wehub-resource-sync
e4dcfc49aa
Tests / Import Check (Python 3.13) (push) Has been cancelled
Tests / Import Check (Python 3.14) (push) Has been cancelled
Tests / Python Tests (Python 3.11) (push) Has been cancelled
Tests / Python Tests (Python 3.12) (push) Has been cancelled
Tests / Python Tests (Python 3.14) (push) Has been cancelled
Tests / Test Summary (push) Has been cancelled
Tests / Lint and Format (push) Has been cancelled
Tests / Web Node Tests (push) Has been cancelled
Tests / Import Check (Python 3.11) (push) Has been cancelled
Tests / Import Check (Python 3.12) (push) Has been cancelled
Tests / Python Tests (Python 3.13) (push) Has been cancelled
1845 lines
74 KiB
Python
1845 lines
74 KiB
Python
"""
|
|
SQLite-backed unified chat session store.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import asyncio
|
|
from collections.abc import Iterator
|
|
from contextlib import contextmanager
|
|
from dataclasses import dataclass
|
|
import json
|
|
import os
|
|
from pathlib import Path
|
|
import re
|
|
import sqlite3
|
|
import time
|
|
from typing import Any
|
|
import uuid
|
|
|
|
from deeptutor.services.path_service import get_path_service
|
|
|
|
|
|
def _json_dumps(value: Any) -> str:
|
|
# default=str: a single non-serializable object inside an event payload
|
|
# (e.g. a dataclass smuggled into tool args) must degrade to its repr,
|
|
# never kill message/event persistence for the whole turn.
|
|
return json.dumps(value, ensure_ascii=False, default=str)
|
|
|
|
|
|
# Sentinel so ``add_message`` can distinguish "caller wants the legacy
|
|
# auto-pick-latest-message default" from "caller explicitly wants the
|
|
# message attached at the session root (parent = NULL)". Both surface as
|
|
# ``None`` in the public ``parent_message_id`` arg, which is why we need
|
|
# a sentinel separate from None.
|
|
class _Unset:
|
|
pass
|
|
|
|
|
|
_PARENT_AUTO = _Unset()
|
|
|
|
|
|
def _json_loads(value: str | None, default: Any) -> Any:
|
|
if not value:
|
|
return default
|
|
try:
|
|
return json.loads(value)
|
|
except json.JSONDecodeError:
|
|
return default
|
|
|
|
|
|
# Imported conversations share the session tables with native chats but carry
|
|
# this id prefix as their discriminator (see ``SQLiteSessionStore._WHERE_*``).
|
|
_IMPORTED_ID_PREFIX = "imported_"
|
|
_ID_SAFE = re.compile(r"[^A-Za-z0-9_-]")
|
|
|
|
|
|
def make_imported_session_id(source: str, external_id: str) -> str:
|
|
"""Build a deterministic, dedup-friendly id for an imported conversation.
|
|
|
|
``source`` (e.g. ``claude_code``/``codex``) namespaces the original
|
|
session uuid so two tools that happen to reuse an id never collide; the
|
|
determinism is what makes re-importing the same folder idempotent.
|
|
"""
|
|
src = _ID_SAFE.sub("-", (source or "external").strip()) or "external"
|
|
ext = _ID_SAFE.sub("-", (external_id or "").strip()) or uuid.uuid4().hex
|
|
return f"{_IMPORTED_ID_PREFIX}{src}_{ext}"
|
|
|
|
|
|
@dataclass
|
|
class TurnRecord:
|
|
id: str
|
|
session_id: str
|
|
capability: str
|
|
status: str
|
|
error: str
|
|
created_at: float
|
|
updated_at: float
|
|
finished_at: float | None
|
|
last_seq: int = 0
|
|
|
|
def to_dict(self) -> dict[str, Any]:
|
|
return {
|
|
"id": self.id,
|
|
"turn_id": self.id,
|
|
"session_id": self.session_id,
|
|
"capability": self.capability,
|
|
"status": self.status,
|
|
"error": self.error,
|
|
"created_at": self.created_at,
|
|
"updated_at": self.updated_at,
|
|
"finished_at": self.finished_at,
|
|
"last_seq": self.last_seq,
|
|
}
|
|
|
|
|
|
class SQLiteSessionStore:
|
|
"""Persist unified chat sessions and messages in a SQLite database."""
|
|
|
|
def __init__(self, db_path: Path | None = None) -> None:
|
|
path_service = get_path_service()
|
|
self.db_path = db_path or path_service.get_chat_history_db()
|
|
self.db_path.parent.mkdir(parents=True, exist_ok=True)
|
|
self._migrate_legacy_db(path_service)
|
|
self._lock = asyncio.Lock()
|
|
self._initialize()
|
|
|
|
def _migrate_legacy_db(self, path_service) -> None:
|
|
"""Move the legacy ``data/chat_history.db`` into ``data/user/`` once."""
|
|
legacy_path = path_service.project_root / "data" / "chat_history.db"
|
|
if self.db_path.exists() or not legacy_path.exists() or legacy_path == self.db_path:
|
|
return
|
|
try:
|
|
os.replace(legacy_path, self.db_path)
|
|
except OSError:
|
|
# Fall back to leaving the legacy DB in place if an OS-level move
|
|
# is not possible; the new DB path will be initialized empty.
|
|
pass
|
|
|
|
def _initialize(self) -> None:
|
|
with self._connect() as conn:
|
|
conn.executescript(
|
|
"""
|
|
CREATE TABLE IF NOT EXISTS sessions (
|
|
id TEXT PRIMARY KEY,
|
|
title TEXT NOT NULL DEFAULT 'New conversation',
|
|
created_at REAL NOT NULL,
|
|
updated_at REAL NOT NULL,
|
|
compressed_summary TEXT DEFAULT '',
|
|
summary_up_to_msg_id INTEGER DEFAULT 0,
|
|
preferences_json TEXT DEFAULT '{}'
|
|
);
|
|
|
|
CREATE TABLE IF NOT EXISTS messages (
|
|
id INTEGER PRIMARY KEY AUTOINCREMENT,
|
|
session_id TEXT NOT NULL REFERENCES sessions(id) ON DELETE CASCADE,
|
|
role TEXT NOT NULL,
|
|
content TEXT NOT NULL DEFAULT '',
|
|
capability TEXT DEFAULT '',
|
|
events_json TEXT DEFAULT '',
|
|
attachments_json TEXT DEFAULT '',
|
|
metadata_json TEXT DEFAULT '{}',
|
|
created_at REAL NOT NULL,
|
|
-- Edit-branching: NULL for the first message in a session;
|
|
-- otherwise the immediately preceding message on the path
|
|
-- this row continues. Siblings (same parent) are alternate
|
|
-- branches the user can switch between.
|
|
parent_message_id INTEGER
|
|
);
|
|
|
|
CREATE INDEX IF NOT EXISTS idx_messages_session_created
|
|
ON messages(session_id, created_at, id);
|
|
-- ``idx_messages_parent`` is created after the
|
|
-- parent_message_id migration runs (see below). Putting it
|
|
-- in this script would fail on legacy DBs where the column
|
|
-- gets added by ALTER TABLE further down.
|
|
|
|
CREATE INDEX IF NOT EXISTS idx_sessions_updated_at
|
|
ON sessions(updated_at DESC);
|
|
|
|
CREATE TABLE IF NOT EXISTS turns (
|
|
id TEXT PRIMARY KEY,
|
|
session_id TEXT NOT NULL REFERENCES sessions(id) ON DELETE CASCADE,
|
|
capability TEXT DEFAULT '',
|
|
status TEXT NOT NULL DEFAULT 'running',
|
|
error TEXT DEFAULT '',
|
|
created_at REAL NOT NULL,
|
|
updated_at REAL NOT NULL,
|
|
finished_at REAL
|
|
);
|
|
|
|
CREATE INDEX IF NOT EXISTS idx_turns_session_updated
|
|
ON turns(session_id, updated_at DESC);
|
|
|
|
CREATE INDEX IF NOT EXISTS idx_turns_session_status
|
|
ON turns(session_id, status, updated_at DESC);
|
|
|
|
CREATE TABLE IF NOT EXISTS turn_events (
|
|
id INTEGER PRIMARY KEY AUTOINCREMENT,
|
|
turn_id TEXT NOT NULL REFERENCES turns(id) ON DELETE CASCADE,
|
|
seq INTEGER NOT NULL,
|
|
type TEXT NOT NULL,
|
|
source TEXT DEFAULT '',
|
|
stage TEXT DEFAULT '',
|
|
content TEXT DEFAULT '',
|
|
metadata_json TEXT DEFAULT '',
|
|
timestamp REAL NOT NULL,
|
|
created_at REAL NOT NULL,
|
|
UNIQUE(turn_id, seq)
|
|
);
|
|
|
|
CREATE INDEX IF NOT EXISTS idx_turn_events_turn_seq
|
|
ON turn_events(turn_id, seq);
|
|
|
|
CREATE TABLE IF NOT EXISTS notebook_entries (
|
|
id INTEGER PRIMARY KEY AUTOINCREMENT,
|
|
session_id TEXT NOT NULL REFERENCES sessions(id) ON DELETE CASCADE,
|
|
turn_id TEXT NOT NULL DEFAULT '',
|
|
question_id TEXT NOT NULL,
|
|
question TEXT NOT NULL,
|
|
question_type TEXT DEFAULT '',
|
|
options_json TEXT DEFAULT '{}',
|
|
correct_answer TEXT DEFAULT '',
|
|
explanation TEXT DEFAULT '',
|
|
difficulty TEXT DEFAULT '',
|
|
user_answer TEXT DEFAULT '',
|
|
user_answer_images_json TEXT DEFAULT '[]',
|
|
is_correct INTEGER DEFAULT 0,
|
|
bookmarked INTEGER DEFAULT 0,
|
|
followup_session_id TEXT DEFAULT '',
|
|
ai_judgment TEXT DEFAULT '',
|
|
created_at REAL NOT NULL,
|
|
updated_at REAL NOT NULL,
|
|
UNIQUE(session_id, turn_id, question_id)
|
|
);
|
|
|
|
CREATE INDEX IF NOT EXISTS idx_notebook_entries_session
|
|
ON notebook_entries(session_id, created_at DESC);
|
|
|
|
CREATE INDEX IF NOT EXISTS idx_notebook_entries_bookmarked
|
|
ON notebook_entries(bookmarked, created_at DESC);
|
|
|
|
CREATE TABLE IF NOT EXISTS notebook_categories (
|
|
id INTEGER PRIMARY KEY AUTOINCREMENT,
|
|
name TEXT NOT NULL UNIQUE,
|
|
created_at REAL NOT NULL
|
|
);
|
|
|
|
CREATE TABLE IF NOT EXISTS notebook_entry_categories (
|
|
entry_id INTEGER NOT NULL REFERENCES notebook_entries(id) ON DELETE CASCADE,
|
|
category_id INTEGER NOT NULL REFERENCES notebook_categories(id) ON DELETE CASCADE,
|
|
PRIMARY KEY (entry_id, category_id)
|
|
);
|
|
"""
|
|
)
|
|
columns = {row[1] for row in conn.execute("PRAGMA table_info(sessions)").fetchall()}
|
|
if "preferences_json" not in columns:
|
|
conn.execute("ALTER TABLE sessions ADD COLUMN preferences_json TEXT DEFAULT '{}'")
|
|
if "kind" in columns:
|
|
try:
|
|
conn.execute("ALTER TABLE sessions DROP COLUMN kind")
|
|
except sqlite3.OperationalError:
|
|
# Older SQLite builds may not support DROP COLUMN. The
|
|
# application no longer reads or writes this legacy field.
|
|
pass
|
|
message_columns = {
|
|
row[1] for row in conn.execute("PRAGMA table_info(messages)").fetchall()
|
|
}
|
|
if "metadata_json" not in message_columns:
|
|
conn.execute("ALTER TABLE messages ADD COLUMN metadata_json TEXT DEFAULT '{}'")
|
|
if "parent_message_id" not in message_columns:
|
|
conn.execute("ALTER TABLE messages ADD COLUMN parent_message_id INTEGER")
|
|
# Backfill: for every existing session, treat the message stream
|
|
# as a single linear path — each row's parent is the previous
|
|
# row (by id) in the same session. Rows with no predecessor stay
|
|
# NULL. We do this per session in pure Python to avoid relying
|
|
# on window functions, which older SQLite builds may not have.
|
|
sessions_rows = conn.execute("SELECT id FROM sessions").fetchall()
|
|
for srow in sessions_rows:
|
|
prev_id: int | None = None
|
|
msg_rows = conn.execute(
|
|
"SELECT id FROM messages WHERE session_id = ? ORDER BY id ASC",
|
|
(srow[0],),
|
|
).fetchall()
|
|
for mrow in msg_rows:
|
|
if prev_id is not None:
|
|
conn.execute(
|
|
"UPDATE messages SET parent_message_id = ? WHERE id = ?",
|
|
(prev_id, mrow[0]),
|
|
)
|
|
prev_id = mrow[0]
|
|
# Always ensure the parent-lookup index exists — covers both
|
|
# the legacy-migration case (just added the column) and the
|
|
# fresh-DB case (created above without the index inline).
|
|
conn.execute(
|
|
"CREATE INDEX IF NOT EXISTS idx_messages_parent "
|
|
"ON messages(session_id, parent_message_id)"
|
|
)
|
|
self._migrate_notebook_entries_add_turn_id(conn)
|
|
self._migrate_notebook_entries_add_user_answer_images(conn)
|
|
self._migrate_notebook_entries_add_ai_judgment(conn)
|
|
conn.commit()
|
|
|
|
@staticmethod
|
|
def _migrate_notebook_entries_add_turn_id(conn: sqlite3.Connection) -> None:
|
|
"""Add ``turn_id`` to legacy notebook_entries and re-scope the UNIQUE
|
|
constraint to ``(session_id, turn_id, question_id)``.
|
|
|
|
The old unique constraint conflated quizzes generated in the same chat
|
|
(issue #487): regenerating a quiz with the same positional
|
|
``question_id`` (e.g. ``q_1``) would collide with the previous quiz's
|
|
notebook entries and the UI hydrated stale answers. Scoping by
|
|
``turn_id`` keeps each quiz isolated.
|
|
"""
|
|
notebook_cols = {
|
|
row[1] for row in conn.execute("PRAGMA table_info(notebook_entries)").fetchall()
|
|
}
|
|
if not notebook_cols:
|
|
return
|
|
if "turn_id" not in notebook_cols:
|
|
conn.execute("ALTER TABLE notebook_entries ADD COLUMN turn_id TEXT NOT NULL DEFAULT ''")
|
|
# SQLite stores table-level UNIQUE constraints as auto-indexes whose
|
|
# names start with ``sqlite_autoindex_notebook_entries_``; the columns
|
|
# they cover live in PRAGMA index_info. Detect whether any existing
|
|
# auto-index still covers only (session_id, question_id) and, if so,
|
|
# rebuild the table to swap in the new scope.
|
|
needs_rebuild = False
|
|
for idx_row in conn.execute("PRAGMA index_list(notebook_entries)").fetchall():
|
|
idx_name = idx_row[1]
|
|
if not idx_name.startswith("sqlite_autoindex_notebook_entries_"):
|
|
continue
|
|
cols = [r[2] for r in conn.execute(f"PRAGMA index_info({idx_name})").fetchall()]
|
|
if cols == ["session_id", "question_id"]:
|
|
needs_rebuild = True
|
|
break
|
|
if not needs_rebuild:
|
|
return
|
|
conn.executescript(
|
|
"""
|
|
CREATE TABLE notebook_entries_new (
|
|
id INTEGER PRIMARY KEY AUTOINCREMENT,
|
|
session_id TEXT NOT NULL REFERENCES sessions(id) ON DELETE CASCADE,
|
|
turn_id TEXT NOT NULL DEFAULT '',
|
|
question_id TEXT NOT NULL,
|
|
question TEXT NOT NULL,
|
|
question_type TEXT DEFAULT '',
|
|
options_json TEXT DEFAULT '{}',
|
|
correct_answer TEXT DEFAULT '',
|
|
explanation TEXT DEFAULT '',
|
|
difficulty TEXT DEFAULT '',
|
|
user_answer TEXT DEFAULT '',
|
|
is_correct INTEGER DEFAULT 0,
|
|
bookmarked INTEGER DEFAULT 0,
|
|
followup_session_id TEXT DEFAULT '',
|
|
created_at REAL NOT NULL,
|
|
updated_at REAL NOT NULL,
|
|
UNIQUE(session_id, turn_id, question_id)
|
|
);
|
|
|
|
INSERT INTO notebook_entries_new (
|
|
id, session_id, turn_id, question_id, question, question_type,
|
|
options_json, correct_answer, explanation, difficulty,
|
|
user_answer, is_correct, bookmarked, followup_session_id,
|
|
created_at, updated_at
|
|
)
|
|
SELECT
|
|
id, session_id, COALESCE(turn_id, ''), question_id, question,
|
|
question_type, options_json, correct_answer, explanation,
|
|
difficulty, user_answer, is_correct, bookmarked,
|
|
followup_session_id, created_at, updated_at
|
|
FROM notebook_entries;
|
|
|
|
DROP TABLE notebook_entries;
|
|
ALTER TABLE notebook_entries_new RENAME TO notebook_entries;
|
|
|
|
CREATE INDEX IF NOT EXISTS idx_notebook_entries_session
|
|
ON notebook_entries(session_id, created_at DESC);
|
|
|
|
CREATE INDEX IF NOT EXISTS idx_notebook_entries_bookmarked
|
|
ON notebook_entries(bookmarked, created_at DESC);
|
|
"""
|
|
)
|
|
|
|
@staticmethod
|
|
def _migrate_notebook_entries_add_user_answer_images(
|
|
conn: sqlite3.Connection,
|
|
) -> None:
|
|
"""Back-fill ``user_answer_images_json`` on legacy DBs.
|
|
|
|
The column stores a JSON array of ``{id, url, filename, mime_type}``
|
|
records for image attachments uploaded as part of the learner's
|
|
answer. The bytes themselves live in the AttachmentStore; we only
|
|
keep references in the row so notebook_entries stays lean.
|
|
"""
|
|
cols = {row[1] for row in conn.execute("PRAGMA table_info(notebook_entries)").fetchall()}
|
|
if not cols:
|
|
return
|
|
if "user_answer_images_json" not in cols:
|
|
conn.execute(
|
|
"ALTER TABLE notebook_entries ADD COLUMN user_answer_images_json TEXT DEFAULT '[]'"
|
|
)
|
|
|
|
@staticmethod
|
|
def _migrate_notebook_entries_add_ai_judgment(
|
|
conn: sqlite3.Connection,
|
|
) -> None:
|
|
"""Back-fill ``ai_judgment`` on legacy DBs.
|
|
|
|
Stores the latest AI-judge text per entry as plain markdown. Empty
|
|
string means the learner has not run the AI judge for this entry
|
|
yet.
|
|
"""
|
|
cols = {row[1] for row in conn.execute("PRAGMA table_info(notebook_entries)").fetchall()}
|
|
if not cols:
|
|
return
|
|
if "ai_judgment" not in cols:
|
|
conn.execute("ALTER TABLE notebook_entries ADD COLUMN ai_judgment TEXT DEFAULT ''")
|
|
|
|
async def _run(self, fn, *args):
|
|
async with self._lock:
|
|
return await asyncio.to_thread(fn, *args)
|
|
|
|
@contextmanager
|
|
def _connect(self) -> Iterator[sqlite3.Connection]:
|
|
# sqlite3.Connection's own context manager commits/rolls back but does
|
|
# NOT close the connection — so naked `with sqlite3.connect(...)` leaks
|
|
# one FD per call until GC. Wrap it so each call site gets both
|
|
# transaction semantics and deterministic close. The inner `with conn`
|
|
# commits on clean exit and rolls back on exception, so call sites do
|
|
# NOT need an explicit conn.commit() (any remaining ones are no-ops).
|
|
conn = sqlite3.connect(self.db_path)
|
|
conn.row_factory = sqlite3.Row
|
|
conn.execute("PRAGMA foreign_keys = ON")
|
|
try:
|
|
with conn:
|
|
yield conn
|
|
finally:
|
|
conn.close()
|
|
|
|
def _create_session_sync(
|
|
self,
|
|
title: str | None = None,
|
|
session_id: str | None = None,
|
|
) -> dict[str, Any]:
|
|
now = time.time()
|
|
resolved_id = session_id or f"unified_{int(now * 1000)}_{uuid.uuid4().hex[:8]}"
|
|
resolved_title = (title or "New conversation").strip() or "New conversation"
|
|
with self._connect() as conn:
|
|
conn.execute(
|
|
"""
|
|
INSERT INTO sessions (
|
|
id, title, created_at, updated_at,
|
|
compressed_summary, summary_up_to_msg_id
|
|
)
|
|
VALUES (?, ?, ?, ?, '', 0)
|
|
""",
|
|
(resolved_id, resolved_title[:100], now, now),
|
|
)
|
|
conn.commit()
|
|
return {
|
|
"id": resolved_id,
|
|
"session_id": resolved_id,
|
|
"title": resolved_title[:100],
|
|
"created_at": now,
|
|
"updated_at": now,
|
|
"compressed_summary": "",
|
|
"summary_up_to_msg_id": 0,
|
|
}
|
|
|
|
async def create_session(
|
|
self,
|
|
title: str | None = None,
|
|
session_id: str | None = None,
|
|
) -> dict[str, Any]:
|
|
return await self._run(self._create_session_sync, title, session_id)
|
|
|
|
def _get_session_sync(self, session_id: str) -> dict[str, Any] | None:
|
|
with self._connect() as conn:
|
|
row = conn.execute(
|
|
"""
|
|
SELECT
|
|
s.id,
|
|
s.title,
|
|
s.created_at,
|
|
s.updated_at,
|
|
s.compressed_summary,
|
|
s.summary_up_to_msg_id,
|
|
s.preferences_json,
|
|
COALESCE(
|
|
(
|
|
SELECT t.status
|
|
FROM turns t
|
|
WHERE t.session_id = s.id
|
|
ORDER BY t.updated_at DESC
|
|
LIMIT 1
|
|
),
|
|
'idle'
|
|
) AS status,
|
|
COALESCE(
|
|
(
|
|
SELECT t.id
|
|
FROM turns t
|
|
WHERE t.session_id = s.id AND t.status = 'running'
|
|
ORDER BY t.updated_at DESC
|
|
LIMIT 1
|
|
),
|
|
''
|
|
) AS active_turn_id,
|
|
COALESCE(
|
|
(
|
|
SELECT t.capability
|
|
FROM turns t
|
|
WHERE t.session_id = s.id
|
|
ORDER BY t.updated_at DESC
|
|
LIMIT 1
|
|
),
|
|
''
|
|
) AS capability
|
|
FROM sessions
|
|
s
|
|
WHERE s.id = ?
|
|
""",
|
|
(session_id,),
|
|
).fetchone()
|
|
if not row:
|
|
return None
|
|
payload = dict(row)
|
|
payload["session_id"] = payload["id"]
|
|
payload["preferences"] = _json_loads(payload.pop("preferences_json", ""), {})
|
|
return payload
|
|
|
|
async def get_session(self, session_id: str) -> dict[str, Any] | None:
|
|
return await self._run(self._get_session_sync, session_id)
|
|
|
|
async def ensure_session(
|
|
self,
|
|
session_id: str | None = None,
|
|
) -> dict[str, Any]:
|
|
if session_id:
|
|
session = await self.get_session(session_id)
|
|
if session is not None:
|
|
return session
|
|
return await self.create_session()
|
|
|
|
@staticmethod
|
|
def _serialize_turn(row: sqlite3.Row) -> dict[str, Any]:
|
|
return TurnRecord(
|
|
id=row["id"],
|
|
session_id=row["session_id"],
|
|
capability=row["capability"] or "",
|
|
status=row["status"] or "running",
|
|
error=row["error"] or "",
|
|
created_at=row["created_at"],
|
|
updated_at=row["updated_at"],
|
|
finished_at=row["finished_at"],
|
|
last_seq=row["last_seq"] if "last_seq" in row.keys() else 0,
|
|
).to_dict()
|
|
|
|
def _create_turn_sync(self, session_id: str, capability: str = "") -> dict[str, Any]:
|
|
now = time.time()
|
|
turn_id = f"turn_{int(now * 1000)}_{uuid.uuid4().hex[:10]}"
|
|
with self._connect() as conn:
|
|
session = conn.execute("SELECT id FROM sessions WHERE id = ?", (session_id,)).fetchone()
|
|
if session is None:
|
|
raise ValueError(f"Session not found: {session_id}")
|
|
active = conn.execute(
|
|
"""
|
|
SELECT id
|
|
FROM turns
|
|
WHERE session_id = ? AND status = 'running'
|
|
ORDER BY updated_at DESC
|
|
LIMIT 1
|
|
""",
|
|
(session_id,),
|
|
).fetchone()
|
|
if active is not None:
|
|
raise RuntimeError(f"Session already has an active turn: {active['id']}")
|
|
conn.execute(
|
|
"""
|
|
INSERT INTO turns (id, session_id, capability, status, error, created_at, updated_at, finished_at)
|
|
VALUES (?, ?, ?, 'running', '', ?, ?, NULL)
|
|
""",
|
|
(turn_id, session_id, capability or "", now, now),
|
|
)
|
|
conn.commit()
|
|
return {
|
|
"id": turn_id,
|
|
"turn_id": turn_id,
|
|
"session_id": session_id,
|
|
"capability": capability or "",
|
|
"status": "running",
|
|
"error": "",
|
|
"created_at": now,
|
|
"updated_at": now,
|
|
"finished_at": None,
|
|
"last_seq": 0,
|
|
}
|
|
|
|
async def create_turn(self, session_id: str, capability: str = "") -> dict[str, Any]:
|
|
return await self._run(self._create_turn_sync, session_id, capability)
|
|
|
|
def _get_turn_sync(self, turn_id: str) -> dict[str, Any] | None:
|
|
with self._connect() as conn:
|
|
row = conn.execute(
|
|
"""
|
|
SELECT
|
|
t.*,
|
|
COALESCE((SELECT MAX(seq) FROM turn_events te WHERE te.turn_id = t.id), 0) AS last_seq
|
|
FROM turns t
|
|
WHERE t.id = ?
|
|
""",
|
|
(turn_id,),
|
|
).fetchone()
|
|
if row is None:
|
|
return None
|
|
return self._serialize_turn(row)
|
|
|
|
async def get_turn(self, turn_id: str) -> dict[str, Any] | None:
|
|
return await self._run(self._get_turn_sync, turn_id)
|
|
|
|
def _get_active_turn_sync(self, session_id: str) -> dict[str, Any] | None:
|
|
with self._connect() as conn:
|
|
row = conn.execute(
|
|
"""
|
|
SELECT
|
|
t.*,
|
|
COALESCE((SELECT MAX(seq) FROM turn_events te WHERE te.turn_id = t.id), 0) AS last_seq
|
|
FROM turns t
|
|
WHERE t.session_id = ? AND t.status = 'running'
|
|
ORDER BY t.updated_at DESC
|
|
LIMIT 1
|
|
""",
|
|
(session_id,),
|
|
).fetchone()
|
|
if row is None:
|
|
return None
|
|
return self._serialize_turn(row)
|
|
|
|
async def get_active_turn(self, session_id: str) -> dict[str, Any] | None:
|
|
return await self._run(self._get_active_turn_sync, session_id)
|
|
|
|
def _list_active_turns_sync(self, session_id: str) -> list[dict[str, Any]]:
|
|
with self._connect() as conn:
|
|
rows = conn.execute(
|
|
"""
|
|
SELECT
|
|
t.*,
|
|
COALESCE((SELECT MAX(seq) FROM turn_events te WHERE te.turn_id = t.id), 0) AS last_seq
|
|
FROM turns t
|
|
WHERE t.session_id = ? AND t.status = 'running'
|
|
ORDER BY t.updated_at DESC
|
|
""",
|
|
(session_id,),
|
|
).fetchall()
|
|
return [self._serialize_turn(row) for row in rows]
|
|
|
|
async def list_active_turns(self, session_id: str) -> list[dict[str, Any]]:
|
|
return await self._run(self._list_active_turns_sync, session_id)
|
|
|
|
def _update_turn_status_sync(self, turn_id: str, status: str, error: str = "") -> bool:
|
|
now = time.time()
|
|
finished_at = now if status in {"completed", "failed", "cancelled"} else None
|
|
with self._connect() as conn:
|
|
cur = conn.execute(
|
|
"""
|
|
UPDATE turns
|
|
SET status = ?, error = ?, updated_at = ?, finished_at = ?
|
|
WHERE id = ?
|
|
""",
|
|
(status, error or "", now, finished_at, turn_id),
|
|
)
|
|
conn.commit()
|
|
return cur.rowcount > 0
|
|
|
|
async def update_turn_status(self, turn_id: str, status: str, error: str = "") -> bool:
|
|
return await self._run(self._update_turn_status_sync, turn_id, status, error)
|
|
|
|
def _append_turn_event_sync(self, turn_id: str, event: dict[str, Any]) -> dict[str, Any]:
|
|
now = time.time()
|
|
with self._connect() as conn:
|
|
turn = conn.execute(
|
|
"SELECT id, session_id FROM turns WHERE id = ?", (turn_id,)
|
|
).fetchone()
|
|
if turn is None:
|
|
raise ValueError(f"Turn not found: {turn_id}")
|
|
provided_seq = int(event.get("seq") or 0)
|
|
if provided_seq > 0:
|
|
seq = provided_seq
|
|
else:
|
|
row = conn.execute(
|
|
"SELECT COALESCE(MAX(seq), 0) AS last_seq FROM turn_events WHERE turn_id = ?",
|
|
(turn_id,),
|
|
).fetchone()
|
|
seq = int(row["last_seq"]) + 1 if row else 1
|
|
payload = dict(event)
|
|
payload["seq"] = seq
|
|
payload["turn_id"] = payload.get("turn_id") or turn_id
|
|
payload["session_id"] = payload.get("session_id") or turn["session_id"]
|
|
conn.execute(
|
|
"""
|
|
INSERT OR REPLACE INTO turn_events (
|
|
turn_id, seq, type, source, stage, content, metadata_json, timestamp, created_at
|
|
) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)
|
|
""",
|
|
(
|
|
turn_id,
|
|
seq,
|
|
payload.get("type", ""),
|
|
payload.get("source", ""),
|
|
payload.get("stage", ""),
|
|
payload.get("content", "") or "",
|
|
_json_dumps(payload.get("metadata", {})),
|
|
float(payload.get("timestamp") or now),
|
|
now,
|
|
),
|
|
)
|
|
conn.execute(
|
|
"UPDATE turns SET updated_at = ? WHERE id = ?",
|
|
(now, turn_id),
|
|
)
|
|
conn.commit()
|
|
return payload
|
|
|
|
async def append_turn_event(self, turn_id: str, event: dict[str, Any]) -> dict[str, Any]:
|
|
return await self._run(self._append_turn_event_sync, turn_id, event)
|
|
|
|
def _get_turn_events_sync(self, turn_id: str, after_seq: int = 0) -> list[dict[str, Any]]:
|
|
with self._connect() as conn:
|
|
rows = conn.execute(
|
|
"""
|
|
SELECT turn_id, seq, type, source, stage, content, metadata_json, timestamp
|
|
FROM turn_events
|
|
WHERE turn_id = ? AND seq > ?
|
|
ORDER BY seq ASC
|
|
""",
|
|
(turn_id, max(0, int(after_seq))),
|
|
).fetchall()
|
|
turn = conn.execute("SELECT session_id FROM turns WHERE id = ?", (turn_id,)).fetchone()
|
|
session_id = turn["session_id"] if turn else ""
|
|
return [
|
|
{
|
|
"type": row["type"],
|
|
"source": row["source"] or "",
|
|
"stage": row["stage"] or "",
|
|
"content": row["content"] or "",
|
|
"metadata": _json_loads(row["metadata_json"], {}),
|
|
"session_id": session_id,
|
|
"turn_id": row["turn_id"],
|
|
"seq": row["seq"],
|
|
"timestamp": row["timestamp"],
|
|
}
|
|
for row in rows
|
|
]
|
|
|
|
async def get_turn_events(self, turn_id: str, after_seq: int = 0) -> list[dict[str, Any]]:
|
|
return await self._run(self._get_turn_events_sync, turn_id, after_seq)
|
|
|
|
def _update_session_title_sync(self, session_id: str, title: str) -> bool:
|
|
with self._connect() as conn:
|
|
cur = conn.execute(
|
|
"""
|
|
UPDATE sessions
|
|
SET title = ?, updated_at = ?
|
|
WHERE id = ?
|
|
""",
|
|
((title.strip() or "New conversation")[:100], time.time(), session_id),
|
|
)
|
|
conn.commit()
|
|
return cur.rowcount > 0
|
|
|
|
async def update_session_title(self, session_id: str, title: str) -> bool:
|
|
return await self._run(self._update_session_title_sync, session_id, title)
|
|
|
|
def _delete_session_sync(self, session_id: str) -> bool:
|
|
with self._connect() as conn:
|
|
cur = conn.execute("DELETE FROM sessions WHERE id = ?", (session_id,))
|
|
conn.commit()
|
|
return cur.rowcount > 0
|
|
|
|
async def delete_session(self, session_id: str) -> bool:
|
|
return await self._run(self._delete_session_sync, session_id)
|
|
|
|
def _add_message_sync(
|
|
self,
|
|
session_id: str,
|
|
role: str,
|
|
content: str,
|
|
capability: str = "",
|
|
events: list[dict[str, Any]] | None = None,
|
|
attachments: list[dict[str, Any]] | None = None,
|
|
metadata: dict[str, Any] | None = None,
|
|
parent_message_id: int | None | _Unset = _PARENT_AUTO,
|
|
) -> int:
|
|
now = time.time()
|
|
with self._connect() as conn:
|
|
session = conn.execute(
|
|
"SELECT id, title FROM sessions WHERE id = ?", (session_id,)
|
|
).fetchone()
|
|
if session is None:
|
|
raise ValueError(f"Session not found: {session_id}")
|
|
|
|
resolved_parent_id: int | None
|
|
if isinstance(parent_message_id, _Unset):
|
|
# Legacy auto-append path: chain off the latest row in the
|
|
# session so the linear thread stays connected.
|
|
last_row = conn.execute(
|
|
"SELECT id FROM messages WHERE session_id = ? ORDER BY id DESC LIMIT 1",
|
|
(session_id,),
|
|
).fetchone()
|
|
resolved_parent_id = int(last_row["id"]) if last_row is not None else None
|
|
else:
|
|
# Caller pinned a parent explicitly — including ``None``,
|
|
# which means "attach at the session root" (used by edits
|
|
# of the very first message in a session).
|
|
resolved_parent_id = (
|
|
int(parent_message_id) if parent_message_id is not None else None
|
|
)
|
|
|
|
cur = conn.execute(
|
|
"""
|
|
INSERT INTO messages (
|
|
session_id, role, content, capability, events_json,
|
|
attachments_json, metadata_json, created_at, parent_message_id
|
|
) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)
|
|
""",
|
|
(
|
|
session_id,
|
|
role,
|
|
content or "",
|
|
capability or "",
|
|
_json_dumps(events or []),
|
|
_json_dumps(attachments or []),
|
|
_json_dumps(metadata or {}),
|
|
now,
|
|
resolved_parent_id,
|
|
),
|
|
)
|
|
|
|
# Title is no longer derived from the first user message — the
|
|
# turn runtime calls an LLM to generate a real summary title
|
|
# once the first user+assistant pair is complete. Until then
|
|
# the session keeps the default sentinel ``New conversation``
|
|
# which the frontend renders as a breathing "New chat" chip.
|
|
conn.execute(
|
|
"UPDATE sessions SET updated_at = ? WHERE id = ?",
|
|
(now, session_id),
|
|
)
|
|
conn.commit()
|
|
return int(cur.lastrowid)
|
|
|
|
async def add_message(
|
|
self,
|
|
session_id: str,
|
|
role: str,
|
|
content: str,
|
|
capability: str = "",
|
|
events: list[dict[str, Any]] | None = None,
|
|
attachments: list[dict[str, Any]] | None = None,
|
|
metadata: dict[str, Any] | None = None,
|
|
parent_message_id: int | None | _Unset = _PARENT_AUTO,
|
|
) -> int:
|
|
return await self._run(
|
|
self._add_message_sync,
|
|
session_id,
|
|
role,
|
|
content,
|
|
capability,
|
|
events,
|
|
attachments,
|
|
metadata,
|
|
parent_message_id,
|
|
)
|
|
|
|
@staticmethod
|
|
def _backfill_import_meta_sync(
|
|
conn: sqlite3.Connection,
|
|
session_id: str,
|
|
current_prefs_json: str | None,
|
|
incoming_prefs: dict[str, Any],
|
|
) -> bool:
|
|
"""Merge agent attribution from a re-import into an existing session's
|
|
``preferences.import`` block, leaving everything else untouched. Returns
|
|
whether anything changed (so the caller can skip a needless write)."""
|
|
incoming_import = (incoming_prefs or {}).get("import") or {}
|
|
if not incoming_import:
|
|
return False
|
|
prefs = _json_loads(current_prefs_json, {})
|
|
if not isinstance(prefs, dict):
|
|
prefs = {}
|
|
meta = dict(prefs.get("import") or {})
|
|
changed = False
|
|
# Only attribution fields propagate on re-import; source/external_id are
|
|
# part of the dedup identity and never change for a given session.
|
|
for key in ("agent_id", "agent_name", "source_cwd"):
|
|
value = incoming_import.get(key)
|
|
if value and meta.get(key) != value:
|
|
meta[key] = value
|
|
changed = True
|
|
if not changed:
|
|
return False
|
|
prefs["import"] = meta
|
|
conn.execute(
|
|
"UPDATE sessions SET preferences_json = ? WHERE id = ?",
|
|
(_json_dumps(prefs), session_id),
|
|
)
|
|
return True
|
|
|
|
def _import_session_sync(
|
|
self,
|
|
session_id: str,
|
|
title: str,
|
|
created_at: float,
|
|
updated_at: float,
|
|
preferences: dict[str, Any],
|
|
messages: list[dict[str, Any]],
|
|
) -> dict[str, Any]:
|
|
with self._connect() as conn:
|
|
existing = conn.execute(
|
|
"SELECT preferences_json FROM sessions WHERE id = ?", (session_id,)
|
|
).fetchone()
|
|
if existing is not None:
|
|
# Idempotent on content: a session imported before keeps its
|
|
# (possibly already-continued) state — re-importing the same
|
|
# folder never duplicates rows or clobbers the user's edits.
|
|
# We do, however, backfill agent attribution (agent_id /
|
|
# agent_name) so re-syncing re-tags conversations that were
|
|
# imported before the agent model existed, and an agent rename
|
|
# propagates. This only touches the ``import`` metadata block.
|
|
updated = self._backfill_import_meta_sync(
|
|
conn, session_id, existing["preferences_json"], preferences
|
|
)
|
|
if updated:
|
|
conn.commit()
|
|
return {
|
|
"session_id": session_id,
|
|
"imported": False,
|
|
"updated": updated,
|
|
"message_count": 0,
|
|
}
|
|
safe_title = (title or "").strip()[:100] or "Imported conversation"
|
|
conn.execute(
|
|
"""
|
|
INSERT INTO sessions (
|
|
id, title, created_at, updated_at,
|
|
compressed_summary, summary_up_to_msg_id, preferences_json
|
|
) VALUES (?, ?, ?, ?, '', 0, ?)
|
|
""",
|
|
(session_id, safe_title, created_at, updated_at, _json_dumps(preferences or {})),
|
|
)
|
|
prev_id: int | None = None
|
|
count = 0
|
|
for msg in messages:
|
|
cur = conn.execute(
|
|
"""
|
|
INSERT INTO messages (
|
|
session_id, role, content, capability, events_json,
|
|
attachments_json, metadata_json, created_at, parent_message_id
|
|
) VALUES (?, ?, ?, '', '[]', '[]', ?, ?, ?)
|
|
""",
|
|
(
|
|
session_id,
|
|
msg.get("role") or "user",
|
|
msg.get("content") or "",
|
|
_json_dumps(msg.get("metadata") or {}),
|
|
float(msg.get("created_at") or created_at),
|
|
prev_id,
|
|
),
|
|
)
|
|
prev_id = int(cur.lastrowid)
|
|
count += 1
|
|
conn.commit()
|
|
return {"session_id": session_id, "imported": True, "message_count": count}
|
|
|
|
async def import_session(
|
|
self,
|
|
session_id: str,
|
|
title: str,
|
|
created_at: float,
|
|
updated_at: float,
|
|
preferences: dict[str, Any] | None,
|
|
messages: list[dict[str, Any]],
|
|
) -> dict[str, Any]:
|
|
"""Persist a pre-existing conversation (imported from an external CLI
|
|
such as Claude Code or Codex) as a normal session, so the chat loop can
|
|
re-open and continue it. ``session_id`` must carry the ``imported_``
|
|
prefix (see :data:`_IMPORTED_ID_PREFIX`). Idempotent by id: a session
|
|
already present is left untouched.
|
|
"""
|
|
return await self._run(
|
|
self._import_session_sync,
|
|
session_id,
|
|
title,
|
|
created_at,
|
|
updated_at,
|
|
preferences or {},
|
|
messages,
|
|
)
|
|
|
|
def _delete_message_sync(self, message_id: int | str) -> bool:
|
|
with self._connect() as conn:
|
|
cur = conn.execute("DELETE FROM messages WHERE id = ?", (int(message_id),))
|
|
conn.commit()
|
|
return cur.rowcount > 0
|
|
|
|
async def delete_message(self, message_id: int | str) -> bool:
|
|
return await self._run(self._delete_message_sync, message_id)
|
|
|
|
def _delete_turn_by_message_sync(self, session_id: str, message_id: int) -> dict[str, Any]:
|
|
with self._connect() as conn:
|
|
msg = conn.execute(
|
|
"""
|
|
SELECT id, session_id, role, attachments_json, created_at
|
|
FROM messages
|
|
WHERE id = ?
|
|
""",
|
|
(int(message_id),),
|
|
).fetchone()
|
|
if msg is None or msg["session_id"] != session_id:
|
|
return {
|
|
"deleted": False,
|
|
"attachment_ids": [],
|
|
"turn_id": None,
|
|
"was_running": False,
|
|
}
|
|
|
|
role = msg["role"]
|
|
paired_msg = None
|
|
if role == "user":
|
|
paired_msg = conn.execute(
|
|
"""
|
|
SELECT id, session_id, role, attachments_json, created_at
|
|
FROM messages
|
|
WHERE session_id = ? AND role = 'assistant' AND id > ?
|
|
ORDER BY id ASC
|
|
LIMIT 1
|
|
""",
|
|
(session_id, int(message_id)),
|
|
).fetchone()
|
|
elif role == "assistant":
|
|
paired_msg = conn.execute(
|
|
"""
|
|
SELECT id, session_id, role, attachments_json, created_at
|
|
FROM messages
|
|
WHERE session_id = ? AND role = 'user' AND id < ?
|
|
ORDER BY id DESC
|
|
LIMIT 1
|
|
""",
|
|
(session_id, int(message_id)),
|
|
).fetchone()
|
|
|
|
user_msg = msg if role == "user" else paired_msg
|
|
turn_id = None
|
|
was_running = False
|
|
if user_msg is not None:
|
|
user_created_at = user_msg["created_at"]
|
|
turn_row = conn.execute(
|
|
"""
|
|
SELECT id, status
|
|
FROM turns
|
|
WHERE session_id = ? AND created_at >= ?
|
|
ORDER BY created_at ASC
|
|
LIMIT 1
|
|
""",
|
|
(session_id, user_created_at),
|
|
).fetchone()
|
|
if turn_row is not None:
|
|
turn_id = turn_row["id"]
|
|
was_running = turn_row["status"] == "running"
|
|
|
|
if was_running:
|
|
return {
|
|
"deleted": False,
|
|
"attachment_ids": [],
|
|
"turn_id": turn_id,
|
|
"was_running": True,
|
|
}
|
|
|
|
attachment_ids: list[str] = []
|
|
for m in [msg, paired_msg]:
|
|
if m is not None:
|
|
atts = _json_loads(m["attachments_json"], [])
|
|
for att in atts:
|
|
aid = att.get("id") or att.get("attachment_id")
|
|
if aid:
|
|
attachment_ids.append(aid)
|
|
|
|
if turn_id is not None:
|
|
conn.execute("DELETE FROM turn_events WHERE turn_id = ?", (turn_id,))
|
|
conn.execute("DELETE FROM turns WHERE id = ?", (turn_id,))
|
|
|
|
ids_to_delete = [int(message_id)]
|
|
if paired_msg is not None:
|
|
ids_to_delete.append(int(paired_msg["id"]))
|
|
conn.execute(
|
|
f"DELETE FROM messages WHERE id IN ({','.join('?' * len(ids_to_delete))})", # nosec B608
|
|
tuple(ids_to_delete),
|
|
)
|
|
|
|
session_row = conn.execute(
|
|
"SELECT summary_up_to_msg_id FROM sessions WHERE id = ?",
|
|
(session_id,),
|
|
).fetchone()
|
|
if session_row is not None:
|
|
summary_up_to = int(session_row["summary_up_to_msg_id"])
|
|
if any(mid <= summary_up_to for mid in ids_to_delete):
|
|
conn.execute(
|
|
"UPDATE sessions SET summary_up_to_msg_id = 0 WHERE id = ?",
|
|
(session_id,),
|
|
)
|
|
|
|
conn.execute(
|
|
"UPDATE sessions SET updated_at = ? WHERE id = ?",
|
|
(time.time(), session_id),
|
|
)
|
|
conn.commit()
|
|
|
|
return {
|
|
"deleted": True,
|
|
"attachment_ids": attachment_ids,
|
|
"turn_id": turn_id,
|
|
"was_running": was_running,
|
|
}
|
|
|
|
async def delete_turn_by_message(self, session_id: str, message_id: int) -> dict[str, Any]:
|
|
return await self._run(self._delete_turn_by_message_sync, session_id, message_id)
|
|
|
|
def _get_last_message_sync(
|
|
self, session_id: str, role: str | None = None
|
|
) -> dict[str, Any] | None:
|
|
with self._connect() as conn:
|
|
if role is None:
|
|
row = conn.execute(
|
|
"""
|
|
SELECT id, session_id, role, content, capability, events_json,
|
|
attachments_json, metadata_json, created_at, parent_message_id
|
|
FROM messages
|
|
WHERE session_id = ?
|
|
ORDER BY id DESC
|
|
LIMIT 1
|
|
""",
|
|
(session_id,),
|
|
).fetchone()
|
|
else:
|
|
row = conn.execute(
|
|
"""
|
|
SELECT id, session_id, role, content, capability, events_json,
|
|
attachments_json, metadata_json, created_at, parent_message_id
|
|
FROM messages
|
|
WHERE session_id = ? AND role = ?
|
|
ORDER BY id DESC
|
|
LIMIT 1
|
|
""",
|
|
(session_id, role),
|
|
).fetchone()
|
|
if row is None:
|
|
return None
|
|
return self._serialize_message(row)
|
|
|
|
async def get_last_message(
|
|
self, session_id: str, role: str | None = None
|
|
) -> dict[str, Any] | None:
|
|
return await self._run(self._get_last_message_sync, session_id, role)
|
|
|
|
def _serialize_message(self, row: sqlite3.Row) -> dict[str, Any]:
|
|
row_keys = row.keys()
|
|
parent_id = row["parent_message_id"] if "parent_message_id" in row_keys else None
|
|
return {
|
|
"id": row["id"],
|
|
"session_id": row["session_id"],
|
|
"role": row["role"],
|
|
"content": row["content"],
|
|
"capability": row["capability"] or "",
|
|
"events": _json_loads(row["events_json"], []),
|
|
"attachments": _json_loads(row["attachments_json"], []),
|
|
"metadata": _json_loads(row["metadata_json"], {}),
|
|
"created_at": row["created_at"],
|
|
"parent_message_id": int(parent_id) if parent_id is not None else None,
|
|
}
|
|
|
|
def _get_messages_sync(self, session_id: str) -> list[dict[str, Any]]:
|
|
with self._connect() as conn:
|
|
rows = conn.execute(
|
|
"""
|
|
SELECT id, session_id, role, content, capability, events_json,
|
|
attachments_json, metadata_json, created_at, parent_message_id
|
|
FROM messages
|
|
WHERE session_id = ?
|
|
ORDER BY id ASC
|
|
""",
|
|
(session_id,),
|
|
).fetchall()
|
|
return [self._serialize_message(row) for row in rows]
|
|
|
|
def _get_message_path_sync(self, session_id: str, leaf_message_id: int) -> list[dict[str, Any]]:
|
|
"""Return the chain of messages from the session root down to
|
|
``leaf_message_id`` (inclusive), in chronological order.
|
|
|
|
Used by the turn runtime to build LLM context for a branched
|
|
re-run: only ancestors of the new user message are included, so
|
|
sibling branches at any depth are excluded.
|
|
"""
|
|
with self._connect() as conn:
|
|
chain: list[dict[str, Any]] = []
|
|
current: int | None = int(leaf_message_id)
|
|
# Bound the walk defensively in case of corrupted parent pointers.
|
|
safety = 10_000
|
|
while current is not None and safety > 0:
|
|
row = conn.execute(
|
|
"""
|
|
SELECT id, session_id, role, content, capability, events_json,
|
|
attachments_json, metadata_json, created_at, parent_message_id
|
|
FROM messages
|
|
WHERE id = ? AND session_id = ?
|
|
""",
|
|
(current, session_id),
|
|
).fetchone()
|
|
if row is None:
|
|
break
|
|
chain.append(self._serialize_message(row))
|
|
parent = row["parent_message_id"]
|
|
current = int(parent) if parent is not None else None
|
|
safety -= 1
|
|
chain.reverse()
|
|
return chain
|
|
|
|
async def get_message_path(self, session_id: str, leaf_message_id: int) -> list[dict[str, Any]]:
|
|
return await self._run(self._get_message_path_sync, session_id, int(leaf_message_id))
|
|
|
|
async def get_messages(self, session_id: str) -> list[dict[str, Any]]:
|
|
return await self._run(self._get_messages_sync, session_id)
|
|
|
|
def _get_messages_for_context_sync(
|
|
self, session_id: str, leaf_message_id: int | None = None
|
|
) -> list[dict[str, Any]]:
|
|
with self._connect() as conn:
|
|
if leaf_message_id is None:
|
|
rows = conn.execute(
|
|
"""
|
|
SELECT id, role, content
|
|
FROM messages
|
|
WHERE session_id = ?
|
|
AND role IN ('user', 'assistant', 'system')
|
|
ORDER BY id ASC
|
|
""",
|
|
(session_id,),
|
|
).fetchall()
|
|
return [
|
|
{
|
|
"id": row["id"],
|
|
"role": row["role"],
|
|
"content": row["content"] or "",
|
|
}
|
|
for row in rows
|
|
]
|
|
# Branch-aware path walk: include only ancestors (+ leaf) so
|
|
# sibling branches at any depth are excluded from LLM context.
|
|
chain: list[dict[str, Any]] = []
|
|
current: int | None = int(leaf_message_id)
|
|
safety = 10_000
|
|
while current is not None and safety > 0:
|
|
row = conn.execute(
|
|
"""
|
|
SELECT id, role, content, parent_message_id
|
|
FROM messages
|
|
WHERE id = ? AND session_id = ?
|
|
AND role IN ('user', 'assistant', 'system')
|
|
""",
|
|
(current, session_id),
|
|
).fetchone()
|
|
if row is None:
|
|
break
|
|
chain.append(
|
|
{
|
|
"id": row["id"],
|
|
"role": row["role"],
|
|
"content": row["content"] or "",
|
|
}
|
|
)
|
|
parent = row["parent_message_id"]
|
|
current = int(parent) if parent is not None else None
|
|
safety -= 1
|
|
chain.reverse()
|
|
return chain
|
|
|
|
async def get_messages_for_context(
|
|
self, session_id: str, leaf_message_id: int | None = None
|
|
) -> list[dict[str, Any]]:
|
|
return await self._run(self._get_messages_for_context_sync, session_id, leaf_message_id)
|
|
|
|
# Imported conversations live in the same tables as native chats (so the
|
|
# chat loop can re-open and continue them) but carry an ``imported_`` id
|
|
# prefix. That prefix is the discriminator — it travels with the primary
|
|
# key, so we filter on it instead of adding a column + migration.
|
|
_SESSION_SUMMARY_SQL = """
|
|
SELECT
|
|
s.id,
|
|
s.title,
|
|
s.created_at,
|
|
s.updated_at,
|
|
s.compressed_summary,
|
|
s.summary_up_to_msg_id,
|
|
s.preferences_json,
|
|
COUNT(m.id) AS message_count,
|
|
COALESCE(
|
|
(SELECT t.status FROM turns t WHERE t.session_id = s.id
|
|
ORDER BY t.updated_at DESC LIMIT 1),
|
|
'idle'
|
|
) AS status,
|
|
COALESCE(
|
|
(SELECT t.id FROM turns t WHERE t.session_id = s.id AND t.status = 'running'
|
|
ORDER BY t.updated_at DESC LIMIT 1),
|
|
''
|
|
) AS active_turn_id,
|
|
COALESCE(
|
|
(SELECT t.capability FROM turns t WHERE t.session_id = s.id
|
|
ORDER BY t.updated_at DESC LIMIT 1),
|
|
''
|
|
) AS capability,
|
|
COALESCE(
|
|
(SELECT m2.content FROM messages m2
|
|
WHERE m2.session_id = s.id AND TRIM(COALESCE(m2.content, '')) != ''
|
|
ORDER BY m2.id DESC LIMIT 1),
|
|
''
|
|
) AS last_message
|
|
FROM sessions s
|
|
LEFT JOIN messages m ON m.session_id = s.id
|
|
{where}
|
|
GROUP BY s.id
|
|
ORDER BY s.updated_at DESC
|
|
LIMIT ? OFFSET ?
|
|
"""
|
|
|
|
# ``ESCAPE '\'`` makes the underscore in ``imported_`` literal rather than
|
|
# the LIKE single-char wildcard.
|
|
_WHERE_NATIVE = r"WHERE s.id NOT LIKE 'imported\_%' ESCAPE '\'"
|
|
_WHERE_IMPORTED = r"WHERE s.id LIKE 'imported\_%' ESCAPE '\'"
|
|
|
|
def _list_session_summaries_sync(
|
|
self, where_sql: str, limit: int, offset: int
|
|
) -> list[dict[str, Any]]:
|
|
with self._connect() as conn:
|
|
rows = conn.execute(
|
|
self._SESSION_SUMMARY_SQL.format(where=where_sql),
|
|
(limit, offset),
|
|
).fetchall()
|
|
sessions = []
|
|
for row in rows:
|
|
payload = dict(row)
|
|
payload["session_id"] = payload["id"]
|
|
payload["preferences"] = _json_loads(payload.pop("preferences_json", ""), {})
|
|
sessions.append(payload)
|
|
return sessions
|
|
|
|
def _list_sessions_sync(
|
|
self,
|
|
limit: int = 50,
|
|
offset: int = 0,
|
|
) -> list[dict[str, Any]]:
|
|
# Native chats only — imported histories surface under their own
|
|
# Space category, not the regular history list.
|
|
return self._list_session_summaries_sync(self._WHERE_NATIVE, limit, offset)
|
|
|
|
def _list_imported_sessions_sync(
|
|
self,
|
|
limit: int = 50,
|
|
offset: int = 0,
|
|
) -> list[dict[str, Any]]:
|
|
return self._list_session_summaries_sync(self._WHERE_IMPORTED, limit, offset)
|
|
|
|
async def list_sessions(
|
|
self,
|
|
limit: int = 50,
|
|
offset: int = 0,
|
|
) -> list[dict[str, Any]]:
|
|
return await self._run(self._list_sessions_sync, limit, offset)
|
|
|
|
async def list_imported_sessions(
|
|
self,
|
|
limit: int = 50,
|
|
offset: int = 0,
|
|
) -> list[dict[str, Any]]:
|
|
return await self._run(self._list_imported_sessions_sync, limit, offset)
|
|
|
|
def _update_summary_sync(self, session_id: str, summary: str, up_to_msg_id: int) -> bool:
|
|
with self._connect() as conn:
|
|
cur = conn.execute(
|
|
"""
|
|
UPDATE sessions
|
|
SET compressed_summary = ?, summary_up_to_msg_id = ?, updated_at = updated_at
|
|
WHERE id = ?
|
|
""",
|
|
(summary, max(0, int(up_to_msg_id)), session_id),
|
|
)
|
|
conn.commit()
|
|
return cur.rowcount > 0
|
|
|
|
async def update_summary(self, session_id: str, summary: str, up_to_msg_id: int) -> bool:
|
|
return await self._run(self._update_summary_sync, session_id, summary, up_to_msg_id)
|
|
|
|
def _update_session_preferences_sync(
|
|
self, session_id: str, preferences: dict[str, Any]
|
|
) -> bool:
|
|
with self._connect() as conn:
|
|
current = conn.execute(
|
|
"SELECT preferences_json FROM sessions WHERE id = ?",
|
|
(session_id,),
|
|
).fetchone()
|
|
if current is None:
|
|
return False
|
|
merged = {
|
|
**_json_loads(current["preferences_json"], {}),
|
|
**(preferences or {}),
|
|
}
|
|
cur = conn.execute(
|
|
"""
|
|
UPDATE sessions
|
|
SET preferences_json = ?, updated_at = ?
|
|
WHERE id = ?
|
|
""",
|
|
(_json_dumps(merged), time.time(), session_id),
|
|
)
|
|
conn.commit()
|
|
return cur.rowcount > 0
|
|
|
|
async def update_session_preferences(
|
|
self, session_id: str, preferences: dict[str, Any]
|
|
) -> bool:
|
|
return await self._run(self._update_session_preferences_sync, session_id, preferences)
|
|
|
|
async def get_session_with_messages(self, session_id: str) -> dict[str, Any] | None:
|
|
session = await self.get_session(session_id)
|
|
if session is None:
|
|
return None
|
|
session["messages"] = await self.get_messages(session_id)
|
|
session["active_turns"] = await self.list_active_turns(session_id)
|
|
return session
|
|
|
|
# ── Notebook entries ──────────────────────────────────────────────
|
|
|
|
def _upsert_notebook_entries_sync(self, session_id: str, items: list[dict[str, Any]]) -> int:
|
|
if not items:
|
|
return 0
|
|
now = time.time()
|
|
with self._connect() as conn:
|
|
if (
|
|
conn.execute("SELECT id FROM sessions WHERE id = ?", (session_id,)).fetchone()
|
|
is None
|
|
):
|
|
raise ValueError(f"Session not found: {session_id}")
|
|
upserted = 0
|
|
for item in items:
|
|
question = (item.get("question") or "").strip()
|
|
question_id = (item.get("question_id") or "").strip()
|
|
if not question or not question_id:
|
|
continue
|
|
turn_id = (item.get("turn_id") or "").strip()
|
|
# ``user_answer_images`` is an optional list of records
|
|
# ``[{id, url, filename, mime_type}, …]``. We serialise it
|
|
# here so callers that only know about text don't need to
|
|
# know JSON. ``None`` keeps the existing column value on
|
|
# UPDATE (avoid clobbering stored images on a partial
|
|
# upsert that only changes ``is_correct``).
|
|
images_value = item.get("user_answer_images")
|
|
images_json = _json_dumps(images_value) if isinstance(images_value, list) else None
|
|
if images_json is None:
|
|
conn.execute(
|
|
"""
|
|
INSERT INTO notebook_entries (
|
|
session_id, turn_id, question_id, question, question_type,
|
|
options_json, correct_answer, explanation, difficulty,
|
|
user_answer, user_answer_images_json, is_correct,
|
|
bookmarked, followup_session_id, created_at, updated_at
|
|
) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, '[]', ?, 0, '', ?, ?)
|
|
ON CONFLICT(session_id, turn_id, question_id) DO UPDATE SET
|
|
user_answer = excluded.user_answer,
|
|
is_correct = excluded.is_correct,
|
|
updated_at = excluded.updated_at
|
|
""",
|
|
(
|
|
session_id,
|
|
turn_id,
|
|
question_id,
|
|
question,
|
|
item.get("question_type") or "",
|
|
_json_dumps(item.get("options") or {}),
|
|
item.get("correct_answer") or "",
|
|
item.get("explanation") or "",
|
|
item.get("difficulty") or "",
|
|
item.get("user_answer") or "",
|
|
1 if item.get("is_correct") else 0,
|
|
now,
|
|
now,
|
|
),
|
|
)
|
|
else:
|
|
conn.execute(
|
|
"""
|
|
INSERT INTO notebook_entries (
|
|
session_id, turn_id, question_id, question, question_type,
|
|
options_json, correct_answer, explanation, difficulty,
|
|
user_answer, user_answer_images_json, is_correct,
|
|
bookmarked, followup_session_id, created_at, updated_at
|
|
) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, 0, '', ?, ?)
|
|
ON CONFLICT(session_id, turn_id, question_id) DO UPDATE SET
|
|
user_answer = excluded.user_answer,
|
|
user_answer_images_json = excluded.user_answer_images_json,
|
|
is_correct = excluded.is_correct,
|
|
updated_at = excluded.updated_at
|
|
""",
|
|
(
|
|
session_id,
|
|
turn_id,
|
|
question_id,
|
|
question,
|
|
item.get("question_type") or "",
|
|
_json_dumps(item.get("options") or {}),
|
|
item.get("correct_answer") or "",
|
|
item.get("explanation") or "",
|
|
item.get("difficulty") or "",
|
|
item.get("user_answer") or "",
|
|
images_json,
|
|
1 if item.get("is_correct") else 0,
|
|
now,
|
|
now,
|
|
),
|
|
)
|
|
upserted += 1
|
|
conn.commit()
|
|
return upserted
|
|
|
|
async def upsert_notebook_entries(self, session_id: str, items: list[dict[str, Any]]) -> int:
|
|
return await self._run(self._upsert_notebook_entries_sync, session_id, items)
|
|
|
|
@staticmethod
|
|
def _serialize_notebook_entry(row: sqlite3.Row) -> dict[str, Any]:
|
|
keys = set(row.keys())
|
|
images: list[dict[str, Any]] = []
|
|
if "user_answer_images_json" in keys:
|
|
raw_images = _json_loads(row["user_answer_images_json"], [])
|
|
if isinstance(raw_images, list):
|
|
images = [r for r in raw_images if isinstance(r, dict)]
|
|
return {
|
|
"id": int(row["id"]),
|
|
"session_id": row["session_id"],
|
|
"session_title": row["session_title"] or "" if "session_title" in keys else "",
|
|
"turn_id": (row["turn_id"] or "") if "turn_id" in keys else "",
|
|
"question_id": row["question_id"] or "",
|
|
"question": row["question"],
|
|
"question_type": row["question_type"] or "",
|
|
"options": _json_loads(row["options_json"], {}),
|
|
"correct_answer": row["correct_answer"] or "",
|
|
"explanation": row["explanation"] or "",
|
|
"difficulty": row["difficulty"] or "",
|
|
"user_answer": row["user_answer"] or "",
|
|
"user_answer_images": images,
|
|
"is_correct": bool(row["is_correct"]),
|
|
"bookmarked": bool(row["bookmarked"]),
|
|
"followup_session_id": row["followup_session_id"] or "",
|
|
"ai_judgment": (row["ai_judgment"] or "") if "ai_judgment" in keys else "",
|
|
"created_at": float(row["created_at"]),
|
|
"updated_at": float(row["updated_at"]),
|
|
}
|
|
|
|
def _list_notebook_entries_sync(
|
|
self,
|
|
category_id: int | None,
|
|
bookmarked: bool | None,
|
|
is_correct: bool | None,
|
|
limit: int,
|
|
offset: int,
|
|
session_id: str | None = None,
|
|
) -> dict[str, Any]:
|
|
base = """
|
|
SELECT
|
|
n.id, n.session_id, COALESCE(s.title, '') AS session_title,
|
|
n.turn_id, n.question_id, n.question, n.question_type, n.options_json,
|
|
n.correct_answer, n.explanation, n.difficulty,
|
|
n.user_answer, n.user_answer_images_json, n.is_correct, n.bookmarked,
|
|
n.followup_session_id, n.ai_judgment, n.created_at, n.updated_at
|
|
FROM notebook_entries n
|
|
LEFT JOIN sessions s ON s.id = n.session_id
|
|
"""
|
|
count_base = "SELECT COUNT(*) AS cnt FROM notebook_entries n"
|
|
conditions: list[str] = []
|
|
params: list[Any] = []
|
|
if category_id is not None:
|
|
join = " INNER JOIN notebook_entry_categories ec ON ec.entry_id = n.id"
|
|
base += join
|
|
count_base += join
|
|
conditions.append("ec.category_id = ?")
|
|
params.append(category_id)
|
|
if bookmarked is not None:
|
|
conditions.append("n.bookmarked = ?")
|
|
params.append(1 if bookmarked else 0)
|
|
if is_correct is not None:
|
|
conditions.append("n.is_correct = ?")
|
|
params.append(1 if is_correct else 0)
|
|
if session_id is not None:
|
|
conditions.append("n.session_id = ?")
|
|
params.append(session_id)
|
|
where = (" WHERE " + " AND ".join(conditions)) if conditions else ""
|
|
with self._connect() as conn:
|
|
total_row = conn.execute(count_base + where, tuple(params)).fetchone()
|
|
total = int(total_row["cnt"]) if total_row else 0
|
|
rows = conn.execute(
|
|
base + where + " ORDER BY n.created_at DESC LIMIT ? OFFSET ?",
|
|
tuple(params) + (limit, offset),
|
|
).fetchall()
|
|
items = [self._serialize_notebook_entry(r) for r in rows]
|
|
return {"items": items, "total": total}
|
|
|
|
async def list_notebook_entries(
|
|
self,
|
|
category_id: int | None = None,
|
|
bookmarked: bool | None = None,
|
|
is_correct: bool | None = None,
|
|
limit: int = 50,
|
|
offset: int = 0,
|
|
*,
|
|
session_id: str | None = None,
|
|
) -> dict[str, Any]:
|
|
return await self._run(
|
|
self._list_notebook_entries_sync,
|
|
category_id,
|
|
bookmarked,
|
|
is_correct,
|
|
limit,
|
|
offset,
|
|
session_id,
|
|
)
|
|
|
|
def _get_notebook_entry_sync(self, entry_id: int) -> dict[str, Any] | None:
|
|
with self._connect() as conn:
|
|
row = conn.execute(
|
|
"""
|
|
SELECT
|
|
n.*, COALESCE(s.title, '') AS session_title
|
|
FROM notebook_entries n
|
|
LEFT JOIN sessions s ON s.id = n.session_id
|
|
WHERE n.id = ?
|
|
""",
|
|
(entry_id,),
|
|
).fetchone()
|
|
if row is None:
|
|
return None
|
|
entry = self._serialize_notebook_entry(row)
|
|
cats = conn.execute(
|
|
"""
|
|
SELECT c.id, c.name
|
|
FROM notebook_categories c
|
|
INNER JOIN notebook_entry_categories ec ON ec.category_id = c.id
|
|
WHERE ec.entry_id = ?
|
|
ORDER BY c.name
|
|
""",
|
|
(entry_id,),
|
|
).fetchall()
|
|
entry["categories"] = [{"id": c["id"], "name": c["name"]} for c in cats]
|
|
return entry
|
|
|
|
async def get_notebook_entry(self, entry_id: int) -> dict[str, Any] | None:
|
|
return await self._run(self._get_notebook_entry_sync, entry_id)
|
|
|
|
def _find_notebook_entry_sync(
|
|
self,
|
|
session_id: str,
|
|
question_id: str,
|
|
turn_id: str | None = None,
|
|
) -> dict[str, Any] | None:
|
|
with self._connect() as conn:
|
|
if turn_id is not None:
|
|
row = conn.execute(
|
|
"""
|
|
SELECT n.*, COALESCE(s.title, '') AS session_title
|
|
FROM notebook_entries n
|
|
LEFT JOIN sessions s ON s.id = n.session_id
|
|
WHERE n.session_id = ?
|
|
AND n.turn_id = ?
|
|
AND n.question_id = ?
|
|
""",
|
|
(session_id, turn_id, question_id),
|
|
).fetchone()
|
|
else:
|
|
# Legacy lookup: return the most recent matching entry across
|
|
# turns. Two quizzes in the same session can share a question_id
|
|
# (positional IDs like ``q_1``), so we explicitly pick the
|
|
# newest one to keep behavior deterministic for callers that
|
|
# don't yet pass a turn_id.
|
|
row = conn.execute(
|
|
"""
|
|
SELECT n.*, COALESCE(s.title, '') AS session_title
|
|
FROM notebook_entries n
|
|
LEFT JOIN sessions s ON s.id = n.session_id
|
|
WHERE n.session_id = ? AND n.question_id = ?
|
|
ORDER BY n.updated_at DESC, n.id DESC
|
|
LIMIT 1
|
|
""",
|
|
(session_id, question_id),
|
|
).fetchone()
|
|
if row is None:
|
|
return None
|
|
return self._serialize_notebook_entry(row)
|
|
|
|
async def find_notebook_entry(
|
|
self,
|
|
session_id: str,
|
|
question_id: str,
|
|
turn_id: str | None = None,
|
|
) -> dict[str, Any] | None:
|
|
return await self._run(self._find_notebook_entry_sync, session_id, question_id, turn_id)
|
|
|
|
def _update_notebook_entry_sync(self, entry_id: int, updates: dict[str, Any]) -> bool:
|
|
allowed = {
|
|
"bookmarked",
|
|
"followup_session_id",
|
|
"user_answer",
|
|
"is_correct",
|
|
"ai_judgment",
|
|
}
|
|
fields = {k: v for k, v in updates.items() if k in allowed}
|
|
if not fields:
|
|
return False
|
|
fields["updated_at"] = time.time()
|
|
if "bookmarked" in fields:
|
|
fields["bookmarked"] = 1 if fields["bookmarked"] else 0
|
|
if "is_correct" in fields:
|
|
fields["is_correct"] = 1 if fields["is_correct"] else 0
|
|
set_clause = ", ".join(f"{k} = ?" for k in fields)
|
|
values = list(fields.values()) + [entry_id]
|
|
with self._connect() as conn:
|
|
cur = conn.execute(
|
|
f"UPDATE notebook_entries SET {set_clause} WHERE id = ?", # nosec B608
|
|
tuple(values),
|
|
)
|
|
conn.commit()
|
|
return cur.rowcount > 0
|
|
|
|
async def update_notebook_entry(self, entry_id: int, updates: dict[str, Any]) -> bool:
|
|
return await self._run(self._update_notebook_entry_sync, entry_id, updates)
|
|
|
|
def _delete_notebook_entry_sync(self, entry_id: int) -> bool:
|
|
with self._connect() as conn:
|
|
cur = conn.execute("DELETE FROM notebook_entries WHERE id = ?", (entry_id,))
|
|
conn.commit()
|
|
return cur.rowcount > 0
|
|
|
|
async def delete_notebook_entry(self, entry_id: int) -> bool:
|
|
return await self._run(self._delete_notebook_entry_sync, entry_id)
|
|
|
|
# ── Notebook categories ────────────────────────────────────────
|
|
|
|
def _create_category_sync(self, name: str) -> dict[str, Any]:
|
|
now = time.time()
|
|
with self._connect() as conn:
|
|
cur = conn.execute(
|
|
"INSERT INTO notebook_categories (name, created_at) VALUES (?, ?)",
|
|
(name.strip(), now),
|
|
)
|
|
conn.commit()
|
|
return {"id": int(cur.lastrowid), "name": name.strip(), "created_at": now}
|
|
|
|
async def create_category(self, name: str) -> dict[str, Any]:
|
|
return await self._run(self._create_category_sync, name)
|
|
|
|
def _list_categories_sync(self) -> list[dict[str, Any]]:
|
|
with self._connect() as conn:
|
|
rows = conn.execute(
|
|
"""
|
|
SELECT c.id, c.name, c.created_at,
|
|
COUNT(ec.entry_id) AS entry_count
|
|
FROM notebook_categories c
|
|
LEFT JOIN notebook_entry_categories ec ON ec.category_id = c.id
|
|
GROUP BY c.id
|
|
ORDER BY c.name
|
|
""",
|
|
).fetchall()
|
|
return [
|
|
{
|
|
"id": r["id"],
|
|
"name": r["name"],
|
|
"created_at": float(r["created_at"]),
|
|
"entry_count": int(r["entry_count"]),
|
|
}
|
|
for r in rows
|
|
]
|
|
|
|
async def list_categories(self) -> list[dict[str, Any]]:
|
|
return await self._run(self._list_categories_sync)
|
|
|
|
def _rename_category_sync(self, category_id: int, name: str) -> bool:
|
|
with self._connect() as conn:
|
|
cur = conn.execute(
|
|
"UPDATE notebook_categories SET name = ? WHERE id = ?",
|
|
(name.strip(), category_id),
|
|
)
|
|
conn.commit()
|
|
return cur.rowcount > 0
|
|
|
|
async def rename_category(self, category_id: int, name: str) -> bool:
|
|
return await self._run(self._rename_category_sync, category_id, name)
|
|
|
|
def _delete_category_sync(self, category_id: int) -> bool:
|
|
with self._connect() as conn:
|
|
cur = conn.execute("DELETE FROM notebook_categories WHERE id = ?", (category_id,))
|
|
conn.commit()
|
|
return cur.rowcount > 0
|
|
|
|
async def delete_category(self, category_id: int) -> bool:
|
|
return await self._run(self._delete_category_sync, category_id)
|
|
|
|
def _add_entry_to_category_sync(self, entry_id: int, category_id: int) -> bool:
|
|
with self._connect() as conn:
|
|
try:
|
|
conn.execute(
|
|
"INSERT OR IGNORE INTO notebook_entry_categories (entry_id, category_id) VALUES (?, ?)",
|
|
(entry_id, category_id),
|
|
)
|
|
conn.commit()
|
|
except sqlite3.IntegrityError:
|
|
return False
|
|
return True
|
|
|
|
async def add_entry_to_category(self, entry_id: int, category_id: int) -> bool:
|
|
return await self._run(self._add_entry_to_category_sync, entry_id, category_id)
|
|
|
|
def _remove_entry_from_category_sync(self, entry_id: int, category_id: int) -> bool:
|
|
with self._connect() as conn:
|
|
cur = conn.execute(
|
|
"DELETE FROM notebook_entry_categories WHERE entry_id = ? AND category_id = ?",
|
|
(entry_id, category_id),
|
|
)
|
|
conn.commit()
|
|
return cur.rowcount > 0
|
|
|
|
async def remove_entry_from_category(self, entry_id: int, category_id: int) -> bool:
|
|
return await self._run(self._remove_entry_from_category_sync, entry_id, category_id)
|
|
|
|
def _get_entry_categories_sync(self, entry_id: int) -> list[dict[str, Any]]:
|
|
with self._connect() as conn:
|
|
rows = conn.execute(
|
|
"""
|
|
SELECT c.id, c.name FROM notebook_categories c
|
|
INNER JOIN notebook_entry_categories ec ON ec.category_id = c.id
|
|
WHERE ec.entry_id = ?
|
|
ORDER BY c.name
|
|
""",
|
|
(entry_id,),
|
|
).fetchall()
|
|
return [{"id": r["id"], "name": r["name"]} for r in rows]
|
|
|
|
async def get_entry_categories(self, entry_id: int) -> list[dict[str, Any]]:
|
|
return await self._run(self._get_entry_categories_sync, entry_id)
|
|
|
|
|
|
_instances: dict[str, SQLiteSessionStore] = {}
|
|
|
|
|
|
def get_sqlite_session_store() -> SQLiteSessionStore:
|
|
db_path = get_path_service().get_chat_history_db().resolve()
|
|
key = str(db_path)
|
|
if key not in _instances:
|
|
_instances[key] = SQLiteSessionStore(db_path=db_path)
|
|
return _instances[key]
|
|
|
|
|
|
__all__ = ["SQLiteSessionStore", "get_sqlite_session_store", "make_imported_session_id"]
|