Ship WARDEN-WP-0030: delegation register for every catalog lane
All checks were successful
CI Smoke / host-smoke (push) Successful in 0s
CI Smoke / container-smoke (push) Successful in 1s

Every execution position is now explicit. Catalog entries carry
delegation.mode (permanent / native / interim) with intended owner and
blocker. warden route gaps lists the interim set. Promotion requires
the ownership question. Doctrine lives in AccessRouting.md; the
register was published to the named owner repos.
This commit is contained in:
tegwick 2026-08-15 20:54:58 +02:00
parent 8d3706fa06
commit c93e3c9b43
11 changed files with 619 additions and 29 deletions

View file

@ -12,12 +12,13 @@ from warden.routing.catalog import (
find_catalog_path,
load_catalog,
)
from warden.routing.models import RouteEntry
from warden.routing.models import Delegation, RouteEntry
__all__ = [
"Catalog",
"CatalogError",
"CatalogFreshness",
"Delegation",
"RouteEntry",
"find_catalog_path",
"load_catalog",

View file

@ -23,7 +23,13 @@ from typing import List, Optional
import yaml
from warden.routing.models import VALID_RISK, RotationGuide, RouteEntry
from warden.routing.models import (
VALID_DELEGATION_MODES,
VALID_RISK,
Delegation,
RotationGuide,
RouteEntry,
)
# Structured handoff string fields (WP-0014) — templates and pointers only.
# Every one is scanned for accidental secret material; see _assert_no_secret_material.
@ -201,6 +207,25 @@ class Catalog:
if is_review_stale(e.reviewed, threshold_days=threshold_days, today=today)
]
def gaps(self, include_draft: bool = False) -> List[RouteEntry]:
"""Interim lanes — the queryable delegation register (WARDEN-WP-0030)."""
return [e for e in self.listed(include_draft=include_draft) if e.is_interim]
def stale_gaps(
self,
include_draft: bool = False,
threshold_days: int = DEFAULT_STALE_DAYS,
*,
today: Optional[date] = None,
) -> List[RouteEntry]:
"""Interim lanes whose delegation review is past the cadence threshold."""
out: List[RouteEntry] = []
for e in self.gaps(include_draft=include_draft):
reviewed = e.effective_delegation.reviewed or e.reviewed
if is_review_stale(reviewed, threshold_days=threshold_days, today=today):
out.append(e)
return out
def freshness(
self,
*,
@ -236,6 +261,15 @@ class Catalog:
f"{stale_count} catalog entr{'y' if stale_count == 1 else 'ies'} "
f"past {stale_threshold_days}d review cadence"
)
stale_interim = len(self.stale_gaps(
include_draft=True, threshold_days=stale_threshold_days, today=today
))
if stale_interim:
warnings.append(
f"{stale_interim} interim delegation"
f"{'' if stale_interim == 1 else 's'} past "
f"{stale_threshold_days}d review — see `warden route gaps`"
)
return CatalogFreshness(
path=str(path),
@ -362,6 +396,66 @@ def _parse_rotation(entry_id: str, raw: Optional[dict]) -> Optional[RotationGuid
)
def _parse_delegation(entry_id: str, raw: Optional[dict]) -> Optional[Delegation]:
"""Parse an optional ``delegation:`` block (WARDEN-WP-0030).
Absence is allowed: the loader treats it as implicit interim with an
unknown owner. When the block *is* present, mode / owner / blocker rules
are enforced so a declared answer cannot be incomplete.
"""
if raw is None:
return None
if not isinstance(raw, dict):
raise CatalogError(f"entry {entry_id!r} `delegation` must be a mapping")
mode = str(raw.get("mode", "")).strip()
if mode not in VALID_DELEGATION_MODES:
raise CatalogError(
f"entry {entry_id!r} delegation.mode {mode!r} invalid "
f"(expected one of {VALID_DELEGATION_MODES})"
)
intended_owner = str(raw.get("intended_owner", "")).strip() or None
if mode != "permanent" and not intended_owner:
raise CatalogError(
f"entry {entry_id!r} delegation.intended_owner is required "
f"unless mode is permanent"
)
blocked_on = str(raw.get("blocked_on", "")).strip() or None
if mode == "interim" and not blocked_on:
raise CatalogError(
f"entry {entry_id!r} delegation.blocked_on is required when mode is interim"
)
reviewed = str(raw.get("reviewed", "")).strip() or None
if not reviewed:
raise CatalogError(f"entry {entry_id!r} delegation.reviewed is required")
try:
date.fromisoformat(reviewed)
except ValueError as e:
raise CatalogError(
f"entry {entry_id!r} delegation.reviewed {reviewed!r} is not YYYY-MM-DD"
) from e
if intended_owner:
_assert_no_secret_material(
entry_id, "delegation.intended_owner", intended_owner, prose=True
)
if blocked_on:
_assert_no_secret_material(
entry_id, "delegation.blocked_on", blocked_on, prose=True
)
return Delegation(
mode=mode,
intended_owner=intended_owner,
blocked_on=blocked_on,
reviewed=reviewed,
implicit=False,
)
def _parse_entry(raw: dict, index: int) -> RouteEntry:
if not isinstance(raw, dict):
raise CatalogError(f"entry #{index} is not a mapping")
@ -452,6 +546,7 @@ def _parse_entry(raw: dict, index: int) -> RouteEntry:
pointer_command=handoff["pointer_command"],
rotation=_parse_rotation(entry_id, raw.get("rotation")),
risk=risk,
delegation=_parse_delegation(entry_id, raw.get("delegation")),
)

View file

@ -34,6 +34,40 @@ class RotationGuide:
# standard — ordinary workload secrets (ESO-fed, non-escrow); normal least-privilege.
VALID_RISK = ("standard", "high")
# Delegation modes (WARDEN-WP-0030). Absence of a block is implicit interim.
# native — intended owner already fronts the lane (route-primary / pointer)
# interim — ops-warden covers a gap; intended_owner + blocked_on required
# permanent — ops-warden is the designed owner of this front door (SSH only today)
VALID_DELEGATION_MODES = ("native", "interim", "permanent")
IMPLICIT_DELEGATION_BLOCKED_ON = (
"unclassified — no delegation block; treat as a question, not a settlement"
)
@dataclass
class Delegation:
"""Who should own this front door, and what is missing (WARDEN-WP-0030).
Pointer-layer only: names the intended owner and the blocker. Does not
restate how that owner will implement their front door.
"""
mode: str # native | interim | permanent
intended_owner: Optional[str] = None
blocked_on: Optional[str] = None
reviewed: Optional[str] = None
implicit: bool = False
def to_dict(self) -> dict:
return {
"mode": self.mode,
"intended_owner": self.intended_owner,
"blocked_on": self.blocked_on,
"reviewed": self.reviewed,
"implicit": self.implicit,
}
@dataclass
class RouteEntry:
@ -78,6 +112,8 @@ class RouteEntry:
rotation: Optional[RotationGuide] = None
# Agent read-boundary risk class (WP-0026 T04). high → agents use wrap/out/exec only.
risk: str = "standard" # "standard" | "high"
# Delegation register (WP-0030). None → implicit interim with unknown owner.
delegation: Optional[Delegation] = None
@property
def is_active(self) -> bool:
@ -111,6 +147,28 @@ class RouteEntry:
"""True when an owner-native exec front door is the primary path for this lane."""
return bool(self.exec_owner and self.exec_command)
@property
def effective_delegation(self) -> Delegation:
"""Declared delegation, or implicit interim with an unknown owner.
Absence of a ``delegation:`` block is a question (WP-0030), not a
settlement that ops-warden owns the front door.
"""
if self.delegation is not None:
return self.delegation
return Delegation(
mode="interim",
intended_owner=None,
blocked_on=IMPLICIT_DELEGATION_BLOCKED_ON,
reviewed=None,
implicit=True,
)
@property
def is_interim(self) -> bool:
"""True when this lane is a tracked gap (explicit or implicit)."""
return self.effective_delegation.mode == "interim"
@property
def has_handoff(self) -> bool:
"""True when structured assist fields are present (advisory richness)."""