feat: meter research runs against account credits (route-0 billing)

dashboard/backend/api/routers/research.py

@@ -19,6 +19,9 @@
 
 import base64
 import os
+import threading
+import time
+import uuid
 from typing import Any, Dict, Optional
 
 import httpx
@@ -28,13 +31,39 @@
 from dashboard.backend.api.auth import get_current_user
 from dashboard.backend.domain.agents import marketplace as marketplace_mod
 from dashboard.backend.domain.agents import research_store
+from dashboard.backend.domain.credits.models import credits_micro_for_cents
+from dashboard.backend.domain.credits.repository_common import (
+    CreditAccountRestrictedStoreError,
+    InsufficientCreditsError,
+)
+from dashboard.backend.domain.credits.service import credits_service
+from dashboard.backend import users as users_module
+from dashboard.backend.domain.credits.repository_common import (
+    CreditAccountRestrictedStoreError,
+    InsufficientCreditsError,
+)
+from dashboard.backend.domain.credits.service import credits_service
+from dashboard.backend import users as users_module
 
 router = APIRouter(prefix="/v1/research", tags=["research"])
 
 POLL_TIMEOUT_SECONDS = 10.0
 MANIFEST_CACHE_SECONDS = 300.0
 _manifest_cache: Dict[str, Any] = {}
 
+# Route-0 billing on the REAL credit rail: reserve a per-run usage ceiling at
+# submit, settle at the same amount on completion, release on failure. When
+# contract v1.1 adds usage reporting, `actual_micro` becomes the reported
+# spend instead of the estimate — the reserve/settle calls stay.
+def _estimate_usd_cents() -> int:
+    raw = (os.getenv("RESEARCH_RUN_ESTIMATE_USD_CENTS") or "").strip()
+    try:
+        value = int(raw)
+    except ValueError:
+        return 100  # $1.00 per Deep Research run, absent operator tuning
+    return value if value > 0 else 100
+
+
 ARTIFACT_CONTENT_TYPES = {
     "docx": "application/vnd.openxmlformats-officedocument.wordprocessingml.document",
     "pdf": "application/pdf",
@@ -175,6 +204,34 @@ def create_research_run(
 
     email_me = bool(body.get("email_me"))
     payload = {"agent_id": _agent_id(template), "settings": settings}
+
+    # Billing: reserve a per-run usage ceiling from the caller's real Credits
+    # balance — the same $1 = 1 credit rail platform-credit backtests settle
+    # on. The store raises InsufficientCreditsError when the balance can't
+    # cover the ceiling, which maps to the 402. (Route-0 interim: completion
+    # settles at the reserved estimate; contract v1.1's usage reporting turns
+    # this into settle-at-actual.)
+    run_id = f"rr_{uuid.uuid4().hex[:12]}"
+    estimate_micro = credits_micro_for_cents(_estimate_usd_cents())
+    try:
+        reservation = credits_service.reserve_llm_credits(
+            user_id=current_user["id"],
+            run_id=run_id,
+            call_index=0,
+            amount_micro=estimate_micro,
+            provider_id=_agent_id(template).replace("-", "_"),
+        )
+    except InsufficientCreditsError as exc:
+        raise HTTPException(status_code=402, detail=str(exc)) from None
+    except CreditAccountRestrictedStoreError as exc:
+        # A paused/restricted Credits account must get the deliberate refusal
+        # other Credits surfaces use, not an uncaught 500.
+        raise HTTPException(status_code=403, detail=str(exc)) from None
+    reservation_id = str(reservation.reservation_id)
+
+    # From here until create_run, any failure must release the just-taken
+    # hold — an unreleased reservation permanently strands the user's
+    # balance (each client retry strands another chunk).
     try:
         response = httpx.post(
             f"{_service_base(template)}/runs",
@@ -184,25 +241,86 @@ def create_research_run(
         )
         response.raise_for_status()
         service_run = response.json()
-    except httpx.HTTPError as exc:
-        raise _service_error(exc, "submitting the research run") from None
-
-    import uuid
+        if not isinstance(service_run, dict):
+            raise ValueError("submit response was not a JSON object")
+    except (httpx.HTTPError, ValueError) as exc:
+        credits_service.release_llm_credits(
+            reservation_id, reason="submit failed; no Deep Research invocation"
+        )
+        if isinstance(exc, httpx.HTTPError):
+            raise _service_error(exc, "submitting the research run") from None
+        raise HTTPException(
+            status_code=502,
+            detail="Research service returned a malformed submit response",
+        ) from None
 
     service_run_id = str(service_run.get("run_id") or "").strip()
-    run_id = f"rr_{service_run_id}" if service_run_id else f"rr_{uuid.uuid4().hex[:12]}"
     research_store.create_run(
         run_id=run_id,
         user_id=current_user["id"],
         template_id=template_id,
         service_run_id=service_run_id,
+        reservation_id=reservation_id,
+        estimate_micro=estimate_micro,
         status=str(service_run.get("status") or "running"),
         settings=settings,
         email_me=email_me,
     )
     return {"run_id": run_id, "status": service_run.get("status") or "running"}
 
 
+def _estimate_micro_from_run(run: Dict[str, Any]) -> int:
+    """Route-0 settle amount: the reserved estimate, from env-tunable cents."""
+    return credits_micro_for_cents(_estimate_usd_cents())
+
+
+# Background sweeper (v1.1): discovers completed/failed runs without relying
+# on the submitting user keeping the page open. Same logic the status route
+# runs, just on a timer; single in-process worker (matches the codebase's
+# single-worker assumption). In-flight guard keeps it from racing a
+# user-driven poll on the same run.
+_SWEEP_INTERVAL_SECONDS = 60
+_sweep_lock = threading.Lock()
+_sweep_inflight: set = set()
+
+
+def _sweep_pending_runs() -> None:
+    for row in research_store.list_nonterminal_runs():
+        run_id = row["run_id"]
+        with _sweep_lock:
+            if run_id in _sweep_inflight:
+                continue
+            _sweep_inflight.add(run_id)
+        try:
+            template = marketplace_mod.get_marketplace_template(row["template_id"])
+            if not template or not marketplace_mod.shelf_is_research(template):
+                continue
+            user = users_module.user_store.get_user_by_id(row["user_id"])
+            if not user:
+                continue
+            fresh = research_store.get_run(run_id, row["user_id"])
+            if not fresh:
+                continue
+            _maybe_complete_run(fresh, template, user)
+        except Exception as exc:  # noqa: BLE001 - one bad run must not kill the sweep
+            print(f"research sweeper: run {run_id} sweep error: {exc}")
+        finally:
+            with _sweep_lock:
+                _sweep_inflight.discard(run_id)
+
+
+def _sweeper_loop() -> None:
+    while True:
+        try:
+            _sweep_pending_runs()
+        except Exception as exc:  # noqa: BLE001
+            print(f"research sweeper: loop error: {exc}")
+        time.sleep(_SWEEP_INTERVAL_SECONDS)
+
+
+threading.Thread(target=_sweeper_loop, name="research-sweeper", daemon=True).start()
+
+
 def _service_run_id(run: Dict[str, Any]) -> str:
     return str(run.get("service_run_id") or "")
 
@@ -230,6 +348,10 @@ def _maybe_complete_run(run: Dict[str, Any], template: Dict[str, Any],
         research_store.update_run_status(run["run_id"], "failed",
                                          error=service_status.get("error") or "Research failed",
                                          completed=True)
+        if run.get("reservation_id"):
+            credits_service.release_llm_credits(
+                run["reservation_id"], reason="research run failed",
+            )
         run["status"] = "failed"
         return run
     if status != "completed":
@@ -244,10 +366,9 @@ def _maybe_complete_run(run: Dict[str, Any], template: Dict[str, Any],
         result.raise_for_status()
         payload = result.json()
     except httpx.HTTPError:
-        research_store.update_run_status(run["run_id"], "failed",
-                                         error="Result could not be retrieved",
-                                         completed=True)
-        run["status"] = "failed"
+        # Result fetch failed but the run COMPLETED on the service — this is
+        # retryable (next poll re-fetches), not a failed run. Making it
+        # terminal here would strand the reservation on a transient 5xx.
         return run
 
     research_store.store_artifacts(
@@ -256,6 +377,22 @@ def _maybe_complete_run(run: Dict[str, Any], template: Dict[str, Any],
         payload.get("evidence"),
         payload.get("report_markdown") or "",
     )
+    reservation_id = run.get("reservation_id")
+    if not reservation_id:
+        # Pre-billing legacy run (route-0 migration): nothing was reserved, so
+        # complete without touching Credits — settling a NULL id would raise.
+        research_store.update_run_status(run["run_id"], "completed", completed=True)
+        run["status"] = "completed"
+        return run
+    # Route-0 interim: settle at the amount actually reserved for THIS run
+    # (persisted at submit — re-reading the env here would settle a run at a
+    # price chosen after it started). Contract v1.1's usage reporting upgrades
+    # this to settle(actual_micro=reported spend).
+    credits_service.settle_llm_credits(
+        reservation_id,
+        actual_micro=int(run.get("estimate_micro") or 0),
+        evidence={"source": "research-agent", "agent_id": run["template_id"]},
+    )
     research_store.update_run_status(run["run_id"], "completed", completed=True)
     run["status"] = "completed"
 

dashboard/backend/domain/agents/research_store.py

@@ -47,6 +47,7 @@ def _init_schema() -> None:
                 user_id INTEGER NOT NULL,
                 template_id TEXT NOT NULL,
                 service_run_id TEXT,
+                reservation_id TEXT,
                 status TEXT NOT NULL DEFAULT 'queued',
                 settings_json TEXT NOT NULL,
                 email_me INTEGER NOT NULL DEFAULT 0,
@@ -68,6 +69,18 @@ def _init_schema() -> None:
                 ON research_runs(user_id, created_at DESC);
             """
         )
+        # Existing installs created the table before billing (route 0) added
+        # the reservation column; CREATE IF NOT EXISTS won't add it there.
+        migrations = (
+            "ALTER TABLE research_runs ADD COLUMN reservation_id TEXT",
+            "ALTER TABLE research_runs ADD COLUMN estimate_micro INTEGER",
+        )
+        for statement in migrations:
+            try:
+                with _connect() as conn:
+                    conn.execute(statement)
+            except sqlite3.OperationalError:
+                pass  # column already exists
 
 
 _init_schema()
@@ -111,19 +124,32 @@ def create_run(
     user_id: int,
     template_id: str,
     service_run_id: str,
+    reservation_id: str,
+    estimate_micro: int,
     status: str,
     settings: Dict[str, Any],
     email_me: bool,
 ) -> None:
     with _connect() as conn:
         conn.execute(
             "INSERT INTO research_runs (run_id, user_id, template_id, service_run_id,"
-            " status, settings_json, email_me) VALUES (?, ?, ?, ?, ?, ?, ?)",
-            (run_id, user_id, template_id, service_run_id, status,
+            " reservation_id, estimate_micro, status, settings_json, email_me)"
+            " VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)",
+            (run_id, user_id, template_id, service_run_id, reservation_id,
+             int(estimate_micro), status,
              json.dumps(settings, ensure_ascii=False), int(email_me)),
         )
 
 
+def list_nonterminal_runs() -> List[Dict[str, Any]]:
+    """All queued/running runs across users — the sweeper's work queue."""
+    with _connect() as conn:
+        rows = conn.execute(
+            "SELECT * FROM research_runs WHERE status IN ('queued', 'running')"
+        ).fetchall()
+    return [dict(row) for row in rows]
+
+
 def get_run(run_id: str, user_id: int) -> Optional[Dict[str, Any]]:
     with _connect() as conn:
         row = conn.execute(

dashboard/backend/domain/entitlements/credits.py

@@ -1,5 +1,13 @@
 """Credit metering for operator-funded LLM spend.
 
+**STATUS (2026-09): NOT wired into any production path.** The live backtest
+billing is the real-money credits system (`domain/credits/`):
+platform-credit runs reserve at accept and settle from actual provider usage
+(`reserve_llm_credits` / `settle_llm_credits`), and BYOK runs bill nothing.
+This module is the reference flat-quota implementation, kept until the
+team decides to wire or retire it — do not call it from new code without
+that discussion.
+
 One credit buys one LLM-driven dashboard backtest.
 
 **What is metered, and why only that.** A credit denominates operator money, so