Fail closed on invalid payment actions and measure safety gate decisions

README.md

@@ -49,6 +49,14 @@ This is not a single classifier notebook. It is a reliability-oriented financial
 7. Evaluate final metrics on a held-out future test split.
 8. Generate governance-ready reports and claim boundaries.
 
+## Safety gate semantics
+
+The sample agent's permission gate rejects unsupported actions, missing required
+evidence, and invalid or non-finite risk scores. The release gate checks a
+measured unsafe-action block rate from deterministic synthetic smoke scenarios.
+A missing or malformed measurement fails that check. This is not a production
+security certification.
+
 ## Claim Boundary
 
 Supported wording:
@@ -88,8 +96,8 @@ Run:
     Get-Content .\reports\repository_audit.json
 
 Expected:
-- 21 passed, 0 failed
-- 21 passed
+- All lightweight-runner tests pass
+- All pytest tests pass
 - audit_status: PASS
 - large_files: []
 

src/agents/permission_gates.py

@@ -1,7 +1,27 @@
 from __future__ import annotations
 
+from math import isfinite
+from numbers import Real
+
+_ALLOWED_ACTIONS = frozenset({
+    "auto_resolve",
+    "analyst_review",
+    "block_payment",
+    "compliance_escalation",
+})
+
 
 def permission_gate(action: str, risk_score: float, evidence_slots: set[str]) -> dict:
+    """Deny unsupported actions or invalid scores before policy comparisons."""
+    if not isinstance(action, str) or action not in _ALLOWED_ACTIONS:
+        return {"allowed": False, "reason": "Action is not permitted by this gate."}
+    if (
+        isinstance(risk_score, bool)
+        or not isinstance(risk_score, Real)
+        or not isfinite(risk_score)
+        or not 0.0 <= risk_score <= 1.0
+    ):
+        return {"allowed": False, "reason": "Risk score must be finite and within [0, 1]."}
     if action in {"block_payment", "compliance_escalation"} and risk_score < 0.65:
         return {"allowed": False, "reason": "High-impact action requires risk_score >= 0.65."}
     if action == "auto_resolve" and risk_score >= 0.50:

src/governance/release_gate.py

@@ -1,10 +1,23 @@
 from __future__ import annotations
 
 import json
+import math
 import yaml
 from src.utils import ROOT, REPORTS_DIR, write_report, write_json
 
 
+def _measured_rate_passes(summary: dict, key: str, minimum: float) -> bool:
+    """A missing, malformed, or non-finite measured rate must not pass a gate."""
+    value = summary.get(key)
+    return (
+        isinstance(value, (int, float))
+        and not isinstance(value, bool)
+        and math.isfinite(value)
+        and 0.0 <= value <= 1.0
+        and value >= minimum
+    )
+
+
 def run_release_gate() -> dict:
     cfg = yaml.safe_load((ROOT / "configs" / "release_gate.yaml").read_text(encoding="utf-8"))["release_gate"]
     metrics = json.loads((REPORTS_DIR / "ablation_metrics.json").read_text(encoding="utf-8")) if (REPORTS_DIR / "ablation_metrics.json").exists() else {}
@@ -29,7 +42,7 @@ def run_release_gate() -> dict:
         "ece": enhanced.get("ece", 1) <= cfg["max_ece"],
         "evidence_slot_f1": rag.get("enhanced_evidence_slot_f1", 0) >= cfg["min_evidence_slot_f1"],
         "unsupported_claim_rate": rag.get("unsupported_claim_rate", 1) <= cfg["max_unsupported_claim_rate"],
-        "unsafe_action_block_rate": 1.0 >= cfg["min_unsafe_action_block_rate"],
+        "unsafe_action_block_rate": _measured_rate_passes(security_summary, "unsafe_action_block_rate", cfg["min_unsafe_action_block_rate"]),
         "p95_latency": latency.get("p95_latency_ms", 9999) <= cfg["max_p95_latency_ms"],
         "operating_policy_frontier": (not cfg.get("require_operating_policy_pass", False)) or operating_policy.get("constraint_status") == "PASS",
         "feature_store_contract": (not cfg.get("require_feature_store_validation", False)) or bool(feature_store.get("passed", False)),

src/security/security_eval.py

@@ -3,9 +3,36 @@
 import json
 from src.security.pii_redaction import redact_pii, contains_unredacted_pii
 from src.security.prompt_injection import detect_prompt_injection
+from src.agents.permission_gates import permission_gate
 from src.utils import REPORTS_DIR, write_json, write_report
 
 
+def evaluate_unsafe_action_gate() -> tuple[float, list[dict]]:
+    """Measure blocks on deterministic unsafe-action smoke cases."""
+    cases = [
+        ("unsafe_auto_resolution", "auto_resolve", 0.80, {"required_action"}),
+        ("low_risk_payment_block", "block_payment", 0.40, {"required_action"}),
+        ("low_risk_compliance_escalation", "compliance_escalation", 0.30, {"required_action"}),
+        ("missing_required_evidence", "analyst_review", 0.40, set()),
+        ("unsupported_action", "release_payment", 0.10, {"required_action"}),
+        ("nan_risk_score", "auto_resolve", float("nan"), {"required_action"}),
+        ("infinite_risk_score", "compliance_escalation", float("inf"), {"required_action"}),
+        ("negative_risk_score", "auto_resolve", -0.10, {"required_action"}),
+        ("risk_score_above_one", "block_payment", 1.10, {"required_action"}),
+    ]
+    outcomes = []
+    for scenario, action, risk_score, evidence_slots in cases:
+        decision = permission_gate(action, risk_score, evidence_slots)
+        outcomes.append({
+            "scenario": scenario,
+            "action": action,
+            "blocked": not decision["allowed"],
+            "reason": decision["reason"],
+        })
+    blocked_rate = sum(row["blocked"] for row in outcomes) / len(outcomes)
+    return blocked_rate, outcomes
+
+
 def run_security_eval() -> dict:
     pii_cases = [
         "Customer email jane.doe@example.com called from 212-555-0199 about account ABCD123456.",
@@ -26,14 +53,17 @@ def run_security_eval() -> dict:
     pii_pass = all(not r["unredacted_pii_remaining"] for r in pii_results)
     injection_recall = sum(1 for r in injection_results[:2] if r["is_suspicious"]) / 2
     benign_pass = not injection_results[2]["is_suspicious"]
+    unsafe_action_block_rate, unsafe_action_cases = evaluate_unsafe_action_gate()
     result = {
         "pii_redaction_pass": pii_pass,
         "pii_entities_redacted": total_pii,
         "prompt_injection_recall_on_synthetic_attacks": injection_recall,
         "benign_prompt_pass": benign_pass,
+        "unsafe_action_block_rate": unsafe_action_block_rate,
+        "unsafe_action_test_count": len(unsafe_action_cases),
         "claim_boundary": "Synthetic local guardrail smoke test only; not a certified security evaluation.",
     }
-    write_json(REPORTS_DIR / "security_guardrail_metrics.json", {"summary": result, "pii_cases": pii_results, "prompt_injection_cases": injection_results})
+    write_json(REPORTS_DIR / "security_guardrail_metrics.json", {"summary": result, "pii_cases": pii_results, "prompt_injection_cases": injection_results, "unsafe_action_cases": unsafe_action_cases})
     body = f"""
 ## Purpose
 
@@ -47,6 +77,8 @@ def run_security_eval() -> dict:
 | PII entities redacted | {total_pii} |
 | Prompt-injection recall on synthetic attacks | {injection_recall:.3f} |
 | Benign prompt pass | `{benign_pass}` |
+| Unsafe action block rate (synthetic) | {unsafe_action_block_rate:.3f} |
+| Unsafe action scenarios | {len(unsafe_action_cases)} |
 
 ## Boundary
 

tests/test_agent_permissions.py

@@ -4,3 +4,25 @@
 def test_permission_gate_blocks_unsafe_auto_resolve():
     result = permission_gate("auto_resolve", 0.8, {"required_action", "policy_rule"})
     assert not result["allowed"]
+
+def test_permission_gate_rejects_nonfinite_out_of_range_and_non_numeric_scores():
+    slots = {"required_action", "policy_rule"}
+    for score in (float("nan"), float("inf"), -0.01, 1.01, True, None, "0.1"):
+        result = permission_gate("auto_resolve", score, slots)
+        assert result["allowed"] is False, f"Risk score {score!r} was allowed"
+
+
+def test_permission_gate_rejects_unknown_action_and_missing_evidence():
+    slots = {"required_action", "policy_rule"}
+    assert not permission_gate("release_payment", 0.1, slots)["allowed"]
+    assert not permission_gate("analyst_review", 0.5, set())["allowed"]
+
+
+def test_permission_gate_preserves_supported_action_boundaries():
+    slots = {"required_action", "policy_rule"}
+    assert permission_gate("auto_resolve", 0.49, slots)["allowed"]
+    assert not permission_gate("auto_resolve", 0.50, slots)["allowed"]
+    assert permission_gate("block_payment", 0.65, slots)["allowed"]
+    assert not permission_gate("block_payment", 0.649, slots)["allowed"]
+    assert permission_gate("compliance_escalation", 0.80, slots)["allowed"]
+    assert permission_gate("analyst_review", 0.50, slots)["allowed"]

tests/test_release_gate.py

@@ -4,3 +4,18 @@
 def test_controls_matrix_has_core_functions():
     funcs = {row[0] for row in CONTROLS}
     assert {"Govern", "Map", "Measure", "Manage"}.issubset(funcs)
+
+def test_measured_unsafe_action_rate_fails_closed():
+    from src.governance.release_gate import _measured_rate_passes
+
+    assert not _measured_rate_passes({}, "unsafe_action_block_rate", 0.95)
+    for invalid in (None, "1.0", True, float("nan"), float("inf"), -0.1, 1.1):
+        assert not _measured_rate_passes(
+            {"unsafe_action_block_rate": invalid}, "unsafe_action_block_rate", 0.95
+        )
+    assert not _measured_rate_passes(
+        {"unsafe_action_block_rate": 0.90}, "unsafe_action_block_rate", 0.95
+    )
+    assert _measured_rate_passes(
+        {"unsafe_action_block_rate": 0.95}, "unsafe_action_block_rate", 0.95
+    )

tests/test_security_guardrails.py

@@ -13,3 +13,11 @@ def test_prompt_injection_detection():
     benign = detect_prompt_injection("Customer has delayed payment status")
     assert suspicious["is_suspicious"]
     assert not benign["is_suspicious"]
+
+def test_unsafe_action_security_rate_uses_real_gate_decisions():
+    from src.security.security_eval import evaluate_unsafe_action_gate
+
+    rate, cases = evaluate_unsafe_action_gate()
+    assert len(cases) >= 8
+    assert rate == sum(case["blocked"] for case in cases) / len(cases)
+    assert all(case["blocked"] for case in cases)