feat(budget): self-calibrating cap — learn from observed limits (audit D4/F17)

.console/log.md

@@ -1,3 +1,21 @@
+## 2026-07-14 — feat(budget): self-calibrating cap — learn from observed limits (audit D4/F17)
+
+Retires the 42M magic constant. The budget guard's cap was a single-sample
+constant (fragile: plan-tier or weight changes silently invalidate it). Now:
+when an ACCOUNT-WIDE claude limit trips (session_5h / global_weekly), the
+on_cooldown hook snapshots the estimator's current trailing-window weighted
+usage — an observed sample of the real cap, measured in the SAME units the
+estimator uses, so systematic estimator bias cancels out — and records it in
+the usage store (best-effort, one sample per episode via a 1h recency guard;
+last 8 kept). `budget_status` cap precedence is now: explicit
+`OC_CLAUDE_BUDGET_CAP_WEIGHTED` env override > learned median (>=2 samples,
+robust to a single anomalous event) > 42M cold-start seed. `usage_store` gains
+`record_budget_cap_sample` + `learned_budget_cap`; `usage_budget` gains
+`_resolve_cap`/`_learned_cap` (lazy store import, best-effort). 9 new tests
+(median/min-samples, recency guard, learned-vs-env precedence, on_cooldown
+records for session_5h but not model_weekly). 38 pass; ruff+ty clean. Next
+audit items: D1 reviewer backend ladder, D2 council, D3 attribution.
+
 ## 2026-07-13 — feat(budget): claude 25% reserve guard + audit fixes (F1/F2/F16)
 
 Lands the operator's 2026-07-13 directive (leave ~25% of every 5h claude bucket

src/operations_center/entrypoints/loop_bridge/main.py

@@ -114,6 +114,7 @@ def on_cooldown(payload_json: str) -> int:
     The engine's ``model`` is the full model id; the usage store wants the
     short name from the backend mapping.
     """
+    from operations_center.backends.limit_classifier import GLOBAL_WEEKLY, SESSION_5H
     from operations_center.execution.usage_store import UsageStore
 
     payload = json.loads(payload_json)
@@ -127,13 +128,26 @@ def on_cooldown(payload_json: str) -> int:
     )
     limit_kind = str(payload.get("limit_kind") or "unknown")
     model = short_model if payload.get("model") else None
-    UsageStore().record_worker_backend_cooldown(
+    now = datetime.now(timezone.utc)
+    store = UsageStore()
+    store.record_worker_backend_cooldown(
         worker_backend=worker_backend,
         reset_at=reset_at,
-        now=datetime.now(timezone.utc),
+        now=now,
         limit_kind=limit_kind,
         model=model,
     )
+    # Self-calibrating budget cap (audit D4): an account-wide claude limit means
+    # the estimator's current weighted usage is an observed sample of the real
+    # cap. Record it (best-effort, one sample per episode) so the budget guard
+    # learns the cap instead of trusting the 42M cold-start constant.
+    if worker_backend == "claude_code" and limit_kind in (SESSION_5H, GLOBAL_WEEKLY):
+        try:
+            from operations_center.execution.usage_budget import budget_status
+
+            store.record_budget_cap_sample(weighted=budget_status(now=now).used_weighted, now=now)
+        except Exception as exc:  # noqa: BLE001 — calibration must not block cooldown recording
+            logger.warning("loop_bridge: budget cap sample not recorded: %s", exc)
     return 0
 
 

src/operations_center/execution/usage_budget.py

@@ -37,7 +37,9 @@
 rather than late.
 
 Env overrides (documented in .env.operations-center.example):
-  OC_CLAUDE_BUDGET_CAP_WEIGHTED  — window capacity in weighted tokens (default 42_000_000)
+  OC_CLAUDE_BUDGET_CAP_WEIGHTED  — window capacity in weighted tokens; overrides the
+                                   learned cap (default: learned from observed limits,
+                                   else the 42_000_000 cold-start seed)
   OC_CLAUDE_BUDGET_RESERVE       — fraction to leave unspent (default 0.25, clamped [0, 0.95])
   OC_CLAUDE_BUDGET_DISABLED      — truthy (1/true/yes/on) → guard reports never-exhausted
 """
@@ -108,6 +110,38 @@ def _truthy(value: str | None) -> bool:
     return str(value or "").strip().lower() in _TRUTHY
 
 
+def _learned_cap() -> float | None:
+    """Best-effort cap learned from observed account-wide limit events (audit D4).
+
+    Lazy import so this module stays dependency-light and importable without the
+    usage store; any failure falls back to the default cap.
+    """
+    try:
+        from operations_center.execution.usage_store import UsageStore
+
+        return UsageStore().learned_budget_cap()
+    except Exception:  # noqa: BLE001 — calibration is best-effort
+        return None
+
+
+def _resolve_cap() -> float:
+    """Cap precedence: explicit env override > learned-from-observed-limits > default.
+
+    The 42M default is only a cold-start seed; once the fleet has observed a
+    couple of real account-wide limits, the learned median replaces it and the
+    magic constant retires (audit F17).
+    """
+    raw = os.environ.get("OC_CLAUDE_BUDGET_CAP_WEIGHTED")
+    if raw is not None and raw.strip() != "":
+        cap = _env_float("OC_CLAUDE_BUDGET_CAP_WEIGHTED", DEFAULT_CAP_WEIGHTED)
+    else:
+        cap = _learned_cap() or DEFAULT_CAP_WEIGHTED
+    if cap <= 0:
+        logger.warning("usage_budget: resolved cap %s <= 0; using default %s", cap, DEFAULT_CAP_WEIGHTED)
+        cap = DEFAULT_CAP_WEIGHTED
+    return cap
+
+
 def _claude_projects_dir() -> Path:
     return Path(os.environ.get("CLAUDE_CONFIG_DIR", str(Path.home() / ".claude"))) / "projects"
 
@@ -205,10 +239,7 @@ def _reset_horizon(
 
 def budget_status(now: datetime | None = None) -> BudgetStatus:
     now = now or datetime.now(timezone.utc)
-    cap = _env_float("OC_CLAUDE_BUDGET_CAP_WEIGHTED", DEFAULT_CAP_WEIGHTED)
-    if cap <= 0:  # a non-positive cap would make the threshold meaningless
-        logger.warning("usage_budget: cap %s <= 0; using default %s", cap, DEFAULT_CAP_WEIGHTED)
-        cap = DEFAULT_CAP_WEIGHTED
+    cap = _resolve_cap()
     reserve = min(max(_env_float("OC_CLAUDE_BUDGET_RESERVE", DEFAULT_RESERVE), 0.0), _MAX_RESERVE)
     disabled = _truthy(os.environ.get("OC_CLAUDE_BUDGET_DISABLED"))
 

src/operations_center/execution/usage_store.py

@@ -487,6 +487,58 @@ def record_worker_backend_cooldown(
                 event["model"] = model
             self._append_event(data, event, now=now)
 
+    def record_budget_cap_sample(
+        self,
+        *,
+        weighted: float,
+        now: datetime,
+        min_gap: timedelta = timedelta(hours=1),
+        keep: int = 8,
+    ) -> None:
+        """Record an observed budget-cap sample (self-calibrating cap, audit D4/F17).
+
+        ``weighted`` is the budget estimator's trailing-window usage at the
+        moment an account-wide claude limit tripped — an observed sample of the
+        real cap, measured in the SAME weighted-token units the estimator uses,
+        so systematic estimator bias cancels out. At most one sample per limit
+        episode (``min_gap`` recency guard, since the engine re-records the
+        cooldown every iteration); the most recent ``keep`` samples are retained.
+        """
+        if weighted <= 0:
+            return
+        with self._exclusive():
+            data = self.load()
+            samples = list(data.get("budget_cap_samples", []))
+            if samples:
+                last_at = samples[-1].get("at")
+                try:
+                    if last_at and now - datetime.fromisoformat(str(last_at)) < min_gap:
+                        return  # same limit episode — one sample per episode
+                except (ValueError, TypeError):
+                    pass
+            samples.append({"weighted": float(weighted), "at": now.isoformat()})
+            data["budget_cap_samples"] = samples[-keep:]
+            self.save(data, now=now)
+
+    def learned_budget_cap(self, *, min_samples: int = 2) -> float | None:
+        """Median of recent observed cap samples, or None if too few to trust.
+
+        The median is robust to a single anomalous limit event; the budget
+        guard's 25% reserve absorbs the remaining error. Returns None until at
+        least ``min_samples`` observations exist (caller falls back to default).
+        """
+        data = self.load()
+        weights = sorted(
+            float(s["weighted"])
+            for s in data.get("budget_cap_samples", [])
+            if isinstance(s, dict) and s.get("weighted")
+        )
+        if len(weights) < min_samples:
+            return None
+        n = len(weights)
+        mid = n // 2
+        return weights[mid] if n % 2 else (weights[mid - 1] + weights[mid]) / 2.0
+
     def worker_backend_cooldown_until(
         self,
         worker_backend: str,

tests/unit/entrypoints/loop_bridge/test_main.py

@@ -79,6 +79,50 @@ def test_on_cooldown_records_into_usage_store(monkeypatch, tmp_path: Path) -> No
     )
 
 
+def test_on_cooldown_session5h_records_budget_cap_sample(monkeypatch, tmp_path: Path) -> None:
+    # audit D4: an account-wide claude limit calibrates the budget cap.
+    monkeypatch.setenv("OPERATIONS_CENTER_EXECUTION_USAGE_PATH", str(tmp_path / "usage.json"))
+    cfg = tmp_path / "cfg"
+    monkeypatch.setenv("CLAUDE_CONFIG_DIR", str(cfg))
+    monkeypatch.delenv("OC_CLAUDE_BUDGET_CAP_WEIGHTED", raising=False)
+    proj = cfg / "projects" / "p1"
+    proj.mkdir(parents=True)
+    now = datetime.now(timezone.utc)
+    proj.joinpath("s.jsonl").write_text(
+        json.dumps(
+            {"timestamp": now.isoformat(), "message": {"model": "claude-sonnet-5", "usage": {"output_tokens": 100}}}
+        )
+        + "\n"
+    )
+    reset_at = now + timedelta(hours=2)
+    payload = json.dumps(
+        {
+            "backend": "claude",
+            "reset_at": reset_at.strftime("%Y-%m-%dT%H:%M:%SZ"),
+            "limit_kind": "session_5h",
+            "model": None,
+        }
+    )
+    assert bridge.on_cooldown(payload) == 0
+    # the estimator's 500 weighted (5*100 output) is captured as a cap observation
+    assert UsageStore().learned_budget_cap(min_samples=1) == 500.0
+
+
+def test_on_cooldown_model_weekly_records_no_cap_sample(monkeypatch, tmp_path: Path) -> None:
+    monkeypatch.setenv("OPERATIONS_CENTER_EXECUTION_USAGE_PATH", str(tmp_path / "usage.json"))
+    reset_at = datetime.now(timezone.utc) + timedelta(hours=2)
+    payload = json.dumps(
+        {
+            "backend": "claude",
+            "reset_at": reset_at.strftime("%Y-%m-%dT%H:%M:%SZ"),
+            "limit_kind": "model_weekly",  # per-model, NOT account-wide → no calibration
+            "model": "claude-sonnet-5",
+        }
+    )
+    assert bridge.on_cooldown(payload) == 0
+    assert UsageStore().learned_budget_cap(min_samples=1) is None
+
+
 def test_on_cooldown_unknown_backend_is_noop(monkeypatch, tmp_path: Path) -> None:
     monkeypatch.setenv("OPERATIONS_CENTER_EXECUTION_USAGE_PATH", str(tmp_path / "usage.json"))
     payload = json.dumps(

tests/unit/execution/test_usage_budget.py

@@ -19,6 +19,7 @@
     DEFAULT_CAP_WEIGHTED,
     budget_status,
 )
+from operations_center.execution.usage_store import UsageStore
 
 NOW = datetime(2026, 7, 13, 19, 0, tzinfo=timezone.utc)
 
@@ -189,6 +190,52 @@ def test_naive_timestamp_is_counted_not_crashed(monkeypatch, tmp_path: Path):
     assert round(s.used_weighted) == 100
 
 
+def _store(monkeypatch, tmp_path: Path) -> UsageStore:
+    monkeypatch.setenv("OPERATIONS_CENTER_EXECUTION_USAGE_PATH", str(tmp_path / "usage.json"))
+    return UsageStore()
+
+
+def test_learned_cap_needs_two_samples_then_median(monkeypatch, tmp_path: Path):
+    # audit D4: cap is learned from observed account-wide limit events.
+    s = _store(monkeypatch, tmp_path)
+    assert s.learned_budget_cap() is None  # no observations
+    s.record_budget_cap_sample(weighted=40_000_000, now=NOW)
+    assert s.learned_budget_cap() is None  # one is not enough to trust
+    s.record_budget_cap_sample(weighted=44_000_000, now=NOW + timedelta(hours=2))
+    assert s.learned_budget_cap() == 42_000_000  # median of two
+    s.record_budget_cap_sample(weighted=30_000_000, now=NOW + timedelta(hours=4))
+    assert s.learned_budget_cap() == 40_000_000  # median of three, robust to the low outlier
+
+
+def test_cap_sample_recency_guard_keeps_one_per_episode(monkeypatch, tmp_path: Path):
+    s = _store(monkeypatch, tmp_path)
+    s.record_budget_cap_sample(weighted=40_000_000, now=NOW)
+    # the engine re-records the same cooldown every iteration — within min_gap, dropped
+    s.record_budget_cap_sample(weighted=99_000_000, now=NOW + timedelta(minutes=20))
+    s.record_budget_cap_sample(weighted=44_000_000, now=NOW + timedelta(hours=2))  # new episode
+    assert s.learned_budget_cap() == 42_000_000  # median of [40M, 44M]; the 99M re-record dropped
+
+
+def test_budget_status_uses_learned_cap_when_no_env_override(monkeypatch, tmp_path: Path):
+    monkeypatch.setenv("CLAUDE_CONFIG_DIR", str(tmp_path))  # no transcripts → used 0
+    monkeypatch.delenv("OC_CLAUDE_BUDGET_CAP_WEIGHTED", raising=False)
+    monkeypatch.delenv("OC_CLAUDE_BUDGET_RESERVE", raising=False)
+    s = _store(monkeypatch, tmp_path)
+    s.record_budget_cap_sample(weighted=1000.0, now=NOW - timedelta(hours=3))
+    s.record_budget_cap_sample(weighted=1000.0, now=NOW - timedelta(hours=1))
+    status = budget_status(now=NOW)
+    assert status.cap_weighted == 1000.0  # learned, not the 42M seed
+
+
+def test_env_cap_override_beats_learned(monkeypatch, tmp_path: Path):
+    monkeypatch.setenv("CLAUDE_CONFIG_DIR", str(tmp_path))
+    monkeypatch.setenv("OC_CLAUDE_BUDGET_CAP_WEIGHTED", "500")  # explicit override
+    s = _store(monkeypatch, tmp_path)
+    s.record_budget_cap_sample(weighted=1000.0, now=NOW - timedelta(hours=3))
+    s.record_budget_cap_sample(weighted=1000.0, now=NOW - timedelta(hours=1))
+    assert budget_status(now=NOW).cap_weighted == 500.0  # env wins over the learned 1000
+
+
 def test_disabled_env_reports_not_exhausted(monkeypatch, tmp_path: Path):
     projects = _env(monkeypatch, tmp_path, cap=10.0)
     _write_transcript(