risk-nexus/tools/register_check.py

#!/usr/bin/env python3
"""Report what the register is not saying out loud.

RISK-WP-0001-T08. Report only: it writes nothing, changes no field,
escalates nothing on its own, and blocks nothing. A human or the custodian
acts on the output. Exit code is 0 unless a finding cannot be read.
"""
from __future__ import annotations

import datetime as dt

import register_lib as lib

NOW = lib.now()


def main() -> int:
    fs = [f for f in lib.findings() if lib.watched(f.get("status"))]
    lines: list[str] = []

    def section(title: str, rows: list[str], quiet: str) -> None:
        lines.append(f"{title}:")
        if rows:
            lines.extend(f"  {r}" for r in rows)
        else:
            lines.append(f"  {quiet}")
        lines.append("")

    seen: dict[str, str] = {}
    dupes = []
    for f in lib.findings():
        prior = seen.get(f["id"])
        if prior:
            dupes.append(f"{f['id']} — filed twice: {prior} and {f['_path'].name}; renumber the later commit")
        seen[f["id"]] = f["_path"].name
    if dupes:
        section("Duplicate ids", dupes, "none")

    unknown = [
        f"{f['id']} — status '{f.get('status')}' is not one of {', '.join(lib.KNOWN_STATUSES)}; watched anyway"
        for f in fs
        if f.get("status") not in lib.KNOWN_STATUSES
    ]
    if unknown:
        section("Unrecognised status", unknown, "none")

    ungraded = [
        f"{f['id']} — {', '.join(k for k in lib.GRADED_FIELDS if str(f.get(k, 'unset')) == 'unset')}"
        for f in fs
        if any(str(f.get(k, "unset")) == "unset" for k in lib.GRADED_FIELDS)
    ]
    section("Ungraded", ungraded, "none — every watched finding carries a grade")

    due, deferred = [], []
    for f in fs:
        if f.get("deferred_to"):
            until = lib.moment(f["deferred_to"])
            if until and NOW < until:
                deferred.append(f"{f['id']} — deferred by the operator until {f['deferred_to']}")
                continue
        when = lib.moment(f.get("next_check"))
        if when is None:
            due.append(f"{f['id']} — no next_check set")
        elif NOW >= when:
            late = NOW - when
            hours = int(late.total_seconds() // 3600)
            rung = f.get("cadence", "instant")
            due.append(
                f"{f['id']} ({f.get('severity')}) — due {f['next_check']}"
                f"{f', {hours}h late' if hours else ''}, cadence {rung}"
            )
    section("Checks due", due, "none")
    if deferred:
        section("Deferred by the operator", deferred, "none")

    # The rung is the signal: how long this matter has held still.
    stability = []
    for f in sorted(fs, key=lambda f: lib.rung_index(f.get("cadence", "instant"))):
        rung = f.get("cadence", "instant")
        streak = f.get("clean_streak", 0)
        note = " — at the ceiling" if rung == lib.TOP_RUNG else ""
        stability.append(f"{f['id']}: {rung} ({streak} clean in a row){note}")
    section("Stability — where each finding sits on the ladder", stability, "nothing watched")

    bottom = [
        f"{f['id']} ({f.get('severity')}) — still at the bottom rung since {f.get('last_checked')}"
        for f in fs
        if lib.rung_index(f.get("cadence", "instant")) == 0
        and (m := lib.moment(f.get("last_checked")))
        and NOW - m > dt.timedelta(days=14)
    ]
    section("Stalled — escalation trigger 5", bottom, "none")

    embargo = [
        f"{f['id']} — lifts when: {f.get('embargo_condition')}"
        for f in fs
        if f.get("disclosure") == "embargoed"
    ]
    section("Embargoed", embargo, "none")

    esc = [
        f"{f['id']} — trigger {f.get('escalation_trigger')}, {f.get('escalation_status')}"
        for f in fs
        if f.get("escalation") == "required" and f.get("escalation_status") != "answered"
    ]
    section("Escalations awaiting the operator", esc, "none")

    rescore = [
        f"{f['id']} — {f.get('severity')} now, {f.get('severity_at_production')} at production"
        for f in fs
        if f.get("production_rescore")
    ]
    section("Owed at the production transition", rescore, "none — no finding is graded lower for build mode")

    print(f"Register check — {NOW:%Y-%m-%d %H:%MZ}\n{len(fs)} live finding(s)\n")
    print("\n".join(lines).rstrip())
    return 0


if __name__ == "__main__":
    raise SystemExit(main())