acdl/core/confidence_signal.py

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


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 []
    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:
            return Signal(0.0, "block", per_input,
                          reasons + [f"CRITICAL_OVERRIDE:{pcr.get('ruleId','?')}"])
        penalty += p

    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"
    return Signal(score, band, per_input, reasons)


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