fix(runtime): make orphan reconciliation lease-aware (#4427)

CHANGELOG.md

@@ -198,6 +198,9 @@ This section accumulates work toward the **2.1.0** milestone
 
 ### Fixed
 
+- **runtime:** Re-check orphan candidates through an atomic, lease-aware takeover
+  claim so a successful heartbeat after the scan keeps the run active and only
+  one reconciler reports recovery. ([#4424])
 - **skills:** Apply `allowed-tools` only to slash-activated or actually loaded
   lead-agent skills, preventing passive enabled skills and evaluation fixtures
   from removing MCP, web, file, and delegation tools from every run. ([#4095],
@@ -1114,3 +1117,4 @@ with **180 merged pull requests** since the first 2.0 milestone tag.
 [#4287]: https://github.com/bytedance/deer-flow/pull/4287
 [#4288]: https://github.com/bytedance/deer-flow/pull/4288
 [#4324]: https://github.com/bytedance/deer-flow/issues/4324
+[#4424]: https://github.com/bytedance/deer-flow/issues/4424

README.md

@@ -276,6 +276,8 @@ DeerFlow still uses `Forwarded` / `X-Forwarded-*` headers to recover the browser
 
 > [!IMPORTANT]
 > The Gateway still owns active run tasks in process, so production defaults to a single Gateway worker (`GATEWAY_WORKERS=1`). Multi-worker deployments require Postgres, the Redis stream bridge (`stream_bridge.type: redis`), and `run_ownership.heartbeat_enabled: true`. The bridge shares SSE delivery and `Last-Event-ID` replay across workers, while lease reconciliation marks runs from dead workers as errors, publishes the terminal stream marker, schedules retained-stream cleanup, and updates the affected thread status. SSE and `/wait` consumers also refresh durable status on heartbeats as a fallback if terminal publication fails. The rolling retained-buffer TTL (`stream_ttl_seconds`) remains a cleanup safety net rather than a run timeout. IM channel state and other process-local services still need their own multi-worker coordination.
+>
+> Reconciliation uses an atomic takeover claim that re-checks the lease after candidate selection, so a successful owner renewal wins over orphan recovery and only one reconciler can report a run as recovered.
 
 See [CONTRIBUTING.md](CONTRIBUTING.md) for detailed Docker development guide.
 

backend/packages/harness/deerflow/runtime/runs/manager.py

@@ -1213,7 +1213,10 @@ class RunManager:
 
         Rows with a still-valid lease are skipped — they belong to another live
         worker. Rows with a NULL lease (pre-ownership data) are reclaimed as
-        well, matching the original single-worker recovery behaviour.
+        well, matching the original single-worker recovery behaviour. The
+        candidate scan is only an optimization: each row is claimed with a
+        lease-aware conditional update so a heartbeat renewal after the scan
+        always wins over reconciliation.
         """
         if self._store is None:
             return []

backend/tests/test_multi_worker_run_ownership.py

@@ -403,6 +403,51 @@ async def test_reconciliation_skips_candidate_when_owner_renews_lease_after_scan
     assert datetime.fromisoformat(stored["lease_expires_at"]) > datetime.now(UTC)
 
 
+@pytest.mark.anyio
+async def test_concurrent_reconcilers_report_one_successful_claim():
+    """Two reapers scanning the same candidate must report one recovery."""
+    store = MemoryRunStore()
+    grace = 10
+    run_id = "contended-orphan"
+    await store.put(
+        run_id,
+        thread_id="thread-1",
+        status="running",
+        owner_worker_id="worker-dead",
+        lease_expires_at=(datetime.now(UTC) - timedelta(seconds=grace + 5)).isoformat(),
+        created_at=(datetime.now(UTC) - timedelta(seconds=120)).isoformat(),
+    )
+
+    original_scan = store.list_inflight_with_expired_lease
+    both_scanned = asyncio.Event()
+    scan_lock = asyncio.Lock()
+    scan_count = 0
+
+    async def synchronized_scan(*, before=None, grace_seconds=10):
+        nonlocal scan_count
+        rows = [dict(row) for row in await original_scan(before=before, grace_seconds=grace_seconds)]
+        async with scan_lock:
+            scan_count += 1
+            if scan_count == 2:
+                both_scanned.set()
+        await asyncio.wait_for(both_scanned.wait(), timeout=1)
+        return rows
+
+    store.list_inflight_with_expired_lease = synchronized_scan
+    managers = [
+        _make_manager(store=store, run_ownership_config=_lease_config(heartbeat_enabled=True, grace_seconds=grace)),
+        _make_manager(store=store, run_ownership_config=_lease_config(heartbeat_enabled=True, grace_seconds=grace)),
+    ]
+
+    results = await asyncio.gather(*(manager.reconcile_orphaned_inflight_runs(error="orphaned") for manager in managers))
+
+    assert sorted(len(recovered) for recovered in results) == [0, 1]
+    assert [record.run_id for recovered in results for record in recovered] == [run_id]
+    row = await store.get(run_id)
+    assert row is not None
+    assert row["status"] == "error"
+
+
 @pytest.mark.anyio
 async def test_reconciliation_claims_null_lease_runs():
     """Pre-ownership rows (NULL lease) must be reclaimed."""

backend/tests/test_run_repository.py

@@ -958,6 +958,47 @@ class TestRunRepository:
         assert row["status"] == "running"
         await _cleanup()
 
+    @pytest.mark.anyio
+    async def test_reconciliation_skips_run_renewed_after_scan(self, tmp_path):
+        """The SQL takeover CAS must reject a candidate renewed after its scan."""
+        repo = await _make_repo(tmp_path)
+        grace = 10
+        run_id = "renewed-after-scan"
+        owner_worker_id = "worker-alive"
+        try:
+            await repo.put(
+                run_id,
+                thread_id="t1",
+                status="running",
+                owner_worker_id=owner_worker_id,
+                lease_expires_at=(datetime.now(UTC) - timedelta(seconds=grace + 5)).isoformat(),
+                created_at=(datetime.now(UTC) - timedelta(seconds=120)).isoformat(),
+            )
+            original_scan = repo.list_inflight_with_expired_lease
+
+            async def scan_then_renew(*, before=None, grace_seconds=10):
+                rows = await original_scan(before=before, grace_seconds=grace_seconds)
+                renewed = await repo.update_lease(
+                    run_id,
+                    owner_worker_id=owner_worker_id,
+                    lease_expires_at=(datetime.now(UTC) + timedelta(seconds=60)).isoformat(),
+                )
+                assert renewed is True
+                return rows
+
+            repo.list_inflight_with_expired_lease = scan_then_renew
+            manager = RunManager(store=repo)
+
+            recovered = await manager.reconcile_orphaned_inflight_runs(error="orphaned")
+
+            row = await repo.get(run_id)
+            assert recovered == []
+            assert row is not None
+            assert row["status"] == "running"
+            assert datetime.fromisoformat(row["lease_expires_at"]) > datetime.now(UTC)
+        finally:
+            await _cleanup()
+
     @pytest.mark.anyio
     async def test_claim_for_takeover_succeeds_with_null_lease(self, tmp_path):
         repo = await _make_repo(tmp_path)