mirror of
https://github.com/usestrix/strix.git
synced 2026-08-28 05:25:00 +00:00
8b464dae5d
One tool was doing three jobs (wait on the user, wait on other agents, and - wrongly - wait for a long-running command), so the driver had to guess which one an agent meant and used parent_id as the proxy: the root waits for a human, everyone else waits for agents. That proxy is wrong, since the user can message any agent from the TUI's agent tree. Tool identity now carries the intent, and the coordinator records it as a wait_kind that survives snapshot/restore: respond_to_user -> wait_kind="user", never auto-resumed (root or not) wait_for_agents -> wait_kind="agents", auto-resumed on a 300s timer recovery exhaust -> wait_kind="stalled" respond_to_user fuses the message and the yield into one call, so there is no way to answer and then forget to stop - the two-step that gpt-4o-mini skipped 2/2 in live testing. Plain text still renders as before. Auto-resume is also bounded now: an agent that re-parks after every timeout burned a model turn every 300s for the rest of the scan (and, since parked children notify their parent, spammed the parent's inbox on the same cycle). After _MAX_IDLE_AUTO_RESUMES it stays parked until a real message arrives.
1003 lines
35 KiB
Python
1003 lines
35 KiB
Python
"""Execution loop for addressable SDK-backed Strix agents."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import asyncio
|
|
import contextlib
|
|
import logging
|
|
import uuid
|
|
from collections.abc import Callable
|
|
from typing import TYPE_CHECKING, Any, cast
|
|
|
|
import litellm
|
|
from agents import RunConfig, Runner
|
|
from agents.exceptions import AgentsException, MaxTurnsExceeded, UserError
|
|
from agents.sandbox.errors import ExecTransportError
|
|
from docker import errors as docker_errors # type: ignore[import-untyped, unused-ignore]
|
|
from openai import (
|
|
APIConnectionError,
|
|
APIError,
|
|
APITimeoutError,
|
|
)
|
|
|
|
from strix.config import codex
|
|
from strix.core.hooks import (
|
|
BudgetExceededError,
|
|
BudgetPausedError,
|
|
SubagentBudgetReservedError,
|
|
)
|
|
from strix.core.inputs import child_initial_input
|
|
from strix.core.sessions import (
|
|
enforce_image_budget,
|
|
open_agent_session,
|
|
replace_session_items,
|
|
seed_initial_input,
|
|
strip_all_images_from_session,
|
|
)
|
|
from strix.llm.compaction import is_context_overflow, maybe_compact
|
|
|
|
|
|
if TYPE_CHECKING:
|
|
from pathlib import Path
|
|
|
|
from agents.items import TResponseInputItem
|
|
from agents.lifecycle import RunHooks
|
|
from agents.memory import Session, SQLiteSession
|
|
from agents.result import RunResultBase
|
|
|
|
from strix.core.agents import AgentCoordinator, Status
|
|
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
StreamEventSink = Callable[[str, Any], None]
|
|
|
|
_INPUT_REJECTION_CODES = frozenset({400, 404, 422})
|
|
_MAX_COMPACTIONS_PER_CYCLE = 2
|
|
|
|
|
|
class ProviderRefusalError(AgentsException):
|
|
"""Raised when a provider returns a structured refusal instead of an exception."""
|
|
|
|
|
|
def _structured_provider_refusal(result: Any) -> str | None:
|
|
for item in getattr(result, "new_items", ()) or ():
|
|
raw_item = getattr(item, "raw_item", None)
|
|
for content in getattr(raw_item, "content", ()) or ():
|
|
if getattr(content, "type", None) != "refusal":
|
|
continue
|
|
refusal = getattr(content, "refusal", None)
|
|
if isinstance(refusal, str) and refusal.strip():
|
|
return refusal.strip()
|
|
return "The model provider refused this request."
|
|
return None
|
|
|
|
|
|
def _run_config_model(run_config: RunConfig) -> str | None:
|
|
return run_config.model if isinstance(run_config.model, str) else None
|
|
|
|
|
|
def _agent_instructions(agent: Any) -> str:
|
|
instructions = getattr(agent, "instructions", None)
|
|
return instructions if isinstance(instructions, str) else ""
|
|
|
|
|
|
def _agent_tools_text(agent: Any) -> str:
|
|
parts: list[str] = []
|
|
for tool in getattr(agent, "tools", []) or []:
|
|
name = getattr(tool, "name", "")
|
|
description = getattr(tool, "description", "") or ""
|
|
schema = getattr(tool, "params_json_schema", "") or ""
|
|
parts.append(f"{name} {description} {schema}")
|
|
return "\n".join(parts)
|
|
|
|
|
|
async def _compact_session(
|
|
agent: Any, session: Session, run_config: RunConfig, *, force: bool
|
|
) -> bool:
|
|
model = _run_config_model(run_config)
|
|
if session is None or model is None:
|
|
return False
|
|
return await maybe_compact(
|
|
session,
|
|
model=model,
|
|
instructions=_agent_instructions(agent),
|
|
tools_text=_agent_tools_text(agent),
|
|
force=force,
|
|
)
|
|
|
|
|
|
_MAX_TRANSIENT_MODEL_RETRIES = 5
|
|
_TRANSIENT_MODEL_RETRY_BASE_DELAY_S = 2.0
|
|
_TRANSIENT_MODEL_RETRY_MAX_DELAY_S = 90.0
|
|
|
|
|
|
def _model_error_status_code(exc: BaseException) -> int | None:
|
|
code = getattr(exc, "status_code", None)
|
|
return code if isinstance(code, int) else None
|
|
|
|
|
|
def _is_transient_model_error(exc: BaseException) -> bool:
|
|
if codex.is_content_guardrail_error(exc):
|
|
return False
|
|
if isinstance(
|
|
exc, APITimeoutError | APIConnectionError | TimeoutError | ConnectionError | OSError
|
|
):
|
|
return True
|
|
code = _model_error_status_code(exc)
|
|
if code is not None:
|
|
return bool(litellm._should_retry(code))
|
|
return isinstance(exc, APIError)
|
|
|
|
|
|
def _transient_model_retry_delay(attempt: int) -> float:
|
|
delay = _TRANSIENT_MODEL_RETRY_BASE_DELAY_S * float(2 ** (attempt - 1))
|
|
return min(delay, _TRANSIENT_MODEL_RETRY_MAX_DELAY_S)
|
|
|
|
|
|
async def _salvage_stream_to_session(
|
|
session: Session,
|
|
pre_run_items: list[Any],
|
|
stream: Any,
|
|
agent_id: str,
|
|
) -> None:
|
|
"""Persist a crashed run's full history so a revived agent loses no context."""
|
|
if stream is None:
|
|
return
|
|
try:
|
|
replay = list(stream.to_input_list())
|
|
except Exception:
|
|
logger.exception("could not build salvage history for %s", agent_id)
|
|
return
|
|
desired = list(pre_run_items) + replay
|
|
if len(desired) <= len(pre_run_items):
|
|
return
|
|
try:
|
|
await replace_session_items(session, desired)
|
|
except Exception:
|
|
logger.exception("salvaging crashed run history failed for %s", agent_id)
|
|
|
|
|
|
async def _seed_and_prepare_first_input(
|
|
session: Session | None, initial_input: Any, *, start_parked: bool
|
|
) -> Any:
|
|
"""Persist the opening input up front so it survives a first-turn crash."""
|
|
if initial_input and session is not None and not start_parked:
|
|
with contextlib.suppress(Exception):
|
|
if await seed_initial_input(session, initial_input):
|
|
return []
|
|
return initial_input
|
|
|
|
|
|
async def run_agent_loop(
|
|
*,
|
|
agent: Any,
|
|
initial_input: Any,
|
|
run_config: RunConfig,
|
|
context: dict[str, Any],
|
|
max_turns: int,
|
|
coordinator: AgentCoordinator,
|
|
agent_id: str,
|
|
interactive: bool,
|
|
session: Session | None = None,
|
|
start_parked: bool = False,
|
|
event_sink: StreamEventSink | None = None,
|
|
hooks: RunHooks[dict[str, Any]] | None = None,
|
|
) -> RunResultBase | None:
|
|
await coordinator.attach_runtime(
|
|
agent_id,
|
|
session=session,
|
|
interrupt_on_message=interactive,
|
|
)
|
|
result: RunResultBase | None = None
|
|
|
|
first_cycle_input = await _seed_and_prepare_first_input(
|
|
session, initial_input, start_parked=start_parked
|
|
)
|
|
|
|
budget_stopped = coordinator.budget_stopped
|
|
reserve_stopped = coordinator.reserve_stopped
|
|
if budget_stopped:
|
|
await coordinator.set_status(agent_id, "stopped")
|
|
raise BudgetExceededError("scan budget reached")
|
|
if reserve_stopped and context.get("parent_id") is not None:
|
|
await coordinator.set_status(agent_id, "stopped")
|
|
raise SubagentBudgetReservedError("scan reached the sub-agent budget reserve")
|
|
|
|
if reserve_stopped and start_parked and interactive and context.get("parent_id") is None:
|
|
await coordinator.send(agent_id, _reserve_notice())
|
|
|
|
if not (start_parked and interactive):
|
|
with contextlib.suppress(BudgetPausedError):
|
|
result = await _run_until_lifecycle(
|
|
agent,
|
|
coordinator,
|
|
agent_id,
|
|
initial_input=first_cycle_input,
|
|
run_config=run_config,
|
|
context=context,
|
|
max_turns=max_turns,
|
|
session=session,
|
|
interactive=interactive,
|
|
event_sink=event_sink,
|
|
hooks=hooks,
|
|
)
|
|
|
|
if not interactive:
|
|
return result
|
|
|
|
while True:
|
|
timeout = await _plain_waiting_timeout(coordinator, agent_id)
|
|
try:
|
|
woke = await coordinator.wait_for_message(agent_id, timeout=timeout)
|
|
except asyncio.CancelledError:
|
|
return result
|
|
|
|
if coordinator.budget_stopped:
|
|
await coordinator.set_status(agent_id, "stopped")
|
|
raise BudgetExceededError("scan budget reached")
|
|
|
|
if coordinator.reserve_stopped and context.get("parent_id") is not None:
|
|
await coordinator.set_status(agent_id, "stopped")
|
|
raise SubagentBudgetReservedError("scan reached the sub-agent budget reserve")
|
|
|
|
if woke:
|
|
# Real input is real progress, so the nudge budget starts over. A bare
|
|
# auto-resume is not: it must not hand a wedged agent a fresh budget.
|
|
await coordinator.reset_recovery(agent_id)
|
|
await coordinator.reset_idle_resumes(agent_id)
|
|
else:
|
|
idle_resumes = await coordinator.record_idle_resume(agent_id)
|
|
if idle_resumes >= _MAX_IDLE_AUTO_RESUMES:
|
|
logger.warning(
|
|
"agent %s auto-resumed %d times without hearing from anyone; "
|
|
"leaving it parked until a real message arrives",
|
|
agent_id,
|
|
idle_resumes,
|
|
)
|
|
await coordinator.park_waiting(agent_id, wait_kind="stalled")
|
|
await _notify_parent_on_stall(coordinator, agent_id)
|
|
continue
|
|
logger.info("agent %s reached its waiting timeout; auto-resuming", agent_id)
|
|
await coordinator.send(
|
|
agent_id,
|
|
{
|
|
"from": "system",
|
|
"type": "auto_resume",
|
|
"content": "Waiting timeout reached. Resuming execution.",
|
|
},
|
|
interrupt=False,
|
|
)
|
|
|
|
await coordinator.consume_pending(agent_id)
|
|
with contextlib.suppress(BudgetPausedError):
|
|
result = await _run_until_lifecycle(
|
|
agent,
|
|
coordinator,
|
|
agent_id,
|
|
initial_input=[],
|
|
run_config=run_config,
|
|
context=context,
|
|
max_turns=max_turns,
|
|
session=session,
|
|
interactive=True,
|
|
event_sink=event_sink,
|
|
hooks=hooks,
|
|
)
|
|
|
|
|
|
async def spawn_child_agent(
|
|
*,
|
|
coordinator: AgentCoordinator,
|
|
factory: Any,
|
|
agents_db_path: Path,
|
|
sessions_to_close: list[SQLiteSession],
|
|
run_config: RunConfig,
|
|
max_turns: int,
|
|
interactive: bool,
|
|
parent_ctx: dict[str, Any],
|
|
name: str,
|
|
task: str,
|
|
skills: list[str],
|
|
parent_history: list[Any],
|
|
event_sink: StreamEventSink | None = None,
|
|
hooks: RunHooks[dict[str, Any]] | None = None,
|
|
) -> dict[str, Any]:
|
|
parent_id = parent_ctx.get("agent_id")
|
|
if not isinstance(parent_id, str):
|
|
raise TypeError("Parent agent_id missing from context")
|
|
|
|
child_id = uuid.uuid4().hex[:8]
|
|
child_agent = factory(name=name, skills=skills)
|
|
await coordinator.register(
|
|
child_id,
|
|
name,
|
|
parent_id,
|
|
task=task,
|
|
skills=skills,
|
|
)
|
|
|
|
await _start_child_runner(
|
|
parent_ctx=parent_ctx,
|
|
coordinator=coordinator,
|
|
agents_db_path=agents_db_path,
|
|
sessions_to_close=sessions_to_close,
|
|
run_config=run_config,
|
|
max_turns=max_turns,
|
|
interactive=interactive,
|
|
child_agent=child_agent,
|
|
child_id=child_id,
|
|
name=name,
|
|
parent_id=parent_id,
|
|
task=task,
|
|
initial_input=child_initial_input(
|
|
name=name,
|
|
child_id=child_id,
|
|
parent_id=parent_id,
|
|
task=task,
|
|
parent_history=parent_history,
|
|
),
|
|
event_sink=event_sink,
|
|
hooks=hooks,
|
|
)
|
|
|
|
return {
|
|
"success": True,
|
|
"agent_id": child_id,
|
|
"name": name,
|
|
"parent_id": parent_id,
|
|
"message": f"Spawned '{name}' ({child_id}) running in parallel.",
|
|
}
|
|
|
|
|
|
async def respawn_subagents(
|
|
*,
|
|
coordinator: AgentCoordinator,
|
|
factory: Any,
|
|
agents_db_path: Path,
|
|
sessions_to_close: list[SQLiteSession],
|
|
run_config: RunConfig,
|
|
max_turns: int,
|
|
interactive: bool,
|
|
parent_ctx: dict[str, Any],
|
|
root_id: str,
|
|
event_sink: StreamEventSink | None = None,
|
|
hooks: RunHooks[dict[str, Any]] | None = None,
|
|
) -> None:
|
|
async with coordinator._lock:
|
|
agents_snapshot = [
|
|
(aid, status, dict(coordinator.metadata.get(aid, {})))
|
|
for aid, status in coordinator.statuses.items()
|
|
]
|
|
candidates: list[tuple[str, str, str | None, dict[str, Any]]] = []
|
|
for aid, status, md in agents_snapshot:
|
|
if not interactive and status not in {"running", "waiting"}:
|
|
continue
|
|
if coordinator.parent_of.get(aid) is None or aid == root_id:
|
|
continue
|
|
md["_restored_status"] = status
|
|
candidates.append(
|
|
(
|
|
aid,
|
|
coordinator.names.get(aid, aid),
|
|
coordinator.parent_of.get(aid),
|
|
md,
|
|
)
|
|
)
|
|
|
|
for child_id, name, parent_id, md in candidates:
|
|
try:
|
|
restored_status = str(md.get("_restored_status") or "running")
|
|
start_parked = interactive and restored_status != "running"
|
|
|
|
if start_parked:
|
|
logger.warning(
|
|
"respawn %s (%s): starting parked from status=%s",
|
|
child_id,
|
|
name,
|
|
restored_status,
|
|
)
|
|
|
|
child_skills = list(md.get("skills") or [])
|
|
child_agent = factory(name=name, skills=child_skills)
|
|
await _start_child_runner(
|
|
parent_ctx=parent_ctx,
|
|
coordinator=coordinator,
|
|
agents_db_path=agents_db_path,
|
|
sessions_to_close=sessions_to_close,
|
|
run_config=run_config,
|
|
max_turns=max_turns,
|
|
interactive=interactive,
|
|
child_agent=child_agent,
|
|
child_id=child_id,
|
|
name=name,
|
|
parent_id=parent_id,
|
|
task=str(md.get("task", "")),
|
|
initial_input=[],
|
|
start_parked=start_parked,
|
|
event_sink=event_sink,
|
|
hooks=hooks,
|
|
)
|
|
logger.info(
|
|
"respawned %s (%s) parent=%s task_len=%d",
|
|
child_id,
|
|
name,
|
|
parent_id or "-",
|
|
len(md.get("task", "")),
|
|
)
|
|
except Exception:
|
|
logger.exception("respawn %s failed; marking crashed", child_id)
|
|
with contextlib.suppress(Exception):
|
|
await coordinator.set_status(child_id, "crashed")
|
|
|
|
|
|
_INTERACTIVE_TOOL_RECOVERY_LIMIT = 3
|
|
|
|
|
|
async def _run_until_lifecycle(
|
|
agent: Any,
|
|
coordinator: AgentCoordinator,
|
|
agent_id: str,
|
|
*,
|
|
initial_input: Any,
|
|
run_config: RunConfig,
|
|
context: dict[str, Any],
|
|
max_turns: int,
|
|
session: Session | None,
|
|
interactive: bool,
|
|
event_sink: StreamEventSink | None,
|
|
hooks: RunHooks[dict[str, Any]] | None,
|
|
) -> RunResultBase | None:
|
|
"""Drive an agent until an explicit lifecycle tool settles its status.
|
|
|
|
A turn that ends without ``finish_scan``, ``agent_finish``,
|
|
``respond_to_user``, or ``wait_for_agents`` leaves the agent ``running``:
|
|
plain text never terminates a run and never yields to the user. Such a turn
|
|
is nudged back into a tool call, bounded by a recovery limit.
|
|
"""
|
|
result: RunResultBase | None = None
|
|
input_data: Any = initial_input
|
|
recovery_limit = _INTERACTIVE_TOOL_RECOVERY_LIMIT if interactive else max(1, max_turns)
|
|
|
|
while True:
|
|
if coordinator.budget_stopped:
|
|
await coordinator.set_status(agent_id, "stopped")
|
|
raise BudgetExceededError("scan budget reached")
|
|
|
|
if coordinator.reserve_stopped and context.get("parent_id") is not None:
|
|
await coordinator.set_status(agent_id, "stopped")
|
|
raise SubagentBudgetReservedError("scan reached the sub-agent budget reserve")
|
|
|
|
if interactive:
|
|
result = await _run_cycle_parked(
|
|
agent,
|
|
coordinator,
|
|
agent_id,
|
|
input_data=input_data,
|
|
run_config=run_config,
|
|
context=context,
|
|
max_turns=max_turns,
|
|
session=session,
|
|
event_sink=event_sink,
|
|
hooks=hooks,
|
|
)
|
|
else:
|
|
result = await _run_cycle(
|
|
agent,
|
|
coordinator,
|
|
agent_id,
|
|
input_data=input_data,
|
|
run_config=run_config,
|
|
context=context,
|
|
max_turns=max_turns,
|
|
session=session,
|
|
interactive=False,
|
|
event_sink=event_sink,
|
|
hooks=hooks,
|
|
)
|
|
|
|
status = await _agent_status(coordinator, agent_id)
|
|
if status != "running":
|
|
await coordinator.reset_recovery(agent_id)
|
|
return result
|
|
|
|
recoveries = await coordinator.record_recovery(agent_id)
|
|
logger.warning(
|
|
"agent %s ended a turn without a lifecycle tool call (interactive=%s); "
|
|
"forcing tool continuation (%d/%d): %s",
|
|
agent_id,
|
|
interactive,
|
|
recoveries,
|
|
recovery_limit,
|
|
_final_output_preview(result),
|
|
)
|
|
|
|
if recoveries >= recovery_limit:
|
|
return await _exhausted_recovery(coordinator, agent_id, result, interactive=interactive)
|
|
|
|
input_data = await _append_tool_required_message(
|
|
session=session,
|
|
context=context,
|
|
attempt=recoveries,
|
|
limit=recovery_limit,
|
|
interactive=interactive,
|
|
)
|
|
|
|
|
|
async def _exhausted_recovery(
|
|
coordinator: AgentCoordinator,
|
|
agent_id: str,
|
|
result: RunResultBase | None,
|
|
*,
|
|
interactive: bool,
|
|
) -> RunResultBase | None:
|
|
"""Settle an agent that never recovered into a tool call.
|
|
|
|
Interactive runs park instead of dying: a human is attached and can message
|
|
any agent, so the scan stays resumable. Autonomous runs have nobody to
|
|
resume them, so they fail loudly.
|
|
"""
|
|
if not interactive:
|
|
await coordinator.set_status(agent_id, "crashed")
|
|
await _notify_parent_on_terminal(coordinator, agent_id, "crashed")
|
|
raise MaxTurnsExceeded(
|
|
"Agent exhausted recovery attempts without calling finish_scan or agent_finish."
|
|
)
|
|
|
|
logger.warning(
|
|
"agent %s exhausted tool-call recovery attempts; parking until a message arrives",
|
|
agent_id,
|
|
)
|
|
await coordinator.park_waiting(agent_id, wait_kind="stalled")
|
|
# A parked child owes its parent a completion report it can no longer send. The
|
|
# parent is an agent, not a watching human, so nothing else tells it to stop
|
|
# waiting and it burns its full timeout on a message that is never coming.
|
|
await _notify_parent_on_stall(coordinator, agent_id)
|
|
return result
|
|
|
|
|
|
_WAITING_AUTO_RESUME_TIMEOUT_S = 300.0
|
|
|
|
# An agent that parks again after every auto-resume makes no progress, so stop
|
|
# spending a model turn per timeout and leave it parked for a real message.
|
|
_MAX_IDLE_AUTO_RESUMES = 3
|
|
|
|
|
|
async def _plain_waiting_timeout(
|
|
coordinator: AgentCoordinator,
|
|
agent_id: str,
|
|
) -> float | None:
|
|
"""Auto-resume timeout for a parked agent; None waits until a message arrives.
|
|
|
|
Driven by what the agent is waiting on, not by where it sits in the graph:
|
|
the user can message any agent, so an agent awaiting a human parks
|
|
indefinitely whether or not it is the root. Only an agent awaiting other
|
|
agents is re-checked on a timer, and only until it has spent its idle
|
|
budget re-parking without hearing anything.
|
|
"""
|
|
async with coordinator._lock:
|
|
status = coordinator.statuses.get(agent_id)
|
|
has_error = agent_id in coordinator.errors
|
|
runtime = coordinator.runtimes.get(agent_id)
|
|
gated = runtime.user_wake_required if runtime is not None else False
|
|
wait_kind = coordinator.wait_kinds.get(agent_id)
|
|
idle_resumes = coordinator.idle_resume_counts.get(agent_id, 0)
|
|
if status != "waiting" or has_error or gated:
|
|
return None
|
|
if wait_kind != "agents" or idle_resumes >= _MAX_IDLE_AUTO_RESUMES:
|
|
return None
|
|
return _WAITING_AUTO_RESUME_TIMEOUT_S
|
|
|
|
|
|
async def _run_cycle_parked(
|
|
agent: Any,
|
|
coordinator: AgentCoordinator,
|
|
agent_id: str,
|
|
*,
|
|
input_data: Any,
|
|
run_config: RunConfig,
|
|
context: dict[str, Any],
|
|
max_turns: int,
|
|
session: Session | None,
|
|
event_sink: StreamEventSink | None,
|
|
hooks: RunHooks[dict[str, Any]] | None,
|
|
) -> RunResultBase | None:
|
|
"""Interactive run cycle that parks on any error instead of killing the runner."""
|
|
try:
|
|
return await _run_cycle(
|
|
agent,
|
|
coordinator,
|
|
agent_id,
|
|
input_data=input_data,
|
|
run_config=run_config,
|
|
context=context,
|
|
max_turns=max_turns,
|
|
session=session,
|
|
interactive=True,
|
|
event_sink=event_sink,
|
|
hooks=hooks,
|
|
)
|
|
except (BudgetExceededError, BudgetPausedError, SubagentBudgetReservedError):
|
|
raise
|
|
except Exception as exc:
|
|
logger.exception("error escaped the run cycle for %s; parking as failed", agent_id)
|
|
await coordinator.set_status(agent_id, "failed", error=str(exc) or type(exc).__name__)
|
|
await _notify_parent_on_terminal(coordinator, agent_id, "failed")
|
|
return None
|
|
|
|
|
|
async def _run_cycle( # noqa: PLR0912, PLR0915
|
|
agent: Any,
|
|
coordinator: AgentCoordinator,
|
|
agent_id: str,
|
|
*,
|
|
input_data: Any,
|
|
run_config: RunConfig,
|
|
context: dict[str, Any],
|
|
max_turns: int,
|
|
session: Session | None,
|
|
interactive: bool,
|
|
event_sink: StreamEventSink | None,
|
|
hooks: RunHooks[dict[str, Any]] | None,
|
|
) -> RunResultBase | None:
|
|
image_strips = 0
|
|
compactions = 0
|
|
model_retries = 0
|
|
while True:
|
|
stream: Any = None
|
|
pre_run_items: list[Any] = []
|
|
try:
|
|
await coordinator.mark_running(agent_id)
|
|
if session is not None:
|
|
max_images = context.get("max_context_images")
|
|
if isinstance(max_images, int):
|
|
try:
|
|
await enforce_image_budget(session, max_images)
|
|
except Exception:
|
|
logger.exception("image-budget enforcement failed for %s", agent_id)
|
|
try:
|
|
await _compact_session(agent, session, run_config, force=False)
|
|
except Exception:
|
|
logger.exception("proactive compaction failed for %s", agent_id)
|
|
with contextlib.suppress(Exception):
|
|
pre_run_items = list(await session.get_items())
|
|
stream = Runner.run_streamed(
|
|
agent,
|
|
input=input_data,
|
|
run_config=run_config,
|
|
context=context,
|
|
max_turns=max_turns,
|
|
session=session,
|
|
hooks=hooks,
|
|
)
|
|
await coordinator.attach_stream(agent_id, stream)
|
|
try:
|
|
try:
|
|
async for event in stream.stream_events():
|
|
if event_sink is not None:
|
|
try:
|
|
event_sink(agent_id, event)
|
|
except Exception:
|
|
logger.exception("stream event sink failed for %s", agent_id)
|
|
if stream.run_loop_exception is not None:
|
|
raise stream.run_loop_exception
|
|
if refusal := _structured_provider_refusal(stream):
|
|
raise ProviderRefusalError(refusal)
|
|
except (BudgetExceededError, BudgetPausedError, SubagentBudgetReservedError):
|
|
raise
|
|
except RuntimeError as stream_exc:
|
|
if "after shutdown" not in str(stream_exc):
|
|
raise
|
|
logger.warning(
|
|
"Ignoring LiteLLM end-of-stream shutdown race for %s",
|
|
agent_id,
|
|
)
|
|
except (ExecTransportError, docker_errors.NotFound):
|
|
if not coordinator.is_shutting_down:
|
|
raise
|
|
logger.warning(
|
|
"Ignoring sandbox container error during teardown for %s",
|
|
agent_id,
|
|
exc_info=True,
|
|
)
|
|
finally:
|
|
await coordinator.detach_stream(agent_id, stream)
|
|
except BudgetPausedError as exc:
|
|
logger.info("agent %s paused at the scan budget limit: %s", agent_id, exc)
|
|
await coordinator.pause_for_budget(agent_id)
|
|
raise
|
|
except SubagentBudgetReservedError as exc:
|
|
logger.info("sub-agent %s stopped at the budget reserve: %s", agent_id, exc)
|
|
await coordinator.set_status(agent_id, "stopped")
|
|
await _notify_root_on_budget_reserve(coordinator)
|
|
raise
|
|
except BudgetExceededError as exc:
|
|
logger.info(
|
|
"agent %s reached the scan budget limit; stopping the scan: %s", agent_id, exc
|
|
)
|
|
await coordinator.set_status(agent_id, "stopped")
|
|
await coordinator.trigger_budget_stop()
|
|
raise
|
|
except Exception as exc:
|
|
if (
|
|
image_strips < 3
|
|
and session is not None
|
|
and getattr(exc, "status_code", None) in _INPUT_REJECTION_CODES
|
|
):
|
|
try:
|
|
stripped = await strip_all_images_from_session(session)
|
|
except Exception:
|
|
logger.exception("image-strip recovery failed for %s", agent_id)
|
|
stripped = False
|
|
if stripped:
|
|
image_strips += 1
|
|
logger.info(
|
|
"Stripped images from %s session after rejection; retrying (%d)",
|
|
agent_id,
|
|
image_strips,
|
|
)
|
|
input_data = []
|
|
continue
|
|
if (
|
|
compactions < _MAX_COMPACTIONS_PER_CYCLE
|
|
and session is not None
|
|
and is_context_overflow(exc)
|
|
):
|
|
try:
|
|
compacted = await _compact_session(agent, session, run_config, force=True)
|
|
except Exception:
|
|
logger.exception("overflow compaction recovery failed for %s", agent_id)
|
|
compacted = False
|
|
if compacted:
|
|
compactions += 1
|
|
logger.info(
|
|
"Compacted %s session after context overflow; retrying (%d)",
|
|
agent_id,
|
|
compactions,
|
|
)
|
|
input_data = []
|
|
continue
|
|
if model_retries < _MAX_TRANSIENT_MODEL_RETRIES and _is_transient_model_error(exc):
|
|
model_retries += 1
|
|
delay = _transient_model_retry_delay(model_retries)
|
|
logger.warning(
|
|
"transient model/provider error for %s; replaying turn "
|
|
"(attempt %d/%d, backoff %.1fs): %r",
|
|
agent_id,
|
|
model_retries,
|
|
_MAX_TRANSIENT_MODEL_RETRIES,
|
|
delay,
|
|
exc,
|
|
)
|
|
await asyncio.sleep(delay)
|
|
if session is not None:
|
|
input_data = []
|
|
continue
|
|
if session is not None:
|
|
await _salvage_stream_to_session(session, pre_run_items, stream, agent_id)
|
|
if isinstance(exc, ProviderRefusalError):
|
|
logger.warning("agent %s refused by the model provider: %s", agent_id, exc)
|
|
await coordinator.set_status(agent_id, "failed", error=str(exc))
|
|
await _notify_parent_on_terminal(coordinator, agent_id, "failed")
|
|
return None
|
|
if not interactive:
|
|
raise
|
|
if isinstance(exc, MaxTurnsExceeded):
|
|
status: Status = "stopped"
|
|
elif isinstance(exc, UserError | AgentsException | APIError):
|
|
status = "failed"
|
|
else:
|
|
status = "crashed"
|
|
logger.exception("agent run failed for %s; parking as %s", agent_id, status)
|
|
await coordinator.set_status(agent_id, status, error=str(exc) or type(exc).__name__)
|
|
await _notify_parent_on_terminal(coordinator, agent_id, status)
|
|
return None
|
|
else:
|
|
return cast("RunResultBase | None", stream)
|
|
|
|
|
|
async def _agent_status(coordinator: AgentCoordinator, agent_id: str) -> Status | None:
|
|
async with coordinator._lock:
|
|
return coordinator.statuses.get(agent_id)
|
|
|
|
|
|
def _final_output_preview(result: RunResultBase | None) -> str:
|
|
final_output = getattr(result, "final_output", None)
|
|
if final_output is None:
|
|
return "<none>"
|
|
text = str(final_output).replace("\n", " ").strip()
|
|
if not text:
|
|
return "<empty>"
|
|
return text[:300]
|
|
|
|
|
|
async def _append_tool_required_message(
|
|
*,
|
|
session: Session | None,
|
|
context: dict[str, Any],
|
|
attempt: int,
|
|
limit: int,
|
|
interactive: bool,
|
|
) -> list[dict[str, str]]:
|
|
finish_tool = "finish_scan" if context.get("parent_id") is None else "agent_finish"
|
|
if interactive:
|
|
message = (
|
|
"Your previous message ended a turn without a tool call. Plain text never ends "
|
|
"execution and never hands control to the user: it is shown to the user, and the "
|
|
"run continues. Continue immediately and call exactly one tool. "
|
|
"If you have something to tell the user and nothing to do until they reply, "
|
|
"call respond_to_user. "
|
|
"If you are blocked waiting for another agent, call wait_for_agents. "
|
|
f"If the whole engagement is complete, call {finish_tool}. "
|
|
"Otherwise use the appropriate execution or planning tool. "
|
|
f"This is recovery attempt {attempt}/{limit}."
|
|
)
|
|
else:
|
|
message = (
|
|
"Your previous response ended the autonomous Strix run without a lifecycle tool "
|
|
"call. That is invalid in non-interactive mode; plain text final answers are "
|
|
"ignored. Continue immediately and call exactly one tool. "
|
|
f"If your work is complete, call {finish_tool}. "
|
|
"If you are blocked waiting for another agent, call wait_for_agents. "
|
|
"Otherwise use the appropriate execution or planning tool. "
|
|
f"This is recovery attempt {attempt}/{limit}."
|
|
)
|
|
item = {"role": "user", "content": message}
|
|
if session is None:
|
|
return [item]
|
|
|
|
await session.add_items([cast("TResponseInputItem", item)])
|
|
return []
|
|
|
|
|
|
_TERMINAL_NOTICE = {
|
|
"crashed": (
|
|
"[Agent crash] {name} ({agent_id}) terminated unexpectedly. "
|
|
"Stop waiting on this child unless you want to message it again."
|
|
),
|
|
"failed": (
|
|
"[Agent failed] {name} ({agent_id}) stopped with an error and will not "
|
|
"send a completion report. Stop waiting on this child unless you want to "
|
|
"message it again."
|
|
),
|
|
"stopped": (
|
|
"[Agent capped] {name} ({agent_id}) hit its turn limit and was stopped "
|
|
"before finishing. It will not send a completion report, so stop waiting "
|
|
"on this child; account for its capped subtask and continue."
|
|
),
|
|
}
|
|
|
|
|
|
_STALL_NOTICE = (
|
|
"[Agent stalled] {name} ({agent_id}) kept ending turns without a tool call and is "
|
|
"parked until it receives a message. It will not send a completion report on its "
|
|
"own: either message it with a concrete next step to unblock it, or stop waiting on "
|
|
"it and account for its unfinished subtask."
|
|
)
|
|
|
|
|
|
async def _notify_parent_on_stall(
|
|
coordinator: AgentCoordinator,
|
|
agent_id: str,
|
|
) -> None:
|
|
"""Tell the parent that a child parked mid-task, so it stops waiting blindly."""
|
|
async with coordinator._lock:
|
|
parent = coordinator.parent_of.get(agent_id)
|
|
name = coordinator.names.get(agent_id, agent_id)
|
|
if parent is None:
|
|
return
|
|
await coordinator.send(
|
|
parent,
|
|
{
|
|
"from": agent_id,
|
|
"type": "stalled",
|
|
"priority": "high",
|
|
"content": _STALL_NOTICE.format(name=name, agent_id=agent_id),
|
|
},
|
|
interrupt=False,
|
|
)
|
|
|
|
|
|
async def _notify_parent_on_terminal(
|
|
coordinator: AgentCoordinator,
|
|
agent_id: str,
|
|
status: str,
|
|
) -> None:
|
|
template = _TERMINAL_NOTICE.get(status)
|
|
if template is None:
|
|
return
|
|
async with coordinator._lock:
|
|
parent = coordinator.parent_of.get(agent_id)
|
|
name = coordinator.names.get(agent_id, agent_id)
|
|
if parent is None:
|
|
return
|
|
await coordinator.send(
|
|
parent,
|
|
{
|
|
"from": agent_id,
|
|
"type": status,
|
|
"priority": "high",
|
|
"content": template.format(name=name, agent_id=agent_id),
|
|
},
|
|
interrupt=False,
|
|
)
|
|
|
|
|
|
def _reserve_notice() -> dict[str, Any]:
|
|
return {
|
|
"from": "system",
|
|
"type": "budget_reserve_stop",
|
|
"priority": "high",
|
|
"content": (
|
|
"[Budget reserve] The scan has reached the sub-agent budget reserve: every "
|
|
"sub-agent is being force-stopped as soon as its in-flight turn completes, and "
|
|
"none will send a completion report. Their confirmed vulnerabilities are "
|
|
"already filed as they were found. Do not wait on any sub-agents and do not "
|
|
"spawn new ones — wrap up now and call finish_scan."
|
|
),
|
|
}
|
|
|
|
|
|
async def _notify_root_on_budget_reserve(coordinator: AgentCoordinator) -> None:
|
|
root = await coordinator.claim_reserve_notification()
|
|
if root is None:
|
|
return
|
|
await coordinator.send(root, _reserve_notice())
|
|
|
|
|
|
async def _start_child_runner(
|
|
*,
|
|
parent_ctx: dict[str, Any],
|
|
coordinator: AgentCoordinator,
|
|
agents_db_path: Path,
|
|
sessions_to_close: list[SQLiteSession],
|
|
run_config: RunConfig,
|
|
max_turns: int,
|
|
interactive: bool,
|
|
child_agent: Any,
|
|
child_id: str,
|
|
name: str,
|
|
parent_id: str | None,
|
|
task: str,
|
|
initial_input: Any,
|
|
start_parked: bool = False,
|
|
event_sink: StreamEventSink | None = None,
|
|
hooks: RunHooks[dict[str, Any]] | None = None,
|
|
) -> None:
|
|
session = open_agent_session(child_id, agents_db_path)
|
|
sessions_to_close.append(session)
|
|
await coordinator.attach_runtime(child_id, session=session)
|
|
|
|
child_ctx: dict[str, Any] = dict(parent_ctx)
|
|
child_ctx["agent_id"] = child_id
|
|
child_ctx["parent_id"] = parent_id
|
|
child_ctx["task"] = task
|
|
|
|
async def _child_loop() -> None:
|
|
# A budget stop is a clean scan-wide shutdown, not a child failure: the
|
|
# child's status and parent notification are already settled in
|
|
# ``_run_cycle``. Swallow it here so the detached task does not surface a
|
|
# spurious "Task exception was never retrieved" warning. The root agent
|
|
# hits the same limit on its next call and tears the scan down.
|
|
try:
|
|
await run_agent_loop(
|
|
agent=child_agent,
|
|
initial_input=initial_input,
|
|
run_config=run_config,
|
|
context=child_ctx,
|
|
max_turns=max_turns,
|
|
coordinator=coordinator,
|
|
agent_id=child_id,
|
|
interactive=interactive,
|
|
session=session,
|
|
start_parked=start_parked,
|
|
event_sink=event_sink,
|
|
hooks=hooks,
|
|
)
|
|
except BudgetExceededError:
|
|
logger.info("child %s stopped after reaching the scan budget limit", child_id)
|
|
except SubagentBudgetReservedError:
|
|
logger.info("child %s stopped at the sub-agent budget reserve", child_id)
|
|
|
|
task_handle = asyncio.create_task(_child_loop(), name=f"agent-{name}-{child_id}")
|
|
await coordinator.attach_runtime(child_id, task=task_handle)
|