"""End-to-end verification of callback lifecycle wiring (task-2631, L1~L4 통합).

성격: **adversarial e2e 검증**. L4 wiring(dispatch.executor_completion_contract)을
빌딩블록 단위가 아니라 **실제 closeout 경로 전체**로 통합 실행하면서, 머지된 L4
wiring 의 실제 결함(fields 10~14 미emit · artifact 미생성 · 분류 불일치 · atomic
write 미보장 · 9-field 훼손 등)을 적극 탐색한다.

실제 closeout 경로 (스펙 task-2631 §구현 가이드):
  9-field callback contract closeout dict (단일소스 `_contract_fields`)
    → `append_lifecycle_fields(...)` → 15 keys(9 + 6 lifecycle field) result dict 검증
      (회장 표기 "fields 10~14"는 (14)=classified_by·applied_count 2키를 묶은 것 → 물리적 6키/총 15)
    → `write_callback_lifecycle_artifact(task_id, evidence, events_dir=<tmp>)`
    → 생성된 `<task>.callback_lifecycle.json` 검증(atomic·idempotent·내용)
    → `callback_stage_separation` 3단계 분리 검증.

필수 검증 10 (회장 verbatim · task-2631):
  1. 정상 closeout result.json 에 fields 10~14 append 확인
  2. callback_lifecycle.json 생성 확인
  3. artifact writer atomic write 확인
  4. callback gate PASS / notification sent / collector received 분리 기록 확인
  5. normal callback received 케이스 분류 확인
  6. fallback collector applied 케이스 분류 확인
  7. ENVELOPE_PREPARED_NOT_FIRED 케이스 분류 확인
  8. SELF_KEY_FIRED_NON_AUTHORITATIVE 케이스 분류 확인
  9. UNKNOWN / INSUFFICIENT_EVIDENCE 케이스 분류 확인
  10. 기존 callback contract 9 fields 유지 확인

필수 fixtures 5: task-2625 / task-2628 / task-2628_plus_1 / unknown_insufficient
  + normal_anu_owned(신규).

frozen anchor (D-SPEC-EXACTNESS):
  A1: 실제 closeout 경로에서 result.json fields 10~14 emit + artifact 생성
  A2: normal/fallback/unknown/self-key 사례가 classifier 와 일치
  A3: 프로덕션 코드 변경 0 — 결함 시 패치 아닌 STOP+보고
  A4: production enforcement 완료 판정 금지(회장 별도) → 본 테스트는 검증만, 판정 0

규칙:
- 순수 함수 + 격리 tmp_path 만 (실 cron 0 · 실 발사 0 · subprocess 0 · live workspace 의존 0)
- frozen fixture 만 입력 (tests/fixtures/callback_lifecycle/) — fixture 수정/날조 0
- 프로덕션 코드 import 만, 수정 0. pytest 로 실행 가능.
"""

import glob
import importlib.util
import json
import os
import sys
from pathlib import Path

# worktree root 를 sys.path 에 추가 (안전망 — 기존 classifier/wiring 테스트와 동일 패턴)
_WORKTREE_ROOT = Path(__file__).resolve().parents[2]
if str(_WORKTREE_ROOT) not in sys.path:
    sys.path.insert(0, str(_WORKTREE_ROOT))

import pytest


def _load_real(modname: str, relpath: str):
    """tests/dispatch 패키지 shadow 를 우회해 실 dispatch 모듈을 파일 경로로 로드.

    (기존 test_callback_lifecycle_wiring_2630.py / test_callback_runtime_enforcement_2626.py
    와 동일 패턴 — dispatch/__init__.py 본문 미접촉.)
    """
    # 절대경로 동일성으로 판별 (Gemini HIGH): endswith(relpath) 는 tests/dispatch 등
    # shadow 디렉토리 경로도 매칭될 수 있어 shadow 모듈을 잘못 반환할 위험이 있다.
    target_path = (_WORKTREE_ROOT / relpath).resolve()
    existing = sys.modules.get(modname)
    if existing is not None:
        existing_path = getattr(existing, "__file__", None)
        if existing_path and Path(existing_path).resolve() == target_path:
            return existing
    spec = importlib.util.spec_from_file_location(modname, _WORKTREE_ROOT / relpath)
    assert spec is not None and spec.loader is not None
    mod = importlib.util.module_from_spec(spec)
    sys.modules[modname] = mod
    spec.loader.exec_module(mod)
    return mod


# 의존 순서대로 실 dispatch 모듈 선등록 (callback_owner_enforcer ← helper ← contract)
_load_real("dispatch.callback_owner_enforcer", "dispatch/callback_owner_enforcer.py")
_load_real("dispatch.normal_fallback_callback_helper", "dispatch/normal_fallback_callback_helper.py")
ecc = _load_real("dispatch.executor_completion_contract", "dispatch/executor_completion_contract.py")

from dispatch.normal_fallback_callback_helper import (  # noqa: E402
    _contract_fields,
    CALLBACK_KIND_NORMAL,
)

# frozen fixture root (live workspace 아님 — repo 내 frozen 자료)
FIXTURE_ROOT = Path(__file__).resolve().parents[1] / "fixtures" / "callback_lifecycle"

# 필수 fixtures 5 (신규 normal_anu_owned 포함). 디렉터리명 → 의미.
ALL_FIXTURES = (
    "normal_anu_owned",     # 신규 — NORMAL_CALLBACK_RECEIVED / NONE / normal
    "task-2625",            # SELF_KEY_FIRED_NON_AUTHORITATIVE (incident)
    "task-2628",            # FINISH_TASK_GIT_GATE_BLOCKED_BEFORE_CALLBACK + FOREIGN_DIRTY_BLOCKER
    "task-2628_plus_1",     # FALLBACK_COLLECTOR_APPLIED + ENVELOPE_PREPARED_NOT_FIRED
    "unknown_insufficient", # UNKNOWN / INSUFFICIENT_EVIDENCE
)


def _load_fixture(fixture_dir_name):
    base = FIXTURE_ROOT / fixture_dir_name
    evidence = json.loads((base / "evidence.json").read_text(encoding="utf-8"))
    expected = json.loads((base / "expected.json").read_text(encoding="utf-8"))
    return evidence, expected


def _real_nine_fields(status="ENVELOPE_PREPARED_NOT_FIRED", cron_id="E2E1234"):
    """단일소스(_contract_fields)에서 실제 9-field callback contract dict 생성.

    빌딩블록 hand-roll 이 아니라 실 contract 경로의 입력을 그대로 사용한다.
    """
    return _contract_fields(
        callback_prompt="task_id=task-2631 e2e closeout",
        kind=CALLBACK_KIND_NORMAL,
        cron_id=cron_id,
        status=status,
        envelope_only=True,
        fallback_prompt="",
        fallback_registered=False,
    )


def _temp_residue(events_dir):
    """events_dir 안의 atomic-write 임시파일 잔존 목록 (있으면 atomic write 위반)."""
    d = str(events_dir)
    return sorted(
        glob.glob(os.path.join(d, ".callback_lifecycle.*"))
        + glob.glob(os.path.join(d, "*.tmp"))
    )


# ═════════════════════════════════════════════════════════════════════════════
# 필수 검증 5~9 + A2: 5 fixture 전부 실 결선 경로 출력 == expected.json 정확 일치
# (classify_completion_lifecycle + build_callback_lifecycle_artifact 양 경로)
# ═════════════════════════════════════════════════════════════════════════════

@pytest.mark.parametrize("fixture_dir", ALL_FIXTURES)
def test_v5_9_all_five_fixtures_match_expected(fixture_dir):
    """A2 — 5 fixture(normal/fallback/unknown/self-key/git-gate) 실 결선 == 회장 매핑.

    classify_completion_lifecycle(결선 진입점) 과 build_callback_lifecycle_artifact
    (artifact 빌더) 가 **동일** 분류를 산출해야 한다(경로 간 불일치 = 결함).
    """
    evidence, expected = _load_fixture(fixture_dir)
    task_id = evidence.get("task_id", fixture_dir)

    res = ecc.classify_completion_lifecycle(evidence)
    art = ecc.build_callback_lifecycle_artifact(task_id, evidence)

    for key in ("delivery_outcome", "normal_callback_miss_cause", "evidence_completeness"):
        assert res[key] == expected[key], (
            f"[{fixture_dir}] classify {key}: got={res[key]!r} want={expected[key]!r}"
        )
        assert art[key] == expected[key], (
            f"[{fixture_dir}] artifact {key}: got={art[key]!r} want={expected[key]!r}"
        )
    # root_cause_tags: 집합 동치 + 길이 동치(중복 0)
    assert set(res["root_cause_tags"]) == set(expected["root_cause_tags"])
    assert len(res["root_cause_tags"]) == len(expected["root_cause_tags"])
    assert res["root_cause_tags"] == art["root_cause_tags"], "결선 두 경로 태그 순서까지 일치"
    # classification (expected 에 있으면)
    if "classification" in expected:
        assert res["classification"] == expected["classification"]
        assert art["classification"] == expected["classification"]


def test_v5_normal_callback_received():
    """필수 검증 5 — 신규 normal_anu_owned fixture → NORMAL / NONE / normal."""
    evidence, _ = _load_fixture("normal_anu_owned")
    res = ecc.classify_completion_lifecycle(evidence)
    assert res["delivery_outcome"] == "NORMAL_CALLBACK_RECEIVED"
    assert res["normal_callback_miss_cause"] == "NONE"
    assert res["classification"] == "normal"
    assert res["root_cause_tags"] == []
    assert res["evidence_completeness"] == "COMPLETE"


def test_v6_fallback_collector_applied():
    """필수 검증 6 — task-2628+1 → FALLBACK_COLLECTOR_APPLIED."""
    evidence, _ = _load_fixture("task-2628_plus_1")
    res = ecc.classify_completion_lifecycle(evidence)
    assert res["delivery_outcome"] == "FALLBACK_COLLECTOR_APPLIED"


def test_v7_envelope_prepared_not_fired():
    """필수 검증 7 — task-2628+1 → ENVELOPE_PREPARED_NOT_FIRED (umbrella) + 3 root cause."""
    evidence, _ = _load_fixture("task-2628_plus_1")
    res = ecc.classify_completion_lifecycle(evidence)
    assert res["normal_callback_miss_cause"] == "ENVELOPE_PREPARED_NOT_FIRED"
    assert set(res["root_cause_tags"]) == {
        "SELF_KEY_FAIL_CLOSED_BEFORE_FIRE", "BOT_APP_TOKEN_ABSENT", "REFLECTION_NOT_MERGED",
    }


def test_v8_self_key_fired_non_authoritative():
    """필수 검증 8 — task-2625 → SELF_KEY_FIRED_NON_AUTHORITATIVE (incident).

    A3 보존: SELF_KEY_FAIL_CLOSED_BEFORE_FIRE(차단 성공) 와 정반대(차단 실패→발사) 로
    분리 유지. 둘이 뭉개지면 결함.
    """
    evidence, _ = _load_fixture("task-2625")
    res = ecc.classify_completion_lifecycle(evidence)
    assert res["normal_callback_miss_cause"] == "SELF_KEY_FIRED_NON_AUTHORITATIVE"
    assert res["normal_callback_miss_cause"] != "SELF_KEY_FAIL_CLOSED_BEFORE_FIRE"
    assert res["classification"] == "incident"


def test_v9_unknown_insufficient_evidence():
    """필수 검증 9 — 증거 결핍 → UNKNOWN/INSUFFICIENT_EVIDENCE + 추정 태그 0."""
    evidence, _ = _load_fixture("unknown_insufficient")
    res = ecc.classify_completion_lifecycle(evidence)
    assert res["delivery_outcome"] == "UNKNOWN_INSUFFICIENT_EVIDENCE"
    assert res["normal_callback_miss_cause"] == "UNKNOWN_INSUFFICIENT_EVIDENCE"
    assert res["evidence_completeness"] == "MISSING"
    assert res["root_cause_tags"] == []


# ═════════════════════════════════════════════════════════════════════════════
# 필수 검증 1 + 10 + A1: 정상 closeout 실경로 → fields 10~14 append · 9-field 유지
# ═════════════════════════════════════════════════════════════════════════════

def test_v1_normal_closeout_appends_fields_10_to_14():
    """필수 검증 1 — 정상 closeout result.json 에 fields 10~14(6 keys) append."""
    evidence, expected = _load_fixture("normal_anu_owned")
    nine = _real_nine_fields()
    merged = ecc.append_lifecycle_fields(nine, evidence)

    # fields 10~14 (6 keys) 전부 존재
    for k in ecc.LIFECYCLE_RESULT_FIELDS:
        assert k in merged, f"field 10~14 누락: {k}"
    # 9-field per-callback contract + 14 logical fields(=6 keys) 동시 존재 → 15 physical keys
    assert len(merged) == 9 + len(ecc.LIFECYCLE_RESULT_FIELDS) == 15
    # 정상 closeout 분류값이 emit 됨
    assert merged["delivery_outcome"] == expected["delivery_outcome"] == "NORMAL_CALLBACK_RECEIVED"
    assert merged["normal_callback_miss_cause"] == "NONE"
    assert isinstance(merged["lifecycle_state_evidence"], dict)
    assert merged["classified_by"] == "auto-classifier"
    assert merged["applied_count"] == 0


def test_v10_nine_fields_preserved_no_overwrite():
    """필수 검증 10 — append 후에도 기존 9 fields 키/값 보존(덮어쓰기 0)."""
    evidence, _ = _load_fixture("normal_anu_owned")
    nine = _real_nine_fields()
    # manifest 가 단일소스 _contract_fields 키와 일치 (drift guard, 순서 비의존)
    assert set(nine.keys()) == set(ecc.CALLBACK_CONTRACT_9_FIELDS)
    merged = ecc.append_lifecycle_fields(nine, evidence)
    for k in ecc.CALLBACK_CONTRACT_9_FIELDS:
        assert k in merged and merged[k] == nine[k], f"9-field 훼손: {k}"
    # 입력 dict 불변(부작용 0)
    assert set(nine.keys()) == set(ecc.CALLBACK_CONTRACT_9_FIELDS)
    assert "delivery_outcome" not in nine


def test_v10b_append_only_collision_rejected():
    """append-only 보증 — 추가하려는 lifecycle 필드가 base 에 이미 존재하면
    ValueError (기존 키 덮어쓰기 절대 0). poisoned 키 normal_callback_miss_cause 는
    9-field 가 아니라 append 대상 lifecycle 필드이며, 그것이 base 에 선존재하는
    충돌 케이스를 검증한다."""
    evidence, _ = _load_fixture("normal_anu_owned")
    poisoned = {"normal_callback_miss_cause": "SHOULD_NOT_BE_OVERWRITTEN"}
    with pytest.raises(ValueError):
        ecc.append_lifecycle_fields(poisoned, evidence)


def test_anchor1_real_closeout_emits_14_fields_and_artifact(tmp_path):
    """A1 — 실제 closeout 경로 1회 통주: 9-field → 14 fields → on-disk artifact 정합.

    빌딩블록 분리 호출이 아니라 result.json fields 와 artifact 가 **같은 evidence
    경로**에서 일관되게 파생되는지(둘이 어긋나면 wiring 결함) 교차검증한다.
    """
    evidence, _ = _load_fixture("normal_anu_owned")
    task_id = evidence["task_id"]
    events_dir = tmp_path / "memory" / "events"

    # (1) 9-field 실 contract → (2) append → 14 logical fields result dict
    nine = _real_nine_fields(status="FIRED", cron_id="A17C9F00")
    result_dict = ecc.append_lifecycle_fields(nine, evidence)

    # (3) artifact 디스크 기록 → (4) 읽어와서 result_dict 의 lifecycle field 와 정합 확인
    path = ecc.write_callback_lifecycle_artifact(task_id, evidence, events_dir=str(events_dir))
    assert Path(path).exists()
    on_disk = json.loads(Path(path).read_text(encoding="utf-8"))

    for k in ("delivery_outcome", "normal_callback_miss_cause", "root_cause_tags",
              "classified_by", "applied_count", "lifecycle_state_evidence"):
        assert on_disk[k] == result_dict[k], (
            f"result.json field 와 artifact 불일치: {k} "
            f"(result={result_dict[k]!r} vs artifact={on_disk[k]!r})"
        )
    # artifact 는 result.json 9-field 를 포함하지 않는다(역할 분리 — lifecycle 분류 전용)
    for k in ecc.CALLBACK_CONTRACT_9_FIELDS:
        assert k not in on_disk


# ═════════════════════════════════════════════════════════════════════════════
# 필수 검증 2 + A1: callback_lifecycle.json 생성 + 내용 + fallback artifact 와 구분
# ═════════════════════════════════════════════════════════════════════════════

def test_v2_artifact_created_with_correct_path_and_content(tmp_path):
    """필수 검증 2 — `<task>.callback_lifecycle.json` 생성 · suffix · 내용 검증."""
    evidence, expected = _load_fixture("normal_anu_owned")
    task_id = evidence["task_id"]
    events_dir = tmp_path / "memory" / "events"

    path = ecc.write_callback_lifecycle_artifact(task_id, evidence, events_dir=str(events_dir))
    assert path.endswith(f"{task_id}.callback_lifecycle.json")
    assert Path(path).exists()

    art = json.loads(Path(path).read_text(encoding="utf-8"))
    assert art["task_id"] == task_id
    assert art["delivery_outcome"] == expected["delivery_outcome"]
    assert art["classification"] == expected["classification"]
    # fallback collector artifact 와 종류/경로 구분 (필수구현 6)
    assert art["artifact_kind"] == ecc.CALLBACK_LIFECYCLE_ARTIFACT_KIND
    assert art["schema"] == ecc.CALLBACK_LIFECYCLE_ARTIFACT_SCHEMA
    assert "independent-anu-collector" not in path
    assert "fallback_collector_applied" not in path
    # lifecycle artifact 단일 파일만 (다른 collector artifact 미생성)
    assert os.listdir(events_dir) == [f"{task_id}.callback_lifecycle.json"]


# ═════════════════════════════════════════════════════════════════════════════
# 필수 검증 3: artifact writer atomic write
#   - idempotent (byte-identical + mtime 보존)
#   - 임시파일 잔존 0
#   - 쓰기 중단 시 원본 보존 (os.replace 실패 주입)
# ═════════════════════════════════════════════════════════════════════════════

def test_v3a_idempotent_byte_identical(tmp_path):
    """동일 입력 2회 → byte-identical + 동일 내용이면 재기록 0(mtime 보존)."""
    evidence, _ = _load_fixture("normal_anu_owned")
    task_id = evidence["task_id"]
    events_dir = tmp_path / "memory" / "events"

    p1 = ecc.write_callback_lifecycle_artifact(task_id, evidence, events_dir=str(events_dir))
    b1, m1 = Path(p1).read_bytes(), os.path.getmtime(p1)
    p2 = ecc.write_callback_lifecycle_artifact(task_id, evidence, events_dir=str(events_dir))
    b2, m2 = Path(p2).read_bytes(), os.path.getmtime(p2)

    assert p1 == p2
    assert b1 == b2, "동일 입력 2회 → byte-identical 이어야 함(idempotent)"
    assert m1 == m2, "동일 내용이면 재기록 금지(mtime 보존)"


def test_v3b_no_temp_file_residue(tmp_path):
    """atomic write 후 임시파일 잔존 0 (artifact 1개만 남음)."""
    evidence, _ = _load_fixture("normal_anu_owned")
    task_id = evidence["task_id"]
    events_dir = tmp_path / "memory" / "events"

    ecc.write_callback_lifecycle_artifact(task_id, evidence, events_dir=str(events_dir))
    assert _temp_residue(events_dir) == [], "atomic write 임시파일 잔존 발견"
    assert os.listdir(events_dir) == [f"{task_id}.callback_lifecycle.json"]


def test_v3c_write_interruption_preserves_original(tmp_path, monkeypatch):
    """쓰기 중단(os.replace 실패) 시 원본 artifact 보존 + 임시파일 잔존 0.

    adversarial atomicity probe: 정상 기록 후, 내용이 바뀌는 2번째 기록 중
    os.replace 가 예외를 던지도록 주입한다. 계약(docstring: '쓰기 도중 중단/오류에도
    artifact 가 손상/불완전 상태로 남지 않는다')대로면:
      - 원본 내용 불변, 임시파일 잔존 0, 예외 전파.
    """
    task_id = "task-NORMAL-anu-owned"
    events_dir = tmp_path / "memory" / "events"

    # 1) 정상(normal) 내용으로 최초 기록 → 원본 확보
    normal_ev, _ = _load_fixture("normal_anu_owned")
    p = ecc.write_callback_lifecycle_artifact(task_id, normal_ev, events_dir=str(events_dir))
    original = Path(p).read_bytes()

    # 2) 내용이 달라지는 evidence(task-2625) 로 같은 path 재기록 시도 → os.replace 실패 주입
    other_ev, _ = _load_fixture("task-2625")

    def _boom_replace(*_a, **_k):
        raise OSError("simulated write interruption (os.replace)")

    monkeypatch.setattr(os, "replace", _boom_replace, raising=True)

    with pytest.raises(OSError):
        ecc.write_callback_lifecycle_artifact(task_id, other_ev, events_dir=str(events_dir))

    # 원본 보존 (부분기록/손상 0)
    assert Path(p).read_bytes() == original, "쓰기 중단 시 원본이 손상/교체됨 — atomic 위반"
    # 임시파일 잔존 0
    assert _temp_residue(events_dir) == [], "쓰기 중단 후 임시파일 잔존 — cleanup 실패"
    assert os.listdir(events_dir) == [f"{task_id}.callback_lifecycle.json"]


# ═════════════════════════════════════════════════════════════════════════════
# 필수 검증 4: callback gate PASS / notification sent / collector received 분리 기록
# ═════════════════════════════════════════════════════════════════════════════

def test_v4_stage_separation_three_keys_present():
    """3단계 키가 항상 분리 존재."""
    for fx in ("normal_anu_owned", "task-2625", "task-2628", "task-2628_plus_1"):
        evidence, _ = _load_fixture(fx)
        art = ecc.build_callback_lifecycle_artifact(evidence.get("task_id", fx), evidence)
        sep = art["callback_stage_separation"]
        assert set(sep.keys()) == {"callback_gate_pass", "notification_sent", "collector_received"}, fx


def test_v4_normal_path_all_three_true():
    """정상 경로(normal_anu_owned): gate PASS · sent · received 전부 True (유일 케이스)."""
    evidence, _ = _load_fixture("normal_anu_owned")
    art = ecc.build_callback_lifecycle_artifact(evidence["task_id"], evidence)
    sep = art["callback_stage_separation"]
    assert sep == {"callback_gate_pass": True, "notification_sent": True, "collector_received": True}


def test_v4_gate_pass_not_equal_fired_not_equal_received():
    """task-2628+1 — gate PASS ≠ fired ≠ received 분리 입증.

    gate 통과했으나 normal 미발사(sent=False)인데 fallback collector 가 수집
    (received=True). 세 신호가 뭉개지면(예: sent 가 received 를 따라 True) 결함.
    """
    evidence, _ = _load_fixture("task-2628_plus_1")
    art = ecc.build_callback_lifecycle_artifact("task-2628+1", evidence)
    sep = art["callback_stage_separation"]
    assert sep["callback_gate_pass"] is True
    assert sep["notification_sent"] is False, "normal 미발사인데 sent=True → 단계 혼동 결함"
    assert sep["collector_received"] is True
    # 분리 입증: sent 와 received 가 다른 값
    assert sep["notification_sent"] != sep["collector_received"]


def test_v4_git_gate_blocked_gate_pass_false():
    """task-2628 — GIT-GATE 차단 시 callback_gate_pass=False (gate 단계서 차단)."""
    evidence, _ = _load_fixture("task-2628")
    art = ecc.build_callback_lifecycle_artifact("task-2628", evidence)
    sep = art["callback_stage_separation"]
    assert sep["callback_gate_pass"] is False
    assert sep["notification_sent"] is False
    assert sep["collector_received"] is False


# ═════════════════════════════════════════════════════════════════════════════
# live workspace 의존 0 · 실 cron/발사/subprocess 0 (스펙 §구현 가이드 + 금지)
# ═════════════════════════════════════════════════════════════════════════════

def test_no_live_workspace_dependency(tmp_path, monkeypatch):
    """WORKSPACE_ROOT override → events dir 가 격리 경로로 해석(live 의존 0)."""
    monkeypatch.setenv("WORKSPACE_ROOT", str(tmp_path))
    assert ecc.default_events_dir() == str(tmp_path / "memory" / "events")
    evidence, _ = _load_fixture("normal_anu_owned")
    # classify 는 입력 dict 만으로 동작(파일/네트워크 read 0)
    assert ecc.classify_completion_lifecycle(evidence)["delivery_outcome"] == "NORMAL_CALLBACK_RECEIVED"
    # 기본 경로 write 도 격리 tmp 아래에만 생성
    p = ecc.write_callback_lifecycle_artifact(evidence["task_id"], evidence)
    assert str(tmp_path) in p and Path(p).exists()


def test_no_callback_refire_no_subprocess(tmp_path, monkeypatch):
    """전체 closeout 경로가 subprocess/os.system 을 단 한 번도 호출하지 않음.

    실 cron 등록/제거 0 · callback 재발사 0 (스펙 금지). public 발사 API 부재도 확인.
    """
    calls = []

    def _boom(*a, **k):
        calls.append(("subprocess", a, k))
        raise AssertionError("subprocess 호출 — callback 재발사 금지 위반")

    def _boom_system(*a, **k):
        calls.append(("os.system", a, k))
        raise AssertionError("os.system 호출 — callback 재발사 금지 위반")

    import subprocess
    # 외부 명령 실행 가능 API 를 폭넓게 모킹 (callback 재발사 금지 검증 범위 확대)
    for _api in ("run", "Popen", "call", "check_call", "check_output", "getoutput", "getstatusoutput"):
        if hasattr(subprocess, _api):
            monkeypatch.setattr(subprocess, _api, _boom, raising=True)
    monkeypatch.setattr(os, "system", _boom_system, raising=True)
    for _osapi in ("popen", "spawnl", "spawnle", "spawnlp", "spawnlpe",
                   "spawnv", "spawnve", "spawnvp", "spawnvpe",
                   "posix_spawn", "posix_spawnp",
                   "execl", "execle", "execlp", "execlpe",
                   "execv", "execve", "execvp", "execvpe"):
        if hasattr(os, _osapi):
            monkeypatch.setattr(os, _osapi, _boom_system, raising=True)

    events_dir = tmp_path / "ev"
    for fx in ALL_FIXTURES:
        evidence, _ = _load_fixture(fx)
        task_id = evidence.get("task_id", fx)
        nine = _real_nine_fields()
        ecc.append_lifecycle_fields(nine, evidence)
        ecc.build_callback_lifecycle_artifact(task_id, evidence)
        ecc.write_callback_lifecycle_artifact(task_id, evidence, events_dir=str(events_dir))

    assert calls == [], f"발사/subprocess 호출 0 이어야 함, got {calls}"
    # 결선 모듈에 cron 발사/등록 public 함수 부재(설계상 데이터-only · A3/A4)
    for forbidden in ("fire_callback", "register_cron", "remove_cron", "send_callback"):
        assert not hasattr(ecc, forbidden), f"금지 API 노출: {forbidden}"


# ═════════════════════════════════════════════════════════════════════════════
# frozen anchor 메타 검증 (A2: 5 fixture 결선 일치 일괄 / A3: 프로덕션 무변경 표식)
# ═════════════════════════════════════════════════════════════════════════════

def test_anchor2_five_cases_classifier_alignment():
    """A2 — normal/fallback/unknown/self-key(+git-gate) 5 사례가 결선 분류와 일치."""
    summary = {}
    for fx in ALL_FIXTURES:
        evidence, expected = _load_fixture(fx)
        res = ecc.classify_completion_lifecycle(evidence)
        assert res["delivery_outcome"] == expected["delivery_outcome"], fx
        assert res["normal_callback_miss_cause"] == expected["normal_callback_miss_cause"], fx
        summary[fx] = (res["delivery_outcome"], res["normal_callback_miss_cause"])
    # 5 사례 (축A × 축B) 조합이 서로 달라 2축 모델 타당성 유지(중복 분류 0)
    assert len(set(summary.values())) == len(ALL_FIXTURES), f"분류 조합 중복: {summary}"