fix(runtime): recover completed thread reservations on admission (#8068)

* fix(runtime): recover completed thread reservations on admission

* Address thread reservation recovery review

* Place reservation recovery note below agent fixes

---------

Co-authored-by: liyinlong <173714821+lilong555@users.noreply.github.com>
Co-authored-by: Willem Jiang <willem.jiang@gmail.com>

CHANGELOG.md

@@ -173,6 +173,12 @@ This section accumulates work toward the **2.3.0** milestone
   model output-length stop. Todo state is preserved so users can continue in a
   later run. ([#7953])
 
+- **runtime:** A thread can accept new runs and mutations after a transient
+  database failure leaves a completed operation's reservation in place. The
+  next request for the thread on the same Gateway worker retries the release,
+  including in single-worker deployments with lease heartbeats disabled.
+  ([#8068](https://github.com/bytedance/deer-flow/pull/8068))
+
 - **channels:** Long Feishu/Lark answers are no longer lost. Card text was
   unbounded and ASCII-escaped, so an answer of about 4,400 CJK characters made
   the card request exceed Feishu's 30 KB limit (`230025`); the final patch and

README.md

@@ -845,6 +845,13 @@ DeerFlow still uses `Forwarded` / `X-Forwarded-*` headers to recover the browser
 >
 > Reconciliation uses an atomic takeover claim that re-checks the lease after candidate selection, so a successful owner renewal wins over orphan recovery and only one reconciler can report a run as recovered. When multiple Gateway workers share the Docker/AIO or E2B sandbox backend, also configure `sandbox.ownership.type: redis`; E2B uses the leases during background startup and periodic reconciliation so duplicate/orphan cleanup cannot terminate a live peer's sandbox.
 
+If a completed thread operation (such as a checkpoint, goal, or artifact write,
+branch, or deletion) fails to release its database reservation, the same Gateway
+worker retries cleanup when it receives a new run or mutation for that thread.
+This also works in single-worker deployments with lease heartbeats disabled.
+If cleanup fails again, admission returns a busy response. Completed requests
+can still replay their saved results by idempotency key.
+
 See [CONTRIBUTING.md](CONTRIBUTING.md) for detailed Docker development guide.
 
 #### Upgrading an existing checkout

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

@@ -15,6 +15,16 @@ namespace/thread bounds, cursor iteration and outage/TTL fallback.
 
 ### Run Admission
 
+Finished thread operations retain their exact run ID and captured owner until
+their durable release succeeds. Retry these releases under the existing
+same-thread admission gate before admitting new work, using the bounded Store
+retry policy. Register completion before cleanup awaits so cancellation remains
+retryable. Live reservations stay exclusive; failed releases return a conflict.
+Replay local and stored idempotent results before cleanup. A conflicting new
+key retries admission once after release; start new leases after gate and cleanup waits.
+The retry ledger holds identities only and schedules no background tasks.
+Coverage: `tests/test_thread_operation_release_recovery.py`.
+
 Goal operation observation preserves producer artifacts when normalizing checkpoint results; page titles never establish failure.
 
 Interactive Queue/Steer inputs belong to the existing RunStore row. Accept only

backend/packages/harness/deerflow/runtime/runs/manager.py

@@ -412,6 +412,10 @@ class RunManager:
         self._admission_locks = AsyncKeyedLockTable[str]()
         self._snapshot_locks = AsyncKeyedLockTable[str]()
         self._pending_admissions: dict[str, dict[str, int]] = {}
+        # Only operations whose bodies have exited enter this ledger. Retain
+        # their captured owner without retaining the completed request task.
+        # A later same-thread admission retries release under its existing gate.
+        self._pending_operation_releases: dict[str, dict[str, str | None]] = {}
         self._store = store
         self._persistence_retry_policy = persistence_retry_policy or PersistenceRetryPolicy()
         self._worker_id = worker_id or _generate_worker_id()
@@ -2295,23 +2299,39 @@ class RunManager:
         execution_fence=None,
     ) -> RunRecord:
         """Atomically admit a normal agent run for a thread."""
-        return await self._admit_thread_operation(
-            thread_id,
-            assistant_id,
-            operation_kind=ThreadOperationKind.run,
-            on_disconnect=on_disconnect,
-            metadata=metadata,
-            kwargs=kwargs,
-            multitask_strategy=multitask_strategy,
-            model_name=model_name,
-            user_id=user_id,
-            idempotency_key=idempotency_key,
-            evidence_origin=evidence_origin,
-            evidence_agent_id=evidence_agent_id,
-            idempotency_request=idempotency_request,
-            run_id=run_id,
-            execution_fence=execution_fence,
-        )
+
+        async def admit() -> RunRecord:
+            return await self._admit_thread_operation(
+                thread_id,
+                assistant_id,
+                operation_kind=ThreadOperationKind.run,
+                on_disconnect=on_disconnect,
+                metadata=metadata,
+                kwargs=kwargs,
+                multitask_strategy=multitask_strategy,
+                model_name=model_name,
+                user_id=user_id,
+                idempotency_key=idempotency_key,
+                evidence_origin=evidence_origin,
+                evidence_agent_id=evidence_agent_id,
+                idempotency_request=idempotency_request,
+                run_id=run_id,
+                execution_fence=execution_fence,
+            )
+
+        try:
+            return await admit()
+        except ConflictError:
+            if idempotency_key is None:
+                raise
+            # Let the Store replay an evicted idempotent result before cleanup.
+            # A new key blocked by a finished reservation gets one recovery
+            # attempt, then returns through the ordinary atomic admission path.
+            async with self._admission_locks.hold(thread_id):
+                if not self._pending_operation_releases.get(thread_id):
+                    raise
+                await self._release_completed_thread_operations(thread_id)
+            return await admit()
 
     async def _close_cancelled_admission(self, record: RunRecord) -> None:
         """Terminalize an unseen replacement and confirm its durable state."""
@@ -2412,7 +2432,6 @@ class RunManager:
         if multitask_strategy not in _supported_strategies:
             raise UnsupportedStrategyError(f"Multitask strategy '{multitask_strategy}' is not yet supported. Supported strategies: {', '.join(_supported_strategies)}")
 
-        lease_expires_at = self._compute_lease_expires_at()
         grace_seconds = self._run_ownership_config.grace_seconds if self._run_ownership_config else 10
 
         if execution_fence is not None and (self._store is None or not _store_accepts_idempotency_parameter(self._store, "execution_fence") or multitask_strategy != "reject"):
@@ -2434,7 +2453,6 @@ class RunManager:
             updated_at=now,
             model_name=model_name,
             owner_worker_id=self._worker_id,
-            lease_expires_at=lease_expires_at,
             idempotency_key=idempotency_key,
             evidence_origin=evidence_origin,
             evidence_agent_id=evidence_agent_id,
@@ -2452,6 +2470,10 @@ class RunManager:
                         existing.idempotency_reused = True
                         return existing
 
+            if idempotency_key is None:
+                await self._release_completed_thread_operations(thread_id)
+            async with self._lock:
+
                 def reuse_idempotent_run(conflict: RunIdempotencyConflict) -> RunRecord:
                     # A locally held record for this key already returned above, so
                     # the conflicting row belongs to a peer or to a run this worker
@@ -2489,6 +2511,10 @@ class RunManager:
 
             # 2) Persist with only the same-thread admission gate held. The
             #    store is the source of truth for cross-process atomicity.
+            # Gate contention and completed-operation release can consume a
+            # whole lease interval. Start the new lease after those waits.
+            lease_expires_at = self._compute_lease_expires_at()
+            record.lease_expires_at = lease_expires_at
             if self._store is not None:
                 if idempotency_request is not None and not _store_accepts_idempotency_parameter(self._store, "idempotency_request"):
                     raise RunIdempotencyUnsupported("The configured RunStore does not support keyed resume idempotency")
@@ -2653,6 +2679,36 @@ class RunManager:
         logger.info("Run created: run_id=%s thread_id=%s", run_id, thread_id)
         return record
 
+    async def _release_completed_thread_operations(self, thread_id: str) -> None:
+        """Retry known finished operations; caller holds the thread admission gate.
+
+        This is request-driven and uses the ordinary bounded Store retry policy.
+        Failed or cancelled deletes retain their exact IDs and captured owners;
+        a missing row is already released, including after an uncertain commit.
+        Live reservations and peer-owned work never enter this local ledger.
+        """
+        pending = self._pending_operation_releases.get(thread_id)
+        if not pending or self._store is None:
+            return
+        for run_id, user_id in tuple(pending.items()):
+            try:
+                await self._call_store_with_retry(
+                    "release thread operation",
+                    run_id,
+                    lambda: self._store.delete_thread_operation(run_id, user_id=user_id),
+                )
+            except Exception as exc:
+                raise ConflictError(f"Thread {thread_id} has a completed operation awaiting cleanup") from exc
+            # Remove both local ownership records only after confirmed release.
+            # There is no await between unregistering and acknowledging the ID.
+            async with self._lock:
+                removed = self._runs.pop(run_id, None)
+                if removed is not None:
+                    self._unindex_run_locked(run_id, removed.thread_id)
+                pending.pop(run_id, None)
+        if not pending:
+            self._pending_operation_releases.pop(thread_id, None)
+
     @asynccontextmanager
     async def reserve_thread_operation(
         self,
@@ -2692,17 +2748,21 @@ class RunManager:
                 raise ConflictError(f"Thread {thread_id} reservation lease was lost") from None
             raise
         finally:
+            if self._store is not None:
+                # Register before the first cleanup await so cancellation also
+                # leaves a retryable release owned by this manager.
+                # The manager's event loop owns the ledger, and each ledger
+                # update is synchronous. The lock in the release path protects
+                # the run registry and its index.
+                self._pending_operation_releases.setdefault(thread_id, {})[record.run_id] = record.user_id
             try:
                 if self._store is not None:
                     try:
-                        await self._call_store_with_retry(
-                            "release thread operation",
-                            record.run_id,
-                            lambda: self._store.delete_thread_operation(record.run_id, user_id=record.user_id),
-                        )
+                        async with self._admission_locks.hold(thread_id):
+                            await self._release_completed_thread_operations(thread_id)
                     except Exception:
                         logger.warning(
-                            "Failed to release persisted thread operation %s; leaving it for orphan reconciliation",
+                            "Failed to release persisted thread operation %s; leaving it for orphan reconciliation or same-thread admission retry",
                             record.run_id,
                             exc_info=True,
                         )

backend/tests/test_thread_operation_release_recovery.py

@@ -0,0 +1,360 @@
+"""Finished thread operations remain recoverable after a failed SQL release."""
+
+import asyncio
+import sqlite3
+from contextlib import asynccontextmanager
+from types import SimpleNamespace
+
+import pytest
+from sqlalchemy.ext.asyncio import async_sessionmaker, create_async_engine
+from starlette.requests import Request
+
+from app.gateway.services import reserve_checkpoint_write
+from deerflow.config.run_ownership_config import RunOwnershipConfig
+from deerflow.persistence.run.model import CompletedRunSnapshotRow, RunChangeClockRow, RunRow
+from deerflow.persistence.run.sql import RunRepository
+from deerflow.runtime import RunManager, RunStatus, ThreadOperationKind
+from deerflow.runtime.runs.manager import ConflictError, PersistenceRetryPolicy
+from deerflow.runtime.user_context import reset_current_user, set_current_user
+
+
+@asynccontextmanager
+async def database(tmp_path, *, run_ownership_config=None):
+    path = tmp_path / "runs.sqlite"
+    engine = create_async_engine(f"sqlite+aiosqlite:///{path}", connect_args={"timeout": 0.01})
+    try:
+        async with engine.begin() as conn:
+            await conn.run_sync(lambda c: RunRow.metadata.create_all(c, tables=[RunRow.__table__, RunChangeClockRow.__table__, CompletedRunSnapshotRow.__table__]))
+        repository = RunRepository(async_sessionmaker(engine, expire_on_commit=False))
+        manager = RunManager(
+            store=repository,
+            persistence_retry_policy=PersistenceRetryPolicy(max_attempts=2, initial_delay=0),
+            run_ownership_config=run_ownership_config,
+        )
+        with sqlite3.connect(path) as blocker:
+            yield manager, repository, blocker
+    finally:
+        await engine.dispose()
+
+
+async def fail_release(manager, blocker, *, kind=ThreadOperationKind.checkpoint_write, body_error=False):
+    try:
+        async with manager.reserve_thread_operation("thread", kind=kind, user_id="alice") as record:
+            blocker.execute("BEGIN IMMEDIATE")
+            if body_error:
+                raise ValueError("original body failure")
+    except ValueError:
+        if not body_error:
+            raise
+    return record
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize("kind", [kind for kind in ThreadOperationKind if kind != ThreadOperationKind.run])
+@pytest.mark.parametrize("strategy", ["reject", "interrupt", "rollback"])
+async def test_unlock_recovers_finished_reservation_before_new_run(tmp_path, kind, strategy):
+    async with database(tmp_path) as (manager, repository, blocker):
+        assert not manager.heartbeat_enabled
+        finished = await fail_release(manager, blocker, kind=kind)
+        blocker.rollback()
+        assert (await repository.get(finished.run_id, user_id="alice"))["status"] == "pending"
+        token = set_current_user(SimpleNamespace(id="different-context"))
+        try:
+            successor = await manager.create_or_reject("thread", user_id="alice", multitask_strategy=strategy)
+        finally:
+            reset_current_user(token)
+        assert successor.status.value == "pending"
+        assert await repository.get(finished.run_id, user_id="alice") is None
+        assert [row["run_id"] for row in await repository.list_inflight()] == [successor.run_id]
+
+
+@pytest.mark.asyncio
+async def test_gateway_checkpoint_boundary_can_retry_after_body_and_release_fail(tmp_path):
+    async with database(tmp_path) as (manager, repository, blocker):
+        app = SimpleNamespace(state=SimpleNamespace(run_manager=manager))
+        request = Request({"type": "http", "app": app})
+        with pytest.raises(ValueError, match="original body failure"):
+            async with reserve_checkpoint_write(request, "thread", user_id="alice"):
+                blocker.execute("BEGIN IMMEDIATE")
+                raise ValueError("original body failure")
+        blocker.rollback()
+        async with reserve_checkpoint_write(request, "thread", user_id="alice"):
+            rows = await repository.list_inflight()
+            assert len(rows) == 1
+            assert rows[0]["operation_kind"] == "checkpoint_write"
+        assert await repository.list_inflight() == []
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize("strategy", ["reject", "interrupt", "rollback"])
+async def test_active_reservation_remains_exclusive(tmp_path, strategy):
+    async with database(tmp_path) as (manager, repository, _blocker):
+        peer = RunManager(store=repository)
+        async with manager.reserve_thread_operation("thread", kind=ThreadOperationKind.checkpoint_write, user_id="alice") as active:
+            for contender in (manager, peer):
+                with pytest.raises(ConflictError):
+                    await contender.create_or_reject("thread", user_id="alice", multitask_strategy=strategy)
+            assert (await repository.get(active.run_id, user_id="alice"))["status"] == "pending"
+        assert await repository.list_inflight() == []
+
+
+@pytest.mark.asyncio
+async def test_recovery_and_concurrent_admission_share_the_thread_gate(tmp_path):
+    async with database(tmp_path) as (manager, repository, blocker):
+        finished = await fail_release(manager, blocker)
+        blocker.rollback()
+        results = await asyncio.gather(*(manager.create_or_reject("thread", user_id="alice") for _ in range(2)), return_exceptions=True)
+        admitted = [result for result in results if not isinstance(result, BaseException)]
+        assert len(admitted) == 1
+        assert sum(isinstance(result, ConflictError) for result in results) == 1
+        assert await repository.get(finished.run_id, user_id="alice") is None
+        assert [row["run_id"] for row in await repository.list_inflight()] == [admitted[0].run_id]
+
+
+@pytest.mark.asyncio
+async def test_persistent_failure_is_bounded_and_does_not_schedule_retries(tmp_path, monkeypatch):
+    async with database(tmp_path) as (manager, repository, blocker):
+        original = repository.delete_thread_operation
+        attempts = []
+
+        async def release(run_id, *, user_id):
+            attempts.append((run_id, user_id))
+            await original(run_id, user_id=user_id)
+
+        monkeypatch.setattr(repository, "delete_thread_operation", release)
+        finished = await fail_release(manager, blocker)
+        tasks_before = asyncio.all_tasks()
+        assert len(attempts) == 2
+        for attempt in range(2):
+            with pytest.raises(ConflictError):
+                await asyncio.wait_for(manager.create_or_reject("thread", user_id="alice"), timeout=1)
+            assert len(attempts) == 4 + 2 * attempt
+        assert asyncio.all_tasks() == tasks_before
+        blocker.rollback()
+        successor = await manager.create_or_reject("thread", user_id="alice")
+        assert await repository.get(finished.run_id, user_id="alice") is None
+        assert successor.status.value == "pending"
+        assert all(owner == "alice" for _run_id, owner in attempts)
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize("phase", ["initial_release", "retry", "after_commit"])
+async def test_cancelled_release_remains_owned_until_a_later_admission(tmp_path, monkeypatch, phase):
+    async with database(tmp_path) as (manager, repository, blocker):
+        if phase == "retry":
+            finished = await fail_release(manager, blocker)
+            blocker.rollback()
+        entered = asyncio.Event()
+        original = repository.delete_thread_operation
+        calls = 0
+
+        async def pause_once(run_id, *, user_id):
+            nonlocal calls
+            calls += 1
+            if calls == 1:
+                if phase == "after_commit":
+                    await original(run_id, user_id=user_id)
+                entered.set()
+                await asyncio.Event().wait()
+            await original(run_id, user_id=user_id)
+
+        monkeypatch.setattr(repository, "delete_thread_operation", pause_once)
+
+        async def operation():
+            nonlocal finished
+            if phase == "retry":
+                return await manager.create_or_reject("thread", user_id="alice")
+            async with manager.reserve_thread_operation("thread", kind=ThreadOperationKind.checkpoint_write, user_id="alice") as finished:
+                pass
+
+        pending = asyncio.create_task(operation())
+        try:
+            await asyncio.wait_for(entered.wait(), timeout=1)
+        finally:
+            pending.cancel()
+            try:
+                await pending
+            except (asyncio.CancelledError, ConflictError):
+                pass
+        successor = await manager.create_or_reject("thread", user_id="alice")
+        assert await repository.get(finished.run_id, user_id="alice") is None
+        assert [row["run_id"] for row in await repository.list_inflight()] == [successor.run_id]
+
+
+@pytest.mark.asyncio
+async def test_shutdown_is_bounded_with_failed_release_and_startup_can_reconcile(tmp_path):
+    async with database(tmp_path) as (manager, repository, blocker):
+        finished = await fail_release(manager, blocker)
+        tasks_before = asyncio.all_tasks()
+        await asyncio.wait_for(manager.shutdown(timeout=0.01), timeout=0.2)
+        assert asyncio.all_tasks() == tasks_before
+        assert (await repository.get(finished.run_id, user_id="alice"))["status"] == "pending"
+        blocker.rollback()
+        restarted = RunManager(store=repository)
+        assert await restarted.reconcile_orphaned_inflight_runs(error="startup recovery") == []
+        successor = await restarted.create_or_reject("thread", user_id="alice")
+        assert successor.status.value == "pending"
+
+
+@pytest.mark.asyncio
+async def test_new_lease_starts_after_slow_release_retry(tmp_path, monkeypatch):
+    from datetime import UTC, datetime, timedelta
+
+    from deerflow.runtime.runs import manager as manager_module
+
+    async with database(tmp_path) as (manager, repository, blocker):
+        # Use heartbeat admission semantics; the background timer is unnecessary
+        # for testing the initial durable deadline after delayed recovery.
+        manager._run_ownership_config = RunOwnershipConfig(heartbeat_enabled=True, lease_seconds=5)
+        finished = await fail_release(manager, blocker)
+        blocker.rollback()
+        original = repository.delete_thread_operation
+        clock_time = datetime.now(UTC)
+
+        class Clock(datetime):
+            @classmethod
+            def now(cls, tz=None):
+                return clock_time.astimezone(tz)
+
+        async def delayed_release(run_id, *, user_id):
+            nonlocal clock_time
+            await asyncio.sleep(0)
+            clock_time += timedelta(seconds=10)
+            await original(run_id, user_id=user_id)
+
+        monkeypatch.setattr(manager_module, "datetime", Clock)
+        monkeypatch.setattr(repository, "delete_thread_operation", delayed_release)
+        try:
+            successor = await manager.create_or_reject("thread", user_id="alice")
+            row = await repository.get(successor.run_id, user_id="alice")
+            deadline = datetime.fromisoformat(row["lease_expires_at"])
+            assert deadline >= clock_time + timedelta(seconds=5)
+            assert successor.lease_expires_at == row["lease_expires_at"]
+            assert await repository.get(finished.run_id, user_id="alice") is None
+        finally:
+            for handle in manager._lease_deadlines.values():
+                handle.cancel()
+
+
+@pytest.mark.asyncio
+async def test_pending_release_conflict_remains_an_im_busy_response(tmp_path):
+    import httpx
+    from langgraph_sdk.errors import ConflictError as SDKConflictError
+
+    from app.channels.manager import _is_thread_busy_error
+
+    async with database(tmp_path) as (manager, _repository, blocker):
+        await fail_release(manager, blocker)
+        with pytest.raises(ConflictError) as failure:
+            await manager.create_or_reject("thread", user_id="alice")
+        detail = str(failure.value)
+        response = httpx.Response(409, request=httpx.Request("POST", "http://gateway/threads/thread/runs"))
+        assert _is_thread_busy_error(SDKConflictError(detail, response=response, body={"detail": detail}))
+        blocker.rollback()
+
+
+@pytest.mark.asyncio
+async def test_completed_key_replays_while_later_reservation_awaits_release(tmp_path):
+    async with database(tmp_path) as (manager, repository, blocker):
+        original = await manager.create_or_reject("thread", user_id="alice", idempotency_key="completed-request")
+        await manager.set_status(original.run_id, RunStatus.success)
+        finished = await fail_release(manager, blocker)
+        try:
+            replay = await manager.create_or_reject("thread", user_id="alice", idempotency_key="completed-request")
+            assert replay is original
+            assert replay.status == RunStatus.success
+            assert replay.idempotency_reused
+            assert (await repository.get(finished.run_id, user_id="alice"))["status"] == "pending"
+        finally:
+            blocker.rollback()
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize("strategy", ["reject", "interrupt", "rollback"])
+async def test_evicted_completed_key_replays_while_later_release_fails(monkeypatch, strategy):
+    from deerflow.runtime.runs.store.memory import MemoryRunStore
+
+    store = MemoryRunStore()
+    manager = RunManager(store=store)
+    original = await manager.create_or_reject("thread", user_id="alice", idempotency_key="completed-request")
+    await manager.set_status(original.run_id, RunStatus.success)
+    await manager.cleanup(original.run_id, delay=0)
+
+    async def unavailable_release(run_id, *, user_id):
+        raise RuntimeError("release unavailable")
+
+    monkeypatch.setattr(store, "delete_thread_operation", unavailable_release)
+    async with manager.reserve_thread_operation("thread", kind=ThreadOperationKind.checkpoint_write, user_id="alice") as finished:
+        pass
+    replay = await manager.create_or_reject("thread", user_id="alice", idempotency_key="completed-request", multitask_strategy=strategy)
+    assert replay.run_id == original.run_id
+    assert replay.status == RunStatus.success
+    assert replay.idempotency_reused and replay.store_only
+    assert (await store.get(finished.run_id))["status"] == "pending"
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize("strategy", ["reject", "interrupt", "rollback"])
+async def test_new_key_recovers_after_release_failure(tmp_path, strategy):
+    async with database(tmp_path) as (manager, repository, blocker):
+        finished = await fail_release(manager, blocker)
+        blocker.rollback()
+        successor = await manager.create_or_reject("thread", user_id="alice", idempotency_key="new-request", multitask_strategy=strategy)
+        replay = await manager.create_or_reject("thread", user_id="alice", idempotency_key="new-request", multitask_strategy=strategy)
+        assert replay.run_id == successor.run_id
+        assert replay.idempotency_reused
+        assert await repository.get(finished.run_id, user_id="alice") is None
+        assert [row["run_id"] for row in await repository.list_inflight()] == [successor.run_id]
+
+
+@pytest.mark.asyncio
+async def test_uncertain_release_keeps_an_active_peer_reservation(tmp_path, monkeypatch):
+    async with database(tmp_path) as (manager, repository, _blocker):
+        original = repository.delete_thread_operation
+        first = True
+
+        async def lose_first_acknowledgement(run_id, *, user_id):
+            nonlocal first
+            await original(run_id, user_id=user_id)
+            if first:
+                first = False
+                raise RuntimeError("release acknowledgement lost")
+
+        monkeypatch.setattr(repository, "delete_thread_operation", lose_first_acknowledgement)
+        async with manager.reserve_thread_operation("thread", kind=ThreadOperationKind.checkpoint_write, user_id="alice") as finished:
+            pass
+        assert await repository.get(finished.run_id, user_id="alice") is None
+        peer = RunManager(store=repository)
+        async with peer.reserve_thread_operation("thread", kind=ThreadOperationKind.checkpoint_write, user_id="alice") as active:
+            with pytest.raises(ConflictError):
+                await manager.create_or_reject("thread", user_id="alice")
+            assert [row["run_id"] for row in await repository.list_inflight()] == [active.run_id]
+        assert await repository.list_inflight() == []
+
+
+@pytest.mark.asyncio
+async def test_release_retry_after_peer_takeover_preserves_its_new_reservation(tmp_path):
+    ownership = RunOwnershipConfig(heartbeat_enabled=True, grace_seconds=0)
+    async with database(tmp_path, run_ownership_config=ownership) as (manager, repository, blocker):
+        peer = RunManager(store=repository, run_ownership_config=ownership)
+        finished = await fail_release(manager, blocker)
+        blocker.rollback()
+
+        assert await peer.reconcile_orphaned_inflight_runs(error="peer takeover") == []
+        pending = await repository.get(finished.run_id, user_id="alice")
+        assert pending["status"] == "pending"
+        assert pending["lease_expires_at"] is not None
+
+        # Expire the durable lease without waiting for the heartbeat interval.
+        blocker.execute("UPDATE runs SET lease_expires_at = datetime('now', '-1 minute') WHERE run_id = ?", (finished.run_id,))
+        blocker.commit()
+        assert await peer.reconcile_orphaned_inflight_runs(error="peer takeover") == []
+        reclaimed = await repository.get(finished.run_id, user_id="alice")
+        assert reclaimed["status"] == "error"
+        assert reclaimed["error"] == "peer takeover"
+
+        async with peer.reserve_thread_operation("thread", kind=ThreadOperationKind.checkpoint_write, user_id="alice") as active:
+            with pytest.raises(ConflictError):
+                await manager.create_or_reject("thread", user_id="alice")
+            assert await repository.get(finished.run_id, user_id="alice") is None
+            assert [row["run_id"] for row in await repository.list_inflight()] == [active.run_id]