ceos/unsloth
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions
unsloth/studio/backend/storage/studio_db.py
danielhanchen 71a07515ff Studio: fix Deep Research review findings (RAG slot bound, orphaned workers, hardening) 
- Bound the shared RAG search slot to one running worker. The search that is
  doing the embedding/index/GPU work now owns the admission slot until it
  finishes, instead of freeing it on caller timeout while the detached worker
  keeps running, which let a second search enter and stack concurrent work
  behind the capacity-of-one semaphore.
- Cancel active research runs before deleting their thread, project, or all
  history. Deleting cascade-drops the run row, but the worker only notices at
  its next lease check, so it could keep doing model/web/RAG work for a run
  that no longer exists; signalling cancel first shortens that window.
- Shield the planner prompt's conversation and question with _shield_untrusted,
  matching the decision and synthesis prompts, so untrusted text cannot forge
  planner delimiters.
- Do not let a research key-revocation failure replace a successful
  non-streaming completion; log it like the streaming path does.
- Include created_at in the protected research-message guard so a client cannot
  reorder server-managed prompt/response messages while leaving the body intact.
- Reject non-scalar ragScope values; a nested container evades the
  sensitive-key scan when its inner keys are unlisted and would reach retrieval
  code that expects a scalar scope id.

Adds regression tests for each.
2026-07-22 13:10:38 +00:00

3130 lines
107 KiB
Python
Raw Blame History

# 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 left the new table empty and the
# rows orphaned in _legacy, and the migration 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()
Reference in a new issue View git blame Copy permalink