diff --git a/core/confidence_signal.py b/core/confidence_signal.py index 3f8821e..13dc775 100644 --- a/core/confidence_signal.py +++ b/core/confidence_signal.py @@ -144,6 +144,7 @@ def compute(contract_id: str, environment: str, penalty = 0.0 policy_input = inputs.get("policy") pcrs = policy_input if isinstance(policy_input, list) else [] + critical_override = False for pcr in pcrs: if not isinstance(pcr, dict): continue @@ -152,20 +153,31 @@ def compute(contract_id: str, environment: str, sev = pcr.get("severity") p = PENALTY.get(sev, 0.0) if p is None: - return Signal(0.0, "block", per_input, - reasons + [f"CRITICAL_OVERRIDE:{pcr.get('ruleId','?')}"]) + # Critical PCR hard override: score = 0, band = block. + # Do NOT early-return — fall through to the event emission + # block below so the SPEC §5.8 evidence stream + # (confidence.computed -> ai.decision.made -> ...) is complete + # even on a critical override (REQ-318: a critical PCR is a + # confidence-driven escalation and must carry escalation_reason). + reasons.append(f"CRITICAL_OVERRIDE:{pcr.get('ruleId','?')}") + critical_override = True + break penalty += p - score = max(0.0, min(1.0, base - penalty)) - threshold = THRESHOLDS[environment] - if score >= threshold: - band = "pass" - elif score < threshold - 0.10: + if critical_override: + score = 0.0 band = "block" else: - band = "warn" - if environment == "dev" and band == "warn": - band = "block" + score = max(0.0, min(1.0, base - penalty)) + threshold = THRESHOLDS[environment] + if score >= threshold: + band = "pass" + elif score < threshold - 0.10: + band = "block" + else: + band = "warn" + if environment == "dev" and band == "warn": + band = "block" signal = Signal(score, band, per_input, reasons) # Emit nova.confidence.computed + nova.ai.decision.made events (D-122). @@ -184,6 +196,17 @@ def compute(contract_id: str, environment: str, "human_override": band == "block", "threshold": THRESHOLDS[environment], } + # REQ-318 (SPEC §5.8): on a `block` band, carry escalation_reason. + # In v1.26 the only value is "confidence" — a block is always + # confidence-driven (the score fell below threshold OR a critical + # PCR fired a hard override). Future milestones may add "policy" + # (a critical PCR that is not confidence-scored); leave the door + # open but only emit "confidence" now. On pass/warn bands the + # field is ABSENT (escalation_reason is only meaningful on a + # block — it is the Post-Pilot Human Escalation Frequency + # denominator). + if band == "block": + decision_data["escalation_reason"] = "confidence" decision_event = make_event("nova.ai.decision.made", run_id, environment, decision_data, contract_id=contract_id, actor_type="confidence-gate", actor_id="confidence_signal") diff --git a/core/metrics/collector.py b/core/metrics/collector.py index 5488d21..bac9a1d 100644 --- a/core/metrics/collector.py +++ b/core/metrics/collector.py @@ -54,7 +54,8 @@ def _init_store(db_path=None): confidence_band TEXT, hitl_block INTEGER, cost_estimate_usd REAL, - decision_id TEXT + decision_id TEXT, + escalation_reason TEXT ); CREATE TABLE IF NOT EXISTS fact_capability ( @@ -110,7 +111,9 @@ def _init_store(db_path=None): confidence REAL, alternatives TEXT, human_override INTEGER, + escalation_reason TEXT, outcome TEXT, + backfilled_at TEXT, event_time TEXT, PRIMARY KEY (decision_id) ); @@ -217,14 +220,15 @@ def collect_run_manifests(db_path=None, runs_dir=None): 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 (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) + 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("cost_estimate_usd", 0), manifest.get("decision_id", ""), + manifest.get("escalation_reason"))) count += 1 conn.commit() conn.close() @@ -232,7 +236,16 @@ def collect_run_manifests(db_path=None, runs_dir=None): def collect_decision_ledger(db_path=None, ledger_db=None): - """Read the Decision Ledger SQLite → fact_decision.""" + """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: @@ -251,15 +264,27 @@ def collect_decision_ledger(db_path=None, ledger_db=None): 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, outcome, event_time) - VALUES (?, ?, ?, ?, ?, ?, ?, ?) + 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("outcome", "pending"), event_time)) + data.get("escalation_reason"), + outcome, backfilled_at, event_time)) count += 1 conn.commit() conn.close() diff --git a/core/metrics/decision_ledger.py b/core/metrics/decision_ledger.py index 1c66dba..8a3b5de 100644 --- a/core/metrics/decision_ledger.py +++ b/core/metrics/decision_ledger.py @@ -224,12 +224,16 @@ def replay_run(run_id, db_path=None): line = f" [{e['seq']}] {e['event_time']} {etype}" if etype == "nova.ai.decision.made": line += f" confidence={data.get('confidence', '?')} band={data.get('chosen_action', '?')} override={data.get('human_override', '?')}" + if data.get("escalation_reason"): + line += f" escalation_reason={data.get('escalation_reason')}" elif etype == "nova.attestation.recorded": line += f" env={data.get('environment', '?')} approver={data.get('approver', '?')} result={data.get('result', '?')}" elif etype == "nova.run.completed": line += f" exit={data.get('exit_code', '?')} outcome={data.get('outcome', '?')}" elif etype == "nova.run.failed": line += f" exit={data.get('exit_code', '?')} outcome=failed" + elif etype == "nova.outcome.backfilled": + line += f" prev={data.get('previous_outcome', '?')} new={data.get('new_outcome', '?')} at={data.get('backfilled_at', '?')}" lines.append(line) lines.append("=== End replay ===") return "\n".join(lines) diff --git a/core/metrics/outcome_backfill.py b/core/metrics/outcome_backfill.py new file mode 100644 index 0000000..06d3a37 --- /dev/null +++ b/core/metrics/outcome_backfill.py @@ -0,0 +1,213 @@ +"""Nova Outcome Backfill (REQ-317, SPEC §5.8, P3 Wave 2). + +The `fact_decision.outcome` column in the metrics cold store is written +`pending` by the collector (it ingests `nova.ai.decision.made` events, +which are emitted *before* the run executes the apply). Once the run +completes (`nova.run.completed`, exit 0) or fails (`nova.run.failed`, +exit non-zero), the outcome must be transitioned `pending -> +succeeded`/`failed` so the Post-Pilot AI Decision Accuracy denominator is +grounded (an outcome that is stuck `pending` cannot be scored). + +Architecture (grounded in what the ledger + collector actually do): + + * The Decision Ledger (`core/metrics/decision_ledger.py`) is an + **append-only hash-chain** of CloudEvents envelopes — there is no + `fact_decision` table *inside* the ledger DB; facts live in the + separate collector cold store (`core/metrics/collector.py`, + `nova_metrics.db`). The ledger is never UPDATEd in place (that would + break the SHA-256 chain — see `verify_chain()`). + * Therefore the backfill does TWO things: + 1. Appends a new audit event `nova.outcome.backfilled` to the + ledger (preserves the hash chain; auditable via `replay_run`). + 2. UPDATEs the `fact_decision` row in the cold store (the row is + keyed by `decision_id`; `outcome` + `backfilled_at` are + mutable — they are facts, not chain events). + +Idempotent + terminal: + * If `outcome` is already `succeeded`/`failed` (i.e. not `pending`), + the call is a no-op and returns `{"status": "already_backfilled", + "existing_outcome": }`. A terminal outcome is NEVER + overwritten (defense against double-backfill and against flipping a + `succeeded` run to `failed` retroactively or vice versa). + * The same `outcome` value is re-asserted harmlessly (still a no-op). + +REQ-317: `outcome` ∈ {"succeeded", "failed"} only — `pending` is the +initial state and may not be written by the backfill (it would undo the +transition). An invalid value raises `ValueError`. + +Future milestones may add `'policy'` to `escalation_reason` (REQ-318); +this module is scoped to outcome only. +""" + +import datetime +import json +import os +import sqlite3 +import sys +from pathlib import Path +from typing import Optional, Dict, Any + +sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) +from core.metrics.event_envelope import make_event, append_event +from core.metrics.decision_ledger import append as ledger_append, _LEDGER_PATH + +# The collector cold store path is mirrored here so the backfill can be +# invoked without importing the collector (avoids a circular import: +# the collector calls into backfill at run.completed/run.failed time). +_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") + +_VALID_OUTCOMES = {"succeeded", "failed"} +_PENDING = "pending" + + +def _iso8601_now(): + return datetime.datetime.now(datetime.timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ") + + +def _resolve_store_path(store_path: Optional[str | Path]) -> str: + if store_path is None: + return _STORE_PATH + return str(store_path) + + +def _resolve_ledger_path(ledger_path: Optional[str | Path]) -> str: + if ledger_path is None: + return _LEDGER_PATH + return str(ledger_path) + + +def _get_fact_decision(decision_id: str, store_path: str) -> Optional[Dict[str, Any]]: + """Read the fact_decision row for decision_id (or None).""" + if not os.path.isfile(store_path): + return None + conn = sqlite3.connect(store_path) + conn.row_factory = sqlite3.Row + row = conn.execute( + "SELECT decision_id, run_id, chosen_action, confidence, alternatives, " + "human_override, outcome, event_time FROM fact_decision WHERE decision_id = ?", + (decision_id,), + ).fetchone() + conn.close() + if row is None: + return None + return dict(row) + + +def backfill( + decision_id: str, + outcome: str, + ledger_path: Optional[str | Path] = None, + store_path: Optional[str | Path] = None, +) -> Dict[str, Any]: + """Transition fact_decision.outcome from `pending` to `outcome`. + + Args: + decision_id: the decision id (== run_id for v1.26). + outcome: the terminal outcome; must be in {"succeeded", "failed"}. + ledger_path: optional override for the Decision Ledger SQLite DB. + store_path: optional override for the collector cold store SQLite DB. + + Returns: + A dict describing the result: + * success: {"status": "backfilled", "decision_id", "previous_outcome", + "new_outcome", "backfilled_at"} + * no-op: {"status": "already_backfilled", "decision_id", + "existing_outcome", "backfilled_at"} + + Raises: + ValueError: if `outcome` is not in {"succeeded", "failed"}. + KeyError: if `decision_id` is not present in fact_decision. + """ + if outcome not in _VALID_OUTCOMES: + raise ValueError( + f"outcome must be one of {sorted(_VALID_OUTCOMES)}, got: {outcome!r}" + ) + + sp = _resolve_store_path(store_path) + lp = _resolve_ledger_path(ledger_path) + + existing = _get_fact_decision(decision_id, sp) + if existing is None: + raise KeyError(decision_id) + + current_outcome = existing.get("outcome") or _PENDING + backfilled_at = _iso8601_now() + + if current_outcome != _PENDING: + # Idempotent + terminal: do NOT overwrite a non-pending outcome. + return { + "status": "already_backfilled", + "decision_id": decision_id, + "existing_outcome": current_outcome, + "backfilled_at": backfilled_at, + } + + run_id = existing.get("run_id") or decision_id + + # 1. UPDATE the fact_decision row in the cold store (mutable fact). + conn = sqlite3.connect(sp) + # Add backfilled_at column idempotently (schema was added in v1.26 P3 W2; + # older cold stores created by P2 lack it — ALTER TABLE is a no-op if + # the column already exists). + try: + conn.execute("ALTER TABLE fact_decision ADD COLUMN backfilled_at TEXT") + except sqlite3.OperationalError: + pass # column already exists + conn.execute( + "UPDATE fact_decision SET outcome = ?, backfilled_at = ? WHERE decision_id = ?", + (outcome, backfilled_at, decision_id), + ) + conn.commit() + conn.close() + + # 2. Append an audit event to the append-only Decision Ledger (preserves + # the hash chain — the ledger is never UPDATEd in place). + try: + backfill_data = { + "decision_id": decision_id, + "previous_outcome": _PENDING, + "new_outcome": outcome, + "backfilled_at": backfilled_at, + } + event = make_event( + "nova.outcome.backfilled", + run_id, + existing.get("environment", ""), + backfill_data, + contract_id=existing.get("contract_id", ""), + actor_type="outcome-backfill", + actor_id="outcome_backfill", + ) + append_event(event) + ledger_append(event, db_path=lp) + except Exception: + # Metrics emission must never break the backfill — the cold store + # UPDATE is the source of truth for the denominator; the ledger + # event is audit chrome. + pass + + return { + "status": "backfilled", + "decision_id": decision_id, + "previous_outcome": _PENDING, + "new_outcome": outcome, + "backfilled_at": backfilled_at, + } + + +if __name__ == "__main__": + if len(sys.argv) < 3: + print("usage: outcome_backfill.py ", file=sys.stderr) + sys.exit(2) + _did = sys.argv[1] + _out = sys.argv[2] + try: + _r = backfill(_did, _out) + print(json.dumps(_r, indent=2)) + except (ValueError, KeyError) as exc: + print(f"error: {exc}", file=sys.stderr) + sys.exit(1) \ No newline at end of file diff --git a/core/metrics/run_manifest.py b/core/metrics/run_manifest.py index 29e9897..61e9237 100644 --- a/core/metrics/run_manifest.py +++ b/core/metrics/run_manifest.py @@ -27,6 +27,27 @@ def _iso8601_now(): return datetime.datetime.now(datetime.timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ") +def _backfill_outcome(decision_id, outcome): + """Transition fact_decision.outcome pending -> outcome (REQ-317). + + Best-effort: logs a warning and skips if decision_id is missing or the + backfill raises. Never raises — the run is already completing/failing + and the manifest write is the source of truth for the run outcome. + """ + if not decision_id: + # A run that failed before ai.decision.made was emitted has no + # decision to backfill (e.g. a schema-validation failure). Skip + # silently rather than pollute stderr on every clean run. + return None + try: + from core.metrics import outcome_backfill + return outcome_backfill.backfill(decision_id, outcome) + except Exception as exc: # pragma: no cover - defensive + print(f"[run_manifest] outcome backfill skipped for {decision_id}: {exc}", + file=sys.stderr) + return None + + def _run_id(): return f"run-{int(time.time())}-{uuid.uuid4().hex[:8]}" @@ -44,7 +65,7 @@ def start_run(contract_id, environment, stages=None): return run_id -def complete_run(run_id, contract_id, environment, stages, exit_code, confidence=None, hitl=None, policy=None, cost_estimate_usd=None, decision_id=None): +def complete_run(run_id, contract_id, environment, stages, exit_code, confidence=None, hitl=None, policy=None, cost_estimate_usd=None, decision_id=None, escalation_reason=None): """Emit nova.run.completed + write the per-run manifest JSON. Args: @@ -58,6 +79,10 @@ def complete_run(run_id, contract_id, environment, stages, exit_code, confidence policy: optional {passed, failed, skipped} cost_estimate_usd: optional float decision_id: optional string (links to the Decision Ledger) + escalation_reason: optional string (REQ-318) — "confidence" when + the ai.decision.made band was block; absent/None otherwise. + Persisted into the manifest so the collector can write it + into fact_run (Post-Pilot Human Escalation Frequency denom). """ started_at = stages[0].get("started_at", _iso8601_now()) if stages else _iso8601_now() completed_at = _iso8601_now() @@ -83,6 +108,8 @@ def complete_run(run_id, contract_id, environment, stages, exit_code, confidence manifest["cost_estimate_usd"] = cost_estimate_usd if decision_id: manifest["decision_id"] = decision_id + if escalation_reason: + manifest["escalation_reason"] = escalation_reason os.makedirs(_RUNS_DIR, exist_ok=True) manifest_path = os.path.join(_RUNS_DIR, f"{run_id}.json") @@ -92,6 +119,14 @@ def complete_run(run_id, contract_id, environment, stages, exit_code, confidence event_type = "nova.run.completed" if exit_code == 0 else "nova.run.failed" emit(event_type, run_id, environment, manifest, contract_id=contract_id) + # REQ-317: backfill fact_decision.outcome pending -> succeeded/failed + # after the run completes. The decision_id links the run to the + # Decision Ledger entry written by ai.decision.made. Best-effort: a + # run that failed before ai.decision.made was emitted has no + # decision_id and the backfill is a no-op (the run outcome is still + # captured in the manifest above). + backfill_result = _backfill_outcome(decision_id, outcome) + return manifest diff --git a/tests/test_confidence_escalation_reason.py b/tests/test_confidence_escalation_reason.py new file mode 100644 index 0000000..e556d64 --- /dev/null +++ b/tests/test_confidence_escalation_reason.py @@ -0,0 +1,211 @@ +"""Tests for escalation_reason on ai.decision.made (REQ-318, SPEC §5.8, P3 W2). + +Covers: + * block band carries escalation_reason == "confidence" + human_override True + * pass band has escalation_reason ABSENT + human_override False + * fact_run persists escalation_reason (collector wiring) + +Follows the fixture pattern in tests/test_metrics_emitters.py. +""" + +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_metrics(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() + events_log = metrics_dir / "events.jsonl" + ledger_db = metrics_dir / "decision_ledger.db" + store_db = metrics_dir / "nova_metrics.db" + + monkeypatch.setattr("core.metrics.event_envelope.METRICS_DIR", str(metrics_dir)) + monkeypatch.setattr("core.metrics.event_envelope.EVENTS_LOG", str(events_log)) + monkeypatch.setattr("core.metrics.run_manifest._METRICS_DIR", str(metrics_dir)) + monkeypatch.setattr("core.metrics.run_manifest._RUNS_DIR", str(runs_dir)) + monkeypatch.setattr("core.metrics.decision_ledger._LEDGER_PATH", str(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.outcome_backfill._METRICS_DIR", str(metrics_dir)) + monkeypatch.setattr("core.metrics.outcome_backfill._STORE_PATH", str(store_db)) + monkeypatch.setattr("core.metrics.outcome_backfill._LEDGER_PATH", str(ledger_db)) + return { + "metrics_dir": metrics_dir, + "events_log": events_log, + "ledger_db": ledger_db, + "store_db": store_db, + "runs_dir": runs_dir, + } + + +def _base_inputs(): + return { + "policy": [{"result": "pass", "severity": "info"}], + "validation": {"schema": True, "stack_resolved": True, + "tf_validated": True, "tf_planned": True}, + "freshness": {"age_days": 0, "max_age_days": 7}, + "source": {"submitter": "dev", "commit_sha": "abc"}, + "history": {"prior_rollbacks": 0, "prior_policy_fails": 0}, + "nfrs": {"conformance": 1.0}, + } + + +def _block_inputs(): + """A critical PCR triggers a hard override (score=0, band=block).""" + inputs = _base_inputs() + inputs["policy"] = [{"result": "fail", "severity": "critical", "ruleId": "CKV_X"}] + return inputs + + +def _read_decision_event(events_log): + lines = events_log.read_text().strip().split("\n") + for line in lines: + ev = json.loads(line) + if ev["type"] == "nova.ai.decision.made": + return ev + return None + + +def test_block_band_has_escalation_reason(tmp_metrics): + """A block (critical PCR hard override) carries escalation_reason='confidence'.""" + from core.confidence_signal import compute + sig = compute("cid-block-1", "dev", _block_inputs()) + assert sig.band == "block" + ev = _read_decision_event(tmp_metrics["events_log"]) + assert ev is not None + data = ev["data"] + assert data["chosen_action"] == "block" + assert data["human_override"] is True + assert data.get("escalation_reason") == "confidence" + + +def test_pass_band_no_escalation_reason(tmp_metrics): + """A clean dev apply (pass band) has NO escalation_reason + human_override False.""" + from core.confidence_signal import compute + sig = compute("cid-pass-1", "dev", _base_inputs()) + assert sig.band == "pass" + ev = _read_decision_event(tmp_metrics["events_log"]) + assert ev is not None + data = ev["data"] + assert data["chosen_action"] == "pass" + assert data["human_override"] is False + # escalation_reason must be ABSENT on a non-block band. + assert "escalation_reason" not in data + + +def test_low_confidence_block_has_escalation_reason(tmp_metrics): + """A score below (threshold - 0.10) blocks on confidence grounds.""" + from core.confidence_signal import compute + # Freshness maximally stale + a high-severity policy fail drags the + # score well below the dev threshold of 0.50 - 0.10 = 0.40. + inputs = _base_inputs() + inputs["freshness"] = {"age_days": 7, "max_age_days": 7} + inputs["policy"] = [{"result": "fail", "severity": "high", "ruleId": "CKV_Y"}] + sig = compute("cid-block-2", "dev", inputs) + assert sig.band == "block" + ev = _read_decision_event(tmp_metrics["events_log"]) + data = ev["data"] + assert data.get("escalation_reason") == "confidence" + + +def test_fact_run_persists_escalation_reason(tmp_metrics): + """The collector persists escalation_reason into fact_run + fact_decision. + + End-to-end: confidence_signal emits ai.decision.made (block) → + run_manifest.complete_run writes the manifest with escalation_reason → + collector.collect_run_manifests + collect_decision_ledger populate + fact_run.escalation_reason + fact_decision.escalation_reason. + """ + from core.confidence_signal import compute + from core.metrics.run_manifest import complete_run + from core.metrics.collector import collect_run_manifests, collect_decision_ledger + + # Emit a block decision. + os.environ["NOVA_RUN_ID"] = "run-esc-1" + try: + sig = compute("cid-esc-1", "dev", _block_inputs()) + assert sig.band == "block" + finally: + os.environ.pop("NOVA_RUN_ID", None) + + # Complete the run with escalation_reason carried into the manifest. + manifest = complete_run( + "run-esc-1", "cid-esc-1", "dev", + stages=[{"name": "apply", "duration_ms": 100, "exit_code": 0}], + exit_code=0, + confidence={"score": sig.score, "band": sig.band, "perInput": sig.perInput}, + decision_id="run-esc-1", + escalation_reason="confidence", + ) + assert manifest["escalation_reason"] == "confidence" + + # Collector reads the manifest → fact_run. + collect_run_manifests() + # Collector reads the ledger → fact_decision. + collect_decision_ledger() + + conn = sqlite3.connect(str(tmp_metrics["store_db"])) + conn.row_factory = sqlite3.Row + run_row = conn.execute( + "SELECT run_id, escalation_reason, decision_id FROM fact_run WHERE run_id = ?", + ("run-esc-1",), + ).fetchone() + dec_row = conn.execute( + "SELECT decision_id, escalation_reason, human_override FROM fact_decision WHERE decision_id = ?", + ("run-esc-1",), + ).fetchone() + conn.close() + + assert run_row is not None + assert run_row["escalation_reason"] == "confidence" + assert run_row["decision_id"] == "run-esc-1" + assert dec_row is not None + assert dec_row["escalation_reason"] == "confidence" + assert dec_row["human_override"] == 1 + + +def test_fact_run_no_escalation_reason_on_pass(tmp_metrics): + """A pass-band run has escalation_reason NULL in fact_run.""" + from core.confidence_signal import compute + from core.metrics.run_manifest import complete_run + from core.metrics.collector import collect_run_manifests + + os.environ["NOVA_RUN_ID"] = "run-esc-pass-1" + try: + sig = compute("cid-esc-pass-1", "dev", _base_inputs()) + assert sig.band == "pass" + finally: + os.environ.pop("NOVA_RUN_ID", None) + + complete_run( + "run-esc-pass-1", "cid-esc-pass-1", "dev", + stages=[{"name": "apply", "duration_ms": 100, "exit_code": 0}], + exit_code=0, + confidence={"score": sig.score, "band": sig.band, "perInput": sig.perInput}, + decision_id="run-esc-pass-1", + # escalation_reason intentionally omitted (pass band). + ) + collect_run_manifests() + conn = sqlite3.connect(str(tmp_metrics["store_db"])) + conn.row_factory = sqlite3.Row + row = conn.execute( + "SELECT run_id, escalation_reason FROM fact_run WHERE run_id = ?", + ("run-esc-pass-1",), + ).fetchone() + conn.close() + assert row is not None + assert row["escalation_reason"] is None \ No newline at end of file diff --git a/tests/test_outcome_backfill.py b/tests/test_outcome_backfill.py new file mode 100644 index 0000000..21c6359 --- /dev/null +++ b/tests/test_outcome_backfill.py @@ -0,0 +1,152 @@ +"""Tests for Nova Outcome Backfill (REQ-317, SPEC §5.8, P3 W2). + +Covers the fact_decision.outcome transition pending -> succeeded/failed: + * happy path (succeeded, failed) + * idempotency (already_backfilled is a no-op) + * terminal defense (does NOT flip succeeded -> failed) + * invalid outcome raises ValueError + * unknown decision_id raises KeyError + +Uses a tmp SQLite cold store + Decision Ledger (does NOT touch the real +metrics/decision_ledger.db). Follows the fixture pattern in +tests/test_metrics_collector.py + tests/test_metrics_emitters.py. +""" + +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_backfill_env(tmp_path, monkeypatch): + """Redirect metrics/ to a tmp dir + seed a fact_decision row (pending).""" + metrics_dir = tmp_path / "metrics" + metrics_dir.mkdir() + store_db = metrics_dir / "nova_metrics.db" + ledger_db = metrics_dir / "decision_ledger.db" + events_log = metrics_dir / "events.jsonl" + + monkeypatch.setattr("core.metrics.event_envelope.METRICS_DIR", str(metrics_dir)) + monkeypatch.setattr("core.metrics.event_envelope.EVENTS_LOG", str(events_log)) + monkeypatch.setattr("core.metrics.decision_ledger._LEDGER_PATH", str(ledger_db)) + monkeypatch.setattr("core.metrics.outcome_backfill._METRICS_DIR", str(metrics_dir)) + monkeypatch.setattr("core.metrics.outcome_backfill._STORE_PATH", str(store_db)) + monkeypatch.setattr("core.metrics.outcome_backfill._LEDGER_PATH", str(ledger_db)) + + # Initialize the cold store schema + a pending fact_decision row. + from core.metrics.collector import _init_store + _init_store(str(store_db)) + + conn = sqlite3.connect(str(store_db)) + conn.execute( + "INSERT INTO fact_decision " + "(decision_id, run_id, chosen_action, confidence, alternatives, " + "human_override, escalation_reason, outcome, backfilled_at, event_time) " + "VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?)", + ("dec-1", "run-1", "block", 0.42, "{}", 1, "confidence", + "pending", None, "2026-08-18T00:00:00Z"), + ) + conn.commit() + conn.close() + + return { + "metrics_dir": metrics_dir, + "store_db": store_db, + "ledger_db": ledger_db, + "events_log": events_log, + "decision_id": "dec-1", + } + + +def _get_fact_decision(store_db, decision_id): + conn = sqlite3.connect(str(store_db)) + conn.row_factory = sqlite3.Row + row = conn.execute( + "SELECT decision_id, outcome, backfilled_at FROM fact_decision WHERE decision_id = ?", + (decision_id,), + ).fetchone() + conn.close() + return dict(row) if row else None + + +def test_backfill_succeeded(tmp_backfill_env): + from core.metrics.outcome_backfill import backfill + result = backfill(tmp_backfill_env["decision_id"], "succeeded") + assert result["status"] == "backfilled" + assert result["previous_outcome"] == "pending" + assert result["new_outcome"] == "succeeded" + assert result["backfilled_at"] + fact = _get_fact_decision(tmp_backfill_env["store_db"], "dec-1") + assert fact["outcome"] == "succeeded" + assert fact["backfilled_at"] == result["backfilled_at"] + + +def test_backfill_failed(tmp_backfill_env): + from core.metrics.outcome_backfill import backfill + result = backfill(tmp_backfill_env["decision_id"], "failed") + assert result["status"] == "backfilled" + assert result["new_outcome"] == "failed" + fact = _get_fact_decision(tmp_backfill_env["store_db"], "dec-1") + assert fact["outcome"] == "failed" + + +def test_backfill_idempotent(tmp_backfill_env): + """Already succeeded → backfill('succeeded') again → no-op.""" + from core.metrics.outcome_backfill import backfill + backfill(tmp_backfill_env["decision_id"], "succeeded") + result = backfill(tmp_backfill_env["decision_id"], "succeeded") + assert result["status"] == "already_backfilled" + assert result["existing_outcome"] == "succeeded" + fact = _get_fact_decision(tmp_backfill_env["store_db"], "dec-1") + assert fact["outcome"] == "succeeded" + + +def test_backfill_does_not_overwrite(tmp_backfill_env): + """Already succeeded → backfill('failed') → must NOT flip to failed. + + A terminal outcome is never overwritten (defense against double-backfill + and against retroactively flipping succeeded -> failed). + """ + from core.metrics.outcome_backfill import backfill + backfill(tmp_backfill_env["decision_id"], "succeeded") + result = backfill(tmp_backfill_env["decision_id"], "failed") + assert result["status"] == "already_backfilled" + assert result["existing_outcome"] == "succeeded" + fact = _get_fact_decision(tmp_backfill_env["store_db"], "dec-1") + assert fact["outcome"] == "succeeded" + + +def test_backfill_invalid_outcome(tmp_backfill_env): + from core.metrics.outcome_backfill import backfill + with pytest.raises(ValueError): + backfill(tmp_backfill_env["decision_id"], "pending") + with pytest.raises(ValueError): + backfill(tmp_backfill_env["decision_id"], "garbage") + + +def test_backfill_unknown_decision(tmp_backfill_env): + from core.metrics.outcome_backfill import backfill + with pytest.raises(KeyError): + backfill("nonexistent-decision-id", "succeeded") + + +def test_backfill_appends_ledger_event(tmp_backfill_env): + """The backfill appends nova.outcome.backfilled to the Decision Ledger + (preserves the hash chain — the ledger is never UPDATEd in place).""" + from core.metrics.outcome_backfill import backfill + from core.metrics.decision_ledger import query_by_run, verify_chain + backfill(tmp_backfill_env["decision_id"], "succeeded") + entries = query_by_run("run-1", db_path=str(tmp_backfill_env["ledger_db"])) + types = [e["event_type"] for e in entries] + assert "nova.outcome.backfilled" in types + # Hash chain still intact. + ok, broken, _ = verify_chain(db_path=str(tmp_backfill_env["ledger_db"])) + assert ok, f"chain broken after backfill: {broken}" + assert broken == 0 \ No newline at end of file