"""tests/regression/test_callback_cancel_operational_integration_2553plus23.py

task-2553+23 — CALLBACK_CANCEL_ON_SUCCESS_OPERATIONAL_INTEGRATION regression.

§7 필수 12 (회장 verbatim 10 + 9-R.2/9-R.3 추가 2) — 전건 PASS.

100% offline. 실 callback cron / cokacdir subprocess / network / git mutation 0.
모든 cron-list 는 **주입 fake lister**, 모든 cron-remove 는 **주입 fake
remover**. 실 subprocess 호출 시 테스트 즉시 FAIL(차단 spy).

frozen anchor utils/anu_delegation_completion_callback.py 는 본 테스트·seam·
verifier 어디에서도 import/호출하지 않는다 — AST 정적 증명(§3, §6).

live `/home/jay/workspace` git tracked HEAD/branch/ref 는 테스트 전후
assertEqual (§6) — 본 task 산출물은 git-untracked batch-internal.
"""
from __future__ import annotations

import ast
import json
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 utils.completion_callback_fallback_cancel import RemoverResult  # noqa: E402
from utils.completion_callback_operational_cancel_seam import (  # noqa: E402
    run_operational_cancel_seam,
)
from utils.live_cron_state_verifier import (  # noqa: E402
    LiveVerifyClassification,
    verify_live_cron_state,
)

FIXDIR = WORKSPACE / "memory" / "fixtures"


def _fx(name: str) -> dict:
    return json.loads((FIXDIR / f"{name}.json").read_text(encoding="utf-8"))


def _materialize(fx: dict, tmp: Path) -> dict:
    paths: dict = {}
    dfm = tmp / f"{fx['task_id']}.dispatch-fired.json"
    dfm.write_text(json.dumps(fx["dispatch_fired_marker"]), encoding="utf-8")
    paths["dispatch_fired_marker_path"] = dfm

    rj = tmp / f"{fx['task_id']}.result.json"
    rj.write_text(json.dumps(fx["result_json"]), encoding="utf-8")
    paths["result_json_path"] = rj

    rep = tmp / f"{fx['task_id']}.report.md"
    rep.write_text(fx.get("report_text", ""), encoding="utf-8")
    paths["report_path"] = rep

    crm = tmp / f"{fx['task_id']}.collector-result.json"
    if fx.get("collector_result_marker_present", True) and "collector_result_marker" in fx:
        crm.write_text(json.dumps(fx["collector_result_marker"]), encoding="utf-8")
    paths["collector_result_marker_path"] = crm

    paths["fallback_cancelled_marker_path"] = tmp / f"{fx['task_id']}.fallback-cancelled.json"
    paths["cancel_lock_path"] = tmp / f"{fx['task_id']}.cancel.lock"
    paths["audit_path"] = tmp / f"{fx['task_id']}.cancel-audit.json"
    return paths


class FakeCronLister:
    """주입 fake — fixture live_cron_entries 를 정규화 형태로 반환. 실 0."""

    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 — 호출 인자/횟수 기록. 실 subprocess 0."""

    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):
    """어떤 테스트에서도 실 subprocess (--cron-list/--cron-remove) 호출 0 강제."""

    def _boom(*a, **k):  # noqa: ANN001, ANN002, ANN003
        raise AssertionError(
            "실 subprocess 호출 금지 (§6 9-R.4) — 주입 fake lister/remover 만 허용"
        )

    monkeypatch.setattr(subprocess, "run", _boom)


def _git_ref():
    """live workspace git tracked HEAD/branch/ref (read-only — mutation 0).

    subprocess 가 차단되어 있으므로 git CLI 대신 .git 파일을 직접 읽는다.
    """
    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():
    """§6 — 테스트 전후 git tracked HEAD/branch/ref assertEqual."""
    before = _git_ref()
    yield
    assert _git_ref() == before, "git HEAD/branch/ref 변경 감지 (§6 위반)"


def _seam(fx: dict, tmp: Path, *, remover, lister):
    paths = _materialize(fx, tmp)
    return run_operational_cancel_seam(
        task_id=fx["task_id"],
        target_cron_id=fx["target_cron_id"],
        callback_contract=fx.get("callback_contract"),
        cron_lister=lister,
        remover=remover,
        operational=False,  # 본 task: 실 cron 비접촉 (dry_run 경로)
        **paths,
    )


# ── §7 필수 12 ──────────────────────────────────────────────────────────────


def test_01_verified_remove_called(tmp_path):
    """1. normal success + live ownership VERIFIED → cron-remove called."""
    fx = _fx("task-2553+23.live-verified")
    spy = SpyRemover("removed")
    lister = FakeCronLister(fx["live_cron_entries"])
    out = _seam(fx, tmp_path, remover=spy, lister=lister)
    assert out.seam_classification == "PLUS9A_CANCELLED"
    assert out.cron_remove_invoked is True
    assert out.fallback_cancelled is True
    assert out.remove_allowed_by_live_verifier is True
    assert spy.calls == [{"cron_id": "FB23-0001", "dry_run": True}]
    assert out.normal_success_preserved is True
    assert lister.calls == 1


def test_02_live_missing_skip_success_preserved(tmp_path):
    """2. normal success + live cron missing → cancel skipped, success 유지."""
    fx = _fx("task-2553+23.live-missing")
    spy = SpyRemover("removed")
    out = _seam(fx, tmp_path, remover=spy, lister=FakeCronLister(fx["live_cron_entries"]))
    assert out.seam_classification == "SKIP_LIVE_SKIP_ALREADY_REMOVED"
    assert out.cron_remove_invoked is False
    assert spy.calls == []
    assert out.normal_success_preserved is True


def test_03_task_id_mismatch_skip(tmp_path):
    """3. normal success + task_id mismatch → cancel skipped."""
    fx = _fx("task-2553+23.task-id-mismatch")
    spy = SpyRemover("removed")
    out = _seam(fx, tmp_path, remover=spy, lister=FakeCronLister(fx["live_cron_entries"]))
    assert out.seam_classification == "SKIP_LIVE_SKIP_MISMATCH"
    assert out.live_verification["checks"]["c1_task_id_match"] is False
    assert spy.calls == []
    assert out.normal_success_preserved is True


def test_04_chat_id_mismatch_skip(tmp_path):
    """4. normal success + chat_id mismatch → cancel skipped."""
    fx = _fx("task-2553+23.chat-id-mismatch")
    spy = SpyRemover("removed")
    out = _seam(fx, tmp_path, remover=spy, lister=FakeCronLister(fx["live_cron_entries"]))
    assert out.seam_classification == "SKIP_LIVE_SKIP_MISMATCH"
    assert out.live_verification["checks"]["c2_chat_id_owned"] is False
    assert spy.calls == []


def test_05_role_not_fallback_skip(tmp_path):
    """5. normal success + role not fallback → cancel skipped."""
    fx = _fx("task-2553+23.role-not-fallback")
    spy = SpyRemover("removed")
    out = _seam(fx, tmp_path, remover=spy, lister=FakeCronLister(fx["live_cron_entries"]))
    assert out.seam_classification == "SKIP_LIVE_SKIP_MISMATCH"
    assert out.live_verification["checks"]["c3_role_fallback"] is False
    assert spy.calls == []


def test_06_normal_failed_skip_fallback_preserved(tmp_path):
    """6. normal failed/HOLD/partial → cancel skipped, fallback 보존."""
    fx = _fx("task-2553+23.normal-failed")
    spy = SpyRemover("removed")
    lister = FakeCronLister(fx["live_cron_entries"])
    out = _seam(fx, tmp_path, remover=spy, lister=lister)
    assert out.seam_classification == "SKIP_DURABLE_NOT_SATISFIED"
    assert out.cron_remove_invoked is False
    assert spy.calls == []
    assert lister.calls == 0  # durable gate 차단 → live 조회조차 안 함
    assert out.normal_success_preserved is True


def test_07_already_fired_idempotent_noop(tmp_path):
    """7. fallback already fired → idempotent no-op, no failure."""
    fx = _fx("task-2553+23.already-fired")
    spy = SpyRemover("removed")
    out = _seam(fx, tmp_path, remover=spy, lister=FakeCronLister(fx["live_cron_entries"]))
    assert out.seam_classification == "SKIP_LIVE_SKIP_ALREADY_FIRED"
    assert out.cron_remove_invoked is False
    assert spy.calls == []
    assert out.normal_success_preserved is True


def test_08_duplicate_callback_path_no_regression(tmp_path):
    """8. +9a/+16 DUPLICATE_CALLBACK_IGNORED 경로 무회귀.

    기존 dedup regression 모듈을 재실행하여 본 +23 seam/verifier 추가가 기존
    duplicate-callback 경로를 회귀시키지 않음을 입증한다.
    """
    import importlib

    sys.path.insert(0, str(WORKSPACE / "tests" / "regression"))
    dup = importlib.import_module(
        "test_completion_callback_dup_ignored_realworld_2553plus1"
    )
    fxd = dup._fixture()
    ackp = tmp_path / "dup.callback-ack.json"
    first = dup.run_completion_callback_collector(
        dup._normal_input(fxd), ackp, post_result_review_fn=dup._mock_review_advisory
    )
    second = dup.run_completion_callback_collector(
        dup._fallback_input(fxd), ackp, post_result_review_fn=dup._mock_review_advisory
    )
    assert first.classification == dup.Classification.PASS
    assert second.classification == dup.Classification.DUPLICATE_CALLBACK_IGNORED
    assert second.closeout_candidate is False


def test_09_real_remove_only_when_verifier_pass(tmp_path):
    """9. 실 remove 호출은 verifier PASS 일 때만 가능.

    verifier SKIP 계열 전수 → remover 절대 미호출. VERIFIED 1건만 호출.
    """
    skip_fixtures = [
        "task-2553+23.live-missing",
        "task-2553+23.task-id-mismatch",
        "task-2553+23.chat-id-mismatch",
        "task-2553+23.role-not-fallback",
        "task-2553+23.already-fired",
        "task-2553+23.marker-id-mismatch",
        "task-2553+23.already-removed",
    ]
    for name in skip_fixtures:
        fx = _fx(name)
        spy = SpyRemover("removed")
        out = _seam(
            fx, tmp_path, remover=spy, lister=FakeCronLister(fx["live_cron_entries"])
        )
        assert out.remove_allowed_by_live_verifier is False, name
        assert spy.calls == [], f"{name}: verifier SKIP 인데 remover 호출됨"
    # VERIFIED 만 호출
    fx = _fx("task-2553+23.live-verified")
    spy = SpyRemover("removed")
    out = _seam(fx, tmp_path, remover=spy, lister=FakeCronLister(fx["live_cron_entries"]))
    assert out.remove_allowed_by_live_verifier is True
    assert len(spy.calls) == 1


def test_10_wrong_cron_id_never_removed(tmp_path):
    """10. wrong cron id 제거 시도 0.

    live 목록에 target 과 다른 id 만 존재 → 어떤 경우에도 그 id 로 remove 0.
    """
    fx = _fx("task-2553+23.live-verified")
    fx["live_cron_entries"] = [
        {"id": "WRONG-OTHER-ID", "task_id": fx["task_id"], "chat_id": 6937032012,
         "role": "fallback", "fired": False, "removed": False}
    ]
    spy = SpyRemover("removed")
    out = _seam(fx, tmp_path, remover=spy, lister=FakeCronLister(fx["live_cron_entries"]))
    assert out.cron_remove_invoked is False
    assert spy.calls == []  # WRONG-OTHER-ID 로도, target 으로도 remove 0
    assert out.seam_classification == "SKIP_LIVE_SKIP_ALREADY_REMOVED"


def test_11_marker_fallback_cron_id_mismatch_skip(tmp_path):
    """11. dispatch-fired marker fallback_cron_id 불일치 → cancel skipped, 실 remove 0 (9-R.2)."""
    fx = _fx("task-2553+23.marker-id-mismatch")
    spy = SpyRemover("removed")
    out = _seam(fx, tmp_path, remover=spy, lister=FakeCronLister(fx["live_cron_entries"]))
    assert out.seam_classification == "SKIP_LIVE_SKIP_MISMATCH"
    assert out.live_verification["checks"]["c4_marker_id_crosscheck"] is False
    assert out.cron_remove_invoked is False
    assert spy.calls == []


def test_12_already_removed_idempotent_noop(tmp_path):
    """12. fallback already removed → idempotent no-op, 이중 remove 0, normal success 불변 (9-R.3)."""
    fx = _fx("task-2553+23.already-removed")
    spy = SpyRemover("removed")
    out = _seam(fx, tmp_path, remover=spy, lister=FakeCronLister(fx["live_cron_entries"]))
    assert out.seam_classification == "SKIP_LIVE_SKIP_ALREADY_REMOVED"
    assert out.cron_remove_invoked is False
    assert spy.calls == []  # 이중 remove 0
    assert out.normal_success_preserved is True
    # case 7 already-fired 와 분기 분리 확인
    assert out.live_verification["classification"] == "SKIP_ALREADY_REMOVED"


# ── 보조 — 분리/디커플/안전 정적 증명 ───────────────────────────────────────


def test_aux_frozen_anchor_not_imported():
    """seam/verifier 가 frozen orchestrator 를 import/호출하지 않음 — AST 증명."""
    for mod in (
        "completion_callback_operational_cancel_seam",
        "live_cron_state_verifier",
    ):
        src = (WORKSPACE / "utils" / f"{mod}.py").read_text(encoding="utf-8")
        tree = ast.parse(src)
        imported: list = []
        for node in ast.walk(tree):
            if isinstance(node, ast.Import):
                imported += [a.name for a in node.names]
            elif isinstance(node, ast.ImportFrom):
                imported.append(node.module or "")
        assert not any(
            "anu_delegation_completion_callback" in (m or "") for m in imported
        ), f"{mod}: frozen orchestrator import 결합 발견 {imported}"


def test_aux_query_failed_skips_preserve(tmp_path):
    """live 조회 실패 → SKIP_QUERY_FAILED, remove_allowed=False (fail-safe)."""
    fx = _fx("task-2553+23.live-verified")
    paths = _materialize(fx, tmp_path)
    lv = verify_live_cron_state(
        task_id=fx["task_id"],
        target_cron_id=fx["target_cron_id"],
        dispatch_fired_marker_path=paths["dispatch_fired_marker_path"],
        cron_lister=FakeCronLister([], status="error"),
    )
    assert lv.classification == LiveVerifyClassification.SKIP_QUERY_FAILED
    assert lv.remove_allowed is False


def test_aux_decouple_audit_written(tmp_path):
    """디커플 + audit: remove skip 이어도 normal success 유지 + audit JSON 기록."""
    fx = _fx("task-2553+23.task-id-mismatch")
    spy = SpyRemover("removed")
    paths = _materialize(fx, tmp_path)
    out = run_operational_cancel_seam(
        task_id=fx["task_id"],
        target_cron_id=fx["target_cron_id"],
        cron_lister=FakeCronLister(fx["live_cron_entries"]),
        remover=spy,
        operational=False,
        **paths,
    )
    assert out.normal_success_preserved is True
    assert out.audit_path is not None
    rec = json.loads(Path(out.audit_path).read_text(encoding="utf-8"))
    assert rec["schema"] == "task-2553+23.cancel-audit_v1"
    assert rec["task_id"] == fx["task_id"]
    assert rec["decision"]["cron_remove_invoked"] is False


def test_aux_raising_now_fn_never_propagates(tmp_path):
    """디커플 절대불변: 주입 now_fn 이 예외를 던져도 collector 비전파,
    normal_success_preserved=True, audit 기록 (independent-review MED fix)."""
    fx = _fx("task-2553+23.live-verified")
    spy = SpyRemover("removed")
    paths = _materialize(fx, tmp_path)

    def _boom_now() -> str:
        raise RuntimeError("clock failure injected")

    out = run_operational_cancel_seam(
        task_id=fx["task_id"],
        target_cron_id=fx["target_cron_id"],
        callback_contract=fx.get("callback_contract"),
        cron_lister=FakeCronLister(fx["live_cron_entries"]),
        remover=spy,
        operational=False,
        now_fn=_boom_now,
        **paths,
    )
    assert out.normal_success_preserved is True
    assert out.audit_path is not None
    rec = json.loads(Path(out.audit_path).read_text(encoding="utf-8"))
    assert rec["schema"] == "task-2553+23.cancel-audit_v1"


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