# -*- coding: utf-8 -*- """Regression — task-2553+39 TRACK B: batch coordinator ← profile decision output 소비. §3.3 verbatim 매핑 + invariant extras (byte-0 / zero-import-coupling / NO-CRON / emission-guard / no-mutation / git-ref / schema-conformance): 1 정상 소비: valid decision.v1 → coordinator-consumable binding 산출 2 engine 부재 → fail-closed safe (DECISION_UNAVAILABLE, 자동확정 0) 3 engine schema mismatch → fail-closed safe (자동확정 0, settle 불가) 4 engine status HOLD_FOR_CHAIR → HOLD 전파 (자동확정 0) 5 런타임 HOLD trigger 관측 → RUNTIME_HOLD_OBSERVED (자동확정 0) 6 coordinator closeout/merge 자동확정 0 — 모든 입력에서 hard-pinned False 7 engine/coordinator(+29/+30/frozen) byte-0 — binding 소비로 무변 8 +29/+30 public API 무회귀 9 zero import coupling — binding 모듈에 anu_v3 import 0 (순수 stdlib) 10 NO-CRON — cron/cokacdir 토큰 0 11 emission hard-guard — frozen/git-tracked/무관 untracked clobber 0 12 소비한 decision 입력 무mutation 13 decision-logic 파일 I/O side effect 0 14 binding output schema 적합 15 live /home/jay/workspace git tracked HEAD/branch/ref 전후 assertEqual """ from __future__ import annotations import hashlib import json import sys import tempfile import unittest from pathlib import Path _ROOT = Path(__file__).resolve().parents[2] if str(_ROOT) not in sys.path: sys.path.insert(0, str(_ROOT)) from anu_v3.coordinator_profile_binding import ( # noqa: E402 ACCEPTED_DECISION_SCHEMA, BINDING_SCHEMA, CONSUME_HOLD_ENGINE, CONSUME_HOLD_RUNTIME, CONSUME_OK, CONSUME_UNAVAILABLE, CoordinatorProfileBinding, CoordinatorProfileBindingError, FrozenWriteRefused, consume_for_coordinator, emit_binding, load_binding_fixture, load_profile_decision, ) FX = _ROOT / "memory" / "fixtures" FX_RESOLVED = FX / "task-2553+39.profile-decision-resolved.json" FX_HOLD = FX / "task-2553+39.profile-decision-hold.json" FX_RUNTIME_HOLD = FX / "task-2553+39.runtime-hold-observed.json" FX_MISMATCH = FX / "task-2553+39.decision-schema-mismatch.json" SCHEMA_FILE = _ROOT / "schemas" / "coordinator_profile_binding.schema.json" BINDING_MODULE_SRC = _ROOT / "anu_v3" / "coordinator_profile_binding.py" # read-only consume 대상 — binding 소비로 byte-0 무변이어야 한다 (§5). FROZEN = { "engine": _ROOT / "anu_v3" / "policy_profile_engine.py", "parallel_batch_coordinator": _ROOT / "anu_v3" / "parallel_batch_coordinator.py", "generic_batch_coordinator": _ROOT / "anu_v3" / "generic_batch_coordinator.py", "parallel_runtime_registry": _ROOT / "anu_v3" / "parallel_runtime_registry.py", "callback_anchor": _ROOT / "utils" / "anu_delegation_completion_callback.py", } # read-only fixtures — binding 소비로 무mutation (§3.2 / §5). RO_ARTIFACTS = [FX_RESOLVED, FX_HOLD, FX_RUNTIME_HOLD, FX_MISMATCH] def _sha(p: Path) -> str: return hashlib.sha256(p.read_bytes()).hexdigest() def _git_ref(): git_dir = _ROOT / ".git" head_txt = (git_dir / "HEAD").read_text(encoding="utf-8").strip() branch = ( head_txt.split("ref: ", 1)[1] if head_txt.startswith("ref:") else head_txt ) ref_path = git_dir / branch if head_txt.startswith("ref:") else None sha = ( (git_dir / branch).read_text(encoding="utf-8").strip() if ref_path and ref_path.exists() else head_txt ) return (head_txt, branch, sha) class CoordinatorProfileBindingRegression(unittest.TestCase): def setUp(self): self._frozen_pre = {k: _sha(p) for k, p in FROZEN.items() if p.is_file()} self._ro_pre = {p: _sha(p) for p in RO_ARTIFACTS if p.is_file()} self._git_pre = _git_ref() def tearDown(self): self.assertEqual( {k: _sha(p) for k, p in FROZEN.items() if p.is_file()}, self._frozen_pre, "engine/coordinator/anchor byte-0 위반 (§5)", ) self.assertEqual( {p: _sha(p) for p in RO_ARTIFACTS if p.is_file()}, self._ro_pre, "read-only fixture mutated (§3.2/§5)", ) self.assertEqual( _git_ref(), self._git_pre, "live /home/jay/workspace git HEAD/branch/ref 변경 (§5 위반)", ) # 1 def test_01_normal_consumption(self): fx = load_binding_fixture(FX_RESOLVED) binding = consume_for_coordinator( fx.decision, runtime_signals=fx.runtime_signals ) self.assertEqual(binding["binding_schema"], BINDING_SCHEMA) self.assertEqual(binding["consumed_decision_schema"], ACCEPTED_DECISION_SCHEMA) view = binding["track_consumption_view"] # engine decision output 의미 보존 소비 self.assertEqual(len(view["gate_condition_names"]), 8) self.assertEqual(len(view["hold_trigger_conditions"]), 8) self.assertTrue(view["allowed_actions"]) self.assertTrue(view["forbidden_actions"]) self.assertIsNotNone(view["completion_packet_meta_ref"]) self.assertIsNotNone(view["evidence_meta_ref"]) self.assertEqual(binding["consumption_decision"]["signal"], CONSUME_OK) # 2 def test_02_engine_absent_fail_closed(self): missing = _ROOT / "memory" / "fixtures" / "task-2553+39.__nope__.json" self.assertFalse(missing.exists()) binding = consume_for_coordinator(missing) self.assertTrue(binding["fail_closed"]) self.assertEqual(binding["consumption_decision"]["signal"], CONSUME_UNAVAILABLE) self.assertEqual(binding["error_code"], "decision_source_unreadable") # raw loader 는 fail-closed 예외 with self.assertRaises(CoordinatorProfileBindingError): load_profile_decision(missing) # 3 def test_03_schema_mismatch_fail_closed(self): fx = load_binding_fixture(FX_MISMATCH) binding = consume_for_coordinator(fx.decision) self.assertTrue(binding["fail_closed"]) self.assertEqual(binding["error_code"], "decision_schema_mismatch") self.assertEqual(binding["consumption_decision"]["signal"], CONSUME_UNAVAILABLE) # settle 불가 — 자동확정 0 cd = binding["consumption_decision"] self.assertFalse(cd["auto_confirm"]) self.assertFalse(cd["closeout_authority"]) self.assertFalse(cd["merge_authority"]) # 4 def test_04_engine_hold_propagated(self): fx = load_binding_fixture(FX_HOLD) binding = consume_for_coordinator(fx.decision) cd = binding["consumption_decision"] self.assertEqual(cd["signal"], CONSUME_HOLD_ENGINE) self.assertTrue(cd["hold"]["hold"]) self.assertTrue(cd["hold"]["engine_status_hold"]) self.assertFalse(cd["auto_confirm"]) # 5 def test_05_runtime_hold_observed(self): fx = load_binding_fixture(FX_RUNTIME_HOLD) binding = consume_for_coordinator( fx.decision, runtime_signals=fx.runtime_signals ) cd = binding["consumption_decision"] self.assertEqual(cd["signal"], CONSUME_HOLD_RUNTIME) self.assertTrue(cd["hold"]["runtime_fired_triggers"]) self.assertFalse(cd["auto_confirm"]) # 6 — coordinator closeout/merge 자동확정 0 (모든 입력 hard-pinned) def test_06_no_auto_confirm_any_input(self): for src in (FX_RESOLVED, FX_HOLD, FX_RUNTIME_HOLD, FX_MISMATCH): fx = load_binding_fixture(src) b = consume_for_coordinator( fx.decision, runtime_signals=fx.runtime_signals ) view = b["track_consumption_view"] cd = b["consumption_decision"] self.assertEqual(view["coordinator_role"], "decision_consumer_only", src.name) for flag_holder in (view, cd): self.assertFalse(flag_holder["closeout_authority"], src.name) self.assertFalse(flag_holder["merge_authority"], src.name) self.assertFalse(flag_holder["auto_confirm"], src.name) self.assertEqual(cd["authority"], "judgment_assist_only") # RESOLVED + CONSUME_OK 여도 binding 은 절대 확정 권한 신호 0: # merge/closeout/auto_confirm 권한 플래그는 전부 False 로 hard-pin. fx = load_binding_fixture(FX_RESOLVED) cpb = CoordinatorProfileBinding(fx.decision) self.assertEqual(cpb.engine_status(), "RESOLVED") view = cpb.track_consumption_view() cd = cpb.consumption_decision() self.assertEqual(cd["signal"], CONSUME_OK) for k in ("closeout_authority", "merge_authority", "auto_confirm"): self.assertFalse(view[k], f"view.{k} must be hard-pinned False") self.assertFalse(cd[k], f"decision.{k} must be hard-pinned False") # 7 — byte-0 (tearDown 이 검증; 여기선 소비 경로 실행) def test_07_consume_does_not_mutate_frozen(self): for src in RO_ARTIFACTS: fx = load_binding_fixture(src) consume_for_coordinator( fx.decision, runtime_signals=fx.runtime_signals ) # 검증은 tearDown 의 byte-0 / git-ref assertEqual # 8 — +29/+30 public API 무회귀 def test_08_2930_api_intact(self): import anu_v3.parallel_runtime_registry as reg import anu_v3.generic_batch_coordinator as gbc for sym in ("ParallelRuntimeRegistry", "TaskRuntimeRecord"): self.assertTrue(hasattr(reg, sym), f"+29 API 회귀: {sym}") for sym in ("GenericBatchCoordinator", "GenericBatchPlan", "GenericTrackPlan", "emit_generic_batch_state", "load_plan_from_fixture"): self.assertTrue(hasattr(gbc, sym), f"+30 API 회귀: {sym}") # +30 closeout 은 여전히 confirmed=False (대체/격상 0) prop_default = gbc.GenericBatchCoordinator.closeout_proposal self.assertTrue(callable(prop_default)) # 9 — zero import coupling (binding 모듈 anu_v3 import 0) def test_09_zero_import_coupling(self): src = BINDING_MODULE_SRC.read_text(encoding="utf-8") self.assertNotIn("import anu_v3", src, "import 결합 0 위반 (§3.1/§5)") self.assertNotIn("from anu_v3", src, "import 결합 0 위반 (§3.1/§5)") # 실제 import 그래프에도 anu_v3 의존 0 import anu_v3.coordinator_profile_binding as m self.assertEqual(m.__name__, "anu_v3.coordinator_profile_binding") # 10 — NO-CRON def test_10_no_cron_tokens(self): src = BINDING_MODULE_SRC.read_text(encoding="utf-8") for tok in ("--cron", "cron-remove", "cron-register", "--cron-remove", "cokacdir"): self.assertNotIn(tok, src, f"NO-CRON 위반: {tok!r}") # 11 — emission hard-guard def test_11_emission_guard(self): fx = load_binding_fixture(FX_RESOLVED) binding = consume_for_coordinator(fx.decision) # chair durable v1 거부 with self.assertRaises(FrozenWriteRefused): emit_binding(binding, _ROOT / "memory" / "events" / "task-2553.parallel-batch-state.json") # git-tracked 거부 with self.assertRaises(FrozenWriteRefused): emit_binding(binding, _ROOT / "anu_v3" / "policy_profile_engine.py") # 무관 untracked non-deliverable clobber 0 with tempfile.TemporaryDirectory() as d: victim = Path(d) / "unrelated.txt" victim.write_text("ORIGINAL", encoding="utf-8") with self.assertRaises(FrozenWriteRefused): emit_binding(binding, victim) self.assertEqual(victim.read_text(encoding="utf-8"), "ORIGINAL") # NEW untracked deliverable 은 허용 + 재emit idempotent out = Path(d) / "x.coordinator-profile-binding.json" p = emit_binding(binding, out) self.assertTrue(p.is_file()) emit_binding(binding, out) # sanctioned re-emit # 12 — 소비한 decision 입력 무mutation def test_12_input_decision_not_mutated(self): fx = load_binding_fixture(FX_RESOLVED) snap = json.dumps(fx.decision, sort_keys=True) cpb = CoordinatorProfileBinding(fx.decision) cpb.build_binding(fx.runtime_signals) cpb.track_consumption_view() cpb.consumption_decision(fx.runtime_signals) self.assertEqual(json.dumps(fx.decision, sort_keys=True), snap) # 13 — decision-logic 파일 I/O side effect 0 def test_13_decision_logic_no_io(self): fx = load_binding_fixture(FX_RESOLVED) with tempfile.TemporaryDirectory() as d: before = set(Path(d).iterdir()) cpb = CoordinatorProfileBinding(fx.decision) cpb.build_binding() cpb.track_consumption_view() cpb.consumption_decision() cpb.evaluate_hold() consume_for_coordinator(fx.decision) self.assertEqual(set(Path(d).iterdir()), before) # 14 — binding output schema 적합 (+ fail-closed 도 적합) def test_14_schema_conformance(self): import jsonschema schema = json.loads(SCHEMA_FILE.read_text(encoding="utf-8")) for src in (FX_RESOLVED, FX_HOLD, FX_RUNTIME_HOLD, FX_MISMATCH): fx = load_binding_fixture(src) b = consume_for_coordinator( fx.decision, runtime_signals=fx.runtime_signals ) jsonschema.validate(b, schema) # 15 — git-ref invariant (tearDown 검증; 여기서 emission 까지 실행) def test_15_git_ref_invariant_under_emit(self): fx = load_binding_fixture(FX_RESOLVED) b = consume_for_coordinator(fx.decision) with tempfile.TemporaryDirectory() as d: emit_binding(b, Path(d) / "t.coordinator-profile-binding.json") # 검증은 tearDown 의 _git_ref assertEqual if __name__ == "__main__": unittest.main(verbosity=2)