diff --git a/.gitignore b/.gitignore index 9833294..119c844 100644 --- a/.gitignore +++ b/.gitignore @@ -68,6 +68,7 @@ next-steps.md # Working folders ai_working/tmp +.next/ # Jupyter checkpoints .ipynb_checkpoints/ diff --git a/modules/tool-delegate/README.md b/modules/tool-delegate/README.md index 11ff3ec..4aa9f99 100644 --- a/modules/tool-delegate/README.md +++ b/modules/tool-delegate/README.md @@ -30,6 +30,56 @@ Resume sessions using the full `session_id` returned by previous delegate calls: session_id: "abc123-def456-..._foundation:explorer" ``` +### Layered bounding: call budget (Layer 1) + wall-clock backstop (Layer 3) + +Delegated sessions are bounded two ways, and they are meant to be read as a +pair, not independently: + +1. **Layer 1 -- per-leg LLM-call budget** (`settings.max_llm_calls`, off by + default). Enforced in the child's own orchestrator loop via + `max_iterations`; exhaustion is a normal turn ending (the child wraps up + with its own summary and the transcript stays complete and resumable). + This is the layer that should actually catch a runaway agent. See the + "Layer 1 call budget" section below. +2. **Layer 2 -- provider HTTP timeouts.** Already shipped by every provider + in the ecosystem (120-600s). Not implemented in this module; a hung + single LLM call self-resolves as an `LLMError` well before Layer 3 would + ever fire. +3. **Layer 3 -- wall-clock backstop** (`settings.timeout`, described below). + This is what this section documents. It is orchestrator-independent and + deliberately generous: it exists for the residual case Layer 1 cannot + cover -- an orchestrator with no call-budget support, or a single + hanging tool call with no internal timeout of its own. If this backstop + fires on a session that has a working Layer 1 budget, treat that as a + bug report about the budget, not evidence the backstop is too loose. + +#### Delegate Timeout (Layer 3) + +Delegated spawn and resume operations time out after **14400 seconds (4 +hours)** by default. This is roughly 12x the measured healthy upper bound +for a delegated sub-session and roughly half the duration of the worst +observed runaway -- generous enough that it should essentially never fire +in front of a working Layer 1 budget, while still bounding the case where +Layer 1 does not apply. Configure `settings.timeout` with a positive finite +number of seconds to change the limit, or set it explicitly to `null` to +disable the delegate-level timeout. + +A timeout returns `success: false` with structured output containing +`status: timed_out`, the child `session_id`, the agent identity when available, +and metadata with `timeout_seconds`, `resumable: false`, and +`resume_status: pending_child_cleanup`. It emits `delegate:error` with +`error_type: delegate_timeout`, not `delegate:agent_completed`, because the +cancelled child may still be cleaning up. + +Do not immediately resume the returned session ID. The coordinated +persistence-capable `amplifier-app-cli` spawner may persist the interrupted +session after cancellation cleanup finishes, but the delegate timeout response +does not claim that persistence is complete or that the session is ready to +resume. (This honest `resumable: false` reporting stays until +`amplifier-app-cli#260` lands -- it is not on the critical path for Layer 1 or +Layer 3 to ship, since Layer 1's own exit is a normal return and its +`resumable: true` is already a fact today.) + ### Tool Inheritance Fix Agent's explicit tool declarations are always honored, even when parent excludes them. Exclusions apply only to inheritance, not explicit declarations. @@ -70,7 +120,7 @@ modules: exclude_tools: - delegate # Default: spawned agents can't further delegate exclude_hooks: [] - timeout: 300 + timeout: 14400 # Layer 3 backstop default (4h); set to null to disable max_llm_calls: null # Layer 1 call budget (spec: 298-replacement). # None/unset (default): ships dark -- no # budget is injected into any child session; diff --git a/modules/tool-delegate/amplifier_module_tool_delegate/__init__.py b/modules/tool-delegate/amplifier_module_tool_delegate/__init__.py index f0e4cf5..c3d71b6 100644 --- a/modules/tool-delegate/amplifier_module_tool_delegate/__init__.py +++ b/modules/tool-delegate/amplifier_module_tool_delegate/__init__.py @@ -19,7 +19,16 @@ - features.provider_selection.enabled: Allow provider preferences (default: True) - settings.exclude_tools: Tools spawned agents should NOT inherit (default: ["tool-delegate"]) - settings.exclude_hooks: Hooks spawned agents should NOT inherit (default: []) -- settings.timeout: Maximum total execution time for child session in seconds (default: None/disabled) +- settings.timeout: Maximum child-session execution time in seconds (default: 14400, + i.e. 4 hours); set explicitly to None/null to disable. This is a Layer 3 + wall-clock BACKSTOP -- orchestrator-independent and intentionally generous + (~12x the measured healthy upper bound). It exists for the cases a per-leg + LLM-call budget (settings.max_llm_calls) cannot cover: an orchestrator with no + budget support, or a single hung call. If a real orchestrator with a call + budget makes this timeout fire in practice, that is a signal the budget itself + needs attention, not that this default is too generous. Timeouts return the + child session ID, but callers must wait for app-layer cancellation cleanup and + persistence before attempting to resume it. - settings.strict_model_role: When True, a model_role that resolves to no candidates raises ModelRoleUnresolvedError instead of silently falling back to the session default model (default: False). Regardless of this @@ -43,10 +52,13 @@ import asyncio import json import logging +import math import re +from collections.abc import Coroutine from typing import Any from amplifier_core import ModuleCoordinator, ToolResult + from amplifier_foundation import ProviderPreference from amplifier_foundation.tracing import generate_sub_session_id @@ -228,6 +240,40 @@ def _validate_call_budget(value: Any) -> int | None: return value or None # 0 -> None (explicit opt-out) +class _DelegateTimeoutExpired(Exception): + """Internal signal that the delegate-owned timeout expired.""" + + +def _validate_timeout(timeout: object) -> int | float | None: + """Return a timeout that asyncio's event loop can represent. + + ``asyncio.wait`` takes a float timeout. Validate that conversion at + configuration time, before spawning a child coroutine, so an oversized + integer cannot fail later after work has begun. + """ + if timeout is None: + return None + if isinstance(timeout, bool) or not isinstance(timeout, (int, float)): + raise TypeError( + "settings.timeout must be null or a positive finite, non-boolean " + "number of seconds" + ) + + try: + event_loop_timeout = float(timeout) + except OverflowError as error: + raise ValueError( + "settings.timeout must be representable as a finite event-loop timeout" + ) from error + + if timeout <= 0 or not math.isfinite(event_loop_timeout): + raise ValueError( + "settings.timeout must be null or a positive finite, non-boolean " + "number of seconds" + ) + return timeout + + async def mount(coordinator: ModuleCoordinator, config: dict[str, Any] | None = None): """Mount the agent delegation tool. @@ -326,7 +372,8 @@ def __init__(self, coordinator: ModuleCoordinator, config: dict[str, Any]): # Settings self.exclude_tools: list[str] = settings.get("exclude_tools", ["tool-delegate"]) self.exclude_hooks: list[str] = settings.get("exclude_hooks", []) - self.timeout: int | None = settings.get("timeout", None) + self.timeout = _validate_timeout(settings.get("timeout", 14400)) + self._detached_child_tasks: set[asyncio.Task[Any]] = set() # When True, model_role resolving to no candidates raises # ModelRoleUnresolvedError instead of silently falling back to the # session default model. Default False preserves existing behavior @@ -431,6 +478,62 @@ def _build_feature_registry(self) -> list[dict[str, Any]]: }, ] + async def _await_child_with_deadline( + self, child_coro: Coroutine[Any, Any, Any] + ) -> Any: + """Await a child while releasing the parent at the configured deadline. + + Unlike ``asyncio.timeout`` and ``asyncio.wait_for``, this does not wait + for a child that catches ``CancelledError`` or performs slow cancellation + cleanup. The child is cancelled, detached, and its terminal result is + consumed by a callback. A cancellation of this parent task follows the + same cleanup path but is re-raised unchanged. + """ + if self.timeout is None: + return await child_coro + + child_task = asyncio.create_task(child_coro) + try: + done, _ = await asyncio.wait( + (child_task,), + timeout=float(self.timeout), + return_when=asyncio.ALL_COMPLETED, + ) + except asyncio.CancelledError: + self._cancel_and_detach_child(child_task) + raise + + if child_task in done: + return child_task.result() + + self._cancel_and_detach_child(child_task) + raise _DelegateTimeoutExpired + + def _cancel_and_detach_child(self, child_task: asyncio.Task[Any]) -> None: + """Cancel a child while retaining it strongly until terminal cleanup.""" + if not child_task.done(): + child_task.cancel() + if child_task.done(): + self._consume_detached_child_result(child_task) + return + + self._detached_child_tasks.add(child_task) + child_task.add_done_callback(self._consume_detached_child_result) + + def _consume_detached_child_result(self, child_task: asyncio.Task[Any]) -> None: + """Consume a detached child result and release its strong reference.""" + try: + child_task.result() + except asyncio.CancelledError: + pass + except BaseException: + logger.debug( + "Detached delegate child finished with an exception after cancellation", + exc_info=True, + ) + finally: + self._detached_child_tasks.discard(child_task) + def _compose_feature_descriptions(self) -> str: """Compose feature descriptions based on enabled state. @@ -1679,11 +1782,7 @@ async def _spawn_new_session( self_delegation_depth=child_self_delegation_depth, session_metadata=session_metadata, ) - if self.timeout is not None: - async with asyncio.timeout(self.timeout): - result = await spawn_coro - else: - result = await spawn_coro + result = await self._await_child_with_deadline(spawn_coro) # Structured delegation return contract: parse the sub-agent's # response once (no-op when the feature is disabled -- see @@ -1795,16 +1894,14 @@ async def _spawn_new_session( ) raise - except TimeoutError: - # asyncio.timeout raises TimeoutError (which may propagate as - # CancelledError internally). Surface the source clearly so the - # caller knows this was a delegation-level wall-clock timeout, - # not a provider or network issue. + except _DelegateTimeoutExpired: + recovery_msg = ( + "Child cancellation cleanup is still in progress; do not resume " + "this session until cleanup and persistence complete." + ) timeout_msg = ( f"Agent '{agent_name}' timed out after {self.timeout}s " - f"(delegate tool session-level timeout). " - f"Increase or disable the timeout in tool-delegate settings " - f"(settings.timeout) to allow longer-running agents." + f"(delegate tool session-level timeout). {recovery_msg}" ) logger.warning(timeout_msg) if hooks: @@ -1815,11 +1912,30 @@ async def _spawn_new_session( "sub_session_id": sub_session_id, "parent_session_id": parent_session_id, "error": timeout_msg, + "error_type": "delegate_timeout", + "status": "timed_out", + "timeout_seconds": self.timeout, + "resumable": False, + "resume_status": "pending_child_cleanup", "tool_call_id": tool_call_id, "parallel_group_id": parallel_group_id, }, ) - return ToolResult(success=False, error={"message": timeout_msg}) + return ToolResult( + success=False, + output={ + "session_id": sub_session_id, + "agent": agent_name, + "status": "timed_out", + "metadata": { + "timeout_seconds": self.timeout, + "resumable": False, + "resume_status": "pending_child_cleanup", + "recovery_message": recovery_msg, + }, + }, + error={"message": timeout_msg}, + ) except Exception as e: # Emit delegate:error event — include the exception type so the @@ -1901,6 +2017,9 @@ async def _resume_existing_session( ToolResult with success status and output or error """ parent_session_id = self.coordinator.session_id + resume_agent = None + if "_" in session_id: + resume_agent = session_id.rsplit("_", 1)[-1] or None # Resolve agent identity BEFORE the try block (and before emitting # any events), from the most reliable in-repo source available @@ -1967,11 +2086,7 @@ async def _resume_existing_session( sub_session_id=full_session_id, instruction=effective_instruction, ) - if self.timeout is not None: - async with asyncio.timeout(self.timeout): - result = await resume_coro - else: - result = await resume_coro + result = await self._await_child_with_deadline(resume_coro) # Structured delegation return contract (see the spawn path for # the full explanation) -- computed once, reused for telemetry @@ -1998,6 +2113,8 @@ async def _resume_existing_session( # Return output with session info. "response" is `cleaned_response` # -- see the spawn path's comment for the exact byte-identity # guarantee this preserves in the disabled/non-conformant paths. + # `agent_name` was already resolved above (before the try block) + # via `_resolve_agent_for_session` -- no re-derivation here. session_id_result = result["session_id"] return ToolResult( success=True, @@ -2066,30 +2183,53 @@ async def _resume_existing_session( ) raise - except TimeoutError: + except _DelegateTimeoutExpired: # Resolve agent name for the message the same way as everywhere # else on this path (cache first, session_id suffix fallback). resume_agent = self._resolve_agent_for_session(session_id) + agent_label = resume_agent or "unknown" + recovery_msg = ( + "Child cancellation cleanup is still in progress; do not resume " + "this session until cleanup and persistence complete." + ) timeout_msg = ( - f"Resumed agent '{resume_agent}' timed out after {self.timeout}s " - f"(delegate tool session-level timeout). " - f"Increase or disable the timeout in tool-delegate settings " - f"(settings.timeout) to allow longer-running agents." + f"Resumed agent '{agent_label}' timed out after {self.timeout}s " + f"(delegate tool session-level timeout). {recovery_msg}" ) logger.warning(timeout_msg) if hooks: - await hooks.emit( - "delegate:error", - { - "agent": resume_agent, - "session_id": session_id, - "parent_session_id": parent_session_id, - "error": timeout_msg, - "tool_call_id": tool_call_id, - "parallel_group_id": parallel_group_id, - }, - ) - return ToolResult(success=False, error={"message": timeout_msg}) + error_payload = { + "session_id": session_id, + "parent_session_id": parent_session_id, + "error": timeout_msg, + "error_type": "delegate_timeout", + "status": "timed_out", + "timeout_seconds": self.timeout, + "resumable": False, + "resume_status": "pending_child_cleanup", + "tool_call_id": tool_call_id, + "parallel_group_id": parallel_group_id, + } + if resume_agent is not None: + error_payload["agent"] = resume_agent + await hooks.emit("delegate:error", error_payload) + timeout_output = { + "session_id": session_id, + "status": "timed_out", + "metadata": { + "timeout_seconds": self.timeout, + "resumable": False, + "resume_status": "pending_child_cleanup", + "recovery_message": recovery_msg, + }, + } + if resume_agent is not None: + timeout_output["agent"] = resume_agent + return ToolResult( + success=False, + output=timeout_output, + error={"message": timeout_msg}, + ) except Exception as e: # Other errors — include exception type for clear source attribution diff --git a/modules/tool-delegate/tests/test_delegate_timeout.py b/modules/tool-delegate/tests/test_delegate_timeout.py new file mode 100644 index 0000000..081d042 --- /dev/null +++ b/modules/tool-delegate/tests/test_delegate_timeout.py @@ -0,0 +1,467 @@ +"""Tests for delegate timeout configuration, results, and lifecycle events.""" + +from __future__ import annotations + +import asyncio +import gc +import logging +from unittest.mock import AsyncMock, MagicMock + +import pytest +from amplifier_module_tool_delegate import DelegateTool + +_ABSENT = object() + + +def _make_tool( + *, + timeout: object = _ABSENT, + spawn_fn=None, + resume_fn=None, +) -> DelegateTool: + coordinator = MagicMock() + coordinator.session_id = "parent-session-123" + coordinator.config = {"agents": {"test-agent": {}}} + coordinator.session_state = {} + coordinator._tool_dispatch_context = {} + coordinator._tool_dispatch_contexts = {} + + capabilities = { + "session.spawn": spawn_fn or AsyncMock(), + "session.resume": resume_fn or AsyncMock(), + "self_delegation_depth": 0, + } + coordinator.get_capability = lambda name: capabilities.get(name) + coordinator.get = MagicMock(return_value=None) + + parent_session = MagicMock() + parent_session.config = {"session": {"orchestrator": {}}} + coordinator.session = parent_session + + settings: dict[str, object] = {"exclude_tools": []} + if timeout is not _ABSENT: + settings["timeout"] = timeout + return DelegateTool(coordinator, {"features": {}, "settings": settings}) + + +def _hooks() -> MagicMock: + hooks = MagicMock() + hooks.emit = AsyncMock() + return hooks + + +async def _never_finishes(**_kwargs): + await asyncio.Future() + + +async def _is_cancelled(**_kwargs): + raise asyncio.CancelledError + + +async def _capability_times_out(**_kwargs): + raise TimeoutError("capability timeout") + + +async def _spawn(tool: DelegateTool, hooks): + return await tool._spawn_new_session( + agent_name="test-agent", + instruction="Do something", + context_depth="none", + context_scope="conversation", + context_turns=5, + provider_preferences=None, + hooks=hooks, + tool_call_id="call-timeout", + parallel_group_id="parallel-timeout", + ) + + +async def _resume(tool: DelegateTool, hooks): + return await tool._resume_existing_session( + session_id="child-session-001_test-agent", + instruction="Continue", + hooks=hooks, + tool_call_id="call-resume-timeout", + parallel_group_id="parallel-resume-timeout", + ) + + +def _emissions(hooks: MagicMock) -> list[tuple[str, dict]]: + return [(args[0], args[1]) for args, _kwargs in hooks.emit.call_args_list] + + +def test_timeout_defaults_only_when_key_is_absent(): + assert _make_tool().timeout == 14400 + assert _make_tool(timeout=None).timeout is None + + +@pytest.mark.parametrize("timeout", [1, 0.5, 14400, 10**100]) +def test_timeout_accepts_positive_finite_non_bool_numbers(timeout): + assert _make_tool(timeout=timeout).timeout == timeout + + +@pytest.mark.parametrize("timeout", [True, False, "1", []]) +def test_timeout_rejects_invalid_types_eagerly(timeout): + with pytest.raises(TypeError, match="settings.timeout"): + _make_tool(timeout=timeout) + + +@pytest.mark.parametrize( + "timeout", + [0, -1, -0.5, float("inf"), float("-inf"), float("nan")], +) +def test_timeout_rejects_invalid_numeric_values_eagerly(timeout): + with pytest.raises(ValueError, match="settings.timeout"): + _make_tool(timeout=timeout) + + +@pytest.mark.parametrize("timeout", [10**1000, float("inf")]) +def test_timeout_rejects_unrepresentable_values_before_creating_a_child( + timeout, recwarn +): + spawn_fn = MagicMock() + + with pytest.raises(ValueError, match="settings.timeout"): + _make_tool(timeout=timeout, spawn_fn=spawn_fn) + + assert spawn_fn.call_count == 0 + assert not recwarn + + +@pytest.mark.asyncio +async def test_spawn_timeout_reports_pending_cleanup_error_without_completed_event(): + spawn_fn = AsyncMock(side_effect=_never_finishes) + hooks = _hooks() + tool = _make_tool(timeout=0.01, spawn_fn=spawn_fn) + + result = await _spawn(tool, hooks) + + child_session_id = spawn_fn.call_args.kwargs["sub_session_id"] + assert result.success is False + assert result.output == { + "session_id": child_session_id, + "agent": "test-agent", + "status": "timed_out", + "metadata": { + "timeout_seconds": 0.01, + "resumable": False, + "resume_status": "pending_child_cleanup", + "recovery_message": ( + "Child cancellation cleanup is still in progress; do not resume " + "this session until cleanup and persistence complete." + ), + }, + } + + emissions = _emissions(hooks) + event_names = [name for name, _payload in emissions] + assert "delegate:agent_cancelled" not in event_names + assert "delegate:agent_completed" not in event_names + error = next(payload for name, payload in emissions if name == "delegate:error") + assert error["agent"] == "test-agent" + assert error["sub_session_id"] == child_session_id + assert error["error_type"] == "delegate_timeout" + assert error["status"] == "timed_out" + assert error["timeout_seconds"] == 0.01 + assert error["resumable"] is False + assert error["resume_status"] == "pending_child_cleanup" + assert error["tool_call_id"] == "call-timeout" + assert error["parallel_group_id"] == "parallel-timeout" + + +@pytest.mark.asyncio +async def test_resume_timeout_reports_pending_cleanup_error_without_completed_event(): + resume_fn = AsyncMock(side_effect=_never_finishes) + hooks = _hooks() + tool = _make_tool(timeout=0.01, resume_fn=resume_fn) + session_id = "child-session-001_test-agent" + + result = await tool._resume_existing_session( + session_id=session_id, + instruction="Continue", + hooks=hooks, + tool_call_id="call-resume-timeout", + parallel_group_id="parallel-resume-timeout", + ) + + assert result.success is False + assert result.output == { + "session_id": session_id, + "agent": "test-agent", + "status": "timed_out", + "metadata": { + "timeout_seconds": 0.01, + "resumable": False, + "resume_status": "pending_child_cleanup", + "recovery_message": ( + "Child cancellation cleanup is still in progress; do not resume " + "this session until cleanup and persistence complete." + ), + }, + } + + emissions = _emissions(hooks) + event_names = [name for name, _payload in emissions] + assert "delegate:agent_cancelled" not in event_names + assert "delegate:agent_completed" not in event_names + error = next(payload for name, payload in emissions if name == "delegate:error") + assert error["agent"] == "test-agent" + assert error["session_id"] == session_id + assert error["error_type"] == "delegate_timeout" + assert error["status"] == "timed_out" + assert error["timeout_seconds"] == 0.01 + assert error["resumable"] is False + assert error["resume_status"] == "pending_child_cleanup" + assert error["tool_call_id"] == "call-resume-timeout" + assert error["parallel_group_id"] == "parallel-resume-timeout" + + +@pytest.mark.timeout(1) +@pytest.mark.asyncio +async def test_spawn_deadline_releases_parent_when_child_suppresses_cancellation(): + child_finished = asyncio.Event() + + async def suppresses_cancellation(**_kwargs): + try: + await asyncio.Future() + except asyncio.CancelledError: + child_finished.set() + return {"session_id": "ignored", "output": "ignored"} + + hooks = _hooks() + tool = _make_tool(timeout=0.02, spawn_fn=suppresses_cancellation) + started_at = asyncio.get_running_loop().time() + + result = await _spawn(tool, hooks) + elapsed = asyncio.get_running_loop().time() - started_at + + assert result.output is not None + assert result.output["status"] == "timed_out" + assert 0.01 <= elapsed < 0.12 + await asyncio.wait_for(child_finished.wait(), timeout=0.5) + + +@pytest.mark.timeout(1) +@pytest.mark.asyncio +async def test_detached_child_registry_survives_gc_and_cleans_up(): + release_child = asyncio.Event() + cancellation_suppressed = asyncio.Event() + loop_errors: list[dict] = [] + loop = asyncio.get_running_loop() + previous_handler = loop.get_exception_handler() + loop.set_exception_handler(lambda _loop, context: loop_errors.append(context)) + + async def remains_pending_after_cancellation(**_kwargs): + try: + await asyncio.Future() + except asyncio.CancelledError: + cancellation_suppressed.set() + await release_child.wait() + return {"session_id": "ignored", "output": "ignored"} + + try: + hooks = _hooks() + tool = _make_tool(timeout=0.02, spawn_fn=remains_pending_after_cancellation) + + result = await _spawn(tool, hooks) + await asyncio.wait_for(cancellation_suppressed.wait(), timeout=0.5) + + assert result.output is not None + assert result.output["status"] == "timed_out" + assert len(tool._detached_child_tasks) == 1 + + gc.collect() + await asyncio.sleep(0) + assert not any( + context.get("message") == "Task was destroyed but it is pending!" + for context in loop_errors + ) + assert len(tool._detached_child_tasks) == 1 + + release_child.set() + for _ in range(10): + if not tool._detached_child_tasks: + break + await asyncio.sleep(0) + assert not tool._detached_child_tasks + finally: + release_child.set() + await asyncio.sleep(0) + loop.set_exception_handler(previous_handler) + + +@pytest.mark.timeout(1) +@pytest.mark.asyncio +async def test_timeout_consumes_late_child_exception(caplog): + child_finished = asyncio.Event() + + async def raises_after_cancellation(**_kwargs): + try: + await asyncio.Future() + except asyncio.CancelledError: + child_finished.set() + raise RuntimeError("late child failure") + + caplog.set_level(logging.ERROR, logger="asyncio") + hooks = _hooks() + tool = _make_tool(timeout=0.02, spawn_fn=raises_after_cancellation) + + result = await _spawn(tool, hooks) + + assert result.output is not None + assert result.output["status"] == "timed_out" + await asyncio.wait_for(child_finished.wait(), timeout=0.5) + await asyncio.sleep(0) + assert "Task exception was never retrieved" not in caplog.text + + +@pytest.mark.timeout(1) +@pytest.mark.asyncio +async def test_resume_deadline_releases_parent_during_slow_cancellation_cleanup(): + cleanup_started = asyncio.Event() + cleanup_finished = asyncio.Event() + + async def slow_cancellation_cleanup(**_kwargs): + try: + await asyncio.Future() + except asyncio.CancelledError: + cleanup_started.set() + await asyncio.sleep(0.2) + cleanup_finished.set() + return {"session_id": "ignored", "output": "ignored"} + + hooks = _hooks() + tool = _make_tool(timeout=0.02, resume_fn=slow_cancellation_cleanup) + started_at = asyncio.get_running_loop().time() + + result = await _resume(tool, hooks) + elapsed = asyncio.get_running_loop().time() - started_at + + assert result.output is not None + assert result.output["status"] == "timed_out" + assert 0.01 <= elapsed < 0.12 + await asyncio.wait_for(cleanup_started.wait(), timeout=0.5) + await asyncio.wait_for(cleanup_finished.wait(), timeout=0.5) + + +@pytest.mark.timeout(1) +@pytest.mark.asyncio +async def test_external_parent_cancellation_reraises_while_cancelling_child(): + child_started = asyncio.Event() + child_cancelled = asyncio.Event() + + async def pending_child(**_kwargs): + child_started.set() + try: + await asyncio.Future() + except asyncio.CancelledError: + child_cancelled.set() + raise + + hooks = _hooks() + tool = _make_tool(timeout=1, spawn_fn=pending_child) + parent_task = asyncio.create_task(_spawn(tool, hooks)) + await asyncio.wait_for(child_started.wait(), timeout=0.5) + + parent_task.cancel() + with pytest.raises(asyncio.CancelledError): + await parent_task + + await asyncio.wait_for(child_cancelled.wait(), timeout=0.5) + event_names = [name for name, _payload in _emissions(hooks)] + assert "delegate:agent_cancelled" in event_names + assert "delegate:agent_completed" not in event_names + + +@pytest.mark.parametrize("timeout", [None, 0.01], ids=["disabled", "enabled"]) +@pytest.mark.asyncio +async def test_spawn_capability_timeout_error_uses_ordinary_error_handling(timeout): + hooks = _hooks() + tool = _make_tool( + timeout=timeout, + spawn_fn=AsyncMock(side_effect=_capability_times_out), + ) + + result = await _spawn(tool, hooks) + + expected_message = "Agent delegation failed (TimeoutError): capability timeout" + assert result.success is False + assert result.output == expected_message + assert result.error == {"message": expected_message} + + emissions = _emissions(hooks) + event_names = [name for name, _payload in emissions] + assert "delegate:error" in event_names + assert "delegate:agent_completed" not in event_names + assert "delegate:agent_cancelled" not in event_names + assert all(payload.get("status") != "timed_out" for _name, payload in emissions) + error_payload = next( + payload for name, payload in emissions if name == "delegate:error" + ) + assert error_payload["error"] == expected_message + + +@pytest.mark.parametrize("timeout", [None, 0.01], ids=["disabled", "enabled"]) +@pytest.mark.asyncio +async def test_resume_capability_timeout_error_uses_ordinary_error_handling(timeout): + hooks = _hooks() + tool = _make_tool( + timeout=timeout, + resume_fn=AsyncMock(side_effect=_capability_times_out), + ) + + result = await tool._resume_existing_session( + session_id="child-session-001_test-agent", + instruction="Continue", + hooks=hooks, + tool_call_id="call-capability-timeout", + parallel_group_id="parallel-capability-timeout", + ) + + expected_message = "Agent resume failed (TimeoutError): capability timeout" + assert result.success is False + assert result.output == expected_message + assert result.error == {"message": expected_message} + + emissions = _emissions(hooks) + event_names = [name for name, _payload in emissions] + assert "delegate:error" in event_names + assert "delegate:agent_completed" not in event_names + assert "delegate:agent_cancelled" not in event_names + assert all(payload.get("status") != "timed_out" for _name, payload in emissions) + error_payload = next( + payload for name, payload in emissions if name == "delegate:error" + ) + assert error_payload["error"] == expected_message + + +@pytest.mark.asyncio +async def test_spawn_cancellation_emits_cancelled_and_reraises(): + hooks = _hooks() + tool = _make_tool(timeout=0.01, spawn_fn=AsyncMock(side_effect=_is_cancelled)) + + with pytest.raises(asyncio.CancelledError): + await _spawn(tool, hooks) + + event_names = [name for name, _payload in _emissions(hooks)] + assert "delegate:agent_cancelled" in event_names + assert "delegate:agent_completed" not in event_names + + +@pytest.mark.asyncio +async def test_resume_cancellation_emits_cancelled_and_reraises(): + hooks = _hooks() + tool = _make_tool(timeout=0.01, resume_fn=AsyncMock(side_effect=_is_cancelled)) + + with pytest.raises(asyncio.CancelledError): + await tool._resume_existing_session( + session_id="child-session-001_test-agent", + instruction="Continue", + hooks=hooks, + tool_call_id="call-cancelled", + parallel_group_id="parallel-cancelled", + ) + + event_names = [name for name, _payload in _emissions(hooks)] + assert "delegate:agent_cancelled" in event_names + assert "delegate:agent_completed" not in event_names diff --git a/pyproject.toml b/pyproject.toml index b9ee017..9ac9118 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -64,7 +64,8 @@ dev = [ ] [tool.pytest.ini_options] -testpaths = ["tests"] +testpaths = ["tests", "modules/tool-delegate/tests"] +pythonpath = ["modules/tool-delegate"] addopts = "--import-mode=importlib" asyncio_mode = "strict" asyncio_default_fixture_loop_scope = "function"