From f7aad831255cdf3bf993db22479cd45764d1a923 Mon Sep 17 00:00:00 2001 From: kina0630 Date: Tue, 5 May 2026 01:37:18 +0800 Subject: [PATCH 1/7] docs(adr): add PR2 subagent hardening packet Ultraworked with [Sisyphus](https://github.com/code-yeongyu/oh-my-openagent) Co-authored-by: Sisyphus --- ...r2-cancellation-failure-summary-routing.md | 57 +++++++++++++++++++ ...ocal-cloud-aware-subagent-orchestration.md | 10 ++++ 2 files changed, 67 insertions(+) create mode 100644 coding-agent/.opencode/prompts/tasks/adr-0018-pr2-cancellation-failure-summary-routing.md 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 From f78acc181b4ee6fcbb8182e9e87ac108c64b6a6c Mon Sep 17 00:00:00 2001 From: kina0630 Date: Tue, 5 May 2026 01:37:18 +0800 Subject: [PATCH 2/7] fix(toolset): preserve fatal tool exceptions Ultraworked with [Sisyphus](https://github.com/code-yeongyu/oh-my-openagent) Co-authored-by: Sisyphus --- coding-agent/src/agentkit/tools/__init__.py | 2 ++ coding-agent/src/agentkit/tools/toolset.py | 10 ++++++++++ 2 files changed, 12 insertions(+) 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, From 265d02b16358664f161808d6d2eb413fd5943283 Mon Sep 17 00:00:00 2001 From: kina0630 Date: Tue, 5 May 2026 01:37:18 +0800 Subject: [PATCH 3/7] fix(session): propagate owner fencing failures Ultraworked with [Sisyphus](https://github.com/code-yeongyu/oh-my-openagent) Co-authored-by: Sisyphus --- .../coding_agent/plugins/parallel_executor.py | 4 ++ .../coding_agent/ui/session_owner_store.py | 3 +- .../plugins/test_parallel_executor.py | 39 +++++++++++++++++++ 3 files changed, 45 insertions(+), 1 deletion(-) diff --git a/coding-agent/src/coding_agent/plugins/parallel_executor.py b/coding-agent/src/coding_agent/plugins/parallel_executor.py index 2b637e053..a7e75b415 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,6 +146,8 @@ 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)}) 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..41f8b27e1 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,41 @@ 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_semaphore_limits_concurrency(self): concurrent_count = 0 From fa9eaeef61459e91c7e8ac653f0011570ff6a3c4 Mon Sep 17 00:00:00 2001 From: kina0630 Date: Tue, 5 May 2026 01:37:18 +0800 Subject: [PATCH 4/7] fix(subagent): route child failure summaries Ultraworked with [Sisyphus](https://github.com/code-yeongyu/oh-my-openagent) Co-authored-by: Sisyphus --- .../src/coding_agent/tools/subagent.py | 41 ++- .../tests/coding_agent/tools/test_subagent.py | 290 ++++++++++++++++++ 2 files changed, 329 insertions(+), 2 deletions(-) diff --git a/coding-agent/src/coding_agent/tools/subagent.py b/coding-agent/src/coding_agent/tools/subagent.py index 08883c0c1..d374d0053 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,47 @@ 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 Exception as exc: + outcome = TurnOutcome(stop_reason=StopReason.ERROR, error=str(exc)) finally: - await child_consumer.close() - await _close_adapter_if_supported(child_adapter) + try: + await child_consumer.close() + except asyncio.CancelledError: + raise + except Exception as exc: + cleanup_error = f"child consumer close failed: {exc}" + try: + await _close_adapter_if_supported(child_adapter) + except asyncio.CancelledError: + raise + 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 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/tests/coding_agent/tools/test_subagent.py b/coding-agent/tests/coding_agent/tools/test_subagent.py index d0f14e1bf..fff08b6d2 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,238 @@ 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") + + 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", + ) + + 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 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 +1089,60 @@ 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_fails_fast_on_empty_parent_session_id( monkeypatch: pytest.MonkeyPatch, From 9585f71b5279410b4656c892d5f7acef3190abe4 Mon Sep 17 00:00:00 2001 From: kina0630 Date: Tue, 5 May 2026 12:19:13 +0800 Subject: [PATCH 5/7] fix(subagent): close adapter on cleanup cancel Ultraworked with [Sisyphus](https://github.com/code-yeongyu/oh-my-openagent) Co-authored-by: Sisyphus --- .../src/coding_agent/tools/subagent.py | 12 ++-- .../tests/coding_agent/tools/test_subagent.py | 56 +++++++++++++++++++ 2 files changed, 64 insertions(+), 4 deletions(-) diff --git a/coding-agent/src/coding_agent/tools/subagent.py b/coding-agent/src/coding_agent/tools/subagent.py index d374d0053..59ee5eec1 100644 --- a/coding-agent/src/coding_agent/tools/subagent.py +++ b/coding-agent/src/coding_agent/tools/subagent.py @@ -312,16 +312,18 @@ async def subagent_dispatch( except Exception as exc: outcome = TurnOutcome(stop_reason=StopReason.ERROR, error=str(exc)) finally: + saved_cancel: asyncio.CancelledError | None = None try: await child_consumer.close() - except asyncio.CancelledError: - raise + 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: - raise + 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 = ( @@ -329,6 +331,8 @@ async def subagent_dispatch( 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: diff --git a/coding-agent/tests/coding_agent/tools/test_subagent.py b/coding-agent/tests/coding_agent/tools/test_subagent.py index fff08b6d2..ff9af4c42 100644 --- a/coding-agent/tests/coding_agent/tools/test_subagent.py +++ b/coding-agent/tests/coding_agent/tools/test_subagent.py @@ -453,6 +453,7 @@ async def test_subagent_tool_publishes_interrupted_summary_to_parent_session( ) -> 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, @@ -484,6 +485,9 @@ async def run_turn(self, _goal: str) -> TurnOutcome: 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) @@ -501,6 +505,7 @@ async def run_turn(self, _goal: str) -> TurnOutcome: 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", @@ -1639,3 +1644,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"] From 22c3b7364e3ef1a05cb03c89371a4d9eb88f942c Mon Sep 17 00:00:00 2001 From: kina0630 Date: Tue, 5 May 2026 12:19:13 +0800 Subject: [PATCH 6/7] fix(parallel): cancel siblings on fatal error Ultraworked with [Sisyphus](https://github.com/code-yeongyu/oh-my-openagent) Co-authored-by: Sisyphus --- .../coding_agent/plugins/parallel_executor.py | 12 +++++-- .../plugins/test_parallel_executor.py | 31 +++++++++++++++++++ 2 files changed, 40 insertions(+), 3 deletions(-) diff --git a/coding-agent/src/coding_agent/plugins/parallel_executor.py b/coding-agent/src/coding_agent/plugins/parallel_executor.py index a7e75b415..26f516339 100644 --- a/coding-agent/src/coding_agent/plugins/parallel_executor.py +++ b/coding-agent/src/coding_agent/plugins/parallel_executor.py @@ -151,9 +151,15 @@ async def _run_one(idx: int, call: ToolCallDict) -> tuple[int, Any]: 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/tests/coding_agent/plugins/test_parallel_executor.py b/coding-agent/tests/coding_agent/plugins/test_parallel_executor.py index 41f8b27e1..e833604a7 100644 --- a/coding-agent/tests/coding_agent/plugins/test_parallel_executor.py +++ b/coding-agent/tests/coding_agent/plugins/test_parallel_executor.py @@ -272,6 +272,37 @@ async def stale_owner_execute( 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 From 47fc18cd09bb54e591bdb127e41f7314a71d4d24 Mon Sep 17 00:00:00 2001 From: kina0630 Date: Tue, 5 May 2026 12:32:40 +0800 Subject: [PATCH 7/7] fix(subagent): fail fast on child owner conflict Ultraworked with [Sisyphus](https://github.com/code-yeongyu/oh-my-openagent) Co-authored-by: Sisyphus --- .../src/coding_agent/tools/subagent.py | 2 + .../tests/coding_agent/tools/test_subagent.py | 44 +++++++++++++++++++ 2 files changed, 46 insertions(+) diff --git a/coding-agent/src/coding_agent/tools/subagent.py b/coding-agent/src/coding_agent/tools/subagent.py index 59ee5eec1..8ab6c79c3 100644 --- a/coding-agent/src/coding_agent/tools/subagent.py +++ b/coding-agent/src/coding_agent/tools/subagent.py @@ -309,6 +309,8 @@ async def subagent_dispatch( timed_out = True except asyncio.CancelledError: raise + except SessionOwnershipConflictError: + raise except Exception as exc: outcome = TurnOutcome(stop_reason=StopReason.ERROR, error=str(exc)) finally: diff --git a/coding-agent/tests/coding_agent/tools/test_subagent.py b/coding-agent/tests/coding_agent/tools/test_subagent.py index ff9af4c42..620aebc0b 100644 --- a/coding-agent/tests/coding_agent/tools/test_subagent.py +++ b/coding-agent/tests/coding_agent/tools/test_subagent.py @@ -1148,6 +1148,50 @@ def child_pipeline_builder(**_kwargs: Any): ) +@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,