feat(subagents): contain unexpected sub-agent crashes as bounded retries (#326)

A delegate that crashes with anything other than an expected soft
degradation (a provider ModelAPIError/FallbackExceptionGroup, a plain
ValueError from a bad tool argument) aborts the whole parent run. For an
orchestrator fanning out to flaky delegates that is too fragile: one
crash kills unrelated in-flight work.

Add opt-in SubAgent(contain_errors=True) / SubAgents(contain_errors=False)
default. When on, the crash is caught and returned to the parent as a
bounded ModelRetry instead of propagating. It stays loud: the exception
rides the retry message, it is logged, and tool_retries still bounds
consecutive crashes into an abort -- a genuine bug is never masked as
success. Orthogonal to on_failure (which only sets the message for
expected soft degradations). Cancellation, a shared UsageLimitExceeded,
pydantic-ai control-flow signals, and UserError always propagate.

pydantic_ai_harness/experimental/subagents/README.md

@@ -93,6 +93,7 @@ orchestrator = Agent(
 | `timeout_seconds` | A wall-clock budget for one delegation. When the child exceeds it, its run is cancelled and the parent gets a soft steering message instead of hanging on the child. The cancelled child's `event_stream_handler` (if any) stops receiving events without a terminal event. |
 | `max_calls` | The maximum number of delegations to this sub-agent per parent run. Once reached, further delegations return a soft budget-exhausted message without running the child. Counts are scoped to one `Agent.run` (a `run_id`) and cleared when it ends, so each parent run and each level of a nested tree budgets independently. |
 | `on_failure` | A steering message returned to the parent for any soft degradation of this delegate, in place of the built-in default. Setting it also makes child failures soft (see below). |
+| `contain_errors` | Whether an unexpected crash in this delegate is caught and returned to the parent as a bounded `ModelRetry` instead of aborting the parent run (see below). Unset inherits the `SubAgents(contain_errors=...)` default (off). |
 
 ## Failure handling
 
@@ -102,6 +103,8 @@ A sub-agent run that fails with a *soft model error* (`ModelRetry`, `UnexpectedM
 
 Hard errors propagate to stop the whole run. A `UsageLimitExceeded` from a child that has *no* per-delegate `usage_limits` (so it shares the parent's accounting) means the whole tree is out of budget and propagates; a child reaching its *own* `usage_limits` is soft, as above.
 
+An *unexpected crash* -- any other exception the child raises, such as a provider `ModelAPIError`/`FallbackExceptionGroup` or a plain `ValueError` from a bad tool argument -- propagates by default and aborts the parent run. Set `contain_errors=True` (per delegate, or as the `SubAgents` default) to catch it and return it to the parent as a bounded `ModelRetry` instead, so one delegate crash cannot kill the whole run. Containment stays loud: the exception rides the retry message (`Sub-agent '<name>' crashed: ...`), it is logged via the standard `logging` module, and `tool_retries` still bounds consecutive crashes into an abort. This is orthogonal to `on_failure` -- a contained crash always raises the loud retry, never the soft `on_failure` return, so a genuine bug is never masked as success. Cancellation, a shared `UsageLimitExceeded`, pydantic-ai control-flow signals (`CallDeferred`, `ApprovalRequired`, the `Skip*` signals), and `UserError` always propagate regardless of `contain_errors`.
+
 ## Discovery
 
 The sub-agents are listed in the system prompt via `get_instructions`, using each agent's `description` (or a `SubAgent(description=...)` override). A sub-agent with no description is listed by name alone.
@@ -191,6 +194,7 @@ SubAgents(
     event_stream_handler=None,  # forwarded to each sub-agent run to stream its events
     tool_name='delegate_task',
     tool_retries=2,        # extra delegate-tool attempts after a sub-agent error before aborting (None inherits the agent default)
+    contain_errors=False,  # default for SubAgent.contain_errors: contain an unexpected crash as a bounded retry
 )
 ```
 
@@ -203,6 +207,7 @@ SubAgent(
     timeout_seconds=None,  # per-delegation wall-clock budget
     max_calls=None,        # max delegations to this sub-agent per parent run
     on_failure=None,       # steering message for soft degradations of this delegate
+    contain_errors=None,   # contain an unexpected crash as a bounded retry; None inherits the SubAgents default
 )
 ```
 

pydantic_ai_harness/experimental/subagents/_capability.py

@@ -140,6 +140,13 @@ class SubAgents(AbstractCapability[AgentDepsT]):
     repeated flaky sub-agent does not abort the parent run on its first repeat;
     set `None` to inherit the parent agent's default tool retries instead."""
 
+    contain_errors: bool = False
+    """Default for `SubAgent.contain_errors`: whether an unexpected sub-agent crash
+    is caught and returned to the parent as a bounded `ModelRetry` instead of
+    aborting the parent run. Off by default, so a crash propagates. Any `SubAgent`
+    can override this per delegate. See `SubAgent.contain_errors` for the
+    containment contract and what always propagates regardless."""
+
     _by_name: dict[str, SubAgent[AgentDepsT]] = field(
         default_factory=dict[str, 'SubAgent[AgentDepsT]'], init=False, repr=False, compare=False
     )
@@ -273,6 +280,7 @@ class SubAgents(AbstractCapability[AgentDepsT]):
             event_stream_handler=self.event_stream_handler,
             tool_name=self.tool_name,
             tool_retries=self.tool_retries,
+            contain_errors=self.contain_errors,
             call_counts=self._call_counts,
         )
 

pydantic_ai_harness/experimental/subagents/_toolset.py

@@ -3,13 +3,24 @@
 from __future__ import annotations
 
 import asyncio
+import logging
 from collections.abc import Mapping, Sequence
 from dataclasses import dataclass
 from typing import Any, Generic
 
 from pydantic_ai.agent import AbstractAgent, EventStreamHandler
 from pydantic_ai.capabilities import AgentCapability
-from pydantic_ai.exceptions import ModelRetry, UnexpectedModelBehavior, UsageLimitExceeded
+from pydantic_ai.exceptions import (
+    ApprovalRequired,
+    CallDeferred,
+    ModelRetry,
+    SkipModelRequest,
+    SkipToolExecution,
+    SkipToolValidation,
+    UnexpectedModelBehavior,
+    UsageLimitExceeded,
+    UserError,
+)
 from pydantic_ai.tools import AgentDepsT, RunContext
 from pydantic_ai.toolsets import AbstractToolset, FunctionToolset
 
@@ -18,6 +29,23 @@ from pydantic_ai.toolsets import AbstractToolset, FunctionToolset
 from pydantic_ai.toolsets._capability_owned import CapabilityOwnedToolset
 from pydantic_ai.usage import UsageLimits
 
+logger = logging.getLogger(__name__)
+
+# Signals that must always reach the parent run, even when a delegate has
+# `contain_errors` on. Containing the first five would break the agent graph
+# (deferred/approval/skip control-flow); a `UserError` is a setup bug that no
+# retry can fix, so masking it into a retry only delays and obscures it.
+# Cancellation (`asyncio.CancelledError`, a `BaseException`) is out of `except
+# Exception`'s reach already, and a shared `UsageLimitExceeded` has its own clause.
+_ALWAYS_PROPAGATE: tuple[type[Exception], ...] = (
+    CallDeferred,
+    ApprovalRequired,
+    SkipModelRequest,
+    SkipToolValidation,
+    SkipToolExecution,
+    UserError,
+)
+
 
 @dataclass(frozen=True)
 class SubAgent(Generic[AgentDepsT]):
@@ -68,6 +96,19 @@ class SubAgent(Generic[AgentDepsT]):
     failures soft: a child error returns this message as a normal tool result
     instead of raising a parent `ModelRetry`."""
 
+    contain_errors: bool | None = None
+    """Whether an unexpected sub-agent crash is contained instead of aborting the
+    parent run. When `True`, an exception the child raises that is not an expected
+    soft degradation (a provider `ModelAPIError`/`FallbackExceptionGroup`, a plain
+    `ValueError` from a bad tool argument, etc.) is caught and returned to the parent
+    as a bounded `ModelRetry`, so one delegate crash cannot kill the whole run. It
+    stays loud: the exception rides the retry message and is logged, and
+    `tool_retries` still bounds consecutive crashes into an abort. Cancellation, a
+    shared usage-limit, pydantic-ai control-flow signals, and `UserError` always
+    propagate regardless. Unset inherits `SubAgents.contain_errors` (default off).
+    Orthogonal to `on_failure`, which only sets the message for expected soft
+    degradations; a contained crash always raises the loud `ModelRetry`."""
+
     @property
     def resolved_name(self) -> str | None:
         """The delegate's name: `name` if set, else the agent's own `name`."""
@@ -108,6 +149,7 @@ class SubAgentToolset(FunctionToolset[AgentDepsT]):
         event_stream_handler: EventStreamHandler[AgentDepsT] | None,
         tool_name: str,
         tool_retries: int | None,
+        contain_errors: bool,
         call_counts: dict[str, dict[str, int]],
     ) -> None:
         super().__init__()
@@ -117,6 +159,7 @@ class SubAgentToolset(FunctionToolset[AgentDepsT]):
         self._shared_capabilities = list(shared_capabilities)
         self._event_stream_handler = event_stream_handler
         self._tool_name = tool_name
+        self._contain_errors = contain_errors
         # Run-scoped delegation counts, keyed by run_id then sub-agent name.
         # Shared with the capability, which clears each run's entry in wrap_run.
         self._call_counts = call_counts
@@ -233,6 +276,20 @@ class SubAgentToolset(FunctionToolset[AgentDepsT]):
                 return sub_agent.on_failure
             # Soft sub-agent failures come back to the parent as a retry it can react to.
             raise ModelRetry(f'Sub-agent {agent_name!r} failed: {exc}') from exc
+        except _ALWAYS_PROPAGATE:
+            raise
+        except Exception as exc:
+            contain = sub_agent.contain_errors if sub_agent.contain_errors is not None else self._contain_errors
+            if not contain:
+                raise
+            # Contain the crash so it cannot abort the parent, but keep it loud: the
+            # exception rides the retry message and is logged, and `tool_retries`
+            # bounds consecutive crashes into an abort.
+            logger.warning('Contained crash from sub-agent %r', agent_name, exc_info=exc)
+            raise ModelRetry(
+                f'Sub-agent {agent_name!r} crashed: {type(exc).__name__}: {exc}. '
+                f'Treat this as a recoverable failure and decide from existing evidence.'
+            ) from exc
         return str(result.output)
 
     @staticmethod

tests/experimental/subagents/test_subagents.py

@@ -10,7 +10,7 @@ from typing import Any
 import pytest
 from pydantic_ai import Agent
 from pydantic_ai.capabilities import AbstractCapability
-from pydantic_ai.exceptions import UnexpectedModelBehavior, UsageLimitExceeded
+from pydantic_ai.exceptions import ModelAPIError, UnexpectedModelBehavior, UsageLimitExceeded, UserError
 from pydantic_ai.messages import (
     AgentStreamEvent,
     ModelMessage,
@@ -97,6 +97,25 @@ def _delegate_n_then_finish(agent_name: str, n: int) -> FunctionModel:
     return FunctionModel(model_fn)
 
 
+def _delegate_two_then_finish(first: str, second: str) -> FunctionModel:
+    """A parent model that delegates to `first`, then `second`, then replies with text."""
+    calls = {'n': 0}
+
+    def model_fn(messages: list[ModelMessage], info: AgentInfo) -> ModelResponse:
+        calls['n'] += 1
+        if calls['n'] == 1:
+            return ModelResponse(
+                parts=[ToolCallPart('delegate_task', {'agent_name': first, 'task': 't'}, tool_call_id='c1')]
+            )
+        if calls['n'] == 2:
+            return ModelResponse(
+                parts=[ToolCallPart('delegate_task', {'agent_name': second, 'task': 't'}, tool_call_id='c2')]
+            )
+        return ModelResponse(parts=[TextPart('all done')])
+
+    return FunctionModel(model_fn)
+
+
 def _delegate_returns(result: Any) -> list[str]:
     """The `delegate_task` tool-return contents from a run result, in order."""
     return [
@@ -308,6 +327,7 @@ class TestDelegation:
             event_stream_handler=None,
             tool_name='delegate_task',
             tool_retries=1,
+            contain_errors=False,
             call_counts={},
         )
         parent: Agent[object, str] = Agent(_delegate_then_finish('worker'), toolsets=[toolset])
@@ -655,6 +675,97 @@ class TestRunControls:
         assert any("Delegate budget for 'worker' is exhausted" in r for r in returns)
 
 
+def _crash(message: str = 'provider down') -> FunctionModel:
+    """A sub-agent model that raises a `ModelAPIError` (an unexpected crash, not a soft failure)."""
+
+    def boom(messages: list[ModelMessage], info: AgentInfo) -> ModelResponse:
+        raise ModelAPIError('m', message)
+
+    return FunctionModel(boom)
+
+
+def _delegate_retries(result: Any) -> list[str]:
+    """The `delegate_task` retry-prompt contents from a run result, in order."""
+    return [
+        str(part.content)
+        for message in result.all_messages()
+        for part in message.parts
+        if isinstance(part, RetryPromptPart) and part.tool_name == 'delegate_task'
+    ]
+
+
+class TestContainErrors:
+    """`contain_errors`: an unexpected sub-agent crash becomes a bounded retry instead of aborting the parent."""
+
+    async def test_crash_propagates_by_default(self) -> None:
+        boomer = Agent(_crash(), name='boomer')
+        parent: Agent[object, str] = Agent(
+            _delegate_then_finish('boomer'),
+            capabilities=[SubAgents(agents=[SubAgent(boomer)])],
+        )
+        with pytest.raises(ModelAPIError):
+            await parent.run('go')
+
+    async def test_contained_crash_becomes_model_retry(self) -> None:
+        boomer = Agent(_crash(), name='boomer')
+        worker = Agent(TestModel(custom_output_text='OK'), name='worker')
+        parent: Agent[object, str] = Agent(
+            _delegate_two_then_finish('boomer', 'worker'),
+            capabilities=[SubAgents(agents=[SubAgent(boomer, contain_errors=True), SubAgent(worker)])],
+        )
+        result = await parent.run('go')
+        assert result.output == 'all done'
+        retries = _delegate_retries(result)
+        assert any('crashed: ModelAPIError' in r and 'provider down' in r for r in retries)
+
+    async def test_capability_default_contains_unset_delegate(self) -> None:
+        boomer = Agent(_crash(), name='boomer')
+        worker = Agent(TestModel(custom_output_text='OK'), name='worker')
+        parent: Agent[object, str] = Agent(
+            _delegate_two_then_finish('boomer', 'worker'),
+            capabilities=[SubAgents(agents=[SubAgent(boomer), SubAgent(worker)], contain_errors=True)],
+        )
+        result = await parent.run('go')
+        assert result.output == 'all done'
+        assert any('crashed' in r for r in _delegate_retries(result))
+
+    async def test_delegate_override_beats_capability_default(self) -> None:
+        boomer = Agent(_crash(), name='boomer')
+        parent: Agent[object, str] = Agent(
+            _delegate_then_finish('boomer'),
+            capabilities=[SubAgents(agents=[SubAgent(boomer, contain_errors=False)], contain_errors=True)],
+        )
+        with pytest.raises(ModelAPIError):
+            await parent.run('go')
+
+    async def test_user_error_always_propagates(self) -> None:
+        def boom(messages: list[ModelMessage], info: AgentInfo) -> ModelResponse:
+            raise UserError('bad setup')
+
+        boomer = Agent(FunctionModel(boom), name='boomer')
+        parent: Agent[object, str] = Agent(
+            _delegate_then_finish('boomer'),
+            capabilities=[SubAgents(agents=[SubAgent(boomer, contain_errors=True)])],
+        )
+        # A setup bug must reach the developer even under containment, not become a retry.
+        with pytest.raises(UserError):
+            await parent.run('go')
+
+    async def test_on_failure_does_not_soften_contained_crash(self) -> None:
+        boomer = Agent(_crash(), name='boomer')
+        worker = Agent(TestModel(custom_output_text='OK'), name='worker')
+        parent: Agent[object, str] = Agent(
+            _delegate_two_then_finish('boomer', 'worker'),
+            capabilities=[
+                SubAgents(agents=[SubAgent(boomer, contain_errors=True, on_failure='soft note'), SubAgent(worker)])
+            ],
+        )
+        result = await parent.run('go')
+        # A crash stays loud: it is a retry, and on_failure's soft message never becomes a delegate return.
+        assert any('crashed' in r for r in _delegate_retries(result))
+        assert 'soft note' not in _delegate_returns(result)
+
+
 class TestNameValidation:
     def test_duplicate_name_raises(self) -> None:
         first = Agent(TestModel(), name='dup')