Coverage for cli / commands / swarm_cmd.py: 100.00%
281 statements
« prev ^ index » next coverage.py v7.13.5, created at 2026-09-14 22:07 +0000
1"""``gco swarm`` — supervise a fleet of child Mission sessions.
3One orchestrator Mission session spawns and drives concurrent child
4Mission sessions through in-process supervisor tools, under hard rails
5(fleet cap, pooled child-iteration budget, concurrency bound, finite
6child budgets), until the orchestrator's deterministic verdict cascade
7reaches a terminal verdict. See ``docs/SWARM.md`` for the model.
9The whole subcommand group is gated by ``GCO_ENABLE_SWARM``: when the
10env var is unset, the group prints a one-line hint and exits with code
112 before dispatching to any subcommand. With the flag set, the
12subcommands talk directly to the persistence backend, the swarm rules
13in ``mission/swarm.py``, and the child runner — no MCP round-trip is
14involved, so the CLI works without the MCP server running.
16Subcommands:
18* ``run`` — scaffold a plan from a directive, start a swarm, prime the
19 fleet, and drive it to completion synchronously.
20* ``start`` — validate inputs and persist a new orchestrator session.
21* ``iterate`` — drive (or resume) an existing swarm's fleet.
22* ``status`` — the one-call fleet rollup document.
23* ``abort`` — terminate the orchestrator and abort every live child.
24* ``list`` — list orchestrator sessions.
25* ``scaffold-plan`` — draft a validated Swarm_Plan without starting.
27Output formats: every subcommand defaults to ``--output json``; pass
28``--output table`` for a human-readable summary where offered.
29"""
31from __future__ import annotations
33import asyncio
34import json
35import os
36import secrets
37import sys
38from datetime import UTC, datetime
39from pathlib import Path
40from typing import Any
42import click
44# The Mission package lives under ``gco_mcp/mission/`` and is imported as
45# ``mission.*``. Match the path-injection pattern used throughout the
46# MCP module surface so the imports below resolve regardless of how this
47# module is loaded.
48sys.path.insert(0, str(Path(__file__).resolve().parent.parent.parent / "gco_mcp"))
50from gco.bedrock import BedrockFTUFormNotAcceptedError # noqa: E402
52_FEATURE_FLAG_HINT = (
53 "Swarm tools are gated. Set GCO_ENABLE_SWARM=true (or GCO_ENABLE_ALL_TOOLS=true) to enable."
54)
57def _flag_enabled() -> bool:
58 """Return True iff ``GCO_ENABLE_SWARM`` (or umbrella) is truthy."""
59 truthy = {"true", "1", "yes", "on"}
60 return (
61 os.environ.get("GCO_ENABLE_SWARM", "").strip().lower() in truthy
62 or os.environ.get("GCO_ENABLE_ALL_TOOLS", "").strip().lower() in truthy
63 )
66def _check_feature_flag() -> None:
67 """Print the hint and exit with code 2 when the gating flag is unset."""
68 if not _flag_enabled():
69 click.echo(_FEATURE_FLAG_HINT, err=True)
70 raise SystemExit(2)
73def _emit_json(payload: Any, *, err: bool = False) -> None:
74 """Emit ``payload`` as a single JSON line."""
75 from ..output import emit_structured_document
77 emit_structured_document(
78 payload,
79 output_format="json",
80 rendered=json.dumps(payload, default=str),
81 err=err,
82 )
85def _emit_json_text(text: str) -> None:
86 """Emit pre-rendered JSON while registering its native document shape."""
87 from ..output import emit_structured_document
89 try:
90 document = json.loads(text)
91 except json.JSONDecodeError:
92 click.echo(text)
93 return
94 emit_structured_document(document, output_format="json", rendered=text)
97def _emit_error(code: str, details: dict[str, Any] | None = None) -> None:
98 """Emit a structured error envelope to stderr."""
99 payload: dict[str, Any] = {"code": code}
100 if details is not None:
101 payload["details"] = details
102 _emit_json(payload, err=True)
105# ---------------------------------------------------------------------------
106# Registry and engine wiring
107# ---------------------------------------------------------------------------
110def _ensure_tool_registry() -> None:
111 """Register every MCP tool against the shared FastMCP server, once.
113 Same on-demand registration the mission CLI performs: the live tool
114 dispatcher and the spawn validators look tools up on the shared
115 FastMCP instance, which starts empty in a plain CLI process.
116 """
117 sys.path.insert(0, str(Path(__file__).resolve().parent.parent.parent / "gco_mcp"))
118 from tools import register_all_tools # noqa: PLC0415
120 register_all_tools()
123def _resolve_registered_tools_for_cli() -> tuple[dict[str, Any], dict[str, set[str]]]:
124 """Snapshot the live registry as ``(name -> Tool, name -> tags)``."""
125 _ensure_tool_registry()
126 from server import mcp # noqa: PLC0415 — lazy
128 async def _list() -> list[Any]:
129 return list(await mcp._list_tools())
131 tools = asyncio.run(_list())
132 registered = {t.name: t for t in tools}
133 tags = {t.name: set(getattr(t, "tags", None) or ()) for t in tools}
134 return registered, tags
137def _tool_docstrings(registered: dict[str, Any]) -> dict[str, str]:
138 return {name: str(getattr(tool, "description", "") or "") for name, tool in registered.items()}
141def _deps_builder(*, dry_run: bool) -> Any:
142 """The runner's per-session engine dependency factory for the CLI path.
144 Orchestrator sessions get the supervisor-tool catalog stubs merged
145 into their sampler metadata (spawn proposals then validate against
146 the catalog); children build plain engines. ``--dry-run`` swaps in
147 the canned-stub dispatcher, mirroring ``gco mission run``.
148 """
149 from mission._engine_factory import build_engine_dependencies # noqa: PLC0415
150 from mission.swarm import ( # noqa: PLC0415
151 SUPERVISOR_TOOL_DOCSTRINGS,
152 SUPERVISOR_TOOL_SCHEMAS,
153 SUPERVISOR_TOOLS,
154 )
156 class _SchemaShim:
157 def __init__(self, schema: dict[str, Any]) -> None:
158 self._schema = schema
160 def model_json_schema(self) -> dict[str, Any]:
161 return self._schema
163 class _SupervisorToolStub:
164 def __init__(self, name: str) -> None:
165 self.name = name
166 self.description = SUPERVISOR_TOOL_DOCSTRINGS[name]
167 self.tags = {"swarm", "supervisor"}
168 self.input_schema = _SchemaShim(SUPERVISOR_TOOL_SCHEMAS[name])
170 async def build(session: Any) -> Any:
171 extra = None
172 if session.get("role") == "orchestrator":
173 extra = (
174 {name: _SupervisorToolStub(name) for name in SUPERVISOR_TOOLS},
175 dict(SUPERVISOR_TOOL_DOCSTRINGS),
176 )
177 return await build_engine_dependencies(
178 session, None, use_stub_dispatcher=dry_run, extra_tool_metadata=extra
179 )
181 return build
184def _make_runner(orchestrator_id: str, *, dry_run: bool) -> Any:
185 from mission.state import get_backend # noqa: PLC0415
186 from mission.swarm_runner import SwarmRunner # noqa: PLC0415
188 registered, tags = _resolve_registered_tools_for_cli()
190 def _stream_verdict(record: Any) -> None:
191 _emit_json(
192 {
193 "event": "swarm.iteration",
194 "iteration_index": record.get("iteration_index"),
195 "verdict": record.get("verdict"),
196 "verdict_reason": record.get("verdict_reason"),
197 },
198 err=True,
199 )
201 return SwarmRunner(
202 backend=get_backend(),
203 orchestrator_id=orchestrator_id,
204 deps_builder=_deps_builder(dry_run=dry_run),
205 registered_tools=registered,
206 registered_tags=tags,
207 on_orchestrator_iteration=_stream_verdict,
208 )
211def _persist_orchestrator(
212 *,
213 directive: str,
214 criteria: list[dict[str, Any]],
215 budget: dict[str, Any],
216 swarm_config: dict[str, Any],
217 tool_allowlist: tuple[str, ...],
218 allow_all_tools: bool,
219 stagnation_threshold: int,
220 use_sampling: bool | None,
221) -> dict[str, Any]:
222 """Validate inputs and persist a new orchestrator session.
224 Exits 1 with the structured envelope on any validation failure —
225 before anything is persisted.
226 """
227 from mission import sampling as mission_sampling # noqa: PLC0415
228 from mission import swarm as swarm_rules # noqa: PLC0415
229 from mission import validation as mission_validation # noqa: PLC0415
230 from mission.state import get_backend # noqa: PLC0415
231 from mission.validation import MissionValidationError # noqa: PLC0415
233 try:
234 directive_clean = mission_validation.validate_directive(directive)
235 criteria_clean = mission_validation.validate_criteria(criteria)
236 swarm_clean = swarm_rules.validate_swarm_config(swarm_config)
237 budget_clean = mission_validation.validate_budget(budget, [], {})
238 cadence_clean = mission_validation.validate_cadence({"kind": "every_iteration"})
239 extra: list[str] = []
240 if allow_all_tools or tool_allowlist:
241 registered, _tags = _resolve_registered_tools_for_cli()
242 extra = mission_validation.resolve_effective_allowlist(
243 allow_all_tools=allow_all_tools,
244 explicit_allowlist=list(tool_allowlist),
245 registered_tools=registered,
246 )
247 except MissionValidationError as exc:
248 _emit_error(exc.code, exc.details)
249 raise SystemExit(1) from exc
251 use_resolved, backend_resolved = mission_sampling.resolve_sampling_state(use_sampling)
252 session = swarm_rules.build_orchestrator_session(
253 session_id=f"mission-{secrets.token_hex(8)}",
254 directive=directive_clean,
255 criteria=criteria_clean,
256 budget=budget_clean,
257 swarm_config=swarm_clean,
258 cadence=cadence_clean,
259 extra_allowlist=extra,
260 stagnation_threshold=stagnation_threshold,
261 use_sampling=use_resolved,
262 sampling_backend_resolved=backend_resolved,
263 created_at=datetime.now(UTC).isoformat(),
264 )
265 get_backend().save_session(session) # type: ignore[arg-type]
266 return session
269def _scaffold_plan(
270 *,
271 directive: str,
272 swarm_config: dict[str, Any],
273 tool_allowlist: tuple[str, ...],
274 allow_all_tools: bool,
275 max_children: int | None,
276 use_sampling: bool | None,
277 retries: int,
278) -> dict[str, Any]:
279 """Produce a validated Swarm_Plan; sampled when a backend resolves.
281 Returns ``{"plan": [...], "sampling_path": bool, "fallback_reason"}``.
282 Exits 1 on config/validation failures. A sampling failure falls back
283 to the deterministic single-worker plan with a one-line warning,
284 except the permanent Anthropic first-time-use gate, which is
285 reported as a hard error (Mission precedent).
286 """
287 from mission import sampling as mission_sampling # noqa: PLC0415
288 from mission import swarm as swarm_rules # noqa: PLC0415
289 from mission import swarm_scaffold # noqa: PLC0415
290 from mission.validation import MissionValidationError # noqa: PLC0415
292 try:
293 config = swarm_rules.validate_swarm_config(swarm_config)
294 except MissionValidationError as exc:
295 _emit_error(exc.code, exc.details)
296 raise SystemExit(1) from exc
297 registered, tags = _resolve_registered_tools_for_cli()
298 use_resolved, backend_name = mission_sampling.resolve_sampling_state(use_sampling)
299 plan: list[dict[str, Any]] | None = None
300 fallback_reason: str | None = None
301 if use_resolved:
302 backend_obj = mission_sampling.select_sampling_backend(None)
303 if backend_obj is not None:
304 try:
305 plan = asyncio.run(
306 swarm_scaffold.generate_sampled_plan(
307 backend_obj,
308 directive,
309 config=config,
310 registered_tools=registered,
311 registered_tags=tags,
312 tool_docstrings=_tool_docstrings(registered),
313 max_children=max_children,
314 tool_allowlist=(None if allow_all_tools else list(tool_allowlist) or None),
315 retries=retries,
316 )
317 )
318 except BedrockFTUFormNotAcceptedError as exc:
319 _emit_error("bedrock_ftu_form_not_accepted", {"message": str(exc)})
320 raise SystemExit(1) from exc
321 except swarm_scaffold.SwarmScaffoldError as exc:
322 fallback_reason = exc.last_reason
323 click.echo(
324 f"Sampled plan rejected ({exc.last_reason}); "
325 "falling back to the deterministic single-worker plan.",
326 err=True,
327 )
328 else:
329 fallback_reason = "sampling_backend_unavailable"
330 if plan is None:
331 try:
332 plan = swarm_scaffold.generate_deterministic_plan(
333 directive,
334 config=config,
335 registered_tools=registered,
336 registered_tags=tags,
337 tool_allowlist=list(tool_allowlist) or None,
338 allow_all_tools=allow_all_tools,
339 )
340 except MissionValidationError as exc:
341 _emit_error(exc.code, exc.details)
342 raise SystemExit(1) from exc
343 return {
344 "plan": plan,
345 "sampling_path": fallback_reason is None and use_resolved,
346 "sampling_backend_resolved": backend_name,
347 "fallback_reason": fallback_reason,
348 }
351async def _prime_and_run(
352 runner: Any, plan: list[dict[str, Any]], *, max_orchestrator_iterations: int | None = None
353) -> dict[str, Any]:
354 """Dispatch the plan's spawns through the runner seam, then drive.
356 Every spawn envelope streams to stderr; a rejected plan entry is a
357 hard failure (the plan was pre-validated, so a rejection here means
358 the world changed underneath it, and silently running a partial
359 fleet would be dishonest).
360 """
361 for request in plan:
362 result = await runner.spawn(request)
363 _emit_json({"event": "swarm.spawn", **result}, err=True)
364 if not result.get("spawned"):
365 raise SystemExit(1)
366 return dict(
367 await runner.run_to_completion(max_orchestrator_iterations=max_orchestrator_iterations)
368 )
371# ---------------------------------------------------------------------------
372# Click group
373# ---------------------------------------------------------------------------
376@click.group("swarm")
377def swarm_cmd() -> None:
378 """Swarm supervision: one orchestrator Mission driving child Missions.
380 Gated by GCO_ENABLE_SWARM. See docs/SWARM.md for the supervisor
381 model, the rails, and the determinism boundary.
382 """
383 _check_feature_flag()
386_SWARM_OPTIONS = [
387 click.option("--max-children", type=int, default=3, show_default=True),
388 click.option("--child-iteration-pool", type=int, default=15, show_default=True),
389 click.option("--max-concurrent-children", type=int, default=3, show_default=True),
390 click.option(
391 "--allow-overlapping-mutating-tools",
392 is_flag=True,
393 default=False,
394 help="Allow two live children to share a non-read-only tool.",
395 ),
396]
399def _swarm_options(func: Any) -> Any:
400 for option in reversed(_SWARM_OPTIONS):
401 func = option(func)
402 return func
405def _swarm_config_from_flags(
406 max_children: int,
407 child_iteration_pool: int,
408 max_concurrent_children: int,
409 allow_overlapping_mutating_tools: bool,
410) -> dict[str, Any]:
411 return {
412 "max_children": max_children,
413 "child_iteration_pool": child_iteration_pool,
414 "max_concurrent_children": max_concurrent_children,
415 "allow_overlapping_mutating_tools": allow_overlapping_mutating_tools,
416 }
419# ---------------------------------------------------------------------------
420# run
421# ---------------------------------------------------------------------------
424@swarm_cmd.command("run")
425@click.option("--directive", required=True, help="The swarm-level goal, natural language.")
426@click.option(
427 "--tool-allowlist",
428 multiple=True,
429 metavar="NAME",
430 help="Tool allowed to scaffolded children (repeatable).",
431)
432@click.option("--allow-all-tools", is_flag=True, default=False)
433@_swarm_options
434@click.option("--max-iterations", type=int, default=25, show_default=True)
435@click.option("--max-wall-clock", type=int, default=1800, show_default=True)
436@click.option("--stagnation-threshold", type=int, default=3, show_default=True)
437@click.option("--use-sampling/--no-sampling", "use_sampling", default=None)
438@click.option("--retries", type=int, default=3, show_default=True)
439@click.option(
440 "--save-plan",
441 type=click.Path(dir_okay=False, writable=True),
442 default=None,
443 help="Also write the scaffolded plan JSON to this path.",
444)
445@click.option(
446 "--dry-run",
447 is_flag=True,
448 default=False,
449 help="Use the canned-stub tool dispatcher (loop mechanics only).",
450)
451def swarm_run_cmd(
452 directive: str,
453 tool_allowlist: tuple[str, ...],
454 allow_all_tools: bool,
455 max_children: int,
456 child_iteration_pool: int,
457 max_concurrent_children: int,
458 allow_overlapping_mutating_tools: bool,
459 max_iterations: int,
460 max_wall_clock: int,
461 stagnation_threshold: int,
462 use_sampling: bool | None,
463 retries: int,
464 save_plan: str | None,
465 dry_run: bool,
466) -> None:
467 """Scaffold a plan, start a swarm, and drive it to completion.
469 Per-iteration verdicts and spawn envelopes stream to stderr as JSON
470 lines; the orchestrator's Final_Report lands on stdout when the
471 swarm reaches a terminal verdict.
472 """
473 swarm_config = _swarm_config_from_flags(
474 max_children,
475 child_iteration_pool,
476 max_concurrent_children,
477 allow_overlapping_mutating_tools,
478 )
479 scaffold = _scaffold_plan(
480 directive=directive,
481 swarm_config=swarm_config,
482 tool_allowlist=tool_allowlist,
483 allow_all_tools=allow_all_tools,
484 max_children=None,
485 use_sampling=use_sampling,
486 retries=retries,
487 )
488 plan = scaffold["plan"]
489 if save_plan:
490 Path(save_plan).write_text(json.dumps(plan, indent=2), encoding="utf-8")
491 # The default orchestrator criterion: every planned slot completed.
492 # Expressed over the fleet metrics so it is deterministic and
493 # readable in the session JSON.
494 criteria = [
495 {
496 "criterion_id": "fleet_completed",
497 "kind": "metric_threshold",
498 "required": True,
499 "metric": "metrics.children_completed",
500 "op": ">=",
501 "target": len(plan),
502 },
503 {
504 "criterion_id": "no_failed_children",
505 "kind": "metric_threshold",
506 "required": True,
507 "metric": "metrics.children_failed",
508 "op": "==",
509 "target": 0,
510 },
511 ]
512 session = _persist_orchestrator(
513 directive=directive,
514 criteria=criteria,
515 budget={"max_iterations": max_iterations, "max_wall_clock_seconds": max_wall_clock},
516 swarm_config=swarm_config,
517 tool_allowlist=(),
518 allow_all_tools=False,
519 stagnation_threshold=stagnation_threshold,
520 use_sampling=use_sampling if not dry_run else False,
521 )
522 _emit_json(
523 {
524 "event": "swarm.run.started",
525 "session_id": session["session_id"],
526 "plan_children": [entry["slot"] for entry in plan],
527 "sampling_path": scaffold["sampling_path"],
528 "fallback_reason": scaffold["fallback_reason"],
529 },
530 err=True,
531 )
532 final = _drive(session["session_id"], plan=plan, dry_run=dry_run)
533 _emit_report(final)
534 raise SystemExit(0 if final.get("final_verdict") == "complete" else 3)
537def _drive(
538 orchestrator_id: str,
539 *,
540 plan: list[dict[str, Any]] | None = None,
541 dry_run: bool = False,
542 max_orchestrator_iterations: int | None = None,
543) -> dict[str, Any]:
544 """Build a runner and drive the swarm, mapping runner errors to exits."""
545 from mission.swarm_runner import SwarmRunnerBusyError # noqa: PLC0415
546 from mission.validation import MissionValidationError # noqa: PLC0415
548 runner = _make_runner(orchestrator_id, dry_run=dry_run)
549 try:
550 return asyncio.run(
551 _prime_and_run(
552 runner,
553 plan or [],
554 max_orchestrator_iterations=max_orchestrator_iterations,
555 )
556 )
557 except SwarmRunnerBusyError as busy:
558 _emit_error(
559 "swarm_runner_active",
560 {"session_id": orchestrator_id, "holder_pid": busy.holder_pid},
561 )
562 raise SystemExit(1) from busy
563 except MissionValidationError as exc:
564 _emit_error(exc.code, exc.details)
565 raise SystemExit(1) from exc
568def _emit_report(final: dict[str, Any]) -> None:
569 """Print the Final_Report JSON to stdout when present, else a summary."""
570 report_path = final.get("final_report_path")
571 if report_path and Path(str(report_path)).exists():
572 _emit_json_text(Path(str(report_path)).read_text(encoding="utf-8"))
573 return
574 _emit_json(
575 {
576 "session_id": final.get("session_id"),
577 "status": final.get("status"),
578 "final_verdict": final.get("final_verdict"),
579 }
580 )
583# ---------------------------------------------------------------------------
584# start / iterate
585# ---------------------------------------------------------------------------
588@swarm_cmd.command("start")
589@click.option("--directive", required=True)
590@click.option(
591 "--criteria-file",
592 type=click.Path(exists=True, dir_okay=False),
593 required=True,
594 help="JSON array of orchestrator criteria (over the fleet metrics).",
595)
596@click.option("--tool-allowlist", multiple=True, metavar="NAME")
597@click.option("--allow-all-tools", is_flag=True, default=False)
598@_swarm_options
599@click.option("--max-iterations", type=int, default=25, show_default=True)
600@click.option("--max-wall-clock", type=int, default=1800, show_default=True)
601@click.option("--stagnation-threshold", type=int, default=3, show_default=True)
602@click.option("--use-sampling/--no-sampling", "use_sampling", default=None)
603def swarm_start_cmd(
604 directive: str,
605 criteria_file: str,
606 tool_allowlist: tuple[str, ...],
607 allow_all_tools: bool,
608 max_children: int,
609 child_iteration_pool: int,
610 max_concurrent_children: int,
611 allow_overlapping_mutating_tools: bool,
612 max_iterations: int,
613 max_wall_clock: int,
614 stagnation_threshold: int,
615 use_sampling: bool | None,
616) -> None:
617 """Persist a new swarm (orchestrator) session without driving it."""
618 try:
619 criteria = json.loads(Path(criteria_file).read_text(encoding="utf-8"))
620 except (OSError, ValueError) as exc:
621 _emit_error("validation_error", {"field": "criteria_file", "reason": str(exc)})
622 raise SystemExit(1) from exc
623 session = _persist_orchestrator(
624 directive=directive,
625 criteria=criteria,
626 budget={"max_iterations": max_iterations, "max_wall_clock_seconds": max_wall_clock},
627 swarm_config=_swarm_config_from_flags(
628 max_children,
629 child_iteration_pool,
630 max_concurrent_children,
631 allow_overlapping_mutating_tools,
632 ),
633 tool_allowlist=tool_allowlist,
634 allow_all_tools=allow_all_tools,
635 stagnation_threshold=stagnation_threshold,
636 use_sampling=use_sampling,
637 )
638 _emit_json(
639 {
640 "session_id": session["session_id"],
641 "status": session["status"],
642 "use_sampling": session["use_sampling"],
643 "swarm": session["swarm"],
644 }
645 )
648@swarm_cmd.command("iterate")
649@click.argument("session_id")
650@click.option(
651 "--max-orchestrator-iterations",
652 type=int,
653 default=None,
654 help="Detach after this many orchestrator iterations (fleet stays resumable).",
655)
656@click.option("--dry-run", is_flag=True, default=False)
657def swarm_iterate_cmd(
658 session_id: str, max_orchestrator_iterations: int | None, dry_run: bool
659) -> None:
660 """Drive (or resume) an existing swarm's fleet.
662 Also the crash-recovery path: a fresh runner re-schedules every live
663 child and evaluates restart policy for children that went terminal
664 while unsupervised.
665 """
666 final = _drive(
667 session_id,
668 dry_run=dry_run,
669 max_orchestrator_iterations=max_orchestrator_iterations,
670 )
671 _emit_json(
672 {
673 "session_id": session_id,
674 "status": final.get("status"),
675 "final_verdict": final.get("final_verdict"),
676 "iterations_run": len(final.get("iterations", [])),
677 }
678 )
681# ---------------------------------------------------------------------------
682# status / abort / list / scaffold-plan
683# ---------------------------------------------------------------------------
686def _load_orchestrator_or_exit(session_id: str) -> tuple[Any, dict[str, Any]]:
687 from mission.state import get_backend # noqa: PLC0415
689 backend = get_backend()
690 session = backend.load_session(session_id)
691 if session is None:
692 _emit_error("session_not_found", {"session_id": session_id})
693 raise SystemExit(1)
694 if session.get("role") != "orchestrator" or "swarm" not in session:
695 _emit_error(
696 "validation_error",
697 {"field": "role", "reason": "not_an_orchestrator", "session_id": session_id},
698 )
699 raise SystemExit(1)
700 return backend, dict(session)
703@swarm_cmd.command("status")
704@click.argument("session_id")
705@click.option("--output", type=click.Choice(["json", "table"]), default="json", show_default=True)
706def swarm_status_cmd(session_id: str, output: str) -> None:
707 """One-call fleet rollup: rails, pool, child table, findings."""
708 from mission.swarm_runner import build_fleet_rollup # noqa: PLC0415
710 backend, session = _load_orchestrator_or_exit(session_id)
711 rollup = build_fleet_rollup(backend, session) # type: ignore[arg-type]
712 if output == "json":
713 _emit_json(rollup)
714 return
715 pool = rollup["pool"]
716 click.echo(f"Swarm {rollup['session_id']} status={rollup['status']}")
717 click.echo(
718 f"Pool: {pool['remaining']}/{pool['pool']} remaining "
719 f"(reserved {pool['reserved']}, consumed {pool['consumed']})"
720 )
721 click.echo(f"Runner: {rollup['runner_state'] or 'none'}")
722 click.echo("Children:")
723 for row in rollup["children"]:
724 verdict = row.get("final_verdict", "-")
725 click.echo(
726 f" {row['slot']:<24} {row['status']:<12} verdict={verdict} "
727 f"respawns={row['respawn_count']}"
728 )
729 for finding in rollup["findings"]:
730 click.echo(f"finding: {finding}")
733@swarm_cmd.command("abort")
734@click.argument("session_id")
735def swarm_abort_cmd(session_id: str) -> None:
736 """Terminate the orchestrator and abort every non-terminal child."""
737 from mission.swarm_runner import abort_swarm # noqa: PLC0415
738 from mission.types import TERMINAL_STATES # noqa: PLC0415
740 backend, session = _load_orchestrator_or_exit(session_id)
741 if session["status"] in TERMINAL_STATES:
742 _emit_error("session_terminal", {"session_id": session_id, "status": session["status"]})
743 raise SystemExit(1)
744 _emit_json(abort_swarm(backend, session)) # type: ignore[arg-type]
747@swarm_cmd.command("list")
748@click.option("--status", default=None, help="Filter by lifecycle status.")
749def swarm_list_cmd(status: str | None) -> None:
750 """List swarm (orchestrator) sessions on the configured backend."""
751 from mission.state import get_backend # noqa: PLC0415
752 from mission.swarm_runner import list_swarms # noqa: PLC0415
754 _emit_json({"swarms": list_swarms(get_backend(), status=status)})
757@swarm_cmd.command("scaffold-plan")
758@click.option("--directive", required=True)
759@click.option("--tool-allowlist", multiple=True, metavar="NAME")
760@click.option("--allow-all-tools", is_flag=True, default=False)
761@_swarm_options
762@click.option("--max-plan-children", "max_plan_children", type=int, default=None)
763@click.option("--use-sampling/--no-sampling", "use_sampling", default=None)
764@click.option("--retries", type=int, default=3, show_default=True)
765@click.option(
766 "--output-file",
767 type=click.Path(dir_okay=False, writable=True),
768 default=None,
769 help="Write the plan JSON here instead of stdout.",
770)
771def swarm_scaffold_plan_cmd(
772 directive: str,
773 tool_allowlist: tuple[str, ...],
774 allow_all_tools: bool,
775 max_children: int,
776 child_iteration_pool: int,
777 max_concurrent_children: int,
778 allow_overlapping_mutating_tools: bool,
779 max_plan_children: int | None,
780 use_sampling: bool | None,
781 retries: int,
782 output_file: str | None,
783) -> None:
784 """Draft a validated Swarm_Plan for review, without starting anything."""
785 scaffold = _scaffold_plan(
786 directive=directive,
787 swarm_config=_swarm_config_from_flags(
788 max_children,
789 child_iteration_pool,
790 max_concurrent_children,
791 allow_overlapping_mutating_tools,
792 ),
793 tool_allowlist=tool_allowlist,
794 allow_all_tools=allow_all_tools,
795 max_children=max_plan_children,
796 use_sampling=use_sampling,
797 retries=retries,
798 )
799 if output_file:
800 Path(output_file).write_text(json.dumps(scaffold["plan"], indent=2), encoding="utf-8")
801 _emit_json(
802 {
803 "written": output_file,
804 "children": len(scaffold["plan"]),
805 "sampling_path": scaffold["sampling_path"],
806 },
807 err=True,
808 )
809 return
810 _emit_json(scaffold)