secrets-engine/src/secrets_engine/decisions.py

"""Decision integration.

Privileged actions require an approved decision (or approved CCR) unless the lane
is explicitly `bootstrap-only` or the caller passes --dry-run. Decisions are
looked up from State Hub by id; when the hub has no record yet (common during the
pilot) a local approval fixture under `.decisions/<ref>.yaml` can stand in, so the
end-to-end chain is testable before the canonical hub decision object exists.

No secret values are ever read from or written to a decision.
"""
from __future__ import annotations

import json
import urllib.error
import urllib.request
from dataclasses import dataclass
from pathlib import Path
from typing import Any

import yaml

from secrets_engine.catalog import CatalogEntry
from secrets_engine.errors import DecisionError

APPROVED_STATUSES = {"resolved", "approved", "accepted"}


@dataclass
class Decision:
    id: str
    title: str
    status: str
    superseded_by: str | None
    source: str          # "hub" | "local-fixture"
    review_url: str = ""
    raw: dict[str, Any] | None = None

    def is_approved(self) -> bool:
        return self.status.lower() in APPROVED_STATUSES and not self.superseded_by


def _hub_get(hub_url: str, decision_id: str) -> dict[str, Any] | None:
    url = hub_url.rstrip("/") + f"/decisions/{decision_id}"
    try:
        with urllib.request.urlopen(url, timeout=3) as resp:
            return json.loads(resp.read())
    except (urllib.error.URLError, OSError, ValueError):
        return None


def _local_fixture(repo_root: Path, ref: str) -> dict[str, Any] | None:
    path = repo_root / ".decisions" / f"{ref}.yaml"
    if not path.exists():
        return None
    try:
        return yaml.safe_load(path.read_text(encoding="utf-8")) or None
    except yaml.YAMLError as e:
        raise DecisionError(f"{path}: invalid decision fixture: {e}") from e


def resolve_decision(
    *,
    hub_url: str,
    repo_root: Path,
    decision_ref: str,
) -> Decision:
    """Resolve a decision by id (hub) or slug (local fixture). Raises if absent."""
    if not decision_ref:
        raise DecisionError("no decision reference provided")

    doc = _hub_get(hub_url, decision_ref)
    if doc:
        return Decision(
            id=doc.get("id", decision_ref),
            title=doc.get("title", ""),
            status=doc.get("status", "unknown"),
            superseded_by=doc.get("superseded_by"),
            source="hub",
            review_url=f"{hub_url.rstrip('/')}/decisions/{doc.get('id', decision_ref)}",
            raw=doc,
        )

    fixture = _local_fixture(repo_root, decision_ref)
    if fixture:
        return Decision(
            id=fixture.get("id", decision_ref),
            title=fixture.get("title", decision_ref),
            status=fixture.get("status", "unknown"),
            superseded_by=fixture.get("superseded_by"),
            source="local-fixture",
            review_url=fixture.get("review_url", ""),
            raw=fixture,
        )

    raise DecisionError(
        f"decision '{decision_ref}' not found in State Hub or local fixtures"
    )


def require_approved(entry: CatalogEntry, decision: Decision | None) -> None:
    """Enforce the lane's approval model. Raises DecisionError if not satisfied."""
    if not entry.approval_required():
        return  # bootstrap-only lane
    if decision is None:
        raise DecisionError(
            f"lane '{entry.id}' requires an approved decision; none resolved"
        )
    if decision.superseded_by:
        raise DecisionError(
            f"decision '{decision.id}' is superseded by '{decision.superseded_by}'"
        )
    if not decision.is_approved():
        raise DecisionError(
            f"decision '{decision.id}' is not approved (status='{decision.status}')"
        )