Jon Chery
51b886f3f6
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 ---
229 lines
8.3 KiB
Python
229 lines
8.3 KiB
Python
"""Nova Confidence Signal (REQ-19).
|
|
|
|
The platform's certified answer to "is this safe to proceed?" (vision
|
|
tenet: "Safety is Computed, Not Assumed"). Every delivery action produces
|
|
a measurable, explainable confidence signal; reliance on operator
|
|
instinct is not a substitute.
|
|
|
|
Inputs (weights sum to 1.0, D-040):
|
|
1. policy_results (0.30) — list[PolicyCheckResult] (schemas/policy_check_result.schema.json)
|
|
2. validation (0.25) — {schema: bool, stack_resolved: bool, tf_validated: bool, tf_planned: bool}
|
|
3. freshness (0.10) — {age_days: float, max_age_days: float}
|
|
4. source (0.15) — {submitter: str, commit_sha: str, signed: bool}
|
|
5. history (0.10) — {prior_rollbacks: int, prior_policy_fails: int}
|
|
6. nfrs (0.10) — {declared: list[str], conformance: float|None}
|
|
|
|
Severity -> penalty (locked, ARCHITECTURE.md §8):
|
|
critical -> hard override (score = 0, block)
|
|
high -> -0.20
|
|
medium -> -0.05
|
|
low -> -0.01
|
|
info -> 0.00
|
|
|
|
Per-env thresholds (locked, ARCHITECTURE.md §8): dev 0.50, qa 0.75, prod 0.90, dr 0.95.
|
|
Output: {score, band, perInput, reasonCodes}.
|
|
Halt with explicit reason on missing input (§8).
|
|
|
|
Spike cold-start (A-6.2): inputs 3 (freshness), 5 (history), 6 (nfrs) are
|
|
'present + neutral 0.5' because the spike is the first submission with no
|
|
history and no declared NFRs. The gate is *presence*, not *conformance* —
|
|
the 'all six inputs present' dev gate (§5) is satisfied by non-null
|
|
per-input scores.
|
|
"""
|
|
|
|
from dataclasses import dataclass, asdict
|
|
from typing import List, Literal, Optional, Dict, Any
|
|
import json
|
|
import os
|
|
import sys
|
|
|
|
sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
|
|
from core.metrics.event_envelope import emit, make_event, append_event
|
|
from core.metrics.decision_ledger import append as ledger_append
|
|
|
|
|
|
WEIGHTS = {
|
|
"policy": 0.30,
|
|
"validation": 0.25,
|
|
"freshness": 0.10,
|
|
"source": 0.15,
|
|
"history": 0.10,
|
|
"nfrs": 0.10,
|
|
}
|
|
|
|
PENALTY = {
|
|
"critical": None,
|
|
"high": 0.20,
|
|
"medium": 0.05,
|
|
"low": 0.01,
|
|
"info": 0.0,
|
|
}
|
|
|
|
THRESHOLDS = {"dev": 0.50, "qa": 0.75, "prod": 0.90, "dr": 0.95}
|
|
|
|
|
|
@dataclass
|
|
class Signal:
|
|
score: float
|
|
band: Literal["pass", "warn", "block"]
|
|
perInput: Dict[str, float]
|
|
reasonCodes: List[str]
|
|
|
|
|
|
def _per_input_score(name: str, raw: Any) -> tuple:
|
|
"""Return (score in [0,1], reasons list). Unknown/missing -> 0.5 + INPUT_MISSING."""
|
|
reasons: List[str] = []
|
|
if raw is None:
|
|
return 0.5, [f"INPUT_MISSING:{name}"]
|
|
if name == "policy":
|
|
pcrs = raw if isinstance(raw, list) else []
|
|
if not pcrs:
|
|
return 0.5, []
|
|
scores = []
|
|
for pcr in pcrs:
|
|
r = pcr.get("result", "skipped")
|
|
if r == "pass" or r == "skipped":
|
|
scores.append(1.0)
|
|
else:
|
|
scores.append(0.0)
|
|
return sum(scores) / len(scores), []
|
|
if name == "validation":
|
|
keys = ("schema", "stack_resolved", "tf_validated", "tf_planned")
|
|
if not isinstance(raw, dict):
|
|
return 0.5, []
|
|
trues = sum(1 for k in keys if raw.get(k))
|
|
return trues / 4.0, []
|
|
if name == "freshness":
|
|
if not isinstance(raw, dict):
|
|
return 0.5, []
|
|
age = float(raw.get("age_days", 0))
|
|
mx = float(raw.get("max_age_days", 1)) or 1
|
|
s = 1.0 - (age / mx)
|
|
return max(0.0, min(1.0, s)), []
|
|
if name == "source":
|
|
if not isinstance(raw, dict):
|
|
return 0.5, []
|
|
if raw.get("submitter") and raw.get("commit_sha"):
|
|
return 1.0, []
|
|
return 0.5, []
|
|
if name == "history":
|
|
if not isinstance(raw, dict):
|
|
return 0.5, []
|
|
rollbacks = int(raw.get("prior_rollbacks", 0))
|
|
fails = int(raw.get("prior_policy_fails", 0))
|
|
s = 1.0 - (rollbacks * 0.2 + fails * 0.1)
|
|
return max(0.0, min(1.0, s)), []
|
|
if name == "nfrs":
|
|
if not isinstance(raw, dict):
|
|
return 0.5, []
|
|
conf = raw.get("conformance")
|
|
if conf is None:
|
|
return 0.5, []
|
|
return float(conf), []
|
|
return 0.5, []
|
|
|
|
|
|
def compute(contract_id: str, environment: str,
|
|
inputs: Dict[str, Any]) -> Signal:
|
|
"""Orchestrate the 6-input weighted sum + severity penalty + band."""
|
|
missing = sorted(set(WEIGHTS.keys()) - set(inputs.keys()))
|
|
if missing:
|
|
return Signal(0.0, "block", {},
|
|
[f"INPUT_MISSING:{m}" for m in missing])
|
|
|
|
per_input: Dict[str, float] = {}
|
|
reasons: List[str] = []
|
|
base = 0.0
|
|
for name, weight in WEIGHTS.items():
|
|
raw = inputs.get(name)
|
|
s, r = _per_input_score(name, raw)
|
|
per_input[name] = s
|
|
reasons.extend(r)
|
|
base += s * weight
|
|
|
|
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
|
|
if pcr.get("result") != "fail":
|
|
continue
|
|
sev = pcr.get("severity")
|
|
p = PENALTY.get(sev, 0.0)
|
|
if p is None:
|
|
# 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
|
|
|
|
if critical_override:
|
|
score = 0.0
|
|
band = "block"
|
|
else:
|
|
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).
|
|
# The "AI decision" is the confidence-gated policy engine, not an LLM.
|
|
# decision_id = run_id (or "cli-<ts>" when called from CLI without a run).
|
|
try:
|
|
run_id = os.environ.get("NOVA_RUN_ID", f"cli-{int(__import__('time').time())}")
|
|
conf_data = {"score": score, "band": band, "perInput": per_input, "reasonCodes": reasons}
|
|
emit("nova.confidence.computed", run_id, environment, conf_data, contract_id=contract_id)
|
|
|
|
decision_data = {
|
|
"decision_id": run_id,
|
|
"chosen_action": band,
|
|
"confidence": score,
|
|
"alternatives": per_input,
|
|
"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")
|
|
append_event(decision_event)
|
|
ledger_append(decision_event)
|
|
except Exception:
|
|
pass # metrics emission must never break the confidence gate
|
|
|
|
return signal
|
|
|
|
|
|
if __name__ == "__main__":
|
|
if len(sys.argv) < 3:
|
|
print("usage: confidence_signal.py <inputs.json> <environment>", file=sys.stderr)
|
|
sys.exit(2)
|
|
env = sys.argv[2]
|
|
with open(sys.argv[1], "r", encoding="utf-8") as fh:
|
|
inputs = json.load(fh)
|
|
sig = compute("cli", env, inputs)
|
|
print(json.dumps(asdict(sig), indent=2)) |