scripts/labora_bridge.py

"""Pure contract primitives for the device-only Labora local bridge 1.2.

This module deliberately contains no authentication, HTTP, provider, filesystem,
or billing integration.  In particular, signed envelopes and transient action
content are never written by the persistence helpers below.
"""

import base64
import hashlib
import hmac
import json
import re
import secrets
import sqlite3
import time
from collections.abc import Mapping, Sequence
from pathlib import PurePosixPath
from typing import Any, Literal

from pydantic import BaseModel, ConfigDict, Field, field_validator, model_validator


CONTRACT_VERSION = "1.2"
ENVELOPE_ISSUER = "synderesis-labora-bridge"
ENVELOPE_AUDIENCE = "labora-native-device"
MAX_ENVELOPE_LIFETIME_SECONDS = 10 * 60
MAX_CLOCK_SKEW_SECONDS = 30
MAX_RUN_LIFETIME_SECONDS = 24 * 60 * 60
MAX_RUN_DECISIONS = 16
MAX_RUN_ACTIONS = 8
MAX_RUN_PROVIDER_CALLS = 16
MAX_USAGE_UNITS = 1_000_000_000
MAX_COST_MICROUNITS = 1_000_000_000_000
MAX_SEARCH_QUERY_BYTES = 8 * 1024
MAX_FINAL_ANSWER_BYTES = 100 * 1024
LOCAL_ACTION_TYPES = (
    "workspace.read_text",
    "workspace.write_text_atomic",
    "git.status",
    "git.diff",
)
CAPABILITY_DESCRIPTORS = (
    {"type": "workspace.read_text", "schema_version": 1},
    {"type": "workspace.write_text_atomic", "schema_version": 1},
    {"type": "git.status", "schema_version": 1},
    {"type": "git.diff", "schema_version": 1},
)
CAPABILITY_DESCRIPTOR_DIGEST = (
    "a556a46ec4dfe7b8ac6d12b7a652362a5bd17d16caf8c2c9828fabc799cd26e9"
)
EMPTY_ROLLING_CHAIN_DIGEST = "0" * 64
PUBLIC_PHASES = (
    "queued",
    "planning",
    "working",
    "reviewing",
    "waiting_for_approval",
    "completed",
    "completed_with_limits",
    "cancelled",
    "failed",
)
TERMINAL_PHASES = frozenset(PUBLIC_PHASES[-4:])
INTERNAL_STATES = frozenset(
    {
        "queued",
        "planning",
        "working",
        "reviewing",
        "waiting_for_approval",
        "provider_returned",
        "metering_pending",
        "metering_recorded",
        "metering_recorded_cancelled",
        "delivery_ready",
        "completed",
        "completed_with_limits",
        "cancelled",
        "failed",
        "provider_outcome_unknown",
    }
)
ACCOUNTING_FENCE_STATES = frozenset(
    {"metering_recorded", "metering_recorded_cancelled"}
)
ACTION_STATES = frozenset(
    {
        "offered",
        "claimed",
        "result_received",
        "consumed",
        "denied",
        "cancelled",
        "expired",
    }
)
EXECUTION_LEASE_STATES = frozenset(
    {
        "prepared",
        "dispatch_committed",
        "returned",
        "abandoned",
    }
)
DIGEST_DOMAINS = frozenset(
    {
        "request",
        "action",
        "approval",
        "result_chain",
        "receipt",
        "response",
        "idempotency",
    }
)
MAX_TEXT_BYTES = 128 * 1024
MAX_GIT_OUTPUT_BYTES = 128 * 1024
OPAQUE_ID_PREFIXES = frozenset({"lrun_", "lact_", "lexe_", "lrcpt_"})

_DIGEST_RE = re.compile(r"^[0-9a-f]{64}$")
_IDEMPOTENCY_RE = re.compile(r"^[A-Za-z0-9._~:+/=-]{16,128}$")
_KEY_ID_RE = re.compile(r"^[A-Za-z0-9._-]{1,64}$")
_IDENTITY_RE = re.compile(r"^[A-Za-z0-9._:@/-]{1,256}$")
_SAFE_STATE_RE = re.compile(r"^[a-z][a-z0-9_]{0,63}$")
_OPERATION_RE = re.compile(r"^[a-z][A-Za-z0-9_.:-]{0,255}$")
_OPAQUE_ID_RE = re.compile(r"^(?:lrun_|lact_|lexe_|lrcpt_)[A-Za-z0-9_-]{22,86}$")
_PRICING_VALUE_RE = re.compile(r"^[A-Za-z0-9._:/-]{1,128}$")
_CREDENTIAL_PART_RE = re.compile(
    r"(?:^|[._-])(env|credentials?|secrets?|tokens?|id_rsa|id_ed25519|netrc)(?:$|[._-])",
    re.IGNORECASE,
)
_ARCHIVE_SUFFIXES = (
    ".7z",
    ".bz2",
    ".gz",
    ".rar",
    ".tar",
    ".tar.bz2",
    ".tar.gz",
    ".tgz",
    ".xz",
    ".zip",
)


class AccountingFenceConflict(RuntimeError):
    """The run still has authority that prevents final accounting."""


class BridgeModel(BaseModel):
    """Strict base for every public bridge schema."""

    model_config = ConfigDict(extra="forbid", strict=True)


class WorkspaceManifest(BridgeModel):
    handle: str = Field(min_length=1, max_length=256)
    manifest_digest: str

    @field_validator("manifest_digest")
    @classmethod
    def digest_is_canonical(cls, value: str) -> str:
        return require_digest(value)


class SourcePolicy(BridgeModel):
    official: Literal["off", "required", "allowed"] = "off"
    web_search: bool = False


class CapabilityDescriptor(BridgeModel):
    type: Literal[
        "workspace.read_text",
        "workspace.write_text_atomic",
        "git.status",
        "git.diff",
    ]
    schema_version: Literal[1]


class StartRunRequest(BridgeModel):
    goal: str = Field(min_length=1, max_length=32_768)
    workspace: WorkspaceManifest
    source_policy: SourcePolicy
    capabilities: list[CapabilityDescriptor] = Field(min_length=4, max_length=4)
    capability_descriptor_digest: str

    @model_validator(mode="after")
    def descriptors_are_frozen(self) -> "StartRunRequest":
        require_capability_descriptors(
            [item.model_dump() for item in self.capabilities],
            self.capability_descriptor_digest,
        )
        return self


class ReadTextAction(BridgeModel):
    type: Literal["workspace.read_text"]
    path: str
    max_bytes: int = Field(default=MAX_TEXT_BYTES, ge=1, le=MAX_TEXT_BYTES)


class WriteTextAtomicAction(BridgeModel):
    type: Literal["workspace.write_text_atomic"]
    path: str
    content: str
    expected_base_sha256: str | None = None
    must_not_exist: bool = False
    output_sha256: str

    @model_validator(mode="after")
    def validate_write(self) -> "WriteTextAtomicAction":
        if (self.expected_base_sha256 is None) == (not self.must_not_exist):
            raise ValueError(
                "exactly one of expected_base_sha256 or must_not_exist is required"
            )
        if self.expected_base_sha256 is not None:
            require_digest(self.expected_base_sha256)
        require_digest(self.output_sha256)
        encoded = self.content.encode("utf-8")
        if len(encoded) > MAX_TEXT_BYTES:
            raise ValueError("replacement exceeds byte limit")
        if not hmac.compare_digest(sha256_bytes(encoded), self.output_sha256):
            raise ValueError("output_sha256 does not match replacement")
        return self


class GitStatusAction(BridgeModel):
    type: Literal["git.status"]


class GitDiffAction(BridgeModel):
    type: Literal["git.diff"]
    scope: Literal["working", "staged"] = "working"
    paths: list[str] = Field(default_factory=list, max_length=128)
    max_bytes: int = Field(default=MAX_GIT_OUTPUT_BYTES, ge=1, le=MAX_GIT_OUTPUT_BYTES)


class ClaimRequest(BridgeModel):
    action_offer: str = Field(min_length=32, max_length=8192)
    approval: Literal["approved", "denied"]


class ResultPayload(BridgeModel):
    status: Literal["ok", "error"]
    content: str = Field(max_length=MAX_TEXT_BYTES)

    @field_validator("content")
    @classmethod
    def content_is_bounded_utf8(cls, value: str) -> str:
        if len(value.encode("utf-8")) > MAX_TEXT_BYTES:
            raise ValueError("result exceeds byte limit")
        return value


class ResultRequest(BridgeModel):
    execution: str = Field(min_length=32, max_length=8192)
    action: dict[str, Any]
    result: ResultPayload
    result_digest: str

    @model_validator(mode="after")
    def result_matches_digest(self) -> "ResultRequest":
        require_digest(self.result_digest)
        if not hmac.compare_digest(
            sha256_text(self.result.content), self.result_digest
        ):
            raise ValueError("result_digest does not match content")
        return self


class ChainItem(BridgeModel):
    action: dict[str, Any]
    result: ResultPayload


class ContinueRequest(StartRunRequest):
    chain: list[ChainItem] = Field(min_length=1, max_length=MAX_RUN_ACTIONS)
    continuation: str = Field(min_length=32, max_length=8192)


class CancelRequest(BridgeModel):
    pass


class BackendSearchDecision(BridgeModel):
    kind: Literal["backend_search"]
    source: Literal["official", "web"]
    query: str = Field(min_length=1, max_length=MAX_SEARCH_QUERY_BYTES)

    @field_validator("query")
    @classmethod
    def query_is_bounded_utf8(cls, value: str) -> str:
        if len(value.encode("utf-8")) > MAX_SEARCH_QUERY_BYTES:
            raise ValueError("search query exceeds byte limit")
        return value


class LocalActionDecision(BridgeModel):
    kind: Literal["local_action"]
    action: dict[str, Any]

    @field_validator("action")
    @classmethod
    def action_is_closed(cls, value: dict[str, Any]) -> dict[str, Any]:
        validated = validate_local_action(value)
        return {
            key: item for key, item in validated.items() if key not in {"argv", "env"}
        }


class FinishDecision(BridgeModel):
    kind: Literal["finish"]
    answer: str = Field(max_length=MAX_FINAL_ANSWER_BYTES)
    citations: list[str] = Field(default_factory=list, max_length=128)

    @field_validator("answer")
    @classmethod
    def answer_is_bounded_utf8(cls, value: str) -> str:
        if len(value.encode("utf-8")) > MAX_FINAL_ANSWER_BYTES:
            raise ValueError("answer exceeds byte limit")
        return value

    @field_validator("citations")
    @classmethod
    def citations_are_bounded(cls, value: list[str]) -> list[str]:
        if value:
            raise ValueError("bridge finish citations are unavailable")
        return value


class NeedsScopeDecision(BridgeModel):
    kind: Literal["needs_scope"]


class UnavailableDecision(BridgeModel):
    kind: Literal["unavailable"]


def validate_coordinator_decision(value: Mapping[str, Any]) -> dict[str, Any]:
    """Accept exactly one bounded controller decision and no provider reasoning."""

    if not isinstance(value, Mapping):
        raise ValueError("coordinator decision must be an object")
    model: type[BridgeModel]
    kind = value.get("kind")
    if kind == "local_action":
        model = LocalActionDecision
    elif kind == "finish":
        model = FinishDecision
    elif kind == "needs_scope":
        model = NeedsScopeDecision
    elif kind == "unavailable":
        model = UnavailableDecision
    else:
        raise ValueError("unsupported coordinator decision")
    return model.model_validate(value).model_dump()


def canonical_json(value: Any) -> bytes:
    """Return the single UTF-8 JSON representation used by every digest/token."""

    try:
        return json.dumps(
            value,
            ensure_ascii=False,
            allow_nan=False,
            sort_keys=True,
            separators=(",", ":"),
        ).encode("utf-8")
    except (TypeError, ValueError) as exc:
        raise ValueError("value is not canonical JSON") from exc


def sha256_bytes(value: bytes) -> str:
    return hashlib.sha256(value).hexdigest()


def sha256_text(value: str) -> str:
    return sha256_bytes(value.encode("utf-8"))


def canonical_digest(value: Any) -> str:
    return sha256_bytes(canonical_json(value))


if canonical_digest(list(CAPABILITY_DESCRIPTORS)) != CAPABILITY_DESCRIPTOR_DIGEST:
    raise RuntimeError("frozen capability descriptor digest mismatch")


def require_capability_descriptors(
    descriptors: Sequence[Mapping[str, Any]], digest: str
) -> str:
    """Require the exact ordered bridge capability contract and its digest."""

    if isinstance(descriptors, (str, bytes)) or not isinstance(descriptors, Sequence):
        raise ValueError("capability descriptors must be an ordered list")
    try:
        clean = [dict(item) for item in descriptors]
    except (TypeError, ValueError) as exc:
        raise ValueError("invalid capability descriptors") from exc
    expected = [dict(item) for item in CAPABILITY_DESCRIPTORS]
    if clean != expected:
        raise ValueError("capability descriptors do not match the frozen contract")
    if not digests_equal(digest, CAPABILITY_DESCRIPTOR_DIGEST):
        raise ValueError("capability descriptor digest mismatch")
    if not digests_equal(canonical_digest(clean), CAPABILITY_DESCRIPTOR_DIGEST):
        raise ValueError("capability descriptors are not canonical")
    return CAPABILITY_DESCRIPTOR_DIGEST


def _fold_rolling_chain_item(
    previous_digest: str, sequence: int, item: Mapping[str, Any]
) -> str:
    require_digest(previous_digest)
    if type(sequence) is not int or not 1 <= sequence <= MAX_RUN_ACTIONS:
        raise ValueError("invalid continuation chain sequence")
    if not isinstance(item, Mapping) or set(item) != {"action", "result"}:
        raise ValueError("invalid continuation chain item")
    action = validate_local_action(item["action"])
    # Executor-only argv/env are derived, not part of the signed action.
    action = {key: value for key, value in action.items() if key not in {"argv", "env"}}
    result = ResultPayload.model_validate(item["result"]).model_dump()
    return _fold_rolling_chain_digests(
        previous_digest,
        sequence,
        canonical_digest(action),
        canonical_digest(result),
    )


def _fold_rolling_chain_digests(
    previous_digest: str,
    sequence: int,
    action_digest: str,
    result_digest: str,
) -> str:
    """Fold already-validated content digests without retaining private content."""

    require_digest(previous_digest)
    require_digest(action_digest)
    require_digest(result_digest)
    if type(sequence) is not int or not 1 <= sequence <= MAX_RUN_ACTIONS:
        raise ValueError("invalid continuation chain sequence")
    return canonical_digest(
        {
            "sequence": sequence,
            "previous_digest": previous_digest,
            "action_digest": action_digest,
            "result_digest": result_digest,
        }
    )


def digest_chain(items: Sequence[Mapping[str, Any]]) -> str:
    """Bind an ordered transient action/result chain without persisting it."""

    if len(items) > MAX_RUN_ACTIONS:
        raise ValueError("continuation chain exceeds run limit")
    previous = EMPTY_ROLLING_CHAIN_DIGEST
    for index, item in enumerate(items, start=1):
        previous = _fold_rolling_chain_item(previous, index, item)
    return previous


def verify_rolling_chain(
    persisted_digest: str,
    persisted_sequence: int,
    chain_items: Sequence[Mapping[str, Any]],
) -> str:
    """Verify the complete ordered action/result history against persisted state."""

    require_digest(persisted_digest)
    if (
        type(persisted_sequence) is not int
        or not 0 <= persisted_sequence <= MAX_RUN_ACTIONS
        or isinstance(chain_items, (str, bytes))
        or not isinstance(chain_items, Sequence)
        or len(chain_items) != persisted_sequence
    ):
        raise ValueError("continuation chain sequence mismatch")
    computed = digest_chain(chain_items)
    if not hmac.compare_digest(computed, persisted_digest):
        raise ValueError("continuation chain digest mismatch")
    return computed


def require_digest(value: str) -> str:
    if not isinstance(value, str) or _DIGEST_RE.fullmatch(value) is None:
        raise ValueError("expected a lowercase SHA-256 digest")
    return value


def digests_equal(left: str, right: str) -> bool:
    return hmac.compare_digest(require_digest(left), require_digest(right))


def new_opaque_id(prefix: str) -> str:
    if prefix not in OPAQUE_ID_PREFIXES:
        raise ValueError("unsupported opaque ID prefix")
    # token_urlsafe(16) carries exactly 128 random bits before encoding.
    return prefix + secrets.token_urlsafe(16)


def validate_idempotency_key(value: str) -> str:
    if not isinstance(value, str) or _IDEMPOTENCY_RE.fullmatch(value) is None:
        raise ValueError("invalid Idempotency-Key")
    return value


def _private_digest_with_key(key_id: str, key: bytes, domain: str, value: Any) -> str:
    if domain not in DIGEST_DOMAINS:
        raise ValueError("unsupported private digest domain")
    if _KEY_ID_RE.fullmatch(key_id) is None or len(key) < 32:
        raise ValueError("invalid private digest key")
    message = b"synderesis:labora-bridge:" + CONTRACT_VERSION.encode("ascii")
    message += b":private-digest\x00" + key_id.encode("ascii") + b"\x00"
    message += domain.encode("ascii") + b"\x00" + canonical_json(value)
    return hmac.new(key, message, hashlib.sha256).hexdigest()


def private_digest(
    keyring: str | Mapping[str, Any],
    domain: str,
    value: Any,
) -> tuple[str, str]:
    """Digest private transient data with the active retained key.

    The return order is ``(digest, key_id)`` so both values can be persisted
    together.  The input value is canonicalized in memory and is never stored.
    """

    ring = parse_keyring(keyring)
    key_id = ring["active_key_id"]
    return _private_digest_with_key(key_id, ring["keys"][key_id], domain, value), key_id


def recompute_private_digest(
    keyring: str | Mapping[str, Any],
    domain: str,
    value: Any,
    key_id: str,
) -> str:
    """Recompute a stored fingerprint under its retained key ID."""

    ring = parse_keyring(keyring)
    key = ring["keys"].get(key_id)
    if key is None:
        raise ValueError("private digest key has been rotated out")
    return _private_digest_with_key(key_id, key, domain, value)


def idempotency_key_candidates(
    keyring: str | Mapping[str, Any],
    value: str,
) -> list[tuple[str, str]]:
    """Return ``(digest, key_id)`` lookup candidates, active key first."""

    validated = validate_idempotency_key(value)
    ring = parse_keyring(keyring)
    active = ring["active_key_id"]
    key_ids = [active, *sorted(key_id for key_id in ring["keys"] if key_id != active)]
    return [
        (
            _private_digest_with_key(
                key_id, ring["keys"][key_id], "idempotency", validated
            ),
            key_id,
        )
        for key_id in key_ids
    ]


def approval_context_digest(
    keyring: str | Mapping[str, Any],
    *,
    customer_id: str,
    principal_id: str,
    run_id: str,
    action_id: str,
    action_digest: str,
    workspace_manifest_digest: str,
    capability_descriptor_digest: str,
    decision: Literal["approved", "denied"],
    exp: int,
) -> tuple[str, str]:
    """Bind an approval decision to its complete closed authority context."""

    _require_identity(customer_id, "customer")
    _require_identity(principal_id, "principal")
    _require_opaque_id(run_id, "lrun_")
    _require_opaque_id(action_id, "lact_")
    require_digest(action_digest)
    require_digest(workspace_manifest_digest)
    if not digests_equal(capability_descriptor_digest, CAPABILITY_DESCRIPTOR_DIGEST):
        raise ValueError("capability descriptor digest mismatch")
    if decision not in {"approved", "denied"}:
        raise ValueError("invalid approval decision")
    if type(exp) is not int or exp < 1:
        raise ValueError("invalid approval expiry")
    return private_digest(
        keyring,
        "approval",
        {
            "customer": customer_id,
            "principal": principal_id,
            "run": run_id,
            "action": action_id,
            "action_digest": action_digest,
            "workspace_manifest_digest": workspace_manifest_digest,
            "capability_descriptor_digest": capability_descriptor_digest,
            "decision": decision,
            "exp": exp,
        },
    )


def _b64encode(value: bytes) -> str:
    return base64.urlsafe_b64encode(value).rstrip(b"=").decode("ascii")


def _b64decode(value: str) -> bytes:
    if (
        not isinstance(value, str)
        or not value
        or re.fullmatch(r"[A-Za-z0-9_-]+", value) is None
    ):
        raise ValueError("invalid base64url")
    try:
        decoded = base64.urlsafe_b64decode(value + "=" * (-len(value) % 4))
    except (ValueError, TypeError) as exc:
        raise ValueError("invalid base64url") from exc
    if not hmac.compare_digest(_b64encode(decoded), value):
        raise ValueError("non-canonical base64url")
    return decoded


def parse_keyring(value: str | Mapping[str, Any]) -> dict[str, Any]:
    """Validate dedicated key configuration without reading environment state."""

    try:
        raw = json.loads(value) if isinstance(value, str) else dict(value)
    except (TypeError, ValueError, json.JSONDecodeError) as exc:
        raise ValueError("invalid Labora bridge keyring") from exc
    if set(raw) != {"active_key_id", "keys"}:
        raise ValueError("keyring schema is closed")
    active = raw.get("active_key_id")
    encoded_keys = raw.get("keys")
    if not isinstance(active, str) or _KEY_ID_RE.fullmatch(active) is None:
        raise ValueError("invalid active key ID")
    if not isinstance(encoded_keys, Mapping) or not encoded_keys:
        raise ValueError("keyring contains no keys")
    keys: dict[str, bytes] = {}
    for key_id, encoded in encoded_keys.items():
        if not isinstance(key_id, str) or _KEY_ID_RE.fullmatch(key_id) is None:
            raise ValueError("invalid key ID")
        key = encoded if isinstance(encoded, bytes) else _b64decode(encoded)
        if len(key) < 32:
            raise ValueError("signing keys must contain at least 256 bits")
        keys[key_id] = key
    if active not in keys:
        raise ValueError("active signing key is unavailable")
    return {"active_key_id": active, "keys": keys}


_ENVELOPE_FIELDS = {
    "action_offer": frozenset(
        {
            "version",
            "type",
            "issuer",
            "audience",
            "customer_id",
            "principal_id",
            "run_id",
            "action_id",
            "sequence",
            "request_digest",
            "action_digest",
            "workspace_manifest_digest",
            "capability_descriptor_digest",
            "rolling_chain_digest",
            "rolling_chain_sequence",
            "iat",
            "exp",
        }
    ),
    "execution": frozenset(
        {
            "version",
            "type",
            "issuer",
            "audience",
            "customer_id",
            "principal_id",
            "run_id",
            "action_id",
            "execution_id",
            "sequence",
            "request_digest",
            "action_digest",
            "approval_digest",
            "workspace_manifest_digest",
            "capability_descriptor_digest",
            "rolling_chain_digest",
            "rolling_chain_sequence",
            "iat",
            "exp",
        }
    ),
    "continuation": frozenset(
        {
            "version",
            "type",
            "issuer",
            "audience",
            "customer_id",
            "principal_id",
            "run_id",
            "action_id",
            "receipt_id",
            "sequence",
            "request_digest",
            "action_digest",
            "result_digest",
            "receipt_digest",
            "workspace_manifest_digest",
            "capability_descriptor_digest",
            "rolling_chain_digest",
            "rolling_chain_sequence",
            "iat",
            "exp",
        }
    ),
}


def _validate_claims(claims: Mapping[str, Any]) -> dict[str, Any]:
    clean = dict(claims)
    envelope_type = clean.get("type")
    expected = _ENVELOPE_FIELDS.get(envelope_type)
    if expected is None or set(clean) != expected:
        raise ValueError("invalid envelope claims")
    if clean["version"] != CONTRACT_VERSION:
        raise ValueError("invalid envelope version")
    if clean["issuer"] != ENVELOPE_ISSUER or clean["audience"] != ENVELOPE_AUDIENCE:
        raise ValueError("invalid envelope authority")
    for field in ("customer_id", "principal_id"):
        if (
            not isinstance(clean[field], str)
            or _IDENTITY_RE.fullmatch(clean[field]) is None
        ):
            raise ValueError(f"invalid {field}")
    for field in expected & {
        "request_digest",
        "action_digest",
        "approval_digest",
        "result_digest",
        "receipt_digest",
        "workspace_manifest_digest",
        "capability_descriptor_digest",
        "rolling_chain_digest",
    }:
        require_digest(clean[field])
    if not digests_equal(
        clean["capability_descriptor_digest"], CAPABILITY_DESCRIPTOR_DIGEST
    ):
        raise ValueError("envelope capability descriptor mismatch")
    for field, prefix in (
        ("run_id", "lrun_"),
        ("action_id", "lact_"),
        ("execution_id", "lexe_"),
        ("receipt_id", "lrcpt_"),
    ):
        if field in clean and (
            not isinstance(clean[field], str)
            or not clean[field].startswith(prefix)
            or _OPAQUE_ID_RE.fullmatch(clean[field]) is None
        ):
            raise ValueError(f"invalid {field}")
    if type(clean.get("sequence")) is not int or clean["sequence"] < 1:
        raise ValueError("invalid sequence")
    rolling_sequence = clean.get("rolling_chain_sequence")
    if (
        type(rolling_sequence) is not int
        or not 0 <= rolling_sequence <= MAX_RUN_ACTIONS
    ):
        raise ValueError("invalid rolling chain sequence")
    expected_rolling_sequence = (
        clean["sequence"] if envelope_type == "continuation" else clean["sequence"] - 1
    )
    if rolling_sequence != expected_rolling_sequence:
        raise ValueError("envelope rolling chain sequence mismatch")
    if rolling_sequence == 0 and not hmac.compare_digest(
        clean["rolling_chain_digest"], EMPTY_ROLLING_CHAIN_DIGEST
    ):
        raise ValueError("invalid empty rolling chain digest")
    issued_at, expires_at = clean.get("iat"), clean.get("exp")
    if (
        type(issued_at) is not int
        or type(expires_at) is not int
        or issued_at < 1
        or expires_at <= issued_at
        or expires_at - issued_at > MAX_ENVELOPE_LIFETIME_SECONDS
    ):
        raise ValueError("invalid envelope lifetime")
    return clean


def sign_envelope(keyring: str | Mapping[str, Any], claims: Mapping[str, Any]) -> str:
    ring = parse_keyring(keyring)
    body = {"v": 1, "kid": ring["active_key_id"], "claims": _validate_claims(claims)}
    payload = canonical_json(body)
    signature = hmac.new(
        ring["keys"][ring["active_key_id"]], payload, hashlib.sha256
    ).digest()
    return f"{_b64encode(payload)}.{_b64encode(signature)}"


def verify_envelope(
    keyring: str | Mapping[str, Any],
    token: str,
    *,
    expected_type: str,
    customer_id: str,
    principal_id: str,
    run_id: str,
    action_id: str,
    sequence: int,
    now: int | None = None,
) -> dict[str, Any]:
    """Authenticate and bind an envelope.

    ``now=None`` is intentionally signature/digest-only for bounded forensic
    inspection of historical authority. Every live production caller must pass
    an explicit current integer timestamp so expiry remains fail closed.
    """

    ring = parse_keyring(keyring)
    if not isinstance(token, str) or token.count(".") != 1:
        raise ValueError("invalid envelope")
    payload_part, signature_part = token.split(".")
    payload, signature = _b64decode(payload_part), _b64decode(signature_part)
    try:
        body = json.loads(payload)
    except (UnicodeDecodeError, json.JSONDecodeError) as exc:
        raise ValueError("invalid envelope") from exc
    if (
        not isinstance(body, dict)
        or set(body) != {"v", "kid", "claims"}
        or body["v"] != 1
    ):
        raise ValueError("unsupported envelope")
    key_id = body.get("kid")
    key = ring["keys"].get(key_id)
    if key is None:
        raise ValueError("envelope key has been rotated out")
    expected_signature = hmac.new(key, payload, hashlib.sha256).digest()
    if not hmac.compare_digest(signature, expected_signature):
        raise ValueError("invalid envelope signature")
    # Reject alternate JSON encodings even when signed by a trusted key.
    if not hmac.compare_digest(payload, canonical_json(body)):
        raise ValueError("non-canonical envelope")
    claims = _validate_claims(body.get("claims", {}))
    bindings = {
        "type": expected_type,
        "customer_id": customer_id,
        "principal_id": principal_id,
        "run_id": run_id,
        "action_id": action_id,
        "sequence": sequence,
    }
    for field, expected in bindings.items():
        actual = claims.get(field)
        if type(actual) is not type(expected) or not hmac.compare_digest(
            str(actual), str(expected)
        ):
            raise ValueError(f"envelope {field} mismatch")
    if now is not None:
        if (
            type(now) is not int
            or claims["exp"] <= now
            or claims["iat"] > now + MAX_CLOCK_SKEW_SECONDS
        ):
            raise ValueError("envelope expired")
    return claims


def _validated_relative_path(value: str) -> str:
    if not isinstance(value, str) or not value or "\\" in value or "\x00" in value:
        raise ValueError("path must be a normalized POSIX relative path")
    path = PurePosixPath(value)
    parts = path.parts
    if (
        path.is_absolute()
        or value != path.as_posix()
        or any(part in {"", ".", ".."} for part in parts)
    ):
        raise ValueError("path escapes or is not normalized")
    lowered = tuple(part.casefold() for part in parts)
    if ".git" in lowered:
        raise ValueError("Git metadata is unavailable")
    if any(_CREDENTIAL_PART_RE.search(part) for part in lowered):
        raise ValueError("credential-like paths are unavailable")
    if lowered[-1].endswith(_ARCHIVE_SUFFIXES):
        raise ValueError("archives are unavailable")
    return path.as_posix()


def validate_workspace_path(
    value: str, *, symlink_components: Sequence[str] = ()
) -> str:
    """Lexically confine a path; the executor supplies any detected symlink parts."""

    normalized = _validated_relative_path(value)
    if symlink_components:
        normalized_parts = normalized.split("/")
        supplied = {str(part).casefold() for part in symlink_components}
        if any(part.casefold() in supplied for part in normalized_parts):
            raise ValueError("symlink/reparse traversal is unavailable")
    return normalized


def validate_local_action(action: Mapping[str, Any]) -> dict[str, Any]:
    """Validate one closed action and return the fixed executor instruction."""

    if not isinstance(action, Mapping):
        raise ValueError("action must be an object")
    kind = action.get("type")
    try:
        if kind == "workspace.read_text":
            parsed = ReadTextAction.model_validate(action)
            return {**parsed.model_dump(), "path": validate_workspace_path(parsed.path)}
        if kind == "workspace.write_text_atomic":
            parsed = WriteTextAtomicAction.model_validate(action)
            return {**parsed.model_dump(), "path": validate_workspace_path(parsed.path)}
        if kind == "git.status":
            GitStatusAction.model_validate(action)
            return {
                "type": kind,
                "argv": [
                    "git",
                    "--no-optional-locks",
                    "status",
                    "--short",
                    "--untracked-files=all",
                ],
                "env": _fixed_git_env(),
                "max_bytes": MAX_GIT_OUTPUT_BYTES,
            }
        if kind == "git.diff":
            parsed = GitDiffAction.model_validate(action)
            paths = [validate_workspace_path(path) for path in parsed.paths]
            argv = [
                "git",
                "--no-optional-locks",
                "diff",
                "--no-ext-diff",
                "--no-textconv",
                "--ignore-submodules=all",
            ]
            if parsed.scope == "staged":
                argv.append("--cached")
            if paths:
                argv.extend(["--", *paths])
            return {
                **parsed.model_dump(),
                "paths": paths,
                "argv": argv,
                "env": _fixed_git_env(),
            }
    except (TypeError, ValueError) as exc:
        raise ValueError("invalid local action") from exc
    raise ValueError("unsupported local action")


def _fixed_git_env() -> dict[str, str]:
    return {
        "GIT_TERMINAL_PROMPT": "0",
        "GIT_CONFIG_NOSYSTEM": "1",
        "GIT_CONFIG_GLOBAL": "/dev/null",
        "GIT_OPTIONAL_LOCKS": "0",
        "GIT_PAGER": "cat",
        "PAGER": "cat",
    }


SCHEMA_SQL = """
CREATE TABLE IF NOT EXISTS labora_runs (
    run_id TEXT PRIMARY KEY,
    customer_id TEXT NOT NULL,
    principal_id TEXT NOT NULL,
    request_digest TEXT NOT NULL,
    request_digest_kid TEXT NOT NULL,
    workspace_manifest_digest TEXT,
    capability_descriptor_digest TEXT,
    rolling_chain_digest TEXT NOT NULL DEFAULT '0000000000000000000000000000000000000000000000000000000000000000',
    rolling_chain_sequence INTEGER NOT NULL DEFAULT 0 CHECK (rolling_chain_sequence BETWEEN 0 AND 8),
    internal_state TEXT NOT NULL,
    public_state TEXT NOT NULL CHECK (public_state IN (
        'queued','planning','working','reviewing','waiting_for_approval',
        'completed','completed_with_limits','cancelled','failed'
    )),
    outcome TEXT NOT NULL DEFAULT '',
    sequence INTEGER NOT NULL DEFAULT 0 CHECK (sequence BETWEEN 0 AND 8),
    decision_count INTEGER NOT NULL DEFAULT 0 CHECK (decision_count BETWEEN 0 AND 16),
    action_count INTEGER NOT NULL DEFAULT 0 CHECK (action_count BETWEEN 0 AND 8),
    provider_call_count INTEGER NOT NULL DEFAULT 0 CHECK (provider_call_count BETWEEN 0 AND 16),
    active_action_id TEXT,
    quota_reservation_id INTEGER,
    input_units INTEGER NOT NULL DEFAULT 0 CHECK (input_units BETWEEN 0 AND 1000000000),
    output_units INTEGER NOT NULL DEFAULT 0 CHECK (output_units BETWEEN 0 AND 1000000000),
    cancellation_requested INTEGER NOT NULL DEFAULT 0 CHECK (cancellation_requested IN (0, 1)),
    provider_dispatched INTEGER NOT NULL DEFAULT 0 CHECK (provider_dispatched IN (0, 1)),
    metering_finalized INTEGER NOT NULL DEFAULT 0 CHECK (metering_finalized IN (0, 1)),
    version INTEGER NOT NULL DEFAULT 0 CHECK (version >= 0),
    ttl_deadline INTEGER NOT NULL,
    expires_at INTEGER NOT NULL CHECK (expires_at >= ttl_deadline),
    created_at INTEGER NOT NULL,
    updated_at INTEGER NOT NULL,
    UNIQUE (run_id, customer_id, principal_id)
);
CREATE INDEX IF NOT EXISTS idx_labora_runs_scope
ON labora_runs(customer_id, principal_id, run_id);

CREATE TABLE IF NOT EXISTS labora_actions (
    action_id TEXT PRIMARY KEY,
    run_id TEXT NOT NULL,
    customer_id TEXT NOT NULL,
    principal_id TEXT NOT NULL,
    sequence INTEGER NOT NULL CHECK (sequence BETWEEN 1 AND 8),
    capability TEXT NOT NULL CHECK (capability IN (
        'workspace.read_text','workspace.write_text_atomic','git.status','git.diff'
    )),
    action_digest TEXT NOT NULL,
    action_digest_kid TEXT NOT NULL,
    approval_mode TEXT NOT NULL CHECK (approval_mode IN ('workspace_grant','action_specific')),
    state TEXT NOT NULL CHECK (state IN (
        'offered','claimed','result_received','consumed','denied','cancelled','expired'
    )),
    execution_id TEXT,
    approval_digest TEXT,
    approval_digest_kid TEXT,
    result_digest TEXT,
    result_chain_digest TEXT,
    result_chain_digest_kid TEXT,
    receipt_id TEXT,
    receipt_digest TEXT,
    receipt_digest_kid TEXT,
    version INTEGER NOT NULL DEFAULT 0 CHECK (version >= 0),
    expires_at INTEGER NOT NULL,
    created_at INTEGER NOT NULL,
    updated_at INTEGER NOT NULL,
    UNIQUE (run_id, customer_id, principal_id, sequence),
    FOREIGN KEY (run_id, customer_id, principal_id)
        REFERENCES labora_runs(run_id, customer_id, principal_id) ON DELETE CASCADE
);
CREATE UNIQUE INDEX IF NOT EXISTS idx_labora_one_active_action
ON labora_actions(customer_id, principal_id, run_id)
WHERE state IN ('offered', 'claimed', 'result_received');
CREATE INDEX IF NOT EXISTS idx_labora_actions_scope
ON labora_actions(customer_id, principal_id, run_id, action_id);

CREATE TABLE IF NOT EXISTS labora_model_executions (
    execution_id TEXT PRIMARY KEY,
    run_id TEXT NOT NULL,
    customer_id TEXT NOT NULL,
    principal_id TEXT NOT NULL,
    operation TEXT NOT NULL,
    request_digest TEXT NOT NULL,
    request_digest_kid TEXT NOT NULL,
    lease_state TEXT NOT NULL CHECK (lease_state IN (
        'prepared','dispatch_committed','returned','abandoned'
    )),
    quota_reservation_id INTEGER,
    provider_dispatched INTEGER NOT NULL DEFAULT 0 CHECK (provider_dispatched IN (0, 1)),
    response_digest TEXT,
    response_digest_kid TEXT,
    input_units INTEGER NOT NULL DEFAULT 0 CHECK (input_units BETWEEN 0 AND 1000000000),
    output_units INTEGER NOT NULL DEFAULT 0 CHECK (output_units BETWEEN 0 AND 1000000000),
    pricing_provider TEXT,
    pricing_funding TEXT CHECK (pricing_funding IN ('paid','complimentary','nonbilling')),
    pricing_kind TEXT,
    cost_microunits INTEGER CHECK (
        cost_microunits IS NULL OR cost_microunits BETWEEN 0 AND 1000000000000
    ),
    outcome TEXT NOT NULL DEFAULT '',
    version INTEGER NOT NULL DEFAULT 0 CHECK (version >= 0),
    expires_at INTEGER NOT NULL,
    created_at INTEGER NOT NULL,
    updated_at INTEGER NOT NULL,
    UNIQUE (run_id, customer_id, principal_id, operation, request_digest_kid, request_digest),
    FOREIGN KEY (run_id, customer_id, principal_id)
        REFERENCES labora_runs(run_id, customer_id, principal_id) ON DELETE CASCADE
);
CREATE INDEX IF NOT EXISTS idx_labora_model_scope
ON labora_model_executions(customer_id, principal_id, run_id, lease_state);

CREATE TABLE IF NOT EXISTS labora_idempotency (
    customer_id TEXT NOT NULL,
    principal_id TEXT NOT NULL,
    operation TEXT NOT NULL,
    key_digest TEXT NOT NULL,
    key_digest_kid TEXT NOT NULL,
    request_digest TEXT NOT NULL,
    request_digest_kid TEXT NOT NULL,
    resource_id TEXT,
    related_id TEXT,
    safe_response_state TEXT NOT NULL,
    version INTEGER NOT NULL DEFAULT 0 CHECK (version >= 0),
    expires_at INTEGER NOT NULL,
    created_at INTEGER NOT NULL,
    updated_at INTEGER NOT NULL,
    PRIMARY KEY (customer_id, principal_id, operation, key_digest_kid, key_digest)
);
CREATE INDEX IF NOT EXISTS idx_labora_idempotency_scope
ON labora_idempotency(customer_id, principal_id, operation, expires_at);
"""

_REQUIRED_SCHEMA_COLUMNS = {
    "labora_runs": {
        "run_id",
        "customer_id",
        "principal_id",
        "request_digest",
        "request_digest_kid",
        "workspace_manifest_digest",
        "capability_descriptor_digest",
        "rolling_chain_digest",
        "rolling_chain_sequence",
        "decision_count",
        "action_count",
        "provider_call_count",
        "ttl_deadline",
        "cancellation_requested",
        "metering_finalized",
    },
    "labora_actions": {
        "action_id",
        "run_id",
        "customer_id",
        "principal_id",
        "action_digest",
        "action_digest_kid",
        "result_chain_digest",
        "result_chain_digest_kid",
    },
    "labora_model_executions": {
        "execution_id",
        "run_id",
        "customer_id",
        "principal_id",
        "lease_state",
        "request_digest_kid",
        "response_digest_kid",
        "pricing_funding",
    },
    "labora_idempotency": {
        "customer_id",
        "principal_id",
        "operation",
        "key_digest",
        "key_digest_kid",
        "request_digest",
        "request_digest_kid",
        "safe_response_state",
    },
}


def _now(value: int | None) -> int:
    timestamp = int(time.time()) if value is None else value
    if type(timestamp) is not int or timestamp < 1:
        raise ValueError("invalid timestamp")
    return timestamp


def _require_identity(value: str, name: str) -> str:
    if not isinstance(value, str) or _IDENTITY_RE.fullmatch(value) is None:
        raise ValueError(f"invalid {name}")
    return value


def _require_opaque_id(value: str, prefix: str) -> str:
    if (
        not isinstance(value, str)
        or not value.startswith(prefix)
        or _OPAQUE_ID_RE.fullmatch(value) is None
    ):
        raise ValueError(f"invalid {prefix[:-1]} ID")
    return value


def _require_key_id(value: str) -> str:
    if not isinstance(value, str) or _KEY_ID_RE.fullmatch(value) is None:
        raise ValueError("invalid digest key ID")
    return value


def _require_operation(value: str) -> str:
    if not isinstance(value, str) or _OPERATION_RE.fullmatch(value) is None:
        raise ValueError("invalid bridge operation")
    return value


def _require_safe_state(value: str) -> str:
    if not isinstance(value, str) or _SAFE_STATE_RE.fullmatch(value) is None:
        raise ValueError("invalid safe response state")
    return value


def _require_resource_id(value: str | None) -> str | None:
    if value is not None and _OPAQUE_ID_RE.fullmatch(value) is None:
        raise ValueError("invalid replay-safe resource ID")
    return value


def _run_immediate(connection: sqlite3.Connection, operation: Any) -> Any:
    if connection.in_transaction:
        raise RuntimeError("self-committing bridge wrapper requires an idle connection")
    connection.execute("BEGIN IMMEDIATE")
    try:
        result = operation()
        connection.commit()
        return result
    except Exception:
        connection.rollback()
        raise


def init_bridge_schema(connection: sqlite3.Connection) -> None:
    """Create and verify the four content-free authoritative bridge tables."""

    connection.execute("PRAGMA foreign_keys = ON")
    connection.executescript(SCHEMA_SQL)
    _migrate_bridge_schema(connection)
    for table, required in _REQUIRED_SCHEMA_COLUMNS.items():
        actual = {
            str(row[1]) for row in connection.execute(f"PRAGMA table_info({table})")
        }
        if not required <= actual:
            raise RuntimeError(
                f"incompatible {table} schema; bridge must remain unavailable"
            )


def _migrate_bridge_schema(connection: sqlite3.Connection) -> None:
    """Idempotently add content-free 1.2 closure columns to legacy databases."""

    existing = {
        str(row[1]) for row in connection.execute("PRAGMA table_info(labora_runs)")
    }
    additions = {
        "workspace_manifest_digest": "TEXT",
        "capability_descriptor_digest": "TEXT",
        "rolling_chain_digest": (
            "TEXT NOT NULL DEFAULT "
            "'0000000000000000000000000000000000000000000000000000000000000000'"
        ),
        "rolling_chain_sequence": (
            "INTEGER NOT NULL DEFAULT 0 CHECK (rolling_chain_sequence BETWEEN 0 AND 8)"
        ),
    }
    for column, declaration in additions.items():
        if column not in existing:
            connection.execute(
                f"ALTER TABLE labora_runs ADD COLUMN {column} {declaration}"
            )


def missing_retained_key_ids(
    connection: sqlite3.Connection,
    keyring: str | Mapping[str, Any],
    *,
    now: int | None = None,
) -> set[str]:
    """Return key IDs referenced by metadata that cleanup must retain."""

    timestamp = _now(now)
    retained = set(parse_keyring(keyring)["keys"])
    key_columns = {
        "labora_runs": ("request_digest_kid",),
        "labora_actions": (
            "action_digest_kid",
            "approval_digest_kid",
            "result_chain_digest_kid",
            "receipt_digest_kid",
        ),
        "labora_model_executions": (
            "request_digest_kid",
            "response_digest_kid",
        ),
        "labora_idempotency": ("key_digest_kid", "request_digest_kid"),
    }
    predicate_columns = {
        "labora_runs": {
            "public_state",
            "metering_finalized",
            "expires_at",
        },
        "labora_actions": {"state", "expires_at"},
        "labora_model_executions": {"lease_state", "outcome", "expires_at"},
        "labora_idempotency": {
            "customer_id",
            "principal_id",
            "resource_id",
            "related_id",
            "expires_at",
        },
    }
    predicates = {
        "labora_runs": """
            (expires_at>? OR public_state NOT IN
                ('completed','completed_with_limits','cancelled','failed')
             OR metering_finalized=0)
        """,
        "labora_actions": """
            (expires_at>? OR state IN ('offered','claimed','result_received'))
        """,
        "labora_model_executions": """
            (expires_at>? OR lease_state IN ('prepared','dispatch_committed')
             OR (lease_state='abandoned' AND outcome='provider_outcome_unknown'))
        """,
        "labora_idempotency": """
            (expires_at>? OR EXISTS (
                SELECT 1 FROM labora_runs r
                WHERE r.customer_id=labora_idempotency.customer_id
                  AND r.principal_id=labora_idempotency.principal_id
                  AND (r.run_id=labora_idempotency.resource_id
                       OR r.run_id=labora_idempotency.related_id)
                  AND (r.expires_at>? OR r.public_state NOT IN
                        ('completed','completed_with_limits','cancelled','failed')
                       OR r.metering_finalized=0)
            ) OR EXISTS (
                SELECT 1 FROM labora_actions a
                WHERE a.customer_id=labora_idempotency.customer_id
                  AND a.principal_id=labora_idempotency.principal_id
                  AND (a.action_id=labora_idempotency.resource_id
                       OR a.action_id=labora_idempotency.related_id)
                  AND (a.expires_at>?
                       OR a.state IN ('offered','claimed','result_received'))
            ) OR EXISTS (
                SELECT 1 FROM labora_model_executions e
                WHERE e.customer_id=labora_idempotency.customer_id
                  AND e.principal_id=labora_idempotency.principal_id
                  AND (e.execution_id=labora_idempotency.resource_id
                       OR e.execution_id=labora_idempotency.related_id)
                  AND (e.expires_at>?
                       OR e.lease_state IN ('prepared','dispatch_committed')
                       OR (e.lease_state='abandoned'
                           AND e.outcome='provider_outcome_unknown'))
            ))
        """,
    }
    referenced: set[str] = set()
    for table, columns in key_columns.items():
        actual = {
            str(row[1]) for row in connection.execute(f"PRAGMA table_info({table})")
        }
        if not (set(columns) | predicate_columns[table]) <= actual:
            raise RuntimeError(
                f"incompatible {table} schema; bridge must remain unavailable"
            )
        parameters = (
            (timestamp, timestamp, timestamp, timestamp)
            if table == "labora_idempotency"
            else (timestamp,)
        )
        for column in columns:
            referenced.update(
                str(row[0])
                for row in connection.execute(
                    f"SELECT DISTINCT {column} FROM {table} "
                    f"WHERE {predicates[table]} AND {column} IS NOT NULL",
                    parameters,
                )
                if str(row[0])
            )
    return referenced - retained


def create_run_tx(
    connection: sqlite3.Connection,
    *,
    run_id: str,
    customer_id: str,
    principal_id: str,
    request_digest: str,
    request_digest_kid: str,
    workspace_manifest_digest: str,
    capability_descriptor_digest: str,
    expires_at: int,
    ttl_deadline: int,
    quota_reservation_id: int | None = None,
    now: int | None = None,
) -> None:
    """Insert a run without committing, for composition with admission work."""

    timestamp = _now(now)
    _require_opaque_id(run_id, "lrun_")
    _require_identity(customer_id, "customer")
    _require_identity(principal_id, "principal")
    require_digest(request_digest)
    _require_key_id(request_digest_kid)
    require_digest(workspace_manifest_digest)
    if not digests_equal(capability_descriptor_digest, CAPABILITY_DESCRIPTOR_DIGEST):
        raise ValueError("capability descriptor digest mismatch")
    if (
        type(ttl_deadline) is not int
        or type(expires_at) is not int
        or ttl_deadline <= timestamp
        or ttl_deadline > timestamp + MAX_RUN_LIFETIME_SECONDS
        or expires_at < ttl_deadline
        or expires_at > timestamp + MAX_RUN_LIFETIME_SECONDS
    ):
        raise ValueError("invalid run lifetime")
    if quota_reservation_id is not None and (
        type(quota_reservation_id) is not int or quota_reservation_id < 1
    ):
        raise ValueError("invalid quota reservation")
    connection.execute(
        """INSERT INTO labora_runs
           (run_id, customer_id, principal_id, request_digest, request_digest_kid,
            workspace_manifest_digest, capability_descriptor_digest,
            internal_state, public_state, quota_reservation_id, ttl_deadline,
            expires_at, created_at, updated_at)
           VALUES (?, ?, ?, ?, ?, ?, ?, 'queued', 'queued', ?, ?, ?, ?, ?)""",
        (
            run_id,
            customer_id,
            principal_id,
            request_digest,
            request_digest_kid,
            workspace_manifest_digest,
            capability_descriptor_digest,
            quota_reservation_id,
            ttl_deadline,
            expires_at,
            timestamp,
            timestamp,
        ),
    )


def create_run(connection: sqlite3.Connection, **kwargs: Any) -> None:
    """Self-committing wrapper around :func:`create_run_tx`."""

    _run_immediate(connection, lambda: create_run_tx(connection, **kwargs))


def cas_run_state_tx(
    connection: sqlite3.Connection,
    *,
    run_id: str,
    customer_id: str,
    principal_id: str,
    expected_state: str,
    expected_version: int,
    new_state: str,
    public_state: str,
    active_action_id: str | None,
    sequence: int,
    now: int | None = None,
) -> int:
    """CAS a non-cancellation run transition without committing."""

    timestamp = _now(now)
    _require_opaque_id(run_id, "lrun_")
    _require_identity(customer_id, "customer")
    _require_identity(principal_id, "principal")
    if (
        expected_state not in INTERNAL_STATES
        or new_state not in INTERNAL_STATES
        or public_state not in PUBLIC_PHASES
        or new_state in TERMINAL_PHASES
        or public_state in TERMINAL_PHASES
        or type(expected_version) is not int
        or expected_version < 0
        or type(sequence) is not int
        or not 0 <= sequence <= MAX_RUN_ACTIONS
    ):
        raise ValueError("invalid run transition")
    if active_action_id is not None:
        _require_opaque_id(active_action_id, "lact_")
    cursor = connection.execute(
        """UPDATE labora_runs
           SET internal_state = ?, public_state = ?, active_action_id = ?, sequence = ?,
               version = version + 1, updated_at = ?
           WHERE run_id = ? AND customer_id = ? AND principal_id = ?
             AND internal_state = ? AND version = ? AND sequence <= ?
             AND cancellation_requested = 0 AND ttl_deadline > ?
             AND public_state NOT IN
                 ('completed','completed_with_limits','cancelled','failed')
             AND metering_finalized = 0
             AND internal_state NOT IN
                 ('metering_recorded','metering_recorded_cancelled')
             AND workspace_manifest_digest IS NOT NULL
             AND capability_descriptor_digest = ?""",
        (
            new_state,
            public_state,
            active_action_id,
            sequence,
            timestamp,
            run_id,
            customer_id,
            principal_id,
            expected_state,
            expected_version,
            sequence,
            timestamp,
            CAPABILITY_DESCRIPTOR_DIGEST,
        ),
    )
    if cursor.rowcount != 1:
        raise RuntimeError("run CAS conflict")
    return expected_version + 1


def cas_run_state(connection: sqlite3.Connection, **kwargs: Any) -> int:
    """Self-committing wrapper around :func:`cas_run_state_tx`."""

    return _run_immediate(connection, lambda: cas_run_state_tx(connection, **kwargs))


def insert_action_tx(
    connection: sqlite3.Connection,
    *,
    action_id: str,
    run_id: str,
    customer_id: str,
    principal_id: str,
    sequence: int,
    capability: str,
    action_digest: str,
    action_digest_kid: str,
    approval_mode: Literal["workspace_grant", "action_specific"],
    expires_at: int,
    now: int | None = None,
) -> None:
    """Offer one action only while its principal-scoped run remains cancellable."""

    timestamp = _now(now)
    _require_opaque_id(action_id, "lact_")
    _require_opaque_id(run_id, "lrun_")
    _require_identity(customer_id, "customer")
    _require_identity(principal_id, "principal")
    require_digest(action_digest)
    _require_key_id(action_digest_kid)
    if capability not in LOCAL_ACTION_TYPES or approval_mode not in {
        "workspace_grant",
        "action_specific",
    }:
        raise ValueError("invalid action metadata")
    if type(sequence) is not int or not 1 <= sequence <= MAX_RUN_ACTIONS:
        raise ValueError("invalid action sequence")
    if (
        type(expires_at) is not int
        or expires_at <= timestamp
        or expires_at > timestamp + MAX_ENVELOPE_LIFETIME_SECONDS
    ):
        raise ValueError("invalid action expiry")
    cursor = connection.execute(
        """INSERT INTO labora_actions
           (action_id, run_id, customer_id, principal_id, sequence, capability,
            action_digest, action_digest_kid, approval_mode, state, expires_at,
            created_at, updated_at)
           SELECT ?, ?, ?, ?, ?, ?, ?, ?, ?, 'offered', ?, ?, ?
           FROM labora_runs
           WHERE run_id = ? AND customer_id = ? AND principal_id = ?
             AND cancellation_requested = 0 AND ttl_deadline > ?
             AND public_state NOT IN
                 ('completed','completed_with_limits','cancelled','failed')
             AND metering_finalized = 0
             AND internal_state NOT IN
                 ('metering_recorded','metering_recorded_cancelled')
             AND workspace_manifest_digest IS NOT NULL
             AND capability_descriptor_digest = ?
             AND active_action_id IS NULL AND sequence = ? AND action_count < ?""",
        (
            action_id,
            run_id,
            customer_id,
            principal_id,
            sequence,
            capability,
            action_digest,
            action_digest_kid,
            approval_mode,
            expires_at,
            timestamp,
            timestamp,
            run_id,
            customer_id,
            principal_id,
            timestamp,
            CAPABILITY_DESCRIPTOR_DIGEST,
            sequence - 1,
            MAX_RUN_ACTIONS,
        ),
    )
    if cursor.rowcount != 1:
        raise RuntimeError("action admission conflict")
    run_cursor = connection.execute(
        """UPDATE labora_runs
           SET internal_state='waiting_for_approval', public_state='waiting_for_approval',
               active_action_id=?, sequence=?, action_count=action_count+1,
               version=version+1, updated_at=?
           WHERE run_id=? AND customer_id=? AND principal_id=?
             AND cancellation_requested=0 AND active_action_id IS NULL
             AND action_count < ?
             AND public_state NOT IN
                 ('completed','completed_with_limits','cancelled','failed')
             AND metering_finalized=0
             AND internal_state NOT IN
                 ('metering_recorded','metering_recorded_cancelled')""",
        (
            action_id,
            sequence,
            timestamp,
            run_id,
            customer_id,
            principal_id,
            MAX_RUN_ACTIONS,
        ),
    )
    if run_cursor.rowcount != 1:
        raise RuntimeError("run changed during action admission")


def cas_action_state_tx(
    connection: sqlite3.Connection,
    *,
    action_id: str,
    run_id: str,
    customer_id: str,
    principal_id: str,
    expected_state: str,
    expected_version: int,
    new_state: str,
    metadata: Mapping[str, str | None] | None = None,
    allow_cancelled_result: bool = False,
    now: int | None = None,
) -> int:
    """Consume one action state; only a late result may cross cancellation."""

    timestamp = _now(now)
    _require_opaque_id(action_id, "lact_")
    _require_opaque_id(run_id, "lrun_")
    _require_identity(customer_id, "customer")
    _require_identity(principal_id, "principal")
    if (
        expected_state not in ACTION_STATES
        or new_state not in ACTION_STATES
        or expected_version < 0
    ):
        raise ValueError("invalid action transition")
    if allow_cancelled_result and not (
        expected_state == "claimed" and new_state == "result_received"
    ):
        raise ValueError("only a late claimed result may cross cancellation")
    allowed = {
        "execution_id",
        "approval_digest",
        "approval_digest_kid",
        "result_digest",
        "result_chain_digest",
        "result_chain_digest_kid",
        "receipt_id",
        "receipt_digest",
        "receipt_digest_kid",
    }
    updates = dict(metadata or {})
    if set(updates) - allowed:
        raise ValueError("invalid action transition metadata")
    for name, value in updates.items():
        if value is None:
            continue
        if name.endswith("_kid"):
            _require_key_id(value)
        elif name.endswith("_digest"):
            require_digest(value)
        elif name == "execution_id":
            _require_opaque_id(value, "lexe_")
        elif name == "receipt_id":
            _require_opaque_id(value, "lrcpt_")
    assignments = ["state = ?", "version = version + 1", "updated_at = ?"]
    values: list[Any] = [new_state, timestamp]
    for name in sorted(updates):
        assignments.append(f"{name} = ?")
        values.append(updates[name])
    allow_cancelled = 1 if allow_cancelled_result else 0
    values.extend(
        [
            action_id,
            run_id,
            customer_id,
            principal_id,
            expected_state,
            expected_version,
            timestamp,
            allow_cancelled,
        ]
    )
    cursor = connection.execute(
        f"UPDATE labora_actions SET {', '.join(assignments)} "
        "WHERE action_id=? AND run_id=? AND customer_id=? AND principal_id=? "
        "AND state=? AND version=? AND EXISTS ("
        "SELECT 1 FROM labora_runs r WHERE r.run_id=labora_actions.run_id "
        "AND r.customer_id=labora_actions.customer_id AND r.principal_id=labora_actions.principal_id "
        "AND r.active_action_id=labora_actions.action_id "
        "AND r.sequence=labora_actions.sequence "
        "AND r.ttl_deadline>? AND ((r.cancellation_requested=0 "
        "AND r.public_state NOT IN "
        "('completed','completed_with_limits','cancelled','failed') "
        "AND r.metering_finalized=0 "
        "AND r.internal_state NOT IN "
        "('metering_recorded','metering_recorded_cancelled')) OR (?=1 "
        "AND r.cancellation_requested=1 AND r.provider_dispatched=1 "
        "AND r.public_state='cancelled' "
        "AND r.internal_state='cancelled' "
        "AND r.outcome='cancelled_after_dispatch')) "
        "AND r.workspace_manifest_digest IS NOT NULL "
        "AND r.capability_descriptor_digest=?)",
        [*values, CAPABILITY_DESCRIPTOR_DIGEST],
    )
    if cursor.rowcount != 1:
        raise RuntimeError("action CAS conflict")
    return expected_version + 1


def cas_action_state(connection: sqlite3.Connection, **kwargs: Any) -> int:
    """Self-committing wrapper around :func:`cas_action_state_tx`."""

    return _run_immediate(connection, lambda: cas_action_state_tx(connection, **kwargs))


def deny_action_tx(
    connection: sqlite3.Connection,
    *,
    action_id: str,
    run_id: str,
    customer_id: str,
    principal_id: str,
    expected_action_version: int,
    approval_digest: str,
    approval_digest_kid: str,
    now: int | None = None,
) -> tuple[int, int]:
    """Irreversibly deny one offer and move its run behind the metering gate."""

    timestamp = _now(now)
    action_version = cas_action_state_tx(
        connection,
        action_id=action_id,
        run_id=run_id,
        customer_id=customer_id,
        principal_id=principal_id,
        expected_state="offered",
        expected_version=expected_action_version,
        new_state="denied",
        metadata={
            "approval_digest": approval_digest,
            "approval_digest_kid": approval_digest_kid,
        },
        now=timestamp,
    )
    cursor = connection.execute(
        """UPDATE labora_runs
           SET internal_state='metering_pending', public_state='working',
               outcome='approval_denied', version=version+1, updated_at=?
           WHERE run_id=? AND customer_id=? AND principal_id=?
             AND active_action_id=? AND cancellation_requested=0
             AND public_state='waiting_for_approval'
             AND metering_finalized=0
             AND internal_state NOT IN
                 ('metering_recorded','metering_recorded_cancelled')
             AND workspace_manifest_digest IS NOT NULL
             AND capability_descriptor_digest=?""",
        (
            timestamp,
            run_id,
            customer_id,
            principal_id,
            action_id,
            CAPABILITY_DESCRIPTOR_DIGEST,
        ),
    )
    if cursor.rowcount != 1:
        raise RuntimeError("run changed during action denial")
    run_version = connection.execute(
        """SELECT version FROM labora_runs
           WHERE run_id=? AND customer_id=? AND principal_id=?""",
        (run_id, customer_id, principal_id),
    ).fetchone()
    if run_version is None:
        raise RuntimeError("denied action run is unavailable")
    return action_version, int(run_version[0])


def claim_idempotency_tx(
    connection: sqlite3.Connection,
    *,
    keyring: str | Mapping[str, Any],
    idempotency_key: str,
    request_value: Any,
    customer_id: str,
    principal_id: str,
    operation: str,
    resource_id: str | None,
    related_id: str | None,
    safe_response_state: str,
    expires_at: int,
    now: int | None = None,
) -> Literal["created", "replay"]:
    """Claim/replay an operation across every retained digest key, without commit."""

    timestamp = _now(now)
    _require_identity(customer_id, "customer")
    _require_identity(principal_id, "principal")
    _require_operation(operation)
    _require_safe_state(safe_response_state)
    _require_resource_id(resource_id)
    _require_resource_id(related_id)
    if (
        type(expires_at) is not int
        or expires_at <= timestamp
        or expires_at > timestamp + MAX_RUN_LIFETIME_SECONDS
    ):
        raise ValueError("invalid idempotency expiry")
    candidates = idempotency_key_candidates(keyring, idempotency_key)
    clauses = " OR ".join("(key_digest=? AND key_digest_kid=?)" for _ in candidates)
    parameters: list[Any] = [customer_id, principal_id, operation]
    for digest, key_id in candidates:
        parameters.extend([digest, key_id])
    rows = connection.execute(
        f"""SELECT key_digest, key_digest_kid, request_digest, request_digest_kid
            FROM labora_idempotency
            WHERE customer_id=? AND principal_id=? AND operation=? AND ({clauses})""",
        parameters,
    ).fetchall()
    if len(rows) > 1:
        raise RuntimeError("ambiguous retained-key idempotency state")
    if rows:
        stored_request = recompute_private_digest(
            keyring, "request", request_value, str(rows[0][3])
        )
        if not hmac.compare_digest(str(rows[0][2]), stored_request):
            raise ValueError("idempotency key conflicts with another request")
        return "replay"
    key_digest, key_id = candidates[0]
    request_digest = recompute_private_digest(keyring, "request", request_value, key_id)
    connection.execute(
        """INSERT INTO labora_idempotency
           (customer_id, principal_id, operation, key_digest, key_digest_kid,
            request_digest, request_digest_kid, resource_id, related_id,
            safe_response_state, expires_at, created_at, updated_at)
           VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)""",
        (
            customer_id,
            principal_id,
            operation,
            key_digest,
            key_id,
            request_digest,
            key_id,
            resource_id,
            related_id,
            safe_response_state,
            expires_at,
            timestamp,
            timestamp,
        ),
    )
    return "created"


def claim_idempotency(
    connection: sqlite3.Connection, **kwargs: Any
) -> Literal["created", "replay"]:
    """Self-committing wrapper around :func:`claim_idempotency_tx`."""

    return _run_immediate(
        connection, lambda: claim_idempotency_tx(connection, **kwargs)
    )


def complete_idempotency_tx(
    connection: sqlite3.Connection,
    *,
    customer_id: str,
    principal_id: str,
    operation: str,
    key_digest: str,
    key_digest_kid: str,
    request_digest: str,
    resource_id: str | None,
    related_id: str | None,
    safe_response_state: str,
    expected_version: int = 0,
    now: int | None = None,
) -> int:
    """Publish only replay-safe IDs/state after the admitted mutation commits."""

    timestamp = _now(now)
    _require_identity(customer_id, "customer")
    _require_identity(principal_id, "principal")
    _require_operation(operation)
    require_digest(key_digest)
    _require_key_id(key_digest_kid)
    require_digest(request_digest)
    _require_safe_state(safe_response_state)
    _require_resource_id(resource_id)
    _require_resource_id(related_id)
    cursor = connection.execute(
        """UPDATE labora_idempotency
           SET resource_id=?, related_id=?, safe_response_state=?,
               version=version+1, updated_at=?
           WHERE customer_id=? AND principal_id=? AND operation=?
             AND key_digest=? AND key_digest_kid=? AND request_digest=?
             AND version=?""",
        (
            resource_id,
            related_id,
            safe_response_state,
            timestamp,
            customer_id,
            principal_id,
            operation,
            key_digest,
            key_digest_kid,
            request_digest,
            expected_version,
        ),
    )
    if cursor.rowcount != 1:
        raise RuntimeError("idempotency completion conflict")
    return expected_version + 1


def insert_prepared_execution_tx(
    connection: sqlite3.Connection,
    *,
    execution_id: str,
    run_id: str,
    customer_id: str,
    principal_id: str,
    operation: str,
    request_digest: str,
    request_digest_kid: str,
    quota_reservation_id: int | None,
    expires_at: int,
    now: int | None = None,
) -> None:
    """Insert the only provider lease eligible for a later dispatch CAS."""

    timestamp = _now(now)
    _require_opaque_id(execution_id, "lexe_")
    _require_opaque_id(run_id, "lrun_")
    _require_identity(customer_id, "customer")
    _require_identity(principal_id, "principal")
    _require_operation(operation)
    require_digest(request_digest)
    _require_key_id(request_digest_kid)
    if quota_reservation_id is not None and (
        type(quota_reservation_id) is not int or quota_reservation_id < 1
    ):
        raise ValueError("invalid quota reservation")
    if (
        type(expires_at) is not int
        or expires_at <= timestamp
        or expires_at > timestamp + MAX_RUN_LIFETIME_SECONDS
    ):
        raise ValueError("invalid execution expiry")
    cursor = connection.execute(
        """INSERT INTO labora_model_executions
           (execution_id, run_id, customer_id, principal_id, operation,
            request_digest, request_digest_kid, lease_state, quota_reservation_id,
            expires_at, created_at, updated_at)
           SELECT ?, ?, ?, ?, ?, ?, ?, 'prepared', ?, ?, ?, ?
           FROM labora_runs
           WHERE run_id=? AND customer_id=? AND principal_id=?
             AND cancellation_requested=0 AND ttl_deadline>? AND provider_call_count<?
             AND decision_count<?
             AND public_state NOT IN
                 ('completed','completed_with_limits','cancelled','failed')
             AND metering_finalized=0
             AND internal_state NOT IN
                 ('metering_recorded','metering_recorded_cancelled')
             AND workspace_manifest_digest IS NOT NULL
             AND capability_descriptor_digest=?""",
        (
            execution_id,
            run_id,
            customer_id,
            principal_id,
            operation,
            request_digest,
            request_digest_kid,
            quota_reservation_id,
            expires_at,
            timestamp,
            timestamp,
            run_id,
            customer_id,
            principal_id,
            timestamp,
            MAX_RUN_PROVIDER_CALLS,
            MAX_RUN_DECISIONS,
            CAPABILITY_DESCRIPTOR_DIGEST,
        ),
    )
    if cursor.rowcount != 1:
        raise RuntimeError("execution admission conflict")


def commit_dispatch_tx(
    connection: sqlite3.Connection,
    *,
    execution_id: str,
    run_id: str,
    customer_id: str,
    principal_id: str,
    expected_version: int = 0,
    now: int | None = None,
) -> int:
    """Irreversibly fence one prepared execution before an outbound call."""

    timestamp = _now(now)
    _require_opaque_id(execution_id, "lexe_")
    _require_opaque_id(run_id, "lrun_")
    _require_identity(customer_id, "customer")
    _require_identity(principal_id, "principal")
    cursor = connection.execute(
        """UPDATE labora_model_executions
           SET lease_state='dispatch_committed', provider_dispatched=1,
               version=version+1, updated_at=?
           WHERE execution_id=? AND run_id=? AND customer_id=? AND principal_id=?
             AND lease_state='prepared' AND provider_dispatched=0 AND version=?
             AND EXISTS (SELECT 1 FROM labora_runs r
                 WHERE r.run_id=labora_model_executions.run_id
                   AND r.customer_id=labora_model_executions.customer_id
                   AND r.principal_id=labora_model_executions.principal_id
                   AND r.cancellation_requested=0 AND r.ttl_deadline>?
                   AND r.provider_call_count<? AND r.decision_count<?
                   AND r.public_state NOT IN
                       ('completed','completed_with_limits','cancelled','failed')
                   AND r.metering_finalized=0
                   AND r.internal_state NOT IN
                       ('metering_recorded','metering_recorded_cancelled')
                   AND r.workspace_manifest_digest IS NOT NULL
                   AND r.capability_descriptor_digest=?)""",
        (
            timestamp,
            execution_id,
            run_id,
            customer_id,
            principal_id,
            expected_version,
            timestamp,
            MAX_RUN_PROVIDER_CALLS,
            MAX_RUN_DECISIONS,
            CAPABILITY_DESCRIPTOR_DIGEST,
        ),
    )
    if cursor.rowcount != 1:
        raise RuntimeError("dispatch CAS conflict")
    run_cursor = connection.execute(
        """UPDATE labora_runs
           SET provider_dispatched=1, provider_call_count=provider_call_count+1,
               internal_state='working', public_state='working',
               version=version+1, updated_at=?
           WHERE run_id=? AND customer_id=? AND principal_id=?
             AND cancellation_requested=0 AND ttl_deadline>?
             AND provider_call_count<? AND decision_count<?
             AND public_state NOT IN
                 ('completed','completed_with_limits','cancelled','failed')
             AND metering_finalized=0
             AND internal_state NOT IN
                 ('metering_recorded','metering_recorded_cancelled')""",
        (
            timestamp,
            run_id,
            customer_id,
            principal_id,
            timestamp,
            MAX_RUN_PROVIDER_CALLS,
            MAX_RUN_DECISIONS,
        ),
    )
    if run_cursor.rowcount != 1:
        raise RuntimeError("run changed during dispatch commit")
    return expected_version + 1


def record_provider_return_tx(
    connection: sqlite3.Connection,
    *,
    execution_id: str,
    run_id: str,
    customer_id: str,
    principal_id: str,
    expected_version: int,
    response_digest: str,
    response_digest_kid: str,
    input_units: int,
    output_units: int,
    pricing_provider: str,
    pricing_funding: Literal["paid", "complimentary", "nonbilling"],
    pricing_kind: str,
    cost_microunits: int,
    next_state: str = "provider_returned",
    public_state: str = "reviewing",
    now: int | None = None,
) -> Literal["returned", "cancelled_after_dispatch"]:
    """Persist bounded provider evidence; never persist request or response content."""

    timestamp = _now(now)
    _require_opaque_id(execution_id, "lexe_")
    _require_opaque_id(run_id, "lrun_")
    _require_identity(customer_id, "customer")
    _require_identity(principal_id, "principal")
    require_digest(response_digest)
    _require_key_id(response_digest_kid)
    if (
        type(input_units) is not int
        or type(output_units) is not int
        or not 0 <= input_units <= MAX_USAGE_UNITS
        or not 0 <= output_units <= MAX_USAGE_UNITS
        or type(cost_microunits) is not int
        or not 0 <= cost_microunits <= MAX_COST_MICROUNITS
        or pricing_funding not in {"paid", "complimentary", "nonbilling"}
        or not isinstance(pricing_provider, str)
        or _PRICING_VALUE_RE.fullmatch(pricing_provider) is None
        or not isinstance(pricing_kind, str)
        or _PRICING_VALUE_RE.fullmatch(pricing_kind) is None
        or next_state not in INTERNAL_STATES
        or public_state not in PUBLIC_PHASES
        or next_state in TERMINAL_PHASES
        or public_state in TERMINAL_PHASES
    ):
        raise ValueError("invalid provider return metadata")
    cursor = connection.execute(
        """UPDATE labora_model_executions
           SET lease_state='returned', response_digest=?, response_digest_kid=?,
               input_units=?, output_units=?, pricing_provider=?, pricing_funding=?,
               pricing_kind=?, cost_microunits=?, outcome='returned',
               version=version+1, updated_at=?
           WHERE execution_id=? AND run_id=? AND customer_id=? AND principal_id=?
             AND lease_state='dispatch_committed' AND provider_dispatched=1
             AND version=?
             AND EXISTS (SELECT 1 FROM labora_runs r
                 WHERE r.run_id=labora_model_executions.run_id
                   AND r.customer_id=labora_model_executions.customer_id
                   AND r.principal_id=labora_model_executions.principal_id
                   AND r.public_state NOT IN
                       ('completed','completed_with_limits','cancelled','failed')
                   AND r.metering_finalized=0
                   AND r.internal_state NOT IN
                       ('metering_recorded','metering_recorded_cancelled'))""",
        (
            response_digest,
            response_digest_kid,
            input_units,
            output_units,
            pricing_provider,
            pricing_funding,
            pricing_kind,
            cost_microunits,
            timestamp,
            execution_id,
            run_id,
            customer_id,
            principal_id,
            expected_version,
        ),
    )
    if cursor.rowcount != 1:
        raise RuntimeError("provider return CAS conflict")
    run = connection.execute(
        """SELECT cancellation_requested, version FROM labora_runs
           WHERE run_id=? AND customer_id=? AND principal_id=?""",
        (run_id, customer_id, principal_id),
    ).fetchone()
    if run is None:
        raise RuntimeError("provider return run is unavailable")
    cancelled = bool(run[0])
    outcome = "cancelled_after_dispatch" if cancelled else ""
    run_cursor = connection.execute(
        """UPDATE labora_runs
           SET input_units=input_units+?, output_units=output_units+?,
               internal_state=?, public_state=?, outcome=?,
               decision_count=decision_count+1, version=version+1, updated_at=?
           WHERE run_id=? AND customer_id=? AND principal_id=? AND version=?
             AND decision_count<?
             AND public_state NOT IN
                 ('completed','completed_with_limits','cancelled','failed')
             AND metering_finalized=0
             AND internal_state NOT IN
                 ('metering_recorded','metering_recorded_cancelled')""",
        (
            input_units,
            output_units,
            "metering_pending" if cancelled else next_state,
            "working" if cancelled else public_state,
            outcome,
            timestamp,
            run_id,
            customer_id,
            principal_id,
            int(run[1]),
            MAX_RUN_DECISIONS,
        ),
    )
    if run_cursor.rowcount != 1:
        raise RuntimeError("run changed during provider return")
    return "cancelled_after_dispatch" if cancelled else "returned"


def advance_rolling_chain_tx(
    connection: sqlite3.Connection,
    *,
    run_id: str,
    action_id: str,
    customer_id: str,
    principal_id: str,
    expected_digest: str,
    expected_sequence: int,
    action: Mapping[str, Any],
    result: Mapping[str, Any],
    now: int | None = None,
) -> tuple[str, int]:
    """Fold one validated pair and CAS only its content-free rolling authority."""

    timestamp = _now(now)
    _require_opaque_id(run_id, "lrun_")
    _require_opaque_id(action_id, "lact_")
    _require_identity(customer_id, "customer")
    _require_identity(principal_id, "principal")
    require_digest(expected_digest)
    if (
        type(expected_sequence) is not int
        or not 0 <= expected_sequence < MAX_RUN_ACTIONS
    ):
        raise ValueError("invalid rolling chain sequence")
    next_sequence = expected_sequence + 1
    next_digest = _fold_rolling_chain_item(
        expected_digest,
        next_sequence,
        {"action": action, "result": result},
    )
    cursor = connection.execute(
        """UPDATE labora_runs
           SET rolling_chain_digest=?, rolling_chain_sequence=?,
               version=version+1, updated_at=?
           WHERE run_id=? AND customer_id=? AND principal_id=?
             AND active_action_id=? AND sequence=? AND cancellation_requested=0
             AND public_state NOT IN
                 ('completed','completed_with_limits','cancelled','failed')
             AND metering_finalized=0
             AND internal_state NOT IN
                 ('metering_recorded','metering_recorded_cancelled')
             AND rolling_chain_digest=? AND rolling_chain_sequence=?
             AND workspace_manifest_digest IS NOT NULL
             AND capability_descriptor_digest=?
             AND EXISTS (SELECT 1 FROM labora_actions a
                 WHERE a.action_id=? AND a.run_id=labora_runs.run_id
                   AND a.customer_id=labora_runs.customer_id
                   AND a.principal_id=labora_runs.principal_id
                   AND a.sequence=? AND a.state='result_received')""",
        (
            next_digest,
            next_sequence,
            timestamp,
            run_id,
            customer_id,
            principal_id,
            action_id,
            next_sequence,
            expected_digest,
            expected_sequence,
            CAPABILITY_DESCRIPTOR_DIGEST,
            action_id,
            next_sequence,
        ),
    )
    if cursor.rowcount != 1:
        raise RuntimeError("rolling chain CAS conflict")
    return next_digest, next_sequence


def consume_continuation_tx(
    connection: sqlite3.Connection,
    *,
    action_id: str,
    run_id: str,
    customer_id: str,
    principal_id: str,
    expected_action_version: int,
    expected_run_version: int,
    now: int | None = None,
) -> tuple[int, int]:
    """Consume one receipt and reopen its run only if cancellation has not won."""

    timestamp = _now(now)
    action_version = cas_action_state_tx(
        connection,
        action_id=action_id,
        run_id=run_id,
        customer_id=customer_id,
        principal_id=principal_id,
        expected_state="result_received",
        expected_version=expected_action_version,
        new_state="consumed",
        now=timestamp,
    )
    cursor = connection.execute(
        """UPDATE labora_runs
           SET internal_state='working', public_state='working', active_action_id=NULL,
               version=version+1, updated_at=?
           WHERE run_id=? AND customer_id=? AND principal_id=?
             AND active_action_id=? AND version=? AND cancellation_requested=0
             AND ttl_deadline>?
             AND public_state NOT IN
                 ('completed','completed_with_limits','cancelled','failed')
             AND metering_finalized=0
             AND internal_state NOT IN
                 ('metering_recorded','metering_recorded_cancelled')""",
        (
            timestamp,
            run_id,
            customer_id,
            principal_id,
            action_id,
            expected_run_version,
            timestamp,
        ),
    )
    if cursor.rowcount != 1:
        raise RuntimeError("continuation CAS conflict")
    return action_version, expected_run_version + 1


def request_cancellation_tx(
    connection: sqlite3.Connection,
    *,
    run_id: str,
    customer_id: str,
    principal_id: str,
    now: int | None = None,
) -> Literal["terminal", "cancelled_before_dispatch", "cancelled_after_dispatch"]:
    """Record monotonic cancellation and classify the irreversible effect boundary."""

    timestamp = _now(now)
    _require_opaque_id(run_id, "lrun_")
    _require_identity(customer_id, "customer")
    _require_identity(principal_id, "principal")
    row = connection.execute(
        """SELECT public_state, cancellation_requested, provider_dispatched,
                  metering_finalized, version, internal_state
           FROM labora_runs WHERE run_id=? AND customer_id=? AND principal_id=?""",
        (run_id, customer_id, principal_id),
    ).fetchone()
    if row is None:
        raise RuntimeError("run is unavailable")
    if str(row[0]) in TERMINAL_PHASES:
        return "terminal"
    dispatched = (
        bool(row[2])
        or connection.execute(
            """SELECT EXISTS(SELECT 1 FROM labora_model_executions
           WHERE run_id=? AND customer_id=? AND principal_id=?
             AND lease_state IN ('dispatch_committed','returned'))""",
            (run_id, customer_id, principal_id),
        ).fetchone()[0]
    )
    outcome = "cancelled_after_dispatch" if dispatched else "cancelled_before_dispatch"
    if bool(row[1]) and str(row[5]) != "metering_recorded":
        return outcome
    internal_state = (
        "metering_recorded_cancelled"
        if str(row[5]) in ACCOUNTING_FENCE_STATES
        else "metering_pending"
        if dispatched
        else "cancelled"
    )
    public_state = "working" if dispatched else "cancelled"
    cursor = connection.execute(
        """UPDATE labora_runs
           SET cancellation_requested=1, internal_state=?, public_state=?, outcome=?,
               version=version+1, updated_at=?
           WHERE run_id=? AND customer_id=? AND principal_id=? AND version=?
             AND public_state NOT IN ('completed','completed_with_limits','cancelled','failed')""",
        (
            internal_state,
            public_state,
            outcome,
            timestamp,
            run_id,
            customer_id,
            principal_id,
            int(row[4]),
        ),
    )
    if cursor.rowcount != 1:
        raise RuntimeError("cancellation CAS conflict")
    connection.execute(
        """UPDATE labora_model_executions SET lease_state='abandoned', outcome='cancelled_before_dispatch',
               version=version+1, updated_at=?
           WHERE run_id=? AND customer_id=? AND principal_id=? AND lease_state='prepared'""",
        (timestamp, run_id, customer_id, principal_id),
    )
    connection.execute(
        """UPDATE labora_actions SET state='cancelled', version=version+1, updated_at=?
           WHERE run_id=? AND customer_id=? AND principal_id=? AND state='offered'""",
        (timestamp, run_id, customer_id, principal_id),
    )
    return outcome


def prepare_metering_recorded_tx(
    connection: sqlite3.Connection,
    *,
    run_id: str,
    customer_id: str,
    principal_id: str,
    now: int | None = None,
) -> Literal["recorded", "replay"]:
    """Fence authoritative accounting before releasing the SQLite write lock."""

    timestamp = _now(now)
    _require_opaque_id(run_id, "lrun_")
    _require_identity(customer_id, "customer")
    _require_identity(principal_id, "principal")
    row = connection.execute(
        """SELECT internal_state,public_state,cancellation_requested,
                  metering_finalized,version
           FROM labora_runs
           WHERE run_id=? AND customer_id=? AND principal_id=?""",
        (run_id, customer_id, principal_id),
    ).fetchone()
    if row is None:
        raise RuntimeError("run is unavailable")
    cancelled = bool(row[2])
    target = (
        "metering_recorded_cancelled" if cancelled else "metering_recorded"
    )
    current = str(row[0])
    if bool(row[3]):
        raise RuntimeError("metering marker precedes accounting fence")
    if current in ACCOUNTING_FENCE_STATES:
        if current != target:
            raise RuntimeError("accounting fence cancellation mismatch")
        return "replay"
    if str(row[1]) in TERMINAL_PHASES:
        raise AccountingFenceConflict("terminal run cannot enter accounting fence")

    blocked_actions = (
        "('offered')" if cancelled else "('offered','claimed','result_received')"
    )
    cursor = connection.execute(
        f"""UPDATE labora_runs
            SET internal_state=?,version=version+1,updated_at=?
            WHERE run_id=? AND customer_id=? AND principal_id=? AND version=?
              AND metering_finalized=0
              AND cancellation_requested=?
              AND public_state NOT IN
                  ('completed','completed_with_limits','cancelled','failed')
              AND internal_state NOT IN
                  ('metering_recorded','metering_recorded_cancelled')
              AND NOT EXISTS (
                  SELECT 1 FROM labora_model_executions e
                  WHERE e.run_id=labora_runs.run_id
                    AND e.customer_id=labora_runs.customer_id
                    AND e.principal_id=labora_runs.principal_id
                    AND e.lease_state IN ('prepared','dispatch_committed')
              )
              AND NOT EXISTS (
                  SELECT 1 FROM labora_actions a
                  WHERE a.run_id=labora_runs.run_id
                    AND a.customer_id=labora_runs.customer_id
                    AND a.principal_id=labora_runs.principal_id
                    AND a.state IN {blocked_actions}
              )""",
        (
            target,
            timestamp,
            run_id,
            customer_id,
            principal_id,
            int(row[4]),
            int(cancelled),
        ),
    )
    if cursor.rowcount != 1:
        raise AccountingFenceConflict("accounting authority is still live")
    return "recorded"


def mark_metering_finalized_tx(
    connection: sqlite3.Connection,
    *,
    run_id: str,
    customer_id: str,
    principal_id: str,
    now: int | None = None,
) -> Literal["finalized", "replay"]:
    """Advance only the marker after the durable accounting fence exists."""

    timestamp = _now(now)
    row = connection.execute(
        """SELECT internal_state,cancellation_requested,metering_finalized,
                  public_state
           FROM labora_runs
           WHERE run_id=? AND customer_id=? AND principal_id=?""",
        (run_id, customer_id, principal_id),
    ).fetchone()
    if row is None:
        raise RuntimeError("run is unavailable")
    target = (
        "metering_recorded_cancelled" if bool(row[1]) else "metering_recorded"
    )
    if bool(row[2]):
        public_state = str(row[3])
        prepublished_cancelled = (
            bool(row[1])
            and public_state == "cancelled"
            and str(row[0]) == "metering_recorded_cancelled"
        )
        valid_replay = (
            str(row[0]) == public_state or prepublished_cancelled
            if public_state in TERMINAL_PHASES
            else str(row[0]) == target
        )
        if not valid_replay:
            raise RuntimeError("metering replay lacks exact authority")
        return "replay"
    cancelled_between_fence_and_marker = (
        bool(row[1])
        and str(row[3]) == "cancelled"
        and str(row[0]) == "metering_recorded_cancelled"
    )
    if (
        not cancelled_between_fence_and_marker
        and (str(row[3]) in TERMINAL_PHASES or str(row[0]) != target)
    ):
        raise RuntimeError("metering marker lacks the exact accounting fence")
    blocked_actions = (
        "('offered')"
        if bool(row[1])
        else "('offered','claimed','result_received')"
    )
    cursor = connection.execute(
        f"""UPDATE labora_runs SET metering_finalized=1, updated_at=?
           WHERE run_id=? AND customer_id=? AND principal_id=? AND metering_finalized=0
             AND internal_state=? AND cancellation_requested=?
             AND NOT EXISTS (
                 SELECT 1 FROM labora_actions a
                 WHERE a.run_id=labora_runs.run_id
                   AND a.customer_id=labora_runs.customer_id
                   AND a.principal_id=labora_runs.principal_id
                   AND a.state IN {blocked_actions}
             )
             AND NOT EXISTS (
                 SELECT 1 FROM labora_model_executions unresolved
                 WHERE unresolved.run_id=labora_runs.run_id
                   AND unresolved.customer_id=labora_runs.customer_id
                   AND unresolved.principal_id=labora_runs.principal_id
                   AND unresolved.lease_state IN ('prepared','dispatch_committed')
             )
             AND (provider_dispatched=0 OR EXISTS (
                 SELECT 1 FROM labora_model_executions e
                 WHERE e.run_id=labora_runs.run_id
                   AND e.customer_id=labora_runs.customer_id
                   AND e.principal_id=labora_runs.principal_id
                   AND (e.lease_state='returned' OR (
                       e.lease_state='abandoned'
                       AND e.outcome='provider_outcome_unknown'
                   ))))""",
        (
            timestamp,
            run_id,
            customer_id,
            principal_id,
            target,
            int(bool(row[1])),
        ),
    )
    if cursor.rowcount == 1:
        return "finalized"
    raise RuntimeError("metering finalization conflict")


def publish_terminal_tx(
    connection: sqlite3.Connection,
    *,
    run_id: str,
    customer_id: str,
    principal_id: str,
    phase: Literal["completed", "completed_with_limits", "failed"],
    outcome: str = "",
    now: int | None = None,
) -> str:
    """Publish a terminal phase only after metering and a final cancel check."""

    timestamp = _now(now)
    if phase not in {"completed", "completed_with_limits", "failed"}:
        raise ValueError("invalid terminal phase")
    _require_safe_state(outcome or "none")
    row = connection.execute(
        """SELECT cancellation_requested, provider_dispatched, metering_finalized,
                  public_state, version, internal_state
           FROM labora_runs WHERE run_id=? AND customer_id=? AND principal_id=?""",
        (run_id, customer_id, principal_id),
    ).fetchone()
    if row is None:
        raise RuntimeError("run is unavailable")
    if str(row[3]) in TERMINAL_PHASES:
        if bool(row[1]) and not bool(row[2]):
            raise RuntimeError("persisted terminal phase precedes metering")
        if (
            str(row[3]) == "cancelled"
            and bool(row[0])
            and bool(row[2])
            and str(row[5]) == "metering_recorded_cancelled"
        ):
            cursor = connection.execute(
                """UPDATE labora_runs
                   SET internal_state='cancelled', version=version+1, updated_at=?
                   WHERE run_id=? AND customer_id=? AND principal_id=? AND version=?
                     AND cancellation_requested=1
                     AND metering_finalized=1
                     AND internal_state='metering_recorded_cancelled'
                     AND public_state='cancelled'""",
                (
                    timestamp,
                    run_id,
                    customer_id,
                    principal_id,
                    int(row[4]),
                ),
            )
            if cursor.rowcount != 1:
                raise RuntimeError("terminal cancellation publication conflict")
            return "cancelled"
        if str(row[5]) != str(row[3]):
            raise RuntimeError("persisted terminal phase lacks exact authority")
        return str(row[3])
    if bool(row[0]):
        if bool(row[1]) and not bool(row[2]):
            raise RuntimeError("post-dispatch cancellation is not metered")
        target_phase, target_outcome = (
            "cancelled",
            (
                "cancelled_after_dispatch"
                if bool(row[1])
                else "cancelled_before_dispatch"
            ),
        )
    else:
        if not bool(row[2]):
            raise RuntimeError("terminal publication precedes metering")
        target_phase, target_outcome = phase, outcome
    expected_fence = (
        "metering_recorded_cancelled" if bool(row[0]) else "metering_recorded"
    )
    if str(row[5]) != expected_fence:
        raise RuntimeError("terminal publication lacks the exact accounting fence")
    cursor = connection.execute(
        """UPDATE labora_runs
           SET internal_state=?, public_state=?, outcome=?, version=version+1, updated_at=?
           WHERE run_id=? AND customer_id=? AND principal_id=? AND version=?
             AND metering_finalized=1
             AND internal_state=?
             AND public_state NOT IN ('completed','completed_with_limits','cancelled','failed')""",
        (
            target_phase,
            target_phase,
            target_outcome,
            timestamp,
            run_id,
            customer_id,
            principal_id,
            int(row[4]),
            expected_fence,
        ),
    )
    if cursor.rowcount != 1:
        raise RuntimeError("terminal publication conflict")
    return target_phase


def reconcile_execution_tx(
    connection: sqlite3.Connection,
    *,
    execution_id: str,
    customer_id: str,
    principal_id: str,
    input_units: int = 0,
    output_units: int = 0,
    pricing_provider: str | None = None,
    pricing_funding: Literal["paid", "complimentary", "nonbilling"] | None = None,
    pricing_kind: str | None = None,
    cost_microunits: int | None = None,
    now: int | None = None,
) -> Literal["abandoned", "provider_outcome_unknown", "returned"]:
    """Reconcile a provider lease without publishing an unmetered terminal state."""

    timestamp = _now(now)
    _require_opaque_id(execution_id, "lexe_")
    _require_identity(customer_id, "customer")
    _require_identity(principal_id, "principal")
    if (
        type(input_units) is not int
        or type(output_units) is not int
        or not 0 <= input_units <= MAX_USAGE_UNITS
        or not 0 <= output_units <= MAX_USAGE_UNITS
        or cost_microunits is not None
        and (
            type(cost_microunits) is not int
            or not 0 <= cost_microunits <= MAX_COST_MICROUNITS
        )
    ):
        raise ValueError("invalid recovery usage metadata")
    pricing_values = (pricing_provider, pricing_funding, pricing_kind)
    if any(value is not None for value in pricing_values) and any(
        value is None for value in pricing_values
    ):
        raise ValueError("incomplete recovery pricing metadata")
    if pricing_funding is not None and pricing_funding not in {
        "paid",
        "complimentary",
        "nonbilling",
    }:
        raise ValueError("invalid recovery funding metadata")
    for value in (pricing_provider, pricing_kind):
        if value is not None and _PRICING_VALUE_RE.fullmatch(value) is None:
            raise ValueError("invalid recovery pricing metadata")
    row = connection.execute(
        """SELECT run_id, lease_state, version FROM labora_model_executions
           WHERE execution_id=? AND customer_id=? AND principal_id=?""",
        (execution_id, customer_id, principal_id),
    ).fetchone()
    if row is None:
        raise RuntimeError("execution is unavailable")
    run_id, state, version = str(row[0]), str(row[1]), int(row[2])
    if state == "returned":
        return "returned"
    if state == "abandoned":
        outcome = connection.execute(
            """SELECT outcome FROM labora_model_executions
               WHERE execution_id=? AND customer_id=? AND principal_id=?""",
            (execution_id, customer_id, principal_id),
        ).fetchone()
        return (
            "provider_outcome_unknown"
            if outcome is not None and str(outcome[0]) == "provider_outcome_unknown"
            else "abandoned"
        )
    if state == "prepared":
        if (
            input_units
            or output_units
            or cost_microunits not in {None, 0}
            or any(value is not None for value in pricing_values)
        ):
            raise ValueError("usage cannot be attached before the dispatch fence")
        cursor = connection.execute(
            """UPDATE labora_model_executions
               SET lease_state='abandoned', outcome='abandoned', version=version+1, updated_at=?
               WHERE execution_id=? AND customer_id=? AND principal_id=?
                 AND lease_state='prepared' AND version=?
                 AND EXISTS (SELECT 1 FROM labora_runs r
                     WHERE r.run_id=labora_model_executions.run_id
                       AND r.customer_id=labora_model_executions.customer_id
                       AND r.principal_id=labora_model_executions.principal_id
                       AND r.public_state NOT IN
                           ('completed','completed_with_limits','cancelled','failed')
                       AND r.metering_finalized=0
                       AND r.internal_state NOT IN
                           ('metering_recorded','metering_recorded_cancelled'))""",
            (timestamp, execution_id, customer_id, principal_id, version),
        )
        if cursor.rowcount != 1:
            raise RuntimeError("prepared recovery CAS conflict")
        return "abandoned"
    if state == "dispatch_committed":
        cursor = connection.execute(
            """UPDATE labora_model_executions
               SET lease_state='abandoned', outcome='provider_outcome_unknown',
                   input_units=?, output_units=?, pricing_provider=?,
                   pricing_funding=?, pricing_kind=?, cost_microunits=?,
                   version=version+1, updated_at=?
               WHERE execution_id=? AND customer_id=? AND principal_id=?
                 AND lease_state='dispatch_committed' AND version=?
                 AND EXISTS (SELECT 1 FROM labora_runs r
                     WHERE r.run_id=labora_model_executions.run_id
                       AND r.customer_id=labora_model_executions.customer_id
                       AND r.principal_id=labora_model_executions.principal_id
                       AND r.public_state NOT IN
                           ('completed','completed_with_limits','cancelled','failed')
                       AND r.metering_finalized=0
                       AND r.internal_state NOT IN
                           ('metering_recorded','metering_recorded_cancelled'))""",
            (
                input_units,
                output_units,
                pricing_provider,
                pricing_funding,
                pricing_kind,
                cost_microunits,
                timestamp,
                execution_id,
                customer_id,
                principal_id,
                version,
            ),
        )
        if cursor.rowcount != 1:
            raise RuntimeError("dispatch recovery CAS conflict")
        run_cursor = connection.execute(
            """UPDATE labora_runs
               SET internal_state='provider_outcome_unknown',
                   public_state='working', outcome='provider_outcome_unknown',
                   input_units=input_units+?, output_units=output_units+?,
                   decision_count=decision_count+1,
                   version=version+1, updated_at=?
               WHERE run_id=? AND customer_id=? AND principal_id=?
                 AND decision_count<?
                 AND public_state NOT IN ('completed','completed_with_limits','cancelled','failed')
                 AND metering_finalized=0
                 AND internal_state NOT IN
                     ('metering_recorded','metering_recorded_cancelled')""",
            (
                input_units,
                output_units,
                timestamp,
                run_id,
                customer_id,
                principal_id,
                MAX_RUN_DECISIONS,
            ),
        )
        if run_cursor.rowcount != 1:
            raise RuntimeError("run changed during dispatch recovery")
        return "provider_outcome_unknown"
    return "abandoned"


def delete_expired_bridge_metadata(
    connection: sqlite3.Connection, *, now: int | None = None
) -> int:
    """Delete only resolved expired roots; unresolved effects retain replay fences."""

    timestamp = _now(now)

    def cleanup() -> int:
        total = 0
        prepared = connection.execute(
            """UPDATE labora_model_executions
               SET lease_state='abandoned', outcome='expired_before_dispatch',
                   version=version+1, updated_at=?
               WHERE expires_at<=? AND lease_state='prepared'
                 AND EXISTS (SELECT 1 FROM labora_runs r
                     WHERE r.run_id=labora_model_executions.run_id
                       AND r.customer_id=labora_model_executions.customer_id
                       AND r.principal_id=labora_model_executions.principal_id
                       AND r.metering_finalized=0
                       AND r.internal_state NOT IN
                           ('metering_recorded','metering_recorded_cancelled')
                       AND r.public_state NOT IN
                           ('completed','completed_with_limits','cancelled','failed'))""",
            (timestamp, timestamp),
        )
        total += prepared.rowcount
        connection.execute(
            """UPDATE labora_runs
               SET internal_state='failed', public_state='failed',
                   outcome='expired_before_dispatch', version=version+1, updated_at=?
               WHERE expires_at<=? AND provider_dispatched=0
                 AND metering_finalized=0
                 AND internal_state NOT IN
                     ('metering_recorded','metering_recorded_cancelled')
                 AND public_state NOT IN ('completed','completed_with_limits','cancelled','failed')
                 AND EXISTS (SELECT 1 FROM labora_model_executions e
                     WHERE e.run_id=labora_runs.run_id
                       AND e.customer_id=labora_runs.customer_id
                       AND e.principal_id=labora_runs.principal_id
                       AND e.outcome='expired_before_dispatch')""",
            (timestamp, timestamp),
        )
        connection.execute(
            """UPDATE labora_model_executions
               SET lease_state='abandoned', outcome='provider_outcome_unknown',
                   cost_microunits=COALESCE(cost_microunits,0),
                   version=version+1, updated_at=?
               WHERE expires_at<=? AND lease_state='dispatch_committed'
                 AND EXISTS (SELECT 1 FROM labora_runs r
                     WHERE r.run_id=labora_model_executions.run_id
                       AND r.customer_id=labora_model_executions.customer_id
                       AND r.principal_id=labora_model_executions.principal_id
                       AND r.metering_finalized=0
                       AND r.internal_state NOT IN
                           ('metering_recorded','metering_recorded_cancelled')
                       AND r.public_state NOT IN
                           ('completed','completed_with_limits','cancelled','failed'))""",
            (timestamp, timestamp),
        )
        connection.execute(
            """UPDATE labora_runs
               SET internal_state='provider_outcome_unknown',
                   public_state='working', outcome='provider_outcome_unknown',
                   version=version+1, updated_at=?
               WHERE expires_at<=? AND public_state NOT IN
                   ('completed','completed_with_limits','cancelled','failed')
                 AND metering_finalized=0
                 AND internal_state NOT IN
                     ('metering_recorded','metering_recorded_cancelled')
                 AND NOT (
                     internal_state='provider_outcome_unknown'
                     AND public_state='working'
                     AND outcome='provider_outcome_unknown'
                 )
                 AND EXISTS (SELECT 1 FROM labora_model_executions e
                     WHERE e.run_id=labora_runs.run_id
                       AND e.customer_id=labora_runs.customer_id
                       AND e.principal_id=labora_runs.principal_id
                       AND e.outcome='provider_outcome_unknown')""",
            (timestamp, timestamp),
        )
        idem = connection.execute(
            """DELETE FROM labora_idempotency
               WHERE expires_at<=? AND (
                   resource_id IS NULL OR resource_id='' OR
                   NOT EXISTS (SELECT 1 FROM labora_runs r
                       WHERE r.run_id=resource_id OR r.run_id=related_id) OR
                   EXISTS (SELECT 1 FROM labora_runs r
                       WHERE (r.run_id=resource_id OR r.run_id=related_id)
                         AND r.public_state IN ('completed','completed_with_limits','cancelled','failed')
                         AND (r.metering_finalized=1 OR r.provider_dispatched=0))
               )""",
            (timestamp,),
        )
        total += idem.rowcount
        runs = connection.execute(
            """DELETE FROM labora_runs
               WHERE expires_at<=?
                 AND public_state IN ('completed','completed_with_limits','cancelled','failed')
                 AND (metering_finalized=1 OR provider_dispatched=0)
                 AND NOT EXISTS (SELECT 1 FROM labora_idempotency i
                     WHERE i.customer_id=labora_runs.customer_id
                       AND i.principal_id=labora_runs.principal_id
                       AND (i.resource_id=labora_runs.run_id OR i.related_id=labora_runs.run_id))""",
            (timestamp,),
        )
        total += runs.rowcount
        return total

    return _run_immediate(connection, cleanup)


def erase_customer_bridge_metadata(
    connection: sqlite3.Connection, customer_id: str
) -> int:
    """Immediately erase all bridge metadata for one exact customer."""

    _require_identity(customer_id, "customer")

    def erase() -> int:
        total = 0
        # Children are explicit so the returned count is useful even with cascades.
        for table in (
            "labora_idempotency",
            "labora_model_executions",
            "labora_actions",
            "labora_runs",
        ):
            cursor = connection.execute(
                f"DELETE FROM {table} WHERE customer_id=?", (customer_id,)
            )
            total += cursor.rowcount
        return total

    return _run_immediate(connection, erase)