"""Sub-agent runtime spec for agent-team sessions.""" from __future__ import annotations import asyncio import contextlib import logging import os import re from collections.abc import Iterator from dataclasses import dataclass, field from pathlib import Path from typing import Any, ClassVar # Reuses ``INCOMPLETE_STOP_REASONS`` because "needs a forced final # answer" and "report is partial" are the same concept; keeping them # tied at the import boundary prevents drift between this module and # the fan-in formatter. from frontier_agent.components.agent_bus.fan_in import ( INCOMPLETE_STOP_REASONS as _FORCED_FINAL_STOP_REASONS, ) from frontier_agent.components.agent_bus.models import ( SubAgentResult, SubAgentRuntimeSpec, SubTask, ) from frontier_agent.components.finalization import ( COMMON_RECOVERY_NUDGE_PREFIXES, build_recovery_context, chat_with_fallback_budget, minimal_best_effort_answer, ) from frontier_agent.components.observers.context_size_guard import ContextSizeGuard from frontier_agent.core.loop_types import ( AgentLoopResult, LoopConfig, LoopPolicy, ToolResult, ) from frontier_agent.core.messages import ( Message, assistant_msg_with_reasoning, is_assistant_msg, is_tool_msg, text_of, user_msg, ) from frontier_agent.core.runtime import registry from frontier_agent.core.runtime.loop.compact import ( URL_RE, KeepLastNToolResultsCompactor, MessageCompactor, ) from frontier_agent.core.runtime.loop.tiered_compact import ( InputTokenGauge, InputTokenThresholdPolicy, TieredCompactor, compaction_trigger_tokens, ) from frontier_agent.core.runtime.loop.tool_exec import ToolResultPostProcessor from frontier_agent.state.event_store.sqlite import EventStore from plugins.tools.bash import BASH_STDERR_SEPARATOR from workflows.agent_team.identity import MAIN_AGENT_ID, SUB_ROLE_ID, llm_session_id from workflows.agent_team.stream_repetition import StreamRepetitionConfig logger = logging.getLogger(__name__) SWARM_SCOPE_KEY = "swarm_subagent_runtime" """ExecutionScope.metadata key carrying :class:`SwarmSubagentRuntime`.""" def render_sandbox_fs_note( *, sandbox_mode: str, inputs_available: bool, audience: str, ) -> str: """Render truthful filesystem context for the active sandbox topology.""" shared = sandbox_mode in ("container", "native") location = "this native workspace" if sandbox_mode == "native" else "this container" workspace_path = ( os.environ.get("FRONTIER_AGENT_WORKSPACE_DIR", "the workspace") if sandbox_mode == "native" else "/workspace" ) inputs_path = ( os.environ.get("FRONTIER_AGENT_INPUTS_DIR", "the inputs directory") if sandbox_mode == "native" else "/inputs" ) outputs_path = ( os.environ.get("FRONTIER_AGENT_OUTPUTS_DIR", "the outputs directory") if sandbox_mode == "native" else "/outputs" ) project_path = os.environ.get("FRONTIER_AGENT_PROJECT_DIR", "").strip() if audience == "sub": workspace = ( f"{workspace_path} is shared by the coordinator and all sub-agents in " f"{location}; use distinctive subdirectories and never assume " "unlabelled files are yours." if shared else f"{workspace_path} is private to this sub-agent." ) else: workspace = ( f"{workspace_path} is shared by the coordinator and all sub-agents in " f"{location}." if shared else f"{workspace_path} contains the coordinator worktree and the " "sub-agents' private worktree directories." ) inputs = ( f"Task input files are available under {inputs_path}." if inputs_available else f"No task input files were provided; {inputs_path} is empty. Do not probe it." ) # Positive read-context guidance: /workspace is not just where YOU write — # in container mode it is the shared read source for teammates' candidates. # In bwrap mode each sub-agent's /workspace is private, so cross-agent reuse # can only come through the coordinator's report text. collab_read = ( " To build on a teammate's work, read their candidate directly from " f"{workspace_path} at the path named in that agent's report — {workspace_path} is " "the team's shared read source for intermediate products, alongside " f"{inputs_path} for external input files." if shared else " To reuse another agent's work, rely on the report text " "the coordinator passes in; each sub-agent's /workspace is private and " "not readable by other agents here." ) optional_dependencies = ( "Scientific and plotting packages are optional. Office writer packages " "used by create_file are already installed. Do not pre-install optional " "packages; install only a package required by the task with " "`python -m pip install `. Native mode stores it in the " "workspace-local dependency overlay. " if sandbox_mode == "native" else "" ) project = "" if project_path and os.path.realpath(project_path) != os.path.realpath(workspace_path): project = ( f"The user's project/coding directory is {project_path}; read or " "edit repository files there, while keeping clones, downloads, " f"drafts, probes, and other intermediate work under {workspace_path}. " ) manifest_namespace = "" if audience == "main" and outputs_path != "/outputs": manifest_namespace = ( " When calling assign_task, however, output_paths is a virtual " "manifest namespace: always declare /outputs/, " f"never {outputs_path}/. The publisher runtime maps " "that declaration back to the physical root automatically." ) return ( "\n\nFILESYSTEM CONTEXT (runtime-resolved): " f"{workspace} {inputs}{collab_read} " f"{optional_dependencies}{project}" "All research notes, calculations, candidate artifacts, temporary " f"files, and verification material belong under {workspace_path}. {outputs_path} " "is the collected final-deliverable root and is read-only by default. " "Only one assignment, the one carrying an exact output_paths " "manifest, may write there; it must publish exactly that manifest and " "must not add README, confirmation, validation, duplicate, or alternate-" "format files unless declared. The one exception is the top-level " f"{outputs_path}/scratch/ directory: writable by every assignment, persisted " "across rounds for intermediate products worth reusing later, never " "shown to the user, with a 512MB quota (over-quota writes fail until " f"you delete files there). Only the literal {outputs_path}/scratch/ prefix " f"qualifies.{manifest_namespace}" ) _THINK_BLOCK_RE = re.compile(r"[\s\S]*?\s*") _AGED_TOOL_SENTINEL = "[aged-tool:" _REPORT_BLOCK_RE = re.compile(r"]*>[\s\S]*?") _STITCH_MAX_CHARS = 60_000 _SUBAGENT_NO_TOOL_NUDGE = ( "Every turn must end with a tool call. When you have gathered enough " "evidence, call `submit_report` with your Scope/Finding/Evidence report. " "Otherwise call a data-gathering tool (web_search, web_fetch, read_file, " "bash). Do not reply in plain text." ) _SUBAGENT_FINALIZATION_MESSAGE = ( "Your assignment is entering its finalization reserve. Stop new " "exploration. Finish and validate the best current artifact in /workspace; " "if this assignment has publish permission, publish exactly its declared " "/outputs manifest. Then call `submit_report` with concrete findings and " "file paths. Submit a useful partial report even if some work remains." ) _RECOVERY_NUDGE_PREFIXES = ( # This workflow's own sub-agent control messages, on top of the framework set. _SUBAGENT_NO_TOOL_NUDGE, _SUBAGENT_FINALIZATION_MESSAGE, "[stop signal] The coordinator has asked you to STOP immediately.", *COMMON_RECOVERY_NUDGE_PREFIXES, ) _RECOVERY_EMPTY_FALLBACK = ( "No reliable intermediate text was recoverable. Give the most useful " "honest answer possible from the original task." ) _RECOVERY_LABELS = { "system": "System guidance", "assistant": "Visible coordinator draft", "tool": "Collected observation or sub-agent report", "user": "User instruction", } # Some text-mode models under long-context fatigue may emit tool calls as raw # text in `content` (e.g. # ``{"name": ...}``) instead of going through the # structured ``tool_calls`` field. The runtime's text-mode parser still # dispatches these correctly, but the raw markup remains in the content # string. When the loop terminates abnormally and `force_final_answer` # nudges the model for a plain-text answer, the model often keeps # emitting tool-call markup because it is stuck in that mode. Strip # anything resembling tool-call XML before treating the response as the # final answer. _LEAKED_TOOL_CALL_BLOCK_RE = re.compile( r"<\s*tool_call[^>]*>[\s\S]*?<\s*/\s*tool_call\s*>", re.IGNORECASE, ) _LEAKED_TOOL_RESPONSE_BLOCK_RE = re.compile( r"<\s*tool_response[^>]*>[\s\S]*?<\s*/\s*tool_response\s*>", re.IGNORECASE, ) # Function-call style: ``...``. _LEAKED_FUNCTION_BLOCK_RE = re.compile( r"<\s*function\s*=[\s\S]*?<\s*/\s*function\s*>", re.IGNORECASE, ) # Stray opening tag with no matching close. _LEAKED_TAG_FRAGMENT_RE = re.compile( r"<\s*(tool_call|tool_response|function)\b[^>]*>?", re.IGNORECASE, ) def _strip_thinking(text: str) -> str: return _THINK_BLOCK_RE.sub("", text or "").strip() def _strip_leaked_tool_calls(text: str) -> str: """Remove text-mode tool-call markup that leaked into content. Returns the cleaned string. If everything was tool-call XML and the remainder is empty/whitespace, returns ``""`` so the caller can decide on a sentinel. """ if not text: return text out = _LEAKED_TOOL_CALL_BLOCK_RE.sub("", text) out = _LEAKED_TOOL_RESPONSE_BLOCK_RE.sub("", out) out = _LEAKED_FUNCTION_BLOCK_RE.sub("", out) out = _LEAKED_TAG_FRAGMENT_RE.sub("", out) return out.strip() def _finalization_prompt( task_description: str, *, structured_report: bool, ) -> str: if structured_report: return ( "Write the final user-facing report now using only the clean " "recovery context below. Directly answer the original task, state " "what was completed, name any deliverables that were produced, " "and clearly qualify remaining gaps. Use structured Markdown when " "helpful. Do not call tools and do not return an error sentinel." ) return ( "Provide the best final answer now using only the clean recovery " "context below. Preserve useful partial conclusions and mention any " "deliverables already produced. Do not call tools and do not return an " "error sentinel." ) def _build_recovery_messages( messages: list[Message], *, task_description: str, structured_report: bool, ) -> list[Message]: """Flatten history so malformed/orphan tool protocol cannot poison rescue.""" recovery_context = build_recovery_context( messages, strip_thinking=_strip_thinking, strip_leaked_tool_calls=_strip_leaked_tool_calls, nudge_prefixes=_RECOVERY_NUDGE_PREFIXES, empty_fallback=_RECOVERY_EMPTY_FALLBACK, labels=_RECOVERY_LABELS, ) prompt = ( f"{_finalization_prompt(task_description, structured_report=structured_report)}" "\n\nThe original tool-call fields were intentionally removed because " "they may be malformed. Treat all observations as partial progress.\n\n" f"Original task:\n{task_description or '(not repeated)'}\n\n" f"Clean recovery context:\n{recovery_context}" ) return [user_msg(prompt)] def _stitch_collected_reports(messages: list[Message]) -> str: """Deterministically preserve collected sub-agent reports without an LLM.""" blocks: list[str] = [] seen: set[str] = set() for message in messages: if not isinstance(message, dict) or not is_tool_msg(message): continue content = text_of(message.get("content")) for match in _REPORT_BLOCK_RE.finditer(content): block = match.group(0) if block in seen: continue seen.add(block) blocks.append(block) if not blocks: return "" stitched = "\n\n".join(blocks) if len(stitched) > _STITCH_MAX_CHARS: stitched = stitched[:_STITCH_MAX_CHARS] + "\n[... truncated ...]" return ( "## Best-effort collected results\n\n" "The synthesis step ended early, so the completed sub-agent reports " "below are delivered directly instead of returning no answer.\n\n" + stitched ) def _minimal_best_effort_answer( task_description: str, stopped_by: str, *, language: str = "", ) -> str: """Always provide a user-facing answer even when the rescue LLM is down.""" return minimal_best_effort_answer( task_description, stopped_by, language=language, ) async def force_final_answer( result: AgentLoopResult, llm: Any, timeout: float, *, task_description: str = "", structured_report: bool = False, language: str = "", ) -> AgentLoopResult: """Run one tool-free LLM call to extract a plain-text final answer. No-op when the loop already produced one (``finalize_answer`` / ``submit_report`` latched ``metadata["final_answer"]``) or when the exit reason is outside :data:`_FORCED_FINAL_STOP_REASONS`. Mutates ``result`` in place. The rescue request is rebuilt as one clean user message, so a malformed tool call in the original history cannot make the finalization request fail the same way. Best-effort — failures fall back to collected reports and finally an explicit user-facing partial result. """ if ( result.metadata.get("final_answer") or result.stopped_by not in _FORCED_FINAL_STOP_REASONS ): return result recovery_messages = _build_recovery_messages( list(result.messages), task_description=task_description, structured_report=structured_report, ) ask_timeout = max(float(timeout), 1.0) async def _ask() -> str: resp = await chat_with_fallback_budget( llm, recovery_messages, per_leg_timeout_s=ask_timeout, ) content = getattr(resp, "content", "") or "" # This rescue assistant turn is appended directly to history. It # bypasses the kernel normalizer, so route reasoning through the # shared format-aware builder (tag default) — a bare # ``reasoning_content`` field would land off the served # checkpoint's expected wire shape. cleaned = _strip_leaked_tool_calls(_strip_thinking(text_of(content))) if cleaned: result.messages.extend(recovery_messages) result.messages.append(assistant_msg_with_reasoning( cleaned, getattr(resp, "reasoning_content", "") or "", thinking_format="tag", )) return cleaned rescue_mode = "deterministic_fallback" try: text = await _ask() if text: rescue_mode = "clean_context_llm" except Exception as exc: logger.warning("force_final_answer failed: %s", exc) text = "" # If the clean recovery call still returns only leaked markup, fall back to # the original ``final_content`` (also stripped), completed sub-agent # reports, or an explicit best-effort status rather than scoring raw # tool-call XML. This is the *terminal* answer, so the coordinator's own # visible draft outranks the raw reports — it is already a synthesis, and # nothing downstream will turn the reports into one. # ``prepare_report_handoff`` deliberately reverses these two: there the # reporter is still to come, and raw reports are the better input for it. if not text: text = _strip_leaked_tool_calls(_strip_thinking(result.final_content)) if text: rescue_mode = "existing_partial" if not text: text = _stitch_collected_reports(result.messages) if text: rescue_mode = "collected_reports" if not text: text = _minimal_best_effort_answer( task_description, result.stopped_by, language=language, ) result.metadata["final_answer"] = text result.metadata["final_answer_rescued"] = True result.metadata["final_answer_rescue_mode"] = rescue_mode result.metadata["final_answer_source"] = rescue_mode result.final_content = text logger.info( "force_final_answer injected after %s (%d chars)", result.stopped_by, len(text), ) return result def prepare_report_handoff( result: AgentLoopResult, *, task_description: str = "", language: str = "", ) -> AgentLoopResult: """Prepare a non-empty baseline before an optional reporter runs. Completed reports outrank a truncated coordinator draft because they carry more evidence and remain usable if the reporter fails open. """ if result.stopped_by not in _FORCED_FINAL_STOP_REASONS: return result source = str(result.metadata.get("final_answer_source") or "") text = _strip_leaked_tool_calls(_strip_thinking(str( result.metadata.get("final_answer") or "", ))) if text and not source: source = "agent" if not text: text = _stitch_collected_reports(result.messages) if text: source = "collected_reports" if not text: text = _strip_leaked_tool_calls(_strip_thinking(result.final_content)) if text: source = "existing_partial" if not text: text = _minimal_best_effort_answer( task_description, result.stopped_by, language=language, ) source = "deterministic_fallback" result.metadata["final_answer"] = text result.metadata["final_answer_source"] = source result.metadata["report_handoff"] = True result.metadata["report_handoff_reason"] = result.stopped_by result.final_content = text logger.info( "prepared reporter handoff after %s (%d chars)", result.stopped_by, len(text), ) return result @dataclass(frozen=True) class SwarmSubagentRuntime: """Per-task config the main agent threads to its sub-agents. The main-agent node populates one of these per loop run and passes it through ``run_agent_loop(scope_metadata={SWARM_SCOPE_KEY: …})``. ``create_subagent`` reads it back to build a :class:`SubAgentRuntimeSpec` aligned with the main agent's profile. """ original_question: str = "" trajectory_dir: Path | None = None sub_agent_llm: Any = None sub_agent_llm_timeout: float | None = None sub_agent_tool_timeout: float | None = None reasoning_only_timeout_s: float | None = None reasoning_only_max_tokens: int | None = None logical_call_timeout_s: float | None = None sub_agent_max_turns: int | None = None # Reserve several tool-enabled turns to finish assigned artifacts and # submit a report before LastTurnForcer strips tools on the landing turn. finalization_reserve_turns: int = 6 sub_agent_thinking_in_history: bool = False sub_agent_thinking_history_max_tokens: int | None = None # Compaction (loaded from profile yaml `agent:` section). Defaults # suit long-context tool use; ``keep_tool_result=-1`` disables. sub_keep_tool_result: int = 10 sub_compact_after_turns: int = 100 sub_context_token_limit: int = 180_000 # Tiered context compaction (mirrors stateful_react); "off" (default) keeps # the legacy keep-last-N path. "tiered" → compact at max_len*0.8 (REAL input # tokens): Tier1 keep-last-N tool results, Tier2 LLM summary if not enough. context_compaction: str = "off" compaction_spill: bool = False max_len: int = 0 max_input_tokens: int | None = None tier1_keep_tool_result: int = 5 keep_recent_turns: int = 5 sub_agent_model_profile: Any = None fs_mode: bool = False # ``None`` (default) disables BudgetObserver — max_turns is the only bound. max_tokens: int | None = None # StuckTargetGuard (0 = off): confirmed failures for one host within the # last ``stuck_target_window`` network turns. Success resets the host; the # escalation threshold quarantines it. With no token budget and # ``sub_max_turns`` at 100, a sub-agent can otherwise spend its allowance on # one unreachable URL. Tighter than the stateful_react defaults (6/10/20) # because a sub-agent works one narrow sub-question. stuck_target_hint_after: int = 5 stuck_target_escalate_after: int = 8 stuck_target_window: int = 15 # Profile's declared per-call OUTPUT cap (``llm.max_tokens``) — the model's # real ``max_completion_tokens`` ceiling. Sub-agents want a big output # budget for full reports, but must never REQUEST more than the model # accepts or the server 400s the very first turn. ``_bind_sub_agent_llm`` # caps its desired budget by this. ``None`` → unknown (no profile) → keep # the legacy hardcoded desired value. llm_max_tokens: int | None = None tool_result_max_chars: int = 15_000 # When True (default, debug-friendly): each task on a sticky session # writes a separate ``.tNN.{json,jsonl}`` so reuse is directly # observable from the filesystem. When False: # all tasks on a session share a single ``.{json,jsonl}`` # that is atomic-replaced after every event, keeping a stable on-disk shape # across reused sessions. trajectory_per_task: bool = True # When set, sub-agents also write # ``__sub_.json`` next to the existing # TrajectoryFileObserver output. The two writers coexist — they # target different files. worker_trace_dir: Path | None = None workflow_id: str = "" # A second, streaming trace: when both are set, sub-agents also attach # the caller's protocol stream observer so their LLM calls / tool calls / # per-call ``usage`` events flow to stdout through the same emitter # (single line-buffered stdout, atomic writes) and feed the same usage # aggregator, so ``final.usage`` covers the main agent plus every # sub-agent instead of the main agent alone. Both objects come from the # optional external protocol layer — see ``workflows/_shared/sdk_shim``. protocol_emitter: Any = None protocol_usage_aggregator: Any = None mcp_tool_names: list[str] | None = None mcp_tool_specs: list[dict[str, Any]] | None = None stream_repetition_config: StreamRepetitionConfig | None = None # Per-run resolved sub-agent tools. ``None`` preserves the role default # for callers that do not set it; agent_team sets this from profile config # without mutating the process-global AgentRegistry. sub_agent_tool_names: list[str] | None = None # Concrete per-run tool objects. This lets profiles replace an implementation # while retaining the public tool name and without mutating the registry. sub_agent_tools: list[Any] | None = None # Fail-closed role id for this workflow's sub-agents. Distinct from swarm's # ``swarm_sub`` so the two workflows' AgentDefinitions don't overwrite each # other when both are dir-scanned into one registry (create_subagent reads # this via ``_resolve_sub_role_id``). sub_role_id: str = SUB_ROLE_ID # Per-task worktree (working dir, NOT the trajectory/log dir). When set # and bwrap is available, each sub-agent runs its tools inside a bwrap # sandbox whose ``/workspace`` is ``worktree_root/`` (private, # siblings invisible). ``sub_binds`` are the extra mounts shared by all # sub-agents — ``(/inputs ro, /outputs rw)`` — resolved by the main node # (subs write their deliverable to /outputs; see _resolve_sandbox_binds). worktree_root: Path | None = None sub_binds: tuple[tuple[str, str, bool], ...] = () # Sandbox isolation mode (set by the main node). ``container`` → every # sub-agent SHARES the one mounted /workspace (``shared_workspace_dir``) and # /outputs via a CurrentSandbox (the container is the isolation boundary, # no per-sub subdir, no bwrap). ``bwrap`` / ``auto`` → the legacy per-sub # bwrap sandbox path below. sandbox_mode: str = "auto" shared_workspace_dir: str = "" # Appended to every sub-agent's system prompt (e.g. the sandbox filesystem # convention). create_subagent reads it off the runtime; empty = no-op. sub_prompt_suffix: str = "" # Output publication is single-owner and manifest-bound. The dataclass is # frozen, so ``assign_task`` mutates this per-run state bag rather than # replacing fields; follow-ups must reuse the same agent and manifest. # Observers every sub-agent loop must also run. A host that wraps this # workflow in its own trust boundary (the HF Space demo's filesystem # containment and secret-argument guards) supplies them through # ``metadata['sdk_sub_agent_extra_observers']``; without this field they # would stop at the coordinator, and a worker's tool calls — which carry # the network tools — would execute unguarded. The host owns the instances, # so they are shared across every sub-agent loop of the run. extra_observers: tuple[Any, ...] = field(default=(), compare=False) publication_state: dict[str, Any] = field(default_factory=dict, compare=False) publication_lock: asyncio.Lock = field( default_factory=asyncio.Lock, compare=False, repr=False, ) class NoOpCompactor(MessageCompactor): """Disable mid-loop compaction — max_turns is the only bound.""" def compact( self, messages: list[Message], keep_recent: int, ) -> list[Message]: return messages class ReasoningStripCompactor(MessageCompactor): """Age old reasoning **and** tool-message bodies. Conservative — never drops messages. Why: ``NoOpCompactor``'s "no compaction needed" assumption breaks for thinking models *and* tool-heavy ReAct loops: 1. Some reasoning models return large reasoning blocks that the next request re-sends (``thinking_in_history=true`` + provider-specific ``reasoning_content`` round-trip). 2. Tool messages can also be large enough to exhaust the context window. Stripping ```` alone may not help when tool messages dominate. What: keeps the message *structure* intact (every ``ToolMessage`` keeps its ``tool_call_id`` so the matching AI ``tool_calls`` doesn't orphan — Azure rejects orphans with HTTP 400) and replaces the bulky parts of messages older than ``keep_recent``: * AI ``…`` blocks in ``content`` (tag-mode) * AI ``reasoning_content`` field (DeepSeek/GPT-5) * Tool ``content`` → one-line stub ``[aged-tool: url=] …`` so the LLM still knows what was fetched and can re-fetch the URL if needed. Recent ``keep_recent`` messages are untouched. Aging is idempotent: a tool stub is detected by its sentinel and left alone, so triggering every turn after ``compact_after_turns`` is cheap. """ _MAX_TOOL_PREVIEW = 120 # chars of body kept in the aged stub def compact( self, messages: list[Message], keep_recent: int, ) -> list[Message]: if len(messages) <= keep_recent: return messages cutoff = len(messages) - keep_recent for i in range(cutoff): msg = messages[i] if is_assistant_msg(msg): content = msg.get("content") if isinstance(content, str) and "" in content: msg["content"] = _THINK_BLOCK_RE.sub("", content) if "reasoning_content" in msg: msg.pop("reasoning_content", None) continue if is_tool_msg(msg): raw = msg.get("content") content = raw if isinstance(raw, str) else str(raw) if content.startswith(_AGED_TOOL_SENTINEL): continue tool_name = msg.get("name") or "tool" m = URL_RE.search(content) first_url = m.group(0) if m else "" preview = content[: self._MAX_TOOL_PREVIEW].replace("\n", " ").strip() head = f"{_AGED_TOOL_SENTINEL} {tool_name}" if first_url: head += f" url={first_url}" msg["content"] = ( f"{head}] {preview}…" if preview else f"{head}] (body aged out — re-fetch if needed)" ) return messages class SwarmToolResultPostProcessor(ToolResultPostProcessor): """Per-tool dispatch for fresh ``ToolMessage`` content. Fires inside the agent loop right after a tool returns and before the result enters the conversation history (``agent_loop.py`` step 8d). Replaces ``DefaultToolResultPostProcessor``'s blunt char cap with per-tool dispatch: content-retrieval tools (web_fetch, web_search, scholar_search) pass through unmodified — academic-paper results sections and long Wikipedia articles need their full body. Compute tools (bash, run_python_code) keep tight head/tail caps because their output is unbounded by design (``find /``, runaway ``print``). Web results pass through because relevant evidence may occur after an abstract or introduction. Cap upstream at ``TOOL_RESULT_MAX_CHARS``, let ``ReasoningStripCompactor`` age old bodies to URL stubs at 200K, and trust the Summary LLM (``info_to_extract`` on web_fetch) for in-call compression when the agent asks for it. Errors (``is_error=True``) bypass all caps — the model needs the full error text to recover. ``delegate_subtask`` and ``submit_report`` pass through unmodified: sub-agent reports are structured and expensive to regenerate, and the terminal-tool ack is just a stub. Adding a new compute-style tool: append one ``_BUDGETS`` entry and a dispatch in ``process`` if it needs custom reshape (like ``bash``'s stderr-aware split); otherwise everything falls through to ``_head_cap`` with the looked-up budget. The hard 150K ``TOOL_RESULT_MAX_CHARS`` cap inside ``execute_tools`` runs first, so this processor sees at most 150K of input. Output is the string that lands in the ``ToolMessage`` — observers / evidence still see the underlying ``ToolResult`` unchanged. """ # Pass-through for content-retrieval tools: information density is # high (paper results sections sit 30K+ chars in), upstream Summary # LLM (web_fetch's info_to_extract path) handles in-call compression # when the agent asks for it, and ReasoningStripCompactor ages old # bodies to URL stubs once total context crosses 200K. The hard # ceiling is now TOOL_RESULT_MAX_CHARS=150K in execute_tools. # bash / run_python_code stay capped — their output is unbounded by # design (find /, big stdout) and head/tail-with-stderr is the right # diet, not lifting it. _PASS_THROUGH = frozenset({ "delegate_subtask", "submit_report", # collect_reports concatenates the (already-structured, expensive-to- # regenerate) sub-agent reports the main agent synthesizes from — never # head-cap it to the 6K default, or the main agent only sees the first # ~1-2 reports when many sub-agents fan in. "collect_reports", "web_fetch", "web_search", "scholar_search", # Paginates itself: the trailing "PARTIAL READ … read_file again with # offset=N" line IS how the caller continues, so head-capping it replaces # a usable offset with generic "re-fetch" advice. Measured on a live # agent-team run before this was added: of 5 read_file results over 6K, # the tail instruction was lost in every one. React whitelisted this from # the start; the omission here was never deliberate. "read_file", # Same contract, same reason as read_file above. Head-capping it would cut # the continuation pointer off the content the call exists to fetch back, # so recovery would silently return less than it says it did. "recover_result", }) _BUDGETS: ClassVar[dict[str, int]] = { "bash": 4_000, "run_python_code": 5_000, # Gate ① overflows these to disk at meta.max_result_chars=8K and puts the # "saved to , use read_file" pointer at the TAIL. The 6K default cut # below that and took the pointer with it, leaving a truncated result with # no way back to the rest. 10K leaves the 8K body and its footer room to # both survive — the value react has carried for this reason all along. "grep_search": 10_000, "glob_search": 10_000, # Belt-and-suspenders: collect_reports is whitelisted above (pass- # through wins), but if it ever leaves the whitelist this large budget # keeps it effectively uncapped (bounded only by the 150K upstream # TOOL_RESULT_MAX_CHARS ceiling) rather than falling back to 6K. "collect_reports": 150_000, } _BUDGET_DEFAULT = 6_000 _BASH_STDERR_KEEP = 3_000 _ELLIPSIS = ( "\n\n[... body truncated for context budget — re-fetch URL " "or call again with a focused query if you need the rest ...]" ) def process(self, tool_result: ToolResult) -> str: content = ( tool_result.result if isinstance(tool_result.result, str) else str(tool_result.result) ) if tool_result.is_error: return content name = tool_result.name or "" if name in self._PASS_THROUGH: return content budget = self._BUDGETS.get(name, self._BUDGET_DEFAULT) if name == "bash": return self._compact_bash(content, budget) return self._head_cap(content, budget) def _head_cap(self, content: str, budget: int) -> str: # Skip the cut when it wouldn't actually shrink — a budget+30 # input becomes budget+len(ELLIPSIS) bytes otherwise. if len(content) <= budget + len(self._ELLIPSIS): return content return content[:budget] + self._ELLIPSIS def _compact_bash(self, content: str, budget: int) -> str: """Keep stderr in full + stdout head/tail with elision marker. Depends on the ``BASH_STDERR_SEPARATOR`` literal emitted by ``plugins/tools/bash.py`` and ``run_python_code.py``; if those producers stop joining stdout/stderr with this exact string, the split silently degrades to "treat the whole blob as stdout". """ if len(content) <= budget: return content if BASH_STDERR_SEPARATOR in content: stdout, stderr = content.split(BASH_STDERR_SEPARATOR, 1) else: stdout, stderr = content, "" stderr_keep = ( stderr[-self._BASH_STDERR_KEEP:] if len(stderr) > self._BASH_STDERR_KEEP else stderr ) overhead = len(BASH_STDERR_SEPARATOR) + 80 if stderr_keep else 0 stdout_budget = max(500, budget - len(stderr_keep) - overhead) if len(stdout) > stdout_budget: half = stdout_budget // 2 stdout_keep = ( stdout[:half] + f"\n[... {len(stdout) - stdout_budget} stdout chars elided ...]\n" + stdout[-half:] ) else: stdout_keep = stdout if stderr_keep: return stdout_keep + BASH_STDERR_SEPARATOR + stderr_keep return stdout_keep def _swarm_loop_policy() -> LoopPolicy: return LoopPolicy( terminal_tool_names=("submit_report",), no_tool_behavior="nudge", no_tool_nudge_message=_SUBAGENT_NO_TOOL_NUDGE, ) def _sub_tool_names(runtime: SwarmSubagentRuntime | None = None) -> list[str]: if runtime is not None and runtime.sub_agent_tool_names is not None: return list(runtime.sub_agent_tool_names) from frontier_agent.core.runtime.resources.manager import ResourceManager rm = registry.get_optional(ResourceManager) if rm is None: return [] role_id = getattr(runtime, "sub_role_id", None) or SUB_ROLE_ID return list(rm.get_tool_names_for_role(role_id)) def _sub_tools(runtime: SwarmSubagentRuntime | None = None) -> list[Any]: """Resolve the actual tool objects bound to a sub-agent role. Used by ``_swarm_observers`` to embed the OpenAI tool schema in each sub-agent's saved trajectory. Returns ``[]`` when the resource manager isn't registered (e.g. unit tests). """ if runtime is not None and runtime.sub_agent_tools is not None: return list(runtime.sub_agent_tools) from frontier_agent.core.runtime.resources.manager import ResourceManager rm = registry.get_optional(ResourceManager) if rm is None: return [] if runtime is not None and runtime.sub_agent_tool_names is not None: all_tools = rm.all_tools return [all_tools[name] for name in runtime.sub_agent_tool_names if name in all_tools] role_id = getattr(runtime, "sub_role_id", None) or SUB_ROLE_ID return list(rm.get_tools_for_role(role_id)) def _swarm_observers( runtime: SwarmSubagentRuntime, *, task_id: str, session_name: str, task_index: int = 1, task_id_for_sse: str | None = None, run_id: str = "", run_type: str = "", ) -> list[Any]: """Build the observer stack a sub-agent runs with. ``task_id`` is the AgentBus / LoopConfig scope id (may be a heavy-mode synthetic id like ``f"{root}.heavy_run_3"``). ``task_id_for_sse`` defaults to ``task_id``; heavy mode passes the root id so SSE / event_store stay scoped to the user-visible task. """ from frontier_agent.components.observers.budget_observer import BudgetObserver from frontier_agent.components.observers.duplicate_query_rollback import ( DuplicateQueryRollbackObserver, ) from frontier_agent.components.observers.finalization_reserve import ( FinalizationReserveObserver, ) from frontier_agent.components.observers.last_turn_forcer import LastTurnForcer from frontier_agent.components.observers.leaked_tool_call_retry import ( LeakedToolCallRetryObserver, ) from frontier_agent.components.observers.react_step_tracker import ReactStepTracker from frontier_agent.components.observers.repetition_guard import RepetitionGuard from frontier_agent.components.observers.sse_observer import SSEObserver from frontier_agent.components.observers.stop_signal_observer import StopSignalObserver from frontier_agent.components.observers.text_repetition_guard import ( TextRepetitionGuard, ) from workflows._shared.research.observers.assertion_observer import AssertionObserver from workflows._shared.research.observers.evidence_observer import EvidenceObserver from workflows._shared.research.observers.finalize_answer_observer import ( FinalizeAnswerObserver, ) from workflows.agent_team.observers.console import SubAgentLoggingObserver sse_task_id = task_id_for_sse if task_id_for_sse is not None else task_id tool_names = _sub_tool_names(runtime) terminal = "submit_report" if "submit_report" in tool_names else None observers: list[Any] = [ SubAgentLoggingObserver(session_name=session_name), LeakedToolCallRetryObserver(tool_names=tool_names), # Repetition stop-loss (the search-specific rollback is mounted # below). ``RepetitionGuard`` hints at three consecutive turns of # byte-identical tool calls and stops at six; it is the ONLY guard # that can end this loop, because the rollback's budget expires into # permanent let-through and ``TextRepetitionGuard`` needs visible # prose the pathology does not produce (the deliberation goes to # ``thinking`` under ``thinking_format: tag``). ``TextRepetitionGuard`` # covers the prose loop and also stops here. Both stop reasons are in # ``fan_in.INCOMPLETE_STOP_REASONS``, so the partial report still # reaches fan-in through ``force_final_answer``. Stopping is only # affordable for a sub-agent — the coordinator IS the run and gets a # hint at most. RepetitionGuard(stop_after=6), TextRepetitionGuard(enable_stop=True), ReactStepTracker(), EvidenceObserver(), AssertionObserver(), # Cooperative stop: main agent's stop_subagent registers a stop for # this session_id; we inject the stop prompt at the next turn boundary. StopSignalObserver(session_id=f"{task_id}::{session_name}"), ] if DuplicateQueryRollbackObserver.DEFAULT_TOOL_NAMES.intersection(tool_names): # Only meaningful for an agent that can search. The rollback pops the # turn before the duplicate search runs, so it costs an LLM call and # never a tool call. observers.append(DuplicateQueryRollbackObserver()) if runtime.max_tokens: observers.append(BudgetObserver(max_tokens=runtime.max_tokens)) if runtime.stuck_target_hint_after > 0: # Nudge, then quarantine, a repeatedly failing host. Deliberately not # mounted on the coordinator: its tool pool carries no network tools. from frontier_agent.components.observers.stuck_target_guard import ( StuckTargetGuard, ) observers.append(StuckTargetGuard( hint_after=runtime.stuck_target_hint_after, escalate_after=runtime.stuck_target_escalate_after, window=runtime.stuck_target_window, )) if runtime.stream_repetition_config is not None: from workflows.agent_team.stream_repetition import StreamRepetitionDeltaSink observers.append(StreamRepetitionDeltaSink()) if terminal: observers.append(FinalizeAnswerObserver(tool_names=[terminal])) observers.append(FinalizationReserveObserver( reserve_turns=runtime.finalization_reserve_turns, message=_SUBAGENT_FINALIZATION_MESSAGE, )) observers.append(LastTurnForcer(terminal_tool=terminal)) event_store = registry.get_optional(EventStore) if event_store is not None: observers.append(SSEObserver( event_store=event_store, task_id=sse_task_id, run_id=run_id, run_type=run_type, )) if runtime.trajectory_dir is not None: from frontier_agent.components.observers.trajectory import ( TrajectoryFileObserver, ) if runtime.trajectory_per_task: traj_filename = f"{session_name}.t{task_index:02d}" else: traj_filename = session_name observers.append(TrajectoryFileObserver( runtime.trajectory_dir / "subagents", filename=traj_filename, tools=_sub_tools(runtime), format_env_vars=("SWARM_TRAJECTORY_FORMATS",), tool_schema_detail="minimal", include_start_tool_names=False, )) # A second, ``worker_trace_dir``-scoped trace writer — coexists with the # TrajectoryFileObserver above rather than replacing it (see the # ``worker_trace_dir`` field docstring). if runtime.worker_trace_dir is not None and runtime.workflow_id: from workflows._shared.sdk_shim import ( make_subagent_trace_observer, ) wt = make_subagent_trace_observer( workflow_id=runtime.workflow_id, parent_agent_id=MAIN_AGENT_ID, sub_agent_id=session_name, session_name=session_name, trace_dir=runtime.worker_trace_dir, mode="frontier_agent_swarm", ) if wt is not None: observers.append(wt) # Last, so ``notify_tool_call``'s "last rewrite wins" gives the host's # guards the final say over any earlier ``rewrite_args`` — a secret must not # be re-introduced after being scrubbed. "First skip wins" is unaffected: an # earlier skip already means the call never runs. observers.extend(runtime.extra_observers) # The optional SDK protocol stream is not part of the OSS runtime. return observers # Cap on per-job denial logs held for the result adapter. Only jobs whose # adapter never ran (cancelled, timed out) linger, and a session tops out at # MAX_TASKS_PER_SESSION tasks, so this is a leak backstop rather than a limit # anything real reaches. _MAX_TRACKED_DENIAL_JOBS = 64 async def _adapt_swarm_result( agent_result: AgentLoopResult, *, job_id: str, item: SubTask, session_name: str, runtime: SwarmSubagentRuntime, denials: tuple[tuple[str, str], ...] = (), ) -> SubAgentResult: """Normalise a sub-agent run into a :class:`SubAgentResult`. ``force_final_answer`` rescue runs in the bus's ``force_finalizer`` hook (BEFORE absorb), not here — see :func:`build_swarm_session_runtime_spec`. By the time we reach this adapter, the rescue (if any) has already mutated ``agent_result.messages`` and ``agent_result.final_content``, and the bus has absorbed the change into the session. """ from workflows._shared.research.result_adapter import adapt_result result = adapt_result(agent_result) prefix = session_name or job_id.rsplit(".", 1)[-1] assertions: list[dict[str, Any]] = [] for idx, assertion in enumerate(result.get("assertions", []), 1): item_dict = dict(assertion) item_dict["id"] = f"as-{prefix}-{idx:03d}" assertions.append(item_dict) metadata = dict(getattr(agent_result, "metadata", {}) or {}) metadata["evidence_cards"] = list(result.get("evidence_cards", [])) metadata["assertions"] = assertions metadata["stopped_by"] = getattr(agent_result, "stopped_by", "") or "" final_content = result.get("final_content", "") if denials: # The refusal itself only ever reached the sub-agent, which cannot # authorize itself; the coordinator can. Prepending rather than # appending keeps it above a long report, matching how # ``fan_in.format_report_block`` places its own [NOTE: ...] line. from plugins.tools._deliverable_policy import render_denial_escalation blocked_writes = [ list(entry) for entry in denials if entry[0] != "unverifiable" ] unverifiable_accesses = [ list(entry) for entry in denials if entry[0] == "unverifiable" ] if blocked_writes: metadata["blocked_output_writes"] = blocked_writes if unverifiable_accesses: metadata["unverifiable_output_accesses"] = unverifiable_accesses final_content = render_denial_escalation(denials) + final_content return SubAgentResult( question=item.question, role_id=item.role_id, final_content=final_content, success=True, job_id=job_id, metadata=metadata, ) def build_swarm_session_runtime_spec( runtime: SwarmSubagentRuntime, *, session_name: str, task_id: str, task_id_for_sse: str | None = None, run_id: str = "", run_type: str = "", ) -> SubAgentRuntimeSpec: """Build a runtime spec for a single sub-agent session. The main-agent node calls this once per ``create_subagent`` to inject agent-team loop config, observers, and result adapter. ``task_id`` is the AgentBus / LoopConfig scope id (heavy mode passes a synthetic per-run id). ``task_id_for_sse`` defaults to ``task_id``; heavy mode passes the root user-facing id so the SSE stream stays keyed by what the user sees. """ # Per-job real-token gauge, shared between _config (the trigger policy) and # _observers (the observer that updates it) via setdefault, so both see the # SAME instance for a given job. _gauges: dict[str, InputTokenGauge] = {} # Blocked /outputs writes per job. ``_context_setup`` opens the log and # ``_adapter`` drains it: the bus runs the adapter AFTER the context manager # has exited (bus.py, around the ``result_adapter`` call), so the contextvar # is already reset by then and the list has to be handed over explicitly. _denials: dict[str, list[tuple[str, str]]] = {} _tiered = runtime.context_compaction == "tiered" and runtime.max_len > 0 from plugins.tools._overflow import spill_compacted_body def _config(_job_id: str, item: SubTask, max_turns: int) -> LoopConfig: if _tiered: gauge = _gauges.setdefault(_job_id, InputTokenGauge()) compactor: Any = TieredCompactor( keep_tool_result=runtime.tier1_keep_tool_result, summary_llm=runtime.sub_agent_llm, relief_target=int(runtime.max_len * 0.6), gauge=gauge, # calibrate relief to real tokens (unit-match trigger) spill=spill_compacted_body if runtime.compaction_spill else None, summary_retry_timeout_s=( runtime.sub_agent_llm_timeout or LoopConfig.llm_timeout ), ) compaction_policy: Any = InputTokenThresholdPolicy( gauge, compaction_trigger_tokens(runtime.max_len), ) keep_recent = max(6, runtime.keep_recent_turns * 3) else: compactor = KeepLastNToolResultsCompactor( keep_tool_result=runtime.sub_keep_tool_result, ) compaction_policy = None keep_recent = LoopConfig.keep_recent return LoopConfig( max_turns=max_turns, task_id=task_id, llm_session_id=llm_session_id(task_id, session_name), role_id=item.role_id, tool_result_max_chars=runtime.tool_result_max_chars, llm_timeout=int(runtime.sub_agent_llm_timeout or LoopConfig.llm_timeout), tool_timeout=int(runtime.sub_agent_tool_timeout or LoopConfig.tool_timeout), reasoning_only_timeout_s=runtime.reasoning_only_timeout_s, reasoning_only_max_tokens=runtime.reasoning_only_max_tokens, logical_call_timeout_s=runtime.logical_call_timeout_s, max_completion_tokens=( runtime.llm_max_tokens or LoopConfig.max_completion_tokens ), context_token_limit=runtime.sub_context_token_limit, compact_after_turns=runtime.sub_compact_after_turns, keep_recent=keep_recent, loop_policy=_swarm_loop_policy(), compactor=compactor, compaction_policy=compaction_policy, tool_result_post_processor=SwarmToolResultPostProcessor(), ) def _observers(_job_id: str, _item: SubTask, task_index: int) -> list[Any]: obs = _swarm_observers( runtime, task_id=task_id, session_name=session_name, task_index=task_index, task_id_for_sse=task_id_for_sse, run_id=run_id, run_type=run_type, ) if _tiered: obs = [*list(obs), _gauges.setdefault(_job_id, InputTokenGauge())] if runtime.max_input_tokens: obs.append(ContextSizeGuard( max_input_tokens=runtime.max_input_tokens, force_compaction_first=True, )) return obs async def _force_finalizer(raw_result: Any, _item: SubTask) -> Any: # Mirror the old _adapt_swarm_result behaviour: skip when no # LLM is bound (unit tests + lightweight call sites). if runtime.sub_agent_llm is None: return raw_result timeout = ( runtime.sub_agent_llm_timeout if runtime.sub_agent_llm_timeout is not None else LoopConfig.llm_timeout ) return await force_final_answer( raw_result, runtime.sub_agent_llm, timeout, task_description=_item.question, ) async def _adapter( agent_result: Any, job_id: str, item: SubTask, ) -> SubAgentResult: return await _adapt_swarm_result( agent_result, job_id=job_id, item=item, session_name=session_name, runtime=runtime, denials=tuple(_denials.pop(job_id, ())), ) @contextlib.contextmanager def _context_setup(job_id: str, _item: SubTask) -> Iterator[None]: """Scope a per-sub-agent bwrap sandbox for the duration of the loop. Binds ``worktree_root/`` (this sub-agent's private dir) plus the shared dir; sibling sub-agents are not bound, so they are invisible to one another. No-op when no worktree is configured or bwrap is unavailable (falls back to the inherited sandbox). """ from frontier_agent.components.agent_bus.stop_signal import get_stop_registry from plugins.tools._bash_policy import reset_policy_mode, set_policy_mode from plugins.tools._deliverable_policy import ( new_output_write_denial_log, reset_deliverable_write_paths, reset_output_write_denial_log, set_deliverable_write_paths, ) from plugins.tools._sandbox import ( BwrapSandbox, bwrap_available, clear_task_sandbox, make_current_sandbox, set_task_sandbox, ) # Reusable session: drop any cooperative-stop flag left over from a # PRIOR task so it can't roll back turn 1 of this new one. A stop for # THIS job is only queued after it starts running (see stop_subagent), # i.e. strictly after this point, so it is preserved. get_stop_registry().clear_stale(f"{task_id}::{session_name}", job_id) # Default sub-agent bash to the command allowlist too. Set BEFORE the # no-worktree / no-bwrap fallback (which just yields) so that path is # covered — not only the bwrap branch. AgentBus builds the sub-agent's # ExecutionScope without our scope_metadata, so the contextvar is the # reliable carrier here. Reset in the outer finally (covers both paths). policy_token = set_policy_mode("enforce") can_publish = _item.metadata.get("can_publish") is True output_paths = ( _item.metadata.get("output_paths", []) if can_publish else [] ) # Manifest entries superseded by a replace_manifest follow-up: the # publisher may remove them from /outputs so a format change does not # leave the old deliverable stranded there. retired_paths = ( _item.metadata.get("retired_paths", []) if can_publish else [] ) deliverable_token = set_deliverable_write_paths( output_paths, retired_paths, ) # A stale entry would attach the previous task's denial to this one, so # the store is keyed per job and overwritten, not accumulated. The # adapter pops its entry, but it never runs for a cancelled or # timed-out job, so drop the oldest keys rather than growing forever. while len(_denials) >= _MAX_TRACKED_DENIAL_JOBS: _denials.pop(next(iter(_denials))) denial_token, _denials[job_id] = new_output_write_denial_log() try: # Container mode: every sub-agent SHARES the one mounted /workspace # (and /outputs) via a CurrentSandbox. AgentBus builds the sub's # ExecutionScope without the main node's scope_metadata, so we set # the shared sandbox on the contextvar explicitly here. No kill on # exit — the surrounding container owns the mount's lifecycle. if runtime.sandbox_mode in ("container", "native") and runtime.shared_workspace_dir: token = set_task_sandbox( make_current_sandbox(runtime.shared_workspace_dir) ) try: yield finally: clear_task_sandbox(token) return if runtime.worktree_root is None or not bwrap_available(): yield return sandbox = BwrapSandbox( workspace=str(runtime.worktree_root / session_name), binds=tuple( (src, dst, ro or (not output_paths and dst == "/outputs")) for src, dst, ro in runtime.sub_binds ), ) token = set_task_sandbox(sandbox) try: yield finally: clear_task_sandbox(token) # Reclaim the sandbox's mkdtemp tmpdir — without this each # sub-agent leaks one private /tmp dir for the worker's # lifetime (the main_agent node path kills its sandbox too). kill = getattr(sandbox, "kill", None) if callable(kill): try: kill() except Exception: logger.warning("agent_team sub-agent sandbox kill failed", exc_info=True) finally: reset_output_write_denial_log(denial_token) reset_deliverable_write_paths(deliverable_token) reset_policy_mode(policy_token) from frontier_agent.core.runtime.loop.model_profile import HistoryPolicy return SubAgentRuntimeSpec( config_builder=_config, observers_builder=_observers, result_adapter=_adapter, force_finalizer=_force_finalizer, context_setup=_context_setup, model_profile=runtime.sub_agent_model_profile, history_policy=HistoryPolicy( thinking_in_history=runtime.sub_agent_thinking_in_history, thinking_history_max_tokens=( runtime.sub_agent_thinking_history_max_tokens ), ), )