"""tests/regression/test_collector_path_wiring_2553plus37.py task-2553+37 — CANCEL-ON-SUCCESS COLLECTOR-PATH WIRING regression (§3 1~15). 9-R.1: 전 케이스 mock/fixture/격리 — 실 cron-list·실 cron-remove·실 schedule_history 무접촉. "cron-remove called" = 주입 SpyRemover 호출 spy assert (실 cron 삭제 아님). 실 subprocess 즉시 FAIL 차단. 실 운영 cron 실제 삭제 0 (§6 / 9-R.1 / 9-R.4). live /home/jay/workspace git tracked HEAD/branch/ref 전후 assertEqual (§5 repo root 기준). §3 매핑: 1 wired-fix → (mocked) cron-remove called 2 fallback-id-missing → cron-remove 0, fallback preserved 3 +23 task-id-mismatch (binding ok) → cron-remove 0 4 +23 chat-id-mismatch (binding ok) → cron-remove 0 5 +23 role-not-fallback (binding ok)→ cron-remove 0 6 non-PASS (HOLD/RESULT_MISSING) → cron-remove 0 7 seam exception → normal success preserved 8 cancel-audit JSON 생성 (wired/binding-invalid 양쪽) 9 DUPLICATE_CALLBACK_IGNORED 무회귀 10 +33/+34/+35 COLLECTOR_PATH_NOT_WIRED 재현 후 수정 경로 PASS 11 +32 mandatory normal callback contract 무회귀 12 +25 operational seam 무회귀 13 registry/checkpoint primary path 대체 없음 14 raw token/credential exposure 0 15 unrelated cron remove attempt 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 utils.anu_delegation_completion_callback import ( # noqa: E402 CallbackInput, CallbackType, Classification, run_completion_callback_collector, ) from utils.completion_callback_fallback_cancel import RemoverResult # noqa: E402 from utils.normal_completion_callback_collector_entrypoint import ( # noqa: E402 NormalCallbackBinding, build_normal_completion_callback_prompt, run_wired_normal_completion_callback_collector, validate_normal_callback_binding, ) from utils.operational_collector_wiring import ( # noqa: E402 OperationalSeamParams, run_operational_completion_callback_collector, ) FIXDIR = WORKSPACE / "memory" / "fixtures" FROZEN = WORKSPACE / "utils" / "anu_delegation_completion_callback.py" FROZEN_SHA = "83b3e307c8207c76a3e311c408aab4951373bd317896e51687d3007907b0c3d4" ANU_KEY_SECRET = "c119085addb0f8b7" # 노출 검사용 — 산출물에 박히면 FAIL def _fx(name: str) -> dict: return json.loads((FIXDIR / f"{name}.json").read_text(encoding="utf-8")) class FakeCronLister: def __init__(self, entries, status="ok"): self.entries = entries self.status = status self.calls = 0 def __call__(self) -> dict: self.calls += 1 if self.status != "ok": return {"status": self.status, "entries": [], "raw": {}} return { "status": "ok", "entries": list(self.entries), "raw": {"fake": True}, } class SpyRemover: """주입 fake remover — 실 cron 삭제 0. 호출 인자 spy 기록.""" def __init__(self, status: str = "removed"): self.status = status self.calls: list = [] def __call__(self, cron_id: str, *, dry_run: bool = True) -> RemoverResult: self.calls.append({"cron_id": cron_id, "dry_run": dry_run}) return RemoverResult(status=self.status, detail=f"fake:{self.status}") @pytest.fixture(autouse=True) def _block_real_subprocess(monkeypatch): def _boom(*a, **k): # noqa: ANN001, ANN002, ANN003 raise AssertionError( "실 subprocess 호출 금지 (§6 9-R.1/9-R.4) — fake lister/remover 만" ) monkeypatch.setattr(subprocess, "run", _boom) def _git_ref(): git_dir = WORKSPACE / ".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) @pytest.fixture(autouse=True) def _git_ref_invariant(): before = _git_ref() yield assert _git_ref() == before, "git HEAD/branch/ref 변경 감지 (§5 위반)" def _materialize(fx: dict, tmp: Path) -> dict: tmp.mkdir(parents=True, exist_ok=True) p: dict = {} dfm = tmp / f"{fx['task_id']}.dispatch-fired.json" dfm.write_text(json.dumps(fx["dispatch_fired_marker"]), encoding="utf-8") p["dispatch_fired_marker_path"] = dfm rj = tmp / f"{fx['task_id']}.result.json" rj.write_text(json.dumps(fx["result_json"]), encoding="utf-8") p["result_json_path"] = rj rep = tmp / f"{fx['task_id']}.report.md" rep.write_text(fx.get("report_text", ""), encoding="utf-8") p["report_path"] = rep crm = tmp / f"{fx['task_id']}.collector-result.json" crm.write_text( json.dumps(fx["collector_result_marker"]), encoding="utf-8" ) p["collector_result_marker_path"] = crm p["fallback_cancelled_marker_path"] = ( tmp / f"{fx['task_id']}.fallback-cancelled.json" ) p["cancel_lock_path"] = tmp / f"{fx['task_id']}.cancel.lock" p["seam_audit_path"] = tmp / f"{fx['task_id']}.plus23-cancel-audit.json" return p def _binding_from_fx(fx: dict, *, fallback_override="__use_fx__"): b = fx.get("binding", {}) fid = ( b.get("fallback_cron_id") if fallback_override == "__use_fx__" else fallback_override ) return NormalCallbackBinding( task_id=b.get("task_id", fx["task_id"]), dispatch_cron_id=b.get("dispatch_cron_id", fx.get("dispatch_cron_id", "D")), normal_collector_cron_id=b.get( "normal_collector_cron_id", fx.get("normal_collector_cron_id", "N") ), fallback_cron_id=fid, chat_id=b.get("chat_id", 6937032012), ) def _pass_input(task_id: str, *, callback_type=CallbackType.NORMAL): return CallbackInput( task_id=task_id, executor="dev4-sim", dispatch_cron_id="DISP37", callback_type=callback_type, callback_cron_id="NORM37", cron_status="ok", task_status="completed", required_closeout_markers={"result_json": True, "report": True}, preservation_anchors={"frozen_anchor": "match"}, dev_sunset=True, ) def _run_wired(fx, tmp, *, inp=None, lister=None, remover=None, binding=None): p = _materialize(fx, tmp) spy = remover or SpyRemover("removed") lst = lister or FakeCronLister(fx["live_cron_entries"]) res = run_wired_normal_completion_callback_collector( inp or _pass_input(fx["task_id"]), tmp / "ack.json", binding=binding or _binding_from_fx(fx), dispatch_fired_marker_path=p["dispatch_fired_marker_path"], result_json_path=p["result_json_path"], report_path=p["report_path"], collector_result_marker_path=p["collector_result_marker_path"], claim_dir=tmp / "claims", cron_lister=lst, remover=spy, fallback_cancelled_marker_path=p["fallback_cancelled_marker_path"], cancel_lock_path=p["cancel_lock_path"], seam_audit_path=p["seam_audit_path"], callback_contract=fx.get("callback_contract"), ) return res, spy, lst # ── §3.1 wired entrypoint → operational_collector_wiring → cron-remove ────── def test_01_wired_success_bound_verifier_pass_remove_called(tmp_path): fx = _fx("task-2553+37.wired-fix") res, spy, lst = _run_wired(fx, tmp_path) assert res.wired_via_operational_collector_wiring is True assert res.binding_valid is True assert res.collector_result.classification == Classification.PASS assert res.durable_success is True assert res.seam_invoked is True assert res.cron_remove_invoked is True assert res.fallback_preserved is False # PRIMARY 결선 증명 — +25 wiring 결과 보유 assert res.wiring_result is not None assert res.wiring_result.seam_outcome.seam_classification == "PLUS9A_CANCELLED" # operational=True → dry_run=False 로 bound id 만 remove 시도 assert spy.calls == [{"cron_id": "FB37-0001", "dry_run": False}] assert lst.calls == 1 assert res.cancel_audit["normal_success_unchanged"] is True # ── §3.2 fallback_cron_id missing → cron-remove 0, fallback preserved ────── def test_02_fallback_id_missing_no_remove_preserved(tmp_path): fx = _fx("task-2553+37.fallback-id-missing") res, spy, _ = _run_wired(fx, tmp_path) assert res.binding_valid is False assert "fallback_cron_id_missing" in res.binding_invalid_reasons assert res.seam_invoked is False assert res.cron_remove_invoked is False assert res.fallback_preserved is True assert spy.calls == [] # 실 remove 0 assert res.cancel_audit["lookup_status"] == "BINDING_MISSING" assert res.cancel_audit["remove_attempted"] is False assert res.cancel_audit["normal_success_unchanged"] is True # 디커플 — collector 분류는 그대로 산출 assert res.collector_result.classification == Classification.PASS # ── §3.x fallback_cron_id marker mismatch → cron-remove 0 ────────────────── def test_02b_fallback_id_marker_mismatch_no_remove(tmp_path): fx = _fx("task-2553+37.fallback-id-mismatch") res, spy, _ = _run_wired(fx, tmp_path) assert res.binding_valid is False assert any( r.startswith("fallback_cron_id_marker_mismatch") for r in res.binding_invalid_reasons ) assert res.seam_invoked is False assert spy.calls == [] assert res.cancel_audit["lookup_status"] == "BINDING_MISMATCH" assert res.cancel_audit["normal_success_unchanged"] is True # ── §3.3/3.4/3.5 task/chat/role mismatch (binding ok) → cron-remove 0 ────── @pytest.mark.parametrize( "name", [ "task-2553+23.task-id-mismatch", "task-2553+23.chat-id-mismatch", "task-2553+23.role-not-fallback", ], ) def test_03_live_mismatch_binding_ok_no_remove(tmp_path, name): fx = _fx(name) # binding 은 marker 와 일치(valid) → seam 진입하되 live verifier 가 SKIP binding = NormalCallbackBinding( task_id=fx["task_id"], dispatch_cron_id="DISP23", normal_collector_cron_id="NORM23", fallback_cron_id=fx["target_cron_id"], chat_id=6937032012, ) res, spy, _ = _run_wired( fx, tmp_path, inp=_pass_input(fx["task_id"]), binding=binding, ) assert res.binding_valid is True assert res.seam_invoked is True # entrypoint→+25 경유(PRIMARY) assert res.cron_remove_invoked is False assert spy.calls == [] # live verifier SKIP_MISMATCH → 실 remove 0 assert res.wiring_result.seam_outcome.seam_classification == ( "SKIP_LIVE_SKIP_MISMATCH" ) assert res.collector_result.classification == Classification.PASS # ── §3.6 non-PASS (HOLD/RESULT_MISSING) → cron-remove 0 ──────────────────── @pytest.mark.parametrize( "mutate,expect_cls", [ ( {"required_closeout_markers": {"result_json": False}}, Classification.RESULT_MISSING, ), ({"chair_gated": True}, Classification.HOLD_FOR_CHAIR), ], ) def test_06_non_pass_no_remove(tmp_path, mutate, expect_cls): fx = _fx("task-2553+37.wired-fix") inp = _pass_input(fx["task_id"]) for k, v in mutate.items(): setattr(inp, k, v) res, spy, _ = _run_wired(fx, tmp_path, inp=inp) assert res.collector_result.classification == expect_cls assert res.durable_success is False assert res.seam_invoked is False assert res.cron_remove_invoked is False assert spy.calls == [] # ── §3.7 seam exception → normal success preserved (디커플) ──────────────── def test_07_seam_exception_decoupled(tmp_path, monkeypatch): fx = _fx("task-2553+37.wired-fix") import utils.operational_collector_wiring as wiring def _boom(**k): # noqa: ANN003 raise RuntimeError("seam blew up") monkeypatch.setattr(wiring, "run_operational_cancel_seam", _boom) res, spy, _ = _run_wired(fx, tmp_path) # collector 성공은 seam 예외와 무관하게 PASS 유지 assert res.collector_result.classification == Classification.PASS assert res.collector_result.closeout_candidate is True assert res.seam_invoked is False assert spy.calls == [] assert res.cancel_audit["normal_success_unchanged"] is True # ── §3.7b ANU-Codex adjudication — frozen collector 단일 호출(ack # single-winner) + +25 단독 디커플 위임(재호출 래퍼 0, Codex MED 수용) ────── def test_07b_single_frozen_call_no_reinvoke_wrapper(): """binding-valid 경로는 frozen collector 를 재호출하는 try/except 래퍼를 두지 않는다 — 동일 ack loser 로 PASS→DUPLICATE 뒤집힘 방지(Codex MED). """ import ast src = ( WORKSPACE / "utils" / "normal_completion_callback_collector_entrypoint.py" ).read_text(encoding="utf-8") tree = ast.parse(src) fn = next( n for n in ast.walk(tree) if isinstance(n, ast.FunctionDef) and n.name == "run_wired_normal_completion_callback_collector" ) # 함수 본문 calls 수집 calls = [ c.func.id for c in ast.walk(fn) if isinstance(c, ast.Call) and isinstance(c.func, ast.Name) ] # frozen collector 직접 호출은 binding-invalid 경로 1곳뿐(재호출 래퍼 0). assert calls.count("run_completion_callback_collector") == 1 assert calls.count("run_operational_completion_callback_collector") == 1 # ── §3.8 cancel-audit JSON 생성 (wired + binding-invalid 양쪽) ───────────── def test_08_cancel_audit_generated_both_paths(tmp_path): req = { "schema", "event_id", "target_cron_id", "lookup_status", "five_condition_results", "remove_attempted", "remove_result", "skip_reason", "already_removed_or_missing", "normal_success_unchanged", } r1, _, _ = _run_wired(_fx("task-2553+37.wired-fix"), tmp_path / "a") assert req.issubset(r1.cancel_audit.keys()) assert Path(r1.cancel_audit_path).exists() r2, _, _ = _run_wired(_fx("task-2553+37.fallback-id-missing"), tmp_path / "b") assert req.issubset(r2.cancel_audit.keys()) assert Path(r2.cancel_audit_path).exists() assert r2.cancel_audit["schema"] == "task-2553+25.cancel-audit_v1" # ── §3.9 DUPLICATE_CALLBACK_IGNORED 무회귀 ───────────────────────────────── def test_09_duplicate_callback_no_regression(tmp_path): fx = _fx("task-2553+37.wired-fix") p = _materialize(fx, tmp_path / "shared") ack = tmp_path / "shared-ack.json" claim = tmp_path / "claims" binding = _binding_from_fx(fx) def _call(ctype, spy): return run_wired_normal_completion_callback_collector( _pass_input(fx["task_id"], callback_type=ctype), ack, binding=binding, dispatch_fired_marker_path=p["dispatch_fired_marker_path"], result_json_path=p["result_json_path"], report_path=p["report_path"], collector_result_marker_path=p["collector_result_marker_path"], claim_dir=claim, cron_lister=FakeCronLister(fx["live_cron_entries"]), remover=spy, fallback_cancelled_marker_path=p["fallback_cancelled_marker_path"], cancel_lock_path=p["cancel_lock_path"], ) first = _call(CallbackType.NORMAL, SpyRemover("removed")) spy2 = SpyRemover("removed") second = _call(CallbackType.FALLBACK_STALE, spy2) assert first.collector_result.classification == Classification.PASS assert first.seam_invoked is True assert ( second.collector_result.classification == Classification.DUPLICATE_CALLBACK_IGNORED ) assert second.durable_success is False assert second.seam_invoked is False assert spy2.calls == [] # 안전망 무회귀 — 후발 seam 미진입 # ── §3.10 COLLECTOR_PATH_NOT_WIRED 재현 후 수정 경로 PASS ────────────────── def test_10_not_wired_reproduction_then_fixed(tmp_path, monkeypatch): fx = _fx("task-2553+37.collector-path-not-wired") p = _materialize(fx, tmp_path / "nw") # (재현) +33/+34/+35 — 자유형 collector 가 frozen collector 만 직접 호출. # operational cancel seam spy 가 호출되지 않음을 입증. import utils.operational_collector_wiring as wiring seam_spy = {"called": False} orig_seam = wiring.run_operational_cancel_seam def _spy_seam(**k): # noqa: ANN003 seam_spy["called"] = True return orig_seam(**k) monkeypatch.setattr(wiring, "run_operational_cancel_seam", _spy_seam) free_form = run_completion_callback_collector( _pass_input(fx["task_id"]), tmp_path / "nw" / "freeform-ack.json" ) assert free_form.classification == Classification.PASS assert seam_spy["called"] is False # 미경유 — seam 호출 0 (버그 재현) nw_audit = tmp_path / "nw" / "task-2553+37.cancel-audit.json" assert not nw_audit.exists() # cancel-audit ABSENT (진단 일치) # (수정) 동일 입력을 표준 wired entrypoint 로 → seam 경유·remove called. seam_spy["called"] = False spy = SpyRemover("removed") res = run_wired_normal_completion_callback_collector( _pass_input(fx["task_id"]), tmp_path / "nw" / "wired-ack.json", binding=_binding_from_fx(fx), dispatch_fired_marker_path=p["dispatch_fired_marker_path"], result_json_path=p["result_json_path"], report_path=p["report_path"], collector_result_marker_path=p["collector_result_marker_path"], claim_dir=tmp_path / "nw" / "claims", cron_lister=FakeCronLister(fx["live_cron_entries"]), remover=spy, ) assert seam_spy["called"] is True # 수정 — seam 경유 assert res.wired_via_operational_collector_wiring is True assert res.seam_invoked is True assert res.cron_remove_invoked is True assert spy.calls == [{"cron_id": "FB37-NW01", "dry_run": False}] assert Path(res.cancel_audit_path).exists() # cancel-audit 생성 # ── §3.11 +32 mandatory normal callback contract 무회귀 ──────────────────── def test_11_plus32_mandatory_contract_no_regression(tmp_path): from dispatch.executor_completion_contract import ( validate_4tuple, ) ok = NormalCallbackBinding( task_id="t", dispatch_cron_id="D", normal_collector_cron_id="N", fallback_cron_id="F", ) assert validate_normal_callback_binding(ok) == [] assert validate_4tuple(ok.as_4tuple()) == [] # normal_collector_cron_id 누락 → +32 mandatory 위반 그대로 검출 bad = NormalCallbackBinding( task_id="t", dispatch_cron_id="D", normal_collector_cron_id=None, fallback_cron_id="F", ) reasons = validate_normal_callback_binding(bad) assert any("normal_collector_cron_id missing" in r for r in reasons) # prompt 빌더는 mandatory 부재 시 ValueError (callback mandatory 약화 0) with pytest.raises(ValueError): build_normal_completion_callback_prompt( NormalCallbackBinding( task_id="t", dispatch_cron_id="D", normal_collector_cron_id="N", fallback_cron_id=None, ) ) # ── §3.12 +25 operational seam 무회귀 ────────────────────────────────────── def test_12_plus25_operational_seam_no_regression(tmp_path): fx = _fx("task-2553+25.dry-run") p = _materialize(fx, tmp_path) sp = OperationalSeamParams( target_cron_id=fx["target_cron_id"], dispatch_fired_marker_path=p["dispatch_fired_marker_path"], result_json_path=p["result_json_path"], report_path=p["report_path"], collector_result_marker_path=p["collector_result_marker_path"], audit_path=p["seam_audit_path"], cron_lister=FakeCronLister(fx["live_cron_entries"]), remover=SpyRemover("removed"), fallback_cancelled_marker_path=p["fallback_cancelled_marker_path"], cancel_lock_path=p["cancel_lock_path"], callback_contract=fx.get("callback_contract"), ) res = run_operational_completion_callback_collector( _pass_input(fx["task_id"]), tmp_path / "ack.json", seam_params=sp, claim_dir=tmp_path / "claims", ) assert res.collector_result.classification == Classification.PASS assert res.seam_invoked is True assert res.seam_outcome.seam_classification == "PLUS9A_CANCELLED" # ── §3.13 registry/checkpoint primary path 대체 없음 ────────────────────── def test_13_registry_not_primary(): import ast src = ( WORKSPACE / "utils" / "normal_completion_callback_collector_entrypoint.py" ).read_text(encoding="utf-8") tree = ast.parse(src) modules = { n.module for n in ast.walk(tree) if isinstance(n, ast.ImportFrom) and n.module } # entrypoint 는 registry/checkpoint(+29/+30/+31) 모듈을 primary 로 # import 하지 않는다 (recovery layer 분리). assert not any( "registry" in m or "checkpoint" in m or "reconcile" in m for m in modules ) contract = build_normal_completion_callback_prompt( NormalCallbackBinding( task_id="t", dispatch_cron_id="D", normal_collector_cron_id="N", fallback_cron_id="F", ) ) assert "PRIMARY" in contract["collector_path_contract"] assert "recovery" in contract["collector_path_contract"] # ── §3.14 raw token/credential exposure 0 ───────────────────────────────── def test_14_no_credential_exposure(tmp_path): contract = build_normal_completion_callback_prompt( NormalCallbackBinding( task_id="t", dispatch_cron_id="D", normal_collector_cron_id="N", fallback_cron_id="F", ) ) blob = json.dumps(contract, ensure_ascii=False) assert ANU_KEY_SECRET not in blob # 실 키 미박제 (placeholder ref 만) res, _, _ = _run_wired(_fx("task-2553+37.wired-fix"), tmp_path) assert ANU_KEY_SECRET not in json.dumps(res.cancel_audit, ensure_ascii=False) # 모듈 소스에도 실 키 하드코딩 0 src = ( WORKSPACE / "utils" / "normal_completion_callback_collector_entrypoint.py" ).read_text(encoding="utf-8") assert ANU_KEY_SECRET not in src # ── §3.15 unrelated cron remove attempt 0 ───────────────────────────────── def test_15_no_unrelated_cron_remove(tmp_path): fx = _fx("task-2553+37.wired-fix") res, spy, _ = _run_wired(fx, tmp_path) # remover 는 오직 bound target_cron_id 로만 1회 — 무관 cron 시도 0 assert [c["cron_id"] for c in spy.calls] == ["FB37-0001"] # binding-invalid 경로는 아예 remover 미호출 res2, spy2, _ = _run_wired( _fx("task-2553+37.fallback-id-missing"), tmp_path / "x" ) assert spy2.calls == [] # ── frozen anchor byte-0 ─────────────────────────────────────────────────── def test_frozen_anchor_byte0(): sha = hashlib.sha256(FROZEN.read_bytes()).hexdigest() assert sha == FROZEN_SHA, f"frozen anchor 변경 감지: {sha}" if __name__ == "__main__": raise SystemExit(pytest.main([__file__, "-q"]))