"""Regression — task-2553+36 TRACK C PHASE-B RECONCILIATION. C1 settled engine (task-2553+33 ACCEPT) 을 정본으로, C1↔C2↔C3 interface contract 를 코드로 정합화한다. 회장 verbatim §6 regression 1~10 을 그대로 이행한다. C1 core(anu_v3/policy_profile_engine.py) 무접촉 — byte-0 (§6.8). 정본 API (변경 불가): parse_goal_request(obj, *, schema_dir) -> resolve_policy(goal_request, *, profile_json_dir, ...) -> PolicyResolution 금지 API: resolve_policy(goal_type=..., boundary=...) (§6.2 FAIL) 이 파일은 expected_files allowlist (tests/regression/test_*_2553plus36*.py) 이며 git-untracked batch-internal — tracked HEAD(20456b5f) 무변. """ from __future__ import annotations import importlib 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") FIXTURES = os.path.join(REPO_ROOT, "memory", "fixtures") PROFILE_DIR = os.path.join(REPO_ROOT, "memory", "policy_profiles") ENGINE_PATH = os.path.join(REPO_ROOT, "anu_v3", "policy_profile_engine.py") ENGINE_MODULE = "anu_v3.policy_profile_engine" C1_BASELINE_SHA256 = ( "2363e291a0a43884892f5e554f115481a077322bd5caa3000fb75bf5b72bc6be" ) C1_BASELINE_BYTES = 28063 C2_FIXTURES = [ "task-2553+34.profile-mismatch.json", "task-2553+34.missing-profile.json", "task-2553+34.stale-profile.json", "task-2553+34.forbidden-boundary.json", "task-2553+34.allow-vs-forbid-conflict.json", ] # Phase-A baseline target -> (goal_type, profile_name, profile_instance_present) TARGETS = { "PR#128 clean replacement merge": ( "merge_clean_replacement_pr", "clean_replacement_pr_merge_v1", False, ), "PR#129 test-only hardening merge": ( "merge_test_only_hardening_pr", "test_only_hardening_pr_merge_v1", True, ), "Gemini thread resolve": ( "gemini_thread_resolution_limited", "gemini_thread_resolve_v1", False, ), "post-merge smoke harness artifact closeout": ( "post_merge_smoke_artifact_closeout", "post_merge_artifact_closeout_v1", False, ), } def _load(path): with open(path, "r", encoding="utf-8") as fh: return json.load(fh) def _engine(): return importlib.import_module(ENGINE_MODULE) def _goal_request(name, statement, profile_name, boundary=None): return { "goal_id": name, "goal_statement": statement, "policy_profile": {"name": profile_name}, "boundary": boundary or ["dry_run", "no_real_merge", "no_github_write"], } class R1_CanonicalHappyPath(unittest.TestCase): """§6.1 — parse_goal_request -> resolve_policy(goal_request) 정상.""" def test_canonical_parse_then_resolve(self): eng = _engine() gr = _goal_request( "r1", "canonical path", "test_only_hardening_pr_merge_v1" ) parsed = eng.parse_goal_request(gr) self.assertEqual(parsed["boundary"], gr["boundary"]) res = eng.resolve_policy(parsed, profile_json_dir=PROFILE_DIR) self.assertIn(res.status, ("RESOLVED", "HOLD_FOR_CHAIR")) self.assertTrue(hasattr(res, "gate")) self.assertTrue(hasattr(res, "hold_conditions")) self.assertIsInstance(res.completion_packet_schema, dict) self.assertIsInstance(res.evidence_schema, dict) # resolve_policy 도 내부에서 parse_goal_request 를 호출하는 단일 경로. res2 = eng.resolve_policy(gr, profile_json_dir=PROFILE_DIR) self.assertEqual(res2.status, res.status) class R2_ForbiddenApiFails(unittest.TestCase): """§6.2 — resolve_policy(goal_type=..., boundary=...) 사용 시 FAIL.""" def test_forbidden_kwargs_raise(self): eng = _engine() with self.assertRaises(TypeError): eng.resolve_policy(goal_type="merge", boundary={"dry_run": True}) def test_forbidden_positional_misuse_fail_closed(self): eng = _engine() # goal_type 문자열을 goal_request 위치에 — fail-closed PolicyEngineError. with self.assertRaises(eng.PolicyEngineError): eng.resolve_policy("merge_test_only_hardening_pr") class R3_C2AdversarialReRunVsSettledEngine(unittest.TestCase): """§6.3 — C2 adversarial fixtures 가 C1 settled engine 기준으로 재실행. Reconciled C1-canonical invariant: 모든 adversarial 입력에 대해 권한확대 0 — hard fail-closed(PolicyEngineError) 또는 allowed∩forbidden=∅. """ def test_all_c2_fixtures_no_privilege_expansion(self): eng = _engine() results = {} for fname in C2_FIXTURES: data = _load(os.path.join(FIXTURES, fname)) gr = data["input"]["goal_request"] try: res = eng.resolve_policy(gr, profile_json_dir=PROFILE_DIR) leak = sorted( set(res.allowed_actions) & set(res.forbidden_actions) ) self.assertEqual( leak, [], f"{fname}: 권한 누수 (fail-closed 위반)" ) results[fname] = f"{res.status}/leak=[]" except eng.PolicyEngineError as exc: self.assertTrue(exc.code, f"{fname}: error code 부재") results[fname] = f"FAIL-CLOSED:{exc.code}" # 5 fixtures 전부 평가됨 (vacuous 아님). self.assertEqual(len(results), len(C2_FIXTURES)) # 최소 하나는 hard fail-closed (missing/stale) 여야 — 비공허 보증. self.assertTrue( any(v.startswith("FAIL-CLOSED") for v in results.values()), results, ) class R4to7_DryRunApplicationReRun(unittest.TestCase): """§6.4~6.7 — 4 dry-run application targets 재실행 (실 side-effect 0). settled engine = contract deriver: profile instance 有 -> 결정론적 RESOLVED/HOLD_FOR_CHAIR, profile instance 無 -> 결정론적 fail-closed. 어떤 경우에도 실 merge / GitHub write / thread resolve 0. """ def _dry_run(self, target): eng = _engine() profile_name, present = TARGETS[target][1], TARGETS[target][2] gr = _goal_request(f"dry-{target[:18]}", target, profile_name) if present: res = eng.resolve_policy(gr, profile_json_dir=PROFILE_DIR) self.assertIn(res.status, ("RESOLVED", "HOLD_FOR_CHAIR"), target) self.assertEqual( sorted(set(res.allowed_actions) & set(res.forbidden_actions)), [], target, ) return ("RESOLVED_CONTRACT", res.status) with self.assertRaises(eng.PolicyEngineError) as ctx: eng.resolve_policy(gr, profile_json_dir=PROFILE_DIR) self.assertTrue(ctx.exception.code, target) return ("FAIL_CLOSED", ctx.exception.code) def test_r4_pr128_clean_replacement(self): kind, _ = self._dry_run("PR#128 clean replacement merge") self.assertEqual(kind, "FAIL_CLOSED") def test_r5_pr129_test_only_hardening(self): kind, status = self._dry_run("PR#129 test-only hardening merge") self.assertEqual(kind, "RESOLVED_CONTRACT") self.assertIn(status, ("RESOLVED", "HOLD_FOR_CHAIR")) def test_r6_gemini_thread_resolve(self): kind, _ = self._dry_run("Gemini thread resolve") self.assertEqual(kind, "FAIL_CLOSED") def test_r7_post_merge_smoke_closeout(self): kind, _ = self._dry_run( "post-merge smoke harness artifact closeout" ) self.assertEqual(kind, "FAIL_CLOSED") def test_no_real_side_effect_engine_is_pure(self): """엔진 모듈 소스에 GitHub/merge/network/file-write 부작용 토큰 부재.""" with open(ENGINE_PATH, "r", encoding="utf-8") as fh: src = fh.read() for bad in ( "requests.post", "subprocess.", "gh pr merge", "open(", "urllib.request", ): self.assertNotIn(bad, src, f"엔진 부작용 토큰 발견: {bad}") class R8_C1CoreByte0(unittest.TestCase): """§6.8 — C1 core byte-0 (전후 sha256 동일).""" def test_engine_sha256_and_size_unchanged(self): with open(ENGINE_PATH, "rb") as fh: blob = fh.read() import hashlib self.assertEqual(len(blob), C1_BASELINE_BYTES) self.assertEqual( hashlib.sha256(blob).hexdigest(), C1_BASELINE_SHA256 ) class R9_CrossTrackContaminationZero(unittest.TestCase): """§6.9 — C2/C3 cross-track contamination 0 (namespace 단독 소유).""" def test_c2_c3_originals_unmodified_additive_only(self): # +34/+35 원본 result/decision/comparison 무변 — 본 task 는 phase-b* # additive 파일만 신규 생성. 원본 파서·terminal marker 보존. c35 = _load( os.path.join(EVENTS, "task-2553+35.dry-run-comparison.json") ) self.assertEqual(c35["task"], "task-2553+35") r34 = _load(os.path.join(EVENTS, "task-2553+34.result.json")) self.assertEqual(r34["status"], "DEFERRED_PENDING_C1") r35 = _load(os.path.join(EVENTS, "task-2553+35.result.json")) self.assertEqual(r35["status"], "DEFERRED_PENDING_C1") def test_plus36_namespace_single_owner(self): ic = _load( os.path.join(FIXTURES, "task-2553+36.interface-contract.json") ) self.assertEqual(ic["ownership"]["c2_c3_crosswrite"], 0) self.assertFalse(ic["ownership"]["contamination"]) self.assertEqual(ic["task"], "task-2553+36") class R10_TrackCConsolidatedPassHold(unittest.TestCase): """§6.10 — Track C consolidated PASS/HOLD 산출.""" def test_consolidated_result_present_and_decidable(self): path = os.path.join(EVENTS, "task-2553+36.result.json") self.assertTrue(os.path.isfile(path), "consolidated result 부재") res = _load(path) self.assertEqual(res["task"], "task-2553+36") self.assertIn(res["track_c_consolidated"], ("PASS", "HOLD")) self.assertIn("hold_for_chair", res) self.assertIn("regression", res) if __name__ == "__main__": unittest.main(verbosity=2)