fix(runtime): serialize checkpoint writes with active runs (#4437)

* fix(runtime): serialize checkpoint writes with active runs

* fix(runtime): address checkpoint reservation reviews

* fix(runtime): address reservation race reviews

* fix(runtime): refine reservation conflict semantics

README.md

@@ -756,11 +756,11 @@ Supported commands:
 
 After each Gateway-backed run, DeerFlow evaluates the visible conversation against the active goal with a non-thinking evaluator model. The evaluator must return a typed blocker (`missing_evidence`, `needs_user_input`, `run_failed`, `external_wait`, or `goal_not_met_yet`) plus visible evidence. DeerFlow only injects a hidden continuation when the latest assistant turn is durably checkpointed, the blocker is `goal_not_met_yet`, the thread did not change during evaluation, and the no-progress breaker has not fired. The safety cap defaults to 8 hidden continuations, and repeated identical non-progress evaluations stop after 2 attempts. `/goal clear` and any user-authored new input win over queued continuations. When the goal is satisfied, DeerFlow clears it automatically and publishes the updated thread state.
 
-The Web UI shows the active goal above the composer. The same command is available from the TUI and supported IM channels. In the Web UI and supported IM channels, setting `/goal <completion condition>` also starts a run with the condition as the task; status and clear commands only manage goal state.
+The Web UI shows the active goal above the composer. The same command is available from the TUI and supported IM channels. In the Web UI and supported IM channels, setting `/goal <completion condition>` also starts a run with the condition as the task; status and clear commands only manage goal state. Setting or clearing a goal is rejected while that thread has a run in flight, including a run owned by another Gateway worker, so the goal checkpoint cannot branch away from an active run's checkpoint lineage.
 
 ### Manual Context Compaction
 
-Use `/compact` in the Web UI composer to summarize older context for the current thread. DeerFlow keeps the full chat visible, but future model calls use the compacted summary plus recent messages. The command is ignored when there is not enough history to compact, and it is blocked while the thread has a run in flight.
+Use `/compact` in the Web UI composer to summarize older context for the current thread. DeerFlow keeps the full chat visible, but future model calls use the compacted summary plus recent messages. The command is ignored when there is not enough history to compact, and it is blocked while the thread has a run in flight, including when that run is owned by another Gateway worker. If a multi-worker reservation loses its lease, DeerFlow cancels the checkpoint writer before the replacing run proceeds and returns a retryable conflict after cleanup. Thread-title edits are serialized through the same state-write boundary and show a conflict without closing the rename dialog when a run is active.
 
 ### Sub-Agents
 

backend/AGENTS.md

@@ -453,6 +453,8 @@ metadata only.
 - `cancel()` returns a :class:`~deerflow.runtime.CancelOutcome` enum: `cancelled` (local cancel), `taken_over` (non-owning worker claimed the run because the owner's lease expired — marks it as `error`), `lease_valid_elsewhere` (owner's lease is still alive — caller should return 409 + `Retry-After`), `not_active_locally` (heartbeat disabled, preserving the old 409 path), `not_cancellable` (terminal state), or `unknown` (not found in memory or store). `create_or_reject(..., multitask_strategy="interrupt"|"rollback")` persists interrupted status through `RunStore.update_status()`, matching normal `set_status()` transitions.
 - Store-only hydrated runs are readable history. In multi-worker mode with heartbeat enabled, cancel on a store-only run can take over (mark `error`) when the owner's lease has expired past the grace window; otherwise it fails with 409 + `Retry-After`. In single-worker mode (heartbeat off), store-only runs still return 409.
 - Startup/orphan reconciliation must claim stale active rows with `RunStore.claim_for_takeover()`, not a plain `update_status()`. The final claim re-checks `status` and lease expiry atomically, so a heartbeat renewal between the candidate scan and the recovery write keeps the run active.
+- Run admission and independent checkpoint writes are first-class thread operations. `runs.operation_kind` distinguishes user-visible `run` rows from internal `checkpoint_write` reservations, while every active kind shares the existing 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 checkpoint 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 checkpoint 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 checkpoint 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.
 - `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).
 - 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.
 - Thread-scoped run creation accepts `checkpoint` / `checkpoint_id`; Gateway validates the checkpoint belongs to the request thread before writing `checkpoint_id` / `checkpoint_ns` into `config.configurable` for LangGraph branching.
@@ -840,6 +842,7 @@ This invokes `alembic revision --autogenerate` against the live ORM models. Revi
 - `migrations/versions/0002_runs_token_usage.py` — fixes issue #3682
 - `migrations/versions/0004_run_ownership.py` — `runs` multi-worker ownership + the `uq_runs_thread_active` partial unique index, with a `_dedupe_active_runs_per_thread()` pre-step so `CREATE UNIQUE INDEX` cannot fail on a field DB that already has duplicate active rows per thread
 - `migrations/versions/0007_scheduled_run_active_index.py` — the `uq_scheduled_task_run_active` partial unique index (at most one queued/running `scheduled_task_runs` row per `task_id`), with a `_dedupe_active_scheduled_runs_per_task()` pre-step (keeps the newest active row per task, supersedes the rest to `interrupted` with an explanatory `error` + `finished_at`) mirroring 0004; chains after `0006_agents`
+- `migrations/versions/0008_thread_operation_kind.py` — adds `runs.operation_kind` for durable non-run thread reservations; chains after `0007_scheduled_run_active_index`
 - `persistence/bootstrap.py` — `bootstrap_schema(engine, backend=...)`, the three-branch decision + locking
 - Tests: `tests/test_persistence_bootstrap.py` (branches), `tests/test_persistence_bootstrap_concurrency.py` (concurrency), `tests/test_persistence_bootstrap_regression.py` (issue #3682), `tests/test_persistence_migrations_env.py` (filter), `tests/blocking_io/test_persistence_bootstrap.py` (asyncio.to_thread anchor), `tests/test_migration_0004_run_ownership_dedupe.py` + `tests/test_migration_0007_scheduled_run_active_dedupe.py` (dedupe-before-unique-index pre-steps)
 
@@ -1135,9 +1138,10 @@ Automatic conversation summarization when approaching token limits:
 - Manual compaction uses `POST /api/threads/{id}/compact`, reuses the same
   `DeerFlowSummarizationMiddleware`, writes a new checkpoint with updated
   `messages` and `summary_text`, and bumps only those channel versions.
-  The route shares the per-thread serialization gate used by `/goal` writes
+  The route uses the shared `reserve_checkpoint_write()` boundary (also used by
-  and run admission so compaction cannot race with goal updates or runs that
+  manual state updates). Its short-lived `checkpoint_write` thread operation
-  read/write checkpoints.
+  shares the durable active-thread uniqueness constraint with run admission,
+  preventing either worker-local or cross-worker checkpoint-write races.
 
 See [docs/summarization.md](docs/summarization.md) for details.
 

backend/app/gateway/routers/console.py

@@ -272,7 +272,9 @@ async def console_stats(request: Request) -> ConsoleStatsResponse:
     """Return the dashboard's headline counters."""
     sf = _session_factory_or_503()
     user_id = await get_current_user(request)
-    run_where = (RunRow.user_id == user_id,) if user_id else ()
+    run_where = (RunRow.operation_kind == "run",)
+    if user_id:
+        run_where += (RunRow.user_id == user_id,)
     thread_where = (ThreadMetaRow.user_id == user_id,) if user_id else ()
 
     pricing = _build_pricing_map()
@@ -347,7 +349,14 @@ async def console_runs(
     sf = _session_factory_or_503()
     user_id = await get_current_user(request)
 
-    stmt = select(RunRow, ThreadMetaRow.display_name).join(ThreadMetaRow, ThreadMetaRow.thread_id == RunRow.thread_id, isouter=True).order_by(RunRow.created_at.desc(), RunRow.run_id.desc()).limit(limit + 1).offset(offset)
+    stmt = (
+        select(RunRow, ThreadMetaRow.display_name)
+        .join(ThreadMetaRow, ThreadMetaRow.thread_id == RunRow.thread_id, isouter=True)
+        .where(RunRow.operation_kind == "run")
+        .order_by(RunRow.created_at.desc(), RunRow.run_id.desc())
+        .limit(limit + 1)
+        .offset(offset)
+    )
     if user_id:
         stmt = stmt.where(RunRow.user_id == user_id)
     if status:
@@ -415,7 +424,7 @@ async def console_usage(
     start_local = today_local - timedelta(days=days - 1)
     window_start_utc = datetime.combine(start_local, time.min, tzinfo=UTC) - tz_delta
 
-    stmt = select(RunRow).where(RunRow.created_at >= window_start_utc)
+    stmt = select(RunRow).where(RunRow.operation_kind == "run", RunRow.created_at >= window_start_utc)
     if user_id:
         stmt = stmt.where(RunRow.user_id == user_id)
 

backend/app/gateway/routers/threads.py

@@ -32,13 +32,14 @@ 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
 from app.gateway.internal_auth import get_trusted_internal_owner_user_id
 from app.gateway.services import (
     build_checkpoint_state_accessor,
     build_checkpoint_state_mutation_accessor,
     build_thread_checkpoint_state_accessor,
     build_thread_checkpoint_state_mutation_accessor,
+    reserve_checkpoint_write,
 )
 from app.gateway.utils import sanitize_log_param
 from deerflow.agents.thread_state import THREAD_STATE_REDUCER_FIELDS
@@ -57,11 +58,11 @@ from deerflow.runtime.goal import (
     DEFAULT_MAX_GOAL_CONTINUATIONS,
     build_goal_state,
     ensure_thread_checkpoint,
-    goal_thread_lock,
     read_thread_goal,
     write_thread_goal,
 )
 from deerflow.runtime.journal import build_branch_history_seed_events
+from deerflow.runtime.runs.manager import ConflictError
 from deerflow.runtime.runs.worker import valid_duration_entry
 from deerflow.runtime.user_context import get_effective_user_id
 from deerflow.utils.file_io import run_file_io
@@ -1053,13 +1054,17 @@ async def set_thread_goal(thread_id: str, body: ThreadGoalRequest, request: Requ
     this endpoint creates the missing thread checkpoint on demand.
     """
     checkpointer = get_checkpointer(request)
-    await _ensure_thread_for_goal(thread_id, request)
     try:
         goal = build_goal_state(body.objective, max_continuations=body.max_continuations)
-        async with goal_thread_lock(thread_id):
+        async with reserve_checkpoint_write(request, thread_id, user_id=get_effective_user_id()):
+            await _ensure_thread_for_goal(thread_id, request)
             await write_thread_goal(checkpointer, thread_id, goal, as_node="goal", create_if_missing=True)
     except ValueError as exc:
         raise HTTPException(status_code=422, detail=str(exc)) from exc
+    except ConflictError:
+        raise HTTPException(status_code=409, detail="Thread has a run in flight. Set the goal after the run finishes.") from None
+    except HTTPException:
+        raise
     except Exception:
         logger.exception("Failed to set goal for thread %s", sanitize_log_param(thread_id))
         raise HTTPException(status_code=500, detail="Failed to set thread goal") from None
@@ -1072,8 +1077,10 @@ async def clear_thread_goal(thread_id: str, request: Request) -> ThreadGoalRespo
     """Clear the active goal for a thread."""
     checkpointer = get_checkpointer(request)
     try:
-        async with goal_thread_lock(thread_id):
+        async with reserve_checkpoint_write(request, thread_id, user_id=get_effective_user_id()):
             await write_thread_goal(checkpointer, thread_id, None, as_node="goal")
+    except ConflictError:
+        raise HTTPException(status_code=409, detail="Thread has a run in flight. Clear the goal after the run finishes.") from None
     except LookupError:
         return ThreadGoalResponse(goal=None)
     except Exception:
@@ -1099,7 +1106,6 @@ def _thread_compact_response(result: ThreadCompactionResult) -> ThreadCompactRes
 @require_permission("threads", "write", owner_check=True, require_existing=True)
 async def compact_thread(thread_id: str, body: ThreadCompactRequest, request: Request) -> ThreadCompactResponse:
     """Manually summarize old thread context while preserving the visible history."""
-    run_manager = get_run_manager(request)
     # Compaction writes only base-schema channels (messages + summary_text);
     # every other channel — including middleware-contributed ones — is carried
     # forward by checkpoint fork inheritance, so the base-schema mutation
@@ -1114,9 +1120,7 @@ async def compact_thread(thread_id: str, body: ThreadCompactRequest, request: Re
         raise _checkpoint_mode_http_error(exc, thread_id) from exc
     keep = body.keep.to_tuple() if body.keep is not None else None
     try:
-        async with goal_thread_lock(thread_id):
+        async with reserve_checkpoint_write(request, thread_id, user_id=get_effective_user_id()):
-            if await run_manager.has_inflight(thread_id):
-                raise HTTPException(status_code=409, detail="Thread has a run in flight. Compact after the run finishes.")
             result = await compact_thread_context(
                 accessor,
                 thread_id,
@@ -1125,6 +1129,8 @@ async def compact_thread(thread_id: str, body: ThreadCompactRequest, request: Re
                 user_id=get_effective_user_id(),
                 agent_name=body.agent_name,
             )
+    except ConflictError:
+        raise HTTPException(status_code=409, detail="Thread has a run in flight. Compact after the run finishes.") from None
     except _CHECKPOINT_MODE_ERRORS as exc:
         raise _checkpoint_mode_http_error(exc, thread_id) from exc
     except ContextCompactionDisabled:
@@ -1232,12 +1238,15 @@ async def update_thread_state(thread_id: str, body: ThreadStateUpdateRequest, re
         reducer_fields = THREAD_STATE_REDUCER_FIELDS
     updates = {key: Overwrite(value) if key in reducer_fields else value for key, value in values.items()}
     try:
-        updated_config = await accessor.aupdate(
+        async with reserve_checkpoint_write(request, thread_id, user_id=get_effective_user_id()):
-            read_config,
+            updated_config = await accessor.aupdate(
-            updates,
+                read_config,
-            as_node=mutation_node,
+                updates,
-        )
+                as_node=mutation_node,
-        snapshot = await accessor.aget(updated_config)
+            )
+            snapshot = await accessor.aget(updated_config)
+    except ConflictError:
+        raise HTTPException(status_code=409, detail="Thread has a run in flight. Update state after the run finishes.") from None
     except _CHECKPOINT_MODE_ERRORS as exc:
         raise _checkpoint_mode_http_error(exc, thread_id) from exc
     except Exception:

backend/app/gateway/services.py

@@ -11,7 +11,8 @@ import asyncio
 import json
 import logging
 import re
-from collections.abc import Mapping
+from collections.abc import AsyncIterator, Mapping
+from contextlib import asynccontextmanager
 from types import SimpleNamespace
 from typing import Any
 
@@ -44,6 +45,7 @@ from deerflow.runtime import (
     RunRecord,
     RunStatus,
     StreamBridge,
+    ThreadOperationKind,
     UnsupportedStrategyError,
     build_state_mutation_graph,
     run_agent,
@@ -64,6 +66,25 @@ from deerflow.utils.messages import ORIGINAL_USER_CONTENT_KEY
 
 logger = logging.getLogger(__name__)
 
+
+@asynccontextmanager
+async def reserve_checkpoint_write(
+    request: Request,
+    thread_id: str,
+    *,
+    user_id: str | None = None,
+) -> AsyncIterator[None]:
+    """Serialize an out-of-run checkpoint writer against all thread operations."""
+    run_manager = get_run_manager(request)
+    async with goal_thread_lock(thread_id):
+        async with run_manager.reserve_thread_operation(
+            thread_id,
+            kind=ThreadOperationKind.checkpoint_write,
+            user_id=user_id,
+        ):
+            yield
+
+
 _TERMINAL_RUN_STATUSES = {
     RunStatus.success,
     RunStatus.error,

backend/packages/harness/deerflow/persistence/migrations/versions/0008_thread_operation_kind.py

@@ -0,0 +1,36 @@
+"""thread operation kind.
+
+Revision ID: 0008_thread_operation_kind
+Revises: 0007_scheduled_run_active_index
+Create Date: 2026-07-24
+"""
+
+from __future__ import annotations
+
+from collections.abc import Sequence
+
+import sqlalchemy as sa
+from alembic import op
+
+revision: str = "0008_thread_operation_kind"
+down_revision: str | Sequence[str] | None = "0007_scheduled_run_active_index"
+branch_labels: str | Sequence[str] | None = None
+depends_on: str | Sequence[str] | None = None
+
+
+def upgrade() -> None:
+    from deerflow.persistence.migrations._helpers import safe_add_column
+
+    safe_add_column(
+        "runs",
+        sa.Column(
+            "operation_kind",
+            sa.String(length=32),
+            nullable=False,
+            server_default=sa.text("'run'"),
+        ),
+    )
+
+
+def downgrade() -> None:
+    op.drop_column("runs", "operation_kind")

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

@@ -19,6 +19,7 @@ class RunRow(Base):
     user_id: Mapped[str | None] = mapped_column(String(64), index=True)
     status: Mapped[str] = mapped_column(String(20), default="pending")
     # "pending" | "running" | "success" | "error" | "timeout" | "interrupted"
+    operation_kind: Mapped[str] = mapped_column(String(32), nullable=False, default="run", server_default=text("'run'"))
 
     model_name: Mapped[str | None] = mapped_column(String(128))
     multitask_strategy: Mapped[str] = mapped_column(String(20), default="reject")

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

@@ -92,6 +92,7 @@ class RunRepository(RunStore):
         user_id: str | None | _AutoSentinel = AUTO,
         model_name: str | None = None,
         status="pending",
+        operation_kind: str = "run",
         multitask_strategy="reject",
         metadata=None,
         kwargs=None,
@@ -118,6 +119,7 @@ class RunRepository(RunStore):
             "user_id": resolved_user_id,
             "model_name": self._normalize_model_name(model_name),
             "status": status,
+            "operation_kind": operation_kind,
             "multitask_strategy": multitask_strategy,
             "metadata_json": self._safe_json(metadata) or {},
             "kwargs_json": self._safe_json(kwargs) or {},
@@ -160,7 +162,7 @@ class RunRepository(RunStore):
         limit=100,
     ):
         resolved_user_id = resolve_user_id(user_id, method_name="RunRepository.list_by_thread")
-        stmt = select(RunRow).where(RunRow.thread_id == thread_id)
+        stmt = select(RunRow).where(RunRow.thread_id == thread_id, RunRow.operation_kind == "run")
         if resolved_user_id is not None:
             stmt = stmt.where(RunRow.user_id == resolved_user_id)
         stmt = stmt.order_by(RunRow.created_at.desc()).limit(limit)
@@ -178,6 +180,7 @@ class RunRepository(RunStore):
         source = RunRow.metadata_json["regenerate_from_run_id"].as_string()
         stmt = select(source).where(
             RunRow.thread_id == thread_id,
+            RunRow.operation_kind == "run",
             RunRow.status == "success",
             source.is_not(None),
             source != "",
@@ -198,7 +201,7 @@ class RunRepository(RunStore):
         if not run_ids:
             return {}
         resolved_user_id = resolve_user_id(user_id, method_name="RunRepository.get_many_by_thread")
-        stmt = select(RunRow).where(RunRow.thread_id == thread_id, RunRow.run_id.in_(run_ids))
+        stmt = select(RunRow).where(RunRow.thread_id == thread_id, RunRow.operation_kind == "run", RunRow.run_id.in_(run_ids))
         if resolved_user_id is not None:
             stmt = stmt.where(RunRow.user_id == resolved_user_id)
         async with self._sf() as session:
@@ -242,6 +245,10 @@ class RunRepository(RunStore):
             await session.delete(row)
             await session.commit()
 
+    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)
+
     async def list_pending(self, *, before=None):
         if before is None:
             before_dt = datetime.now(UTC)
@@ -249,7 +256,7 @@ class RunRepository(RunStore):
             before_dt = before
         else:
             before_dt = datetime.fromisoformat(before)
-        stmt = select(RunRow).where(RunRow.status == "pending", RunRow.created_at <= before_dt).order_by(RunRow.created_at.asc())
+        stmt = select(RunRow).where(RunRow.operation_kind == "run", RunRow.status == "pending", RunRow.created_at <= before_dt).order_by(RunRow.created_at.asc())
         async with self._sf() as session:
             result = await session.execute(stmt)
             return [self._row_to_dict(r) for r in result.scalars()]
@@ -378,6 +385,7 @@ class RunRepository(RunStore):
         statuses = ("success", "error", "running") if include_active else ("success", "error")
         _completed = RunRow.status.in_(statuses)
         _thread = RunRow.thread_id == thread_id
+        _run_operation = RunRow.operation_kind == "run"
 
         stmt = select(
             RunRow.model_name,
@@ -388,7 +396,7 @@ class RunRepository(RunStore):
             RunRow.subagent_tokens,
             RunRow.middleware_tokens,
             RunRow.token_usage_by_model,
-        ).where(_thread, _completed)
+        ).where(_thread, _run_operation, _completed)
 
         async with self._sf() as session:
             rows = (await session.execute(stmt)).all()
@@ -510,13 +518,14 @@ class RunRepository(RunStore):
             result = await session.execute(stmt)
             return [self._row_to_dict(r) for r in result.scalars()]
 
-    async def create_run_atomic(
+    async def create_thread_operation_atomic(
         self,
         run_id: str,
         *,
         thread_id: str,
         owner_worker_id: str,
         lease_expires_at: str | None,
+        operation_kind: str = "run",
         multitask_strategy: str = "reject",
         assistant_id: str | None = None,
         user_id: str | None = None,
@@ -541,7 +550,7 @@ class RunRepository(RunStore):
         """
         from deerflow.runtime.runs.manager import ConflictError
 
-        resolved_user_id = resolve_user_id(user_id or AUTO, method_name="RunRepository.create_run_atomic")
+        resolved_user_id = resolve_user_id(user_id or AUTO, method_name="RunRepository.create_thread_operation_atomic")
         now = datetime.now(UTC)
         created = datetime.fromisoformat(created_at) if created_at else now
         lease_dt = datetime.fromisoformat(lease_expires_at) if lease_expires_at else None
@@ -553,6 +562,7 @@ class RunRepository(RunStore):
             "user_id": resolved_user_id,
             "model_name": self._normalize_model_name(model_name),
             "status": "pending",
+            "operation_kind": operation_kind,
             "multitask_strategy": multitask_strategy,
             "metadata_json": self._safe_json(metadata) or {},
             "kwargs_json": self._safe_json(kwargs) or {},
@@ -576,6 +586,7 @@ class RunRepository(RunStore):
                 )
                 result = await session.execute(stmt)
                 for row in result.scalars():
+                    lease_expired = False
                     if row.lease_expires_at is not None:
                         # SQLite drops tzinfo on read despite
                         # ``DateTime(timezone=True)`` (see ``_row_to_dict``).
@@ -588,6 +599,7 @@ class RunRepository(RunStore):
                         row_lease = row.lease_expires_at
                         if row_lease.tzinfo is None:
                             row_lease = row_lease.replace(tzinfo=UTC)
+                        lease_expired = row_lease < cutoff
                         if row_lease >= cutoff and row.owner_worker_id != owner_worker_id:
                             # Live run owned by another worker — we cannot
                             # interrupt it and the partial unique index would
@@ -595,6 +607,8 @@ class RunRepository(RunStore):
                             # ConflictError so the caller gets a clean signal
                             # instead of a retry loop on IntegrityError.
                             raise ConflictError(f"Thread {thread_id} already has an active run owned by another worker")
+                    if row.operation_kind != "run" and not lease_expired:
+                        raise ConflictError(f"Thread {thread_id} has an active checkpoint write")
                     row.status = "interrupted"
                     row.error = "Cancelled by newer run"
                     row.owner_worker_id = owner_worker_id

backend/packages/harness/deerflow/runtime/__init__.py

@@ -7,7 +7,7 @@ directly from ``deerflow.runtime``.
 
 from .checkpoint_state import CheckpointStateAccessor, build_state_mutation_graph
 from .checkpointer import checkpointer_context, get_checkpointer, make_checkpointer, reset_checkpointer
-from .runs import ORPHAN_RECOVERY_STOP_REASON, STARTUP_ORPHAN_RECOVERY_ERROR, CancelOutcome, ConflictError, DisconnectMode, RunContext, RunManager, RunRecord, RunStatus, UnsupportedStrategyError, run_agent
+from .runs import ORPHAN_RECOVERY_STOP_REASON, STARTUP_ORPHAN_RECOVERY_ERROR, CancelOutcome, ConflictError, DisconnectMode, RunContext, RunManager, RunRecord, RunStatus, ThreadOperationKind, UnsupportedStrategyError, run_agent
 from .serialization import serialize, serialize_channel_values, serialize_channel_values_for_api, serialize_lc_object, serialize_messages_tuple, strip_data_url_image_blocks
 from .store import get_store, make_store, reset_store, store_context
 
@@ -34,6 +34,7 @@ __all__ = [
     "RunManager",
     "RunRecord",
     "RunStatus",
+    "ThreadOperationKind",
     "STARTUP_ORPHAN_RECOVERY_ERROR",
     "UnsupportedStrategyError",
     "run_agent",

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

@@ -1,7 +1,7 @@
 """Run lifecycle management for LangGraph Platform API compatibility."""
 
 from .manager import ORPHAN_RECOVERY_STOP_REASON, STARTUP_ORPHAN_RECOVERY_ERROR, CancelOutcome, ConflictError, RunManager, RunRecord, UnsupportedStrategyError
-from .schemas import DisconnectMode, RunStatus
+from .schemas import DisconnectMode, RunStatus, ThreadOperationKind
 from .worker import RunContext, run_agent
 
 __all__ = [
@@ -13,6 +13,7 @@ __all__ = [
     "RunManager",
     "RunRecord",
     "RunStatus",
+    "ThreadOperationKind",
     "STARTUP_ORPHAN_RECOVERY_ERROR",
     "UnsupportedStrategyError",
     "run_agent",

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

@@ -7,7 +7,8 @@ import logging
 import socket
 import sqlite3
 import uuid
-from collections.abc import Awaitable, Callable
+from collections.abc import AsyncIterator, Awaitable, Callable
+from contextlib import asynccontextmanager
 from dataclasses import dataclass, field
 from datetime import UTC, datetime, timedelta
 from enum import StrEnum
@@ -19,7 +20,7 @@ from deerflow.runtime.user_context import AUTO, _AutoSentinel, resolve_user_id
 from deerflow.utils.time import is_lease_expired
 from deerflow.utils.time import now_iso as _now_iso
 
-from .schemas import DisconnectMode, RunStatus
+from .schemas import DisconnectMode, RunStatus, ThreadOperationKind
 
 if TYPE_CHECKING:
     from deerflow.config.run_ownership_config import RunOwnershipConfig
@@ -158,6 +159,7 @@ class RunRecord:
     assistant_id: str | None
     status: RunStatus
     on_disconnect: DisconnectMode
+    operation_kind: ThreadOperationKind = ThreadOperationKind.run
     multitask_strategy: str = "reject"
     metadata: dict = field(default_factory=dict)
     kwargs: dict = field(default_factory=dict)
@@ -260,6 +262,7 @@ class RunManager:
             "thread_id": record.thread_id,
             "assistant_id": record.assistant_id,
             "status": record.status.value,
+            "operation_kind": record.operation_kind.value,
             "multitask_strategy": record.multitask_strategy,
             "metadata": record.metadata or {},
             "kwargs": record.kwargs or {},
@@ -398,6 +401,7 @@ class RunManager:
             assistant_id=row.get("assistant_id"),
             status=RunStatus(row.get("status") or RunStatus.pending.value),
             on_disconnect=DisconnectMode(row.get("on_disconnect") or DisconnectMode.cancel.value),
+            operation_kind=ThreadOperationKind(row.get("operation_kind") or ThreadOperationKind.run.value),
             multitask_strategy=row.get("multitask_strategy") or "reject",
             metadata=row.get("metadata") or {},
             kwargs=row.get("kwargs") or {},
@@ -493,7 +497,7 @@ class RunManager:
 
         Note: this method assumes no active run exists for the thread. It
         persists via ``store.put`` (upsert) rather than the atomic
-        ``create_run_atomic`` primitive, so a concurrent insert for the
+        ``create_thread_operation_atomic`` primitive, so a concurrent insert for the
         same thread will hit the partial unique index and surface as a
         raw ``IntegrityError`` instead of a ``ConflictError``. Production
         callers should use :meth:`create_or_reject`.
@@ -586,7 +590,7 @@ class RunManager:
             limit: Maximum number of runs to return.
         """
         async with self._lock:
-            memory_records = self._thread_records_locked(thread_id)
+            memory_records = [record for record in self._thread_records_locked(thread_id) if record.operation_kind == ThreadOperationKind.run]
         if self._store is None:
             return sorted(memory_records, key=lambda r: r.created_at, reverse=True)[:limit]
         records_by_id = {record.run_id: record for record in memory_records}
@@ -626,7 +630,7 @@ class RunManager:
         """
         resolved_user_id = resolve_user_id(user_id, method_name="RunManager.list_successful_regenerate_sources")
         async with self._lock:
-            memory_records = [record for record in self._thread_records_locked(thread_id) if resolved_user_id is None or record.user_id == resolved_user_id]
+            memory_records = [record for record in self._thread_records_locked(thread_id) if record.operation_kind == ThreadOperationKind.run and (resolved_user_id is None or record.user_id == resolved_user_id)]
 
         sources = set(await self._store.list_successful_regenerate_sources(thread_id, user_id=resolved_user_id)) if self._store is not None else set()
         # _thread_records_locked preserves the insertion order of the thread
@@ -654,7 +658,9 @@ class RunManager:
             return {}
         resolved_user_id = resolve_user_id(user_id, method_name="RunManager.get_many_by_thread")
         async with self._lock:
-            records_by_id = {record.run_id: record for record in self._thread_records_locked(thread_id) if record.run_id in run_ids and (resolved_user_id is None or record.user_id == resolved_user_id)}
+            records_by_id = {
+                record.run_id: record for record in self._thread_records_locked(thread_id) if record.operation_kind == ThreadOperationKind.run and record.run_id in run_ids and (resolved_user_id is None or record.user_id == resolved_user_id)
+            }
         if self._store is None:
             return records_by_id
 
@@ -943,6 +949,32 @@ class RunManager:
         multitask_strategy: str = "reject",
         model_name: str | None = None,
         user_id: str | None = 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,
+        )
+
+    async def _admit_thread_operation(
+        self,
+        thread_id: str,
+        assistant_id: str | None = None,
+        *,
+        operation_kind: ThreadOperationKind,
+        on_disconnect: DisconnectMode = DisconnectMode.cancel,
+        metadata: dict | None = None,
+        kwargs: dict | None = None,
+        multitask_strategy: str = "reject",
+        model_name: str | None = None,
+        user_id: str | None = None,
     ) -> RunRecord:
         """Atomically check for inflight runs and create a new one.
 
@@ -975,6 +1007,7 @@ class RunManager:
             assistant_id=assistant_id,
             status=RunStatus.pending,
             on_disconnect=on_disconnect,
+            operation_kind=operation_kind,
             multitask_strategy=multitask_strategy,
             metadata=metadata or {},
             kwargs=kwargs or {},
@@ -991,6 +1024,9 @@ class RunManager:
             #    store's partial unique index below).
             local_inflight = [r for r in self._thread_records_locked(thread_id) if r.status in (RunStatus.pending, RunStatus.running) or r.finalizing]
 
+            if multitask_strategy in ("interrupt", "rollback") and any(record.operation_kind != ThreadOperationKind.run for record in local_inflight):
+                raise ConflictError(f"Thread {thread_id} has an active checkpoint write")
+
             if multitask_strategy == "reject" and local_inflight:
                 raise ConflictError(f"Thread {thread_id} already has an active run")
 
@@ -1008,13 +1044,14 @@ class RunManager:
                 if multitask_strategy == "reject":
                     try:
                         await self._call_store_with_retry(
-                            "create_run_atomic",
+                            "create_thread_operation_atomic",
                             run_id,
-                            lambda: self._store.create_run_atomic(
+                            lambda: self._store.create_thread_operation_atomic(
                                 run_id=run_id,
                                 thread_id=thread_id,
                                 owner_worker_id=self._worker_id,
                                 lease_expires_at=lease_expires_at,
+                                operation_kind=operation_kind.value,
                                 multitask_strategy="reject",
                                 assistant_id=assistant_id,
                                 user_id=user_id,
@@ -1039,13 +1076,14 @@ class RunManager:
                     for attempt in range(max_retries):
                         try:
                             await self._call_store_with_retry(
-                                "create_run_atomic",
+                                "create_thread_operation_atomic",
                                 run_id,
-                                lambda: self._store.create_run_atomic(
+                                lambda: self._store.create_thread_operation_atomic(
                                     run_id=run_id,
                                     thread_id=thread_id,
                                     owner_worker_id=self._worker_id,
                                     lease_expires_at=lease_expires_at,
+                                    operation_kind=operation_kind.value,
                                     multitask_strategy=multitask_strategy,
                                     assistant_id=assistant_id,
                                     user_id=user_id,
@@ -1068,7 +1106,7 @@ class RunManager:
                                 # worker won the race for this thread.
                                 raise ConflictError(f"Thread {thread_id} already has an active run") from exc
                             raise
-                    # ``create_run_atomic`` already marked any claimed store
+                    # ``create_thread_operation_atomic`` already marked any claimed store
                     # rows as interrupted in the same transaction; no extra
                     # store write is needed for them.
 
@@ -1100,6 +1138,65 @@ class RunManager:
         logger.info("Run created: run_id=%s thread_id=%s", run_id, thread_id)
         return record
 
+    @asynccontextmanager
+    async def reserve_thread_operation(
+        self,
+        thread_id: str,
+        *,
+        kind: ThreadOperationKind,
+        user_id: str | None = None,
+    ) -> AsyncIterator[None]:
+        """Hold exclusive durable admission for a non-run thread operation.
+
+        The reservation is a short-lived pending row, so the same durable
+        uniqueness constraint used by ``create_or_reject`` closes both sides of
+        the race across Gateway workers.
+        """
+        if kind == ThreadOperationKind.run:
+            raise ValueError("Normal runs must be admitted with create_or_reject()")
+        record = await self._admit_thread_operation(
+            thread_id,
+            operation_kind=kind,
+            multitask_strategy="reject",
+            user_id=user_id,
+        )
+        try:
+            reservation_task = asyncio.current_task()
+            if reservation_task is None:
+                raise RuntimeError("Thread operation reservation requires an active asyncio task")
+            lease_lost = True
+            async with self._lock:
+                if self._runs.get(record.run_id) is record:
+                    record.task = reservation_task
+                    lease_lost = record.abort_event.is_set()
+            if lease_lost:
+                raise asyncio.CancelledError()
+            yield
+        except asyncio.CancelledError:
+            if record.abort_event.is_set():
+                raise ConflictError(f"Thread {thread_id} reservation lease was lost") from None
+            raise
+        finally:
+            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),
+                        )
+                    except Exception:
+                        logger.warning(
+                            "Failed to release persisted thread operation %s; leaving it for orphan reconciliation",
+                            record.run_id,
+                            exc_info=True,
+                        )
+            finally:
+                async with self._lock:
+                    removed = self._runs.pop(record.run_id, None)
+                    if removed is not None:
+                        self._unindex_run_locked(record.run_id, removed.thread_id)
+
     async def reconcile_orphaned_inflight_runs(
         self,
         *,
@@ -1170,7 +1267,8 @@ class RunManager:
             record.error = error
             record.stop_reason = stop_reason
             record.updated_at = now
-            recovered.append(record)
+            if record.operation_kind == ThreadOperationKind.run:
+                recovered.append(record)
 
         if recovered:
             logger.warning("Recovered %d orphaned inflight run(s) as error", len(recovered))
@@ -1179,7 +1277,7 @@ class RunManager:
     async def has_inflight(self, thread_id: str) -> bool:
         """Return ``True`` if *thread_id* has a pending or running run."""
         async with self._lock:
-            return any(r.status in (RunStatus.pending, RunStatus.running) or r.finalizing for r in self._thread_records_locked(thread_id))
+            return any(r.operation_kind == ThreadOperationKind.run and (r.status in (RunStatus.pending, RunStatus.running) or r.finalizing) for r in self._thread_records_locked(thread_id))
 
     async def cleanup(self, run_id: str, *, delay: float = 300) -> None:
         """Remove a run record after an optional delay."""
@@ -1304,7 +1402,7 @@ class RunManager:
             # Renew any pending/running run owned by this worker unless its
             # background task has already completed. A pending run whose task
             # has not been spawned yet (``task is None``) is still live from
-            # this worker's perspective — between ``create_run_atomic``
+            # this worker's perspective — between ``create_thread_operation_atomic``
             # inserting the row and the worker layer spawning the agent task
             # there is a brief window. If we drop those records here and the
             # window stretches past ``lease_seconds`` (e.g. event-loop
@@ -1338,16 +1436,18 @@ class RunManager:
                     # or ``owner_worker_id`` changed). Stop the local task so
                     # we don't waste CPU or overwrite the takeover status on
                     # finalisation.
-                    logger.warning(
+                    async with self._lock:
-                        "Run %s lease renewal failed (status=%s,owner=%s) – worker likely taken over; aborting local task",
+                        still_active = self._runs.get(run_id) is record and record.status in (RunStatus.pending, RunStatus.running) and record.owner_worker_id == self._worker_id and (record.task is None or not record.task.done())
-                        run_id,
+                        if still_active:
-                        record.status.value,
+                            logger.warning(
-                        record.owner_worker_id,
+                                "Run %s lease renewal failed (status=%s,owner=%s) – worker likely taken over; aborting local task",
-                    )
+                                run_id,
-                    record.abort_event.set()
+                                record.status.value,
-                    task_active = record.task is not None and not record.task.done()
+                                record.owner_worker_id,
-                    if task_active:
+                            )
-                        record.task.cancel()
+                            record.abort_event.set()
+                            if record.task is not None:
+                                record.task.cancel()
             except Exception:
                 logger.warning("Failed to renew lease for run %s", run_id, exc_info=True)
 

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

@@ -3,6 +3,13 @@
 from enum import StrEnum
 
 
+class ThreadOperationKind(StrEnum):
+    """Kind of operation holding exclusive admission for a thread."""
+
+    run = "run"
+    checkpoint_write = "checkpoint_write"
+
+
 class RunStatus(StrEnum):
     """Lifecycle status of a single run."""
 

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

@@ -25,6 +25,7 @@ class RunStore(abc.ABC):
         user_id: str | None = None,
         model_name: str | None = None,
         status: str = "pending",
+        operation_kind: str = "run",
         multitask_strategy: str = "reject",
         metadata: dict[str, Any] | None = None,
         kwargs: dict[str, Any] | None = None,
@@ -98,6 +99,15 @@ class RunStore(abc.ABC):
     async def delete(self, run_id: str) -> None:
         pass
 
+    async def delete_thread_operation(self, run_id: str, *, user_id: str | None) -> None:
+        """Release an admitted thread operation for its recorded owner.
+
+        The default keeps legacy stores compatible: older implementations only
+        accepted ``run_id``. User-aware stores should override this method so
+        cleanup never depends on ambient request context.
+        """
+        await self.delete(run_id)
+
     @abc.abstractmethod
     async def update_model_name(
         self,
@@ -215,7 +225,53 @@ class RunStore(abc.ABC):
         """Return active runs whose lease has expired (or is NULL for pre-ownership rows)."""
         pass
 
-    @abc.abstractmethod
+    async def create_thread_operation_atomic(
+        self,
+        run_id: str,
+        *,
+        thread_id: str,
+        owner_worker_id: str,
+        lease_expires_at: str | None,
+        operation_kind: str = "run",
+        multitask_strategy: str = "reject",
+        assistant_id: str | None = None,
+        user_id: str | None = None,
+        model_name: str | None = None,
+        metadata: dict[str, Any] | None = None,
+        kwargs: dict[str, Any] | None = None,
+        created_at: str | None = None,
+        grace_seconds: int = 10,
+    ) -> tuple[dict[str, Any], list[dict[str, Any]]]:
+        """Atomically create an active thread operation with cross-process uniqueness.
+
+        The default implementation preserves compatibility with stores that
+        still implement the former ``create_run_atomic`` interface. Legacy
+        stores support only normal run rows; internal operation kinds require
+        an implementation of this method.
+
+        Returns ``(new_run_dict, claimed_run_dicts)``.
+        Raises ``IntegrityError`` on conflict for ``reject`` strategy.
+        """
+        legacy_impl = type(self).create_run_atomic
+        if legacy_impl is RunStore.create_run_atomic:
+            raise NotImplementedError("RunStore must implement create_thread_operation_atomic() or create_run_atomic()")
+        if operation_kind != "run":
+            raise NotImplementedError("Legacy RunStore.create_run_atomic() cannot create non-run thread operations")
+        return await self.create_run_atomic(
+            run_id,
+            thread_id=thread_id,
+            owner_worker_id=owner_worker_id,
+            lease_expires_at=lease_expires_at,
+            multitask_strategy=multitask_strategy,
+            assistant_id=assistant_id,
+            user_id=user_id,
+            model_name=model_name,
+            metadata=metadata,
+            kwargs=kwargs,
+            created_at=created_at,
+            grace_seconds=grace_seconds,
+        )
+
     async def create_run_atomic(
         self,
         run_id: str,
@@ -232,9 +288,22 @@ class RunStore(abc.ABC):
         created_at: str | None = None,
         grace_seconds: int = 10,
     ) -> tuple[dict[str, Any], list[dict[str, Any]]]:
-        """Atomically create a run row with cross-process thread-uniqueness.
+        """Deprecated compatibility alias for normal-run admission."""
-
+        operation_impl = type(self).create_thread_operation_atomic
-        Returns ``(new_run_dict, claimed_run_dicts)``.
+        if operation_impl is RunStore.create_thread_operation_atomic:
-        Raises ``IntegrityError`` on conflict for ``reject`` strategy.
+            raise NotImplementedError("RunStore must implement create_thread_operation_atomic() or create_run_atomic()")
-        """
+        return await self.create_thread_operation_atomic(
-        pass
+            run_id,
+            thread_id=thread_id,
+            owner_worker_id=owner_worker_id,
+            lease_expires_at=lease_expires_at,
+            operation_kind="run",
+            multitask_strategy=multitask_strategy,
+            assistant_id=assistant_id,
+            user_id=user_id,
+            model_name=model_name,
+            metadata=metadata,
+            kwargs=kwargs,
+            created_at=created_at,
+            grace_seconds=grace_seconds,
+        )

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

@@ -41,6 +41,7 @@ class MemoryRunStore(RunStore):
         user_id=None,
         model_name=None,
         status="pending",
+        operation_kind="run",
         multitask_strategy="reject",
         metadata=None,
         kwargs=None,
@@ -58,6 +59,7 @@ class MemoryRunStore(RunStore):
             "user_id": user_id,
             "model_name": model_name,
             "status": status,
+            "operation_kind": operation_kind,
             "multitask_strategy": multitask_strategy,
             "metadata": metadata or {},
             "kwargs": kwargs or {},
@@ -85,7 +87,7 @@ class MemoryRunStore(RunStore):
         run_ids = self._runs_by_thread.get(thread_id)
         if not run_ids:
             return []
-        results = [run for run_id in run_ids if (run := self._runs.get(run_id)) is not None and (user_id is None or run.get("user_id") == user_id)]
+        results = [run for run_id in run_ids if (run := self._runs.get(run_id)) is not None and run.get("operation_kind", "run") == "run" and (user_id is None or run.get("user_id") == user_id)]
         results.sort(key=lambda r: r["created_at"], reverse=True)
         return results[:limit]
 
@@ -94,7 +96,7 @@ class MemoryRunStore(RunStore):
         sources: set[str] = set()
         for run_id in run_ids:
             run = self._runs.get(run_id)
-            if run is None or run.get("status") != "success":
+            if run is None or run.get("operation_kind", "run") != "run" or run.get("status") != "success":
                 continue
             if user_id is not None and run.get("user_id") != user_id:
                 continue
@@ -105,7 +107,7 @@ class MemoryRunStore(RunStore):
 
     async def get_many_by_thread(self, thread_id, run_ids, *, user_id=None):
         thread_run_ids = self._runs_by_thread.get(thread_id) or ()
-        return {run_id: run for run_id in thread_run_ids if run_id in run_ids and (run := self._runs.get(run_id)) is not None and (user_id is None or run.get("user_id") == user_id)}
+        return {run_id: run for run_id in thread_run_ids if run_id in run_ids and (run := self._runs.get(run_id)) is not None and run.get("operation_kind", "run") == "run" and (user_id is None or run.get("user_id") == user_id)}
 
     async def update_status(self, run_id, status, *, error=None, stop_reason=None):
         run = self._runs.get(run_id)
@@ -128,7 +130,7 @@ class MemoryRunStore(RunStore):
             self._runs[run_id]["model_name"] = model_name
             self._runs[run_id]["updated_at"] = datetime.now(UTC).isoformat()
 
-    async def delete(self, run_id):
+    async def delete(self, run_id, *, user_id=None):
         run = self._runs.pop(run_id, None)
         if run is not None:
             self._unindex_run(run_id, run["thread_id"])
@@ -152,7 +154,7 @@ class MemoryRunStore(RunStore):
 
     async def list_pending(self, *, before=None):
         now = before or datetime.now(UTC).isoformat()
-        results = [r for r in self._runs.values() if r["status"] == "pending" and r["created_at"] <= now]
+        results = [r for r in self._runs.values() if r.get("operation_kind", "run") == "run" and r["status"] == "pending" and r["created_at"] <= now]
         results.sort(key=lambda r: r["created_at"])
         return results
 
@@ -167,7 +169,7 @@ class MemoryRunStore(RunStore):
         # Use the thread index for an O(runs-in-thread) lookup instead of
         # scanning every run in the process (mirrors ``list_by_thread``).
         run_ids = self._runs_by_thread.get(thread_id) or ()
-        completed = [run for run_id in run_ids if (run := self._runs.get(run_id)) is not None and run.get("status") in statuses]
+        completed = [run for run_id in run_ids if (run := self._runs.get(run_id)) is not None and run.get("operation_kind", "run") == "run" and run.get("status") in statuses]
         by_model: dict[str, dict] = {}
         for r in completed:
             usage_by_model = r.get("token_usage_by_model") or {}
@@ -288,13 +290,14 @@ class MemoryRunStore(RunStore):
         results.sort(key=lambda r: r["created_at"])
         return results
 
-    async def create_run_atomic(
+    async def create_thread_operation_atomic(
         self,
         run_id: str,
         *,
         thread_id: str,
         owner_worker_id: str,
         lease_expires_at: str | None,
+        operation_kind: str = "run",
         multitask_strategy: str = "reject",
         assistant_id: str | None = None,
         user_id: str | None = None,
@@ -330,6 +333,7 @@ class MemoryRunStore(RunStore):
                     continue
                 if r["status"] not in ("pending", "running"):
                     continue
+                lease_expired = False
                 existing_lease = r.get("lease_expires_at")
                 if existing_lease is not None:
                     try:
@@ -340,6 +344,7 @@ class MemoryRunStore(RunStore):
                         # raise ``TypeError``.
                         if lease_dt.tzinfo is None:
                             lease_dt = lease_dt.replace(tzinfo=UTC)
+                        lease_expired = lease_dt < cutoff
                         if lease_dt >= cutoff and r.get("owner_worker_id") != owner_worker_id:
                             # Live run owned by another worker — cannot
                             # interrupt, and the partial unique index would
@@ -349,6 +354,8 @@ class MemoryRunStore(RunStore):
                             raise ConflictError(f"Thread {thread_id} already has an active run owned by another worker")
                     except (ValueError, TypeError):
                         pass
+                if r.get("operation_kind", "run") != "run" and not lease_expired:
+                    raise ConflictError(f"Thread {thread_id} has an active checkpoint write")
                 candidates.append(r)
             for r in candidates:
                 r["status"] = "interrupted"
@@ -364,6 +371,7 @@ class MemoryRunStore(RunStore):
             "user_id": user_id,
             "model_name": model_name,
             "status": "pending",
+            "operation_kind": operation_kind,
             "multitask_strategy": multitask_strategy,
             "metadata": metadata or {},
             "kwargs": kwargs or {},

backend/tests/test_console_router.py

@@ -126,6 +126,15 @@ def _seed_rows() -> tuple[list[ThreadMetaRow], list[RunRow]]:
             created_at=NOW - timedelta(hours=2),
             updated_at=NOW - timedelta(hours=2) + timedelta(seconds=8),
         ),
+        RunRow(
+            run_id="checkpoint-write-1",
+            thread_id="t1",
+            user_id="user-a",
+            operation_kind="checkpoint_write",
+            status="error",
+            created_at=NOW - timedelta(minutes=30),
+            updated_at=NOW - timedelta(minutes=29),
+        ),
     ]
     return threads, runs
 

backend/tests/test_gateway_checkpoint_mode.py

@@ -29,7 +29,9 @@ from app.gateway.routers import threads
 from deerflow.agents.thread_state import get_thread_state_schema
 from deerflow.config.app_config import AppConfig, reset_app_config, set_app_config
 from deerflow.persistence.thread_meta.memory import MemoryThreadMetaStore
+from deerflow.runtime import RunManager
 from deerflow.runtime.checkpoint_mode import checkpoint_metadata_uses_delta, inject_checkpoint_mode
+from deerflow.runtime.runs.store.memory import MemoryRunStore
 
 _THREAD_ID = "thread-gateway-parity"
 
@@ -128,6 +130,7 @@ def test_full_mode_gateway_rejects_delta_thread_with_409(_stub_app_config, monke
     app.state.thread_store.get = AsyncMock(return_value=None)
     app.state.checkpoint_channel_mode = "full"
     app.state.run_event_store = SimpleNamespace()
+    app.state.run_manager = RunManager(store=MemoryRunStore())
     app.include_router(threads.router)
 
     full_graph = _build_reply_graph("full", checkpointer)

backend/tests/test_migration_0004_run_ownership_dedupe.py

@@ -157,7 +157,7 @@ async def test_migration_dedupes_duplicate_active_rows_before_unique_index(tmp_p
         with sqlite3.connect(db_path) as raw:
             version_row = raw.execute("SELECT version_num FROM alembic_version").fetchone()
         # Bootstrap upgrades through the later revisions after 0004.
-        assert version_row[0] == "0007_scheduled_run_active_index"
+        assert version_row[0] == "0008_thread_operation_kind"
 
         # Sanity: the invariant the index enforces is now true — at most one
         # active row per thread.

backend/tests/test_migration_0007_scheduled_run_active_dedupe.py

@@ -169,7 +169,7 @@ async def test_migration_supersedes_duplicate_active_runs_before_unique_index(tm
 
         with sqlite3.connect(db_path) as raw:
             version_row = raw.execute("SELECT version_num FROM alembic_version").fetchone()
-        assert version_row[0] == "0007_scheduled_run_active_index"
+        assert version_row[0] == "0008_thread_operation_kind"
 
         # Sanity: the invariant the index enforces now holds — at most one
         # active row per task_id.

backend/tests/test_multi_worker_run_ownership.py

@@ -3,7 +3,7 @@
 Coverage:
 - create_or_reject with reject strategy blocks duplicate active runs
 - create_or_reject with interrupt strategy claims and cancels old runs
-- create_run_atomic refuses to interrupt a run owned by another live worker
+- create_thread_operation_atomic refuses to interrupt a run owned by another live worker
 - reconcile_orphaned_inflight_runs uses lease-based detection
 - periodic reconciliation notifies Gateway recovery orchestration
 - Worker reconciliation skips runs with unexpired leases
@@ -20,7 +20,7 @@ from unittest.mock import AsyncMock
 import pytest
 
 from deerflow.config.run_ownership_config import RunOwnershipConfig
-from deerflow.runtime import ORPHAN_RECOVERY_STOP_REASON, RunManager, RunStatus
+from deerflow.runtime import ORPHAN_RECOVERY_STOP_REASON, RunManager, RunStatus, ThreadOperationKind
 from deerflow.runtime.runs.manager import CancelOutcome, ConflictError, _generate_worker_id
 from deerflow.runtime.runs.store.memory import MemoryRunStore
 
@@ -74,6 +74,76 @@ async def test_reject_succeeds_when_no_active_run():
     assert record.lease_expires_at is not None
 
 
+@pytest.mark.anyio
+async def test_checkpoint_write_reservation_rejects_nonowning_worker_while_run_is_active():
+    """A durable run owned by worker A must block worker B's checkpoint writer."""
+    store = MemoryRunStore()
+    owner = _make_manager(store=store, worker_id="worker-a")
+    non_owner = _make_manager(store=store, worker_id="worker-b")
+    active = await owner.create_or_reject("thread-1")
+    await owner.set_status(active.run_id, RunStatus.running)
+
+    with pytest.raises(ConflictError, match="already has an active run"):
+        async with non_owner.reserve_thread_operation("thread-1", kind=ThreadOperationKind.checkpoint_write):
+            pytest.fail("the checkpoint mutation guard must not be acquired")
+
+    stored = await store.get(active.run_id)
+    assert stored is not None
+    assert stored["status"] == "running"
+
+
+@pytest.mark.anyio
+async def test_checkpoint_write_reservation_blocks_new_runs_until_mutation_finishes():
+    """The durable guard closes the check-then-write window in both directions."""
+    store = MemoryRunStore()
+    compaction_worker = _make_manager(store=store, worker_id="worker-a")
+    run_worker = _make_manager(store=store, worker_id="worker-b")
+
+    async with compaction_worker.reserve_thread_operation("thread-1", kind=ThreadOperationKind.checkpoint_write):
+        inflight = await store.list_inflight()
+        assert len(inflight) == 1
+        assert inflight[0]["operation_kind"] == ThreadOperationKind.checkpoint_write
+        assert inflight[0]["metadata"] == {}
+
+        with pytest.raises(ConflictError, match="checkpoint write"):
+            await run_worker.create_or_reject("thread-1", multitask_strategy="interrupt")
+
+    assert await store.list_inflight() == []
+    assert await store.list_by_thread("thread-1") == []
+
+    admitted = await run_worker.create_or_reject("thread-1")
+    assert admitted.status == RunStatus.pending
+
+
+@pytest.mark.anyio
+async def test_interrupt_reclaims_expired_checkpoint_write_reservation():
+    """A dead checkpoint writer must not wait for periodic reconciliation."""
+    store = MemoryRunStore()
+    expired = (datetime.now(UTC) - timedelta(seconds=30)).isoformat()
+    await store.put(
+        "checkpoint-write-1",
+        thread_id="thread-1",
+        status="pending",
+        operation_kind=ThreadOperationKind.checkpoint_write,
+        owner_worker_id="dead-worker",
+        lease_expires_at=expired,
+        created_at=expired,
+    )
+    manager = _make_manager(
+        store=store,
+        worker_id="worker-b",
+        run_ownership_config=_lease_config(grace_seconds=10),
+    )
+
+    admitted = await manager.create_or_reject("thread-1", multitask_strategy="interrupt")
+
+    assert admitted.status == RunStatus.pending
+    stale = await store.get("checkpoint-write-1")
+    assert stale is not None
+    assert stale["status"] == "interrupted"
+    assert stale["owner_worker_id"] == "worker-b"
+
+
 @pytest.mark.anyio
 async def test_reject_blocks_reentrant_same_thread_locally():
     """reject must also block when a local in-memory active run exists."""
@@ -139,7 +209,7 @@ async def test_interrupt_exhausted_retries_surface_as_conflict_error():
     import sqlite3
 
     class _AlwaysUniqueViolationStore(MemoryRunStore):
-        """MemoryRunStore whose ``create_run_atomic`` always raises a
+        """MemoryRunStore whose ``create_thread_operation_atomic`` always raises a
         real-flavoured unique-violation IntegrityError, simulating a worker
         that keeps losing the cross-worker race for the same thread."""
 
@@ -147,7 +217,7 @@ async def test_interrupt_exhausted_retries_surface_as_conflict_error():
             super().__init__()
             self.atomic_call_count = 0
 
-        async def create_run_atomic(self, *args, **kwargs):
+        async def create_thread_operation_atomic(self, *args, **kwargs):
             self.atomic_call_count += 1
             err = sqlite3.IntegrityError("UNIQUE constraint failed: runs.uq_runs_thread_active")
             err.sqlite_errorcode = sqlite3.SQLITE_CONSTRAINT_UNIQUE
@@ -184,6 +254,7 @@ async def test_run_record_stores_owner_and_lease():
     assert stored is not None
     assert stored["owner_worker_id"] == manager.worker_id
     assert stored["lease_expires_at"] is not None
+    assert stored["operation_kind"] == ThreadOperationKind.run
 
 
 @pytest.mark.anyio
@@ -197,6 +268,34 @@ async def test_store_row_roundtrips_ownership_fields():
     assert hydrated is not None
     assert hydrated.owner_worker_id == manager.worker_id
     assert hydrated.lease_expires_at is not None
+    assert hydrated.operation_kind == ThreadOperationKind.run
+
+
+@pytest.mark.anyio
+async def test_reconciliation_releases_expired_internal_operation_without_reporting_run():
+    """Expired internal reservations release admission without becoming failed runs."""
+    store = MemoryRunStore()
+    expired = (datetime.now(UTC) - timedelta(seconds=30)).isoformat()
+    await store.put(
+        "checkpoint-write-1",
+        thread_id="thread-1",
+        status="pending",
+        operation_kind=ThreadOperationKind.checkpoint_write,
+        owner_worker_id="dead-worker",
+        lease_expires_at=expired,
+        created_at=expired,
+    )
+    manager = _make_manager(
+        store=store,
+        run_ownership_config=_lease_config(heartbeat_enabled=True, grace_seconds=10),
+    )
+
+    recovered = await manager.reconcile_orphaned_inflight_runs(error="owner expired")
+
+    assert recovered == []
+    stored = await store.get("checkpoint-write-1")
+    assert stored is not None
+    assert stored["status"] == "error"
 
 
 # ---------------------------------------------------------------------------
@@ -678,12 +777,12 @@ async def test_heartbeat_renews_active_run_leases():
 
 @pytest.mark.anyio
 async def test_heartbeat_renews_pending_run_before_task_is_spawned():
-    """A run sitting in ``pending`` between ``create_run_atomic`` and task
+    """A run sitting in ``pending`` between ``create_thread_operation_atomic`` and task
     spawn must still have its lease renewed.
 
     Pre-fix the renewal filter required ``record.task is not None``, so a
     pending run with no task yet (the brief window after
-    ``create_run_atomic`` inserts the row before the worker layer spawns
+    ``create_thread_operation_atomic`` inserts the row before the worker layer spawns
     the agent task) was silently skipped. If that window stretched past
     ``lease_seconds`` — e.g. event-loop saturation, slow checkpoint
     hydrate — peer reconciliation reclaimed the run as an orphan and
@@ -859,14 +958,14 @@ def test_two_managers_have_different_default_ids():
 
 
 @pytest.mark.anyio
-async def test_create_run_atomic_reject_prevents_duplicate():
+async def test_create_thread_operation_atomic_reject_prevents_duplicate():
-    """store.create_run_atomic with reject must raise ConflictError on duplicate."""
+    """Atomic thread-operation creation must reject a duplicate."""
     store = MemoryRunStore()
     config = _lease_config()
 
-    store.create_run_atomic = AsyncMock(wraps=store.create_run_atomic)
+    store.create_thread_operation_atomic = AsyncMock(wraps=store.create_thread_operation_atomic)
 
-    await store.create_run_atomic(
+    await store.create_thread_operation_atomic(
         run_id="run-1",
         thread_id="thread-1",
         owner_worker_id="w1",
@@ -876,7 +975,7 @@ async def test_create_run_atomic_reject_prevents_duplicate():
     )
 
     with pytest.raises(ConflictError, match="already has an active run"):
-        await store.create_run_atomic(
+        await store.create_thread_operation_atomic(
             run_id="run-2",
             thread_id="thread-1",
             owner_worker_id="w2",
@@ -887,14 +986,14 @@ async def test_create_run_atomic_reject_prevents_duplicate():
 
 
 @pytest.mark.anyio
-async def test_create_run_atomic_interrupt_claims_and_creates():
+async def test_create_thread_operation_atomic_interrupt_claims_and_creates():
-    """store.create_run_atomic with interrupt must claim old and create new."""
+    """Atomic thread-operation creation with interrupt must claim and replace."""
     store = MemoryRunStore()
     config = _lease_config()
     # Create an active run with an expired lease (simulating a crashed worker)
     expired_lease = (datetime.now(UTC) - timedelta(seconds=60)).isoformat()
 
-    await store.create_run_atomic(
+    await store.create_thread_operation_atomic(
         run_id="run-old",
         thread_id="thread-1",
         owner_worker_id="w1",
@@ -903,7 +1002,7 @@ async def test_create_run_atomic_interrupt_claims_and_creates():
         grace_seconds=config.grace_seconds,
     )
 
-    new_row, claimed = await store.create_run_atomic(
+    new_row, claimed = await store.create_thread_operation_atomic(
         run_id="run-new",
         thread_id="thread-1",
         owner_worker_id="w2",
@@ -923,7 +1022,7 @@ async def test_create_run_atomic_interrupt_claims_and_creates():
 
 
 @pytest.mark.anyio
-async def test_create_run_atomic_interrupt_rejects_other_worker_valid_lease():
+async def test_create_thread_operation_atomic_interrupt_rejects_other_worker_valid_lease():
     """Interrupt must raise ConflictError when a valid-lease run is owned by another worker.
 
     The partial unique index ``uq_runs_thread_active`` would reject the INSERT
@@ -934,7 +1033,7 @@ async def test_create_run_atomic_interrupt_rejects_other_worker_valid_lease():
     config = _lease_config(grace_seconds=10)
     valid_lease = (datetime.now(UTC) + timedelta(seconds=30)).isoformat()
 
-    await store.create_run_atomic(
+    await store.create_thread_operation_atomic(
         run_id="valid-lease-run",
         thread_id="thread-1",
         owner_worker_id="other-worker",
@@ -944,7 +1043,7 @@ async def test_create_run_atomic_interrupt_rejects_other_worker_valid_lease():
     )
 
     with pytest.raises(ConflictError, match="another worker"):
-        await store.create_run_atomic(
+        await store.create_thread_operation_atomic(
             run_id="run-new",
             thread_id="thread-1",
             owner_worker_id="w2",
@@ -960,13 +1059,13 @@ async def test_create_run_atomic_interrupt_rejects_other_worker_valid_lease():
 
 
 @pytest.mark.anyio
-async def test_create_run_atomic_interrupt_allows_self_owned_valid_lease():
+async def test_create_thread_operation_atomic_interrupt_allows_self_owned_valid_lease():
     """Interrupt must succeed when the existing valid-lease run is owned by this worker."""
     store = MemoryRunStore()
     config = _lease_config(grace_seconds=10)
     valid_lease = (datetime.now(UTC) + timedelta(seconds=30)).isoformat()
 
-    await store.create_run_atomic(
+    await store.create_thread_operation_atomic(
         run_id="self-run",
         thread_id="thread-1",
         owner_worker_id="w1",
@@ -975,7 +1074,7 @@ async def test_create_run_atomic_interrupt_allows_self_owned_valid_lease():
         grace_seconds=config.grace_seconds,
     )
 
-    new_row, claimed = await store.create_run_atomic(
+    new_row, claimed = await store.create_thread_operation_atomic(
         run_id="run-new",
         thread_id="thread-1",
         owner_worker_id="w1",  # same worker
@@ -991,7 +1090,7 @@ async def test_create_run_atomic_interrupt_allows_self_owned_valid_lease():
 
 
 @pytest.mark.anyio
-async def test_create_run_atomic_interrupt_rolls_back_earlier_mutations_on_conflict():
+async def test_create_thread_operation_atomic_interrupt_rolls_back_earlier_mutations_on_conflict():
     """Interrupt must not leave earlier candidates interrupted when a later
     candidate raises ConflictError.
 
@@ -1012,7 +1111,7 @@ async def test_create_run_atomic_interrupt_rolls_back_earlier_mutations_on_confl
     expired_lease = (datetime.now(UTC) - timedelta(seconds=60)).isoformat()
     valid_lease = (datetime.now(UTC) + timedelta(seconds=30)).isoformat()
 
-    # Seed both active rows directly via ``put`` (bypassing create_run_atomic's
+    # Seed both active rows directly via ``put`` (bypassing atomic operation creation's
     # reject check, which would refuse the second row). Insert the
     # interruptible run first so dict iteration visits it first — that's the
     # ordering that exposes the half-interrupted divergence in a naive
@@ -1033,7 +1132,7 @@ async def test_create_run_atomic_interrupt_rolls_back_earlier_mutations_on_confl
     )
 
     with pytest.raises(ConflictError, match="another worker"):
-        await store.create_run_atomic(
+        await store.create_thread_operation_atomic(
             run_id="run-new",
             thread_id="thread-1",
             owner_worker_id="w1",

backend/tests/test_persistence_bootstrap.py

@@ -48,7 +48,7 @@ from deerflow.persistence.migrations._helpers import _normalize_default
 asyncio_test = pytest.mark.asyncio
 
 
-HEAD = "0007_scheduled_run_active_index"
+HEAD = "0008_thread_operation_kind"
 BASELINE = "0001_baseline"
 
 
@@ -147,6 +147,8 @@ async def test_empty_branch_creates_all_and_stamps_head(tmp_path: Path) -> None:
         }:
             assert required in tables, f"missing table: {required}"
         assert "token_usage_by_model" in await _runs_columns(engine)
+        operation_kind = await _runs_column_meta(engine, "operation_kind")
+        assert operation_kind["nullable"] is False
         assert await _alembic_version(engine) == HEAD
         # The partial unique index on (thread_id WHERE status IN pending/running)
         # must exist on a fresh DB because the empty-branch stamps head without

backend/tests/test_persistence_bootstrap_concurrency.py

@@ -28,7 +28,7 @@ from deerflow.persistence.bootstrap import bootstrap_schema
 pytestmark = pytest.mark.asyncio
 
 
-HEAD = "0007_scheduled_run_active_index"
+HEAD = "0008_thread_operation_kind"
 
 
 def _url(tmp_path: Path) -> str:

backend/tests/test_persistence_bootstrap_regression.py

@@ -76,7 +76,7 @@ async def test_legacy_database_recovers_token_usage_column(tmp_path: Path) -> No
             cols = {row[1] for row in raw.execute("PRAGMA table_info(runs)").fetchall()}
             assert "token_usage_by_model" in cols
             version_row = raw.execute("SELECT version_num FROM alembic_version").fetchone()
-            assert version_row[0] == "0007_scheduled_run_active_index"
+            assert version_row[0] == "0008_thread_operation_kind"
 
         # And the read path that originally 500'd must now succeed.
         sf = get_session_factory()
@@ -116,6 +116,6 @@ async def test_legacy_database_with_manual_alter_still_bootstraps(tmp_path: Path
             # No duplicate column -- list, not set, to catch dupes.
             assert cols.count("token_usage_by_model") == 1
             version_row = raw.execute("SELECT version_num FROM alembic_version").fetchone()
-            assert version_row[0] == "0007_scheduled_run_active_index"
+            assert version_row[0] == "0008_thread_operation_kind"
     finally:
         await close_engine()

backend/tests/test_run_manager.py

@@ -9,7 +9,8 @@ from typing import Any
 import pytest
 from sqlalchemy.exc import DatabaseError as SQLAlchemyDatabaseError
 
-from deerflow.runtime import DisconnectMode, RunManager, RunStatus
+from deerflow.config.run_ownership_config import RunOwnershipConfig
+from deerflow.runtime import DisconnectMode, RunManager, RunStatus, ThreadOperationKind
 from deerflow.runtime.runs.manager import CancelOutcome, ConflictError, PersistenceRetryPolicy
 from deerflow.runtime.runs.store.memory import MemoryRunStore
 
@@ -99,6 +100,34 @@ class AlwaysMissingCompletionRunStore(MemoryRunStore):
         return False
 
 
+class FailingDeleteRunStore(MemoryRunStore):
+    """Run store that cannot release a persisted thread-operation row."""
+
+    async def delete(self, run_id, *, user_id=None):
+        raise RuntimeError("delete failed")
+
+
+class LostLeaseRunStore(MemoryRunStore):
+    """Run store that reports a reservation was taken over."""
+
+    async def update_lease(self, run_id, *, owner_worker_id, lease_expires_at):
+        return False
+
+
+class PausedLostLeaseRunStore(MemoryRunStore):
+    """Run store whose failed renewal can be released after reservation cleanup."""
+
+    def __init__(self) -> None:
+        super().__init__()
+        self.renewal_started = asyncio.Event()
+        self.finish_renewal = asyncio.Event()
+
+    async def update_lease(self, run_id, *, owner_worker_id, lease_expires_at):
+        self.renewal_started.set()
+        await self.finish_renewal.wait()
+        return False
+
+
 async def _stored_statuses(store: MemoryRunStore, *run_ids: str) -> dict[str, Any]:
     rows = {}
     for run_id in run_ids:
@@ -107,6 +136,142 @@ async def _stored_statuses(store: MemoryRunStore, *run_ids: str) -> dict[str, An
     return rows
 
 
+@pytest.mark.anyio
+async def test_reservation_delete_failure_preserves_body_error_and_clears_local_record(caplog):
+    store = FailingDeleteRunStore()
+    manager = RunManager(
+        store=store,
+        persistence_retry_policy=PersistenceRetryPolicy(max_attempts=1, initial_delay=0),
+    )
+
+    with caplog.at_level(logging.WARNING), pytest.raises(ValueError, match="body failed"):
+        async with manager.reserve_thread_operation(
+            "thread-1",
+            kind=ThreadOperationKind.checkpoint_write,
+        ):
+            raise ValueError("body failed")
+
+    assert not await manager.has_inflight("thread-1")
+    assert manager._runs == {}
+    assert manager._runs_by_thread == {}
+    assert len(await store.list_inflight()) == 1
+    assert "leaving it for orphan reconciliation" in caplog.text
+
+
+@pytest.mark.anyio
+async def test_reservation_lease_loss_surfaces_as_conflict_after_cancelling_body():
+    store = LostLeaseRunStore()
+    manager = RunManager(
+        store=store,
+        run_ownership_config=RunOwnershipConfig(
+            lease_seconds=30,
+            grace_seconds=10,
+            heartbeat_enabled=True,
+        ),
+    )
+    entered = asyncio.Event()
+
+    async def hold_reservation() -> None:
+        async with manager.reserve_thread_operation(
+            "thread-1",
+            kind=ThreadOperationKind.checkpoint_write,
+        ):
+            entered.set()
+            await asyncio.Event().wait()
+
+    task = asyncio.create_task(hold_reservation())
+    await entered.wait()
+
+    await manager._renew_leases()
+
+    with pytest.raises(ConflictError, match="reservation lease was lost"):
+        await task
+    assert not await manager.has_inflight("thread-1")
+    assert await store.list_inflight() == []
+
+
+@pytest.mark.anyio
+async def test_reservation_cancelled_while_attaching_task_is_released(monkeypatch):
+    store = MemoryRunStore()
+    manager = RunManager(store=store)
+    admitted = asyncio.Event()
+    return_from_admission = asyncio.Event()
+    original_admit = manager._admit_thread_operation
+
+    async def pause_after_admission(*args, **kwargs):
+        record = await original_admit(*args, **kwargs)
+        admitted.set()
+        await return_from_admission.wait()
+        return record
+
+    monkeypatch.setattr(manager, "_admit_thread_operation", pause_after_admission)
+
+    async def reserve() -> None:
+        async with manager.reserve_thread_operation(
+            "thread-1",
+            kind=ThreadOperationKind.checkpoint_write,
+        ):
+            raise AssertionError("cancelled reservation must not enter its body")
+
+    task = asyncio.create_task(reserve())
+    await admitted.wait()
+    await manager._lock.acquire()
+    return_from_admission.set()
+    await asyncio.sleep(0)
+    task.cancel()
+    manager._lock.release()
+
+    with pytest.raises(asyncio.CancelledError):
+        await task
+    assert not await manager.has_inflight("thread-1")
+    assert manager._runs == {}
+    assert manager._runs_by_thread == {}
+    assert await store.list_inflight() == []
+
+
+@pytest.mark.anyio
+async def test_late_failed_renewal_does_not_cancel_released_reservation():
+    store = PausedLostLeaseRunStore()
+    manager = RunManager(
+        store=store,
+        run_ownership_config=RunOwnershipConfig(
+            lease_seconds=30,
+            grace_seconds=10,
+            heartbeat_enabled=True,
+        ),
+    )
+    entered = asyncio.Event()
+    leave_body = asyncio.Event()
+    context_exited = asyncio.Event()
+    finish_request = asyncio.Event()
+
+    async def request() -> None:
+        async with manager.reserve_thread_operation(
+            "thread-1",
+            kind=ThreadOperationKind.checkpoint_write,
+        ):
+            entered.set()
+            await leave_body.wait()
+        context_exited.set()
+        await finish_request.wait()
+
+    request_task = asyncio.create_task(request())
+    await entered.wait()
+    renewal_task = asyncio.create_task(manager._renew_leases())
+    await store.renewal_started.wait()
+
+    leave_body.set()
+    await context_exited.wait()
+    assert not await manager.has_inflight("thread-1")
+
+    store.finish_renewal.set()
+    await renewal_task
+    assert not request_task.done()
+
+    finish_request.set()
+    await request_task
+
+
 @pytest.mark.anyio
 async def test_create_and_get(manager: RunManager):
     """Created run should be retrievable with new fields."""
@@ -367,6 +532,16 @@ async def test_has_inflight(manager: RunManager):
     assert await manager.has_inflight("thread-1") is False
 
 
+@pytest.mark.anyio
+async def test_has_inflight_ignores_checkpoint_write_reservation(manager: RunManager):
+    """Internal checkpoint writers are not user-visible runs."""
+    async with manager.reserve_thread_operation(
+        "thread-1",
+        kind=ThreadOperationKind.checkpoint_write,
+    ):
+        assert await manager.has_inflight("thread-1") is False
+
+
 @pytest.mark.anyio
 async def test_cleanup(manager: RunManager):
     """After cleanup, the run should be gone."""
@@ -662,7 +837,7 @@ async def test_create_or_reject_does_not_interrupt_old_run_when_new_run_store_wr
     manager = RunManager(store=store)
     old = await manager.create("thread-1")
     await manager.set_status(old.run_id, RunStatus.running)
-    store.create_run_atomic = AsyncMock(side_effect=RuntimeError("db down"))
+    store.create_thread_operation_atomic = AsyncMock(side_effect=RuntimeError("db down"))
 
     with pytest.raises(RuntimeError, match="db down"):
         await manager.create_or_reject("thread-1", multitask_strategy="interrupt")
@@ -686,7 +861,7 @@ async def test_create_or_reject_does_not_interrupt_old_run_when_new_run_store_wr
     async def cancelled_create(run_id, **kwargs):
         raise asyncio.CancelledError
 
-    store.create_run_atomic = cancelled_create
+    store.create_thread_operation_atomic = cancelled_create
 
     with pytest.raises(asyncio.CancelledError):
         await manager.create_or_reject("thread-1", multitask_strategy="interrupt")
@@ -900,12 +1075,12 @@ async def test_list_by_thread_falls_back_to_store_with_user_filter():
 
 
 class _FailingPutRunStore(MemoryRunStore):
-    """Memory run store whose every ``put`` and ``create_run_atomic`` fails (non-retryably)."""
+    """Memory run store whose every ``put`` and atomic operation create fails."""
 
     async def put(self, run_id, **kwargs):
         raise ValueError("simulated persist failure")
 
-    async def create_run_atomic(self, run_id, **kwargs):
+    async def create_thread_operation_atomic(self, run_id, **kwargs):
         raise ValueError("simulated persist failure")
 
 

backend/tests/test_run_repository.py

@@ -9,7 +9,7 @@ import pytest
 from sqlalchemy.dialects import postgresql
 
 from deerflow.persistence.run import RunRepository
-from deerflow.runtime import CancelOutcome, RunManager, RunStatus
+from deerflow.runtime import CancelOutcome, RunManager, RunStatus, ThreadOperationKind
 from deerflow.runtime.runs.manager import ConflictError
 from deerflow.runtime.runs.store.base import RunStore
 
@@ -29,6 +29,9 @@ async def _cleanup():
 
 
 class _CustomRunStoreWithoutProgress(RunStore):
+    def __init__(self):
+        self.legacy_atomic_calls = 0
+
     async def put(self, *args, **kwargs):
         return None
 
@@ -66,6 +69,7 @@ class _CustomRunStoreWithoutProgress(RunStore):
         return []
 
     async def create_run_atomic(self, *args, **kwargs):
+        self.legacy_atomic_calls += 1
         return {}, []
 
     async def claim_for_takeover(self, *args, **kwargs):
@@ -79,6 +83,28 @@ async def test_update_run_progress_defaults_to_noop_for_custom_store():
     await store.update_run_progress("r1", total_tokens=1)
 
 
+@pytest.mark.anyio
+async def test_legacy_create_run_atomic_store_remains_compatible():
+    store = _CustomRunStoreWithoutProgress()
+
+    await store.create_thread_operation_atomic(
+        "r1",
+        thread_id="t1",
+        owner_worker_id="worker-1",
+        lease_expires_at=None,
+    )
+
+    assert store.legacy_atomic_calls == 1
+    with pytest.raises(NotImplementedError, match="cannot create non-run"):
+        await store.create_thread_operation_atomic(
+            "checkpoint-write-1",
+            thread_id="t1",
+            owner_worker_id="worker-1",
+            lease_expires_at=None,
+            operation_kind=ThreadOperationKind.checkpoint_write,
+        )
+
+
 class TestRunRepository:
     @pytest.mark.anyio
     async def test_put_and_get(self, tmp_path):
@@ -148,6 +174,22 @@ class TestRunRepository:
         assert all(r["thread_id"] == "t1" for r in rows)
         await _cleanup()
 
+    @pytest.mark.anyio
+    async def test_run_history_excludes_internal_thread_operations(self, tmp_path):
+        repo = await _make_repo(tmp_path)
+        await repo.put("r1", thread_id="t1", status="success")
+        await repo.put(
+            "checkpoint-write-1",
+            thread_id="t1",
+            status="error",
+            operation_kind=ThreadOperationKind.checkpoint_write,
+        )
+
+        rows = await repo.list_by_thread("t1")
+
+        assert [row["run_id"] for row in rows] == ["r1"]
+        await _cleanup()
+
     @pytest.mark.anyio
     async def test_list_by_thread_owner_filter(self, tmp_path):
         repo = await _make_repo(tmp_path)
@@ -648,7 +690,7 @@ class TestRunRepository:
         await _cleanup()
 
     @pytest.mark.anyio
-    async def test_create_run_atomic_reject_propagates_conflict_on_unique_violation(self, tmp_path):
+    async def test_create_thread_operation_atomic_rejects_unique_violation(self, tmp_path):
         """reject path against a real SQLite-backed store must surface as ConflictError, not raw IntegrityError.
 
         The partial unique index ``uq_runs_thread_active`` is created by
@@ -678,7 +720,7 @@ class TestRunRepository:
         # Pre-insert an active run on thread T directly through the store so
         # the partial unique index has something to enforce on the second insert.
         lease = (datetime.now(UTC) + timedelta(seconds=30)).isoformat()
-        await repo.create_run_atomic(
+        await repo.create_thread_operation_atomic(
             "run-A",
             thread_id="thread-T",
             owner_worker_id="worker-A",
@@ -697,6 +739,65 @@ class TestRunRepository:
 
         await _cleanup()
 
+    @pytest.mark.anyio
+    async def test_checkpoint_write_reservation_blocks_interrupt_run_on_sql_store(self, tmp_path):
+        """An interrupt-strategy run cannot displace a durable checkpoint writer."""
+        repo = await _make_repo(tmp_path)
+        compaction_worker = RunManager(store=repo, worker_id="worker-a")
+        run_worker = RunManager(store=repo, worker_id="worker-b")
+
+        async with compaction_worker.reserve_thread_operation("thread-T", kind=ThreadOperationKind.checkpoint_write):
+            with pytest.raises(ConflictError, match="checkpoint write"):
+                await run_worker.create_or_reject("thread-T", multitask_strategy="interrupt")
+
+        assert await repo.list_by_thread("thread-T") == []
+        admitted = await run_worker.create_or_reject("thread-T")
+        assert admitted.status == RunStatus.pending
+        await _cleanup()
+
+    @pytest.mark.anyio
+    async def test_reservation_release_uses_record_user_without_ambient_context(self, tmp_path):
+        """Release must not depend on the request ContextVar still being set."""
+        repo = await _make_repo(tmp_path)
+        manager = RunManager(store=repo)
+
+        async with manager.reserve_thread_operation(
+            "thread-T",
+            kind=ThreadOperationKind.checkpoint_write,
+            user_id="reservation-owner",
+        ):
+            inflight = await repo.list_inflight()
+            assert len(inflight) == 1
+            assert inflight[0]["user_id"] == "reservation-owner"
+
+        assert await repo.list_inflight() == []
+        await _cleanup()
+
+    @pytest.mark.anyio
+    async def test_interrupt_reclaims_expired_checkpoint_write_reservation_on_sql_store(self, tmp_path):
+        """An expired durable checkpoint writer is immediately reclaimable."""
+        repo = await _make_repo(tmp_path)
+        expired = (datetime.now(UTC) - timedelta(seconds=30)).isoformat()
+        await repo.put(
+            "checkpoint-write-1",
+            thread_id="thread-T",
+            status="pending",
+            operation_kind=ThreadOperationKind.checkpoint_write,
+            owner_worker_id="dead-worker",
+            lease_expires_at=expired,
+            created_at=expired,
+        )
+        manager = RunManager(store=repo, worker_id="worker-b")
+
+        admitted = await manager.create_or_reject("thread-T", multitask_strategy="interrupt")
+
+        assert admitted.status == RunStatus.pending
+        stale = await repo.get("checkpoint-write-1")
+        assert stale is not None
+        assert stale["status"] == "interrupted"
+        assert stale["owner_worker_id"] == "worker-b"
+        await _cleanup()
+
     @pytest.mark.anyio
     async def test_is_unique_violation_detects_real_sqlite_integrity_error(self, tmp_path):
         """``_is_unique_violation`` must return True for a real SQLite IntegrityError.
@@ -777,12 +878,12 @@ class TestRunRepository:
         assert _is_unique_violation(sa_err) is True
 
     @pytest.mark.anyio
-    async def test_create_run_atomic_interrupt_tolerates_tz_naive_lease_on_sqlite(self, tmp_path):
+    async def test_create_thread_operation_atomic_tolerates_tz_naive_lease_on_sqlite(self, tmp_path):
         """Interrupt path must not raise TypeError comparing naive vs aware datetimes.
 
         SQLite drops tzinfo on read despite ``DateTime(timezone=True)`` (see
         the comment in ``RunRepository._row_to_dict``). The interrupt branch
-        of ``create_run_atomic`` compares ``row.lease_expires_at`` against
+        of ``create_thread_operation_atomic`` compares ``row.lease_expires_at`` against
         the aware ``cutoff = datetime.now(UTC) - ...`` in Python. Under
         default config (heartbeat disabled) leases are always NULL so the
         ``is not None`` check short-circuits, but there is no guard against
@@ -801,7 +902,7 @@ class TestRunRepository:
         # The lease value is stored as ISO; SQLite reads it back as a tz-naive
         # datetime — exactly the shape that triggered the bug.
         valid_lease = (datetime.now(UTC) + timedelta(seconds=30)).isoformat()
-        await repo.create_run_atomic(
+        await repo.create_thread_operation_atomic(
             "valid-lease-run",
             thread_id="thread-T",
             owner_worker_id="other-worker",
@@ -813,7 +914,7 @@ class TestRunRepository:
         # The interrupt path must surface a clean ConflictError, not a
         # TypeError from the naive-vs-aware comparison.
         with pytest.raises(ConflictError, match="another worker"):
-            await repo.create_run_atomic(
+            await repo.create_thread_operation_atomic(
                 "run-new",
                 thread_id="thread-T",
                 owner_worker_id="w1",

backend/tests/test_threads_router.py

@@ -1,5 +1,6 @@
 import asyncio
 import re
+from contextlib import asynccontextmanager
 from types import SimpleNamespace
 from unittest.mock import AsyncMock, patch
 
@@ -52,6 +53,15 @@ class _PermissiveThreadMetaStore(MemoryThreadMetaStore):
         return await super().search(metadata=metadata, status=status, limit=limit, offset=offset, user_id=None)
 
 
+class _ThreadTestRunManager:
+    async def list_by_thread(self, _thread_id: str, *, user_id=None, limit: int = 100) -> list:
+        return []
+
+    @asynccontextmanager
+    async def reserve_thread_operation(self, _thread_id: str, **_kwargs):
+        yield
+
+
 def _build_thread_app() -> tuple[FastAPI, InMemoryStore, InMemorySaver]:
     """Build a stub-authed FastAPI app wired with an in-memory ThreadMetaStore.
 
@@ -66,11 +76,82 @@ def _build_thread_app() -> tuple[FastAPI, InMemoryStore, InMemorySaver]:
     checkpointer = InMemorySaver()
     app.state.store = store
     app.state.checkpointer = checkpointer
+    app.state.run_manager = _ThreadTestRunManager()
     app.state.thread_store = _PermissiveThreadMetaStore(store)
     app.include_router(threads.router)
     return app, store, checkpointer
 
 
+def test_compact_rejects_run_owned_by_another_worker(monkeypatch) -> None:
+    """The HTTP guard must consult the shared store, not only local run memory."""
+    from deerflow.runtime import RunManager, RunStatus
+    from deerflow.runtime.runs.store.memory import MemoryRunStore
+
+    app, _store, _checkpointer = _build_thread_app()
+    run_store = MemoryRunStore()
+    owner = RunManager(store=run_store, worker_id="worker-a")
+    non_owner = RunManager(store=run_store, worker_id="worker-b")
+    app.state.run_manager = non_owner
+    monkeypatch.setattr(
+        threads,
+        "build_checkpoint_state_mutation_accessor",
+        lambda *_args, **_kwargs: (SimpleNamespace(), None),
+    )
+
+    async def _seed_active_run() -> None:
+        active = await owner.create_or_reject("thread-compact-race")
+        await owner.set_status(active.run_id, RunStatus.running)
+
+    asyncio.run(_seed_active_run())
+
+    with TestClient(app) as client:
+        created = client.post("/api/threads", json={"thread_id": "thread-compact-race"})
+        assert created.status_code == 200, created.text
+        response = client.post("/api/threads/thread-compact-race/compact", json={"force": True})
+
+    assert response.status_code == 409
+    assert response.json()["detail"] == "Thread has a run in flight. Compact after the run finishes."
+
+
+def test_update_state_rejects_run_owned_by_another_worker(monkeypatch) -> None:
+    """All out-of-run writes share the same durable thread-operation admission."""
+    from deerflow.runtime import RunManager, RunStatus
+    from deerflow.runtime.runs.store.memory import MemoryRunStore
+
+    app, _store, _checkpointer = _build_thread_app()
+    run_store = MemoryRunStore()
+    owner = RunManager(store=run_store, worker_id="worker-a")
+    app.state.run_manager = RunManager(store=run_store, worker_id="worker-b")
+    accessor = SimpleNamespace(
+        graph=None,
+        aupdate=AsyncMock(side_effect=AssertionError("state write must not run")),
+        aget=AsyncMock(),
+    )
+    monkeypatch.setattr(
+        threads,
+        "build_thread_checkpoint_state_mutation_accessor",
+        AsyncMock(return_value=(accessor, {"configurable": {"thread_id": "thread-state-race"}})),
+    )
+
+    async def _seed_active_run() -> None:
+        active = await owner.create_or_reject("thread-state-race")
+        await owner.set_status(active.run_id, RunStatus.running)
+
+    asyncio.run(_seed_active_run())
+
+    with TestClient(app) as client:
+        created = client.post("/api/threads", json={"thread_id": "thread-state-race"})
+        assert created.status_code == 200, created.text
+        response = client.post(
+            "/api/threads/thread-state-race/state",
+            json={"values": {"title": "must not write"}},
+        )
+
+    assert response.status_code == 409
+    assert response.json()["detail"] == "Thread has a run in flight. Update state after the run finishes."
+    accessor.aupdate.assert_not_awaited()
+
+
 class _RawStateAccessor:
     def __init__(self, checkpointer: InMemorySaver):
         self.checkpointer = checkpointer
@@ -593,6 +674,44 @@ def test_goal_status_and_clear_round_trip() -> None:
     assert "goal" not in state_response.json()["values"]
 
 
+def test_goal_mutations_reject_run_owned_by_another_worker() -> None:
+    """PUT and DELETE goal writes share the durable thread-operation boundary."""
+    from deerflow.runtime import RunManager, RunStatus
+    from deerflow.runtime.runs.store.memory import MemoryRunStore
+
+    app, _store, _checkpointer = _build_thread_app()
+    run_store = MemoryRunStore()
+    owner = RunManager(store=run_store, worker_id="worker-a")
+    app.state.run_manager = RunManager(store=run_store, worker_id="worker-b")
+    thread_id = "thread-goal-race"
+
+    async def _seed_active_run() -> None:
+        active = await owner.create_or_reject(thread_id)
+        await owner.set_status(active.run_id, RunStatus.running)
+
+    with TestClient(app) as client:
+        created = client.post("/api/threads", json={"thread_id": thread_id})
+        assert created.status_code == 200, created.text
+        initial_goal = client.put(
+            f"/api/threads/{thread_id}/goal",
+            json={"objective": "Original goal"},
+        )
+        assert initial_goal.status_code == 200, initial_goal.text
+        assert client.portal is not None
+        client.portal.call(_seed_active_run)
+        put_response = client.put(
+            f"/api/threads/{thread_id}/goal",
+            json={"objective": "Must not be written"},
+        )
+        delete_response = client.delete(f"/api/threads/{thread_id}/goal")
+        goal_response = client.get(f"/api/threads/{thread_id}/goal")
+
+    assert put_response.status_code == 409, put_response.text
+    assert delete_response.status_code == 409, delete_response.text
+    assert goal_response.status_code == 200, goal_response.text
+    assert goal_response.json()["goal"]["objective"] == "Original goal"
+
+
 def test_internal_owner_header_assigns_thread_to_owner() -> None:
     import asyncio
 
@@ -1116,7 +1235,7 @@ def test_branch_thread_can_prepare_regenerate_without_branch_run_events() -> Non
         return []
 
     app.state.run_event_store = SimpleNamespace(list_messages=list_messages)
-    app.state.run_manager = SimpleNamespace(list_by_thread=list_by_thread)
+    app.state.run_manager.list_by_thread = list_by_thread
 
     human = HumanMessage(id="human-1", content="Question", additional_kwargs={"run_id": source_run_id})
     ai = AIMessage(id="ai-1", content="Answer")
@@ -1741,7 +1860,7 @@ def test_state_endpoints_preserve_extension_reducer_channels(monkeypatch, mode)
         return []
 
     app.state.run_event_store = SimpleNamespace(list_messages=list_messages)
-    app.state.run_manager = SimpleNamespace(list_by_thread=list_by_thread)
+    app.state.run_manager.list_by_thread = list_by_thread
 
     recorded_updates: list[dict] = []
     real_mutation_builder = gateway_services.build_checkpoint_state_mutation_accessor

frontend/AGENTS.md

@@ -77,7 +77,7 @@ Composer drafts are tab-scoped browser state. `core/threads/composer-draft.ts` s
 
 Auth UI note: the login page's "keep me signed in" option submits only `remember_me` to the Gateway and may persist only the email address through `core/auth/remember-login.ts`. Passwords and tokens must never be stored in frontend storage; the `HttpOnly access_token` and readable `csrf_token` cookies remain Gateway-owned.
 
-`/goal` and `/compact` are built-in composer commands, not skill activations. `src/components/workspace/input-box.tsx` intercepts `/goal`, `/goal clear`, and `/goal <condition>` before normal chat submission, calling Gateway `GET/PUT/DELETE /api/threads/{thread_id}/goal`. Setting `/goal <condition>` also submits the condition text as the next user task so the agent starts running immediately; status and clear do not start a run. Goal and compact requests are tied to the current `threadId` with an `AbortController`, so switching threads or unmounting the composer aborts in-flight requests and stale responses cannot update the new thread's composer state. The chat pages render `GoalStatus` above the composer from `AgentThreadState.goal`, with local optimistic state until the next stream `values` update arrives. `/compact` calls `POST /api/threads/{thread_id}/compact` to summarize older active context while leaving the full visible chat history intact; it is skipped on new/empty threads and blocked server-side while a run is in flight.
+`/goal` and `/compact` are built-in composer commands, not skill activations. `src/components/workspace/input-box.tsx` intercepts `/goal`, `/goal clear`, and `/goal <condition>` before normal chat submission, calling Gateway `GET/PUT/DELETE /api/threads/{thread_id}/goal`. Setting `/goal <condition>` also submits the condition text as the next user task so the agent starts running immediately; status and clear do not start a run. Goal and compact requests are tied to the current `threadId` with an `AbortController`, so switching threads or unmounting the composer aborts in-flight requests and stale responses cannot update the new thread's composer state. The chat pages render `GoalStatus` above the composer from `AgentThreadState.goal`, with local optimistic state until the next stream `values` update arrives. `/compact` calls `POST /api/threads/{thread_id}/compact` to summarize older active context while leaving the full visible chat history intact; it is skipped on new/empty threads and blocked server-side while a run is in flight. Thread rename uses the same serialized state-write route; the rename dialog stays open and surfaces the server error when an active run returns 409.
 
 Human input requests are a structured message protocol layered on normal chat history. The backend writes request payloads to `ToolMessage.artifact.human_input`, `src/core/messages/human-input.ts` owns the runtime validators/types, and `src/components/workspace/messages/human-input-card.tsx` renders the reusable card. `MessageList` owns answered/latest/pending state for visible cards, but derives answered responses from raw `thread.messages` because replies are hidden; pending cards clear when the hidden reply appears, when dispatch is dropped, or when a new `thread.error` reports an async stream failure. Page-level submit callbacks must send a normal human message and put `hide_from_ui: true` plus the response payload in the fourth `sendMessage(..., options)` argument as `options.additionalKwargs`; the third argument remains run context such as `{ agent_name }`. Composer entry points should disable normal bottom input while `hasOpenHumanInputRequest(...)` is true so users answer through the card and preserve response metadata.
 

frontend/src/components/workspace/recent-chat-list.tsx

@@ -167,12 +167,25 @@ export function RecentChatList() {
 
   const handleRenameSubmit = useCallback(() => {
     if (renameThreadId && renameValue.trim()) {
-      renameThread({ threadId: renameThreadId, title: renameValue.trim() });
+      renameThread(
-      setRenameDialogOpen(false);
+        { threadId: renameThreadId, title: renameValue.trim() },
-      setRenameThreadId(null);
+        {
-      setRenameValue("");
+          onSuccess: () => {
+            setRenameDialogOpen(false);
+            setRenameThreadId(null);
+            setRenameValue("");
+          },
+          onError: (error) => {
+            toast.error(
+              error instanceof Error && error.message
+                ? error.message
+                : t.common.renameFailed,
+            );
+          },
+        },
+      );
     }
-  }, [renameThread, renameThreadId, renameValue]);
+  }, [renameThread, renameThreadId, renameValue, t.common.renameFailed]);
 
   const handleShare = useCallback(
     async (thread: AgentThread) => {

frontend/src/core/i18n/locales/en-US.ts

@@ -24,6 +24,7 @@ export const enUS: Translations = {
     delete: "Delete",
     edit: "Edit",
     rename: "Rename",
+    renameFailed: "Failed to rename thread.",
     share: "Share",
     openInNewWindow: "Open in new window",
     close: "Close",

frontend/src/core/i18n/locales/types.ts

@@ -13,6 +13,7 @@ export interface Translations {
     delete: string;
     edit: string;
     rename: string;
+    renameFailed: string;
     share: string;
     openInNewWindow: string;
     close: string;

frontend/src/core/i18n/locales/zh-CN.ts

@@ -24,6 +24,7 @@ export const zhCN: Translations = {
     delete: "删除",
     edit: "编辑",
     rename: "重命名",
+    renameFailed: "重命名会话失败。",
     share: "分享",
     openInNewWindow: "在新窗口打开",
     close: "关闭",

frontend/tests/unit/core/threads/api.test.ts

@@ -155,3 +155,19 @@ test("compactThreadContext posts agent attribution and abort signal", async () =
     },
   );
 });
+
+test("compactThreadContext surfaces an active-run conflict", async () => {
+  fetchWithAuth.mockResolvedValue({
+    ok: false,
+    status: 409,
+    json: async () => ({
+      detail: "Thread has a run in flight. Compact after the run finishes.",
+    }),
+  });
+
+  const { compactThreadContext } = await import("@/core/threads/api");
+
+  await expect(compactThreadContext("thread-1")).rejects.toThrow(
+    "Thread has a run in flight. Compact after the run finishes.",
+  );
+});