mirror of
https://github.com/usestrix/strix.git
synced 2026-08-16 17:27:26 +02:00
58df71d3db
* let an agent wait on what it already said An agent that answers in plain text is nudged to call a tool, and the only tool that hands control back takes a required message. So it says the same thing twice: once as text the user has already read, once as the argument it had to supply to stop. Seen on a run whose whole instruction was "hi" - a greeting, then the same greeting again through respond_to_user. message is optional now. The nudge arms the tool with the text that was delivered and says not to repeat it, so an agent that has said its piece can park on it with an empty call. Anything it does want to add it passes normally. Parking still cannot leave the user on silence: an empty call is refused unless something was actually said, and the arming is single use - execution clears it as soon as a turn ends any other way. The interactive prompt now also says to answer and stop in one respond_to_user call, which is what avoids the nudge in the first place. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com> * drop the worked example from the interactive prompt "the user greeted you, asked something you can answer outright, or you need a decision" was the run I had been reading, written into a rule that holds whatever the reason. The rule is that replying and stopping is one call; listing occasions only invites the model to check whether this is one of them. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com> * drop the arming flag; an empty message just waits Passing the delivered text from execution into the tool, and refusing an empty call without it, was machinery guarding against an agent parking having said nothing. That leaves the user looking at "waiting for your reply" with a cursor in front of them - they type. It does not need a mechanism. What is left is the default on message, and the nudge saying the text already landed. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com> * only offer waiting on words that were written The nudge told every agent its text had already been delivered, but it fires whenever a turn leaves the agent running, and a turn can end with no tool call and no text at all - _final_output_preview has carried <none> and <empty> branches all along. An agent that said nothing was being invited to wait on an answer the user never received, leaving them at a bare prompt. It now reads the turn: waiting on what was said is offered only when something was, and otherwise the agent is told plainly that the user has read nothing and to send its message. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com> * leave the continuation nudge alone Rewording it meant asserting from the outside whether the agent had spoken, and the nudge fires whenever a turn leaves the agent running - text or no text. The agent knows which it did without being told, so the guidance belongs in its prompt, where the condition is its own to read. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com> * say it in the nudge, where the agent is reading An agent stranded by the nudge reasons off the nudge. Told only to call respond_to_user, it supplies a message, and since it has just answered in plain text that message is the same answer again. The system prompt saying otherwise sits thousands of tokens earlier and loses. The clause goes on the line the agent acts on: call respond_to_user, with no message if it has already said it. That reads true whatever the turn did, including one that produced no text, because the agent is the one who knows which — nothing here has to work it out from the outside. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com> --------- Co-authored-by: Claude Opus 4.8 <noreply@anthropic.com>
1033 lines
37 KiB
Python
1033 lines
37 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 isinstance(exc, MaxTurnsExceeded):
|
|
status: Status = "stopped"
|
|
elif isinstance(exc, UserError | AgentsException | APIError):
|
|
status = "failed"
|
|
else:
|
|
status = "crashed"
|
|
logger.exception("agent run failed for %s; marking %s", agent_id, status)
|
|
# Settle the status and wake the parent before the exception unwinds a
|
|
# non-interactive agent's task: a child that dies still owes its parent a
|
|
# report, and the parent would otherwise wait out its timeout on a message
|
|
# the dead child can no longer send.
|
|
await coordinator.set_status(agent_id, status, error=str(exc) or type(exc).__name__)
|
|
await notify_parent_on_terminal(coordinator, agent_id, status)
|
|
if not interactive:
|
|
raise
|
|
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 — with no message if you have already said it. "
|
|
"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 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 = {
|
|
"completed": (
|
|
"[Agent completed] {name} ({agent_id}) finished and is no longer running, but it "
|
|
"sent no completion report. Stop waiting on this child; ask it directly if you "
|
|
"need its results."
|
|
),
|
|
"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 stopped] {name} ({agent_id}) was stopped before finishing (turn limit "
|
|
"or an explicit stop). It will not send a completion report, so stop waiting "
|
|
"on this child; account for its unfinished 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
|
|
if not await coordinator.claim_parent_notice(agent_id):
|
|
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 _notify_parent_on_exit(
|
|
coordinator: AgentCoordinator,
|
|
agent_id: str,
|
|
) -> None:
|
|
"""Backstop for a child whose loop ended without telling its parent.
|
|
|
|
Every terminal state counts, including ``completed``: a child that skips its
|
|
completion report leaves the parent waiting on a message nobody will send.
|
|
"""
|
|
status = await _agent_status(coordinator, agent_id)
|
|
if status is None:
|
|
return
|
|
await notify_parent_on_terminal(coordinator, agent_id, status)
|
|
|
|
|
|
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)
|
|
finally:
|
|
if not coordinator.is_shutting_down:
|
|
await _notify_parent_on_exit(coordinator, child_id)
|
|
|
|
task_handle = asyncio.create_task(_child_loop(), name=f"agent-{name}-{child_id}")
|
|
await coordinator.attach_runtime(child_id, task=task_handle)
|