"""tests/regression/test_cancel_on_success_live_observation_2553plus28.py task-2553+28 — CALLBACK CANCEL-ON-SUCCESS LIVE OBSERVATION FIXTURE regression. §3.2 6-step assertion (전건 PASS, 100% mock/fixture): ① normal success → seam 1회 ② verifier 5조건 AND PASS → remove called ③ 각 mismatch/SKIP → preserve (remove 0) ④ fallback 발화 0 (spy schedule_history) ⑤ cancel-audit JSON schema 정합 (필수 8필드 + 5각 + normal_success_unchanged) ⑥ exact-once O_EXCL (동시 2호출 → 1 invoke·1 no-op, +25 9-R.3 정합) ⑦ 디커플 (cron-remove 실패 → normal collector success 불변) 부가 단언: • frozen anchor byte-0 (sha 전후 동일, §7) • 격리 강제 self-test: subprocess 차단·실 cron API 차단·live-path 차단 • mock-only / 실 운영 무접촉 evidence • observation-fixture-decision.json ↔ harness 매핑 정합 • live `/home/jay/workspace` git tracked HEAD/branch/ref 전후 assertEqual (§5 / 9-R.5 — task-2553+28.* marker=untracked, 위반 아님) 100% offline. 실 callback cron / cokacdir subprocess / 실 schedule_history / 실 4-tuple / network / git mutation 0. 실 운영 cron 실제 삭제·실 발화 0. """ from __future__ import annotations import builtins import hashlib import json import os 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 tests.fixtures.cancel_on_success_observation_harness import ( # noqa: E402 ANU_CHAT_ID, CONTRACT_PATH, FROZEN_ANCHOR, FROZEN_SHA, LIVE_SCHEDULE_HISTORY, FakeCronLister, SpyRemover, SpyScheduleHistory, attempt_exact_once_claim, install_isolation_guards, materialize_scenario, observe_six_step, simulate_concurrent_double, ) DECISION_JSON = ( WORKSPACE / "memory" / "events" / "task-2553+28.observation-fixture-decision.json" ) # ── git ref invariant (§5 / 9-R.5 — repo root 기준) ────────────────────────── 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 _scenario(tmp_path: Path, **kw): return materialize_scenario(tmp_path / "sandbox", **kw) def _fakes(scenario, *, remover_status="removed", history_fired=None): lister = FakeCronLister( scenario["verifier_entries"], drop_id_after_first=scenario["target_cron_id"], ) remover = SpyRemover(status=remover_status) history = SpyScheduleHistory( scenario["sandbox"], fired_ids=history_fired or [] ) return lister, remover, history # ════════════════════════════════════════════════════════════════════════════ # ① normal success → seam 1회 # ════════════════════════════════════════════════════════════════════════════ def test_step1_normal_success_to_seam_once(tmp_path, monkeypatch): install_isolation_guards(monkeypatch, tmp_path) sc = _scenario(tmp_path) lister, remover, history = _fakes(sc) obs = observe_six_step(sc, lister=lister, remover=remover, history=history) assert obs.step1_normal_collector_success is True assert obs.step2_seam_invoked_once is True assert obs.seam_invoke_count == 1 # ════════════════════════════════════════════════════════════════════════════ # ② verifier 5조건 AND PASS → remove called # ════════════════════════════════════════════════════════════════════════════ def test_step2_verifier_five_and_pass_then_remove(tmp_path, monkeypatch): install_isolation_guards(monkeypatch, tmp_path) sc = _scenario(tmp_path) lister, remover, history = _fakes(sc) obs = observe_six_step(sc, lister=lister, remover=remover, history=history) assert obs.step3_verifier_five_and_pass is True five = obs.cancel_audit["five_condition_results"] assert all( five[k] is True for k in ( "c1_task_id_match", "c2_chat_id_owned", "c3_role_fallback", "c4_marker_id_crosscheck", "c5_pending_not_fired_not_removed", ) ) assert obs.step4_fallback_cron_remove_ok is True assert obs.remove_call_count == 1 assert remover.calls[0]["dry_run"] is False # operational=True # ════════════════════════════════════════════════════════════════════════════ # ③ 각 mismatch/SKIP → preserve (remove 0) # ════════════════════════════════════════════════════════════════════════════ @pytest.mark.parametrize( "mutate,label", [ (lambda e: e.update(task_id="task-OTHER"), "task_id mismatch"), (lambda e: e.update(chat_id=111), "chat_id not owned"), (lambda e: e.update(role="normal"), "role not fallback"), (lambda e: e.update(fired=True), "already fired"), (lambda e: e.update(removed=True), "already removed"), ], ) def test_step3_mismatch_skip_preserves(tmp_path, monkeypatch, mutate, label): install_isolation_guards(monkeypatch, tmp_path) sc = _scenario(tmp_path) mutate(sc["verifier_entries"][0]) lister, remover, history = _fakes(sc) obs = observe_six_step(sc, lister=lister, remover=remover, history=history) # SKIP/mismatch → live verifier 5각 미충족, remove 0, fallback 보존. assert obs.step3_verifier_five_and_pass is False, label assert obs.remove_call_count == 0, f"{label}: 실 remove 발생 (preserve 위반)" # 디커플 — normal collector success 는 mismatch 와 무관하게 불변. assert obs.step1_normal_collector_success is True, label assert obs.cancel_audit["normal_success_unchanged"] is True, label # ════════════════════════════════════════════════════════════════════════════ # ④ fallback 발화 0 (spy schedule_history) # ════════════════════════════════════════════════════════════════════════════ def test_step4_no_fallback_firing(tmp_path, monkeypatch): install_isolation_guards(monkeypatch, tmp_path) sc = _scenario(tmp_path) lister, remover, history = _fakes(sc) obs = observe_six_step(sc, lister=lister, remover=remover, history=history) assert obs.step5_no_fallback_firing is True assert history.no_firing(sc["target_cron_id"]) is True # spy 는 격리 sandbox 만 — 실 schedule_history 미접촉. assert history.reads, "schedule_history spy 미조회 (passive 관측 누락)" # ════════════════════════════════════════════════════════════════════════════ # ⑤ cancel-audit JSON schema 정합 # ════════════════════════════════════════════════════════════════════════════ def test_step5_cancel_audit_schema(tmp_path, monkeypatch): install_isolation_guards(monkeypatch, tmp_path) sc = _scenario(tmp_path) lister, remover, history = _fakes(sc) obs = observe_six_step(sc, lister=lister, remover=remover, history=history) a = obs.cancel_audit for k in ( "schema", "event_id", "five_condition_results", "remove_attempted", "remove_result", "skip_reason", "already_removed_or_missing", "normal_success_unchanged", ): assert k in a, f"cancel-audit 필수 필드 누락: {k}" assert set(a["five_condition_results"]) >= { "c1_task_id_match", "c2_chat_id_owned", "c3_role_fallback", "c4_marker_id_crosscheck", "c5_pending_not_fired_not_removed", } assert a["normal_success_unchanged"] is True assert a["remove_attempted"] is True assert a["remove_result"] == "CANCELLED" assert obs.step6_cancel_audit_schema_ok is True # ════════════════════════════════════════════════════════════════════════════ # ⑥ exact-once O_EXCL (동시 2호출 → 1 invoke·1 no-op, +25 9-R.3 정합) # ════════════════════════════════════════════════════════════════════════════ def test_step6_exact_once_oexcl(tmp_path, monkeypatch): install_isolation_guards(monkeypatch, tmp_path) claim_dir = tmp_path / "sandbox" / "claims" eid = hashlib.sha256(b"task-2553+28-event").hexdigest() res = simulate_concurrent_double(claim_dir, eid) assert res["first_claimed"] is True assert res["second_claimed"] is False assert res["seam_invoke_count"] == 1 assert res["second_is_noop"] is True assert res["exact_once_ok"] is True # 명시 claim artifact (O_EXCL) 가 실제 FS 에 1개만 — in-memory mutex 아님. claims = list(claim_dir.glob("*.seam-claim.*.json")) assert len(claims) == 1 payload = json.loads(claims[0].read_text(encoding="utf-8")) assert payload["atomic_create_method"] == "O_CREAT|O_EXCL" assert payload["event_id"] == eid # 동일 key 재시도 → 즉시 no-op (idempotent). assert attempt_exact_once_claim(claim_dir, eid) is False # ════════════════════════════════════════════════════════════════════════════ # ⑦ 디커플 — cron-remove 실패 → normal collector success 불변 # ════════════════════════════════════════════════════════════════════════════ def test_step7_decouple_remove_failure(tmp_path, monkeypatch): install_isolation_guards(monkeypatch, tmp_path) sc = _scenario(tmp_path) lister, remover, history = _fakes(sc, remover_status="failed") obs = observe_six_step(sc, lister=lister, remover=remover, history=history) # remove 실패해도 normal collector success 불변. assert obs.step1_normal_collector_success is True assert obs.cancel_audit["normal_success_unchanged"] is True # remove 시도는 했으나 cancel 실패 → step4 미충족, 그러나 디커플 유지. assert obs.step4_fallback_cron_remove_ok is False assert obs.cancel_audit["remove_result"] != "CANCELLED" # ════════════════════════════════════════════════════════════════════════════ # 부가: frozen byte-0 (§7) # ════════════════════════════════════════════════════════════════════════════ def test_frozen_anchor_byte0(): sha = hashlib.sha256(FROZEN_ANCHOR.read_bytes()).hexdigest() assert sha == FROZEN_SHA, "frozen anchor byte-0 변경 (§7 위반)" # ════════════════════════════════════════════════════════════════════════════ # 부가: 9-R.3 격리 강제 self-test (subprocess / 실 cron API / live-path) # ════════════════════════════════════════════════════════════════════════════ def test_isolation_blocks_subprocess(tmp_path, monkeypatch): install_isolation_guards(monkeypatch, tmp_path) with pytest.raises(AssertionError): subprocess.run(["echo", "x"]) with pytest.raises(AssertionError): os.system("echo x") def test_isolation_blocks_real_cron_api(tmp_path, monkeypatch): install_isolation_guards(monkeypatch, tmp_path) import utils.completion_callback_fallback_cancel as fc import utils.live_cron_state_verifier as lv with pytest.raises(AssertionError): lv.RealCokacdirCronLister()() with pytest.raises(AssertionError): fc.RealCokacdirCronRemover()("FB1", dry_run=False) def test_isolation_blocks_live_paths(tmp_path, monkeypatch): install_isolation_guards(monkeypatch, tmp_path) # 실 schedule_history write/read 차단. with pytest.raises(AssertionError): builtins.open(str(LIVE_SCHEDULE_HISTORY / "FBX.log"), "w") with pytest.raises(AssertionError): os.open(str(LIVE_SCHEDULE_HISTORY / "FBX.log"), os.O_RDONLY) # sandbox 내부는 허용. p = tmp_path / "ok.txt" with builtins.open(str(p), "w") as f: f.write("ok") assert p.read_text() == "ok" # contract 는 read-only allowlist (참조 허용, write 금지). with builtins.open(str(CONTRACT_PATH), "r") as f: assert f.read(1) with pytest.raises(AssertionError): builtins.open(str(CONTRACT_PATH), "w") # ════════════════════════════════════════════════════════════════════════════ # 부가: mock-only / 실 운영 무접촉 evidence # ════════════════════════════════════════════════════════════════════════════ def test_mock_only_full_six_step(tmp_path, monkeypatch): install_isolation_guards(monkeypatch, tmp_path) sc = _scenario(tmp_path) lister, remover, history = _fakes(sc) obs = observe_six_step(sc, lister=lister, remover=remover, history=history) assert obs.all_pass is True, obs.step_map() assert obs.mock_only is True # 실 발화·실 제거 0 — 전부 주입 Fake/Spy. assert all(c["dry_run"] is False for c in remover.calls) assert lister.calls >= 1 and not lister.id_present(sc["target_cron_id"]) # ════════════════════════════════════════════════════════════════════════════ # 부가: observation-fixture-decision.json ↔ harness 매핑 정합 # ════════════════════════════════════════════════════════════════════════════ def test_decision_json_consistency(tmp_path, monkeypatch): assert DECISION_JSON.exists(), "observation-fixture-decision.json 부재" dec = json.loads(DECISION_JSON.read_text(encoding="utf-8")) assert dec["mock_only"] is True assert dec["real_ops_zero_touch"] is True assert dec["future_real_task_applicable"] is True assert dec["anu_chat_id"] == ANU_CHAT_ID mapping = dec["contract_six_step_to_harness"] assert set(mapping) == {f"step{i}" for i in range(1, 7)} install_isolation_guards(monkeypatch, tmp_path) sc = _scenario(tmp_path) lister, remover, history = _fakes(sc) obs = observe_six_step(sc, lister=lister, remover=remover, history=history) # decision 이 선언한 6-step 가 실제 harness 관측과 정합. for step, ok in obs.step_map().items(): assert ok is True, f"{step} 관측 실패 — decision 정합 위반" assert mapping[step]["harness_component"] assert mapping[step]["mock_artifact"]