"""tests/regression/test_cancel_on_success_live_observation_2553plus41.py

task-2553+41 — TRACK D: +37 CANCEL-ON-SUCCESS LIVE OBSERVATION FIXTURE
regression (read-only observation only).

§3.2 6-step assertion (전건 PASS, 100% mock/fixture, +37 entrypoint 경유):
  ① normal success                      (collector PASS + durable-success)
  ② wired entrypoint                    (binding-valid, +37 표준 entrypoint)
  ③ operational_collector_wiring        (PRIMARY 경유 — +25 wiring 결과 보유)
  ④ run_operational_cancel_seam         (operational=True, seam 1회 + claim)
  ⑤ live verifier 5조건 AND PASS        (c1~c5 전부 True)
  ⑥ bound fallback cron-remove + audit  (CANCELLED + schema 정합)

부가 단언:
  • binding-invalid (missing / marker mismatch) → seam 미진입·cron-remove
    0·fallback 보존·BINDING_* cancel-audit·디커플 (+37 §2.5, read-only)
  • verifier mismatch/SKIP → remove 0·preserve (디커플 불변)
  • fallback 발화 0 (spy schedule_history, 격리)
  • exact-once O_EXCL (동시 2호출 → 1 invoke·1 no-op)
  • 디커플 (cron-remove 실패 → normal collector success 불변)
  • read-only consume 증거: +37 entrypoint / +25 wiring / frozen anchor
    byte-0 (sha 전후 동일, §5 / §7)
  • 격리 강제 self-test: subprocess 차단·실 cron API 차단·live-path 차단
  • observation-decision.json ↔ harness 매핑 정합
  • live /home/jay/workspace git tracked HEAD/branch/ref 전후 assertEqual
    (§5 — task-2553+41.* / 신규 fixture/tests = 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_live_observation_harness_2553plus41 import (  # noqa: E402,E501
    ANU_CHAT_ID,
    FROZEN_ANCHOR,
    FROZEN_SHA,
    LIVE_SCHEDULE_HISTORY,
    PLUS25_WIRING_SRC,
    WIRED_ENTRYPOINT_SRC,
    FakeCronLister,
    SpyRemover,
    SpyScheduleHistory,
    attempt_exact_once_claim,
    binding_invalid_reasons,
    install_isolation_guards,
    make_binding,
    materialize_scenario,
    observe_binding_invalid,
    observe_six_step,
    simulate_concurrent_double,
)
from utils.anu_delegation_completion_callback import Classification  # noqa: E402

DECISION_JSON = (
    WORKSPACE
    / "memory"
    / "events"
    / "task-2553+41.observation-decision.json"
)
#: +37 wired entrypoint / +25 wiring read-only consume 무수정 검증용 sha.
WIRED_ENTRYPOINT_SHA = hashlib.sha256(
    WIRED_ENTRYPOINT_SRC.read_bytes()
).hexdigest()
PLUS25_WIRING_SHA = hashlib.sha256(PLUS25_WIRING_SRC.read_bytes()).hexdigest()
ANU_KEY_SECRET = "c119085addb0f8b7"  # 노출 검사용 — 산출물에 박히면 FAIL


# ── git ref invariant (§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


def _valid_binding(sc):
    return make_binding(sc["task_id"], sc["target_cron_id"])


# ════════════════════════════════════════════════════════════════════════════
# ① normal success → ② wired entrypoint → ③ operational_collector_wiring
# ════════════════════════════════════════════════════════════════════════════
def test_step123_normal_success_wired_via_plus25(tmp_path, monkeypatch):
    install_isolation_guards(monkeypatch, tmp_path)
    sc = _scenario(tmp_path)
    lister, remover, history = _fakes(sc)
    obs = observe_six_step(
        sc,
        binding=_valid_binding(sc),
        lister=lister,
        remover=remover,
        history=history,
    )
    assert obs.step1_normal_success is True
    assert obs.step2_wired_entrypoint is True
    assert obs.binding_valid is True
    assert obs.step3_operational_collector_wiring is True


# ════════════════════════════════════════════════════════════════════════════
# ④ run_operational_cancel_seam(operational=True) — seam 1회 + claim
# ════════════════════════════════════════════════════════════════════════════
def test_step4_operational_cancel_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,
        binding=_valid_binding(sc),
        lister=lister,
        remover=remover,
        history=history,
    )
    assert obs.step4_run_operational_cancel_seam is True
    assert obs.seam_invoke_count == 1


# ════════════════════════════════════════════════════════════════════════════
# ⑤ live verifier 5조건 AND PASS
# ════════════════════════════════════════════════════════════════════════════
def test_step5_live_verifier_five_and_pass(tmp_path, monkeypatch):
    install_isolation_guards(monkeypatch, tmp_path)
    sc = _scenario(tmp_path)
    lister, remover, history = _fakes(sc)
    obs = observe_six_step(
        sc,
        binding=_valid_binding(sc),
        lister=lister,
        remover=remover,
        history=history,
    )
    assert obs.step5_live_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",
        )
    )


# ════════════════════════════════════════════════════════════════════════════
# ⑥ bound fallback cron-remove + cancel-audit
# ════════════════════════════════════════════════════════════════════════════
def test_step6_bound_fallback_remove_and_audit(tmp_path, monkeypatch):
    install_isolation_guards(monkeypatch, tmp_path)
    sc = _scenario(tmp_path)
    lister, remover, history = _fakes(sc)
    obs = observe_six_step(
        sc,
        binding=_valid_binding(sc),
        lister=lister,
        remover=remover,
        history=history,
    )
    assert obs.step6_bound_fallback_remove_and_audit is True
    assert obs.remove_call_count == 1
    assert remover.calls[0]["cron_id"] == sc["target_cron_id"]
    assert remover.calls[0]["dry_run"] is False  # operational=True
    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 a["remove_attempted"] is True
    assert a["remove_result"] == "CANCELLED"
    assert a["normal_success_unchanged"] is True


# ════════════════════════════════════════════════════════════════════════════
# full 6-step all-pass + mock-only / 실 운영 무접촉 evidence
# ════════════════════════════════════════════════════════════════════════════
def test_full_six_step_all_pass_mock_only(tmp_path, monkeypatch):
    install_isolation_guards(monkeypatch, tmp_path)
    sc = _scenario(tmp_path)
    lister, remover, history = _fakes(sc)
    obs = observe_six_step(
        sc,
        binding=_valid_binding(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"])
    assert history.reads, "schedule_history spy 미조회 (passive 관측 누락)"


# ════════════════════════════════════════════════════════════════════════════
# binding-invalid (missing) → seam 미진입·cron-remove 0·fallback 보존 (+37 §2.5)
# ════════════════════════════════════════════════════════════════════════════
def test_binding_missing_no_seam_no_remove_preserved(tmp_path, monkeypatch):
    install_isolation_guards(monkeypatch, tmp_path)
    sc = _scenario(tmp_path)
    lister, remover, _ = _fakes(sc)
    binding = make_binding(
        sc["task_id"], sc["target_cron_id"], fallback_cron_id=None
    )
    res = observe_binding_invalid(
        sc, binding=binding, lister=lister, remover=remover
    )
    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 remover.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


# ════════════════════════════════════════════════════════════════════════════
# binding-invalid (marker mismatch) → BINDING_MISMATCH, cron-remove 0
# ════════════════════════════════════════════════════════════════════════════
def test_binding_marker_mismatch_no_remove(tmp_path, monkeypatch):
    install_isolation_guards(monkeypatch, tmp_path)
    sc = _scenario(tmp_path)
    lister, remover, _ = _fakes(sc)
    # marker.fallback_callback_cron_id = target, binding 은 다른 id → mismatch
    binding = make_binding(
        sc["task_id"], sc["target_cron_id"], fallback_cron_id="FB-OTHER-41"
    )
    pre = binding_invalid_reasons(
        binding, dispatch_fired_marker_path=sc["dispatch_fired_marker_path"]
    )
    assert any(r.startswith("fallback_cron_id_marker_mismatch") for r in pre)
    res = observe_binding_invalid(
        sc, binding=binding, lister=lister, remover=remover
    )
    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 remover.calls == []
    assert res.cancel_audit["lookup_status"] == "BINDING_MISMATCH"
    assert res.cancel_audit["normal_success_unchanged"] is True


# ════════════════════════════════════════════════════════════════════════════
# verifier mismatch/SKIP (binding ok) → remove 0·preserve·디커플 불변
# ════════════════════════════════════════════════════════════════════════════
@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_verifier_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,
        binding=_valid_binding(sc),
        lister=lister,
        remover=remover,
        history=history,
    )
    # binding-valid → seam 진입하되 live verifier 5각 미충족 → remove 0.
    assert obs.binding_valid is True, label
    assert obs.step5_live_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_success is True, label
    assert obs.cancel_audit["normal_success_unchanged"] is True, label


# ════════════════════════════════════════════════════════════════════════════
# 디커플 — cron-remove 실패 → normal collector success 불변
# ════════════════════════════════════════════════════════════════════════════
def test_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,
        binding=_valid_binding(sc),
        lister=lister,
        remover=remover,
        history=history,
    )
    # remove 실패해도 normal collector success 불변 (디커플 절대불변).
    assert obs.step1_normal_success is True
    assert obs.cancel_audit["normal_success_unchanged"] is True
    # remove 시도는 했으나 cancel 실패 → step6 미충족.
    assert obs.step6_bound_fallback_remove_and_audit is False
    assert obs.cancel_audit["remove_result"] != "CANCELLED"


# ════════════════════════════════════════════════════════════════════════════
# exact-once O_EXCL (동시 2호출 → 1 invoke·1 no-op)
# ════════════════════════════════════════════════════════════════════════════
def test_exact_once_oexcl(tmp_path, monkeypatch):
    install_isolation_guards(monkeypatch, tmp_path)
    claim_dir = tmp_path / "sandbox" / "claims"
    eid = hashlib.sha256(b"task-2553+41-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
    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
    assert attempt_exact_once_claim(claim_dir, eid) is False


# ════════════════════════════════════════════════════════════════════════════
# read-only consume 증거 — +37 entrypoint / +25 wiring / frozen byte-0 (§5/§7)
# ════════════════════════════════════════════════════════════════════════════
def test_readonly_consume_byte0_invariant():
    assert (
        hashlib.sha256(FROZEN_ANCHOR.read_bytes()).hexdigest() == FROZEN_SHA
    ), "frozen anchor byte-0 변경 (§5/§7 위반)"
    assert (
        hashlib.sha256(WIRED_ENTRYPOINT_SRC.read_bytes()).hexdigest()
        == WIRED_ENTRYPOINT_SHA
    ), "+37 wired entrypoint 변경 감지 (callback/collector 경로 수정 — §5 위반)"
    assert (
        hashlib.sha256(PLUS25_WIRING_SRC.read_bytes()).hexdigest()
        == PLUS25_WIRING_SHA
    ), "+25 operational_collector_wiring 변경 감지 (§5 위반)"


# ════════════════════════════════════════════════════════════════════════════
# 격리 강제 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)
    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"
    # +37 entrypoint 소스는 read-only allowlist (참조 허용, write 금지).
    with builtins.open(str(WIRED_ENTRYPOINT_SRC), "r") as f:
        assert f.read(1)
    with pytest.raises(AssertionError):
        builtins.open(str(WIRED_ENTRYPOINT_SRC), "w")


# ════════════════════════════════════════════════════════════════════════════
# observation-decision.json ↔ harness 매핑 정합
# ════════════════════════════════════════════════════════════════════════════
def test_decision_json_consistency(tmp_path, monkeypatch):
    assert DECISION_JSON.exists(), "observation-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["callback_collector_path_unmodified"] 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,
        binding=_valid_binding(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"]
    # 실 키 미박제 (placeholder ref 만)
    assert ANU_KEY_SECRET not in json.dumps(dec, ensure_ascii=False)


if __name__ == "__main__":
    raise SystemExit(pytest.main([__file__, "-q"]))