def inspect_custom_fields_registry(connection: sqlite3.Connection) -> dict[str, object]:
    columns = table_info(connection, "documents")
    column_map = {row["name"]: row for row in columns}
    actual_custom_fields = [
        name
        for name in column_map
        if name not in BUILTIN_FIELD_TYPES
        and name not in INTERNAL_DOCUMENT_COLUMNS
        and name not in {LEGACY_METADATA_LOCKS_COLUMN}
    ]
    registry_rows = connection.execute(
        """
        SELECT field_name, field_type, instruction, created_at
        FROM custom_fields_registry
        ORDER BY field_name ASC
        """
    ).fetchall()
    registry_fields = {row["field_name"]: row for row in registry_rows}
    missing_registry = sorted(name for name in actual_custom_fields if name not in registry_fields)
    orphaned_registry = sorted(name for name in registry_fields if name not in column_map)
    return {
        "actual_custom_fields": sorted(actual_custom_fields),
        "missing_registry": missing_registry,
        "orphaned_registry": orphaned_registry,
    }


def reconcile_custom_fields_registry(connection: sqlite3.Connection, repair: bool) -> dict[str, object]:
    status = inspect_custom_fields_registry(connection)
    repaired: list[str] = []
    if repair:
        now = utc_now()
        for field_name in status["missing_registry"]:
            sqlite_type = next(
                (row["type"] for row in table_info(connection, "documents") if row["name"] == field_name),
                "",
            )
            connection.execute(
                """
                INSERT INTO custom_fields_registry (field_name, field_type, instruction, created_at)
                VALUES (?, ?, ?, ?)
                ON CONFLICT(field_name) DO NOTHING
                """,
                (field_name, infer_registry_field_type(sqlite_type), None, now),
            )
            repaired.append(field_name)
        if repaired:
            connection.commit()
        status = inspect_custom_fields_registry(connection)
    status["repaired_registry"] = repaired
    return status


def merge_legacy_field_locks(connection: sqlite3.Connection) -> int:
    columns = table_columns(connection, "documents")
    if MANUAL_FIELD_LOCKS_COLUMN not in columns or LEGACY_METADATA_LOCKS_COLUMN not in columns:
        return 0

    rows = connection.execute(
        f"""
        SELECT id, {quote_identifier(MANUAL_FIELD_LOCKS_COLUMN)} AS manual_locks,
               {quote_identifier(LEGACY_METADATA_LOCKS_COLUMN)} AS legacy_locks
        FROM documents
        """
    ).fetchall()

    merged_count = 0
    for row in rows:
        manual_locks = normalize_string_list(row["manual_locks"])
        legacy_locks = normalize_string_list(row["legacy_locks"])
        if not legacy_locks:
            continue

        merged = list(manual_locks)
        for field_name in legacy_locks:
            if field_name not in merged:
                merged.append(field_name)
        if merged == manual_locks:
            continue

        connection.execute(
            f"""
            UPDATE documents
            SET {quote_identifier(MANUAL_FIELD_LOCKS_COLUMN)} = ?
            WHERE id = ?
            """,
            (json.dumps(merged), row["id"]),
        )
        merged_count += 1
    return merged_count


def backfill_content_type(connection: sqlite3.Connection) -> int:
    columns = table_columns(connection, "documents")
    if "content_type" not in columns:
        return 0

    rows = connection.execute(
        """
        SELECT id, file_type, content_type
        FROM documents
        WHERE content_type IS NULL OR TRIM(content_type) = ''
        """
    ).fetchall()
    updated = 0
    for row in rows:
        inferred = infer_content_type_from_extension((row["file_type"] or "").lower())
        if not inferred:
            continue
        connection.execute(
            "UPDATE documents SET content_type = ? WHERE id = ?",
            (inferred, row["id"]),
        )
        updated += 1
    return updated


def backfill_child_document_kinds(connection: sqlite3.Connection) -> int:
    columns = table_columns(connection, "documents")
    required = {"parent_document_id", "child_document_kind"}
    if not required.issubset(columns):
        return 0
    cursor = connection.execute(
        """
        UPDATE documents
        SET child_document_kind = ?
        WHERE parent_document_id IS NOT NULL
          AND (child_document_kind IS NULL OR TRIM(child_document_kind) = '')
        """,
        (CHILD_DOCUMENT_KIND_ATTACHMENT,),
    )
    return int(cursor.rowcount or 0)


def backfill_conversation_assignment_modes(connection: sqlite3.Connection) -> int:
    columns = table_columns(connection, "documents")
    if "conversation_assignment_mode" not in columns:
        return 0
    cursor = connection.execute(
        """
        UPDATE documents
        SET conversation_assignment_mode = ?
        WHERE conversation_assignment_mode IS NULL
          OR TRIM(conversation_assignment_mode) = ''
        """,
        (CONVERSATION_ASSIGNMENT_MODE_AUTO,),
    )
    return int(cursor.rowcount or 0)


def backfill_custodian(connection: sqlite3.Connection) -> int:
    columns = table_columns(connection, "documents")
    required = {"custodian", "source_kind", "source_rel_path", "parent_document_id"}
    if not required.issubset(columns):
        return 0

    rows = connection.execute(
        """
        SELECT id, source_kind, source_rel_path, parent_document_id
        FROM documents
        WHERE custodian IS NULL OR TRIM(custodian) = ''
        ORDER BY CASE WHEN parent_document_id IS NULL THEN 0 ELSE 1 END ASC, id ASC
        """
    ).fetchall()
    updated = 0
    for row in rows:
        parent_custodian = None
        parent_document_id = row["parent_document_id"]
        if parent_document_id is not None:
            parent_row = connection.execute(
                "SELECT custodian FROM documents WHERE id = ?",
                (parent_document_id,),
            ).fetchone()
            if parent_row is not None:
                parent_custodian = parent_row["custodian"]
        custodian = infer_source_custodian(
            source_kind=row["source_kind"],
            source_rel_path=row["source_rel_path"],
            parent_custodian=parent_custodian,
        )
        if not custodian:
            continue
        connection.execute(
            "UPDATE documents SET custodian = ? WHERE id = ?",
            (custodian, row["id"]),
        )
        updated += 1
    return updated


def backfill_dataset_ids(connection: sqlite3.Connection, root: Path | None = None) -> int:
    if not table_exists(connection, "datasets") or not table_exists(connection, "documents"):
        return 0

    productions_by_id: dict[int, sqlite3.Row] = {}
    if table_exists(connection, "productions"):
        production_rows = connection.execute(
            """
            SELECT id, dataset_id, rel_root, production_name
            FROM productions
            ORDER BY id ASC
            """
        ).fetchall()
        for row in production_rows:
            productions_by_id[int(row["id"])] = row

    updated = 0

    if productions_by_id:
        for row in productions_by_id.values():
            if row["dataset_id"] is not None:
                continue
            dataset_id = ensure_dataset_row(
                connection,
                source_kind=PRODUCTION_SOURCE_KIND,
                dataset_locator=str(row["rel_root"]),
                dataset_name=production_dataset_name(str(row["rel_root"]), str(row["production_name"] or "")),
            )
            connection.execute(
                "UPDATE productions SET dataset_id = ? WHERE id = ?",
                (dataset_id, row["id"]),
            )
            updated += 1

    if table_exists(connection, "container_sources"):
        container_rows = connection.execute(
            """
            SELECT id, dataset_id, source_kind, source_rel_path
            FROM container_sources
            ORDER BY id ASC
            """
        ).fetchall()
        for row in container_rows:
            if row["dataset_id"] is not None:
                continue
            source_kind = str(row["source_kind"] or "")
            source_rel_path = str(row["source_rel_path"] or "")
            if not source_kind or not source_rel_path:
                continue
            dataset_name = (
                pst_dataset_name(source_rel_path)
                if source_kind == PST_SOURCE_KIND
                else mbox_dataset_name(source_rel_path)
                if source_kind == MBOX_SOURCE_KIND
                else source_rel_path
            )
            dataset_id = ensure_dataset_row(
                connection,
                source_kind=source_kind,
                dataset_locator=source_rel_path,
                dataset_name=dataset_name,
            )
            connection.execute(
                "UPDATE container_sources SET dataset_id = ? WHERE id = ?",
                (dataset_id, row["id"]),
            )
            updated += 1

    document_columns = table_columns(connection, "documents")
    required_document_columns = {"id", "dataset_id", "parent_document_id", "source_kind", "source_rel_path", "production_id"}
    if not required_document_columns.issubset(document_columns):
        return updated

    filesystem_dataset_id: int | None = None
    rows = connection.execute(
        """
        SELECT id, dataset_id, parent_document_id, source_kind, source_rel_path, production_id
        FROM documents
        WHERE dataset_id IS NULL
        ORDER BY CASE WHEN parent_document_id IS NULL THEN 0 ELSE 1 END ASC, id ASC
        """
    ).fetchall()
    for row in rows:
        dataset_id: int | None = None
        parent_document_id = row["parent_document_id"]
        if parent_document_id is not None:
            parent_row = connection.execute(
                "SELECT dataset_id FROM documents WHERE id = ?",
                (parent_document_id,),
            ).fetchone()
            if parent_row is not None and parent_row["dataset_id"] is not None:
                dataset_id = int(parent_row["dataset_id"])
        if dataset_id is None and row["production_id"] is not None:
            production_row = productions_by_id.get(int(row["production_id"]))
            if production_row is not None:
                if production_row["dataset_id"] is None:
                    dataset_id = ensure_dataset_row(
                        connection,
                        source_kind=PRODUCTION_SOURCE_KIND,
                        dataset_locator=str(production_row["rel_root"]),
                        dataset_name=production_dataset_name(
                            str(production_row["rel_root"]),
                            str(production_row["production_name"] or ""),
                        ),
                    )
                    connection.execute(
                        "UPDATE productions SET dataset_id = ? WHERE id = ?",
                        (dataset_id, production_row["id"]),
                    )
                    productions_by_id[int(production_row["id"])] = connection.execute(
                        "SELECT id, dataset_id, rel_root, production_name FROM productions WHERE id = ?",
                        (production_row["id"],),
                    ).fetchone()
                    updated += 1
                else:
                    dataset_id = int(production_row["dataset_id"])
        source_kind = normalize_whitespace(str(row["source_kind"] or "")).lower()
        source_rel_path = normalize_whitespace(str(row["source_rel_path"] or ""))
        if dataset_id is None and source_kind in {PST_SOURCE_KIND, MBOX_SOURCE_KIND} and source_rel_path:
            dataset_id = ensure_dataset_row(
                connection,
                source_kind=source_kind,
                dataset_locator=source_rel_path,
                dataset_name=(
                    pst_dataset_name(source_rel_path)
                    if source_kind == PST_SOURCE_KIND
                    else mbox_dataset_name(source_rel_path)
                ),
            )
        if dataset_id is None:
            if filesystem_dataset_id is None:
                filesystem_dataset_id = ensure_dataset_row(
                    connection,
                    source_kind=FILESYSTEM_SOURCE_KIND,
                    dataset_locator=filesystem_dataset_locator(),
                    dataset_name=filesystem_dataset_name(root),
                )
            dataset_id = filesystem_dataset_id
        connection.execute(
            "UPDATE documents SET dataset_id = ? WHERE id = ?",
            (dataset_id, row["id"]),
        )
        updated += 1

    return updated


def backfill_dataset_memberships(connection: sqlite3.Connection) -> int:
    if not table_exists(connection, "dataset_documents") or not table_exists(connection, "datasets"):
        return 0

    updated = 0
    dataset_rows = connection.execute(
        """
        SELECT id, source_kind, dataset_locator
        FROM datasets
        ORDER BY id ASC
        """
    ).fetchall()
    for row in dataset_rows:
        source_kind = normalize_whitespace(str(row["source_kind"] or "")).lower()
        source_locator = normalize_whitespace(str(row["dataset_locator"] or ""))
        if not source_kind or not source_locator or source_kind == MANUAL_DATASET_SOURCE_KIND:
            continue
        ensure_dataset_source_row(
            connection,
            dataset_id=int(row["id"]),
            source_kind=source_kind,
            source_locator=source_locator,
        )

    existing_membership_count = int(
        connection.execute("SELECT COUNT(*) AS count FROM dataset_documents").fetchone()["count"] or 0
    )
    if existing_membership_count > 0:
        return updated

    productions_by_id: dict[int, sqlite3.Row] = {}
    if table_exists(connection, "productions"):
        for row in connection.execute(
            """
            SELECT id, dataset_id, rel_root
            FROM productions
            ORDER BY id ASC
            """
        ).fetchall():
            productions_by_id[int(row["id"])] = row

    document_columns = table_columns(connection, "documents")
    required_document_columns = {"id", "dataset_id", "parent_document_id", "source_kind", "source_rel_path", "production_id"}
    if not required_document_columns.issubset(document_columns):
        return updated

    rows = connection.execute(
        """
        SELECT id, dataset_id, parent_document_id, source_kind, source_rel_path, production_id
        FROM documents
        WHERE dataset_id IS NOT NULL
        ORDER BY CASE WHEN parent_document_id IS NULL THEN 0 ELSE 1 END ASC, id ASC
        """
    ).fetchall()
    for row in rows:
        dataset_id = int(row["dataset_id"])
        document_id = int(row["id"])
        source_membership_ids: list[int] = []

        if row["parent_document_id"] is not None:
            parent_memberships = connection.execute(
                """
                SELECT dataset_source_id
                FROM dataset_documents
                WHERE dataset_id = ? AND document_id = ? AND dataset_source_id IS NOT NULL
                ORDER BY dataset_source_id ASC
                """,
                (dataset_id, int(row["parent_document_id"])),
            ).fetchall()
            source_membership_ids = [int(item["dataset_source_id"]) for item in parent_memberships]

        if not source_membership_ids and row["production_id"] is not None:
            production_row = productions_by_id.get(int(row["production_id"]))
            if production_row is not None:
                dataset_source_row = get_dataset_source_row(
                    connection,
                    source_kind=PRODUCTION_SOURCE_KIND,
                    source_locator=str(production_row["rel_root"]),
                )
                if dataset_source_row is not None and int(dataset_source_row["dataset_id"]) == dataset_id:
                    source_membership_ids.append(int(dataset_source_row["id"]))

        source_kind = normalize_whitespace(str(row["source_kind"] or "")).lower()
        source_rel_path = normalize_whitespace(str(row["source_rel_path"] or ""))
        if not source_membership_ids and source_kind in {PST_SOURCE_KIND, MBOX_SOURCE_KIND} and source_rel_path:
            dataset_source_row = get_dataset_source_row(
                connection,
                source_kind=source_kind,
                source_locator=source_rel_path,
            )
            if dataset_source_row is not None and int(dataset_source_row["dataset_id"]) == dataset_id:
                source_membership_ids.append(int(dataset_source_row["id"]))

        if not source_membership_ids and source_kind in {FILESYSTEM_SOURCE_KIND, EMAIL_ATTACHMENT_SOURCE_KIND}:
            dataset_source_row = get_dataset_source_row(
                connection,
                source_kind=FILESYSTEM_SOURCE_KIND,
                source_locator=filesystem_dataset_locator(),
            )
            if dataset_source_row is not None and int(dataset_source_row["dataset_id"]) == dataset_id:
                source_membership_ids.append(int(dataset_source_row["id"]))

        if source_membership_ids:
            for dataset_source_id in source_membership_ids:
                ensure_dataset_document_membership(
                    connection,
                    dataset_id=dataset_id,
                    document_id=document_id,
                    dataset_source_id=dataset_source_id,
                )
                updated += 1
            continue

        ensure_dataset_document_membership(
            connection,
            dataset_id=dataset_id,
            document_id=document_id,
            dataset_source_id=None,
        )
        updated += 1

    return updated


def backfill_occurrence_dataset_source_ids(connection: sqlite3.Connection) -> int:
    if not table_exists(connection, "document_occurrences") or not table_exists(connection, "dataset_sources"):
        return 0
    columns = table_columns(connection, "document_occurrences")
    if "dataset_source_id" not in columns:
        return 0
    rows = connection.execute(
        """
        SELECT *
        FROM document_occurrences
        WHERE dataset_source_id IS NULL
        ORDER BY id ASC
        """
    ).fetchall()
    updated = 0
    for row in rows:
        dataset_source_row = dataset_source_row_for_occurrence(connection, row)
        if dataset_source_row is None:
            continue
        connection.execute(
            """
            UPDATE document_occurrences
            SET dataset_source_id = ?, updated_at = COALESCE(updated_at, ?)
            WHERE id = ?
            """,
            (int(dataset_source_row["id"]), utc_now(), int(row["id"])),
        )
        updated += 1
    return updated


def backfill_dataset_name_normalized(connection: sqlite3.Connection) -> int:
    if not table_exists(connection, "datasets"):
        return 0
    columns = table_columns(connection, "datasets")
    if "dataset_name" not in columns or "dataset_name_normalized" not in columns:
        return 0
    rows = connection.execute(
        """
        SELECT id, dataset_name, dataset_name_normalized
        FROM datasets
        ORDER BY id ASC
        """
    ).fetchall()
    updated = 0
    for row in rows:
        dataset_name = normalized_dataset_name_or_default(str(row["dataset_name"] or ""))
        normalized_name = normalize_dataset_name_for_compare(dataset_name)
        if (
            normalize_inline_whitespace(str(row["dataset_name"] or "")) == dataset_name
            and normalize_inline_whitespace(str(row["dataset_name_normalized"] or "")) == normalized_name
        ):
            continue
        connection.execute(
            """
            UPDATE datasets
            SET dataset_name = ?, dataset_name_normalized = ?, updated_at = ?
            WHERE id = ?
            """,
            (dataset_name, normalized_name, utc_now(), int(row["id"])),
        )
        updated += 1
    return updated


def merge_dataset_identity_duplicates(connection: sqlite3.Connection) -> int:
    if not table_exists(connection, "datasets"):
        return 0
    rows = connection.execute(
        """
        SELECT id, source_kind, dataset_locator
        FROM datasets
        ORDER BY LOWER(source_kind) ASC, dataset_locator ASC, id ASC
        """
    ).fetchall()
    ids_by_identity: dict[tuple[str, str], list[int]] = defaultdict(list)
    for row in rows:
        identity = (
            normalize_whitespace(str(row["source_kind"] or "")).lower(),
            normalize_whitespace(str(row["dataset_locator"] or "")),
        )
        ids_by_identity[identity].append(int(row["id"]))

    merged = 0
    for duplicate_ids in ids_by_identity.values():
        if len(duplicate_ids) < 2:
            continue
        keep_id = min(duplicate_ids)
        for drop_id in sorted(dataset_id for dataset_id in duplicate_ids if dataset_id != keep_id):
            source_rows = connection.execute(
                """
                SELECT id, source_kind, source_locator, created_at, updated_at
                FROM dataset_sources
                WHERE dataset_id = ?
                ORDER BY id ASC
                """,
                (drop_id,),
            ).fetchall()
            source_id_map: dict[int, int] = {}
            for source_row in source_rows:
                new_source_id = ensure_dataset_source_row(
                    connection,
                    dataset_id=keep_id,
                    source_kind=str(source_row["source_kind"]),
                    source_locator=str(source_row["source_locator"]),
                )
                source_id_map[int(source_row["id"])] = new_source_id

            membership_rows = connection.execute(
                """
                SELECT id, document_id, dataset_source_id, created_at, updated_at
                FROM dataset_documents
                WHERE dataset_id = ?
                ORDER BY id ASC
                """,
                (drop_id,),
            ).fetchall()
            for membership_row in membership_rows:
                dataset_source_id = membership_row["dataset_source_id"]
                remapped_source_id = (
                    source_id_map.get(int(dataset_source_id))
                    if dataset_source_id is not None
                    else None
                )
                connection.execute(
                    """
                    INSERT OR IGNORE INTO dataset_documents (
                      dataset_id, document_id, dataset_source_id, created_at, updated_at
                    ) VALUES (?, ?, ?, ?, ?)
                    """,
                    (
                        keep_id,
                        int(membership_row["document_id"]),
                        remapped_source_id,
                        membership_row["created_at"] or utc_now(),
                        membership_row["updated_at"] or utc_now(),
                    ),
                )
                connection.execute(
                    "DELETE FROM dataset_documents WHERE id = ?",
                    (int(membership_row["id"]),),
                )

            connection.execute("UPDATE documents SET dataset_id = ? WHERE dataset_id = ?", (keep_id, drop_id))
            connection.execute("UPDATE productions SET dataset_id = ? WHERE dataset_id = ?", (keep_id, drop_id))
            connection.execute("UPDATE container_sources SET dataset_id = ? WHERE dataset_id = ?", (keep_id, drop_id))
            connection.execute("DELETE FROM dataset_sources WHERE dataset_id = ?", (drop_id,))
            connection.execute("DELETE FROM datasets WHERE id = ?", (drop_id,))
            merged += 1
    return merged


def suffix_dataset_name_collisions(connection: sqlite3.Connection) -> int:
    if not table_exists(connection, "datasets"):
        return 0
    rows = connection.execute(
        """
        SELECT id, dataset_name
        FROM datasets
        ORDER BY id ASC
        """
    ).fetchall()
    renamed = 0
    used_names: set[str] = set()
    for row in rows:
        dataset_id = int(row["id"])
        base_name = normalized_dataset_name_or_default(str(row["dataset_name"] or ""))
        candidate = base_name
        suffix = 2
        normalized_candidate = normalize_dataset_name_for_compare(candidate)
        while normalized_candidate in used_names:
            candidate = f"{base_name}_{suffix}"
            normalized_candidate = normalize_dataset_name_for_compare(candidate)
            suffix += 1
        used_names.add(normalized_candidate)
        if candidate == base_name and normalize_inline_whitespace(str(row["dataset_name"] or "")) == base_name:
            connection.execute(
                """
                UPDATE datasets
                SET dataset_name_normalized = ?, updated_at = ?
                WHERE id = ?
                """,
                (normalized_candidate, utc_now(), dataset_id),
            )
            continue
        connection.execute(
            """
            UPDATE datasets
            SET dataset_name = ?, dataset_name_normalized = ?, updated_at = ?
            WHERE id = ?
            """,
            (candidate, normalized_candidate, utc_now(), dataset_id),
        )
        if candidate != base_name or normalize_inline_whitespace(str(row["dataset_name"] or "")) != base_name:
            renamed += 1
    return renamed


def ensure_control_number_batch_row(
    connection: sqlite3.Connection,
    batch_number: int,
    next_family_sequence: int,
) -> int:
    row = connection.execute(
        """
        SELECT next_family_sequence
        FROM control_number_batches
        WHERE batch_number = ?
        """,
        (batch_number,),
    ).fetchone()
    now = utc_now()
    if row is None:
        connection.execute(
            """
            INSERT INTO control_number_batches (batch_number, next_family_sequence, created_at, updated_at)
            VALUES (?, ?, ?, ?)
            """,
            (batch_number, next_family_sequence, now, now),
        )
        return next_family_sequence
    normalized_next = max(int(row["next_family_sequence"]), next_family_sequence)
    if normalized_next != int(row["next_family_sequence"]):
        connection.execute(
            """
            UPDATE control_number_batches
            SET next_family_sequence = ?, updated_at = ?
            WHERE batch_number = ?
            """,
            (normalized_next, now, batch_number),
        )
    return normalized_next


def backfill_control_number_batches(connection: sqlite3.Connection) -> int:
    columns = table_columns(connection, "documents")
    required = {"control_number_batch", "control_number_family_sequence"}
    if not required.issubset(columns):
        return 0
    rows = connection.execute(
        """
        SELECT control_number_batch, MAX(control_number_family_sequence) AS max_family_sequence
        FROM documents
        WHERE control_number_batch IS NOT NULL AND control_number_family_sequence IS NOT NULL
        GROUP BY control_number_batch
        """
    ).fetchall()
    updated = 0
    for row in rows:
        before = connection.execute(
            """
            SELECT next_family_sequence
            FROM control_number_batches
            WHERE batch_number = ?
            """,
            (row["control_number_batch"],),
        ).fetchone()
        next_sequence = int(row["max_family_sequence"] or 0) + 1
        ensure_control_number_batch_row(connection, int(row["control_number_batch"]), next_sequence)
        if before is None or int(before["next_family_sequence"]) != next_sequence:
            updated += 1
    return updated


def backfill_document_occurrences(connection: sqlite3.Connection) -> int:
    if not table_exists(connection, "documents") or not table_exists(connection, "document_occurrences"):
        return 0

    missing_document_rows = connection.execute(
        """
        SELECT *
        FROM documents
        WHERE NOT EXISTS (
          SELECT 1
          FROM document_occurrences
          WHERE document_occurrences.document_id = documents.id
        )
          AND canonical_status != ?
        ORDER BY CASE WHEN parent_document_id IS NULL THEN 0 ELSE 1 END ASC, id ASC
        """,
        (CANONICAL_STATUS_MERGED,),
    ).fetchall()
    preview_counts = {
        int(row["document_id"]): int(row["count"] or 0)
        for row in connection.execute(
            """
            SELECT document_id, COUNT(*) AS count
            FROM document_previews
            WHERE document_id IN (
              SELECT id
              FROM documents
              WHERE NOT EXISTS (
                SELECT 1
                FROM document_occurrences
                WHERE document_occurrences.document_id = documents.id
              )
            )
            GROUP BY document_id
            """
        ).fetchall()
    } if missing_document_rows else {}

    inserted = 0
    for row in missing_document_rows:
        document_id = int(row["id"])
        legacy_custodian = None
        if "custodian" in row.keys():
            legacy_custodian = row["custodian"]
        elif "custodians_json" in row.keys():
            custodian_values = parse_document_custodians_json(row["custodians_json"])
            legacy_custodian = custodian_values[0] if len(custodian_values) == 1 else None
        occurrence_id = upsert_document_occurrence(
            connection,
            document_id=document_id,
            existing_occurrence_id=None,
            parent_occurrence_id=None,
            occurrence_control_number=row["control_number"],
            source_kind=row["source_kind"],
            source_rel_path=row["source_rel_path"] or row["rel_path"],
            source_item_id=row["source_item_id"],
            source_folder_path=row["source_folder_path"],
            production_id=row["production_id"],
            begin_bates=row["begin_bates"],
            end_bates=row["end_bates"],
            begin_attachment=row["begin_attachment"],
            end_attachment=row["end_attachment"],
            rel_path=str(row["rel_path"]),
            file_name=str(row["file_name"]),
            file_type=row["file_type"],
            mime_type=None,
            file_size=row["file_size"],
            file_hash=row["file_hash"],
            custodian=legacy_custodian,
            fs_created_at=None,
            fs_modified_at=None,
            extracted={
                "author": row["author"],
                "title": row["title"],
                "subject": row["subject"],
                "participants": row["participants"],
                "recipients": row["recipients"],
                "date_created": row["date_created"],
                "date_modified": row["date_modified"],
                "content_type": row["content_type"],
            },
            has_preview=preview_counts.get(document_id, 0) > 0,
            text_status=str(row["text_status"] or "ok"),
            ingested_at=str(row["ingested_at"] or row["updated_at"] or utc_now()),
            last_seen_at=str(row["last_seen_at"] or row["ingested_at"] or row["updated_at"] or utc_now()),
            updated_at=str(row["updated_at"] or row["last_seen_at"] or row["ingested_at"] or utc_now()),
        )
        inserted += 1

    child_rows = connection.execute(
        """
        SELECT child.id, child.parent_document_id, child_occurrence.id AS child_occurrence_id
        FROM documents child
        JOIN document_occurrences child_occurrence ON child_occurrence.document_id = child.id
        WHERE child.parent_document_id IS NOT NULL
          AND child_occurrence.parent_occurrence_id IS NULL
        ORDER BY child.id ASC, child_occurrence.id ASC
        """
    ).fetchall()
    for row in child_rows:
        parent_occurrence = connection.execute(
            """
            SELECT id
            FROM document_occurrences
            WHERE document_id = ?
            ORDER BY id ASC
            LIMIT 1
            """,
            (int(row["parent_document_id"]),),
        ).fetchone()
        if parent_occurrence is None:
            continue
        connection.execute(
            """
            UPDATE document_occurrences
            SET parent_occurrence_id = ?
            WHERE id = ? AND parent_occurrence_id IS NULL
            """,
            (int(parent_occurrence["id"]), int(row["child_occurrence_id"])),
        )
    return inserted


def drop_document_field_locks(connection: sqlite3.Connection, field_name: str) -> int:
    if not table_exists(connection, "documents"):
        return 0
    rows = connection.execute(
        f"""
        SELECT id, {quote_identifier(MANUAL_FIELD_LOCKS_COLUMN)} AS locks_json
        FROM documents
        WHERE {quote_identifier(MANUAL_FIELD_LOCKS_COLUMN)} LIKE ?
        """,
        (f'%"{field_name}"%',),
    ).fetchall()
    updated = 0
    for row in rows:
        locks = normalize_string_list(row["locks_json"])
        filtered_locks = [lock_name for lock_name in locks if lock_name != field_name]
        if filtered_locks == locks:
            continue
        connection.execute(
            f"""
            UPDATE documents
            SET {quote_identifier(MANUAL_FIELD_LOCKS_COLUMN)} = ?
            WHERE id = ?
            """,
            (json.dumps(filtered_locks), int(row["id"])),
        )
        updated += 1
    return updated


def remove_documents_custodian_column(connection: sqlite3.Connection) -> bool:
    if "custodian" not in table_columns(connection, "documents"):
        return False
    connection.execute("ALTER TABLE documents DROP COLUMN custodian")
    return True


def backfill_document_dedupe_keys(connection: sqlite3.Connection) -> int:
    if not table_exists(connection, "documents") or not table_exists(connection, "document_dedupe_keys"):
        return 0
    before = connection.total_changes
    rows = connection.execute(
        """
        WITH desired_keys AS (
          SELECT
            file_hash,
            MAX(id) AS document_id
          FROM documents
          WHERE parent_document_id IS NULL
            AND file_hash IS NOT NULL
            AND TRIM(file_hash) != ''
            AND lifecycle_status != 'deleted'
            AND LOWER(COALESCE(source_kind, '')) IN (?, ?, ?)
          GROUP BY file_hash
        )
        SELECT desired_keys.document_id AS id, desired_keys.file_hash
        FROM desired_keys
        LEFT JOIN document_dedupe_keys existing_key
          ON existing_key.basis = 'file_hash'
         AND existing_key.key_value = desired_keys.file_hash
        WHERE existing_key.document_id IS NULL
           OR existing_key.document_id != desired_keys.document_id
        ORDER BY desired_keys.document_id ASC
        """,
        (FILESYSTEM_SOURCE_KIND, PST_SOURCE_KIND, MBOX_SOURCE_KIND),
    ).fetchall()
    for row in rows:
        bind_document_dedupe_key(
            connection,
            basis="file_hash",
            key_value=str(row["file_hash"]),
            document_id=int(row["id"]),
        )
    return connection.total_changes - before


def refresh_documents_from_occurrences(connection: sqlite3.Connection) -> int:
    if not table_exists(connection, "documents") or not table_exists(connection, "document_occurrences"):
        return 0
    rows = connection.execute(
        """
        SELECT id
        FROM documents
        ORDER BY CASE WHEN parent_document_id IS NULL THEN 0 ELSE 1 END ASC, id ASC
        """
    ).fetchall()
    for row in rows:
        refresh_document_from_occurrences(connection, int(row["id"]))
    return len(rows)


def should_refresh_document_occurrence_caches(
    *,
    prior_schema_version: int | None,
    backfilled_occurrence_dataset_source_ids: int,
    backfilled_document_occurrences: int,
) -> bool:
    # A full occurrence refresh also rebuilds document entity links; keep it off
    # the read-only command path once the workspace schema is current.
    if prior_schema_version != SCHEMA_VERSION:
        return True
    return backfilled_occurrence_dataset_source_ids > 0 or backfilled_document_occurrences > 0


def backfill_control_numbers(connection: sqlite3.Connection) -> int:
    columns = table_columns(connection, "documents")
    required = {
        "control_number",
        "parent_document_id",
        "control_number_batch",
        "control_number_family_sequence",
        "control_number_attachment_sequence",
        "source_kind",
    }
    if not required.issubset(columns):
        return 0

    backfill_control_number_batches(connection)
    updated = 0

    top_level_rows = connection.execute(
        """
        SELECT id, control_number
        FROM documents
        WHERE parent_document_id IS NULL
          AND COALESCE(source_kind, ?) != ?
          AND (
            control_number IS NULL OR TRIM(control_number) = ''
            OR control_number_batch IS NULL
            OR control_number_family_sequence IS NULL
          )
        ORDER BY id ASC
        """
    , (FILESYSTEM_SOURCE_KIND, PRODUCTION_SOURCE_KIND)).fetchall()
    if top_level_rows:
        next_family_sequence = ensure_control_number_batch_row(connection, 1, 1)
        for row in top_level_rows:
            parsed_identity = parse_control_number(row["control_number"])
            if parsed_identity is not None and parsed_identity[2] is None:
                batch_number, family_sequence, _ = parsed_identity
                control_number = str(row["control_number"]).strip()
                ensure_control_number_batch_row(connection, batch_number, family_sequence + 1)
                if batch_number == 1:
                    next_family_sequence = max(next_family_sequence, family_sequence + 1)
            else:
                batch_number = 1
                family_sequence = next_family_sequence
                control_number = format_control_number(batch_number, family_sequence)
                next_family_sequence += 1
            connection.execute(
                """
                UPDATE documents
                SET control_number = ?, control_number_batch = ?, control_number_family_sequence = ?, control_number_attachment_sequence = NULL
                WHERE id = ?
                """,
                (control_number, batch_number, family_sequence, row["id"]),
            )
            updated += 1
        ensure_control_number_batch_row(connection, 1, next_family_sequence)

    parent_identities = {
        int(row["id"]): (int(row["control_number_batch"]), int(row["control_number_family_sequence"]))
        for row in connection.execute(
            """
            SELECT id, control_number_batch, control_number_family_sequence
            FROM documents
            WHERE parent_document_id IS NULL
              AND COALESCE(source_kind, ?) != ?
              AND control_number_batch IS NOT NULL
              AND control_number_family_sequence IS NOT NULL
            """
        , (FILESYSTEM_SOURCE_KIND, PRODUCTION_SOURCE_KIND)).fetchall()
    }
    attachment_counters: dict[int, int] = {}
    child_rows = connection.execute(
        """
        SELECT id, parent_document_id, control_number
        FROM documents
        WHERE parent_document_id IS NOT NULL
          AND COALESCE(source_kind, ?) != ?
          AND (
            control_number IS NULL OR TRIM(control_number) = ''
            OR control_number_batch IS NULL
            OR control_number_family_sequence IS NULL
            OR control_number_attachment_sequence IS NULL
          )
        ORDER BY parent_document_id ASC, id ASC
        """
    , (EMAIL_ATTACHMENT_SOURCE_KIND, PRODUCTION_SOURCE_KIND)).fetchall()
    for row in child_rows:
        parent_document_id = int(row["parent_document_id"])
        parent_identity = parent_identities.get(parent_document_id)
        if parent_identity is None:
            continue
        parsed_identity = parse_control_number(row["control_number"])
        batch_number, family_sequence = parent_identity
        if parsed_identity is not None and parsed_identity[2] is not None:
            _, _, attachment_sequence = parsed_identity
        else:
            if parent_document_id not in attachment_counters:
                max_sequence = connection.execute(
                    """
                    SELECT MAX(control_number_attachment_sequence) AS max_attachment_sequence
                    FROM documents
                    WHERE parent_document_id = ?
                    """,
                    (parent_document_id,),
                ).fetchone()
                attachment_counters[parent_document_id] = int(max_sequence["max_attachment_sequence"] or 0)
            attachment_counters[parent_document_id] += 1
            attachment_sequence = attachment_counters[parent_document_id]
        connection.execute(
            """
            UPDATE documents
            SET control_number = ?, control_number_batch = ?, control_number_family_sequence = ?, control_number_attachment_sequence = ?
            WHERE id = ?
            """,
            (
                format_control_number(batch_number, family_sequence, attachment_sequence),
                batch_number,
                family_sequence,
                attachment_sequence,
                row["id"],
            ),
        )
        updated += 1

    return updated


def allocate_ingestion_batch_number(connection: sqlite3.Connection) -> int:
    row = connection.execute("SELECT MAX(batch_number) AS max_batch_number FROM control_number_batches").fetchone()
    batch_number = int(row["max_batch_number"] or 0) + 1
    ensure_control_number_batch_row(connection, batch_number, 1)
    return batch_number


def reserve_control_number_family_sequence(connection: sqlite3.Connection, batch_number: int) -> int:
    next_family_sequence = ensure_control_number_batch_row(connection, batch_number, 1)
    connection.execute(
        """
        UPDATE control_number_batches
        SET next_family_sequence = ?, updated_at = ?
        WHERE batch_number = ?
        """,
        (next_family_sequence + 1, utc_now(), batch_number),
    )
    return next_family_sequence


def next_attachment_sequence(connection: sqlite3.Connection, parent_document_id: int) -> int:
    row = connection.execute(
        """
        SELECT MAX(control_number_attachment_sequence) AS max_attachment_sequence
        FROM documents
        WHERE parent_document_id = ?
        """,
        (parent_document_id,),
    ).fetchone()
    return int(row["max_attachment_sequence"] or 0) + 1


def backfill_internal_rel_path_prefix(connection: sqlite3.Connection) -> int:
    """Rewrite synthetic rel_paths that still use the legacy ``.retriever/`` prefix.

    Container-derived messages, production logical documents, and attachment
    blobs used to be stored with a leading ``.retriever/`` so that
    ``<root>/<rel_path>`` would happen to resolve to the real file under the
    state directory. That made it look like the workspace's opaque state
    directory contained indexed documents, which confused scans and searches.
    The canonical synthetic prefix is now ``_retriever/`` and path resolution
    translates that back to the state directory explicitly.
    """
    if not table_exists(connection, "documents"):
        return 0
    cursor = connection.execute(
        """
        UPDATE documents
        SET rel_path = '_retriever/' || substr(rel_path, length('.retriever/') + 1)
        WHERE rel_path LIKE '.retriever/%'
        """
    )
    return int(cursor.rowcount or 0)


def ensure_documents_fts(connection: sqlite3.Connection) -> bool:
    expected_columns = {"document_id", "file_name", "title", "subject", "author", "custodian", "participants", "recipients"}
    existing_columns = table_columns(connection, "documents_fts")
    if existing_columns == expected_columns:
        return False

    connection.execute("DROP TABLE IF EXISTS documents_fts")
    connection.execute(
        """
        CREATE VIRTUAL TABLE documents_fts USING fts5(
          document_id UNINDEXED,
          file_name,
          title,
          subject,
          author,
          custodian,
          participants,
          recipients
        )
        """
    )
    rows = connection.execute(
        """
        SELECT id, file_name, title, subject, author, custodians_json, participants, recipients
        FROM documents
        """
    ).fetchall()
    if rows:
        connection.executemany(
            """
            INSERT INTO documents_fts (document_id, file_name, title, subject, author, custodian, participants, recipients)
            VALUES (?, ?, ?, ?, ?, ?, ?, ?)
            """,
            [
                (
                    row["id"],
                    row["file_name"],
                    row["title"],
                    row["subject"],
                    row["author"],
                    document_custodian_display_text_from_row(row),
                    row["participants"],
                    row["recipients"],
                )
                for row in rows
            ],
        )
    return True


def apply_schema(connection: sqlite3.Connection, root: Path | None = None) -> dict[str, object]:
    prior_schema_version: int | None = None
    if table_exists(connection, "workspace_meta"):
        workspace_meta_row = connection.execute(
            """
            SELECT schema_version
            FROM workspace_meta
            WHERE id = 1
            """
        ).fetchone()
        if workspace_meta_row is not None and workspace_meta_row["schema_version"] is not None:
            prior_schema_version = int(workspace_meta_row["schema_version"])

    rename_table_if_needed(connection, "display_id_batches", "control_number_batches")
    for statement in SCHEMA_STATEMENTS:
        connection.execute(statement)
    migrated_export_work_items_nullable_document_id = ensure_export_work_items_nullable_document_id(connection)

    if table_exists(connection, "documents"):
        rename_column_if_needed(connection, "documents", "display_id", "control_number")
        rename_column_if_needed(connection, "documents", "display_batch", "control_number_batch")
        rename_column_if_needed(connection, "documents", "display_family_sequence", "control_number_family_sequence")
        rename_column_if_needed(connection, "documents", "display_attachment_sequence", "control_number_attachment_sequence")

    ensure_column(connection, "documents", f"{MANUAL_FIELD_LOCKS_COLUMN} TEXT NOT NULL DEFAULT '[]'")
    ensure_column(connection, "documents", "canonical_kind TEXT NOT NULL DEFAULT 'unknown'")
    ensure_column(connection, "documents", "canonical_status TEXT NOT NULL DEFAULT 'active'")
    ensure_column(connection, "documents", "merged_into_document_id INTEGER REFERENCES documents(id) ON DELETE SET NULL")
    ensure_column(connection, "documents", "content_type TEXT")
    ensure_column(connection, "documents", "custodians_json TEXT NOT NULL DEFAULT '[]'")
    ensure_column(connection, "documents", "participants TEXT")
    ensure_column(connection, "documents", "control_number TEXT")
    ensure_column(connection, "documents", "conversation_id INTEGER REFERENCES conversations(id) ON DELETE SET NULL")
    ensure_column(connection, "documents", f"conversation_assignment_mode TEXT NOT NULL DEFAULT '{CONVERSATION_ASSIGNMENT_MODE_AUTO}'")
    ensure_column(connection, "documents", "dataset_id INTEGER REFERENCES datasets(id) ON DELETE SET NULL")
    ensure_column(connection, "documents", "parent_document_id INTEGER REFERENCES documents(id) ON DELETE CASCADE")
    ensure_column(connection, "documents", "child_document_kind TEXT")
    ensure_column(connection, "documents", "source_kind TEXT")
    ensure_column(connection, "documents", "source_rel_path TEXT")
    ensure_column(connection, "documents", "source_item_id TEXT")
    ensure_column(connection, "documents", "root_message_key TEXT")
    ensure_column(connection, "documents", "source_folder_path TEXT")
    ensure_column(connection, "documents", "production_id INTEGER REFERENCES productions(id) ON DELETE SET NULL")
    ensure_column(connection, "documents", "begin_bates TEXT")
    ensure_column(connection, "documents", "end_bates TEXT")
    ensure_column(connection, "documents", "begin_attachment TEXT")
    ensure_column(connection, "documents", "end_attachment TEXT")
    ensure_column(connection, "documents", "control_number_batch INTEGER")
    ensure_column(connection, "documents", "control_number_family_sequence INTEGER")
    ensure_column(connection, "documents", "control_number_attachment_sequence INTEGER")
    ensure_column(connection, "documents", "source_text_revision_id INTEGER REFERENCES text_revisions(id) ON DELETE SET NULL")
    ensure_column(connection, "documents", "active_search_text_revision_id INTEGER REFERENCES text_revisions(id) ON DELETE SET NULL")
    ensure_column(connection, "documents", "active_text_source_kind TEXT")
    ensure_column(connection, "documents", "active_text_language TEXT")
    ensure_column(connection, "documents", "active_text_quality_score REAL")
    ensure_column(connection, "document_occurrences", "mime_type TEXT")
    ensure_column(connection, "document_occurrences", "dataset_source_id INTEGER REFERENCES dataset_sources(id) ON DELETE SET NULL")
    ensure_column(connection, "document_occurrences", "fs_created_at TEXT")
    ensure_column(connection, "document_occurrences", "fs_modified_at TEXT")
    ensure_column(connection, "document_occurrences", "has_preview INTEGER NOT NULL DEFAULT 0")
    ensure_column(connection, "document_occurrences", "extracted_author TEXT")
    ensure_column(connection, "document_occurrences", "extracted_title TEXT")
    ensure_column(connection, "document_occurrences", "extracted_subject TEXT")
    ensure_column(connection, "document_occurrences", "extracted_participants TEXT")
    ensure_column(connection, "document_occurrences", "extracted_recipients TEXT")
    ensure_column(connection, "document_occurrences", "extracted_doc_authored_at TEXT")
    ensure_column(connection, "document_occurrences", "extracted_doc_modified_at TEXT")
    ensure_column(connection, "document_occurrences", "extracted_content_type TEXT")
    ensure_column(connection, "document_occurrences", "extracted_kind TEXT")
    ensure_column(connection, "document_occurrences", "entity_hints_json TEXT NOT NULL DEFAULT '{}'")
    ensure_column(connection, "document_email_threading", "message_id TEXT")
    ensure_column(connection, "document_email_threading", "in_reply_to TEXT")
    ensure_column(connection, "document_email_threading", "references_json TEXT NOT NULL DEFAULT '[]'")
    ensure_column(connection, "document_email_threading", "conversation_index TEXT")
    ensure_column(connection, "document_email_threading", "conversation_topic TEXT")
    ensure_column(connection, "document_email_threading", "normalized_subject TEXT")
    ensure_column(connection, "document_email_threading", "updated_at TEXT NOT NULL DEFAULT ''")
    ensure_column(connection, "document_chat_threading", "thread_id TEXT")
    ensure_column(connection, "document_chat_threading", "message_id TEXT")
    ensure_column(connection, "document_chat_threading", "parent_message_id TEXT")
    ensure_column(connection, "document_chat_threading", "thread_type TEXT")
    ensure_column(connection, "document_chat_threading", "participants_json TEXT NOT NULL DEFAULT '[]'")
    ensure_column(connection, "document_chat_threading", "updated_at TEXT NOT NULL DEFAULT ''")
    ensure_column(connection, "document_previews", "target_fragment TEXT")
    ensure_column(connection, "job_versions", "capability TEXT NOT NULL DEFAULT ''")
    ensure_column(connection, "runs", "activation_policy TEXT NOT NULL DEFAULT 'manual'")
    ensure_column(connection, "run_items", "result_id INTEGER REFERENCES results(id) ON DELETE SET NULL")
    ensure_column(connection, "run_items", "page_number INTEGER")
    ensure_column(connection, "run_items", "input_artifact_rel_path TEXT")
    ensure_column(connection, "run_items", "claimed_by TEXT")
    ensure_column(connection, "run_items", "claimed_at TEXT")
    ensure_column(connection, "run_items", "last_heartbeat_at TEXT")
    ensure_column(connection, "productions", "dataset_id INTEGER REFERENCES datasets(id) ON DELETE SET NULL")
    ensure_column(connection, "container_sources", "dataset_id INTEGER REFERENCES datasets(id) ON DELETE SET NULL")
    ensure_column(connection, "datasets", "dataset_name_normalized TEXT")
    ensure_column(connection, "datasets", "allow_auto_merge INTEGER NOT NULL DEFAULT 1")
    ensure_column(connection, "datasets", "email_auto_merge INTEGER NOT NULL DEFAULT 1")
    ensure_column(connection, "datasets", "handle_auto_merge INTEGER NOT NULL DEFAULT 1")
    ensure_column(connection, "datasets", "phone_auto_merge INTEGER NOT NULL DEFAULT 0")
    ensure_column(connection, "datasets", "name_auto_merge INTEGER NOT NULL DEFAULT 0")
    ensure_column(connection, "datasets", "external_id_auto_merge_names_json TEXT NOT NULL DEFAULT '[]'")
    backfilled_legacy_control_number = backfill_legacy_column(
        connection,
        "documents",
        "display_id",
        "control_number",
        treat_blank_as_missing=True,
    )
    backfilled_legacy_control_number_batch = backfill_legacy_column(
        connection,
        "documents",
        "display_batch",
        "control_number_batch",
    )
    backfilled_legacy_control_number_family_sequence = backfill_legacy_column(
        connection,
        "documents",
        "display_family_sequence",
        "control_number_family_sequence",
    )
    backfilled_legacy_control_number_attachment_sequence = backfill_legacy_column(
        connection,
        "documents",
        "display_attachment_sequence",
        "control_number_attachment_sequence",
    )
    if table_exists(connection, "job_versions") and table_exists(connection, "jobs"):
        blank_capability_rows = connection.execute(
            """
            SELECT jv.id, j.job_kind
            FROM job_versions jv
            JOIN jobs j ON j.id = jv.job_id
            WHERE jv.capability IS NULL OR TRIM(jv.capability) = ''
            """
        ).fetchall()
        for row in blank_capability_rows:
            job_kind = normalize_whitespace(str(row["job_kind"] or "")).lower()
            try:
                capability = default_job_capability_for_kind(job_kind)
            except RetrieverError:
                capability = "text_structured"
            connection.execute(
                "UPDATE job_versions SET capability = ? WHERE id = ?",
                (capability, int(row["id"])),
            )
    connection.execute("CREATE INDEX IF NOT EXISTS idx_documents_conversation_id ON documents(conversation_id)")
    connection.execute("CREATE INDEX IF NOT EXISTS idx_documents_conversation_assignment_mode ON documents(conversation_assignment_mode)")
    connection.execute("CREATE INDEX IF NOT EXISTS idx_documents_parent_document_id ON documents(parent_document_id)")
    connection.execute("CREATE INDEX IF NOT EXISTS idx_documents_child_document_kind ON documents(child_document_kind)")
    connection.execute("CREATE INDEX IF NOT EXISTS idx_documents_dataset_id ON documents(dataset_id)")
    connection.execute("CREATE INDEX IF NOT EXISTS idx_documents_canonical_status ON documents(canonical_status)")
    connection.execute("CREATE INDEX IF NOT EXISTS idx_documents_merged_into_document_id ON documents(merged_into_document_id)")
    connection.execute("CREATE INDEX IF NOT EXISTS idx_documents_canonical_status_content_hash ON documents(canonical_status, content_hash)")
    connection.execute("CREATE INDEX IF NOT EXISTS idx_documents_source_kind ON documents(source_kind)")
    connection.execute("CREATE INDEX IF NOT EXISTS idx_documents_source_rel_path ON documents(source_rel_path)")
    connection.execute("CREATE INDEX IF NOT EXISTS idx_documents_root_message_key ON documents(root_message_key)")
    connection.execute("CREATE INDEX IF NOT EXISTS idx_documents_production_id ON documents(production_id)")
    connection.execute("CREATE INDEX IF NOT EXISTS idx_documents_begin_bates ON documents(begin_bates)")
    connection.execute("CREATE INDEX IF NOT EXISTS idx_documents_end_bates ON documents(end_bates)")
    connection.execute("CREATE INDEX IF NOT EXISTS idx_documents_source_text_revision_id ON documents(source_text_revision_id)")
    connection.execute(
        "CREATE INDEX IF NOT EXISTS idx_documents_active_search_text_revision_id ON documents(active_search_text_revision_id)"
    )
    connection.execute("CREATE INDEX IF NOT EXISTS idx_document_occurrences_document_id ON document_occurrences(document_id)")
    connection.execute("CREATE INDEX IF NOT EXISTS idx_document_occurrences_document_status_ingested ON document_occurrences(document_id, lifecycle_status, ingested_at)")
    connection.execute("CREATE INDEX IF NOT EXISTS idx_document_occurrences_document_kind_status ON document_occurrences(document_id, extracted_kind, text_status)")
    connection.execute("CREATE INDEX IF NOT EXISTS idx_document_occurrences_file_hash ON document_occurrences(file_hash)")
    connection.execute("CREATE INDEX IF NOT EXISTS idx_document_occurrences_source_rel_path ON document_occurrences(source_rel_path)")
    connection.execute("CREATE INDEX IF NOT EXISTS idx_document_occurrences_dataset_source_id ON document_occurrences(dataset_source_id)")
    connection.execute("CREATE UNIQUE INDEX IF NOT EXISTS idx_document_occurrences_active_rel_path ON document_occurrences(rel_path) WHERE lifecycle_status = 'active'")
    connection.execute(
        """
        CREATE UNIQUE INDEX IF NOT EXISTS idx_document_occurrences_active_source_identity
        ON document_occurrences(source_kind, COALESCE(custodian, ''), COALESCE(source_rel_path, ''), COALESCE(source_item_id, ''))
        WHERE lifecycle_status = 'active'
        """
    )
    connection.execute("CREATE UNIQUE INDEX IF NOT EXISTS idx_document_dedupe_keys_basis_value ON document_dedupe_keys(basis, key_value)")
    connection.execute("CREATE INDEX IF NOT EXISTS idx_document_dedupe_keys_document_id ON document_dedupe_keys(document_id)")
    connection.execute("CREATE INDEX IF NOT EXISTS idx_canonical_metadata_conflicts_document_field ON canonical_metadata_conflicts(document_id, field_name)")
    connection.execute("CREATE INDEX IF NOT EXISTS idx_document_control_number_aliases_alias_value ON document_control_number_aliases(alias_value)")
    connection.execute("CREATE INDEX IF NOT EXISTS idx_document_merge_events_survivor_loser_created ON document_merge_events(survivor_document_id, loser_document_id, created_at)")
    connection.execute("CREATE INDEX IF NOT EXISTS idx_document_field_conflicts_merge_event_id ON document_field_conflicts(merge_event_id)")
    connection.execute("CREATE INDEX IF NOT EXISTS idx_dataset_sources_dataset_id ON dataset_sources(dataset_id)")
    connection.execute("CREATE UNIQUE INDEX IF NOT EXISTS idx_dataset_sources_locator_unique ON dataset_sources(source_kind, source_locator)")
    connection.execute("DROP INDEX IF EXISTS idx_conversations_source_kind_key")
    connection.execute(
        """
        CREATE UNIQUE INDEX IF NOT EXISTS idx_conversations_source_locator_key_unique
        ON conversations(source_kind, source_locator, conversation_key)
        """
    )
    connection.execute("CREATE INDEX IF NOT EXISTS idx_conversations_source_locator ON conversations(source_locator)")
    connection.execute("CREATE INDEX IF NOT EXISTS idx_dataset_documents_document_id ON dataset_documents(document_id)")
    connection.execute("CREATE INDEX IF NOT EXISTS idx_dataset_documents_dataset_id ON dataset_documents(dataset_id)")
    connection.execute("CREATE INDEX IF NOT EXISTS idx_dataset_documents_source_id ON dataset_documents(dataset_source_id)")
    connection.execute(
        """
        CREATE UNIQUE INDEX IF NOT EXISTS idx_dataset_documents_membership_unique
        ON dataset_documents(dataset_id, document_id, COALESCE(dataset_source_id, 0))
        """
    )
    connection.execute("CREATE INDEX IF NOT EXISTS idx_productions_dataset_id ON productions(dataset_id)")
    connection.execute("CREATE INDEX IF NOT EXISTS idx_container_sources_dataset_id ON container_sources(dataset_id)")
    connection.execute("CREATE INDEX IF NOT EXISTS idx_container_sources_source_kind ON container_sources(source_kind)")
    connection.execute("CREATE INDEX IF NOT EXISTS idx_entities_status_type ON entities(canonical_status, entity_type)")
    connection.execute("CREATE INDEX IF NOT EXISTS idx_entities_merged_into ON entities(merged_into_entity_id)")
    connection.execute("CREATE INDEX IF NOT EXISTS idx_entities_label ON entities(display_name, primary_email)")
    connection.execute("CREATE INDEX IF NOT EXISTS idx_entity_identifiers_entity_type ON entity_identifiers(entity_id, identifier_type)")
    connection.execute("CREATE INDEX IF NOT EXISTS idx_entity_identifiers_lookup ON entity_identifiers(identifier_type, normalized_value)")
    connection.execute(
        """
        CREATE INDEX IF NOT EXISTS idx_entity_identifiers_scoped_handle
        ON entity_identifiers(provider, provider_scope, normalized_value)
        WHERE identifier_type = 'handle'
        """
    )
    connection.execute(
        """
        CREATE INDEX IF NOT EXISTS idx_entity_identifiers_external_id
        ON entity_identifiers(identifier_name, identifier_scope, normalized_value)
        WHERE identifier_type = 'external_id'
        """
    )
    connection.execute("CREATE INDEX IF NOT EXISTS idx_entity_resolution_keys_entity ON entity_resolution_keys(entity_id)")
    connection.execute("CREATE INDEX IF NOT EXISTS idx_entity_resolution_keys_lookup ON entity_resolution_keys(key_type, normalized_value)")
    connection.execute(
        """
        CREATE UNIQUE INDEX IF NOT EXISTS idx_entity_resolution_keys_email_unique
        ON entity_resolution_keys(key_type, normalized_value)
        WHERE key_type = 'email'
        """
    )
    connection.execute(
        """
        CREATE UNIQUE INDEX IF NOT EXISTS idx_entity_resolution_keys_handle_unique
        ON entity_resolution_keys(key_type, provider, provider_scope, normalized_value)
        WHERE key_type = 'handle'
        """
    )
    connection.execute("DROP INDEX IF EXISTS idx_entity_resolution_keys_external_id_unique")
    connection.execute(
        """
        CREATE UNIQUE INDEX IF NOT EXISTS idx_entity_resolution_keys_external_id_unique
        ON entity_resolution_keys(key_type, identifier_name, COALESCE(identifier_scope, ''), normalized_value)
        WHERE key_type = 'external_id'
        """
    )
    connection.execute("CREATE INDEX IF NOT EXISTS idx_document_entities_document_role ON document_entities(document_id, role, ordinal)")
    connection.execute("CREATE INDEX IF NOT EXISTS idx_document_entities_entity_role ON document_entities(entity_id, role)")
    connection.execute(
        """
        CREATE UNIQUE INDEX IF NOT EXISTS idx_document_entities_unique
        ON document_entities(document_id, role, entity_id)
        """
    )
    connection.execute("CREATE UNIQUE INDEX IF NOT EXISTS idx_entity_merge_blocks_pair ON entity_merge_blocks(left_entity_id, right_entity_id)")
    connection.execute("CREATE INDEX IF NOT EXISTS idx_entity_overrides_source_entity ON entity_overrides(source_entity_id)")
    connection.execute("CREATE INDEX IF NOT EXISTS idx_entity_overrides_replacement_entity ON entity_overrides(replacement_entity_id)")
    connection.execute(
        """
        CREATE UNIQUE INDEX IF NOT EXISTS idx_entity_overrides_document_effect
        ON entity_overrides(
          scope_type,
          COALESCE(scope_id, 0),
          COALESCE(role, ''),
          COALESCE(source_entity_id, 0),
          COALESCE(normalized_candidate_key, ''),
          override_effect
        )
        WHERE scope_type = 'document'
        """
    )
    connection.execute(
        """
        CREATE UNIQUE INDEX IF NOT EXISTS idx_entity_overrides_global_ignore
        ON entity_overrides(
          scope_type,
          COALESCE(role, ''),
          COALESCE(source_entity_id, 0),
          COALESCE(normalized_candidate_key, ''),
          COALESCE(source_hint, ''),
          override_effect
        )
        WHERE scope_type = 'global' AND override_effect = 'ignore'
        """
    )
    connection.execute("CREATE INDEX IF NOT EXISTS idx_document_email_threading_message_id ON document_email_threading(message_id)")
    connection.execute(
        "CREATE INDEX IF NOT EXISTS idx_document_email_threading_conversation_index ON document_email_threading(conversation_index)"
    )
    connection.execute(
        "CREATE INDEX IF NOT EXISTS idx_document_email_threading_conversation_topic ON document_email_threading(conversation_topic)"
    )
    connection.execute(
        "CREATE INDEX IF NOT EXISTS idx_document_email_threading_normalized_subject ON document_email_threading(normalized_subject)"
    )
    connection.execute("CREATE INDEX IF NOT EXISTS idx_document_chat_threading_thread_id ON document_chat_threading(thread_id)")
    connection.execute("CREATE INDEX IF NOT EXISTS idx_document_chat_threading_message_id ON document_chat_threading(message_id)")
    connection.execute(
        "CREATE INDEX IF NOT EXISTS idx_document_chat_threading_parent_message_id ON document_chat_threading(parent_message_id)"
    )
    connection.execute("DROP INDEX IF EXISTS idx_documents_display_sort")
    connection.execute("CREATE UNIQUE INDEX IF NOT EXISTS idx_documents_control_number_unique ON documents(control_number)")
    connection.execute(
        """
        CREATE UNIQUE INDEX IF NOT EXISTS idx_documents_source_identity_unique
        ON documents(source_rel_path, source_item_id)
        WHERE source_kind IS NOT NULL AND source_item_id IS NOT NULL
        """
    )
    connection.execute(
        """
        CREATE INDEX IF NOT EXISTS idx_documents_control_number_sort
        ON documents(control_number_batch, control_number_family_sequence, control_number_attachment_sequence)
        """
    )
    connection.execute("CREATE INDEX IF NOT EXISTS idx_job_outputs_job_id ON job_outputs(job_id, ordinal)")
    connection.execute("CREATE INDEX IF NOT EXISTS idx_job_versions_job_id ON job_versions(job_id, version DESC)")
    connection.execute("CREATE INDEX IF NOT EXISTS idx_job_versions_capability ON job_versions(capability)")
    connection.execute("CREATE INDEX IF NOT EXISTS idx_runs_job_version_id ON runs(job_version_id)")
    connection.execute("CREATE INDEX IF NOT EXISTS idx_runs_from_run_id ON runs(from_run_id)")
    connection.execute(
        """
        CREATE INDEX IF NOT EXISTS idx_run_snapshot_documents_run_id_ordinal
        ON run_snapshot_documents(run_id, ordinal, id)
        """
    )
    connection.execute(
        "CREATE INDEX IF NOT EXISTS idx_run_snapshot_documents_document_id ON run_snapshot_documents(document_id)"
    )
    connection.execute("CREATE INDEX IF NOT EXISTS idx_run_items_run_id_status ON run_items(run_id, status)")
    connection.execute("CREATE INDEX IF NOT EXISTS idx_run_items_document_id ON run_items(document_id)")
    connection.execute("CREATE INDEX IF NOT EXISTS idx_run_items_result_id ON run_items(result_id)")
    connection.execute("CREATE INDEX IF NOT EXISTS idx_run_items_page_number ON run_items(run_id, document_id, page_number)")
    connection.execute("CREATE INDEX IF NOT EXISTS idx_run_items_run_claim ON run_items(run_id, status, claimed_by)")
    connection.execute("CREATE INDEX IF NOT EXISTS idx_run_items_heartbeat ON run_items(last_heartbeat_at)")
    connection.execute("CREATE INDEX IF NOT EXISTS idx_run_workers_run_id ON run_workers(run_id, status)")
    connection.execute("CREATE INDEX IF NOT EXISTS idx_run_workers_claimed_by ON run_workers(run_id, claimed_by)")
    connection.execute("CREATE INDEX IF NOT EXISTS idx_run_workers_task_id ON run_workers(worker_task_id)")
    connection.execute("DROP INDEX IF EXISTS idx_run_items_snapshot_kind_unique")
    connection.execute(
        """
        CREATE UNIQUE INDEX IF NOT EXISTS idx_run_items_snapshot_kind_unique
        ON run_items(
          run_id,
          COALESCE(run_snapshot_document_id, 0),
          item_kind,
          COALESCE(page_number, 0),
          COALESCE(segment_id, 0)
        )
        """
    )
    connection.execute("CREATE INDEX IF NOT EXISTS idx_attempts_run_item_id ON attempts(run_item_id, attempt_number)")
    connection.execute("CREATE INDEX IF NOT EXISTS idx_ocr_page_outputs_run_id_doc_page ON ocr_page_outputs(run_id, document_id, page_number)")
    connection.execute(
        "CREATE INDEX IF NOT EXISTS idx_image_description_page_outputs_run_id_doc_page "
        "ON image_description_page_outputs(run_id, document_id, page_number)"
    )
    connection.execute("CREATE INDEX IF NOT EXISTS idx_results_job_version_id ON results(job_version_id, created_at)")
    connection.execute("CREATE INDEX IF NOT EXISTS idx_results_document_id ON results(document_id, created_at)")
    connection.execute("CREATE INDEX IF NOT EXISTS idx_results_input_revision_id ON results(input_revision_id)")
    connection.execute(
        """
        CREATE UNIQUE INDEX IF NOT EXISTS idx_results_active_identity_unique
        ON results(document_id, job_version_id, input_identity)
        WHERE retracted_at IS NULL
        """
    )
    connection.execute("CREATE INDEX IF NOT EXISTS idx_result_outputs_result_id ON result_outputs(result_id)")
    connection.execute(
        """
        CREATE INDEX IF NOT EXISTS idx_text_revisions_document_created_at
        ON text_revisions(document_id, created_at, id)
        """
    )
    connection.execute(
        """
        CREATE INDEX IF NOT EXISTS idx_text_revision_segments_revision_profile
        ON text_revision_segments(revision_id, segment_profile, level, ordinal)
        """
    )
    connection.execute(
        """
        CREATE UNIQUE INDEX IF NOT EXISTS idx_embedding_vectors_active_segment_unique
        ON embedding_vectors(job_version_id, segment_id)
        WHERE retracted_at IS NULL
        """
    )
    connection.execute(
        """
        CREATE INDEX IF NOT EXISTS idx_publications_document_field
        ON publications(document_id, custom_field_name, published_at)
        """
    )
    connection.execute(
        """
        CREATE INDEX IF NOT EXISTS idx_text_revision_activation_events_document_created_at
        ON text_revision_activation_events(document_id, created_at, id)
        """
    )
    merged_legacy_locks = merge_legacy_field_locks(connection)
    backfilled_content_type = backfill_content_type(connection)
    backfilled_child_document_kinds = backfill_child_document_kinds(connection)
    backfilled_conversation_assignment_modes = backfill_conversation_assignment_modes(connection)
    backfilled_source_kinds = backfill_source_kinds(connection)
    rewrote_internal_rel_path_prefix = backfill_internal_rel_path_prefix(connection)
    dataset_membership_migration_needed = prior_schema_version is None or prior_schema_version < 12
    backfilled_dataset_ids = backfill_dataset_ids(connection, root) if dataset_membership_migration_needed else 0
    backfilled_dataset_memberships = backfill_dataset_memberships(connection) if dataset_membership_migration_needed else 0
    backfilled_occurrence_dataset_source_ids = backfill_occurrence_dataset_source_ids(connection)
    backfilled_dataset_name_normalized = backfill_dataset_name_normalized(connection)
    merged_duplicate_dataset_identities = merge_dataset_identity_duplicates(connection)
    suffixed_dataset_name_collisions = suffix_dataset_name_collisions(connection)
    connection.execute("CREATE UNIQUE INDEX IF NOT EXISTS idx_datasets_source_locator_unique ON datasets(source_kind, dataset_locator)")
    connection.execute("CREATE UNIQUE INDEX IF NOT EXISTS idx_datasets_name_normalized_unique ON datasets(dataset_name_normalized)")
    backfilled_custodian = backfill_custodian(connection)
    backfilled_control_numbers = backfill_control_numbers(connection)
    rebuilt_control_number_batches = backfill_control_number_batches(connection)
    backfilled_document_occurrences = backfill_document_occurrences(connection)
    backfilled_document_dedupe_keys = backfill_document_dedupe_keys(connection)
    refresh_document_occurrence_caches = should_refresh_document_occurrence_caches(
        prior_schema_version=prior_schema_version,
        backfilled_occurrence_dataset_source_ids=backfilled_occurrence_dataset_source_ids,
        backfilled_document_occurrences=backfilled_document_occurrences,
    )
    refreshed_document_occurrence_caches = (
        refresh_documents_from_occurrences(connection)
        if refresh_document_occurrence_caches
        else 0
    )
    removed_custodian_locks = drop_document_field_locks(connection, "custodian")
    removed_documents_custodian_column = remove_documents_custodian_column(connection)
    rebuilt_documents_fts = ensure_documents_fts(connection)
    connection.commit()
    return {
        "schema_version": SCHEMA_VERSION,
        "backfilled_content_type": backfilled_content_type,
        "backfilled_child_document_kinds": backfilled_child_document_kinds,
        "backfilled_conversation_assignment_modes": backfilled_conversation_assignment_modes,
        "backfilled_custodian": backfilled_custodian,
        "backfilled_dataset_ids": backfilled_dataset_ids,
        "backfilled_dataset_memberships": backfilled_dataset_memberships,
        "backfilled_occurrence_dataset_source_ids": backfilled_occurrence_dataset_source_ids,
        "backfilled_dataset_name_normalized": backfilled_dataset_name_normalized,
        "merged_duplicate_dataset_identities": merged_duplicate_dataset_identities,
        "suffixed_dataset_name_collisions": suffixed_dataset_name_collisions,
        "backfilled_source_kinds": backfilled_source_kinds,
        "rewrote_internal_rel_path_prefix": rewrote_internal_rel_path_prefix,
        "backfilled_control_numbers": backfilled_control_numbers,
        "rebuilt_control_number_batches": rebuilt_control_number_batches,
        "backfilled_document_occurrences": backfilled_document_occurrences,
        "migrated_export_work_items_nullable_document_id": migrated_export_work_items_nullable_document_id,
        "backfilled_document_dedupe_keys": backfilled_document_dedupe_keys,
        "refresh_document_occurrence_caches": refresh_document_occurrence_caches,
        "refreshed_document_occurrence_caches": refreshed_document_occurrence_caches,
        "removed_custodian_locks": removed_custodian_locks,
        "removed_documents_custodian_column": removed_documents_custodian_column,
        "backfilled_legacy_control_number": backfilled_legacy_control_number,
        "backfilled_legacy_control_number_batch": backfilled_legacy_control_number_batch,
        "backfilled_legacy_control_number_family_sequence": backfilled_legacy_control_number_family_sequence,
        "backfilled_legacy_control_number_attachment_sequence": backfilled_legacy_control_number_attachment_sequence,
        "rebuilt_documents_fts": rebuilt_documents_fts,
        "merged_legacy_locks": merged_legacy_locks,
    }


def read_runtime(path: Path) -> dict[str, object] | None:
    if not path.exists():
        return None
    return json.loads(path.read_text(encoding="utf-8"))


def read_plugin_runtime_requirements_version(paths: dict[str, Path] | None) -> str | None:
    if paths is None:
        return None
    marker_path = paths["requirements_marker_path"]
    if not marker_path.exists():
        return None
    return normalize_whitespace(marker_path.read_text(encoding="utf-8")) or None


def write_plugin_runtime_requirements_version(paths: dict[str, Path], version: str) -> None:
    paths["requirements_marker_path"].write_text(f"{version}\n", encoding="utf-8")


def activate_plugin_site_packages(paths: dict[str, Path] | None) -> bool:
    if paths is None:
        return False
    site_packages_path = first_existing_plugin_runtime_site_packages(paths)
    if site_packages_path is None:
        return False
    try:
        resolved = str(site_packages_path.resolve())
    except OSError:
        resolved = str(site_packages_path)
    if resolved in ACTIVATED_PLUGIN_SITE_PACKAGES or resolved in sys.path:
        ACTIVATED_PLUGIN_SITE_PACKAGES.add(resolved)
        return True
    site.addsitedir(resolved)
    ACTIVATED_PLUGIN_SITE_PACKAGES.add(resolved)
    return True


def describe_plugin_runtime(paths: dict[str, Path] | None) -> dict[str, object]:
    if paths is None:
        return {
            "status": "unavailable",
            "detail": "Shared plugin runtime path could not be determined from the current plugin installation.",
            "mode": "shared",
            "plugin_root": None,
            "runtime_root": None,
            "venv_present": False,
            "python_executable": None,
            "site_packages_path": None,
            "requirements_version": None,
        }
    venv_dir_present = paths["venv_dir"].exists()
    python_present = paths["venv_python_path"].exists()
    site_packages_path = first_existing_plugin_runtime_site_packages(paths)
    installed_requirements_version = read_plugin_runtime_requirements_version(paths)
    if python_present and installed_requirements_version == REQUIREMENTS_VERSION:
        status = "pass"
        detail = "Shared plugin runtime and pinned requirements are ready."
    elif python_present:
        status = "partial"
        detail = "Shared plugin runtime exists, but the pinned requirements are not fully installed yet."
    elif venv_dir_present:
        status = "partial"
        detail = "Shared plugin runtime directory exists, but the Python executable is missing."
    else:
        status = "missing"
        detail = "Shared plugin runtime has not been initialized yet."
    return {
        "status": status,
        "detail": detail,
        "mode": "shared",
        "plugin_root": str(paths["plugin_root"]),
        "runtime_root": str(paths["runtime_root"]),
        "venv_present": venv_dir_present,
        "python_executable": str(paths["venv_python_path"]) if python_present else None,
        "site_packages_path": str(site_packages_path) if site_packages_path is not None else None,
        "requirements_version": installed_requirements_version,
    }


def ensure_plugin_runtime_layout(paths: dict[str, Path]) -> None:
    paths["runtime_root"].mkdir(parents=True, exist_ok=True)
    paths["locks_dir"].mkdir(parents=True, exist_ok=True)


def ensure_plugin_venv(paths: dict[str, Path]) -> dict[str, object]:
    ensure_plugin_runtime_layout(paths)
    venv_dir = paths["venv_dir"]
    created = not paths["venv_python_path"].exists()
    if created:
        builder = venv.EnvBuilder(with_pip=True)
        builder.create(venv_dir)
    if not paths["venv_python_path"].exists():
        raise RetrieverError(
            f"Plugin runtime initialization did not create a Python executable at {paths['venv_python_path']}"
        )
    return {
        "created": created,
        "python_executable": str(paths["venv_python_path"]),
    }


def install_plugin_runtime_requirements(
    paths: dict[str, Path],
    requirements: list[str] | tuple[str, ...],
    *,
    reason: str,
) -> dict[str, object]:
    python_path = paths["venv_python_path"]
    if not python_path.exists():
        raise RetrieverError(f"Plugin runtime Python not found at {python_path}")
    command = [
        str(python_path),
        "-m",
        "pip",
        "install",
        "--disable-pip-version-check",
        *list(requirements),
    ]
    completed = subprocess.run(command, capture_output=True, text=True)
    if completed.returncode != 0:
        detail = normalize_whitespace(completed.stderr or completed.stdout) or f"pip exited with status {completed.returncode}"
        raise RetrieverError(f"Plugin runtime install failed for {reason}: {detail}")
    write_plugin_runtime_requirements_version(paths, REQUIREMENTS_VERSION)
    return {
        "installed": True,
        "reason": reason,
        "command": command,
        "requirements_version": REQUIREMENTS_VERSION,
    }


def ensure_plugin_runtime(
    paths: dict[str, Path] | None,
    *,
    install_requirements: bool,
    force_requirements_install: bool = False,
    reason: str,
) -> dict[str, object]:
    if paths is None:
        raise RetrieverError("Could not determine the shared plugin runtime location for this Retriever installation.")
    ensure_plugin_runtime_layout(paths)
    lock_handle = acquire_plugin_runtime_install_lock(paths)
    try:
        venv_result = ensure_plugin_venv(paths)
        activate_plugin_site_packages(paths)
        installed_requirements_version = read_plugin_runtime_requirements_version(paths)
        requirements_installed = False
        if install_requirements and (
            force_requirements_install
            or installed_requirements_version != REQUIREMENTS_VERSION
        ):
            install_plugin_runtime_requirements(paths, PINNED_RUNTIME_REQUIREMENTS, reason=reason)
            installed_requirements_version = REQUIREMENTS_VERSION
            requirements_installed = True
            activate_plugin_site_packages(paths)
        runtime_status = describe_plugin_runtime(paths)
        return {
            "status": runtime_status["status"],
            "detail": runtime_status["detail"],
            "venv_created": bool(venv_result["created"]),
            "requirements_installed": requirements_installed,
            "requirements_version": installed_requirements_version,
            "python_executable": runtime_status["python_executable"],
        }
    finally:
        release_plugin_runtime_install_lock(lock_handle)


def read_workspace_meta(connection: sqlite3.Connection) -> dict[str, object] | None:
    if not table_exists(connection, "workspace_meta"):
        return None
    row = connection.execute(
        """
        SELECT
          id, schema_version, tool_version, requirements_version,
          template_source, template_sha256, created_at, updated_at
        FROM workspace_meta
        WHERE id = 1
        """
    ).fetchone()
    if row is None:
        return None
    return {key: row[key] for key in row.keys()}


def write_runtime(paths: dict[str, Path], tool_sha256: str | None) -> dict[str, object]:
    now = utc_now()
    plugin_runtime_paths_for_workspace = plugin_runtime_paths(root=paths["root"])
    runtime = {
        "tool_version": TOOL_VERSION,
        "schema_version": SCHEMA_VERSION,
        "requirements_version": REQUIREMENTS_VERSION,
        "template_source": TEMPLATE_SOURCE,
        "template_sha256": tool_sha256,
        "python_version": platform.python_version(),
        "generated_at": now,
        "last_verified_at": now,
        "plugin_runtime": describe_plugin_runtime(plugin_runtime_paths_for_workspace),
    }
    paths["runtime_path"].write_text(json.dumps(runtime, indent=2, sort_keys=True), encoding="utf-8")
    return runtime


def probe_fts5() -> dict[str, str]:
    connection = sqlite3.connect(":memory:")
    try:
        connection.execute("CREATE VIRTUAL TABLE t USING fts5(x)")
        return {"status": "pass", "detail": "FTS5 virtual table created successfully"}
    except Exception as exc:  # pragma: no cover - runtime probe
        return {"status": "fail", "detail": f"{type(exc).__name__}: {exc}"}
    finally:
        connection.close()


def detect_platform() -> dict[str, str]:
    return {
        "platform": platform.platform(),
        "system": platform.system(),
        "release": platform.release(),
        "machine": platform.machine(),
    }


def probe_processing_providers(connection: sqlite3.Connection | None) -> dict[str, object]:
    configured_providers: list[str] = []
    configured_capabilities: list[str] = []
    if connection is not None and table_exists(connection, "job_versions"):
        configured_providers = sorted(
            {
                normalize_whitespace(str(row["provider"] or "")).lower()
                for row in connection.execute(
                    """
                    SELECT DISTINCT provider
                    FROM job_versions
                    WHERE provider IS NOT NULL AND TRIM(provider) != ''
                    """
                ).fetchall()
                if normalize_whitespace(str(row["provider"] or ""))
            }
        )
        configured_capabilities = sorted(
            {
                normalize_whitespace(str(row["capability"] or "")).lower()
                for row in connection.execute(
                    """
                    SELECT DISTINCT capability
                    FROM job_versions
                    WHERE capability IS NOT NULL AND TRIM(capability) != ''
                    """
                ).fetchall()
                if normalize_whitespace(str(row["capability"] or ""))
            }
        )
    return {
        "configured_providers": configured_providers,
        "configured_capabilities": configured_capabilities,
        "cowork_runtime": {
            "status": "pass",
            "detail": "Cowork agent execution is the primary runtime path for processing jobs.",
        },
        "external_providers": {
            "status": "warn" if configured_providers else "pass",
            "detail": (
                "External provider identifiers are stored for future integrations, but are not required for Cowork execution."
                if configured_providers
                else "No external provider identifiers are configured."
            ),
        },
    }


def determine_workspace_state(paths: dict[str, Path]) -> str:
    state_dir = paths["state_dir"].exists()
    db_path = paths["db_path"].exists()
    db_usable = db_path and (file_size_bytes(paths["db_path"]) or 0) > 0
    runtime_path = paths["runtime_path"].exists()
    if all((state_dir, db_usable, runtime_path)):
        return "initialized"
    if any((state_dir, db_path, runtime_path)):
        return "partial"
    return "missing"


def workspace_status(root: Path, quick: bool) -> dict[str, object]:
    set_active_workspace_root(root)
    paths = workspace_paths(root)
    runtime_paths = plugin_runtime_paths(root=root)
    plugin_runtime = describe_plugin_runtime(runtime_paths)
    fts5 = probe_fts5()
    pip_python = (
        runtime_paths["venv_python_path"]
        if runtime_paths is not None and runtime_paths["venv_python_path"].exists()
        else Path(sys.executable)
    )
    pip_ok, pip_version = run_command([str(pip_python), "-m", "pip", "--version"])
    pst_backend = dependency_status(
        "pypff",
        package_name="libpff-python",
        import_name="pypff",
        detail_label="PST backend",
        probe_if_unloaded=True,
        allow_auto_install=False,
    )
    runtime = read_runtime(paths["runtime_path"])
    canonical_tool_path = locate_canonical_plugin_tool_or_self()
    current_sha = sha256_file(canonical_tool_path) if canonical_tool_path is not None else None
    stored_sha = None if runtime is None else runtime.get("template_sha256")
    workspace_state = determine_workspace_state(paths)
    registry_status = None
    workspace_inventory = None
    processing_providers = probe_processing_providers(None)
    journal_mode = None
    db_error = None
    workspace_meta = None
    actual_schema_version = None
    schema_needs_migration = None
    workspace_inventory_error = None
    registry_error = None

    if paths["db_path"].exists() and (file_size_bytes(paths["db_path"]) or 0) > 0:
        try:
            connection = connect_db(paths["db_path"])
            try:
                journal_mode = current_journal_mode(connection)
                workspace_meta = read_workspace_meta(connection)
                if workspace_meta is not None and workspace_meta.get("schema_version") is not None:
                    actual_schema_version = int(workspace_meta["schema_version"])
                    schema_needs_migration = actual_schema_version < SCHEMA_VERSION
                elif workspace_state != "missing":
                    schema_needs_migration = True

                if schema_needs_migration is not True:
                    if table_exists(connection, "documents"):
                        try:
                            workspace_inventory = document_inventory_counts(connection)
                        except Exception as exc:
                            workspace_inventory_error = f"{type(exc).__name__}: {exc}"
                    if table_exists(connection, "documents") and table_exists(connection, "custom_fields_registry"):
                        try:
                            registry_status = reconcile_custom_fields_registry(connection, repair=False)
                        except Exception as exc:
                            registry_error = f"{type(exc).__name__}: {exc}"

                processing_providers = probe_processing_providers(connection)
            finally:
                connection.close()
        except Exception as exc:
            db_error = f"{type(exc).__name__}: {exc}"

    overall = "pass"
    if fts5["status"] != "pass":
        overall = "fail"
    elif db_error is not None:
        overall = "fail"
    elif (
        workspace_state == "partial"
        or schema_needs_migration is True
        or workspace_inventory_error is not None
        or registry_error is not None
    ):
        overall = "partial"

    result: dict[str, object] = {
        "overall": overall,
        "tool_version": TOOL_VERSION,
        "schema_version": SCHEMA_VERSION,
        "workspace_schema_version": actual_schema_version,
        "schema_needs_migration": schema_needs_migration,
        "python_version": platform.python_version(),
        "pip_version": pip_version if pip_ok else None,
        "pip_status": "pass" if pip_ok else "fail",
        "sqlite_version": sqlite3.sqlite_version,
        "fts5": fts5,
        "pst_backend": pst_backend,
        "processing_providers": processing_providers,
        "platform": detect_platform(),
        "plugin_runtime": plugin_runtime,
        "workspace": {
            "root": str(root.resolve()),
            "state": workspace_state,
            "db_present": paths["db_path"].exists(),
            "db_size_bytes": file_size_bytes(paths["db_path"]),
            "runtime_present": paths["runtime_path"].exists(),
            "canonical_tool_present": canonical_tool_path is not None,
        },
        "tool_integrity": {
            "current_sha256": current_sha,
            "runtime_sha256": stored_sha,
            "matches_runtime": current_sha == stored_sha if current_sha and stored_sha else None,
        },
    }
    if journal_mode is not None:
        result["sqlite_journal_mode"] = journal_mode
    if db_error is not None:
        result["db_error"] = db_error
    if workspace_inventory is not None:
        result["workspace_inventory"] = workspace_inventory
    if workspace_inventory_error is not None:
        result["workspace_inventory_error"] = workspace_inventory_error
    if registry_error is not None:
        result["custom_field_registry_error"] = registry_error

    if not quick:
        result["paths"] = {key: str(value) for key, value in paths.items()}
        if runtime is not None:
            result["runtime"] = runtime
        if workspace_meta is not None:
            result["workspace_meta"] = workspace_meta
        if registry_status is not None:
            result["custom_field_registry"] = registry_status
    return result


def doctor(root: Path, quick: bool, *, repair_stale_sidecars: bool = False) -> dict[str, object]:
    set_active_workspace_root(root)
    paths = workspace_paths(root)
    runtime_paths = plugin_runtime_paths(root=root)
    plugin_runtime = describe_plugin_runtime(runtime_paths)
    fts5 = probe_fts5()
    pip_python = (
        runtime_paths["venv_python_path"]
        if runtime_paths is not None and runtime_paths["venv_python_path"].exists()
        else Path(sys.executable)
    )
    pip_ok, pip_version = run_command([str(pip_python), "-m", "pip", "--version"])
    pst_backend = dependency_status(
        "pypff",
        package_name="libpff-python",
        import_name="pypff",
        detail_label="PST backend",
        probe_if_unloaded=True,
        allow_auto_install=False,
    )
    runtime = read_runtime(paths["runtime_path"])
    canonical_tool_path = locate_canonical_plugin_tool_or_self()
    current_sha = sha256_file(canonical_tool_path) if canonical_tool_path is not None else None
    stored_sha = None if runtime is None else runtime.get("template_sha256")
    workspace_state = determine_workspace_state(paths)
    registry_status = None
    schema_status = None
    workspace_inventory = None
    processing_providers = probe_processing_providers(None)
    journal_mode = None
    db_error = None
    integrity_check: dict[str, object] | None = None
    stale_artifacts_before = [str(path) for path in stale_sqlite_artifact_paths(paths["db_path"])]
    removed_stale_artifacts: list[str] = []

    if repair_stale_sidecars and stale_artifacts_before:
        removed_stale_artifacts = remove_stale_sqlite_artifacts(paths["db_path"])

    if paths["db_path"].exists():
        try:
            connection = connect_db(paths["db_path"])
            try:
                journal_mode = current_journal_mode(connection)
                try:
                    integrity_rows = connection.execute("PRAGMA integrity_check").fetchall()
                    integrity_messages = [str(row[0]) for row in integrity_rows]
                    integrity_check = {
                        "status": "ok" if integrity_messages == ["ok"] else "error",
                        "messages": integrity_messages,
                    }
                except Exception as exc:
                    integrity_check = {
                        "status": "failed",
                        "messages": [f"{type(exc).__name__}: {exc}"],
                    }
                if integrity_check is None or str(integrity_check.get("status")) == "ok":
                    schema_status = apply_schema(connection, root)
                    registry_status = reconcile_custom_fields_registry(connection, repair=True)
                    workspace_inventory = document_inventory_counts(connection)
                    processing_providers = probe_processing_providers(connection)
            finally:
                connection.close()
        except Exception as exc:
            db_error = f"{type(exc).__name__}: {exc}"

    overall = "pass"
    if fts5["status"] != "pass":
        overall = "fail"
    elif db_error is not None:
        overall = "fail"
    elif integrity_check is not None and str(integrity_check.get("status") or "") != "ok":
        overall = "fail"
    elif workspace_state == "partial":
        overall = "partial"
    elif stale_artifacts_before and not removed_stale_artifacts:
        overall = "partial"

    result: dict[str, object] = {
        "overall": overall,
        "tool_version": TOOL_VERSION,
        "schema_version": SCHEMA_VERSION,
        "python_version": platform.python_version(),
        "pip_version": pip_version if pip_ok else None,
        "pip_status": "pass" if pip_ok else "fail",
        "sqlite_version": sqlite3.sqlite_version,
        "fts5": fts5,
        "pst_backend": pst_backend,
        "processing_providers": processing_providers,
        "platform": detect_platform(),
        "plugin_runtime": plugin_runtime,
        "workspace": {
            "root": str(root.resolve()),
            "state": workspace_state,
            "db_present": paths["db_path"].exists(),
            "db_size_bytes": file_size_bytes(paths["db_path"]),
            "runtime_present": paths["runtime_path"].exists(),
            "canonical_tool_present": canonical_tool_path is not None,
        },
        "tool_integrity": {
            "current_sha256": current_sha,
            "runtime_sha256": stored_sha,
            "matches_runtime": current_sha == stored_sha if current_sha and stored_sha else None,
        },
        "db": {
            "path": str(paths["db_path"]),
            "stale_sqlite_artifacts_before": stale_artifacts_before,
            "removed_stale_sqlite_artifacts": removed_stale_artifacts,
            "stale_sqlite_artifacts_after": [str(path) for path in stale_sqlite_artifact_paths(paths["db_path"])],
        },
    }
    if journal_mode is not None:
        result["sqlite_journal_mode"] = journal_mode
    if db_error is not None:
        result["db_error"] = db_error
    if workspace_inventory is not None:
        result["workspace_inventory"] = workspace_inventory
    if integrity_check is not None:
        result["db"]["integrity_check"] = integrity_check

    if not quick:
        result["paths"] = {key: str(value) for key, value in paths.items()}
        if runtime is not None:
            result["runtime"] = runtime
        if schema_status is not None:
            result["schema_apply"] = schema_status
        if registry_status is not None:
            result["custom_field_registry"] = registry_status
    return result


def bootstrap(root: Path) -> dict[str, object]:
    set_active_workspace_root(root)
    paths = workspace_paths(root)
    ensure_layout(paths)
    canonical_tool_path = locate_canonical_plugin_tool_or_self()
    tool_sha = (
        sha256_file(canonical_tool_path)
        if canonical_tool_path is not None
        else sha256_file(Path(__file__).resolve())
    )
    recovered_sqlite_artifacts = remove_stale_sqlite_artifacts(paths["db_path"])
    seeded_sqlite_db: dict[str, object] | None = None
    last_error: Exception | None = None
    for attempt in range(4):
        try:
            connection = connect_db(paths["db_path"])
            try:
                journal_mode = current_journal_mode(connection)
                apply_schema(connection, root)
                registry_status = reconcile_custom_fields_registry(connection, repair=True)
                write_workspace_meta(connection, tool_sha)
            finally:
                connection.close()
            write_runtime(paths, tool_sha)
            result = {
                "status": "initialized" if paths["runtime_path"].exists() else "failed",
                "workspace_root": str(root.resolve()),
                "schema_version": SCHEMA_VERSION,
                "tool_version": TOOL_VERSION,
                "requirements_version": REQUIREMENTS_VERSION,
                "custom_field_registry": registry_status,
            }
            if journal_mode is not None:
                result["journal_mode"] = journal_mode
            if recovered_sqlite_artifacts:
                result["recovered_sqlite_artifacts"] = recovered_sqlite_artifacts
            if seeded_sqlite_db is not None:
                result["seeded_sqlite_db"] = seeded_sqlite_db
            return result
        except Exception as exc:
            last_error = exc
            retry_artifacts = remove_stale_sqlite_artifacts(paths["db_path"])
            if retry_artifacts:
                for artifact in retry_artifacts:
                    if artifact not in recovered_sqlite_artifacts:
                        recovered_sqlite_artifacts.append(artifact)
                continue
            if seeded_sqlite_db is None and sqlite_bootstrap_seed_required(paths["db_path"], exc):
                seeded_sqlite_db = seed_sqlite_db_from_local_temp(paths["db_path"])
                for artifact in list(seeded_sqlite_db.get("reset_artifacts") or []):
                    if artifact not in recovered_sqlite_artifacts:
                        recovered_sqlite_artifacts.append(str(artifact))
                continue
            break
    detail = f"{type(last_error).__name__}: {last_error}" if last_error is not None else "unknown bootstrap failure"
    if recovered_sqlite_artifacts:
        detail = (
            f"{detail}. Removed stale SQLite artifacts before retry: "
            f"{', '.join(recovered_sqlite_artifacts)}"
        )
    if seeded_sqlite_db is not None:
        detail = (
            f"{detail}. Seeded SQLite DB from local temp with journal mode "
            f"{seeded_sqlite_db.get('journal_mode') or 'delete'} before retry"
        )
    raise RetrieverError(f"Bootstrap failed for {paths['db_path']}: {detail}") from last_error


# Commands that must not trigger an auto-upgrade. `schema-version` needs to
# work even when a workspace does not exist yet. `workspace` owns the
# explicit init/status/update flows, so it should not transparently rewrite
# itself before dispatch. `slash` is intentionally not exempt so user-facing
# slash commands also benefit from clean-but-stale auto-upgrades.
AUTO_UPGRADE_EXEMPT_COMMANDS = frozenset(
    {
        "schema-version",
        "workspace",
    }
)


def locate_canonical_plugin_tool(current_file: str | None = None) -> Path | None:
    """Find the canonical ``skills/tool-template/tools.py`` bundle."""

    def _safe_resolve(path: Path) -> Path | None:
        try:
            return path.resolve()
        except OSError:
            return None

    env_path = os.environ.get("RETRIEVER_CANONICAL_TOOL_PATH")
    if env_path:
        candidate = Path(env_path).expanduser()
        if candidate.is_file():
            resolved = _safe_resolve(candidate)
            if resolved is not None:
                return resolved

    candidate_path = current_file or __file__
    try:
        start = Path(candidate_path).resolve()
    except OSError:
        start = None
    if start is None:
        return None

    sibling_candidate = start.with_name("tools.py")
    sibling_resolved = _safe_resolve(sibling_candidate)
    if sibling_resolved is not None and sibling_resolved.is_file():
        return sibling_resolved

    for parent in [start.parent, *start.parents]:
        candidate = parent / "skills" / "tool-template" / "tools.py"
        resolved = _safe_resolve(candidate)
        if resolved is not None and resolved.is_file():
            return resolved
    return None


def locate_canonical_plugin_tool_or_self(current_file: str | None = None) -> Path | None:
    located = locate_canonical_plugin_tool(current_file)
    if located is not None:
        return located

    candidate_path = current_file or __file__
    try:
        candidate = Path(candidate_path).resolve()
    except OSError:
        return None
    if (
        candidate.is_file()
        and candidate.parent.name == "tool-template"
        and candidate.parent.parent.name == "skills"
        and candidate.name == "tools.py"
    ):
        return candidate
    return None
    if (
        candidate.is_file()
        and candidate.parent.name == "tool-template"
        and candidate.parent.parent.name == "skills"
    ):
        return candidate
    return None


def upgrade_workspace_tool(
    root: Path,
    canonical_path: Path,
    *,
    force: bool = False,
    reason: str = "manual",
) -> dict[str, object]:
    """Refresh workspace runtime metadata from the canonical tool bundle."""
    del force
    paths = workspace_paths(root)
    ensure_layout(paths)
    runtime = read_runtime(paths["runtime_path"])
    runtime_sha = runtime.get("template_sha256") if isinstance(runtime, dict) else None
    runtime_version = runtime.get("tool_version") if isinstance(runtime, dict) else None
    canonical_sha = sha256_file(canonical_path)
    if canonical_sha is None:
        raise RetrieverError(f"Canonical tool not readable at {canonical_path}")

    write_runtime(paths, canonical_sha)

    meta_updated = False
    meta_error: str | None = None
    try:
        connection = connect_db(paths["db_path"])
        try:
            write_workspace_meta(connection, canonical_sha)
            meta_updated = True
        finally:
            connection.close()
    except Exception as exc:  # pragma: no cover - best-effort path
        meta_error = f"{type(exc).__name__}: {exc}"

    result: dict[str, object] = {
        "status": "no-op" if runtime_sha == canonical_sha else "updated-runtime",
        "reason": reason,
        "previous_tool_sha256": runtime_sha,
        "previous_tool_version": runtime_version,
        "new_tool_sha256": canonical_sha,
        "new_tool_version": TOOL_VERSION,
        "canonical_path": str(canonical_path),
        "workspace_meta_updated": meta_updated,
    }
    if meta_error is not None:
        result["workspace_meta_error"] = meta_error
    return result


def maybe_upgrade_workspace_tool(root: Path) -> dict[str, object] | None:
    """Best-effort runtime metadata refresh for a workspace."""
    paths = workspace_paths(root)
    if not paths["state_dir"].exists():
        return None
    runtime = read_runtime(paths["runtime_path"])
    if not isinstance(runtime, dict):
        return None

    canonical_path = locate_canonical_plugin_tool_or_self()
    if canonical_path is None:
        return None
    canonical_sha = sha256_file(canonical_path)
    if canonical_sha is None:
        return None
    if runtime.get("template_sha256") == canonical_sha and runtime.get("template_source") == TEMPLATE_SOURCE:
        return None
    return upgrade_workspace_tool(root, canonical_path, reason="auto-runtime-sync")


def init_workspace(root: Path, quick: bool = False) -> dict[str, object]:
    set_active_workspace_root(root)
    paths = workspace_paths(root)
    runtime_paths = plugin_runtime_paths(root=root)
    tool_update: dict[str, object] | None = None
    canonical_path = locate_canonical_plugin_tool_or_self()

    if canonical_path is not None:
        if not paths["runtime_path"].exists():
            tool_update = upgrade_workspace_tool(
                root,
                canonical_path,
                force=False,
                reason="workspace-init",
            )
        else:
            maybe_update = maybe_upgrade_workspace_tool(root)
            if maybe_update is not None:
                tool_update = maybe_update

    runtime_init = ensure_plugin_runtime(
        runtime_paths,
        install_requirements=True,
        force_requirements_install=False,
        reason="init",
    )
    initialization_payload = bootstrap(root)
    result: dict[str, object] = {
        "action": "init",
        "status": "ready",
        "workspace_root": str(root.resolve()),
        "initialization": initialization_payload,
        "status_report": workspace_status(root, quick=quick),
    }
    if tool_update is not None:
        result["tool_update"] = tool_update
    if payload_has_meaningful_value(runtime_init):
        result["runtime_init"] = runtime_init
    return result


def write_workspace_meta(connection: sqlite3.Connection, tool_sha256: str | None) -> None:
    now = utc_now()
    connection.execute(
        """
        INSERT INTO workspace_meta (
          id, schema_version, tool_version, requirements_version,
          template_source, template_sha256, created_at, updated_at
        ) VALUES (1, ?, ?, ?, ?, ?, ?, ?)
        ON CONFLICT(id) DO UPDATE SET
          schema_version = excluded.schema_version,
          tool_version = excluded.tool_version,
          requirements_version = excluded.requirements_version,
          template_source = excluded.template_source,
          template_sha256 = excluded.template_sha256,
          updated_at = excluded.updated_at
        """,
        (
            SCHEMA_VERSION,
            TOOL_VERSION,
            REQUIREMENTS_VERSION,
            TEMPLATE_SOURCE,
            tool_sha256,
            now,
            now,
        ),
    )
    connection.commit()


def workspace_runtime_metadata_needs_materialization(
    paths: dict[str, Path],
    connection: sqlite3.Connection,
) -> bool:
    if not paths["runtime_path"].exists():
        return True
    workspace_meta = read_workspace_meta(connection)
    if workspace_meta is None:
        return True
    try:
        return int(workspace_meta.get("schema_version")) != SCHEMA_VERSION
    except (TypeError, ValueError):
        return True


def ensure_workspace_runtime_metadata(root: Path, connection: sqlite3.Connection) -> None:
    paths = workspace_paths(root)
    if not workspace_runtime_metadata_needs_materialization(paths, connection):
        return
    canonical_tool_path = locate_canonical_plugin_tool_or_self()
    tool_path = canonical_tool_path or Path(__file__).resolve()
    tool_sha = sha256_file(tool_path)
    if tool_sha is None:
        raise RetrieverError(f"Canonical tool not readable at {tool_path}")
    write_workspace_meta(connection, tool_sha)
    write_runtime(paths, tool_sha)


def resolve_production_root_argument(workspace_root: Path, raw_production_root: str | Path) -> Path:
    candidate = Path(raw_production_root).expanduser()
    if not candidate.is_absolute():
        candidate = workspace_root / candidate
    candidate = candidate.resolve()
    try:
        candidate.relative_to(workspace_root.resolve())
    except ValueError as exc:
        raise RetrieverError("Production root must be inside the workspace root for Phase 4 ingest.") from exc
    return candidate


def production_row_signature(
    existing_row: sqlite3.Row | None,
    *,
    rel_path: str,
    file_name: str,
    source_kind: str,
    production_id: int,
    begin_bates: str,
    end_bates: str,
    begin_attachment: str | None,
    end_attachment: str | None,
    extracted: dict[str, object],
    source_parts: list[dict[str, object]],
) -> tuple[object, ...]:
    existing_locks = existing_row[MANUAL_FIELD_LOCKS_COLUMN] if existing_row is not None else "[]"
    return (
        rel_path,
        file_name,
        source_kind,
        production_id,
        begin_bates,
        end_bates,
        begin_attachment,
        end_attachment,
        extracted.get("title"),
        extracted.get("content_type"),
        extracted.get("text_status"),
        sha256_text(str(extracted.get("text_content") or "")),
        extracted.get("page_count"),
        tuple(
            sorted(
                (part["part_kind"], part["rel_source_path"], int(part.get("ordinal", 0)))
                for part in source_parts
            )
        ),
        bool(extracted.get("preview_artifacts")),
        existing_locks,
    )


def existing_production_row_signature(connection: sqlite3.Connection, row: sqlite3.Row | None) -> tuple[object, ...] | None:
    if row is None:
        return None
    source_parts = connection.execute(
        """
        SELECT part_kind, rel_source_path, ordinal
        FROM document_source_parts
        WHERE document_id = ?
        ORDER BY part_kind ASC, ordinal ASC, id ASC
        """,
        (row["id"],),
    ).fetchall()
    return (
        row["rel_path"],
        row["file_name"],
        row["source_kind"],
        row["production_id"],
        row["begin_bates"],
        row["end_bates"],
        row["begin_attachment"],
        row["end_attachment"],
        row["title"],
        row["content_type"],
        row["text_status"],
        row["content_hash"],
        row["page_count"],
        tuple((part["part_kind"], part["rel_source_path"], int(part["ordinal"])) for part in source_parts),
        connection.execute("SELECT COUNT(*) AS count FROM document_previews WHERE document_id = ?", (row["id"],)).fetchone()["count"] > 0,
        row[MANUAL_FIELD_LOCKS_COLUMN],
    )