apodex/frontier-agent-demo
14
1"""Tool: create_subagent — register persistent sub-agent sessions."""2 3from __future__ import annotations4 5import importlib6import inspect7import logging8import os9from typing import Any10 11from frontier_agent.components.agent_bus import AgentBus12from frontier_agent.core.execution_context import get_current_execution_scope13from frontier_agent.core.runtime.loop.message_trimmer import TaskBoundaryTrimmer14from frontier_agent.core.runtime.registries import services as registry15from frontier_agent.core.tool import tool16from plugins.tools._bus_scope import resolve_bus_task_id, resolve_root_task_id17from plugins.tools._coerce import coerce_json_list18 19logger = logging.getLogger(__name__)20 21# create_subagent only registers sessions; actual concurrent execution is22# bounded by SpawnGuard(max_parallel). Raise via FRONTIER_AGENT_MAX_SUBAGENTS_PER_DISPATCH.23MAX_SUBAGENTS_PER_DISPATCH: int = int(24 os.environ.get("FRONTIER_AGENT_MAX_SUBAGENTS_PER_DISPATCH", "20")25)26 27SUB_ROLE_ID = "swarm_sub"28SUB_MAX_TURNS_DEFAULT = 10029 30 31def _resolve_sub_role_id(runtime: Any | None) -> str:32 """Sub-agent role id, workflow-overridable.33 34 Defaults to the shared ``swarm_sub`` registration (the name is35 historical — see ``workflows/agent_team/README.md``); a workflow can set36 ``sub_role_id`` on its runtime (agent_team → ``agent_team_sub``) so its37 sub-agents resolve their own fail-closed tool pool instead of sharing it.38 """39 return getattr(runtime, "sub_role_id", None) or SUB_ROLE_ID40 41 42# Sentinel: caller did not specify task_types → ask the workflow for its tuple43# (the behaviour assign_task / scope-rewrite callers rely on). Distinct from an44# explicit ``None`` (lenient role-label path) that create_subagent passes for45# agent_team.46_TASK_TYPES_UNSET: Any = object()47 48 49def _normalize_agent_name(50 name: str, task_types: tuple[str, ...] | None = _TASK_TYPES_UNSET,51) -> str:52 """Auto-fix underscored topic to dashed topic (strict-naming convention).53 54 The convention is ``{topic}_{task_type}[_{N}]`` where the topic segment55 uses dashes for multi-word descriptors. LLMs often emit56 ``gpu_market_research_1`` instead of canonical ``gpu-market_research_1``;57 rather than reject, we rewrite the topic span when the suffix58 unambiguously identifies a ``task_type``. Names whose last segment isn't a59 known task_type are returned unchanged so :func:`_validate_agent_name`60 surfaces the real error.61 62 ``task_types is None`` (workflows like agent_team that use free-form63 role-label names) → return ``name`` unchanged: there is no64 ``{topic}_{task_type}`` structure to normalize.65 """66 if task_types is _TASK_TYPES_UNSET:67 task_types = _resolve_task_types(None) # caller did not say → ask the workflow68 if not task_types:69 return name70 71 parts = name.split("_")72 if len(parts) < 2:73 return name74 75 last = parts[-1]76 if last.isdigit():77 if len(parts) < 3:78 return name79 task_type = parts[-2]80 topic_parts = parts[:-2]81 suffix = f"_{task_type}_{last}"82 else:83 task_type = last84 topic_parts = parts[:-1]85 suffix = f"_{task_type}"86 87 if task_type not in task_types:88 return name89 if len(topic_parts) <= 1:90 return name91 return "-".join(topic_parts) + suffix92 93 94def _validate_agent_name(95 name: str, task_types: tuple[str, ...] | None = _TASK_TYPES_UNSET,96) -> str | None:97 """Return an error message if ``name`` violates the naming convention.98 99 With ``task_types`` (strict mode): valid format is ``{topic}_{task_type}`` or100 ``{topic}_{task_type}_{N}`` where the topic uses dashes and ``task_type``101 is one of ``task_types``.102 103 With ``task_types is None`` (agent_team and any workflow using free-form104 ROLE-label sub-agent names, resolved by substring in its prompts): use a105 LENIENT check — accept any safe identifier. The previous behaviour106 hard-coded a fixed task-type tuple here, so it rejected EVERY agent_team role107 name (``final_verifier`` / ``lit_search`` / ``match_researcher`` / …),108 created zero sub-agents, and deadlocked the planning-mode finalize gate.109 110 Returns ``None`` when valid. Callers should run :func:`_normalize_agent_name`111 first (a no-op in the lenient case).112 """113 if task_types is _TASK_TYPES_UNSET:114 task_types = _resolve_task_types(None) # caller did not say → ask the workflow115 if not task_types:116 # Lenient role-label path: a safe identifier is all that's required;117 # the workflow's get_subagent_system_prompt resolves the specialist by118 # substring, so any reasonable label is valid. Reject only empty names119 # or shell-metacharacter/space injection in the session name.120 if (121 not name122 or not name[0].isalnum()123 or not all(c.isalnum() or c in "_-" for c in name)124 ):125 return (126 f"Agent name {name!r} must be a non-empty identifier "127 f"(letters/digits/_/-, starting alphanumeric; no spaces or "128 f"shell metacharacters)."129 )130 return None131 132 SUBAGENT_TASK_TYPES = task_types133 parts = name.split("_")134 # Need at least topic + task_type → 2 segments minimum.135 if len(parts) < 2:136 return (137 f"Agent name {name!r} must follow {{topic}}_{{task_type}}[_{{N}}]. "138 f"Valid task_types: {', '.join(SUBAGENT_TASK_TYPES)}. "139 f"Use dashes inside the topic: e.g. 'gpu-market_research_1'."140 )141 142 last = parts[-1]143 if last.isdigit():144 # Format: topic_tasktype_N — need at least 3 segments.145 if len(parts) < 3:146 return (147 f"Agent name {name!r}: numeric suffix requires a task_type before it. "148 f"Format: {{topic}}_{{task_type}}_{{N}}. "149 f"Valid task_types: {', '.join(SUBAGENT_TASK_TYPES)}."150 )151 task_type = parts[-2]152 topic_parts = parts[:-2]153 else:154 task_type = last155 topic_parts = parts[:-1]156 157 if task_type not in SUBAGENT_TASK_TYPES:158 return (159 f"Agent name {name!r} has invalid task_type {task_type!r}. "160 f"Must be one of: {', '.join(SUBAGENT_TASK_TYPES)}. "161 f"Format: {{topic}}_{{task_type}}[_{{N}}] — use dashes inside the topic. "162 f"Examples: 'gpu-market_research_1', 'draft-answer_verify', "163 f"'conflicting-claims_lverify'."164 )165 166 # After _normalize_agent_name(), a well-typed topic is a single167 # dash-joined token. If we still see extra ``_`` segments here it168 # means normalization bailed (unknown task_type path) — surface a169 # clear error instead of silently mangling the name.170 if len(topic_parts) != 1:171 bad_topic = "_".join(topic_parts)172 good_topic = "-".join(topic_parts)173 return (174 f"Agent name {name!r}: topic {bad_topic!r} contains underscores. "175 f"Use dashes instead: '{good_topic}_{task_type}'. "176 f"Format: {{topic}}_{{task_type}}[_{{N}}] — topic uses dashes only."177 )178 return None179 180 181def _resolve_runtime(scope_metadata: dict[str, Any]) -> Any | None:182 """Pull the sub-agent runtime config from the active ExecutionScope.183 184 Returns ``None`` outside a multi-agent run so the tool can no-op185 cleanly (still registers the session, just without the workflow's186 observers). The scope key is shared verbatim by every workflow that187 spawns sub-agents, so this resolves whichever one stashed its runtime.188 """189 from plugins.tools._bus_scope import SWARM_SCOPE_KEY190 return scope_metadata.get(SWARM_SCOPE_KEY)191 192 193def _runtime_workflow_pkg(runtime: Any | None) -> str:194 """Resolve the workflow package that owns ``runtime``.195 196 ``create_subagent`` is shared across coordinator-style workflows that197 each ship their own ``subagent_runtime`` / ``prompts`` /198 ``stream_repetition`` modules. The runtime dataclass lives in its199 workflow's ``subagent_runtime`` module, so its ``__module__`` (e.g.200 ``workflows.agent_team.subagent_runtime``) names the package to201 dispatch to. ``None`` / anything not under a ``workflows.<pkg>``202 namespace falls back to ``workflows.agent_team``.203 """204 mod = type(runtime).__module__ if runtime is not None else ""205 if mod.startswith("workflows.") and mod.count(".") >= 2:206 return mod.rsplit(".", 1)[0]207 return "workflows.agent_team"208 209 210def _resolve_task_types(runtime: Any | None) -> tuple[str, ...] | None:211 """Return the active workflow's ``SUBAGENT_TASK_TYPES``, or ``None`` when the212 workflow does not use the strict ``{topic}_{task_type}`` naming convention.213 214 No workflow in this repository declares it, so the strict path is an215 extension seam rather than a live branch here.216 217 A workflow opts into strict ``{topic}_{task_type}[_N]`` names by218 defining ``SUBAGENT_TASK_TYPES`` in its prompts module. ``agent_team``219 does not: it uses free-form ROLE-label names (``final_verifier`` /220 ``lit_search`` / ``match_researcher`` / …) that its221 ``get_subagent_system_prompt`` resolves by substring. Returning222 ``None`` switches name handling to the lenient path below, instead of223 rejecting every role-label name — which created ZERO sub-agents and224 deadlocked the planning-mode finalize gate.225 """226 pkg = _runtime_workflow_pkg(runtime)227 try:228 mod = importlib.import_module(f"{pkg}.prompts")229 tt = getattr(mod, "SUBAGENT_TASK_TYPES", None)230 return tuple(tt) if tt else None231 except Exception:232 return None233 234 235def _build_runtime_spec(236 runtime: Any,237 *,238 session_name: str,239 task_id: str,240 task_id_for_sse: str | None = None,241 run_id: str = "",242 run_type: str = "",243) -> Any:244 mod = importlib.import_module(f"{_runtime_workflow_pkg(runtime)}.subagent_runtime")245 return mod.build_swarm_session_runtime_spec(246 runtime,247 session_name=session_name,248 task_id=task_id,249 task_id_for_sse=task_id_for_sse,250 run_id=run_id,251 run_type=run_type,252 )253 254 255def _resolve_specialist_prompt(256 name: str,257 role_hint: str,258 *,259 fs_mode: bool,260 enhancements: bool = False,261 mcp_tool_names: list[str] | None = None,262 mcp_tool_specs: list[dict[str, Any]] | None = None,263 runtime: Any | None = None,264) -> str:265 """Route a sub-agent to its workflow's specialist system prompt by name."""266 mod = importlib.import_module(f"{_runtime_workflow_pkg(runtime)}.prompts")267 fn = mod.get_subagent_system_prompt268 kwargs: dict[str, Any] = dict(269 name=name,270 role=role_hint,271 include_domain_guide=fs_mode,272 mcp_tool_names=mcp_tool_names or (),273 mcp_tool_specs=mcp_tool_specs or (),274 )275 params = inspect.signature(fn).parameters276 # Optional prompt-builder knobs are capability-detected so the shared tool277 # remains compatible with workflows that do not expose them.278 if "enhancements" in params:279 kwargs["enhancements"] = enhancements280 # Two blocks of runtime facts the static templates cannot know. Both are281 # task-level constants — identical for every sub-agent of one task — so282 # both belong in the shared KV-cache prefix, ahead of the per-agent role.283 #284 # ``sub_prompt_suffix``: for agent_team this is the filesystem-convention285 # note286 # (scratch → /workspace, final deliverable → /outputs) when a sandbox is287 # active — the main agent gets it inline, and sub-agents share the same288 # mounts, so they need the same convention or a sub could drop a final into289 # its /workspace cwd and it would never be collected. In strict mode it is the290 # task's /inputs listing, since read_file cannot list a directory.291 #292 # ``notice``: the research/verifier templates hard-code web_search /293 # web_fetch and a web-centric methodology. If the active tool policy294 # disabled those tools, an explicit override keeps the model from trying295 # tools that will not be in its list.296 runtime_block = (297 str(getattr(runtime, "sub_prompt_suffix", "") or "")298 + _disabled_web_tools_notice(runtime)299 )300 accepts_suffix = "runtime_suffix" in params301 if accepts_suffix:302 kwargs["runtime_suffix"] = runtime_block303 prompt = fn(**kwargs)304 # Capable builders place the block before the role; legacy builders can305 # only have it appended, so adding a new workflow cannot crash. Same block,306 # same internal order either way.307 return prompt if accepts_suffix else prompt + runtime_block308 309 310def _runtime_tool_names(runtime: Any | None) -> set[str] | None:311 names = getattr(runtime, "sub_agent_tool_names", None)312 if names is None:313 return None314 return {str(name) for name in names}315 316 317def _runtime_tools_override(runtime: Any | None) -> list[Any] | None:318 tools = getattr(runtime, "sub_agent_tools", None)319 if tools is not None:320 return list(tools)321 names = getattr(runtime, "sub_agent_tool_names", None)322 if names is None:323 return None324 from frontier_agent.core.runtime.resources.manager import ResourceManager325 326 rm = registry.get_optional(ResourceManager)327 if rm is None:328 return []329 all_tools = rm.all_tools330 return [all_tools[name] for name in names if name in all_tools]331 332 333def _disabled_web_tools_notice(runtime: Any | None = None) -> str:334 """Override block emitted when the tool policy disabled web access for335 sub-agents — keeps the web-centric research prompt coherent without it.336 337 ``check_permission`` runs through the ResourceManager's effective context338 (which layers the global allow/deny policy), so this returns non-empty339 exactly when ``web_search`` / ``web_fetch`` were switched off for this run.340 341 The wording is position-independent because this task-level constant sits342 in the shared prompt prefix, before the per-agent role.343 """344 from frontier_agent.core.runtime.resources.manager import ResourceManager345 346 rm = registry.get_optional(ResourceManager)347 runtime_names = _runtime_tool_names(runtime)348 if rm is None and runtime_names is None:349 return ""350 if runtime_names is not None:351 # A lambda rather than ``.__contains__``: the bound method accepts352 # object, which does not match the (name: str) -> bool signature the353 # else-branch defines.354 def has_tool(name: str) -> bool:355 return name in runtime_names356 else:357 assert rm is not None358 359 def has_tool(name: str) -> bool:360 return rm.check_permission(_resolve_sub_role_id(runtime), name)361 disabled = [362 name for name in ("web_search", "web_fetch")363 if not has_tool(name)364 ]365 if not disabled:366 return ""367 if has_tool("bash"):368 fallback = " Use `bash` for local computation or data work."369 elif has_tool("run_python_code"):370 fallback = " Use `run_python_code` for local computation or data work."371 else:372 fallback = ""373 tools_str = " and ".join(f"`{n}`" for n in disabled)374 verb = "are" if len(disabled) > 1 else "is"375 return (376 "\n\n# Tool Availability Override (READ FIRST)\n"377 f"{tools_str} {verb} DISABLED for this run and will NOT appear in your "378 "tool list. Ignore every other instruction in this prompt that tells "379 "you to search the web or fetch pages, wherever it appears — before or "380 "after this section, your role included. Do not attempt them, and do not "381 "use code to issue web requests as a workaround." + fallback +382 " Answer from your own knowledge and reasoning; when a fact cannot be "383 "verified without the disabled tools, state it as unverified rather "384 "than fabricating a source."385 )386 387 388def _bind_sub_agent_llm(runtime: Any | None) -> Any | None:389 """Pick the sub-agent LLM and bind ``max_tokens`` for full reports."""390 if runtime is not None and runtime.sub_agent_llm is not None:391 llm = runtime.sub_agent_llm392 else:393 from frontier_agent.core.runtime.resources.manager import ResourceManager394 resource_mgr = registry.get_optional(ResourceManager)395 if resource_mgr is None:396 return None397 try:398 llm = resource_mgr.get_llm(_resolve_sub_role_id(runtime))399 except Exception:400 return None401 # Native clients carry no langchain ``Runnable.bind``; bind the402 # per-call ``max_tokens`` knob via the kernel loop's ``_BoundLLM``403 # shim (mirrors ``llm_client._bind_reduced_max_tokens``) so the404 # sub-agent loop threads a big output budget into every request for405 # full-length reports.406 #407 # BUT never request MORE output than the model actually accepts: the408 # profile's declared ``llm.max_tokens`` (threaded as409 # ``runtime.llm_max_tokens``) is the model's real410 # ``max_completion_tokens`` ceiling. Over-requesting (e.g. the legacy411 # hardcoded 65536 against a 32768-cap model) makes the server 400 the412 # sub-agent on turn 1 (``stopped_by=llm_error``, empty report). Cap the413 # desired budget by the profile ceiling when known; fall back to the414 # legacy desired value when unknown (no profile).415 desired_max_tokens = 65536416 profile_cap = getattr(runtime, "llm_max_tokens", None)417 if isinstance(profile_cap, int) and profile_cap > 0:418 eff_max_tokens = min(desired_max_tokens, profile_cap)419 if eff_max_tokens < desired_max_tokens:420 logger.info(421 "sub-agent LLM: capping max_tokens %d→%d per profile "422 "llm.max_tokens (model output ceiling)",423 desired_max_tokens, eff_max_tokens,424 )425 else:426 # No profile ceiling known: fall back to the legacy desired value.427 # This is the path that historically 400'd sub-agents when the model's428 # real output cap was below 65536 — warn so it's greppable in logs.429 eff_max_tokens = desired_max_tokens430 logger.warning(431 "sub-agent LLM: no profile llm.max_tokens known; requesting "432 "max_tokens=%d unbounded — if the model rejects it the sub-agent "433 "dies on turn 1 with stopped_by=llm_error and an empty report",434 eff_max_tokens,435 )436 try:437 from dataclasses import replace438 439 from frontier_agent.core.runtime.loop.llm_client import _ensure_bound440 bound = replace(_ensure_bound(llm), max_tokens=eff_max_tokens)441 except Exception:442 bound = llm443 stream_cfg = getattr(runtime, "stream_repetition_config", None)444 if stream_cfg is None:445 return bound446 _sr = importlib.import_module(447 f"{_runtime_workflow_pkg(runtime)}.stream_repetition"448 )449 wrap_llm_for_stream_repetition = _sr.wrap_llm_for_stream_repetition450 451 wrapped, _observer = wrap_llm_for_stream_repetition(452 bound,453 config=stream_cfg,454 role_id=SUB_ROLE_ID,455 label="subagent",456 )457 return wrapped458 459 460@tool461async def create_subagent(agents: list[Any] | str = "") -> str:462 """Create one or more persistent sub-agents.463 464 Each sub-agent is a long-lived session that can accept multiple465 ``assign_task`` calls, accumulating history across tasks. The466 session's history is trimmed between tasks (system + each completed467 task's prompt + final report) so reused agents stay context-efficient.468 469 Args:470 agents: list of dicts with:471 - ``name`` (required): Unique sub-agent name following472 ``{topic}_{task_type}[_{N}]``, e.g. ``lit-review_research``.473 - ``system_prompt`` (required): Custom system prompt for this474 instance — describes its specialty/focus.475 476 Returns:477 Confirmation text listing the created sub-agents.478 """479 # ``agents`` defaults to ``""`` so models emitting empty ``{}`` args480 # land on the actionable error path instead of crashing pydantic481 # with ``Field required``. JSON-string serialisation is handled by482 # ``coerce_json_list``.483 agents = coerce_json_list(agents) or []484 if not agents:485 return "Error: create_subagent requires at least one agent spec."486 487 if len(agents) > MAX_SUBAGENTS_PER_DISPATCH:488 return (489 f"Error: create_subagent supports at most "490 f"{MAX_SUBAGENTS_PER_DISPATCH} agents per call; you passed "491 f"{len(agents)}. Split the list across multiple calls so no "492 f"agent is silently dropped."493 )494 495 scope = get_current_execution_scope()496 if scope is None:497 return (498 "Error: create_subagent can only be called inside an "499 "active ReAct execution."500 )501 502 runtime = _resolve_runtime(scope.metadata)503 sub_role_id = _resolve_sub_role_id(runtime)504 fs_mode = bool(getattr(runtime, "fs_mode", False))505 # Online prompt enhancements gate — same ``sdk_protocol_emitter`` signal506 # the main agent reads (set by serve.py / run.py, absent on benchmark507 # eval). On → online-tuned sub-agent templates; off → ``*_lean`` baseline.508 online_prompt = bool((scope.metadata or {}).get("sdk_protocol_emitter"))509 bus = registry.get(AgentBus)510 bus_task_id = resolve_bus_task_id(scope)511 sse_task_id = resolve_root_task_id(scope)512 # Heavy-mode tags carried in scope_metadata by main_agent_node513 # (S0 plumbing). Empty in a normal run — passed through to514 # SSEObserver so per-run sub-agent events are tagged with the515 # owning run_id.516 run_id = str((scope.metadata or {}).get("run_id") or "")517 run_type = str((scope.metadata or {}).get("run_type") or "")518 bound_llm = _bind_sub_agent_llm(runtime)519 520 created: list[str] = []521 errors: list[str] = []522 renamed: list[tuple[str, str]] = []523 524 # Workflow-aware name validation: strict mode enforces {topic}_{task_type};525 # agent_team (no SUBAGENT_TASK_TYPES) uses lenient role-label names.526 task_types = _resolve_task_types(runtime)527 528 for spec in agents:529 if not isinstance(spec, dict):530 errors.append(f"Skipping non-dict agent spec: {spec!r}")531 continue532 raw_name = str(spec.get("name", "")).strip()533 prompt = str(spec.get("system_prompt", "")).strip()534 if not raw_name:535 errors.append("Skipping agent with no name")536 continue537 name = _normalize_agent_name(raw_name, task_types)538 if name != raw_name:539 renamed.append((raw_name, name))540 name_error = _validate_agent_name(name, task_types)541 if name_error:542 errors.append(name_error)543 continue544 if not prompt:545 errors.append(f"Skipping agent {name!r} with no system_prompt")546 continue547 548 try:549 effective_prompt = _resolve_specialist_prompt(550 name,551 prompt,552 fs_mode=fs_mode,553 enhancements=online_prompt,554 mcp_tool_names=getattr(runtime, "mcp_tool_names", None),555 mcp_tool_specs=getattr(runtime, "mcp_tool_specs", None),556 runtime=runtime,557 )558 runtime_spec = (559 _build_runtime_spec(560 runtime,561 session_name=name,562 task_id=bus_task_id,563 task_id_for_sse=sse_task_id,564 run_id=run_id,565 run_type=run_type,566 )567 if runtime is not None568 else None569 )570 571 await bus.create_session(572 task_id=bus_task_id,573 name=name,574 role_id=sub_role_id,575 system_prompt=effective_prompt,576 tools_override=_runtime_tools_override(runtime),577 trimmer=TaskBoundaryTrimmer(),578 max_turns=(579 int(runtime.sub_agent_max_turns)580 if runtime is not None581 and getattr(runtime, "sub_agent_max_turns", None)582 else SUB_MAX_TURNS_DEFAULT583 ),584 llm_override=bound_llm,585 tool_result_max_chars=getattr(runtime, "tool_result_max_chars", 15_000),586 runtime_spec=runtime_spec,587 )588 created.append(name)589 except Exception as exc:590 logger.warning(591 "create_subagent: failed to create %s: %s", name, exc,592 )593 errors.append(f"Failed to create {name!r}: {exc}")594 595 if not created and errors:596 return "Error: " + "; ".join(errors)597 598 lines = [f"Created {len(created)} sub-agent(s):"]599 lines.extend(f" - {n}" for n in created)600 if renamed:601 lines.append("")602 lines.append(603 "Note: normalized underscores in topic → dashes "604 "(canonical form is {topic}_{task_type}[_{N}] with dash-only topic):"605 )606 lines.extend(f" - {raw} → {fixed}" for raw, fixed in renamed)607 if errors:608 lines.append("")609 lines.append("Warnings:")610 lines.extend(f" - {e}" for e in errors)611 lines.append("")612 lines.append(613 "Call assign_task(tasks=[{agent:NAME, prompt:...}]) to give them work."614 )615 return "\n".join(lines)616 617 618__all__ = ["create_subagent"]619