feat(telemetry): implement empirical call telemetry, metrics aggregates, user pricing, and update state contract

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`

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

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

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",
 ]

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():

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,

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,

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)

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