Video-Player-Controller/controller/store.py

"""Transactional SQLite persistence and non-destructive v1 YAML migration."""

from __future__ import annotations

import contextlib
import datetime as dt
import fcntl
import hashlib
import json
import os
import sqlite3
import tempfile
import threading
import time
import uuid
from pathlib import Path
from typing import Any, Iterable, Iterator, Mapping, Sequence

import yaml

from .config import StorageConfig
from .contracts import SLOT_IDS
from .models import (
    AudioLeaseRecord,
    DeleteToken,
    DownloadJob,
    DownloadReservation,
    MigrationError,
    MigrationReport,
    PlayerObservation,
    RouterObservation,
    SlotGuard,
    SlotProtectedError,
    SlotRecord,
    StaleOperationError,
)


SCHEMA_VERSION = 2
LEGACY_MIGRATION_NAME = "legacy_playlists_yaml_v1"
MAX_LEGACY_YAML_BYTES = 8 * 1024 * 1024


SCHEMA_SQL = """
CREATE TABLE IF NOT EXISTS schema_migrations (
    version INTEGER PRIMARY KEY,
    name TEXT NOT NULL UNIQUE,
    applied_at REAL NOT NULL
);

CREATE TABLE IF NOT EXISTS data_migrations (
    name TEXT PRIMARY KEY,
    applied_at REAL NOT NULL,
    source_path TEXT,
    source_sha256 TEXT,
    backup_path TEXT,
    details_json TEXT NOT NULL DEFAULT '{}'
);

CREATE TABLE IF NOT EXISTS slots (
    slot_number INTEGER PRIMARY KEY CHECK (slot_number BETWEEN 1 AND 10),
    generation INTEGER NOT NULL DEFAULT 0 CHECK (generation >= 0),
    media_state TEXT NOT NULL DEFAULT 'empty'
        CHECK (media_state IN ('empty','queued','downloading','ready','error','deleting')),
    url TEXT,
    youtube_id TEXT,
    title TEXT NOT NULL DEFAULT '',
    item_count INTEGER NOT NULL DEFAULT 0 CHECK (item_count >= 0),
    published_item_count INTEGER NOT NULL DEFAULT 0 CHECK (published_item_count >= 0),
    created_at REAL,
    ready_at REAL,
    expires_at REAL,
    last_error TEXT,
    updated_at REAL NOT NULL,
    CHECK (media_state <> 'empty' OR (url IS NULL AND item_count = 0))
);

CREATE TABLE IF NOT EXISTS slot_items (
    slot_number INTEGER NOT NULL REFERENCES slots(slot_number),
    position INTEGER NOT NULL CHECK (position >= 1),
    generation INTEGER NOT NULL CHECK (generation >= 0),
    title TEXT NOT NULL,
    relative_path TEXT NOT NULL,
    PRIMARY KEY (slot_number, position)
);

CREATE TABLE IF NOT EXISTS download_jobs (
    job_id TEXT PRIMARY KEY,
    slot_number INTEGER NOT NULL REFERENCES slots(slot_number),
    generation INTEGER NOT NULL CHECK (generation >= 1),
    url TEXT NOT NULL,
    status TEXT NOT NULL
        CHECK (status IN ('reserved','running','succeeded','failed','cancelled','interrupted','stale')),
    submitter TEXT,
    note TEXT,
    requested_at REAL NOT NULL,
    started_at REAL,
    finished_at REAL,
    error TEXT,
    UNIQUE (slot_number, generation)
);

CREATE UNIQUE INDEX IF NOT EXISTS one_live_download_per_slot
ON download_jobs(slot_number)
WHERE status IN ('reserved','running');

CREATE TABLE IF NOT EXISTS slot_guards (
    guard_id TEXT PRIMARY KEY,
    slot_number INTEGER NOT NULL REFERENCES slots(slot_number),
    kind TEXT NOT NULL CHECK (kind IN
        ('player_active','pending_play','restart_uncertain','legacy_uncertain')),
    state TEXT NOT NULL CHECK (state IN ('active','review_required','cleared')),
    created_at REAL NOT NULL,
    review_after REAL,
    cleared_at REAL,
    cleared_by TEXT,
    clear_reason TEXT,
    details_json TEXT NOT NULL DEFAULT '{}'
);

CREATE UNIQUE INDEX IF NOT EXISTS one_live_guard_per_slot
ON slot_guards(slot_number)
WHERE state IN ('active','review_required');

CREATE TABLE IF NOT EXISTS player_observation (
    singleton INTEGER PRIMARY KEY CHECK (singleton = 1),
    availability TEXT NOT NULL CHECK (availability IN ('unknown','online','offline')),
    phase TEXT NOT NULL CHECK (phase IN ('unknown','idle','playing','paused','error')),
    active_slot INTEGER REFERENCES slots(slot_number),
    active_path TEXT,
    paused INTEGER CHECK (paused IS NULL OR paused IN (0,1)),
    generation INTEGER NOT NULL DEFAULT 0,
    reconciliation_state TEXT NOT NULL CHECK (reconciliation_state IN
        ('clear','confirmed','awaiting_evidence','operator_review','admin_cleared','unknown')),
    reconciliation_due_at REAL,
    reconciliation_id TEXT,
    last_status_at REAL,
    last_telemetry_at REAL,
    last_status_json TEXT,
    last_telemetry_json TEXT
);

CREATE TABLE IF NOT EXISTS audio_router_observation (
    singleton INTEGER PRIMARY KEY CHECK (singleton = 1),
    mode TEXT,
    freshness TEXT NOT NULL CHECK (freshness IN ('unknown','current','stale')),
    observed_at REAL,
    last_warning TEXT
);

CREATE TABLE IF NOT EXISTS audio_leases (
    lease_id TEXT PRIMARY KEY,
    operation_id TEXT,
    playback_generation INTEGER,
    state TEXT NOT NULL CHECK (state IN
        ('acquiring','held','unconfirmed','ownership_lost','restore_pending','released','abandoned')),
    previous_mode TEXT,
    target_mode TEXT NOT NULL,
    restoration_mode TEXT,
    restoration_source TEXT NOT NULL CHECK (restoration_source IN
        ('observed_previous','configured_fallback','none')),
    controller_changed INTEGER NOT NULL DEFAULT 0 CHECK (controller_changed IN (0,1)),
    confirmation_deadline REAL,
    created_at REAL NOT NULL,
    acquired_at REAL,
    released_at REAL,
    warning TEXT,
    error TEXT
);

CREATE UNIQUE INDEX IF NOT EXISTS one_live_audio_lease
ON audio_leases((1))
WHERE state IN ('acquiring','held','unconfirmed','restore_pending');

CREATE TABLE IF NOT EXISTS audit_events (
    event_id INTEGER PRIMARY KEY AUTOINCREMENT,
    occurred_at REAL NOT NULL,
    event_type TEXT NOT NULL,
    slot_number INTEGER REFERENCES slots(slot_number),
    actor TEXT,
    details_json TEXT NOT NULL DEFAULT '{}'
);
"""


class _UniqueKeyLoader(yaml.SafeLoader):
    pass


def _construct_unique_mapping(
    loader: _UniqueKeyLoader,
    node: yaml.MappingNode,
    deep: bool = False,
) -> dict[Any, Any]:
    result: dict[Any, Any] = {}
    for key_node, value_node in node.value:
        key = loader.construct_object(key_node, deep=deep)
        try:
            duplicate = key in result
        except TypeError as exc:
            raise MigrationError("legacy YAML contains an unhashable mapping key") from exc
        if duplicate:
            raise MigrationError(f"legacy YAML contains duplicate key {key!r}")
        result[key] = loader.construct_object(value_node, deep=deep)
    return result


_UniqueKeyLoader.add_constructor(
    yaml.resolver.BaseResolver.DEFAULT_MAPPING_TAG,
    _construct_unique_mapping,
)


def _utc_timestamp(value: Any, field: str, warnings: list[str]) -> float | None:
    if value in (None, ""):
        return None
    if isinstance(value, (int, float)) and not isinstance(value, bool):
        return float(value)
    if isinstance(value, dt.datetime):
        if value.tzinfo is None:
            value = value.replace(tzinfo=dt.timezone.utc)
        return value.timestamp()
    if isinstance(value, dt.date):
        return dt.datetime.combine(value, dt.time(), tzinfo=dt.timezone.utc).timestamp()
    if isinstance(value, str):
        try:
            parsed = dt.datetime.fromisoformat(value.replace("Z", "+00:00"))
            if parsed.tzinfo is None:
                parsed = parsed.replace(tzinfo=dt.timezone.utc)
            return parsed.timestamp()
        except ValueError:
            pass
    warnings.append(f"ignored invalid {field}: {value!r}")
    return None


def _coerce_count(value: Any, slot: int, warnings: list[str]) -> int:
    if isinstance(value, bool):
        warnings.append(f"slot {slot}: invalid item_count {value!r}; used 0")
        return 0
    try:
        count = int(value)
    except (TypeError, ValueError, OverflowError):
        count = -1
    if count < 0:
        warnings.append(f"slot {slot}: invalid item_count {value!r}; used 0")
        return 0
    return count


def _row_to_slot(row: sqlite3.Row) -> SlotRecord:
    return SlotRecord(
        slot=row["slot_number"],
        generation=row["generation"],
        media_state=row["media_state"],
        url=row["url"],
        yt_id=row["youtube_id"],
        title=row["title"],
        item_count=row["item_count"],
        published_item_count=row["published_item_count"],
        created_at=row["created_at"],
        ready_at=row["ready_at"],
        expires_at=row["expires_at"],
        last_error=row["last_error"],
        updated_at=row["updated_at"],
    )


class ControllerStore:
    """Thread-safe transactional store.

    Controller v2's coordinator is the intended sole writer.  The re-entrant
    lock also makes direct calls safe in HTTP/tests and SQLite transactions are
    the final cross-connection reservation boundary.
    """

    def __init__(self, config: StorageConfig, *, now=time.time):
        self.config = config
        self.path = config.database_path
        self.legacy_path = config.legacy_yaml_path
        self._now = now
        self._lock = threading.RLock()
        self._connection: sqlite3.Connection | None = None
        self.migration_report = MigrationReport(migrated=False)

    def open(self) -> "ControllerStore":
        with self._lock:
            if self._connection is not None:
                return self
            self.config.data_base.mkdir(parents=True, exist_ok=True)
            lock_path = self.config.data_base / ".controller-store.lock"
            with lock_path.open("a+b") as lock_file:
                fcntl.flock(lock_file.fileno(), fcntl.LOCK_EX)
                try:
                    # Recheck only after taking the process lock. This prevents
                    # two WSGI processes from both staging and replacing the
                    # authoritative database during first boot or migration.
                    if not self.path.exists():
                        if self.legacy_path.exists():
                            self.migration_report = self._migrate_legacy_yaml()
                        else:
                            self._create_new_database()
                    self._connection = self._connect(self.path)
                    self._upgrade_database(self._connection)
                    self._validate_database(self._connection)
                    self._recover_interrupted_jobs()
                except BaseException:
                    if self._connection is not None:
                        self._connection.close()
                        self._connection = None
                    raise
                finally:
                    fcntl.flock(lock_file.fileno(), fcntl.LOCK_UN)
            return self

    def close(self) -> None:
        with self._lock:
            if self._connection is not None:
                self._connection.close()
                self._connection = None

    def __enter__(self) -> "ControllerStore":
        return self.open()

    def __exit__(self, exc_type, exc, traceback) -> None:
        self.close()

    @property
    def connection(self) -> sqlite3.Connection:
        if self._connection is None:
            raise RuntimeError("ControllerStore is not open")
        return self._connection

    @staticmethod
    def _connect(path: Path, *, runtime: bool = True) -> sqlite3.Connection:
        connection = sqlite3.connect(
            path,
            timeout=5.0,
            isolation_level=None,
            check_same_thread=False,
        )
        connection.row_factory = sqlite3.Row
        connection.execute("PRAGMA foreign_keys = ON")
        connection.execute("PRAGMA busy_timeout = 5000")
        connection.execute("PRAGMA synchronous = FULL")
        if runtime:
            connection.execute("PRAGMA journal_mode = WAL")
        else:
            connection.execute("PRAGMA journal_mode = DELETE")
        return connection

    @contextlib.contextmanager
    def _transaction(self) -> Iterator[sqlite3.Connection]:
        with self._lock:
            connection = self.connection
            connection.execute("BEGIN IMMEDIATE")
            try:
                yield connection
            except BaseException:
                connection.rollback()
                raise
            else:
                connection.commit()

    def _initialize_schema(self, connection: sqlite3.Connection) -> None:
        connection.executescript(SCHEMA_SQL)
        now = self._now()
        connection.executemany(
            """
            INSERT OR IGNORE INTO slots
                (slot_number, generation, media_state, updated_at)
            VALUES (?, 0, 'empty', ?)
            """,
            ((slot, now) for slot in SLOT_IDS),
        )
        connection.execute(
            """
            INSERT OR IGNORE INTO player_observation
                (singleton, availability, phase, generation, reconciliation_state)
            VALUES (1, 'unknown', 'unknown', 0, 'clear')
            """
        )
        connection.execute(
            """
            INSERT OR IGNORE INTO audio_router_observation
                (singleton, freshness)
            VALUES (1, 'unknown')
            """
        )
        connection.execute(
            """
            INSERT OR IGNORE INTO schema_migrations(version, name, applied_at)
            VALUES (1, 'controller_v2_initial', ?)
            """,
            (now,),
        )
        connection.execute(
            """
            INSERT OR IGNORE INTO schema_migrations(version, name, applied_at)
            VALUES (2, 'external_playback_reconciliation', ?)
            """,
            (now,),
        )
        connection.execute(f"PRAGMA user_version = {SCHEMA_VERSION}")

    def _upgrade_database(self, connection: sqlite3.Connection) -> None:
        """Apply bounded internal schema upgrades under the process lock."""

        version = connection.execute("PRAGMA user_version").fetchone()[0]
        if version == SCHEMA_VERSION:
            return
        if version != 1:
            raise MigrationError(
                f"unsupported Controller database schema {version}; expected 1 or {SCHEMA_VERSION}"
            )
        connection.execute("BEGIN IMMEDIATE")
        try:
            columns = {
                row["name"]
                for row in connection.execute(
                    "PRAGMA table_info(player_observation)"
                ).fetchall()
            }
            if "reconciliation_id" not in columns:
                connection.execute(
                    "ALTER TABLE player_observation ADD COLUMN reconciliation_id TEXT"
                )
            connection.execute(
                """
                INSERT OR IGNORE INTO schema_migrations(version, name, applied_at)
                VALUES (2, 'external_playback_reconciliation', ?)
                """,
                (self._now(),),
            )
            connection.execute(f"PRAGMA user_version = {SCHEMA_VERSION}")
        except BaseException:
            connection.rollback()
            raise
        else:
            connection.commit()

    def _create_new_database(self) -> None:
        staging = self._staging_path()
        try:
            connection = self._connect(staging, runtime=False)
            try:
                self._initialize_schema(connection)
                connection.execute(
                    """
                    INSERT INTO data_migrations
                        (name, applied_at, details_json)
                    VALUES ('fresh_database_v1', ?, '{}')
                    """,
                    (self._now(),),
                )
                self._validate_database(connection)
            finally:
                connection.close()
            self._fsync_file(staging)
            os.replace(staging, self.path)
            self._fsync_directory(self.path.parent)
        except BaseException:
            staging.unlink(missing_ok=True)
            raise

    def _staging_path(self) -> Path:
        descriptor, raw_path = tempfile.mkstemp(
            prefix=f".{self.path.name}.",
            suffix=".staging",
            dir=self.path.parent,
        )
        os.close(descriptor)
        return Path(raw_path)

    @staticmethod
    def _fsync_file(path: Path) -> None:
        descriptor = os.open(path, os.O_RDONLY)
        try:
            os.fsync(descriptor)
        finally:
            os.close(descriptor)

    @staticmethod
    def _fsync_directory(path: Path) -> None:
        descriptor = os.open(path, os.O_RDONLY | getattr(os, "O_DIRECTORY", 0))
        try:
            os.fsync(descriptor)
        finally:
            os.close(descriptor)

    def _create_backup(self, source: bytes, digest: str) -> Path:
        backup = self.legacy_path.with_name(
            f"{self.legacy_path.name}.pre-v2.{digest}.bak"
        )
        if backup.exists():
            if hashlib.sha256(backup.read_bytes()).hexdigest() != digest:
                raise MigrationError(f"existing migration backup is invalid: {backup}")
            return backup

        descriptor, raw_temp = tempfile.mkstemp(
            prefix=f".{backup.name}.",
            suffix=".tmp",
            dir=backup.parent,
        )
        temp = Path(raw_temp)
        try:
            with os.fdopen(descriptor, "wb") as stream:
                stream.write(source)
                stream.flush()
                os.fsync(stream.fileno())
            try:
                os.link(temp, backup)
            except FileExistsError:
                if hashlib.sha256(backup.read_bytes()).hexdigest() != digest:
                    raise MigrationError(f"existing migration backup is invalid: {backup}")
            self._fsync_directory(backup.parent)
        finally:
            temp.unlink(missing_ok=True)

        if hashlib.sha256(backup.read_bytes()).hexdigest() != digest:
            raise MigrationError("migration backup verification failed")
        return backup

    def _parse_legacy(self, source: bytes) -> tuple[dict[int, Mapping[str, Any]], list[str]]:
        try:
            text = source.decode("utf-8")
        except UnicodeDecodeError as exc:
            raise MigrationError("legacy YAML is not valid UTF-8") from exc
        try:
            raw = yaml.load(text, Loader=_UniqueKeyLoader)
        except (yaml.YAMLError, MigrationError) as exc:
            if isinstance(exc, MigrationError):
                raise
            raise MigrationError(f"legacy YAML cannot be parsed: {exc}") from exc
        if raw is None:
            raw = {}
        if not isinstance(raw, Mapping):
            raise MigrationError("legacy YAML root must be a mapping")
        raw_slots = raw.get("slots", {})
        if not isinstance(raw_slots, Mapping):
            raise MigrationError("legacy YAML slots must be a mapping")

        normalized: dict[int, Mapping[str, Any]] = {}
        original_keys: dict[int, Any] = {}
        for key, value in raw_slots.items():
            if isinstance(key, bool):
                raise MigrationError(f"invalid legacy slot key {key!r}")
            try:
                slot = int(key)
            except (TypeError, ValueError, OverflowError) as exc:
                raise MigrationError(f"invalid legacy slot key {key!r}") from exc
            if slot not in SLOT_IDS:
                raise MigrationError(f"legacy slot {slot} is outside stable range 1..10")
            if slot in normalized:
                raise MigrationError(
                    f"legacy slot keys {original_keys[slot]!r} and {key!r} normalize to {slot}"
                )
            if not isinstance(value, Mapping):
                raise MigrationError(f"legacy slot {slot} must be a mapping")
            embedded = value.get("slot", slot)
            if isinstance(embedded, bool):
                raise MigrationError(f"legacy slot {slot} has invalid embedded identity")
            try:
                embedded_slot = int(embedded)
            except (TypeError, ValueError, OverflowError) as exc:
                raise MigrationError(f"legacy slot {slot} has invalid embedded identity") from exc
            if embedded_slot != slot:
                raise MigrationError(
                    f"legacy slot key {slot} conflicts with embedded slot {embedded_slot}"
                )
            normalized[slot] = value
            original_keys[slot] = key
        return normalized, []

    def _migrate_legacy_yaml(self) -> MigrationReport:
        before = self.legacy_path.stat()
        with self.legacy_path.open("rb") as stream:
            opened = os.fstat(stream.fileno())
            source = stream.read(MAX_LEGACY_YAML_BYTES + 1)
            closed = os.fstat(stream.fileno())
        if len(source) > MAX_LEGACY_YAML_BYTES:
            raise MigrationError("legacy YAML exceeds migration size limit")
        after = self.legacy_path.stat()
        identity_before = (
            before.st_dev,
            before.st_ino,
            before.st_size,
            before.st_mtime_ns,
        )
        identity_opened = (
            opened.st_dev,
            opened.st_ino,
            opened.st_size,
            opened.st_mtime_ns,
        )
        identity_closed = (
            closed.st_dev,
            closed.st_ino,
            closed.st_size,
            closed.st_mtime_ns,
        )
        identity_after = (
            after.st_dev,
            after.st_ino,
            after.st_size,
            after.st_mtime_ns,
        )
        if not (
            identity_before
            == identity_opened
            == identity_closed
            == identity_after
        ):
            raise MigrationError("legacy YAML changed while it was being read")
        digest = hashlib.sha256(source).hexdigest()
        backup = self._create_backup(source, digest)
        slots, warnings = self._parse_legacy(source)
        staging = self._staging_path()
        try:
            connection = self._connect(staging, runtime=False)
            try:
                self._initialize_schema(connection)
                connection.execute("BEGIN IMMEDIATE")
                try:
                    now = self._now()
                    for slot_number, legacy in sorted(slots.items()):
                        self._import_legacy_slot(
                            connection,
                            slot_number,
                            legacy,
                            warnings,
                            now,
                        )
                    details = json.dumps({"warnings": warnings}, sort_keys=True)
                    connection.execute(
                        """
                        INSERT INTO data_migrations
                            (name, applied_at, source_path, source_sha256,
                             backup_path, details_json)
                        VALUES (?, ?, ?, ?, ?, ?)
                        """,
                        (
                            LEGACY_MIGRATION_NAME,
                            now,
                            str(self.legacy_path),
                            digest,
                            str(backup),
                            details,
                        ),
                    )
                    connection.commit()
                except BaseException:
                    connection.rollback()
                    raise
                self._validate_database(connection)
            finally:
                connection.close()
            self._fsync_file(staging)
            os.replace(staging, self.path)
            self._fsync_directory(self.path.parent)
        except BaseException:
            staging.unlink(missing_ok=True)
            raise
        return MigrationReport(
            migrated=True,
            source_path=str(self.legacy_path),
            source_sha256=digest,
            backup_path=str(backup),
            warnings=tuple(warnings),
        )

    def _import_legacy_slot(
        self,
        connection: sqlite3.Connection,
        slot: int,
        legacy: Mapping[str, Any],
        warnings: list[str],
        now: float,
    ) -> None:
        raw_state = str(legacy.get("state", "ready")).strip().lower()
        error: str | None = None
        guarded = raw_state in {"playing", "paused", "busy"}
        interrupted = raw_state in {"queued", "downloading"}
        if raw_state in {"ready", "idle", "finished", "playing", "paused", "busy"}:
            media_state = "ready"
        elif raw_state == "error":
            media_state = "error"
            error = str(legacy.get("last_error") or "legacy download error")
        elif interrupted:
            media_state = "error"
            error = "download interrupted during Controller v2 migration"
            warnings.append(f"slot {slot}: legacy {raw_state} download marked interrupted")
        else:
            media_state = "error"
            error = f"unknown legacy state: {raw_state}"
            warnings.append(f"slot {slot}: unknown legacy state {raw_state!r}")

        url_value = legacy.get("url")
        url = str(url_value) if url_value not in (None, "") else None
        if url is None:
            media_state = "error"
            error = error or "legacy slot has no URL"
            warnings.append(f"slot {slot}: missing URL")
        title = str(legacy.get("title") or "")
        count = _coerce_count(legacy.get("item_count", 0), slot, warnings)
        created_at = _utc_timestamp(legacy.get("created_at"), f"slot {slot} created_at", warnings)
        expires_at = _utc_timestamp(
            legacy.get("expires_at"),
            f"slot {slot} expires_at",
            warnings,
        )
        if media_state == "error" and expires_at is None:
            expires_at = now + self.config.retention_seconds
            warnings.append(
                f"slot {slot}: missing error expiry defaulted to retention window"
            )
        generation = 1
        connection.execute(
            """
            UPDATE slots
            SET generation = ?, media_state = ?, url = ?, youtube_id = ?,
                title = ?, item_count = ?, published_item_count = ?,
                created_at = ?, ready_at = ?, expires_at = ?, last_error = ?,
                updated_at = ?
            WHERE slot_number = ?
            """,
            (
                generation,
                media_state,
                url,
                legacy.get("yt_id"),
                title,
                count,
                count,
                created_at,
                created_at if media_state == "ready" else None,
                expires_at,
                error,
                now,
                slot,
            ),
        )
        if guarded:
            guard_id = f"legacy-{slot}-{generation}"
            connection.execute(
                """
                INSERT INTO slot_guards
                    (guard_id, slot_number, kind, state, created_at,
                     review_after, details_json)
                VALUES (?, ?, 'legacy_uncertain', 'review_required', ?, NULL, ?)
                """,
                (
                    guard_id,
                    slot,
                    now,
                    json.dumps({"legacy_state": raw_state}, sort_keys=True),
                ),
            )
        if interrupted and url is not None:
            connection.execute(
                """
                INSERT INTO download_jobs
                    (job_id, slot_number, generation, url, status,
                     requested_at, finished_at, error)
                VALUES (?, ?, ?, ?, 'interrupted', ?, ?, ?)
                """,
                (
                    f"legacy-interrupted-{slot}-{generation}",
                    slot,
                    generation,
                    url,
                    created_at or now,
                    now,
                    error,
                ),
            )
        connection.execute(
            """
            INSERT INTO audit_events
                (occurred_at, event_type, slot_number, actor, details_json)
            VALUES (?, 'legacy_slot_migrated', ?, 'migration', ?)
            """,
            (now, slot, json.dumps({"legacy_state": raw_state}, sort_keys=True)),
        )

    @staticmethod
    def _validate_database(connection: sqlite3.Connection) -> None:
        version = connection.execute("PRAGMA user_version").fetchone()[0]
        if version != SCHEMA_VERSION:
            raise MigrationError(
                f"unsupported Controller database schema {version}; expected {SCHEMA_VERSION}"
            )
        violations = connection.execute("PRAGMA foreign_key_check").fetchall()
        if violations:
            raise MigrationError(f"Controller database foreign-key failure: {violations!r}")
        integrity = connection.execute("PRAGMA integrity_check").fetchone()[0]
        if integrity != "ok":
            raise MigrationError(f"Controller database integrity failure: {integrity}")
        slots = connection.execute(
            "SELECT slot_number FROM slots ORDER BY slot_number"
        ).fetchall()
        if [row[0] for row in slots] != list(SLOT_IDS):
            raise MigrationError("Controller database must contain stable slots 1..10")

    def _recover_interrupted_jobs(self) -> None:
        with self._transaction() as connection:
            now = self._now()
            live = connection.execute(
                """
                SELECT job_id, slot_number, generation
                FROM download_jobs
                WHERE status IN ('reserved','running')
                """
            ).fetchall()
            for row in live:
                message = "download interrupted by Controller restart"
                connection.execute(
                    """
                    UPDATE download_jobs
                    SET status = 'interrupted', finished_at = ?, error = ?
                    WHERE job_id = ?
                    """,
                    (now, message, row["job_id"]),
                )
                connection.execute(
                    """
                    UPDATE slots
                    SET media_state = 'error', last_error = ?,
                        expires_at = COALESCE(expires_at, ?), updated_at = ?
                    WHERE slot_number = ? AND generation = ?
                    """,
                    (
                        message,
                        now + self.config.retention_seconds,
                        now,
                        row["slot_number"],
                        row["generation"],
                    ),
                )

    def list_slots(self) -> list[SlotRecord]:
        with self._lock:
            rows = self.connection.execute(
                "SELECT * FROM slots ORDER BY slot_number"
            ).fetchall()
            return [_row_to_slot(row) for row in rows]

    def get_slot(self, slot: int) -> SlotRecord:
        if slot not in SLOT_IDS:
            raise ValueError(f"slot must be in 1..10, got {slot!r}")
        with self._lock:
            row = self.connection.execute(
                "SELECT * FROM slots WHERE slot_number = ?", (slot,)
            ).fetchone()
            assert row is not None
            return _row_to_slot(row)

    def get_slot_items(self, slot: int) -> list[tuple[str, str]]:
        with self._lock:
            rows = self.connection.execute(
                """
                SELECT title, relative_path FROM slot_items
                WHERE slot_number = ? ORDER BY position
                """,
                (slot,),
            ).fetchall()
            return [(row["title"], row["relative_path"]) for row in rows]

    def reserve_download(
        self,
        url: str,
        submitter: str | None,
        note: str | None,
        *,
        job_id: str | None = None,
    ) -> DownloadReservation:
        if not isinstance(url, str) or not url.strip():
            raise ValueError("download URL must be non-empty")
        url = url.strip()
        job_id = job_id or uuid.uuid4().hex
        with self._transaction() as connection:
            existing = connection.execute(
                """
                SELECT s.* FROM slots s
                WHERE s.url = ? AND s.media_state <> 'empty'
                ORDER BY s.slot_number LIMIT 1
                """,
                (url,),
            ).fetchone()
            reused = existing is not None
            if existing is not None:
                slot = existing["slot_number"]
                if existing["media_state"] in {"queued", "downloading", "deleting"}:
                    return DownloadReservation(status="slot_busy", slot=slot, reused_slot=True)
                if self._slot_has_live_guard(connection, slot):
                    return DownloadReservation(status="slot_protected", slot=slot, reused_slot=True)
            else:
                free = connection.execute(
                    """
                    SELECT s.slot_number FROM slots s
                    WHERE s.media_state = 'empty'
                      AND NOT EXISTS (
                          SELECT 1 FROM slot_guards g
                          WHERE g.slot_number = s.slot_number
                            AND g.state IN ('active','review_required')
                      )
                    ORDER BY s.slot_number LIMIT 1
                    """
                ).fetchone()
                if free is None:
                    oldest = connection.execute(
                        """
                        SELECT MIN(expires_at) FROM slots
                        WHERE expires_at IS NOT NULL
                        """
                    ).fetchone()[0]
                    return DownloadReservation(
                        status="queue_full",
                        oldest_expires_at=oldest,
                    )
                slot = free["slot_number"]

            current = connection.execute(
                "SELECT generation FROM slots WHERE slot_number = ?", (slot,)
            ).fetchone()
            generation = current["generation"] + 1
            now = self._now()
            connection.execute(
                """
                UPDATE slots
                SET generation = ?, media_state = 'queued', url = ?,
                    created_at = COALESCE(created_at, ?), last_error = NULL,
                    updated_at = ?
                WHERE slot_number = ?
                """,
                (generation, url, now, now, slot),
            )
            connection.execute(
                """
                INSERT INTO download_jobs
                    (job_id, slot_number, generation, url, status,
                     submitter, note, requested_at)
                VALUES (?, ?, ?, ?, 'reserved', ?, ?, ?)
                """,
                (job_id, slot, generation, url, submitter, note, now),
            )
            return DownloadReservation(
                status="reserved",
                slot=slot,
                generation=generation,
                job_id=job_id,
                reused_slot=reused,
            )

    @staticmethod
    def _slot_has_live_guard(connection: sqlite3.Connection, slot: int) -> bool:
        return connection.execute(
            """
            SELECT 1 FROM slot_guards
            WHERE slot_number = ? AND state IN ('active','review_required')
            """,
            (slot,),
        ).fetchone() is not None

    def find_occupied_slot_for_url(self, url: str) -> int | None:
        """Return the slot already holding ``url``, if any.

        A non-``None`` result means ingest for this exact URL would reuse or
        refresh a slot that already holds media, rather than land in a slot
        that is genuinely empty. Callers use this to decide whether an
        ingest request needs fresh Player evidence before proceeding.
        """

        if not isinstance(url, str):
            return None
        url = url.strip()
        if not url:
            return None
        with self._lock:
            row = self.connection.execute(
                """
                SELECT slot_number FROM slots
                WHERE url = ? AND media_state <> 'empty'
                ORDER BY slot_number LIMIT 1
                """,
                (url,),
            ).fetchone()
            return None if row is None else int(row["slot_number"])

    def mark_download_running(self, job_id: str, slot: int, generation: int) -> bool:
        with self._transaction() as connection:
            now = self._now()
            updated = connection.execute(
                """
                UPDATE download_jobs SET status = 'running', started_at = ?
                WHERE job_id = ? AND slot_number = ? AND generation = ?
                  AND status = 'reserved'
                """,
                (now, job_id, slot, generation),
            ).rowcount
            if not updated:
                return False
            slot_updated = connection.execute(
                """
                UPDATE slots SET media_state = 'downloading', updated_at = ?
                WHERE slot_number = ? AND generation = ? AND media_state = 'queued'
                """,
                (now, slot, generation),
            ).rowcount
            if not slot_updated:
                raise StaleOperationError("download reservation no longer owns the slot")
            return True

    def complete_download(
        self,
        job_id: str,
        slot: int,
        generation: int,
        *,
        title: str,
        items: Sequence[tuple[str, str]],
        yt_id: str | None = None,
    ) -> bool:
        with self._transaction() as connection:
            job = connection.execute(
                """
                SELECT status FROM download_jobs
                WHERE job_id = ? AND slot_number = ? AND generation = ?
                """,
                (job_id, slot, generation),
            ).fetchone()
            if job is None or job["status"] not in {"reserved", "running"}:
                return False
            current = connection.execute(
                "SELECT generation FROM slots WHERE slot_number = ?", (slot,)
            ).fetchone()
            if current["generation"] != generation:
                connection.execute(
                    """
                    UPDATE download_jobs SET status = 'stale', finished_at = ?,
                        error = 'stale worker completion'
                    WHERE job_id = ?
                    """,
                    (self._now(), job_id),
                )
                return False
            now = self._now()
            expires = now + self.config.retention_seconds
            connection.execute("DELETE FROM slot_items WHERE slot_number = ?", (slot,))
            connection.executemany(
                """
                INSERT INTO slot_items
                    (slot_number, position, generation, title, relative_path)
                VALUES (?, ?, ?, ?, ?)
                """,
                (
                    (slot, position, generation, item_title, relative_path)
                    for position, (item_title, relative_path) in enumerate(items, start=1)
                ),
            )
            connection.execute(
                """
                UPDATE slots
                SET media_state = 'ready', youtube_id = ?, title = ?,
                    item_count = ?, ready_at = ?, expires_at = ?,
                    last_error = NULL, updated_at = ?
                WHERE slot_number = ? AND generation = ?
                """,
                (yt_id, title, len(items), now, expires, now, slot, generation),
            )
            connection.execute(
                """
                UPDATE download_jobs
                SET status = 'succeeded', finished_at = ?, error = NULL
                WHERE job_id = ?
                """,
                (now, job_id),
            )
            return True

    def fail_download(
        self,
        job_id: str,
        slot: int,
        generation: int,
        error: str,
    ) -> bool:
        with self._transaction() as connection:
            now = self._now()
            updated = connection.execute(
                """
                UPDATE download_jobs
                SET status = 'failed', finished_at = ?, error = ?
                WHERE job_id = ? AND slot_number = ? AND generation = ?
                  AND status IN ('reserved','running')
                """,
                (now, error, job_id, slot, generation),
            ).rowcount
            if not updated:
                return False
            connection.execute(
                """
                UPDATE slots
                SET media_state = 'error', last_error = ?,
                    expires_at = COALESCE(expires_at, ?), updated_at = ?
                WHERE slot_number = ? AND generation = ?
                """,
                (
                    error,
                    now + self.config.retention_seconds,
                    now,
                    slot,
                    generation,
                ),
            )
            return True

    def get_job(self, job_id: str) -> DownloadJob | None:
        with self._lock:
            row = self.connection.execute(
                "SELECT * FROM download_jobs WHERE job_id = ?", (job_id,)
            ).fetchone()
            if row is None:
                return None
            return DownloadJob(
                job_id=row["job_id"],
                slot=row["slot_number"],
                generation=row["generation"],
                url=row["url"],
                status=row["status"],
                submitter=row["submitter"],
                note=row["note"],
                requested_at=row["requested_at"],
                started_at=row["started_at"],
                finished_at=row["finished_at"],
                error=row["error"],
            )

    def protect_slot(
        self,
        slot: int,
        kind: str,
        *,
        guard_id: str | None = None,
        review_after: float | None = None,
        details: Mapping[str, Any] | None = None,
    ) -> SlotGuard:
        if slot not in SLOT_IDS:
            raise ValueError("slot must be in 1..10")
        guard_id = guard_id or uuid.uuid4().hex
        with self._transaction() as connection:
            existing = connection.execute(
                """
                SELECT * FROM slot_guards
                WHERE slot_number = ? AND state IN ('active','review_required')
                """,
                (slot,),
            ).fetchone()
            if existing is not None:
                return self._row_to_guard(existing)
            now = self._now()
            connection.execute(
                """
                INSERT INTO slot_guards
                    (guard_id, slot_number, kind, state, created_at,
                     review_after, details_json)
                VALUES (?, ?, ?, 'active', ?, ?, ?)
                """,
                (
                    guard_id,
                    slot,
                    kind,
                    now,
                    review_after,
                    json.dumps(details or {}, sort_keys=True),
                ),
            )
            row = connection.execute(
                "SELECT * FROM slot_guards WHERE guard_id = ?", (guard_id,)
            ).fetchone()
            assert row is not None
            return self._row_to_guard(row)

    @staticmethod
    def _row_to_guard(row: sqlite3.Row) -> SlotGuard:
        return SlotGuard(
            guard_id=row["guard_id"],
            slot=row["slot_number"],
            kind=row["kind"],
            state=row["state"],
            created_at=row["created_at"],
            review_after=row["review_after"],
            cleared_at=row["cleared_at"],
            cleared_by=row["cleared_by"],
            clear_reason=row["clear_reason"],
        )

    def list_live_guards(self) -> list[SlotGuard]:
        with self._lock:
            rows = self.connection.execute(
                """
                SELECT * FROM slot_guards
                WHERE state IN ('active','review_required')
                ORDER BY slot_number
                """
            ).fetchall()
            return [self._row_to_guard(row) for row in rows]

    def reconcile_active_guard(
        self,
        slot: int,
        *,
        kind: str = "player_active",
        review_after: float | None = None,
        details: Mapping[str, Any] | None = None,
    ) -> SlotGuard:
        """Reconcile guards to one idempotent authoritative guard.

        A pending play for another slot is preserved until its in-memory
        operation is explicitly failed. That closes the brief deletion window
        that otherwise exists between contradictory evidence and reduction of
        the pending operation. Repeated status/telemetry for the same slot
        updates the live row in place instead of appending guard history.
        """
        if slot not in SLOT_IDS:
            raise ValueError("slot must be in 1..10")
        with self._transaction() as connection:
            now = self._now()
            existing = connection.execute(
                """
                SELECT * FROM slot_guards
                WHERE slot_number = ? AND state IN ('active','review_required')
                """,
                (slot,),
            ).fetchone()
            connection.execute(
                """
                UPDATE slot_guards
                SET state = 'cleared', cleared_at = ?, cleared_by = 'player',
                    clear_reason = 'reconciled by fresh active evidence'
                WHERE state IN ('active','review_required')
                  AND slot_number <> ? AND kind <> 'pending_play'
                """,
                (now, slot),
            )
            encoded = json.dumps(details or {}, sort_keys=True)
            if existing is None:
                guard_id = uuid.uuid4().hex
                connection.execute(
                    """
                    INSERT INTO slot_guards
                        (guard_id, slot_number, kind, state, created_at,
                         review_after, details_json)
                    VALUES (?, ?, ?, 'active', ?, ?, ?)
                    """,
                    (guard_id, slot, kind, now, review_after, encoded),
                )
            else:
                guard_id = existing["guard_id"]
                connection.execute(
                    """
                    UPDATE slot_guards
                    SET kind = ?, state = 'active', review_after = ?,
                        cleared_at = NULL, cleared_by = NULL,
                        clear_reason = NULL, details_json = ?
                    WHERE guard_id = ?
                    """,
                    (kind, review_after, encoded, guard_id),
                )
            row = connection.execute(
                "SELECT * FROM slot_guards WHERE guard_id = ?", (guard_id,)
            ).fetchone()
            assert row is not None
            return self._row_to_guard(row)

    def mark_external_uncertain(
        self,
        path: str,
        *,
        review_after: float | None,
        availability: str = "unknown",
        details: Mapping[str, Any] | None = None,
        status_payload: Mapping[str, Any] | None = None,
    ) -> PlayerObservation:
        """Persist an exact review token for path-only playback uncertainty."""

        if not path:
            raise ValueError("external playback reconciliation requires a path")
        now = self._now()
        with self._transaction() as connection:
            row = connection.execute(
                "SELECT * FROM player_observation WHERE singleton = 1"
            ).fetchone()
            assert row is not None
            same_review = (
                row["active_slot"] is None
                and row["active_path"] == path
                and row["reconciliation_id"] is not None
                and row["reconciliation_state"]
                in {"awaiting_evidence", "operator_review"}
            )
            review_id = row["reconciliation_id"] if same_review else uuid.uuid4().hex
            deadline = row["reconciliation_due_at"] if same_review else None
            if deadline is None:
                deadline = review_after
            elif review_after is not None:
                deadline = min(deadline, review_after)
            state = (
                "operator_review"
                if same_review and row["reconciliation_state"] == "operator_review"
                else "awaiting_evidence"
            )
            connection.execute(
                """
                UPDATE player_observation
                SET availability = ?, phase = 'unknown', active_slot = NULL,
                    active_path = ?, paused = NULL, generation = generation + 1,
                    reconciliation_state = ?, reconciliation_due_at = ?,
                    reconciliation_id = ?,
                    last_status_at = CASE WHEN ? IS NULL THEN last_status_at ELSE ? END,
                    last_status_json = COALESCE(?, last_status_json)
                WHERE singleton = 1
                """,
                (
                    availability,
                    path,
                    state,
                    deadline,
                    review_id,
                    None if status_payload is None else 1,
                    now,
                    None
                    if status_payload is None
                    else json.dumps(status_payload, sort_keys=True),
                ),
            )
            if not same_review:
                connection.execute(
                    """
                    INSERT INTO audit_events
                        (occurred_at, event_type, actor, details_json)
                    VALUES (?, 'external_playback_uncertain', 'controller', ?)
                    """,
                    (
                        now,
                        json.dumps(
                            {
                                "path": path,
                                "review_id": review_id,
                                **dict(details or {}),
                            },
                            sort_keys=True,
                        ),
                    ),
                )
        return self.player_observation()

    def mark_external_reconciliation_due(
        self, now: float | None = None
    ) -> tuple[str, str] | None:
        """Move path-only uncertainty to operator review without clearing it."""

        now = self._now() if now is None else now
        with self._transaction() as connection:
            row = connection.execute(
                """
                SELECT * FROM player_observation
                WHERE singleton = 1 AND active_slot IS NULL
                  AND active_path IS NOT NULL AND reconciliation_id IS NOT NULL
                  AND reconciliation_state = 'awaiting_evidence'
                  AND reconciliation_due_at IS NOT NULL
                  AND reconciliation_due_at <= ?
                """,
                (now,),
            ).fetchone()
            if row is None:
                return None
            connection.execute(
                """
                UPDATE player_observation
                SET phase = 'unknown', reconciliation_state = 'operator_review',
                    generation = generation + 1
                WHERE singleton = 1
                """
            )
            connection.execute(
                """
                INSERT INTO audit_events
                    (occurred_at, event_type, actor, details_json)
                VALUES (?, 'external_playback_operator_review', 'controller', ?)
                """,
                (
                    now,
                    json.dumps(
                        {
                            "path": row["active_path"],
                            "review_id": row["reconciliation_id"],
                        },
                        sort_keys=True,
                    ),
                ),
            )
            return row["reconciliation_id"], row["active_path"]

    def clear_external_reconciliation(
        self,
        review_id: str,
        *,
        actor: str,
        reason: str,
    ) -> bool:
        """Administratively clear one exact path-only uncertainty token."""

        if not actor.strip() or not reason.strip():
            raise ValueError("administrative reconciliation requires actor and reason")
        with self._transaction() as connection:
            row = connection.execute(
                """
                SELECT * FROM player_observation
                WHERE singleton = 1 AND active_slot IS NULL
                  AND active_path IS NOT NULL AND reconciliation_id = ?
                  AND reconciliation_state IN ('awaiting_evidence','operator_review')
                """,
                (review_id,),
            ).fetchone()
            if row is None:
                return False
            now = self._now()
            connection.execute(
                """
                UPDATE player_observation
                SET phase = 'unknown', active_path = NULL, paused = NULL,
                    generation = generation + 1,
                    reconciliation_state = 'admin_cleared',
                    reconciliation_due_at = NULL, reconciliation_id = NULL
                WHERE singleton = 1
                """
            )
            connection.execute(
                """
                INSERT INTO audit_events
                    (occurred_at, event_type, actor, details_json)
                VALUES (?, 'external_playback_admin_cleared', ?, ?)
                """,
                (
                    now,
                    actor.strip(),
                    json.dumps(
                        {
                            "path": row["active_path"],
                            "reason": reason.strip(),
                            "review_id": review_id,
                        },
                        sort_keys=True,
                    ),
                ),
            )
            return True

    def cancel_external_intent(self, review_id: str, *, reason: str) -> bool:
        """Clear one pre-evidence external intent after a definite failure."""

        if not reason.strip():
            raise ValueError("external intent cancellation requires a reason")
        with self._transaction() as connection:
            row = connection.execute(
                """
                SELECT * FROM player_observation
                WHERE singleton = 1 AND active_slot IS NULL
                  AND active_path IS NOT NULL AND reconciliation_id = ?
                  AND reconciliation_state = 'awaiting_evidence'
                """,
                (review_id,),
            ).fetchone()
            if row is None:
                return False
            now = self._now()
            connection.execute(
                """
                UPDATE player_observation
                SET phase = 'unknown', active_path = NULL, paused = NULL,
                    generation = generation + 1,
                    reconciliation_state = 'clear',
                    reconciliation_due_at = NULL, reconciliation_id = NULL
                WHERE singleton = 1
                """
            )
            connection.execute(
                """
                INSERT INTO audit_events
                    (occurred_at, event_type, actor, details_json)
                VALUES (?, 'external_playback_intent_cancelled', 'controller', ?)
                """,
                (
                    now,
                    json.dumps(
                        {
                            "path": row["active_path"],
                            "reason": reason.strip(),
                            "review_id": review_id,
                        },
                        sort_keys=True,
                    ),
                ),
            )
            return True

    def reconcile_external_active(self) -> None:
        """Clear stale slot ownership when fresh evidence is path-only."""

        with self._transaction() as connection:
            now = self._now()
            connection.execute(
                """
                UPDATE slot_guards
                SET state = 'cleared', cleared_at = ?, cleared_by = 'player',
                    clear_reason = 'fresh Player evidence is outside managed slots'
                WHERE state IN ('active','review_required')
                  AND kind <> 'pending_play'
                """,
                (now,),
            )

    def mark_slot_uncertain(
        self,
        slot: int,
        *,
        review_after: float | None,
        details: Mapping[str, Any] | None = None,
    ) -> SlotGuard:
        """Conservatively convert one live guard to restart uncertainty.

        Other guarded slots are untouched and an existing review deadline is
        never moved later by duplicate offline/disconnect evidence.
        """

        if slot not in SLOT_IDS:
            raise ValueError("slot must be in 1..10")
        with self._transaction() as connection:
            now = self._now()
            row = connection.execute(
                """
                SELECT * FROM slot_guards
                WHERE slot_number = ? AND state IN ('active','review_required')
                """,
                (slot,),
            ).fetchone()
            encoded = json.dumps(details or {}, sort_keys=True)
            if row is None:
                guard_id = uuid.uuid4().hex
                connection.execute(
                    """
                    INSERT INTO slot_guards
                        (guard_id, slot_number, kind, state, created_at,
                         review_after, details_json)
                    VALUES (?, ?, 'restart_uncertain', 'active', ?, ?, ?)
                    """,
                    (guard_id, slot, now, review_after, encoded),
                )
            else:
                guard_id = row["guard_id"]
                # Legacy uncertainty is already an explicit operator review;
                # retain that identity/state. Otherwise make the old active or
                # pending guard administratively reconcilable.
                kind = (
                    "legacy_uncertain"
                    if row["kind"] == "legacy_uncertain"
                    else "restart_uncertain"
                )
                deadline = row["review_after"]
                if deadline is None:
                    deadline = review_after
                elif review_after is not None:
                    deadline = min(deadline, review_after)
                connection.execute(
                    """
                    UPDATE slot_guards
                    SET kind = ?, review_after = ?, details_json = ?
                    WHERE guard_id = ?
                    """,
                    (kind, deadline, encoded, guard_id),
                )
            updated = connection.execute(
                "SELECT * FROM slot_guards WHERE guard_id = ?", (guard_id,)
            ).fetchone()
            assert updated is not None
            return self._row_to_guard(updated)

    def clear_playback_guards(
        self,
        *,
        reason: str,
        slot: int | None = None,
        kind: str | None = None,
    ) -> int:
        clauses = ["state IN ('active','review_required')"]
        parameters: list[Any] = []
        if slot is not None:
            clauses.append("slot_number = ?")
            parameters.append(slot)
        if kind is not None:
            clauses.append("kind = ?")
            parameters.append(kind)
        with self._transaction() as connection:
            now = self._now()
            result = connection.execute(
                f"""
                UPDATE slot_guards
                SET state = 'cleared', cleared_at = ?, cleared_by = 'controller',
                    clear_reason = ?
                WHERE {' AND '.join(clauses)}
                """,
                (now, reason, *parameters),
            )
            return result.rowcount

    def mark_due_guards_for_review(self, now: float | None = None) -> list[SlotGuard]:
        now = self._now() if now is None else now
        with self._transaction() as connection:
            due_rows = connection.execute(
                """
                SELECT * FROM slot_guards
                WHERE state = 'active' AND review_after IS NOT NULL
                  AND review_after <= ?
                ORDER BY slot_number
                """,
                (now,),
            ).fetchall()
            if not due_rows:
                return []
            guard_ids = [row["guard_id"] for row in due_rows]
            placeholders = ",".join("?" for _ in guard_ids)
            connection.execute(
                """
                UPDATE slot_guards SET state = 'review_required'
                WHERE guard_id IN ("""
                + placeholders
                + ")",
                guard_ids,
            )
            rows = connection.execute(
                "SELECT * FROM slot_guards WHERE guard_id IN ("
                + placeholders
                + ") ORDER BY slot_number",
                guard_ids,
            ).fetchall()
            return [self._row_to_guard(row) for row in rows]

    def clear_guard(
        self,
        slot: int,
        guard_id: str,
        *,
        actor: str,
        reason: str,
    ) -> bool:
        if not actor.strip() or not reason.strip():
            raise ValueError("administrative guard clear requires actor and reason")
        with self._transaction() as connection:
            now = self._now()
            updated = connection.execute(
                """
                UPDATE slot_guards
                SET state = 'cleared', cleared_at = ?, cleared_by = ?,
                    clear_reason = ?
                WHERE guard_id = ? AND slot_number = ?
                  AND state IN ('active','review_required')
                """,
                (now, actor.strip(), reason.strip(), guard_id, slot),
            ).rowcount
            if not updated:
                return False
            connection.execute(
                """
                UPDATE player_observation
                SET phase = 'unknown', active_slot = NULL, active_path = NULL,
                    paused = NULL, reconciliation_state = 'admin_cleared',
                    reconciliation_due_at = NULL, reconciliation_id = NULL,
                    generation = generation + 1
                WHERE singleton = 1 AND active_slot = ?
                  AND (reconciliation_id IS NULL OR reconciliation_id = ?)
                """,
                (slot, guard_id),
            )
            connection.execute(
                """
                INSERT INTO audit_events
                    (occurred_at, event_type, slot_number, actor, details_json)
                VALUES (?, 'playback_guard_admin_cleared', ?, ?, ?)
                """,
                (now, slot, actor.strip(), json.dumps({"reason": reason.strip()})),
            )
            return True

    def begin_delete(
        self,
        slot: int,
        *,
        reason: str,
        expected_url: str | None = None,
    ) -> DeleteToken:
        if slot not in SLOT_IDS:
            raise ValueError("slot must be in 1..10")
        with self._transaction() as connection:
            row = connection.execute(
                "SELECT * FROM slots WHERE slot_number = ?", (slot,)
            ).fetchone()
            assert row is not None
            if row["media_state"] == "empty":
                raise StaleOperationError(f"slot {slot} is already empty")
            if expected_url is not None and row["url"] != expected_url:
                raise StaleOperationError(f"slot {slot} URL no longer matches")
            if row["media_state"] in {"queued", "downloading", "deleting"}:
                raise SlotProtectedError(f"slot {slot} is {row['media_state']}")
            if self._slot_has_live_guard(connection, slot):
                raise SlotProtectedError(f"slot {slot} has active playback protection")
            live_job = connection.execute(
                """
                SELECT 1 FROM download_jobs
                WHERE slot_number = ? AND status IN ('reserved','running')
                """,
                (slot,),
            ).fetchone()
            if live_job is not None:
                raise SlotProtectedError(f"slot {slot} has an active download")
            generation = row["generation"] + 1
            now = self._now()
            connection.execute(
                """
                UPDATE slots
                SET generation = ?, media_state = 'deleting', updated_at = ?
                WHERE slot_number = ?
                """,
                (generation, now, slot),
            )
            return DeleteToken(slot=slot, generation=generation, reason=reason)

    def finish_delete(self, token: DeleteToken) -> bool:
        with self._transaction() as connection:
            now = self._now()
            updated = connection.execute(
                """
                UPDATE slots
                SET media_state = 'empty', url = NULL, youtube_id = NULL,
                    title = '', item_count = 0, created_at = NULL,
                    ready_at = NULL, expires_at = NULL, last_error = NULL,
                    updated_at = ?
                WHERE slot_number = ? AND generation = ?
                  AND media_state = 'deleting'
                """,
                (now, token.slot, token.generation),
            ).rowcount
            if not updated:
                return False
            connection.execute(
                "DELETE FROM slot_items WHERE slot_number = ?", (token.slot,)
            )
            connection.execute(
                """
                INSERT INTO audit_events
                    (occurred_at, event_type, slot_number, actor, details_json)
                VALUES (?, 'slot_deleted', ?, 'controller', ?)
                """,
                (now, token.slot, json.dumps({"reason": token.reason})),
            )
            return True

    def pending_deletions(self) -> list[DeleteToken]:
        with self._lock:
            rows = self.connection.execute(
                """
                SELECT slot_number, generation FROM slots
                WHERE media_state = 'deleting' ORDER BY slot_number
                """
            ).fetchall()
            return [
                DeleteToken(
                    slot=row["slot_number"],
                    generation=row["generation"],
                    reason="recovered_interrupted_deletion",
                )
                for row in rows
            ]

    def begin_expired_deletions(self, now: float | None = None) -> list[DeleteToken]:
        now = self._now() if now is None else now
        tokens: list[DeleteToken] = []
        for slot in SLOT_IDS:
            record = self.get_slot(slot)
            if (
                record.media_state in {"ready", "error"}
                and record.expires_at is not None
                and record.expires_at <= now
            ):
                try:
                    tokens.append(self.begin_delete(slot, reason="expired"))
                except SlotProtectedError:
                    continue
        return tokens

    def player_observation(self) -> PlayerObservation:
        with self._lock:
            row = self.connection.execute(
                "SELECT * FROM player_observation WHERE singleton = 1"
            ).fetchone()
            assert row is not None
            return PlayerObservation(
                availability=row["availability"],
                phase=row["phase"],
                active_slot=row["active_slot"],
                active_path=row["active_path"],
                paused=None if row["paused"] is None else bool(row["paused"]),
                generation=row["generation"],
                reconciliation_state=row["reconciliation_state"],
                reconciliation_due_at=row["reconciliation_due_at"],
                reconciliation_id=row["reconciliation_id"],
                last_status_at=row["last_status_at"],
                last_telemetry_at=row["last_telemetry_at"],
            )

    def set_player_observation(
        self,
        *,
        availability: str,
        phase: str,
        active_slot: int | None,
        active_path: str | None,
        paused: bool | None,
        reconciliation_state: str,
        reconciliation_due_at: float | None,
        reconciliation_id: str | None,
        status_payload: Mapping[str, Any] | None = None,
        telemetry_payload: Mapping[str, Any] | None = None,
    ) -> PlayerObservation:
        now = self._now()
        with self._transaction() as connection:
            connection.execute(
                """
                UPDATE player_observation
                SET availability = ?, phase = ?, active_slot = ?,
                    active_path = ?, paused = ?, generation = generation + 1,
                    reconciliation_state = ?, reconciliation_due_at = ?,
                    reconciliation_id = ?,
                    last_status_at = CASE WHEN ? IS NULL THEN last_status_at ELSE ? END,
                    last_telemetry_at = CASE WHEN ? IS NULL THEN last_telemetry_at ELSE ? END,
                    last_status_json = COALESCE(?, last_status_json),
                    last_telemetry_json = COALESCE(?, last_telemetry_json)
                WHERE singleton = 1
                """,
                (
                    availability,
                    phase,
                    active_slot,
                    active_path,
                    None if paused is None else int(paused),
                    reconciliation_state,
                    reconciliation_due_at,
                    reconciliation_id,
                    None if status_payload is None else 1,
                    now,
                    None if telemetry_payload is None else 1,
                    now,
                    None if status_payload is None else json.dumps(status_payload, sort_keys=True),
                    None if telemetry_payload is None else json.dumps(telemetry_payload, sort_keys=True),
                ),
            )
        return self.player_observation()

    def mark_restart_uncertain(self) -> SlotGuard | None:
        """Convert every pre-restart active/pending guard conservatively.

        The operation that owned a pending guard was process-local, so after a
        restart both a prior active Player and a command awaiting evidence are
        uncertain. Deadlines are filled only when absent and therefore cannot
        be postponed by restart churn.
        """

        with self._transaction() as connection:
            now = self._now()
            requested_due = None
            if self.config.uncertain_review_after_seconds > 0:
                requested_due = now + self.config.uncertain_review_after_seconds
            observation = connection.execute(
                "SELECT * FROM player_observation WHERE singleton = 1"
            ).fetchone()
            assert observation is not None
            rows = connection.execute(
                """
                SELECT * FROM slot_guards
                WHERE state IN ('active','review_required')
                  AND kind IN (
                      'player_active','pending_play','restart_uncertain','legacy_uncertain'
                  )
                ORDER BY created_at, slot_number
                """
            ).fetchall()

            active_slot = observation["active_slot"]
            active_path = observation["active_path"]
            if active_slot is not None and not any(
                row["slot_number"] == active_slot for row in rows
            ):
                guard_id = uuid.uuid4().hex
                connection.execute(
                    """
                    INSERT INTO slot_guards
                        (guard_id, slot_number, kind, state, created_at,
                         review_after, details_json)
                    VALUES (?, ?, 'restart_uncertain', 'active', ?, ?, ?)
                    """,
                    (
                        guard_id,
                        active_slot,
                        now,
                        requested_due,
                        json.dumps(
                            {"previous_phase": observation["phase"]},
                            sort_keys=True,
                        ),
                    ),
                )
                rows = connection.execute(
                    "SELECT * FROM slot_guards WHERE guard_id = ?", (guard_id,)
                ).fetchall() + rows

            converted: list[sqlite3.Row] = []
            for row in rows:
                if row["kind"] == "legacy_uncertain":
                    converted.append(row)
                    continue
                deadline = row["review_after"]
                if deadline is None:
                    deadline = requested_due
                connection.execute(
                    """
                    UPDATE slot_guards
                    SET kind = 'restart_uncertain', review_after = ?
                    WHERE guard_id = ?
                    """,
                    (deadline, row["guard_id"]),
                )
                converted_row = connection.execute(
                    "SELECT * FROM slot_guards WHERE guard_id = ?",
                    (row["guard_id"],),
                ).fetchone()
                assert converted_row is not None
                converted.append(converted_row)

            if active_slot is None and active_path is None and len(converted) == 1:
                active_slot = converted[0]["slot_number"]
                active_path = None
            active_guard = next(
                (
                    row
                    for row in converted
                    if row["slot_number"] == active_slot
                ),
                (
                    converted[0]
                    if active_path is None and len(converted) == 1
                    else None
                ),
            )
            external_review_id = None
            external_due = None
            external_state = None
            if active_slot is None and active_path is not None:
                existing_external = (
                    observation["reconciliation_id"] is not None
                    and observation["reconciliation_state"]
                    in {"awaiting_evidence", "operator_review"}
                )
                external_review_id = (
                    observation["reconciliation_id"]
                    if existing_external
                    else uuid.uuid4().hex
                )
                external_due = (
                    observation["reconciliation_due_at"]
                    if existing_external
                    else requested_due
                )
                external_state = (
                    "operator_review"
                    if existing_external
                    and observation["reconciliation_state"] == "operator_review"
                    else "awaiting_evidence"
                )

            if active_slot is not None or active_path is not None or converted:
                reconciliation_state = "awaiting_evidence"
                reconciliation_id = None
                reconciliation_due_at = None
                if active_guard is not None:
                    reconciliation_id = active_guard["guard_id"]
                    reconciliation_due_at = active_guard["review_after"]
                    if active_guard["state"] == "review_required":
                        reconciliation_state = "operator_review"
                elif external_review_id is not None:
                    reconciliation_id = external_review_id
                    reconciliation_due_at = external_due
                    reconciliation_state = external_state
                connection.execute(
                    """
                    UPDATE player_observation
                    SET availability = 'unknown', phase = 'unknown',
                        active_slot = ?, active_path = ?, paused = NULL,
                        generation = generation + 1,
                        reconciliation_state = ?,
                        reconciliation_due_at = ?, reconciliation_id = ?
                    WHERE singleton = 1
                    """,
                    (
                        active_slot,
                        active_path,
                        reconciliation_state,
                        reconciliation_due_at,
                        reconciliation_id,
                    ),
                )
                if external_review_id is not None and not existing_external:
                    connection.execute(
                        """
                        INSERT INTO audit_events
                            (occurred_at, event_type, actor, details_json)
                        VALUES (?, 'external_playback_uncertain', 'controller', ?)
                        """,
                        (
                            now,
                            json.dumps(
                                {
                                    "path": active_path,
                                    "review_id": external_review_id,
                                    "reason": "Controller restart",
                                },
                                sort_keys=True,
                            ),
                        ),
                    )
            return (
                None
                if active_guard is None
                else self._row_to_guard(active_guard)
            )

    def set_published_item_count(self, slot: int, count: int) -> None:
        with self._transaction() as connection:
            connection.execute(
                """
                UPDATE slots SET published_item_count = ?, updated_at = ?
                WHERE slot_number = ?
                """,
                (count, self._now(), slot),
            )

    def router_observation(self) -> RouterObservation:
        with self._lock:
            row = self.connection.execute(
                "SELECT * FROM audio_router_observation WHERE singleton = 1"
            ).fetchone()
            assert row is not None
            return RouterObservation(
                mode=row["mode"],
                freshness=row["freshness"],
                observed_at=row["observed_at"],
                last_warning=row["last_warning"],
            )

    def set_router_observation(
        self,
        *,
        mode: str | None,
        freshness: str,
        warning: str | None = None,
    ) -> RouterObservation:
        with self._transaction() as connection:
            connection.execute(
                """
                UPDATE audio_router_observation
                SET mode = ?, freshness = ?, observed_at = ?, last_warning = ?
                WHERE singleton = 1
                """,
                (mode, freshness, self._now(), warning),
            )
        return self.router_observation()

    @staticmethod
    def _row_to_audio_lease(row: sqlite3.Row) -> AudioLeaseRecord:
        return AudioLeaseRecord(
            lease_id=row["lease_id"],
            operation_id=row["operation_id"],
            playback_generation=row["playback_generation"],
            state=row["state"],
            previous_mode=row["previous_mode"],
            target_mode=row["target_mode"],
            restoration_mode=row["restoration_mode"],
            restoration_source=row["restoration_source"],
            controller_changed=bool(row["controller_changed"]),
            confirmation_deadline=row["confirmation_deadline"],
            created_at=row["created_at"],
            acquired_at=row["acquired_at"],
            released_at=row["released_at"],
            warning=row["warning"],
            error=row["error"],
        )

    def create_audio_lease(
        self,
        *,
        lease_id: str,
        operation_id: str,
        state: str,
        previous_mode: str | None,
        target_mode: str,
        restoration_mode: str | None,
        restoration_source: str,
        controller_changed: bool,
        confirmation_deadline: float | None,
        warning: str | None = None,
        error: str | None = None,
    ) -> AudioLeaseRecord:
        with self._transaction() as connection:
            now = self._now()
            connection.execute(
                """
                INSERT INTO audio_leases
                    (lease_id, operation_id, state, previous_mode, target_mode,
                     restoration_mode, restoration_source, controller_changed,
                     confirmation_deadline, created_at, warning, error)
                VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
                """,
                (
                    lease_id,
                    operation_id,
                    state,
                    previous_mode,
                    target_mode,
                    restoration_mode,
                    restoration_source,
                    int(controller_changed),
                    confirmation_deadline,
                    now,
                    warning,
                    error,
                ),
            )
            row = connection.execute(
                "SELECT * FROM audio_leases WHERE lease_id = ?", (lease_id,)
            ).fetchone()
            assert row is not None
            return self._row_to_audio_lease(row)

    def update_audio_lease(
        self,
        lease_id: str,
        *,
        state: str,
        confirmation_deadline: float | None = None,
        warning: str | None = None,
        error: str | None = None,
        acquired: bool = False,
        released: bool = False,
    ) -> AudioLeaseRecord | None:
        with self._transaction() as connection:
            now = self._now()
            updated = connection.execute(
                """
                UPDATE audio_leases
                SET state = ?, confirmation_deadline = ?,
                    warning = COALESCE(?, warning), error = COALESCE(?, error),
                    acquired_at = CASE WHEN ? THEN COALESCE(acquired_at, ?) ELSE acquired_at END,
                    released_at = CASE WHEN ? THEN ? ELSE released_at END
                WHERE lease_id = ?
                """,
                (
                    state,
                    confirmation_deadline,
                    warning,
                    error,
                    int(acquired),
                    now,
                    int(released),
                    now,
                    lease_id,
                ),
            ).rowcount
            if not updated:
                return None
            row = connection.execute(
                "SELECT * FROM audio_leases WHERE lease_id = ?", (lease_id,)
            ).fetchone()
            assert row is not None
            return self._row_to_audio_lease(row)

    def get_audio_lease(self, lease_id: str) -> AudioLeaseRecord | None:
        with self._lock:
            row = self.connection.execute(
                "SELECT * FROM audio_leases WHERE lease_id = ?", (lease_id,)
            ).fetchone()
            return None if row is None else self._row_to_audio_lease(row)

    def live_audio_lease(self) -> AudioLeaseRecord | None:
        with self._lock:
            row = self.connection.execute(
                """
                SELECT * FROM audio_leases
                WHERE state IN ('acquiring','held','unconfirmed','restore_pending')
                ORDER BY created_at DESC LIMIT 1
                """
            ).fetchone()
            return None if row is None else self._row_to_audio_lease(row)

    def abandon_audio_leases_on_restart(self) -> int:
        with self._transaction() as connection:
            now = self._now()
            return connection.execute(
                """
                UPDATE audio_leases
                SET state = 'abandoned', released_at = ?,
                    warning = 'Controller restarted; audio ownership was relinquished'
                WHERE state IN ('acquiring','held','unconfirmed','restore_pending')
                """,
                (now,),
            ).rowcount

    def record_audit_event(
        self,
        event_type: str,
        *,
        slot: int | None = None,
        actor: str | None = None,
        details: Mapping[str, Any] | None = None,
    ) -> int:
        with self._transaction() as connection:
            cursor = connection.execute(
                """
                INSERT INTO audit_events
                    (occurred_at, event_type, slot_number, actor, details_json)
                VALUES (?, ?, ?, ?, ?)
                """,
                (
                    self._now(),
                    event_type,
                    slot,
                    actor,
                    json.dumps(details or {}, sort_keys=True),
                ),
            )
            return int(cursor.lastrowid)