"""Tests for the routing pointer catalog and `warden route` CLI.

No test here requires a live subsystem — routing is a read-only pointer layer.
"""
import json
import re
from datetime import date
from pathlib import Path

import pytest
import yaml
from typer.testing import CliRunner

from warden.cli import app
from warden.routing import CatalogError, load_catalog
from warden.routing.catalog import days_since_review, find_catalog_path, is_review_stale
from warden.scorecard import check_catalog_rotation_coverage

runner = CliRunner()


def _repo_catalog() -> Path:
    return find_catalog_path()


def _write_catalog(tmp_path: Path, entries: list[dict]) -> Path:
    path = tmp_path / "catalog.yaml"
    path.write_text(yaml.dump({"version": 1, "entries": entries}))
    return path


SSH_ENTRY = {
    "id": "ssh-cert-host-access",
    "title": "SSH cert",
    "need_keywords": ["ssh", "cert", "sign"],
    "owner_repo": "ops-warden",
    "subsystem": "ops-warden",
    "warden_executes": True,
    "wiki_ref": "wiki/AccessRouting.md#issue-vs-route",
    "canon_ref": "net-kingdom/docs/x.md",
    "reviewed": "2026-06-18",
    "status": "active",
    "workload_ref": {
        "applicability": "not-applicable",
        "reason": "generic certificate action",
    },
    "cert_command": "warden sign <actor> --pubkey <path>",
    "steps": ["confirm inventory", "sign"],
}

ROUTED_ENTRY = {
    "id": "openbao-api-key",
    "title": "API key",
    "need_keywords": ["api", "key", "openbao"],
    "owner_repo": "railiance-platform",
    "subsystem": "OpenBao",
    "warden_executes": False,
    "wiki_ref": "wiki/CredentialRouting.md#routing-table",
    "canon_ref": "net-kingdom/docs/x.md",
    "reviewed": "2026-06-18",
    "status": "active",
    "workload_ref": {
        "applicability": "not-applicable",
        "reason": "generic credential pattern",
    },
}


# ---------------------------------------------------------------------------
# Catalog load + validation
# ---------------------------------------------------------------------------

def test_real_catalog_loads():
    catalog = load_catalog(_repo_catalog())
    assert len(catalog.entries) >= 6
    ssh = catalog.get("ssh-cert-host-access")
    assert ssh is not None and ssh.warden_executes is True
    assert ssh.cert_command and "warden sign" in ssh.cert_command


def test_real_catalog_has_one_executed_lane():
    catalog = load_catalog(_repo_catalog())
    executed = [e for e in catalog.entries if e.warden_executes]
    assert [e.id for e in executed] == ["ssh-cert-host-access"]


def test_every_catalog_lane_declares_workload_applicability():
    catalog = load_catalog(_repo_catalog())
    assert all(entry.workload_ref is not None for entry in catalog.entries)
    assert {entry.workload_ref.resolution for entry in catalog.entries} == {
        "resolved", "unknown", "not-applicable"
    }


def test_managed_and_operational_workload_references_parse():
    catalog = load_catalog(_repo_catalog())
    managed = catalog.get("issue-core-ingestion-api-key").workload_ref
    assert managed.resolution == "resolved"
    assert (managed.rapp_id, managed.name, managed.deployable) == (
        "rapp-issue-core", "issue-core", "issue-core"
    )
    operational = catalog.get("ops-warden-warden-sign-token").workload_ref
    assert operational.resolution == "resolved"
    assert operational.rapp_id is None
    assert operational.name == "ops-warden"
    assert operational.declaration_ref == "tenancy.yaml"


def test_workload_reference_rejects_ambiguous_absence(tmp_path):
    bad = dict(ROUTED_ENTRY)
    bad.pop("workload_ref")
    with pytest.raises(CatalogError, match="workload_ref"):
        load_catalog(_write_catalog(tmp_path, [bad]))


def test_workload_reference_rejects_malformed_managed_target(tmp_path):
    bad = dict(ROUTED_ENTRY)
    bad["workload_ref"] = {
        "applicability": "applicable",
        "rapp_id": "rapp-issue-core",
    }
    with pytest.raises(CatalogError, match="requires name"):
        load_catalog(_write_catalog(tmp_path, [bad]))


def test_ops_warden_warden_sign_lane_has_native_exec():
    """RAILIANCE-WP-0005 T08 — broker lane routes to railiance-platform credential exec."""
    catalog = load_catalog(_repo_catalog())
    e = catalog.get("ops-warden-warden-sign-token")
    assert e is not None and e.is_active and e.owner_repo == "railiance-platform"
    assert e.has_native_exec is True
    assert e.exec_owner == "railiance-platform"
    assert "credential.py exec" in e.exec_command
    assert "ops-warden/warden-sign" in e.exec_command
    assert "credential-exec-ops-warden-smoke" in e.pointer_command
    assert e.warden_executes is False
    assert e.resolvable is False  # broker lane — owner exec, not warden access --fetch


def test_route_find_vault_token_ops_warden_prefers_broker_lane():
    catalog = load_catalog(_repo_catalog())
    matches = catalog.find("VAULT_TOKEN ops-warden warden sign", limit=3)
    assert matches[0].id == "ops-warden-warden-sign-token"


def test_whynot_design_npm_lane_is_concrete_and_resolvable():
    """The provisioned npm publish lane has no placeholders and reports resolvable."""
    catalog = load_catalog(_repo_catalog())
    e = catalog.get("whynot-design-npm-publish")
    assert e is not None and e.is_active and e.exec_capable
    assert e.resolvable is True
    assert "<" not in e.fetch_command and ">" not in e.fetch_command
    assert "platform/workloads/coulomb/whynot-design/npm-publish" in e.fetch_command


def test_policy_nexus_source_read_lane_is_exact_high_risk_and_resolvable():
    catalog = load_catalog(_repo_catalog())
    entry = catalog.get("policy-nexus-forgejo-source-read")
    assert entry is not None and entry.is_active and entry.exec_capable
    assert entry.resolvable is True
    assert entry.risk == "high"
    assert entry.owner_repo == "railiance-platform"
    assert entry.fetch_command == (
        "bao kv get -field=FORGEJO_SOURCE_TOKEN "
        "platform/workloads/policy-nexus/forgejo-source-read"
    )
    assert entry.path_template == "platform/workloads/policy-nexus/forgejo-source-read"
    assert entry.auth_method.endswith(
        "role=policy-nexus-forgejo-source-workload-kv-read"
    )
    assert entry.delegation is not None and entry.delegation.mode == "native"


def test_route_find_policy_nexus_source_read_prefers_concrete_lane():
    catalog = load_catalog(_repo_catalog())
    matches = catalog.find(
        "policy nexus Forgejo private source repository read token Actions", limit=1
    )
    assert matches[0].id == "policy-nexus-forgejo-source-read"


def test_generic_and_template_lanes_not_resolvable():
    catalog = load_catalog(_repo_catalog())
    # generic openbao lane has <FIELD>/<path_template>; login lane has <domain>.
    assert catalog.get("openbao-api-key").resolvable is False
    assert catalog.get("key-cape-oidc-login").resolvable is False


def test_platform_admin_login_lane_is_exact_and_non_value_bearing():
    entry = load_catalog(_repo_catalog()).get("openbao-platform-admin-login")
    assert entry.lane == "login"
    assert entry.risk == "high"
    assert entry.fetch_command == (
        "bao login -no-print -method=oidc -path=netkingdom role=platform-admin"
    )
    assert entry.workload_ref.resolution == "not-applicable"


def test_netkingdom_sso_bind_lanes_are_routed_but_not_resolvable():
    catalog = load_catalog(_repo_catalog())
    for lane_id in (
        "net-kingdom-lldap-bind-credential",
        "net-kingdom-privacyidea-admin-token",
    ):
        entry = catalog.get(lane_id)
        assert entry is not None
        assert entry.owner_repo == "railiance-platform"
        assert entry.risk == "high"
        assert entry.warden_executes is False
        assert entry.exec_capable is False
        assert entry.resolvable is False
        assert entry.delegation.blocked_on
        assert "net-kingdom-sso-bind-credentials.md#worker-checklist" in entry.wiki_ref


def test_openbao_recovery_ceremony_is_non_value_bearing_owner_pointer():
    entry = load_catalog(_repo_catalog()).get("openbao-shamir-recovery-ceremony")
    assert entry.lane == "ceremony"
    assert entry.risk == "high"
    assert entry.owner_repo == "railiance-platform"
    assert entry.warden_executes is False
    assert entry.exec_capable is False
    assert entry.has_handoff is False
    assert entry.vends_secret is False
    assert entry.workload_ref.resolution == "not-applicable"


def test_find_exact_id_wins_over_keyword_collision():
    catalog = load_catalog(_repo_catalog())
    # "npm" alone collides with openbao-api-key; the exact id must resolve uniquely.
    assert catalog.find("whynot-design-npm-publish", limit=1)[0].id == "whynot-design-npm-publish"


def test_native_exec_owner_on_npm_lane():
    """secrets-engine is the owner-native exec front door for the npm lane (WP-0019)."""
    catalog = load_catalog(_repo_catalog())
    e = catalog.get("whynot-design-npm-publish")
    assert e.has_native_exec is True
    assert e.exec_owner == "secrets-engine"
    assert "secrets-engine exec --catalog whynot-design-npm-publish" in e.exec_command
    assert "secrets-engine route" in e.pointer_command
    # The proxy fallback is still available (exec_capable + resolvable).
    assert e.exec_capable is True and e.resolvable is True


def test_lanes_without_native_exec():
    catalog = load_catalog(_repo_catalog())
    assert catalog.get("openbao-api-key").has_native_exec is False
    assert catalog.get("ssh-cert-host-access").has_native_exec is False


def test_cli_show_native_exec_json(repo_catalog_env):
    result = runner.invoke(app, ["route", "show", "whynot-design-npm-publish", "--json"])
    data = json.loads(result.stdout)
    assert data["exec_owner"] == "secrets-engine"
    assert "secrets-engine exec" in data["exec_command"]
    assert "primary" in data["next_action"] and "secrets-engine" in data["next_action"]


def test_cli_show_warden_sign_broker_json(repo_catalog_env):
    result = runner.invoke(app, ["route", "show", "ops-warden-warden-sign-token", "--json"])
    assert result.exit_code == 0
    data = json.loads(result.stdout)
    assert data["owner_repo"] == "railiance-platform"
    assert data["exec_owner"] == "railiance-platform"
    assert "credential.py exec" in data["exec_command"]
    assert "primary" in data["next_action"] and "railiance-platform" in data["next_action"]


def test_no_double_source_rule_rejects_routed_steps(tmp_path):
    bad = dict(ROUTED_ENTRY)
    bad["steps"] = ["do a thing on OpenBao"]  # non-SSH entry must not carry steps
    path = _write_catalog(tmp_path, [SSH_ENTRY, bad])
    with pytest.raises(CatalogError, match="no-double-source"):
        load_catalog(path)


def test_routed_cert_command_rejected(tmp_path):
    bad = dict(ROUTED_ENTRY)
    bad["cert_command"] = "warden secret get"
    path = _write_catalog(tmp_path, [bad])
    with pytest.raises(CatalogError, match="cert_command"):
        load_catalog(path)


def test_duplicate_id_rejected(tmp_path):
    path = _write_catalog(tmp_path, [ROUTED_ENTRY, dict(ROUTED_ENTRY)])
    with pytest.raises(CatalogError, match="duplicate"):
        load_catalog(path)


def test_missing_field_rejected(tmp_path):
    bad = {k: v for k, v in ROUTED_ENTRY.items() if k != "owner_repo"}
    path = _write_catalog(tmp_path, [bad])
    with pytest.raises(CatalogError, match="owner_repo"):
        load_catalog(path)


def test_missing_catalog_file():
    with pytest.raises(CatalogError):
        load_catalog(Path("/nonexistent/catalog.yaml"))


# ---------------------------------------------------------------------------
# Structured handoff fields (WP-0014, T1)
# ---------------------------------------------------------------------------

def test_handoff_fields_parse_on_routed_entry(tmp_path):
    entry = dict(ROUTED_ENTRY)
    entry["auth_method"] = "key-cape OIDC → bao login -method=oidc role=<domain>"
    entry["path_template"] = "platform/workloads/<domain>/<workload>/<bundle>"
    entry["fetch_command"] = "bao kv get -field=<FIELD> <path_template>"
    entry["policy_ref"] = "flex-auth check secret.read:<domain>"
    entry["exec_capable"] = True
    catalog = load_catalog(_write_catalog(tmp_path, [entry]))
    e = catalog.get("openbao-api-key")
    assert e.has_handoff is True
    assert e.exec_capable is True
    assert e.path_template.startswith("platform/workloads/")


def test_real_catalog_openbao_entry_has_handoff():
    e = load_catalog(_repo_catalog()).get("openbao-api-key")
    assert e is not None and e.has_handoff and e.exec_capable
    assert "<" in e.path_template and "<" in e.fetch_command  # templates, not values


def test_exec_capable_without_fetch_command_rejected(tmp_path):
    bad = dict(ROUTED_ENTRY)
    bad["exec_capable"] = True  # no fetch_command
    with pytest.raises(CatalogError, match="fetch_command"):
        load_catalog(_write_catalog(tmp_path, [bad]))


@pytest.mark.parametrize(
    "leaked",
    [
        "bao write x token=ghp_abcdef0123456789abcdef0123",  # github token prefix
        "x=AKIAIOSFODNN7EXAMPLE",                             # aws key id
        "header=eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9",       # jwt prefix
        "val=ZmFrZXNlY3JldDEyMzQ1Njc4OWFiY2RlZmdoaWprbA",    # high-entropy run
    ],
)
def test_handoff_secret_material_rejected(tmp_path, leaked):
    bad = dict(ROUTED_ENTRY)
    bad["fetch_command"] = leaked
    with pytest.raises(CatalogError, match="secret|high-entropy"):
        load_catalog(_write_catalog(tmp_path, [bad]))


def test_handoff_template_with_placeholders_accepted(tmp_path):
    ok = dict(ROUTED_ENTRY)
    ok["fetch_command"] = "bao kv get -field=<FIELD> platform/workloads/<domain>/<bundle>"
    catalog = load_catalog(_write_catalog(tmp_path, [ok]))
    assert catalog.get("openbao-api-key").fetch_command.startswith("bao kv get")


# ---------------------------------------------------------------------------
# find ranking
# ---------------------------------------------------------------------------

def test_find_active_excludes_draft():
    catalog = load_catalog(_repo_catalog())
    ids = [e.id for e in catalog.find("s3 temporary credentials")]
    assert "object-storage-sts" not in ids


def test_find_all_includes_draft():
    catalog = load_catalog(_repo_catalog())
    ids = [e.id for e in catalog.find("s3 temporary credentials", include_draft=True)]
    assert "object-storage-sts" in ids


def test_find_issue_core_lane_active():
    catalog = load_catalog(_repo_catalog())
    ids = [e.id for e in catalog.find("issue core api key")]
    assert "issue-core-ingestion-api-key" in ids


def test_find_ssh_tunnel_top_match():
    catalog = load_catalog(_repo_catalog())
    matches = catalog.find("ssh tunnel")
    assert matches and matches[0].id == "ops-bridge-tunnel"


def test_find_openrouter_key():
    catalog = load_catalog(_repo_catalog())
    matches = catalog.find("openrouter api key", include_draft=True)
    assert matches and matches[0].id == "openrouter-llm-connect"


def test_find_reuse_surface_hub_write_token():
    catalog = load_catalog(_repo_catalog())
    matches = catalog.find("REUSE_SURFACE_TOKEN", limit=3)
    assert matches and matches[0].id == "reuse-surface-hub-write-token"


def test_reuse_surface_hub_write_token_lane_is_resolvable():
    catalog = load_catalog(_repo_catalog())
    e = catalog.get("reuse-surface-hub-write-token")
    assert e is not None and e.is_active and e.exec_capable
    assert e.resolvable is True
    assert e.owner_repo == "railiance-platform"
    assert "platform/workloads/reuse/reuse-surface/runtime-secrets" in e.fetch_command


def test_find_object_storage_sts():
    catalog = load_catalog(_repo_catalog())
    matches = catalog.find("s3 temporary credentials", include_draft=True)
    assert matches and matches[0].id == "object-storage-sts"


# ---------------------------------------------------------------------------
# Review staleness
# ---------------------------------------------------------------------------

def test_days_since_review():
    assert days_since_review("2026-06-01", today=date(2026, 6, 24)) == 23


def test_is_review_stale_past_threshold():
    assert is_review_stale("2026-01-01", threshold_days=90, today=date(2026, 6, 24))


def test_is_review_stale_within_threshold():
    assert not is_review_stale("2026-06-01", threshold_days=90, today=date(2026, 6, 24))


def test_catalog_stale_filters_entries():
    catalog = load_catalog(_repo_catalog())
    stale = catalog.stale(threshold_days=0, today=date(2026, 6, 25))
    assert stale
    assert all(e.reviewed <= "2026-06-24" for e in stale)


# ---------------------------------------------------------------------------
# CLI (uses the repo catalog via env override)
# ---------------------------------------------------------------------------

@pytest.fixture
def repo_catalog_env(monkeypatch):
    monkeypatch.setenv("WARDEN_ROUTING_CATALOG", str(_repo_catalog()))


def test_cli_list_active_only(repo_catalog_env):
    result = runner.invoke(app, ["route", "list", "--json"])
    assert result.exit_code == 0
    ids = [e["id"] for e in json.loads(result.stdout)]
    assert "object-storage-sts" not in ids
    assert "issue-core-ingestion-api-key" in ids


def test_cli_list_all_includes_draft(repo_catalog_env):
    result = runner.invoke(app, ["route", "list", "--all", "--json"])
    ids = [e["id"] for e in json.loads(result.stdout)]
    assert "object-storage-sts" in ids


def test_cli_show_ssh_json_includes_cert_pattern(repo_catalog_env):
    result = runner.invoke(app, ["route", "show", "ssh-cert-host-access", "--json"])
    assert result.exit_code == 0
    data = json.loads(result.stdout)
    assert data["warden_executes"] is True
    assert data["warden_role"] == "issue"
    assert "warden sign" in data["cert_command"]
    assert data["steps"]


def test_cli_show_routed_has_next_action_not_steps(repo_catalog_env):
    result = runner.invoke(app, ["route", "show", "openbao-api-key", "--json"])
    data = json.loads(result.stdout)
    assert data["warden_executes"] is False
    # exec_capable lane surfaces as an "assist" role so agents see it is proxyable.
    assert data["warden_role"] == "assist"
    assert data["exec_capable"] is True
    assert "steps" not in data
    assert "next_action" in data
    assert "proxy" in data["next_action"]


def test_cli_show_unknown_exits_one(repo_catalog_env):
    result = runner.invoke(app, ["route", "show", "does-not-exist"])
    assert result.exit_code == 1


def test_cli_find_json(repo_catalog_env):
    result = runner.invoke(app, ["route", "find", "ssh tunnel", "--json"])
    assert result.exit_code == 0
    ids = [e["id"] for e in json.loads(result.stdout)]
    assert "ops-bridge-tunnel" in ids


def test_cli_list_stale_json(repo_catalog_env):
    result = runner.invoke(
        app, ["route", "list", "--stale", "--stale-days", "1", "--json"]
    )
    assert result.exit_code == 0
    data = json.loads(result.stdout)
    assert data
    assert all("days_since_review" in row for row in data)
    assert all(row["stale_threshold_days"] == 1 for row in data)


def test_cli_list_stale_empty_with_high_threshold(repo_catalog_env):
    result = runner.invoke(
        app, ["route", "list", "--stale", "--stale-days", "9999"]
    )
    assert result.exit_code == 0
    assert "No stale" in result.output


def test_cli_find_openrouter_draft_only_with_all(repo_catalog_env):
    result = runner.invoke(
        app, ["route", "find", "openrouter api key", "--all", "--json"]
    )
    assert result.exit_code == 0
    ids = [e["id"] for e in json.loads(result.stdout)]
    assert "openrouter-llm-connect" in ids


# ---------------------------------------------------------------------------
# T5 drift guard — every wiki_ref anchor resolves, every entry has a reviewed date
# ---------------------------------------------------------------------------

def _github_slug(heading: str) -> str:
    """Approximate GitHub's heading-anchor slug algorithm."""
    text = heading.strip().lower()
    text = re.sub(r"[^\w\s-]", "", text)  # drop punctuation (em-dash, parens, etc.)
    text = text.replace(" ", "-")
    return text


def _heading_anchors(md_path: Path) -> set[str]:
    anchors: set[str] = set()
    for line in md_path.read_text().splitlines():
        m = re.match(r"^#{1,6}\s+(.*)$", line)
        if m:
            anchors.add(_github_slug(m.group(1)))
    return anchors


def test_every_wiki_ref_anchor_resolves():
    catalog = load_catalog(_repo_catalog())
    repo_root = _repo_catalog().parents[2]  # registry/routing/catalog.yaml -> repo root
    failures = []
    for entry in catalog.entries:
        rel, _, anchor = entry.wiki_ref.partition("#")
        md_path = repo_root / rel
        if not md_path.exists():
            failures.append(f"{entry.id}: wiki file missing: {rel}")
            continue
        if anchor and anchor not in _heading_anchors(md_path):
            failures.append(f"{entry.id}: anchor #{anchor} not found in {rel}")
    assert not failures, "\n".join(failures)


def test_every_entry_has_reviewed_date():
    catalog = load_catalog(_repo_catalog())
    for entry in catalog.entries:
        assert re.match(r"^\d{4}-\d{2}-\d{2}$", entry.reviewed), (
            f"{entry.id}: reviewed must be YYYY-MM-DD, got {entry.reviewed!r}"
        )


# ---------------------------------------------------------------------------
# Rotation / re-establishment guidance registry (WARDEN-WP-0026 T06)
# ---------------------------------------------------------------------------

def test_every_active_vending_lane_has_rotation_guidance():
    """Coverage gate: an active lane that vends a secret must say how to renew it."""
    catalog = load_catalog(_repo_catalog())
    missing = [e.id for e in catalog.entries if e.is_active and e.vends_secret and not e.has_rotation]
    assert not missing, f"active vending lanes lacking rotation guidance: {missing}"


def test_scorecard_rotation_coverage_check_passes_on_repo_catalog():
    result = check_catalog_rotation_coverage()
    assert result.passed, result.detail


def test_non_vending_lanes_are_exempt_from_rotation():
    """SSH (issue), login, and pointer-only lanes carry no rotation block."""
    catalog = load_catalog(_repo_catalog())
    assert catalog.get("ssh-cert-host-access").vends_secret is False   # issue lane
    assert catalog.get("key-cape-oidc-login").vends_secret is False    # login lane
    assert catalog.get("ops-bridge-tunnel").vends_secret is False      # pointer only


def test_rotation_block_parses_fields():
    catalog = load_catalog(_repo_catalog())
    rot = catalog.get("forgejo-admin-api-token").rotation
    assert rot is not None
    assert rot.method in ("rotate", "re-establish")
    assert rot.owner == "railiance-platform"
    assert rot.steps and all(isinstance(s, str) for s in rot.steps)


def test_re_establish_method_on_backup_lane():
    catalog = load_catalog(_repo_catalog())
    rot = catalog.get("railiance-backup-offsite-lane").rotation
    assert rot is not None and rot.method == "re-establish"


def test_invalid_rotation_method_rejected(tmp_path):
    entry = dict(ROUTED_ENTRY, rotation={"method": "renew", "owner": "x", "steps": ["a"]})
    with pytest.raises(CatalogError, match="rotation.method"):
        load_catalog(_write_catalog(tmp_path, [SSH_ENTRY, entry]))


def test_rotation_steps_screened_for_pasted_token(tmp_path):
    """A high-entropy pasted token in prose is rejected; ordinary prose is allowed."""
    leak = dict(ROUTED_ENTRY, rotation={
        "method": "rotate", "owner": "x",
        "steps": ["set the value to ghp_" + "aB3dE5" * 6],  # mixed alnum → high-entropy run
    })
    with pytest.raises(CatalogError, match="high-entropy|secret"):
        load_catalog(_write_catalog(tmp_path, [SSH_ENTRY, leak]))


def test_rotation_prose_allows_ordinary_sentences(tmp_path):
    """Words like 'exists.' must not trip the terse 's.' prefix screen."""
    ok = dict(ROUTED_ENTRY, rotation={
        "method": "rotate", "owner": "railiance-platform",
        "steps": ["Rotate per the concrete workload's entry when one exists."],
    })
    catalog = load_catalog(_write_catalog(tmp_path, [SSH_ENTRY, ok]))
    assert catalog.get("openbao-api-key").rotation.steps


def test_rotate_guide_cli_json():
    result = runner.invoke(app, ["rotate-guide", "forgejo-admin-api-token", "--json"])
    assert result.exit_code == 0
    payload = json.loads(result.stdout)
    assert payload["method"] == "rotate"
    assert payload["owner"] == "railiance-platform"
    assert payload["steps"]


def test_rotate_guide_cli_ssh_lane_is_graceful():
    # SSH renewal is re-issuance, not a static rotation — exit 0, not an error.
    result = runner.invoke(app, ["rotate-guide", "ssh-cert-host-access"])
    assert result.exit_code == 0


# ---------------------------------------------------------------------------
# Agent read-boundary + risk class (WARDEN-WP-0026 T04)
# ---------------------------------------------------------------------------

def test_high_risk_lanes_classified():
    catalog = load_catalog(_repo_catalog())
    high = {e.id for e in catalog.entries if e.is_high_risk}
    assert "railiance-backup-offsite-lane" in high
    assert "forgejo-admin-api-token" in high
    assert "openrouter-llm-connect" in high
    # WARDEN-WP-0033-T02: these two were asserted standard here, and the assertion
    # held a defective grade still. Both paths carry a second credential the grade
    # ignored -- GITEA_BACKEND_TOKEN (CCR-2026-0002) and the dual-consumer webhook
    # HMAC (CCR-2026-0005). A read discloses every field at a path, so the grade
    # must cover the union, not the headline field.
    assert "issue-core-ingestion-api-key" in high
    assert "reuse-surface-hub-write-token" in high


def test_invalid_risk_rejected(tmp_path):
    bad = dict(ROUTED_ENTRY, risk="critical")
    with pytest.raises(CatalogError, match="risk"):
        load_catalog(_write_catalog(tmp_path, [SSH_ENTRY, bad]))


def test_backup_lane_promoted_and_resolvable():
    """WP-0026 T07 — CCR-2026-0004 lane is active, resolvable, high-risk, has rotation."""
    catalog = load_catalog(_repo_catalog())
    e = catalog.get("railiance-backup-offsite-lane")
    assert e is not None
    assert e.status == "active"
    assert e.resolvable is True
    assert e.is_high_risk is True
    assert e.has_rotation is True
    assert e.rotation.method == "re-establish"
    assert "NC_WEBDAV_TOKEN" in (e.fetch_command or "")
    assert "<" not in (e.fetch_command or "")


def test_route_show_json_includes_risk():
    result = runner.invoke(app, ["route", "show", "railiance-backup-offsite-lane", "--json"])
    assert result.exit_code == 0
    payload = json.loads(result.stdout)
    assert payload["risk"] == "high"
    assert payload["high_risk"] is True
    assert payload["resolvable"] is True
    assert payload["status"] == "active"


# ---------------------------------------------------------------------------
# Delegation register (WARDEN-WP-0030)
# ---------------------------------------------------------------------------

def test_every_catalog_entry_declares_delegation():
    catalog = load_catalog(_repo_catalog())
    missing = [e.id for e in catalog.entries if e.delegation is None]
    assert missing == [], f"entries missing delegation block: {missing}"


def test_every_proxy_declares_delegation():
    """A new exec_capable proxy cannot land without answering the ownership question."""
    catalog = load_catalog(_repo_catalog())
    missing = [
        e.id
        for e in catalog.entries
        if e.exec_capable and not e.warden_executes and e.delegation is None
    ]
    assert missing == [], f"proxy lanes missing delegation: {missing}"


def test_ssh_lane_is_permanent_delegation():
    e = load_catalog(_repo_catalog()).get("ssh-cert-host-access")
    assert e.delegation is not None
    assert e.delegation.mode == "permanent"
    assert e.delegation.intended_owner is None
    assert e.is_interim is False


def test_native_exec_lanes_are_native_delegation():
    catalog = load_catalog(_repo_catalog())
    for eid, owner in (
        ("whynot-design-npm-publish", "secrets-engine"),
        ("ops-warden-warden-sign-token", "railiance-platform"),
    ):
        e = catalog.get(eid)
        assert e.delegation is not None
        assert e.delegation.mode == "native"
        assert e.delegation.intended_owner == owner


def test_founder_interim_lanes_classified():
    catalog = load_catalog(_repo_catalog())
    expected = {
        "rapp-qonto-keycape-client": "key-cape",
        "binky-company-email-imap": "tenant-engine",
        "binky-qonto-api": "tenant-engine",
        "railiance-backup-offsite-lane": "railiance-platform",
        "agent-harness-forgejo-deploy": "railiance-platform",
    }
    for eid, owner in expected.items():
        e = catalog.get(eid)
        assert e is not None and e.delegation is not None
        assert e.delegation.mode == "interim"
        assert e.delegation.intended_owner == owner
        assert e.delegation.blocked_on


def test_missing_delegation_is_implicit_interim(tmp_path):
    catalog = load_catalog(_write_catalog(tmp_path, [dict(ROUTED_ENTRY)]))
    e = catalog.get("openbao-api-key")
    d = e.effective_delegation
    assert e.delegation is None
    assert d.implicit is True
    assert d.mode == "interim"
    assert d.intended_owner is None
    assert "unclassified" in (d.blocked_on or "")


def test_interim_without_blocked_on_rejected(tmp_path):
    bad = dict(
        ROUTED_ENTRY,
        delegation={
            "mode": "interim",
            "intended_owner": "secrets-engine",
            "reviewed": "2026-08-15",
        },
    )
    with pytest.raises(CatalogError, match="blocked_on"):
        load_catalog(_write_catalog(tmp_path, [bad]))


def test_non_permanent_without_owner_rejected(tmp_path):
    bad = dict(
        ROUTED_ENTRY,
        delegation={"mode": "native", "reviewed": "2026-08-15"},
    )
    with pytest.raises(CatalogError, match="intended_owner"):
        load_catalog(_write_catalog(tmp_path, [bad]))


def test_invalid_delegation_mode_rejected(tmp_path):
    bad = dict(
        ROUTED_ENTRY,
        delegation={"mode": "maybe", "intended_owner": "x", "reviewed": "2026-08-15"},
    )
    with pytest.raises(CatalogError, match="delegation.mode"):
        load_catalog(_write_catalog(tmp_path, [bad]))


def test_catalog_gaps_lists_only_interim():
    catalog = load_catalog(_repo_catalog())
    gap_ids = {e.id for e in catalog.gaps(include_draft=True)}
    assert "ssh-cert-host-access" not in gap_ids
    assert "whynot-design-npm-publish" not in gap_ids
    assert "binky-company-email-imap" in gap_ids
    # WARDEN-WP-0033: openbao-api-key was listed here as an interim cover. It is
    # not one -- its path_template is a <domain>/<workload>/<bundle> routing
    # pattern rather than a single secret lane, so there is no front door for
    # anyone to take over. secrets-engine refused it on exactly that ground and
    # ops-warden agrees. A pointer to OpenBao is not a gap ops-warden is holding,
    # and counting it as one overstated the interim surface by a lane.
    assert "openbao-api-key" not in gap_ids
    assert all(catalog.get(i).is_interim for i in gap_ids)


def test_cli_route_gaps_json(repo_catalog_env):
    result = runner.invoke(app, ["route", "gaps", "--json"])
    assert result.exit_code == 0
    data = json.loads(result.stdout)
    assert data
    ids = {row["id"] for row in data}
    assert "binky-company-email-imap" in ids
    assert "ssh-cert-host-access" not in ids
    for row in data:
        assert row["mode"] == "interim"
        assert "intended_owner" in row
        assert "blocked_on" in row
        assert "days_since_review" in row


def test_cli_route_show_includes_delegation(repo_catalog_env):
    result = runner.invoke(app, ["route", "show", "binky-qonto-api", "--json"])
    assert result.exit_code == 0
    data = json.loads(result.stdout)
    assert data["delegation"]["mode"] == "interim"
    assert data["delegation"]["intended_owner"] == "tenant-engine"
    assert data["delegation"]["implicit"] is False


# --- ADR-0007: absence is not a grade (WARDEN-WP-0032-T06) ------------------


def _bare_entry(**overrides):
    """A minimal RouteEntry, so these tests exercise defaults and nothing else."""
    from warden.routing.models import RouteEntry

    fields = dict(
        id="x",
        title="t",
        need_keywords=[],
        owner_repo="r",
        subsystem="s",
        warden_executes=False,
        wiki_ref="w",
        canon_ref="c",
        reviewed="2026-08-20",
        status="active",
    )
    fields.update(overrides)
    return RouteEntry(**fields)


def test_every_repo_catalog_lane_is_explicitly_graded():
    """The CI gate. A lane added without a `risk` grade is a defect (ADR-0007)."""
    catalog = load_catalog(_repo_catalog())
    ungraded = sorted(e.id for e in catalog.entries if not e.is_graded)
    assert ungraded == [], (
        f"{len(ungraded)} catalog lane(s) carry no explicit risk grade: {ungraded}. "
        "ADR-0007: absence is not a grade — grade the lane on merit."
    )


def test_ungraded_lane_fails_safe_to_high_risk():
    """RISK-F-0003 regression: an omitted grade must not wave a lane through.

    Before ADR-0007 the dataclass default was "standard", so a lane that simply
    omitted the field landed outside the agent read-boundary silently.
    """
    entry = _bare_entry()
    assert entry.risk == "ungraded"
    assert entry.is_graded is False
    assert entry.is_high_risk is True


def test_unrecognised_grade_is_treated_as_high():
    """A grade from a newer catalog must not be read as permission."""
    entry = _bare_entry(risk="spicy")
    assert entry.is_high_risk is True
    assert entry.is_graded is False


def test_ungraded_risk_uses_maturity_derived_zone_default():
    entry = _bare_entry()
    assert entry.risk_for_zone(
        effective_zone="z0-experimental",
        admission="satisfied",
        synthetic_only=True,
    ) == "standard"
    assert entry.risk_for_zone(
        effective_zone="z0-experimental",
        admission="unknown",
        synthetic_only=True,
    ) == "high"
    assert entry.risk_for_zone(
        effective_zone="z3-critical",
        admission="satisfied",
    ) == "critical"
    assert entry.risk_for_zone(effective_zone="unknown") == "high"


def test_explicit_risk_grade_always_wins_over_zone_default():
    entry = _bare_entry(risk="standard")
    assert entry.risk_for_zone(
        effective_zone="z3-critical", admission="satisfied"
    ) == "standard"


def test_low_risk_vocabulary_is_explicit():
    for grade in ("standard", "low", "accepted"):
        entry = _bare_entry(risk=grade)
        assert entry.is_high_risk is False, grade
        assert entry.is_graded is True, grade


# ---------------------------------------------------------------------------
# Blocker staleness cadence + verification (WARDEN-WP-0033-T05)
# ---------------------------------------------------------------------------

def test_blocker_cadence_is_separate_from_pointer_cadence():
    """Two claims with different half-lives must not share one threshold.

    "Is this still the right owner and page?" is quarterly. "Has the owner
    answered yet?" is not. Sharing 90 days made the second one inert -- the
    register was six days old, so it could not have fired for months.
    """
    from warden.routing.catalog import DEFAULT_BLOCKER_STALE_DAYS, DEFAULT_STALE_DAYS

    assert DEFAULT_STALE_DAYS == 90
    assert DEFAULT_BLOCKER_STALE_DAYS == 14
    assert DEFAULT_BLOCKER_STALE_DAYS < DEFAULT_STALE_DAYS


def test_blocker_window_scales_with_risk_and_matches_risk_nexus():
    """risk-nexus stall windows: 14d critical/high, 30d medium, 60d low.

    They offered the convention instead of a joint tool, so the two registers
    agree only for as long as these numbers do.
    """
    from warden.routing.catalog import blocker_stale_days

    assert blocker_stale_days("high") == 14
    assert blocker_stale_days("standard") == 30
    assert blocker_stale_days("low") == 60
    # An ungraded lane gets the SHORTEST window, not the longest -- ADR-0007 makes
    # an absent grade a defect, so its blocker is the least trustworthy of all.
    assert blocker_stale_days("ungraded") == 14
    assert blocker_stale_days(None) == 14
    # An explicit --stale-days still wins.
    assert blocker_stale_days("low", 7) == 7


def test_asked_and_waiting_is_not_verification():
    """The failure this whole change exists to catch.

    A lane asked today reads as reviewed today. The secrets-engine blocker sat
    in exactly that state for ten days while looking current.
    """
    from warden.routing.models import Delegation

    asked = Delegation(mode="interim", intended_owner="x", blocked_on="y",
                       reviewed="2026-08-21", verified="asked-and-waiting")
    assert asked.is_verified is False

    for method in ("owner-confirmed", "source-read"):
        d = Delegation(mode="interim", intended_owner="x", blocked_on="y",
                       reviewed="2026-08-21", verified=method)
        assert d.is_verified is True, method


def test_stale_gaps_flags_unverified_even_when_the_date_is_today():
    catalog = load_catalog(_repo_catalog())
    stale = {e.id for e in catalog.stale_gaps(include_draft=True, today=date(2026, 8, 21))}
    # Asked of key-cape on 2026-08-21 and unanswered -- zero days old, still stale.
    assert "key-cape-oidc-login" in stale
    # Confirmed by the owner the same day -- fresh.
    assert "issue-core-ingestion-api-key" not in stale


def test_invalid_verification_method_rejected(tmp_path):
    entry = dict(ROUTED_ENTRY)
    entry["delegation"] = {
        "mode": "interim", "intended_owner": "secrets-engine",
        "blocked_on": "pending", "reviewed": "2026-08-21", "verified": "probably-fine",
    }
    with pytest.raises(CatalogError, match="verified"):
        load_catalog(_write_catalog(tmp_path, [SSH_ENTRY, entry]))


def test_every_interim_lane_records_how_it_was_verified():
    """Structural, not time-based, so it never fails on a calendar day alone."""
    catalog = load_catalog(_repo_catalog())
    missing = [
        e.id for e in catalog.gaps(include_draft=True)
        if e.effective_delegation.verified is None
    ]
    assert not missing, f"interim lanes with no `verified`: {missing}"


def test_cli_route_gaps_fail_on_stale_exits_3(repo_catalog_env):
    result = runner.invoke(app, ["route", "gaps", "--fail-on-stale", "--json"])
    assert result.exit_code == 3
    rows = json.loads(result.stdout)
    assert any(r["stale"] for r in rows)
    # An asked-and-waiting lane stays stale until it is verified, regardless of
    # how many calendar days have elapsed since the request.
    assert any(
        r["stale"]
        and r["verified"] == "asked-and-waiting"
        for r in rows
    )