from __future__ import annotations import os import subprocess import sys import tempfile import threading import time import traceback import warnings from dataclasses import replace from pathlib import Path from typing import Any, Callable, TypeVar from xml.sax.saxutils import escape from account_lock import AccountLockError from account_registry import Account, AccountRegistry, AccountRegistryKind, RESERVED_SLUGS, ROUTING_RULES_FILENAMES from claude_auth_operation import ClaudeAuthOperation from claude_health_client import ClaudeHealthClient from device_auth import DeviceAuthFlow, DeviceAuthPrompt, DeviceAuthSession, terminate_process from device_auth_dialog import ClaudeCodeEntryDialog, DeviceAuthDialog from device_auth_operation import DeviceAuthOperation from factory_decision_notification_store import FactoryDecisionNotificationStore from factory_menu import build_factory_menu_items from factory_status import FactoryAvailability, FactoryStatusSnapshot, deck_factory_url, load_factory_status from grok_auth_operation import GrokAuthOperation from grok_health_client import GrokHealthClient from health_client import AccountHealthClient, AccountSnapshot, HealthStatus from health_store import HealthSnapshotStore from quota_notification_store import QuotaNotificationStore from routing_resolver import load_rules import limit_warning from popup_window import PopupWindow from provider_services import CodexHealthAdapter, CodexLifecycleAdapter, ProviderServiceMap, ProviderServices from runtime_paths import runtime_dir from scheduler import RefreshScheduler, default_idle_add, start_background_thread from tray_model import TrayMenuItem, TrayMenuItemKind, format_account_label, format_title_part from ui.view_model import ( AccountInput, AlertKey, DashboardCallbacks, DashboardVM, ProviderInput, build_dashboard_vm, ) APPINDICATOR_ID = os.environ.get( "CODEX_TRAY_APPINDICATOR_ID", "systray-ai", ) APPINDICATOR_ICON = os.environ.get( "CODEX_TRAY_ICON_NAME", "codex-account-switcher", ) APPINDICATOR_ICON_WARNING = os.environ.get( "CODEX_TRAY_ICON_WARNING_NAME", "codex-account-switcher-warning", ) APPINDICATOR_ICON_WARNING_LIVE = os.environ.get( "CODEX_TRAY_ICON_WARNING_LIVE_NAME", "codex-account-switcher-warning-live", ) APP_DISPLAY_NAME = "Systray AI" HEALTH_STALE_SECONDS = 180 def _load_gtk_modules() -> tuple[Any, Any]: import gi gi.require_version("Gtk", "3.0") gi.require_version("AyatanaAppIndicator3", "0.1") from gi.repository import AyatanaAppIndicator3, Gtk return Gtk, AyatanaAppIndicator3 def _load_glib_module() -> Any | None: try: import gi warnings.filterwarnings( "ignore", message="GLib.unix_signal_add_full is deprecated; use GLibUnix.signal_add_full instead", category=DeprecationWarning, ) gi.require_version("GLib", "2.0") from gi.repository import GLib return GLib except Exception: return None _T = TypeVar("_T") def _expired(reset_at: float | None, now: float) -> bool: """A reading describes its window only until that window's reset boundary.""" return reset_at is not None and reset_at <= now def _unexpired(value: _T | None, reset_at: float | None, now: float) -> _T | None: return None if _expired(reset_at, now) else value def _percent_after_reset( used_pct: int | None, reset_at: float | None, now: float ) -> int | None: """Past its boundary a window's counter has zeroed, so the whole window is free.""" if used_pct is None: return None return 0 if _expired(reset_at, now) else used_pct def _runtime_tray_dir() -> Path: directory = runtime_dir() try: directory.mkdir(parents=True, exist_ok=True) return directory except OSError: fallback = Path(tempfile.gettempdir()) / "overdeck-systray" fallback.mkdir(parents=True, exist_ok=True) return fallback class Indicator: def __init__( self, registry: AccountRegistry, health_client: AccountHealthClient, claude_registry: AccountRegistry | None = None, grok_registry: AccountRegistry | None = None, provider_services: ProviderServiceMap | None = None, gtk_module: Any | None = None, appindicator_module: Any | None = None, device_auth_flow: DeviceAuthFlow | None = None, device_auth_operation: DeviceAuthOperation | None = None, dialog_factory: Callable[..., Any] | None = None, idle_add: Callable[..., Any] | None = None, thread_runner: Callable[[Callable[[], None]], Any] | None = None, popup_window_factory: Callable[..., Any] | None = None, dashboard_factory: Callable[..., Any] | None = None, settings_dialog_factory: Callable[..., Any] | None = None, limits_dialog_factory: Callable[..., Any] | None = None, settings_store: Any | None = None, health_snapshot_store: HealthSnapshotStore | None = None, claude_health_snapshot_store: HealthSnapshotStore | None = None, quota_notification_store: QuotaNotificationStore | None = None, factory_status_loader: Callable[[], FactoryStatusSnapshot] | None = None, factory_url_opener: Callable[[str], None] | None = None, factory_decision_notification_store: FactoryDecisionNotificationStore | None = None, ) -> None: self.health_client = health_client self._gtk = gtk_module self._appindicator = appindicator_module self._device_auth_flow = device_auth_flow or DeviceAuthFlow() self._device_auth_operation = device_auth_operation or DeviceAuthOperation( registry, self._device_auth_flow, ) self._provider_services = provider_services or self._build_provider_services( registry, health_client, claude_registry, grok_registry, self._device_auth_operation, ) self.registry = self._provider_services.for_tool(AccountRegistryKind.CODEX).registry self.claude_registry = self._provider_registry(AccountRegistryKind.CLAUDE) self.grok_registry = self._provider_registry(AccountRegistryKind.GROK) self._dialog_factory = dialog_factory or DeviceAuthDialog self._idle_add = idle_add or default_idle_add self._thread_runner = thread_runner or start_background_thread self._popup_window_factory = popup_window_factory or PopupWindow self._dashboard_factory = dashboard_factory self._settings_dialog_factory = settings_dialog_factory self._limits_dialog_factory = limits_dialog_factory self._settings_store = settings_store or self._default_settings_store() self._settings = self._settings_store.read() self._health_store = health_snapshot_store or self._default_health_store() self._claude_health_store = ( claude_health_snapshot_store or self._default_claude_health_store() ) self._grok_health_store = self._default_grok_health_store() self._quota_notification_store = quota_notification_store or QuotaNotificationStore( _runtime_tray_dir() / "quota_notifications.json" ) self._glib: Any | None = None self._time: Callable[[], float] = time.time self._scheduler = RefreshScheduler( idle_add=self._idle_add, thread_runner=self._thread_runner, time_fn=lambda: self._time(), ) self.indicator: Any | None = None self.menu: Any | None = None self.popup: Any | None = None self._dashboard: Any | None = None self._health_timer_id: Any | None = None self.snapshots: dict[str, AccountSnapshot] = {} self._dismissed_alert_keys: set[AlertKey] = set() self._icon_warning_active: bool = False self._warning_icon_remaining: int | None = None self._warning_icon_theme_path: Path = _runtime_tray_dir() self._warning_icon_name: str = APPINDICATOR_ICON_WARNING_LIVE self._warning_icon_path: Path = self._warning_icon_theme_path / f"{self._warning_icon_name}.svg" self.last_fetched: dict[str, float] = {} self._menu_items: dict[str, Any] = {} self._busy_actions: dict[str, set[str]] = {} self._provider_busy_actions: dict[str, set[str]] = {} self._factory_status_loader = factory_status_loader or load_factory_status self._factory_url_opener = factory_url_opener or self._default_factory_url_opener self._factory_decision_notification_store = ( factory_decision_notification_store or FactoryDecisionNotificationStore(_runtime_tray_dir() / "factory_decision_notifications.json") ) self._factory_snapshot: FactoryStatusSnapshot = FactoryStatusSnapshot.unavailable() self._factory_refresh_inflight = False def build(self) -> None: gtk, appindicator = self._modules() self.indicator = appindicator.Indicator.new( APPINDICATOR_ID, APPINDICATOR_ICON, appindicator.IndicatorCategory.APPLICATION_STATUS, ) self.indicator.set_status(appindicator.IndicatorStatus.ACTIVE) if hasattr(self.indicator, "connect"): try: self.indicator.connect("activate", lambda *_args: self._toggle_popup()) except Exception: pass self._load_health_cache() self._rebuild_menu() self._update_title() self._refresh_all(force=True) glib = self._glib_module() if glib is not None and hasattr(glib, "timeout_add_seconds"): self._health_timer_id = glib.timeout_add_seconds( self._settings.refresh_interval_seconds, self._refresh_all_timer ) def run(self) -> None: gtk, _appindicator = self._modules() gtk.main() def _modules(self) -> tuple[Any, Any]: if self._gtk is None or self._appindicator is None: self._gtk, self._appindicator = _load_gtk_modules() return self._gtk, self._appindicator @staticmethod def _build_provider_services( registry: AccountRegistry, health_client: AccountHealthClient, claude_registry: AccountRegistry | None, grok_registry: AccountRegistry | None, device_auth_operation: DeviceAuthOperation, ) -> ProviderServiceMap: providers = [ ProviderServices( tool=AccountRegistryKind.CODEX, registry=registry, lifecycle=CodexLifecycleAdapter(device_auth_operation), health=CodexHealthAdapter(health_client), ) ] if claude_registry is not None: providers.append( ProviderServices( tool=AccountRegistryKind.CLAUDE, registry=claude_registry, lifecycle=ClaudeAuthOperation(claude_registry), health=ClaudeHealthClient(), ) ) if grok_registry is not None: providers.append( ProviderServices( tool=AccountRegistryKind.GROK, registry=grok_registry, lifecycle=GrokAuthOperation(grok_registry), health=GrokHealthClient(), ) ) return ProviderServiceMap(providers) def _provider_registry(self, tool: AccountRegistryKind) -> AccountRegistry | None: try: return self._provider_services.for_tool(tool).registry except LookupError: return None def _providers(self) -> tuple[ProviderServices, ...]: providers: list[ProviderServices] = [] for tool in ( AccountRegistryKind.CODEX, AccountRegistryKind.CLAUDE, AccountRegistryKind.GROK, ): try: providers.append(self._provider_services.for_tool(tool)) except LookupError: continue return tuple(providers) def _account_lookup(self) -> dict[str, tuple[Account, AccountRegistry]]: lookup: dict[str, tuple[Account, AccountRegistry]] = {} for provider in self._providers(): lookup.update( { self._account_key(account): (account, provider.registry) for account in provider.registry.list() } ) return lookup def _build_menu_items(self) -> tuple[TrayMenuItem, ...]: providers = self._providers() factory_items = list(build_factory_menu_items(self._factory_snapshot)) if len(providers) == 1: provider = providers[0] items = self._provider_account_items(provider) if items: items.append(self._separator()) items.extend(factory_items) items.append(self._separator()) items.append(self._action_item("Open dashboard", action="open_dashboard")) items.append(self._submenu_item("Manage accounts", self._manage_items(provider))) items.append(self._action_item("Quit", action="quit")) return tuple(items) items: list[TrayMenuItem] = [ self._submenu_item(self._provider_label(provider.tool), self._provider_menu(provider)) for provider in providers ] items.append(self._separator()) items.extend(factory_items) items.append(self._separator()) items.append(self._action_item("Open dashboard", action="open_dashboard")) items.append(self._action_item("Quit", action="quit")) return tuple(items) def _provider_menu(self, provider: ProviderServices) -> tuple[TrayMenuItem, ...]: items = self._provider_account_items(provider) if items: items.append(self._separator()) items.append(self._submenu_item("Manage accounts", self._manage_items(provider))) return tuple(items) def _provider_account_items(self, provider: ProviderServices) -> list[TrayMenuItem]: items: list[TrayMenuItem] = [] default_slug = provider.registry.default_slug() for account in provider.registry.list(): locked = self._account_is_locked(provider.registry, account.slug) snapshot = ( self.snapshots.get(self._account_key(account)) if self._show_health_in_menu(account) else None ) label = self._account_label( account, snapshot, show_health=self._show_health_in_menu(account), now=self._time(), locked=locked, ) items.append( TrayMenuItem( kind=TrayMenuItemKind.RADIO, label=label, active=account.slug == default_slug, sensitive=not locked, account_key=account.tray_key, ) ) return items def _manage_items(self, provider: ProviderServices) -> tuple[TrayMenuItem, ...]: items: list[TrayMenuItem] = [ self._action_item( "Add account…", action=f"add_account:{provider.tool.value}", sensitive=not self._provider_action_busy(provider.tool.value, "add_account"), ) ] accounts = provider.registry.list() if accounts: items.append(self._separator()) default_slug = provider.registry.default_slug() for account in accounts: key = self._account_key(account) locked = self._account_is_locked(provider.registry, account.slug) items.extend( [ self._action_item( f"{'Unlock' if locked else 'Lock'} {account.alias}", action="toggle_account_lock", account_key=key, sensitive=not self._account_action_busy(account, "account_lock"), ), self._action_item( f"Set default {account.alias}", action="set_default", account_key=key, sensitive=account.slug != default_slug and not locked, ), self._action_item( f"Reload {account.alias}", action="reload_account", account_key=key, sensitive=( not locked and not self._account_action_busy(account, "reload") ), ), self._action_item( f"Re-authenticate {account.alias}…", action="repair_account", account_key=key, sensitive=( not locked and not self._account_action_busy(account, "reauthenticate") ), ), self._action_item( f"Rename {account.alias}…", action="rename_account", account_key=key, ), self._action_item( f"Remove {account.alias}…", action="remove_account", account_key=key, ), ] ) return tuple(items) @staticmethod def _action_item( label: str, *, action: str, account_key: str | None = None, sensitive: bool = True, ) -> TrayMenuItem: return TrayMenuItem( kind=TrayMenuItemKind.ACTION, label=label, action=action, account_key=account_key, sensitive=sensitive, ) @staticmethod def _submenu_item(label: str, children: tuple[TrayMenuItem, ...]) -> TrayMenuItem: return TrayMenuItem( kind=TrayMenuItemKind.SUBMENU, label=label, children=children, ) @staticmethod def _separator() -> TrayMenuItem: return TrayMenuItem(kind=TrayMenuItemKind.SEPARATOR) def _append_rendered_menu_items( self, gtk: Any, menu: Any, items: tuple[TrayMenuItem, ...], account_lookup: dict[str, tuple[Account, AccountRegistry]], ) -> None: group = None for item in items: rendered = self._render_menu_item(gtk, item, account_lookup, group) menu.append(rendered) if item.kind == TrayMenuItemKind.RADIO and group is None: group = rendered def _render_menu_item( self, gtk: Any, item: TrayMenuItem, account_lookup: dict[str, tuple[Account, AccountRegistry]], group: Any | None, ) -> Any: if item.kind == TrayMenuItemKind.SEPARATOR: return gtk.SeparatorMenuItem() if item.kind == TrayMenuItemKind.SUBMENU: menu_item = gtk.MenuItem(label=item.label) submenu = gtk.Menu() self._append_rendered_menu_items(gtk, submenu, item.children, account_lookup) menu_item.set_submenu(submenu) if hasattr(menu_item, "set_sensitive"): menu_item.set_sensitive(item.sensitive) return menu_item if item.kind == TrayMenuItemKind.RADIO: menu_item = gtk.RadioMenuItem(label=item.label) if group is not None: menu_item.join_group(group) if item.account_key is not None: account_state = account_lookup.get(item.account_key) if account_state is not None: account, registry = account_state handle = menu_item.connect("activate", self._make_activate_handler(account, registry)) if item.active: menu_item.handler_block(handle) menu_item.set_active(True) menu_item.handler_unblock(handle) self._menu_items[item.account_key] = menu_item if hasattr(menu_item, "set_sensitive"): menu_item.set_sensitive(item.sensitive) return menu_item callback = self._menu_action_callback(item, account_lookup) menu_item = gtk.MenuItem(label=item.label) if callback is not None: menu_item.connect("activate", callback) if hasattr(menu_item, "set_sensitive"): menu_item.set_sensitive(item.sensitive) return menu_item def _menu_action_callback( self, item: TrayMenuItem, account_lookup: dict[str, tuple[Account, AccountRegistry]], ) -> Callable[..., Any] | None: if item.action == "quit": return lambda _item: self._modules()[0].main_quit() if item.action == "open_dashboard": return lambda _item: self._open_popup() if item.action == "open_factory": return lambda _item: self._open_factory_in_deck() if item.action is not None and item.action.startswith("add_account:"): tool = item.action.split(":", 1)[1] return lambda _item, tool=tool: self._prompt_and_add_account_for_tool(tool) if item.account_key is None: return None account_state = account_lookup.get(item.account_key) if account_state is None: return None account, _registry = account_state if item.action == "toggle_account_lock": return lambda _item, account=account: self._toggle_account_lock(account) if item.action == "set_default": return lambda _item, account=account: self._set_default_account(account) if item.action == "reload_account": return lambda _item, account=account: self._refresh_account(account) if item.action == "repair_account": return lambda _item, account=account: self._repair_account(account) if item.action == "rename_account": return lambda _item, account=account: self._rename_account(account) if item.action == "remove_account": return lambda _item, account=account: self._remove_account(account) return None @staticmethod def _registry_tool(registry: AccountRegistry | None) -> str: kind = getattr(registry, "kind", None) tool = getattr(kind, "value", None) if isinstance(tool, str) and tool: return tool return "codex" def _account_key(self, account: Account) -> str: return account.tray_key def _rekey_snapshots(self, tool: str, slug: str, new_slug: str) -> None: old_key = f"{tool}:{slug}" new_key = f"{tool}:{new_slug}" for store in (self.snapshots, self.last_fetched): if old_key in store: store[new_key] = store.pop(old_key) def _normalize_snapshot_keys( self, snapshots: dict[str, AccountSnapshot], registry: AccountRegistry | None = None, ) -> dict[str, AccountSnapshot]: normalized: dict[str, AccountSnapshot] = {} tool = self._registry_tool(registry) for key, snapshot in snapshots.items(): normalized[key if ":" in key else f"{tool}:{key}"] = snapshot return normalized def _bare_slug_snapshots(self, registry: AccountRegistry | None) -> dict[str, AccountSnapshot]: if registry is None: return {} prefix = f"{self._registry_tool(registry)}:" active_keys = {self._account_key(account) for account in registry.list()} return { key[len(prefix):]: snapshot for key, snapshot in self.snapshots.items() if key.startswith(prefix) and key in active_keys } def _make_activate_handler(self, account: Account, registry: AccountRegistry | None = None): def on_activate(item: Any) -> None: if not item.get_active(): return target_registry = registry or self.registry if self._account_is_locked(target_registry, account.slug): self._notify(f"Account {account.alias} is locked") self._rebuild_menu() return snapshot = self.snapshots.get(self._account_key(account)) if ( target_registry is self.registry and snapshot is not None and snapshot.status == HealthStatus.BROKEN ): self._repair_account(account) return self._set_default_account(account) return on_activate def _update_title(self) -> None: if self.indicator is None: return codex_default = self._default_account(self.registry) codex_title = self._title_part( "Codex", codex_default, locked=( codex_default is not None and self._account_is_locked(self.registry, codex_default.slug) ), ) if self.claude_registry is None: self.indicator.set_title(codex_title) return claude_default = self._default_account(self.claude_registry) claude_title = self._title_part( "Claude", claude_default, locked=( claude_default is not None and self._account_is_locked(self.claude_registry, claude_default.slug) ), ) self.indicator.set_title(f"{codex_title} · {claude_title}") @staticmethod def _title_part( label: str, account: Account | None, *, locked: bool = False, ) -> str: return format_title_part(label, account, locked=locked) def _default_account(self, registry: AccountRegistry | None = None) -> Account | None: target_registry = registry or self.registry default_slug = target_registry.default_slug() if default_slug is None: return None for account in target_registry.list(): if account.slug == default_slug: return account return None @staticmethod def _account_label( account: Account, snapshot: AccountSnapshot | None = None, *, show_health: bool = False, now: float | None = None, locked: bool | None = None, ) -> str: return format_account_label( account, snapshot, show_health=show_health, now=now, locked=locked, ) @staticmethod def _account_is_locked(registry: AccountRegistry, slug: str) -> bool: checker = getattr(registry, "is_locked", None) if not callable(checker): return False try: return bool(checker(slug)) except (AccountLockError, OSError) as exc: print(f"systray-ai: account lock state unavailable: {exc}", file=sys.stderr) return True def _show_health_in_menu(self, account: Account) -> bool: return True @staticmethod def _provider_label(tool: AccountRegistryKind | str) -> str: normalized = tool.value if isinstance(tool, AccountRegistryKind) else tool if normalized == "claude": return "Claude Code" if normalized == "grok": return "Grok" return "Codex" @staticmethod def _notify(message: str) -> None: try: subprocess.run( ["notify-send", APP_DISPLAY_NAME, message], check=False, ) except FileNotFoundError: return def _account_action_busy(self, account: Account, action: str) -> bool: return action in self._busy_actions.get(self._account_key(account), set()) def _provider_action_busy(self, tool: str, action: str) -> bool: return action in self._provider_busy_actions.get(tool, set()) def _set_account_action_busy(self, account: Account, action: str, busy: bool) -> None: key = self._account_key(account) actions = self._busy_actions.setdefault(key, set()) if busy: actions.add(action) else: actions.discard(action) if not actions: self._busy_actions.pop(key, None) self._update_dashboard_account(account) def _set_provider_action_busy(self, tool: str, action: str, busy: bool) -> None: actions = self._provider_busy_actions.setdefault(tool, set()) if busy: actions.add(action) else: actions.discard(action) if not actions: self._provider_busy_actions.pop(tool, None) self._rebuild_popup_if_open() def _toggle_account_lock(self, account: Account) -> None: registry = self._provider_services.for_account(account).registry locked = self._account_is_locked(registry, account.slug) if locked: gtk, _appindicator = self._modules() dialog = gtk.MessageDialog( transient_for=None, flags=0, message_type=gtk.MessageType.QUESTION, buttons=gtk.ButtonsType.YES_NO, text=f"Unlock account {account.alias}?", ) if hasattr(dialog, "format_secondary_text"): dialog.format_secondary_text( "Agents and commands may select this account for new sessions." ) response = dialog.run() dialog.destroy() if response != gtk.ResponseType.YES: return try: registry.set_locked(account.slug, not locked) except (KeyError, OSError, RuntimeError) as exc: self._log_exception(exc) self._notify(self._failure_message("Could not change account lock", exc)) self._rebuild_menu() self._rebuild_popup_if_open() return self._rebuild_menu() self._rebuild_popup_if_open() self._update_title() self._update_warning_icon() state = "locked" if not locked else "unlocked" detail = ( "New sessions are blocked." if not locked else "New sessions may use it again." ) self._notify(f"Account {account.alias} {state}. {detail}") def _set_default_account(self, account: Account) -> None: registry = self._provider_services.for_account(account).registry if self._account_is_locked(registry, account.slug): self._notify(f"Account {account.alias} is locked") self._rebuild_menu() self._rebuild_popup_if_open() return try: registry.set_default(account) except AccountLockError as exc: self._notify(str(exc)) self._rebuild_menu() self._rebuild_popup_if_open() return self._rebuild_menu() self._rebuild_popup_if_open() self._update_title() self._update_warning_icon() self._notify(f"Default account set to {account.alias}") def _breach_evaluation_snapshot(self, snapshot: AccountSnapshot) -> AccountSnapshot: now = self._time() snapshot = replace( snapshot, primary_used_pct=_percent_after_reset( snapshot.primary_used_pct, snapshot.primary_reset_at, now ), secondary_used_pct=_percent_after_reset( snapshot.secondary_used_pct, snapshot.secondary_reset_at, now ), ) if ( snapshot.status == HealthStatus.UNKNOWN and ( snapshot.primary_used_pct is not None or snapshot.secondary_used_pct is not None ) ): return replace(snapshot, status=HealthStatus.OK) return snapshot def _check_limit_warning( self, account: Account, merged_snapshot: AccountSnapshot, previous_snapshot: AccountSnapshot | None = None, ) -> None: registry = self._provider_services.for_account(account).registry if registry.default_slug() != account.slug: return key = self._account_key(account) evaluated = self._breach_evaluation_snapshot(merged_snapshot) if evaluated.status != HealthStatus.OK: return for window, used_pct, previous_used_pct, threshold in ( ( limit_warning.WINDOW_5H, evaluated.primary_used_pct, previous_snapshot.primary_used_pct if previous_snapshot else None, self._settings.warning_threshold_pct, ), ( limit_warning.WINDOW_7D, evaluated.secondary_used_pct, previous_snapshot.secondary_used_pct if previous_snapshot else None, limit_warning.SEVEN_DAY_WARNING_THRESHOLD_PCT, ), ): if used_pct is None: continue remaining = 100 - min(max(used_pct, 0), 100) if limit_warning.window_reset(previous_used_pct, used_pct): if self._quota_notification_store.record_warning_reset(key, window): self._notify( f"{account.tool.capitalize()} account '{account.alias}' {window} limit " f"reset — {remaining}% available" ) elif remaining < threshold and self._quota_notification_store.record_warning(key, window): self._notify( f"{account.tool.capitalize()} account '{account.alias}' is under " f"{threshold}% remaining on its {window} limit ({remaining}% left)" ) self._check_quota_lifecycle(account, evaluated) self._update_warning_icon() # Hysteresis band: exhaustion notifies at >=100% used, recovery re-arms # only once usage falls back to this level or below. QUOTA_REARM_USED_PCT = 95 def _check_quota_lifecycle(self, account: Account, snapshot: AccountSnapshot) -> None: for window, used_pct, reset_at in ( ("5h", snapshot.primary_used_pct, snapshot.primary_reset_at), ("7d", snapshot.secondary_used_pct, snapshot.secondary_reset_at), ): if used_pct is None or reset_at is None: continue reset_cycle = str(reset_at) if used_pct >= 100: if self._quota_notification_store.record_exhaustion( self._account_key(account), window, reset_cycle ): self._notify( f"{account.tool.capitalize()} account '{account.alias}' has 0% left on its {window} limit" ) elif used_pct <= self.QUOTA_REARM_USED_PCT: self._quota_notification_store.record_reset( self._account_key(account), window, reset_cycle ) def _update_warning_icon(self) -> None: if self.indicator is None: return warning_remaining = self._warning_icon_remaining_pct() warning_active = warning_remaining is not None if warning_active != self._icon_warning_active: if warning_active: if hasattr(self.indicator, "set_icon_full"): icon_path = self._render_warning_icon(warning_remaining) self._set_indicator_icon_theme(icon_path.parent) self.indicator.set_icon_full( self._warning_icon_name, f"Usage limit warning (!) {warning_remaining}% left", ) else: if hasattr(self.indicator, "set_icon_full"): self.indicator.set_icon_full(APPINDICATOR_ICON, APP_DISPLAY_NAME) self._icon_warning_active = warning_active self._warning_icon_remaining = warning_remaining return if warning_active and warning_remaining != self._warning_icon_remaining and hasattr( self.indicator, "set_icon_full" ): icon_path = self._render_warning_icon(warning_remaining) self._set_indicator_icon_theme(icon_path.parent) self.indicator.set_icon_full( self._warning_icon_name, f"Usage limit warning (!) {warning_remaining}% left", ) self._warning_icon_remaining = warning_remaining def _warning_icon_remaining_pct(self) -> int | None: remaining_values: list[int] = [] registries = [ registry for registry in (self.registry, self.claude_registry, self.grok_registry) if registry is not None ] for registry in registries: account = self._default_account(registry) if account is None: continue snapshot = self.snapshots.get(self._account_key(account)) if snapshot is None: continue evaluated = self._breach_evaluation_snapshot(snapshot) breached = limit_warning.breached_windows(evaluated) if limit_warning.WINDOW_5H in breached and evaluated.primary_used_pct is not None: remaining_values.append(100 - min(max(evaluated.primary_used_pct, 0), 100)) if limit_warning.WINDOW_7D in breached and evaluated.secondary_used_pct is not None: remaining_values.append(100 - min(max(evaluated.secondary_used_pct, 0), 100)) if not remaining_values: return None return min(remaining_values) def _render_warning_icon(self, remaining_pct: int) -> Path: text = escape(f"{remaining_pct}%") x = 15.5 if remaining_pct >= 10 else 17.0 font_size = "4.6" if remaining_pct >= 10 else "6.2" svg = f""" {text} """ self._warning_icon_path.write_text(svg, encoding="utf-8") return self._warning_icon_path def _set_indicator_icon_theme(self, theme_path: Path) -> None: if self.indicator is None or not hasattr(self.indicator, "set_icon_theme_path"): return self.indicator.set_icon_theme_path(str(theme_path)) def _toggle_popup(self) -> None: if self.popup is not None and not self.popup.is_closed: self.popup.close() return self._open_popup() def _open_popup(self) -> None: self._ensure_popup() if self.popup is None: return if self._dashboard is not None: self._dashboard.show_tab(self._build_dashboard_vm().active_tab) if hasattr(self.popup, "show"): self.popup.show() if hasattr(self.popup, "show_all"): self.popup.show_all() self._refresh_all(force=True) def _ensure_popup(self) -> None: if self.popup is None: self.popup = self._popup_window_factory( title="Accounts", auto_close=self._settings.auto_close, transient_for=self.menu, ) if self._dashboard is None: factory = self._dashboard_factory if factory is None: from ui.dashboard import Dashboard factory = Dashboard self._dashboard = factory(self._dashboard_callbacks()) content = self.popup.get_content_area() widget = getattr(self._dashboard, "widget", self._dashboard) if hasattr(content, "add"): content.add(widget) else: content.pack_start(widget, True, True, 0) self._apply_dashboard_theme() self._rebuild_popup() def _apply_dashboard_theme(self) -> None: if self._dashboard is None: return set_theme = getattr(self._dashboard, "set_theme", None) if not callable(set_theme): return from ui.theme import resolve_theme preference = getattr(self._settings, "theme", "system") set_theme(resolve_theme(preference)) def _rebuild_popup_if_open(self) -> None: if self.popup is None: return self._rebuild_popup() def _rebuild_popup(self) -> None: if self.popup is None or self._dashboard is None: return self._apply_dashboard_theme() self._dashboard.update(self._build_dashboard_vm()) def _build_dashboard_vm(self) -> DashboardVM: account_caps_by_tool = { provider.tool.value: self._account_caps_for_tool(provider.tool.value) for provider in self._providers() } providers: list[ProviderInput] = [] for provider in self._providers(): default_slug = provider.registry.default_slug() tool_caps = account_caps_by_tool.get(provider.tool.value, {}) accounts = tuple( AccountInput( tool=provider.tool.value, slug=account.slug, alias=account.alias, plan=account.plan, is_default=account.slug == default_slug, snapshot=self.snapshots.get(self._account_key(account)), checked_at=self.last_fetched.get(self._account_key(account)), busy_actions=frozenset( self._busy_actions.get(self._account_key(account), set()) ), account_caps=tool_caps.get(account.slug, {}), locked=self._account_is_locked(provider.registry, account.slug), ) for account in provider.registry.list() ) providers.append( ProviderInput( tool=provider.tool.value, label=self._provider_label(provider.tool), accounts=accounts, ) ) return build_dashboard_vm( providers, now=self._time(), last_used_tab=self._settings.last_tab or None, warning_threshold_pct=self._settings.warning_threshold_pct, busy_add_tools=frozenset( tool for tool, actions in self._provider_busy_actions.items() if "add_account" in actions ), dismissed_alert_keys=frozenset(self._dismissed_alert_keys), ) def _dashboard_callbacks(self) -> DashboardCallbacks: return DashboardCallbacks( on_set_default=lambda tool, slug: self._with_account( tool, slug, self._set_default_account ), on_reload=self._dashboard_reload, on_repair=lambda tool, slug: self._with_account(tool, slug, self._repair_account), on_rename=lambda tool, slug: self._with_account(tool, slug, self._rename_account), on_remove=lambda tool, slug: self._with_account(tool, slug, self._remove_account), on_add=lambda tool: self._prompt_and_add_account_for_tool(tool), on_open_settings=self._open_settings_dialog, on_set_limits=lambda tool, slug: self._with_account( tool, slug, self._open_limits_dialog ), on_toggle_lock=lambda tool, slug: self._with_account( tool, slug, self._toggle_account_lock ), on_tab_changed=self._persist_last_tab, on_dismiss_alert=self._dismiss_alert, ) def _dismiss_alert(self, kind: str, tool: str, slug: str) -> None: self._dismissed_alert_keys.add((kind, tool, slug)) self._rebuild_popup_if_open() if self.popup is not None and self.popup.is_visible: resize_to_content = getattr(self.popup, "resize_to_content", None) if callable(resize_to_content): resize_to_content() def _with_account( self, tool: str, slug: str, action: Callable[[Account], None], ) -> None: account = self._resolve_account(tool, slug) if account is None: return action(account) def _resolve_account(self, tool: str, slug: str) -> Account | None: try: registry = self._provider_services.for_tool(tool).registry except LookupError: return None for account in registry.list(): if account.slug == slug: return account return None def _dashboard_reload(self, tool: str | None, slug: str | None) -> None: if tool is None: self._refresh_all(force=True) return if slug is None: try: registry = self._provider_services.for_tool(tool).registry except LookupError: return for account in registry.list(): self._refresh_account(account) return self._with_account(tool, slug, self._refresh_account) def _persist_last_tab(self, tab_id: str) -> None: if all(provider.tool.value != tab_id for provider in self._providers()): return if self._settings.last_tab == tab_id: return self._settings = replace(self._settings, last_tab=tab_id) self._settings_store.write(self._settings) def _open_settings_dialog(self) -> None: factory = self._settings_dialog_factory if factory is None: from ui.settings_dialog import SettingsDialog factory = SettingsDialog dialog = factory( self._settings, on_save=self._apply_settings, transient_for=getattr(self.popup, "window", None), ) dialog.open() def _routing_rules_path_for_tool(self, tool: str) -> Path | None: try: kind = AccountRegistryKind(tool) except ValueError: return None if kind not in ROUTING_RULES_FILENAMES: return None registry = self._provider_services.for_tool(tool).registry return registry.base_dir / ROUTING_RULES_FILENAMES[kind] def _account_caps_for_tool(self, tool: str) -> dict[str, dict[str, int]]: path = self._routing_rules_path_for_tool(tool) if path is None or not path.is_file(): return {} registry = self._provider_services.for_tool(tool).registry known_slugs = {account.slug for account in registry.list()} try: rules = load_rules(path) except (OSError, ValueError) as exc: print(f"systray: {tool} spend caps unreadable: {exc}", file=sys.stderr) return {} return { slug: dict(windows) for slug, windows in rules.account_caps.items() if slug in known_slugs } def _open_limits_dialog(self, account: Account) -> None: tool = account.tool path = self._routing_rules_path_for_tool(tool) if path is None: return snapshot = self.snapshots.get(self._account_key(account)) if snapshot is None: return caps = self._account_caps_for_tool(tool).get(account.slug, {}) factory = self._limits_dialog_factory if factory is None: from ui.limits_dialog import LimitsDialog factory = LimitsDialog registry = self._provider_services.for_tool(tool).registry known_slugs = {entry.slug for entry in registry.list()} def on_save(window: str, cap_pct: int | None) -> None: from routing_resolver import update_account_cap update_account_cap( path, account.slug, window, cap_pct, known_slugs=known_slugs, ) self._rebuild_popup_if_open() dialog = factory( account.slug, snapshot, caps, on_save, transient_for=getattr(self.popup, "window", None), ) dialog.open() def _apply_settings(self, settings: Any) -> None: self._settings = settings self._settings_store.write(settings) if self.popup is not None and hasattr(self.popup, "set_auto_close"): self.popup.set_auto_close(settings.auto_close) self._reschedule_health_timer(settings.refresh_interval_seconds) self._rebuild_popup_if_open() def _reschedule_health_timer(self, interval_seconds: int) -> None: glib = self._glib_module() if glib is None or not hasattr(glib, "timeout_add_seconds"): return if self._health_timer_id is not None and hasattr(glib, "source_remove"): glib.source_remove(self._health_timer_id) self._health_timer_id = glib.timeout_add_seconds( interval_seconds, self._refresh_all_timer ) def _default_settings_store(self) -> Any: from ui.settings_store import SettingsStore base_dir = getattr(self.registry, "base_dir", None) directory = Path(base_dir) if base_dir is not None else runtime_dir() return SettingsStore(directory / "settings.json") def _update_dashboard_account(self, _account: Account) -> None: self._rebuild_popup_if_open() @staticmethod def _failure_message(prefix: str, exc: Exception) -> str: detail = str(exc) if not detail: return prefix return f"{prefix}: {detail}" @staticmethod def _log_exception(exc: Exception) -> None: traceback.print_exception(type(exc), exc, exc.__traceback__) def _repair_account(self, account: Account) -> None: registry = self._provider_services.for_account(account).registry if self._account_is_locked(registry, account.slug): self._notify(f"Account {account.alias} is locked") return self._set_account_action_busy(account, "reauthenticate", True) self._thread_runner(lambda: self._run_repair_flow(account)) def _add_account(self, alias: str, slug: str | None = None) -> None: self._add_account_for_tool(self._registry_tool(self.registry), alias, slug=slug) def _prompt_and_add_account_for_tool(self, tool: str) -> None: slug = self._prompt_account_slug(tool) if slug is None: return self._add_account_for_tool(tool, slug, slug=slug) def _prompt_account_slug(self, tool: str) -> str | None: gtk, _appindicator = self._modules() try: provider = self._provider_services.for_tool(tool) except LookupError: return None registry = provider.registry dialog = gtk.Dialog(title="Add account") dialog.add_button("Cancel", gtk.ResponseType.CANCEL) dialog.add_button("Add", gtk.ResponseType.OK) content = dialog.get_content_area() content.add(gtk.Label(label="Account slug (letters, numbers, hyphens):")) entry = gtk.Entry() content.add(entry) dialog.show_all() response = dialog.run() raw_slug = entry.get_text().strip() dialog.destroy() if response != gtk.ResponseType.OK or not raw_slug: return None try: slug = registry.normalize_slug(raw_slug) except ValueError as exc: self._notify(str(exc)) return None if slug in RESERVED_SLUGS: self._notify(f"Reserved account slug: {slug}") return None known_slugs = {account.slug for account in registry.list()} if slug in known_slugs or (registry.accounts_dir / slug).exists(): self._notify(f"Account slug already exists: {slug}") return None return slug def _add_account_for_tool( self, tool: str, alias: str, *, slug: str | None = None, login_hint: str | None = None, ) -> None: if self._provider_action_busy(tool, "add_account"): return self._set_provider_action_busy(tool, "add_account", True) self._thread_runner( lambda: self._run_add_account_flow(tool, alias, slug=slug, login_hint=login_hint) ) def _run_repair_flow(self, account: Account) -> None: dialog = None provider = self._provider_services.for_account(account) if self._account_is_locked(provider.registry, account.slug): self._idle_add(self._set_account_action_busy, account, "reauthenticate", False) self._idle_add(self._notify, f"Account {account.alias} is locked") return for event in provider.lifecycle.reauthenticate(account): if event.kind in {"process_started", "started", "browser-waiting", "commit", "rollback"}: continue if event.kind == "prompt_ready": if event.session is None: message = event.message or ( event.prompt.raw_text if event.prompt is not None else f"Complete re-auth for {account.alias}" ) self._idle_add(self._notify, message) continue try: dialog = self._show_repair_dialog(event.prompt, account, event.session) except Exception as exc: self._log_exception(exc) if event.session is not None: terminate_process(event.session.process) self._device_auth_flow.rollback(account.account_home) self._idle_add( self._notify, self._failure_message( f"Re-authentication failed for {account.alias}", exc, ), ) self._idle_add(self._set_account_action_busy, account, "reauthenticate", False) self._idle_add(self._rebuild_menu) self._idle_add(self._rebuild_popup_if_open) return continue if event.kind == "code_required": try: dialog = self._create_claude_code_entry_dialog( event.prompt.url, on_submit=lambda code: provider.lifecycle.submit_code(event.session, code), on_cancel=lambda: terminate_process(event.session.process), ) except Exception as exc: self._log_exception(exc) if event.session is not None: terminate_process(event.session.process) self._idle_add( self._notify, self._failure_message( f"Re-authentication failed for {account.alias}", exc, ), ) self._idle_add(self._set_account_action_busy, account, "reauthenticate", False) self._idle_add(self._rebuild_menu) self._idle_add(self._rebuild_popup_if_open) return continue if event.kind == "collision": if event.collision is not None: self._mark_account_broken(event.collision) self._idle_add(self._finish_repair_failure, dialog, account) return if event.kind == "success": self._idle_add(self._finish_repair_success, dialog, account) return if event.kind == "failure": if dialog is None: self._idle_add( self._notify, event.message or self._failure_message( f"Re-authentication failed for {account.alias}", event.error or RuntimeError("repair failed"), ), ) self._idle_add(self._set_account_action_busy, account, "reauthenticate", False) self._idle_add(self._rebuild_menu) self._idle_add(self._rebuild_popup_if_open) return self._idle_add(self._finish_repair_failure, dialog, account) return def _run_add_account_flow( self, tool: str, alias: str, *, slug: str | None = None, login_hint: str | None = None, ) -> None: dialog = None provider = self._provider_services.for_tool(tool) for event in provider.lifecycle.add(alias, slug=slug, login_hint=login_hint): if event.kind in {"process_started", "started", "browser-waiting"}: continue if event.kind == "prompt_ready": # Grok device-code: no subprocess session — notify URL/code only. if event.session is None: message = event.message or ( event.prompt.raw_text if event.prompt is not None else f"Complete sign-in for {alias}" ) self._idle_add(self._notify, message) continue try: dialog = self._show_add_dialog( event.prompt, event.existing_accounts or [], event.session, ) except Exception as exc: self._log_exception(exc) if event.session is not None: terminate_process(event.session.process) if event.account is not None and event.account.slug != alias: try: provider.registry.remove(event.account.slug) except Exception: pass self._idle_add( self._notify, self._failure_message(f"Adding account {alias} failed", exc), ) self._idle_add(self._set_provider_action_busy, tool, "add_account", False) self._idle_add(self._rebuild_menu) self._idle_add(self._rebuild_popup_if_open) return continue if event.kind == "code_required": try: dialog = self._create_claude_code_entry_dialog( event.prompt.url, on_submit=lambda code: provider.lifecycle.submit_code( event.session, code, ), on_cancel=lambda: terminate_process(event.session.process), ) except Exception as exc: self._log_exception(exc) if event.session is not None: terminate_process(event.session.process) provider.registry.remove( event.account.slug if event.account is not None else alias ) self._idle_add( self._notify, self._failure_message(f"Adding account {alias} failed", exc), ) self._idle_add(self._set_provider_action_busy, tool, "add_account", False) self._idle_add(self._rebuild_menu) self._idle_add(self._rebuild_popup_if_open) return continue if event.kind == "rollback": continue if event.kind == "collision": if event.collision is not None: self._mark_account_broken(event.collision) self._idle_add( self._finish_add_collision, dialog, alias, tool, event.collision, ) return if event.kind == "success": self._idle_add(self._finish_add_success, dialog, alias, tool) return if event.kind == "failure": if dialog is None: self._idle_add( self._notify, event.message or self._failure_message( f"Adding account {alias} failed", event.error or RuntimeError("add failed"), ), ) self._idle_add(self._set_provider_action_busy, tool, "add_account", False) self._idle_add(self._rebuild_menu) self._idle_add(self._rebuild_popup_if_open) return self._idle_add(self._finish_add_failure, dialog, alias, tool) return def _show_repair_dialog( self, prompt: DeviceAuthPrompt, account: Account, session: DeviceAuthSession, ) -> Any: return self._create_device_auth_dialog( prompt, on_retry=lambda: self._repair_account(account), on_cancel=lambda: terminate_process(session.process), existing_accounts=None, ) def _show_add_dialog( self, prompt: DeviceAuthPrompt, existing_accounts: list[tuple[str, str]], session: DeviceAuthSession, ) -> Any: return self._create_device_auth_dialog( prompt, on_retry=None, on_cancel=lambda: terminate_process(session.process), existing_accounts=existing_accounts, ) def _create_device_auth_dialog( self, prompt: DeviceAuthPrompt, on_retry: Callable[[], None] | None, on_cancel: Callable[[], None] | None, existing_accounts: list[tuple[str, str]] | None, ) -> Any: done = threading.Event() result: dict[str, Any] = {} def build_dialog() -> bool: try: result["dialog"] = self._dialog_factory( prompt, on_retry=on_retry, on_cancel=on_cancel, existing_accounts=existing_accounts, ) except Exception as exc: result["error"] = exc finally: done.set() return False self._idle_add(build_dialog) done.wait() error = result.get("error") if error is not None: raise error return result["dialog"] def _create_claude_code_entry_dialog( self, url: str, *, on_submit: Callable[[str], None], on_cancel: Callable[[], None], ) -> Any: done = threading.Event() result: dict[str, Any] = {} def build_dialog() -> bool: try: result["dialog"] = ClaudeCodeEntryDialog( url, on_submit=on_submit, on_cancel=on_cancel, ) except Exception as exc: result["error"] = exc finally: done.set() return False self._idle_add(build_dialog) done.wait() error = result.get("error") if error is not None: raise error return result["dialog"] def _finish_repair_success(self, dialog: Any, account: Account) -> None: if hasattr(dialog, "close_success"): dialog.close_success() key = self._account_key(account) self._set_account_action_busy(account, "reauthenticate", False) merged = self._merge_snapshot( key, AccountSnapshot(HealthStatus.OK, None, None, checked_at=self._time()), ) if merged is not None: self.snapshots[key] = merged self.last_fetched[key] = merged.checked_at if merged.checked_at is not None else self._time() self._rebuild_menu() self._rebuild_popup_if_open() self._write_health_cache() self._update_title() self._notify(f"Re-authenticated {account.alias}") def _finish_repair_failure(self, dialog: Any, account: Account) -> None: self._set_account_action_busy(account, "reauthenticate", False) if hasattr(dialog, "show_failure"): dialog.show_failure(retry_available=True) self._rebuild_menu() self._rebuild_popup_if_open() self._notify(f"Re-authentication failed for {account.alias}") def _finish_add_success(self, dialog: Any, alias: str, tool: str) -> None: self._set_provider_action_busy(tool, "add_account", False) if hasattr(dialog, "close_success"): dialog.close_success() self._rebuild_menu() self._rebuild_popup_if_open() self._update_title() self._notify(f"Added account {alias}") def _finish_add_failure(self, dialog: Any, alias: str, tool: str) -> None: self._set_provider_action_busy(tool, "add_account", False) if hasattr(dialog, "show_failure"): dialog.show_failure(retry_available=False) self._rebuild_menu() self._rebuild_popup_if_open() self._notify(f"Adding account {alias} failed") def _finish_add_collision(self, dialog: Any, alias: str, tool: str, collision: Account) -> None: self._set_provider_action_busy(tool, "add_account", False) if hasattr(dialog, "show_failure"): dialog.show_failure(retry_available=False) self._rebuild_menu() self._rebuild_popup_if_open() self._notify( f"Adding account {alias} collided with existing account {collision.alias}" ) def _existing_accounts_for_add(self, tool: str | None = None) -> list[tuple[str, str]]: provider = self._provider_services.for_tool(tool or self._registry_tool(self.registry)) return [ (account.alias, account.email) for account in provider.registry.list() if account.email is not None ] def _find_account_collision(self, new_slug: str) -> Account | None: new_account = None for account in self.registry.list(): if account.slug == new_slug: new_account = account break if new_account is None or not new_account.account_id: return None for account in self.registry.list(): if account.slug != new_slug and account.account_id == new_account.account_id: return account return None def _mark_account_broken(self, account: Account) -> None: key = self._account_key(account) merged = self._merge_snapshot( key, AccountSnapshot(HealthStatus.BROKEN, None, None, checked_at=self._time()), ) if merged is not None: self.snapshots[key] = merged self.last_fetched[key] = merged.checked_at if merged.checked_at is not None else self._time() self._write_health_cache() self._update_dashboard_account(account) def _rename_account(self, account: Account) -> None: gtk, _appindicator = self._modules() dialog = gtk.Dialog(title="Rename account") dialog.add_button("Cancel", gtk.ResponseType.CANCEL) dialog.add_button("OK", gtk.ResponseType.OK) content = dialog.get_content_area() content.add(gtk.Label(label=f"Rename account: {account.alias}")) entry = gtk.Entry() entry.set_text(account.alias) content.add(entry) dialog.show_all() response = dialog.run() new_alias = entry.get_text().strip() dialog.destroy() if response != gtk.ResponseType.OK or not new_alias: return registry = self._provider_services.for_account(account).registry tool, slug = account.tool, account.slug new_slug = registry.rename(slug, new_alias) if new_slug and new_slug != slug: self._rekey_snapshots(tool, slug, new_slug) self._rebuild_menu() self._rebuild_popup_if_open() self._update_title() def _remove_account(self, account: Account) -> None: gtk, _appindicator = self._modules() dialog = gtk.MessageDialog( transient_for=None, flags=0, message_type=gtk.MessageType.QUESTION, buttons=gtk.ButtonsType.YES_NO, text=f"Remove account {account.alias}?", ) response = dialog.run() dialog.destroy() if response != gtk.ResponseType.YES: return self._provider_services.for_account(account).registry.remove(account.slug) self._rebuild_menu() self._rebuild_popup_if_open() self._update_title() def _rebuild_menu(self) -> None: gtk, _appindicator = self._modules() self.menu = gtk.Menu() self._menu_items = {} if hasattr(self.menu, "connect"): self.menu.connect("show", lambda *_args: self._refresh_all(force=True)) account_lookup = self._account_lookup() self._append_rendered_menu_items(gtk, self.menu, self._build_menu_items(), account_lookup) self.menu.show_all() if self.indicator is not None: self.indicator.set_menu(self.menu) self._rebuild_popup_if_open() @staticmethod def _new_radio_menu_item(gtk: Any, group: Any, label: str) -> Any: radio_menu_item = gtk.RadioMenuItem factory = getattr(radio_menu_item, "new_with_label_from_widget", None) if callable(factory): return factory(group, label) try: return radio_menu_item(label=label) except TypeError: return radio_menu_item(group, label) def _glib_module(self) -> Any | None: if self._glib is None: self._glib = _load_glib_module() return self._glib def _refresh_all_timer(self) -> bool: # An exception escaping a GLib timeout callback removes the source, so one # failed tick would stop every later health refresh for the process lifetime. try: self._refresh_all(force=False) self._update_warning_icon() except Exception as exc: self._log_exception(exc) return True def _refresh_account(self, account: Account) -> None: registry = self._provider_services.for_account(account).registry if self._account_is_locked(registry, account.slug): self._notify(f"Account {account.alias} is locked") return self._set_account_action_busy(account, "reload", True) self._scheduler.schedule_refresh_account(account, None, self._run_refresh_account) def _schedule_refresh_account(self, account: Account, generation: int | None) -> None: registry = self._provider_services.for_account(account).registry if self._account_is_locked(registry, account.slug): if generation is not None: self._finish_refresh(generation) return self._set_account_action_busy(account, "reload", True) self._scheduler.schedule_refresh_account(account, generation, self._run_refresh_account) def _run_refresh_account(self, account: Account, generation: int | None) -> None: registry = self._provider_services.for_account(account).registry if self._account_is_locked(registry, account.slug): self._idle_add(self._set_account_action_busy, account, "reload", False) if generation is not None: self._idle_add(self._finish_refresh, generation) return self._scheduler.run_refresh_account( account, generation, lambda target_account: self._provider_services.for_account(target_account).health.fetch( target_account ), self._apply_snapshot, self._handle_refresh_error, ) def _handle_refresh_error( self, account: Account, generation: int | None, exc: Exception, ) -> None: self._log_exception(exc) self._set_account_action_busy(account, "reload", False) if generation is not None: self._finish_refresh(generation) def _apply_snapshot( self, account: Account, snapshot: AccountSnapshot, generation: int | None, ) -> None: key = self._account_key(account) checked_at = snapshot.checked_at if snapshot.checked_at is not None else self._time() self.last_fetched[key] = checked_at previous_snapshot = self.snapshots.get(key) merged_snapshot = self._merge_snapshot(key, replace(snapshot, checked_at=checked_at)) if merged_snapshot is not None: self.snapshots[key] = merged_snapshot self._check_limit_warning(account, merged_snapshot, previous_snapshot) self._set_account_action_busy(account, "reload", False) self._update_account_menu_item(account) self._update_dashboard_account(account) if generation is not None: self._finish_refresh(generation) return self._write_health_cache() def _update_account_menu_item(self, account: Account) -> None: key = self._account_key(account) item = self._menu_items.get(key) if item is None: return model_item = self._account_label( account, self.snapshots.get(key) if self._show_health_in_menu(account) else None, show_health=self._show_health_in_menu(account), now=self._time(), locked=self._account_is_locked( self._provider_services.for_account(account).registry, account.slug, ), ) if hasattr(item, "set_label"): item.set_label(model_item) else: item.label = model_item if hasattr(item, "set_sensitive"): item.set_sensitive( not self._account_is_locked( self._provider_services.for_account(account).registry, account.slug, ) ) def _refresh_all(self, force: bool = False) -> None: self._schedule_factory_refresh() accounts: list[Account] = [] for provider in self._providers(): accounts.extend( account for account in provider.registry.list() if not self._account_is_locked(provider.registry, account.slug) ) self._scheduler.refresh_all( accounts, lambda account: self.last_fetched.get(self._account_key(account)), HEALTH_STALE_SECONDS, force, self._schedule_refresh_account, self._write_health_cache, ) @staticmethod def _default_factory_url_opener(url: str) -> None: subprocess.Popen(["firefox", url]) def _open_factory_in_deck(self) -> None: self._factory_url_opener(deck_factory_url()) def _schedule_factory_refresh(self) -> None: if self._factory_refresh_inflight: return self._factory_refresh_inflight = True self._thread_runner(self._run_factory_refresh) def _run_factory_refresh(self) -> None: try: snapshot = self._factory_status_loader() except Exception as exc: self._log_exception(exc) snapshot = FactoryStatusSnapshot.unavailable() self._idle_add(self._apply_factory_snapshot, snapshot) def _apply_factory_snapshot(self, snapshot: FactoryStatusSnapshot) -> None: self._factory_refresh_inflight = False if snapshot == self._factory_snapshot: return self._factory_snapshot = snapshot self._notify_new_factory_decisions(snapshot) self._rebuild_menu() def _notify_new_factory_decisions(self, snapshot: FactoryStatusSnapshot) -> None: if snapshot.availability is not FactoryAvailability.AVAILABLE: return store = self._factory_decision_notification_store for decision in snapshot.pending_decisions: if store.record_notified(decision.decision_id): question = decision.question.strip() or decision.decision_id self._notify(f"Factory needs a decision: {question}") store.prune(snapshot.all_pending_decision_ids) def _finish_refresh(self, generation: int) -> None: self._scheduler.finish_refresh(generation, self._write_health_cache) def _write_health_cache(self) -> None: store = self._health_store if store is not None: store.write(self._bare_slug_snapshots(self.registry)) claude_store = self._claude_health_store if claude_store is not None: claude_store.write(self._bare_slug_snapshots(self.claude_registry)) grok_store = getattr(self, "_grok_health_store", None) if grok_store is not None and self.grok_registry is not None: grok_store.write(self._bare_slug_snapshots(self.grok_registry)) def _load_health_cache(self) -> None: self._load_registry_health_cache(self._health_store, self.registry) if self.claude_registry is not None: self._load_registry_health_cache(self._claude_health_store, self.claude_registry) if self.grok_registry is not None: self._load_registry_health_cache( getattr(self, "_grok_health_store", None), self.grok_registry, ) def _load_registry_health_cache( self, store: HealthSnapshotStore | None, registry: AccountRegistry | None, ) -> None: if store is None: return snapshots = store.read() if not snapshots: return normalized = self._normalize_snapshot_keys(snapshots, registry) self.snapshots.update(normalized) cache_path = getattr(store, "path", None) if cache_path is not None: try: fetched_at = Path(cache_path).stat().st_mtime except OSError: fetched_at = self._time() else: fetched_at = self._time() for key, snapshot in normalized.items(): self.last_fetched[key] = snapshot.checked_at if snapshot.checked_at is not None else fetched_at def _default_health_store(self) -> HealthSnapshotStore | None: path = getattr(self.registry, "health_cache_path", None) if path is None: return None return HealthSnapshotStore(Path(path), stale_after_s=0) def _default_claude_health_store(self) -> HealthSnapshotStore | None: if self.claude_registry is None: return None path = getattr(self.claude_registry, "health_cache_path", None) if path is None: return None return HealthSnapshotStore(Path(path), stale_after_s=0) def _default_grok_health_store(self) -> HealthSnapshotStore | None: if self.grok_registry is None: return None path = getattr(self.grok_registry, "health_cache_path", None) if path is None: return None return HealthSnapshotStore(Path(path), stale_after_s=0) def _merge_snapshot(self, key: str, snapshot: AccountSnapshot) -> AccountSnapshot | None: previous = self.snapshots.get(key) if previous is None or snapshot.status in {HealthStatus.OK, HealthStatus.BROKEN}: return snapshot now = self._time() carried_primary = _unexpired( previous.primary_used_pct, previous.primary_reset_at, now ) carried_secondary = _unexpired( previous.secondary_used_pct, previous.secondary_reset_at, now ) snapshot = replace( snapshot, primary_used_pct=( snapshot.primary_used_pct if snapshot.primary_used_pct is not None else carried_primary ), secondary_used_pct=( snapshot.secondary_used_pct if snapshot.secondary_used_pct is not None else carried_secondary ), primary_reset_at=( snapshot.primary_reset_at if snapshot.primary_reset_at is not None else None if _expired(previous.primary_reset_at, now) else previous.primary_reset_at ), secondary_reset_at=( snapshot.secondary_reset_at if snapshot.secondary_reset_at is not None else None if _expired(previous.secondary_reset_at, now) else previous.secondary_reset_at ), checked_at=( min(snapshot.checked_at, previous.checked_at) if snapshot.checked_at is not None and previous.checked_at is not None else snapshot.checked_at if snapshot.checked_at is not None else previous.checked_at ), named_limits=snapshot.named_limits or previous.named_limits, extra_usage=snapshot.extra_usage if snapshot.extra_usage is not None else previous.extra_usage, spend=snapshot.spend if snapshot.spend is not None else previous.spend, ) return snapshot