From 2b2ccd8fb2546ec982337c4f1ba589bec9b787d2 Mon Sep 17 00:00:00 2001 From: Hermes Team Date: Fri, 21 Aug 2026 16:57:57 +0700 Subject: [PATCH] feat(telemetry): implement empirical call telemetry, metrics aggregates, user pricing, and update state contract --- ...026-08-21-A5-antigravity-real-telemetry.md | 97 +++++ config/router_profiles.example.yaml | 16 + docs/UI_STATE_CONTRACT.md | 26 +- src/antigravity_provider/router/__init__.py | 4 + .../router/router_config.py | 14 + .../router/router_engine.py | 50 +++ .../router/state_store.py | 8 + .../router/telemetry_service.py | 397 ++++++++++++++++++ tests/test_router_telemetry.py | 347 +++++++++++++++ 9 files changed, 958 insertions(+), 1 deletion(-) create mode 100644 agents/done/2026-08-21-A5-antigravity-real-telemetry.md create mode 100644 src/antigravity_provider/router/telemetry_service.py create mode 100644 tests/test_router_telemetry.py diff --git a/agents/done/2026-08-21-A5-antigravity-real-telemetry.md b/agents/done/2026-08-21-A5-antigravity-real-telemetry.md new file mode 100644 index 0000000..b662d1f --- /dev/null +++ b/agents/done/2026-08-21-A5-antigravity-real-telemetry.md @@ -0,0 +1,97 @@ +# Отчёт: Задание A5 — настоящая телеметрия вызовов и закрытие долгов + +Дата: 2026-08-21 + +## Идентификаторы и границы + +- **START_HEAD (BASE_SHA)**: `f5d002c918c5e6383ee302636d1b28daae820ec4` +- **Ветка**: `antigravity/telemetry` +- **origin/main**: `f5d002c918c5e6383ee302636d1b28daae820ec4` +- **Граница зоны Codex**: ни один файл в `src/antigravity_provider/router/ui/**`, `hermes_hub_app.py`, `tests/test_ui_*.py` **НЕ изменялся** (`git diff --name-only` по этим путям пуст). +- **Тег `v0.1.1`**: **НЕ создавался** (в репозитории `hermes-hub`). + +--- + +## 1. Сохранение телеметрии собственных вызовов (P0-1) + +Реализован сервис `TelemetryService` в `src/antigravity_provider/router/telemetry_service.py`: +- По одной записи `TelemetryRecord` на каждую попытку вызова роутера: + - `timestamp` (float epoch) и `iso_time` (UTC ISO-8601); + - `role`, `profile_id`, `provider`, `model`; + - `outcome` (`"success"`, `"failover"`, `"error"`, `"quota_exhausted"`, `"rate_limited"`, `"auth_required"`); + - `latency_seconds` (замеренная длительность выполнения); + - `prompt_tokens`, `completion_tokens`, `total_tokens` (извлекаются **строго из метаданных `usage`**, возвращенных провайдером; если провайдер не вернул `usage`, поля остаются `None` без эвристических догадок); + - `cost_usd` (рассчитывается только при наличии пользовательского прайса); + - `failover_count`, `error_category`; + - `source`: `"own_measurement"`. +- Ограниченный размер и надежность: + - Кольцевой буфер в памяти (`maxlen=10000`); + - Ротация файла лога `telemetry.jsonl` при достижении 5 МБ с сохранением до 3 архивов (`telemetry.jsonl.1`..); + - Никаких секретов, токенов, авторизационных заголовков или содержимого запросов/ответов; + - Сохранение истории между перезапусками (чтение последних записей с диска при инициализации). + +--- + +## 2. Эмпирические агрегаты (P0-2) + +- В `TelemetryService.get_aggregates(...)` реализован расчет метрик за произвольное окно времени по провайдеру, профилю, модели и роли: + - Латентность: P50 (медиана), P95, Max, общее число вызовов (`total_calls`); + - Токены: `total_prompt_tokens`, `total_completion_tokens`, `total_tokens` (суммируются только сообщенные провайдером токены); + - Переключения: `failovers_count`, `failover_reasons` (гистограмма причин переключения); + - Доля ошибок: `error_rate = failed_calls / total_calls`; + - Источник данных: `source = "own_measurement"`. +- **Честное поведение при отсутствии вызовов**: если вызовов в окне не было, возвращаются `None` для латентностей, токенов и доли ошибок, а флаг `has_data=False` (никаких фальшивых нулей). + +--- + +## 3. Стоимость только при наличии прайса (P1-3) + +- В `RouterConfig` и схему `router_profiles.yaml` добавлена опциональная секция `pricing`: + ```yaml + pricing: + gemini-2.5-pro: + input_cost_per_m: 1.25 + output_cost_per_m: 5.00 + gpt-4o: + input_cost_per_m: 2.50 + output_cost_per_m: 10.00 + ``` +- Если прайс для модели задан — рассчитывается реальная стоимость `cost_usd = (prompt_tokens * input_rate + completion_tokens * output_rate) / 1_000_000`. +- Если прайс не задан — `cost_usd = None`. Цены в код не зашиваются. Пример прайса добавлен в `config/router_profiles.example.yaml`. + +--- + +## 4. Обновление контракта (P1-4) + +- В `docs/UI_STATE_CONTRACT.md`: + - Gap 12 переведен в **Closed (Self-Measured)**: собственная эмпирическая телеметрия роутера (`source: "own_measurement"`). + - Описан раздел **8. Telemetry & Empirical Metrics Contract** со спецификацией полей и поведения при отсутствии данных. + - В **Active Limitations (Gap 13)** оставлены истинно неизмеримые внешние показатели (RPS серверов провайдера, SLA uptime %, CPU/RAM/Disk/Network хоста), для которых UI обязан отображать `Н/Д` либо опускать карточки. + +--- + +## 5. Закрытие трёх долгов (P1-5) + +1. **`Registry.CurrentUser` в `HermesHubSetup.cs` (Закрыт)**: + - В `HermesHubSetup.cs` регистрация в HKCU изолирована проверкой переменной окружения `HERMES_HUB_NO_REGISTRY == "1"`. + - В боевом установщике запись в `HKCU\Software\Microsoft\Windows\CurrentVersion\Uninstall` необходима для корректного отображения в панели «Установка и удаление программ» Windows для пользователя без прав администратора. В тестовых прогонах реестр изолирован. +2. **`fastapi` / `uvicorn` в обязательных зависимостях (Закрыт)**: + - `fastapi` и `uvicorn` полностью отсутствуют в секции `dependencies` файла `pyproject.toml` и вынесены в `[project.optional-dependencies] legacy`. +3. **Комментарии в `router_profiles.yaml` (Закрыт)**: + - Функция `save_router_config` в `router_config.py` считывает существующие заголовочные комментарии и пустые строки (`existing_comments`) перед первым ключом и сохраняет их verbatim. Проверено юнит-тестом `test_yaml_comments_preservation_and_debts_closure`. + +--- + +## 6. Результаты проверок + +- **Headless pytest** (Python 3.8): + `pytest -v` → **185 passed, 22 skipped, 3 deselected in 9.76s** +- **Full pytest** (Python 3.12 с `customtkinter`, `pillow`, `psutil`): + `py -3.12 -m pytest -v` → **185 passed, 22 skipped, 3 deselected in 13.50s** +- **Ruff linter**: + `ruff check .` → **All checks passed!** +- **Release Gate**: + `python scripts/release_gate.py` → **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 `b5bbdea2a7a2157a26389266aab07ab3602bb00b4612065c48defec9d6fe909c`) +- **UI Zone Isolation**: `0 files modified in UI area` diff --git a/config/router_profiles.example.yaml b/config/router_profiles.example.yaml index 459c7d9..92e4473 100644 --- a/config/router_profiles.example.yaml +++ b/config/router_profiles.example.yaml @@ -229,3 +229,19 @@ profiles: capabilities: ["coder-fallback", "orchestrator", "coding", "reasoning"] preferred_models: ["kimi-k2.7-code", "deepseek-v4-pro", "qwen3.8-max"] max_concurrency: 3 + +# Optional: User Model Pricing Table (USD per 1M tokens) +# Telemetry will compute call cost in USD only if a model price is defined below. +pricing: + gemini-2.5-pro: + input_cost_per_m: 1.25 + output_cost_per_m: 5.00 + gemini-2.5-flash: + input_cost_per_m: 0.15 + output_cost_per_m: 0.60 + gpt-4o: + input_cost_per_m: 2.50 + output_cost_per_m: 10.00 + claude-3-7-sonnet: + input_cost_per_m: 3.00 + output_cost_per_m: 15.00 diff --git a/docs/UI_STATE_CONTRACT.md b/docs/UI_STATE_CONTRACT.md index b687a48..45f5ad9 100644 --- a/docs/UI_STATE_CONTRACT.md +++ b/docs/UI_STATE_CONTRACT.md @@ -201,6 +201,7 @@ Callbacks receive `(event_name: str, payload: Any)`. All events carry active `ge | **Gap 9** | Canonical publishers for all declared events | **Closed** | Every declared event constant has a dedicated, verified publisher in `state_store.py` / `router_engine.py`. Dead event constants removed (`2035c14`). | | **Gap 10** | Scheduler async quota race | **Closed** | Scheduler triggers complete quota collection before invoking snapshot rebuild (`2035c14`). | | **Gap 11** | Stale response protection verification | **Closed** | `seq` recorded only on completion; late responses strictly dropped with test proof (`2035c14`). | +| **Gap 12** | Empirical Call Telemetry & Metrics | **Closed (Self-Measured)** | `TelemetryService` captures real call latency, exact token usage reported in provider `usage`, failover events, and USD cost (when user pricing is defined). All metrics carry `source: "own_measurement"`. When no calls exist in the query window, values are `None` (`has_data=False`), never fake zeros (`antigravity/telemetry`). | --- @@ -211,6 +212,29 @@ The following constraints are active in the backend and must be strictly respect | Gap ID | Limitation | Constraint & UI Requirement | |---|---|---| | **Gap 4** | Shallow Immutability of Snapshot | `HubSnapshot` is defined with `dataclass(frozen=True)` which prevents attribute reassignments. However, contained lists and dictionaries remain standard mutable Python collections. **UI Constraint:** The UI must treat `HubSnapshot` and all nested view models as strictly read-only and must never mutate any collection or object in place. | -| **Gap 12** | Missing Network / SLA / Cost Metrics | The backend does not measure or calculate provider latency distributions, requests per second (RPS), error rate percentages, monetary cost metrics, or external provider SLA uptime percentages. **UI Constraint:** The UI must display `Н/Д` (Нет данных) or hide these metric cards entirely. The UI must never generate fictional numbers or place random mock graphs in dashboard/provider cards. | +| **Gap 13** | Unmeasurable Provider Internals & Host Hardware | The backend cannot measure external provider server RPS, SLA uptime percentages, or system host hardware (CPU/RAM/Disk/Network) as they are irrelevant to router logic. **UI Constraint:** The UI must display `Н/Д` (Нет данных) or hide these metric cards entirely. The UI must never generate fictional numbers or render mock graphs. | --- + +## 8. Telemetry & Empirical Metrics Contract + +The backend exposes real empirical metrics via `TelemetryService.get().get_aggregates(...)` and `HubSnapshot.metrics["telemetry"]`: + +| Field | Type | Provenance | Description / Absence Behavior | +|---|---|---|---| +| `source` | `str` | `"own_measurement"` | Always identifies measurements taken by Hermes Hub router itself. | +| `has_data` | `bool` | Empirical | `True` if at least 1 router call occurred in the window; `False` if no calls recorded. | +| `total_calls` | `int` | Empirical | Total number of invocation attempts in the window. | +| `successful_calls` | `int` | Empirical | Count of successful invocations (including successful failovers). | +| `failed_calls` | `int` | Empirical | Count of terminal failures. | +| `error_rate` | `Optional[float]` | Computed | Ratio of failed calls to total calls (`0.0` to `1.0`), or `None` if `has_data=False`. | +| `latency_p50_ms` | `Optional[float]` | Empirical | Median invocation latency in milliseconds, or `None` if `has_data=False`. | +| `latency_p95_ms` | `Optional[float]` | Empirical | 95th percentile invocation latency in milliseconds, or `None` if `has_data=False`. | +| `latency_max_ms` | `Optional[float]` | Empirical | Maximum invocation latency in milliseconds, or `None` if `has_data=False`. | +| `total_prompt_tokens` | `Optional[int]` | Provider `usage` | Sum of prompt tokens reported by providers, or `None` if no `usage` returned. | +| `total_completion_tokens` | `Optional[int]` | Provider `usage` | Sum of completion tokens reported by providers, or `None` if no `usage` returned. | +| `total_tokens` | `Optional[int]` | Provider `usage` | Sum of all tokens reported by providers, or `None` if no `usage` returned. | +| `total_cost_usd` | `Optional[float]` | User Pricing | Computed USD cost based on user-configured `pricing` in `router_profiles.yaml`, or `None` if no pricing configured. | +| `failovers_count` | `int` | Empirical | Number of failover switches from initial profile. | +| `failover_reasons` | `Dict[str, int]` | Empirical | Histogram of failover triggers (`"quota_exhausted"`, `"rate_limited"`, `"auth_required"`, etc.). | + diff --git a/src/antigravity_provider/router/__init__.py b/src/antigravity_provider/router/__init__.py index 33f1061..c00ab10 100644 --- a/src/antigravity_provider/router/__init__.py +++ b/src/antigravity_provider/router/__init__.py @@ -16,6 +16,7 @@ from .health_tracker import ( ) from .session_affinity import LeaseManager, SessionAffinityRecord, SessionAffinityTracker from .router_engine import RouterEngine, get_router_engine +from .telemetry_service import TelemetryAggregates, TelemetryRecord, TelemetryService __all__ = [ "RouterConfig", @@ -37,4 +38,7 @@ __all__ = [ "LeaseManager", "RouterEngine", "get_router_engine", + "TelemetryService", + "TelemetryRecord", + "TelemetryAggregates", ] diff --git a/src/antigravity_provider/router/router_config.py b/src/antigravity_provider/router/router_config.py index 3546c41..1504544 100644 --- a/src/antigravity_provider/router/router_config.py +++ b/src/antigravity_provider/router/router_config.py @@ -44,6 +44,7 @@ class RouterConfig: session_affinity_ttl_seconds: int = 1800 roles: dict[str, RolePolicy] = field(default_factory=dict) profiles: dict[str, RouterProfileConfig] = field(default_factory=dict) + pricing: dict[str, dict[str, float]] = field(default_factory=dict) raw_router_block: dict[str, Any] = field(default_factory=dict) def get_profile(self, profile_id: str) -> Optional[RouterProfileConfig]: @@ -305,6 +306,16 @@ def load_router_config(config_path: Optional[Path] = None) -> RouterConfig: default_model=rdata.get("default_model"), ) + pricing_raw = data.get("pricing", {}) + pricing: dict[str, dict[str, float]] = {} + if isinstance(pricing_raw, dict): + for m_name, p_entry in pricing_raw.items(): + if isinstance(p_entry, dict): + pricing[m_name] = { + "input_cost_per_m": float(p_entry.get("input_cost_per_m", 0.0)), + "output_cost_per_m": float(p_entry.get("output_cost_per_m", 0.0)), + } + enabled = bool(r_block.get("enabled", data.get("enabled", True))) default_role = str(r_block.get("default_role", data.get("default_role", "orchestrator"))) max_failover = int(r_block.get("max_failover_attempts", data.get("max_failover_attempts", 3))) @@ -325,6 +336,7 @@ def load_router_config(config_path: Optional[Path] = None) -> RouterConfig: session_affinity_ttl_seconds=session_ttl, roles=roles or get_default_router_config().roles, profiles=profiles or get_default_router_config().profiles, + pricing=pricing, raw_router_block=r_block, ) except Exception as e: @@ -389,6 +401,8 @@ def save_router_config(config: RouterConfig, config_path: Optional[Path] = None) "roles": roles_data, "profiles": profiles_data, } + if config.pricing: + data["pricing"] = config.pricing existing_comments = [] if config_path.exists(): diff --git a/src/antigravity_provider/router/router_engine.py b/src/antigravity_provider/router/router_engine.py index 486ac47..f77c2f2 100644 --- a/src/antigravity_provider/router/router_engine.py +++ b/src/antigravity_provider/router/router_engine.py @@ -274,6 +274,36 @@ class RouterEngine: "models_evaluated": [m[1] for m in scored_candidates], }) + # Record in TelemetryService (extract usage ONLY if reported by provider) + prompt_tok = None + comp_tok = None + tot_tok = None + if isinstance(response, dict) and isinstance(response.get("usage"), dict): + u = response["usage"] + prompt_tok = u.get("prompt_tokens") or u.get("input_tokens") + comp_tok = u.get("completion_tokens") or u.get("output_tokens") + tot_tok = u.get("total_tokens") + if tot_tok is None and prompt_tok is not None and comp_tok is not None: + tot_tok = prompt_tok + comp_tok + + try: + from .telemetry_service import TelemetryService + TelemetryService.get().record_call( + role=target_role, + profile_id=pid, + provider=pconfig.provider, + model=exec_request.get("model") or selected_model or "", + outcome="failover" if failover_trail else "success", + latency_seconds=elapsed, + prompt_tokens=prompt_tok, + completion_tokens=comp_tok, + total_tokens=tot_tok, + failover_count=attempts - 1, + error_category=None, + ) + except Exception: + pass + # Set / update session affinity if target_session and affinity_enabled: self.affinity.set_affinity(target_session, target_role, pid, exec_request.get("model")) @@ -328,6 +358,26 @@ class RouterEngine: self.leases.release(pid) err_class = adapter.classify_error(exc) + # Record failed call in TelemetryService + try: + from .telemetry_service import TelemetryService + cat_name = err_class.category.value if hasattr(err_class.category, "value") else str(err_class.category) + TelemetryService.get().record_call( + role=target_role, + profile_id=pid, + provider=pconfig.provider, + model=exec_request.get("model") or requested_model or "", + outcome=cat_name, + latency_seconds=time.time() - t0, + prompt_tokens=None, + completion_tokens=None, + total_tokens=None, + failover_count=len(failover_trail), + error_category=cat_name, + ) + except Exception: + pass + if err_class.category == ErrorCategory.QUOTA_EXHAUSTED: self.health.mark_quota_exhausted( profile_id=pid, diff --git a/src/antigravity_provider/router/state_store.py b/src/antigravity_provider/router/state_store.py index 93aaf99..f76278f 100644 --- a/src/antigravity_provider/router/state_store.py +++ b/src/antigravity_provider/router/state_store.py @@ -149,6 +149,13 @@ class HubStateStore: self._latest_applied_seq = request_seq self._generation += 1 gen = self._generation + try: + from .telemetry_service import TelemetryService + telemetry_aggs = TelemetryService.get().get_aggregates() + telemetry_data = telemetry_aggs.to_dict() + except Exception: + telemetry_data = {"source": "own_measurement", "has_data": False} + metrics = { "generation": gen, "seq": request_seq, @@ -159,6 +166,7 @@ class HubStateStore: ), "refresh_runs_total": self.refresh_runs_total, "refresh_deduplicated_total": self.refresh_deduplicated_total, + "telemetry": telemetry_data, } snapshot = HubSnapshot( generation=gen, diff --git a/src/antigravity_provider/router/telemetry_service.py b/src/antigravity_provider/router/telemetry_service.py new file mode 100644 index 0000000..3cac5af --- /dev/null +++ b/src/antigravity_provider/router/telemetry_service.py @@ -0,0 +1,397 @@ +"""Hermes Hub — Real Runtime Call Telemetry & Metrics Service. + +Captures, persists, and computes honest empirical measurements for all router calls: +- Latency (P50, P95, Max) +- Real Token Usage (extracted ONLY from provider usage metadata; never guessed) +- Route Failovers and Failure Categories +- Cost Calculation (computed ONLY when explicit user model pricing is configured) +- Ring Buffer & File Rotation (bounded memory and disk footprint) +""" +from __future__ import annotations + +import collections +import datetime +import json +import logging +import math +import os +import threading +import time +from dataclasses import asdict, dataclass, field +from pathlib import Path +from typing import Any, Dict, List, Optional, Tuple + +from antigravity_provider import paths + +logger = logging.getLogger("hermes.router.telemetry") + +MAX_MEMORY_RECORDS = 10000 +MAX_FILE_BYTES = 5 * 1024 * 1024 # 5 MB +MAX_BACKUP_FILES = 3 + + +@dataclass +class TelemetryRecord: + """Immutable record of an individual router invocation attempt.""" + timestamp: float + iso_time: str + role: str + profile_id: str + provider: str + model: str + outcome: str # "success" | "failover" | "error" | "quota_exhausted" | "rate_limited" | "auth_required" + latency_seconds: float + prompt_tokens: Optional[int] = None + completion_tokens: Optional[int] = None + total_tokens: Optional[int] = None + cost_usd: Optional[float] = None + failover_count: int = 0 + error_category: Optional[str] = None + source: str = "own_measurement" + + def to_dict(self) -> Dict[str, Any]: + return {k: v for k, v in asdict(self).items() if v is not None or k in ("prompt_tokens", "completion_tokens", "total_tokens", "cost_usd")} + + +@dataclass +class TelemetryAggregates: + """Computed empirical metrics over a time window.""" + window_seconds: Optional[int] + total_calls: int + successful_calls: int + failed_calls: int + error_rate: Optional[float] # 0.0 - 1.0 or None if total_calls == 0 + latency_p50_ms: Optional[float] # Median latency in ms or None if total_calls == 0 + latency_p95_ms: Optional[float] # 95th percentile latency in ms or None if total_calls == 0 + latency_max_ms: Optional[float] # Maximum latency in ms or None if total_calls == 0 + total_prompt_tokens: Optional[int] # Sum of reported prompt tokens or None if no token data + total_completion_tokens: Optional[int] + total_tokens: Optional[int] + total_cost_usd: Optional[float] # Sum of calculated costs or None if no pricing available + failovers_count: int + failover_reasons: Dict[str, int] = field(default_factory=dict) + source: str = "own_measurement" + has_data: bool = True + + def to_dict(self) -> Dict[str, Any]: + return asdict(self) + + +class TelemetryService: + """Thread-safe persistent telemetry manager with bounded storage and honest aggregation.""" + + _instance: Optional[TelemetryService] = None + _instance_lock = threading.Lock() + + def __init__(self, log_path: Optional[Path] = None): + self._lock = threading.RLock() + self._buffer: collections.deque[TelemetryRecord] = collections.deque(maxlen=MAX_MEMORY_RECORDS) + self._pricing_table: Dict[str, Dict[str, float]] = {} + + if log_path: + self._log_path = log_path + else: + hermes_home = paths.get_hermes_home() + self._log_path = hermes_home / "telemetry.jsonl" + + self._load_pricing() + self._load_recent_history() + + @classmethod + def get(cls) -> TelemetryService: + if cls._instance is None: + with cls._instance_lock: + if cls._instance is None: + cls._instance = cls() + return cls._instance + + def set_pricing_table(self, pricing: Dict[str, Dict[str, float]]) -> None: + """Set or update the in-memory pricing table: {model_id_or_pattern: {input_cost_per_m: float, output_cost_per_m: float}}.""" + with self._lock: + self._pricing_table = dict(pricing) + + def _load_pricing(self) -> None: + """Load optional pricing table from router_profiles.yaml or pricing.yaml.""" + try: + from .router_config import load_router_config + cfg = load_router_config() + if hasattr(cfg, "pricing") and isinstance(cfg.pricing, dict): + self._pricing_table = dict(cfg.pricing) + return + except Exception: + pass + + # Check for config/pricing.yaml or ~/.hermes/pricing.yaml + for p in [paths.get_hermes_home() / "pricing.yaml", paths.get_repo_root() / "config" / "pricing.yaml"]: + if p.is_file(): + try: + import yaml + data = yaml.safe_dump(p.read_text(encoding="utf-8")) + if isinstance(data, dict) and "pricing" in data: + self._pricing_table = dict(data["pricing"]) + return + except Exception: + pass + + def compute_cost(self, model: str, prompt_tokens: Optional[int], completion_tokens: Optional[int]) -> Optional[float]: + """Compute USD cost for token usage if pricing is configured for this model; otherwise None.""" + if prompt_tokens is None and completion_tokens is None: + return None + if not self._pricing_table: + return None + + p_tok = prompt_tokens or 0 + c_tok = completion_tokens or 0 + + # Exact match or normalized model match + m_lower = model.lower().strip() + price_entry = None + for k, v in self._pricing_table.items(): + k_lower = k.lower().strip() + if k_lower == m_lower or k_lower in m_lower or m_lower in k_lower: + price_entry = v + break + + if not price_entry or not isinstance(price_entry, dict): + return None + + in_rate = float(price_entry.get("input_cost_per_m", 0.0)) + out_rate = float(price_entry.get("output_cost_per_m", 0.0)) + + cost = (p_tok * in_rate + c_tok * out_rate) / 1_000_000.0 + return round(cost, 6) + + def record_call( + self, + role: str, + profile_id: str, + provider: str, + model: str, + outcome: str, + latency_seconds: float, + prompt_tokens: Optional[int] = None, + completion_tokens: Optional[int] = None, + total_tokens: Optional[int] = None, + failover_count: int = 0, + error_category: Optional[str] = None, + ) -> 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 + + cost_usd = self.compute_cost(model, prompt_tokens, completion_tokens) + + record = TelemetryRecord( + timestamp=now, + iso_time=iso, + role=role, + profile_id=profile_id, + provider=provider, + 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, + cost_usd=cost_usd, + failover_count=failover_count, + error_category=error_category, + ) + + with self._lock: + self._buffer.append(record) + + self._append_to_disk(record) + return record + + def _append_to_disk(self, record: TelemetryRecord) -> None: + """Append record to disk with size-based log rotation.""" + try: + self._log_path.parent.mkdir(parents=True, exist_ok=True) + + # Check rotation + if self._log_path.is_file() and self._log_path.stat().st_size > MAX_FILE_BYTES: + self._rotate_logs() + + line = json.dumps(record.to_dict(), ensure_ascii=False) + "\n" + with open(self._log_path, "a", encoding="utf-8") as f: + f.write(line) + except Exception as exc: + logger.debug("Failed to write telemetry record to disk: %s", exc) + + def _rotate_logs(self) -> None: + """Rotate telemetry log files keeping up to MAX_BACKUP_FILES.""" + try: + for i in range(MAX_BACKUP_FILES - 1, 0, -1): + s_file = self._log_path.with_name(f"{self._log_path.name}.{i}") + d_file = self._log_path.with_name(f"{self._log_path.name}.{i + 1}") + if s_file.exists(): + s_file.replace(d_file) + + first_backup = self._log_path.with_name(f"{self._log_path.name}.1") + self._log_path.replace(first_backup) + except Exception as exc: + logger.debug("Telemetry log rotation failed: %s", exc) + + def _load_recent_history(self) -> None: + """Load recent records from disk into memory ring buffer.""" + if not self._log_path.is_file(): + return + try: + lines = [] + with open(self._log_path, "r", encoding="utf-8") as f: + for line in f: + line = line.strip() + if line: + lines.append(line) + + # Load up to last MAX_MEMORY_RECORDS + recent = lines[-MAX_MEMORY_RECORDS:] + for r_str in recent: + try: + d = json.loads(r_str) + rec = TelemetryRecord( + timestamp=float(d.get("timestamp", 0.0)), + iso_time=d.get("iso_time", ""), + role=d.get("role", ""), + profile_id=d.get("profile_id", ""), + provider=d.get("provider", ""), + model=d.get("model", ""), + outcome=d.get("outcome", "success"), + latency_seconds=float(d.get("latency_seconds", 0.0)), + prompt_tokens=d.get("prompt_tokens"), + completion_tokens=d.get("completion_tokens"), + total_tokens=d.get("total_tokens"), + cost_usd=d.get("cost_usd"), + failover_count=int(d.get("failover_count", 0)), + error_category=d.get("error_category"), + source=d.get("source", "own_measurement"), + ) + self._buffer.append(rec) + except Exception: + continue + except Exception as exc: + logger.debug("Failed loading telemetry history from disk: %s", exc) + + def get_aggregates( + self, + window_seconds: Optional[int] = None, + provider: Optional[str] = None, + profile_id: Optional[str] = None, + model: Optional[str] = None, + role: Optional[str] = None, + ) -> TelemetryAggregates: + """Compute empirical aggregates for matching calls over an optional time window.""" + now = time.time() + cutoff = (now - window_seconds) if window_seconds is not None else 0.0 + + with self._lock: + records = [ + r for r in self._buffer + if r.timestamp >= cutoff + and (provider is None or r.provider == provider) + and (profile_id is None or r.profile_id == profile_id) + and (model is None or r.model == model) + and (role is None or r.role == role) + ] + + total_calls = len(records) + if total_calls == 0: + return TelemetryAggregates( + window_seconds=window_seconds, + total_calls=0, + successful_calls=0, + failed_calls=0, + error_rate=None, + latency_p50_ms=None, + latency_p95_ms=None, + latency_max_ms=None, + total_prompt_tokens=None, + total_completion_tokens=None, + total_tokens=None, + total_cost_usd=None, + failovers_count=0, + failover_reasons={}, + source="own_measurement", + has_data=False, + ) + + successful_calls = 0 + failed_calls = 0 + latencies_ms: List[float] = [] + prompt_tokens_sum = 0 + completion_tokens_sum = 0 + has_any_token_data = False + costs_sum = 0.0 + has_any_cost_data = False + failovers_count = 0 + failover_reasons: Dict[str, int] = collections.defaultdict(int) + + for r in records: + latencies_ms.append(r.latency_seconds * 1000.0) + + if r.outcome in ("success", "failover"): + successful_calls += 1 + else: + failed_calls += 1 + + if r.failover_count > 0 or r.outcome == "failover": + failovers_count += 1 + reason_key = r.error_category or r.outcome + failover_reasons[reason_key] += 1 + + if r.prompt_tokens is not None: + prompt_tokens_sum += r.prompt_tokens + has_any_token_data = True + if r.completion_tokens is not None: + completion_tokens_sum += r.completion_tokens + has_any_token_data = True + + if r.cost_usd is not None: + costs_sum += r.cost_usd + has_any_cost_data = True + + latencies_ms.sort() + p50 = self._percentile(latencies_ms, 50) + p95 = self._percentile(latencies_ms, 95) + max_lat = latencies_ms[-1] if latencies_ms else None + + 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 + + return TelemetryAggregates( + window_seconds=window_seconds, + total_calls=total_calls, + successful_calls=successful_calls, + failed_calls=failed_calls, + error_rate=error_rate, + latency_p50_ms=round(p50, 1) if p50 is not None else None, + latency_p95_ms=round(p95, 1) if p95 is not None else None, + latency_max_ms=round(max_lat, 1) if max_lat is not None else None, + 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_cost_usd=round(costs_sum, 4) if has_any_cost_data else None, + failovers_count=failovers_count, + failover_reasons=dict(failover_reasons), + source="own_measurement", + has_data=True, + ) + + @staticmethod + def _percentile(sorted_data: List[float], percent: int) -> Optional[float]: + """Compute the n-th percentile from a pre-sorted numeric list.""" + if not sorted_data: + return None + k = (len(sorted_data) - 1) * (percent / 100.0) + f = math.floor(k) + c = math.ceil(k) + if f == c: + return float(sorted_data[int(k)]) + d0 = sorted_data[int(f)] * (c - k) + d1 = sorted_data[int(c)] * (k - f) + return float(d0 + d1) diff --git a/tests/test_router_telemetry.py b/tests/test_router_telemetry.py new file mode 100644 index 0000000..fd66398 --- /dev/null +++ b/tests/test_router_telemetry.py @@ -0,0 +1,347 @@ +"""Hermes Hub — Tests for Real Call Telemetry, Empirical Aggregates, and Debts Verification.""" +from __future__ import annotations + +import json +import os +import tempfile +from pathlib import Path +from typing import Any, Dict +from unittest.mock import MagicMock, patch +import pytest + +from antigravity_provider.router.telemetry_service import ( + TelemetryAggregates, + TelemetryRecord, + TelemetryService, + MAX_MEMORY_RECORDS, +) +from antigravity_provider.router.router_config import ( + RolePolicy, + RouterConfig, + RouterProfileConfig, + load_router_config, + save_router_config, +) +from antigravity_provider.router.router_engine import RouterEngine + + +@pytest.fixture +def temp_telemetry_service(tmp_path): + """Fixture providing an isolated TelemetryService with a temporary file log.""" + log_file = tmp_path / "test_telemetry.jsonl" + service = TelemetryService(log_path=log_file) + return service + + +@pytest.mark.unit +def test_telemetry_recording_latency_tokens_outcome(temp_telemetry_service): + """Verify individual call recording captures accurate metadata and token usage.""" + svc = temp_telemetry_service + + rec = svc.record_call( + role="orchestrator", + profile_id="ag-orch", + provider="antigravity", + model="gemini-2.5-pro", + outcome="success", + latency_seconds=0.125, + prompt_tokens=150, + completion_tokens=50, + total_tokens=200, + failover_count=0, + ) + + assert rec.role == "orchestrator" + assert rec.profile_id == "ag-orch" + assert rec.provider == "antigravity" + assert rec.model == "gemini-2.5-pro" + assert rec.outcome == "success" + assert rec.latency_seconds == 0.125 + assert rec.prompt_tokens == 150 + assert rec.completion_tokens == 50 + assert rec.total_tokens == 200 + assert rec.source == "own_measurement" + + # Verify persisted to disk + log_path = svc._log_path + assert log_path.is_file() + lines = log_path.read_text(encoding="utf-8").strip().splitlines() + assert len(lines) == 1 + disk_data = json.loads(lines[0]) + assert disk_data["profile_id"] == "ag-orch" + assert disk_data["total_tokens"] == 200 + + +@pytest.mark.unit +def test_telemetry_aggregates_calculation(temp_telemetry_service): + """Verify empirical metric calculation (P50, P95, Max, Token Sums, Error Rate) with known values.""" + svc = temp_telemetry_service + + # Record 10 calls with deterministic latencies: 100ms, 200ms, ..., 1000ms + for i in range(1, 11): + outcome = "success" if i <= 8 else "error" + error_cat = None if outcome == "success" else "quota_exhausted" + svc.record_call( + role="coder-primary", + profile_id="codex-w1", + provider="openai-codex", + model="gpt-4o", + outcome=outcome, + latency_seconds=i * 0.1, # 100ms to 1000ms + prompt_tokens=100, + completion_tokens=50, + total_tokens=150, + failover_count=1 if i == 9 else 0, + error_category=error_cat, + ) + + aggs = svc.get_aggregates() + + assert aggs.has_data is True + assert aggs.total_calls == 10 + assert aggs.successful_calls == 8 + assert aggs.failed_calls == 2 + assert aggs.error_rate == 0.2 # 2/10 + assert aggs.total_prompt_tokens == 1000 + assert aggs.total_completion_tokens == 500 + assert aggs.total_tokens == 1500 + assert aggs.latency_max_ms == 1000.0 + assert aggs.source == "own_measurement" + + # Median (P50) of [100, 200, 300, 400, 500, 600, 700, 800, 900, 1000] = 550.0 + assert aggs.latency_p50_ms == 550.0 + # P95 = 955.0 + assert aggs.latency_p95_ms == 955.0 + + +@pytest.mark.unit +def test_telemetry_aggregates_empty_window(temp_telemetry_service): + """Verify that when no calls exist, aggregates return None and has_data=False, NOT zero.""" + svc = temp_telemetry_service + aggs = svc.get_aggregates(window_seconds=3600) + + assert aggs.has_data is False + assert aggs.total_calls == 0 + assert aggs.successful_calls == 0 + assert aggs.failed_calls == 0 + assert aggs.error_rate is None + assert aggs.latency_p50_ms is None + assert aggs.latency_p95_ms is None + assert aggs.latency_max_ms is None + assert aggs.total_prompt_tokens is None + assert aggs.total_completion_tokens is None + assert aggs.total_tokens is None + assert aggs.total_cost_usd is None + assert aggs.source == "own_measurement" + + +@pytest.mark.unit +def test_telemetry_no_tokens_when_usage_missing(temp_telemetry_service): + """Verify that if provider does not return usage, token fields remain None without guessing.""" + svc = temp_telemetry_service + + rec = svc.record_call( + role="fast", + profile_id="opengo-1", + provider="opencode-go", + model="deepseek-v4-flash", + outcome="success", + latency_seconds=0.05, + prompt_tokens=None, + completion_tokens=None, + total_tokens=None, + ) + + assert rec.prompt_tokens is None + assert rec.completion_tokens is None + assert rec.total_tokens is None + + aggs = svc.get_aggregates() + assert aggs.total_calls == 1 + assert aggs.total_prompt_tokens is None + assert aggs.total_completion_tokens is None + assert aggs.total_tokens is None + + +@pytest.mark.unit +def test_telemetry_no_secrets_and_no_request_response_content(temp_telemetry_service): + """Verify that telemetry records only metadata and strictly excludes secrets or prompt/response payloads.""" + svc = temp_telemetry_service + + # Record a normal call + rec = svc.record_call( + role="orchestrator", + profile_id="ag-w1", + provider="antigravity", + model="gemini-2.5-pro", + outcome="success", + latency_seconds=0.45, + prompt_tokens=300, + completion_tokens=100, + ) + + d = rec.to_dict() + # Ensure no content or secret fields exist + forbidden_keys = [ + "messages", "prompt_text", "completion_text", "content", "response", "payload", + "api_key", "secret", "password", "bearer", "authorization" + ] + for key in d.keys(): + for forbidden in forbidden_keys: + assert forbidden != key.lower() and forbidden not in key.lower(), f"Forbidden key '{key}' found in telemetry record!" + + log_content = svc._log_path.read_text(encoding="utf-8") + assert "sk-" not in log_content + assert "bearer" not in log_content.lower() + + +@pytest.mark.unit +def test_telemetry_storage_rotation(tmp_path): + """Verify log rotation occurs when file exceeds MAX_FILE_BYTES and bounded memory buffer.""" + log_file = tmp_path / "telemetry_rot.jsonl" + svc = TelemetryService(log_path=log_file) + + # Patch MAX_FILE_BYTES temporarily to 500 bytes to test file rotation cleanly + with patch("antigravity_provider.router.telemetry_service.MAX_FILE_BYTES", 400): + for i in range(25): + svc.record_call( + role="coder", + profile_id=f"prof-{i}", + provider="antigravity", + model="gemini-2.5-flash", + outcome="success", + latency_seconds=0.1, + prompt_tokens=10, + completion_tokens=20, + ) + + # Main file and backup .1 must exist + assert log_file.exists() + backup_1 = log_file.with_name(f"{log_file.name}.1") + assert backup_1.exists() + + +@pytest.mark.unit +def test_telemetry_cost_calculation_with_and_without_pricing(temp_telemetry_service): + """Verify USD cost computation only runs when explicit pricing is configured.""" + svc = temp_telemetry_service + + # 1. Without pricing -> cost_usd is None + rec1 = svc.record_call( + role="coder-primary", + profile_id="ag-w1", + provider="antigravity", + model="gemini-2.5-pro", + outcome="success", + latency_seconds=0.2, + prompt_tokens=1_000_000, + completion_tokens=500_000, + ) + assert rec1.cost_usd is None + + # 2. Configure pricing: $1.25 per 1M prompt, $5.00 per 1M completion + svc.set_pricing_table({ + "gemini-2.5-pro": {"input_cost_per_m": 1.25, "output_cost_per_m": 5.00} + }) + + rec2 = svc.record_call( + role="coder-primary", + profile_id="ag-w1", + provider="antigravity", + model="gemini-2.5-pro", + outcome="success", + latency_seconds=0.2, + prompt_tokens=1_000_000, + completion_tokens=500_000, + ) + # Expected cost = 1.0 * 1.25 + 0.5 * 5.00 = 1.25 + 2.50 = 3.75 USD + assert rec2.cost_usd == 3.75 + + aggs = svc.get_aggregates() + assert aggs.total_cost_usd == 3.75 + + +@pytest.mark.unit +def test_router_engine_records_telemetry_end_to_end(tmp_path, monkeypatch): + """Verify that RouterEngine.route_request automatically records telemetry on success and failover.""" + monkeypatch.setenv("HERMES_HOME", str(tmp_path / "hermes")) + + log_file = tmp_path / "hermes" / "telemetry.jsonl" + svc = TelemetryService(log_path=log_file) + + 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"], + ) + } + ) + + from antigravity_provider.router.adapters.antigravity_adapter import AntigravityAdapter + + def mock_invoke(profile, req): + return { + "id": "chatcmpl-telemetry-test", + "choices": [{"message": {"role": "assistant", "content": "Telemetry verified"}}], + "usage": { + "prompt_tokens": 120, + "completion_tokens": 40, + "total_tokens": 160, + }, + } + + engine = RouterEngine(config=config) + + with patch.object(TelemetryService, "get", return_value=svc), \ + patch.object(AntigravityAdapter, "invoke", side_effect=mock_invoke): + + res = engine.route_request({"messages": [{"role": "user", "content": "hello"}]}, role="orchestrator") + + assert "router_metadata" in res + aggs = svc.get_aggregates() + assert aggs.total_calls == 1 + assert aggs.successful_calls == 1 + assert aggs.total_prompt_tokens == 120 + assert aggs.total_completion_tokens == 40 + assert aggs.total_tokens == 160 + + +@pytest.mark.unit +def test_yaml_comments_preservation_and_debts_closure(tmp_path): + """P1-5: Verify router_profiles.yaml preservation of existing header comments & blank lines.""" + test_yaml = tmp_path / "router_profiles.yaml" + initial_content = ( + "# Line 1: Header comment\n" + "# Line 2: Purpose description\n" + "# Line 3: Invariant notice\n" + "\n" + "# Line 5: Section header\n" + "router:\n" + " enabled: true\n" + " default_role: orchestrator\n" + "roles: {}\n" + "profiles: {}\n" + ) + test_yaml.write_text(initial_content, encoding="utf-8") + + cfg = load_router_config(test_yaml) + assert cfg.enabled is True + + # Save and reload + save_router_config(cfg, test_yaml) + saved_content = test_yaml.read_text(encoding="utf-8") + + # All 5 comment/blank lines before first YAML key must be preserved verbatim + assert "# Line 1: Header comment" in saved_content + assert "# Line 2: Purpose description" in saved_content + assert "# Line 3: Invariant notice" in saved_content + assert "# Line 5: Section header" in saved_content