"""Regression — task-2553+35 TRACK C3: policy_profile_engine dry-run application. Phase-B contract: * C1 engine (anu_v3/policy_profile_engine) PRESENT -> run dry-run against the 4 task-2553 targets and assert engine-derived gate/HOLD/completion-packet schema matches the Phase-A read-only baseline. NO real merge / GitHub write / thread resolve is ever performed (dry-run only). * C1 engine ABSENT -> skip the engine-consuming assertions and assert the deferral is recorded as the explicit terminal status DEFERRED_PENDING_C1 (NOT HOLD, NOT FAIL) with a next_action. Phase-A baseline assertions always run (engine-independent): they re-derive the expected gate/HOLD/packet keys from the existing task-2553 actual-result events read-only and assert the comparison artifact encodes them faithfully. """ import importlib.util import json import os import unittest REPO_ROOT = os.path.abspath(os.path.join(os.path.dirname(__file__), "..", "..")) EVENTS = os.path.join(REPO_ROOT, "memory", "events") COMPARISON = os.path.join(EVENTS, "task-2553+35.dry-run-comparison.json") RESULT = os.path.join(EVENTS, "task-2553+35.result.json") ENGINE_MODULE = "anu_v3.policy_profile_engine" ENGINE_PATH = os.path.join(REPO_ROOT, "anu_v3", "policy_profile_engine.py") TARGETS = [ "PR#128 clean replacement merge", "PR#129 test-only hardening merge", "Gemini thread resolve", "post-merge smoke harness artifact closeout", ] def _load(path): with open(path, "r", encoding="utf-8") as fh: return json.load(fh) def _engine_present(): """C1 engine presence probe — import spec AND file on disk.""" if not os.path.isfile(ENGINE_PATH): return False try: return importlib.util.find_spec(ENGINE_MODULE) is not None except (ImportError, ValueError, ModuleNotFoundError): return False class PhaseABaselineTest(unittest.TestCase): """Engine-independent — always runs.""" @classmethod def setUpClass(cls): cls.cmp = _load(COMPARISON) def test_comparison_artifact_schema(self): self.assertEqual(self.cmp["task"], "task-2553+35") self.assertEqual( self.cmp["task_md_sha256"], "0a0415868cb288a7ad2678022c2c2d3c690f8914bda559eec655cf9acc1cd799", ) self.assertIn("phase_a", self.cmp) self.assertIn("phase_b", self.cmp) def test_phase_a_covers_four_targets(self): got = [t["target"] for t in self.cmp["phase_a"]["targets"]] self.assertEqual(got, TARGETS) def test_phase_a_each_target_has_profile_gate_hold_packet(self): for t in self.cmp["phase_a"]["targets"]: self.assertIn("profile_selection", t, t["target"]) self.assertIn("goal_type", t["profile_selection"], t["target"]) self.assertIn("expected_gate", t, t["target"]) self.assertIn("expected_hold", t, t["target"]) self.assertIn( "expected_completion_packet_schema", t, t["target"] ) self.assertIn( "actual_terminal_status", t["expected_hold"], t["target"] ) def test_phase_a_baseline_matches_actual_result_events(self): """Re-derive the actual terminal status from the source events read-only and assert the baseline encodes it faithfully.""" by_target = { t["target"]: t for t in self.cmp["phase_a"]["targets"] } plus12 = _load(os.path.join(EVENTS, "task-2553+12.result.json")) self.assertEqual(plus12["status"], "HOLD_FOR_CHAIR") self.assertEqual( by_target["PR#128 clean replacement merge"]["expected_hold"][ "actual_terminal_status" ], "HOLD_FOR_CHAIR", ) plus26 = _load(os.path.join(EVENTS, "task-2553+26.result.json")) self.assertEqual(plus26["outcome"], "COMPLETE_MERGED") self.assertFalse( plus26["consolidated_summary_9"]["9_hold_for_chair"] ) self.assertEqual( by_target["PR#129 test-only hardening merge"]["expected_hold"][ "actual_terminal_status" ], "COMPLETE_MERGED", ) plus11 = _load(os.path.join(EVENTS, "task-2553+11.result.json")) self.assertEqual( plus11["final_status"], "RESOLVED_THREAD__MERGE_READY (merge NOT performed; chair packet only)", ) self.assertFalse(plus11["scope_compliance"]["hold_triggered"]) self.assertEqual( by_target["Gemini thread resolve"]["expected_hold"][ "actual_terminal_status" ], "RESOLVED_THREAD__MERGE_READY", ) plus13 = _load(os.path.join(EVENTS, "task-2553+13.result.json")) self.assertEqual(plus13["status"], "DONE") self.assertFalse(plus13["hold"]) self.assertEqual( by_target["post-merge smoke harness artifact closeout"][ "expected_hold" ]["actual_terminal_status"], "DONE", ) def test_dry_run_invariants_zero_real_side_effects(self): inv = self.cmp["dry_run_invariants"] self.assertEqual(inv["real_merge"], 0) self.assertEqual(inv["real_github_write"], 0) self.assertEqual(inv["real_thread_resolve"], 0) self.assertEqual(inv["c1_core_modified"], 0) self.assertEqual(inv["existing_2553_artifacts_modified"], 0) self.assertFalse(inv["contamination"]) def test_source_events_unmodified_read_only(self): """Phase-A is read-only: source events must still parse and keep their terminal markers (no mutation introduced by C3).""" for name, key, val in [ ("task-2553+12.result.json", "status", "HOLD_FOR_CHAIR"), ("task-2553+13.result.json", "status", "DONE"), ]: doc = _load(os.path.join(EVENTS, name)) self.assertEqual(doc[key], val, name) class PhaseBEngineConsumingTest(unittest.TestCase): """Engine-consuming — dry-run only, never real merge/write.""" @classmethod def setUpClass(cls): cls.cmp = _load(COMPARISON) cls.present = _engine_present() def test_phase_b_deferred_when_engine_absent(self): if self.present: self.skipTest("C1 engine present — handled by dry-run test") pb = self.cmp["phase_b"] self.assertEqual(pb["state"], "DEFERRED_PENDING_C1") self.assertFalse(pb["executed"]) self.assertNotEqual(pb["state"], "HOLD_FOR_CHAIR") self.assertTrue(pb["next_action"]) self.assertFalse(self.cmp["hold_for_chair"]) if os.path.isfile(RESULT): res = _load(RESULT) self.assertEqual(res["status"], "DEFERRED_PENDING_C1") self.assertNotIn(res["status"], ("HOLD", "FAIL")) self.assertTrue(res.get("next_action")) def test_engine_dry_run_matches_phase_a_baseline(self): """task-2553+36 TRACK C PHASE-B RECONCILIATION — C3 harness corrected to the C1 *settled* canonical API (task-2553+33 ACCEPT). C1 canonical contract (회장 verbatim, 변경 불가): parse_goal_request(obj, *, schema_dir) -> resolve_policy(goal_request, *, profile_json_dir, ...) -> PolicyResolution The pre-+36 contract (resolve_policy(goal_type=..., boundary=...) echoing a historical *runtime terminal status*) was the documented C1<->C3 API mismatch: the settled engine is a PURE CONTRACT DERIVER. It returns PolicyResolution.status in {RESOLVED, HOLD_FOR_CHAIR} OR fail-closes with PolicyEngineError — it never replays the historical runtime terminal status, and never performs a real merge / GitHub write / thread resolve. Reconciled dry-run contract (in-memory only, zero side effects): * target WITH a real policy_profile instance -> canonical parse_goal_request -> resolve_policy(goal_request, profile_json_dir=...) succeeds; PolicyResolution exposes gate / hold / completion-packet / evidence schema; status in {RESOLVED, HOLD_FOR_CHAIR}. * target WITHOUT a profile instance -> the canonical engine DETERMINISTICALLY fail-closes (PolicyEngineError) — the correct dry-run outcome, NOT an engine defect and NOT a real side effect. * the forbidden API resolve_policy(goal_type=..., boundary=...) MUST raise (regression §6.2). """ if not self.present: self.skipTest( "C1 engine absent — Phase-B DEFERRED_PENDING_C1 (not HOLD/FAIL)" ) # Engine present: load the C1 *canonical* API. Pure / in-memory only — # NO network / merge / GitHub write / thread resolve. engine = importlib.import_module(ENGINE_MODULE) self.assertTrue( hasattr(engine, "parse_goal_request") and hasattr(engine, "resolve_policy"), "C1 settled canonical API parse_goal_request/resolve_policy 부재 " "(contract drift → §9 HOLD)", ) parse_goal_request = engine.parse_goal_request resolve_policy = engine.resolve_policy PolicyEngineError = engine.PolicyEngineError profile_dir = os.path.join(REPO_ROOT, "memory", "policy_profiles") # §6.2 — forbidden legacy API must FAIL (not silently accepted). with self.assertRaises(TypeError): resolve_policy(goal_type="x", boundary={"dry_run": True}) baseline = {t["target"]: t for t in self.cmp["phase_a"]["targets"]} for tgt, base in baseline.items(): sel = base["profile_selection"] profile_name = sel.get("expected_policy_profile") present = bool(sel.get("profile_instance_present_in_repo")) and ( profile_name is not None ) goal_request = { "goal_id": f"dryrun-2553p36-{tgt[:24]}", "goal_statement": f"dry-run application: {tgt}", "policy_profile": {"name": profile_name or "__absent__"}, "boundary": ["dry_run", "no_real_merge", "no_github_write"], } # parse_goal_request -> resolve_policy(goal_request) canonical path. if present: gr = parse_goal_request(goal_request) self.assertEqual(gr["boundary"], goal_request["boundary"], tgt) res = resolve_policy(gr, profile_json_dir=profile_dir) self.assertIn( res.status, ("RESOLVED", "HOLD_FOR_CHAIR"), f"{tgt}: canonical status out of contract domain", ) # PolicyResolution must expose the derived contract surfaces. self.assertTrue(hasattr(res, "gate"), tgt) self.assertTrue(hasattr(res, "hold_conditions"), tgt) self.assertIsInstance(res.completion_packet_schema, dict, tgt) self.assertIsInstance(res.evidence_schema, dict, tgt) # Pure derivation: no privilege leak (allowed ∩ forbidden = ∅). self.assertEqual( sorted(set(res.allowed_actions) & set(res.forbidden_actions)), [], f"{tgt}: allowed∩forbidden 누수 (fail-closed 위반)", ) else: # No profile instance -> deterministic fail-closed, never a # real merge/write, never the historical runtime status. with self.assertRaises(PolicyEngineError) as ctx: resolve_policy( parse_goal_request(goal_request), profile_json_dir=profile_dir, ) self.assertTrue( ctx.exception.code, f"{tgt}: fail-closed PolicyEngineError must carry a code", ) if __name__ == "__main__": unittest.main(verbosity=2)