diff --git a/CHANGELOG.md b/CHANGELOG.md index 48a4b57c5a..e4067b7bc7 100644 --- a/CHANGELOG.md +++ b/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 diff --git a/README.md b/README.md index aefea9965d..6750e409c5 100644 --- a/README.md +++ b/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 diff --git a/backend/packages/harness/deerflow/runtime/AGENTS.md b/backend/packages/harness/deerflow/runtime/AGENTS.md index 1029f7bd58..0481bd6b7f 100644 --- a/backend/packages/harness/deerflow/runtime/AGENTS.md +++ b/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 diff --git a/backend/packages/harness/deerflow/runtime/runs/manager.py b/backend/packages/harness/deerflow/runtime/runs/manager.py index 356abf861d..88fe07dd23 100644 --- a/backend/packages/harness/deerflow/runtime/runs/manager.py +++ b/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, ) diff --git a/backend/tests/test_thread_operation_release_recovery.py b/backend/tests/test_thread_operation_release_recovery.py new file mode 100644 index 0000000000..66e05cd0a8 --- /dev/null +++ b/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]