diff --git a/hermes_state_portability.py b/hermes_state_portability.py index 265f237c10..a8396fc909 100644 --- a/hermes_state_portability.py +++ b/hermes_state_portability.py @@ -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: diff --git a/hermes_state_schema.py b/hermes_state_schema.py index 06ae79c64e..7945013d43 100644 --- a/hermes_state_schema.py +++ b/hermes_state_schema.py @@ -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) diff --git a/hermes_state_search.py b/hermes_state_search.py index d0dbc2eb53..832602d58c 100644 --- a/hermes_state_search.py +++ b/hermes_state_search.py @@ -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: diff --git a/hermes_state_telegram.py b/hermes_state_telegram.py index f413943bc7..a7172a3370 100644 --- a/hermes_state_telegram.py +++ b/hermes_state_telegram.py @@ -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 diff --git a/hermes_state_titles.py b/hermes_state_titles.py index 0b3b912098..0297ddf11b 100644 --- a/hermes_state_titles.py +++ b/hermes_state_titles.py @@ -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 diff --git a/hermes_state_usage.py b/hermes_state_usage.py index 4e61d85edb..61396dc6b8 100644 --- a/hermes_state_usage.py +++ b/hermes_state_usage.py @@ -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