Private
Public Access
0
0

feat(vendor-state): add vendor_quota/last_error/token_tracker to AppController

- AppController.__init__: public vendor_quota: Dict[str,Any], last_error: Optional[Dict[str,str]], token_tracker: Dict[str,Any]
- set_vendor_quota(provider, remaining_pct, reset_at): public API for ai_client quota paths
- clear_last_error(): reset hook
- _refresh_api_metrics: read vendor_quota and error from payload, populate state

ai_client per-provider quota wire-up deferred to a future track (per-provider
signals differ; this commit establishes the state shape and read path).
This commit is contained in:
2026-06-03 11:53:56 -04:00
parent 375e30e427
commit 7aeed5e01f
+20
View File
@@ -863,6 +863,9 @@ class AppController:
self._tier_stream_last_len: Dict[str, int] = {}
self._current_session_usage = None
self._current_mma_tier_usage = None
self.vendor_quota: Dict[str, Any] = {}
self.last_error: Optional[Dict[str, str]] = None
self.token_tracker: Dict[str, Any] = {"used": 0, "limit": 0, "cache_hits": 0, "cache_misses": 0}
self._current_token_history = None
self._current_session_start_time = None
self._inject_file_path: str = ""
@@ -3356,8 +3359,25 @@ class AppController:
count = cache_stats.get("cache_count", 0)
size_bytes = cache_stats.get("total_size_bytes", 0)
self._gemini_cache_text = f"Gemini Caches: {count} ({size_bytes / 1024:.1f} KB)"
quota = payload.get("vendor_quota")
if isinstance(quota, dict) and quota:
self.vendor_quota = quota
if "error" in payload and isinstance(payload["error"], dict):
self.last_error = payload["error"]
self._update_cached_stats()
def set_vendor_quota(self, provider: str, remaining_pct: float, reset_at: str = "") -> None:
"""Update vendor quota state from a quota-bearing API response.
[C: src/vendor_state.py:get_vendor_state]
"""
self.vendor_quota = {"provider": provider, "remaining_pct": remaining_pct, "reset_at": reset_at}
def clear_last_error(self) -> None:
"""Reset last_error after a successful response cycle.
[C: src/vendor_state.py:get_vendor_state]
"""
self.last_error = None
def _update_cached_stats(self) -> None:
from src import ai_client
self._cached_cache_stats = ai_client.get_gemini_cache_stats()