codex-app-server/cloud/tests/test_metrics_loop.py

"""Tests for cloud.metrics: collection, dual-column cost, feedback.

Every metric has a positive case and an unknown/edge case; unknown
is never reported as zero, and an unapproved golden set never makes
a report decision-ready.

Only the standard library; no real model, no network, no secrets.
"""

import os
import sys
import unittest

sys.path.insert(0, os.path.dirname(os.path.dirname(
    os.path.dirname(os.path.abspath(__file__)))))

from cloud.arch_conformance import check_contract_version  # noqa: E402
from cloud.metrics import (  # noqa: E402
    build_report, compute_metrics, evaluate_feedback)


def _record(task_id, status, attempts=None, defects=None, category=None):
    return {"id": task_id, "status": status, "attempts": attempts,
            "defects": defects, "defect_category": category}


class TestComputeMetrics(unittest.TestCase):
    def test_first_pass_rework_defect_rates(self):
        records = [
            _record("a", "completed", attempts=1, defects=0),
            _record("b", "merged", attempts=3, defects=2,
                    category="contract"),
            _record("c", "failed", attempts=2, defects=1,
                    category="contract"),
            _record("d", "queued", attempts=1),  # not terminal: excluded
            _record("e", "interrupted", attempts=1),  # not terminal
        ]
        m = compute_metrics(records)
        self.assertEqual(m["terminal_tasks"], 3)
        self.assertAlmostEqual(m["first_pass_rate"], 1 / 3)
        self.assertAlmostEqual(m["rework_rate"], 2 / 3)
        self.assertAlmostEqual(m["defect_rate"], 2 / 3)
        self.assertEqual(m["defect_by_category"], {"contract": 2})

    def test_unknown_attempts_and_defects_are_none_not_zero(self):
        records = [_record("a", "completed"),
                   _record("b", "failed")]
        m = compute_metrics(records)
        self.assertIsNone(m["first_pass_rate"])
        self.assertIsNone(m["rework_rate"])
        self.assertIsNone(m["defect_rate"])
        self.assertFalse(m["first_pass_known"])
        self.assertFalse(m["rework_known"])
        self.assertFalse(m["defect_known"])

    def test_empty_set_rates_are_none(self):
        m = compute_metrics([])
        self.assertEqual(m["terminal_tasks"], 0)
        self.assertIsNone(m["first_pass_rate"])
        self.assertIsNone(m["defect_rate"])
        self.assertIsNone(m["total_usd_micros"])
        self.assertFalse(m["cost_complete"])


class TestDualColumnCost(unittest.TestCase):
    def test_complete_cost_totals(self):
        records = [_record("a", "completed", attempts=1)]
        costs = {"a": {"priced_usd_micros": 5000,
                        "unpriced_actual_tokens": 0, "has_rows": True}}
        m = compute_metrics(records, costs)
        self.assertEqual(m["total_usd_micros"], 5000)
        self.assertTrue(m["cost_complete"])

    def test_unpriced_cost_never_folds_into_zero_or_total(self):
        records = [_record("a", "completed", attempts=1)]
        costs = {"a": {"priced_usd_micros": 5000,
                        "unpriced_actual_tokens": 700, "has_rows": True}}
        m = compute_metrics(records, costs)
        self.assertIsNone(m["total_usd_micros"])
        self.assertEqual(m["priced_usd_micros_partial"], 5000)
        self.assertFalse(m["cost_complete"])
        self.assertEqual(m["per_task_cost"][0]["unpriced_actual_tokens"],
                         700)

    def test_missing_cost_rows_make_total_unknown(self):
        records = [_record("a", "completed", attempts=1)]
        m = compute_metrics(records, costs={})
        self.assertIsNone(m["total_usd_micros"])
        self.assertFalse(m["cost_complete"])


class TestFeedbackRules(unittest.TestCase):
    def test_defect_breach_tightens_check(self):
        m = compute_metrics([
            _record("a", "completed", attempts=1),
            _record("b", "failed", attempts=1, defects=1,
                    category="contract"),
        ])
        actions = evaluate_feedback(
            m, thresholds={"defect_rate": 0.20,
                           "first_pass_rate_min": 0.50,
                           "check_overturn_rate": 0.30})
        tighten = [a for a in actions if a["kind"] == "tighten_check"]
        self.assertEqual(len(tighten), 1)
        self.assertEqual(tighten[0]["target"], "contract")
        self.assertFalse(tighten[0]["auto_apply"])

    def test_no_breach_no_action(self):
        m = compute_metrics([_record("a", "completed", attempts=1)])
        actions = evaluate_feedback(
            m, thresholds={"defect_rate": 0.20,
                           "first_pass_rate_min": 0.50,
                           "check_overturn_rate": 0.30})
        self.assertEqual(actions, [])

    def test_low_first_pass_routes_with_extra_audit(self):
        m = compute_metrics([
            _record("a", "completed", attempts=3),
            _record("b", "failed", attempts=2),
        ])
        actions = evaluate_feedback(
            m, thresholds={"defect_rate": 0.20,
                           "first_pass_rate_min": 0.50,
                           "check_overturn_rate": 0.30})
        self.assertTrue(any(a["kind"] == "route_with_extra_audit"
                            for a in actions))

    def test_high_overturn_recalibrates_without_auto_apply(self):
        m = compute_metrics([_record("a", "completed", attempts=1)])
        actions = evaluate_feedback(
            m, thresholds={"defect_rate": 0.20,
                           "first_pass_rate_min": 0.50,
                           "check_overturn_rate": 0.30},
            check_stats=[{"check_id": "forbidden-dir", "violations": 10,
                          "overturned": 5}])
        recal = [a for a in actions if a["kind"] == "recalibrate_check"]
        self.assertEqual(len(recal), 1)
        self.assertFalse(recal[0]["auto_apply"])

    def test_low_overturn_no_recalibration(self):
        m = compute_metrics([_record("a", "completed", attempts=1)])
        actions = evaluate_feedback(
            m, check_stats=[{"check_id": "forbidden-dir",
                             "violations": 10, "overturned": 1}])
        self.assertEqual(
            [a for a in actions if a["kind"] == "recalibrate_check"], [])


class TestGoldenSetAndStandardsLoop(unittest.TestCase):
    def test_unapproved_golden_set_is_not_decision_ready(self):
        m = compute_metrics([_record("a", "completed", attempts=1)])
        report = build_report(m, golden={"tasks": ["a", "b"]})
        self.assertFalse(report["decision_ready"])
        self.assertEqual(report["golden_set"]["status"],
                         "pending_human_approval")

    def test_approved_golden_set_is_decision_ready(self):
        m = compute_metrics([_record("a", "completed", attempts=1)])
        report = build_report(
            m, golden={"tasks": ["a"], "approved_by": "human",
                       "approved_at": "2026-10-31"},
            thresholds_source="human_approved")
        self.assertTrue(report["decision_ready"])
        self.assertEqual(report["thresholds_source"], "human_approved")

    def test_sample_only_set_must_not_be_presented_as_golden(self):
        # The set that exists today is sample-only: no approval fields
        # exist for it, so any report built from it stays pending.
        m = compute_metrics([_record("a", "completed", attempts=1)])
        report = build_report(m, golden={"tasks": ["sample-1"]})
        self.assertFalse(report["decision_ready"])
        self.assertIn("NOT human-approved", report["decision_note"])

    def test_standards_change_goes_through_contract_protocol(self):
        # The loop closes through the same protocol: an audit-standard
        # version bump without approval fails; with approval it passes.
        self.assertFalse(
            check_contract_version("v1", "v2")["ok"])
        self.assertTrue(check_contract_version(
            "v1", "v2",
            change={"approved": True, "backward_compatible": True,
                    "affected_clients": ["all"]})["ok"])


if __name__ == "__main__":
    unittest.main()