diff --git a/backend/app/gateway/routers/threads.py b/backend/app/gateway/routers/threads.py index 529281352b..d75642471e 100644 --- a/backend/app/gateway/routers/threads.py +++ b/backend/app/gateway/routers/threads.py @@ -810,6 +810,7 @@ async def _delete_thread_data_with_reservation(thread_id: str, request: Request) # persisted runs/events/feedback and the thread_meta row all belong to the # same owner, so they must not resolve their scope independently. user_id = get_effective_user_id() + run_manager = get_run_manager(request) batch_repo = getattr(request.app.state, "subagent_batch_repo", None) if batch_repo is not None: await batch_repo.suppress_thread_continuations(thread_id) @@ -852,6 +853,16 @@ async def _delete_thread_data_with_reservation(thread_id: str, request: Request) except Exception: logger.debug("Could not delete run records for thread %s (not critical)", sanitize_log_param(thread_id)) + # Drop this worker's in-memory copies too. ``RunManager.get()`` and + # ``list_by_thread()`` answer from the process-local registry before the + # store, so without this the deleted thread's runs — including ``kwargs`` + # with the user's prompt — stay readable through ``GET /threads/{id}/runs`` + # until the Gateway restarts, while the thread itself already 404s. + try: + await run_manager.discard_thread_runs(thread_id) + except Exception: + logger.debug("Could not evict in-memory runs for thread %s (not critical)", sanitize_log_param(thread_id)) + # Remove persisted run events (best-effort). These are the user-visible # conversation history, not a cache: leaving them behind makes a deleted # thread's feed readable again through GET /threads/{id}/messages. A legacy diff --git a/backend/packages/harness/deerflow/runtime/runs/manager.py b/backend/packages/harness/deerflow/runtime/runs/manager.py index d6cd7a1573..356abf861d 100644 --- a/backend/packages/harness/deerflow/runtime/runs/manager.py +++ b/backend/packages/harness/deerflow/runtime/runs/manager.py @@ -343,6 +343,12 @@ class RunRecord: idempotency_reused: bool = False evidence_origin: str = "unknown" evidence_agent_id: str | None = None + # Set by a thread deletion that arrived while this record's worker task was + # still running its finalize tail. The record stays registered so that tail + # keeps a target for its manager-mediated writes, but every public read + # treats it as gone and the task's completion evicts it immediately instead + # of waiting for ``cleanup()``'s delay. + evicted: bool = field(default=False, repr=False) @dataclass(frozen=True) @@ -422,6 +428,9 @@ class RunManager: self._lease_checks: dict[str, asyncio.Task[None]] = {} self._lease_operations: dict[str, asyncio.Task] = {} self._lease_deadlines: dict[str, asyncio.TimerHandle] = {} + # Evictions queued by a deleted thread's finishing finalize tail. Held so + # the loop cannot garbage-collect them before they drop the record. + self._tail_evictions: set[asyncio.Task] = set() @asynccontextmanager async def _admission_scope(self, thread_id: str, run_id: str) -> AsyncIterator[None]: @@ -1176,8 +1185,13 @@ class RunManager: """ async with self._run_registry_lock(run_id): record = self._runs.get(run_id) - if record is not None: + if record is not None and not record.evicted: return record + # A hidden record is authoritative for "this run is deleted", so never + # resurrect it from a stale store read. Settle that before any store call + # so the read stays side-effect free and store-less managers stop here. + if record is not None and record.evicted: + return None if self._store is None: return None try: @@ -1191,9 +1205,9 @@ class RunManager: # in-memory record while the store call was in flight. async with self._run_registry_lock(run_id): record = self._runs.get(run_id) - if record is not None: + if record is not None and not record.evicted: return record - if row is None: + if record is not None or row is None: return None try: return self._record_from_store(row) @@ -1263,6 +1277,7 @@ class RunManager: record for record in self._thread_records_locked(thread_id) if record.operation_kind == ThreadOperationKind.run + and not record.evicted and run_is_before_cursor( record.created_at, record.run_id, @@ -1290,7 +1305,15 @@ class RunManager: return _page(memory_records) for row in rows: run_id = row.get("run_id") - if run_id and run_id not in records_by_id: + if not run_id: + continue + known = self._runs.get(run_id) + # A hidden record means this thread was deleted while the record + # still had a live finalize tail; the durable row is already gone, + # so a store row here can only be a stale or partial read. + if known is not None and known.evicted: + continue + if run_id not in records_by_id: try: records_by_id[run_id] = self._record_from_store(row) except Exception: @@ -1313,7 +1336,7 @@ class RunManager: """ resolved_user_id = resolve_user_id(user_id, method_name="RunManager.list_successful_regenerate_sources") async with self._thread_registry_lock(thread_id): - memory_records = [record for record in self._thread_records_locked(thread_id) if record.operation_kind == ThreadOperationKind.run and (resolved_user_id is None or record.user_id == resolved_user_id)] + memory_records = [record for record in self._thread_records_locked(thread_id) if record.operation_kind == ThreadOperationKind.run and not record.evicted and (resolved_user_id is None or record.user_id == resolved_user_id)] sources = set(await self._store.list_successful_regenerate_sources(thread_id, user_id=resolved_user_id)) if self._store is not None else set() # _thread_records_locked preserves the insertion order of the thread @@ -1384,7 +1407,7 @@ class RunManager: records_by_id[record.run_id] = record async with self._thread_registry_lock(thread_id): - memory_records = [record for record in self._thread_records_locked(thread_id) if resolved_user_id is None or record.user_id == resolved_user_id] + memory_records = [record for record in self._thread_records_locked(thread_id) if not record.evicted and (resolved_user_id is None or record.user_id == resolved_user_id)] for record in memory_records: records_by_id[record.run_id] = record @@ -1475,12 +1498,22 @@ class RunManager: resolved_user_id = resolve_user_id(user_id, method_name="RunManager.get_many_by_thread") async with self._thread_registry_lock(thread_id): records_by_id = { - record.run_id: record for record in self._thread_records_locked(thread_id) if record.operation_kind == ThreadOperationKind.run and record.run_id in run_ids and (resolved_user_id is None or record.user_id == resolved_user_id) + record.run_id: record + for record in self._thread_records_locked(thread_id) + if record.operation_kind == ThreadOperationKind.run and not record.evicted and record.run_id in run_ids and (resolved_user_id is None or record.user_id == resolved_user_id) } if self._store is None: return records_by_id remaining = run_ids - records_by_id.keys() + if not remaining: + return records_by_id + # Mirror ``list_by_thread``: a hidden record means the thread was deleted + # while its finalize tail was live, so a store row for it can only be a + # stale or partial read and must never be rebuilt into a visible record. + async with self._thread_registry_lock(thread_id): + hidden_run_ids = {record.run_id for record in self._thread_records_locked(thread_id) if record.operation_kind == ThreadOperationKind.run and record.evicted} + remaining -= hidden_run_ids if not remaining: return records_by_id try: @@ -2796,6 +2829,94 @@ class RunManager: async with self._thread_registry_lock(thread_id): 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 discard_thread_runs(self, thread_id: str) -> int: + """Drop this worker's in-memory run records for *thread_id*. + + A thread deletion removes the durable rows through + ``RunStore.delete_by_thread``, but ``get()`` and ``list_by_thread()`` + answer from ``_runs`` / ``_runs_by_thread`` first. Leaving those copies + in place makes a deleted thread's runs — including ``kwargs`` with the + user's prompt — readable until the process restarts, even though the + thread itself already answers 404. + + The caller must already hold the thread's exclusive admission, so no + run for it can be admitted concurrently here. A finished record can + still own a live finalize tail (``worker.py`` clears ``finalizing``, + then publishes the terminal frame, closes the journal and releases the + sandbox lease before scheduling ``cleanup()``), so records whose + ``task`` is still running stay indexed here: the bound is one final + tail, and later manager-mediated writes land on a record whose durable + rows are already gone. + + This covers this worker only. A Gateway peer that admitted the same + thread keeps its own ``_runs`` copies until its deferred + ``cleanup(run_id)`` fires (the default 300s delay), so a DELETE served + elsewhere bounds multi-worker residues rather than removing them. + """ + async with self._thread_registry_lock(thread_id): + run_ids = list(self._runs_by_thread.get(thread_id, {})) + dropped = 0 + deferred: list[asyncio.Task] = [] + for run_id in run_ids: + record = self._runs.get(run_id) + if record is None or record.thread_id != thread_id: + self._unindex_run_locked(run_id, thread_id) + continue + # Historical runs only. Internal thread-operation rows are held + # by their own release path (this deletion holds one of them). + if record.operation_kind != ThreadOperationKind.run: + continue + task = record.task + if task is not None and not task.done(): + # A finalize tail still owns this record. Hide it from every + # public read now, and let the task's own completion evict + # it — otherwise the run and its prompt stay readable + # through the run endpoints until ``cleanup()``'s delay. + record.evicted = True + deferred.append(task) + continue + self._runs.pop(run_id, None) + self._unindex_run_locked(run_id, thread_id) + dropped += 1 + for task in deferred: + task.add_done_callback(lambda completed: self._evict_after_task(thread_id, completed)) + return dropped + + def _evict_after_task(self, thread_id: str, completed: asyncio.Task) -> None: + """Evict a record shielded only by *completed*, once its tail is done. + + ``discard_thread_runs`` flags a record instead of popping it so the + finalize tail keeps a write target. The tail is the record's ``task``, + so its completion is the earliest point at which popping is safe; + ``cleanup()``'s 300s delay is far too late for a deleted thread. + + Eviction runs as its own task because this callback cannot await the + registry lock. The loop holds only weak references to tasks, so the + eviction is tracked here until it settles. + """ + try: + task = asyncio.get_running_loop().create_task(self._evict_completed_tail(thread_id)) + except RuntimeError: + # Fail-closed: the record stays hidden from reads because it is + # still flagged, and the next discard pass evicts it outright. + logger.debug("No running loop to evict the thread %s finalize tail", thread_id, exc_info=True) + return + self._tail_evictions.add(task) + task.add_done_callback(self._tail_evictions.discard) + + async def _evict_completed_tail(self, thread_id: str) -> None: + """Drop records a deletion marked ``evicted`` once their tail is done.""" + async with self._thread_registry_lock(thread_id): + for run_id, record in list(self._runs.items()): + if record.thread_id != thread_id or not record.evicted: + continue + if record.operation_kind != ThreadOperationKind.run: + continue + if record.task is not None and not record.task.done(): + continue + self._runs.pop(run_id, None) + self._unindex_run_locked(run_id, thread_id) + async def cleanup(self, run_id: str, *, delay: float = 300) -> None: """Remove a run record after an optional delay. diff --git a/backend/tests/test_run_manager.py b/backend/tests/test_run_manager.py index f3f854d531..881ee1693e 100644 --- a/backend/tests/test_run_manager.py +++ b/backend/tests/test_run_manager.py @@ -724,6 +724,181 @@ async def test_cleanup_without_store_preserves_history(manager: RunManager): assert [r.run_id for r in await manager.list_by_thread("thread-1")] == [run_id] +@pytest.mark.anyio +async def test_discard_thread_runs_keeps_internal_operation_rows(manager_with_store: RunManager): + """Only the thread's historical runs go; the release path owns the rest. + + The deletion that calls this holds a ``delete`` reservation, and a + checkpoint write or a branch operation may be indexed for the same thread. + Inverting the ``operation_kind`` guard must fail here rather than being + absorbed by a later release: the reservation's own ``finally`` would prune + its bucket anyway, which is why the router-level tests cannot pin it. + """ + mgr = manager_with_store + thread_id = "thread-1" + finished = await mgr.create(thread_id, kwargs={"input": "keep me out of memory"}) + await mgr.set_status(finished.run_id, RunStatus.success) + + async with mgr.reserve_thread_operation(thread_id, kind=ThreadOperationKind.delete): + dropped = await mgr.discard_thread_runs(thread_id) + + assert dropped == 1 + assert finished.run_id not in mgr._runs + assert finished.run_id not in mgr._runs_by_thread.get(thread_id, {}) + internal = [record for record in mgr._runs.values() if record.thread_id == thread_id] + assert [record.operation_kind for record in internal] == [ThreadOperationKind.delete] + # The reservation still protects the request that is running. + assert [record.run_id for record in internal] == [next(iter(mgr._runs_by_thread[thread_id]))] + + # Its own release path removes it once the reservation exits. + assert thread_id not in mgr._runs_by_thread + # The run's history survives through the store fallback. + hydrated = await mgr.get(finished.run_id, user_id=finished.user_id) + assert hydrated is not None + assert hydrated.status is RunStatus.success + + +@pytest.mark.anyio +async def test_discard_thread_runs_leaves_a_live_finalize_tail_attached(manager: RunManager): + """A finished record whose task is mid-tail is not severed. + + ``worker.py`` clears ``finalizing`` before publishing the terminal frame and + releasing the sandbox lease, so popping here would turn the tail's + manager-mediated writes into ``set_status called for unknown run`` warnings. + """ + record = await manager.create("thread-1") + await manager.set_status(record.run_id, RunStatus.success) + + async def tail() -> None: + await asyncio.sleep(0) + + tail_task = asyncio.create_task(tail()) + record.task = tail_task + try: + assert await manager.discard_thread_runs("thread-1") == 0 + assert manager._runs.get(record.run_id) is record + finally: + await tail_task + + # Once the tail finishes, the next pass drops it. + assert await manager.discard_thread_runs("thread-1") == 1 + assert record.run_id not in manager._runs + + +@pytest.mark.anyio +async def test_discard_thread_runs_hides_a_live_finalize_tail_from_reads(manager_with_store: RunManager): + """A delete during the finalize tail must not leave the run readable. + + ``discard_thread_runs`` cannot pop the record while ``worker.py``'s tail is + still running, so the record stays registered as its write target. Every + public read must treat it as deleted anyway: the durable row is gone and the + user only sees a 404 for the thread they just deleted. + """ + mgr = manager_with_store + record = await mgr.create("thread-1", kwargs={"input": "the user's prompt"}) + await mgr.set_status(record.run_id, RunStatus.success) + + async def tail() -> None: + await asyncio.sleep(0.05) + + tail_task = asyncio.create_task(tail()) + record.task = tail_task + # The router deletes the durable rows before evicting this worker's copies. + await mgr._store.delete_by_thread("thread-1") + try: + assert await mgr.discard_thread_runs("thread-1") == 0 + + assert await mgr.get(record.run_id, user_id=record.user_id) is None + assert await mgr.list_by_thread("thread-1", user_id=record.user_id) == [] + assert await mgr.get_many_by_thread("thread-1", {record.run_id}) == {} + finally: + await tail_task + + # The tail's completion evicts it; no second delete pass and no 300s wait. + await asyncio.wait_for(_wait_for_eviction(mgr, record.run_id), timeout=1) + assert record.run_id not in mgr._runs + assert "thread-1" not in mgr._runs_by_thread + + +async def _wait_for_eviction(manager: RunManager, run_id: str) -> None: + while run_id in manager._runs: + await asyncio.sleep(0.01) + + +@pytest.mark.anyio +async def test_hidden_record_read_stays_quiet_without_a_store(manager: RunManager, caplog: pytest.LogCaptureFixture): + """A store-less manager must not reach for a store it does not have. + + The hidden record is authoritative for "this run is deleted", so ``get`` + settles on it before any store call. Narrowing that to a missing record let + a store-less read fall through to ``None.get`` and log a hydration warning + with a traceback, misattributing a local invariant to a store failure. + """ + record = await manager.create("thread-1") + await manager.set_status(record.run_id, RunStatus.success) + + async def tail() -> None: + await asyncio.sleep(0.05) + + record.task = asyncio.create_task(tail()) + try: + assert await manager.discard_thread_runs("thread-1") == 0 + assert record.evicted + + with caplog.at_level(logging.WARNING): + assert await manager.get(record.run_id) is None + assert await manager.aget(record.run_id) is None + assert [r.message for r in caplog.records if "Failed to hydrate" in r.message] == [] + finally: + await record.task + + +@pytest.mark.anyio +async def test_get_many_by_thread_never_rebuilds_a_hidden_record(manager_with_store: RunManager): + """Batch reads skip evicted runs even when the store still has a row. + + ``list_by_thread`` already refuses to rebuild a hidden record from a store + row, because the durable row is deleted before the flag is set and anything + left is a stale or partial read. The batch reader must keep that invariant + uniform, or a stale row resurrects a run the user just deleted. + """ + mgr = manager_with_store + record = await mgr.create("thread-1", kwargs={"input": "the user's prompt"}) + await mgr.set_status(record.run_id, RunStatus.success) + + async def tail() -> None: + await asyncio.sleep(0.05) + + record.task = asyncio.create_task(tail()) + try: + assert await mgr.discard_thread_runs("thread-1") == 0 + # A stale row surviving the delete must not satisfy the batch read. + await mgr._store.put( + record.run_id, + thread_id="thread-1", + assistant_id=record.assistant_id, + status="success", + multitask_strategy=record.multitask_strategy, + metadata={}, + user_id=record.user_id, + ) + assert await mgr.get_many_by_thread("thread-1", {record.run_id}) == {} + finally: + await record.task + + +@pytest.mark.anyio +async def test_discard_thread_runs_ignores_other_threads(manager: RunManager): + mine = await manager.create("thread-1") + theirs = await manager.create("thread-2") + + assert await manager.discard_thread_runs("thread-1") == 1 + + assert mine.run_id not in manager._runs + assert manager._runs.get(theirs.run_id) is theirs + assert [r.run_id for r in await manager.list_by_thread("thread-2")] == [theirs.run_id] + + @pytest.mark.anyio async def test_set_status_with_error(manager: RunManager): """Error message should be stored on the record.""" diff --git a/backend/tests/test_threads_router.py b/backend/tests/test_threads_router.py index c641827925..fc816ec656 100644 --- a/backend/tests/test_threads_router.py +++ b/backend/tests/test_threads_router.py @@ -589,6 +589,102 @@ def _persistence_cleanup_app(tmp_path, *, run_store, event_store, feedback_repo) return app +def test_delete_thread_evicts_this_workers_in_memory_run_registry(tmp_path): + """Deleting a thread must drop the runs this worker still holds in memory. + + ``RunStore.delete_by_thread`` removes the durable rows, but ``RunManager`` + keeps its own process-local copies and answers ``get()`` / ``list_by_thread()`` + from them before consulting the store. Without an eviction the deleted + thread's runs — including ``kwargs`` with the user's prompt — stay readable + through ``GET /api/threads/{id}/runs`` until the Gateway restarts, while the + thread itself already answers 404. + """ + from deerflow.runtime import RunManager, RunStatus + from deerflow.runtime.runs.store.memory import MemoryRunStore + + paths = Paths(tmp_path) + run_store = MemoryRunStore() + run_manager = RunManager(store=run_store) + + async def _seed() -> str: + record = await run_manager.create_or_reject("thread-runs-gone") + await run_manager.set_status(record.run_id, RunStatus.success) + return record.run_id + + run_id = asyncio.run(_seed()) + assert asyncio.run(run_manager.get(run_id)) is not None + + event_store = MagicMock() + event_store.delete_by_thread = AsyncMock(return_value=1) + feedback_repo = MagicMock() + feedback_repo.delete_by_thread = AsyncMock(return_value=0) + + app = make_authed_test_app() + app.state.run_manager = run_manager + app.state.run_store = run_store + app.state.run_event_store = event_store + app.state.feedback_repo = feedback_repo + app.include_router(threads.router) + + with patch("app.gateway.routers.threads.get_paths", return_value=paths): + with TestClient(app) as client: + response = client.delete("/api/threads/thread-runs-gone") + + assert response.status_code == 200, response.text + # The durable row is gone, so the in-memory copy is the only remaining + # answer — and it must be gone too. + assert asyncio.run(run_store.list_by_thread("thread-runs-gone", user_id=None)) == [] + assert asyncio.run(run_manager.get(run_id)) is None + assert asyncio.run(run_manager.list_by_thread("thread-runs-gone", user_id=None)) == [] + assert "thread-runs-gone" not in run_manager._runs_by_thread + + +def test_deleted_thread_runs_are_no_longer_readable_over_http(tmp_path): + """The reported sequence: delete 200 → runs list ``[]`` → single run 404.""" + from deerflow.runtime import RunManager, RunStatus + from deerflow.runtime.runs.store.memory import MemoryRunStore + + paths = Paths(tmp_path) + run_store = MemoryRunStore() + run_manager = RunManager(store=run_store) + + async def _seed() -> str: + record = await run_manager.create_or_reject("thread-http-gone", kwargs={"input": {"messages": [{"role": "user", "content": "probe-marker-1"}]}}) + await run_manager.set_status(record.run_id, RunStatus.success) + return record.run_id + + run_id = asyncio.run(_seed()) + + event_store = MagicMock() + event_store.delete_by_thread = AsyncMock(return_value=1) + feedback_repo = MagicMock() + feedback_repo.delete_by_thread = AsyncMock(return_value=0) + + app = make_authed_test_app() + app.state.run_manager = run_manager + app.state.run_store = run_store + app.state.run_event_store = event_store + app.state.feedback_repo = feedback_repo + app.include_router(threads.router) + app.include_router(thread_runs.router) + + with patch("app.gateway.routers.threads.get_paths", return_value=paths): + with TestClient(app) as client: + before = client.get("/api/threads/thread-http-gone/runs") + assert before.status_code == 200, before.text + assert [row["run_id"] for row in before.json()] == [run_id] + + deleted = client.delete("/api/threads/thread-http-gone") + assert deleted.status_code == 200, deleted.text + + listing = client.get("/api/threads/thread-http-gone/runs") + single = client.get(f"/api/threads/thread-http-gone/runs/{run_id}") + + assert listing.status_code == 200, listing.text + assert listing.json() == [] + assert single.status_code == 404, single.text + + def test_delete_thread_route_cleans_persisted_records(tmp_path): """run_events, historical runs and feedback are cleaned under the reservation.""" from deerflow.runtime.user_context import get_effective_user_id