DataOps-platform/app/core/edge_gateway/queue.py

"""Durable SQLite task and outbound-event queue for an offline edge gateway."""

from __future__ import annotations

import hashlib
import json
import re
import sqlite3
import uuid
from collections.abc import Callable, Iterator, Mapping
from contextlib import contextmanager
from dataclasses import dataclass
from datetime import UTC, datetime, timedelta
from pathlib import Path
from types import MappingProxyType

from .contracts import (
    EdgeContractError,
    EdgeEventContract,
    EdgeTaskContract,
    SignedTaskEnvelope,
    canonical_sha256,
    canonical_timestamp,
    strict_json_bytes,
)
from .policy import EdgeEgressPolicy, EdgePolicyError


class EdgeQueueError(RuntimeError):
    """Base error for durable edge queue operations."""


class EdgeQueueConflictError(EdgeQueueError):
    """Raised when an existing stable ID is replayed with different content."""


class EdgeQueueLeaseError(EdgeQueueError):
    """Raised when a mutation is attempted without the current lease token."""


class EdgeQueueSchemaError(EdgeQueueError):
    """Raised when an edge queue database schema cannot be trusted."""


SCHEMA_VERSION = 2

_V1_TASK_TABLE_SQL = """CREATE TABLE edge_tasks (
    task_id TEXT PRIMARY KEY,
    digest TEXT NOT NULL,
    task_json TEXT NOT NULL,
    status TEXT NOT NULL CHECK (
        status IN ('pending', 'leased', 'completed', 'failed', 'cancelled')
    ),
    attempt_count INTEGER NOT NULL DEFAULT 0 CHECK (attempt_count >= 0),
    available_at TEXT NOT NULL,
    deadline_at TEXT NOT NULL,
    deadline_epoch_us INTEGER NOT NULL,
    lease_owner TEXT,
    lease_token TEXT,
    lease_expires_at TEXT,
    error_code TEXT,
    created_at TEXT NOT NULL,
    updated_at TEXT NOT NULL
)"""
_TASK_TABLE_SQL = _V1_TASK_TABLE_SQL.rstrip()[:-1].rstrip() + ",\n" + (
    "    authority_key_id TEXT,\n"
    "    authority_digest TEXT,\n"
    "    authority_json TEXT,\n"
    "    remote_lease_token_digest TEXT,\n"
    "    remote_lease_ciphertext TEXT,\n"
    "    remote_lease_expires_at TEXT,\n"
    "    remote_attempt INTEGER CHECK (remote_attempt IS NULL OR remote_attempt >= 1)\n"
    ")"
)
_TASK_INDEX_SQL = """CREATE INDEX edge_tasks_claim_idx ON edge_tasks(
    status, deadline_epoch_us, available_at, lease_expires_at, created_at
)"""
_V1_EVENT_TABLE_SQL = """CREATE TABLE edge_outbound_events (
    event_id TEXT PRIMARY KEY,
    digest TEXT NOT NULL,
    event_json TEXT NOT NULL,
    status TEXT NOT NULL CHECK (
        status IN ('pending', 'sending', 'acknowledged', 'failed')
    ),
    attempt_count INTEGER NOT NULL DEFAULT 0 CHECK (attempt_count >= 0),
    available_at TEXT NOT NULL,
    lease_owner TEXT,
    lease_token TEXT,
    lease_expires_at TEXT,
    error_code TEXT,
    acknowledgement_json TEXT,
    ack_digest TEXT,
    lease_token_digest TEXT,
    created_at TEXT NOT NULL,
    updated_at TEXT NOT NULL
)"""
_EVENT_TABLE_SQL = _V1_EVENT_TABLE_SQL.rstrip()[:-1].rstrip() + ",\n" + (
    "    task_digest TEXT,\n"
    "    remote_lease_token_digest TEXT,\n"
    "    control_lease_token_digest TEXT\n"
    ")"
)
_EVENT_INDEX_SQL = """CREATE INDEX edge_events_claim_idx ON edge_outbound_events(
    status, available_at, lease_expires_at, created_at
)"""
_ARTIFACT_TABLE_SQL = """CREATE TABLE edge_local_artifacts (
    artifact_digest TEXT PRIMARY KEY,
    artifact_ref TEXT,
    artifact_ref_hash TEXT NOT NULL,
    classification TEXT NOT NULL CHECK (
        classification IN ('raw', 'recent_detail', 'restricted',
                           'desensitized_metadata', 'statistics', 'lineage', 'evidence')
    ),
    retention_until TEXT NOT NULL,
    task_id TEXT NOT NULL,
    event_id TEXT NOT NULL,
    cleanup_status TEXT NOT NULL CHECK (
        cleanup_status IN ('pending', 'deleting', 'deleted', 'failed')
    ),
    attempt_count INTEGER NOT NULL DEFAULT 0 CHECK (attempt_count >= 0),
    available_at TEXT NOT NULL,
    lease_owner TEXT,
    lease_token TEXT,
    lease_expires_at TEXT,
    error_code TEXT,
    deletion_receipt_json TEXT,
    created_at TEXT NOT NULL,
    updated_at TEXT NOT NULL,
    FOREIGN KEY(task_id) REFERENCES edge_tasks(task_id),
    FOREIGN KEY(event_id) REFERENCES edge_outbound_events(event_id)
)"""
_ARTIFACT_INDEX_SQL = """CREATE INDEX edge_artifacts_task_idx ON edge_local_artifacts(
    cleanup_status, retention_until, available_at, lease_expires_at, created_at
)"""
_RELEASE_TABLE_SQL = """CREATE TABLE edge_release_state (
    release_id TEXT PRIMARY KEY,
    manifest_digest TEXT NOT NULL,
    version TEXT NOT NULL,
    rollback_version TEXT NOT NULL,
    status TEXT NOT NULL CHECK (
        status IN ('offered', 'accepted', 'candidate', 'installed', 'failed', 'rolled_back')
    ),
    previous_version TEXT,
    current_version TEXT,
    updated_at TEXT NOT NULL
)"""
_RELEASE_INDEX_SQL = """CREATE INDEX edge_release_status_idx ON edge_release_state(
    status, updated_at, release_id
)"""
_OUTCOME_TABLE_SQL = """CREATE TABLE edge_task_outcomes (
    task_id TEXT PRIMARY KEY,
    digest TEXT NOT NULL,
    outcome_json TEXT NOT NULL,
    status TEXT NOT NULL CHECK (
        status IN ('pending', 'sending', 'acknowledged', 'failed')
    ),
    attempt_count INTEGER NOT NULL DEFAULT 0 CHECK (attempt_count >= 0),
    available_at TEXT NOT NULL,
    lease_owner TEXT,
    lease_token TEXT,
    lease_expires_at TEXT,
    error_code TEXT,
    acknowledgement_json TEXT,
    ack_digest TEXT,
    remote_lease_token_digest TEXT NOT NULL,
    created_at TEXT NOT NULL,
    updated_at TEXT NOT NULL,
    FOREIGN KEY(task_id) REFERENCES edge_tasks(task_id)
)"""
_OUTCOME_INDEX_SQL = """CREATE INDEX edge_outcomes_claim_idx ON edge_task_outcomes(
    status, available_at, lease_expires_at, created_at
)"""
_EXPECTED_TABLE_INFO = {
    "edge_tasks": (
        ("task_id", "TEXT", 0, None, 1),
        ("digest", "TEXT", 1, None, 0),
        ("task_json", "TEXT", 1, None, 0),
        ("status", "TEXT", 1, None, 0),
        ("attempt_count", "INTEGER", 1, "0", 0),
        ("available_at", "TEXT", 1, None, 0),
        ("deadline_at", "TEXT", 1, None, 0),
        ("deadline_epoch_us", "INTEGER", 1, None, 0),
        ("lease_owner", "TEXT", 0, None, 0),
        ("lease_token", "TEXT", 0, None, 0),
        ("lease_expires_at", "TEXT", 0, None, 0),
        ("error_code", "TEXT", 0, None, 0),
        ("created_at", "TEXT", 1, None, 0),
        ("updated_at", "TEXT", 1, None, 0),
        ("authority_key_id", "TEXT", 0, None, 0),
        ("authority_digest", "TEXT", 0, None, 0),
        ("authority_json", "TEXT", 0, None, 0),
        ("remote_lease_token_digest", "TEXT", 0, None, 0),
        ("remote_lease_ciphertext", "TEXT", 0, None, 0),
        ("remote_lease_expires_at", "TEXT", 0, None, 0),
        ("remote_attempt", "INTEGER", 0, None, 0),
    ),
    "edge_outbound_events": (
        ("event_id", "TEXT", 0, None, 1),
        ("digest", "TEXT", 1, None, 0),
        ("event_json", "TEXT", 1, None, 0),
        ("status", "TEXT", 1, None, 0),
        ("attempt_count", "INTEGER", 1, "0", 0),
        ("available_at", "TEXT", 1, None, 0),
        ("lease_owner", "TEXT", 0, None, 0),
        ("lease_token", "TEXT", 0, None, 0),
        ("lease_expires_at", "TEXT", 0, None, 0),
        ("error_code", "TEXT", 0, None, 0),
        ("acknowledgement_json", "TEXT", 0, None, 0),
        ("ack_digest", "TEXT", 0, None, 0),
        ("lease_token_digest", "TEXT", 0, None, 0),
        ("created_at", "TEXT", 1, None, 0),
        ("updated_at", "TEXT", 1, None, 0),
        ("task_digest", "TEXT", 0, None, 0),
        ("remote_lease_token_digest", "TEXT", 0, None, 0),
        ("control_lease_token_digest", "TEXT", 0, None, 0),
    ),
    "edge_local_artifacts": (
        ("artifact_digest", "TEXT", 0, None, 1),
        ("artifact_ref", "TEXT", 0, None, 0),
        ("artifact_ref_hash", "TEXT", 1, None, 0),
        ("classification", "TEXT", 1, None, 0),
        ("retention_until", "TEXT", 1, None, 0),
        ("task_id", "TEXT", 1, None, 0),
        ("event_id", "TEXT", 1, None, 0),
        ("cleanup_status", "TEXT", 1, None, 0),
        ("attempt_count", "INTEGER", 1, "0", 0),
        ("available_at", "TEXT", 1, None, 0),
        ("lease_owner", "TEXT", 0, None, 0),
        ("lease_token", "TEXT", 0, None, 0),
        ("lease_expires_at", "TEXT", 0, None, 0),
        ("error_code", "TEXT", 0, None, 0),
        ("deletion_receipt_json", "TEXT", 0, None, 0),
        ("created_at", "TEXT", 1, None, 0),
        ("updated_at", "TEXT", 1, None, 0),
    ),
    "edge_release_state": (
        ("release_id", "TEXT", 0, None, 1),
        ("manifest_digest", "TEXT", 1, None, 0),
        ("version", "TEXT", 1, None, 0),
        ("rollback_version", "TEXT", 1, None, 0),
        ("status", "TEXT", 1, None, 0),
        ("previous_version", "TEXT", 0, None, 0),
        ("current_version", "TEXT", 0, None, 0),
        ("updated_at", "TEXT", 1, None, 0),
    ),
    "edge_task_outcomes": (
        ("task_id", "TEXT", 0, None, 1),
        ("digest", "TEXT", 1, None, 0),
        ("outcome_json", "TEXT", 1, None, 0),
        ("status", "TEXT", 1, None, 0),
        ("attempt_count", "INTEGER", 1, "0", 0),
        ("available_at", "TEXT", 1, None, 0),
        ("lease_owner", "TEXT", 0, None, 0),
        ("lease_token", "TEXT", 0, None, 0),
        ("lease_expires_at", "TEXT", 0, None, 0),
        ("error_code", "TEXT", 0, None, 0),
        ("acknowledgement_json", "TEXT", 0, None, 0),
        ("ack_digest", "TEXT", 0, None, 0),
        ("remote_lease_token_digest", "TEXT", 1, None, 0),
        ("created_at", "TEXT", 1, None, 0),
        ("updated_at", "TEXT", 1, None, 0),
    ),
}
_EXPECTED_INDEXES = {
    "edge_tasks": {
        "edge_tasks_claim_idx": (
            0,
            "c",
            0,
            ("status", "deadline_epoch_us", "available_at", "lease_expires_at", "created_at"),
        ),
    },
    "edge_outbound_events": {
        "edge_events_claim_idx": (
            0,
            "c",
            0,
            ("status", "available_at", "lease_expires_at", "created_at"),
        ),
    },
    "edge_local_artifacts": {
        "edge_artifacts_task_idx": (
            0, "c", 0,
            ("cleanup_status", "retention_until", "available_at", "lease_expires_at", "created_at"),
        ),
    },
    "edge_release_state": {
        "edge_release_status_idx": (0, "c", 0, ("status", "updated_at", "release_id")),
    },
    "edge_task_outcomes": {
        "edge_outcomes_claim_idx": (
            0, "c", 0, ("status", "available_at", "lease_expires_at", "created_at")
        ),
    },
}
_EXPECTED_TABLE_SQL = {
    "edge_tasks": _TASK_TABLE_SQL,
    "edge_outbound_events": _EVENT_TABLE_SQL,
    "edge_local_artifacts": _ARTIFACT_TABLE_SQL,
    "edge_release_state": _RELEASE_TABLE_SQL,
    "edge_task_outcomes": _OUTCOME_TABLE_SQL,
}

_V1_EXPECTED_TABLE_INFO = {
    table: tuple(row for row in info if row[0] not in {
        "authority_key_id", "authority_digest", "authority_json",
        "remote_lease_token_digest", "remote_lease_ciphertext", "remote_lease_expires_at", "remote_attempt",
        "task_digest", "control_lease_token_digest",
    })
    for table, info in _EXPECTED_TABLE_INFO.items()
    if table not in {"edge_local_artifacts", "edge_release_state", "edge_task_outcomes"}
}
_V1_EXPECTED_TABLE_SQL = {
    "edge_tasks": _V1_TASK_TABLE_SQL,
    "edge_outbound_events": _V1_EVENT_TABLE_SQL,
}


@dataclass(frozen=True, slots=True)
class QueuedTask:
    task_id: str
    digest: str
    status: str
    task: Mapping[str, object]
    attempt_count: int
    available_at: str
    lease_owner: str | None
    lease_token: str | None
    lease_expires_at: str | None
    error_code: str | None
    authority_key_id: str | None
    authority_digest: str | None
    remote_lease_token_digest: str | None
    remote_lease_expires_at: str | None
    remote_attempt: int | None
    created_at: str
    updated_at: str


@dataclass(frozen=True, slots=True)
class QueuedEvent:
    event_id: str
    digest: str
    status: str
    event: Mapping[str, object]
    attempt_count: int
    available_at: str
    lease_owner: str | None
    lease_token: str | None
    lease_expires_at: str | None
    error_code: str | None
    acknowledgement: Mapping[str, object] | None
    task_digest: str | None
    remote_lease_token_digest: str | None
    control_lease_token_digest: str | None
    created_at: str
    updated_at: str


@dataclass(frozen=True, slots=True)
class LocalArtifact:
    artifact_digest: str
    artifact_ref: str | None
    artifact_ref_hash: str
    classification: str
    retention_until: str
    task_id: str
    event_id: str
    cleanup_status: str
    attempt_count: int
    available_at: str
    lease_owner: str | None
    lease_token: str | None
    lease_expires_at: str | None
    error_code: str | None
    deletion_receipt: Mapping[str, object] | None
    created_at: str
    updated_at: str


@dataclass(frozen=True, slots=True)
class ReleaseState:
    release_id: str
    manifest_digest: str
    version: str
    rollback_version: str
    status: str
    previous_version: str | None
    current_version: str | None
    updated_at: str


@dataclass(frozen=True, slots=True)
class QueuedOutcome:
    task_id: str
    digest: str
    status: str
    outcome: Mapping[str, object]
    attempt_count: int
    available_at: str
    lease_owner: str | None
    lease_token: str | None
    lease_expires_at: str | None
    error_code: str | None
    acknowledgement: Mapping[str, object] | None
    remote_lease_token_digest: str
    created_at: str
    updated_at: str


_ERROR_CODE = re.compile(r"^[a-z][a-z0-9_.-]{0,127}$")
_SENSITIVE_ERROR_PARTS = ("password", "passwd", "secret", "token", "credential")


def _utc_now(clock: Callable[[], object]) -> datetime:
    value = clock()
    if isinstance(value, bool):
        raise ValueError("queue clock returned an invalid value")
    if isinstance(value, (int, float)):
        value = datetime.fromtimestamp(value, tz=UTC)
    if not isinstance(value, datetime):
        raise ValueError("queue clock must return a datetime or timestamp")
    if value.tzinfo is None or value.utcoffset() is None:
        raise ValueError("queue clock must return a timezone-aware value")
    return value.astimezone(UTC)


def _timestamp(value: datetime) -> str:
    utc = value.astimezone(UTC)
    canonical = utc.strftime("%Y-%m-%dT%H:%M:%S")
    if utc.microsecond:
        canonical += f".{utc.microsecond:06d}".rstrip("0")
    return f"{canonical}Z"


def _epoch_microseconds(value: datetime) -> int:
    epoch = datetime(1970, 1, 1, tzinfo=UTC)
    delta = value.astimezone(UTC) - epoch
    return (
        (delta.days * 86_400 + delta.seconds) * 1_000_000
        + delta.microseconds
    )


def _normalize_schema_sql(value: str | None) -> str:
    if not isinstance(value, str):
        return ""
    normalized = re.sub(r"\s+", " ", value.strip()).lower()
    return re.sub(r"\s*([(),])\s*", r"\1", normalized)


def _freeze_mapping(value: Mapping[str, object]) -> Mapping[str, object]:
    def freeze(item: object) -> object:
        if isinstance(item, dict):
            return MappingProxyType({key: freeze(child) for key, child in item.items()})
        if isinstance(item, list):
            return tuple(freeze(child) for child in item)
        return item

    return freeze(dict(value))  # type: ignore[return-value]


class SqliteEdgeQueue:
    """An on-disk, WAL-backed queue with fenced leases and exact replay."""

    def __init__(
        self,
        db_path: str | Path,
        *,
        clock: Callable[[], object] | None = None,
        lease_seconds: int = 60,
        max_attempts: int = 5,
        retry_base_seconds: int = 2,
        retry_max_seconds: int = 300,
        busy_timeout_ms: int = 5_000,
        egress_policy: EdgeEgressPolicy | None = None,
        random_source: Callable[[], float] | None = None,
        remote_lease_encoder: Callable[[str], str] | None = None,
        remote_lease_decoder: Callable[[str], str] | None = None,
    ) -> None:
        if not isinstance(db_path, (str, Path)) or not str(db_path).strip():
            raise ValueError("an explicit SQLite database path is required")
        path_text = str(db_path)
        if path_text == ":memory:" or "mode=memory" in path_text:
            raise ValueError("an in-memory database is not allowed")
        for label, value in {
            "lease_seconds": lease_seconds,
            "max_attempts": max_attempts,
            "retry_base_seconds": retry_base_seconds,
            "retry_max_seconds": retry_max_seconds,
            "busy_timeout_ms": busy_timeout_ms,
        }.items():
            if isinstance(value, bool) or not isinstance(value, int) or value < 1:
                raise ValueError(f"{label} must be a positive integer")
        if retry_base_seconds > retry_max_seconds:
            raise ValueError("retry_base_seconds cannot exceed retry_max_seconds")
        self.db_path = str(Path(db_path).expanduser().resolve())
        self._clock = clock or (lambda: datetime.now(UTC))
        self.lease_seconds = lease_seconds
        self.max_attempts = max_attempts
        self.retry_base_seconds = retry_base_seconds
        self.retry_max_seconds = retry_max_seconds
        self.busy_timeout_ms = busy_timeout_ms
        self.egress_policy = egress_policy or EdgeEgressPolicy(
            allowed_control_hosts=set()
        )
        # The injectable source makes retry timing testable; 0.5 is neutral jitter.
        self._random = random_source or (lambda: 0.5)
        if (remote_lease_encoder is None) != (remote_lease_decoder is None):
            raise ValueError("remote lease encoder and decoder must be paired")
        self._remote_lease_encoder = remote_lease_encoder
        self._remote_lease_decoder = remote_lease_decoder
        Path(self.db_path).parent.mkdir(parents=True, exist_ok=True)
        self._initialize()

    def _connect(self) -> sqlite3.Connection:
        connection = sqlite3.connect(
            self.db_path,
            timeout=self.busy_timeout_ms / 1_000,
            isolation_level=None,
        )
        connection.row_factory = sqlite3.Row
        connection.execute(f"PRAGMA busy_timeout = {self.busy_timeout_ms}")
        connection.execute("PRAGMA foreign_keys = ON")
        return connection

    def _initialize(self) -> None:
        connection = self._connect()
        try:
            version = connection.execute("PRAGMA user_version").fetchone()[0]
            edge_tables = {
                row[0]
                for row in connection.execute(
                    """
                    SELECT name FROM sqlite_master
                    WHERE type = 'table' AND name LIKE 'edge_%'
                    """
                ).fetchall()
            }
            if version == 0 and edge_tables:
                raise EdgeQueueSchemaError("unversioned edge queue schema is unsafe")
            if version not in {0, 1, SCHEMA_VERSION}:
                raise EdgeQueueSchemaError("edge queue schema version is unsupported")
            connection.execute("PRAGMA journal_mode = WAL")
            connection.execute("PRAGMA synchronous = FULL")
            connection.execute("BEGIN IMMEDIATE")
            if version == 0:
                for statement in (
                    _TASK_TABLE_SQL,
                    _TASK_INDEX_SQL,
                    _EVENT_TABLE_SQL,
                    _EVENT_INDEX_SQL,
                    _ARTIFACT_TABLE_SQL,
                    _ARTIFACT_INDEX_SQL,
                    _RELEASE_TABLE_SQL,
                    _RELEASE_INDEX_SQL,
                    _OUTCOME_TABLE_SQL,
                    _OUTCOME_INDEX_SQL,
                ):
                    connection.execute(statement)
                connection.execute(f"PRAGMA user_version = {SCHEMA_VERSION}")
            elif version == 1:
                self._validate_schema(
                    connection,
                    expected_info=_V1_EXPECTED_TABLE_INFO,
                    expected_sql=_V1_EXPECTED_TABLE_SQL,
                    expected_indexes={
                        key: _EXPECTED_INDEXES[key]
                        for key in ("edge_tasks", "edge_outbound_events")
                    },
                )
                active_unsigned = connection.execute(
                    """SELECT EXISTS(
                    SELECT 1 FROM edge_tasks WHERE status IN ('pending','leased')
                    )"""
                ).fetchone()[0]
                unacknowledged_events = connection.execute(
                    """SELECT EXISTS(
                    SELECT 1 FROM edge_outbound_events WHERE status <> 'acknowledged'
                    )"""
                ).fetchone()[0]
                if active_unsigned or unacknowledged_events:
                    raise EdgeQueueSchemaError(
                        "v1 edge queue must drain active tasks and unacknowledged events"
                    )
                for statement in (
                    "ALTER TABLE edge_tasks ADD COLUMN authority_key_id TEXT",
                    "ALTER TABLE edge_tasks ADD COLUMN authority_digest TEXT",
                    "ALTER TABLE edge_tasks ADD COLUMN authority_json TEXT",
                    "ALTER TABLE edge_tasks ADD COLUMN remote_lease_token_digest TEXT",
                    "ALTER TABLE edge_tasks ADD COLUMN remote_lease_ciphertext TEXT",
                    "ALTER TABLE edge_tasks ADD COLUMN remote_lease_expires_at TEXT",
                    "ALTER TABLE edge_tasks ADD COLUMN remote_attempt INTEGER CHECK (remote_attempt IS NULL OR remote_attempt >= 1)",
                    "ALTER TABLE edge_outbound_events ADD COLUMN task_digest TEXT",
                    "ALTER TABLE edge_outbound_events ADD COLUMN remote_lease_token_digest TEXT",
                    "ALTER TABLE edge_outbound_events ADD COLUMN control_lease_token_digest TEXT",
                    _ARTIFACT_TABLE_SQL,
                    _ARTIFACT_INDEX_SQL,
                    _RELEASE_TABLE_SQL,
                    _RELEASE_INDEX_SQL,
                    _OUTCOME_TABLE_SQL,
                    _OUTCOME_INDEX_SQL,
                ):
                    connection.execute(statement)
                connection.execute(f"PRAGMA user_version = {SCHEMA_VERSION}")
            self._validate_schema(connection)
            if connection.in_transaction:
                connection.execute("COMMIT")
        except Exception:
            if connection.in_transaction:
                connection.execute("ROLLBACK")
            raise
        finally:
            connection.close()

    @staticmethod
    def _validate_schema(
        connection: sqlite3.Connection,
        *,
        expected_info: Mapping[str, tuple] = _EXPECTED_TABLE_INFO,
        expected_sql: Mapping[str, str] = _EXPECTED_TABLE_SQL,
        expected_indexes: Mapping[str, Mapping[str, tuple]] = _EXPECTED_INDEXES,
    ) -> None:
        primary_keys = {
            "edge_tasks": "task_id",
            "edge_outbound_events": "event_id",
            "edge_local_artifacts": "artifact_digest",
            "edge_release_state": "release_id",
            "edge_task_outcomes": "task_id",
        }
        for table, table_expected_info in expected_info.items():
            rows = connection.execute(f"PRAGMA table_info({table})").fetchall()
            actual_info = tuple(
                (row[1], row[2].upper(), row[3], row[4], row[5]) for row in rows
            )
            if actual_info != table_expected_info:
                raise EdgeQueueSchemaError("edge queue schema columns are invalid")
            table_sql = connection.execute(
                "SELECT sql FROM sqlite_master WHERE type = 'table' AND name = ?",
                (table,),
            ).fetchone()
            if table_sql is None or _normalize_schema_sql(
                table_sql[0]
            ) != _normalize_schema_sql(expected_sql[table]):
                raise EdgeQueueSchemaError(
                    "edge queue schema constraints are invalid"
                )

            index_rows = connection.execute(f"PRAGMA index_list({table})").fetchall()
            custom_rows = {row[1]: row for row in index_rows if row[3] == "c"}
            table_expected_indexes = expected_indexes[table]
            if set(custom_rows) != set(table_expected_indexes):
                raise EdgeQueueSchemaError("edge queue schema indexes are invalid")
            for name, (unique, origin, partial, columns) in table_expected_indexes.items():
                row = custom_rows[name]
                actual_columns = tuple(
                    item[2]
                    for item in connection.execute(
                        f"PRAGMA index_info({name})"
                    ).fetchall()
                )
                if (row[2], row[3], row[4], actual_columns) != (
                    unique,
                    origin,
                    partial,
                    columns,
                ):
                    raise EdgeQueueSchemaError(
                        "edge queue schema index definition is invalid"
                    )
            primary_indexes = [row for row in index_rows if row[3] == "pk"]
            if len(primary_indexes) != 1:
                raise EdgeQueueSchemaError("edge queue schema primary key is invalid")
            primary = primary_indexes[0]
            primary_columns = tuple(
                item[2]
                for item in connection.execute(
                    f"PRAGMA index_info({primary[1]})"
                ).fetchall()
            )
            if (primary[2], primary[4], primary_columns) != (
                1,
                0,
                (primary_keys[table],),
            ):
                raise EdgeQueueSchemaError("edge queue schema primary key is invalid")

    @contextmanager
    def _transaction(self) -> Iterator[sqlite3.Connection]:
        connection = self._connect()
        try:
            connection.execute("BEGIN IMMEDIATE")
            yield connection
            connection.execute("COMMIT")
        except Exception:
            if connection.in_transaction:
                connection.execute("ROLLBACK")
            raise
        finally:
            connection.close()

    def enqueue(self, task: EdgeTaskContract | Mapping[str, object]) -> QueuedTask:
        contract = EdgeTaskContract.from_mapping(
            task.to_mapping() if isinstance(task, EdgeTaskContract) else task
        )
        return self._enqueue_contract(contract, authority=None)

    def enqueue_signed(self, envelope: SignedTaskEnvelope) -> QueuedTask:
        """Atomically persist a previously verified authority envelope and task."""

        if not isinstance(envelope, SignedTaskEnvelope) or not envelope.is_verified:
            raise EdgeContractError("a verified SignedTaskEnvelope is required")
        # Reconstruct the task so object.__setattr__ cannot bypass validation.
        contract = EdgeTaskContract.from_mapping(envelope.task.to_mapping())
        if envelope.contract_digest != contract.digest:
            raise EdgeContractError("signed authority no longer binds the task")
        return self._enqueue_contract(contract, authority=envelope)

    def accept_signed_task(
        self,
        envelope: SignedTaskEnvelope,
        *,
        remote_lease_token: str,
        remote_lease_expires_at: str,
        remote_attempt: int,
    ) -> QueuedTask:
        """Atomically accept authority and an encrypted recoverable control lease."""

        if self._remote_lease_encoder is None:
            raise EdgeQueueLeaseError("secure remote lease codec is required")
        if not isinstance(envelope, SignedTaskEnvelope) or not envelope.is_verified:
            raise EdgeContractError("a verified SignedTaskEnvelope is required")
        return self._enqueue_contract(
            EdgeTaskContract.from_mapping(envelope.task.to_mapping()),
            authority=envelope,
            remote_binding=self._normalize_remote_binding(
                remote_lease_token, remote_lease_expires_at, remote_attempt
            ),
        )

    def _enqueue_contract(
        self,
        contract: EdgeTaskContract,
        *,
        authority: SignedTaskEnvelope | None,
        remote_binding: tuple[str, str | None, str, int] | None = None,
    ) -> QueuedTask:
        task_mapping = contract.to_mapping()
        encoded = strict_json_bytes(task_mapping).decode("utf-8")
        digest = canonical_sha256(task_mapping)
        if remote_binding is not None and remote_binding[3] != contract.attempt:
            raise EdgeQueueLeaseError(
                "remote lease attempt does not bind the task contract"
            )
        with self._transaction() as connection:
            existing = connection.execute(
                "SELECT * FROM edge_tasks WHERE task_id = ?", (contract.task_id,)
            ).fetchone()
            if existing is not None:
                if existing["digest"] != digest:
                    raise EdgeQueueConflictError(
                        "task ID already exists with a different digest"
                    )
                if authority is not None and existing["authority_digest"] != authority.digest:
                    raise EdgeQueueConflictError(
                        "task ID already exists with a different authority digest"
                    )
                if remote_binding is not None:
                    now = _utc_now(self._clock)
                    old_expiry_text = existing["remote_lease_expires_at"]
                    old_expiry = (
                        datetime.fromisoformat(old_expiry_text.replace("Z", "+00:00"))
                        if old_expiry_text is not None else None
                    )
                    if (
                        existing["remote_lease_token_digest"] != remote_binding[0]
                        and old_expiry is not None and old_expiry > now
                    ):
                        raise EdgeQueueConflictError("active remote lease binding conflicts")
                    new_expiry = datetime.fromisoformat(remote_binding[2].replace("Z", "+00:00"))
                    if new_expiry <= now:
                        raise EdgeQueueLeaseError("remote lease has expired")
                    if (
                        existing["status"] in {"failed", "cancelled"}
                        and not (
                            existing["status"] == "failed"
                            and existing["error_code"] == "execution_lease_expired"
                        )
                    ):
                        raise EdgeQueueConflictError(
                            "terminal task cannot be rebound to a remote lease"
                        )
                    changed_remote_lease = (
                        existing["remote_lease_token_digest"] != remote_binding[0]
                    )
                    if changed_remote_lease:
                        event_rows = connection.execute(
                            """SELECT * FROM edge_outbound_events
                            WHERE status IN ('pending','sending')
                              AND json_extract(event_json,'$.task_id')=?""",
                            (contract.task_id,),
                        ).fetchall()
                        for event_row in event_rows:
                            event_mapping = json.loads(event_row["event_json"])
                            if (
                                event_row["task_digest"] != digest
                                or event_row["remote_lease_token_digest"]
                                != existing["remote_lease_token_digest"]
                                or event_mapping.get("task_id") != contract.task_id
                                or event_mapping.get("gateway_id") != contract.gateway_id
                                or canonical_sha256(event_mapping) != event_row["digest"]
                            ):
                                raise EdgeQueueConflictError(
                                    "pending event does not bind the task lease"
                                )
                        connection.execute(
                            """UPDATE edge_outbound_events
                            SET remote_lease_token_digest=?,updated_at=?
                            WHERE status IN ('pending','sending')
                              AND json_extract(event_json,'$.task_id')=?""",
                            (
                                remote_binding[0],
                                _timestamp(now),
                                contract.task_id,
                            ),
                        )
                    recoverable_failure = (
                        existing["status"] == "failed"
                        and existing["error_code"] == "execution_lease_expired"
                    )
                    connection.execute(
                        """UPDATE edge_tasks SET remote_lease_token_digest=?,
                        remote_lease_ciphertext=?,remote_lease_expires_at=?,remote_attempt=?,
                        status=CASE WHEN ? THEN 'pending' ELSE status END,
                        available_at=CASE WHEN ? THEN ? ELSE available_at END,
                        error_code=CASE WHEN ? THEN NULL ELSE error_code END,
                        updated_at=?
                        WHERE task_id=? AND digest=? AND authority_digest=?""",
                        (
                            remote_binding[0], remote_binding[1], remote_binding[2],
                            remote_binding[3], recoverable_failure,
                            recoverable_failure, _timestamp(now), recoverable_failure,
                            _timestamp(now), contract.task_id,
                            digest, authority.digest if authority is not None else None,
                        ),
                    )
                    existing = connection.execute(
                        "SELECT * FROM edge_tasks WHERE task_id=?", (contract.task_id,)
                    ).fetchone()
                return self._task_record(existing)
            now = _utc_now(self._clock)
            deadline = datetime.fromisoformat(
                contract.deadline_at.replace("Z", "+00:00")
            ).astimezone(UTC)
            if deadline <= now:
                raise EdgeContractError("task deadline has expired")
            if remote_binding is not None:
                remote_expiry = datetime.fromisoformat(remote_binding[2].replace("Z", "+00:00"))
                if remote_expiry <= now or remote_expiry > deadline:
                    raise EdgeQueueLeaseError("remote lease validity window is invalid")
            now_text = _timestamp(now)
            connection.execute(
                """
                INSERT INTO edge_tasks (
                    task_id, digest, task_json, status, attempt_count,
                    available_at, deadline_at, deadline_epoch_us,
                    authority_key_id, authority_digest, authority_json,
                    remote_lease_token_digest, remote_lease_ciphertext,
                    remote_lease_expires_at, remote_attempt,
                    created_at, updated_at
                ) VALUES (?, ?, ?, 'pending', 0, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
                """,
                (
                    contract.task_id,
                    digest,
                    encoded,
                    now_text,
                    contract.deadline_at,
                    _epoch_microseconds(deadline),
                    authority.authority_key_id if authority is not None else None,
                    authority.digest if authority is not None else None,
                    strict_json_bytes(authority.to_mapping()).decode("utf-8")
                    if authority is not None else None,
                    remote_binding[0] if remote_binding is not None else None,
                    remote_binding[1] if remote_binding is not None else None,
                    remote_binding[2] if remote_binding is not None else None,
                    remote_binding[3] if remote_binding is not None else None,
                    now_text,
                    now_text,
                ),
            )
            row = connection.execute(
                "SELECT * FROM edge_tasks WHERE task_id = ?", (contract.task_id,)
            ).fetchone()
            return self._task_record(row)

    def claim(self, lease_owner: str) -> QueuedTask | None:
        owner = self._lease_owner(lease_owner)
        with self._transaction() as connection:
            now = _utc_now(self._clock)
            now_text = _timestamp(now)
            now_epoch_us = _epoch_microseconds(now)
            lease_token = str(uuid.uuid4())
            expires_at = _timestamp(now + timedelta(seconds=self.lease_seconds))
            connection.execute(
                """
                UPDATE edge_tasks
                SET status = 'failed', lease_owner = NULL, lease_token = NULL,
                    lease_expires_at = NULL,
                    error_code = 'task_deadline_expired', updated_at = ?
                WHERE status IN ('pending', 'leased') AND deadline_epoch_us <= ?
                """,
                (now_text, now_epoch_us),
            )
            connection.execute(
                """
                UPDATE edge_tasks
                SET status = 'failed', lease_owner = NULL, lease_token = NULL,
                    lease_expires_at = NULL,
                    error_code = 'lease_attempts_exhausted', updated_at = ?
                WHERE status = 'leased'
                  AND julianday(lease_expires_at) <= julianday(?)
                  AND attempt_count >= ?
                """,
                (now_text, now_text, self.max_attempts),
            )
            row = connection.execute(
                """
                SELECT * FROM edge_tasks
                WHERE deadline_epoch_us > ? AND attempt_count < ? AND (
                    (status = 'pending' AND julianday(available_at) <= julianday(?))
                    OR (status = 'leased'
                        AND julianday(lease_expires_at) <= julianday(?))
                )
                ORDER BY created_at, task_id
                LIMIT 1
                """,
                (now_epoch_us, self.max_attempts, now_text, now_text),
            ).fetchone()
            if row is None:
                return None
            updated = connection.execute(
                """
                UPDATE edge_tasks
                SET status = 'leased', attempt_count = attempt_count + 1,
                    lease_owner = ?, lease_token = ?, lease_expires_at = ?,
                    error_code = NULL, updated_at = ?
                WHERE task_id = ? AND deadline_epoch_us > ?
                  AND attempt_count = ? AND (
                    (status = 'pending' AND julianday(available_at) <= julianday(?))
                    OR (status = 'leased'
                        AND julianday(lease_expires_at) <= julianday(?))
                )
                """,
                (
                    owner,
                    lease_token,
                    expires_at,
                    now_text,
                    row["task_id"],
                    now_epoch_us,
                    row["attempt_count"],
                    now_text,
                    now_text,
                ),
            )
            if updated.rowcount != 1:
                return None
            claimed = connection.execute(
                "SELECT * FROM edge_tasks WHERE task_id = ?", (row["task_id"],)
            ).fetchone()
            return self._task_record(claimed)

    def bind_remote_lease(
        self,
        task_id: str,
        *,
        lease_token: str,
        remote_lease_token: str,
        remote_lease_expires_at: str,
        remote_attempt: int,
    ) -> QueuedTask:
        """Bind control-plane authority to the signed contract attempt."""

        digest, ciphertext, expiry_text, remote_attempt = self._normalize_remote_binding(
            remote_lease_token, remote_lease_expires_at, remote_attempt
        )
        expiry = datetime.fromisoformat(expiry_text.replace("Z", "+00:00"))
        with self._transaction() as connection:
            row = self._leased_row(connection, "edge_tasks", "task_id", task_id, lease_token)
            now = _utc_now(self._clock)
            if expiry <= now:
                raise EdgeQueueLeaseError("remote lease has expired")
            task_mapping = json.loads(row["task_json"])
            if remote_attempt != task_mapping["attempt"]:
                raise EdgeQueueLeaseError("remote lease attempt does not bind task contract")
            existing = row["remote_lease_token_digest"]
            if existing is not None and (
                existing != digest
                or row["remote_lease_expires_at"] != expiry_text
                or row["remote_attempt"] != remote_attempt
            ):
                raise EdgeQueueConflictError("remote lease binding conflicts")
            connection.execute(
                """
                UPDATE edge_tasks
                SET remote_lease_token_digest=?, remote_lease_ciphertext=?, remote_lease_expires_at=?,
                    remote_attempt=?, updated_at=?
                WHERE task_id=? AND status='leased' AND lease_token=?
                """,
                (digest, ciphertext, expiry_text, remote_attempt, _timestamp(now), task_id, lease_token),
            )
            updated = connection.execute(
                "SELECT * FROM edge_tasks WHERE task_id=?", (task_id,)
            ).fetchone()
            return self._task_record(updated)

    def recover_remote_lease(self, task_id: str) -> str:
        """Recover a control lease through the injected secure decoder without exposing it in records."""

        if self._remote_lease_decoder is None:
            raise EdgeQueueLeaseError("secure remote lease codec is required")
        connection = self._connect()
        try:
            row = connection.execute(
                "SELECT remote_lease_ciphertext,remote_lease_token_digest FROM edge_tasks WHERE task_id=?",
                (task_id,),
            ).fetchone()
        finally:
            connection.close()
        if row is None or row["remote_lease_ciphertext"] is None:
            raise EdgeQueueLeaseError("recoverable remote lease does not exist")
        try:
            token = self._remote_lease_decoder(row["remote_lease_ciphertext"])
        except Exception as exc:
            raise EdgeQueueLeaseError("remote lease recovery failed") from exc
        if not isinstance(token, str) or self._opaque_token_digest(token) != row["remote_lease_token_digest"]:
            raise EdgeQueueLeaseError("recovered remote lease failed integrity validation")
        return token

    def complete_with_event(
        self,
        task_id: str,
        *,
        lease_token: str,
        remote_lease_token: str,
        event: EdgeEventContract | Mapping[str, object],
        artifacts: list[Mapping[str, object]] | tuple[Mapping[str, object], ...] = (),
    ) -> tuple[QueuedTask, QueuedEvent]:
        """Atomically commit a safe event, local artifact metadata, and task terminal state."""

        contract = EdgeEventContract.from_mapping(
            event.to_mapping() if isinstance(event, EdgeEventContract) else event
        )
        approved = self.egress_policy.approve_event(contract.to_mapping())
        encoded_event = strict_json_bytes(approved).decode("utf-8")
        normalized_artifacts = [self._validate_local_artifact(item) for item in artifacts]
        if not isinstance(remote_lease_token, str) or not remote_lease_token:
            raise EdgeQueueLeaseError("remote lease token is invalid")
        remote_digest = self._opaque_token_digest(remote_lease_token)
        with self._transaction() as connection:
            now = _utc_now(self._clock)
            now_text = _timestamp(now)
            row = self._leased_row(connection, "edge_tasks", "task_id", task_id, lease_token)
            if row["deadline_epoch_us"] <= _epoch_microseconds(now):
                raise EdgeQueueLeaseError("task deadline has expired")
            if row["remote_lease_token_digest"] != remote_digest:
                raise EdgeQueueLeaseError("current remote lease token is required")
            remote_expiry = row["remote_lease_expires_at"]
            if remote_expiry is None or datetime.fromisoformat(
                remote_expiry.replace("Z", "+00:00")
            ) <= now:
                raise EdgeQueueLeaseError("remote lease has expired")
            task_mapping = json.loads(row["task_json"])
            if row["remote_attempt"] != task_mapping["attempt"]:
                raise EdgeQueueLeaseError("remote lease attempt is stale")
            for field in ("task_id", "gateway_id", "environment", "network_zone", "purpose", "policy_digest"):
                if contract.to_mapping()[field] != task_mapping[field]:
                    raise EdgeQueueConflictError(f"event {field} does not bind the task")
            if contract.attempt != task_mapping["attempt"]:
                raise EdgeQueueConflictError("event attempt does not bind the task")
            existing = connection.execute(
                "SELECT digest FROM edge_outbound_events WHERE event_id=?", (contract.event_id,)
            ).fetchone()
            if existing is not None:
                raise EdgeQueueConflictError("event ID already exists before task completion")
            connection.execute(
                """
                INSERT INTO edge_outbound_events (
                    event_id,digest,event_json,status,attempt_count,available_at,
                    task_digest,remote_lease_token_digest,created_at,updated_at
                ) VALUES (?,?,?,'pending',0,?,?,?,?,?)
                """,
                (
                    contract.event_id, contract.digest, encoded_event, now_text,
                    row["digest"], remote_digest, now_text, now_text,
                ),
            )
            for artifact in normalized_artifacts:
                connection.execute(
                    """
                    INSERT INTO edge_local_artifacts (
                        artifact_digest,artifact_ref,artifact_ref_hash,classification,
                        retention_until,task_id,event_id,cleanup_status,attempt_count,
                        available_at,created_at,updated_at
                    ) VALUES (?,?,?,?,?,?,?,'pending',0,?,?,?)
                    """,
                    (
                        artifact["artifact_digest"], artifact["artifact_ref"],
                        artifact["artifact_ref_hash"], artifact["classification"],
                        artifact["retention_until"], task_id, contract.event_id,
                        now_text, now_text, now_text,
                    ),
                )
            updated = connection.execute(
                """
                UPDATE edge_tasks SET status='completed', lease_owner=NULL,
                    lease_token=NULL, lease_expires_at=NULL, error_code=NULL, updated_at=?
                WHERE task_id=? AND status='leased' AND lease_token=?
                """,
                (now_text, task_id, lease_token),
            )
            if updated.rowcount != 1:
                raise EdgeQueueLeaseError("task completion lost its lease")
            task_row = connection.execute("SELECT * FROM edge_tasks WHERE task_id=?", (task_id,)).fetchone()
            event_row = connection.execute("SELECT * FROM edge_outbound_events WHERE event_id=?", (contract.event_id,)).fetchone()
            return self._task_record(task_row), self._event_record(event_row)

    def list_local_artifacts(self, *, task_id: str) -> tuple[LocalArtifact, ...]:
        connection = self._connect()
        try:
            rows = connection.execute(
                "SELECT * FROM edge_local_artifacts WHERE task_id=? ORDER BY created_at,artifact_digest",
                (task_id,),
            ).fetchall()
        finally:
            connection.close()
        return tuple(self._artifact_record(row) for row in rows)

    def claim_artifact_cleanup(self, lease_owner: str) -> LocalArtifact | None:
        owner = self._lease_owner(lease_owner)
        with self._transaction() as connection:
            now = _utc_now(self._clock)
            now_text = _timestamp(now)
            expires_at = _timestamp(now + timedelta(seconds=self.lease_seconds))
            connection.execute(
                """UPDATE edge_local_artifacts SET cleanup_status='failed',
                lease_owner=NULL,lease_token=NULL,lease_expires_at=NULL,
                error_code='lease_attempts_exhausted',updated_at=?
                WHERE cleanup_status='deleting'
                AND julianday(lease_expires_at)<=julianday(?) AND attempt_count>=?""",
                (now_text, now_text, self.max_attempts),
            )
            row = connection.execute(
                """SELECT * FROM edge_local_artifacts WHERE attempt_count<?
                AND julianday(retention_until)<=julianday(?) AND (
                  (cleanup_status='pending' AND julianday(available_at)<=julianday(?))
                  OR (cleanup_status='deleting'
                      AND julianday(lease_expires_at)<=julianday(?)))
                ORDER BY retention_until,created_at,artifact_digest LIMIT 1""",
                (self.max_attempts, now_text, now_text, now_text),
            ).fetchone()
            if row is None:
                return None
            lease_token = str(uuid.uuid4())
            connection.execute(
                """UPDATE edge_local_artifacts SET cleanup_status='deleting',
                attempt_count=attempt_count+1,lease_owner=?,lease_token=?,
                lease_expires_at=?,error_code=NULL,updated_at=?
                WHERE artifact_digest=? AND attempt_count=? AND (
                  (cleanup_status='pending' AND julianday(available_at)<=julianday(?))
                  OR (cleanup_status='deleting'
                      AND julianday(lease_expires_at)<=julianday(?)))""",
                (
                    owner, lease_token, expires_at, now_text, row["artifact_digest"],
                    row["attempt_count"], now_text, now_text,
                ),
            )
            claimed = connection.execute(
                "SELECT * FROM edge_local_artifacts WHERE artifact_digest=?",
                (row["artifact_digest"],),
            ).fetchone()
            return self._artifact_record(claimed)

    def acknowledge_artifact_cleanup(
        self,
        artifact_digest: str,
        *,
        lease_token: str,
        receipt: Mapping[str, object],
    ) -> LocalArtifact:
        required = {"artifact_digest", "artifact_ref_hash", "deleted_at", "status"}
        with self._transaction() as connection:
            row = self._leased_row(
                connection, "edge_local_artifacts", "artifact_digest",
                artifact_digest, lease_token, status="deleting",
                status_column="cleanup_status",
            )
            if (
                not isinstance(receipt, Mapping)
                or set(receipt) != required
                or receipt.get("artifact_digest") != artifact_digest
                or receipt.get("artifact_ref_hash") != row["artifact_ref_hash"]
                or receipt.get("status") != "deleted"
            ):
                raise EdgePolicyError("artifact deletion receipt is invalid")
            try:
                deleted_at = canonical_timestamp(receipt["deleted_at"], "deleted_at")
            except EdgeContractError as exc:
                raise EdgePolicyError("artifact deletion receipt is invalid") from exc
            normalized = {
                "artifact_digest": artifact_digest,
                "artifact_ref_hash": row["artifact_ref_hash"],
                "deleted_at": deleted_at,
                "status": "deleted",
            }
            now_text = _timestamp(_utc_now(self._clock))
            connection.execute(
                """UPDATE edge_local_artifacts SET artifact_ref=NULL,
                cleanup_status='deleted',deletion_receipt_json=?,lease_owner=NULL,
                lease_token=NULL,lease_expires_at=NULL,error_code=NULL,updated_at=?
                WHERE artifact_digest=? AND cleanup_status='deleting' AND lease_token=?""",
                (
                    strict_json_bytes(normalized).decode("utf-8"), now_text,
                    artifact_digest, lease_token,
                ),
            )
            updated = connection.execute(
                "SELECT * FROM edge_local_artifacts WHERE artifact_digest=?",
                (artifact_digest,),
            ).fetchone()
            return self._artifact_record(updated)

    def fail_artifact_cleanup(
        self, artifact_digest: str, *, lease_token: str, error_code: str
    ) -> LocalArtifact:
        safe_error = self._error_code(error_code)
        with self._transaction() as connection:
            now = _utc_now(self._clock)
            now_text = _timestamp(now)
            row = self._leased_row(
                connection, "edge_local_artifacts", "artifact_digest",
                artifact_digest, lease_token, status="deleting",
                status_column="cleanup_status",
            )
            terminal = row["attempt_count"] >= self.max_attempts
            status = "failed" if terminal else "pending"
            available_at = now_text if terminal else _timestamp(
                now + timedelta(seconds=self._retry_delay(row["attempt_count"]))
            )
            connection.execute(
                """UPDATE edge_local_artifacts SET cleanup_status=?,available_at=?,
                lease_owner=NULL,lease_token=NULL,lease_expires_at=NULL,error_code=?,
                updated_at=? WHERE artifact_digest=? AND cleanup_status='deleting'
                AND lease_token=?""",
                (
                    status, available_at, safe_error, now_text,
                    artifact_digest, lease_token,
                ),
            )
            updated = connection.execute(
                "SELECT * FROM edge_local_artifacts WHERE artifact_digest=?",
                (artifact_digest,),
            ).fetchone()
            return self._artifact_record(updated)

    def put_release_state(
        self,
        *,
        release_id: str,
        manifest_digest: str,
        version: str,
        rollback_version: str,
    ) -> ReleaseState:
        release_id = self._bounded_identifier(release_id, "release_id")
        manifest_digest = self._sha256(manifest_digest, "manifest_digest")
        version = self._bounded_identifier(version, "version")
        rollback_version = self._bounded_identifier(rollback_version, "rollback_version")
        with self._transaction() as connection:
            existing = connection.execute(
                "SELECT * FROM edge_release_state WHERE release_id=?", (release_id,)
            ).fetchone()
            if existing is not None:
                if (
                    existing["manifest_digest"], existing["version"], existing["rollback_version"]
                ) != (manifest_digest, version, rollback_version):
                    raise EdgeQueueConflictError("release ID has different immutable content")
                return ReleaseState(**dict(existing))
            now_text = _timestamp(_utc_now(self._clock))
            connection.execute(
                """INSERT INTO edge_release_state
                (release_id,manifest_digest,version,rollback_version,status,updated_at)
                VALUES (?,?,?,?,'offered',?)""",
                (release_id, manifest_digest, version, rollback_version, now_text),
            )
            row = connection.execute(
                "SELECT * FROM edge_release_state WHERE release_id=?", (release_id,)
            ).fetchone()
            return ReleaseState(**dict(row))

    def compare_and_set_release_state(
        self,
        release_id: str,
        *,
        expected_status: str,
        target_status: str,
        previous_version: str | None = None,
        current_version: str | None = None,
    ) -> ReleaseState:
        release_id = self._bounded_identifier(release_id, "release_id")
        allowed = {"offered", "accepted", "candidate", "installed", "failed", "rolled_back"}
        if expected_status not in allowed or target_status not in allowed:
            raise EdgeQueueConflictError("release status is invalid")
        transitions = {
            "offered": {"accepted", "failed"},
            "accepted": {"candidate", "failed"},
            "candidate": {"installed", "failed", "rolled_back"},
            "installed": {"rolled_back", "failed"},
            "failed": {"rolled_back"},
            "rolled_back": set(),
        }
        if target_status not in transitions[expected_status]:
            raise EdgeQueueConflictError("release state transition is not approved")
        if previous_version is not None:
            previous_version = self._bounded_identifier(previous_version, "previous_version")
        if current_version is not None:
            current_version = self._bounded_identifier(current_version, "current_version")
        with self._transaction() as connection:
            now_text = _timestamp(_utc_now(self._clock))
            updated = connection.execute(
                """UPDATE edge_release_state
                SET status=?,previous_version=COALESCE(?,previous_version),
                    current_version=COALESCE(?,current_version),updated_at=?
                WHERE release_id=? AND status=?""",
                (
                    target_status, previous_version, current_version, now_text,
                    release_id, expected_status,
                ),
            )
            if updated.rowcount != 1:
                row = connection.execute(
                    "SELECT status FROM edge_release_state WHERE release_id=?", (release_id,)
                ).fetchone()
                if row is None:
                    raise EdgeQueueError("release does not exist")
                raise EdgeQueueConflictError("release state compare-and-set lost")
            row = connection.execute(
                "SELECT * FROM edge_release_state WHERE release_id=?", (release_id,)
            ).fetchone()
            return ReleaseState(**dict(row))

    def get_release_state(self, release_id: str) -> ReleaseState | None:
        release_id = self._bounded_identifier(release_id, "release_id")
        connection = self._connect()
        try:
            row = connection.execute(
                "SELECT * FROM edge_release_state WHERE release_id=?", (release_id,)
            ).fetchone()
        finally:
            connection.close()
        return ReleaseState(**dict(row)) if row is not None else None

    def list_release_states(self) -> tuple[ReleaseState, ...]:
        connection = self._connect()
        try:
            rows = connection.execute(
                "SELECT * FROM edge_release_state ORDER BY updated_at,release_id"
            ).fetchall()
        finally:
            connection.close()
        return tuple(ReleaseState(**dict(row)) for row in rows)

    def complete(self, task_id: str, *, lease_token: str) -> QueuedTask:
        return self._finish_task(
            task_id,
            lease_token=lease_token,
            target_status="completed",
            error_code=None,
        )

    def fail(
        self,
        task_id: str,
        *,
        lease_token: str,
        error_code: str,
        retryable: bool = True,
    ) -> QueuedTask:
        safe_error = self._error_code(error_code)
        if not isinstance(retryable, bool):
            raise ValueError("retryable must be a boolean")
        with self._transaction() as connection:
            now = _utc_now(self._clock)
            now_text = _timestamp(now)
            row = self._leased_row(
                connection, "edge_tasks", "task_id", task_id, lease_token
            )
            terminal = not retryable or row["attempt_count"] >= self.max_attempts
            status = "failed" if terminal else "pending"
            available_at = now_text
            if not terminal:
                available_at = _timestamp(
                    now + timedelta(seconds=self._retry_delay(row["attempt_count"]))
                )
            connection.execute(
                """
                UPDATE edge_tasks
                SET status = ?, available_at = ?, lease_owner = NULL,
                    lease_token = NULL, lease_expires_at = NULL,
                    error_code = ?, updated_at = ?
                WHERE task_id = ? AND status = 'leased' AND lease_token = ?
                """,
                (status, available_at, safe_error, now_text, task_id, lease_token),
            )
            updated = connection.execute(
                "SELECT * FROM edge_tasks WHERE task_id = ?", (task_id,)
            ).fetchone()
            return self._task_record(updated)

    def _finish_task(
        self,
        task_id: str,
        *,
        lease_token: str,
        target_status: str,
        error_code: str | None,
    ) -> QueuedTask:
        with self._transaction() as connection:
            now_text = _timestamp(_utc_now(self._clock))
            self._leased_row(
                connection, "edge_tasks", "task_id", task_id, lease_token
            )
            connection.execute(
                """
                UPDATE edge_tasks
                SET status = ?, lease_owner = NULL, lease_token = NULL,
                    lease_expires_at = NULL, error_code = ?, updated_at = ?
                WHERE task_id = ? AND status = 'leased' AND lease_token = ?
                """,
                (target_status, error_code, now_text, task_id, lease_token),
            )
            row = connection.execute(
                "SELECT * FROM edge_tasks WHERE task_id = ?", (task_id,)
            ).fetchone()
            return self._task_record(row)

    def cancel(self, task_id: str) -> QueuedTask:
        with self._transaction() as connection:
            now_text = _timestamp(_utc_now(self._clock))
            row = connection.execute(
                "SELECT * FROM edge_tasks WHERE task_id = ?", (task_id,)
            ).fetchone()
            if row is None:
                raise EdgeQueueError("task does not exist")
            if row["status"] in {"pending", "leased"}:
                connection.execute(
                    """
                    UPDATE edge_tasks
                    SET status = 'cancelled', lease_owner = NULL,
                        lease_token = NULL, lease_expires_at = NULL,
                        updated_at = ?
                    WHERE task_id = ? AND status IN ('pending', 'leased')
                    """,
                    (now_text, task_id),
                )
                row = connection.execute(
                    "SELECT * FROM edge_tasks WHERE task_id = ?", (task_id,)
                ).fetchone()
            return self._task_record(row)

    def cancel_with_outcome(self, task_id: str) -> QueuedOutcome:
        """Atomically fence local work and persist the cancelled control outcome."""

        task_id = self._bounded_identifier(task_id, "task_id")
        with self._transaction() as connection:
            now_text = _timestamp(_utc_now(self._clock))
            row = connection.execute(
                "SELECT * FROM edge_tasks WHERE task_id = ?", (task_id,)
            ).fetchone()
            if row is None:
                raise EdgeQueueError("task does not exist")
            if row["status"] in {"pending", "leased"}:
                connection.execute(
                    """UPDATE edge_tasks SET status='cancelled',lease_owner=NULL,
                    lease_token=NULL,lease_expires_at=NULL,updated_at=?
                    WHERE task_id=? AND status IN ('pending','leased')""",
                    (now_text, task_id),
                )
                row = connection.execute(
                    "SELECT * FROM edge_tasks WHERE task_id=?", (task_id,)
                ).fetchone()
            if row["status"] != "cancelled":
                raise EdgeQueueConflictError("only a cancelled task can emit cancellation")
            remote_digest = row["remote_lease_token_digest"]
            if not isinstance(remote_digest, str):
                raise EdgeQueueLeaseError("cancel outcome requires a remote lease binding")
            outcome = {
                "task_id": task_id,
                "outcome": "cancelled",
                "safe_summary": {"task_status": "cancelled"},
            }
            digest = canonical_sha256(outcome)
            encoded = strict_json_bytes(outcome).decode("utf-8")
            existing = connection.execute(
                "SELECT * FROM edge_task_outcomes WHERE task_id=?", (task_id,)
            ).fetchone()
            if existing is not None:
                if (
                    existing["digest"] != digest
                    or existing["remote_lease_token_digest"] != remote_digest
                ):
                    raise EdgeQueueConflictError("cancel outcome replay conflicts")
                return self._outcome_record(existing)
            connection.execute(
                """INSERT INTO edge_task_outcomes(
                task_id,digest,outcome_json,status,attempt_count,available_at,
                remote_lease_token_digest,created_at,updated_at
                ) VALUES (?,?,?,'pending',0,?,?,?,?)""",
                (task_id, digest, encoded, now_text, remote_digest, now_text, now_text),
            )
            inserted = connection.execute(
                "SELECT * FROM edge_task_outcomes WHERE task_id=?", (task_id,)
            ).fetchone()
            return self._outcome_record(inserted)

    def claim_outcome(self, lease_owner: str) -> QueuedOutcome | None:
        owner = self._lease_owner(lease_owner)
        with self._transaction() as connection:
            now = _utc_now(self._clock)
            now_text = _timestamp(now)
            expires_at = _timestamp(now + timedelta(seconds=self.lease_seconds))
            connection.execute(
                """UPDATE edge_task_outcomes SET status='failed',lease_owner=NULL,
                lease_token=NULL,lease_expires_at=NULL,error_code='lease_attempts_exhausted',
                updated_at=? WHERE status='sending'
                AND julianday(lease_expires_at)<=julianday(?) AND attempt_count>=?""",
                (now_text, now_text, self.max_attempts),
            )
            row = connection.execute(
                """SELECT * FROM edge_task_outcomes WHERE attempt_count<? AND (
                (status='pending' AND julianday(available_at)<=julianday(?)) OR
                (status='sending' AND julianday(lease_expires_at)<=julianday(?)))
                ORDER BY created_at,task_id LIMIT 1""",
                (self.max_attempts, now_text, now_text),
            ).fetchone()
            if row is None:
                return None
            lease_token = str(uuid.uuid4())
            connection.execute(
                """UPDATE edge_task_outcomes SET status='sending',
                attempt_count=attempt_count+1,lease_owner=?,lease_token=?,
                lease_expires_at=?,error_code=NULL,updated_at=?
                WHERE task_id=? AND attempt_count=? AND (
                (status='pending' AND julianday(available_at)<=julianday(?)) OR
                (status='sending' AND julianday(lease_expires_at)<=julianday(?)))""",
                (
                    owner, lease_token, expires_at, now_text, row["task_id"],
                    row["attempt_count"], now_text, now_text,
                ),
            )
            claimed = connection.execute(
                "SELECT * FROM edge_task_outcomes WHERE task_id=?", (row["task_id"],)
            ).fetchone()
            return self._outcome_record(claimed)

    def acknowledge_outcome(
        self,
        task_id: str,
        *,
        lease_token: str,
        acknowledgement: Mapping[str, object],
    ) -> QueuedOutcome:
        required = {"task_id", "status", "replayed"}
        if (
            not isinstance(acknowledgement, Mapping)
            or set(acknowledgement) != required
            or acknowledgement.get("task_id") != task_id
            or acknowledgement.get("status") != "cancelled"
            or not isinstance(acknowledgement.get("replayed"), bool)
        ):
            raise EdgePolicyError("cancel outcome acknowledgement is invalid")
        encoded = strict_json_bytes(acknowledgement).decode("utf-8")
        ack_digest = canonical_sha256(acknowledgement)
        with self._transaction() as connection:
            now_text = _timestamp(_utc_now(self._clock))
            self._leased_row(
                connection, "edge_task_outcomes", "task_id", task_id, lease_token,
                status="sending",
            )
            connection.execute(
                """UPDATE edge_task_outcomes SET status='acknowledged',
                acknowledgement_json=?,ack_digest=?,lease_owner=NULL,lease_token=NULL,
                lease_expires_at=NULL,updated_at=?
                WHERE task_id=? AND status='sending' AND lease_token=?""",
                (encoded, ack_digest, now_text, task_id, lease_token),
            )
            row = connection.execute(
                "SELECT * FROM edge_task_outcomes WHERE task_id=?", (task_id,)
            ).fetchone()
            return self._outcome_record(row)

    def fail_outcome(
        self, task_id: str, *, lease_token: str, error_code: str
    ) -> QueuedOutcome:
        safe_error = self._error_code(error_code)
        with self._transaction() as connection:
            now = _utc_now(self._clock)
            now_text = _timestamp(now)
            row = self._leased_row(
                connection, "edge_task_outcomes", "task_id", task_id, lease_token,
                status="sending",
            )
            terminal = row["attempt_count"] >= self.max_attempts
            status = "failed" if terminal else "pending"
            available_at = now_text if terminal else _timestamp(
                now + timedelta(seconds=self._retry_delay(row["attempt_count"]))
            )
            connection.execute(
                """UPDATE edge_task_outcomes SET status=?,available_at=?,lease_owner=NULL,
                lease_token=NULL,lease_expires_at=NULL,error_code=?,updated_at=?
                WHERE task_id=? AND status='sending' AND lease_token=?""",
                (
                    status, available_at, safe_error, now_text, task_id, lease_token,
                ),
            )
            updated = connection.execute(
                "SELECT * FROM edge_task_outcomes WHERE task_id=?", (task_id,)
            ).fetchone()
            return self._outcome_record(updated)

    def get_outcome(self, task_id: str) -> QueuedOutcome | None:
        connection = self._connect()
        try:
            row = connection.execute(
                "SELECT * FROM edge_task_outcomes WHERE task_id=?", (task_id,)
            ).fetchone()
        finally:
            connection.close()
        return self._outcome_record(row) if row is not None else None

    def get_task(self, task_id: str) -> QueuedTask | None:
        connection = self._connect()
        try:
            row = connection.execute(
                "SELECT * FROM edge_tasks WHERE task_id = ?", (task_id,)
            ).fetchone()
        finally:
            connection.close()
        return self._task_record(row) if row is not None else None

    def persist_event(
        self, event: EdgeEventContract | Mapping[str, object]
    ) -> QueuedEvent:
        raw = event.to_mapping() if isinstance(event, EdgeEventContract) else dict(event)
        try:
            contract = EdgeEventContract.from_mapping(raw)
        except EdgeContractError as exc:
            raw_id = raw.get("event_id")
            if isinstance(raw_id, str) and raw_id:
                with self._transaction() as connection:
                    existing = connection.execute(
                        "SELECT 1 FROM edge_outbound_events WHERE event_id = ?",
                        (raw_id,),
                    ).fetchone()
                if existing is not None:
                    raise EdgeQueueConflictError(
                        "event ID already exists with a different digest"
                    ) from exc
            raise
        normalized = contract.to_mapping()
        digest = contract.digest
        with self._transaction() as connection:
            now_text = _timestamp(_utc_now(self._clock))
            existing = connection.execute(
                "SELECT * FROM edge_outbound_events WHERE event_id = ?",
                (contract.event_id,),
            ).fetchone()
            if existing is not None:
                if existing["digest"] != digest:
                    raise EdgeQueueConflictError(
                        "event ID already exists with a different digest"
                    )
                return self._event_record(existing)
            approved = self.egress_policy.approve_event(normalized)
            encoded = strict_json_bytes(approved).decode("utf-8")
            connection.execute(
                """
                INSERT INTO edge_outbound_events (
                    event_id, digest, event_json, status, attempt_count,
                    available_at, created_at, updated_at
                ) VALUES (?, ?, ?, 'pending', 0, ?, ?, ?)
                """,
                (contract.event_id, digest, encoded, now_text, now_text, now_text),
            )
            row = connection.execute(
                "SELECT * FROM edge_outbound_events WHERE event_id = ?",
                (contract.event_id,),
            ).fetchone()
            return self._event_record(row)

    def claim_event(self, lease_owner: str) -> QueuedEvent | None:
        owner = self._lease_owner(lease_owner)
        with self._transaction() as connection:
            now = _utc_now(self._clock)
            now_text = _timestamp(now)
            lease_token = str(uuid.uuid4())
            expires_at = _timestamp(now + timedelta(seconds=self.lease_seconds))
            connection.execute(
                """
                UPDATE edge_outbound_events
                SET status = 'failed', lease_owner = NULL, lease_token = NULL,
                    lease_expires_at = NULL,
                    error_code = 'lease_attempts_exhausted', updated_at = ?
                WHERE status = 'sending'
                  AND julianday(lease_expires_at) <= julianday(?)
                  AND attempt_count >= ?
                """,
                (now_text, now_text, self.max_attempts),
            )
            row = connection.execute(
                """
                SELECT * FROM edge_outbound_events
                WHERE attempt_count < ? AND (
                    (status = 'pending' AND julianday(available_at) <= julianday(?))
                    OR (status = 'sending'
                        AND julianday(lease_expires_at) <= julianday(?))
                )
                ORDER BY created_at, event_id
                LIMIT 1
                """,
                (self.max_attempts, now_text, now_text),
            ).fetchone()
            if row is None:
                return None
            connection.execute(
                """
                UPDATE edge_outbound_events
                SET status = 'sending', attempt_count = attempt_count + 1,
                    lease_owner = ?, lease_token = ?, lease_expires_at = ?,
                    error_code = NULL, updated_at = ?
                WHERE event_id = ? AND attempt_count = ? AND (
                    (status = 'pending' AND julianday(available_at) <= julianday(?))
                    OR (status = 'sending'
                        AND julianday(lease_expires_at) <= julianday(?))
                )
                """,
                (
                    owner,
                    lease_token,
                    expires_at,
                    now_text,
                    row["event_id"],
                    row["attempt_count"],
                    now_text,
                    now_text,
                ),
            )
            claimed = connection.execute(
                "SELECT * FROM edge_outbound_events WHERE event_id = ?",
                (row["event_id"],),
            ).fetchone()
            return self._event_record(claimed)

    def acknowledge_event(
        self,
        event_id: str,
        *,
        lease_token: str,
        acknowledgement: Mapping[str, object],
    ) -> QueuedEvent:
        token_digest = canonical_sha256(lease_token)
        with self._transaction() as connection:
            now_text = _timestamp(_utc_now(self._clock))
            existing = connection.execute(
                "SELECT * FROM edge_outbound_events WHERE event_id = ?", (event_id,)
            ).fetchone()
            if existing is None:
                raise EdgeQueueError("event does not exist")
            normalized_ack = self._validate_acknowledgement(
                acknowledgement,
                event_id=event_id,
                event_digest=existing["digest"],
                remote_lease_token_digest=existing["remote_lease_token_digest"],
            )
            encoded = strict_json_bytes(normalized_ack)
            if len(encoded) > 4_096:
                raise EdgePolicyError("acknowledgement exceeds the byte limit")
            ack_digest = canonical_sha256(normalized_ack)
            if existing["status"] == "acknowledged":
                if existing["lease_token_digest"] != token_digest:
                    raise EdgeQueueLeaseError(
                        "acknowledgement replay requires the original lease token"
                    )
                if existing["ack_digest"] != ack_digest:
                    raise EdgeQueueConflictError(
                        "acknowledgement replay has a different digest"
                    )
                return self._event_record(existing)
            self._leased_row(
                connection,
                "edge_outbound_events",
                "event_id",
                event_id,
                lease_token,
                status="sending",
            )
            connection.execute(
                """
                UPDATE edge_outbound_events
                SET status = 'acknowledged', acknowledgement_json = ?,
                    ack_digest = ?, lease_token_digest = ?,
                    control_lease_token_digest = ?,
                    lease_owner = NULL, lease_token = NULL,
                    lease_expires_at = NULL, updated_at = ?
                WHERE event_id = ? AND status = 'sending' AND lease_token = ?
                """,
                (
                    encoded.decode("utf-8"),
                    ack_digest,
                    token_digest,
                    token_digest,
                    now_text,
                    event_id,
                    lease_token,
                ),
            )
            row = connection.execute(
                "SELECT * FROM edge_outbound_events WHERE event_id = ?", (event_id,)
            ).fetchone()
            return self._event_record(row)

    def fail_event(
        self, event_id: str, *, lease_token: str, error_code: str
    ) -> QueuedEvent:
        safe_error = self._error_code(error_code)
        with self._transaction() as connection:
            now = _utc_now(self._clock)
            now_text = _timestamp(now)
            row = self._leased_row(
                connection,
                "edge_outbound_events",
                "event_id",
                event_id,
                lease_token,
                status="sending",
            )
            terminal = row["attempt_count"] >= self.max_attempts
            status = "failed" if terminal else "pending"
            available_at = now_text
            if not terminal:
                available_at = _timestamp(
                    now + timedelta(seconds=self._retry_delay(row["attempt_count"]))
                )
            connection.execute(
                """
                UPDATE edge_outbound_events
                SET status = ?, available_at = ?, lease_owner = NULL,
                    lease_token = NULL, lease_expires_at = NULL,
                    error_code = ?, updated_at = ?
                WHERE event_id = ? AND status = 'sending' AND lease_token = ?
                """,
                (status, available_at, safe_error, now_text, event_id, lease_token),
            )
            updated = connection.execute(
                "SELECT * FROM edge_outbound_events WHERE event_id = ?", (event_id,)
            ).fetchone()
            return self._event_record(updated)

    def get_event(self, event_id: str) -> QueuedEvent | None:
        connection = self._connect()
        try:
            row = connection.execute(
                "SELECT * FROM edge_outbound_events WHERE event_id = ?", (event_id,)
            ).fetchone()
        finally:
            connection.close()
        return self._event_record(row) if row is not None else None

    @staticmethod
    def _validate_local_artifact(value: Mapping[str, object]) -> dict[str, str]:
        required = {
            "artifact_digest", "artifact_ref", "artifact_ref_hash",
            "classification", "retention_until",
        }
        if not isinstance(value, Mapping) or set(value) != required:
            raise EdgePolicyError("local artifact requires exact metadata fields")
        digest = value["artifact_digest"]
        reference = value["artifact_ref"]
        reference_hash = value["artifact_ref_hash"]
        classification = value["classification"]
        if not isinstance(digest, str) or not re.fullmatch(r"[0-9a-f]{64}", digest):
            raise EdgePolicyError("local artifact digest is invalid")
        if (
            not isinstance(reference, str)
            or not Path(reference).is_absolute()
            or len(reference.encode("utf-8")) > 2_048
            or not isinstance(reference_hash, str)
            or not re.fullmatch(r"[0-9a-f]{64}", reference_hash)
            or reference_hash != canonical_sha256(reference)
        ):
            raise EdgePolicyError("local artifact reference is invalid")
        if classification not in {
            "raw", "recent_detail", "restricted", "desensitized_metadata",
            "statistics", "lineage", "evidence",
        }:
            raise EdgePolicyError("local artifact classification is invalid")
        try:
            retention = canonical_timestamp(value["retention_until"], "retention_until")
        except EdgeContractError as exc:
            raise EdgePolicyError("local artifact retention is invalid") from exc
        return {
            "artifact_digest": digest,
            "artifact_ref": reference,  # type: ignore[dict-item]
            "artifact_ref_hash": reference_hash,  # type: ignore[dict-item]
            "classification": classification,  # type: ignore[dict-item]
            "retention_until": retention,
        }

    def _normalize_remote_binding(
        self, token: object, expires_at: object, attempt: object
    ) -> tuple[str, str | None, str, int]:
        if (
            not isinstance(token, str) or not token.strip()
            or len(token.encode("utf-8")) > 1024
        ):
            raise EdgeQueueLeaseError("remote lease token is invalid")
        if isinstance(attempt, bool) or not isinstance(attempt, int) or attempt < 1:
            raise EdgeQueueLeaseError("remote lease attempt is invalid")
        try:
            expiry = canonical_timestamp(expires_at, "remote_lease_expires_at")
        except EdgeContractError as exc:
            raise EdgeQueueLeaseError("remote lease expiry is invalid") from exc
        ciphertext: str | None = None
        if self._remote_lease_encoder is not None:
            try:
                ciphertext = self._remote_lease_encoder(token)
            except Exception as exc:
                raise EdgeQueueLeaseError("remote lease protection failed") from exc
            if (
                not isinstance(ciphertext, str) or not ciphertext
                or ciphertext == token or len(ciphertext.encode("utf-8")) > 4096
            ):
                raise EdgeQueueLeaseError("remote lease protection produced invalid ciphertext")
        return self._opaque_token_digest(token), ciphertext, expiry, attempt

    @staticmethod
    def _opaque_token_digest(token: str) -> str:
        return hashlib.sha256(token.encode("utf-8")).hexdigest()

    def _retry_delay(self, attempt_count: int) -> int:
        base = min(
            self.retry_max_seconds,
            self.retry_base_seconds * (2 ** max(0, attempt_count - 1)),
        )
        sample = self._random()
        if isinstance(sample, bool) or not isinstance(sample, (int, float)) or not 0 <= sample <= 1:
            raise EdgeQueueError("retry random source returned an invalid value")
        return min(
            self.retry_max_seconds,
            max(1, round(base * (0.5 + float(sample)))),
        )

    @staticmethod
    def _bounded_identifier(value: object, label: str) -> str:
        if (
            not isinstance(value, str) or not value or value.strip() != value
            or len(value.encode("utf-8")) > 255
            or not re.fullmatch(r"[A-Za-z0-9][A-Za-z0-9._:-]{0,254}", value)
        ):
            raise EdgeQueueError(f"{label} is invalid")
        return value

    @staticmethod
    def _sha256(value: object, label: str) -> str:
        if not isinstance(value, str) or not re.fullmatch(r"[0-9a-f]{64}", value):
            raise EdgeQueueError(f"{label} is invalid")
        return value

    @staticmethod
    def _lease_owner(value: object) -> str:
        if (
            not isinstance(value, str)
            or not value
            or len(value.encode("utf-8")) > 255
            or value.strip() != value
        ):
            raise ValueError("lease_owner is invalid")
        return value

    @staticmethod
    def _error_code(value: object) -> str:
        if not isinstance(value, str) or not _ERROR_CODE.fullmatch(value):
            raise EdgePolicyError("error_code must be a bounded code")
        if any(part in value for part in _SENSITIVE_ERROR_PARTS):
            raise EdgePolicyError("error_code contains sensitive content")
        return value

    def _leased_row(
        self,
        connection: sqlite3.Connection,
        table: str,
        id_column: str,
        identity: str,
        lease_token: str,
        *,
        status: str = "leased",
        status_column: str = "status",
    ) -> sqlite3.Row:
        row = connection.execute(
            f"SELECT * FROM {table} WHERE {id_column} = ?",  # noqa: S608
            (identity,),
        ).fetchone()
        if row is None:
            raise EdgeQueueError("queue item does not exist")
        if row[status_column] != status or row["lease_token"] != lease_token:
            raise EdgeQueueLeaseError("current lease token is required")
        lease_expires_at = row["lease_expires_at"]
        if lease_expires_at is None:
            raise EdgeQueueLeaseError("lease has expired")
        parsed_expiry = datetime.fromisoformat(lease_expires_at.replace("Z", "+00:00"))
        if parsed_expiry <= _utc_now(self._clock):
            raise EdgeQueueLeaseError("lease has expired")
        return row

    @staticmethod
    def _validate_acknowledgement(
        value: Mapping[str, object],
        *,
        event_id: str,
        event_digest: str,
        remote_lease_token_digest: str | None,
    ) -> dict[str, str]:
        if remote_lease_token_digest is not None:
            required = {"event_id", "event_digest", "remote_lease_digest", "received_at", "status"}
            if not isinstance(value, Mapping) or set(value) != required:
                raise EdgePolicyError("bound acknowledgement requires exact approved fields")
            if value["event_id"] != event_id or value["event_digest"] != event_digest:
                raise EdgePolicyError("acknowledgement does not bind the event")
            if value["remote_lease_digest"] != remote_lease_token_digest:
                raise EdgePolicyError("acknowledgement remote lease is invalid")
            if value["status"] != "accepted":
                raise EdgePolicyError("acknowledgement status is invalid")
            try:
                received_at = canonical_timestamp(value["received_at"], "received_at")
            except EdgeContractError as exc:
                raise EdgePolicyError("acknowledgement received_at is invalid") from exc
            return {
                "event_id": event_id,
                "event_digest": event_digest,
                "remote_lease_digest": remote_lease_token_digest,
                "received_at": received_at,
                "status": "accepted",
            }
        if not isinstance(value, Mapping) or set(value) != {
            "message_id",
            "received_at",
            "status",
        }:
            raise EdgePolicyError("acknowledgement requires exact approved fields")
        message_id = value["message_id"]
        status = value["status"]
        if (
            not isinstance(message_id, str)
            or not message_id
            or message_id.strip() != message_id
            or len(message_id.encode("utf-8")) > 255
        ):
            raise EdgePolicyError("acknowledgement message_id is invalid")
        if status != "accepted":
            raise EdgePolicyError("acknowledgement status is invalid")
        try:
            received_at = canonical_timestamp(value["received_at"], "received_at")
        except EdgeContractError as exc:
            raise EdgePolicyError("acknowledgement received_at is invalid") from exc
        return {
            "message_id": message_id,
            "received_at": received_at,
            "status": status,
        }

    @staticmethod
    def _task_record(row: sqlite3.Row) -> QueuedTask:
        task = json.loads(row["task_json"])
        return QueuedTask(
            task_id=row["task_id"],
            digest=row["digest"],
            status=row["status"],
            task=_freeze_mapping(task),
            attempt_count=row["attempt_count"],
            available_at=row["available_at"],
            lease_owner=row["lease_owner"],
            lease_token=row["lease_token"],
            lease_expires_at=row["lease_expires_at"],
            error_code=row["error_code"],
            authority_key_id=row["authority_key_id"],
            authority_digest=row["authority_digest"],
            remote_lease_token_digest=row["remote_lease_token_digest"],
            remote_lease_expires_at=row["remote_lease_expires_at"],
            remote_attempt=row["remote_attempt"],
            created_at=row["created_at"],
            updated_at=row["updated_at"],
        )

    @staticmethod
    def _event_record(row: sqlite3.Row) -> QueuedEvent:
        event = json.loads(row["event_json"])
        acknowledgement = (
            json.loads(row["acknowledgement_json"])
            if row["acknowledgement_json"] is not None
            else None
        )
        return QueuedEvent(
            event_id=row["event_id"],
            digest=row["digest"],
            status=row["status"],
            event=_freeze_mapping(event),
            attempt_count=row["attempt_count"],
            available_at=row["available_at"],
            lease_owner=row["lease_owner"],
            lease_token=row["lease_token"],
            lease_expires_at=row["lease_expires_at"],
            error_code=row["error_code"],
            acknowledgement=(
                _freeze_mapping(acknowledgement)
                if acknowledgement is not None
                else None
            ),
            task_digest=row["task_digest"],
            remote_lease_token_digest=row["remote_lease_token_digest"],
            control_lease_token_digest=row["control_lease_token_digest"],
            created_at=row["created_at"],
            updated_at=row["updated_at"],
        )

    @staticmethod
    def _artifact_record(row: sqlite3.Row) -> LocalArtifact:
        receipt = (
            json.loads(row["deletion_receipt_json"])
            if row["deletion_receipt_json"] is not None
            else None
        )
        return LocalArtifact(
            artifact_digest=row["artifact_digest"],
            artifact_ref=row["artifact_ref"],
            artifact_ref_hash=row["artifact_ref_hash"],
            classification=row["classification"],
            retention_until=row["retention_until"],
            task_id=row["task_id"],
            event_id=row["event_id"],
            cleanup_status=row["cleanup_status"],
            attempt_count=row["attempt_count"],
            available_at=row["available_at"],
            lease_owner=row["lease_owner"],
            lease_token=row["lease_token"],
            lease_expires_at=row["lease_expires_at"],
            error_code=row["error_code"],
            deletion_receipt=_freeze_mapping(receipt) if receipt is not None else None,
            created_at=row["created_at"],
            updated_at=row["updated_at"],
        )

    @staticmethod
    def _outcome_record(row: sqlite3.Row) -> QueuedOutcome:
        outcome = json.loads(row["outcome_json"])
        acknowledgement = (
            json.loads(row["acknowledgement_json"])
            if row["acknowledgement_json"] is not None
            else None
        )
        return QueuedOutcome(
            task_id=row["task_id"],
            digest=row["digest"],
            status=row["status"],
            outcome=_freeze_mapping(outcome),
            attempt_count=row["attempt_count"],
            available_at=row["available_at"],
            lease_owner=row["lease_owner"],
            lease_token=row["lease_token"],
            lease_expires_at=row["lease_expires_at"],
            error_code=row["error_code"],
            acknowledgement=(
                _freeze_mapping(acknowledgement)
                if acknowledgement is not None
                else None
            ),
            remote_lease_token_digest=row["remote_lease_token_digest"],
            created_at=row["created_at"],
            updated_at=row["updated_at"],
        )