apodex/frontier-agent-demo
14
1"""Tool: assign_task — non-blocking task dispatch to persistent sub-agents.2 3Wraps AgentBus.submit_task_to_session. Each call queues a task on a4previously-created session; the task runs in the background. Use5``collect_reports`` to fetch results (FIRST_COMPLETED semantics).6"""7 8from __future__ import annotations9 10import logging11import os12import re13from typing import Any14 15from pydantic import (16 BaseModel,17 ConfigDict,18 Field,19 StrictBool,20 StrictStr,21 ValidationError,22 field_validator,23 model_validator,24)25 26from frontier_agent.components.agent_bus import AgentBus27from frontier_agent.core.execution_context import (28 get_current_execution_scope,29 get_current_tool_call_id,30)31from frontier_agent.core.runtime.registries import services as registry32from frontier_agent.core.tool import tool33from plugins.tools._bus_scope import resolve_bus_task_id34from plugins.tools._coerce import coerce_json_list35from plugins.tools._deliverable_policy import (36 normalise_output_paths,37 output_write_directives,38 render_publish_assignment,39 render_retirement_note,40 render_workspace_assignment,41)42from plugins.tools.create_subagent import (43 _normalize_agent_name,44 _resolve_runtime,45 _resolve_task_types,46)47 48logger = logging.getLogger(__name__)49 50# Actual concurrency is bounded by SpawnGuard(max_parallel).51# Raise via FRONTIER_AGENT_MAX_TASKS_PER_DISPATCH.52MAX_TASKS_PER_DISPATCH: int = int(53 os.environ.get("FRONTIER_AGENT_MAX_TASKS_PER_DISPATCH", "20")54)55 56# Hard cap on tasks per session. Past ~5 reuses the message history is57# dominated by stale tool stubs and earlier conclusions, and the agent58# starts anchoring on a prior (often wrong) answer instead of doing fresh59# work — heavily-reused sessions score far below fresh ones. The limit60# forces the main agent to spawn a new specialist instead of poisoning a61# saturated session.62MAX_TASKS_PER_SESSION = 563 64 65def _session_at_task_cap(session: Any) -> bool:66 """Whether this session can no longer accept an assignment.67 68 Counts dispatched tasks AND queued-but-not-yet-dispatched ones:69 ``total_task_count`` only increments at dispatch time70 (``bus.py:_dispatch_session_task``), so tasks parked in ``pending_tasks``71 would otherwise slip past the cap. A missing session counts as capped —72 nothing can be assigned to a name the bus does not know.73 """74 if session is None:75 return True76 return (77 getattr(session, "total_task_count", 0)78 + len(getattr(session, "pending_tasks", ()) or ())79 ) >= MAX_TASKS_PER_SESSION80 81 82def _session_has_publish_work(bus: AgentBus, session: Any) -> bool:83 """Whether a session still has a running or queued publishing task.84 85 Publication authorization is copied into each dispatched task. Changing86 ``publication_state`` therefore cannot revoke an incumbent task that is87 already running or queued; transferring the role while such work exists88 would leave two agents authorized to write the same manifest.89 90 Metadata lookup failures are treated conservatively. A delayed transfer is91 recoverable on the next coordinator turn, while an unsafe transfer can92 corrupt the final deliverable.93 """94 if session is None:95 return False96 97 if getattr(session, "current_job_id", None) is not None:98 if not hasattr(bus, "current_job_metadata"):99 return True100 try:101 metadata = dict(bus.current_job_metadata(session.session_id) or {})102 if metadata.get("can_publish") is True:103 return True104 if "can_publish" not in metadata:105 return True106 except Exception:107 logger.warning(108 "assign_task: could not inspect current job metadata for %s",109 getattr(session, "session_id", "<unknown>"),110 exc_info=True,111 )112 return True113 114 for pending in getattr(session, "pending_tasks", ()) or ():115 metadata = getattr(pending, "task_metadata", None) or {}116 if metadata.get("can_publish") is True:117 return True118 if "can_publish" not in metadata:119 return True120 return False121 122 123# Cross-agent report attachment:124# <attach agent="q1_lit"/>125# The main agent puts these tags inside a task prompt to feed another126# session's last report into this new task. We expand them here before127# dispatch so the downstream sub-agent sees the literal report text.128_ATTACH_RE = re.compile(r'<attach\s+agent="([^"]+)"\s*/>')129 130 131class AssignmentSpec(BaseModel):132 """The permissive assignment shape exposed by the shared registry tool."""133 134 # The registry-level tool has always ignored unrelated item keys, and135 # workflows other than agent_team may still bind it. Keep that runtime136 # leniency here without advertising agent-team-only publication controls137 # on the shared model-facing schema — agent_team installs its own138 # stricter Tool object in its loop instead.139 model_config = ConfigDict(extra="ignore")140 141 agent: StrictStr = Field(142 description=(143 "Name of a sub-agent created via create_subagent during this "144 "execution. Agent names mentioned in prior task history do not "145 "exist automatically."146 ),147 min_length=1,148 )149 prompt: StrictStr = Field(150 description="Concrete task prompt for that sub-agent.",151 min_length=1,152 )153 154 155# ``publish`` used to be the authorization field and ``output_paths`` its156# attachment. They were never independent: ``validate_publish_contract``157# rejected true-without-paths and paths-without-true alike, so the boolean158# carried nothing the manifest did not already carry. Requiring it only added a159# key the coordinator omitted on roughly a quarter of its assignments (60/236160# and 63/225 across the two arms of the 30-task APEX replay), where the default161# silently absorbed it as false -- and an omission next to a correct manifest162# was a hard rejection rather than a publisher. The manifest is now the grant;163# ``publish`` survives only so callers that still send it are not rejected.164#165# NOTE: this class's docstring is rendered verbatim into the model-facing tool166# schema, so keep it to what the coordinator needs to read.167class AgentTeamAssignmentSpec(AssignmentSpec):168 """One assignment. Carrying ``output_paths`` is what authorizes it to write169 those ``/outputs`` paths; without them the task runs workspace-only."""170 171 model_config = ConfigDict(extra="forbid")172 173 publish: StrictBool | None = Field(174 default=None,175 # Marked deprecated in the advertised schema only. Pydantic's own176 # ``deprecated=True`` warns on every attribute read, and177 # ``validate_publish_contract`` below reads it twice per assignment --178 # that is a DeprecationWarning per validated task in every run.179 json_schema_extra={"deprecated": True},180 description=(181 "Deprecated, omit it. output_paths alone grants publication. If "182 "sent it must agree with the manifest."183 ),184 )185 output_paths: list[StrictStr] = Field(186 default_factory=list,187 description=(188 "Exact absolute final file paths under /outputs. Supplying them IS "189 "the publication grant, so set them on exactly one final-publisher "190 "assignment and omit them on every other. Pass a JSON array even "191 "for one path."192 ),193 )194 replace_manifest: StrictBool = Field(195 default=False,196 description=(197 "Set true only on a follow-up to the existing publisher when the "198 "required final output formats genuinely changed."199 ),200 )201 202 @property203 def can_publish(self) -> bool:204 """Whether this assignment may write ``/outputs``.205 206 Derived from the manifest alone: a path list is the grant.207 """208 return bool(self.output_paths)209 210 @field_validator("publish", mode="before")211 @classmethod212 def normalise_publish_boolean(cls, value: Any) -> Any:213 """Accept common JSON boolean strings in the structured field."""214 if isinstance(value, str):215 normalised = value.strip().lower()216 if normalised in {"false", "true"}:217 return normalised == "true"218 return value219 220 @field_validator("output_paths", mode="before")221 @classmethod222 def normalise_null_output_paths(cls, value: Any) -> Any:223 """Treat an explicitly unused optional manifest like an omitted one."""224 return [] if value is None else value225 226 @field_validator("replace_manifest", mode="before")227 @classmethod228 def normalise_null_replace_manifest(cls, value: Any) -> Any:229 """Treat an explicitly unused optional replacement flag as false."""230 return False if value is None else value231 232 @model_validator(mode="after")233 def validate_publish_contract(self) -> AgentTeamAssignmentSpec:234 """Reject a compatibility flag that contradicts the manifest.235 236 A contradiction is raised rather than resolved in either direction:237 honouring the flag would drop a manifest the coordinator asked for238 (the shape that loses the deliverable), and honouring the manifest239 would widen authority on the strength of a call that says not to.240 """241 if not self.output_paths:242 if self.publish is True:243 raise ValueError(244 "publish=true requires at least one exact absolute "245 "output_paths entry"246 )247 if self.replace_manifest:248 raise ValueError("replace_manifest requires output_paths")249 return self250 251 if self.publish is False:252 raise ValueError(253 "publish=false contradicts output_paths; output_paths is the "254 "publication grant, so omit publish to authorize this manifest "255 "or drop output_paths for workspace-only work"256 )257 self.output_paths = list(normalise_output_paths(self.output_paths))258 return self259 260 261def _assignment_validation_error(index: int, exc: ValidationError) -> str:262 issues: list[str] = []263 for error in exc.errors(include_url=False, include_input=False):264 error_location = error.get("loc", ())265 location = ".".join(str(part) for part in error_location) or "task"266 issues.append(f"{location}: {error.get('msg', 'invalid value')}")267 detail = "; ".join(issues) or "invalid assignment object"268 return f"task {index}: {detail}"269 270 271def _resolve_original_question(scope_metadata: dict[str, Any]) -> str:272 from plugins.tools._bus_scope import SWARM_SCOPE_KEY273 runtime = scope_metadata.get(SWARM_SCOPE_KEY)274 return getattr(runtime, "original_question", "").strip()275 276 277def _unknown_agent_validation_errors(278 raw_tasks: list[Any],279 *,280 bus: AgentBus,281 bus_task_id: str,282 task_types: tuple[str, ...],283) -> list[str]:284 """Surface lifecycle errors alongside structured metadata errors.285 286 Pydantic validation used to return before session lookup. A call that had287 both a bad manifest and a stale agent name therefore needed two retries:288 fixing the path merely uncovered ``Unknown agent`` on the next turn.289 """290 errors: list[str] = []291 for index, raw_spec in enumerate(raw_tasks, start=1):292 if not isinstance(raw_spec, dict):293 continue294 raw_name = raw_spec.get("agent")295 if not isinstance(raw_name, str) or not raw_name.strip():296 continue297 agent_name = _normalize_agent_name(raw_name.strip(), task_types)298 if agent_name and bus.get_session(f"{bus_task_id}::{agent_name}") is None:299 errors.append(300 f"task {index}: agent: Unknown agent {agent_name!r}; call "301 "create_subagent first. Sub-agents are scoped to the current "302 "execution and names from prior task history are not active"303 )304 return errors305 306 307def _expand_attach_tags(task_prompt: str, task_id: str, bus: AgentBus) -> str:308 """Replace ``<attach agent="NAME"/>`` with the named session's last report.309 310 If the referenced agent doesn't exist or has no report yet, leave a311 visible placeholder rather than silently dropping the tag — that312 gives the sub-agent a chance to say "referenced report missing" in313 its output instead of blindly proceeding on an incomplete task.314 """315 if "<attach" not in task_prompt:316 return task_prompt317 318 def _sub(match: re.Match[str]) -> str:319 name = match.group(1).strip()320 session = bus.get_session(f"{task_id}::{name}")321 if session is None:322 return (323 f"[attach agent={name!r}: agent not found — "324 f"main must create it before attaching]"325 )326 report = (session.last_report or "").strip()327 if not report:328 return (329 f"[attach agent={name!r}: no report yet — "330 f"the agent has not completed a task]"331 )332 return (333 f"\n\n--- BEGIN REPORT FROM {name} ---\n"334 f"{report}\n"335 f"--- END REPORT FROM {name} ---\n\n"336 )337 338 return _ATTACH_RE.sub(_sub, task_prompt)339 340 341@tool342async def assign_task(tasks: list[AssignmentSpec] | str = "") -> str:343 """Assign tasks to previously-created sub-agents. Non-blocking.344 345 Each task is submitted to an existing session; the session runs its346 tasks strictly serially (one at a time). Submitting a second task347 while another is in flight queues it FIFO behind the running one —348 it starts automatically as soon as the predecessor finalises, and349 its report flows through ``collect_reports`` like any other.350 351 Args:352 tasks: list of dicts with:353 - ``agent`` (required): Name of a sub-agent previously354 created via ``create_subagent``.355 - ``prompt`` (required): The task prompt for this sub-agent.356 - ``output_paths`` (agent-team only): Exact absolute final file357 paths under ``/outputs``. Supplying them authorizes this358 assignment to write exactly those paths, so set them on the359 single final publishing assignment only.360 - ``publish`` (optional, agent-team only): Compatibility only.361 Authority comes from ``output_paths``; if sent it must agree362 with the manifest.363 - ``replace_manifest`` (optional): Set true on a follow-up to the364 existing publisher when the required final formats have changed.365 Dropped entries become removable so the publisher can clear the366 superseded files out of ``/outputs``.367 368 Returns:369 Summary of successful submissions and any errors.370 """371 # ``tasks`` defaults to ``""`` so models that emit an empty ``{}``372 # args object (qwen35-397B occasionally does, before it has decided373 # what to assign) hit our actionable error path instead of a raw374 # pydantic ``Field required`` ValidationError that costs a retry.375 # Some models also serialise the list as a JSON-encoded string —376 # ``coerce_json_list`` handles that.377 raw_tasks = coerce_json_list(tasks) or []378 if not raw_tasks:379 return "Error: assign_task requires at least one task."380 if not isinstance(raw_tasks, list):381 return (382 "Error: assign_task.tasks must be a JSON array of assignment "383 "objects (or a JSON-encoded array)."384 )385 386 if len(raw_tasks) > MAX_TASKS_PER_DISPATCH:387 return (388 f"Error: assign_task supports at most "389 f"{MAX_TASKS_PER_DISPATCH} tasks per call; you passed "390 f"{len(raw_tasks)}. Split the list across multiple calls so no "391 f"assignment is silently dropped."392 )393 394 scope = get_current_execution_scope()395 if scope is None:396 return (397 "Error: assign_task can only be called inside an "398 "active ReAct execution."399 )400 401 bus = registry.get(AgentBus)402 bus_task_id = resolve_bus_task_id(scope)403 original_question = _resolve_original_question(scope.metadata)404 # Resolve the active workflow's naming convention the SAME way405 # create_subagent does, so create and assign normalize names identically.406 # Without this, assign_task defaulted to swarm's ``{topic}_{task_type}``407 # convention and silently rewrote free-form agent_team role names408 # (e.g. ``taiwan_visa_verify`` → ``taiwan-visa_verify``,409 # ``taiwan__visa__verify`` → ``taiwan--visa-_verify``). create_subagent410 # (lenient for agent_team) stored the literal name, so the rewritten411 # lookup missed the just-created session or hit a saturated near-duplicate412 # — the main agent could never reliably reach its own sub-agents and413 # spiralled into an unbounded create/assign loop. A workflow that does414 # declare task types resolves to its own tuple → behaviour unchanged415 # for it.416 runtime = _resolve_runtime(scope.metadata)417 task_types = _resolve_task_types(runtime)418 is_agent_team = (419 runtime is not None420 and hasattr(runtime, "publication_state")421 and hasattr(runtime, "publication_lock")422 )423 spec_type = AgentTeamAssignmentSpec if is_agent_team else AssignmentSpec424 specs: list[AssignmentSpec] = []425 validation_errors: list[str] = []426 for index, raw_spec in enumerate(raw_tasks, start=1):427 try:428 # Backward-compatible runtime coercion for historical callers that429 # supplied one path string. The model-facing schema remains the430 # stronger ``array[string]`` shape so new tool calls learn the431 # canonical structure.432 if isinstance(raw_spec, dict) and "output_paths" in raw_spec:433 raw_output_paths = coerce_json_list(raw_spec["output_paths"])434 if isinstance(raw_output_paths, str):435 raw_output_paths = [raw_output_paths]436 raw_spec = {**raw_spec, "output_paths": raw_output_paths}437 validation_input = (438 raw_spec.model_dump()439 if isinstance(raw_spec, BaseModel)440 and not isinstance(raw_spec, spec_type)441 else raw_spec442 )443 spec = (444 raw_spec445 if isinstance(raw_spec, spec_type)446 else spec_type.model_validate(validation_input)447 )448 except ValidationError as exc:449 validation_errors.append(_assignment_validation_error(index, exc))450 continue451 specs.append(spec)452 if is_agent_team and validation_errors:453 validation_errors.extend(_unknown_agent_validation_errors(454 raw_tasks,455 bus=bus,456 bus_task_id=bus_task_id,457 task_types=task_types or (),458 ))459 return (460 "Error: invalid agent-team assignment metadata: "461 + "; ".join(validation_errors)462 + ". No tasks were dispatched."463 )464 if not specs:465 return "Error: invalid assignment metadata: " + "; ".join(validation_errors)466 467 submitted: list[dict[str, str]] = []468 errors: list[str] = []469 notices: list[str] = []470 # Deliverable paths the ORIGINAL QUESTION names. It is prepended verbatim to471 # every dispatched prompt below, so every workspace-only sub-agent inherits472 # the user's own instruction to write them -- see the note on473 # ``output_write_directives``. Computed once: the text is the same for all474 # assignments in the run.475 question_directives = output_write_directives(original_question)476 # Keyed on the manifest, not on ``publish``: the manifest is the grant.477 # A plain ``AssignmentSpec`` has no such field, hence the getattr default.478 publish_specs = [spec for spec in specs if getattr(spec, "output_paths", ())]479 # Whether the coordinator DECLARED a publisher -- one in this dispatch, or480 # one the run already recorded. This is safe only as prompt context: a spec481 # can still fail to dispatch and its manifest may not cover the inherited482 # paths. Actual authorization is derived from publication_state after all483 # submissions finish below.484 recorded_publisher = ""485 if is_agent_team and runtime is not None:486 recorded_publisher = str(487 (getattr(runtime, "publication_state", None) or {}).get(488 "publisher_agent_name"489 )490 or ""491 )492 publisher_declared = bool(publish_specs) or bool(recorded_publisher)493 if is_agent_team and len(publish_specs) > 1:494 return (495 "Error: only one publishing assignment is allowed per dispatch. "496 "Choose one final integrator and one exact output manifest."497 )498 499 for spec in specs:500 agent_name = _normalize_agent_name(501 spec.agent.strip(), task_types,502 )503 task_prompt = spec.prompt.strip()504 if not agent_name:505 errors.append("Skipping task with no 'agent' field")506 continue507 if not task_prompt:508 errors.append(f"Skipping empty task for {agent_name!r}")509 continue510 session_id = f"{bus_task_id}::{agent_name}"511 session = bus.get_session(session_id)512 if session is None:513 errors.append(514 f"Unknown agent {agent_name!r} — call create_subagent first. "515 "Sub-agents are scoped to the current execution; a name from "516 "prior task history is not active automatically"517 )518 continue519 520 task_metadata: dict[str, Any] = {}521 publication_claim: tuple[str, ...] = ()522 publication_state: dict[str, Any] | None = None523 previous_publisher = ""524 previous_manifest: tuple[str, ...] = ()525 replace_manifest = False526 if (527 is_agent_team528 and runtime is not None529 and isinstance(spec, AgentTeamAssignmentSpec)530 ):531 if spec.can_publish:532 if "verifier" in agent_name.lower():533 errors.append(534 f"{agent_name}: verifier tasks cannot publish files; "535 "return verification as report text"536 )537 continue538 output_paths = tuple(spec.output_paths)539 publication_state = runtime.publication_state540 replace_manifest = spec.replace_manifest541 publication_claim = output_paths542 task_metadata = {543 "can_publish": True,544 "output_paths": list(output_paths),545 }546 task_prompt += render_publish_assignment(output_paths)547 else:548 task_metadata = {"can_publish": False, "output_paths": []}549 # The coordinator's own wording can direct the same write the550 # question does. Both are neutralised the same way; only this551 # one is worth reporting back, because only this one is a552 # contradiction the coordinator authored and can fix.553 own_directives = output_write_directives(task_prompt)554 task_prompt += render_workspace_assignment(555 inherited_paths=(*own_directives, *question_directives),556 publisher_declared=publisher_declared,557 )558 if own_directives:559 named = ", ".join(own_directives)560 notices.append(561 f"{agent_name}: dispatched workspace-only, but its "562 f"prompt tells it to write {named}. That write is "563 f"blocked for a non-publisher and the agent may report "564 f"success after routing it elsewhere. If this agent is "565 f"meant to produce the deliverable, re-assign it with "566 f"output_paths={list(own_directives)!r}; "567 f"otherwise expect its result under /workspace"568 )569 570 if _session_at_task_cap(session):571 errors.append(572 f"{agent_name}: session has reached the {MAX_TASKS_PER_SESSION}"573 f"-task limit (dispatched={session.total_task_count}, "574 f"queued={len(session.pending_tasks)}). Long sticky sessions "575 f"anchor on prior conclusions and degrade accuracy. If you "576 f"still need NEW information, create a fresh sub-agent (e.g. "577 f"{agent_name}_v2 or a role-renamed variant) and assign the "578 f"task to it. But if this session already reported what you "579 f"need — or you were only trying to wrap up / gave it a "580 f"trivial task — do NOT spawn more agents: deliver your final "581 f"answer directly as plain text now."582 )583 continue584 585 # Expand any cross-agent <attach agent="..."/> tags before586 # dispatch so the sub-agent sees the actual report text.587 expanded_prompt = _expand_attach_tags(588 task_prompt, bus_task_id, bus,589 )590 if original_question and original_question[:100] not in expanded_prompt:591 expanded_prompt = (592 f"# Original Question\n{original_question}\n\n"593 f"# Your Task\n{expanded_prompt}"594 )595 596 # Build the spawn_context dict so the sub-agent's trace file is597 # stamped with the delegation lineage. ``parent_run_id`` is the598 # parent loop's run_id (set by599 # per-run state on fan-out paths; empty for600 # single-loop SDK callers). ``delegation_prompt`` is the verbatim601 # text the sub-agent sees, post attach-expansion.602 try:603 _allowed_tools = [604 getattr(t, "name", str(t))605 for t in (getattr(session, "tools", None) or [])606 ]607 except Exception:608 _allowed_tools = []609 scope_md = scope.metadata or {}610 spawn_context = {611 "parent_run_id": str(scope_md.get("run_id") or ""),612 "parent_agent_id": str(613 scope_md.get("agent_id") or scope_md.get("role_id") or ""614 ),615 "parent_turn": int(scope_md.get("current_turn") or 0),616 "spawned_by_llm_call_id": str(617 scope_md.get("last_llm_call_id") or ""618 ),619 "spawned_by_tool_call_id": str(get_current_tool_call_id() or ""),620 "delegation_prompt": expanded_prompt,621 "allowed_tools": _allowed_tools,622 "depth": 1,623 "budget": {624 "max_turns": int(getattr(session, "max_turns", 0) or 0),625 },626 }627 try:628 if (629 publication_claim630 and publication_state is not None631 and runtime is not None632 ):633 # Serialize the check, claim, and queue submission. The bus634 # await only enqueues work; it does not wait for the sub-agent.635 # Recording first closes the check-then-set race, while the636 # rollback preserves an earlier manifest if enqueueing fails.637 async with runtime.publication_lock:638 previous_publisher = str(639 publication_state.get("publisher_agent_name") or ""640 )641 previous_manifest = tuple(642 publication_state.get("deliverable_manifest") or ()643 )644 previous_retired = tuple(645 publication_state.get("retired_paths") or ()646 )647 if previous_publisher and previous_publisher != agent_name:648 # The lock exists so two sub-agents cannot race on the649 # same deliverable — NOT to make the role permanent. An650 # incumbent that can no longer be dispatched used to651 # deadlock the run outright: the task cap above told the652 # coordinator to "create a fresh sub-agent" while this653 # branch told it to "reuse" the capped one, so nothing654 # could ever write /outputs again. Seen for real — a655 # trial spent its last 6 turns alternating between the656 # two errors and shipped no deliverable at all, despite657 # having a finished answer in the workspace. Same trap658 # ``finalize_answer._finalize_gate`` already documents659 # for unassigned agents.660 incumbent = bus.get_session(661 f"{bus_task_id}::{previous_publisher}"662 )663 if not _session_at_task_cap(incumbent):664 errors.append(665 f"{agent_name}: publisher already assigned to "666 f"{previous_publisher!r}, which can still take "667 f"work — one publisher per run, so reuse that "668 f"agent for the deliverable"669 )670 continue671 if _session_has_publish_work(bus, incumbent):672 errors.append(673 f"{agent_name}: publisher {previous_publisher!r} "674 "still has a publishing task running or queued; "675 "wait for it with collect_reports before "676 "transferring the publisher role"677 )678 continue679 logger.warning(680 "assign_task: transferring publisher role %r -> %r "681 "(incumbent can no longer be dispatched)",682 previous_publisher, agent_name,683 )684 if (685 previous_manifest686 and previous_manifest != publication_claim687 and not replace_manifest688 ):689 errors.append(690 f"{agent_name}: output manifest is already fixed as "691 f"{list(previous_manifest)!r}; set "692 "replace_manifest=true only if the required final "693 "formats genuinely changed"694 )695 continue696 # Entries dropped by a replacement would otherwise be697 # stranded: every write path to them stays blocked, so the698 # run would end with the old AND the new format present.699 # Carrying them as ``retired_paths`` lets the publisher —700 # and only the publisher — delete or move them out.701 retired_paths = tuple(702 path703 for path in dict.fromkeys(704 (*previous_retired, *previous_manifest)705 )706 if path not in publication_claim707 )708 publication_state["publisher_agent_name"] = agent_name709 publication_state["deliverable_manifest"] = publication_claim710 publication_state["retired_paths"] = retired_paths711 task_metadata["retired_paths"] = list(retired_paths)712 publish_prompt = expanded_prompt713 if retired_paths:714 publish_prompt += render_retirement_note(retired_paths)715 spawn_context["delegation_prompt"] = publish_prompt716 try:717 job_id = await bus.submit_task_to_session(718 session_id,719 publish_prompt,720 spawn_context=spawn_context,721 task_metadata=task_metadata,722 )723 except Exception:724 if previous_publisher:725 publication_state["publisher_agent_name"] = (726 previous_publisher727 )728 else:729 publication_state.pop("publisher_agent_name", None)730 if previous_manifest:731 publication_state["deliverable_manifest"] = (732 previous_manifest733 )734 else:735 publication_state.pop("deliverable_manifest", None)736 if previous_retired:737 publication_state["retired_paths"] = previous_retired738 else:739 publication_state.pop("retired_paths", None)740 raise741 else:742 job_id = await bus.submit_task_to_session(743 session_id,744 expanded_prompt,745 spawn_context=spawn_context,746 task_metadata=task_metadata,747 )748 submitted.append({"agent": agent_name, "job_id": job_id})749 except RuntimeError as exc:750 errors.append(f"{agent_name}: {exc}")751 except Exception as exc:752 logger.warning(753 "assign_task: failed for %s: %s", agent_name, exc,754 )755 errors.append(f"{agent_name}: {exc}")756 757 all_errors = [*validation_errors, *errors]758 if not submitted and all_errors:759 return "Error: " + "; ".join(all_errors)760 761 # Derive effective authority only after submission: merely carrying a762 # manifest spec does not establish it (the agent may be unknown or763 # capped, the manifest may conflict, or enqueueing may fail). The manifest764 # is also path-specific: a publisher for report.pdf cannot write answer.md.765 effective_manifest: tuple[str, ...] = ()766 if is_agent_team and runtime is not None:767 effective_manifest = tuple(768 (getattr(runtime, "publication_state", None) or {}).get(769 "deliverable_manifest"770 )771 or ()772 )773 uncovered_directives = tuple(774 path for path in question_directives if path not in effective_manifest775 )776 # A research-only round is a legitimate reason to have no authorized path,777 # so this is a notice, not an error. Left unsaid, it is exactly the shape778 # that loses the deliverable at finalization.779 if is_agent_team and submitted and uncovered_directives:780 named = ", ".join(uncovered_directives)781 if effective_manifest:782 replacement_manifest = list(dict.fromkeys(783 (*effective_manifest, *uncovered_directives),784 ))785 notices.append(786 f"No agent in this run can write {named}, which the question "787 "names as the deliverable. The recorded publisher manifest "788 f"covers {list(effective_manifest)!r}, not every required "789 "path. Reuse that publisher with replace_manifest=true and "790 f"output_paths={replacement_manifest!r}."791 )792 else:793 notices.append(794 f"No agent in this run can write {named}, which the question "795 "names as the deliverable. That is expected for a research or "796 "verification round; before the run ends, collect the "797 "workspace paths from these reports and assign one task "798 f"with output_paths={list(uncovered_directives)!r}."799 )800 801 lines = [802 f"Submitted {len(submitted)} task(s) in parallel "803 f"(agents run concurrently in the background):"804 ]805 for s in submitted:806 lines.append(f" - {s['agent']}")807 if all_errors:808 lines.append("")809 lines.append("Warnings:")810 for e in all_errors:811 lines.append(f" - {e}")812 if notices:813 lines.append("")814 lines.append("Publishing notices:")815 for n in notices:816 lines.append(f" - {n}")817 lines.append("")818 lines.append(819 "Reports arrive automatically between turns. Call collect_reports() "820 "only if you have no useful local work and need a running agent's "821 "result before deciding."822 )823 return "\n".join(lines)824 825 826_AGENT_TEAM_ASSIGN_TASK_PARAMETERS = {827 "type": "object",828 "properties": {829 "tasks": {830 "type": "array",831 "items": AgentTeamAssignmentSpec.model_json_schema(),832 "description": (833 "Assignment objects. Give output_paths to the one final "834 "publisher; omit it for workspace-only work."835 ),836 },837 },838 "required": ["tasks"],839}840 841 842def _top_level_grants_authority(843 publish: bool | str | None,844 output_paths: list[str] | str | None,845) -> bool:846 """Whether top-level metadata would hand out ``/outputs`` write authority.847 848 Read by both the fold and the guard in front of it. Keeping the test in one849 place is deliberate: a literal check on one side and a normalising check on850 the other is what previously let ``"True"`` slip past the guard while the851 fold refused to expand it, so the batch ran with the publish intent and its852 manifest silently dropped.853 854 ``output_paths`` counts on its own now that it is the grant -- without this855 a top-level manifest would be folded into every item of a research batch856 and authorize all of them. Before, each item's ``publish: false`` collided857 with the folded manifest and the contract rejected the call; with the858 boolean optional there is nothing left to collide with.859 """860 if AgentTeamAssignmentSpec.normalise_publish_boolean(publish) is True:861 return True862 if output_paths is None:863 return False864 coerced = coerce_json_list(output_paths)865 if isinstance(coerced, str):866 return bool(coerced.strip())867 return bool(coerced)868 869 870def _fold_top_level_publish_metadata(871 tasks: list[AgentTeamAssignmentSpec | dict[str, Any]] | str,872 *,873 publish: bool | str | None = None,874 output_paths: list[str] | str | None = None,875 replace_manifest: bool | str | None = None,876) -> list[AgentTeamAssignmentSpec | dict[str, Any]] | str:877 """Recover a common model formatting error without widening authority.878 879 The canonical schema keeps publication metadata inside each ``tasks[]``880 item. Models occasionally emit the same metadata beside ``tasks``. Metadata881 that grants nothing can safely apply to every item; anything that grants882 ``/outputs`` authority is accepted only for a single assignment, so it883 cannot accidentally grant publication rights to a batch of researchers.884 885 Only *missing* keys are filled in. A call that already carries correct886 per-item metadata and merely echoes a summary value at the top level must887 come through unchanged: overwriting would demote the real publisher and888 then fail the contract on its now-contradictory ``output_paths``.889 """890 if publish is None and output_paths is None and replace_manifest is None:891 return tasks892 raw_tasks = coerce_json_list(tasks)893 if not isinstance(raw_tasks, list) or not raw_tasks:894 return tasks895 # A non-mapping item is the model's error to hear about, not something to896 # crash on: leave the list alone so per-task validation reports it.897 if not all(isinstance(raw, (BaseModel, dict)) for raw in raw_tasks):898 return tasks899 publish_value = AgentTeamAssignmentSpec.normalise_publish_boolean(publish)900 if _top_level_grants_authority(publish, output_paths) and len(raw_tasks) != 1:901 return tasks902 folded: list[AgentTeamAssignmentSpec | dict[str, Any]] = []903 for raw in raw_tasks:904 item = raw.model_dump() if isinstance(raw, BaseModel) else dict(raw)905 if publish is not None:906 item.setdefault("publish", publish_value)907 if output_paths is not None:908 item.setdefault("output_paths", output_paths)909 if replace_manifest is not None:910 item.setdefault("replace_manifest", replace_manifest)911 folded.append(item)912 return folded913 914 915@tool(916 name="assign_task",917 description=assign_task.description,918 parameters=_AGENT_TEAM_ASSIGN_TASK_PARAMETERS,919)920async def agent_team_assign_task(921 tasks: list[AgentTeamAssignmentSpec | dict[str, Any]] | str = "",922 publish: bool | str | None = None,923 output_paths: list[str] | str | None = None,924 replace_manifest: bool | str | None = None,925) -> str:926 """Assign agent-team tasks, authorizing at most one of them to publish.927 928 Args:929 tasks: Assignment objects. The one final publisher carries its exact930 absolute ``output_paths`` manifest, which is what grants it write931 access to those paths; every other item omits ``output_paths`` and932 runs workspace-only.933 """934 tasks = _fold_top_level_publish_metadata(935 tasks,936 publish=publish,937 output_paths=output_paths,938 replace_manifest=replace_manifest,939 )940 if _top_level_grants_authority(publish, output_paths):941 # The fold silently declines to expand an authority-granting value942 # across a batch. Saying so is the difference between the coordinator943 # re-sending the manifest on the right item and a round that runs with944 # no publisher and no error to act on.945 parsed = coerce_json_list(tasks)946 if not isinstance(parsed, list) or len(parsed) != 1:947 return (948 "Error: top-level output_paths (or publish=true) is accepted "949 "only for one task; put the output_paths manifest inside the "950 "single tasks[] item that should publish."951 )952 raw_item = parsed[0]953 item = (954 raw_item.model_dump()955 if isinstance(raw_item, BaseModel)956 else raw_item957 )958 if isinstance(item, dict) and not _top_level_grants_authority(959 item.get("publish"), item.get("output_paths"),960 ):961 return (962 "Error: top-level output_paths (or publish=true) conflicts "963 "with the single tasks[] item, whose existing publish or "964 "output_paths value prevents that authority from being "965 "applied. Put the complete output_paths manifest inside that "966 "task and omit the top-level publication fields."967 )968 return await assign_task.func(tasks=tasks)969 970 971__all__ = [972 "AgentTeamAssignmentSpec",973 "AssignmentSpec",974 "_fold_top_level_publish_metadata",975 "_top_level_grants_authority",976 "agent_team_assign_task",977 "assign_task",978]979