from __future__ import annotations import copy from datetime import datetime, timezone import unittest from existence_possibility_compiler_v02 import SCHEMA, compile_plan, validate, digest NOW = datetime(2026, 11, 3, 12, 0, tzinfo=timezone.utc) DIMS = ("discover", "create", "preserve", "evolve") def rg(low, base, high): return {"low": low, "base": base, "high": high} def dr(v=0): return {d: rg(v, v, v) for d in DIMS} def cost(cpu=(0, 0, 0), disk=(0, 0, 0), cash=(0, 0, 0)): return { "cpu": rg(*cpu), "disk": rg(*disk), "cash": rg(*cash), } def action(aid, *, requires=None, provides=None, effects=None, externalities=None, risk=(.01, .02, .04), confidence=(.8, .9, .98), costs=None, reversible=True, approval=False, observations=None, conflicts=None): return { "id": aid, "title": aid.replace("_", " ").title(), "requires": requires or [], "provides": provides or [aid + "_cap"], "requires_observations": observations or [], "conflicts_with": conflicts or [], "cost": costs or cost((.1, .2, .3), (.01, .02, .03)), "effects": effects or dr(5), "externalities": externalities or dr(0), "risk": rg(*risk), "confidence": rg(*confidence), "evidence_grade": "tested", "reversible": reversible, "approval_required": approval, "preconditions": ["baseline exists"], "verification": ["independent check passes"], "success_metrics": ["declared threshold passes"], "rollback": "remove isolated artefact and retain baseline" } def base_problem(actions=None): return { "schema": SCHEMA, "title": "v0.2 unit fixture", "resource_budget": { "cpu": {"limit": 8, "shadow_price": .2}, "disk": {"limit": 4, "shadow_price": .5}, "cash": {"limit": 0, "shadow_price": 1} }, "mission_weights": {"discover": .25, "create": .25, "preserve": .25, "evolve": .25}, "current_capabilities": ["base"], "target_capabilities": ["target"], "observations": [{"id": "source", "status": "verified", "source": "fixture:source"}], "scenarios": [ {"id": "stress", "effect_case": "low", "externality_case": "high", "cost_case": "high", "risk_case": "high", "confidence_case": "low"}, {"id": "reference", "effect_case": "base", "externality_case": "base", "cost_case": "base", "risk_case": "base", "confidence_case": "base"}, {"id": "favourable", "effect_case": "high", "externality_case": "low", "cost_case": "low", "risk_case": "low", "confidence_case": "high"} ], "constraints": { "max_risk_per_action": .65, "max_total_risk": .7, "max_weighted_externality": 50, "max_plan_steps": 3, "regret_weight": .25, "require_reversible": True, "allow_approval_required": False }, "actions": actions or [ action("prepare", requires=["base"], provides=["prepared"]), action("experiment", requires=["prepared"], provides=["target"], observations=["source"]), action("next_option", requires=["target"], provides=["next"]), ] } class RobustPossibilityTests(unittest.TestCase): def test_dependency_chain_reaches_target(self): p = base_problem() p["constraints"]["max_plan_steps"] = 2 r = compile_plan(p, now=NOW) self.assertEqual(r["status"], "target_reached") self.assertEqual(r["recommended_plan"]["actions"], ["prepare", "experiment"]) self.assertFalse(r["execution_performed"]) def test_output_uses_robust_v2_schema(self): r = compile_plan(base_problem(), now=NOW) self.assertEqual(r["schema"], "darkoza.existence-plan.v2") self.assertEqual(r["engine"], "darkoza-epc-0.2") def test_persistent_option_is_unlocked_in_every_scenario(self): p = base_problem() p["constraints"]["max_plan_steps"] = 2 r = compile_plan(p, now=NOW) ids = [x["id"] for x in r["recommended_plan"]["persistent_new_options"]] self.assertIn("next_option", ids) def test_stress_only_option_is_not_credited(self): p = base_problem([ action("prepare", requires=["base"], provides=["prepared"]), action("experiment", requires=["prepared"], provides=["target"]), action("fragile", requires=["target"], provides=["fragile_result"], costs=cost((1, 3, 9), (0, 1, 5))) ]) r = compile_plan(p, now=NOW) plan = r["recommended_plan"] self.assertEqual(plan["actions"], ["prepare", "experiment"]) self.assertIn("fragile", [x["id"] for x in plan["fragile_new_options"]]) self.assertNotIn("fragile", [x["id"] for x in plan["persistent_new_options"]]) def test_fragile_option_lists_where_it_is_available(self): p = base_problem([ action("prepare", requires=["base"], provides=["prepared"]), action("experiment", requires=["prepared"], provides=["target"]), action("fragile", requires=["target"], provides=["fragile_result"], costs=cost((1, 3, 9), (0, 1, 5))) ]) plan = compile_plan(p, now=NOW)["recommended_plan"] item = next(x for x in plan["fragile_new_options"] if x["id"] == "fragile") self.assertEqual(item["available_in_scenarios"], ["favourable", "reference"]) def test_no_scenario_probabilities_are_assumed(self): r = compile_plan(base_problem(), now=NOW) self.assertFalse(r["scenario_method"]["probabilities_assumed"]) self.assertIn("not an expected value", r["scenario_method"]["mean_warning"]) def test_regret_is_computed_for_each_scenario(self): r = compile_plan(base_problem(), now=NOW) p = r["recommended_plan"] self.assertEqual(set(p["scenario_regrets"]), {"stress", "reference", "favourable"}) self.assertGreaterEqual(p["max_regret"], 0) self.assertIn(p["max_regret_scenario"], p["scenario_regrets"]) def test_scenario_winners_are_reported(self): r = compile_plan(base_problem(), now=NOW) self.assertEqual(set(r["scenario_winners"]), {"stress", "reference", "favourable"}) def test_robust_choice_can_differ_from_favourable_scenario_winner(self): volatile = action("volatile", requires=["base"], provides=["target"], effects={d: rg(1, 5, 10) for d in DIMS}, conflicts=["steady"]) steady = action("steady", requires=["base"], provides=["target"], effects={d: rg(4, 5, 6) for d in DIMS}, conflicts=["volatile"]) r = compile_plan(base_problem([volatile, steady]), now=NOW) self.assertEqual(r["recommended_plan"]["actions"], ["steady"]) self.assertEqual(r["scenario_winners"]["stress"], ["steady"]) self.assertEqual(r["scenario_winners"]["favourable"], ["volatile"]) def test_rank_stability_is_bounded(self): p = compile_plan(base_problem(), now=NOW)["recommended_plan"] self.assertGreaterEqual(p["scenario_rank_stability"], 0) self.assertLessEqual(p["scenario_rank_stability"], 1) def test_negative_externality_is_reported(self): harmful = action("harmful", requires=["base"], provides=["target"], externalities={d: rg(1, 2, 4) for d in DIMS}) r = compile_plan(base_problem([harmful]), now=NOW) self.assertGreater(r["recommended_plan"]["worst_case_externality"], 0) def test_externality_limit_blocks_action(self): bad = action("harmful", requires=["base"], provides=["target"], externalities={d: rg(1, 2, 4) for d in DIMS}) p = base_problem([bad]) p["constraints"]["max_weighted_externality"] = .5 r = compile_plan(p, now=NOW) self.assertEqual(r["status"], "no_safe_feasible_action") self.assertTrue(any("externality_limit" in x for x in r["actions_blocked_from_initial_state"][0]["reasons"])) def test_high_case_risk_fails_closed(self): risky = action("risky", requires=["base"], provides=["target"], risk=(.1, .3, .9)) r = compile_plan(base_problem([risky]), now=NOW) self.assertEqual(r["status"], "no_safe_feasible_action") self.assertTrue(any("per_action_risk_limit:stress" in x for x in r["actions_blocked_from_initial_state"][0]["reasons"])) def test_high_case_resource_cost_fails_closed(self): expensive = action("expensive", requires=["base"], provides=["target"], costs=cost((1, 2, 10), (0, 0, 0))) r = compile_plan(base_problem([expensive]), now=NOW) self.assertEqual(r["status"], "no_safe_feasible_action") self.assertTrue(any("resource_limit:cpu:stress" in x for x in r["actions_blocked_from_initial_state"][0]["reasons"])) def test_confidence_range_changes_scenario_value(self): a = action("a", requires=["base"], provides=["target"], confidence=(.2, .6, .95)) r = compile_plan(base_problem([a]), now=NOW) scores = r["recommended_plan"]["scenario_scores"] self.assertLess(scores["stress"]["direct_value"], scores["favourable"]["direct_value"]) def test_shadow_price_is_explicitly_applied(self): p = base_problem([action("costly", requires=["base"], provides=["target"], costs=cost((1, 2, 3), (1, 2, 3)))]) r = compile_plan(p, now=NOW) ref = r["recommended_plan"]["scenario_scores"]["reference"] self.assertGreater(ref["resource_shadow_cost"], 0) def test_shadow_price_can_change_selected_action(self): cheap = action("cheap", requires=["base"], provides=["target"], effects=dr(5), costs=cost((.1, .1, .1), (0, 0, 0))) rich = action("rich", requires=["base"], provides=["target"], effects=dr(6), costs=cost((5, 5, 5), (0, 0, 0))) p = base_problem([cheap, rich]) p["resource_budget"]["cpu"]["shadow_price"] = 0 first = compile_plan(p, now=NOW)["recommended_plan"]["actions"] p["resource_budget"]["cpu"]["shadow_price"] = 15 second = compile_plan(p, now=NOW)["recommended_plan"]["actions"] self.assertNotEqual(first, second) self.assertIn("cheap", second) def test_unverified_observation_blocks_action_in_all_scenarios(self): p = base_problem([action("needs_source", requires=["base"], provides=["target"], observations=["source"])]) p["observations"][0]["status"] = "hypothesis" r = compile_plan(p, now=NOW) self.assertEqual(r["status"], "no_safe_feasible_action") self.assertTrue(any(x.startswith("unverified_observations:") for x in r["actions_blocked_from_initial_state"][0]["reasons"])) def test_verified_observation_requires_source(self): p = base_problem() p["observations"][0]["source"] = "" with self.assertRaises(ValueError): validate(p) def test_invalid_range_order_rejected(self): p = base_problem() p["actions"][0]["risk"] = rg(.4, .2, .5) with self.assertRaises(ValueError): validate(p) def test_nan_range_rejected(self): p = base_problem() p["actions"][0]["confidence"]["base"] = float("nan") with self.assertRaises(ValueError): validate(p) def test_invalid_scenario_case_rejected(self): p = base_problem() p["scenarios"][0]["risk_case"] = "catastrophic" with self.assertRaises(ValueError): validate(p) def test_duplicate_scenario_id_rejected(self): p = base_problem() p["scenarios"][1]["id"] = p["scenarios"][0]["id"] with self.assertRaises(ValueError): validate(p) def test_scenario_count_bounds_enforced(self): p = base_problem() p["scenarios"] = p["scenarios"][:1] with self.assertRaises(ValueError): validate(p) def test_duplicate_action_id_rejected(self): p = base_problem() p["actions"].append(copy.deepcopy(p["actions"][0])) with self.assertRaises(ValueError): validate(p) def test_dependency_cycle_fails_closed(self): p = base_problem([ action("a", requires=["b_cap"], provides=["a_cap"]), action("b", requires=["a_cap"], provides=["target"]) ]) r = compile_plan(p, now=NOW) self.assertEqual(r["status"], "no_safe_feasible_action") def test_conflicting_actions_not_selected_together(self): p = base_problem([ action("one", requires=["base"], provides=["one_cap"], conflicts=["two"]), action("two", requires=["base"], provides=["target"]) ]) r = compile_plan(p, now=NOW) self.assertLessEqual(len(set(r["recommended_plan"]["actions"]) & {"one", "two"}), 1) def test_irreversible_action_blocked_by_default(self): p = base_problem([action("irreversible", requires=["base"], provides=["target"], reversible=False, approval=True)]) r = compile_plan(p, now=NOW) reasons = r["actions_blocked_from_initial_state"][0]["reasons"] self.assertIn("reversibility_required", reasons) self.assertIn("approval_not_permitted", reasons) def test_approval_gated_action_can_only_be_proposed_when_enabled(self): p = base_problem([action("approval_action", requires=["base"], provides=["target"], approval=True)]) p["constraints"]["allow_approval_required"] = True r = compile_plan(p, now=NOW) self.assertTrue(r["recommended_plan"]["actions"]) self.assertTrue(r["experiment_contracts"][0]["approval_required"]) self.assertFalse(r["execution_performed"]) def test_plan_is_never_executed(self): r = compile_plan(base_problem(), now=NOW) self.assertEqual(r["decision"], "PLAN_ONLY_NO_EXECUTION") self.assertFalse(r["execution_performed"]) self.assertTrue(all(c["status"] == "proposed_not_executed" for c in r["experiment_contracts"])) def test_reproducible_decision_digest(self): p = base_problem() a = compile_plan(p, now=NOW) b = compile_plan(p, now=NOW) self.assertEqual(a["decision_digest_sha256"], b["decision_digest_sha256"]) self.assertEqual(a["input_digest_sha256"], digest(p)) def test_input_change_changes_digest(self): p = base_problem() a = compile_plan(p, now=NOW) p["resource_budget"]["cpu"]["shadow_price"] += 1 b = compile_plan(p, now=NOW) self.assertNotEqual(a["input_digest_sha256"], b["input_digest_sha256"]) def test_pareto_alternatives_are_present_and_bounded(self): r = compile_plan(base_problem(), now=NOW) self.assertTrue(r["pareto_alternatives"]) self.assertLessEqual(len(r["pareto_alternatives"]), 8) def test_effective_frontier_values_are_nonnegative(self): r = compile_plan(base_problem(), now=NOW) for o in r["recommended_plan"]["persistent_new_options"]: self.assertGreaterEqual(o["worst_scenario_option_value"], 0) if __name__ == "__main__": unittest.main(verbosity=2)