"""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))