Function _run_agent_loop

libs/code/deepagents_code/non_interactive.py:776–908

Run the agent and handle HITL interrupts until the task completes. The loop is capped at `max_turns` when set, otherwise `_MAX_HITL_ITERATIONS`, to prevent runaway retries (e.g. the agent repeatedly attempting rejected commands). Args: agent: The agent (Pregel or RemoteAgen

(
    agent: Any,  # noqa: ANN401
    message: str,
    config: RunnableConfig,
    console: Console,
    file_op_tracker: FileOpTracker,
    *,
    quiet: bool = False,
    stream: bool = True,
    message_kwargs: dict[str, Any] | None = None,
    thread_url_lookup: ThreadUrlLookupState | None = None,
    max_turns: int | None = None,
    rubric: str | None = None,
    show_rubric_iterations: bool = False,
)

async def _run_agent_loop(
    agent: Any,  # noqa: ANN401
    message: str,
    config: RunnableConfig,
    console: Console,
    file_op_tracker: FileOpTracker,
    *,
    quiet: bool = False,
    stream: bool = True,
    message_kwargs: dict[str, Any] | None = None,
    thread_url_lookup: ThreadUrlLookupState | None = None,
    max_turns: int | None = None,
    rubric: str | None = None,
    show_rubric_iterations: bool = False,
) -> None:
    """Run the agent and handle HITL interrupts until the task completes.

    The loop is capped at `max_turns` when set,
    otherwise `_MAX_HITL_ITERATIONS`, to prevent runaway retries
    (e.g. the agent repeatedly attempting rejected commands).

    Args:
        agent: The agent (Pregel or RemoteAgent).
        message: The user's task message.
        config: LangGraph runnable config.
        console: Rich console for formatted output.
        file_op_tracker: Tracker for file-operation diffs.
        quiet: Suppress diagnostic formatting on stdout.
        stream: When `True`, text is written to stdout as it arrives.

            When `False`, the full response is buffered and flushed at
            the end.
        message_kwargs: Extra fields merged into the initial HumanMessage
            dict (e.g., `additional_kwargs` for persisted skill metadata).
        thread_url_lookup: Optional non-blocking lookup state for rendering
            a fast-follow LangSmith thread link.
        max_turns: Optional cap on total agentic turns (initial response plus
            HITL resumes).

            When `None`, falls back to `_MAX_HITL_ITERATIONS`.
        rubric: Acceptance criteria supplied to `RubricMiddleware` via the
            graph's `rubric` state field.

            `None` leaves it unset (no grading).
        show_rubric_iterations: Whether rubric lifecycle messages should include
            iteration numbers.

    Raises:
        HITLIterationLimitError: If the effective turn limit is exceeded.
    """
    spinner = None if quiet else _ConsoleSpinner(console)
    state = StreamState(
        quiet=quiet,
        stream=stream,
        spinner=spinner,
        show_rubric_iterations=show_rubric_iterations,
    )
    user_msg: dict[str, Any] = {"role": "user", "content": message}