Models

stirrup.core.models

RESOLUTION_1MP module-attribute

RESOLUTION_1MP = 1000000

RESOLUTION_480P module-attribute

RESOLUTION_480P = 640 * 480

__all__ module-attribute

__all__ = [
    "Addable",
    "AnyReasoningBlock",
    "AssistantBlock",
    "AssistantMessage",
    "AudioContentBlock",
    "BinaryContentBlock",
    "ChatMessage",
    "Content",
    "ContentBlock",
    "EmptyParams",
    "EncryptedReasoningBlock",
    "ImageContentBlock",
    "LLMClient",
    "OpaqueBlock",
    "Reasoning",
    "ReasoningBlock",
    "ReasoningRefBlock",
    "RedactedReasoningBlock",
    "SignedReasoningBlock",
    "SubAgentMetadata",
    "SummaryMessage",
    "SystemMessage",
    "TextBlock",
    "TokenUsage",
    "Tool",
    "ToolCall",
    "ToolMessage",
    "ToolProvider",
    "ToolResult",
    "ToolUseCountMetadata",
    "TurnWarningMessage",
    "UserMessage",
    "VideoContentBlock",
    "aggregate_metadata",
    "final_text",
    "joined_text",
    "reasoning_blocks",
    "tool_call_blocks",
]

Base64Bytes module-attribute

Base64Bytes = Annotated[
    bytes,
    PlainValidator(_b64_to_bytes),
    PlainSerializer(_bytes_to_b64, when_used="json"),
]

REASONING_BLOCK_TYPES module-attribute

REASONING_BLOCK_TYPES = (
    ReasoningBlock,
    SignedReasoningBlock,
    RedactedReasoningBlock,
    ReasoningRefBlock,
    EncryptedReasoningBlock,
)

Runtime mirror of AnyReasoningBlock for isinstance checks — keep in lockstep.

_CHANNEL_PROJECTION_DEPRECATION module-attribute

_CHANNEL_PROJECTION_DEPRECATION = "AssistantMessage.{channel} is deprecated; {replacement}. The compatibility projection will be removed in a future release."

ContentBlock

ContentBlock = (
    ImageContentBlock
    | VideoContentBlock
    | AudioContentBlock
    | str
)

Union of all content block types (image, video, audio, or text).

Content

Content = list[ContentBlock] | str

Message content: either a plain string or list of mixed content blocks.

AnyReasoningBlock

AnyReasoningBlock = (
    ReasoningBlock
    | SignedReasoningBlock
    | RedactedReasoningBlock
    | ReasoningRefBlock
    | EncryptedReasoningBlock
)

The reasoning family: one kind per passback mechanism.

AssistantBlock

AssistantBlock = Annotated[
    TextBlock
    | ReasoningBlock
    | SignedReasoningBlock
    | RedactedReasoningBlock
    | ReasoningRefBlock
    | EncryptedReasoningBlock
    | OpaqueBlock
    | ToolCall
    | ImageContentBlock
    | VideoContentBlock
    | AudioContentBlock,
    Field(discriminator=kind),
]

One block of an assistant turn, discriminated on kind.

_MediaBlock

_MediaBlock = (
    ImageContentBlock
    | VideoContentBlock
    | AudioContentBlock
)

UserRoleMessage

UserRoleMessage = Annotated[
    UserMessage | SummaryMessage | TurnWarningMessage,
    Field(discriminator=kind),
    BeforeValidator(_reject_untagged_user_message),
]

User-role messages, discriminated on kind — the agent-injected UserMessage subclasses share role="user", so dumped histories need the nested discriminator to rehydrate them as their own types (e.g. SummaryMessage.replaced_ids).

ChatMessage

ChatMessage = Annotated[
    SystemMessage
    | UserRoleMessage
    | AssistantMessage
    | ToolMessage,
    Field(discriminator=role),
]

Discriminated union of all message types, automatically parsed based on role field.

BinaryContentBlock

Bases: BaseModel, ABC

Base class for binary content (images, video, audio) with MIME type validation.

mime_type property

mime_type: str

MIME type for data based on headers.

extension property

extension: str

File extension for the content (e.g., 'png', 'mp4', 'mp3') without leading dot.

ImageContentBlock

Bases: BinaryContentBlock

Image content supporting PNG, JPEG, WebP, PSD formats with automatic downscaling.

to_base64_url

to_base64_url(
    max_pixels: int | None = RESOLUTION_1MP,
) -> str

Convert image to base64 data URL, optionally resizing to max pixel count.

Source code in src/stirrup/core/models.py

def to_base64_url(self, max_pixels: int | None = RESOLUTION_1MP) -> str:
    """Convert image to base64 data URL, optionally resizing to max pixel count."""
    img: Image.Image = Image.open(BytesIO(self.data))
    if max_pixels is not None and img.width * img.height > max_pixels:
        tw, th = downscale_image(img.width, img.height, max_pixels)
        img.thumbnail((tw, th), Image.Resampling.LANCZOS)
    if img.mode != "RGB":
        img = img.convert("RGB")
    buf = BytesIO()
    img.save(buf, format="PNG")
    return f"data:image/png;base64,{b64encode(buf.getvalue()).decode()}"

VideoContentBlock

Bases: BinaryContentBlock

MP4 video content with automatic transcoding and resolution downscaling.

to_base64_url

to_base64_url(
    max_pixels: int | None = RESOLUTION_480P,
    fps: int | None = None,
) -> str

Transcode to MP4 and return base64 data URL.

Source code in src/stirrup/core/models.py

def to_base64_url(self, max_pixels: int | None = RESOLUTION_480P, fps: int | None = None) -> str:
    """Transcode to MP4 and return base64 data URL."""
    with warnings.catch_warnings():
        warnings.filterwarnings("ignore", category=UserWarning, module="moviepy.*")
        with NamedTemporaryFile(suffix=".mp4") as fin, NamedTemporaryFile(suffix=".mp4") as fout:
            fin.write(self.data)
            fin.flush()
            clip = VideoFileClip(fin.name)
            tw, th = downscale_image(int(clip.w), int(clip.h), max_pixels)
            clip = clip.with_effects([Resize(new_size=(tw, th))])

            clip.write_videofile(
                fout.name,
                codec="libx264",
                fps=fps,
                audio=clip.audio is not None,
                audio_codec="aac",
                preset="veryfast",
                logger=None,
            )
            clip.close()
            return f"data:video/mp4;base64,{b64encode(fout.read()).decode()}"

AudioContentBlock

Bases: BinaryContentBlock

Audio content supporting MPEG, WAV, AAC, and other common audio formats.

to_base64_url

to_base64_url(bitrate: str = '192k') -> str

Transcode to MP3 and return base64 data URL.

Source code in src/stirrup/core/models.py

def to_base64_url(self, bitrate: str = "192k") -> str:
    """Transcode to MP3 and return base64 data URL."""
    with warnings.catch_warnings():
        warnings.filterwarnings("ignore", category=UserWarning, module="moviepy.*")
        with NamedTemporaryFile(suffix=".bin") as fin, NamedTemporaryFile(suffix=".mp3") as fout:
            fin.write(self.data)
            fin.flush()
            clip = AudioFileClip(fin.name)
            clip.write_audiofile(fout.name, codec="libmp3lame", bitrate=bitrate, logger=None)
            clip.close()
            return f"data:audio/mpeg;base64,{b64encode(fout.read()).decode()}"

Addable

Bases: Protocol

Protocol for types that support aggregation via add.

TokenUsage

Bases: BaseModel

Token counts for LLM usage.

Token terminology: output = reasoning + answer.

output property

output: int

Total output tokens (reasoning + answer).

total property

total: int

Total token count across input, answer, and reasoning.

__add__

__add__(other: TokenUsage) -> TokenUsage

Add two TokenUsage objects together, summing each field independently.

Source code in src/stirrup/core/models.py

def __add__(self, other: "TokenUsage") -> "TokenUsage":
    """Add two TokenUsage objects together, summing each field independently."""
    return TokenUsage(
        input=self.input + other.input,
        answer=self.answer + other.answer,
        reasoning=self.reasoning + other.reasoning,
    )

ToolUseCountMetadata

Bases: BaseModel

Generic metadata tracking tool usage count.

Implements Addable protocol for aggregation. Use this for tools that only need to track how many times they were called.

Subclasses can override add with their own type thanks to Self typing.

ToolResult

Bases: BaseModel

Result from a tool executor with optional metadata.

Generic over metadata type M. M should implement Addable protocol for aggregation support, but this is not enforced at the class level due to Pydantic schema generation limitations.

Attributes:

Name	Type	Description
content	Content	The result content (string, list of content blocks, or images)
success	bool	Whether the tool call was successful. For finish tools, controls if agent terminates.
metadata	M | None	Optional metadata (e.g., usage stats) that implements Addable for aggregation

EmptyParams

Bases: BaseModel

Empty parameter model for tools that don't require parameters.

Tool

Bases: BaseModel

Tool definition with name, description, parameter schema, and executor function.

Generic over

P: Parameter model type (Pydantic BaseModel subclass, or EmptyParams for parameterless tools) M: Metadata type (should implement Addable for aggregation; use None for tools without metadata)

Tools are simple, stateless callables. For tools requiring lifecycle management (setup/teardown, resource pooling), use a ToolProvider instead.

Example with parameters

class CalcParams(BaseModel):
    expression: str

calc_tool = Tool[CalcParams, None](
    name="calc",
    description="Evaluate math",
    parameters=CalcParams,
    executor=lambda p: ToolResult(content=str(eval(p.expression))),
)

Example without parameters (uses EmptyParams by default):

time_tool = Tool[EmptyParams, None](
    name="time",
    description="Get current time",
    executor=lambda _: ToolResult(content=datetime.now().isoformat()),
)

ToolProvider

Bases: ABC

Abstract base class for tool providers with lifecycle management.

ToolProviders manage resources (HTTP clients, sandboxes, server connections) and return Tool instances when entering their async context. They implement the async context manager protocol.

Use ToolProvider for: - Tools requiring setup/teardown (connections, temp directories) - Tools that return multiple Tool instances (e.g., MCP servers) - Tools with shared state across calls (e.g., HTTP client pooling)

Example

class MyToolProvider(ToolProvider): async def aenter(self) -> Tool | list[Tool]: # Setup resources and return tool(s) return self._create_tool()

# __aexit__ is optional - default is no-op

Agent automatically manages ToolProvider lifecycle via its session() context.

__aenter__ abstractmethod async

__aenter__() -> Tool | list[Tool]

Enter async context: setup resources and return tool(s).

Returns:

Type	Description
Tool | list[Tool]	A single Tool instance, or a list of Tool instances for providers
Tool | list[Tool]	that expose multiple tools (e.g., MCP servers).

Source code in src/stirrup/core/models.py

@abstractmethod
async def __aenter__(self) -> "Tool | list[Tool]":
    """Enter async context: setup resources and return tool(s).

    Returns:
        A single Tool instance, or a list of Tool instances for providers
        that expose multiple tools (e.g., MCP servers).
    """
    ...

__aexit__ async

__aexit__(
    exc_type: type[BaseException] | None,
    exc_val: BaseException | None,
    exc_tb: TracebackType | None,
) -> None

Exit async context: cleanup resources. Default: no-op.

Source code in src/stirrup/core/models.py

async def __aexit__(  # noqa: B027
    self,
    exc_type: type[BaseException] | None,
    exc_val: BaseException | None,
    exc_tb: TracebackType | None,
) -> None:
    """Exit async context: cleanup resources. Default: no-op."""

LLMClient

Bases: Protocol

Protocol defining the interface for LLM client implementations.

Any LLM client must implement this protocol to work with the Agent class. Provides text generation with tool support and model capability inspection.

ToolCall

Bases: BaseModel

Represents a tool invocation request from the LLM.

Also a member of the AssistantBlock union: the kind discriminator is defaulted, so legacy payloads without the key still validate anywhere ToolCall is used as a plain input, and new dumps always carry it.

Attributes:

Name	Type	Description
name	str	Name of the tool to invoke
arguments	str	JSON string containing tool parameters
tool_call_id	str	Unique identifier for tracking this tool call and its result

signature class-attribute instance-attribute

signature: str | None = None

Opaque passback state attached to this exact block, e.g. a Google thought signature.

has_provider_tool_call_id class-attribute instance-attribute

has_provider_tool_call_id: bool = True

Whether tool_call_id was present on the provider's original block.

A client may synthesize tool_call_id for internal call/result matching while retaining that it must be omitted from provider-attached passback.

from_provider classmethod

from_provider(
    *,
    provider_id: str | None,
    name: str,
    arguments: str,
    signature: str | None = None,
) -> Self

Capture one provider call with a stable internal correlation ID.

Source code in src/stirrup/core/models.py

@classmethod
def from_provider(
    cls,
    *,
    provider_id: str | None,
    name: str,
    arguments: str,
    signature: str | None = None,
) -> Self:
    """Capture one provider call with a stable internal correlation ID."""
    native_id = provider_id or None
    return cls(
        tool_call_id=native_id or uuid4().hex,
        has_provider_tool_call_id=native_id is not None,
        name=name,
        arguments=arguments,
        signature=signature,
    )

SystemMessage

Bases: BaseModel

System-level instructions and context for the LLM.

UserMessage

Bases: BaseModel

User input message to the LLM.

SummaryMessage

Bases: UserMessage

Summary message bridging summarized-away conversation context.

Attributes:

Name	Type	Description
replaced_ids	list[str]	Ids of the AssistantMessages this summary replaced, so consumers of dumped histories can reconstruct lineage offline.

TurnWarningMessage

Bases: UserMessage

Warning message injected when the agent is close to max_turns.

Reasoning

Bases: BaseModel

Channel-era reasoning shape accepted only while reading serialized v0.1 messages.

Deprecated as a standalone type: match on the AnyReasoningBlock kinds (ReasoningBlock / SignedReasoningBlock / RedactedReasoningBlock / ReasoningRefBlock / EncryptedReasoningBlock) instead.

TextBlock

Bases: BaseModel

One contiguous run of answer text in an assistant turn.

signature carries opaque passback state attached to this exact block, e.g. a Google thought signature emitted on a visible text part. A client that cannot re-emit the signature must reject passback rather than silently stripping it.

ReasoningBlock

Bases: BaseModel

In-band reasoning text with no passback token.

E.g. reasoning_content on Chat Completions-compatible hosts, or -tag extraction.

SignedReasoningBlock

Bases: BaseModel

Reasoning bound to an opaque provider signature re-emitted verbatim on passback.

E.g. Anthropic signed thinking blocks.

RedactedReasoningBlock

Bases: BaseModel

Reasoning the provider withheld, replaced by an opaque payload.

E.g. Anthropic redacted_thinking: data must be re-emitted verbatim as a redacted_thinking block on passback. Carries no readable content.

ReasoningRefBlock

Bases: BaseModel

Reasoning held provider-side and passed back by reference.

This is retained for providers that require an item-level handle on replay. OpenAI Responses continuation uses AssistantMessage.provider_response_id instead and does not create this block.

EncryptedReasoningBlock

Bases: BaseModel

Reasoning returned as an opaque encrypted payload for stateless passback.

E.g. OpenAI Responses reasoning items requested with include: ["reasoning.encrypted_content"] (store=false / zero-data-retention): the item — id, summary parts, and encrypted payload — is re-emitted verbatim in position on passback. The payload is opaque and non-inspectable.

OpaqueBlock

Bases: BaseModel

Provider-native block the framework carries uninterpreted.

For provider-issued marker/control blocks that must round-trip untouched: data holds the block's raw JSON (self-describing — the provider's own type field travels inside it). The framework preserves it in position through history, projections, and serialization so a client that understands the payload can re-emit it verbatim on passback; other clients fail loudly.

AssistantMessage

Bases: BaseModel

LLM response message: an ordered sequence of assistant blocks.

blocks is the only stored content. The channel-era content and tool_calls attributes remain deprecated views; reasoning raises because an ordered reasoning block sequence has no faithful channel-shaped projection. Serialized v0.1 payloads upgrade to blocks during validation. Channel-shaped construction is not part of the v0.2 API; new code constructs blocks directly. Mixing blocks with non-empty legacy channel keys raises.

provider_response_id class-attribute instance-attribute

provider_response_id: str | None = None

Provider-attached continuation state, e.g. an OpenAI Responses resp_... id.

This is turn metadata rather than emitted assistant content, so it lives beside blocks instead of inside their emission order. It is distinct from id (Stirrup's message identity) and ReasoningRefBlock.id (an emitted reasoning item handle).

content property

content: list[AssistantBlock] | str

Bare text for one text block, empty text for no blocks, or the block list.

reasoning property

reasoning: Reasoning | None

Deprecated channel accessor retained only to fail with migration guidance.

tool_calls property

tool_calls: list[ToolCall]

Tool calls in emission order.

e2e_otps property

e2e_otps: float | None

End-to-end output tokens per second.

ToolMessage

Bases: BaseModel

Tool execution result returned to the LLM.

Attributes:

Name	Type	Description
role	Literal['tool']	Always "tool"
content	Content	The tool result content
tool_call_id	str	ID linking this result to the corresponding tool call
name	str | None	Name of the tool that was called
args_was_valid	bool	Whether the tool arguments were valid
success	bool	Whether the tool executed successfully (used by finish tool to control termination)

tool_duration property

tool_duration: float | None

Tool execution duration in seconds.

SubAgentMetadata

Bases: BaseModel

Metadata from sub-agent execution including token usage, message history, and child run metadata.

Implements Addable protocol to support aggregation across multiple subagent calls.

__add__

__add__(other: SubAgentMetadata) -> SubAgentMetadata

Combine metadata from multiple subagent calls.

Source code in src/stirrup/core/models.py

def __add__(self, other: "SubAgentMetadata") -> "SubAgentMetadata":
    """Combine metadata from multiple subagent calls."""
    # Concatenate message histories
    combined_history = self.message_history + other.message_history
    # Merge run metadata (concatenate lists per key, keep last for non-list internal keys)
    combined_meta: dict[str, Any] = dict(self.run_metadata)
    for key, value in other.run_metadata.items():
        if key in combined_meta and isinstance(combined_meta[key], list) and isinstance(value, list):
            combined_meta[key] = combined_meta[key] + value
        else:
            combined_meta[key] = value
    return SubAgentMetadata(
        message_history=combined_history,
        run_metadata=combined_meta,
    )

_bytes_to_b64

_bytes_to_b64(v: bytes) -> str

Source code in src/stirrup/core/models.py

def _bytes_to_b64(v: bytes) -> str:
    return base64.b64encode(v).decode("ascii")

_b64_to_bytes

_b64_to_bytes(v: bytes | str) -> bytes

Source code in src/stirrup/core/models.py

def _b64_to_bytes(v: bytes | str) -> bytes:
    if isinstance(v, bytes):
        return v
    if isinstance(v, str):
        return base64.b64decode(v.encode("ascii"))
    raise TypeError("Invalid bytes value")

downscale_image

downscale_image(
    w: int, h: int, max_pixels: int | None = 1000000
) -> tuple[int, int]

Downscale image dimensions to fit within max pixel count while maintaining aspect ratio.

Returns even dimensions with minimum 2x2 size.

Source code in src/stirrup/core/models.py

def downscale_image(w: int, h: int, max_pixels: int | None = 1_000_000) -> tuple[int, int]:
    """Downscale image dimensions to fit within max pixel count while maintaining aspect ratio.

    Returns even dimensions with minimum 2x2 size.
    """
    s = 1.0 if max_pixels is None or w * h <= max_pixels else sqrt(max_pixels / (w * h))
    nw, nh = int(w * s) // 2 * 2, int(h * s) // 2 * 2
    return max(nw, 2), max(nh, 2)

_merge_dicts

_merge_dicts(a: dict, b: dict) -> dict

Deep merge two dicts, recursively merging nested dicts and summing numbers.

Source code in src/stirrup/core/models.py

def _merge_dicts(a: dict, b: dict) -> dict:
    """Deep merge two dicts, recursively merging nested dicts and summing numbers."""
    merged = dict(a)
    for key, value in b.items():
        if key in merged:
            existing = merged[key]
            if isinstance(existing, dict) and isinstance(value, dict):
                merged[key] = _merge_dicts(existing, value)
            elif (isinstance(existing, int | float) and isinstance(value, int | float)) or (
                isinstance(existing, list) and isinstance(value, list)
            ):
                merged[key] = existing + value
            else:
                merged[key] = value
        else:
            merged[key] = value
    return merged

_aggregate_list

_aggregate_list(metadata_list: list[T]) -> T | None

Aggregate a list of metadata using add, with fallback for dicts.

Source code in src/stirrup/core/models.py

def _aggregate_list[T: Addable](metadata_list: list[T]) -> T | None:
    """Aggregate a list of metadata using __add__, with fallback for dicts."""
    if not metadata_list:
        return None
    aggregated: T = metadata_list[0]
    for m in metadata_list[1:]:
        if isinstance(aggregated, dict) and isinstance(m, dict):
            aggregated = _merge_dicts(aggregated, m)  # ty: ignore[invalid-assignment]
        else:
            aggregated = aggregated + m
    return aggregated

to_json_serializable

to_json_serializable(value: object) -> object

Source code in src/stirrup/core/models.py

def to_json_serializable(value: object) -> object:
    # None and JSON primitives
    if value is None or isinstance(value, str | int | bool):
        return value

    # Floats need special handling for nan/inf
    if isinstance(value, float):
        if isnan(value) or isinf(value):
            raise ValueError(f"Cannot serialize {value} to JSON")
        return value

    # Pydantic models
    if isinstance(value, BaseModel):
        return value.model_dump(mode="json")

    # Common non-serializable types
    if isinstance(value, datetime | date | time):
        return value.isoformat()

    if isinstance(value, timedelta):
        return value.total_seconds()

    if isinstance(value, Decimal):
        return float(value)

    if isinstance(value, dict):
        return {k: to_json_serializable(v) for k, v in value.items()}

    if isinstance(value, list | tuple | set | frozenset):
        return [to_json_serializable(v) for v in value]

    # We have not implemented other cases (e.g. Bytes, Enum, etc.)
    raise TypeError(f"Cannot serialize {type(value).__name__} to JSON: {value!r}")

_collect_all_token_usage

_collect_all_token_usage(result: dict) -> TokenUsage

Recursively collect all token_usage from a flattened aggregate_metadata result.

Parameters:

Name	Type	Description	Default
result	dict	The flattened dict from aggregate_metadata (before JSON serialization)	required

Returns:

Type	Description
TokenUsage	Combined TokenUsage from all entries (direct and nested sub-agents)

Source code in src/stirrup/core/models.py

def _collect_all_token_usage(result: dict) -> "TokenUsage":
    """Recursively collect all token_usage from a flattened aggregate_metadata result.

    Args:
        result: The flattened dict from aggregate_metadata (before JSON serialization)

    Returns:
        Combined TokenUsage from all entries (direct and nested sub-agents)
    """
    total = TokenUsage()

    for key, value in result.items():
        if key == "token_usage" and isinstance(value, TokenUsage):
            # Direct token_usage at this level
            total = total + value
        elif isinstance(value, dict):
            # This could be a sub-agent's tool dict - check for nested token_usage
            nested_token_usage = value.get("token_usage")
            if isinstance(nested_token_usage, TokenUsage):
                total = total + nested_token_usage

    return total

aggregate_metadata

aggregate_metadata(
    metadata_dict: dict[str, list[Any]],
    prefix: str = "",
    return_json_serializable: Literal[True] = True,
) -> object

aggregate_metadata(
    metadata_dict: dict[str, list[Any]],
    prefix: str = "",
    return_json_serializable: Literal[False] = ...,
) -> dict

aggregate_metadata(
    metadata_dict: dict[str, list[Any]],
    prefix: str = "",
    return_json_serializable: bool = True,
) -> dict | object

Aggregate metadata lists and flatten sub-agents into a single-level dict with hierarchical keys.

For entries with nested run_metadata (e.g., SubAgentMetadata), flattens sub-agents using dot notation. Each sub-agent's value is a dict mapping its direct tool names to their aggregated metadata (excluding nested sub-agent data, which gets its own top-level key).

At the root level, token_usage is rolled up to include all sub-agent token usage.

Parameters:

Name	Type	Description	Default
metadata_dict	dict[str, list[Any]]	Dict mapping names (tools or agents) to lists of metadata instances	required
prefix	str	Key prefix for nested calls (used internally for recursion)	''

Returns:

Name	Type	Description
	dict | object	Flat dict with dot-notation keys for sub-agents.
Example	dict | object	{ "token_usage": , "web_browsing_sub_agent": {"web_search": , "token_usage": }, "web_browsing_sub_agent.web_fetch_sub_agent": {"fetch_web_page": , "token_usage": }
	dict | object	}

Source code in src/stirrup/core/models.py

def aggregate_metadata(
    metadata_dict: dict[str, list[Any]], prefix: str = "", return_json_serializable: bool = True
) -> dict | object:
    """Aggregate metadata lists and flatten sub-agents into a single-level dict with hierarchical keys.

    For entries with nested run_metadata (e.g., SubAgentMetadata), flattens sub-agents using dot notation.
    Each sub-agent's value is a dict mapping its direct tool names to their aggregated metadata
    (excluding nested sub-agent data, which gets its own top-level key).

    At the root level, token_usage is rolled up to include all sub-agent token usage.

    Args:
        metadata_dict: Dict mapping names (tools or agents) to lists of metadata instances
        prefix: Key prefix for nested calls (used internally for recursion)

    Returns:
        Flat dict with dot-notation keys for sub-agents.
        Example: {
            "token_usage": <combined from all agents>,
            "web_browsing_sub_agent": {"web_search": <aggregated>, "token_usage": <aggregated>},
            "web_browsing_sub_agent.web_fetch_sub_agent": {"fetch_web_page": <aggregated>, "token_usage": <aggregated>}
        }
    """
    result: dict = {}

    # First pass: aggregate all entries in this level (skip internal keys prefixed with _)
    aggregated_level: dict = {}
    for name, metadata_list in metadata_dict.items():
        if name.startswith("_") or not metadata_list:
            continue
        aggregated_level[name] = _aggregate_list(metadata_list)

    # Second pass: separate nested sub-agents from direct tools, and recurse
    direct_tools: dict = {}
    for name, aggregated in aggregated_level.items():
        if hasattr(aggregated, "run_metadata") and isinstance(aggregated.run_metadata, dict):
            # This is a sub-agent - recurse into it
            full_key = f"{prefix}.{name}" if prefix else name
            nested = aggregate_metadata(aggregated.run_metadata, prefix=full_key, return_json_serializable=False)
            result.update(nested)
        else:
            # This is a direct tool/metadata - keep it at this level
            direct_tools[name] = aggregated

    # Store direct tools under the current prefix
    if prefix:
        result[prefix] = direct_tools
    else:
        # At root level, merge direct tools into result
        result.update(direct_tools)

    # At root level, roll up all token_usage from sub-agents
    if not prefix:
        total_token_usage = _collect_all_token_usage(result)
        if total_token_usage.total > 0:
            result["token_usage"] = [total_token_usage]

    if return_json_serializable:
        # Convert all Pydantic models to JSON-serializable dicts
        return to_json_serializable(result)
    return result

joined_text

joined_text(blocks: Sequence[AssistantBlock]) -> str | None

All answer text across text blocks, directly concatenated; None when absent.

Source code in src/stirrup/core/models.py

def joined_text(blocks: Sequence[AssistantBlock]) -> str | None:
    """All answer text across text blocks, directly concatenated; None when absent."""
    texts = [block.text for block in blocks if isinstance(block, TextBlock)]
    if not texts:
        return None
    return "".join(texts)

final_text

final_text(blocks: Sequence[AssistantBlock]) -> str | None

Text of the last text block — the "answer" in thinking→text→thinking→text turns.

Source code in src/stirrup/core/models.py

def final_text(blocks: Sequence[AssistantBlock]) -> str | None:
    """Text of the last text block — the "answer" in thinking→text→thinking→text turns."""
    for block in reversed(blocks):
        if isinstance(block, TextBlock):
            return block.text
    return None

tool_call_blocks

tool_call_blocks(
    blocks: Sequence[AssistantBlock],
) -> list[ToolCall]

Tool calls in emission order.

Source code in src/stirrup/core/models.py

def tool_call_blocks(blocks: Sequence[AssistantBlock]) -> list[ToolCall]:
    """Tool calls in emission order."""
    return [block for block in blocks if isinstance(block, ToolCall)]

reasoning_blocks

reasoning_blocks(
    blocks: Sequence[AssistantBlock],
) -> list[AnyReasoningBlock]

Reasoning blocks (any kind) in emission order.

Source code in src/stirrup/core/models.py

def reasoning_blocks(blocks: Sequence[AssistantBlock]) -> list[AnyReasoningBlock]:
    """Reasoning blocks (any kind) in emission order."""
    return [block for block in blocks if isinstance(block, REASONING_BLOCK_TYPES)]

_reasoning_to_block

_reasoning_to_block(reasoning: object) -> object

Upgrade a flat channel-era Reasoning value to its block equivalent.

A signature means signed passback; bare content is in-band reasoning.

Source code in src/stirrup/core/models.py

def _reasoning_to_block(reasoning: object) -> object:
    """Upgrade a flat channel-era ``Reasoning`` value to its block equivalent.

    A signature means signed passback; bare content is in-band reasoning.
    """
    match reasoning:
        case Reasoning(signature=signature, content=content):
            pass
        case {**mapping}:
            signature = mapping.get("signature")
            content = mapping.get("content")
        case _:
            return reasoning  # let block validation fail loudly
    if signature is not None:
        if not isinstance(signature, str):
            return reasoning
        return SignedReasoningBlock(signature=signature, content=content if isinstance(content, str) else "")
    return ReasoningBlock(content=content if isinstance(content, str) else "")

_warn_channel_projection

_warn_channel_projection(
    channel: str, replacement: str
) -> None

Source code in src/stirrup/core/models.py

def _warn_channel_projection(channel: str, replacement: str) -> None:
    warnings.warn(
        _CHANNEL_PROJECTION_DEPRECATION.format(channel=channel, replacement=replacement),
        DeprecationWarning,
        stacklevel=3,
    )

_upgrade_legacy_assistant_message

_upgrade_legacy_assistant_message(data: object) -> object

Read a serialized v0.1 assistant message into canonical blocks.

Source code in src/stirrup/core/models.py

def _upgrade_legacy_assistant_message(data: object) -> object:
    """Read a serialized v0.1 assistant message into canonical blocks."""
    if not isinstance(data, dict):
        return data
    upgraded = dict(data)
    content = upgraded.pop("content", None)
    reasoning = upgraded.pop("reasoning", None)
    tool_calls = upgraded.pop("tool_calls", None)

    if "blocks" in upgraded:
        if content not in (None, "", []) or reasoning or tool_calls:
            raise ValueError("AssistantMessage cannot mix 'blocks' with legacy channels")
        return upgraded

    blocks: list[object] = []
    if reasoning:
        blocks.append(_reasoning_to_block(reasoning))
    if isinstance(content, str):
        if content:
            blocks.append(TextBlock(text=content))
    elif isinstance(content, list):
        blocks.extend(TextBlock(text=item) if isinstance(item, str) else item for item in content if item != "")
    elif content is not None:
        raise ValueError(f"AssistantMessage 'content' must be a string or list, got {type(content).__name__}")
    if tool_calls is not None and not isinstance(tool_calls, list | tuple):
        raise ValueError("AssistantMessage 'tool_calls' must be a sequence")
    blocks.extend({**call, "kind": "tool_call"} if isinstance(call, dict) else call for call in tool_calls or [])
    upgraded["blocks"] = blocks
    return upgraded

_reject_untagged_user_message

_reject_untagged_user_message(data: object) -> object

Source code in src/stirrup/core/models.py

def _reject_untagged_user_message(data: object) -> object:
    if isinstance(data, dict) and "kind" not in data:
        raise ValueError(
            "Cannot load a cached user message without the required 'kind' discriminator. "
            "This cache predates typed user-message kinds and cannot be resumed safely; "
            "delete the old Stirrup cache and restart the run."
        )
    return data

_upgrade_legacy_message_sequence

_upgrade_legacy_message_sequence(
    messages: object,
) -> object

Correlate nullable v0.1 tool-call/result IDs before per-message validation.

v0.1 allowed both sides of tool correlation to omit their ID. A single-message validator cannot recover that relationship, so sequence readers assign a shared ID to each idless call and copy it to the following null-ID result in emission order. Explicit provider IDs are never rewritten.

Source code in src/stirrup/core/models.py

def _upgrade_legacy_message_sequence(messages: object) -> object:
    """Correlate nullable v0.1 tool-call/result IDs before per-message validation.

    v0.1 allowed both sides of tool correlation to omit their ID. A single-message
    validator cannot recover that relationship, so sequence readers assign a shared
    ID to each idless call and copy it to the following null-ID result in emission
    order. Explicit provider IDs are never rewritten.
    """
    if not isinstance(messages, list | tuple):
        return messages

    upgraded_messages: list[object] = []
    pending_calls: list[tuple[str, str | None]] = []
    for message in messages:
        if not isinstance(message, dict):
            pending_calls.clear()
            upgraded_messages.append(message)
            continue

        upgraded = dict(message)
        match upgraded.get("role"):
            case "assistant":
                pending_calls.clear()
                container_key = "blocks" if isinstance(upgraded.get("blocks"), list) else "tool_calls"
                items = upgraded.get(container_key)
                if isinstance(items, list):
                    migrated_items: list[object] = []
                    for item in items:
                        is_tool_call = container_key == "tool_calls" or (
                            isinstance(item, dict) and item.get("kind") == "tool_call"
                        )
                        if not is_tool_call or not isinstance(item, dict):
                            migrated_items.append(item)
                            continue

                        call = dict(item)
                        call_id = call.get("tool_call_id")
                        if call_id in (None, ""):
                            call_id = uuid4().hex
                            call["tool_call_id"] = call_id
                            call["has_provider_tool_call_id"] = False
                        if not isinstance(call_id, str):
                            raise ValueError("Legacy tool_call_id must be a string or null")
                        call_name = call.get("name")
                        pending_calls.append((call_id, call_name if isinstance(call_name, str) else None))
                        migrated_items.append(call)
                    upgraded[container_key] = migrated_items

            case "tool":
                result_id = upgraded.get("tool_call_id")
                if result_id in (None, ""):
                    if not pending_calls:
                        raise ValueError("Cannot correlate a null-ID v0.1 tool result with a preceding tool call")
                    expected_id, expected_name = pending_calls.pop(0)
                    result_name = upgraded.get("name")
                    if isinstance(result_name, str) and expected_name is not None and result_name != expected_name:
                        raise ValueError(
                            f"Cannot correlate v0.1 tool result {result_name!r} with preceding call {expected_name!r}"
                        )
                    upgraded["tool_call_id"] = expected_id
                elif isinstance(result_id, str):
                    pending_calls = [call for call in pending_calls if call[0] != result_id]

            case "user":
                pending_calls.clear()

            case _:
                pending_calls.clear()

        upgraded_messages.append(upgraded)
    return upgraded_messages