refactor(hermes_state): AST-neutral packing of state mixins (120 cols)

This commit is contained in:
Teknium
2026-09-02 19:10:31 -07:00
parent 5cca3715f4
commit c1620901ae
6 changed files with 62 additions and 120 deletions

View File

@@ -17,9 +17,8 @@ from hermes_state_common import SCHEMA_SQL, _PREVIEW_RAW_SUBQUERY_SQL, _shape_pr
logger = logging.getLogger("hermes_state")
_IMPORT_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",
"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",
)
# ``role`` is validated separately (non-empty string).
_IMPORT_MESSAGE_TEXT_FIELDS = (
@@ -51,13 +50,11 @@ _IMPORT_SESSION_INSERT_SQL = """INSERT INTO sessions (
)"""
# Columns copied verbatim from the payload; typed columns are converted below.
_IMPORT_PASSTHROUGH_COLS = (
"user_id", "model", "model_config", "end_reason", "cwd", "git_branch", "git_repo_root",
"billing_provider", "billing_base_url", "billing_mode", "cost_status", "cost_source",
"pricing_version", "title",
"user_id", "model", "model_config", "end_reason", "cwd", "git_branch", "git_repo_root", "billing_provider",
"billing_base_url", "billing_mode", "cost_status", "cost_source", "pricing_version", "title",
)
_IMPORT_INT_COLS = (
"input_tokens", "output_tokens", "cache_read_tokens", "cache_write_tokens", "reasoning_tokens",
"api_call_count",
"input_tokens", "output_tokens", "cache_read_tokens", "cache_write_tokens", "reasoning_tokens", "api_call_count",
)
_IMPORT_FLOAT_COLS = ("ended_at", "estimated_cost_usd", "actual_cost_usd")
@@ -165,8 +162,7 @@ class SessionPortabilityMixin:
# Same read-your-writes guarantee as list_sessions_rich.
self.flush_token_counts()
query = _rich_select(
self._compact_session_cols() if compact_rows else "s.*",
f"s.id IN ({','.join('?' for _ in ids)})",
self._compact_session_cols() if compact_rows else "s.*", f"s.id IN ({','.join('?' for _ in ids)})",
prompt_select=None if compact_rows else f", {_PROMPT_RESOLVED_SQL}",
)
return {s["id"]: s for s in self._rich_rows(query, ids)}
@@ -222,9 +218,7 @@ class SessionPortabilityMixin:
"""Export all sessions (with messages) as dicts, e.g. for JSONL backup."""
return [self._with_messages(s) for s in self.search_sessions(source=source, limit=100000)]
def adopt_session_lineage_from(
self, donor_db: Any, session_id: str, *, retire_donor: bool = True
) -> Dict[str, Any]:
def adopt_session_lineage_from(self, donor_db: Any, session_id: str, *, retire_donor: bool = True) -> Dict[str, Any]:
"""Adopt *session_id*'s full compression lineage from *donor_db* (stranded-bot-session
heal: a profile bot's rows accumulated in the DEFAULT profile's state.db before the
desktop routed session RPCs by target session). Pure composition
@@ -272,9 +266,7 @@ class SessionPortabilityMixin:
donor_retired = False
if adopted and retire_donor and not donor_ahead:
donor_retired = all(
self._retire_donor_segment(donor_db, seg["id"]) for seg in segments if seg.get("id")
)
donor_retired = all(self._retire_donor_segment(donor_db, seg["id"]) for seg in segments if seg.get("id"))
return {**result, "adopted": adopted, "donor_retired": donor_retired}
def _retire_donor_segment(self, donor_db: Any, seg_id: str) -> bool:

View File

@@ -114,8 +114,7 @@ _TITLE_UNIQUE_INDEX_SQL = (
"CREATE UNIQUE INDEX IF NOT EXISTS idx_sessions_title_unique ON sessions(title) WHERE title IS NOT NULL"
)
_STALE_KEY_UPSERT_SQL = (
"INSERT INTO state_meta (key, value) VALUES (?, '1') "
"ON CONFLICT(key) DO UPDATE SET value = excluded.value"
"INSERT INTO state_meta (key, value) VALUES (?, '1') ON CONFLICT(key) DO UPDATE SET value = excluded.value"
)
_CLEAR_REBUILD_MARKERS_SQL = "DELETE FROM state_meta WHERE key IN ('fts_rebuild_high_water', 'fts_rebuild_progress')"
@@ -165,9 +164,7 @@ class SessionSchemaMixin:
(the legacy column stays a read fallback; next init resumes), whereas propagating
left the version below 25 and re-ran this on every open (gateway crash loop)."""
try:
rows = cursor.execute(
"SELECT id, system_prompt FROM sessions WHERE system_prompt IS NOT NULL"
).fetchall()
rows = cursor.execute("SELECT id, system_prompt FROM sessions WHERE system_prompt IS NOT NULL").fetchall()
except sqlite3.OperationalError:
return
for session_id, prompt in rows:
@@ -180,8 +177,7 @@ class SessionSchemaMixin:
except sqlite3.OperationalError as exc:
logger.warning(
"v25 prompt dedupe paused after contention (%s); "
"unmigrated rows keep the legacy inline prompt and the "
"next schema init resumes the migration.",
"unmigrated rows keep the legacy inline prompt and the next schema init resumes the migration.",
exc,
)
return
@@ -212,8 +208,7 @@ class SessionSchemaMixin:
return False # "name IN ()" is a SQLite syntax error
placeholders = ",".join("?" for _ in names)
row = cursor.execute(
f"SELECT COUNT(*) FROM sqlite_master WHERE type = 'trigger' AND name IN ({placeholders})",
tuple(names),
f"SELECT COUNT(*) FROM sqlite_master WHERE type = 'trigger' AND name IN ({placeholders})", tuple(names),
).fetchone()
return int(row[0]) < len(names)
@@ -237,8 +232,7 @@ class SessionSchemaMixin:
update_names += ("messages_fts_cjk_update",)
placeholders = ", ".join("?" for _ in update_names)
rows = cursor.execute(
f"SELECT name, sql FROM sqlite_master WHERE type = 'trigger' AND name IN ({placeholders})",
update_names,
f"SELECT name, sql FROM sqlite_master WHERE type = 'trigger' AND name IN ({placeholders})", update_names,
).fetchall()
to_drop = [name for name, sql in rows if self._fts_update_trigger_needs_narrowing(sql)]
if not to_drop:
@@ -265,9 +259,7 @@ class SessionSchemaMixin:
"CJK FTS UPDATE trigger missing or still broad after "
"UPDATE OF migration; marked stale and unavailable"
)
logger.info(
"Migrated %d broad FTS UPDATE trigger(s) to AFTER UPDATE OF (no rebuild required)", len(to_drop),
)
logger.info("Migrated %d broad FTS UPDATE trigger(s) to AFTER UPDATE OF (no rebuild required)", len(to_drop))
return len(to_drop)
def _cjk_update_trigger_is_narrowed(self, cursor: sqlite3.Cursor) -> bool:
@@ -373,8 +365,7 @@ class SessionSchemaMixin:
"holder_pids": sorted({pid for pid, _path in foreign_holders if pid > 0}),
}
cursor.execute(
"INSERT INTO state_meta (key, value) VALUES (?, ?) "
"ON CONFLICT(key) DO UPDATE SET value = excluded.value",
"INSERT INTO state_meta (key, value) VALUES (?, ?) ON CONFLICT(key) DO UPDATE SET value = excluded.value",
(FTS_REBUILD_DEFERRAL_KEY, json.dumps(diagnostic, sort_keys=True)),
)
if attempts >= _FTS_HOLDER_ESCALATE_ATTEMPTS and now - first_seen >= _FTS_HOLDER_ESCALATE_SECONDS:
@@ -402,8 +393,7 @@ class SessionSchemaMixin:
return False
logger.warning(
"Deferred stale state.db FTS rebuild while foreign processes "
"hold the database or WAL sidecars (%s); canonical writes and "
"LIKE search remain available (deferral %d).",
"hold the database or WAL sidecars (%s); canonical writes and LIKE search remain available (deferral %d).",
foreign_holders,
attempts,
)
@@ -462,8 +452,7 @@ class SessionSchemaMixin:
return recovered
except Exception: # noqa: BLE001 - background retry must never raise
logger.warning(
"In-process retry of the deferred stale state.db FTS rebuild failed; will retry later.",
exc_info=True,
"In-process retry of the deferred stale state.db FTS rebuild failed; will retry later.", exc_info=True,
)
return False
@@ -527,9 +516,7 @@ class SessionSchemaMixin:
self._fts_stale = False
self._fts_enabled = True
self._trigram_available = include_trigram
logger.warning(
"Rebuilt stale state.db FTS indexes from canonical messages and restored sync triggers."
)
logger.warning("Rebuilt stale state.db FTS indexes from canonical messages and restored sync triggers.")
return True
# ── Declarative column reconciliation ──────────────────────────────────
@@ -616,8 +603,7 @@ class SessionSchemaMixin:
raise
# Anything else permanently strands the store behind SCHEMA_SQL — be loud.
logger.warning(
"reconcile %s.%s failed; store remains behind "
"SCHEMA_SQL: %s", table_name, col_name, exc,
"reconcile %s.%s failed; store remains behind SCHEMA_SQL: %s", table_name, col_name, exc,
)
@staticmethod
@@ -653,8 +639,7 @@ class SessionSchemaMixin:
if pk_cols is None or pk_cols == ["scope", "session_key"]:
return
logger.info(
"gateway_routing has legacy primary key %r; rebuilding with composite (scope, session_key) key",
pk_cols,
"gateway_routing has legacy primary key %r; rebuilding with composite (scope, session_key) key", pk_cols,
)
self._rebuild_table(
cursor, "gateway_routing", "gateway_routing_legacy_pk",
@@ -683,8 +668,7 @@ class SessionSchemaMixin:
if pk_cols is None or "task" in pk_cols:
return
logger.info(
"session_model_usage has legacy primary key %r (missing task); "
"rebuilding with composite 6-column key",
"session_model_usage has legacy primary key %r (missing task); rebuilding with composite 6-column key",
sorted(pk_cols),
)
cursor.execute("PRAGMA foreign_keys=OFF")
@@ -799,8 +783,7 @@ class SessionSchemaMixin:
)
cursor.execute(
"UPDATE sessions SET model_config = json_set("
"COALESCE(model_config, '{}'), '$._delegate_from', '__orphaned__') "
"WHERE parent_session_id IS NULL "
"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 "
@@ -962,8 +945,7 @@ class SessionSchemaMixin:
return
logger.warning(
"Deferred startup FTS rebuild: another process holds the "
"rebuild authority for this state.db; detaching FTS sync "
"until the stale-index recovery path rebuilds it."
"rebuild authority for this state.db; detaching FTS sync until the stale-index recovery path rebuilds it."
)
cursor.execute(_STALE_KEY_UPSERT_SQL, (FTS_STALE_KEY,))
self._drop_all_fts_triggers(cursor)

View File

@@ -103,8 +103,7 @@ def _non_operator_tokens(raw_query: str) -> List[str]:
def _quote_fts_tokens(raw_query: str) -> str:
"""Quote each non-operator token (neutralising FTS5 special characters), keeping AND/OR/NOT."""
return " ".join(
tok if tok.upper() in _FTS_OPERATORS else '"' + tok.replace('"', '""') + '"'
for tok in raw_query.split()
tok if tok.upper() in _FTS_OPERATORS else '"' + tok.replace('"', '""') + '"' for tok in raw_query.split()
)
@@ -166,8 +165,7 @@ class SessionSearchMixin:
"""See module docstring — mixin for SessionDB (Search cluster)."""
_SEARCH_MESSAGE_RESULT_FIELDS = (
"id", "session_id", "role", "snippet", "timestamp", "tool_name", "source", "model",
"session_started", "context",
"id", "session_id", "role", "snippet", "timestamp", "tool_name", "source", "model", "session_started", "context"
)
@classmethod
@@ -207,8 +205,7 @@ class SessionSearchMixin:
def _rebuild_status(self, prefix: str) -> Optional[Dict[str, Any]]:
rows = self._read_all(
"SELECT key, value FROM state_meta WHERE key IN (?, ?)",
(f"{prefix}_high_water", f"{prefix}_progress"),
"SELECT key, value FROM state_meta WHERE key IN (?, ?)", (f"{prefix}_high_water", f"{prefix}_progress"),
)
meta = {r["key"]: r["value"] for r in rows}
high_water = meta.get(f"{prefix}_high_water")
@@ -221,8 +218,7 @@ class SessionSearchMixin:
# per indexed doc, so the anti-join is exact.
_BOUNDARY_SWEEP_SQL = (
"INSERT INTO {table}(rowid, content, tool_name, tool_calls) "
"SELECT m.id, m.content, m.tool_name, m.tool_calls FROM messages m "
"WHERE m.id > ? AND m.id <= ? {extra}"
"SELECT m.id, m.content, m.tool_name, m.tool_calls FROM messages m WHERE m.id > ? AND m.id <= ? {extra}"
"AND NOT EXISTS (SELECT 1 FROM {table}_docsize d WHERE d.id = m.id)"
)
_CHUNK_INSERT_SQL = (
@@ -279,8 +275,7 @@ class SessionSearchMixin:
if not self._fts_enabled or not self._fts_cjk_loaded:
return False
return self._rebuild_step(
"fts_cjk_rebuild",
[self._CHUNK_INSERT_SQL.format(table="messages_fts_cjk", extra=" AND role <> 'tool'")],
"fts_cjk_rebuild", [self._CHUNK_INSERT_SQL.format(table="messages_fts_cjk", extra=" AND role <> 'tool'")],
fail_msg="CJK FTS rebuild chunk failed (will retry): %s", finish=self._fts_cjk_rebuild_finish,
)
@@ -350,8 +345,7 @@ class SessionSearchMixin:
high_water = int(row[0]) if row is not None else 0
# Claim the LAST row of the LIMIT window so a full chunk goes per step.
upper_rows = conn.execute(
f"SELECT {key} FROM {tbl} WHERE {key} > ? "
f"ORDER BY {key} LIMIT {self._FTS_REBUILD_CHUNK_ROWS}",
f"SELECT {key} FROM {tbl} WHERE {key} > ? ORDER BY {key} LIMIT {self._FTS_REBUILD_CHUNK_ROWS}",
(high_water,),
).fetchall()
if not upper_rows:
@@ -362,8 +356,7 @@ class SessionSearchMixin:
self.set_meta(marker_key, str(upper), cursor=conn)
return True
cur = conn.execute(
f"DELETE FROM {tbl} WHERE ({key}) IN "
f"(SELECT {key} FROM {tbl} LIMIT {self._FTS_REBUILD_CHUNK_ROWS})"
f"DELETE FROM {tbl} WHERE ({key}) IN (SELECT {key} FROM {tbl} LIMIT {self._FTS_REBUILD_CHUNK_ROWS})"
)
if cur.rowcount == 0:
return _drop(conn)
@@ -502,16 +495,14 @@ class SessionSearchMixin:
self._drop_fts_triggers(conn)
conn.execute("DROP VIEW IF EXISTS messages_fts_trigram_src")
had = bool(conn.execute(
"SELECT 1 FROM sqlite_master WHERE type = 'table' "
"AND name IN ('messages_fts', 'messages_fts_trigram') "
"SELECT 1 FROM sqlite_master WHERE type = 'table' AND name IN ('messages_fts', 'messages_fts_trigram') "
"AND sql LIKE 'CREATE VIRTUAL TABLE%' LIMIT 1"
).fetchone())
if had:
conn.execute("PRAGMA writable_schema=ON")
conn.execute(
"DELETE FROM sqlite_master WHERE type = 'table' "
"AND name IN ('messages_fts', 'messages_fts_trigram') "
"AND sql LIKE 'CREATE VIRTUAL TABLE%'"
"AND name IN ('messages_fts', 'messages_fts_trigram') AND sql LIKE 'CREATE VIRTUAL TABLE%'"
)
conn.execute("PRAGMA writable_schema=RESET")
shadows = [
@@ -585,9 +576,7 @@ class SessionSearchMixin:
return refusal
self.set_meta("fts_storage_version", str(FTS_STORAGE_VERSION), cursor=conn)
_delete_meta(conn, "fts_optimize_available")
conn.execute(
"UPDATE schema_version SET version = ? WHERE version < ?", (SCHEMA_VERSION, SCHEMA_VERSION),
)
conn.execute("UPDATE schema_version SET version = ? WHERE version < ?", (SCHEMA_VERSION, SCHEMA_VERSION))
return None
def optimize_fts_storage(
@@ -844,11 +833,7 @@ class SessionSearchMixin:
def _trigram_route_ok(self, raw_query: str) -> bool:
"""Per-token CJK length gate for the trigram index: ``广西 OR 桂林 OR 漓江`` has 6
CJK chars total but 2 per token, so trigram returns 0."""
return (
self._count_cjk(raw_query) >= 3
and not self._has_short_cjk_token(raw_query)
and self._trigram_available
)
return(self._count_cjk(raw_query) >= 3 and not self._has_short_cjk_token(raw_query) and self._trigram_available)
def _describe_search_path(self, query: str) -> str:
"""Best-effort name of the routing path a query takes (log-only)."""
@@ -873,9 +858,8 @@ class SessionSearchMixin:
@staticmethod
def _fts_match_sql(
table: str, match_query: str, order_by_sql: str, *, include_inactive: bool,
source_filter: Optional[List[str]], exclude_sources: Optional[List[str]],
role_filter: Optional[List[str]], limit: int, offset: int,
table: str, match_query: str, order_by_sql: str, *, include_inactive: bool, source_filter: Optional[List[str]],
exclude_sources: Optional[List[str]], role_filter: Optional[List[str]], limit: int, offset: int,
) -> Tuple[str, list]:
"""MATCH query + params against one FTS5 index joined to messages/sessions."""
where = [f"{table} MATCH ?"]
@@ -1032,9 +1016,8 @@ class SessionSearchMixin:
elapsed_ms = (time.time() - started) * 1000.0
if elapsed_ms >= env_float("HERMES_SEARCH_SLOW_MS", 1000.0):
logger.info(
"slow session search: path=%s elapsed=%.0fms rows=%s query=%r",
self._describe_search_path(query), elapsed_ms,
len(rows) if rows is not None else "err", query[:200],
"slow session search: path=%s elapsed=%.0fms rows=%s query=%r", self._describe_search_path(query),
elapsed_ms, len(rows) if rows is not None else "err", query[: 200],
)
def _search_messages_impl(
@@ -1231,8 +1214,7 @@ class SessionSearchMixin:
with fts_rebuild_admission(self.db_path) as admitted:
if not admitted:
logger.warning(
"Deferred in-place FTS rebuild: another process holds "
"the rebuild authority for this state.db."
"Deferred in-place FTS rebuild: another process holds the rebuild authority for this state.db."
)
return 0
with self._lock:

View File

@@ -27,8 +27,7 @@ _TOPIC_TABLES = (
(
"telegram_dm_topic_mode",
"profile_name, chat_id, user_id, enabled, activated_at, updated_at, "
"has_topics_enabled, allows_users_to_create_topics, "
"capability_checked_at, intro_message_id, pinned_message_id",
"has_topics_enabled, allows_users_to_create_topics, capability_checked_at, intro_message_id, pinned_message_id",
"""
profile_name TEXT NOT NULL DEFAULT 'default',
chat_id TEXT NOT NULL,
@@ -46,8 +45,7 @@ _TOPIC_TABLES = (
),
(
"telegram_dm_topic_bindings",
"profile_name, chat_id, thread_id, user_id, session_key, "
"session_id, managed_mode, linked_at, updated_at",
"profile_name, chat_id, thread_id, user_id, session_key, session_id, managed_mode, linked_at, updated_at",
"""
profile_name TEXT NOT NULL DEFAULT 'default',
chat_id TEXT NOT NULL,
@@ -158,9 +156,8 @@ class SessionTelegramTopicsMixin:
self._execute_write(_do)
def enable_telegram_topic_mode(
self, *, chat_id: str, user_id: str, profile_name: str = "default",
has_topics_enabled: Optional[bool] = None,
allows_users_to_create_topics: Optional[bool] = None,
self, *, chat_id: str, user_id: str, profile_name: str="default", has_topics_enabled: Optional[bool]=None,
allows_users_to_create_topics: Optional[bool]=None,
) -> None:
"""Enable Telegram DM topic mode for one private chat/user. Owns the explicit topic
migration; SessionDB startup must not create these tables."""
@@ -214,9 +211,7 @@ class SessionTelegramTopicsMixin:
return
self._execute_write(_do)
def is_telegram_topic_mode_enabled(
self, *, chat_id: str, user_id: str, profile_name: str = "default"
) -> bool:
def is_telegram_topic_mode_enabled(self, *, chat_id: str, user_id: str, profile_name: str = "default") -> bool:
"""Return whether Telegram DM topic mode is enabled for this chat/user."""
profile_name = _normalize_telegram_topic_profile_name(profile_name)
row = self._topic_read_one(
@@ -248,8 +243,7 @@ class SessionTelegramTopicsMixin:
"""All bindings for one chat, newest first ([] when the table is absent)."""
profile_name = _normalize_telegram_topic_profile_name(profile_name)
rows = self._topic_read_all(
"SELECT * FROM telegram_dm_topic_bindings WHERE profile_name = ? AND chat_id = ? "
"ORDER BY updated_at DESC",
"SELECT * FROM telegram_dm_topic_bindings WHERE profile_name = ? AND chat_id = ? ORDER BY updated_at DESC",
(profile_name, str(chat_id)),
)
return [dict(row) for row in rows]
@@ -265,9 +259,7 @@ class SessionTelegramTopicsMixin:
)
return dict(row) if row else None
def delete_telegram_topic_binding(
self, *, chat_id: str, thread_id: str, profile_name: str = "default"
) -> int:
def delete_telegram_topic_binding(self, *, chat_id: str, thread_id: str, profile_name: str = "default") -> int:
"""Remove the binding row for one (chat, thread) pair. Called when the Bot API confirms
a topic was deleted externally (``Thread not found`` after the same-thread retry
failed); otherwise ``gateway.run._recover_telegram_topic_thread_id`` keeps

View File

@@ -120,8 +120,7 @@ class SessionTitlesMixin:
# CAS on the values just read (``IS`` is NULL-safe): a concurrent write between
# the SELECT and here loses instead of being overwritten.
cursor = conn.execute(
"UPDATE sessions SET title = ?, title_source = ? "
"WHERE id = ? AND title IS ? AND title_source IS ?",
"UPDATE sessions SET title = ?, title_source = ? WHERE id = ? AND title IS ? AND title_source IS ?",
(title, source if title else None, session_id, current["title"], current["title_source"]),
)
return cursor.rowcount

View File

@@ -274,13 +274,11 @@ class SessionUsageMixin:
pass # never fatal at interpreter shutdown
def update_token_counts(
self, session_id: str, input_tokens: int = 0, output_tokens: int = 0, model: str = None,
cache_read_tokens: int = 0, cache_write_tokens: int = 0, reasoning_tokens: int = 0,
estimated_cost_usd: Optional[float] = None, actual_cost_usd: Optional[float] = None,
cost_status: Optional[str] = None, cost_source: Optional[str] = None,
pricing_version: Optional[str] = None, billing_provider: Optional[str] = None,
billing_base_url: Optional[str] = None, billing_mode: Optional[str] = None,
api_call_count: int = 0, absolute: bool = False,
self, session_id: str, input_tokens: int=0, output_tokens: int=0, model: str=None, cache_read_tokens: int=0,
cache_write_tokens: int=0, reasoning_tokens: int=0, estimated_cost_usd: Optional[float]=None,
actual_cost_usd: Optional[float]=None, cost_status: Optional[str]=None, cost_source: Optional[str]=None,
pricing_version: Optional[str]=None, billing_provider: Optional[str]=None, billing_base_url: Optional[str]=None,
billing_mode: Optional[str]=None, api_call_count: int=0, absolute: bool=False,
) -> None:
"""Update token counters and backfill model if unset. *absolute*=False increments
(per-API-call deltas, CLI path); *absolute*=True sets directly (gateway path,
@@ -335,12 +333,11 @@ class SessionUsageMixin:
self._execute_write(_do)
def _record_model_usage(
self, conn, session_id: str, *, model: Optional[str] = None, billing_provider: Optional[str] = None,
billing_base_url: Optional[str] = None, billing_mode: Optional[str] = None, input_tokens: int = 0,
output_tokens: int = 0, cache_read_tokens: int = 0, cache_write_tokens: int = 0,
reasoning_tokens: int = 0, estimated_cost_usd: Optional[float] = None,
actual_cost_usd: Optional[float] = None, cost_status: Optional[str] = None,
cost_source: Optional[str] = None, api_call_count: int = 0, task: str = "",
self, conn, session_id: str, *, model: Optional[str]=None, billing_provider: Optional[str]=None,
billing_base_url: Optional[str]=None, billing_mode: Optional[str]=None, input_tokens: int=0,
output_tokens: int=0, cache_read_tokens: int=0, cache_write_tokens: int=0, reasoning_tokens: int=0,
estimated_cost_usd: Optional[float]=None, actual_cost_usd: Optional[float]=None,
cost_status: Optional[str]=None, cost_source: Optional[str]=None, api_call_count: int=0, task: str="",
) -> None:
"""Accumulate a per-API-call usage delta into session_model_usage, inside the caller's
write txn after the ``sessions`` UPDATE. A missing model/provider falls back to
@@ -348,8 +345,7 @@ class SessionUsageMixin:
main-loop route (vision on gemini while the main loop runs anthropic): missing
info stays 'unknown'/empty."""
row = conn.execute(
"SELECT model, billing_provider, billing_base_url, billing_mode "
"FROM sessions WHERE id = ?", (session_id,),
"SELECT model, billing_provider, billing_base_url, billing_mode FROM sessions WHERE id = ?", (session_id,),
).fetchone()
sess = dict(row) if (row is not None and not task) else {}
counts = [v or 0 for v in (input_tokens, output_tokens, cache_read_tokens, cache_write_tokens, reasoning_tokens)]
@@ -366,11 +362,10 @@ class SessionUsageMixin:
cost_status, cost_source, now, now))
def record_auxiliary_usage(
self, session_id: str, task: str, *, model: Optional[str] = None,
billing_provider: Optional[str] = None, billing_base_url: Optional[str] = None,
input_tokens: int = 0, output_tokens: int = 0, cache_read_tokens: int = 0,
cache_write_tokens: int = 0, reasoning_tokens: int = 0,
estimated_cost_usd: Optional[float] = None, api_call_count: int = 1,
self, session_id: str, task: str, *, model: Optional[str]=None, billing_provider: Optional[str]=None,
billing_base_url: Optional[str]=None, input_tokens: int=0, output_tokens: int=0, cache_read_tokens: int=0,
cache_write_tokens: int=0, reasoning_tokens: int=0, estimated_cost_usd: Optional[float]=None,
api_call_count: int=1,
) -> None:
"""Record an auxiliary LLM call's usage (vision, compression, title generation, ...)
as a per-(model, provider, task) delta in ``session_model_usage`` WITHOUT touching