From cb7f6bbb2e5696bae2c595ca8a3a52cefb6a4a7e Mon Sep 17 00:00:00 2001 From: Thomas Connally <51974392+tcconnally@users.noreply.github.com> Date: Tue, 23 Jun 2026 21:23:43 -0500 Subject: [PATCH] feat(agent): track per-model token usage for mid-session model switches MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The `sessions` table records only the initial (model, billing_provider) for a session, so when a user switches models mid-session (via `/model` or programmatically) every token — including the switched model's — is attributed to the first model. Insights/billing reports then hide the cost of the new model entirely (e.g. a session that started on deepseek and switched to opus shows $0 for opus). Add a `session_model_usage` table keyed (session_id, model, billing_provider) that accumulates each per-API-call delta under the model active at the time of the call. `update_token_counts()` is the single chokepoint every per-call delta flows through (CLI, gateway, cron, delegated, codex), so recording there captures accurate attribution on every platform. Only the incremental path records — the gateway's `absolute=True` summary overwrite is skipped to avoid double-counting cumulative totals that can't be split per model. When a call omits the model, it falls back to the session's recorded model, matching the existing COALESCE-from-session summary behaviour. Insights `_compute_model_breakdown` now aggregates tokens and cost from `session_model_usage`, so a switched session splits correctly across models, with a defensive fallback to the per-session aggregate for any session lacking usage rows. A v17 migration backfills one usage row per existing token-bearing session from its aggregate totals (idempotent via INSERT OR IGNORE), validated lossless against a 1.3 GB production DB. Tests: per-model recording, mid-session split, model fallback, absolute no-double-count, v17 backfill, and an insights-level switch breakdown. Fixes #51607. Co-Authored-By: Claude Opus 4.8 --- agent/insights.py | 140 +++++++++++++++++++++++++---- hermes_state.py | 166 ++++++++++++++++++++++++++++++++++- tests/agent/test_insights.py | 29 ++++++ tests/test_hermes_state.py | 123 ++++++++++++++++++++++++++ 4 files changed, 438 insertions(+), 20 deletions(-) diff --git a/agent/insights.py b/agent/insights.py index 9977010549c..b996cd405d9 100644 --- a/agent/insights.py +++ b/agent/insights.py @@ -17,6 +17,7 @@ Usage: """ import json +import sqlite3 import time from collections import Counter, defaultdict from datetime import datetime @@ -142,7 +143,7 @@ class InsightsEngine: # Compute insights overview = self._compute_overview(sessions, message_stats) - models = self._compute_model_breakdown(sessions) + models = self._compute_model_breakdown(sessions, cutoff, source) platforms = self._compute_platform_breakdown(sessions) tools = self._compute_tool_breakdown(tool_usage) skills = self._compute_skill_breakdown(skill_usage) @@ -473,39 +474,140 @@ class InsightsEngine: "included_cost_sessions": included_cost_sessions, } - def _compute_model_breakdown(self, sessions: List[Dict]) -> List[Dict]: - """Break down usage by model.""" + _GET_MODEL_USAGE_WITH_SOURCE = ( + "SELECT u.session_id, u.model, u.billing_provider, u.billing_base_url," + " u.api_call_count, u.input_tokens, u.output_tokens," + " u.cache_read_tokens, u.cache_write_tokens, u.reasoning_tokens," + " u.estimated_cost_usd" + " FROM session_model_usage u" + " JOIN sessions s ON s.id = u.session_id" + " WHERE s.started_at >= ? AND s.source = ?" + ) + _GET_MODEL_USAGE_ALL = ( + "SELECT u.session_id, u.model, u.billing_provider, u.billing_base_url," + " u.api_call_count, u.input_tokens, u.output_tokens," + " u.cache_read_tokens, u.cache_write_tokens, u.reasoning_tokens," + " u.estimated_cost_usd" + " FROM session_model_usage u" + " JOIN sessions s ON s.id = u.session_id" + " WHERE s.started_at >= ?" + ) + + def _get_model_usage(self, cutoff: float, source: str = None) -> List[Dict]: + """Fetch per-model usage rows within the window (issue #51607). + + Returns an empty list when the table is missing (e.g. a DB opened by + older code that never created it) so the caller can fall back to the + per-session aggregate. + """ + try: + if source: + cursor = self._conn.execute( + self._GET_MODEL_USAGE_WITH_SOURCE, (cutoff, source) + ) + else: + cursor = self._conn.execute(self._GET_MODEL_USAGE_ALL, (cutoff,)) + return [dict(row) for row in cursor.fetchall()] + except sqlite3.OperationalError: + return [] + + def _compute_model_breakdown( + self, sessions: List[Dict], cutoff: float, source: str = None + ) -> List[Dict]: + """Break down token usage and cost by model. + + Tokens and cost are attributed per model from session_model_usage, so a + session that switched models mid-flight (via ``/model``) splits across + every model it used instead of dumping everything on the initial model + (issue #51607). Sessions without per-model rows — e.g. data written + before this table existed and not yet backfilled — fall back to their + single recorded (model, billing_provider) aggregate so nothing is lost. + + Tool calls aren't tied to a specific API invocation, so they stay + attributed to the session's recorded model. + """ model_data = defaultdict(lambda: { - "sessions": 0, "input_tokens": 0, "output_tokens": 0, + "sessions": set(), "input_tokens": 0, "output_tokens": 0, "cache_read_tokens": 0, "cache_write_tokens": 0, - "total_tokens": 0, "tool_calls": 0, "cost": 0.0, + "reasoning_tokens": 0, "total_tokens": 0, "api_calls": 0, + "tool_calls": 0, "cost": 0.0, }) - for s in sessions: - model = s.get("model") or "unknown" + def _accumulate(model, provider, base_url, session_id, inp, out, + cache_read, cache_write, reasoning): + model = model or "unknown" # Normalize: strip provider prefix for display display_model = model.split("/")[-1] if "/" in model else model d = model_data[display_model] - d["sessions"] += 1 - inp = s.get("input_tokens") or 0 - out = s.get("output_tokens") or 0 - cache_read = s.get("cache_read_tokens") or 0 - cache_write = s.get("cache_write_tokens") or 0 + d["sessions"].add(session_id) d["input_tokens"] += inp d["output_tokens"] += out d["cache_read_tokens"] += cache_read d["cache_write_tokens"] += cache_write + d["reasoning_tokens"] += reasoning d["total_tokens"] += inp + out + cache_read + cache_write - d["tool_calls"] += s.get("tool_call_count") or 0 - estimate, status = _estimate_cost(s) + estimate, status = _estimate_cost( + model, inp, out, + cache_read_tokens=cache_read, cache_write_tokens=cache_write, + provider=provider or None, base_url=base_url, + ) d["cost"] += estimate - d["has_pricing"] = has_known_pricing(model, s.get("billing_provider"), s.get("billing_base_url")) d["cost_status"] = status + if has_known_pricing(model, provider or None, base_url): + d["has_pricing"] = True + else: + d.setdefault("has_pricing", False) + return display_model - result = [ - {"model": model, **data} - for model, data in model_data.items() - ] + usage_rows = self._get_model_usage(cutoff, source) + covered: set = set() + for r in usage_rows: + covered.add(r["session_id"]) + d = _accumulate( + r["model"], r["billing_provider"], r.get("billing_base_url"), + r["session_id"], r["input_tokens"] or 0, r["output_tokens"] or 0, + r["cache_read_tokens"] or 0, r["cache_write_tokens"] or 0, + r["reasoning_tokens"] or 0, + ) + model_data[d]["api_calls"] += r["api_call_count"] or 0 + + # Fallback for sessions with token totals but no per-model rows + # (legacy data not covered by the v17 backfill). Attribute their + # aggregate to the single recorded model so totals never regress. + for s in sessions: + if s["id"] in covered: + continue + inp = s.get("input_tokens") or 0 + out = s.get("output_tokens") or 0 + cache_read = s.get("cache_read_tokens") or 0 + cache_write = s.get("cache_write_tokens") or 0 + if not (inp or out or cache_read or cache_write): + continue + _accumulate( + s.get("model"), s.get("billing_provider"), + s.get("billing_base_url"), s["id"], + inp, out, cache_read, cache_write, 0, + ) + + # Tool calls are attributed by the session's recorded model. + for s in sessions: + tool_calls = s.get("tool_call_count") or 0 + if not tool_calls: + continue + model = s.get("model") or "unknown" + display_model = model.split("/")[-1] if "/" in model else model + model_data[display_model]["tool_calls"] += tool_calls + + result = [] + for model, data in model_data.items(): + entry = {"model": model, **data} + entry["sessions"] = len(data["sessions"]) + # Models that surfaced only via tool-call attribution (no token + # rows) won't have these set by _accumulate — default them so the + # output shape is uniform for downstream/JSON consumers. + entry.setdefault("has_pricing", False) + entry.setdefault("cost_status", "unknown") + result.append(entry) # Sort by tokens first, fall back to session count when tokens are 0 result.sort(key=lambda x: (x["total_tokens"], x["sessions"]), reverse=True) return result diff --git a/hermes_state.py b/hermes_state.py index f1ef14cb5fc..13802967879 100644 --- a/hermes_state.py +++ b/hermes_state.py @@ -122,7 +122,7 @@ T = TypeVar("T") DEFAULT_DB_PATH = get_hermes_home() / "state.db" -SCHEMA_VERSION = 19 +SCHEMA_VERSION = 20 # Cap on user-controlled FTS5 query input before regex/sanitizer processing. # Search queries do not need to be arbitrarily large, and bounding them keeps @@ -768,6 +768,23 @@ CREATE TABLE IF NOT EXISTS messages ( compacted INTEGER NOT NULL DEFAULT 0 ); +CREATE TABLE IF NOT EXISTS session_model_usage ( + session_id TEXT NOT NULL REFERENCES sessions(id), + model TEXT NOT NULL, + billing_provider TEXT NOT NULL DEFAULT '', + billing_base_url TEXT, + 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, + first_seen REAL, + last_seen REAL, + PRIMARY KEY (session_id, model, billing_provider) +); + CREATE TABLE IF NOT EXISTS state_meta ( key TEXT PRIMARY KEY, value TEXT @@ -794,6 +811,8 @@ 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); """ # Indexes that reference columns added in later schema versions must be @@ -1554,6 +1573,45 @@ class SessionDB: # 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, api_call_count, input_tokens, + output_tokens, cache_read_tokens, + cache_write_tokens, reasoning_tokens, + estimated_cost_usd, first_seen, last_seen + ) + SELECT id, COALESCE(model, 'unknown'), + COALESCE(billing_provider, ''), + billing_base_url, + 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), + 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 < SCHEMA_VERSION and fts_migrations_complete: cursor.execute( "UPDATE schema_version SET version = ?", @@ -2500,10 +2558,116 @@ class SessionDB: api_call_count, session_id, ) + # Per-model usage attribution. ``update_token_counts`` is the single + # chokepoint every per-API-call delta flows through (CLI, gateway, cron, + # delegated runs — see conversation_loop / codex_runtime), and each call + # carries the model/provider *active at the time of that call*. The + # ``sessions`` row only keeps one (model, billing_provider) pair, so a + # mid-session ``/model`` switch otherwise attributes every token to the + # initial model (issue #51607). Recording the per-call delta into + # session_model_usage keyed by the live model preserves an accurate + # per-model breakdown regardless of how many times the user switches. + # + # Only the incremental path records here: the gateway also issues an + # ``absolute=True`` call that overwrites the sessions summary totals + # with the cached agent's cumulative figures — folding those in would + # double-count, and cumulative totals can't be split back per model. + record_model_usage = (not absolute) and ( + input_tokens or output_tokens or cache_read_tokens + or cache_write_tokens or reasoning_tokens or api_call_count + or estimated_cost_usd + ) + def _do(conn): conn.execute(sql, params) + if record_model_usage: + self._record_model_usage( + conn, + session_id, + model=model, + billing_provider=billing_provider, + billing_base_url=billing_base_url, + input_tokens=input_tokens, + output_tokens=output_tokens, + cache_read_tokens=cache_read_tokens, + cache_write_tokens=cache_write_tokens, + reasoning_tokens=reasoning_tokens, + estimated_cost_usd=estimated_cost_usd, + api_call_count=api_call_count, + ) self._execute_write(_do) + def _record_model_usage( + self, + conn, + session_id: str, + *, + model: Optional[str], + billing_provider: Optional[str], + billing_base_url: Optional[str], + input_tokens: int, + output_tokens: int, + cache_read_tokens: int, + cache_write_tokens: int, + reasoning_tokens: int, + estimated_cost_usd: Optional[float], + api_call_count: int, + ) -> None: + """Accumulate a per-API-call usage delta into session_model_usage. + + Runs inside the caller's write transaction (after the ``sessions`` + UPDATE) so the per-model rows stay consistent with the summary row. + When the caller omits the model/provider (some paths only pass token + deltas), fall back to the values already recorded on the session row — + the same COALESCE-from-session behaviour the summary update uses. + """ + row = conn.execute( + "SELECT model, billing_provider, billing_base_url " + "FROM sessions WHERE id = ?", + (session_id,), + ).fetchone() + sess_model = row["model"] if row is not None else None + sess_provider = row["billing_provider"] if row is not None else None + sess_base_url = row["billing_base_url"] if row is not None else None + + eff_model = model or sess_model or "unknown" + eff_provider = billing_provider or sess_provider or "" + eff_base_url = billing_base_url or sess_base_url + now = time.time() + conn.execute( + """INSERT INTO session_model_usage ( + session_id, model, billing_provider, billing_base_url, + api_call_count, input_tokens, output_tokens, + cache_read_tokens, cache_write_tokens, reasoning_tokens, + estimated_cost_usd, first_seen, last_seen + ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) + ON CONFLICT(session_id, model, billing_provider) DO UPDATE SET + api_call_count = api_call_count + excluded.api_call_count, + input_tokens = input_tokens + excluded.input_tokens, + output_tokens = output_tokens + excluded.output_tokens, + cache_read_tokens = cache_read_tokens + excluded.cache_read_tokens, + cache_write_tokens = cache_write_tokens + excluded.cache_write_tokens, + reasoning_tokens = reasoning_tokens + excluded.reasoning_tokens, + estimated_cost_usd = estimated_cost_usd + excluded.estimated_cost_usd, + billing_base_url = COALESCE(excluded.billing_base_url, billing_base_url), + last_seen = excluded.last_seen""", + ( + session_id, + eff_model, + eff_provider, + eff_base_url, + api_call_count or 0, + input_tokens or 0, + output_tokens or 0, + cache_read_tokens or 0, + cache_write_tokens or 0, + reasoning_tokens or 0, + float(estimated_cost_usd or 0.0), + now, + now, + ), + ) + def ensure_session( self, session_id: str, diff --git a/tests/agent/test_insights.py b/tests/agent/test_insights.py index e0aad522227..22cc0814f36 100644 --- a/tests/agent/test_insights.py +++ b/tests/agent/test_insights.py @@ -320,6 +320,35 @@ class TestInsightsPopulated: claude = next(m for m in models if "claude-sonnet" in m["model"]) assert claude["sessions"] == 2 + def test_model_breakdown_splits_mid_session_switch(self, db): + """A session that switches models mid-flight is split across both + models in the breakdown, not dumped on the initial model (#51607). + """ + now = time.time() + db.create_session(session_id="sw", source="cli", + model="deepseek/deepseek-v4-pro") + # 40k tokens on deepseek, then switch and 50k on opus. + db.update_token_counts("sw", input_tokens=40000, output_tokens=8000, + model="deepseek/deepseek-v4-pro", + billing_provider="deepseek", api_call_count=2) + db.update_session_model("sw", "anthropic/claude-opus-4.8") + db.update_token_counts("sw", input_tokens=50000, output_tokens=4000, + model="anthropic/claude-opus-4.8", + billing_provider="openrouter", api_call_count=3) + db._conn.commit() + + report = InsightsEngine(db).generate(days=30) + models = {m["model"]: m for m in report["models"]} + assert "deepseek-v4-pro" in models + assert "claude-opus-4.8" in models + # Tokens attributed to the model that actually incurred them. + assert models["deepseek-v4-pro"]["input_tokens"] == 40000 + assert models["claude-opus-4.8"]["input_tokens"] == 50000 + assert models["claude-opus-4.8"]["api_calls"] == 3 + # The summary row's single model would have hidden one of these. + assert models["deepseek-v4-pro"]["total_tokens"] == 48000 + assert models["claude-opus-4.8"]["total_tokens"] == 54000 + def test_platform_breakdown(self, populated_db): engine = InsightsEngine(populated_db) report = engine.generate(days=30) diff --git a/tests/test_hermes_state.py b/tests/test_hermes_state.py index b2697c2ab6e..18c7e695029 100644 --- a/tests/test_hermes_state.py +++ b/tests/test_hermes_state.py @@ -351,6 +351,129 @@ class TestSessionLifecycle: assert sess["billing_provider"] == "openai" assert sess["billing_mode"] == "local" # preserved (COALESCE on None) + def test_per_model_usage_recorded_for_single_model(self, db): + """Each per-call delta lands in session_model_usage (#51607).""" + db.create_session(session_id="s1", source="cli") + db.update_token_counts("s1", input_tokens=200, output_tokens=100, + model="anthropic/claude-opus-4.8", + billing_provider="anthropic", api_call_count=1) + db.update_token_counts("s1", input_tokens=100, output_tokens=50, + model="anthropic/claude-opus-4.8", + billing_provider="anthropic", api_call_count=1) + + rows = db._conn.execute( + "SELECT model, billing_provider, api_call_count, input_tokens, " + "output_tokens FROM session_model_usage WHERE session_id = 's1'" + ).fetchall() + assert len(rows) == 1 + row = rows[0] + assert row["model"] == "anthropic/claude-opus-4.8" + assert row["billing_provider"] == "anthropic" + assert row["api_call_count"] == 2 + assert row["input_tokens"] == 300 + assert row["output_tokens"] == 150 + + def test_mid_session_switch_splits_per_model_usage(self, db): + """The headline #51607 case: tokens after a /model switch are + attributed to the new model, not the session's initial model. + + The ``sessions`` summary row still holds combined totals + the latest + model, but session_model_usage keeps an accurate per-model split. + """ + db.create_session(session_id="s1", source="cli", + model="deepseek/deepseek-v4-pro") + # Pre-switch calls on deepseek. + db.update_token_counts("s1", input_tokens=40_000, output_tokens=8_000, + model="deepseek/deepseek-v4-pro", + billing_provider="deepseek", api_call_count=2) + # User runs /model — the gateway persists the new model … + db.update_session_model("s1", "anthropic/claude-opus-4.8") + # … and subsequent per-call deltas carry the new model/provider. + db.update_token_counts("s1", input_tokens=50_000, output_tokens=4_000, + model="anthropic/claude-opus-4.8", + billing_provider="openrouter", api_call_count=3) + + rows = { + r["model"]: r + for r in db._conn.execute( + "SELECT model, billing_provider, input_tokens, output_tokens, " + "api_call_count FROM session_model_usage WHERE session_id = 's1'" + ).fetchall() + } + assert set(rows) == {"deepseek/deepseek-v4-pro", + "anthropic/claude-opus-4.8"} + assert rows["deepseek/deepseek-v4-pro"]["input_tokens"] == 40_000 + assert rows["deepseek/deepseek-v4-pro"]["api_call_count"] == 2 + assert rows["anthropic/claude-opus-4.8"]["input_tokens"] == 50_000 + assert rows["anthropic/claude-opus-4.8"]["billing_provider"] == "openrouter" + assert rows["anthropic/claude-opus-4.8"]["api_call_count"] == 3 + + # Summary row: latest model + combined totals (unchanged behaviour). + session = db.get_session("s1") + assert session["model"] == "anthropic/claude-opus-4.8" + assert session["input_tokens"] == 90_000 + assert session["output_tokens"] == 12_000 + + def test_per_model_usage_falls_back_to_session_model(self, db): + """When a call omits the model, attribute it to the session's + recorded model — matches the COALESCE-from-session summary behaviour + and keeps existing callers (which pass no model) working. + """ + db.create_session(session_id="s1", source="cli", + model="gpt-4o", ) + db.update_token_counts("s1", input_tokens=10, output_tokens=5) + + rows = db._conn.execute( + "SELECT model FROM session_model_usage WHERE session_id = 's1'" + ).fetchall() + assert len(rows) == 1 + assert rows[0]["model"] == "gpt-4o" + + def test_absolute_update_does_not_record_per_model(self, db): + """absolute=True overwrites the cumulative summary row (gateway path) + and must NOT add per-model rows — those are accumulated from the + per-call incremental path, so recording here would double-count. + """ + db.create_session(session_id="s1", source="cli", model="gpt-4o") + db.update_token_counts("s1", input_tokens=500, output_tokens=200, + model="gpt-4o", absolute=True) + + rows = db._conn.execute( + "SELECT COUNT(*) AS n FROM session_model_usage WHERE session_id = 's1'" + ).fetchone() + assert rows["n"] == 0 + + def test_v17_backfill_seeds_existing_session_usage(self, tmp_path): + """A DB upgraded from <17 seeds one usage row per historical session + from its aggregate totals, so insights read uniformly from the table. + """ + db_path = tmp_path / "legacy.db" + db = SessionDB(db_path=db_path) + db.create_session(session_id="legacy1", source="cli", model="gpt-4o") + db.update_token_counts("legacy1", input_tokens=1234, output_tokens=567, + model="gpt-4o", billing_provider="openai") + # Simulate a pre-v17 database: drop the per-model rows and roll the + # recorded schema version back so the backfill migration re-runs. + db._conn.execute("DELETE FROM session_model_usage") + db._conn.execute("UPDATE schema_version SET version = 16") + db._conn.commit() + db.close() + + # Reopen — _init_schema should backfill from the sessions aggregate. + db2 = SessionDB(db_path=db_path) + try: + rows = db2._conn.execute( + "SELECT model, billing_provider, input_tokens, output_tokens " + "FROM session_model_usage WHERE session_id = 'legacy1'" + ).fetchall() + assert len(rows) == 1 + assert rows[0]["model"] == "gpt-4o" + assert rows[0]["billing_provider"] == "openai" + assert rows[0]["input_tokens"] == 1234 + assert rows[0]["output_tokens"] == 567 + finally: + db2.close() + def test_parent_session(self, db): db.create_session(session_id="parent", source="cli") db.create_session(session_id="child", source="cli", parent_session_id="parent")