acdl/core/metrics/collector.py

"""Nova Metrics Collector (REQ-189, P2).

Reads all grounded signals (REGRESSION_REPORT.json, per-run manifests,
junit XML, pcr.json, signal.json, COST.md, decision ledger, coverage.json)
and normalizes them into a SQLite cold store at metrics/nova_metrics.db.

D-120: Nova-native (SQLite, no ClickHouse/BigQuery).
D-125: hybrid model — reads files + events → SQLite.
D-126: cold-only (no hot path; hot path deferred D-096).
D-128: metrics/ at repo root.

Idempotent: re-running the collector against the same inputs produces
identical row counts (REQ-200). The collector uses INSERT OR REPLACE
on fact tables keyed by natural keys.
"""

import datetime
import json
import os
import sqlite3
import sys
import xml.etree.ElementTree as ET

_METRICS_DIR = os.path.join(os.path.dirname(os.path.dirname(os.path.dirname(os.path.abspath(__file__)))), "metrics")
_STORE_PATH = os.path.join(_METRICS_DIR, "nova_metrics.db")
_REPO_ROOT = os.path.dirname(os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
_REGRESSION_REPORT = os.path.join(_REPO_ROOT, ".ciagent", "REGRESSION_REPORT.json")
_RUNS_DIR = os.path.join(_METRICS_DIR, "runs")
_LEDGER_DB = os.path.join(_METRICS_DIR, "decision_ledger.db")
_COVERAGE_JSON = os.path.join(_METRICS_DIR, "coverage.json")
_TEST_RESULTS_XML = os.path.join(_METRICS_DIR, "test-results.xml")


def _iso8601_now():
    return datetime.datetime.now(datetime.timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ")


def _init_store(db_path=None):
    """Create the fact/dim tables in the SQLite cold store."""
    if db_path is None:
        db_path = _STORE_PATH
    os.makedirs(os.path.dirname(db_path), exist_ok=True)
    conn = sqlite3.connect(db_path)
    conn.executescript("""
        CREATE TABLE IF NOT EXISTS fact_run (
            run_id TEXT PRIMARY KEY,
            contract_id TEXT,
            environment TEXT,
            started_at TEXT,
            completed_at TEXT,
            exit_code INTEGER,
            outcome TEXT,
            confidence_score REAL,
            confidence_band TEXT,
            hitl_block INTEGER,
            cost_estimate_usd REAL,
            decision_id TEXT,
            escalation_reason TEXT
        );

        CREATE TABLE IF NOT EXISTS fact_capability (
            capability_id TEXT,
            run_id TEXT,
            name TEXT,
            status TEXT,
            tier TEXT,
            duration_ms REAL,
            detail TEXT,
            run_at_utc TEXT,
            PRIMARY KEY (capability_id, run_id)
        );

        CREATE TABLE IF NOT EXISTS fact_policy_check (
            run_id TEXT,
            rule_id TEXT,
            severity TEXT,
            result TEXT,
            resource_ref TEXT,
            evaluated_at TEXT,
            PRIMARY KEY (run_id, rule_id, resource_ref)
        );

        CREATE TABLE IF NOT EXISTS fact_confidence (
            run_id TEXT,
            score REAL,
            band TEXT,
            per_input TEXT,
            reason_codes TEXT,
            environment TEXT,
            computed_at TEXT,
            PRIMARY KEY (run_id)
        );

        CREATE TABLE IF NOT EXISTS fact_test (
            run_id TEXT,
            total_tests INTEGER,
            passed INTEGER,
            failed INTEGER,
            errors INTEGER,
            skipped INTEGER,
            duration_s REAL,
            coverage_pct REAL,
            collected_at TEXT,
            PRIMARY KEY (run_id)
        );

        CREATE TABLE IF NOT EXISTS fact_decision (
            decision_id TEXT,
            run_id TEXT,
            chosen_action TEXT,
            confidence REAL,
            alternatives TEXT,
            human_override INTEGER,
            escalation_reason TEXT,
            outcome TEXT,
            backfilled_at TEXT,
            event_time TEXT,
            PRIMARY KEY (decision_id)
        );

        CREATE TABLE IF NOT EXISTS fact_cost_estimate (
            run_id TEXT,
            delta_usd REAL,
            total_monthly_usd REAL,
            available INTEGER,
            estimated_at TEXT,
            PRIMARY KEY (run_id)
        );

        CREATE TABLE IF NOT EXISTS fact_lifecycle (
            module TEXT,
            environment TEXT,
            phase TEXT,
            result TEXT,
            duration_ms REAL,
            run_at TEXT,
            PRIMARY KEY (module, environment, phase, run_at)
        );

        CREATE TABLE IF NOT EXISTS dim_capability (
            capability_id TEXT PRIMARY KEY,
            name TEXT,
            tier TEXT,
            source_milestone TEXT
        );

        CREATE TABLE IF NOT EXISTS dim_milestone (
            milestone TEXT PRIMARY KEY,
            phase INTEGER,
            tag TEXT,
            completed_at TEXT
        );
    """)
    conn.commit()
    conn.close()


def collect_regression_report(db_path=None, report_path=None):
    """Read REGRESSION_REPORT.json → fact_capability + dim_capability."""
    if db_path is None:
        db_path = _STORE_PATH
    if report_path is None:
        report_path = _REGRESSION_REPORT
    if not os.path.isfile(report_path):
        return 0
    _init_store(db_path)
    with open(report_path) as f:
        report = json.load(f)
    run_id = report.get("run_id", f"regr-{report.get('run_at_utc','')}")
    run_at = report.get("run_at_utc", _iso8601_now())
    milestone = report.get("milestone", "")
    conn = sqlite3.connect(db_path)
    for result in report.get("results", []):
        cap_id = result.get("capability_id", "")
        conn.execute("""
            INSERT OR REPLACE INTO fact_capability
            (capability_id, run_id, name, status, tier, duration_ms, detail, run_at_utc)
            VALUES (?, ?, ?, ?, ?, ?, ?, ?)
        """, (cap_id, run_id, result.get("name", ""), result.get("status", ""),
              result.get("tier", ""), result.get("duration_ms", 0),
              result.get("detail", ""), run_at))
        conn.execute("""
            INSERT OR REPLACE INTO dim_capability
            (capability_id, name, tier, source_milestone)
            VALUES (?, ?, ?, ?)
        """, (cap_id, result.get("name", ""), result.get("tier", ""), milestone))
    conn.execute("""
        INSERT OR REPLACE INTO dim_milestone
        (milestone, phase, tag, completed_at)
        VALUES (?, ?, ?, ?)
    """, (milestone, report.get("phase", 0), "", run_at))
    conn.commit()
    conn.close()
    return len(report.get("results", []))


def collect_run_manifests(db_path=None, runs_dir=None):
    """Read per-run manifests from metrics/runs/*.json → fact_run."""
    if db_path is None:
        db_path = _STORE_PATH
    if runs_dir is None:
        runs_dir = _RUNS_DIR
    if not os.path.isdir(runs_dir):
        return 0
    _init_store(db_path)
    count = 0
    conn = sqlite3.connect(db_path)
    for fname in sorted(os.listdir(runs_dir)):
        if not fname.endswith(".json"):
            continue
        fpath = os.path.join(runs_dir, fname)
        if os.path.isdir(fpath):
            continue
        with open(fpath) as f:
            manifest = json.load(f)
        run_id = manifest.get("run_id", fname.replace(".json", ""))
        conf = manifest.get("confidence", {})
        hitl = manifest.get("hitl", {})
        conn.execute("""
            INSERT OR REPLACE INTO fact_run
            (run_id, contract_id, environment, started_at, completed_at,
             exit_code, outcome, confidence_score, confidence_band,
             hitl_block, cost_estimate_usd, decision_id, escalation_reason)
            VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
        """, (run_id, manifest.get("contract_id", ""), manifest.get("environment", ""),
              manifest.get("started_at", ""), manifest.get("completed_at", ""),
              manifest.get("exit_code", 0), manifest.get("outcome", ""),
              conf.get("score", 0), conf.get("band", ""),
              1 if hitl.get("block") else 0,
              manifest.get("cost_estimate_usd", 0), manifest.get("decision_id", ""),
              manifest.get("escalation_reason")))
        count += 1
    conn.commit()
    conn.close()
    return count


def collect_decision_ledger(db_path=None, ledger_db=None):
    """Read the Decision Ledger SQLite → fact_decision.

    REQ-317: preserves a backfilled outcome. The ledger is append-only
    and the `nova.ai.decision.made` event always carries outcome=pending
    (it is emitted before apply). Once `outcome_backfill.backfill()` has
    transitioned the `fact_decision` row to succeeded/failed, a re-run of
    the collector must NOT clobber it back to pending. We therefore
    coalesce: if the existing row has a non-pending outcome, keep it +
    its backfilled_at; otherwise write pending (the event default).
    """
    if db_path is None:
        db_path = _STORE_PATH
    if ledger_db is None:
        ledger_db = _LEDGER_DB
    if not os.path.isfile(ledger_db):
        return 0
    _init_store(db_path)
    ledger_conn = sqlite3.connect(ledger_db)
    rows = ledger_conn.execute(
        "SELECT event_type, run_id, event_time, payload FROM decision_ledger WHERE event_type = 'nova.ai.decision.made' ORDER BY seq"
    ).fetchall()
    ledger_conn.close()
    conn = sqlite3.connect(db_path)
    count = 0
    for etype, run_id, event_time, payload_json in rows:
        payload = json.loads(payload_json)
        data = payload.get("data", {})
        decision_id = data.get("decision_id", run_id)
        # Preserve a backfilled outcome across collector re-runs (REQ-317).
        existing = conn.execute(
            "SELECT outcome, backfilled_at FROM fact_decision WHERE decision_id = ?",
            (decision_id,),
        ).fetchone()
        if existing and existing[0] and existing[0] != "pending":
            outcome = existing[0]
            backfilled_at = existing[1]
        else:
            outcome = data.get("outcome", "pending")
            backfilled_at = data.get("backfilled_at")
        conn.execute("""
            INSERT OR REPLACE INTO fact_decision
            (decision_id, run_id, chosen_action, confidence, alternatives,
             human_override, escalation_reason, outcome, backfilled_at, event_time)
            VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
        """, (decision_id, run_id, data.get("chosen_action", ""),
              data.get("confidence", 0), json.dumps(data.get("alternatives", {})),
              1 if data.get("human_override") else 0,
              data.get("escalation_reason"),
              outcome, backfilled_at, event_time))
        count += 1
    conn.commit()
    conn.close()
    return count


def collect_test_results(db_path=None, junit_path=None, coverage_path=None):
    """Read junit XML + coverage.json → fact_test."""
    if db_path is None:
        db_path = _STORE_PATH
    if junit_path is None:
        junit_path = _TEST_RESULTS_XML
    if coverage_path is None:
        coverage_path = _COVERAGE_JSON
    if not os.path.isfile(junit_path):
        return 0
    _init_store(db_path)
    run_id = f"test-{_iso8601_now()}"
    total = passed = failed = errors = skipped = 0
    duration = 0.0
    try:
        tree = ET.parse(junit_path)
        root = tree.getroot()
        for suite in root.iter("testsuite"):
            total += int(suite.get("tests", 0))
            failed += int(suite.get("failures", 0))
            errors += int(suite.get("errors", 0))
            skipped += int(suite.get("skipped", 0))
            duration += float(suite.get("time", 0))
        passed = total - failed - errors - skipped
    except Exception:
        pass

    coverage_pct = 0.0
    if os.path.isfile(coverage_path):
        try:
            with open(coverage_path) as f:
                cov = json.load(f)
            coverage_pct = cov.get("totals", {}).get("percent_covered", 0.0)
        except Exception:
            pass

    conn = sqlite3.connect(db_path)
    conn.execute("""
        INSERT OR REPLACE INTO fact_test
        (run_id, total_tests, passed, failed, errors, skipped, duration_s, coverage_pct, collected_at)
        VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)
    """, (run_id, total, passed, failed, errors, skipped, duration, coverage_pct, _iso8601_now()))
    conn.commit()
    conn.close()
    return 1


def collect_lifecycle_reports(db_path=None, lifecycle_dir=None):
    """Read metrics/lifecycle/*.json → fact_lifecycle."""
    if db_path is None:
        db_path = _STORE_PATH
    if lifecycle_dir is None:
        lifecycle_dir = os.path.join(_METRICS_DIR, "lifecycle")
    if not os.path.isdir(lifecycle_dir):
        return 0
    _init_store(db_path)
    count = 0
    conn = sqlite3.connect(db_path)
    for fname in sorted(os.listdir(lifecycle_dir)):
        if not fname.endswith(".json"):
            continue
        fpath = os.path.join(lifecycle_dir, fname)
        with open(fpath) as f:
            report = json.load(f)
        conn.execute("""
            INSERT OR REPLACE INTO fact_lifecycle
            (module, environment, phase, result, duration_ms, run_at)
            VALUES (?, ?, ?, ?, ?, ?)
        """, (report.get("module", ""), report.get("environment", ""),
              report.get("phase", ""), report.get("result", ""),
              report.get("duration_ms", 0), report.get("run_at", _iso8601_now())))
        count += 1
    conn.commit()
    conn.close()
    return count


def collect_all(db_path=None):
    """Run all collectors. Returns a summary dict."""
    if db_path is None:
        db_path = _STORE_PATH
    _init_store(db_path)
    summary = {
        "capabilities": collect_regression_report(db_path),
        "runs": collect_run_manifests(db_path),
        "decisions": collect_decision_ledger(db_path),
        "tests": collect_test_results(db_path),
        "lifecycle": collect_lifecycle_reports(db_path),
        "collected_at": _iso8601_now(),
    }
    return summary


if __name__ == "__main__":
    result = collect_all()
    print(json.dumps(result, indent=2))