from __future__ import annotations as _annotations

import asyncio
import contextvars
import dataclasses
import functools
import inspect
import warnings
from collections.abc import (
    AsyncGenerator,
    AsyncIterable,
    AsyncIterator,
    Awaitable,
    Callable,
    Collection,
    Generator,
    Sequence,
)
from contextlib import (
    AbstractAsyncContextManager,
    AsyncExitStack,
    asynccontextmanager,
    contextmanager,
)
from contextvars import ContextVar
from copy import copy
from dataclasses import replace
from pathlib import Path
from typing import TYPE_CHECKING, Any, ClassVar, Generic, Literal, NamedTuple, cast, overload
from uuid import uuid4

import anyio
from opentelemetry.trace import NoOpTracer
from pydantic.alias_generators import to_snake
from pydantic.json_schema import GenerateJsonSchema
from typing_extensions import Self, TypeIs, TypeVar

from pydantic_ai._instrumentation import DEFAULT_INSTRUMENTATION_VERSION
from pydantic_ai._spec import load_from_registry
from pydantic_ai.capabilities._deferred_capability_loader import DeferredCapabilityLoader

from .. import (
    _agent_graph,
    _display,
    _enqueue,
    _instructions,
    _output,
    _system_prompt,
    _utils,
    concurrency as _concurrency,
    exceptions,
    messages as _messages,
    models,
    usage as _usage,
)
from .._agent_graph import (
    CallToolsNode,
    EndStrategy,
    ModelRequestNode,
    UserPromptNode,
    build_run_context,
    capture_run_messages,
)
from .._cancel import CancellationToken, RunBinding, RunCancellation, take_run_binding
from .._deferred_capabilities import registered_loaded_capability_ids
from .._instructions import AgentInstructions
from .._output import OutputToolset
from .._run_context import dispatch_event_stream, set_current_run_context
from .._template import validate_from_spec_args
from .._warnings import PydanticAIDeprecationWarning
from ..capabilities import (
    AbstractCapability,
    AgentCapability,
    AgentModel,
    CombinedCapability,
    ModelSelection,
    ModelSelector,
    ToolSearch as ToolSearchCap,
)
from ..capabilities._dynamic import wrap_capability_funcs
from ..capabilities._ordering import has_capability_type
from ..capabilities._pending_messages import PendingMessageDrainCapability
from ..capabilities.abstract import (
    _combine_duplicate_capabilities,  # pyright: ignore[reportPrivateUsage]
    _declares_default_id,  # pyright: ignore[reportPrivateUsage]
    _reject_class_crossing_id,  # pyright: ignore[reportPrivateUsage]
    _repeated_id_message,  # pyright: ignore[reportPrivateUsage]
    leaf_capabilities,
)
from ..capabilities.combined import bind_capabilities_tier
from ..capabilities.hooks import EventT, Hooks, OnEventHookFunc
from ..capabilities.instrumentation import Instrumentation as InstrumentationCap
from ..models.instrumented import InstrumentationSettings, InstrumentedModel
from ..native_tools import AbstractNativeTool
from ..native_tools._tool_search import ToolSearchTool
from ..output import OutputDataT, OutputSpec, StructuredDict
from ..run import AgentRun, AgentRunResult
from ..settings import ModelSettings, merge_model_settings
from ..template import TemplateStr
from ..tool_manager import ParallelExecutionMode, ToolManager
from ..tools import (
    AgentDepsT,
    AgentNativeTool,
    ArgsValidatorFunc,
    DeferredToolResults,
    DocstringFormat,
    GenerateToolJsonSchema,
    NativeToolFunc,
    RunContext,
    SystemPromptFunc,
    Tool,
    ToolDefinition,
    ToolDenied,
    ToolFuncContext,
    ToolFuncEither,
    ToolFuncPlain,
    ToolParams,
    ToolPrepareFunc,
    ToolsPrepareFunc,
)
from ..toolsets import AbstractToolset, AgentToolset
from ..toolsets._dynamic import (
    DynamicToolset,
    ToolsetFunc,
)
from ..toolsets._instruction_collection import collect_toolset_instructions
from ..toolsets._tool_search import parse_discovered_tools
from ..toolsets.abstract import AGENT_TOOLSET_ID
from ..toolsets.combined import CombinedToolset
from ..toolsets.function import FunctionToolset
from ..toolsets.prepared import PreparedToolset
from .abstract import (
    AbstractAgent,
    AgentMetadata,
    AgentModelSettings,
    AgentRealtime,
    AgentRetries,
    AgentRunEvents,
    EventStreamHandler,
    EventStreamProcessor,
    RunOutputDataT,
    _RealtimeSessionLifecycle,  # pyright: ignore[reportPrivateUsage]
    _RealtimeSessionResolution,  # pyright: ignore[reportPrivateUsage]
)
from .spec import AgentSpec, get_capability_registry
from .wrapper import WrapperAgent

if TYPE_CHECKING:
    from starlette.applications import Starlette

    from pydantic_graph import Graph, GraphRunContext

    from .. import result as _result
    from ..realtime import (
        AudioRetention,
        KnownRealtimeModelName,
        RealtimeEvent as RealtimeEvent,
        RealtimeModel,
        RealtimeModelSettings,
        RealtimeProviderSession,
        RealtimeSession,
    )
    from ..ui._web import ModelsParam

__all__ = (
    'AbstractAgent',
    'Agent',
    'AgentModelSettings',
    'AgentRealtime',
    'AgentRetries',
    'AgentRun',
    'AgentRunEvents',
    'AgentRunResult',
    'NativeToolFunc',
    'CallToolsNode',
    'EndStrategy',
    'EventStreamHandler',
    'EventStreamProcessor',
    'InstrumentationSettings',
    'ModelRequestNode',
    'ParallelExecutionMode',
    'UserPromptNode',
    'WrapperAgent',
    'capture_run_messages',
    'PydanticAIDeprecationWarning',
    'ToolsPrepareFunc',
    'ToolDenied',
    'RealtimeEvent',
)


@dataclasses.dataclass(frozen=True)
class _ResolvedAgentRetries:
    """Fully resolved retry budgets used internally."""

    tools: int
    output: int


@dataclasses.dataclass(frozen=True)
class _RunLifecycle:
    short_circuited: bool


@asynccontextmanager
async def _run_lifecycle_hooks(  # noqa: C901
    run_capability: AbstractCapability[Any],
    run_ctx: RunContext[Any],
    *,
    build_result: Callable[[], AgentRunResult[Any]],
    finalize: Callable[[AgentRunResult[Any]], Awaitable[None]],
    extract_error: Callable[[BaseException], BaseException] | None = None,
    result_ready: Callable[[], bool] | None = None,
    restore_context_on: AsyncExitStack | None = None,
) -> AsyncGenerator[_RunLifecycle]:
    """Dispatch run hooks around a caller-owned run body.

    `restore_context_on` hands ownership of the propagated-ContextVar restore to the caller's exit
    stack: registered before the caller pushes later entries (like its toolset), so restore happens
    in LIFO order after those exit — resources entered inside the propagated context (e.g. the run
    span set by `Instrumentation.wrap_run`) also *exit* inside it. Without it, this context manager
    restores on its own exit.
    """
    # wrap_run cooperative hand-off protocol:
    #
    # 1. _do_run() calls before_run, sets _run_ready, then awaits _run_done.
    # 2. wrap_run wraps _do_run via the capability middleware chain.
    # 3. We await either _run_ready (handler started) or _wrap_task completion
    #    (short-circuit: wrap_run returned without calling handler).
    # 4. We yield to the caller to execute the run body.
    # 5. When the caller finishes (or an error occurs), we set _run_done.
    # 6. _do_run resumes: returns the result (success) or re-raises the error.
    # 7. If wrap_run catches the error and returns a recovery result, we use it.
    #    Otherwise the original error propagates.
    _run_ready = asyncio.Event()
    _run_done = asyncio.Event()
    _run_error: BaseException | None = None
    _wrap_context: list[tuple[ContextVar[Any], Any]] | None = None

    async def _do_run() -> AgentRunResult[Any]:
        nonlocal _wrap_context
        with set_current_run_context(run_ctx):
            await run_capability.before_run(run_ctx)
            current_ctx = contextvars.copy_context()
        # Capture context vars set by wrap_run/before_run so they can be propagated to the
        # caller's task, where the run body and any child tasks execute.
        _wrap_context = [
            (var, current_ctx[var])
            for var in current_ctx
            if var not in outer_context or outer_context[var] is not current_ctx[var]
        ]
        _run_ready.set()
        await _run_done.wait()
        if _run_error is not None:
            raise extract_error(_run_error) if extract_error is not None else _run_error
        if result_ready is not None and not result_ready():  # pragma: no cover
            # The caller finished without a result (e.g. `break` out of iteration): there is
            # nothing to return, so park until the wrap task is cancelled below. Normally the
            # cancellation is delivered at this task's resume point before this line runs, so
            # it's only reached if a `wrap_run` implementation absorbed the cancellation.
            await asyncio.Future[AgentRunResult[Any]]()
        return build_result()

    # Before `wrap_run`, not inside the handler it wraps: a `wrap_run` implementation may call a
    # durable operation before it awaits the handler, and one that short-circuits never awaits it at
    # all, so dispatch has to be installed by the time the chain is entered.
    run_capabilities_by_id = {
        capability.id: capability for capability in leaf_capabilities(run_capability) if capability.id is not None
    }
    # Mutated in place where the run already shares one mapping by reference with every `RunContext`
    # it builds (see `GraphAgentDeps.run_capabilities_by_id`); a realtime session has no graph to
    # share one, so it gets this mapping directly.
    if (existing := run_ctx._run_capabilities_by_id) is None:  # pyright: ignore[reportPrivateUsage]
        run_ctx._run_capabilities_by_id = run_capabilities_by_id  # pyright: ignore[reportPrivateUsage]
    else:
        existing.clear()
        existing.update(run_capabilities_by_id)
    run_capability._prepare_run_context(run_ctx)  # pyright: ignore[reportPrivateUsage]

    outer_context = contextvars.copy_context()
    _wrap_task = asyncio.create_task(run_capability.wrap_run(run_ctx, handler=_do_run))
    # Wait for handler to start or wrap_run to complete (short-circuit).
    _ready_waiter = asyncio.create_task(_run_ready.wait())
    try:
        await asyncio.wait({_ready_waiter, _wrap_task}, return_when=asyncio.FIRST_COMPLETED)
    except BaseException as exc:
        # Unblock `_do_run` before draining, mirroring the streaming handoff: if
        # `before_run`'s durable step absorbed the CancelledError (e.g. Temporal's
        # cooperative cancellation) and returned, `_do_run` is parked on
        # `_run_done.wait()`. Set `_run_error` so the survivor re-raises this error
        # instead of asserting on a not-yet-produced result, then set `_run_done` so
        # it can exit and `cancel_and_drain`'s gather can complete (it discards the
        # survivor's exception). Harmless no-op when `_wrap_task` really died
        # cancelled — it's already unwinding. See https://github.com/pydantic/pydantic-ai/issues/6422.
        _run_error = exc
        _run_done.set()
        await _utils.cancel_and_drain(_ready_waiter, _wrap_task)
        raise
    else:
        await _utils.cancel_and_drain(_ready_waiter)

    # Propagate context vars set by wrap_run/before_run to the caller's task.
    context_tokens: list[tuple[ContextVar[Any], contextvars.Token[Any]]] = []
    # Indexing instead of tuple unpacking because pyright can't resolve types through
    # nonlocal + Optional unpacking.
    for cv_pair in _wrap_context or ():
        context_tokens.append((cv_pair[0], cv_pair[0].set(cv_pair[1])))

    def _restore_context_vars() -> None:
        for var, token in context_tokens:
            var.reset(token)

    if restore_context_on is not None:
        restore_context_on.callback(_restore_context_vars)

    async def _finalize_result(result: AgentRunResult[Any]) -> None:
        nonlocal _run_error
        result = await run_capability.after_run(run_ctx, result=result)
        # Every completion path funnels through here — including `wrap_run`/`on_run_error`
        # recovering from the very `CancelledError` an external cancel delivered. If that
        # cancellation is still pending on this task, re-assert it rather than let the run
        # finalize as a success.
        _utils.raise_if_cancelling()
        await finalize(result)
        _run_error = None

    short_circuited = _wrap_task.done() and not _run_ready.is_set()
    if short_circuited:
        await _finalize_result(_wrap_task.result())

    try:
        try:
            yield _RunLifecycle(short_circuited=short_circuited)
        except BaseException as exc:
            _run_error = extract_error(exc) if extract_error is not None else exc
            # Don't attempt recovery for GeneratorExit/KeyboardInterrupt — awaiting
            # `_wrap_task` during cleanup could delay shutdown.
            if isinstance(_run_error, (GeneratorExit, KeyboardInterrupt)):
                raise
            # Don't re-raise yet — give wrap_run a chance to recover. If wrap_run catches
            # the error from handler() and returns a recovery result, it is suppressed.
        finally:
            if not short_circuited:
                _run_done.set()
                if _run_error is None and (result_ready is None or result_ready()):
                    await _finalize_result(await _wrap_task)
                elif _run_error is not None:
                    # Error path: await wrap_run to see if it recovers. `_do_run()` re-raises
                    # `_run_error`; if wrap_run catches it and returns a result, recovery succeeds.
                    try:
                        await _finalize_result(await _wrap_task)
                    except BaseException as wrap_exc:
                        # Attach wrap_run's own errors as context so they're visible in tracebacks
                        # (but don't mask the original). Skip CancelledError: it's expected
                        # cancellation propagation, and setting __context__ on it causes hangs on
                        # Python 3.10.
                        if not isinstance(wrap_exc, asyncio.CancelledError) and wrap_exc is not _run_error:
                            # Only fires for bugs in `wrap_run` implementations.
                            _run_error.__context__ = wrap_exc  # pragma: no cover
                # `_run_done.set()` can't complete `_wrap_task` synchronously, so the task is
                # always still pending here.
                elif not _wrap_task.done():  # pragma: no branch
                    _wrap_task.cancel()
                    try:
                        await _wrap_task
                    except (asyncio.CancelledError, BaseException):
                        pass

        # If wrap_run didn't recover, give on_run_error a chance.
        if _run_error is not None:
            try:
                result = await run_capability.on_run_error(run_ctx, error=_run_error)
            except BaseException as on_error_exc:
                _run_error = on_error_exc
            else:
                await _finalize_result(result)

        # If on_run_error didn't recover either, re-raise. In an @asynccontextmanager,
        # not re-raising suppresses the exception.
        if _run_error is not None:
            raise _run_error
    finally:
        if restore_context_on is None:
            _restore_context_vars()


def _is_model(value: object) -> TypeIs[models.Model[Any]]:
    """Narrow a value to a concrete model without losing its client type to `Unknown`."""
    return isinstance(value, models.Model)


def _agent_instruction_source(instruction: _instructions.AgentInstruction[Any]) -> _messages.InstructionSource | None:
    """The source to attribute one of the agent's configured instructions to.

    Only the literal instructions the agent was built with are addressed by the agent's own key: an
    instruction function becomes addressable when `@agent.instructions(name=...)` names it, so a bare
    callable speaks for nobody and stays unidentified.
    """
    return _messages.AgentInstructionSource() if isinstance(instruction, (str, _messages.InstructionPart)) else None


def _normalize_agent_retries(retries: AgentRetries, *, default: int = 1) -> _ResolvedAgentRetries:
    """Resolve normalized retry overrides into concrete retry budgets.

    Missing keys in an `AgentRetries` dict fall back to `default`, so internal code can work with a
    single concrete shape.
    """
    return _ResolvedAgentRetries(tools=retries.get('tools', default), output=retries.get('output', default))


def _normalize_agent_retry_overrides(retries: int | AgentRetries | None) -> AgentRetries:
    """Normalize retry input without filling missing keys.

    Used while merging layered configuration. A bare `int` sets both the `tools` and `output`
    budgets — the same shorthand at every call site (construction, run, override, spec).
    """
    if retries is None:
        return {}
    if isinstance(retries, int):
        return {'tools': retries, 'output': retries}
    return retries.copy()


T = TypeVar('T')
S = TypeVar('S')
_PreparedDepsT = TypeVar('_PreparedDepsT')
_PreparedOutputT = TypeVar('_PreparedOutputT')
NoneType = type(None)


@dataclasses.dataclass
class _ResolvedSpec:
    """Result of resolving an AgentSpec for use at run/override time."""

    capability: CombinedCapability[Any] | None
    instructions: list[_instructions.AgentInstruction[Any]]
    model: str | None
    model_settings: ModelSettings | None
    metadata: dict[str, Any] | None
    name: str | None
    output_retries: int | None
    tool_retries: int | None


@dataclasses.dataclass(init=False)
class Agent(AbstractAgent[AgentDepsT, OutputDataT]):
    """Class for defining "agents" - a way to have a specific type of "conversation" with an LLM.

    Agents are generic in the dependency type they take [`AgentDepsT`][pydantic_ai.tools.AgentDepsT]
    and the output type they return, [`OutputDataT`][pydantic_ai.output.OutputDataT].

    By default, if neither generic parameter is customised, agents have type `Agent[object, str]`.

    Minimal usage example:

    ```python
    from pydantic_ai import Agent

    agent = Agent('openai:gpt-5.2')
    result = agent.run_sync('What is the capital of France?')
    print(result.output)
    #> The capital of France is Paris.
    ```
    """

    _model: models.Model | models.KnownModelName | str | None

    _name: str | None
    _description: TemplateStr[AgentDepsT] | str | None
    end_strategy: EndStrategy
    """The strategy for handling function tool calls the model requests alongside a result that ends the run.

    That result usually comes from an output tool call, but with `NativeOutput`, `PromptedOutput`, or image
    output it comes from the structured text or image the model returns in the same response. Plain,
    unstructured text (`str` or `TextOutput`) is not treated as such a result: since the model isn't told
    its text is final, `end_strategy` never skips tools on its account, even under `'early'`.

    Defaults to `'graceful'`. See [`EndStrategy`][pydantic_ai.agent.EndStrategy] for the behavior of
    each strategy.
    """

    model_settings: AgentModelSettings[AgentDepsT] | None
    """Optional model request settings to use for this agent's runs, by default.

    Can be a static `ModelSettings` dict or a callable that takes a
    [`RunContext`][pydantic_ai.tools.RunContext] and returns `ModelSettings`.
    Callables are called before each model request, allowing dynamic per-step settings.

    Note, if `model_settings` is also provided at run time, those settings will be merged
    on top of the agent-level settings, with the run-level argument taking priority.
    """

    _output_type: OutputSpec[OutputDataT]

    _instrument: InstrumentationSettings | bool | None
    """Backing store for the `instrument` attribute. Read internally by
    `_resolve_instrumentation_settings()`; the public `agent.instrument` property reads/writes
    this field directly."""

    _instrument_default: ClassVar[InstrumentationSettings | bool] = False
    _metadata: AgentMetadata[AgentDepsT] | None = dataclasses.field(repr=False)

    _deps_type: type[AgentDepsT] = dataclasses.field(repr=False)
    _output_schema: _output.OutputSchema[OutputDataT] = dataclasses.field(repr=False)
    _output_validators: list[_output.OutputValidator[AgentDepsT, OutputDataT]] = dataclasses.field(repr=False)
    _instructions: list[_instructions.SourcedInstruction[AgentDepsT]] = dataclasses.field(repr=False)
    _system_prompts: tuple[str, ...] = dataclasses.field(repr=False)
    _system_prompt_functions: list[_system_prompt.SystemPromptRunner[AgentDepsT]] = dataclasses.field(repr=False)
    _system_prompt_dynamic_functions: dict[str, _system_prompt.SystemPromptRunner[AgentDepsT]] = dataclasses.field(
        repr=False
    )
    _function_toolset: FunctionToolset[AgentDepsT] = dataclasses.field(repr=False)
    _output_toolset: OutputToolset[AgentDepsT] | None = dataclasses.field(repr=False)
    _user_toolsets: list[AbstractToolset[AgentDepsT]] = dataclasses.field(repr=False)
    _max_output_retries: int = dataclasses.field(repr=False)
    _max_tool_retries: int = dataclasses.field(repr=False)
    _tool_timeout: float | None = dataclasses.field(repr=False)
    _validation_context: Any | Callable[[RunContext[AgentDepsT]], Any] = dataclasses.field(repr=False)

    _event_stream_handler: EventStreamHandler[AgentDepsT] | None = dataclasses.field(repr=False)

    _concurrency_limiter: _concurrency.AbstractConcurrencyLimiter | None = dataclasses.field(repr=False)

    _entered_count: int = dataclasses.field(repr=False)
    _exit_stack: AsyncExitStack | None = dataclasses.field(repr=False)

    @functools.cached_property
    def _enter_lock(self) -> anyio.Lock:
        # We use a cached_property for this because `anyio.Lock` binds to the event loop on which
        # it's first used; deferring creation until first access ensures it binds to the correct
        # running loop and avoids issues with Temporal's workflow sandbox.
        return anyio.Lock()

    # `__init__` keeps an overload pair purely so Pyright resolves a class-union `output_type`
    # (`Foo | Bar`) as `type[Foo | Bar]` rather than a bare `UnionType`; on a non-overloaded
    # signature Pyright rejects the union argument. The two overloads are intentionally
    # identical, so the second one overlaps the first.
    @overload
    def __init__(
        self,
        model: models.Model | models.KnownModelName | str | None = None,
        *,
        output_type: OutputSpec[OutputDataT] = str,
        instructions: AgentInstructions[AgentDepsT] = None,
        system_prompt: str | Sequence[str] = (),
        deps_type: type[AgentDepsT] = object,
        name: str | None = None,
        description: TemplateStr[AgentDepsT] | str | None = None,
        model_settings: AgentModelSettings[AgentDepsT] | None = None,
        retries: int | AgentRetries | None = None,
        validation_context: Any | Callable[[RunContext[AgentDepsT]], Any] = None,
        tools: Sequence[Tool[AgentDepsT] | ToolFuncEither[AgentDepsT, ...]] = (),
        toolsets: Sequence[AgentToolset[AgentDepsT]] | None = None,
        defer_model_check: bool = False,
        end_strategy: EndStrategy = 'graceful',
        metadata: AgentMetadata[AgentDepsT] | None = None,
        tool_timeout: float | None = None,
        max_concurrency: _concurrency.AnyConcurrencyLimit = None,
        capabilities: Sequence[AgentCapability[AgentDepsT]] | None = None,
    ) -> None: ...

    @overload
    def __init__(  # pyright: ignore[reportOverlappingOverload]
        self,
        model: models.Model | models.KnownModelName | str | None = None,
        *,
        output_type: OutputSpec[OutputDataT] = str,
        instructions: AgentInstructions[AgentDepsT] = None,
        system_prompt: str | Sequence[str] = (),
        deps_type: type[AgentDepsT] = object,
        name: str | None = None,
        description: TemplateStr[AgentDepsT] | str | None = None,
        model_settings: AgentModelSettings[AgentDepsT] | None = None,
        retries: int | AgentRetries | None = None,
        validation_context: Any | Callable[[RunContext[AgentDepsT]], Any] = None,
        tools: Sequence[Tool[AgentDepsT] | ToolFuncEither[AgentDepsT, ...]] = (),
        toolsets: Sequence[AgentToolset[AgentDepsT]] | None = None,
        defer_model_check: bool = False,
        end_strategy: EndStrategy = 'graceful',
        metadata: AgentMetadata[AgentDepsT] | None = None,
        tool_timeout: float | None = None,
        max_concurrency: _concurrency.AnyConcurrencyLimit = None,
        capabilities: Sequence[AgentCapability[AgentDepsT]] | None = None,
    ) -> None: ...

    def __init__(
        self,
        model: models.Model | models.KnownModelName | str | None = None,
        *,
        output_type: OutputSpec[OutputDataT] = str,
        instructions: AgentInstructions[AgentDepsT] = None,
        system_prompt: str | Sequence[str] = (),
        deps_type: type[AgentDepsT] = object,
        name: str | None = None,
        description: TemplateStr[AgentDepsT] | str | None = None,
        model_settings: AgentModelSettings[AgentDepsT] | None = None,
        retries: int | AgentRetries | None = None,
        validation_context: Any | Callable[[RunContext[AgentDepsT]], Any] = None,
        tools: Sequence[Tool[AgentDepsT] | ToolFuncEither[AgentDepsT, ...]] = (),
        toolsets: Sequence[AgentToolset[AgentDepsT]] | None = None,
        defer_model_check: bool = False,
        end_strategy: EndStrategy = 'graceful',
        metadata: AgentMetadata[AgentDepsT] | None = None,
        tool_timeout: float | None = None,
        max_concurrency: _concurrency.AnyConcurrencyLimit = None,
        capabilities: Sequence[AgentCapability[AgentDepsT]] | None = None,
    ) -> None:
        """Create an agent.

        Args:
            model: The default model to use for this agent, if not provided,
                you must provide the model when calling it. We allow `str` here since the actual list of allowed models changes frequently.
            output_type: The type of the output data, used to validate the data returned by the model,
                defaults to `str`.
            instructions: Instructions to use for this agent, you can also register instructions via a function with
                [`instructions`][pydantic_ai.agent.Agent.instructions] or pass additional, temporary, instructions when executing a run.
            system_prompt: Static system prompts to use for this agent, you can also register system
                prompts via a function with [`system_prompt`][pydantic_ai.agent.Agent.system_prompt].
            deps_type: The type used for dependency injection, this parameter exists solely to allow you to fully
                parameterize the agent, and therefore get the best out of static type checking.
                If you're not using deps, but want type checking to pass, you can set `deps=None` to satisfy Pyright
                or add a type hint `: Agent[object, <return type>]`.
            name: The name of the agent, used for logging. If `None`, we try to infer the agent name from the call frame
                when the agent is first run.
            description: A human-readable description of the agent, attached to the agent run span as
                `gen_ai.agent.description` when instrumentation is enabled.
            model_settings: Optional model request settings to use for this agent's runs, by default.
                Can be a static `ModelSettings` dict or a callable that takes a
                [`RunContext`][pydantic_ai.tools.RunContext] and returns `ModelSettings`.
                Callables are called before each model request, allowing dynamic per-step settings.
            retries: Per-category retry budgets for tools and output validation. Pass an `int` to set the same
                budget for both, or an [`AgentRetries`][pydantic_ai.AgentRetries] dict to set them
                individually (e.g. `retries={'tools': 3, 'output': 1}`). Defaults to 1 for both.
                On the text path, `output` is a global budget shared across all output-validation retries
                in a run; on the tool path it is the default per-tool `max_retries` for each output tool,
                overridable via [`ToolOutput(max_retries=...)`][pydantic_ai.output.ToolOutput.max_retries].
                Both budgets can be overridden per run via `agent.run(retries=...)` (and friends), passing
                an `AgentRetries` dict (e.g. `retries={'tools': 3}`) for per-category control.
                For model request retries, see the [transport retries](../retries.md#transport-retries) documentation.
            validation_context: Pydantic [validation context](https://docs.pydantic.dev/latest/concepts/validators/#validation-context) used to validate tool arguments and outputs.
            tools: Tools to register with the agent, you can also register tools via the decorators
                [`@agent.tool`][pydantic_ai.agent.Agent.tool] and [`@agent.tool_plain`][pydantic_ai.agent.Agent.tool_plain].
            toolsets: Toolsets to register with the agent, including MCP servers and functions which take a run context
                and return a toolset. See [`ToolsetFunc`][pydantic_ai.toolsets.ToolsetFunc] for more information.
            defer_model_check: by default, if you provide a [named][pydantic_ai.models.KnownModelName] model,
                it's evaluated to create a [`Model`][pydantic_ai.models.Model] instance immediately,
                which checks for the necessary environment variables. Set this to `True`
                to defer the evaluation until the first run. Useful if you want to
                [override the model][pydantic_ai.agent.Agent.override] for testing.
            end_strategy: Strategy for handling tool calls that are requested alongside a final result.
                See [`EndStrategy`][pydantic_ai.agent.EndStrategy] for more information.
            metadata: Optional metadata to store with each run.
                Provide a dictionary of primitives, or a callable returning one
                computed from the [`RunContext`][pydantic_ai.tools.RunContext] on each run.
                Metadata is resolved when a run starts and recomputed after a successful run finishes so it
                can reflect the final state.
                Resolved metadata can be read after the run completes via
                [`AgentRun.metadata`][pydantic_ai.agent.AgentRun],
                [`AgentRunResult.metadata`][pydantic_ai.agent.AgentRunResult], and
                [`StreamedRunResult.metadata`][pydantic_ai.result.StreamedRunResult],
                and is attached to the agent run span when instrumentation is enabled.
            tool_timeout: Default timeout in seconds for tool execution. If a tool takes longer than this,
                the tool is considered to have failed and a retry prompt is returned to the model (counting towards the retry limit).
                Individual tools can override this with their own timeout. Defaults to None (no timeout).
            max_concurrency: Optional limit on concurrent agent runs. Can be an integer for simple limiting,
                a [`ConcurrencyLimit`][pydantic_ai.ConcurrencyLimit] for advanced configuration with backpressure,
                a [`ConcurrencyLimiter`][pydantic_ai.ConcurrencyLimiter] for sharing limits across
                multiple agents, or None (default) for no limiting. When the limit is reached, additional calls
                to `run()` or `iter()` will wait until a slot becomes available.
            capabilities: Optional list of [capabilities](https://pydantic.dev/docs/ai/capabilities/overview/) to configure the agent with,
                including functions which take a run context and return a capability.
                See [`CapabilityFunc`][pydantic_ai.capabilities.CapabilityFunc] for more information.
                Custom capabilities can be created by subclassing
                [`AbstractCapability`][pydantic_ai.capabilities.AbstractCapability].
        """
        self._name = name
        self._description = description
        self.end_strategy = end_strategy

        capabilities = wrap_capability_funcs(capabilities)

        _inject_auto_capabilities(capabilities)

        # Listeners registered with `@agent.on_event` live here rather than in the root capability,
        # so they survive an overridden root capability the way `@agent.tool` tools survive
        # `override(toolsets=...)`. `on_event` is the only way to reach it, so it never contributes
        # instructions, tools or model settings, and while it holds no listeners `listens_to()` is
        # False and the dispatch gate skips it entirely.
        self._event_hooks: Hooks[AgentDepsT] = Hooks()

        self._root_capability = CombinedCapability(capabilities)
        _validate_capability_ids(self._root_capability.capabilities)
        # Two capabilities the agent itself was given meet under their shared id here, before
        # anything reads what they contribute: `for_agent`, toolset extraction and native-tool
        # validation all run below, and each would otherwise see a pair that is really one --
        # native tools most visibly, since two differently configured `WebSearch` instances carry
        # the same native tool id and reading them as two makes that a conflict. Duplicates
        # *across* layers stay with run setup, where the run-level list first exists.
        combined = _combine_duplicate_capabilities(self._root_capability, [capabilities])
        # `visit_and_replace` on a container rebuilds a container, and combining keeps one
        # occurrence of every id, so it can neither change shape nor empty the tree.
        assert isinstance(combined, CombinedCapability), 'combining the agent capabilities kept a container'
        self._root_capability = combined
        _validate_instruction_source_ids([self._root_capability])

        # Keep the constructor value untouched while capabilities bind. A capability may interpret
        # model IDs itself, so eagerly inferring a string here could construct the wrong provider
        # (and perform its authentication/configuration side effects) before `for_agent()` can add
        # the appropriate resolver. Durability capabilities tolerate a raw-string model in their
        # `for_agent` and rebuild it worker-side, so they no longer need a concrete `Model` here.
        self._model = model

        self.model_settings = model_settings

        self._output_type = output_type
        self._instrument = None
        self._metadata = metadata
        self._deps_type = deps_type

        self._output_schema = _output.OutputSchema[OutputDataT].build(output_type)
        self._output_validators = []

        # The agent's own literal instructions are one addressable part; instruction functions are
        # only addressable if `@agent.instructions(name=...)` names them.
        self._instructions = [
            _instructions.sourced_instruction(instruction, _agent_instruction_source(instruction))
            for instruction in _instructions.normalize_instructions(instructions)
        ]

        self._system_prompts = (system_prompt,) if isinstance(system_prompt, str) else tuple(system_prompt)
        self._system_prompt_functions = []
        self._system_prompt_dynamic_functions = {}

        retry_overrides = _normalize_agent_retry_overrides(retries)
        resolved_retries = _normalize_agent_retries(retry_overrides)
        self._max_tool_retries = resolved_retries.tools
        self._max_output_retries = resolved_retries.output
        self._tool_timeout = tool_timeout
        if self._tool_timeout is not None and self._tool_timeout <= 0:
            raise exceptions.UserError(f'tool_timeout must be > 0, got {self._tool_timeout}')

        self._validation_context = validation_context

        self._output_toolset = self._output_schema.toolset
        if self._output_toolset and self._output_toolset.max_retries is None:
            self._output_toolset.max_retries = self._max_output_retries

        # Leave `max_retries=None` so the agent-level tool-retry default resolves per-run via
        # `ToolManager.default_max_retries` -> `RunContext.max_retries` (overridable at `run`/`iter`/`override`)
        # rather than baked onto each tool; explicit per-tool/per-toolset budgets still win through the
        # `tool.max_retries -> toolset.max_retries -> ctx.max_retries` fallback chain.
        self._function_toolset = _AgentFunctionToolset(
            tools,
            max_retries=None,
            timeout=self._tool_timeout,
            output_schema=self._output_schema,
        )

        # Agent-direct toolsets
        agent_toolsets = list(toolsets or [])
        self._dynamic_toolsets = [
            DynamicToolset[AgentDepsT](toolset_func=toolset)
            for toolset in agent_toolsets
            if not isinstance(toolset, AbstractToolset)
        ]
        self._constructor_dynamic_toolset_count = len(self._dynamic_toolsets)
        self._user_toolsets = [toolset for toolset in agent_toolsets if isinstance(toolset, AbstractToolset)]

        # Populated by durable-execution subclasses; base agents use the run-level kwarg.
        self._event_stream_handler = None

        self._concurrency_limiter = _concurrency.normalize_to_limiter(max_concurrency)

        self._override_name: ContextVar[_utils.Option[str]] = ContextVar('_override_name', default=None)
        self._override_deps: ContextVar[_utils.Option[AgentDepsT]] = ContextVar('_override_deps', default=None)
        self._override_model: ContextVar[_utils.Option[models.Model | models.KnownModelName | str]] = ContextVar(
            '_override_model', default=None
        )
        self._override_toolsets: ContextVar[_utils.Option[Sequence[AbstractToolset[AgentDepsT]]]] = ContextVar(
            '_override_toolsets', default=None
        )
        self._override_tools: ContextVar[
            _utils.Option[Sequence[Tool[AgentDepsT] | ToolFuncEither[AgentDepsT, ...]]]
        ] = ContextVar('_override_tools', default=None)
        self._override_native_tools: ContextVar[_utils.Option[Sequence[AgentNativeTool[AgentDepsT]]]] = ContextVar(
            '_override_native_tools', default=None
        )
        self._override_instructions: ContextVar[_utils.Option[list[_instructions.AgentInstruction[AgentDepsT]]]] = (
            ContextVar('_override_instructions', default=None)
        )
        self._override_metadata: ContextVar[_utils.Option[AgentMetadata[AgentDepsT]]] = ContextVar(
            '_override_metadata', default=None
        )
        self._override_model_settings: ContextVar[_utils.Option[AgentModelSettings[AgentDepsT]]] = ContextVar(
            '_override_model_settings', default=None
        )
        self._override_output_retries: ContextVar[_utils.Option[int]] = ContextVar(
            '_override_output_retries', default=None
        )
        self._override_tool_retries: ContextVar[_utils.Option[int]] = ContextVar('_override_tool_retries', default=None)
        self._override_root_capability: ContextVar[_utils.Option[CombinedCapability[AgentDepsT]]] = ContextVar(
            '_override_root_capability', default=None
        )
        self._entered_count = 0
        self._exit_stack = None
        self._entered_model_ids: set[int] = set()
        self._entered_models_by_selection: dict[tuple[int, str], models.Model] = {}

        # Initialize capability-contributed fields before binding so `for_agent` can safely
        # inspect `agent.toolsets`. Contributions from the bound capability are extracted below.
        self._cap_toolsets: list[AgentToolset[AgentDepsT]] = []
        self._cap_instructions: list[_instructions.SourcedInstruction[AgentDepsT]] = []
        self._cap_native_tools: list[AgentNativeTool[AgentDepsT]] = []
        self._cap_model_settings: AgentModelSettings[AgentDepsT] | None = None

        # Let capabilities bind to this agent (discover model, name, toolsets, etc.).
        # Binding happens in two phases: capabilities in the `innermost` ordering tier
        # (i.e. durability capabilities) wrap all of the agent's toolsets in their
        # `for_agent`, so they only bind after every other capability has bound and had
        # its contributed toolsets extracted into `self._cap_toolsets` (and thereby
        # `self.toolsets`). The flip side is that `innermost` capabilities can't
        # contribute toolsets of their own.
        self._root_capability = bind_capabilities_tier(self._root_capability, self, innermost=False)
        cap_toolset = self._root_capability.get_toolset()
        if cap_toolset is not None:
            self._cap_toolsets = [cap_toolset]
        self._root_capability = bind_capabilities_tier(self._root_capability, self, innermost=True)

        if model is not None and not defer_model_check and not self._root_capability.has_resolve_model_id:
            self._model = models.infer_model(model)

        # Validate the bound tree so a replacement returned by `for_agent` is subject to the
        # same eager ID checks as the capability originally passed to the constructor.
        static_capabilities: list[AbstractCapability[AgentDepsT]] = []
        self._root_capability.apply(static_capabilities.append)
        _validate_capability_ids(static_capabilities)
        _validate_instruction_source_ids([self._root_capability])

        # Extract capability-contributed configuration (after for_agent so caps can provide instructions etc.)
        self._cap_instructions = self._root_capability._collect_instructions()  # pyright: ignore[reportPrivateUsage]
        self._cap_native_tools = list(self._root_capability.get_native_tools())
        _validate_native_tool_ids(self._cap_native_tools, source='agent capabilities')
        self._cap_model_settings = self._root_capability.get_model_settings()

        # Constructing the combined view validates stable toolset identities at registration time,
        # before a run attempts to mint instruction ids or dispatch tools.
        CombinedToolset(self.toolsets)

    @overload
    @classmethod
    def from_spec(
        cls,
        spec: dict[str, Any] | AgentSpec,
        *,
        custom_capability_types: Sequence[type[AbstractCapability[Any]]] = (),
        model: models.Model | models.KnownModelName | str | None = None,
        output_type: OutputSpec[Any] = str,
        instructions: AgentInstructions[Any] = None,
        system_prompt: str | Sequence[str] = (),
        name: str | None = None,
        description: TemplateStr[Any] | str | None = None,
        model_settings: ModelSettings | None = None,
        retries: int | AgentRetries | None = None,
        validation_context: Any = None,
        tools: Sequence[Tool[Any] | ToolFuncEither[Any, ...]] = (),
        toolsets: Sequence[AgentToolset[Any]] | None = None,
        defer_model_check: bool = False,
        end_strategy: EndStrategy | None = None,
        metadata: AgentMetadata[Any] | None = None,
        tool_timeout: float | None = None,
        max_concurrency: _concurrency.AnyConcurrencyLimit = None,
        capabilities: Sequence[AgentCapability[Any]] | None = None,
    ) -> Agent[object, str]: ...

    @overload
    @classmethod
    def from_spec(
        cls,
        spec: dict[str, Any] | AgentSpec,
        *,
        deps_type: type[T],
        custom_capability_types: Sequence[type[AbstractCapability[Any]]] = (),
        model: models.Model | models.KnownModelName | str | None = None,
        output_type: OutputSpec[Any] = str,
        instructions: AgentInstructions[Any] = None,
        system_prompt: str | Sequence[str] = (),
        name: str | None = None,
        description: TemplateStr[Any] | str | None = None,
        model_settings: ModelSettings | None = None,
        retries: int | AgentRetries | None = None,
        validation_context: Any = None,
        tools: Sequence[Tool[Any] | ToolFuncEither[Any, ...]] = (),
        toolsets: Sequence[AgentToolset[Any]] | None = None,
        defer_model_check: bool = False,
        end_strategy: EndStrategy | None = None,
        metadata: AgentMetadata[Any] | None = None,
        tool_timeout: float | None = None,
        max_concurrency: _concurrency.AnyConcurrencyLimit = None,
        capabilities: Sequence[AgentCapability[Any]] | None = None,
    ) -> Agent[T, str]: ...

    @classmethod
    def from_spec(
        cls,
        spec: dict[str, Any] | AgentSpec,
        *,
        deps_type: type[Any] = type(None),
        custom_capability_types: Sequence[type[AbstractCapability[Any]]] = (),
        model: models.Model | models.KnownModelName | str | None = None,
        output_type: OutputSpec[Any] = str,
        instructions: AgentInstructions[Any] = None,
        system_prompt: str | Sequence[str] = (),
        name: str | None = None,
        description: TemplateStr[Any] | str | None = None,
        model_settings: ModelSettings | None = None,
        retries: int | AgentRetries | None = None,
        validation_context: Any = None,
        tools: Sequence[Tool[Any] | ToolFuncEither[Any, ...]] = (),
        toolsets: Sequence[AgentToolset[Any]] | None = None,
        defer_model_check: bool = False,
        end_strategy: EndStrategy | None = None,
        metadata: AgentMetadata[Any] | None = None,
        tool_timeout: float | None = None,
        max_concurrency: _concurrency.AnyConcurrencyLimit = None,
        capabilities: Sequence[AgentCapability[Any]] | None = None,
    ) -> Agent[Any, Any]:
        """Construct an Agent from a spec dict or `AgentSpec`.

        This allows defining agents declaratively in YAML/JSON/dict form.
        Keyword arguments supplement the spec: scalar spec fields (like `name`,
        `retries`) are used as defaults that explicit arguments override, while
        `capabilities` from both sources are merged.

        Args:
            spec: The agent specification, either a dict or an `AgentSpec` instance.
            deps_type: The type of the dependencies for the agent. When provided,
                template strings in capabilities (e.g. `"Hello {{name}}"`) are
                compiled and validated against this type.
            custom_capability_types: Additional capability classes to make available
                beyond the built-in defaults.
            model: Override the model from the spec.
            output_type: The type of the output data, defaults to `str`.
            instructions: Instructions for the agent.
            system_prompt: Static system prompts.
            name: The agent name, overrides spec `name` if provided.
            description: The agent description, overrides spec `description` if provided.
            model_settings: Model request settings.
            retries: Retry budgets for tools and output validation. Pass an `int` to set the same budget
                for both, or an [`AgentRetries`][pydantic_ai.AgentRetries] dict to set them individually.
                Overrides spec `retries` if provided.
            validation_context: Pydantic validation context for tool arguments and outputs.
            tools: Tools to register with the agent.
            toolsets: Toolsets to register with the agent.
            defer_model_check: Defer model evaluation until first run.
            end_strategy: Strategy for tool calls alongside a final result, overrides spec `end_strategy` if provided.
            metadata: Metadata to store with each run, overrides spec `metadata` if provided.
            tool_timeout: Default timeout for tool execution, overrides spec `tool_timeout` if provided.

            max_concurrency: Limit on concurrent agent runs.
            capabilities: Additional capabilities merged with those from the spec.

        Returns:
            A new Agent instance.
        """
        validated_spec, template_context = _validate_spec(spec, deps_type)

        effective_output_type: OutputSpec[Any]
        if output_type is not str:
            effective_output_type = output_type
        elif validated_spec.output_schema is not None:
            effective_output_type = StructuredDict(validated_spec.output_schema)
        else:
            effective_output_type = str

        # Merge instructions from spec and arg
        merged_instructions = _instructions.normalize_instructions(validated_spec.instructions)
        merged_instructions.extend(_instructions.normalize_instructions(instructions))

        all_capabilities: list[AgentCapability[Any]] = list(
            _capabilities_from_spec(validated_spec, custom_capability_types, template_context)
        )
        if capabilities:
            all_capabilities.extend(capabilities)

        effective_model = model or validated_spec.model

        agent = Agent(
            model=effective_model,
            output_type=effective_output_type,
            instructions=merged_instructions or None,
            system_prompt=system_prompt,
            deps_type=deps_type,
            name=name or validated_spec.name,
            description=description or validated_spec.description,
            model_settings=merge_model_settings(
                cast(ModelSettings, validated_spec.model_settings) if validated_spec.model_settings else None,
                model_settings,
            ),
            retries=_merge_retries_with_spec(retries, validated_spec),
            validation_context=validation_context,
            tools=tools,
            toolsets=toolsets,
            defer_model_check=defer_model_check,
            end_strategy=end_strategy if end_strategy is not None else validated_spec.end_strategy,
            metadata=metadata if metadata is not None else validated_spec.metadata,
            tool_timeout=tool_timeout if tool_timeout is not None else validated_spec.tool_timeout,
            max_concurrency=max_concurrency,
            capabilities=all_capabilities,
        )
        return agent

    @overload
    @classmethod
    def from_file(
        cls,
        path: Path | str,
        *,
        fmt: Literal['yaml', 'json'] | None = None,
        custom_capability_types: Sequence[type[AbstractCapability[Any]]] = (),
        model: models.Model | models.KnownModelName | str | None = None,
        output_type: OutputSpec[Any] = str,
        instructions: AgentInstructions[Any] = None,
        system_prompt: str | Sequence[str] = (),
        name: str | None = None,
        description: TemplateStr[Any] | str | None = None,
        model_settings: ModelSettings | None = None,
        retries: int | AgentRetries | None = None,
        validation_context: Any = None,
        tools: Sequence[Tool[Any] | ToolFuncEither[Any, ...]] = (),
        toolsets: Sequence[AgentToolset[Any]] | None = None,
        defer_model_check: bool = False,
        end_strategy: EndStrategy | None = None,
        metadata: AgentMetadata[Any] | None = None,
        tool_timeout: float | None = None,
        max_concurrency: _concurrency.AnyConcurrencyLimit = None,
        capabilities: Sequence[AgentCapability[Any]] | None = None,
    ) -> Agent[object, str]: ...

    @overload
    @classmethod
    def from_file(
        cls,
        path: Path | str,
        *,
        fmt: Literal['yaml', 'json'] | None = None,
        deps_type: type[T],
        custom_capability_types: Sequence[type[AbstractCapability[Any]]] = (),
        model: models.Model | models.KnownModelName | str | None = None,
        output_type: OutputSpec[Any] = str,
        instructions: AgentInstructions[Any] = None,
        system_prompt: str | Sequence[str] = (),
        name: str | None = None,
        description: TemplateStr[Any] | str | None = None,
        model_settings: ModelSettings | None = None,
        retries: int | AgentRetries | None = None,
        validation_context: Any = None,
        tools: Sequence[Tool[Any] | ToolFuncEither[Any, ...]] = (),
        toolsets: Sequence[AgentToolset[Any]] | None = None,
        defer_model_check: bool = False,
        end_strategy: EndStrategy | None = None,
        metadata: AgentMetadata[Any] | None = None,
        tool_timeout: float | None = None,
        max_concurrency: _concurrency.AnyConcurrencyLimit = None,
        capabilities: Sequence[AgentCapability[Any]] | None = None,
    ) -> Agent[T, str]: ...

    @classmethod
    def from_file(
        cls,
        path: Path | str,
        *,
        fmt: Literal['yaml', 'json'] | None = None,
        deps_type: type[Any] = type(None),
        custom_capability_types: Sequence[type[AbstractCapability[Any]]] = (),
        model: models.Model | models.KnownModelName | str | None = None,
        output_type: OutputSpec[Any] = str,
        instructions: AgentInstructions[Any] = None,
        system_prompt: str | Sequence[str] = (),
        name: str | None = None,
        description: TemplateStr[Any] | str | None = None,
        model_settings: ModelSettings | None = None,
        retries: int | AgentRetries | None = None,
        validation_context: Any = None,
        tools: Sequence[Tool[Any] | ToolFuncEither[Any, ...]] = (),
        toolsets: Sequence[AgentToolset[Any]] | None = None,
        defer_model_check: bool = False,
        end_strategy: EndStrategy | None = None,
        metadata: AgentMetadata[Any] | None = None,
        tool_timeout: float | None = None,
        max_concurrency: _concurrency.AnyConcurrencyLimit = None,
        capabilities: Sequence[AgentCapability[Any]] | None = None,
    ) -> Agent[Any, Any]:
        """Construct an Agent from a YAML or JSON spec file.

        This is a convenience method equivalent to
        `Agent.from_spec(AgentSpec.from_file(path), ...)`.

        The file format is inferred from the extension (`.yaml`/`.yml` or `.json`)
        unless overridden with the `fmt` argument.

        All other arguments are forwarded to [`from_spec`][pydantic_ai.agent.Agent.from_spec].
        """
        spec = AgentSpec.from_file(path, fmt=fmt)
        agent = cls.from_spec(
            spec,
            deps_type=deps_type,
            custom_capability_types=custom_capability_types,
            model=model,
            output_type=output_type,
            instructions=instructions,
            system_prompt=system_prompt,
            name=name,
            description=description,
            model_settings=model_settings,
            retries=retries,
            validation_context=validation_context,
            tools=tools,
            toolsets=toolsets,
            defer_model_check=defer_model_check,
            end_strategy=end_strategy,
            metadata=metadata,
            tool_timeout=tool_timeout,
            max_concurrency=max_concurrency,
            capabilities=capabilities,
        )
        return agent

    @staticmethod
    def instrument_all(instrument: InstrumentationSettings | bool = True) -> None:
        """Set the instrumentation options for all agents that don't explicitly add an `Instrumentation` capability."""
        Agent._instrument_default = instrument

    @property
    def instrument(self) -> InstrumentationSettings | bool | None:
        """Instrumentation settings applied to this agent."""
        return self._instrument

    @instrument.setter
    def instrument(self, value: InstrumentationSettings | bool | None) -> None:
        self._instrument = value

    @property
    def model(self) -> models.Model | models.KnownModelName | str | None:
        """The default model configured for this agent."""
        return self._model

    @model.setter
    def model(self, value: models.Model | models.KnownModelName | str | None) -> None:
        """Set the default model configured for this agent.

        We allow `str` here since the actual list of allowed models changes frequently.
        """
        self._model = value

    @property
    def name(self) -> str | None:
        """The name of the agent, used for logging.

        If `None`, we try to infer the agent name from the call frame when the agent is first run.
        """
        name_ = self._override_name.get()
        return name_.value if name_ else self._name

    @name.setter
    def name(self, value: str | None) -> None:
        """Set the name of the agent, used for logging."""
        self._name = value

    @property
    def description(self) -> str | None:
        """A human-readable description of the agent.

        If the description is a TemplateStr, returns the raw template source.
        The rendered description is available at runtime via OTel span attributes.
        """
        if self._description is None:
            return None
        return str(self._description)

    @description.setter
    def description(self, value: TemplateStr[AgentDepsT] | str | None) -> None:
        """Set the description of the agent."""
        self._description = value

    def render_description(self, deps: AgentDepsT = None) -> str | None:
        """Return the agent description, rendering any TemplateStr with the given deps."""
        if self._description is None:
            return None
        if isinstance(self._description, TemplateStr):
            return self._description.render(deps)
        return self._description

    @property
    def deps_type(self) -> type:
        """The type of dependencies used by the agent."""
        return self._deps_type

    @property
    def output_type(self) -> OutputSpec[OutputDataT]:
        """The type of data output by agent runs, used to validate the data returned by the model, defaults to `str`."""
        return self._output_type

    @property
    def event_stream_handler(self) -> EventStreamHandler[AgentDepsT] | None:
        """Optional handler for events from the model's streaming response and the agent's execution of tools."""
        return self._event_stream_handler

    @property
    def validation_context(self) -> Any | Callable[[RunContext[AgentDepsT]], Any]:
        """The Pydantic validation context used to validate tool arguments and outputs.

        Set this when validators need values from [`ValidationInfo.context`][pydantic.ValidationInfo.context].
        A callable can build the context from the current [`RunContext`][pydantic_ai.tools.RunContext].
        """
        return self._validation_context

    def _get_validation_context(self) -> Any | Callable[[RunContext[AgentDepsT]], Any]:
        return self._validation_context

    def __repr__(self) -> str:
        return f'{type(self).__name__}(model={self.model!r}, name={self.name!r}, end_strategy={self.end_strategy!r}, model_settings={self.model_settings!r}, output_type={self.output_type!r})'

    @overload
    def iter(
        self,
        user_prompt: str | Sequence[_messages.UserContent] | None = None,
        *,
        output_type: None = None,
        message_history: Sequence[_messages.ModelMessage] | None = None,
        deferred_tool_results: DeferredToolResults | None = None,
        conversation_id: str | None = None,
        run_id: str | None = None,
        model: models.Model | models.KnownModelName | str | None = None,
        instructions: AgentInstructions[AgentDepsT] = None,
        deps: AgentDepsT = None,
        model_settings: AgentModelSettings[AgentDepsT] | None = None,
        usage_limits: _usage.UsageLimits | None = None,
        cancellation_token: CancellationToken | None = None,
        usage: _usage.RunUsage | None = None,
        metadata: AgentMetadata[AgentDepsT] | None = None,
        retries: int | AgentRetries | None = None,
        infer_name: bool = True,
        toolsets: Sequence[AbstractToolset[AgentDepsT]] | None = None,
        capabilities: Sequence[AgentCapability[AgentDepsT]] | None = None,
        spec: dict[str, Any] | AgentSpec | None = None,
    ) -> AbstractAsyncContextManager[AgentRun[AgentDepsT, OutputDataT]]: ...

    @overload
    def iter(
        self,
        user_prompt: str | Sequence[_messages.UserContent] | None = None,
        *,
        output_type: OutputSpec[RunOutputDataT],
        message_history: Sequence[_messages.ModelMessage] | None = None,
        deferred_tool_results: DeferredToolResults | None = None,
        conversation_id: str | None = None,
        run_id: str | None = None,
        model: models.Model | models.KnownModelName | str | None = None,
        instructions: AgentInstructions[AgentDepsT] = None,
        deps: AgentDepsT = None,
        model_settings: AgentModelSettings[AgentDepsT] | None = None,
        usage_limits: _usage.UsageLimits | None = None,
        cancellation_token: CancellationToken | None = None,
        usage: _usage.RunUsage | None = None,
        metadata: AgentMetadata[AgentDepsT] | None = None,
        retries: int | AgentRetries | None = None,
        infer_name: bool = True,
        toolsets: Sequence[AbstractToolset[AgentDepsT]] | None = None,
        capabilities: Sequence[AgentCapability[AgentDepsT]] | None = None,
        spec: dict[str, Any] | AgentSpec | None = None,
    ) -> AbstractAsyncContextManager[AgentRun[AgentDepsT, RunOutputDataT]]: ...

    @asynccontextmanager
    async def iter(
        self,
        user_prompt: str | Sequence[_messages.UserContent] | None = None,
        *,
        output_type: OutputSpec[Any] | None = None,
        message_history: Sequence[_messages.ModelMessage] | None = None,
        deferred_tool_results: DeferredToolResults | None = None,
        conversation_id: str | None = None,
        run_id: str | None = None,
        model: models.Model | models.KnownModelName | str | None = None,
        instructions: AgentInstructions[AgentDepsT] = None,
        deps: AgentDepsT = None,
        model_settings: AgentModelSettings[AgentDepsT] | None = None,
        usage_limits: _usage.UsageLimits | None = None,
        cancellation_token: CancellationToken | None = None,
        usage: _usage.RunUsage | None = None,
        metadata: AgentMetadata[AgentDepsT] | None = None,
        retries: int | AgentRetries | None = None,
        infer_name: bool = True,
        toolsets: Sequence[AbstractToolset[AgentDepsT]] | None = None,
        capabilities: Sequence[AgentCapability[AgentDepsT]] | None = None,
        spec: dict[str, Any] | AgentSpec | None = None,
    ) -> AsyncGenerator[AgentRun[AgentDepsT, Any]]:
        """A contextmanager which can be used to iterate over the agent graph's nodes as they are executed.

        This method builds an internal agent graph (using system prompts, tools and output schemas) and then returns an
        `AgentRun` object. The `AgentRun` can be used to async-iterate over the nodes of the graph as they are
        executed. This is the API to use if you want to consume the outputs coming from each LLM model response, or the
        stream of events coming from the execution of tools.

        The `AgentRun` also provides methods to access the full message history, new messages, and usage statistics,
        and the final result of the run once it has completed.

        For more details, see the documentation of `AgentRun`.

        Example:
        ```python
        from pydantic_ai import Agent

        agent = Agent('openai:gpt-5.2')

        async def main():
            nodes = []
            async with agent.iter('What is the capital of France?') as agent_run:
                async for node in agent_run:
                    nodes.append(node)
            print(nodes)
            '''
            [
                UserPromptNode(
                    user_prompt='What is the capital of France?',
                    instructions_functions=[],
                    system_prompts=(),
                    system_prompt_functions=[],
                    system_prompt_dynamic_functions={},
                ),
                ModelRequestNode(
                    request=ModelRequest(
                        parts=[
                            UserPromptPart(
                                content='What is the capital of France?',
                                timestamp=datetime.datetime(...),
                            )
                        ],
                        timestamp=datetime.datetime(...),
                        run_id='...',
                        conversation_id='...',
                    )
                ),
                CallToolsNode(
                    model_response=ModelResponse(
                        parts=[TextPart(content='The capital of France is Paris.')],
                        usage=RequestUsage(
                            cost=Decimal('0.000196'), input_tokens=56, output_tokens=7
                        ),
                        model_name='gpt-5.2',
                        timestamp=datetime.datetime(...),
                        run_id='...',
                        conversation_id='...',
                    )
                ),
                End(data=FinalResult(output='The capital of France is Paris.')),
            ]
            '''
            print(agent_run.result.output)
            #> The capital of France is Paris.
        ```

        Args:
            user_prompt: User input to start/continue the conversation.
            output_type: Custom output type to use for this run, `output_type` may only be used if the agent has no
                output validators since output validators would expect an argument that matches the agent's output type.
            message_history: History of the conversation so far.
            deferred_tool_results: Optional results for deferred tool calls in the message history.
            conversation_id: ID of the conversation this run belongs to. Pass `'new'` to start a fresh conversation, ignoring any `conversation_id` already on `message_history`. If omitted, falls back to the most recent `conversation_id` on `message_history` or a freshly generated UUID7.
            run_id: Optional ID for this agent run. Unlike `conversation_id`, never inherited from `message_history`. Passing an empty string, or a value that already appears on `message_history`, raises `UserError` because both break `new_messages()`; use `conversation_id` to correlate across turns or deferred-tool resume. If omitted, a fresh UUID7 is generated.
            model: Optional model to use for this run, required if `model` was not set when creating the agent.
            instructions: Optional additional instructions to use for this run.
            deps: Optional dependencies to use for this run.
            model_settings: Optional settings to use for this model's request, or a callable
                that receives [`RunContext`][pydantic_ai.tools.RunContext] and returns settings.
                Callables are called before each model request, allowing dynamic per-step settings.
            usage_limits: Optional limits on model request count or token usage.
            cancellation_token: Token used to cancel this run from another task or thread. Single-use:
                mint a fresh token per run, as a reused (already-cancelled) token prevents the run from starting.
            usage: Optional usage to start with, useful for resuming a conversation or agents used in tools.
            metadata: Optional metadata to attach to this run. Accepts a dictionary or a callable taking
                [`RunContext`][pydantic_ai.tools.RunContext]; merged with the agent's configured metadata.
            retries: Override the agent-level retry budgets for this run. Pass an `int` to override both
                the tool-retry and output budgets, or an [`AgentRetries`][pydantic_ai.AgentRetries] dict to
                override just one (e.g. `retries={'tools': 3}`). See
                [`Agent.__init__`][pydantic_ai.agent.Agent.__init__] for semantics of the two enforcement paths.
            infer_name: Whether to try to infer the agent name from the call frame if it's not set.
            toolsets: Optional additional toolsets for this run.
            capabilities: Optional additional [capabilities](https://pydantic.dev/docs/ai/capabilities/overview/) for this run, merged with the agent's configured capabilities.
            spec: Optional agent spec to apply for this run. At run time, spec values are additive.

        Returns:
            The result of the run.
        """
        if infer_name and self.name is None:
            self._infer_name(inspect.currentframe())

        prepared = await self._prepare_run(
            user_prompt,
            output_type=output_type,
            message_history=message_history,
            deferred_tool_results=deferred_tool_results,
            conversation_id=conversation_id,
            run_id=run_id,
            model=model,
            instructions=instructions,
            deps=deps,
            model_settings=model_settings,
            usage_limits=usage_limits,
            cancellation_token=cancellation_token,
            usage=usage,
            metadata=metadata,
            retries=retries,
            toolsets=toolsets,
            capabilities=capabilities,
            spec=spec,
        )
        async with prepared.open() as agent_run:
            yield agent_run

    async def _prepare_run(  # noqa: C901
        self,
        user_prompt: str | Sequence[_messages.UserContent] | None = None,
        *,
        output_type: OutputSpec[Any] | None = None,
        message_history: Sequence[_messages.ModelMessage] | None = None,
        deferred_tool_results: DeferredToolResults | None = None,
        conversation_id: str | None = None,
        run_id: str | None = None,
        model: models.Model | models.KnownModelName | str | None = None,
        instructions: AgentInstructions[AgentDepsT] = None,
        deps: AgentDepsT = None,
        model_settings: AgentModelSettings[AgentDepsT] | None = None,
        usage_limits: _usage.UsageLimits | None = None,
        cancellation_token: CancellationToken | None = None,
        usage: _usage.RunUsage | None = None,
        metadata: AgentMetadata[AgentDepsT] | None = None,
        retries: int | AgentRetries | None = None,
        toolsets: Sequence[AbstractToolset[AgentDepsT]] | None = None,
        capabilities: Sequence[AgentCapability[AgentDepsT]] | None = None,
        spec: dict[str, Any] | AgentSpec | None = None,
    ) -> _PreparedAgentRun[AgentDepsT, Any]:
        # Consume the pending `AgentRunEvents` binding before ANY user-supplied code (capability /
        # toolset `for_run()` hooks below) runs in this context: a hook that starts a nested agent
        # run would otherwise consume it and attach the outer handle to the wrong run.
        binding = take_run_binding()

        # The controller likewise exists before any user-supplied setup code, so `RunContext.cancel()`
        # from a capability/toolset `for_run()` hook records the request instead of raising. Delivery
        # still waits for `bind()` below: setup hooks are never interrupted, and a request recorded
        # here ends the run at the first await after binding, before any model request (#7386).
        cancellation = binding.cancellation if binding is not None else RunCancellation()

        # A bare `int` overrides both budgets; a partial `retries={'tools': ...}` / `{'output': ...}`
        # dict overrides only the named budget for this run (riding `ToolManager.default_max_retries`).
        retry_overrides = _normalize_agent_retry_overrides(retries)

        # Resolve the root capability (override > agent default) up front: it's needed both for the
        # capability-supplied model fallback below and for run-time capability assembly further down.
        base_capability, base_is_override = self._base_run_capability()

        # Resolve spec contributions (additive at run time)
        resolved = self._resolve_spec(spec)

        effective_output_retries = retry_overrides.get('output')
        effective_tool_retries = retry_overrides.get('tools')
        if resolved is not None:
            # Model: spec as fallback (run param > spec > agent)
            if model is None and resolved.model is not None:
                model = resolved.model
            # Output retries: run param > spec > agent default
            if effective_output_retries is None and resolved.output_retries is not None:
                effective_output_retries = resolved.output_retries
            # Tool retries: run param > spec > agent default
            if effective_tool_retries is None and resolved.tool_retries is not None:
                effective_tool_retries = resolved.tool_retries
            # Instructions: spec instructions are additional
            if resolved.instructions:
                extra = resolved.instructions
                if instructions is not None:
                    existing = _instructions.normalize_instructions(instructions)
                    existing.extend(extra)
                    instructions = existing
                else:
                    instructions = extra
            # Model settings: merge spec settings under run settings (only static dicts)
            if resolved.model_settings is not None:
                if model_settings is None or not callable(model_settings):
                    model_settings = merge_model_settings(resolved.model_settings, model_settings)
                # If model_settings is a callable, spec model_settings are handled via the capability layer
            # Metadata: merge spec metadata under run metadata
            if resolved.metadata is not None:
                if metadata is not None:
                    if callable(metadata):
                        _spec_meta = resolved.metadata
                        _orig_metadata = metadata

                        def _merged_meta(ctx: RunContext[AgentDepsT]) -> dict[str, Any]:
                            return {**(_spec_meta or {}), **_orig_metadata(ctx)}

                        metadata = _merged_meta
                    else:
                        metadata = {**resolved.metadata, **metadata}
                else:
                    metadata = resolved.metadata

        # `override(retries=...)` wins over the run kwarg + spec, matching the precedence
        # of `model`/`deps`/`instructions`/etc. (see `Agent._get_model`). This keeps testing
        # fixtures that wrap call sites in `agent.override(retries=N)` effective even when
        # production code passes its own `run(retries=...)`.
        override_output_retries = self._override_output_retries.get()
        if override_output_retries is not None:
            effective_output_retries = override_output_retries.value
        effective_tool_retries_resolved = self._resolve_tool_retries(effective_tool_retries)

        deps = self._get_deps(deps)
        usage = usage or _usage.RunUsage()

        # Run/spec capabilities that already exist before `for_run()` can participate in
        # bootstrap model selection and ID resolution. A capability function may only
        # contribute after a bootstrap model exists because its contract requires RunContext.
        extra_capabilities: list[AbstractCapability[AgentDepsT]] = []
        if resolved is not None and resolved.capability is not None:
            extra_capabilities.append(resolved.capability)
        extra_capabilities.extend(wrap_capability_funcs(capabilities))
        extra_capabilities = self._bind_run_capabilities(extra_capabilities)
        model_layers: list[AbstractCapability[AgentDepsT]] = [base_capability, *extra_capabilities]
        bootstrap_capability: AbstractCapability[AgentDepsT]
        if len(model_layers) > 1:
            bootstrap_capability = CombinedCapability(model_layers)
        else:
            bootstrap_capability = model_layers[0]
        resolved_models_by_selection: dict[tuple[int, str], models.Model] = {}

        # Explicit run/spec/override models are authoritative. Otherwise the capability model
        # contribution selects the initial model needed to construct RunContext and resolve
        # `for_run()`; dynamic contributions are evaluated again for later request steps.
        model_is_explicit = model is not None or self._override_model.get() is not None
        model_contribution = None if model_is_explicit else bootstrap_capability.get_model()
        self._check_dynamic_model_resume(model_contribution, message_history)

        has_default_model = self._has_model(model)

        # The string the run's model was selected from, if any — carried through to
        # `ModelRequestContext.model_id` so durable-execution capabilities can round-trip
        # the original selection token (e.g. an alias only a `resolve_model_id` capability
        # can resolve) across the activity/step/task boundary.
        model_id: str | None = None
        default_model_id: str | None = None
        default_model: models.Model | None = None
        if has_default_model:
            raw_model = self._pick_raw_model(model)
            model_id = raw_model if isinstance(raw_model, str) else None
            default_model_id = model_id
            default_model = await self._resolve_model_selection(
                raw_model,
                capability=bootstrap_capability,
                deps=deps,
                resolved_models=resolved_models_by_selection,
            )
        if model_contribution is not None:
            selection_ctx = models.ModelSelectionContext(
                agent=self,
                deps=deps,
                model=default_model,
                run_step=1,
                messages=list(message_history) if message_history else [],
                usage=usage,
            )
            model_used, model_id = await self._evaluate_model_contribution(
                model_contribution,
                capability=bootstrap_capability,
                ctx=selection_ctx,
                resolved_models=resolved_models_by_selection,
            )
        elif default_model is not None:
            model_used = default_model
        else:
            raise exceptions.UserError('`model` must either be set on the agent or included when calling it.')
        del model
        output_schema = self._prepare_output_schema(output_type)

        output_type_ = output_type or self.output_type

        # We consider it a user error if a user tries to restrict the result type while having an output validator that
        # may change the result type from the restricted type to something else. Therefore, we consider the following
        # typecast reasonable, even though it is possible to violate it with otherwise-type-checked code.
        output_validators = self._output_validators

        # Resolve the effective per-output-tool default: run arg > spec > agent init default
        effective_output_toolset_max_retries = (
            effective_output_retries if effective_output_retries is not None else self._max_output_retries
        )

        output_toolset = self._output_toolset
        if output_schema != self._output_schema or output_validators:
            output_toolset = output_schema.toolset
            if output_toolset:
                # Clone before mutating max_retries when the toolset is the shared agent-level
                # instance (output_schema == self._output_schema, branch hit via output_validators);
                # when output_schema differs, output_schema.toolset is already a fresh per-run instance.
                if output_toolset is self._output_toolset and effective_output_retries is not None:
                    output_toolset = copy(output_toolset)
                if output_toolset.max_retries is None or effective_output_retries is not None:
                    output_toolset.max_retries = effective_output_toolset_max_retries
                output_toolset.output_validators = output_validators
        elif output_toolset is not None and effective_output_retries is not None:
            # Clone before mutating max_retries so concurrent runs don't race on the
            # shared agent-level toolset.
            output_toolset = copy(output_toolset)
            output_toolset.max_retries = effective_output_toolset_max_retries

        # Build the graph
        graph = _agent_graph.build_agent_graph(self.name, self._deps_type, output_type_)

        # Build the initial state
        state = _agent_graph.GraphAgentState(
            message_history=list(message_history) if message_history else [],
            usage=usage,
            output_retries_used=0,
            run_step=0,
            run_id=_agent_graph.resolve_run_id(run_id, message_history),
            conversation_id=_agent_graph.resolve_conversation_id(conversation_id, message_history),
        )

        # Build a resolver that computes model settings per-step, in order of precedence: run > agent > model
        model_settings_override = self._override_model_settings.get()
        agent_model_settings = (
            model_settings_override.value if model_settings_override is not None else self.model_settings
        )
        run_model_settings = model_settings if model_settings_override is None else None

        # Validate `tool_choice` on the static baseline. Callable layers (agent-level callable,
        # run-level callable, capability-supplied) may inject `'required'` or `list[str]` per-step
        # and are trusted to adapt across steps; static dict values would lock every step into a
        # tool call and prevent the agent from producing a final response.
        baseline_settings: ModelSettings | None = model_used.settings
        if not callable(agent_model_settings):
            baseline_settings = merge_model_settings(baseline_settings, agent_model_settings)
        if not callable(run_model_settings):
            baseline_settings = merge_model_settings(baseline_settings, run_model_settings)
        if baseline_settings:
            tool_choice = baseline_settings.get('tool_choice')
            if tool_choice == 'required' or isinstance(tool_choice, list):
                raise exceptions.UserError(
                    f'`tool_choice={tool_choice!r}` prevents the agent from producing a final response '
                    f'because output tools are excluded. Use `ToolOrOutput` to combine specific function '
                    f"tools with output capability, return a callable from a capability's "
                    f'`get_model_settings()` to vary `tool_choice` per step, or use '
                    f'`pydantic_ai.direct.model_request` for single-shot model calls.'
                )

        usage_limits = usage_limits or _usage.UsageLimits()

        # Resolve instrumentation: an explicit `InstrumentedModel` (passed by the user
        # to `Agent(model=...)`, e.g. by `logfire.instrument_pydantic_ai(model)`) wins,
        # then `Agent.instrument_all()` / `agent.instrument = ...` (read via
        # `_resolve_instrumentation_settings`). When detected, unwrap so the rest of the
        # run uses the plain model — the `Instrumentation` capability injected below
        # provides the spans.
        if isinstance(model_used, InstrumentedModel):
            instrumentation_settings: InstrumentationSettings | None = model_used.instrumentation_settings
            model_used = model_used.wrapped
        else:
            instrumentation_settings = self._resolve_instrumentation_settings()

        if instrumentation_settings is not None:
            tracer = instrumentation_settings.tracer
            instrumentation_cap: InstrumentationCap | None = InstrumentationCap(settings=instrumentation_settings)
        else:
            tracer = NoOpTracer()
            instrumentation_cap = None

        # Allocated here rather than with the graph deps below, so the context `for_run` receives
        # shares the very mappings the run fills at setup. A capability that holds on to that
        # context and later passes it to a durable operation then dispatches like any other caller,
        # instead of silently running the operation inline.
        durable_operations: dict[tuple[str, str], Callable[..., Awaitable[Any]]] = {}
        run_capabilities_by_id: dict[str, AbstractCapability[AgentDepsT]] = {}

        # Build initial RunContext for for_run lifecycle hooks. Includes every
        # field that's already known here — `tool_manager` and `validation_context`
        # are populated later by `build_run_context` once the run is iterating.
        initial_ctx = RunContext[AgentDepsT](
            deps=deps,
            agent=self,
            _durable_operations=durable_operations,
            _run_capabilities_by_id=run_capabilities_by_id,
            model=model_used,
            _model_id=model_id,
            usage=usage,
            usage_limits=usage_limits,
            prompt=user_prompt,
            messages=state.message_history,
            tracer=tracer,
            trace_include_content=instrumentation_settings is not None and instrumentation_settings.include_content,
            instrumentation_version=instrumentation_settings.version
            if instrumentation_settings
            else DEFAULT_INSTRUMENTATION_VERSION,
            run_step=0,
            pending_messages=state.pending_messages,
            run_id=state.run_id,
            conversation_id=state.conversation_id,
            _cancellation=cancellation,
        )

        # Resolve run metadata up front so capability and toolset `for_run` hooks
        # can see it on `RunContext.metadata`. Metadata factories receive the
        # `initial_ctx` above (no `tool_manager` / `validation_context` yet); they
        # will be invoked again at the end of the run with the full final state,
        # so any field that becomes available later still ends up reflected in
        # `agent_run.metadata`. Factories should be pure mappings over the run
        # context, not perform IO or have side effects.
        state.metadata = self._get_metadata(initial_ctx, metadata)
        initial_ctx.metadata = state.metadata

        # Resolve the capability layers and extract their per-run contributions. Shared with
        # `realtime_session` via `_resolve_run_capabilities` so both wire capabilities up identically;
        # this call site keeps the graph-only surroundings: the `InstrumentedModel` unwrap and
        # instrumentation-settings resolution above, the deferred loader (`inject_deferred_loader=True`),
        # the output toolset below, and the layered `get_model_settings` closure. Keep those in sync
        # with the realtime call site.
        resolved_caps = await self._resolve_run_capabilities(
            initial_ctx,
            base_capability=base_capability,
            extra_capabilities=extra_capabilities,
            instrumentation_cap=instrumentation_cap,
            inject_deferred_loader=True,
            base_is_override=base_is_override,
        )
        run_capability = resolved_caps.run_capability
        capabilities_dict = resolved_caps.capabilities
        cap_instructions = resolved_caps.instructions
        cap_native_tools = resolved_caps.native_tools
        cap_model_settings = resolved_caps.model_settings
        cap_toolsets = resolved_caps.toolsets

        # Whether any capability's `for_run` swapped a model-layer contribution during resolution; the
        # per-step model-selection block below keys off this. The model layers are the tail of the
        # resolved layers (the `Instrumentation` capability, when injected, sits at the front).
        resolved_layers = resolved_caps.resolved_layers
        model_layer_start = len(resolved_layers) - len(model_layers)
        model_layers_unchanged = all(
            resolved_layers[model_layer_start + index] is layer for index, layer in enumerate(model_layers)
        )

        # Build model settings resolver using per-run capability. Shared with `realtime_session` via
        # `_layer_model_settings` (agent -> capability -> run order; the model's own settings are the
        # base for a graph run). Resolved per model-request step here; once at connect in a session.
        def get_model_settings(run_context: RunContext[AgentDepsT]) -> ModelSettings | None:
            # A capability can select a different model per step, so the base is the step's live
            # `run_context.model` settings, not a captured initial model. A graph run always uses a
            # request-response `Model` here; realtime has its own settings path and never reaches this.
            # (Hoisted to a local first so pyright narrows cleanly after `RunContext.model` widened to
            # `AbstractModel`; member access on the narrowed attribute directly trips a false positive.)
            step_model = run_context.model
            return _layer_model_settings(
                run_context,
                (agent_model_settings, cap_model_settings, run_model_settings),
                base=step_model.settings if isinstance(step_model, models.Model) else None,
            )

        # Build toolset with per-run capability contributions
        toolset = self._get_toolset(
            output_toolset=output_toolset,
            additional_toolsets=toolsets,
            cap_toolsets=cap_toolsets,
            run_capability=run_capability,
            max_output_retries=effective_output_toolset_max_retries,
        )
        toolset = await toolset.for_run(initial_ctx)
        tool_manager = ToolManager[AgentDepsT](
            toolset, root_capability=run_capability, default_max_retries=effective_tool_retries_resolved
        )

        # Build instructions with per-run capability contributions
        sourced_instructions = self._get_instructions(
            additional_instructions=instructions,
            cap_instructions=cap_instructions,
        )

        async def get_instructions(
            run_context: RunContext[AgentDepsT],
        ) -> list[_messages.InstructionPart] | None:
            return await _instructions.resolve_sourced_instructions(sourced_instructions, run_context) or None

        # The deferred capabilities the model has already loaded in prior steps; the graph
        # refreshes this from history before each model request, so the seed only matters
        # for pre-first-step access. Non-deferred capabilities are folded in by the
        # `RunContext.active_capability_ids` property.
        loaded_capability_ids = (
            registered_loaded_capability_ids(message_history, capabilities_dict.keys())
            if message_history
            else set[str]()
        )
        discovered_tool_names = parse_discovered_tools(message_history) if message_history else set[str]()

        run_model_contribution = None if model_is_explicit else run_capability.get_model()
        self._check_dynamic_model_resume(run_model_contribution, message_history)
        model_selector: ModelSelector[AgentDepsT] | None
        model_selected_for_step: int | None
        capability_owns_current_model: bool
        if model_layers_unchanged:
            model_selector = (
                model_contribution if callable(model_contribution) and not _is_model(model_contribution) else None
            )
            model_selected_for_step = 1 if model_selector is not None else None
            capability_owns_current_model = model_contribution is not None
        elif callable(run_model_contribution) and not _is_model(run_model_contribution):
            # The bootstrap model was only needed to construct RunContext for `for_run`.
            # The replacement selector makes the authoritative step-one choice in the graph,
            # but the discarded bootstrap model still needs its lifecycle managed.
            model_selector = run_model_contribution
            model_selected_for_step = None
            capability_owns_current_model = True
        elif run_model_contribution is not None:
            model_used = await self._resolve_model_selection(
                run_model_contribution,
                capability=run_capability,
                deps=deps,
                resolved_models=resolved_models_by_selection,
            )
            model_id = run_model_contribution if isinstance(run_model_contribution, str) else None
            model_selector = None
            model_selected_for_step = None
            capability_owns_current_model = True
        elif default_model is not None:
            model_used = default_model
            # The bootstrap contribution was withdrawn in `for_run`, so provenance reverts to the run's default.
            model_id = default_model_id
            model_selector = None
            model_selected_for_step = None
            capability_owns_current_model = False
        else:
            raise exceptions.UserError(
                'A capability removed the bootstrap model in `for_run()` but the agent has no default model.'
            )

        async def evaluate_model_selector(
            selector: ModelSelector[AgentDepsT], selection_ctx: models.ModelSelectionContext[AgentDepsT]
        ) -> tuple[models.Model, str | None]:
            return await self._evaluate_model_contribution(
                selector,
                capability=run_capability,
                ctx=selection_ctx,
                resolved_models=resolved_models_by_selection,
            )

        def display_banner(*, model: str, tools: int) -> None:
            # Called by the graph once the run's first step has resolved the model it will actually
            # use and the tools it will actually offer, and only when there is a banner to show;
            # everything else is settled here and now.
            _display.display_agent_banner(
                name=self.name,
                model=model,
                # A run-level `output_type=` overrides what the agent was built with.
                output_type=output_type_,
                tools=tools,
                capabilities=_registered_capability_count(bootstrap_capability),
            )

        model_resources = _RunModelResources(self._entered_model_ids.copy())
        graph_deps = _agent_graph.GraphAgentDeps[AgentDepsT, OutputDataT](
            user_deps=deps,
            agent=self,
            prompt=user_prompt,
            new_message_index=len(message_history) if message_history else 0,
            resumed_request=None,
            resumed_request_index=None,
            model=model_used,
            model_id=model_id,
            model_selector=model_selector,
            model_selected_for_step=model_selected_for_step,
            evaluate_model_selector=evaluate_model_selector,
            enter_model=model_resources.enter_model,
            get_model_settings=get_model_settings,
            usage_limits=usage_limits,
            max_output_retries=effective_output_toolset_max_retries,
            end_strategy=self.end_strategy,
            output_schema=output_schema,
            output_validators=output_validators,
            validation_context=self._validation_context,
            root_capability=run_capability,
            capabilities=capabilities_dict,
            loaded_capability_ids=loaded_capability_ids,
            discovered_tool_names=discovered_tool_names,
            durable_operations=durable_operations,
            run_capabilities_by_id=run_capabilities_by_id,
            native_tools=cap_native_tools,
            tool_manager=tool_manager,
            display_banner=display_banner,
            tracer=tracer,
            get_instructions=get_instructions,
            instrumentation_settings=instrumentation_settings,
            cancellation=cancellation,
        )

        user_prompt_node = _agent_graph.UserPromptNode[AgentDepsT](
            user_prompt=user_prompt,
            deferred_tool_results=deferred_tool_results,
            instructions=None,
            instructions_functions=[],
            system_prompts=self._system_prompts,
            system_prompt_functions=self._system_prompt_functions,
            system_prompt_dynamic_functions=self._system_prompt_dynamic_functions,
        )

        return _PreparedAgentRun[AgentDepsT, Any](
            graph=graph,
            state=state,
            graph_deps=graph_deps,
            user_prompt_node=user_prompt_node,
            agent_name=self.name or 'agent',
            binding=binding,
            cancellation_token=cancellation_token,
            model=model_used,
            capability_owns_current_model=capability_owns_current_model,
            model_resources=model_resources,
            run_capability=run_capability,
            toolset=toolset,
            usage_limits=usage_limits,
            concurrency_limiter=self._concurrency_limiter,
            resolve_metadata=functools.partial(self._resolve_and_store_metadata, metadata=metadata),
        )

    def _get_metadata(
        self,
        ctx: RunContext[AgentDepsT],
        additional_metadata: AgentMetadata[AgentDepsT] | None = None,
    ) -> dict[str, Any] | None:
        metadata_override = self._override_metadata.get()
        if metadata_override is not None:
            return self._resolve_metadata_config(metadata_override.value, ctx)

        base_metadata = self._resolve_metadata_config(self._metadata, ctx)
        run_metadata = self._resolve_metadata_config(additional_metadata, ctx)

        if base_metadata and run_metadata:
            return {**base_metadata, **run_metadata}
        return run_metadata or base_metadata

    def _resolve_metadata_config(
        self,
        config: AgentMetadata[AgentDepsT] | None,
        ctx: RunContext[AgentDepsT],
    ) -> dict[str, Any] | None:
        if config is None:
            return None
        metadata = config(ctx) if callable(config) else config
        return metadata

    def _resolve_and_store_metadata(
        self,
        graph_run_ctx: GraphRunContext[_agent_graph.GraphAgentState, _agent_graph.GraphAgentDeps[AgentDepsT, Any]],
        metadata: AgentMetadata[AgentDepsT] | None,
    ) -> dict[str, Any] | None:
        run_context = build_run_context(graph_run_ctx)
        resolved_metadata = self._get_metadata(run_context, metadata)
        graph_run_ctx.state.metadata = resolved_metadata
        return resolved_metadata

    def _resolve_spec(
        self,
        spec: dict[str, Any] | AgentSpec | None,
        custom_capability_types: Sequence[type[AbstractCapability[Any]]] = (),
    ) -> _ResolvedSpec | None:
        """Validate and instantiate capabilities from a spec, returning contributions.

        Returns None if spec is None.
        """
        if spec is None:
            return None

        validated_spec, template_context = _validate_spec(spec, self._deps_type)

        capabilities = list(_capabilities_from_spec(validated_spec, custom_capability_types, template_context))
        combined = CombinedCapability(capabilities) if capabilities else None

        retry_overrides = _retry_overrides_from_spec(validated_spec)

        # Warn for unsupported fields with non-default values. Read via `__dict__` to avoid
        # triggering pydantic deprecation warnings on deprecated spec fields.
        for field_name in _UNSUPPORTED_SPEC_FIELDS:
            field_info = type(validated_spec).model_fields[field_name]
            if validated_spec.__dict__[field_name] != field_info.default:
                warnings.warn(
                    f'AgentSpec field {field_name!r} is not supported at run/override time and will be ignored',
                    UserWarning,
                    stacklevel=3,
                )

        return _ResolvedSpec(
            capability=combined,
            instructions=_instructions.normalize_instructions(validated_spec.instructions)
            if validated_spec.instructions
            else [],
            model=validated_spec.model,
            model_settings=cast(ModelSettings, validated_spec.model_settings)
            if validated_spec.model_settings
            else None,
            metadata=validated_spec.metadata,
            name=validated_spec.name,
            output_retries=retry_overrides.get('output'),
            tool_retries=retry_overrides.get('tools'),
        )

    @contextmanager
    def override(  # noqa: C901
        self,
        *,
        name: str | _utils.Unset = _utils.UNSET,
        deps: AgentDepsT | _utils.Unset = _utils.UNSET,
        model: models.Model | models.KnownModelName | str | _utils.Unset = _utils.UNSET,
        toolsets: Sequence[AbstractToolset[AgentDepsT]] | _utils.Unset = _utils.UNSET,
        tools: Sequence[Tool[AgentDepsT] | ToolFuncEither[AgentDepsT, ...]] | _utils.Unset = _utils.UNSET,
        native_tools: Sequence[AgentNativeTool[AgentDepsT]] | _utils.Unset = _utils.UNSET,
        instructions: AgentInstructions[AgentDepsT] | _utils.Unset = _utils.UNSET,
        metadata: AgentMetadata[AgentDepsT] | _utils.Unset = _utils.UNSET,
        model_settings: AgentModelSettings[AgentDepsT] | _utils.Unset = _utils.UNSET,
        retries: int | AgentRetries | _utils.Unset = _utils.UNSET,
        spec: dict[str, Any] | AgentSpec | None = None,
    ) -> Generator[None]:
        """Context manager to temporarily override agent configuration.

        This is particularly useful when testing.
        You can find an example of this [here](../testing.md#overriding-model-via-pytest-fixtures).

        Args:
            name: The name to use instead of the name passed to the agent constructor and agent run.
            deps: The dependencies to use instead of the dependencies passed to the agent run.
            model: The model to use instead of the model passed to the agent run.
            toolsets: The toolsets to use instead of the toolsets passed to the agent constructor and agent run.
            tools: The tools to use instead of the tools registered with the agent.
            native_tools: The native tools to use instead of the agent's configured native tools.
            instructions: The instructions to use instead of the instructions registered with the agent.
                Note: this also replaces capability-contributed instructions (e.g. from
                [`get_instructions`][pydantic_ai.capabilities.AbstractCapability.get_instructions]).
            metadata: The metadata to use instead of the metadata passed to the agent constructor. When set, any
                per-run `metadata` argument is ignored.
            model_settings: The model settings to use instead of the model settings passed to the agent constructor.
                When set, any per-run `model_settings` argument is ignored.
            retries: The retry budgets to use instead of the agent-level configuration. Pass an `int` to
                override both the tool-retry and output budgets, or an [`AgentRetries`][pydantic_ai.AgentRetries]
                dict to override just one (e.g. `retries={'tools': 3}`).
                When set, any per-run `retries` argument is ignored.
            spec: Optional agent spec providing defaults for override. Explicit params take precedence
                over spec values. When the spec includes `capabilities`, they replace (not merge with)
                the agent's existing capabilities. To add capabilities without replacing, pass `spec`
                to `run()` or `iter()` instead.
        """
        # A bare `int` overrides both budgets; a partial `retries={'tools': ...}` / `{'output': ...}`
        # dict overrides only the named budget, so an unset budget still falls through to the run
        # kwarg, spec, or agent default.
        override_output_retries: int | _utils.Unset
        override_tool_retries: int | _utils.Unset
        if _utils.is_set(retries):
            retry_overrides = _normalize_agent_retry_overrides(retries)
            override_output_retries = retry_overrides.get('output', _utils.UNSET)
            override_tool_retries = retry_overrides.get('tools', _utils.UNSET)
        else:
            override_output_retries = _utils.UNSET
            override_tool_retries = _utils.UNSET

        resolved = self._resolve_spec(spec)

        # A spec capability replaces the agent's root capability for the duration of the
        # override. Build it before resolving an overridden model so custom model IDs can
        # be preserved for that capability's async, deps-aware resolver.
        if resolved is not None and resolved.capability is not None:
            override_caps = list(resolved.capability.capabilities)
            _inject_auto_capabilities(override_caps)
            override_capability: CombinedCapability[AgentDepsT] | None = CombinedCapability(override_caps).for_agent(
                self
            )
        else:
            override_capability = None

        # Apply spec values as defaults where explicit params are not set
        if resolved is not None:
            if not _utils.is_set(name) and resolved.name is not None:
                name = resolved.name
            if not _utils.is_set(model) and resolved.model is not None:
                model = resolved.model
            if not _utils.is_set(instructions) and resolved.instructions:
                instructions = resolved.instructions
            if not _utils.is_set(model_settings) and resolved.model_settings is not None:
                model_settings = resolved.model_settings
            if not _utils.is_set(metadata) and resolved.metadata is not None:
                metadata = resolved.metadata
            if not _utils.is_set(override_output_retries) and resolved.output_retries is not None:
                override_output_retries = resolved.output_retries
            if not _utils.is_set(override_tool_retries) and resolved.tool_retries is not None:
                override_tool_retries = resolved.tool_retries

        if _utils.is_set(name):
            name_token = self._override_name.set(_utils.Some(name))
        else:
            name_token = None

        if _utils.is_set(deps):
            deps_token = self._override_deps.set(_utils.Some(deps))
        else:
            deps_token = None

        if _utils.is_set(model):
            model_capability = override_capability or self._effective_root_capability()
            override_model = (
                model if isinstance(model, str) and model_capability.has_resolve_model_id else models.infer_model(model)
            )
            model_token = self._override_model.set(_utils.Some(override_model))
        else:
            model_token = None

        if _utils.is_set(toolsets):
            toolsets_token = self._override_toolsets.set(_utils.Some(toolsets))
        else:
            toolsets_token = None

        if _utils.is_set(tools):
            tools_token = self._override_tools.set(_utils.Some(tools))
        else:
            tools_token = None

        if _utils.is_set(native_tools):
            native_tools_token = self._override_native_tools.set(_utils.Some(native_tools))
        else:
            native_tools_token = None

        if _utils.is_set(instructions):
            normalized_instructions = _instructions.normalize_instructions(instructions)
            instructions_token = self._override_instructions.set(_utils.Some(normalized_instructions))
        else:
            instructions_token = None

        if _utils.is_set(metadata):
            metadata_token = self._override_metadata.set(_utils.Some(metadata))
        else:
            metadata_token = None

        if _utils.is_set(model_settings):
            model_settings_token = self._override_model_settings.set(_utils.Some(model_settings))
        else:
            model_settings_token = None

        if _utils.is_set(override_output_retries):
            output_retries_token = self._override_output_retries.set(_utils.Some(override_output_retries))
        else:
            output_retries_token = None

        if _utils.is_set(override_tool_retries):
            tool_retries_token = self._override_tool_retries.set(_utils.Some(override_tool_retries))
        else:
            tool_retries_token = None

        # Set capability from spec, replacing the agent's existing root capability.
        # Auto-inject infrastructure capabilities since the override replaces
        # (not merges with) the agent's root capability.
        if override_capability is not None:
            cap_token = self._override_root_capability.set(_utils.Some(override_capability))
        else:
            cap_token = None

        try:
            yield
        finally:
            if name_token is not None:
                self._override_name.reset(name_token)
            if deps_token is not None:
                self._override_deps.reset(deps_token)
            if model_token is not None:
                self._override_model.reset(model_token)
            if toolsets_token is not None:
                self._override_toolsets.reset(toolsets_token)
            if tools_token is not None:
                self._override_tools.reset(tools_token)
            if native_tools_token is not None:
                self._override_native_tools.reset(native_tools_token)
            if instructions_token is not None:
                self._override_instructions.reset(instructions_token)
            if metadata_token is not None:
                self._override_metadata.reset(metadata_token)
            if model_settings_token is not None:
                self._override_model_settings.reset(model_settings_token)
            if output_retries_token is not None:
                self._override_output_retries.reset(output_retries_token)
            if tool_retries_token is not None:
                self._override_tool_retries.reset(tool_retries_token)
            if cap_token is not None:
                self._override_root_capability.reset(cap_token)

    @overload
    def instructions(
        self, func: Callable[[RunContext[AgentDepsT]], str | None], /
    ) -> Callable[[RunContext[AgentDepsT]], str | None]: ...

    @overload
    def instructions(
        self, func: Callable[[RunContext[AgentDepsT]], Awaitable[str | None]], /
    ) -> Callable[[RunContext[AgentDepsT]], Awaitable[str | None]]: ...

    @overload
    def instructions(self, func: Callable[[], str | None], /) -> Callable[[], str | None]: ...

    @overload
    def instructions(self, func: Callable[[], Awaitable[str | None]], /) -> Callable[[], Awaitable[str | None]]: ...

    @overload
    def instructions(
        self, /, *, name: str | None = None
    ) -> Callable[[SystemPromptFunc[AgentDepsT]], SystemPromptFunc[AgentDepsT]]: ...

    def instructions(
        self,
        func: SystemPromptFunc[AgentDepsT] | None = None,
        /,
        *,
        name: str | None = None,
    ) -> Callable[[SystemPromptFunc[AgentDepsT]], SystemPromptFunc[AgentDepsT]] | SystemPromptFunc[AgentDepsT]:
        """Decorator to register an instructions function.

        Optionally takes [`RunContext`][pydantic_ai.tools.RunContext] as its only argument.
        Can decorate a sync or async functions.

        The decorator can be used bare (`agent.instructions`).

        Overloads for every possible signature of `instructions` are included so the decorator doesn't obscure
        the type of the function.

        Example:
        ```python
        from pydantic_ai import Agent, RunContext

        agent = Agent('test', deps_type=str)

        @agent.instructions
        def simple_instructions() -> str:
            return 'foobar'

        @agent.instructions
        async def async_instructions(ctx: RunContext[str]) -> str:
            return f'{ctx.deps} is the best'
        ```

        Args:
            func: The instructions function to register.
            name: An optional name for the instruction part this function produces, keyed as
                `'agent:<name>'` on [`InstructionPart.id`][pydantic_ai.messages.InstructionPart.id] so an
                application can address this part specifically, where the bare `'agent'` key addresses
                the agent's literal instructions. See [instruction parts](../agent.md#instruction-parts).
        """
        if name is not None:
            _instructions.validate_instruction_name(name)
        instruction_id = (
            _messages.InstructionId(_messages.AgentInstructionSource(), name=name) if name is not None else None
        )

        def decorator(
            func_: SystemPromptFunc[AgentDepsT],
        ) -> SystemPromptFunc[AgentDepsT]:
            self._instructions.append(
                _instructions.SourcedInstruction(func_, name=name, id=instruction_id, dynamic=True)
            )
            return func_

        return decorator if func is None else decorator(func)

    async def system_prompt_parts(
        self,
        *,
        deps: AgentDepsT = None,
        model: models.Model | models.KnownModelName | str | None = None,
        message_history: Sequence[_messages.ModelMessage] | None = None,
        prompt: str | Sequence[_messages.UserContent] | None = None,
        usage: _usage.RunUsage | None = None,
        model_settings: ModelSettings | None = None,
    ) -> list[_messages.SystemPromptPart]:
        """Resolve the agent's configured system prompts into `SystemPromptPart`s.

        See [`AbstractAgent.system_prompt_parts`][pydantic_ai.agent.AbstractAgent.system_prompt_parts].
        """
        deps = self._get_deps(deps)
        usage = usage or _usage.RunUsage()
        messages = list(message_history or [])
        capability = self._effective_root_capability()
        has_default_model = self._has_model(model)
        default_model = (
            await self._resolve_model_selection(self._pick_raw_model(model), capability=capability, deps=deps)
            if has_default_model
            else None
        )
        if model is None and self._override_model.get() is None:
            contribution = capability.get_model()
            if contribution is not None:
                selection_ctx = models.ModelSelectionContext(
                    agent=self,
                    deps=deps,
                    model=default_model,
                    run_step=1,
                    messages=messages,
                    usage=usage,
                )
                selected_model, _ = await self._evaluate_model_contribution(
                    contribution, capability=capability, ctx=selection_ctx
                )
            elif default_model is not None:
                selected_model = default_model
            else:
                raise exceptions.UserError('`model` must either be set on the agent or supplied by a capability.')
        else:
            assert default_model is not None
            selected_model = default_model
        run_context = RunContext[AgentDepsT](
            deps=deps,
            agent=self,
            model=selected_model,
            usage=usage,
            prompt=prompt,
            messages=messages,
            model_settings=model_settings,
            run_step=1,
        )
        return await _system_prompt.resolve_system_prompts(
            self._system_prompts, self._system_prompt_functions, run_context
        )

    @overload
    def system_prompt(
        self, func: Callable[[RunContext[AgentDepsT]], str | None], /
    ) -> Callable[[RunContext[AgentDepsT]], str | None]: ...

    @overload
    def system_prompt(
        self, func: Callable[[RunContext[AgentDepsT]], Awaitable[str | None]], /
    ) -> Callable[[RunContext[AgentDepsT]], Awaitable[str | None]]: ...

    @overload
    def system_prompt(self, func: Callable[[], str | None], /) -> Callable[[], str | None]: ...

    @overload
    def system_prompt(self, func: Callable[[], Awaitable[str | None]], /) -> Callable[[], Awaitable[str | None]]: ...

    @overload
    def system_prompt(
        self, /, *, dynamic: bool = False
    ) -> Callable[[SystemPromptFunc[AgentDepsT]], SystemPromptFunc[AgentDepsT]]: ...

    def system_prompt(
        self,
        func: SystemPromptFunc[AgentDepsT] | None = None,
        /,
        *,
        dynamic: bool = False,
    ) -> Callable[[SystemPromptFunc[AgentDepsT]], SystemPromptFunc[AgentDepsT]] | SystemPromptFunc[AgentDepsT]:
        """Decorator to register a system prompt function.

        Optionally takes [`RunContext`][pydantic_ai.tools.RunContext] as its only argument.
        Can decorate a sync or async functions.

        The decorator can be used either bare (`agent.system_prompt`) or as a function call
        (`agent.system_prompt(...)`), see the examples below.

        Overloads for every possible signature of `system_prompt` are included so the decorator doesn't obscure
        the type of the function, see `tests/typed_agent.py` for tests.

        Args:
            func: The function to decorate
            dynamic: If True, the system prompt will be reevaluated even when `messages_history` is provided,
                see [`SystemPromptPart.dynamic_ref`][pydantic_ai.messages.SystemPromptPart.dynamic_ref]

        Example:
        ```python
        from pydantic_ai import Agent, RunContext

        agent = Agent('test', deps_type=str)

        @agent.system_prompt
        def simple_system_prompt() -> str:
            return 'foobar'

        @agent.system_prompt(dynamic=True)
        async def async_system_prompt(ctx: RunContext[str]) -> str:
            return f'{ctx.deps} is the best'
        ```
        """
        if func is None:

            def decorator(
                func_: SystemPromptFunc[AgentDepsT],
            ) -> SystemPromptFunc[AgentDepsT]:
                runner = _system_prompt.SystemPromptRunner[AgentDepsT](func_, dynamic=dynamic)
                self._system_prompt_functions.append(runner)
                if dynamic:  # pragma: lax no cover
                    self._system_prompt_dynamic_functions[func_.__qualname__] = runner
                return func_

            return decorator
        else:
            assert not dynamic, "dynamic can't be True in this case"
            self._system_prompt_functions.append(_system_prompt.SystemPromptRunner[AgentDepsT](func, dynamic=dynamic))
            return func

    @overload
    def output_validator(
        self, func: Callable[[RunContext[AgentDepsT], OutputDataT], OutputDataT], /
    ) -> Callable[[RunContext[AgentDepsT], OutputDataT], OutputDataT]: ...

    @overload
    def output_validator(
        self, func: Callable[[RunContext[AgentDepsT], OutputDataT], Awaitable[OutputDataT]], /
    ) -> Callable[[RunContext[AgentDepsT], OutputDataT], Awaitable[OutputDataT]]: ...

    @overload
    def output_validator(
        self, func: Callable[[OutputDataT], OutputDataT], /
    ) -> Callable[[OutputDataT], OutputDataT]: ...

    @overload
    def output_validator(
        self, func: Callable[[OutputDataT], Awaitable[OutputDataT]], /
    ) -> Callable[[OutputDataT], Awaitable[OutputDataT]]: ...

    def output_validator(
        self, func: _output.OutputValidatorFunc[AgentDepsT, OutputDataT], /
    ) -> _output.OutputValidatorFunc[AgentDepsT, OutputDataT]:
        """Decorator to register an output validator function.

        Optionally takes [`RunContext`][pydantic_ai.tools.RunContext] as its first argument.
        Can decorate a sync or async functions.

        Overloads for every possible signature of `output_validator` are included so the decorator doesn't obscure
        the type of the function, see `tests/typed_agent.py` for tests.

        Example:
        ```python
        from pydantic_ai import Agent, ModelRetry, RunContext

        agent = Agent('test', deps_type=str)

        @agent.output_validator
        def output_validator_simple(data: str) -> str:
            if 'wrong' in data:
                raise ModelRetry('wrong response')
            return data

        @agent.output_validator
        async def output_validator_deps(ctx: RunContext[str], data: str) -> str:
            if ctx.deps in data:
                raise ModelRetry('wrong response')
            return data

        result = agent.run_sync('foobar', deps='spam')
        print(result.output)
        #> success (no tool calls)
        ```
        """
        self._output_validators.append(_output.OutputValidator[AgentDepsT, Any](func))
        return func

    @overload
    def on_event(
        self, func: OnEventHookFunc[_messages.AgentStreamEvent], /
    ) -> OnEventHookFunc[_messages.AgentStreamEvent]: ...

    @overload
    def on_event(
        self, *event_types: type[EventT], timeout: float | None = None
    ) -> Callable[[OnEventHookFunc[EventT]], OnEventHookFunc[EventT]]: ...

    def on_event(
        self,
        func_or_event_type: OnEventHookFunc[_messages.AgentStreamEvent] | type[EventT] | None = None,
        *event_types: type[EventT],
        timeout: float | None = None,
    ) -> Any:
        """Decorator to register a listener for events on this agent's run event stream.

        Every event on the stream can be listened for: the framework's own model and tool events,
        the application's [`CustomEvent`][pydantic_ai.messages.CustomEvent]s, and the
        [`CapabilityEvent`][pydantic_ai.messages.CapabilityEvent]s published by the agent's
        capabilities. Naming event classes narrows the `event` argument to their union and lets
        dispatch skip the agent's listeners for anything else; a bare `@agent.on_event` sees every
        [`AgentStreamEvent`][pydantic_ai.messages.AgentStreamEvent].

        This is the application-level counterpart to
        [`@on_event`][pydantic_ai.capabilities.on_event] on a capability, and it dispatches at the
        same point. These listeners join after the agent's own capabilities, so they see the events
        those emitted, and they survive an overridden root capability. Capability ordering still
        applies: one asking for `position='innermost'` keeps that position and its listeners run
        after these.

        Dispatch happens *upstream* of
        [`wrap_run_event_stream()`][pydantic_ai.capabilities.AbstractCapability.wrap_run_event_stream],
        so a listener sees each event as it was emitted, not as it is finally delivered. A
        capability that rewrites, replaces or drops events in its stream wrapper does so after
        every listener has already run, which means a listener can see an event that no stream
        consumer ever receives. To act on the delivered stream instead, wrap it yourself with
        `wrap_run_event_stream` on a [`Hooks`][pydantic_ai.capabilities.Hooks] capability, or
        consume [`run_stream_events()`][pydantic_ai.agent.AbstractAgent.run_stream_events].

        Being application code, a listener may emit a `CustomEvent` of its own. That is how a
        capability's internal event reaches a frontend: capability events are deliberately not
        forwarded by the [AG-UI](../ui/ag-ui.md) and [Vercel AI](../ui/vercel-ai.md) adapters, so
        you republish the part of one that is public.

        For hook families other than events, pass a [`Hooks`][pydantic_ai.capabilities.Hooks]
        capability to `capabilities=`, where its position among the other capabilities — which
        decides where it sits in each wrap chain — is yours to choose.

        Example:
        ```python
        from dataclasses import dataclass

        from pydantic_ai import Agent, CapabilityEvent, CustomEvent, RunContext

        agent = Agent('test')


        @dataclass(kw_only=True)
        class IndexRebuiltEvent(CapabilityEvent, namespace='indexer'):
            documents: int


        @dataclass(kw_only=True)
        class SearchReadyEvent(CustomEvent):
            documents: int


        @agent.on_event(IndexRebuiltEvent)
        async def republish(ctx: RunContext[None], event: IndexRebuiltEvent) -> None:
            await ctx.emit(SearchReadyEvent(documents=event.documents))
        ```
        """
        # `Hooks.on.event` already sorts the bare form from the filtered one; forward verbatim so
        # there is one implementation of that split. Typed `Any` because the overloads above carry
        # the signature, as they do for the registrar itself.
        registrar: Any = self._event_hooks.on
        if func_or_event_type is None:
            return registrar.event(*event_types, timeout=timeout)
        return registrar.event(func_or_event_type, *event_types, timeout=timeout)

    @overload
    def tool(self, func: ToolFuncContext[AgentDepsT, ToolParams], /) -> ToolFuncContext[AgentDepsT, ToolParams]: ...

    @overload
    def tool(
        self,
        /,
        *,
        name: str | None = None,
        description: str | None = None,
        retries: int | None = None,
        prepare: ToolPrepareFunc[AgentDepsT] | None = None,
        args_validator: ArgsValidatorFunc[AgentDepsT, ToolParams] | None = None,
        docstring_format: DocstringFormat = 'auto',
        require_parameter_descriptions: bool = False,
        schema_generator: type[GenerateJsonSchema] = GenerateToolJsonSchema,
        strict: bool | None = None,
        sequential: bool = False,
        requires_approval: bool = False,
        metadata: dict[str, Any] | None = None,
        timeout: float | None = None,
        defer_loading: bool = False,
        include_return_schema: bool | None = None,
    ) -> Callable[[ToolFuncContext[AgentDepsT, ToolParams]], ToolFuncContext[AgentDepsT, ToolParams]]: ...

    def tool(
        self,
        func: ToolFuncContext[AgentDepsT, ToolParams] | None = None,
        /,
        *,
        name: str | None = None,
        description: str | None = None,
        retries: int | None = None,
        prepare: ToolPrepareFunc[AgentDepsT] | None = None,
        args_validator: ArgsValidatorFunc[AgentDepsT, ToolParams] | None = None,
        docstring_format: DocstringFormat = 'auto',
        require_parameter_descriptions: bool = False,
        schema_generator: type[GenerateJsonSchema] = GenerateToolJsonSchema,
        strict: bool | None = None,
        sequential: bool = False,
        requires_approval: bool = False,
        metadata: dict[str, Any] | None = None,
        timeout: float | None = None,
        defer_loading: bool = False,
        include_return_schema: bool | None = None,
    ) -> Any:
        """Decorator to register a tool function which takes [`RunContext`][pydantic_ai.tools.RunContext] as its first argument.

        Can decorate a sync or async functions.

        The docstring is inspected to extract both the tool description and description of each parameter,
        [learn more](../tools.md#function-tools-and-schema).

        We can't add overloads for every possible signature of tool, since the return type is a recursive union
        so the signature of functions decorated with `@agent.tool` is obscured.

        Example:
        ```python
        from pydantic_ai import Agent, RunContext

        agent = Agent('test', deps_type=int)

        @agent.tool
        def foobar(ctx: RunContext[int], x: int) -> int:
            return ctx.deps + x

        @agent.tool(retries=2)
        async def spam(ctx: RunContext[str], y: float) -> float:
            return ctx.deps + y

        result = agent.run_sync('foobar', deps=1)
        print(result.output)
        #> {"foobar":1,"spam":1.0}
        ```

        Args:
            func: The tool function to register.
            name: The name of the tool, defaults to the function name.
            description: The description of the tool, defaults to the function docstring.
            retries: The number of retries to allow for this tool, defaults to the agent's default retries,
                which defaults to 1.
            prepare: custom method to prepare the tool definition for each step, return `None` to omit this
                tool from a given step. This is useful if you want to customise a tool at call time,
                or omit it completely from a step. See [`ToolPrepareFunc`][pydantic_ai.tools.ToolPrepareFunc].
            args_validator: custom method to validate tool arguments after schema validation has passed,
                before execution. The validator receives the already-validated and type-converted parameters,
                with `RunContext` as the first argument.
                Raise [`ModelRetry`][pydantic_ai.exceptions.ModelRetry] to ask the model to correct the
                arguments and try again, or [`ToolFailed`][pydantic_ai.exceptions.ToolFailed] to report a
                terminal failure the model should adapt to instead of retrying. Return `None` on success.
                See [`ArgsValidatorFunc`][pydantic_ai.tools.ArgsValidatorFunc].
            docstring_format: The format of the docstring, see [`DocstringFormat`][pydantic_ai.tools.DocstringFormat].
                Defaults to `'auto'`, such that the format is inferred from the structure of the docstring.
            require_parameter_descriptions: If True, raise an error if a parameter description is missing. Defaults to False.
            schema_generator: The JSON schema generator class to use for this tool. Defaults to `GenerateToolJsonSchema`.
            strict: Whether to enforce (vendor-specific) strict schema adherence for tool calls (supported by OpenAI, Anthropic, Google, and Bedrock).
                See [`ToolDefinition`][pydantic_ai.tools.ToolDefinition] for more info.
            sequential: Whether this tool acts as a barrier that runs alone, not overlapping with other tool calls.
                See [`ToolDefinition`][pydantic_ai.tools.ToolDefinition] for more info. Defaults to False.
            requires_approval: Whether this tool requires human-in-the-loop approval. Defaults to False.
                See the [tools documentation](../deferred-tools.md#human-in-the-loop-tool-approval) for more info.
            metadata: Optional metadata for the tool. This is not sent to the model but can be used for filtering and tool behavior customization.
            timeout: Timeout in seconds for tool execution. If the tool takes longer, a retry prompt is returned to the model.
                Overrides the agent-level `tool_timeout` if set. Defaults to None (no timeout).
            defer_loading: Whether to hide this tool until it's revealed by tool search, `load_capability`,
                or another tool's `ToolReturn.tools`. Defaults to False.
                See [Tool Search](../tools-advanced.md#tool-search) for more info.
            include_return_schema: Whether to include the return schema in the tool definition sent to the model.
                If `None`, defaults to `False` unless the [`IncludeToolReturnSchemas`][pydantic_ai.capabilities.IncludeToolReturnSchemas] capability is used.
        """

        def tool_decorator(
            func_: ToolFuncContext[AgentDepsT, ToolParams],
        ) -> ToolFuncContext[AgentDepsT, ToolParams]:
            # noinspection PyTypeChecker
            self._function_toolset.add_function(
                func_,
                takes_ctx=True,
                name=name,
                description=description,
                retries=retries,
                prepare=prepare,
                args_validator=args_validator,
                docstring_format=docstring_format,
                require_parameter_descriptions=require_parameter_descriptions,
                schema_generator=schema_generator,
                strict=strict,
                sequential=sequential,
                requires_approval=requires_approval,
                metadata=metadata,
                timeout=timeout,
                defer_loading=defer_loading,
                include_return_schema=include_return_schema,
            )
            return func_

        return tool_decorator if func is None else tool_decorator(func)

    @overload
    def tool_plain(self, func: ToolFuncPlain[ToolParams], /) -> ToolFuncPlain[ToolParams]: ...

    @overload
    def tool_plain(
        self,
        /,
        *,
        name: str | None = None,
        description: str | None = None,
        retries: int | None = None,
        prepare: ToolPrepareFunc[AgentDepsT] | None = None,
        args_validator: ArgsValidatorFunc[AgentDepsT, ToolParams] | None = None,
        docstring_format: DocstringFormat = 'auto',
        require_parameter_descriptions: bool = False,
        schema_generator: type[GenerateJsonSchema] = GenerateToolJsonSchema,
        strict: bool | None = None,
        sequential: bool = False,
        requires_approval: bool = False,
        metadata: dict[str, Any] | None = None,
        timeout: float | None = None,
        defer_loading: bool = False,
        include_return_schema: bool | None = None,
    ) -> Callable[[ToolFuncPlain[ToolParams]], ToolFuncPlain[ToolParams]]: ...

    def tool_plain(
        self,
        func: ToolFuncPlain[ToolParams] | None = None,
        /,
        *,
        name: str | None = None,
        description: str | None = None,
        retries: int | None = None,
        prepare: ToolPrepareFunc[AgentDepsT] | None = None,
        args_validator: ArgsValidatorFunc[AgentDepsT, ToolParams] | None = None,
        docstring_format: DocstringFormat = 'auto',
        require_parameter_descriptions: bool = False,
        schema_generator: type[GenerateJsonSchema] = GenerateToolJsonSchema,
        strict: bool | None = None,
        sequential: bool = False,
        requires_approval: bool = False,
        metadata: dict[str, Any] | None = None,
        timeout: float | None = None,
        defer_loading: bool = False,
        include_return_schema: bool | None = None,
    ) -> Any:
        """Decorator to register a tool function which DOES NOT take `RunContext` as an argument.

        Can decorate a sync or async functions.

        The docstring is inspected to extract both the tool description and description of each parameter,
        [learn more](../tools.md#function-tools-and-schema).

        We can't add overloads for every possible signature of tool, since the return type is a recursive union
        so the signature of functions decorated with `@agent.tool` is obscured.

        Example:
        ```python
        from pydantic_ai import Agent, RunContext

        agent = Agent('test')

        @agent.tool
        def foobar(ctx: RunContext[int]) -> int:
            return 123

        @agent.tool(retries=2)
        async def spam(ctx: RunContext[str]) -> float:
            return 3.14

        result = agent.run_sync('foobar', deps=1)
        print(result.output)
        #> {"foobar":123,"spam":3.14}
        ```

        Args:
            func: The tool function to register.
            name: The name of the tool, defaults to the function name.
            description: The description of the tool, defaults to the function docstring.
            retries: The number of retries to allow for this tool, defaults to the agent's default retries,
                which defaults to 1.
            prepare: custom method to prepare the tool definition for each step, return `None` to omit this
                tool from a given step. This is useful if you want to customise a tool at call time,
                or omit it completely from a step. See [`ToolPrepareFunc`][pydantic_ai.tools.ToolPrepareFunc].
            args_validator: custom method to validate tool arguments after schema validation has passed,
                before execution. The validator receives the already-validated and type-converted parameters,
                with [`RunContext`][pydantic_ai.tools.RunContext] as the first argument — even though the
                tool function itself does not take `RunContext` when using `tool_plain`.
                Raise [`ModelRetry`][pydantic_ai.exceptions.ModelRetry] to ask the model to correct the
                arguments and try again, or [`ToolFailed`][pydantic_ai.exceptions.ToolFailed] to report a
                terminal failure the model should adapt to instead of retrying. Return `None` on success.
                See [`ArgsValidatorFunc`][pydantic_ai.tools.ArgsValidatorFunc].
            docstring_format: The format of the docstring, see [`DocstringFormat`][pydantic_ai.tools.DocstringFormat].
                Defaults to `'auto'`, such that the format is inferred from the structure of the docstring.
            require_parameter_descriptions: If True, raise an error if a parameter description is missing. Defaults to False.
            schema_generator: The JSON schema generator class to use for this tool. Defaults to `GenerateToolJsonSchema`.
            strict: Whether to enforce (vendor-specific) strict schema adherence for tool calls (supported by OpenAI, Anthropic, Google, and Bedrock).
                See [`ToolDefinition`][pydantic_ai.tools.ToolDefinition] for more info.
            sequential: Whether this tool acts as a barrier that runs alone, not overlapping with other tool calls.
                See [`ToolDefinition`][pydantic_ai.tools.ToolDefinition] for more info. Defaults to False.
            requires_approval: Whether this tool requires human-in-the-loop approval. Defaults to False.
                See the [tools documentation](../deferred-tools.md#human-in-the-loop-tool-approval) for more info.
            metadata: Optional metadata for the tool. This is not sent to the model but can be used for filtering and tool behavior customization.
            timeout: Timeout in seconds for tool execution. If the tool takes longer, a retry prompt is returned to the model.
                Overrides the agent-level `tool_timeout` if set. Defaults to None (no timeout).
            defer_loading: Whether to hide this tool until it's revealed by tool search, `load_capability`,
                or another tool's `ToolReturn.tools`. Defaults to False.
                See [Tool Search](../tools-advanced.md#tool-search) for more info.
            include_return_schema: Whether to include the return schema in the tool definition sent to the model.
                If `None`, defaults to `False` unless the [`IncludeToolReturnSchemas`][pydantic_ai.capabilities.IncludeToolReturnSchemas] capability is used.
        """

        def tool_decorator(func_: ToolFuncPlain[ToolParams]) -> ToolFuncPlain[ToolParams]:
            # noinspection PyTypeChecker
            self._function_toolset.add_function(
                func_,
                takes_ctx=False,
                name=name,
                description=description,
                retries=retries,
                prepare=prepare,
                args_validator=args_validator,
                docstring_format=docstring_format,
                require_parameter_descriptions=require_parameter_descriptions,
                schema_generator=schema_generator,
                strict=strict,
                sequential=sequential,
                requires_approval=requires_approval,
                metadata=metadata,
                timeout=timeout,
                defer_loading=defer_loading,
                include_return_schema=include_return_schema,
            )
            return func_

        return tool_decorator if func is None else tool_decorator(func)

    @overload
    def toolset(self, func: ToolsetFunc[AgentDepsT], /) -> ToolsetFunc[AgentDepsT]: ...

    @overload
    def toolset(
        self,
        /,
        *,
        per_run_step: bool = True,
        id: str | None = None,
    ) -> Callable[[ToolsetFunc[AgentDepsT]], ToolsetFunc[AgentDepsT]]: ...

    def toolset(
        self,
        func: ToolsetFunc[AgentDepsT] | None = None,
        /,
        *,
        per_run_step: bool = True,
        id: str | None = None,
    ) -> Any:
        """Decorator to register a toolset function which takes [`RunContext`][pydantic_ai.tools.RunContext] as its only argument.

        Can decorate a sync or async functions.

        The decorator can be used bare (`agent.toolset`).

        Example:
        ```python
        from pydantic_ai import AbstractToolset, Agent, FunctionToolset, RunContext

        agent = Agent('test', deps_type=str)

        @agent.toolset
        async def simple_toolset(ctx: RunContext[str]) -> AbstractToolset[str]:
            return FunctionToolset()
        ```

        Args:
            func: The toolset function to register.
            per_run_step: Whether to re-evaluate the toolset for each run step. Defaults to True.
            id: An optional unique ID for the dynamic toolset. Under durable execution, construct a
                [`DynamicToolset`][pydantic_ai.toolsets.DynamicToolset] with this ID and pass it to
                `Agent(toolsets=[...])` instead; decorator registrations cannot be used inside a
                workflow or flow because they happen after durable units are created.
        """

        def toolset_decorator(func_: ToolsetFunc[AgentDepsT]) -> ToolsetFunc[AgentDepsT]:
            self._dynamic_toolsets.append(DynamicToolset(func_, per_run_step=per_run_step, id=id))
            return func_

        return toolset_decorator if func is None else toolset_decorator(func)

    def _has_model(self, model: models.Model | models.KnownModelName | str | None) -> bool:
        """Whether a run given `model` would have one to use, counting an `override(model=...)`.

        What `_pick_raw_model` answers for, asked ahead of it by callers that would rather not have
        it raise. A capability can still contribute a model when this is False.
        """
        return model is not None or self._override_model.get() is not None or self.model is not None

    def _pick_raw_model(
        self, model: models.Model | models.KnownModelName | str | None
    ) -> models.Model | models.KnownModelName | str:
        if some_model := self._override_model.get():
            return some_model.value
        if model is not None:
            return model
        if self.model is not None:
            return self.model
        raise exceptions.UserError('`model` must either be set on the agent or included when calling it.')

    def _effective_root_capability(self) -> CombinedCapability[AgentDepsT]:
        """Return the override capability when present, otherwise the configured root."""
        override = self._override_root_capability.get()
        return override.value if override is not None else self._root_capability

    def _startup_banner_details(
        self,
        model: models.Model | models.KnownModelName | str | None,
        toolsets: Sequence[AbstractToolset[AgentDepsT]] | None = None,
    ) -> _StartupBannerDetails:
        """Describe the session `_cli` is about to open, before any run has resolved anything.

        Resolved the way a run resolves it rather than read off the agent as configured: an
        `override()` in force, or instrumentation switched on globally by `Agent.instrument_all()`,
        would otherwise have the banner describe a different session than the one about to start.

        Args:
            model: Model the session was asked to use, if not the agent's own.
            toolsets: Toolsets the session will pass to each run, which aren't on the agent.
        """
        chat_model = self._pick_raw_model(model)
        # `self.toolsets` is override-aware, so an `override(toolsets=...)` is reflected.
        session_toolsets = [*self.toolsets, *(toolsets or [])]
        # Only a `FunctionToolset` holds its tools synchronously; every other toolset answers
        # `get_tools()` given a `RunContext`, and an MCP server would have to be connected to first.
        countable = [toolset for toolset in session_toolsets if isinstance(toolset, FunctionToolset)]
        return _StartupBannerDetails(
            model=chat_model.model_id if isinstance(chat_model, models.Model) else chat_model,
            # Rather than report a number that's wrong — `clai --mcp-config` would have said
            # `tools: 0` next to a session full of MCP tools — the banner leaves the count out
            # entirely. A run's own banner counts what the model is really offered.
            tools=sum(len(toolset.tools) for toolset in countable) if len(countable) == len(session_toolsets) else None,
            capabilities=_registered_capability_count(self._effective_root_capability()),
            instrumented=(
                isinstance(chat_model, InstrumentedModel) or self._resolve_instrumentation_settings() is not None
            ),
        )

    def _resolve_tool_retries(self, retries: int | None = None) -> int:
        """Resolve the effective tool-retry default: override > run/spec > agent default."""
        override = self._override_tool_retries.get()
        if override is not None:
            return override.value
        return retries if retries is not None else self._max_tool_retries

    def _base_run_capability(self) -> tuple[CombinedCapability[AgentDepsT], bool]:
        """The base capability layer for a run, plus whether it came from `override(root_capability=...)`.

        `iter` and `realtime_session` both resolve the base layer through this so the override is honored
        identically — KEEP the two call sites in sync (a realtime session that ignored the override would
        silently drop a `with agent.override(root_capability=...):` block).
        """
        override_cap = self._override_root_capability.get()
        base = self._effective_root_capability()
        if self._event_hooks.has_on_event:
            # Wrapped here rather than in `_root_capability` so an override of it cannot drop the
            # listeners; `Hooks` binds to itself, so it needs no `for_agent` pass. `CombinedCapability`
            # re-sorts, so this joins after the agent's own capabilities but still yields to one that
            # asks for `position='innermost'` — see `on_event`'s docstring.
            base = CombinedCapability([base, self._event_hooks])
        return base, override_cap is not None

    def _bind_run_capabilities(
        self, extra_capabilities: list[AbstractCapability[AgentDepsT]]
    ) -> list[AbstractCapability[AgentDepsT]]:
        """Bind per-run capabilities to this agent via `for_agent` before capability resolution.

        `_resolve_run_capabilities` only ever calls `for_run`, so binding the per-run layer via
        `for_agent` is the caller's responsibility. `iter` and `realtime_session` both MUST call this —
        skipping it uses a capability that overrides `for_agent` (e.g. the durability capabilities)
        unbound, a silent divergence. KEEP the two call sites in sync.
        """
        return [capability.for_agent(self) for capability in extra_capabilities]

    async def _resolve_model_selection(
        self,
        selection: ModelSelection,
        *,
        capability: AbstractCapability[AgentDepsT],
        deps: AgentDepsT,
        resolved_models: dict[tuple[int, str], models.Model] | None = None,
    ) -> models.Model:
        """Resolve a concrete model selection through the capability chain."""
        if not isinstance(selection, str):
            return selection
        cache_key = (id(capability), selection)
        if resolved_models is not None and (resolved_model := resolved_models.get(cache_key)) is not None:
            return resolved_model
        if entered_model := self._entered_models_by_selection.get(cache_key):
            if resolved_models is not None:
                resolved_models[cache_key] = entered_model
            return entered_model
        resolution_ctx = models.ModelResolutionContext(agent=self, deps=deps)
        resolved = await capability.resolve_model_id(resolution_ctx, model_id=selection)
        resolved_model = resolved if resolved is not None else models.infer_model(selection)
        if resolved_models is not None:
            resolved_models[cache_key] = resolved_model
        return resolved_model

    async def _evaluate_model_contribution(
        self,
        contribution: AgentModel[AgentDepsT],
        *,
        capability: AbstractCapability[AgentDepsT],
        ctx: models.ModelSelectionContext[AgentDepsT],
        resolved_models: dict[tuple[int, str], models.Model] | None = None,
    ) -> tuple[models.Model, str | None]:
        """Evaluate a static or dynamic model contribution and resolve its result."""
        selection = contribution(ctx) if callable(contribution) and not _is_model(contribution) else contribution
        if inspect.isawaitable(selection):
            selection = await selection
        model = await self._resolve_model_selection(
            selection, capability=capability, deps=ctx.deps, resolved_models=resolved_models
        )
        return model, selection if isinstance(selection, str) else None

    @staticmethod
    def _check_dynamic_model_resume(
        contribution: AgentModel[AgentDepsT] | None,
        message_history: Sequence[_messages.ModelMessage] | None,
    ) -> None:
        """Reject cross-run continuation when a selector cannot reconstruct the pinned model."""
        if (
            callable(contribution)
            and not _is_model(contribution)
            and message_history
            and isinstance(message_history[-1], _messages.ModelResponse)
            and message_history[-1].state == 'suspended'
        ):
            raise exceptions.UserError(
                'Cannot resume a suspended response with a dynamic capability model: the model '
                'that created the provider-side job cannot be reconstructed unambiguously. Pass '
                'that model explicitly to `run(model=...)` when resuming.'
            )

    def _get_model_outside_run(self, model: models.Model | models.KnownModelName | str | None = None) -> models.Model:
        """Resolve a configured or static capability model where run deps are unavailable."""
        capability = self._effective_root_capability()
        if model is not None or self._override_model.get() is not None:
            selection = self._pick_raw_model(model)
            return selection if _is_model(selection) else models.infer_model(selection)
        contribution = capability.get_model()
        if callable(contribution) and not _is_model(contribution):
            raise exceptions.UserError(
                'The capability model is dynamic and can only be selected during a run with run dependencies. '
                'Pass a concrete model explicitly.'
            )
        selection = contribution if contribution is not None else self._pick_raw_model(None)
        if isinstance(selection, str) and capability.has_resolve_model_id:
            raise exceptions.UserError(
                'The configured model ID is resolved by a capability using run dependencies. '
                'Pass a concrete model explicitly.'
            )
        return selection if _is_model(selection) else models.infer_model(selection)

    def _resolve_instrumentation_settings(self) -> InstrumentationSettings | None:
        """Resolve effective `InstrumentationSettings` from `Agent.instrument_all` / `agent.instrument`."""
        instrument = self._instrument if self._instrument is not None else self._instrument_default
        if not instrument:
            return None
        return InstrumentationSettings() if instrument is True else instrument

    def _get_deps(self: Agent[T, OutputDataT], deps: T) -> T:
        """Get deps for a run.

        If we've overridden deps via `_override_deps`, use that, otherwise use the deps passed to the call.

        We could do runtime type checking of deps against `self._deps_type`, but that's a slippery slope.
        """
        if some_deps := self._override_deps.get():
            return some_deps.value
        else:
            return deps

    async def _resolve_run_capabilities(
        self,
        ctx: RunContext[AgentDepsT],
        *,
        base_capability: AbstractCapability[AgentDepsT],
        extra_capabilities: list[AbstractCapability[AgentDepsT]],
        instrumentation_cap: InstrumentationCap | None,
        inject_deferred_loader: bool,
        base_is_override: bool,
    ) -> _ResolvedRunCapabilities[AgentDepsT]:
        """Resolve the per-run capability layers and extract their contributions.

        Shared by [`iter`][pydantic_ai.agent.AbstractAgent.iter] / [`run`][pydantic_ai.agent.AbstractAgent.run]
        and [`realtime_session`][pydantic_ai.agent.Agent.realtime_session] so both wire capabilities up
        identically: the outermost `Instrumentation` injection, per-layer `for_run` resolution (never
        composing first — see below), optional deferred-loader injection, and the native-tool /
        instruction / model-settings / toolset contributions with `override(native_tools=...)` folded
        in. Each caller keeps its own surrounding logic (instrumentation *settings* resolution,
        model-settings layering, output toolset, the graph vs. the connection) and cross-references
        this method so the two stay in sync.

        `ctx` must already carry `metadata` (and, when instrumented, `tracer` / `trace_include_content`),
        since capability `for_run` hooks observe it. `base_is_override` is whether `base_capability`
        came from `override(root_capability=...)` (only the graph run supports that today), used solely
        for the native-tool validation error's `source`.
        """
        run_layers: list[AbstractCapability[AgentDepsT]] = [base_capability, *extra_capabilities]
        # Prepend `Instrumentation` (outermost, so its spans wrap everything) unless the user already
        # added one themselves — mirroring the explicit-capability-wins precedence.
        if instrumentation_cap is not None and not has_capability_type(run_layers, InstrumentationCap):
            run_layers.insert(0, instrumentation_cap)

        # Resolve `for_run` per layer instead of composing a `CombinedCapability` first (which would
        # gather over the same children): the `override(native_tools=...)` merge below needs the
        # *resolved* extras — their native tools may only materialize in `for_run`, e.g. from a
        # capability function's returned capability — and the composed tree re-flattens/re-sorts its
        # children and can't hand them back. Nor can the extras be re-resolved afterwards to peek:
        # `for_run` is documented as called once per run and may have per-run side effects (the
        # durable-exec integrations rely on this for deterministic replay). Composing from the resolved
        # pieces yields the same structure as resolving a pre-composed tree, since the same
        # flatten-and-sort runs on the same resolved children either way.
        resolved_layers = await _utils.gather(*(cap.for_run(ctx) for cap in run_layers))
        # The extras are the tail of `run_layers` (instrumentation, if added, is at the front). Slicing
        # from the front avoids the `[-0:]` full-list pitfall when there are no extras.
        resolved_extras = resolved_layers[len(resolved_layers) - len(extra_capabilities) :]
        base_capability._validate_runtime_capabilities(  # pyright: ignore[reportPrivateUsage]
            ctx,
            [capability for extra in resolved_extras for capability in leaf_capabilities(extra)],
        )
        # Two capabilities under one `id` name the same thing, so the tree is resolved down to one
        # each before anything reads it. Duplicates *within* a layer are one configuration stated
        # twice and `combine` settles them; they are combined here, exactly once, and the merged
        # survivor is the only form anyone reads -- the per-layer native-tool validation below and
        # the contributions above draw on the same trees, so validation and advertisement cannot
        # see different merged instances. The agent's capabilities and the run's are then settled
        # against each other as separate layers: a run-level capability overrides its agent-level
        # namesake outright -- `run(capabilities=[WebSearch(allowed_domains=[...])])` states what
        # this run may reach, and merging it into the agent's list would widen the restriction it
        # was passed to impose.
        combined_layers = [
            _combine_duplicate_capabilities(CombinedCapability(list(layer)) if len(layer) > 1 else layer[0], [layer])
            for layer in (resolved_layers[: len(resolved_layers) - len(extra_capabilities)], resolved_extras)
            if layer
        ]
        run_capability = (
            _combine_duplicate_capabilities(
                CombinedCapability(combined_layers) if len(combined_layers) > 1 else combined_layers[0],
                [[layer] for layer in combined_layers],
            )
            if len(combined_layers) > 1
            else combined_layers[0]
        )
        # Not covered by the construction-time check: a run's capabilities compose with a retained
        # overriding container exactly as a registered sibling does, and `for_run` may hand back a
        # capability whose `id` differs from the one that was validated, so the resolved tree is
        # checked even when no additional layer was composed. Reads the *combined* tree, so a
        # duplicate `combine` has already resolved is not reported twice over.
        _validate_instruction_source_ids([run_capability])

        # Re-extract get_*() from the resolved capability if anything is contributed per-run.
        capabilities = _build_run_capabilities(run_capability)
        # Inject the loader only if a deferred capability is present AND `for_run` didn't already return
        # one, or a second loader toolset then errors on the reserved `load_capability` name
        # (cf. https://github.com/pydantic/pydantic-ai/issues/5047).
        if inject_deferred_loader and (
            any(capability.defer_loading is True for capability in capabilities.values())
            and not has_capability_type([run_capability], DeferredCapabilityLoader)
        ):
            run_capability = CombinedCapability([run_capability, DeferredCapabilityLoader()])
            capabilities = _build_run_capabilities(run_capability)

        # Register the run's capabilities on the context so the `toolset.for_run(ctx)` each caller runs
        # next evaluates dynamic-toolset factories against them (e.g. a `DynamicCapability`'s contributed
        # toolset reuses the capability instance `for_run` already resolved).
        ctx.capabilities = capabilities

        # Only read contributions from the resolved tree when the run actually has capabilities beyond
        # the plain agent default; otherwise fall back to the init-time snapshots.
        if run_capability is not base_capability or base_is_override:
            source_cap: AbstractCapability[AgentDepsT] | None = run_capability
        else:
            source_cap = None
        if source_cap is not None:
            instructions = source_cap._collect_instructions()  # pyright: ignore[reportPrivateUsage]
            native_tools = list(source_cap.get_native_tools())
            model_settings = source_cap.get_model_settings()
            cap_toolset = source_cap.get_toolset()
            toolsets: list[AgentToolset[AgentDepsT]] | None = [cap_toolset] if cap_toolset is not None else []
        else:
            instructions = None  # use init-time defaults
            native_tools = self._cap_native_tools
            model_settings = self._cap_model_settings
            toolsets = None

        # Native tool ids are validated per layer, off each layer's *combined* form above -- the
        # same merged trees the run tree is built from, so validation and advertisement read one
        # instance. Conflicting definitions sharing a `unique_id` *within* a layer are ambiguous;
        # last-wins *across* layers is the intentional override mechanism. Instrumentation
        # contributes no native tools.
        base_native_tools = list(combined_layers[0].get_native_tools())
        _validate_native_tool_ids(
            base_native_tools,
            source='override spec capabilities' if base_is_override else 'agent capabilities',
        )
        extra_native_tools = list(combined_layers[1].get_native_tools()) if len(combined_layers) > 1 else []
        _validate_native_tool_ids(extra_native_tools, source='run capabilities')

        # `override(native_tools=...)` replaces the agent's *baseline* native tools while still
        # preserving any additional per-run capability-contributed native tools on top.
        if some_native_tools := self._override_native_tools.get():
            _validate_native_tool_ids(some_native_tools.value, source='override native_tools')
            native_tools = [*some_native_tools.value, *extra_native_tools]

        return _ResolvedRunCapabilities(
            run_capability=run_capability,
            capabilities=capabilities,
            instructions=instructions,
            native_tools=native_tools,
            model_settings=model_settings,
            toolsets=toolsets,
            resolved_layers=resolved_layers,
        )

    def _get_instructions(
        self,
        additional_instructions: AgentInstructions[AgentDepsT] = None,
        cap_instructions: list[_instructions.SourcedInstruction[AgentDepsT]] | None = None,
    ) -> list[_instructions.SourcedInstruction[AgentDepsT]]:
        """Collect the lazy agent-level instructions for final request resolution.

        Toolset instructions are collected separately during run execution.

        Args:
            additional_instructions: Additional instructions to include for this run.
            cap_instructions: Instructions from capabilities, resolved at run time.

        """
        override_instructions = self._override_instructions.get()
        instructions: list[_instructions.SourcedInstruction[AgentDepsT]]
        if override_instructions:
            # Override replaces all instructions, including capability contributions, so what it
            # provides takes the place of (and the id of) the agent's own instructions.
            instructions = [
                _instructions.sourced_instruction(instruction, _agent_instruction_source(instruction))
                for instruction in override_instructions.value
            ]
        else:
            instructions = [*self._instructions]
            instructions.extend(cap_instructions if cap_instructions is not None else self._cap_instructions)
            if additional_instructions is not None:
                # Instructions passed to a specific run are already the caller's to change, and
                # aren't part of the agent's own configured instructions.
                instructions.extend(
                    _instructions.sourced_instruction(instruction, None)
                    for instruction in _instructions.normalize_instructions(additional_instructions)
                )

        return instructions

    def _get_toolset(
        self,
        output_toolset: AbstractToolset[AgentDepsT] | None | _utils.Unset = _utils.UNSET,
        additional_toolsets: Sequence[AbstractToolset[AgentDepsT]] | None = None,
        cap_toolsets: Sequence[AgentToolset[AgentDepsT]] | None = None,
        run_capability: AbstractCapability[AgentDepsT] | None = None,
        max_output_retries: int | None = None,
    ) -> AbstractToolset[AgentDepsT]:
        """Get the complete toolset.

        Args:
            output_toolset: The output toolset to use instead of the one built at agent construction time.
            additional_toolsets: Additional toolsets to add, unless toolsets have been overridden.
            cap_toolsets: Per-run capability toolsets to use instead of the init-time capability toolsets.
            run_capability: The per-run capability instance, used to apply wrapper toolsets.
            max_output_retries: The effective output retry budget for this run (run kwarg / spec / agent default).
                Used as `ctx.max_retries` for the `prepare_output_tools` capability hook so it sees the
                same budget the run will actually enforce. Falls back to the agent-level default.
        """
        toolsets = list(self._build_toolset_list(cap_toolsets=cap_toolsets))
        # Don't add additional toolsets if the toolsets have been overridden
        if additional_toolsets and self._override_toolsets.get() is None:
            toolsets = [*toolsets, *additional_toolsets]

        toolset: AbstractToolset[AgentDepsT] = CombinedToolset(toolsets)

        if run_capability is not None:
            # Dispatch the `prepare_tools` capability hook through a `PreparedToolset` wrapped
            # **inside** any other capability `get_wrapper_toolset` results (e.g. `ToolSearch`,
            # `CodeMode`): filter/modify defs first, let other toolset transformations layer on
            # top. The hook sees **function** tools only — output tools route through
            # `prepare_output_tools` below.
            fn_cap = run_capability

            async def _dispatch_prepare_tools(
                ctx: RunContext[AgentDepsT], tool_defs: list[ToolDefinition]
            ) -> list[ToolDefinition]:
                return await fn_cap.prepare_tools(ctx, tool_defs)

            toolset = PreparedToolset(toolset, _dispatch_prepare_tools)

            # Capability wrapper toolsets (including ToolSearch and CodeMode) are
            # applied here via get_wrapper_toolset, around the prepare_tools wrap above.
            toolset = run_capability.get_wrapper_toolset(toolset) or toolset

        output_toolset = output_toolset if _utils.is_set(output_toolset) else self._output_toolset
        if output_toolset is not None:
            if run_capability is not None:
                # Dispatch the new `prepare_output_tools` capability hook through a `PreparedToolset`
                # wrapped around the output toolset specifically — so the hook only sees output
                # tools, and the filtered/modified defs flow into `ToolManager.tools` and the model
                # request parameters together. Override `ctx.max_retries` to the agent's output
                # retry budget (matches `_build_output_run_context`'s contract — see https://github.com/pydantic/pydantic-ai/issues/4745).
                # `output_toolset.max_retries` is set to `max_output_retries` at agent construction.
                output_cap = run_capability
                effective_max_output_retries = (
                    max_output_retries if max_output_retries is not None else self._max_output_retries
                )

                async def _dispatch_prepare_output_tools(
                    ctx: RunContext[AgentDepsT], tool_defs: list[ToolDefinition]
                ) -> list[ToolDefinition]:
                    output_ctx = replace(ctx, max_retries=effective_max_output_retries)
                    return await output_cap.prepare_output_tools(output_ctx, tool_defs)

                output_toolset = PreparedToolset(output_toolset, _dispatch_prepare_output_tools)
            toolset = CombinedToolset([output_toolset, toolset])

        return toolset

    @property
    def root_capability(self) -> CombinedCapability[AgentDepsT]:
        """The root capability of the agent, containing all registered capabilities."""
        return self._root_capability

    @property
    def toolsets(self) -> Sequence[AbstractToolset[AgentDepsT]]:
        """All toolsets registered on the agent, including a function toolset holding tools that were registered on the agent directly.

        Output tools are not included.
        """
        return self._build_toolset_list()

    @property
    def _construction_toolsets(self) -> Sequence[AbstractToolset[AgentDepsT]]:
        """The toolsets this agent was built with, ignoring anything added afterwards.

        Read by `pydantic_ai.durable_exec` through `construction_toolsets`, which is where the
        reason it exists is written down.
        """
        return self._build_toolset_list(ignore_overrides=True)

    def _build_toolset_list(
        self,
        cap_toolsets: Sequence[AgentToolset[AgentDepsT]] | None = None,
        *,
        ignore_overrides: bool = False,
    ) -> list[AbstractToolset[AgentDepsT]]:
        """Build the list of toolsets, optionally with per-run capability toolsets.

        With `ignore_overrides`, active `override(tools=...)`/`override(toolsets=...)` values and
        dynamic toolsets added with `@agent.toolset` after construction are skipped, so the result
        is what the agent was built with. See
        `Agent._construction_toolsets`.
        """
        toolsets: list[AbstractToolset[AgentDepsT]] = []

        if not ignore_overrides and (some_tools := self._override_tools.get()):
            # `max_retries=None` for the same reason as the agent's own function toolset: the
            # tool-retry default rides `ToolManager.default_max_retries` rather than being baked here.
            function_toolset = _AgentFunctionToolset(
                some_tools.value,
                max_retries=None,
                timeout=self._tool_timeout,
                output_schema=self._output_schema,
            )
        else:
            function_toolset = self._function_toolset
        toolsets.append(function_toolset)

        if not ignore_overrides and (some_user_toolsets := self._override_toolsets.get()):
            toolsets.extend(some_user_toolsets.value)
        else:
            toolsets.extend(self._user_toolsets)
            dynamic_toolsets = (
                self._dynamic_toolsets[: self._constructor_dynamic_toolset_count]
                if ignore_overrides
                else self._dynamic_toolsets
            )
            toolsets.extend(dynamic_toolsets)
            for cap_ts in cap_toolsets if cap_toolsets is not None else self._cap_toolsets:
                if isinstance(cap_ts, AbstractToolset):
                    toolsets.append(cap_ts)  # pyright: ignore[reportUnknownArgumentType]
                else:  # pragma: no cover
                    # `get_toolset()` always returns an `AbstractToolset`.
                    toolsets.append(DynamicToolset(cap_ts))

        return toolsets

    @overload
    def _prepare_output_schema(self, output_type: None) -> _output.OutputSchema[OutputDataT]: ...

    @overload
    def _prepare_output_schema(
        self, output_type: OutputSpec[RunOutputDataT]
    ) -> _output.OutputSchema[RunOutputDataT]: ...

    def _prepare_output_schema(self, output_type: OutputSpec[Any] | None) -> _output.OutputSchema[Any]:
        if output_type is not None:
            if self._output_validators:
                raise exceptions.UserError('Cannot set a custom run `output_type` when the agent has output validators')
            schema = _output.OutputSchema.build(output_type)
        else:
            schema = self._output_schema

        return schema

    @asynccontextmanager
    async def _resolve_realtime_session(  # noqa: C901
        self,
        model: RealtimeModel | KnownRealtimeModelName | str,
        *,
        deps: AgentDepsT = None,
        model_settings: RealtimeModelSettings | None = None,
        instructions: _instructions.AgentInstructions[AgentDepsT] = None,
        toolsets: Sequence[AbstractToolset[AgentDepsT]] | None = None,
        capabilities: Sequence[AgentCapability[AgentDepsT]] | None = None,
        usage: _usage.RunUsage | None = None,
        usage_limits: _usage.UsageLimits | None = None,
        metadata: AgentMetadata[AgentDepsT] | None = None,
        conversation_id: str | None = None,
        run_id: str | None = None,
        message_history: Sequence[_messages.ModelMessage] | None = None,
        run_lifecycle: bool = False,
    ) -> AsyncGenerator[_RealtimeSessionResolution[AgentDepsT]]:
        """Resolve the agent configuration shared by realtime sessions and WebRTC signaling.

        With `run_lifecycle`, the run-lifecycle hooks are dispatched around the resolved configuration
        as well, so that they wrap the toolset — and the session the caller opens inside them — exactly
        as `iter` does. Only [`_open_realtime_session`][pydantic_ai.agent.Agent._open_realtime_session]
        asks for that: signaling only reads back the instructions and tools a session would advertise,
        and is not itself a run.
        """
        from ..realtime import RealtimeModel, infer_realtime_model

        if not isinstance(model, RealtimeModel):
            model = infer_realtime_model(model)

        deps = self._get_deps(deps)
        # Resolved, not passed through: a session inherits the conversation it continues and mints a fresh
        # id otherwise, so telemetry and a later handoff line up with a classic run — and `'new'` means
        # "fork off this history" here too rather than becoming a literal id shared by every such session.
        conversation_id = _agent_graph.resolve_conversation_id(conversation_id, message_history)
        run_id = _agent_graph.resolve_run_id(run_id, message_history)
        max_tool_retries = self._resolve_tool_retries()
        cancellation = RunCancellation() if run_lifecycle else None
        run_context = RunContext[AgentDepsT](
            deps=deps,
            agent=self,
            model=model,
            usage=usage if usage is not None else _usage.RunUsage(),
            # `RunContext.usage_limits` is documented as always set during a run. Unlike a classic run,
            # an unset `usage_limits` enforces nothing here (limits are opt-in for a session — a whole
            # conversation, where the classic 50-request default would cut a long voice conversation
            # short), so the context carries an explicitly limitless `UsageLimits` rather than the
            # classic default: what hooks read must match what the session enforces.
            usage_limits=usage_limits if usage_limits is not None else _usage.UsageLimits(request_limit=None),
            model_settings=None,
            conversation_id=conversation_id,
            run_id=run_id,
            # Seed `ctx.messages` from `message_history` like `iter` does, so dynamic `@agent.instructions`
            # functions and capability `for_run` hooks see the prior conversation. KEEP IN SYNC with `iter`.
            messages=list(message_history) if message_history else [],
            max_retries=max_tool_retries,
        )
        # Both need the context that only exists once it's built, so they're assigned rather than passed —
        # the graph does the same via `replace`. Without them a tool validated in a session sees a
        # different Pydantic context, and a capability introspecting the chain sees none, than in a run.
        run_context.validation_context = _agent_graph.build_validation_context(self._validation_context, run_context)

        # Instrumentation: inject an `Instrumentation` capability (outermost) so tool spans flow through
        # `ToolManager.handle_call`'s `wrap_tool_execute` hook — the single, canonical source of tool
        # spans — exactly as `run`/`iter` do (via `_resolve_run_capabilities` below). A realtime model is
        # a `RealtimeModel`, never an `InstrumentedModel`, so there's no wrapped model to unwrap; the
        # settings come straight from `_resolve_instrumentation_settings()`. The helper skips injection if
        # the user already supplied an `Instrumentation` capability (agent- or call-level).
        extra_capabilities = self._bind_run_capabilities(wrap_capability_funcs(capabilities))
        instrumentation_settings = self._resolve_instrumentation_settings()
        instrumentation_cap = (
            InstrumentationCap(settings=instrumentation_settings) if instrumentation_settings is not None else None
        )

        # The session-level `realtime` span and per-response `chat` spans are hand-managed by
        # `RealtimeSession`; run-lifecycle hooks have no per-exchange boundary on which to build them.
        # Drive them from the settings that
        # will actually win: an explicit `Instrumentation` capability's (agent- or call-level) over the
        # `instrument=`-derived ones, matching the precedence `_resolve_run_capabilities` applies to the
        # tool spans.
        # The winner is *selected* here, not combined: `combine` settles duplicates *within* one
        # layer and never runs across the agent-to-run boundary, so the session reads the
        # configuration off the instance `_resolve_run_capabilities` would keep -- the last
        # explicit `Instrumentation` in application order (a call-level one supersedes the
        # agent-level one, and within a layer the guarded combine resolves to the last instance's
        # values). Taking the first match would drive the session spans from settings the
        # effective configuration had already turned off. Id guards apply once, where the layers
        # actually combine.
        instrumentation_layers = [self._effective_root_capability(), *extra_capabilities]
        explicit_instrumentations = [
            leaf
            for layer in instrumentation_layers
            for leaf in leaf_capabilities(layer)
            if isinstance(leaf, InstrumentationCap)
        ]
        explicit_instrumentation = explicit_instrumentations[-1] if explicit_instrumentations else None
        session_instrumentation_settings = (
            explicit_instrumentation.settings if explicit_instrumentation is not None else instrumentation_settings
        )
        # Mirror `iter`'s `RunContext`: expose the resolved tracer (a `NoOpTracer` when uninstrumented)
        # and content-tracing flag, and resolve metadata before `for_run` so capability/toolset hooks see
        # it (same ordering as the graph run).
        run_context.tracer = (
            session_instrumentation_settings.tracer if session_instrumentation_settings is not None else NoOpTracer()
        )
        run_context.trace_include_content = (
            session_instrumentation_settings is not None and session_instrumentation_settings.include_content
        )
        if session_instrumentation_settings is not None:
            run_context.instrumentation_version = session_instrumentation_settings.version
        run_context.metadata = self._get_metadata(run_context, metadata)

        # Resolve the capability layers and extract their contributions, exactly as `run`/`iter` do via
        # the shared helpers (`_base_run_capability` honors `override(root_capability=...)` the same way).
        # Realtime keeps its own surroundings: no `InstrumentedModel` unwrap, once-only model settings
        # (below), and the `_keep_native` drop plus the shared native ↔ local-tool swap (below). Keep
        # this in sync with the `iter` call site.
        base_capability, base_is_override = self._base_run_capability()
        resolved_caps = await self._resolve_run_capabilities(
            run_context,
            base_capability=base_capability,
            extra_capabilities=extra_capabilities,
            instrumentation_cap=instrumentation_cap,
            inject_deferred_loader=True,
            base_is_override=base_is_override,
        )
        run_capability = resolved_caps.run_capability
        # Deferred capabilities load in a session the same way they do in a graph run: the catalog
        # renders into the connect-time instructions and the loaded instructions come back as the
        # `load_capability` tool's own result, which every provider supports. What no provider
        # connection can do (yet) is advertise NEW tools mid-session — tools are fixed at connect —
        # so a deferred capability whose loading would have to reveal tools can't be honored, and
        # fails up front rather than silently loading less than it promised. The direct hook calls
        # probe the same contributions extraction reads; their results are discarded.
        undeliverable_capability_ids = [
            capability_id
            for capability_id, capability in run_context.capabilities.items()
            if capability.defer_loading is True
            and (capability.get_toolset() is not None or (capability.get_native_tools() or ()))
        ]
        if undeliverable_capability_ids:
            formatted_ids = ', '.join(repr(capability_id) for capability_id in undeliverable_capability_ids)
            raise exceptions.UserError(
                'Realtime sessions cannot reveal tools mid-session, so deferred capabilities that '
                'contribute tools or native tools are not supported; remove `defer_loading=True` '
                f'from: {formatted_ids}.'
            )
        # `_resolve_run_capabilities` already registered `run_context.capabilities` for the toolset/connect
        # `for_run` below, exactly as the graph run relies on. The root of that chain has to be published
        # too, or the context contradicts itself: `capabilities` populated while `root_capability` is
        # `None`, so anything delegating through the effective chain sees none of it.
        run_context.root_capability = run_capability
        # The state an `AgentRunResult` is built from when the session closes. `run_id` and
        # `conversation_id` are the ones already resolved above, so the result, the run context, and
        # every message the session stamps all agree.
        result_state = _agent_graph.GraphAgentState(
            message_history=[],
            usage=run_context.usage,
            run_id=run_id,
            conversation_id=conversation_id,
            metadata=run_context.metadata,
        )

        # Regular agent and capability model settings intentionally do not apply to realtime sessions.
        # A future capability hook dedicated to realtime settings can add that behavior deliberately.
        effective_model_settings: RealtimeModelSettings | None = model.settings.copy() if model.settings else None
        if model_settings:
            if effective_model_settings is None:
                effective_model_settings = model_settings.copy()
            else:
                effective_model_settings.update(model_settings)
        # Realtime settings are fixed at connect time, so the merged settings hold for the whole
        # session — unlike a classic run, where this is re-stamped before each model request.
        run_context.model_settings = effective_model_settings

        # Native (provider built-in) tools, e.g. via `capabilities=[NativeTool(WebSearchTool())]`. Dynamic
        # native-tool functions are resolved once against the connect-time context — like dynamic
        # instructions, since a session's tool list is fixed from the moment the connection opens. The
        # auto-injected optional `ToolSearchTool` is dropped (mirroring the graph's corpus-empty drop):
        # there's no tool-search corpus and realtime providers don't support it. The helper already
        # folded in `override(native_tools=...)` and any per-call capability native tools.
        # KEEP IN SYNC with the graph's resolution in `_prepare_request_parameters`.
        native_tools: list[AbstractNativeTool] = []
        for native_tool in resolved_caps.native_tools:
            if not isinstance(native_tool, AbstractNativeTool):
                resolved_native = native_tool(run_context)
                if inspect.isawaitable(resolved_native):
                    resolved_native = await resolved_native
                if resolved_native is None:
                    continue
                native_tool = resolved_native
            if not (isinstance(native_tool, ToolSearchTool) and native_tool.optional):
                native_tools.append(native_tool)
        model_profile = model.profile

        toolset = self._get_toolset(
            output_toolset=None,
            additional_toolsets=toolsets,
            cap_toolsets=resolved_caps.toolsets,
            run_capability=run_capability,
        )
        toolset = await toolset.for_run(run_context)

        # The hooks run before the session exists — and a `wrap_run` that short-circuits means it never
        # will — so the session and the result standing in for it are handed back through this holder.
        lifecycle_state = _RealtimeSessionLifecycle() if run_lifecycle else None

        def _build_session_result() -> AgentRunResult[Any]:
            assert lifecycle_state is not None and lifecycle_state.session is not None
            return lifecycle_state.session._build_run_result(result_state)  # pyright: ignore[reportPrivateUsage]

        async def _finalize_session_result(result: AgentRunResult[Any]) -> None:
            assert lifecycle_state is not None
            if cancellation is not None and cancellation.cancel_requested:
                raise asyncio.CancelledError('pydantic-ai: re-asserting a requested run cancellation')
            if lifecycle_state.session is None:
                lifecycle_state.short_result = result
            else:
                lifecycle_state.session._result = result  # pyright: ignore[reportPrivateUsage]

        @asynccontextmanager
        async def _translate_cancellation() -> AsyncGenerator[None]:
            assert cancellation is not None and lifecycle_state is not None

            def _run_cancelled(message: str) -> exceptions.RunCancelled:
                assert lifecycle_state is not None
                session = lifecycle_state.session
                return exceptions.RunCancelled(
                    message,
                    messages=session.all_messages() if session is not None else message_history or (),
                    new_message_index=len(message_history or ()),
                    usage=run_context.usage,
                    metadata=run_context.metadata,
                    run_id=run_id,
                    conversation_id=conversation_id,
                )

            try:
                yield
            except exceptions.RunCancelled as exc:
                # Match classic runs: a nested run carries its own history, but this session's
                # caller must receive the outer conversation it can actually resume.
                raise _run_cancelled('The agent run was cancelled by a nested run.') from exc
            except asyncio.CancelledError as exc:
                cancelled = _run_cancelled('The agent run was cancelled.')
                if cancellation.resolve():
                    raise cancelled from exc
                cancelled._attach_to(exc)  # pyright: ignore[reportPrivateUsage]
                raise
            finally:
                cancellation.release_issued()

        yielded = False
        async with AsyncExitStack() as session_stack:
            if lifecycle_state is not None:
                assert cancellation is not None
                # Setup-time `for_run` callbacks have the same context contract as a classic run:
                # cancellation becomes available only once the run lifecycle has an owning task.
                run_context._cancellation = cancellation  # pyright: ignore[reportPrivateUsage]
                await session_stack.enter_async_context(_translate_cancellation())
                cancellation.bind()
                session_stack.callback(cancellation.finish)
                lifecycle = await session_stack.enter_async_context(
                    _run_lifecycle_hooks(
                        run_capability,
                        run_context,
                        build_result=_build_session_result,
                        finalize=_finalize_session_result,
                    )
                )
                if lifecycle.short_circuited:
                    # Nothing below this is resolved: the toolset is never entered and the caller
                    # yields a closed session in place of connecting. The empty `ToolManager` is the
                    # one that session is built on. The `yielded` guard mirrors the normal path: if the
                    # caller's body raises and the lifecycle hooks recover it (`on_run_error` runs even
                    # after a short-circuit), the error is suppressed at the exit stack and execution
                    # resumes below — without this, that clean resume would yield a second time and
                    # `asynccontextmanager` would raise `RuntimeError: generator didn't stop after athrow()`.
                    try:
                        yield _RealtimeSessionResolution(
                            model=model,
                            run_context=run_context,
                            tool_manager=ToolManager(FunctionToolset()),
                            model_request_parameters=models.ModelRequestParameters(),
                            model_settings=effective_model_settings,
                            instructions=None,
                            request_messages=[],
                            model_profile=model_profile,
                            instrumentation_settings=session_instrumentation_settings,
                            conversation_id=conversation_id,
                            run_id=run_id,
                            lifecycle=lifecycle_state,
                            short_circuited=True,
                        )
                    finally:
                        yielded = True
                    return
            await session_stack.enter_async_context(toolset)
            tool_manager = await ToolManager[AgentDepsT](
                toolset, root_capability=run_capability, default_max_retries=max_tool_retries
            ).for_run_step(run_context)
            tool_defs = tool_manager.tool_defs

            # Resolve authored instructions, then fold in toolset-contributed
            # instructions, mirroring the run/iter graph. Capability-contributed instructions come from
            # the resolved capabilities (like `iter`), not just the init-time snapshot.
            sourced_instructions = self._get_instructions(
                additional_instructions=instructions, cap_instructions=resolved_caps.instructions
            )
            # Build `InstructionPart`s (static parts first, then dynamic functions, then dynamic toolset
            # instructions) and join with the canonical `InstructionPart.join` — same double-newline
            # separator, static-before-dynamic ordering, and per-source `id`s as the graph run.
            # KEEP IN SYNC with the graph's `_get_instructions` / `ModelRequestNode`.
            instruction_parts = await _instructions.resolve_sourced_instructions(sourced_instructions, run_context)
            instruction_parts.extend(await collect_toolset_instructions(tool_manager.toolset, run_context))
            resolved_instructions = _messages.InstructionPart.join(_messages.InstructionPart.sorted(instruction_parts))
            request_messages = [
                *(message_history or ()),
                _messages.ModelRequest(parts=[], instructions=resolved_instructions or None),
            ]
            model_request_parameters = models.ModelRequestParameters(
                function_tools=tool_defs,
                native_tools=native_tools,
            )
            # Resolve `include_return_schema` exactly as `Model.prepare_request` does: return schemas
            # are cleared on tools that didn't opt in, kept as-is for a model that renders them
            # natively (Gemini Live's function-declaration `response` schema), and injected into the
            # tool description elsewhere.
            model_request_parameters = models.prepare_return_schemas(
                model_request_parameters,
                supports_tool_return_schema=model_profile.get('supports_tool_return_schema', False),
            )
            # Run the same native ↔ local-tool fallback swap the classic agent-run path applies (via
            # `Model._resolve_request_tools`): drop an unsupported native tool when a local fallback
            # (stamped `unless_native=...` by the capability's toolset) is present, drop the redundant
            # local tool when the native tool IS supported, and raise the shared `UserError` (suggesting
            # `local=...`) only when unsupported with no local fallback. Realtime models genuinely default
            # to supporting no native tools, so the default here is `frozenset()`, not `SUPPORTED_NATIVE_TOOLS`.
            # No `can_withhold_tool_schemas`/`tool_addition_mode`: a realtime connection can neither
            # withhold a schema nor reveal a tool later, so every deferred unrevealed tool resolves
            # to `'withheld'` in the visibility table.
            model_request_parameters = models.resolve_request_tools(
                model_request_parameters, model_profile.get('supported_native_tools', frozenset())
            )
            # A `defer_loading=True` tool is hidden until tool search reveals it, which a session whose
            # tools are fixed at connect can never do — the model would be handed a `search_tools`
            # affordance that finds the tool and then can't have it. Rejected up front for the same
            # reason a deferred *capability* that contributes tools is (see `_resolve_run_capabilities`
            # above), rather than silently advertising a search that leads nowhere.
            deferred_tool_names = [tool.name for tool in model_request_parameters.function_tools if tool.defer_loading]
            if deferred_tool_names:
                formatted_names = ', '.join(repr(name) for name in deferred_tool_names)
                raise exceptions.UserError(
                    'Realtime sessions cannot reveal tools mid-session, so tools with '
                    f'`defer_loading=True` are not supported; remove it from: {formatted_names}.'
                )
            # Realtime codecs read `function_tools` directly, so hidden tools are dropped from the
            # connect-time advertisement entirely — the same physical removal this path applied
            # before visibility became a resolved table.
            model_request_parameters = dataclasses.replace(
                model_request_parameters,
                function_tools=model_request_parameters.declared_function_tools,
            )

            wrap_event_stream: (
                Callable[
                    [AsyncIterable[_messages.AgentStreamEvent]],
                    AsyncIterable[_messages.AgentStreamEvent],
                ]
                | None
            ) = (
                (
                    lambda stream: run_capability.wrap_run_event_stream(
                        run_context, stream=dispatch_event_stream(run_context, stream)
                    )
                )
                if run_capability.has_wrap_run_event_stream or run_capability.has_on_event
                else None
            )

            resolution = _RealtimeSessionResolution(
                model=model,
                run_context=run_context,
                tool_manager=tool_manager,
                model_request_parameters=model_request_parameters,
                model_settings=effective_model_settings,
                instructions=resolved_instructions or None,
                request_messages=request_messages,
                model_profile=model_profile,
                instrumentation_settings=session_instrumentation_settings,
                conversation_id=conversation_id,
                run_id=run_id,
                wrap_event_stream=wrap_event_stream,
                lifecycle=lifecycle_state,
            )
            try:
                yield resolution
            finally:
                yielded = True

        if yielded:
            # Reached on the normal path, and when the run-lifecycle hooks suppressed a caller-side
            # error that `wrap_run`/`on_run_error` recovered — the caller sees a clean exit either way.
            return
        # Entering the toolset or resolving the session configuration failed after the run-lifecycle
        # hooks were entered, and `wrap_run`/`on_run_error` recovered with a result: the hooks
        # suppressed the error above with nothing yielded yet, which `asynccontextmanager` would
        # report as `RuntimeError: generator didn't yield`. Yield the short-circuit resolution shape
        # instead, so `_open_realtime_session` yields its closed placeholder session carrying the
        # recovery result.
        assert lifecycle_state is not None and lifecycle_state.short_result is not None
        yield _RealtimeSessionResolution(
            model=model,
            run_context=run_context,
            tool_manager=ToolManager(FunctionToolset()),
            model_request_parameters=models.ModelRequestParameters(),
            model_settings=effective_model_settings,
            instructions=None,
            request_messages=[],
            model_profile=model_profile,
            instrumentation_settings=session_instrumentation_settings,
            conversation_id=conversation_id,
            run_id=run_id,
            lifecycle=lifecycle_state,
            short_circuited=True,
        )

    @asynccontextmanager
    async def _open_realtime_session(
        self,
        model: RealtimeModel | KnownRealtimeModelName | str,
        *,
        deps: AgentDepsT = None,
        model_settings: RealtimeModelSettings | None = None,
        instructions: _instructions.AgentInstructions[AgentDepsT] = None,
        toolsets: Sequence[AbstractToolset[AgentDepsT]] | None = None,
        capabilities: Sequence[AgentCapability[AgentDepsT]] | None = None,
        usage: _usage.RunUsage | None = None,
        usage_limits: _usage.UsageLimits | None = None,
        metadata: AgentMetadata[AgentDepsT] | None = None,
        conversation_id: str | None = None,
        run_id: str | None = None,
        message_history: Sequence[_messages.ModelMessage] | None = None,
        audio_retention: AudioRetention = 'transcript_only',
        handle_barge_in: bool = False,
        retain_images_every_n: int = 1,
        retain_images_max: int | None = 100,
        provider_session: RealtimeProviderSession | None = None,
    ) -> AsyncGenerator[RealtimeSession]:
        """Worker behind [`AgentRealtime.session`][pydantic_ai.agent.AgentRealtime.session].

        Opens the realtime session and drives the agent's tools. Users go through
        [`agent.realtime(model).session()`][pydantic_ai.agent.AbstractAgent.realtime]; see
        [`realtime`][pydantic_ai.agent.AbstractAgent.realtime] for the parameter reference.
        """
        from ..realtime import RealtimeSession
        from ..realtime.codec import RealtimeCodecEvent, RealtimeConnection, RealtimeInput

        # A WebRTC sideband session doesn't own the audio transport: the browser streams audio to the
        # provider directly, so the session disables its audio methods and retains no audio bytes.
        owns_media = provider_session is None
        if not owns_media and audio_retention != 'transcript_only':
            # Reject an audio-retention request that can never be satisfied (no audio bytes flow here).
            raise exceptions.UserError(
                "A WebRTC sideband session can't retain audio: the browser exchanges audio with the "
                'provider directly, so no audio bytes reach this connection. Leave `audio_retention` at '
                "'transcript_only' (transcripts still build the conversation history)."
            )

        yielded = False
        async with self._resolve_realtime_session(
            model,
            deps=deps,
            model_settings=model_settings,
            instructions=instructions,
            toolsets=toolsets,
            capabilities=capabilities,
            usage=usage,
            usage_limits=usage_limits,
            metadata=metadata,
            conversation_id=conversation_id,
            run_id=run_id,
            message_history=message_history,
            run_lifecycle=True,
        ) as resolved:
            lifecycle = resolved.lifecycle
            assert lifecycle is not None

            def _closed_session(result: AgentRunResult[Any]) -> RealtimeSession:
                # The session is yielded closed, so `_ensure_not_closed` rejects every public
                # entry point before the connection is touched; these raises are defense-in-depth
                # for the abstract methods the base class requires.
                class _SkippedRealtimeConnection(RealtimeConnection):
                    async def send(self, content: RealtimeInput) -> None:
                        raise exceptions.UserError(  # pragma: no cover
                            'This realtime session was short-circuited before connecting.'
                        )

                    def __aiter__(self) -> AsyncIterator[RealtimeCodecEvent]:
                        raise exceptions.UserError(  # pragma: no cover
                            'This realtime session was short-circuited before connecting.'
                        )

                session = RealtimeSession(
                    _SkippedRealtimeConnection(),
                    model=resolved.model,
                    tool_manager=resolved.tool_manager,
                    usage=resolved.run_context.usage,
                    usage_limits=usage_limits,
                    message_history=message_history,
                    conversation_id=resolved.conversation_id,
                    run_id=resolved.run_id,
                    metadata=resolved.run_context.metadata,
                )
                session._result = result  # pyright: ignore[reportPrivateUsage]
                session._closed = True  # pyright: ignore[reportPrivateUsage]
                return session

            if resolved.short_circuited:
                assert lifecycle.short_result is not None
                # `yielded` guard as in the normal path below: if the caller's body raises and the
                # resolver's lifecycle hooks recover it, the error is suppressed at the `async with`
                # above and execution resumes below. Without this, that resume would yield a second
                # closed session and `asynccontextmanager` would raise `generator didn't stop after athrow()`.
                try:
                    yield _closed_session(lifecycle.short_result)
                finally:
                    yielded = True
                return

            if message_history and not resolved.model_profile.get('supports_session_seeding', False):
                raise exceptions.UserError(
                    f'The {resolved.model.model_name!r} realtime model does not support seeding a session with '
                    '`message_history`.'
                )

            output_modality = (resolved.model_settings or {}).get('output_modality', 'audio')
            # Unlike a setting the provider merely ignores, this one changes what the caller gets back:
            # a model that can't do it either fails the handshake (Gemini) or answers with speech anyway
            # (xAI), so it is rejected here rather than after a connection is open.
            if output_modality == 'text' and not resolved.model_profile.get('supports_text_output', True):
                raise exceptions.UserError(
                    f"The {resolved.model.model_name!r} realtime model does not support `output_modality='text'`; "
                    'it only generates audio. Read the spoken answer from the transcript on the '
                    '`SpeechPart` instead.'
                )

            connection_manager = (
                resolved.model.connect(
                    messages=resolved.request_messages,
                    model_settings=resolved.model_settings,
                    model_request_parameters=resolved.model_request_parameters,
                )
                if provider_session is None
                else resolved.model.connect_webrtc(
                    provider_session,
                    messages=resolved.request_messages,
                    model_settings=resolved.model_settings,
                    model_request_parameters=resolved.model_request_parameters,
                )
            )
            async with connection_manager as connection:
                session = RealtimeSession(
                    connection,
                    model=resolved.model,
                    tool_manager=resolved.tool_manager,
                    owns_media=owns_media,
                    instrumentation=resolved.instrumentation_settings,
                    # Fall back to 'agent' like the classic run span (see `capabilities/instrumentation.py`)
                    # so the session span always carries an `agent_name`; backends that group runs by it
                    # (e.g. Logfire's Runs view) would otherwise skip an unnamed agent's realtime session.
                    agent_name=self.name or 'agent',
                    usage=resolved.run_context.usage,
                    usage_limits=usage_limits,
                    audio_retention=audio_retention,
                    handle_barge_in=handle_barge_in,
                    retain_images_every_n=retain_images_every_n,
                    retain_images_max=retain_images_max,
                    message_history=message_history,
                    conversation_id=resolved.conversation_id,
                    run_id=resolved.run_id,
                    instructions=resolved.instructions,
                    metadata=resolved.run_context.metadata,
                    agent_description=(
                        self.render_description(resolved.run_context.deps)
                        if resolved.instrumentation_settings is not None
                        else None
                    ),
                    output_modality=output_modality,
                    # Surfaced on the session span so the session's configured native tools and realtime
                    # settings are inspectable, respecting `include_model_request_parameters`.
                    model_request_parameters=resolved.model_request_parameters,
                    model_settings=resolved.model_settings,
                    wrap_event_stream=resolved.wrap_event_stream,
                )
                lifecycle.session = session
                resolved.run_context.realtime_session = session
                async with session:
                    try:
                        yield session
                    finally:
                        yielded = True

        if yielded:
            # Reached on the normal path, and when the run-lifecycle hooks suppressed a session error
            # that `wrap_run`/`on_run_error` recovered — the caller sees a clean exit either way.
            return
        # Opening the connection or building the session failed and `wrap_run`/`on_run_error`
        # recovered with a result: the resolver's lifecycle hooks suppressed the error with nothing
        # yielded yet, which `asynccontextmanager` would report as `RuntimeError: generator didn't
        # yield`. Yield the short-circuit path's closed placeholder carrying the recovery result.
        assert lifecycle.short_result is not None
        yield _closed_session(lifecycle.short_result)

    async def __aenter__(self) -> Self:
        """Enter the agent context.

        This will start all [`MCPToolset`s][pydantic_ai.mcp.MCPToolset] registered as `toolsets` so they are ready to be used,
        and enter the model so the provider's HTTP client will be closed cleanly on exit.

        This is a no-op if the agent has already been entered.
        """
        async with self._enter_lock:
            if self._entered_count == 0:
                async with AsyncExitStack() as exit_stack:
                    toolset = self._get_toolset()
                    await exit_stack.enter_async_context(toolset)

                    capability = self._effective_root_capability()
                    capability_model = capability.get_model()
                    override_model = self._override_model.get()
                    if override_model is not None:
                        static_selection = override_model.value
                    elif callable(capability_model) and not _is_model(capability_model):
                        # Dynamic capability models are entered by the run that selects them.
                        static_selection = None
                    elif capability_model is not None:
                        static_selection = capability_model
                    else:
                        static_selection = self.model
                    if static_selection is not None and not (
                        isinstance(static_selection, str) and capability.has_resolve_model_id
                    ):
                        model = (
                            static_selection if _is_model(static_selection) else models.infer_model(static_selection)
                        )
                        await exit_stack.enter_async_context(model)
                        self._entered_model_ids.add(id(model))
                        if isinstance(static_selection, str):
                            self._entered_models_by_selection[id(capability), static_selection] = model

                    self._exit_stack = exit_stack.pop_all()
            self._entered_count += 1
        return self

    async def __aexit__(self, *args: Any) -> bool | None:
        async with self._enter_lock:
            self._entered_count -= 1
            if self._entered_count == 0 and self._exit_stack is not None:
                try:
                    await self._exit_stack.aclose()
                finally:
                    self._exit_stack = None
                    self._entered_model_ids.clear()
                    self._entered_models_by_selection.clear()

    def set_mcp_sampling_model(self, model: models.Model | models.KnownModelName | str | None = None) -> None:
        """Set the sampling model on all [`MCPToolset`s][pydantic_ai.mcp.MCPToolset] registered with the agent.

        If no sampling model is provided, the agent's model will be used.
        """
        try:
            sampling_model = models.infer_model(model) if model else self._get_model_outside_run()
        except exceptions.UserError as e:
            capability = self._effective_root_capability()
            if model is None and (callable(capability.get_model()) or capability.has_resolve_model_id):
                raise exceptions.UserError(
                    'The capability model requires run dependencies and cannot be used for MCP sampling setup. '
                    'Pass a concrete model explicitly.'
                ) from e
            raise exceptions.UserError('No sampling model provided and no model set on the agent.') from e

        from ..mcp import MCPToolset

        def _set_sampling_model(toolset: AbstractToolset[AgentDepsT]) -> None:
            if isinstance(toolset, MCPToolset):
                toolset.set_sampling_model(sampling_model)

        self._get_toolset().apply(_set_sampling_model)

    def to_web(
        self,
        *,
        models: ModelsParam = None,
        deps: AgentDepsT = None,
        model_settings: ModelSettings | None = None,
        instructions: str | None = None,
        html_source: str | Path | None = None,
        allowed_hosts: Sequence[str] | None = None,
    ) -> Starlette:
        """Create a Starlette app that serves a web chat UI for this agent.

        This method returns a pre-configured Starlette application that provides a web-based
        chat interface for interacting with the agent. By default, the UI is fetched from a
        CDN and cached on first use.

        The returned Starlette application can be mounted into a FastAPI app or run directly
        with any ASGI server (uvicorn, hypercorn, etc.).

        Note that the `deps` and `model_settings` will be the same for each request.
        To provide different `deps` for each request use the lower-level adapters directly.

        The agent's configured native tools (registered via `capabilities=[NativeTool(...)]`
        or higher-level capabilities like `WebSearch()`) are automatically exposed as
        options in the UI.

        Args:
            models: Additional models to make available in the UI. Can be:
                - A sequence of model names/instances (e.g., `['openai:gpt-5', 'anthropic:claude-sonnet-4-6']`)
                - A dict mapping display labels to model names/instances
                  (e.g., `{'GPT 5': 'openai:gpt-5', 'Claude': 'anthropic:claude-sonnet-4-6'}`)
                The agent's model is always included. Native tool support is automatically
                determined from each model's profile.
            deps: Optional dependencies to use for all requests.
            model_settings: Optional settings to use for all model requests.
            instructions: Optional extra instructions to pass to each agent run.
            html_source: Path or URL for the chat UI HTML. Can be:
                - None (default): Fetches from CDN and caches locally
                - A Path instance: Reads from the local file
                - A URL string (http:// or https://): Fetches from the URL
                - A file path string: Reads from the local file
            allowed_hosts: Additional hostnames to answer to, e.g. `['ui.example.com']` or
                `['*.example.com']` (subdomains only, so list the apex separately if you serve it).
                IP addresses and `localhost` are always allowed; any other `Host` header is refused
                with a `421`, so that a website cannot reach the UI on your machine by pointing a
                hostname it controls at you (DNS rebinding). Pass `['*']` to answer to any host,
                only if something in front of the app already authenticates requests.

        Returns:
            A configured Starlette application ready to be served (e.g., with uvicorn)

        Example:
            ```python
            from pydantic_ai import Agent
            from pydantic_ai.capabilities import NativeTool
            from pydantic_ai.native_tools import WebSearchTool

            agent = Agent('openai:gpt-5', capabilities=[NativeTool(WebSearchTool())])

            # Simple usage - uses agent's model and native tools
            app = agent.to_web()

            # Or provide additional models for UI selection
            app = agent.to_web(models=['openai:gpt-5', 'anthropic:claude-sonnet-4-6'])

            # Then run with: uvicorn app:app --reload
            ```
        """
        from ..ui._web import create_web_app

        return create_web_app(
            self,
            models=models,
            deps=deps,
            model_settings=model_settings,
            instructions=instructions,
            html_source=html_source,
            allowed_hosts=allowed_hosts,
        )


@dataclasses.dataclass
class _RunModelResources:
    """Enter every model selected by a run on its shared resource stack."""

    entered_model_ids: set[int]
    _stack: AsyncExitStack | None = dataclasses.field(default=None, init=False, repr=False)

    def bind_stack(self, stack: AsyncExitStack) -> None:
        assert self._stack is None
        self._stack = stack

    async def enter_model(self, selected_model: models.Model) -> None:
        model_identity = id(selected_model)
        if model_identity in self.entered_model_ids:
            return
        assert self._stack is not None
        await self._stack.enter_async_context(selected_model)
        self.entered_model_ids.add(model_identity)


@dataclasses.dataclass
class _PreparedAgentRun(Generic[_PreparedDepsT, _PreparedOutputT]):
    """The fully assembled inputs and resources for one graph-based agent run."""

    graph: Graph[
        _agent_graph.GraphAgentState,
        _agent_graph.GraphAgentDeps[_PreparedDepsT, _PreparedOutputT],
        _agent_graph.UserPromptNode[_PreparedDepsT, _PreparedOutputT],
        _result.FinalResult[_PreparedOutputT],
    ]
    state: _agent_graph.GraphAgentState
    graph_deps: _agent_graph.GraphAgentDeps[_PreparedDepsT, _PreparedOutputT]
    user_prompt_node: _agent_graph.UserPromptNode[_PreparedDepsT, _PreparedOutputT]
    agent_name: str
    binding: RunBinding | None
    cancellation_token: CancellationToken | None
    model: models.Model
    capability_owns_current_model: bool
    model_resources: _RunModelResources
    run_capability: AbstractCapability[_PreparedDepsT]
    toolset: AbstractToolset[_PreparedDepsT]
    usage_limits: _usage.UsageLimits
    concurrency_limiter: _concurrency.AbstractConcurrencyLimiter | None
    resolve_metadata: Callable[
        [GraphRunContext[_agent_graph.GraphAgentState, _agent_graph.GraphAgentDeps[_PreparedDepsT, _PreparedOutputT]]],
        dict[str, Any] | None,
    ]

    @asynccontextmanager
    async def open(self) -> AsyncGenerator[AgentRun[_PreparedDepsT, _PreparedOutputT]]:
        graph_deps = self.graph_deps
        state = self.state
        pending_message_queue = state.pending_messages
        assert isinstance(pending_message_queue, _enqueue.PendingMessageQueue)

        @asynccontextmanager
        async def _translate_cancellation() -> AsyncGenerator[None]:
            def _run_cancelled(message: str) -> exceptions.RunCancelled:
                return _agent_graph.run_cancelled_snapshot(message, state, graph_deps)

            try:
                yield
            except exceptions.RunCancelled as exc:
                # A `RunCancelled` reaching this run's outer edge from below — e.g. a delegate tool
                # awaited a sub-agent run that cancelled itself via `cancel()` — carries the
                # *nested* run's history, not this run's. Presenting it to this run's caller
                # unchanged would make `RunCancelled.all_messages()` (and a resume from it) silently
                # use the wrong conversation. Re-stamp it with this run's state, keeping the nested
                # cancellation as the cause. Whether a nested cancellation should terminate this run
                # at all, or be isolated as a tool failure, is a separate semantics question tracked
                # in https://github.com/pydantic/pydantic-ai/issues/7199.
                raise _run_cancelled('The agent run was cancelled by a nested run.') from exc
            except asyncio.CancelledError as exc:
                first_party = graph_deps.cancellation.resolve()
                if first_party:
                    raise _run_cancelled('The agent run was cancelled.') from exc
                # An external cancellation must keep propagating as `CancelledError`, but the run
                # state rides along on the exception instance for `RunCancelled.from_cancellation()`.
                # Nested runs attach to the same propagating exception; the outermost run attaches
                # last and wins, giving its awaiter the outer run's history.
                _run_cancelled('The agent run was cancelled by an external asyncio cancellation.')._attach_to(exc)  # pyright: ignore[reportPrivateUsage]
                raise
            finally:
                # On every exit path — translation above, a clean exit after user code swallowed a
                # requested cancellation, a superseded driving task, or a non-cancellation error
                # overtaking a requested cancel — an issued-but-unresolved cancellation must not
                # leak an elevated `Task.cancelling()` count past the run: it would spuriously
                # cancel unrelated later work on the task that drove the run.
                graph_deps.cancellation.release_issued()

        async with AsyncExitStack() as stack:
            # Enter first so cancellation is classified only after every other context has torn down.
            await stack.enter_async_context(_translate_cancellation())

            # Bind the run's cancellation controller to this task and register the token BEFORE any
            # potentially-blocking setup (the concurrency limiter, model entry): a run queued behind
            # the concurrency limiter must still be cancellable via its token or `cancel()`, and a
            # pre-cancelled token must prevent it from starting. `finish` neutralizes `cancel()` once
            # the run is over so it can never cancel unrelated later work on this task.
            graph_deps.cancellation.bind()
            stack.callback(graph_deps.cancellation.finish)
            # Nothing drains the queue once the graph stops, so reject later enqueues instead of
            # stranding them. A normal finish already closed it inside `drain_at_end`.
            stack.callback(pending_message_queue.close)
            if self.cancellation_token is not None:
                graph_deps.cancellation.attach_token(self.cancellation_token)

            self.model_resources.bind_stack(stack)
            await stack.enter_async_context(
                _concurrency.get_concurrency_context(self.concurrency_limiter, f'agent:{self.agent_name}')
            )
            if self.capability_owns_current_model:
                await self.model_resources.enter_model(self.model)
            graph_run = await stack.enter_async_context(
                self.graph.iter(
                    inputs=self.user_prompt_node,
                    state=state,
                    deps=graph_deps,
                    span=None,
                    infer_name=False,
                )
            )
            agent_run = AgentRun(graph_run)
            if self.binding is not None:
                # Hand the live `AgentRun` to the `AgentRunEvents` handle that started this run, so
                # its `cancel()`/run-state accessors reach the run. The controller was already bound
                # to this task above, before `wrap_run`/`before_run`.
                self.binding.agent_run = agent_run
            self.resolve_metadata(agent_run.ctx)

            # Build RunContext for run lifecycle hooks
            run_ctx = _agent_graph.build_run_context(agent_run.ctx)

            async def _finalize_result(result: AgentRunResult[Any]) -> None:
                self.usage_limits.check_cost(result.usage)
                # A first-party cancellation request remains terminal even if a hook consumed the
                # task's cancellation counter (past the helper's `raise_if_cancelling` backstop).
                if graph_deps.cancellation.cancel_requested:
                    raise asyncio.CancelledError('pydantic-ai: re-asserting a requested run cancellation')
                agent_run._result_override = result  # pyright: ignore[reportPrivateUsage]

            def _extract_error(error: BaseException) -> BaseException:
                # Use the original node error if available, since context manager __aexit__ chains
                # (GraphRun → anyio TaskGroup) may transform it into CancelledError or ExceptionGroup.
                return agent_run._node_error or error  # pyright: ignore[reportPrivateUsage]

            def _build_result() -> AgentRunResult[Any]:
                result = agent_run.result
                assert result is not None
                return result

            async with _run_lifecycle_hooks(
                self.run_capability,
                run_ctx,
                build_result=_build_result,
                finalize=_finalize_result,
                extract_error=_extract_error,
                result_ready=lambda: agent_run.result is not None,
                # Restore on `stack` in LIFO order (after toolset exit, before graph run exit).
                restore_context_on=stack,
            ):
                # Enter toolset AFTER context vars are propagated so that toolset
                # __aenter__/__aexit__ run inside the run span context.
                await stack.enter_async_context(self.toolset)
                try:
                    yield agent_run
                finally:
                    if agent_run.result is not None:
                        self.resolve_metadata(agent_run.ctx)


def _merge_retries_with_spec(
    explicit: int | AgentRetries | None,
    spec: AgentSpec,
) -> AgentRetries | None:
    """Merge an explicit `retries=` value with the retry fields on an `AgentSpec`.

    Explicit kwarg keys win over spec keys.
    """
    merged = _retry_overrides_from_spec(spec)
    merged.update(_normalize_agent_retry_overrides(explicit))
    if not merged:
        return None
    return merged


def _retry_overrides_from_spec(spec: AgentSpec) -> AgentRetries:
    """Return retry fields explicitly configured on an `AgentSpec`."""
    return _normalize_agent_retry_overrides(spec.retries) if 'retries' in spec.model_fields_set else {}


_UNSUPPORTED_SPEC_FIELDS: tuple[str, ...] = (
    'description',
    'end_strategy',
    'tool_timeout',
    'output_schema',
    'deps_schema',
)
"""AgentSpec fields that are not supported at run/override time."""

_AUTO_INJECT_CAPABILITY_TYPES: tuple[type[AbstractCapability[Any]], ...] = (
    ToolSearchCap,
    PendingMessageDrainCapability,
)
"""Infrastructure capabilities auto-injected when not already present."""


def _registered_capability_count(capability: AbstractCapability[Any]) -> int:
    """Count the capabilities the user registered, for a `pydantic_ai._display` banner.

    Infrastructure capabilities are injected for every agent, so counting those would say nothing
    about the agent the user actually wrote.
    """
    return sum(not isinstance(leaf, _AUTO_INJECT_CAPABILITY_TYPES) for leaf in leaf_capabilities(capability))


class _StartupBannerDetails(NamedTuple):
    """What a chat session can say about itself before any run has resolved anything."""

    model: str
    tools: int | None
    """`None` when the session holds a toolset whose tools can't be counted without connecting."""
    capabilities: int
    instrumented: bool


def _inject_auto_capabilities(capabilities: list[AbstractCapability[Any]]) -> None:
    """Ensure all auto-injected infrastructure capabilities are present.

    Each capability's own `CapabilityOrdering` (e.g. `position='outermost'`)
    determines its final placement, so insertion order here doesn't matter.
    """
    for cap_type in _AUTO_INJECT_CAPABILITY_TYPES:
        if not has_capability_type(capabilities, cap_type):
            capabilities.append(cap_type())


def _validate_capability_ids(capabilities: Sequence[AbstractCapability[Any]]) -> set[str]:
    """Validate capability `id`s and return the set of explicit ones.

    Rejects deferred capabilities that lack an explicit `id`, and ids shared by capabilities that
    have not said how they compose. Capabilities whose class declares a default `id` (what
    `_declares_default_id` reads off the class body -- `Thinking` declares one and overrides
    nothing) are allowed to repeat: their duplication is resolved over the whole composed tree at
    run setup by `_combine_duplicate_capabilities`, the only place `combine` is called. Rejecting
    the rest here means the common mistake still surfaces in `Agent(...)` rather than on the first
    run.

    Shared by construction-time validation over the statically-provided capabilities and run-time
    assembly, which also covers capabilities supplied per-run or returned by `for_run` and so can't
    be checked at construction.
    """
    owners: dict[str, type[AbstractCapability[Any]]] = {}
    for cap in capabilities:
        if cap.defer_loading is True and cap.id is None:
            raise exceptions.UserError(
                'Deferred capabilities must use stable explicit `id` values. '
                'Pass `id=...` when using `defer_loading=True`.'
            )
        if cap.id is None:
            continue
        _instructions.validate_instruction_id_segment(cap.id, kind='Capability id')
        # Both classes decide, not just the one that happens to come second: whether an id can
        # repeat is a property of the pair, so reading it off the later capability alone would let
        # one order through and reject the other.
        owner = owners.get(cap.id)
        if owner is not None:
            _reject_class_crossing_id(cap.id, {owner, type(cap)})
            if not _declares_default_id(type(cap)):
                raise exceptions.UserError(_repeated_id_message(cap.id))
        owners.setdefault(cap.id, type(cap))
    return set(owners)


def _validate_instruction_source_ids(capabilities: Sequence[AbstractCapability[Any]]) -> None:
    """Reject two instruction sources that would contribute parts under one `capability:<id>` key.

    `_validate_capability_ids` walks the flattened capabilities, but a `CombinedCapability` subclass
    that overrides `get_instructions` contributes as a source in its own right and is deliberately
    retained rather than splatted, so it never appears in that list. Its `id` therefore went
    unchecked, and a sibling could share it -- leaving an application unable to tell whose text
    `capability:<id>` addresses, which is the one thing the key exists to make unambiguous.

    Runs at construction, after `for_agent`, and again on the tree a run actually resolves to.
    The last of those is not redundant: a capability passed to `run()` joins the retained container
    the same way a registered sibling does, and `for_run` may hand back a capability carrying a
    different `id` than the one construction saw.

    Sources whose class declares a default `id` are exempt at construction for the same reason they
    are in `_validate_capability_ids`: the run resolves them to one source before any instructions
    are collected, so the `capability:<id>` key is unambiguous by the time it is used. The run-setup
    call sees the already-combined tree, where a surviving duplicate is a genuine conflict.
    """
    sources_by_id: dict[str, AbstractCapability[Any]] = {}
    for capability in capabilities:
        sources = (
            capability._instruction_sources  # pyright: ignore[reportPrivateUsage]
            if isinstance(capability, CombinedCapability)
            else (capability,)
        )
        for source in sources:
            if source.id is None:
                continue
            if (existing := sources_by_id.setdefault(source.id, source)) is not source and not _declares_default_id(
                type(source)
            ):
                raise exceptions.UserError(
                    f'Capability id {existing.id!r} is used by multiple capabilities that contribute '
                    'instructions. Capability ids must be unique within a run.'
                )


def _validate_native_tool_ids(native_tools: Sequence[AgentNativeTool[Any]], *, source: str) -> None:
    """Reject native tools that share a `unique_id` but carry conflicting definitions.

    Native tools are keyed by `unique_id` when request parameters are deduplicated (see
    `Model.prepare_request`). That dedup is intentionally last-wins *across* layers, so a run-level
    native tool can override an agent-level default with the same id. *Within* a single layer,
    though, two different tools sharing an id are ambiguous: the silent last-wins would bind a
    stable id (e.g. an `MCPServerTool` id) to an unexpected definition such as a different server
    URL or authorization token. Fail fast here instead. Identical duplicates are allowed and
    collapsed later.

    `NativeToolFunc` callables are skipped: they have no stable `unique_id` to key on.
    """
    seen: dict[str, AbstractNativeTool] = {}
    for tool in native_tools:
        if not isinstance(tool, AbstractNativeTool):
            continue
        existing = seen.setdefault(tool.unique_id, tool)
        if existing is not tool and existing != tool:
            raise exceptions.UserError(
                f'Native tool id {tool.unique_id!r} maps to conflicting definitions in {source}. '
                'Native tool ids must be unique within a capability layer.'
            )


@dataclasses.dataclass
class _ResolvedRunCapabilities(Generic[AgentDepsT]):
    """The per-run capability state shared by `run`/`iter` and `realtime_session`.

    Produced by [`Agent._resolve_run_capabilities`][]: the resolved capability tree plus the
    contributions extracted from it (instructions, native tools, model settings, toolsets), so both a
    graph run and a realtime session wire capabilities up identically. See the cross-references on the
    two call sites for the surrounding logic each keeps to itself.
    """

    run_capability: AbstractCapability[AgentDepsT]
    capabilities: dict[str, AbstractCapability[AgentDepsT]]
    instructions: list[_instructions.SourcedInstruction[AgentDepsT]] | None
    native_tools: list[AgentNativeTool[AgentDepsT]]
    model_settings: AgentModelSettings[AgentDepsT] | None
    toolsets: list[AgentToolset[AgentDepsT]] | None
    resolved_layers: list[AbstractCapability[AgentDepsT]]
    """Each run layer after `for_run`, in order (instrumentation first when injected). The graph run
    compares the model-layer slice against its pre-resolution `model_layers` to detect whether any
    capability changed the model contribution during resolution (`model_layers_unchanged`)."""


def _layer_model_settings(
    run_context: RunContext[AgentDepsT],
    layers: Sequence[AgentModelSettings[AgentDepsT] | None],
    *,
    base: ModelSettings | None = None,
) -> ModelSettings | None:
    """Merge model-settings layers left-to-right, stamping `run_context.model_settings` before each.

    Each layer is a static `ModelSettings`, a callable resolved against the run context, or `None`.
    Stamping the merged-so-far onto `run_context.model_settings` before a callable layer runs lets it
    observe the previous layers — the agent -> capability -> run order both `iter` (per model-request
    step) and `realtime_session` (once, at connect) rely on. `base` is the model's own settings for a
    graph run; a realtime model has none, so it defaults to `None`.
    """
    merged = base
    run_context.model_settings = merged
    for layer in layers:
        resolved = layer(run_context) if callable(layer) else layer
        merged = merge_model_settings(merged, resolved)
        run_context.model_settings = merged
    return merged


def _build_run_capabilities(capability: AbstractCapability[AgentDepsT]) -> dict[str, AbstractCapability[AgentDepsT]]:
    capabilities: list[AbstractCapability[AgentDepsT]] = []
    capability.apply(capabilities.append)

    # Runs on the tree `_combine_duplicate_capabilities` has already resolved, so a shared id that
    # survives to here is one no `combine` accepted. Still needed at run time, not just at
    # construction: `defer_loading` and `id` can be set after the agent was built, and `for_run` may
    # hand back a capability carrying neither of the values construction saw.
    explicit_ids = _validate_capability_ids(capabilities)

    by_id: dict[str, AbstractCapability[AgentDepsT]] = {}
    for cap in capabilities:
        capability_id = cap.id
        if capability_id is None:
            capability_id = _synthetic_capability_id(type(cap), taken=by_id.keys() | explicit_ids)
        by_id[capability_id] = cap

    return by_id


def _synthetic_capability_id(cls: type[AbstractCapability[Any]], *, taken: Collection[str]) -> str:
    """A key for a capability the user never named, shaped so nobody mistakes it for a name.

    Every capability in a run needs a key, including the ones with no `id`, or `ctx.capabilities`
    could not list them. Nothing carries such a key out of the run: the paths that persist one --
    the load records progressive disclosure reads back out of message history -- all require
    `defer_loading=True`, which in turn requires an explicit `id`. So this is a run-local handle,
    and the only thing wrong with the `thinking` / `thinking_2` it used to be was that it looked
    exactly like an `id` somebody chose.

    It looked stable, too, and wasn't: the ordinal counted from attachment order, so listing two
    anonymous capabilities of one class the other way round swapped which of them owned
    `thinking_2`. Nothing depended on that, because nothing may -- but a reader who copied the key
    out of `ctx.capabilities` had no way to tell.

    Angle brackets say "the framework named this", the way `'<agent>'` and `'<output>'` already do
    for toolsets. The class name keeps the registry and a traceback readable. The random suffix is
    the part that matters: it differs every run, so relying on it fails immediately and visibly
    rather than the next time someone reorders a list.
    """
    base_id = to_snake(cls.__name__)
    while True:
        candidate = f'<{base_id}:{uuid4().hex[:6]}>'
        if candidate not in taken:
            return candidate


def _validate_spec(
    spec: dict[str, Any] | AgentSpec,
    deps_type: type[Any],
) -> tuple[AgentSpec, dict[str, Any]]:
    """Validate a spec dict/object and build the template context.

    Shared by `Agent.from_spec()` and `Agent._resolve_spec()`.

    Returns:
        A tuple of (validated_spec, template_context).
    """
    template_context: dict[str, Any] = {
        'deps_type': deps_type if deps_type is not type(None) else None,
    }
    if isinstance(spec, dict):
        validated_spec = AgentSpec.model_validate(spec, context=template_context)
    else:
        validated_spec = spec
    template_context['deps_schema'] = validated_spec.deps_schema
    return validated_spec, template_context


def _capabilities_from_spec(
    spec: AgentSpec,
    custom_capability_types: Sequence[type[AbstractCapability[Any]]],
    template_context: dict[str, Any],
) -> list[AbstractCapability[Any]]:
    """Instantiate capabilities from an AgentSpec using the capability registry.

    Shared by `Agent.from_spec()` and `Agent._resolve_spec()`.
    """
    from pydantic_ai.agent import spec as _agent_spec

    registry = get_capability_registry(custom_capability_types)

    def _instantiate_cap(
        cap_cls: type[AbstractCapability[Any]],
        args: tuple[Any, ...],
        kwargs: dict[str, Any],
    ) -> AbstractCapability[Any]:
        args, kwargs = validate_from_spec_args(cap_cls, args, kwargs, template_context)
        return cap_cls.from_spec(*args, **kwargs)

    # Set context so nested from_spec calls (e.g. PrefixTools) can reuse the registry
    ctx = _agent_spec.CapabilitySpecContext(registry=registry, instantiate=_instantiate_cap)
    token = _agent_spec.capability_spec_context.set(ctx)
    try:
        capabilities: list[AbstractCapability[Any]] = []
        for cap_spec in spec.capabilities:
            capability = load_from_registry(
                registry,
                cap_spec,
                label='capability',
                custom_types_param='custom_capability_types',
                instantiate=_instantiate_cap,
            )
            capabilities.append(capability)
        return capabilities
    finally:
        _agent_spec.capability_spec_context.reset(token)


@dataclasses.dataclass(init=False)
class _AgentFunctionToolset(FunctionToolset[AgentDepsT]):
    output_schema: _output.OutputSchema[Any]

    def __init__(
        self,
        tools: Sequence[Tool[AgentDepsT] | ToolFuncEither[AgentDepsT, ...]] = [],
        *,
        max_retries: int | None = None,
        timeout: float | None = None,
        id: str | None = None,
        output_schema: _output.OutputSchema[Any],
    ):
        self.output_schema = output_schema
        super().__init__(tools, max_retries=max_retries, timeout=timeout, id=id)

    @property
    def id(self) -> str:
        return AGENT_TOOLSET_ID

    @property
    def label(self) -> str:
        return 'the agent'