diff --git a/backend/AGENTS.md b/backend/AGENTS.md index e40688f10..ff2cd14c7 100644 --- a/backend/AGENTS.md +++ b/backend/AGENTS.md @@ -441,6 +441,7 @@ metadata only. - When a persistent `RunStore` is configured, `get()` and `list_by_thread()` hydrate historical runs from the store. In-memory records win for the same `run_id` so task, abort, and stream-control state stays attached to active local runs. - `cancel()` returns a :class:`~deerflow.runtime.CancelOutcome` enum: `cancelled` (local cancel), `taken_over` (non-owning worker claimed the run because the owner's lease expired — marks it as `error`), `lease_valid_elsewhere` (owner's lease is still alive — caller should return 409 + `Retry-After`), `not_active_locally` (heartbeat disabled, preserving the old 409 path), `not_cancellable` (terminal state), or `unknown` (not found in memory or store). `create_or_reject(..., multitask_strategy="interrupt"|"rollback")` persists interrupted status through `RunStore.update_status()`, matching normal `set_status()` transitions. - Store-only hydrated runs are readable history. In multi-worker mode with heartbeat enabled, cancel on a store-only run can take over (mark `error`) when the owner's lease has expired past the grace window; otherwise it fails with 409 + `Retry-After`. In single-worker mode (heartbeat off), store-only runs still return 409. +- Startup/orphan reconciliation must claim stale active rows with `RunStore.claim_for_takeover()`, not a plain `update_status()`. The final claim re-checks `status` and lease expiry atomically, so a heartbeat renewal between the candidate scan and the recovery write keeps the run active. - `POST /wait` (both thread-scoped and `/api/runs/wait`) drains the stream bridge via `wait_for_run_completion()` instead of bare `await record.task`, so it honours the run's `on_disconnect` setting and cancels the background run on real client disconnect rather than returning a stale checkpoint (issue #3265). - Redis `StreamBridge` keys use a rolling retained-buffer TTL (`stream_bridge.stream_ttl_seconds`, refreshed on `publish()` / `publish_end()`) as a leak safety net, not as a run timeout. Startup orphan recovery publishes `END_SENTINEL` and schedules stream cleanup for recovered runs; malformed `Last-Event-ID` reconnect values live-tail new Redis events rather than replaying the retained buffer. Do not broaden this into a shared-database multi-pod reaper without adding worker ownership/liveness first. - Thread-scoped run creation accepts `checkpoint` / `checkpoint_id`; Gateway validates the checkpoint belongs to the request thread before writing `checkpoint_id` / `checkpoint_ns` into `config.configurable` for LangGraph branching. diff --git a/backend/packages/harness/deerflow/runtime/runs/manager.py b/backend/packages/harness/deerflow/runtime/runs/manager.py index 588498a94..0a8b6aad6 100644 --- a/backend/packages/harness/deerflow/runtime/runs/manager.py +++ b/backend/packages/harness/deerflow/runtime/runs/manager.py @@ -1130,13 +1130,28 @@ class RunManager: # Still owned by a local task — skip continue + try: + claimed = await self._call_store_with_retry( + "claim_for_takeover", + record.run_id, + lambda: self._store.claim_for_takeover( + record.run_id, + grace_seconds=grace_seconds, + error=error, + ), + ) + except Exception: + logger.warning("Failed to claim orphaned run %s for reconciliation", record.run_id, exc_info=True) + continue + if not claimed: + logger.info( + "Skipped orphaned run %s recovery because the takeover claim no longer matched", + record.run_id, + ) + continue record.status = RunStatus.error record.error = error record.updated_at = now - persisted = await self._persist_status(record, RunStatus.error, error=error) - if not persisted: - logger.warning("Skipped orphaned run %s recovery because error status was not persisted", record.run_id) - continue recovered.append(record) if recovered: diff --git a/backend/tests/test_multi_worker_run_ownership.py b/backend/tests/test_multi_worker_run_ownership.py index aadd99456..7f298c663 100644 --- a/backend/tests/test_multi_worker_run_ownership.py +++ b/backend/tests/test_multi_worker_run_ownership.py @@ -260,6 +260,49 @@ async def test_reconciliation_skips_active_lease_runs(): assert stored["status"] == "running" +@pytest.mark.anyio +async def test_reconciliation_skips_candidate_when_owner_renews_lease_after_scan(): + """A renewed lease between scan and claim must keep the run active.""" + store = MemoryRunStore() + grace = 10 + expired_lease = (datetime.now(UTC) - timedelta(seconds=grace + 5)).isoformat() + await store.put( + "race-run", + thread_id="thread-1", + status="running", + owner_worker_id="worker-alive", + lease_expires_at=expired_lease, + created_at=(datetime.now(UTC) - timedelta(seconds=120)).isoformat(), + ) + original_list = store.list_inflight_with_expired_lease + + async def list_then_owner_renews(*, before=None, grace_seconds=10): + rows = [dict(row) for row in await original_list(before=before, grace_seconds=grace_seconds)] + renewed_lease = (datetime.now(UTC) + timedelta(seconds=60)).isoformat() + updated = await store.update_lease( + "race-run", + owner_worker_id="worker-alive", + lease_expires_at=renewed_lease, + ) + assert updated is True + return rows + + store.list_inflight_with_expired_lease = list_then_owner_renews + manager = _make_manager( + store=store, + run_ownership_config=_lease_config(heartbeat_enabled=True, grace_seconds=grace), + ) + + recovered = await manager.reconcile_orphaned_inflight_runs( + error="Gateway restarted before this run reached a durable final state.", + ) + + assert recovered == [] + stored = await store.get("race-run") + assert stored["status"] == "running" + assert datetime.fromisoformat(stored["lease_expires_at"]) > datetime.now(UTC) + + @pytest.mark.anyio async def test_reconciliation_claims_null_lease_runs(): """Pre-ownership rows (NULL lease) must be reclaimed.""" diff --git a/backend/tests/test_run_manager.py b/backend/tests/test_run_manager.py index 5a61e3fa9..9a3433749 100644 --- a/backend/tests/test_run_manager.py +++ b/backend/tests/test_run_manager.py @@ -61,15 +61,15 @@ class PermanentStatusRunStore(MemoryRunStore): ) -class FailingStatusRunStore(MemoryRunStore): - """Memory run store that always fails status updates.""" +class FailingTakeoverRunStore(MemoryRunStore): + """Memory run store that always fails takeover claims.""" def __init__(self) -> None: super().__init__() - self.status_update_attempts = 0 + self.takeover_attempts = 0 - async def update_status(self, run_id, status, *, error=None, stop_reason=None): - self.status_update_attempts += 1 + async def claim_for_takeover(self, run_id, *, grace_seconds, error): + self.takeover_attempts += 1 raise sqlite3.OperationalError("database is locked") @@ -301,9 +301,9 @@ async def test_reconcile_orphaned_inflight_runs_skips_live_local_run(): @pytest.mark.anyio -async def test_reconcile_orphaned_inflight_runs_skips_rows_when_error_status_is_not_persisted(): - """Startup recovery must not report a row as recovered if the error update failed.""" - store = FailingStatusRunStore() +async def test_reconcile_orphaned_inflight_runs_skips_rows_when_takeover_claim_fails(): + """Startup recovery must not report a row as recovered if the takeover claim failed.""" + store = FailingTakeoverRunStore() await store.put("running-run", thread_id="thread-1", status="running", created_at="2026-01-01T00:00:00+00:00") manager = RunManager( store=store, @@ -318,7 +318,7 @@ async def test_reconcile_orphaned_inflight_runs_skips_rows_when_error_status_is_ stored = await store.get("running-run") assert recovered == [] assert stored["status"] == "running" - assert store.status_update_attempts == 2 + assert store.takeover_attempts == 2 @pytest.mark.anyio