"""Swarm convergence protocol — consensus, conflict resolution, and a termination function. Issue #12: today a swarm run reaches a terminal state in `orchestrator/main.py:refresh_swarm_run_status` purely by task bookkeeping — "all known tasks reached a terminal state, and (optionally, behind ENABLE_REVIEW_LOOP) the master critic accepted the work". There is NO explicit convergence model: no consensus score, no first-class conflict detection / resolution pass, and no machine-readable `termination_reason` that explains WHY the swarm stopped. A reader of the run cannot tell "quality reached" apart from "budget exhausted" apart from "blocked by an unresolved risk". This module adds that missing model as a PURE, side-effect-free layer: * `evaluate_convergence(run_state) -> ConvergenceReport` derives a run status, a single explanatory `termination_reason`, a consensus score, and the detected / resolved / unresolved conflicts from plain inputs (task states, review verdicts, budget, risks). It performs NO I/O and never mutates the run — the caller decides whether/when to persist or emit. * Conflict detectors for artifact mismatch, test failure, review disagreement, and dependency inconsistency, plus a `resolve_conflicts` pass that marks each conflict resolved or unresolved. * Event-payload builders matching the SwarmRuntime emit_event convention (Manager `event_type` + plain dict payload). UNCONDITIONAL: `evaluate_convergence` is wired into `refresh_swarm_run_status` and runs on every terminal swarm run (no enable flag — this repo is the swarm runtime). It produces the report + `termination_reason` stored on the run and surfaced on `timeline.updated`. Honesty (org rule #9): the report is currently EXPLANATORY — it derives `termination_reason` / consensus / conflicts from real run inputs but does NOT override `run.status` (today next_status from task bookkeeping and the report agree on completed/failed). Authoritative status-override is a documented follow-on. Code is English; the companion design doc is Simplified Chinese (docs/ style). """ from __future__ import annotations from dataclasses import dataclass, field from enum import Enum from typing import Any, Dict, List, Optional # --- enums ---------------------------------------------------------------- class TerminationReason(str, Enum): """Why the swarm stopped converging. Exactly one is attached to every terminal ConvergenceReport. `tasks_completed` is the honest fallback that mirrors today's only real termination signal ("all tasks done + review passed"); the others require their respective inputs (quality grade, budget state, round counter, blocking risk) to be present, and are advisory until those inputs are wired in. """ QUALITY_REACHED = "quality_reached" # acceptance/quality gate satisfied BUDGET_EXHAUSTED = "budget_exhausted" # token / cost / duration budget spent MAX_ROUNDS_REACHED = "max_rounds_reached" # review/redo cycle cap hit RISK_BLOCKED = "risk_blocked" # an unresolved blocking risk stopped the run TASKS_COMPLETED = "tasks_completed" # fallback: all tasks terminal, nothing else to explain class ConflictType(str, Enum): """Kinds of inter-agent disagreement the swarm can detect.""" ARTIFACT_MISMATCH = "artifact_mismatch" # two tasks wrote conflicting content to one path TEST_FAILURE = "test_failure" # a task's reported tests did not pass REVIEW_DISAGREEMENT = "review_disagreement" # master critic rejected accepted-looking work DEPENDENCY_INCONSISTENCY = "dependency_inconsistency" # task completed before/without its dependency class ConvergenceStatus(str, Enum): """Run-level convergence verdict. Maps onto SwarmRun.status: CONVERGED->completed, FAILED->failed, BLOCKED->blocked, RUNNING->running (not yet terminal). """ RUNNING = "running" CONVERGED = "converged" FAILED = "failed" BLOCKED = "blocked" # --- data structures ------------------------------------------------------ @dataclass class Conflict: """One detected disagreement. `resolved` is set by resolve_conflicts.""" conflict_id: str type: ConflictType description: str task_ids: List[str] = field(default_factory=list) detail: Dict[str, Any] = field(default_factory=dict) resolved: bool = False resolution: Optional[str] = None # how it was resolved, or why it could not be def to_dict(self) -> Dict[str, Any]: return { "conflict_id": self.conflict_id, "type": self.type.value, "description": self.description, "task_ids": list(self.task_ids), "detail": dict(self.detail), "resolved": self.resolved, "resolution": self.resolution, } @dataclass class ConvergenceReport: """Machine-readable explanation of why (and whether) a swarm converged.""" status: ConvergenceStatus termination_reason: Optional[TerminationReason] # None while still RUNNING consensus_score: float # [0,100]; share of work in agreement conflicts: List[Conflict] = field(default_factory=list) resolved_conflicts: List[Conflict] = field(default_factory=list) unresolved_risks: List[Dict[str, Any]] = field(default_factory=list) budget_state: Dict[str, Any] = field(default_factory=dict) quality_state: Dict[str, Any] = field(default_factory=dict) def is_terminal(self) -> bool: return self.status in { ConvergenceStatus.CONVERGED, ConvergenceStatus.FAILED, ConvergenceStatus.BLOCKED, } def to_dict(self) -> Dict[str, Any]: return { "status": self.status.value, "termination_reason": ( self.termination_reason.value if self.termination_reason else None ), "consensus_score": self.consensus_score, "conflicts": [c.to_dict() for c in self.conflicts], "resolved_conflicts": [c.to_dict() for c in self.resolved_conflicts], "unresolved_risks": [dict(r) for r in self.unresolved_risks], "budget_state": dict(self.budget_state), "quality_state": dict(self.quality_state), } # --- terminal status constants (avoid importing TaskStatus to keep this pure) --- _TASK_TERMINAL_OK = {"completed"} _TASK_TERMINAL_FAIL = {"failed"} _TASK_ACTIVE = {"pending", "assigned", "in_progress"} _TASK_BLOCKED = {"blocked"} def _task_status(task: Any) -> str: """Normalize a task's status to a plain lowercase string. Accepts a plain dict (test inputs), a pydantic Task, or an enum value. """ if isinstance(task, dict): status = task.get("status") else: status = getattr(task, "status", None) value = getattr(status, "value", status) return str(value or "").lower() def _task_id(task: Any) -> str: if isinstance(task, dict): return str(task.get("task_id") or "") return str(getattr(task, "task_id", "") or "") def _task_result(task: Any) -> Dict[str, Any]: """Pull a structured result dict off a task (dict or model). Empty if absent.""" if isinstance(task, dict): result = task.get("result") else: result = getattr(task, "result", None) if isinstance(result, dict): return result return {} # --- conflict detectors --------------------------------------------------- def detect_artifact_mismatch(tasks: List[Any]) -> List[Conflict]: """Two completed tasks claim to have written different content to the same path. Looks at each task result's `files_modified` (paths) and the optional per-path `file_contents` / `file_hashes` map. If two tasks touch the same path with a differing recorded content/hash, that is an artifact mismatch. """ conflicts: List[Conflict] = [] # path -> list of (task_id, fingerprint) by_path: Dict[str, List[tuple]] = {} for task in tasks: if _task_status(task) not in _TASK_TERMINAL_OK: continue result = _task_result(task) tid = _task_id(task) contents = result.get("file_contents") or {} hashes = result.get("file_hashes") or {} for path in result.get("files_modified") or []: fingerprint = contents.get(path, hashes.get(path)) by_path.setdefault(path, []).append((tid, fingerprint)) for path, entries in by_path.items(): if len(entries) < 2: continue fingerprints = {fp for _, fp in entries if fp is not None} if len(fingerprints) > 1: ids = [tid for tid, _ in entries] conflicts.append(Conflict( conflict_id=f"conf-artifact-{path}", type=ConflictType.ARTIFACT_MISMATCH, description=f"Conflicting writes to {path} by {', '.join(ids)}", task_ids=ids, detail={"path": path, "fingerprints": sorted(str(fp) for fp in fingerprints)}, )) return conflicts def detect_test_failures(tasks: List[Any]) -> List[Conflict]: """A task reported a test result that did not pass. Reads `tests_passed` (bool) or `test_pass_rate` (0..1 or 0..100) from the task result. No test signal at all is NOT a conflict (rule #9: absence is not failure). """ conflicts: List[Conflict] = [] for task in tasks: result = _task_result(task) tid = _task_id(task) passed = result.get("tests_passed") rate = result.get("test_pass_rate") failing = False if passed is False: failing = True elif isinstance(rate, (int, float)): normalized = rate if rate <= 1 else rate / 100.0 failing = normalized < 1.0 if failing: conflicts.append(Conflict( conflict_id=f"conf-test-{tid}", type=ConflictType.TEST_FAILURE, description=f"Task {tid} reported failing tests", task_ids=[tid], detail={"tests_passed": passed, "test_pass_rate": rate}, )) return conflicts def detect_review_disagreement( tasks: List[Any], review_verdict: Optional[Dict[str, Any]], ) -> List[Conflict]: """The master critic rejected work that otherwise looks complete. `review_verdict` is the shape produced by master_agent.review_and_decide: {accepted: bool, retry_tasks: [..], summary: str}. A rejection is a review-disagreement conflict over the tasks the critic flagged for redo. """ if not review_verdict or review_verdict.get("accepted", True): return [] retry_tasks = list(review_verdict.get("retry_tasks") or []) if not retry_tasks: # Rejected but nothing actionable flagged — still a (run-wide) disagreement. retry_tasks = [_task_id(t) for t in tasks if _task_status(t) in _TASK_TERMINAL_OK] return [Conflict( conflict_id="conf-review", type=ConflictType.REVIEW_DISAGREEMENT, description=review_verdict.get("summary") or "Master critic rejected the work", task_ids=retry_tasks, detail={"summary": review_verdict.get("summary")}, )] def detect_dependency_inconsistency(tasks: List[Any]) -> List[Conflict]: """A task completed while a declared dependency did not complete. Reads `depends_on` from each task and checks the dependency's terminal status within the same run. A completed task whose dependency failed / is still active is an inconsistency. """ by_id = {_task_id(t): t for t in tasks} conflicts: List[Conflict] = [] for task in tasks: if _task_status(task) not in _TASK_TERMINAL_OK: continue tid = _task_id(task) depends_on = task.get("depends_on") if isinstance(task, dict) else getattr(task, "depends_on", None) for dep_id in depends_on or []: dep = by_id.get(dep_id) if dep is None: continue # dependency not in this run's task set; not our call to judge if _task_status(dep) not in _TASK_TERMINAL_OK: conflicts.append(Conflict( conflict_id=f"conf-dep-{tid}-{dep_id}", type=ConflictType.DEPENDENCY_INCONSISTENCY, description=f"Task {tid} completed but dependency {dep_id} is {_task_status(dep)}", task_ids=[tid, dep_id], detail={"task_id": tid, "dependency_id": dep_id, "dependency_status": _task_status(dep)}, )) return conflicts def detect_conflicts( tasks: List[Any], review_verdict: Optional[Dict[str, Any]] = None, ) -> List[Conflict]: """Run every detector and return the combined conflict list.""" conflicts: List[Conflict] = [] conflicts.extend(detect_artifact_mismatch(tasks)) conflicts.extend(detect_test_failures(tasks)) conflicts.extend(detect_review_disagreement(tasks, review_verdict)) conflicts.extend(detect_dependency_inconsistency(tasks)) return conflicts # --- conflict resolution -------------------------------------------------- def resolve_conflicts( conflicts: List[Conflict], review_verdict: Optional[Dict[str, Any]] = None, ) -> tuple[List[Conflict], List[Conflict]]: """Mark each conflict resolved or unresolved with a simple, explainable pass. This is a deliberately conservative first pass, not an auto-merge engine: * ARTIFACT_MISMATCH: resolved only if the review verdict accepted the work (the critic implicitly picked a winning version); otherwise unresolved. * REVIEW_DISAGREEMENT: resolved if the verdict names actionable `retry_tasks` (the swarm CAN act on it by reopening them); a blanket rejection with nothing actionable stays unresolved. * TEST_FAILURE / DEPENDENCY_INCONSISTENCY: treated as hard blockers — not auto-resolvable here; left unresolved for a human / redo cycle. Returns (resolved, unresolved). Each input Conflict is mutated in place to carry its `resolved` flag and `resolution` note. """ resolved: List[Conflict] = [] unresolved: List[Conflict] = [] accepted = bool(review_verdict and review_verdict.get("accepted")) actionable = bool(review_verdict and review_verdict.get("retry_tasks")) for conflict in conflicts: if conflict.type == ConflictType.ARTIFACT_MISMATCH and accepted: conflict.resolved = True conflict.resolution = "Master review accepted a winning version" elif conflict.type == ConflictType.REVIEW_DISAGREEMENT and actionable: conflict.resolved = True conflict.resolution = "Reopened flagged tasks for a redo cycle" else: conflict.resolved = False conflict.resolution = "No automatic resolution; requires redo or human review" (resolved if conflict.resolved else unresolved).append(conflict) return resolved, unresolved # --- budget / quality / risk derivation ----------------------------------- def derive_budget_state(run_state: Dict[str, Any]) -> Dict[str, Any]: """Summarize budget consumption from the run state inputs (no I/O). Recognized inputs under `run_state["budget"]`: max_tokens/token_limit, max_cost_usd, max_duration_seconds/duration_seconds and under `run_state["usage"]`: total_tokens, total_cost_usd, elapsed_seconds. `exhausted` is True if any present limit is met or exceeded. """ budget = run_state.get("budget") or {} usage = run_state.get("usage") or {} max_tokens = budget.get("max_tokens") or budget.get("token_limit") max_cost = budget.get("max_cost_usd") max_duration = budget.get("max_duration_seconds") or budget.get("duration_seconds") used_tokens = usage.get("total_tokens") used_cost = usage.get("total_cost_usd") elapsed = usage.get("elapsed_seconds") def _exhausted(limit, used): return ( isinstance(limit, (int, float)) and limit > 0 and isinstance(used, (int, float)) and used >= limit ) exhausted = ( _exhausted(max_tokens, used_tokens) or _exhausted(max_cost, used_cost) or _exhausted(max_duration, elapsed) ) return { "max_tokens": max_tokens, "used_tokens": used_tokens, "max_cost_usd": max_cost, "used_cost_usd": used_cost, "max_duration_seconds": max_duration, "elapsed_seconds": elapsed, "exhausted": exhausted, } def derive_quality_state(run_state: Dict[str, Any]) -> Dict[str, Any]: """Summarize the run quality gate from inputs (no I/O). `run_state["quality"]` mirrors SwarmRun.quality (Group B fixture grade): {test_pass_rate, graded, ...}. `reached` is True when a grade is present and meets `acceptance_threshold` (default 1.0 == all fixture tests pass). Absent grade -> reached False, graded False (rule #9: not graded != passed). """ quality = run_state.get("quality") or {} threshold = run_state.get("acceptance_threshold", 1.0) rate = quality.get("test_pass_rate") graded = quality.get("graded", rate is not None) reached = False if isinstance(rate, (int, float)): normalized = rate if rate <= 1 else rate / 100.0 reached = normalized >= threshold return { "graded": bool(graded), "test_pass_rate": rate, "acceptance_threshold": threshold, "reached": reached, } def derive_unresolved_risks( run_state: Dict[str, Any], unresolved_conflicts: List[Conflict], ) -> List[Dict[str, Any]]: """Collect blocking risks: explicit input risks + unresolved hard conflicts. `run_state["risks"]` is a list of {id, description, blocking: bool}. Any risk with blocking=True, plus every unresolved TEST_FAILURE / DEPENDENCY_INCONSISTENCY / unresolved REVIEW_DISAGREEMENT, surfaces here. These drive RISK_BLOCKED. """ risks: List[Dict[str, Any]] = [] for risk in run_state.get("risks") or []: if isinstance(risk, dict) and risk.get("blocking"): risks.append({ "id": risk.get("id"), "description": risk.get("description"), "source": "input_risk", }) for conflict in unresolved_conflicts: risks.append({ "id": conflict.conflict_id, "description": conflict.description, "source": f"conflict:{conflict.type.value}", }) return risks def compute_consensus_score(tasks: List[Any], conflicts: List[Conflict]) -> float: """Share of completed work that is NOT entangled in a conflict, as 0..100. consensus = 100 * (completed_tasks_without_conflict / completed_tasks). No completed tasks -> 0.0 (nothing has been agreed yet). This is a concrete, explainable proxy for "how much of the swarm's output is in agreement", derived only from real task states and detected conflicts. """ completed = [t for t in tasks if _task_status(t) in _TASK_TERMINAL_OK] if not completed: return 0.0 conflicted_ids = set() for conflict in conflicts: conflicted_ids.update(conflict.task_ids) agreeing = [t for t in completed if _task_id(t) not in conflicted_ids] return round(100.0 * len(agreeing) / len(completed), 2) # --- the termination function --------------------------------------------- def evaluate_convergence(run_state: Dict[str, Any]) -> ConvergenceReport: """Derive a ConvergenceReport from a run-state snapshot (pure, no I/O). `run_state` is a plain dict (the caller assembles it from SwarmRun + tasks): tasks: list of tasks (dicts or models) with status/result/depends_on review_verdict: optional master critic verdict budget: optional budget limits usage: optional consumed usage quality: optional Group B fixture grade risks: optional list of blocking risks review_cycles: int cycles already run max_review_cycles: int cap acceptance_threshold:optional quality threshold (default 1.0) Decision order (first match wins) for terminal runs: 1. Any active/blocked task -> RUNNING (no termination_reason yet) 2. Any blocking risk / unresolved hard conflict -> BLOCKED, RISK_BLOCKED 3. Any failed task -> FAILED, BUDGET_EXHAUSTED if budget spent else MAX_ROUNDS_REACHED if the redo cap was hit else TASKS_COMPLETED 4. Budget exhausted (all tasks done) -> CONVERGED, BUDGET_EXHAUSTED 5. Review cap hit -> CONVERGED, MAX_ROUNDS_REACHED 6. Quality gate reached -> CONVERGED, QUALITY_REACHED 7. Otherwise (all tasks done) -> CONVERGED, TASKS_COMPLETED (the honest fallback == today's only real signal) Every terminal report carries exactly one non-None termination_reason. """ tasks = list(run_state.get("tasks") or []) review_verdict = run_state.get("review_verdict") conflicts = detect_conflicts(tasks, review_verdict) resolved, unresolved = resolve_conflicts(conflicts, review_verdict) consensus_score = compute_consensus_score(tasks, conflicts) budget_state = derive_budget_state(run_state) quality_state = derive_quality_state(run_state) unresolved_risks = derive_unresolved_risks(run_state, unresolved) def _report(status: ConvergenceStatus, reason: Optional[TerminationReason]) -> ConvergenceReport: return ConvergenceReport( status=status, termination_reason=reason, consensus_score=consensus_score, conflicts=conflicts, resolved_conflicts=resolved, unresolved_risks=unresolved_risks, budget_state=budget_state, quality_state=quality_state, ) statuses = [_task_status(t) for t in tasks] # 1. still running — not terminal yet. if not tasks or any(s in _TASK_ACTIVE for s in statuses): return _report(ConvergenceStatus.RUNNING, None) # A blocked task with no unresolved risk is still "in flight" (mirrors # refresh_swarm_run_status keeping blocked-with-active-child runs running). # 2. blocking risk / unresolved hard conflict -> blocked. if unresolved_risks: return _report(ConvergenceStatus.BLOCKED, TerminationReason.RISK_BLOCKED) if any(s in _TASK_BLOCKED for s in statuses): return _report(ConvergenceStatus.RUNNING, None) cycles = int(run_state.get("review_cycles", 0) or 0) max_cycles = int(run_state.get("max_review_cycles", 0) or 0) rounds_hit = max_cycles > 0 and cycles >= max_cycles # 3. a failed task -> failed run, with the best available explanation. if any(s in _TASK_TERMINAL_FAIL for s in statuses): if budget_state.get("exhausted"): reason = TerminationReason.BUDGET_EXHAUSTED elif rounds_hit: reason = TerminationReason.MAX_ROUNDS_REACHED else: reason = TerminationReason.TASKS_COMPLETED return _report(ConvergenceStatus.FAILED, reason) # All tasks completed. Explain WHY it stopped, most-specific reason first. # 4. budget exhausted. if budget_state.get("exhausted"): return _report(ConvergenceStatus.CONVERGED, TerminationReason.BUDGET_EXHAUSTED) # 5. review/redo cap hit. if rounds_hit: return _report(ConvergenceStatus.CONVERGED, TerminationReason.MAX_ROUNDS_REACHED) # 6. explicit quality gate satisfied. if quality_state.get("reached"): return _report(ConvergenceStatus.CONVERGED, TerminationReason.QUALITY_REACHED) # 7. honest fallback: nothing else to explain beyond "all tasks done". return _report(ConvergenceStatus.CONVERGED, TerminationReason.TASKS_COMPLETED) # --- event payload builders ----------------------------------------------- # Match SwarmRuntime.emit_event(run, event_type, payload=...) convention: each # builder returns (event_type, payload) so a caller can splat it. Payloads are # plain dicts of JSON-safe primitives. These event types are NOT yet in the # Manager event registry — see the doc's integration notes before subscribing. def event_convergence_started(run_state: Dict[str, Any]) -> tuple[str, Dict[str, Any]]: """convergence.started — emitted once when the convergence pass begins.""" return "convergence.started", { "summary": "Convergence evaluation started", "task_count": len(run_state.get("tasks") or []), } def event_conflict_detected(conflict: Conflict) -> tuple[str, Dict[str, Any]]: """conflict.detected — one per detected conflict.""" return "conflict.detected", { "summary": conflict.description, **conflict.to_dict(), } def event_conflict_resolved(conflict: Conflict) -> tuple[str, Dict[str, Any]]: """conflict.resolved — one per conflict that resolve_conflicts closed.""" return "conflict.resolved", { "summary": conflict.resolution or "Conflict resolved", **conflict.to_dict(), } def event_consensus_updated(report: ConvergenceReport) -> tuple[str, Dict[str, Any]]: """consensus.updated — current consensus score and conflict tallies.""" return "consensus.updated", { "summary": f"Consensus {report.consensus_score}%", "consensus_score": report.consensus_score, "conflicts_total": len(report.conflicts), "conflicts_resolved": len(report.resolved_conflicts), "conflicts_unresolved": len(report.unresolved_risks), } def event_convergence_reached(report: ConvergenceReport) -> tuple[str, Dict[str, Any]]: """convergence.reached — terminal success with its termination_reason.""" return "convergence.reached", { "summary": "Swarm converged", **report.to_dict(), } def event_convergence_failed(report: ConvergenceReport) -> tuple[str, Dict[str, Any]]: """convergence.failed — terminal failure/block with its termination_reason.""" return "convergence.failed", { "summary": f"Swarm did not converge: {report.status.value}", **report.to_dict(), }