Adaptive check cadence: the interval is earned, not assigned
Operator ruling 2026-08-20. Severity no longer sets the review interval. A check that comes back clean climbs one rung — instant, 1h, 8h, 24h, 48h, 96h, 7d, 14d, 1mo, 1q — and anything wrong drops straight back to instant. A quarter is the ceiling. The operator may defer an instant finding to a stated date; that is the only other way off the bottom rung. The rung is the point: it says how stable the estate has been on that matter, which is information severity does not carry. Volatile things get attention automatically; quiet things stop consuming it; neither judgement has to be made by a person who might be busy. Escalation trigger 5 rebased onto the ladder — fourteen days at the bottom rung, whether that is failing checks or no checks. Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
This commit is contained in:
parent
8b204d0411
commit
42bbf5d2dc
17 changed files with 429 additions and 148 deletions
Binary file not shown.
|
|
@ -11,21 +11,11 @@ import datetime as dt
|
|||
|
||||
import register_lib as lib
|
||||
|
||||
TODAY = dt.date.today()
|
||||
|
||||
|
||||
def date(value):
|
||||
if isinstance(value, dt.date):
|
||||
return value
|
||||
if isinstance(value, str) and value not in ("", "unset"):
|
||||
return dt.date.fromisoformat(value)
|
||||
return None
|
||||
NOW = lib.now()
|
||||
|
||||
|
||||
def main() -> int:
|
||||
# "accepted" is deliberately carried, not closed: it keeps its review
|
||||
# interval and its production re-score, so the nag must keep watching it.
|
||||
fs = [f for f in lib.findings() if f.get("status") in ("open", "accepted")]
|
||||
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:
|
||||
|
|
@ -36,44 +26,67 @@ def main() -> int:
|
|||
lines.append(f" {quiet}")
|
||||
lines.append("")
|
||||
|
||||
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 open finding carries a grade")
|
||||
section("Ungraded", ungraded, "none — every watched finding carries a grade")
|
||||
|
||||
overdue = []
|
||||
due, deferred = [], []
|
||||
for f in fs:
|
||||
due = date(f.get("review_by"))
|
||||
if due and TODAY > due:
|
||||
overdue.append(f"{f['id']} — review was due {due} ({(TODAY - due).days}d ago)")
|
||||
section("Overdue review", overdue, "none")
|
||||
|
||||
stale = []
|
||||
for f in fs:
|
||||
seen = date(f.get("last_reviewed"))
|
||||
interval = lib.REVIEW_INTERVAL_DAYS.get(f.get("severity"))
|
||||
if not seen or not interval or f.get("severity") == "low":
|
||||
continue
|
||||
limit = seen + dt.timedelta(days=interval * lib.STALE_MULTIPLIER)
|
||||
if TODAY > limit:
|
||||
stale.append(
|
||||
f"{f['id']} ({f.get('severity')}) — untouched since {seen}; "
|
||||
f"escalation trigger 5 fires (limit was {limit})"
|
||||
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("Stalled — escalation trigger 5", stale, "none")
|
||||
section("Checks due", due, "none")
|
||||
if deferred:
|
||||
section("Deferred by the operator", deferred, "none")
|
||||
|
||||
embargo = []
|
||||
for f in fs:
|
||||
if f.get("disclosure") != "embargoed":
|
||||
continue
|
||||
due = date(f.get("embargo_review"))
|
||||
if due and TODAY > due:
|
||||
embargo.append(
|
||||
f"{f['id']} — embargo not re-decided since {due}; lifts when: {f.get('embargo_condition')}"
|
||||
)
|
||||
section("Embargo overdue for re-decision", embargo, "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')}"
|
||||
|
|
@ -89,7 +102,7 @@ def main() -> int:
|
|||
]
|
||||
section("Owed at the production transition", rescore, "none — no finding is graded lower for build mode")
|
||||
|
||||
print(f"Register check — {TODAY}\n{len(fs)} open or accepted finding(s)\n")
|
||||
print(f"Register check — {NOW:%Y-%m-%d %H:%MZ}\n{len(fs)} live finding(s)\n")
|
||||
print("\n".join(lines).rstrip())
|
||||
return 0
|
||||
|
||||
|
|
|
|||
|
|
@ -22,14 +22,19 @@ def date(value) -> dt.date | None:
|
|||
return None
|
||||
|
||||
|
||||
def review_cell(f: dict) -> str:
|
||||
due = date(f.get("review_by"))
|
||||
if due is None:
|
||||
def check_cell(f: dict) -> str:
|
||||
when = lib.moment(f.get("next_check"))
|
||||
if when is None:
|
||||
return "—"
|
||||
late = (TODAY - due).days
|
||||
if late > 0:
|
||||
return f"**{due} (overdue {late}d)**"
|
||||
return str(due)
|
||||
if lib.now() >= when:
|
||||
return "**due**"
|
||||
return when.strftime("%Y-%m-%d %H:%MZ")
|
||||
|
||||
|
||||
def cadence_cell(f: dict) -> str:
|
||||
rung = f.get("cadence", "instant")
|
||||
streak = f.get("clean_streak", 0)
|
||||
return f"{rung} ({streak})"
|
||||
|
||||
|
||||
def sev_cell(f: dict) -> str:
|
||||
|
|
@ -50,23 +55,23 @@ def esc_cell(f: dict) -> str:
|
|||
def render() -> str:
|
||||
fs = lib.findings()
|
||||
ns = lib.notes()
|
||||
open_fs = [f for f in fs if f.get("status") == "open"]
|
||||
open_fs = [f for f in fs if lib.watched(f.get("status"))]
|
||||
|
||||
out = [
|
||||
"# Register",
|
||||
"",
|
||||
f"Generated by `tools/register_index.py` from `findings/`. Do not edit by hand. Last built {TODAY}.",
|
||||
"",
|
||||
f"{len(open_fs)} open of {len(fs)} findings; {len(ns)} notes below the floor.",
|
||||
f"{len(open_fs)} live of {len(fs)} findings; {len(ns)} notes below the floor.",
|
||||
"",
|
||||
"## Findings",
|
||||
"",
|
||||
"| ID | Finding | System | Severity | Disclosure | Escalation | Fix owner | Status | Review by |",
|
||||
"| --- | --- | --- | --- | --- | --- | --- | --- | --- |",
|
||||
"| ID | Finding | System | Severity | Disclosure | Escalation | Fix owner | Status | Cadence | Next check |",
|
||||
"| --- | --- | --- | --- | --- | --- | --- | --- | --- | --- |",
|
||||
]
|
||||
for f in fs:
|
||||
out.append(
|
||||
"| [{id}]({path}) | {title} | {system} | {sev} | {disc} | {esc} | {owner} | {status} | {review} |".format(
|
||||
"| [{id}]({path}) | {title} | {system} | {sev} | {disc} | {esc} | {owner} | {status} | {cadence} | {check} |".format(
|
||||
id=f["id"],
|
||||
path=f["_path"].relative_to(lib.REPO),
|
||||
title=str(f.get("title", "")).strip('"'),
|
||||
|
|
@ -76,7 +81,8 @@ def render() -> str:
|
|||
esc=esc_cell(f),
|
||||
owner=f.get("fix_owner", "—"),
|
||||
status=f.get("status", "—"),
|
||||
review=review_cell(f),
|
||||
cadence=cadence_cell(f),
|
||||
check=check_cell(f),
|
||||
)
|
||||
)
|
||||
|
||||
|
|
@ -112,7 +118,9 @@ def render() -> str:
|
|||
|
||||
out += ["", "## How to read this", "",
|
||||
"Severity is `docs/method/severity.md`; disclosure `docs/method/disclosure.md`;",
|
||||
"escalation `docs/method/escalation.md`; review dates `docs/method/review.md`.",
|
||||
"escalation `docs/method/escalation.md`; the check cadence `docs/method/review.md`.",
|
||||
"Cadence is the ladder rung and the count of consecutive clean checks — a finding at `1q (9)`",
|
||||
"has held still for a long time; one at `instant (0)` moved recently. Anything wrong resets it.",
|
||||
"A constraint may be graded higher than the finding that carries it — read both.",
|
||||
""]
|
||||
return "\n".join(out)
|
||||
|
|
|
|||
|
|
@ -4,6 +4,7 @@ The finding files are the source of truth. Nothing here writes to them.
|
|||
"""
|
||||
from __future__ import annotations
|
||||
|
||||
import datetime as dt
|
||||
import pathlib
|
||||
import yaml
|
||||
|
||||
|
|
@ -12,9 +13,53 @@ FINDINGS = REPO / "findings"
|
|||
NOTES = REPO / "notes"
|
||||
|
||||
SEVERITIES = ["critical", "high", "medium", "low"]
|
||||
REVIEW_INTERVAL_DAYS = {"critical": 7, "high": 30, "medium": 90, "low": 180}
|
||||
STALE_MULTIPLIER = 2
|
||||
GRADED_FIELDS = ["severity", "disclosure", "escalation", "review_by", "last_reviewed"]
|
||||
GRADED_FIELDS = ["severity", "disclosure", "escalation", "next_check", "cadence"]
|
||||
# A finding leaves the watch list only by being genuinely finished. Anything
|
||||
# else — including a word the register has not seen before — stays watched, and
|
||||
# the unknown word is reported. A finding must never fall out of the nag because
|
||||
# somebody used a status the tooling did not recognise.
|
||||
CLOSED_STATUSES = ("fixed", "withdrawn")
|
||||
KNOWN_STATUSES = ("open", "accepted", "mitigated") + CLOSED_STATUSES
|
||||
|
||||
|
||||
def watched(status: str | None) -> bool:
|
||||
return status not in CLOSED_STATUSES
|
||||
|
||||
# The adaptive cadence ladder (operator ruling, 2026-08-20). A clean check
|
||||
# climbs one rung; anything wrong drops straight back to `instant`. The rung a
|
||||
# finding sits on is itself the signal: how stable this matter has been.
|
||||
CADENCE = [
|
||||
("instant", dt.timedelta(0)),
|
||||
("1h", dt.timedelta(hours=1)),
|
||||
("8h", dt.timedelta(hours=8)),
|
||||
("24h", dt.timedelta(hours=24)),
|
||||
("48h", dt.timedelta(hours=48)),
|
||||
("96h", dt.timedelta(hours=96)),
|
||||
("7d", dt.timedelta(days=7)),
|
||||
("14d", dt.timedelta(days=14)),
|
||||
("1mo", dt.timedelta(days=30)),
|
||||
("1q", dt.timedelta(days=90)),
|
||||
]
|
||||
CADENCE_NAMES = [name for name, _ in CADENCE]
|
||||
TOP_RUNG = CADENCE_NAMES[-1]
|
||||
|
||||
|
||||
def rung_index(name: str) -> int:
|
||||
return CADENCE_NAMES.index(name) if name in CADENCE_NAMES else 0
|
||||
|
||||
|
||||
def interval(name: str) -> dt.timedelta:
|
||||
return CADENCE[rung_index(name)][1]
|
||||
|
||||
|
||||
def climb(name: str) -> str:
|
||||
"""One clean check: up one rung, never past the quarter."""
|
||||
return CADENCE_NAMES[min(rung_index(name) + 1, len(CADENCE) - 1)]
|
||||
|
||||
|
||||
def reset() -> str:
|
||||
"""Anything wrong: back to the bottom."""
|
||||
return CADENCE_NAMES[0]
|
||||
|
||||
|
||||
def load(path: pathlib.Path) -> dict:
|
||||
|
|
@ -38,5 +83,20 @@ def notes() -> list[dict]:
|
|||
return sorted((load(p) for p in NOTES.glob("RISK-N-*.md")), key=lambda n: n["id"], reverse=True)
|
||||
|
||||
|
||||
def moment(value) -> dt.datetime | None:
|
||||
"""Parse a date or datetime front-matter value as UTC."""
|
||||
if isinstance(value, dt.datetime):
|
||||
return value if value.tzinfo else value.replace(tzinfo=dt.timezone.utc)
|
||||
if isinstance(value, dt.date):
|
||||
return dt.datetime.combine(value, dt.time(0, 0), tzinfo=dt.timezone.utc)
|
||||
if isinstance(value, str) and value not in ("", "unset"):
|
||||
return dt.datetime.fromisoformat(value.replace("Z", "+00:00")).astimezone(dt.timezone.utc)
|
||||
return None
|
||||
|
||||
|
||||
def now() -> dt.datetime:
|
||||
return dt.datetime.now(dt.timezone.utc)
|
||||
|
||||
|
||||
def sev_rank(sev: str) -> int:
|
||||
return SEVERITIES.index(sev) if sev in SEVERITIES else len(SEVERITIES)
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue