Spaces:
Paused
Paused
| """Policy engine: determinism, version hashing, pre-authorized stop-ship, | |
| INDETERMINATE fail-safe, severity taxonomy, the full option matrix and the | |
| safety-as-hard-constraint guarantee.""" | |
| from __future__ import annotations | |
| import copy | |
| from sensorflow.agentic import policy as policy_mod | |
| from sensorflow.agentic.policy import PolicyInput | |
| def _base_input(**overrides) -> PolicyInput: | |
| kwargs = dict( | |
| failure_id="fail-test", | |
| safety_critical_class=True, | |
| behavioral_evidence="observed_unsafe", | |
| downstream_contained=False, | |
| rate=1.0e-4, rate_ci=[7.0e-5, 1.6e-4], denominator=240_000, | |
| significant=True, small_sample=False, | |
| novelty="novel", evidence_confidence="CONFIRMED", | |
| downstream_consequence="observed_critical", | |
| gt_available=True, lineage_complete=True, telemetry_available=True, | |
| fusion_verdict="multi_modal_supported", agent_conflict=False, | |
| ) | |
| kwargs.update(overrides) | |
| return PolicyInput(**kwargs) | |
| # ------------------------------------------------------------------ versioning | |
| def test_policy_hash_versioning_and_determinism(): | |
| doc = policy_mod.save_policy(copy.deepcopy(policy_mod.DEFAULT_POLICY)) | |
| assert doc["policy_version"] == policy_mod.policy_hash(doc) | |
| # identical content -> identical hash; changed content -> new hash | |
| again = policy_mod.policy_hash(copy.deepcopy(policy_mod.DEFAULT_POLICY)) | |
| assert again == doc["policy_version"] | |
| mutated = copy.deepcopy(policy_mod.DEFAULT_POLICY) | |
| mutated["expected_loss"]["delay_cost"] = 999.0 | |
| assert policy_mod.policy_hash(mutated) != doc["policy_version"] | |
| # persisted + retrievable by version | |
| fetched = policy_mod.get_policy(doc["policy_version"]) | |
| assert fetched["policy_name"] == doc["policy_name"] | |
| # example thresholds must be flagged as placeholders | |
| assert doc["placeholder_values"] is True | |
| def test_policy_evaluation_is_deterministic(): | |
| inp = _base_input() | |
| r1 = policy_mod.evaluate(inp) | |
| r2 = policy_mod.evaluate(inp) | |
| for key in ("outcome", "severity", "recommended_option", "policy_version"): | |
| assert r1[key] == r2[key] | |
| assert r1["expected_loss_table"] == r2["expected_loss_table"] | |
| # ------------------------------------------------------------------ stop-ship | |
| def test_automatic_stop_ship_fires_only_on_preauthorized_conditions(): | |
| # ASS-1: S4 + CONFIRMED + significant -> fires | |
| r = policy_mod.evaluate(_base_input()) | |
| assert r["severity"] == "S4" | |
| assert r["outcome"] == "AUTOMATIC_STOP_SHIP" | |
| assert r["automatic_stop_ship_condition"]["condition_id"] == "ASS-1" | |
| assert r["recommended_option"] == "STOP_SHIP" | |
| # same facts but only LIKELY confidence -> NOT pre-authorized | |
| r2 = policy_mod.evaluate(_base_input(evidence_confidence="LIKELY")) | |
| assert r2["outcome"] != "AUTOMATIC_STOP_SHIP" | |
| # not significant -> NOT pre-authorized | |
| r3 = policy_mod.evaluate(_base_input(significant=False)) | |
| assert r3["outcome"] != "AUTOMATIC_STOP_SHIP" | |
| # ------------------------------------------------------------------ INDETERMINATE | |
| def test_indeterminate_on_missing_gt_lineage_telemetry_and_conflict(): | |
| cases = [ | |
| {"gt_available": False}, | |
| {"lineage_complete": False}, | |
| {"telemetry_available": False, "denominator": None}, | |
| {"fusion_verdict": "verification_failed"}, | |
| {"agent_conflict": True, "conflict_details": ["vlm vs fusion"]}, | |
| ] | |
| for overrides in cases: | |
| r = policy_mod.evaluate(_base_input(**overrides)) | |
| assert r["outcome"] == "INDETERMINATE", overrides | |
| # never a pass, always escalates to a human | |
| assert r["outcome"] != "NO_LAUNCH_IMPACT" | |
| assert r["recommended_option"] == "HUMAN_SAFETY_REVIEW" | |
| assert r["indeterminate_reasons"] | |
| # ------------------------------------------------------------------ severity | |
| def test_severity_taxonomy_criteria(): | |
| doc = policy_mod.get_policy() | |
| assert set(doc["severity_taxonomy"]) == {"S0", "S1", "S2", "S3", "S4", "S5"} | |
| def sev(**kw): | |
| return policy_mod.assign_severity(doc, _base_input(**kw))["severity"] | |
| assert sev(collision_observed=True) == "S5" | |
| assert sev(behavioral_evidence="observed_unsafe") == "S4" | |
| assert sev(behavioral_evidence="none", downstream_contained=False) == "S3" | |
| assert sev(behavioral_evidence="none", downstream_contained=True) == "S2" | |
| assert sev(safety_critical_class=False, behavioral_evidence="none", | |
| functional_impact=True) == "S1" | |
| assert sev(safety_critical_class=False, behavioral_evidence="none", | |
| functional_impact=False) == "S0" | |
| # ------------------------------------------------------------------ option matrix | |
| def test_option_matrix_all_seven_rows(): | |
| # row 1: gate violated -> STOP_SHIP | |
| r = policy_mod.evaluate(_base_input(gate_violated=True, | |
| evidence_confidence="LIKELY")) | |
| assert r["matrix_row_fired"]["row"] == 1 | |
| assert r["recommended_option"] == "STOP_SHIP" | |
| # row 2: high-confidence uncontained critical -> OPTION_A_DELAY | |
| r = policy_mod.evaluate(_base_input(evidence_confidence="LIKELY")) | |
| assert r["matrix_row_fired"]["row"] == 2 | |
| assert r["recommended_option"] == "OPTION_A_DELAY" | |
| # row 3: validated mitigation -> OPTION_B_MITIGATION | |
| r = policy_mod.evaluate(_base_input(evidence_confidence="LIKELY", | |
| mitigation_validated=True, | |
| mitigation_description="class-gate patch")) | |
| assert r["matrix_row_fired"]["row"] == 3 | |
| assert r["recommended_option"] == "OPTION_B_MITIGATION" | |
| # row 4: reliably detectable ODD concentration -> OPTION_C_REDUCED_ODD | |
| r = policy_mod.evaluate(_base_input( | |
| behavioral_evidence="observed_contained", downstream_contained=True, | |
| downstream_consequence="observed_degraded", | |
| evidence_confidence="LIKELY", | |
| concentrated=True, concentration_dimension="lighting", | |
| odd_detector_recall=0.999, odd_exclusion_share=0.10, | |
| residual_rate_outside_odd=4.0e-6)) | |
| assert r["matrix_row_fired"]["row"] == 4 | |
| assert r["recommended_option"] == "OPTION_C_REDUCED_ODD" | |
| assert r["option_c_evaluation"]["feasible"] is True | |
| # row 4 must NOT fire when the dimension is not instrumented in the ODD | |
| # taxonomy (geography), even with perfect detector numbers | |
| r_geo = policy_mod.evaluate(_base_input( | |
| behavioral_evidence="observed_contained", downstream_contained=True, | |
| downstream_consequence="observed_degraded", | |
| evidence_confidence="LIKELY", | |
| concentrated=True, concentration_dimension="geo_bucket", | |
| odd_detector_recall=0.999, odd_exclusion_share=0.10, | |
| residual_rate_outside_odd=4.0e-6)) | |
| assert r_geo["option_c_evaluation"]["feasible"] is False | |
| assert r_geo["recommended_option"] != "OPTION_C_REDUCED_ODD" | |
| # row 5: insufficient evidence -> EXPAND_EVALUATION | |
| r = policy_mod.evaluate(_base_input( | |
| behavioral_evidence="none", downstream_contained=True, | |
| significant=False, small_sample=True, | |
| evidence_confidence="INSUFFICIENT_EVIDENCE", | |
| downstream_consequence="uncertain")) | |
| assert r["matrix_row_fired"]["row"] == 5 | |
| assert r["recommended_option"] == "EXPAND_EVALUATION" | |
| # row 6: modality conflict -> HUMAN_SAFETY_REVIEW | |
| r = policy_mod.evaluate(_base_input( | |
| safety_critical_class=False, behavioral_evidence="none", | |
| evidence_confidence="LIKELY", rate=1.0e-6, | |
| fusion_verdict="modality_conflict", | |
| downstream_consequence="uncertain")) | |
| assert r["matrix_row_fired"]["row"] == 6 | |
| assert r["recommended_option"] == "HUMAN_SAFETY_REVIEW" | |
| # row 7: no safety impact -> PROCEED | |
| r = policy_mod.evaluate(_base_input( | |
| safety_critical_class=False, behavioral_evidence="none", | |
| significant=False, rate=1.0e-6, novelty="known_stable", | |
| evidence_confidence="LIKELY", | |
| downstream_consequence="none_observed")) | |
| assert r["matrix_row_fired"]["row"] == 7 | |
| assert r["recommended_option"] == "PROCEED" | |
| # ------------------------------------------------------------------ hard constraint | |
| def test_cheaper_unsafe_option_never_selected(): | |
| # Craft a policy where PROCEED is by far the cheapest option. | |
| doc = copy.deepcopy(policy_mod.DEFAULT_POLICY) | |
| doc["expected_loss"]["incident_cost"] = 0.0 | |
| doc["expected_loss"]["delay_cost"] = 10_000.0 | |
| doc = policy_mod.save_policy(doc) | |
| # S3 severity, POSSIBLE confidence: rows 1-6 all miss, row 7 (PROCEED) | |
| # matches — but PROCEED is infeasible under the hard constraint because | |
| # the residual rate exceeds MAX_ACCEPTABLE_RESIDUAL_RISK at S3+. | |
| # (odd_exclusion_share gives every alternative a nonzero business cost, | |
| # so PROCEED is STRICTLY cheapest.) | |
| inp = _base_input(behavioral_evidence="none", downstream_contained=False, | |
| evidence_confidence="POSSIBLE", significant=True, | |
| small_sample=False, rate=1.0e-4, | |
| odd_exclusion_share=0.5, | |
| downstream_consequence="uncertain") | |
| r = policy_mod.evaluate(inp, policy=doc) | |
| table = {row["option"]: row for row in r["expected_loss_table"]} | |
| # PROCEED is the cheapest option in the table... | |
| cheapest = min(r["expected_loss_table"], key=lambda x: x["expected_loss"]) | |
| assert cheapest["option"] == "PROCEED" | |
| # ...but it is infeasible and therefore NOT selected. | |
| assert table["PROCEED"]["feasible"] is False | |
| assert r["recommended_option"] != "PROCEED" | |
| assert r["recommended_option"] == "HUMAN_SAFETY_REVIEW" # fail-safe | |
| # restore the default active policy for the rest of the suite | |
| policy_mod.save_policy(copy.deepcopy(policy_mod.DEFAULT_POLICY)) | |