fix(subagent): harden child failure routing

coding-agent/.opencode/prompts/tasks/adr-0018-pr2-cancellation-failure-summary-routing.md

@@ -0,0 +1,57 @@
+Goal:
+Harden ADR-0018 PR 2 so child subagent timeout, interruption, adapter failures, cleanup failures, and summary routing are deterministic at the parent surface.
+
+Scope:
+- Convert direct child adapter run exceptions into the existing subagent summary path instead of leaking raw adapter errors past the parent tool result.
+- Keep `asyncio.CancelledError` as cancellation, not a swallowed child failure, while still releasing write leases and closing adapters.
+- Make child interruption and adapter cleanup failures model-visible through deterministic subagent summaries.
+- Preserve hidden child tape trace entries with `skip_context=True` and `subagent_child=True`.
+- Preserve summary publication semantics: ordinary publisher failures are best-effort, but parent session ownership failures remain fail-fast.
+- Ensure parent session ownership failures remain fail-fast even when subagent runs through generic tool execution or parallel batch execution.
+
+Out of scope:
+- Do not add PG-backed or provider-backed distributed child write leases.
+- Do not run child agents in a separate process, queue worker, pod, or cloud callback handler.
+- Do not add a real cloud vendor backend or credentials.
+- Do not move subagent orchestration, cloud clients, owner routing, or approval policy into `agentkit`.
+- Do not enable nested subagents.
+
+Context:
+- ADRs:
+  - `docs/adr/0013-define-phase2-multi-instance-session-ownership-for-pg-http-sessions.md`
+  - `docs/adr/0015-enforce-owner-routed-http-event-streams.md`
+  - `docs/adr/0016-stage-environment-runtime-and-tool-orchestration-prs.md`
+  - `docs/adr/0017-cloud-workspace-execution.md`
+  - `docs/adr/0018-owner-local-cloud-aware-subagent-orchestration.md`
+- Relevant files:
+  - `src/coding_agent/tools/subagent.py`
+  - `src/coding_agent/subagents/coordinator.py`
+  - `src/coding_agent/adapter.py`
+  - `src/coding_agent/adapter_types.py`
+  - `src/agentkit/tools/toolset.py`
+  - `src/coding_agent/plugins/parallel_executor.py`
+  - `src/coding_agent/ui/session_manager.py`
+  - `tests/coding_agent/tools/test_subagent.py`
+  - `tests/coding_agent/subagents/test_coordinator.py`
+  - `tests/coding_agent/plugins/test_parallel_executor.py`
+  - `tests/ui/test_session_manager_owner_checks.py`
+
+Target tests:
+- `uv run pytest tests/agentkit/tools/test_toolset.py -v`
+- `uv run pytest tests/coding_agent/plugins/test_parallel_executor.py -v`
+- `uv run pytest tests/coding_agent/tools/test_subagent.py tests/coding_agent/subagents/test_coordinator.py -v`
+- `uv run pytest tests/coding_agent/test_bootstrap.py tests/ui/test_session_manager_public_api.py tests/ui/test_session_manager_owner_checks.py -k "subagent or cloud" -v`
+- `uv run python -m compileall -q src tests/coding_agent tests/ui`
+
+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.
+- Stop and write a new ADR if implementation needs cross-process child execution, distributed leases, or real cloud vendor transport.
+- Ignore non-blocking optimization suggestions.

coding-agent/docs/adr/0018-owner-local-cloud-aware-subagent-orchestration.md

@@ -160,6 +160,16 @@ fencing semantics, retry/idempotency model, and cloud workspace write policy.
 - [x] `test_subagent_tool_publishes_completion_summary_to_parent_session`
 - [x] `test_subagent_summary_publish_rejects_stale_owner`
 - [x] `uv run pytest tests/coding_agent/tools/test_subagent.py tests/coding_agent/subagents/test_coordinator.py tests/coding_agent/test_bootstrap.py tests/ui/test_session_manager_public_api.py tests/ui/test_session_manager_owner_checks.py -k "subagent or cloud" -v`
+- [x] `test_subagent_tool_publishes_interrupted_summary_to_parent_session`
+- [x] `test_subagent_tool_publishes_run_exception_summary_to_parent_session`
+- [x] `test_subagent_tool_publishes_adapter_close_failure_summary_to_parent_session`
+- [x] `test_subagent_tool_releases_write_lease_and_closes_adapter_on_cancellation`
+- [x] `test_subagent_tool_stale_owner_publish_escapes_toolset_execution`
+- [x] `test_toolset_parallel_batch_propagates_owner_conflict`
+- [x] `uv run pytest tests/agentkit/tools/test_toolset.py -v`
+- [x] `uv run pytest tests/coding_agent/plugins/test_parallel_executor.py -v`
+- [x] `uv run pytest tests/coding_agent/tools/test_subagent.py tests/coding_agent/subagents/test_coordinator.py -v`
+- [x] `uv run pytest tests/coding_agent/test_bootstrap.py tests/ui/test_session_manager_public_api.py tests/ui/test_session_manager_owner_checks.py -k "subagent or cloud" -v`
 
 ## References
 

coding-agent/src/agentkit/tools/__init__.py

@@ -3,6 +3,7 @@
 from agentkit.tools.schema import ToolSchema
 from agentkit.tools.toolset import (
     CALL_TOOL_NAME,
+    FatalToolExecutionError,
     SEARCH_TOOLS_NAME,
     ProxyToolProvider,
     ToolApprovalPolicy,
@@ -29,6 +30,7 @@
     "Toolset",
     "ProxyToolProvider",
     "CALL_TOOL_NAME",
+    "FatalToolExecutionError",
     "SEARCH_TOOLS_NAME",
     "UNHANDLED_TOOL_RESULT",
     "tool",

coding-agent/src/agentkit/tools/toolset.py

@@ -142,6 +142,10 @@ def error_message(self) -> str:
         return ""
 
 
+class FatalToolExecutionError(RuntimeError):
+    """Tool exception that must escape generic tool-result wrapping."""
+
+
 class Toolset:
     """Coordinates tool schema validation, approval, and execution hooks."""
 
@@ -540,6 +544,8 @@ async def _execute_batch(
                         timeout_seconds=options.timeout_seconds,
                     )
                     break
+                except FatalToolExecutionError:
+                    raise
                 except Exception as exc:
                     if attempt == options.max_retries:
                         return self._error_results(calls, exc)
@@ -606,6 +612,8 @@ async def _execute_one_from_hook(
                         timeout_seconds=options.timeout_seconds,
                     )
                     break
+                except FatalToolExecutionError:
+                    raise
                 except Exception as exc:
                     if attempt == options.max_retries:
                         return ToolExecutionResult(
@@ -758,6 +766,8 @@ def _envelope_raw_result(
         call: ToolCallRequest,
         result: Any,
     ) -> ToolExecutionResult:
+        if isinstance(result, FatalToolExecutionError):
+            raise result
         if isinstance(result, Exception):
             return ToolExecutionResult(
                 tool_call_id=call.tool_call_id,

coding-agent/src/coding_agent/plugins/parallel_executor.py

@@ -5,6 +5,8 @@
 import json
 from typing import Any, Awaitable, Callable, cast
 
+from agentkit.tools import FatalToolExecutionError
+
 _FILE_TOOLS = frozenset({"file_read", "file_write", "file_replace"})
 
 _CONFLICT_PAIRS: frozenset[tuple[str, str]] = frozenset(
@@ -144,12 +146,20 @@ async def _run_one(idx: int, call: ToolCallDict) -> tuple[int, Any]:
                                 call["arguments"],
                             )
                         return idx, result
+                    except FatalToolExecutionError:
+                        raise
                     except Exception as exc:
                         return idx, json.dumps({"error": str(exc)})
 
-            batch_results = await asyncio.gather(
-                *[_run_one(idx, call) for idx, call in batch]
-            )
+            tasks: dict[int, asyncio.Task[tuple[int, Any]]] = {}
+            try:
+                async with asyncio.TaskGroup() as task_group:
+                    for idx, call in batch:
+                        tasks[idx] = task_group.create_task(_run_one(idx, call))
+            except* FatalToolExecutionError as exc_group:
+                raise exc_group.exceptions[0]
+
+            batch_results = [tasks[idx].result() for idx, _ in batch]
             for idx, result in batch_results:
                 results_by_index[idx] = result
 

coding-agent/src/coding_agent/tools/subagent.py

@@ -125,6 +125,11 @@ def _summarize_subagent_outcome(outcome: TurnOutcome) -> str:
             raise ValueError("subagent error outcome missing error message")
         return f"Subagent failed: {outcome.error}"
 
+    if outcome.stop_reason == StopReason.INTERRUPTED:
+        if outcome.error:
+            return f"Subagent interrupted: {outcome.error}"
+        return "Subagent interrupted"
+
     if outcome.final_message:
         return f"Subagent completed: {outcome.final_message}"
 
@@ -295,15 +300,53 @@ async def subagent_dispatch(
         )
         outcome: TurnOutcome | None = None
         timed_out = False
+        cleanup_error: str | None = None
         try:
             outcome = await asyncio.wait_for(
                 child_adapter.run_turn(goal), timeout=timeout_seconds
             )
         except asyncio.TimeoutError:
             timed_out = True
+        except asyncio.CancelledError:
+            raise
+        except SessionOwnershipConflictError:
+            raise
+        except Exception as exc:
+            outcome = TurnOutcome(stop_reason=StopReason.ERROR, error=str(exc))
         finally:
-            await child_consumer.close()
-            await _close_adapter_if_supported(child_adapter)
+            saved_cancel: asyncio.CancelledError | None = None
+            try:
+                await child_consumer.close()
+            except asyncio.CancelledError as exc:
+                saved_cancel = exc
+            except Exception as exc:
+                cleanup_error = f"child consumer close failed: {exc}"
+            try:
+                await _close_adapter_if_supported(child_adapter)
+            except asyncio.CancelledError as exc:
+                if saved_cancel is None:
+                    saved_cancel = exc
+            except Exception as exc:
+                adapter_close_error = f"child adapter close failed: {exc}"
+                cleanup_error = (
+                    adapter_close_error
+                    if cleanup_error is None
+                    else f"{cleanup_error}; {adapter_close_error}"
+                )
+            if saved_cancel is not None:
+                raise saved_cancel
+
+        if cleanup_error is not None:
+            if timed_out:
+                cleanup_error = (
+                    f"timed out after {timeout_seconds:g} seconds; {cleanup_error}"
+                )
+            elif outcome is not None and outcome.stop_reason == StopReason.ERROR:
+                if outcome.error is None:
+                    raise ValueError("subagent error outcome missing error message")
+                cleanup_error = f"{outcome.error}; {cleanup_error}"
+            outcome = TurnOutcome(stop_reason=StopReason.ERROR, error=cleanup_error)
+            timed_out = False
 
         if timed_out:
             summary = f"Subagent timed out after {timeout_seconds:g} seconds"

coding-agent/src/coding_agent/ui/session_owner_store.py

@@ -6,6 +6,7 @@
 from typing import Protocol
 
 from agentkit.storage.pg import PGPool, PGSessionOwnerStore
+from agentkit.tools import FatalToolExecutionError
 
 
 class SessionOwnershipConflictReason(StrEnum):
@@ -14,7 +15,7 @@ class SessionOwnershipConflictReason(StrEnum):
     EXPIRED_LEASE = "expired_lease"
 
 
-class SessionOwnershipConflictError(RuntimeError):
+class SessionOwnershipConflictError(FatalToolExecutionError):
     """Raised when the current instance is not authorized to mutate a session."""
 
     def __init__(

coding-agent/tests/coding_agent/plugins/test_parallel_executor.py

@@ -7,10 +7,14 @@
 
 import pytest
 
+from agentkit.plugin.registry import PluginRegistry
+from agentkit.runtime.hook_runtime import HookRuntime
+from agentkit.tools import ToolCallRequest, Toolset
 from coding_agent.plugins.parallel_executor import (
     DependencyAnalyzer,
     ParallelExecutorPlugin,
 )
+from coding_agent.ui.session_owner_store import SessionOwnershipConflictError
 
 
 class TestDependencyAnalyzer:
@@ -233,6 +237,72 @@ async def failing(name: str, arguments: dict[str, Any]) -> str:
         assert "error" in results[0].lower()
         assert "boom" in results[0].lower()
 
+    @pytest.mark.asyncio
+    async def test_toolset_parallel_batch_propagates_owner_conflict(self):
+        async def stale_owner_execute(
+            name: str, arguments: dict[str, Any]
+        ) -> str:
+            del name, arguments
+            raise SessionOwnershipConflictError(
+                "stale owner or fencing token rejected"
+            )
+
+        plugin = ParallelExecutorPlugin(execute_fn=stale_owner_execute)
+        registry = PluginRegistry()
+        registry.register(plugin)
+        toolset = Toolset(runtime=HookRuntime(registry))
+
+        with pytest.raises(
+            SessionOwnershipConflictError,
+            match="stale owner or fencing token rejected",
+        ):
+            await toolset.execute_tools(
+                [
+                    ToolCallRequest(
+                        tool_call_id="tc-1",
+                        name="file_read",
+                        arguments={"path": "a.py"},
+                    ),
+                    ToolCallRequest(
+                        tool_call_id="tc-2",
+                        name="glob_files",
+                        arguments={"pattern": "*.py"},
+                    ),
+                ],
+                ctx=object(),
+            )
+
+    @pytest.mark.asyncio
+    async def test_execute_batch_cancels_sibling_tasks_on_fatal_error(self):
+        sibling_cancelled = asyncio.Event()
+
+        async def execute(name: str, arguments: dict[str, Any]) -> str:
+            del arguments
+            if name == "fatal":
+                raise SessionOwnershipConflictError("stale owner or fencing token rejected")
+            try:
+                await asyncio.sleep(1)
+            except asyncio.CancelledError:
+                sibling_cancelled.set()
+                raise
+            return "unexpected"
+
+        plugin = ParallelExecutorPlugin(execute_fn=execute)
+
+        with pytest.raises(
+            SessionOwnershipConflictError,
+            match="stale owner or fencing token rejected",
+        ):
+            await plugin.execute_batch(
+                tool_calls=[
+                    {"name": "fatal", "arguments": {}},
+                    {"name": "grep", "arguments": {"pattern": "x"}},
+                ]
+            )
+
+        await asyncio.sleep(0)
+        assert sibling_cancelled.is_set() is True
+
     @pytest.mark.asyncio
     async def test_semaphore_limits_concurrency(self):
         concurrent_count = 0