railiance-platform/scripts/task_execution_journal.py

"""Source-only admission primitive; never authorizes or dispatches native work.

Trusted owner mediation must validate claim/PDP/consume before transitions.
One approval admits one operation. A persisted reservation cannot be resumed.
The process lock is held across the whole owner invocation, including cleanup.
"""
from __future__ import annotations

import fcntl
import os
import re
import sqlite3
import stat
import uuid
from contextlib import contextmanager
from dataclasses import dataclass
from pathlib import Path


class Refused(Exception):
    pass


@dataclass(frozen=True)
class Binding:
    approval_id: str
    request_digest: str
    operation_id: str
    plan_digest: str

    def validate(self):
        if any(type(v) is not str for v in (self.approval_id, self.request_digest, self.operation_id, self.plan_digest)):
            raise Refused("binding fields must be strings")
        if not re.fullmatch(r"[A-Za-z0-9][A-Za-z0-9_-]{0,127}", self.approval_id):
            raise Refused("invalid approval identifier")
        # Preserve the exact served PDP/Approval value; never silently strip it.
        if not re.fullmatch(r"sha256:[0-9a-f]{64}", self.request_digest):
            raise Refused("invalid canonical request digest")
        if not re.fullmatch(r"[0-9a-f]{64}", self.plan_digest):
            raise Refused("invalid owner plan content digest")
        try:
            if str(uuid.UUID(self.operation_id)) != self.operation_id:
                raise ValueError()
        except (ValueError, AttributeError):
            raise Refused("invalid operation identifier") from None


_NEXT = {
    "reserved": {"consume-confirmed", "failed", "unknown"},
    "consume-confirmed": {"authenticating", "failed", "unknown"},
    "authenticating": {"grant-issuing", "failed", "unknown"},
    "grant-issuing": {"grant-issued", "failed", "unknown"},
    "grant-issued": {"applying", "failed", "unknown"},
    "applying": {"readback", "failed", "unknown"},
    "readback": {"verified", "failed", "unknown"},
    "verified": {"cleanup-confirmed", "unknown"},
}


class Journal:
    def __init__(self, directory: Path):
        # Require an existing owner-private directory: deployment creates it.
        if directory.is_symlink() or directory.resolve() != directory.absolute():
            raise Refused("journal path is indirect")
        self.directory = directory
        self._private(directory.stat(), directory=True)
        self._lock_fd = self._open("execution.lock")
        try:
            fd = self._open("journal.sqlite")
        except BaseException:
            os.close(self._lock_fd)
            del self._lock_fd
            raise
        original = os.fstat(fd)
        os.close(fd)
        self.db = sqlite3.connect(directory / "journal.sqlite", isolation_level=None)
        current = (directory / "journal.sqlite").stat()
        if (original.st_dev, original.st_ino) != (current.st_dev, current.st_ino):
            self.close()
            raise Refused("journal changed during open")
        self.db.execute("PRAGMA synchronous=FULL")
        self.db.execute("PRAGMA journal_mode=DELETE")
        self.db.execute("CREATE TABLE IF NOT EXISTS operations (approval TEXT PRIMARY KEY, request TEXT NOT NULL, operation TEXT UNIQUE NOT NULL, plan TEXT NOT NULL, state TEXT NOT NULL, evidence TEXT)")
        self.db.execute("CREATE TABLE IF NOT EXISTS calls (operation TEXT NOT NULL, ordinal INTEGER NOT NULL, call_digest TEXT NOT NULL, outcome TEXT NOT NULL, evidence TEXT, PRIMARY KEY(operation,ordinal))")
        self._held = False
        self._admission = None
        self._sync_directory()

    @staticmethod
    def _private(s, directory=False):
        kind = stat.S_ISDIR(s.st_mode) if directory else stat.S_ISREG(s.st_mode)
        if not kind or s.st_uid != os.getuid() or s.st_mode & 0o077:
            raise Refused("journal is not owner private")
        if not directory and s.st_nlink != 1:
            raise Refused("journal file has multiple links")

    def _open(self, name):
        fd = os.open(self.directory / name, os.O_CREAT | os.O_RDWR | os.O_NOFOLLOW, 0o600)
        try:
            self._private(os.fstat(fd))
        except BaseException:
            os.close(fd)
            raise
        return fd

    def _sync_directory(self):
        fd = os.open(self.directory, os.O_RDONLY | os.O_DIRECTORY | os.O_NOFOLLOW)
        try:
            os.fsync(fd)
        finally:
            os.close(fd)

    @contextmanager
    def locked(self):
        if self._held:
            raise Refused("nested journal admission")
        try:
            fcntl.flock(self._lock_fd, fcntl.LOCK_EX | fcntl.LOCK_NB)
        except BlockingIOError:
            raise Refused("owner invocation already active") from None
        self._held = True
        try:
            yield self
        finally:
            self._admission = None
            self._held = False
            fcntl.flock(self._lock_fd, fcntl.LOCK_UN)

    def reserve(self, binding: Binding):
        binding.validate()
        if not self._held or self._admission is not None:
            raise Refused("reservation requires fresh exclusive invocation")
        try:
            self.db.execute("BEGIN IMMEDIATE")
            self.db.execute("INSERT INTO operations VALUES (?,?,?,?,?,NULL)",
                            (binding.approval_id, binding.request_digest,
                             binding.operation_id, binding.plan_digest, "reserved"))
            self.db.execute("COMMIT")
            self._sync_directory()
        except sqlite3.IntegrityError:
            self.db.execute("ROLLBACK")
            raise Refused("approval or operation already admitted") from None
        except BaseException:
            if self.db.in_transaction:
                self.db.execute("ROLLBACK")
            raise
        self._admission = binding

    def transition(self, binding: Binding, expected: str, target: str, evidence_digest=None):
        if not self._held or self._admission != binding:
            raise Refused("no live admission")
        if target not in _NEXT.get(expected, set()):
            raise Refused("invalid state transition")
        if target == "cleanup-confirmed" and evidence_digest is None:
            raise Refused("cleanup requires evidence")
        if evidence_digest is not None and not re.fullmatch(r"[0-9a-f]{64}", evidence_digest):
            raise Refused("invalid evidence digest")
        try:
            self.db.execute("BEGIN IMMEDIATE")
            pending=self.db.execute("SELECT COUNT(*) FROM calls WHERE operation=? AND outcome='pending'",(binding.operation_id,)).fetchone()[0]
            if pending and target != 'unknown':
                raise Refused("unconfirmed external call requires unknown")
            changed = self.db.execute(
                "UPDATE operations SET state=?, evidence=? WHERE approval=? AND request=? AND operation=? AND plan=? AND state=?",
                (target, evidence_digest, binding.approval_id, binding.request_digest,
                 binding.operation_id, binding.plan_digest, expected)).rowcount
            if changed != 1:
                raise Refused("journal state changed")
            self.db.execute("COMMIT")
            self._sync_directory()
        except BaseException:
            if self.db.in_transaction:
                self.db.execute("ROLLBACK")
            raise

    def observe(self, binding: Binding):
        binding.validate()
        row = self.db.execute("SELECT request,operation,plan,state,evidence FROM operations WHERE approval=?", (binding.approval_id,)).fetchone()
        if row is None:
            return None
        if row[:3] != (binding.request_digest, binding.operation_id, binding.plan_digest):
            raise Refused("journal binding mismatch")
        # Readback cannot create a live admission, even for an interrupted state.
        pending = self.db.execute("SELECT COUNT(*) FROM calls WHERE operation=? AND outcome='pending'", (binding.operation_id,)).fetchone()[0]
        return {"state": row[3], "evidence_digest": row[4], "dispatch_allowed": False,
                "external_outcome_unknown": bool(pending)}

    def before_call(self, binding: Binding, ordinal: int, call_digest: str):
        """Persist one exact plan-call intent before the mediator issues it.

        Caller must derive ordinal/digest from the independently pinned owner plan.
        This primitive doesn't validate method/body or decide whether a call is allowed.
        """
        if not self._held or self._admission != binding:
            raise Refused("no live admission")
        if type(ordinal) is not int or ordinal < 0 or ordinal > 63:
            raise Refused("invalid fixed-plan ordinal")
        if type(call_digest) is not str or not re.fullmatch(r"[0-9a-f]{64}", call_digest):
            raise Refused("invalid call digest")
        try:
            self.db.execute("BEGIN IMMEDIATE")
            count,pending = self.db.execute("SELECT COUNT(*), COALESCE(SUM(outcome='pending'),0) FROM calls WHERE operation=?",(binding.operation_id,)).fetchone()
            state=self.db.execute("SELECT state FROM operations WHERE operation=?",(binding.operation_id,)).fetchone()[0]
            if ordinal != count or pending or state in {'failed','unknown','cleanup-confirmed'}:
                raise Refused("call is replayed, uncertain or terminal")
            self.db.execute("INSERT INTO calls VALUES (?,?,?,'pending',NULL)",(binding.operation_id,ordinal,call_digest))
            self.db.execute("COMMIT")
            self._sync_directory()
        except BaseException:
            if self.db.in_transaction: self.db.execute("ROLLBACK")
            raise

    def after_call(self, binding: Binding, ordinal: int, call_digest: str, evidence_digest: str):
        if not self._held or self._admission != binding:
            raise Refused("no live admission")
        if type(ordinal) is not int or ordinal < 0 or ordinal > 63:
            raise Refused("invalid fixed-plan ordinal")
        if type(call_digest) is not str or not re.fullmatch(r"[0-9a-f]{64}", call_digest):
            raise Refused("invalid call digest")
        if type(evidence_digest) is not str or not re.fullmatch(r"[0-9a-f]{64}", evidence_digest):
            raise Refused("invalid evidence digest")
        try:
            self.db.execute("BEGIN IMMEDIATE")
            n=self.db.execute("UPDATE calls SET outcome='complete', evidence=? WHERE operation=? AND ordinal=? AND call_digest=? AND outcome='pending'",(evidence_digest,binding.operation_id,ordinal,call_digest)).rowcount
            if n != 1: raise Refused("call confirmation mismatch")
            self.db.execute("COMMIT")
            self._sync_directory()
        except BaseException:
            if self.db.in_transaction: self.db.execute("ROLLBACK")
            raise

    def close(self):
        if hasattr(self, "db"):
            self.db.close()
        if hasattr(self, "_lock_fd"):
            os.close(self._lock_fd)