From cda56aa2825e9afa79c49933efed66a74139bac7 Mon Sep 17 00:00:00 2001 From: Fengmin Li <2080291162@qq.com> Date: Thu, 17 Sep 2026 07:38:38 +0800 Subject: [PATCH] fix(runtime): keep store-less run history through cleanup (#5453) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit * fix(runtime): keep store-less run history through cleanup run_agent schedules cleanup for every terminal run, but cleanup evicted the in-memory record unconditionally. With a durable RunStore that is safe: the store fallback in get()/list_by_thread() still serves the run. Without one there is no fallback, so an embedded consumer that builds RunManager() with its default store=None lost completed runs from history entirely. Gate eviction on a backing store, matching the retain-forever behaviour documented for memory-only mode, and cover it: the two index/cleanup tests now use a store, plus a regression asserting a store-less cleanup keeps the record readable. * test(runtime): terminalize the store-backed cleanup record first Mirror the production sequence — run_agent only schedules cleanup once the run is terminal and its store row is finalized — and pin that a terminal store row hydrates with its status intact. --- .../harness/deerflow/runtime/AGENTS.md | 2 +- .../harness/deerflow/runtime/runs/manager.py | 11 +++- backend/tests/test_run_manager.py | 54 ++++++++++++++----- 3 files changed, 53 insertions(+), 14 deletions(-) diff --git a/backend/packages/harness/deerflow/runtime/AGENTS.md b/backend/packages/harness/deerflow/runtime/AGENTS.md index c74a5a037..06e3db651 100644 --- a/backend/packages/harness/deerflow/runtime/AGENTS.md +++ b/backend/packages/harness/deerflow/runtime/AGENTS.md @@ -219,7 +219,7 @@ The first `RunManager.list_by_thread()` hydration page uses a 100-row floor or the number of required IDs, whichever is larger; missing exact runs use targeted `get()` calls. -**Terminal run cleanup explicitly breaks graph-scoped references while preserving the existing `RunRecord` grace period.** Every `agent.astream()` iterator is closed in `_stream_once`, including abort/exception/early-break paths. A close failure after an abort is warning-only and cannot replace the user-requested `interrupted` outcome; normal-completion close failures still surface, and an in-flight stream exception remains authoritative over a secondary close failure. Journal construction and cancellable preflight work (including MCP task projection and the prior-finalization wait) live inside the worker's guarded body, so cancellation before agent startup still terminalizes the run and closes its stream. `run_agent()` wraps the complete terminal-finalization sequence in an outer teardown guard, so cancellation or failure from any terminal-stage await cannot skip `RunJournal.close()`, removal of the journal, `__pregel_runtime`, and internal runtime-context values from every runnable config, or release of local graph/payload references. That guard schedules bridge cleanup, run-record cleanup, and cyclic GC even when interruption happens before the terminal stream marker or terminal publication itself fails, so neither a cancelled observer nor a delivery-backend outage can strand process-local run state. A non-`Exception` `BaseException` caught while awaiting the completion hook or task-stop notification (including host-task cancellation) is deferred through the ordinary remaining finalization, with the first interruption preserved and every caught host-task `CancelledError` balanced by calling `Task.uncancel()` until the current task’s cumulative cancellation count is clear. Task-stop fan-out runs in one child task and every host wait uses `shield`, so repeated cancellation of the worker cannot cancel that fan-out or skip later observers; the worker keeps awaiting the same child task. A rogue observer that raises its own `CancelledError` remains contained by the extension dispatcher and distinguishable from host cancellation. This guarantee applies only to cancellation caught during those hook stages: clearing the finalizing barrier and publishing END remain direct awaits, so another cancellation in the subsequent critical tail retains forceful-termination semantics instead of creating an unbounded shield. If that tail completes without another interruption, the first deferred interruption is re-raised after END; a barrier-clear failure prevents END publication, while an END failure is raised after the barrier is clear. `RunJournal.flush()` clears its `_pending_progress_task` after awaiting or cancelling it; ordinary `close()` detaches the event store/progress reporter and clears callback bookkeeping only after that flush succeeds, preserving the buffer for retry on a transient store failure. A fenced worker instead calls `close(flush=False)`, which cancels pending journal work and detaches without initiating another event-store write after lease ownership is lost; its final detach runs even if a second cancellation interrupts pending-task shutdown. `RunManager.cleanup(run_id)` retains the process-local `RunRecord`, completed task, and request payload for its default 300-second local join/status window before releasing them. Durable history remains in `RunStore`; `StreamBridge` data keeps its separate 60-second late-subscriber window, and both cleanup coroutines run in a fresh empty `contextvars.Context`. A contextless full cyclic-GC pass, coalesced to at most once every 10 seconds and dispatched through the default executor, bounds the lifetime of unreachable LangGraph callback/loop cycles without synchronously walking the heap in the event-loop timer; passes taking at least 100 ms are logged at INFO because CPython GC may still impose interpreter-level pauses. +**Terminal run cleanup explicitly breaks graph-scoped references while preserving the existing `RunRecord` grace period.** Every `agent.astream()` iterator is closed in `_stream_once`, including abort/exception/early-break paths. A close failure after an abort is warning-only and cannot replace the user-requested `interrupted` outcome; normal-completion close failures still surface, and an in-flight stream exception remains authoritative over a secondary close failure. Journal construction and cancellable preflight work (including MCP task projection and the prior-finalization wait) live inside the worker's guarded body, so cancellation before agent startup still terminalizes the run and closes its stream. `run_agent()` wraps the complete terminal-finalization sequence in an outer teardown guard, so cancellation or failure from any terminal-stage await cannot skip `RunJournal.close()`, removal of the journal, `__pregel_runtime`, and internal runtime-context values from every runnable config, or release of local graph/payload references. That guard schedules bridge cleanup, run-record cleanup, and cyclic GC even when interruption happens before the terminal stream marker or terminal publication itself fails, so neither a cancelled observer nor a delivery-backend outage can strand process-local run state. A non-`Exception` `BaseException` caught while awaiting the completion hook or task-stop notification (including host-task cancellation) is deferred through the ordinary remaining finalization, with the first interruption preserved and every caught host-task `CancelledError` balanced by calling `Task.uncancel()` until the current task’s cumulative cancellation count is clear. Task-stop fan-out runs in one child task and every host wait uses `shield`, so repeated cancellation of the worker cannot cancel that fan-out or skip later observers; the worker keeps awaiting the same child task. A rogue observer that raises its own `CancelledError` remains contained by the extension dispatcher and distinguishable from host cancellation. This guarantee applies only to cancellation caught during those hook stages: clearing the finalizing barrier and publishing END remain direct awaits, so another cancellation in the subsequent critical tail retains forceful-termination semantics instead of creating an unbounded shield. If that tail completes without another interruption, the first deferred interruption is re-raised after END; a barrier-clear failure prevents END publication, while an END failure is raised after the barrier is clear. `RunJournal.flush()` clears its `_pending_progress_task` after awaiting or cancelling it; ordinary `close()` detaches the event store/progress reporter and clears callback bookkeeping only after that flush succeeds, preserving the buffer for retry on a transient store failure. A fenced worker instead calls `close(flush=False)`, which cancels pending journal work and detaches without initiating another event-store write after lease ownership is lost; its final detach runs even if a second cancellation interrupts pending-task shutdown. `RunManager.cleanup(run_id)` retains the process-local `RunRecord`, completed task, and request payload for its default 300-second local join/status window before releasing them, and evicts only when a `RunStore` backs the manager — without one there is no hydration fallback, so the record is retained rather than silently dropping the run's history. Durable history remains in `RunStore`; `StreamBridge` data keeps its separate 60-second late-subscriber window, and both cleanup coroutines run in a fresh empty `contextvars.Context`. A contextless full cyclic-GC pass, coalesced to at most once every 10 seconds and dispatched through the default executor, bounds the lifetime of unreachable LangGraph callback/loop cycles without synchronously walking the heap in the event-loop timer; passes taking at least 100 ms are logged at INFO because CPython GC may still impose interpreter-level pauses. **`RunManager._runs` holds only records this worker admitted.** A cross-worker idempotent reuse returns the `store_only` row from `_record_from_store()` unregistered: the peer never finalizes or `cleanup()`s it, so a registered copy stays `pending`/`running`, 409s later same-thread admissions, hides the owner's orphan from reconciliation, and sends a peer `cancel()` down the local-owner path. Pinned by `test_peer_idempotent_reuse_*` and `test_peer_cancel_of_reused_run_*` (`tests/test_multi_worker_run_ownership.py`) plus `tests/test_gateway_services.py::test_start_run_peer_idempotent_reuse_*`. diff --git a/backend/packages/harness/deerflow/runtime/runs/manager.py b/backend/packages/harness/deerflow/runtime/runs/manager.py index b2b4a98aa..c3eddb0ac 100644 --- a/backend/packages/harness/deerflow/runtime/runs/manager.py +++ b/backend/packages/harness/deerflow/runtime/runs/manager.py @@ -1950,7 +1950,16 @@ class RunManager: return any(r.operation_kind == ThreadOperationKind.run and (r.status in (RunStatus.pending, RunStatus.running) or r.finalizing) for r in self._thread_records_locked(thread_id)) async def cleanup(self, run_id: str, *, delay: float = 300) -> None: - """Remove a run record after an optional delay.""" + """Remove a run record after an optional delay. + + Eviction is only safe when a ``RunStore`` backs this manager: history + then stays readable through the store fallback in ``get()`` / + ``list_by_thread()``. Without one, dropping the record would erase the + run's history entirely, so a store-less manager keeps the previous + retain-forever behaviour and this returns immediately. + """ + if self._store is None: + return if delay > 0: await asyncio.sleep(delay) async with self._lock: diff --git a/backend/tests/test_run_manager.py b/backend/tests/test_run_manager.py index ef042e971..afb0f4ddb 100644 --- a/backend/tests/test_run_manager.py +++ b/backend/tests/test_run_manager.py @@ -674,13 +674,40 @@ async def test_has_inflight_ignores_checkpoint_write_reservation(manager: RunMan @pytest.mark.anyio -async def test_cleanup(manager: RunManager): - """After cleanup, the run should be gone.""" +async def test_cleanup_evicts_with_store(manager_with_store: RunManager): + """With a store, cleanup releases the record and history stays readable.""" + mgr = manager_with_store + record = await mgr.create("thread-1") + run_id = record.run_id + # Mirrors the production sequence: run_agent only schedules cleanup once + # the run is terminal and its store row has been finalized. + await mgr.set_status(run_id, RunStatus.success) + + await mgr.cleanup(run_id, delay=0) + assert run_id not in mgr._runs + hydrated = await mgr.get(run_id, user_id=record.user_id) + assert hydrated is not None + assert hydrated.run_id == run_id + assert hydrated.status is RunStatus.success + + +@pytest.mark.anyio +async def test_cleanup_without_store_preserves_history(manager: RunManager): + """Without a store there is no fallback, so cleanup must not erase history. + + ``run_agent`` schedules cleanup for every terminal run. Evicting in + memory-only mode would drop the record from ``_runs`` with nothing left to + hydrate it from, making completed runs disappear from history instead of + being released from a durable copy. + """ record = await manager.create("thread-1") run_id = record.run_id + await manager.set_status(run_id, RunStatus.success) await manager.cleanup(run_id, delay=0) - assert await manager.get(run_id) is None + + assert await manager.get(run_id) is record + assert [r.run_id for r in await manager.list_by_thread("thread-1")] == [run_id] @pytest.mark.anyio @@ -1595,18 +1622,21 @@ async def test_thread_index_preserves_insertion_order(manager: RunManager): @pytest.mark.anyio -async def test_thread_index_cleanup_prunes_run_and_empty_bucket(manager: RunManager): - a1 = await manager.create("thread-a") - a2 = await manager.create("thread-a") +async def test_thread_index_cleanup_prunes_run_and_empty_bucket(manager_with_store: RunManager): + mgr = manager_with_store + a1 = await mgr.create("thread-a") + a2 = await mgr.create("thread-a") - await manager.cleanup(a1.run_id, delay=0) - assert a1.run_id not in manager._runs - assert set(manager._runs_by_thread["thread-a"]) == {a2.run_id} + await mgr.cleanup(a1.run_id, delay=0) + assert a1.run_id not in mgr._runs + assert set(mgr._runs_by_thread["thread-a"]) == {a2.run_id} - await manager.cleanup(a2.run_id, delay=0) + await mgr.cleanup(a2.run_id, delay=0) # Empty buckets are pruned so the index cannot grow without bound. - assert "thread-a" not in manager._runs_by_thread - assert await manager.list_by_thread("thread-a") == [] + assert "thread-a" not in mgr._runs_by_thread + # Both records survive as store-only history; the store does not promise + # to preserve the in-memory insertion order. + assert {r.run_id for r in await mgr.list_by_thread("thread-a")} == {a1.run_id, a2.run_id} @pytest.mark.anyio