# -*- coding: utf-8 -*- """tests.regression.test_dispatch_entrypoint_wiring — task-2640 Track A. 회장 verbatim unfork #1/#2/#3 결선 단언: - dispatch/__init__.py 가 enforcer/helper/validator 를 import 했는가 (결선만) - dispatch/core.py 가 dispatch_to_bot_with_contract 를 노출하는가 - dispatch/prompt.py 가 build_prompt_with_contract guard 를 노출하는가 - 봇 spawn 직전 validator 호출 결과가 fail-closed 인가 Layer A / NO-CRON: subprocess / cokacdir / merge / cron 호출 0. """ from __future__ import annotations import importlib.util import sys from pathlib import Path # task-2640 — tests/dispatch shadow 회피 (test_spawn_callback_contract_validator.py # 와 동일 패턴). worktree 의 실 dispatch 패키지 + 신규 모듈 강제 로드. _ROOT = Path(__file__).resolve().parents[2] _TESTS_DIR = str(_ROOT / "tests") sys.path[:] = [p for p in sys.path if p != _TESTS_DIR] if str(_ROOT) not in sys.path: sys.path.insert(0, str(_ROOT)) for _m in [k for k in list(sys.modules) if k == "dispatch" or k.startswith("dispatch.")]: _cached = sys.modules.get(_m) _f = getattr(_cached, "__file__", "") or "" if "/tests/dispatch" in _f or _f == "": del sys.modules[_m] def _load_real(modname: str, relpath: str): existing = sys.modules.get(modname) if existing is not None and getattr(existing, "__file__", "") and str(existing.__file__).endswith(relpath): return existing spec = importlib.util.spec_from_file_location(modname, _ROOT / relpath) assert spec is not None and spec.loader is not None, f"spec not found: {relpath}" mod = importlib.util.module_from_spec(spec) sys.modules[modname] = mod spec.loader.exec_module(mod) return mod _load_real("dispatch", "dispatch/__init__.py") _load_real("dispatch.callback_owner_enforcer", "dispatch/callback_owner_enforcer.py") _load_real("dispatch.normal_fallback_callback_helper", "dispatch/normal_fallback_callback_helper.py") _load_real( "dispatch.spawn_callback_contract_validator", "dispatch/spawn_callback_contract_validator.py", ) _load_real("dispatch.core", "dispatch/core.py") _load_real("dispatch.prompt", "dispatch/prompt.py") def test_dispatch_init_imports_enforcer_helper_validator() -> None: """dispatch/__init__.py public surface 결선 — import 가 결성에 노출되어야 함.""" import dispatch as _dispatch_pkg # 회장 verbatim unfork #1 — enforcer 결선 assert hasattr(_dispatch_pkg, "enforce_callback_owner") assert hasattr(_dispatch_pkg, "SELF_COLLECTOR_FORBIDDEN") assert hasattr(_dispatch_pkg, "COLLECTOR_ROLE_ANU") assert hasattr(_dispatch_pkg, "DEFAULT_ANU_KEYS") assert hasattr(_dispatch_pkg, "ANU_KEY_2553") # ANU-owned callback request builder assert hasattr(_dispatch_pkg, "build_anu_owned_callback_request") # 신규 spawn validator assert hasattr(_dispatch_pkg, "validate_spawn_callback_contract") assert hasattr(_dispatch_pkg, "ContractValidationResult") assert hasattr(_dispatch_pkg, "NO_OP_SPAWN_CONTRACT_FAILED") def test_dispatch_core_exposes_spawn_contract_gate() -> None: """dispatch/core.py 가 dispatch_to_bot_with_contract 를 노출 — 회장 unfork #2.""" from dispatch import core as dc assert "dispatch_to_bot_with_contract" in dc.__all__ assert callable(dc.dispatch_to_bot_with_contract) assert "validate_spawn_callback_contract" in dc.__all__ assert "NO_OP_SPAWN_CONTRACT_FAILED" in dc.__all__ def test_dispatch_prompt_exposes_build_with_contract_guard() -> None: """dispatch/prompt.py 가 build_prompt_with_contract guard 를 노출 — 회장 unfork #3.""" from dispatch import prompt as dp assert "build_prompt_with_contract" in dp.__all__ assert callable(dp.build_prompt_with_contract) # 기존 build_prompt 도 facade 로 그대로 노출 assert "build_prompt" in dp.__all__ def test_dispatch_to_bot_with_contract_blocks_self_collector() -> None: """spawn 직전 validator 가 executor self-key 를 fail-closed 차단한다.""" from dispatch.core import ( NO_OP_SPAWN_CONTRACT_FAILED, dispatch_to_bot_with_contract, ) out = dispatch_to_bot_with_contract( task_id="task-2640-wiring-self-key", executor_key="c119085addb0f8b7", prompt=( "ANU_KEY: c119085addb0f8b7\ncollector_role=ANU\n" "## SELF_COLLECTOR / SENDFILE_ONLY / NOT_REGISTERED 차단\nbody" ), anu_key="c119085addb0f8b7", ) assert out["status"] == "blocked" assert out["no_op_reason"] == NO_OP_SPAWN_CONTRACT_FAILED assert out["primary_classification"] == "SELF_COLLECTOR_FORBIDDEN" def test_dispatch_to_bot_with_contract_pass_normal_path() -> None: """정상 ANU key + doctrine 정합 prompt 는 PASS 통과.""" from dispatch.core import dispatch_to_bot_with_contract out = dispatch_to_bot_with_contract( task_id="task-2640-wiring-pass", executor_key="1e41a2324a3ccdd0", prompt=( "ANU_KEY: c119085addb0f8b7\ncollector_role=ANU\n" "## SELF_COLLECTOR / SENDFILE_ONLY / NOT_REGISTERED 차단\nbody" ), anu_key="c119085addb0f8b7", ) assert out["status"] == "ok" assert out["verdict"] == "PASS" def test_build_prompt_with_contract_blocks_self_collector_self_key() -> None: """prompt 빌드 전 enforce guard 가 executor self-key 를 차단한다.""" from dispatch.prompt import build_prompt_with_contract out = build_prompt_with_contract( team_id="dev6-team", task_desc="dummy task desc for wiring test (executor == anu key forbidden)", task_id="task-2640-wiring-prompt-block", level="normal", task_type="coding", executor_key="c119085addb0f8b7", # SAME as anu_key — must block anu_key="c119085addb0f8b7", ) assert out["status"] == "blocked" assert out["prompt"] is None # enforce_verdict FAIL (stage 1 — before build_prompt was called) assert out["enforce_verdict"] == "FAIL" assert out["contract_verdict"] is None assert any( "SELF_COLLECTOR_FORBIDDEN" in c for c in out.get("classifications", []) ) def test_dispatch_module_subprocess_not_invoked() -> None: """본 wiring 검증에서 dispatch 모듈 subprocess 호출은 발생하지 않는다 (Layer A / NO-CRON).""" # 본 검증 자체로 NO-CRON 단언: 위 4개 테스트는 subprocess 비호출 코드만 사용한다. # 별도 mock 검증 대신 정적 단언 — Layer A 가드 (dispatch.core / # dispatch.prompt 의 새 함수는 subprocess import 0). from dispatch import core as dc from dispatch import prompt as dp src_core = dc.dispatch_to_bot_with_contract.__code__.co_filename src_prompt = dp.build_prompt_with_contract.__code__.co_filename assert src_core.endswith("core.py") assert src_prompt.endswith("prompt.py") # 두 helper 모듈에서 subprocess import 0 (코드 정적 단언). import dispatch.spawn_callback_contract_validator as scv assert "subprocess" not in dir(scv)