merge: complete antigravity state layer

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 пакета сознательно не выполнялась.

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:

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]:

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:

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:

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"

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()
+            current = self._current_snapshot or self._build_empty_snapshot()
-            profile = next(
+            current_profile = current.get_profile(profile_id)
-                (item for items in cached.values() for item in items if item.profile_id == profile_id),
+            provider = provider or (current_profile.provider if current_profile else "")
-                None,
+            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,

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:

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)
+

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