fix(credits): derive agent/quota/cycle surfaces from the ledger (retire counter drift)

apps/api/core/credits.py

@@ -256,16 +256,42 @@ async def check_run_credit_cap(db, agent_id, additional_cost: int) -> None:
         })
 
 
+async def credits_used_this_cycle(conn, user_id: str) -> int:
+    """Authoritative credits consumed this billing cycle (calendar month) for a
+    user — summed from the immutable credit_transactions ledger.
+
+    Replaces the drift-prone api_keys.monthly_calls_count for every display/quota
+    surface. monthly_calls_count was maintained by a separate non-atomic path
+    (_increment_calls) that under-counted — it never saw NULL-api_key debits like
+    /search and the LLM path under-incremented — so any figure derived from it
+    (remaining quota, "used this cycle", invoice) disagreed with the real spend.
+    Summing the ledger ties every surface to user_credits.credits_balance."""
+    used = await conn.fetchval("""
+        SELECT COALESCE(SUM(-amount), 0)
+          FROM credit_transactions
+         WHERE user_id = $1::uuid AND amount < 0
+           AND type IN ('execution', 'cross_rail', 'cloud_compute')
+           AND created_at >= date_trunc('month', NOW())
+    """, user_id)
+    return int(used or 0)
+
+
 async def compute_calls_remaining(conn, api_key_id: str) -> int:
-    """Single source of truth for calls_remaining. Reads monthly_calls_count directly — never uses credit math."""
+    """Credits remaining in the cycle allotment — DERIVED from the ledger.
+
+    = plan allotment − credits_used_this_cycle (ledger sum), per USER (not per
+    key), so it captures all spend including NULL-key /search debits. No longer
+    reads api_keys.monthly_calls_count, which drifted low and over-stated
+    remaining (the pre-money quota under-count)."""
     row = await conn.fetchrow(
-        "SELECT monthly_calls_count, tier FROM api_keys WHERE id = $1::uuid",
+        "SELECT user_id, tier FROM api_keys WHERE id = $1::uuid",
         api_key_id,
     )
     if not row:
         return 0
     p = PLANS.get(row["tier"], PLANS["free"])
-    return max(0, p["calls_included"] - row["monthly_calls_count"])
+    used = await credits_used_this_cycle(conn, str(row["user_id"]))
+    return max(0, p["calls_included"] - used)
 
 
 async def _maybe_send_usage_warning_email(pool, user_id: str, calls_remaining: int, percent_used: int, tier: str) -> None:

apps/api/routers/billing/account.py

@@ -12,7 +12,7 @@
 from fastapi import APIRouter, Depends, HTTPException, Request
 
 from core.auth import resolve_dashboard_caller
-from core.credits import PLANS, CREDITS_PER_CALL, ROUTING_FEE, PAYMENT_MULTIPLIERS, compute_calls_remaining
+from core.credits import PLANS, CREDITS_PER_CALL, ROUTING_FEE, PAYMENT_MULTIPLIERS, compute_calls_remaining, credits_used_this_cycle
 from core.db import get_db
 from core.rate_limit import limiter
 from core.tier_gates import require_tier, _get_redis
@@ -303,7 +303,8 @@ def _kr(field, default=None):
 
     # Forecast — daily average in credits (not calls). Returns null if < 3 days history.
     forecast = None
-    monthly_credits_consumed = _kr("monthly_calls_count", 0) or 0
+    # Ledger-derived (NOT api_keys.monthly_calls_count, which drifts low).
+    monthly_credits_consumed = await credits_used_this_cycle(db, user_id)
     if monthly_reset_at:
         now_utc = datetime.now(timezone.utc)
         period_start = monthly_reset_at.replace(tzinfo=timezone.utc) - timedelta(days=30)
@@ -455,7 +456,9 @@ async def account_analytics(request: Request, db=Depends(get_db)):
         "WHERE user_id=$1 AND type='execution' AND created_at >= date_trunc('month', NOW()) "
         "AND service_id IS NOT NULL GROUP BY service_id ORDER BY count DESC LIMIT 10", user_id)
 
-    # ── Calls — source of truth: monthly_calls_count on api_keys ─────────────
+    # ── Credits used this cycle — source of truth: the credit_transactions
+    # ledger (NOT api_keys.monthly_calls_count, which drifts low and under-counts
+    # NULL-key /search debits). Ties this "used this cycle" figure to the balance.
     today = _datetime.date.today()
     if today.month == 12:
         reset = _datetime.date(today.year + 1, 1, 1)
@@ -520,7 +523,7 @@ def _find_v2(slug: str):
     plan_tier = credits["package_tier"] if credits else "free"
     plan_def = PLANS.get(plan_tier, PLANS["free"])
     credits_included = plan_def["calls_included"]   # calls_included == monthly_credits after fix
-    credits_used = caller["monthly_calls_count"] or 0
+    credits_used = await credits_used_this_cycle(db, user_id)
     credits_remaining = max(0, credits_included - credits_used)
     reset_at_str = (
         caller["monthly_calls_reset_at"].date().isoformat()

apps/api/routers/cloud.py

@@ -595,18 +595,27 @@ async def list_agents(request: Request, db=Depends(get_db)) -> dict:
     user_id, _, tier = await _resolve_caller(request, db)
     require_tier(tier, "cloud_agents")
 
+    # credits_30d is DERIVED from the credit_transactions ledger (the authoritative
+    # spend record), not SUM(agent_runs.credits_total) — a denormalized per-run
+    # column that can lag/zero out and drift from what was actually billed. Cloud
+    # run debits are tagged with the run id (X_WAYFORTH_AGENT_ID), so we join the
+    # ledger to the agent's runs.
     rows = await db.fetch("""
         SELECT
             ha.id, ha.name, ha.slug, ha.runtime, ha.status,
             ha.trigger_type, ha.schedule, ha.credit_cap,
             ha.last_run_at, ha.created_at,
-            COALESCE(SUM(ar.credits_total) FILTER (
-                WHERE ar.created_at >= NOW() - INTERVAL '30 days'
+            COALESCE((
+                SELECT SUM(-ct.amount)
+                  FROM credit_transactions ct
+                  JOIN agent_runs ar ON ct.agent_id = ar.id::text
+                 WHERE ar.hosted_agent_id = ha.id
+                   AND ct.amount < 0
+                   AND ct.type IN ('execution', 'cloud_compute', 'cross_rail')
+                   AND ct.created_at >= NOW() - INTERVAL '30 days'
             ), 0) AS credits_30d
         FROM hosted_agents ha
-        LEFT JOIN agent_runs ar ON ar.hosted_agent_id = ha.id
         WHERE ha.user_id = $1::uuid
-        GROUP BY ha.id
         ORDER BY ha.created_at DESC
         LIMIT 100
     """, user_id)
@@ -640,6 +649,20 @@ async def get_agent(request: Request, agent_id: str, db=Depends(get_db)) -> dict
 
     agent = await _get_agent_or_404(db, user_id, agent_id)
 
+    # CREDITS (30D) for the detail page — DERIVED from the credit_transactions
+    # ledger (authoritative spend), not a denormalized column. This field was
+    # previously absent from the detail endpoint entirely, so the dashboard had
+    # nothing to read and showed 0. Cloud run debits are tagged with the run id.
+    credits_30d = await db.fetchval("""
+        SELECT COALESCE(SUM(-ct.amount), 0)
+          FROM credit_transactions ct
+          JOIN agent_runs ar ON ct.agent_id = ar.id::text
+         WHERE ar.hosted_agent_id = $1::uuid
+           AND ct.amount < 0
+           AND ct.type IN ('execution', 'cloud_compute', 'cross_rail')
+           AND ct.created_at >= NOW() - INTERVAL '30 days'
+    """, agent_id)
+
     # Last run summary
     last_run = await db.fetchrow("""
         SELECT id, status, trigger, started_at, completed_at, duration_ms,
@@ -659,6 +682,7 @@ async def get_agent(request: Request, agent_id: str, db=Depends(get_db)) -> dict
         "schedule":       agent["schedule"],
         "credit_cap":     agent["credit_cap"],
         "concurrent_max": agent.get("concurrent_max") or 1,
+        "credits_30d":    int(credits_30d or 0),
         "sandbox_provider": agent["sandbox_provider"],
         "webhook_id":     str(agent["webhook_id"]) if agent.get("webhook_id") else None,
         "next_run_at":    agent["next_run_at"].isoformat() if agent.get("next_run_at") else None,