feat(P03 W2): outcome backfill (REQ-317) + escalation_reason (REQ-318)
REQ-317: core/metrics/outcome_backfill.py backfills fact_decision.outcome pending -> succeeded/failed after run.completed/run.failed; idempotent + terminal (does not overwrite a non-pending outcome); wired into the collector. The Post-Pilot AI Decision Accuracy denominator is now grounded (fact_decision.outcome is not stuck pending). REQ-318: ai.decision.made on a block band carries escalation_reason: 'confidence' (the only value in v1.26 — a block is always confidence- driven; future milestones may add 'policy'). Persisted into fact_run by the collector. The Post-Pilot Human Escalation Frequency denominator is now grounded. ---ci--- project: acdl phase: 3 milestone: v1.26 status: execute wave: W2 ---
This commit is contained in:
Jon Chery
+33
-10
@@ -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")
|
||||
|
||||
@@ -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()
|
||||
|
||||
@@ -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)
|
||||
|
||||
@@ -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": <current>}`. 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 <decision_id> <succeeded|failed>", 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)
|
||||
@@ -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
|
||||
|
||||
|
||||
|
||||
@@ -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
|
||||
@@ -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
|
||||
Reference in New Issue
Block a user