FastheiandClaude Opus 4.8
1edf73aba4
真机端到端实测(#70)暴露三问题,本 PR 全部修复(仅 agent + orchestrator,不跨仓): 问题1【阻断】单任务超时只有 60s,生成类任务必挂 - agent/main.py: TASK_TIMEOUT_SECONDS 默认 60→300(仅对外部/独立启动 agent 生效)。 - agent_launcher.py: 新增 DEFAULT_TASK_TIMEOUT_SECONDS=300、_budget_duration_seconds、 resolve_task_timeout(base=env 默认 300,与 run budget.duration_seconds 取较小); plan_launch_specs 把 TASK_TIMEOUT_SECONDS 透传进每个 agent env(非敏感,inline, k8s 不进 Secret)。 问题2【体验】事件时间线全是内部噪音(纯附加,未碰冻结契约) - swarm_runtime.py: is_client_visible(=event_type∈FROZEN_CLIENT_EVENT_TYPES,单一真源); emit_event 给 envelope 加 metadata.client_visible 布尔 + 关键客户端事件回填可选 payload.message(人话进度,仅取已有字段,不伪造)。task.heartbeat/retried/ deployment.status_changed/timeline/budget 标 client_visible=false,仍持久化+回调 但客户端据此过滤出时间线。冻结事件集/类型/sequence/artifact 形状一字未动。 - event-schema.md: 文档化两个附加字段 + 新增 §6.1,明确未解冻。 问题3【正确性】失败/超时 termination_reason 仍报 "tasks_completed" - convergence.py: 新增 TIMEOUT/MAX_RETRIES_EXCEEDED/TASK_FAILED;classify_failure_reason 按 timeout→max_retries→task_failed 取最具体(仅凭真实 per-task 信号);FAILED 分支 再不会返回 tasks_completed(该 reason 仅用于成功),budget/rounds 仅在通用失败时才覆盖。 - task_queue.py: fail_task 永久失败时把 reason 落到 task.result({"success":false,"error":reason}), 不覆盖已有结果,供 convergence 读取。 - main.py: compute_convergence_report 快照补 retry_count/max_retries。 测试:新增 test_resolve_task_timeout、扩 test-convergence(failed_timeout/max_retries/ generic + "FAILED 永不报 tasks_completed"不变量)。本地全过:test-agent-launcher / test-convergence / test-runtime-contract / test-contract-freeze / test-merge-smoke / test-workflow-e2e / test-security-boundary。 影响:agent + orchestrator + 文档;不动 Manager↔Swarm 冻结契约字段(问题2 纯附加)。 栈在 #64(agent_swarm git 注入)之上,#64 合并后本 PR base 自动转 main。 Closes #70 Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
683 lines
29 KiB
Python
683 lines
29 KiB
Python
"""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
|
|
TIMEOUT = "timeout" # a task exhausted its execution timeout
|
|
MAX_RETRIES_EXCEEDED = "max_retries_exceeded" # a task failed after its retry budget
|
|
TASK_FAILED = "task_failed" # a task failed for another (non-timeout) reason
|
|
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 {}
|
|
|
|
|
|
def _task_field(task: Any, key: str) -> Any:
|
|
"""Read an arbitrary field off a task (dict or model)."""
|
|
if isinstance(task, dict):
|
|
return task.get(key)
|
|
return getattr(task, key, None)
|
|
|
|
|
|
def _is_timeout_failure(task: Any) -> bool:
|
|
"""Whether a failed task's recorded reason indicates an execution timeout.
|
|
|
|
The agent reports a timeout as ``result={"error": "timeout", ...}`` (agent/main.py); the
|
|
orchestrator persists that reason on the failed task. We match the canonical "timeout" token
|
|
(case-insensitive) in the task's result ``error``/``reason`` so we never guess.
|
|
"""
|
|
result = _task_result(task)
|
|
text = " ".join(
|
|
str(result.get(k) or "") for k in ("error", "reason", "summary")
|
|
).lower()
|
|
return "timeout" in text
|
|
|
|
|
|
def _is_max_retries_failure(task: Any) -> bool:
|
|
"""Whether a failed task exhausted its retry budget (retry_count >= max_retries)."""
|
|
try:
|
|
retry_count = int(_task_field(task, "retry_count") or 0)
|
|
max_retries = int(_task_field(task, "max_retries") or 0)
|
|
except (TypeError, ValueError):
|
|
return False
|
|
return max_retries > 0 and retry_count >= max_retries
|
|
|
|
|
|
def classify_failure_reason(tasks: List[Any]) -> TerminationReason:
|
|
"""Pick the most-specific termination reason for a FAILED run from its failed tasks.
|
|
|
|
Precedence (most specific first), derived only from real per-task signals (never fabricated):
|
|
1. ``timeout`` — any failed task whose recorded error is a timeout.
|
|
2. ``max_retries_exceeded`` — any failed task that exhausted its retry budget.
|
|
3. ``task_failed`` — a task failed for another reason (the honest generic fail).
|
|
Returns ``TASK_FAILED`` when no failed task is present (caller should only invoke this on a
|
|
FAILED run, but we stay safe).
|
|
"""
|
|
failed = [t for t in tasks if _task_status(t) in _TASK_TERMINAL_FAIL]
|
|
if any(_is_timeout_failure(t) for t in failed):
|
|
return TerminationReason.TIMEOUT
|
|
if any(_is_max_retries_failure(t) for t in failed):
|
|
return TerminationReason.MAX_RETRIES_EXCEEDED
|
|
return TerminationReason.TASK_FAILED
|
|
|
|
|
|
# --- 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, with the most accurate reason: TIMEOUT if a
|
|
failed task timed out, else MAX_RETRIES_EXCEEDED if a
|
|
failed task exhausted its retry budget, else
|
|
BUDGET_EXHAUSTED if budget spent, else
|
|
MAX_ROUNDS_REACHED if the redo cap was hit, else
|
|
TASK_FAILED (never TASKS_COMPLETED — that is a
|
|
success-only reason; see agent_swarm#70)
|
|
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. A FAILED run must NEVER
|
|
# report `tasks_completed` (that reason is success-only and was the agent_swarm#70 bug):
|
|
# classify the failure (timeout / max_retries_exceeded / task_failed) and let a more specific
|
|
# resource cause (budget / rounds) win only when no concrete failure signal is present.
|
|
if any(s in _TASK_TERMINAL_FAIL for s in statuses):
|
|
failure_reason = classify_failure_reason(tasks)
|
|
if failure_reason is TerminationReason.TASK_FAILED and budget_state.get("exhausted"):
|
|
reason = TerminationReason.BUDGET_EXHAUSTED
|
|
elif failure_reason is TerminationReason.TASK_FAILED and rounds_hit:
|
|
reason = TerminationReason.MAX_ROUNDS_REACHED
|
|
else:
|
|
reason = failure_reason
|
|
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(),
|
|
}
|