diff --git a/coding-agent/.opencode/prompts/tasks/local-daemon-pr2-session-record-runtime-handle.md b/coding-agent/.opencode/prompts/tasks/local-daemon-pr2-session-record-runtime-handle.md new file mode 100644 index 000000000..c52a20831 --- /dev/null +++ b/coding-agent/.opencode/prompts/tasks/local-daemon-pr2-session-record-runtime-handle.md @@ -0,0 +1,61 @@ +Goal: +Split durable session metadata from process-local runtime handles as the next +ADR-0058 implementation slice, without changing runtime behavior, persisted +session payloads, or public HTTP/CLI APIs. + +Scope: +- Introduce a narrow `SessionRuntimeHandle` structure for process-local fields + such as task, runtime pipeline/context/adapter, runtime message bus, approval + cursor, event queues, and approval coordination objects. +- Introduce or prepare a `SessionRecord` boundary for durable session metadata + while preserving the existing stored session payload shape. +- Move access inside `SessionManager` toward explicit durable-record vs runtime + handle usage in the smallest behavior-preserving slice. +- Add focused regression coverage proving persisted session data excludes + process-local handles and hydrated sessions can reacquire runtime state. + +Out of scope: +- Do not change HTTP request/response schemas. +- Do not change session store, runtime store, tape store, or checkpoint store + persisted formats. +- Do not introduce `RunCoordinator` or `LocalDaemonExecutor` behavior yet. +- Do not replace `ExecutionBinding` persistence with `RunTarget` persistence in + this slice. +- Do not change `coding_agent run` behavior. + +Context: +- ADRs: + - `docs/adr/0058-local-daemon-control-plane-executor-architecture.md` + - `docs/adr/0055-session-resume-and-interrupted-run-semantics.md` + - `docs/adr/0015-enforce-owner-routed-http-event-streams.md` +- Postmortem patterns: + - `postmortem/patterns/PM-0015-require-store-backed-requests-across-http-approval-flow.md` + - `postmortem/patterns/PM-0021-guard-event-stream-registration-against-disappearing-sessions.md` + - `postmortem/patterns/PM-0022-revalidate-event-stream-ownership-after-queue-attach.md` + - `postmortem/patterns/PM-0023-make-event-stream-cleanup-and-teardown-idempotent.md` +- Relevant files: + - `src/coding_agent/server/session_manager.py` + - `src/coding_agent/server/stores/session_store.py` + - `tests/ui/test_session_persistence.py` + - `tests/ui/test_session_manager_runtime.py` + - `tests/ui/test_session_manager_public_api.py` + - `tests/ui/test_http_server_failover.py` + +Target tests: +- `uv run pytest tests/agentkit/ tests/coding_agent/ tests/cli/ -k "session or runtime or resume or checkpoint" -v` +- `uv run pytest tests/agentkit/ tests/coding_agent/ tests/cli/ -k "event or queue or owner or close or delete" -v` +- `uv run ruff check src/coding_agent/server/session_manager.py tests/coding_agent/ tests/cli/` + +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: +- Stop if the split requires changing persisted session payloads or public API + schemas. +- Stop if `RunCoordinator`, executor ownership, or daemon lifecycle design is + needed to proceed. +- Stop after one review/fix/retest cycle. 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 952585097..64d897d2d 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 @@ -1,6 +1,6 @@ # ADR-0058: Local daemon control plane and executor architecture -**Status**: Proposed +**Status**: Accepted **Date**: 2026-06-02 ## Context @@ -121,20 +121,29 @@ This ADR does not implement that path. ## Acceptance Criteria -- [ ] `RunTarget`, `WorkspaceRef`, `ExecutorRef`, `IsolationPolicy`, and +- [x] `RunTarget`, `WorkspaceRef`, `ExecutorRef`, `IsolationPolicy`, and `RunConstraints` exist under `src/coding_agent/runs/`. -- [ ] `ExecutionBinding -> RunTarget` adapter maps local, cloud, +- [x] `ExecutionBinding -> RunTarget` adapter maps local, cloud, external-worker, and local-attached compatibility bindings without changing existing serialized `ExecutionBinding` payloads. -- [ ] New code in the run-placement skeleton imports canonical binding models +- [x] New code in the run-placement skeleton imports canonical binding models from `coding_agent.environment.execution_binding`, not server/ui aliases. -- [ ] `test_local_execution_binding_maps_to_local_daemon_run_target` -- [ ] `test_cloud_execution_binding_maps_to_managed_pool_run_target` -- [ ] `test_external_worker_binding_maps_to_external_worker_run_target` -- [ ] `test_local_attached_binding_maps_to_local_attached_run_target` -- [ ] `test_run_target_rejects_empty_annotations_key` -- [ ] `uv run pytest tests/coding_agent/test_run_target.py tests/ui/test_execution_binding.py -v` -- [ ] `uv run ruff check src/coding_agent/runs tests/coding_agent/test_run_target.py` +- [x] `test_local_execution_binding_maps_to_local_daemon_run_target` +- [x] `test_cloud_execution_binding_maps_to_managed_pool_run_target` +- [x] `test_external_worker_binding_maps_to_external_worker_run_target` +- [x] `test_local_attached_binding_maps_to_local_attached_run_target` +- [x] `test_run_target_rejects_empty_annotations_key` +- [x] `uv run pytest tests/coding_agent/test_run_target.py tests/ui/test_execution_binding.py -v` +- [x] `uv run ruff check src/coding_agent/runs tests/coding_agent/test_run_target.py` + +## Follow-up Implementation Tasks + +- Split durable `SessionRecord` data from process-local `SessionRuntimeHandle` + state. +- Introduce a `RunCoordinator` boundary that selects an executor from + `RunTarget`. +- Move runtime ownership behind `LocalDaemonExecutor`. +- Demote `coding_agent run` to an inline testkit/devkit compatibility path. ## References diff --git a/coding-agent/src/coding_agent/server/session_manager.py b/coding-agent/src/coding_agent/server/session_manager.py index 3c9562cf6..17cab9ccb 100644 --- a/coding-agent/src/coding_agent/server/session_manager.py +++ b/coding-agent/src/coding_agent/server/session_manager.py @@ -18,7 +18,7 @@ from inspect import isawaitable from math import isfinite from pathlib import Path -from typing import Any, Literal, Protocol, TypeVar, cast +from typing import Any, ClassVar, Literal, Protocol, TypeVar, cast from agentkit.environment import Environment from agentkit.storage.checkpoint_fs import FSCheckpointStore @@ -697,6 +697,150 @@ class EventBroadcastResult: failed_pruned_count: int +@dataclass +class SessionRuntimeHandle: + """Process-local runtime state associated with a session record.""" + + approval_coordinator: ApprovalCoordinator + task: asyncio.Task[Any] | None = None + pending_approval: dict[str, Any] | None = None + approval_event: asyncio.Event = field(default_factory=asyncio.Event) + approval_response: dict[str, Any] | None = None + event_queues: list[asyncio.Queue[dict[str, Any]]] = field(default_factory=list) + runtime_pipeline: Any | None = None + runtime_ctx: Any | None = None + runtime_adapter: Any | None = None + runtime_message_bus: RuntimeMessageBus = field( + default_factory=InMemoryRuntimeMessageBus + ) + approval_decision_cursor: RuntimeMessageCursor = field( + default_factory=RuntimeMessageCursor + ) + + +@dataclass(frozen=True) +class SessionRecord: + """Durable session metadata stored across process restarts.""" + + id: str + created_at: datetime + last_activity: datetime + repo_path: Path | None + origin: dict[str, str] | None + execution_binding: ExecutionBinding + approval_policy: ApprovalPolicy + provider_name: str | None + model_name: str | None + base_url: str | None + max_steps: int + tape_id: str | None + last_failure_details: str | None + + def to_store_data(self) -> dict[str, Any]: + return { + "id": self.id, + "created_at": self.created_at.isoformat(), + "last_activity": self.last_activity.isoformat(), + # repo_path remains backward-compatible metadata and seeds the + # default local binding when execution_binding is omitted. + "repo_path": None if self.repo_path is None else str(self.repo_path), + "origin": None if self.origin is None else dict(self.origin), + "execution_binding": self.execution_binding.to_dict(), + "approval_policy": self.approval_policy.value, + "provider_name": self.provider_name, + "model_name": self.model_name, + "base_url": self.base_url, + "max_steps": self.max_steps, + "tape_id": self.tape_id, + "last_failure_details": self.last_failure_details, + } + + @classmethod + def from_store_data(cls, data: dict[str, Any]) -> SessionRecord: + repo_path_raw = data.get("repo_path") + if repo_path_raw is not None and not isinstance(repo_path_raw, str): + raise TypeError("session metadata has invalid repo_path") + origin_raw = data.get("origin") + if origin_raw is not None: + if not isinstance(origin_raw, dict): + raise TypeError("session metadata has invalid origin") + origin = { + key: _required_session_str(cast(dict[str, Any], origin_raw), key) + for key in cast(dict[str, Any], origin_raw) + } + else: + origin = None + approval_policy_raw = data.get("approval_policy") + if not isinstance(approval_policy_raw, str): + raise TypeError("session metadata is missing approval_policy") + provider_name_raw = data.get("provider_name") + if provider_name_raw is not None and not isinstance(provider_name_raw, str): + raise TypeError("session metadata has invalid provider_name") + model_name_raw = data.get("model_name") + if model_name_raw is not None and not isinstance(model_name_raw, str): + raise TypeError("session metadata has invalid model_name") + base_url_raw = data.get("base_url") + if base_url_raw is not None and not isinstance(base_url_raw, str): + raise TypeError("session metadata has invalid base_url") + tape_id_raw = data.get("tape_id") + if tape_id_raw is not None and not isinstance(tape_id_raw, str): + raise TypeError("session metadata has invalid tape_id") + last_failure_details_raw = data.get("last_failure_details") + if last_failure_details_raw is not None and not isinstance( + last_failure_details_raw, str + ): + raise TypeError("session metadata has invalid last_failure_details") + binding_raw = data.get("execution_binding") + if binding_raw is not None: + if not isinstance(binding_raw, dict): + raise TypeError("session metadata has invalid execution_binding") + execution_binding = ExecutionBinding.from_dict(binding_raw) + else: + workspace_root = ( + str(Path(repo_path_raw).resolve()) + if repo_path_raw is not None + else str(Path.cwd().resolve()) + ) + execution_binding = LocalExecutionBinding(workspace_root=workspace_root) + return cls( + id=_required_session_str(data, "id"), + created_at=datetime.fromisoformat( + _required_session_str(data, "created_at") + ), + last_activity=datetime.fromisoformat( + _required_session_str(data, "last_activity") + ), + repo_path=None if repo_path_raw is None else Path(repo_path_raw), + origin=origin, + execution_binding=execution_binding, + approval_policy=ApprovalPolicy(approval_policy_raw), + provider_name=provider_name_raw, + model_name=model_name_raw, + base_url=base_url_raw, + max_steps=_required_session_int(data, "max_steps"), + tape_id=tape_id_raw, + last_failure_details=last_failure_details_raw, + ) + + def to_session(self) -> Session: + return Session( + id=self.id, + created_at=self.created_at, + last_activity=self.last_activity, + approval_store=ApprovalStore(), + repo_path=self.repo_path, + origin=self.origin, + execution_binding=self.execution_binding, + approval_policy=self.approval_policy, + provider_name=self.provider_name, + model_name=self.model_name, + base_url=self.base_url, + max_steps=self.max_steps, + tape_id=self.tape_id, + last_failure_details=self.last_failure_details, + ) + + @dataclass class Session: """A managed agent session. @@ -722,6 +866,7 @@ class Session: model_name: str | None = None base_url: str | None = None max_steps: int = 30 + runtime_handle: SessionRuntimeHandle = field(init=False, repr=False) task: asyncio.Task[Any] | None = None turn_in_progress: bool = False turn_status: TurnStatus = "idle" @@ -743,6 +888,48 @@ class Session: ) approval_coordinator: ApprovalCoordinator = field(init=False) + _RUNTIME_HANDLE_FIELD_NAMES: ClassVar[frozenset[str]] = frozenset( + { + "task", + "pending_approval", + "approval_event", + "approval_response", + "event_queues", + "runtime_pipeline", + "runtime_ctx", + "runtime_adapter", + "runtime_message_bus", + "approval_decision_cursor", + "approval_coordinator", + } + ) + + def __getattribute__(self, name: str) -> Any: + if name in object.__getattribute__(self, "_RUNTIME_HANDLE_FIELD_NAMES"): + instance_dict = object.__getattribute__(self, "__dict__") + handle = instance_dict.get("runtime_handle") + if handle is not None: + return getattr(handle, name) + return object.__getattribute__(self, name) + + def __setattr__(self, name: str, value: object) -> None: + if name == "approval_store": + object.__setattr__(self, name, value) + instance_dict = object.__getattribute__(self, "__dict__") + handle = instance_dict.get("runtime_handle") + if handle is not None: + handle.approval_coordinator = ApprovalCoordinator( + cast(ApprovalStore, value) + ) + return + if name in self._RUNTIME_HANDLE_FIELD_NAMES: + instance_dict = object.__getattribute__(self, "__dict__") + handle = instance_dict.get("runtime_handle") + if handle is not None: + setattr(handle, name, value) + return + object.__setattr__(self, name, value) + def __post_init__(self) -> None: if self.execution_binding is cast(ExecutionBinding, _DEFAULT_EXECUTION_BINDING): workspace_root = ( @@ -754,7 +941,23 @@ def __post_init__(self) -> None: workspace_root=workspace_root ) self.wire = LocalWire(self.id) - self.approval_coordinator = ApprovalCoordinator(self.approval_store) + handle = SessionRuntimeHandle( + approval_coordinator=ApprovalCoordinator(self.approval_store), + task=self.task, + pending_approval=self.pending_approval, + approval_event=self.approval_event, + approval_response=self.approval_response, + event_queues=self.event_queues, + runtime_pipeline=self.runtime_pipeline, + runtime_ctx=self.runtime_ctx, + runtime_adapter=self.runtime_adapter, + runtime_message_bus=self.runtime_message_bus, + approval_decision_cursor=self.approval_decision_cursor, + ) + object.__setattr__(self, "runtime_handle", handle) + instance_dict = object.__getattribute__(self, "__dict__") + for field_name in self._RUNTIME_HANDLE_FIELD_NAMES: + instance_dict.pop(field_name, None) def broadcast_event_nowait(self, event: dict[str, Any]) -> EventBroadcastResult: active_queues: list[asyncio.Queue[dict[str, Any]]] = [] @@ -829,91 +1032,28 @@ def as_dict(self) -> dict[str, Any]: } def to_store_data(self) -> dict[str, Any]: - return { - "id": self.id, - "created_at": self.created_at.isoformat(), - "last_activity": self.last_activity.isoformat(), - # repo_path remains backward-compatible metadata and seeds the - # default local binding when execution_binding is omitted. - "repo_path": None if self.repo_path is None else str(self.repo_path), - "origin": None if self.origin is None else dict(self.origin), - "execution_binding": self.execution_binding.to_dict(), - "approval_policy": self.approval_policy.value, - "provider_name": self.provider_name, - "model_name": self.model_name, - "base_url": self.base_url, - "max_steps": self.max_steps, - "tape_id": self.tape_id, - "last_failure_details": self.last_failure_details, - } + return self.to_record().to_store_data() + + def to_record(self) -> SessionRecord: + return SessionRecord( + id=self.id, + created_at=self.created_at, + last_activity=self.last_activity, + repo_path=self.repo_path, + origin=None if self.origin is None else dict(self.origin), + execution_binding=self.execution_binding, + approval_policy=self.approval_policy, + provider_name=self.provider_name, + model_name=self.model_name, + base_url=self.base_url, + max_steps=self.max_steps, + tape_id=self.tape_id, + last_failure_details=self.last_failure_details, + ) @classmethod def from_store_data(cls, data: dict[str, Any]) -> Session: - repo_path_raw = data.get("repo_path") - if repo_path_raw is not None and not isinstance(repo_path_raw, str): - raise TypeError("session metadata has invalid repo_path") - origin_raw = data.get("origin") - if origin_raw is not None: - if not isinstance(origin_raw, dict): - raise TypeError("session metadata has invalid origin") - origin = { - key: _required_session_str(cast(dict[str, Any], origin_raw), key) - for key in cast(dict[str, Any], origin_raw) - } - else: - origin = None - approval_policy_raw = data.get("approval_policy") - if not isinstance(approval_policy_raw, str): - raise TypeError("session metadata is missing approval_policy") - provider_name_raw = data.get("provider_name") - if provider_name_raw is not None and not isinstance(provider_name_raw, str): - raise TypeError("session metadata has invalid provider_name") - model_name_raw = data.get("model_name") - if model_name_raw is not None and not isinstance(model_name_raw, str): - raise TypeError("session metadata has invalid model_name") - base_url_raw = data.get("base_url") - if base_url_raw is not None and not isinstance(base_url_raw, str): - raise TypeError("session metadata has invalid base_url") - tape_id_raw = data.get("tape_id") - if tape_id_raw is not None and not isinstance(tape_id_raw, str): - raise TypeError("session metadata has invalid tape_id") - last_failure_details_raw = data.get("last_failure_details") - if last_failure_details_raw is not None and not isinstance( - last_failure_details_raw, str - ): - raise TypeError("session metadata has invalid last_failure_details") - binding_raw = data.get("execution_binding") - if binding_raw is not None: - if not isinstance(binding_raw, dict): - raise TypeError("session metadata has invalid execution_binding") - execution_binding = ExecutionBinding.from_dict(binding_raw) - else: - workspace_root = ( - str(Path(repo_path_raw).resolve()) - if repo_path_raw is not None - else str(Path.cwd().resolve()) - ) - execution_binding = LocalExecutionBinding(workspace_root=workspace_root) - session = cls( - id=_required_session_str(data, "id"), - created_at=datetime.fromisoformat( - _required_session_str(data, "created_at") - ), - last_activity=datetime.fromisoformat( - _required_session_str(data, "last_activity") - ), - approval_store=ApprovalStore(), - repo_path=None if repo_path_raw is None else Path(repo_path_raw), - origin=origin, - execution_binding=execution_binding, - approval_policy=ApprovalPolicy(approval_policy_raw), - provider_name=provider_name_raw, - model_name=model_name_raw, - base_url=base_url_raw, - max_steps=_required_session_int(data, "max_steps"), - tape_id=tape_id_raw, - last_failure_details=last_failure_details_raw, - ) + session = SessionRecord.from_store_data(data).to_session() session.turn_in_progress = False session.pending_approval = None session.approval_response = None diff --git a/coding-agent/tests/ui/test_session_persistence.py b/coding-agent/tests/ui/test_session_persistence.py index 88dd41a28..07ae98196 100644 --- a/coding-agent/tests/ui/test_session_persistence.py +++ b/coding-agent/tests/ui/test_session_persistence.py @@ -11,7 +11,12 @@ import pytest from coding_agent.approval.store import ApprovalStore -from coding_agent.server.session_manager import Session, SessionManager +from coding_agent.server.session_manager import ( + Session, + SessionManager, + SessionRecord, + SessionRuntimeHandle, +) from coding_agent.server.stores.session_store import ( InMemorySessionStore, PGSessionMetadataStore, @@ -23,6 +28,7 @@ PGWorkspaceMetadataStore, WorkspaceRecord, ) +from coding_agent.wire.protocol import ApprovalRequest def test_register_session_uses_public_api() -> None: @@ -1240,6 +1246,133 @@ def test_rehydrate_clears_non_restart_safe_runtime_state() -> None: assert reloaded.approval_response is None +def test_session_runtime_state_lives_on_runtime_handle() -> None: + session = Session( + id="runtime-handle-session", + created_at=datetime.now(), + last_activity=datetime.now(), + approval_store=ApprovalStore(), + ) + runtime_ctx = object() + runtime_adapter = object() + + session.runtime_ctx = runtime_ctx + session.runtime_adapter = runtime_adapter + session.event_queues.append(asyncio.Queue()) + session.pending_approval = {"request_id": "req-123", "tool_name": "bash"} + + assert isinstance(session.runtime_handle, SessionRuntimeHandle) + assert session.runtime_handle.runtime_ctx is runtime_ctx + assert session.runtime_handle.runtime_adapter is runtime_adapter + assert session.runtime_handle.event_queues == session.event_queues + assert session.runtime_handle.pending_approval == { + "request_id": "req-123", + "tool_name": "bash", + } + instance_dict = object.__getattribute__(session, "__dict__") + assert "runtime_ctx" not in instance_dict + assert "runtime_adapter" not in instance_dict + assert "event_queues" not in instance_dict + assert "pending_approval" not in instance_dict + + +def test_session_store_data_excludes_runtime_handle_state() -> None: + session = Session( + id="runtime-handle-session", + created_at=datetime.now(), + last_activity=datetime.now(), + approval_store=ApprovalStore(), + ) + session.runtime_pipeline = object() + session.runtime_ctx = object() + session.runtime_adapter = object() + session.event_queues.append(asyncio.Queue()) + session.pending_approval = {"request_id": "req-123", "tool_name": "bash"} + + payload = session.to_store_data() + + assert "runtime_handle" not in payload + assert "runtime_pipeline" not in payload + assert "runtime_ctx" not in payload + assert "runtime_adapter" not in payload + assert "event_queues" not in payload + assert "pending_approval" not in payload + + +def test_session_record_round_trips_existing_store_payload() -> None: + session = Session( + id="record-session", + created_at=datetime.now(UTC), + last_activity=datetime.now(UTC), + approval_store=ApprovalStore(), + provider_name="openai", + model_name="gpt-4o", + base_url="https://api.example.test", + max_steps=12, + tape_id="tape-123", + last_failure_details="previous failure", + ) + + payload = session.to_store_data() + record = SessionRecord.from_store_data(payload) + reloaded = record.to_session() + + assert record.to_store_data() == payload + assert reloaded.id == session.id + assert reloaded.provider_name == "openai" + assert reloaded.model_name == "gpt-4o" + assert reloaded.base_url == "https://api.example.test" + assert reloaded.max_steps == 12 + assert reloaded.tape_id == "tape-123" + assert reloaded.last_failure_details == "previous failure" + assert isinstance(reloaded.runtime_handle, SessionRuntimeHandle) + assert reloaded.runtime_ctx is None + assert reloaded.event_queues == [] + + +def test_session_record_excludes_process_local_runtime_state() -> None: + session = Session( + id="record-session", + created_at=datetime.now(UTC), + last_activity=datetime.now(UTC), + approval_store=ApprovalStore(), + ) + session.runtime_ctx = object() + session.runtime_adapter = object() + session.event_queues.append(asyncio.Queue()) + session.pending_approval = {"request_id": "req-123", "tool_name": "bash"} + + record = session.to_record() + + assert isinstance(record, SessionRecord) + assert not hasattr(record, "runtime_handle") + assert not hasattr(record, "runtime_ctx") + assert not hasattr(record, "event_queues") + assert not hasattr(record, "pending_approval") + + +def test_session_rebinds_approval_coordinator_when_store_changes() -> None: + session = Session( + id="approval-rebind-session", + created_at=datetime.now(UTC), + last_activity=datetime.now(UTC), + approval_store=ApprovalStore(), + ) + replacement_store = ApprovalStore() + request = ApprovalRequest( + session_id=session.id, + request_id="req-rebound", + tool="bash", + args={"cmd": "pwd"}, + ) + + session.approval_store = replacement_store + session.approval_coordinator.add_request(request) + + assert session.runtime_handle.approval_coordinator is session.approval_coordinator + assert replacement_store.get_request("req-rebound") == request + + def test_redis_session_store_reports_health_from_ping() -> None: class FakeRedisClient: def ping(self) -> bool: