Scale the blocker window by lane risk, converging with risk-nexus
All checks were successful
CI Smoke / host-smoke (push) Successful in 0s
CI Smoke / container-smoke (push) Successful in 1s

risk-nexus accepted the offer to match their convention rather than grow a second
one, and published it: 14d critical/high, 30d medium, 60d low, nothing auto-closing
on staleness alone. Their preference — point warden route gaps at those windows and
the two registers agree without a shared mechanism — is better than a joint tool.

blocker_stale_days() now maps lane risk onto those windows. A flat 14 would have
been wrong in both directions: too aggressive for a low-risk pointer, and it treated
an admin PAT lane the same as one.

ungraded takes the shortest window, not the longest. ADR-0007 makes an absent grade
a defect and ADR-0008 makes a grade cover the whole path, so a lane nobody graded is
the one whose blocker deserves least trust. Encoding that as 60 days would have been
the fail-open default this repo already fixed once.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
This commit is contained in:
tegwick 2026-08-21 13:29:29 +02:00
parent 55f0f47a02
commit e18894ee5b
7 changed files with 96 additions and 24 deletions

View file

@ -710,7 +710,7 @@ def route_list(
bool, typer.Option("--stale", help="Show entries past review cadence (see --stale-days)")
] = False,
stale_days: Annotated[
int,
Optional[int],
typer.Option(
"--stale-days",
help="Days since reviewed before an entry is stale (default 90)",
@ -771,10 +771,10 @@ def route_list(
)
from warden.routing.catalog import DEFAULT_BLOCKER_STALE_DAYS
from warden.routing.catalog import blocker_stale_days
def _gap_is_stale(delegation, reviewed: str, stale_days: int) -> bool:
def _gap_is_stale(entry, delegation, reviewed: str, stale_days) -> bool:
"""An interim lane needs attention on either of two independent grounds.
Age is the obvious one. The other is that the review was never a
@ -784,7 +784,7 @@ def _gap_is_stale(delegation, reviewed: str, stale_days: int) -> bool:
"""
from warden.routing.catalog import days_since_review
if days_since_review(reviewed) > stale_days:
if days_since_review(reviewed) > blocker_stale_days(entry.risk, stale_days):
return True
return delegation.verified is not None and not delegation.is_verified
@ -797,11 +797,11 @@ def route_gaps(
int,
typer.Option(
"--stale-days",
help="Days since a blocker was verified before an interim lane is stale "
"(default 14 — see DEFAULT_BLOCKER_STALE_DAYS)",
help="Override the risk-scaled blocker window (default: 14d high/ungraded, "
"30d standard, 60d low — matches risk-nexus stall windows)",
min=1,
),
] = DEFAULT_BLOCKER_STALE_DAYS,
] = None,
fail_on_stale: Annotated[
bool,
typer.Option(
@ -834,7 +834,8 @@ def route_gaps(
"verified": d.verified,
"is_verified": d.is_verified,
"implicit": d.implicit,
"stale": _gap_is_stale(d, reviewed, stale_days),
"window_days": blocker_stale_days(e.risk, stale_days),
"stale": _gap_is_stale(e, d, reviewed, stale_days),
}
)
print(json.dumps(payload, indent=2))
@ -858,7 +859,7 @@ def route_gaps(
d = e.effective_delegation
reviewed = d.reviewed or e.reviewed
days = days_since_review(reviewed)
stale = _gap_is_stale(d, reviewed, stale_days)
stale = _gap_is_stale(e, d, reviewed, stale_days)
reviewed_styled = f"[yellow]{reviewed}[/yellow]" if stale else reviewed
days_styled = f"[yellow]{days}[/yellow]" if stale else str(days)
verified_styled = (
@ -879,6 +880,7 @@ def route_gaps(
stale_entries = [
e for e in entries
if _gap_is_stale(
e,
e.effective_delegation,
e.effective_delegation.reviewed or e.reviewed,
stale_days,
@ -890,12 +892,13 @@ def route_gaps(
# asked-and-waiting lane reads as reviewed.
aged = [
e for e in stale_entries
if days_since_review(e.effective_delegation.reviewed or e.reviewed) > stale_days
if days_since_review(e.effective_delegation.reviewed or e.reviewed)
> blocker_stale_days(e.risk, stale_days)
]
unverified = [e for e in stale_entries if e not in aged]
if aged:
console.print(
f"[yellow]{len(aged)} interim lane(s) past the {stale_days}d blocker "
f"[yellow]{len(aged)} interim lane(s) past their blocker "
f"cadence — re-check the blocker, do not just bump the date.[/yellow]"
)
if unverified:

View file

@ -87,6 +87,29 @@ DEFAULT_STALE_DAYS = 90
# ~15 interim lanes, surfaces about one lane a day rather than a wall of them.
DEFAULT_BLOCKER_STALE_DAYS = 14
# Scaled by the lane's own risk grade, matching risk-nexus's stall windows
# (14d critical/high, 30d medium, 60d low — docs/method/check-procedure.md).
# They offered the convention rather than a joint tool: point `warden route gaps`
# at the same windows and the two registers agree without a shared mechanism.
#
# `ungraded` gets the shortest window, not the longest. ADR-0007 already decided
# an absent grade is a defect and ADR-0008 that a grade covers the whole path;
# a lane nobody has graded is exactly the one whose blocker is least trustworthy.
BLOCKER_STALE_DAYS_BY_RISK = {
"high": 14,
"ungraded": 14,
"standard": 30,
"accepted": 60,
"low": 60,
}
def blocker_stale_days(risk: Optional[str], override: Optional[int] = None) -> int:
"""Days a lane's blocker may go unverified, scaled by what the lane holds."""
if override is not None:
return override
return BLOCKER_STALE_DAYS_BY_RISK.get(risk or "ungraded", DEFAULT_BLOCKER_STALE_DAYS)
def days_since_review(reviewed: str, *, today: Optional[date] = None) -> int:
"""Calendar days between reviewed date (YYYY-MM-DD) and today."""
@ -229,12 +252,16 @@ class Catalog:
def stale_gaps(
self,
include_draft: bool = False,
threshold_days: int = DEFAULT_BLOCKER_STALE_DAYS,
threshold_days: Optional[int] = None,
*,
today: Optional[date] = None,
) -> List[RouteEntry]:
"""Interim lanes whose blocker is due a re-check.
The window scales with the lane's risk grade unless `threshold_days`
overrides it -- a blocker on a lane holding an admin PAT should not go
unverified as long as one on a low-risk pointer.
A lane counts as stale when its review date is past the threshold **or**
when the review was never a verification at all. An `asked-and-waiting`
entry is the case that motivated this: it looks freshly reviewed on the
@ -244,7 +271,8 @@ class Catalog:
for e in self.gaps(include_draft=include_draft):
d = e.effective_delegation
reviewed = d.reviewed or e.reviewed
if is_review_stale(reviewed, threshold_days=threshold_days, today=today):
window = blocker_stale_days(e.risk, threshold_days)
if is_review_stale(reviewed, threshold_days=window, today=today):
out.append(e)
elif d.verified is not None and not d.is_verified:
out.append(e)
@ -287,14 +315,12 @@ class Catalog:
)
# Interim blockers run on their own, much shorter cadence -- a stale
# pointer and an unanswered blocker are not the same kind of drift.
stale_interim = len(self.stale_gaps(
include_draft=True, threshold_days=DEFAULT_BLOCKER_STALE_DAYS, today=today
))
stale_interim = len(self.stale_gaps(include_draft=True, today=today))
if stale_interim:
warnings.append(
f"{stale_interim} interim delegation"
f"{'' if stale_interim == 1 else 's'} need re-verifying "
f"({DEFAULT_BLOCKER_STALE_DAYS}d blocker cadence) — see `warden route gaps`"
f"(risk-scaled blocker cadence) — see `warden route gaps`"
)
return CatalogFreshness(