feat: extract local runtime provider

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.

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.

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

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.

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