Reset-aware per-turn cost accounting + backfill of swallowed turn costs

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)

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),
+    )

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({

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.