"""tests/regression/test_dispatch_profile_selection_2553plus38.py task-2553+38 — TRACK A: profile engine → dispatch selection 연결 regression. Spec: memory/tasks/task-2553+38.md (sha256 643ae31b0d12d225603e9e81fcd961023dcbe4ced41d88908271f014d651c313). §3 검증: 1. 정상 selection (SELECTED, profile auto-bound, engine read-only 소비) 2. profile 부재·mismatch → 안전 거부(SELECTION_REFUSED, 자동 적용 0) 3. engine HOLD → HOLD_FOR_CHAIR 전파 (자동 적용 0, fail-closed) 4. C1 engine byte-0 (anu_v3/policy_profile_engine.py sha256 무변) 5. +33 정본 API(parse_goal_request / resolve_policy) 무회귀 (직접 호출) 6. dispatch lifecycle 무파괴 (seam 순수 — 파일 write 0, profile mutation 0, 예외 비전파, 멱등) 7. seam 산출이 dispatch_profile_selection.schema.json 충족 8. fixture 전 케이스 expect 일치 9. live /home/jay/workspace git tracked HEAD/branch/ref 전후 assertEqual (§5 repo root 기준 — 신규 untracked 산물은 위반 아님) 모든 테스트 100% offline — network / git mutation / GitHub API 호출 0. """ from __future__ import annotations import hashlib import json import subprocess import sys from pathlib import Path import pytest WORKSPACE = Path(__file__).resolve().parent.parent.parent if str(WORKSPACE) in sys.path: sys.path.remove(str(WORKSPACE)) sys.path.insert(0, str(WORKSPACE)) from anu_v3.dispatch_profile_selection import ( # noqa: E402 # pyright: ignore[reportMissingImports] SELECTION_SCHEMA_ID, DispatchSelectionRequest, build_selection_request, run_dispatch_profile_selection, select_profile_for_dispatch, ) from anu_v3.policy_profile_engine import ( # noqa: E402 # pyright: ignore[reportMissingImports] PolicyEngineError, PolicyResolution, parse_goal_request, resolve_policy, validate_against_meta, ) SCHEMA_DIR = WORKSPACE / "schemas" PROFILE_JSON_DIR = WORKSPACE / "memory" / "policy_profiles" PROFILE_SCHEMA_DIR = WORKSPACE / "schemas" / "policy_profiles" ENGINE_PY = WORKSPACE / "anu_v3" / "policy_profile_engine.py" PROFILE_JSON = PROFILE_JSON_DIR / "test_only_hardening_pr_merge_v1.json" FIXTURE = WORKSPACE / "memory" / "fixtures" / "task-2553+38.cases.json" SELECTION_SCHEMA = SCHEMA_DIR / "dispatch_profile_selection.schema.json" # byte-0 baseline (task md §5 / §6 — engine 무수정 입증). ENGINE_SHA256_BASELINE = ( "2363e291a0a43884892f5e554f115481a077322bd5caa3000fb75bf5b72bc6be" ) PROFILE_SHA256_BASELINE = ( "7e161d7dd579aae025d9c2c202e9f226839dcbbfdea312cb55f624e4a6582a13" ) SANCTIONED_HEAD = "20456b5f83fc039f2fd6f50f4b94095c29b41bfb" SANCTIONED_BRANCH = "task/task-2553p1-f1-clean-replacement" def _sha256(p: Path) -> str: return hashlib.sha256(p.read_bytes()).hexdigest() def _git(*args: str) -> str: return subprocess.run( ["git", "-C", str(WORKSPACE), *args], capture_output=True, text=True, check=True, ).stdout.strip() def _dirs() -> dict: return { "profile_json_dir": PROFILE_JSON_DIR, "profile_schema_dir": PROFILE_SCHEMA_DIR, "schema_dir": SCHEMA_DIR, } @pytest.fixture(scope="module") def cases(): return json.loads(FIXTURE.read_text(encoding="utf-8"))["cases"] # --- 1. 정상 selection ------------------------------------------------------- def test_normal_selection_binds_profile(): sel = select_profile_for_dispatch( DispatchSelectionRequest( goal_id="t1", goal_statement="connect engine to dispatch", policy_profile_name="test_only_hardening_pr_merge_v1", ), **_dirs(), ) assert sel.status == "SELECTED" assert sel.profile_bound is True assert sel.auto_apply is True assert sel.profile_id == "test_only_hardening_pr_merge_v1" assert sel.gate_condition_names # gate 존재 assert sel.dispatch_lifecycle_effect == "none" b = sel.selection_binding() assert b["profile_id"] == "test_only_hardening_pr_merge_v1" assert b["status"] == "SELECTED" def test_selection_binding_matches_engine_resolution(): """seam binding == engine 정본 어댑터(to_coordinator_binding) 동치 (engine read-only 소비 증거).""" gr = { "goal_id": "t2", "goal_statement": "x", "policy_profile": {"name": "test_only_hardening_pr_merge_v1"}, } res = resolve_policy(gr, **_dirs()) # 정본 API 직접 eng = res.to_coordinator_binding() sel = run_dispatch_profile_selection(gr, **_dirs()) assert sel["status"] == "SELECTED" assert sel["gate_condition_names"] == eng["gate_condition_names"] assert sel["hold_trigger_conditions"] == eng["hold_trigger_conditions"] assert sel["allowed_actions"] == eng["allowed_actions"] assert sel["forbidden_actions"] == eng["forbidden_actions"] assert sel["completion_packet_meta_ref"] == eng["completion_packet_meta_ref"] # --- 2. 안전 거부 (profile 부재 / mismatch) ---------------------------------- @pytest.mark.parametrize( "raw,code", [ ({"goal_id": "a", "goal_statement": "x", "policy_profile": {"name": "no_such_profile_xyz"}}, "profile_load_fail"), (42, "selection_request_not_mapping"), ({"goal_id": "a", "goal_statement": "x", "policy_profile": {}}, "selection_policy_profile_name_missing"), ({"goal_id": "", "goal_statement": "x", "policy_profile": {"name": "test_only_hardening_pr_merge_v1"}}, "goal_request_schema_fail"), ], ) def test_absent_or_mismatch_is_safe_refusal(raw, code): d = run_dispatch_profile_selection(raw, **_dirs()) assert d["status"] == "SELECTION_REFUSED" assert d["profile_bound"] is False assert d["auto_apply"] is False # 자동 적용 0 assert d["refusal_code"] == code assert d["dispatch_lifecycle_effect"] == "none" def test_refusal_does_not_raise_into_dispatch(): """seam 은 engine 예외를 dispatch lifecycle 로 전파하지 않는다.""" try: run_dispatch_profile_selection( {"goal_id": "a", "goal_statement": "x", "policy_profile": {"name": "no_such_profile_xyz"}}, **_dirs()) except Exception as e: # pragma: no cover pytest.fail(f"seam leaked exception into dispatch: {e!r}") # --- 3. engine HOLD 전파 ----------------------------------------------------- def test_engine_hold_propagated_no_auto_apply(): d = run_dispatch_profile_selection( { "goal_id": "h1", "goal_statement": "boundary deny contradicts allowed", "policy_profile": {"name": "test_only_hardening_pr_merge_v1"}, "boundary": ["forbid:gate_evaluate"], }, **_dirs(), ) assert d["status"] == "HOLD_FOR_CHAIR" assert d["profile_bound"] is False assert d["auto_apply"] is False assert d["refusal_code"] == "engine_hold_for_chair" assert d["engine_decision"]["status"] == "HOLD_FOR_CHAIR" # --- 4. engine byte-0 -------------------------------------------------------- def test_engine_byte0_unchanged(): assert _sha256(ENGINE_PY) == ENGINE_SHA256_BASELINE def test_profile_json_unchanged_after_selection(): before = _sha256(PROFILE_JSON) for _ in range(3): run_dispatch_profile_selection( {"goal_id": "p", "goal_statement": "x", "policy_profile": {"name": "test_only_hardening_pr_merge_v1"}}, **_dirs()) assert _sha256(PROFILE_JSON) == before == PROFILE_SHA256_BASELINE # --- 5. +33 정본 API 무회귀 -------------------------------------------------- def test_canonical_engine_api_no_regression(): gr = parse_goal_request( {"goal_id": "r", "goal_statement": "x", "policy_profile": {"name": "test_only_hardening_pr_merge_v1"}}, schema_dir=SCHEMA_DIR, ) assert gr["goal_id"] == "r" res = resolve_policy(gr, **_dirs()) assert isinstance(res, PolicyResolution) assert res.status == "RESOLVED" with pytest.raises(PolicyEngineError): resolve_policy( {"goal_id": "r", "goal_statement": "x", "policy_profile": {"name": "no_such_profile_xyz"}}, **_dirs()) # --- 6. dispatch lifecycle 무파괴 (순수·멱등) -------------------------------- def test_seam_is_pure_and_idempotent(): raw = {"goal_id": "i", "goal_statement": "x", "policy_profile": {"name": "test_only_hardening_pr_merge_v1"}} a = run_dispatch_profile_selection(raw, **_dirs()) b = run_dispatch_profile_selection(raw, **_dirs()) assert a == b assert raw == {"goal_id": "i", "goal_statement": "x", "policy_profile": {"name": "test_only_hardening_pr_merge_v1"}} def test_build_selection_request_normalizes(): req = build_selection_request( {"goal_id": "n", "goal_statement": "x", "goal_type": " custom ", "boundary": ["a", 1], "policy_profile": {"name": "p"}}) assert req.goal_type == "custom" assert req.boundary == ["a", "1"] assert req.to_goal_request()["policy_profile"] == {"name": "p"} # --- 7. schema 충족 ---------------------------------------------------------- def test_selection_output_satisfies_schema(cases): schema = json.loads(SELECTION_SCHEMA.read_text(encoding="utf-8")) sel_schema = schema["properties"]["selection"] for c in cases: d = run_dispatch_profile_selection(c["request"], **_dirs()) errs = validate_against_meta(d, sel_schema) assert not errs, f"{c['name']}: schema errs {errs}" assert d["schema"] == SELECTION_SCHEMA_ID # --- 8. fixture 전 케이스 ---------------------------------------------------- def test_all_fixture_cases(cases): assert len(cases) == 6 for c in cases: d = run_dispatch_profile_selection(c["request"], **_dirs()) for k, v in c["expect"].items(): assert d[k] == v, f"{c['name']}.{k}: {d[k]!r} != {v!r}" # --- 9. live repo HEAD/branch/ref 전후 동일 ---------------------------------- def test_live_repo_head_branch_ref_unchanged(): head_before = _git("rev-parse", "HEAD") branch_before = _git("rev-parse", "--abbrev-ref", "HEAD") # 본 task 산물은 전부 신규 (untracked) — selection seam 재실행. run_dispatch_profile_selection( {"goal_id": "z", "goal_statement": "x", "policy_profile": {"name": "test_only_hardening_pr_merge_v1"}}, **_dirs()) assert _git("rev-parse", "HEAD") == head_before == SANCTIONED_HEAD assert _git("rev-parse", "--abbrev-ref", "HEAD") == branch_before == SANCTIONED_BRANCH tracked = subprocess.run( ["git", "-C", str(WORKSPACE), "ls-files", "anu_v3/dispatch_profile_selection.py", "schemas/dispatch_profile_selection.schema.json", "tests/regression/test_dispatch_profile_selection_2553plus38.py"], capture_output=True, text=True, check=True).stdout.strip() assert tracked == "", f"seam 산물이 tracked 됨(위반): {tracked!r}"