fix: billed-cost reconciliation + no-pause speed keepalive

scripts/analyze_spread.py

@@ -122,6 +122,10 @@ def main() -> None:
         delta = [round(a - b, 4) for a, b in zip(traj, base_slice)]
         ticks_above = sum(1 for x in delta if x > 0)
         c = m["summary"]["cost_snapshot"]
+        # Billed ground truth (run_scenario reconciles via OpenRouter's
+        # generation API) — same schema we hand-patched into the v0.3.0
+        # leaderboard from the dashboard. Estimates stay for comparison.
+        billed = m["summary"].get("billed_cost")
         rows.append({
             "model": m["summary"]["model"],
             "name": name,
@@ -140,6 +144,10 @@ def main() -> None:
             "avg_confidence": m["deliberation"]["avg_confidence"],
             "wall_min": round(c["wall_time_s"] / 60, 1),
             "est_cost_usd": round(c["estimated_cost_usd"], 2),
+            "real_cost_usd": (
+                round(billed["billed_cost_usd"], 3) if billed else None
+            ),
+            "cost_source": billed["source"] if billed else None,
             "trajectory": traj,
             "days": days,
             "baseline_slice": [round(x, 4) for x in base_slice],

scripts/run_benchmark.py

@@ -6,6 +6,7 @@
 import asyncio
 import json
 import logging
+import os
 import random
 import time
 from pathlib import Path
@@ -35,7 +36,7 @@
 from rle.scoring.composite import CompositeScorer
 from rle.scoring.delta import PairedResult
 from rle.scoring.recorder import TimeSeriesRecorder
-from rle.tracking.cost_tracker import CostTracker, create_cost_tracker
+from rle.tracking.cost_tracker import CostTracker, create_cost_tracker, fetch_billed_costs
 from rle.tracking.event_log import EventLog
 from rle.tracking.hf_logger import HFLogger
 from rle.tracking.history import append_history, get_run_dir, update_baseline
@@ -692,6 +693,19 @@ async def main(args: argparse.Namespace) -> None:
         else:
             _print_leaderboard(results, model=args.model)
 
+        # Reconcile estimates against OpenRouter's billed ground truth
+        # (token-count estimates diverged up to 4x on the v0.3.0 spread).
+        billed_report = None
+        effective_base_url = args.base_url or config.provider_base_url or ""
+        openai_key = os.environ.get("OPENAI_API_KEY", "")
+        generation_ids = cost_tracker.generation_ids
+        if "openrouter.ai" in effective_base_url and openai_key and generation_ids:
+            print(
+                f"\nReconciling billed cost for {len(generation_ids)} "
+                "generations against OpenRouter...",
+            )
+            billed_report = await fetch_billed_costs(generation_ids, openai_key)
+
         # Build enriched summary with metadata
         metadata = collect_metadata(random_seed=args.seed)
         summary: dict[str, Any] = {
@@ -708,6 +722,8 @@ async def main(args: argparse.Namespace) -> None:
             "scenarios": results,
             "cost_snapshot": cost_tracker.snapshot().model_dump(),
         }
+        if billed_report:
+            summary["billed_cost"] = billed_report.model_dump()
         if event_log:
             summary["event_summary"] = event_log.summary().model_dump()
         if is_paired and paired_results:

scripts/run_scenario.py

@@ -6,6 +6,7 @@
 import asyncio
 import json
 import logging
+import os
 import random
 import sys
 from pathlib import Path
@@ -31,7 +32,7 @@
 from rle.scenarios.loader import list_scenarios, load_scenario
 from rle.scoring.composite import CompositeScorer
 from rle.scoring.recorder import TimeSeriesRecorder
-from rle.tracking.cost_tracker import create_cost_tracker
+from rle.tracking.cost_tracker import create_cost_tracker, fetch_billed_costs
 from rle.tracking.event_log import EventLog
 from rle.tracking.history import append_history
 from rle.tracking.metadata import collect_metadata
@@ -78,6 +79,7 @@ def _build_run_summary(  # noqa: PLR0913
     ticks_run: int,
     cost_snapshot_dict: dict[str, object],
     event_summary_dict: dict[str, object] | None,
+    billed_cost_dict: dict[str, object] | None = None,
 ) -> dict[str, object]:
     """Compose the per-scenario summary JSON (metadata + config + result)."""
     summary: dict[str, object] = {
@@ -98,6 +100,8 @@ def _build_run_summary(  # noqa: PLR0913
         "ticks_run": ticks_run,
         "cost_snapshot": cost_snapshot_dict,
     }
+    if billed_cost_dict is not None:
+        summary["billed_cost"] = billed_cost_dict
     if event_summary_dict is not None:
         summary["event_summary"] = event_summary_dict
     return summary
@@ -316,6 +320,20 @@ async def main(args: argparse.Namespace) -> None:
     # Output
     _print_results(loop, recorder)
 
+    # Reconcile estimates against OpenRouter's billed ground truth — the
+    # token-count estimator diverged up to 4x in both directions on the
+    # v0.3.0 spread (thinking-model usage shapes, caching discounts).
+    billed_report = None
+    effective_base_url = args.base_url or config.provider_base_url or ""
+    openai_key = os.environ.get("OPENAI_API_KEY", "")
+    generation_ids = cost_tracker.generation_ids
+    if "openrouter.ai" in effective_base_url and openai_key and generation_ids:
+        print(
+            f"\nReconciling billed cost for {len(generation_ids)} generations "
+            "against OpenRouter...",
+        )
+        billed_report = await fetch_billed_costs(generation_ids, openai_key)
+
     if args.output:
         output_dir = Path(args.output)
         output_dir.mkdir(parents=True, exist_ok=True)
@@ -350,6 +368,9 @@ async def main(args: argparse.Namespace) -> None:
             event_summary_dict=(
                 event_log.summary().model_dump() if event_log else None
             ),
+            billed_cost_dict=(
+                billed_report.model_dump() if billed_report else None
+            ),
         )
         summary_path = output_dir / f"{scenario_path.stem}_summary.json"
         summary_path.write_text(json.dumps(summary, indent=2, default=str))
@@ -382,6 +403,16 @@ async def main(args: argparse.Namespace) -> None:
             f"| Est. cost: ${snap.estimated_cost_usd:.4f} "
             f"| Wall time: {snap.wall_time_s:.1f}s",
         )
+    if billed_report:
+        unbilled = (
+            f" ({billed_report.missing_generations} unbilled — lower bound)"
+            if billed_report.missing_generations else ""
+        )
+        print(
+            f"Billed cost (OpenRouter ground truth): "
+            f"${billed_report.billed_cost_usd:.4f} over "
+            f"{billed_report.billed_generations} generations{unbilled}",
+        )
 
     if event_log:
         event_log.close()

src/rle/agents/base_role.py

@@ -180,6 +180,10 @@ def __init__(
         self._pending_events: list[RimAPIEvent] = []
         self._last_usage: dict[str, int] | None = None
         self._last_raw_output: str | None = None
+        # Provider generation IDs since the last drain — one per completion
+        # call (parse retries included), consumed by the game loop for
+        # billed-cost reconciliation against OpenRouter.
+        self._generation_ids: list[str] = []
         self._weave_op: Any = None  # Set via enable_weave() for LLM call tracing
 
     def set_provider_kwargs(self, **kwargs: Any) -> None:
@@ -259,8 +263,37 @@ def _record_completion(self, result: CompletionResult) -> CompletionResult:
             if reasoning:
                 usage["reasoning_tokens"] = reasoning
         self._last_usage = usage
+        gen_id = self._extract_generation_id(getattr(result, "raw_response", None))
+        if gen_id is not None:
+            self._generation_ids.append(gen_id)
         return result
 
+    def drain_generation_ids(self) -> list[str]:
+        """Return and clear generation IDs accumulated since the last drain.
+
+        Captures EVERY provider call — including parse retries and
+        deliberations that later failed to parse — because those bill tokens
+        too. The game loop drains this each tick into the cost tracker.
+        """
+        ids, self._generation_ids = self._generation_ids, []
+        return ids
+
+    @staticmethod
+    def _extract_generation_id(raw_response: Any) -> str | None:
+        """Pull the provider's generation/completion ID from the raw response.
+
+        OpenRouter's chat completion ``id`` (``gen-...``) keys its
+        ``/api/v1/generation`` billing endpoint. Handles both the OpenAI SDK
+        object shape and the plain dict shape. None when absent.
+        """
+        if raw_response is None:
+            return None
+        if isinstance(raw_response, dict):
+            gen_id = raw_response.get("id")
+        else:
+            gen_id = getattr(raw_response, "id", None)
+        return gen_id if isinstance(gen_id, str) and gen_id else None
+
     @staticmethod
     def _extract_reasoning_tokens(raw_response: Any) -> int:
         """Pull reasoning-token count from a provider's raw usage payload.

src/rle/orchestration/game_loop.py

@@ -3,6 +3,7 @@
 from __future__ import annotations
 
 import asyncio
+import contextlib
 import json
 import logging
 import time as _time
@@ -86,6 +87,7 @@ def __init__(
         screenshots_enabled: bool = False,
         auto_dismiss_dialogs: bool = True,
         camera_director: CameraDirector | None = None,
+        speed_keepalive_s: float = 10.0,
     ) -> None:
         self._config = config
         self._client = client
@@ -129,6 +131,7 @@ def __init__(
         self._screenshots_enabled = screenshots_enabled
         self._auto_dismiss_dialogs = auto_dismiss_dialogs
         self._camera_director = camera_director
+        self._speed_keepalive_s = speed_keepalive_s
 
         # Hub-spoke communication — agents read messages from their spokes
         self._hub = CentralPost(max_agents=len(agents))
@@ -619,6 +622,15 @@ async def run_tick(self) -> TickResult:
                         },
                     )
 
+            # Capture generation IDs from every provider call this tick —
+            # parse retries and failed deliberations bill tokens too — for
+            # billed-cost reconciliation against OpenRouter (the token-count
+            # estimator diverged up to 4x on the v0.3.0 spread).
+            if self._cost_tracker:
+                for any_agent in self._agents:
+                    for gen_id in any_agent.drain_generation_ids():
+                        self._cost_tracker.record_generation_id(gen_id)
+
             # Resolve conflicts
             resolved, resolver_stats = self._resolver.resolve(plans, state)
             self._metric_context.conflicts_total += resolver_stats.conflicts_total
@@ -745,6 +757,24 @@ async def run_tick(self) -> TickResult:
 
         return result
 
+    async def _speed_keepalive(self) -> None:
+        """Re-assert game speed periodically (no-pause mode only).
+
+        RimWorld force-pauses on threat letters (mad animal, raid). With
+        --no-pause the loop previously re-asserted speed only at tick
+        boundaries, so a slow model left the game frozen for its entire
+        deliberation window (minutes on kimi/qwen in the v0.3.0 spread) —
+        frozen footage and wildly non-uniform game-time per tick. Setting
+        the speed is idempotent, so this just fires every few seconds for
+        the whole run. Cancelled by run().
+        """
+        while True:
+            await asyncio.sleep(self._speed_keepalive_s)
+            try:
+                await self._client.unpause_game()
+            except Exception:
+                logger.debug("Speed keepalive failed", exc_info=True)
+
     async def run(self, max_ticks: int | None = None) -> list[TickResult]:
         """Run the game loop for N ticks or until stopped."""
         self._running = True
@@ -756,26 +786,41 @@ async def run(self, max_ticks: int | None = None) -> list[TickResult]:
                 "the colony-name dialog and dev debug log to need manual "
                 "dismissal.",
             )
+        keepalive: asyncio.Task[None] | None = None
         if self._no_pause:
             await self._client.unpause_game()
+            if self._speed_keepalive_s > 0:
+                keepalive = asyncio.create_task(self._speed_keepalive())
         tick_count = 0
-        while self._running:
-            result = await self.run_tick()
-            tick_count += 1
-            score_str = ""
-            if result.score:
-                score_str = f" | score={result.score.composite:.3f}"
-            logger.info(
-                "Tick %d (day %d): %d actions, %d executed%s",
-                tick_count,
-                result.day,
-                result.execution.total,
-                result.execution.executed,
-                score_str,
-            )
-            if max_ticks and tick_count >= max_ticks:
-                break
-            await asyncio.sleep(self._config.tick_interval)
+        try:
+            while self._running:
+                result = await self.run_tick()
+                tick_count += 1
+                score_str = ""
+                if result.score:
+                    score_str = f" | score={result.score.composite:.3f}"
+                logger.info(
+                    "Tick %d (day %d): %d actions, %d executed%s",
+                    tick_count,
+                    result.day,
+                    result.execution.total,
+                    result.execution.executed,
+                    score_str,
+                )
+                if max_ticks and tick_count >= max_ticks:
+                    break
+                await asyncio.sleep(self._config.tick_interval)
+        finally:
+            if keepalive is not None:
+                keepalive.cancel()
+                with contextlib.suppress(asyncio.CancelledError):
+                    await keepalive
+            # Final drain: a deliberation that timed out on the last tick may
+            # have completed in its worker thread after the per-tick drain.
+            if self._cost_tracker:
+                for agent in self._agents:
+                    for gen_id in agent.drain_generation_ids():
+                        self._cost_tracker.record_generation_id(gen_id)
         return self._tick_results
 
     def stop(self) -> None: