"""Closed local execution kernel v1; no installed adapters or live CLI dispatch.

An owner must acknowledge each descriptor and provide an independent provenance
verifier. Descriptive case types remain plan-only. Nothing executes shell prose.
"""

from __future__ import annotations

from contextlib import contextmanager, redirect_stderr, redirect_stdout
from dataclasses import asdict, dataclass, replace
import fcntl
import hashlib
import json
import os
from pathlib import Path
import re
import sqlite3
import threading
import uuid
from typing import Callable

import yaml
from .case_receipt_state import (
    Outcome,
    Pins,
    Receipt,
    Refused,
    StepState,
    begin,
    finish,
    recover,
    begin_rollback,
    finish_rollback,
)

IDENTIFIER = re.compile(r"[a-zA-Z0-9][a-zA-Z0-9_.:/@-]{0,191}\Z")
MAX_SOURCE_BYTES = 262144
MAX_HISTORY = 128
OUTPUT_LOCK = threading.RLock()


def bounded_source(value: bytes):
    if type(value) is not bytes or len(value) > MAX_SOURCE_BYTES:
        raise Refused("source exceeds bounded input contract")


@contextmanager
def contained_output():
    """Discard Python and inherited OS stdout/stderr; serialize process fd changes."""
    with OUTPUT_LOCK:
        saved = []
        null = os.open(os.devnull, os.O_WRONLY)
        try:
            for target in (1, 2):
                saved.append((target, os.dup(target)))
                os.dup2(null, target)
            with redirect_stdout(_Discard()), redirect_stderr(_Discard()):
                try:
                    yield
                except Exception:
                    raise Refused("owner callback failed") from None
                except BaseException:
                    raise OwnerInterrupted() from None
        finally:
            for target, original in saved:
                os.dup2(original, target)
                os.close(original)
            os.close(null)


DIGEST = re.compile(r"sha256:[0-9a-f]{64}\Z")


def digest(value: bytes) -> str:
    return "sha256:" + hashlib.sha256(value).hexdigest()


@dataclass(frozen=True, slots=True)
class Descriptor:
    version: str
    step: str
    owner: str
    adapter: str
    adapter_source: str
    actor: str
    lane: str
    target_parameter: str
    effect: str
    allowed_created: tuple[str, ...]
    retry_safe: bool
    revocation_scopes: tuple[str, ...]

    @classmethod
    def decode(cls, source: bytes) -> Descriptor:
        bounded_source(source)
        try:
            obj = json.loads(source)
            if set(obj) != set(cls.__dataclass_fields__):
                raise ValueError
            if not isinstance(obj["allowed_created"], list):
                raise ValueError
            obj["allowed_created"] = tuple(obj["allowed_created"])
            if not isinstance(obj["revocation_scopes"], list):
                raise ValueError
            obj["revocation_scopes"] = tuple(obj["revocation_scopes"])
            result = cls(**obj)
            identifiers = (
                result.step,
                result.owner,
                result.adapter,
                result.lane,
                result.target_parameter,
                *result.allowed_created,
            )
            if result.version != "warden.case.execution.v1" or not all(
                isinstance(v, str) and IDENTIFIER.fullmatch(v) for v in identifiers
            ):
                raise ValueError
            if not DIGEST.fullmatch(result.adapter_source) or result.actor not in (
                "human",
                "agent",
            ):
                raise ValueError
            if (
                result.effect not in ("read", "change", "mint")
                or type(result.retry_safe) is not bool
            ):
                raise ValueError
            scopes = set(result.revocation_scopes)
            if (
                len(scopes) != len(result.revocation_scopes)
                or not scopes <= {"parent-session", "broker-grant"}
                or "parent-session" not in scopes
            ):
                raise ValueError
            if result.effect == "mint" and "broker-grant" not in scopes:
                raise ValueError
            if len(result.allowed_created) > 32:
                raise ValueError
            if len(set(result.allowed_created)) != len(result.allowed_created):
                raise ValueError
            return result
        except (ValueError, TypeError, KeyError):
            raise Refused("invalid closed execution descriptor") from None


@dataclass(frozen=True, slots=True)
class OwnerResult:
    outcome: Outcome
    evidence: str
    created: tuple[str, ...] = ()
    revoked: bool = False
    session_revocation_evidence: str | None = None
    grant_revocation_evidence: str | None = None
    attempt_id: str = ""
    operation: str = ""


@dataclass(frozen=True, slots=True)
class Invocation:
    step: str
    owner: str
    target: str
    actor: str
    principal: str
    lane: str
    pins: Pins
    attempt_id: str = ""
    operation: str = "admission"


@dataclass(frozen=True, slots=True)
class Adapter:
    reference: str
    source: str
    invoke: Callable[[Invocation], OwnerResult]
    verify: Callable[[Invocation, OwnerResult], bool]
    rollback: Callable[[Invocation, tuple[str, ...]], OwnerResult] | None = None
    reconcile: Callable[[Invocation], OwnerResult] | None = None


class Journal:
    """Private SQLite journal plus process lock; transaction commits before dispatch."""

    def __init__(self, directory: Path):
        directory.mkdir(mode=0o700, parents=True, exist_ok=True)
        if (
            directory.is_symlink()
            or directory.stat().st_uid != os.getuid()
            or directory.stat().st_mode & 0o077
        ):
            raise Refused("journal directory is not private")
        self.directory = directory
        for name in ("journal.sqlite", "execution.lock"):
            path = directory / name
            fd = os.open(path, os.O_CREAT | os.O_RDWR | os.O_NOFOLLOW, 0o600)
            try:
                if os.fstat(fd).st_uid != os.getuid() or os.fstat(fd).st_mode & 0o077:
                    raise Refused("journal file is not private")
            finally:
                os.close(fd)
        self.db = sqlite3.connect(directory / "journal.sqlite", isolation_level=None)
        self.db.execute("PRAGMA synchronous=FULL")
        self.db.execute(
            "CREATE TABLE IF NOT EXISTS receipts (seq INTEGER PRIMARY KEY, binding TEXT NOT NULL, payload TEXT NOT NULL)"
        )

    @contextmanager
    def locked(self):
        with (self.directory / "execution.lock").open("rb") as fh:
            try:
                fcntl.flock(fh, fcntl.LOCK_EX | fcntl.LOCK_NB)
            except BlockingIOError:
                raise Refused("case execution already active") from None
            try:
                yield
            finally:
                fcntl.flock(fh, fcntl.LOCK_UN)

    def load(self, binding: str, initial: StepState) -> StepState:
        row = self.db.execute(
            "SELECT payload FROM receipts WHERE binding=? ORDER BY seq DESC LIMIT 1",
            (binding,),
        ).fetchone()
        if row is None:
            return initial
        obj = json.loads(row[0])
        expected = {
            "step_id",
            "owner",
            "pins",
            "allowed_created",
            "retry_safe",
            "receipts",
            "journal_schema",
            "evidence_digest",
            "session_revoked",
            "session_revocation_digest",
            "grant_revocation_digest",
            "attempt_id",
            "operation",
        }
        if (
            set(obj) != expected
            or obj["journal_schema"] != "warden.case.step-journal.v1"
        ):
            raise Refused("unknown journal receipt schema")
        state = StepState(
            obj["step_id"],
            obj["owner"],
            Pins(**obj["pins"]),
            tuple(obj["allowed_created"]),
            obj["retry_safe"],
            tuple(
                Receipt(r["attempt"], Outcome(r["outcome"]), tuple(r["created"]))
                for r in obj["receipts"]
            ),
        )
        if (
            state.step_id,
            state.owner,
            state.pins,
            state.allowed_created,
            state.retry_safe,
        ) != (
            initial.step_id,
            initial.owner,
            initial.pins,
            initial.allowed_created,
            initial.retry_safe,
        ):
            raise Refused("journal pins changed")
        return state

    def save(
        self,
        binding: str,
        state: StepState,
        evidence: OwnerResult | None = None,
        invocation: Invocation | None = None,
    ):
        self.db.execute("BEGIN IMMEDIATE")
        try:
            self.db.execute(
                "INSERT INTO receipts(binding,payload) VALUES (?,?)",
                (
                    binding,
                    json.dumps(
                        asdict(state)
                        | {
                            "journal_schema": "warden.case.step-journal.v1",
                            "attempt_id": invocation.attempt_id if invocation else None,
                            "operation": invocation.operation if invocation else None,
                            "evidence_digest": digest(evidence.evidence.encode())
                            if evidence
                            else None,
                            "session_revoked": evidence.revoked if evidence else None,
                            "session_revocation_digest": digest(
                                evidence.session_revocation_evidence.encode()
                            )
                            if evidence and evidence.session_revocation_evidence
                            else None,
                            "grant_revocation_digest": digest(
                                evidence.grant_revocation_evidence.encode()
                            )
                            if evidence and evidence.grant_revocation_evidence
                            else None,
                        },
                        separators=(",", ":"),
                    ),
                ),
            )
            self.db.execute("COMMIT")
        except BaseException:
            self.db.execute("ROLLBACK")
            raise


class Executor:
    """Injected fixed owner adapters and provenance verifier; none ship by default.

    evidence verifier must authenticate owner custody/provenance or explicitly
    attest attended manual evidence; a bare JSON reference is never sufficient.
    It receives purpose, exact source digest, and named evidence reference.
    """

    def __init__(
        self,
        journal: Journal,
        adapters: dict[str, Adapter],
        evidence_verifier: Callable[[str, str, str], bool],
    ):
        self.journal, self.adapters, self.evidence_verifier = (
            journal,
            dict(adapters),
            evidence_verifier,
        )
        self._admitted = {}

    def admit(
        self,
        case_source: bytes,
        type_source: bytes,
        descriptor_source: bytes,
        catalog_source: bytes,
        *,
        principal: str,
        actor: str,
        lane: str,
        acceptance_evidence: str,
        materialization_evidence: str,
        descriptor_evidence: str,
        session_evidence: str,
    ) -> tuple[Descriptor, Invocation, StepState]:
        for source in (case_source, type_source, descriptor_source, catalog_source):
            bounded_source(source)
        descriptor = Descriptor.decode(descriptor_source)
        try:
            case = yaml.safe_load(case_source)["spec"]
            case_type = yaml.safe_load(type_source)["spec"]
            if not isinstance(case, dict) or not isinstance(case_type, dict):
                raise ValueError
        except (yaml.YAMLError, KeyError, TypeError, ValueError):
            raise Refused("invalid case or type source shape") from None
        if case.get("status") != "materialized" or not case.get("acceptance"):
            raise Refused("case is not accepted and structurally materialized")
        if case_type.get("status") not in ("acknowledged", "implemented"):
            raise Refused("type is not acknowledged")
        if case.get("type") != f"{case_type['id']}@{case_type['version']}":
            raise Refused("case type pin mismatch")
        if case_type.get("execution_descriptors", {}).get(descriptor.step) != digest(
            descriptor_source
        ):
            raise Refused(
                "type has no exact acknowledged machine descriptor; plan only"
            )
        for purpose, source, evidence in (
            ("case-acceptance", digest(case_source), acceptance_evidence),
            (
                "ops-mason-materialization",
                digest(case_source),
                materialization_evidence,
            ),
            (
                "owner-descriptor-acceptance:" + descriptor.owner,
                digest(descriptor_source),
                descriptor_evidence,
            ),
            ("type-machine-seam-acceptance", digest(type_source), descriptor_evidence),
            ("catalog-routing-acceptance", digest(catalog_source), descriptor_evidence),
            (
                f"caller-session:{actor}:{principal}:{lane}",
                digest(case_source),
                session_evidence,
            ),
        ):
            if (
                not isinstance(evidence, str)
                or not IDENTIFIER.fullmatch(evidence)
                or self._verify_evidence(purpose, source, evidence) is not True
            ):
                raise Refused("independent owner evidence not verified")
        if (
            not IDENTIFIER.fullmatch(principal)
            or actor != descriptor.actor
            or lane != descriptor.lane
        ):
            raise Refused("principal actor or lane mismatch")
        if descriptor.effect == "mint" and (
            actor != "human" or os.getenv("WARDEN_AGENT_ID")
        ):
            raise Refused("agent mint is forbidden")
        parameters = case.get("parameters", {})
        declared = {
            item["name"]: item
            for item in case_type.get("parameters", [])
            if isinstance(item, dict) and isinstance(item.get("name"), str)
        }
        if (
            not isinstance(parameters, dict)
            or descriptor.target_parameter not in declared
        ):
            raise Refused("target parameter is not declared by acknowledged type")
        for name in parameters:
            definition = declared.get(name)
            if (
                definition is None
                or definition.get("sensitivity") not in ("public", "operational")
                or definition.get("case_settable") is False
            ):
                raise Refused("unknown, fixed or secret-sensitivity case parameter")
        target = parameters.get(descriptor.target_parameter)
        if not isinstance(target, str) or not IDENTIFIER.fullmatch(target):
            raise Refused("target must be exact case metadata")
        adapter = self.adapters.get(descriptor.adapter)
        if adapter is None or (adapter.reference, adapter.source) != (
            descriptor.adapter,
            descriptor.adapter_source,
        ):
            raise Refused("unknown adapter or owner source pin")
        sheet = json.dumps(
            [
                descriptor.step,
                descriptor.owner,
                target,
                actor,
                principal,
                lane,
                acceptance_evidence,
                materialization_evidence,
                descriptor_evidence,
                session_evidence,
            ],
            separators=(",", ":"),
        ).encode()
        pins = Pins(
            digest(case_source),
            digest(type_source),
            digest(descriptor_source),
            digest(catalog_source),
            digest(acceptance_evidence.encode()),
            digest(sheet),
        )
        invocation = Invocation(
            descriptor.step, descriptor.owner, target, actor, principal, lane, pins
        )
        initial = StepState(
            descriptor.step,
            descriptor.owner,
            pins,
            descriptor.allowed_created,
            descriptor.retry_safe,
        )
        self._admitted[invocation] = (descriptor, initial, session_evidence)
        return descriptor, invocation, initial

    def _result(
        self, adapter: Adapter, invocation: Invocation, result: OwnerResult
    ) -> OwnerResult:
        if type(result) is not OwnerResult or not isinstance(result.outcome, Outcome):
            raise Refused("unknown owner result shape")
        if type(result.created) is not tuple or not all(
            isinstance(v, str) and IDENTIFIER.fullmatch(v) for v in result.created
        ):
            raise Refused("owner created-object result invalid")
        if not isinstance(result.evidence, str) or not IDENTIFIER.fullmatch(
            result.evidence
        ):
            raise Refused("owner evidence identifier invalid")
        if adapter.verify(invocation, result) is not True or result.revoked is not True:
            raise Refused("owner result or session revocation unverified")
        if (result.attempt_id, result.operation) != (
            invocation.attempt_id,
            invocation.operation,
        ):
            raise Refused("owner receipt belongs to another attempt")
        descriptor = self._admitted[
            replace(invocation, attempt_id="", operation="admission")
        ][0]
        proof_fields = {
            "parent-session": (
                f"session-revocation:{invocation.lane}:{invocation.operation}:{invocation.attempt_id}",
                result.session_revocation_evidence,
            ),
            "broker-grant": (
                f"grant-revocation:{invocation.lane}:{invocation.operation}:{invocation.attempt_id}",
                result.grant_revocation_evidence,
            ),
        }
        proofs = [proof_fields[scope] for scope in descriptor.revocation_scopes]
        for purpose, evidence in proofs:
            if (
                not isinstance(evidence, str)
                or not IDENTIFIER.fullmatch(evidence)
                or self._verify_evidence(purpose, invocation.pins.case, evidence)
                is not True
            ):
                raise Refused("independent grant or parent-session revocation absent")
        return result

    def _verify_evidence(self, purpose, source, evidence):
        try:
            with contained_output():
                return self.evidence_verifier(purpose, source, evidence) is True
        except Exception:
            raise Refused("independent evidence verification failed") from None

    def _admission(
        self, descriptor: Descriptor, invocation: Invocation, initial: StepState
    ):
        if self._admitted.get(invocation, ())[:2] != (descriptor, initial):
            raise Refused("execution requires verified admission on this executor")
        adapter = self.adapters.get(descriptor.adapter)
        if adapter is None or (adapter.reference, adapter.source) != (
            descriptor.adapter,
            descriptor.adapter_source,
        ):
            raise Refused("adapter changed after admission")
        session = self._admitted[invocation][2]
        if (
            self._verify_evidence(
                f"caller-session:{invocation.actor}:{invocation.principal}:{invocation.lane}",
                invocation.pins.case,
                session,
            )
            is not True
        ):
            raise Refused("caller session no longer verified")
        if descriptor.effect == "mint" and os.getenv("WARDEN_AGENT_ID"):
            raise Refused("agent mint is forbidden")
        return adapter

    def _attempt(self, invocation: Invocation, state: StepState, operation: str):
        identity = json.dumps(
            [asdict(invocation.pins), operation, state.last.attempt, uuid.uuid4().hex],
            separators=(",", ":"),
        ).encode()
        return replace(
            invocation,
            attempt_id="attempt-" + hashlib.sha256(identity).hexdigest(),
            operation=operation,
        )

    def run(
        self,
        descriptor: Descriptor,
        invocation: Invocation,
        initial: StepState,
        *,
        retry=False,
    ) -> StepState:
        adapter = self._admission(descriptor, invocation, initial)
        with self.journal.locked():
            binding = f"{invocation.step}:{invocation.target}"
            state = self.journal.load(binding, initial)
            recovered = recover(state)
            if recovered != state:
                self.journal.save(binding, recovered)
                state = recovered
            if len(state.receipts) > MAX_HISTORY - 4:
                raise Refused("bounded journal history requires owner archival")
            running = begin(state, invocation.pins, retry=retry)
            dispatched = self._attempt(invocation, running, "run")
            self.journal.save(binding, running, invocation=dispatched)
            evidence = None
            try:
                with contained_output():
                    result = self._result(
                        adapter, dispatched, adapter.invoke(dispatched)
                    )
                completed = finish(
                    running, invocation.pins, result.outcome, created=result.created
                )
                evidence = result
            except Exception:
                completed = finish(running, invocation.pins, Outcome.UNCERTAIN)
            self.journal.save(binding, completed, evidence, dispatched)
            return completed

    def rollback(
        self, descriptor: Descriptor, invocation: Invocation, initial: StepState
    ) -> StepState:
        adapter = self._admission(descriptor, invocation, initial)
        with self.journal.locked():
            binding = f"{invocation.step}:{invocation.target}"
            state = self.journal.load(binding, initial)
            if recover(state) != state:
                self.journal.save(binding, recover(state))
                raise Refused("lost rollback result requires owner reconciliation")
            if adapter.rollback is None:
                raise Refused("owner has no reviewed undo adapter")
            if len(state.receipts) > MAX_HISTORY - 4:
                raise Refused("bounded journal history requires owner archival")
            running = begin_rollback(state, invocation.pins)
            dispatched = self._attempt(invocation, running, "rollback")
            self.journal.save(binding, running, invocation=dispatched)
            evidence = None
            try:
                with contained_output():
                    result = self._result(
                        adapter,
                        dispatched,
                        adapter.rollback(dispatched, running.last.created),
                    )
                verified = result.outcome is Outcome.ROLLED_BACK and not result.created
                evidence = result if verified else None
            except Exception:
                verified = False
            completed = finish_rollback(running, invocation.pins, verified=verified)
            self.journal.save(binding, completed, evidence, dispatched)
            return completed

    def reconcile(
        self, descriptor: Descriptor, invocation: Invocation, initial: StepState
    ) -> StepState:
        """Explicit owner-native state inspection; never redispatches the mutation."""
        adapter = self._admission(descriptor, invocation, initial)
        if adapter.reconcile is None:
            raise Refused("owner reconciliation adapter absent")
        with self.journal.locked():
            binding = f"{invocation.step}:{invocation.target}"
            state = recover(self.journal.load(binding, initial))
            if not state.last or state.last.outcome not in (
                Outcome.UNCERTAIN,
                Outcome.FAILED_AFTER,
            ):
                raise Refused("no uncertain mutation to reconcile")
            if len(state.receipts) > MAX_HISTORY - 4:
                raise Refused("bounded journal history requires owner archival")
            state = replace(
                state,
                receipts=(
                    *state.receipts,
                    Receipt(len(state.receipts) + 1, Outcome.UNCERTAIN),
                ),
            )
            dispatched = self._attempt(invocation, state, "reconcile")
            self.journal.save(binding, state, invocation=dispatched)
            with contained_output():
                result = self._result(
                    adapter, dispatched, adapter.reconcile(dispatched)
                )
            if result.outcome not in (
                Outcome.NO_CHANGE,
                Outcome.SUCCESS,
                Outcome.FAILED_BEFORE,
            ):
                raise Refused("owner did not reconcile target state")
            # Validate created-object scope using the same closed result reducer.
            finish(
                begin(
                    StepState(
                        state.step_id,
                        state.owner,
                        state.pins,
                        state.allowed_created,
                        state.retry_safe,
                    ),
                    state.pins,
                ),
                state.pins,
                result.outcome,
                created=result.created,
            )
            reconciled = StepState(
                state.step_id,
                state.owner,
                state.pins,
                state.allowed_created,
                state.retry_safe,
                (
                    *state.receipts,
                    Receipt(len(state.receipts) + 1, result.outcome, result.created),
                ),
            )
            self.journal.save(binding, reconciled, result, dispatched)
            return reconciled


class OwnerInterrupted(KeyboardInterrupt):
    """Sanitized interruption preserves persisted running/crash recovery semantics."""

    def __init__(self):
        super().__init__("owner execution interrupted")


class _Discard:
    """Never retain or forward adapter output, including unbounded malicious text."""

    def write(self, text):
        return len(text)

    def flush(self):
        pass