diff --git a/src/vendor_state.py b/src/vendor_state.py new file mode 100644 index 00000000..3ac94c80 --- /dev/null +++ b/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 diff --git a/tests/test_vendor_state.py b/tests/test_vendor_state.py new file mode 100644 index 00000000..c7a187ec --- /dev/null +++ b/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