feat: execute local runtime through daemon executor

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.

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",
 ]

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,
-        )

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)

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)
+        )