Coverage for src/keel/swarm.py: 100%

1"""Keel Swarm — Deterministic static dependency analysis & conflict clustering. 

2 

3Pure, stdlib-first dependency graph analysis for multi-agent parallel execution. 

4Partitions candidate issues into orthogonal (disjoint) Waves and independent Clusters, 

5enabling Direct Batch Landing for non-overlapping diff trees and adaptive merge 

6funneling with automated conflict recovery for dependent trees. 

7 

8All functions here are pure and deterministic — no subprocess, no network. 

9""" 

10 

11from __future__ import annotations 

12 

13import datetime 

14import fnmatch 

15import json 

16import posixpath 

17import re 

18from dataclasses import dataclass, field 

19from pathlib import Path 

20from typing import Any 

21 

22from .config import ProjectConfig 

23 

24#: Regex to capture backticked file paths or common file paths in issue text 

25_PATH_BACKTICK_RE = re.compile(r"`([a-zA-Z0-9_\-./]+\.[a-zA-Z0-9_\-]+)`") 

26_PATH_GENERAL_RE = re.compile( 

27 r"(?:^|[\s(\[])([a-zA-Z0-9_\-./]+/(?:[a-zA-Z0-9_\-./]+\.[a-zA-Z0-9_\-]+|[a-zA-Z0-9_\-]+/|\*))" 

28) 

29 

30 

31@dataclass(frozen=True) 

32class IssueScope: 

33 """Normalized predicted blast radius and role for a single backlog issue.""" 

34 

35 issue: int 

36 title: str = "" 

37 body: str = "" 

38 labels: tuple[str, ...] = () 

39 declared_files: tuple[str, ...] = () 

40 predicted_files: tuple[str, ...] = () 

41 role: str = "core" 

42 

43 def to_dict(self) -> dict[str, Any]: 

44 return { 

45 "issue": self.issue, 

46 "title": self.title, 

47 "role": self.role, 

48 "labels": list(self.labels), 

49 "declared_files": list(self.declared_files), 

50 "predicted_files": list(self.predicted_files), 

51 } 

52 

53 

54@dataclass(frozen=True) 

55class SwarmCluster: 

56 """A unit of work within a Swarm Wave consisting of one or more related issues.""" 

57 

58 cluster_id: str 

59 issues: tuple[int, ...] 

60 role: str 

61 combined_scope: tuple[str, ...] 

62 depends_on_issues: tuple[int, ...] = () 

63 

64 def to_dict(self) -> dict[str, Any]: 

65 return { 

66 "cluster_id": self.cluster_id, 

67 "issues": list(self.issues), 

68 "role": self.role, 

69 "combined_scope": list(self.combined_scope), 

70 "depends_on_issues": list(self.depends_on_issues), 

71 } 

72 

73 

74@dataclass(frozen=True) 

75class SwarmWave: 

76 """A parallel execution wave containing mutually independent or sequenced clusters.""" 

77 

78 wave_index: int 

79 mode: str # "orthogonal_parallel" or "sequential_dependent" 

80 eligible_direct_landing: bool 

81 clusters: tuple[SwarmCluster, ...] 

82 

83 def to_dict(self) -> dict[str, Any]: 

84 return { 

85 "wave_index": self.wave_index, 

86 "mode": self.mode, 

87 "eligible_direct_landing": self.eligible_direct_landing, 

88 "clusters": [c.to_dict() for c in self.clusters], 

89 } 

90 

91 

92@dataclass(frozen=True) 

93class SwarmPlan: 

94 """Complete deterministic partition and execution plan for a swarm run.""" 

95 

96 swarm_id: str 

97 total_issues: int 

98 waves: tuple[SwarmWave, ...] 

99 conflict_map: dict[int, tuple[int, ...]] = field(default_factory=dict) 

100 issue_scopes: dict[int, IssueScope] = field(default_factory=dict) 

101 

102 def to_dict(self) -> dict[str, Any]: 

103 return { 

104 "swarm_id": self.swarm_id, 

105 "total_issues": self.total_issues, 

106 "waves": [w.to_dict() for w in self.waves], 

107 "conflict_map": {str(k): list(v) for k, v in self.conflict_map.items()}, 

108 "issue_scopes": {str(k): v.to_dict() for k, v in self.issue_scopes.items()}, 

109 } 

110 

111 

112@dataclass(frozen=True) 

113class SwarmWorkerStatus: 

114 """Live state for an individual worker operating on a swarm cluster.""" 

115 

116 cluster_id: str 

117 issue: int 

118 role: str 

119 agent: str = "claude" 

120 model: str = "default" 

121 step: str = "s0" 

122 status: str = "queued" # queued, running, passed, failed, merged 

123 updated_at: str = "" 

124 details: str = "" 

125 

126 def to_dict(self) -> dict[str, Any]: 

127 return { 

128 "cluster_id": self.cluster_id, 

129 "issue": self.issue, 

130 "role": self.role, 

131 "agent": self.agent, 

132 "model": self.model, 

133 "step": self.step, 

134 "status": self.status, 

135 "updated_at": self.updated_at, 

136 "details": self.details, 

137 } 

138 

139 

140@dataclass(frozen=True) 

141class SwarmRunState: 

142 """State tracking for a live or completed swarm execution.""" 

143 

144 swarm_id: str 

145 total_workers: int 

146 active_wave: int = 1 

147 workers: tuple[SwarmWorkerStatus, ...] = () 

148 started_at: str = "" 

149 completed_at: str | None = None 

150 

151 def to_dict(self) -> dict[str, Any]: 

152 return { 

153 "swarm_id": self.swarm_id, 

154 "total_workers": self.total_workers, 

155 "active_wave": self.active_wave, 

156 "workers": [w.to_dict() for w in self.workers], 

157 "started_at": self.started_at, 

158 "completed_at": self.completed_at, 

159 } 

160 

161 

162@dataclass(frozen=True) 

163class SwarmRunResult: 

164 """Outcome summary for a complete or partial swarm execution.""" 

165 

166 swarm_id: str 

167 status: str # "success", "partial_failure", "failed" 

168 total_workers: int 

169 passed_count: int 

170 failed_count: int 

171 dry_run: bool 

172 wave_results: tuple[dict[str, Any], ...] = () 

173 

174 def to_dict(self) -> dict[str, Any]: 

175 return { 

176 "swarm_id": self.swarm_id, 

177 "status": self.status, 

178 "total_workers": self.total_workers, 

179 "passed_count": self.passed_count, 

180 "failed_count": self.failed_count, 

181 "dry_run": self.dry_run, 

182 "wave_results": list(self.wave_results), 

183 } 

184 

185 

186@dataclass(frozen=True) 

187class LandingDecision: 

188 """Evaluation of whether a wave can directly land or requires sequential funneling.""" 

189 

190 mode: str # "direct_batch" or "sequential_funnel" 

191 eligible: bool 

192 cluster_ids: tuple[str, ...] 

193 reason: str = "orthogonal_diffs" 

194 

195 def to_dict(self) -> dict[str, Any]: 

196 return { 

197 "mode": self.mode, 

198 "eligible": self.eligible, 

199 "cluster_ids": list(self.cluster_ids), 

200 "reason": self.reason, 

201 } 

202 

203 

204@dataclass(frozen=True) 

205class SwarmLandingResult: 

206 """Outcome report for landing a swarm wave.""" 

207 

208 swarm_id: str 

209 wave_index: int 

210 mode: str 

211 landed_clusters: tuple[str, ...] 

212 healed_clusters: tuple[str, ...] 

213 failed_clusters: tuple[str, ...] 

214 status: str # "success", "partial_failure", "failed" 

215 

216 def to_dict(self) -> dict[str, Any]: 

217 return { 

218 "swarm_id": self.swarm_id, 

219 "wave_index": self.wave_index, 

220 "mode": self.mode, 

221 "landed_clusters": list(self.landed_clusters), 

222 "healed_clusters": list(self.healed_clusters), 

223 "failed_clusters": list(self.failed_clusters), 

224 "status": self.status, 

225 } 

226 

227 

228def _normalize_path(p: str) -> str: 

229 cleaned = p.strip("`'\" \t\r\n.,;:()") 

230 cleaned = cleaned.replace("\\", "/").removeprefix("./").removeprefix("/") 

231 return posixpath.normpath(cleaned) if cleaned else "" 

232 

233 

234def extract_predicted_paths(text: str) -> list[str]: 

235 """Extract candidate file and directory paths from issue title or markdown body.""" 

236 found: set[str] = set() 

237 for match in _PATH_BACKTICK_RE.finditer(text): 

238 found.add(_normalize_path(match.group(1))) 

239 for match in _PATH_GENERAL_RE.finditer(text): 

240 found.add(_normalize_path(match.group(1))) 

241 found.discard("") 

242 return sorted(found) 

243 

244 

245def extract_issue_scope( 

246 issue: int, 

247 *, 

248 title: str = "", 

249 body: str = "", 

250 labels: list[str] | tuple[str, ...] | None = None, 

251 declared_files: list[str] | tuple[str, ...] | None = None, 

252 config: ProjectConfig | None = None, 

253) -> IssueScope: 

254 """Extract and normalize predicted files and roles for an issue.""" 

255 norm_labels = tuple(sorted(set(labels or ()))) 

256 norm_declared = tuple( 

257 sorted(set(_normalize_path(f) for f in (declared_files or ()) if _normalize_path(f))) 

258 ) 

259 

260 predicted = set(norm_declared) 

261 combined_text = f"{title}\n{body}" 

262 predicted.update(extract_predicted_paths(combined_text)) 

263 

264 # Resolve role from labels or config 

265 resolved_role = "core" 

266 for label in norm_labels: 

267 if label.startswith("role:"): 

268 resolved_role = label.removeprefix("role:") 

269 break 

270 if label.startswith("area:"): 

271 resolved_role = label.removeprefix("area:") 

272 break 

273 

274 if config and config.knobs.implementer_agents: 

275 if resolved_role not in config.knobs.implementer_agents: 

276 resolved_role = "core" if "core" in config.knobs.implementer_agents else resolved_role 

277 

278 # Default directory hints based on role or labels if no specific files found 

279 if not predicted: 

280 if "visual" in resolved_role or any("visual" in lbl for lbl in norm_labels): 

281 predicted.add("keel-visual/*") 

282 elif "docs" in resolved_role or any("docs" in lbl for lbl in norm_labels): 

283 predicted.add("docs/*") 

284 elif "website" in resolved_role or any("website" in lbl for lbl in norm_labels): 

285 predicted.add("website/*") 

286 elif "cli" in resolved_role or any("cli" in lbl for lbl in norm_labels): 

287 predicted.add("src/keel/cli.py") 

288 else: 

289 predicted.add(f"scope/issue-{issue}/*") 

290 

291 return IssueScope( 

292 issue=issue, 

293 title=title.strip(), 

294 body=body.strip(), 

295 labels=norm_labels, 

296 declared_files=norm_declared, 

297 predicted_files=tuple(sorted(predicted)), 

298 role=resolved_role, 

299 ) 

300 

301 

302def paths_intersect(path_a: str, path_b: str) -> bool: 

303 """True if path_a and path_b refer to the same file or overlapping glob/directory.""" 

304 a = _normalize_path(path_a) 

305 b = _normalize_path(path_b) 

306 if not a or not b: 

307 return False 

308 if a == b: 

309 return True 

310 if a == "*" or b == "*": 

311 return True 

312 # Directory prefix overlap 

313 a_dir = a if a.endswith("/") else a + "/" 

314 b_dir = b if b.endswith("/") else b + "/" 

315 if b.startswith(a_dir) or a.startswith(b_dir): 

316 return True 

317 # Glob matching 

318 if "*" in a and fnmatch.fnmatch(b, a): 

319 return True 

320 if "*" in b and fnmatch.fnmatch(a, b): 

321 return True 

322 return False 

323 

324 

325def scopes_intersect(scope_a: IssueScope, scope_b: IssueScope) -> tuple[str, ...]: 

326 """Return common overlapping paths between two issue scopes.""" 

327 overlaps: set[str] = set() 

328 for fa in scope_a.predicted_files: 

329 for fb in scope_b.predicted_files: 

330 if paths_intersect(fa, fb): 

331 overlaps.add(fa if len(fa) <= len(fb) else fb) 

332 return tuple(sorted(overlaps)) 

333 

334 

335def build_swarm_plan( 

336 issue_scopes: list[IssueScope] | tuple[IssueScope, ...], 

337 *, 

338 swarm_id: str | None = None, 

339 config: ProjectConfig | None = None, 

340) -> SwarmPlan: 

341 """Deterministically partition candidate issues into orthogonal Waves and Clusters.""" 

342 if not issue_scopes: 

343 now_id = swarm_id or ( 

344 "swarm-" + datetime.datetime.now(datetime.UTC).strftime("%Y%m%d-%H%M%S") 

345 ) 

346 return SwarmPlan( 

347 swarm_id=now_id, 

348 total_issues=0, 

349 waves=(), 

350 conflict_map={}, 

351 issue_scopes={}, 

352 ) 

353 

354 # Sort issues for determinism 

355 sorted_scopes = sorted(issue_scopes, key=lambda s: s.issue) 

356 scope_by_id = {s.issue: s for s in sorted_scopes} 

357 

358 # Compute pairwise conflict graph 

359 conflict_map: dict[int, list[int]] = {s.issue: [] for s in sorted_scopes} 

360 for i, sa in enumerate(sorted_scopes): 

361 for sb in sorted_scopes[i + 1 :]: 

362 if scopes_intersect(sa, sb): 

363 conflict_map[sa.issue].append(sb.issue) 

364 conflict_map[sb.issue].append(sa.issue) 

365 

366 frozen_conflicts: dict[int, tuple[int, ...]] = { 

367 k: tuple(sorted(v)) for k, v in conflict_map.items() 

368 } 

369 

370 # Greedy wave partitioning (independent set partitioning) 

371 remaining_issues = [s.issue for s in sorted_scopes] 

372 waves: list[SwarmWave] = [] 

373 wave_idx = 1 

374 

375 assigned_prior_issues: set[int] = set() 

376 

377 while remaining_issues: 

378 current_wave_issues: list[int] = [] 

379 current_wave_scope_set: set[str] = set() 

380 

381 for issue_num in list(remaining_issues): 

382 scope = scope_by_id[issue_num] 

383 # Check if this issue conflicts with anything already placed in current wave 

384 has_wave_conflict = False 

385 for placed_num in current_wave_issues: 

386 if issue_num in frozen_conflicts.get(placed_num, ()): 

387 has_wave_conflict = True 

388 break 

389 

390 if not has_wave_conflict: 

391 current_wave_issues.append(issue_num) 

392 current_wave_scope_set.update(scope.predicted_files) 

393 remaining_issues.remove(issue_num) 

394 

395 # Build clusters for this wave 

396 clusters: list[SwarmCluster] = [] 

397 for issue_num in current_wave_issues: 

398 sc = scope_by_id[issue_num] 

399 # Find dependencies on previously assigned waves 

400 deps = tuple( 

401 sorted( 

402 dep 

403 for dep in frozen_conflicts.get(issue_num, ()) 

404 if dep in assigned_prior_issues 

405 ) 

406 ) 

407 clusters.append( 

408 SwarmCluster( 

409 cluster_id=f"cluster-{wave_idx}-{issue_num}", 

410 issues=(issue_num,), 

411 role=sc.role, 

412 combined_scope=sc.predicted_files, 

413 depends_on_issues=deps, 

414 ) 

415 ) 

416 

417 assigned_prior_issues.update(current_wave_issues) 

418 

419 # Determine mode & direct landing eligibility 

420 is_orthogonal = len(current_wave_issues) > 0 

421 waves.append( 

422 SwarmWave( 

423 wave_index=wave_idx, 

424 mode="orthogonal_parallel" if is_orthogonal else "sequential_dependent", 

425 eligible_direct_landing=is_orthogonal, 

426 clusters=tuple(clusters), 

427 ) 

428 ) 

429 wave_idx += 1 

430 

431 now_id = swarm_id or ( 

432 "swarm-" + datetime.datetime.now(datetime.UTC).strftime("%Y%m%d-%H%M%S") 

433 ) 

434 return SwarmPlan( 

435 swarm_id=now_id, 

436 total_issues=len(sorted_scopes), 

437 waves=tuple(waves), 

438 conflict_map=frozen_conflicts, 

439 issue_scopes=scope_by_id, 

440 ) 

441 

442 

443def render_swarm_plan_text(plan: SwarmPlan) -> str: 

444 """Render human-readable tabular text summary of the SwarmPlan.""" 

445 lines: list[str] = [ 

446 f"keel swarm plan — {plan.swarm_id}", 

447 f" total issues : {plan.total_issues}", 

448 f" total waves : {len(plan.waves)}", 

449 "", 

450 ] 

451 for w in plan.waves: 

452 landing = ( 

453 "eligible for direct batch landing" 

454 if w.eligible_direct_landing 

455 else "sequential merge funnel" 

456 ) 

457 lines.append(f"Wave {w.wave_index} [{w.mode}] — {landing}:") 

458 for c in w.clusters: 

459 issues_str = ", ".join(f"#{i}" for i in c.issues) 

460 scope_prefix = c.combined_scope[:3] 

461 scope_str = ", ".join(scope_prefix) + ( 

462 "..." if len(c.combined_scope) > 3 else "" 

463 ) 

464 dep_str = ( 

465 f" (depends on: {', '.join(f'#{d}' for d in c.depends_on_issues)})" 

466 if c.depends_on_issues 

467 else "" 

468 ) 

469 lines.append( 

470 f" • Cluster {c.cluster_id}: {issues_str} [{c.role}] → {scope_str}{dep_str}" 

471 ) 

472 lines.append("") 

473 return "\n".join(lines).strip() 

474 

475 

476def render_swarm_plan_tree(plan: SwarmPlan) -> str: 

477 """Render a visual ASCII/Unicode DAG dependency tree of the SwarmPlan.""" 

478 if plan.total_issues == 0: 

479 return f"keel swarm plan — {plan.swarm_id} (0 issues)" 

480 

481 direct_count = sum(1 for w in plan.waves if w.eligible_direct_landing) 

482 hdr_a = f"│ 🐝 Keel Swarm Plan — {plan.swarm_id:<38} │" 

483 hdr_b = ( 

484 f"│ Issues: {plan.total_issues:<3} │ Waves: {len(plan.waves):<3} " 

485 f"│ Direct Landing Waves: {direct_count:<2} │" 

486 ) 

487 lines: list[str] = [ 

488 "╭" + "─" * 62 + "╮", 

489 hdr_a, 

490 hdr_b, 

491 "╰" + "─" * 62 + "╯", 

492 "", 

493 ] 

494 

495 for w in plan.waves: 

496 mode_icon = "⚡" if w.eligible_direct_landing else "⏳" 

497 landing_label = ( 

498 "Direct Batch Landing" if w.eligible_direct_landing else "Sequential Funnel" 

499 ) 

500 lines.append(f"{mode_icon} Wave {w.wave_index} [{w.mode}] — {landing_label}") 

501 

502 num_clusters = len(w.clusters) 

503 for i, c in enumerate(w.clusters): 

504 is_last_cluster = i == num_clusters - 1 

505 c_prefix = "└── " if is_last_cluster else "├── " 

506 c_indent = " " if is_last_cluster else "│ " 

507 

508 issues_str = ", ".join(f"#{num}" for num in c.issues) 

509 lines.append(f"{c_prefix}📦 Cluster {c.cluster_id} ({issues_str}) [{c.role}]") 

510 

511 scope_items = list(c.combined_scope) 

512 has_deps = bool(c.depends_on_issues) 

513 

514 scope_branch = "├── " if has_deps else "└── " 

515 lines.append( 

516 f"{c_indent}{scope_branch}Scope: {', '.join(scope_items[:3])}" 

517 f"{'...' if len(scope_items) > 3 else ''}" 

518 ) 

519 

520 if has_deps: 

521 dep_issues = ", ".join(f"#{d}" for d in c.depends_on_issues) 

522 lines.append(f"{c_indent}└── ⛓️ Depends on: {dep_issues}") 

523 lines.append("") 

524 

525 return "\n".join(lines).rstrip() 

526 

527 

528def render_swarm_status_dashboard(state: SwarmRunState | None) -> str: 

529 """Render a live terminal ASCII matrix status board of the swarm run.""" 

530 if state is None: 

531 return "keel swarm status — no active or recent swarm run found." 

532 

533 status_badges = { 

534 "queued": "[QUEUED ⏳]", 

535 "running": "[RUNNING ⚙️]", 

536 "passed": "[PASSED ✓]", 

537 "failed": "[FAILED ✗]", 

538 "merged": "[MERGED 🚢]", 

539 } 

540 

541 start_str = state.started_at[:19] if state.started_at else "pending" 

542 hdr_info = ( 

543 f"│ Active Wave: {state.active_wave:<3} │ Total Workers: {state.total_workers:<3} " 

544 f"│ Started: {start_str:<24} │" 

545 ) 

546 cols_hdr = ( 

547 f"│ {'Cluster':<16} │ {'Issue':<6} │ {'Role':<8} │ {'Step':<5} " 

548 f"│ {'Agent / Model':<16} │ {'Status':<10} │" 

549 ) 

550 lines = [ 

551 "╭" + "─" * 74 + "╮", 

552 f"│ 🐝 Keel Swarm Live Status — {state.swarm_id:<44} │", 

553 hdr_info, 

554 "├" + "─" * 74 + "┤", 

555 cols_hdr, 

556 "├" + "─" * 74 + "┤", 

557 ] 

558 

559 for w in state.workers: 

560 badge = status_badges.get(w.status, f"[{w.status.upper()}]") 

561 agent_str = f"{w.agent}:{w.model}"[:16] 

562 row_str = ( 

563 f"│ {w.cluster_id:<16} │ #{w.issue:<5} │ {w.role:<8} │ {w.step:<5} " 

564 f"│ {agent_str:<16} │ {badge:<10} │" 

565 ) 

566 lines.append(row_str) 

567 

568 lines.append("╰" + "─" * 74 + "╯") 

569 return "\n".join(lines) 

570 

571 

572def resolve_swarm_state_dir(root: str | Path = ".") -> Path: 

573 """Ensure and return the path to `.keel/state/swarm/` directory.""" 

574 p = Path(root) / ".keel" / "state" / "swarm" 

575 p.mkdir(parents=True, exist_ok=True) 

576 return p 

577 

578 

579def save_swarm_state(state: SwarmRunState, root: str | Path = ".") -> Path: 

580 """Persist a SwarmRunState JSON snapshot to `.keel/state/swarm/<swarm_id>.json`.""" 

581 state_dir = resolve_swarm_state_dir(root) 

582 file_path = state_dir / f"{state.swarm_id}.json" 

583 file_path.write_text(json.dumps(state.to_dict(), indent=2), encoding="utf-8") 

584 return file_path 

585 

586 

587def load_swarm_state(swarm_id: str, root: str | Path = ".") -> SwarmRunState | None: 

588 """Load a SwarmRunState from disk if present.""" 

589 state_dir = Path(root) / ".keel" / "state" / "swarm" 

590 file_path = state_dir / f"{swarm_id}.json" 

591 if not file_path.exists(): 

592 return None 

593 try: 

594 data = json.loads(file_path.read_text(encoding="utf-8")) 

595 workers = tuple( 

596 SwarmWorkerStatus( 

597 cluster_id=str(w.get("cluster_id", "")), 

598 issue=int(w.get("issue", 0)), 

599 role=str(w.get("role", "core")), 

600 agent=str(w.get("agent", "claude")), 

601 model=str(w.get("model", "default")), 

602 step=str(w.get("step", "s0")), 

603 status=str(w.get("status", "queued")), 

604 updated_at=str(w.get("updated_at", "")), 

605 details=str(w.get("details", "")), 

606 ) 

607 for w in data.get("workers", []) 

608 ) 

609 return SwarmRunState( 

610 swarm_id=str(data.get("swarm_id", swarm_id)), 

611 total_workers=int(data.get("total_workers", len(workers))), 

612 active_wave=int(data.get("active_wave", 1)), 

613 workers=workers, 

614 started_at=str(data.get("started_at", "")), 

615 completed_at=data.get("completed_at"), 

616 ) 

617 except (json.JSONDecodeError, ValueError, KeyError): 

618 return None 

619 

620 

621def update_worker_state( 

622 state: SwarmRunState, 

623 cluster_id: str, 

624 *, 

625 step: str | None = None, 

626 status: str | None = None, 

627 details: str | None = None, 

628) -> SwarmRunState: 

629 """Return a new SwarmRunState with the specified worker's fields updated.""" 

630 updated_workers = [] 

631 for w in state.workers: 

632 if w.cluster_id == cluster_id: 

633 updated_workers.append( 

634 SwarmWorkerStatus( 

635 cluster_id=w.cluster_id, 

636 issue=w.issue, 

637 role=w.role, 

638 agent=w.agent, 

639 model=w.model, 

640 step=step if step is not None else w.step, 

641 status=status if status is not None else w.status, 

642 updated_at=datetime.datetime.now(datetime.UTC).isoformat(), 

643 details=details if details is not None else w.details, 

644 ) 

645 ) 

646 else: 

647 updated_workers.append(w) 

648 

649 return SwarmRunState( 

650 swarm_id=state.swarm_id, 

651 total_workers=state.total_workers, 

652 active_wave=state.active_wave, 

653 workers=tuple(updated_workers), 

654 started_at=state.started_at, 

655 completed_at=state.completed_at, 

656 ) 

657 

658 

659def rebalance_swarm_plan(plan: SwarmPlan, failed_issue: int) -> SwarmPlan: 

660 """Dynamically recalculate a SwarmPlan when an issue fails during execution. 

661 

662 Any subsequent wave clusters that depended on ``failed_issue`` will have 

663 the failed dependency omitted, while independent disjoint clusters 

664 proceed without interruption. 

665 """ 

666 new_waves = [] 

667 for w in plan.waves: 

668 new_clusters = [] 

669 for c in w.clusters: 

670 if failed_issue in c.issues: 

671 continue 

672 new_clusters.append(c) 

673 if new_clusters: 

674 new_waves.append( 

675 SwarmWave( 

676 wave_index=w.wave_index, 

677 mode=w.mode, 

678 eligible_direct_landing=w.eligible_direct_landing, 

679 clusters=tuple(new_clusters), 

680 ) 

681 ) 

682 

683 return SwarmPlan( 

684 swarm_id=plan.swarm_id, 

685 total_issues=sum(len(c.issues) for w in new_waves for c in w.clusters), 

686 waves=tuple(new_waves), 

687 conflict_map=plan.conflict_map, 

688 issue_scopes=plan.issue_scopes, 

689 ) 

690 

691 

692def render_swarm_run_result(result: SwarmRunResult) -> str: 

693 """Render a human-readable text summary of a SwarmRunResult.""" 

694 status_icon = ( 

695 "✓" 

696 if result.status == "success" 

697 else ("⚠️" if result.status == "partial_failure" else "✗") 

698 ) 

699 lines = [ 

700 f"keel swarm run — {result.swarm_id}", 

701 f" status : {result.status} {status_icon}", 

702 f" total workers : {result.total_workers}", 

703 f" passed : {result.passed_count}", 

704 f" failed : {result.failed_count}", 

705 f" dry-run : {'true' if result.dry_run else 'false'}", 

706 f" total waves : {len(result.wave_results)}", 

707 ] 

708 return "\n".join(lines) 

709 

710 

711def evaluate_wave_landing_mode( 

712 wave: SwarmWave, 

713 pr_diff_map: dict[str, list[str] | tuple[str, ...]], 

714) -> LandingDecision: 

715 """Evaluate whether a wave can directly land or requires sequential funneling.""" 

716 cluster_ids = tuple(c.cluster_id for c in wave.clusters) 

717 if len(cluster_ids) <= 1: 

718 return LandingDecision( 

719 mode="direct_batch", 

720 eligible=True, 

721 cluster_ids=cluster_ids, 

722 reason="single_cluster", 

723 ) 

724 

725 # Check pairwise disjointness of actual diff files 

726 has_conflict = False 

727 for i, c1 in enumerate(wave.clusters): 

728 diff1 = pr_diff_map.get(c1.cluster_id, c1.combined_scope) 

729 for c2 in wave.clusters[i + 1 :]: 

730 diff2 = pr_diff_map.get(c2.cluster_id, c2.combined_scope) 

731 for f1 in diff1: 

732 for f2 in diff2: 

733 if paths_intersect(f1, f2): 

734 has_conflict = True 

735 break 

736 if has_conflict: 

737 break 

738 if has_conflict: 

739 break 

740 if has_conflict: 

741 break 

742 

743 if not has_conflict: 

744 return LandingDecision( 

745 mode="direct_batch", 

746 eligible=True,