diff --git a/core/metrics/collector.py b/core/metrics/collector.py new file mode 100644 index 0000000..5488d21 --- /dev/null +++ b/core/metrics/collector.py @@ -0,0 +1,364 @@ +"""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 + ); + + 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, + outcome 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) + 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", ""))) + 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.""" + 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) + conn.execute(""" + INSERT OR REPLACE INTO fact_decision + (decision_id, run_id, chosen_action, confidence, alternatives, + human_override, outcome, 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("outcome", "pending"), 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)) \ No newline at end of file diff --git a/core/metrics/decision_ledger_cli.py b/core/metrics/decision_ledger_cli.py new file mode 100644 index 0000000..5d5b95c --- /dev/null +++ b/core/metrics/decision_ledger_cli.py @@ -0,0 +1,39 @@ +"""Nova Decision Ledger CLI (REQ-207). + +Subcommands: query, verify-chain, stats, export, replay. +Read-only CLI for the Decision Ledger SQLite hash-chain. +""" + +import json +import os +import sys + +sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.dirname(os.path.abspath(__file__))))) +from core.metrics.decision_ledger import query_by_run, verify_chain, stats, export_since, replay_run + + +def main(): + if len(sys.argv) < 2: + print("usage: decision_ledger_cli.py [args]", file=sys.stderr) + sys.exit(2) + cmd = sys.argv[1] + if cmd == "query" and len(sys.argv) >= 3: + print(json.dumps(query_by_run(sys.argv[2]), indent=2)) + elif cmd == "verify-chain": + ok, broken, details = verify_chain() + print(f"chain_ok={ok} broken={broken} details={details}") + sys.exit(0 if ok else 1) + elif cmd == "stats": + print(json.dumps(stats(), indent=2)) + elif cmd == "export" and len(sys.argv) >= 3: + fmt = sys.argv[3] if len(sys.argv) >= 4 else "json" + print(export_since(sys.argv[2], fmt=fmt)) + elif cmd == "replay" and len(sys.argv) >= 3: + print(replay_run(sys.argv[2])) + else: + print(f"unknown command: {cmd}", file=sys.stderr) + sys.exit(2) + + +if __name__ == "__main__": + main() \ No newline at end of file diff --git a/tests/test_metrics_collector.py b/tests/test_metrics_collector.py new file mode 100644 index 0000000..821ddc6 --- /dev/null +++ b/tests/test_metrics_collector.py @@ -0,0 +1,200 @@ +"""Tests for Nova metrics collector (P2, REQ-189/200). + +Tests the collector's idempotent re-run property (REQ-200) and the +SQLite cold store schema. +""" + +import json +import os +import sqlite3 +import sys +from pathlib import Path + +import pytest + +ROOT = Path(__file__).resolve().parent.parent +sys.path.insert(0, str(ROOT)) + + +@pytest.fixture +def tmp_store(tmp_path, monkeypatch): + """Redirect metrics/ to a tmp dir for isolated testing.""" + metrics_dir = tmp_path / "metrics" + metrics_dir.mkdir() + runs_dir = metrics_dir / "runs" + runs_dir.mkdir() + lifecycle_dir = metrics_dir / "lifecycle" + lifecycle_dir.mkdir() + store_db = metrics_dir / "nova_metrics.db" + ledger_db = metrics_dir / "decision_ledger.db" + + monkeypatch.setattr("core.metrics.collector._METRICS_DIR", str(metrics_dir)) + monkeypatch.setattr("core.metrics.collector._STORE_PATH", str(store_db)) + monkeypatch.setattr("core.metrics.collector._RUNS_DIR", str(runs_dir)) + monkeypatch.setattr("core.metrics.collector._LEDGER_DB", str(ledger_db)) + monkeypatch.setattr("core.metrics.collector._REPO_ROOT", str(tmp_path)) + monkeypatch.setattr("core.metrics.collector._REGRESSION_REPORT", str(tmp_path / "REGRESSION_REPORT.json")) + monkeypatch.setattr("core.metrics.collector._COVERAGE_JSON", str(metrics_dir / "coverage.json")) + monkeypatch.setattr("core.metrics.collector._TEST_RESULTS_XML", str(metrics_dir / "test-results.xml")) + monkeypatch.setattr("core.metrics.decision_ledger._LEDGER_PATH", str(ledger_db)) + return {"metrics_dir": metrics_dir, "store_db": store_db, "ledger_db": ledger_db, "runs_dir": runs_dir} + + +def _write_regression_report(path, run_id="regr-test-1"): + report = { + "run_id": run_id, + "run_at_utc": "2026-08-04T12:00:00Z", + "milestone": "v1.17", + "phase": 0, + "summary": {"Verified": 18, "Decayed": 0, "Broken": 0, "Skipped": 4}, + "passed": True, + "results": [ + {"capability_id": "CAP-001", "name": "test cap", "status": "Verified", "tier": "local", "duration_ms": 100, "detail": "ok"}, + {"capability_id": "CAP-002", "name": "test cap 2", "status": "Skipped", "tier": "live-aws", "duration_ms": 50, "detail": "D-096"}, + ], + } + with open(path, "w") as f: + json.dump(report, f) + + +def _write_run_manifest(runs_dir, run_id="run-test-1"): + manifest = { + "run_id": run_id, + "contract_id": "cid-1", + "environment": "dev", + "started_at": "2026-08-04T12:00:00Z", + "completed_at": "2026-08-04T12:01:00Z", + "exit_code": 0, + "stages": [{"name": "resolve", "duration_ms": 100, "exit_code": 0}], + "outcome": "succeeded", + "confidence": {"score": 0.9, "band": "pass", "perInput": {"policy": 1.0}}, + "hitl": {"gate": "dev", "result": "autonomous", "block": False}, + "cost_estimate_usd": -12.5, + "decision_id": run_id, + } + with open(runs_dir / f"{run_id}.json", "w") as f: + json.dump(manifest, f) + + +def _write_junit(path): + xml = """ + + + + +""" + path.write_text(xml) + + +def _write_coverage(path): + with open(path, "w") as f: + json.dump({"totals": {"percent_covered": 85.5}}, f) + + +def test_collector_init(tmp_store): + from core.metrics.collector import _init_store + _init_store() + assert tmp_store["store_db"].exists() + conn = sqlite3.connect(str(tmp_store["store_db"])) + tables = conn.execute("SELECT name FROM sqlite_master WHERE type='table'").fetchall() + conn.close() + table_names = [t[0] for t in tables] + assert "fact_run" in table_names + assert "fact_capability" in table_names + assert "fact_decision" in table_names + assert "dim_capability" in table_names + assert "dim_milestone" in table_names + + +def test_collector_regression_report(tmp_store): + from core.metrics.collector import collect_regression_report + _write_regression_report(tmp_store["metrics_dir"].parent / "REGRESSION_REPORT.json") + count = collect_regression_report() + assert count == 2 + conn = sqlite3.connect(str(tmp_store["store_db"])) + rows = conn.execute("SELECT capability_id, status FROM fact_capability").fetchall() + conn.close() + assert len(rows) == 2 + assert rows[0][0] == "CAP-001" + + +def test_collector_run_manifests(tmp_store): + from core.metrics.collector import collect_run_manifests + _write_run_manifest(tmp_store["runs_dir"]) + count = collect_run_manifests() + assert count == 1 + conn = sqlite3.connect(str(tmp_store["store_db"])) + row = conn.execute("SELECT run_id, confidence_score, cost_estimate_usd FROM fact_run").fetchone() + conn.close() + assert row[0] == "run-test-1" + assert row[1] == 0.9 + assert row[2] == -12.5 + + +def test_collector_idempotent(tmp_store): + """REQ-200: re-running the collector produces identical row counts.""" + from core.metrics.collector import collect_all + _write_regression_report(tmp_store["metrics_dir"].parent / "REGRESSION_REPORT.json") + _write_run_manifest(tmp_store["runs_dir"]) + _write_junit(tmp_store["metrics_dir"] / "test-results.xml") + _write_coverage(tmp_store["metrics_dir"] / "coverage.json") + + result1 = collect_all() + conn = sqlite3.connect(str(tmp_store["store_db"])) + cap_count_1 = conn.execute("SELECT COUNT(*) FROM fact_capability").fetchone()[0] + run_count_1 = conn.execute("SELECT COUNT(*) FROM fact_run").fetchone()[0] + conn.close() + + result2 = collect_all() + conn = sqlite3.connect(str(tmp_store["store_db"])) + cap_count_2 = conn.execute("SELECT COUNT(*) FROM fact_capability").fetchone()[0] + run_count_2 = conn.execute("SELECT COUNT(*) FROM fact_run").fetchone()[0] + conn.close() + + assert cap_count_1 == cap_count_2 + assert run_count_1 == run_count_2 + + +def test_collector_decision_ledger(tmp_store): + from core.metrics.event_envelope import make_event + from core.metrics.decision_ledger import append + from core.metrics.collector import collect_decision_ledger + ev = make_event("nova.ai.decision.made", "run-dl-collect-1", "dev", + {"decision_id": "run-dl-collect-1", "chosen_action": "pass", + "confidence": 0.94, "alternatives": {"policy": 1.0}, + "human_override": False, "outcome": "succeeded"}) + append(ev) + count = collect_decision_ledger() + assert count == 1 + conn = sqlite3.connect(str(tmp_store["store_db"])) + row = conn.execute("SELECT decision_id, confidence, chosen_action FROM fact_decision").fetchone() + conn.close() + assert row[0] == "run-dl-collect-1" + assert row[1] == 0.94 + assert row[2] == "pass" + + +def test_collector_test_results(tmp_store): + from core.metrics.collector import collect_test_results + _write_junit(tmp_store["metrics_dir"] / "test-results.xml") + _write_coverage(tmp_store["metrics_dir"] / "coverage.json") + count = collect_test_results() + assert count == 1 + conn = sqlite3.connect(str(tmp_store["store_db"])) + row = conn.execute("SELECT total_tests, passed, coverage_pct FROM fact_test").fetchone() + conn.close() + assert row[0] == 10 + assert row[1] == 10 + assert row[2] == 85.5 + + +def test_collector_all(tmp_store): + from core.metrics.collector import collect_all + _write_regression_report(tmp_store["metrics_dir"].parent / "REGRESSION_REPORT.json") + _write_run_manifest(tmp_store["runs_dir"]) + _write_junit(tmp_store["metrics_dir"] / "test-results.xml") + _write_coverage(tmp_store["metrics_dir"] / "coverage.json") + result = collect_all() + assert result["capabilities"] == 2 + assert result["runs"] == 1 + assert result["tests"] == 1 \ No newline at end of file