"""tests/regression/test_cancel_on_success_live_e2e_2553plus48.py task-2553+48 — cancel-on-success live remove END-TO-END regression (§7 1~10). 9-R.1 / 회장 §5: 전 케이스 전용 격리 FakeCronWorld(in-memory) + WorldSpyRemover 주입. 실 cron-list·실 cron-remove·실 schedule_history·실 durable ledger(canonical) 무접촉. "cron-remove called" = 주입 WorldSpyRemover 호출 spy assert (실 운영 cron 강제삭제 0 — 회장 §5 준수, Layer A). 실 subprocess 즉시 FAIL 차단. durable 4-tuple ledger 는 tmp JSONL 격리. live /home/jay/workspace git tracked HEAD/branch/ref 전후 assertEqual (§4). bound·verified single cancel 1건만 (Layer B). §7 매핑: 1 properly-bound 4-tuple(live==ledger·role=fallback)+5조건 AND PASS → cron-remove 실제 호출·cancel-audit 생성 2 normal success 후 fallback 발화 0 end-to-end 재현 3 ROLE/TRACK/CHAT mismatch → 보존(무회귀) 4 5조건 1개 fail → 보존 5 cron-remove 실패 → normal success decouple 유지 6 DUPLICATE_CALLBACK_IGNORED safety 무회귀 7 실 운영 cron 무접촉(Fake/Spy 격리) 8 +37/+25/+23/+44/+45/+46/+47 byte-0·regression 무회귀 9 +32 callback mandatory 무회귀 10 raw token/credential exposure 0 """ from __future__ import annotations import hashlib import importlib 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_e2e import ( # noqa: E402 E2E_DECOUPLED, E2E_PASS, E2E_PRESERVED, FakeCronWorld, ProperlyBound4Tuple, WorldSpyRemover, build_properly_bound_4tuple, run_cancel_on_success_live_e2e, verify_4tuple_binding_integrity, ) from anu_v3.cancel_on_success_live_wiring import ( # noqa: E402 LOOKUP_VERIFIED, ) from anu_v3.callback_4tuple_registry import ( # noqa: E402 Callback4TupleRegistry, ) from utils.anu_delegation_completion_callback import ( # noqa: E402 CallbackInput, CallbackType, Classification, ) FIXDIR = WORKSPACE / "memory" / "fixtures" SCHEMA_E2E = json.loads( ( WORKSPACE / "schemas" / "cancel_on_success_e2e_audit.schema.json" ).read_text(encoding="utf-8") ) SCHEMA_AUDIT = json.loads( (WORKSPACE / "schemas" / "cancel_on_success_audit.schema.json").read_text( encoding="utf-8" ) ) ANU_KEY_SECRET = "c119085addb0f8b7" # 노출 검사용 — 산출물에 박히면 FAIL # 회장 §4/§6/§7.8 — frozen 모듈 byte-0 baseline (within-run tautology 금지: # 리터럴 핀). regression 후 EQUAL 이어야 무회귀. FROZEN_SHA256 = { "anu_v3/cancel_on_success_live_wiring.py": ( "ea33514d2289f0097a5344d2d5bbc0735b6385c3bbfe03aa5c6d9f24f970bf0a" ), "anu_v3/callback_4tuple_registry.py": ( "774d550628410d36962c23a7663c4b6dbf72789de7c7fd940871e9ad8280e5ab" ), "anu_v3/callback_event_trigger.py": ( "352ad0f570e55040e7c1e4a32cbfe0f076cbd53529b4db6222a8da1a4bee9cc5" ), "anu_v3/cancel_audit_writer.py": ( "0c5e10e616fe8919492f6c774045e8ff12f3ebfd203ecd25f35ad9ebd58c14d9" ), "anu_v3/artifact_root_resolver.py": ( "0f1900f46f60fe504ec89b5c84967506e15250d71d6f413b31d805c783a613fb" ), "utils/normal_completion_callback_collector_entrypoint.py": ( "41af1b601f42f337301ca7a45c185349385dc625ca02df43810541baa9953db9" ), "utils/operational_collector_wiring.py": ( "0636b99cf5d7e69d880c426bd03c0ff77e97987d1780912c05cc23899b832c6f" ), "utils/live_cron_state_verifier.py": ( "6bba56bd0dfc97e9155244251e00cdf3dcef2981a75dc5e2b966aa0b746d8d13" ), "utils/anu_delegation_completion_callback.py": ( "83b3e307c8207c76a3e311c408aab4951373bd317896e51687d3007907b0c3d4" ), "dispatch/executor_completion_contract.py": ( "364caa11904285657abd716d78c5493b1f8b519318387d0f864fb6a136dca0b4" ), } def _fx(name: str) -> dict: return json.loads((FIXDIR / f"{name}.json").read_text(encoding="utf-8")) 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 / 9-R.1 Layer A) — 전용 격리 " "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 변경 감지 (§4 위반)" @pytest.fixture(autouse=True) def _frozen_byte0_invariant(): """모든 케이스 실행 전후 frozen 모듈 byte-0 (§7.8 무회귀).""" 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)" assert _sha(rel) == pinned, f"{rel} byte-0 위반 (post, §7.8)" 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 _tup(fx: dict) -> ProperlyBound4Tuple: t = fx["properly_bound_4tuple"] return ProperlyBound4Tuple( task_id=t["task_id"], dispatch_id=t["dispatch_id"], dispatch_cron_id=t["dispatch_cron_id"], executor=t["executor"], chat_id=t["chat_id"], normal_collector_cron_id=t["normal_collector_cron_id"], fallback_callback_cron_id=t["fallback_callback_cron_id"], role=t["role"], ) def _inp(task_id: str, *, callback_type=CallbackType.NORMAL): return CallbackInput( task_id=task_id, executor="dev-sim", dispatch_cron_id="DISP4800", callback_type=callback_type, callback_cron_id="NORM4800", cron_status="ok", task_status="completed", required_closeout_markers={"result_json": True, "report": True}, preservation_anchors={"frozen_anchor": "match"}, dev_sunset=True, ) def _run_e2e( fx, tmp, *, inp=None, remover_status="removed", skip_binding=False, world_entries=None, expected_chat_id=6937032012, expected_role="fallback", ): p = _materialize(fx, tmp) world = FakeCronWorld( world_entries if world_entries is not None else fx["live_cron_entries"] ) spy = WorldSpyRemover(world, status=remover_status) res = run_cancel_on_success_live_e2e( inp or _inp(fx["task_id"]), tmp / "ack.json", _tup(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", ledger_path=tmp / "ledger" / "callback_4tuple_index.jsonl", dispatch_marker=fx["dispatch_fired_marker"], world=world, remover=spy, e2e_audit_path=tmp / "e2e-audit.json", 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"), expected_chat_id=expected_chat_id, expected_role=expected_role, skip_binding_construction=skip_binding, ) return res, spy, world # ── §7.1 properly-bound + 5조건 AND PASS → cron-remove·cancel-audit ──────── def test_01_properly_bound_e2e_pass_remove_and_audit(tmp_path): fx = _fx("task-2553plus48.properly-bound-4tuple") exp = fx["resolved_via_plus48"] res, spy, world = _run_e2e(fx, tmp_path) # properly-bound 4-tuple = +44 guard REGISTERED + +47 write-back COMPLETED bc = res.binding_construction assert bc.guard_registered_appended is True assert bc.writeback_status == exp["expected_writeback_status"] assert bc.writeback_appended is True bi = res.binding_integrity assert bi.ok is exp["expected_binding_integrity_ok"] assert bi.live_eq_ledger is exp["expected_live_eq_ledger"] assert bi.marker_eq_ledger is exp["expected_marker_eq_ledger"] assert bi.role_is_fallback is exp["expected_role_is_fallback"] # +45 경유 — LOOKUP_VERIFIED → 5조건 AND PASS → cron-remove 실제 호출 assert res.properly_bound is True assert res.live_result.lookup.status == LOOKUP_VERIFIED assert res.durable_success is exp["expected_durable_success"] assert res.seam_invoked is exp["expected_seam_invoked"] assert res.cron_remove_invoked is exp["expected_cron_remove_invoked"] assert res.fallback_cancelled is exp["expected_fallback_cancelled"] assert res.fallback_preserved is exp["expected_fallback_preserved"] so = res.live_result.wired_result.wiring_result.seam_outcome assert so.seam_classification == exp["expected_seam_classification"] five = res.e2e_audit["five_condition_results"] assert five["all_satisfied"] is True # bound·verified single cancel 1건만 (Layer B) — 격리 spy 1회·정확 id assert spy.calls == [{"cron_id": "PB48FALLBACK", "dry_run": False}] assert world.list_calls == 1 # cancel-audit 생성 + e2e schema valid + nested +45 audit schema valid assert res.verdict == E2E_PASS assert Path(res.e2e_audit_path).exists() jsonschema.validate(res.e2e_audit, SCHEMA_E2E) jsonschema.validate(res.e2e_audit["cancel_audit"], SCHEMA_AUDIT) assert res.e2e_audit["normal_success_unchanged"] is True # ── §7.2 normal success 후 fallback 발화 0 end-to-end ───────────────────── def test_02_fallback_fire_zero_end_to_end(tmp_path): fx = _fx("task-2553plus48.properly-bound-4tuple") res, spy, world = _run_e2e(fx, tmp_path) fp = res.fallback_fire_proof # 실 제거 후 normal success 후 fallback 발화 시도 전부 미발화(0) assert fp.cron_removed is True assert fp.post_success_fire_attempts == 3 assert fp.fallback_fired_count == 0 assert res.e2e_audit["fallback_fire_proof"]["fallback_fired_count"] == 0 # 격리 세계의 bound fallback entry 가 실제 removed (live==ledger 1건만) entry = next(e for e in world.entries if e["id"] == "PB48FALLBACK") assert entry["removed"] is True assert entry.get("fired") in (False, None) # 재현(보존): properly-bound 미구성 → seam 미진입 → bound fallback # 미제거(보존) → normal success 후 fallback 이 뒤늦게 발화(=1). 이것이 # +45 self-test ROLE_MISMATCH 보존으로 live remove 가 미실증이던 정확한 # 결함. +48 properly-bound 경로에서 발화 0 으로 해소. 두 경로 모두 # normal success 는 디커플로 무변. fx2 = _fx("task-2553plus48.properly-bound-4tuple") repro = fx2["reproduction_before_plus48"] # 외부 ledger 비움 + binding 구성 skip → LOOKUP_NO_LEDGER_RECORD 보존 res2, spy2, world2 = _run_e2e( fx2, tmp_path / "repro", skip_binding=True ) assert res2.verdict == repro["expected_verdict"] assert res2.seam_invoked is repro["expected_seam_invoked"] assert res2.cron_remove_invoked is repro["expected_cron_remove_invoked"] assert res2.fallback_preserved is repro["expected_fallback_preserved"] assert res2.fallback_fire_proof.cron_removed is False # 미제거 fallback 은 뒤늦게 발화 — 이것이 +48 이 닫는 그 갭(=1). assert res2.fallback_fire_proof.fallback_fired_count == ( repro["expected_fallback_fired_count"] ) # 그러나 normal success 는 디커플로 무변 (회장 §3 절대불변). assert res2.normal_success_unchanged is True assert res2.e2e_audit["normal_success_unchanged"] is True assert spy2.calls == [] # ── §7.3 ROLE/TRACK/CHAT mismatch → 보존 (무회귀) ───────────────────────── @pytest.mark.parametrize( "mut,expect_lookup", [ ({"role": "normal"}, "LOOKUP_ROLE_MISMATCH"), ({"chat_id": "9999999999"}, "LOOKUP_CHAT_MISMATCH"), ({"task_id": "task-2553+48-OTHER"}, "LOOKUP_NO_LEDGER_RECORD"), ], ) def test_03_mismatch_preserved_no_regression(tmp_path, mut, expect_lookup): fx = _fx("task-2553plus48.properly-bound-4tuple") tup = _tup(fx) for k, v in mut.items(): setattr(tup, k, v) p = _materialize(fx, tmp_path) world = FakeCronWorld(fx["live_cron_entries"]) spy = WorldSpyRemover(world) res = run_cancel_on_success_live_e2e( _inp(fx["task_id"]), tmp_path / "ack.json", tup, 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 / "claims", ledger_path=tmp_path / "led" / "callback_4tuple_index.jsonl", dispatch_marker=fx["dispatch_fired_marker"], world=world, remover=spy, e2e_audit_path=tmp_path / "e2e.json", callback_contract=fx.get("callback_contract"), ) assert res.live_result.lookup.status == expect_lookup assert res.verdict == E2E_PRESERVED assert res.seam_invoked is False assert res.cron_remove_invoked is False assert res.fallback_preserved is True # 보존 경로 — bound fallback 미제거(보수 가드 무회귀). normal success # 는 디커플로 무변 (회장 §3). live remove 강제 0 = +45 보수 가드 무회귀. assert res.fallback_fire_proof.cron_removed is False assert res.normal_success_unchanged is True assert res.e2e_audit["normal_success_unchanged"] is True assert spy.calls == [] jsonschema.validate(res.e2e_audit, SCHEMA_E2E) # ── §7.4 5조건 1개 fail → 보존 ──────────────────────────────────────────── @pytest.mark.parametrize( "mutate", [ {"task_id": "task-2553+OTHER"}, {"chat_id": 1234}, {"role": "normal"}, {"fired": True}, ], ) def test_04_five_condition_one_fail_preserved(tmp_path, mutate): fx = _fx("task-2553plus48.properly-bound-4tuple") entry = dict(fx["live_cron_entries"][0]) entry.update(mutate) res, spy, _ = _run_e2e(fx, tmp_path, world_entries=[entry]) # durable lookup VERIFIED 이나 live verifier 5조건 미충족 → remove 0 assert res.live_result.lookup.status == LOOKUP_VERIFIED assert res.seam_invoked is True assert res.cron_remove_invoked is False assert res.fallback_cancelled is False assert res.verdict == E2E_PRESERVED # 5조건 1개라도 fail → live remove 미시도(보존), normal success 디커플. assert res.fallback_fire_proof.cron_removed is False assert res.normal_success_unchanged is True assert spy.calls == [] # ── §7.5 cron-remove 실패 → normal success decouple 유지 ────────────────── def test_05_remove_failed_decoupled(tmp_path): fx = _fx("task-2553plus48.properly-bound-4tuple") res, spy, world = _run_e2e(fx, tmp_path, remover_status="failed") # remover 시도했으나 실패 → fallback 보존, normal success 불변(디커플) assert res.live_result.lookup.status == LOOKUP_VERIFIED assert res.seam_invoked is True assert res.cron_remove_invoked is True assert res.fallback_cancelled is False assert res.fallback_preserved is True assert res.durable_success is True assert res.normal_success_unchanged is True assert res.e2e_audit["normal_success_unchanged"] is True assert res.verdict == E2E_DECOUPLED # remover 호출은 됐으나 실패 — 격리 세계 entry 미제거 assert spy.calls == [{"cron_id": "PB48FALLBACK", "dry_run": False}] entry = next(e for e in world.entries if e["id"] == "PB48FALLBACK") assert entry["removed"] is False # ── §7.6 DUPLICATE_CALLBACK_IGNORED safety 무회귀 ───────────────────────── def test_06_duplicate_callback_safety_no_regression(tmp_path): fx = _fx("task-2553plus48.properly-bound-4tuple") p = _materialize(fx, tmp_path / "shared") ack = tmp_path / "shared-ack.json" claim = tmp_path / "claims" ledger = tmp_path / "led" / "callback_4tuple_index.jsonl" world = FakeCronWorld(fx["live_cron_entries"]) def _call(ctype, spy): return run_cancel_on_success_live_e2e( _inp(fx["task_id"], callback_type=ctype), ack, _tup(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=claim, ledger_path=ledger, dispatch_marker=fx["dispatch_fired_marker"], world=world, remover=spy, e2e_audit_path=tmp_path / f"e2e-{ctype.name}.json", fallback_cancelled_marker_path=p[ "fallback_cancelled_marker_path" ], cancel_lock_path=p["cancel_lock_path"], ) first = _call(CallbackType.NORMAL, WorldSpyRemover(world)) spy2 = WorldSpyRemover(world) second = _call(CallbackType.FALLBACK_STALE, spy2) assert first.live_result.collector_result.classification == ( Classification.PASS ) assert first.seam_invoked is True assert first.verdict == E2E_PASS assert second.live_result.collector_result.classification == ( Classification.DUPLICATE_CALLBACK_IGNORED ) assert second.durable_success is False assert second.seam_invoked is False # 후발 fallback seam 미진입 — 안전망 무회귀, 추가 remove 0 assert spy2.calls == [] assert second.normal_success_unchanged is True # ── §7.7 실 운영 cron 무접촉 (Fake/Spy 격리) ────────────────────────────── def test_07_no_real_ops_cron_contact(tmp_path): import ast src = ( WORKSPACE / "anu_v3" / "cancel_on_success_live_e2e.py" ).read_text(encoding="utf-8") tree = ast.parse(src) # 모듈 import 면에서 subprocess/os.system/cokacdir 직접 import 0. imported = { n.module for n in ast.walk(tree) if isinstance(n, ast.ImportFrom) and n.module } assert "subprocess" not in imported assert not any("cokacdir" in (m or "") for m in imported) # 실행 문자열에 실 cron register/remove 명령 0. assert "cokacdir --cron" not in src assert "subprocess.run" not in src # 실증은 전용 격리 FakeCronWorld 로만 — 실제 cron 삭제 0. fx = _fx("task-2553plus48.properly-bound-4tuple") res, spy, world = _run_e2e(fx, tmp_path) assert isinstance(world, FakeCronWorld) assert world.entries is not fx["live_cron_entries"] # 깊은 복사(부작용 0) assert [c["cron_id"] for c in spy.calls] == ["PB48FALLBACK"] # ── §7.8 +37/+25/+23/+44/+45/+46/+47 byte-0·regression 무회귀 ───────────── def test_08_frozen_byte0_and_cross_regression_no_regression(): # byte-0 (autouse _frozen_byte0_invariant 가 전후 강제) — 명시 재확인 for rel, pinned in FROZEN_SHA256.items(): assert _sha(rel) == pinned, f"{rel} byte-0 위반 (§7.8)" # 선행 regression 모듈 import 가능·핵심 케이스 보유 (무회귀 — 동일 심볼) m45 = importlib.import_module( "tests.regression.test_cancel_on_success_live_wiring_2553plus45" ) m47 = importlib.import_module( "tests.regression.test_callback_event_trigger_2553plus47" ) m44 = importlib.import_module( "tests.regression.test_callback_4tuple_registry_2553plus44" ) import unittest assert hasattr( m45, "test_01_success_bound_verifier_pass_remove_called" ) # +47/+44 는 unittest.TestCase 기반 — TestCase 서브클래스 보유로 # 무회귀 import 확인(모듈 정상 로드 + 테스트 케이스 실재). def _has_testcase(mod) -> bool: return any( isinstance(getattr(mod, n), type) and issubclass(getattr(mod, n), unittest.TestCase) for n in dir(mod) ) assert _has_testcase(m47) assert _has_testcase(m44) # +45 진입점·+47 write-back 을 +48 이 약화하지 않음 (read-only 재사용) from anu_v3.cancel_on_success_live_wiring import ( run_cancel_on_success_live_wiring, ) from anu_v3.callback_event_trigger import write_back_completed assert callable(run_cancel_on_success_live_wiring) assert callable(write_back_completed) # ── §7.9 +32 callback mandatory 무회귀 ──────────────────────────────────── def test_09_plus32_callback_mandatory_no_regression(tmp_path): from dispatch.executor_completion_contract import ( Callback4Tuple, validate_4tuple, ) fx = _fx("task-2553plus48.properly-bound-4tuple") t = fx["properly_bound_4tuple"] ok = Callback4Tuple( task_id=t["task_id"], dispatch_cron_id=t["dispatch_cron_id"], normal_collector_cron_id=t["normal_collector_cron_id"], fallback_callback_cron_id=t["fallback_callback_cron_id"], ) assert validate_4tuple(ok) == [] # normal_collector_cron_id 누락 → +47 write-back CALLBACK_MANDATORY_ # VIOLATION (callback mandatory rule 약화 0) reg = Callback4TupleRegistry( tmp_path / "mand" / "callback_4tuple_index.jsonl" ) bad_tup = ProperlyBound4Tuple( task_id="t-mand", dispatch_id="D", dispatch_cron_id="DC", executor="dev-sim", chat_id="6937032012", normal_collector_cron_id="", fallback_callback_cron_id="FBX", role="fallback", ) bc = build_properly_bound_4tuple(reg, bad_tup) assert bc.writeback_status == "CALLBACK_MANDATORY_VIOLATION" assert bc.writeback_appended is False # properly-bound 정상 경로는 4-tuple mandatory 충족 + role=fallback 보유 reg2 = Callback4TupleRegistry( tmp_path / "ok" / "callback_4tuple_index.jsonl" ) bc2 = build_properly_bound_4tuple(reg2, _tup(fx)) assert bc2.writeback_status == "WRITEBACK_COMPLETED" integ = verify_4tuple_binding_integrity( reg2, task_id=fx["task_id"], live_cron_entries=fx["live_cron_entries"], dispatch_marker=fx["dispatch_fired_marker"], ) assert integ.ok is True assert integ.role_is_fallback is True # ── §7.4b Codex F1 — 모호 multi-fallback → properly-bound 불성립·보존 ───── def test_04b_ambiguous_multi_fallback_not_pass(tmp_path): """task/role 매칭 live fallback 2건 → single bound·verified 깨짐 → properly-bound 불성립, E2E_PASS 차단, sibling 발화로 fire>0 (Codex F1).""" fx = _fx("task-2553plus48.properly-bound-4tuple") bound = dict(fx["live_cron_entries"][0]) # id=PB48FALLBACK (ledger 일치) sibling = dict(bound) sibling["id"] = "PB48FALLBACK-SIBLING" # 같은 task/role, ledger 불일치 res, spy, world = _run_e2e( fx, tmp_path, world_entries=[bound, sibling] ) bi = res.binding_integrity assert bi.single_live_fallback is False assert bi.ok is False assert res.properly_bound is False assert res.verdict != E2E_PASS # sibling fallback 이 armed 로 남아 발화 → fire count > 0 (단일 보장 깨짐) assert res.fallback_fire_proof.fallback_fired_count >= 1 assert res.normal_success_unchanged is True jsonschema.validate(res.e2e_audit, SCHEMA_E2E) assert res.e2e_audit["binding_integrity"]["single_live_fallback"] is False # ── §7.10 raw token/credential exposure 0 ───────────────────────────────── def test_10_no_credential_exposure(tmp_path): fx = _fx("task-2553plus48.properly-bound-4tuple") res, _, _ = _run_e2e(fx, tmp_path) blob = json.dumps(res.e2e_audit, ensure_ascii=False) assert ANU_KEY_SECRET not in blob # 산출 모듈·schema 소스에 raw anu_key 박힘 0 for rel in ( "anu_v3/cancel_on_success_live_e2e.py", "schemas/cancel_on_success_e2e_audit.schema.json", "tests/regression/test_cancel_on_success_live_e2e_2553plus48.py", ): src = (WORKSPACE / rel).read_text(encoding="utf-8") # 본 테스트 파일은 노출검사 상수로만 보유(코드/감사기록 박제 아님). if rel.endswith("test_cancel_on_success_live_e2e_2553plus48.py"): assert src.count(ANU_KEY_SECRET) == 1 else: assert ANU_KEY_SECRET not in src # e2e audit on-disk 산출물에도 raw token 0 disk = Path(res.e2e_audit_path).read_text(encoding="utf-8") assert ANU_KEY_SECRET not in disk if __name__ == "__main__": raise SystemExit(pytest.main([__file__, "-q"]))