feat(budget): operator budget signal — operations-center.sh budget (audit D1)

.console/log.md

@@ -1,3 +1,20 @@
+## 2026-07-14 — feat(budget): operator budget signal `operations-center.sh budget` (audit D1)
+
+Voluntary operator readout (D1 part 3). A human session can't be hard-gated, so
+per the operator's decision the fleet is throttled and the operator gets a
+SIGNAL instead. `operations-center.sh budget` (read-only, skips janitor) prints
+one line via `python -m operations_center.execution.usage_budget`:
+`claude budget: <ok|THROTTLING|DISABLED> — N% of reserve threshold used | XM
+weighted before the fleet throttles | YM before the hard 5h limit | window
+…→… | cap …`. New testable `format_status_line(BudgetStatus)` +
+`__main__`. Answers the real question: how much room before the fleet throttles
+vs before the hard 5h limit stops everything. 1 test; ruff+ty clean.
+Remaining D1: reviewer backend ladder + codex fallback (backend-agnostic
+harness — the big one, designed next), standalone budget writer (F3, decouple
+from the loop). NOTE: audit F9 was partly wrong — WATCH_INTERVAL_* ARE read by
+operations-center.sh watch-role dispatch (lines ~354-358), just not by the
+Python side; re-triage F9 before acting.
+
 ## 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

scripts/operations-center.sh

@@ -729,7 +729,7 @@ shift || true
 cd "${ROOT_DIR}"
 # Skip janitor for read-only / stop commands — they're fast and don't need it.
 case "${cmd}" in
-  watch-all-status|dev-status|watch-all-stop|watch-stop|watchdog-stop|plane-status|providers-status|doctor|status|worker-backend-status|worker-backend-probe|loop-start|loop-stop|loop-status|loop-log) ;;
+  watch-all-status|dev-status|watch-all-stop|watch-stop|watchdog-stop|plane-status|providers-status|doctor|status|worker-backend-status|worker-backend-probe|loop-start|loop-stop|loop-status|loop-log|budget) ;;
   *) run_janitor ;;
 esac
 
@@ -1069,6 +1069,12 @@ PYEOF
   loop-status)
     loop_status
     ;;
+  budget)
+    # Operator budget signal (audit D1): voluntary readout of how much claude
+    # subscription room is left before the fleet throttles / the hard 5h limit.
+    load_env_file
+    "${VENV_DIR}/bin/python" -m operations_center.execution.usage_budget
+    ;;
   loop-log)
     loop_log
     ;;

src/operations_center/execution/usage_budget.py

@@ -266,10 +266,41 @@ def budget_status(now: datetime | None = None) -> BudgetStatus:
     )
 
 
+def _mtok(x: float) -> str:
+    return f"{x / 1e6:.1f}M"
+
+
+def format_status_line(s: BudgetStatus) -> str:
+    """One-line, human-readable budget readout for the operator (audit D1).
+
+    Voluntary signal only — a human session can't be hard-gated. It answers the
+    operator's real question: how much room is left before the FLEET throttles,
+    and before the HARD 5h limit stops everything.
+    """
+    pct = 100.0 * s.used_weighted / max(s.threshold_weighted, 1.0)
+    if s.disabled:
+        state = "DISABLED (guard off)"
+    elif s.exhausted:
+        state = "THROTTLING — fleet diverts to codex / parks reviews"
+    else:
+        state = "ok"
+    return (
+        f"claude budget: {state} — {pct:.0f}% of reserve threshold used | "
+        f"{_mtok(s.threshold_weighted - s.used_weighted)} weighted before the fleet throttles | "
+        f"{_mtok(s.cap_weighted - s.used_weighted)} before the hard 5h limit | "
+        f"window {s.bucket_start:%H:%MZ}->{s.bucket_end:%H:%MZ} | cap {_mtok(s.cap_weighted)}"
+    )
+
+
 __all__ = [
     "BudgetStatus",
     "budget_status",
+    "format_status_line",
     "BUCKET_SPAN",
     "DEFAULT_CAP_WEIGHTED",
     "DEFAULT_RESERVE",
 ]
+
+
+if __name__ == "__main__":  # `python -m operations_center.execution.usage_budget`
+    print(format_status_line(budget_status()))  # noqa: T201 — CLI readout is the point

tests/unit/execution/test_usage_budget.py

@@ -17,7 +17,9 @@
 from operations_center.execution.usage_budget import (
     BUCKET_SPAN,
     DEFAULT_CAP_WEIGHTED,
+    BudgetStatus,
     budget_status,
+    format_status_line,
 )
 from operations_center.execution.usage_store import UsageStore
 
@@ -190,6 +192,28 @@ def test_naive_timestamp_is_counted_not_crashed(monkeypatch, tmp_path: Path):
     assert round(s.used_weighted) == 100
 
 
+def _status(**kw) -> BudgetStatus:
+    base = dict(
+        bucket_start=NOW,
+        bucket_end=NOW + BUCKET_SPAN,
+        used_weighted=1000.0,
+        cap_weighted=10000.0,
+        reserve_fraction=0.25,
+        exhausted=False,
+        disabled=False,
+    )
+    base.update(kw)
+    return BudgetStatus(**base)
+
+
+def test_format_status_line_reflects_state(monkeypatch):
+    ok = format_status_line(_status())
+    assert "ok" in ok
+    assert "before the fleet throttles" in ok and "before the hard 5h limit" in ok
+    assert "THROTTLING" in format_status_line(_status(used_weighted=9000.0, exhausted=True))
+    assert "DISABLED" in format_status_line(_status(disabled=True))
+
+
 def _store(monkeypatch, tmp_path: Path) -> UsageStore:
     monkeypatch.setenv("OPERATIONS_CENTER_EXECUTION_USAGE_PATH", str(tmp_path / "usage.json"))
     return UsageStore()