diff --git a/coding-agent/.opencode/prompts/tasks/runtime-maintenance-admission-service-task-packet.md b/coding-agent/.opencode/prompts/tasks/runtime-maintenance-admission-service-task-packet.md new file mode 100644 index 000000000..1b9843613 --- /dev/null +++ b/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. diff --git a/coding-agent/docs/adr/0058-local-daemon-control-plane-executor-architecture.md b/coding-agent/docs/adr/0058-local-daemon-control-plane-executor-architecture.md index 6f7f2c9aa..c89033fa8 100644 --- a/coding-agent/docs/adr/0058-local-daemon-control-plane-executor-architecture.md +++ b/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 diff --git a/coding-agent/src/coding_agent/runs/__init__.py b/coding-agent/src/coding_agent/runs/__init__.py index c6ce08f82..883a66484 100644 --- a/coding-agent/src/coding_agent/runs/__init__.py +++ b/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", diff --git a/coding-agent/src/coding_agent/runs/lifecycle.py b/coding-agent/src/coding_agent/runs/lifecycle.py index 32b6e0225..7eb163d89 100644 --- a/coding-agent/src/coding_agent/runs/lifecycle.py +++ b/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", diff --git a/coding-agent/src/coding_agent/server/session_manager.py b/coding-agent/src/coding_agent/server/session_manager.py index b2b3e062b..b98e5322c 100644 --- a/coding-agent/src/coding_agent/server/session_manager.py +++ b/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,25 +2739,23 @@ 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, @@ -2759,16 +2763,8 @@ async def capture_checkpoint( 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: @@ -2776,15 +2772,23 @@ async def capture_checkpoint( 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, + ) diff --git a/coding-agent/tests/coding_agent/test_runtime_maintenance_admission_service.py b/coding-agent/tests/coding_agent/test_runtime_maintenance_admission_service.py new file mode 100644 index 000000000..616999cc4 --- /dev/null +++ b/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