Add OpenRouter cost history backend

api/providers.py

@@ -1441,6 +1441,255 @@ def _provider_is_oauth(provider_id: str) -> bool:
     return provider_id in _OAUTH_PROVIDERS
 
 
+# ── OpenRouter cost-history snapshot helpers (#692) ──────────────────────────
+
+_COST_SNAPSHOTS_DIR_NAME = "cost-snapshots"
+_COST_SNAPSHOT_MAX_DAYS = 365  # hard cap to prevent unbounded growth
+
+
+def _cost_snapshots_dir() -> Path:
+    """Return the directory for cost-snapshot JSON files.
+
+    Uses the Hermes home directory (profile-aware) so snapshots are
+    isolated per profile, matching the existing STATE_DIR convention.
+    """
+    return _get_hermes_home() / _COST_SNAPSHOTS_DIR_NAME
+
+
+def _fetch_openrouter_key_usage(api_key: str) -> dict[str, Any] | None:
+    """Fetch current usage/limit from the OpenRouter ``/auth/key`` endpoint.
+
+    Returns a dict with ``usage``, ``limit``, ``label`` on success, or
+    ``None`` on any failure.  Never raises; callers handle the None case.
+    """
+    req = urllib.request.Request(
+        _OPENROUTER_KEY_URL,
+        headers={
+            "Authorization": f"Bearer {api_key}",
+            "Accept": "application/json",
+        },
+    )
+    try:
+        with urllib.request.urlopen(req, timeout=_PROVIDER_QUOTA_TIMEOUT_SECONDS) as resp:
+            raw = resp.read()
+        payload = json.loads(raw.decode("utf-8") if isinstance(raw, (bytes, bytearray)) else raw)
+        sanitized = _sanitize_openrouter_quota(payload)
+        label = None
+        if isinstance(payload, dict):
+            data = payload.get("data", payload)
+            if isinstance(data, dict):
+                label = str(data.get("label") or "").strip() or None
+        return {
+            "usage": sanitized.get("usage"),
+            "limit": sanitized.get("limit"),
+            "label": label,
+        }
+    except Exception:
+        logger.debug("OpenRouter key usage fetch failed for cost-history", exc_info=True)
+        return None
+
+
+def _read_cost_snapshots(provider: str) -> list[dict[str, Any]]:
+    """Read persisted daily snapshots for *provider* from disk.
+
+    Returns a list of ``{date, used, limit}`` dicts sorted by date
+    ascending.  Returns an empty list if the file does not exist or is
+    corrupt.
+    """
+    path = _cost_snapshots_dir() / f"{provider}.json"
+    if not path.exists():
+        return []
+    try:
+        raw = path.read_text(encoding="utf-8")
+        data = json.loads(raw)
+    except (OSError, json.JSONDecodeError, ValueError):
+        return []
+    if not isinstance(data, dict):
+        return []
+    snapshots = data.get("snapshots")
+    if not isinstance(snapshots, list):
+        return []
+    # Validate and sort
+    valid = []
+    for entry in snapshots:
+        if not isinstance(entry, dict):
+            continue
+        date = str(entry.get("date") or "").strip()
+        if not date:
+            continue
+        valid.append({
+            "date": date,
+            "used": _quota_number(entry.get("used")),
+            "limit": _quota_number(entry.get("limit")),
+        })
+    valid.sort(key=lambda e: e["date"])
+    return valid
+
+
+def _write_cost_snapshots(provider: str, snapshots: list[dict[str, Any]]) -> None:
+    """Persist daily snapshots for *provider* to disk atomically."""
+    snap_dir = _cost_snapshots_dir()
+    snap_dir.mkdir(parents=True, exist_ok=True)
+    path = snap_dir / f"{provider}.json"
+    payload = {"provider": provider, "snapshots": snapshots}
+    body = json.dumps(payload, ensure_ascii=False, indent=2)
+    import tempfile as _tempfile
+    _tmp_fd, _tmp_path = _tempfile.mkstemp(
+        dir=str(snap_dir), prefix=f".{provider}_", suffix=".tmp"
+    )
+    try:
+        with os.fdopen(_tmp_fd, "w", encoding="utf-8") as _f:
+            _f.write(body)
+            _f.flush()
+            os.fsync(_f.fileno())
+        os.replace(_tmp_path, path)
+    except BaseException:
+        try:
+            os.unlink(_tmp_path)
+        except OSError:
+            pass
+        raise
+
+
+def _append_cost_snapshot(provider: str, usage: int | float | None, limit: int | float | None) -> list[dict[str, Any]]:
+    """Append today's snapshot and return the updated list.
+
+    If a snapshot for today already exists it is updated in-place so
+    repeated calls within the same day are idempotent.
+    """
+    today = datetime.now(timezone.utc).strftime("%Y-%m-%d")
+    snapshots = _read_cost_snapshots(provider)
+    # Update or append today's entry
+    updated = False
+    for entry in snapshots:
+        if entry["date"] == today:
+            entry["used"] = usage
+            entry["limit"] = limit
+            updated = True
+            break
+    if not updated:
+        snapshots.append({"date": today, "used": usage, "limit": limit})
+    snapshots.sort(key=lambda e: e["date"])
+    # Cap to _COST_SNAPSHOT_MAX_DAYS entries (keep most recent)
+    if len(snapshots) > _COST_SNAPSHOT_MAX_DAYS:
+        snapshots = snapshots[-_COST_SNAPSHOT_MAX_DAYS:]
+    _write_cost_snapshots(provider, snapshots)
+    return snapshots
+
+
+def _compute_deltas(snapshots: list[dict[str, Any]], window_days: int) -> list[dict[str, Any]]:
+    """Compute daily deltas from cumulative usage snapshots.
+
+    Each snapshot carries cumulative ``used``; the delta for a day is
+    the difference between that day's cumulative value and the previous
+    day's.  The oldest day in the window has ``delta=None`` (no
+    previous baseline).
+    """
+    # Take only the last *window_days* entries
+    window = snapshots[-window_days:] if len(snapshots) > window_days else list(snapshots)
+    result: list[dict[str, Any]] = []
+    for i, entry in enumerate(window):
+        delta = None
+        if i > 0 and entry.get("used") is not None and window[i - 1].get("used") is not None:
+            delta = float(entry["used"]) - float(window[i - 1]["used"])
+            # Rounding: avoid -0.0 and tiny floating-point noise
+            if abs(delta) < 1e-9:
+                delta = 0.0
+            else:
+                delta = round(delta, 6)
+        result.append({
+            "date": entry["date"],
+            "used": entry.get("used"),
+            "delta": delta,
+        })
+    return result
+
+
+def get_provider_cost_history(provider_id: str | None = None, days: int = 7) -> dict[str, Any]:
+    """Return daily cost-history snapshots with deltas for a provider.
+
+    Currently only ``openrouter`` is supported.  On each call the
+    endpoint fetches the current cumulative usage from the OpenRouter
+    ``/auth/key`` endpoint, appends/updates today's snapshot, and
+    returns the last *days* snapshots with per-day deltas.
+
+    Returns a dict matching the existing API style (``ok``, ``provider``,
+    ``status``, ``message``, …).
+    """
+    provider = (provider_id or "").strip().lower()
+    if not provider:
+        return {
+            "ok": False,
+            "provider": None,
+            "status": "missing_provider",
+            "message": "Provider parameter is required.  Use ?provider=openrouter",
+        }
+
+    if provider != "openrouter":
+        display_name = _PROVIDER_DISPLAY.get(provider, provider.replace("-", " ").title())
+        return {
+            "ok": False,
+            "provider": provider,
+            "display_name": display_name,
+            "supported": False,
+            "status": "unsupported",
+            "message": f"Cost history is not available for {display_name}. Only openrouter is supported in this release.",
+        }
+
+    display_name = _PROVIDER_DISPLAY.get("openrouter", "OpenRouter")
+    api_key = _get_provider_api_key("openrouter")
+    if not api_key:
+        return {
+            "ok": False,
+            "provider": "openrouter",
+            "display_name": display_name,
+            "supported": True,
+            "status": "no_key",
+            "message": "OpenRouter cost history needs an OPENROUTER_API_KEY configured on the server.",
+        }
+
+    # Fetch current cumulative usage from OpenRouter
+    key_info = _fetch_openrouter_key_usage(api_key)
+    if key_info is None:
+        # Upstream failure — still return any previously persisted snapshots
+        # so the chart degrades gracefully instead of going blank.
+        snapshots = _read_cost_snapshots("openrouter")
+        deltas = _compute_deltas(snapshots, days)
+        return {
+            "ok": False,
+            "provider": "openrouter",
+            "display_name": display_name,
+            "supported": True,
+            "status": "unavailable",
+            "window_days": days,
+            "snapshots": deltas,
+            "limit": None,
+            "label": None,
+            "message": "OpenRouter cost history is temporarily unavailable. Showing last known data.",
+        }
+
+    # Persist today's snapshot
+    try:
+        snapshots = _append_cost_snapshot("openrouter", key_info["usage"], key_info["limit"])
+    except Exception:
+        logger.debug("Failed to persist cost snapshot for openrouter", exc_info=True)
+        snapshots = _read_cost_snapshots("openrouter")
+
+    deltas = _compute_deltas(snapshots, days)
+    return {
+        "ok": True,
+        "provider": "openrouter",
+        "display_name": display_name,
+        "supported": True,
+        "status": "available",
+        "window_days": days,
+        "snapshots": deltas,
+        "limit": key_info.get("limit"),
+        "label": key_info.get("label") or "OpenRouter credits",
+        "message": "OpenRouter cost history loaded.",
+    }
+
+
 # SECTION: Public API
 
 

api/routes.py

@@ -2033,7 +2033,7 @@ from api.run_journal import (
     read_run_events,
     stale_interrupted_event,
 )
-from api.providers import get_providers, get_provider_quota, set_provider_key, remove_provider_key
+from api.providers import get_providers, get_provider_quota, get_provider_cost_history, set_provider_key, remove_provider_key
 from api.onboarding import (
     apply_onboarding_setup,
     get_onboarding_status,
@@ -3323,6 +3323,16 @@ def handle_get(handler, parsed) -> bool:
         refresh = (query.get("refresh", [""])[0] or "").strip().lower() in {"1", "true", "yes", "on"}
         return j(handler, get_provider_quota(provider_id, refresh=refresh))
 
+    if parsed.path == "/api/provider/cost-history":
+        query = parse_qs(parsed.query)
+        provider_id = (query.get("provider", [""])[0] or None)
+        days_raw = (query.get("days", ["7"])[0] or "7").strip()
+        try:
+            days = max(1, min(int(days_raw), 365))
+        except (ValueError, TypeError):
+            days = 7
+        return j(handler, get_provider_cost_history(provider_id, days))
+
     if parsed.path == "/api/settings":
         settings = load_settings()
         # Never expose the stored password hash to clients

tests/test_provider_cost_history.py

@@ -0,0 +1,484 @@
+"""Regression coverage for OpenRouter cost-history endpoint (#692).
+
+Tests cover:
+  - Happy-path snapshot append and delta computation
+  - Missing credentials (no_key)
+  - Unsupported provider (non-openrouter)
+  - Upstream failure (graceful degradation with stale data)
+  - Malformed / corrupt snapshot file on disk
+  - Idempotent same-day updates
+  - No real network calls or private credential leakage
+"""
+
+from __future__ import annotations
+
+import json
+import os
+import sys
+import types
+import urllib.error
+from datetime import datetime, timezone
+from io import BytesIO
+from pathlib import Path
+from types import SimpleNamespace
+
+import api.config as config
+import api.profiles as profiles
+
+ROOT = Path(__file__).resolve().parents[1]
+
+
+class _FakeResponse:
+    """Minimal stand-in for urllib.request.urlopen context manager."""
+
+    def __init__(self, payload: bytes):
+        self._payload = payload
+
+    def __enter__(self):
+        return self
+
+    def __exit__(self, *exc):
+        return False
+
+    def read(self):
+        return self._payload
+
+
+def _with_config(model=None, providers=None):
+    old_cfg = dict(config.cfg)
+    old_mtime = config._cfg_mtime
+    config.cfg.clear()
+    config.cfg["model"] = model or {}
+    if providers is not None:
+        config.cfg["providers"] = providers
+    try:
+        config._cfg_mtime = config.Path(config._get_config_path()).stat().st_mtime
+    except Exception:
+        config._cfg_mtime = 0.0
+    return old_cfg, old_mtime
+
+
+def _restore_config(old_cfg, old_mtime):
+    config.cfg.clear()
+    config.cfg.update(old_cfg)
+    config._cfg_mtime = old_mtime
+
+
+# ── Happy path: snapshot append + delta response ──────────────────────────────
+
+
+def test_openrouter_cost_history_happy_path(monkeypatch, tmp_path):
+    """On-demand snapshot append returns deltas from cumulative usage."""
+    monkeypatch.setattr(profiles, "get_active_hermes_home", lambda: tmp_path)
+    monkeypatch.delenv("OPENROUTER_API_KEY", raising=False)
+    (tmp_path / ".env").write_text("OPENROUTER_API_KEY=test-or-key\n", encoding="utf-8")
+    old_cfg, old_mtime = _with_config(model={"provider": "openrouter"})
+
+    import api.providers as providers
+
+    call_count = {"n": 0}
+
+    def fake_urlopen(req, timeout):
+        call_count["n"] += 1
+        # Simulate cumulative usage of 5.0 credits used out of 20 limit
+        payload = {
+            "data": {
+                "limit_remaining": 15.0,
+                "usage": 5.0,
+                "limit": 20,
+                "label": "Test Label",
+                "key": "must-not-leak",
+            }
+        }
+        return _FakeResponse(json.dumps(payload).encode("utf-8"))
+
+    monkeypatch.setattr(providers.urllib.request, "urlopen", fake_urlopen)
+
+    # Freeze "today" so the test is deterministic
+    fake_today = "2030-04-15"
+    monkeypatch.setattr(providers, "datetime", type("DT", (), {
+        "now": staticmethod(lambda tz=None: datetime(2030, 4, 15, 12, 0, 0, tzinfo=tz or timezone.utc)),
+        "strftime": datetime.strftime,
+    }))
+
+    try:
+        result = providers.get_provider_cost_history("openrouter", days=7)
+    finally:
+        _restore_config(old_cfg, old_mtime)
+
+    assert result["ok"] is True
+    assert result["provider"] == "openrouter"
+    assert result["supported"] is True
+    assert result["status"] == "available"
+    assert result["window_days"] == 7
+    assert result["limit"] == 20
+    assert result["label"] == "Test Label"
+    assert result["message"] == "OpenRouter cost history loaded."
+    # One snapshot for today
+    assert len(result["snapshots"]) == 1
+    snap = result["snapshots"][0]
+    assert snap["date"] == fake_today
+    assert snap["used"] == 5.0
+    assert snap["delta"] is None  # first entry has no previous baseline
+    # Verify the snapshot file was persisted
+    snap_file = tmp_path / "cost-snapshots" / "openrouter.json"
+    assert snap_file.exists()
+    persisted = json.loads(snap_file.read_text(encoding="utf-8"))
+    assert len(persisted["snapshots"]) == 1
+    # No credential leakage
+    assert "test-or-key" not in repr(result)
+    assert "must-not-leak" not in repr(result)
+
+
+def test_openrouter_cost_history_deltas_from_cumulative(monkeypatch, tmp_path):
+    """Deltas are computed as differences between consecutive cumulative values."""
+    monkeypatch.setattr(profiles, "get_active_hermes_home", lambda: tmp_path)
+    monkeypatch.delenv("OPENROUTER_API_KEY", raising=False)
+    (tmp_path / ".env").write_text("OPENROUTER_API_KEY=test-or-key\n", encoding="utf-8")
+    old_cfg, old_mtime = _with_config(model={"provider": "openrouter"})
+
+    import api.providers as providers
+
+    # Pre-seed two historical snapshots
+    snap_dir = tmp_path / "cost-snapshots"
+    snap_dir.mkdir(parents=True, exist_ok=True)
+    historical = {
+        "provider": "openrouter",
+        "snapshots": [
+            {"date": "2030-04-13", "used": 3.0, "limit": 20},
+            {"date": "2030-04-14", "used": 4.5, "limit": 20},
+        ],
+    }
+    (snap_dir / "openrouter.json").write_text(json.dumps(historical), encoding="utf-8")
+
+    call_count = {"n": 0}
+
+    def fake_urlopen(req, timeout):
+        call_count["n"] += 1
+        # Current cumulative usage is 7.0
+        payload = {"data": {"usage": 7.0, "limit": 20, "label": "Credits"}}
+        return _FakeResponse(json.dumps(payload).encode("utf-8"))
+
+    monkeypatch.setattr(providers.urllib.request, "urlopen", fake_urlopen)
+
+    # Freeze "today"
+    monkeypatch.setattr(providers, "datetime", type("DT", (), {
+        "now": staticmethod(lambda tz=None: datetime(2030, 4, 15, 12, 0, 0, tzinfo=tz or timezone.utc)),
+        "strftime": datetime.strftime,
+    }))
+
+    try:
+        result = providers.get_provider_cost_history("openrouter", days=7)
+    finally:
+        _restore_config(old_cfg, old_mtime)
+
+    assert result["ok"] is True
+    snaps = result["snapshots"]
+    assert len(snaps) == 3
+    # Day 1: no delta (baseline)
+    assert snaps[0]["date"] == "2030-04-13"
+    assert snaps[0]["used"] == 3.0
+    assert snaps[0]["delta"] is None
+    # Day 2: delta = 4.5 - 3.0 = 1.5
+    assert snaps[1]["date"] == "2030-04-14"
+    assert snaps[1]["used"] == 4.5
+    assert snaps[1]["delta"] == 1.5
+    # Day 3 (today): delta = 7.0 - 4.5 = 2.5
+    assert snaps[2]["date"] == "2030-04-15"
+    assert snaps[2]["used"] == 7.0
+    assert snaps[2]["delta"] == 2.5
+
+
+# ── Missing credentials ───────────────────────────────────────────────────────
+
+
+def test_openrouter_cost_history_no_key(monkeypatch, tmp_path):
+    """No API key → safe no_key response without network call."""
+    monkeypatch.setattr(profiles, "get_active_hermes_home", lambda: tmp_path)
+    monkeypatch.delenv("OPENROUTER_API_KEY", raising=False)
+    old_cfg, old_mtime = _with_config(model={"provider": "openrouter"})
+
+    import api.providers as providers
+
+    def explode(*_a, **_kw):
+        raise AssertionError("should not call network without a key")
+
+    monkeypatch.setattr(providers.urllib.request, "urlopen", explode)
+
+    try:
+        result = providers.get_provider_cost_history("openrouter", days=7)
+    finally:
+        _restore_config(old_cfg, old_mtime)
+
+    assert result["ok"] is False
+    assert result["provider"] == "openrouter"
+    assert result["supported"] is True
+    assert result["status"] == "no_key"
+    assert "OPENROUTER_API_KEY" in result["message"]
+
+
+# ── Unsupported provider ──────────────────────────────────────────────────────
+
+
+def test_cost_history_unsupported_provider(monkeypatch, tmp_path):
+    """Non-openrouter providers return a clear unsupported response."""
+    monkeypatch.setattr(profiles, "get_active_hermes_home", lambda: tmp_path)
+    old_cfg, old_mtime = _with_config(model={"provider": "anthropic"})
+
+    import api.providers as providers
+
+    try:
+        result = providers.get_provider_cost_history("anthropic", days=7)
+    finally:
+        _restore_config(old_cfg, old_mtime)
+
+    assert result["ok"] is False
+    assert result["provider"] == "anthropic"
+    assert result["supported"] is False
+    assert result["status"] == "unsupported"
+    assert "openrouter" in result["message"].lower()
+
+
+def test_cost_history_missing_provider_param(monkeypatch, tmp_path):
+    """Empty provider parameter returns a clear error."""
+    monkeypatch.setattr(profiles, "get_active_hermes_home", lambda: tmp_path)
+
+    import api.providers as providers
+
+    result = providers.get_provider_cost_history("", days=7)
+    assert result["ok"] is False
+    assert result["status"] == "missing_provider"
+
+    result2 = providers.get_provider_cost_history(None, days=7)
+    assert result2["ok"] is False
+    assert result2["status"] == "missing_provider"
+
+
+# ── Upstream failure / graceful degradation ────────────────────────────────────
+
+
+def test_openrouter_cost_history_upstream_failure_degrades_gracefully(monkeypatch, tmp_path):
+    """When OpenRouter API fails, previously persisted snapshots are still returned."""
+    monkeypatch.setattr(profiles, "get_active_hermes_home", lambda: tmp_path)
+    monkeypatch.delenv("OPENROUTER_API_KEY", raising=False)
+    (tmp_path / ".env").write_text("OPENROUTER_API_KEY=test-or-key\n", encoding="utf-8")
+    old_cfg, old_mtime = _with_config(model={"provider": "openrouter"})
+
+    import api.providers as providers
+
+    # Pre-seed a snapshot
+    snap_dir = tmp_path / "cost-snapshots"
+    snap_dir.mkdir(parents=True, exist_ok=True)
+    historical = {
+        "provider": "openrouter",
+        "snapshots": [
+            {"date": "2030-04-13", "used": 3.0, "limit": 20},
+        ],
+    }
+    (snap_dir / "openrouter.json").write_text(json.dumps(historical), encoding="utf-8")
+
+    req = providers.urllib.request.Request("https://openrouter.ai/api/v1/key")
+    def fake_urlopen(_req, timeout=None):
+        raise urllib.error.HTTPError(req.full_url, 500, "Server Error", {}, BytesIO(b"error"))
+
+    monkeypatch.setattr(providers.urllib.request, "urlopen", fake_urlopen)
+
+    try:
+        result = providers.get_provider_cost_history("openrouter", days=7)
+    finally:
+        _restore_config(old_cfg, old_mtime)
+
+    assert result["ok"] is False
+    assert result["status"] == "unavailable"
+    # Still returns previously persisted data
+    assert len(result["snapshots"]) == 1
+    assert result["snapshots"][0]["date"] == "2030-04-13"
+    assert "temporarily unavailable" in result["message"].lower()
+
+
+def test_openrouter_cost_history_timeout_is_safe(monkeypatch, tmp_path):
+    """Timeout from OpenRouter does not produce a traceback or leak secrets."""
+    monkeypatch.setattr(profiles, "get_active_hermes_home", lambda: tmp_path)
+    monkeypatch.delenv("OPENROUTER_API_KEY", raising=False)
+    (tmp_path / ".env").write_text("OPENROUTER_API_KEY=test-or-key\n", encoding="utf-8")
+    old_cfg, old_mtime = _with_config(model={"provider": "openrouter"})
+
+    import api.providers as providers
+
+    def fake_urlopen(_req, timeout=None):
+        raise TimeoutError("slow secret")
+
+    monkeypatch.setattr(providers.urllib.request, "urlopen", fake_urlopen)
+
+    try:
+        result = providers.get_provider_cost_history("openrouter", days=7)
+    finally:
+        _restore_config(old_cfg, old_mtime)
+
+    assert result["ok"] is False
+    assert result["status"] == "unavailable"
+    assert "test-or-key" not in repr(result)
+    assert "secret" not in repr(result).lower()
+
+
+# ── Malformed / corrupt snapshot file ─────────────────────────────────────────
+
+
+def test_openrouter_cost_history_corrupt_snapshot_file(monkeypatch, tmp_path):
+    """A corrupt snapshot file on disk is handled gracefully (treated as empty)."""
+    monkeypatch.setattr(profiles, "get_active_hermes_home", lambda: tmp_path)
+    monkeypatch.delenv("OPENROUTER_API_KEY", raising=False)
+    (tmp_path / ".env").write_text("OPENROUTER_API_KEY=test-or-key\n", encoding="utf-8")
+    old_cfg, old_mtime = _with_config(model={"provider": "openrouter"})
+
+    import api.providers as providers
+
+    # Write a corrupt file
+    snap_dir = tmp_path / "cost-snapshots"
+    snap_dir.mkdir(parents=True, exist_ok=True)
+    (snap_dir / "openrouter.json").write_text("NOT VALID JSON{{{{", encoding="utf-8")
+
+    def fake_urlopen(req, timeout):
+        payload = {"data": {"usage": 2.0, "limit": 10, "label": "Credits"}}
+        return _FakeResponse(json.dumps(payload).encode("utf-8"))
+
+    monkeypatch.setattr(providers.urllib.request, "urlopen", fake_urlopen)
+
+    # Freeze "today"
+    monkeypatch.setattr(providers, "datetime", type("DT", (), {
+        "now": staticmethod(lambda tz=None: datetime(2030, 4, 15, 12, 0, 0, tzinfo=tz or timezone.utc)),
+        "strftime": datetime.strftime,
+    }))
+
+    try:
+        result = providers.get_provider_cost_history("openrouter", days=7)
+    finally:
+        _restore_config(old_cfg, old_mtime)
+
+    # Corrupt file is ignored; fresh snapshot is created
+    assert result["ok"] is True
+    assert len(result["snapshots"]) == 1
+    assert result["snapshots"][0]["date"] == "2030-04-15"
+
+
+# ── Idempotent same-day updates ───────────────────────────────────────────────
+
+
+def test_openrouter_cost_history_same_day_idempotent(monkeypatch, tmp_path):
+    """Repeated calls on the same day update the snapshot in-place."""
+    monkeypatch.setattr(profiles, "get_active_hermes_home", lambda: tmp_path)
+    monkeypatch.delenv("OPENROUTER_API_KEY", raising=False)
+    (tmp_path / ".env").write_text("OPENROUTER_API_KEY=test-or-key\n", encoding="utf-8")
+    old_cfg, old_mtime = _with_config(model={"provider": "openrouter"})
+
+    import api.providers as providers
+
+    call_count = {"n": 0}
+
+    def fake_urlopen(req, timeout):
+        call_count["n"] += 1
+        usage = 5.0 + call_count["n"]  # usage grows each call
+        payload = {"data": {"usage": usage, "limit": 20, "label": "Credits"}}
+        return _FakeResponse(json.dumps(payload).encode("utf-8"))
+
+    monkeypatch.setattr(providers.urllib.request, "urlopen", fake_urlopen)
+
+    # Freeze "today"
+    monkeypatch.setattr(providers, "datetime", type("DT", (), {
+        "now": staticmethod(lambda tz=None: datetime(2030, 4, 15, 12, 0, 0, tzinfo=tz or timezone.utc)),
+        "strftime": datetime.strftime,
+    }))
+
+    try:
+        r1 = providers.get_provider_cost_history("openrouter", days=7)
+        r2 = providers.get_provider_cost_history("openrouter", days=7)
+    finally:
+        _restore_config(old_cfg, old_mtime)
+
+    # Both calls succeed; only one snapshot date (today)
+    assert r1["ok"] is True
+    assert r2["ok"] is True
+    assert len(r1["snapshots"]) == 1
+    assert len(r2["snapshots"]) == 1
+    # Second call updated the same day's used value
+    assert r2["snapshots"][0]["used"] == 7.0  # 5.0 + 2 (second call)
+    # Verify persisted file has only one entry for today
+    snap_file = tmp_path / "cost-snapshots" / "openrouter.json"
+    persisted = json.loads(snap_file.read_text(encoding="utf-8"))
+    assert len(persisted["snapshots"]) == 1
+
+
+# ── Window days parameter ─────────────────────────────────────────────────────
+
+
+def test_openrouter_cost_history_window_days_truncation(monkeypatch, tmp_path):
+    """The window_days parameter limits how many snapshots are returned."""
+    monkeypatch.setattr(profiles, "get_active_hermes_home", lambda: tmp_path)
+    monkeypatch.delenv("OPENROUTER_API_KEY", raising=False)
+    (tmp_path / ".env").write_text("OPENROUTER_API_KEY=test-or-key\n", encoding="utf-8")
+    old_cfg, old_mtime = _with_config(model={"provider": "openrouter"})
+
+    import api.providers as providers
+
+    # Pre-seed 5 historical snapshots
+    snap_dir = tmp_path / "cost-snapshots"
+    snap_dir.mkdir(parents=True, exist_ok=True)
+    historical = {
+        "provider": "openrouter",
+        "snapshots": [
+            {"date": f"2030-04-{d:02d}", "used": float(d), "limit": 20}
+            for d in range(10, 15)
+        ],
+    }
+    (snap_dir / "openrouter.json").write_text(json.dumps(historical), encoding="utf-8")
+
+    def fake_urlopen(req, timeout):
+        payload = {"data": {"usage": 15.0, "limit": 20, "label": "Credits"}}
+        return _FakeResponse(json.dumps(payload).encode("utf-8"))
+
+    monkeypatch.setattr(providers.urllib.request, "urlopen", fake_urlopen)
+
+    # Freeze "today"
+    monkeypatch.setattr(providers, "datetime", type("DT", (), {
+        "now": staticmethod(lambda tz=None: datetime(2030, 4, 15, 12, 0, 0, tzinfo=tz or timezone.utc)),
+        "strftime": datetime.strftime,
+    }))
+
+    try:
+        result = providers.get_provider_cost_history("openrouter", days=3)
+    finally:
+        _restore_config(old_cfg, old_mtime)
+
+    # 5 historical + 1 today = 6 total, but window_days=3 returns last 3
+    assert result["ok"] is True
+    assert result["window_days"] == 3
+    assert len(result["snapshots"]) == 3
+    # The returned snapshots are the most recent 3
+    assert result["snapshots"][0]["date"] == "2030-04-13"
+    assert result["snapshots"][1]["date"] == "2030-04-14"
+    assert result["snapshots"][2]["date"] == "2030-04-15"
+
+
+# ── No real network calls ─────────────────────────────────────────────────────
+
+
+def test_cost_history_uses_no_real_network(monkeypatch, tmp_path):
+    """Every test path must monkeypatch urlopen; verify no real calls escape."""
+    monkeypatch.setattr(profiles, "get_active_hermes_home", lambda: tmp_path)
+    monkeypatch.delenv("OPENROUTER_API_KEY", raising=False)
+    old_cfg, old_mtime = _with_config(model={"provider": "openrouter"})
+
+    import api.providers as providers
+
+    # Without a key, no network call is made at all
+    def explode(*_a, **_kw):
+        raise AssertionError("real network call detected")
+
+    monkeypatch.setattr(providers.urllib.request, "urlopen", explode)
+
+    try:
+        result = providers.get_provider_cost_history("openrouter", days=7)
+    finally:
+        _restore_config(old_cfg, old_mtime)
+
+    assert result["status"] == "no_key"