Agentswarm/orchestrator/convergence.py

"""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(),
    }