"""tests/regression/test_cancel_on_success_live_wiring_2553plus45.py task-2553+45 — CANCEL-ON-SUCCESS LIVE INTEGRATION regression (§6 1~20). 9-R.1: 전 케이스 mock/fixture/격리 — 실 cron-list·실 cron-remove·실 schedule_history·실 durable ledger(canonical) 무접촉. "cron-remove called" = 주입 SpyRemover 호출 spy assert (실 cron 삭제 아님). 실 subprocess 즉시 FAIL 차단. durable 4-tuple ledger 는 tmp JSONL 로 격리 주입. live /home/jay/workspace git tracked HEAD/branch/ref 전후 assertEqual (§5). §6 매핑: 1 normal success + fallback bound + live verifier PASS → cron-remove called 2 normal success + fallback_cron_id missing → cron-remove 0·preserved 3 task_id mismatch (durable lookup) → cron-remove 0 4 chat_id mismatch (durable lookup) → cron-remove 0 5 role not fallback (durable lookup) → cron-remove 0 6 normal collector failure/HOLD/partial → cron-remove 0 7 cron-remove exception → normal success preserved (디커플) 8 cancel-audit JSON 생성 + schema(jsonschema) valid 9 fallback duplicate safety path 무회귀 10 +41 fixture normal success 후 fallback 발화 재현 11 +41 fixture 수정 경로 cancel-on-success PASS 12 +39 fixture normal success 후 fallback 발화 재현 13 +39 fixture 수정 경로 cancel-on-success PASS 14 unrelated cron remove attempt 0 15 +32 mandatory normal callback contract 무회귀 16 +37 cancel wiring regression 무회귀 17 +44/+46 4-tuple/root resolver 호환 18 raw token/credential exposure 0 19 runtime checkpoint primary path 대체 없음 20 normal collector success 와 cancel failure decoupled """ from __future__ import annotations import json import subprocess import sys from pathlib import Path import jsonschema 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.cancel_on_success_live_wiring import ( # noqa: E402 LOOKUP_CHAT_MISMATCH, LOOKUP_NO_FALLBACK_BOUND, LOOKUP_NO_LEDGER_RECORD, LOOKUP_ROLE_MISMATCH, LOOKUP_VERIFIED, lookup_fallback_from_durable_registry, run_cancel_on_success_live_wiring, ) from anu_v3.operational_collector_wiring import ( # noqa: E402 LIVE_COLLECTOR_ENTRYPOINT, live_wiring_contract, resolve_live_collector_entrypoint, ) from anu_v3.cancel_audit_writer import ( # noqa: E402 REQUIRED_AUDIT_FIELDS, audit_is_complete, ) from utils.anu_delegation_completion_callback import ( # noqa: E402 CallbackInput, CallbackType, Classification, ) from utils.completion_callback_fallback_cancel import RemoverResult # noqa: E402 FIXDIR = WORKSPACE / "memory" / "fixtures" SCHEMA_AUDIT = json.loads( (WORKSPACE / "schemas" / "cancel_on_success_audit.schema.json").read_text( encoding="utf-8" ) ) SCHEMA_VERIFIER = json.loads( ( WORKSPACE / "schemas" / "live_cancel_verifier_result.schema.json" ).read_text(encoding="utf-8") ) 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 _ledger(tmp: Path, record: dict | None) -> Path: """tmp durable 4-tuple JSONL ledger (격리 — canonical ledger 무접촉).""" lp = tmp / "callback_4tuple_index.jsonl" lp.parent.mkdir(parents=True, exist_ok=True) if record is not None: lp.write_text( json.dumps(record, ensure_ascii=False, sort_keys=True) + "\n", encoding="utf-8", ) else: lp.write_text("", encoding="utf-8") return lp def _pass_input(task_id: str, *, callback_type=CallbackType.NORMAL): return CallbackInput( task_id=task_id, executor="dev-sim", dispatch_cron_id="DISP45", callback_type=callback_type, callback_cron_id="NORM45", cron_status="ok", task_status="completed", required_closeout_markers={"result_json": True, "report": True}, preservation_anchors={"frozen_anchor": "match"}, dev_sunset=True, ) def _run_live( fx, tmp, *, record="__fx__", inp=None, lister=None, remover=None, ): p = _materialize(fx, tmp) rec = fx["durable_4tuple_record"] if record == "__fx__" else record lp = _ledger(tmp / "ledger", rec) spy = remover or SpyRemover("removed") lst = lister or FakeCronLister(fx["live_cron_entries"]) res = run_cancel_on_success_live_wiring( inp or _pass_input(fx["task_id"]), tmp / "ack.json", 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", ledger_path=lp, 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 # ── §6.1 normal success + bound + live verifier PASS → cron-remove called ── def test_01_success_bound_verifier_pass_remove_called(tmp_path): fx = _fx("task-2553plus41.fallback-fired-after-normal-success") res, spy, lst = _run_live(fx, tmp_path) assert res.lookup.status == LOOKUP_VERIFIED assert res.wired_via_operational_collector_wiring 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 # operational=True → dry_run=False, bound id 1건만 remove 시도 assert spy.calls == [{"cron_id": "8A0E088E", "dry_run": False}] assert lst.calls == 1 assert res.cancel_audit["normal_success_unchanged"] is True assert res.cancel_audit["lookup_source"] == "durable_4tuple_registry" # ── §6.2 fallback_cron_id missing → cron-remove 0·preserved ─────────────── def test_02_fallback_missing_no_remove_preserved(tmp_path): fx = _fx("task-2553plus41.fallback-fired-after-normal-success") rec = dict(fx["durable_4tuple_record"]) rec["fallback_callback_cron_id"] = None rec["no_fallback"] = True res, spy, _ = _run_live(fx, tmp_path, record=rec) assert res.lookup.status == LOOKUP_NO_FALLBACK_BOUND assert res.seam_invoked is False assert res.cron_remove_invoked is False assert res.fallback_preserved is True assert spy.calls == [] assert res.cancel_audit["remove_attempted"] is False assert res.cancel_audit["normal_success_unchanged"] is True # ── §6.2b durable ledger record 부재 → cron-remove 0·preserved ──────────── def test_02b_no_ledger_record_no_remove(tmp_path): fx = _fx("task-2553plus41.fallback-fired-after-normal-success") res, spy, _ = _run_live(fx, tmp_path, record=None) assert res.lookup.status == LOOKUP_NO_LEDGER_RECORD assert res.seam_invoked is False assert spy.calls == [] assert res.fallback_preserved is True assert res.cancel_audit["lookup_status"] == LOOKUP_NO_LEDGER_RECORD # Codex HIGH 수용: lookup 실패에도 normal completion callback collector # 는 MANDATORY 로 1회 실행 — collector_result 보존(우회 0, §3 약화 0). assert res.collector_result is not None assert res.collector_result.classification == Classification.PASS assert res.durable_success is True assert res.normal_success_unchanged is True # ── §6.3 task_id mismatch (durable lookup) → cron-remove 0 ──────────────── def test_03_task_id_mismatch_no_remove(tmp_path): fx = _fx("task-2553plus41.fallback-fired-after-normal-success") rec = dict(fx["durable_4tuple_record"]) rec["task_id"] = "task-2553+UNRELATED-OTHER" # ledger 에는 unrelated task 만 → 본 task_id 조회 시 record 부재. res, spy, _ = _run_live(fx, tmp_path, record=rec) assert res.lookup.status == LOOKUP_NO_LEDGER_RECORD assert res.cron_remove_invoked is False assert spy.calls == [] # unrelated record 인용 0 # ── §6.4 chat_id mismatch (durable lookup) → cron-remove 0 ──────────────── def test_04_chat_id_mismatch_no_remove(tmp_path): fx = _fx("task-2553plus41.fallback-fired-after-normal-success") rec = dict(fx["durable_4tuple_record"]) rec["chat_id"] = "9999999999" res, spy, _ = _run_live(fx, tmp_path, record=rec) assert res.lookup.status == LOOKUP_CHAT_MISMATCH assert res.cron_remove_invoked is False assert spy.calls == [] assert res.cancel_audit["lookup_status"] == LOOKUP_CHAT_MISMATCH # ── §6.5 role not fallback (durable lookup) → cron-remove 0 ─────────────── def test_05_role_not_fallback_no_remove(tmp_path): fx = _fx("task-2553plus41.fallback-fired-after-normal-success") rec = dict(fx["durable_4tuple_record"]) rec["role"] = "normal" res, spy, _ = _run_live(fx, tmp_path, record=rec) assert res.lookup.status == LOOKUP_ROLE_MISMATCH assert res.cron_remove_invoked is False assert spy.calls == [] # ── §6.5b live verifier 5조건 mismatch (binding ok) → cron-remove 0 ─────── @pytest.mark.parametrize( "mutate", [ {"task_id": "task-2553+OTHER"}, {"chat_id": 1234}, {"role": "normal"}, {"fired": True}, ], ) def test_05b_live_verifier_mismatch_no_remove(tmp_path, mutate): fx = _fx("task-2553plus41.fallback-fired-after-normal-success") entry = dict(fx["live_cron_entries"][0]) entry.update(mutate) res, spy, _ = _run_live( fx, tmp_path, lister=FakeCronLister([entry]) ) # durable lookup 은 VERIFIED 이나 live verifier 5조건 미충족 → remove 0 assert res.lookup.status == LOOKUP_VERIFIED assert res.seam_invoked is True assert res.cron_remove_invoked is False assert spy.calls == [] # ── §6.6 normal collector failure/HOLD/partial → 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-2553plus41.fallback-fired-after-normal-success") inp = _pass_input(fx["task_id"]) for k, v in mutate.items(): setattr(inp, k, v) res, spy, _ = _run_live(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 == [] assert res.cancel_audit["normal_success_unchanged"] is True # ── §6.7 cron-remove exception → normal success preserved (디커플) ──────── def test_07_seam_exception_decoupled(tmp_path, monkeypatch): fx = _fx("task-2553plus41.fallback-fired-after-normal-success") 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_live(fx, tmp_path) # collector 성공은 seam 예외와 무관하게 PASS 유지 assert res.collector_result.classification == Classification.PASS assert res.collector_result.closeout_candidate is True assert res.normal_success_unchanged is True assert res.seam_invoked is False assert spy.calls == [] assert res.cancel_audit["normal_success_unchanged"] is True # ── §6.8 cancel-audit JSON 생성 + schema valid ──────────────────────────── def test_08_cancel_audit_generated_schema_valid(tmp_path): # bound·verified 경로 r1, _, _ = _run_live( _fx("task-2553plus41.fallback-fired-after-normal-success"), tmp_path / "a", ) assert audit_is_complete(r1.cancel_audit) assert set(REQUIRED_AUDIT_FIELDS).issubset(r1.cancel_audit.keys()) assert Path(r1.cancel_audit_path).exists() jsonschema.validate(r1.cancel_audit, SCHEMA_AUDIT) # lookup-skip 경로 r2, _, _ = _run_live( _fx("task-2553plus41.fallback-fired-after-normal-success"), tmp_path / "b", record=None, ) assert audit_is_complete(r2.cancel_audit) jsonschema.validate(r2.cancel_audit, SCHEMA_AUDIT) # live verifier 결과도 +23 schema 정합 so = r1.wired_result.wiring_result.seam_outcome jsonschema.validate(so.live_verification, SCHEMA_VERIFIER) # ── §6.9 fallback duplicate safety path 무회귀 ──────────────────────────── def test_09_duplicate_callback_no_regression(tmp_path): fx = _fx("task-2553plus41.fallback-fired-after-normal-success") p = _materialize(fx, tmp_path / "shared") lp = _ledger(tmp_path / "led", fx["durable_4tuple_record"]) ack = tmp_path / "shared-ack.json" claim = tmp_path / "claims" def _call(ctype, spy): return run_cancel_on_success_live_wiring( _pass_input(fx["task_id"], callback_type=ctype), ack, 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, ledger_path=lp, 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 미진입 # ── §6.10 +41 fixture normal success 후 fallback 발화 재현 ──────────────── def test_10_plus41_reproduction(tmp_path): fx = _fx("task-2553plus41.fallback-fired-after-normal-success") repro = fx["reproduction_before_plus45"] # 재현: durable record 부재 → lookup 불가 → seam 미진입 → fallback 잔존 res, spy, _ = _run_live(fx, tmp_path, record=None) assert res.lookup.status == repro["expected_lookup_status"] assert res.seam_invoked is repro["expected_seam_invoked"] assert res.cron_remove_invoked is repro["expected_cron_remove_invoked"] assert res.fallback_preserved is repro["expected_fallback_preserved"] assert spy.calls == [] # fallback 잔존 → 뒤늦게 발화하는 그 상태 # ── §6.11 +41 fixture 수정 경로 cancel-on-success PASS ──────────────────── def test_11_plus41_resolved(tmp_path): fx = _fx("task-2553plus41.fallback-fired-after-normal-success") fixed = fx["resolved_via_plus45"] res, spy, _ = _run_live(fx, tmp_path) assert res.lookup.status == fixed["expected_lookup_status"] assert ( res.wired_via_operational_collector_wiring is fixed["expected_wired_via_operational_collector_wiring"] ) assert res.durable_success is fixed["expected_durable_success"] assert res.seam_invoked is fixed["expected_seam_invoked"] assert res.cron_remove_invoked is fixed["expected_cron_remove_invoked"] assert res.fallback_preserved is fixed["expected_fallback_preserved"] so = res.wired_result.wiring_result.seam_outcome assert so.seam_classification == fixed["expected_seam_classification"] assert spy.calls == [{"cron_id": "8A0E088E", "dry_run": False}] # ── §6.12 +39 fixture normal success 후 fallback 발화 재현 ──────────────── def test_12_plus39_reproduction(tmp_path): fx = _fx("task-2553plus39.fallback-fired-after-normal-success") repro = fx["reproduction_before_plus45"] res, spy, _ = _run_live(fx, tmp_path, record=None) assert res.lookup.status == repro["expected_lookup_status"] assert res.seam_invoked is repro["expected_seam_invoked"] assert res.cron_remove_invoked is repro["expected_cron_remove_invoked"] assert res.fallback_preserved is repro["expected_fallback_preserved"] assert spy.calls == [] # ── §6.13 +39 fixture 수정 경로 cancel-on-success PASS ──────────────────── def test_13_plus39_resolved(tmp_path): fx = _fx("task-2553plus39.fallback-fired-after-normal-success") fixed = fx["resolved_via_plus45"] res, spy, _ = _run_live(fx, tmp_path) assert res.lookup.status == fixed["expected_lookup_status"] assert res.durable_success is fixed["expected_durable_success"] assert res.seam_invoked is fixed["expected_seam_invoked"] assert res.cron_remove_invoked is fixed["expected_cron_remove_invoked"] assert res.fallback_preserved is fixed["expected_fallback_preserved"] so = res.wired_result.wiring_result.seam_outcome assert so.seam_classification == fixed["expected_seam_classification"] assert spy.calls == [{"cron_id": "FB39-DEADMAN", "dry_run": False}] # +46 canonical-first root 사용 증명 (autoset-cwd false-missing 차단) assert res.lookup.canonical_root == "/home/jay/workspace" # ── §6.14 unrelated cron remove attempt 0 ───────────────────────────────── def test_14_no_unrelated_cron_remove(tmp_path): fx = _fx("task-2553plus41.fallback-fired-after-normal-success") res, spy, _ = _run_live(fx, tmp_path) # remover 는 오직 durable-bound target 으로만 1회 — 무관 cron 시도 0 assert [c["cron_id"] for c in spy.calls] == ["8A0E088E"] # lookup-skip 경로는 아예 remover 미호출 _, spy2, _ = _run_live(fx, tmp_path / "x", record=None) assert spy2.calls == [] # ── §6.15 +32 mandatory normal callback contract 무회귀 ─────────────────── def test_15_plus32_mandatory_contract_no_regression(tmp_path): from dispatch.executor_completion_contract import ( Callback4Tuple, validate_4tuple, ) fx = _fx("task-2553plus41.fallback-fired-after-normal-success") rec = fx["durable_4tuple_record"] ok = Callback4Tuple( task_id=rec["task_id"], dispatch_cron_id=rec["dispatch_cron_id"], normal_collector_cron_id=rec["normal_collector_cron_id"], fallback_callback_cron_id=rec["fallback_callback_cron_id"], ) assert validate_4tuple(ok) == [] # normal_collector_cron_id 누락 → +32 mandatory 위반 그대로 검출 bad = Callback4Tuple( task_id="t", dispatch_cron_id="D", normal_collector_cron_id=None, fallback_callback_cron_id="F", ) assert any( "normal_collector_cron_id missing" in r for r in validate_4tuple(bad) ) # durable lookup 이 mandatory 를 약화하지 않음 — VERIFIED 시 4-tuple 보유 res, _, _ = _run_live(fx, tmp_path) assert res.lookup.normal_collector_cron_id == "NORM4100" # ── §6.16 +37 cancel wiring regression 무회귀 ───────────────────────────── def test_16_plus37_wiring_no_regression(): import importlib m37 = importlib.import_module( "tests.regression.test_collector_path_wiring_2553plus37" ) # +37 표준 entrypoint·바인딩 검증기를 +45 가 약화하지 않음(동일 심볼). from utils.normal_completion_callback_collector_entrypoint import ( run_wired_normal_completion_callback_collector, ) assert callable(run_wired_normal_completion_callback_collector) assert hasattr(m37, "test_01_wired_success_bound_verifier_pass_remove_called") # ── §6.17 +44/+46 4-tuple/root resolver 호환 ────────────────────────────── def test_17_plus44_46_compat(tmp_path): from anu_v3.artifact_root_resolver import canonical_root, resolve_roots from anu_v3.callback_4tuple_registry import ( Callback4TupleRegistry, make_record, ) rec = make_record( task_id="task-2553+45-compat", dispatch_id="D", dispatch_cron_id="DC", executor="dev-sim", chat_id="6937032012", normal_collector_cron_id="N", fallback_callback_cron_id="FBX", role="fallback", status="COMPLETED", ) lp = tmp_path / "compat.jsonl" Callback4TupleRegistry(lp).append(rec) look = lookup_fallback_from_durable_registry( task_id="task-2553+45-compat", ledger_path=lp ) assert look.ok is True assert look.status == LOOKUP_VERIFIED assert look.fallback_cron_id == "FBX" # +46 canonical-first 사용 (autoset-cwd 단독 false-missing 금지) roots = resolve_roots(autoset_cwd=tmp_path) assert roots.search_order[0] == str(canonical_root()) assert look.canonical_root == str(canonical_root()) # ── §6.18 raw token/credential exposure 0 ───────────────────────────────── def test_18_no_credential_exposure(tmp_path): res, _, _ = _run_live( _fx("task-2553plus41.fallback-fired-after-normal-success"), tmp_path ) blob = json.dumps(res.cancel_audit, ensure_ascii=False) assert ANU_KEY_SECRET not in blob for mod in ( "cancel_on_success_live_wiring.py", "operational_collector_wiring.py", "cancel_audit_writer.py", ): src = (WORKSPACE / "anu_v3" / mod).read_text(encoding="utf-8") assert ANU_KEY_SECRET not in src contract = live_wiring_contract() assert ANU_KEY_SECRET not in json.dumps(contract, ensure_ascii=False) # ── §6.19 runtime checkpoint primary path 대체 없음 ────────────────────── def test_19_registry_checkpoint_not_primary(): import ast src = ( WORKSPACE / "anu_v3" / "cancel_on_success_live_wiring.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 } # runtime checkpoint/reconcile 모듈을 primary 로 import 하지 않는다. assert not any( "checkpoint" in m or "reconcile" in m for m in modules ) # live entrypoint 가 primary — registry/checkpoint = recovery layer assert resolve_live_collector_entrypoint() is run_cancel_on_success_live_wiring c = live_wiring_contract() assert "recovery layer only" in c["primary_path"] assert c["live_collector_entrypoint"] == LIVE_COLLECTOR_ENTRYPOINT # ── §6.20 normal collector success 와 cancel failure decoupled ──────────── def test_20_success_decoupled_from_cancel_failure(tmp_path): fx = _fx("task-2553plus41.fallback-fired-after-normal-success") # remover 가 실패를 반환해도 normal collector success 불변 res, spy, _ = _run_live( fx, tmp_path, remover=SpyRemover("failed") ) assert res.collector_result.classification == Classification.PASS assert res.durable_success is True assert res.normal_success_unchanged is True assert res.cancel_audit["normal_success_unchanged"] is True # remover 호출은 됐으나(시도) 실패 — fallback 보존, success 무변 assert spy.calls == [{"cron_id": "8A0E088E", "dry_run": False}] assert res.fallback_preserved is True if __name__ == "__main__": raise SystemExit(pytest.main([__file__, "-q"]))