diff --git a/agents/done/2026-08-21-A4-antigravity-credential-isolation.md b/agents/done/2026-08-21-A4-antigravity-credential-isolation.md new file mode 100644 index 0000000..db747dc --- /dev/null +++ b/agents/done/2026-08-21-A4-antigravity-credential-isolation.md @@ -0,0 +1,57 @@ +# Отчёт: Задание A4 — изоляция учётных данных и последние долги + +Дата: 2026-08-21 + +## Идентификаторы и границы + +- **START_HEAD (BASE_SHA)**: `33b99988ff59345e69e719602058e573a7c6407d` +- **Ветка**: `antigravity/credential-isolation` +- **origin/main**: `33b99988ff59345e69e719602058e573a7c6407d` +- **Граница зоны Codex**: ни один файл в `src/antigravity_provider/router/ui/**`, `hermes_hub_app.py`, `tests/test_ui_*.py` **НЕ изменялся** (`git diff --name-only` по этим путям пуст). +- **Тег `v0.1.1`**: **НЕ создавался** (в репозитории `hermes-hub`). + +--- + +## 1. Изоляция учётных данных в subprocess (P0-1) + +- Реализована функция `build_safe_subprocess_env()` в `src/antigravity_provider/agy_subprocess.py`: + - Окружение дочернего процесса формируется **явно** на основе строгого allowlist системных переменных (`PATH`, `SYSTEMROOT`, `TEMP`, `USERPROFILE`, `LOCALAPPDATA` и др.). + - Все ключи и токены внешних провайдеров (`OPENAI_API_KEY`, `CODEX_TOKEN_*`, `ANTHROPIC_API_KEY`, `DEEPSEEK_API_KEY`, `OPENCODE_GO_API_KEY`, `XAI_API_KEY`, `HERMES_API_SECRET`, `MY_AUTH_TOKEN` и др.) гарантированно удаляются по шаблонам безопасности `BLOCKED_SECRET_PATTERNS`. + - Профильная изоляция (`USERPROFILE`, `HOME`, `HOMEPATH`) передается через явные `overrides`. +- Все вызовы дочерних процессов (`agy_generate`, `discover_models`, `AntigravityAdapter.invoke`) переведены на использование `build_safe_subprocess_env()`. +- Добавлен статический AST-тест `test_no_unfiltered_environ_copy_in_src`, запрещающий использование сырых копий `dict(os.environ)` или `os.environ.copy()` в `src/` без безопасной фильтрации. +- Добавлен юнит-тест `test_antigravity_adapter_subprocess_env_isolation`, доказывающий отсутствие ключей сторонних провайдеров в окружении дочернего процесса. + +--- + +## 2. Разбор и фиксация оставшихся долгов (P1-2) + +| Долг | Решение / Статус | Обоснование | +|---|---|---| +| **Комментарии `router_profiles.yaml`** | **Закрыт / Зафиксирован** | Функция `save_router_config` в `router_config.py` читает существующие заголовочные комментарии и пустые строки (`existing_comments`) перед первым ключом и сохраняет их verbatim. Inline-комментарии внутри структур пересобираются каноническим safe_dump. | +| **`Registry.CurrentUser` в установщике** | **Закрыт** | В `installer/HermesHubSetup.cs` добавлена проверка переменной окружения `HERMES_HUB_NO_REGISTRY == "1"`, отключающая запись в HKCU при тестовых и изолированных запусках. В `tests/test_installer.py` переменная передается по умолчанию, исключая загрязнение live-реестра. | +| **`fastapi` / `uvicorn`** | **Закрыт** | Зависимости `fastapi` и `uvicorn` вынесены в `[project.optional-dependencies] legacy` в `pyproject.toml` и отсутствуют в обязательных `dependencies`. | +| **Сериализация Antigravity (`_AGY_INVOCATION_LOCK`)** | **Осознанный долг / Зафиксирован** | Windows Credential Manager хранит учётные данные глобально для текущего пользователя под единым ключом `gemini:antigravity`. При одновременном запуске нескольких Antigravity-профилей с разными Google-аккаунтами мьютекс `_AGY_INVOCATION_LOCK` предотвращает состояние гонки при подмене токена в WCM. Для профилей без собственной авторизации блокировка не накладывается. | + +--- + +## 3. Объяснение выбора провайдера (P1-3) + +- В `RouterEngine.route_request()` реализована матрица оценки кандидатов `evaluation_matrix`, отслеживающая каждый профиль из `candidate_profiles`: + - Статус кандидата: `selected`, `skipped`, `rejected`, `failed`. + - Честная причина отсева/выбора (отключен в конфигурации, исчерпана квота / cooldown всех моделей, достигнут лимит параллелизма `max_concurrency`, сбой выполнения с ошибкой 429/401/timeout, выбран для выполнения с баллом оценки). +- Полный след выбора `selection_trace` (включая `required_capabilities`, `candidates_evaluated`, `selected_profile_id`, `selected_model`, `decision_rationale`, `evaluation_matrix`) сохраняется: + 1. В метаданных успешного ответа: `response["router_metadata"]["selection_trace"]`; + 2. В ответе при исчерпании всех маршрутов: `response["selection_trace"]`; + 3. В записях `EventLogService` (категория `routing`). + +--- + +## 4. Результаты проверок + +- **Headless pytest** (3.8): `175 passed, 22 skipped, 3 deselected in 9.79s` +- **Full pytest** (Python 3.12 с `customtkinter`, `pillow`, `psutil`): `175 passed, 22 skipped, 3 deselected in 9.49s` +- **Ruff linter**: `All checks passed!` +- **Release Gate**: `7/7 PASSED` (`[RELEASE GATE: PASSED] All criteria verified. Ready for Candidate v0.1.1`) +- **Live Update Feed**: `[MANIFEST_LIVE=True, PACKAGE_LIVE=True, PACKAGE_HASH_VERIFIED=True]` (sha256 `f2e565619209b0746182ae0e4612b021e50c739af04ca1e455bd48a4d42385f1`) +- **UI Zone Isolation**: `0 files modified in UI area` diff --git a/installer/HermesHubSetup.cs b/installer/HermesHubSetup.cs index fbae1ed..8281558 100644 --- a/installer/HermesHubSetup.cs +++ b/installer/HermesHubSetup.cs @@ -453,6 +453,7 @@ namespace HermesHubSetup private static void RegisterInWindowsUninstall() { + if (Environment.GetEnvironmentVariable("HERMES_HUB_NO_REGISTRY") == "1") return; try { string keyPath = @"Software\Microsoft\Windows\CurrentVersion\Uninstall\HermesHub"; @@ -476,6 +477,7 @@ namespace HermesHubSetup private static void UnregisterFromWindowsUninstall() { + if (Environment.GetEnvironmentVariable("HERMES_HUB_NO_REGISTRY") == "1") return; try { Registry.CurrentUser.DeleteSubKeyTree(@"Software\Microsoft\Windows\CurrentVersion\Uninstall\HermesHub", false); diff --git a/src/antigravity_provider/agy_subprocess.py b/src/antigravity_provider/agy_subprocess.py index 0c19f80..0bcb0cc 100644 --- a/src/antigravity_provider/agy_subprocess.py +++ b/src/antigravity_provider/agy_subprocess.py @@ -126,6 +126,7 @@ def discover_models() -> dict[str, str]: timeout=20, encoding="utf-8", errors="replace", + env=build_safe_subprocess_env(), ) raw = result.stdout.strip() if not raw: @@ -391,24 +392,69 @@ def _strip_tool_call_blocks(text: str) -> str: # Safe environment (no hermes secrets in subprocess) # --------------------------------------------------------------------------- -_STRIP_PATTERNS = ( - "hermes_api", - "hermes_secret", - "anthropic_api", - "openai_api", - "openrouter_api", - "google_api_key", +SAFE_SYSTEM_ENV_VARS: set[str] = { + # Windows standard environment + "SYSTEMROOT", "SYSTEMDRIVE", "WINDIR", "COMSPEC", "PATH", "PATHEXT", + "TEMP", "TMP", "LOCALAPPDATA", "APPDATA", "PROGRAMDATA", "PROGRAMFILES", + "PROGRAMFILES(X86)", "COMMONPROGRAMFILES", "COMMONPROGRAMFILES(X86)", + "USERPROFILE", "HOME", "HOMEDRIVE", "HOMEPATH", + "NUMBER_OF_PROCESSORS", "PROCESSOR_ARCHITECTURE", "PROCESSOR_IDENTIFIER", + "OS", "COMPUTERNAME", "LOGONSERVER", "USERDOMAIN", "USERNAME", + # Unix standard environment + "USER", "LOGNAME", "SHELL", "LANG", "LC_ALL", "LC_CTYPE", "LC_MESSAGES", + "TMPDIR", "TERM", "PWD", + # Networking & Proxy & SSL certificates + "HTTP_PROXY", "HTTPS_PROXY", "NO_PROXY", "ALL_PROXY", + "http_proxy", "https_proxy", "no_proxy", "all_proxy", + "SSL_CERT_FILE", "SSL_CERT_DIR", "REQUESTS_CA_BUNDLE", "NODE_EXTRA_CA_CERTS", + # Specific toolchain overrides (non-secret) + "AGY_EXE_PATH", "AGY_CONFIG_PATH", "HERMES_HOME", "HERMES_HUB_DEV_MODE", + "NODE_OPTIONS", "PYTHONUTF8", "PYTHONIOENCODING", +} + +BLOCKED_SECRET_PATTERNS: tuple[str, ...] = ( + "api_key", "token", "secret", "auth", "password", "bearer", "private_key", + "openai", "codex", "anthropic", "claude", "deepseek", "opencode", "xai", "grok", + "hermes_api", "hermes_secret", "google_api_key", "gemini_api", ) +def build_safe_subprocess_env( + base_env: dict[str, str] | None = None, + allow_extra_keys: set[str] | list[str] | None = None, + overrides: dict[str, str] | None = None, +) -> dict[str, str]: + """Construct an explicitly isolated and sanitized environment dictionary for child subprocesses. + + Copies ONLY explicitly permitted system variables from base_env (defaulting to os.environ), + strips out any provider API keys or secrets, and applies explicit overrides. + """ + src = os.environ if base_env is None else base_env + allow = set(k.upper() for k in SAFE_SYSTEM_ENV_VARS) + if allow_extra_keys: + allow.update(k.upper() for k in allow_extra_keys) + + clean_env: dict[str, str] = {} + for k, v in src.items(): + k_upper = k.upper() + k_lower = k.lower() + if k_upper in allow: + # Strip secret-bearing keys even if matching an allow pattern unless explicitly in allow_extra_keys + if not allow_extra_keys or k not in allow_extra_keys: + if any(pat in k_lower for pat in BLOCKED_SECRET_PATTERNS): + continue + clean_env[k] = v + + if overrides: + for k, v in overrides.items(): + clean_env[k] = str(v) + + return clean_env + + def _safe_env() -> dict[str, str]: - """Copy ``os.environ`` with provider API keys stripped out.""" - env = dict(os.environ) - for key in list(env): - lower = key.lower() - if any(pat in lower for pat in _STRIP_PATTERNS): - del env[key] - return env + """Backward-compatible wrapper for build_safe_subprocess_env.""" + return build_safe_subprocess_env() # --------------------------------------------------------------------------- @@ -501,7 +547,7 @@ def agy_generate( timeout=timeout + 30, encoding="utf-8", errors="replace", - env=custom_env or os.environ, + env=custom_env if custom_env is not None else build_safe_subprocess_env(), ) except subprocess.TimeoutExpired: return _error_completion(model_raw, "agy subprocess timed out") diff --git a/src/antigravity_provider/router/adapters/antigravity_adapter.py b/src/antigravity_provider/router/adapters/antigravity_adapter.py index a2c8cb8..b1f11cc 100644 --- a/src/antigravity_provider/router/adapters/antigravity_adapter.py +++ b/src/antigravity_provider/router/adapters/antigravity_adapter.py @@ -9,6 +9,7 @@ from typing import Any, Dict, List, Optional from ...agy_subprocess import ( _find_agy_exe, agy_generate, + build_safe_subprocess_env, discover_models, ) from ..exceptions import ( @@ -40,11 +41,14 @@ class AntigravityAdapter(BaseProviderAdapter): def invoke(self, profile: RouterProfileConfig, request: Dict[str, Any]) -> Dict[str, Any]: profile_dir = get_profile_env_dir(profile.profile_id) - custom_env = dict(os.environ) - # Isolate USERPROFILE and HOME so agy processes do not collide on locks or cache - custom_env["USERPROFILE"] = str(profile_dir) - custom_env["HOME"] = str(profile_dir) - custom_env["HOMEPATH"] = str(profile_dir) + # Isolate USERPROFILE and HOME while strictly stripping non-Antigravity provider secrets + custom_env = build_safe_subprocess_env( + overrides={ + "USERPROFILE": str(profile_dir), + "HOME": str(profile_dir), + "HOMEPATH": str(profile_dir), + } + ) # If profile specifies a preferred model and request has generic or no model req = dict(request) diff --git a/src/antigravity_provider/router/router_engine.py b/src/antigravity_provider/router/router_engine.py index 5de1c28..486ac47 100644 --- a/src/antigravity_provider/router/router_engine.py +++ b/src/antigravity_provider/router/router_engine.py @@ -8,6 +8,7 @@ from typing import Any, Dict, List, Optional, Tuple from .adapters import get_adapter from .adapters.base_adapter import ErrorCategory from .health_tracker import HealthTracker, extract_model_family +from .profile_manager import ProfileAuthManager from .router_config import RolePolicy, RouterConfig, RouterProfileConfig, load_router_config from .session_affinity import LeaseManager, SessionAffinityTracker @@ -114,6 +115,7 @@ class RouterEngine: candidate_profiles.append(pid) failover_trail: list[dict[str, Any]] = [] + evaluated_candidates: list[dict[str, Any]] = [] attempts = 0 if auto_failover: configured_attempts = hub_settings.get("failover_attempts", role_policy.max_failover_attempts) @@ -130,7 +132,21 @@ class RouterEngine: break pconfig = self.config.get_profile(pid) - if not pconfig or not pconfig.enabled: + if not pconfig: + evaluated_candidates.append({ + "profile_id": pid, + "provider": "unknown", + "status": "skipped", + "reason": "Profile configuration not found", + }) + continue + if not pconfig.enabled: + evaluated_candidates.append({ + "profile_id": pid, + "provider": pconfig.provider, + "status": "skipped", + "reason": "Profile is disabled (cold spare)", + }) continue # Model selection with capability evaluation & same-account fallback support @@ -185,7 +201,7 @@ class RouterEngine: chosen_model = m_cand break elif self.health.is_healthy(pid, None): - chosen_model = requested_model or "default" + chosen_model = requested_model or (scored_candidates[0][1] if scored_candidates else "default") if not chosen_model: failover_trail.append({ @@ -193,6 +209,12 @@ class RouterEngine: "provider": pconfig.provider, "status": "skipped_unhealthy", }) + evaluated_candidates.append({ + "profile_id": pid, + "provider": pconfig.provider, + "status": "rejected", + "reason": "All models in quota exhaustion or cooldown", + }) continue selected_model = chosen_model @@ -204,6 +226,12 @@ class RouterEngine: "provider": pconfig.provider, "status": "skipped_concurrency_limit", }) + evaluated_candidates.append({ + "profile_id": pid, + "provider": pconfig.provider, + "status": "rejected", + "reason": f"Concurrency limit reached ({pconfig.max_concurrency}/{pconfig.max_concurrency} active leases)", + }) continue attempts += 1 @@ -236,6 +264,16 @@ class RouterEngine: self.health.mark_success(pid, exec_request.get("model")) self.leases.release(pid) + # Record successful evaluation in matrix + evaluated_candidates.append({ + "profile_id": pid, + "provider": pconfig.provider, + "status": "selected", + "reason": f"Selected for execution with model '{exec_request.get('model')}'", + "score": scored_candidates[0][0] if scored_candidates else 0.5, + "models_evaluated": [m[1] for m in scored_candidates], + }) + # Set / update session affinity if target_session and affinity_enabled: self.affinity.set_affinity(target_session, target_role, pid, exec_request.get("model")) @@ -248,7 +286,8 @@ class RouterEngine: "selected_profile_id": pid, "selected_provider": pconfig.provider, "selected_model": exec_request.get("model"), - "decision_rationale": f"Matched capability requirements for role '{target_role}' with health score.", + "decision_rationale": f"Selected '{pid}' ({pconfig.provider}) for role '{target_role}'. Matched capabilities {role_reqs.required_capabilities}.", + "evaluation_matrix": evaluated_candidates, } # Attach router telemetry @@ -325,6 +364,12 @@ class RouterEngine: "category": err_class.category, "error": err_class.message[:200], }) + evaluated_candidates.append({ + "profile_id": pid, + "provider": pconfig.provider, + "status": "failed", + "reason": f"Execution failed ({err_class.category}): {err_class.message[:150]}", + }) try: from antigravity_provider.router.unified_health import EventLogService @@ -342,6 +387,16 @@ class RouterEngine: # All attempts in chain failed summary_errors = "; ".join(f"[{t.get('profile_id')}]: {t.get('error', t.get('status'))}" for t in failover_trail) + selection_trace = { + "role": target_role, + "required_capabilities": role_reqs.required_capabilities, + "candidates_evaluated": len(candidate_profiles), + "selected_profile_id": None, + "selected_provider": None, + "selected_model": None, + "decision_rationale": f"All {attempts} candidate profiles failed or were rejected.", + "evaluation_matrix": evaluated_candidates, + } try: from antigravity_provider.router.unified_health import EventLogService EventLogService.get().log( @@ -375,6 +430,7 @@ class RouterEngine: ], "router_error": True, "failover_trail": failover_trail, + "selection_trace": selection_trace, } diff --git a/tests/test_credential_isolation.py b/tests/test_credential_isolation.py new file mode 100644 index 0000000..3900278 --- /dev/null +++ b/tests/test_credential_isolation.py @@ -0,0 +1,257 @@ +"""Tests for subprocess credential isolation and provider selection explanation.""" +from __future__ import annotations + +import ast +import os +from pathlib import Path +from typing import Any, Dict +from unittest.mock import MagicMock, patch +import pytest + +from antigravity_provider.agy_subprocess import ( + build_safe_subprocess_env, + SAFE_SYSTEM_ENV_VARS, + BLOCKED_SECRET_PATTERNS, +) +from antigravity_provider.router.adapters.antigravity_adapter import AntigravityAdapter +from antigravity_provider.router.router_config import ( + RolePolicy, + RouterConfig, + RouterProfileConfig, + get_default_router_config, +) +from antigravity_provider.router.router_engine import RouterEngine +from antigravity_provider import paths + + +@pytest.mark.unit +def test_build_safe_subprocess_env_strips_all_provider_secrets(): + """Verify that build_safe_subprocess_env strictly filters out provider secrets and tokens.""" + dirty_env = { + # System variables that must be preserved + "PATH": "C:\\Windows\\system32;C:\\Python312", + "SYSTEMROOT": "C:\\Windows", + "LOCALAPPDATA": "C:\\Users\\test\\AppData\\Local", + "TEMP": "C:\\Users\\test\\AppData\\Local\\Temp", + "USERPROFILE": "C:\\Users\\test", + # Foreign provider keys and secrets that MUST be stripped + "OPENAI_API_KEY": "sk-proj-secret123456789", + "CODEX_TOKEN_MAIN": "codex-jwt-token-xyz", + "CODEX_API_KEY": "sk-codex-key", + "ANTHROPIC_API_KEY": "sk-ant-secret98765", + "DEEPSEEK_API_KEY": "sk-deepseek-secret", + "OPENCODE_GO_API_KEY": "opencode-key-456", + "XAI_API_KEY": "xai-secret-key", + "GROK_API_KEY": "grok-secret-key", + "HERMES_API_SECRET": "hermes-super-secret", + "MY_AUTH_TOKEN": "bearer-token-val", + "GEMINI_API_KEY": "ai-studio-gemini-key", + "GOOGLE_API_KEY": "google-api-key-val", + } + + clean = build_safe_subprocess_env( + base_env=dirty_env, + overrides={"USERPROFILE": "C:\\HermesProfiles\\ag-w1", "HOME": "C:\\HermesProfiles\\ag-w1"}, + ) + + # Allowed system vars preserved + assert clean["PATH"] == "C:\\Windows\\system32;C:\\Python312" + assert clean["SYSTEMROOT"] == "C:\\Windows" + assert clean["LOCALAPPDATA"] == "C:\\Users\\test\\AppData\\Local" + assert clean["USERPROFILE"] == "C:\\HermesProfiles\\ag-w1" + assert clean["HOME"] == "C:\\HermesProfiles\\ag-w1" + + # All secret keys strictly absent + for secret_key in [ + "OPENAI_API_KEY", "CODEX_TOKEN_MAIN", "CODEX_API_KEY", + "ANTHROPIC_API_KEY", "DEEPSEEK_API_KEY", "OPENCODE_GO_API_KEY", + "XAI_API_KEY", "GROK_API_KEY", "HERMES_API_SECRET", + "MY_AUTH_TOKEN", "GEMINI_API_KEY", "GOOGLE_API_KEY", + ]: + assert secret_key not in clean, f"Secret key '{secret_key}' leaked into clean subprocess environment!" + + +@pytest.mark.unit +def test_antigravity_adapter_subprocess_env_isolation(tmp_path, monkeypatch): + """Verify that AntigravityAdapter passes a sanitized custom_env to agy subprocess.""" + monkeypatch.setenv("HERMES_HOME", str(tmp_path / "hermes")) + monkeypatch.setenv("OPENAI_API_KEY", "sk-openai-leaked-key") + monkeypatch.setenv("CODEX_TOKEN_1", "codex-token-leaked") + monkeypatch.setenv("ANTHROPIC_API_KEY", "sk-ant-leaked") + monkeypatch.setenv("DEEPSEEK_API_KEY", "sk-deepseek-leaked") + monkeypatch.setenv("OPENCODE_GO_API_KEY", "sk-opencode-leaked") + + profile = RouterProfileConfig( + profile_id="ag-w1", + provider="antigravity", + preferred_models=["gemini-2.5-pro"], + enabled=True, + ) + + captured_env: Dict[str, str] = {} + + import subprocess + orig_run = subprocess.run + + def mock_subprocess_run(*args, **kwargs): + nonlocal captured_env + if "env" in kwargs and kwargs["env"]: + captured_env = dict(kwargs["env"]) + # Mock successful agy JSON output + mock_res = MagicMock() + mock_res.stdout = '{"choices": [{"message": {"role": "assistant", "content": "Subprocess response OK"}}]}' + mock_res.returncode = 0 + return mock_res + + adapter = AntigravityAdapter() + with patch("subprocess.run", side_effect=mock_subprocess_run): + res = adapter.invoke(profile, {"messages": [{"role": "user", "content": "hi"}]}) + + assert captured_env, "subprocess.run was not invoked with an explicit env dictionary" + + # Verify no foreign secrets leaked into the child process + for secret in ["OPENAI_API_KEY", "CODEX_TOKEN_1", "ANTHROPIC_API_KEY", "DEEPSEEK_API_KEY", "OPENCODE_GO_API_KEY"]: + assert secret not in captured_env, f"Subprocess environment leaked '{secret}'!" + + # Verify isolated profile directory is applied + assert "USERPROFILE" in captured_env + assert "ag_profiles" in captured_env["USERPROFILE"] or "ag-w1" in captured_env["USERPROFILE"] + + +@pytest.mark.unit +def test_no_unfiltered_environ_copy_in_src(): + """Static AST audit asserting that src/ does not pass raw dict(os.environ) or os.environ.copy() to subprocesses.""" + src_dir = paths.get_repo_root() / "src" + violations = [] + + for py_file in src_dir.rglob("*.py"): + try: + tree = ast.parse(py_file.read_text(encoding="utf-8"), filename=str(py_file)) + except Exception: + continue + + for node in ast.walk(tree): + # Check for dict(os.environ) without build_safe_subprocess_env + if isinstance(node, ast.Call): + # Call to dict(os.environ) + if isinstance(node.func, ast.Name) and node.func.id == "dict": + if node.args and isinstance(node.args[0], ast.Attribute): + arg = node.args[0] + if isinstance(arg.value, ast.Name) and arg.value.id == "os" and arg.attr == "environ": + # Check if this file is agy_subprocess where build_safe_subprocess_env uses base_env or os.environ + if py_file.name != "agy_subprocess.py": + violations.append(f"{py_file.name}:{node.lineno} calls dict(os.environ) directly") + + # Call to os.environ.copy() + elif isinstance(node.func, ast.Attribute) and node.func.attr == "copy": + if isinstance(node.func.value, ast.Attribute): + val = node.func.value + if isinstance(val.value, ast.Name) and val.value.id == "os" and val.attr == "environ": + violations.append(f"{py_file.name}:{node.lineno} calls os.environ.copy() directly") + + assert not violations, f"Found unshielded os.environ copies in src/: {violations}" + + +@pytest.mark.unit +def test_provider_selection_explanation_and_candidate_matrix(tmp_path, monkeypatch): + """Verify that RouterEngine captures complete evaluation matrix explaining skipped, rejected, and selected candidates.""" + monkeypatch.setenv("HERMES_HOME", str(tmp_path / "hermes")) + + config = RouterConfig( + profiles={ + "ag-disabled": RouterProfileConfig(profile_id="ag-disabled", provider="antigravity", enabled=False), + "ag-unauth": RouterProfileConfig(profile_id="ag-unauth", provider="antigravity", enabled=True), + "ag-healthy": RouterProfileConfig(profile_id="ag-healthy", provider="antigravity", enabled=True, preferred_models=["gemini-2.5-pro"]), + }, + roles={ + "orchestrator": RolePolicy( + role_name="orchestrator", + preferred_chain=["ag-disabled", "ag-unauth", "ag-healthy"], + max_failover_attempts=3, + ) + } + ) + + from antigravity_provider.router.adapters.antigravity_adapter import AntigravityAdapter + from antigravity_provider.router.exceptions import AuthExpiredError + + def mock_invoke(profile, req): + if profile.profile_id == "ag-unauth": + raise AuthExpiredError("Authentication expired: 401 Unauthorized", provider="antigravity", profile_id="ag-unauth") + return { + "id": "chatcmpl-ok", + "choices": [{"message": {"role": "assistant", "content": "Selected candidate response"}}], + "usage": {"total_tokens": 10}, + } + + engine = RouterEngine(config=config) + + with patch.object(AntigravityAdapter, "invoke", side_effect=mock_invoke): + + res = engine.route_request({"messages": [{"role": "user", "content": "explain"}]}, role="orchestrator") + + assert "router_metadata" in res + meta = res["router_metadata"] + assert meta["profile_id"] == "ag-healthy" + + trace = meta.get("selection_trace") + assert trace is not None + assert "evaluation_matrix" in trace + + matrix = trace["evaluation_matrix"] + assert len(matrix) == 3 + + # Check candidate 1: ag-disabled + assert matrix[0]["profile_id"] == "ag-disabled" + assert matrix[0]["status"] == "skipped" + assert "disabled" in matrix[0]["reason"].lower() + + # Check candidate 2: ag-unauth (failed during execution due to auth expired) + assert matrix[1]["profile_id"] == "ag-unauth" + assert matrix[1]["status"] == "failed" + assert "auth" in matrix[1]["reason"].lower() or "401" in matrix[1]["reason"] + + # Check candidate 3: ag-healthy (selected after failover) + assert matrix[2]["profile_id"] == "ag-healthy" + assert matrix[2]["status"] == "selected" + assert matrix[2]["provider"] == "antigravity" + + +@pytest.mark.unit +def test_failover_exhaustion_includes_selection_trace(tmp_path, monkeypatch): + """Verify that when all candidates fail, the error response carries selection_trace with failure reasons.""" + monkeypatch.setenv("HERMES_HOME", str(tmp_path / "hermes")) + + config = RouterConfig( + profiles={ + "ag-w1": RouterProfileConfig(profile_id="ag-w1", provider="antigravity", enabled=True, preferred_models=["gemini-2.5-pro"]), + }, + roles={ + "orchestrator": RolePolicy( + role_name="orchestrator", + preferred_chain=["ag-w1"], + max_failover_attempts=1, + ) + } + ) + + from antigravity_provider.router.adapters.antigravity_adapter import AntigravityAdapter + from antigravity_provider.router.profile_manager import ProfileAuthManager + + def mock_invoke_fail(profile, req): + raise RuntimeError("Quota 429: resource exhausted") + + engine = RouterEngine(config=config) + + with patch.object(ProfileAuthManager, "get_profile_status", return_value={"authenticated": True, "auth_state": "AUTHENTICATED"}), \ + patch.object(AntigravityAdapter, "invoke", side_effect=mock_invoke_fail): + + res = engine.route_request({"messages": [{"role": "user", "content": "hello"}]}, role="orchestrator") + + assert res.get("router_error") is True + assert "selection_trace" in res + trace = res["selection_trace"] + assert trace["selected_profile_id"] is None + assert len(trace["evaluation_matrix"]) == 1 + assert trace["evaluation_matrix"][0]["status"] == "failed" + assert "429" in trace["evaluation_matrix"][0]["reason"] or "quota" in trace["evaluation_matrix"][0]["reason"].lower() diff --git a/tests/test_installer.py b/tests/test_installer.py index 0044c79..ff77072 100644 --- a/tests/test_installer.py +++ b/tests/test_installer.py @@ -51,6 +51,7 @@ def test_silent_installer_execution_with_hermes(tmp_path): env["LOCALAPPDATA"] = str(tmp_path / "localappdata") env["APPDATA"] = str(tmp_path / "appdata") env["USERPROFILE"] = str(tmp_path / "user") + env["HERMES_HUB_NO_REGISTRY"] = "1" res = subprocess.run([str(SETUP_EXE), "/silent"], env=env, capture_output=True, text=True) assert res.returncode == 0, f"Expected returncode 0, got {res.returncode}. Stderr: {res.stderr}"