fix: show live quotas in fixed account cards

This commit is contained in:
Hermes Team 2026-08-22 23:45:37 +07:00
parent 40b3466558
commit 7d77877358
6 changed files with 288 additions and 108 deletions

View file

@ -150,6 +150,11 @@ class AccountQuotaService:
except Exception as e: except Exception as e:
logger.warning("Error fetching quota for %s/%s: %s", provider, profile_id, e) logger.warning("Error fetching quota for %s/%s: %s", provider, profile_id, e)
snap = self._generate_baseline_snapshot(provider, profile_id) snap = self._generate_baseline_snapshot(provider, profile_id)
status = getattr(e, "status", None)
if status == 401 or "401" in str(e) or "unauthenticated" in str(e).lower():
snap.unavailable_reason = "Авторизация истекла — обновите подключение"
else:
snap.unavailable_reason = "Провайдер не вернул данные лимитов"
with self._cache_lock: with self._cache_lock:
self._snapshots[key] = snap self._snapshots[key] = snap
@ -330,9 +335,32 @@ class AccountQuotaService:
if not access_token: if not access_token:
raise RuntimeError("В профиле Antigravity отсутствует access token") raise RuntimeError("В профиле Antigravity отсутствует access token")
refresh_token = token_data.get("refresh_token") or token_data.get("refresh")
def _refresh_and_save() -> str:
if not refresh_token:
raise RuntimeError("OAuth-сессия истекла, refresh token отсутствует")
refreshed = refresh_access_token(str(refresh_token))
token_data.update(refreshed)
auth_data["token"] = token_data
ProfileAuthManager.save_profile_auth("antigravity", profile_id, auth_data)
return str(refreshed["access_token"])
expiry = _parse_datetime(token_data.get("expires_at") or token_data.get("expiry"))
if expiry and expiry <= now + timedelta(seconds=60):
access_token = _refresh_and_save()
project_id = auth_data.get("project_id") or auth_data.get("projectId") project_id = auth_data.get("project_id") or auth_data.get("projectId")
if not project_id: if not project_id:
project_id = load_or_onboard_project(str(access_token)) try:
project_id = load_or_onboard_project(str(access_token))
except Exception as exc:
if (getattr(exc, "status", None) != 401 and "401" not in str(exc)) or not refresh_token:
raise
access_token = _refresh_and_save()
project_id = load_or_onboard_project(str(access_token))
auth_data["project_id"] = project_id
ProfileAuthManager.save_profile_auth("antigravity", profile_id, auth_data)
def _fetch(token: str) -> dict[str, Any]: def _fetch(token: str) -> dict[str, Any]:
request = urllib.request.Request( request = urllib.request.Request(
@ -351,15 +379,9 @@ class AccountQuotaService:
try: try:
payload = _fetch(str(access_token)) payload = _fetch(str(access_token))
except urllib.error.HTTPError as exc: except urllib.error.HTTPError as exc:
refresh_token = token_data.get("refresh_token") or token_data.get("refresh")
if exc.code != 401 or not refresh_token: if exc.code != 401 or not refresh_token:
raise raise
refreshed = refresh_access_token(str(refresh_token)) payload = _fetch(_refresh_and_save())
token_data.update(refreshed)
auth_data["token"] = token_data
auth_data["project_id"] = project_id
ProfileAuthManager.save_profile_auth("antigravity", profile_id, auth_data)
payload = _fetch(str(refreshed["access_token"]))
models = payload.get("models") or {} models = payload.get("models") or {}
if not isinstance(models, dict): if not isinstance(models, dict):
@ -440,44 +462,85 @@ class AccountQuotaService:
) )
def _collect_opencode_quota(self, profile_id: str, auth_data: dict) -> QuotaSnapshot: def _collect_opencode_quota(self, profile_id: str, auth_data: dict) -> QuotaSnapshot:
"""Collect Sliding, Weekly, and Monthly usage for OpenCode Go.""" """Validate OpenCode Go entitlement and read usage when its API exposes it."""
now = _utc_now() now = _utc_now()
b_sliding = QuotaBucket( api_key = auth_data.get("api_key")
id="opencode.sliding", if not api_key:
display_name="Скользящее", raise RuntimeError("Ключ OpenCode Go не сохранён")
model_family="opencode",
used_percent=None, base_url = "https://opencode.ai/zen/go/v1"
remaining_percent=None,
period="sliding", def _get(path: str) -> dict[str, Any]:
status="unknown", request = urllib.request.Request(
) f"{base_url}{path}",
b_weekly = QuotaBucket( headers={
id="opencode.weekly", "Authorization": f"Bearer {api_key}",
display_name="Недельное", "Accept": "application/json",
model_family="opencode", "User-Agent": "hermes-hub/1.0",
used_percent=None, },
remaining_percent=None, )
period="7d", with urllib.request.urlopen(request, timeout=20) as response:
reset_at=now + timedelta(days=7), return json.loads(response.read().decode("utf-8") or "{}")
status="unknown",
) # /models is the documented read-only endpoint and confirms that the
b_monthly = QuotaBucket( # key is accepted without spending a request from the user's limit.
id="opencode.monthly", _get("/models")
display_name="Ежемесячное", usage: dict[str, Any] = {}
model_family="opencode", unavailable_reason: Optional[str] = None
used_percent=None, try:
remaining_percent=None, usage = _get("/usage")
period="30d", except urllib.error.HTTPError as exc:
reset_at=now + timedelta(days=30), raw = exc.read().decode("utf-8", "replace")
status="unknown", if exc.code == 403 and "subscription required" in raw.lower():
unavailable_reason = "Для этого ключа не активна подписка OpenCode Go"
elif exc.code in (403, 404):
unavailable_reason = "OpenCode Go не предоставляет остаток через публичный API"
else:
raise
def _metric(*names: str) -> dict[str, Any]:
for name in names:
value = usage.get(name)
if isinstance(value, dict):
return value
return {}
buckets: list[QuotaBucket] = []
specs = (
("5h", "Лимит 5 часов", 12, _metric("five_hour", "fiveHour", "sliding")),
("7d", "Недельный лимит", 30, _metric("weekly", "seven_day", "sevenDay")),
("30d", "Месячный лимит", 60, _metric("monthly", "thirty_day", "thirtyDay")),
) )
for period, label, limit_value, metric in specs:
remaining_percent = metric.get("remaining_percent", metric.get("remainingPercentage"))
remaining_absolute = metric.get("remaining", metric.get("remaining_amount"))
used_absolute = metric.get("used", metric.get("used_amount"))
reset_at = _parse_datetime(metric.get("reset_at") or metric.get("resetTime"))
buckets.append(
QuotaBucket(
id=f"opencode.{period}",
display_name=label,
model_family="opencode",
remaining_percent=float(remaining_percent) if isinstance(remaining_percent, (int, float)) else None,
used_absolute=int(used_absolute) if isinstance(used_absolute, (int, float)) else None,
remaining_absolute=(
int(remaining_absolute) if isinstance(remaining_absolute, (int, float)) else None
),
limit_absolute=limit_value,
reset_at=reset_at,
period=period,
unit="USD",
scope="account",
)
)
return QuotaSnapshot( return QuotaSnapshot(
account_id=profile_id, account_id=profile_id,
provider="opencode-go", provider="opencode-go",
buckets=[b_sliding, b_weekly, b_monthly], buckets=buckets,
fetched_at=now, fetched_at=now,
source="baseline", source="provider_api",
unavailable_reason=unavailable_reason,
) )
def _collect_claude_quota(self, profile_id: str, auth_data: dict) -> QuotaSnapshot: def _collect_claude_quota(self, profile_id: str, auth_data: dict) -> QuotaSnapshot:

View file

@ -868,7 +868,7 @@ class AccountCardWidget(HubCard):
self.profile_id = profile_id self.profile_id = profile_id
self.profile_model: Any = None self.profile_model: Any = None
self.on_action = on_action self.on_action = on_action
self.compact = compact self.compact = True
self._quota_widgets: Dict[str, QuotaBucketWidget] = {} self._quota_widgets: Dict[str, QuotaBucketWidget] = {}
self.widgets_created = 0 self.widgets_created = 0
self.widgets_destroyed = 0 self.widgets_destroyed = 0
@ -877,8 +877,8 @@ class AccountCardWidget(HubCard):
top.pack(fill="x", padx=Theme.CARD_PAD_X, pady=(Theme.CARD_PAD_Y, Theme.SPACE_XS)) top.pack(fill="x", padx=Theme.CARD_PAD_X, pady=(Theme.CARD_PAD_Y, Theme.SPACE_XS))
self.provider = ctk.CTkLabel(top, text=provider, font=Theme.font_micro(), text_color=Theme.TEXT_MUTED) self.provider = ctk.CTkLabel(top, text=provider, font=Theme.font_micro(), text_color=Theme.TEXT_MUTED)
self.provider.pack(side="left") self.provider.pack(side="left")
self.toggle = IconButton(top, text="", command=self.toggle_compact) self.profile_mark = ctk.CTkLabel(top, text="", font=Theme.font_caption(), text_color=Theme.TEXT_ACCENT)
self.toggle.pack(side="right") self.profile_mark.pack(side="right")
self.identity = EllipsizedLabel(self, text=identity, font=Theme.font_body_bold(), text_color=Theme.TEXT_PRIMARY) self.identity = EllipsizedLabel(self, text=identity, font=Theme.font_body_bold(), text_color=Theme.TEXT_PRIMARY)
self.identity.pack(anchor="w", padx=Theme.CARD_PAD_X) self.identity.pack(anchor="w", padx=Theme.CARD_PAD_X)
@ -889,21 +889,37 @@ class AccountCardWidget(HubCard):
self.status.pack(anchor="w", padx=Theme.CARD_PAD_X, pady=Theme.SPACE_SM) self.status.pack(anchor="w", padx=Theme.CARD_PAD_X, pady=Theme.SPACE_SM)
self.compact_quota = ctk.CTkFrame(self, fg_color="transparent") self.compact_quota = ctk.CTkFrame(self, fg_color="transparent")
self.compact_quota_labels = [ self.compact_quota_cells: list[dict[str, Any]] = []
ctk.CTkLabel( for index in range(4):
self.compact_quota, cell = ctk.CTkFrame(self.compact_quota, fg_color="transparent")
text="", cell.grid(row=index // 2, column=index % 2, sticky="nsew", padx=(0, 10), pady=(2, 6))
font=Theme.font_micro(), title = ctk.CTkLabel(cell, text="", font=Theme.font_micro(), text_color=Theme.TEXT_PRIMARY, anchor="w")
text_color=Theme.TEXT_SECONDARY, title.pack(fill="x")
anchor="w", value = ctk.CTkLabel(cell, text="", font=Theme.font_caption(), text_color=Theme.TEXT_SECONDARY, anchor="e")
value.pack(fill="x")
progress = ctk.CTkProgressBar(
cell,
height=5,
corner_radius=3,
fg_color=Theme.SURFACE_MUTED,
progress_color=Theme.STATUS_HEALTHY,
)
progress.pack(fill="x", pady=(2, 1))
progress.set(0)
reset = ctk.CTkLabel(cell, text="", font=Theme.font_micro(), text_color=Theme.TEXT_MUTED, anchor="w")
reset.pack(fill="x")
self.compact_quota_cells.append(
{"frame": cell, "title": title, "value": value, "progress": progress, "reset": reset}
) )
for _ in range(4)
]
for index, label in enumerate(self.compact_quota_labels):
label.grid(row=index // 2, column=index % 2, sticky="ew", padx=(0, 8), pady=2)
self.compact_quota.grid_columnconfigure(index % 2, weight=1) self.compact_quota.grid_columnconfigure(index % 2, weight=1)
self.compact_actions = ctk.CTkFrame(self, fg_color="transparent") self.compact_actions = ctk.CTkFrame(self, fg_color="transparent")
for text, action in (("", "test"), ("", "refresh_account"), ("Роль", "assign_role")): for text, action in (
("", "test"),
("", "set_main"),
("", "set_orchestrator"),
("Роль", "assign_role"),
("", "refresh_account"),
):
ActionButton( ActionButton(
self.compact_actions, self.compact_actions,
text=text, text=text,
@ -961,7 +977,7 @@ class AccountCardWidget(HubCard):
width=72, width=72,
command=lambda: self._trigger("delete_credentials"), command=lambda: self._trigger("delete_credentials"),
).pack(side="right") ).pack(side="right")
self.set_compact(compact) self.set_compact(True)
@staticmethod @staticmethod
def resolve_identity(profile: Any) -> str: def resolve_identity(profile: Any) -> str:
@ -992,19 +1008,14 @@ class AccountCardWidget(HubCard):
self.action_feedback.pack(fill="x", pady=(0, Theme.SPACE_XS), before=self.actions.winfo_children()[1]) self.action_feedback.pack(fill="x", pady=(0, Theme.SPACE_XS), before=self.actions.winfo_children()[1])
def toggle_compact(self) -> None: def toggle_compact(self) -> None:
self.set_compact(not self.compact) # Account cards intentionally have one fixed Cockpit-style layout.
self.set_compact(True)
def set_compact(self, compact: bool) -> None: def set_compact(self, compact: bool) -> None:
self.compact = compact self.compact = True
self.toggle.configure(text="" if compact else "") self.details.pack_forget()
if compact: self.compact_quota.pack(fill="x", padx=Theme.CARD_PAD_X, pady=(0, Theme.SPACE_XS))
self.details.pack_forget() self.compact_actions.pack(fill="x", padx=Theme.CARD_PAD_X, pady=(0, Theme.CARD_PAD_Y))
self.compact_quota.pack(fill="x", padx=Theme.CARD_PAD_X, pady=(0, Theme.SPACE_XS))
self.compact_actions.pack(fill="x", padx=Theme.CARD_PAD_X, pady=(0, Theme.CARD_PAD_Y))
else:
self.compact_quota.pack_forget()
self.compact_actions.pack_forget()
self.details.pack(fill="x")
def update_account(self, profile: Any, quota_snapshot: Optional[Any] = None) -> None: def update_account(self, profile: Any, quota_snapshot: Optional[Any] = None) -> None:
self.profile_model = profile self.profile_model = profile
@ -1024,28 +1035,53 @@ class AccountCardWidget(HubCard):
) )
else: else:
self.plan_badge.pack_forget() self.plan_badge.pack_forget()
self.status.set_status(profile.health_state, getattr(profile, "health_label_ru", None))
self.configure(border_color=Theme.BORDER_ACCENT if profile.is_main_account else Theme.BORDER) self.configure(border_color=Theme.BORDER_ACCENT if profile.is_main_account else Theme.BORDER)
snapshot = quota_snapshot or getattr(profile, "quota_snapshot", None) snapshot = quota_snapshot or getattr(profile, "quota_snapshot", None)
buckets = list(getattr(snapshot, "buckets", None) or []) buckets = list(getattr(snapshot, "buckets", None) or [])
estimated = bool(getattr(snapshot, "is_estimated", True)) if snapshot else True estimated = bool(getattr(snapshot, "is_estimated", True)) if snapshot else True
for index, label in enumerate(self.compact_quota_labels): unavailable_reason = getattr(snapshot, "unavailable_reason", None) if snapshot else None
measured_remaining = [
float(bucket.remaining_percent)
for bucket in buckets
if getattr(bucket, "remaining_percent", None) is not None
]
if measured_remaining and not estimated:
if all(remaining <= 0 for remaining in measured_remaining):
card_health, card_label = "quota_exhausted", "Квота исчерпана"
elif any(remaining <= 0 for remaining in measured_remaining):
card_health, card_label = "warning", "Часть квот исчерпана"
else:
card_health, card_label = "healthy", "Работает"
else:
card_health = profile.health_state
card_label = getattr(profile, "health_label_ru", None)
self.status.set_status(card_health, card_label)
for index, cell in enumerate(self.compact_quota_cells):
if index < len(buckets): if index < len(buckets):
bucket = buckets[index] bucket = buckets[index]
label.configure( remaining = getattr(bucket, "remaining_percent", None)
text=f"{bucket.display_name}: {bucket.formatted_remaining()}", color = (
text_color=( Theme.STATUS_HEALTHY
Theme.STATUS_HEALTHY if getattr(bucket, "status", "unknown") == "healthy"
if getattr(bucket, "status", "unknown") == "healthy" else Theme.STATUS_WARNING
else Theme.STATUS_WARNING if getattr(bucket, "status", "unknown") == "warning"
if getattr(bucket, "status", "unknown") == "warning" else Theme.TEXT_MUTED
else Theme.TEXT_MUTED
),
) )
label.grid() detail = bucket.formatted_remaining()
if detail == "Н" and unavailable_reason:
limit = getattr(bucket, "limit_absolute", None)
unit = getattr(bucket, "unit", None)
prefix = f"Лимит ${limit}" if limit is not None and unit == "USD" else ""
detail = f"{prefix}{unavailable_reason}"
cell["title"].configure(text=bucket.display_name)
cell["value"].configure(text=detail, text_color=color)
cell["progress"].configure(progress_color=color)
cell["progress"].set(float(remaining) / 100.0 if remaining is not None else 0)
cell["reset"].configure(text=bucket.formatted_reset() or "Период указан провайдером")
cell["frame"].grid()
else: else:
label.grid_remove() cell["frame"].grid_remove()
seen: set[str] = set() seen: set[str] = set()
for bucket in buckets: for bucket in buckets:
key = str(getattr(bucket, "id", "") or getattr(bucket, "display_name", "bucket")) key = str(getattr(bucket, "id", "") or getattr(bucket, "display_name", "bucket"))

View file

@ -29,7 +29,7 @@ PROVIDER_LABELS = {
class ProviderGroup(ctk.CTkFrame): class ProviderGroup(ctk.CTkFrame):
"""Collapsible provider section that retains its account cards while hidden.""" """Fixed provider section with always-visible Cockpit-style cards."""
def __init__(self, master: Any, provider: str): def __init__(self, master: Any, provider: str):
super().__init__(master, fg_color="transparent") super().__init__(master, fg_color="transparent")
@ -37,21 +37,13 @@ class ProviderGroup(ctk.CTkFrame):
self.collapsed = False self.collapsed = False
self.header = ctk.CTkFrame(self, fg_color=Theme.BG_HEADER, corner_radius=Theme.RADIUS_SM) self.header = ctk.CTkFrame(self, fg_color=Theme.BG_HEADER, corner_radius=Theme.RADIUS_SM)
self.header.pack(fill="x", pady=(Theme.SPACE_SM, Theme.SPACE_XS)) self.header.pack(fill="x", pady=(Theme.SPACE_SM, Theme.SPACE_XS))
self.toggle = ActionButton(
self.header,
text="",
variant="ghost",
width=34,
command=self.toggle_collapsed,
)
self.toggle.pack(side="left", padx=(Theme.SPACE_SM, 0))
self.title = ctk.CTkLabel( self.title = ctk.CTkLabel(
self.header, self.header,
text=PROVIDER_LABELS.get(provider, provider), text=PROVIDER_LABELS.get(provider, provider),
font=Theme.font_heading(), font=Theme.font_heading(),
text_color=Theme.TEXT_PRIMARY, text_color=Theme.TEXT_PRIMARY,
) )
self.title.pack(side="left", padx=Theme.SPACE_SM, pady=Theme.SPACE_SM) self.title.pack(side="left", padx=Theme.SPACE_MD, pady=Theme.SPACE_SM)
self.count = ctk.CTkLabel(self.header, text="0", font=Theme.font_caption(), text_color=Theme.TEXT_MUTED) self.count = ctk.CTkLabel(self.header, text="0", font=Theme.font_caption(), text_color=Theme.TEXT_MUTED)
self.count.pack(side="right", padx=Theme.SPACE_MD) self.count.pack(side="right", padx=Theme.SPACE_MD)
self.body = ctk.CTkFrame(self, fg_color="transparent") self.body = ctk.CTkFrame(self, fg_color="transparent")
@ -60,12 +52,8 @@ class ProviderGroup(ctk.CTkFrame):
self.body.grid_columnconfigure(column, weight=1) self.body.grid_columnconfigure(column, weight=1)
def toggle_collapsed(self) -> None: def toggle_collapsed(self) -> None:
self.collapsed = not self.collapsed self.collapsed = False
self.toggle.configure(text="" if self.collapsed else "") self.body.pack(fill="x")
if self.collapsed:
self.body.pack_forget()
else:
self.body.pack(fill="x")
class AccountsView(ctk.CTkFrame): class AccountsView(ctk.CTkFrame):
@ -91,7 +79,7 @@ class AccountsView(ctk.CTkFrame):
header = SectionHeader( header = SectionHeader(
self, self,
title="Аккаунты и квоты", title="Аккаунты и квоты",
subtitle="Реальные идентичности, независимые лимитные корзины и резервные роли", subtitle="Компактные карточки аккаунтов, реальные остатки и быстрые действия",
action_text="+ Добавить аккаунт", action_text="+ Добавить аккаунт",
action_cmd=lambda: self._emit("add_account", {}), action_cmd=lambda: self._emit("add_account", {}),
) )
@ -203,7 +191,6 @@ class AccountsView(ctk.CTkFrame):
return return
self._snapshot = snapshot self._snapshot = snapshot
live_ids = {profile.profile_id for profile in self._profiles()} live_ids = {profile.profile_id for profile in self._profiles()}
collapse_new_collection = len(self._cards) <= 1 and len(live_ids) > 1
for profile_id in list(self._cards): for profile_id in list(self._cards):
if profile_id not in live_ids: if profile_id not in live_ids:
self._cards.pop(profile_id).destroy() self._cards.pop(profile_id).destroy()
@ -221,15 +208,12 @@ class AccountsView(ctk.CTkFrame):
profile.profile_id, profile.profile_id,
AccountCardWidget.resolve_identity(profile), AccountCardWidget.resolve_identity(profile),
profile.provider_display_name, profile.provider_display_name,
compact=len(live_ids) > 1, compact=True,
on_action=self._emit, on_action=self._emit,
) )
self._cards[profile.profile_id] = card self._cards[profile.profile_id] = card
self.cards_created += 1 self.cards_created += 1
card.update_account(profile, snapshot.quotas.get(profile.profile_id)) card.update_account(profile, snapshot.quotas.get(profile.profile_id))
if collapse_new_collection:
for card in self._cards.values():
card.set_compact(True)
self._render_visibility() self._render_visibility()
def _render_visibility(self) -> None: def _render_visibility(self) -> None:

View file

@ -83,10 +83,10 @@ class _EndpointCard(HubCard):
self.subtitle.pack(anchor="w", pady=(0, 1)) self.subtitle.pack(anchor="w", pady=(0, 1))
self.status = ctk.CTkLabel(text, text="", font=Theme.font_micro(), text_color=Theme.STATUS_HEALTHY) self.status = ctk.CTkLabel(text, text="", font=Theme.font_micro(), text_color=Theme.STATUS_HEALTHY)
self.status.pack(anchor="w") self.status.pack(anchor="w")
quota = ctk.CTkFrame(self, fg_color="transparent", width=58) quota = ctk.CTkFrame(self, fg_color="transparent", width=92)
quota.pack(side="right", fill="y", padx=(3, 8), pady=8) quota.pack(side="right", fill="y", padx=(3, 8), pady=8)
quota.pack_propagate(False) quota.pack_propagate(False)
self.quota_label = ctk.CTkLabel(quota, text="Н", font=Theme.font_micro(), text_color=Theme.TEXT_SECONDARY) self.quota_label = ctk.CTkLabel(quota, text="", font=Theme.font_micro(), text_color=Theme.TEXT_SECONDARY)
self.quota_label.pack(anchor="e") self.quota_label.pack(anchor="e")
self.progress = ctk.CTkProgressBar( self.progress = ctk.CTkProgressBar(
quota, height=4, corner_radius=2, progress_color=Theme.STATUS_HEALTHY, fg_color=Theme.SURFACE_MUTED quota, height=4, corner_radius=2, progress_color=Theme.STATUS_HEALTHY, fg_color=Theme.SURFACE_MUTED
@ -195,7 +195,10 @@ class _RouteDiagram(ctk.CTkFrame):
self.context, text="▤ Хранилище контекста", font=Theme.font_caption(), text_color=Theme.TEXT_PRIMARY self.context, text="▤ Хранилище контекста", font=Theme.font_caption(), text_color=Theme.TEXT_PRIMARY
).pack(pady=(6, 0)) ).pack(pady=(6, 0))
self.context_status = ctk.CTkLabel( self.context_status = ctk.CTkLabel(
self.context, text="● Состояние: Н", font=Theme.font_micro(), text_color=Theme.TEXT_MUTED self.context,
text="● Нет телеметрии хранилища",
font=Theme.font_micro(),
text_color=Theme.TEXT_MUTED,
) )
self.context_status.pack() self.context_status.pack()
self._left_labels = ["", "", ""] self._left_labels = ["", "", ""]
@ -513,7 +516,7 @@ class DashboardView(ctk.CTkFrame):
for bucket in quota.buckets: for bucket in quota.buckets:
if bucket.remaining_percent is not None: if bucket.remaining_percent is not None:
return f"{bucket.remaining_percent:.0f}%", float(bucket.remaining_percent) return f"{bucket.remaining_percent:.0f}%", float(bucket.remaining_percent)
return "Н", None return "Нет данных API", None
@staticmethod @staticmethod
def _sync_endpoint_cards( def _sync_endpoint_cards(
@ -633,7 +636,7 @@ class DashboardView(ctk.CTkFrame):
agent.role_name_ru, agent.role_name_ru,
f"{agent.provider_display_name}{agent.model}", f"{agent.provider_display_name}{agent.model}",
"Здорово" if agent.is_active else agent.status_label_ru, "Здорово" if agent.is_active else agent.status_label_ru,
agent.active_quota_label or "Н", agent.active_quota_label or "Нет данных API",
None, None,
"healthy" if agent.is_active else "warning", "healthy" if agent.is_active else "warning",
) )
@ -658,12 +661,12 @@ class DashboardView(ctk.CTkFrame):
left_labels: list[str] = [] left_labels: list[str] = []
for provider in providers: for provider in providers:
share = dict(provider_telemetry.get(provider.provider_id) or {}).get("call_share") share = dict(provider_telemetry.get(provider.provider_id) or {}).get("call_share")
left_labels.append(f"{share:.0%}" if share is not None else "Н") left_labels.append(f"{share:.0%}" if share is not None else "")
right_labels: list[str] = [] right_labels: list[str] = []
for agent in agents: for agent in agents:
measured = dict(role_telemetry.get(agent.role_id) or {}) measured = dict(role_telemetry.get(agent.role_id) or {})
calls = measured.get("total_calls") if measured.get("has_data") else None calls = measured.get("total_calls") if measured.get("has_data") else None
right_labels.append(f"{calls} выз." if calls is not None else "Н") right_labels.append(f"{calls} выз." if calls is not None else "")
self.route_diagram.set_labels(left_labels, right_labels) self.route_diagram.set_labels(left_labels, right_labels)
live_provider_ids = {provider.provider_id for provider in providers} live_provider_ids = {provider.provider_id for provider in providers}

View file

@ -2,9 +2,11 @@
from __future__ import annotations from __future__ import annotations
import json import json
import io
import os import os
import tempfile import tempfile
import time import time
import urllib.error
from datetime import datetime, timedelta, timezone from datetime import datetime, timedelta, timezone
from pathlib import Path from pathlib import Path
from unittest.mock import MagicMock, patch from unittest.mock import MagicMock, patch
@ -222,6 +224,75 @@ def test_antigravity_separate_claude_and_gemini_buckets():
assert b_g.remaining_percent == 87.0 assert b_g.remaining_percent == 87.0
def test_antigravity_refreshes_expired_token_before_project_discovery():
service = AccountQuotaService()
response = MagicMock()
response.__enter__.return_value.read.return_value = json.dumps(
{
"models": {
"claude-sonnet-4-6": {
"quotaInfo": {"remainingFraction": 0.64, "resetTime": "2026-08-23T00:00:00Z"}
},
"gemini-3.7-flash": {
"quotaInfo": {"remainingFraction": 0.91, "resetTime": "2026-08-23T00:00:00Z"}
},
}
}
).encode()
auth_data = {
"token": {
"access_token": "expired-token",
"refresh_token": "refresh-token",
"expires_at": "2026-08-22T00:00:00Z",
}
}
with patch(
"antigravity_provider.oauth.refresh_access_token",
return_value={"access_token": "fresh-token", "expires_at": "2026-08-23T00:00:00Z"},
) as refresh, patch(
"antigravity_provider.cloudcode.load_or_onboard_project", return_value="project-1"
) as discover, patch(
"antigravity_provider.router.quota_collector.ProfileAuthManager.save_profile_auth"
) as save, patch(
"antigravity_provider.router.quota_collector.urllib.request.urlopen", return_value=response
) as urlopen:
snapshot = service._collect_antigravity_quota("ag-w1", auth_data)
refresh.assert_called_once_with("refresh-token")
discover.assert_called_once_with("fresh-token")
assert "Bearer fresh-token" == urlopen.call_args.args[0].headers["Authorization"]
assert save.called
assert auth_data["project_id"] == "project-1"
assert snapshot.source == "provider_api"
assert snapshot.get_bucket_for_model("claude-sonnet-4-6").remaining_percent == 64.0
def test_opencode_shows_published_limits_and_subscription_error():
service = AccountQuotaService()
models_response = MagicMock()
models_response.__enter__.return_value.read.return_value = b'{"data": []}'
entitlement_error = urllib.error.HTTPError(
"https://opencode.ai/zen/go/v1/usage",
403,
"Forbidden",
{},
io.BytesIO(b'{"message":"OpenCode Go subscription required"}'),
)
with patch(
"antigravity_provider.router.quota_collector.urllib.request.urlopen",
side_effect=[models_response, entitlement_error],
):
snapshot = service._collect_opencode_quota("opengo-1", {"api_key": "test-key"})
assert snapshot.source == "provider_api"
assert snapshot.unavailable_reason == "Для этого ключа не активна подписка OpenCode Go"
assert [bucket.limit_absolute for bucket in snapshot.buckets] == [12, 30, 60]
assert [bucket.period for bucket in snapshot.buckets] == ["5h", "7d", "30d"]
assert all(bucket.remaining_percent is None for bucket in snapshot.buckets)
def test_health_tracker_antigravity_claude_exhaustion_does_not_block_gemini(tmp_path): def test_health_tracker_antigravity_claude_exhaustion_does_not_block_gemini(tmp_path):
state_file = tmp_path / "router_state.json" state_file = tmp_path / "router_state.json"
tracker = HealthTracker(state_file=state_file) tracker = HealthTracker(state_file=state_file)

View file

@ -97,6 +97,29 @@ def test_plan_badge_distinguishes_trusted_inferred_and_unknown(ui_root) -> None:
card.destroy() card.destroy()
@pytest.mark.ui
def test_live_quota_overrides_stale_exhausted_card_status(ui_root) -> None:
card = AccountCardWidget(ui_root, "account-1", "user@example.test", "Antigravity")
profile = replace(_profile(), health_state="quota_exhausted", health_label_ru="Квота исчерпана")
snapshot = QuotaSnapshot(
account_id="account-1",
provider="antigravity",
source="provider_api",
buckets=[
QuotaBucket(id="claude", display_name="Claude", remaining_percent=100.0),
QuotaBucket(id="gemini", display_name="Gemini", remaining_percent=100.0),
],
)
try:
card.pack()
card.update_account(profile, snapshot)
ui_root.update_idletasks()
assert card.status.label.cget("text") == "Работает"
assert card.status.dot.cget("text_color") == Theme.STATUS_HEALTHY
finally:
card.destroy()
@pytest.mark.ui @pytest.mark.ui
def test_quota_missing_is_not_rendered_as_zero_and_reason_is_visible(ui_root) -> None: def test_quota_missing_is_not_rendered_as_zero_and_reason_is_visible(ui_root) -> None:
widget = QuotaBucketWidget(ui_root, "bucket", "Claude 5h") widget = QuotaBucketWidget(ui_root, "bucket", "Claude 5h")