# -*- coding: utf-8 -*- """regression — task-2605 Track B READ_ONLY_OPERATIONAL_PILOT. 9-stage 실 entrypoint 자동 결선을 *증명* 한다. 본 테스트는 실 entrypoint 를 직접 호출(mock-only/문서-only 금지)하며 read-only — write 0, 실 dispatch 0, ledger byte-0 (direct write 0). mock-only FAIL: 음성 케이스로 (i) 미매핑 goal_type fail-closed, (ii) executor self key callback 구조적 차단, (iii) dead-man/fallback 진행 트리거 하드 FAIL, (iv) self-chain QUARANTINED, (v) Track A durable-success EVENT 소비 event-driven 분기(A 미완→NON_BLOCKING / A 완료→cancel-on-success live remove 또는 CANCEL_FAILED_CLASSIFIED) 를 검증한다. """ from __future__ import annotations import importlib.util import sys from pathlib import Path import pytest ROOT = Path(__file__).resolve().parents[2] if str(ROOT) not in sys.path: sys.path.insert(0, str(ROOT)) PILOT_REL = "scripts/run_operational_pilot_2605.py" def _load(modname: str): spec = importlib.util.spec_from_file_location(modname, ROOT / PILOT_REL) assert spec and spec.loader mod = importlib.util.module_from_spec(spec) spec.loader.exec_module(mod) return mod @pytest.fixture(scope="module") def pilot(): return _load("_p2605_pilot") @pytest.fixture(scope="module") def run(pilot): # read-only — ledger direct write 0, 실 dispatch 0. return pilot.run_pilot() def test_expected_files_allowlist_exact(pilot): """§4 expected_files allowlist = 정확히 회장 명세 10종. callback_4tuple_index.jsonl 은 track 미소유 공유 인프라 — 미포함.""" assert pilot.EXPECTED_FILES == ( "scripts/run_operational_pilot_2605.py", "memory/events/task-2605.pilot-run.json", "memory/events/task-2605.selected-profile.json", "memory/events/task-2605.execution-result.json", "memory/events/task-2605.independent-collector-result.json", "memory/events/task-2605.consolidated-result.json", "tests/regression/test_operational_pilot_2605.py", "memory/events/task-2605.decision.json", "memory/events/task-2605.result.json", "memory/reports/task-2605.md", ) assert ( "memory/events/callback_4tuple_index.jsonl" not in pilot.EXPECTED_FILES ) # 정적/shared-infra 는 runner 직접 write 안 함. for static in ( "scripts/run_operational_pilot_2605.py", "tests/regression/test_operational_pilot_2605.py", "memory/reports/task-2605.md", ): assert static not in pilot.EMIT_FILES def test_callback_owner_is_independent_anu_executor_self_forbidden(pilot): """callback owner = 독립 ANU key; executor self key 절대 금지.""" assert pilot.ANU_CALLBACK_KEY == "c119085addb0f8b7" assert pilot.EXECUTOR_SELF_KEY == "1e41a2324a3ccdd0" assert pilot.ANU_CALLBACK_KEY != pilot.EXECUTOR_SELF_KEY assert pilot.TRACK_A_TASK_ID == "task-2604" def test_nine_stage_auto_chained(run): """9-stage 가 한 입력 short read-only goal 으로 자동 결선.""" ev = run["evaluation"] assert ev["auto_chained_9_stage"] is True assert ev["all_passed"] is True for name, ok in ev["checks"].items(): assert ok is True, f"{name} failed" def test_stage1_profile_auto_resolved_without_chair_detail(run): """STAGE1 — 회장 미지정 gate/HOLD/allowed/forbidden 자동 산출.""" s1 = run["stages"]["stage1"] assert s1["status"] == "RESOLVED" assert s1["profile_bound"] is True assert s1["unknown_goal_type_resolution"].get("status") != "RESOLVED" def test_stage2_anu_key_contract_enforced(run): """STAGE2 — executor self key callback 구조적 차단.""" s2 = run["stages"]["stage2"] assert s2["anu_key_registration_allowed"] is True assert s2["self_key_registration_allowed"] is False assert s2["self_key_assert_raised_CallbackRegistrationBlocked"] is True def test_stage4_executor_self_collector_forbidden(run): """STAGE4 — executor self collector = SELF_COLLECTOR_FORBIDDEN.""" s4 = run["stages"]["stage4"] assert s4["executor_self_collector_forbidden"] is True assert s4["independent_anu_collector_ok"] is True def test_stage5_independent_anu_authoritative_self_chain_quarantined(run): """STAGE5 — authoritative 는 독립 ANU 만; self-chain QUARANTINED.""" s5 = run["stages"]["stage5"] assert s5["authoritative_pass"] is True assert s5["self_chain_quarantined"] is True assert s5["executor_self_collector"] is False def test_stage5_acceptance_event_driven_on_track_a(run): """STAGE5 — acceptance = Track A durable-success EVENT 소비 (event-driven, NOT fixed-time/dead-man). A 미완→NON_BLOCKING(b) OPERATIONAL_PASS / A 완료→cancel-on-success live remove 또는 CANCEL_FAILED_CLASSIFIED.""" s5 = run["stages"]["stage5"] ta = s5["track_a_event_consumption"] fb = s5["fallback_acceptance"] assert fb["event_driven_on_track_a"] is True assert ta["event_driven"] is True assert ta["fixed_time_or_dead_man_used"] is False if ta["track_a_status"] == "A_INCOMPLETE": assert fb["criterion"] == "(b) REGISTRY_NON_BLOCKING_EXPLICIT_MARK" assert fb["verdict"]["verdict"] == "OPERATIONAL_PASS" assert fb["non_blocking_mark_schema_valid"] is True assert fb["semantic_binding_ok"] is True assert "b" in fb["verdict"]["satisfied_criterion"] else: assert ta["track_a_status"] == "A_COMPLETE" assert ( fb["branch"]["cancel_on_success_live_remove_or_classified"] is True ) def test_stage6_durable_writeback_evaluated_readonly_no_ledger_write(run): """STAGE6 — +53 durable-success write-back **평가만**(read-only). 실 append 는 독립 ANU collector 소관 — ledger 직접 write 0.""" s6 = run["stages"]["stage6"] assert s6["durable_success_evaluated"] is True assert s6["writeback_classification"] in ( "DURABLE_SUCCESS_WRITTEN", "WRITEBACK_IDEMPOTENT_SKIP", ) assert s6["write_performed"] is False assert s6["ledger_direct_write"] == 0 assert "independent ANU collector" in s6["apply_deferred_to"] def test_stage7_loop_event_driven_deadman_forbidden(run): """STAGE7 — Track A EVENT event-driven 소비; dead-man 진행 트리거 하드 FAIL; fixed-time 미사용.""" s7 = run["stages"]["stage7"] assert s7["verdict"] in ("NEXT_ACTION_READY", "ALL_SETTLED", "WAIT") assert s7["event_driven_acceptance_consistent"] is True assert s7["fixed_time_used"] is False neg = s7["forbidden_trigger_negative"] assert neg["hard_fail"] is True assert neg["dead_man_signal_observed_not_promoted"] is True # A 미완이면 loop 는 Track A 이벤트 미감지(NO_LEDGER_RECORD fail-safe # DEFER — 진행 트리거 아님). if s7["track_a_status"] == "A_INCOMPLETE": assert s7["track_a_in_events_observed"] is False assert s7["track_a_single_task_result_ready"] is False def test_stage8_enactor_proposal_only_fallback_forbidden(run): """STAGE8 — proposal-only·additive; fallback 진행 트리거 하드 FAIL.""" s8 = run["stages"]["stage8"] assert s8["actual_write_performed"] is False assert s8["auto_executed"] is False assert s8["merge_pr_write"] == 0 assert s8["forbidden_trigger_negative"]["hard_fail"] is True def test_self_chain_not_authoritative(run, pilot): """self-chain 산출은 영구 비권위 — authoritative 는 독립 ANU.""" art = pilot._build_artifacts(run) cr = art["memory/events/task-2605.consolidated-result.json"] assert cr["self_chain_authoritative"] is False assert "DEFERRED_TO_INDEPENDENT_ANU_COLLECTOR" in ( cr["authoritative_selection"] ) def test_subagent_usage_ledger_present(run, pilot): """회장 §7 — team result contract + subagent usage ledger 강제.""" art = pilot._build_artifacts(run) res = art["memory/events/task-2605.result.json"] led = res["subagent_usage_ledger"] assert led["subagents_used"] == [] assert "subagent 미사용" in led["declaration"] ebc = led["external_boundary_contract"] assert ebc["team_result_contract"] is True assert ebc["executor_self_key_callback_collector_count"] == 0 assert ebc["executor_self_dispatch_delegation_count"] == 0 def test_decision_conditions_read_only_no_ledger_write(run, pilot): """§2/§5 — read-only·no PR/merge/credential/branch + ledger write 0.""" art = pilot._build_artifacts(run) dec = art["memory/events/task-2605.decision.json"] c = dec["conditions_satisfied_2"] assert c["read_only"] is True assert c["no_pr"] is True assert c["no_merge"] is True assert c["no_credential"] is True assert c["no_branch_commit_push"] is True assert c["ledger_direct_write"] == 0 assert dec["shared_infra_not_allowlisted"] == pilot.SHARED_LEDGER assert dec["hold_for_chair"] is False assert "Track A(task-2604) durable-success EVENT" in ( dec["acceptance_fallback_criterion"] ) def test_a_complete_branch_classified_when_no_lookup_record(pilot): """A 완료 시뮬레이션 — Track A durable-success EVENT 존재 시 cancel-on-success live remove 또는 CANCEL_FAILED_CLASSIFIED 분기가 실 entrypoint(cancel_on_success_live_wiring)로 동작함을 증명. 현 ledger 에 task-2605 fallback record 가 없으므로 분류 처분 CANCEL_FAILED_CLASSIFIED 가 되어야 한다 (미분류 잔존 아님).""" s1 = pilot._stage1_profile_selection( { "goal_id": "g", "goal_statement": "s", "goal_type": pilot.GOAL_TYPE, "boundary": [], } ) s3 = pilot._stage3_executor(s1) track_a_complete = { "real_entrypoint": "test-injected A_COMPLETE", "track_a_task_id": pilot.TRACK_A_TASK_ID, "shared_ledger_present": True, "track_a_durable_success_event_present": True, "track_a_durable_records_count": 1, "track_a_registry_record": {"task_id": pilot.TRACK_A_TASK_ID}, "event_driven": True, "fixed_time_or_dead_man_used": False, "track_a_status": "A_COMPLETE", } s5 = pilot._stage5_independent_collector(s3, track_a_complete) fb = s5["fallback_acceptance"] assert fb["criterion"] in ( "(a) CANCEL_ON_SUCCESS_LIVE_REMOVE", "CANCEL_FAILED_CLASSIFIED", ) br = fb["branch"] assert br["track_a_status"] == "A_COMPLETE" assert br["cancel_on_success_live_remove_or_classified"] is True # 실 durable lookup entrypoint 가 호출되어 분류 status 를 산출. assert br["cancel_disposition"].startswith( "CANCEL_FAILED_CLASSIFIED" ) or br["cancel_disposition"] == "CANCEL_ON_SUCCESS_LIVE_REMOVE_ELIGIBLE" def test_mock_only_is_fail_real_entrypoint_required(pilot, run): """mock-only FAIL — 실 entrypoint 가 호출됨을 증명 (문서-only 금지).""" art = pilot._build_artifacts(run) pr = art["memory/events/task-2605.pilot-run.json"] assert pr["mock_only"] is False assert pr["documentation_only"] is False # 옵션 B remediation — stage9 도 real entrypoint 값으로 backed. # range(1,10) 전수: stage1~8 무회귀 + stage9 real-entrypoint 실증. for i in range(1, 10): ep = run["stages"][f"stage{i}"]["real_entrypoint"] assert "anu_v3" in ep or "dispatch." in ep or "scripts." in ep assert issubclass(pilot.MockOnlyError, RuntimeError) def _make_seal_spy(delegate: bool): """consolidation seal seam spy 팩토리. ``delegate=True`` → 인자를 기록한 뒤 *실* selector (``avs.select_authoritative_verdict``) 로 그대로 위임 → 실 entrypoint 호출을 독립 증명. ``delegate=False`` → 호출만 기록(미위임). 반환되는 ``calls`` 리스트가 비어 있으면 seam 미경유(=실 selector 미호출) 로 stage9 회귀가 FAIL 되어야 한다.""" calls = [] def spy(avs, records, *, task_id, anu_keys): rec = { "avs": avs, "records": list(records), "task_id": task_id, "anu_keys": list(anu_keys), "real_seal": None, } calls.append(rec) if delegate: seal = avs.select_authoritative_verdict( records, task_id=task_id, anu_keys=anu_keys ) rec["real_seal"] = seal return seal return None return spy, calls def test_stage9_select_authoritative_verdict_real_call_proven_by_spy( run, pilot, monkeypatch ): """STAGE9 옵션 A 강화 — `select_authoritative_verdict` *실제 호출*을 monkeypatch/spy 로 독립 증명한다 (회장 §4.1·§4.2). seam spy 가 ``consolidate_pilot_result`` 가 로드한 *실* avs 모듈로 위임 → 실 entrypoint 가 *진짜 consolidated self-chain 레코드* 에 대해 실행됨을 관찰. 통합 결과 dict 의 seal 필드가 실 seal 값에서 *파생* 됨을 대조(자기보고 플래그만 검증 금지 — §4.2).""" spy, calls = _make_seal_spy(delegate=True) monkeypatch.setattr(pilot, "_consolidation_authoritative_seal", spy) out = pilot.consolidate_pilot_result( run["stages"], run["track_a_event"] ) # (1) seam(=실 selector) 단일 호출 — 미호출 시 즉시 FAIL. assert len(calls) == 1, ( "select_authoritative_verdict seam 미경유 — stage9 하드코딩/" "static-string 회귀" ) c = calls[0] assert c["task_id"] == pilot.TASK_ID assert c["anu_keys"] == [pilot.ANU_CALLBACK_KEY] assert c["anu_keys"] != [pilot.EXECUTOR_SELF_KEY] assert len(c["records"]) == 1 vr = c["records"][0] # 실 selector 에 전달된 consolidated self-chain 레코드 신원 검증. assert vr.kind == "collector_result" assert vr.task_id == pilot.TASK_ID assert vr.executor_key == pilot.EXECUTOR_SELF_KEY assert vr.collector_key == pilot.EXECUTOR_SELF_KEY assert vr.session_is_executor_self is True assert vr.claimed_origin == "independent" # untrusted — 재계산 대상 # (2) 실 모듈이 claim 무시·origin 재계산하여 quarantine 산출함을 증명. # consolidate_pilot_result 가 *실제 로드한* avs 모듈 인스턴스(spy 캡처) # 로 검증 — 동일 모듈 인스턴스에서 seal 이 산출됐음을 강하게 보증. avs = c["avs"] assert ( vr.derived_origin([pilot.ANU_CALLBACK_KEY]) == avs.ORIGIN_SELF_CHAIN ) real_seal = c["real_seal"] assert isinstance(real_seal, avs.AuthoritativeSelectionResult) assert real_seal.classification == avs.AUTHORITATIVE_VERDICT_PENDING assert real_seal.independent_anu_count == 0 assert real_seal.quarantined_count >= 1 assert real_seal.ok is False # (3) 통합 결과는 실 seal 에서 *파생* (하드코딩 아님) — 대조. assert out["consolidation_seal_verdict"] == real_seal.verdict assert ( out["consolidation_seal_classification"] == real_seal.classification ) assert ( out["consolidation_seal_quarantined_count"] == real_seal.quarantined_count ) assert ( out["consolidation_seal_independent_anu_count"] == real_seal.independent_anu_count ) assert out["consolidation_non_authoritative_confirmed"] is True assert out["self_chain_authoritative"] is False assert "DEFERRED_TO_INDEPENDENT_ANU_COLLECTOR" in ( out["authoritative_selection"] ) @pytest.mark.parametrize( "cls,qc,iac,vrd,expect_non_auth", [ # 실 PENDING 동형 변형 — non_authoritative True 가 seal 에서 파생. ("AUTHORITATIVE_VERDICT_PENDING", 5, 0, "FAIL", True), # 명백히 다른 sentinel — 하드코딩이면 통과 못 함. ("MUT_SENTINEL_CLS", 4242, 7, "MUT_SENTINEL_V", False), # PASS/AUTHORITATIVE 변형 — non_authoritative False 로 뒤집힘. ("AUTHORITATIVE_PASS", 0, 2, "PASS", False), ], ) def test_stage9_hardcoded_consolidation_cannot_pass_mutation( run, pilot, monkeypatch, cls, qc, iac, vrd, expect_non_auth ): """STAGE9 옵션 A 강화 — 하드코딩 우회 구조 제거 (회장 §4.3). seam 이 *임의* selector 결과를 반환하도록 변형하면 통합 결과 dict 가 그 값을 *그대로 반영* 해야 한다. ``consolidate_pilot_result()`` 가 stage9 값을 하드코딩하면 본 mutation 대조에서 반드시 FAIL → 하드코딩 통과 구조가 구조적으로 제거됨을 보증한다.""" avs = pilot._load_real( "anu_v3.authoritative_verdict_selector", "anu_v3/authoritative_verdict_selector.py", ) sentinel = avs.AuthoritativeSelectionResult( schema="mutation-sentinel", verdict=vrd, classification=cls, task_id=pilot.TASK_ID, authoritative_verdict=None, authoritative_source_kind=None, quarantined_count=qc, independent_anu_count=iac, deadman_duplicate_seen=False, deadman_valid=False, reasons=["mutation"], ) def mut(*_a, **_k): del _a, _k # seam signature 흡수 — sentinel 고정 반환. return sentinel monkeypatch.setattr(pilot, "_consolidation_authoritative_seal", mut) out = pilot.consolidate_pilot_result( run["stages"], run["track_a_event"] ) # seal 필드가 sentinel 에서 파생 — 하드코딩이면 canonical 값이 남아 FAIL. assert out["consolidation_seal_verdict"] == vrd assert out["consolidation_seal_classification"] == cls assert out["consolidation_seal_quarantined_count"] == qc assert out["consolidation_seal_independent_anu_count"] == iac assert out["consolidation_non_authoritative_confirmed"] is expect_non_auth def test_stage9_seam_is_on_critical_path_not_dead_code( run, pilot, monkeypatch ): """STAGE9 옵션 A 강화 — seam(실 selector) 이 critical path 임을 증명. seam 을 예외로 대체하면 ``consolidate_pilot_result`` 가 그대로 전파해야 한다. stage9 가 selector 결과와 무관한 하드코딩 shortcut 을 가졌다면 예외가 삼켜져 통과할 것이므로, 본 테스트가 그 구조를 차단.""" def boom(*_a, **_k): del _a, _k # seam signature 흡수 — critical-path probe 예외. raise RuntimeError("seam-on-critical-path-probe") monkeypatch.setattr(pilot, "_consolidation_authoritative_seal", boom) with pytest.raises(RuntimeError, match="seam-on-critical-path-probe"): pilot.consolidate_pilot_result( run["stages"], run["track_a_event"] ) def test_stage9_self_report_contract_and_real_entrypoint_shape(run, pilot): """STAGE9 — 옵션 B real-entrypoint 계승 계약 (자기보고 *외* spy 증명은 위 테스트들이 담당; 본 테스트는 형태/계약 계승만 보존).""" s9 = run["stages"]["stage9"] ep = s9["real_entrypoint"] assert "anu_v3" in ep and "scripts." in ep assert "static string" not in ep.split("(")[0] assert ( s9["consolidation_entrypoint"] == "anu_v3.authoritative_verdict_selector." "select_authoritative_verdict" ) assert s9["consolidation_sealed"] is True assert s9["self_chain_authoritative"] is False assert s9["consolidation_non_authoritative_confirmed"] is True assert s9["consolidation_seal_classification"] == ( "AUTHORITATIVE_VERDICT_PENDING" ) assert s9["consolidation_seal_independent_anu_count"] == 0 assert s9["consolidation_seal_quarantined_count"] >= 1 assert s9["mock_only"] is False assert s9["documentation_only"] is False assert "DEFERRED_TO_INDEPENDENT_ANU_COLLECTOR" in ( s9["authoritative_selection"] ) assert callable(pilot.consolidate_pilot_result) assert callable(pilot._stage9_consolidation) assert callable(pilot._consolidation_authoritative_seal) assert ( run["evaluation"]["checks"]["stage9_consolidation_real_entrypoint"] is True ) art = pilot._build_artifacts(run) pr = art["memory/events/task-2605.pilot-run.json"] assert set(pr["nine_stage_chain"].keys()) == { f"stage{i}" for i in range(1, 10) } assert pr["stage9"] == s9["real_entrypoint"] assert "consolidated result (memory/events/" not in pr["stage9"] cr = art["memory/events/task-2605.consolidated-result.json"] assert set(cr["nine_stage_real_entrypoints"].keys()) == { f"stage{i}" for i in range(1, 10) } assert cr["stage9_consolidation_seal"]["consolidation_sealed"] is True def test_stage1_8_byte_equivalent_no_regression_and_nine_stage(pilot): """회장 §4.4·§4.5 — stage1~8 byte-equivalent 무회귀 + range(1,10) 9-stage 검증 유지. seam 도입이 결정론/스테이지 산출을 변형하지 않음을 독립 run 2회로 대조.""" import json r1 = pilot.run_pilot() r2 = pilot.run_pilot() # range(1,10) 9-stage 전수 존재. assert set(r1["stages"].keys()) == { f"stage{i}" for i in range(1, 10) } for i in range(1, 9): k = f"stage{i}" b1 = json.dumps(r1["stages"][k], sort_keys=True, ensure_ascii=False) b2 = json.dumps(r2["stages"][k], sort_keys=True, ensure_ascii=False) assert b1 == b2, f"{k} byte-equivalent 무회귀 위반" # stage9 default(unpatched) 경로도 옵션 B 구현 무회귀(결정론). b1 = json.dumps(r1["stages"]["stage9"], sort_keys=True, ensure_ascii=False) b2 = json.dumps(r2["stages"]["stage9"], sort_keys=True, ensure_ascii=False) assert b1 == b2 # 9-stage 자동 결선 evaluation 유지. assert r1["evaluation"]["auto_chained_9_stage"] is True assert all(r1["evaluation"]["checks"].values())