"""Lab 01 — The control plane: KYA, policy-as-code, continuous authorization.

Build the layer that knows every agent in the bank.

Work top to bottom; each section's tests pass once that section is done:

  1.  classification ordering
  2.  rule matching
  3.  the policy engine — default-deny, deny-overrides
  4.  bundle integrity — signing and validation
  5.  distribution and fail-static
  6.  the agent registry — the KYA database
  7.  authorization-aware discovery
  8.  posture checks
  9.  continuous authorization
  10. evaluation as an authorization input
  11. tracing and lineage

    pytest                     # against your work
    LAB_MODULE=solution pytest # against the reference

Determinism rules: the clock is injected, every collection you return is sorted, and
identifiers are derived (a counter, a hash) rather than random.
"""

from __future__ import annotations

import hashlib
import hmac
import json
from dataclasses import dataclass, field, replace
from enum import Enum
from typing import Callable, Dict, Iterable, List, Mapping, Optional, Sequence, Set, Tuple

# ======================================================================================
# 1. The decision inputs — subject, action, resource, environment
# ======================================================================================


_CLASSIFICATION_ORDER = ("public", "internal", "confidential", "restricted")


def classification_rank(name: str) -> int:
    """Return the ordinal of ``name`` in ``_CLASSIFICATION_ORDER``.

    TODO: raise ``ValueError`` for an unknown name. Do NOT default to a safe value:
    an unknown classification is a bug in the caller, and silently treating it as
    "public" is the wrong kind of quiet.
    """
    raise NotImplementedError


@dataclass(frozen=True)
class Subject:
    """The BLENDED principal: an agent acting for a user.

    Both halves constrain what is allowed, and both appear in the decision record. A
    subject with an agent and no user is a workload acting on its own behalf — a
    different, narrower thing, and the policy must be able to tell.
    """

    agent_id: str
    user_id: Optional[str]
    tenant: str
    scopes: Tuple[str, ...] = ()
    delegation_chain: Tuple[str, ...] = ()
    clearance: str = "internal"

    @property
    def is_user_scoped(self) -> bool:
        # TODO: true when a user is present.
        raise NotImplementedError


@dataclass(frozen=True)
class Resource:
    resource_type: str                  # "tool" | "data" | "model"
    resource_id: str
    tenant: str
    classification: str = "internal"
    barrier: Optional[str] = None


@dataclass(frozen=True)
class Environment:
    """Everything about *now* that the policy may read.

    ``anomaly_score`` and ``evaluation_age_ticks`` are the posture signals: they are what
    make authorization *continuous* rather than a fact established at session start.
    """

    tick: int = 0
    channel: str = "api"
    anomaly_score: float = 0.0
    evaluation_age_ticks: int = 0
    approvals: Tuple[str, ...] = ()
    step_index: int = 1
    amount_micros: int = 0


@dataclass(frozen=True)
class Request:
    subject: Subject
    action: str
    resource: Resource
    environment: Environment = Environment()


# ======================================================================================
# 2. Policy — default-deny, deny-overrides
# ======================================================================================


class Effect(str, Enum):
    ALLOW = "allow"
    DENY = "deny"


#: A condition reads the whole request and returns a boolean. Keeping conditions as
#: callables avoids writing an expression parser — the EVALUATION SEMANTICS are the
#: lesson, not the grammar.
Condition = Callable[["Request"], bool]


@dataclass(frozen=True)
class Rule:
    """One policy rule.

    ``actions``/``resource_types``/``tenants`` are cheap structural matches; ``condition``
    is the expressive part. Empty tuples mean "any".
    """

    name: str
    effect: Effect
    actions: Tuple[str, ...] = ()
    resource_types: Tuple[str, ...] = ()
    tenants: Tuple[str, ...] = ()
    condition: Optional[Condition] = None
    reason: str = ""

    def matches(self, request: Request) -> bool:
        """TODO: ALL populated facets must match — not any.

        An empty tuple means "any". A ``None`` condition means "no extra constraint".
        Getting this wrong in the `any` direction produces a rule that fires far more
        often than its author intended, which for a DENY is merely annoying and for an
        ALLOW is a hole.
        """
        raise NotImplementedError


def _action_matches(pattern: str, action: str) -> bool:
    """TODO: exact match, or a single trailing wildcard.

    ``payments.*`` matches ``payments.release`` but ``pay.*`` must NOT match
    ``payments.release`` — a prefix check on the wrong string is the classic bug here.
    """
    raise NotImplementedError


@dataclass(frozen=True)
class Decision:
    """The artifact. Not a boolean — a boolean cannot be shown to an examiner.

    ``policy_version`` is the field that makes this evidence: six months later, "which
    policy allowed this?" has an answer that can be looked up.
    """

    effect: Effect
    reason: str
    rule_name: str
    policy_version: str
    obligations: Tuple[str, ...] = ()
    matched_rules: Tuple[str, ...] = ()
    evaluated_at: int = 0

    @property
    def allowed(self) -> bool:
        raise NotImplementedError


# ======================================================================================
# 3. The policy bundle — versioned, signed, atomically activated
# ======================================================================================


class BundleError(Exception):
    pass


@dataclass(frozen=True)
class PolicyBundle:
    """A versioned, signed set of rules.

    Signing matters more than it looks: **atomic activation of a verified bundle** is the
    mechanism that makes fail-static real.
    """

    version: str
    rules: Tuple[Rule, ...]
    created_tick: int
    signature: str = ""

    def digest(self) -> str:
        """TODO: a stable sha256 hex digest over the version and the rules' STRUCTURE.

        Serialize with ``sort_keys=True`` so two equal bundles hash equally. Conditions
        are callables and cannot be hashed — record only whether one is present, and note
        the limit: a production bundle carries Rego or Cedar *source*, which can be.
        """
        raise NotImplementedError

    def sign(self, secret: bytes) -> "PolicyBundle":
        """TODO: return a copy carrying an HMAC-SHA256 of ``digest()``."""
        raise NotImplementedError

    def verify(self, secret: bytes) -> None:
        """TODO: raise ``BundleError`` when unsigned or when the signature does not match.

        Compare with ``hmac.compare_digest``, never ``==``.
        """
        raise NotImplementedError

    def validate(self) -> None:
        """TODO: structural checks. Raise ``BundleError`` on:

          * duplicate rule names (report them sorted);
          * an ALLOW rule with no actions, no resource types, no tenants and no condition
            — an unconditional allow-everything, which is never intended and is silent.

        An unconditional DENY-everything is legitimate: it is the panic bundle.
        """
        raise NotImplementedError


# ======================================================================================
# 4. The policy decision point
# ======================================================================================


class PolicyEngine:
    """Default-deny, deny-overrides.

    Two combining rules, both non-negotiable in a bank:

      * **default-deny** — no matching rule means DENY;
      * **deny-overrides** — any matching DENY beats every ALLOW, and evaluation does not
        stop at the first allow. Order-independence is what makes a policy set reviewable.
    """

    def __init__(self, bundle: PolicyBundle) -> None:
        # TODO: validate the bundle before accepting it, then store it.
        raise NotImplementedError

    def evaluate(self, request: Request) -> Decision:
        """TODO: collect every matching rule, then:

          * any DENY  -> DENY, naming the FIRST deny in bundle order (deterministic);
          * else any ALLOW -> ALLOW, naming the first allow;
          * else DENY with rule name ``"-"`` and a reason mentioning "default deny".

        Every returned decision carries ``policy_version``, ``matched_rules`` (sorted, all
        of them, not just the winner) and ``evaluated_at`` from the request environment.
        """
        raise NotImplementedError


# ======================================================================================
# 5. Bundle distribution — and the third posture
# ======================================================================================


class Posture(str, Enum):
    """What the data plane does when the control plane is unreachable.

    FAIL_OPEN is a security hole. FAIL_SHUT makes the control plane's availability
    multiply into the data plane's — the Phase 00 composition trap. FAIL_STATIC is the
    option people forget exists, and it is the right one.
    """

    FAIL_OPEN = "fail_open"
    FAIL_SHUT = "fail_shut"
    FAIL_STATIC = "fail_static"


@dataclass(frozen=True)
class DistributorStatus:
    version: str
    last_successful_activation: int
    staleness_ticks: int
    stale: bool
    hard_stopped: bool


class BundleDistributor:
    """Polls for a bundle, verifies it, activates it atomically.

    Three thresholds are the design: the refresh interval (how often we try), the
    staleness alarm (how long before somebody is told) and the hard stop (how long before
    we refuse to serve).
    """

    def __init__(self, *, secret: bytes, now: Callable[[], int],
                 initial: PolicyBundle,
                 staleness_alarm_ticks: int = 300,
                 hard_stop_ticks: int = 1800,
                 posture: Posture = Posture.FAIL_STATIC) -> None:
        # TODO: verify AND validate the initial bundle before accepting it — a
        # distributor that starts from an unverified bundle has no root of trust.
        # Record the activation time from the injected clock; keep an alarm list and a
        # failure counter.
        raise NotImplementedError

    @property
    def active(self) -> PolicyBundle:
        raise NotImplementedError

    def offer(self, bundle: PolicyBundle) -> bool:
        """TODO: try to activate. Return True on success, False on rejection.

        Reject and keep the previous bundle live when the offered one is unsigned,
        tampered, structurally invalid, or OLDER than the active one (a replayed old
        bundle is a policy rollback). Every rejection appends an alarm and increments the
        failure counter, and must NOT reset the staleness clock.

        That is the whole of fail-static: a bad push degrades to STALE, never to BROKEN.
        """
        raise NotImplementedError

    def status(self) -> DistributorStatus:
        """TODO: age since the last SUCCESSFUL activation, plus the two threshold flags."""
        raise NotImplementedError

    def engine(self) -> PolicyEngine:
        """TODO: return an engine over the active bundle — but raise ``BundleError``
        mentioning "hard stop" once the age reaches ``hard_stop_ticks``.

        This is the one case where fail-static becomes fail-shut, deliberately and with a
        stated threshold.
        """
        raise NotImplementedError


# ======================================================================================
# 6. The agent registry — the KYA database
# ======================================================================================


class AgentState(str, Enum):
    DRAFT = "draft"
    APPROVED = "approved"
    ACTIVE = "active"
    SUSPENDED = "suspended"
    RETIRED = "retired"


AGENT_TRANSITIONS: Mapping[AgentState, frozenset] = {
    # TODO: fill in the legal transitions.
    #   draft     -> approved, retired
    #   approved  -> active, retired
    #   active    -> suspended, retired
    #   suspended -> active, retired
    #   retired   -> (terminal; omit the key entirely)
}


@dataclass(frozen=True)
class AgentRecord:
    """Everything KYA must answer, at runtime, for every agent in production."""

    agent_id: str
    owner: str                              # a HUMAN
    tenant: str
    permitted_tools: Tuple[str, ...]
    max_classification: str = "internal"
    model_deployment: str = ""
    model_version: str = ""                 # pinned; an unpinned model is a finding
    autonomy_band: str = "read_only"        # read_only | assisted | autonomous
    state: AgentState = AgentState.DRAFT
    last_evaluation_tick: int = 0
    evaluation_score: float = 0.0
    max_action_micros: int = 0


class AgentRegistry:
    """The authoritative inventory. The reason the platform exists."""

    def __init__(self) -> None:
        self._agents: Dict[str, AgentRecord] = {}

    def register(self, record: AgentRecord) -> AgentRecord:
        """TODO: reject a duplicate id, a record with no human owner, and a record with
        no pinned model version (an unpinned model changes under you). Validate the
        classification. New agents start in DRAFT — do not silently promote."""
        raise NotImplementedError

    def get(self, agent_id: str) -> AgentRecord:
        """TODO: raise ``KeyError`` naming the unknown agent."""
        raise NotImplementedError

    def transition(self, agent_id: str, target: AgentState) -> AgentRecord:
        """TODO: enforce ``AGENT_TRANSITIONS``; raise ``ValueError`` mentioning
        "not a legal transition" otherwise. Records are frozen — build a new one."""
        raise NotImplementedError

    def record_evaluation(self, agent_id: str, *, tick: int, score: float) -> AgentRecord:
        raise NotImplementedError

    def by_owner(self, owner: str) -> List[AgentRecord]:
        """TODO: sorted by agent id — the access-review query."""
        raise NotImplementedError

    def all(self) -> List[AgentRecord]:
        raise NotImplementedError


# ======================================================================================
# 7. Tool registry and authorization-aware capability discovery
# ======================================================================================


class SideEffect(str, Enum):
    READ = "read"
    WRITE_IDEMPOTENT = "write_idempotent"
    WRITE_NON_IDEMPOTENT = "write_non_idempotent"
    IRREVERSIBLE = "irreversible"


@dataclass(frozen=True)
class ToolRecord:
    tool_id: str
    description: str
    side_effect: SideEffect
    required_scopes: Tuple[str, ...] = ()
    classification: str = "internal"
    tenants: Tuple[str, ...] = ()
    owner: str = "unknown"


class ToolRegistry:
    def __init__(self) -> None:
        self._tools: Dict[str, ToolRecord] = {}

    def publish(self, tool: ToolRecord) -> None:
        """TODO: reject a duplicate id; validate the classification."""
        raise NotImplementedError

    def get(self, tool_id: str) -> Optional[ToolRecord]:
        raise NotImplementedError

    def all(self) -> List[ToolRecord]:
        """TODO: sorted by tool id."""
        raise NotImplementedError


@dataclass(frozen=True)
class Capability:
    tool_id: str
    description: str
    side_effect: SideEffect


@dataclass(frozen=True)
class PostureFinding:
    """A KYA check that failed, and how hard it bites.

    The distinction is the whole subtlety of posture. Some failures are **categorical** —
    a suspended agent should do nothing at all. Others are **graduated**: an agent whose
    evaluation went stale this morning is not dangerous to *read* with, but it has no
    business releasing a payment.

    Collapsing the two fails in both directions. Treat everything as categorical and a
    routine eval lapse takes the fleet down, which teaches operators to raise the
    thresholds until they never fire. Treat everything as graduated and a suspended agent
    keeps reading customer data.
    """

    check: str
    detail: str
    blocks_reads: bool

    def __str__(self) -> str:
        return self.detail


class ControlPlane:
    """The composition: registries + policy + posture, answering two questions.

      * ``discover`` — what may this agent see RIGHT NOW?
      * ``authorize`` — may it do this specific thing RIGHT NOW?
    """

    def __init__(self, *, agents: AgentRegistry, tools: ToolRegistry,
                 distributor: BundleDistributor, now: Callable[[], int],
                 max_evaluation_age_ticks: int = 1000,
                 anomaly_block_writes: float = 0.5,
                 anomaly_block_all: float = 0.8) -> None:
        # TODO: store the collaborators and thresholds; open an empty decision log.
        raise NotImplementedError

    # -- KYA posture ---------------------------------------------------------------
    def posture_checks(self, record: AgentRecord,
                       environment: Environment) -> List[PostureFinding]:
        """TODO: the KYA questions, evaluated NOW. Return every failure, stable order:

        | check | condition | `blocks_reads` |
        |---|---|---|
        | `state` | the agent is not ACTIVE | True |
        | `owner` | the agent has no owner | True |
        | `model` | the agent has no pinned model version | True |
        | `evaluation` | `tick - last_evaluation_tick` exceeds the max age | **False** |
        | `anomaly` | score >= `anomaly_block_all` | True |
        | `anomaly` | score >= `anomaly_block_writes` | **False** |

        The two anomaly rows are exclusive — emit one finding, not two.

        Return every failure, not the first: an operator fixing one at a time is an
        operator who learns about the next one on the next attempt.
        """
        raise NotImplementedError

    def blocking_findings(self, record: AgentRecord, environment: Environment, *,
                          high_impact: bool) -> List[PostureFinding]:
        """TODO: the subset that actually stops THIS request — everything that blocks
        reads, plus everything else when the action is high-impact."""
        raise NotImplementedError

    # -- discovery ------------------------------------------------------------------
    def discover(self, subject: Subject,
                 environment: Environment = Environment()) -> List[Capability]:
        """TODO: the tools this subject may use right now.

        Filter in order — registry membership, posture, tenant, classification ceiling,
        held scopes, then a real policy probe per surviving tool. Discovery is an
        AUTHORIZATION DECISION, not a lookup: a capability the principal cannot use must
        not be visible, because its name is information and a model that can see a tool
        will eventually try to call it.

        Posture degrades the answer the same way it degrades authorization. A
        **categorical** failure returns nothing at all. A **graduated** one returns the
        READ capabilities only — so the model is never offered a tool the next call
        would refuse.
        """
        raise NotImplementedError

    # -- authorization ---------------------------------------------------------------
    def authorize(self, request: Request, *, high_impact: bool = False) -> Decision:
        """TODO: posture first (cheap, and it short-circuits), then policy.

          * unknown agent -> DENY, rule ``"kya:unregistered"``;
          * blocking posture findings -> DENY, rule ``"kya:posture"``, reason joining
            their details with ``"; "``;
          * otherwise the distributor's engine decides.

        Every outcome is appended to the decision log, and every one carries a policy
        version — including the posture denials, which is easy to forget precisely
        because policy was never consulted, and is exactly the record an examiner asks
        for.
        """
        raise NotImplementedError


# ======================================================================================
# 8. Continuous authorization
# ======================================================================================


@dataclass(frozen=True)
class Lease:
    """A cached decision with a TTL.

    Caching is not an optimization here; without it the PDP sits on the synchronous path
    of every step and its availability multiplies into the platform's. The TTL is the
    revocation-latency budget, stated as a number.
    """

    decision: Decision
    issued_at: int
    ttl_ticks: int
    request_fingerprint: str
    agent_id: str

    def is_live(self, tick: int) -> bool:
        """TODO: live while ``tick - issued_at < ttl_ticks`` — expiry lands ON the TTL."""
        raise NotImplementedError


def fingerprint(request: Request) -> str:
    """TODO: a stable digest of what makes two requests 'the same decision'.

    Include the subject (agent, user, tenant, sorted scopes), the action and the resource
    (type, id, tenant, classification). Deliberately EXCLUDE the volatile environment —
    tick, anomaly score — because including them would make every lease a miss. Which is
    precisely why a lease also has a TTL and an explicit revocation path.

    Use a derived digest (``hashlib.blake2b``), never ``hash()``: Python salts string
    hashing per process, so ``hash()`` is not stable across runs.
    """
    raise NotImplementedError


class ContinuousAuthorizer:
    """Re-evaluates during a long-running task, not only at session start.

    An agent task can run for an hour while entitlements, agent posture and risk signals
    all change. A decision made at minute zero and cached for the session is a decision
    about conditions that no longer hold.
    """

    def __init__(self, *, control_plane: ControlPlane, now: Callable[[], int],
                 lease_ttl_ticks: int = 60) -> None:
        # TODO: leases keyed by fingerprint, a revoked-agent set, and two counters
        # (``evaluations``, ``lease_hits``) so the tests can see which path ran.
        raise NotImplementedError

    def authorize(self, request: Request, *, high_impact: bool = False) -> Decision:
        """TODO: serve from a live lease, otherwise re-evaluate.

        Four conditions must ALL hold to reuse a lease:
          * one exists and is live at ``request.environment.tick``;
          * the action is not high-impact (those are never leased, in either direction);
          * the agent is not revoked;
          * the lease's policy version equals the distributor's ACTIVE version — a new
            bundle invalidates every live lease, which is what makes a policy push take
            effect now rather than one TTL from now.

        Pass ``high_impact`` through to the control plane: it is what turns a graduated
        posture finding into a block.

        Cache only ALLOW decisions, and only when not high-impact. Caching a deny is
        tempting and wrong: it delays reinstatement by a TTL.
        """
        raise NotImplementedError

    def revoke_agent(self, agent_id: str) -> int:
        """TODO: the kill switch. Return the number of leases dropped.

        Two halves, and both are needed. Dropping the leases is what makes revocation
        *fast*; the revoked set is what keeps it fast, because a request arriving one tick
        later would otherwise mint a fresh lease from a control plane that has not yet
        caught up.

        'We suspended it but it kept acting for sixty seconds' is a sentence you do not
        want to say to an examiner.
        """
        raise NotImplementedError

    def reinstate_agent(self, agent_id: str) -> None:
        raise NotImplementedError


# ======================================================================================
# 9. Evaluation pipeline — quality as an authorization input
# ======================================================================================


@dataclass(frozen=True)
class EvalCase:
    case_id: str
    kind: str                 # "golden" | "safety" | "regression"
    expected: str


@dataclass(frozen=True)
class EvalResult:
    agent_id: str
    tick: int
    total: int
    passed: int
    safety_failures: int

    @property
    def score(self) -> float:
        """TODO: pass rate; 0.0 for an empty suite rather than a ZeroDivisionError."""
        raise NotImplementedError


@dataclass(frozen=True)
class PromotionVerdict:
    promoted: bool
    reasons: Tuple[str, ...]


class EvaluationPipeline:
    """Golden sets, safety suites, regression gates — and the link that matters.

    Evaluation status is an *authorization input* (via the posture check), not a
    dashboard. That single link is what makes quality a control rather than a report.
    """

    def __init__(self, *, registry: AgentRegistry, now: Callable[[], int],
                 min_score: float = 0.85, allow_safety_failures: int = 0) -> None:
        raise NotImplementedError

    def run(self, agent_id: str, cases: Sequence[EvalCase],
            respond: Callable[[EvalCase], str]) -> EvalResult:
        """TODO: score the suite, count safety failures separately, append to history,
        and PUSH the freshness back into the registry — that write is the whole point."""
        raise NotImplementedError

    def gate(self, result: EvalResult, *,
             baseline: Optional[EvalResult] = None) -> PromotionVerdict:
        """TODO: return every blocking reason, not the first:

          * more safety failures than allowed — disqualifying REGARDLESS of score, because
            an aggregate that averages away a safety failure is a gate that does not gate;
          * score below ``min_score``;
          * a score strictly below the baseline (equal is not a regression).
        """
        raise NotImplementedError


# ======================================================================================
# 10. Tracing and lineage at agent and tool granularity
# ======================================================================================


@dataclass(frozen=True)
class Span:
    """One unit of work, with the identity and decision attached."""

    span_id: str
    parent_id: Optional[str]
    trace_id: str
    name: str
    kind: str                          # "agent" | "tool" | "model" | "policy"
    agent_id: str
    user_id: Optional[str]
    tenant: str
    started_at: int
    ended_at: int
    policy_version: str = ""
    decision: Optional[str] = None
    model_version: str = ""
    attributes: Mapping[str, str] = field(default_factory=dict)

    @property
    def duration(self) -> int:
        raise NotImplementedError


class Tracer:
    def __init__(self, *, now: Callable[[], int]) -> None:
        # TODO: a span list and a counter. Span ids are DERIVED (``span-1``, ``span-2``),
        # never random — a test that cannot predict an id cannot assert on it.
        raise NotImplementedError

    def record(self, *, trace_id: str, parent_id: Optional[str], name: str, kind: str,
               subject: Subject, started_at: int, policy_version: str = "",
               decision: Optional[str] = None, model_version: str = "",
               **attributes: str) -> Span:
        """TODO: build and store a span, ending it at ``now()``.

        Take ``started_at`` from the caller — but note the caller must read it from the
        SAME injected clock. Two clocks is how you get negative durations in production.
        """
        raise NotImplementedError

    def trace(self, trace_id: str) -> List[Span]:
        raise NotImplementedError

    def lineage(self, trace_id: str) -> Dict[str, object]:
        """TODO: what produced this outcome — the agents, users, tools, model versions,
        policy versions, decisions and span count for one trace, each sorted.

        This is the query an examiner's question reduces to, and it is only answerable
        because every span carries identity and version.
        """
        raise NotImplementedError


# ======================================================================================
# Worked example
# ======================================================================================


def main() -> None:  # pragma: no cover
    """TODO (optional): once the tests pass, build the eight-section demo.

    Compare against ``python solution.py`` — but only after your own runs.
    """
    raise NotImplementedError


if __name__ == "__main__":  # pragma: no cover
    main()