feat(vendor-state): add pure aggregator with stable metric keys

src/vendor_state.py

@@ -0,0 +1,80 @@
+from dataclasses import dataclass
+
+@dataclass(frozen=True)
+class VendorMetric:
+ """Atomic vendor-state metric.
+ [C: src/gui_2.py:render_vendor_state]
+ """
+ key: str
+ label: str
+ value: str
+ state: str
+ tooltip: str
+
+def get_vendor_state(app) -> list[VendorMetric]:
+ """Aggregate per-vendor session state for the Operations Hub Vendor State tab.
+ [C: src/gui_2.py:render_vendor_state]
+ """
+ out: list[VendorMetric] = []
+ out.append(VendorMetric(
+  key="provider_model",
+  label="Provider / Model",
+  value=f"{app.current_provider} / {app.current_model}",
+  state="info",
+  tooltip="The vendor and model that will handle the next request."
+ ))
+ ctrl = getattr(app, "controller", None)
+ tt = getattr(ctrl, "token_tracker", None) if ctrl else None
+ if tt and getattr(tt, "limit", 0):
+  pct = 100.0 * getattr(tt, "used", 0) / tt.limit
+  state = "warn" if pct > 75 else "ok"
+  out.append(VendorMetric(
+   key="context_window",
+   label="Context Window",
+   value=f"{tt.used:,} / {tt.limit:,} ({pct:.0f}%)",
+   state=state,
+   tooltip="Used vs total context window for the current session."
+  ))
+ else:
+  out.append(VendorMetric(
+   key="context_window", label="Context Window", value="—", state="info",
+   tooltip="No token tracker attached for the current provider."
+  ))
+ if tt is not None:
+  hits = getattr(tt, "cache_hits", 0)
+  miss = getattr(tt, "cache_misses", 0)
+  total = hits + miss
+  rate = (100.0 * hits / total) if total else 0.0
+  out.append(VendorMetric(
+   key="cache", label="Cache Hit Rate",
+   value=f"{rate:.0f}% ({hits:,}/{total:,})",
+   state="ok" if rate > 50 else "info",
+   tooltip="Server-side prompt cache hit rate for the current session."
+  ))
+ else:
+  out.append(VendorMetric(
+   key="cache", label="Cache Hit Rate", value="—", state="info",
+   tooltip="No token tracker attached for the current provider."
+  ))
+ quota = (getattr(ctrl, "vendor_quota", {}) or {}) if ctrl else {}
+ pct_left = quota.get("remaining_pct")
+ if pct_left is None:
+  out.append(VendorMetric(
+   key="quota", label="Vendor Quota", value="—", state="info",
+   tooltip="Vendor did not report quota for the current billing period."
+  ))
+ else:
+  out.append(VendorMetric(
+   key="quota", label="Vendor Quota",
+   value=f"{pct_left}% remaining",
+   state="ok" if pct_left > 25 else "warn",
+   tooltip="Approximate quota remaining for the current billing period."
+  ))
+ err = getattr(ctrl, "last_error", None) if ctrl else None
+ out.append(VendorMetric(
+  key="last_error", label="Last Error",
+  value=err.get("class", "none") if err else "none",
+  state="error" if err else "ok",
+  tooltip=err.get("message", "No error since session start.") if err else "No error since session start."
+ ))
+ return out

tests/test_vendor_state.py

@@ -0,0 +1,56 @@
+from src.vendor_state import get_vendor_state, VendorMetric
+
+class _StubTT:
+ def __init__(self, used=0, limit=0, cache_hits=0, cache_misses=0):
+  self.used = used
+  self.limit = limit
+  self.cache_hits = cache_hits
+  self.cache_misses = cache_misses
+
+class _StubController:
+ def __init__(self, token_tracker=None, last_error=None, vendor_quota=None):
+  self.token_tracker = token_tracker
+  self.last_error = last_error
+  self.vendor_quota = vendor_quota if vendor_quota is not None else {}
+
+class _StubApp:
+ def __init__(self, controller):
+  self.current_provider = "Anthropic"
+  self.current_model = "claude-opus-4"
+  self.controller = controller
+
+def test_get_vendor_state_returns_core_metrics():
+ ctrl = _StubController(token_tracker=_StubTT(used=78234, limit=200000, cache_hits=1200, cache_misses=80), vendor_quota={"remaining_pct": 87})
+ app = _StubApp(ctrl)
+ metrics = get_vendor_state(app)
+ keys = {m.key for m in metrics}
+ assert "provider_model" in keys
+ assert "context_window" in keys
+ assert "cache" in keys
+ assert "quota" in keys
+ assert "last_error" in keys
+
+def test_missing_data_renders_em_dash_not_crash():
+ ctrl = _StubController(token_tracker=None, last_error=None, vendor_quota={})
+ app = _StubApp(ctrl)
+ metrics = get_vendor_state(app)
+ assert len(metrics) > 0
+ for m in metrics:
+  assert m.value is not None
+  assert m.value != ""
+
+def test_context_window_state_warn_above_75_percent():
+ ctrl = _StubController(token_tracker=_StubTT(used=160000, limit=200000, cache_hits=0, cache_misses=0))
+ app = _StubApp(ctrl)
+ metrics = get_vendor_state(app)
+ cw = next(m for m in metrics if m.key == "context_window")
+ assert cw.state == "warn"
+ assert "80%" in cw.value
+
+def test_last_error_state_error_when_present():
+ ctrl = _StubController(token_tracker=None, last_error={"class": "ProviderError", "message": "rate limit"})
+ app = _StubApp(ctrl)
+ metrics = get_vendor_state(app)
+ err = next(m for m in metrics if m.key == "last_error")
+ assert err.state == "error"
+ assert "ProviderError" in err.value