11 Commits

Author SHA1 Message Date
lau0708
ef3c1c2aee
fix(runtime): share one change position across an atomic thread operation (#5663)
* fix(runtime): share one change position across an atomic thread operation

MemoryRunStore.create_thread_operation_atomic marked each row it touched
separately, so one interrupt-and-replace consumed two positions in the
(change_seq, run_id) cursor that list_changed consumers page with. The SQL
store allocates a single position for the whole set, and runtime/AGENTS.md
documents that as the contract for an atomic thread operation.

The memory store is the backend an install runs whenever no durable database
is configured, so a cursor reader could observe the interrupted row and its
replacement under two positions instead of one, ordered by run_id.

Allocate one position after the raise-only scan -- so a rejected operation
still consumes nothing -- and reuse it for the claimed rows and the new row.

* test(runtime): pin that a rejected thread operation consumes no position

create_thread_operation_atomic allocates its change position only after the
raise-only candidate scan, so a ConflictError consumes nothing -- the memory
store's counterpart of the SQL store rolling back and leaving its clock
untouched. Nothing asserted that ordering.

A consumed-but-unused position leaves no trace in the rows themselves, so a
test comparing list_changed output passes even with the allocation hoisted
above the scan. Assert instead on the position the next accepted operation
lands on: that surfaces a hoist as a gap.
2026-09-22 11:38:43 +08:00
Hyeonsang Cho
28a81452ce
fix(runtime): stop idempotent reuse from blocking the thread on a peer worker (#5393)
* fix(runtime): stop idempotent reuse from blocking the thread on a peer worker

When an idempotent run admission lands on a worker that does not own the
run, RunManager hydrates the stored row and returns it as the reused
record. It also registered that row in the worker's local run map, but
only the owning worker's task lifecycle finalizes and cleans up local
records, so the copy kept its admission-time pending/running status for
the life of the process. On that worker every later reject-strategy
admission for the thread returned 409 until a restart, run reads kept
reporting the stale status, orphan reconciliation skipped the run as
locally live if the owner crashed, and a cancel took the local-owner
path and marked the owner's still-running row interrupted.

Return the hydrated row as a detached store-only handle instead of
registering it. A local record for the key is already returned before
the store insert, so the removed lookup of an existing local record was
unreachable. get(), cancel() and reconciliation now read the durable row
on the peer, matching the documented non-owner contract.

* test(runtime): pin keyed retries of a terminal reused run on the SQL store

Review follow-up on #5393: the post-cleanup release relied on a keyed
retry resolving through the terminal row's idempotency conflict, but
only MemoryRunStore pinned that path, and no test retried on the owner
after its local record was cleaned up.

The SQL repository test now retries the key on the peer and on the
owner once the run is terminal and cleaned up, asserting both get the
same run back as a store-only reused handle before the keyless
follow-up is admitted.
2026-09-13 18:16:22 +08:00
MiaoRuidx
8a78c264b7
fix(runtime): cancel runs across live gateway workers (#4500)
* docs(runtime): design cross-worker cancellation

* fix(runtime): cancel runs across gateway workers

* fix(runtime): harden cross-worker cancellation races

让取消请求与 owner 终态写入通过持久化 CAS 决定先后,保证首次取消 action 在不同 worker 路由下保持一致。\n\n将 heartbeat 收敛为续租后仅发送本地中止信号,并补齐完成竞态与路由重试的回归用例。

* docs(runtime): drop implementation plan from PR

移除仅用于实现过程的跨 worker 取消设计记录,保留 README 和 backend/AGENTS.md 中面向最终行为的文档。

* fix(runtime): preserve local cancel fallback

* test(runtime): adapt worker run manager fakes

* docs(runtime): fix run cancel migration registry

---------

Co-authored-by: MiaoRuidx <12540796+MiaoRuidx@users.noreply.github.com>
2026-07-28 21:45:20 +08:00
Huixin615
090e80c1dd
fix(runtime): fail-stop runs when lease ownership cannot be confirmed (#4431)
* fix(runtime): fail-stop runs after lease expiry

* test(runtime): cover late successful lease renewal

---------

Co-authored-by: Willem Jiang <willem.jiang@gmail.com>
2026-07-27 07:25:34 +08:00
Huixin615
8af760fc30
fix(runtime): make orphan reconciliation lease-aware (#4427) 2026-07-25 23:26:17 +08:00
Vanzeren
3c8b82c594
fix(runtime): serialize checkpoint writes with active runs (#4437)
* fix(runtime): serialize checkpoint writes with active runs

* fix(runtime): address checkpoint reservation reviews

* fix(runtime): address reservation race reviews

* fix(runtime): refine reservation conflict semantics
2026-07-25 23:18:34 +08:00
H Haidong
c7538cfb35
fix(runs): terminate orphaned streams after lease recovery (#4420)
* fix(runs): terminate orphaned streams after lease recovery

* fix(runs): include recovered ids in callback warnings

* fix(runs): harden orphan recovery lifecycle
2026-07-24 19:34:20 +08:00
MiaoRuidx
80c06414f8
fix: make orphan reconciliation lease-aware (#4434)
让启动/孤儿 run 恢复在最终写入前通过 claim_for_takeover 原子重查 lease,避免 owner 在扫描后续约成功仍被误标为 error。

补充扫描后续约的回归测试,并把 reconciliation 写失败测试迁移到 takeover claim 路径。

Co-authored-by: MiaoRuidx <12540796+MiaoRuidx@users.noreply.github.com>
Co-authored-by: Willem Jiang <willem.jiang@gmail.com>
2026-07-24 09:48:48 +08:00
Vanzeren
1769b2de0d
fix: read run stop_reason from runtime context (#4188)
* fix: read run stop_reason from runtime context

* fix: address review feedback for #4188 stop_reason integration

   - migration 0005: use safe_add_column for consistency and drift detection
   - worker: clear runtime.context stop_reason at start of each _stream_once
     turn so a clean continuation doesn't inherit a prior cap reason
   - tests: replace circular unit test with real middleware integration
     tests that exercise LoopDetectionMiddleware._apply and
     TokenBudgetMiddleware._apply through the worker, proving the full
     middleware → runtime.context → persist pipeline

* fix(test): resume conftest

* fix: stamp stop_reason in all guard middlewares, fix clearing semantics
2026-07-16 08:19:52 +08:00
heart-scalpel
b53c1ae0e0
fix(runs): cancel degrades to lease takeover for multi-worker (#4064)
* fix(runs): cancel degrades to lease takeover for multi-worker

Work item 4 of the multi-worker ownership epic
(https://github.com/bytedance/deer-flow/issues/3948).

Problem: POST /runs/{run_id}/cancel landing on a non-owning worker
returns 409 — the cancel button silently fails under GATEWAY_WORKERS>1
with no sticky routing. cancel() required the current worker to hold
the in-memory task/abort_event, which any non-owner pod cannot satisfy.

Changes:
- RunManager.cancel() returns CancelOutcome enum (cancelled /
  taken_over / lease_valid_elsewhere / not_active_locally /
  not_cancellable / unknown) instead of bool, so the router can map
  each outcome to the right HTTP response.
- New store primitive claim_for_takeover(): a single atomic
  conditional UPDATE that marks a run as error only when
  status IN (pending, running) AND (lease IS NULL OR
  lease < now - grace). Closes the stale-read / concurrent-heartbeat
  race — if the owner renews between our read and write, the UPDATE
  matches 0 rows and we surface lease_valid_elsewhere.
- HTTP cancel + stream-join endpoints route on CancelOutcome:
  cancelled -> 202 (or 204 with wait=true); taken_over -> 202
  immediately (no SSE streaming — the run is terminal on another
  worker, streaming would hang); lease_valid_elsewhere -> 409 +
  Retry-After header computed from lease_expires_at + grace_seconds.
- RunManager.grace_seconds exposed as a public property; the router
  no longer reaches into _run_ownership_config.
- _is_lease_expired extracted to a module-level function, shared by
  RunManager.cancel() and MemoryRunStore.claim_for_takeover().
- GATEWAY_WORKERS=1 + heartbeat_enabled=false is zero-regression:
  the non-local path short-circuits to not_active_locally, preserving
  the original 409 behaviour the existing tests pin.

Tests: 12 new (5 store primitive + 4 cancel-takeover unit + 3 HTTP
including a regression guard verifying POST /stream?action=interrupt
on a dead-owner run returns 202 instead of hanging on SSE).
244 directly-related tests pass; 36/36 blocking-IO gate pass.

Co-Authored-By: Claude Opus 4.7 <noreply@anthropic.com>

* fix(runs): guard update_status and self-terminate on takeover

Two defenses close a split-brain window where the original owner
could overwrite a peer's takeover status:

- update_status (SQL + memory store) now guards on
  status IN ('pending','running'). When takeover already set
  the row to 'error', the owner's final status write matches
  0 rows and is dropped.

- _persist_status: when update_status returns False, check
  whether the row exists before attempting recovery via put().
  If the row exists (takeover by another worker), skip recovery
  instead of blindly upserting over the takeover.

- Heartbeat _renew_leases: when update_lease returns False
  (row no longer pending/running or owner changed), cancel the
  local task so wasted CPU is bounded to the next heartbeat
  tick (~10s) instead of the full task lifetime.

Also fix three reviewer feedback items:

- Re-fetch the store row when cancel() returns
  lease_valid_elsewhere, so Retry-After uses the owner's
  freshly-renewed lease instead of a stale value from
  request start.

- Fallback 'unknown' in takeover error message when
  owner_worker_id is NULL (pre-ownership data).

- Remove dead else-10 branch from grace_seconds property
  (unreachable — all callers are downstream of the
  heartbeat_enabled guard).

Co-Authored-By: Claude Opus 4.7 <noreply@anthropic.com>

* test(runs): pin split-brain defences from update_status guard + heartbeat

Three tests lock down the takeover authoritativeness so a
late-running owner cannot overwrite a peer's claim:

- update_status must reject writes when the store row is already
  terminal (taken over by another worker).
- _persist_status must skip row-recovery via put() when the row
  exists but has been taken over.
- Heartbeat _renew_leases must cancel the local task when
  update_lease returns False (row claimed by another worker).

Co-Authored-By: Claude Opus 4.7 <noreply@anthropic.com>

* fix(runs): precise outcome + log when local cancel loses to peer takeover

Two reviewer precision nits on the split-brain defence:

- _persist_status: branch the skip-reason log on existing["status"].
  error → WARNING "peer takeover" (anomalous); interrupted/success →
  INFO "local cancel/completion race" (expected when user hits stop
  as the run finishes). Stops noisy false-positive takeover warnings
  in operator logs.

- cancel() local path: when _persist_status returns False, re-check
  the store. If a peer's claim_for_takeover flipped the row to error
  between our in-memory cancel and the guarded update_status, surface
  taken_over instead of cancelled so the client sees a status
  consistent with the store.

Test: test_cancel_returns_taken_over_when_peer_claims_during_local_cancel
pins the race outcome.

Co-Authored-By: Claude Opus 4.7 <noreply@anthropic.com>

* fix(runs): widen update_status guard, de-duplicate lease helpers, add coverage

Round 3 of reviewer feedback:

- Widen update_status guard to status IN ('pending','running','interrupted').
  The original guard blocked interrupted→error (the rollback finalize path),
  losing the "Rolled back by user" message. interrupted is now permitted
  while error/success stay locked — takeover protection unchanged.

- claim_for_takeover False now re-reads the store row to distinguish causes:
  owner renewed lease → lease_valid_elsewhere; row went terminal →
  not_cancellable; another worker already took it over → taken_over.

- Extract _raise_lease_valid_elsewhere() helper to de-duplicate the
  409+Retry-After block shared across cancel_run and stream_existing_run.

- Extract _lease_expired_or_null() in persistence/run/sql.py to
  de-duplicate the lease-expiry SQL WHERE clause shared by
  claim_for_takeover and list_inflight_with_expired_lease.

- 11 new tests: 5 SQL-layer claim_for_takeover (expired/valid/NULL/
  terminal/nonexistent), 3 _compute_retry_after unit (NULL/unparseable/
  normal), 2 claim re-read precision (terminal/takeover), 1 stream
  endpoint 409+Retry-After.

Not addressed (non-blocking, reviewer agreed):
- The 2–3 store.gets in the takeover cold path: optimizing the API to
  accept a pre-fetched record would couple the router to the manager
  more tightly than justified by the perf gain.
- The lease-expiry inline loop in MemoryRunStore.list_inflight_with_-
  expired_lease pre-computes cutoff once for all rows; switching to the
  shared _is_lease_expired helper would recompute datetime.now() per row
  with no real benefit.

260 related tests pass; 36/36 blocking-IO gate pass; ruff clean.

Co-Authored-By: Claude Opus 4.7 <noreply@anthropic.com>

* fix(runs): de-duplicate lease-expiry helper, restore defensive fallback

Address final round of review feedback:

- Extract is_lease_expired to deerflow.utils.time (no _ prefix, public
  utility). Manager and MemoryRunStore now import from the same place
  instead of the store reaching backward into the manager for a private
  function.

- Restore defensive else-10 fallback in grace_seconds property (removed
  in an earlier round). The guard is unreachable for current callers but
  protects future ones from AttributeError.

- Comment the transient in-memory interrupted vs store error state when
  a local cancel is superseded by a peer takeover.

- Comment the max(1, ...) floor in _compute_retry_after — the floor is
  a lower bound, not a poll interval; clients should apply jitter.

Co-Authored-By: Claude Opus 4.7 <noreply@anthropic.com>

---------

Co-authored-by: Claude Opus 4.7 <noreply@anthropic.com>
Co-authored-by: rayhpeng <rayhpeng@gmail.com>
2026-07-14 07:37:59 +08:00
heart-scalpel
3bc3af2530
fix(runs): close multi-worker ownership gaps in run atomicity (#3948) (#4003)
* feat(runs): cross-process run ownership with lease + reconciliation (#3948)

Implements work items 2 and 3 of the multi-worker P0 plan
(docs/multi_worker.md). Work item 1 (Postgres startup gate, #3960)
already landed; this PR makes run creation race-safe across worker
processes and lets Postgres deployments recover orphaned inflight runs
from crashed workers without mis-marking live runs as orphans.

Work item 2 — cross-process atomic create_or_reject

- Alembic revision 0004_run_ownership adds runs.owner_worker_id,
  runs.lease_expires_at, idx_runs_lease, and a partial unique index
  uq_runs_thread_active (one pending/running run per thread). The
  index is declared on RunRow.__table_args__ with sqlite_where +
  postgresql_where (mirroring uq_channel_connection_active_identity)
  so the empty-DB bootstrap path — which runs Base.metadata.create_all
  + alembic stamp head without executing any revision's upgrade() —
  also lands it on fresh deployments. Migration 0004 additionally
  creates it idempotently for legacy/versioned upgrades.
- RunRepository.create_run_atomic is the new atomic primitive:
  - reject: INSERT directly; the partial unique index catches
    duplicate active runs; the manager surfaces the result as
    ConflictError.
  - interrupt/rollback: SELECT FOR UPDATE the conflicting rows,
    skip rows whose lease is still valid AND owned by another live
    worker (raise ConflictError — the INSERT would have failed on
    the index anyway, and a retry loop cannot make progress),
    cancel the rest in the same transaction, then INSERT the new
    row. Rows owned by this worker are interruptible regardless of
    lease state.
- RunManager.create_or_reject dispatches to the store under the
  existing local lock; same-worker in-memory cancellation runs after
  the store commit succeeds. MemoryRunStore mirrors the same
  semantics for tests and database.backend=memory.

Work item 3 — lease heartbeat + Postgres reconciliation

- RunOwnershipConfig (lease_seconds=30, grace_seconds=10,
  heartbeat_enabled=false by default), registered as startup-only in
  reload_boundary.STARTUP_ONLY_FIELDS because the heartbeat background
  task is created once in langgraph_runtime() and is not rebuilt on
  config.yaml edits.
- When heartbeat_enabled, each worker renewes leases on its own
  active runs with interval = lease_seconds / 3. The loop is bounded
  and stop-event-cancellable so shutdown is prompt.
- reconcile_orphaned_inflight_runs now runs on every backend — the
  sqlite-only gate in app/gateway/deps.py is dropped in the same
  commit so there is no window where Postgres would mis-mark live
  Worker A runs as orphans. Reconciliation errors only runs whose
  lease is NULL (legacy pre-ownership rows) or older than
  grace_seconds. In single-worker mode (heartbeat off, NULL leases)
  all inflight rows reclaim immediately, preserving the pre-ownership
  recovery latency.
- Heartbeat starts AFTER startup reconciliation and stops BEFORE the
  in-flight run drain on shutdown so the two cannot race.

GATEWAY_WORKERS=1 with heartbeat_enabled=false keeps current behavior.

Verified: 170 related tests + full backend suite (minus Docker-gated
live tests) green; ruff check + ruff format clean.

* fix(runs): tighten unique-violation handling and document clock-sync budget

Three follow-up fixes to the cross-process run ownership work in #3948,
surfacing during review.

1. _is_unique_violation: detect by driver-native signal, not message text

   The previous substring heuristic ("unique" + "violat", or "duplicate")
   missed SQLite's actual phrasing "UNIQUE constraint failed: <table>.<index>"
   — SQLite says "failed", not "violates", and never "duplicate". On SQLite
   the detector returned False, the reject path re-raised the raw
   IntegrityError, and clients saw HTTP 500 instead of ConflictError 409.
   The conversion is the load-bearing piece of the "store is source of
   truth" design but was untested — every atomic test used MemoryRunStore,
   which raises ConflictError directly and never reached this branch.

   Now prefers driver-native signals: psycopg pgcode/sqlcode "23505" and
   sqlite3 sqlite_errorcode SQLITE_CONSTRAINT_UNIQUE (reachable through
   SQLAlchemy IntegrityError.orig). Message matching stays as a fallback
   with SQLite's exact "unique constraint failed" phrase added.

2. interrupt/rollback: convert exhausted-retry IntegrityError to ConflictError

   The reject branch converts unique violations to ConflictError. The
   interrupt/rollback retry loop did not — on the 3rd attempt it re-raised
   the raw IntegrityError, leaking HTTP 500 for the same race condition
   that reject surfaces as 409. Symmetric conversion added after the loop;
   callers now see a consistent ConflictError regardless of strategy.

3. Document clock-sync requirement for multi-worker lease reconciliation

   reconcile_orphaned_inflight_runs compares another worker's UTC
   lease_expires_at against this worker's datetime.now(UTC). The only skew
   budget is grace_seconds (default 10s) — worst case, with the owning
   worker's heartbeat just about to fire, a peer whose clock is more than
   ~grace_seconds ahead can mis-reclaim a still-live run as an orphan.

   Documented in RunOwnershipConfig's docstring (with the math) and in
   config.example.yaml (with operational guidance), so operators in
   NTP-poor environments know to raise grace_seconds. Default unchanged:
   10s is reasonable for NTP-synced K8s/cloud, and bumping it would slow
   recovery of genuinely dead workers (lease_seconds + grace_seconds from
   last heartbeat to reclaim).

Tests:
- test_create_run_atomic_reject_propagates_conflict_on_unique_violation:
  end-to-end against a real SQLite-backed RunRepository, pre-inserts an
  active run, asserts reject-strategy create surfaces as ConflictError
  rather than raw IntegrityError.
- test_is_unique_violation_detects_real_sqlite_integrity_error: unit test
  for the detector against a real SQLite-raised IntegrityError; asserts
  driver-level sqlite_errorcode is SQLITE_CONSTRAINT_UNIQUE.
- test_interrupt_exhausted_retries_surface_as_conflict_error: pins the
  symmetric 409 behavior after the retry loop exhausts.

Verified: ruff check + ruff format clean; multi-worker + run_repository
+ owner_isolation + reload_boundary suites green.

Co-Authored-By: Claude Opus 4.7 <noreply@anthropic.com>

* fix(runs): close multi-worker ownership gaps in lease heartbeat and unique-violation detection

Five code-review fixes from docs/multi_worker.md:

1. Drop unused ``claim_inflight_runs`` primitive — no caller anywhere.
   ``create_run_atomic`` does its own inline claim (SELECT FOR UPDATE +
   cancel) inside the INSERT transaction; a separate claim primitive
   would split that into two transactions and open a claim→INSERT race.
   Removes ~40 lines across base.py / memory.py / sql.py plus the
   unused ``now_iso`` parameter, freeing future RunStore implementations
   from providing it.

2. Broaden ``_renew_leases`` filter to renew pending/running runs owned
   by this worker even when ``record.task is None``. The previous
   ``task is not None`` requirement skipped the brief window between
   ``create_run_atomic`` inserting the row and the worker spawning the
   agent task; under event-loop load that window can approach
   ``lease_seconds``, after which peer reconciliation marks the run
   ``error`` (visible) or a peer's ``create_or_reject("interrupt")``
   silently kills the queued run. Filter now:
   ``task is None or not task.done()``.

3. Document the unsynchronised ``record.lease_expires_at = new_expiry``
   write. ``lease_expires_at`` is the only field on an existing record
   this path mutates; ``set_status`` / ``_persist_status`` touch other
   fields, so there is no concurrent writer to race against. Re-acquiring
   ``self._lock`` would serialise unrelated run mutations for no gain.

4. Gate ``_is_unique_violation`` message fallbacks on
   ``isinstance(current, (SAIntegrityError, sqlite3.IntegrityError))``.
   The driver-code path (pgcode/sqlite_errorcode) remains load-bearing;
   substring fallbacks are now belt-and-suspenders only for cases where
   the driver attribute isn't reachable through the cause chain. Without
   the gate, any application exception whose ``str()`` happens to contain
   "duplicate key" / "unique" + "violat" (CHECK constraint, validation
   error) would silently surface as HTTP 409 instead of 500.

5. Route ``update_lease`` through ``_call_store_with_retry`` for
   consistency with every other store call, and wrap
   ``await self._renew_leases()`` in ``_heartbeat_loop`` with
   ``except Exception: logger.warning(...)``. Previously a transient
   error from the snapshot path or an unexpected exception would kill
   the heartbeat task silently — after which no lease is ever renewed
   again and every active run eventually looks orphaned.
   ``except Exception`` lets ``CancelledError`` (BaseException since
   3.8) propagate so shutdown cancellation still works.

Regression tests:
- ``test_heartbeat_renews_pending_run_before_task_is_spawned``
- ``test_is_unique_violation_does_not_misclassify_application_exception``

Co-Authored-By: Claude Opus 4.7 <noreply@anthropic.com>

* fix(runs): harden multi-worker migration, memory atomicity, and tz-naive lease comparison

Three follow-up fixes to the multi-worker run ownership work:

- migration 0004 dedupe pass: cancel superseded duplicate active rows per
  thread before creating the partial UNIQUE index ``uq_runs_thread_active``
  so dirty DBs (Postgres deployments that had reconciliation skipped by the
  old sqlite-only gate, or any env that ran GATEWAY_WORKERS>1 before this PR)
  do not abort the alembic upgrade and block gateway startup. Keeps the
  newest active row per thread, marks the rest as error with an explanatory
  message.

- MemoryRunStore.create_run_atomic interrupt/rollback path: split the single-
  pass loop into two passes (collect candidates, validate, then mutate) so a
  ConflictError raised on a later candidate does not leave earlier candidates
  half-interrupted. Mirrors the SQL store's transactional rollback semantics;
  the entire test_multi_worker_run_ownership.py suite runs against memory so
  this divergence was giving false confidence.

- RunRepository.create_run_atomic interrupt path: coerce tz-naive
  ``row.lease_expires_at`` to UTC before comparing against the aware
  ``cutoff``. SQLite drops tzinfo on read despite ``DateTime(timezone=True)``
  (this file's own comment acknowledges it), so the Python-side comparison
  raised ``TypeError: can't compare offset-naive and offset-aware datetimes``
  whenever heartbeat was enabled on SQLite and a lease was non-NULL. Defaults
  (heartbeat off -> leases always NULL) masked it, but there was no guard
  against the combination. Follows the existing "naive is UTC" convention
  from ``coerce_iso``.

Each fix ships with a regression test pinning the behavior.

Co-Authored-By: heart-scalpel <heart-scalpel@users.noreply.github.com>
Co-Authored-By: Claude Opus 4.7 <noreply@anthropic.com>

* fix(runs): enforce heartbeat for multi-worker, fix memory-store datetime comparison, lazy-import ConflictError in store layer

Three fixes from code review:

1. Extend the startup gate (GATEWAY_WORKERS>1) to also require
   run_ownership.heartbeat_enabled=true. Without heartbeat every run has
   a NULL lease, so reconciliation treats all inflight rows as orphans
   and Worker B would kill Worker A's live runs on every rolling update
   or scale-up.

2. Fix MemoryRunStore.list_inflight_with_expired_lease to parse
   created_at as datetime instead of ISO string lexical comparison,
   and handle tz-naive lease values uniformly with the SQL store.

3. Store layer (sql.py, memory.py) now lazy-imports ConflictError
   inside create_run_atomic instead of importing from the higher
   RunManager layer at module level.

Co-Authored-By: Claude Opus 4.7 <noreply@anthropic.com>

* fix(runs): add owner check to update_lease, document create() assumption, restore deleted comment

- update_lease (SQL + memory) now requires owner_worker_id match in WHERE
  clause so the primitive is safe by construction against misuse
- create() docstring notes it bypasses atomic create_run_atomic and
  assumes no active run exists for the thread
- restore explanatory comment in MemoryRunStore.aggregate_tokens_by_thread
  that was dropped in an earlier commit

Co-Authored-By: Claude Opus 4.7 <noreply@anthropic.com>

* fix(runs): add psycopg3 sqlstate detection and periodic orphan reconciliation

- _is_unique_violation now checks sqlstate attribute (psycopg3 uses this
  instead of pgcode). On Postgres, the only supported multi-worker backend,
  detection was falling through to the message-substring fallback.
- _heartbeat_loop now runs reconcile_orphaned_inflight_runs every 3rd
  cycle (every lease_seconds) to catch orphans whose lease expires between
  pod restarts. Single-worker deployments are unaffected (heartbeat off).

Co-Authored-By: Claude Opus 4.7 <noreply@anthropic.com>

---------

Co-authored-by: Claude Opus 4.7 <noreply@anthropic.com>
Co-authored-by: heart-scalpel <heart-scalpel@users.noreply.github.com>
Co-authored-by: Willem Jiang <willem.jiang@gmail.com>
2026-07-11 16:05:30 +08:00