unsloth/studio/backend/storage/studio_db.py

# SPDX-License-Identifier: AGPL-3.0-only
# Copyright 2026-present the Unsloth AI Inc. team. All rights reserved. See /studio/LICENSE.AGPL-3.0

"""SQLite storage for training run history and metrics.

Like auth/storage.py (module-level functions, raw sqlite3, per-function
connections) plus WAL mode and PRAGMA foreign_keys = ON for CASCADE deletes.
"""

import hashlib
import json
import logging
import os
import platform
import re
import shutil
import sqlite3
import threading
from datetime import datetime, timezone
from pathlib import Path

logger = logging.getLogger(__name__)
from typing import Any, Iterable, Optional


from utils.paths import (
    ensure_dir,
    project_workspaces_root,
    studio_db_path,
)
from utils.paths.external_media import is_linux_run_media_path, is_local_filesystem_root
from utils.paths.sensitive import (
    contains_sensitive_path_component as _shared_contains_sensitive_path_component,
)
from utils.training_runs import extract_project_name


def _extract_project_name_from_config_json(config_json: Optional[str]) -> Optional[str]:
    if not config_json:
        return None
    try:
        return extract_project_name(json.loads(config_json))
    except (json.JSONDecodeError, TypeError):
        return None


def _denied_path_prefixes() -> list[str]:
    """Platform-aware denylist of system directories."""
    system = platform.system()
    if system == "Linux":
        return ["/proc", "/sys", "/dev", "/etc", "/boot", "/run"]
    if system == "Darwin":
        # macOS realpath() resolves /etc -> /private/etc etc; include /private variants.
        return [
            "/System",
            "/Library",
            "/dev",
            "/etc",
            "/private/etc",
            "/tmp",
            "/private/tmp",
            "/var",
            "/private/var",
        ]
    if system == "Windows":
        win = os.environ.get("SystemRoot", r"C:\Windows")
        pf = os.environ.get("ProgramFiles", r"C:\Program Files")
        pf86 = os.environ.get("ProgramFiles(x86)", r"C:\Program Files (x86)")
        return [os.path.normcase(p) for p in [win, pf, pf86]]
    return []


def is_denied_system_path(path: str) -> bool:
    """True if *path* is, or descends from, a denied system directory.

    Mirrors the denylist add_scan_folder() enforces at registration so the
    browser refuses /etc, /proc, C:\\Windows, etc. even when the allowlist holds
    a broad root (a Windows drive root C:\\ or a legacy-registered / root). The
    /run carve-out keeps Linux removable-media mounts browseable. Expects an
    already-resolved (realpath) path so symlinks cannot escape into a denied subtree.
    """
    is_win = platform.system() == "Windows"
    check = os.path.normcase(path) if is_win else path
    for prefix in _denied_path_prefixes():
        if check == prefix or check.startswith(prefix + os.sep):
            if prefix == "/run" and is_linux_run_media_path(check):
                continue
            return True
    return False


def _contains_sensitive_path_component(path: str) -> bool:
    return _shared_contains_sensitive_path_component(path)


def contains_sensitive_path_component(path: str) -> bool:
    return _contains_sensitive_path_component(path)


_schema_lock = threading.Lock()
_schema_ready = False
_SQLITE_IN_CHUNK_SIZE = 900
_PROJECT_WORKSPACE_SUBDIRS = ("sandbox",)
_CHAT_ATTACHMENT_INVENTORY_VERSION = 1


def _project_slug(name: str) -> str:
    slug = re.sub(r"[^A-Za-z0-9._-]+", "-", name.strip()).strip(".-_")
    return slug[:48] or "project"


def _default_project_root(project: dict) -> str:
    project_id = str(project["id"])
    suffix = re.sub(r"[^A-Za-z0-9_-]+", "-", project_id)[:8].strip("-_") or "project"
    folder_name = f"{_project_slug(str(project.get('name') or 'Project'))}-{suffix}"
    return str(project_workspaces_root() / folder_name)


def _ensure_project_workspace(root_path: str) -> str:
    root = Path(root_path).expanduser()
    root_resolved = ensure_dir(root).resolve()
    for subdir in _PROJECT_WORKSPACE_SUBDIRS:
        ensure_dir(root_resolved / subdir)
    return str(root_resolved)


def _delete_project_workspace(project: dict) -> None:
    root_path = project.get("rootPath")
    if not root_path:
        return
    root = Path(root_path).expanduser()
    try:
        root_resolved = root.resolve(strict = False)
    except (OSError, RuntimeError, ValueError):
        logger.warning("Skipping project workspace delete for invalid path %r", root_path)
        return

    project_id = str(project["id"])
    suffix = re.sub(r"[^A-Za-z0-9_-]+", "-", project_id)[:8].strip("-_") or "project"
    if not root_resolved.name.endswith(f"-{suffix}"):
        logger.warning(
            "Skipping project workspace delete for unexpected project path %s",
            root_resolved,
        )
        return
    if root_resolved.parent == root_resolved or root_resolved == Path.home().resolve():
        logger.warning(
            "Skipping project workspace delete for unsafe project path %s",
            root_resolved,
        )
        return
    check = (
        os.path.normcase(str(root_resolved))
        if platform.system() == "Windows"
        else str(root_resolved)
    )
    for prefix in _denied_path_prefixes():
        if check == prefix or check.startswith(prefix + os.sep):
            logger.warning(
                "Skipping project workspace delete under denied path %s",
                root_resolved,
            )
            return
    if not root_resolved.exists():
        return
    if root_resolved.is_symlink() or not root_resolved.is_dir():
        logger.warning(
            "Skipping project workspace delete for non-directory path %s",
            root_resolved,
        )
        return
    shutil.rmtree(root_resolved)


def _ensure_schema(conn: sqlite3.Connection) -> None:
    """Create tables and indexes if they don't exist. Called once per process."""
    conn.execute("PRAGMA journal_mode=WAL")
    conn.execute(
        """
        CREATE TABLE IF NOT EXISTS training_runs (
            id TEXT NOT NULL PRIMARY KEY,
            status TEXT NOT NULL DEFAULT 'running',
            model_name TEXT NOT NULL,
            dataset_name TEXT NOT NULL,
            config_json TEXT NOT NULL,
            started_at TEXT NOT NULL,
            ended_at TEXT,
            total_steps INTEGER,
            final_step INTEGER,
            final_loss REAL,
            output_dir TEXT,
            error_message TEXT,
            duration_seconds REAL,
            loss_sparkline TEXT,
            display_name TEXT,
            resume_blocked INTEGER NOT NULL DEFAULT 0
        )
        """
    )
    existing_cols = {row[1] for row in conn.execute("PRAGMA table_info(training_runs)").fetchall()}
    if "display_name" not in existing_cols:
        conn.execute("ALTER TABLE training_runs ADD COLUMN display_name TEXT")
    if "resume_blocked" not in existing_cols:
        conn.execute(
            "ALTER TABLE training_runs ADD COLUMN resume_blocked INTEGER NOT NULL DEFAULT 0"
        )
    conn.execute(
        """
        CREATE TABLE IF NOT EXISTS training_metrics (
            id INTEGER PRIMARY KEY AUTOINCREMENT,
            run_id TEXT NOT NULL REFERENCES training_runs(id) ON DELETE CASCADE,
            step INTEGER NOT NULL,
            loss REAL,
            learning_rate REAL,
            grad_norm REAL,
            eval_loss REAL,
            epoch REAL,
            num_tokens INTEGER,
            elapsed_seconds REAL,
            UNIQUE(run_id, step)
        )
        """
    )
    conn.execute("CREATE INDEX IF NOT EXISTS idx_metrics_run_id ON training_metrics(run_id)")
    # Windows: COLLATE NOCASE so C:\Models and c:\models dedup. Elsewhere keep
    # case-sensitive BINARY so /Models and /models stay distinct.
    collation = "COLLATE NOCASE" if platform.system() == "Windows" else ""
    conn.execute(
        f"""
        CREATE TABLE IF NOT EXISTS scan_folders (
            id INTEGER PRIMARY KEY AUTOINCREMENT,
            path TEXT NOT NULL UNIQUE {collation},
            created_at TEXT NOT NULL
        )
        """
    )
    conn.execute(
        """
        CREATE TABLE IF NOT EXISTS chat_projects (
            id TEXT NOT NULL PRIMARY KEY,
            name TEXT NOT NULL,
            instructions TEXT,
            root_path TEXT,
            archived INTEGER NOT NULL DEFAULT 0,
            created_at INTEGER NOT NULL,
            updated_at INTEGER NOT NULL
        )
        """
    )
    chat_project_cols = {
        row[1] for row in conn.execute("PRAGMA table_info(chat_projects)").fetchall()
    }
    if "root_path" not in chat_project_cols:
        conn.execute("ALTER TABLE chat_projects ADD COLUMN root_path TEXT")
    conn.execute(
        "CREATE INDEX IF NOT EXISTS idx_chat_projects_archived_updated_at ON chat_projects(archived, updated_at)"
    )
    conn.execute(
        """
        CREATE TABLE IF NOT EXISTS chat_threads (
            id TEXT NOT NULL PRIMARY KEY,
            title TEXT NOT NULL,
            model_type TEXT NOT NULL,
            model_id TEXT,
            pair_id TEXT,
            project_id TEXT,
            archived INTEGER NOT NULL DEFAULT 0,
            created_at INTEGER NOT NULL,
            updated_at INTEGER,
            openai_code_exec_container_id TEXT,
            anthropic_code_exec_container_id TEXT,
            forked_from_thread_id TEXT,
            forked_from_message_id TEXT,
            FOREIGN KEY(project_id) REFERENCES chat_projects(id) ON DELETE CASCADE
        )
        """
    )
    chat_thread_cols = {
        row[1] for row in conn.execute("PRAGMA table_info(chat_threads)").fetchall()
    }
    if "project_id" not in chat_thread_cols:
        conn.execute("ALTER TABLE chat_threads ADD COLUMN project_id TEXT")
    if "openai_code_exec_container_id" not in chat_thread_cols:
        conn.execute("ALTER TABLE chat_threads ADD COLUMN openai_code_exec_container_id TEXT")
    if "anthropic_code_exec_container_id" not in chat_thread_cols:
        conn.execute("ALTER TABLE chat_threads ADD COLUMN anthropic_code_exec_container_id TEXT")
    if "forked_from_thread_id" not in chat_thread_cols:
        conn.execute("ALTER TABLE chat_threads ADD COLUMN forked_from_thread_id TEXT")
    if "forked_from_message_id" not in chat_thread_cols:
        conn.execute("ALTER TABLE chat_threads ADD COLUMN forked_from_message_id TEXT")
    if "updated_at" not in chat_thread_cols:
        conn.execute("ALTER TABLE chat_threads ADD COLUMN updated_at INTEGER")
        # Floor at created_at: forked threads copy older ancestor messages,
        # so the fork's creation time must win over the branch message times.
        conn.execute(
            """
            UPDATE chat_threads SET updated_at = MAX(
                COALESCE(
                    (
                        SELECT MAX(m.created_at) FROM chat_messages m
                        WHERE m.thread_id = chat_threads.id
                    ),
                    created_at
                ),
                created_at
            )
            """
        )
    conn.execute(
        """
        CREATE TABLE IF NOT EXISTS chat_messages (
            id TEXT NOT NULL PRIMARY KEY,
            thread_id TEXT NOT NULL REFERENCES chat_threads(id) ON DELETE CASCADE,
            parent_id TEXT,
            role TEXT NOT NULL,
            content_json TEXT NOT NULL,
            attachments_json TEXT,
            metadata_json TEXT,
            created_at INTEGER NOT NULL
        )
        """
    )
    tombstone_schema = """
        CREATE TABLE chat_attachment_tombstones (
            thread_id TEXT NOT NULL REFERENCES chat_threads(id) ON DELETE CASCADE,
            message_id TEXT NOT NULL,
            attachment_id TEXT NOT NULL,
            deleted_at INTEGER NOT NULL,
            PRIMARY KEY(thread_id, message_id, attachment_id)
        ) WITHOUT ROWID
    """
    tombstone_table = conn.execute(
        """
        SELECT 1 FROM sqlite_master
        WHERE type = 'table' AND name = 'chat_attachment_tombstones'
        """
    ).fetchone()
    if tombstone_table is None:
        conn.execute(tombstone_schema)
    else:
        tombstone_columns = {
            row[1] for row in conn.execute("PRAGMA table_info(chat_attachment_tombstones)")
        }
        tombstone_fk_targets = {
            row[2] for row in conn.execute("PRAGMA foreign_key_list(chat_attachment_tombstones)")
        }
        if "thread_id" not in tombstone_columns or "chat_threads" not in tombstone_fk_targets:
            # The first implementation cascaded through chat_messages, which
            # erased deletion knowledge during pruneMissing. Rebuild once,
            # retaining every tombstone whose owning thread still exists.
            conn.execute("SAVEPOINT migrate_chat_attachment_tombstones")
            try:
                conn.execute(
                    "ALTER TABLE chat_attachment_tombstones "
                    "RENAME TO chat_attachment_tombstones_legacy"
                )
                conn.execute(tombstone_schema)
                if "thread_id" in tombstone_columns:
                    conn.execute(
                        """
                        INSERT OR IGNORE INTO chat_attachment_tombstones
                            (thread_id, message_id, attachment_id, deleted_at)
                        SELECT legacy.thread_id, legacy.message_id,
                               legacy.attachment_id, legacy.deleted_at
                        FROM chat_attachment_tombstones_legacy legacy
                        JOIN chat_threads thread ON thread.id = legacy.thread_id
                        """
                    )
                else:
                    conn.execute(
                        """
                        INSERT OR IGNORE INTO chat_attachment_tombstones
                            (thread_id, message_id, attachment_id, deleted_at)
                        SELECT message.thread_id, legacy.message_id,
                               legacy.attachment_id, legacy.deleted_at
                        FROM chat_attachment_tombstones_legacy legacy
                        JOIN chat_messages message ON message.id = legacy.message_id
                        """
                    )
                conn.execute("DROP TABLE chat_attachment_tombstones_legacy")
                conn.execute("RELEASE SAVEPOINT migrate_chat_attachment_tombstones")
            except Exception:
                conn.execute("ROLLBACK TO SAVEPOINT migrate_chat_attachment_tombstones")
                conn.execute("RELEASE SAVEPOINT migrate_chat_attachment_tombstones")
                raise
    conn.execute(
        """
        CREATE TABLE IF NOT EXISTS chat_attachment_inventory (
            message_id TEXT NOT NULL REFERENCES chat_messages(id) ON DELETE CASCADE,
            attachment_id TEXT NOT NULL,
            name TEXT NOT NULL,
            type TEXT,
            content_type TEXT,
            size_bytes INTEGER,
            PRIMARY KEY(message_id, attachment_id)
        ) WITHOUT ROWID
        """
    )
    conn.execute(
        """
        CREATE TABLE IF NOT EXISTS chat_attachment_inventory_state (
            singleton INTEGER NOT NULL PRIMARY KEY CHECK(singleton = 1),
            inventory_version INTEGER NOT NULL DEFAULT 0,
            dirty INTEGER NOT NULL DEFAULT 1,
            backfilled_at INTEGER NOT NULL
        )
        """
    )
    inventory_state_columns = {
        row[1] for row in conn.execute("PRAGMA table_info(chat_attachment_inventory_state)")
    }
    if "inventory_version" not in inventory_state_columns:
        conn.execute(
            "ALTER TABLE chat_attachment_inventory_state "
            "ADD COLUMN inventory_version INTEGER NOT NULL DEFAULT 0"
        )
    if "dirty" not in inventory_state_columns:
        conn.execute(
            "ALTER TABLE chat_attachment_inventory_state "
            "ADD COLUMN dirty INTEGER NOT NULL DEFAULT 1"
        )
    conn.execute(
        """
        CREATE TRIGGER IF NOT EXISTS chat_attachment_inventory_dirty_insert
        AFTER INSERT ON chat_messages
        BEGIN
            INSERT INTO chat_attachment_inventory_state
                (singleton, inventory_version, dirty, backfilled_at)
            VALUES (1, 0, 1, 0)
            ON CONFLICT(singleton) DO UPDATE SET dirty = 1;
        END
        """
    )
    conn.execute(
        """
        CREATE TRIGGER IF NOT EXISTS chat_attachment_inventory_dirty_update
        AFTER UPDATE ON chat_messages
        BEGIN
            INSERT INTO chat_attachment_inventory_state
                (singleton, inventory_version, dirty, backfilled_at)
            VALUES (1, 0, 1, 0)
            ON CONFLICT(singleton) DO UPDATE SET dirty = 1;
        END
        """
    )
    conn.execute(
        """
        CREATE TRIGGER IF NOT EXISTS chat_attachment_inventory_dirty_delete
        AFTER DELETE ON chat_messages
        BEGIN
            INSERT INTO chat_attachment_inventory_state
                (singleton, inventory_version, dirty, backfilled_at)
            VALUES (1, 0, 1, 0)
            ON CONFLICT(singleton) DO UPDATE SET dirty = 1;
        END
        """
    )
    conn.execute(
        "CREATE INDEX IF NOT EXISTS idx_chat_threads_model_type_created_at ON chat_threads(model_type, created_at)"
    )
    conn.execute("CREATE INDEX IF NOT EXISTS idx_chat_threads_pair_id ON chat_threads(pair_id)")
    conn.execute(
        "CREATE INDEX IF NOT EXISTS idx_chat_threads_project_id ON chat_threads(project_id)"
    )
    conn.execute(
        "CREATE INDEX IF NOT EXISTS idx_chat_messages_thread_id_created_at ON chat_messages(thread_id, created_at)"
    )
    conn.execute(
        """
        CREATE TABLE IF NOT EXISTS chat_settings (
            key TEXT NOT NULL PRIMARY KEY,
            value_json TEXT NOT NULL,
            updated_at TEXT NOT NULL
        )
        """
    )
    conn.execute(
        """
        CREATE TABLE IF NOT EXISTS app_settings (
            key TEXT NOT NULL PRIMARY KEY,
            value_json TEXT NOT NULL,
            updated_at TEXT NOT NULL
        )
        """
    )
    conn.execute(
        """
        CREATE TABLE IF NOT EXISTS chat_settings_quarantine (
            id INTEGER PRIMARY KEY AUTOINCREMENT,
            key TEXT NOT NULL,
            value_json TEXT NOT NULL,
            reason TEXT NOT NULL,
            quarantined_at TEXT NOT NULL
        )
        """
    )
    # Import ledger inside studio.db (vs. a localStorage boolean) so a db wipe
    # re-triggers the legacy Dexie import instead of silently hiding threads.
    # Keyed by legacy thread id; Dexie is read-only so per-thread suffices.
    conn.execute(
        """
        CREATE TABLE IF NOT EXISTS chat_legacy_imports (
            legacy_thread_id TEXT NOT NULL PRIMARY KEY,
            imported_at INTEGER NOT NULL
        ) WITHOUT ROWID
        """
    )
    conn.execute(
        """
        CREATE TABLE IF NOT EXISTS prompt_entries (
            id TEXT NOT NULL PRIMARY KEY,
            name TEXT NOT NULL,
            text TEXT NOT NULL,
            created_at INTEGER NOT NULL,
            updated_at INTEGER NOT NULL
        )
        """
    )
    conn.execute(
        "CREATE INDEX IF NOT EXISTS idx_prompt_entries_created_at ON prompt_entries(created_at)"
    )
    conn.execute(
        """
        CREATE TABLE IF NOT EXISTS prompt_lists (
            id TEXT NOT NULL PRIMARY KEY,
            name TEXT NOT NULL,
            items_json TEXT NOT NULL,
            created_at INTEGER NOT NULL,
            updated_at INTEGER NOT NULL
        )
        """
    )
    conn.execute(
        "CREATE INDEX IF NOT EXISTS idx_prompt_lists_created_at ON prompt_lists(created_at)"
    )
    conn.execute(
        """
        CREATE TABLE IF NOT EXISTS research_runs (
            id TEXT NOT NULL PRIMARY KEY,
            owner_subject TEXT NOT NULL,
            thread_id TEXT NOT NULL REFERENCES chat_threads(id) ON DELETE CASCADE,
            user_message_id TEXT NOT NULL REFERENCES chat_messages(id) ON DELETE CASCADE,
            assistant_message_id TEXT REFERENCES chat_messages(id) ON DELETE SET NULL,
            status TEXT NOT NULL CHECK(status IN (
                'planning', 'awaiting_approval', 'queued', 'running', 'paused',
                'cancelling', 'cancelled', 'completed', 'failed'
            )),
            plan_json TEXT,
            plan_revision INTEGER NOT NULL DEFAULT 0,
            plan_hash TEXT,
            config_json TEXT NOT NULL,
            cancel_requested INTEGER NOT NULL DEFAULT 0,
            lease_owner TEXT,
            lease_expires_at INTEGER,
            heartbeat_at INTEGER,
            retry_count INTEGER NOT NULL DEFAULT 0,
            error_message TEXT,
            report_text TEXT,
            created_at INTEGER NOT NULL,
            updated_at INTEGER NOT NULL,
            started_at INTEGER,
            completed_at INTEGER,
            next_event_seq INTEGER NOT NULL DEFAULT 1
        )
        """
    )
    research_run_cols = {
        row[1] for row in conn.execute("PRAGMA table_info(research_runs)").fetchall()
    }
    if "report_text" not in research_run_cols:
        conn.execute("ALTER TABLE research_runs ADD COLUMN report_text TEXT")
    conn.execute(
        """
        CREATE TABLE IF NOT EXISTS research_thread_claims (
            owner_subject TEXT NOT NULL,
            thread_id TEXT NOT NULL PRIMARY KEY REFERENCES chat_threads(id) ON DELETE CASCADE,
            created_at INTEGER NOT NULL
        ) WITHOUT ROWID
        """
    )
    claim_pk = [
        row[1]
        for row in sorted(
            conn.execute("PRAGMA table_info(research_thread_claims)").fetchall(),
            key = lambda row: int(row[5] or 0),
        )
        if int(row[5] or 0) > 0
    ]
    if claim_pk != ["thread_id"]:
        # Rebuild the claims table (legacy owner_subject+thread_id PK -> thread_id PK) atomically.
        # Without an explicit transaction the RENAME/CREATE/INSERT/DROP run in autocommit, so an
        # interruption after CREATE orphaned the rows in _legacy and never re-triggered.
        conn.commit()
        conn.execute("BEGIN IMMEDIATE")
        try:
            conn.execute(
                "ALTER TABLE research_thread_claims RENAME TO research_thread_claims_legacy"
            )
            conn.execute(
                """
                CREATE TABLE research_thread_claims (
                    owner_subject TEXT NOT NULL,
                    thread_id TEXT NOT NULL PRIMARY KEY REFERENCES chat_threads(id) ON DELETE CASCADE,
                    created_at INTEGER NOT NULL
                ) WITHOUT ROWID
                """
            )
            conn.execute(
                """INSERT OR IGNORE INTO research_thread_claims
                   (owner_subject, thread_id, created_at)
                   SELECT owner_subject, thread_id, created_at
                   FROM research_thread_claims_legacy
                   ORDER BY created_at, owner_subject"""
            )
            conn.execute("DROP TABLE research_thread_claims_legacy")
            conn.commit()
        except Exception:
            conn.rollback()
            raise
    conn.execute(
        """INSERT OR IGNORE INTO research_thread_claims
           (owner_subject, thread_id, created_at)
           SELECT owner_subject, thread_id, created_at
           FROM research_runs ORDER BY created_at, id"""
    )
    conn.execute(
        """UPDATE research_runs
           SET status='failed', error_message='Superseded by the global thread research claim',
               lease_owner=NULL, lease_expires_at=NULL, completed_at=COALESCE(completed_at, updated_at)
           WHERE status IN ('planning','awaiting_approval','queued','running','paused','cancelling')
             AND EXISTS (
                 SELECT 1 FROM research_thread_claims c
                 WHERE c.thread_id=research_runs.thread_id
                   AND c.owner_subject<>research_runs.owner_subject
             )"""
    )
    conn.execute(
        """
        CREATE TABLE IF NOT EXISTS research_plan_steps (
            run_id TEXT NOT NULL REFERENCES research_runs(id) ON DELETE CASCADE,
            position INTEGER NOT NULL,
            title TEXT NOT NULL,
            query TEXT NOT NULL,
            status TEXT NOT NULL DEFAULT 'pending',
            result_json TEXT,
            started_at INTEGER,
            completed_at INTEGER,
            PRIMARY KEY(run_id, position)
        ) WITHOUT ROWID
        """
    )
    conn.execute(
        """
        CREATE TABLE IF NOT EXISTS research_sources (
            id INTEGER PRIMARY KEY AUTOINCREMENT,
            run_id TEXT NOT NULL REFERENCES research_runs(id) ON DELETE CASCADE,
            step_position INTEGER,
            url TEXT NOT NULL,
            title TEXT,
            snippet TEXT,
            fetched_at INTEGER NOT NULL,
            UNIQUE(run_id, url)
        )
        """
    )
    conn.execute(
        """
        CREATE TABLE IF NOT EXISTS research_document_sources (
            id INTEGER PRIMARY KEY AUTOINCREMENT,
            run_id TEXT NOT NULL REFERENCES research_runs(id) ON DELETE CASCADE,
            step_position INTEGER,
            source_key TEXT NOT NULL,
            document_id TEXT,
            chunk_id TEXT,
            filename TEXT NOT NULL,
            page INTEGER,
            score REAL,
            snippet TEXT,
            fetched_at INTEGER NOT NULL,
            UNIQUE(run_id, source_key)
        )
        """
    )
    conn.execute(
        """
        CREATE TABLE IF NOT EXISTS research_events (
            run_id TEXT NOT NULL REFERENCES research_runs(id) ON DELETE CASCADE,
            seq INTEGER NOT NULL,
            event_type TEXT NOT NULL,
            data_json TEXT NOT NULL,
            created_at INTEGER NOT NULL,
            PRIMARY KEY(run_id, seq)
        ) WITHOUT ROWID
        """
    )
    conn.execute(
        "CREATE INDEX IF NOT EXISTS idx_research_runs_owner_thread_status "
        "ON research_runs(owner_subject, thread_id, status)"
    )
    conn.execute(
        "CREATE INDEX IF NOT EXISTS idx_research_runs_lease "
        "ON research_runs(status, lease_expires_at)"
    )
    conn.execute(
        "CREATE INDEX IF NOT EXISTS idx_research_sources_run ON research_sources(run_id, id)"
    )
    conn.execute(
        "CREATE INDEX IF NOT EXISTS idx_research_document_sources_run "
        "ON research_document_sources(run_id, id)"
    )
    inventory_state = conn.execute(
        """
        SELECT inventory_version, dirty
        FROM chat_attachment_inventory_state
        WHERE singleton = 1
        """
    ).fetchone()
    # Positional read: works for raw tuple or sqlite3.Row (no row_factory precondition).
    if (
        inventory_state is None
        or inventory_state[0] != _CHAT_ATTACHMENT_INVENTORY_VERSION
        or inventory_state[1]
    ):
        _rebuild_chat_attachment_inventory(conn)
        _mark_chat_attachment_inventory_clean(conn)
    conn.commit()


def _prompt_entry_from_row(row: sqlite3.Row) -> dict:
    return {
        "id": row["id"],
        "name": row["name"],
        "text": row["text"],
        "createdAt": row["created_at"],
        "updatedAt": row["updated_at"],
    }


def list_prompt_entries() -> list[dict]:
    conn = get_connection()
    try:
        rows = conn.execute("SELECT * FROM prompt_entries ORDER BY created_at DESC").fetchall()
        return [_prompt_entry_from_row(r) for r in rows]
    finally:
        conn.close()


def upsert_prompt_entry(entry: dict) -> dict:
    conn = get_connection()
    try:
        conn.execute(
            """
            INSERT INTO prompt_entries (id, name, text, created_at, updated_at)
            VALUES (?, ?, ?, ?, ?)
            ON CONFLICT(id) DO UPDATE SET
                name = excluded.name,
                text = excluded.text,
                updated_at = excluded.updated_at
            """,
            (
                entry["id"],
                entry["name"],
                entry["text"],
                entry["createdAt"],
                entry["updatedAt"],
            ),
        )
        conn.commit()
        return entry
    finally:
        conn.close()


def delete_prompt_entry(entry_id: str) -> None:
    conn = get_connection()
    try:
        conn.execute("DELETE FROM prompt_entries WHERE id = ?", (entry_id,))
        conn.commit()
    finally:
        conn.close()


def bulk_upsert_prompt_entries(entries: list[dict]) -> int:
    if not entries:
        return 0
    conn = get_connection()
    try:
        conn.executemany(
            """
            INSERT INTO prompt_entries (id, name, text, created_at, updated_at)
            VALUES (?, ?, ?, ?, ?)
            ON CONFLICT(id) DO UPDATE SET
                name = excluded.name,
                text = excluded.text,
                updated_at = excluded.updated_at
            """,
            [(e["id"], e["name"], e["text"], e["createdAt"], e["updatedAt"]) for e in entries],
        )
        conn.commit()
        return len(entries)
    finally:
        conn.close()


def _prompt_list_from_row(row: sqlite3.Row) -> dict:
    return {
        "id": row["id"],
        "name": row["name"],
        "items": json.loads(row["items_json"]),
        "createdAt": row["created_at"],
        "updatedAt": row["updated_at"],
    }


def list_prompt_lists_db() -> list[dict]:
    conn = get_connection()
    try:
        rows = conn.execute("SELECT * FROM prompt_lists ORDER BY created_at DESC").fetchall()
        return [_prompt_list_from_row(r) for r in rows]
    finally:
        conn.close()


def upsert_prompt_list(lst: dict) -> dict:
    conn = get_connection()
    try:
        conn.execute(
            """
            INSERT INTO prompt_lists (id, name, items_json, created_at, updated_at)
            VALUES (?, ?, ?, ?, ?)
            ON CONFLICT(id) DO UPDATE SET
                name = excluded.name,
                items_json = excluded.items_json,
                updated_at = excluded.updated_at
            """,
            (
                lst["id"],
                lst["name"],
                json.dumps(lst["items"]),
                lst["createdAt"],
                lst["updatedAt"],
            ),
        )
        conn.commit()
        return lst
    finally:
        conn.close()


def delete_prompt_list_db(list_id: str) -> None:
    conn = get_connection()
    try:
        conn.execute("DELETE FROM prompt_lists WHERE id = ?", (list_id,))
        conn.commit()
    finally:
        conn.close()


def bulk_upsert_prompt_lists(lists: list[dict]) -> int:
    if not lists:
        return 0
    conn = get_connection()
    try:
        conn.executemany(
            """
            INSERT INTO prompt_lists (id, name, items_json, created_at, updated_at)
            VALUES (?, ?, ?, ?, ?)
            ON CONFLICT(id) DO UPDATE SET
                name = excluded.name,
                items_json = excluded.items_json,
                updated_at = excluded.updated_at
            """,
            [
                (
                    lst["id"],
                    lst["name"],
                    json.dumps(lst["items"]),
                    lst["createdAt"],
                    lst["updatedAt"],
                )
                for lst in lists
            ],
        )
        conn.commit()
        return len(lists)
    finally:
        conn.close()


def get_connection() -> sqlite3.Connection:
    """Open studio.db with WAL mode, create tables once per process, enable foreign keys."""
    global _schema_ready
    db_path = studio_db_path()
    ensure_dir(db_path.parent)
    conn = sqlite3.connect(str(db_path))
    conn.row_factory = sqlite3.Row
    # foreign_keys is session-scoped; set per connection
    conn.execute("PRAGMA foreign_keys=ON")
    if not _schema_ready:
        with _schema_lock:
            if not _schema_ready:
                try:
                    _ensure_schema(conn)
                    conn.commit()
                    _schema_ready = True
                except Exception:
                    conn.close()
                    raise
    return conn


def create_run(
    id: str,
    model_name: str,
    dataset_name: str,
    config_json: str,
    started_at: str,
    total_steps: Optional[int],
    *,
    output_dir: Optional[str] = None,
    cancel_requested: bool = False,
    resumed_from_run_id: Optional[str] = None,
) -> None:
    conn = get_connection()
    try:
        conn.execute(
            """
            INSERT INTO training_runs (
                id, model_name, dataset_name, config_json, started_at, total_steps,
                output_dir, resume_blocked
            )
            VALUES (?, ?, ?, ?, ?, ?, ?, ?)
            """,
            (
                id,
                model_name,
                dataset_name,
                config_json,
                started_at,
                total_steps,
                None if cancel_requested else output_dir,
                int(cancel_requested),
            ),
        )
        if resumed_from_run_id:
            claimed = conn.execute(
                """
                UPDATE training_runs SET resume_blocked = 1
                WHERE id = ? AND status IN ('stopped', 'error')
                  AND output_dir = ? AND resume_blocked = 0
                """,
                (resumed_from_run_id, output_dir),
            )
            if claimed.rowcount != 1:
                raise RuntimeError("Resume source is no longer available")
        conn.commit()
    finally:
        conn.close()


def update_run_total_steps(id: str, total_steps: int) -> None:
    conn = get_connection()
    try:
        conn.execute(
            "UPDATE training_runs SET total_steps = ? WHERE id = ?",
            (total_steps, id),
        )
        conn.commit()
    finally:
        conn.close()


def update_run_progress(
    id: str, step: int, loss: Optional[float], duration_seconds: Optional[float]
) -> None:
    """Update current progress on a running training run (called on each metric flush)."""
    conn = get_connection()
    try:
        conn.execute(
            "UPDATE training_runs SET final_step = ?, final_loss = ?, duration_seconds = ? WHERE id = ?",
            (step, loss, duration_seconds, id),
        )
        conn.commit()
    finally:
        conn.close()


def finish_run(
    id: str,
    status: str,
    ended_at: str,
    final_step: Optional[int],
    final_loss: Optional[float],
    duration_seconds: Optional[float],
    loss_sparkline: Optional[str] = None,
    output_dir: Optional[str] = None,
    error_message: Optional[str] = None,
    clear_output_dir: bool = False,
    resume_blocked: bool = False,
) -> None:
    conn = get_connection()
    try:
        conn.execute(
            """
            UPDATE training_runs
            SET status = ?, ended_at = ?, final_step = ?, final_loss = ?,
                duration_seconds = ?, loss_sparkline = ?,
                output_dir = CASE
                    WHEN resume_blocked = 1 OR ? = 1 THEN NULL
                    WHEN ? IS NOT NULL THEN ?
                    WHEN ? IN ('error', 'stopped') THEN output_dir
                    ELSE NULL
                END,
                error_message = ?,
                resume_blocked = CASE WHEN resume_blocked = 1 OR ? = 1 THEN 1 ELSE ? END
            WHERE id = ? AND status = 'running'
            """,
            (
                status,
                ended_at,
                final_step,
                final_loss,
                duration_seconds,
                loss_sparkline,
                int(clear_output_dir),
                output_dir,
                output_dir,
                status,
                error_message,
                int(clear_output_dir),
                int(resume_blocked),
                id,
            ),
        )
        conn.commit()
    finally:
        conn.close()


def insert_metrics_batch(run_id: str, metrics: list[dict]) -> None:
    if not metrics:
        return
    conn = get_connection()
    try:
        conn.executemany(
            """
            INSERT INTO training_metrics
                (run_id, step, loss, learning_rate, grad_norm, eval_loss, epoch, num_tokens, elapsed_seconds)
            VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)
            ON CONFLICT(run_id, step) DO UPDATE SET
                loss = COALESCE(excluded.loss, loss),
                learning_rate = COALESCE(excluded.learning_rate, learning_rate),
                grad_norm = COALESCE(excluded.grad_norm, grad_norm),
                eval_loss = COALESCE(excluded.eval_loss, eval_loss),
                epoch = COALESCE(excluded.epoch, epoch),
                num_tokens = COALESCE(excluded.num_tokens, num_tokens),
                elapsed_seconds = COALESCE(excluded.elapsed_seconds, elapsed_seconds)
            """,
            [
                (
                    run_id,
                    m.get("step"),
                    m.get("loss"),
                    m.get("learning_rate"),
                    m.get("grad_norm"),
                    m.get("eval_loss"),
                    m.get("epoch"),
                    m.get("num_tokens"),
                    m.get("elapsed_seconds"),
                )
                for m in metrics
            ],
        )
        conn.commit()
    finally:
        conn.close()


def update_run_display_name(id: str, display_name: Optional[str]) -> None:
    conn = get_connection()
    try:
        conn.execute(
            "UPDATE training_runs SET display_name = ? WHERE id = ?",
            (display_name, id),
        )
        conn.commit()
    finally:
        conn.close()


def update_run_output_dir(id: str, output_dir: Optional[str]) -> None:
    conn = get_connection()
    try:
        conn.execute(
            """
            UPDATE training_runs SET output_dir = ?
            WHERE id = ? AND status = 'running' AND resume_blocked = 0
            """,
            (output_dir, id),
        )
        conn.commit()
    finally:
        conn.close()


def mark_run_cancel_requested(id: str) -> bool:
    """Clear resume/export state only while the exact run is still active."""
    conn = get_connection()
    try:
        cursor = conn.execute(
            """
            UPDATE training_runs SET output_dir = NULL, resume_blocked = 1
            WHERE id = ? AND status = 'running'
            """,
            (id,),
        )
        conn.commit()
        return cursor.rowcount > 0
    finally:
        conn.close()


def list_runs(limit: int = 50, offset: int = 0) -> dict:
    conn = get_connection()
    try:
        total = conn.execute("SELECT COUNT(*) FROM training_runs").fetchone()[0]
        rows = conn.execute(
            """
            SELECT r.id, r.status, r.model_name, r.dataset_name, r.started_at,
                   r.ended_at, r.total_steps, r.final_step, r.final_loss,
                   r.output_dir, r.duration_seconds, r.error_message,
                   r.loss_sparkline, r.display_name, r.config_json, r.resume_blocked,
                   CASE
                       WHEN r.status IN ('stopped', 'error')
                            AND r.output_dir IS NOT NULL
                            AND EXISTS (
                                SELECT 1
                                FROM training_runs newer
                                WHERE newer.output_dir = r.output_dir
                                  AND newer.status IN ('stopped', 'completed', 'error', 'running')
                                  AND newer.started_at > r.started_at
                            )
                       THEN 1 ELSE 0
                   END AS resumed_later
            FROM training_runs r
            ORDER BY started_at DESC
            LIMIT ? OFFSET ?
            """,
            (limit, offset),
        ).fetchall()
        runs = []
        for row in rows:
            run = dict(row)
            run["project_name"] = _extract_project_name_from_config_json(run.get("config_json"))
            sparkline = run.get("loss_sparkline")
            if sparkline:
                try:
                    run["loss_sparkline"] = json.loads(sparkline)
                except (json.JSONDecodeError, TypeError):
                    logger.debug("Failed to parse loss_sparkline for run %s", run.get("id"))
                    run["loss_sparkline"] = None
            runs.append(run)
        return {"runs": runs, "total": total}
    finally:
        conn.close()


def get_run(id: str) -> Optional[dict]:
    conn = get_connection()
    try:
        row = conn.execute(
            """
            SELECT r.*,
                   CASE
                       WHEN r.status IN ('stopped', 'error')
                            AND r.output_dir IS NOT NULL
                            AND EXISTS (
                                SELECT 1
                                FROM training_runs newer
                                WHERE newer.output_dir = r.output_dir
                                  AND newer.status IN ('stopped', 'completed', 'error', 'running')
                                  AND newer.started_at > r.started_at
                            )
                       THEN 1 ELSE 0
                   END AS resumed_later
            FROM training_runs r
            WHERE r.id = ?
            """,
            (id,),
        ).fetchone()
        if row is None:
            return None
        run = dict(row)
        run["project_name"] = _extract_project_name_from_config_json(run.get("config_json"))
        sparkline = run.get("loss_sparkline")
        if sparkline:
            try:
                run["loss_sparkline"] = json.loads(sparkline)
            except (json.JSONDecodeError, TypeError):
                logger.debug("Failed to parse loss_sparkline for run %s", id)
                run["loss_sparkline"] = None
        return run
    finally:
        conn.close()


def get_resumable_run_by_output_dir(output_dir: str) -> Optional[dict]:
    conn = get_connection()
    try:
        row = conn.execute(
            """
            SELECT r.*,
                   0 AS resumed_later
            FROM training_runs r
            WHERE r.output_dir = ?
              AND r.status IN ('stopped', 'error')
              AND NOT EXISTS (
                  SELECT 1
                  FROM training_runs newer
                  WHERE newer.output_dir = r.output_dir
                    AND newer.status IN ('stopped', 'completed', 'error', 'running')
                    AND newer.started_at > r.started_at
              )
            ORDER BY r.started_at DESC
            LIMIT 1
            """,
            (output_dir,),
        ).fetchone()
        if row is None:
            return None
        run = dict(row)
        sparkline = run.get("loss_sparkline")
        if sparkline:
            try:
                run["loss_sparkline"] = json.loads(sparkline)
            except (json.JSONDecodeError, TypeError):
                logger.debug("Failed to parse loss_sparkline for output_dir %s", output_dir)
                run["loss_sparkline"] = None
        return run
    finally:
        conn.close()


def get_run_metrics(id: str) -> dict:
    """Return metric arrays for a run, using paired step arrays per metric."""
    conn = get_connection()
    try:
        rows = conn.execute(
            """
            SELECT step, loss, learning_rate, grad_norm, eval_loss, epoch,
                   num_tokens, elapsed_seconds
            FROM training_metrics
            WHERE run_id = ?
            ORDER BY step
            """,
            (id,),
        ).fetchall()

        step_history: list[int] = []
        loss_history: list[float] = []
        loss_step_history: list[int] = []
        lr_history: list[float] = []
        lr_step_history: list[int] = []
        grad_norm_history: list[float] = []
        grad_norm_step_history: list[int] = []
        eval_loss_history: list[float] = []
        eval_step_history: list[int] = []
        final_epoch: float | None = None
        final_num_tokens: int | None = None

        for row in rows:
            step = row["step"]
            step_history.append(step)
            if step > 0 and row["loss"] is not None:
                loss_history.append(row["loss"])
                loss_step_history.append(step)
            if step > 0 and row["learning_rate"] is not None:
                lr_history.append(row["learning_rate"])
                lr_step_history.append(step)
            if step > 0 and row["grad_norm"] is not None:
                grad_norm_history.append(row["grad_norm"])
                grad_norm_step_history.append(step)
            if step > 0 and row["eval_loss"] is not None:
                eval_loss_history.append(row["eval_loss"])
                eval_step_history.append(step)
            if row["epoch"] is not None:
                final_epoch = row["epoch"]
            if row["num_tokens"] is not None:
                final_num_tokens = row["num_tokens"]

        return {
            "step_history": step_history,
            "loss_history": loss_history,
            "loss_step_history": loss_step_history,
            "lr_history": lr_history,
            "lr_step_history": lr_step_history,
            "grad_norm_history": grad_norm_history,
            "grad_norm_step_history": grad_norm_step_history,
            "eval_loss_history": eval_loss_history,
            "eval_step_history": eval_step_history,
            "final_epoch": final_epoch,
            "final_num_tokens": final_num_tokens,
        }
    finally:
        conn.close()


def delete_run(id: str) -> None:
    conn = get_connection()
    try:
        conn.execute("DELETE FROM training_runs WHERE id = ?", (id,))
        conn.commit()
    finally:
        conn.close()


def cleanup_orphaned_runs() -> None:
    """Mark any 'running' rows as errored on startup (server restarted mid-training)."""
    conn = get_connection()
    try:
        conn.execute(
            """
            UPDATE training_runs
            SET status = CASE WHEN resume_blocked = 1 THEN 'stopped' ELSE 'error' END,
                error_message = CASE
                    WHEN resume_blocked = 1 THEN NULL
                    ELSE 'Server restarted during training'
                END,
                output_dir = CASE WHEN resume_blocked = 1 THEN NULL ELSE output_dir END,
                ended_at = ?
            WHERE status = 'running'
            """,
            (datetime.now(timezone.utc).isoformat(),),
        )
        conn.commit()
    finally:
        conn.close()


def list_scan_folders() -> list[dict]:
    conn = get_connection()
    try:
        rows = conn.execute(
            "SELECT id, path, created_at FROM scan_folders ORDER BY created_at"
        ).fetchall()
        return [dict(row) for row in rows]
    finally:
        conn.close()


def add_scan_folder(path: str) -> dict:
    """Add a directory to the custom scan folder list. Returns the row."""
    if not path or not path.strip():
        raise ValueError("Path cannot be empty")
    normalized = os.path.realpath(os.path.expanduser(path.strip()))

    # Validate the path is an existing, readable directory before persisting.
    if not os.path.exists(normalized):
        raise ValueError("Path does not exist")
    if not os.path.isdir(normalized):
        raise ValueError("Path must be a directory, not a file")
    if not os.access(normalized, os.R_OK | os.X_OK):
        raise ValueError("Path is not readable")
    # Reject a local filesystem root ("/", or a bare Windows drive root "C:\\"):
    # registering one seeds the browse allowlist with a root above denied system
    # dirs. A UNC share root (\\server\share) has none under it and was
    # registerable before this guard, so it stays allowed. Mirrors scan_folders.py.
    if is_local_filesystem_root(normalized):
        raise ValueError("The filesystem root cannot be registered")
    if _contains_sensitive_path_component(normalized):
        raise ValueError("Credential or configuration directories are not allowed")

    # Windows: normcase for the denylist check but store original casing
    # so consumers see the native drive-letter casing (e.g. C:\Models).
    is_win = platform.system() == "Windows"
    check = os.path.normcase(normalized) if is_win else normalized
    for prefix in _denied_path_prefixes():
        if check == prefix or check.startswith(prefix + os.sep):
            if prefix == "/run" and is_linux_run_media_path(check):
                continue
            raise ValueError(f"Path under {prefix} is not allowed")

    conn = get_connection()
    try:
        now = datetime.now(timezone.utc).isoformat()
        # Windows: case-insensitive lookup so C:\Models and c:\models dedup.
        if is_win:
            existing = conn.execute(
                "SELECT id, path, created_at FROM scan_folders WHERE path = ? COLLATE NOCASE",
                (normalized,),
            ).fetchone()
        else:
            existing = conn.execute(
                "SELECT id, path, created_at FROM scan_folders WHERE path = ?",
                (normalized,),
            ).fetchone()
        if existing is not None:
            return dict(existing)
        try:
            conn.execute(
                "INSERT INTO scan_folders (path, created_at) VALUES (?, ?)",
                (normalized, now),
            )
            conn.commit()
        except sqlite3.IntegrityError:
            pass  # duplicate; fall through to SELECT
        # Same collation as the pre-check to catch concurrent writes (Windows).
        fallback_sql = (
            "SELECT id, path, created_at FROM scan_folders WHERE path = ? COLLATE NOCASE"
            if is_win
            else "SELECT id, path, created_at FROM scan_folders WHERE path = ?"
        )
        row = conn.execute(fallback_sql, (normalized,)).fetchone()
        if row is None:
            raise ValueError("Folder was concurrently removed")
        return dict(row)
    finally:
        conn.close()


def remove_scan_folder(id: int) -> None:
    conn = get_connection()
    try:
        conn.execute("DELETE FROM scan_folders WHERE id = ?", (id,))
        conn.commit()
    finally:
        conn.close()


def _json_loads(value: str | None, fallback):
    if value is None:
        return fallback
    try:
        return json.loads(value)
    except (json.JSONDecodeError, TypeError):
        return fallback


def _chat_thread_from_row(row: sqlite3.Row) -> dict:
    data = dict(row)
    return {
        "id": data["id"],
        "title": data["title"],
        "modelType": data["model_type"],
        "modelId": data.get("model_id") or "",
        "pairId": data.get("pair_id") or None,
        "projectId": data.get("project_id") or None,
        "archived": bool(data["archived"]),
        "createdAt": data["created_at"],
        "updatedAt": data.get("updated_at")
        if data.get("updated_at") is not None
        else data["created_at"],
        "openaiCodeExecContainerId": data.get("openai_code_exec_container_id"),
        "anthropicCodeExecContainerId": data.get("anthropic_code_exec_container_id"),
        "forkedFromThreadId": data.get("forked_from_thread_id"),
        "forkedFromMessageId": data.get("forked_from_message_id"),
    }


def _chat_project_from_row(row: sqlite3.Row) -> dict:
    data = dict(row)
    root_path = data.get("root_path")
    return {
        "id": data["id"],
        "name": data["name"],
        "instructions": data.get("instructions") or "",
        "rootPath": root_path or None,
        "sandboxPath": os.path.join(root_path, "sandbox") if root_path else None,
        "archived": bool(data["archived"]),
        "createdAt": data["created_at"],
        "updatedAt": data["updated_at"],
    }


def _chat_message_from_row(row: sqlite3.Row) -> dict:
    data = dict(row)
    message = {
        "id": data["id"],
        "threadId": data["thread_id"],
        "parentId": data.get("parent_id"),
        "role": data["role"],
        "content": _json_loads(data.get("content_json"), []),
        "createdAt": data["created_at"],
    }
    attachments = _json_loads(data.get("attachments_json"), None)
    metadata = _json_loads(data.get("metadata_json"), None)
    if attachments is not None:
        message["attachments"] = attachments
    if metadata is not None:
        message["metadata"] = metadata
    return message


def upsert_chat_thread(thread: dict) -> dict:
    conn = get_connection()
    try:
        conn.execute(
            """
            INSERT INTO chat_threads
                (id, title, model_type, model_id, pair_id, project_id, archived, created_at, updated_at, openai_code_exec_container_id, anthropic_code_exec_container_id, forked_from_thread_id, forked_from_message_id)
            VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
            ON CONFLICT(id) DO UPDATE SET
                title = excluded.title,
                model_type = excluded.model_type,
                model_id = excluded.model_id,
                pair_id = excluded.pair_id,
                project_id = excluded.project_id,
                archived = excluded.archived,
                created_at = excluded.created_at,
                updated_at = COALESCE(excluded.updated_at, chat_threads.updated_at),
                openai_code_exec_container_id = excluded.openai_code_exec_container_id,
                anthropic_code_exec_container_id = excluded.anthropic_code_exec_container_id,
                forked_from_thread_id = excluded.forked_from_thread_id,
                forked_from_message_id = excluded.forked_from_message_id
            """,
            (
                thread["id"],
                thread.get("title") or "New Chat",
                thread["modelType"],
                thread.get("modelId") or "",
                thread.get("pairId"),
                thread.get("projectId"),
                1 if thread.get("archived") else 0,
                int(thread["createdAt"]),
                int(thread["updatedAt"]) if thread.get("updatedAt") is not None else None,
                thread.get("openaiCodeExecContainerId"),
                thread.get("anthropicCodeExecContainerId"),
                thread.get("forkedFromThreadId"),
                thread.get("forkedFromMessageId"),
            ),
        )
        conn.commit()
        return get_chat_thread(thread["id"]) or thread
    finally:
        conn.close()


def update_chat_thread(id: str, patch: dict) -> Optional[dict]:
    allowed = {
        "title": ("title", patch.get("title")),
        "modelType": ("model_type", patch.get("modelType")),
        "modelId": ("model_id", patch.get("modelId")),
        "pairId": ("pair_id", patch.get("pairId")),
        "projectId": ("project_id", patch.get("projectId")),
        "archived": ("archived", 1 if patch.get("archived") else 0),
        "createdAt": ("created_at", patch.get("createdAt")),
        "updatedAt": ("updated_at", patch.get("updatedAt")),
        "openaiCodeExecContainerId": (
            "openai_code_exec_container_id",
            patch.get("openaiCodeExecContainerId"),
        ),
        "anthropicCodeExecContainerId": (
            "anthropic_code_exec_container_id",
            patch.get("anthropicCodeExecContainerId"),
        ),
        "forkedFromThreadId": (
            "forked_from_thread_id",
            patch.get("forkedFromThreadId"),
        ),
        "forkedFromMessageId": (
            "forked_from_message_id",
            patch.get("forkedFromMessageId"),
        ),
    }
    assignments = []
    values = []
    for key, (column, value) in allowed.items():
        if key in patch:
            assignments.append(f"{column} = ?")
            values.append(value)
    if not assignments:
        return get_chat_thread(id)

    conn = get_connection()
    try:
        conn.execute(
            f"UPDATE chat_threads SET {', '.join(assignments)} WHERE id = ?",
            (*values, id),
        )
        conn.commit()
        row = conn.execute("SELECT * FROM chat_threads WHERE id = ?", (id,)).fetchone()
        return _chat_thread_from_row(row) if row is not None else None
    finally:
        conn.close()


def get_chat_thread(id: str) -> Optional[dict]:
    conn = get_connection()
    try:
        row = conn.execute("SELECT * FROM chat_threads WHERE id = ?", (id,)).fetchone()
        return _chat_thread_from_row(row) if row is not None else None
    finally:
        conn.close()


def list_chat_threads(
    model_type: str | None = None,
    pair_id: str | None = None,
    project_id: str | None = None,
    include_archived: bool = True,
) -> list[dict]:
    clauses = []
    values: list[object] = []
    if model_type is not None:
        clauses.append("model_type = ?")
        values.append(model_type)
    if pair_id is not None:
        clauses.append("pair_id = ?")
        values.append(pair_id)
    if project_id is not None:
        clauses.append("project_id = ?")
        values.append(project_id)
    if not include_archived:
        clauses.append("archived = 0")
    where = f"WHERE {' AND '.join(clauses)}" if clauses else ""
    conn = get_connection()
    try:
        rows = conn.execute(
            f"SELECT * FROM chat_threads {where} "
            "ORDER BY COALESCE(updated_at, created_at) DESC, created_at DESC",
            values,
        ).fetchall()
        return [_chat_thread_from_row(row) for row in rows]
    finally:
        conn.close()


def delete_chat_threads(ids: list[str]) -> None:
    if not ids:
        return
    conn = get_connection()
    try:
        conn.execute("BEGIN IMMEDIATE")
        _ensure_chat_attachment_inventory_current(conn)
        conn.executemany(
            "DELETE FROM chat_attachment_tombstones WHERE thread_id = ?",
            [(id,) for id in ids],
        )
        conn.executemany("DELETE FROM chat_threads WHERE id = ?", [(id,) for id in ids])
        _mark_chat_attachment_inventory_clean(conn)
        conn.commit()
    finally:
        conn.close()


def clear_chat_history() -> None:
    conn = get_connection()
    try:
        conn.execute("BEGIN IMMEDIATE")
        _ensure_chat_attachment_inventory_current(conn)
        conn.execute("DELETE FROM chat_attachment_tombstones")
        conn.execute("DELETE FROM chat_threads")
        _mark_chat_attachment_inventory_clean(conn)
        conn.commit()
    finally:
        conn.close()


def count_chat_threads() -> int:
    conn = get_connection()
    try:
        return int(conn.execute("SELECT COUNT(*) FROM chat_threads").fetchone()[0])
    finally:
        conn.close()


def upsert_chat_project(project: dict) -> dict:
    existing = get_chat_project(project["id"])
    root_path = existing.get("rootPath") if existing else None
    if not root_path:
        root_path = _default_project_root(project)
    root_path = _ensure_project_workspace(root_path)
    conn = get_connection()
    try:
        conn.execute(
            """
            INSERT INTO chat_projects
                (id, name, instructions, root_path, archived, created_at, updated_at)
            VALUES (?, ?, ?, ?, ?, ?, ?)
            ON CONFLICT(id) DO UPDATE SET
                name = excluded.name,
                instructions = excluded.instructions,
                root_path = COALESCE(chat_projects.root_path, excluded.root_path),
                archived = excluded.archived,
                created_at = excluded.created_at,
                updated_at = excluded.updated_at
            """,
            (
                project["id"],
                project["name"],
                project.get("instructions") or "",
                root_path,
                1 if project.get("archived") else 0,
                int(project["createdAt"]),
                int(project["updatedAt"]),
            ),
        )
        conn.commit()
        return get_chat_project(project["id"]) or project
    finally:
        conn.close()


def update_chat_project(id: str, patch: dict) -> Optional[dict]:
    allowed = {
        "name": ("name", patch.get("name")),
        "instructions": ("instructions", patch.get("instructions")),
        "archived": ("archived", 1 if patch.get("archived") else 0),
        "createdAt": ("created_at", patch.get("createdAt")),
        "updatedAt": ("updated_at", patch.get("updatedAt")),
    }
    assignments = []
    values = []
    for key, (column, value) in allowed.items():
        if key in patch:
            assignments.append(f"{column} = ?")
            values.append(value)
    if not assignments:
        return get_chat_project(id)

    conn = get_connection()
    try:
        conn.execute(
            f"UPDATE chat_projects SET {', '.join(assignments)} WHERE id = ?",
            (*values, id),
        )
        conn.commit()
        row = conn.execute("SELECT * FROM chat_projects WHERE id = ?", (id,)).fetchone()
        return _chat_project_from_row(row) if row is not None else None
    finally:
        conn.close()


def ensure_chat_project_workspace(id: str) -> Optional[dict]:
    project = get_chat_project(id)
    if project is None:
        return None
    root_path = project.get("rootPath") or _default_project_root(project)
    root_path = _ensure_project_workspace(root_path)
    if project.get("rootPath") == root_path:
        return project
    conn = get_connection()
    try:
        conn.execute(
            "UPDATE chat_projects SET root_path = ? WHERE id = ?",
            (root_path, id),
        )
        conn.commit()
    finally:
        conn.close()
    return get_chat_project(id)


def get_chat_project(id: str) -> Optional[dict]:
    conn = get_connection()
    try:
        row = conn.execute("SELECT * FROM chat_projects WHERE id = ?", (id,)).fetchone()
        return _chat_project_from_row(row) if row is not None else None
    finally:
        conn.close()


def list_chat_projects(include_archived: bool = False) -> list[dict]:
    conn = get_connection()
    try:
        where = "" if include_archived else "WHERE archived = 0"
        rows = conn.execute(
            f"SELECT * FROM chat_projects {where} ORDER BY updated_at DESC"
        ).fetchall()
        return [_chat_project_from_row(row) for row in rows]
    finally:
        conn.close()


def delete_chat_project(id: str, delete_files: bool = False) -> Optional[dict]:
    conn = get_connection()
    try:
        conn.execute("BEGIN IMMEDIATE")
        _ensure_chat_attachment_inventory_current(conn)
        row = conn.execute("SELECT * FROM chat_projects WHERE id = ?", (id,)).fetchone()
        if row is None:
            conn.rollback()
            return None
        project = _chat_project_from_row(row)
        conn.execute("DELETE FROM chat_threads WHERE project_id = ?", (id,))
        conn.execute("DELETE FROM chat_projects WHERE id = ?", (id,))
        _mark_chat_attachment_inventory_clean(conn)
        conn.commit()
        if delete_files:
            _delete_project_workspace(project)
        return project
    except Exception:
        conn.rollback()
        raise
    finally:
        conn.close()


class ChatMessageConflictError(RuntimeError):
    """Raised when a chat message id already belongs to another thread."""


class ChatMessageProtectedError(RuntimeError):
    """Raised when pruning would remove a message owned by a durable feature."""


class CorruptSettingsError(RuntimeError):
    """Raised when a partial settings patch would overwrite corrupt settings."""


def _parse_chat_setting_json(key: str, value_json: str) -> tuple[bool, Any]:
    try:
        return True, json.loads(value_json)
    except (json.JSONDecodeError, TypeError) as exc:
        logger.warning(
            "Corrupt chat_settings JSON; quarantining key=%s error=%s",
            key,
            exc,
        )
        return False, None


def _load_chat_settings_for_merge(conn: sqlite3.Connection) -> tuple[dict[str, Any], set[str]]:
    rows = conn.execute("SELECT key, value_json FROM chat_settings").fetchall()
    current: dict[str, Any] = {}
    corrupt: set[str] = set()
    now = datetime.now(timezone.utc).isoformat()
    for row in rows:
        ok, value = _parse_chat_setting_json(row["key"], row["value_json"])
        if ok:
            current[row["key"]] = value
            continue
        corrupt.add(row["key"])
        conn.execute(
            """
            INSERT INTO chat_settings_quarantine
                (key, value_json, reason, quarantined_at)
            VALUES (?, ?, ?, ?)
            """,
            (row["key"], row["value_json"], "json_decode_error", now),
        )
        conn.execute(
            "DELETE FROM chat_settings WHERE key = ? AND value_json = ?",
            (row["key"], row["value_json"]),
        )
    return current, corrupt


def _raise_if_chat_message_thread_conflicts(
    conn: sqlite3.Connection, thread_id: str, message_ids: list[str]
) -> None:
    unique_ids = list(dict.fromkeys(message_ids))
    if not unique_ids:
        return
    conflicts: list[str] = []
    for start in range(0, len(unique_ids), _SQLITE_IN_CHUNK_SIZE):
        chunk = unique_ids[start : start + _SQLITE_IN_CHUNK_SIZE]
        placeholders = ",".join("?" for _ in chunk)
        rows = conn.execute(
            f"""
            SELECT id FROM chat_messages
            WHERE id IN ({placeholders}) AND thread_id != ?
            ORDER BY id
            """,
            (*chunk, thread_id),
        ).fetchall()
        conflicts.extend(row["id"] for row in rows)
    if conflicts:
        preview = ", ".join(conflicts[:5])
        suffix = "" if len(conflicts) <= 5 else f" (+{len(conflicts) - 5} more)"
        raise ChatMessageConflictError(
            f"Message id already belongs to another thread: {preview}{suffix}"
        )


def _bump_chat_thread_updated_at(
    conn: sqlite3.Connection, thread_id: str, message_created_at: int
) -> None:
    conn.execute(
        """
        UPDATE chat_threads
        SET updated_at = MAX(COALESCE(updated_at, created_at), ?)
        WHERE id = ?
        """,
        (message_created_at, thread_id),
    )


def _recompute_chat_thread_updated_at(conn: sqlite3.Connection, thread_id: str) -> None:
    """Set updated_at from the remaining messages, floored at created_at.

    Unlike the ratchet-only bump, this can lower updated_at -- needed after
    pruning, which may delete the thread's newest message.
    """
    conn.execute(
        """
        UPDATE chat_threads
        SET updated_at = MAX(
            COALESCE(
                (
                    SELECT MAX(m.created_at) FROM chat_messages m
                    WHERE m.thread_id = chat_threads.id
                ),
                created_at
            ),
            created_at
        )
        WHERE id = ?
        """,
        (thread_id,),
    )


def _research_message_ids(conn: sqlite3.Connection, thread_id: str) -> set[str]:
    return {
        str(message_id)
        for row in conn.execute(
            "SELECT user_message_id, assistant_message_id FROM research_runs WHERE thread_id = ?",
            (thread_id,),
        ).fetchall()
        for message_id in row
        if message_id is not None
    }


def _research_message_would_change(conn: sqlite3.Connection, thread_id: str, message: dict) -> bool:
    row = conn.execute(
        "SELECT parent_id, role, content_json, metadata_json, attachments_json, created_at "
        "FROM chat_messages WHERE thread_id = ? AND id = ?",
        (thread_id, str(message["id"])),
    ).fetchone()
    if row is None:
        return False

    def canon(value: object) -> str | None:
        return json.dumps(value, sort_keys = True) if value is not None else None

    # created_at is compared too: without it a client could re-upsert a protected message with an
    # unchanged body but a different timestamp and silently reorder the server-managed research
    # prompt/response pair. Absent createdAt defaults to the stored value (a no-op re-sync).
    return (
        canon(message.get("content", [])) != canon(json.loads(row["content_json"] or "[]"))
        or canon(message.get("metadata"))
        != canon(json.loads(row["metadata_json"]) if row["metadata_json"] else None)
        or canon(message.get("attachments"))
        != canon(json.loads(row["attachments_json"]) if row["attachments_json"] else None)
        or (message.get("parentId") or None) != (row["parent_id"] or None)
        or str(message.get("role")) != str(row["role"])
        or int(message.get("createdAt", row["created_at"])) != int(row["created_at"])
    )


def _guard_research_messages(
    conn: sqlite3.Connection, thread_id: str, messages: list[dict]
) -> None:
    protected = _research_message_ids(conn, thread_id)
    if not protected:
        return
    for message in messages:
        if str(message["id"]) in protected and _research_message_would_change(
            conn, thread_id, message
        ):
            raise ChatMessageProtectedError(
                "Research prompts and responses are server-managed and cannot be edited"
            )


_CONTENT_PART_ID_PREFIX = "content-part-sha256-"
_URI_SCHEME_RE = re.compile(r"^[A-Za-z][A-Za-z0-9+.-]*:")


def _is_locally_stored_blob(value: str) -> bool:
    """True for data URIs or bare base64, never external/blob URI references."""
    candidate = value.lstrip()
    if not candidate:
        return False
    if candidate[:5].lower() == "data:":
        return True
    if candidate.startswith(("//", "\\\\")):
        return False
    return _URI_SCHEME_RE.match(candidate) is None


def _managed_content_part_payload(part: dict) -> Optional[tuple[str, Any]]:
    """Return the locally stored blob payload used to identify a content part."""
    image = part.get("image")
    if isinstance(image, str) and image[:5].lower() == "data:":
        return "image", image

    audio = part.get("audio")
    if isinstance(audio, str) and _is_locally_stored_blob(audio):
        return "audio", audio
    if isinstance(audio, dict):
        data = audio.get("data")
        if isinstance(data, str) and _is_locally_stored_blob(data):
            return "audio", audio
    return None


def _content_part_id(part: dict) -> Optional[str]:
    """Stable managed id derived from blob data, without mutating inference content."""
    payload = _managed_content_part_payload(part)
    if payload is None:
        return None
    canonical = json.dumps(
        payload,
        ensure_ascii = False,
        separators = (",", ":"),
        sort_keys = True,
    ).encode("utf-8")
    return f"{_CONTENT_PART_ID_PREFIX}{hashlib.sha256(canonical).hexdigest()}"


def _chat_attachment_tombstones_for_messages(
    conn: sqlite3.Connection, thread_id: str, message_ids: list[str]
) -> dict[str, set[str]]:
    tombstones = {message_id: set() for message_id in message_ids}
    unique_ids = list(dict.fromkeys(message_ids))
    for start in range(0, len(unique_ids), _SQLITE_IN_CHUNK_SIZE):
        chunk = unique_ids[start : start + _SQLITE_IN_CHUNK_SIZE]
        placeholders = ",".join("?" for _ in chunk)
        rows = conn.execute(
            f"""
            SELECT message_id, attachment_id
            FROM chat_attachment_tombstones
            WHERE thread_id = ? AND message_id IN ({placeholders})
            """,
            (thread_id, *chunk),
        ).fetchall()
        for row in rows:
            tombstones[row["message_id"]].add(row["attachment_id"])
    return tombstones


def _reconcile_chat_message_uploads(message: dict, tombstones: set[str]) -> dict:
    """Strip uploads previously deleted through the Data tab from a stale write."""
    if not tombstones:
        return message

    reconciled = dict(message)
    attachments = message.get("attachments")
    if isinstance(attachments, list):
        reconciled["attachments"] = [
            attachment
            for attachment in attachments
            if not (isinstance(attachment, dict) and str(attachment.get("id") or "") in tombstones)
        ]

    content = message.get("content")
    if isinstance(content, list):
        reconciled["content"] = [
            part
            for part in content
            if not (isinstance(part, dict) and (_content_part_id(part) or "") in tombstones)
        ]
    return reconciled


def _chat_attachment_metadata_text(value, fallback: Optional[str] = None) -> Optional[str]:
    """Keep untyped legacy/import metadata safe for SQLite binding."""
    if value is None:
        return fallback
    if isinstance(value, str):
        return value or fallback
    if isinstance(value, (bool, int, float)):
        return str(value)
    # Objects and arrays are not useful display metadata and sqlite3 rejects
    # binding them directly.
    return fallback


def _chat_attachment_inventory_entries(
    attachments_json: Optional[str],
    content_json: Optional[str],
    tombstones: Optional[set[str]] = None,
) -> list[dict]:
    tombstones = tombstones or set()
    attachments = _json_loads(attachments_json, None)
    if not isinstance(attachments, list):
        attachments = []
    attachments = [
        attachment
        for attachment in attachments
        if isinstance(attachment, dict) and attachment.get("id")
    ]
    attachments.extend(_content_part_attachments(content_json))

    entries: list[dict] = []
    seen: set[str] = set()
    for attachment in attachments:
        attachment_id = str(attachment["id"])
        if attachment_id in seen or attachment_id in tombstones:
            continue
        seen.add(attachment_id)
        entries.append(
            {
                "id": attachment_id,
                "name": _chat_attachment_metadata_text(attachment.get("name"), "attachment"),
                "type": _chat_attachment_metadata_text(attachment.get("type")),
                "contentType": _chat_attachment_metadata_text(attachment.get("contentType")),
                "sizeBytes": _chat_attachment_size_bytes(attachment),
            }
        )
    return entries


def _replace_chat_attachment_inventory(
    conn: sqlite3.Connection,
    message_id: str,
    attachments_json: Optional[str],
    content_json: Optional[str],
    tombstones: Optional[set[str]] = None,
) -> None:
    conn.execute("DELETE FROM chat_attachment_inventory WHERE message_id = ?", (message_id,))
    entries = _chat_attachment_inventory_entries(
        attachments_json,
        content_json,
        tombstones,
    )
    conn.executemany(
        """
        INSERT INTO chat_attachment_inventory
            (message_id, attachment_id, name, type, content_type, size_bytes)
        VALUES (?, ?, ?, ?, ?, ?)
        """,
        [
            (
                message_id,
                entry["id"],
                entry["name"],
                entry["type"],
                entry["contentType"],
                entry["sizeBytes"],
            )
            for entry in entries
        ],
    )


def _mark_chat_attachment_inventory_clean(conn: sqlite3.Connection) -> None:
    conn.execute(
        """
        INSERT INTO chat_attachment_inventory_state
            (singleton, inventory_version, dirty, backfilled_at)
        VALUES (1, ?, 0, ?)
        ON CONFLICT(singleton) DO UPDATE SET
            inventory_version = excluded.inventory_version,
            dirty = 0,
            backfilled_at = excluded.backfilled_at
        """,
        (
            _CHAT_ATTACHMENT_INVENTORY_VERSION,
            int(datetime.now(timezone.utc).timestamp() * 1000),
        ),
    )


def _rebuild_chat_attachment_inventory(conn: sqlite3.Connection) -> None:
    """Rebuild after schema upgrade or a write from an older Studio build."""
    conn.execute("DELETE FROM chat_attachment_inventory")
    tombstones: dict[tuple[str, str], set[str]] = {}
    for row in conn.execute(
        "SELECT thread_id, message_id, attachment_id FROM chat_attachment_tombstones"
    ).fetchall():
        tombstones.setdefault((row["thread_id"], row["message_id"]), set()).add(
            row["attachment_id"]
        )
    rows = conn.execute(
        "SELECT id, thread_id, attachments_json, content_json FROM chat_messages"
    ).fetchall()
    for row in rows:
        _replace_chat_attachment_inventory(
            conn,
            row["id"],
            row["attachments_json"],
            row["content_json"],
            tombstones.get((row["thread_id"], row["id"]), set()),
        )


def _ensure_chat_attachment_inventory_current(conn: sqlite3.Connection) -> None:
    state = conn.execute(
        """
        SELECT inventory_version, dirty
        FROM chat_attachment_inventory_state
        WHERE singleton = 1
        """
    ).fetchone()
    if (
        state is not None
        and state["inventory_version"] == _CHAT_ATTACHMENT_INVENTORY_VERSION
        and not state["dirty"]
    ):
        return

    owns_transaction = not conn.in_transaction
    if owns_transaction:
        conn.execute("BEGIN IMMEDIATE")
    try:
        state = conn.execute(
            """
            SELECT inventory_version, dirty
            FROM chat_attachment_inventory_state
            WHERE singleton = 1
            """
        ).fetchone()
        if (
            state is None
            or state["inventory_version"] != _CHAT_ATTACHMENT_INVENTORY_VERSION
            or state["dirty"]
        ):
            _rebuild_chat_attachment_inventory(conn)
            _mark_chat_attachment_inventory_clean(conn)
        if owns_transaction:
            conn.commit()
    except Exception:
        if owns_transaction:
            conn.rollback()
        raise


def upsert_chat_message(message: dict, *, allow_research_update: bool = False) -> dict:
    conn = get_connection()
    try:
        conn.execute("BEGIN IMMEDIATE")
        _ensure_chat_attachment_inventory_current(conn)
        if not allow_research_update:
            _guard_research_messages(conn, message["threadId"], [message])
        _raise_if_chat_message_thread_conflicts(
            conn,
            message["threadId"],
            [message["id"]],
        )
        tombstones = _chat_attachment_tombstones_for_messages(
            conn,
            message["threadId"],
            [message["id"]],
        )
        reconciled = _reconcile_chat_message_uploads(
            message,
            tombstones.get(message["id"], set()),
        )
        content_json = json.dumps(reconciled.get("content", []))
        attachments_json = (
            json.dumps(reconciled.get("attachments"))
            if reconciled.get("attachments") is not None
            else None
        )
        conn.execute(
            """
            INSERT INTO chat_messages
                (id, thread_id, parent_id, role, content_json, attachments_json, metadata_json, created_at)
            VALUES (?, ?, ?, ?, ?, ?, ?, ?)
            ON CONFLICT(id) DO UPDATE SET
                parent_id = excluded.parent_id,
                role = excluded.role,
                content_json = excluded.content_json,
                attachments_json = excluded.attachments_json,
                metadata_json = excluded.metadata_json,
                created_at = excluded.created_at
            WHERE excluded.thread_id = chat_messages.thread_id
            """,
            (
                reconciled["id"],
                reconciled["threadId"],
                reconciled.get("parentId"),
                reconciled["role"],
                content_json,
                attachments_json,
                json.dumps(reconciled.get("metadata"))
                if reconciled.get("metadata") is not None
                else None,
                int(reconciled["createdAt"]),
            ),
        )
        _replace_chat_attachment_inventory(
            conn,
            reconciled["id"],
            attachments_json,
            content_json,
        )
        _bump_chat_thread_updated_at(
            conn,
            reconciled["threadId"],
            int(reconciled["createdAt"]),
        )
        _mark_chat_attachment_inventory_clean(conn)
        conn.commit()
        return reconciled
    except Exception:
        conn.rollback()
        raise
    finally:
        conn.close()


def sync_chat_messages(
    thread_id: str,
    messages: list[dict],
    prune_missing: bool = False,
    *,
    allow_research_update: bool = False,
) -> list[dict]:
    conn = get_connection()
    try:
        conn.execute("BEGIN IMMEDIATE")
        _ensure_chat_attachment_inventory_current(conn)
        if not allow_research_update:
            _guard_research_messages(conn, thread_id, messages)
        _raise_if_chat_message_thread_conflicts(
            conn,
            thread_id,
            [m["id"] for m in messages],
        )
        tombstones = _chat_attachment_tombstones_for_messages(
            conn,
            thread_id,
            [m["id"] for m in messages],
        )
        reconciled_messages = [
            _reconcile_chat_message_uploads(m, tombstones.get(m["id"], set())) for m in messages
        ]
        serialized_messages = [
            (
                m,
                json.dumps(m.get("content", [])),
                json.dumps(m.get("attachments")) if m.get("attachments") is not None else None,
            )
            for m in reconciled_messages
        ]
        conn.executemany(
            """
            INSERT INTO chat_messages
                (id, thread_id, parent_id, role, content_json, attachments_json, metadata_json, created_at)
            VALUES (?, ?, ?, ?, ?, ?, ?, ?)
            ON CONFLICT(id) DO UPDATE SET
                parent_id = excluded.parent_id,
                role = excluded.role,
                content_json = excluded.content_json,
                attachments_json = excluded.attachments_json,
                metadata_json = excluded.metadata_json,
                created_at = excluded.created_at
            WHERE excluded.thread_id = chat_messages.thread_id
            """,
            [
                (
                    m["id"],
                    thread_id,
                    m.get("parentId"),
                    m["role"],
                    content_json,
                    attachments_json,
                    json.dumps(m.get("metadata")) if m.get("metadata") is not None else None,
                    int(m["createdAt"]),
                )
                for m, content_json, attachments_json in serialized_messages
            ],
        )
        for m, content_json, attachments_json in serialized_messages:
            _replace_chat_attachment_inventory(
                conn,
                m["id"],
                attachments_json,
                content_json,
            )
        if prune_missing:
            retained_ids = {m["id"] for m in reconciled_messages}
            existing_ids = {
                row["id"]
                for row in conn.execute(
                    "SELECT id FROM chat_messages WHERE thread_id = ?",
                    (thread_id,),
                ).fetchall()
            }
            missing_ids = sorted(existing_ids - retained_ids)
            if set(missing_ids) & _research_message_ids(conn, thread_id):
                raise ChatMessageProtectedError(
                    "Research prompts and responses cannot be deleted from their original thread"
                )
            for start in range(0, len(missing_ids), _SQLITE_IN_CHUNK_SIZE):
                chunk = missing_ids[start : start + _SQLITE_IN_CHUNK_SIZE]
                placeholders = ",".join("?" for _ in chunk)
                conn.execute(
                    f"DELETE FROM chat_messages WHERE thread_id = ? AND id IN ({placeholders})",
                    (thread_id, *chunk),
                )
            _recompute_chat_thread_updated_at(conn, thread_id)
        elif reconciled_messages:
            _bump_chat_thread_updated_at(
                conn,
                thread_id,
                max(int(m["createdAt"]) for m in reconciled_messages),
            )
        _mark_chat_attachment_inventory_clean(conn)
        conn.commit()
        return list_chat_messages(thread_id)
    except (ChatMessageConflictError, ChatMessageProtectedError):
        conn.rollback()
        raise
    except sqlite3.Error:
        logger.exception("Failed to sync chat messages for thread %s", thread_id)
        conn.rollback()
        raise
    finally:
        conn.close()


_RESEARCH_LINK_KEYS = {
    "researchRunId",
    "researchRun",
    "researchStatus",
    "researchPlanRevision",
    "serverManaged",
}


def _detach_research_message_json(
    content_json: str, metadata_json: str | None
) -> tuple[str, str | None]:
    content = _json_loads(content_json, [])
    metadata = _json_loads(metadata_json, None)
    custom = metadata.get("custom") if isinstance(metadata, dict) else None
    linked = (
        isinstance(metadata, dict)
        and any(key in metadata for key in _RESEARCH_LINK_KEYS)
        or isinstance(custom, dict)
        and any(key in custom for key in _RESEARCH_LINK_KEYS)
        or isinstance(content, list)
        and any(
            isinstance(part, dict) and any(key in part for key in _RESEARCH_LINK_KEYS)
            for part in content
        )
    )
    if not linked:
        return content_json, metadata_json

    if isinstance(content, list):
        content = [
            {key: value for key, value in part.items() if key not in _RESEARCH_LINK_KEYS}
            if isinstance(part, dict)
            else part
            for part in content
        ]
    if isinstance(metadata, dict):
        metadata = {key: value for key, value in metadata.items() if key not in _RESEARCH_LINK_KEYS}
        custom = metadata.get("custom")
        if isinstance(custom, dict):
            metadata["custom"] = {
                key: value for key, value in custom.items() if key not in _RESEARCH_LINK_KEYS
            }
    return (
        json.dumps(content, ensure_ascii = False),
        json.dumps(metadata, ensure_ascii = False) if metadata is not None else None,
    )


def fork_chat_thread(
    source_thread_id: str,
    branch_message_id: str,
    new_thread_id: str,
    new_title: str,
    created_at: int,
    id_factory,
) -> Optional[dict]:
    """Atomically clone thread + ancestor msgs `[root..branch_message_id]`
    into a new thread. Returns the new thread dict (with messages copied)
    or None if source missing.

    Reset both code-exec container ids -- per-provider snapshot is handled
    by the route layer (best-effort, OpenAI only).

    `id_factory()` produces fresh message uuids; injected for testability.
    """
    conn = get_connection()
    try:
        conn.execute("BEGIN IMMEDIATE")
        _ensure_chat_attachment_inventory_current(conn)
        src = conn.execute(
            "SELECT * FROM chat_threads WHERE id = ?", (source_thread_id,)
        ).fetchone()
        if src is None:
            conn.rollback()
            return None
        # Verify branch msg belongs to source thread.
        branch_row = conn.execute(
            "SELECT * FROM chat_messages WHERE thread_id = ? AND id = ?",
            (source_thread_id, branch_message_id),
        ).fetchone()
        if branch_row is None:
            conn.rollback()
            return None
        # Walk ancestry from branch msg back to root via parent_id chain.
        ancestry: list[sqlite3.Row] = []
        cursor_row = branch_row
        seen: set[str] = set()
        while cursor_row is not None and cursor_row["id"] not in seen:
            ancestry.append(cursor_row)
            seen.add(cursor_row["id"])
            parent = cursor_row["parent_id"]
            if not parent:
                break
            cursor_row = conn.execute(
                "SELECT * FROM chat_messages WHERE thread_id = ? AND id = ?",
                (source_thread_id, parent),
            ).fetchone()
        ancestry.reverse()  # root .. branch msg
        # Map old msg id -> new msg id for parent_id rewriting.
        id_map: dict[str, str] = {row["id"]: id_factory() for row in ancestry}
        src_dict = dict(src)
        conn.execute(
            """
            INSERT INTO chat_threads
                (id, title, model_type, model_id, pair_id, project_id, archived, created_at,
                 openai_code_exec_container_id, anthropic_code_exec_container_id,
                 forked_from_thread_id, forked_from_message_id)
            VALUES (?, ?, ?, ?, ?, ?, 0, ?, NULL, NULL, ?, ?)
            """,
            (
                new_thread_id,
                new_title,
                src_dict["model_type"],
                src_dict.get("model_id") or "",
                None,  # pairId: forks always standalone (compare-mode disabled v1)
                src_dict.get("project_id"),
                int(created_at),
                source_thread_id,
                branch_message_id,
            ),
        )
        fork_messages = []
        for row in ancestry:
            content_json, metadata_json = _detach_research_message_json(
                row["content_json"], row["metadata_json"]
            )
            fork_messages.append(
                (
                    id_map[row["id"]],
                    new_thread_id,
                    id_map.get(row["parent_id"]) if row["parent_id"] else None,
                    row["role"],
                    content_json,
                    row["attachments_json"],
                    metadata_json,
                    int(row["created_at"]),
                )
            )
        conn.executemany(
            """
            INSERT INTO chat_messages
                (id, thread_id, parent_id, role, content_json, attachments_json,
                 metadata_json, created_at)
            VALUES (?, ?, ?, ?, ?, ?, ?, ?)
            """,
            fork_messages,
        )
        for row in ancestry:
            _replace_chat_attachment_inventory(
                conn,
                id_map[row["id"]],
                row["attachments_json"],
                row["content_json"],
            )
        _mark_chat_attachment_inventory_clean(conn)
        conn.commit()
        thread_row = conn.execute(
            "SELECT * FROM chat_threads WHERE id = ?", (new_thread_id,)
        ).fetchone()
        return _chat_thread_from_row(thread_row) if thread_row is not None else None
    except Exception:
        conn.rollback()
        raise
    finally:
        conn.close()


def count_forks_for_message(thread_id: str, message_id: str) -> int:
    conn = get_connection()
    try:
        row = conn.execute(
            """
            SELECT COUNT(*) FROM chat_threads
            WHERE forked_from_thread_id = ? AND forked_from_message_id = ?
            """,
            (thread_id, message_id),
        ).fetchone()
        return int(row[0]) if row is not None else 0
    finally:
        conn.close()


def list_chat_messages(thread_id: str) -> list[dict]:
    conn = get_connection()
    try:
        rows = conn.execute(
            """
            SELECT * FROM chat_messages
            WHERE thread_id = ?
            ORDER BY created_at ASC, id ASC
            """,
            (thread_id,),
        ).fetchall()
        return [_chat_message_from_row(row) for row in rows]
    finally:
        conn.close()


def get_chat_message(thread_id: str, message_id: str) -> Optional[dict]:
    conn = get_connection()
    try:
        row = conn.execute(
            """
            SELECT * FROM chat_messages
            WHERE thread_id = ? AND id = ?
            """,
            (thread_id, message_id),
        ).fetchone()
        return _chat_message_from_row(row) if row is not None else None
    finally:
        conn.close()


def _blob_part_base64_len(part: dict) -> int:
    """Base64 payload length of an image or audio content part, or 0."""
    image = part.get("image")
    if isinstance(image, str) and image[:5].lower() == "data:":
        return len(image.rsplit(",", 1)[-1])
    audio = part.get("audio")
    if isinstance(audio, str) and _is_locally_stored_blob(audio):
        return len(audio.rsplit(",", 1)[-1])
    if isinstance(audio, dict):
        data = audio.get("data")
        if isinstance(data, str) and _is_locally_stored_blob(data):
            return len(data)
    return 0


def _chat_attachment_size_bytes(attachment: dict) -> Optional[int]:
    """Approximate stored size of one attachment's content parts.

    Image and audio parts hold base64 payloads (decoded bytes ~= 3/4 of the
    encoded length); text parts count their character length. None when there
    is no sizable content (e.g. a stripped/legacy attachment).
    """
    total = 0
    found = False
    for part in attachment.get("content") or []:
        if not isinstance(part, dict):
            continue
        blob_len = _blob_part_base64_len(part)
        if blob_len > 0:
            total += (blob_len * 3) // 4
            found = True
            continue
        text = part.get("text")
        if isinstance(text, str) and text:
            total += len(text.encode("utf-8", errors = "ignore"))
            found = True
    return total if found else None


def _content_part_attachments(content_json: Optional[str]) -> list[dict]:
    """Managed local blobs stored in content_json, with stable payload ids.

    Exact duplicate blobs intentionally share one inventory id. Deleting that
    id removes every identical copy, avoiding ambiguous index-based addressing.
    """
    content = _json_loads(content_json, None)
    if not isinstance(content, list):
        return []
    out: list[dict] = []
    seen: set[str] = set()
    for part in content:
        if not isinstance(part, dict):
            continue
        attachment_id = _content_part_id(part)
        payload = _managed_content_part_payload(part)
        if attachment_id is None or payload is None or attachment_id in seen:
            continue
        seen.add(attachment_id)
        kind, value = payload
        content_type = None
        if kind == "image" and isinstance(value, str):
            content_type = value[5:].split(";", 1)[0].split(",", 1)[0] or None
        out.append(
            {
                "id": attachment_id,
                "type": kind,
                "name": "Chat image" if kind == "image" else "Chat audio",
                "contentType": content_type,
                "content": [part],
            }
        )
    return out


def list_chat_attachments_page(
    limit: int = 50, offset: int = 0
) -> tuple[list[dict], Optional[int]]:
    """One bounded page from the normalized attachment inventory."""
    if not 1 <= limit <= 100:
        raise ValueError("limit must be between 1 and 100")
    if offset < 0:
        raise ValueError("offset must be non-negative")

    conn = get_connection()
    try:
        _ensure_chat_attachment_inventory_current(conn)
        rows = conn.execute(
            """
            SELECT i.attachment_id, i.name, i.type, i.content_type,
                   i.size_bytes, m.id AS message_id, m.thread_id,
                   m.created_at, t.title AS thread_title, t.pair_id
            FROM chat_attachment_inventory i
            JOIN chat_messages m ON m.id = i.message_id
            LEFT JOIN chat_threads t ON t.id = m.thread_id
            ORDER BY m.created_at DESC, m.id ASC, i.attachment_id ASC
            LIMIT ? OFFSET ?
            """,
            (limit + 1, offset),
        ).fetchall()
    finally:
        conn.close()

    has_more = len(rows) > limit
    page_rows = rows[:limit]
    attachments = [
        {
            "id": row["attachment_id"],
            "messageId": row["message_id"],
            "threadId": row["thread_id"],
            "pairId": row["pair_id"],
            "threadTitle": row["thread_title"],
            "name": row["name"],
            "type": row["type"],
            "contentType": row["content_type"],
            "sizeBytes": row["size_bytes"],
            "createdAt": row["created_at"],
        }
        for row in page_rows
    ]
    return attachments, offset + limit if has_more else None


def list_chat_attachments() -> list[dict]:
    """Compatibility helper returning the full normalized inventory."""
    attachments: list[dict] = []
    offset = 0
    while True:
        page, next_offset = list_chat_attachments_page(limit = 100, offset = offset)
        attachments.extend(page)
        if next_offset is None:
            return attachments
        offset = next_offset


def get_chat_attachment(message_id: str, attachment_id: str) -> Optional[dict]:
    """One attachment record (full content) from a message, or None."""
    conn = get_connection()
    try:
        row = conn.execute(
            """
            SELECT message.attachments_json, message.content_json,
                   EXISTS(
                       SELECT 1 FROM chat_attachment_tombstones tombstone
                       WHERE tombstone.thread_id = message.thread_id
                         AND tombstone.message_id = message.id
                         AND tombstone.attachment_id = ?
                   ) AS tombstoned
            FROM chat_messages message
            WHERE message.id = ?
            """,
            (attachment_id, message_id),
        ).fetchone()
    finally:
        conn.close()
    if row is None or row["tombstoned"]:
        return None
    attachments = _json_loads(row["attachments_json"], None)
    if isinstance(attachments, list):
        for attachment in attachments:
            if isinstance(attachment, dict) and str(attachment.get("id") or "") == attachment_id:
                return attachment
    if attachment_id.startswith(_CONTENT_PART_ID_PREFIX):
        for attachment in _content_part_attachments(row["content_json"]):
            if attachment["id"] == attachment_id:
                return attachment
    return None


def _record_chat_attachment_tombstone(
    conn: sqlite3.Connection, thread_id: str, message_id: str, attachment_id: str
) -> None:
    conn.execute(
        """
        INSERT INTO chat_attachment_tombstones
            (thread_id, message_id, attachment_id, deleted_at)
        VALUES (?, ?, ?, ?)
        ON CONFLICT(thread_id, message_id, attachment_id) DO UPDATE SET
            deleted_at = excluded.deleted_at
        """,
        (
            thread_id,
            message_id,
            attachment_id,
            int(datetime.now(timezone.utc).timestamp() * 1000),
        ),
    )


def delete_chat_attachment(message_id: str, attachment_id: str) -> bool:
    """Remove one stored upload from a message.

    The tombstone is retained while the thread exists, so pruning and later
    recreating the same message id cannot restore the deleted upload. If an
    ordinary attachment id collides with a content-blob id, both are deleted as
    one managed item.
    """
    conn = get_connection()
    try:
        conn.execute("BEGIN IMMEDIATE")
        _ensure_chat_attachment_inventory_current(conn)
        row = conn.execute(
            """
            SELECT thread_id, attachments_json, content_json
            FROM chat_messages WHERE id = ?
            """,
            (message_id,),
        ).fetchone()
        if row is None:
            conn.rollback()
            return False
        if str(message_id) in _research_message_ids(conn, str(row["thread_id"])):
            conn.rollback()
            raise ChatMessageProtectedError(
                "Research prompts and responses are server-managed and cannot be edited"
            )

        attachments = _json_loads(row["attachments_json"], None)
        updated_attachments_json = row["attachments_json"]
        deleted_attachment = False
        if isinstance(attachments, list):
            remaining_attachments = [
                attachment
                for attachment in attachments
                if not (
                    isinstance(attachment, dict)
                    and str(attachment.get("id") or "") == attachment_id
                )
            ]
            deleted_attachment = len(remaining_attachments) != len(attachments)
            if deleted_attachment:
                updated_attachments_json = json.dumps(remaining_attachments)

        content = _json_loads(row["content_json"], None)
        updated_content_json = row["content_json"]
        deleted_content = False
        if attachment_id.startswith(_CONTENT_PART_ID_PREFIX) and isinstance(content, list):
            remaining_content = [
                part
                for part in content
                if not (isinstance(part, dict) and _content_part_id(part) == attachment_id)
            ]
            deleted_content = len(remaining_content) != len(content)
            if deleted_content:
                updated_content_json = json.dumps(remaining_content)

        if not deleted_attachment and not deleted_content:
            conn.rollback()
            return False
        conn.execute(
            """
            UPDATE chat_messages
            SET attachments_json = ?, content_json = ?
            WHERE id = ?
            """,
            (updated_attachments_json, updated_content_json, message_id),
        )
        _record_chat_attachment_tombstone(
            conn,
            row["thread_id"],
            message_id,
            attachment_id,
        )
        _replace_chat_attachment_inventory(
            conn,
            message_id,
            updated_attachments_json,
            updated_content_json,
        )
        _mark_chat_attachment_inventory_clean(conn)
        conn.commit()
        return True
    except Exception:
        conn.rollback()
        raise
    finally:
        conn.close()


def list_chat_messages_for_threads(thread_ids: list[str]) -> list[dict]:
    if not thread_ids:
        return []
    unique_thread_ids = list(dict.fromkeys(thread_ids))
    messages: list[dict] = []
    conn = get_connection()
    try:
        for start in range(0, len(unique_thread_ids), _SQLITE_IN_CHUNK_SIZE):
            chunk = unique_thread_ids[start : start + _SQLITE_IN_CHUNK_SIZE]
            placeholders = ",".join("?" for _ in chunk)
            rows = conn.execute(
                f"""
                SELECT * FROM chat_messages
                WHERE thread_id IN ({placeholders})
                ORDER BY created_at ASC, id ASC
                """,
                chunk,
            ).fetchall()
            messages.extend(_chat_message_from_row(row) for row in rows)
        return sorted(
            messages,
            key = lambda message: (message["createdAt"], message["id"]),
        )
    finally:
        conn.close()


def get_app_setting(key: str, fallback = None):
    conn = get_connection()
    try:
        row = conn.execute("SELECT value_json FROM app_settings WHERE key = ?", (key,)).fetchone()
        if row is None:
            return fallback
        return _json_loads(row["value_json"], fallback)
    finally:
        conn.close()


def upsert_app_settings(settings: dict[str, Any]) -> dict[str, Any]:
    if not settings:
        return {}
    conn = get_connection()
    try:
        now = datetime.now(timezone.utc).isoformat()
        conn.executemany(
            """
            INSERT INTO app_settings (key, value_json, updated_at)
            VALUES (?, ?, ?)
            ON CONFLICT(key) DO UPDATE SET
                value_json = excluded.value_json,
                updated_at = excluded.updated_at
            """,
            [(key, json.dumps(value), now) for key, value in settings.items()],
        )
        conn.commit()
        rows = conn.execute("SELECT key, value_json FROM app_settings ORDER BY key").fetchall()
        return {row["key"]: _json_loads(row["value_json"], None) for row in rows}
    finally:
        conn.close()


def upsert_app_setting_map_entry(
    key: str, entry_key: str, entry_value: dict[str, Any] | None
) -> dict[str, Any]:
    """Set (or delete, when entry_value is falsy) one sub-entry of a dict-valued
    app setting, atomically under BEGIN IMMEDIATE so concurrent writers to other
    sub-entries cannot drop each other's updates."""
    conn = get_connection()
    try:
        conn.execute("BEGIN IMMEDIATE")
        row = conn.execute("SELECT value_json FROM app_settings WHERE key = ?", (key,)).fetchone()
        current = _json_loads(row["value_json"], {}) if row else {}
        if not isinstance(current, dict):
            current = {}
        if entry_value:
            current[entry_key] = entry_value
        else:
            current.pop(entry_key, None)
        now = datetime.now(timezone.utc).isoformat()
        conn.execute(
            """
            INSERT INTO app_settings (key, value_json, updated_at)
            VALUES (?, ?, ?)
            ON CONFLICT(key) DO UPDATE SET
                value_json = excluded.value_json,
                updated_at = excluded.updated_at
            """,
            (key, json.dumps(current), now),
        )
        conn.commit()
        return current
    except Exception:
        conn.rollback()
        raise
    finally:
        conn.close()


def list_chat_settings() -> dict[str, Any]:
    conn = get_connection()
    try:
        rows = conn.execute("SELECT key, value_json FROM chat_settings ORDER BY key").fetchall()
        settings: dict[str, Any] = {}
        for row in rows:
            settings[row["key"]] = _json_loads(row["value_json"], None)
        return settings
    finally:
        conn.close()


def upsert_chat_settings(settings: dict[str, Any]) -> dict[str, Any]:
    if not settings:
        return list_chat_settings()
    conn = get_connection()
    try:
        now = datetime.now(timezone.utc).isoformat()
        conn.executemany(
            """
            INSERT INTO chat_settings (key, value_json, updated_at)
            VALUES (?, ?, ?)
            ON CONFLICT(key) DO UPDATE SET
                value_json = excluded.value_json,
                updated_at = excluded.updated_at
            """,
            [(key, json.dumps(value), now) for key, value in settings.items()],
        )
        conn.commit()
        return list_chat_settings()
    finally:
        conn.close()


def _deep_merge_settings(current: dict[str, Any], updates: dict[str, Any]) -> dict[str, Any]:
    merged = dict(current)
    for key, value in updates.items():
        current_value = merged.get(key)
        if isinstance(current_value, dict) and isinstance(value, dict):
            merged[key] = _deep_merge_settings(current_value, value)
        else:
            merged[key] = value
    return merged


def upsert_chat_settings_merge(updates: dict[str, Any]) -> dict[str, Any]:
    """Atomic read-merge-write under BEGIN IMMEDIATE so concurrent writers
    cannot drop each other's updates."""
    if not updates:
        return list_chat_settings()
    conn = get_connection()
    try:
        conn.execute("BEGIN IMMEDIATE")
        current, corrupt = _load_chat_settings_for_merge(conn)
        unsafe_partial_keys = [
            key for key, value in updates.items() if key in corrupt and isinstance(value, dict)
        ]
        if unsafe_partial_keys:
            conn.commit()
            keys = ", ".join(sorted(unsafe_partial_keys))
            raise CorruptSettingsError(
                f"Cannot apply partial settings patch to corrupt key(s): {keys}"
            )
        merged = _deep_merge_settings(current, updates)
        now = datetime.now(timezone.utc).isoformat()
        conn.executemany(
            """
            INSERT INTO chat_settings (key, value_json, updated_at)
            VALUES (?, ?, ?)
            ON CONFLICT(key) DO UPDATE SET
                value_json = excluded.value_json,
                updated_at = excluded.updated_at
            """,
            [(key, json.dumps(value), now) for key, value in merged.items()],
        )
        conn.commit()
        return merged
    except CorruptSettingsError:
        raise
    except Exception:
        conn.rollback()
        raise
    finally:
        conn.close()


# ---------------------------------------------------------------------------
# Legacy Dexie import ledger
# ---------------------------------------------------------------------------
# See the schema comment in _ensure_schema() for the recovery rationale.


def list_chat_legacy_imports() -> list[str]:
    """Return the legacy_thread_id of every thread already imported."""
    conn = get_connection()
    try:
        rows = conn.execute("SELECT legacy_thread_id FROM chat_legacy_imports").fetchall()
        return [row[0] for row in rows]
    finally:
        conn.close()


def upsert_chat_legacy_imports(legacy_thread_ids: list[str]) -> tuple[int, int]:
    """Mark each given legacy thread id as imported. Idempotent.

    Returns (accepted, inserted): count of deduped non-empty input ids, and
    count of rows actually new. RETURNING lets callers tell first-time imports
    from idempotent re-runs without an extra SELECT.
    """
    ids = list(dict.fromkeys(tid for tid in legacy_thread_ids if tid))
    if not ids:
        return 0, 0
    ts = int(datetime.now(timezone.utc).timestamp() * 1000)
    conn = get_connection()
    try:
        inserted = 0
        for tid in ids:
            row = conn.execute(
                """
                INSERT INTO chat_legacy_imports (legacy_thread_id, imported_at)
                VALUES (?, ?)
                ON CONFLICT(legacy_thread_id) DO NOTHING
                RETURNING legacy_thread_id
                """,
                (tid, ts),
            ).fetchone()
            if row is not None:
                inserted += 1
        conn.commit()
        return len(ids), inserted
    finally:
        conn.close()