Coverage for gco_mcp / mission / swarm_runner.py: 100.00%

1"""The swarm Child_Runner: concurrent child driving under one supervisor. 

2 

3This is the impure counterpart to :mod:`mission.swarm` (which holds every 

4pure rule). The runner owns one orchestrator session's fleet for the 

5lifetime of a drive call: 

6 

7* **Supervisor tools.** ``mission_spawn`` / ``children_status`` / 

8 ``child_abort`` exist only as entries in the dispatcher wrapper built 

9 by :meth:`SwarmRunner.wrap_dispatcher` — injected into the orchestrator 

10 engine, invisible to FastMCP, unreachable from child or standalone 

11 sessions. Spawn admission is entirely :func:`mission.swarm.validate_spawn`. 

12* **Concurrent drivers.** Each spawned child gets an asyncio task looping 

13 the child's :class:`~mission.engine.MissionEngine` to a terminal state, 

14 bounded by an ``asyncio.Semaphore(max_concurrent_children)``. 

15* **Single-writer registry discipline.** The orchestrator's engine saves 

16 the orchestrator session during its own iterations, so the runner never 

17 writes the registry mid-iteration: mutations accumulate in memory and 

18 flush at iteration boundaries (and at finalization) onto a freshly 

19 loaded copy. Child drivers write only their own child sessions. A crash 

20 between spawn and flush leaves an orphan child session; the 

21 startup reconciliation pass adopts any persisted child whose 

22 ``parent_session_id`` matches but is missing from the registry. 

23* **Heartbeats and the single-runner guard.** One 

24 :class:`~tools._task_status.TaskStatusWriter` record per swarm 

25 (``swarm-{session_id}``) plus one per slot 

26 (``swarm-{session_id}-{slot}``). The swarm record doubles as the 

27 advisory same-host lock: a ``running`` record under a live foreign PID 

28 refuses startup; a dead PID reads as orphaned and is taken over. 

29* **Terminal cascade.** Whatever terminal verdict the orchestrator's 

30 unchanged cascade produces, the runner cancels drivers, aborts every 

31 non-terminal child through the same status transition 

32 ``mission_abort`` performs, settles and refunds each slot, emits 

33 lifecycle audit, and only then finishes its heartbeat. 

34 

35Everything decision-shaped in here delegates to the pure module: 

36admission (:func:`~mission.swarm.validate_spawn`), pool arithmetic 

37(:func:`~mission.swarm.settle_entry` / :func:`~mission.swarm.respawn_entry`), 

38and the restart table (:func:`~mission.swarm.should_respawn`). 

39""" 

40 

41from __future__ import annotations 

42 

43import asyncio 

44import contextlib 

45import json 

46import os 

47import secrets 

48import sys 

49from collections.abc import Awaitable, Callable, Mapping 

50from datetime import UTC, datetime 

51from pathlib import Path 

52from typing import Any, cast 

53 

54# <pyflowchart-code-diagram> BEGIN - auto-inserted, do not edit 

55# Generated at (UTC): 2026-09-03T18:56:22Z 

56# Generated from Git commit: 37fd4384775eeebf18fea3e5e085cef9645077be 

57# Flowchart(s) generated from this file: 

58# * ``SwarmRunner.run_to_completion`` -> ``diagrams/code_diagrams/gco_mcp/mission/swarm_runner.SwarmRunner_run_to_completion.html`` 

59# (PNG: ``diagrams/code_diagrams/gco_mcp/mission/swarm_runner.SwarmRunner_run_to_completion.png``) 

60# Regenerate with ``SOURCE_DATE_EPOCH=<unix-seconds> GCO_DIAGRAM_SOURCE_COMMIT=<40-char-sha> python diagrams/generate.py --code-only``. 

61# <pyflowchart-code-diagram> END 

62 

63 

64# Match the package's path-injection pattern (see _engine_factory.py): 

65# gco_mcp/ modules import each other with gco_mcp/ itself on sys.path. 

66sys.path.insert(0, str(Path(__file__).resolve().parent.parent)) 

67 

68from tools._task_status import TaskStatusWriter, get_task # noqa: E402 

69 

70from mission import audit as mission_audit # noqa: E402 

71from mission import final_report # noqa: E402 

72from mission import swarm as swarm_rules # noqa: E402 

73from mission._engine_factory import EngineDependencies # noqa: E402 

74from mission.engine import MissionEngine, ObservationAugmenter # noqa: E402 

75from mission.types import ( # noqa: E402 

76 SCHEMA_VERSION, 

77 TERMINAL_STATES, 

78 ChildRegistryEntry, 

79 SessionState, 

80 SwarmConfig, 

81) 

82from mission.validation import MissionValidationError # noqa: E402 

83 

84__all__ = [ 

85 "DepsBuilder", 

86 "SwarmRunner", 

87 "SwarmRunnerBusyError", 

88 "abort_swarm", 

89 "build_children_snapshot", 

90 "build_fleet_rollup", 

91 "list_swarms", 

92] 

93 

94#: Async factory the runner calls once per engine it constructs. The CLI 

95#: binds this to ``build_engine_dependencies`` (live or ``--dry-run`` stub); 

96#: tests bind a stub. Receiving the session lets the builder resolve 

97#: sampling and sandbox wiring per session role. 

98DepsBuilder = Callable[[Mapping[str, Any]], Awaitable[EngineDependencies]] 

99 

100#: Optional async reviser for ``on_failure_with_revision`` respawns. Takes 

101#: the failed child's terminal session, returns replacement directive text 

102#: or ``None`` (fall back to the original directive). Wired by the 

103#: scaffolder layer; the runner treats it as advisory text supply only — 

104#: the respawn *decision* never consults it. 

105DirectiveReviser = Callable[[SessionState], Awaitable[str | None]] 

106 

107#: Seconds the orchestrator waits for observable fleet progress before 

108#: taking another iteration anyway. Bounds patience without removing it: 

109#: a wedged fleet still reaches the stagnation cascade, it just takes 

110#: ``stagnation_threshold`` windows to get there instead of spinning 

111#: through them in one event-loop turn. 

112DEFAULT_FLEET_PROGRESS_TIMEOUT = 30.0 

113 

114 

115class SwarmRunnerBusyError(RuntimeError): 

116 """Another live process already drives this swarm. 

117 

118 Carries the holding record's PID so operator surfaces can print an 

119 actionable refusal instead of a bare failure. 

120 """ 

121 

122 def __init__(self, swarm_session_id: str, holder_pid: int | None) -> None: 

123 self.swarm_session_id = swarm_session_id 

124 self.holder_pid = holder_pid 

125 super().__init__(f"swarm {swarm_session_id} is already driven by live pid {holder_pid}") 

126 

127 

128def _now_iso() -> str: 

129 return datetime.now(UTC).isoformat() 

130 

131 

132def build_children_snapshot( 

133 config: SwarmConfig, 

134 children: list[ChildRegistryEntry], 

135 load_child: Callable[[str], SessionState | None], 

136) -> dict[str, Any]: 

137 """Build the deterministic Children_Observation contribution. 

138 

139 Pure given its inputs: slot-ordered entries plus the aggregate 

140 metrics, built from the registry and whatever ``load_child`` returns. 

141 A child whose session cannot be loaded surfaces with the distinct 

142 ``"unreadable"`` status token rather than being omitted, so criteria 

143 over the fleet read unmet/inconclusive instead of falsely met. 

144 """ 

145 entries: list[dict[str, Any]] = [] 

146 counts = {"running": 0, "completed": 0, "failed": 0} 

147 for entry in sorted(children, key=lambda e: e["slot"]): 

148 child = load_child(entry["session_id"]) 

149 row: dict[str, Any] = { 

150 "slot": entry["slot"], 

151 "session_id": entry["session_id"], 

152 "respawn_count": entry["respawn_count"], 

153 } 

154 if child is None: 

155 row["status"] = "unreadable" 

156 counts["failed"] += 1 

157 else: 

158 status = str(child.get("status", "unreadable")) 

159 row["iterations_consumed"] = len(child.get("iterations", [])) 

160 final_verdict = child.get("final_verdict") 

161 if final_verdict is not None: 

162 row["final_verdict"] = final_verdict 

163 # Supervision-aware status: a slot whose session ended 

164 # unmet but whose restart policy still owes it a respawn is 

165 # "respawning", not "failed" — otherwise a fleet criterion 

166 # like ``children_failed >= 1`` fires in the window between 

167 # a child's terminal save and its replacement, and the 

168 # orchestrator completes out from under its own policy. 

169 # Deterministic: computed purely from the entry + the 

170 # persisted child status through the same restart table the 

171 # runner itself uses. 

172 if status == "completed": 

173 row["status"] = status 

174 counts["completed"] += 1 

175 elif status in ("terminated", "failed"): 

176 wants_respawn, _reason = swarm_rules.should_respawn(entry, status) 

177 if wants_respawn: 

178 row["status"] = "respawning" 

179 counts["running"] += 1 

180 else: 

181 row["status"] = status 

182 counts["failed"] += 1 

183 else: 

184 row["status"] = status 

185 counts["running"] += 1 

186 entries.append(row) 

187 balance = swarm_rules.compute_pool_balance(config["child_iteration_pool"], children) 

188 return { 

189 "children": entries, 

190 "metrics": { 

191 "children_total": len(entries), 

192 "children_running": counts["running"], 

193 "children_completed": counts["completed"], 

194 "children_failed": counts["failed"], 

195 "iteration_pool_remaining": balance["remaining"], 

196 }, 

197 } 

198 

199 

200class SwarmRunner: 

201 """Drives one orchestrator session's fleet to a terminal verdict.""" 

202 

203 def __init__( 

204 self, 

205 *, 

206 backend: Any, 

207 orchestrator_id: str, 

208 deps_builder: DepsBuilder, 

209 registered_tools: dict[str, Any], 

210 registered_tags: Mapping[str, set[str]], 

211 flag_lookup: dict[str, str] | None = None, 

212 revise_directive: DirectiveReviser | None = None, 

213 on_orchestrator_iteration: Callable[[Mapping[str, Any]], None] | None = None, 

214 fleet_progress_timeout: float = DEFAULT_FLEET_PROGRESS_TIMEOUT, 

215 ) -> None: 

216 session = backend.load_session(orchestrator_id) 

217 if session is None: 

218 raise MissionValidationError( 

219 "session_not_found", details={"session_id": orchestrator_id} 

220 ) 

221 if session.get("role") != "orchestrator" or "swarm" not in session: 

222 raise MissionValidationError( 

223 "validation_error", 

224 details={"field": "role", "reason": "not_an_orchestrator"}, 

225 ) 

226 self._backend = backend 

227 self._orchestrator_id = orchestrator_id 

228 self._deps_builder = deps_builder 

229 self._registered_tools = registered_tools 

230 self._registered_tags = registered_tags 

231 self._flag_lookup = flag_lookup 

232 self._revise_directive = revise_directive 

233 # Optional per-iteration observer (the CLI's JSON-line verdict 

234 # stream). Called after each orchestrator iteration with the 

235 # iteration record; exceptions are the caller's problem by 

236 # design — a broken observer should fail the drive loudly. 

237 self._on_orchestrator_iteration = on_orchestrator_iteration 

238 self._config: SwarmConfig = session["swarm"] 

239 self._registry: list[ChildRegistryEntry] = list(session.get("children", [])) 

240 self._allowlists: dict[str, list[str]] = {} 

241 self._semaphore = asyncio.Semaphore(self._config["max_concurrent_children"]) 

242 self._tasks: dict[str, asyncio.Task[None]] = {} 

243 # Fleet-progress signal. The orchestrator observes children 

244 # through their persisted sessions, so it must not out-run them: 

245 # a real child iteration suspends many times, and an orchestrator 

246 # that only yielded one event-loop turn per iteration would spend 

247 # its whole stagnation window watching an untouched "pending" 

248 # fleet and terminate a swarm that was working fine. Drivers bump 

249 # the tick on every observable change (iteration recorded, slot 

250 # settled) and set the event to wake a waiting orchestrator. 

251 self._fleet_progress_timeout = fleet_progress_timeout 

252 self._progress_ticks = 0 

253 self._progress_event = asyncio.Event() 

254 self._dirty = False 

255 self._swarm_writer: TaskStatusWriter | None = None 

256 self._child_writers: dict[str, TaskStatusWriter] = {} 

257 

258 # ------------------------------------------------------------------ # 

259 # Registry helpers (single-writer: this class, at await boundaries) 

260 # ------------------------------------------------------------------ # 

261 

262 def _entry_index(self, slot: str) -> int: 

263 for index, entry in enumerate(self._registry): 

264 if entry["slot"] == slot: 

265 return index 

266 raise KeyError(slot) 

267 

268 def _live_entries(self) -> list[ChildRegistryEntry]: 

269 return [entry for entry in self._registry if not entry.get("settled")] 

270 

271 def _flush_registry(self) -> None: 

272 """Persist the in-memory registry onto a freshly loaded session. 

273 

274 Loading fresh means the orchestrator engine's own most-recent 

275 save (iterations, status, report paths) is preserved and only 

276 the ``children`` key is replaced — the runner is the single 

277 writer of that key, the engine of everything else. 

278 """ 

279 if not self._dirty: 

280 return 

281 session = self._backend.load_session(self._orchestrator_id) 

282 if session is None: 

283 return 

284 session["children"] = list(self._registry) 

285 self._backend.save_session(session) 

286 self._dirty = False 

287 

288 def _reconcile_orphans(self) -> None: 

289 """Adopt persisted children missing from the registry. 

290 

291 A crash between a spawn's child-session write and the next 

292 registry flush leaves a child on disk that the registry never 

293 recorded. Adoption rebuilds a conservative entry (reservation = 

294 the child's own iteration cap, restart ``never``) so pool 

295 accounting stays honest across the crash. 

296 """ 

297 known_ids = {entry["session_id"] for entry in self._registry} 

298 for entry in self._registry: 

299 known_ids.update(entry.get("prior_session_ids", [])) 

300 # The backend's list filter supports ``status`` only and its 

301 # summaries carry no parent linkage, so this is deliberately 

302 # list-all followed by a load-and-verify per candidate — the 

303 # state protocol stays untouched. 

304 listed = self._backend.list_sessions() 

305 for summary in listed: 

306 child_id = summary.get("session_id") 

307 if not child_id or child_id in known_ids: 

308 continue 

309 child = self._backend.load_session(child_id) 

310 if child is None or child.get("parent_session_id") != self._orchestrator_id: 

311 continue 

312 slot = f"adopted-{child_id[-8:]}" 

313 adopted: ChildRegistryEntry = { 

314 "slot": slot, 

315 "session_id": child_id, 

316 "spawned_at": str(child.get("created_at", _now_iso())), 

317 "reserved_iterations": int(child["budget"]["max_iterations"]), 

318 "restart_policy": "never", 

319 "max_respawns": 0, 

320 "respawn_count": 0, 

321 "consumed_iterations": 0, 

322 } 

323 self._registry.append(adopted) 

324 self._dirty = True 

325 mission_audit.emit_child_lifecycle_event( 

326 self._orchestrator_id, child_id, slot, "spawned", reason="adopted_orphan" 

327 ) 

328 

329 # ------------------------------------------------------------------ # 

330 # Heartbeats and the single-runner guard 

331 # ------------------------------------------------------------------ # 

332 

333 def _swarm_task_id(self) -> str: 

334 return f"swarm-{self._orchestrator_id}" 

335 

336 def _child_task_id(self, slot: str) -> str: 

337 return f"swarm-{self._orchestrator_id}-{slot}" 

338 

339 def _acquire_guard(self) -> None: 

340 """Refuse startup when a live foreign process drives this swarm. 

341 

342 The probe is the swarm heartbeat record itself: ``get_task`` 

343 already performs PID-liveness orphan rewriting, so a record 

344 still reading ``state=running`` with ``is_alive`` under another 

345 PID means a genuinely live runner. A dead PID reads as orphaned 

346 and is taken over. Advisory and same-host by design — the scope 

347 of the disk-backed task channel. 

348 """ 

349 record = get_task(self._swarm_task_id()) 

350 if ( 

351 record is not None 

352 and record.get("state") == "running" 

353 and record.get("is_alive") 

354 and record.get("pid") not in (None, os.getpid()) 

355 ): 

356 raise SwarmRunnerBusyError(self._orchestrator_id, record.get("pid")) 

357 self._swarm_writer = TaskStatusWriter( 

358 self._swarm_task_id(), 

359 "swarm_run", 

360 [self._orchestrator_id], 

361 pid=os.getpid(), 

362 ) 

363 

364 def _child_writer(self, slot: str) -> TaskStatusWriter: 

365 writer = self._child_writers.get(slot) 

366 if writer is None: 

367 writer = TaskStatusWriter( 

368 self._child_task_id(slot), 

369 "swarm_child", 

370 [self._orchestrator_id, slot], 

371 pid=os.getpid(), 

372 ) 

373 self._child_writers[slot] = writer 

374 return writer 

375 

376 def _heartbeat(self, line: str) -> None: 

377 if self._swarm_writer is not None: 

378 self._swarm_writer.record_line(line, stream="stdout") 

379 

380 # ------------------------------------------------------------------ # 

381 # Supervisor tools 

382 # ------------------------------------------------------------------ # 

383 

384 def wrap_dispatcher(self, inner: Any) -> Any: 

385 """Route supervisor names in-process; everything else falls through.""" 

386 

387 async def dispatch(tool_name: str, args: dict[str, Any], ctx: Any) -> Any: 

388 if tool_name == "mission_spawn": 

389 return await self.spawn(args) 

390 if tool_name == "children_status": 

391 return self.children_status() 

392 if tool_name == "child_abort": 

393 return await self.abort_child(str(args.get("slot", ""))) 

394 return await inner(tool_name, args, ctx) 

395 

396 return dispatch 

397 

398 def observation_augmenter(self) -> ObservationAugmenter: 

399 """The Children_Observation contribution for the orchestrator engine.""" 

400 

401 def augment(session: SessionState) -> dict[str, Any]: 

402 del session # snapshot reads the runner's authoritative registry 

403 return self.children_status() 

404 

405 return augment 

406 

407 def children_status(self) -> dict[str, Any]: 

408 """Deterministic fleet snapshot (tool result and augmenter payload).""" 

409 return build_children_snapshot(self._config, self._registry, self._backend.load_session) 

410 

411 async def spawn( 

412 self, request: Mapping[str, Any], *, respawn_of_slot: str | None = None 

413 ) -> dict[str, Any]: 

414 """Admit, persist, register, and schedule one child. 

415 

416 Returns the spawn result envelope on success or the standard 

417 ``{"code", "details"}`` envelope on rejection — the orchestrator 

418 iteration continues either way, and a sampled strategy sees the 

419 precise rejection reason in its next revision prompt. 

420 """ 

421 sibling_allowlists = { 

422 entry["slot"]: self._allowlists.get(entry["slot"], []) for entry in self._live_entries() 

423 } 

424 try: 

425 spec = swarm_rules.validate_spawn( 

426 parent_role="orchestrator", 

427 config=self._config, 

428 children=self._registry, 

429 request=request, 

430 registered_tools=self._registered_tools, 

431 registered_tags=self._registered_tags, 

432 sibling_allowlists=sibling_allowlists, 

433 flag_lookup=self._flag_lookup, 

434 respawn_of_slot=respawn_of_slot, 

435 ) 

436 except MissionValidationError as err: 

437 if respawn_of_slot is not None: 

438 mission_audit.emit_child_lifecycle_event( 

439 self._orchestrator_id, 

440 None, 

441 respawn_of_slot, 

442 "respawn_denied", 

443 reason=str((err.details or {}).get("reason", err.code)), 

444 ) 

445 return {"code": err.code, "details": err.details} 

446 

447 child_id = f"mission-{secrets.token_hex(8)}" 

448 child: dict[str, Any] = { 

449 "version": SCHEMA_VERSION, 

450 "session_id": child_id, 

451 "directive_text": spec["directive"], 

452 "criteria": _strip_parsed_asts(spec["criteria"]), 

453 "budget": spec["budget"], 

454 "tool_allowlist": spec["tool_allowlist"], 

455 "checkpoint_cadence": spec["checkpoint_cadence"], 

456 "stagnation_threshold": 3, 

457 "use_sampling": spec["use_sampling"], 

458 "sampling_backend_resolved": "none" if not spec["use_sampling"] else "bedrock", 

459 "allow_scripted_strategies": False, 

460 "status": "pending", 

461 "created_at": _now_iso(), 

462 "iterations": [], 

463 "no_progress_counter": 0, 

464 "role": "child", 

465 "parent_session_id": self._orchestrator_id, 

466 } 

467 self._backend.save_session(cast("SessionState", child)) 

468 

469 if respawn_of_slot is None: 

470 entry = swarm_rules.new_registry_entry(spec, child_id, _now_iso()) 

471 self._registry.append(entry) 

472 action = "spawned" 

473 else: 

474 index = self._entry_index(respawn_of_slot) 

475 entry = swarm_rules.respawn_entry( 

476 self._registry[index], 

477 new_session_id=child_id, 

478 reserved_iterations=spec["budget"]["max_iterations"], 

479 spawned_at=_now_iso(), 

480 ) 

481 self._registry[index] = entry 

482 action = "respawned" 

483 self._allowlists[spec["slot"]] = list(spec["tool_allowlist"]) 

484 self._dirty = True 

485 mission_audit.emit_child_lifecycle_event( 

486 self._orchestrator_id, child_id, spec["slot"], action 

487 ) 

488 self._schedule_child(spec["slot"]) 

489 balance = swarm_rules.compute_pool_balance( 

490 self._config["child_iteration_pool"], self._registry 

491 ) 

492 return { 

493 "spawned": True, 

494 "slot": spec["slot"], 

495 "child_session_id": child_id, 

496 "pool_remaining": balance["remaining"], 

497 } 

498 

499 async def abort_child(self, slot: str) -> dict[str, Any]: 

500 """Abort one live slot: cancel its driver, terminate, settle.""" 

501 try: 

502 index = self._entry_index(slot) 

503 except KeyError: 

504 return { 

505 "code": "validation_error", 

506 "details": {"field": "spawn", "reason": "unknown_slot", "slot": slot}, 

507 } 

508 task = self._tasks.get(slot) 

509 if task is not None and not task.done(): 

510 task.cancel() 

511 with contextlib.suppress(asyncio.CancelledError): 

512 await task 

513 entry = self._registry[index] 

514 if not entry.get("settled"): 

515 consumed = self._terminate_child_session(entry["session_id"]) 

516 self._registry[index] = swarm_rules.settle_entry(entry, consumed) 

517 self._dirty = True 

518 mission_audit.emit_child_lifecycle_event( 

519 self._orchestrator_id, entry["session_id"], slot, "aborted" 

520 ) 

521 return {"aborted": True, "slot": slot} 

522 

523 def _terminate_child_session(self, child_id: str) -> int: 

524 """Apply the ``mission_abort`` terminal transition to a child. 

525 

526 Returns the child's recorded iteration count for settlement; a 

527 missing session settles at zero consumption (full refund) since 

528 nothing demonstrably ran. 

529 """ 

530 child = self._backend.load_session(child_id) 

531 if child is None: 

532 return 0 

533 if child["status"] not in TERMINAL_STATES: 

534 child["status"] = "terminated" 

535 child["final_verdict"] = "terminate" 

536 child["ended_at"] = _now_iso() 

537 self._backend.save_session(child) 

538 return len(child.get("iterations", [])) 

539 

540 # ------------------------------------------------------------------ # 

541 # Child drivers 

542 # ------------------------------------------------------------------ # 

543 

544 def _note_fleet_progress(self) -> None: 

545 """Record observable child progress and wake a waiting orchestrator.""" 

546 self._progress_ticks += 1 

547 self._progress_event.set() 

548 

549 async def _await_fleet_progress(self, observed_ticks: int) -> None: 

550 """Yield until the fleet changes under the orchestrator's feet. 

551 

552 ``observed_ticks`` is the tick count captured *before* the 

553 orchestrator iteration that just ran, so progress that landed 

554 while that iteration was in flight counts and costs no wait. 

555 

556 With no live children there is nothing to wait for: yield one 

557 turn and let the orchestrator's own criteria and cascade decide. 

558 A wedged fleet falls through on timeout, which is the honest 

559 outcome — the orchestrator waited and nothing moved. 

560 """ 

561 if not self._live_entries() or self._progress_ticks != observed_ticks: 

562 await asyncio.sleep(0) 

563 return 

564 if not any(not task.done() for task in self._tasks.values()): 

565 # Live slots but no running driver (e.g. a cancelled driver 

566 # left a slot unsettled): nothing can produce progress, so 

567 # waiting would only stall the cascade. 

568 await asyncio.sleep(0) 

569 return 

570 # No await between the tick check and the clear, so no driver can 

571 # interleave and have its signal dropped (single-threaded loop). 

572 self._progress_event.clear() 

573 with contextlib.suppress(TimeoutError): 

574 await asyncio.wait_for( 

575 self._progress_event.wait(), timeout=self._fleet_progress_timeout 

576 ) 

577 

578 def _schedule_child(self, slot: str) -> None: 

579 existing = self._tasks.get(slot) 

580 if existing is not None and not existing.done(): 

581 return 

582 self._tasks[slot] = asyncio.get_running_loop().create_task(self._drive_child(slot)) 

583 

584 async def _drive_child(self, slot: str) -> None: 

585 writer = self._child_writer(slot) 

586 entry = self._registry[self._entry_index(slot)] 

587 child_id = entry["session_id"] 

588 try: 

589 child = self._backend.load_session(child_id) 

590 if child is None: 

591 self._settle_slot(slot, consumed=entry["reserved_iterations"], status="failed") 

592 writer.finish(state="failed", error="child session unreadable") 

593 return 

594 engine = await self._build_child_engine(child) 

595 while True: 

596 current = self._backend.load_session(child_id) 

597 if current is None or current["status"] in TERMINAL_STATES: 

598 break 

599 async with self._semaphore: 

600 record = await engine.run_iteration(child_id) 

601 writer.record_line( 

602 f"slot={slot} verdict={record.get('verdict')}" 

603 f" reason={record.get('verdict_reason')}", 

604 stream="stdout", 

605 ) 

606 self._note_fleet_progress() 

607 # No explicit yield here: when the iteration that just ran 

608 # was the child's last, the next loop pass must reach the 

609 # settle + restart-policy step without the orchestrator 

610 # observing the raw terminal session in between. Real 

611 # engine work yields at its own await points; the 

612 # orchestrator loop carries the fairness yield. 

613 final = self._backend.load_session(child_id) 

614 final_status = str(final["status"]) if final is not None else "failed" 

615 consumed = len(final.get("iterations", [])) if final is not None else 0 

616 self._settle_slot(slot, consumed=consumed, status=final_status) 

617 writer.finish(state="succeeded" if final_status == "completed" else "failed") 

618 await self._maybe_respawn(slot, final_status, final) 

619 except asyncio.CancelledError: 

620 writer.finish(state="cancelled") 

621 raise 

622 except Exception as exc: # noqa: BLE001 — a driver bug must not kill the swarm 

623 # Persist the standard abort transition before settling the slot. 

624 consumed = self._terminate_child_session(child_id) 

625 self._settle_slot(slot, consumed=consumed, status="failed") 

626 writer.finish(state="failed", error=str(exc)) 

627 

628 def _settle_slot(self, slot: str, *, consumed: int, status: str) -> None: 

629 index = self._entry_index(slot) 

630 entry = self._registry[index] 

631 if entry.get("settled"): 

632 return 

633 self._registry[index] = swarm_rules.settle_entry(entry, consumed) 

634 self._dirty = True 

635 self._note_fleet_progress() 

636 mission_audit.emit_child_lifecycle_event( 

637 self._orchestrator_id, 

638 entry["session_id"], 

639 slot, 

640 "terminal", 

641 final_status=status, 

642 ) 

643 

644 async def _maybe_respawn( 

645 self, slot: str, final_status: str, final_session: SessionState | None 

646 ) -> None: 

647 entry = self._registry[self._entry_index(slot)] 

648 decision, reason = swarm_rules.should_respawn(entry, final_status) 

649 if not decision: 

650 return 

651 directive = str(final_session.get("directive_text", "")) if final_session else "" 

652 if ( 

653 entry["restart_policy"] == "on_failure_with_revision" 

654 and self._revise_directive is not None 

655 and final_session is not None 

656 ): 

657 with contextlib.suppress(Exception): 

658 revised = await self._revise_directive(final_session) 

659 if revised: 

660 directive = revised 

661 request: dict[str, Any] = { 

662 "slot": slot, 

663 "directive": directive, 

664 "criteria": _strip_parsed_asts(list(final_session.get("criteria", []))) 

665 if final_session 

666 else [], 

667 "budget": dict(final_session["budget"]) if final_session else {}, 

668 "tool_allowlist": list(final_session.get("tool_allowlist", [])) 

669 if final_session 

670 else [], 

671 "restart_policy": entry["restart_policy"], 

672 "max_respawns": entry["max_respawns"], 

673 "use_sampling": bool(final_session.get("use_sampling", False)) 

674 if final_session 

675 else False, 

676 } 

677 await self.spawn(request, respawn_of_slot=slot) 

678 

679 async def _build_child_engine(self, child: SessionState) -> MissionEngine: 

680 deps = await self._deps_builder(child) 

681 return MissionEngine( 

682 backend=self._backend, 

683 tool_dispatcher=deps.tool_dispatcher, 

684 sampling_callable=deps.sampling_callable, 

685 sandbox_runner=deps.sandbox_runner, 

686 final_lessons_callable=deps.final_lessons_callable, 

687 memory_store=deps.memory_store, 

688 ) 

689 

690 async def _build_orchestrator_engine(self, session: SessionState) -> MissionEngine: 

691 deps = await self._deps_builder(session) 

692 return MissionEngine( 

693 backend=self._backend, 

694 tool_dispatcher=self.wrap_dispatcher(deps.tool_dispatcher), 

695 sampling_callable=deps.sampling_callable, 

696 sandbox_runner=deps.sandbox_runner, 

697 final_lessons_callable=deps.final_lessons_callable, 

698 memory_store=deps.memory_store, 

699 observation_augmenters=[self.observation_augmenter()], 

700 ) 

701 

702 # ------------------------------------------------------------------ # 

703 # Main loop 

704 # ------------------------------------------------------------------ # 

705 

706 def _load_orchestrator_session(self) -> SessionState: 

707 """Load the orchestrator or report concurrent deletion consistently.""" 

708 session = self._backend.load_session(self._orchestrator_id) 

709 if session is None: 

710 raise MissionValidationError( 

711 "session_not_found", 

712 details={"session_id": self._orchestrator_id}, 

713 ) 

714 return cast("SessionState", session) 

715 

716 async def run_to_completion( 

717 self, *, max_orchestrator_iterations: int | None = None 

718 ) -> SessionState: 

719 """Drive orchestrator and fleet until the orchestrator is terminal. 

720 

721 Also the resume path: on startup every live registry slot gets a 

722 driver scheduled, and children that went terminal while 

723 unsupervised get their restart policy evaluated on settlement. 

724 

725 ``max_orchestrator_iterations`` bounds one call's orchestrator 

726 iterations (the ``mission_iterate`` shape). Hitting the bound 

727 **detaches** rather than terminates: drivers are cancelled, 

728 children stay non-terminal and resumable, no abort cascade runs, 

729 and the swarm heartbeat finishes ``cancelled`` so the next 

730 runner's guard sees a released fleet. The abort cascade runs 

731 only on a genuinely terminal (or already-terminal) orchestrator. 

732 """ 

733 self._acquire_guard() 

734 try: 

735 self._reconcile_orphans() 

736 self._flush_registry() 

737 session = self._load_orchestrator_session() 

738 engine = await self._build_orchestrator_engine(session) 

739 for entry in self._live_entries(): 

740 self._schedule_child(entry["slot"]) 

741 terminal = False 

742 ran = 0 

743 observed_ticks = self._progress_ticks 

744 while True: 

745 current = self._load_orchestrator_session() 

746 if current["status"] in TERMINAL_STATES: 

747 terminal = True 

748 break 

749 if current["status"] == "paused": 

750 break 

751 if max_orchestrator_iterations is not None and ran >= max_orchestrator_iterations: 

752 break 

753 # Gate *before* the next iteration, never after the last 

754 # one: every exit above must leave without paying the 

755 # wait, or bounded (``mission_iterate``-shaped) calls and 

756 # terminal cascades would stall on a fleet nobody is 

757 # waiting for. The first iteration is the orchestrator's 

758 # initial assessment and never waits. 

759 if ran > 0: 

760 await self._await_fleet_progress(observed_ticks) 

761 observed_ticks = self._progress_ticks 

762 record = await engine.run_iteration(self._orchestrator_id) 

763 ran += 1 

764 self._flush_registry() 

765 self._heartbeat( 

766 f"iteration verdict={record.get('verdict')}" 

767 f" reason={record.get('verdict_reason')}" 

768 ) 

769 if self._on_orchestrator_iteration is not None: 

770 self._on_orchestrator_iteration(record) 

771 if record.get("verdict") in ("complete", "terminate"): 

772 terminal = True 

773 break 

774 if terminal: 

775 await self._cascade_shutdown() 

776 self._flush_registry() 

777 final = self._load_orchestrator_session() 

778 self._refresh_report_children(final) 

779 if self._swarm_writer is not None: 

780 succeeded = final.get("final_verdict") == "complete" 

781 self._swarm_writer.finish(state="succeeded" if succeeded else "failed") 

782 return final 

783 # Detached (iteration bound or pause): leave the fleet 

784 # resumable and release the guard record. 

785 self._flush_registry() 

786 detached = self._load_orchestrator_session() 

787 if self._swarm_writer is not None: 

788 self._swarm_writer.finish(state="cancelled") 

789 return detached 

790 finally: 

791 for task in self._tasks.values(): 

792 if not task.done(): 

793 task.cancel() 

794 if self._tasks: 

795 await asyncio.gather(*self._tasks.values(), return_exceptions=True) 

796 self._flush_registry() 

797 

798 def _refresh_report_children(self, final: SessionState) -> None: 

799 """Rewrite the report's per-child table with post-cascade states. 

800 

801 The engine writes the Final_Report during terminal finalization, 

802 which runs *before* the abort cascade settles the last registry 

803 entries. Refreshing the ``swarm_children`` table afterwards makes 

804 the durable artifact reflect final supervision outcomes.