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

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

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

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)

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

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