# -*- coding: utf-8 -*-
# pyright: reportMissingImports=false, reportFunctionMemberAccess=false
"""task-2756 provenance ADOPTED → ANU-owned callback bridge 회귀 테스트.

검증 대상:
  dispatch.anu_pickup_driver.process_one 의 provenance ADOPTED 분기에서
  동작하는 _provenance_callback_bridge 를 포함한 end-to-end 통합 검증.

절대 제약:
  - 실제 ANU key 는 raw literal 금지 — ANU_KEY 상수 변수(callback_owner_enforcer
    .ANU_KEY_2553 import)를 통해서만 참조한다. 실 builder 통합 검증도 상수 변수로만.
  - pickup_once 실호출 0 — 전부 mock 주입.
  - tmp_path fixture 만 사용. 실제 memory 디렉토리에 쓰지 않는다.
  - dispatch/anu_pickup_driver.py 수정 0 (읽기만).
"""
from __future__ import annotations

import json
import os
import sys
import time as _time
import types
from pathlib import Path

# ── sys.path 부트스트랩 (test_provenance_path_2753.py 패턴 동일) ─────────────
_ROOT = Path(__file__).resolve().parents[2]
if str(_ROOT) not in sys.path:
    sys.path.insert(0, str(_ROOT))

import importlib.util as _ilu

_real_init = _ROOT / "dispatch" / "__init__.py"
_cached = sys.modules.get("dispatch")
if _cached is None or (getattr(_cached, "__file__", "") or "") != str(_real_init):
    for _k in [k for k in list(sys.modules) if k == "dispatch" or k.startswith("dispatch.")]:
        del sys.modules[_k]
    _spec = _ilu.spec_from_file_location(
        "dispatch", _real_init, submodule_search_locations=[str(_ROOT / "dispatch")]
    )
    assert _spec is not None and _spec.loader is not None
    _pkg = _ilu.module_from_spec(_spec)
    sys.modules["dispatch"] = _pkg
    _spec.loader.exec_module(_pkg)

from dispatch import anu_pickup_driver as drv  # noqa: E402
from dispatch import anu_executor_provenance as prov  # noqa: E402

# ANU key: callback_owner_enforcer 에서 읽음 (literal 0 — 상수 변수 참조만)
from dispatch.callback_owner_enforcer import ANU_KEY_2553 as _ANU_KEY  # noqa: E402

# 독립 ANU key (task-2756 명시: 이것만 build_anu_owned_callback_request PASS)
ANU_KEY: str = _ANU_KEY

# ── 상수 ──────────────────────────────────────────────────────────────────────
TASK_ID = "task-2756"
CAPSULE_SHA = "abcdef01" * 8  # 64-hex 테스트 sha (실제 키 아님)
EXECUTOR_SELF_KEY = "executor-self-key-2756"   # ANU 가 아닌 executor 키 (mock 용)
NON_ANU_KEY = "deadbeef12345678"              # ANU 아닌 임의 키


def _valid_result() -> dict:
    """provenance-path 통과용 valid executor_result (envelope 없음)."""
    return {
        "task_id": TASK_ID,
        "completion_signal": "EXECUTOR_RESULT_WRITTEN",
        "terminal_status": "PASS",
        "capsule_sha256": CAPSULE_SHA,
        "report_path": "report.md",
        # collector_envelope 없음 → provenance-path 진입.
    }


def _cap_reader(*, task_id=TASK_ID, source_sha256=CAPSULE_SHA):
    return lambda tid: {"task_id": task_id, "source_sha256": source_sha256}


# ── 헬퍼 (test_provenance_path_2753.py 패턴 재사용) ──────────────────────────
def _make_dirs(tmp_path: Path) -> Path:
    (tmp_path / "memory" / "events" / "p0b_inbox").mkdir(parents=True, exist_ok=True)
    (tmp_path / "memory" / "state").mkdir(parents=True, exist_ok=True)
    (tmp_path / "memory" / "capabilities").mkdir(parents=True, exist_ok=True)
    return tmp_path


def _events_dir(root: Path) -> Path:
    return root / "memory" / "events" / "p0b_inbox"


def _write_result(root: Path, name=None, payload=None) -> Path:
    if name is None:
        name = f"{TASK_ID}.result.json"
    if payload is None:
        payload = _valid_result()
    p = _events_dir(root) / name
    p.write_text(json.dumps(payload), encoding="utf-8")
    return p


def _enable_activation(root: Path) -> None:
    flag = root / "memory" / "state" / "p0b_driver_enabled"
    flag.parent.mkdir(parents=True, exist_ok=True)
    flag.write_text(drv.ACTIVATION_ENABLED + "\n", encoding="utf-8")


def _age(path, seconds: float = 10.0) -> None:
    past = _time.time() - seconds
    os.utime(str(path), (past, past))


_NO_SLEEP = lambda *a, **k: None  # noqa: E731


def make_pickup_mock(verdict: str = "WAKE_BUILT"):
    calls: list = []

    def _pickup(*args, **kwargs):
        calls.append((args, kwargs))
        return types.SimpleNamespace(
            verdict=verdict, ok=(verdict == "WAKE_BUILT"),
            argv=["x"], task_id=TASK_ID, reasons=[],
        )

    _pickup.calls = calls  # type: ignore[attr-defined]
    return _pickup


def make_launch_spy():
    """callback_launch_fn spy — 호출 인자(request)를 기록."""
    calls: list = []
    captured: list = []  # 첫 번째 인자(request 객체) 저장

    def _launch(request, *args, **kwargs):
        calls.append((request, args, kwargs))
        captured.append(request)
        return types.SimpleNamespace(decision="LAUNCHED")

    _launch.calls = calls  # type: ignore[attr-defined]
    _launch.captured = captured  # type: ignore[attr-defined]
    return _launch


def make_collector_writer_spy():
    calls: list = []

    def _writer(collector, path):
        calls.append((collector, path))
        return path

    _writer.calls = calls  # type: ignore[attr-defined]
    return _writer


# ── gate mock 헬퍼 ────────────────────────────────────────────────────────────

class _GateReady:
    """owner_gate_fn 이 PASS 를 반환하는 mock gate 결과."""
    ready = True
    status = drv.GATE_READY_STATUS


class _GateNotReady:
    """owner_gate_fn 이 FAIL 을 반환하는 mock gate 결과."""
    ready = False
    status = "OWNER_PICKUP_NOT_READY"


def gate_pass(result_path, **kw):
    return _GateReady()


def gate_fail(result_path, **kw):
    return _GateNotReady()


# ── process_one 을 ADOPTED 분기로 진입시키는 공통 헬퍼 ──────────────────────

def _run_adopted(
    tmp_path,
    *,
    callback_launch_fn=None,
    owner_gate_fn=None,
    callback_owner_key: str = "",
    callback_executor_key: str = "",
    callback_chat_id: str = "",
    callback_at: str = "",
    callback_dedupe_seen=None,
    payload=None,
):
    """provenance-path ADOPTED 분기를 통해 process_one 을 호출하고 DriverRecord 반환.

    test_provenance_path_2753.py 의 test_integration_provenance_adopted 패턴을 그대로 가져옴.
    provenance_fn 은 SpyResult(ok=True) 반환 함수로 고정한다.

    cron_id 는 process_one 파라미터로 직접 주입한다(wrapper 없이 실제 production 경로 검증).
    callback_request_builder 는 기본 None(= 실제 build_anu_owned_callback_request 사용).
    """
    root = _make_dirs(tmp_path)
    _enable_activation(root)
    pickup = make_pickup_mock()
    writer = make_collector_writer_spy()

    p = _write_result(root, payload=payload)
    _age(p)

    # sha256 계산 (provenance_fn spy 가 반환할 source_result_sha256)
    import hashlib
    sha = hashlib.sha256(p.read_bytes()).hexdigest()

    def _provenance_fn(result, result_path, **kwargs):
        return prov.ProvenanceResult(ok=True, source_result_sha256=sha)

    rec = drv.process_one(
        str(p),
        root=str(root),
        pickup_fn=pickup,
        sleep_fn=_NO_SLEEP,
        provenance_path_enabled=True,
        capability_reader=_cap_reader(),
        collector_writer_fn=writer,
        provenance_fn=_provenance_fn,
        callback_launch_fn=callback_launch_fn,
        owner_gate_fn=owner_gate_fn,
        callback_owner_key=callback_owner_key,
        callback_executor_key=callback_executor_key,
        callback_chat_id=callback_chat_id,
        callback_at=callback_at,
        callback_request_builder=None,
        callback_dedupe_seen=callback_dedupe_seen,
        callback_dispatch_cron_id="dispatch-cron-2756",
        callback_normal_collector_cron_id="normal-collector-cron-2756",
        callback_no_fallback=True,
    )
    return rec, sha


# ═════════════════════════════════════════════════════════════════════════════
# 테스트 케이스 9개 + α
# ═════════════════════════════════════════════════════════════════════════════

# ── 테스트 1 ──
def test_bridge_fires_on_adopted_with_gate_pass(tmp_path):
    """callback_launch_fn + owner_gate_fn(ready=True) + ANU owner_key → fire 정확히 1회.
    request.verdict==PASS, DriverRecord.callback_launch_verdict==BRIDGE_CALLBACK_REQUEST_BUILT,
    verdict==PROVENANCE_ADOPTED.
    """
    launch_spy = make_launch_spy()

    rec, _sha = _run_adopted(
        tmp_path,
        callback_launch_fn=launch_spy,
        owner_gate_fn=gate_pass,
        callback_owner_key=ANU_KEY,
        callback_executor_key=EXECUTOR_SELF_KEY,  # ANU 가 아닌 executor key
        callback_chat_id="chat-2756",
        callback_at="10m",
    )

    # ① fire 정확히 1회
    assert len(launch_spy.calls) == 1, (
        f"callback_launch_fn 호출 횟수 {len(launch_spy.calls)} != 1"
    )
    # ② 캡처된 request.verdict == "PASS"
    req = launch_spy.captured[0]
    assert getattr(req, "verdict", None) == "PASS", (
        f"request.verdict={getattr(req, 'verdict', None)!r} != 'PASS'"
    )
    # ③ DriverRecord.callback_launch_verdict == BRIDGE_CALLBACK_REQUEST_BUILT
    assert rec.callback_launch_verdict == drv.BRIDGE_FIRED, (
        f"callback_launch_verdict={rec.callback_launch_verdict!r}"
    )
    # ④ verdict == PROVENANCE_ADOPTED (bridge 가 verdict 를 바꾸지 않음)
    assert rec.verdict == drv.VERDICT_PROVENANCE_ADOPTED


# ── 테스트 2 ──
def test_bridge_payload_contains_required_fields(tmp_path):
    """bridge request.argv 직렬화 문자열에 필수 필드 포함 확인.
    task_id / "PROVENANCE_ADOPTED"(terminal_status) / source_result_sha256 /
    "ANU"(collector_role) / "OWNER_ANU"(owner_proof) 가 argv join 문자열에 있어야 함.
    """
    launch_spy = make_launch_spy()

    rec, sha = _run_adopted(
        tmp_path,
        callback_launch_fn=launch_spy,
        owner_gate_fn=gate_pass,
        callback_owner_key=ANU_KEY,
        callback_executor_key=EXECUTOR_SELF_KEY,
        callback_chat_id="chat-2756",
        callback_at="10m",
    )

    assert len(launch_spy.captured) == 1
    req = launch_spy.captured[0]
    # argv 가 있어야 함(PASS → argv is not None)
    argv = getattr(req, "argv", None)
    assert argv is not None, "request.argv is None (FAIL 상태?)"
    argv_str = " ".join(str(a) for a in argv)

    # 필수 필드 검증
    assert TASK_ID in argv_str, f"task_id '{TASK_ID}' not in argv"
    assert drv.VERDICT_PROVENANCE_ADOPTED in argv_str, \
        f"'PROVENANCE_ADOPTED' (terminal_status) not in argv"
    assert sha in argv_str, f"source_result_sha256 '{sha}' not in argv"
    assert "ANU" in argv_str, "'ANU' (collector_role) not in argv"
    assert "OWNER_ANU" in argv_str, "'OWNER_ANU' (owner_proof) not in argv"


# ── 테스트 3 ──
def test_bridge_self_key_fails(tmp_path):
    """callback_owner_key == callback_executor_key (self-callback) → fire 0,
    callback_launch_verdict==BRIDGE_CALLBACK_OWNER_FAIL.
    실제 build_anu_owned_callback_request 를 경유하여 enforce 가 FAIL 내는지 검증.
    cron_id 는 valid 하게 주입하되 owner_key 가 self-key(non-ANU) 이므로 FAIL.
    """
    launch_spy = make_launch_spy()
    # self-key: owner_key 와 executor_key 를 동일하게 주입
    # (ANU 가 아닌 임의 key 를 owner+executor 동일로 설정 → enforce 가 self-key + non-ANU 로 이중 FAIL)
    rec, _ = _run_adopted(
        tmp_path,
        callback_launch_fn=launch_spy,
        owner_gate_fn=gate_pass,
        callback_owner_key=EXECUTOR_SELF_KEY,   # non-ANU key
        callback_executor_key=EXECUTOR_SELF_KEY,  # same → self-callback
        callback_chat_id="chat-2756",
        callback_at="10m",
    )

    assert len(launch_spy.calls) == 0, \
        f"self-key 케이스에서 fire {len(launch_spy.calls)}회 발생 (0 이어야 함)"
    assert rec.callback_launch_verdict == drv.BRIDGE_CALLBACK_OWNER_FAIL, \
        f"callback_launch_verdict={rec.callback_launch_verdict!r}"


# ── 테스트 4 ──
def test_bridge_non_anu_owner_key_fails(tmp_path):
    """callback_owner_key='deadbeef...'(비 ANU key) → fire 0, BRIDGE_CALLBACK_OWNER_FAIL.
    실제 build_anu_owned_callback_request 사용 — enforce 가 ANU 아닌 key 를 거부.
    cron_id 는 valid 하게 주입하되 owner_key 가 비-ANU 이므로 FAIL.
    """
    launch_spy = make_launch_spy()

    rec, _ = _run_adopted(
        tmp_path,
        callback_launch_fn=launch_spy,
        owner_gate_fn=gate_pass,
        callback_owner_key=NON_ANU_KEY,
        callback_executor_key=EXECUTOR_SELF_KEY,
        callback_chat_id="chat-2756",
        callback_at="10m",
    )

    assert len(launch_spy.calls) == 0, \
        f"비 ANU key 케이스에서 fire {len(launch_spy.calls)}회 발생 (0 이어야 함)"
    assert rec.callback_launch_verdict == drv.BRIDGE_CALLBACK_OWNER_FAIL, \
        f"callback_launch_verdict={rec.callback_launch_verdict!r}"


# ── 테스트 5 ──
def test_bridge_owner_gate_missing_fail_closed(tmp_path):
    """callback_launch_fn 주입 + owner_gate_fn=None → fail-closed: fire 0,
    callback_launch_verdict==BRIDGE_OWNER_GATE_BLOCKED.

    주의: process_one 의 task-2741-r2 MANDATORY fail-closed(envelope 경로) 는
    ADOPTED 분기(envelope 부재) 에서는 적용되지 않는다.
    bridge 내부의 fail-closed 검사(② 단계)로 BRIDGE_OWNER_GATE_BLOCKED 가 반환됨.
    """
    launch_spy = make_launch_spy()

    rec, _ = _run_adopted(
        tmp_path,
        callback_launch_fn=launch_spy,
        owner_gate_fn=None,  # 부재 → fail-closed
        callback_owner_key=ANU_KEY,
        callback_executor_key=EXECUTOR_SELF_KEY,
        callback_chat_id="chat-2756",
        callback_at="10m",
    )

    assert len(launch_spy.calls) == 0, \
        f"owner_gate_fn 부재 케이스에서 fire {len(launch_spy.calls)}회 발생 (0 이어야 함)"
    assert rec.callback_launch_verdict == drv.BRIDGE_OWNER_GATE_BLOCKED, \
        f"callback_launch_verdict={rec.callback_launch_verdict!r}"


# ── 테스트 6 ──
def test_bridge_owner_gate_fail_no_fire(tmp_path):
    """owner_gate_fn 이 ready=False 결과 반환 → fire 0, BRIDGE_OWNER_GATE_FAIL."""
    launch_spy = make_launch_spy()

    rec, _ = _run_adopted(
        tmp_path,
        callback_launch_fn=launch_spy,
        owner_gate_fn=gate_fail,  # gate not ready
        callback_owner_key=ANU_KEY,
        callback_executor_key=EXECUTOR_SELF_KEY,
        callback_chat_id="chat-2756",
        callback_at="10m",
    )

    assert len(launch_spy.calls) == 0, \
        f"gate FAIL 케이스에서 fire {len(launch_spy.calls)}회 발생 (0 이어야 함)"
    assert rec.callback_launch_verdict == drv.BRIDGE_OWNER_GATE_FAIL, \
        f"callback_launch_verdict={rec.callback_launch_verdict!r}"


# ── 테스트 7 ──
def test_bridge_off_when_no_launch_fn(tmp_path):
    """callback_launch_fn=None (기본, 무회귀) → fire 0, callback_launch_verdict is None,
    verdict==PROVENANCE_ADOPTED 정상.
    """
    launch_spy = make_launch_spy()

    # callback_launch_fn 을 주입하지 않는다
    rec, _ = _run_adopted(
        tmp_path,
        callback_launch_fn=None,  # 명시적으로 None
        owner_gate_fn=gate_pass,
        callback_owner_key=ANU_KEY,
        callback_executor_key=EXECUTOR_SELF_KEY,
        callback_chat_id="chat-2756",
        callback_at="10m",
    )

    # launch_spy 는 주입하지 않았으므로 calls 는 없지만, 여기서는 rec 를 검증
    assert rec.callback_launch_verdict is None, \
        f"callback_launch_verdict={rec.callback_launch_verdict!r} (None 이어야 함)"
    assert rec.verdict == drv.VERDICT_PROVENANCE_ADOPTED, \
        f"verdict={rec.verdict!r}"
    # fire 는 0 (launch_spy 가 없으므로 검증 간접적으로 verdict 와 callback_launch_verdict 로 확인)


# ── 테스트 8 ──
def test_bridge_dedupe_skips_duplicate(tmp_path):
    """dedupe_seen set 양방향 검증.
    1) 처음 호출(빈 set) → fire 1회 + set 에 key 추가됨.
    2) 동일 set 두 번째 호출 → fire 0 + BRIDGE_DEDUPED.
    """
    # ── 첫 번째 호출: 빈 set → fire 1회 ──
    dedupe_seen: set = set()
    launch_spy = make_launch_spy()

    root1 = _make_dirs(tmp_path / "run1")
    _enable_activation(root1)
    p1 = _write_result(root1)
    _age(p1)

    import hashlib
    sha1 = hashlib.sha256(p1.read_bytes()).hexdigest()
    expected_dedupe_key = f"{TASK_ID}:{sha1}"

    def _prov_fn_1(result, result_path, **kwargs):
        return prov.ProvenanceResult(ok=True, source_result_sha256=sha1)

    rec1 = drv.process_one(
        str(p1),
        root=str(root1),
        pickup_fn=make_pickup_mock(),
        sleep_fn=_NO_SLEEP,
        provenance_path_enabled=True,
        capability_reader=_cap_reader(),
        collector_writer_fn=make_collector_writer_spy(),
        provenance_fn=_prov_fn_1,
        callback_launch_fn=launch_spy,
        owner_gate_fn=gate_pass,
        callback_owner_key=ANU_KEY,
        callback_executor_key=EXECUTOR_SELF_KEY,
        callback_chat_id="chat-2756",
        callback_at="10m",
        callback_request_builder=None,
        callback_dedupe_seen=dedupe_seen,
        callback_dispatch_cron_id="dispatch-cron-2756",
        callback_normal_collector_cron_id="normal-collector-cron-2756",
        callback_no_fallback=True,
    )

    # 첫 번째 호출: fire 1회
    assert len(launch_spy.calls) == 1, \
        f"첫 번째 호출에서 fire 횟수 {len(launch_spy.calls)} != 1"
    # dedupe_seen 에 key 가 추가됨
    assert expected_dedupe_key in dedupe_seen, \
        f"dedupe_seen 에 '{expected_dedupe_key}' 가 추가되지 않음. set={dedupe_seen}"
    assert rec1.callback_launch_verdict == drv.BRIDGE_FIRED

    # ── 두 번째 호출: 동일 set 재사용 → fire 0 + BRIDGE_DEDUPED ──
    root2 = _make_dirs(tmp_path / "run2")
    _enable_activation(root2)
    p2 = _write_result(root2)
    _age(p2)

    _sha2 = hashlib.sha256(p2.read_bytes()).hexdigest()  # noqa: F841

    def _prov_fn_2(result, result_path, **kwargs):
        # dedupe 는 task_id+sha 기준. 두 번째 호출도 동일한 sha1 을 반환하여 dedupe 유발.
        return prov.ProvenanceResult(ok=True, source_result_sha256=sha1)

    rec2 = drv.process_one(
        str(p2),
        root=str(root2),
        pickup_fn=make_pickup_mock(),
        sleep_fn=_NO_SLEEP,
        provenance_path_enabled=True,
        capability_reader=_cap_reader(),
        collector_writer_fn=make_collector_writer_spy(),
        provenance_fn=_prov_fn_2,
        callback_launch_fn=launch_spy,
        owner_gate_fn=gate_pass,
        callback_owner_key=ANU_KEY,
        callback_executor_key=EXECUTOR_SELF_KEY,
        callback_chat_id="chat-2756",
        callback_at="10m",
        callback_request_builder=None,
        callback_dedupe_seen=dedupe_seen,  # 동일 set 재사용
        callback_dispatch_cron_id="dispatch-cron-2756",
        callback_normal_collector_cron_id="normal-collector-cron-2756",
        callback_no_fallback=True,
    )

    # 두 번째 호출: fire 추가 없음(총 여전히 1회)
    assert len(launch_spy.calls) == 1, \
        f"두 번째 호출 후 fire 횟수 {len(launch_spy.calls)} != 1 (dedupe 실패)"
    assert rec2.callback_launch_verdict == drv.BRIDGE_DEDUPED, \
        f"두 번째 callback_launch_verdict={rec2.callback_launch_verdict!r} (BRIDGE_DEDUPED 이어야 함)"


# ── 테스트 9 ──
def test_bridge_not_called_on_quarantine_and_invalid(tmp_path):
    """ADOPTED 아닌 분기(quarantine/invalid)에서는 bridge fire 가 없음 검증.

    케이스 A: provenance_path_enabled=False + envelope 없음 → QUARANTINE(owner_unprovable)
    케이스 B: provenance_fn 이 ok=False → QUARANTINE(provenance_fail)
    양쪽 모두 callback_launch_fn 이 0회 호출되어야 함.
    """
    # ── 케이스 A: provenance_path_enabled=False → owner_unprovable quarantine ──
    launch_spy_a = make_launch_spy()
    root_a = _make_dirs(tmp_path / "case_a")
    _enable_activation(root_a)
    p_a = _write_result(root_a)
    _age(p_a)

    rec_a = drv.process_one(
        str(p_a),
        root=str(root_a),
        pickup_fn=make_pickup_mock(),
        sleep_fn=_NO_SLEEP,
        provenance_path_enabled=False,  # flag OFF
        capability_reader=_cap_reader(),
        callback_launch_fn=launch_spy_a,
        owner_gate_fn=gate_pass,
        callback_owner_key=ANU_KEY,
        callback_executor_key=EXECUTOR_SELF_KEY,
    )

    assert rec_a.verdict == drv.VERDICT_QUARANTINE, \
        f"케이스A verdict={rec_a.verdict!r}"
    assert len(launch_spy_a.calls) == 0, \
        f"케이스A: owner_unprovable quarantine 에서 fire {len(launch_spy_a.calls)}회 발생"

    # ── 케이스 B: provenance_fn ok=False → quarantine ──
    launch_spy_b = make_launch_spy()
    root_b = _make_dirs(tmp_path / "case_b")
    _enable_activation(root_b)
    p_b = _write_result(root_b)
    _age(p_b)

    def _prov_fn_fail(result, result_path, **kwargs):
        return prov.ProvenanceResult(ok=False, reason=prov.REASON_CAPSULE_SHA_MISMATCH)

    rec_b = drv.process_one(
        str(p_b),
        root=str(root_b),
        pickup_fn=make_pickup_mock(),
        sleep_fn=_NO_SLEEP,
        provenance_path_enabled=True,
        capability_reader=_cap_reader(),
        provenance_fn=_prov_fn_fail,
        collector_writer_fn=make_collector_writer_spy(),
        callback_launch_fn=launch_spy_b,
        owner_gate_fn=gate_pass,
        callback_owner_key=ANU_KEY,
        callback_executor_key=EXECUTOR_SELF_KEY,
    )

    assert rec_b.verdict == drv.VERDICT_QUARANTINE, \
        f"케이스B verdict={rec_b.verdict!r}"
    assert len(launch_spy_b.calls) == 0, \
        f"케이스B: provenance FAIL quarantine 에서 fire {len(launch_spy_b.calls)}회 발생"


# ── 추가 테스트: bridge verdict 가 PROVENANCE_ADOPTED verdict 에 영향 없음 재확인 ──

def test_bridge_callback_launch_verdict_does_not_change_main_verdict(tmp_path):
    """bridge 결과(BRIDGE_OWNER_GATE_FAIL 등)가 record.verdict 를 바꾸지 않음.
    ADOPTED 분기에서 어떤 bridge 결과라도 verdict 는 항상 PROVENANCE_ADOPTED.
    """
    # gate_fail → BRIDGE_OWNER_GATE_FAIL, 그래도 verdict == PROVENANCE_ADOPTED
    launch_spy = make_launch_spy()
    rec, _ = _run_adopted(
        tmp_path,
        callback_launch_fn=launch_spy,
        owner_gate_fn=gate_fail,
        callback_owner_key=ANU_KEY,
        callback_executor_key=EXECUTOR_SELF_KEY,
        callback_chat_id="chat-2756",
        callback_at="10m",
    )
    assert rec.verdict == drv.VERDICT_PROVENANCE_ADOPTED
    assert rec.callback_launch_verdict == drv.BRIDGE_OWNER_GATE_FAIL
    assert len(launch_spy.calls) == 0


def test_bridge_fires_exactly_once_not_twice(tmp_path):
    """ADOPTED 분기에서 fire 가 정확히 1회(중복 없음) 발생함을 재확인."""
    launch_spy = make_launch_spy()
    rec, _ = _run_adopted(
        tmp_path,
        callback_launch_fn=launch_spy,
        owner_gate_fn=gate_pass,
        callback_owner_key=ANU_KEY,
        callback_executor_key=EXECUTOR_SELF_KEY,
        callback_chat_id="chat-2756",
        callback_at="10m",
    )
    assert len(launch_spy.calls) == 1
    assert rec.callback_launch_verdict == drv.BRIDGE_FIRED
    assert rec.verdict == drv.VERDICT_PROVENANCE_ADOPTED


# ═════════════════════════════════════════════════════════════════════════════
# task-2756+1 추가 테스트: HIGH 예외 graceful + Medium-1 cron_id + Medium-2 move_err
# ═════════════════════════════════════════════════════════════════════════════

def make_raising_launch():
    """callback_launch_fn 이 예외를 raise 하는 spy (HIGH fix 검증용)."""
    calls: list = []

    def _launch(request, *args, **kwargs):
        calls.append(request)
        raise RuntimeError("boom-2756 launch failure")

    _launch.calls = calls  # type: ignore[attr-defined]
    return _launch


def test_bridge_launch_exception_graceful_no_crash(tmp_path):
    """★ HIGH: callback_launch_fn 이 예외 raise → driver crash 0.
    bridge_verdict(callback_launch_verdict)==None, DriverRecord.error 에 'callback launch 실패'
    라벨 기록(영구 누락 대신 관측 가능), verdict==PROVENANCE_ADOPTED 불변, launch 시도 1회.
    """
    raising = make_raising_launch()

    rec, _sha = _run_adopted(
        tmp_path,
        callback_launch_fn=raising,
        owner_gate_fn=gate_pass,
        callback_owner_key=ANU_KEY,
        callback_executor_key=EXECUTOR_SELF_KEY,
        callback_chat_id="chat-2756",
        callback_at="10m",
    )

    # ① driver crash 0 → DriverRecord 반환됨
    assert rec is not None
    # ② 예외 graceful capture → callback_launch_verdict is None
    assert rec.callback_launch_verdict is None, (
        f"callback_launch_verdict={rec.callback_launch_verdict!r} (None 이어야 함)"
    )
    # ③ error 라벨 관측 가능
    assert rec.error is not None and "callback launch 실패" in rec.error, (
        f"rec.error={rec.error!r} ('callback launch 실패' 포함되어야 함)"
    )
    # ④ verdict 불변
    assert rec.verdict == drv.VERDICT_PROVENANCE_ADOPTED
    # ⑤ launch 가 실제 1회 시도됨
    assert len(raising.calls) == 1


def test_bridge_request_cron_id_consistency(tmp_path):
    """★ Medium-1: 실제 builder 경유 시 request.cron_id == normal_collector_cron_id 일관성.
    _run_adopted 는 callback_normal_collector_cron_id='normal-collector-cron-2756' 주입."""
    launch_spy = make_launch_spy()

    rec, _ = _run_adopted(
        tmp_path,
        callback_launch_fn=launch_spy,
        owner_gate_fn=gate_pass,
        callback_owner_key=ANU_KEY,
        callback_executor_key=EXECUTOR_SELF_KEY,
        callback_chat_id="chat-2756",
        callback_at="10m",
    )

    assert len(launch_spy.captured) == 1
    req = launch_spy.captured[0]
    assert getattr(req, "cron_id", None) == "normal-collector-cron-2756", (
        f"request.cron_id={getattr(req, 'cron_id', None)!r} != 'normal-collector-cron-2756' "
        "(Medium-1 cron_id 일관성 미반영)"
    )
    # PASS request 이므로 callback_launch_verdict == BRIDGE_FIRED
    assert rec.callback_launch_verdict == drv.BRIDGE_FIRED


def test_bridge_move_err_reprocess_no_duplicate(tmp_path, monkeypatch):
    """★ Medium-2: _move_processed 실패(move_err) 시 result inbox 보존 + 재처리 시
    dedupe 가 callback duplicate 0 보장. callback 은 write 성공 후·move 전에 fire 됨.
    """
    import hashlib

    # _move_processed 가 항상 에러 문자열 반환 → move 실패 시뮬레이션
    monkeypatch.setattr(
        drv, "_move_processed",
        lambda *a, **k: "processed move 실패: simulated-2756",
    )

    dedupe_seen: set = set()
    launch_spy = make_launch_spy()

    root1 = _make_dirs(tmp_path / "c1")
    _enable_activation(root1)
    p1 = _write_result(root1)
    _age(p1)
    sha = hashlib.sha256(p1.read_bytes()).hexdigest()
    dedupe_key = f"{TASK_ID}:{sha}"

    def _prov(result, result_path, **kwargs):
        return prov.ProvenanceResult(ok=True, source_result_sha256=sha)

    common = dict(
        root=str(root1),
        sleep_fn=_NO_SLEEP,
        provenance_path_enabled=True,
        capability_reader=_cap_reader(),
        provenance_fn=_prov,
        callback_launch_fn=launch_spy,
        owner_gate_fn=gate_pass,
        callback_owner_key=ANU_KEY,
        callback_executor_key=EXECUTOR_SELF_KEY,
        callback_chat_id="chat-2756",
        callback_at="10m",
        callback_request_builder=None,
        callback_dedupe_seen=dedupe_seen,
        callback_dispatch_cron_id="dispatch-cron-2756",
        callback_normal_collector_cron_id="normal-collector-cron-2756",
        callback_no_fallback=True,
    )

    # ── cycle 1: write OK, move FAIL → fire 1회, result inbox 보존 ──
    rec1 = drv.process_one(
        str(p1), pickup_fn=make_pickup_mock(),
        collector_writer_fn=make_collector_writer_spy(), **common,
    )
    assert len(launch_spy.calls) == 1, f"cycle1 fire {len(launch_spy.calls)} != 1"
    assert rec1.callback_launch_verdict == drv.BRIDGE_FIRED
    assert rec1.error is not None and "processed move 실패" in rec1.error
    assert dedupe_key in dedupe_seen
    assert p1.exists(), "move 실패 시 result 는 inbox 보존되어야 함"

    # ── cycle 2: 동일 result 재처리(같은 sha + 공유 dedupe_seen) → DEDUPED, fire 추가 0 ──
    rec2 = drv.process_one(
        str(p1), pickup_fn=make_pickup_mock(),
        collector_writer_fn=make_collector_writer_spy(), **common,
    )
    assert len(launch_spy.calls) == 1, (
        f"재처리 후 fire {len(launch_spy.calls)} != 1 (callback duplicate 발생)"
    )
    assert rec2.callback_launch_verdict == drv.BRIDGE_DEDUPED, (
        f"cycle2 callback_launch_verdict={rec2.callback_launch_verdict!r} (BRIDGE_DEDUPED 이어야 함)"
    )


# ═════════════════════════════════════════════════════════════════════════════
# task-2756+2 신규 테스트: callback 실패 경로 전체 정합 (HIGH move 보존 + fix④⑤)
# ═════════════════════════════════════════════════════════════════════════════

def test_bridge_launch_exception_preserves_result_and_retries(tmp_path):
    """★ HIGH fix①: callback_launch_fn 예외 → result inbox 보존(move 0) + 다음 cycle
    재처리(재시도) 가능 + dedupe 미기록. _move_processed 는 mock 하지 않는다(실제 move 경로 검증).
    """
    import hashlib

    dedupe_seen: set = set()
    raising = make_raising_launch()

    root1 = _make_dirs(tmp_path / "x1")
    _enable_activation(root1)
    p1 = _write_result(root1)
    _age(p1)
    sha = hashlib.sha256(p1.read_bytes()).hexdigest()
    dedupe_key = f"{TASK_ID}:{sha}"

    def _prov(result, result_path, **kwargs):
        return prov.ProvenanceResult(ok=True, source_result_sha256=sha)

    common = dict(
        root=str(root1),
        sleep_fn=_NO_SLEEP,
        provenance_path_enabled=True,
        capability_reader=_cap_reader(),
        provenance_fn=_prov,
        callback_launch_fn=raising,
        owner_gate_fn=gate_pass,
        callback_owner_key=ANU_KEY,
        callback_executor_key=EXECUTOR_SELF_KEY,
        callback_chat_id="chat-2756",
        callback_at="10m",
        callback_request_builder=None,
        callback_dedupe_seen=dedupe_seen,
        callback_dispatch_cron_id="dispatch-cron-2756",
        callback_normal_collector_cron_id="normal-collector-cron-2756",
        callback_no_fallback=True,
    )

    # ── cycle 1: launch 예외 → crash 0, result inbox 보존(move 0), dedupe 미기록 ──
    rec1 = drv.process_one(
        str(p1), pickup_fn=make_pickup_mock(),
        collector_writer_fn=make_collector_writer_spy(), **common,
    )
    assert rec1 is not None  # driver crash 0
    assert rec1.callback_launch_verdict is None
    assert rec1.error is not None and "callback launch 실패" in rec1.error
    assert rec1.verdict == drv.VERDICT_PROVENANCE_ADOPTED
    assert len(raising.calls) == 1, f"cycle1 launch 시도 {len(raising.calls)} != 1"
    # ★ HIGH: callback 실패 시 result 는 inbox 보존(move 0)
    assert p1.exists(), "callback 실패 시 result 는 inbox 에 보존되어야 함(영구 누락 0)"
    # dedupe 미기록 → 재시도 가능
    assert dedupe_key not in dedupe_seen, "callback 실패 시 dedupe 미기록이어야 함"

    # ── cycle 2: 동일 result 재처리 → callback 재시도(fire 시도 2회), 여전히 보존 ──
    rec2 = drv.process_one(
        str(p1), pickup_fn=make_pickup_mock(),
        collector_writer_fn=make_collector_writer_spy(), **common,
    )
    assert len(raising.calls) == 2, (
        f"재처리 시 callback 재시도가 일어나야 함(누적 launch {len(raising.calls)} != 2)"
    )
    assert rec2.callback_launch_verdict is None
    assert p1.exists(), "재처리 후에도 result inbox 보존"


def test_bridge_owner_gate_exception_fail_closed(tmp_path):
    """★ fix④: owner_gate_fn 예외 → fail-closed(fire 0·crash 0), BRIDGE_OWNER_GATE_FAIL,
    error 라벨 관측, verdict 불변, result inbox 보존(bridge_err → move 0).
    """
    import hashlib

    def gate_raises(result_path, **kw):
        raise RuntimeError("gate-boom-2756")

    dedupe_seen: set = set()
    launch_spy = make_launch_spy()

    root1 = _make_dirs(tmp_path / "g1")
    _enable_activation(root1)
    p1 = _write_result(root1)
    _age(p1)
    sha = hashlib.sha256(p1.read_bytes()).hexdigest()

    def _prov(result, result_path, **kwargs):
        return prov.ProvenanceResult(ok=True, source_result_sha256=sha)

    rec = drv.process_one(
        str(p1),
        root=str(root1),
        pickup_fn=make_pickup_mock(),
        sleep_fn=_NO_SLEEP,
        provenance_path_enabled=True,
        capability_reader=_cap_reader(),
        provenance_fn=_prov,
        collector_writer_fn=make_collector_writer_spy(),
        callback_launch_fn=launch_spy,
        owner_gate_fn=gate_raises,  # 예외 발생 gate
        callback_owner_key=ANU_KEY,
        callback_executor_key=EXECUTOR_SELF_KEY,
        callback_chat_id="chat-2756",
        callback_at="10m",
        callback_request_builder=None,
        callback_dedupe_seen=dedupe_seen,
        callback_dispatch_cron_id="dispatch-cron-2756",
        callback_normal_collector_cron_id="normal-collector-cron-2756",
        callback_no_fallback=True,
    )

    assert rec is not None  # crash 0
    assert len(launch_spy.calls) == 0, f"gate 예외 시 fire {len(launch_spy.calls)} != 0"
    assert rec.callback_launch_verdict == drv.BRIDGE_OWNER_GATE_FAIL, (
        f"callback_launch_verdict={rec.callback_launch_verdict!r}"
    )
    assert rec.error is not None and "owner_gate 예외" in rec.error
    assert rec.verdict == drv.VERDICT_PROVENANCE_ADOPTED
    # bridge_err(예외) → move 0 → result 보존
    assert p1.exists(), "gate 예외(bridge_err) 시 result inbox 보존"
    # dedupe 미기록
    assert len(dedupe_seen) == 0


def test_bridge_dedupe_recorded_only_on_success(tmp_path):
    """★ fix②: callback launch 성공(BRIDGE_FIRED) 시에만 dedupe 기록. 실패 경로
    (gate FAIL / owner FAIL)에서는 dedupe 미기록.
    """
    # ── 성공: dedupe 기록 ──
    dedupe_ok: set = set()
    launch_ok = make_launch_spy()
    rec_ok, sha_ok = _run_adopted(
        tmp_path / "ok",
        callback_launch_fn=launch_ok,
        owner_gate_fn=gate_pass,
        callback_owner_key=ANU_KEY,
        callback_executor_key=EXECUTOR_SELF_KEY,
        callback_chat_id="chat-2756",
        callback_at="10m",
        callback_dedupe_seen=dedupe_ok,
    )
    assert rec_ok.callback_launch_verdict == drv.BRIDGE_FIRED
    assert f"{TASK_ID}:{sha_ok}" in dedupe_ok, "성공 시 dedupe 기록되어야 함"

    # ── gate FAIL: dedupe 미기록 ──
    dedupe_fail: set = set()
    launch_fail = make_launch_spy()
    rec_fail, _ = _run_adopted(
        tmp_path / "fail",
        callback_launch_fn=launch_fail,
        owner_gate_fn=gate_fail,
        callback_owner_key=ANU_KEY,
        callback_executor_key=EXECUTOR_SELF_KEY,
        callback_chat_id="chat-2756",
        callback_at="10m",
        callback_dedupe_seen=dedupe_fail,
    )
    assert rec_fail.callback_launch_verdict == drv.BRIDGE_OWNER_GATE_FAIL
    assert len(dedupe_fail) == 0, "gate FAIL 시 dedupe 미기록이어야 함"

    # ── owner_key FAIL(비 ANU): dedupe 미기록 ──
    dedupe_owner: set = set()
    launch_owner = make_launch_spy()
    rec_owner, _ = _run_adopted(
        tmp_path / "owner",
        callback_launch_fn=launch_owner,
        owner_gate_fn=gate_pass,
        callback_owner_key=NON_ANU_KEY,
        callback_executor_key=EXECUTOR_SELF_KEY,
        callback_chat_id="chat-2756",
        callback_at="10m",
        callback_dedupe_seen=dedupe_owner,
    )
    assert rec_owner.callback_launch_verdict == drv.BRIDGE_CALLBACK_OWNER_FAIL
    assert len(dedupe_owner) == 0, "owner FAIL 시 dedupe 미기록이어야 함"


def test_bounded_dedupe_set_evicts_oldest(tmp_path):
    """★ fix⑤: _BoundedDedupeSet 가 maxlen 초과 시 가장 오래된 키부터 evict(무한 증가 0).
    build_callback_dedupe_seen 헬퍼로 생성. bridge 와 동일 인터페이스(in, add) 동작 확인.
    """
    s = drv.build_callback_dedupe_seen(maxlen=3)
    s.add("k1")
    s.add("k2")
    s.add("k3")
    assert "k1" in s and "k2" in s and "k3" in s
    assert len(s) == 3
    # maxlen 초과 → 가장 오래된 k1 evict
    s.add("k4")
    assert len(s) == 3, f"maxlen 상한 위반: len={len(s)}"
    assert "k1" not in s, "가장 오래된 키(k1)가 evict 되어야 함"
    assert "k4" in s and "k3" in s and "k2" in s
    # 다수 추가해도 상한 유지(무한 증가 0)
    for i in range(100):
        s.add(f"bulk-{i}")
    assert len(s) == 3, f"대량 추가 후에도 상한 유지되어야 함: len={len(s)}"

    # bridge 가 이 bounded set 을 그대로 dedupe 로 사용 가능한지(in/add 인터페이스) 통합 확인
    bounded = drv.build_callback_dedupe_seen(maxlen=5)
    launch_spy = make_launch_spy()
    rec, sha = _run_adopted(
        tmp_path / "bounded",
        callback_launch_fn=launch_spy,
        owner_gate_fn=gate_pass,
        callback_owner_key=ANU_KEY,
        callback_executor_key=EXECUTOR_SELF_KEY,
        callback_chat_id="chat-2756",
        callback_at="10m",
        callback_dedupe_seen=bounded,
    )
    assert rec.callback_launch_verdict == drv.BRIDGE_FIRED
    assert f"{TASK_ID}:{sha}" in bounded