From 5096df905700333abe132e0b9570552a42168d9a Mon Sep 17 00:00:00 2001 From: kina0630 Date: Tue, 2 Jun 2026 22:39:09 +0800 Subject: [PATCH 1/2] feat: execute local runtime through daemon executor --- ...daemon-pr6-runtime-invocation-ownership.md | 60 ++++++++++++++ .../src/coding_agent/executors/__init__.py | 7 +- .../coding_agent/executors/local_daemon.py | 31 ++++++-- .../coding_agent/server/session_manager.py | 30 +++++-- .../test_local_daemon_executor.py | 55 ++++++++++++- .../tests/ui/test_session_manager_runtime.py | 79 +++++++++++++++++++ 6 files changed, 248 insertions(+), 14 deletions(-) create mode 100644 coding-agent/.opencode/prompts/tasks/local-daemon-pr6-runtime-invocation-ownership.md diff --git a/coding-agent/.opencode/prompts/tasks/local-daemon-pr6-runtime-invocation-ownership.md b/coding-agent/.opencode/prompts/tasks/local-daemon-pr6-runtime-invocation-ownership.md new file mode 100644 index 000000000..b9be5da9d --- /dev/null +++ b/coding-agent/.opencode/prompts/tasks/local-daemon-pr6-runtime-invocation-ownership.md @@ -0,0 +1,60 @@ +Goal: +Move the local runtime invocation call behind `LocalDaemonExecutor` as the next +ADR-0058 executor-ownership slice. + +Scope: +- Add `LocalDaemonRuntimeExecution` to describe a local daemon runtime execution + request plus the runtime callable. +- Add `LocalDaemonExecutor.execute_runtime(...)` and validate the same + `LocalDaemonExecutorRef + LocalPathWorkspaceRef` target contract used by + `submit_run(...)`. +- Let `SessionManager` accept an injectable `LocalDaemonExecutor`. +- Configure the default `RunCoordinator` with the same local executor instance + used for runtime invocation. +- Route local `SessionManager.run_agent()` runtime calls through + `LocalDaemonExecutor.execute_runtime(...)`. +- Preserve existing run lifecycle, runtime store, HTTP, CLI, approval, and + checkpoint behavior. + +Out of scope: +- Do not move the full pipeline/context/adapter preparation out of + `SessionManager` in this slice. +- Do not implement daemon IPC, background worker lifecycle, or reconnect. +- Do not change persisted session payloads or runtime store schemas. +- Do not change public HTTP or CLI APIs. +- Do not implement managed pool, local attached, or cloud executor execution. +- Do not demote `coding_agent run` in this slice. + +Context: +- ADRs: + - `docs/adr/0058-local-daemon-control-plane-executor-architecture.md` + - `docs/adr/0054-executor-runtime-terminology-and-resume-first-direction.md` +- Relevant files: + - `src/coding_agent/executors/local_daemon.py` + - `src/coding_agent/server/session_manager.py` + - `tests/coding_agent/test_local_daemon_executor.py` + - `tests/ui/test_session_manager_runtime.py` + +Target tests: +- `uv run pytest tests/coding_agent/test_local_daemon_executor.py tests/coding_agent/test_run_coordinator.py tests/coding_agent/test_run_target.py -v` +- `uv run pytest tests/ui/test_session_manager_runtime.py -v` +- `uv run ruff check src/coding_agent/executors src/coding_agent/runs src/coding_agent/server/session_manager.py tests/coding_agent/test_local_daemon_executor.py tests/coding_agent/test_run_coordinator.py tests/coding_agent/test_run_target.py tests/ui/test_session_manager_runtime.py` +- `uv run basedpyright src/coding_agent/executors src/coding_agent/runs tests/coding_agent/test_local_daemon_executor.py tests/coding_agent/test_run_coordinator.py tests/coding_agent/test_run_target.py` +- `git diff --check` + +Known type-check caveat: +- `session_manager.py` and `test_session_manager_runtime.py` still have + pre-existing strict typing debt. This slice type-checks the executor and run + surfaces directly. + +Loop policy: +- Engineer implements the smallest runtime invocation ownership shift and runs + the target tests. +- Reviewer reports only P1/P2 correctness, safety, or scope issues. +- Engineer fixes only accepted P1/P2 findings and reruns the same target tests. +- Stop after one review/fix/retest cycle unless a human expands the scope. + +Stop conditions: +- Stop if full runtime preparation migration is required. +- Stop if daemon lifecycle or IPC design is required. +- Stop if persisted schemas or public APIs must change. diff --git a/coding-agent/src/coding_agent/executors/__init__.py b/coding-agent/src/coding_agent/executors/__init__.py index 84616fb3c..202089548 100644 --- a/coding-agent/src/coding_agent/executors/__init__.py +++ b/coding-agent/src/coding_agent/executors/__init__.py @@ -1,6 +1,11 @@ -from .local_daemon import LocalDaemonExecutor, RunExecutorTargetError +from .local_daemon import ( + LocalDaemonExecutor, + LocalDaemonRuntimeExecution, + RunExecutorTargetError, +) __all__ = [ "LocalDaemonExecutor", + "LocalDaemonRuntimeExecution", "RunExecutorTargetError", ] diff --git a/coding-agent/src/coding_agent/executors/local_daemon.py b/coding-agent/src/coding_agent/executors/local_daemon.py index a1b89256f..2ba629ae4 100644 --- a/coding-agent/src/coding_agent/executors/local_daemon.py +++ b/coding-agent/src/coding_agent/executors/local_daemon.py @@ -1,5 +1,6 @@ from __future__ import annotations +from collections.abc import Awaitable, Callable from dataclasses import dataclass from coding_agent.runs import ( @@ -10,10 +11,19 @@ ) +RuntimeTurnCallable = Callable[[], Awaitable[object]] + + class RunExecutorTargetError(ValueError): """Raised when a run executor receives an incompatible target.""" +@dataclass(frozen=True) +class LocalDaemonRuntimeExecution: + request: RunRequest + run: RuntimeTurnCallable + + @dataclass(frozen=True) class LocalDaemonExecutor: """Local daemon run executor boundary. @@ -23,6 +33,20 @@ class LocalDaemonExecutor: """ async def submit_run(self, request: RunRequest) -> RunSubmission: + self._validate_request_target(request) + return RunSubmission( + session_id=request.session_id, + run_id=request.run_id, + target=request.target, + executor=request.target.executor, + metadata=request.metadata, + ) + + async def execute_runtime(self, execution: LocalDaemonRuntimeExecution) -> object: + self._validate_request_target(execution.request) + return await execution.run() + + def _validate_request_target(self, request: RunRequest) -> None: if not isinstance(request.target.executor, LocalDaemonExecutorRef): raise RunExecutorTargetError( "LocalDaemonExecutor requires a local_daemon executor target" @@ -31,10 +55,3 @@ async def submit_run(self, request: RunRequest) -> RunSubmission: raise RunExecutorTargetError( "LocalDaemonExecutor requires a local_path workspace target" ) - return RunSubmission( - session_id=request.session_id, - run_id=request.run_id, - target=request.target, - executor=request.target.executor, - metadata=request.metadata, - ) diff --git a/coding-agent/src/coding_agent/server/session_manager.py b/coding-agent/src/coding_agent/server/session_manager.py index a1566b737..1f70f7556 100644 --- a/coding-agent/src/coding_agent/server/session_manager.py +++ b/coding-agent/src/coding_agent/server/session_manager.py @@ -73,9 +73,10 @@ SQLiteRuntimeStore, JSONValue as RuntimeJSONValue, ) -from coding_agent.executors import LocalDaemonExecutor +from coding_agent.executors import LocalDaemonExecutor, LocalDaemonRuntimeExecution from coding_agent.runs import ( DefaultRunCoordinator, + LocalDaemonExecutorRef, RunCoordinator, RunRequest, run_target_from_execution_binding, @@ -1229,6 +1230,7 @@ def __init__( fencing_token: int | None = None, owner_lease_seconds: float = 30.0, run_coordinator: RunCoordinator | None = None, + local_daemon_executor: LocalDaemonExecutor | None = None, ): self._storage_config = storage_config or {} self._pg_pool = pg_pool @@ -1253,8 +1255,13 @@ def __init__( ) self._create_agent = create_agent_fn self._binding_resolver = binding_resolver or DefaultBindingResolver() + self._local_daemon_executor = ( + LocalDaemonExecutor() + if local_daemon_executor is None + else local_daemon_executor + ) self._run_coordinator = ( - DefaultRunCoordinator(local_daemon_executor=LocalDaemonExecutor()) + DefaultRunCoordinator(local_daemon_executor=self._local_daemon_executor) if run_coordinator is None else run_coordinator ) @@ -3368,7 +3375,7 @@ async def _submit_runtime_run_request( run_id: str, prompt: str, resume_context: SessionResumeContext | None = None, - ) -> None: + ) -> RunRequest: input_summary = prompt if prompt.strip() else None request = RunRequest( session_id=session.id, @@ -3380,6 +3387,7 @@ async def _submit_runtime_run_request( ), ) await self._run_coordinator.submit_run(request) + return request def _invalidate_runtime(self, session: Session) -> None: session.runtime_pipeline = None @@ -3878,7 +3886,7 @@ async def run_agent( started_at=started_at, resume_context=resume_context, ) - await self._submit_runtime_run_request( + run_request = await self._submit_runtime_run_request( session, run_id=run_id, prompt=prompt, @@ -3907,7 +3915,19 @@ async def run_agent( prompt=prompt, resume_context=resume_context, ) - outcome = await adapter.run_turn(prompt) + + async def run_runtime_turn() -> object: + return await adapter.run_turn(prompt) + + if isinstance(run_request.target.executor, LocalDaemonExecutorRef): + outcome = await self._local_daemon_executor.execute_runtime( + LocalDaemonRuntimeExecution( + request=run_request, + run=run_runtime_turn, + ) + ) + else: + outcome = await run_runtime_turn() if self._runtime_store is not None: turn_outcome = self._require_turn_outcome(outcome) turn_status = self._status_from_turn_outcome(turn_outcome) 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 e99100745..e714ba285 100644 --- a/coding-agent/tests/coding_agent/test_local_daemon_executor.py +++ b/coding-agent/tests/coding_agent/test_local_daemon_executor.py @@ -2,7 +2,11 @@ import pytest -from coding_agent.executors import LocalDaemonExecutor, RunExecutorTargetError +from coding_agent.executors import ( + LocalDaemonExecutor, + LocalDaemonRuntimeExecution, + RunExecutorTargetError, +) from coding_agent.runs import ( CloudWorkspaceRef, IsolationPolicy, @@ -40,6 +44,23 @@ async def test_local_daemon_executor_accepts_local_daemon_target() -> None: assert submission.status == "accepted" +@pytest.mark.asyncio +async def test_local_daemon_executor_executes_runtime_callable() -> None: + request = _local_request() + executed: list[str] = [] + + async def run_runtime() -> str: + executed.append(request.run_id) + return "completed" + + result = await LocalDaemonExecutor().execute_runtime( + LocalDaemonRuntimeExecution(request=request, run=run_runtime) + ) + + assert result == "completed" + assert executed == ["run-1"] + + @pytest.mark.asyncio async def test_local_daemon_executor_rejects_non_local_daemon_executor() -> None: request = RunRequest( @@ -87,3 +108,35 @@ async def test_local_daemon_executor_rejects_non_local_workspace() -> None: match="LocalDaemonExecutor requires a local_path workspace target", ): _ = await LocalDaemonExecutor().submit_run(request) + + +@pytest.mark.asyncio +async def test_local_daemon_executor_rejects_runtime_for_non_local_target() -> None: + request = RunRequest( + session_id="session-1", + run_id="run-1", + target=RunTarget( + workspace=CloudWorkspaceRef( + workspace_url="docker://workspace/ws-1", + workspace_id="ws-1", + ), + executor=ManagedPoolExecutorRef(), + isolation=IsolationPolicy( + kind="provider_sandbox", + network="provider_managed", + filesystem="provider_managed", + secrets="provider_managed", + ), + ), + ) + + async def run_runtime() -> str: + raise AssertionError("runtime should not execute") + + with pytest.raises( + RunExecutorTargetError, + match="LocalDaemonExecutor requires a local_daemon executor target", + ): + _ = await LocalDaemonExecutor().execute_runtime( + LocalDaemonRuntimeExecution(request=request, run=run_runtime) + ) diff --git a/coding-agent/tests/ui/test_session_manager_runtime.py b/coding-agent/tests/ui/test_session_manager_runtime.py index 4655a4933..f1dd26098 100644 --- a/coding-agent/tests/ui/test_session_manager_runtime.py +++ b/coding-agent/tests/ui/test_session_manager_runtime.py @@ -24,6 +24,7 @@ CloudEnvironment, LocalEnvironment, ) +from coding_agent.executors import LocalDaemonExecutor, LocalDaemonRuntimeExecution from coding_agent.runtime_store import ( AgentInteractionRecord, AgentRunRecord, @@ -321,6 +322,15 @@ async def submit_run(self, request: RunRequest) -> RunSubmission: ) +class RecordingLocalDaemonExecutor(LocalDaemonExecutor): + def __init__(self) -> None: + self.executions: list[LocalDaemonRuntimeExecution] = [] + + async def execute_runtime(self, execution: LocalDaemonRuntimeExecution) -> object: + self.executions.append(execution) + return await super().execute_runtime(execution) + + @pytest.mark.asyncio async def test_run_agent_does_not_hardcode_api_key() -> None: manager = SessionManager() @@ -637,6 +647,75 @@ def fake_create_agent(**kwargs): assert runtime_store.updated[-1]["status"] == "completed" +@pytest.mark.asyncio +async def test_run_agent_executes_local_runtime_through_local_daemon_executor( + tmp_path: Path, +) -> None: + runtime_store = FakeRuntimeStore() + local_executor = RecordingLocalDaemonExecutor() + workspace = tmp_path / "repo" + workspace.mkdir() + manager = SessionManager( + runtime_store=cast(Any, runtime_store), + local_daemon_executor=local_executor, + ) + session_id = await manager.create_session( + execution_binding=LocalExecutionBinding(workspace_root=str(workspace)), + ) + + class FakeAdapter: + def __init__(self, pipeline, ctx, consumer) -> None: + del pipeline, consumer + self.ctx = ctx + + async def run_turn(self, prompt: str) -> TurnOutcome: + del prompt + return TurnOutcome( + stop_reason=StopReason.NO_TOOL_CALLS, + steps_taken=1, + ) + + fake_pipeline = types.SimpleNamespace( + _registry=types.SimpleNamespace( + get=lambda _: types.SimpleNamespace(_instance=None) + ) + ) + + def fake_create_agent(**kwargs): + environment = kwargs["environment"] + if not isinstance(environment, LocalEnvironment): + raise TypeError("expected local environment") + return fake_pipeline, types.SimpleNamespace( + session_id=kwargs["session_id_override"], + config={}, + tape=kwargs.get("tape") or Tape(tape_id="stable-tape"), + run_context=AgentRunContext( + session_id=kwargs["session_id_override"], + run_id=kwargs["run_id_override"], + agent_id=None, + environment=environment, + ), + ) + + with pytest.MonkeyPatch.context() as mp: + mp.setattr("coding_agent.__main__.create_agent", fake_create_agent) + mp.setattr("coding_agent.server.session_manager.PipelineAdapter", FakeAdapter) + + await manager.run_agent(session_id, "implement runtime ownership") + + created_run = runtime_store.created[0] + assert len(local_executor.executions) == 1 + execution = local_executor.executions[0] + assert execution.request.session_id == session_id + assert execution.request.run_id == created_run.run_id + assert execution.request.input_summary == "implement runtime ownership" + assert isinstance(execution.request.target.executor, LocalDaemonExecutorRef) + assert isinstance(execution.request.target.workspace, LocalPathWorkspaceRef) + assert execution.request.target.workspace.path == str(workspace.resolve()) + assert runtime_store.updated[0]["status"] == "running" + assert runtime_store.updated[-1]["status"] == "completed" + + @pytest.mark.asyncio async def test_run_agent_marks_agent_run_failed_when_turn_outcome_errors() -> None: runtime_store = FakeRuntimeStore() From 6c97821a37be7bccb9f384da941442f80f374c99 Mon Sep 17 00:00:00 2001 From: kina0630 Date: Tue, 2 Jun 2026 22:49:34 +0800 Subject: [PATCH 2/2] test: cover non-local daemon executor routing --- .../tests/ui/test_session_manager_runtime.py | 65 +++++++++++++++++++ 1 file changed, 65 insertions(+) diff --git a/coding-agent/tests/ui/test_session_manager_runtime.py b/coding-agent/tests/ui/test_session_manager_runtime.py index f1dd26098..c677265d4 100644 --- a/coding-agent/tests/ui/test_session_manager_runtime.py +++ b/coding-agent/tests/ui/test_session_manager_runtime.py @@ -3336,6 +3336,71 @@ async def run_turn(self, prompt: str) -> None: assert runtime_store.created[0].metadata["execution_plane"] == "control_plane" +@pytest.mark.asyncio +async def test_run_agent_does_not_route_cloud_runtime_through_local_daemon_executor() -> None: + local_executor = RecordingLocalDaemonExecutor() + runtime_store = FakeRuntimeStore() + fake_pipeline = types.SimpleNamespace( + _registry=types.SimpleNamespace( + get=lambda _: types.SimpleNamespace(_instance=None) + ) + ) + + def fake_create_agent(**kwargs): + environment = kwargs["environment"] + if not isinstance(environment, CloudEnvironment): + raise TypeError("expected cloud environment") + return fake_pipeline, types.SimpleNamespace( + session_id=kwargs["session_id_override"], + config={}, + tape=kwargs.get("tape") or Tape(tape_id="stable-tape"), + run_context=AgentRunContext( + session_id=kwargs["session_id_override"], + run_id=kwargs["run_id_override"], + agent_id=None, + environment=environment, + ), + ) + + manager = SessionManager( + store=InMemorySessionStore(), + runtime_store=cast(Any, runtime_store), + create_agent_fn=fake_create_agent, + local_daemon_executor=local_executor, + binding_resolver=DefaultBindingResolver( + cloud_client_factory=lambda binding: FakeCloudClient() + ), + ) + session_id = await manager.create_session( + execution_binding=CloudWorkspaceBinding( + workspace_url="https://workspace.example.com", + workspace_id="ws-123", + ) + ) + + class FakeAdapter: + def __init__(self, pipeline, ctx, consumer) -> None: + del pipeline, consumer + self.ctx = ctx + + async def run_turn(self, prompt: str) -> TurnOutcome: + del prompt + return TurnOutcome( + stop_reason=StopReason.NO_TOOL_CALLS, + steps_taken=1, + ) + + with pytest.MonkeyPatch.context() as mp: + mp.setattr("coding_agent.server.session_manager.PipelineAdapter", FakeAdapter) + + await manager.run_agent(session_id, "hello cloud") + + assert local_executor.executions == [] + assert runtime_store.created[0].metadata["workspace_surface"] == "cloud_workspace" + assert runtime_store.updated[0]["status"] == "running" + assert runtime_store.updated[-1]["status"] == "completed" + + @pytest.mark.asyncio async def test_restore_checkpoint_preserves_cloud_execution_binding() -> None: captured_kwargs: dict[str, object] = {}