athena-oracle/oracle/metrics.py

"""Metrics layer — structured telemetry for the ingestion pipeline.

Tracks per-run metrics, adapter health, verdict distribution, and score
statistics.  All stored in SQLite alongside entries for zero-cost persistence.

Usage inside a pipeline run:
    from oracle.metrics import MetricsRun
    m = MetricsRun(db_path)
    m.log_adapter('arxiv', fetched=42, errors=0, runtime_ms=1200)
    m.log_verdicts(publish=2, watch=28, archive=8, drop=4)
    m.log_scores([5.2, 6.1, 3.0, ...])
    m.log_summaries(useful=38, fallback=4)
    m.save()
"""
import json
import sqlite3
import time
from datetime import datetime, timezone
from pathlib import Path
from typing import Optional


SCHEMA = """
CREATE TABLE IF NOT EXISTS metrics (
    id INTEGER PRIMARY KEY,
    run_id INTEGER NOT NULL,
    metric_name TEXT NOT NULL,
    value REAL,
    metadata TEXT,
    recorded_at TEXT NOT NULL
);

CREATE TABLE IF NOT EXISTS adapter_health (
    id INTEGER PRIMARY KEY,
    run_id INTEGER NOT NULL,
    adapter_name TEXT NOT NULL,
    items_fetched INTEGER DEFAULT 0,
    errors INTEGER DEFAULT 0,
    runtime_ms INTEGER DEFAULT 0,
    consecutive_failures INTEGER DEFAULT 0,
    last_error TEXT,
    recorded_at TEXT NOT NULL
);

-- Index for fast time-range queries
CREATE INDEX IF NOT EXISTS idx_metrics_run_id ON metrics(run_id);
CREATE INDEX IF NOT EXISTS idx_metrics_name ON metrics(metric_name);
CREATE INDEX IF NOT EXISTS idx_adapter_health_name ON adapter_health(adapter_name);
CREATE INDEX IF NOT EXISTS idx_adapter_health_run ON adapter_health(run_id);
"""


def migrate_metrics(conn: sqlite3.Connection) -> bool:
    """Idempotently create metrics tables. Returns True if tables were new."""
    cur = conn.execute(
        "SELECT name FROM sqlite_master WHERE type='table' AND name='metrics'"
    )
    was_missing = cur.fetchone() is None

    conn.executescript(SCHEMA)
    conn.commit()
    return was_missing


class MetricsRun:
    """Accumulate metrics for a single pipeline run, then persist atomically."""

    def __init__(self, db_path: str | Path):
        self.db_path = str(db_path)
        self.conn = sqlite3.connect(self.db_path)
        self.conn.row_factory = sqlite3.Row
        migrate_metrics(self.conn)

        # Generate a run_id from run_log
        self.run_id = self._resolve_run_id()
        self.now = datetime.now(timezone.utc).isoformat()

        self._adapters: list[dict] = []
        self._metrics: list[dict] = []
        self._score_values: list[float] = []
        self._verdict_counts: dict[str, int] = {}
        self._summary_counts: dict[str, int] = {"useful": 0, "fallback": 0}

    def _resolve_run_id(self) -> int:
        """Get or create the current run entry in run_log."""
        cur = self.conn.execute(
            "SELECT MAX(id) as rid FROM run_log WHERE date(run_time) = date('now', 'utc')"
        )
        rid = cur.fetchone()["rid"]
        if rid is None:
            self.conn.execute(
                "INSERT INTO run_log (run_time, total_fetched, total_stored, sources_ok, sources_failed, notes, failure_class) "
                "VALUES (?, 0, 0, '[]', '[]', 'metrics-session', '')",
                (self.now,),
            )
            self.conn.commit()
            return self.conn.execute("SELECT last_insert_rowid()").fetchone()[0]
        return rid

    # ── Public API ──

    def log_adapter(
        self,
        name: str,
        fetched: int = 0,
        errors: int = 0,
        runtime_ms: int = 0,
        last_error: str | None = None,
    ) -> None:
        """Record per-adapter health metrics."""
        # Calculate consecutive failures
        consecutive = self._consecutive_failures(name)
        if fetched > 0 or errors == 0:
            consecutive = 0  # Reset on success

        self._adapters.append({
            "run_id": self.run_id,
            "adapter_name": name,
            "items_fetched": fetched,
            "errors": errors,
            "runtime_ms": runtime_ms,
            "consecutive_failures": consecutive,
            "last_error": last_error,
            "recorded_at": self.now,
        })

    def log_verdicts(
        self,
        publish: int = 0,
        watch: int = 0,
        archive: int = 0,
        drop: int = 0,
    ) -> None:
        """Record verdict distribution for this run."""
        self._verdict_counts = {
            "PUBLISH": publish,
            "WATCH": watch,
            "ARCHIVE": archive,
            "DROP": drop,
        }
        total = publish + watch + archive + drop
        self._metrics.append({
            "run_id": self.run_id,
            "metric_name": "publish_rate",
            "value": publish / total if total else 0,
            "metadata": json.dumps({
                "publish": publish,
                "watch": watch,
                "archive": archive,
                "drop": drop,
                "total": total,
            }),
            "recorded_at": self.now,
        })

    def log_scores(self, scores: list[float]) -> None:
        """Record signal score distribution."""
        if not scores:
            return
        self._score_values = scores
        sorted_scores = sorted(scores)
        n = len(sorted_scores)
        self._metrics.append({
            "run_id": self.run_id,
            "metric_name": "score_mean",
            "value": sum(scores) / n,
            "metadata": json.dumps({
                "count": n,
                "min": sorted_scores[0],
                "max": sorted_scores[-1],
                "p25": sorted_scores[n // 4],
                "p50": sorted_scores[n // 2],
                "p90": sorted_scores[int(n * 0.9)],
                "p99": sorted_scores[int(n * 0.99)] if n > 100 else sorted_scores[-1],
            }),
            "recorded_at": self.now,
        })

    def log_summaries(self, useful: int = 0, fallback: int = 0) -> None:
        """Record summarization quality metrics."""
        total = useful + fallback
        self._summary_counts = {"useful": useful, "fallback": fallback}
        if total:
            self._metrics.append({
                "run_id": self.run_id,
                "metric_name": "summary_quality_rate",
                "value": useful / total,
                "metadata": json.dumps({
                    "useful": useful,
                    "fallback": fallback,
                    "total": total,
                }),
                "recorded_at": self.now,
            })

    def log_metric(
        self,
        name: str,
        value: float,
        metadata: dict | None = None,
    ) -> None:
        """Record an arbitrary metric."""
        self._metrics.append({
            "run_id": self.run_id,
            "metric_name": name,
            "value": value,
            "metadata": json.dumps(metadata) if metadata else None,
            "recorded_at": self.now,
        })

    def save(self) -> int:
        """Persist all accumulated metrics. Returns run_id."""
        # Write adapter health
        for a in self._adapters:
            self.conn.execute(
                "INSERT INTO adapter_health "
                "(run_id, adapter_name, items_fetched, errors, runtime_ms, "
                " consecutive_failures, last_error, recorded_at) "
                "VALUES (?, ?, ?, ?, ?, ?, ?, ?)",
                (
                    a["run_id"], a["adapter_name"], a["items_fetched"],
                    a["errors"], a["runtime_ms"], a["consecutive_failures"],
                    a["last_error"], a["recorded_at"],
                ),
            )

        # Write metrics
        for m in self._metrics:
            self.conn.execute(
                "INSERT INTO metrics "
                "(run_id, metric_name, value, metadata, recorded_at) "
                "VALUES (?, ?, ?, ?, ?)",
                (
                    m["run_id"], m["metric_name"], m["value"],
                    m["metadata"], m["recorded_at"],
                ),
            )

        # Update run_log totals
        total_fetched = sum(a["items_fetched"] for a in self._adapters)
        total_errors = sum(a["errors"] for a in self._adapters)
        sources_ok = [a["adapter_name"] for a in self._adapters if a["items_fetched"] > 0]
        sources_failed = [a["adapter_name"] for a in self._adapters if a["errors"] > 0]

        self.conn.execute(
            "UPDATE run_log SET total_fetched = ?, total_stored = ?, "
            "sources_ok = ?, sources_failed = ?, notes = ? "
            "WHERE id = ?",
            (
                total_fetched,
                total_fetched - total_errors,
                json.dumps(sources_ok),
                json.dumps(sources_failed),
                f"publish_rate={self._verdict_counts.get('PUBLISH', 0)}, "
                f"summary_quality={self._summary_counts.get('useful', 0)}/{sum(self._summary_counts.values()) or 1}",
                self.run_id,
            ),
        )

        self.conn.commit()
        return self.run_id

    def close(self) -> None:
        self.conn.close()

    # ── Helpers ──

    def _consecutive_failures(self, adapter_name: str) -> int:
        """Count consecutive runs where adapter returned 0 items."""
        cur = self.conn.execute(
            "SELECT items_fetched FROM adapter_health "
            "WHERE adapter_name = ? AND run_id < ? "
            "ORDER BY run_id DESC LIMIT 5",
            (adapter_name, self.run_id),
        )
        count = 0
        for row in cur:
            if row["items_fetched"] == 0:
                count += 1
            else:
                break
        return count


# ── Aggregation queries ──


def get_publish_trend(db_path: str, days: int = 30) -> list[dict]:
    """Daily PUBLISH rate over the last N days."""
    conn = sqlite3.connect(db_path)
    conn.row_factory = sqlite3.Row

    cur = conn.execute(
        "SELECT date(m.recorded_at) as day, "
        "ROUND(AVG(m.value), 4) as avg_publish_rate, "
        "COUNT(*) as runs "
        "FROM metrics m "
        "WHERE m.metric_name = 'publish_rate' "
        "AND m.recorded_at >= datetime('now', ?) "
        "GROUP BY day ORDER BY day DESC",
        (f"-{days} days",),
    )
    results = [dict(r) for r in cur.fetchall()]
    conn.close()
    return results


def get_adapter_health(db_path: str, days: int = 7) -> list[dict]:
    """Adapter health summary over last N days."""
    conn = sqlite3.connect(db_path)
    conn.row_factory = sqlite3.Row

    cur = conn.execute(
        "SELECT adapter_name, "
        "COUNT(*) as runs, "
        "SUM(items_fetched) as total_fetched, "
        "SUM(errors) as total_errors, "
        "MAX(consecutive_failures) as max_consecutive_failures, "
        "AVG(runtime_ms) as avg_runtime_ms, "
        "MAX(last_error) as last_error "
        "FROM adapter_health "
        "WHERE recorded_at >= datetime('now', ?) "
        "GROUP BY adapter_name",
        (f"-{days} days",),
    )
    results = [dict(r) for r in cur.fetchall()]
    conn.close()
    return results


def get_score_distribution(db_path: str, days: int = 30) -> list[dict]:
    """Score distribution percentiles over last N days."""
    conn = sqlite3.connect(db_path)
    conn.row_factory = sqlite3.Row

    cur = conn.execute(
        "SELECT date(recorded_at) as day, value as mean, metadata "
        "FROM metrics "
        "WHERE metric_name = 'score_mean' "
        "AND recorded_at >= datetime('now', ?) "
        "ORDER BY day DESC",
        (f"-{days} days",),
    )
    results = []
    for r in cur:
        meta = json.loads(r["metadata"]) if r["metadata"] else {}
        results.append({
            "day": r["day"],
            "mean": r["mean"],
            "min": meta.get("min"),
            "max": meta.get("max"),
            "p25": meta.get("p25"),
            "p50": meta.get("p50"),
            "p90": meta.get("p90"),
        })
    conn.close()
    return results


def get_summary_quality(db_path: str, days: int = 30) -> list[dict]:
    """Summarization quality rate over time."""
    conn = sqlite3.connect(db_path)
    conn.row_factory = sqlite3.Row

    cur = conn.execute(
        "SELECT date(recorded_at) as day, ROUND(AVG(value), 4) as avg_quality "
        "FROM metrics "
        "WHERE metric_name = 'summary_quality_rate' "
        "AND recorded_at >= datetime('now', ?) "
        "GROUP BY day ORDER BY day DESC",
        (f"-{days} days",),
    )
    results = [dict(r) for r in cur.fetchall()]
    conn.close()
    return results


def get_metric_alerts(db_path: str) -> list[str]:
    """Generate alert strings based on metric anomalies."""
    alerts = []

    # Adapter failures
    health = get_adapter_health(db_path, days=7)
    for h in health:
        if h["max_consecutive_failures"] and h["max_consecutive_failures"] >= 3:
            alerts.append(
                f"⚠ {h['adapter_name']}: {h['max_consecutive_failures']} consecutive "
                f"failed runs. Last error: {h.get('last_error', 'unknown')}"
            )
        if h["total_errors"] and h["runs"]:
            err_rate = h["total_errors"] / h["runs"] * 100
            if err_rate > 50:
                alerts.append(
                    f"⚠ {h['adapter_name']}: {err_rate:.0f}% error rate "
                    f"({h['total_errors']}/{h['runs']} runs)"
                )

    # Publish starvation
    trend = get_publish_trend(db_path, days=7)
    if trend:
        latest = trend[0]["avg_publish_rate"]
        if latest == 0:
            alerts.append("🔴 PUBLISH rate is 0% — scoring threshold may be too strict")
        elif latest < 0.01:
            alerts.append(
                f"🟡 PUBLISH rate is {latest:.1%} — "
                "consider lowering the signal threshold from 6.0"
            )

    # Summary quality
    quality = get_summary_quality(db_path, days=7)
    if quality:
        latest = quality[0]["avg_quality"]
        if latest < 0.5:
            alerts.append(
                f"🟡 Summary quality at {latest:.0%} — "
                "most items falling back to title-only"
            )

    return alerts