From 68d90c08a73c650b14af28b83346ff4cd3ffafe9 Mon Sep 17 00:00:00 2001 From: Jon Chery Date: Tue, 21 Jul 2026 18:46:13 +0000 Subject: [PATCH] phase: 7, status: plan-as-execute, persona: backend-engineer+security-engineer, task: T-7.9 ---ci--- project: acdl phase: 7 milestone: v1.1 status: plan-as-execute persona: backend-engineer+security-engineer task: T-7.9 requirements.covered: [REQ-19] ---/ci--- Wave 4 (co-authored): platform/confidence_signal.py (REQ-19). backend-engineer: 6-input weighted sum (WEIGHTS = policy 0.30, validation 0.25, freshness 0.10, source 0.15, history 0.10, nfrs 0.10 per D-040; sums to 1.0), per-env thresholds (dev 0.50, qa 0.75, prod 0.90, dr 0.95), _per_input_score rules, compute() orchestration, Signal dataclass, dev-warn->block flip (autonomous, no reviewer). security-engineer: PENALTY dict (critical=None hard-override sentinel, high 0.20, medium 0.05, low 0.01, info 0.0), critical-override short-circuit return in compute(), severity penalty loop over the policy input's failing PolicyCheckResult records. Verified: missing input -> block + INPUT_MISSING:; critical fail -> 0.0 block + CRITICAL_OVERRIDE:; cold-start spike dev (ACDL _TAG_NAMING skipped, all validation true, nfrs neutral 0.5) -> 0.95 pass. stdlib only. --- platform/confidence_signal.py | 175 ++++++++++++++++++++++++++++++++++ 1 file changed, 175 insertions(+) create mode 100644 platform/confidence_signal.py diff --git a/platform/confidence_signal.py b/platform/confidence_signal.py new file mode 100644 index 0000000..c5d3502 --- /dev/null +++ b/platform/confidence_signal.py @@ -0,0 +1,175 @@ +"""ACDL 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, ir_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", "ir_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 ", 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)) \ No newline at end of file