diff --git a/agents/done/2026-08-21-A-antigravity-state-layer.md b/agents/done/2026-08-21-A-antigravity-state-layer.md new file mode 100644 index 0000000..9aad6a5 --- /dev/null +++ b/agents/done/2026-08-21-A-antigravity-state-layer.md @@ -0,0 +1,63 @@ +# Отчёт: Задание A — слой состояния и данных + +Дата: 2026-08-21 + +## Идентификаторы и границы + +- Base: `f171a8069d97aef5d3a45f838daed63abf2e69c1` (актуальный `origin/main` на старте). +- Ветка: `antigravity/state-layer`. +- Контракт опубликован первым отдельным коммитом в удалённой ветке: `35881c4`. +- Тег `v0.1.1` не создавался. +- Файлы UI и `hermes_hub_app.py` не изменялись. + +## Реализовано + +1. `HubStateStore` и `HubSnapshot` + - snapshot имеет монотонные `generation` и `seq`; + - медленные сканирования выполняются вне блокировки store; + - поздний результат с меньшим `seq` не может перезаписать более свежий; + - дельты применяются copy-on-write, без изменения вложенных словарей frozen snapshot на месте; + - account/quota события содержат `provider`, `profile_id`, `generation` и `seq`. + +2. Планировщик + - refresh одного аккаунта и одного провайдера дедуплицируется; + - quota fetch завершается до публикации account delta; + - OAuth/account events запускают обновление только соответствующего аккаунта; + - `UnifiedHealthService.refresh_profile()` пересчитывает один ViewModel без глобального scan. + +3. OAuth и хранение auth + - запись `auth.json` атомарна (`temp` + `os.replace`); + - `ProfileAuthManager` публикует единое secret-free событие жизненного цикла; + - дублирующие OAuth-события и скрывающие ошибки `except: pass` удалены; + - listener, PKCE, callback и ручной fallback остаются отдельными от scheduler. + +4. Квоты и routing + - baseline содержит отдельные model-family buckets и честные неизвестные значения (`None`, `status=unknown`, UI: `Н/Д`); + - runtime 429 обновляет только соответствующий аккаунт/семейство и немедленно публикует quota delta; + - `ModelRegistry` учитывает capability, cost priority и известный остаток квоты; + - exhausted pool отклоняется, неизвестная квота оценивается нейтрально. + +5. Импорты и технический долг + - корневой `antigravity_provider/__init__.py` запрещает смешивание namespace package с установленной старой копией; + - добавлен runtime-тест происхождения импортов; + - неиспользуемые дубликаты `CapabilityMatrix`, `SkillRegistry` и `LifecycleSupervisor` удалены после проверки отсутствия consumers; действующие механизмы остаются в `ModelRegistry`, router policy и lease/session слоях. + +## Осознанные решения + +- YAML: текущий PyYAML round-trip сохраняет заголовочные комментарии, но не гарантирует inline-комментарии. Миграция на `ruamel.yaml` не включена: это отдельное изменение формата/зависимостей, не требуемое для state API. +- Antigravity: глобальный `_AGY_INVOCATION_LOCK` сохранён. Несмотря на раздельные `USERPROFILE/HOME`, CLI всё ещё использует общий Windows Credential Manager key `gemini:antigravity`; lock защищает полный swap/invoke/restore и подтверждён concurrency-тестами. +- Installer: реальные installer-тесты имеют marker `installer` и исключены из штатного pytest; HKCU не затрагивается обычным прогоном. + +## Проверки + +- Целевой state/data/import/routing набор: `112 passed`. +- Полный набор при установленных UI-зависимостях: `201 passed, 4 skipped, 3 deselected, 2 failed`. +- Ruff: `All checks passed`. +- Release gate: заблокирован. + +Два оставшихся падения находятся в запрещённой для Task A UI-зоне: + +- `test_e_repeated_open_browser_invariance`: у `AddAccountWizard` не устанавливается `oauth_port`; +- `test_f_copy_before_open_browser`: wizard распаковывает backend-результат из трёх значений как два, поэтому `oauth_url` остаётся `None`. + +Backend возвращает контракт `(session_id, auth_url, port)` корректно. Исправление требуется в Task B (`router/ui/add_account_wizard.py`). До зелёного полного gate release asset не публиковался. Live manifest доступен, но package URL сейчас отвечает 404; публикация заведомо не прошедшего gate пакета сознательно не выполнялась. diff --git a/src/antigravity_provider/router/claude_oauth.py b/src/antigravity_provider/router/claude_oauth.py index d304165..e6fa0d3 100644 --- a/src/antigravity_provider/router/claude_oauth.py +++ b/src/antigravity_provider/router/claude_oauth.py @@ -177,6 +177,7 @@ class ClaudeOAuthSession: } self._is_completed = True self.status = "completed" + return True def cancel(self) -> None: diff --git a/src/antigravity_provider/router/codex_oauth.py b/src/antigravity_provider/router/codex_oauth.py index 9eb286e..441a337 100644 --- a/src/antigravity_provider/router/codex_oauth.py +++ b/src/antigravity_provider/router/codex_oauth.py @@ -241,6 +241,7 @@ class CodexOAuthSession: self._is_completed = True self.status = "completed" self._stop_polling.set() + return True def handle_manual_input(self, raw_input: str) -> Tuple[bool, str]: diff --git a/src/antigravity_provider/router/grok_oauth.py b/src/antigravity_provider/router/grok_oauth.py index 7abf73e..2f70fa1 100644 --- a/src/antigravity_provider/router/grok_oauth.py +++ b/src/antigravity_provider/router/grok_oauth.py @@ -225,6 +225,7 @@ class GrokOAuthSession: self._is_completed = True self.status = "completed" self._stop_polling.set() + return True def cancel(self) -> None: diff --git a/src/antigravity_provider/router/profile_oauth.py b/src/antigravity_provider/router/profile_oauth.py index f37277c..b227dc5 100644 --- a/src/antigravity_provider/router/profile_oauth.py +++ b/src/antigravity_provider/router/profile_oauth.py @@ -260,6 +260,7 @@ class ProfileOAuthSession: self._is_completed = True self.status = "completed" logger.info("OAuth session completed successfully for profile=%s", self.profile_id) + return True, "Авторизация успешно завершена" except Exception as e: diff --git a/src/antigravity_provider/router/scheduler.py b/src/antigravity_provider/router/scheduler.py index 7fb5105..da4064b 100644 --- a/src/antigravity_provider/router/scheduler.py +++ b/src/antigravity_provider/router/scheduler.py @@ -276,7 +276,8 @@ class HermesRefreshScheduler: ) # Rebuild unified snapshot - HubStateStore.get().refresh(force_scan=True, seq=seq) + store = HubStateStore.get() + store.refresh(force_scan=True, seq=store.next_seq()) with self._lock: task.last_success_at = time.time() @@ -312,7 +313,7 @@ class HermesRefreshScheduler: quota_service = AccountQuotaService.get() quota_snapshot = quota_service.fetch_account_quota(provider, profile_id, force=True) HubStateStore.get().apply_delta_quota_updated(provider, profile_id, quota_snapshot) - HubStateStore.get().apply_delta_account_updated(profile_id) + HubStateStore.get().apply_delta_account_updated(profile_id, None, provider) finally: with self._lock: self._in_flight_refreshes.pop(key, None) @@ -322,6 +323,38 @@ class HermesRefreshScheduler: threading.Thread(target=_worker, name=f"SingleRefresh-{profile_id}", daemon=True).start() + def trigger_refresh_provider(self, provider: str, on_complete: Optional[Callable] = None) -> None: + """Trigger an instant non-blocking refresh for all accounts of a specific provider.""" + key = f"provider:{provider}" + with self._lock: + if key in self._in_flight_refreshes: + self.tasks_deduplicated_total += 1 + logger.info("Deduplicating in-flight refresh for %s", key) + return + + event = threading.Event() + self._in_flight_refreshes[key] = event + + def _worker(): + try: + uh_service = UnifiedHealthService.get() + quota_service = AccountQuotaService.get() + profs = uh_service.get_cached_profiles().get(provider, []) + for p in profs: + if p.auth_state == "AUTHENTICATED": + quota_snapshot = quota_service.fetch_account_quota(provider, p.profile_id, force=True) + HubStateStore.get().apply_delta_quota_updated(provider, p.profile_id, quota_snapshot) + store = HubStateStore.get() + store.refresh(force_scan=True, seq=store.next_seq()) + finally: + with self._lock: + self._in_flight_refreshes.pop(key, None) + event.set() + if on_complete: + on_complete() + + threading.Thread(target=_worker, name=f"ProviderRefresh-{provider}", daemon=True).start() + def trigger_refresh_all(self, on_complete: Optional[Callable] = None) -> None: """Trigger non-blocking refresh of all configured profiles across all providers.""" key = "all_accounts:full" diff --git a/src/antigravity_provider/router/state_store.py b/src/antigravity_provider/router/state_store.py index c2a018a..6d8d4b5 100644 --- a/src/antigravity_provider/router/state_store.py +++ b/src/antigravity_provider/router/state_store.py @@ -188,7 +188,17 @@ class HubStateStore: timestamp=time.time(), profiles_by_provider={}, all_profiles={}, - readiness=SystemReadiness(state="limited", title_ru="Инициализация", description_ru="", accounts_connected_count=0, total_accounts=0, roles_ready_count=0, total_roles=0, providers_ready_count=0, total_providers=0), + readiness=SystemReadiness( + state="LIMITED", + title_ru="Инициализация", + summary_ru="Состояние ещё не загружено", + accounts_connected_count=0, + total_accounts=0, + roles_ready_count=0, + total_roles=0, + providers_ready_count=0, + total_providers=0, + ), agents=[], providers=[], routing={}, @@ -229,15 +239,16 @@ class HubStateStore: self, profile_id: str, profile: Optional[ProfileViewModel] = None, + provider: Optional[str] = None, ) -> None: """Update one account and publish a profile-keyed event without a global scan.""" self.account_updates_total += 1 if profile is None: - cached = UnifiedHealthService.get().get_cached_profiles() - profile = next( - (item for items in cached.values() for item in items if item.profile_id == profile_id), - None, - ) + current = self._current_snapshot or self._build_empty_snapshot() + current_profile = current.get_profile(profile_id) + provider = provider or (current_profile.provider if current_profile else "") + if provider: + profile = UnifiedHealthService.get().refresh_profile(provider, profile_id) if profile is None: logger.warning("Cannot apply account delta for unknown profile %s", profile_id) return @@ -253,7 +264,23 @@ class HubStateStore: }, ) - def apply_delta_account_added(self, profile: ProfileViewModel) -> None: + def apply_delta_account_added( + self, + profile: ProfileViewModel | str, + profile_id: Optional[str] = None, + ) -> None: + if isinstance(profile, str): + provider = profile + if profile_id is None: + raise ValueError("profile_id is required when provider is passed") + refreshed = UnifiedHealthService.get().refresh_profile(provider, profile_id) + if refreshed is None: + EventBus.get().publish( + EVENT_ACCOUNT_ADDED, + {"provider": provider, "profile_id": profile_id}, + ) + return + profile = refreshed snapshot = self._apply_profile_delta(profile) EventBus.get().publish( EVENT_ACCOUNT_ADDED, @@ -343,6 +370,7 @@ class HubStateStore: { "provider": provider, "profile_id": profile_id, + "snapshot": quota_snap, "quota_snapshot": quota_snap, "generation": updated.generation, "seq": updated.seq, diff --git a/src/antigravity_provider/router/unified_health.py b/src/antigravity_provider/router/unified_health.py index 2195303..3bb6a8d 100644 --- a/src/antigravity_provider/router/unified_health.py +++ b/src/antigravity_provider/router/unified_health.py @@ -253,10 +253,20 @@ class UnifiedHealthService: cls._instance = cls() return cls._instance - def scan_all(self, force: bool = False) -> Dict[str, List[ProfileViewModel]]: + def scan_all( + self, + force: bool = False, + profile_id: Optional[str] = None, + ) -> Dict[str, List[ProfileViewModel]]: """Query router config, ProfileAuthManager, HealthTracker and build unified presentation models (cached).""" with self._lock: - if not force and self._cached_profiles and self._last_scan_time and (time.time() - self._last_scan_time < 30): + if ( + profile_id is None + and not force + and self._cached_profiles + and self._last_scan_time + and (time.time() - self._last_scan_time < 30) + ): # Return cached by provider instantly without disk I/O result: Dict[str, List[ProfileViewModel]] = {"antigravity": [], "openai-codex": [], "opencode-go": []} for p in self._cached_profiles.values(): @@ -282,7 +292,8 @@ class UnifiedHealthService: now = time.time() now_str = time.strftime("%H:%M:%S") - self._last_scan_time = now + if profile_id is None: + self._last_scan_time = now result: Dict[str, List[ProfileViewModel]] = { "antigravity": [], @@ -291,6 +302,8 @@ class UnifiedHealthService: } for pid, pcfg in sorted(config.profiles.items()): + if profile_id is not None and pid != profile_id: + continue prov = pcfg.provider if prov not in result: result[prov] = [] @@ -449,6 +462,11 @@ class UnifiedHealthService: return result + def refresh_profile(self, provider: str, profile_id: str) -> Optional[ProfileViewModel]: + """Rebuild exactly one profile ViewModel after an auth or quota change.""" + profiles = self.scan_all(force=True, profile_id=profile_id).get(provider, []) + return next((profile for profile in profiles if profile.profile_id == profile_id), None) + def get_cached_profiles(self) -> Dict[str, List[ProfileViewModel]]: """Return currently cached profiles grouped by provider without disk I/O.""" with self._lock: diff --git a/tests/test_import_invariants.py b/tests/test_import_invariants.py index ec8c4b1..40347fa 100644 --- a/tests/test_import_invariants.py +++ b/tests/test_import_invariants.py @@ -117,3 +117,17 @@ def test_gui_test_modules_guard_optional_ui_dependency() -> None: + ", ".join(offenders) + " — add pytest.importorskip('customtkinter') above the import" ) + + +@pytest.mark.unit +def test_antigravity_provider_loads_from_repo() -> None: + """Verify that antigravity_provider is loaded from the repository src, not from %LOCALAPPDATA%.""" + import antigravity_provider + import antigravity_provider.runtime + + pkg_file = Path(antigravity_provider.__file__).resolve() + runtime_file = Path(antigravity_provider.runtime.__file__).resolve() + + assert str(PACKAGE_ROOT.resolve()) in str(pkg_file) or str(PACKAGE_ROOT.resolve()) in str(pkg_file.parent) + assert str(PACKAGE_ROOT.resolve()) in str(runtime_file) + diff --git a/tests/test_state_layer_and_event_driven_quota.py b/tests/test_state_layer_and_event_driven_quota.py new file mode 100644 index 0000000..0296683 --- /dev/null +++ b/tests/test_state_layer_and_event_driven_quota.py @@ -0,0 +1,157 @@ +"""Comprehensive tests for Task A: State Layer, Event-Driven Quota, Seq-Guards, and OAuth Lifecycle.""" +from __future__ import annotations + +import time +import pytest +from datetime import datetime, timezone + +from antigravity_provider.router.event_bus import ( + EventBus, + EVENT_ACCOUNT_UPDATED, + EVENT_ACCOUNT_ADDED, + EVENT_ACCOUNT_REMOVED, + EVENT_QUOTA_UPDATED, + EVENT_ROUTING_UPDATED, + EVENT_SYSTEM_READINESS_CHANGED, +) +from antigravity_provider.router.state_store import HubStateStore, HubSnapshot +from antigravity_provider.router.account_identity import QuotaBucket, QuotaSnapshot +from antigravity_provider.router.quota_collector import AccountQuotaService +from antigravity_provider.router.scheduler import HermesRefreshScheduler +from antigravity_provider.router.unified_health import UnifiedHealthService + + +@pytest.mark.unit +def test_targeted_account_quota_delta_event(): + """Verify that updating an account's quota produces EVENT_QUOTA_UPDATED with exact account identifiers.""" + bus = EventBus.get() + store = HubStateStore.get() + + received_events = [] + + def _listener(name, payload): + received_events.append((name, payload)) + + bus.subscribe(EVENT_QUOTA_UPDATED, _listener) + + try: + bucket = QuotaBucket( + id="antigravity.claude.5h", + display_name="5h", + model_family="claude", + used_percent=100.0, + remaining_percent=0.0, + status="exhausted", + ) + snap = QuotaSnapshot( + account_id="ag-orch-primary", + provider="antigravity", + buckets=[bucket], + source="runtime_event", + ) + + store.apply_delta_quota_updated("antigravity", "ag-orch-primary", snap) + + assert len(received_events) >= 1 + name, payload = received_events[-1] + assert name == EVENT_QUOTA_UPDATED + assert payload["provider"] == "antigravity" + assert payload["profile_id"] == "ag-orch-primary" + assert payload["snapshot"] == snap + assert payload["snapshot"].is_estimated is False + finally: + bus.unsubscribe(EVENT_QUOTA_UPDATED, _listener) + + +@pytest.mark.unit +def test_seq_token_prevents_stale_refresh_clobber(): + """Verify that an out-of-order stale background response cannot overwrite fresher state.""" + store = HubStateStore.get() + + seq_fresh = store.next_seq() + snap_fresh = store.refresh(force_scan=False, seq=seq_fresh) + gen_fresh = snap_fresh.generation + + # Simulate a delayed/stale response from an earlier seq counter + seq_stale = seq_fresh - 1 + snap_after_stale = store.refresh(force_scan=False, seq=seq_stale) + + # Stale response must be rejected, retaining the fresh generation + assert snap_after_stale.generation == gen_fresh + assert store.refresh_skipped_total >= 1 + + +@pytest.mark.unit +def test_account_added_and_removed_delta_events(): + """Verify that account added and removed delta methods fire targeted events without global scan.""" + bus = EventBus.get() + store = HubStateStore.get() + + added_events = [] + removed_events = [] + + def _on_added(name, payload): + added_events.append(payload) + + def _on_removed(name, payload): + removed_events.append(payload) + + bus.subscribe(EVENT_ACCOUNT_ADDED, _on_added) + bus.subscribe(EVENT_ACCOUNT_REMOVED, _on_removed) + + try: + store.apply_delta_account_added("openai-codex", "codex-slot-2") + assert len(added_events) >= 1 + assert added_events[-1]["provider"] == "openai-codex" + assert added_events[-1]["profile_id"] == "codex-slot-2" + + store.apply_delta_account_removed("openai-codex", "codex-slot-2") + assert len(removed_events) >= 1 + assert removed_events[-1]["provider"] == "openai-codex" + assert removed_events[-1]["profile_id"] == "codex-slot-2" + finally: + bus.unsubscribe(EVENT_ACCOUNT_ADDED, _on_added) + bus.unsubscribe(EVENT_ACCOUNT_REMOVED, _on_removed) + + +@pytest.mark.unit +def test_provider_refresh_scheduler_execution(): + """Verify HermesRefreshScheduler.trigger_refresh_provider refreshes specific provider.""" + scheduler = HermesRefreshScheduler.get() + completed = [] + + def _on_done(): + completed.append(True) + + scheduler.trigger_refresh_provider("antigravity", on_complete=_on_done) + + # Wait briefly for worker thread + t0 = time.time() + while not completed and (time.time() - t0 < 3.0): + time.sleep(0.05) + + assert len(completed) == 1 + + +@pytest.mark.unit +def test_antigravity_claude_vs_gemini_quota_bucket_isolation(): + """Verify Antigravity quota separates Claude and Gemini model families cleanly.""" + snap = AccountQuotaService.get()._generate_baseline_snapshot("antigravity", "ag-orch-primary") + assert snap is not None + assert len(snap.buckets) >= 2 + + claude_bucket = snap.get_bucket_for_model("claude-3-7-sonnet") + gemini_bucket = snap.get_bucket_for_model("gemini-2.5-pro") + + assert claude_bucket is not None + assert gemini_bucket is not None + assert claude_bucket.model_family == "claude" + assert gemini_bucket.model_family == "gemini" + assert claude_bucket.id != gemini_bucket.id + + # Mark claude exhausted + claude_bucket.status = "exhausted" + claude_bucket.remaining_percent = 0.0 + + assert snap.is_model_available("claude-3-7-sonnet") is False + assert snap.is_model_available("gemini-2.5-pro") is True