# -*- coding: utf-8 -*- """Regression — task-2553+42 STEP 2 profile engine operational adoption planner. 증명 대상 (회장 §3.4 / 9-R.1 Layer A): - plan 산출 정상 (expected_files allowlist / conflict / risk tier) - seam(+38) · binding(+39) · engine(C1) byte-0 (planner 사용 전후 hash EQUAL) - callback / collector 경로 무수정 (frozen 83b3e307 byte-0, source 무접촉) - 실 write · merge · cron · PR 0 / dry-run 부작용 0 (applied=0) - frozen anchor byte-0 / git tracked HEAD·branch 전후 EQUAL - emit hard-guard: frozen/tracked write REFUSE Layer A 한정 (9-R.1): executor 의 §8 normal completion callback 발사는 검증 대상 아님 (별개 lifecycle 신호). 본 테스트는 산출 모듈의 callback/ collector 소스 무수정 · 실 adoption 0 · dry-run 부작용 0 만 증명. """ from __future__ import annotations import hashlib import json import subprocess import unittest from pathlib import Path import anu_v3.coordinator_profile_binding as cpb import anu_v3.dispatch_profile_selection as dps import anu_v3.policy_profile_engine as ppe import anu_v3.profile_adoption_planner as planner REPO = Path(__file__).resolve().parents[2] FIXTURE = REPO / "memory" / "fixtures" / "task-2553+42.json" SCHEMA = REPO / "schemas" / "profile_adoption_plan.schema.json" # §5 frozen / byte-0 anchor (실 파일 경로). _BYTE0 = [ "anu_v3/policy_profile_engine.py", "anu_v3/dispatch_profile_selection.py", "anu_v3/coordinator_profile_binding.py", "anu_v3/parallel_batch_coordinator.py", "utils/anu_delegation_completion_callback.py", ] _CALLBACK_FROZEN_SHA = ( "83b3e307c8207c76a3e311c408aab4951373bd317896e51687d3007907b0c3d4" ) def _sha(p: Path) -> str: return hashlib.sha256(p.read_bytes()).hexdigest() class ProfileAdoptionPlanner2553Plus42(unittest.TestCase): def setUp(self) -> None: self.fixture = json.loads(FIXTURE.read_text(encoding="utf-8")) self.inv = self.fixture["expected_invariants"] self._base_hashes = { rel: _sha(REPO / rel) for rel in _BYTE0 if (REPO / rel).exists() } # ── plan 산출 정상 ────────────────────────────────────────────── def test_plan_built_and_well_formed(self) -> None: plan = planner.build_adoption_plan() self.assertEqual(plan["plan_schema"], planner.PLAN_SCHEMA) self.assertEqual(plan["task"], "task-2553+42") self.assertEqual(plan["adoption_lifecycle_effect"], "none") self.assertEqual(plan["real_in_place_adoption_count"], 0) self.assertGreaterEqual( len(plan["touchpoints"]), self.inv["min_touchpoints"] ) ids = [t["touchpoint_id"] for t in plan["touchpoints"]] self.assertEqual(ids, self.inv["expected_touchpoint_ids"]) def test_expected_files_allowlist_matches_spec(self) -> None: plan = planner.build_adoption_plan() allow = plan["expected_files_allowlist"] for required in ( "anu_v3/profile_adoption_planner.py", "schemas/profile_adoption_plan.schema.json", "tests/regression/test_profile_adoption_planner_2553plus42.py", "memory/events/task-2553+42.adoption-plan.json", "memory/reports/task-2553+42.md", ): self.assertIn(required, allow) def test_conflict_set_flags_frozen_parallel_coordinator(self) -> None: plan = planner.build_adoption_plan() conf = plan["conflict"] self.assertGreaterEqual( conf["conflict_count"], self.fixture["cases"][0]["expect_conflict_count_min"] ) cands = [c["candidate"] for c in conf["frozen_anchor_conflicts"]] self.assertIn(self.inv["expected_blocked_frozen_route"], cands) def test_risk_tier_overall_high(self) -> None: plan = planner.build_adoption_plan() self.assertEqual( plan["risk"]["overall_tier"], self.inv["overall_risk_tier"] ) per = plan["risk"]["per_touchpoint"] self.assertEqual(per["TE.engine_decision_emit"], "LOW") self.assertEqual(per["TA.dispatch_selection_wire"], "MED") self.assertEqual(per["TB.coordinator_binding_consume"], "HIGH") def test_schema_validates_plan_and_dry_run(self) -> None: meta = json.loads(SCHEMA.read_text(encoding="utf-8")) plan = planner.build_adoption_plan() self.assertEqual(ppe.validate_against_meta(plan, meta), []) dry = planner.dry_run_adoption(plan) self.assertEqual( ppe.validate_against_meta(dry, meta["definitions"]["dry_run"]), [] ) # ── dry-run 부작용 0 / 실 write·merge·cron·PR 0 ──────────────── def test_dry_run_zero_side_effects(self) -> None: dry = planner.dry_run_adoption() self.assertEqual(dry["mode"], "READ_ONLY_SIMULATION") self.assertEqual(dry["applied_count"], 0) self.assertEqual(dry["writes"], 0) self.assertEqual(dry["merges"], 0) self.assertEqual(dry["cron_ops"], 0) self.assertEqual(dry["pr_ops"], 0) self.assertFalse(dry["callback_collector_touched"]) self.assertFalse(dry["frozen_anchor_touched"]) for d in dry["simulated_diffs"]: self.assertFalse(d["applied"]) blocked = [b["candidate"] for b in dry["blocked_frozen_routes"]] self.assertIn(self.inv["expected_blocked_frozen_route"], blocked) def test_bundle_entrypoint_pure(self) -> None: b = planner.run_adoption_planner() self.assertFalse(b["real_in_place_adoption"]) self.assertTrue(b["callback_collector_untouched"]) # ── seam/engine/coordinator/callback byte-0 (전후 EQUAL) ─────── def test_byte0_anchors_unchanged_after_planner_use(self) -> None: planner.run_adoption_planner() # read-only consume planner.introspect_seams() for rel, base in self._base_hashes.items(): self.assertEqual(_sha(REPO / rel), base, f"byte-0 깨짐: {rel}") def test_callback_frozen_anchor_sha(self) -> None: cb = REPO / "utils" / "anu_delegation_completion_callback.py" self.assertEqual(_sha(cb), _CALLBACK_FROZEN_SHA) def test_planner_does_not_reference_callback_collector_source(self) -> None: src = (REPO / "anu_v3" / "profile_adoption_planner.py").read_text( encoding="utf-8" ) # callback/collector 경로를 import 하거나 읽지 않는다 (Layer A 무접촉). self.assertNotIn("import utils.anu_delegation_completion_callback", src) self.assertNotIn("anu_delegation_completion_callback import", src) self.assertNotIn("read_text", src.split("def emit_adoption_plan")[0] .split("def load_planner_fixture")[0]) # ── git tracked HEAD / branch 전후 EQUAL ─────────────────────── def test_git_head_and_branch_unchanged(self) -> None: head = subprocess.run( ["git", "-C", str(REPO), "rev-parse", "HEAD"], capture_output=True, text=True, ).stdout.strip() branch = subprocess.run( ["git", "-C", str(REPO), "branch", "--show-current"], capture_output=True, text=True, ).stdout.strip() self.assertEqual(head, "20456b5f83fc039f2fd6f50f4b94095c29b41bfb") self.assertEqual(branch, "task/task-2553p1-f1-clean-replacement") # 신규 산출물은 untracked 여야 함 (tracked 코드 무변). tracked = subprocess.run( ["git", "-C", str(REPO), "ls-files", "anu_v3/profile_adoption_planner.py"], capture_output=True, text=True, ).stdout.strip() self.assertEqual(tracked, "") # ── emit hard-guard (decision 경계 밖 — frozen/tracked REFUSE) ── def test_emit_refuses_frozen_durable_v1(self) -> None: with self.assertRaises(planner.FrozenWriteRefused): planner.emit_adoption_plan({}, "task-2553.parallel-batch-state.json") def test_emit_refuses_frozen_anchor_basename(self) -> None: with self.assertRaises(planner.FrozenWriteRefused): planner.emit_adoption_plan({}, "policy_profile_engine.py") def test_emit_refuses_git_tracked_path(self) -> None: # AGENTS.md 류 tracked 파일 write 거부 (tracked HEAD invariant). tracked = subprocess.run( ["git", "-C", str(REPO), "ls-files"], capture_output=True, text=True, ).stdout.splitlines() if tracked: with self.assertRaises(planner.FrozenWriteRefused): planner.emit_adoption_plan({}, REPO / tracked[0]) # ── Codex 수렴 회귀 (c/e/f fix) ──────────────────────────────── def test_emit_refuses_callback_collector_path(self) -> None: for cc in planner.CALLBACK_COLLECTOR_PATHS: with self.assertRaises(planner.FrozenWriteRefused): planner.emit_adoption_plan({}, cc) with self.assertRaises(planner.FrozenWriteRefused): planner.emit_adoption_plan({}, REPO / cc) def test_conflict_count_deduped_no_double_count(self) -> None: # callback ∩ frozen 인 candidate 를 동시 보유한 합성 touchpoint — # conflict_count 가 distinct route 로만 계수되는지 검증. tp = planner.AdoptionTouchpoint( touchpoint_id="SYN.dup", track="X", seam_module="m", seam_entrypoint="e", consumes="c", live_target_candidates=[ "utils/anu_delegation_completion_callback.py" # frozen ∩ callback ], adoption_kind="k", in_place_edit_required=True, touches_frozen_anchor=True, touches_callback_collector=True, risk_tier=planner.RISK_HIGH, rationale="r", ) conf = planner.classify_conflicts([tp]) self.assertEqual(conf["conflict_count"], 1) def test_dry_run_callback_route_blocked_not_simulated(self) -> None: tp = planner.AdoptionTouchpoint( touchpoint_id="SYN.cb", track="X", seam_module="m", seam_entrypoint="e", consumes="c", live_target_candidates=["anu_v3/executor_callback_contract.py"], adoption_kind="k", in_place_edit_required=True, touches_frozen_anchor=False, touches_callback_collector=True, risk_tier=planner.RISK_HIGH, rationale="r", ) plan = planner.build_adoption_plan() plan["touchpoints"] = [tp.to_dict()] dry = planner.dry_run_adoption(plan) self.assertFalse(dry["callback_collector_touched"]) cb_blocked = [b["candidate"] for b in dry["blocked_callback_routes"]] self.assertIn("anu_v3/executor_callback_contract.py", cb_blocked) # callback route 는 simulated diff 에 +call-site 로 표현되지 않음. for d in dry["simulated_diffs"]: self.assertNotIn("executor_callback_contract", d["change"]) self.assertFalse(d["applied"]) # ── seam/engine API 미러 정합성 (read-only introspection) ────── def test_introspection_mirrors_seam_contracts(self) -> None: intro = planner.introspect_seams() self.assertEqual(intro["dispatch_seam"]["module"], dps.SEAM_MODULE) self.assertEqual( intro["dispatch_seam"]["lifecycle_effect"], dps.DISPATCH_LIFECYCLE_EFFECT, ) self.assertEqual( intro["coordinator_binding_seam"]["binding_schema"], cpb.BINDING_SCHEMA, ) self.assertEqual(intro["engine"]["version"], ppe.ENGINE_VERSION) if __name__ == "__main__": unittest.main(verbosity=2)