"""Metrics collection and feedback loop for agent-task quality. Zero third-party dependencies; standard library only. This module imports no ``cloud.*`` module: it computes over plain record dicts so the collection logic is testable without a database, and so the report shape is identical whether the records came from the control plane (``server._metrics_report`` reading the attempt ledger), a test fixture, or a hand-built record list. Definitions (pinned here, not left to each caller): - A task counts toward a rate only when it reached a terminal status (``completed`` / ``merged`` / ``failed`` / ``merge-failed``). ``interrupted`` is deliberately NOT terminal: platform-interrupted tasks auto-resume as fresh runs, so counting them as failures would punish a platform event, not the agent. - First-pass green: terminal task whose ``attempts`` == 1 and whose status is ``completed`` / ``merged``. - Defect: a task with ``defects`` > 0, where ``defects`` is the count of failed CI/audit rounds or follow-up rounds for that task. Since audit gap 3 these numbers come from the ``task_attempts`` ledger (``cloud/attempt_ledger.py``: one row per real attempt, aggregated by the endpoint); a task with no ledger rows must still pass ``None`` -- and the rates then come back as ``None`` with ``*_known`` False. An unknown rate is never reported as 0: 0 would claim "no defects", which is a decision-grade lie. - Rework: terminal task with ``attempts`` > 1. Same None discipline. - Per-task cost reuses the #212 dual-column discipline from ``cloud.cost``: priced USD micros and unpriced actual tokens are reported in separate columns. ``total_usd_micros`` is ``None`` whenever any task's cost is incomplete (no cost rows, or unpriced actual tokens > 0); the priced partial sum is still shown, labelled ``priced_usd_micros_partial``, so nobody mistakes it for a total. Unpriced cost is never folded into 0. Feedback (``evaluate_feedback``) is deliberately a pure function returning action records, not an auto-applier: an action says WHAT should change and WHY, with its evidence attached; applying it to routing or to the audit standards is a separate, approved step (see ``docs/arch-conformance-metrics.md``). The rules are testable: - defect rate for a category above threshold -> ``tighten_check`` for that category's check; - first-pass rate below its floor -> ``route_with_extra_audit``; - a check whose violations are overturned on review above threshold -> ``recalibrate_check`` (review the threshold, do NOT auto-apply a new one -- a check that fires wrongly must not silently widen itself). Thresholds passed in are labelled by source in the report. The defaults in ``DEFAULT_THRESHOLDS`` are starting values for tests, NOT human-approved production thresholds: the report marks them ``thresholds_source: "default_unapproved"`` until a human approves a set, and a report built on unapproved thresholds is still usable for collection -- it is just not decision-grade on its own. Golden tasks: a report is ``decision_ready`` only when a golden set has been human-approved (``approved_by`` + ``approved_at`` present and at least one golden task listed). Until then the golden section reads ``status: "pending_human_approval"`` and the report says so plainly; the sample-only set that exists today must never be presented as decision-ready. This module never invents golden tasks. """ import json import time TERMINAL_STATUSES = frozenset( {"completed", "merged", "failed", "merge-failed"}) SUCCESS_STATUSES = frozenset({"completed", "merged"}) DEFAULT_THRESHOLDS = { "defect_rate": 0.20, "first_pass_rate_min": 0.50, "check_overturn_rate": 0.30, } def _rate(numerator, denominator): if denominator is None or denominator == 0: return None return numerator / denominator def compute_metrics(records, costs=None): """``records``: list of dicts with keys ``id``, ``status`` and optional ``attempts`` (int), ``defects`` (int), ``defect_category`` (str). ``costs``: dict task_id -> dict with ``priced_usd_micros`` (int|None), ``unpriced_actual_tokens`` (int), ``has_rows`` (bool). Returns the metrics dict.""" costs = costs or {} terminal = [r for r in records if r.get("status") in TERMINAL_STATUSES] known_attempts = [r for r in terminal if r.get("attempts") is not None] first_pass = [r for r in known_attempts if r["attempts"] == 1 and r.get("status") in SUCCESS_STATUSES] rework = [r for r in known_attempts if r["attempts"] > 1] known_defects = [r for r in terminal if r.get("defects") is not None] defective = [r for r in known_defects if r["defects"] > 0] defect_by_category = {} for r in defective: cat = r.get("defect_category") or "uncategorized" defect_by_category[cat] = defect_by_category.get(cat, 0) + 1 attempts_total = sum(r["attempts"] for r in known_attempts) per_task_cost = [] priced_partial = 0 cost_complete = True for r in terminal: c = costs.get(r["id"]) or {} priced = c.get("priced_usd_micros") unpriced = int(c.get("unpriced_actual_tokens") or 0) has_rows = bool(c.get("has_rows")) complete = has_rows and unpriced == 0 and priced is not None if not complete: cost_complete = False if priced is not None: priced_partial += int(priced) per_task_cost.append({ "task_id": r["id"], "priced_usd_micros": priced, "unpriced_actual_tokens": unpriced, "cost_complete": complete, }) if not terminal: cost_complete = False return { "tasks_total": len(records), "terminal_tasks": len(terminal), "first_pass_tasks": len(first_pass), "first_pass_rate": _rate(len(first_pass), len(known_attempts)), "first_pass_known": bool(known_attempts), "rework_tasks": len(rework), "rework_rate": _rate(len(rework), len(known_attempts)), "rework_known": bool(known_attempts), "avg_attempts": (attempts_total / len(known_attempts) if known_attempts else None), "defect_tasks": len(defective), "defect_rate": _rate(len(defective), len(known_defects)), "defect_known": bool(known_defects), "defect_by_category": defect_by_category, "per_task_cost": per_task_cost, # Dual-column total: None unless every terminal task's cost # is complete; the partial sum is shown separately, labelled. "priced_usd_micros_partial": priced_partial, "total_usd_micros": priced_partial if cost_complete else None, "cost_complete": cost_complete, } def evaluate_feedback(metrics, thresholds=None, check_stats=None): """Testable feedback rules. Returns a list of action dicts; each carries its evidence so a reviewer can audit why it fired.""" thresholds = thresholds or DEFAULT_THRESHOLDS actions = [] defect_rate = metrics.get("defect_rate") if metrics.get("defect_known") and defect_rate is not None: limit = thresholds.get("defect_rate") if limit is not None and defect_rate > limit: for cat, count in sorted( metrics.get("defect_by_category", {}).items()): actions.append({ "kind": "tighten_check", "target": cat, "reason": "defect_rate %.3f > threshold %.3f " "(%d defective tasks in category)" % (defect_rate, limit, count), "evidence": {"defect_rate": defect_rate, "threshold": limit, "category_tasks": count}, "auto_apply": False, }) first_pass = metrics.get("first_pass_rate") floor = thresholds.get("first_pass_rate_min") if first_pass is not None and floor is not None and first_pass < floor: actions.append({ "kind": "route_with_extra_audit", "target": "all", "reason": "first_pass_rate %.3f < floor %.3f" % (first_pass, floor), "evidence": {"first_pass_rate": first_pass, "floor": floor}, "auto_apply": False, }) overturn_limit = thresholds.get("check_overturn_rate") for stat in check_stats or []: violations = stat.get("violations") or 0 overturned = stat.get("overturned") or 0 if violations and overturn_limit is not None: rate = overturned / violations if rate > overturn_limit: actions.append({ "kind": "recalibrate_check", "target": stat.get("check_id"), "reason": "overturn_rate %.3f > threshold %.3f " "(%d/%d violations overturned on review)" % (rate, overturn_limit, overturned, violations), "evidence": {"overturn_rate": rate, "threshold": overturn_limit, "violations": violations, "overturned": overturned}, "auto_apply": False, }) return actions def build_report(metrics, golden=None, thresholds=None, thresholds_source="default_unapproved", check_stats=None, sampling_plan=None): """Assemble the JSON report document. ``golden``: None, or a dict with ``tasks`` (list of task ids), ``approved_by`` and ``approved_at``. Only a fully approved golden set makes the report decision-ready; anything else is pending. """ golden = golden or {} tasks = golden.get("tasks") or [] approved = bool(tasks) and bool(golden.get("approved_by")) \ and bool(golden.get("approved_at")) return { "generated_at": time.time(), "golden_set": { "status": "approved" if approved else "pending_human_approval", "tasks": list(tasks), "approved_by": golden.get("approved_by"), "approved_at": golden.get("approved_at"), }, "decision_ready": approved, "decision_note": ("golden set human-approved; metrics may inform " "decisions" if approved else "golden set NOT human-approved: this report " "collects metrics only and must not be used " "as a decision basis"), "metrics": metrics, "thresholds": dict(thresholds or DEFAULT_THRESHOLDS), "thresholds_source": thresholds_source, "feedback_actions": evaluate_feedback( metrics, thresholds=thresholds, check_stats=check_stats), # How the loop closes (see docs/arch-conformance-metrics.md): # feedback actions are consumed by (a) routing -- tasks in a # flagged category are routed with an extra audit gate, and # (b) audit standards -- a tighten/recalibrate action opens a # standards-review item; the standards document itself changes # only through the contract-change protocol # (arch_conformance.check_contract_version) with coordinator # approval, never by auto-applying an action record. "feedback_consumers": ["routing", "audit_standards"], # Audit-of-audit sampling (cloud/audit_sampling.py): when a # plan is supplied it is carried verbatim; otherwise the # report states sampling is pending rather than implying # the audit was re-examined. "sampling": sampling_plan or { "status": "pending", "note": "no audit-of-audit sample plan supplied; audit " "verdicts in this report have not been " "re-examined"}, } def render_report(report): """Canonical JSON rendering of a report document.""" return json.dumps(report, indent=2, sort_keys=True, default=str)