fix(threads): evict in-memory runs when a thread is deleted (#8036)

* 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

@@ -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

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.
 

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."""

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