refactor: delegate runtime maintenance admission

coding-agent/.opencode/prompts/tasks/runtime-maintenance-admission-service-task-packet.md

@@ -0,0 +1,51 @@
+Goal:
+Move local runtime maintenance turn admission out of `SessionManager` so
+checkpoint capture, checkpoint restore, and runtime config replacement share a
+small service boundary for turn-lock, owner, session-load, and active-turn
+checks.
+
+Scope:
+- Add a `RuntimeMaintenanceAdmissionService` under `coding_agent.runs` with an
+  exclusive admission helper for maintenance operations.
+- Delegate `SessionManager.replace_session_runtime_config()`,
+  `SessionManager.capture_checkpoint()`, and
+  `SessionManager.restore_checkpoint()` admission checks to the new service.
+- Add focused service tests and update ADR-0058 follow-up status.
+
+Out of scope:
+- Do not change checkpoint payload shape, tape truncation, runtime replacement,
+  HTTP event stream behavior, daemon CLI behavior, or executor selection.
+- Do not change `run_agent()` or `prepare_session_turn()` behavior already owned
+  by `RuntimeTurnAdmissionService`.
+- Do not touch untracked `webui/` or contract fixture workspaces.
+
+Context:
+- ADRs:
+  - `docs/adr/0058-local-daemon-control-plane-executor-architecture.md`
+- Relevant files:
+  - `src/coding_agent/runs/lifecycle.py`
+  - `src/coding_agent/runs/__init__.py`
+  - `src/coding_agent/server/session_manager.py`
+  - `tests/coding_agent/test_runtime_maintenance_admission_service.py`
+  - `tests/ui/test_session_manager_runtime.py`
+- Postmortem notes:
+  - `postmortem/patterns/PM-0022-revalidate-event-stream-ownership-after-queue-attach.md`
+  - `postmortem/patterns/PM-0023-make-event-stream-cleanup-and-teardown-idempotent.md`
+
+Target tests:
+- `uv run pytest tests/coding_agent/test_runtime_maintenance_admission_service.py -v`
+- `uv run pytest tests/ui/test_session_manager_runtime.py -k "checkpoint or runtime_config or turn_in_progress" -v`
+- `uv run ruff check src/coding_agent/runs/lifecycle.py src/coding_agent/runs/__init__.py src/coding_agent/server/session_manager.py tests/coding_agent/test_runtime_maintenance_admission_service.py`
+- `uv run ruff format --check src/coding_agent/runs/lifecycle.py src/coding_agent/runs/__init__.py src/coding_agent/server/session_manager.py tests/coding_agent/test_runtime_maintenance_admission_service.py`
+
+Loop policy:
+- Engineer implements the smallest correct change and runs the target tests.
+- Reviewer reviews only the resulting diff and affected tests.
+- Reviewer reports only P1/P2 findings.
+- Engineer fixes only accepted P1/P2 findings and reruns the same target tests.
+- Verifier reruns the exact target tests and reports pass/fail only.
+
+Stop conditions:
+- At most one review/fix/retest cycle.
+- Escalate architectural redirection or scope expansion to the human.
+- Ignore non-blocking optimization suggestions.

coding-agent/docs/adr/0058-local-daemon-control-plane-executor-architecture.md

@@ -346,6 +346,10 @@ This ADR does not implement that path.
     active-turn checks, is delegated to `RuntimeTurnAdmissionService` instead
     of living inline in `SessionManager.prepare_session_turn()` and
     `SessionManager.run_agent()`.
+  - Completed: local runtime maintenance admission for runtime config
+    replacement, checkpoint capture, and checkpoint restore is delegated to
+    `RuntimeMaintenanceAdmissionService` instead of living inline in
+    `SessionManager`.
   - Completed: runtime control service construction for persistence, recovery,
     query, resume, attached-executor, cancel, and task-stopper services is
     delegated to `RuntimeControlServices` instead of `SessionManager` private

coding-agent/src/coding_agent/runs/__init__.py

@@ -55,6 +55,9 @@
     RuntimeEnsureSession,
 )
 from .lifecycle import (
+    RuntimeMaintenanceAdmissionBody,
+    RuntimeMaintenanceAdmissionService,
+    RuntimeMaintenanceAdmissionSession,
     RuntimeMessageSnapshotSaver,
     RuntimeObservationCompleter,
     RuntimeObservationStarter,
@@ -241,6 +244,9 @@
     "SessionRuntimeHandle",
     "RuntimeExecution",
     "RuntimeMessageSnapshotSession",
+    "RuntimeMaintenanceAdmissionBody",
+    "RuntimeMaintenanceAdmissionService",
+    "RuntimeMaintenanceAdmissionSession",
     "RuntimeMessageSnapshotSaver",
     "RuntimeMetadataSession",
     "RuntimeObservationService",

coding-agent/src/coding_agent/runs/lifecycle.py

@@ -152,6 +152,11 @@ class RuntimeTurnAdmissionSession(Protocol):
     [RuntimeTurnAdmissionSession],
     Awaitable[Any],
 ]
+RuntimeMaintenanceAdmissionSession = RuntimeTurnAdmissionSession
+RuntimeMaintenanceAdmissionBody = Callable[
+    [RuntimeMaintenanceAdmissionSession],
+    Awaitable[Any],
+]
 
 
 class RuntimeHandleSession(Protocol):
@@ -331,6 +336,31 @@ async def run_exclusive(
             return await body(session)
 
 
+@dataclass(frozen=True)
+class RuntimeMaintenanceAdmissionService:
+    turn_lock_for: RuntimeTurnLockProvider
+    assert_owner: RuntimeOwnerAsserter
+    load_session: RuntimeTurnAdmissionSessionLoader
+
+    async def run_exclusive(
+        self,
+        session_id: str,
+        body: RuntimeMaintenanceAdmissionBody,
+    ) -> Any:
+        lock = self.turn_lock_for(session_id)
+        if lock.locked():
+            raise RuntimeError("turn already in progress")
+
+        async with lock:
+            await self.assert_owner(session_id)
+            session = await self.load_session(session_id)
+            if session.turn_in_progress or (
+                session.task is not None and not session.task.done()
+            ):
+                raise RuntimeError("turn already in progress")
+            return await body(session)
+
+
 @dataclass(frozen=True)
 class RuntimeRunLifecycle:
     store: RuntimeRunLifecycleStore | None
@@ -828,6 +858,9 @@ async def complete(
 
 
 __all__ = [
+    "RuntimeMaintenanceAdmissionBody",
+    "RuntimeMaintenanceAdmissionService",
+    "RuntimeMaintenanceAdmissionSession",
     "RuntimeMessageSnapshotSaver",
     "RuntimeObservationCompleter",
     "RuntimeRunLifecycle",

coding-agent/src/coding_agent/server/session_manager.py

@@ -79,6 +79,7 @@
     RuntimeCloser,
     RuntimeContextBindingService,
     RuntimeEnsureService,
+    RuntimeMaintenanceAdmissionService,
     RuntimeRunMetadataService,
     RuntimeObservationService,
     RuntimePreparationRequestService,
@@ -825,6 +826,11 @@ async def runtime_session_is_recoverable(session_id: str) -> bool:
             assert_owner=self._assert_owner,
             load_session=self.get_session_async,
         )
+        self._runtime_maintenance_admission = RuntimeMaintenanceAdmissionService(
+            turn_lock_for=self._turn_lock_for,
+            assert_owner=self._assert_owner,
+            load_session=self.get_session_async,
+        )
         self._runtime_closer = RuntimeCloser()
         self._runtime_agent_factory_service = RuntimeAgentFactoryService(
             create_agent=self._create_agent,
@@ -2733,58 +2739,56 @@ async def replace_session_runtime_config(
         *,
         model_name: str,
     ) -> Session:
-        turn_lock = self._turn_lock_for(session_id)
-        if turn_lock.locked():
-            raise RuntimeError("turn already in progress")
-
-        async with turn_lock:
-            await self._assert_owner(session_id)
-            session = await self.get_session_async(session_id)
-            if session.task and not session.task.done():
-                raise RuntimeError("turn already in progress")
-            if session.turn_in_progress:
-                raise RuntimeError("turn already in progress")
-
+        async def replace_admitted_runtime(session: object) -> Session:
+            admitted_session = cast(Session, session)
             return await self._runtime_replacement_service.replace_runtime_config(
-                session,
+                admitted_session,
                 model_name=model_name,
                 build_runtime=self._build_session_runtime,
                 persist_session=self._persist_session_async,
             )
 
+        return cast(
+            Session,
+            await self._runtime_maintenance_admission.run_exclusive(
+                session_id,
+                replace_admitted_runtime,
+            ),
+        )
+
     async def capture_checkpoint(
         self,
         session_id: str,
         *,
         label: str | None = None,
         extra: dict[str, Any] | None = None,
     ) -> CheckpointMeta:
-        turn_lock = self._turn_lock_for(session_id)
-        if turn_lock.locked():
-            raise RuntimeError("turn already in progress")
-
-        async with turn_lock:
-            await self._assert_owner(session_id)
-            session = await self.get_session_async(session_id)
-            if session.turn_in_progress or (session.task and not session.task.done()):
-                raise RuntimeError("turn already in progress")
-
+        async def capture_admitted_checkpoint(session: object) -> CheckpointMeta:
+            admitted_session = cast(Session, session)
             ctx = await self.ensure_session_runtime(session_id)
             payload = dict(extra or {})
             if _CHECKPOINT_SESSION_CONFIG_KEY in payload:
                 raise ValueError(
                     f"'{_CHECKPOINT_SESSION_CONFIG_KEY}' is a reserved checkpoint metadata key and cannot be provided via extra"
                 )
             payload[_CHECKPOINT_SESSION_CONFIG_KEY] = (
-                serialize_checkpoint_session_config(session)
+                serialize_checkpoint_session_config(admitted_session)
             )
             checkpoint = await self._checkpoint_service.capture(
                 ctx, label=label, extra=payload
             )
-            session.tape_id = ctx.tape.tape_id
-            await self._persist_session_async(session)
+            admitted_session.tape_id = ctx.tape.tape_id
+            await self._persist_session_async(admitted_session)
             return checkpoint
 
+        return cast(
+            CheckpointMeta,
+            await self._runtime_maintenance_admission.run_exclusive(
+                session_id,
+                capture_admitted_checkpoint,
+            ),
+        )
+
     async def export_workspace_archive(
         self,
         session_id: str,
@@ -2815,13 +2819,10 @@ async def list_checkpoints(self, session_id: str) -> list[CheckpointMeta]:
         return await self._checkpoint_service.list(session.tape_id)
 
     async def restore_checkpoint(self, session_id: str, checkpoint_id: str) -> None:
-        turn_lock = self._turn_lock_for(session_id)
-        if turn_lock.locked():
-            raise RuntimeError("turn already in progress")
+        async def restore_admitted_checkpoint(session: object) -> None:
+            await self._restore_checkpoint(cast(Session, session), checkpoint_id)
 
-        async with turn_lock:
-            await self._assert_owner(session_id)
-            session = await self.get_session_async(session_id)
-            if session.turn_in_progress or (session.task and not session.task.done()):
-                raise RuntimeError("turn already in progress")
-            await self._restore_checkpoint(session, checkpoint_id)
+        await self._runtime_maintenance_admission.run_exclusive(
+            session_id,
+            restore_admitted_checkpoint,
+        )

coding-agent/tests/coding_agent/test_runtime_maintenance_admission_service.py

@@ -0,0 +1,130 @@
+from __future__ import annotations
+
+import asyncio
+from dataclasses import dataclass
+
+import pytest
+
+from coding_agent.runs import RuntimeMaintenanceAdmissionService
+
+
+@dataclass
+class FakeTask:
+    is_done: bool = False
+
+    def cancel(self) -> None:
+        pass
+
+    def done(self) -> bool:
+        return self.is_done
+
+    def __await__(self):
+        if False:
+            yield None
+        return None
+
+
+@dataclass
+class FakeSession:
+    turn_in_progress: bool = False
+    task: FakeTask | None = None
+
+
+def _service(
+    *,
+    session: FakeSession | None = None,
+    lock: asyncio.Lock | None = None,
+    calls: list[str] | None = None,
+) -> RuntimeMaintenanceAdmissionService:
+    current_session = session or FakeSession()
+    current_lock = lock or asyncio.Lock()
+
+    async def assert_owner(session_id: str) -> None:
+        if calls is not None:
+            calls.append(f"owner:{session_id}")
+
+    async def load_session(session_id: str) -> FakeSession:
+        if calls is not None:
+            calls.append(f"load:{session_id}")
+        return current_session
+
+    return RuntimeMaintenanceAdmissionService(
+        turn_lock_for=lambda session_id: current_lock,
+        assert_owner=assert_owner,
+        load_session=load_session,
+    )
+
+
+@pytest.mark.asyncio
+async def test_run_exclusive_runs_body_under_turn_lock() -> None:
+    lock = asyncio.Lock()
+    calls: list[str] = []
+
+    async def body(session: object) -> str:
+        calls.append(f"body_locked:{lock.locked()}")
+        assert isinstance(session, FakeSession)
+        return "done"
+
+    result = await _service(lock=lock, calls=calls).run_exclusive("session-1", body)
+
+    assert result == "done"
+    assert calls == ["owner:session-1", "load:session-1", "body_locked:True"]
+    assert lock.locked() is False
+
+
+@pytest.mark.asyncio
+async def test_run_exclusive_rejects_active_lock_without_owner_or_load() -> None:
+    lock = asyncio.Lock()
+    await lock.acquire()
+    calls: list[str] = []
+
+    async def body(session: object) -> None:
+        del session
+        calls.append("body")
+
+    try:
+        with pytest.raises(RuntimeError, match="turn already in progress"):
+            await _service(lock=lock, calls=calls).run_exclusive("session-1", body)
+    finally:
+        lock.release()
+
+    assert calls == []
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize(
+    "session",
+    [
+        FakeSession(turn_in_progress=True),
+        FakeSession(task=FakeTask(is_done=False)),
+    ],
+)
+async def test_run_exclusive_rejects_active_session_before_body(
+    session: FakeSession,
+) -> None:
+    calls: list[str] = []
+
+    async def body(admitted_session: object) -> None:
+        del admitted_session
+        calls.append("body")
+
+    with pytest.raises(RuntimeError, match="turn already in progress"):
+        await _service(session=session, calls=calls).run_exclusive(
+            "session-1",
+            body,
+        )
+
+    assert calls == ["owner:session-1", "load:session-1"]
+
+
+@pytest.mark.asyncio
+async def test_run_exclusive_allows_completed_task() -> None:
+    session = FakeSession(task=FakeTask(is_done=True))
+
+    async def body(admitted_session: object) -> FakeSession:
+        assert admitted_session is session
+        return session
+
+    admitted = await _service(session=session).run_exclusive("session-1", body)
+
+    assert admitted is session