fix(runtime): keep renewing the lease while a staged terminal run finalizes (#6263)

* fix(runtime): keep renewing the lease while a staged terminal run finalizes

With an event store, run_agent stages the terminal status in memory and
commits it to the run store only after the journal flush, delivery receipt,
workspace scan and duration checkpoint. _renew_leases selected runs by local
status, so it stopped renewing as soon as the status was staged. A
finalization longer than lease_seconds + grace_seconds let a peer, or this
worker's own reconciler, claim the still-active row and record the
successful run as an orphan error.

The worker now marks RunRecord.terminal_commit_pending until the deferred
commit is attempted, and lease renewal treats it like the existing
scheduled-goal barrier: keep renewing, release on a confirmed own commit,
and fence when a peer claims the row. A fenced run stops renewing.

* docs(changelog): reference #6263 for the staged terminal lease renewal fix

---------

Co-authored-by: Willem Jiang <willem.jiang@gmail.com>

CHANGELOG.md

@@ -466,6 +466,16 @@ This release closes that milestone with **301 merged pull requests**.
 
 ### Fixed
 
+- **runtime:** A multi-worker run that finished successfully is no longer
+  reclaimed as an orphan `error` while its worker is still finalizing. With an
+  event store, the worker records the terminal status in memory first and
+  writes it to the run store only after the journal flush, delivery receipt,
+  workspace scan and duration checkpoint. Lease renewal skipped runs whose
+  local status was already terminal, so a finalization longer than the lease
+  plus grace (about 30–40 seconds by default) let a peer, or the worker's own
+  reconciler, claim the still-active row. The heartbeat now keeps renewing
+  until that deferred write is attempted, and fences the run if a peer claims
+  it. Affects only `run_ownership.heartbeat_enabled` deployments. ([#6263])
 - **channels:** Buzz now follows a thread without a fresh mention for authors
   bound with `/connect`. With `channel_connections.enabled`, the manager maps a
   bound author's threads only in the connection repository, but Buzz's
@@ -7711,3 +7721,4 @@ with **180 merged pull requests** since the first 2.0 milestone tag.
 [#6232]: https://github.com/bytedance/deer-flow/pull/6232
 [#6238]: https://github.com/bytedance/deer-flow/pull/6238
 [#6255]: https://github.com/bytedance/deer-flow/pull/6255
+[#6263]: https://github.com/bytedance/deer-flow/pull/6263

CHANGELOG_zh.md

@@ -423,6 +423,13 @@
 
 ### 修复
 
+- **运行时：** 多 worker 部署中已成功结束的运行，不会再在其 worker 仍在收尾时被
+  当作孤儿运行回收为 `error`。配置了事件存储时，worker 先在内存中记录终态，直到
+  journal 刷新、交付回执、工作区扫描与时长 checkpoint 写入完成后才写入运行存储。
+  此前租约续期会跳过本地状态已是终态的运行，收尾一旦超过租约加宽限期（默认约
+  30–40 秒），其他 worker 或本 worker 自己的回收器就会接管这条仍处于活动状态的
+  记录。心跳现在会持续续期直到这次延迟写入被执行，若期间被其他 worker 接管则隔离
+  本地运行。仅影响启用 `run_ownership.heartbeat_enabled` 的部署。([#6263])
 - **渠道：** 通过 `/connect` 绑定的 Buzz 作者在已参与的话题中回复时，无需再次提及
   机器人。开启 `channel_connections.enabled` 后，管理器只在连接仓库中记录已绑定作者
   的话题映射，而 Buzz 的话题跟随判断只读取 JSON 渠道存储，导致机器人正在回复的话题
@@ -6437,3 +6444,4 @@ DeerFlow 2.0 是围绕"超级智能体"框架的彻底重写，核心包含子
 [#6232]: https://github.com/bytedance/deer-flow/pull/6232
 [#6238]: https://github.com/bytedance/deer-flow/pull/6238
 [#6255]: https://github.com/bytedance/deer-flow/pull/6255
+[#6263]: https://github.com/bytedance/deer-flow/pull/6263

backend/docs/runtime-guidance-details.md

@@ -55,7 +55,8 @@ cancellation while waiting for prior finalization still emit a zero-delivery
 receipt. The worker flushes ordinary journal events, idempotently persists the
 run-scoped receipt, and only then persists the staged terminal run status. A
 receipt failure is retried on a short bounded schedule while the owning worker
-still knows the real outcome and holds the lease. Delivery candidates are every
+still knows the real outcome and holds the lease; `terminal_commit_pending` keeps
+heartbeat renewal running until the staged status is committed. Delivery candidates are every
 regular file created or modified under `/mnt/user-data/outputs`; internal
 process-feedback files are excluded (the scanner's `EXCLUDED_DIR_NAMES` plus
 the configured `tool_output.storage_subdir`), so a run that only externalized

backend/packages/harness/deerflow/runtime/AGENTS.md

@@ -72,6 +72,13 @@ answers for paginated history. See `docs/skill-usage-ui.md`.
 
 **Run delivery receipts:** Journal artifact evidence and terminal status must finalize before an ordinary satisfied goal is cleared. That cleanup uses a durable checkpoint-write reservation; delivery failure retains the ordinary goal without another continuation. The scheduled-only exception is described below. Details: `backend/docs/runtime-guidance-details.md`.
 
+**Deferred terminal commit:** With an event store, the worker stages its terminal
+status locally and commits it only after finalization's receipt and duration
+writes. `RunRecord.terminal_commit_pending` keeps `_renew_leases()` renewing that
+still-active row until the commit is attempted; a renewal rejected by the worker's
+own commit is confirmed by re-reading the row, while a peer claim fences the run.
+Never select runs for renewal by local status alone.
+
 **Deferred-tool promotion event deduplication** (`runtime/journal.py`): one
 `RunJournal` owns the lead graph's run-scoped atomic promotion claim. Parallel
 `tool_search` Sends read the same pre-step state, so state diffing alone can

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

@@ -230,6 +230,10 @@ class RunRecord:
     goal_verdict: dict[str, Any] | None = None
     # Process-local finalization barrier for a worker-installed scheduled goal.
     scheduled_goal_cleanup_pending: bool = False
+    # Process-local renewal barrier: the worker stages its terminal status in
+    # memory and commits it only at the end of finalization, so the durable row
+    # stays active and its lease must keep renewing until that commit.
+    terminal_commit_pending: bool = False
     idempotency_key: str | None = None
     # True only on the caller that recovered an existing idempotent admission;
     # that caller must not attach a second worker to the durable run.
@@ -2111,6 +2115,8 @@ class RunManager:
             if record.ownership_lost:
                 return True
             record.ownership_lost = True
+            # A fenced worker must not keep the peer-recoverable row alive.
+            record.terminal_commit_pending = False
             record.abort_event.set()
             record.status = RunStatus.error
             record.error = reason
@@ -2198,13 +2204,18 @@ class RunManager:
             if cycle % 3 == 0:
                 self._schedule_orphan_reconciliation()
 
-    async def _scheduled_terminal_write_confirmed(self, record: RunRecord) -> bool:
+    @staticmethod
+    def _awaits_terminal_commit(record: RunRecord) -> bool:
+        """Whether a run keeps its active durable row until a deferred terminal commit."""
+        return record.scheduled_goal_cleanup_pending or record.terminal_commit_pending
+
+    async def _terminal_write_confirmed(self, record: RunRecord) -> bool:
         """Recognize this worker's acknowledged outcome at the commit boundary.
 
         A peer takeover may preserve owner_worker_id, so compare every terminal
         outcome field rather than accepting any terminal row with that owner.
         """
-        if not record.scheduled_goal_cleanup_pending or record.status in (RunStatus.pending, RunStatus.running) or self._store is None:
+        if not self._awaits_terminal_commit(record) or record.status in (RunStatus.pending, RunStatus.running) or self._store is None:
             return False
         row = await self._store.get(record.run_id, user_id=record.user_id)
         return (
@@ -2218,16 +2229,17 @@ class RunManager:
             and row.get("goal_verdict") == record.goal_verdict
         )
 
-    async def _release_confirmed_scheduled_barrier(self, record: RunRecord, *, timeout: float) -> bool:
-        if not record.scheduled_goal_cleanup_pending or record.status in (RunStatus.pending, RunStatus.running):
+    async def _release_confirmed_terminal_barrier(self, record: RunRecord, *, timeout: float) -> bool:
+        if not self._awaits_terminal_commit(record) or record.status in (RunStatus.pending, RunStatus.running):
             return False
         try:
             async with asyncio.timeout(timeout):
-                if await self._scheduled_terminal_write_confirmed(record):
+                if await self._terminal_write_confirmed(record):
                     record.scheduled_goal_cleanup_pending = False
+                    record.terminal_commit_pending = False
                     return True
         except Exception:
-            logger.warning("Unable to verify scheduled terminal commit for run %s", record.run_id, exc_info=True)
+            logger.warning("Unable to verify terminal commit for run %s", record.run_id, exc_info=True)
         return False
 
     async def _renew_leases(self) -> None:
@@ -2254,30 +2266,32 @@ class RunManager:
             # saturation, slow checkpoint hydrate on a fresh worker), peer
             # reconciliation will reclaim the run as an orphan and mark it
             # ``error`` even though this worker still intends to execute it.
+            # The same holds for a locally terminal run whose worker is still
+            # finalizing before its deferred terminal commit.
             active_runs = [
                 (rid, record)
                 for rid, record in self._runs.items()
-                if (record.status in (RunStatus.pending, RunStatus.running) or record.scheduled_goal_cleanup_pending) and record.owner_worker_id == self._worker_id and (record.task is None or not record.task.done())
+                if (record.status in (RunStatus.pending, RunStatus.running) or self._awaits_terminal_commit(record)) and record.owner_worker_id == self._worker_id and (record.task is None or not record.task.done())
             ]
 
         for run_id, record in active_runs:
-            if record.status not in (RunStatus.pending, RunStatus.running) and not record.scheduled_goal_cleanup_pending:
+            if record.status not in (RunStatus.pending, RunStatus.running) and not self._awaits_terminal_commit(record):
                 continue
             confirmed_deadline = self._parse_lease_deadline(record.lease_expires_at)
-            if record.scheduled_goal_cleanup_pending and record.status not in (RunStatus.pending, RunStatus.running):
+            if self._awaits_terminal_commit(record) and record.status not in (RunStatus.pending, RunStatus.running):
                 # The DB may have committed the terminal CAS while the worker
                 # is still receiving its acknowledgement. It no longer needs
                 # renewal and must not be mistaken for a peer takeover.
                 remaining = (confirmed_deadline - datetime.now(UTC)).total_seconds() if confirmed_deadline is not None else 0
-                if await self._release_confirmed_scheduled_barrier(record, timeout=remaining if remaining > 0 else lease_seconds):
+                if await self._release_confirmed_terminal_barrier(record, timeout=remaining if remaining > 0 else lease_seconds):
                     continue
             if confirmed_deadline is None or confirmed_deadline <= datetime.now(UTC):
-                if await self._release_confirmed_scheduled_barrier(record, timeout=lease_seconds):
+                if await self._release_confirmed_terminal_barrier(record, timeout=lease_seconds):
                     continue
                 await self._mark_ownership_lost(
                     record,
                     reason="Lease ownership could not be confirmed before the last confirmed lease expired.",
-                    require_active=not record.scheduled_goal_cleanup_pending,
+                    require_active=not self._awaits_terminal_commit(record),
                 )
                 continue
 
@@ -2296,12 +2310,12 @@ class RunManager:
                     )
                 if renewal.renewed:
                     if confirmed_deadline <= datetime.now(UTC):
-                        if await self._release_confirmed_scheduled_barrier(record, timeout=lease_seconds):
+                        if await self._release_confirmed_terminal_barrier(record, timeout=lease_seconds):
                             continue
                         await self._mark_ownership_lost(
                             record,
                             reason="Lease renewal completed after the last confirmed lease had already expired.",
-                            require_active=not record.scheduled_goal_cleanup_pending,
+                            require_active=not self._awaits_terminal_commit(record),
                         )
                         continue
                     # Unsynced write is benign: ``lease_expires_at`` is the
@@ -2330,7 +2344,7 @@ class RunManager:
                     async with self._lock:
                         still_active = (
                             self._runs.get(run_id) is record
-                            and (record.status in (RunStatus.pending, RunStatus.running) or record.scheduled_goal_cleanup_pending)
+                            and (record.status in (RunStatus.pending, RunStatus.running) or self._awaits_terminal_commit(record))
                             and record.owner_worker_id == self._worker_id
                             and (record.task is None or not record.task.done())
                         )
@@ -2338,7 +2352,7 @@ class RunManager:
                         # A detached precheck may have observed running just
                         # before our terminal CAS committed. Re-read before
                         # treating the rejected renewal as lost ownership.
-                        if await self._release_confirmed_scheduled_barrier(record, timeout=lease_seconds):
+                        if await self._release_confirmed_terminal_barrier(record, timeout=lease_seconds):
                             continue
                         logger.warning(
                             "Run %s lease renewal failed (status=%s,owner=%s) – worker likely taken over; aborting local task",
@@ -2349,16 +2363,16 @@ class RunManager:
                         await self._mark_ownership_lost(
                             record,
                             reason="The durable store rejected lease renewal for this worker.",
-                            require_active=not record.scheduled_goal_cleanup_pending,
+                            require_active=not self._awaits_terminal_commit(record),
                         )
             except Exception:
                 if confirmed_deadline <= datetime.now(UTC):
-                    if await self._release_confirmed_scheduled_barrier(record, timeout=lease_seconds):
+                    if await self._release_confirmed_terminal_barrier(record, timeout=lease_seconds):
                         continue
                     await self._mark_ownership_lost(
                         record,
                         reason="Lease ownership could not be confirmed before the last confirmed lease expired.",
-                        require_active=not record.scheduled_goal_cleanup_pending,
+                        require_active=not self._awaits_terminal_commit(record),
                     )
                 else:
                     logger.warning(

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

@@ -913,6 +913,9 @@ async def run_agent(
     run_events_config = ctx.run_events_config
     thread_store = ctx.thread_store
     terminal_status_kwargs = {"persist": False} if event_store is not None else {}
+    # Staged terminal statuses commit only after receipt and duration writes;
+    # keep the still-active durable row's lease renewed until then.
+    record.terminal_commit_pending = event_store is not None
 
     run_id = record.run_id
     thread_id = record.thread_id
@@ -1773,6 +1776,9 @@ async def run_agent(
                             await run_manager.persist_current_status(run_id)
                 except Exception:
                     logger.warning("Failed to persist terminal status for run %s after delivery receipt attempts", run_id, exc_info=True)
+            # The deferred commit has been attempted. A failed write is left to
+            # completion fallback or lease recovery, as before.
+            record.terminal_commit_pending = False
 
             if not record.ownership_lost and journal is not None and persist_completion:
                 try:

backend/tests/test_multi_worker_run_ownership.py

@@ -8,6 +8,7 @@ Coverage:
 - periodic reconciliation notifies Gateway recovery orchestration
 - Worker reconciliation skips runs with unexpired leases
 - Lease heartbeat renews active run leases
+- Lease heartbeat keeps renewing a staged terminal run until its commit
 - GATEWAY_WORKERS=1 + heartbeat_enabled=false behaviour unchanged
 """
 
@@ -15,14 +16,17 @@ from __future__ import annotations
 
 import asyncio
 from datetime import UTC, datetime, timedelta
+from types import SimpleNamespace
 from unittest.mock import AsyncMock
 
 import pytest
 
 from deerflow.config.run_ownership_config import RunOwnershipConfig
 from deerflow.runtime import ORPHAN_RECOVERY_STOP_REASON, RunManager, RunStatus, ThreadOperationKind
+from deerflow.runtime.events.store.memory import MemoryRunEventStore
 from deerflow.runtime.runs.manager import CancelOutcome, ConflictError, _generate_worker_id
 from deerflow.runtime.runs.store.memory import MemoryRunStore
+from deerflow.runtime.runs.worker import RunContext, run_agent
 
 # ---------------------------------------------------------------------------
 # Helpers
@@ -1114,6 +1118,139 @@ async def test_heartbeat_skips_runs_not_owned_by_this_worker():
     assert stored["lease_expires_at"] == old_lease
 
 
+@pytest.mark.anyio
+async def test_staged_terminal_run_keeps_lease_through_worker_finalization():
+    """A worker finalizing a staged success must keep its lease renewed.
+
+    With an event store the worker stages the terminal status locally and
+    commits it only after its receipt and duration writes. The durable row
+    stays active meanwhile, so an unrenewed lease would let a peer reclaim
+    the successful run as an orphan ``error``.
+    """
+    config = _lease_config(lease_seconds=30, grace_seconds=0, heartbeat_enabled=True)
+    store = MemoryRunStore()
+    owner = _make_manager(store=store, worker_id="worker-a", run_ownership_config=config)
+    peer = _make_manager(store=store, worker_id="worker-b", run_ownership_config=config)
+    record = await owner.create_or_reject("thread-finalize")
+    bridge = SimpleNamespace(publish=AsyncMock(), publish_end=AsyncMock(), cleanup=AsyncMock())
+    finalizing = asyncio.Event()
+    release = asyncio.Event()
+
+    class SlowReceiptEventStore(MemoryRunEventStore):
+        async def put_if_absent(self, **event):
+            if event["event_type"] == "run.delivery":
+                finalizing.set()
+                await release.wait()
+            return await super().put_if_absent(**event)
+
+    class FinishingAgent:
+        async def astream(self, graph_input, config=None, stream_mode=None, subgraphs=False):
+            del graph_input, config, stream_mode, subgraphs
+            yield {"messages": []}
+
+    task = asyncio.create_task(
+        run_agent(
+            bridge,
+            owner,
+            record,
+            ctx=RunContext(checkpointer=None, event_store=SlowReceiptEventStore()),
+            agent_factory=lambda **_kwargs: FinishingAgent(),
+            graph_input={},
+            config={},
+        )
+    )
+    record.task = task
+
+    try:
+        await asyncio.wait_for(finalizing.wait(), timeout=1)
+        assert record.status == RunStatus.success
+        assert (await store.get(record.run_id))["status"] == "running"
+
+        # Shrink the confirmed lease so it lapses during the test; only a
+        # heartbeat renewal can keep the peer from claiming the row.
+        near_expiry = (datetime.now(UTC) + timedelta(milliseconds=50)).isoformat()
+        record.lease_expires_at = near_expiry
+        store._runs[record.run_id]["lease_expires_at"] = near_expiry
+        await owner._renew_leases()
+        await asyncio.sleep(0.1)
+
+        assert await peer.reconcile_orphaned_inflight_runs(error="orphaned") == []
+    finally:
+        release.set()
+        await asyncio.wait_for(task, timeout=1)
+
+    assert record.ownership_lost is False
+    assert record.terminal_commit_pending is False
+    assert (await store.get(record.run_id))["status"] == "success"
+
+
+@pytest.mark.anyio
+async def test_renewal_rejected_by_own_terminal_commit_releases_barrier():
+    """A renewal racing this worker's terminal commit must not fence the run."""
+    config = _lease_config(lease_seconds=30, heartbeat_enabled=True)
+    store = MemoryRunStore()
+    manager = _make_manager(store=store, worker_id="worker-a", run_ownership_config=config)
+    record = await manager.create_or_reject("thread-1")
+    record.task = asyncio.create_task(asyncio.sleep(3600))
+    record.terminal_commit_pending = True
+    await manager.set_status(record.run_id, RunStatus.success, persist=False)
+    # The terminal write lands before the worker clears its barrier.
+    assert (await store.finalize_if_not_cancelled(record.run_id, status="success")).finalized
+
+    try:
+        await manager._renew_leases()
+
+        assert record.ownership_lost is False
+        assert record.terminal_commit_pending is False
+        assert record.status == RunStatus.success
+    finally:
+        record.task.cancel()
+        with pytest.raises(asyncio.CancelledError):
+            await record.task
+
+
+@pytest.mark.anyio
+async def test_renewal_rejected_by_peer_fences_staged_terminal_run():
+    """A peer claim during finalization must fence the staged outcome."""
+    config = _lease_config(lease_seconds=30, heartbeat_enabled=True)
+    store = MemoryRunStore()
+    manager = _make_manager(store=store, worker_id="worker-a", run_ownership_config=config)
+    record = await manager.create_or_reject("thread-1")
+    record.task = asyncio.create_task(asyncio.sleep(3600))
+    record.terminal_commit_pending = True
+    await manager.set_status(record.run_id, RunStatus.success, persist=False)
+    store._runs[record.run_id]["owner_worker_id"] = "worker-b"
+
+    await manager._renew_leases()
+    await asyncio.sleep(0)
+
+    assert record.ownership_lost is True
+    assert record.task.cancelled()
+
+
+@pytest.mark.anyio
+async def test_fenced_staged_terminal_run_stops_renewing():
+    """Fencing ends the terminal-commit barrier, so its lease can lapse."""
+    config = _lease_config(lease_seconds=30, heartbeat_enabled=True)
+    store = MemoryRunStore()
+    manager = _make_manager(store=store, worker_id="worker-a", run_ownership_config=config)
+    record = await manager.create_or_reject("thread-1")
+    record.task = asyncio.create_task(asyncio.sleep(3600))
+    record.terminal_commit_pending = True
+    await manager.set_status(record.run_id, RunStatus.success, persist=False)
+    await manager._mark_ownership_lost(record, reason="Test fence", require_active=False)
+    store.update_lease = AsyncMock(wraps=store.update_lease)
+
+    try:
+        await manager._renew_leases()
+
+        store.update_lease.assert_not_awaited()
+        assert record.terminal_commit_pending is False
+    finally:
+        with pytest.raises(asyncio.CancelledError):
+            await record.task
+
+
 @pytest.mark.anyio
 async def test_heartbeat_not_started_when_disabled():
     """When heartbeat_enabled is False, start_heartbeat must be a no-op."""