Class StreamState

libs/code/deepagents_code/non_interactive.py:234–282

Mutable state accumulated while iterating over the agent stream.

@dataclass
class StreamState:
    """Mutable state accumulated while iterating over the agent stream."""

    quiet: bool = False
    """When `True`, stream-time diagnostics (the tool-call and file-operation
    notifications, plus the stdout separator newline preceding a tool call) are
    suppressed, so stdout carries only agent response text."""

    stream: bool = True
    """When `True` (default), text chunks are written to stdout as they arrive.

    When `False`, text is buffered in `full_response` and flushed after the
    agent finishes.
    """

    full_response: list[str] = field(default_factory=list)
    """Accumulated text fragments from the AI message stream."""

    tool_call_buffers: dict[int | str, dict[str, str | None]] = field(
        default_factory=dict
    )
    """Maps a tool-call index or ID to its name/ID metadata for in-progress
    tool calls."""

    pending_interrupts: dict[str, HITLRequest] = field(default_factory=dict)
    """Maps interrupt IDs to their validated HITL requests that are awaiting
    decisions."""

    hitl_response: dict[str, dict[str, list[dict[str, str]]]] = field(
        default_factory=dict
    )
    """Maps interrupt IDs to dicts containing a `'decisions'` key with a list of
    decision dicts (each having a `'type'` key of `'approve'` or `'reject'`).

    Used to resume the agent after HITL processing.
    """

    interrupt_occurred: bool = False
    """Flag indicating whether any HITL interrupt was received during the
    current stream pass."""

    stats: SessionStats = field(default_factory=SessionStats)
    """Accumulated model usage stats for this stream."""

    spinner: _ConsoleSpinner | None = None
    """Optional animated spinner shown during agent work in verbose mode."""

    show_rubric_iterations: bool = False
    """Whether rubric lifecycle messages should include iteration numbers."""


@dataclass