"""LockedIn FDE · Canary evidence lab · Python 3.11+ · Edition 1.

python3 fde-canary-evidence-lab.py test | demo | <case>
Original synthetic count packets, not live traffic, a deployment controller,
statistical significance test, SLO recommendation or human release authority.
No packages, network, model, credentials, file writes or external effects are used.
The fixed thresholds below are teaching policy, not industry defaults.
"""
from copy import deepcopy
from fractions import Fraction
import json
import math
import sys
import unittest

SEGMENTS = ("routine", "exception")
COHORTS = ("control", "candidate")
MIN_SAMPLE = 20
RATE_DELTA = Fraction(5, 100)
LATENCY_RATIO = Fraction(3, 2)
CASES = ("baseline-ready", "aggregate-green-segment-red", "all-refusal", "missing-observation", "missing-segment", "mixed-window", "small-sample", "latency-regression", "safety-event", "duplicate-observation", "malformed-scalar")


def make_case(name):
    if name not in CASES:
        raise ValueError("unknown-case")
    rows = []
    counts = {cohort: {} for cohort in COHORTS}
    for cohort in COHORTS:
        for segment in SEGMENTS:
            count = 100 if segment == "routine" else 20
            errors = 2 if cohort == "control" or segment == "exception" else 1
            accepted = (95 if cohort == "control" else 96) if segment == "routine" else 18
            counts[cohort][segment] = count
            for i in range(count):
                rows.append({"id": f"{cohort}-{segment}-{i}", "cohort": cohort, "segment": segment, "second": i % 60, "error": i < errors, "accepted": errors <= i < errors + accepted, "latency_ms": (100 if cohort == "control" else 110) + i % 7, "unsafe": False})
    packet = {"windows": {cohort: [0, 60] for cohort in COHORTS}, "admitted": counts, "observations": rows}
    if name == "aggregate-green-segment-red":
        for i, row in enumerate(r for r in rows if r["cohort"] == "candidate" and r["segment"] == "exception"):
            row.update(error=i < 6, accepted=i >= 6)
    elif name == "all-refusal":
        for row in rows:
            if row["cohort"] == "candidate":
                row.update(error=False, accepted=False)
    elif name == "missing-observation":
        rows.pop()
    elif name == "missing-segment":
        packet["observations"] = [r for r in rows if r["segment"] != "exception"]
        for cohort in COHORTS:
            counts[cohort]["exception"] = 0
    elif name == "mixed-window":
        packet["windows"]["candidate"] = [60, 120]
    elif name == "small-sample":
        packet["observations"] = [r for r in rows if int(r["id"].split("-")[-1]) < 5]
        for cohort in COHORTS:
            counts[cohort] = {segment: 5 for segment in SEGMENTS}
    elif name == "latency-regression":
        for row in rows:
            if row["cohort"] == "candidate" and row["segment"] == "exception":
                row["latency_ms"] = 900
    elif name == "safety-event":
        rows[-1]["unsafe"] = True
    elif name == "duplicate-observation":
        rows[-1]["id"] = rows[0]["id"]
    elif name == "malformed-scalar":
        rows[-1]["latency_ms"] = True
    return packet


def metrics(rows):
    n = len(rows)
    return {"observed": n, "errors": sum(r["error"] for r in rows), "accepted": sum(r["accepted"] for r in rows), "p95_ms": sorted(r["latency_ms"] for r in rows)[math.ceil(n * .95) - 1]}


def evaluate(packet):
    """Review recommendation only. All observations and acceptance labels are fixtures."""
    def result(status, reasons, comparisons=None):
        return {"status": status, "reasons": reasons, "comparisons": comparisons or [], "origin": "synthetic-teaching-packet", "release_authorized": False}
    try:
        if not isinstance(packet, dict) or set(packet) != {"windows", "admitted", "observations"}:
            raise ValueError()
        windows, admitted, rows = packet["windows"], packet["admitted"], packet["observations"]
        if not isinstance(windows, dict) or set(windows) != set(COHORTS) or not isinstance(admitted, dict) or set(admitted) != set(COHORTS):
            raise ValueError()
        for cohort in COHORTS:
            window = windows[cohort]
            if not isinstance(window, list) or len(window) != 2 or any(type(t) is not int for t in window) or not 0 <= window[0] < window[1] <= 86400:
                raise ValueError()
            if not isinstance(admitted[cohort], dict) or set(admitted[cohort]) != set(SEGMENTS) or any(type(n) is not int or not 0 <= n <= 1000 for n in admitted[cohort].values()):
                raise ValueError()
        if not isinstance(rows, list) or len(rows) > 2000:
            raise ValueError()
        seen = set()
        for row in rows:
            if not isinstance(row, dict) or set(row) != {"id", "cohort", "segment", "second", "error", "accepted", "latency_ms", "unsafe"}:
                raise ValueError()
            if not isinstance(row["id"], str) or not 1 <= len(row["id"]) <= 80 or row["id"] in seen or row["cohort"] not in COHORTS or row["segment"] not in SEGMENTS:
                raise ValueError()
            seen.add(row["id"])
            if any(type(row[k]) is not bool for k in ("error", "accepted", "unsafe")) or (row["error"] and row["accepted"]):
                raise ValueError()
            if type(row["second"]) is not int or type(row["latency_ms"]) is not int or not 1 <= row["latency_ms"] <= 60000:
                raise ValueError()
    except (KeyError, TypeError, ValueError):
        return result("hold", ["invalid-or-duplicate-observation"])
    # A supplied critical event cannot be averaged away or excused by low volume.
    if any(row["unsafe"] for row in rows):
        return result("stop", ["critical-safety-observation"])
    if windows["control"] != windows["candidate"]:
        return result("hold", ["non-comparable-windows"])
    if any(not windows[r["cohort"]][0] <= r["second"] < windows[r["cohort"]][1] for r in rows):
        return result("hold", ["out-of-window-observation"])
    groups = {(c, s): [r for r in rows if r["cohort"] == c and r["segment"] == s] for c in COHORTS for s in SEGMENTS}
    if any(len(groups[c, s]) != admitted[c][s] for c in COHORTS for s in SEGMENTS):
        return result("hold", ["admitted-observed-mismatch"])
    if any(len(group) < MIN_SAMPLE for group in groups.values()):
        return result("hold", ["insufficient-segment-evidence"])
    comparisons, reasons = [], []
    for segment in SEGMENTS:
        control, candidate = (metrics(groups[c, segment]) for c in COHORTS)
        comparisons.append({"segment": segment, "control": control, "candidate": candidate})
        if Fraction(candidate["errors"], candidate["observed"]) - Fraction(control["errors"], control["observed"]) > RATE_DELTA:
            reasons.append(f"{segment}:error-regression")
        if Fraction(control["accepted"], control["observed"]) - Fraction(candidate["accepted"], candidate["observed"]) > RATE_DELTA:
            reasons.append(f"{segment}:accepted-work-regression")
        if Fraction(candidate["p95_ms"], control["p95_ms"]) > LATENCY_RATIO:
            reasons.append(f"{segment}:latency-regression")
    return result("stop" if reasons else "ready-for-human-review", reasons or ["authored-thresholds-met"], comparisons)


class LabTests(unittest.TestCase):
    def test_useful_comparison_does_not_grant_release_authority(self):
        r = evaluate(make_case("baseline-ready"))
        self.assertEqual(r["status"], "ready-for-human-review")
        self.assertEqual(len(r["comparisons"]), 2)
        self.assertFalse(r["release_authorized"])

    def test_aggregate_green_cannot_hide_a_segment_failure(self):
        p = make_case("aggregate-green-segment-red")
        counts = [metrics([r for r in p["observations"] if r["cohort"] == c]) for c in COHORTS]
        self.assertLess(Fraction(counts[1]["errors"] - counts[0]["errors"], counts[0]["observed"]), RATE_DELTA)
        self.assertIn("exception:error-regression", evaluate(p)["reasons"])

    def test_refusing_everything_does_not_pass(self):
        r = evaluate(make_case("all-refusal"))
        self.assertEqual(r["status"], "stop")
        self.assertIn("routine:accepted-work-regression", r["reasons"])

    def test_missing_small_and_mixed_evidence_hold(self):
        for case in ("missing-observation", "missing-segment", "small-sample", "mixed-window"):
            with self.subTest(case=case):
                self.assertEqual(evaluate(make_case(case))["status"], "hold")

    def test_slow_tail_is_retained(self):
        self.assertIn("exception:latency-regression", evaluate(make_case("latency-regression"))["reasons"])

    def test_critical_event_stops_even_with_small_sample(self):
        p = make_case("small-sample"); p["observations"][0]["unsafe"] = True
        self.assertEqual(evaluate(p)["status"], "stop")

    def test_malformed_and_duplicate_rows_are_not_silent_drops(self):
        for case in ("malformed-scalar", "duplicate-observation"):
            self.assertEqual(evaluate(make_case(case))["reasons"], ["invalid-or-duplicate-observation"])
        p = make_case("baseline-ready"); p["observations"][0]["second"] = 60
        self.assertEqual(evaluate(p)["reasons"], ["out-of-window-observation"])

    def test_exact_rate_boundary_and_input_preservation(self):
        p = make_case("baseline-ready"); original = deepcopy(p)
        evaluate(p); self.assertEqual(p, original)
        group = [r for r in p["observations"] if r["cohort"] == "candidate" and r["segment"] == "exception"]
        group[2].update(error=True, accepted=False)
        self.assertNotIn("exception:error-regression", evaluate(p)["reasons"])
        self.assertEqual(evaluate(p)["status"], "ready-for-human-review")

    def test_all_named_packets_have_explicit_non_authorizing_results(self):
        for name in CASES:
            r = evaluate(make_case(name))
            self.assertIn(r["status"], ("hold", "stop", "ready-for-human-review"))
            self.assertFalse(r["release_authorized"])


if __name__ == "__main__":
    command = sys.argv[1] if len(sys.argv) == 2 else ""
    if command == "test":
        unittest.main(argv=[sys.argv[0]])
    elif command == "demo" or command in CASES:
        for case in CASES if command == "demo" else (command,):
            print(json.dumps({"case": case, **evaluate(make_case(case))}, sort_keys=True))
    else:
        sys.stderr.write("Use test, demo, or one named case from CASES.\n")
        sys.exit(2)