[claude] Add sovereignty metrics tracking + dashboard panel (#981) (#1083)

src/config.py

@@ -152,6 +152,10 @@ class Settings(BaseSettings):
     # Default is False (telemetry disabled) to align with sovereign AI vision.
     telemetry_enabled: bool = False
 
+    # ── Sovereignty Metrics ──────────────────────────────────────────────
+    # Alert when API cost per research task exceeds this threshold (USD).
+    sovereignty_api_cost_alert_threshold: float = 1.00
+
     # CORS allowed origins for the web chat interface (Gitea Pages, etc.)
     # Set CORS_ORIGINS as a comma-separated list, e.g. "http://localhost:3000,https://example.com"
     cors_origins: list[str] = [

src/dashboard/app.py

@@ -45,6 +45,7 @@ from dashboard.routes.models import api_router as models_api_router
 from dashboard.routes.models import router as models_router
 from dashboard.routes.quests import router as quests_router
 from dashboard.routes.scorecards import router as scorecards_router
+from dashboard.routes.sovereignty_metrics import router as sovereignty_metrics_router
 from dashboard.routes.spark import router as spark_router
 from dashboard.routes.system import router as system_router
 from dashboard.routes.tasks import router as tasks_router
@@ -631,6 +632,7 @@ app.include_router(tower_router)
 app.include_router(daily_run_router)
 app.include_router(quests_router)
 app.include_router(scorecards_router)
+app.include_router(sovereignty_metrics_router)
 
 
 @app.websocket("/ws")

src/dashboard/routes/sovereignty_metrics.py

@@ -0,0 +1,74 @@
+"""Sovereignty metrics dashboard routes.
+
+Provides API endpoints and HTMX partials for tracking research
+sovereignty progress against graduation targets.
+
+Refs: #981
+"""
+
+import logging
+from typing import Any
+
+from fastapi import APIRouter, Request
+from fastapi.responses import HTMLResponse
+
+from config import settings
+from dashboard.templating import templates
+from infrastructure.sovereignty_metrics import (
+    GRADUATION_TARGETS,
+    get_sovereignty_store,
+)
+
+logger = logging.getLogger(__name__)
+
+router = APIRouter(prefix="/sovereignty", tags=["sovereignty"])
+
+
+@router.get("/metrics")
+async def sovereignty_metrics_api() -> dict[str, Any]:
+    """JSON API: full sovereignty metrics summary with trends."""
+    store = get_sovereignty_store()
+    summary = store.get_summary()
+    alerts = store.get_alerts(unacknowledged_only=True)
+    return {
+        "metrics": summary,
+        "alerts": alerts,
+        "targets": GRADUATION_TARGETS,
+        "cost_threshold": settings.sovereignty_api_cost_alert_threshold,
+    }
+
+
+@router.get("/metrics/panel", response_class=HTMLResponse)
+async def sovereignty_metrics_panel(request: Request) -> HTMLResponse:
+    """HTMX partial: sovereignty metrics progress panel."""
+    store = get_sovereignty_store()
+    summary = store.get_summary()
+    alerts = store.get_alerts(unacknowledged_only=True)
+
+    return templates.TemplateResponse(
+        request,
+        "partials/sovereignty_metrics.html",
+        {
+            "metrics": summary,
+            "alerts": alerts,
+            "targets": GRADUATION_TARGETS,
+        },
+    )
+
+
+@router.get("/alerts")
+async def sovereignty_alerts_api() -> dict[str, Any]:
+    """JSON API: sovereignty alerts."""
+    store = get_sovereignty_store()
+    return {
+        "alerts": store.get_alerts(unacknowledged_only=False),
+        "unacknowledged": store.get_alerts(unacknowledged_only=True),
+    }
+
+
+@router.post("/alerts/{alert_id}/acknowledge")
+async def acknowledge_alert(alert_id: int) -> dict[str, bool]:
+    """Acknowledge a sovereignty alert."""
+    store = get_sovereignty_store()
+    success = store.acknowledge_alert(alert_id)
+    return {"success": success}

src/dashboard/templates/mission_control.html

@@ -179,6 +179,13 @@
   </div>
 </div>
 
+<!-- Sovereignty Metrics -->
+{% call panel("SOVEREIGNTY METRICS", id="sovereignty-metrics-panel",
+              hx_get="/sovereignty/metrics/panel",
+              hx_trigger="load, every 30s") %}
+  <p class="chat-history-placeholder">Loading sovereignty metrics...</p>
+{% endcall %}
+
 <!-- Chat History -->
 <div class="card mc-card-spaced">
     <div class="card-header">

src/dashboard/templates/partials/sovereignty_metrics.html

@@ -0,0 +1,63 @@
+{# HTMX partial: Sovereignty Metrics Progress Panel
+   Loaded via hx-get="/sovereignty/metrics/panel"
+   Refs: #981
+#}
+{% set phase_labels = {"pre-start": "Pre-start", "week1": "Week 1", "month1": "Month 1", "month3": "Month 3", "graduated": "Graduated"} %}
+{% set phase_colors = {"pre-start": "var(--text-dim)", "week1": "var(--red)", "month1": "var(--amber)", "month3": "var(--green)", "graduated": "var(--purple)"} %}
+
+{% set metric_labels = {
+  "cache_hit_rate": "Cache Hit Rate",
+  "api_cost": "API Cost / Task",
+  "time_to_report": "Time to Report",
+  "human_involvement": "Human Involvement",
+  "local_artifacts": "Local Artifacts"
+} %}
+
+{% set metric_units = {
+  "cache_hit_rate": "%",
+  "api_cost": "$",
+  "time_to_report": "min",
+  "human_involvement": "%",
+  "local_artifacts": ""
+} %}
+
+{% if alerts %}
+<div class="sov-alerts">
+  {% for alert in alerts %}
+  <div class="sov-alert-item">
+    <span class="sov-alert-icon">!</span>
+    <span>{{ alert.message }}</span>
+  </div>
+  {% endfor %}
+</div>
+{% endif %}
+
+<div class="grid grid-3">
+{% for key, data in metrics.items() %}
+  {% set label = metric_labels.get(key, key) %}
+  {% set unit = metric_units.get(key, "") %}
+  {% set phase = data.phase %}
+  {% set color = phase_colors.get(phase, "var(--text-dim)") %}
+  <div class="stat">
+    <div class="stat-value" style="color: {{ color }}">
+      {% if data.current is not none %}
+        {% if key == "cache_hit_rate" or key == "human_involvement" %}
+          {{ "%.0f"|format(data.current * 100) }}{{ unit }}
+        {% elif key == "api_cost" %}
+          {{ unit }}{{ "%.2f"|format(data.current) }}
+        {% elif key == "time_to_report" %}
+          {{ "%.1f"|format(data.current) }}{{ unit }}
+        {% else %}
+          {{ data.current|int }}
+        {% endif %}
+      {% else %}
+        --
+      {% endif %}
+    </div>
+    <div class="stat-label">{{ label }}</div>
+    <div class="stat-label" style="font-size: 0.7rem; color: {{ color }}">
+      {{ phase_labels.get(phase, phase) }}
+    </div>
+  </div>
+{% endfor %}
+</div>

src/infrastructure/sovereignty_metrics.py

@@ -0,0 +1,307 @@
+"""Sovereignty metrics collector and store.
+
+Tracks research sovereignty progress: cache hit rate, API cost,
+time-to-report, and human involvement. Persists to SQLite for
+trend analysis and dashboard display.
+
+Refs: #981
+"""
+
+import json
+import logging
+import sqlite3
+from contextlib import closing
+from dataclasses import dataclass, field
+from datetime import UTC, datetime
+from pathlib import Path
+from typing import Any
+
+from config import settings
+
+logger = logging.getLogger(__name__)
+
+DB_PATH = Path(settings.repo_root) / "data" / "sovereignty_metrics.db"
+
+_SCHEMA = """
+CREATE TABLE IF NOT EXISTS sovereignty_metrics (
+    id INTEGER PRIMARY KEY AUTOINCREMENT,
+    timestamp TEXT NOT NULL,
+    metric_type TEXT NOT NULL,
+    value REAL NOT NULL,
+    metadata TEXT DEFAULT '{}'
+);
+CREATE INDEX IF NOT EXISTS idx_sm_type ON sovereignty_metrics(metric_type);
+CREATE INDEX IF NOT EXISTS idx_sm_ts ON sovereignty_metrics(timestamp);
+
+CREATE TABLE IF NOT EXISTS sovereignty_alerts (
+    id INTEGER PRIMARY KEY AUTOINCREMENT,
+    timestamp TEXT NOT NULL,
+    alert_type TEXT NOT NULL,
+    message TEXT NOT NULL,
+    value REAL NOT NULL,
+    threshold REAL NOT NULL,
+    acknowledged INTEGER DEFAULT 0
+);
+CREATE INDEX IF NOT EXISTS idx_sa_ts ON sovereignty_alerts(timestamp);
+CREATE INDEX IF NOT EXISTS idx_sa_ack ON sovereignty_alerts(acknowledged);
+"""
+
+
+@dataclass
+class SovereigntyMetric:
+    """A single sovereignty metric data point."""
+
+    metric_type: str  # cache_hit_rate, api_cost, time_to_report, human_involvement
+    value: float
+    timestamp: str = field(default_factory=lambda: datetime.now(UTC).isoformat())
+    metadata: dict[str, Any] = field(default_factory=dict)
+
+
+@dataclass
+class SovereigntyAlert:
+    """An alert triggered when a metric exceeds a threshold."""
+
+    alert_type: str
+    message: str
+    value: float
+    threshold: float
+    timestamp: str = field(default_factory=lambda: datetime.now(UTC).isoformat())
+    acknowledged: bool = False
+
+
+# Graduation targets from issue #981
+GRADUATION_TARGETS = {
+    "cache_hit_rate": {"week1": 0.10, "month1": 0.40, "month3": 0.80, "graduation": 0.90},
+    "api_cost": {"week1": 1.50, "month1": 0.50, "month3": 0.10, "graduation": 0.01},
+    "time_to_report": {"week1": 180.0, "month1": 30.0, "month3": 5.0, "graduation": 1.0},
+    "human_involvement": {"week1": 1.0, "month1": 0.5, "month3": 0.25, "graduation": 0.0},
+    "local_artifacts": {"week1": 6, "month1": 30, "month3": 100, "graduation": 500},
+}
+
+
+class SovereigntyMetricsStore:
+    """SQLite-backed sovereignty metrics store.
+
+    Thread-safe: creates a new connection per operation.
+    """
+
+    def __init__(self, db_path: Path | None = None) -> None:
+        self._db_path = db_path or DB_PATH
+        self._init_db()
+
+    def _init_db(self) -> None:
+        """Initialize the database schema."""
+        try:
+            self._db_path.parent.mkdir(parents=True, exist_ok=True)
+            with closing(sqlite3.connect(str(self._db_path))) as conn:
+                conn.execute("PRAGMA journal_mode=WAL")
+                conn.execute(f"PRAGMA busy_timeout={settings.db_busy_timeout_ms}")
+                conn.executescript(_SCHEMA)
+                conn.commit()
+        except Exception as exc:
+            logger.warning("Failed to initialize sovereignty metrics DB: %s", exc)
+
+    def _connect(self) -> sqlite3.Connection:
+        """Get a new connection."""
+        conn = sqlite3.connect(str(self._db_path))
+        conn.row_factory = sqlite3.Row
+        conn.execute(f"PRAGMA busy_timeout={settings.db_busy_timeout_ms}")
+        return conn
+
+    def record(self, metric: SovereigntyMetric) -> None:
+        """Record a sovereignty metric data point."""
+        try:
+            with closing(self._connect()) as conn:
+                conn.execute(
+                    "INSERT INTO sovereignty_metrics (timestamp, metric_type, value, metadata) "
+                    "VALUES (?, ?, ?, ?)",
+                    (
+                        metric.timestamp,
+                        metric.metric_type,
+                        metric.value,
+                        json.dumps(metric.metadata),
+                    ),
+                )
+                conn.commit()
+        except Exception as exc:
+            logger.warning("Failed to record sovereignty metric: %s", exc)
+
+        # Check thresholds for alerts
+        self._check_alert(metric)
+
+    def _check_alert(self, metric: SovereigntyMetric) -> None:
+        """Check if a metric triggers an alert."""
+        threshold = settings.sovereignty_api_cost_alert_threshold
+        if metric.metric_type == "api_cost" and metric.value > threshold:
+            alert = SovereigntyAlert(
+                alert_type="api_cost_exceeded",
+                message=f"API cost ${metric.value:.2f} exceeds threshold ${threshold:.2f}",
+                value=metric.value,
+                threshold=threshold,
+            )
+            self._record_alert(alert)
+
+    def _record_alert(self, alert: SovereigntyAlert) -> None:
+        """Persist an alert."""
+        try:
+            with closing(self._connect()) as conn:
+                conn.execute(
+                    "INSERT INTO sovereignty_alerts "
+                    "(timestamp, alert_type, message, value, threshold) "
+                    "VALUES (?, ?, ?, ?, ?)",
+                    (
+                        alert.timestamp,
+                        alert.alert_type,
+                        alert.message,
+                        alert.value,
+                        alert.threshold,
+                    ),
+                )
+                conn.commit()
+            logger.warning("Sovereignty alert: %s", alert.message)
+        except Exception as exc:
+            logger.warning("Failed to record sovereignty alert: %s", exc)
+
+    def get_latest(self, metric_type: str, limit: int = 50) -> list[dict]:
+        """Get the most recent metric values for a given type."""
+        try:
+            with closing(self._connect()) as conn:
+                rows = conn.execute(
+                    "SELECT timestamp, value, metadata FROM sovereignty_metrics "
+                    "WHERE metric_type = ? ORDER BY timestamp DESC LIMIT ?",
+                    (metric_type, limit),
+                ).fetchall()
+                return [
+                    {
+                        "timestamp": row["timestamp"],
+                        "value": row["value"],
+                        "metadata": json.loads(row["metadata"]) if row["metadata"] else {},
+                    }
+                    for row in rows
+                ]
+        except Exception as exc:
+            logger.warning("Failed to query sovereignty metrics: %s", exc)
+            return []
+
+    def get_summary(self) -> dict[str, Any]:
+        """Get a summary of current sovereignty metrics progress."""
+        summary: dict[str, Any] = {}
+        for metric_type in GRADUATION_TARGETS:
+            latest = self.get_latest(metric_type, limit=1)
+            history = self.get_latest(metric_type, limit=30)
+
+            current_value = latest[0]["value"] if latest else None
+            targets = GRADUATION_TARGETS[metric_type]
+
+            # Determine current phase based on value
+            phase = "pre-start"
+            if current_value is not None:
+                if metric_type in ("api_cost", "time_to_report", "human_involvement"):
+                    # Lower is better
+                    if current_value <= targets["graduation"]:
+                        phase = "graduated"
+                    elif current_value <= targets["month3"]:
+                        phase = "month3"
+                    elif current_value <= targets["month1"]:
+                        phase = "month1"
+                    elif current_value <= targets["week1"]:
+                        phase = "week1"
+                    else:
+                        phase = "pre-start"
+                else:
+                    # Higher is better
+                    if current_value >= targets["graduation"]:
+                        phase = "graduated"
+                    elif current_value >= targets["month3"]:
+                        phase = "month3"
+                    elif current_value >= targets["month1"]:
+                        phase = "month1"
+                    elif current_value >= targets["week1"]:
+                        phase = "week1"
+                    else:
+                        phase = "pre-start"
+
+            summary[metric_type] = {
+                "current": current_value,
+                "phase": phase,
+                "targets": targets,
+                "trend": [{"t": h["timestamp"], "v": h["value"]} for h in reversed(history)],
+            }
+
+        return summary
+
+    def get_alerts(self, unacknowledged_only: bool = True, limit: int = 20) -> list[dict]:
+        """Get sovereignty alerts."""
+        try:
+            with closing(self._connect()) as conn:
+                if unacknowledged_only:
+                    rows = conn.execute(
+                        "SELECT * FROM sovereignty_alerts "
+                        "WHERE acknowledged = 0 ORDER BY timestamp DESC LIMIT ?",
+                        (limit,),
+                    ).fetchall()
+                else:
+                    rows = conn.execute(
+                        "SELECT * FROM sovereignty_alerts "
+                        "ORDER BY timestamp DESC LIMIT ?",
+                        (limit,),
+                    ).fetchall()
+                return [dict(row) for row in rows]
+        except Exception as exc:
+            logger.warning("Failed to query sovereignty alerts: %s", exc)
+            return []
+
+    def acknowledge_alert(self, alert_id: int) -> bool:
+        """Acknowledge an alert."""
+        try:
+            with closing(self._connect()) as conn:
+                conn.execute(
+                    "UPDATE sovereignty_alerts SET acknowledged = 1 WHERE id = ?",
+                    (alert_id,),
+                )
+                conn.commit()
+                return True
+        except Exception as exc:
+            logger.warning("Failed to acknowledge alert: %s", exc)
+            return False
+
+
+# ── Module-level singleton ─────────────────────────────────────────────────
+_store: SovereigntyMetricsStore | None = None
+
+
+def get_sovereignty_store() -> SovereigntyMetricsStore:
+    """Return the module-level store, creating it on first access."""
+    global _store
+    if _store is None:
+        _store = SovereigntyMetricsStore()
+    return _store
+
+
+async def emit_sovereignty_metric(
+    metric_type: str,
+    value: float,
+    metadata: dict[str, Any] | None = None,
+) -> None:
+    """Convenience function to record a sovereignty metric and emit an event.
+
+    Also publishes to the event bus for real-time subscribers.
+    """
+    import asyncio
+
+    from infrastructure.events.bus import emit
+
+    metric = SovereigntyMetric(
+        metric_type=metric_type,
+        value=value,
+        metadata=metadata or {},
+    )
+    # Record to SQLite in thread to avoid blocking event loop
+    await asyncio.to_thread(get_sovereignty_store().record, metric)
+
+    # Publish to event bus for real-time consumers
+    await emit(
+        f"sovereignty.metric.{metric_type}",
+        source="sovereignty_metrics",
+        data={"metric_type": metric_type, "value": value, **(metadata or {})},
+    )