#!/usr/bin/env python3
"""Read-only audit of a revenue workflow from normalized history and request files.

Usage:
  python3 workflow_audit.py transitions --config workflow-config.json \
      --transitions transitions.csv --as-of 2026-09-30T20:00:00Z [--since 2026-07-01] [--json]
  python3 workflow_audit.py sla --config workflow-config.json \
      --requests requests.csv --as-of 2026-09-30T20:00:00Z [--json]
  python3 workflow_audit.py self-test

transitions.csv: record_id,object,field,value,changed_at,source
  (normalize_history.py writes this shape from CRM history exports)
requests.csv: request_id,record_id,request_class,received_at,crm_seen_at,assigned_at,
  last_assigned_at,first_response_at,disposition,disposition_reason,disposition_at,
  owner_changes,excluded_reason

Rules (thresholds and config keys: references/reading-the-crm.md):
  T01 history-coverage  T02 unknown-values  T03 backward-moves  T04 gates-and-skips
  T05 writer-ping-pong  T06 writers  T07 aging  T08 conversion-by-cohort  T09 velocity
  T10 re-entry-loops
  S01 timestamps  S02 compliance  S03 unanswered  S04 clock-decomposition
  S05 owner-churn  S06 owner-vs-buyer-clock  S07 unowned

Output: "<SEVERITY> <RULE> <field or class> [<IDs>]: <detail>", INFO lines for metrics,
then "<n> FAIL, <n> WARN, <n> PASS". --json prints the same findings as JSON. Rows dated
after --as-of are ignored, so a run is repeatable. Business minutes come from the
calendar of the team that owns each class (classes.<class>.calendar names an entry in
calendars; without it the single calendar block is used). IANA zones, so daylight
saving is handled. A day takes one span ["09:00", "17:00"] or a list of spans, and
"24:00" closes at midnight. Email-like sources and values in transitions.csv become
UNMAPPED_USER:<short hash> labels, as in normalize_history.py, so no address is printed.

Standard library only. Reads local files and writes to stdout only (the self-test uses a
temporary folder it deletes). Exit 0 with no FAIL, 1 with at least one FAIL, 2 on bad
input, including a file with no rows to audit, a --since after every change and a requests
file with no eligible request.
"""
import argparse
import contextlib
import csv
import hashlib
import io
import json
import math
import os
import re
import sys
import tempfile
from collections import Counter, defaultdict
from datetime import date, datetime, time as dtime, timedelta, timezone

try:
    from zoneinfo import ZoneInfo
except ImportError:  # Python older than 3.9
    ZoneInfo = None

DAY_KEYS = ["mon", "tue", "wed", "thu", "fri", "sat", "sun"]
DEFAULT_CAPS = {"contacts": 45, "companies": 20, "deals": 20, "tickets": 20}
THRESHOLDS = {"missing_start_fail_rate": 0.01, "ingest_lag_p90_minutes": 5,
              "owner_changes_warn": 2, "capped_fail_rate": 0.05}
TRANSITION_COLUMNS = ["record_id", "object", "field", "value", "changed_at", "source"]
EMAIL_RE = re.compile(r"[^@\s]+@[^@\s]+\.[A-Za-z]{2,}")
ISO_RE = re.compile(
    r"^(\d{4})-(\d{2})-(\d{2})[T ](\d{2}):(\d{2})(?::(\d{2})(?:[.,](\d{1,9}))?)?"
    r"\s*(Z|[+-]\d{2}:?\d{2})?$", re.I)
UTC = timezone.utc


class InputError(Exception):
    pass


# ---------- parsing and small helpers ----------

def offset_tz(text):
    m = re.match(r"^([+-])(\d{2}):?(\d{2})$", text or "")
    if not m:
        raise InputError("offset %r must look like -05:00" % text)
    sign = 1 if m.group(1) == "+" else -1
    try:
        return timezone(sign * timedelta(hours=int(m.group(2)), minutes=int(m.group(3))))
    except ValueError:
        raise InputError("offset %r is out of range" % text)


def parse_time(text):
    """Aware UTC datetime, or None for empty or unparseable text. No offset means UTC."""
    s = (text or "").strip()
    if not s:
        return None
    if re.match(r"^\d{4}-\d{2}-\d{2}$", s):
        s += "T00:00:00"
    m = ISO_RE.match(s)
    if not m:
        return None
    parts = [int(m.group(i)) for i in range(1, 6)] + [int(m.group(6) or 0)]
    micro = int((m.group(7) or "0")[:6].ljust(6, "0"))
    zone = m.group(8)
    try:
        tz = UTC if zone is None or zone.upper() == "Z" else offset_tz(zone)
        return datetime(*parts, micro, tzinfo=tz).astimezone(UTC)
    except (ValueError, InputError):
        return None


def median(values):
    v = sorted(values)
    if not v:
        return None
    mid = len(v) // 2
    return v[mid] if len(v) % 2 else (v[mid - 1] + v[mid]) / 2.0


def p90(values):
    v = sorted(values)
    return v[max(1, math.ceil(0.9 * len(v))) - 1] if v else None


def num(x, places=1):
    if x is None:
        return "n/a"
    if abs(x - round(x)) < 1e-9:
        return str(int(round(x)))
    return ("%." + str(places) + "f") % x


def pct(part, whole):
    return "%.1f%%" % (100.0 * part / whole) if whole else "n/a"


def day(dt):
    return dt.strftime("%Y-%m-%d")


def matches(source, prefixes):
    return any(source.startswith(p) for p in prefixes or [])


def examples(events, limit=3):
    """events: [(record, text)] -> 'text; text; text; and N more'."""
    more = len(events) - limit
    return "; ".join(text for _, text in events[:limit]) + ("; and %d more" % more if more > 0 else "")


class Report:
    def __init__(self):
        self.items = []
        self.audited = 0  # SLA classes with at least one eligible request

    def add(self, severity, rule, scope, ids, detail):
        self.items.append({"severity": severity, "rule": rule, "scope": scope,
                           "ids": list(ids), "detail": detail})

    def counts(self):
        c = Counter(i["severity"] for i in self.items)
        return c["FAIL"], c["WARN"], c["PASS"]

    def render(self, header, as_json):
        fails, warns, passes = self.counts()
        if as_json:
            print(json.dumps({"input": header, "findings": self.items,
                              "summary": {"FAIL": fails, "WARN": warns, "PASS": passes}}, indent=2))
            return
        for line in header:
            print(line)
        for i in self.items:
            ids = i["ids"][:6] + (["(+%d more)" % (len(i["ids"]) - 6)] if len(i["ids"]) > 6 else [])
            print("%s %s %s%s: %s" % (i["severity"], i["rule"], i["scope"],
                                      (" " + ", ".join(ids)) if ids else "", i["detail"]))
        print("%d FAIL, %d WARN, %d PASS. Nothing was changed." % (fails, warns, passes))


def load_config(path):
    try:
        with open(path, encoding="utf-8") as fh:
            cfg = json.load(fh)
    except (OSError, ValueError) as exc:
        raise InputError("cannot read config %s: %s" % (path, exc))
    if not isinstance(cfg, dict):
        raise InputError("config must be a JSON object")
    validate_config(cfg)
    th = dict(THRESHOLDS)
    th.update(cfg.get("thresholds") or {})
    cfg["thresholds"] = th
    return cfg


def validate_config(cfg):
    """Raise InputError for shapes the audit cannot use; keys starting with _ are notes."""
    def is_num(x):
        return isinstance(x, (int, float)) and not isinstance(x, bool)

    def need(ok, where, what):
        if not ok:
            raise InputError("config %s must be %s" % (where, what))

    def str_list(x):
        return isinstance(x, list) and all(isinstance(i, str) for i in x)

    need(isinstance(cfg.get("fields", {}), dict), "fields", "an object")
    for f, fc in (cfg.get("fields") or {}).items():
        if f.startswith("_"):
            continue
        need(isinstance(fc, dict), "fields.%s" % f, "an object")
        for key in ("ordered", "terminal", "unranked", "allowed_skips", "loop_values"):
            need(str_list(fc.get(key, [])), "fields.%s.%s" % (f, key), "a list of strings")
        need(isinstance(fc.get("gates", {}), dict) and all(str_list(v) for v in (fc.get("gates") or {}).values()),
             "fields.%s.gates" % f, "an object of value -> list of required prior values")
        need(isinstance(fc.get("max_dwell_days", {}), dict)
             and all(is_num(v) for v in (fc.get("max_dwell_days") or {}).values()),
             "fields.%s.max_dwell_days" % f, "an object of value -> days")
        need(isinstance(fc.get("max_entries_90d", 2), int), "fields.%s.max_entries_90d" % f, "a whole number")
    need(isinstance(cfg.get("writers", {}), dict), "writers", "an object")
    for f, rules in (cfg.get("writers") or {}).items():
        if f.startswith("_"):
            continue
        need(isinstance(rules, dict) and all(str_list(v) for v in rules.values()),
             "writers.%s" % f, "an object of value -> list of source prefixes")
    for key in ("manual_source_prefixes", "correction_source_prefixes"):
        need(str_list(cfg.get(key, [])), key, "a list of strings")
    for key in ("clear_set_seconds", "pingpong_hours"):
        need(is_num(cfg.get(key, 0)), key, "a number")
    need(isinstance(cfg.get("revision_caps", {}), dict)
         and all(v is None or isinstance(v, int) for v in (cfg.get("revision_caps") or {}).values()),
         "revision_caps", "an object of object -> whole number or null")
    for i, pair in enumerate(cfg.get("conversions") or []):
        need(isinstance(pair, list) and len(pair) == 4 and all(isinstance(x, str) for x in pair[:3])
             and is_num(pair[3]), "conversions[%d]" % i, "[field, from, to, window_days]")
    calendars = cfg.get("calendars", {})
    need(isinstance(calendars, dict) and all(isinstance(v, dict) for k, v in (calendars or {}).items()
                                             if not k.startswith("_")),
         "calendars", "an object of calendar name -> calendar block")
    need(isinstance(cfg.get("classes", {}), dict), "classes", "an object")
    for name, cc in (cfg.get("classes") or {}).items():
        if name.startswith("_"):
            continue
        need(isinstance(cc, dict), "classes.%s" % name, "an object")
        for key in ("target_business_minutes", "max_unassigned_business_minutes"):
            need(cc.get(key) is None or is_num(cc.get(key)), "classes.%s.%s" % (name, key), "a number or null")
        need(cc.get("calendar") is None or (isinstance(cc.get("calendar"), str) and cc["calendar"] in (calendars or {})),
             "classes.%s.calendar" % name, "the name of an entry in calendars")
        need(is_num(cc.get("target_compliance", 0.9)) and 0 < cc.get("target_compliance", 0.9) <= 1,
             "classes.%s.target_compliance" % name, "a share between 0 and 1, such as 0.9")
    need(isinstance(cfg.get("thresholds", {}), dict)
         and all(is_num(v) for v in (cfg.get("thresholds") or {}).values()), "thresholds", "an object of numbers")


# ---------- coverage calendar ----------

def clock_minutes(text, closing):
    """'09:30' -> 570. '24:00' is allowed only as a closing time (midnight at the end of the day)."""
    m = re.match(r"^(\d{1,2}):(\d{2})$", text.strip()) if isinstance(text, str) else None
    if not m:
        raise ValueError(text)
    hours, mins = int(m.group(1)), int(m.group(2))
    if mins > 59 or hours > 24 or (hours == 24 and (mins or not closing)):
        raise ValueError(text)
    return hours * 60 + mins


class Calendar:
    def __init__(self, cfg, label="calendar"):
        if not isinstance(cfg, dict) or not isinstance(cfg.get("hours"), dict) or not cfg.get("hours"):
            raise InputError("config needs %s.timezone and %s.hours" % (label, label))
        self.warning, self.tz, self.label = None, None, label
        name = cfg.get("timezone")
        if name and ZoneInfo is not None:
            try:
                self.tz = ZoneInfo(name)
            except Exception:
                self.tz = None
        if self.tz is None:
            if not cfg.get("utc_offset"):
                raise InputError("time zone %r could not be loaded; set %s.utc_offset as a fallback" % (name, label))
            self.tz = offset_tz(cfg["utc_offset"])
            self.warning = ("%s uses the fixed offset %s: daylight saving changes are not handled"
                            % (label, cfg["utc_offset"]))
        self.zone = name if self.warning is None else cfg["utc_offset"]
        self.hours = {}
        for index, key in enumerate(DAY_KEYS):
            spec = cfg["hours"].get(key)
            if not spec:
                continue
            spans = spec if isinstance(spec, list) and all(isinstance(s, list) for s in spec) else [spec]
            parsed = []
            for span in spans:
                try:
                    if not isinstance(span, list) or len(span) != 2:
                        raise ValueError(span)
                    opening, closing = clock_minutes(span[0], False), clock_minutes(span[1], True)
                except (TypeError, ValueError, AttributeError):
                    raise InputError('%s.hours.%s must look like ["09:00", "17:00"], or a list of spans such as '
                                     '[["00:00", "06:00"], ["22:00", "24:00"]]' % (label, key))
                if closing <= opening:
                    raise InputError("%s.hours.%s must close after it opens" % (label, key))
                parsed.append((opening, closing))
            parsed.sort()
            if any(b[0] < a[1] for a, b in zip(parsed, parsed[1:])):
                raise InputError("%s.hours.%s has overlapping spans" % (label, key))
            self.hours[index] = parsed
        try:
            self.holidays = set(date.fromisoformat(d) for d in cfg.get("holidays") or [])
        except (TypeError, ValueError) as exc:
            raise InputError("%s.holidays: %s" % (label, exc))

    def at(self, day, minute):
        """UTC time of a local wall-clock minute on a day; minute 1440 is the next midnight."""
        if minute >= 1440:
            day, minute = day + timedelta(days=1), minute - 1440
        return datetime.combine(day, dtime(minute // 60, minute % 60), tzinfo=self.tz).astimezone(UTC)

    def minutes(self, start, end):
        """Business minutes between two aware datetimes."""
        if end <= start:
            return 0.0
        total, current = 0.0, start.astimezone(self.tz).date()
        last = end.astimezone(self.tz).date()
        while current <= last:
            if current not in self.holidays:
                for opening, closing in self.hours.get(current.weekday(), []):
                    low, high = max(start, self.at(current, opening)), min(end, self.at(current, closing))
                    if high > low:
                        total += (high - low).total_seconds() / 60.0
            current += timedelta(days=1)
        return total


# ---------- transitions ----------

def mask_email(text, masked):
    """An email-like value becomes UNMAPPED_USER:<first 8 hex of sha256>, the label normalize_history.py uses."""
    if text and EMAIL_RE.search(text):
        label = "UNMAPPED_USER:" + hashlib.sha256(text.lower().encode("utf-8")).hexdigest()[:8]
        masked.add(label)
        return label
    return text


def load_transitions(path, as_of):
    rows, skipped, masked = [], Counter(), set()
    try:
        with open(path, newline="", encoding="utf-8-sig") as fh:
            reader = csv.DictReader(fh)
            missing = [c for c in TRANSITION_COLUMNS if c not in (reader.fieldnames or [])]
            if missing:
                raise InputError("%s is missing columns: %s" % (path, ", ".join(missing)))
            for index, row in enumerate(reader):
                when = parse_time(row["changed_at"])
                if when is None:
                    skipped["unparseable changed_at"] += 1
                elif when > as_of:
                    skipped["after --as-of"] += 1
                else:
                    rows.append({"record": (row["record_id"] or "").strip(), "object": (row["object"] or "").strip(),
                                 "field": (row["field"] or "").strip(),
                                 "value": mask_email((row["value"] or "").strip(), masked),
                                 "time": when, "source": mask_email((row["source"] or "").strip(), masked),
                                 "index": index})
    except OSError as exc:
        raise InputError("cannot read %s: %s" % (path, exc))
    return rows, skipped, len(masked)


def build_histories(rows, clear_set_seconds):
    """{(field, record): {"object", "raw", "changes", "entries"}} with clear-then-set collapsed."""
    grouped = defaultdict(list)
    for r in rows:
        grouped[(r["field"], r["record"])].append(r)
    histories = {}
    for key, items in grouped.items():
        items.sort(key=lambda r: (r["time"], r["index"]))
        changes, i = [], 0
        while i < len(items):
            cur, nxt = items[i], items[i + 1] if i + 1 < len(items) else None
            if (cur["value"] == "" and nxt is not None and nxt["value"] != "" and nxt["source"] == cur["source"]
                    and (nxt["time"] - cur["time"]).total_seconds() <= clear_set_seconds):
                changes.append({"value": nxt["value"], "time": nxt["time"], "source": nxt["source"], "clear_set": True})
                i += 2
                continue
            changes.append({"value": cur["value"], "time": cur["time"], "source": cur["source"], "clear_set": False})
            i += 1
        entries, prev = [], None
        for ch in changes:
            if ch["value"] and ch["value"] != prev:
                entries.append(ch)
            prev = ch["value"]
        histories[key] = {"object": items[0]["object"], "raw": items, "changes": changes, "entries": entries}
    return histories


def first_entry(h, value, after=None):
    for e in h["entries"]:
        if e["value"] == value and (after is None or e["time"] > after):
            return e
    return None


def audit_transitions(cfg, rows, skipped, as_of, since, report):
    fields_cfg = cfg.get("fields") or {}
    writers_cfg = cfg.get("writers") or {}
    th = cfg["thresholds"]
    caps = dict(DEFAULT_CAPS)
    caps.update(cfg.get("revision_caps") or {})
    manual = cfg.get("manual_source_prefixes", ["CRM_UI:"])
    correction = cfg.get("correction_source_prefixes", [])
    histories = build_histories(rows, cfg.get("clear_set_seconds", 60))
    present = sorted(set(f for f, _ in histories))
    fields = [f for f in fields_cfg if f in present] + [f for f in present if f not in fields_cfg]
    by_field = defaultdict(list)
    for (f, record), h in sorted(histories.items()):
        by_field[f].append((record, h))
    in_scope = (lambda t: since is None or t >= since)

    # T01 history coverage
    for f in fields:
        obj = (fields_cfg.get(f) or {}).get("object") or by_field[f][0][1]["object"]
        cap = caps.get(obj)
        counts = {record: len(h["raw"]) for record, h in by_field[f]}
        if cap is None:
            report.add("INFO", "T01", f, [], "no revision cap set for %s (unknown or not applicable); check one "
                       "busy record against the CRM before trusting transition counts" % obj)
            continue
        at_cap = sorted(r for r, n in counts.items() if n >= cap)
        rate = len(at_cap) / float(len(counts))
        if at_cap:
            sev = "FAIL" if rate > th["capped_fail_rate"] else "WARN"
            report.add(sev, "T01", f, at_cap, "%d of %d records (%s) have %d or more revisions, so older history "
                       "may be missing" % (len(at_cap), len(counts), pct(len(at_cap), len(counts)), cap))
        else:
            report.add("PASS", "T01", f, [], "no record at the %s revision cap (%d); most revisions on one "
                       "record: %d" % (obj, cap, max(counts.values())))
    for f in [f for f in fields_cfg if not f.startswith("_") and f not in present]:
        report.add("WARN", "T01", f, [], "configured in fields, but the file has no rows for it, so no rule ran for "
                   "it; check the field name in the file and any --field-map")
    for f in [f for f in present if f not in fields_cfg]:
        report.add("WARN", "T01", f, [], "%d rows in the file but no fields entry, so T02 to T04, T07 and T10 did not "
                   "run for it%s; fix the name or add it to fields" % (
                       sum(len(h["raw"]) for _, h in by_field[f]),
                       "" if f in writers_cfg else " and T06 only lists its writers"))
    if skipped.get("unparseable changed_at"):
        report.add("WARN", "T01", "input", [], "%d rows skipped with an unparseable changed_at"
                   % skipped["unparseable changed_at"])

    # T02 unknown values
    for f in fields:
        fc = fields_cfg.get(f)
        known = set((fc or {}).get("ordered") or []) | set((fc or {}).get("terminal") or []) \
            | set((fc or {}).get("unranked") or [])
        if not known:
            report.add("INFO", "T02", f, [], "no value list configured; values not checked")
            continue
        unknown = defaultdict(lambda: {"records": set(), "sources": Counter()})
        for record, h in by_field[f]:
            for ch in h["changes"]:
                if ch["value"] and ch["value"] not in known and in_scope(ch["time"]):
                    unknown[ch["value"]]["records"].add(record)
                    unknown[ch["value"]]["sources"][ch["source"]] += 1
        total = len(by_field[f])
        for value, info in sorted(unknown.items()):
            report.add("WARN", "T02", f, sorted(info["records"]), "unknown value '%s' set %d time(s) by %s on %d of "
                       "%d records (%s)" % (value, sum(info["sources"].values()), ", ".join(sorted(info["sources"])),
                                            len(info["records"]), total, pct(len(info["records"]), total)))
        if not unknown:
            report.add("PASS", "T02", f, [], "every value is in the configured lists")

    # T03 backward moves, T04 gates and skips
    t03, t04 = [], []
    for f in fields:
        fc = fields_cfg.get(f) or {}
        ordered, terminal = fc.get("ordered") or [], set(fc.get("terminal") or [])
        if not ordered:
            t03.append(("INFO", f, [], "no ordered list; backward moves, gates and skips not checked"))
            continue
        rank = {v: i for i, v in enumerate(ordered)}
        for v in terminal:
            rank.setdefault(v, len(ordered))
        gates, skips_ok = fc.get("gates") or {}, set(fc.get("allowed_skips") or [])
        back, gate_hits, skip_hits = defaultdict(list), defaultdict(list), defaultdict(list)
        entered_gate = defaultdict(set)
        total = len(by_field[f])
        for record, h in by_field[f]:
            last, seen, prev = None, set(), None
            for ch in h["changes"]:
                v, when = ch["value"], ch["time"]
                if not v:
                    prev = ""
                    continue
                entry = v != prev
                if entry and v in gates and in_scope(when):
                    entered_gate[v].add(record)
                # Backward: a lower rank, or leaving an end state (a terminal value or the last ordered value,
                # such as closedwon) for any other ranked value. Terminal values share one rank.
                backward = last is not None and v in rank and (
                    rank[v] < rank[last] or (v != last and (last in terminal or last == ordered[-1])))
                if backward and in_scope(when):
                    sev = "PASS" if matches(ch["source"], correction) else (
                        "WARN" if matches(ch["source"], manual) else "FAIL")
                    back[(sev, ch["source"])].append((record, "%s %s to %s on %s%s" % (
                        record, last, v, day(when), " (clear-then-set)" if ch["clear_set"] else "")))
                if entry and v in gates and not (seen & set(gates[v])) and in_scope(when):
                    gate_hits[(v, ch["source"])].append((record, "%s on %s%s" % (
                        record, day(when), " as its first value" if not seen else "")))
                if (entry and v in rank and last is not None and rank[v] > rank[last] + 1 and v not in terminal
                        and last not in terminal and "%s>%s" % (last, v) not in skips_ok and in_scope(when)):
                    skip_hits[(last, v)].append((record, "%s by %s on %s" % (record, ch["source"], day(when))))
                if v in rank:
                    last = v
                seen.add(v)
                prev = v
        for (sev, source), events in sorted(back.items(), key=lambda kv: ({"FAIL": 0, "WARN": 1}.get(kv[0][0], 2), kv[0][1])):
            if sev == "PASS":
                continue
            records = sorted(set(r for r, _ in events))
            kind = "automated, outside the correction path" if sev == "FAIL" else "manual, no correction source"
            t03.append((sev, f, records, "%d backward move(s) by %s (%s) on %d of %d records: %s" % (
                len(events), source, kind, len(records), total, examples(events))))
        through = sum(len(e) for (sev, _), e in back.items() if sev == "PASS")
        if not any(sev != "PASS" for sev, _ in back):
            t03.append(("PASS", f, [], "no backward moves outside the correction path"
                        + (" (%d through it)" % through if through else "")))
        for (v, source), events in sorted(gate_hits.items()):
            records = sorted(set(r for r, _ in events))
            entered = len(entered_gate[v])
            t04.append(("FAIL", f, records, "%d of %d records entering %s (%s) had no prior %s, set by %s: %s" % (
                len(records), entered, v, pct(len(records), entered), " or ".join(gates[v]), source,
                examples(events))))
        for (a, b), events in sorted(skip_hits.items()):
            records = sorted(set(r for r, _ in events))
            t04.append(("WARN", f, records, "%d move(s) from %s to %s on %d of %d records (%s) skip %s, which "
                        "allowed_skips does not list: %s" % (
                            len(events), a, b, len(records), total, pct(len(records), total),
                            " and ".join(ordered[rank[a] + 1:rank[b]]), examples(events))))
        if not gate_hits and not skip_hits:
            t04.append(("PASS", f, [], "no gate violations and no unexpected skips"))
    for rule, items in (("T03", t03), ("T04", t04)):
        for sev, f, ids, detail in items:
            report.add(sev, rule, f, ids, detail)

    # T05 writer ping-pong
    window = timedelta(hours=cfg.get("pingpong_hours", 24))
    for f in fields:
        flips = defaultdict(list)
        for record, h in by_field[f]:
            seq = []
            for e in h["entries"]:
                if not seq or seq[-1]["value"] != e["value"]:
                    seq.append(e)
            for a, b, c in zip(seq, seq[1:], seq[2:]):
                if (c["value"] == a["value"] != b["value"] and c["time"] - b["time"] <= window
                        and b["source"] != c["source"] and in_scope(c["time"])):
                    flips[" and ".join(sorted([b["source"], c["source"]]))].append((record, "%s %s to %s (%s) and back "
                        "(%s) within %s h" % (record, a["value"], b["value"], b["source"], c["source"],
                                             num((c["time"] - b["time"]).total_seconds() / 3600.0, 2))))
        total = len(by_field[f])
        for pair, events in sorted(flips.items()):
            records = sorted(set(r for r, _ in events))
            report.add("FAIL", "T05", f, records, "%d flip(s) between %s on %d of %d records (%s): %s" % (
                len(events), pair, len(records), total, pct(len(records), total), examples(events)))
        if not flips:
            report.add("PASS", "T05", f, [], "no A to B to A flips between different sources within %s h"
                       % num(window.total_seconds() / 3600.0))

    # T06 writers
    for f in fields:
        inventory = Counter()
        rules = writers_cfg.get(f)
        bad = defaultdict(lambda: {"count": 0, "values": Counter(), "records": set(), "clear_set": 0})
        for record, h in by_field[f]:
            for ch in h["changes"]:
                if not in_scope(ch["time"]):
                    continue
                inventory[ch["source"]] += 1
                if rules is None:
                    continue
                v = ch["value"]
                allowed = rules.get(v, rules.get("*", [])) if v else rules.get("", rules.get("*", []))
                if not matches(ch["source"], allowed):
                    b = bad[ch["source"]]
                    b["count"] += 1
                    b["values"][v or "(cleared)"] += 1
                    b["records"].add(record)
                    b["clear_set"] += 1 if ch["clear_set"] else 0
        report.add("INFO", "T06", f, [], "writers seen: " + ", ".join(
            "%s %d" % (s or "(blank source)", n) for s, n in sorted(inventory.items())))
        if rules is None:
            continue
        for source, b in sorted(bad.items()):
            values = ", ".join("%s x%d" % (v, n) for v, n in sorted(b["values"].items()))
            note = " (%d clear-then-set collapsed)" % b["clear_set"] if b["clear_set"] else ""
            changes = sum(inventory.values())
            report.add("FAIL", "T06", f, sorted(b["records"]), "%s wrote values it may not write: %s; %d of %d "
                       "changes (%s)%s" % (source or "(blank source)", values, b["count"], changes,
                                           pct(b["count"], changes), note))
        if not bad:
            report.add("PASS", "T06", f, [], "every change came from an allowed writer (%s)" % ", ".join(
                sorted(set(p for plist in rules.values() for p in plist))))

    # T07 aging
    for f in fields:
        limits = (fields_cfg.get(f) or {}).get("max_dwell_days") or {}
        if not limits:
            continue
        now_in, old = Counter(), defaultdict(list)
        for record, h in by_field[f]:
            if not h["changes"] or not h["changes"][-1]["value"] or not h["entries"]:
                continue
            current = h["entries"][-1]
            if current["value"] != h["changes"][-1]["value"] or current["value"] not in limits:
                continue
            now_in[current["value"]] += 1
            days = (as_of - current["time"]).total_seconds() / 86400.0
            if days > limits[current["value"]]:
                old[current["value"]].append((days, record))
        for value in [v for v in limits if v in old]:
            items = sorted(old[value], reverse=True)
            report.add("WARN", "T07", f, [r for _, r in items], "%d of %d records now in %s have been there longer "
                       "than %s days; oldest: %s" % (len(items), now_in[value], value, num(limits[value]),
                                                    ", ".join("%s %.2f days" % (r, d) for d, r in items[:5])))
        if not old:
            report.add("PASS", "T07", f, [], "no record over its stage dwell limit")

    # T08 conversion by cohort, T09 velocity
    for pair in cfg.get("conversions") or []:
        f, src, dst, window_days = pair
        cohorts, durations = defaultdict(lambda: [0, 0]), []
        for record, h in by_field.get(f, []):
            start = first_entry(h, src)
            if start is None or not in_scope(start["time"]):
                continue
            end = first_entry(h, dst, after=start["time"])
            month = start["time"].strftime("%Y-%m")
            cohorts[month][0] += 1
            if end is not None:
                gap = (end["time"] - start["time"]).total_seconds() / 86400.0
                durations.append(gap)
                if gap <= window_days:
                    cohorts[month][1] += 1
        label = "%s to %s within %d days" % (src, dst, window_days)
        if not cohorts:
            report.add("INFO", "T08", f, [], "%s: no record entered %s%s" % (
                label, src, " on or after --since" if since else ""))
        for month in sorted(cohorts):
            entered, reached = cohorts[month]
            y, m = int(month[:4]), int(month[5:])
            mature_at = datetime(y + (m == 12), m % 12 + 1, 1, tzinfo=UTC) + timedelta(days=window_days)
            if mature_at <= as_of:
                report.add("INFO", "T08", f, [], "%s, cohort %s: entered %d, reached %d, %s" % (
                    label, month, entered, reached, pct(reached, entered)))
            else:
                report.add("INFO", "T08", f, [], "%s, cohort %s: entered %d, immature until %s" % (
                    label, month, entered, day(mature_at)))
        if durations:
            report.add("INFO", "T09", f, [], "%s to %s: median %.2f days, p90 %.2f days (n=%d)" % (
                src, dst, median(durations), p90(durations), len(durations)))

    # T10 re-entry loops
    for f in fields:
        fc = fields_cfg.get(f) or {}
        loop_values, limit = fc.get("loop_values") or [], fc.get("max_entries_90d", 2)
        if not loop_values:
            continue
        loops, entered_loop = defaultdict(list), defaultdict(set)
        for record, h in by_field[f]:
            for value in loop_values:
                times = [e["time"] for e in h["entries"] if e["value"] == value]
                if any(in_scope(t) for t in times):
                    entered_loop[value].add(record)
                best = []
                for i in range(len(times)):
                    inside = [t for t in times[i:] if t - times[i] <= timedelta(days=90)]
                    if len(inside) > len(best):
                        best = inside
                if len(best) > limit and in_scope(best[-1]):
                    loops[value].append((record, "%s %d entries (%s)" % (
                        record, len(best), ", ".join(day(t) for t in best))))
        for value, events in sorted(loops.items()):
            entered = len(entered_loop[value])
            report.add("WARN", "T10", f, [r for r, _ in events], "%d of %d records entering %s (%s) entered it more "
                       "than %d times within 90 days: %s" % (len(events), entered, value, pct(len(events), entered),
                                                            limit, examples(events)))
        if not loops:
            report.add("PASS", "T10", f, [], "no record re-entered %s more than %d times in 90 days"
                       % (" or ".join(loop_values), limit))
    return histories


# ---------- SLA ----------

def load_requests(path):
    try:
        with open(path, newline="", encoding="utf-8-sig") as fh:
            reader = csv.DictReader(fh)
            columns = reader.fieldnames or []
            for needed in ("request_id", "request_class"):
                if needed not in columns:
                    raise InputError("%s is missing the column %s" % (path, needed))
            return list(reader), columns
    except OSError as exc:
        raise InputError("cannot read %s: %s" % (path, exc))


def class_calendar(cfg, name, cc, cache):
    """The calendar of the team that owns the class: calendars.<classes.<class>.calendar>, else calendar."""
    key = cc.get("calendar")
    if key:
        block, label = (cfg.get("calendars") or {}).get(key), "calendars.%s" % key
    elif cfg.get("calendar") is not None:
        block, label = cfg.get("calendar"), "calendar"
    else:
        raise InputError("class %s has no calendar: set classes.%s.calendar to an entry in calendars, or add a "
                         "calendar block" % (name, name))
    if label not in cache:
        cache[label] = Calendar(block, label)
    return cache[label]


def audit_sla(cfg, rows, columns, as_of, report):
    th, classes = cfg["thresholds"], cfg.get("classes") or {}
    names = [c for c in classes if not c.startswith("_")] + sorted(set(r["request_class"] for r in rows) - set(classes))
    notes, calendars_used, cache = [], [], {}
    for name in names:
        group = [r for r in rows if r["request_class"] == name]
        if not group:
            report.add("INFO", "S01", name, [], "configured, but the requests file has no rows of this class, so it "
                       "was not audited")
            continue
        cc = classes.get(name) or {}
        start_f, stop_f = cc.get("start", "received_at"), cc.get("stop", "first_response_at")
        owner_f, target = cc.get("owner_start", "last_assigned_at"), cc.get("target_business_minutes")
        goal = cc.get("target_compliance", 0.9)
        max_unowned = cc.get("max_unassigned_business_minutes")
        max_unowned = target if max_unowned is None else max_unowned
        for col in (start_f, stop_f, owner_f):
            if col not in columns:
                raise InputError("class %s needs the column %s" % (name, col))
        if target is None:
            report.add("FAIL", "S02", name, [], "no target_business_minutes configured, so this class cannot be audited")
            continue
        cal = class_calendar(cfg, name, cc, cache)
        calendars_used.append("%s on %s (%s)" % (name, cal.label.replace("calendars.", ""), cal.zone))
        if cal.warning and "Note: " + cal.warning not in notes:
            notes.append("Note: " + cal.warning)

        def ts(row, col):
            t = parse_time(row.get(col))
            return None if t is None or t > as_of else t

        excluded, missing_start, skew, no_crm, measured = Counter(), [], [], [], []
        for r in group:
            start = parse_time(r.get(start_f))
            if start is not None and start > as_of:
                continue
            if (r.get("excluded_reason") or "").strip():
                excluded[r["excluded_reason"].strip()] += 1
                continue
            stop = ts(r, stop_f)
            if start is None:
                missing_start.append(r["request_id"])
            elif stop is not None and stop < start:
                skew.append(r["request_id"])
            else:
                if ts(r, "crm_seen_at") is None:
                    no_crm.append(r["request_id"])
                measured.append((r, start, stop))
        eligible = len(measured) + len(missing_start) + len(skew)
        if not eligible:  # every row is after --as-of or excluded: no finding below would mean anything
            report.add("INFO", "S01", name, [], "no eligible request as of this run (%d excluded), so it was not audited"
                       % sum(excluded.values()))
            continue
        report.audited += 1

        # S01 timestamps
        if skew or (eligible and len(missing_start) / float(eligible) > th["missing_start_fail_rate"]):
            report.add("FAIL", "S01", name, missing_start + skew, "%d of %d eligible requests miss %s; %d stop "
                       "before they start (clock skew); both are left out of compliance" % (
                           len(missing_start), eligible, start_f, len(skew)))
        elif missing_start or no_crm:
            report.add("WARN", "S01", name, missing_start + no_crm, "%d miss %s; %d miss crm_seen_at, so their "
                       "delay cannot be split" % (len(missing_start), start_f, len(no_crm)))
        else:
            report.add("PASS", "S01", name, [], "all %d eligible requests have %s and crm_seen_at; no stop "
                       "before start" % (eligible, start_f))

        # S02 compliance and S03 unanswered
        on_time, late, overdue, not_due = [], [], [], []
        for r, start, stop in measured:
            if stop is not None:
                used = cal.minutes(start, stop)
                (on_time if used <= target else late).append((r["request_id"], used))
            else:
                age = cal.minutes(start, as_of)
                (overdue if age > target else not_due).append((r["request_id"], age))
        due = len(measured) - len(not_due)
        line = "%d requests: %d excluded, %d eligible; on time %d, late %d, overdue %d, not yet due %d" % (
            eligible + sum(excluded.values()), sum(excluded.values()), eligible, len(on_time), len(late),
            len(overdue), len(not_due))
        if missing_start or skew:
            line += "; %d left out of compliance for timestamp problems (S01)" % (len(missing_start) + len(skew))
        if due == 0:
            report.add("INFO", "S02", name, [], "nothing due yet; " + line)
        else:
            rate = len(on_time) / float(due)
            report.add("FAIL" if rate < goal else "PASS", "S02", name, [], "compliance %s (%d on time of %d due) "
                       "against a %s target; %s" % (pct(len(on_time), due), len(on_time), due,
                                                    "%g%%" % (goal * 100), line))
        if late:
            report.add("INFO", "S02", name, [], "late: %s business minutes (target %s)" % (
                ", ".join("%s %s" % (i, num(m)) for i, m in late), num(target)))
        if not_due:
            report.add("INFO", "S02", name, [], "not yet due: %s (business minutes used of %s)" % (
                ", ".join("%s %s" % (i, num(m)) for i, m in not_due), num(target)))
        answered = len(on_time) + len(late)
        if answered:
            report.add("INFO", "S02", name, [], "answered-only rate %s (%d of %d answered) is not compliance" % (
                pct(len(on_time), answered), len(on_time), answered))
        if excluded:
            report.add("INFO", "S02", name, [], "excluded: " + ", ".join(
                "%s %d" % (k, v) for k, v in sorted(excluded.items())))
        if overdue:
            overdue.sort(key=lambda x: -x[1])
            report.add("FAIL", "S03", name, [i for i, _ in overdue], "no %s and past the target: %d of %d due (%s); "
                       "business age %s minutes" % (stop_f, len(overdue), due, pct(len(overdue), due),
                                                    ", ".join("%s %s" % (i, num(m)) for i, m in overdue)))
        else:
            report.add("PASS", "S03", name, [], "no overdue unanswered request")

        # S04 clock decomposition
        ingest, to_owner, owner_to_stop, buyer_b, buyer_w, gaps = [], [], [], [], [], []
        for r, start, stop in measured:
            crm, assigned = ts(r, "crm_seen_at"), ts(r, "assigned_at")
            if crm is not None and crm >= start:
                ingest.append((r["request_id"], (crm - start).total_seconds() / 60.0))
            if crm is not None and assigned is not None and assigned >= crm:
                to_owner.append((assigned - crm).total_seconds() / 60.0)
            if assigned is not None and stop is not None and stop >= assigned:
                owner_to_stop.append(cal.minutes(assigned, stop))
            if stop is not None:
                buyer_b.append(cal.minutes(start, stop))
                buyer_w.append((stop - start).total_seconds() / 60.0)
                gaps.append((buyer_w[-1] - buyer_b[-1], r["request_id"], buyer_w[-1], buyer_b[-1]))
        lag = [m for _, m in ingest]
        if lag and p90(lag) > th["ingest_lag_p90_minutes"]:
            slow = sorted([x for x in ingest if x[1] > th["ingest_lag_p90_minutes"]], key=lambda x: -x[1])
            report.add("WARN", "S04", name, [i for i, _ in slow], "received-to-CRM p90 %s minutes is above %s "
                       "(median %s, n=%d); slowest: %s" % (num(p90(lag)), num(th["ingest_lag_p90_minutes"]),
                                                          num(median(lag)), len(lag),
                                                          ", ".join("%s %s" % (i, num(m)) for i, m in slow)))
        elif lag:
            report.add("PASS", "S04", name, [], "received-to-CRM p90 %s minutes is within %s (n=%d)" % (
                num(p90(lag)), num(th["ingest_lag_p90_minutes"]), len(lag)))
        report.add("INFO", "S04", name, [], "CRM to assigned: median %s, p90 %s wall-clock minutes (n=%d); "
                   "assigned to %s: median %s, p90 %s business minutes (n=%d)" % (
                       num(median(to_owner)), num(p90(to_owner)), len(to_owner), stop_f,
                       num(median(owner_to_stop)), num(p90(owner_to_stop)), len(owner_to_stop)))
        if buyer_b:
            report.add("INFO", "S04", name, [], "buyer clock among answered (n=%d): median %s, p90 %s business "
                       "minutes; wall-clock median %s, p90 %s minutes" % (
                           len(buyer_b), num(median(buyer_b)), num(p90(buyer_b)),
                           num(median(buyer_w)), num(p90(buyer_w))))
        baseline = [m for _, m in on_time + late + overdue]
        if baseline:
            report.add("INFO", "S04", name, [], "baseline for a first target, every due request with overdue ones "
                       "at their age (n=%d, %d overdue): median %s, p90 %s business minutes%s" % (
                           len(baseline), len(overdue), num(median(baseline)), num(p90(baseline)),
                           "; fewer than 30 due, so use the buyer promise or a labelled proposal" if len(baseline) < 30
                           else ""))
        wide_gaps = sorted([g for g in gaps if g[0] >= 60], reverse=True)[:3]
        if wide_gaps:
            report.add("INFO", "S04", name, [g[1] for g in wide_gaps], "outside coverage hours the buyer waited "
                       "longer than the SLA shows: %s" % ", ".join(
                           "%s %s wall-clock for %s business minutes" % (i, num(w), num(b)) for _, i, w, b in wide_gaps))

        # S05 owner churn
        churn = []
        for r, _, _ in measured:
            try:
                changes = int((r.get("owner_changes") or "0").strip() or 0)
            except ValueError:
                changes = 0
            if changes >= th["owner_changes_warn"]:
                churn.append((r["request_id"], changes))
        if churn:
            report.add("WARN", "S05", name, [i for i, _ in churn], "%d of %d measured requests (%s) with %d or more "
                       "owner changes before the stop event: %s" % (
                           len(churn), len(measured), pct(len(churn), len(measured)), th["owner_changes_warn"],
                           ", ".join("%s %d" % c for c in churn)))
        if not churn:
            report.add("PASS", "S05", name, [], "no request with %d or more owner changes" % th["owner_changes_warn"])

        # S06 owner clock versus buyer clock
        owner_ok, buyer_ok, hidden, counted = 0, 0, [], 0
        for r, start, stop in measured:
            owner_start = ts(r, owner_f)
            if stop is None or owner_start is None or owner_start > stop:
                continue
            counted += 1
            owner_good = cal.minutes(owner_start, stop) <= target
            buyer_good = cal.minutes(start, stop) <= target
            owner_ok += owner_good
            buyer_ok += buyer_good
            if owner_good and not buyer_good:
                hidden.append(r["request_id"])
        if hidden:
            report.add("WARN", "S06", name, hidden, "on time by the owner clock (from %s) but late for the buyer; "
                       "a report that starts at assignment shows these as on time" % owner_f)
        else:
            report.add("PASS", "S06", name, [], "no request on time for the owner but late for the buyer")
        if counted:
            report.add("INFO", "S06", name, [], "owner clock on time for %d of %d answered (%s); buyer clock on "
                       "time for %d of %d (%s)" % (owner_ok, counted, pct(owner_ok, counted),
                                                  buyer_ok, counted, pct(buyer_ok, counted)))

        # S07 unowned, and assigned after the class limit
        if "assigned_at" not in columns:
            report.add("INFO", "S07", name, [], "not checked: the requests file has no assigned_at column")
            continue
        unowned, assigned_late = [], []
        for r, start, stop in measured:
            assigned = ts(r, "assigned_at")
            if assigned is None:
                age = cal.minutes(start, stop or as_of)
                if age > max_unowned:
                    unowned.append((r["request_id"], age))
            elif cal.minutes(start, assigned) > max_unowned:
                assigned_late.append((r["request_id"], cal.minutes(start, assigned)))
        if unowned:
            unowned.sort(key=lambda x: -x[1])
            report.add("FAIL", "S07", name, [i for i, _ in unowned], "%d of %d measured requests (%s) with no owner "
                       "past %s business minutes; business age %s" % (
                           len(unowned), len(measured), pct(len(unowned), len(measured)), num(max_unowned),
                           ", ".join("%s %s" % (i, num(a)) for i, a in unowned)))
        if assigned_late:
            assigned_late.sort(key=lambda x: -x[1])
            report.add("WARN", "S07", name, [i for i, _ in assigned_late], "%d of %d measured requests (%s) assigned "
                       "after %s business minutes: %s" % (
                           len(assigned_late), len(measured), pct(len(assigned_late), len(measured)),
                           num(max_unowned), ", ".join("%s %s" % (i, num(m)) for i, m in assigned_late)))
        if not unowned and not assigned_late:
            report.add("PASS", "S07", name, [], "no request waited more than %s business minutes for an owner"
                       % num(max_unowned))
    if calendars_used:
        notes.insert(0, "Calendars: " + "; ".join(calendars_used) + ".")
    return notes


# ---------- self-test ----------

def self_test():
    checks = []

    def check(name, ok):
        checks.append((name, bool(ok)))

    def t(s):
        return parse_time(s)

    week = {k: ["09:00", "17:00"] for k in DAY_KEYS[:5]}
    cal_cfg = {"timezone": "America/New_York", "hours": week, "holidays": ["2026-09-07"]}
    cal = Calendar(cal_cfg)
    check("business minutes across the DST change", cal.minutes(t("2026-10-30T20:50:00Z"), t("2026-11-02T14:10:00Z")) == 20)
    check("business minutes from a Saturday", cal.minutes(t("2026-08-08T15:00:00Z"), t("2026-08-10T13:10:00Z")) == 10)
    check("business minutes from a holiday", cal.minutes(t("2026-09-07T15:00:00Z"), t("2026-09-08T13:05:00Z")) == 5)

    def sla(rows, cols=None, extra=None):
        cfg = {"calendar": cal_cfg, "classes": {"hr": {"target_business_minutes": 15}}, "thresholds": dict(THRESHOLDS)}
        cfg.update(extra or {})
        validate_config(cfg)
        cols = cols or ["request_id", "request_class", "received_at", "crm_seen_at", "assigned_at",
                        "last_assigned_at", "first_response_at", "owner_changes", "excluded_reason"]
        full = [dict({c: "" for c in cols}, **dict({"request_class": "hr"}, **r)) for r in rows]
        rep = Report()
        audit_sla(cfg, full, cols, t("2026-08-11T20:00:00Z"), rep)
        return rep.items

    base = {"crm_seen_at": "2026-08-10T14:00:00Z", "assigned_at": "2026-08-10T14:00:00Z",
            "last_assigned_at": "2026-08-10T14:00:00Z"}
    items = sla([dict(base, request_id="a", received_at="2026-08-10T14:00:00Z", first_response_at="2026-08-10T14:15:00Z")])
    check("a response exactly at the target is on time",
          any(i["rule"] == "S02" and i["severity"] == "PASS" and "1 on time of 1" in i["detail"] for i in items))
    items = sla([dict(base, request_id="b", received_at="2026-08-10T14:00:00Z", first_response_at="2026-08-10T13:00:00Z")])
    check("stop before start fails S01", any(i["rule"] == "S01" and i["severity"] == "FAIL" for i in items))
    items = sla([dict(base, request_id="c", received_at="", first_response_at="2026-08-10T14:05:00Z"),
                 dict(base, request_id="d", received_at="2026-08-10T14:00:00Z", first_response_at="2026-08-10T14:05:00Z")])
    check("missing received_at is out of compliance and in S01",
          any(i["rule"] == "S01" and "c" in i["ids"] for i in items)
          and any(i["rule"] == "S02" and "1 on time of 1 due" in i["detail"] for i in items))
    items = sla([{"request_id": "e", "received_at": "2026-08-10T14:00:00Z", "crm_seen_at": "2026-08-10T14:00:00Z",
                  "last_assigned_at": "2026-08-10T14:00:00Z", "first_response_at": "2026-08-10T14:05:00Z"}],
                ["request_id", "request_class", "received_at", "crm_seen_at", "last_assigned_at", "first_response_at"])
    check("S07 is skipped, not failed, without an assigned_at column",
          any(i["rule"] == "S07" and i["severity"] == "INFO" for i in items)
          and not any(i["rule"] == "S07" and i["severity"] == "FAIL" for i in items))
    items = sla([{"request_id": "f", "received_at": "2026-08-10T14:00:00Z", "crm_seen_at": "2026-08-10T14:00:00Z"}],
                extra={"classes": {"hr": {"target_business_minutes": 15, "max_unassigned_business_minutes": None}}})
    check("a null max_unassigned_business_minutes falls back to the target",
          any(i["rule"] == "S07" and i["severity"] == "FAIL" and "past 15 business" in i["detail"] for i in items))
    two = {"calendars": {"ny": cal_cfg, "ldn": dict(cal_cfg, timezone="Europe/London", holidays=[])},
           "classes": {"a": {"target_business_minutes": 60, "calendar": "ny"},
                       "b": {"target_business_minutes": 60, "calendar": "ldn"}}}
    pair = dict(base, received_at="2026-08-10T12:00:00Z", first_response_at="2026-08-10T13:30:00Z")
    items = sla([dict(pair, request_id="g", request_class="a"), dict(pair, request_id="h", request_class="b")], extra=two)
    check("each class uses its own team calendar (New York 30, London 90 business minutes)",
          any(i["rule"] == "S02" and i["scope"] == "a" and "1 on time of 1" in i["detail"] for i in items)
          and any(i["rule"] == "S02" and i["scope"] == "b" and "late: h 90" in i["detail"] for i in items))
    items = sla([dict(base, request_id="i", received_at="2026-08-10T14:00:00Z", first_response_at="2026-08-10T14:10:00Z"),
                 dict(base, request_id="j", received_at="2026-08-10T14:00:00Z")])
    check("the baseline counts an overdue request at its business age",
          any(i["rule"] == "S04" and "(n=2, 1 overdue): median 425" in i["detail"] for i in items))
    for bad, what in (({"conversions": [["stage", "a", "b"]]}, "a malformed conversions entry"),
                      ({"classes": {"a": {"calendar": "missing"}}}, "a class calendar missing from calendars")):
        try:
            validate_config(bad)
            rejected = False
        except InputError:
            rejected = True
        check("%s is bad input" % what, rejected)
    check("an out-of-range offset is unparseable, not a crash", parse_time("2026-08-10T14:00:00+25:00") is None)

    def trans(rows, cfg_extra, as_of="2026-12-31T00:00:00Z"):
        cfg = {"fields": {"stage": dict({"object": "contacts", "ordered": ["a", "b", "c", "d"], "terminal": ["x"]},
                                        **cfg_extra)}, "thresholds": dict(THRESHOLDS),
               "correction_source_prefixes": ["WORKFLOW:fix"], "manual_source_prefixes": ["CRM_UI:"]}
        cfg.update(cfg_extra.get("_top", {}))
        parsed = [{"record": r[0], "object": "contacts", "field": "stage", "value": r[1], "time": t(r[2]),
                   "source": r[3], "index": i} for i, r in enumerate(rows)]
        rep = Report()
        audit_transitions(cfg, parsed, Counter(), t(as_of), None, rep)
        return rep.items

    items = trans([("r1", "b", "2026-07-01T10:00:00Z", "WORKFLOW:score"), ("r1", "", "2026-07-05T10:00:00Z", "WORKFLOW:recycle"),
                   ("r1", "a", "2026-07-05T10:00:30Z", "WORKFLOW:recycle")], {})
    check("clear then set within 60 seconds is one backward move",
          any(i["rule"] == "T03" and i["severity"] == "FAIL" and i["detail"].startswith("1 backward") for i in items))
    items = trans([("r2", "c", "2026-07-01T10:00:00Z", "FORM:x")], {"gates": {"c": ["b"]}})
    check("a first value entering a gated stage fails T04", any(i["rule"] == "T04" and i["severity"] == "FAIL" for i in items))
    pingpong = [("r3", "a", "2026-07-01T10:00:00Z", "S1"), ("r3", "b", "2026-07-01T11:00:00Z", "S2"),
                ("r3", "a", "2026-07-01T12:00:00Z", "S3")]
    items = trans(pingpong, {})
    check("A to B to A from different sources fails T05", any(i["rule"] == "T05" and i["severity"] == "FAIL" for i in items))
    items = trans([(r, v, w, "S1") for r, v, w, _ in pingpong], {})
    check("A to B to A from one source is not ping-pong", not any(i["rule"] == "T05" and i["severity"] == "FAIL" for i in items))
    items = trans([("r4", "a", "2026-07-01T10:00:00Z", "S1"), ("r4", "x", "2026-07-02T10:00:00Z", "S1")], {})
    check("a terminal value produces no skip warning", not any(i["rule"] == "T04" and i["severity"] == "WARN" for i in items))
    busy = [("r5", v, "2026-07-0%dT10:00:00Z" % (n + 1), "S1") for n, v in enumerate("abab")]
    quiet = [("q%02d" % n, "a", "2026-07-01T10:00:00Z", "S1") for n in range(19)]
    items = trans(busy + quiet, {"_top": {"revision_caps": {"contacts": 4}}})
    check("one record of 20 at the revision cap warns T01",
          any(i["rule"] == "T01" and i["severity"] == "WARN" and i["ids"] == ["r5"] for i in items))
    items = trans(busy + [(r.replace("r5", "r7"), v, w, x) for r, v, w, x in busy] + quiet[:18],
                  {"_top": {"revision_caps": {"contacts": 4}}})
    check("two records of 20 at the revision cap fail T01", any(i["rule"] == "T01" and i["severity"] == "FAIL" for i in items))
    items = trans([("r6", "a", "2026-09-10T10:00:00Z", "S1")],
                  {"_top": {"conversions": [["stage", "a", "b", 30]]}}, as_of="2026-09-30T20:00:00Z")
    check("an open cohort prints immature", any(i["rule"] == "T08" and "immature until 2026-10-31" in i["detail"] for i in items))
    items = trans([("r8", "d", "2026-07-01T10:00:00Z", "S1"), ("r8", "x", "2026-07-02T10:00:00Z", "CRM_UI:rep_c")], {})
    check("leaving the last ordered value for a terminal value warns T03 (closedwon to closedlost by hand)",
          any(i["rule"] == "T03" and i["severity"] == "WARN" and "r8 d to x" in i["detail"] for i in items))
    items = trans([("r9", "x", "2026-07-01T10:00:00Z", "S1"), ("r9", "y", "2026-07-02T10:00:00Z", "S2")],
                  {"terminal": ["x", "y"]})
    check("a move between two terminal values fails T03 although they share a rank",
          any(i["rule"] == "T03" and i["severity"] == "FAIL" and "r9 x to y" in i["detail"] for i in items))
    items = trans([("r4", "a", "2026-07-01T10:00:00Z", "S1"), ("r4", "x", "2026-07-02T10:00:00Z", "S2")], {})
    check("entering a terminal value from an open stage is not a backward move",
          not any(i["rule"] == "T03" and i["severity"] in ("FAIL", "WARN") for i in items))
    rep = Report()
    audit_transitions({"fields": {"stage": {"object": "contacts", "ordered": ["a", "b"]}}, "thresholds": dict(THRESHOLDS)},
                      [{"record": "r1", "object": "contacts", "field": "lifecycle_stage", "value": "a",
                        "time": t("2026-07-01T10:00:00Z"), "source": "S1", "index": 0}],
                      Counter(), t("2026-12-31T00:00:00Z"), None, rep)
    check("a configured field with no rows and a field with no fields entry both warn T01",
          any(i["rule"] == "T01" and i["severity"] == "WARN" and i["scope"] == "stage" for i in rep.items)
          and any(i["rule"] == "T01" and i["severity"] == "WARN" and i["scope"] == "lifecycle_stage" for i in rep.items))
    items = sla([dict(base, request_id="k", received_at="2026-08-10T14:00:00Z", assigned_at="2026-08-10T14:40:00Z",
                      last_assigned_at="2026-08-10T14:40:00Z", first_response_at="2026-08-10T14:45:00Z")],
                extra={"classes": {"hr": {"target_business_minutes": 15}, "cw": {"target_business_minutes": 60}}})
    check("an owner assigned after the class limit warns S07 and S07 does not pass",
          any(i["rule"] == "S07" and i["severity"] == "WARN" and "assigned after 15 business minutes: k 40" in i["detail"]
              for i in items) and not any(i["rule"] == "S07" and i["severity"] == "PASS" for i in items))
    check("a configured class with no rows prints INFO, not PASS",
          any(i["rule"] == "S01" and i["scope"] == "cw" and i["severity"] == "INFO" for i in items)
          and not any(i["scope"] == "cw" and i["severity"] != "INFO" for i in items))
    split = Calendar({"timezone": "America/New_York", "hours": {"mon": [["09:00", "12:00"], ["13:00", "17:00"]]}})
    check("a day with two spans skips the break", split.minutes(t("2026-08-10T15:00:00Z"), t("2026-08-10T18:00:00Z")) == 120)
    night = Calendar({"timezone": "America/New_York",
                      "hours": {k: [["00:00", "06:00"], ["22:00", "24:00"]] for k in DAY_KEYS}})
    check("an overnight shift as two spans, with 24:00 as midnight",
          night.minutes(t("2026-08-11T03:00:00Z"), t("2026-08-11T05:00:00Z")) == 120)
    for hours, what in (({"mon": [["09:00", "13:00"], ["12:00", "17:00"]]}, "overlapping spans are"),
                        ({"mon": ["24:00", "17:00"]}, "24:00 as an opening time is")):
        try:
            Calendar({"timezone": "America/New_York", "hours": hours})
            rejected = False
        except InputError:
            rejected = True
        check("%s bad input" % what, rejected)
    with tempfile.TemporaryDirectory() as tmp:
        config, transitions, requests = (os.path.join(tmp, n) for n in ("c.json", "t.csv", "r.csv"))
        with open(config, "w") as fh:
            json.dump({"calendar": cal_cfg, "classes": {"hr": {"target_business_minutes": 15}}}, fh)
        with open(transitions, "w", newline="") as fh:
            csv.writer(fh).writerow(TRANSITION_COLUMNS)
        with open(requests, "w", newline="") as fh:
            csv.writer(fh).writerow(["request_id", "request_class", "received_at"])
        errors = io.StringIO()
        with contextlib.redirect_stderr(errors):
            codes = (main(["transitions", "--config", config, "--transitions", transitions, "--as-of", "2026-09-30"]),
                     main(["sla", "--config", config, "--requests", requests, "--as-of", "2026-09-30"]))
        check("header-only transitions and requests files exit 2",
              codes == (2, 2) and "no rows to audit" in errors.getvalue() and "no requests to audit" in errors.getvalue())
        later = os.path.join(tmp, "later.csv")
        with open(later, "w", newline="") as fh:
            csv.writer(fh).writerows([["request_id", "request_class", "received_at", "crm_seen_at", "first_response_at",
                                       "last_assigned_at", "excluded_reason"],
                                      ["l-1", "hr", "2026-10-05T14:00:00Z", "", "", "", ""],
                                      ["l-2", "hr", "2026-09-01T14:00:00Z", "", "", "", "spam"]])
        errors = io.StringIO()
        with contextlib.redirect_stderr(errors):
            code = main(["sla", "--config", config, "--requests", later, "--as-of", "2026-09-30T20:00:00Z"])
        check("requests all after --as-of or excluded exit 2 instead of passing",
              code == 2 and "no eligible requests" in errors.getvalue())
        with open(transitions, "a", newline="") as fh:
            csv.writer(fh).writerow(["c-1", "contacts", "owner", "rep.b@example.com", "2026-07-01T10:00:00Z",
                                     "rep.a@example.com"])
        got, _, masked = load_transitions(transitions, t("2026-09-30T20:00:00Z"))
        check("email-like sources and values become the UNMAPPED_USER labels normalize_history.py writes",
              masked == 2 and "@" not in got[0]["value"] + got[0]["source"]
              and got[0]["source"] == "UNMAPPED_USER:" + hashlib.sha256(b"rep.a@example.com").hexdigest()[:8])
        errors = io.StringIO()
        with contextlib.redirect_stderr(errors):
            code = main(["transitions", "--config", config, "--transitions", transitions,
                         "--as-of", "2026-09-30T20:00:00Z", "--since", "2026-09-01"])
        check("a --since after every change exits 2 instead of passing",
              code == 2 and "nothing would be audited" in errors.getvalue())
    failed = [name for name, ok in checks if not ok]
    for name in failed:
        print("FAILED: %s" % name)
    print("self-test %s (%d checks)" % ("passed" if not failed else "failed", len(checks)))
    return 1 if failed else 0


# ---------- main ----------

def main(argv):
    if argv[:1] == ["self-test"]:
        return self_test()
    parser = argparse.ArgumentParser(description="Read-only revenue workflow audit.")
    parser.add_argument("mode", choices=["transitions", "sla"])
    parser.add_argument("--config", required=True)
    parser.add_argument("--transitions")
    parser.add_argument("--requests")
    parser.add_argument("--as-of", required=True, help="ISO 8601 time, e.g. 2026-09-30T20:00:00Z")
    parser.add_argument("--since", help="YYYY-MM-DD: report changes on or after this date only")
    parser.add_argument("--json", action="store_true")
    args = parser.parse_args(argv)
    try:
        as_of = parse_time(args.as_of)
        if as_of is None:
            raise InputError("--as-of %r is not an ISO 8601 time" % args.as_of)
        since = parse_time(args.since) if args.since else None
        if args.since and since is None:
            raise InputError("--since %r is not a date" % args.since)
        cfg = load_config(args.config)
        report = Report()
        if args.mode == "transitions":
            if not args.transitions:
                raise InputError("transitions mode needs --transitions FILE")
            rows, skipped, masked = load_transitions(args.transitions, as_of)
            if not rows:
                raise InputError("no rows to audit in %s (%s)" % (os.path.basename(args.transitions), ", ".join(
                    "%d %s" % (n, k) for k, n in skipped.items()) or "the file has a header only"))
            if since is not None and not any(r["time"] >= since for r in rows):
                raise InputError("no change in %s on or after --since %s, so nothing would be audited"
                                 % (os.path.basename(args.transitions), args.since))
            histories = audit_transitions(cfg, rows, skipped, as_of, since, report)
            header = ["Transitions: %s, %d rows used (%s), %d records, %d fields. As of %s%s." % (
                os.path.basename(args.transitions), len(rows),
                ", ".join("%d %s" % (n, k) for k, n in skipped.items()) or "none skipped",
                len(set(r["record"] for r in rows)), len(set(f for f, _ in histories)), args.as_of,
                ", since %s" % args.since if args.since else "")]
            if masked:
                header.append("Note: %d email-like sources or values were replaced with UNMAPPED_USER labels; map "
                              "them to role labels with normalize_history.py --source-map or --value-map." % masked)
        else:
            if not args.requests:
                raise InputError("sla mode needs --requests FILE")
            rows, columns = load_requests(args.requests)
            if not rows:
                raise InputError("no requests to audit in %s (the file has a header only)"
                                 % os.path.basename(args.requests))
            notes = audit_sla(cfg, rows, columns, as_of, report)
            if not report.audited and not report.counts()[0]:
                raise InputError("no eligible requests to audit in %s (every row is after --as-of or excluded)"
                                 % os.path.basename(args.requests))
            header = ["Requests: %s, %d rows, classes %s. As of %s." % (
                os.path.basename(args.requests), len(rows),
                ", ".join(sorted(set(r["request_class"] for r in rows))), args.as_of)]
            header += notes
    except InputError as exc:
        print("error: %s" % exc, file=sys.stderr)
        return 2
    report.render(header, args.json)
    return 1 if report.counts()[0] else 0


if __name__ == "__main__":
    try:
        code = main(sys.argv[1:])
    except Exception as exc:  # malformed input that slipped past validation: never exit 1, which means FAIL
        print("error: could not process the input (%s: %s)" % (type(exc).__name__, exc), file=sys.stderr)
        code = 2
    sys.exit(code)