diff --git a/coding-agent/.opencode/prompts/tasks/adr-0018-pr2-cancellation-failure-summary-routing.md b/coding-agent/.opencode/prompts/tasks/adr-0018-pr2-cancellation-failure-summary-routing.md new file mode 100644 index 000000000..4f99cc52c --- /dev/null +++ b/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. diff --git a/coding-agent/docs/adr/0018-owner-local-cloud-aware-subagent-orchestration.md b/coding-agent/docs/adr/0018-owner-local-cloud-aware-subagent-orchestration.md index 46cce6a47..0997e4fc8 100644 --- a/coding-agent/docs/adr/0018-owner-local-cloud-aware-subagent-orchestration.md +++ b/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 diff --git a/coding-agent/src/agentkit/tools/__init__.py b/coding-agent/src/agentkit/tools/__init__.py index 132d891ff..b07cf7e16 100644 --- a/coding-agent/src/agentkit/tools/__init__.py +++ b/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", diff --git a/coding-agent/src/agentkit/tools/toolset.py b/coding-agent/src/agentkit/tools/toolset.py index 15c28ad07..c64e23469 100644 --- a/coding-agent/src/agentkit/tools/toolset.py +++ b/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, diff --git a/coding-agent/src/coding_agent/plugins/parallel_executor.py b/coding-agent/src/coding_agent/plugins/parallel_executor.py index 2b637e053..26f516339 100644 --- a/coding-agent/src/coding_agent/plugins/parallel_executor.py +++ b/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 diff --git a/coding-agent/src/coding_agent/tools/subagent.py b/coding-agent/src/coding_agent/tools/subagent.py index 08883c0c1..8ab6c79c3 100644 --- a/coding-agent/src/coding_agent/tools/subagent.py +++ b/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" diff --git a/coding-agent/src/coding_agent/ui/session_owner_store.py b/coding-agent/src/coding_agent/ui/session_owner_store.py index bf7c9f6c8..380da56f7 100644 --- a/coding-agent/src/coding_agent/ui/session_owner_store.py +++ b/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__( diff --git a/coding-agent/tests/coding_agent/plugins/test_parallel_executor.py b/coding-agent/tests/coding_agent/plugins/test_parallel_executor.py index ba445afdf..e833604a7 100644 --- a/coding-agent/tests/coding_agent/plugins/test_parallel_executor.py +++ b/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 diff --git a/coding-agent/tests/coding_agent/tools/test_subagent.py b/coding-agent/tests/coding_agent/tools/test_subagent.py index d0f14e1bf..620aebc0b 100644 --- a/coding-agent/tests/coding_agent/tools/test_subagent.py +++ b/coding-agent/tests/coding_agent/tools/test_subagent.py @@ -6,14 +6,18 @@ import pytest +from agentkit.plugin.registry import PluginRegistry from agentkit.providers.models import DoneEvent, TextEvent, ToolCallEvent from agentkit.runtime import AgentRunContext, ContextBudget +from agentkit.runtime.hook_runtime import HookRuntime from agentkit.runtime.pipeline import Pipeline, PipelineContext from agentkit.tape.models import Entry from agentkit.tape.tape import Tape +from agentkit.tools import ToolCallRequest, Toolset from coding_agent.adapter_types import StopReason, TurnOutcome from coding_agent.__main__ import create_agent, create_child_pipeline from coding_agent.environment import CloudCommandResult, CloudEnvironment, LocalEnvironment +from coding_agent.plugins.core_tools import CoreToolsPlugin from coding_agent.ui.session_owner_store import SessionOwnershipConflictError from coding_agent.wire.protocol import StreamDelta, TurnEnd, WireMessage from coding_agent.tools.subagent import build_subagent_tool @@ -443,6 +447,243 @@ async def run_turn(self, _goal: str) -> TurnOutcome: ] +@pytest.mark.asyncio +async def test_subagent_tool_publishes_interrupted_summary_to_parent_session( + monkeypatch: pytest.MonkeyPatch, +) -> None: + calls: list[dict[str, object]] = [] + child_ctx = PipelineContext(tape=Tape(), session_id="parent-session") + events: list[str] = [] + + async def publish_subagent_message( + session_id: str, + text: str, + *, + message_id: str | None = None, + metadata: dict[str, Any] | None = None, + ) -> bool: + calls.append( + { + "session_id": session_id, + "text": text, + "message_id": message_id, + "metadata": metadata, + } + ) + return True + + def child_pipeline_builder(**_kwargs: Any): + return cast(Pipeline, object()), child_ctx + + class InterruptedAdapter: + def __init__(self, **_kwargs: Any) -> None: + pass + + async def run_turn(self, _goal: str) -> TurnOutcome: + return TurnOutcome( + stop_reason=StopReason.INTERRUPTED, + error="Cancelled by user", + ) + + async def close(self) -> None: + events.append("adapter-close") + + monkeypatch.setattr("coding_agent.tools.subagent.PipelineAdapter", InterruptedAdapter) + + tool_fn = build_subagent_tool(child_pipeline_builder) + parent_ctx = PipelineContext( + tape=Tape(), + session_id="parent-session", + config={ + "agent_id": "parent-agent", + "subagent_timeout": 30.0, + "child_worker_coordinator": _StubCoordinator(), + "subagent_message_publisher": publish_subagent_message, + }, + ) + + result = await tool_fn(goal="Inspect", __pipeline_ctx__=parent_ctx) + + assert result == "Subagent interrupted: Cancelled by user" + assert events == ["adapter-close"] + assert calls == [ + { + "session_id": "parent-session", + "text": "Subagent interrupted: Cancelled by user", + "message_id": None, + "metadata": { + "source": "subagent", + "child_agent_id": "parent-agent.child-1", + }, + } + ] + + +@pytest.mark.asyncio +async def test_subagent_tool_publishes_run_exception_summary_to_parent_session( + monkeypatch: pytest.MonkeyPatch, +) -> None: + calls: list[dict[str, object]] = [] + child_ctx = PipelineContext(tape=Tape(), session_id="parent-session") + events: list[str] = [] + + async def publish_subagent_message( + session_id: str, + text: str, + *, + message_id: str | None = None, + metadata: dict[str, Any] | None = None, + ) -> bool: + calls.append( + { + "session_id": session_id, + "text": text, + "message_id": message_id, + "metadata": metadata, + } + ) + return True + + def child_pipeline_builder(**_kwargs: Any): + return cast(Pipeline, object()), child_ctx + + class RaisingAdapter: + def __init__(self, **_kwargs: Any) -> None: + pass + + async def run_turn(self, _goal: str) -> TurnOutcome: + child_ctx.tape.append( + Entry( + kind="message", + payload={"role": "user", "content": "Inspect"}, + ) + ) + raise RuntimeError("adapter mount failed") + + async def close(self) -> None: + events.append("adapter-close") + + monkeypatch.setattr("coding_agent.tools.subagent.PipelineAdapter", RaisingAdapter) + + tool_fn = build_subagent_tool(child_pipeline_builder) + parent_tape = Tape() + parent_ctx = PipelineContext( + tape=parent_tape, + session_id="parent-session", + config={ + "agent_id": "parent-agent", + "subagent_timeout": 30.0, + "child_worker_coordinator": _StubCoordinator(), + "subagent_message_publisher": publish_subagent_message, + }, + ) + + result = await tool_fn(goal="Inspect", __pipeline_ctx__=parent_ctx) + + assert result == "Subagent failed: adapter mount failed" + assert events == ["adapter-close"] + appended_entries = list(parent_tape) + assert len(appended_entries) == 1 + assert appended_entries[0].meta["skip_context"] is True + assert appended_entries[0].meta["subagent_child"] is True + assert appended_entries[0].payload["content"] == "Inspect" + assert calls == [ + { + "session_id": "parent-session", + "text": "Subagent failed: adapter mount failed", + "message_id": None, + "metadata": { + "source": "subagent", + "child_agent_id": "parent-agent.child-1", + }, + } + ] + + +@pytest.mark.asyncio +async def test_subagent_tool_publishes_adapter_close_failure_summary_to_parent_session( + monkeypatch: pytest.MonkeyPatch, +) -> None: + calls: list[dict[str, object]] = [] + child_ctx = PipelineContext(tape=Tape(), session_id="parent-session") + + async def publish_subagent_message( + session_id: str, + text: str, + *, + message_id: str | None = None, + metadata: dict[str, Any] | None = None, + ) -> bool: + calls.append( + { + "session_id": session_id, + "text": text, + "message_id": message_id, + "metadata": metadata, + } + ) + return True + + def child_pipeline_builder(**_kwargs: Any): + return cast(Pipeline, object()), child_ctx + + class CloseFailingAdapter: + def __init__(self, **_kwargs: Any) -> None: + pass + + async def run_turn(self, _goal: str) -> TurnOutcome: + child_ctx.tape.append( + Entry( + kind="message", + payload={"role": "assistant", "content": "Child finished"}, + ) + ) + return TurnOutcome( + stop_reason=StopReason.NO_TOOL_CALLS, + final_message="Child finished", + ) + + async def close(self) -> None: + raise RuntimeError("adapter close failed") + + monkeypatch.setattr( + "coding_agent.tools.subagent.PipelineAdapter", CloseFailingAdapter + ) + + tool_fn = build_subagent_tool(child_pipeline_builder) + parent_tape = Tape() + parent_ctx = PipelineContext( + tape=parent_tape, + session_id="parent-session", + config={ + "agent_id": "parent-agent", + "subagent_timeout": 30.0, + "child_worker_coordinator": _StubCoordinator(), + "subagent_message_publisher": publish_subagent_message, + }, + ) + + result = await tool_fn(goal="Inspect", __pipeline_ctx__=parent_ctx) + + assert result == "Subagent failed: child adapter close failed: adapter close failed" + appended_entries = list(parent_tape) + assert len(appended_entries) == 1 + assert appended_entries[0].meta["skip_context"] is True + assert appended_entries[0].meta["subagent_child"] is True + assert appended_entries[0].payload["content"] == "Child finished" + assert calls == [ + { + "session_id": "parent-session", + "text": "Subagent failed: child adapter close failed: adapter close failed", + "message_id": None, + "metadata": { + "source": "subagent", + "child_agent_id": "parent-agent.child-1", + }, + } + ] + + def test_create_agent_injects_default_child_worker_coordinator() -> None: _pipeline, ctx = create_agent(session_id_override="test-session") @@ -853,6 +1094,104 @@ def child_pipeline_builder(**_kwargs: Any): await tool_fn(goal="Inspect", __pipeline_ctx__=parent_ctx) +@pytest.mark.asyncio +async def test_subagent_tool_stale_owner_publish_escapes_toolset_execution( + monkeypatch: pytest.MonkeyPatch, +) -> None: + child_ctx = PipelineContext(tape=Tape(), session_id="parent-session") + + async def publish_subagent_message( + session_id: str, + text: str, + *, + message_id: str | None = None, + metadata: dict[str, Any] | None = None, + ) -> bool: + del session_id, text, message_id, metadata + raise SessionOwnershipConflictError("stale owner or fencing token rejected") + + def child_pipeline_builder(**_kwargs: Any): + return cast(Pipeline, object()), child_ctx + + monkeypatch.setattr( + "coding_agent.tools.subagent.PipelineAdapter", _make_immediate_adapter() + ) + + core_tools = CoreToolsPlugin(child_pipeline_builder=child_pipeline_builder) + registry = PluginRegistry() + registry.register(core_tools) + toolset = Toolset(runtime=HookRuntime(registry)) + parent_ctx = PipelineContext( + tape=Tape(), + session_id="parent-session", + config={ + "agent_id": "parent-agent", + "subagent_timeout": 30.0, + "child_worker_coordinator": _StubCoordinator(), + "subagent_message_publisher": publish_subagent_message, + }, + ) + + with pytest.raises( + SessionOwnershipConflictError, + match="stale owner or fencing token rejected", + ): + await toolset.execute_tools( + [ + ToolCallRequest( + tool_call_id="tc-subagent", + name="subagent", + arguments={"goal": "Inspect"}, + ) + ], + ctx=parent_ctx, + ) + + +@pytest.mark.asyncio +async def test_subagent_tool_reraises_stale_owner_from_child_run_turn( + monkeypatch: pytest.MonkeyPatch, +) -> None: + child_ctx = PipelineContext(tape=Tape(), session_id="parent-session") + events: list[str] = [] + + def child_pipeline_builder(**_kwargs: Any): + return cast(Pipeline, object()), child_ctx + + class StaleOwnerAdapter: + def __init__(self, **_kwargs: Any) -> None: + pass + + async def run_turn(self, _goal: str) -> TurnOutcome: + raise SessionOwnershipConflictError("stale owner inside child run") + + async def close(self) -> None: + events.append("adapter-close") + + monkeypatch.setattr( + "coding_agent.tools.subagent.PipelineAdapter", StaleOwnerAdapter + ) + + tool_fn = build_subagent_tool(child_pipeline_builder) + parent_ctx = PipelineContext( + tape=Tape(), + session_id="parent-session", + config={ + "agent_id": "parent-agent", + "subagent_timeout": 30.0, + "child_worker_coordinator": _StubCoordinator(), + }, + ) + + with pytest.raises( + SessionOwnershipConflictError, + match="stale owner inside child run", + ): + await tool_fn(goal="Inspect", __pipeline_ctx__=parent_ctx) + + assert events == ["adapter-close"] + + @pytest.mark.asyncio async def test_subagent_tool_fails_fast_on_empty_parent_session_id( monkeypatch: pytest.MonkeyPatch, @@ -1349,3 +1688,54 @@ async def close(self) -> None: await task assert events == ["lease-enter", "lease-exit", "adapter-close"] + + +@pytest.mark.asyncio +async def test_subagent_tool_closes_adapter_when_consumer_close_is_cancelled( + monkeypatch: pytest.MonkeyPatch, +): + events: list[str] = [] + child_ctx = PipelineContext(tape=Tape(), session_id="parent-session") + + def child_pipeline_builder(**_kwargs: Any): + return cast(Pipeline, object()), child_ctx + + class SuccessfulAdapter: + def __init__(self, **_kwargs: Any) -> None: + pass + + async def run_turn(self, _goal: str) -> TurnOutcome: + return TurnOutcome( + stop_reason=StopReason.NO_TOOL_CALLS, + final_message="Child finished", + ) + + async def close(self) -> None: + events.append("adapter-close") + + async def cancelled_close(self) -> None: + del self + events.append("consumer-close") + raise asyncio.CancelledError() + + monkeypatch.setattr("coding_agent.tools.subagent.PipelineAdapter", SuccessfulAdapter) + monkeypatch.setattr( + "coding_agent.tools.subagent._ChildWriteLeaseConsumer.close", + cancelled_close, + ) + + tool_fn = build_subagent_tool(child_pipeline_builder) + parent_ctx = PipelineContext( + tape=Tape(), + session_id="parent-session", + config={ + "agent_id": "parent-agent", + "subagent_timeout": 30.0, + "child_worker_coordinator": _StubCoordinator(), + }, + ) + + with pytest.raises(asyncio.CancelledError): + await tool_fn(goal="Inspect", __pipeline_ctx__=parent_ctx) + + assert events == ["consumer-close", "adapter-close"]