fix(threads): clean persisted records safely on thread deletion (#5535)

* fix(events): serialize DB deletion with thread writers

* fix(runs): delete thread history without dropping reservations

* fix(feedback): support owner-scoped thread cleanup

* fix(threads): clean persisted records on deletion

* fix(threads): correct the feedback cleanup rationale

* test(runs): drop the wall-clock probe from the in-flight delete test

* docs: record the thread-delete and event-store fence contracts

* fix(threads): preserve legacy event-store delete compatibility

backend/app/gateway/AGENTS.md

@@ -170,6 +170,7 @@ startup gate rejects process-local memory and JSONL event stores when
 - Run admission and independent writes are first-class thread operations. `runs.operation_kind` distinguishes `run` from `checkpoint_write`, `artifact_write`, `artifact_archive`, `branch`, and `delete`; every active kind shares the durable active-thread uniqueness constraint. New operation kinds must go through `RunStore.create_thread_operation_atomic()` and `RunManager.reserve_thread_operation()` rather than adding another lock or metadata marker. Live and lease-less reservations are non-interruptible; an expired leased reservation can be reclaimed immediately by interrupt/rollback admission without waiting for orphan reconciliation. Lease-less rows stay fail-closed because the store cannot distinguish a stale row from a live writer in another heartbeat-disabled worker; a rare failed delete therefore requires startup reconciliation, and heartbeat-disabled multi-worker deployment remains unsupported. Reservation bodies are attached to their caller task so loss detected by lease renewal cancels the writer before it can continue after takeover; the context manager translates that lease-loss cancellation to `ConflictError` after cleanup so Gateway mutation routes return a retryable 409 instead of dropping the HTTP request. The cleanup scope begins immediately after durable admission, including the await that attaches the caller task, so cancellation cannot strand a locally renewed pending reservation. A failed renewal is revalidated under the manager lock before cancellation; if the reservation completed and unregistered while the store update was in flight, its request task must not be cancelled after the write. Reservations are excluded from run history/reporting and from run-only helpers such as `list_by_thread()` and `has_inflight()`, release uses the captured owner rather than ambient user context, and local cleanup still runs when the best-effort store delete fails. `RunStore.create_run_atomic()` remains a deprecated compatibility shim for external stores that only admit normal runs; new stores must implement `create_thread_operation_atomic()` to support internal operation kinds.
 - Gateway checkpoint mutations outside run execution must use `services.reserve_checkpoint_write()`, which composes the process-local thread lock with the durable `checkpoint_write` reservation. Manual compaction, `POST /threads/{id}/state`, and both goal mutation routes (`PUT` / `DELETE /threads/{id}/goal`, including creation of a missing goal checkpoint) use this boundary, so an existing run blocks the write and the reservation blocks new reject/interrupt/rollback runs across workers.
 - Branch/state-update checkpoints copy only the source checkpoint's persisted `deerflow_agent_name`, never request metadata; missing or malformed bindings stay unbound so compaction fails closed.
+- `DELETE /api/threads/{id}` holds a durable `delete` reservation for the whole cleanup and removes thread filesystem data, checkpoints, owner-scoped historical runs, run events, feedback and thread metadata best-effort. Historical-run cleanup removes only `operation_kind="run"` rows, so internal thread-operation rows — including the reservation protecting that request — stay durable until `RunManager.reserve_thread_operation()` exits. This cleans persisted rows; preventing an already-admitted write from re-creating state for a deleted thread is a separate thread-incarnation contract, not part of this cleanup.
 - `POST /wait` (both thread-scoped and `/api/runs/wait`) drains the stream bridge via `wait_for_run_completion()` instead of bare `await record.task`, so it honours the run's `on_disconnect` setting and cancels the background run on real client disconnect rather than returning a stale checkpoint (issue #3265).
 - Memory and Redis `StreamBridge` implementations retain only `stream_bridge.queue_maxsize` data events. A syntactically valid `Last-Event-ID` older than the retained watermark, or a live subscriber that falls behind it, yields `StreamGap` before any partial replay. `sse_consumer` maps that control item to an id-less SSE `gap` payload (`stream_replay_gap`) and intentionally leaves the run active; internal `/wait` consumers resume from its latest retained ID because they only need terminal completion. Redis checks bounds plus the non-blocking read in one transaction, using blocking `XREAD` only as a wake-up before repeating the atomic snapshot. For a no-cursor subscriber that established a wait on an empty stream, the first wake response remains provisional until that next snapshot verifies its tail is still retained; this closes the pre-first-delivery trimming window without changing malformed-cursor live tailing. The correctness tradeoff is one three-command snapshot pipeline per poll plus the blocking wake round trip while idle. Malformed cursor behavior remains backend-specific. Memory treats a syntactically numeric cursor below its watermark conservatively as a gap even when the evicted timestamp can no longer be verified; unknown ids at or above the watermark retain the legacy replay-from-earliest policy.
 - Redis `StreamBridge` keys use a rolling retained-buffer TTL (`stream_bridge.stream_ttl_seconds`, refreshed on `publish()` / `publish_end()`) as a leak safety net, not as a run timeout. Startup and lease-driven periodic orphan recovery share one Gateway stream-terminalization path: after `RunManager` durably marks a run `error` with `stop_reason=orphan_recovered`, Gateway publishes `END_SENTINEL` and schedules stream cleanup. The periodic store scan, per-row status writes, and Gateway callback run as one supervised single-flight task, so a slow pass is skipped at the next interval instead of piling up or pausing the sole lease-renewal loop. Store retries have bounded attempts/backoff; an individual operation still relies on the database driver/pool timeout. `RunManager.shutdown()` gives active user runs priority within its shared deadline, then drains or cancels orphan recovery. Gateway tracks delayed recovered-stream cleanups and converts unfinished delays to immediate deletes before closing the bridge; the Redis TTL remains the outage safety net. Only startup recovery, before the runtime yields to requests, projects the latest affected thread to `error`; periodic recovery deliberately avoids that non-atomic projection because `ThreadMetaStore` has no `latest_run_id` conditional-update contract. Store-only SSE and `/wait` consumers wait for the bridge's real END marker after an ordinary durable terminal status, because status persistence can precede tail events. The explicit `orphan_recovered` signal is the only heartbeat fallback: its publisher is known to be gone, so it supplies the liveness boundary if END publication fails or the retained key expires. Malformed `Last-Event-ID` reconnect values live-tail new Redis events rather than replaying the retained buffer. Keep cross-component recovery orchestration in Gateway through the generic `RunManager.on_orphans_recovered` callback; do not introduce a harness-to-app dependency. Callback failure warnings include every recovered `run_id` so operators can identify rows whose Gateway-side terminalization needs inspection.

backend/app/gateway/routers/threads.py

@@ -12,6 +12,7 @@ matching the LangGraph Platform wire format expected by the
 
 from __future__ import annotations
 
+import inspect
 import logging
 import shutil
 import uuid
@@ -32,7 +33,7 @@ from app.gateway.checkpoint_lineage import (
     find_checkpoint_before_message_chronologically,
     is_duration_only_checkpoint,
 )
-from app.gateway.deps import get_checkpointer, get_run_event_store, get_run_manager
+from app.gateway.deps import get_checkpointer, get_run_event_store, get_run_manager, get_run_store
 from app.gateway.internal_auth import get_trusted_internal_owner_user_id
 from app.gateway.services import (
     abuild_checkpoint_state_accessor,
@@ -727,10 +728,33 @@ async def delete_thread_data(thread_id: str, request: Request) -> ThreadDeleteRe
         ) from None
 
 
+def _event_delete_owner_kwargs(delete_by_thread: Any, user_id: str) -> dict[str, str]:
+    """Pass owner scope only when an event store accepts that keyword.
+
+    Third-party ``RunEventStore`` implementations may still expose the legacy
+    ``delete_by_thread(thread_id)`` contract; they must keep deleting, just
+    without the owner filter (their storage is not user-scoped). An
+    uninspectable callable keeps the old call contract, and a ``TypeError``
+    raised inside the backend must never trigger a retry.
+    """
+    try:
+        parameters = inspect.signature(delete_by_thread).parameters.values()
+    except (TypeError, ValueError):
+        return {}
+    if any(parameter.kind == inspect.Parameter.VAR_KEYWORD or (parameter.name == "user_id" and parameter.kind in (inspect.Parameter.POSITIONAL_OR_KEYWORD, inspect.Parameter.KEYWORD_ONLY)) for parameter in parameters):
+        return {"user_id": user_id}
+    return {}
+
+
 async def _delete_thread_data_with_reservation(thread_id: str, request: Request) -> ThreadDeleteResponse:
     """Delete a thread while its durable exclusive reservation is held."""
     from app.gateway.deps import get_thread_store
 
+    # One owner identity for every cleanup step below: the filesystem bucket, the
+    # 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()
+
     # Legacy IDs may predate the canonical filesystem-safe contract. They can
     # still be removed from metadata/checkpoint stores, but must never be
     # interpolated into a host path during cleanup.
@@ -742,7 +766,7 @@ async def _delete_thread_data_with_reservation(thread_id: str, request: Request)
             message="Skipped local data cleanup for legacy thread ID",
         )
     else:
-        response = _delete_thread_data(thread_id, user_id=get_effective_user_id())
+        response = _delete_thread_data(thread_id, user_id=user_id)
 
     # Remove checkpoints (best-effort)
     checkpointer = getattr(request.app.state, "checkpointer", None)
@@ -753,11 +777,44 @@ async def _delete_thread_data_with_reservation(thread_id: str, request: Request)
         except Exception:
             logger.debug("Could not delete checkpoints for thread %s (not critical)", sanitize_log_param(thread_id))
 
+    # Remove historical runs (best-effort). Only ``operation_kind == "run"`` rows
+    # are deleted, so the durable thread-operation reservation protecting this
+    # very request survives until ``reserve_thread_operation`` exits. Third-party
+    # RunStore implementations that predate the capability are skipped.
+    try:
+        delete_runs = getattr(get_run_store(request), "delete_by_thread", None)
+        if delete_runs is not None:
+            await delete_runs(thread_id, user_id=user_id)
+    except Exception:
+        logger.debug("Could not delete run records 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
+    # store that predates the owner-scoped signature is still called, with the
+    # old contract.
+    try:
+        delete_events = get_run_event_store(request).delete_by_thread
+        await delete_events(thread_id, **_event_delete_owner_kwargs(delete_events, user_id))
+    except Exception:
+        logger.debug("Could not delete run events for thread %s (not critical)", sanitize_log_param(thread_id))
+
+    # Remove persisted feedback best-effort. This cleans existing rows; fencing
+    # already-admitted writes across thread deletion is a separate lifecycle
+    # concern. The memory backend legitimately sets ``feedback_repo = None``, so
+    # the optional accessor is used here.
+    try:
+        feedback_repo = getattr(request.app.state, "feedback_repo", None)
+        if feedback_repo is not None:
+            await feedback_repo.delete_by_thread(thread_id, user_id=user_id)
+    except Exception:
+        logger.debug("Could not delete feedback for thread %s (not critical)", sanitize_log_param(thread_id))
+
     # Remove thread_meta row (best-effort) — required for sqlite backend
     # so the deleted thread no longer appears in /threads/search.
     try:
         thread_store = get_thread_store(request)
-        await thread_store.delete(thread_id)
+        await thread_store.delete(thread_id, user_id=user_id)
     except Exception:
         logger.debug("Could not delete thread_meta for %s (not critical)", sanitize_log_param(thread_id))
 

backend/packages/harness/deerflow/persistence/feedback/sql.py

@@ -8,7 +8,7 @@ from __future__ import annotations
 import uuid
 from datetime import UTC, datetime
 
-from sqlalchemy import case, func, select
+from sqlalchemy import case, delete, func, select
 from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker
 
 from deerflow.persistence.feedback.model import FeedbackRow
@@ -187,6 +187,31 @@ class FeedbackRepository:
             await session.commit()
             return True
 
+    async def delete_by_thread(
+        self,
+        thread_id: str,
+        *,
+        user_id: str | None | _AutoSentinel = AUTO,
+    ) -> int:
+        """Delete the owner's feedback for every run of a thread.
+
+        ``user_id`` keeps the repository's three-state convention: ``AUTO``
+        resolves the request context, an explicit id scopes the delete to that
+        owner, and ``None`` removes every owner's rows (migration/CLI callers).
+        """
+        resolved_user_id = resolve_user_id(user_id, method_name="FeedbackRepository.delete_by_thread")
+
+        conditions = [FeedbackRow.thread_id == thread_id]
+        if resolved_user_id is not None:
+            conditions.append(FeedbackRow.user_id == resolved_user_id)
+
+        async with self._sf() as session:
+            count = await session.scalar(select(func.count()).select_from(FeedbackRow).where(*conditions)) or 0
+            if count:
+                await session.execute(delete(FeedbackRow).where(*conditions))
+            await session.commit()
+            return count
+
     async def list_by_thread_grouped(
         self,
         thread_id: str,

backend/packages/harness/deerflow/persistence/run/sql.py

@@ -11,7 +11,7 @@ import json
 from datetime import UTC, datetime, timedelta
 from typing import Any
 
-from sqlalchemy import and_, case, or_, select, update
+from sqlalchemy import and_, case, delete, func, or_, select, update
 from sqlalchemy.exc import IntegrityError
 from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker
 
@@ -358,6 +358,44 @@ class RunRepository(RunStore):
             await session.delete(row)
             await session.commit()
 
+    async def delete_by_thread(
+        self,
+        thread_id: str,
+        *,
+        user_id: str | None | _AutoSentinel = AUTO,
+    ) -> int:
+        """Delete a thread's historical runs, keeping internal operation rows.
+
+        Only ``operation_kind == "run"`` rows are removed. The same table holds
+        durable thread-operation reservations (checkpoint writes, artifact
+        writes, branches and thread deletions) that their own release path owns
+        through ``delete_thread_operation``; deleting the reservation currently
+        protecting a thread deletion would drop the cross-worker exclusion in
+        the middle of that request.
+
+        The bulk delete deliberately does not bump the run-change clock, matching
+        the single-row :meth:`delete` and keeping thread cleanup away from
+        ``run_change_clock`` (#5516). ``user_id`` follows the same three-state
+        convention as the rest of this repository: ``AUTO`` reads the request
+        context (raising when there is none), an explicit id scopes the delete,
+        and ``None`` skips the owner filter for migration/CLI callers.
+        """
+        resolved_user_id = resolve_user_id(user_id, method_name="RunRepository.delete_by_thread")
+
+        conditions = [
+            RunRow.thread_id == thread_id,
+            RunRow.operation_kind == "run",
+        ]
+        if resolved_user_id is not None:
+            conditions.append(RunRow.user_id == resolved_user_id)
+
+        async with self._sf() as session:
+            count = await session.scalar(select(func.count()).select_from(RunRow).where(*conditions)) or 0
+            if count:
+                await session.execute(delete(RunRow).where(*conditions))
+            await session.commit()
+            return count
+
     async def delete_thread_operation(self, run_id: str, *, user_id: str | None) -> None:
         """Release a reservation using its captured owner, not request context."""
         await self.delete(run_id, user_id=user_id)

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

@@ -170,6 +170,19 @@ with one complete thread-log read because each JSONL page would otherwise
 rescan every run file. The default and JSONL paths share the public
 `normalize_message_ids()` and `match_ai_message_run_id()` helpers from
 `events/store/base.py`. Database owner filtering is inherited on every page.
+
+**Event-store mutation fence** (`runtime/events/store/`): every thread mutation —
+`put`, `put_batch`, `put_if_absent`, `delete_by_thread`, `delete_by_run` — shares
+one serialization domain: the per-thread `asyncio` lock, plus (on PostgreSQL) the
+transaction-scoped advisory lock keyed by `thread_id` that
+`DbRunEventStore._acquire_thread_mutation_fence()` takes before any read or write.
+Deletion therefore cannot interleave with an admitted writer and re-create rows for
+a deleted thread, and all backends accept the same owner-scoped delete signature
+(`user_id` filters on the DB store; memory/JSONL accept it for parity). This is
+serialization, not an incarnation fence: a mutation already admitted before a
+deletion may still run afterwards, and preventing old-incarnation resurrection
+needs a separate durable generation contract. `JsonlRunEventStore` keeps its own
+equivalent guarantee through `_run_mutation`.
 Callers may use a missing key as proof that no valid AI event exists only after
 an ordinary return, never after an exception. A caller that crosses a run or
 checkpoint-write admission boundary must repeat the complete audit after

backend/packages/harness/deerflow/runtime/events/store/base.py

@@ -269,9 +269,30 @@ class RunEventStore(abc.ABC):
         """
 
     @abc.abstractmethod
-    async def delete_by_thread(self, thread_id: str) -> int:
-        """Delete all events for a thread. Return the number of deleted events."""
+    async def delete_by_thread(
+        self,
+        thread_id: str,
+        *,
+        user_id: str | None | _AutoSentinel = AUTO,
+    ) -> int:
+        """Delete all events for a thread. Return the number of deleted events.
+
+        ``user_id`` follows the same three-state convention as the read methods:
+        ``AUTO`` resolves the caller's context, an explicit id scopes the delete
+        to that owner, and ``None`` removes every owner's rows. Backends whose
+        storage is not user-scoped (memory, JSONL) accept the parameter for
+        interface parity and ignore it, so callers can delete uniformly.
+        """
 
     @abc.abstractmethod
-    async def delete_by_run(self, thread_id: str, run_id: str) -> int:
-        """Delete all events for a specific run. Return the number of deleted events."""
+    async def delete_by_run(
+        self,
+        thread_id: str,
+        run_id: str,
+        *,
+        user_id: str | None | _AutoSentinel = AUTO,
+    ) -> int:
+        """Delete all events for a specific run. Return the number of deleted events.
+
+        ``user_id`` follows the same convention as :meth:`delete_by_thread`.
+        """

backend/packages/harness/deerflow/runtime/events/store/db.py

@@ -142,15 +142,21 @@ class DbRunEventStore(RunEventStore):
         return ids
 
     @staticmethod
-    async def _max_seq_for_thread(session: AsyncSession, thread_id: str) -> int | None:
-        """Return the current max seq while serializing writers per thread.
+    async def _acquire_thread_mutation_fence(session: AsyncSession, thread_id: str) -> None:
+        """Take the cross-process thread mutation fence, if the dialect has one.
 
         PostgreSQL rejects ``SELECT max(...) FOR UPDATE`` because aggregate
-        results are not lockable rows. As a release-safe workaround, take a
-        transaction-level advisory lock keyed by thread_id before reading the
-        aggregate. Other dialects keep the existing row-locking statement.
+        results are not lockable rows, so it serializes a thread's mutations with
+        a transaction-level advisory lock keyed by ``thread_id``. This is the
+        database half of the contract whose in-process half is
+        ``_get_write_lock()``: every thread mutation — ``put``, ``put_batch``,
+        ``put_if_absent`` and both deletions — takes this fence before touching
+        rows, so an admitted writer can never land a row between a deletion's
+        count and its commit.
+
+        Dialects without a cross-process fence (SQLite) rely on the in-process
+        per-thread lock alone, so this is a no-op there.
         """
-        stmt = select(func.max(RunEventRow.seq)).where(RunEventRow.thread_id == thread_id)
         bind = session.get_bind()
         dialect_name = bind.dialect.name if bind is not None else ""
 
@@ -159,6 +165,22 @@ class DbRunEventStore(RunEventStore):
                 text("SELECT pg_advisory_xact_lock(hashtext(CAST(:thread_id AS text))::bigint)"),
                 {"thread_id": thread_id},
             )
+
+    @staticmethod
+    async def _max_seq_for_thread(session: AsyncSession, thread_id: str) -> int | None:
+        """Return the current max seq while serializing writers per thread.
+
+        Takes the shared thread mutation fence before reading the aggregate, so
+        the read is ordered against every other mutation of the same thread.
+        Other dialects keep the existing row-locking statement.
+        """
+        await DbRunEventStore._acquire_thread_mutation_fence(session, thread_id)
+
+        stmt = select(func.max(RunEventRow.seq)).where(RunEventRow.thread_id == thread_id)
+        bind = session.get_bind()
+        dialect_name = bind.dialect.name if bind is not None else ""
+
+        if dialect_name == "postgresql":
             return await session.scalar(stmt)
 
         return await session.scalar(stmt.with_for_update())
@@ -487,16 +509,26 @@ class DbRunEventStore(RunEventStore):
         *,
         user_id: str | None | _AutoSentinel = AUTO,
     ):
+        """Delete every event of *thread_id* inside the thread mutation fence.
+
+        Deletion takes the same critical section as the writers — the in-process
+        per-thread lock plus, on PostgreSQL, the transaction advisory lock — so a
+        writer admitted before this call can no longer land a row between the
+        count below and the commit, which would resurrect a deleted thread. The
+        JSONL store serializes deletion the same way (``_run_mutation``).
+        """
         resolved_user_id = resolve_user_id(user_id, method_name="DbRunEventStore.delete_by_thread")
-        async with self._sf() as session:
-            count_conditions = [RunEventRow.thread_id == thread_id]
-            if resolved_user_id is not None:
-                count_conditions.append(RunEventRow.user_id == resolved_user_id)
-            count_stmt = select(func.count()).select_from(RunEventRow).where(*count_conditions)
-            count = await session.scalar(count_stmt) or 0
-            if count > 0:
-                await session.execute(delete(RunEventRow).where(*count_conditions))
-                await session.commit()
+        async with self._get_write_lock(thread_id):
+            async with self._sf() as session:
+                async with session.begin():
+                    await self._acquire_thread_mutation_fence(session, thread_id)
+                    count_conditions = [RunEventRow.thread_id == thread_id]
+                    if resolved_user_id is not None:
+                        count_conditions.append(RunEventRow.user_id == resolved_user_id)
+                    count_stmt = select(func.count()).select_from(RunEventRow).where(*count_conditions)
+                    count = await session.scalar(count_stmt) or 0
+                    if count > 0:
+                        await session.execute(delete(RunEventRow).where(*count_conditions))
             # Retire the live-thread pin, but never remove the weak registry
             # entry directly. asyncio.Lock.release() clears ``locked()`` before
             # a queued waiter resumes, so an unlocked check can observe the
@@ -505,7 +537,7 @@ class DbRunEventStore(RunEventStore):
             # later caller therefore resolves that same lock instead of racing
             # it with a fresh one.
             self._write_lock_pins.pop(thread_id, None)
-            return count
+        return count
 
     async def delete_by_run(
         self,
@@ -514,14 +546,21 @@ class DbRunEventStore(RunEventStore):
         *,
         user_id: str | None | _AutoSentinel = AUTO,
     ):
+        """Delete one run's events inside the thread mutation fence.
+
+        Shares ``delete_by_thread``'s critical section; deleting a single run
+        leaves the thread alive, so the write-lock pin is deliberately kept.
+        """
         resolved_user_id = resolve_user_id(user_id, method_name="DbRunEventStore.delete_by_run")
-        async with self._sf() as session:
-            count_conditions = [RunEventRow.thread_id == thread_id, RunEventRow.run_id == run_id]
-            if resolved_user_id is not None:
-                count_conditions.append(RunEventRow.user_id == resolved_user_id)
-            count_stmt = select(func.count()).select_from(RunEventRow).where(*count_conditions)
-            count = await session.scalar(count_stmt) or 0
-            if count > 0:
-                await session.execute(delete(RunEventRow).where(*count_conditions))
-                await session.commit()
-            return count
+        async with self._get_write_lock(thread_id):
+            async with self._sf() as session:
+                async with session.begin():
+                    await self._acquire_thread_mutation_fence(session, thread_id)
+                    count_conditions = [RunEventRow.thread_id == thread_id, RunEventRow.run_id == run_id]
+                    if resolved_user_id is not None:
+                        count_conditions.append(RunEventRow.user_id == resolved_user_id)
+                    count_stmt = select(func.count()).select_from(RunEventRow).where(*count_conditions)
+                    count = await session.scalar(count_stmt) or 0
+                    if count > 0:
+                        await session.execute(delete(RunEventRow).where(*count_conditions))
+        return count

backend/packages/harness/deerflow/runtime/events/store/jsonl.py

@@ -428,7 +428,14 @@ class JsonlRunEventStore(RunEventStore):
                     break
         return found
 
-    async def delete_by_thread(self, thread_id):
+    async def delete_by_thread(self, thread_id, *, user_id: str | None | _AutoSentinel = AUTO):
+        """Delete every event of a thread.
+
+        Run files are keyed by thread, not by owner, so ``user_id`` is accepted
+        for interface parity with the user-scoped backends and ignored — the same
+        convention as this store's read methods.
+        """
+
         async def mutate():
             all_events = await asyncio.to_thread(self._read_thread_events, thread_id)
             count = len(all_events)
@@ -440,7 +447,9 @@ class JsonlRunEventStore(RunEventStore):
 
         return await self._run_mutation(thread_id, mutate)
 
-    async def delete_by_run(self, thread_id, run_id):
+    async def delete_by_run(self, thread_id, run_id, *, user_id: str | None | _AutoSentinel = AUTO):
+        """Delete one run's events; ``user_id`` is accepted for parity only."""
+
         async def mutate():
             events = await asyncio.to_thread(self._read_run_events, thread_id, run_id)
             count = len(events)

backend/packages/harness/deerflow/runtime/events/store/memory.py

@@ -211,7 +211,13 @@ class MemoryRunEventStore(RunEventStore):
                     break
         return found
 
-    async def delete_by_thread(self, thread_id):
+    async def delete_by_thread(self, thread_id, *, user_id: str | None | _AutoSentinel = AUTO):
+        """Delete every event of a thread.
+
+        Events live in process memory without an owner column, so ``user_id`` is
+        accepted for interface parity with the user-scoped backends and ignored
+        — the same convention as this store's read methods.
+        """
         events = self._events.pop(thread_id, [])
         self._messages.pop(thread_id, None)
         self._events_by_run.pop(thread_id, None)
@@ -219,7 +225,8 @@ class MemoryRunEventStore(RunEventStore):
         self._seq_counters.pop(thread_id, None)
         return len(events)
 
-    async def delete_by_run(self, thread_id, run_id):
+    async def delete_by_run(self, thread_id, run_id, *, user_id: str | None | _AutoSentinel = AUTO):
+        """Delete one run's events; ``user_id`` is accepted for parity only."""
         all_events = self._events.get(thread_id, [])
         if not all_events:
             return 0

backend/packages/harness/deerflow/runtime/runs/store/memory.py

@@ -215,6 +215,25 @@ class MemoryRunStore(RunStore):
         if run is not None:
             self._unindex_run(run_id, run["thread_id"])
 
+    async def delete_by_thread(self, thread_id: str, *, user_id=None) -> int:
+        """Delete a thread's historical runs, keeping internal operation rows.
+
+        Mirrors ``RunRepository.delete_by_thread``: only ``operation_kind ==
+        "run"`` rows are removed, so durable thread-operation reservations keep
+        protecting the thread until their own release path drops them.
+        """
+        removed = 0
+        for run_id in list(self._runs_by_thread.get(thread_id, {})):
+            run = self._runs.get(run_id)
+            if run is None or run.get("operation_kind", "run") != "run":
+                continue
+            if user_id is not None and run.get("user_id") != user_id:
+                continue
+            self._runs.pop(run_id, None)
+            self._unindex_run(run_id, run["thread_id"])
+            removed += 1
+        return removed
+
     async def update_run_completion(self, run_id, *, status, **kwargs):
         run = self._runs.get(run_id)
         if run is None:

backend/tests/test_db_event_store_lock_lifecycle.py

@@ -7,6 +7,14 @@ from deerflow.runtime.events.store.db import DbRunEventStore
 
 
 class _PausedDeleteSession:
+    """Fake session that pauses the deletion on its first aggregate read.
+
+    The fixed deletion path opens the session, takes the thread mutation fence
+    (a no-op on this fake's dialect) and only then reads the expected count, so
+    pausing in ``scalar`` proves the caller already owns the thread's mutation
+    critical section.
+    """
+
     def __init__(self, scalar_started: asyncio.Event, allow_scalar: asyncio.Event) -> None:
         self._scalar_started = scalar_started
         self._allow_scalar = allow_scalar
@@ -17,12 +25,20 @@ class _PausedDeleteSession:
     async def __aexit__(self, exc_type, exc, tb):
         return False
 
+    def get_bind(self):
+        # Non-PostgreSQL dialect: ``_acquire_thread_mutation_fence`` must not
+        # emit advisory-lock SQL here — the in-process lock is the only fence.
+        return None
+
+    def begin(self):
+        return self
+
     async def scalar(self, _stmt):
         self._scalar_started.set()
         await self._allow_scalar.wait()
         return 1
 
-    async def execute(self, _stmt):
+    async def execute(self, _stmt, _params=None):
         return None
 
     async def commit(self) -> None:
@@ -51,21 +67,31 @@ async def test_delete_waiter_handoff_keeps_one_write_lock_generation():
             waiter_entered.set()
             await release_waiter.wait()
 
+    delete_task = asyncio.create_task(store.delete_by_thread("t1", user_id=None))
+    await asyncio.sleep(0)
+
+    # B1 contract: deletion is queued behind the existing holder instead of
+    # running concurrently with it.
+    assert not scalar_started.is_set()
+    assert not delete_task.done()
+
     waiter_task = asyncio.create_task(queued_writer())
     await waiter_resolved.wait()
     await asyncio.sleep(0)
     assert not waiter_entered.is_set()
 
-    delete_task = asyncio.create_task(store.delete_by_thread("t1", user_id=None))
-    await scalar_started.wait()
-
     # Resume deletion first, then release the holder. asyncio.Lock.release()
     # marks the lock unlocked before the queued waiter resumes, so the old
     # implementation can evict the registry entry in that handoff window.
-    allow_scalar.set()
     old_lock.release()
     del old_lock
 
+    # Delete now owns the same generation, and the queued writer is still
+    # waiting behind it.
+    await scalar_started.wait()
+    assert not waiter_entered.is_set()
+
+    allow_scalar.set()
     await delete_task
     await waiter_entered.wait()
 
@@ -76,3 +102,53 @@ async def test_delete_waiter_handoff_keeps_one_write_lock_generation():
     finally:
         release_waiter.set()
         await waiter_task
+
+
+@pytest.mark.anyio
+async def test_delete_by_thread_waits_for_thread_write_lock():
+    """Deletion must not enter its DB mutation while a writer holds the lock."""
+    scalar_started = asyncio.Event()
+    allow_scalar = asyncio.Event()
+    allow_scalar.set()
+
+    session = _PausedDeleteSession(scalar_started, allow_scalar)
+    store = DbRunEventStore(lambda: session)
+
+    lock = store._get_write_lock("t1")
+    await lock.acquire()
+
+    task = asyncio.create_task(store.delete_by_thread("t1", user_id=None))
+    await asyncio.sleep(0)
+
+    assert not scalar_started.is_set()
+    assert not task.done()
+
+    lock.release()
+
+    assert await task == 1
+    assert scalar_started.is_set()
+
+
+@pytest.mark.anyio
+async def test_delete_by_run_waits_for_thread_write_lock():
+    """delete_by_run shares the same fence as delete_by_thread."""
+    scalar_started = asyncio.Event()
+    allow_scalar = asyncio.Event()
+    allow_scalar.set()
+
+    session = _PausedDeleteSession(scalar_started, allow_scalar)
+    store = DbRunEventStore(lambda: session)
+
+    lock = store._get_write_lock("t1")
+    await lock.acquire()
+
+    task = asyncio.create_task(store.delete_by_run("t1", "r1", user_id=None))
+    await asyncio.sleep(0)
+
+    assert not scalar_started.is_set()
+    assert not task.done()
+
+    lock.release()
+
+    assert await task == 1
+    assert scalar_started.is_set()

backend/tests/test_feedback.py

@@ -335,3 +335,36 @@ class TestFollowUpAssociation:
         if recent and recent[0].get("status") == "success":
             follow_up = recent[0]["run_id"]
         assert follow_up == "r3"
+
+
+class TestDeleteByThread:
+    """Thread deletion clears feedback without crossing owner or thread scope."""
+
+    @pytest.mark.anyio
+    async def test_deletes_only_the_owners_rows_for_that_thread(self, tmp_path):
+        repo = await _make_feedback_repo(tmp_path)
+        try:
+            await repo.create(run_id="r1", thread_id="t1", rating=1, user_id="alice")
+            await repo.create(run_id="r2", thread_id="t1", rating=-1, user_id="alice")
+            await repo.create(run_id="r1", thread_id="t1", rating=1, user_id="bob")
+            await repo.create(run_id="r1", thread_id="t2", rating=1, user_id="alice")
+
+            count = await repo.delete_by_thread("t1", user_id="alice")
+
+            assert count == 2
+            assert await repo.list_by_thread("t1", user_id="alice") == []
+            assert len(await repo.list_by_thread("t1", user_id="bob")) == 1
+            assert len(await repo.list_by_thread("t2", user_id="alice")) == 1
+        finally:
+            await _cleanup()
+
+    @pytest.mark.anyio
+    async def test_second_call_returns_zero(self, tmp_path):
+        repo = await _make_feedback_repo(tmp_path)
+        try:
+            await repo.create(run_id="r1", thread_id="t1", rating=1, user_id="alice")
+
+            assert await repo.delete_by_thread("t1", user_id="alice") == 1
+            assert await repo.delete_by_thread("t1", user_id="alice") == 0
+        finally:
+            await _cleanup()

backend/tests/test_jsonl_event_store_async_io.py

@@ -362,6 +362,20 @@ async def test_delete_by_run_removes_run_events():
     assert events == []
 
 
+@pytest.mark.anyio
+async def test_delete_methods_accept_the_gateway_owner_scope():
+    """The Gateway calls deletion with an explicit owner on every backend."""
+    with tempfile.TemporaryDirectory() as tmp:
+        store = _make_store(Path(tmp))
+        await store.put(thread_id="t1", run_id="r1", event_type="human_message", category="message")
+        await store.put(thread_id="t1", run_id="r2", event_type="human_message", category="message")
+
+        assert await store.delete_by_run("t1", "r1", user_id="alice") == 1
+        assert await store.count_messages("t1") == 1
+        assert await store.delete_by_thread("t1", user_id="alice") == 1
+        assert await store.count_messages("t1") == 0
+
+
 # ---------------------------------------------------------------------------
 # DB put_batch: rejects mixed-thread batches
 # ---------------------------------------------------------------------------

backend/tests/test_run_event_store.py

@@ -457,6 +457,31 @@ class TestDelete:
         assert len(messages) == 1
         assert messages[0]["run_id"] == "r1"
 
+    @pytest.mark.anyio
+    async def test_delete_by_thread_accepts_owner_scope(self, store):
+        """Every backend accepts the owner scope the Gateway passes (#2803 wiring).
+
+        User-scoped backends apply the filter; the in-memory store is not
+        user-scoped and accepts it for interface parity.
+        """
+        await store.put(thread_id="t1", run_id="r1", event_type="human_message", category="message")
+
+        count = await store.delete_by_thread("t1", user_id="alice")
+
+        assert count == 1
+        assert await store.count_messages("t1") == 0
+
+    @pytest.mark.anyio
+    async def test_delete_by_run_accepts_owner_scope(self, store):
+        await store.put(thread_id="t1", run_id="r1", event_type="human_message", category="message")
+        await store.put(thread_id="t1", run_id="r2", event_type="human_message", category="message")
+
+        count = await store.delete_by_run("t1", "r1", user_id="alice")
+
+        assert count == 1
+        messages = await store.list_messages("t1")
+        assert [message["run_id"] for message in messages] == ["r2"]
+
     @pytest.mark.anyio
     async def test_delete_nonexistent_thread_returns_zero(self, store):
         assert await store.delete_by_thread("nope") == 0
@@ -527,6 +552,90 @@ class TestDbRunEventStore:
         compiled = str(session.scalar_stmt.compile(dialect=postgresql.dialect()))
         assert "FOR UPDATE" not in compiled
 
+    @pytest.mark.anyio
+    async def test_delete_by_thread_takes_postgres_advisory_lock(self):
+        """Deletion must enter the same cross-process fence as writers (#5530)."""
+        from sqlalchemy.dialects import postgresql
+
+        from deerflow.runtime.events.store.db import DbRunEventStore
+
+        class FakeSession:
+            def __init__(self):
+                self.dialect = postgresql.dialect()
+                self.execute_calls = []
+
+            def get_bind(self):
+                return self
+
+            async def __aenter__(self):
+                return self
+
+            async def __aexit__(self, exc_type, exc, tb):
+                return False
+
+            def begin(self):
+                return self
+
+            async def execute(self, stmt, params=None):
+                self.execute_calls.append((stmt, params))
+
+            async def scalar(self, _stmt):
+                return 3
+
+            async def commit(self) -> None:
+                return None
+
+        session = FakeSession()
+
+        count = await DbRunEventStore(lambda: session).delete_by_thread("thread-1", user_id=None)
+
+        assert count == 3
+        assert session.execute_calls
+        assert "pg_advisory_xact_lock" in str(session.execute_calls[0][0])
+        assert session.execute_calls[0][1] == {"thread_id": "thread-1"}
+
+    @pytest.mark.anyio
+    async def test_delete_by_run_takes_postgres_advisory_lock(self):
+        """delete_by_run shares the cross-process fence as well (#5530)."""
+        from sqlalchemy.dialects import postgresql
+
+        from deerflow.runtime.events.store.db import DbRunEventStore
+
+        class FakeSession:
+            def __init__(self):
+                self.dialect = postgresql.dialect()
+                self.execute_calls = []
+
+            def get_bind(self):
+                return self
+
+            async def __aenter__(self):
+                return self
+
+            async def __aexit__(self, exc_type, exc, tb):
+                return False
+
+            def begin(self):
+                return self
+
+            async def execute(self, stmt, params=None):
+                self.execute_calls.append((stmt, params))
+
+            async def scalar(self, _stmt):
+                return 2
+
+            async def commit(self) -> None:
+                return None
+
+        session = FakeSession()
+
+        count = await DbRunEventStore(lambda: session).delete_by_run("thread-1", "run-1", user_id=None)
+
+        assert count == 2
+        assert session.execute_calls
+        assert "pg_advisory_xact_lock" in str(session.execute_calls[0][0])
+        assert session.execute_calls[0][1] == {"thread_id": "thread-1"}
+
     @pytest.mark.anyio
     async def test_basic_crud(self, tmp_path):
         from deerflow.persistence.engine import close_engine, get_session_factory, init_engine
@@ -952,6 +1061,8 @@ class TestDbRunEventStoreWriteLock:
 
     @pytest.mark.anyio
     async def test_delete_by_thread_keeps_lock_held_by_inflight_writer(self, tmp_path):
+        import asyncio
+
         from deerflow.persistence.engine import close_engine, get_session_factory, init_engine
         from deerflow.runtime.events.store.db import DbRunEventStore
 
@@ -959,16 +1070,27 @@ class TestDbRunEventStoreWriteLock:
         await init_engine("sqlite", url=url, sqlite_dir=str(tmp_path))
         s = DbRunEventStore(get_session_factory())
 
-        # Simulate a writer mid-flight by holding the lock; the eviction must
-        # not drop a lock another coroutine is actively using.
+        # Simulate a writer mid-flight by holding the lock. Deletion now shares
+        # the fence, so it must queue behind the in-flight writer instead of
+        # running concurrently with it. The strict ordering guarantee is pinned
+        # without wall-clock timing by the Event-driven tests in
+        # tests/test_db_event_store_lock_lifecycle.py; this test covers the real
+        # SQLite deletion path and the registry state it leaves behind.
         lock = s._get_write_lock("t1")
         await lock.acquire()
-        try:
-            await s.delete_by_thread("t1")
-            assert "t1" in s._write_locks
-            assert s._write_locks["t1"] is lock
-        finally:
-            lock.release()
+
+        delete_task = asyncio.create_task(s.delete_by_thread("t1"))
+        await asyncio.sleep(0)
+        assert not delete_task.done()
+
+        lock.release()
+
+        await delete_task
+
+        # The eviction must not drop a lock another coroutine still holds: the
+        # generation this test references stays resolvable for later writers.
+        assert "t1" in s._write_locks
+        assert s._write_locks["t1"] is lock
 
         await close_engine()
 

backend/tests/test_run_repository.py

@@ -1250,3 +1250,109 @@ class TestRunRepository:
         ok = await repo.claim_for_takeover("no-such-run", grace_seconds=10, error="claimed")
         assert ok is False
         await _cleanup()
+
+
+class TestRunRepositoryDeleteByThread:
+    """Bulk thread cleanup must never delete internal thread-operation rows."""
+
+    @pytest.mark.anyio
+    async def test_preserves_thread_operation_reservation(self, tmp_path):
+        repo = await _make_repo(tmp_path)
+        try:
+            await repo.put("historical-run", thread_id="t1", user_id="alice", status="success", operation_kind="run")
+            await repo.put(
+                "delete-reservation",
+                thread_id="t1",
+                user_id="alice",
+                status="pending",
+                operation_kind=ThreadOperationKind.delete,
+            )
+
+            count = await repo.delete_by_thread("t1", user_id="alice")
+
+            assert count == 1
+            assert await repo.get("historical-run", user_id="alice") is None
+            reservation = await repo.get("delete-reservation", user_id="alice")
+            assert reservation is not None
+            assert reservation["operation_kind"] == "delete"
+        finally:
+            await _cleanup()
+
+    @pytest.mark.anyio
+    async def test_is_owner_and_thread_scoped(self, tmp_path):
+        repo = await _make_repo(tmp_path)
+        try:
+            await repo.put("alice-t1", thread_id="t1", user_id="alice", status="success")
+            await repo.put("bob-t1", thread_id="t1", user_id="bob", status="success")
+            await repo.put("alice-t2", thread_id="t2", user_id="alice", status="success")
+
+            count = await repo.delete_by_thread("t1", user_id="alice")
+
+            assert count == 1
+            assert await repo.get("alice-t1", user_id="alice") is None
+            assert await repo.get("bob-t1", user_id="bob") is not None
+            assert await repo.get("alice-t2", user_id="alice") is not None
+        finally:
+            await _cleanup()
+
+    @pytest.mark.anyio
+    async def test_second_call_returns_zero(self, tmp_path):
+        repo = await _make_repo(tmp_path)
+        try:
+            await repo.put("alice-t1", thread_id="t1", user_id="alice", status="success")
+
+            assert await repo.delete_by_thread("t1", user_id="alice") == 1
+            assert await repo.delete_by_thread("t1", user_id="alice") == 0
+        finally:
+            await _cleanup()
+
+
+class TestMemoryRunStoreDeleteByThread:
+    """MemoryRunStore mirrors the SQL store's cleanup semantics."""
+
+    @pytest.mark.anyio
+    async def test_removes_only_run_operations_for_the_owner(self):
+        from deerflow.runtime.runs.store.memory import MemoryRunStore
+
+        store = MemoryRunStore()
+        await store.put("alice-run", thread_id="t1", user_id="alice", status="success")
+        await store.put(
+            "delete-reservation",
+            thread_id="t1",
+            user_id="alice",
+            status="pending",
+            operation_kind=ThreadOperationKind.delete,
+        )
+        await store.put("bob-run", thread_id="t1", user_id="bob", status="success")
+        await store.put("alice-other-thread", thread_id="t2", user_id="alice", status="success")
+
+        count = await store.delete_by_thread("t1", user_id="alice")
+
+        assert count == 1
+        assert await store.get("alice-run", user_id="alice") is None
+        assert await store.get("delete-reservation", user_id="alice") is not None
+        assert await store.get("bob-run", user_id="bob") is not None
+        assert await store.get("alice-other-thread", user_id="alice") is not None
+
+    @pytest.mark.anyio
+    async def test_keeps_the_thread_index_consistent(self):
+        from deerflow.runtime.runs.store.memory import MemoryRunStore
+
+        store = MemoryRunStore()
+        await store.put("alice-run", thread_id="t1", user_id="alice", status="success")
+        await store.put(
+            "delete-reservation",
+            thread_id="t1",
+            user_id="alice",
+            status="pending",
+            operation_kind=ThreadOperationKind.delete,
+        )
+
+        await store.delete_by_thread("t1", user_id="alice")
+
+        assert "alice-run" not in store._runs
+        assert "alice-run" not in store._runs_by_thread.get("t1", {})
+        assert "delete-reservation" in store._runs_by_thread.get("t1", {})
+        # list_by_thread is run-scoped, so the surviving reservation is not listed
+        # even though the row is still tracked.
+        assert await store.list_by_thread("t1", user_id="alice") == []

backend/tests/test_threads_router.py

@@ -2,7 +2,7 @@ import asyncio
 import re
 from contextlib import asynccontextmanager
 from types import SimpleNamespace
-from unittest.mock import AsyncMock, patch
+from unittest.mock import AsyncMock, MagicMock, patch
 
 import anyio
 import pytest
@@ -434,6 +434,131 @@ def test_delete_thread_route_closes_browser_session(tmp_path):
     manager.close_session.assert_awaited_once_with("thread-browser")
 
 
+def _persistence_cleanup_app(tmp_path, *, run_store, event_store, feedback_repo):
+    app = make_authed_test_app()
+    app.state.run_manager = _ThreadTestRunManager()
+    app.state.run_store = run_store
+    app.state.run_event_store = event_store
+    app.state.feedback_repo = feedback_repo
+    app.include_router(threads.router)
+    return app
+
+
+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
+
+    paths = Paths(tmp_path)
+    user_id = get_effective_user_id()
+    run_store = MagicMock()
+    run_store.delete_by_thread = AsyncMock(return_value=2)
+    event_store = MagicMock()
+    event_store.delete_by_thread = AsyncMock(return_value=4)
+    feedback_repo = MagicMock()
+    feedback_repo.delete_by_thread = AsyncMock(return_value=1)
+
+    app = _persistence_cleanup_app(
+        tmp_path,
+        run_store=run_store,
+        event_store=event_store,
+        feedback_repo=feedback_repo,
+    )
+
+    with patch("app.gateway.routers.threads.get_paths", return_value=paths):
+        with TestClient(app) as client:
+            response = client.delete("/api/threads/thread-cleanup")
+
+    assert response.status_code == 200
+    run_store.delete_by_thread.assert_awaited_once_with("thread-cleanup", user_id=user_id)
+    event_store.delete_by_thread.assert_awaited_once_with("thread-cleanup", user_id=user_id)
+    feedback_repo.delete_by_thread.assert_awaited_once_with("thread-cleanup", user_id=user_id)
+
+
+def test_delete_thread_route_isolates_failing_persistence_cleanup(tmp_path):
+    """One failing store must not stop the remaining cleanup attempts."""
+    paths = Paths(tmp_path)
+    run_store = MagicMock()
+    run_store.delete_by_thread = AsyncMock(side_effect=RuntimeError("simulated cleanup failure"))
+    event_store = MagicMock()
+    event_store.delete_by_thread = AsyncMock(return_value=4)
+    feedback_repo = MagicMock()
+    feedback_repo.delete_by_thread = AsyncMock(return_value=1)
+
+    app = _persistence_cleanup_app(
+        tmp_path,
+        run_store=run_store,
+        event_store=event_store,
+        feedback_repo=feedback_repo,
+    )
+
+    with patch("app.gateway.routers.threads.get_paths", return_value=paths):
+        with TestClient(app) as client:
+            response = client.delete("/api/threads/thread-cleanup")
+
+    assert response.status_code == 200
+    event_store.delete_by_thread.assert_awaited_once()
+    feedback_repo.delete_by_thread.assert_awaited_once()
+
+
+def test_delete_thread_route_tolerates_store_without_bulk_cleanup(tmp_path):
+    """Third-party RunStore implementations without the capability stay usable."""
+    paths = Paths(tmp_path)
+    run_store = SimpleNamespace()  # no delete_by_thread attribute
+    event_store = MagicMock()
+    event_store.delete_by_thread = AsyncMock(return_value=0)
+    feedback_repo = MagicMock()
+    feedback_repo.delete_by_thread = AsyncMock(return_value=0)
+
+    app = _persistence_cleanup_app(
+        tmp_path,
+        run_store=run_store,
+        event_store=event_store,
+        feedback_repo=feedback_repo,
+    )
+
+    with patch("app.gateway.routers.threads.get_paths", return_value=paths):
+        with TestClient(app) as client:
+            response = client.delete("/api/threads/thread-cleanup")
+
+    assert response.status_code == 200
+    event_store.delete_by_thread.assert_awaited_once()
+
+
+class _LegacyRunEventStore:
+    """Event store still on the pre-owner-scope delete contract.
+
+    ``RunEventStore`` is an ABC, but Python never validates override signatures,
+    so this store satisfies it while rejecting the new ``user_id`` keyword.
+    """
+
+    def __init__(self) -> None:
+        self.deleted_threads: list[str] = []
+
+    async def delete_by_thread(self, thread_id: str) -> int:
+        self.deleted_threads.append(thread_id)
+        return 1
+
+
+def test_delete_thread_route_supports_legacy_event_store_delete_signature(tmp_path):
+    """A legacy event store still deletes; the owner kwarg is only passed when accepted."""
+    paths = Paths(tmp_path)
+    event_store = _LegacyRunEventStore()
+
+    app = _persistence_cleanup_app(
+        tmp_path,
+        run_store=SimpleNamespace(),
+        event_store=event_store,
+        feedback_repo=None,
+    )
+
+    with patch("app.gateway.routers.threads.get_paths", return_value=paths):
+        with TestClient(app) as client:
+            response = client.delete("/api/threads/thread-cleanup")
+
+    assert response.status_code == 200
+    assert event_store.deleted_threads == ["thread-cleanup"]
+
+
 def test_delete_thread_route_rejects_invalid_thread_id(tmp_path):
     paths = Paths(tmp_path)