tegwick
623f7ccb2e
RISK-F-0002's fix_tracking now names ZONE-WP-0001 / WARDEN-WP-0032, with the superseded pair kept beside it — the register had told ops-warden the path had changed and then left the old ids in the file. An unresolvable fix_tracking (TEN-IN-0002 is an intake, not a workplan) printed nothing at all, which reads identically to a finding with no tracking. It now says so by name. Both ZONE-WP-0001 and WARDEN-WP-0032 read proposed, which is the true state of the successor. Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
107 lines
4.2 KiB
Python
107 lines
4.2 KiB
Python
#!/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 ""),
|
|
"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"):
|
|
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 = str(r["state"]).lower() in ("done", "finished", "cancel")
|
|
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)
|