#!/usr/bin/env python3
"""Bounded executor: core runner (NK-WP-0056-T11).

Runs one Bounded Execution Admission (canon/standards/bounded-execution-admission_v0.1.md) exactly once:

  preflight (no effect)  the admission is valid (BEA-1..14); the executed source matches its pins; the
                         trust states hold; every operation and check is known; the approval carried by
                         approval-engine binds this admission's digest, names this executor as actor, was
                         approved by listed HUMAN approvers at aal2, is unconsumed, and the window is open
  consume                once, with request_digest = admission digest and decision_id = run id; an
                         unclear outcome is resolved by reading the approval back, never by consuming again
  started                `execution.run.started` to the audit sink; no custody, no run
  steps                  in plan order, one consumer at a time; checks after each mutating step, after
                         their settle time; the kill switch and the budget are checked between steps
  failure                a failed check or an exhausted budget reverts the mutated steps in reverse
                         order (or stops, as the admission says) and re-runs their checks; an error of
                         unknown outcome stops without guessing; a failed revert stops and alerts
  receipt                one `execution.attribution.receipt` per run (PDP `not-evaluated` in the pilot)

Every effect goes through an injected interface: Carrier, operation and check callables, TrustOracle,
AuditSink, Alert, clocks. The runner never prints, logs or stores a value an operation handles.
"""

from __future__ import annotations

import dataclasses
import datetime as dt
import hashlib
import importlib.util
import json
import pathlib
import time
import uuid
from typing import Callable, Protocol

HERE = pathlib.Path(__file__).resolve().parent
REPO = HERE.parent
_spec = importlib.util.spec_from_file_location("bea_validate", REPO / "tools/bounded-execution-admission/validate.py")
bea = importlib.util.module_from_spec(_spec)
_spec.loader.exec_module(bea)

ACTION = "bounded-execution"
STARTED = "execution.run.started"
RECEIPT = "execution.attribution.receipt"


class Refused(Exception):
    """Refused before any effect."""


class Unknown(Exception):
    """An operation's outcome is unknown: stop, do not guess, do not revert blindly."""


class Carrier(Protocol):
    def get(self, approval_id: str) -> dict: ...
    def consume(self, approval_id: str, request_digest: str, decision_id: str) -> dict: ...


class AuditSink(Protocol):
    def emit(self, event: dict) -> None: ...


Operation = Callable[[dict, "Context"], None]   # (step, context) -> None, raises Unknown on unclear outcome
Check = Callable[[dict, "Context"], bool]       # (check, context) -> passed


@dataclasses.dataclass
class Config:
    executor_identity: str               # must equal spec.executor.identity and the approval's actor
    runtime_principal: str               # e.g. system:serviceaccount:sso:nk-bind-executor
    tenant: str
    environment: str
    approver_subjects: dict[str, str]    # admission approver name -> carrier subject_id (no guessing)
    source_root: pathlib.Path = REPO
    clock_source: str = "system-utc"


@dataclasses.dataclass
class Context:
    """What operations and checks may see. Values they fetch stay inside them."""
    run_id: str
    admission_id: str
    started_monotonic: float
    data: dict = dataclasses.field(default_factory=dict)  # operation-private state (e.g. versions), never emitted


@dataclasses.dataclass
class Deps:
    carrier: Carrier
    operations: dict[str, Operation]
    checks: dict[str, Check]
    trust: Callable[[str], bool]
    audit: AuditSink
    alert: Callable[[str, dict], None]
    should_stop: Callable[[], bool] = lambda: False
    wall: Callable[[], dt.datetime] = lambda: dt.datetime.now(dt.UTC)
    monotonic: Callable[[], float] = time.monotonic
    sleep: Callable[[float], None] = time.sleep


def iso(t: dt.datetime) -> str:
    return t.astimezone(dt.UTC).isoformat(timespec="seconds").replace("+00:00", "Z")


class Run:
    def __init__(self, admission: dict, approval_id: str, config: Config, deps: Deps, schema: dict | None = None):
        self.doc = admission
        self.spec = admission.get("spec", {})
        self.approval_id = approval_id
        self.cfg = config
        self.d = deps
        self.schema = schema or json.loads(bea.SCHEMA.read_text())
        self.run_id = str(uuid.uuid4())
        self.steps: list[dict] = []
        self.approver_subject: str | None = None
        self.consumed_at: str | None = None
        self.start_wall: dt.datetime | None = None
        self.start_mono: float | None = None

    # ------------------------------------------------------------------ preflight (no effect)
    def preflight(self) -> None:
        report = bea.evaluate(self.doc, self.schema)
        if not report["valid"]:
            raise Refused("admission_invalid:" + ",".join(sorted({f["rule"] for f in report["findings"]})))
        spec = self.spec
        if spec["executor"]["identity"] != self.cfg.executor_identity:
            raise Refused("not_this_executor")
        for pinned in spec["plan"]["source"]["files"]:
            path = self.cfg.source_root / pinned["path"]
            if not path.is_file() or hashlib.sha256(path.read_bytes()).hexdigest() != pinned["sha256"]:
                raise Refused("source_drift")
        for state in spec["trust_requires"]:
            if not self.d.trust(state):
                raise Refused("trust_not_satisfied:" + state)
        for step in spec["plan"]["steps"] + spec["plan"].get("revert_steps", []):
            if step["operation"] not in self.d.operations:
                raise Refused("unknown_operation")
        for check in spec["checks"]:
            if check["id"] not in self.d.checks:
                raise Refused("unknown_check")
        self._check_window()
        self._check_approval(self.d.carrier.get(self.approval_id), consumed_ok=False)

    def _check_window(self) -> None:
        now = self.d.wall()
        start, end = (bea.parse_time(self.spec["window"][k]) for k in ("not_before", "not_after"))
        if not start <= now < end:
            raise Refused("outside_window")
        if now + dt.timedelta(seconds=self.spec["budget"]["wall_time_seconds"]) > end:
            raise Refused("window_shorter_than_budget")  # a run must not spill past its admission

    def _check_approval(self, approval: dict, consumed_ok: bool) -> None:
        spec = self.spec
        binding = approval.get("binding") or {}
        target = binding.get("target") or {}
        if (binding.get("action") != ACTION or binding.get("actor") != self.cfg.executor_identity
                or target.get("admission_id") != spec["id"]
                or target.get("admission_digest") != bea.admission_digest(self.doc)
                or binding.get("human_control") is not True):
            raise Refused("approval_binding")
        status = approval.get("status")
        if status != "approved" and not (consumed_ok and status == "consumed"):
            raise Refused("approval_status")
        if not consumed_ok and approval.get("consumption"):
            raise Refused("approval_already_consumed")
        allowed = {self.cfg.approver_subjects.get(name) for name in spec["approval"]["approvers"]} - {None}
        entries = approval.get("entries") or []
        if not entries:
            raise Refused("approval_no_entries")
        for entry in entries:  # aal2 is recorded, not enforced, by approval-engine today: check every entry
            if (entry.get("subject_id") not in allowed or str(entry.get("principal_type", "")).lower() != "human"
                    or entry.get("assurance") != "aal2" or entry.get("subject_id") == self.cfg.executor_identity):
                raise Refused("approval_entry")
        self.approver_subject = entries[0]["subject_id"]
        window_end = bea.parse_time(spec["window"]["not_after"])
        expires = (approval.get("validity") or {}).get("expires_at")
        if expires and bea.parse_time(expires) < window_end:
            raise Refused("approval_expires_before_window_end")

    # ------------------------------------------------------------------ consume (the one irreversible act)
    def consume(self) -> None:
        self._check_window()  # again, right before the one irreversible act
        digest = bea.admission_digest(self.doc)
        try:
            response = self.d.carrier.consume(self.approval_id, digest, self.run_id)
        except Exception:
            response = None  # unclear outcome: read back, never consume twice
        if response is not None:
            if (response.get("status") != "consumed" or response.get("request_digest") != digest
                    or response.get("decision_id") != self.run_id or response.get("idempotent") is not False):
                raise Refused("consume_response")
            self.consumed_at = response.get("consumed_at")
        approval = self.d.carrier.get(self.approval_id)
        self._check_approval(approval, consumed_ok=True)
        consumption = approval.get("consumption") or {}
        if approval.get("status") != "consumed" or consumption.get("request_digest") != digest \
                or consumption.get("decision_id") != self.run_id:
            raise Refused("consume_not_confirmed")
        self.consumed_at = consumption.get("consumed_at", self.consumed_at)

    # ------------------------------------------------------------------ execution
    def _budget_left(self) -> float:
        return self.spec["budget"]["wall_time_seconds"] - (self.d.monotonic() - self.start_mono)

    def _record(self, step: dict, kind: str, outcome: str, checks: list[dict] | None = None) -> None:
        self.steps.append({"id": step["id"], "kind": kind, "operation": step["operation"], "target": step["target"],
                           "consumer": step["consumer"], "outcome": outcome, "checks": checks or []})

    def _run_checks(self, ids: list[str], ctx: Context) -> tuple[bool, list[dict]]:
        results = []
        for cid in ids:
            check = next(c for c in self.spec["checks"] if c["id"] == cid)
            settle = check.get("settle_seconds", 0)
            if settle:
                if self._budget_left() < settle + check["timeout_seconds"]:
                    results.append({"id": cid, "result": "budget"})
                    return False, results
                self.d.sleep(settle)
            began = self.d.monotonic()
            try:
                passed = bool(self.d.checks[cid](check, ctx))
            except Exception:
                passed = False
            if self.d.monotonic() - began > check["timeout_seconds"]:
                passed = False
            results.append({"id": cid, "result": "pass" if passed else "fail"})
            if not passed:
                return False, results
        return True, results

    def _revert(self, done: list[dict], ctx: Context) -> bool:
        """Undo the mutated steps in reverse order; each revert step must pass its own checks."""
        reverts = {s["id"]: s for s in self.spec["plan"].get("revert_steps", [])}
        for step in reversed(done):
            if not step["mutates"]:
                continue
            if step.get("irreversible"):
                return False
            revert = reverts[step["revert"]]
            try:
                self.d.operations[revert["operation"]](revert, ctx)
            except Exception:
                self._record(revert, "revert", "error")
                return False
            ok, results = self._run_checks(revert.get("checks_after", []), ctx)
            self._record(revert, "revert", "done" if ok else "check_failed", results)
            if not ok:
                return False
        return True

    def _attempt(self, ctx: Context) -> str:
        """One pass over the plan. Returns completed | reverted | revert_failed | stopped | unknown | budget."""
        policy = self.spec["stop_and_revert"]
        done: list[dict] = []
        for step in self.spec["plan"]["steps"]:
            if self.d.should_stop() or self._budget_left() <= 0:
                reason = "stopped" if self.d.should_stop() else "budget"
                action = policy["on_budget_exhausted"] if reason == "budget" else "revert"
                if action == "revert" and done:
                    return reason if self._revert(done, ctx) else "revert_failed"
                return reason
            try:
                self.d.operations[step["operation"]](step, ctx)
            except Unknown:
                self._record(step, "step", "unknown")
                return "unknown"
            except Exception:
                self._record(step, "step", "unknown")
                return "unknown"
            done.append(step)
            ok, results = self._run_checks(step.get("checks_after", []), ctx) if step["mutates"] else (True, [])
            self._record(step, "step", "done" if ok else "check_failed", results)
            if not ok:
                if policy["on_check_failure"] == "revert":
                    return "reverted" if self._revert(done, ctx) else "revert_failed"
                return "stopped"
        return "completed"

    # ------------------------------------------------------------------ evidence
    def _event(self, kind: str, body: dict) -> dict:
        return {"id": f"run:{self.run_id}:{'started' if kind == STARTED else 'receipt'}", "type": kind,
                "correlation_id": self.run_id, "subject": f"admission:{self.spec['id']}",
                "source": "net-kingdom.bounded-executor", "tenant": self.cfg.tenant, "data": body}

    def _receipt(self, outcome: str, attempts: int) -> dict:
        end = self.d.wall()
        return self._event(RECEIPT, {
            "run_id": self.run_id,
            "initiator": {"sub": self.approver_subject, "role": "approver"},
            "executing_workload": self.cfg.executor_identity,
            "runtime_principal": self.cfg.runtime_principal,
            "tenant": self.cfg.tenant, "environment": self.cfg.environment,
            "source_commit": self.spec["plan"]["source"]["commit"],
            "artifact_digest": self.spec["plan_digest"],
            "target": f"admission:{self.spec['id']}", "action": ACTION,
            "decision_refs": [
                {"label": "admission", "ref": self.spec["id"]},
                {"label": "admission_digest", "ref": bea.admission_digest(self.doc)},
                {"label": "approval", "ref": self.approval_id},
            ],
            "pdp": "not-evaluated",
            "time_interval": {"start": iso(self.start_wall), "end": iso(end), "clock_source": self.cfg.clock_source},
            "outcome": outcome,
            "reverted": any(s["kind"] == "revert" for s in self.steps),
            "attempts": attempts,
            "budget": {"limit": {"wall_time_seconds": self.spec["budget"]["wall_time_seconds"],
                                 "attempts": self.spec["budget"]["attempts"]},
                       "used": {"wall_time_seconds": round(self.d.monotonic() - self.start_mono, 3),
                                "attempts": attempts}},
            "steps": self.steps,
        })

    # ------------------------------------------------------------------ the run
    def execute(self) -> dict:
        try:
            self.preflight()
        except Refused as refusal:
            return {"outcome": "refused", "reason": str(refusal), "run_id": None, "effects": False}
        self.start_wall, self.start_mono = self.d.wall(), self.d.monotonic()
        try:
            self.consume()
        except Refused as refusal:  # consumed or not, nothing else may happen now
            self.d.alert("consume_unconfirmed", {"admission": self.spec["id"], "run_id": self.run_id})
            return {"outcome": "refused", "reason": str(refusal), "run_id": self.run_id, "effects": "consume-only"}
        try:
            self.d.audit.emit(self._event(STARTED, {"run_id": self.run_id, "admission": self.spec["id"],
                                                    "approval": self.approval_id, "consumed_at": self.consumed_at,
                                                    "executing_workload": self.cfg.executor_identity}))
        except Exception:
            self.d.alert("started_not_in_custody", {"admission": self.spec["id"], "run_id": self.run_id})
            return {"outcome": "stopped", "reason": "audit_unavailable", "run_id": self.run_id, "effects": "consume-only"}
        ctx = Context(self.run_id, self.spec["id"], self.start_mono)
        attempts, outcome = 0, "budget"
        while attempts < self.spec["budget"]["attempts"]:
            attempts += 1
            outcome = self._attempt(ctx)
            if outcome != "reverted" or self._budget_left() <= 0:
                break  # only a clean revert may be retried, and only within budget
        if outcome in ("unknown", "revert_failed"):
            self.d.alert(outcome, {"admission": self.spec["id"], "run_id": self.run_id})
        receipt = self._receipt(outcome, attempts)
        try:
            self.d.audit.emit(receipt)
        except Exception:
            self.d.alert("receipt_not_in_custody", {"admission": self.spec["id"], "run_id": self.run_id})
            return {"outcome": outcome, "reason": "receipt_not_in_custody", "run_id": self.run_id, "receipt": receipt}
        return {"outcome": outcome, "run_id": self.run_id, "receipt": receipt}