From 59d57a41ef3c409ded22498282f8dc49380619e7 Mon Sep 17 00:00:00 2001 From: Hermes Team Date: Wed, 26 Aug 2026 09:27:36 +0700 Subject: [PATCH] feat(a31): preflight dependency agent, workflow run state recovery, local concurrency and context guard, PII email masking, and cost-controller token honesty --- docs/web-api/CONTRACT.md | 6 +- src/antigravity_provider/paths.py | 5 + .../router/action_handler.py | 7 + .../router/adapters/local_adapter.py | 110 ++++- .../router/preflight_service.py | 394 +++++++++++++++++ .../router/role_registry.py | 23 + .../router/router_config.py | 8 +- .../router/settings_service.py | 7 + .../router/telemetry_service.py | 143 ++++++- src/antigravity_provider/router/web/server.py | 31 +- .../router/web/static/app.js | 80 ++++ .../router/web/static/index.html | 27 ++ .../router/workflow_service.py | 141 ++++++- .../test_a31_preflight_state_batching_pii.py | 397 ++++++++++++++++++ 14 files changed, 1365 insertions(+), 14 deletions(-) create mode 100644 src/antigravity_provider/router/preflight_service.py create mode 100644 tests/test_a31_preflight_state_batching_pii.py diff --git a/docs/web-api/CONTRACT.md b/docs/web-api/CONTRACT.md index 542088b..0ac8825 100644 --- a/docs/web-api/CONTRACT.md +++ b/docs/web-api/CONTRACT.md @@ -102,6 +102,7 @@ readiness, agents, providers, routing, quotas, metrics, is_stale | `save_workflow` | Валидировать и сохранить узлы, рёбра, layout и предел итераций | `edges`, `agents`, `max_iterations` | | `start_workflow` | Запустить реальную задачу через RouterEngine | `task` | | `stop_workflow` | Запросить остановку текущего запуска | — | +| `run_preflight` | Запустить zero-quota проверку зависимостей, CLI, Python окружения и локальных серверов | — | Состояние графа и LIVE-журнал приходят в поле `workflow` ответа `GET /api/snapshot`. `workflow.run.status=loading` означает загрузку; @@ -113,10 +114,11 @@ account_details add_account agent_settings apply_update assign_role auto_assign_all check_updates delete_credentials edit_route get_update_status oauth open_routing refresh_account refresh_all refresh_data refresh_models -reorder_chain save_chain save_settings set_main -set_model set_orchestrator test +reorder_chain run_preflight save_chain save_settings +set_main set_model set_orchestrator test ``` + Ответ: ```json diff --git a/src/antigravity_provider/paths.py b/src/antigravity_provider/paths.py index 5088675..e116928 100644 --- a/src/antigravity_provider/paths.py +++ b/src/antigravity_provider/paths.py @@ -95,6 +95,11 @@ def get_workflow_state_path() -> Path: return get_config_dir() / "workflow_state.json" +def get_workflow_run_state_path() -> Path: + """Return the persisted active workflow run state file.""" + return get_config_dir() / "workflow_run_state.json" + + def get_agent_files_dir() -> Path: """Return the user-editable directory containing real Agent Files.""" directory = get_hermes_home() / "agents" diff --git a/src/antigravity_provider/router/action_handler.py b/src/antigravity_provider/router/action_handler.py index d9f2aba..f68fd23 100644 --- a/src/antigravity_provider/router/action_handler.py +++ b/src/antigravity_provider/router/action_handler.py @@ -630,5 +630,12 @@ class ActionExecutor: status = mgr.get_status_dict() return {'ok': True, 'message': status.get('message') or 'Статус получен', 'data': status} + elif action == 'run_preflight': + from antigravity_provider.router.preflight_service import PreflightCheckService + service = PreflightCheckService.get() + report = service.run_all_checks() + msg = f"Проверка готовности: {report.passed_count} успешно, {report.failed_count} ошибок, {report.warn_count} предупреждений" + return {'ok': report.success, 'message': msg, 'data': report.to_dict()} + else: return {'ok': False, 'message': f'Неизвестное действие: {action}', 'unknown': True} diff --git a/src/antigravity_provider/router/adapters/local_adapter.py b/src/antigravity_provider/router/adapters/local_adapter.py index ad61621..a632cad 100644 --- a/src/antigravity_provider/router/adapters/local_adapter.py +++ b/src/antigravity_provider/router/adapters/local_adapter.py @@ -46,6 +46,79 @@ class LocalLLMAdapter(BaseProviderAdapter): return val return None + _context_window_cache: Dict[str, int] = {} + + def get_context_window( + self, + profile: RouterProfileConfig, + model: Optional[str] = None, + query_remote: bool = False, + ) -> Optional[int]: + """Fetch actual context_window / max_context_length from profile config or /models endpoint. + + Never invents or hardcodes defaults. Returns None if unknown. + """ + # 1. Profile auth_config / custom settings + for key in ("context_window", "context_length", "max_context_length", "max_tokens_limit", "n_ctx"): + if key in profile.auth_config and profile.auth_config[key]: + try: + return int(profile.auth_config[key]) + except (ValueError, TypeError): + pass + + # 2. In-memory cache from previous model discovery + cache_key = f"{profile.profile_id}:{model or 'default'}" + if cache_key in self._context_window_cache: + return self._context_window_cache[cache_key] + if f"{profile.profile_id}:all" in self._context_window_cache: + return self._context_window_cache[f"{profile.profile_id}:all"] + + if not query_remote: + return None + + # 3. Query /models endpoint + base_url = self._resolve_base_url(profile) + api_key = self._resolve_api_key(profile) + headers = {"Accept": "application/json", "User-Agent": "hermes-router/1.0"} + if api_key: + headers["Authorization"] = f"Bearer {api_key}" + + req = urllib.request.Request(f"{base_url}/models", headers=headers, method="GET") + try: + with urllib.request.urlopen(req, timeout=3) as resp: + data = json.loads(resp.read().decode("utf-8", errors="replace")) + items = data.get("data") or data.get("models") or [] + if isinstance(items, list): + for m in items: + if isinstance(m, dict): + m_id = str(m.get("id") or m.get("name") or "") + for ck in ("context_window", "context_length", "max_model_len", "max_context_length", "n_ctx"): + if ck in m and m[ck]: + try: + ctx_val = int(m[ck]) + self._context_window_cache[f"{profile.profile_id}:{m_id}"] = ctx_val + self._context_window_cache[f"{profile.profile_id}:all"] = ctx_val + if not model or m_id == model or model in m_id or m_id in model or len(items) == 1: + return ctx_val + except (ValueError, TypeError): + pass + meta = m.get("meta") or {} + if isinstance(meta, dict): + for ck in ("n_ctx", "context_length", "max_context_length"): + if ck in meta and meta[ck]: + try: + ctx_val = int(meta[ck]) + self._context_window_cache[f"{profile.profile_id}:{m_id}"] = ctx_val + self._context_window_cache[f"{profile.profile_id}:all"] = ctx_val + if not model or m_id == model or model in m_id or m_id in model or len(items) == 1: + return ctx_val + except (ValueError, TypeError): + pass + except Exception as exc: + logger.debug("Failed to query context window from server for %s: %s", profile.profile_id, exc) + + return None + def invoke(self, profile: RouterProfileConfig, request: Dict[str, Any]) -> Dict[str, Any]: base_url = self._resolve_base_url(profile) api_key = self._resolve_api_key(profile) @@ -54,9 +127,43 @@ class LocalLLMAdapter(BaseProviderAdapter): if not model or model == "default": model = profile.preferred_models[0] if profile.preferred_models else "default" + messages = list(request.get("messages", [])) + + # Context Truncation Guard: safely bound prompt if context_window is known to prevent VRAM overflow + context_window = self.get_context_window(profile, model, query_remote=False) + if context_window is not None and context_window > 0 and len(messages) > 1: + max_tok = int(request.get("max_tokens", 0) or 0) + token_budget = context_window - max_tok - 64 + if token_budget > 100: + def _est_tok(msgs: list) -> int: + total_chars = sum(len(str(m.get("content", ""))) for m in msgs if isinstance(m, dict)) + return int(total_chars / 3.5) + len(msgs) * 4 + + if _est_tok(messages) > token_budget: + logger.warning( + "Context truncation guard active for %s: prompt exceeds context window (%d). Truncating middle messages.", + profile.profile_id, + context_window, + ) + system_msg = [messages[0]] if messages and messages[0].get("role") == "system" else [] + last_msg = messages[-1] + middle = messages[1:-1] if system_msg else messages[:-1] + + while middle and _est_tok(system_msg + middle + [last_msg]) > token_budget: + middle.pop(0) + + if _est_tok(system_msg + middle + [last_msg]) > token_budget: + avail_chars = max(100, int(token_budget * 3.0)) + last_copy = dict(last_msg) + last_copy["content"] = str(last_copy.get("content", ""))[-avail_chars:] + messages = system_msg + middle + [last_copy] + else: + messages = system_msg + middle + [last_msg] + + payload: Dict[str, Any] = { "model": model, - "messages": request.get("messages", []), + "messages": messages, "temperature": request.get("temperature", 0.7), } if "tools" in request and request["tools"]: @@ -105,6 +212,7 @@ class LocalLLMAdapter(BaseProviderAdapter): return data + @staticmethod def _reject_empty_answer(data: Dict[str, Any]) -> None: """Пустой ответ — это отказ, а не успех. diff --git a/src/antigravity_provider/router/preflight_service.py b/src/antigravity_provider/router/preflight_service.py new file mode 100644 index 0000000..e22c00b --- /dev/null +++ b/src/antigravity_provider/router/preflight_service.py @@ -0,0 +1,394 @@ +"""Hermes Hub — Preflight Dependency & Readiness Check Service (Dependency Agent). + +Performs comprehensive zero-quota preflight validation of local environment, CLI tools, +Python dependencies, local inference endpoints, role chain credentials, and disk permissions. +""" +from __future__ import annotations + +import importlib.util +import logging +import os +import shutil +import urllib.error +import urllib.request +from dataclasses import asdict, dataclass, field +from typing import Any, Dict, List, Optional + +from antigravity_provider import paths +from antigravity_provider.router.profile_manager import ProfileAuthManager +from antigravity_provider.router.router_config import load_router_config + +logger = logging.getLogger("hermes.router.preflight") + + +@dataclass +class PreflightItem: + check_id: str + name: str + status: str # "PASS" | "FAIL" | "WARN" + message: str + remediation: Optional[str] = None + + def to_dict(self) -> Dict[str, Any]: + return asdict(self) + + +@dataclass +class PreflightReport: + success: bool + passed_count: int + failed_count: int + warn_count: int + checks: List[PreflightItem] = field(default_factory=list) + + def to_dict(self) -> Dict[str, Any]: + return { + "success": self.success, + "passed_count": self.passed_count, + "failed_count": self.failed_count, + "warn_count": self.warn_count, + "checks": [c.to_dict() for c in self.checks], + } + + +class PreflightCheckService: + """Zero-quota dependency and readiness inspection service.""" + + _instance: Optional[PreflightCheckService] = None + + @classmethod + def get(cls) -> PreflightCheckService: + if cls._instance is None: + cls._instance = cls() + return cls._instance + + def check_cli_dependencies(self) -> List[PreflightItem]: + """Check for external CLI executables and critical Python packages.""" + items: List[PreflightItem] = [] + + # 1. Antigravity CLI (agy) + agy_path = shutil.which("agy") or shutil.which("agy.exe") + if agy_path: + items.append( + PreflightItem( + check_id="cli_agy", + name="CLI Antigravity (agy)", + status="PASS", + message=f"Исполняемый файл agy найден: {agy_path}", + ) + ) + else: + items.append( + PreflightItem( + check_id="cli_agy", + name="CLI Antigravity (agy)", + status="WARN", + message="Утилита 'agy' не найдена в системном PATH.", + remediation="Установите agy CLI или добавьте каталог установки в системную переменную PATH.", + ) + ) + + # 2. Python package: fastapi + fastapi_spec = importlib.util.find_spec("fastapi") + if fastapi_spec is not None: + items.append( + PreflightItem( + check_id="pkg_fastapi", + name="Библиотека FastAPI", + status="PASS", + message="Пакет fastapi успешно импортируется в окружении.", + ) + ) + else: + items.append( + PreflightItem( + check_id="pkg_fastapi", + name="Библиотека FastAPI", + status="FAIL", + message="Пакет 'fastapi' не установлен в текущем Python окружении.", + remediation="Выполните 'pip install fastapi' для работы веб-интерфейса и REST API.", + ) + ) + + # 3. Python package: uvicorn + uvicorn_spec = importlib.util.find_spec("uvicorn") + if uvicorn_spec is not None: + items.append( + PreflightItem( + check_id="pkg_uvicorn", + name="Библиотека Uvicorn", + status="PASS", + message="Пакет uvicorn успешно импортируется в окружении.", + ) + ) + else: + items.append( + PreflightItem( + check_id="pkg_uvicorn", + name="Библиотека Uvicorn", + status="FAIL", + message="Пакет 'uvicorn' не установлен в текущем Python окружении.", + remediation="Выполните 'pip install uvicorn' для запуска веб-сервера.", + ) + ) + + return items + + def check_local_servers(self) -> List[PreflightItem]: + """Poll {base_url}/models with 2.0s timeout for active local provider profiles.""" + items: List[PreflightItem] = [] + config = load_router_config() + local_profiles = [p for p in config.profiles.values() if p.provider == "local" and p.enabled] + + if not local_profiles: + items.append( + PreflightItem( + check_id="local_servers_none", + name="Локальные серверы LLM", + status="PASS", + message="Активные локальные профили (llama.cpp/vLLM) не настроены.", + ) + ) + return items + + from antigravity_provider.router.adapters.local_adapter import LocalLLMAdapter + + adapter = LocalLLMAdapter() + + for pcfg in local_profiles: + base_url = adapter._resolve_base_url(pcfg) + api_key = adapter._resolve_api_key(pcfg) + headers = {"Accept": "application/json", "User-Agent": "hermes-preflight/1.0"} + if api_key: + headers["Authorization"] = f"Bearer {api_key}" + + models_url = f"{base_url}/models" + try: + req = urllib.request.Request(models_url, headers=headers, method="GET") + with urllib.request.urlopen(req, timeout=2.0) as resp: + if resp.status in (200, 204): + items.append( + PreflightItem( + check_id=f"local_srv_{pcfg.profile_id}", + name=f"Локальный сервер {pcfg.profile_id} ({base_url})", + status="PASS", + message=f"Локальный сервер доступен (HTTP {resp.status}).", + ) + ) + else: + items.append( + PreflightItem( + check_id=f"local_srv_{pcfg.profile_id}", + name=f"Локальный сервер {pcfg.profile_id} ({base_url})", + status="FAIL", + message=f"Сервер вернул неожиданный статус HTTP {resp.status}", + remediation=f"Проверьте настройки и логи сервера {base_url}.", + ) + ) + except urllib.error.HTTPError as http_err: + items.append( + PreflightItem( + check_id=f"local_srv_{pcfg.profile_id}", + name=f"Локальный сервер {pcfg.profile_id} ({base_url})", + status="FAIL", + message=f"HTTP ошибка при обращении к {models_url}: {http_err.code} {http_err.reason}", + remediation=f"Убедитесь, что сервер на {base_url} поддерживает OpenAI-совместимый эндпоинт /v1/models.", + ) + ) + except Exception as exc: + items.append( + PreflightItem( + check_id=f"local_srv_{pcfg.profile_id}", + name=f"Локальный сервер {pcfg.profile_id} ({base_url})", + status="FAIL", + message=f"Не удалось подключиться к {base_url}: {exc}", + remediation=f"Запустите локальный сервер llama.cpp / vLLM / Ollama по адресу {base_url}.", + ) + ) + + return items + + def check_auth_credentials(self) -> List[PreflightItem]: + """Verify credential presence for all profiles referenced in active role chains. + + ZERO QUOTA BURN: Only inspects local auth files and keyring status. Never calls paid APIs. + """ + items: List[PreflightItem] = [] + config = load_router_config() + + # Collect all profile IDs in active role chains + referenced_pids: set[str] = set() + for role_policy in config.roles.values(): + for pid in role_policy.preferred_chain: + referenced_pids.add(pid) + + if not referenced_pids: + items.append( + PreflightItem( + check_id="auth_chains_empty", + name="Учетные данные цепочек ролей", + status="WARN", + message="В активных ролях не настроены цепочки профилей.", + remediation="Настройте цепочки профилей в разделе Маршрутизация.", + ) + ) + return items + + for pid in sorted(referenced_pids): + pcfg = config.get_profile(pid) + if not pcfg: + items.append( + PreflightItem( + check_id=f"auth_{pid}", + name=f"Профиль {pid}", + status="FAIL", + message=f"Профиль '{pid}' указан в цепочке роли, но отсутствует в конфигурации.", + remediation=f"Удалите '{pid}' из цепочки роли или настройте профиль в router_profiles.yaml.", + ) + ) + continue + + status = ProfileAuthManager.get_profile_status(pcfg.provider, pid) + is_authenticated = status.get("authenticated", False) + is_expired = status.get("is_expired", False) or status.get("expired", False) or status.get("status") == "EXPIRED" + + if is_authenticated and not is_expired: + items.append( + PreflightItem( + check_id=f"auth_{pid}", + name=f"Авторизация {pid} ({pcfg.provider})", + status="PASS", + message="Учетные данные действительны и сохранены локально.", + ) + ) + elif is_expired: + items.append( + PreflightItem( + check_id=f"auth_{pid}", + name=f"Авторизация {pid} ({pcfg.provider})", + status="FAIL", + message=f"Срок действия авторизации для профиля '{pid}' истек.", + remediation=f"Выполните повторный вход для профиля {pid} в разделе Аккаунты.", + ) + ) + else: + items.append( + PreflightItem( + check_id=f"auth_{pid}", + name=f"Авторизация {pid} ({pcfg.provider})", + status="FAIL", + message=f"Учетные данные для профиля '{pid}' ({pcfg.provider}) не найдены.", + remediation=f"Подключите профиль {pid} через кнопку 'Добавить аккаунт' или 'hermes router login'.", + ) + ) + + return items + + def check_system_environment(self) -> List[PreflightItem]: + """Verify HERMES_HOME presence and read/write permissions for config and logs.""" + items: List[PreflightItem] = [] + + # 1. HERMES_HOME directory + try: + home_dir = paths.get_hermes_home() + if home_dir.is_dir(): + items.append( + PreflightItem( + check_id="env_hermes_home", + name="Каталог HERMES_HOME", + status="PASS", + message=f"Каталог существует: {home_dir}", + ) + ) + else: + items.append( + PreflightItem( + check_id="env_hermes_home", + name="Каталог HERMES_HOME", + status="FAIL", + message=f"Каталог {home_dir} не существует или не является директорией.", + remediation="Проверьте права доступа и создайте каталог HERMES_HOME.", + ) + ) + except Exception as exc: + items.append( + PreflightItem( + check_id="env_hermes_home", + name="Каталог HERMES_HOME", + status="FAIL", + message=f"Ошибка доступа к HERMES_HOME: {exc}", + remediation="Убедитесь, что переменная HERMES_HOME указывает на корректный доступный путь.", + ) + ) + + # 2. Config Directory Write Test + try: + config_dir = paths.get_config_dir() + test_file = config_dir / f".preflight_probe_{os.getpid()}.tmp" + test_file.write_text("probe", encoding="utf-8") + test_file.unlink() + items.append( + PreflightItem( + check_id="env_config_writable", + name="Права на запись в каталог конфигурации", + status="PASS", + message=f"Права на запись в {config_dir} подтверждены.", + ) + ) + except Exception as exc: + items.append( + PreflightItem( + check_id="env_config_writable", + name="Права на запись в каталог конфигурации", + status="FAIL", + message=f"Нет прав на запись в {paths.get_config_dir()}: {exc}", + remediation="Предоставьте текущему пользователю права на запись в каталог конфигурации.", + ) + ) + + # 3. Logs Directory Write Test + try: + logs_dir = paths.get_logs_dir() + test_file = logs_dir / f".preflight_probe_{os.getpid()}.tmp" + test_file.write_text("probe", encoding="utf-8") + test_file.unlink() + items.append( + PreflightItem( + check_id="env_logs_writable", + name="Права на запись в каталог логов", + status="PASS", + message=f"Права на запись в {logs_dir} подтверждены.", + ) + ) + except Exception as exc: + items.append( + PreflightItem( + check_id="env_logs_writable", + name="Права на запись в каталог логов", + status="FAIL", + message=f"Нет прав на запись в {paths.get_logs_dir()}: {exc}", + remediation="Предоставьте текущему пользователю права на запись в каталог логов.", + ) + ) + + return items + + def run_all_checks(self) -> PreflightReport: + """Run all readiness checks and return aggregated PreflightReport.""" + all_items: List[PreflightItem] = [] + all_items.extend(self.check_cli_dependencies()) + all_items.extend(self.check_system_environment()) + all_items.extend(self.check_auth_credentials()) + all_items.extend(self.check_local_servers()) + + passed = sum(1 for item in all_items if item.status == "PASS") + failed = sum(1 for item in all_items if item.status == "FAIL") + warn = sum(1 for item in all_items if item.status == "WARN") + + return PreflightReport( + success=(failed == 0), + passed_count=passed, + failed_count=failed, + warn_count=warn, + checks=all_items, + ) diff --git a/src/antigravity_provider/router/role_registry.py b/src/antigravity_provider/router/role_registry.py index e133560..fb749b6 100644 --- a/src/antigravity_provider/router/role_registry.py +++ b/src/antigravity_provider/router/role_registry.py @@ -178,6 +178,18 @@ CANONICAL_ROLES: Dict[str, RoleDefinition] = { max_failover_attempts=3, tier="expert", ), + "dependency-agent": RoleDefinition( + role_id="dependency-agent", + display_name_ru="Проверяющий готовность", + short_name_ru="Готовность", + description_ru="До начала задачи убеждается, что на месте всё необходимое — исполняемые файлы и CLI, библиотеки, учётные данные, права доступа, доступность локальных серверов. Сообщает о нехватке до запуска.", + is_implemented=True, + capabilities=["dependency-agent", "preflight", "environment", "system_checks", "fast"], + fallback_capabilities=["dependency-agent", "preflight"], + default_preferred_chain=["opengo-1", "ag-w1", "codex-worker-1"], + max_failover_attempts=3, + tier="qa_doc", + ), } _CANONICAL_ROLE_ALIASES: Dict[str, str] = { @@ -215,6 +227,13 @@ _CANONICAL_ROLE_ALIASES: Dict[str, str] = { "специалист по интеграции": "integration-expert", "безопасность": "security-expert", "специалист по безопасности": "security-expert", + "dependency-agent": "dependency-agent", + "dependency_agent": "dependency-agent", + "preflight": "dependency-agent", + "проверяющий готовность": "dependency-agent", + "агент зависимостей": "dependency-agent", + "готовность": "dependency-agent", + "dependency": "dependency-agent", } class RoleRegistry: @@ -272,6 +291,10 @@ class RoleRegistry: clean = name_or_alias.strip().lower() return _CANONICAL_ROLE_ALIASES.get(clean, clean) + @classmethod + def resolve_role_name(cls, name_or_alias: str) -> str: + return cls.resolve_canonical_role(name_or_alias) + @classmethod def get_canonical_role_map(cls) -> Dict[str, str]: return dict(_CANONICAL_ROLE_ALIASES) diff --git a/src/antigravity_provider/router/router_config.py b/src/antigravity_provider/router/router_config.py index 5b76a5f..465bfe0 100644 --- a/src/antigravity_provider/router/router_config.py +++ b/src/antigravity_provider/router/router_config.py @@ -303,16 +303,20 @@ def load_router_config(config_path: Optional[Path] = None) -> RouterConfig: profiles_raw = data.get("profiles", {}) profiles: dict[str, RouterProfileConfig] = {} for pid, pdata in profiles_raw.items(): + provider = pdata.get("provider", "antigravity") + max_concurrency = int(pdata.get("max_concurrency", 1)) + if provider == "local": + max_concurrency = 1 profiles[pid] = RouterProfileConfig( profile_id=pid, - provider=pdata.get("provider", "antigravity"), + provider=provider, account_id=pdata.get("account_id", pid), capabilities=list(pdata.get("capabilities", [])), preferred_models=list(pdata.get("preferred_models", [])), fallback_models=list(pdata.get("fallback_models", [])), auth_config=dict(pdata.get("auth_config", {})), enabled=bool(pdata.get("enabled", True)), - max_concurrency=int(pdata.get("max_concurrency", 1)), + max_concurrency=max_concurrency, custom_base_url=pdata.get("custom_base_url"), ) diff --git a/src/antigravity_provider/router/settings_service.py b/src/antigravity_provider/router/settings_service.py index 0e7182e..1fe66e3 100644 --- a/src/antigravity_provider/router/settings_service.py +++ b/src/antigravity_provider/router/settings_service.py @@ -22,9 +22,11 @@ DEFAULT_SETTINGS: Dict[str, Any] = { "monitoring_interval_seconds": 30, "quota_threshold_percent": 10.0, "quota_threshold_action": "notify", + "email_masking_mode": "none", } + _SETTINGS_CACHE: Dict[str, Any] | None = None _SETTINGS_CACHE_MTIME: float = -1.0 _SETTINGS_CACHE_PATH: str = "" @@ -98,6 +100,11 @@ def get_hub_settings() -> Dict[str, Any]: action = "notify" merged["quota_threshold_action"] = action + email_mode = str(merged.get("email_masking_mode", "none")).strip().lower() + if email_mode not in ("none", "partial", "full"): + email_mode = "none" + merged["email_masking_mode"] = email_mode + _SETTINGS_CACHE = dict(merged) _SETTINGS_CACHE_MTIME = current_mtime _SETTINGS_CACHE_PATH = sfile_str diff --git a/src/antigravity_provider/router/telemetry_service.py b/src/antigravity_provider/router/telemetry_service.py index 838c720..5562078 100644 --- a/src/antigravity_provider/router/telemetry_service.py +++ b/src/antigravity_provider/router/telemetry_service.py @@ -31,6 +31,15 @@ MAX_FILE_BYTES = 5 * 1024 * 1024 # 5 MB MAX_BACKUP_FILES = 3 +def format_token_count(measured: Optional[int], estimated: Optional[int]) -> Optional[str]: + """Format token count distinguishing measured exact counts from estimations (~).""" + if measured is not None: + return str(measured) + elif estimated is not None: + return f"~{estimated}" + return None + + @dataclass class TelemetryRecord: """Immutable record of an individual router invocation attempt.""" @@ -45,6 +54,13 @@ class TelemetryRecord: prompt_tokens: Optional[int] = None completion_tokens: Optional[int] = None total_tokens: Optional[int] = None + prompt_tokens_measured: Optional[int] = None + prompt_tokens_estimated: Optional[int] = None + completion_tokens_measured: Optional[int] = None + completion_tokens_estimated: Optional[int] = None + total_tokens_measured: Optional[int] = None + total_tokens_estimated: Optional[int] = None + is_estimated: bool = False cost_usd: Optional[float] = None failover_count: int = 0 error_category: Optional[str] = None @@ -69,6 +85,14 @@ class TelemetryAggregates: total_prompt_tokens: Optional[int] = None # Sum of reported prompt tokens or None if no token data total_completion_tokens: Optional[int] = None total_tokens: Optional[int] = None + total_prompt_tokens_measured: Optional[int] = None + total_prompt_tokens_estimated: Optional[int] = None + total_completion_tokens_measured: Optional[int] = None + total_completion_tokens_estimated: Optional[int] = None + total_tokens_measured: Optional[int] = None + total_tokens_estimated: Optional[int] = None + tokens_display: Optional[str] = None + has_estimated_tokens: bool = False total_cost_usd: Optional[float] = None # Sum of calculated costs or None if no pricing available failovers_count: int = 0 failover_reasons: Dict[str, int] = field(default_factory=dict) @@ -79,6 +103,7 @@ class TelemetryAggregates: return asdict(self) + class TelemetryService: """Thread-safe persistent telemetry manager with bounded storage and honest aggregation.""" @@ -176,16 +201,42 @@ class TelemetryService: total_tokens: Optional[int] = None, failover_count: int = 0, error_category: Optional[str] = None, + prompt_tokens_measured: Optional[int] = None, + prompt_tokens_estimated: Optional[int] = None, + completion_tokens_measured: Optional[int] = None, + completion_tokens_estimated: Optional[int] = None, + total_tokens_measured: Optional[int] = None, + total_tokens_estimated: Optional[int] = None, + is_estimated: bool = False, ) -> TelemetryRecord: """Record an invocation attempt into memory and rotated log.""" now = time.time() iso = datetime.datetime.fromtimestamp(now, datetime.timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ") - # Derive total tokens if prompt/completion available - if total_tokens is None and prompt_tokens is not None and completion_tokens is not None: - total_tokens = prompt_tokens + completion_tokens + if is_estimated: + p_est = prompt_tokens_estimated if prompt_tokens_estimated is not None else prompt_tokens + c_est = completion_tokens_estimated if completion_tokens_estimated is not None else completion_tokens + t_est = total_tokens_estimated if total_tokens_estimated is not None else total_tokens + if t_est is None and p_est is not None and c_est is not None: + t_est = p_est + c_est + p_meas, c_meas, t_meas = None, None, None + p_tok, c_tok, t_tok = p_est, c_est, t_est + else: + p_meas = prompt_tokens_measured if prompt_tokens_measured is not None else prompt_tokens + c_meas = completion_tokens_measured if completion_tokens_measured is not None else completion_tokens + t_meas = total_tokens_measured if total_tokens_measured is not None else total_tokens + if t_meas is None and p_meas is not None and c_meas is not None: + t_meas = p_meas + c_meas + p_est = prompt_tokens_estimated + c_est = completion_tokens_estimated + t_est = total_tokens_estimated + if t_est is None and p_est is not None and c_est is not None: + t_est = p_est + c_est + p_tok = p_meas if p_meas is not None else p_est + c_tok = c_meas if c_meas is not None else c_est + t_tok = t_meas if t_meas is not None else t_est - cost_usd = self.compute_cost(model, prompt_tokens, completion_tokens) + cost_usd = self.compute_cost(model, p_tok, c_tok) record = TelemetryRecord( timestamp=now, @@ -196,9 +247,16 @@ class TelemetryService: model=model, outcome=outcome, latency_seconds=round(max(0.0, float(latency_seconds)), 4), - prompt_tokens=prompt_tokens, - completion_tokens=completion_tokens, - total_tokens=total_tokens, + prompt_tokens=p_tok, + completion_tokens=c_tok, + total_tokens=t_tok, + prompt_tokens_measured=p_meas, + prompt_tokens_estimated=p_est, + completion_tokens_measured=c_meas, + completion_tokens_estimated=c_est, + total_tokens_measured=t_meas, + total_tokens_estimated=t_est, + is_estimated=is_estimated or (p_est is not None and p_meas is None), cost_usd=cost_usd, failover_count=failover_count, error_category=error_category, @@ -268,6 +326,13 @@ class TelemetryService: prompt_tokens=d.get("prompt_tokens"), completion_tokens=d.get("completion_tokens"), total_tokens=d.get("total_tokens"), + prompt_tokens_measured=d.get("prompt_tokens_measured"), + prompt_tokens_estimated=d.get("prompt_tokens_estimated"), + completion_tokens_measured=d.get("completion_tokens_measured"), + completion_tokens_estimated=d.get("completion_tokens_estimated"), + total_tokens_measured=d.get("total_tokens_measured"), + total_tokens_estimated=d.get("total_tokens_estimated"), + is_estimated=bool(d.get("is_estimated", False)), cost_usd=d.get("cost_usd"), failover_count=int(d.get("failover_count", 0)), error_category=d.get("error_category"), @@ -318,6 +383,14 @@ class TelemetryService: total_prompt_tokens=None, total_completion_tokens=None, total_tokens=None, + total_prompt_tokens_measured=None, + total_prompt_tokens_estimated=None, + total_completion_tokens_measured=None, + total_completion_tokens_estimated=None, + total_tokens_measured=None, + total_tokens_estimated=None, + tokens_display=None, + has_estimated_tokens=False, total_cost_usd=None, failovers_count=0, failover_reasons={}, @@ -333,6 +406,15 @@ class TelemetryService: prompt_tokens_sum = 0 completion_tokens_sum = 0 has_any_token_data = False + prompt_meas_sum = 0 + prompt_est_sum = 0 + has_meas_prompt = False + has_est_prompt = False + comp_meas_sum = 0 + comp_est_sum = 0 + has_meas_comp = False + has_est_comp = False + has_estimated_tokens = False costs_sum = 0.0 has_any_cost_data = False failovers_count = 0 @@ -358,6 +440,41 @@ class TelemetryService: completion_tokens_sum += r.completion_tokens has_any_token_data = True + if r.prompt_tokens_measured is not None: + prompt_meas_sum += r.prompt_tokens_measured + has_meas_prompt = True + elif r.prompt_tokens is not None and not r.is_estimated: + prompt_meas_sum += r.prompt_tokens + has_meas_prompt = True + + if r.prompt_tokens_estimated is not None: + prompt_est_sum += r.prompt_tokens_estimated + has_est_prompt = True + has_estimated_tokens = True + elif r.prompt_tokens is not None and r.is_estimated: + prompt_est_sum += r.prompt_tokens + has_est_prompt = True + has_estimated_tokens = True + + if r.completion_tokens_measured is not None: + comp_meas_sum += r.completion_tokens_measured + has_meas_comp = True + elif r.completion_tokens is not None and not r.is_estimated: + comp_meas_sum += r.completion_tokens + has_meas_comp = True + + if r.completion_tokens_estimated is not None: + comp_est_sum += r.completion_tokens_estimated + has_est_comp = True + has_estimated_tokens = True + elif r.completion_tokens is not None and r.is_estimated: + comp_est_sum += r.completion_tokens + has_est_comp = True + has_estimated_tokens = True + + if r.is_estimated: + has_estimated_tokens = True + if r.cost_usd is not None: costs_sum += r.cost_usd has_any_cost_data = True @@ -369,6 +486,9 @@ class TelemetryService: error_rate = round(failed_calls / total_calls, 4) if total_calls > 0 else 0.0 total_tokens_sum = (prompt_tokens_sum + completion_tokens_sum) if has_any_token_data else None + tot_meas_sum = (prompt_meas_sum + comp_meas_sum) if (has_meas_prompt or has_meas_comp) else None + tot_est_sum = (prompt_est_sum + comp_est_sum) if (has_est_prompt or has_est_comp) else None + tokens_display = format_token_count(tot_meas_sum, tot_est_sum) return TelemetryAggregates( window_seconds=window_seconds, @@ -383,6 +503,14 @@ class TelemetryService: total_prompt_tokens=prompt_tokens_sum if has_any_token_data else None, total_completion_tokens=completion_tokens_sum if has_any_token_data else None, total_tokens=total_tokens_sum, + total_prompt_tokens_measured=prompt_meas_sum if has_meas_prompt else None, + total_prompt_tokens_estimated=prompt_est_sum if has_est_prompt else None, + total_completion_tokens_measured=comp_meas_sum if has_meas_comp else None, + total_completion_tokens_estimated=comp_est_sum if has_est_comp else None, + total_tokens_measured=tot_meas_sum, + total_tokens_estimated=tot_est_sum, + tokens_display=tokens_display, + has_estimated_tokens=has_estimated_tokens, total_cost_usd=round(costs_sum, 4) if has_any_cost_data else None, failovers_count=failovers_count, failover_reasons=dict(failover_reasons), @@ -390,6 +518,7 @@ class TelemetryService: has_data=True, ) + def get_breakdown( self, window_seconds: Optional[int] = 86400, diff --git a/src/antigravity_provider/router/web/server.py b/src/antigravity_provider/router/web/server.py index 5d7e7a4..69569f6 100644 --- a/src/antigravity_provider/router/web/server.py +++ b/src/antigravity_provider/router/web/server.py @@ -125,8 +125,17 @@ def health_check(): } } -def sanitize_snapshot(snap_dict: Any) -> Any: +def sanitize_snapshot(snap_dict: Any, email_masking_mode: Optional[str] = None) -> Any: import re + if email_masking_mode is None: + try: + from antigravity_provider.router.settings_service import get_hub_settings + email_masking_mode = get_hub_settings().get("email_masking_mode", "none") + except Exception: + email_masking_mode = "none" + + mode = str(email_masking_mode or "none").strip().lower() + secret_patterns = [ re.compile(r'((?:access_token|refresh_token|api_key|token|password|secret|key)=)([^\s&,"]+)', re.IGNORECASE), re.compile(r'(sk-[a-zA-Z0-9_\-]{8,})'), @@ -134,6 +143,23 @@ def sanitize_snapshot(snap_dict: Any) -> Any: re.compile(r'(Bearer\s+)([a-zA-Z0-9_\-\.]{8,})', re.IGNORECASE), ] + email_pattern = re.compile(r'\b([A-Za-z0-9._%+-]+)@([A-Za-z0-9.-]+\.[A-Za-z]{2,})\b') + + def _mask_email_match(match: re.Match) -> str: + local_part = match.group(1) + domain_part = match.group(2) + if mode == "full": + return "***@***.***" + elif mode == "partial": + if len(local_part) > 2: + masked = f"{local_part[0]}***{local_part[-1]}" + elif local_part: + masked = f"{local_part[0]}***" + else: + masked = "***" + return f"{masked}@{domain_part}" + return match.group(0) + def _mask_str(val: str) -> str: res = val for pat in secret_patterns: @@ -141,6 +167,8 @@ def sanitize_snapshot(snap_dict: Any) -> Any: res = pat.sub(r'\g<1>***', res) elif pat.groups == 1: res = pat.sub(r'***', res) + if mode in ("partial", "full"): + res = email_pattern.sub(_mask_email_match, res) return res def _sanitize(node): @@ -156,6 +184,7 @@ def sanitize_snapshot(snap_dict: Any) -> Any: return node return _sanitize(snap_dict) + @app.get("/api/snapshot") def get_snapshot(authorized: bool = Depends(get_auth_token)): snapshot = HubStateStore.get().get_snapshot() diff --git a/src/antigravity_provider/router/web/static/app.js b/src/antigravity_provider/router/web/static/app.js index c699e18..1c4958b 100644 --- a/src/antigravity_provider/router/web/static/app.js +++ b/src/antigravity_provider/router/web/static/app.js @@ -182,8 +182,15 @@ function initEventListeners() { if (btnApplyUpdate) { btnApplyUpdate.addEventListener('click', () => applyUpdate()); } + + // Preflight check listener + const btnPreflight = document.getElementById('btn-run-preflight'); + if (btnPreflight) { + btnPreflight.addEventListener('click', () => runPreflightChecks()); + } } + // ── SNAPSHOT INGESTION & MONOTONIC SEQ ── async function fetchSnapshot() { const urlParams = new URLSearchParams(window.location.search); @@ -1313,6 +1320,7 @@ function renderSettingsView() { const quotaThresholdSel = document.getElementById('setting-quota-threshold-percent'); const quotaActionSel = document.getElementById('setting-quota-threshold-action'); + const emailMaskingSel = document.getElementById('setting-email-masking-mode'); const monitorIntervalInput = document.getElementById('setting-monitoring-interval'); if (quotaThresholdSel && s.quota_threshold_percent !== undefined) { @@ -1321,6 +1329,9 @@ function renderSettingsView() { if (quotaActionSel && s.quota_threshold_action) { quotaActionSel.value = s.quota_threshold_action; } + if (emailMaskingSel && s.email_masking_mode) { + emailMaskingSel.value = s.email_masking_mode; + } if (monitorIntervalInput && s.monitoring_interval_seconds !== undefined) { monitorIntervalInput.value = s.monitoring_interval_seconds; } @@ -1329,11 +1340,13 @@ function renderSettingsView() { async function saveHubServerSettings() { const quotaThresholdSel = document.getElementById('setting-quota-threshold-percent'); const quotaActionSel = document.getElementById('setting-quota-threshold-action'); + const emailMaskingSel = document.getElementById('setting-email-masking-mode'); const monitorIntervalInput = document.getElementById('setting-monitoring-interval'); const newSettings = { quota_threshold_percent: quotaThresholdSel ? parseFloat(quotaThresholdSel.value) || 10.0 : 10.0, quota_threshold_action: quotaActionSel ? quotaActionSel.value : 'notify', + email_masking_mode: emailMaskingSel ? emailMaskingSel.value : 'none', monitoring_interval_seconds: monitorIntervalInput ? parseInt(monitorIntervalInput.value, 10) || 30 : 30, }; @@ -1347,6 +1360,73 @@ async function saveHubServerSettings() { } } +// ── PREFLIGHT READINESS CHECKS ── +async function runPreflightChecks() { + const container = document.getElementById('preflight-results-container'); + const btn = document.getElementById('btn-run-preflight'); + if (btn) btn.disabled = true; + if (container) { + container.innerHTML = '
⏳ Запуск zero-quota проверки зависимостей и окружения...
'; + } + try { + const res = await executeAction('run_preflight', {}); + if (!res) throw new Error('Сервер не вернул ответ'); + const report = res.data || {}; + renderPreflightReport(report, container); + showToast(res.message || 'Проверка готовности завершена', res.ok ? 'success' : 'warning'); + } catch (err) { + if (container) { + container.innerHTML = `
❌ Ошибка выполнения проверки: ${escapeHtml(err.message || String(err))}
`; + } + showToast('Ошибка при запуске проверки готовности', 'error'); + } finally { + if (btn) btn.disabled = false; + } +} + +function renderPreflightReport(report, container) { + if (!container) return; + const checks = report.checks || []; + const passed = report.passed_count || 0; + const failed = report.failed_count || 0; + const warn = report.warn_count || 0; + + const statusBadge = `${failed === 0 ? 'Все проверки пройдены' : `Обнаружено ошибок: ${failed}`}`; + + let html = ` +
+
Результат: ${statusBadge}
+
+ Пройдено: ${passed} • + Ошибок: ${failed} • + Предупреждений: ${warn} +
+
+
+ `; + + checks.forEach((item) => { + const badgeClass = item.status === 'PASS' ? 'healthy' : (item.status === 'WARN' ? 'warning' : 'error'); + const icon = item.status === 'PASS' ? '✓' : (item.status === 'WARN' ? '⚠' : '✕'); + html += ` +
+
+
+ ${icon} ${escapeHtml(item.status)} + ${escapeHtml(item.name || item.check_id)} +
+
+
${escapeHtml(item.message || '')}
+ ${item.remediation ? `
💡 Рекомендация: ${escapeHtml(item.remediation)}
` : ''} +
+ `; + }); + + html += '
'; + container.innerHTML = html; +} + + function initSettings() { const btnSave = document.getElementById('btn-save-client-settings'); const tokenInput = document.getElementById('setting-client-token-input'); diff --git a/src/antigravity_provider/router/web/static/index.html b/src/antigravity_provider/router/web/static/index.html index a935ec2..92f0fec 100644 --- a/src/antigravity_provider/router/web/static/index.html +++ b/src/antigravity_provider/router/web/static/index.html @@ -277,9 +277,23 @@ +
+
+
+
Проверка готовности (Preflight / Dependency Agent)
+
Zero-quota аудит окружения, зависимостей, прав доступа и локальных серверов
+
+ +
+
+ +
+
+
Ресурсы хост-системы
+
Мониторинг CPU, памяти, диска и сетевых параметров
@@ -402,6 +416,19 @@
+
+
+
Маскирование почты (PII)
+
Уровень маскирования адресов электронной почты в снапшотах и логах
+
+
+ +
+
Тема оформления
diff --git a/src/antigravity_provider/router/workflow_service.py b/src/antigravity_provider/router/workflow_service.py index b0db3b8..25deca1 100644 --- a/src/antigravity_provider/router/workflow_service.py +++ b/src/antigravity_provider/router/workflow_service.py @@ -35,6 +35,47 @@ def _slug(value: str) -> str: return result or f"agent-{uuid.uuid4().hex[:8]}" +def sanitize_run_data(node: Any) -> Any: + """Recursively strip or mask any credentials or secret keys from workflow run state.""" + secret_key_substrings = [ + "api_key", "token", "password", "secret", "jwt", "bearer", + "access_token", "refresh_token", "client_secret", "authorization", + ] + if isinstance(node, dict): + sanitized: dict[str, Any] = {} + for k, v in node.items(): + k_lower = str(k).lower() + if any(s in k_lower for s in secret_key_substrings) and k_lower not in ("auth_status", "author", "auth_required"): + sanitized[k] = "***" + else: + sanitized[k] = sanitize_run_data(v) + return sanitized + elif isinstance(node, list): + return [sanitize_run_data(x) for x in node] + elif isinstance(node, str): + val = node + val = re.sub(r'Bearer\s+[a-zA-Z0-9_\-\.]{8,}', 'Bearer ***', val, flags=re.IGNORECASE) + val = re.sub(r'sk-[a-zA-Z0-9_\-]{8,}', 'sk-***', val) + val = re.sub(r'gho_[a-zA-Z0-9_\-]{8,}', 'gho_***', val) + val = re.sub(r'((?:access_token|refresh_token|api_key|token|password|secret|key)=)([^\s&,"]+)', r'\g<1>***', val, flags=re.IGNORECASE) + return val + return node + + +def get_last_run_state(run_state_path: Optional[Path] = None) -> Optional[dict[str, Any]]: + """Return the last saved workflow run state from workflow_run_state.json, if any.""" + p = run_state_path or paths.get_workflow_run_state_path() + if not p.is_file(): + return None + try: + data = json.loads(p.read_text(encoding="utf-8")) + if isinstance(data, dict): + return sanitize_run_data(data) + except Exception: + pass + return None + + def _safe_agent_file(value: str, agent_id: str) -> tuple[Path, str]: """Resolve an Agent File below HERMES_HOME/agents and reject traversal.""" root = paths.get_agent_files_dir().resolve() @@ -103,8 +144,9 @@ class WorkflowService: _instance: Optional["WorkflowService"] = None _instance_lock = threading.Lock() - def __init__(self, state_path: Optional[Path] = None) -> None: + def __init__(self, state_path: Optional[Path] = None, run_state_path: Optional[Path] = None) -> None: self.state_path = state_path or paths.get_workflow_state_path() + self.run_state_path = run_state_path or paths.get_workflow_run_state_path() self._lock = threading.RLock() self._stop = threading.Event() self._thread: Optional[threading.Thread] = None @@ -112,6 +154,7 @@ class WorkflowService: self.workflow = WorkflowDefinition() self.events: list[WorkflowEvent] = [] self.run: dict[str, Any] = self._idle_run() + self._completed_steps: list[dict[str, Any]] = [] self._load() self._migrate_router_roles() @@ -140,6 +183,25 @@ class WorkflowService: } def _load(self) -> None: + # 1. Load run state from workflow_run_state.json + if self.run_state_path.is_file(): + try: + run_state = json.loads(self.run_state_path.read_text(encoding="utf-8")) + if isinstance(run_state, dict): + self._completed_steps = list(run_state.get("completed_steps", [])) + if run_state.get("status") in {"RUNNING", "running", "STOPPING", "stopping"}: + run_state["status"] = "INTERRUPTED" + run_state["interruption_reason"] = "Прогон был прерван перезапуском сервера или сбоем процесса" + run_state["updated_at"] = _utc_timestamp() + sanitized = sanitize_run_data(run_state) + self.run_state_path.parent.mkdir(parents=True, exist_ok=True) + temp = self.run_state_path.with_suffix(".tmp") + temp.write_text(json.dumps(sanitized, ensure_ascii=False, indent=2), encoding="utf-8") + temp.replace(self.run_state_path) + except Exception: + self._completed_steps = [] + + # 2. Load workflow definition and events from workflow_state.json if not self.state_path.is_file(): return try: @@ -164,6 +226,42 @@ class WorkflowService: self.events = [] self.run = self._idle_run() + def _save_run_state( + self, + status: str, + step_index: int = 0, + current_agent: Optional[str] = None, + iteration: int = 1, + completed_step: Optional[dict[str, Any]] = None, + interruption_reason: Optional[str] = None, + ) -> None: + if completed_step: + self._completed_steps.append(sanitize_run_data(completed_step)) + + state_payload = { + "run_id": self.run.get("id"), + "status": status.upper(), + "started_at": self.run.get("started_at"), + "updated_at": _utc_timestamp(), + "current_step_index": step_index, + "current_agent_id": current_agent, + "iteration_count": iteration, + "completed_steps": list(self._completed_steps), + "interruption_reason": interruption_reason, + } + sanitized = sanitize_run_data(state_payload) + try: + self.run_state_path.parent.mkdir(parents=True, exist_ok=True) + temp = self.run_state_path.with_suffix(".tmp") + temp.write_text(json.dumps(sanitized, ensure_ascii=False, indent=2), encoding="utf-8") + temp.replace(self.run_state_path) + except Exception: + pass + + def get_last_run_state(self) -> Optional[dict[str, Any]]: + """Return the last saved run state from workflow_run_state.json.""" + return get_last_run_state(self.run_state_path) + def _save(self) -> None: self.state_path.parent.mkdir(parents=True, exist_ok=True) payload = { @@ -425,6 +523,7 @@ class WorkflowService: if not start_id or start_id not in self.agents: raise ValueError("В workflow нет стартового агента") self._stop.clear() + self._completed_steps = [] self.run = self._idle_run() self.run.update({ "id": uuid.uuid4().hex, @@ -436,6 +535,7 @@ class WorkflowService: }) self._event("WORKFLOW_STARTED", "Workflow запущен", run_id=self.run["id"], iteration=1) self._save() + self._save_run_state("RUNNING", step_index=0, current_agent=start_id, iteration=1) self._thread = threading.Thread(target=self._execute, name="HermesWorkflow", daemon=True) self._thread.start() return dict(self.run) @@ -448,6 +548,7 @@ class WorkflowService: self._stop.set() self._event("WORKFLOW_STOP_REQUESTED", "Запрошена остановка workflow", level="warning") self._save() + self._save_run_state("STOPPED", step_index=len(self._completed_steps), current_agent=self.run.get("current_agent_id"), iteration=self.run.get("iteration", 1), interruption_reason="Остановлено пользователем") return dict(self.run) def _execute(self) -> None: @@ -466,10 +567,12 @@ class WorkflowService: visited[current] = visited.get(current, 0) + 1 iteration = max(visited.values()) self.run.update({"current_agent_id": current, "iteration": iteration}) + step_idx = len(self._completed_steps) if iteration > self.workflow.max_iterations: message = f"Достигнут предел итераций: {self.workflow.max_iterations}" self.run.update({"status": "failed", "error": message}) self._event("WORKFLOW_MAX_ITERATIONS", message, level="error", agent_id=current, iteration=iteration) + self._save_run_state("FAILED", step_index=step_idx, current_agent=current, iteration=iteration, interruption_reason=message) break file_data = self.read_agent_file(current) if not file_data["exists"]: @@ -479,6 +582,7 @@ class WorkflowService: ) self._event("AGENT_STARTED", f"{agent.name} начал выполнение", agent_id=current, iteration=iteration) self._save() + self._save_run_state("RUNNING", step_index=step_idx, current_agent=current, iteration=iteration) request = { "model": self._execution_config(agent).get("model"), "messages": [ @@ -515,6 +619,19 @@ class WorkflowService: duration_seconds=duration, error=text if status == "ERROR" else None, ) + step_summary = { + "step_index": step_idx, + "agent_id": current, + "agent_name": agent.name, + "iteration": iteration, + "status": status, + "duration_seconds": duration, + "provider": metadata.get("provider"), + "account": metadata.get("profile_id"), + "model": metadata.get("selected_model") or (metadata.get("selection_trace") or {}).get("selected_model"), + "error": text if status in {"ERROR", "REVIEW_FAILED"} else None, + "timestamp": _utc_timestamp(), + } edge = next( (item for item in self.workflow.edges if item.source == current and item.condition in {status, "ALWAYS"}), None, @@ -548,6 +665,14 @@ class WorkflowService: ) except Exception: pass + self._save_run_state( + "FAILED" if self.run["status"] == "failed" else "COMPLETED", + step_index=step_idx + 1, + current_agent=current, + iteration=iteration, + completed_step=step_summary, + interruption_reason=self.run.get("error") if self.run["status"] == "failed" else None, + ) break self._event( "WORKFLOW_TRANSITION", @@ -562,10 +687,18 @@ class WorkflowService: }, ensure_ascii=False) current = edge.target self._save() + self._save_run_state( + "RUNNING", + step_index=step_idx + 1, + current_agent=current, + iteration=iteration, + completed_step=step_summary, + ) with self._lock: if self._stop.is_set(): self.run.update({"status": "stopped", "error": "Остановлено пользователем"}) self._event("WORKFLOW_STOPPED", "Workflow остановлен пользователем", level="warning") + self._save_run_state("STOPPED", step_index=len(self._completed_steps), current_agent=current, iteration=iteration, interruption_reason="Остановлено пользователем") except Exception as exc: with self._lock: self.run.update({"status": "failed", "error": str(exc)}) @@ -576,6 +709,7 @@ class WorkflowService: EventLogService.get().log("workflow", "Ошибка выполнения workflow", details=str(exc), level="error") except Exception: pass + self._save_run_state("FAILED", step_index=len(self._completed_steps), current_agent=current, iteration=visited.get(current, 1), interruption_reason=str(exc)) finally: with self._lock: self.run["finished_at"] = _utc_timestamp() @@ -583,6 +717,7 @@ class WorkflowService: self.run["current_agent_id"] = current or self.run.get("current_agent_id") self._save() + @staticmethod def _response_text(response: Any) -> str: if not isinstance(response, dict): @@ -644,3 +779,7 @@ def execute_workflow_action(action: str, data: dict[str, Any]) -> dict[str, Any] except Exception: pass return {"ok": True, "message": "Выполнено", "data": result} + + +# Alias for execution service compatibility +WorkflowExecutionService = WorkflowService diff --git a/tests/test_a31_preflight_state_batching_pii.py b/tests/test_a31_preflight_state_batching_pii.py new file mode 100644 index 0000000..cbe35a0 --- /dev/null +++ b/tests/test_a31_preflight_state_batching_pii.py @@ -0,0 +1,397 @@ +"""Tests for Task A31: Preflight Dependency Agent, Workflow Run State, Local Concurrency & Context Window, PII Masking, and Cost Controller Honesty. +""" +from __future__ import annotations + +import json +from pathlib import Path +from unittest.mock import MagicMock, patch + +import pytest + +from antigravity_provider import paths +from antigravity_provider.router.action_handler import ActionExecutor +from antigravity_provider.router.adapters.local_adapter import LocalLLMAdapter +from antigravity_provider.router.preflight_service import PreflightCheckService, PreflightReport +from antigravity_provider.router.role_registry import CANONICAL_ROLES, RoleRegistry +from antigravity_provider.router.router_config import ( + RouterConfig, + RouterProfileConfig, + get_default_router_config, + load_router_config, +) +from antigravity_provider.router.settings_service import ( + DEFAULT_SETTINGS, + get_hub_settings, + invalidate_settings_cache, + save_hub_settings, +) +from antigravity_provider.router.telemetry_service import ( + TelemetryAggregates, + TelemetryRecord, + TelemetryService, + format_token_count, +) +from antigravity_provider.router.web.server import sanitize_snapshot +from antigravity_provider.router.workflow_service import ( + AgentDefinition, + WorkflowDefinition, + WorkflowExecutionService, + WorkflowService, + get_last_run_state, + sanitize_run_data, +) + + +# ============================================================================ +# P0-1: Preflight Dependency Agent +# ============================================================================ + + +def test_dependency_agent_role_registered(): + """Verify 13th role 'dependency-agent' and its canonical aliases in RoleRegistry.""" + assert len(CANONICAL_ROLES) == 13 + assert "dependency-agent" in CANONICAL_ROLES + + role_def = CANONICAL_ROLES["dependency-agent"] + assert role_def.role_id == "dependency-agent" + assert role_def.display_name_ru == "Проверяющий готовность" + assert role_def.short_name_ru == "Готовность" + assert role_def.is_implemented is True + assert "preflight" in role_def.capabilities + + # Test alias resolution + aliases = [ + "dependency-agent", + "dependency_agent", + "preflight", + "проверяющий готовность", + "агент зависимостей", + "готовность", + "dependency", + ] + for alias in aliases: + canonical = RoleRegistry.resolve_role_name(alias) + assert canonical == "dependency-agent", f"Alias '{alias}' resolved to '{canonical}'" + + +def test_preflight_service_cli_and_environment(): + """Verify CLI tools and environment checks.""" + service = PreflightCheckService.get() + + cli_items = service.check_cli_dependencies() + assert len(cli_items) >= 3 + ids = {item.check_id for item in cli_items} + assert "cli_agy" in ids + assert "pkg_fastapi" in ids + assert "pkg_uvicorn" in ids + + env_items = service.check_system_environment() + assert len(env_items) >= 3 + env_ids = {item.check_id for item in env_items} + assert "env_hermes_home" in env_ids + assert "env_config_writable" in env_ids + assert "env_logs_writable" in env_ids + + +def test_preflight_service_run_all_and_action(): + """Verify run_all_checks and action execution.""" + service = PreflightCheckService.get() + report = service.run_all_checks() + + assert isinstance(report, PreflightReport) + assert isinstance(report.passed_count, int) + assert isinstance(report.failed_count, int) + assert isinstance(report.warn_count, int) + assert len(report.checks) > 0 + + report_dict = report.to_dict() + assert "success" in report_dict + assert "checks" in report_dict + assert isinstance(report_dict["checks"], list) + + # Test via ActionExecutor + action_res = ActionExecutor.execute("run_preflight", {}) + assert "ok" in action_res + assert "message" in action_res + assert "data" in action_res + assert "checks" in action_res["data"] + + +# ============================================================================ +# P0-2: Workflow Run State Manager +# ============================================================================ + + +def test_workflow_run_state_sanitization(): + """Verify recursive secret stripping in workflow run state.""" + raw_state = { + "run_id": "test-run-123", + "status": "RUNNING", + "api_key": "sk-1234567890abcdef", + "token": "gho_secrettoken123456", + "nested": { + "password": "supersecretpass", + "auth_status": "ok", + "message": "Bearer eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9.eyJzdWIiOiIxMjM0NTY3ODkwIn0", + "extra_url": "https://example.com/callback?access_token=secret12345&foo=bar", + }, + "step_list": [ + {"account": "acc-1", "client_secret": "my-client-secret-123"}, + {"safe_field": "public_data"}, + ], + } + + sanitized = sanitize_run_data(raw_state) + + assert sanitized["api_key"] == "***" + assert sanitized["token"] == "***" + assert sanitized["nested"]["password"] == "***" + assert sanitized["nested"]["auth_status"] == "ok" + assert "Bearer ***" in sanitized["nested"]["message"] + assert "access_token=***" in sanitized["nested"]["extra_url"] + assert sanitized["step_list"][0]["client_secret"] == "***" + assert sanitized["step_list"][1]["safe_field"] == "public_data" + + +def test_workflow_run_state_interrupted_on_startup(tmp_path: Path): + """Verify that a RUNNING state in workflow_run_state.json transitions to INTERRUPTED on reload.""" + state_file = tmp_path / "workflow_state.json" + run_state_file = tmp_path / "workflow_run_state.json" + + # Pre-populate run state with RUNNING status + initial_run_state = { + "run_id": "run-crash-test", + "status": "RUNNING", + "started_at": "2026-08-26T00:00:00Z", + "updated_at": "2026-08-26T00:00:00Z", + "current_step_index": 2, + "current_agent_id": "developer-1", + "iteration_count": 1, + "completed_steps": [ + {"step_index": 0, "agent_id": "manager", "status": "SUCCESS"}, + {"step_index": 1, "agent_id": "developer-1", "status": "WORKING"}, + ], + "interruption_reason": None, + } + run_state_file.write_text(json.dumps(initial_run_state), encoding="utf-8") + + # Initialize WorkflowService + service = WorkflowService(state_path=state_file, run_state_path=run_state_file) + + # Check that state transitioned to INTERRUPTED + last_state = service.get_last_run_state() + assert last_state is not None + assert last_state["status"] == "INTERRUPTED" + assert last_state["interruption_reason"] == "Прогон был прерван перезапуском сервера или сбоем процесса" + assert len(last_state["completed_steps"]) == 2 + + # Verify top-level function + assert get_last_run_state(run_state_file)["status"] == "INTERRUPTED" + + +def test_workflow_execution_service_alias(): + """Verify WorkflowExecutionService is an alias of WorkflowService.""" + assert WorkflowExecutionService is WorkflowService + + +# ============================================================================ +# P0-3: Local Concurrency & Context Window +# ============================================================================ + + +def test_local_profile_max_concurrency_is_one(): + """Verify all local provider profiles have max_concurrency = 1.""" + config = get_default_router_config() + for pid, pcfg in config.profiles.items(): + if pcfg.provider == "local": + assert pcfg.max_concurrency == 1, f"Local profile {pid} has max_concurrency={pcfg.max_concurrency}" + + # Verify loaded config also enforces max_concurrency = 1 for local profiles + loaded = load_router_config() + for pid, pcfg in loaded.profiles.items(): + if pcfg.provider == "local": + assert pcfg.max_concurrency == 1 + + +def test_local_adapter_get_context_window(): + """Verify LocalLLMAdapter retrieves context window accurately without hallucinating defaults.""" + adapter = LocalLLMAdapter() + + # Profile with explicit context_window in auth_config + prof_with_cfg = RouterProfileConfig( + profile_id="local-test-1", + provider="local", + account_id="acc-1", + auth_config={"context_window": 8192}, + ) + assert adapter.get_context_window(prof_with_cfg) == 8192 + + # Profile without context length and with non-responding server + prof_empty = RouterProfileConfig( + profile_id="local-test-2", + provider="local", + account_id="acc-2", + custom_base_url="http://127.0.0.1:9999/v1", + ) + # Must return None instead of inventing fake numbers + assert adapter.get_context_window(prof_empty) is None + + +def test_local_adapter_context_truncation_guard(): + """Verify context truncation guard protects against VRAM overflow when context_window is known.""" + adapter = LocalLLMAdapter() + + prof = RouterProfileConfig( + profile_id="local-small-ctx", + provider="local", + account_id="acc-1", + auth_config={"context_window": 500}, + custom_base_url="http://127.0.0.1:12345/v1", + ) + + # Huge prompt exceeding 500 tokens + long_middle_content = "important historical dialogue step " * 100 + messages = [ + {"role": "system", "content": "You are a helpful assistant."}, + {"role": "user", "content": "Initial prompt 1"}, + {"role": "assistant", "content": long_middle_content}, + {"role": "user", "content": "Initial prompt 2"}, + {"role": "assistant", "content": long_middle_content}, + {"role": "user", "content": "Latest user task to execute."}, + ] + + mock_resp = { + "choices": [{"message": {"role": "assistant", "content": "Truncated prompt executed successfully."}}], + "usage": {"prompt_tokens": 200, "completion_tokens": 10}, + } + + with patch("urllib.request.urlopen") as mock_urlopen: + mock_cm = MagicMock() + mock_cm.read.return_value = json.dumps(mock_resp).encode("utf-8") + mock_urlopen.return_value.__enter__.return_value = mock_cm + + res = adapter.invoke(prof, {"messages": messages, "max_tokens": 100}) + assert res["choices"][0]["message"]["content"] == "Truncated prompt executed successfully." + + # Verify sent payload messages were truncated + args, kwargs = mock_urlopen.call_args + sent_req = args[0] + sent_body = json.loads(sent_req.data.decode("utf-8")) + sent_messages = sent_body["messages"] + + assert sent_messages[0]["role"] == "system" + assert sent_messages[-1]["content"] == "Latest user task to execute." + # Total count of messages should be pruned + assert len(sent_messages) < len(messages) + + +# ============================================================================ +# P0-4: PII Email Masking +# ============================================================================ + + +def test_settings_email_masking_mode(): + """Verify email_masking_mode in default settings and persistence.""" + assert DEFAULT_SETTINGS["email_masking_mode"] == "none" + + settings = get_hub_settings() + assert settings.get("email_masking_mode") in ("none", "partial", "full") + + +def test_sanitize_snapshot_email_masking_modes(): + """Verify email masking behavior across 'none', 'partial', and 'full' modes.""" + snapshot_data = { + "user_email": "vasya.pupkin@example.com", + "account_id": "google-user-1", + "api_key": "sk-secret123456789", + "nested": { + "developer": "developer.one@domain.org", + "reviewer": "r@test.com", + }, + } + + # 1. Mode: none (emails unchanged, secrets masked) + san_none = sanitize_snapshot(snapshot_data, email_masking_mode="none") + assert san_none["user_email"] == "vasya.pupkin@example.com" + assert san_none["nested"]["developer"] == "developer.one@domain.org" + assert san_none["nested"]["reviewer"] == "r@test.com" + assert "api_key" not in san_none + + # 2. Mode: partial (preserves first and last char of local part + domain for differentiation) + san_partial = sanitize_snapshot(snapshot_data, email_masking_mode="partial") + assert san_partial["user_email"] == "v***n@example.com" + assert san_partial["nested"]["developer"] == "d***e@domain.org" + assert san_partial["nested"]["reviewer"] == "r***@test.com" + assert "api_key" not in san_partial + + # 3. Mode: full (***@***.***) + san_full = sanitize_snapshot(snapshot_data, email_masking_mode="full") + assert san_full["user_email"] == "***@***.***" + assert san_full["nested"]["developer"] == "***@***.***" + assert san_full["nested"]["reviewer"] == "***@***.***" + assert "api_key" not in san_full + + +# ============================================================================ +# P0-5: Cost Controller Token Honesty +# ============================================================================ + + +def test_telemetry_measured_vs_estimated_tokens(tmp_path: Path): + """Verify telemetry distinguishes measured exact tokens from estimated tokens with ~.""" + log_file = tmp_path / "telemetry_test.jsonl" + service = TelemetryService(log_path=log_file) + + # 1. Record measured call + rec1 = service.record_call( + role="developer-1", + profile_id="ag-w1", + provider="antigravity", + model="claude-3-7-sonnet", + outcome="success", + latency_seconds=1.25, + prompt_tokens_measured=500, + completion_tokens_measured=150, + is_estimated=False, + ) + assert rec1.prompt_tokens_measured == 500 + assert rec1.prompt_tokens_estimated is None + assert rec1.is_estimated is False + assert rec1.total_tokens == 650 + + # 2. Record estimated call + rec2 = service.record_call( + role="tester", + profile_id="local-1", + provider="local", + model="Qwen3.8-27B-Q4_K_M.gguf", + outcome="success", + latency_seconds=0.85, + prompt_tokens_estimated=300, + completion_tokens_estimated=50, + is_estimated=True, + ) + assert rec2.prompt_tokens_measured is None + assert rec2.prompt_tokens_estimated == 300 + assert rec2.is_estimated is True + assert rec2.total_tokens == 350 + + # 3. Aggregates for measured only + agg_measured = service.get_aggregates(profile_id="ag-w1") + assert agg_measured.total_tokens_measured == 650 + assert agg_measured.tokens_display == "650" + assert agg_measured.has_estimated_tokens is False + + # 4. Aggregates for estimated only + agg_est = service.get_aggregates(profile_id="local-1") + assert agg_est.total_tokens_estimated == 350 + assert agg_est.tokens_display == "~350" + assert agg_est.has_estimated_tokens is True + + +def test_format_token_count(): + """Verify format_token_count formatting helper.""" + assert format_token_count(1250, None) == "1250" + assert format_token_count(None, 1250) == "~1250" + assert format_token_count(1000, 250) == "1000" + assert format_token_count(None, None) is None