From 22fbf11521ec67ebefd35f68865f57c4895cd68c Mon Sep 17 00:00:00 2001 From: kina0630 Date: Wed, 3 Jun 2026 06:14:43 +0800 Subject: [PATCH] feat: extract runtime turn error handler --- .../local-daemon-pr35-turn-error-handler.md | 48 ++++++ ...mon-control-plane-executor-architecture.md | 3 + .../src/coding_agent/runs/__init__.py | 4 + .../src/coding_agent/runs/lifecycle.py | 76 ++++++++++ .../coding_agent/server/session_manager.py | 65 ++++----- .../tests/coding_agent/test_run_lifecycle.py | 138 ++++++++++++++++++ 6 files changed, 294 insertions(+), 40 deletions(-) create mode 100644 coding-agent/.opencode/prompts/tasks/local-daemon-pr35-turn-error-handler.md diff --git a/coding-agent/.opencode/prompts/tasks/local-daemon-pr35-turn-error-handler.md b/coding-agent/.opencode/prompts/tasks/local-daemon-pr35-turn-error-handler.md new file mode 100644 index 000000000..3f4fd548d --- /dev/null +++ b/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. 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 3b5c25300..b6075f3d2 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 @@ -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 diff --git a/coding-agent/src/coding_agent/runs/__init__.py b/coding-agent/src/coding_agent/runs/__init__.py index d24796b07..d2cf53021 100644 --- a/coding-agent/src/coding_agent/runs/__init__.py +++ b/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", diff --git a/coding-agent/src/coding_agent/runs/lifecycle.py b/coding-agent/src/coding_agent/runs/lifecycle.py index 0fd474e50..3abc4aeec 100644 --- a/coding-agent/src/coding_agent/runs/lifecycle.py +++ b/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", diff --git a/coding-agent/src/coding_agent/server/session_manager.py b/coding-agent/src/coding_agent/server/session_manager.py index 7dc9794a8..3b90eaa6b 100644 --- a/coding-agent/src/coding_agent/server/session_manager.py +++ b/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,6 +3949,14 @@ 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, @@ -3982,17 +3964,17 @@ async def _on_local_daemon_turn_error( 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: diff --git a/coding-agent/tests/coding_agent/test_run_lifecycle.py b/coding-agent/tests/coding_agent/test_run_lifecycle.py index 277e7af50..0058705c7 100644 --- a/coding-agent/tests/coding_agent/test_run_lifecycle.py +++ b/coding-agent/tests/coding_agent/test_run_lifecycle.py @@ -10,6 +10,7 @@ from coding_agent.runtime_store import AgentRunRecord, JSONObject from coding_agent.runs import ( RuntimeRunLifecycle, + RuntimeTurnErrorHandler, RuntimeTurnErrorState, RuntimeTurnFinalizer, RuntimeTurnRunTracker, @@ -386,6 +387,143 @@ async def fail_error_handler() -> None: assert state.handler_failed is True +@pytest.mark.asyncio +async def test_runtime_turn_error_handler_records_generic_failure() -> None: + store = RecordingRuntimeStore() + lifecycle = RuntimeRunLifecycle( + store=store, + metadata_for_session=lambda session, *, resume_context=None: { + "session_id": session.id, + }, + ) + tracker = RuntimeTurnRunTracker( + lifecycle=lifecycle, + run_id="run-1", + started_at=datetime(2026, 1, 1, tzinfo=UTC), + ) + session = FakeTurnSession(id="session-1", tape_id="tape-1") + observation_failures: list[str] = [] + closed: list[str] = [] + notifications: list[tuple[str, str]] = [] + + async def close_runtime(session: FakeTurnSession) -> None: + closed.append(session.id) + + async def notify_generic_error( + session: FakeTurnSession, + exc: Exception, + ) -> None: + notifications.append((session.id, str(exc))) + + await tracker.ensure_started(session) + handler = RuntimeTurnErrorHandler( + turn_run=tracker, + close_runtime=close_runtime, + notify_generic_error=notify_generic_error, + fail_observation=observation_failures.append, + ) + + await handler.handle_generic(session, RuntimeError("boom")) + + assert session.turn_status == "failed" + assert session.last_failure_details == "HTTP session turn failed: boom" + assert observation_failures == ["RuntimeError"] + assert closed == ["session-1"] + assert notifications == [("session-1", "boom")] + assert [update["status"] for update in store.updated] == ["running", "failed"] + assert store.updated[-1]["error"] == "boom" + + +@pytest.mark.asyncio +async def test_runtime_turn_error_handler_records_cancel_without_closing_runtime() -> ( + None +): + store = RecordingRuntimeStore() + lifecycle = RuntimeRunLifecycle( + store=store, + metadata_for_session=lambda session, *, resume_context=None: { + "session_id": session.id, + }, + ) + tracker = RuntimeTurnRunTracker( + lifecycle=lifecycle, + run_id="run-1", + started_at=datetime(2026, 1, 1, tzinfo=UTC), + ) + session = FakeTurnSession(id="session-1", tape_id="tape-1") + cancelled: list[str] = [] + closed: list[str] = [] + notifications: list[str] = [] + + async def close_runtime(session: FakeTurnSession) -> None: + closed.append(session.id) + + async def notify_generic_error( + session: FakeTurnSession, + exc: Exception, + ) -> None: + del session + notifications.append(str(exc)) + + await tracker.ensure_started(session) + handler = RuntimeTurnErrorHandler( + turn_run=tracker, + close_runtime=close_runtime, + notify_generic_error=notify_generic_error, + cancel_observation=lambda: cancelled.append("cancelled"), + ) + + await handler.handle_cancelled(session) + + assert cancelled == ["cancelled"] + assert closed == [] + assert notifications == [] + assert store.updated[-1]["status"] == "cancelled" + assert store.updated[-1]["error"] == "cancelled" + + +@pytest.mark.asyncio +async def test_runtime_turn_error_handler_can_start_before_generic_failure() -> None: + store = RecordingRuntimeStore() + lifecycle = RuntimeRunLifecycle( + store=store, + metadata_for_session=lambda session, *, resume_context=None: { + "session_id": session.id, + }, + ) + tracker = RuntimeTurnRunTracker( + lifecycle=lifecycle, + run_id="run-1", + started_at=datetime(2026, 1, 1, tzinfo=UTC), + ) + session = FakeTurnSession(id="session-1", tape_id="tape-1") + + async def close_runtime(session: FakeTurnSession) -> None: + del session + + async def notify_generic_error( + session: FakeTurnSession, + exc: Exception, + ) -> None: + del session, exc + + handler = RuntimeTurnErrorHandler( + turn_run=tracker, + close_runtime=close_runtime, + notify_generic_error=notify_generic_error, + ) + + await handler.handle_generic( + session, + RuntimeError("submit failed"), + ensure_started=True, + ) + + assert tracker.created is True + assert store.created[0].status == "queued" + assert [update["status"] for update in store.updated] == ["running", "failed"] + + @pytest.mark.asyncio async def test_runtime_turn_run_tracker_starts_only_once() -> None: store = RecordingRuntimeStore()