"""tests/regression/test_cancel_on_success_live_adoption_2604.py task-2604 Track A — CANCEL_ON_SUCCESS LIVE ADOPTION regression (§1~§12). 문서-only 금지: 본 regression 은 **실 모듈 + 실 entrypoint** 를 실행한다 (mock-only FAIL — 모듈/entrypoint 자체를 mock 으로 대체하면 §11 meta-가드가 FAIL). cron-list/remove DI 만 전용 격리 FakeCronWorld/WorldSpyRemover 주입 (실 cron-list·실 cron-remove·실 schedule_history 무접촉). 실 subprocess 즉시 FAIL 차단. 실 shared callback_4tuple_index.jsonl(track 미소유 공유 durable 인프라) 은 전 케이스 byte-0 — self-proof 는 격리 tmp ledger 만 구성. live /home/jay/workspace git tracked HEAD/branch/ref 전후 assertEqual. § 매핑: 1 properly-bound 격리 ledger + 5조건 AND PASS → CANCEL_ADOPTED (cron-remove 실 호출[주입 spy]·fallback_cancelled·cron_removed) 2 spy "failed" → +9a REMOVE_FAILED_WARNING → CANCEL_FAILED_CLASSIFIED (normal success decouple 유지 — 무효화 0) 3 ownership mismatch(task_id) → unrelated cron 제거 방지 → NON_BLOCKING 4 no ledger record → 미식별 → NON_BLOCKING (보존) 5 live 5조건 미충족(entry fired) → seam SKIP_LIVE → NON_BLOCKING 6 실 entrypoint CLI main() self-proof → exit0·audit/decision/result 산출·schema valid·CANCEL_ADOPTED 7 decouple 불변: 전 분류에서 normal_success_unchanged=True 8 unrelated cron 미접촉: spy 는 bound target id 로만 호출 9 frozen +9a/+23/+25/+44/+45/+48/anchor byte-0 무회귀 10 shared callback_4tuple_index.jsonl track-scoped invariant — 이 track 이 task-2604 row append 0 + pre-existing line 변조 0 + append-only 성장(기존 prefix byte 보존). 정상 +53/+54 sanctioned cross-track append 는 무FAIL (글로벌 SHA 하드핀 폐기 — task-2604+1 HOLD remediation) 11 mock-only FAIL meta-가드 (실 모듈/실 entrypoint 실행 입증) 12 raw ANU key adoption-audit 노출 0 """ from __future__ import annotations import hashlib 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)) import anu_v3.cancel_on_success_live_adoption_2604 as ADOPT_MOD # noqa: E402 from anu_v3.cancel_on_success_live_adoption_2604 import ( # noqa: E402 ADOPT_CANCEL_ADOPTED, ADOPT_CANCEL_FAILED_CLASSIFIED, ADOPT_NON_BLOCKING, CancelOnSuccessAdoptionResult, run_cancel_on_success_live_adoption, ) from anu_v3.callback_4tuple_registry import ( # noqa: E402 Callback4TupleRegistry, ) from anu_v3.cancel_on_success_live_e2e import ( # noqa: E402 FakeCronWorld, ProperlyBound4Tuple, WorldSpyRemover, build_properly_bound_4tuple, ) import scripts.run_cancel_on_success_live_adoption_2604 as CLI # noqa: E402 from utils.anu_delegation_completion_callback import ( # noqa: E402 CallbackInput, CallbackType, ) SCHEMA = json.loads( ( WORKSPACE / "schemas" / "cancel_on_success_live_adoption_2604.schema.json" ).read_text(encoding="utf-8") ) ANU_KEY_SECRET = "c119085addb0f8b7" # §12 — adoption-audit 노출 시 FAIL # §9 frozen byte-0 baseline (리터럴 핀 — within-run tautology 금지). FROZEN_SHA256 = { "anu_v3/cancel_on_success_live_wiring.py": ( "ea33514d2289f0097a5344d2d5bbc0735b6385c3bbfe03aa5c6d9f24f970bf0a" ), "anu_v3/callback_4tuple_registry.py": ( "774d550628410d36962c23a7663c4b6dbf72789de7c7fd940871e9ad8280e5ab" ), "anu_v3/cancel_on_success_live_e2e.py": ( "fa1421fef4faf2d6f4dc0cb1ce0d026f5c7086b046056e39dd343fb629b3af9b" ), "anu_v3/cancel_audit_writer.py": ( "0c5e10e616fe8919492f6c774045e8ff12f3ebfd203ecd25f35ad9ebd58c14d9" ), "anu_v3/artifact_root_resolver.py": ( "0f1900f46f60fe504ec89b5c84967506e15250d71d6f413b31d805c783a613fb" ), "utils/completion_callback_fallback_cancel.py": ( "2ef661c49128ec6f1bd96bfbe6c1a227317e87ae31bf6deac025c66fd0e96d37" ), "utils/live_cron_state_verifier.py": ( "6bba56bd0dfc97e9155244251e00cdf3dcef2981a75dc5e2b966aa0b746d8d13" ), "utils/completion_callback_operational_cancel_seam.py": ( "14d6a0fb29e9de9276fcbadde832fe348e02333d21d57f29bd583407f1542590" ), "utils/normal_completion_callback_collector_entrypoint.py": ( "41af1b601f42f337301ca7a45c185349385dc625ca02df43810541baa9953db9" ), "utils/operational_collector_wiring.py": ( "0636b99cf5d7e69d880c426bd03c0ff77e97987d1780912c05cc23899b832c6f" ), "utils/anu_delegation_completion_callback.py": ( "83b3e307c8207c76a3e311c408aab4951373bd317896e51687d3007907b0c3d4" ), } # §10 — 실 shared durable 인프라: track 미소유 공유 (track 직접 write 0). # task-2604+1 HOLD remediation: 글로벌 SHA 하드핀 폐기. 정상 +53/+54 # sanctioned cross-track append 로 SHA 가 바뀌어도 본 track invariant 는 # track-scoped (이 track 의 task-2604 row append 0 / pre-existing prefix # byte 보존 / append-only 성장) 이므로 무FAIL. append-only 위반은 FAIL. SHARED_INDEX_REL = "memory/events/callback_4tuple_index.jsonl" def _sha(rel: str) -> str: return hashlib.sha256((WORKSPACE / rel).read_bytes()).hexdigest() @pytest.fixture(autouse=True) def _block_real_subprocess(monkeypatch): def _boom(*a, **k): # noqa: ANN001, ANN002, ANN003 raise AssertionError( "실 subprocess 호출 금지 (회장 §5) — 전용 격리 FakeCronWorld/" "WorldSpyRemover 만, 실 운영 cron 무접촉" ) 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 변경 감지 (위반)" @pytest.fixture(autouse=True) def _frozen_byte0_invariant(): before = {rel: _sha(rel) for rel in FROZEN_SHA256} yield for rel, pinned in FROZEN_SHA256.items(): assert before[rel] == pinned, f"{rel} baseline drift (pre §9)" assert _sha(rel) == pinned, f"{rel} byte-0 위반 (post §9)" def _shared_index_lines() -> list[str]: """실 shared ledger 의 현재 라인 스냅샷 (keepends — byte 단위 비교).""" return ( (WORKSPACE / SHARED_INDEX_REL) .read_text(encoding="utf-8") .splitlines(keepends=True) ) def _is_task2604_owned_row(line: str) -> bool: """이 track(task-2604) 소유 row 인가. 공유 인프라(callback_4tuple_index.jsonl)엔 본 track 직접 write 0 이 어야 하므로, append 된 라인 중 task_id 가 task-2604(또는 -ut/+N 변형) 인 것이 1건이라도 있으면 track 직접 write 위반. 다른 track(2606/2607/ 2609 등)의 sanctioned +53/+54 durable-success append 는 비대상.""" try: rec = json.loads(line) except (ValueError, TypeError): return False tid = str(rec.get("task_id", "")) return ( tid == "task-2604" or tid.startswith("task-2604-") or tid.startswith("task-2604+") ) @pytest.fixture(autouse=True) def _shared_index_track_scoped_invariant(): """§10 (task-2604+1 HOLD remediation) — track-scoped invariant. 글로벌 callback_4tuple_index.jsonl SHA 동일성 하드핀을 폐기한다(정상 +53/+54 sanctioned cross-track append 를 FAIL 처리하던 spec §4/§8 9-R.1 자가모순 해소). 본 track 검증: (a) 이 track 이 shared ledger 에 task-2604 row 를 append 0 (b) pre-existing line 변조 0 (기존 prefix byte 보존) (c) 파일은 append-only 로만 성장 (축소·prefix 변형 = FAIL) sibling/authorized +53/+54 durable-success append(다른 track 소유)는 비대상 — 무FAIL. isolated /tmp ledger self-proof 는 본 fixture 와 독립(테스트 본문 — cron_removed/spy_calls 그대로 유지). """ before = _shared_index_lines() n_before = len(before) yield after = _shared_index_lines() # (c) append-only — 축소/재작성 = FAIL assert len(after) >= n_before, ( "shared callback_4tuple_index.jsonl 축소/재작성 감지 — " "append-only 위반 (§10 / 회장 §3 — append-only 위반은 FAIL)" ) # (b) pre-existing prefix byte 보존 (기존 line 변조 0) = FAIL assert after[:n_before] == before, ( "shared callback_4tuple_index.jsonl pre-existing line 변조 — " "기존 prefix 보존 위반 (§10 / 회장 §3)" ) # (a) 본 track 이 task-2604 row 를 공유 인프라에 append 0 offending = [ln for ln in after[n_before:] if _is_task2604_owned_row(ln)] assert not offending, ( "track 직접 write 0 위반 — task-2604 소유 row 가 shared " f"callback_4tuple_index.jsonl 에 append 됨: {offending!r} " "(§10 / 회장 §4 — track 미소유 공유 인프라, +53/+54 소관 only)" ) def _inp(task_id: str) -> CallbackInput: return CallbackInput( task_id=task_id, executor="dev-sim", dispatch_cron_id="DISP2604", callback_type=CallbackType.NORMAL, callback_cron_id="NORM2604", cron_status="ok", task_status="completed", required_closeout_markers={"result_json": True, "report": True}, preservation_anchors={"frozen_anchor": "match"}, dev_sunset=True, ) def _marker(task_id: str, fb_cron: str) -> dict: return { "ts_utc": "2026-05-19T00:00:00Z", "marker_kind": "dispatch fired", "dispatch_cron_id": "DISP2604", "callback_policy_a": { "standardized": True, "normal_callback_cron_id": "NORM2604", "fallback_callback_cron_id": fb_cron, "fallback_role": "fallback", "chat_id": 6937032012, "anu_key": ANU_KEY_SECRET, }, "task_id": task_id, } def _materialize(tmp: Path, task_id: str, fb_cron: str) -> dict: tmp.mkdir(parents=True, exist_ok=True) dfm = tmp / "dispatch-fired.json" dfm.write_text(json.dumps(_marker(task_id, fb_cron)), encoding="utf-8") rj = tmp / "result.json" rj.write_text( json.dumps( { "schema": "anu_delegation_result_v1", "task_id": task_id, "status": "completed", "classification": "PASS", } ), encoding="utf-8", ) rep = tmp / "report.md" rep.write_text(f"# {task_id}\nPASS durable-success\n", encoding="utf-8") crm = tmp / "collector-result.json" crm.write_text( json.dumps( { "schema": "callback_ack_dedupe_v1", "task_id": task_id, "classification": "PASS", "ack_acquired": True, } ), encoding="utf-8", ) return { "dispatch_fired_marker_path": dfm, "result_json_path": rj, "report_path": rep, "collector_result_marker_path": crm, "fallback_cancelled_marker_path": tmp / "fb-cancelled.json", "cancel_lock_path": tmp / "cancel.lock", "seam_audit_path": tmp / "plus23-cancel-audit.json", } def _build_isolated_ledger( tmp: Path, task_id: str, fb_cron: str, *, role: str = "fallback" ) -> Path: """격리 tmp ledger 에만 properly-bound 4-tuple 구성 (실 shared index 무접촉 — +44 guard + +47 write-back, read-only consume API).""" ledger = tmp / "isolated_callback_4tuple_index.jsonl" registry = Callback4TupleRegistry(ledger) tup = ProperlyBound4Tuple( task_id=task_id, dispatch_id="DSP-2604-PB", dispatch_cron_id="DISP2604", executor="dev-sim", chat_id="6937032012", normal_collector_cron_id="NORM2604", fallback_callback_cron_id=fb_cron, role=role, ) build_properly_bound_4tuple(registry, tup, ts_kst="2026-05-19 09:00 KST") return ledger def _world(task_id: str, fb_cron: str, *, fired: bool = False, extra=None): entries = [ { "id": fb_cron, "task_id": task_id, "chat_id": 6937032012, "role": "fallback", "fired": fired, "removed": False, } ] if extra: entries.extend(extra) return FakeCronWorld(entries) def _run( tmp: Path, *, task_id="task-2604-ut", fb_cron="UT2604FB", ledger=None, world=None, spy_status="removed", world_fired=False, ) -> tuple[CancelOnSuccessAdoptionResult, WorldSpyRemover, FakeCronWorld]: paths = _materialize(tmp / "claim", task_id, fb_cron) world = world or _world(task_id, fb_cron, fired=world_fired) spy = WorldSpyRemover(world, status=spy_status) res = run_cancel_on_success_live_adoption( _inp(task_id), tmp / "ack.marker", dispatch_fired_marker_path=paths["dispatch_fired_marker_path"], result_json_path=paths["result_json_path"], report_path=paths["report_path"], collector_result_marker_path=paths["collector_result_marker_path"], claim_dir=tmp / "claim", ledger_path=ledger, adoption_audit_path=tmp / "adoption-audit.json", cron_lister=world.lister, remover=spy, fallback_cancelled_marker_path=paths["fallback_cancelled_marker_path"], cancel_lock_path=paths["cancel_lock_path"], seam_audit_path=paths["seam_audit_path"], callback_contract={ "schema": "completion_callback_contract_v1", "fallback_callback_cron_id": fb_cron, }, ) return res, spy, world # ── §1 CANCEL_ADOPTED ──────────────────────────────────────────────────────── def test_s1_cancel_adopted_end_to_end(tmp_path): ledger = _build_isolated_ledger(tmp_path, "task-2604-ut", "UT2604FB") res, spy, world = _run(tmp_path, ledger=ledger) assert res.adoption_classification == ADOPT_CANCEL_ADOPTED assert res.cancel_adopted is True assert res.identification.identified is True assert res.identification.ownership_all_satisfied is True assert res.lookup_status == "LOOKUP_VERIFIED" assert res.seam_invoked is True assert res.cron_remove_invoked is True assert res.fallback_cancelled is True # 실 cancel/remove 입증: 주입 spy 실호출 + 격리 세계 제거. assert len(spy.calls) == 1 assert spy.calls[0]["cron_id"] == "UT2604FB" assert any( e["id"] == "UT2604FB" and e["removed"] for e in world.entries ) assert res.normal_success_unchanged is True assert res.decouple_invariant_held is True jsonschema.validate(res.adoption_audit, SCHEMA) # ── §2 CANCEL_FAILED_CLASSIFIED (decouple) ─────────────────────────────────── def test_s2_remove_failed_classified_decoupled(tmp_path): ledger = _build_isolated_ledger(tmp_path, "task-2604-ut", "UT2604FB") res, spy, world = _run(tmp_path, ledger=ledger, spy_status="failed") assert res.adoption_classification == ADOPT_CANCEL_FAILED_CLASSIFIED assert res.cancel_failed_classified is True assert res.remove_result == "REMOVE_FAILED_CLASSIFIED" assert len(spy.calls) == 1 # remove 시도는 했음 # remove 실패가 normal collector durable-success 를 뒤집지 0 (디커플). assert res.normal_success_unchanged is True assert res.decouple_invariant_held is True assert res.durable_success is True jsonschema.validate(res.adoption_audit, SCHEMA) # ── §3 ownership mismatch → unrelated cron 제거 방지 → NON_BLOCKING ─────────── def test_s3_ownership_mismatch_preserves(tmp_path): # ledger 는 다른 task_id 로 구성 → 처리 task_id 와 4-tuple 불일치. ledger = _build_isolated_ledger( tmp_path, "task-2604-OTHER", "UT2604FB" ) res, spy, world = _run(tmp_path, task_id="task-2604-ut", ledger=ledger) assert res.adoption_classification == ADOPT_NON_BLOCKING assert res.non_blocking_marked is True assert res.cron_remove_invoked is False assert len(spy.calls) == 0 # unrelated cron 제거 0 assert res.fallback_preserved is True assert res.normal_success_unchanged is True jsonschema.validate(res.adoption_audit, SCHEMA) # ── §4 no ledger record → 미식별 → NON_BLOCKING ────────────────────────────── def test_s4_no_ledger_record_preserves(tmp_path): empty = tmp_path / "empty_ledger.jsonl" empty.write_text("", encoding="utf-8") res, spy, _ = _run(tmp_path, ledger=empty) assert res.adoption_classification == ADOPT_NON_BLOCKING assert res.identification.identified is False assert res.identification.ident_status == "NO_LEDGER_RECORD" assert len(spy.calls) == 0 assert res.normal_success_unchanged is True jsonschema.validate(res.adoption_audit, SCHEMA) # ── §5 live 5조건 미충족 (entry fired) → NON_BLOCKING ───────────────────────── def test_s5_live_five_condition_fail_preserves(tmp_path): ledger = _build_isolated_ledger(tmp_path, "task-2604-ut", "UT2604FB") res, spy, world = _run(tmp_path, ledger=ledger, world_fired=True) assert res.adoption_classification == ADOPT_NON_BLOCKING assert res.cron_remove_invoked is False assert len(spy.calls) == 0 assert res.fallback_preserved is True assert res.normal_success_unchanged is True jsonschema.validate(res.adoption_audit, SCHEMA) # ── §6 실 entrypoint CLI main() self-proof ─────────────────────────────────── def test_s6_real_entrypoint_self_proof(tmp_path, monkeypatch): out = tmp_path / "task-2604.cancel-adoption-audit.json" decision = tmp_path / "task-2604.decision.json" result = tmp_path / "task-2604.result.json" monkeypatch.setattr(CLI, "ADOPTION_AUDIT_OUT", out) monkeypatch.setattr(CLI, "DECISION_OUT", decision) monkeypatch.setattr(CLI, "RESULT_OUT", result) monkeypatch.setattr(CLI, "EVENTS", tmp_path) rc = CLI.main(["--mode", "self-proof", "--out", str(out)]) assert rc == 0 assert out.exists() and decision.exists() and result.exists() audit = json.loads(out.read_text(encoding="utf-8")) jsonschema.validate(audit, SCHEMA) assert audit["adoption_classification"] == ADOPT_CANCEL_ADOPTED assert audit["normal_success_unchanged"] is True assert audit["decouple_invariant_held"] is True rj = json.loads(result.read_text(encoding="utf-8")) assert rj["status"] == "ok" assert rj["cancel_adopted"] is True # ── §7 decouple 불변 (전 분류) ─────────────────────────────────────────────── def test_s7_decouple_invariant_all_classes(tmp_path): led = _build_isolated_ledger(tmp_path / "a", "task-2604-ut", "FB1") r_adopt, _, _ = _run(tmp_path / "a", fb_cron="FB1", ledger=led) led2 = _build_isolated_ledger(tmp_path / "b", "task-2604-ut", "FB2") r_fail, _, _ = _run( tmp_path / "b", fb_cron="FB2", ledger=led2, spy_status="failed" ) r_nb, _, _ = _run( tmp_path / "c", fb_cron="FB3", ledger=(tmp_path / "c" / "noledger.jsonl"), ) for r in (r_adopt, r_fail, r_nb): assert r.normal_success_unchanged is True assert r.decouple_invariant_held is True assert r.adoption_audit["normal_success_unchanged"] is True # ── §8 unrelated cron 미접촉 (spy bound target only) ────────────────────────── def test_s8_unrelated_cron_untouched(tmp_path): ledger = _build_isolated_ledger(tmp_path, "task-2604-ut", "UT2604FB") world = _world( "task-2604-ut", "UT2604FB", extra=[ { "id": "UNRELATED-CRON-9", "task_id": "task-OTHER", "chat_id": 6937032012, "role": "fallback", "fired": False, "removed": False, } ], ) res, spy, world = _run(tmp_path, ledger=ledger, world=world) assert res.adoption_classification == ADOPT_CANCEL_ADOPTED assert [c["cron_id"] for c in spy.calls] == ["UT2604FB"] assert not any( e["id"] == "UNRELATED-CRON-9" and e["removed"] for e in world.entries ) # ── §11 mock-only FAIL meta-가드 (실 모듈/실 entrypoint 실행 입증) ──────────── def test_s11_real_module_not_mock(tmp_path): # 모듈/entrypoint 가 WORKSPACE 실파일에서 로드됐는가 (mock 대체 아님). assert Path(ADOPT_MOD.__file__).resolve() == ( WORKSPACE / "anu_v3" / "cancel_on_success_live_adoption_2604.py" ) assert Path(CLI.__file__).resolve() == ( WORKSPACE / "scripts" / "run_cancel_on_success_live_adoption_2604.py" ) ledger = _build_isolated_ledger(tmp_path, "task-2604-ut", "UT2604FB") res, spy, _ = _run(tmp_path, ledger=ledger) # 실 결과 타입·실 +45 underlying audit 결선 입증 (mock-only 면 부재). assert isinstance(res, CancelOnSuccessAdoptionResult) assert res.live_result is not None assert res.underlying_cancel_audit.get("schema", "").startswith( "task-2553+45" ) assert res.wired_via_operational_collector_wiring if hasattr( res, "wired_via_operational_collector_wiring" ) else True # ── §12 raw ANU key adoption-audit 노출 0 ──────────────────────────────────── def test_s12_no_raw_anu_key_in_adoption_fields(tmp_path): ledger = _build_isolated_ledger(tmp_path, "task-2604-ut", "UT2604FB") res, _, _ = _run(tmp_path, ledger=ledger) # adoption 자체 산출 필드(식별/분류/사유/notes)에 raw key 박힘 0. adoption_view = { k: v for k, v in res.adoption_audit.items() if k != "underlying_cancel_audit" } assert ANU_KEY_SECRET not in json.dumps(adoption_view, ensure_ascii=False)