RISK-WP-0005-T01: read the fix records, and two findings moved
fix_tracker.py resolves fix_tracking against the owning repo's workplan file — the ADR-001 source of truth — and uses the file's last commit date as the honest answer to 'has this moved', independent of whether the register looked. Archived workplans are searched too, so a finished fix that was filed away does not read as missing. First run, three findings it should have known about: RISK-F-0005 — AUDIT-WP-0008-T04 has read done since 2026-08-18. The fix this finding asked for has landed and the register spent three days not knowing. Now mitigated, embargo lifted, disclosure public. Not fixed: that needs a probe, and T05's adversarial evidence artifact still reads wait. RISK-F-0002 — both tracked records were closed before the finding was filed: WARDEN-WP-0007 archived 2026-07-08, FLEX-WP-0007 finished 2026-06-29, against a finding of 2026-08-18 that names FLEX-WP-0007 as the blocker. Routed as a question, not a conclusion. Four findings carry no fix tracking at all, which the report now says out loud rather than leaving as an empty field. Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
This commit is contained in:
parent
0ff87c22d8
commit
ba3b3f686d
7 changed files with 217 additions and 13 deletions
101
tools/fix_tracker.py
Normal file
101
tools/fix_tracker.py
Normal file
|
|
@ -0,0 +1,101 @@
|
|||
#!/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
|
||||
for r in resolve(tracking):
|
||||
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)
|
||||
|
|
@ -9,6 +9,7 @@ from __future__ import annotations
|
|||
|
||||
import datetime as dt
|
||||
|
||||
import fix_tracker
|
||||
import register_lib as lib
|
||||
|
||||
NOW = lib.now()
|
||||
|
|
@ -143,6 +144,9 @@ def main() -> int:
|
|||
if ref in waiting_ids and f.get("waiting_on"):
|
||||
deep.append(f"{f['id']} waits, and points at {ref} which also waits — depth two, cut one")
|
||||
section("Waiting on someone", waiting, "nothing is waiting on anyone")
|
||||
# RISK-WP-0005-T01: the fix's own state, read from the owner's workplan
|
||||
# file rather than from our memory of what they told us.
|
||||
section("Fix state", fix_tracker.report(), "no finding claims a tracked fix")
|
||||
if due_defaults:
|
||||
section("Defaults now due — apply them", due_defaults, "none")
|
||||
if deep:
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue