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>

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.

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:

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."""

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