From 4e09e63a68f55bf5203838598b399fd03b2e80bd Mon Sep 17 00:00:00 2001 From: 2SinceJuly <2929362966@qq.com> Date: Mon, 9 Nov 2026 20:21:07 +0800 Subject: [PATCH] fix(threads): evict in-memory runs when a thread is deleted (#8036) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit * fix(threads): evict in-memory runs when a thread is deleted RunManager.get() and list_by_thread() answer from the process-local _runs registry before consulting the RunStore. DELETE /api/threads/{id} only removed the durable rows, so a worker that had admitted the thread's runs kept serving them: after the delete returned 200 and the thread itself answered 404, GET /api/threads/{id}/runs still listed the run and GET /api/threads/{id}/runs/{run_id} still returned 200 with the original prompt intact in kwargs. The residue disappeared only on restart, which is what made it look like a cache rather than a leak. Add RunManager.discard_thread_runs() to drop this worker's copies for the thread and call it from the delete path, next to the durable run deletion. Internal thread-operation rows are left alone: the reservation protecting the deletion is one of them and its own release path owns it. Historical runs, the thread row, run events and checkpoints are unchanged. * fix(threads): keep a live finalize tail attached when evicting runs discard_thread_runs() popped every record it found, but worker.py clears finalizing before the tail that publishes the terminal frame, closes the journal and releases the sandbox lease. Popping mid-tail turns the tail's manager-mediated writes into warnings on a record that no longer exists. Skip records whose task is still running, mirroring what cleanup() already does through shield(), and pin the guard with a unit test that seeds both a finished run and a delete reservation. Document that a DELETE served by one Gateway bounds, rather than removes, a peer's copies: they survive until that peer's deferred cleanup(run_id) fires. * fix(threads): hide a deleted run from reads during its finalize tail A DELETE that arrives after worker.py clears `finalizing` but before the worker task finishes cannot drop the record: the tail still publishes the terminal frame, closes the journal and releases the sandbox lease through it, so popping turns those writes into `set_status called for unknown run`. Leaving the record readable instead keeps the run and its prompt reachable through `GET /threads/{id}/runs` until `cleanup()`'s 300s delay fires, which is exactly the window #8034 reports. Records that a deletion reaches mid-tail are now flagged instead of skipped. `get()`, `list_by_thread()`, `get_many_by_thread()`, the regenerate-source and edit-replay reads skip a flagged record (and never rebuild it from a store row, since the durable rows were deleted first), and the tail's own completion evicts it — no second delete request and no timer. * fix(threads): keep the tail eviction task referenced until it settles The done-callback eviction is a separate task because it cannot await the registry lock inline. The loop holds only weak references to tasks, so an unreferenced eviction could be collected before it drops the record and the deleted run would come back within `cleanup()`'s 300s window. Track it in a manager-owned set and discard on completion. * fix(threads): keep hidden runs hidden on every read path The deletion flag set by discard_thread_runs is authoritative for "this run is deleted", but two reads still treated it as a hint. get() narrowed its store-less guard to a missing record, so a flagged record on a store-less RunManager fell through to self._store.get() on None; the AttributeError was swallowed by the broad except below while every read in the flag window logged a hydration warning with a full traceback. The record now settles the read before any store call, which also keeps the read side-effect free as the comment intends. aget() shares the path. get_many_by_thread() rebuilt flagged runs from store rows. list_by_thread already refuses that, because the durable row is deleted before the flag is set and anything left is a stale or partial read; the batch read now applies the same guard so the invariant is uniform. --------- Co-authored-by: 2SinceJuly <202121119+2SinceJuly@users.noreply.github.com> --- backend/app/gateway/routers/threads.py | 11 ++ .../harness/deerflow/runtime/runs/manager.py | 135 +++++++++++++- backend/tests/test_run_manager.py | 175 ++++++++++++++++++ backend/tests/test_threads_router.py | 96 ++++++++++ 4 files changed, 410 insertions(+), 7 deletions(-) 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