from __future__ import annotations import copy from datetime import datetime, timezone import json import unittest from existence_possibility_compiler import ( SCHEMA, compile_plan, validate, digest ) def action(aid, *, requires=None, provides=None, risk=0.05, reversible=True, approval=False, cost=None, observations=None, conflicts=None, confidence=0.9, effects=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": cost or {"cpu": 1}, "effects": effects or {"discover": 5, "create": 5, "preserve": 5, "evolve": 5}, "confidence": confidence, "risk": risk, "evidence_grade": "tested", "reversible": reversible, "approval_required": approval, "preconditions": ["baseline is present"], "verification": ["test passes"], "success_metrics": ["metric is met"], "rollback": "remove the temporary artifact" } def problem(actions=None): return { "schema": SCHEMA, "title": "unit fixture", "resource_budget": {"cpu": 8}, "mission_weights": {"discover": .25, "create": .25, "preserve": .25, "evolve": .25}, "current_capabilities": ["base"], "target_capabilities": ["target"], "observations": [{"id": "obs", "status": "verified", "source": "fixture:unit-test"}], "constraints": {"max_risk_per_action": .5, "max_total_risk": 1, "max_plan_steps": 3, "require_reversible": True, "allow_approval_required": False}, "actions": actions or [ action("prepare", requires=["base"], provides=["prepared"], cost={"cpu": 1}), action("experiment", requires=["prepared"], provides=["target"], cost={"cpu": 2}, observations=["obs"]), action("package", requires=["target"], provides=["package"], cost={"cpu": 1}) ] } FIXED_NOW = datetime(2026, 11, 3, 12, 0, tzinfo=timezone.utc) class ExistenceEngineTests(unittest.TestCase): def test_finds_dependency_chain_and_target(self): result = compile_plan(problem(), now=FIXED_NOW) self.assertEqual(result["recommended_plan"]["actions"][:2], ["prepare", "experiment"]) self.assertEqual(result["status"], "target_reached") self.assertFalse(result["execution_performed"]) def test_future_option_value_rewards_unlocking(self): p = problem() p["constraints"]["max_plan_steps"] = 2 result = compile_plan(p, now=FIXED_NOW) plan = result["recommended_plan"] self.assertIn("package", [x["id"] for x in plan["newly_unlocked_options"]]) self.assertGreater(plan["new_option_value"], 0) def test_option_frontier_breaks_direct_value_tie(self): p = problem([ action("future_enabler", requires=["base"], provides=["branch"]), action("future_option", requires=["branch"], provides=["next"], effects={"discover": 10, "create": 10, "preserve": 10, "evolve": 10}), action("direct_only", requires=["base"], provides=["other"]) ]) p["target_capabilities"] = [] p["constraints"]["max_plan_steps"] = 1 result = compile_plan(p, now=FIXED_NOW) self.assertEqual(result["recommended_plan"]["actions"], ["future_enabler"]) self.assertIn("future_option", [x["id"] for x in result["recommended_plan"]["newly_unlocked_options"]]) def test_resource_budget_blocks_oversized_action(self): p = problem([action("too_big", requires=["base"], provides=["target"], cost={"cpu": 9})]) result = compile_plan(p, now=FIXED_NOW) self.assertEqual(result["status"], "no_safe_feasible_action") self.assertIn("resource_limit:cpu", result["actions_blocked_from_initial_state"][0]["reasons"]) def test_unverified_observation_fails_closed(self): p = problem([action("needs_evidence", requires=["base"], provides=["target"], observations=["obs"])]) p["observations"][0]["status"] = "hypothesis" result = compile_plan(p, now=FIXED_NOW) self.assertEqual(result["status"], "no_safe_feasible_action") self.assertTrue(any(x.startswith("unverified_observations:") for x in result["actions_blocked_from_initial_state"][0]["reasons"])) def test_verified_observation_requires_source(self): p = problem() p["observations"][0]["source"] = "" with self.assertRaises(ValueError): validate(p) def test_irreversible_action_blocked_by_default(self): p = problem([action("irreversible", requires=["base"], provides=["target"], reversible=False, approval=True)]) result = compile_plan(p, now=FIXED_NOW) reasons = result["actions_blocked_from_initial_state"][0]["reasons"] self.assertIn("reversibility_required", reasons) self.assertIn("approval_not_permitted", reasons) def test_approval_action_can_be_proposed_only_when_explicitly_allowed(self): p = problem([action("approval_step", requires=["base"], provides=["target"], approval=True)]) p["constraints"]["allow_approval_required"] = True result = compile_plan(p, now=FIXED_NOW) self.assertEqual(result["recommended_plan"]["actions"], ["approval_step"]) self.assertTrue(result["recommended_plan"]["requires_human_approval"]) self.assertEqual(result["experiment_contracts"][0]["status"], "proposed_not_executed") def test_per_action_risk_limit_blocks(self): p = problem([action("risky", requires=["base"], provides=["target"], risk=.8)]) result = compile_plan(p, now=FIXED_NOW) self.assertIn("per_action_risk_limit", result["actions_blocked_from_initial_state"][0]["reasons"]) def test_cumulative_risk_limit_blocks_combination(self): p = problem([ action("one", requires=["base"], provides=["one_cap"], risk=.3), action("two", requires=["one_cap"], provides=["target"], risk=.3) ]) p["constraints"]["max_total_risk"] = .5 result = compile_plan(p, now=FIXED_NOW) self.assertNotIn("two", result["recommended_plan"]["actions"]) def test_conflicting_actions_not_selected_together(self): p = problem([ action("one", requires=["base"], provides=["one_cap"], conflicts=["two"]), action("two", requires=["base"], provides=["target"]) ]) result = compile_plan(p, now=FIXED_NOW) self.assertLessEqual(len(set(result["recommended_plan"]["actions"]) & {"one", "two"}), 1) def test_unknown_observation_reference_rejected(self): p = problem([action("bad", requires=["base"], provides=["target"], observations=["missing"])]) with self.assertRaises(ValueError): validate(p) def test_non_finite_numbers_rejected(self): p = problem() p["actions"][0]["confidence"] = float("nan") with self.assertRaises(ValueError): validate(p) def test_duplicate_action_ids_rejected(self): p = problem() p["actions"].append(copy.deepcopy(p["actions"][0])) with self.assertRaises(ValueError): validate(p) def test_dependency_cycle_yields_no_safe_plan(self): p = problem([ action("a", requires=["b_cap"], provides=["a_cap"]), action("b", requires=["a_cap"], provides=["b_cap"]) ]) result = compile_plan(p, now=FIXED_NOW) self.assertEqual(result["status"], "no_safe_feasible_action") def test_deterministic_digest_for_same_input_and_clock(self): p = problem() a = compile_plan(p, now=FIXED_NOW) b = compile_plan(p, now=FIXED_NOW) self.assertEqual(a["decision_digest_sha256"], b["decision_digest_sha256"]) self.assertEqual(a["input_digest_sha256"], digest(p)) def test_changes_in_input_change_input_digest(self): p = problem() a = compile_plan(p, now=FIXED_NOW) p["resource_budget"]["cpu"] = 7 b = compile_plan(p, now=FIXED_NOW) self.assertNotEqual(a["input_digest_sha256"], b["input_digest_sha256"]) def test_missing_target_is_reported_without_false_claim(self): p = problem([action("safe_non_target", requires=["base"], provides=["other"])]) result = compile_plan(p, now=FIXED_NOW) self.assertEqual(result["status"], "feasible_plan") self.assertEqual(result["recommended_plan"]["target_progress"], 0) def test_pareto_alternatives_are_present(self): result = compile_plan(problem(), now=FIXED_NOW) self.assertTrue(result["pareto_alternatives"]) self.assertLessEqual(len(result["pareto_alternatives"]), 8) def test_experiment_contract_includes_rollback_and_verification(self): result = compile_plan(problem(), now=FIXED_NOW) for contract in result["experiment_contracts"]: self.assertTrue(contract["rollback"]) self.assertTrue(contract["verification"]) self.assertEqual(contract["status"], "proposed_not_executed") def test_constraints_reject_invalid_boolean_type(self): p = problem() p["constraints"]["require_reversible"] = "false" with self.assertRaises(ValueError): validate(p) if __name__ == "__main__": unittest.main(verbosity=2)