ops-warden/src/warden/routing/catalog.py

"""Load and validate the routing pointer catalog.

The catalog lives at ``registry/routing/catalog.yaml`` in the repo root. Resolution
order:

1. ``WARDEN_ROUTING_CATALOG`` env var, if set (used by tests / overrides).
2. Walk upward from this module looking for ``registry/routing/catalog.yaml``.

Validation enforces the **no-double-source rule**: only ``warden_executes: true``
entries may carry an authored ``steps`` block or a ``cert_command``. Any non-SSH
entry that does so is a validation error — ops-warden points at the owner's doc, it
never restates another subsystem's procedure.
"""
from __future__ import annotations

import hashlib
import os
import re
from dataclasses import dataclass, field
from datetime import date, datetime, timezone
from pathlib import Path
from typing import List, Optional

import yaml

from warden.routing.models import (
    VALID_DELEGATION_MODES,
    VALID_RISK,
    VALID_WORKLOAD_APPLICABILITY,
    Delegation,
    RotationGuide,
    RouteEntry,
    WorkloadReference,
)

# Structured handoff string fields (WP-0014) — templates and pointers only.
# Every one is scanned for accidental secret material; see _assert_no_secret_material.
_HANDOFF_STR_FIELDS = (
    "auth_method", "path_template", "fetch_command", "policy_ref",
    # Owner-native exec front door (WP-0019) — pointer commands, screened too.
    "exec_command", "pointer_command",
)

# Known secret-bearing token prefixes — a literal here means a value leaked into
# the catalog (which is git-tracked and agent-visible). Templates use `<...>`.
_SECRET_PREFIXES = (
    "ghp_", "gho_", "ghs_", "github_pat_",  # GitHub
    "sk-", "sk_live_", "sk_test_",           # OpenAI / Stripe
    "xoxb-", "xoxp-",                          # Slack
    "AKIA", "ASIA",                            # AWS access key ids
    "hvs.", "hvb.", "s.",                      # Vault/OpenBao service tokens
    "AIza",                                    # Google
    "eyJ",                                     # JWT
)
# A long unbroken high-entropy run that is not a placeholder — likely a raw value.
_HIGH_ENTROPY_RUN = re.compile(r"[A-Za-z0-9_\-]{32,}")

_REQUIRED_FIELDS = (
    "id",
    "title",
    "need_keywords",
    "owner_repo",
    "subsystem",
    "warden_executes",
    "wiki_ref",
    "canon_ref",
    "reviewed",
    "status",
    "workload_ref",
)
_VALID_STATUS = ("active", "draft")
_VALID_LANES = ("secret", "login", "ceremony")
_VALID_ROTATION_METHODS = ("rotate", "re-establish")

# Default review cadence for a catalog pointer — "is this still the right owner
# and page?" That is a genuinely quarterly question, so 90 days is right for it.
# See wiki/AccessRouting.md#drift-review-cadence
DEFAULT_STALE_DAYS = 90

# Cadence for an interim lane's *blocker*, which is a different kind of claim
# with a much shorter half-life: "has the intended owner answered / can they
# front this yet?" (WARDEN-WP-0033-T05).
#
# 14 rather than 90 because 90 was never a loose default, it was an inert one --
# the delegation register was created 2026-08-15, so a 90-day threshold could not
# fire before November and never had. Calibrated instead against blockers that
# actually went stale: the secrets-engine lanes cost ten days, RISK-F-0001
# invalidated an ops-warden blocker in one, and the FLEX-WP-0007 claim was
# repeated by two repos for roughly fifty. 14 catches the ten-day cases and, at
# ~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."""
    reviewed_date = date.fromisoformat(reviewed)
    ref = today or date.today()
    return (ref - reviewed_date).days


def is_review_stale(
    reviewed: str,
    *,
    threshold_days: int = DEFAULT_STALE_DAYS,
    today: Optional[date] = None,
) -> bool:
    """True when reviewed date is older than the cadence threshold."""
    return days_since_review(reviewed, today=today) > threshold_days


class CatalogError(Exception):
    """Raised when the routing catalog is missing or invalid."""


def find_catalog_path(start: Optional[Path] = None) -> Path:
    """Locate registry/routing/catalog.yaml.

    Honors WARDEN_ROUTING_CATALOG first; otherwise walks up from `start`
    (default: this module) until a repo root containing the catalog is found.
    """
    override = os.environ.get("WARDEN_ROUTING_CATALOG")
    if override:
        return Path(os.path.expanduser(override))

    rel = Path("registry") / "routing" / "catalog.yaml"
    here = (start or Path(__file__)).resolve()
    for parent in [here, *here.parents]:
        candidate = parent / rel
        if candidate.exists():
            return candidate
    # Fallback: registry bundled into the installed wheel (warden/_registry/...).
    bundled = Path(__file__).resolve().parent.parent / "_registry" / "routing" / "catalog.yaml"
    if bundled.exists():
        return bundled
    raise CatalogError(
        f"Routing catalog not found ({rel}). Set WARDEN_ROUTING_CATALOG to override."
    )


@dataclass
class CatalogFreshness:
    """Install vs source freshness for the routing catalog (WARDEN-WP-0029 T05).

    Surfaces the path that was loaded, whether it is the wheel-bundled fallback
    (the stale-CLI failure mode), a content hash, and entry review age. Never
    carries secret material.
    """

    path: str
    source: str  # "override" | "repo" | "bundled"
    content_hash: str
    mtime_iso: str
    package_version: str
    entry_count: int
    active_count: int
    newest_reviewed: Optional[str]
    oldest_reviewed: Optional[str]
    stale_entry_count: int
    using_bundled: bool
    warnings: List[str] = field(default_factory=list)

    def to_dict(self) -> dict:
        return {
            "path": self.path,
            "source": self.source,
            "content_hash": self.content_hash,
            "mtime_iso": self.mtime_iso,
            "package_version": self.package_version,
            "entry_count": self.entry_count,
            "active_count": self.active_count,
            "newest_reviewed": self.newest_reviewed,
            "oldest_reviewed": self.oldest_reviewed,
            "stale_entry_count": self.stale_entry_count,
            "using_bundled": self.using_bundled,
            "warnings": list(self.warnings),
        }


@dataclass
class Catalog:
    path: Path
    entries: List[RouteEntry]

    # --- lookup helpers ---------------------------------------------------

    def get(self, entry_id: str) -> Optional[RouteEntry]:
        for e in self.entries:
            if e.id == entry_id:
                return e
        return None

    def listed(self, include_draft: bool = False) -> List[RouteEntry]:
        if include_draft:
            return list(self.entries)
        return [e for e in self.entries if e.is_active]

    def find(self, query: str, include_draft: bool = False, limit: int = 5) -> List[RouteEntry]:
        """Rank entries by keyword overlap with the query. Highest first.

        An exact catalog-id match wins outright — this is what makes a stable keyed
        command (`warden access whynot-design-npm-publish`) resolve deterministically
        regardless of keyword collisions with other lanes.
        """
        exact = self.get(query.strip())
        if exact is not None and (include_draft or exact.is_active):
            return [exact]
        tokens = [t for t in query.lower().replace("-", " ").split() if t]
        pool = self.listed(include_draft=include_draft)
        scored = [(e.match_score(tokens), e) for e in pool]
        scored = [(s, e) for s, e in scored if s > 0]
        scored.sort(key=lambda pair: (-pair[0], pair[1].id))
        return [e for _, e in scored[:limit]]

    def stale(
        self,
        include_draft: bool = False,
        threshold_days: int = DEFAULT_STALE_DAYS,
        *,
        today: Optional[date] = None,
    ) -> List[RouteEntry]:
        """Entries whose reviewed date is past the cadence threshold."""
        return [
            e
            for e in self.listed(include_draft=include_draft)
            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: 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
        day the question is asked and stays that way while nobody answers.
        """
        out: List[RouteEntry] = []
        for e in self.gaps(include_draft=include_draft):
            d = e.effective_delegation
            reviewed = d.reviewed or e.reviewed
            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)
        return out

    def freshness(
        self,
        *,
        stale_threshold_days: int = DEFAULT_STALE_DAYS,
        today: Optional[date] = None,
    ) -> CatalogFreshness:
        """Describe which catalog was loaded and how fresh it is (WP-0029 T05)."""
        path = self.path.resolve()
        text = path.read_text(encoding="utf-8") if path.exists() else ""
        digest = hashlib.sha256(text.encode("utf-8")).hexdigest()[:12]
        mtime_iso = ""
        if path.exists():
            mtime_iso = datetime.fromtimestamp(
                path.stat().st_mtime, tz=timezone.utc
            ).isoformat()

        source = _classify_catalog_source(path)
        using_bundled = source == "bundled"
        reviewed_dates = [e.reviewed for e in self.entries if e.reviewed]
        newest = max(reviewed_dates) if reviewed_dates else None
        oldest = min(reviewed_dates) if reviewed_dates else None
        stale_count = len(self.stale(include_draft=True, threshold_days=stale_threshold_days, today=today))

        package_version = _package_version()
        warnings: List[str] = []
        if using_bundled:
            warnings.append(
                "using wheel-bundled catalog fallback — reinstall from checkout "
                "(`uv tool install -e .` or `pip install -e .`) if lanes look missing"
            )
        if stale_count:
            warnings.append(
                f"{stale_count} catalog entr{'y' if stale_count == 1 else 'ies'} "
                f"past {stale_threshold_days}d review cadence"
            )
        # 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, today=today))
        if stale_interim:
            warnings.append(
                f"{stale_interim} interim delegation"
                f"{'' if stale_interim == 1 else 's'} need re-verifying "
                f"(risk-scaled blocker cadence) — see `warden route gaps`"
            )

        return CatalogFreshness(
            path=str(path),
            source=source,
            content_hash=digest,
            mtime_iso=mtime_iso,
            package_version=package_version,
            entry_count=len(self.entries),
            active_count=len(self.listed(include_draft=False)),
            newest_reviewed=newest,
            oldest_reviewed=oldest,
            stale_entry_count=stale_count,
            using_bundled=using_bundled,
            warnings=warnings,
        )


def _package_version() -> str:
    try:
        from importlib.metadata import version

        return version("ops-warden")
    except Exception:  # noqa: BLE001
        try:
            from warden import __version__

            return str(__version__)
        except Exception:  # noqa: BLE001
            return "unknown"


def _classify_catalog_source(path: Path) -> str:
    """Classify catalog load path for freshness warnings."""
    if os.environ.get("WARDEN_ROUTING_CATALOG"):
        return "override"
    resolved = str(path.resolve())
    if "/_registry/" in resolved or resolved.endswith("/warden/_registry/routing/catalog.yaml"):
        return "bundled"
    # hatch force-include places registry at warden/_registry
    parts = path.resolve().parts
    if "_registry" in parts:
        return "bundled"
    return "repo"


def _assert_no_secret_material(
    entry_id: str, field_name: str, value: str, *, prose: bool = False
) -> None:
    """Reject a handoff field that appears to embed a literal secret value.

    The structured handoff fields are command/path *templates*: concrete values
    must be placeholders (`<...>`) or field names, never a real credential. The
    catalog is git-tracked and agent-visible, so a leaked value here is the exact
    custody failure WP-0014 forbids. We screen for known token prefixes and for a
    long high-entropy run that is not a placeholder.

    ``prose=True`` (rotation guidance steps, WP-0026 T06) skips the *substring*
    prefix screen — short prefixes like ``s.`` or ``eyJ`` collide with ordinary
    English ("exists.", "artifacts.") — and relies on the high-entropy-run detector,
    which catches an actually-pasted token (a real ``hvs.``/``ghp_``/``sk-`` value
    carries a long high-entropy tail) while allowing plain sentences.
    """
    lowered = value.lower()
    if not prose:
        for prefix in _SECRET_PREFIXES:
            if prefix.lower() in lowered:
                raise CatalogError(
                    f"entry {entry_id!r} field {field_name!r} appears to contain a literal "
                    f"secret (matched {prefix!r}). Handoff fields are templates — use "
                    "placeholders like <FIELD>/<PATH>, never a real value."
                )
    for run in _HIGH_ENTROPY_RUN.findall(value):
        # Allow long placeholder/path/identifier tokens; flag anything else.
        if "<" in run or ">" in run:
            continue
        if run.replace("_", "").replace("-", "").isalpha():
            continue  # all-letters run (e.g. a long word) — not a credential
        raise CatalogError(
            f"entry {entry_id!r} field {field_name!r} contains a high-entropy token "
            f"({run[:8]}…) that is not a placeholder — suspected leaked secret value."
        )


def _parse_rotation(entry_id: str, raw: Optional[dict]) -> Optional[RotationGuide]:
    """Parse and validate an optional ``rotation:`` block (WP-0026 T06).

    Advisory renewal guidance only — screened for secret material like every other
    catalog string. ``method`` must be rotate | re-establish; ``steps`` a non-empty
    list; ``owner`` required.
    """
    if raw is None:
        return None
    if not isinstance(raw, dict):
        raise CatalogError(f"entry {entry_id!r} `rotation` must be a mapping")

    method = str(raw.get("method", "")).strip()
    if method not in _VALID_ROTATION_METHODS:
        raise CatalogError(
            f"entry {entry_id!r} rotation.method {method!r} invalid "
            f"(expected one of {_VALID_ROTATION_METHODS})"
        )

    steps_raw = raw.get("steps")
    if not isinstance(steps_raw, list) or not steps_raw:
        raise CatalogError(
            f"entry {entry_id!r} rotation.steps must be a non-empty list of steps"
        )
    steps = [str(s) for s in steps_raw]

    owner = str(raw.get("owner", "")).strip()
    if not owner:
        raise CatalogError(f"entry {entry_id!r} rotation.owner is required")

    # Screen advisory prose for accidental secret material (git-tracked, agent-visible).
    for i, step in enumerate(steps):
        _assert_no_secret_material(entry_id, f"rotation.steps[{i}]", step, prose=True)
    _assert_no_secret_material(entry_id, "rotation.owner", owner, prose=True)

    return RotationGuide(
        method=method,
        steps=steps,
        owner=owner,
        automatable=bool(raw.get("automatable", False)),
    )


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
        )

    verified = str(raw.get("verified", "")).strip() or None
    if verified is not None and verified not in Delegation.VERIFICATION_METHODS:
        raise CatalogError(
            f"entry {entry_id!r} delegation.verified {verified!r} invalid "
            f"(expected one of {Delegation.VERIFICATION_METHODS})"
        )

    return Delegation(
        mode=mode,
        intended_owner=intended_owner,
        blocked_on=blocked_on,
        reviewed=reviewed,
        verified=verified,
        implicit=False,
    )


def _parse_workload_ref(entry_id: str, raw: object) -> WorkloadReference:
    """Parse an explicit workload join without attempting identity inference."""
    if not isinstance(raw, dict):
        raise CatalogError(
            f"entry {entry_id!r} workload_ref must be a mapping; every lane must "
            "declare applicable or not-applicable"
        )

    applicability = str(raw.get("applicability", "")).strip()
    if applicability not in VALID_WORKLOAD_APPLICABILITY:
        raise CatalogError(
            f"entry {entry_id!r} workload_ref.applicability {applicability!r} invalid "
            f"(expected one of {VALID_WORKLOAD_APPLICABILITY})"
        )

    def optional(name: str) -> Optional[str]:
        value = raw.get(name)
        return str(value).strip() if value is not None and str(value).strip() else None

    ref = WorkloadReference(
        applicability=applicability,
        rapp_id=optional("rapp_id"),
        name=optional("name"),
        deployable=optional("deployable"),
        declaration_ref=optional("declaration_ref"),
        reason=optional("reason"),
        unknown_reason=optional("unknown_reason"),
    )

    target_fields = (ref.rapp_id, ref.name, ref.deployable, ref.declaration_ref)
    if applicability == "not-applicable":
        if not ref.reason:
            raise CatalogError(
                f"entry {entry_id!r} workload_ref.reason is required for not-applicable"
            )
        if any(target_fields) or ref.unknown_reason:
            raise CatalogError(
                f"entry {entry_id!r} not-applicable workload_ref must not carry a "
                "workload target or unknown_reason"
            )
        return ref

    if ref.unknown_reason:
        if any(target_fields) or ref.reason:
            raise CatalogError(
                f"entry {entry_id!r} unknown workload_ref must carry only "
                "applicability and unknown_reason"
            )
        return ref

    if not ref.name:
        raise CatalogError(
            f"entry {entry_id!r} applicable workload_ref requires name or "
            "unknown_reason"
        )
    if ref.rapp_id:
        if ref.declaration_ref:
            raise CatalogError(
                f"entry {entry_id!r} managed workload_ref must not also carry "
                "declaration_ref"
            )
    elif not ref.declaration_ref:
        raise CatalogError(
            f"entry {entry_id!r} operational workload_ref requires declaration_ref"
        )
    if ref.deployable and not ref.rapp_id:
        raise CatalogError(
            f"entry {entry_id!r} workload_ref.deployable requires rapp_id"
        )
    if ref.reason:
        raise CatalogError(
            f"entry {entry_id!r} applicable workload_ref must not carry reason"
        )
    return ref


def _parse_entry(raw: dict, index: int) -> RouteEntry:
    if not isinstance(raw, dict):
        raise CatalogError(f"entry #{index} is not a mapping")

    missing = [f for f in _REQUIRED_FIELDS if f not in raw]
    if missing:
        ident = raw.get("id", f"#{index}")
        raise CatalogError(f"entry {ident!r} missing required field(s): {', '.join(missing)}")

    warden_executes = bool(raw["warden_executes"])
    steps = raw.get("steps") or []
    cert_command = raw.get("cert_command")
    status = str(raw["status"])

    if status not in _VALID_STATUS:
        raise CatalogError(
            f"entry {raw['id']!r} has invalid status {status!r} (expected one of {_VALID_STATUS})"
        )

    # No-double-source rule: authored procedure only on the SSH lane.
    if not warden_executes and steps:
        raise CatalogError(
            f"entry {raw['id']!r} is not warden_executes but carries a `steps` block "
            "— routed needs point at the owner's doc; they must not restate procedure "
            "(no-double-source rule)."
        )
    if not warden_executes and cert_command:
        raise CatalogError(
            f"entry {raw['id']!r} is not warden_executes but carries a `cert_command`."
        )

    if not isinstance(raw["need_keywords"], list):
        raise CatalogError(f"entry {raw['id']!r} need_keywords must be a list")

    # Structured handoff fields (WP-0014) — optional, screened for secret material.
    entry_id = str(raw["id"])
    handoff: dict[str, Optional[str]] = {}
    for fname in _HANDOFF_STR_FIELDS:
        val = raw.get(fname)
        if val is None or val == "":
            handoff[fname] = None
            continue
        sval = str(val)
        _assert_no_secret_material(entry_id, fname, sval)
        handoff[fname] = sval

    exec_capable = bool(raw.get("exec_capable", False))
    # A lane cannot be proxy-executable without a fetch_command to run.
    if exec_capable and not handoff["fetch_command"]:
        raise CatalogError(
            f"entry {entry_id!r} sets exec_capable: true but has no fetch_command — "
            "a proxyable lane must declare the command warden runs as the caller."
        )

    lane = str(raw.get("lane", "secret"))
    if lane not in _VALID_LANES:
        raise CatalogError(
            f"entry {entry_id!r} has invalid lane {lane!r} (expected one of {_VALID_LANES})"
        )

    risk_value = raw.get("risk")
    risk = str(risk_value).strip() if risk_value is not None else "ungraded"
    risk = risk or "ungraded"
    if risk != "ungraded" and risk not in VALID_RISK:
        raise CatalogError(
            f"entry {entry_id!r} has invalid risk {risk!r} (expected one of {VALID_RISK})"
        )

    workload_ref = _parse_workload_ref(entry_id, raw.get("workload_ref"))

    return RouteEntry(
        id=entry_id,
        title=str(raw["title"]),
        need_keywords=[str(k) for k in raw["need_keywords"]],
        owner_repo=str(raw["owner_repo"]),
        subsystem=str(raw["subsystem"]),
        warden_executes=warden_executes,
        wiki_ref=str(raw["wiki_ref"]),
        canon_ref=str(raw["canon_ref"]),
        reviewed=str(raw["reviewed"]),
        status=status,
        workload_ref=workload_ref,
        steps=[str(s) for s in steps],
        cert_command=str(cert_command) if cert_command else None,
        auth_method=handoff["auth_method"],
        path_template=handoff["path_template"],
        fetch_command=handoff["fetch_command"],
        exec_capable=exec_capable,
        policy_ref=handoff["policy_ref"],
        lane=lane,
        exec_owner=str(raw["exec_owner"]) if raw.get("exec_owner") else None,
        exec_command=handoff["exec_command"],
        pointer_command=handoff["pointer_command"],
        rotation=_parse_rotation(entry_id, raw.get("rotation")),
        risk=risk,
        delegation=_parse_delegation(entry_id, raw.get("delegation")),
    )


def load_catalog(path: Optional[Path] = None) -> Catalog:
    """Load, parse, and validate the routing catalog."""
    catalog_path = path or find_catalog_path()
    if not catalog_path.exists():
        raise CatalogError(f"Routing catalog not found: {catalog_path}")

    try:
        with catalog_path.open() as f:
            raw = yaml.safe_load(f)
    except yaml.YAMLError as e:
        raise CatalogError(f"Invalid YAML in {catalog_path}: {e}") from e

    if not isinstance(raw, dict):
        raise CatalogError("Catalog must be a YAML mapping")

    raw_entries = raw.get("entries")
    if not isinstance(raw_entries, list) or not raw_entries:
        raise CatalogError("Catalog has no `entries` list")

    entries: List[RouteEntry] = []
    seen: set[str] = set()
    for i, raw_entry in enumerate(raw_entries):
        entry = _parse_entry(raw_entry, i)
        if entry.id in seen:
            raise CatalogError(f"duplicate entry id: {entry.id!r}")
        seen.add(entry.id)
        entries.append(entry)

    return Catalog(path=catalog_path, entries=entries)