risk-nexus/tools/fix_tracker.py

#!/usr/bin/env python3
"""Read the state of the fixes this register claims to be tracking.

RISK-WP-0005-T01. Until now `fix_tracking` held ids like `FLEX-WP-0015-T02`
that this repo wrote down and never read, so an owner who had gone quiet and a
fix that had gone quiet were indistinguishable.

Resolution is against the owning repo's workplan **file**, which is the
ADR-001 source of truth, and the file's last commit date is the honest answer
to "has this moved" — independent of whether this register happened to look.
"""
from __future__ import annotations

import datetime as dt
import pathlib
import re
import subprocess

import register_lib as lib

ESTATE = pathlib.Path("/home/worsch")
WP_ID = re.compile(r"\b([A-Z]+-WP-[0-9]{4})(-T[0-9]{2,3})?\b")

# How long a fix may sit unchanged before the register says so, by severity.
STALL_DAYS = {"critical": 14, "high": 14, "medium": 30, "low": 60}


def find_workplan(wp_id: str) -> pathlib.Path | None:
    hits = sorted(ESTATE.glob(f"*/workplans/{wp_id}-*.md"))
    if hits:
        return hits[0]
    # Archived workplans are still the record. A fix that finished and was
    # archived must not read as "not found in the estate" — that is the
    # register inventing a gap out of a filing convention.
    archived = sorted(ESTATE.glob(f"*/workplans/archived/*{wp_id}-*.md"))
    return archived[0] if archived else None


def last_commit(path: pathlib.Path) -> dt.datetime | None:
    try:
        out = subprocess.run(
            ["git", "-C", str(path.parent.parent), "log", "-1", "--format=%cI", "--", str(path)],
            capture_output=True, text=True, timeout=15,
        ).stdout.strip()
        return lib.moment(out) if out else None
    except (subprocess.SubprocessError, OSError):
        return None


def task_status(path: pathlib.Path, task_id: str) -> str | None:
    m = re.search(rf"id: {re.escape(task_id)}\s*\nstatus: (\w+)", path.read_text(encoding="utf-8"))
    return m.group(1) if m else None


def resolve(fix_tracking: str) -> list[dict]:
    """One entry per work-record id named in the finding's fix_tracking."""
    out = []
    for wp, task in WP_ID.findall(str(fix_tracking or "")):
        path = find_workplan(wp)
        if path is None:
            out.append({"id": wp + (task or ""), "state": "not found in the estate", "moved": None})
            continue
        fm = lib.load(path)
        archived = "archived" in path.parts
        full = wp + task if task else wp
        state = task_status(path, full) if task else fm.get("status")
        out.append({
            "id": full,
            "repo": fm.get("repo", path.parent.parent.name),
            "state": (state or "unreadable") + (" [archived]" if archived else ""),
            "completed": str(state).lower() in ("done", "finished", "cancel"),
            "workplan_status": fm.get("status"),
            "moved": last_commit(path),
        })
    return out


def report() -> list[str]:
    lines, now = [], lib.now()
    for f in lib.findings():
        if not lib.watched(f.get("status")):
            continue
        tracking = f.get("fix_tracking")
        if not tracking or str(tracking).startswith("unset"):
            if f.get("status") == "accepted":
                missing = [key for key in ("accepted_by", "accepted_until", "determination")
                           if not f.get(key) or str(f[key]) == "unset"]
                if lib.moment(f.get("next_check")) is None:
                    missing.append("next_check")
                determinations = {r.get("id") for r in lib.regulatory()}
                if f.get("determination") and f["determination"] not in determinations:
                    missing.append("resolvable determination")
                if not missing:
                    lines.append(
                        f"{f['id']} — RECORDED ACCEPTANCE by {f['accepted_by']}; "
                        f"ends when: {f['accepted_until']}; basis {f['determination']}; "
                        f"review {f['next_check']} (not a fix or renewed acceptance)"
                    )
                    continue
                lines.append(f"{f['id']} — ACCEPTANCE INCOMPLETE: {', '.join(missing)}")
            lines.append(f"{f['id']} ({f.get('severity')}) — NO FIX TRACKING; owner {f.get('fix_owner')}")
            continue
        resolved = resolve(tracking)
        if not resolved:
            # An id the tracker cannot parse is not the same as no tracking, and
            # silently printing neither is how a gap hides.
            lines.append(f"{f['id']} — fix_tracking '{tracking}' names no workplan this tool can resolve")
            continue
        for r in resolved:
            if r["moved"]:
                age = (now - r["moved"]).days
                window = STALL_DAYS.get(f.get("severity"), 30)
                done = r.get("completed", False)
                flag = "" if done or age <= window else f"  ← UNCHANGED {age}d (window {window}d)"
                lines.append(f"{f['id']} — {r['id']}: {r['state']}, last changed {r['moved']:%Y-%m-%d}{flag}")
            else:
                lines.append(f"{f['id']} — {r['id']}: {r['state']}")
    return lines


if __name__ == "__main__":
    print("Fix state, read from the owning repo's workplan file\n")
    for line in report():
        print(" ", line)