"""Regression — task-2553+40 TRACK C: 기존 task-2553 사례 policy_profile 인스턴스화 + C1 engine dry-run 검증. 계약 (task-2553+40.md §3 / §5 / §6, +36 settled C1 canonical API): * 4 신규 policy_profile 인스턴스(memory/policy_profiles/task-2553+40.*.json)는 C1 generic meta-schema 로 valid 하며 C1 engine 이 이름만으로 로딩 가능. * C1 canonical API parse_goal_request -> resolve_policy(profile_json_dir=...) 로 각 인스턴스를 dry-run 하면 PolicyResolution.status ∈ {RESOLVED, HOLD_FOR_CHAIR} (PURE CONTRACT DERIVER — 역사적 runtime terminal status 를 재생하지 않음). gate 조건 집합이 +35 Phase-A 기준선(= 기존 실제 결과로부터 read-only 도출, +35 regression 검증) 및 profile 인코딩과 일치. * 실 merge / 실 GitHub write / 실 thread resolve 0 (정적 입증): C1 engine 모듈에 네트워크/merge 호출 토큰 부재 + 비교 artifact invariants=0. * C1 engine byte-0 + 기존 거버넌스 profile byte-0 (전후 sha256 동일). * mismatch → FAIL (엔진 결함 시사 — §6 HOLD 로 에스컬레이션할 신호). """ import ast import hashlib import importlib 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") PROFILE_DIR = os.path.join(REPO_ROOT, "memory", "policy_profiles") COMPARISON = os.path.join(EVENTS, "task-2553+40.dry-run-comparison.json") DECISION = os.path.join(EVENTS, "task-2553+40.decision.json") RESULT = os.path.join(EVENTS, "task-2553+40.result.json") ENGINE_MODULE = "anu_v3.policy_profile_engine" ENGINE_PATH = os.path.join(REPO_ROOT, "anu_v3", "policy_profile_engine.py") GOV_PROFILE = os.path.join(PROFILE_DIR, "test_only_hardening_pr_merge_v1.json") TASK_MD_SHA = "6f96c78ba77f5bd2354fd0658aa4fd3690a13686dc26e4f6db4ac48fc2852ad3" ENGINE_SHA_PIN = "2363e291a0a43884892f5e554f115481a077322bd5caa3000fb75bf5b72bc6be" GOV_PROFILE_SHA_PIN = ( "7e161d7dd579aae025d9c2c202e9f226839dcbbfdea312cb55f624e4a6582a13" ) INSTANCES = { "PR#128 clean replacement merge": "task-2553+40.clean_replacement_pr_merge", "PR#129 test-only hardening merge": "task-2553+40.test_only_hardening_pr_merge", "review thread cleanup": "task-2553+40.review_thread_cleanup", "post-merge smoke harness artifact closeout": ( "task-2553+40.post_merge_smoke_artifact_closeout" ), } GOAL_TYPE = { "PR#128 clean replacement merge": "merge_clean_replacement_pr", "PR#129 test-only hardening merge": "merge_test_only_hardening_pr", "review thread cleanup": "gemini_thread_resolution_limited", "post-merge smoke harness artifact closeout": ( "post_merge_smoke_artifact_closeout" ), } def _load(path): with open(path, "r", encoding="utf-8") as fh: return json.load(fh) def _sha256(path): with open(path, "rb") as fh: return hashlib.sha256(fh.read()).hexdigest() def _engine_present(): 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 ArtifactSchemaTest(unittest.TestCase): """Engine-independent — 비교/decision artifact 무결성.""" @classmethod def setUpClass(cls): cls.cmp = _load(COMPARISON) cls.dec = _load(DECISION) def test_comparison_identity(self): self.assertEqual(self.cmp["task"], "task-2553+40") self.assertEqual(self.cmp["task_md_sha256"], TASK_MD_SHA) self.assertTrue(self.cmp["engine_present"]) self.assertEqual(self.cmp["phase_b"]["state"], "EXECUTED") def test_four_instances_present(self): for name in INSTANCES.values(): p = os.path.join(PROFILE_DIR, name + ".json") self.assertTrue(os.path.isfile(p), p) doc = _load(p) self.assertIn("profile_id", doc) self.assertRegex(doc["version"], r"^v[0-9]+$") self.assertIn("gate_predicate", doc) 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.assertTrue(inv["existing_governance_profile_byte0"]) self.assertFalse(inv["contamination"]) def test_all_targets_match_no_hold(self): self.assertTrue(self.cmp["phase_b"]["all_targets_match_phase_a"]) self.assertEqual(self.cmp["phase_b"]["mismatch_targets"], []) self.assertFalse(self.cmp["hold_for_chair"]) self.assertEqual(self.dec["decision"], "TRACK_C_PASS") def test_engine_byte0_recorded(self): self.assertTrue(self.cmp["engine_byte0"]) self.assertEqual( self.cmp["engine_sha256_before"], self.cmp["engine_sha256_after"] ) class EngineByteZeroTest(unittest.TestCase): """C1 engine + 기존 거버넌스 profile = byte-0 (read-only consume).""" def test_engine_sha_pinned(self): self.assertEqual(_sha256(ENGINE_PATH), ENGINE_SHA_PIN) def test_governance_profile_sha_pinned(self): self.assertEqual(_sha256(GOV_PROFILE), GOV_PROFILE_SHA_PIN) class NoRealMergeStaticTest(unittest.TestCase): """정적 입증 — C1 engine 모듈에 실 merge/network/GitHub write 호출 부재.""" def test_engine_module_has_no_network_or_merge_calls(self): src = open(ENGINE_PATH, "r", encoding="utf-8").read() tree = ast.parse(src) banned = { "requests", "urllib", "httpx", "subprocess", "socket", "github", "Github", "gh", "merge_pull_request", "merge", } bad = [] for node in ast.walk(tree): if isinstance(node, ast.Import): for a in node.names: if a.name.split(".")[0] in banned: bad.append(a.name) elif isinstance(node, ast.ImportFrom): if (node.module or "").split(".")[0] in banned: bad.append(node.module) self.assertEqual( bad, [], f"C1 engine 에 부작용 import 발견(순수성 위반): {bad}" ) # 모듈 docstring 의 순수성 선언 존속. self.assertIn("부작용 0", src) self.assertIn("GitHub API 0", src) class EngineDryRunReproductionTest(unittest.TestCase): """Engine-consuming — dry-run only, 실 merge/write/thread resolve 0.""" @classmethod def setUpClass(cls): cls.present = _engine_present() cls.cmp = _load(COMPARISON) b35 = _load( os.path.join(EVENTS, "task-2553+35.dry-run-comparison.json") ) cls.b35 = {t["target"]: t for t in b35["phase_a"]["targets"]} cls.b35_alias = { "review thread cleanup": "Gemini thread resolve", } _META = { "kind", "actual_ALL_PASS", "actual_failed", "all_keys_independently_proven", "note", } def _baseline_gate_names(self, eg): names = [] for k, v in eg.items(): if k in self._META: continue if isinstance(v, list): names += list(v) elif isinstance(v, bool): names.append(k) return names def test_canonical_dry_run_matches_phase_a_baseline(self): if not self.present: self.skipTest("C1 engine absent") engine = importlib.import_module(ENGINE_MODULE) parse_goal_request = engine.parse_goal_request resolve_policy = engine.resolve_policy PolicyEngineError = engine.PolicyEngineError # forbidden legacy API must FAIL (contract-drift guard). with self.assertRaises(TypeError): resolve_policy(goal_type="x", boundary={"dry_run": True}) for tgt, pname in INSTANCES.items(): b35key = self.b35_alias.get(tgt, tgt) base = self.b35[b35key] goal_request = { "goal_id": f"dryrun-2553p40-{pname}", "goal_statement": f"dry-run application: {tgt}", "goal_type": GOAL_TYPE[tgt], "policy_profile": {"name": pname}, "boundary": [ "dry_run", "no_real_merge", "no_github_write", "no_thread_resolve", ], } gr = parse_goal_request(goal_request) self.assertEqual(gr["boundary"], goal_request["boundary"], tgt) res = resolve_policy(gr, profile_json_dir=PROFILE_DIR) # PURE CONTRACT DERIVER — status 도메인 한정. self.assertIn( res.status, ("RESOLVED", "HOLD_FOR_CHAIR"), f"{tgt}: status out of contract domain ({res.status})", ) # gate 조건: engine == profile 인코딩 == +35 baseline. engine_gate = [g.name for g in res.gate] prof = _load(os.path.join(PROFILE_DIR, pname + ".json")) encoded = list(prof["gate_predicate"].keys()) base_gate = self._baseline_gate_names(base["expected_gate"]) self.assertEqual(engine_gate, encoded, f"{tgt}: gate roundtrip") self.assertEqual( len(engine_gate), len(base_gate), f"{tgt}: gate count vs +35 baseline " f"({len(engine_gate)} != {len(base_gate)})", ) # 권한 누수 0 (allowed ∩ forbidden = ∅, fail-closed). self.assertEqual( sorted(set(res.allowed_actions) & set(res.forbidden_actions)), [], f"{tgt}: allowed∩forbidden 누수", ) # profile allowed/forbidden ⊆ engine 산출. self.assertTrue( set(prof["allowed_actions"]).issubset(set(res.allowed_actions)), f"{tgt}: allowed superset", ) self.assertTrue( set(prof["forbidden_actions"]).issubset( set(res.forbidden_actions) ), f"{tgt}: forbidden superset", ) # packet/evidence schema 해소. self.assertIn("concrete", res.completion_packet_schema, tgt) self.assertIn("concrete", res.evidence_schema, tgt) # 미존재 profile -> deterministic fail-closed (실 부작용 아님). with self.assertRaises(PolicyEngineError) as ctx: resolve_policy( parse_goal_request({ "goal_id": "dryrun-2553p40-absent", "goal_statement": "absent profile fail-closed", "policy_profile": {"name": "__task2553p40_absent__"}, }), profile_json_dir=PROFILE_DIR, ) self.assertTrue(ctx.exception.code) def test_actual_terminal_status_provenance_readonly(self): """source event terminal status 를 독립 재도출 → comparison 충실 인코딩.""" rederive = { "PR#128 clean replacement merge": ( "task-2553+12.result.json", "status", "HOLD_FOR_CHAIR"), "PR#129 test-only hardening merge": ( "task-2553+26.result.json", "outcome", "COMPLETE_MERGED"), "post-merge smoke harness artifact closeout": ( "task-2553+13.result.json", "status", "DONE"), } per = { r["target"]: r for r in self.cmp["phase_b"]["dry_run_results_per_target"] } for tgt, (fn, key, expect) in rederive.items(): doc = _load(os.path.join(EVENTS, fn)) self.assertEqual(doc[key], expect, fn) self.assertEqual( per[tgt]["actual_terminal_status_provenance_readonly"], expect, tgt, ) if __name__ == "__main__": unittest.main(verbosity=2)