diff --git a/nerve/agent/engine.py b/nerve/agent/engine.py index 9dddda1..5583ec8 100644 --- a/nerve/agent/engine.py +++ b/nerve/agent/engine.py @@ -1701,6 +1701,13 @@ async def _get_or_create_client( ) await self.db.update_session_fields(session_id, {"model": resolved_model}) + # A brand-new CLI process starts its cumulative cost counter at + # zero — zero the persisted baseline so the first turn's delta + # is exact. The reset inference in compute_turn_cost remains + # as a backstop for any recycle path that bypasses this (e.g. + # a `nerve restart` racing an in-flight turn). + await self._reset_cost_baseline(session_id) + logger.info( "Created persistent client for session %s%s", session_id, @@ -1709,6 +1716,31 @@ async def _get_or_create_client( ) return client + async def _reset_cost_baseline(self, session_id: str) -> None: + """Zero the persisted SDK cumulative-cost baseline for a session. + + Called right after a new CLI client is created: the fresh process + reports ``total_cost_usd`` cumulatively from zero, so the stored + high-water mark from the previous client must not be diffed + against it (see compute_turn_cost in nerve.db.usage). + + Re-reads the session row so concurrent metadata writers are not + clobbered; accounting must never break a turn, so failures are + logged and swallowed. + """ + try: + session = await self.db.get_session(session_id) + if not session: + return + meta = json.loads(session.get("metadata") or "{}") + if meta.get("_sdk_cumulative_cost"): + meta["_sdk_cumulative_cost"] = 0 + await self.db.update_session_metadata(session_id, meta) + except Exception as e: + logger.warning( + "Failed to reset cost baseline for %s: %s", session_id, e, + ) + async def _discard_client( self, session_id: str, clear_resume: bool = False, background_memorize: bool = False, @@ -2447,22 +2479,38 @@ async def _finalize_turn( web_fetch = server_tool.get("web_fetch_requests", 0) # Calculate per-turn cost. - # NOTE: The SDK's total_cost_usd is *cumulative* across the - # entire SDK session, NOT per-invocation. We track the last - # known cumulative value in session metadata so we can compute - # the delta for this turn. - from nerve.db.usage import estimate_turn_cost, extract_cache_ttl_split + # NOTE: The SDK's total_cost_usd is *cumulative* per CLI client + # process, NOT per-invocation. We track the last known + # cumulative value in session metadata and compute the delta + # for this turn. The counter resets whenever the client is + # recycled (idle sweep, oneshot cron teardown, restart, model + # switch) — compute_turn_cost detects the reset and attributes + # the new cumulative to this turn instead of recording $0. + from nerve.db.usage import compute_turn_cost, extract_cache_ttl_split sdk_cost = (st.result_meta or {}).get("total_cost_usd") current_session_cost = ( session_record.get("total_cost_usd", 0) if session_record else 0 ) or 0 + prev_cumulative = meta.get("_sdk_cumulative_cost", 0) or 0 + turn_cost, cost_source = compute_turn_cost( + sdk_cost, prev_cumulative, st.last_usage, model=st.last_model, + ) + if cost_source == "sdk_reset": + logger.info( + "SDK cost counter reset for %s (%.4f < %.4f) — client " + "recycle; attributing new cumulative to this turn", + session_id, sdk_cost, prev_cumulative, + ) + elif cost_source == "estimate_backstop": + logger.warning( + "SDK reported no turn cost (cumulative %.4f, prev %.4f) " + "despite token traffic for %s — using token-based " + "estimate $%.4f", + sdk_cost, prev_cumulative, session_id, turn_cost, + ) if sdk_cost is not None: - prev_cumulative = meta.get("_sdk_cumulative_cost", 0) or 0 - turn_cost = max(sdk_cost - prev_cumulative, 0) meta["_sdk_cumulative_cost"] = sdk_cost - else: - turn_cost = estimate_turn_cost(st.last_usage, model=st.last_model) # Save metadata (includes _sdk_cumulative_cost update) await self.db.update_session_metadata(session_id, meta) diff --git a/nerve/db/migrations/v036_backfill_zero_cost_turns.py b/nerve/db/migrations/v036_backfill_zero_cost_turns.py new file mode 100644 index 0000000..9d4dfa1 --- /dev/null +++ b/nerve/db/migrations/v036_backfill_zero_cost_turns.py @@ -0,0 +1,178 @@ +"""V36: Backfill per-turn costs swallowed by SDK client recycling. + +The Claude Agent SDK's ``ResultMessage.total_cost_usd`` is *cumulative* +per CLI client process. The engine derives per-turn cost by diffing +consecutive cumulative values, persisting the last seen value in session +metadata. The metadata survives client recycling but the counter does +not: every recycle (idle sweep, oneshot cron teardown, restart, model +switch) starts a fresh CLI process whose cumulative begins near zero — +*below* the persisted high-water mark. The old ``max(delta, 0)`` clamp +turned that negative diff into $0, silently swallowing the first turn on +every new client. Persistent crons tear the client down after every +run, so every cron turn recorded $0. + +(V24 fixed the inverse bug — treating the cumulative as per-turn, a +massive over-count. The delta logic introduced then caused this +under-count. The engine now uses reset-aware deltas via +``nerve.db.usage.compute_turn_cost``; this migration repairs the rows +written while the clamp was live.) + +Fix: +1. Recompute ``session_usage.cost_usd`` from token counts (same math as + ``nerve.db.usage.estimate_turn_cost``, honoring the 5m/1h cache-write + split when present) for rows that recorded zero cost despite nonzero + token traffic. +2. Add each session's recovered amount to ``sessions.total_cost_usd``. + NOTE: deliberately additive, NOT a re-sum of ``session_usage`` — + telemetry pruning (``prune_telemetry``) deletes old usage rows while + keeping the session row, so a blanket re-sum would erase the pruned + turns' legitimate accumulated cost. The zero-cost target rows + contributed exactly $0 to the stored totals, so adding their + recomputed cost is exact. + +Idempotent: rerunning finds no remaining zero-cost rows with token +traffic (rows whose recompute legitimately rounds to zero recompute to +the same zero and contribute a zero delta). +""" + +from __future__ import annotations + +import logging + +import aiosqlite + +logger = logging.getLogger(__name__) + +# Model pricing (per 1M tokens, USD): +# (input, output, cache_read, cache_write_5m, cache_write_1h, web_search_per_req) +# Snapshot of nerve/db/usage.py MODEL_PRICING at migration time. +_PRICING: dict[str, tuple[float, float, float, float, float, float]] = { + "fable-5": (10, 50, 1.00, 12.50, 20.00, 0.01), + "opus-4-8": (5, 25, 0.50, 6.25, 10.00, 0.01), + "opus-4-7": (5, 25, 0.50, 6.25, 10.00, 0.01), + "opus-4-6": (5, 25, 0.50, 6.25, 10.00, 0.01), + "opus-4-5": (5, 25, 0.50, 6.25, 10.00, 0.01), + "opus-4-1": (15, 75, 1.50, 18.75, 30.00, 0.01), + "opus-4": (15, 75, 1.50, 18.75, 30.00, 0.01), + "sonnet-4": (3, 15, 0.30, 3.75, 6.00, 0.01), + "haiku-4-5": (1, 5, 0.10, 1.25, 2.00, 0.01), + "haiku-3-5": (0.8, 4, 0.08, 1.00, 1.60, 0.01), +} +_DEFAULT = (5, 25, 0.50, 6.25, 10.00, 0.01) # Opus 4.x standard fallback + +# A row is a backfill target when it recorded no cost despite real +# token traffic. +_TARGET_WHERE = ( + "COALESCE(cost_usd, 0) = 0 " + "AND COALESCE(input_tokens, 0) + COALESCE(output_tokens, 0) " + " + COALESCE(cache_read_input_tokens, 0) " + " + COALESCE(cache_creation_input_tokens, 0) > 0" +) + + +def _pricing_for(model: str | None) -> tuple[float, float, float, float, float, float]: + if not model: + return _DEFAULT + m = model.lower() + for key, p in _PRICING.items(): + if key in m: + return p + return _DEFAULT + + +async def up(db: aiosqlite.Connection) -> None: + # -- Step 0: snapshot the target rows ----------------------------------- + # After the UPDATE the repaired rows are indistinguishable from rows + # that always had a cost, so capture their rowids (and sessions) first. + await db.execute("DROP TABLE IF EXISTS _v036_targets") + await db.execute( + f""" + CREATE TEMP TABLE _v036_targets AS + SELECT rowid AS rid, session_id + FROM session_usage + WHERE {_TARGET_WHERE} + """ + ) + async with db.execute("SELECT COUNT(*) FROM _v036_targets") as cur: + target_count = (await cur.fetchone())[0] + + if not target_count: + await db.execute("DROP TABLE IF EXISTS _v036_targets") + await db.commit() + logger.info("V36 migration: no zero-cost turns with token traffic — nothing to backfill") + return + + # -- Step 1: recompute cost_usd from token counts, per model ------------ + # SQLite can't substring-match models in a CASE, so resolve pricing + # per distinct model and bulk-update (the v024 pattern). Cache + # writes honor the 5m/1h TTL split when recorded; legacy rows + # without the split bill the aggregate at the 5m rate — identical + # to estimate_turn_cost. + async with db.execute( + "SELECT DISTINCT model FROM session_usage " + "WHERE rowid IN (SELECT rid FROM _v036_targets)" + ) as cur: + models = [row[0] for row in await cur.fetchall()] + + for model in models: + p_in, p_out, p_cr, p_c5m, p_c1h, p_ws = _pricing_for(model) + if model is None: + model_where = "model IS NULL" + params: tuple = () + else: + model_where = "model = ?" + params = (model,) + + await db.execute( + f""" + UPDATE session_usage + SET cost_usd = ROUND( + COALESCE(input_tokens, 0) * {p_in} / 1000000.0 + + COALESCE(output_tokens, 0) * {p_out} / 1000000.0 + + COALESCE(cache_read_input_tokens, 0) * {p_cr} / 1000000.0 + + CASE + WHEN COALESCE(cache_creation_5m_input_tokens, 0) + + COALESCE(cache_creation_1h_input_tokens, 0) > 0 + THEN COALESCE(cache_creation_5m_input_tokens, 0) * {p_c5m} / 1000000.0 + + COALESCE(cache_creation_1h_input_tokens, 0) * {p_c1h} / 1000000.0 + ELSE COALESCE(cache_creation_input_tokens, 0) * {p_c5m} / 1000000.0 + END + + COALESCE(web_search_requests, 0) * {p_ws}, + 6 + ) + WHERE rowid IN (SELECT rid FROM _v036_targets) AND {model_where} + """, + params, + ) + + # -- Step 2: add the recovered cost to sessions.total_cost_usd ---------- + # The target rows contributed $0 to the stored session totals, so the + # recovered per-session sum is exactly the correction to add. + async with db.execute( + """ + SELECT t.session_id, COALESCE(SUM(u.cost_usd), 0) AS recovered + FROM _v036_targets t + JOIN session_usage u ON u.rowid = t.rid + GROUP BY t.session_id + HAVING recovered > 0 + """ + ) as cur: + per_session = await cur.fetchall() + + total_recovered = 0.0 + for session_id, recovered in per_session: + total_recovered += recovered + await db.execute( + "UPDATE sessions " + "SET total_cost_usd = ROUND(COALESCE(total_cost_usd, 0) + ?, 6) " + "WHERE id = ?", + (recovered, session_id), + ) + + await db.execute("DROP TABLE IF EXISTS _v036_targets") + await db.commit() + logger.info( + "V36 migration: backfilled %d zero-cost turns from token counts " + "(%.2f USD recovered across %d sessions)", + target_count, total_recovered, len(per_session), + ) diff --git a/nerve/db/tasks.py b/nerve/db/tasks.py index 0047ed4..ad3d160 100644 --- a/nerve/db/tasks.py +++ b/nerve/db/tasks.py @@ -149,8 +149,15 @@ async def update_task_tags(self, task_id: str, tags: str) -> None: # ── FTS query building ─────────────────────────────────────────────── - # Characters that are FTS5 syntax or punctuation — replaced with spaces. - _FTS_CLEAN_RE = re.compile(r'["\*\(\)\-:/\\#\.\,\;\'\[\]\{\}@!?\^~`]') + # Anything that is NOT a word character (unicode letters, digits, + # underscore) is FTS5 syntax or punctuation — replaced with spaces. + # Allowlist rather than blocklist: the old punctuation blocklist + # missed characters like `$`, `+`, `=`, `&`, `%`, which slipped into + # the MATCH expression as barewords and raised `fts5: syntax error` + # (e.g. any query containing a dollar amount). Every char `\w` + # keeps is legal in an unquoted FTS5 bareword: ASCII alphanumerics, + # underscore, and codepoints > 127. + _FTS_CLEAN_RE = re.compile(r"[^\w]") # Common stop words filtered from search queries. _FTS_STOP_WORDS = frozenset({ diff --git a/nerve/db/usage.py b/nerve/db/usage.py index 7a61407..a5defa8 100644 --- a/nerve/db/usage.py +++ b/nerve/db/usage.py @@ -350,6 +350,66 @@ def estimate_turn_cost(usage: dict, model: str | None = None) -> float: return round(cost, 6) +def compute_turn_cost( + sdk_cost: float | None, + prev_cumulative: float, + usage: dict | None, + model: str | None = None, +) -> tuple[float, str]: + """Derive this turn's cost from the SDK's cumulative cost counter. + + The Claude Agent SDK reports ``ResultMessage.total_cost_usd`` as a + *cumulative* total per CLI client process. The caller persists the + last seen cumulative value (``_sdk_cumulative_cost`` in session + metadata) and passes it as ``prev_cumulative``; the normal per-turn + cost is the delta between the two. + + Two counter failure modes are handled explicitly (this logic has + produced two accounting incidents — v024 fixed over-counting when + the cumulative was treated as per-turn; the reset handling below + fixes under-counting): + + * **Counter reset** — every client recycle (idle sweep, oneshot + cron teardown, ``nerve restart``, model switch) starts a fresh + CLI process whose cumulative begins near zero, i.e. *below* the + persisted high-water mark. A naive ``max(delta, 0)`` clamp + records $0 for the first turn on the new client — typically the + most expensive turn, since it replays the resumed context. The + new process's cumulative IS its spend so far, so attribute it to + this turn instead. + + * **Zero with traffic** — the counter reports no spend while the + turn moved real tokens (stuck counter, provider that doesn't + price usage). Fall back to the token-based estimate so the row + never records a silent zero. + + Returns ``(turn_cost, source)`` where ``source`` is one of: + + - ``"sdk_delta"`` — normal cumulative delta (also covers genuinely + free/tiny turns, which legitimately stay at ~0); + - ``"sdk_reset"`` — counter went backwards, new cumulative used; + - ``"estimate"`` — SDK reported no cost at all; + - ``"estimate_backstop"`` — SDK cost was zero-ish despite real + token traffic, token-based estimate used. + """ + usage = usage or {} + if sdk_cost is None: + return estimate_turn_cost(usage, model=model), "estimate" + + if sdk_cost >= prev_cumulative: + turn_cost, source = sdk_cost - prev_cumulative, "sdk_delta" + else: + turn_cost, source = sdk_cost, "sdk_reset" + + # Backstop: zero-ish cost with real token traffic → the counter lied + # some other way; trust the token-based estimate instead. + if turn_cost <= 0: + est = estimate_turn_cost(usage, model=model) + if est > 0.0005: + return est, "estimate_backstop" + return turn_cost, source + + def estimate_cost_from_totals(totals: dict, model: str | None = None) -> float: """Estimate USD cost from aggregated totals. diff --git a/tests/test_cost_accounting.py b/tests/test_cost_accounting.py new file mode 100644 index 0000000..2713e01 --- /dev/null +++ b/tests/test_cost_accounting.py @@ -0,0 +1,384 @@ +"""Tests for reset-aware per-turn cost accounting. + +The SDK's ``ResultMessage.total_cost_usd`` is cumulative per CLI client +process. This subsystem has produced two accounting incidents: treating +the cumulative as per-turn (over-count, fixed by v024) and clamping the +delta with ``max(delta, 0)`` (under-count on every client recycle — +recorded $0 for the first turn of each new client, i.e. every persistent +cron run). These tests pin the reset-aware semantics +(``nerve.db.usage.compute_turn_cost``), the engine's baseline reset on +client creation, the v036 backfill migration, and the FTS5 query +sanitization that broke while filing the incident. +""" + +from __future__ import annotations + +import json +from types import SimpleNamespace + +import pytest + +from nerve.db import Database +from nerve.db.migrations.v036_backfill_zero_cost_turns import up as v036_up +from nerve.db.usage import compute_turn_cost, estimate_turn_cost + +# A usage dict with real token traffic (estimate well above the backstop +# threshold on any pricing tier). +_BUSY_USAGE = { + "input_tokens": 1_000, + "output_tokens": 500, + "cache_read_input_tokens": 200_000, + "cache_creation_input_tokens": 20_000, +} + + +# --------------------------------------------------------------------------- +# compute_turn_cost: reset-aware delta semantics +# --------------------------------------------------------------------------- + + +class TestComputeTurnCost: + def test_monotonic_increase_is_exact_delta(self): + cost, source = compute_turn_cost(5.2, 5.0, _BUSY_USAGE, model="claude-fable-5") + assert cost == pytest.approx(0.2) + assert source == "sdk_delta" + + def test_fresh_session_uses_full_cumulative(self): + cost, source = compute_turn_cost(0.3, 0, _BUSY_USAGE, model="claude-fable-5") + assert cost == pytest.approx(0.3) + # prev=0 is the fresh-session baseline: plain delta, no reset. + assert source == "sdk_delta" + + def test_counter_reset_attributes_new_cumulative(self): + # Client recycled: cumulative dropped from 5.0 to 0.3. The old + # max(delta, 0) clamp recorded $0 here — the core of the bug. + cost, source = compute_turn_cost(0.3, 5.0, _BUSY_USAGE, model="claude-fable-5") + assert cost == pytest.approx(0.3) + assert source == "sdk_reset" + + def test_no_sdk_cost_falls_back_to_estimate(self): + cost, source = compute_turn_cost(None, 5.0, _BUSY_USAGE, model="claude-fable-5") + assert cost == estimate_turn_cost(_BUSY_USAGE, model="claude-fable-5") + assert cost > 0 + assert source == "estimate" + + def test_zero_cost_with_traffic_uses_estimate_backstop(self): + # Counter stuck: same cumulative as before despite real tokens. + cost, source = compute_turn_cost(5.0, 5.0, _BUSY_USAGE, model="claude-fable-5") + assert cost == estimate_turn_cost(_BUSY_USAGE, model="claude-fable-5") + assert cost > 0 + assert source == "estimate_backstop" + + def test_reset_to_zero_with_traffic_uses_estimate_backstop(self): + # Recycled client that reports 0.0 cumulative on its first turn + # (e.g. a provider that does not price usage). + cost, source = compute_turn_cost(0.0, 5.0, _BUSY_USAGE, model="claude-fable-5") + assert cost == estimate_turn_cost(_BUSY_USAGE, model="claude-fable-5") + assert source == "estimate_backstop" + + def test_genuinely_free_turn_stays_zero(self): + # No token traffic → the zero is real, no false-positive backstop. + cost, source = compute_turn_cost(5.0, 5.0, {}, model="claude-fable-5") + assert cost == 0 + assert source == "sdk_delta" + + def test_tiny_turn_below_threshold_keeps_sdk_zero(self): + # A handful of tokens estimates below the 0.0005 backstop + # threshold — trust the SDK's zero. + tiny = {"input_tokens": 10, "output_tokens": 5} + cost, source = compute_turn_cost(5.0, 5.0, tiny, model="claude-haiku-4-5") + assert cost == 0 + assert source == "sdk_delta" + + def test_reset_with_nonzero_first_turn_not_backstopped(self): + # Reset with a real first-turn cost: use the SDK number, not the + # estimate, even if they differ. + cost, source = compute_turn_cost(0.003, 9.9, _BUSY_USAGE, model="claude-fable-5") + assert cost == pytest.approx(0.003) + assert source == "sdk_reset" + + def test_none_usage_is_safe(self): + cost, source = compute_turn_cost(None, 0, None, model=None) + assert cost == 0 + assert source == "estimate" + + +# --------------------------------------------------------------------------- +# Engine: baseline zeroed when a new CLI client is created +# --------------------------------------------------------------------------- + + +@pytest.mark.asyncio +class TestResetCostBaseline: + async def _shim(self, db: Database): + return SimpleNamespace(db=db) + + async def test_baseline_zeroed(self, db: Database): + from nerve.agent.engine import AgentEngine + + await db.create_session("sess-base", source="web") + await db.update_session_metadata( + "sess-base", {"_sdk_cumulative_cost": 4.2, "other_key": "kept"}, + ) + + await AgentEngine._reset_cost_baseline(await self._shim(db), "sess-base") + + session = await db.get_session("sess-base") + meta = json.loads(session["metadata"]) + assert meta["_sdk_cumulative_cost"] == 0 + # Sibling metadata keys survive the rewrite. + assert meta["other_key"] == "kept" + + async def test_noop_when_baseline_absent(self, db: Database): + from nerve.agent.engine import AgentEngine + + await db.create_session("sess-nobase", source="web") + await db.update_session_metadata("sess-nobase", {"other_key": "kept"}) + + await AgentEngine._reset_cost_baseline(await self._shim(db), "sess-nobase") + + meta = json.loads((await db.get_session("sess-nobase"))["metadata"]) + assert meta == {"other_key": "kept"} + + async def test_missing_session_is_safe(self, db: Database): + from nerve.agent.engine import AgentEngine + + await AgentEngine._reset_cost_baseline(await self._shim(db), "no-such-session") + + +# --------------------------------------------------------------------------- +# v036 migration: backfill zero-cost turns from token counts +# --------------------------------------------------------------------------- + +# Seed rows for the "repaired" session. Costs recompute from token +# counts with model pricing — cross-checked against estimate_turn_cost. +_ROW_SPLIT = { + "input_tokens": 1_000, + "output_tokens": 500, + "cache_read_input_tokens": 100_000, + "cache_creation_input_tokens": 20_000, + "cache_creation": { + "ephemeral_5m_input_tokens": 15_000, + "ephemeral_1h_input_tokens": 5_000, + }, +} +_ROW_AGGREGATE = { + "input_tokens": 2_000, + "output_tokens": 1_000, + "cache_read_input_tokens": 50_000, + "cache_creation_input_tokens": 10_000, + "server_tool_use": {"web_search_requests": 3}, +} + + +async def _fetch_costs(db: Database, session_id: str) -> list[float]: + async with db.db.execute( + "SELECT cost_usd FROM session_usage WHERE session_id = ? ORDER BY rowid", + (session_id,), + ) as cur: + return [row[0] async for row in cur] + + +async def _session_total(db: Database, session_id: str) -> float: + session = await db.get_session(session_id) + return session["total_cost_usd"] or 0 + + +@pytest.mark.asyncio +class TestBackfillMigration: + async def _seed(self, db: Database) -> None: + # Session A: two swallowed turns (one with the 5m/1h split, one + # aggregate-only + web searches) plus one correctly-priced turn. + await db.create_session("sess-a", source="cron") + await db.record_turn_usage( + session_id="sess-a", + input_tokens=_ROW_SPLIT["input_tokens"], + output_tokens=_ROW_SPLIT["output_tokens"], + cache_creation=_ROW_SPLIT["cache_creation_input_tokens"], + cache_read=_ROW_SPLIT["cache_read_input_tokens"], + cache_creation_5m=15_000, + cache_creation_1h=5_000, + max_context=200_000, + model="claude-fable-5", + cost_usd=0, + ) + await db.record_turn_usage( + session_id="sess-a", + input_tokens=_ROW_AGGREGATE["input_tokens"], + output_tokens=_ROW_AGGREGATE["output_tokens"], + cache_creation=_ROW_AGGREGATE["cache_creation_input_tokens"], + cache_read=_ROW_AGGREGATE["cache_read_input_tokens"], + max_context=200_000, + model="claude-opus-4-7", + cost_usd=0, + web_search_requests=3, + ) + await db.record_turn_usage( + session_id="sess-a", + input_tokens=100, output_tokens=100, + cache_creation=0, cache_read=0, + max_context=200_000, + model="claude-fable-5", + cost_usd=1.0, + ) + # Stored total only ever saw the correctly-priced turn. + await db.update_session_fields("sess-a", {"total_cost_usd": 1.0}) + + # Session B: a zero-cost row with zero tokens — NOT a target. + await db.create_session("sess-b", source="web") + await db.record_turn_usage( + session_id="sess-b", + input_tokens=0, output_tokens=0, + cache_creation=0, cache_read=0, + max_context=200_000, + cost_usd=0, + ) + await db.update_session_fields("sess-b", {"total_cost_usd": 0}) + + # Session C: swallowed turn with unknown model → default pricing. + await db.create_session("sess-c", source="telegram") + await db.record_turn_usage( + session_id="sess-c", + input_tokens=1_000_000, output_tokens=0, + cache_creation=0, cache_read=0, + max_context=200_000, + model=None, + cost_usd=0, + ) + await db.update_session_fields("sess-c", {"total_cost_usd": 0}) + + # Session D: total accumulated from usage rows that telemetry + # pruning has since deleted. The migration must NOT re-sum this + # to zero — pruned history is legitimate spend. + await db.create_session("sess-d", source="web") + await db.update_session_fields("sess-d", {"total_cost_usd": 10.0}) + + async def test_backfill_recomputes_from_tokens(self, db: Database): + await self._seed(db) + await v036_up(db.db) + + costs = await _fetch_costs(db, "sess-a") + expected_split = estimate_turn_cost(_ROW_SPLIT, model="claude-fable-5") + expected_agg = estimate_turn_cost(_ROW_AGGREGATE, model="claude-opus-4-7") + assert costs[0] == pytest.approx(expected_split) + assert costs[1] == pytest.approx(expected_agg) + assert costs[0] > 0 and costs[1] > 0 + # The correctly-priced row is untouched. + assert costs[2] == pytest.approx(1.0) + + async def test_backfill_uses_default_pricing_for_unknown_model(self, db: Database): + await self._seed(db) + await v036_up(db.db) + + costs = await _fetch_costs(db, "sess-c") + # 1M fresh input tokens at the default tier ($5/MTok). + assert costs[0] == pytest.approx(5.0) + + async def test_zero_token_rows_untouched(self, db: Database): + await self._seed(db) + await v036_up(db.db) + + costs = await _fetch_costs(db, "sess-b") + assert costs == [0] + + async def test_session_totals_add_recovered_delta(self, db: Database): + await self._seed(db) + await v036_up(db.db) + + expected_split = estimate_turn_cost(_ROW_SPLIT, model="claude-fable-5") + expected_agg = estimate_turn_cost(_ROW_AGGREGATE, model="claude-opus-4-7") + assert await _session_total(db, "sess-a") == pytest.approx( + 1.0 + expected_split + expected_agg, + ) + assert await _session_total(db, "sess-c") == pytest.approx(5.0) + assert await _session_total(db, "sess-b") == 0 + # Totals now match the per-turn sum for sessions with full history. + totals_a = await db.get_session_usage_totals("sess-a") + assert totals_a["total_cost_usd"] == pytest.approx( + await _session_total(db, "sess-a"), + ) + + async def test_pruned_session_total_preserved(self, db: Database): + await self._seed(db) + await v036_up(db.db) + + assert await _session_total(db, "sess-d") == pytest.approx(10.0) + + async def test_rerun_is_idempotent(self, db: Database): + await self._seed(db) + await v036_up(db.db) + + costs_first = { + sid: await _fetch_costs(db, sid) + for sid in ("sess-a", "sess-b", "sess-c") + } + totals_first = { + sid: await _session_total(db, sid) + for sid in ("sess-a", "sess-b", "sess-c", "sess-d") + } + + await v036_up(db.db) + + for sid, costs in costs_first.items(): + assert await _fetch_costs(db, sid) == costs + for sid, total in totals_first.items(): + assert await _session_total(db, sid) == pytest.approx(total) + + async def test_empty_db_is_noop(self, db: Database): + await v036_up(db.db) + + +# --------------------------------------------------------------------------- +# FTS5 query sanitization (the dup-check crash found while filing this) +# --------------------------------------------------------------------------- + + +class TestFtsQueryBuilding: + def test_dollar_sign_stripped_from_fts_query(self): + q = Database._build_fts_query("cost recorded as $0 per turn") + assert q # real terms survive + assert "$" not in q + + def test_operator_characters_stripped(self): + q = Database._build_fts_query("a+b=c & 100% | $money NEAR near") + assert q + for ch in "+=&%<>|$": + assert ch not in q + + def test_unicode_words_preserved(self): + words = Database._tokenize_query("стоимость сессии $55") + assert "стоимость" in words + assert "сессии" in words + + +@pytest.mark.asyncio +class TestFtsSanitization: + async def _seed_task(self, db: Database) -> None: + await db.upsert_task( + task_id="2026-01-01-turn-cost-recorded-as-zero", + file_path="tasks/2026-01-01-turn-cost-recorded-as-zero.md", + title="Turn cost recorded as $0 after client recycle", + content="Cost counter resets swallow the first turn (about $100).", + tags="accounting,cost", + ) + + async def test_search_tasks_with_dollar_query(self, db: Database): + await self._seed_task(db) + # Previously raised `fts5: syntax error near "$"`. + results = await db.search_tasks("cost recorded as $0") + assert any(t["id"] == "2026-01-01-turn-cost-recorded-as-zero" for t in results) + + async def test_search_tasks_similar_with_special_chars(self, db: Database): + await self._seed_task(db) + # The duplicate-check path used by task_create. + results = await db.search_tasks_similar( + "Per-turn cost silently recorded as $0 — max(delta,0) clamp " + "swallows ~$9.99 & 100% of cron runs", + ) + assert any(t["id"] == "2026-01-01-turn-cost-recorded-as-zero" for t in results) + + async def test_search_all_punctuation_query_is_safe(self, db: Database): + await self._seed_task(db) + # Nothing but FTS operators/punctuation → no crash, graceful result. + await db.search_tasks("$$$ +++ ===") + await db.search_tasks_similar("$$$ +++ ===")