feat: extract runtime turn error handler

coding-agent/.opencode/prompts/tasks/local-daemon-pr35-turn-error-handler.md

@@ -0,0 +1,48 @@
+Goal:
+Extract local daemon turn error actions from `SessionManager.run_agent` into a
+runtime lifecycle helper so error handling is owned by the runtime turn boundary,
+not by the session manager loop body.
+
+Scope:
+- Add a `RuntimeTurnErrorHandler` helper under `coding_agent.runs.lifecycle`.
+- Move fatal, cancelled, and generic turn error actions out of
+  `SessionManager.run_agent`.
+- Preserve existing behavior for run finish status, observation state, runtime
+  close, wire error notification, and outer exception propagation.
+- Update ADR-0058 follow-up notes.
+- Add focused unit coverage for the new lifecycle helper.
+
+Out of scope:
+- Changing runtime store payloads or status names.
+- Moving checkpoint restore, display projection, or final cleanup policy.
+- Changing HTTP/CLI API behavior.
+- Changing LocalDaemonExecutor execution semantics.
+
+Context:
+- ADRs:
+  - docs/adr/0058-local-daemon-control-plane-executor-architecture.md
+- Relevant files:
+  - src/coding_agent/runs/lifecycle.py
+  - src/coding_agent/runs/__init__.py
+  - src/coding_agent/server/session_manager.py
+  - tests/coding_agent/test_run_lifecycle.py
+  - tests/ui/test_session_manager_runtime.py
+
+Target tests:
+- uv run pytest tests/coding_agent/test_run_lifecycle.py -v
+- uv run pytest tests/ui/test_session_manager_runtime.py -k "run_agent_marks_agent_run_failed_when_turn_outcome_errors or run_agent_records_error_outcome_without_runtime_store or run_agent_reraises_fatal_tool_execution_error_without_sending_error_turn or run_agent_persists_agent_run_lifecycle_when_store_configured" -v
+- uv run ruff check src/coding_agent/runs src/coding_agent/server/session_manager.py tests/coding_agent/test_run_lifecycle.py tests/ui/test_session_manager_runtime.py
+- uv run ruff format --check src/coding_agent/runs/lifecycle.py src/coding_agent/runs/__init__.py src/coding_agent/server/session_manager.py tests/coding_agent/test_run_lifecycle.py
+- git diff --check
+
+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:
+- At most one review/fix/retest cycle.
+- Escalate architectural redirection or scope expansion to the human.
+- Ignore non-blocking optimization suggestions.

coding-agent/docs/adr/0058-local-daemon-control-plane-executor-architecture.md

@@ -167,6 +167,9 @@ This ADR does not implement that path.
   - Completed: local daemon before-turn runtime wiring is delegated to
     `RuntimeTurnStarter` instead of living inside `run_agent`'s before-turn
     closure.
+  - Completed: local daemon fatal/cancelled/generic turn error actions are
+    delegated to `RuntimeTurnErrorHandler` instead of living inside
+    `run_agent`.
   - Remaining: `SessionManager` still owns run lifecycle bookkeeping,
     checkpoint restore preparation details, observation callbacks, wire consumer
     setup, and some runtime close/error policy. These should move behind

coding-agent/src/coding_agent/runs/__init__.py

@@ -24,6 +24,8 @@
     RuntimeSubagentMessagePublisherBinder,
     RuntimeTurnBinding,
     RuntimeTurnErrorAction,
+    RuntimeTurnErrorHandler,
+    RuntimeTurnErrorSession,
     RuntimeTurnErrorState,
     RuntimeTurnFinalizer,
     RuntimeTurnRunTracker,
@@ -97,6 +99,8 @@
     "RuntimeSubagentMessagePublisherBinder",
     "RuntimeTurnBinding",
     "RuntimeTurnErrorAction",
+    "RuntimeTurnErrorHandler",
+    "RuntimeTurnErrorSession",
     "RuntimeTurnErrorState",
     "RuntimeTurnFinalizer",
     "RuntimeTurnRunTracker",

coding-agent/src/coding_agent/runs/lifecycle.py

@@ -27,6 +27,10 @@ class RuntimeTurnSession(Protocol):
     last_failure_details: str | None
 
 
+class RuntimeTurnErrorSession(RuntimeTurnSession, RuntimeRunSession, Protocol):
+    pass
+
+
 class RuntimeTurnStartSession(RuntimeRunSession, Protocol):
     runtime_message_bus: object
 
@@ -71,6 +75,18 @@ class RuntimeTurnErrorAction(Protocol):
     async def __call__(self) -> None: ...
 
 
+class RuntimeSessionCloser(Protocol):
+    async def __call__(self, session: RuntimeTurnErrorSession) -> None: ...
+
+
+class RuntimeGenericErrorNotifier(Protocol):
+    async def __call__(
+        self,
+        session: RuntimeTurnErrorSession,
+        exc: Exception,
+    ) -> None: ...
+
+
 class RuntimeRootRunIdentityBinder(Protocol):
     def __call__(
         self,
@@ -308,6 +324,64 @@ async def handle(self, action: RuntimeTurnErrorAction) -> None:
         self.handled = True
 
 
+@dataclass(frozen=True)
+class RuntimeTurnErrorHandler:
+    turn_run: RuntimeTurnRunTracker
+    close_runtime: RuntimeSessionCloser
+    notify_generic_error: RuntimeGenericErrorNotifier
+    fail_observation: Callable[[str], None] | None = None
+    cancel_observation: Callable[[], None] | None = None
+
+    async def handle_fatal(
+        self,
+        session: RuntimeTurnErrorSession,
+        exc: BaseException,
+    ) -> None:
+        if self.fail_observation is not None:
+            self.fail_observation(type(exc).__name__)
+        await self.turn_run.finish_if_started(
+            session,
+            status="failed",
+            result={},
+            error=str(exc),
+        )
+        session.turn_status = "failed"
+        session.last_failure_details = f"Fatal tool execution failed: {exc}"
+        await self.close_runtime(session)
+
+    async def handle_cancelled(self, session: RuntimeTurnErrorSession) -> None:
+        if self.cancel_observation is not None:
+            self.cancel_observation()
+        await self.turn_run.finish_if_started(
+            session,
+            status="cancelled",
+            result={},
+            error="cancelled",
+        )
+
+    async def handle_generic(
+        self,
+        session: RuntimeTurnErrorSession,
+        exc: Exception,
+        *,
+        ensure_started: bool = False,
+    ) -> None:
+        if ensure_started:
+            await self.turn_run.ensure_started(session)
+        if self.fail_observation is not None:
+            self.fail_observation(type(exc).__name__)
+        await self.turn_run.finish_if_started(
+            session,
+            status="failed",
+            result={},
+            error=str(exc),
+        )
+        session.turn_status = "failed"
+        session.last_failure_details = f"HTTP session turn failed: {exc}"
+        await self.close_runtime(session)
+        await self.notify_generic_error(session, exc)
+
+
 @dataclass(frozen=True)
 class RuntimeTurnStarter:
     turn_run: RuntimeTurnRunTracker
@@ -415,6 +489,8 @@ async def complete(
     "RuntimeSubagentMessagePublisherBinder",
     "RuntimeTurnBinding",
     "RuntimeTurnErrorAction",
+    "RuntimeTurnErrorHandler",
+    "RuntimeTurnErrorSession",
     "RuntimeTurnErrorState",
     "RuntimeTurnFinalizer",
     "RuntimeTurnRunTracker",

coding-agent/src/coding_agent/server/session_manager.py

@@ -92,6 +92,7 @@
     RuntimeObservationCompleter,
     RuntimeRunLifecycle,
     RuntimeTurnErrorState,
+    RuntimeTurnErrorHandler,
     RuntimeTurnFinalizer,
     RuntimeTurnRunTracker,
     RuntimeTurnStarter,
@@ -3917,45 +3918,18 @@ async def run_agent(
             observation_recorder: AgentObservationRecorder | None = None
             turn_error_state = RuntimeTurnErrorState()
 
-            async def _handle_fatal_local_daemon_turn_error(
-                exc: FatalToolExecutionError,
-            ) -> None:
+            def _fail_local_daemon_turn_observation(error_type: str) -> None:
                 if observation_recorder is not None:
-                    observation_recorder.fail_turn(error_type=type(exc).__name__)
-                await turn_run.finish_if_started(
-                    session,
-                    status="failed",
-                    result={},
-                    error=str(exc),
-                )
-                session.turn_status = "failed"
-                session.last_failure_details = f"Fatal tool execution failed: {exc}"
-                await self._close_runtime(session)
+                    observation_recorder.fail_turn(error_type=error_type)
 
-            async def _handle_cancelled_local_daemon_turn_error() -> None:
+            def _cancel_local_daemon_turn_observation() -> None:
                 if observation_recorder is not None:
                     observation_recorder.cancel_turn()
-                await turn_run.finish_if_started(
-                    session,
-                    status="cancelled",
-                    result={},
-                    error="cancelled",
-                )
 
-            async def _handle_generic_local_daemon_turn_error(
+            async def _notify_generic_local_daemon_turn_error(
+                session: Session,
                 exc: Exception,
             ) -> None:
-                if observation_recorder is not None:
-                    observation_recorder.fail_turn(error_type=type(exc).__name__)
-                await turn_run.finish_if_started(
-                    session,
-                    status="failed",
-                    result={},
-                    error=str(exc),
-                )
-                session.turn_status = "failed"
-                session.last_failure_details = f"HTTP session turn failed: {exc}"
-                await self._close_runtime(session)
                 logger.exception("HTTP session turn failed")
                 await self._send_session_wire_message(
                     session,
@@ -3975,24 +3949,32 @@ async def _handle_generic_local_daemon_turn_error(
                     ),
                 )
 
+            turn_error_handler = RuntimeTurnErrorHandler(
+                turn_run=turn_run,
+                close_runtime=self._close_runtime,
+                notify_generic_error=_notify_generic_local_daemon_turn_error,
+                fail_observation=_fail_local_daemon_turn_observation,
+                cancel_observation=_cancel_local_daemon_turn_observation,
+            )
+
             async def _on_local_daemon_turn_error(
                 binding: LocalDaemonRuntimeBinding,
                 exc: BaseException,
             ) -> None:
                 del binding
                 if isinstance(exc, FatalToolExecutionError):
                     await turn_error_state.handle(
-                        lambda: _handle_fatal_local_daemon_turn_error(exc)
+                        lambda: turn_error_handler.handle_fatal(session, exc)
                     )
                     return
                 if isinstance(exc, asyncio.CancelledError):
                     await turn_error_state.handle(
-                        _handle_cancelled_local_daemon_turn_error
+                        lambda: turn_error_handler.handle_cancelled(session)
                     )
                     return
                 if isinstance(exc, Exception):
                     await turn_error_state.handle(
-                        lambda: _handle_generic_local_daemon_turn_error(exc)
+                        lambda: turn_error_handler.handle_generic(session, exc)
                     )
 
             try:
@@ -4070,25 +4052,28 @@ def _complete_local_daemon_turn_observation(
                 if turn_error_state.handler_failed:
                     raise
                 if not turn_error_state.handled:
-                    await _handle_fatal_local_daemon_turn_error(exc)
+                    await turn_error_handler.handle_fatal(session, exc)
                 raise
             except asyncio.CancelledError:
                 if turn_error_state.handler_failed:
                     raise
                 if not turn_error_state.handled:
-                    await _handle_cancelled_local_daemon_turn_error()
+                    await turn_error_handler.handle_cancelled(session)
                 raise
             except RunCoordinatorError as exc:
                 if turn_error_state.handler_failed:
                     raise
                 if not turn_error_state.handled:
-                    await turn_run.ensure_started(session)
-                    await _handle_generic_local_daemon_turn_error(exc)
+                    await turn_error_handler.handle_generic(
+                        session,
+                        exc,
+                        ensure_started=True,
+                    )
             except Exception as exc:
                 if turn_error_state.handler_failed:
                     raise
                 if not turn_error_state.handled:
-                    await _handle_generic_local_daemon_turn_error(exc)
+                    await turn_error_handler.handle_generic(session, exc)
             finally:
                 current_task = asyncio.current_task()
                 if session.task is None or session.task is not current_task: