Instrument dataset usage and billing analytics (#316)

* Instrument session usage analytics

Add durable usage_metrics snapshots, backend HF billing refreshes, and shared dataset row usage columns.

Co-authored-by: OpenAI Codex <codex@openai.com>

* Address usage billing review feedback

Timebox billing refreshes, keep user-facing saves fire-and-forget, and share sandbox lifecycle aggregation semantics.

Co-authored-by: OpenAI Codex <codex@openai.com>

* Simplify HF Jobs usage label

Display HF Jobs without the estimated qualifier while preserving fallback usage logic.

Co-authored-by: OpenAI Codex <codex@openai.com>

---------

Co-authored-by: OpenAI Codex <codex@openai.com>

agent/core/session.py

@@ -162,6 +162,8 @@ class Session:
         self.usage_warning_next_threshold_usd: float = USAGE_WARNING_FIRST_THRESHOLD_USD
         self.usage_threshold_checker: Any | None = None
         self.yolo_budget_checker: Any | None = None
+        self.usage_hf_billing_snapshot: dict[str, Any] | None = None
+        self.usage_metrics: dict[str, Any] | None = None
 
         # Session trajectory logging
         self.logged_events: list[dict] = []
@@ -467,6 +469,8 @@ class Session:
         self.pending_approval = None
         self.auto_approval_estimated_spend_usd = 0.0
         self._yolo_budget_reservations = {}
+        self.usage_hf_billing_snapshot = None
+        self.usage_metrics = None
         self.reset_cancel()
 
         # Previous-session metadata is intentionally included for event
@@ -535,6 +539,18 @@ class Session:
             for e in self.logged_events
             if e.get("event_type") == "llm_call"
         )
+        try:
+            from agent.core.usage_metrics import summarize_usage_events
+
+            usage_metrics = summarize_usage_events(
+                self.logged_events,
+                session_id=self.session_id,
+                hf_billing_snapshot=self.usage_hf_billing_snapshot,
+            )
+            self.usage_metrics = usage_metrics
+        except Exception as e:
+            logger.debug("Usage metrics summary failed for %s: %s", self.session_id, e)
+            usage_metrics = self.usage_metrics or {}
         return {
             "session_id": self.session_id,
             "user_id": self.user_id,
@@ -543,6 +559,7 @@ class Session:
             "session_end_time": datetime.now().isoformat(),
             "model_name": self.config.model_name,
             "total_cost_usd": total_cost_usd,
+            "usage_metrics": usage_metrics,
             "messages": [msg.model_dump() for msg in self.context_manager.items],
             "events": self.logged_events,
             "tools": tools,

agent/core/session_uploader.py

@@ -26,6 +26,11 @@ from typing import Any
 
 from dotenv import load_dotenv
 
+from agent.core.usage_metrics import (
+    summarize_usage_events,
+    usage_metric_scalar_fields,
+)
+
 load_dotenv()
 
 # Token resolution for the org KPI dataset. Fallback chain (least-privilege
@@ -261,9 +266,27 @@ def _scrub_session_for_upload(data: dict) -> dict:
     return scrubbed
 
 
+def _usage_metrics_for_row(data: dict) -> dict:
+    metrics = data.get("usage_metrics")
+    if isinstance(metrics, str):
+        try:
+            parsed = json.loads(metrics)
+            metrics = parsed if isinstance(parsed, dict) else None
+        except (json.JSONDecodeError, TypeError):
+            metrics = None
+    if isinstance(metrics, dict):
+        return metrics
+    events = data.get("events")
+    return summarize_usage_events(
+        events if isinstance(events, list) else [],
+        session_id=data.get("session_id"),
+    )
+
+
 def _write_row_payload(data: dict, tmp_path: str) -> None:
     """Single-row JSONL (existing format) — used by KPI scheduler."""
     scrubbed = _scrub_session_for_upload(data)
+    usage_metrics = _usage_metrics_for_row(data)
     session_row = {
         "session_id": data["session_id"],
         "user_id": data.get("user_id"),
@@ -274,7 +297,9 @@ def _write_row_payload(data: dict, tmp_path: str) -> None:
         "messages": json.dumps(scrubbed["messages"]),
         "events": json.dumps(scrubbed["events"]),
         "tools": json.dumps(scrubbed["tools"]),
+        "usage_metrics": json.dumps(_scrub(usage_metrics)),
     }
+    session_row.update(usage_metric_scalar_fields(usage_metrics))
 
     with open(tmp_path, "w") as tmp:
         json.dump(session_row, tmp)

agent/core/usage_metrics.py

@@ -0,0 +1,448 @@
+"""Pure usage/billing summaries for session trajectory analytics."""
+
+from collections import Counter, defaultdict
+from datetime import UTC, datetime, timedelta
+from math import isfinite
+from typing import Any
+
+from agent.core.cost_estimation import SPACE_PRICE_USD_PER_HOUR
+
+USAGE_METRICS_VERSION = 1
+BILLING_SCOPE_ACCOUNT_WINDOW_DELTA = "account_window_delta"
+
+_USAGE_SCALAR_KEYS = (
+    "usage_total_usd",
+    "usage_total_usd_source",
+    "usage_app_total_usd",
+    "usage_hf_billing_total_usd",
+    "usage_llm_calls",
+    "usage_total_tokens",
+    "usage_hf_job_submits",
+    "usage_hf_job_status_snapshots",
+    "usage_sandbox_creates",
+    "usage_sandbox_pairs",
+)
+
+
+def _coerce_float(value: Any) -> float:
+    if isinstance(value, bool) or value is None:
+        return 0.0
+    try:
+        parsed = float(value)
+    except (TypeError, ValueError):
+        return 0.0
+    return parsed if isfinite(parsed) else 0.0
+
+
+def _coerce_optional_float(value: Any) -> float | None:
+    if isinstance(value, bool) or value is None:
+        return None
+    try:
+        parsed = float(value)
+    except (TypeError, ValueError):
+        return None
+    return parsed if isfinite(parsed) else None
+
+
+def _coerce_int(value: Any) -> int:
+    if isinstance(value, bool) or value is None:
+        return 0
+    try:
+        return int(value)
+    except (TypeError, ValueError):
+        return 0
+
+
+def _round_usd(value: Any) -> float:
+    return round(_coerce_float(value), 6)
+
+
+def _parse_timestamp(value: Any) -> datetime | None:
+    if isinstance(value, datetime):
+        dt = value
+    elif isinstance(value, str) and value:
+        try:
+            dt = datetime.fromisoformat(value.replace("Z", "+00:00"))
+        except ValueError:
+            return None
+    else:
+        return None
+    if dt.tzinfo is None:
+        return dt.replace(tzinfo=UTC)
+    return dt.astimezone(UTC)
+
+
+def event_created_at(event: dict[str, Any]) -> datetime | None:
+    return _parse_timestamp(event.get("created_at") or event.get("timestamp"))
+
+
+def _event_data(event: dict[str, Any]) -> dict[str, Any]:
+    data = event.get("data") or {}
+    return data if isinstance(data, dict) else {}
+
+
+def _has_number(value: Any) -> bool:
+    return _coerce_optional_float(value) is not None
+
+
+def _counter_dict(counter: Counter[str]) -> dict[str, int]:
+    return dict(sorted(counter.items()))
+
+
+def _empty_app_bucket(session_id: str | None) -> dict[str, Any]:
+    return {
+        "session_id": session_id,
+        "total_usd": 0.0,
+        "inference_usd": 0.0,
+        "hf_jobs_estimated_usd": 0.0,
+        "sandbox_estimated_usd": 0.0,
+        "llm_calls": 0,
+        "hf_jobs_count": 0,
+        "sandbox_count": 0,
+        "prompt_tokens": 0,
+        "completion_tokens": 0,
+        "cache_read_tokens": 0,
+        "cache_creation_tokens": 0,
+        "total_tokens": 0,
+        "hf_jobs_billable_seconds_estimate": 0,
+        "sandbox_billable_seconds_estimate": 0,
+    }
+
+
+def _sandbox_id(event: dict[str, Any]) -> str | None:
+    sandbox_id = _event_data(event).get("sandbox_id")
+    return sandbox_id if isinstance(sandbox_id, str) and sandbox_id else None
+
+
+def _sandbox_duration_seconds(
+    create_event: dict[str, Any],
+    destroy_event: dict[str, Any],
+) -> int:
+    create_data = _event_data(create_event)
+    destroy_data = _event_data(destroy_event)
+    lifetime_s = _coerce_int(destroy_data.get("lifetime_s"))
+    if lifetime_s > 0:
+        return lifetime_s
+
+    create_at = event_created_at(create_event)
+    destroy_at = event_created_at(destroy_event)
+    if create_at is None or destroy_at is None:
+        return 0
+    create_latency_s = max(0, _coerce_int(create_data.get("create_latency_s")))
+    interval_start = create_at - timedelta(seconds=create_latency_s)
+    if destroy_at <= interval_start:
+        return 0
+    return int((destroy_at - interval_start).total_seconds())
+
+
+def summarize_sandbox_lifecycle(
+    lifecycle_events: list[tuple[int, dict[str, Any]]],
+) -> dict[str, Any]:
+    """Pair sandbox lifecycle events and estimate billed usage.
+
+    Shared by dataset usage metrics and backend usage responses so sandbox
+    pricing and create/destroy pairing semantics cannot drift.
+    """
+    ordered_events = [
+        event
+        for _, event in sorted(
+            lifecycle_events,
+            key=lambda indexed: (
+                event_created_at(indexed[1]) is None,
+                event_created_at(indexed[1]) or datetime.min.replace(tzinfo=UTC),
+                indexed[0],
+            ),
+        )
+    ]
+    active_creates: dict[str, list[dict[str, Any]]] = defaultdict(list)
+    matched_pairs = 0
+    unpaired_destroys = 0
+    estimated_usd = 0.0
+    billable_seconds = 0
+
+    for event in ordered_events:
+        event_type = event.get("event_type")
+        sandbox_id = _sandbox_id(event)
+        if sandbox_id is None:
+            continue
+        if event_type == "sandbox_create":
+            active_creates[sandbox_id].append(event)
+            continue
+        if event_type != "sandbox_destroy":
+            continue
+
+        creates = active_creates.get(sandbox_id)
+        if not creates:
+            unpaired_destroys += 1
+            continue
+
+        create_event = creates.pop()
+        if not creates:
+            active_creates.pop(sandbox_id, None)
+
+        hardware = str(_event_data(create_event).get("hardware") or "cpu-basic")
+        seconds = _sandbox_duration_seconds(create_event, event)
+        price_usd_per_hour = _coerce_float(SPACE_PRICE_USD_PER_HOUR.get(hardware))
+        matched_pairs += 1
+        if price_usd_per_hour > 0:
+            billable_seconds += seconds
+        estimated_usd += price_usd_per_hour * (seconds / 3600)
+
+    return {
+        "matched_pairs": matched_pairs,
+        "unpaired_creates": sum(len(events) for events in active_creates.values()),
+        "unpaired_destroys": unpaired_destroys,
+        "estimated_usd": _round_usd(estimated_usd),
+        "billable_seconds_estimate": billable_seconds,
+    }
+
+
+def normalize_hf_billing_snapshot(snapshot: dict[str, Any] | None) -> dict[str, Any]:
+    """Return a dataset-safe HF billing snapshot.
+
+    Only current-session window rollups are retained. Monthly account totals,
+    credit limits, and any caller-provided extra fields are intentionally
+    dropped before the snapshot can be serialized into session artifacts.
+    """
+    hf_billing = snapshot.get("hf_billing") if isinstance(snapshot, dict) else None
+    hf_billing = hf_billing if isinstance(hf_billing, dict) else {}
+    current_session = hf_billing.get("current_session")
+    current_session = current_session if isinstance(current_session, dict) else None
+
+    sanitized_current = None
+    if current_session is not None:
+        sanitized_current = {
+            "window_start": current_session.get("window_start"),
+            "window_end": current_session.get("window_end"),
+            "timezone": current_session.get("timezone"),
+            "total_usd": _round_usd(current_session.get("total_usd")),
+            "inference_providers_usd": _round_usd(
+                current_session.get("inference_providers_usd")
+            ),
+            "hf_jobs_usd": _round_usd(current_session.get("hf_jobs_usd")),
+            "inference_provider_requests": _coerce_int(
+                current_session.get("inference_provider_requests")
+            ),
+            "hf_jobs_minutes": round(
+                _coerce_float(current_session.get("hf_jobs_minutes")), 3
+            ),
+        }
+
+    available = bool(hf_billing.get("available") and sanitized_current is not None)
+    return {
+        "billing_scope": BILLING_SCOPE_ACCOUNT_WINDOW_DELTA,
+        "hf_billing": {
+            "source": str(hf_billing.get("source") or "hf_billing_usage_v2"),
+            "available": available,
+            "error": None if available else hf_billing.get("error"),
+            "current_session": sanitized_current if available else None,
+        },
+    }
+
+
+def summarize_usage_events(
+    events: list[dict[str, Any]],
+    *,
+    session_id: str | None = None,
+    hf_billing_snapshot: dict[str, Any] | None = None,
+) -> dict[str, Any]:
+    app = _empty_app_bucket(session_id)
+    llm_by_kind: Counter[str] = Counter()
+    llm_by_model: Counter[str] = Counter()
+    job_statuses: Counter[str] = Counter()
+    job_submit_flavors: Counter[str] = Counter()
+    job_status_flavors: Counter[str] = Counter()
+    sandbox_hardware: Counter[str] = Counter()
+    lifecycle_events: list[tuple[int, dict[str, Any]]] = []
+
+    event_count = 0
+    events_without_timestamp = 0
+    llm_calls_with_cost_usd = 0
+    llm_calls_with_nonzero_cost_usd = 0
+    job_submits = 0
+    job_status_snapshots = 0
+    job_snapshots_with_estimated_cost = 0
+    job_snapshots_with_nonzero_estimated_cost = 0
+    sandbox_creates = 0
+    sandbox_destroys = 0
+    turn_complete_count = 0
+    assistant_stream_end_count = 0
+
+    for index, event in enumerate(events or []):
+        if not isinstance(event, dict):
+            continue
+        event_count += 1
+        if event_created_at(event) is None:
+            events_without_timestamp += 1
+
+        event_type = event.get("event_type")
+        data = _event_data(event)
+        if event_type == "llm_call":
+            app["llm_calls"] += 1
+            if "cost_usd" in data:
+                llm_calls_with_cost_usd += 1
+            cost_usd = _coerce_float(data.get("cost_usd"))
+            if cost_usd > 0:
+                llm_calls_with_nonzero_cost_usd += 1
+            app["inference_usd"] += cost_usd
+
+            prompt_tokens = _coerce_int(data.get("prompt_tokens"))
+            completion_tokens = _coerce_int(data.get("completion_tokens"))
+            cache_read_tokens = _coerce_int(data.get("cache_read_tokens"))
+            cache_creation_tokens = _coerce_int(data.get("cache_creation_tokens"))
+            total_tokens = _coerce_int(data.get("total_tokens")) or (
+                prompt_tokens
+                + completion_tokens
+                + cache_read_tokens
+                + cache_creation_tokens
+            )
+            app["prompt_tokens"] += prompt_tokens
+            app["completion_tokens"] += completion_tokens
+            app["cache_read_tokens"] += cache_read_tokens
+            app["cache_creation_tokens"] += cache_creation_tokens
+            app["total_tokens"] += total_tokens
+            llm_by_kind[str(data.get("kind") or "unknown")] += 1
+            llm_by_model[str(data.get("model") or "unknown")] += 1
+        elif event_type == "hf_job_submit":
+            job_submits += 1
+            job_submit_flavors[str(data.get("flavor") or "unknown")] += 1
+        elif event_type == "hf_job_complete":
+            job_status_snapshots += 1
+            app["hf_jobs_count"] += 1
+            estimated_cost = _coerce_float(data.get("estimated_cost_usd"))
+            app["hf_jobs_estimated_usd"] += estimated_cost
+            app["hf_jobs_billable_seconds_estimate"] += _coerce_int(
+                data.get("billable_seconds_estimate") or data.get("wall_time_s")
+            )
+            if _has_number(data.get("estimated_cost_usd")):
+                job_snapshots_with_estimated_cost += 1
+            if estimated_cost > 0:
+                job_snapshots_with_nonzero_estimated_cost += 1
+            job_statuses[str(data.get("final_status") or "unknown")] += 1
+            job_status_flavors[str(data.get("flavor") or "unknown")] += 1
+        elif event_type == "sandbox_create":
+            sandbox_creates += 1
+            sandbox_hardware[str(data.get("hardware") or "cpu-basic")] += 1
+            lifecycle_events.append((index, event))
+        elif event_type == "sandbox_destroy":
+            sandbox_destroys += 1
+            lifecycle_events.append((index, event))
+        elif event_type == "turn_complete":
+            turn_complete_count += 1
+        elif event_type == "assistant_stream_end":
+            assistant_stream_end_count += 1
+
+    sandbox = summarize_sandbox_lifecycle(lifecycle_events)
+    app["sandbox_count"] = sandbox["matched_pairs"]
+    app["sandbox_estimated_usd"] = sandbox["estimated_usd"]
+    app["sandbox_billable_seconds_estimate"] = sandbox["billable_seconds_estimate"]
+    app["inference_usd"] = _round_usd(app["inference_usd"])
+    app["hf_jobs_estimated_usd"] = _round_usd(app["hf_jobs_estimated_usd"])
+    app["total_usd"] = _round_usd(
+        app["inference_usd"]
+        + app["hf_jobs_estimated_usd"]
+        + app["sandbox_estimated_usd"]
+    )
+
+    billing = normalize_hf_billing_snapshot(hf_billing_snapshot)
+    current_billing = billing["hf_billing"]["current_session"]
+    hf_billing_total = None
+    if billing["hf_billing"]["available"] and current_billing is not None:
+        hf_billing_total = _round_usd(current_billing.get("total_usd"))
+        usage_total = _round_usd(hf_billing_total + app["sandbox_estimated_usd"])
+        usage_total_source = "hf_billing_plus_sandbox_estimate"
+    else:
+        usage_total = app["total_usd"]
+        usage_total_source = "app_telemetry_fallback"
+
+    job_flavors = job_submit_flavors + job_status_flavors
+
+    return {
+        "version": USAGE_METRICS_VERSION,
+        "session_id": session_id,
+        "billing_scope": BILLING_SCOPE_ACCOUNT_WINDOW_DELTA,
+        "total_usd": usage_total,
+        "total_usd_source": usage_total_source,
+        "app_total_usd": app["total_usd"],
+        "hf_billing_total_usd": hf_billing_total,
+        "app_telemetry": app,
+        "hf_billing": billing["hf_billing"],
+        "llm": {
+            "calls": app["llm_calls"],
+            "calls_by_kind": _counter_dict(llm_by_kind),
+            "calls_by_model": _counter_dict(llm_by_model),
+            "prompt_tokens": app["prompt_tokens"],
+            "completion_tokens": app["completion_tokens"],
+            "cache_read_tokens": app["cache_read_tokens"],
+            "cache_creation_tokens": app["cache_creation_tokens"],
+            "total_tokens": app["total_tokens"],
+        },
+        "turns": {
+            "turn_complete_count": turn_complete_count,
+            "assistant_stream_end_count": assistant_stream_end_count,
+        },
+        "hf_jobs": {
+            "submits": job_submits,
+            "status_snapshots": job_status_snapshots,
+            "statuses": _counter_dict(job_statuses),
+            "flavors": _counter_dict(job_flavors),
+            "submit_flavors": _counter_dict(job_submit_flavors),
+            "status_snapshot_flavors": _counter_dict(job_status_flavors),
+            "estimated_usd": app["hf_jobs_estimated_usd"],
+            "billable_seconds_estimate": app["hf_jobs_billable_seconds_estimate"],
+            "snapshots_with_estimated_cost": job_snapshots_with_estimated_cost,
+            "snapshots_with_nonzero_estimated_cost": (
+                job_snapshots_with_nonzero_estimated_cost
+            ),
+        },
+        "sandboxes": {
+            "creates": sandbox_creates,
+            "destroys": sandbox_destroys,
+            "matched_pairs": sandbox["matched_pairs"],
+            "unpaired_creates": sandbox["unpaired_creates"],
+            "unpaired_destroys": sandbox["unpaired_destroys"],
+            "hardware": _counter_dict(sandbox_hardware),
+            "estimated_usd": app["sandbox_estimated_usd"],
+            "billable_seconds_estimate": app["sandbox_billable_seconds_estimate"],
+        },
+        "data_quality": {
+            "event_count": event_count,
+            "events_without_timestamp": events_without_timestamp,
+            "llm_calls_with_cost_usd": llm_calls_with_cost_usd,
+            "llm_calls_with_nonzero_cost_usd": llm_calls_with_nonzero_cost_usd,
+            "job_snapshots_with_estimated_cost": job_snapshots_with_estimated_cost,
+            "job_snapshots_missing_estimated_cost": (
+                job_status_snapshots - job_snapshots_with_estimated_cost
+            ),
+        },
+    }
+
+
+def usage_metric_scalar_fields(metrics: dict[str, Any]) -> dict[str, Any]:
+    app = metrics.get("app_telemetry") if isinstance(metrics, dict) else {}
+    llm = metrics.get("llm") if isinstance(metrics, dict) else {}
+    jobs = metrics.get("hf_jobs") if isinstance(metrics, dict) else {}
+    sandboxes = metrics.get("sandboxes") if isinstance(metrics, dict) else {}
+    values = {
+        "usage_total_usd": metrics.get("total_usd"),
+        "usage_total_usd_source": metrics.get("total_usd_source"),
+        "usage_app_total_usd": metrics.get("app_total_usd"),
+        "usage_hf_billing_total_usd": metrics.get("hf_billing_total_usd"),
+        "usage_llm_calls": app.get("llm_calls") if isinstance(app, dict) else None,
+        "usage_total_tokens": llm.get("total_tokens")
+        if isinstance(llm, dict)
+        else None,
+        "usage_hf_job_submits": (
+            jobs.get("submits") if isinstance(jobs, dict) else None
+        ),
+        "usage_hf_job_status_snapshots": (
+            jobs.get("status_snapshots") if isinstance(jobs, dict) else None
+        ),
+        "usage_sandbox_creates": (
+            sandboxes.get("creates") if isinstance(sandboxes, dict) else None
+        ),
+        "usage_sandbox_pairs": (
+            sandboxes.get("matched_pairs") if isinstance(sandboxes, dict) else None
+        ),
+    }
+    return {key: values.get(key) for key in _USAGE_SCALAR_KEYS}

backend/main.py

@@ -1,5 +1,6 @@
 """FastAPI application for HF Agent web interface."""
 
+import asyncio
 import logging
 import os
 from contextlib import asynccontextmanager
@@ -24,6 +25,23 @@ logging.basicConfig(
     format="%(asctime)s - %(name)s - %(levelname)s - %(message)s",
 )
 logger = logging.getLogger(__name__)
+SHUTDOWN_USAGE_REFRESH_CONCURRENCY = 32
+
+
+async def _flush_session_on_shutdown(sid: str, agent_session, semaphore) -> None:
+    sess = agent_session.session
+    if not sess.config.save_sessions:
+        return
+    try:
+        async with semaphore:
+            await session_manager.refresh_session_usage_metrics(
+                agent_session,
+                error_code="lifespan_billing_snapshot_error",
+            )
+            sess.save_and_upload_detached(sess.config.session_dataset_repo)
+            logger.info("Flushed session %s on shutdown", sid)
+    except Exception as e:
+        logger.warning("Failed to flush session %s: %s", sid, e)
 
 
 @asynccontextmanager
@@ -50,16 +68,16 @@ async def lifespan(app: FastAPI):
         logger.warning("KPI scheduler shutdown failed: %s", e)
 
     # Final-flush: save every still-active session so we don't lose traces on
-    # server restart. Uploads are detached subprocesses — this is fast.
+    # server restart. Billing refreshes are timeboxed and bounded; uploads are
+    # detached subprocesses.
     try:
-        for sid, agent_session in list(session_manager.sessions.items()):
-            sess = agent_session.session
-            if sess.config.save_sessions:
-                try:
-                    sess.save_and_upload_detached(sess.config.session_dataset_repo)
-                    logger.info("Flushed session %s on shutdown", sid)
-                except Exception as e:
-                    logger.warning("Failed to flush session %s: %s", sid, e)
+        semaphore = asyncio.Semaphore(SHUTDOWN_USAGE_REFRESH_CONCURRENCY)
+        await asyncio.gather(
+            *(
+                _flush_session_on_shutdown(sid, agent_session, semaphore)
+                for sid, agent_session in list(session_manager.sessions.items())
+            )
+        )
     except Exception as e:
         logger.warning("Lifespan final-flush skipped: %s", e)
     await session_manager.close()

backend/routes/agent.py

@@ -70,7 +70,7 @@ from usage import build_usage_response
 logger = logging.getLogger(__name__)
 
 router = APIRouter(prefix="/api", tags=["agent"])
-_background_teardown_tasks: set[asyncio.Task] = set()
+_background_route_tasks: set[asyncio.Task] = set()
 
 DEFAULT_OPUS_MODEL_ID = CLAUDE_OPUS_48_MODEL_ID
 DEFAULT_GPT_MODEL_ID = GPT_55_MODEL_ID
@@ -85,6 +85,38 @@ async def _reset_usage_window(session_id: str) -> dict[str, Any] | None:
     )
 
 
+async def _refresh_usage_and_upload(
+    agent_session: AgentSession,
+    *,
+    error_code: str,
+) -> None:
+    session = agent_session.session
+    try:
+        await session_manager.refresh_session_usage_metrics(
+            agent_session,
+            error_code=error_code,
+        )
+        session.save_and_upload_detached(session.config.session_dataset_repo)
+    except Exception as e:
+        logger.warning(
+            "Background usage refresh/upload failed for %s: %s",
+            agent_session.session_id,
+            e,
+        )
+
+
+def _schedule_usage_refresh_and_upload(
+    agent_session: AgentSession,
+    *,
+    error_code: str,
+) -> None:
+    task = asyncio.create_task(
+        _refresh_usage_and_upload(agent_session, error_code=error_code)
+    )
+    _background_route_tasks.add(task)
+    task.add_done_callback(_background_route_tasks.discard)
+
+
 def _available_models() -> list[dict[str, Any]]:
     models = [
         {
@@ -759,8 +791,8 @@ async def teardown_session_sandbox(
     """Best-effort sandbox teardown that preserves durable chat history."""
     await _check_session_access(session_id, user, preload_sandbox=False)
     task = asyncio.create_task(session_manager.teardown_sandbox(session_id))
-    _background_teardown_tasks.add(task)
-    task.add_done_callback(_background_teardown_tasks.discard)
+    _background_route_tasks.add(task)
+    task.add_done_callback(_background_route_tasks.discard)
     return {"status": "teardown_requested", "session_id": session_id}
 
 
@@ -910,8 +942,9 @@ async def record_pro_click(
         target=str(body.get("target") or "pro_pricing"),
     )
     if agent_session.session.config.save_sessions:
-        agent_session.session.save_and_upload_detached(
-            agent_session.session.config.session_dataset_repo
+        _schedule_usage_refresh_and_upload(
+            agent_session,
+            error_code="pro_click_billing_snapshot_error",
         )
     return {"status": "ok"}
 
@@ -1154,7 +1187,8 @@ async def submit_feedback(
     # Fire-and-forget save so feedback reaches the dataset even if the user
     # closes the tab right after clicking.
     if agent_session.session.config.save_sessions:
-        agent_session.session.save_and_upload_detached(
-            agent_session.session.config.session_dataset_repo
+        _schedule_usage_refresh_and_upload(
+            agent_session,
+            error_code="feedback_billing_snapshot_error",
         )
     return {"status": "ok"}

backend/session_manager.py

@@ -42,6 +42,7 @@ from agent.messaging.gateway import NotificationGateway
 PROJECT_ROOT = Path(__file__).parent.parent
 DEFAULT_CONFIG_PATH = str(PROJECT_ROOT / "configs" / "frontend_agent_config.json")
 USAGE_WARNING_SPEND_CACHE_TTL_SECONDS = 30.0
+USAGE_BILLING_REFRESH_TIMEOUT_SECONDS = 2.0
 
 
 # These dataclasses match agent/main.py structure
@@ -567,6 +568,72 @@ class SessionManager:
         }
         return spend, billing_source
 
+    @staticmethod
+    def _fallback_hf_billing_snapshot(error: str) -> dict[str, Any]:
+        return {
+            "billing_scope": "account_window_delta",
+            "hf_billing": {
+                "source": "hf_billing_usage_v2",
+                "available": False,
+                "error": error,
+                "current_session": None,
+            },
+        }
+
+    async def refresh_session_usage_metrics(
+        self,
+        agent_session: AgentSession,
+        *,
+        error_code: str = "billing_snapshot_error",
+        billing_timeout_s: float | None = USAGE_BILLING_REFRESH_TIMEOUT_SECONDS,
+    ) -> dict[str, Any]:
+        """Refresh the dataset usage snapshot stored on the runtime session."""
+        from agent.core.usage_metrics import (
+            normalize_hf_billing_snapshot,
+            summarize_usage_events,
+        )
+        from usage import build_hf_billing_snapshot
+
+        session = agent_session.session
+        try:
+            billing_snapshot = build_hf_billing_snapshot(
+                self,
+                hf_token=agent_session.hf_token or getattr(session, "hf_token", None),
+                session_id=agent_session.session_id,
+                timezone_name="UTC",
+            )
+            if billing_timeout_s is not None and billing_timeout_s > 0:
+                hf_billing_snapshot = await asyncio.wait_for(
+                    billing_snapshot,
+                    timeout=billing_timeout_s,
+                )
+            else:
+                hf_billing_snapshot = await billing_snapshot
+        except TimeoutError:
+            logger.debug(
+                "HF billing snapshot refresh timed out for %s after %.2fs",
+                agent_session.session_id,
+                billing_timeout_s or 0,
+            )
+            hf_billing_snapshot = self._fallback_hf_billing_snapshot(error_code)
+        except Exception as e:
+            logger.debug(
+                "HF billing snapshot refresh failed for %s: %s",
+                agent_session.session_id,
+                e,
+            )
+            hf_billing_snapshot = self._fallback_hf_billing_snapshot(error_code)
+
+        hf_billing_snapshot = normalize_hf_billing_snapshot(hf_billing_snapshot)
+        session.usage_hf_billing_snapshot = hf_billing_snapshot
+        metrics = summarize_usage_events(
+            getattr(session, "logged_events", []) or [],
+            session_id=agent_session.session_id,
+            hf_billing_snapshot=hf_billing_snapshot,
+        )
+        session.usage_metrics = metrics
+        return metrics
+
     @staticmethod
     def _runtime_session_usage_spend(agent_session: AgentSession) -> float:
         from usage import aggregate_usage_events, event_created_at
@@ -1634,6 +1701,8 @@ class SessionManager:
                             # than the idle window would otherwise be reaped the
                             # instant it completes.
                             self._touch(agent_session)
+                            if session.config.save_sessions:
+                                await self.refresh_session_usage_metrics(agent_session)
                             await self.persist_session_snapshot(agent_session)
                         if not should_continue:
                             break
@@ -1662,6 +1731,10 @@ class SessionManager:
             # Idempotent via session_id key; detached subprocess.
             if session.config.save_sessions:
                 try:
+                    await self.refresh_session_usage_metrics(
+                        agent_session,
+                        error_code="final_billing_snapshot_error",
+                    )
                     session.save_and_upload_detached(
                         session.config.session_dataset_repo
                     )

backend/usage.py

@@ -8,7 +8,7 @@ from zoneinfo import ZoneInfo, ZoneInfoNotFoundError
 
 import httpx
 
-from agent.core.cost_estimation import SPACE_PRICE_USD_PER_HOUR
+from agent.core.usage_metrics import summarize_sandbox_lifecycle
 
 USAGE_EVENT_TYPES = (
     "llm_call",
@@ -239,47 +239,6 @@ def aggregate_usage_events(
     return bucket
 
 
-def _event_sort_key(
-    indexed_event: tuple[int, dict[str, Any]],
-) -> tuple[bool, datetime, int]:
-    index, event = indexed_event
-    created_at = event_created_at(event)
-    return (
-        created_at is None,
-        created_at or datetime.min.replace(tzinfo=UTC),
-        index,
-    )
-
-
-def _sandbox_id(event: dict[str, Any]) -> str | None:
-    data = event.get("data") or {}
-    sandbox_id = data.get("sandbox_id")
-    return sandbox_id if isinstance(sandbox_id, str) and sandbox_id else None
-
-
-def _sandbox_duration_seconds(
-    create_event: dict[str, Any],
-    destroy_event: dict[str, Any],
-) -> int:
-    create_data = create_event.get("data") or {}
-    destroy_data = destroy_event.get("data") or {}
-    lifetime_s = _coerce_int(destroy_data.get("lifetime_s"))
-
-    if lifetime_s > 0:
-        # Telemetry starts the lifetime clock before create latency elapses.
-        return lifetime_s
-
-    create_at = event_created_at(create_event)
-    destroy_at = event_created_at(destroy_event)
-    if create_at is None or destroy_at is None:
-        return 0
-    create_latency_s = max(0, _coerce_int(create_data.get("create_latency_s")))
-    interval_start = create_at - timedelta(seconds=create_latency_s)
-    if destroy_at <= interval_start:
-        return 0
-    return int((destroy_at - interval_start).total_seconds())
-
-
 def _aggregate_sandbox_usage(
     events: list[dict[str, Any]],
     bucket: dict[str, Any],
@@ -289,41 +248,10 @@ def _aggregate_sandbox_usage(
         for index, event in enumerate(events)
         if event.get("event_type") in {"sandbox_create", "sandbox_destroy"}
     ]
-    ordered_events = [
-        event
-        for _, event in sorted(
-            lifecycle_events,
-            key=_event_sort_key,
-        )
-    ]
-    active_creates: dict[str, dict[str, Any]] = {}
-
-    for event in ordered_events:
-        event_type = event.get("event_type")
-        sandbox_id = _sandbox_id(event)
-        if sandbox_id is None:
-            continue
-
-        if event_type == "sandbox_create":
-            active_creates[sandbox_id] = event
-            continue
-
-        if event_type != "sandbox_destroy":
-            continue
-
-        create_event = active_creates.pop(sandbox_id, None)
-        if create_event is None:
-            continue
-
-        create_data = create_event.get("data") or {}
-        hardware = str(create_data.get("hardware") or "cpu-basic")
-        price_usd_per_hour = SPACE_PRICE_USD_PER_HOUR.get(hardware, 0.0)
-        seconds = _sandbox_duration_seconds(create_event, event)
-
-        bucket["sandbox_count"] += 1
-        if price_usd_per_hour > 0:
-            bucket["sandbox_billable_seconds_estimate"] += seconds
-        bucket["sandbox_estimated_usd"] += price_usd_per_hour * (seconds / 3600)
+    sandbox = summarize_sandbox_lifecycle(lifecycle_events)
+    bucket["sandbox_count"] += sandbox["matched_pairs"]
+    bucket["sandbox_billable_seconds_estimate"] += sandbox["billable_seconds_estimate"]
+    bucket["sandbox_estimated_usd"] += sandbox["estimated_usd"]
 
 
 def _account_bucket_from_billing_usage(
@@ -628,6 +556,69 @@ async def _build_hf_account_usage(
     return account_usage
 
 
+async def build_hf_billing_snapshot(
+    manager: Any,
+    *,
+    hf_token: str | None,
+    session_id: str | None,
+    timezone_name: str | None = None,
+    now: datetime | None = None,
+) -> dict[str, Any]:
+    """Return a dataset-safe HF billing rollup for the session window.
+
+    This intentionally omits monthly account totals and credit-limit details.
+    The snapshot is an account-window delta, not per-call attribution.
+    """
+    windows = resolve_usage_windows(timezone_name, now=now)
+    timezone = str(windows["timezone"])
+    now_utc = windows["now_utc"]
+    snapshot: dict[str, Any] = {
+        "billing_scope": "account_window_delta",
+        "hf_billing": {
+            "source": "hf_billing_usage_v2",
+            "available": False,
+            "error": None,
+            "current_session": None,
+        },
+    }
+    hf_billing = snapshot["hf_billing"]
+
+    if not hf_token:
+        hf_billing["error"] = "missing_hf_token"
+        return snapshot
+    if not session_id:
+        hf_billing["error"] = "missing_session_id"
+        return snapshot
+
+    session_start = _session_usage_window_started_at(manager, session_id)
+    if session_start is None:
+        session_start, _ = await _load_persisted_session_usage_window_metadata(
+            manager,
+            session_id,
+        )
+    if session_start is None:
+        hf_billing["error"] = "missing_session_window"
+        return snapshot
+
+    payload = await _fetch_hf_billing_usage_v2(
+        hf_token,
+        start=session_start,
+        end=now_utc,
+    )
+    if not isinstance(payload, dict):
+        hf_billing["error"] = "billing_usage_unavailable"
+        return snapshot
+
+    hf_billing["available"] = True
+    hf_billing["current_session"] = _account_bucket_from_billing_usage(
+        payload,
+        window_start=session_start,
+        window_end=now_utc,
+        timezone=timezone,
+    )
+    return snapshot
+
+
 def _event_in_window(
     event: dict[str, Any],
     *,

frontend/src/components/UsageMeter.tsx

@@ -120,7 +120,7 @@ function AccountUsageSection({
           strong
         />
         <UsageRow
-          label={useJobEstimate ? 'HF Jobs estimated' : 'HF Jobs'}
+          label="HF Jobs"
           value={formatUsd(
             useJobEstimate ? telemetry?.hf_jobs_estimated_usd : account?.hf_jobs_usd,
           )}

tests/unit/test_session_manager_persistence.py

@@ -324,6 +324,183 @@ def test_usage_spend_falls_back_when_hf_total_is_unavailable():
     assert source == "app_telemetry_session"
 
 
+@pytest.mark.asyncio
+async def test_refresh_usage_metrics_uses_hf_billing_plus_sandbox(monkeypatch):
+    manager = _manager_with_store(NoopSessionStore())
+    agent_session = _runtime_agent_session("s1", hf_token="owner-token")
+    agent_session.session.logged_events = [
+        {
+            "timestamp": "2026-06-01T12:00:00+00:00",
+            "event_type": "llm_call",
+            "data": {"cost_usd": 0.5, "total_tokens": 42},
+        },
+        {
+            "timestamp": "2026-06-01T12:05:00+00:00",
+            "event_type": "hf_job_complete",
+            "data": {"estimated_cost_usd": 1.0},
+        },
+        {
+            "timestamp": "2026-06-01T12:10:00+00:00",
+            "event_type": "sandbox_create",
+            "data": {"sandbox_id": "owner/sandbox-1", "hardware": "t4-small"},
+        },
+        {
+            "timestamp": "2026-06-01T12:40:00+00:00",
+            "event_type": "sandbox_destroy",
+            "data": {"sandbox_id": "owner/sandbox-1", "lifetime_s": 1800},
+        },
+    ]
+
+    async def fake_billing_snapshot(_manager, *, hf_token, session_id, timezone_name):
+        assert hf_token == "owner-token"
+        assert session_id == "s1"
+        assert timezone_name == "UTC"
+        return {
+            "billing_scope": "account_window_delta",
+            "hf_billing": {
+                "source": "hf_billing_usage_v2",
+                "available": True,
+                "current_session": {
+                    "window_start": "2026-06-01T12:00:00Z",
+                    "window_end": "2026-06-01T12:40:00Z",
+                    "timezone": "UTC",
+                    "total_usd": 4.0,
+                    "inference_providers_usd": 3.0,
+                    "hf_jobs_usd": 1.0,
+                    "inference_provider_requests": 6,
+                    "hf_jobs_minutes": 2.0,
+                    "access_token": "must-not-persist",
+                },
+            },
+            "month": {"total_usd": 999},
+            "inference_providers_credits": {"limit_usd": 999},
+        }
+
+    monkeypatch.setattr("usage.build_hf_billing_snapshot", fake_billing_snapshot)
+
+    metrics = await manager.refresh_session_usage_metrics(agent_session)
+
+    assert metrics["total_usd"] == 4.3
+    assert metrics["total_usd_source"] == "hf_billing_plus_sandbox_estimate"
+    assert metrics["app_total_usd"] == 1.8
+    assert metrics["hf_billing_total_usd"] == 4.0
+    assert agent_session.session.usage_metrics == metrics
+    assert agent_session.session.usage_hf_billing_snapshot == {
+        "billing_scope": "account_window_delta",
+        "hf_billing": {
+            "source": "hf_billing_usage_v2",
+            "available": True,
+            "error": None,
+            "current_session": {
+                "window_start": "2026-06-01T12:00:00Z",
+                "window_end": "2026-06-01T12:40:00Z",
+                "timezone": "UTC",
+                "total_usd": 4.0,
+                "inference_providers_usd": 3.0,
+                "hf_jobs_usd": 1.0,
+                "inference_provider_requests": 6,
+                "hf_jobs_minutes": 2.0,
+            },
+        },
+    }
+
+
+@pytest.mark.asyncio
+async def test_refresh_usage_metrics_missing_token_falls_back_to_app_telemetry():
+    manager = _manager_with_store(NoopSessionStore())
+    agent_session = _runtime_agent_session("s1", hf_token=None)
+    agent_session.session.hf_token = None
+    agent_session.session.logged_events = [
+        {
+            "timestamp": "2026-06-01T12:00:00+00:00",
+            "event_type": "llm_call",
+            "data": {"cost_usd": 2.0, "total_tokens": 10},
+        }
+    ]
+
+    metrics = await manager.refresh_session_usage_metrics(agent_session)
+
+    assert metrics["total_usd"] == 2.0
+    assert metrics["total_usd_source"] == "app_telemetry_fallback"
+    assert metrics["hf_billing"] == {
+        "source": "hf_billing_usage_v2",
+        "available": False,
+        "error": "missing_hf_token",
+        "current_session": None,
+    }
+
+
+@pytest.mark.asyncio
+async def test_refresh_usage_metrics_failure_records_error_code(monkeypatch):
+    manager = _manager_with_store(NoopSessionStore())
+    agent_session = _runtime_agent_session("s1", hf_token="owner-token")
+    agent_session.session.logged_events = [
+        {
+            "timestamp": "2026-06-01T12:00:00+00:00",
+            "event_type": "llm_call",
+            "data": {"cost_usd": 2.0, "total_tokens": 10},
+        }
+    ]
+
+    async def fail_billing_snapshot(*_args, **_kwargs):
+        raise RuntimeError("boom")
+
+    monkeypatch.setattr("usage.build_hf_billing_snapshot", fail_billing_snapshot)
+
+    metrics = await manager.refresh_session_usage_metrics(
+        agent_session,
+        error_code="unit_billing_error",
+    )
+
+    assert metrics["total_usd"] == 2.0
+    assert metrics["total_usd_source"] == "app_telemetry_fallback"
+    assert metrics["hf_billing"] == {
+        "source": "hf_billing_usage_v2",
+        "available": False,
+        "error": "unit_billing_error",
+        "current_session": None,
+    }
+
+
+@pytest.mark.asyncio
+async def test_refresh_usage_metrics_timeout_records_error_code(monkeypatch):
+    manager = _manager_with_store(NoopSessionStore())
+    agent_session = _runtime_agent_session("s1", hf_token="owner-token")
+    agent_session.session.logged_events = [
+        {
+            "timestamp": "2026-06-01T12:00:00+00:00",
+            "event_type": "llm_call",
+            "data": {"cost_usd": 2.0, "total_tokens": 10},
+        }
+    ]
+
+    async def slow_billing_snapshot(*_args, **_kwargs):
+        await asyncio.sleep(0.05)
+        return {
+            "hf_billing": {
+                "available": True,
+                "current_session": {"total_usd": 999},
+            }
+        }
+
+    monkeypatch.setattr("usage.build_hf_billing_snapshot", slow_billing_snapshot)
+
+    metrics = await manager.refresh_session_usage_metrics(
+        agent_session,
+        error_code="unit_billing_timeout",
+        billing_timeout_s=0.001,
+    )
+
+    assert metrics["total_usd"] == 2.0
+    assert metrics["total_usd_source"] == "app_telemetry_fallback"
+    assert metrics["hf_billing"] == {
+        "source": "hf_billing_usage_v2",
+        "available": False,
+        "error": "unit_billing_timeout",
+        "current_session": None,
+    }
+
+
 @pytest.mark.asyncio
 async def test_usage_threshold_checker_creates_synthetic_pending_approval(monkeypatch):
     manager = _manager_with_store(NoopSessionStore())

tests/unit/test_session_uploader.py

@@ -158,6 +158,67 @@ def test_row_payload_scrubs_messages_events_and_tools(tmp_path):
     assert "GITHUB_TOKEN=[REDACTED]" in payload
 
 
+def test_row_payload_includes_usage_scalars_and_parseable_metrics(tmp_path):
+    tmp_file = tmp_path / "row.jsonl"
+    data = {
+        "session_id": "session-123",
+        "user_id": "lewtun",
+        "session_start_time": "2026-01-01T00:00:00",
+        "session_end_time": "2026-01-01T00:30:00",
+        "model_name": "anthropic/claude-opus-4.8:fal-ai",
+        "total_cost_usd": 0.01,
+        "messages": [{"role": "user", "content": "hello"}],
+        "events": [
+            {
+                "timestamp": "2026-01-01T00:00:01+00:00",
+                "event_type": "llm_call",
+                "data": {"cost_usd": 0.01, "total_tokens": 42},
+            },
+            {
+                "timestamp": "2026-01-01T00:00:02+00:00",
+                "event_type": "hf_job_submit",
+                "data": {"flavor": "cpu-basic"},
+            },
+            {
+                "timestamp": "2026-01-01T00:00:03+00:00",
+                "event_type": "turn_complete",
+                "data": {},
+            },
+        ],
+        "tools": [{"name": "bash"}],
+    }
+
+    _write_row_payload(data, str(tmp_file))
+
+    row = json.loads(tmp_file.read_text())
+    assert row["session_id"] == "session-123"
+    assert row["user_id"] == "lewtun"
+    assert row["session_start_time"] == "2026-01-01T00:00:00"
+    assert row["session_end_time"] == "2026-01-01T00:30:00"
+    assert row["model_name"] == "anthropic/claude-opus-4.8:fal-ai"
+    assert row["total_cost_usd"] == 0.01
+    assert json.loads(row["messages"]) == data["messages"]
+    assert json.loads(row["events"]) == data["events"]
+    assert json.loads(row["tools"]) == data["tools"]
+
+    metrics = json.loads(row["usage_metrics"])
+    assert metrics["version"] == 1
+    assert metrics["llm"]["calls"] == 1
+    assert metrics["llm"]["total_tokens"] == 42
+    assert metrics["hf_jobs"]["submits"] == 1
+    assert metrics["turns"]["turn_complete_count"] == 1
+    assert row["usage_total_usd"] == 0.01
+    assert row["usage_total_usd_source"] == "app_telemetry_fallback"
+    assert row["usage_app_total_usd"] == 0.01
+    assert row["usage_hf_billing_total_usd"] is None
+    assert row["usage_llm_calls"] == 1
+    assert row["usage_total_tokens"] == 42
+    assert row["usage_hf_job_submits"] == 1
+    assert row["usage_hf_job_status_snapshots"] == 0
+    assert row["usage_sandbox_creates"] == 0
+    assert row["usage_sandbox_pairs"] == 0
+
+
 def test_claude_code_payload_scrubs_messages_before_conversion(tmp_path):
     tmp_file = tmp_path / "claude_code.jsonl"
     data = {
@@ -202,3 +263,40 @@ def test_claude_code_payload_scrubs_messages_before_conversion(tmp_path):
     assert "[REDACTED_HF_TOKEN]" in payload
     assert "[REDACTED_PROVIDER_API_KEY]" in payload
     assert "GITHUB_TOKEN=[REDACTED]" in payload
+
+
+def test_claude_code_payload_ignores_usage_metrics(tmp_path):
+    base = {
+        "session_id": "session-123",
+        "model_name": "anthropic/claude-opus-4.8:fal-ai",
+        "session_start_time": "2026-01-01T00:00:00",
+        "messages": [
+            {
+                "role": "user",
+                "content": "hello",
+                "timestamp": "2026-01-01T00:00:01",
+            },
+            {
+                "role": "assistant",
+                "content": "hi",
+                "timestamp": "2026-01-01T00:00:02",
+            },
+        ],
+    }
+    without_metrics = tmp_path / "without.jsonl"
+    with_metrics = tmp_path / "with.jsonl"
+
+    _write_claude_code_payload(base, str(without_metrics))
+    _write_claude_code_payload(
+        {
+            **base,
+            "usage_metrics": {
+                "version": 1,
+                "total_usd": 123,
+                "hf_billing": {"available": True},
+            },
+        },
+        str(with_metrics),
+    )
+
+    assert with_metrics.read_text() == without_metrics.read_text()

tests/unit/test_usage.py

@@ -14,10 +14,15 @@ from usage import (  # noqa: E402
     _account_bucket_from_billing_usage,
     _session_bucket_from_inference_session_usage,
     aggregate_usage_events,
+    build_hf_billing_snapshot,
     build_usage_response,
     resolve_usage_windows,
 )
 from agent.core import session_persistence  # noqa: E402
+from agent.core.usage_metrics import (  # noqa: E402
+    summarize_usage_events,
+    usage_metric_scalar_fields,
+)
 
 BILLING_SESSION_ID = "00000000-0000-4000-8000-000000000001"
 
@@ -173,6 +178,42 @@ def test_aggregate_usage_events_falls_back_to_sandbox_timestamps():
     assert usage["total_usd"] == 0.3
 
 
+def test_sandbox_lifecycle_pairing_is_shared_for_duplicate_creates():
+    events = [
+        _event(
+            "sandbox_create",
+            {"sandbox_id": "alice/sandbox-reused", "hardware": "t4-small"},
+            created_at="2026-06-01T12:00:00+00:00",
+        ),
+        _event(
+            "sandbox_create",
+            {"sandbox_id": "alice/sandbox-reused", "hardware": "cpu-basic"},
+            created_at="2026-06-01T12:05:00+00:00",
+        ),
+        _event(
+            "sandbox_destroy",
+            {"sandbox_id": "alice/sandbox-reused", "lifetime_s": 300},
+            created_at="2026-06-01T12:10:00+00:00",
+        ),
+        _event(
+            "sandbox_destroy",
+            {"sandbox_id": "alice/sandbox-reused", "lifetime_s": 1200},
+            created_at="2026-06-01T12:20:00+00:00",
+        ),
+    ]
+
+    usage = aggregate_usage_events(events, session_id="s1")
+    metrics = summarize_usage_events(events, session_id="s1")
+
+    assert usage["sandbox_count"] == 2
+    assert usage["sandbox_billable_seconds_estimate"] == 1200
+    assert usage["sandbox_estimated_usd"] == 0.2
+    assert metrics["sandboxes"]["matched_pairs"] == usage["sandbox_count"]
+    assert metrics["sandboxes"]["unpaired_creates"] == 0
+    assert metrics["sandboxes"]["unpaired_destroys"] == 0
+    assert metrics["sandboxes"]["estimated_usd"] == usage["sandbox_estimated_usd"]
+
+
 def test_usage_event_type_allowlists_include_sandbox_lifecycle():
     assert set(USAGE_EVENT_TYPES) >= {"sandbox_create", "sandbox_destroy"}
     assert set(session_persistence.USAGE_EVENT_TYPES) >= {
@@ -181,6 +222,194 @@ def test_usage_event_type_allowlists_include_sandbox_lifecycle():
     }
 
 
+def test_summarize_usage_events_aggregates_dataset_analytics():
+    events = [
+        _event(
+            "llm_call",
+            {
+                "model": "model-a",
+                "kind": "main",
+                "cost_usd": 0,
+                "prompt_tokens": 10,
+                "completion_tokens": 5,
+                "cache_read_tokens": 2,
+            },
+        ),
+        _event(
+            "llm_call",
+            {
+                "model": "model-b",
+                "kind": "research",
+                "cost_usd": 0.125,
+                "prompt_tokens": 20,
+                "completion_tokens": 10,
+                "cache_creation_tokens": 3,
+                "total_tokens": 40,
+            },
+        ),
+        _event("hf_job_submit", {"flavor": "a10g-small"}),
+        _event(
+            "hf_job_complete",
+            {
+                "flavor": "a10g-small",
+                "final_status": "succeeded",
+                "estimated_cost_usd": 0.5,
+                "billable_seconds_estimate": 600,
+            },
+        ),
+        _event(
+            "hf_job_complete",
+            {
+                "flavor": "cpu-basic",
+                "final_status": "failed",
+                "wall_time_s": 30,
+            },
+        ),
+        _event(
+            "sandbox_create",
+            {"sandbox_id": "alice/sandbox-1", "hardware": "t4-small"},
+            created_at="2026-06-01T12:00:00+00:00",
+        ),
+        _event(
+            "sandbox_destroy",
+            {"sandbox_id": "alice/sandbox-1", "lifetime_s": 1800},
+            created_at="2026-06-01T12:30:00+00:00",
+        ),
+        _event(
+            "sandbox_create",
+            {"sandbox_id": "alice/sandbox-2", "hardware": "a100-large"},
+            created_at="2026-06-01T13:00:00+00:00",
+        ),
+        _event(
+            "sandbox_destroy",
+            {"sandbox_id": "alice/sandbox-missing", "lifetime_s": 60},
+            created_at="2026-06-01T13:05:00+00:00",
+        ),
+        _event("turn_complete"),
+        _event("assistant_stream_end"),
+        {"event_type": "debug", "data": {}},
+    ]
+
+    metrics = summarize_usage_events(events, session_id="s1")
+
+    assert metrics["version"] == 1
+    assert metrics["total_usd_source"] == "app_telemetry_fallback"
+    assert metrics["total_usd"] == 0.925
+    assert metrics["llm"] == {
+        "calls": 2,
+        "calls_by_kind": {"main": 1, "research": 1},
+        "calls_by_model": {"model-a": 1, "model-b": 1},
+        "prompt_tokens": 30,
+        "completion_tokens": 15,
+        "cache_read_tokens": 2,
+        "cache_creation_tokens": 3,
+        "total_tokens": 57,
+    }
+    assert metrics["turns"] == {
+        "turn_complete_count": 1,
+        "assistant_stream_end_count": 1,
+    }
+    assert metrics["hf_jobs"]["submits"] == 1
+    assert metrics["hf_jobs"]["status_snapshots"] == 2
+    assert metrics["hf_jobs"]["statuses"] == {"failed": 1, "succeeded": 1}
+    assert metrics["hf_jobs"]["flavors"] == {
+        "a10g-small": 2,
+        "cpu-basic": 1,
+    }
+    assert metrics["hf_jobs"]["estimated_usd"] == 0.5
+    assert metrics["hf_jobs"]["billable_seconds_estimate"] == 630
+    assert metrics["sandboxes"]["creates"] == 2
+    assert metrics["sandboxes"]["destroys"] == 2
+    assert metrics["sandboxes"]["matched_pairs"] == 1
+    assert metrics["sandboxes"]["unpaired_creates"] == 1
+    assert metrics["sandboxes"]["unpaired_destroys"] == 1
+    assert metrics["sandboxes"]["hardware"] == {"a100-large": 1, "t4-small": 1}
+    assert metrics["sandboxes"]["estimated_usd"] == 0.3
+    assert metrics["sandboxes"]["billable_seconds_estimate"] == 1800
+    assert metrics["data_quality"] == {
+        "event_count": 12,
+        "events_without_timestamp": 1,
+        "llm_calls_with_cost_usd": 2,
+        "llm_calls_with_nonzero_cost_usd": 1,
+        "job_snapshots_with_estimated_cost": 1,
+        "job_snapshots_missing_estimated_cost": 1,
+    }
+
+    assert usage_metric_scalar_fields(metrics) == {
+        "usage_total_usd": 0.925,
+        "usage_total_usd_source": "app_telemetry_fallback",
+        "usage_app_total_usd": 0.925,
+        "usage_hf_billing_total_usd": None,
+        "usage_llm_calls": 2,
+        "usage_total_tokens": 57,
+        "usage_hf_job_submits": 1,
+        "usage_hf_job_status_snapshots": 2,
+        "usage_sandbox_creates": 2,
+        "usage_sandbox_pairs": 1,
+    }
+
+
+def test_summarize_usage_events_uses_hf_billing_plus_sandbox_when_available():
+    events = [
+        _event("llm_call", {"cost_usd": 99.0, "total_tokens": 10}),
+        _event("hf_job_complete", {"estimated_cost_usd": 99.0}),
+        _event(
+            "sandbox_create",
+            {"sandbox_id": "alice/sandbox-1", "hardware": "t4-small"},
+            created_at="2026-06-01T12:00:00+00:00",
+        ),
+        _event(
+            "sandbox_destroy",
+            {"sandbox_id": "alice/sandbox-1", "lifetime_s": 1800},
+            created_at="2026-06-01T12:30:00+00:00",
+        ),
+    ]
+
+    metrics = summarize_usage_events(
+        events,
+        session_id="s1",
+        hf_billing_snapshot={
+            "hf_billing": {
+                "source": "hf_billing_usage_v2",
+                "available": True,
+                "current_session": {
+                    "window_start": "2026-06-01T12:00:00Z",
+                    "window_end": "2026-06-01T12:30:00Z",
+                    "timezone": "UTC",
+                    "total_usd": 1.25,
+                    "inference_providers_usd": 1.0,
+                    "hf_jobs_usd": 0.25,
+                    "inference_provider_requests": 3,
+                    "hf_jobs_minutes": 1.5,
+                    "unexpected": "dropped",
+                },
+            },
+            "month": {"total_usd": 999},
+            "inference_providers_credits": {"limit_usd": 999},
+        },
+    )
+
+    assert metrics["total_usd"] == 1.55
+    assert metrics["total_usd_source"] == "hf_billing_plus_sandbox_estimate"
+    assert metrics["app_total_usd"] == 198.3
+    assert metrics["hf_billing_total_usd"] == 1.25
+    assert metrics["hf_billing"] == {
+        "source": "hf_billing_usage_v2",
+        "available": True,
+        "error": None,
+        "current_session": {
+            "window_start": "2026-06-01T12:00:00Z",
+            "window_end": "2026-06-01T12:30:00Z",
+            "timezone": "UTC",
+            "total_usd": 1.25,
+            "inference_providers_usd": 1.0,
+            "hf_jobs_usd": 0.25,
+            "inference_provider_requests": 3,
+            "hf_jobs_minutes": 1.5,
+        },
+    }
+
+
 def test_account_bucket_from_hf_billing_usage_v2():
     usage = _account_bucket_from_billing_usage(
         {
@@ -317,6 +546,69 @@ def _agent_session(session_id, user_id, events):
     )
 
 
+@pytest.mark.asyncio
+async def test_hf_billing_snapshot_uses_session_window_without_account_totals(
+    monkeypatch,
+):
+    usage_window_started_at = datetime(2026, 6, 5, 12, 30, tzinfo=UTC)
+    manager = _Manager(
+        {
+            "s1": SimpleNamespace(
+                session_id="s1",
+                user_id="owner",
+                created_at=datetime(2026, 6, 5, 12, 0, tzinfo=UTC),
+                usage_window_started_at=usage_window_started_at,
+                session=SimpleNamespace(logged_events=[]),
+            )
+        }
+    )
+    calls = []
+
+    async def fake_usage_v2(_token, *, start, end):
+        calls.append((start, end))
+        return {
+            "usage": {
+                "inferenceProviders": {
+                    "usedNanoUsd": 1_500_000_000,
+                    "includedNanoUsd": 2_000_000_000,
+                    "limitNanoUsd": 5_000_000_000,
+                    "numRequests": 4,
+                },
+                "jobs": {"usedMicroUsd": 250_000, "totalMinutes": 3.5},
+            }
+        }
+
+    monkeypatch.setattr("usage._fetch_hf_billing_usage_v2", fake_usage_v2)
+
+    snapshot = await build_hf_billing_snapshot(
+        manager,
+        hf_token="hf_fake",
+        session_id="s1",
+        timezone_name="UTC",
+        now=datetime(2026, 6, 5, 13, 0, tzinfo=UTC),
+    )
+
+    assert calls == [(usage_window_started_at, datetime(2026, 6, 5, 13, 0, tzinfo=UTC))]
+    assert snapshot == {
+        "billing_scope": "account_window_delta",
+        "hf_billing": {
+            "source": "hf_billing_usage_v2",
+            "available": True,
+            "error": None,
+            "current_session": {
+                "window_start": "2026-06-05T12:30:00Z",
+                "window_end": "2026-06-05T13:00:00Z",
+                "timezone": "UTC",
+                "total_usd": 1.75,
+                "inference_providers_usd": 1.5,
+                "hf_jobs_usd": 0.25,
+                "inference_provider_requests": 4,
+                "hf_jobs_minutes": 3.5,
+            },
+        },
+    }
+
+
 @pytest.mark.asyncio
 async def test_usage_response_omits_app_rollups_without_session():
     manager = _Manager(