"""tests/regression/test_default_profile_resolver_2553plus52.py task-2553+52 — TRACK 3: policy profile DEFAULT DISPATCH ADOPTION regression. Spec: memory/tasks/task-2553+52.md (sha256 41029c90209afb982f4a7f2689d741ffb2b2ec44860afbc31287773ec7726f9f). §3 검증 (1~9): 1. default profile resolver — goal_type + boundary 만으로 profile 선택 (chair 가 policy_profile.name 안 줘도 RESOLVED, auto-bound) 2. goal_type → policy_profile mapping 단일 결정 테이블 조회 3. boundary → gate/HOLD/forbidden expansion 표면화 (engine read-only) 4. profile decision → dispatch planning adapter (plan_only hard-pinned) 5. profile decision → batch coordinator input adapter (closeout/merge/ auto_confirm hard-pinned False) 6. missing / unknown profile fail-closed (추측·날조 0) 7. profile conflict fail-closed (explicit 충돌 + engine allowed∩forbidden) 8. selected profile evidence JSON 산출 + schema 충족 9. C1 engine byte-0 + +38 seam byte-0 + +39 binding byte-0 + 기존 profile/anchor 무변, live repo HEAD/branch 전후 동일, 신규 산물 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.default_profile_resolver import ( # noqa: E402 # pyright: ignore[reportMissingImports] EVIDENCE_SCHEMA_ID, RESOLUTION_SCHEMA_ID, DefaultDispatchRequest, build_default_request, load_goal_type_mapping, resolve_default_profile, resolve_profile_name, run_default_profile_resolution, run_selected_profile_evidence, ) from anu_v3.profile_coordinator_input_adapter import ( # noqa: E402 # pyright: ignore[reportMissingImports] adapt_for_coordinator_input, ) from anu_v3.profile_dispatch_planning_adapter import ( # noqa: E402 # pyright: ignore[reportMissingImports] adapt_for_dispatch_planning, ) 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" MAPPING_PATH = PROFILE_JSON_DIR / "goal_type_profile_mapping.json" ENGINE_PY = WORKSPACE / "anu_v3" / "policy_profile_engine.py" DPS_PY = WORKSPACE / "anu_v3" / "dispatch_profile_selection.py" CPB_PY = WORKSPACE / "anu_v3" / "coordinator_profile_binding.py" EXISTING_PROFILE = PROFILE_JSON_DIR / "test_only_hardening_pr_merge_v1.json" FIXTURE = WORKSPACE / "memory" / "fixtures" / "task-2553plus52.cases.json" RES_SCHEMA = SCHEMA_DIR / "default_profile_resolution.schema.json" MAP_SCHEMA = SCHEMA_DIR / "goal_type_profile_mapping.schema.json" # byte-0 baseline (§5/§6 — engine·+38·+39 무수정 입증). ENGINE_SHA256_BASELINE = ( "2363e291a0a43884892f5e554f115481a077322bd5caa3000fb75bf5b72bc6be" ) DPS_SHA256_BASELINE = ( "54d47cd013b62343dca98cfb946eae1608560102da24ed708a3ce641f1ce1c1a" ) CPB_SHA256_BASELINE = ( "ee6341040cb8b8fd60d341c79253b242ad2762e0c71b7855474ba40826d20c19" ) EXISTING_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 { "mapping_path": MAPPING_PATH, "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. default resolver: goal_type + boundary 만으로 profile 선택 ---------- def test_default_path_goal_type_plus_boundary_only(): """chair 가 policy_profile.name 을 주지 않아도 (goal_type + boundary 만으로) profile 이 자동 선택·바인딩된다 (회장 §1 핵심).""" res = resolve_default_profile( DefaultDispatchRequest( goal_id="t1", goal_statement="default path", goal_type="task_2553_final_closeout", boundary=["forbid:production_write"], ), **_dirs(), ) assert res.status == "RESOLVED" assert res.profile_bound is True assert res.auto_apply is True assert res.default_path is True assert res.resolved_profile_name == "task_2553_final_closeout" assert res.profile_id == "task_2553_final_closeout_v1" assert res.dispatch_lifecycle_effect == "none" assert res.boundary_expansion["gate_condition_names"] # gate 존재 # --- 2. goal_type → policy_profile mapping 단일 결정 테이블 ------------------ def test_mapping_is_deterministic_single_table(): mapping = load_goal_type_mapping( mapping_path=MAPPING_PATH, schema_dir=SCHEMA_DIR ) assert mapping["default_path"] is True for gt in ( "task_2553_final_closeout", "runtime_structure_smoke_pilot", "policy_profile_default_dispatch_adoption", ): assert resolve_profile_name(mapping, gt) == gt # 1:1 deterministic # --- 3. boundary → gate/HOLD/forbidden expansion (engine read-only) --------- def test_boundary_expansion_surfaces_engine_resolution(): """resolver 가 engine 산출을 재구현하지 않고 read-only 표면화하는지: boundary deny 가 forbidden 으로 확장되고, engine 정본 어댑터와 동치.""" req = DefaultDispatchRequest( goal_id="t3", goal_statement="x", goal_type="task_2553_final_closeout", boundary=["forbid:profile_select"], ) res = resolve_default_profile(req, **_dirs()) # boundary deny 'forbid:profile_select' 가 profile allowed 와 충돌 → # engine allowed∩forbidden → PROFILE_CONFLICT (fail-closed §3.7). assert res.status == "PROFILE_CONFLICT" assert "forbid:profile_select" in res.boundary_expansion["explicit_boundary"] # 충돌 없는 boundary 는 engine 정본 어댑터와 동치 표면화. gr = { "goal_id": "t3b", "goal_statement": "x", "goal_type": "task_2553_final_closeout", "boundary": ["forbid:production_write"], "policy_profile": {"name": "task_2553_final_closeout"}, } eng = resolve_policy( gr, profile_json_dir=PROFILE_JSON_DIR, profile_schema_dir=PROFILE_SCHEMA_DIR, schema_dir=SCHEMA_DIR, ).to_coordinator_binding() ok = resolve_default_profile( DefaultDispatchRequest( goal_id="t3b", goal_statement="x", goal_type="task_2553_final_closeout", boundary=["forbid:production_write"], ), **_dirs(), ) assert ok.status == "RESOLVED" be = ok.boundary_expansion assert be["gate_condition_names"] == eng["gate_condition_names"] assert be["hold_trigger_conditions"] == eng["hold_trigger_conditions"] assert be["allowed_actions"] == eng["allowed_actions"] assert be["forbidden_actions"] == eng["forbidden_actions"] # --- 4. dispatch planning adapter (plan_only hard-pinned) ------------------- def test_dispatch_planning_adapter_plan_only_hard_pinned(): d = run_default_profile_resolution( {"goal_id": "p1", "goal_statement": "x", "goal_type": "runtime_structure_smoke_pilot"}, **_dirs()) pin = adapt_for_dispatch_planning(d) assert pin["signal"] == "PLAN_INPUT_OK" assert pin["plan_admissible"] is True # hard-pinned — 실 dispatch/PR/merge/branch 자동확정 0 (§2/§6). assert pin["plan_only"] is True assert pin["write_authority"] is False assert pin["merge_authority"] is False assert pin["pr_branch_authority"] is False assert pin["auto_dispatch"] is False assert pin["dispatch_lifecycle_effect"] == "none" def test_dispatch_planning_adapter_fail_closed_on_refused(): d = run_default_profile_resolution( {"goal_id": "p2", "goal_statement": "x", "goal_type": "no_such_goal_type"}, **_dirs()) pin = adapt_for_dispatch_planning(d) assert pin["plan_admissible"] is False assert pin["signal"] == "PLAN_FAIL_CLOSED" assert pin["plan_only"] is True assert pin["merge_authority"] is False # decision 부재 → UNAVAILABLE (예외를 planner 진행신호로 오인 불가). bad = adapt_for_dispatch_planning({"schema": "wrong"}) assert bad["signal"] == "PLAN_DECISION_UNAVAILABLE" assert bad["plan_admissible"] is False # --- 5. batch coordinator input adapter (자동확정 0 hard-pinned) ----------- def test_coordinator_input_adapter_no_auto_confirm(): d = run_default_profile_resolution( {"goal_id": "c1", "goal_statement": "x", "goal_type": "policy_profile_default_dispatch_adoption", "boundary": ["forbid:production_write"]}, **_dirs()) cin = adapt_for_coordinator_input(d) assert cin["signal"] == "COORD_INPUT_OK" assert cin["consumable"] is True assert cin["closeout_authority"] is False assert cin["merge_authority"] is False assert cin["auto_confirm"] is False assert cin["coordinator_role"] == "decision_consumer_only" @pytest.mark.parametrize( "req,sig", [ ({"goal_id": "c", "goal_statement": "x", "goal_type": "no_such"}, "COORD_RESOLVER_REFUSED"), ({"goal_id": "c", "goal_statement": "x", "goal_type": "task_2553_final_closeout", "policy_profile": {"name": "runtime_structure_smoke_pilot"}}, "COORD_RESOLVER_CONFLICT"), ], ) def test_coordinator_input_adapter_fail_closed(req, sig): cin = adapt_for_coordinator_input(run_default_profile_resolution(req, **_dirs())) assert cin["signal"] == sig assert cin["consumable"] is False assert cin["closeout_authority"] is False assert cin["merge_authority"] is False assert cin["auto_confirm"] is False def test_coordinator_input_adapter_unavailable_on_bad_decision(): cin = adapt_for_coordinator_input({"schema": "anu_v3.something.else"}) assert cin["signal"] == "COORD_DECISION_UNAVAILABLE" assert cin["closeout_authority"] is False assert cin["auto_confirm"] is False # --- 6. missing / unknown fail-closed (추측·날조 0) ------------------------- @pytest.mark.parametrize( "raw,status,code", [ ({"goal_id": "a", "goal_statement": "x"}, "DEFAULT_RESOLUTION_REFUSED", "goal_type_missing"), ({"goal_id": "a", "goal_statement": "x", "goal_type": "unknown_xyz"}, "DEFAULT_RESOLUTION_REFUSED", "goal_type_not_mapped"), (42, "DEFAULT_RESOLUTION_REFUSED", "default_request_not_mapping"), ], ) def test_missing_unknown_fail_closed(raw, status, code): d = run_default_profile_resolution(raw, **_dirs()) assert d["status"] == status assert d["profile_bound"] is False assert d["auto_apply"] is False assert d["refusal_code"] == code def test_mapped_but_profile_file_absent_fail_closed(tmp_path): """mapping 은 가리키지만 profile 파일 부재 → engine fail-closed 전파.""" bad_map = tmp_path / "m.json" bad_map.write_text(json.dumps({ "schema_id": "x", "version": "v1", "default_path": True, "mappings": {"ghost": {"policy_profile": "no_such_profile_zzz"}}, }), encoding="utf-8") d = run_default_profile_resolution( {"goal_id": "g", "goal_statement": "x", "goal_type": "ghost"}, mapping_path=bad_map, profile_json_dir=PROFILE_JSON_DIR, profile_schema_dir=PROFILE_SCHEMA_DIR, schema_dir=SCHEMA_DIR) assert d["status"] == "DEFAULT_RESOLUTION_REFUSED" assert d["profile_bound"] is False assert d["refusal_code"] == "profile_load_fail" # --- 7. profile conflict fail-closed --------------------------------------- def test_explicit_profile_conflict_fail_closed(): d = run_default_profile_resolution( {"goal_id": "x", "goal_statement": "x", "goal_type": "task_2553_final_closeout", "policy_profile": {"name": "runtime_structure_smoke_pilot"}}, **_dirs()) assert d["status"] == "PROFILE_CONFLICT" assert d["profile_bound"] is False assert d["auto_apply"] is False assert d["refusal_code"] == "explicit_profile_conflict" def test_engine_allowed_forbidden_conflict_fail_closed(): d = run_default_profile_resolution( {"goal_id": "x", "goal_statement": "x", "goal_type": "policy_profile_default_dispatch_adoption", "boundary": ["forbid:gate_evaluate"]}, **_dirs()) assert d["status"] == "PROFILE_CONFLICT" assert d["profile_bound"] is False assert d["auto_apply"] is False assert d["refusal_code"] == "engine_allowed_forbidden_conflict" assert d["engine_decision"]["status"] == "HOLD_FOR_CHAIR" def test_ambiguous_mapping_conflict_fail_closed(tmp_path): amb = tmp_path / "m.json" amb.write_text(json.dumps({ "schema_id": "x", "version": "v1", "default_path": True, "mappings": {"amb": [{"policy_profile": "a"}, {"policy_profile": "b"}]}, }), encoding="utf-8") d = run_default_profile_resolution( {"goal_id": "g", "goal_statement": "x", "goal_type": "amb"}, mapping_path=amb, profile_json_dir=PROFILE_JSON_DIR, profile_schema_dir=PROFILE_SCHEMA_DIR, schema_dir=SCHEMA_DIR) assert d["status"] == "PROFILE_CONFLICT" assert d["refusal_code"] == "ambiguous_mapping_conflict" # --- 8. selected profile evidence JSON + schema 충족 ----------------------- def test_selected_profile_evidence_json(): ev = run_selected_profile_evidence( {"goal_id": "e1", "goal_statement": "x", "goal_type": "task_2553_final_closeout", "boundary": ["forbid:production_write"]}, **_dirs()) assert ev["schema"] == EVIDENCE_SCHEMA_ID assert ev["default_path"] is True assert ev["selection_input"]["goal_type"] == "task_2553_final_closeout" assert ev["selection_input"]["chair_supplied_policy_profile_name"] is False assert ev["selected"]["resolved_profile_name"] == "task_2553_final_closeout" assert ev["selected"]["profile_bound"] is True assert ev["fail_closed"] is False assert "parse_goal_request -> resolve_policy" in ev["engine_consumed_read_only"] def test_decision_satisfies_schema(cases): schema = json.loads(RES_SCHEMA.read_text(encoding="utf-8")) for c in cases: d = run_default_profile_resolution(c["request"], **_dirs()) errs = validate_against_meta(d, schema) assert not errs, f"{c['name']}: schema errs {errs}" assert d["schema"] == RESOLUTION_SCHEMA_ID def test_mapping_satisfies_schema(): schema = json.loads(MAP_SCHEMA.read_text(encoding="utf-8")) mapping = json.loads(MAPPING_PATH.read_text(encoding="utf-8")) assert not validate_against_meta(mapping, schema) def test_all_fixture_cases(cases): assert len(cases) == 8 for c in cases: d = run_default_profile_resolution(c["request"], **_dirs()) for k, v in c["expect"].items(): assert d[k] == v, f"{c['name']}.{k}: {d[k]!r} != {v!r}" # --- 9. byte-0 / 무변 / live repo / untracked ------------------------------ def test_engine_and_seam_byte0_unchanged(): assert _sha256(ENGINE_PY) == ENGINE_SHA256_BASELINE assert _sha256(DPS_PY) == DPS_SHA256_BASELINE assert _sha256(CPB_PY) == CPB_SHA256_BASELINE def test_existing_profile_and_inputs_unchanged_idempotent(): before = _sha256(EXISTING_PROFILE) map_before = _sha256(MAPPING_PATH) for _ in range(3): run_default_profile_resolution( {"goal_id": "i", "goal_statement": "x", "goal_type": "task_2553_final_closeout"}, **_dirs()) assert _sha256(EXISTING_PROFILE) == before == EXISTING_PROFILE_SHA256_BASELINE assert _sha256(MAPPING_PATH) == map_before # mapping read-only def test_canonical_engine_api_no_regression(): """+33 정본 API parse_goal_request / resolve_policy 무회귀.""" gr = parse_goal_request( {"goal_id": "r", "goal_statement": "x", "goal_type": "x", "policy_profile": {"name": "task_2553_final_closeout"}}, schema_dir=SCHEMA_DIR) assert gr["goal_id"] == "r" res = resolve_policy( gr, profile_json_dir=PROFILE_JSON_DIR, profile_schema_dir=PROFILE_SCHEMA_DIR, schema_dir=SCHEMA_DIR) 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"}}, profile_json_dir=PROFILE_JSON_DIR, profile_schema_dir=PROFILE_SCHEMA_DIR, schema_dir=SCHEMA_DIR) def test_resolver_is_pure_and_idempotent(): raw = {"goal_id": "i", "goal_statement": "x", "goal_type": "task_2553_final_closeout", "boundary": ["forbid:production_write"]} a = run_default_profile_resolution(raw, **_dirs()) b = run_default_profile_resolution(raw, **_dirs()) assert a == b assert raw["goal_type"] == "task_2553_final_closeout" def test_resolver_does_not_raise_into_dispatch(): try: run_default_profile_resolution( {"goal_id": "a", "goal_statement": "x", "goal_type": "no_such_goal_type"}, **_dirs()) run_default_profile_resolution(42, **_dirs()) except Exception as e: # pragma: no cover pytest.fail(f"resolver leaked exception into dispatch: {e!r}") def test_build_default_request_normalizes(): req = build_default_request( {"goal_id": "n", "goal_statement": "x", "goal_type": " gt ", "boundary": ["a", 1], "policy_profile": {"name": " p "}}) assert req.goal_type == "gt" assert req.boundary == ["a", "1"] assert req.explicit_policy_profile_name == "p" def test_live_repo_head_branch_ref_unchanged(): head_before = _git("rev-parse", "HEAD") branch_before = _git("rev-parse", "--abbrev-ref", "HEAD") run_default_profile_resolution( {"goal_id": "z", "goal_statement": "x", "goal_type": "task_2553_final_closeout"}, **_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/default_profile_resolver.py", "anu_v3/profile_dispatch_planning_adapter.py", "anu_v3/profile_coordinator_input_adapter.py", "memory/policy_profiles/goal_type_profile_mapping.json", "schemas/default_profile_resolution.schema.json", "tests/regression/test_default_profile_resolver_2553plus52.py"], capture_output=True, text=True, check=True).stdout.strip() assert tracked == "", f"Track3 산물이 tracked 됨(위반): {tracked!r}"