fix(delegate): bound delegated sessions by default

.gitignore

@@ -68,6 +68,7 @@ next-steps.md
 
 # Working folders
 ai_working/tmp
+.next/
 
 # Jupyter checkpoints
 .ipynb_checkpoints/

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;

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