refactor: split SessionDB into Search/Schema/Portability mixins (mechanical move, ~2.9K LOC out of hermes_state.py)
This commit is contained in:
+30
-3733
File diff suppressed because it is too large
Load Diff
@@ -0,0 +1,523 @@
|
|||||||
|
"""Shared module-level constants for the SessionDB family of modules.
|
||||||
|
|
||||||
|
Extracted verbatim from hermes_state.py so the SessionDB mixin modules
|
||||||
|
(hermes_state_search / hermes_state_schema / hermes_state_portability) can
|
||||||
|
reference them without importing hermes_state (which would be a cycle).
|
||||||
|
hermes_state re-imports every name here for backward compatibility.
|
||||||
|
"""
|
||||||
|
|
||||||
|
from typing import Any
|
||||||
|
|
||||||
|
from agent.skill_commands import (
|
||||||
|
SKILL_EXCERPT_JOINT,
|
||||||
|
SKILL_SCAFFOLD_SQL_LIKE,
|
||||||
|
describe_skill_invocation,
|
||||||
|
)
|
||||||
|
|
||||||
|
|
||||||
|
# Session preview = the head of the first user message, shown wherever a
|
||||||
|
# session has no title (sidebar rows, pickers, exports, the desktop's
|
||||||
|
# `sessionTitle` fallback).
|
||||||
|
#
|
||||||
|
# A /skill invocation expands into a message that embeds the whole skill body,
|
||||||
|
# so the plain head of it previews the SKILL's opening prose as if the user had
|
||||||
|
# written it. Scaffolded rows therefore carry a wider excerpt so
|
||||||
|
# ``_shape_preview`` can hand it to ``describe_skill_invocation`` and recover
|
||||||
|
# ``/work — fix the title leak``: the whole message while it stays under the
|
||||||
|
# budget, and head + tail (where the typed instruction lands) once it doesn't.
|
||||||
|
_PREVIEW_HEAD_CHARS = 63
|
||||||
|
|
||||||
|
|
||||||
|
_PREVIEW_SCAFFOLD_WINDOW = 400
|
||||||
|
|
||||||
|
|
||||||
|
_PREVIEW_MAX_CHARS = 60
|
||||||
|
|
||||||
|
|
||||||
|
_PREVIEW_CONTENT_SQL = "REPLACE(REPLACE(m.content, X'0A', ' '), X'0D', ' ')"
|
||||||
|
|
||||||
|
|
||||||
|
_PREVIEW_SCAFFOLDED_SQL = f"m.content LIKE '{SKILL_SCAFFOLD_SQL_LIKE}'"
|
||||||
|
|
||||||
|
|
||||||
|
# The shared ``_preview_raw`` SELECT expression, interpolated by every listing
|
||||||
|
# query. A scaffolded row gets a wider excerpt: the whole message while it fits
|
||||||
|
# the budget, else head + tail (where the typed instruction lands) spliced
|
||||||
|
# around SKILL_EXCERPT_JOINT.
|
||||||
|
_PREVIEW_RAW_SELECT = (
|
||||||
|
f"CASE WHEN {_PREVIEW_SCAFFOLDED_SQL}"
|
||||||
|
f" AND LENGTH(m.content) > {_PREVIEW_SCAFFOLD_WINDOW * 2}"
|
||||||
|
f" THEN SUBSTR({_PREVIEW_CONTENT_SQL}, 1, {_PREVIEW_SCAFFOLD_WINDOW})"
|
||||||
|
f" || '{SKILL_EXCERPT_JOINT}'"
|
||||||
|
f" || SUBSTR({_PREVIEW_CONTENT_SQL}, -{_PREVIEW_SCAFFOLD_WINDOW})"
|
||||||
|
f" WHEN {_PREVIEW_SCAFFOLDED_SQL}"
|
||||||
|
f" THEN SUBSTR({_PREVIEW_CONTENT_SQL}, 1, {_PREVIEW_SCAFFOLD_WINDOW * 2})"
|
||||||
|
f" ELSE SUBSTR({_PREVIEW_CONTENT_SQL}, 1, {_PREVIEW_HEAD_CHARS}) END"
|
||||||
|
)
|
||||||
|
|
||||||
|
|
||||||
|
def _shape_preview(raw: Any) -> str:
|
||||||
|
"""Turn a ``_preview_raw`` column into the short preview callers show."""
|
||||||
|
text = str(raw or "").strip()
|
||||||
|
if not text:
|
||||||
|
return ""
|
||||||
|
described = describe_skill_invocation(text)
|
||||||
|
text = described if described is not None else text.split(SKILL_EXCERPT_JOINT)[0]
|
||||||
|
if len(text) > _PREVIEW_MAX_CHARS:
|
||||||
|
return text[:_PREVIEW_MAX_CHARS] + "..."
|
||||||
|
return text
|
||||||
|
|
||||||
|
|
||||||
|
# A child session counts as a /branch (kept visible, never cascade-deleted) if
|
||||||
|
# it carries the stable marker OR the legacy end_reason heuristic holds.
|
||||||
|
_BRANCH_CHILD_SQL = (
|
||||||
|
"json_extract(COALESCE({a}.model_config, '{{}}'), '$._branched_from') IS NOT NULL"
|
||||||
|
" OR EXISTS (SELECT 1 FROM sessions p"
|
||||||
|
" WHERE p.id = {a}.parent_session_id"
|
||||||
|
" AND p.end_reason = 'branched'"
|
||||||
|
" AND {a}.started_at >= p.ended_at)"
|
||||||
|
)
|
||||||
|
|
||||||
|
|
||||||
|
_COMPRESSION_CHILD_SQL = (
|
||||||
|
"EXISTS (SELECT 1 FROM sessions p"
|
||||||
|
" WHERE p.id = {a}.parent_session_id"
|
||||||
|
" AND p.end_reason = 'compression')"
|
||||||
|
)
|
||||||
|
|
||||||
|
|
||||||
|
# Rows that surface in pickers: roots + branch children (subagent runs and
|
||||||
|
# compression continuations stay hidden).
|
||||||
|
_LISTABLE_CHILD_SQL = f"(s.parent_session_id IS NULL OR {_BRANCH_CHILD_SQL.format(a='s')})"
|
||||||
|
|
||||||
|
|
||||||
|
def _ephemeral_child_sql(alias: str = "s") -> str:
|
||||||
|
"""Subagent runs (cascade-delete targets), not branches or compression tips."""
|
||||||
|
branch = _BRANCH_CHILD_SQL.format(a=alias)
|
||||||
|
compression = _COMPRESSION_CHILD_SQL.format(a=alias)
|
||||||
|
return (
|
||||||
|
f"({alias}.parent_session_id IS NOT NULL"
|
||||||
|
f" AND NOT ({branch})"
|
||||||
|
f" AND NOT ({compression}))"
|
||||||
|
)
|
||||||
|
|
||||||
|
|
||||||
|
SCHEMA_VERSION = 23
|
||||||
|
|
||||||
|
|
||||||
|
# FTS storage-layout version, tracked INDEPENDENTLY of SCHEMA_VERSION in the
|
||||||
|
# state_meta key ``fts_storage_version``. The main schema version advances
|
||||||
|
# freely on open (so future migrations always land); the FTS *layout* only
|
||||||
|
# reaches the current version when a DB is either born fresh or explicitly
|
||||||
|
# optimized via ``hermes sessions optimize-storage``. A legacy DB sits at
|
||||||
|
# layout 0 (marker absent) with a working inline index until the user opts in.
|
||||||
|
# 1 = v23 external-content layout (content/tool_name/tool_calls,
|
||||||
|
# tool-row-excluded trigram)
|
||||||
|
FTS_STORAGE_VERSION = 1
|
||||||
|
|
||||||
|
|
||||||
|
# Cap on user-controlled FTS5 query input before regex/sanitizer processing.
|
||||||
|
# Search queries do not need to be arbitrarily large, and bounding them keeps
|
||||||
|
# sanitizer/runtime behavior predictable under adversarial input.
|
||||||
|
MAX_FTS5_QUERY_CHARS = 2_048
|
||||||
|
|
||||||
|
|
||||||
|
_FTS_TRIGGERS = (
|
||||||
|
"messages_fts_insert",
|
||||||
|
"messages_fts_delete",
|
||||||
|
"messages_fts_update",
|
||||||
|
"messages_fts_trigram_insert",
|
||||||
|
"messages_fts_trigram_delete",
|
||||||
|
"messages_fts_trigram_update",
|
||||||
|
)
|
||||||
|
|
||||||
|
|
||||||
|
SCHEMA_SQL = """
|
||||||
|
CREATE TABLE IF NOT EXISTS schema_version (
|
||||||
|
version INTEGER NOT NULL
|
||||||
|
);
|
||||||
|
|
||||||
|
CREATE TABLE IF NOT EXISTS sessions (
|
||||||
|
id TEXT PRIMARY KEY,
|
||||||
|
source TEXT NOT NULL,
|
||||||
|
user_id TEXT,
|
||||||
|
session_key TEXT,
|
||||||
|
chat_id TEXT,
|
||||||
|
chat_type TEXT,
|
||||||
|
thread_id TEXT,
|
||||||
|
display_name TEXT,
|
||||||
|
origin_json TEXT,
|
||||||
|
expiry_finalized INTEGER DEFAULT 0,
|
||||||
|
model TEXT,
|
||||||
|
model_config TEXT,
|
||||||
|
system_prompt TEXT,
|
||||||
|
parent_session_id TEXT,
|
||||||
|
started_at REAL NOT NULL,
|
||||||
|
ended_at REAL,
|
||||||
|
end_reason TEXT,
|
||||||
|
message_count INTEGER DEFAULT 0,
|
||||||
|
tool_call_count INTEGER DEFAULT 0,
|
||||||
|
input_tokens INTEGER DEFAULT 0,
|
||||||
|
output_tokens INTEGER DEFAULT 0,
|
||||||
|
cache_read_tokens INTEGER DEFAULT 0,
|
||||||
|
cache_write_tokens INTEGER DEFAULT 0,
|
||||||
|
reasoning_tokens INTEGER DEFAULT 0,
|
||||||
|
cwd TEXT,
|
||||||
|
git_branch TEXT,
|
||||||
|
git_repo_root TEXT,
|
||||||
|
billing_provider TEXT,
|
||||||
|
billing_base_url TEXT,
|
||||||
|
billing_mode TEXT,
|
||||||
|
estimated_cost_usd REAL,
|
||||||
|
actual_cost_usd REAL,
|
||||||
|
cost_status TEXT,
|
||||||
|
cost_source TEXT,
|
||||||
|
pricing_version TEXT,
|
||||||
|
title TEXT,
|
||||||
|
api_call_count INTEGER DEFAULT 0,
|
||||||
|
handoff_state TEXT,
|
||||||
|
handoff_platform TEXT,
|
||||||
|
handoff_error TEXT,
|
||||||
|
compression_failure_cooldown_until REAL,
|
||||||
|
compression_failure_error TEXT,
|
||||||
|
compression_fallback_streak INTEGER NOT NULL DEFAULT 0,
|
||||||
|
compression_ineffective_count INTEGER NOT NULL DEFAULT 0,
|
||||||
|
profile_name TEXT,
|
||||||
|
rewind_count INTEGER NOT NULL DEFAULT 0,
|
||||||
|
archived INTEGER NOT NULL DEFAULT 0,
|
||||||
|
pinned INTEGER NOT NULL DEFAULT 0,
|
||||||
|
FOREIGN KEY (parent_session_id) REFERENCES sessions(id)
|
||||||
|
);
|
||||||
|
|
||||||
|
CREATE TABLE IF NOT EXISTS messages (
|
||||||
|
id INTEGER PRIMARY KEY AUTOINCREMENT,
|
||||||
|
session_id TEXT NOT NULL REFERENCES sessions(id),
|
||||||
|
role TEXT NOT NULL,
|
||||||
|
content TEXT,
|
||||||
|
tool_call_id TEXT,
|
||||||
|
tool_calls TEXT,
|
||||||
|
tool_name TEXT,
|
||||||
|
effect_disposition TEXT,
|
||||||
|
timestamp REAL NOT NULL,
|
||||||
|
token_count INTEGER,
|
||||||
|
finish_reason TEXT,
|
||||||
|
reasoning TEXT,
|
||||||
|
reasoning_content TEXT,
|
||||||
|
reasoning_details TEXT,
|
||||||
|
codex_reasoning_items TEXT,
|
||||||
|
codex_message_items TEXT,
|
||||||
|
platform_message_id TEXT,
|
||||||
|
observed INTEGER DEFAULT 0,
|
||||||
|
active INTEGER NOT NULL DEFAULT 1,
|
||||||
|
compacted INTEGER NOT NULL DEFAULT 0,
|
||||||
|
api_content TEXT,
|
||||||
|
display_kind TEXT,
|
||||||
|
display_metadata TEXT
|
||||||
|
);
|
||||||
|
|
||||||
|
CREATE TABLE IF NOT EXISTS session_model_usage (
|
||||||
|
session_id TEXT NOT NULL REFERENCES sessions(id) ON DELETE CASCADE,
|
||||||
|
model TEXT NOT NULL,
|
||||||
|
billing_provider TEXT NOT NULL DEFAULT '',
|
||||||
|
billing_base_url TEXT NOT NULL DEFAULT '',
|
||||||
|
billing_mode TEXT NOT NULL DEFAULT '',
|
||||||
|
task TEXT NOT NULL DEFAULT '',
|
||||||
|
api_call_count INTEGER NOT NULL DEFAULT 0,
|
||||||
|
input_tokens INTEGER NOT NULL DEFAULT 0,
|
||||||
|
output_tokens INTEGER NOT NULL DEFAULT 0,
|
||||||
|
cache_read_tokens INTEGER NOT NULL DEFAULT 0,
|
||||||
|
cache_write_tokens INTEGER NOT NULL DEFAULT 0,
|
||||||
|
reasoning_tokens INTEGER NOT NULL DEFAULT 0,
|
||||||
|
estimated_cost_usd REAL NOT NULL DEFAULT 0,
|
||||||
|
actual_cost_usd REAL NOT NULL DEFAULT 0,
|
||||||
|
cost_status TEXT,
|
||||||
|
cost_source TEXT,
|
||||||
|
first_seen REAL,
|
||||||
|
last_seen REAL,
|
||||||
|
PRIMARY KEY (session_id, model, billing_provider, billing_base_url, billing_mode, task)
|
||||||
|
);
|
||||||
|
|
||||||
|
CREATE TABLE IF NOT EXISTS state_meta (
|
||||||
|
key TEXT PRIMARY KEY,
|
||||||
|
value TEXT
|
||||||
|
);
|
||||||
|
|
||||||
|
CREATE TABLE IF NOT EXISTS gateway_routing (
|
||||||
|
scope TEXT NOT NULL DEFAULT '',
|
||||||
|
session_key TEXT NOT NULL,
|
||||||
|
entry_json TEXT NOT NULL,
|
||||||
|
updated_at REAL NOT NULL,
|
||||||
|
PRIMARY KEY (scope, session_key)
|
||||||
|
);
|
||||||
|
|
||||||
|
CREATE TABLE IF NOT EXISTS compression_locks (
|
||||||
|
session_id TEXT PRIMARY KEY,
|
||||||
|
holder TEXT NOT NULL,
|
||||||
|
acquired_at REAL NOT NULL,
|
||||||
|
expires_at REAL NOT NULL
|
||||||
|
);
|
||||||
|
|
||||||
|
CREATE TABLE IF NOT EXISTS async_delegations (
|
||||||
|
delegation_id TEXT PRIMARY KEY,
|
||||||
|
origin_session TEXT NOT NULL,
|
||||||
|
origin_ui_session_id TEXT NOT NULL DEFAULT '',
|
||||||
|
parent_session_id TEXT,
|
||||||
|
state TEXT NOT NULL,
|
||||||
|
dispatched_at REAL NOT NULL,
|
||||||
|
completed_at REAL,
|
||||||
|
updated_at REAL NOT NULL,
|
||||||
|
event_json TEXT,
|
||||||
|
result_json TEXT,
|
||||||
|
delivery_state TEXT NOT NULL DEFAULT 'pending',
|
||||||
|
delivery_attempts INTEGER NOT NULL DEFAULT 0,
|
||||||
|
delivered_at REAL,
|
||||||
|
owner_pid INTEGER,
|
||||||
|
owner_started_at INTEGER,
|
||||||
|
task_json TEXT,
|
||||||
|
delivery_claim TEXT,
|
||||||
|
delivery_claimed_at REAL
|
||||||
|
);
|
||||||
|
|
||||||
|
CREATE INDEX IF NOT EXISTS idx_sessions_source ON sessions(source);
|
||||||
|
CREATE INDEX IF NOT EXISTS idx_sessions_source_id ON sessions(source, id);
|
||||||
|
CREATE INDEX IF NOT EXISTS idx_sessions_parent ON sessions(parent_session_id);
|
||||||
|
CREATE INDEX IF NOT EXISTS idx_sessions_started ON sessions(started_at DESC);
|
||||||
|
CREATE INDEX IF NOT EXISTS idx_messages_session ON messages(session_id, timestamp);
|
||||||
|
CREATE INDEX IF NOT EXISTS idx_compression_locks_expires ON compression_locks(expires_at);
|
||||||
|
CREATE INDEX IF NOT EXISTS idx_session_model_usage_session ON session_model_usage(session_id);
|
||||||
|
CREATE INDEX IF NOT EXISTS idx_session_model_usage_model ON session_model_usage(model);
|
||||||
|
CREATE INDEX IF NOT EXISTS idx_async_delegations_delivery
|
||||||
|
ON async_delegations(delivery_state, completed_at);
|
||||||
|
"""
|
||||||
|
|
||||||
|
|
||||||
|
# Indexes that reference columns added in later schema versions must be
|
||||||
|
# created AFTER _reconcile_columns() has had a chance to ADD them on
|
||||||
|
# existing databases. SCHEMA_SQL above is run by sqlite executescript
|
||||||
|
# which would otherwise fail on legacy DBs ("no such column: active").
|
||||||
|
DEFERRED_INDEX_SQL = """
|
||||||
|
CREATE INDEX IF NOT EXISTS idx_messages_session_active
|
||||||
|
ON messages(session_id, active, timestamp);
|
||||||
|
CREATE INDEX IF NOT EXISTS idx_messages_active_null
|
||||||
|
ON messages(active) WHERE active IS NULL;
|
||||||
|
CREATE INDEX IF NOT EXISTS idx_sessions_session_key
|
||||||
|
ON sessions(session_key, started_at DESC);
|
||||||
|
CREATE INDEX IF NOT EXISTS idx_sessions_gateway_peer
|
||||||
|
ON sessions(source, user_id, chat_id, chat_type, thread_id, started_at DESC);
|
||||||
|
CREATE INDEX IF NOT EXISTS idx_sessions_handoff_state
|
||||||
|
ON sessions(handoff_state, started_at);
|
||||||
|
"""
|
||||||
|
|
||||||
|
|
||||||
|
# ── Deferred FTS rebuild bookkeeping (schema v23) ──
|
||||||
|
# While a background index rebuild is pending, two state_meta keys define
|
||||||
|
# which message rows are currently IN the FTS indexes:
|
||||||
|
#
|
||||||
|
# fts_rebuild_high_water H — MAX(messages.id) at the moment the old
|
||||||
|
# indexes were dropped
|
||||||
|
# fts_rebuild_progress P — highest id the chunked backfill has indexed
|
||||||
|
#
|
||||||
|
# A row is indexed iff id <= P (backfilled) OR id > H (inserted after
|
||||||
|
# the drop; ids are AUTOINCREMENT so new rows are always > H and the insert
|
||||||
|
# triggers index them live). Rows in (P, H] are not yet indexed.
|
||||||
|
#
|
||||||
|
# Every trigger below gates on that same predicate: firing an FTS5
|
||||||
|
# external-content 'delete' for a row that is NOT in the index corrupts the
|
||||||
|
# index, and skipping it for a row that IS indexed leaves a stale entry.
|
||||||
|
# When no rebuild is pending both keys are absent and COALESCE turns the
|
||||||
|
# predicate into a tautology (id > -1 OR id <= -1), i.e. normal operation.
|
||||||
|
# The two state_meta PK probes per write are negligible next to the FTS
|
||||||
|
# insert itself.
|
||||||
|
FTS_SQL = """
|
||||||
|
CREATE VIRTUAL TABLE IF NOT EXISTS messages_fts USING fts5(
|
||||||
|
content,
|
||||||
|
tool_name,
|
||||||
|
tool_calls,
|
||||||
|
content='messages',
|
||||||
|
content_rowid='id'
|
||||||
|
);
|
||||||
|
|
||||||
|
CREATE TRIGGER IF NOT EXISTS messages_fts_insert AFTER INSERT ON messages
|
||||||
|
WHEN (new.id > COALESCE((SELECT CAST(value AS INTEGER) FROM state_meta
|
||||||
|
WHERE key = 'fts_rebuild_high_water'), -1)
|
||||||
|
OR new.id <= COALESCE((SELECT CAST(value AS INTEGER) FROM state_meta
|
||||||
|
WHERE key = 'fts_rebuild_progress'), -1))
|
||||||
|
BEGIN
|
||||||
|
INSERT INTO messages_fts(rowid, content, tool_name, tool_calls)
|
||||||
|
VALUES (new.id, new.content, new.tool_name, new.tool_calls);
|
||||||
|
END;
|
||||||
|
|
||||||
|
CREATE TRIGGER IF NOT EXISTS messages_fts_delete AFTER DELETE ON messages
|
||||||
|
WHEN (old.id > COALESCE((SELECT CAST(value AS INTEGER) FROM state_meta
|
||||||
|
WHERE key = 'fts_rebuild_high_water'), -1)
|
||||||
|
OR old.id <= COALESCE((SELECT CAST(value AS INTEGER) FROM state_meta
|
||||||
|
WHERE key = 'fts_rebuild_progress'), -1))
|
||||||
|
BEGIN
|
||||||
|
INSERT INTO messages_fts(messages_fts, rowid, content, tool_name, tool_calls)
|
||||||
|
VALUES ('delete', old.id, old.content, old.tool_name, old.tool_calls);
|
||||||
|
END;
|
||||||
|
|
||||||
|
CREATE TRIGGER IF NOT EXISTS messages_fts_update AFTER UPDATE ON messages
|
||||||
|
WHEN (old.content IS NOT new.content
|
||||||
|
OR old.tool_name IS NOT new.tool_name
|
||||||
|
OR old.tool_calls IS NOT new.tool_calls)
|
||||||
|
AND (old.id > COALESCE((SELECT CAST(value AS INTEGER) FROM state_meta
|
||||||
|
WHERE key = 'fts_rebuild_high_water'), -1)
|
||||||
|
OR old.id <= COALESCE((SELECT CAST(value AS INTEGER) FROM state_meta
|
||||||
|
WHERE key = 'fts_rebuild_progress'), -1))
|
||||||
|
BEGIN
|
||||||
|
INSERT INTO messages_fts(messages_fts, rowid, content, tool_name, tool_calls)
|
||||||
|
VALUES ('delete', old.id, old.content, old.tool_name, old.tool_calls);
|
||||||
|
INSERT INTO messages_fts(rowid, content, tool_name, tool_calls)
|
||||||
|
VALUES (new.id, new.content, new.tool_name, new.tool_calls);
|
||||||
|
END;
|
||||||
|
"""
|
||||||
|
|
||||||
|
|
||||||
|
# Trigram FTS5 table for CJK substring search. The default unicode61
|
||||||
|
# tokenizer splits CJK characters into individual tokens, breaking phrase
|
||||||
|
# matching. The trigram tokenizer creates overlapping 3-byte sequences so
|
||||||
|
# substring queries work natively for any script (CJK, Thai, etc.).
|
||||||
|
#
|
||||||
|
# The trigram index is the most expensive index in state.db (~2.6x the size
|
||||||
|
# of the text it covers), and ``role='tool'`` rows are ~90% of message bytes
|
||||||
|
# while being almost entirely machine noise (base64 payloads, file dumps,
|
||||||
|
# delegation transcripts). The index therefore reads through
|
||||||
|
# ``messages_fts_trigram_src``, a view that excludes tool rows — they stay
|
||||||
|
# fully stored in ``messages`` and fully searchable via the standard
|
||||||
|
# ``messages_fts`` index; they just don't get trigram (CJK substring)
|
||||||
|
# treatment. ``search_messages`` routes CJK queries that filter on
|
||||||
|
# ``role='tool'`` to the LIKE fallback for the same reason.
|
||||||
|
FTS_TRIGRAM_SQL = """
|
||||||
|
CREATE VIEW IF NOT EXISTS messages_fts_trigram_src AS
|
||||||
|
SELECT id, role, content, tool_name, tool_calls
|
||||||
|
FROM messages
|
||||||
|
WHERE role <> 'tool';
|
||||||
|
|
||||||
|
CREATE VIRTUAL TABLE IF NOT EXISTS messages_fts_trigram USING fts5(
|
||||||
|
content,
|
||||||
|
tool_name,
|
||||||
|
tool_calls,
|
||||||
|
content='messages_fts_trigram_src',
|
||||||
|
content_rowid='id',
|
||||||
|
tokenize='trigram'
|
||||||
|
);
|
||||||
|
|
||||||
|
CREATE TRIGGER IF NOT EXISTS messages_fts_trigram_insert AFTER INSERT ON messages
|
||||||
|
WHEN new.role <> 'tool'
|
||||||
|
AND (new.id > COALESCE((SELECT CAST(value AS INTEGER) FROM state_meta
|
||||||
|
WHERE key = 'fts_rebuild_high_water'), -1)
|
||||||
|
OR new.id <= COALESCE((SELECT CAST(value AS INTEGER) FROM state_meta
|
||||||
|
WHERE key = 'fts_rebuild_progress'), -1))
|
||||||
|
BEGIN
|
||||||
|
INSERT INTO messages_fts_trigram(rowid, content, tool_name, tool_calls)
|
||||||
|
VALUES (new.id, new.content, new.tool_name, new.tool_calls);
|
||||||
|
END;
|
||||||
|
|
||||||
|
CREATE TRIGGER IF NOT EXISTS messages_fts_trigram_delete AFTER DELETE ON messages
|
||||||
|
WHEN old.role <> 'tool'
|
||||||
|
AND (old.id > COALESCE((SELECT CAST(value AS INTEGER) FROM state_meta
|
||||||
|
WHERE key = 'fts_rebuild_high_water'), -1)
|
||||||
|
OR old.id <= COALESCE((SELECT CAST(value AS INTEGER) FROM state_meta
|
||||||
|
WHERE key = 'fts_rebuild_progress'), -1))
|
||||||
|
BEGIN
|
||||||
|
INSERT INTO messages_fts_trigram(messages_fts_trigram, rowid, content, tool_name, tool_calls)
|
||||||
|
VALUES ('delete', old.id, old.content, old.tool_name, old.tool_calls);
|
||||||
|
END;
|
||||||
|
|
||||||
|
CREATE TRIGGER IF NOT EXISTS messages_fts_trigram_update AFTER UPDATE ON messages
|
||||||
|
WHEN (old.content IS NOT new.content
|
||||||
|
OR old.tool_name IS NOT new.tool_name
|
||||||
|
OR old.tool_calls IS NOT new.tool_calls
|
||||||
|
OR old.role IS NOT new.role)
|
||||||
|
AND (old.id > COALESCE((SELECT CAST(value AS INTEGER) FROM state_meta
|
||||||
|
WHERE key = 'fts_rebuild_high_water'), -1)
|
||||||
|
OR old.id <= COALESCE((SELECT CAST(value AS INTEGER) FROM state_meta
|
||||||
|
WHERE key = 'fts_rebuild_progress'), -1))
|
||||||
|
BEGIN
|
||||||
|
INSERT INTO messages_fts_trigram(messages_fts_trigram, rowid, content, tool_name, tool_calls)
|
||||||
|
SELECT 'delete', old.id, old.content, old.tool_name, old.tool_calls
|
||||||
|
WHERE old.role <> 'tool';
|
||||||
|
INSERT INTO messages_fts_trigram(rowid, content, tool_name, tool_calls)
|
||||||
|
SELECT new.id, new.content, new.tool_name, new.tool_calls
|
||||||
|
WHERE new.role <> 'tool';
|
||||||
|
END;
|
||||||
|
"""
|
||||||
|
|
||||||
|
|
||||||
|
_FTS_CJK_TRIGGERS = (
|
||||||
|
"messages_fts_cjk_insert",
|
||||||
|
"messages_fts_cjk_delete",
|
||||||
|
"messages_fts_cjk_update",
|
||||||
|
)
|
||||||
|
|
||||||
|
|
||||||
|
# state_meta breadcrumb set when a tokenizer-less process had to drop the
|
||||||
|
# cjk triggers to keep message writes alive: rows written from that moment
|
||||||
|
# on are missing from the cjk index, so it must not serve reads until
|
||||||
|
# `hermes sessions optimize-storage` rebuilds it on a capable host.
|
||||||
|
FTS_CJK_STALE_KEY = "fts_cjk_stale"
|
||||||
|
|
||||||
|
|
||||||
|
# ── Legacy (v22 / inline-content) FTS DDL ──────────────────────────────
|
||||||
|
# Used ONLY to keep an existing pre-v23 install's search working and its
|
||||||
|
# triggers repairable UNTIL the user opts into `hermes db optimize`. This is
|
||||||
|
# the exact inline shape v11..v22 shipped: each virtual table stores its own
|
||||||
|
# copy of ``content || tool_name || tool_calls`` and the trigram table indexes
|
||||||
|
# every row (including role='tool'). We never CREATE these on a fresh install —
|
||||||
|
# fresh installs are born on the v23 external-content schema above. These
|
||||||
|
# constants exist so a legacy DB is never accidentally handed the v23 DDL
|
||||||
|
# (which would create the external-content trigram source VIEW and leave the
|
||||||
|
# DB in a mixed, broken state). `optimize_fts_storage()` is what migrates a
|
||||||
|
# legacy DB to the v23 shape.
|
||||||
|
LEGACY_FTS_SQL = """
|
||||||
|
CREATE VIRTUAL TABLE IF NOT EXISTS messages_fts USING fts5(
|
||||||
|
content
|
||||||
|
);
|
||||||
|
|
||||||
|
CREATE TRIGGER IF NOT EXISTS messages_fts_insert AFTER INSERT ON messages BEGIN
|
||||||
|
INSERT INTO messages_fts(rowid, content) VALUES (
|
||||||
|
new.id,
|
||||||
|
COALESCE(new.content, '') || ' ' || COALESCE(new.tool_name, '') || ' ' || COALESCE(new.tool_calls, '')
|
||||||
|
);
|
||||||
|
END;
|
||||||
|
|
||||||
|
CREATE TRIGGER IF NOT EXISTS messages_fts_delete AFTER DELETE ON messages BEGIN
|
||||||
|
DELETE FROM messages_fts WHERE rowid = old.id;
|
||||||
|
END;
|
||||||
|
|
||||||
|
CREATE TRIGGER IF NOT EXISTS messages_fts_update AFTER UPDATE ON messages BEGIN
|
||||||
|
DELETE FROM messages_fts WHERE rowid = old.id;
|
||||||
|
INSERT INTO messages_fts(rowid, content) VALUES (
|
||||||
|
new.id,
|
||||||
|
COALESCE(new.content, '') || ' ' || COALESCE(new.tool_name, '') || ' ' || COALESCE(new.tool_calls, '')
|
||||||
|
);
|
||||||
|
END;
|
||||||
|
"""
|
||||||
|
|
||||||
|
|
||||||
|
LEGACY_FTS_TRIGRAM_SQL = """
|
||||||
|
CREATE VIRTUAL TABLE IF NOT EXISTS messages_fts_trigram USING fts5(
|
||||||
|
content,
|
||||||
|
tokenize='trigram'
|
||||||
|
);
|
||||||
|
|
||||||
|
CREATE TRIGGER IF NOT EXISTS messages_fts_trigram_insert AFTER INSERT ON messages BEGIN
|
||||||
|
INSERT INTO messages_fts_trigram(rowid, content) VALUES (
|
||||||
|
new.id,
|
||||||
|
COALESCE(new.content, '') || ' ' || COALESCE(new.tool_name, '') || ' ' || COALESCE(new.tool_calls, '')
|
||||||
|
);
|
||||||
|
END;
|
||||||
|
|
||||||
|
CREATE TRIGGER IF NOT EXISTS messages_fts_trigram_delete AFTER DELETE ON messages BEGIN
|
||||||
|
DELETE FROM messages_fts_trigram WHERE rowid = old.id;
|
||||||
|
END;
|
||||||
|
|
||||||
|
CREATE TRIGGER IF NOT EXISTS messages_fts_trigram_update AFTER UPDATE ON messages BEGIN
|
||||||
|
DELETE FROM messages_fts_trigram WHERE rowid = old.id;
|
||||||
|
INSERT INTO messages_fts_trigram(rowid, content) VALUES (
|
||||||
|
new.id,
|
||||||
|
COALESCE(new.content, '') || ' ' || COALESCE(new.tool_name, '') || ' ' || COALESCE(new.tool_calls, '')
|
||||||
|
);
|
||||||
|
END;
|
||||||
|
"""
|
||||||
@@ -0,0 +1,656 @@
|
|||||||
|
"""Session listing/rich rows, export, and import (portability) for SessionDB.
|
||||||
|
|
||||||
|
Mixin contract: this is a plain mixin class consumed by
|
||||||
|
``hermes_state.SessionDB``. It defines no ``__init__`` and no state of its
|
||||||
|
own; methods access the host's attributes (``self._conn``, ``self.db_path``,
|
||||||
|
``self._execute_write`` and other SessionDB methods) established by
|
||||||
|
``SessionDB.__init__``. It must never import hermes_state (cycle) — shared
|
||||||
|
module-level constants live in hermes_state_common.
|
||||||
|
"""
|
||||||
|
|
||||||
|
import logging
|
||||||
|
import json
|
||||||
|
import time
|
||||||
|
from typing import Any, Dict, List, Optional
|
||||||
|
|
||||||
|
from agent.skill_commands import SKILL_SCAFFOLD_SQL_LIKE
|
||||||
|
from hermes_state_common import (
|
||||||
|
SCHEMA_SQL,
|
||||||
|
_PREVIEW_RAW_SELECT,
|
||||||
|
_shape_preview,
|
||||||
|
)
|
||||||
|
|
||||||
|
# Moved methods logged under the "hermes_state" logger before the split;
|
||||||
|
# keep that logger identity so log filtering/capture behavior is unchanged.
|
||||||
|
logger = logging.getLogger("hermes_state")
|
||||||
|
|
||||||
|
|
||||||
|
class SessionPortabilityMixin:
|
||||||
|
"""See module docstring — mixin for SessionDB (Port cluster)."""
|
||||||
|
|
||||||
|
@classmethod
|
||||||
|
def _compact_session_cols(cls) -> str:
|
||||||
|
"""SELECT list for compact_rows: every ``sessions`` column declared in
|
||||||
|
SCHEMA_SQL except the ``system_prompt`` blob, aliased with the ``s``
|
||||||
|
prefix used by list_sessions_rich/_get_session_rich_row queries."""
|
||||||
|
if cls._session_compact_cols_sql is None:
|
||||||
|
declared = cls._parse_schema_columns(SCHEMA_SQL)["sessions"]
|
||||||
|
cls._session_compact_cols_sql = ", ".join(
|
||||||
|
f"s.{name}" for name in declared
|
||||||
|
if name not in cls._SESSION_COMPACT_EXCLUDED
|
||||||
|
)
|
||||||
|
return cls._session_compact_cols_sql
|
||||||
|
|
||||||
|
def distinct_session_cwds(self, include_archived: bool = False) -> List[Dict[str, Any]]:
|
||||||
|
"""Distinct non-empty session cwds with usage stats, for repo discovery.
|
||||||
|
|
||||||
|
Aggregates across ALL session history (not a single page), so the desktop
|
||||||
|
can surface every git repo the user has worked in — not just the repos
|
||||||
|
that happen to be in the currently-loaded recents. Children/branches
|
||||||
|
count: a worktree session is still a real workspace signal.
|
||||||
|
"""
|
||||||
|
where = "cwd IS NOT NULL AND TRIM(cwd) != ''"
|
||||||
|
if not include_archived:
|
||||||
|
where += " AND archived = 0"
|
||||||
|
with self._lock:
|
||||||
|
rows = self._conn.execute(
|
||||||
|
"SELECT cwd AS cwd, COUNT(*) AS sessions, "
|
||||||
|
"MAX(COALESCE(ended_at, started_at, 0)) AS last_active "
|
||||||
|
f"FROM sessions WHERE {where} GROUP BY cwd"
|
||||||
|
).fetchall()
|
||||||
|
return [
|
||||||
|
{
|
||||||
|
"cwd": r["cwd"],
|
||||||
|
"sessions": int(r["sessions"] or 0),
|
||||||
|
"last_active": float(r["last_active"] or 0),
|
||||||
|
}
|
||||||
|
for r in rows
|
||||||
|
]
|
||||||
|
|
||||||
|
def list_cron_job_runs(
|
||||||
|
self,
|
||||||
|
job_id: str,
|
||||||
|
limit: int = 20,
|
||||||
|
offset: int = 0,
|
||||||
|
) -> List[Dict[str, Any]]:
|
||||||
|
"""List the run sessions produced by a single cron job, newest first.
|
||||||
|
|
||||||
|
Cron runs are flat, independent sessions whose id is
|
||||||
|
``cron_{job_id}_{timestamp}`` (see ``cron/scheduler.run_job``). They are
|
||||||
|
never compression roots and never branch, so this deliberately skips the
|
||||||
|
``list_sessions_rich`` recursive compression-chain CTE / leading-wildcard
|
||||||
|
``id_query`` path — that path seeds from *every* ``source='cron'`` row in
|
||||||
|
the DB and only filters to one job's runs after the scan, so it scales
|
||||||
|
with the whole cron pile (a heavy history makes the desktop run-history
|
||||||
|
endpoint time out before it eventually populates).
|
||||||
|
|
||||||
|
Instead this binds to one job with a ``[prefix, prefix_hi)`` range over
|
||||||
|
the id (an index range scan, not a ``%...%`` substring), filters
|
||||||
|
``source='cron'``, and orders by ``started_at DESC``. Work scales with
|
||||||
|
the requested window, not the total cron history.
|
||||||
|
|
||||||
|
Returns the same enriched row shape as ``list_sessions_rich`` (adds
|
||||||
|
``preview`` + ``last_active``) so callers can reuse it.
|
||||||
|
"""
|
||||||
|
prefix = f"cron_{job_id}_"
|
||||||
|
# Half-open upper bound for an index range scan: increment the final
|
||||||
|
# byte of the prefix so the range covers exactly the ids that start
|
||||||
|
# with ``prefix`` and nothing else. ``prefix`` always ends in '_', but
|
||||||
|
# compute it generically rather than hardcoding the successor char.
|
||||||
|
prefix_hi = prefix[:-1] + chr(ord(prefix[-1]) + 1)
|
||||||
|
|
||||||
|
query = f"""
|
||||||
|
SELECT s.*,
|
||||||
|
COALESCE(
|
||||||
|
(SELECT {_PREVIEW_RAW_SELECT}
|
||||||
|
FROM messages m
|
||||||
|
WHERE m.session_id = s.id AND m.role = 'user' AND m.content IS NOT NULL
|
||||||
|
ORDER BY m.timestamp, m.id LIMIT 1),
|
||||||
|
''
|
||||||
|
) AS _preview_raw,
|
||||||
|
COALESCE(
|
||||||
|
(SELECT MAX(m2.timestamp) FROM messages m2 WHERE m2.session_id = s.id),
|
||||||
|
s.started_at
|
||||||
|
) AS last_active
|
||||||
|
FROM sessions s
|
||||||
|
WHERE s.source = 'cron' AND s.id >= ? AND s.id < ?
|
||||||
|
ORDER BY s.started_at DESC, s.id DESC
|
||||||
|
LIMIT ? OFFSET ?
|
||||||
|
"""
|
||||||
|
with self._lock:
|
||||||
|
cursor = self._conn.execute(query, (prefix, prefix_hi, limit, offset))
|
||||||
|
rows = cursor.fetchall()
|
||||||
|
|
||||||
|
runs: List[Dict[str, Any]] = []
|
||||||
|
for row in rows:
|
||||||
|
s = dict(row)
|
||||||
|
s["preview"] = _shape_preview(s.pop("_preview_raw", ""))
|
||||||
|
runs.append(s)
|
||||||
|
return runs
|
||||||
|
|
||||||
|
def _get_session_rich_row(self, session_id: str, compact_rows: bool = False) -> Optional[Dict[str, Any]]:
|
||||||
|
"""Fetch a single session with the same enriched columns as
|
||||||
|
``list_sessions_rich`` (preview + last_active). Returns None if the
|
||||||
|
session doesn't exist.
|
||||||
|
|
||||||
|
Pass ``compact_rows=True`` to omit the ``system_prompt`` blob (see
|
||||||
|
``list_sessions_rich`` for details).
|
||||||
|
"""
|
||||||
|
# Same read-your-writes guarantee as list_sessions_rich.
|
||||||
|
self.flush_token_counts()
|
||||||
|
_sel = self._compact_session_cols() if compact_rows else "s.*"
|
||||||
|
query = f"""
|
||||||
|
SELECT {_sel},
|
||||||
|
COALESCE(
|
||||||
|
(SELECT {_PREVIEW_RAW_SELECT}
|
||||||
|
FROM messages m
|
||||||
|
WHERE m.session_id = s.id AND m.role = 'user' AND m.content IS NOT NULL
|
||||||
|
ORDER BY m.timestamp, m.id LIMIT 1),
|
||||||
|
''
|
||||||
|
) AS _preview_raw,
|
||||||
|
COALESCE(
|
||||||
|
(SELECT MAX(m2.timestamp) FROM messages m2 WHERE m2.session_id = s.id),
|
||||||
|
s.started_at
|
||||||
|
) AS last_active
|
||||||
|
FROM sessions s
|
||||||
|
WHERE s.id = ?
|
||||||
|
"""
|
||||||
|
with self._lock:
|
||||||
|
cursor = self._conn.execute(query, (session_id,))
|
||||||
|
row = cursor.fetchone()
|
||||||
|
if not row:
|
||||||
|
return None
|
||||||
|
s = dict(row)
|
||||||
|
s["preview"] = _shape_preview(s.pop("_preview_raw", ""))
|
||||||
|
return s
|
||||||
|
|
||||||
|
def get_session_rich_row(self, session_id: str, compact_rows: bool = False) -> Optional[Dict[str, Any]]:
|
||||||
|
"""Public wrapper for :meth:`_get_session_rich_row`.
|
||||||
|
|
||||||
|
Exposes the single-session enriched row (same columns as
|
||||||
|
``list_sessions_rich``: preview + last_active) for callers outside
|
||||||
|
this module, e.g. the web server's session-search hydration.
|
||||||
|
"""
|
||||||
|
return self._get_session_rich_row(session_id, compact_rows=compact_rows)
|
||||||
|
|
||||||
|
def list_skill_scaffolded_sessions(self, limit: int = 200) -> List[Dict[str, Any]]:
|
||||||
|
"""Titled sessions whose first user turn was a ``/skill`` invocation.
|
||||||
|
|
||||||
|
Those titles were generated from the expanded message, which embeds the
|
||||||
|
whole skill body — so they describe the skill rather than the request.
|
||||||
|
Returns ``id``, ``title``, and the full first-turn ``content`` so a
|
||||||
|
caller can re-derive what the user typed. Newest first.
|
||||||
|
"""
|
||||||
|
with self._lock:
|
||||||
|
rows = self._conn.execute(
|
||||||
|
"""
|
||||||
|
SELECT s.id, s.title, m.content
|
||||||
|
FROM sessions s
|
||||||
|
JOIN messages m ON m.id = (
|
||||||
|
SELECT m2.id FROM messages m2
|
||||||
|
WHERE m2.session_id = s.id AND m2.role = 'user'
|
||||||
|
AND m2.content IS NOT NULL
|
||||||
|
ORDER BY m2.timestamp, m2.id LIMIT 1
|
||||||
|
)
|
||||||
|
WHERE s.title IS NOT NULL AND m.content LIKE ?
|
||||||
|
ORDER BY s.started_at DESC
|
||||||
|
LIMIT ?
|
||||||
|
""",
|
||||||
|
(SKILL_SCAFFOLD_SQL_LIKE, int(limit)),
|
||||||
|
).fetchall()
|
||||||
|
return [dict(row) for row in rows]
|
||||||
|
|
||||||
|
def get_first_assistant_text(self, session_id: str) -> str:
|
||||||
|
"""The session's first assistant reply as plain text ('' when none).
|
||||||
|
|
||||||
|
Pairs with :meth:`list_skill_scaffolded_sessions` so a re-title can feed
|
||||||
|
the titler the same (request, reply) shape the live path uses.
|
||||||
|
"""
|
||||||
|
with self._lock:
|
||||||
|
row = self._conn.execute(
|
||||||
|
"SELECT content FROM messages "
|
||||||
|
"WHERE session_id = ? AND role = 'assistant' AND content IS NOT NULL "
|
||||||
|
"ORDER BY timestamp, id LIMIT 1",
|
||||||
|
(session_id,),
|
||||||
|
).fetchone()
|
||||||
|
if not row:
|
||||||
|
return ""
|
||||||
|
decoded = self._decode_content(row["content"])
|
||||||
|
return decoded if isinstance(decoded, str) else ""
|
||||||
|
|
||||||
|
def export_session(self, session_id: str) -> Optional[Dict[str, Any]]:
|
||||||
|
"""Export a single session with all its messages as a dict."""
|
||||||
|
session = self.get_session(session_id)
|
||||||
|
if not session:
|
||||||
|
return None
|
||||||
|
messages = self.get_messages(session_id)
|
||||||
|
return {**session, "messages": messages}
|
||||||
|
|
||||||
|
def export_session_lineage(self, session_id: str) -> Optional[Dict[str, Any]]:
|
||||||
|
"""Export a compression lineage as one logical session dict."""
|
||||||
|
lineage_ids = self.get_compression_lineage(session_id)
|
||||||
|
if not lineage_ids:
|
||||||
|
return None
|
||||||
|
segments = []
|
||||||
|
for sid in lineage_ids:
|
||||||
|
segment = self.export_session(sid)
|
||||||
|
if segment:
|
||||||
|
segments.append(segment)
|
||||||
|
if not segments:
|
||||||
|
return None
|
||||||
|
base = dict(segments[-1])
|
||||||
|
total_messages = sum(len(seg.get("messages") or []) for seg in segments)
|
||||||
|
base["segments"] = segments
|
||||||
|
base["lineage_session_ids"] = [seg["id"] for seg in segments]
|
||||||
|
base["message_count"] = total_messages
|
||||||
|
base["messages"] = [msg for seg in segments for msg in (seg.get("messages") or [])]
|
||||||
|
return base
|
||||||
|
|
||||||
|
def export_all(self, source: str = None) -> List[Dict[str, Any]]:
|
||||||
|
"""
|
||||||
|
Export all sessions (with messages) as a list of dicts.
|
||||||
|
Suitable for writing to a JSONL file for backup/analysis.
|
||||||
|
"""
|
||||||
|
sessions = self.search_sessions(source=source, limit=100000)
|
||||||
|
results = []
|
||||||
|
for session in sessions:
|
||||||
|
messages = self.get_messages(session["id"])
|
||||||
|
results.append({**session, "messages": messages})
|
||||||
|
return results
|
||||||
|
|
||||||
|
@staticmethod
|
||||||
|
def _import_text_or_none(value: Any, field: str) -> Optional[str]:
|
||||||
|
if value is None:
|
||||||
|
return None
|
||||||
|
if isinstance(value, str):
|
||||||
|
return value
|
||||||
|
raise ValueError(f"{field} must be a string")
|
||||||
|
|
||||||
|
@staticmethod
|
||||||
|
def _import_json_object_or_none(value: Any, field: str) -> Optional[str]:
|
||||||
|
if value is None:
|
||||||
|
return None
|
||||||
|
if isinstance(value, str):
|
||||||
|
try:
|
||||||
|
parsed = json.loads(value)
|
||||||
|
except json.JSONDecodeError as exc:
|
||||||
|
raise ValueError(f"{field} must be valid JSON") from exc
|
||||||
|
if not isinstance(parsed, dict):
|
||||||
|
raise ValueError(f"{field} must be a JSON object")
|
||||||
|
return value
|
||||||
|
if not isinstance(value, dict):
|
||||||
|
raise ValueError(f"{field} must be a JSON object")
|
||||||
|
try:
|
||||||
|
return json.dumps(value)
|
||||||
|
except (TypeError, ValueError) as exc:
|
||||||
|
raise ValueError(f"{field} must be JSON serializable") from exc
|
||||||
|
|
||||||
|
@staticmethod
|
||||||
|
def _float_or_none(value: Any) -> Optional[float]:
|
||||||
|
if value is None:
|
||||||
|
return None
|
||||||
|
try:
|
||||||
|
return float(value)
|
||||||
|
except (TypeError, ValueError):
|
||||||
|
return None
|
||||||
|
|
||||||
|
@staticmethod
|
||||||
|
def _import_int_or_none(value: Any, field: str) -> Optional[int]:
|
||||||
|
if value is None:
|
||||||
|
return None
|
||||||
|
try:
|
||||||
|
return int(value)
|
||||||
|
except (TypeError, ValueError) as exc:
|
||||||
|
raise ValueError(f"{field} must be an integer") from exc
|
||||||
|
|
||||||
|
@staticmethod
|
||||||
|
def _int_or_default(value: Any, default: int = 0) -> int:
|
||||||
|
if value is None:
|
||||||
|
return default
|
||||||
|
try:
|
||||||
|
return int(value)
|
||||||
|
except (TypeError, ValueError):
|
||||||
|
return default
|
||||||
|
|
||||||
|
@staticmethod
|
||||||
|
def _reasoning_json_value(value: Any) -> Any:
|
||||||
|
if not isinstance(value, str):
|
||||||
|
return value
|
||||||
|
try:
|
||||||
|
return json.loads(value)
|
||||||
|
except (json.JSONDecodeError, TypeError):
|
||||||
|
return value
|
||||||
|
|
||||||
|
@staticmethod
|
||||||
|
def _import_error(index: int, session_id: str, error: str) -> Dict[str, Any]:
|
||||||
|
item: Dict[str, Any] = {"index": index, "error": error}
|
||||||
|
if session_id:
|
||||||
|
item["session_id"] = session_id
|
||||||
|
return item
|
||||||
|
|
||||||
|
def import_sessions(self, sessions: List[Dict[str, Any]]) -> Dict[str, Any]:
|
||||||
|
"""Import sessions exported by :meth:`export_session` or ``export_all``.
|
||||||
|
|
||||||
|
Existing session IDs are skipped. Imported child sessions keep their
|
||||||
|
parent only when that parent already exists or is included in the same
|
||||||
|
import payload; otherwise the child is detached so partial imports don't
|
||||||
|
fail foreign-key validation. Gateway routing, handoff, rewind, and other
|
||||||
|
live runtime state are intentionally reset: this restores conversation
|
||||||
|
history, not ownership of a live channel or process.
|
||||||
|
"""
|
||||||
|
if not isinstance(sessions, list):
|
||||||
|
raise ValueError("sessions must be a list")
|
||||||
|
if len(sessions) > self._IMPORT_MAX_SESSIONS:
|
||||||
|
raise ValueError(
|
||||||
|
f"sessions must contain at most {self._IMPORT_MAX_SESSIONS} entries"
|
||||||
|
)
|
||||||
|
|
||||||
|
normalized: List[Dict[str, Any]] = []
|
||||||
|
errors: List[Dict[str, Any]] = []
|
||||||
|
seen_ids: set[str] = set()
|
||||||
|
total_messages = 0
|
||||||
|
total_bytes = 0
|
||||||
|
session_text_fields = (
|
||||||
|
"source",
|
||||||
|
"user_id",
|
||||||
|
"model",
|
||||||
|
"system_prompt",
|
||||||
|
"end_reason",
|
||||||
|
"cwd",
|
||||||
|
"git_branch",
|
||||||
|
"git_repo_root",
|
||||||
|
"billing_provider",
|
||||||
|
"billing_base_url",
|
||||||
|
"billing_mode",
|
||||||
|
"cost_status",
|
||||||
|
"cost_source",
|
||||||
|
"pricing_version",
|
||||||
|
"title",
|
||||||
|
)
|
||||||
|
message_text_fields = (
|
||||||
|
"role",
|
||||||
|
"tool_call_id",
|
||||||
|
"tool_name",
|
||||||
|
"effect_disposition",
|
||||||
|
"finish_reason",
|
||||||
|
"reasoning",
|
||||||
|
"reasoning_content",
|
||||||
|
"platform_message_id",
|
||||||
|
"message_id",
|
||||||
|
)
|
||||||
|
|
||||||
|
for index, raw in enumerate(sessions):
|
||||||
|
if not isinstance(raw, dict):
|
||||||
|
errors.append(self._import_error(index, "", "session must be an object"))
|
||||||
|
continue
|
||||||
|
session_id = str(raw.get("id") or "").strip()
|
||||||
|
if not session_id:
|
||||||
|
errors.append(self._import_error(index, "", "session id is required"))
|
||||||
|
continue
|
||||||
|
if session_id in seen_ids:
|
||||||
|
errors.append(self._import_error(index, session_id, "duplicate session id"))
|
||||||
|
continue
|
||||||
|
messages = raw.get("messages") or []
|
||||||
|
if not isinstance(messages, list):
|
||||||
|
errors.append(self._import_error(index, session_id, "messages must be a list"))
|
||||||
|
continue
|
||||||
|
if len(messages) > self._IMPORT_MAX_MESSAGES_PER_SESSION:
|
||||||
|
errors.append(
|
||||||
|
self._import_error(
|
||||||
|
index,
|
||||||
|
session_id,
|
||||||
|
"messages exceeds the per-session import limit",
|
||||||
|
)
|
||||||
|
)
|
||||||
|
continue
|
||||||
|
if any(not isinstance(msg, dict) for msg in messages):
|
||||||
|
errors.append(
|
||||||
|
self._import_error(
|
||||||
|
index,
|
||||||
|
session_id,
|
||||||
|
"messages must contain only objects",
|
||||||
|
)
|
||||||
|
)
|
||||||
|
continue
|
||||||
|
|
||||||
|
try:
|
||||||
|
session_bytes = len(
|
||||||
|
json.dumps(raw, ensure_ascii=False, separators=(",", ":")).encode("utf-8")
|
||||||
|
)
|
||||||
|
except (TypeError, ValueError):
|
||||||
|
errors.append(
|
||||||
|
self._import_error(index, session_id, "session must be JSON serializable")
|
||||||
|
)
|
||||||
|
continue
|
||||||
|
if session_bytes > self._IMPORT_MAX_SESSION_BYTES:
|
||||||
|
errors.append(
|
||||||
|
self._import_error(index, session_id, "session exceeds the import size limit")
|
||||||
|
)
|
||||||
|
continue
|
||||||
|
total_bytes += session_bytes
|
||||||
|
if total_bytes > self._IMPORT_MAX_TOTAL_BYTES:
|
||||||
|
errors.append(
|
||||||
|
self._import_error(index, session_id, "import exceeds the total size limit")
|
||||||
|
)
|
||||||
|
continue
|
||||||
|
|
||||||
|
try:
|
||||||
|
clean_session = dict(raw)
|
||||||
|
clean_session["id"] = session_id
|
||||||
|
clean_session["model_config"] = self._import_json_object_or_none(
|
||||||
|
clean_session.get("model_config"), "model_config"
|
||||||
|
)
|
||||||
|
clean_session["parent_session_id"] = self._import_text_or_none(
|
||||||
|
clean_session.get("parent_session_id"), "parent_session_id"
|
||||||
|
)
|
||||||
|
for field in session_text_fields:
|
||||||
|
clean_session[field] = self._import_text_or_none(
|
||||||
|
clean_session.get(field), field
|
||||||
|
)
|
||||||
|
|
||||||
|
clean_messages: List[Dict[str, Any]] = []
|
||||||
|
for message_index, message in enumerate(messages):
|
||||||
|
clean_message = dict(message)
|
||||||
|
role = clean_message.get("role")
|
||||||
|
if not isinstance(role, str) or not role:
|
||||||
|
raise ValueError(f"messages[{message_index}].role must be a non-empty string")
|
||||||
|
for field in message_text_fields:
|
||||||
|
if field == "role":
|
||||||
|
continue
|
||||||
|
clean_message[field] = self._import_text_or_none(
|
||||||
|
clean_message.get(field), field
|
||||||
|
)
|
||||||
|
clean_message["token_count"] = self._import_int_or_none(
|
||||||
|
clean_message.get("token_count"), "token_count"
|
||||||
|
)
|
||||||
|
clean_messages.append(clean_message)
|
||||||
|
except ValueError as exc:
|
||||||
|
errors.append(self._import_error(index, session_id, str(exc)))
|
||||||
|
continue
|
||||||
|
|
||||||
|
total_messages += len(clean_messages)
|
||||||
|
if total_messages > self._IMPORT_MAX_TOTAL_MESSAGES:
|
||||||
|
errors.append(
|
||||||
|
self._import_error(
|
||||||
|
index,
|
||||||
|
session_id,
|
||||||
|
"messages exceeds the total import limit",
|
||||||
|
)
|
||||||
|
)
|
||||||
|
continue
|
||||||
|
seen_ids.add(session_id)
|
||||||
|
normalized.append(
|
||||||
|
{"index": index, "session": clean_session, "messages": clean_messages}
|
||||||
|
)
|
||||||
|
|
||||||
|
if errors:
|
||||||
|
return {
|
||||||
|
"ok": False,
|
||||||
|
"imported": 0,
|
||||||
|
"skipped": 0,
|
||||||
|
"detached": 0,
|
||||||
|
"errors": errors,
|
||||||
|
}
|
||||||
|
|
||||||
|
def _do(conn):
|
||||||
|
imported_ids: List[str] = []
|
||||||
|
skipped_ids: List[str] = []
|
||||||
|
parent_updates: List[tuple[str, str]] = []
|
||||||
|
detached = 0
|
||||||
|
|
||||||
|
for item in normalized:
|
||||||
|
raw = item["session"]
|
||||||
|
messages = item["messages"]
|
||||||
|
session_id = str(raw.get("id") or "").strip()
|
||||||
|
exists = conn.execute(
|
||||||
|
"SELECT 1 FROM sessions WHERE id = ? LIMIT 1",
|
||||||
|
(session_id,),
|
||||||
|
).fetchone()
|
||||||
|
if exists:
|
||||||
|
skipped_ids.append(session_id)
|
||||||
|
continue
|
||||||
|
|
||||||
|
started_at = self._float_or_none(raw.get("started_at"))
|
||||||
|
if started_at is None:
|
||||||
|
started_at = time.time()
|
||||||
|
archived = 1 if raw.get("archived") else 0
|
||||||
|
|
||||||
|
conn.execute(
|
||||||
|
"""INSERT INTO sessions (
|
||||||
|
id, source, user_id, model, model_config, system_prompt,
|
||||||
|
parent_session_id, started_at, ended_at, end_reason,
|
||||||
|
message_count, tool_call_count, input_tokens, output_tokens,
|
||||||
|
cache_read_tokens, cache_write_tokens, reasoning_tokens,
|
||||||
|
cwd, git_branch, git_repo_root,
|
||||||
|
billing_provider, billing_base_url, billing_mode,
|
||||||
|
estimated_cost_usd, actual_cost_usd, cost_status, cost_source,
|
||||||
|
pricing_version, title, api_call_count, archived
|
||||||
|
)
|
||||||
|
VALUES (
|
||||||
|
:id, :source, :user_id, :model, :model_config,
|
||||||
|
:system_prompt, NULL, :started_at, :ended_at,
|
||||||
|
:end_reason, 0, 0, :input_tokens, :output_tokens,
|
||||||
|
:cache_read_tokens, :cache_write_tokens,
|
||||||
|
:reasoning_tokens, :cwd, :git_branch, :git_repo_root,
|
||||||
|
:billing_provider, :billing_base_url, :billing_mode,
|
||||||
|
:estimated_cost_usd, :actual_cost_usd, :cost_status,
|
||||||
|
:cost_source, :pricing_version, :title,
|
||||||
|
:api_call_count, :archived
|
||||||
|
)""",
|
||||||
|
{
|
||||||
|
"id": session_id,
|
||||||
|
"source": str(raw.get("source") or "import"),
|
||||||
|
"user_id": raw.get("user_id"),
|
||||||
|
"model": raw.get("model"),
|
||||||
|
"model_config": raw.get("model_config"),
|
||||||
|
"system_prompt": raw.get("system_prompt"),
|
||||||
|
"started_at": started_at,
|
||||||
|
"ended_at": self._float_or_none(raw.get("ended_at")),
|
||||||
|
"end_reason": raw.get("end_reason"),
|
||||||
|
"input_tokens": self._int_or_default(raw.get("input_tokens")),
|
||||||
|
"output_tokens": self._int_or_default(raw.get("output_tokens")),
|
||||||
|
"cache_read_tokens": self._int_or_default(
|
||||||
|
raw.get("cache_read_tokens")
|
||||||
|
),
|
||||||
|
"cache_write_tokens": self._int_or_default(
|
||||||
|
raw.get("cache_write_tokens")
|
||||||
|
),
|
||||||
|
"reasoning_tokens": self._int_or_default(
|
||||||
|
raw.get("reasoning_tokens")
|
||||||
|
),
|
||||||
|
"cwd": raw.get("cwd"),
|
||||||
|
"git_branch": raw.get("git_branch"),
|
||||||
|
"git_repo_root": raw.get("git_repo_root"),
|
||||||
|
"billing_provider": raw.get("billing_provider"),
|
||||||
|
"billing_base_url": raw.get("billing_base_url"),
|
||||||
|
"billing_mode": raw.get("billing_mode"),
|
||||||
|
"estimated_cost_usd": self._float_or_none(
|
||||||
|
raw.get("estimated_cost_usd")
|
||||||
|
),
|
||||||
|
"actual_cost_usd": self._float_or_none(
|
||||||
|
raw.get("actual_cost_usd")
|
||||||
|
),
|
||||||
|
"cost_status": raw.get("cost_status"),
|
||||||
|
"cost_source": raw.get("cost_source"),
|
||||||
|
"pricing_version": raw.get("pricing_version"),
|
||||||
|
"title": raw.get("title"),
|
||||||
|
"api_call_count": self._int_or_default(raw.get("api_call_count")),
|
||||||
|
"archived": archived,
|
||||||
|
},
|
||||||
|
)
|
||||||
|
|
||||||
|
sanitized_messages: List[Dict[str, Any]] = []
|
||||||
|
for msg in messages:
|
||||||
|
clean = dict(msg)
|
||||||
|
for key in (
|
||||||
|
"reasoning_details",
|
||||||
|
"codex_reasoning_items",
|
||||||
|
"codex_message_items",
|
||||||
|
):
|
||||||
|
clean[key] = self._reasoning_json_value(clean.get(key))
|
||||||
|
sanitized_messages.append(clean)
|
||||||
|
|
||||||
|
total_messages, total_tool_calls = self._insert_message_rows(
|
||||||
|
conn,
|
||||||
|
session_id,
|
||||||
|
sanitized_messages,
|
||||||
|
)
|
||||||
|
conn.execute(
|
||||||
|
"UPDATE sessions SET message_count = ?, tool_call_count = ? WHERE id = ?",
|
||||||
|
(total_messages, total_tool_calls, session_id),
|
||||||
|
)
|
||||||
|
|
||||||
|
parent_id = str(raw.get("parent_session_id") or "").strip()
|
||||||
|
if parent_id:
|
||||||
|
parent_updates.append((session_id, parent_id))
|
||||||
|
imported_ids.append(session_id)
|
||||||
|
|
||||||
|
parent_by_child = dict(parent_updates)
|
||||||
|
|
||||||
|
def _would_create_cycle(session_id: str, parent_id: str) -> bool:
|
||||||
|
seen = {session_id}
|
||||||
|
current = parent_id
|
||||||
|
while current:
|
||||||
|
if current in seen:
|
||||||
|
return True
|
||||||
|
seen.add(current)
|
||||||
|
if current in parent_by_child:
|
||||||
|
current = parent_by_child[current]
|
||||||
|
continue
|
||||||
|
row = conn.execute(
|
||||||
|
"SELECT parent_session_id FROM sessions WHERE id = ? LIMIT 1",
|
||||||
|
(current,),
|
||||||
|
).fetchone()
|
||||||
|
if row is None:
|
||||||
|
return False
|
||||||
|
current = row["parent_session_id"]
|
||||||
|
return False
|
||||||
|
|
||||||
|
for session_id, parent_id in parent_updates:
|
||||||
|
parent_exists = conn.execute(
|
||||||
|
"SELECT 1 FROM sessions WHERE id = ? LIMIT 1",
|
||||||
|
(parent_id,),
|
||||||
|
).fetchone()
|
||||||
|
if parent_exists and not _would_create_cycle(session_id, parent_id):
|
||||||
|
conn.execute(
|
||||||
|
"UPDATE sessions SET parent_session_id = ? WHERE id = ?",
|
||||||
|
(parent_id, session_id),
|
||||||
|
)
|
||||||
|
else:
|
||||||
|
# Drop only the closing edge. Later entries can still attach
|
||||||
|
# to this now-root session, preserving the acyclic portion
|
||||||
|
# of a malformed imported lineage.
|
||||||
|
parent_by_child.pop(session_id, None)
|
||||||
|
detached += 1
|
||||||
|
|
||||||
|
return {
|
||||||
|
"ok": True,
|
||||||
|
"imported": len(imported_ids),
|
||||||
|
"skipped": len(skipped_ids),
|
||||||
|
"detached": detached,
|
||||||
|
"imported_ids": imported_ids,
|
||||||
|
"skipped_ids": skipped_ids,
|
||||||
|
"errors": [],
|
||||||
|
}
|
||||||
|
|
||||||
|
return self._execute_write(_do)
|
||||||
@@ -0,0 +1,779 @@
|
|||||||
|
"""Schema creation, column reconciliation, and FTS DDL management for SessionDB.
|
||||||
|
|
||||||
|
Mixin contract: this is a plain mixin class consumed by
|
||||||
|
``hermes_state.SessionDB``. It defines no ``__init__`` and no state of its
|
||||||
|
own; methods access the host's attributes (``self._conn``, ``self.db_path``,
|
||||||
|
``self._execute_write`` and other SessionDB methods) established by
|
||||||
|
``SessionDB.__init__``. It must never import hermes_state (cycle) — shared
|
||||||
|
module-level constants live in hermes_state_common.
|
||||||
|
"""
|
||||||
|
|
||||||
|
import logging
|
||||||
|
import json
|
||||||
|
import sqlite3
|
||||||
|
from typing import Dict, Optional
|
||||||
|
|
||||||
|
from hermes_constants import get_hermes_home
|
||||||
|
from hermes_state_common import (
|
||||||
|
DEFERRED_INDEX_SQL,
|
||||||
|
FTS_SQL,
|
||||||
|
FTS_STORAGE_VERSION,
|
||||||
|
FTS_TRIGRAM_SQL,
|
||||||
|
LEGACY_FTS_SQL,
|
||||||
|
LEGACY_FTS_TRIGRAM_SQL,
|
||||||
|
SCHEMA_SQL,
|
||||||
|
SCHEMA_VERSION,
|
||||||
|
_FTS_TRIGGERS,
|
||||||
|
_ephemeral_child_sql,
|
||||||
|
)
|
||||||
|
|
||||||
|
# Moved methods logged under the "hermes_state" logger before the split;
|
||||||
|
# keep that logger identity so log filtering/capture behavior is unchanged.
|
||||||
|
logger = logging.getLogger("hermes_state")
|
||||||
|
|
||||||
|
|
||||||
|
class SessionSchemaMixin:
|
||||||
|
"""See module docstring — mixin for SessionDB (Schema cluster)."""
|
||||||
|
|
||||||
|
def _sqlite_supports_fts5(self, cursor: sqlite3.Cursor) -> bool:
|
||||||
|
try:
|
||||||
|
cursor.execute("CREATE VIRTUAL TABLE temp._hermes_fts5_probe USING fts5(x)")
|
||||||
|
cursor.execute("DROP TABLE temp._hermes_fts5_probe")
|
||||||
|
return True
|
||||||
|
except sqlite3.OperationalError as exc:
|
||||||
|
if not self._is_fts5_unavailable_error(exc):
|
||||||
|
raise
|
||||||
|
self._warn_fts5_unavailable(exc)
|
||||||
|
return False
|
||||||
|
|
||||||
|
@staticmethod
|
||||||
|
def _fts_trigger_count(cursor: sqlite3.Cursor) -> int:
|
||||||
|
placeholders = ",".join("?" for _ in _FTS_TRIGGERS)
|
||||||
|
row = cursor.execute(
|
||||||
|
f"SELECT COUNT(*) FROM sqlite_master "
|
||||||
|
f"WHERE type = 'trigger' AND name IN ({placeholders})",
|
||||||
|
_FTS_TRIGGERS,
|
||||||
|
).fetchone()
|
||||||
|
return int(row[0] if not isinstance(row, sqlite3.Row) else row[0])
|
||||||
|
|
||||||
|
@staticmethod
|
||||||
|
def _rebuild_fts_indexes(
|
||||||
|
cursor: sqlite3.Cursor,
|
||||||
|
*,
|
||||||
|
include_trigram: bool = True,
|
||||||
|
) -> None:
|
||||||
|
# Both FTS tables are external-content (v23+): the special 'rebuild'
|
||||||
|
# command wipes the inverted index and repopulates it from the
|
||||||
|
# content source (messages for the standard index, the tool-row-
|
||||||
|
# excluding messages_fts_trigram_src view for the trigram index).
|
||||||
|
cursor.execute("INSERT INTO messages_fts(messages_fts) VALUES('rebuild')")
|
||||||
|
if include_trigram:
|
||||||
|
cursor.execute(
|
||||||
|
"INSERT INTO messages_fts_trigram(messages_fts_trigram) VALUES('rebuild')"
|
||||||
|
)
|
||||||
|
# 'rebuild' indexes EVERY row, so any deferred-backfill markers are
|
||||||
|
# now satisfied — clear them, otherwise the background worker would
|
||||||
|
# re-insert rows the rebuild already covered (duplicate entries).
|
||||||
|
cursor.execute(
|
||||||
|
"DELETE FROM state_meta WHERE key IN "
|
||||||
|
"('fts_rebuild_high_water', 'fts_rebuild_progress')"
|
||||||
|
)
|
||||||
|
|
||||||
|
@staticmethod
|
||||||
|
def _rebuild_legacy_fts_indexes(
|
||||||
|
cursor: sqlite3.Cursor,
|
||||||
|
*,
|
||||||
|
include_trigram: bool = True,
|
||||||
|
) -> None:
|
||||||
|
"""Rebuild the LEGACY inline FTS indexes (pre-v23) from messages.
|
||||||
|
|
||||||
|
Used only to repair a legacy DB whose triggers degraded under an
|
||||||
|
earlier no-FTS5 runtime. Inline tables have no external-content
|
||||||
|
'rebuild' source, so we DELETE + reinsert the concatenated content
|
||||||
|
the legacy triggers produced. Never touches the v23 shape.
|
||||||
|
"""
|
||||||
|
cursor.execute("DELETE FROM messages_fts")
|
||||||
|
cursor.execute(
|
||||||
|
"INSERT INTO messages_fts(rowid, content) "
|
||||||
|
"SELECT id, "
|
||||||
|
"COALESCE(content, '') || ' ' || "
|
||||||
|
"COALESCE(tool_name, '') || ' ' || "
|
||||||
|
"COALESCE(tool_calls, '') "
|
||||||
|
"FROM messages"
|
||||||
|
)
|
||||||
|
if not include_trigram:
|
||||||
|
return
|
||||||
|
cursor.execute("DELETE FROM messages_fts_trigram")
|
||||||
|
cursor.execute(
|
||||||
|
"INSERT INTO messages_fts_trigram(rowid, content) "
|
||||||
|
"SELECT id, "
|
||||||
|
"COALESCE(content, '') || ' ' || "
|
||||||
|
"COALESCE(tool_name, '') || ' ' || "
|
||||||
|
"COALESCE(tool_calls, '') "
|
||||||
|
"FROM messages"
|
||||||
|
)
|
||||||
|
|
||||||
|
def _fts_table_probe(self, cursor: sqlite3.Cursor, table_name: str) -> Optional[bool]:
|
||||||
|
try:
|
||||||
|
cursor.execute(f"SELECT * FROM {table_name} LIMIT 0")
|
||||||
|
return True
|
||||||
|
except sqlite3.OperationalError as exc:
|
||||||
|
if self._is_fts5_unavailable_error(exc):
|
||||||
|
# Only disable FTS entirely when the whole module is missing.
|
||||||
|
# A missing trigram tokenizer only affects trigram searches.
|
||||||
|
if self._is_trigram_unavailable_error(exc):
|
||||||
|
self._warn_trigram_unavailable(exc)
|
||||||
|
else:
|
||||||
|
self._warn_fts5_unavailable(exc)
|
||||||
|
return None
|
||||||
|
if "no such table" in str(exc).lower():
|
||||||
|
return False
|
||||||
|
raise
|
||||||
|
|
||||||
|
@staticmethod
|
||||||
|
def _parse_schema_columns(schema_sql: str) -> Dict[str, Dict[str, str]]:
|
||||||
|
"""Extract expected columns per table from SCHEMA_SQL.
|
||||||
|
|
||||||
|
Uses an in-memory SQLite database to parse the SQL — SQLite itself
|
||||||
|
handles all syntax (DEFAULT expressions with commas, inline
|
||||||
|
REFERENCES, CHECK constraints, etc.) so there are zero regex
|
||||||
|
edge cases. The in-memory DB is opened, the schema DDL is
|
||||||
|
executed, and PRAGMA table_info extracts the column metadata.
|
||||||
|
|
||||||
|
Adding a column to SCHEMA_SQL is all that's needed; the
|
||||||
|
reconciliation loop picks it up automatically.
|
||||||
|
"""
|
||||||
|
ref = sqlite3.connect(":memory:")
|
||||||
|
try:
|
||||||
|
ref.executescript(schema_sql)
|
||||||
|
table_columns: Dict[str, Dict[str, str]] = {}
|
||||||
|
for (tbl,) in ref.execute(
|
||||||
|
"SELECT name FROM sqlite_master "
|
||||||
|
"WHERE type='table' AND name NOT LIKE 'sqlite_%'"
|
||||||
|
).fetchall():
|
||||||
|
cols: Dict[str, str] = {}
|
||||||
|
for row in ref.execute(
|
||||||
|
f'PRAGMA table_info("{tbl}")'
|
||||||
|
).fetchall():
|
||||||
|
# row: (cid, name, type, notnull, dflt_value, pk)
|
||||||
|
col_name = row[1]
|
||||||
|
col_type = row[2] or ""
|
||||||
|
notnull = row[3]
|
||||||
|
default = row[4]
|
||||||
|
pk = row[5]
|
||||||
|
# Reconstruct the type expression for ALTER TABLE ADD COLUMN
|
||||||
|
parts = [col_type] if col_type else []
|
||||||
|
if notnull and not pk:
|
||||||
|
parts.append("NOT NULL")
|
||||||
|
if default is not None:
|
||||||
|
parts.append(f"DEFAULT {default}")
|
||||||
|
cols[col_name] = " ".join(parts)
|
||||||
|
table_columns[tbl] = cols
|
||||||
|
return table_columns
|
||||||
|
finally:
|
||||||
|
ref.close()
|
||||||
|
|
||||||
|
def _reconcile_columns(self, cursor: sqlite3.Cursor) -> None:
|
||||||
|
"""Ensure live tables have every column declared in SCHEMA_SQL.
|
||||||
|
|
||||||
|
Follows the Beets/sqlite-utils pattern: the CREATE TABLE definition
|
||||||
|
in SCHEMA_SQL is the single source of truth for the desired schema.
|
||||||
|
On every startup this method diffs the live columns (via PRAGMA
|
||||||
|
table_info) against the declared columns, and ADDs any that are
|
||||||
|
missing.
|
||||||
|
|
||||||
|
This makes column additions a declarative operation — just add
|
||||||
|
the column to SCHEMA_SQL and it appears on the next startup.
|
||||||
|
Version-gated migration blocks are no longer needed for ADD COLUMN.
|
||||||
|
"""
|
||||||
|
expected = self._parse_schema_columns(SCHEMA_SQL)
|
||||||
|
for table_name, declared_cols in expected.items():
|
||||||
|
# Get current columns from the live table
|
||||||
|
try:
|
||||||
|
rows = cursor.execute(
|
||||||
|
f'PRAGMA table_info("{table_name}")'
|
||||||
|
).fetchall()
|
||||||
|
except sqlite3.OperationalError:
|
||||||
|
continue # Table doesn't exist yet (shouldn't happen after executescript)
|
||||||
|
live_cols = set()
|
||||||
|
for row in rows:
|
||||||
|
# PRAGMA table_info returns (cid, name, type, notnull, dflt_value, pk)
|
||||||
|
name = row[1] if isinstance(row, (tuple, list)) else row["name"]
|
||||||
|
live_cols.add(name)
|
||||||
|
|
||||||
|
for col_name, col_type in declared_cols.items():
|
||||||
|
if col_name not in live_cols:
|
||||||
|
safe_name = col_name.replace('"', '""')
|
||||||
|
try:
|
||||||
|
cursor.execute(
|
||||||
|
f'ALTER TABLE "{table_name}" ADD COLUMN "{safe_name}" {col_type}'
|
||||||
|
)
|
||||||
|
except sqlite3.OperationalError as exc:
|
||||||
|
# Expected: "duplicate column name" from a race or
|
||||||
|
# re-run. Unexpected: "Cannot add a NOT NULL column
|
||||||
|
# with default value NULL" from a schema mistake.
|
||||||
|
# Log at DEBUG so it's visible in agent.log.
|
||||||
|
logger.debug(
|
||||||
|
"reconcile %s.%s: %s", table_name, col_name, exc,
|
||||||
|
)
|
||||||
|
|
||||||
|
def _heal_gateway_routing_pk(self, cursor: sqlite3.Cursor) -> None:
|
||||||
|
"""Rebuild ``gateway_routing`` when its PRIMARY KEY predates scoping.
|
||||||
|
|
||||||
|
Early builds of the routing-index migration (#59203) created the
|
||||||
|
table with ``session_key TEXT PRIMARY KEY`` and no ``scope`` column.
|
||||||
|
``_reconcile_columns()`` ADDs the missing ``scope`` column on those
|
||||||
|
databases, but SQLite cannot ALTER a primary key, so the shipped
|
||||||
|
composite ``PRIMARY KEY (scope, session_key)`` never lands. On such
|
||||||
|
tables every write path is broken:
|
||||||
|
|
||||||
|
* ``save_gateway_routing_entry`` fails with "ON CONFLICT clause does
|
||||||
|
not match any PRIMARY KEY or UNIQUE constraint" (its upsert targets
|
||||||
|
the composite key), and
|
||||||
|
* ``replace_gateway_routing_entries`` fails with "UNIQUE constraint
|
||||||
|
failed: gateway_routing.session_key" whenever the same session_key
|
||||||
|
exists under a different scope — the exact isolation the composite
|
||||||
|
key exists to provide.
|
||||||
|
|
||||||
|
Each failed save logs a warning and falls back to sessions.json,
|
||||||
|
so a legacy-shaped table produces endless per-save warning spam.
|
||||||
|
Rebuild it once, preserving rows. On a session_key collision across
|
||||||
|
scopes (possible while the PK was wrong) the newest row wins.
|
||||||
|
"""
|
||||||
|
try:
|
||||||
|
rows = cursor.execute(
|
||||||
|
'PRAGMA table_info("gateway_routing")'
|
||||||
|
).fetchall()
|
||||||
|
except sqlite3.OperationalError:
|
||||||
|
return
|
||||||
|
if not rows:
|
||||||
|
return
|
||||||
|
|
||||||
|
def _col(row, idx, name):
|
||||||
|
return row[idx] if isinstance(row, (tuple, list)) else row[name]
|
||||||
|
|
||||||
|
pk_cols = [
|
||||||
|
_col(r, 1, "name")
|
||||||
|
for r in sorted(
|
||||||
|
(r for r in rows if _col(r, 5, "pk")),
|
||||||
|
key=lambda r: _col(r, 5, "pk"),
|
||||||
|
)
|
||||||
|
]
|
||||||
|
if pk_cols == ["scope", "session_key"]:
|
||||||
|
return
|
||||||
|
|
||||||
|
logger.info(
|
||||||
|
"gateway_routing has legacy primary key %r; rebuilding with "
|
||||||
|
"composite (scope, session_key) key",
|
||||||
|
pk_cols,
|
||||||
|
)
|
||||||
|
cursor.execute(
|
||||||
|
"ALTER TABLE gateway_routing RENAME TO gateway_routing_legacy_pk"
|
||||||
|
)
|
||||||
|
cursor.execute(
|
||||||
|
"""CREATE TABLE gateway_routing (
|
||||||
|
scope TEXT NOT NULL DEFAULT '',
|
||||||
|
session_key TEXT NOT NULL,
|
||||||
|
entry_json TEXT NOT NULL,
|
||||||
|
updated_at REAL NOT NULL,
|
||||||
|
PRIMARY KEY (scope, session_key)
|
||||||
|
)"""
|
||||||
|
)
|
||||||
|
# INSERT OR REPLACE + updated_at ordering: if the broken PK ever let
|
||||||
|
# two scopes race over one session_key, keep the newest row per
|
||||||
|
# (scope, session_key) pair.
|
||||||
|
cursor.execute(
|
||||||
|
"INSERT OR REPLACE INTO gateway_routing "
|
||||||
|
"(scope, session_key, entry_json, updated_at) "
|
||||||
|
"SELECT COALESCE(scope, ''), session_key, entry_json, updated_at "
|
||||||
|
"FROM gateway_routing_legacy_pk ORDER BY updated_at ASC"
|
||||||
|
)
|
||||||
|
cursor.execute("DROP TABLE gateway_routing_legacy_pk")
|
||||||
|
|
||||||
|
def _init_schema(self):
|
||||||
|
"""Create tables and FTS if they don't exist, reconcile columns.
|
||||||
|
|
||||||
|
Schema management follows the declarative reconciliation pattern
|
||||||
|
(Beets, sqlite-utils): SCHEMA_SQL is the single source of truth.
|
||||||
|
On existing databases, _reconcile_columns() diffs live columns
|
||||||
|
against SCHEMA_SQL and ADDs any missing ones. This eliminates
|
||||||
|
the version-gated migration chain for column additions, making
|
||||||
|
it impossible for reordered or inserted migrations to skip columns.
|
||||||
|
|
||||||
|
The schema_version table is retained for future data migrations
|
||||||
|
(transforming existing rows) which cannot be handled declaratively.
|
||||||
|
"""
|
||||||
|
cursor = self._conn.cursor()
|
||||||
|
|
||||||
|
cursor.executescript(SCHEMA_SQL)
|
||||||
|
|
||||||
|
# ── Declarative column reconciliation ──────────────────────────
|
||||||
|
# Diff live tables against SCHEMA_SQL and ADD any missing columns.
|
||||||
|
# This is idempotent and self-healing: even if a version-gated
|
||||||
|
# migration was skipped (e.g. due to version renumbering), the
|
||||||
|
# column gets created here.
|
||||||
|
self._reconcile_columns(cursor)
|
||||||
|
|
||||||
|
# Rebuild gateway_routing if it still carries the pre-scope PRIMARY
|
||||||
|
# KEY (session_key alone). ADD COLUMN cannot fix a PK, so this is
|
||||||
|
# the one table-shape repair reconciliation can't express.
|
||||||
|
self._heal_gateway_routing_pk(cursor)
|
||||||
|
|
||||||
|
# Indexes that reference reconciler-added columns must be created
|
||||||
|
# AFTER _reconcile_columns runs — declaring them in SCHEMA_SQL
|
||||||
|
# makes the initial executescript fail on legacy DBs (the index's
|
||||||
|
# WHERE clause references a column that doesn't exist yet).
|
||||||
|
try:
|
||||||
|
cursor.execute(
|
||||||
|
"CREATE INDEX IF NOT EXISTS idx_messages_platform_msg_id "
|
||||||
|
"ON messages(session_id, platform_message_id) "
|
||||||
|
"WHERE platform_message_id IS NOT NULL"
|
||||||
|
)
|
||||||
|
except sqlite3.OperationalError as exc:
|
||||||
|
logger.debug("idx_messages_platform_msg_id create skipped: %s", exc)
|
||||||
|
|
||||||
|
# Deferred indexes that reference the reconciler-added ``active``
|
||||||
|
# column (idx_messages_session_active) — same ordering constraint.
|
||||||
|
cursor.executescript(DEFERRED_INDEX_SQL)
|
||||||
|
|
||||||
|
# Heal NULL ``active`` rows unconditionally on every startup.
|
||||||
|
# On real-world DBs the reconciler-added ``active`` column can lack
|
||||||
|
# its NOT NULL DEFAULT 1 (older reconciler builds reconstructed the
|
||||||
|
# type without the default — see #51646: PRAGMA shows
|
||||||
|
# (17,'active','INTEGER',0,None,0) in the wild), so INSERTs that
|
||||||
|
# omitted the column wrote NULL and the ``WHERE active = 1``
|
||||||
|
# transcript loaders hid the whole history. The INSERTs now set
|
||||||
|
# active=1 explicitly; this idempotent repair un-hides rows written
|
||||||
|
# before the fix. It was previously gated at ``current_version <
|
||||||
|
# 12`` which never re-ran for already-v12+ databases.
|
||||||
|
try:
|
||||||
|
cursor.execute(
|
||||||
|
"UPDATE messages SET active = 1 WHERE active IS NULL"
|
||||||
|
)
|
||||||
|
except sqlite3.OperationalError:
|
||||||
|
pass
|
||||||
|
|
||||||
|
fts5_available = self._sqlite_supports_fts5(cursor)
|
||||||
|
fts_migrations_complete = True
|
||||||
|
if not fts5_available:
|
||||||
|
# Existing FTS triggers can still fire on messages INSERT/UPDATE
|
||||||
|
# even though the current sqlite runtime cannot read the virtual
|
||||||
|
# tables they target. Drop only the triggers so core persistence
|
||||||
|
# continues; if a future runtime has FTS5, _ensure_fts_schema()
|
||||||
|
# recreates them.
|
||||||
|
self._drop_fts_triggers(cursor)
|
||||||
|
|
||||||
|
# ── Schema version bookkeeping ─────────────────────────────────
|
||||||
|
# Bump to current so future data migrations (if any) can gate on
|
||||||
|
# version. No version-gated column additions remain.
|
||||||
|
cursor.execute("SELECT version FROM schema_version LIMIT 1")
|
||||||
|
row = cursor.fetchone()
|
||||||
|
if row is None:
|
||||||
|
cursor.execute(
|
||||||
|
"INSERT INTO schema_version (version) VALUES (?)",
|
||||||
|
(SCHEMA_VERSION,),
|
||||||
|
)
|
||||||
|
else:
|
||||||
|
current_version = row["version"] if isinstance(row, sqlite3.Row) else row[0]
|
||||||
|
# Data migrations that can't be expressed declaratively (row
|
||||||
|
# backfills, index changes tied to a specific version step) stay
|
||||||
|
# in a version-gated chain. Column additions are handled by
|
||||||
|
# _reconcile_columns() above and no longer need entries here.
|
||||||
|
if current_version < 10 and SCHEMA_VERSION == 10:
|
||||||
|
# v10: trigram FTS5 table for CJK/substring search. The
|
||||||
|
# virtual table + triggers are created unconditionally via
|
||||||
|
# FTS_TRIGRAM_SQL below, but existing rows need a one-time
|
||||||
|
# backfill into the FTS index.
|
||||||
|
#
|
||||||
|
# Only run this when v10 itself is the target schema. Current
|
||||||
|
# v11+ code drops and rebuilds both FTS tables below, so doing
|
||||||
|
# the v10-only trigram backfill first only burns startup time
|
||||||
|
# and WAL space before v11 throws the work away.
|
||||||
|
if fts5_available:
|
||||||
|
_fts_trigram_exists = self._fts_table_probe(
|
||||||
|
cursor, "messages_fts_trigram"
|
||||||
|
)
|
||||||
|
if _fts_trigram_exists is False:
|
||||||
|
if self._ensure_fts_schema(
|
||||||
|
cursor, "messages_fts_trigram", FTS_TRIGRAM_SQL
|
||||||
|
):
|
||||||
|
cursor.execute(
|
||||||
|
"INSERT INTO messages_fts_trigram(rowid, content) "
|
||||||
|
"SELECT id, content FROM messages WHERE content IS NOT NULL"
|
||||||
|
)
|
||||||
|
else:
|
||||||
|
fts_migrations_complete = False
|
||||||
|
elif _fts_trigram_exists is None:
|
||||||
|
fts_migrations_complete = False
|
||||||
|
else:
|
||||||
|
fts_migrations_complete = False
|
||||||
|
if current_version < 11 and SCHEMA_VERSION < 23:
|
||||||
|
# v11 (SUPERSEDED by v23): re-index FTS5 tables to cover
|
||||||
|
# tool_name + tool_calls in inline mode (#16751). v23 drops
|
||||||
|
# and rebuilds both FTS tables in external-content form, so
|
||||||
|
# running the v11 inline backfill first would only burn
|
||||||
|
# startup time and WAL space before v23 throws the work
|
||||||
|
# away — and its inline INSERT shape no longer matches the
|
||||||
|
# current external-content FTS_SQL anyway. Kept only for
|
||||||
|
# source archaeology; unreachable while SCHEMA_VERSION >= 23.
|
||||||
|
pass
|
||||||
|
if current_version < 16:
|
||||||
|
# v16: tag delegate subagent rows so pickers stay clean after
|
||||||
|
# parent deletes that used to orphan them (parent_session_id → NULL).
|
||||||
|
try:
|
||||||
|
cursor.execute(
|
||||||
|
"UPDATE sessions SET model_config = json_set("
|
||||||
|
"COALESCE(model_config, '{}'), '$._delegate_from', parent_session_id) "
|
||||||
|
f"WHERE parent_session_id IS NOT NULL "
|
||||||
|
"AND json_extract(COALESCE(model_config, '{}'), '$._delegate_from') IS NULL "
|
||||||
|
f"AND {_ephemeral_child_sql('sessions')}"
|
||||||
|
)
|
||||||
|
cursor.execute(
|
||||||
|
"UPDATE sessions SET model_config = json_set("
|
||||||
|
"COALESCE(model_config, '{}'), '$._delegate_from', '__orphaned__') "
|
||||||
|
"WHERE parent_session_id IS NULL "
|
||||||
|
"AND json_extract(COALESCE(model_config, '{}'), '$._delegate_from') IS NULL "
|
||||||
|
"AND json_extract(COALESCE(model_config, '{}'), '$._branched_from') IS NULL "
|
||||||
|
"AND title IS NULL "
|
||||||
|
"AND message_count <= 25 "
|
||||||
|
"AND EXISTS (SELECT 1 FROM messages m "
|
||||||
|
" WHERE m.session_id = sessions.id AND m.role = 'tool') "
|
||||||
|
"AND NOT EXISTS (SELECT 1 FROM sessions ch "
|
||||||
|
" WHERE ch.parent_session_id = sessions.id)"
|
||||||
|
)
|
||||||
|
except sqlite3.OperationalError:
|
||||||
|
pass
|
||||||
|
if current_version < 18:
|
||||||
|
# v18: gateway metadata consolidation (#9006). Backfill
|
||||||
|
# display_name / origin_json / expiry_finalized from
|
||||||
|
# sessions.json so pre-migration gateway sessions are
|
||||||
|
# discoverable from state.db without the JSON index.
|
||||||
|
try:
|
||||||
|
self._backfill_gateway_metadata_from_sessions_json(cursor)
|
||||||
|
except Exception as exc:
|
||||||
|
# Backfill is best-effort: sessions.json may be absent,
|
||||||
|
# corrupted, or partially stale. Missing metadata simply
|
||||||
|
# means consumers fall back to sessions.json for those
|
||||||
|
# rows until the gateway rewrites them.
|
||||||
|
logger.debug("v18 gateway metadata backfill skipped: %s", exc)
|
||||||
|
if current_version < 20:
|
||||||
|
# v20: per-model usage attribution (issue #51607). Going
|
||||||
|
# forward update_token_counts() records each API call into
|
||||||
|
# session_model_usage keyed by the live model, but existing
|
||||||
|
# sessions only have their aggregate totals on the sessions
|
||||||
|
# row. Seed one usage row per historical session from those
|
||||||
|
# aggregates so insights reads uniformly from the new table.
|
||||||
|
# INSERT OR IGNORE keeps it idempotent: if newer code already
|
||||||
|
# wrote a (session_id, model, provider) row for a session, the
|
||||||
|
# PK conflict skips the stale aggregate rather than doubling it.
|
||||||
|
try:
|
||||||
|
cursor.execute(
|
||||||
|
"""INSERT OR IGNORE INTO session_model_usage (
|
||||||
|
session_id, model, billing_provider,
|
||||||
|
billing_base_url, billing_mode,
|
||||||
|
api_call_count, input_tokens,
|
||||||
|
output_tokens, cache_read_tokens,
|
||||||
|
cache_write_tokens, reasoning_tokens,
|
||||||
|
estimated_cost_usd, actual_cost_usd,
|
||||||
|
cost_status, cost_source, first_seen, last_seen
|
||||||
|
)
|
||||||
|
SELECT id, COALESCE(model, 'unknown'),
|
||||||
|
COALESCE(billing_provider, ''),
|
||||||
|
COALESCE(billing_base_url, ''),
|
||||||
|
COALESCE(billing_mode, ''),
|
||||||
|
COALESCE(api_call_count, 0),
|
||||||
|
COALESCE(input_tokens, 0),
|
||||||
|
COALESCE(output_tokens, 0),
|
||||||
|
COALESCE(cache_read_tokens, 0),
|
||||||
|
COALESCE(cache_write_tokens, 0),
|
||||||
|
COALESCE(reasoning_tokens, 0),
|
||||||
|
COALESCE(estimated_cost_usd, 0),
|
||||||
|
COALESCE(actual_cost_usd, 0),
|
||||||
|
cost_status, cost_source,
|
||||||
|
started_at, COALESCE(ended_at, started_at)
|
||||||
|
FROM sessions
|
||||||
|
WHERE COALESCE(input_tokens, 0)
|
||||||
|
+ COALESCE(output_tokens, 0)
|
||||||
|
+ COALESCE(cache_read_tokens, 0)
|
||||||
|
+ COALESCE(cache_write_tokens, 0)
|
||||||
|
+ COALESCE(reasoning_tokens, 0) > 0"""
|
||||||
|
)
|
||||||
|
except sqlite3.OperationalError:
|
||||||
|
pass
|
||||||
|
if current_version < 22:
|
||||||
|
# v22: task-dimension usage attribution (issue #23270).
|
||||||
|
# session_model_usage gains a ``task`` column ('' = main agent
|
||||||
|
# loop; 'vision'/'compression'/'title_generation'/... =
|
||||||
|
# auxiliary calls) so aux model spend is visible in analytics.
|
||||||
|
# The column participates in the PRIMARY KEY and SQLite cannot
|
||||||
|
# ALTER a PK, so rebuild the table. The reconciler will have
|
||||||
|
# already ADDed the plain column on legacy DBs (harmless);
|
||||||
|
# the rebuild bakes it into the PK properly. Existing rows are
|
||||||
|
# main-loop accounting by definition → task=''.
|
||||||
|
try:
|
||||||
|
legacy_pk = cursor.execute(
|
||||||
|
"SELECT COUNT(*) FROM pragma_table_info('session_model_usage') "
|
||||||
|
"WHERE name = 'task' AND pk > 0"
|
||||||
|
).fetchone()[0]
|
||||||
|
if not legacy_pk:
|
||||||
|
cursor.execute("ALTER TABLE session_model_usage RENAME TO session_model_usage_v21")
|
||||||
|
cursor.execute(
|
||||||
|
"""CREATE TABLE session_model_usage (
|
||||||
|
session_id TEXT NOT NULL REFERENCES sessions(id) ON DELETE CASCADE,
|
||||||
|
model TEXT NOT NULL,
|
||||||
|
billing_provider TEXT NOT NULL DEFAULT '',
|
||||||
|
billing_base_url TEXT NOT NULL DEFAULT '',
|
||||||
|
billing_mode TEXT NOT NULL DEFAULT '',
|
||||||
|
task TEXT NOT NULL DEFAULT '',
|
||||||
|
api_call_count INTEGER NOT NULL DEFAULT 0,
|
||||||
|
input_tokens INTEGER NOT NULL DEFAULT 0,
|
||||||
|
output_tokens INTEGER NOT NULL DEFAULT 0,
|
||||||
|
cache_read_tokens INTEGER NOT NULL DEFAULT 0,
|
||||||
|
cache_write_tokens INTEGER NOT NULL DEFAULT 0,
|
||||||
|
reasoning_tokens INTEGER NOT NULL DEFAULT 0,
|
||||||
|
estimated_cost_usd REAL NOT NULL DEFAULT 0,
|
||||||
|
actual_cost_usd REAL NOT NULL DEFAULT 0,
|
||||||
|
cost_status TEXT,
|
||||||
|
cost_source TEXT,
|
||||||
|
first_seen REAL,
|
||||||
|
last_seen REAL,
|
||||||
|
PRIMARY KEY (session_id, model, billing_provider, billing_base_url, billing_mode, task)
|
||||||
|
)"""
|
||||||
|
)
|
||||||
|
cursor.execute(
|
||||||
|
"""INSERT INTO session_model_usage (
|
||||||
|
session_id, model, billing_provider, billing_base_url,
|
||||||
|
billing_mode, task, api_call_count, input_tokens,
|
||||||
|
output_tokens, cache_read_tokens, cache_write_tokens,
|
||||||
|
reasoning_tokens, estimated_cost_usd, actual_cost_usd,
|
||||||
|
cost_status, cost_source, first_seen, last_seen
|
||||||
|
)
|
||||||
|
SELECT session_id, model, billing_provider, billing_base_url,
|
||||||
|
billing_mode, '', api_call_count, input_tokens,
|
||||||
|
output_tokens, cache_read_tokens, cache_write_tokens,
|
||||||
|
reasoning_tokens, estimated_cost_usd, actual_cost_usd,
|
||||||
|
cost_status, cost_source, first_seen, last_seen
|
||||||
|
FROM session_model_usage_v21"""
|
||||||
|
)
|
||||||
|
cursor.execute("DROP TABLE session_model_usage_v21")
|
||||||
|
cursor.execute(
|
||||||
|
"CREATE INDEX IF NOT EXISTS idx_session_model_usage_session "
|
||||||
|
"ON session_model_usage(session_id)"
|
||||||
|
)
|
||||||
|
cursor.execute(
|
||||||
|
"CREATE INDEX IF NOT EXISTS idx_session_model_usage_model "
|
||||||
|
"ON session_model_usage(model)"
|
||||||
|
)
|
||||||
|
except sqlite3.OperationalError as exc:
|
||||||
|
logger.debug("v22 session_model_usage rebuild skipped: %s", exc)
|
||||||
|
if current_version < 23:
|
||||||
|
# v23: FTS storage redesign (issues #22478, #43690, #55233).
|
||||||
|
# The v11 inline-mode FTS tables each store a full private
|
||||||
|
# copy of every message (content || tool_name || tool_calls),
|
||||||
|
# and the trigram index additionally covers role='tool' rows
|
||||||
|
# (~90% of message bytes: base64 payloads, file dumps) at
|
||||||
|
# ~2.6x amplification — together ~75% of state.db on heavy
|
||||||
|
# installs (observed: 18.9 GB of a 25 GB DB).
|
||||||
|
#
|
||||||
|
# OPT-IN, NOT AUTOMATIC. The transition (demote old vtables →
|
||||||
|
# new external-content schema → backfill → teardown → VACUUM)
|
||||||
|
# is disk-heavy (transient ~2x file size to fully reclaim via
|
||||||
|
# VACUUM) and long (~1-2h background on a 25 GB DB). Doing it
|
||||||
|
# silently on every big user's next open — with a completeness
|
||||||
|
# guarantee that depends on the process staying alive long
|
||||||
|
# enough — is the wrong default. So on an EXISTING install we
|
||||||
|
# touch nothing here: the v22 inline FTS keeps working exactly
|
||||||
|
# as before, and we only record a flag advertising that the
|
||||||
|
# optimization is available. `hermes sessions optimize-storage`
|
||||||
|
# performs the whole transition as one deliberate, disk-checked,
|
||||||
|
# progress-reported foreground operation.
|
||||||
|
#
|
||||||
|
# DECOUPLED VERSIONING. Crucially, this does NOT hold back the
|
||||||
|
# main schema_version. The FTS storage LAYOUT is tracked by an
|
||||||
|
# independent `fts_storage_version` marker (see
|
||||||
|
# _fts_storage_version / SETTLE below), so schema_version
|
||||||
|
# advances to SCHEMA_VERSION here like every other migration —
|
||||||
|
# future v24+ migrations land automatically for legacy-FTS
|
||||||
|
# users too. Only the FTS *layout* waits for opt-in.
|
||||||
|
if fts5_available and self._db_has_legacy_inline_fts(cursor):
|
||||||
|
self.set_meta("fts_optimize_available", "1", cursor=cursor)
|
||||||
|
|
||||||
|
# The FTS storage layout is versioned independently of the main
|
||||||
|
# schema (see the v23 note above). Stamp the current layout so the
|
||||||
|
# main version can always advance: a fresh/optimized DB is at
|
||||||
|
# FTS_STORAGE_VERSION; a legacy DB is left at whatever it had
|
||||||
|
# (absent/0) until `optimize-storage` runs. An INTERRUPTED
|
||||||
|
# optimize (legacy vtables already demoted, but rebuild markers
|
||||||
|
# or demoted trash tables still present) is NOT stamped either —
|
||||||
|
# the marker is the source of truth for "fully optimized", and
|
||||||
|
# `fts_optimize_available()` keeps offering the resume until the
|
||||||
|
# transition actually completes.
|
||||||
|
if (
|
||||||
|
fts5_available
|
||||||
|
and not self._db_has_legacy_inline_fts(cursor)
|
||||||
|
and cursor.execute(
|
||||||
|
"SELECT 1 FROM state_meta "
|
||||||
|
"WHERE key = 'fts_rebuild_high_water' LIMIT 1"
|
||||||
|
).fetchone() is None
|
||||||
|
and not self._has_fts_trash(cursor)
|
||||||
|
):
|
||||||
|
self.set_meta(
|
||||||
|
"fts_storage_version", str(FTS_STORAGE_VERSION), cursor=cursor
|
||||||
|
)
|
||||||
|
|
||||||
|
# Advance schema_version to current for ALL non-FTS-layout
|
||||||
|
# migrations. This is deliberately NOT gated on the FTS opt-in —
|
||||||
|
# holding the whole version back would block every future schema
|
||||||
|
# migration for a user who never optimizes. FTS5 being unavailable
|
||||||
|
# is the one case we skip (we can't have created the current FTS
|
||||||
|
# objects, so claiming the current schema would be a lie).
|
||||||
|
if (
|
||||||
|
current_version < SCHEMA_VERSION
|
||||||
|
and fts_migrations_complete
|
||||||
|
and fts5_available
|
||||||
|
):
|
||||||
|
cursor.execute(
|
||||||
|
"UPDATE schema_version SET version = ?",
|
||||||
|
(SCHEMA_VERSION,),
|
||||||
|
)
|
||||||
|
|
||||||
|
# Unique title index — always ensure it exists. Older databases may
|
||||||
|
# contain duplicate aliases from before the constraint was enforced;
|
||||||
|
# preserve every session while letting the newest one retain the alias.
|
||||||
|
title_index_sql = (
|
||||||
|
"CREATE UNIQUE INDEX IF NOT EXISTS idx_sessions_title_unique "
|
||||||
|
"ON sessions(title) WHERE title IS NOT NULL"
|
||||||
|
)
|
||||||
|
try:
|
||||||
|
cursor.execute(title_index_sql)
|
||||||
|
except sqlite3.IntegrityError:
|
||||||
|
# The index is an optimization — its creation must never abort
|
||||||
|
# opening the database, so the repair itself is also guarded.
|
||||||
|
try:
|
||||||
|
cursor.execute(
|
||||||
|
"""UPDATE sessions AS older
|
||||||
|
SET title = NULL
|
||||||
|
WHERE title IS NOT NULL
|
||||||
|
AND EXISTS (
|
||||||
|
SELECT 1 FROM sessions AS newer
|
||||||
|
WHERE newer.title = older.title
|
||||||
|
AND newer.rowid > older.rowid
|
||||||
|
)"""
|
||||||
|
)
|
||||||
|
logger.warning(
|
||||||
|
"Cleared %d duplicate session title(s) while restoring the unique index",
|
||||||
|
cursor.rowcount,
|
||||||
|
)
|
||||||
|
cursor.execute(title_index_sql)
|
||||||
|
except sqlite3.Error:
|
||||||
|
logger.exception(
|
||||||
|
"Could not repair duplicate session titles; "
|
||||||
|
"unique title index not created"
|
||||||
|
)
|
||||||
|
except sqlite3.OperationalError:
|
||||||
|
pass # Index already exists
|
||||||
|
|
||||||
|
if fts5_available:
|
||||||
|
# FTS5 setup. Run the DDL even when the virtual table exists so
|
||||||
|
# CREATE TRIGGER IF NOT EXISTS repairs trigger-only degradation from
|
||||||
|
# an earlier no-FTS5 runtime.
|
||||||
|
#
|
||||||
|
# OPT-IN v23 boundary: a legacy v22 install (inline-content FTS,
|
||||||
|
# not yet opted into `hermes db optimize`) must keep its EXISTING
|
||||||
|
# inline schema + triggers. Running the v23 external-content DDL
|
||||||
|
# here would create the trigram source VIEW and leave the DB in a
|
||||||
|
# mixed inline/external state. So for a legacy DB we only ensure
|
||||||
|
# its inline triggers exist (via the legacy DDL), and skip the
|
||||||
|
# v23 view/external tables entirely. Fresh installs and opted-in
|
||||||
|
# DBs have no legacy inline FTS, so they get the v23 DDL.
|
||||||
|
if self._db_has_legacy_inline_fts(cursor):
|
||||||
|
triggers_need_repair = (
|
||||||
|
self._fts_trigger_count(cursor) < len(_FTS_TRIGGERS)
|
||||||
|
)
|
||||||
|
self._fts_enabled = self._ensure_fts_schema(
|
||||||
|
cursor, "messages_fts", LEGACY_FTS_SQL
|
||||||
|
)
|
||||||
|
if self._fts_enabled:
|
||||||
|
trigram_enabled = self._ensure_fts_schema(
|
||||||
|
cursor, "messages_fts_trigram", LEGACY_FTS_TRIGRAM_SQL
|
||||||
|
)
|
||||||
|
self._trigram_available = trigram_enabled
|
||||||
|
if triggers_need_repair:
|
||||||
|
self._rebuild_legacy_fts_indexes(
|
||||||
|
cursor, include_trigram=trigram_enabled
|
||||||
|
)
|
||||||
|
else:
|
||||||
|
triggers_need_repair = (
|
||||||
|
self._fts_trigger_count(cursor) < len(_FTS_TRIGGERS)
|
||||||
|
)
|
||||||
|
self._fts_enabled = self._ensure_fts_schema(
|
||||||
|
cursor, "messages_fts", FTS_SQL
|
||||||
|
)
|
||||||
|
|
||||||
|
# Trigram FTS5 for CJK/substring search. This is optional
|
||||||
|
# relative to the main FTS table; if it cannot be created,
|
||||||
|
# CJK search falls back to LIKE.
|
||||||
|
if self._fts_enabled:
|
||||||
|
trigram_enabled = self._ensure_fts_schema(
|
||||||
|
cursor, "messages_fts_trigram", FTS_TRIGRAM_SQL
|
||||||
|
)
|
||||||
|
self._trigram_available = trigram_enabled
|
||||||
|
if triggers_need_repair:
|
||||||
|
self._rebuild_fts_indexes(
|
||||||
|
cursor,
|
||||||
|
include_trigram=trigram_enabled,
|
||||||
|
)
|
||||||
|
# CJK-bigram index (cjk_unicode61). Strictly additive to
|
||||||
|
# the surfaces above and gated on the loadable tokenizer:
|
||||||
|
self._ensure_fts_cjk_schema(cursor)
|
||||||
|
|
||||||
|
self._conn.commit()
|
||||||
|
|
||||||
|
def _backfill_gateway_metadata_from_sessions_json(
|
||||||
|
self, cursor: sqlite3.Cursor
|
||||||
|
) -> None:
|
||||||
|
"""One-time v18 backfill of gateway metadata from sessions.json.
|
||||||
|
|
||||||
|
Existing gateway sessions predate the display_name / origin_json /
|
||||||
|
expiry_finalized columns; copy what sessions.json knows so consumers
|
||||||
|
can switch to state.db without losing pre-migration sessions.
|
||||||
|
Only fills NULL columns — never overwrites data written by newer code.
|
||||||
|
"""
|
||||||
|
sessions_file = get_hermes_home() / "sessions" / "sessions.json"
|
||||||
|
if not sessions_file.exists():
|
||||||
|
return
|
||||||
|
with open(sessions_file, "r", encoding="utf-8") as f:
|
||||||
|
data = json.load(f)
|
||||||
|
if not isinstance(data, dict):
|
||||||
|
return
|
||||||
|
for key, entry in data.items():
|
||||||
|
if str(key).startswith("_") or not isinstance(entry, dict):
|
||||||
|
continue
|
||||||
|
session_id = entry.get("session_id")
|
||||||
|
if not session_id:
|
||||||
|
continue
|
||||||
|
origin = entry.get("origin")
|
||||||
|
cursor.execute(
|
||||||
|
"""UPDATE sessions
|
||||||
|
SET session_key = COALESCE(session_key, ?),
|
||||||
|
chat_id = COALESCE(chat_id, ?),
|
||||||
|
chat_type = COALESCE(chat_type, ?),
|
||||||
|
thread_id = COALESCE(thread_id, ?),
|
||||||
|
display_name = COALESCE(display_name, ?),
|
||||||
|
origin_json = COALESCE(origin_json, ?),
|
||||||
|
expiry_finalized = CASE
|
||||||
|
WHEN COALESCE(expiry_finalized, 0) = 0 AND ? = 1 THEN 1
|
||||||
|
ELSE expiry_finalized
|
||||||
|
END
|
||||||
|
WHERE id = ?""",
|
||||||
|
(
|
||||||
|
entry.get("session_key") or key,
|
||||||
|
(origin or {}).get("chat_id") if isinstance(origin, dict) else None,
|
||||||
|
entry.get("chat_type"),
|
||||||
|
(origin or {}).get("thread_id") if isinstance(origin, dict) else None,
|
||||||
|
entry.get("display_name"),
|
||||||
|
json.dumps(origin) if isinstance(origin, dict) else None,
|
||||||
|
1 if entry.get("expiry_finalized") or entry.get("memory_flushed") else 0,
|
||||||
|
str(session_id),
|
||||||
|
),
|
||||||
|
)
|
||||||
File diff suppressed because it is too large
Load Diff
Reference in New Issue
Block a user