violin/plugins/violin_guard/skill_receipts.py

"""Locked, receipt-backed skill delivery state.

The module is deliberately latent in v3-02: callers may prepare and bind
deliveries, but existing execution guards do not consume the receipts until
the command-gate migration branch.
"""

from __future__ import annotations

import hashlib
import json
import uuid
from collections.abc import Callable
from dataclasses import dataclass
from datetime import UTC, datetime
from pathlib import Path
from typing import Any

from . import state
from .skill_policy import validate_skill_selection

__all__ = [
    "DeliveryReservation",
    "HermesSkillViewAdapter",
    "SkillViewResult",
    "advance_context_generation",
    "bind_task",
    "complete_delivery",
    "get_binding",
    "binding_readiness",
    "get_delivery",
    "record_binding_turn",
    "record_delivery_turn",
    "prepare_delivery",
]

_FILE_NAME = "skills.json"
_SCHEMA_VERSION = 1
_MAX_DELIVERIES = 200
_PREPARING_TTL_SECONDS = 300


def _now() -> str:
    return datetime.now(UTC).isoformat().replace("+00:00", "Z")


def _digest(value: str) -> str:
    return "sha256:" + hashlib.sha256(value.encode("utf-8")).hexdigest()


def _preparing_expired(entry: dict[str, Any]) -> bool:
    value = str(entry.get("expires_at") or "").strip()
    if value:
        try:
            return datetime.fromisoformat(value.replace("Z", "+00:00")) <= datetime.now(UTC)
        except ValueError:
            return True
    updated = str(entry.get("updated_at") or entry.get("created_at") or "").strip()
    if not updated:
        return True
    try:
        started = datetime.fromisoformat(updated.replace("Z", "+00:00"))
    except ValueError:
        return True
    return (datetime.now(UTC) - started).total_seconds() >= _PREPARING_TTL_SECONDS


def _preparing_expires_at() -> str:
    from datetime import timedelta

    return (
        (datetime.now(UTC) + timedelta(seconds=_PREPARING_TTL_SECONDS))
        .isoformat()
        .replace("+00:00", "Z")
    )


def _path(eng_dir: str | Path) -> Path:
    return state.resolve_eng_dir(eng_dir) / "state" / _FILE_NAME


def _empty(session_id: str = "", generation: int = 0) -> dict[str, Any]:
    return {
        "schema_version": _SCHEMA_VERSION,
        "context": {"session_id": session_id, "generation": generation},
        "deliveries": {},
        "bindings": {},
    }


def _load(path: Path) -> tuple[dict[str, Any], bool]:
    """Load valid receipt state; recover malformed/old documents fail-closed."""

    if not path.exists():
        return _empty(), False
    try:
        raw = json.loads(path.read_text(encoding="utf-8"))
    except (OSError, json.JSONDecodeError):
        return _empty(), True
    if not isinstance(raw, dict) or raw.get("schema_version") != _SCHEMA_VERSION:
        return _empty(), True
    if not isinstance(raw.get("context"), dict) or not isinstance(raw.get("deliveries"), dict):
        return _empty(), True
    raw.setdefault("bindings", {})
    return raw, False


def _mutate(eng_dir: str | Path, mutation: Callable[[dict[str, Any]], Any]) -> Any:
    path = _path(eng_dir)
    state.ensure_dir(path.parent)
    with state.lock_file(path):
        data, recovered = _load(path)
        if recovered:
            data["recovered_at"] = _now()
        result = mutation(data)
        state.atomic_json(path, data)
        return result


def _context(data: dict[str, Any], session_id: str) -> tuple[str, int]:
    context = data.setdefault("context", {"session_id": "", "generation": 0})
    recorded = str(context.get("session_id") or "")
    if recorded and recorded != session_id:
        # A new Hermes session is a new context, never proof of the old load.
        context["session_id"] = session_id
        context["generation"] = int(context.get("generation") or 0) + 1
    elif not recorded:
        context["session_id"] = session_id
    return str(context["session_id"]), int(context.get("generation") or 0)


def _delivery_key(session_id: str, generation: int, skill: str, bundle_digest: str) -> str:
    return _digest(f"{session_id}\0{generation}\0{skill}\0{bundle_digest}")


def _prune(data: dict[str, Any]) -> None:
    deliveries = data["deliveries"]
    if len(deliveries) <= _MAX_DELIVERIES:
        return
    bound = {str(binding.get("delivery_id") or "") for binding in data["bindings"].values()}
    removable = sorted(
        (entry for entry in deliveries.values() if entry.get("id") not in bound),
        key=lambda entry: str(entry.get("updated_at") or ""),
    )
    for entry in removable[: max(0, len(deliveries) - _MAX_DELIVERIES)]:
        deliveries.pop(entry["id"], None)


@dataclass(frozen=True)
class DeliveryReservation:
    id: str
    status: str
    skill: str
    bundle_digest: str
    session_id: str
    context_generation: int
    owner: bool
    owner_token: str = ""


@dataclass(frozen=True)
class SkillViewResult:
    ready: bool
    content: str = ""
    error: str = ""
    path: str = ""


class HermesSkillViewAdapter:
    """Small adapter around Hermes's JSON-returning ``skill_view`` helper."""

    def __init__(self, view: Callable[..., str] | None = None):
        self._view = view

    def view(self, skill: str, task_id: str | None = None) -> SkillViewResult:
        try:
            view = self._view
            if view is None:
                from tools.skills_tool import skill_view as view  # type: ignore[import-not-found]
            raw = view(skill, task_id=task_id)
            payload = json.loads(raw) if isinstance(raw, str) else raw
            if not isinstance(payload, dict) or not payload.get("success"):
                return SkillViewResult(
                    False, error=str((payload or {}).get("error") or "skill_view failed")
                )
            content = str(payload.get("content") or "")
            if not content:
                return SkillViewResult(False, error="skill_view returned no skill content")
            return SkillViewResult(True, content=content, path=str(payload.get("path") or ""))
        except Exception as exc:  # Hermes availability is an external dependency.
            return SkillViewResult(False, error=f"skill_view unavailable: {exc}")


def prepare_delivery(
    eng_dir: str | Path,
    *,
    session_id: str,
    skill: str,
    bundle_digest: str,
    phase: str,
    vulnerability_class: str | None = None,
    candidate_source: str | None = None,
) -> DeliveryReservation:
    """Reserve exactly one delivery for a semantic receipt key.

    Only the caller receiving ``owner=True`` may invoke Hermes and then call
    :func:`complete_delivery`; concurrent callers see the same ``preparing``
    receipt and never receive duplicate skill content.
    """

    if not session_id.strip() or not bundle_digest.startswith("sha256:"):
        raise ValueError("session_id and a sha256 bundle_digest are required")
    policy = validate_skill_selection(skill, phase, vulnerability_class, candidate_source)
    if policy.mismatch_reasons:
        raise ValueError("; ".join(policy.mismatch_reasons))

    def reserve(data: dict[str, Any]) -> DeliveryReservation:
        current_session, generation = _context(data, session_id.strip())
        identifier = _delivery_key(current_session, generation, skill, bundle_digest)
        existing = data["deliveries"].get(identifier)
        if existing and existing.get("status") == "delivered":
            return DeliveryReservation(
                identifier,
                existing["status"],
                skill,
                bundle_digest,
                current_session,
                generation,
                False,
            )
        if existing and existing.get("status") == "preparing" and not _preparing_expired(existing):
            return DeliveryReservation(
                identifier,
                existing["status"],
                skill,
                bundle_digest,
                current_session,
                generation,
                False,
            )
        owner_token = uuid.uuid4().hex
        now = _now()
        data["deliveries"][identifier] = {
            "id": identifier,
            "skill": skill,
            "bundle_digest": bundle_digest,
            "session_id": current_session,
            "context_generation": generation,
            "status": "preparing",
            "created_at": now,
            "updated_at": now,
            "expires_at": _preparing_expires_at(),
            "owner_token": owner_token,
            "attempts": int((existing or {}).get("attempts") or 0) + 1,
        }
        _prune(data)
        return DeliveryReservation(
            identifier,
            "preparing",
            skill,
            bundle_digest,
            current_session,
            generation,
            True,
            owner_token,
        )

    return _mutate(eng_dir, reserve)


def complete_delivery(
    eng_dir: str | Path,
    reservation: DeliveryReservation,
    result: SkillViewResult,
    *,
    delivered_turn_id: str | None = None,
) -> DeliveryReservation:
    """Complete a reservation after an actual Hermes ``skill_view`` call."""

    def complete(data: dict[str, Any]) -> DeliveryReservation:
        entry = data["deliveries"].get(reservation.id)
        if not entry or entry.get("status") != "preparing":
            raise ValueError("delivery reservation is no longer active")
        if not reservation.owner or not reservation.owner_token:
            raise ValueError("only the reservation owner may complete delivery")
        if entry.get("owner_token") != reservation.owner_token:
            raise ValueError("delivery reservation owner is stale; prepare a new delivery")
        entry["status"] = "delivered" if result.ready else "failed"
        entry["updated_at"] = _now()
        entry["delivered_turn_id"] = delivered_turn_id if result.ready else None
        entry["content_digest"] = _digest(result.content) if result.ready else None
        entry["error"] = result.error if not result.ready else None
        data["deliveries"][reservation.id] = entry
        return DeliveryReservation(
            reservation.id,
            entry["status"],
            reservation.skill,
            reservation.bundle_digest,
            reservation.session_id,
            reservation.context_generation,
            False,
        )

    return _mutate(eng_dir, complete)


def get_delivery(eng_dir: str | Path, delivery_id: str) -> dict[str, Any] | None:
    data, _ = _load(_path(eng_dir))
    return data["deliveries"].get(delivery_id)


def advance_context_generation(eng_dir: str | Path, session_id: str) -> int:
    """Invalidate active bindings after Hermes context reset/compression."""

    def advance(data: dict[str, Any]) -> int:
        _context(data, session_id.strip())
        data["context"]["generation"] = int(data["context"].get("generation") or 0) + 1
        return data["context"]["generation"]

    return _mutate(eng_dir, advance)


def bind_task(
    eng_dir: str | Path,
    *,
    task_id: str,
    delivery_id: str,
    hypothesis_id: str | None = None,
    technique: str = "",
) -> dict[str, Any]:
    """Bind a delivered receipt to one task/hypothesis for later enforcement."""

    def bind(data: dict[str, Any]) -> dict[str, Any]:
        delivery = data["deliveries"].get(delivery_id)
        if not delivery or delivery.get("status") != "delivered":
            raise ValueError("a delivered skill receipt is required before binding")
        binding = {
            "task_id": task_id,
            "delivery_id": delivery_id,
            "skill": delivery["skill"],
            "bundle_digest": delivery["bundle_digest"],
            "session_id": delivery["session_id"],
            "context_generation": delivery["context_generation"],
            "hypothesis_id": hypothesis_id or "",
            "technique": technique.strip(),
            "bound_at": _now(),
        }
        data["bindings"][task_id] = binding
        return binding

    return _mutate(eng_dir, bind)


def get_binding(eng_dir: str | Path, task_id: str) -> dict[str, Any] | None:
    data, _ = _load(_path(eng_dir))
    return data["bindings"].get(task_id)


def binding_readiness(
    eng_dir: str | Path, *, task_id: str, session_id: str
) -> tuple[dict[str, Any] | None, str | None]:
    """Return the active receipt binding, or a precise fail-closed reason."""

    data, recovered = _load(_path(eng_dir))
    if recovered:
        return None, "skill receipt state is unavailable; prepare the selected skill again"
    context = data.get("context") or {}
    binding = (data.get("bindings") or {}).get(task_id)
    if not binding:
        return None, "the active PTT task has no delivered skill binding"
    if str(context.get("session_id") or "") != session_id:
        return None, "the skill binding belongs to a different session"
    if int(binding.get("context_generation", -1)) != int(context.get("generation") or 0):
        return None, "the skill binding is stale after a context reset"
    delivery = (data.get("deliveries") or {}).get(binding.get("delivery_id"))
    if not delivery or delivery.get("status") != "delivered":
        return None, "the bound skill delivery is not ready"
    if delivery.get("bundle_digest") != binding.get("bundle_digest"):
        return None, "the bound skill digest no longer matches its delivery"
    return {
        **binding,
        "delivered_turn_id": delivery.get("delivered_turn_id"),
        "delivered_api_request_id": delivery.get("delivered_api_request_id"),
    }, None


def record_delivery_turn(
    eng_dir: str | Path,
    *,
    delivery_id: str,
    turn_id: str,
    api_request_id: str = "",
) -> None:
    """Associate a successful ``skill_view`` result with its model call."""

    if not turn_id and not api_request_id:
        return

    def record(data: dict[str, Any]) -> None:
        entry = data["deliveries"].get(delivery_id)
        if entry and entry.get("status") == "delivered":
            entry["delivered_turn_id"] = turn_id
            entry["delivered_api_request_id"] = api_request_id
            entry["updated_at"] = _now()

    _mutate(eng_dir, record)


def record_binding_turn(
    eng_dir: str | Path,
    *,
    task_id: str,
    turn_id: str,
    api_request_id: str = "",
) -> None:
    """Associate a binding commit with its model call."""

    if not turn_id and not api_request_id:
        return

    def record(data: dict[str, Any]) -> None:
        binding = data["bindings"].get(task_id)
        if binding:
            binding["bound_turn_id"] = turn_id
            binding["bound_api_request_id"] = api_request_id
            binding["bound_at"] = _now()

    _mutate(eng_dir, record)