pi-agent-config/scenarios/curator/backend/curator/db.py

from __future__ import annotations

import json
import shutil
import sqlite3
import unicodedata
from contextlib import closing, contextmanager
from dataclasses import dataclass
from datetime import UTC, datetime, timedelta
from pathlib import Path
from typing import Any, Callable, Iterator


SCHEMA = """
PRAGMA foreign_keys = ON;

CREATE TABLE IF NOT EXISTS works (
    id INTEGER PRIMARY KEY,
    title TEXT NOT NULL,
    author TEXT NOT NULL DEFAULT '',
    normalized_title TEXT NOT NULL,
    normalized_author TEXT NOT NULL,
    media_type TEXT NOT NULL DEFAULT 'book',
    status TEXT NOT NULL DEFAULT 'owned',
    created_at TEXT NOT NULL,
    updated_at TEXT NOT NULL,
    UNIQUE(media_type, normalized_title, normalized_author)
);

CREATE TABLE IF NOT EXISTS editions (
    id INTEGER PRIMARY KEY,
    work_id INTEGER NOT NULL REFERENCES works(id) ON DELETE CASCADE,
    language TEXT NOT NULL DEFAULT 'und',
    variant TEXT NOT NULL DEFAULT 'original',
    isbn TEXT NOT NULL DEFAULT '',
    publisher TEXT NOT NULL DEFAULT '',
    published_year INTEGER,
    source TEXT NOT NULL DEFAULT 'upload',
    created_at TEXT NOT NULL,
    UNIQUE(work_id, language, variant, isbn)
);

CREATE TABLE IF NOT EXISTS assets (
    id INTEGER PRIMARY KEY,
    edition_id INTEGER NOT NULL REFERENCES editions(id) ON DELETE CASCADE,
    format TEXT NOT NULL,
    filename TEXT NOT NULL,
    path TEXT NOT NULL UNIQUE,
    sha256 TEXT NOT NULL UNIQUE,
    size_bytes INTEGER NOT NULL,
    mime_type TEXT NOT NULL,
    metadata_json TEXT NOT NULL DEFAULT '{}',
    created_at TEXT NOT NULL
);

CREATE TABLE IF NOT EXISTS wanted_books (
    id INTEGER PRIMARY KEY,
    query TEXT NOT NULL,
    title TEXT NOT NULL DEFAULT '',
    author TEXT NOT NULL DEFAULT '',
    language TEXT NOT NULL DEFAULT 'und',
    status TEXT NOT NULL DEFAULT 'wanted',
    next_check_at TEXT,
    last_checked_at TEXT,
    created_at TEXT NOT NULL
);

CREATE TABLE IF NOT EXISTS inbox_items (
    id INTEGER PRIMARY KEY,
    source_url TEXT NOT NULL UNIQUE,
    media_type TEXT NOT NULL DEFAULT 'unknown',
    title TEXT NOT NULL DEFAULT '',
    creator TEXT NOT NULL DEFAULT '',
    recommendation TEXT NOT NULL DEFAULT 'unknown',
    summary TEXT NOT NULL DEFAULT '',
    reasons_json TEXT NOT NULL DEFAULT '[]',
    suggested_action TEXT NOT NULL DEFAULT '',
    metadata_json TEXT NOT NULL DEFAULT '{}',
    status TEXT NOT NULL DEFAULT 'evaluated',
    created_at TEXT NOT NULL,
    updated_at TEXT NOT NULL
);

CREATE TABLE IF NOT EXISTS media_candidates (
    id INTEGER PRIMARY KEY,
    inbox_item_id INTEGER NOT NULL REFERENCES inbox_items(id) ON DELETE CASCADE,
    media_type TEXT NOT NULL,
    title TEXT NOT NULL,
    normalized_title TEXT NOT NULL,
    creator TEXT NOT NULL DEFAULT '',
    normalized_creator TEXT NOT NULL DEFAULT '',
    original_title TEXT NOT NULL DEFAULT '',
    year INTEGER,
    role TEXT NOT NULL DEFAULT 'primary',
    evidence TEXT NOT NULL DEFAULT '',
    recommendation TEXT NOT NULL DEFAULT 'unknown',
    summary TEXT NOT NULL DEFAULT '',
    reasons_json TEXT NOT NULL DEFAULT '[]',
    suggested_action TEXT NOT NULL DEFAULT 'ignore',
    library_state TEXT NOT NULL DEFAULT 'unknown',
    library_matches_json TEXT NOT NULL DEFAULT '[]',
    metadata_json TEXT NOT NULL DEFAULT '{}',
    status TEXT NOT NULL DEFAULT 'evaluated',
    created_at TEXT NOT NULL,
    updated_at TEXT NOT NULL,
    UNIQUE(inbox_item_id, media_type, normalized_title, normalized_creator)
);

CREATE TABLE IF NOT EXISTS telegram_chat_state (
    chat_id INTEGER PRIMARY KEY,
    last_source_url TEXT NOT NULL DEFAULT '',
    last_source_title TEXT NOT NULL DEFAULT '',
    last_action TEXT NOT NULL DEFAULT '',
    updated_at TEXT NOT NULL
);

CREATE TABLE IF NOT EXISTS control_intents (
    id INTEGER PRIMARY KEY,
    channel TEXT NOT NULL,
    conversation_id TEXT NOT NULL DEFAULT '',
    message TEXT NOT NULL,
    plan_json TEXT NOT NULL DEFAULT '{}',
    status TEXT NOT NULL DEFAULT 'interpreted',
    created_at TEXT NOT NULL,
    updated_at TEXT NOT NULL
);

CREATE TABLE IF NOT EXISTS control_plans (
    id INTEGER PRIMARY KEY,
    intent_id INTEGER REFERENCES control_intents(id) ON DELETE SET NULL,
    media_type TEXT NOT NULL,
    action TEXT NOT NULL,
    risk TEXT NOT NULL DEFAULT 'read',
    status TEXT NOT NULL DEFAULT 'proposed',
    idempotency_key TEXT NOT NULL UNIQUE,
    payload_json TEXT NOT NULL DEFAULT '{}',
    created_at TEXT NOT NULL,
    updated_at TEXT NOT NULL
);

CREATE TABLE IF NOT EXISTS control_commands (
    id INTEGER PRIMARY KEY,
    plan_id INTEGER NOT NULL REFERENCES control_plans(id) ON DELETE CASCADE,
    adapter TEXT NOT NULL,
    action TEXT NOT NULL,
    position INTEGER NOT NULL DEFAULT 0,
    status TEXT NOT NULL DEFAULT 'pending',
    request_json TEXT NOT NULL DEFAULT '{}',
    result_json TEXT NOT NULL DEFAULT '{}',
    error TEXT NOT NULL DEFAULT '',
    created_at TEXT NOT NULL,
    updated_at TEXT NOT NULL,
    UNIQUE(plan_id, position)
);

CREATE TABLE IF NOT EXISTS workflow_jobs (
    id INTEGER PRIMARY KEY,
    intent_id INTEGER REFERENCES control_intents(id) ON DELETE SET NULL,
    plan_id INTEGER REFERENCES control_plans(id) ON DELETE SET NULL,
    kind TEXT NOT NULL,
    status TEXT NOT NULL DEFAULT 'requested',
    detail TEXT NOT NULL DEFAULT '',
    error TEXT NOT NULL DEFAULT '',
    created_at TEXT NOT NULL,
    updated_at TEXT NOT NULL
);

CREATE TABLE IF NOT EXISTS control_events (
    id INTEGER PRIMARY KEY,
    intent_id INTEGER REFERENCES control_intents(id) ON DELETE SET NULL,
    plan_id INTEGER REFERENCES control_plans(id) ON DELETE SET NULL,
    job_id INTEGER REFERENCES workflow_jobs(id) ON DELETE SET NULL,
    event_type TEXT NOT NULL,
    payload_json TEXT NOT NULL DEFAULT '{}',
    created_at TEXT NOT NULL
);

CREATE TABLE IF NOT EXISTS query_cache (
    namespace TEXT NOT NULL,
    cache_key TEXT NOT NULL,
    value_json TEXT NOT NULL,
    expires_at TEXT NOT NULL,
    updated_at TEXT NOT NULL,
    PRIMARY KEY(namespace, cache_key)
);

CREATE INDEX IF NOT EXISTS idx_assets_edition ON assets(edition_id);
CREATE INDEX IF NOT EXISTS idx_editions_work ON editions(work_id);
CREATE INDEX IF NOT EXISTS idx_inbox_created ON inbox_items(created_at DESC);
CREATE INDEX IF NOT EXISTS idx_candidates_inbox ON media_candidates(inbox_item_id);
CREATE INDEX IF NOT EXISTS idx_control_intents_created ON control_intents(created_at DESC);
CREATE INDEX IF NOT EXISTS idx_control_plans_intent ON control_plans(intent_id);
CREATE INDEX IF NOT EXISTS idx_workflow_jobs_status ON workflow_jobs(status, updated_at DESC);
CREATE INDEX IF NOT EXISTS idx_control_events_job ON control_events(job_id, created_at);
"""


# ---------------------------------------------------------------------------
# Migrations
# ---------------------------------------------------------------------------
# Schema changes are ordered and recorded in PRAGMA user_version. Before this
# existed there were three ad-hoc ALTER statements in initialize(), each guarded
# by a PRAGMA table_info check:
#
#   - nothing recorded which changes had run, so the only way to know the shape
#     of a database was to re-derive it from the guards;
#   - the guards only worked for adding a column. A rename, a backfill, or a new
#     constraint has no equivalent check, so the next change had nowhere to go;
#   - and there was no snapshot, so a half-applied change left no way back.
#
# A migration is applied inside a transaction, and the whole run is preceded by a
# snapshot of the database file.


@dataclass(frozen=True)
class Migration:
    version: int
    name: str
    apply: Callable[[sqlite3.Connection], None]


def _has_column(connection: sqlite3.Connection, table: str, column: str) -> bool:
    return any(row["name"] == column for row in connection.execute(f"PRAGMA table_info({table})"))


def _migration_0001_baseline(connection: sqlite3.Connection) -> None:
    """Adopt whatever the pre-migration code produced.

    The live database was created by executescript(SCHEMA) plus three guarded
    ALTERs, and reports user_version 0. Re-running both is safe -- every
    statement is IF NOT EXISTS or column-guarded -- so this migration is the
    baseline for both a fresh database and the existing one.
    """
    connection.executescript(SCHEMA)
    for column, statement in (
        ("year", "ALTER TABLE media_candidates ADD COLUMN year INTEGER"),
        ("library_state", "ALTER TABLE media_candidates ADD COLUMN library_state TEXT NOT NULL DEFAULT 'unknown'"),
        ("library_matches_json", "ALTER TABLE media_candidates ADD COLUMN library_matches_json TEXT NOT NULL DEFAULT '[]'"),
    ):
        if not _has_column(connection, "media_candidates", column):
            connection.execute(statement)


def _migration_0002_identifier_projection(connection: sqlite3.Connection) -> None:
    """Project asset source identifiers into an indexed table, and backfill.

    book_work_by_source_identifiers read every book asset row and JSON-parsed
    each metadata blob in Python. That is a full scan of the library on every
    import, and it cannot use an index because the data is inside a JSON string.
    """
    connection.executescript(
        """
        CREATE TABLE IF NOT EXISTS asset_identifiers (
            asset_id   INTEGER NOT NULL REFERENCES assets(id) ON DELETE CASCADE,
            identifier TEXT NOT NULL,
            PRIMARY KEY(asset_id, identifier)
        );
        CREATE INDEX IF NOT EXISTS idx_asset_identifiers_identifier
            ON asset_identifiers(identifier);
        """
    )
    rows = connection.execute("SELECT id, metadata_json FROM assets").fetchall()
    payload: list[tuple[int, str]] = []
    for row in rows:
        try:
            metadata = json.loads(row["metadata_json"] or "{}")
        except json.JSONDecodeError:
            continue
        if not isinstance(metadata, dict):
            continue
        payload.extend(
            (int(row["id"]), key) for key in _identifier_keys(metadata.get("source_identifiers"))
        )
    connection.executemany(
        "INSERT OR IGNORE INTO asset_identifiers(asset_id, identifier) VALUES (?, ?)",
        payload,
    )


def _migration_0003_wanted_dedupe_and_indexes(connection: sqlite3.Connection) -> None:
    """Give wanted_books normalised columns and a real uniqueness constraint.

    add_wanted deduplicated with SELECT-then-INSERT on trim(title)/trim(author),
    which compares raw text: "三体" and " 三体 " matched, but differing width or
    case did not, and two concurrent callers could both miss the SELECT and
    insert twice. A partial unique index over the normalised columns, scoped to
    status='wanted', lets the insert itself resolve the conflict.
    """
    for column in ("normalized_title", "normalized_author"):
        if not _has_column(connection, "wanted_books", column):
            connection.execute(
                f"ALTER TABLE wanted_books ADD COLUMN {column} TEXT NOT NULL DEFAULT ''"
            )

    for row in connection.execute("SELECT id, title, author, query FROM wanted_books").fetchall():
        # Rows created from a bare query have no title; fall back to the query so
        # that the dedupe key is never empty for them.
        title = str(row["title"] or row["query"] or "")
        connection.execute(
            "UPDATE wanted_books SET normalized_title=?, normalized_author=? WHERE id=?",
            (normalize(title), normalize(str(row["author"] or "")), row["id"]),
        )

    # Collapse rows that the old text comparison had let through as distinct,
    # keeping the lowest id. The unique index cannot be created while they exist.
    connection.execute(
        """UPDATE wanted_books SET status='superseded'
           WHERE status='wanted' AND id NOT IN (
               SELECT min(id) FROM wanted_books WHERE status='wanted'
               GROUP BY normalized_title, normalized_author
           )"""
    )
    connection.executescript(
        """
        CREATE UNIQUE INDEX IF NOT EXISTS idx_wanted_books_active
            ON wanted_books(normalized_title, normalized_author) WHERE status='wanted';
        CREATE INDEX IF NOT EXISTS idx_wanted_books_status ON wanted_books(status, created_at DESC);
        CREATE INDEX IF NOT EXISTS idx_media_candidates_status
            ON media_candidates(status, updated_at DESC);
        CREATE INDEX IF NOT EXISTS idx_media_candidates_lookup
            ON media_candidates(media_type, normalized_title, normalized_creator);
        CREATE INDEX IF NOT EXISTS idx_control_commands_plan ON control_commands(plan_id, position);
        CREATE INDEX IF NOT EXISTS idx_control_events_intent ON control_events(intent_id, created_at);
        CREATE INDEX IF NOT EXISTS idx_query_cache_expiry ON query_cache(expires_at);
        """
    )


# Every status a row may hold, enforced by CHECK constraints from migration 4.
# Declared here rather than only in prose so that a typo in a status string
# fails at write time instead of producing a row nothing queries.
WORKFLOW_STATUSES: tuple[str, ...] = (
    "requested", "running", "succeeded", "submitted", "failed", "cancelled",
)
PLAN_STATUSES: tuple[str, ...] = (
    "proposed", "approved", "running", "submitted", "succeeded", "failed", "refused", "cancelled",
)
COMMAND_STATUSES: tuple[str, ...] = (
    "pending", "running", "submitted", "succeeded", "failed", "cancelled",
)
CANDIDATE_STATUSES: tuple[str, ...] = (
    "pending", "selected", "ignored", "owned", "wanted", "tracked",
    "superseded", "not_recommended", "unknown",
)
WANTED_STATUSES: tuple[str, ...] = (
    "wanted", "acquired", "duplicate", "misclassified", "superseded", "cancelled",
)


def _check_clause(column: str, values: tuple[str, ...]) -> str:
    joined = ",".join(f"'{value}'" for value in values)
    return f"CHECK({column} IN ({joined}))"


def _migration_0004_state_machine_and_dead_tables(connection: sqlite3.Connection) -> None:
    """Constrain every status column, and retire three tables.

    Statuses were free-form text. The code wrote "succeeded" in one place and
    "success" in another for the same idea, and nothing objected -- a typo simply
    produced a row that no query would ever match again.

    SQLite cannot add a CHECK constraint to an existing table, so each table is
    rebuilt. Existing values are mapped onto the enumeration first; anything
    still unrecognised is failed loudly rather than silently coerced, because a
    value nobody anticipated is a bug worth seeing.

    Retired here:
      - source_candidates and download_jobs: created for a download pipeline that
        was never built. Zero rows, and referenced only by the schema itself.
      - activity_jobs: superseded by workflow_jobs, which links to the control
        ledger. Its 60 rows of history are migrated across rather than dropped.
    """
    # --- fold activity_jobs history into workflow_jobs ---------------------
    if _has_table(connection, "activity_jobs"):
        connection.execute(
            """INSERT INTO workflow_jobs(kind, status, detail, created_at, updated_at)
               SELECT kind,
                      CASE status
                          WHEN 'success' THEN 'succeeded'
                          WHEN 'failed'  THEN 'failed'
                          WHEN 'running' THEN 'running'
                          ELSE 'cancelled'
                      END,
                      detail, created_at, updated_at
               FROM activity_jobs
               WHERE NOT EXISTS (
                   SELECT 1 FROM workflow_jobs w
                   WHERE w.kind = activity_jobs.kind AND w.created_at = activity_jobs.created_at
               )"""
        )

    connection.executescript(
        """
        DROP TABLE IF EXISTS download_jobs;
        DROP TABLE IF EXISTS source_candidates;
        DROP TABLE IF EXISTS activity_jobs;
        DROP INDEX IF EXISTS idx_jobs_created;
        """
    )

    # --- normalise then constrain -----------------------------------------
    connection.execute("UPDATE workflow_jobs SET status='succeeded' WHERE status='success'")
    connection.execute("UPDATE control_plans SET status='proposed' WHERE status=''")

    for table, column, values in (
        ("workflow_jobs", "status", WORKFLOW_STATUSES),
        ("control_plans", "status", PLAN_STATUSES),
        ("control_commands", "status", COMMAND_STATUSES),
        ("media_candidates", "status", CANDIDATE_STATUSES),
        ("wanted_books", "status", WANTED_STATUSES),
    ):
        unexpected = [
            str(row[0])
            for row in connection.execute(
                f"SELECT DISTINCT {column} FROM {table} "
                f"WHERE {column} NOT IN ({','.join('?' for _ in values)})",
                values,
            )
        ]
        if unexpected:
            raise RuntimeError(
                f"{table}.{column} holds values outside the enumeration: {unexpected}. "
                "Add them to the tuple in db.py or correct the data; refusing to guess."
            )
        _rebuild_with_check(connection, table, column, values)


def _has_table(connection: sqlite3.Connection, name: str) -> bool:
    return connection.execute(
        "SELECT 1 FROM sqlite_master WHERE type='table' AND name=?", (name,)
    ).fetchone() is not None


def _rebuild_with_check(
    connection: sqlite3.Connection, table: str, column: str, values: tuple[str, ...]
) -> None:
    """Recreate a table with a CHECK constraint on one column, preserving data.

    SQLite has no ALTER TABLE ADD CONSTRAINT, so the table is rebuilt from its
    own stored DDL with the clause spliced in. Indexes are recreated afterwards
    because DROP TABLE takes them with it.
    """
    ddl = connection.execute(
        "SELECT sql FROM sqlite_master WHERE type='table' AND name=?", (table,)
    ).fetchone()
    if ddl is None:
        return
    statement = str(ddl[0])
    if "CHECK(" + column in statement.replace(" ", ""):
        return
    indexes = [
        str(row[0])
        for row in connection.execute(
            "SELECT sql FROM sqlite_master WHERE type='index' AND tbl_name=? AND sql IS NOT NULL",
            (table,),
        )
    ]
    columns = [row["name"] for row in connection.execute(f"PRAGMA table_info({table})")]
    column_list = ",".join(f'"{name}"' for name in columns)

    # Splice the CHECK in before the final closing parenthesis.
    cut = statement.rfind(")")
    rebuilt = (
        statement[:cut].rstrip().rstrip(",")
        + ",\n    "
        + _check_clause(column, values)
        + "\n)"
    )
    rebuilt = rebuilt.replace(f"TABLE IF NOT EXISTS {table}", f"TABLE {table}__new", 1)
    rebuilt = rebuilt.replace(f"TABLE {table} ", f"TABLE {table}__new ", 1)
    if f"{table}__new" not in rebuilt:
        raise RuntimeError(f"could not derive a rebuild statement for {table}")

    connection.execute(rebuilt)
    connection.execute(f"INSERT INTO {table}__new({column_list}) SELECT {column_list} FROM {table}")
    connection.execute(f"DROP TABLE {table}")
    connection.execute(f"ALTER TABLE {table}__new RENAME TO {table}")
    for index_sql in indexes:
        connection.execute(index_sql)


MIGRATIONS: tuple[Migration, ...] = (
    Migration(1, "baseline", _migration_0001_baseline),
    Migration(2, "identifier_projection", _migration_0002_identifier_projection),
    Migration(3, "wanted_dedupe_and_indexes", _migration_0003_wanted_dedupe_and_indexes),
    Migration(4, "state_machine_and_dead_tables", _migration_0004_state_machine_and_dead_tables),
)

SCHEMA_VERSION = MIGRATIONS[-1].version


def now() -> str:
    return datetime.now(UTC).replace(microsecond=0).isoformat()


def normalize(value: str) -> str:
    value = unicodedata.normalize("NFKC", value).casefold().strip()
    return " ".join(value.split())


def _identifier_keys(identifiers: Any) -> list[str]:
    """Normalise source identifiers for indexed comparison."""
    if not isinstance(identifiers, (list, tuple)):
        return []
    keys = []
    for value in identifiers:
        key = str(value).casefold().strip()
        if key and key not in keys:
            keys.append(key)
    return keys


def _run_metadata(meta: Any) -> dict[str, Any]:
    """Normalise a pi_agent.RunMeta into stored metadata.

    Accepts None so that callers with no model involvement (imports, manual
    entry) record an explicit empty provenance rather than omitting the keys.
    """
    if meta is None:
        return {"model_used": "", "fallback": False, "primary_error": "", "catalog_errors": []}
    return meta.as_metadata()


class Database:
    def __init__(self, path: Path):
        self.path = path

    @contextmanager
    def connect(self) -> Iterator[sqlite3.Connection]:
        connection = sqlite3.connect(self.path, timeout=30)
        connection.row_factory = sqlite3.Row
        connection.execute("PRAGMA foreign_keys = ON")
        connection.execute("PRAGMA busy_timeout = 30000")
        try:
            yield connection
            connection.commit()
        finally:
            connection.close()

    @contextmanager
    def transaction(self) -> Iterator[sqlite3.Connection]:
        """Run several writes as one unit.

        Methods that accept a `connection` argument join the caller's
        transaction instead of opening their own. Without this, a multi-step
        write was several independent transactions and a failure part-way left
        the earlier steps committed -- import_file compensated by deleting the
        orphaned work afterwards, which only works if the compensation itself
        succeeds.

        Rolls back on any exception, including KeyboardInterrupt.
        """
        connection = sqlite3.connect(self.path, timeout=30)
        connection.row_factory = sqlite3.Row
        connection.execute("PRAGMA foreign_keys = ON")
        connection.execute("PRAGMA busy_timeout = 30000")
        try:
            connection.execute("BEGIN IMMEDIATE")
            yield connection
            connection.commit()
        except BaseException:
            connection.rollback()
            raise
        finally:
            connection.close()

    @contextmanager
    def _writer(self, connection: sqlite3.Connection | None) -> Iterator[sqlite3.Connection]:
        """Join an existing transaction, or open a self-contained one."""
        if connection is not None:
            yield connection
            return
        with self.connect() as own:
            yield own

    def schema_version(self) -> int:
        with self.connect() as connection:
            return int(connection.execute("PRAGMA user_version").fetchone()[0])

    def initialize(self) -> None:
        """Bring the database up to SCHEMA_VERSION, snapshotting first."""
        self.path.parent.mkdir(parents=True, exist_ok=True)
        with self.connect() as connection:
            connection.execute("PRAGMA journal_mode = WAL")
            current = int(connection.execute("PRAGMA user_version").fetchone()[0])

        # Order matters: the newer-schema check has to come first. Checked after
        # the early return, it would never fire, because a database ahead of this
        # build has no pending migrations -- and an older build would then go on
        # to write to a schema it does not understand.
        if current > SCHEMA_VERSION:
            raise RuntimeError(
                f"database at {self.path} reports schema version {current}, "
                f"newer than this code understands ({SCHEMA_VERSION}). Refusing to run: "
                "an older build must not write to a newer schema."
            )
        pending = [migration for migration in MIGRATIONS if migration.version > current]
        if not pending:
            return

        self._snapshot_before_migration(current)
        for migration in pending:
            # One transaction per migration, so a failure leaves the version at
            # the last fully applied one rather than somewhere in between.
            connection = sqlite3.connect(self.path, timeout=30)
            connection.row_factory = sqlite3.Row
            try:
                connection.execute("PRAGMA foreign_keys = OFF")
                connection.execute("BEGIN")
                migration.apply(connection)
                # PRAGMA user_version does not accept a parameter binding.
                connection.execute(f"PRAGMA user_version = {int(migration.version)}")
                connection.commit()
            except Exception as exc:
                connection.rollback()
                raise RuntimeError(
                    f"migration {migration.version} ({migration.name}) failed: {exc}. "
                    f"Database left at version {migration.version - 1}."
                ) from exc
            finally:
                connection.close()

    def _snapshot_before_migration(self, current_version: int) -> None:
        """Copy the database before migrating, keeping the last few snapshots.

        Uses the online backup API rather than a file copy so the snapshot is
        consistent even with a WAL in progress.

        Skipped when the database holds no user table yet. Testing the file's
        existence is not enough: connect() creates the file, so a first-run
        initialize() would otherwise snapshot an empty database.
        """
        if not self.path.exists():
            return
        with closing(sqlite3.connect(self.path, timeout=30)) as probe:
            tables = probe.execute(
                "SELECT count(*) FROM sqlite_master WHERE type='table' AND name NOT LIKE 'sqlite_%'"
            ).fetchone()[0]
        if not tables:
            return
        directory = self.path.parent / "migrations"
        directory.mkdir(parents=True, exist_ok=True)
        stamp = datetime.now(UTC).strftime("%Y%m%dT%H%M%SZ")
        target = directory / f"{self.path.stem}-v{current_version}-{stamp}.sqlite3"
        with closing(sqlite3.connect(self.path, timeout=30)) as source:
            with closing(sqlite3.connect(target)) as destination:
                source.backup(destination)
        snapshots = sorted(directory.glob(f"{self.path.stem}-v*.sqlite3"))
        for old in snapshots[:-5]:
            old.unlink(missing_ok=True)

    def create_job(self, kind: str, detail: str = "") -> int:
        stamp = now()
        with self.connect() as connection:
            cursor = connection.execute(
                "INSERT INTO workflow_jobs(kind, status, detail, created_at, updated_at) VALUES (?, 'running', ?, ?, ?)",
                (kind, detail, stamp, stamp),
            )
            return int(cursor.lastrowid)

    def finish_job(self, job_id: int, status: str, detail: str) -> None:
        with self.connect() as connection:
            connection.execute(
                "UPDATE workflow_jobs SET status = ?, detail = ?, updated_at = ? WHERE id = ?",
                (status, detail, now(), job_id),
            )

    def update_job(self, job_id: int, detail: str) -> None:
        with self.connect() as connection:
            connection.execute(
                "UPDATE workflow_jobs SET detail = ?, updated_at = ? WHERE id = ?",
                (detail, now(), job_id),
            )

    def record_intent(
        self,
        *,
        channel: str,
        conversation_id: str,
        message: str,
        plan: dict[str, Any],
        status: str = "interpreted",
    ) -> int:
        stamp = now()
        with self.connect() as connection:
            cursor = connection.execute(
                """INSERT INTO control_intents(
                       channel, conversation_id, message, plan_json, status, created_at, updated_at
                   ) VALUES (?, ?, ?, ?, ?, ?, ?)""",
                (
                    channel,
                    conversation_id,
                    message,
                    json.dumps(plan, ensure_ascii=False, sort_keys=True),
                    status,
                    stamp,
                    stamp,
                ),
            )
            return int(cursor.lastrowid)


    def create_control_plan(
        self,
        *,
        intent_id: int | None,
        media_type: str,
        action: str,
        risk: str,
        idempotency_key: str,
        payload: dict[str, Any],
        status: str = "proposed",
        connection: sqlite3.Connection | None = None,
    ) -> tuple[int, bool]:
        """Create a plan, or return the existing one for this idempotency key.

        Returns (plan_id, created). The insert itself resolves the conflict
        against the UNIQUE index. The previous SELECT-then-INSERT was the
        idempotency guard for every write, and two Telegram messages arriving
        together could both pass the SELECT -- after which the second raised
        IntegrityError instead of reporting "already planned", so a duplicate
        request surfaced as a failure rather than as a no-op.
        """
        stamp = now()
        with self._writer(connection) as active:
            row = active.execute(
                """INSERT INTO control_plans(
                       intent_id, media_type, action, risk, status, idempotency_key,
                       payload_json, created_at, updated_at
                   ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)
                   ON CONFLICT(idempotency_key) DO NOTHING
                   RETURNING id""",
                (
                    intent_id,
                    media_type,
                    action,
                    risk,
                    status,
                    idempotency_key,
                    json.dumps(payload, ensure_ascii=False, sort_keys=True),
                    stamp,
                    stamp,
                ),
            ).fetchone()
            if row is not None:
                return int(row["id"]), True
            # DO NOTHING returns no row, so the pre-existing plan is fetched here.
            existing = active.execute(
                "SELECT id FROM control_plans WHERE idempotency_key=?",
                (idempotency_key,),
            ).fetchone()
            assert existing is not None
            return int(existing["id"]), False

    def control_plan(self, plan_id: int) -> sqlite3.Row | None:
        with self.connect() as connection:
            return connection.execute("SELECT * FROM control_plans WHERE id=?", (plan_id,)).fetchone()

    def update_control_plan(
        self, plan_id: int, status: str, *, connection: sqlite3.Connection | None = None
    ) -> None:
        with self._writer(connection) as active:
            active.execute(
                "UPDATE control_plans SET status=?, updated_at=? WHERE id=?",
                (status, now(), plan_id),
            )

    def add_control_command(
        self,
        *,
        plan_id: int,
        adapter: str,
        action: str,
        position: int,
        request: dict[str, Any],
        connection: sqlite3.Connection | None = None,
    ) -> int:
        stamp = now()
        with self._writer(connection) as active:
            active.execute(
                """INSERT INTO control_commands(
                       plan_id, adapter, action, position, request_json, created_at, updated_at
                   ) VALUES (?, ?, ?, ?, ?, ?, ?)
                   ON CONFLICT(plan_id, position) DO UPDATE SET
                       adapter=excluded.adapter, action=excluded.action,
                       request_json=excluded.request_json, updated_at=excluded.updated_at""",
                (
                    plan_id,
                    adapter,
                    action,
                    position,
                    json.dumps(request, ensure_ascii=False, sort_keys=True),
                    stamp,
                    stamp,
                ),
            )
            row = active.execute(
                "SELECT id FROM control_commands WHERE plan_id=? AND position=?",
                (plan_id, position),
            ).fetchone()
            assert row is not None
            return int(row["id"])

    def update_control_command(
        self,
        command_id: int,
        status: str,
        *,
        result: dict[str, Any] | None = None,
        error: str = "",
        connection: sqlite3.Connection | None = None,
    ) -> None:
        with self._writer(connection) as active:
            active.execute(
                """UPDATE control_commands
                   SET status=?, result_json=?, error=?, updated_at=? WHERE id=?""",
                (
                    status,
                    json.dumps(result or {}, ensure_ascii=False, sort_keys=True),
                    error,
                    now(),
                    command_id,
                ),
            )

    def create_workflow_job(
        self,
        *,
        kind: str,
        intent_id: int | None = None,
        plan_id: int | None = None,
        status: str = "requested",
        detail: str = "",
    ) -> int:
        stamp = now()
        with self.connect() as connection:
            cursor = connection.execute(
                """INSERT INTO workflow_jobs(
                       intent_id, plan_id, kind, status, detail, created_at, updated_at
                   ) VALUES (?, ?, ?, ?, ?, ?, ?)""",
                (intent_id, plan_id, kind, status, detail, stamp, stamp),
            )
            return int(cursor.lastrowid)

    def update_workflow_job(self, job_id: int, status: str, detail: str = "", error: str = "") -> None:
        with self.connect() as connection:
            connection.execute(
                """UPDATE workflow_jobs
                   SET status=?, detail=?, error=?, updated_at=? WHERE id=?""",
                (status, detail, error, now(), job_id),
            )

    def append_control_event(
        self,
        event_type: str,
        payload: dict[str, Any],
        *,
        intent_id: int | None = None,
        plan_id: int | None = None,
        job_id: int | None = None,
        connection: sqlite3.Connection | None = None,
    ) -> int:
        with self._writer(connection) as active:
            cursor = active.execute(
                """INSERT INTO control_events(
                       intent_id, plan_id, job_id, event_type, payload_json, created_at
                   ) VALUES (?, ?, ?, ?, ?, ?)""",
                (
                    intent_id,
                    plan_id,
                    job_id,
                    event_type,
                    json.dumps(payload, ensure_ascii=False, sort_keys=True),
                    now(),
                ),
            )
            return int(cursor.lastrowid)

    def cache_get(self, namespace: str, cache_key: str) -> dict[str, Any] | None:
        with self.connect() as connection:
            row = connection.execute(
                "SELECT value_json, expires_at FROM query_cache WHERE namespace=? AND cache_key=?",
                (namespace, cache_key),
            ).fetchone()
            if not row:
                return None
            if datetime.fromisoformat(str(row["expires_at"])) <= datetime.now(UTC):
                connection.execute(
                    "DELETE FROM query_cache WHERE namespace=? AND cache_key=?",
                    (namespace, cache_key),
                )
                return None
            value = json.loads(str(row["value_json"]))
            return value if isinstance(value, dict) else None

    def cache_put(self, namespace: str, cache_key: str, value: dict[str, Any], ttl_seconds: int) -> None:
        stamp = now()
        expires = (datetime.now(UTC) + timedelta(seconds=max(1, ttl_seconds))).replace(microsecond=0).isoformat()
        with self.connect() as connection:
            connection.execute(
                """INSERT INTO query_cache(namespace, cache_key, value_json, expires_at, updated_at)
                   VALUES (?, ?, ?, ?, ?)
                   ON CONFLICT(namespace, cache_key) DO UPDATE SET
                       value_json=excluded.value_json, expires_at=excluded.expires_at,
                       updated_at=excluded.updated_at""",
                (
                    namespace,
                    cache_key,
                    json.dumps(value, ensure_ascii=False, sort_keys=True),
                    expires,
                    stamp,
                ),
            )

    def add_wanted(
        self,
        query: str,
        title: str = "",
        author: str = "",
        language: str = "und",
        *,
        connection: sqlite3.Connection | None = None,
    ) -> int:
        """Add a wanted book, returning the existing row if one is already active.

        Deduplication is delegated to the partial unique index over the
        normalised columns (migration 3). The previous SELECT-then-INSERT
        compared raw trimmed text, so it missed width and case variants, and two
        concurrent callers could both pass the SELECT and insert twice.
        """
        clean_query = query.strip()
        clean_title = title.strip()
        clean_author = author.strip()
        # Rows created from a bare query carry no title; the dedupe key falls
        # back to the query so it is never empty.
        title_key = normalize(clean_title or clean_query)
        author_key = normalize(clean_author)
        with self._writer(connection) as active:
            row = active.execute(
                """INSERT INTO wanted_books(
                       query, title, author, language, normalized_title, normalized_author, created_at)
                   VALUES (?, ?, ?, ?, ?, ?, ?)
                   ON CONFLICT(normalized_title, normalized_author) WHERE status='wanted' DO UPDATE SET
                       title=CASE WHEN excluded.title!='' THEN excluded.title ELSE wanted_books.title END,
                       author=CASE WHEN excluded.author!='' THEN excluded.author ELSE wanted_books.author END
                   RETURNING id""",
                (
                    clean_query,
                    clean_title,
                    clean_author,
                    language.strip() or "und",
                    title_key,
                    author_key,
                    now(),
                ),
            ).fetchone()
            assert row is not None
            return int(row["id"])

    def set_chat_source(self, chat_id: int, source_url: str, source_title: str = "") -> None:
        with self.connect() as connection:
            connection.execute(
                """INSERT INTO telegram_chat_state(chat_id, last_source_url, last_source_title, last_action, updated_at)
                   VALUES (?, ?, ?, 'source', ?)
                   ON CONFLICT(chat_id) DO UPDATE SET
                       last_source_url=excluded.last_source_url,
                       last_source_title=CASE WHEN excluded.last_source_title!='' THEN excluded.last_source_title
                                              ELSE telegram_chat_state.last_source_title END,
                       last_action='source', updated_at=excluded.updated_at""",
                (chat_id, source_url.strip(), source_title.strip(), now()),
            )

    def chat_state(self, chat_id: int) -> sqlite3.Row | None:
        with self.connect() as connection:
            return connection.execute(
                "SELECT * FROM telegram_chat_state WHERE chat_id=?",
                (chat_id,),
            ).fetchone()

    def update_wanted_status(self, wanted_id: int, status: str) -> None:
        with self.connect() as connection:
            connection.execute("UPDATE wanted_books SET status=? WHERE id=?", (status, wanted_id))

    def reconcile_imported_book(self, title: str, author: str = "") -> None:
        title_key = normalize(title)
        author_key = normalize(author)
        stamp = now()
        with self.connect() as connection:
            for row in connection.execute("SELECT id,title,query,author FROM wanted_books WHERE status='wanted'"):
                wanted_title = normalize(str(row["title"] or row["query"] or ""))
                wanted_author = normalize(str(row["author"] or ""))
                if wanted_title != title_key or (author_key and wanted_author and wanted_author != author_key):
                    continue
                connection.execute("UPDATE wanted_books SET status='acquired' WHERE id=?", (row["id"],))
            connection.execute(
                """UPDATE media_candidates SET status='owned', library_state='owned', updated_at=?
                   WHERE media_type='book' AND normalized_title=?
                     AND (?='' OR normalized_creator='' OR normalized_creator=?)""",
                (stamp, title_key, author_key, author_key),
            )

    def save_inbox_item(self, source_url: str, evaluation: dict[str, Any], *, meta: Any = None) -> int:
        stamp = now()
        reasons = evaluation.get("reasons") or []
        metadata = {
            "mentioned_works": evaluation.get("mentioned_works") or [],
            "source_title": evaluation.get("source_title") or "",
            # Provenance arrives as an argument, not mixed into the payload the
            # model authored. See pi_agent.RunMeta.
            **_run_metadata(meta),
        }
        values = (
            source_url,
            str(evaluation.get("media_type") or "unknown"),
            str(evaluation.get("title") or ""),
            str(evaluation.get("creator") or ""),
            str(evaluation.get("recommendation") or "unknown"),
            str(evaluation.get("summary") or ""),
            json.dumps(reasons, ensure_ascii=False),
            str(evaluation.get("suggested_action") or ""),
            json.dumps(metadata, ensure_ascii=False),
            stamp,
            stamp,
        )
        with self.connect() as connection:
            connection.execute(
                """INSERT INTO inbox_items(
                       source_url, media_type, title, creator, recommendation, summary,
                       reasons_json, suggested_action, metadata_json, created_at, updated_at
                   ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
                   ON CONFLICT(source_url) DO UPDATE SET
                       media_type=excluded.media_type, title=excluded.title, creator=excluded.creator,
                       recommendation=excluded.recommendation, summary=excluded.summary,
                       reasons_json=excluded.reasons_json, suggested_action=excluded.suggested_action,
                       metadata_json=excluded.metadata_json, status='evaluated', updated_at=excluded.updated_at""",
                values,
            )
            row = connection.execute("SELECT id FROM inbox_items WHERE source_url = ?", (source_url,)).fetchone()
            assert row is not None
            return int(row["id"])

    def save_source_evaluation(
        self,
        source_url: str,
        evaluation: dict[str, Any],
        *,
        meta: Any = None,
    ) -> tuple[int, list[int]]:
        stamp = now()
        source_title = str(evaluation.get("source_title") or source_url)
        source_summary = str(evaluation.get("source_summary") or "")
        run_metadata = _run_metadata(meta)
        metadata = {
            **run_metadata,
            "no_items_reason": evaluation.get("no_items_reason") or "",
        }
        with self.connect() as connection:
            connection.execute(
                """INSERT INTO inbox_items(
                       source_url, media_type, title, creator, recommendation, summary,
                       reasons_json, suggested_action, metadata_json, status, created_at, updated_at
                   ) VALUES (?, 'source', ?, '', 'not_applicable', ?, '[]', '', ?, 'evaluated', ?, ?)
                   ON CONFLICT(source_url) DO UPDATE SET
                       media_type='source', title=excluded.title, creator='', recommendation='not_applicable',
                       summary=excluded.summary, reasons_json='[]', suggested_action='',
                       metadata_json=excluded.metadata_json, status='evaluated', updated_at=excluded.updated_at""",
                (source_url, source_title, source_summary, json.dumps(metadata, ensure_ascii=False), stamp, stamp),
            )
            row = connection.execute("SELECT id FROM inbox_items WHERE source_url = ?", (source_url,)).fetchone()
            assert row is not None
            inbox_id = int(row["id"])
            connection.execute(
                "UPDATE media_candidates SET status='superseded', updated_at=? WHERE inbox_item_id=?",
                (stamp, inbox_id),
            )
            candidate_ids: list[int] = []
            for item in evaluation.get("items") or []:
                if not isinstance(item, dict):
                    continue
                media_type = str(item.get("media_type") or "unknown")
                title = str(item.get("title") or "").strip()
                creator = str(item.get("creator") or "").strip()
                if media_type not in {"book", "movie", "tv", "music"} or not title:
                    continue
                item_metadata = {
                    "model_used": run_metadata["model_used"],
                    "fallback": run_metadata["fallback"],
                    "aliases": item.get("aliases") or [],
                    "external_ids": item.get("external_ids") or {},
                    "book_reviews": item.get("book_reviews") or [],
                    "book_review_errors": item.get("book_review_errors") or [],
                    "book_review_providers_checked": item.get("book_review_providers_checked") or [],
                    "book_reviews_updated_at": stamp if item.get("media_type") == "book" else "",
                    "book_web_review_evidence": item.get("book_web_review_evidence") or [],
                    "book_web_review_errors": item.get("book_web_review_errors") or [],
                    "book_web_review_providers_checked": item.get("book_web_review_providers_checked") or [],
                    "book_web_review": item.get("book_web_review") or {},
                    "book_web_review_model": item.get("book_web_review_model") or "",
                    "book_web_reviews_updated_at": stamp if item.get("media_type") == "book" else "",
                }
                library_state = str(item.get("library_state") or "unknown")
                initial_status = {
                    "owned": "owned",
                    "wanted": "wanted",
                    "tracked": "tracked",
                }.get(library_state, "not_recommended" if item.get("recommendation") == "skip" else "pending")
                values = (
                    inbox_id,
                    media_type,
                    title,
                    normalize(title),
                    creator,
                    normalize(creator),
                    str(item.get("original_title") or ""),
                    item.get("year") if isinstance(item.get("year"), int) else None,
                    str(item.get("role") or "primary"),
                    str(item.get("evidence") or ""),
                    str(item.get("recommendation") or "unknown"),
                    str(item.get("summary") or ""),
                    json.dumps(item.get("reasons") or [], ensure_ascii=False),
                    str(item.get("suggested_action") or "ignore"),
                    library_state,
                    json.dumps(item.get("library_matches") or [], ensure_ascii=False),
                    json.dumps(item_metadata, ensure_ascii=False),
                    initial_status,
                    stamp,
                    stamp,
                )
                connection.execute(
                    """INSERT INTO media_candidates(
                           inbox_item_id, media_type, title, normalized_title, creator, normalized_creator,
                           original_title, year, role, evidence, recommendation, summary, reasons_json,
                           suggested_action, library_state, library_matches_json, metadata_json, status, created_at, updated_at
                       ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
                       ON CONFLICT(inbox_item_id, media_type, normalized_title, normalized_creator) DO UPDATE SET
                           original_title=excluded.original_title, year=excluded.year, role=excluded.role, evidence=excluded.evidence,
                           recommendation=excluded.recommendation, summary=excluded.summary,
                           reasons_json=excluded.reasons_json, suggested_action=excluded.suggested_action,
                           library_state=excluded.library_state, library_matches_json=excluded.library_matches_json,
                           metadata_json=excluded.metadata_json,
                           status=CASE WHEN media_candidates.status IN ('selected','ignored')
                                       AND excluded.library_state NOT IN ('owned','wanted','tracked')
                                       THEN media_candidates.status ELSE excluded.status END,
                           updated_at=excluded.updated_at""",
                    values,
                )
                candidate = connection.execute(
                    """SELECT id FROM media_candidates
                       WHERE inbox_item_id=? AND media_type=? AND normalized_title=? AND normalized_creator=?""",
                    (inbox_id, media_type, normalize(title), normalize(creator)),
                ).fetchone()
                assert candidate is not None
                candidate_ids.append(int(candidate["id"]))
            return inbox_id, candidate_ids

    def media_candidate(self, candidate_id: int) -> sqlite3.Row | None:
        with self.connect() as connection:
            return connection.execute(
                """SELECT c.*, i.source_url, i.title AS source_title
                   FROM media_candidates c JOIN inbox_items i ON i.id=c.inbox_item_id
                   WHERE c.id=?""",
                (candidate_id,),
            ).fetchone()

    def media_candidates(self, inbox_item_id: int) -> list[sqlite3.Row]:
        with self.connect() as connection:
            return list(
                connection.execute(
                    "SELECT * FROM media_candidates WHERE inbox_item_id=? ORDER BY role, id",
                    (inbox_item_id,),
                )
            )

    def sources(self, limit: int = 100) -> list[sqlite3.Row]:
        with self.connect() as connection:
            return list(
                connection.execute(
                    """SELECT i.*,
                              COUNT(c.id) AS discovered_count,
                              SUM(CASE WHEN c.status='pending' THEN 1 ELSE 0 END) AS pending_count,
                              SUM(CASE WHEN c.status IN ('owned','wanted','tracked') THEN 1 ELSE 0 END) AS existing_count
                       FROM inbox_items i LEFT JOIN media_candidates c
                         ON c.inbox_item_id=i.id AND c.status!='superseded'
                       WHERE i.media_type='source'
                       GROUP BY i.id ORDER BY i.updated_at DESC LIMIT ?""",
                    (limit,),
                )
            )

    def all_media_candidates(self, limit: int = 200, status: str = "") -> list[sqlite3.Row]:
        query = """SELECT c.*, i.source_url, i.title AS source_title
                   FROM media_candidates c JOIN inbox_items i ON i.id=c.inbox_item_id
                   WHERE c.status!='superseded'"""
        values: list[Any] = []
        if status:
            query += " AND c.status=?"
            values.append(status)
        query += " ORDER BY c.updated_at DESC, c.id DESC LIMIT ?"
        values.append(limit)
        with self.connect() as connection:
            return list(connection.execute(query, values))

    def update_candidate_status(self, candidate_id: int, status: str) -> None:
        with self.connect() as connection:
            connection.execute(
                "UPDATE media_candidates SET status=?, updated_at=? WHERE id=?",
                (status, now(), candidate_id),
            )

    def update_candidate_catalog_state(
        self,
        candidate_id: int,
        status: str,
        library_state: str,
        matches: list[dict[str, Any]],
    ) -> None:
        with self.connect() as connection:
            connection.execute(
                """UPDATE media_candidates
                   SET status=?, library_state=?, library_matches_json=?, updated_at=?
                   WHERE id=?""",
                (status, library_state, json.dumps(matches, ensure_ascii=False), now(), candidate_id),
            )

    def update_candidate_reviews(
        self,
        candidate_id: int,
        reviews: list[dict[str, Any]],
        errors: list[str],
        providers_checked: list[str],
        web_evidence: list[dict[str, Any]] | None = None,
        web_errors: list[str] | None = None,
        web_providers_checked: list[str] | None = None,
        web_review: dict[str, Any] | None = None,
        web_review_model: str = "",
    ) -> None:
        with self.connect() as connection:
            row = connection.execute(
                "SELECT metadata_json FROM media_candidates WHERE id=?",
                (candidate_id,),
            ).fetchone()
            if not row:
                raise ValueError(f"candidate {candidate_id} does not exist")
            try:
                metadata = json.loads(str(row["metadata_json"] or "{}"))
            except json.JSONDecodeError:
                metadata = {}
            metadata.update({
                "book_reviews": reviews,
                "book_review_errors": errors,
                "book_review_providers_checked": providers_checked,
                "book_reviews_updated_at": now(),
                "book_web_review_evidence": web_evidence or [],
                "book_web_review_errors": web_errors or [],
                "book_web_review_providers_checked": web_providers_checked or [],
                "book_web_review": web_review or {},
                "book_web_review_model": web_review_model,
                "book_web_reviews_updated_at": now(),
            })
            connection.execute(
                "UPDATE media_candidates SET metadata_json=?, updated_at=? WHERE id=?",
                (json.dumps(metadata, ensure_ascii=False, sort_keys=True), now(), candidate_id),
            )

    def inbox_item(self, item_id: int) -> sqlite3.Row | None:
        with self.connect() as connection:
            return connection.execute("SELECT * FROM inbox_items WHERE id = ?", (item_id,)).fetchone()

    def update_inbox_status(self, item_id: int, status: str) -> None:
        with self.connect() as connection:
            connection.execute(
                "UPDATE inbox_items SET status = ?, updated_at = ? WHERE id = ?",
                (status, now(), item_id),
            )

    def upsert_work(
        self,
        title: str,
        author: str,
        media_type: str = "book",
        *,
        connection: sqlite3.Connection | None = None,
    ) -> int:
        """Insert a work, or touch the existing one, and return its id.

        A single statement against the UNIQUE(media_type, normalized_title,
        normalized_author) constraint. The previous SELECT-then-INSERT could
        interleave with a concurrent caller between the two statements and raise
        IntegrityError; two Telegram chats importing the same book, or an import
        racing the web upload handler, was enough.
        """
        stamp = now()
        title_key = normalize(title)
        author_key = normalize(author)
        with self._writer(connection) as active:
            row = active.execute(
                """INSERT INTO works(title, author, normalized_title, normalized_author, media_type, created_at, updated_at)
                   VALUES (?, ?, ?, ?, ?, ?, ?)
                   ON CONFLICT(media_type, normalized_title, normalized_author) DO UPDATE SET
                       title=CASE WHEN excluded.title!='' THEN excluded.title ELSE works.title END,
                       author=CASE WHEN excluded.author!='' THEN excluded.author ELSE works.author END,
                       updated_at=excluded.updated_at
                   RETURNING id""",
                (title, author, title_key, author_key, media_type, stamp, stamp),
            ).fetchone()
            assert row is not None
            return int(row["id"])

    def book_work_by_source_identifiers(
        self,
        identifiers: tuple[str, ...],
        *,
        connection: sqlite3.Connection | None = None,
    ) -> int | None:
        """Find the book work that already owns any of these identifiers.

        An indexed join. This used to read every book asset in the library and
        JSON-parse each metadata blob in Python to find one match.
        """
        keys = _identifier_keys(identifiers)
        if not keys:
            return None
        placeholders = ",".join("?" for _ in keys)
        with self._writer(connection) as active:
            row = active.execute(
                f"""SELECT e.work_id FROM asset_identifiers i
                    JOIN assets a ON a.id=i.asset_id
                    JOIN editions e ON e.id=a.edition_id
                    JOIN works w ON w.id=e.work_id
                    WHERE w.media_type='book' AND i.identifier IN ({placeholders})
                    ORDER BY e.work_id LIMIT 1""",
                keys,
            ).fetchone()
        return int(row["work_id"]) if row else None

    def upsert_edition(
        self,
        work_id: int,
        language: str,
        variant: str,
        isbn: str,
        publisher: str,
        published_year: int | None,
        source: str,
        *,
        connection: sqlite3.Connection | None = None,
    ) -> int:
        with self._writer(connection) as active:
            row = active.execute(
                """INSERT INTO editions(work_id, language, variant, isbn, publisher, published_year, source, created_at)
                   VALUES (?, ?, ?, ?, ?, ?, ?, ?)
                   ON CONFLICT(work_id, language, variant, isbn) DO UPDATE SET
                       publisher=CASE WHEN excluded.publisher!='' THEN excluded.publisher ELSE editions.publisher END,
                       published_year=COALESCE(excluded.published_year, editions.published_year),
                       source=CASE WHEN excluded.source!='' THEN excluded.source ELSE editions.source END
                   RETURNING id""",
                (work_id, language, variant, isbn, publisher, published_year, source, now()),
            ).fetchone()
            assert row is not None
            return int(row["id"])

    def asset_by_hash(self, sha256: str) -> sqlite3.Row | None:
        with self.connect() as connection:
            return connection.execute("SELECT * FROM assets WHERE sha256 = ?", (sha256,)).fetchone()

    def add_asset(
        self,
        edition_id: int,
        fmt: str,
        filename: str,
        path: Path,
        sha256: str,
        size_bytes: int,
        mime_type: str,
        metadata: dict[str, Any],
        *,
        connection: sqlite3.Connection | None = None,
    ) -> int:
        with self._writer(connection) as active:
            cursor = active.execute(
                """INSERT INTO assets(edition_id, format, filename, path, sha256, size_bytes, mime_type, metadata_json, created_at)
                   VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)""",
                (
                    edition_id,
                    fmt,
                    filename,
                    str(path),
                    sha256,
                    size_bytes,
                    mime_type,
                    json.dumps(metadata, ensure_ascii=False, sort_keys=True),
                    now(),
                ),
            )
            asset_id = int(cursor.lastrowid)
            # Project the identifiers out of the JSON blob into an indexed table.
            # Looking them up used to mean reading every book asset row and
            # parsing its metadata in Python.
            active.executemany(
                "INSERT OR IGNORE INTO asset_identifiers(asset_id, identifier) VALUES (?, ?)",
                [
                    (asset_id, key)
                    for key in _identifier_keys(metadata.get("source_identifiers"))
                ],
            )
            return asset_id

    def cleanup_empty_work(self, work_id: int) -> None:
        with self.connect() as connection:
            connection.execute(
                "DELETE FROM editions WHERE work_id=? AND NOT EXISTS (SELECT 1 FROM assets WHERE assets.edition_id=editions.id)",
                (work_id,),
            )
            connection.execute(
                "DELETE FROM works WHERE id=? AND NOT EXISTS (SELECT 1 FROM editions WHERE editions.work_id=works.id)",
                (work_id,),
            )

    def works(self) -> list[sqlite3.Row]:
        with self.connect() as connection:
            return list(
                connection.execute(
                    """SELECT w.*, COUNT(DISTINCT e.id) AS edition_count, COUNT(a.id) AS asset_count
                       FROM works w
                       LEFT JOIN editions e ON e.work_id = w.id
                       LEFT JOIN assets a ON a.edition_id = e.id
                       GROUP BY w.id ORDER BY w.updated_at DESC, w.id DESC"""
                )
            )

    def work(self, work_id: int, *, connection: sqlite3.Connection | None = None) -> sqlite3.Row | None:
        with self._writer(connection) as active:
            return active.execute("SELECT * FROM works WHERE id = ?", (work_id,)).fetchone()

    def work_assets(self, work_id: int) -> list[sqlite3.Row]:
        with self.connect() as connection:
            return list(
                connection.execute(
                    """SELECT a.*, e.language, e.variant, e.isbn, e.publisher, e.published_year, e.source
                       FROM assets a JOIN editions e ON e.id = a.edition_id
                       WHERE e.work_id = ? ORDER BY e.language, e.variant, a.format""",
                    (work_id,),
                )
            )

    def asset(self, asset_id: int) -> sqlite3.Row | None:
        with self.connect() as connection:
            return connection.execute(
                """SELECT a.*, e.language, e.variant, e.work_id, w.title, w.author
                   FROM assets a JOIN editions e ON e.id = a.edition_id JOIN works w ON w.id = e.work_id
                   WHERE a.id = ?""",
                (asset_id,),
            ).fetchone()

    def recent_jobs(self, limit: int = 20) -> list[sqlite3.Row]:
        with self.connect() as connection:
            # Reads the control ledger. workflow_jobs links to control_intents and
            # control_plans, so an activity row can be traced to the intent that
            # produced it -- activity_jobs was a parallel log with no such link.
            return list(
                connection.execute(
                    """SELECT j.*, p.action AS plan_action, p.risk AS plan_risk
                       FROM workflow_jobs j
                       LEFT JOIN control_plans p ON p.id = j.plan_id
                       ORDER BY j.id DESC LIMIT ?""",
                    (limit,),
                )
            )

    def wanted(self, limit: int = 100) -> list[sqlite3.Row]:
        with self.connect() as connection:
            return list(connection.execute("SELECT * FROM wanted_books ORDER BY id DESC LIMIT ?", (limit,)))

    def counts(self) -> dict[str, int]:
        with self.connect() as connection:
            result = {}
            for table in ("works", "editions", "assets"):
                result[table] = int(connection.execute(f"SELECT COUNT(*) FROM {table}").fetchone()[0])
            result["inbox_items"] = int(
                connection.execute("SELECT COUNT(*) FROM inbox_items WHERE media_type='source'").fetchone()[0]
            )
            result["media_candidates"] = int(
                connection.execute("SELECT COUNT(*) FROM media_candidates WHERE status!='superseded'").fetchone()[0]
            )
            result["wanted_books"] = int(
                connection.execute("SELECT COUNT(*) FROM wanted_books WHERE status = 'wanted'").fetchone()[0]
            )
            result["pending_candidates"] = int(
                connection.execute("SELECT COUNT(*) FROM media_candidates WHERE status='pending'").fetchone()[0]
            )
            return result

    def backup(self, destination: Path) -> None:
        # closing() is required: sqlite3's own context manager commits or rolls
        # back the transaction and does NOT close the connection. Written as
        # `with sqlite3.connect(...) as target:` this leaked a file handle on
        # every backup -- twice per daily maintenance run, and once per call in
        # the tests, which is where ResourceWarning surfaced it.
        destination.parent.mkdir(parents=True, exist_ok=True)
        with self.connect() as source, closing(sqlite3.connect(destination)) as target:
            source.backup(target)
        with closing(sqlite3.connect(destination)) as check:
            result = check.execute("PRAGMA integrity_check").fetchone()[0]
            if result != "ok":
                destination.unlink(missing_ok=True)
                raise RuntimeError(f"backup integrity check failed: {result}")