apodex/frontier-agent-demo
14
1"""task_board — a per-run task board for the coordinator (main agent)."""2 3from __future__ import annotations4 5import logging6from typing import Any7 8from frontier_agent.components.agent_bus import AgentBus9from frontier_agent.components.observers.task_board import TaskBoardObserver10from frontier_agent.components.task_board_types import (11 BOARD_TOOLS,12 RESOLUTION_MARKS,13 VALID_RESOLUTION,14 BoardCounts,15 count_resolutions,16)17from frontier_agent.core.execution_context import get_current_execution_scope18from frontier_agent.core.runtime import registry19from frontier_agent.core.tool import tool20from plugins.tools._bus_scope import resolve_bus_task_id21from plugins.tools._coerce import coerce_json_list, coerce_json_object22 23logger = logging.getLogger(__name__)24 25# task_id -> {"seq": int, "tasks": {id: {description, resolution, owners, group}}}26_BOARDS: dict[str, dict[str, Any]] = {}27 28# task_id -> list of pending board write-ops the stream observer has not yet29# drained. Each op is ``{"op": "add"|"update"|"finish_planning", "ids": [...],30# "phase": "planning"|"execution"}``. The board tools append here on every write;31# A task-board stream observer drains them and emits one32# ``response.swarm.task_board`` frame per op. Recording is unconditional + cheap (the33# eval / HTTP paths simply never drain), and ``clear_board`` empties it at run end34# so it can never leak across trials.35_PENDING_OPS: dict[str, list[dict[str, Any]]] = {}36 37 38def _record_op(task_id: str, op: str, ids: list[str] | None = None) -> None:39 """Append a board write-op for the stream observer to drain.40 41 The op stamps the phase AT WRITE TIME. In the agent-team two-loop profile the42 stream observer is not attached to the planning loop, so ops written there are43 drained late (once the execution loop runs) — by then the phase has flipped to44 ``execution``. Freezing the write-time phase keeps a planning-loop ``add_task``45 rendered as ``phase=planning`` rather than the phase current at drain time.46 """47 _PENDING_OPS.setdefault(task_id, []).append(48 {"op": op, "ids": list(ids or []), "phase": current_phase(task_id)}49 )50 51 52# ── State helpers (also used by the observer + finalize_answer gate) ────────53 54def _board(task_id: str) -> dict[str, Any]:55 return _BOARDS.setdefault(task_id, {"seq": 0, "tasks": {}})56 57 58def board_size(task_id: str) -> int:59 """Number of tasks on this task's board (0 if no board)."""60 b = _BOARDS.get(task_id)61 return len(b["tasks"]) if b else 062 63 64def build_task_board_observer(*, cooldown_turns: int = 5) -> TaskBoardObserver:65 """Bind the shared observer to this plugin's task-board state."""66 return TaskBoardObserver(67 board_size=board_size,68 render_board=lambda task_id, bus_task_id: render_board(69 task_id, bus_task_id=bus_task_id,70 ),71 resolve_bus_task_id=resolve_bus_task_id,72 cooldown_turns=cooldown_turns,73 )74 75 76def clear_board(task_id: str) -> None:77 """Drop a task's board + phase — called at the end of the main-agent run."""78 _BOARDS.pop(task_id, None)79 _PHASE.pop(task_id, None)80 _PENDING_OPS.pop(task_id, None)81 82 83# ── Task-board stream projection — pure read helpers ─────────────────────────84# Consumed by the protocol layer's task-board stream observer85# to build the ``task_board.*`` wire frames. Kept here (next to the state they86# read) and side-effect-free so the observer stays a thin emitter.87 88def current_phase(task_id: str) -> str:89 """The board's coordinator phase — ``planning`` or ``execution``.90 91 Defaults to ``execution`` when the run never entered Planning Mode92 (``planning_mode:false`` profiles — apodex production — never call93 ``start_planning``, so ``_PHASE`` has no entry and the board is live in94 execution from the first ``add_task``)."""95 return _PHASE.get(task_id, "execution")96 97 98def serialize_tasks(task_id: str, ids: list[str]) -> list[dict[str, Any]]:99 """Render the given task ids to the wire shape, reading CURRENT board state.100 101 Reading the live ``resolution`` (rather than assuming ``open``) matters: an102 ``add_task`` that de-dups onto an already-``resolved`` id must carry that103 real resolution, or the frontend's whole-row replace would roll the task104 back to ``open`` and regress progress. Ids no longer on the105 board (raced clear) are skipped rather than emitted as ghosts."""106 b = _BOARDS.get(task_id)107 if not b:108 return []109 out: list[dict[str, Any]] = []110 for tid in ids:111 t = b["tasks"].get(tid)112 if t is None:113 continue114 out.append({115 "id": tid,116 "description": t.get("description", ""),117 "resolution": t.get("resolution", "open"),118 "owners": list(t.get("owners") or []),119 "group": t.get("group", ""),120 })121 return out122 123 124def snapshot_tasks(task_id: str) -> list[dict[str, Any]]:125 """Return the complete board in display order for UI projections.126 127 Unlike :func:`render_board`, this keeps task data structured so a client128 can render a real board instead of scraping the human-readable tool result.129 """130 b = _BOARDS.get(task_id)131 if not b:132 return []133 return serialize_tasks(task_id, list(b["tasks"]))134 135 136def drain_board_ops(task_id: str) -> list[dict[str, Any]]:137 """Pop and return all pending board write-ops for this task (FIFO)."""138 return _PENDING_OPS.pop(task_id, [])139 140 141# ── Planning Mode (two-phase state machine) ─────────────────────────────────142# task_id -> "planning" | "execution". Default (no entry) = execution, so ONLY143# the agent-team main agent (which calls start_planning) is ever gated; every144# other caller / workflow is unaffected.145_PHASE: dict[str, str] = {}146# In Planning Mode the agent is a PLANNER, not a solver: it may use only147# READ-ONLY tools (to understand the problem / look up a basic term) plus the148# board tools. EVERYTHING ELSE — team building, dispatch, code/file writes,149# finalize — is blocked until it calls finish_planning. This is an ALLOWLIST150# (the complement is blocked) so a newly-added write tool is denied by default.151_PLANNING_ALLOWED = (152 # read-only inspection / understanding153 "grep_search", "glob_search", "read_file", "read_text", "view_image",154 "web_search", "web_fetch",155 # board tools156 "add_task", "update_task", "finish_planning",157)158 159 160def start_planning(task_id: str) -> None:161 _PHASE[task_id] = "planning"162 163 164def force_finish_planning(task_id: str) -> None:165 """Flip a task out of Planning Mode WITHOUT the empty-board guard the166 ``finish_planning`` tool enforces. Used by the planning-turn-cap path167 (auto-finish at ``planning_max_turns``) and by the two-loop driver after a168 planning loop that ended on max_turns rather than an explicit finish."""169 if task_id in _PHASE:170 _PHASE[task_id] = "execution"171 172 173def is_planning_allowed(tool_name: str) -> bool:174 """True if ``tool_name`` may run during Planning Mode (read-only / board)."""175 return tool_name in _PLANNING_ALLOWED176 177 178def in_planning(task_id: str) -> bool:179 return _PHASE.get(task_id) == "planning"180 181 182def planning_enabled(task_id: str) -> bool:183 """True if this run enabled Planning Mode (start_planning was called) —184 stays True after finish_planning, so the no-solo / verify finalize gates185 keep applying through execution. ``False`` for non-planning runs."""186 return task_id in _PHASE187 188 189def planning_block_message(task_id: str, tool_name: str) -> str | None:190 """Error string to return when ``tool_name`` is called during Planning Mode;191 ``None`` when the call is allowed. No-op unless start_planning() was called192 for this task (i.e. only the agent-team main agent)."""193 if in_planning(task_id) and not is_planning_allowed(tool_name):194 return (195 f"Blocked: `{tool_name}` is unavailable in PLANNING MODE. You are a "196 "PLANNER here — use only READ-ONLY tools (grep_search / glob_search "197 "/ web_search to understand the problem) and the board tools "198 "(add_task / update_task). List EVERY sub-question via add_task, "199 "then call finish_planning() to unlock the team."200 )201 return None202 203 204def unresolved_task_ids(task_id: str) -> list[str]:205 """Ids not yet finished — open or in_progress. ``[]`` if no board.206 207 Cancelled tasks are retracted work (dropped on purpose / created in error),208 so they do NOT count as unresolved and never hold back the finalize gate."""209 b = _BOARDS.get(task_id)210 if not b:211 return []212 return [213 tid for tid, t in b["tasks"].items()214 if t.get("resolution") not in ("resolved", "cancelled")215 ]216 217 218def _exec_status_by_owner(task_id: str) -> dict[str, str]:219 """Map owner sub-agent name -> coarse execution status, derived from the bus.220 221 This is the system-owned half of the board: the model never writes it, so222 the model's ``resolution`` verdict can never clobber the live run state.223 """224 bus = registry.get_optional(AgentBus)225 if bus is None:226 return {}227 out: dict[str, str] = {}228 try:229 sessions = bus.list_sessions_for_task(task_id)230 except Exception:231 return {}232 for s in sessions:233 if getattr(s, "current_job_id", None) is not None:234 st = "running"235 elif getattr(s, "pending_tasks", None):236 st = "queued"237 elif getattr(s, "total_task_count", 0) == 0:238 st = "created"239 else:240 st = "reported"241 out[s.name] = st242 return out243 244 245def render_board(task_id: str, *, bus_task_id: str | None = None) -> str:246 """Render the board, joining model fields with live bus exec-status."""247 b = _BOARDS.get(task_id)248 if not b or not b["tasks"]:249 return "[task board] empty — call add_task to register sub-questions."250 tasks = b["tasks"]251 c = count_resolutions(t["resolution"] for t in tasks.values())252 lines = [253 f"[task board] resolved {c.resolved}/{c.active} · "254 f"in-progress {c.in_progress} · "255 f"open {c.open} · cancelled {c.cancelled}"256 ]257 # The owners/exec column only makes sense when there IS a team: in a solo258 # run (e.g. react) no task is ever assigned, so showing "agents=[unassigned]"259 # on every row is pure noise. Drop the whole column unless at least one task260 # has an owner — then skip the bus query too (nothing to join against).261 show_owners = any(t.get("owners") for t in tasks.values())262 exec_by_owner = _exec_status_by_owner(bus_task_id or task_id) if show_owners else {}263 for tid, t in tasks.items():264 row = (265 f" {RESOLUTION_MARKS.get(t['resolution'], '○')} {tid} "266 f"{t['resolution']:<11} "267 )268 if show_owners:269 owners = t.get("owners") or []270 # one "name:exec_status" per owning agent, so the coordinator sees271 # exactly who is on this task and how far each has got.272 agents = (273 " · ".join(f"{o}:{exec_by_owner.get(o, '?')}" for o in owners)274 if owners else "unassigned"275 )276 row += f"agents=[{agents}] "277 lines.append(f"{row}{t['description'][:80]}")278 return "\n".join(lines)279 280 281# ── Tools ───────────────────────────────────────────────────────────────────282 283def _as_owner_list(val: Any) -> list[str]:284 """Normalise an ``owner`` field (a name, a comma-string, or a list of names)285 into a clean list of agent names. One task can have MANY owners — several286 agents attacking the SAME sub-question from different angles for287 corroboration are all owners of that one task (not separate tasks)."""288 if val is None:289 return []290 items = val if isinstance(val, list) else str(val).split(",")291 out: list[str] = []292 for x in items:293 name = str(x).strip()294 if name and name not in out:295 out.append(name)296 return out297 298 299@tool300async def add_task(tasks: list[Any]) -> str:301 """Register the work items / sub-questions for this run on the task board.302 303 This is your plan and external memory. Call it up front — before doing any304 real work (fetching, running code, gathering evidence) — once you've broken305 the question into the concrete steps you'll work through, and again whenever306 a new sub-question emerges. (Reasoning, and a few clarifying searches to307 understand the problem, may come first.) Duplicate descriptions are308 de-duplicated (you get the existing id).309 310 Args:311 tasks: a list, each item ``{"description": str}``:312 - description (required): ONE concrete, checkable work item, e.g.313 "Verify the Markov condition Δω ≫ system rate holds" — not a vague314 area like "look into the math".315 - owner (OPTIONAL, multi-agent runs only): if a teammate sub-agent is316 already assigned to this item, name it here (a name, comma-string,317 or list — a task may have many). In a solo run, just OMIT it.318 319 Returns:320 The assigned ids plus the rendered board.321 """322 items = coerce_json_list(tasks) or []323 scope = get_current_execution_scope()324 if scope is None:325 return "Error: add_task can only be called inside an active run."326 if not items:327 return "Error: add_task requires at least one {description} item."328 b = _board(scope.task_id)329 existing = {t["description"].strip(): tid for tid, t in b["tasks"].items()}330 new_ids: list[str] = []331 skipped = 0 # items that weren't usable {description} objects332 for raw in items:333 it = coerce_json_object(raw)334 if it is None:335 skipped += 1336 continue337 desc = str(it.get("description", "")).strip()338 if not desc:339 skipped += 1340 continue341 if desc in existing: # dedup re-decomposition342 new_ids.append(existing[desc])343 continue344 b["seq"] += 1345 tid = f"t{b['seq']}"346 b["tasks"][tid] = {347 "description": desc,348 "resolution": "open",349 "owners": _as_owner_list(it.get("owner") or it.get("owners")),350 "group": str(it.get("group", "")).strip(),351 }352 existing[desc] = tid353 new_ids.append(tid)354 logger.info(355 "add_task(task=%s): +%d skipped=%d (ids=%s)",356 scope.task_id, len(new_ids), skipped, new_ids,357 )358 # Nothing landed but items were passed → the shape was wrong. Return a359 # corrective error (not a silent "Added []") so the model re-sends the right360 # shape instead of burning turns repeating the mistake.361 if not new_ids:362 return (363 'Error: add_task expects a list of objects like '364 '[{"description": "..."}]; none of the items were usable, so nothing '365 'was added. Re-call with each task as its own '366 '{"description": "<one concrete work item>"}.'367 )368 _record_op(scope.task_id, "add", new_ids)369 msg = f"Added {new_ids}.\n{render_board(scope.task_id, bus_task_id=resolve_bus_task_id(scope))}"370 if skipped:371 msg += (372 f'\nNote: {skipped} item(s) were skipped (not a {{"description": ...}} '373 "object). Re-add them with that shape if still needed."374 )375 return msg376 377 378@tool379async def update_task(updates: list[Any]) -> str:380 """Mark progress on the board — call this the MOMENT a task finishes.381 382 Update each task as soon as it is done, before starting the next one; don't383 let finished tasks pile up. The arg is a LIST only to cover the case where384 two tasks finished in the SAME turn (resolve both at once) — it is NOT a385 reason to batch resolutions across turns. Returns the full updated board.386 387 Args:388 updates: a list of ``{"id", "resolution"?}`` items:389 - id (required): task id from add_task, e.g. "t3".390 - resolution: "in_progress" (set this the MOMENT you start working a391 task — it marks the one you are on now) | "resolved" (the work item392 is answered AND corroborated — your judgment, not merely "I glanced393 at it") | "cancelled" (retract a task you no longer need or created394 in error — it stops counting toward unresolved work; the id stays395 on the board for the trail) | "open" (the default; not started).396 - owner (OPTIONAL, multi-agent runs only): teammate sub-agent397 name(s) now working it — ADDED to the task's owner list (a name,398 comma-string, or list). Set "replace_owners": true to overwrite399 instead of add. Omit entirely in a solo run.400 401 Returns:402 The full updated task board (+ any per-item errors).403 """404 items = coerce_json_list(updates) or []405 scope = get_current_execution_scope()406 if scope is None:407 return "Error: update_task can only be called inside an active run."408 b = _BOARDS.get(scope.task_id)409 if not b:410 return "Error: no task board yet — call add_task first."411 changed: list[str] = []412 errors: list[str] = []413 for raw in items:414 u = coerce_json_object(raw)415 if u is None:416 continue417 tid = str(u.get("id", ""))418 if tid not in b["tasks"]:419 errors.append(f"{tid or '?'}: no such task")420 continue421 res = str(u.get("resolution", "")).strip()422 if res and res not in VALID_RESOLUTION:423 errors.append(f"{tid}: bad resolution {res!r} (use {VALID_RESOLUTION})")424 continue425 t = b["tasks"][tid]426 t.setdefault("owners", [])427 if res:428 t["resolution"] = res429 # owners ACCUMULATE — assigning another agent to the same task adds it,430 # it does not replace the existing owner(s). ``replace_owners: true``431 # overwrites (e.g. to drop an agent that was reassigned elsewhere).432 new_owners = _as_owner_list(u.get("owner") or u.get("owners"))433 if new_owners:434 if u.get("replace_owners"):435 t["owners"] = new_owners436 else:437 for o in new_owners:438 if o not in t["owners"]:439 t["owners"].append(o)440 changed.append(tid)441 if not changed and not errors:442 return ("Error: update_task needs a list of "443 "{id, resolution?, owner?} items.")444 # Emit ONLY the actually-changed ids — rejected ids (bad id /445 # bad resolution) stay off the wire so the frontend never builds ghost rows.446 if changed:447 _record_op(scope.task_id, "update", changed)448 msg = f"Updated {changed}.\n{render_board(scope.task_id, bus_task_id=resolve_bus_task_id(scope))}"449 if errors:450 msg += "\nerrors: " + "; ".join(errors)451 return msg452 453 454@tool455async def finish_planning() -> str:456 """Leave Planning Mode and start building the team.457 458 While planning, only add_task / update_task are available. Call this once459 your task board lists EVERY sub-question; afterwards create_subagent /460 assign_task / collect_reports become available (you can still add_task /461 update_task to refine the plan as the investigation unfolds).462 """463 scope = get_current_execution_scope()464 if scope is None:465 return "Error: finish_planning can only be called inside an active run."466 if board_size(scope.task_id) == 0:467 return (468 "Error: the task board is empty — call add_task to list the "469 "sub-questions before finishing planning."470 )471 _PHASE[scope.task_id] = "execution"472 _record_op(scope.task_id, "finish_planning")473 return (474 "Planning complete — now in EXECUTION mode. You may create_subagent / "475 "assign_task to build and dispatch the team.\n"476 + render_board(scope.task_id, bus_task_id=resolve_bus_task_id(scope))477 )478 479 480__all__ = [481 "BOARD_TOOLS",482 "RESOLUTION_MARKS",483 "VALID_RESOLUTION",484 "BoardCounts",485 "add_task",486 "board_size",487 "build_task_board_observer",488 "clear_board",489 "count_resolutions",490 "current_phase",491 "drain_board_ops",492 "finish_planning",493 "force_finish_planning",494 "in_planning",495 "is_planning_allowed",496 "planning_block_message",497 "planning_enabled",498 "render_board",499 "serialize_tasks",500 "snapshot_tasks",501 "start_planning",502 "unresolved_task_ids",503 "update_task",504]505