diff --git a/coding-agent/.opencode/prompts/tasks/local-daemon-pr30-local-runtime-provider.md b/coding-agent/.opencode/prompts/tasks/local-daemon-pr30-local-runtime-provider.md new file mode 100644 index 000000000..310ba834f --- /dev/null +++ b/coding-agent/.opencode/prompts/tasks/local-daemon-pr30-local-runtime-provider.md @@ -0,0 +1,47 @@ +Goal: +Move local daemon runtime preparation out of SessionManager into a dedicated +runtime provider boundary, while preserving the current run path behavior. + +Scope: +- Add a LocalDaemonSessionRuntimeProvider under coding_agent.executors. +- Move local target environment resolution, cached runtime reuse validation, + local runtime creation, tape restoration, consumer binding setup, and binding + construction from SessionManager into that provider. +- Keep SessionManager.run_agent() as the public control-plane method, but make + it delegate runtime preparation through the provider. +- Keep LocalDaemonExecutor and RunCoordinator behavior unchanged. +- Refresh ADR-0058 follow-up status for the runtime preparation ownership slice. + +Out of scope: +- Change local daemon runtime execution semantics. +- Change checkpoint restore runtime preparation. +- Move run lifecycle bookkeeping, observation callbacks, approval handling, or + message snapshot persistence out of SessionManager. +- Add a daemon process/client product path. +- Change store schemas, HTTP routes, CLI behavior, or live stream protocols. + +Context: +- ADRs: + - docs/adr/0058-local-daemon-control-plane-executor-architecture.md +- Relevant files: + - src/coding_agent/executors/local_daemon.py + - src/coding_agent/server/session_manager.py + - tests/ui/test_session_manager_runtime.py + +Target tests: +- uv run pytest tests/coding_agent/test_local_daemon_executor.py -v +- uv run pytest tests/ui/test_session_manager_runtime.py -k "local_daemon_executor or coordinator_integration or default_run_target_workspace or rebuilds_live_runtime_when_default_run_target_changes" -v +- uv run ruff check src/coding_agent/executors src/coding_agent/server/session_manager.py tests/ui/test_session_manager_runtime.py +- uv run ruff format --check src/coding_agent/executors src/coding_agent/server/session_manager.py tests/coding_agent/test_local_daemon_executor.py +- git diff --check + +Loop policy: +- Engineer implements the smallest correct change and runs target tests. +- Reviewer reports only P1/P2 runtime ownership, behavior regression, + compatibility, import-boundary, or test-gap findings. +- Engineer fixes only accepted P1/P2 findings and reruns the same target tests. + +Stop conditions: +- Stop if this requires changing public HTTP/CLI behavior. +- Stop if this requires changing persisted payloads or runtime store schemas. +- Stop if this expands into daemon process/client implementation. 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 66aa0c018..358cf6c46 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 @@ -151,12 +151,17 @@ This ADR does not implement that path. - Unsupported managed runtime execution is rejected by `RunCoordinator`, not prefiltered by `SessionManager`. - [~] Move runtime ownership behind `LocalDaemonExecutor`. - - Completed: local daemon runtime preparation, checkpoint restore runtime - preparation, and runtime execution route through `LocalDaemonExecutor`. + - Completed: local daemon runtime execution routes through + `RunCoordinator.execute_runtime()` and `LocalDaemonExecutor`. + - Completed: normal local runtime preparation is delegated from + `SessionManager` to `LocalDaemonSessionRuntimeProvider`. + - Completed: checkpoint restore runtime preparation still routes through + `LocalDaemonExecutor.prepare_runtime()`. - Remaining: `SessionManager` still owns run lifecycle bookkeeping, - observation callbacks, wire consumer setup, and some runtime close/error - policy. These should move behind narrower RunService/EventStore/Executor - lifecycle boundaries before this item is considered complete. + checkpoint restore preparation details, observation callbacks, wire consumer + setup, and some runtime close/error policy. These should move behind + narrower RunService/EventStore/Executor lifecycle boundaries before this + item is considered complete. - [x] Demote `coding_agent run` to an inline testkit/devkit compatibility path. - `run` records `origin.mode = inline_testkit`. - CLI and README describe `run` as dev/testkit one-shot compatibility. diff --git a/coding-agent/src/coding_agent/executors/__init__.py b/coding-agent/src/coding_agent/executors/__init__.py index f26b38e7f..d396e92fc 100644 --- a/coding-agent/src/coding_agent/executors/__init__.py +++ b/coding-agent/src/coding_agent/executors/__init__.py @@ -5,6 +5,8 @@ LocalDaemonRuntimePreparation, LocalDaemonRuntimeProvider, LocalDaemonRuntimeResult, + LocalDaemonRuntimeSession, + LocalDaemonSessionRuntimeProvider, RuntimeTurnCompletedHook, RuntimeTurnFailedHook, RuntimeTurnAdapter, @@ -18,6 +20,8 @@ "LocalDaemonRuntimePreparation", "LocalDaemonRuntimeProvider", "LocalDaemonRuntimeResult", + "LocalDaemonRuntimeSession", + "LocalDaemonSessionRuntimeProvider", "RuntimeTurnCompletedHook", "RuntimeTurnFailedHook", "RuntimeTurnAdapter", diff --git a/coding-agent/src/coding_agent/executors/local_daemon.py b/coding-agent/src/coding_agent/executors/local_daemon.py index 52d3545cd..cd69fc095 100644 --- a/coding-agent/src/coding_agent/executors/local_daemon.py +++ b/coding-agent/src/coding_agent/executors/local_daemon.py @@ -3,12 +3,15 @@ from collections.abc import Awaitable, Callable from dataclasses import dataclass from inspect import isawaitable -from typing import Protocol +from pathlib import Path +from typing import Protocol, cast +from coding_agent.approval import ApprovalPolicy from coding_agent.runs import ( LocalDaemonExecutorRef, LocalPathWorkspaceRef, RunRequest, + RunTarget, RunSubmission, ) @@ -29,8 +32,34 @@ class LocalDaemonRuntimeBinding: class LocalDaemonRuntimeProvider(Protocol): - async def prepare_runtime(self, request: RunRequest) -> LocalDaemonRuntimeBinding: - ... + async def prepare_runtime( + self, request: RunRequest + ) -> LocalDaemonRuntimeBinding: ... + + +class LocalDaemonRuntimeSession(Protocol): + id: str + model_name: str | None + provider_name: str | None + base_url: str | None + max_steps: int | None + approval_policy: ApprovalPolicy + provider: object | None + tape_id: str | None + runtime_message_bus: object + runtime_pipeline: object | None + runtime_ctx: object | None + runtime_adapter: RuntimeTurnAdapter | None + + +RuntimeEnvironmentResolver = Callable[[RunTarget], object] +RuntimeWorkspaceRootResolver = Callable[[object], Path | None] +RuntimeContextWorkspaceRootResolver = Callable[[object], Path | None] +RuntimeCloseHook = Callable[[LocalDaemonRuntimeSession], Awaitable[None]] +RuntimePersistHook = Callable[[LocalDaemonRuntimeSession], Awaitable[None]] +RuntimeTapeLoader = Callable[[str | None], Awaitable[object]] +RuntimeFactory = Callable[..., tuple[object, object]] +RuntimeAdapterFactory = Callable[[object, object], RuntimeTurnAdapter] RuntimePreparedHook = Callable[ @@ -69,6 +98,76 @@ class LocalDaemonRuntimeResult: outcome: object +@dataclass(frozen=True) +class LocalDaemonSessionRuntimeProvider: + session: LocalDaemonRuntimeSession + resolve_environment: RuntimeEnvironmentResolver + workspace_root_for_environment: RuntimeWorkspaceRootResolver + workspace_root_for_runtime: RuntimeContextWorkspaceRootResolver + close_runtime: RuntimeCloseHook + create_agent_for_session: RuntimeFactory + restore_tape: RuntimeTapeLoader + persist_session: RuntimePersistHook + adapter_factory: RuntimeAdapterFactory + + async def prepare_runtime(self, request: RunRequest) -> LocalDaemonRuntimeBinding: + session = self.session + pipeline = session.runtime_pipeline + ctx = session.runtime_ctx + adapter = session.runtime_adapter + environment = self.resolve_environment(request.target) + workspace_root = self.workspace_root_for_environment(environment) + if pipeline is not None and ctx is not None and adapter is not None: + cached_workspace_root = self.workspace_root_for_runtime(ctx) + if ( + cached_workspace_root is not None + and workspace_root is not None + and cached_workspace_root != workspace_root + ): + await self.close_runtime(session) + pipeline = None + ctx = None + adapter = None + if pipeline is None or ctx is None or adapter is None: + approval_mode_map = { + ApprovalPolicy.YOLO: "yolo", + ApprovalPolicy.INTERACTIVE: "interactive", + ApprovalPolicy.AUTO: "auto", + } + pipeline, ctx = self.create_agent_for_session( + workspace_root=workspace_root, + environment=environment, + model_override=session.model_name, + provider_override=session.provider_name, + base_url_override=session.base_url, + max_steps_override=session.max_steps, + approval_mode_override=approval_mode_map[session.approval_policy], + session_id_override=session.id, + run_id_override=request.run_id, + api_key=None, + tape=await self.restore_tape(session.tape_id), + ) + session.tape_id = ctx.tape.tape_id + await self.persist_session(session) + ctx.runtime_message_bus = session.runtime_message_bus + ctx.config["wire_consumer"] = None + ctx.config["agent_id"] = "" + + llm_plugin = pipeline._registry.get("llm_provider") + if session.provider is not None: + llm_plugin._instance = session.provider + + adapter = self.adapter_factory(pipeline, ctx) + session.runtime_pipeline = pipeline + session.runtime_ctx = ctx + session.runtime_adapter = adapter + return LocalDaemonRuntimeBinding( + pipeline=pipeline, + ctx=ctx, + adapter=cast(RuntimeTurnAdapter, adapter), + ) + + @dataclass(frozen=True) class LocalDaemonExecutor: """Local daemon run executor boundary. diff --git a/coding-agent/src/coding_agent/server/session_manager.py b/coding-agent/src/coding_agent/server/session_manager.py index 39b974047..559d38730 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 @@ LocalDaemonRuntimeBinding, LocalDaemonRuntimeExecution, LocalDaemonRuntimePreparation, + LocalDaemonSessionRuntimeProvider, ) from coding_agent.events import DisplayEvent, RuntimeEventReplayService from coding_agent.runs import ( @@ -3457,60 +3458,21 @@ async def _prepare_local_daemon_runtime( consumer: _WireConsumer, request: RunRequest, ) -> LocalDaemonRuntimeBinding: - pipeline = session.runtime_pipeline - ctx = session.runtime_ctx - adapter = session.runtime_adapter - environment = self._resolve_local_daemon_environment(request.target) - workspace_root = self._environment_workspace_root(environment) - if pipeline is not None and ctx is not None and adapter is not None: - cached_workspace_root = self._runtime_environment_workspace_root(ctx) - if ( - cached_workspace_root is not None - and workspace_root is not None - and cached_workspace_root != workspace_root - ): - await self._close_runtime(session) - pipeline = None - ctx = None - adapter = None - if pipeline is None or ctx is None or adapter is None: - approval_mode_map = { - ApprovalPolicy.YOLO: "yolo", - ApprovalPolicy.INTERACTIVE: "interactive", - ApprovalPolicy.AUTO: "auto", - } - pipeline, ctx = self._create_agent_for_session( - workspace_root=workspace_root, - environment=environment, - model_override=session.model_name, - provider_override=session.provider_name, - base_url_override=session.base_url, - max_steps_override=session.max_steps, - approval_mode_override=approval_mode_map[session.approval_policy], - session_id_override=session.id, - run_id_override=request.run_id, - api_key=None, - tape=await self._restore_tape(session.tape_id), - ) - session.tape_id = ctx.tape.tape_id - await self._persist_session_async(session) - ctx.runtime_message_bus = session.runtime_message_bus - ctx.config["wire_consumer"] = None - ctx.config["agent_id"] = "" - - llm_plugin = pipeline._registry.get("llm_provider") - if session.provider is not None: - llm_plugin._instance = session.provider - - adapter = PipelineAdapter(pipeline=pipeline, ctx=ctx, consumer=consumer) - session.runtime_pipeline = pipeline - session.runtime_ctx = ctx - session.runtime_adapter = adapter - return LocalDaemonRuntimeBinding( - pipeline=pipeline, - ctx=ctx, - adapter=adapter, - ) + return await LocalDaemonSessionRuntimeProvider( + session=session, + resolve_environment=self._resolve_local_daemon_environment, + workspace_root_for_environment=self._environment_workspace_root, + workspace_root_for_runtime=self._runtime_environment_workspace_root, + close_runtime=self._close_runtime, + create_agent_for_session=self._create_agent_for_session, + restore_tape=self._restore_tape, + persist_session=self._persist_session_async, + adapter_factory=lambda pipeline, ctx: PipelineAdapter( + pipeline=pipeline, + ctx=ctx, + consumer=consumer, + ), + ).prepare_runtime(request) def _invalidate_runtime(self, session: Session) -> None: session.runtime_pipeline = None diff --git a/coding-agent/tests/coding_agent/test_local_daemon_executor.py b/coding-agent/tests/coding_agent/test_local_daemon_executor.py index bc97346fb..15d6d0a53 100644 --- a/coding-agent/tests/coding_agent/test_local_daemon_executor.py +++ b/coding-agent/tests/coding_agent/test_local_daemon_executor.py @@ -1,14 +1,18 @@ from __future__ import annotations import asyncio +import types +from pathlib import Path import pytest +from coding_agent.approval import ApprovalPolicy from coding_agent.executors import ( LocalDaemonExecutor, LocalDaemonRuntimeBinding, LocalDaemonRuntimeExecution, LocalDaemonRuntimePreparation, + LocalDaemonSessionRuntimeProvider, RunExecutorTargetError, ) from coding_agent.runs import ( @@ -23,11 +27,15 @@ def _local_request() -> RunRequest: + return _local_request_for_path("/repo") + + +def _local_request_for_path(path: str) -> RunRequest: return RunRequest( session_id="session-1", run_id="run-1", target=RunTarget( - workspace=LocalPathWorkspaceRef(path="/repo"), + workspace=LocalPathWorkspaceRef(path=path), executor=LocalDaemonExecutorRef(), isolation=IsolationPolicy(kind="default_local_sandbox"), ), @@ -65,6 +73,151 @@ async def run_turn(self, prompt: str) -> object: raise self.exc +class FakeRuntimeSession: + def __init__(self) -> None: + self.id = "session-1" + self.model_name = "model-1" + self.provider_name = "provider-1" + self.base_url = "https://example.test" + self.max_steps = 7 + self.approval_policy = ApprovalPolicy.AUTO + self.provider = None + self.tape_id: str | None = "tape-old" + self.runtime_message_bus = object() + self.runtime_pipeline: object | None = None + self.runtime_ctx: object | None = None + self.runtime_adapter: FakeRuntimeAdapter | None = None + + +@pytest.mark.asyncio +async def test_local_daemon_session_runtime_provider_reuses_cached_runtime( + tmp_path: Path, +) -> None: + workspace = tmp_path / "repo" + workspace.mkdir() + session = FakeRuntimeSession() + cached_pipeline = object() + cached_ctx = types.SimpleNamespace( + config={}, tape=types.SimpleNamespace(tape_id="tape-old") + ) + cached_adapter = FakeRuntimeAdapter() + session.runtime_pipeline = cached_pipeline + session.runtime_ctx = cached_ctx + session.runtime_adapter = cached_adapter + + async def close_runtime(runtime_session) -> None: + del runtime_session + raise AssertionError("cached runtime should be reused") + + async def restore_tape(tape_id: str | None) -> object: + del tape_id + raise AssertionError("cached runtime should not restore tape") + + async def persist_session(runtime_session) -> None: + del runtime_session + raise AssertionError("cached runtime should not persist") + + def create_agent_for_session(**kwargs): + raise AssertionError(f"cached runtime should not create agent: {kwargs!r}") + + binding = await LocalDaemonSessionRuntimeProvider( + session=session, + resolve_environment=lambda target: target, + workspace_root_for_environment=lambda environment: workspace.resolve(), + workspace_root_for_runtime=lambda ctx: workspace.resolve(), + close_runtime=close_runtime, + create_agent_for_session=create_agent_for_session, + restore_tape=restore_tape, + persist_session=persist_session, + adapter_factory=lambda pipeline, ctx: FakeRuntimeAdapter(), + ).prepare_runtime(_local_request_for_path(str(workspace))) + + assert binding.pipeline is cached_pipeline + assert binding.ctx is cached_ctx + assert binding.adapter is cached_adapter + + +@pytest.mark.asyncio +async def test_local_daemon_session_runtime_provider_rebuilds_changed_workspace( + tmp_path: Path, +) -> None: + old_workspace = tmp_path / "old" + old_workspace.mkdir() + new_workspace = tmp_path / "new" + new_workspace.mkdir() + session = FakeRuntimeSession() + old_adapter = FakeRuntimeAdapter() + session.runtime_pipeline = object() + session.runtime_ctx = types.SimpleNamespace( + config={}, tape=types.SimpleNamespace(tape_id="tape-old") + ) + session.runtime_adapter = old_adapter + closed: list[str] = [] + persisted: list[str | None] = [] + restored: list[str | None] = [] + create_kwargs: dict[str, object] = {} + adapter = FakeRuntimeAdapter() + fake_pipeline = types.SimpleNamespace( + _registry=types.SimpleNamespace( + get=lambda name: types.SimpleNamespace(name=name, _instance=None) + ) + ) + fake_ctx = types.SimpleNamespace( + config={}, tape=types.SimpleNamespace(tape_id="tape-new") + ) + + async def close_runtime(runtime_session) -> None: + closed.append(runtime_session.id) + runtime_session.runtime_pipeline = None + runtime_session.runtime_ctx = None + runtime_session.runtime_adapter = None + + async def restore_tape(tape_id: str | None) -> object: + restored.append(tape_id) + return types.SimpleNamespace(tape_id=tape_id) + + async def persist_session(runtime_session) -> None: + persisted.append(runtime_session.tape_id) + + def create_agent_for_session(**kwargs): + create_kwargs.update(kwargs) + return fake_pipeline, fake_ctx + + binding = await LocalDaemonSessionRuntimeProvider( + session=session, + resolve_environment=lambda target: target, + workspace_root_for_environment=lambda environment: new_workspace.resolve(), + workspace_root_for_runtime=lambda ctx: old_workspace.resolve(), + close_runtime=close_runtime, + create_agent_for_session=create_agent_for_session, + restore_tape=restore_tape, + persist_session=persist_session, + adapter_factory=lambda pipeline, ctx: adapter, + ).prepare_runtime(_local_request_for_path(str(new_workspace))) + + assert closed == ["session-1"] + assert restored == ["tape-old"] + assert persisted == ["tape-new"] + assert create_kwargs["workspace_root"] == new_workspace.resolve() + assert create_kwargs["model_override"] == "model-1" + assert create_kwargs["provider_override"] == "provider-1" + assert create_kwargs["base_url_override"] == "https://example.test" + assert create_kwargs["max_steps_override"] == 7 + assert create_kwargs["approval_mode_override"] == "auto" + assert create_kwargs["session_id_override"] == "session-1" + assert create_kwargs["run_id_override"] == "run-1" + assert create_kwargs["api_key"] is None + assert session.tape_id == "tape-new" + assert fake_ctx.runtime_message_bus is session.runtime_message_bus + assert fake_ctx.config == {"wire_consumer": None, "agent_id": ""} + assert session.runtime_pipeline is fake_pipeline + assert session.runtime_ctx is fake_ctx + assert session.runtime_adapter is adapter + assert binding.pipeline is fake_pipeline + assert binding.ctx is fake_ctx + assert binding.adapter is adapter + + @pytest.mark.asyncio async def test_local_daemon_executor_accepts_local_daemon_target() -> None: request = _local_request() @@ -185,9 +338,7 @@ async def after_turn( ) assert result.outcome == {"status": "completed"} - assert events == [ - ("after_turn", ["implement the task"], {"status": "completed"}) - ] + assert events == [("after_turn", ["implement the task"], {"status": "completed"})] assert adapter.prompts == ["implement the task"] @@ -353,7 +504,9 @@ async def test_local_daemon_executor_rejects_runtime_for_non_local_target() -> N @pytest.mark.asyncio -async def test_local_daemon_executor_validates_runtime_before_overridden_prepare() -> None: +async def test_local_daemon_executor_validates_runtime_before_overridden_prepare() -> ( + None +): request = RunRequest( session_id="session-1", run_id="run-1", @@ -406,7 +559,9 @@ async def prepare_runtime( @pytest.mark.asyncio -async def test_local_daemon_executor_rejects_runtime_preparation_for_non_local_target() -> None: +async def test_local_daemon_executor_rejects_runtime_preparation_for_non_local_target() -> ( + None +): request = RunRequest( session_id="session-1", run_id="run-1", @@ -450,7 +605,9 @@ async def test_local_daemon_executor_rejects_runtime_preparation_for_non_local_t @pytest.mark.asyncio -async def test_local_daemon_executor_rejects_runtime_preparation_for_non_local_workspace() -> None: +async def test_local_daemon_executor_rejects_runtime_preparation_for_non_local_workspace() -> ( + None +): request = RunRequest( session_id="session-1", run_id="run-1",