#!/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, an exhausted budget or the kill switch runs the revert steps of
                         the mutated steps in the admission's declared order (or stops, as it says), each
                         proving the restored state with its own 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 StepFailed(Exception):
    """A definite, known-bad outcome (e.g. a rollout that never became ready): handled like a failed check.
    Operations outside this module may instead set `definite_failure = True` on their exception class."""
    definite_failure = True


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,
                error: BaseException | None = None) -> None:
        entry = {"id": step["id"], "kind": kind, "operation": step["operation"], "target": step["target"],
                 "consumer": step["consumer"], "outcome": outcome, "checks": checks or []}
        if error is not None:
            entry["error_type"] = type(error).__name__  # the type only: messages can carry values
        self.steps.append(entry)

    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. Revert steps run in the order the admission declares them (the author knows
        the recovery order: data before delivery before restart), limited to those whose forward step ran.
        Each must pass its own checks; an irreversible step that ran makes the revert fail."""
        mutated = [s for s in done if s["mutates"]]
        if any(s.get("irreversible") for s in mutated):
            return False
        wanted = {s["revert"] for s in mutated}
        for revert in self.spec["plan"].get("revert_steps", []):
            if revert["id"] not in wanted:
                continue
            try:
                self.d.operations[revert["operation"]](revert, ctx)
            except Exception as error:
                self._record(revert, "revert", "error", error=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 Exception as error:
                if getattr(error, "definite_failure", False) and step["mutates"]:
                    # known-bad state: treat like a failed check, the step counts as done for the revert
                    done.append(step)
                    self._record(step, "step", "failed", error=error)
                    if policy["on_check_failure"] == "revert":
                        return "reverted" if self._revert(done, ctx) else "revert_failed"
                    return "stopped"
                self._record(step, "step", "unknown", error=error)  # outcome not established: stop
                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}