# -*- coding: utf-8 -*-
"""task-2732 — P0B_MINIMUM_PICKUP reporting path regression.

driver(scan_once) 수거 verdict → 회장 chat PASS/HOLD/CALLBACK_MISSING 자동 보고.
네트워크 0, 전부 fake sender/tmpdir(isolated). canonical events(memory/events) 0 touch.
ANU/봇 key literal 절대 노출 0 — 보고는 sendfile only(key 불요), redaction helper 검증.

커버리지(완료조건):
  ①/② verdict → PASS/HOLD/CALLBACK_MISSING 매핑(단위) + NOOP_*/매핑 밖 → 보고 안 함.
  ③ key 노출 0: 모듈 소스에 키 literal 0 · redact_argv 마스킹 · 보고 argv --cron/--key 0.
  ④ launcher_fn=None 유지(wake 0): build_launcher_fn 기본 None · main → scan_once launcher_fn None
     · reporter 가 launch_wake/cron prompt 호출 0(소스/sender argv).
  ⑤ 중복 보고 0: 동일 task_id 2회 → 전송 1회(dedupe ledger).
  ⑥ canonical write 0: 보고 경로가 memory/events 변형 0(isolated tmp root).
"""
from __future__ import annotations

import importlib.util
import json
import os
import sys
import tempfile
import unittest
from datetime import datetime, timezone
from pathlib import Path

_ROOT = Path(__file__).resolve().parent.parent
if str(_ROOT) not in sys.path:
    sys.path.insert(0, str(_ROOT))


def _load(modname: str, relpath: str):
    if modname in sys.modules:
        return sys.modules[modname]
    spec = importlib.util.spec_from_file_location(modname, _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


# 의존 실 로드 (closeout 테스트 패턴 동일)
_load("dispatch.callback_owner_enforcer", "dispatch/callback_owner_enforcer.py")
_load("dispatch.normal_fallback_callback_helper",
      "dispatch/normal_fallback_callback_helper.py")
_load("dispatch.anu_owned_callback_enforcement",
      "dispatch/anu_owned_callback_enforcement.py")
_load("dispatch.anu_collector_result", "dispatch/anu_collector_result.py")
TR = _load("dispatch.anu_terminal_relay", "dispatch/anu_terminal_relay.py")
_load("dispatch.anu_result_pickup_runner", "dispatch/anu_result_pickup_runner.py")
_load("dispatch.anu_pickup_wake_launcher", "dispatch/anu_pickup_wake_launcher.py")
# task-2731: main() lazy-import 대상(callback launch 강제 fn) — 사전 로드(테스트 결선).
_load("dispatch.anu_callback_launch_audit", "dispatch/anu_callback_launch_audit.py")
R = _load("dispatch.anu_pickup_reporter", "dispatch/anu_pickup_reporter.py")
DRV = _load("dispatch.anu_pickup_driver", "dispatch/anu_pickup_driver.py")

_CLOCK = lambda: datetime(2026, 6, 11, 6, 0, 0, tzinfo=timezone.utc)


class _FakeSender:
    """전송 호출을 기록하는 fake sender. ok 반환 제어 가능."""

    def __init__(self, ok: bool = True):
        self.ok = ok
        self.calls = []  # [(file_path, line_contents)]

    def __call__(self, file_path):
        try:
            with open(file_path, "r", encoding="utf-8") as fh:
                contents = fh.read()
        except OSError:
            contents = ""
        self.calls.append((file_path, contents))
        return self.ok


# ── ①/② verdict → 보고 매핑 (단위) ───────────────────────────────────────────
class TestMapReport(unittest.TestCase):
    def test_closeout_done_pass(self):
        self.assertEqual(R.map_report("CLOSEOUT_DONE"), R.REPORT_PASS)

    def test_authoritative_pass(self):
        self.assertEqual(R.map_report("AUTHORITATIVE"), R.REPORT_PASS)
        self.assertEqual(R.map_report(R.VERDICT_AUTHORITATIVE), R.REPORT_PASS)

    def test_quarantine_hold(self):
        self.assertEqual(R.map_report("QUARANTINE"), R.REPORT_HOLD)

    def test_callback_schedule_created_violation(self):
        # CLOSEOUT_DONE 라도 callback_schedule_created=true → CALLBACK_MISSING 우선.
        self.assertEqual(
            R.map_report("CLOSEOUT_DONE", {"callback_schedule_created": True}),
            R.REPORT_CALLBACK_MISSING,
        )

    def test_callback_absent_signal(self):
        self.assertEqual(
            R.map_report("CLOSEOUT_DONE", {"callback_present": False}),
            R.REPORT_CALLBACK_MISSING,
        )

    def test_callback_ok_stays_pass(self):
        # 정상 contract(false) → PASS 유지(오인 0).
        self.assertEqual(
            R.map_report("CLOSEOUT_DONE", {"callback_schedule_created": False}),
            R.REPORT_PASS,
        )
        self.assertEqual(
            R.map_report("AUTHORITATIVE", {"callback_present": True}),
            R.REPORT_PASS,
        )

    def test_noop_not_reported(self):
        for v in ("NOOP_DISABLED", "NOOP_NOT_TARGET", "NOOP_NOT_READY",
                  "NOOP_LEGACY_SKIP", "NOOP_MAX_FILES_DEFER"):
            self.assertIsNone(R.map_report(v), v)

    def test_out_of_spec_verdicts_not_reported(self):
        # spec 매핑 밖(WAKE_BUILT/PICKUP_SKIP/FIRE_FAILED) → 보고 안 함.
        for v in ("WAKE_BUILT", "PICKUP_SKIP", "FIRE_FAILED", "", None):
            self.assertIsNone(R.map_report(v), repr(v))

    def test_driver_verdict_constants_align(self):
        # driver 의 verdict 상수와 reporter 매핑이 어긋나지 않음.
        self.assertEqual(R.map_report(DRV.VERDICT_CLOSEOUT_DONE), R.REPORT_PASS)
        self.assertEqual(R.map_report(DRV.VERDICT_QUARANTINE), R.REPORT_HOLD)
        self.assertIsNone(R.map_report(DRV.VERDICT_NOOP_DISABLED))


# ── 보고 한 줄 포맷 ───────────────────────────────────────────────────────────
class TestReportLine(unittest.TestCase):
    def test_line_format(self):
        line = R.build_report_line("task-2732", R.REPORT_PASS, "CLOSEOUT_DONE",
                                   "abcdef0123456789")
        self.assertEqual(line, "task-2732: PASS · CLOSEOUT_DONE · abcdef012345")

    def test_task_id_from_path(self):
        self.assertEqual(
            R.task_id_from_path("/x/memory/events/task-2732.result.json"), "task-2732")
        self.assertEqual(R.task_id_from_path("/x/foo.txt"), "")


# ── ③ key 노출 0 ─────────────────────────────────────────────────────────────
class TestKeyExposureZero(unittest.TestCase):
    def test_redact_argv_masks_key(self):
        argv = ["/usr/local/bin/cokacdir", "--cron", "msg", "--key", "SECRET123"]
        red = R.redact_argv(argv)
        self.assertNotIn("SECRET123", red)
        self.assertIn(R.REDACTED, red)

    def test_redact_argv_eq_form(self):
        red = R.redact_argv(["--key=SECRET123"])
        self.assertNotIn("SECRET123", "".join(red))

    def test_module_source_no_key_literal(self):
        src = (_ROOT / "dispatch" / "anu_pickup_reporter.py").read_text(encoding="utf-8")
        # 16-hex 봇/ANU key literal 패턴 0 (sealed loader/argv 어디에도).
        import re
        hits = re.findall(r"\b[0-9a-f]{16}\b", src)
        self.assertEqual(hits, [], f"key-like literal 발견: {hits}")

    def test_report_argv_is_sendfile_no_key(self):
        # 보고 argv = --sendfile only. --cron/--key 0 (wake 0·key 0).
        argv = TR.build_send_report_argv("/tmp/x.report.txt", R.ANU_CHAT_ID)
        self.assertEqual(argv[1], "--sendfile")
        self.assertNotIn("--key", argv)
        self.assertNotIn("--cron", argv)


# ── ④ launcher_fn=None 유지(wake 0) 회귀 ─────────────────────────────────────
class TestWakeZero(unittest.TestCase):
    def test_build_launcher_fn_default_none(self):
        # real-wake flag off(부재) → None (wake 0).
        self.assertIsNone(DRV.build_launcher_fn(
            "/nonexistent-root", real_wake_reader=lambda: None))
        self.assertIsNone(DRV.build_launcher_fn(
            "/nonexistent-root", real_wake_reader=lambda: "disabled"))

    def test_main_launch_zero_and_reports(self):
        # ANCHOR-1 (launch-0 posture): real-wake/ACTIVE flag 부재(tmp root)에서 main() 이
        # scan_once 에 넘기는 launcher_fn 은 None(build_launcher_fn flag OFF → None, wake 0)
        # 이고, 그 records 가 reporter 로 결선됨을 검증. canonical 0 touch(scan_once mock).
        with tempfile.TemporaryDirectory() as td:
            captured: dict = {}

            def fake_scan_once(root, *, legacy_cutoff=False, launcher_fn="UNSET",
                               callback_launch_fn="UNSET", **kw):
                captured["root"] = root
                captured["launcher_fn"] = launcher_fn
                captured["callback_launch_fn"] = callback_launch_fn
                return [{"verdict": "CLOSEOUT_DONE",
                         "result_path": os.path.join(td, "task-1.result.json")}]

            def fake_report_records(records, **kw):
                captured["records"] = records
                return []

            orig_root = getattr(DRV, "CANONICAL_ROOT")
            orig_scan = getattr(DRV, "scan_once")
            orig_report = getattr(DRV, "report_records")
            try:
                setattr(DRV, "CANONICAL_ROOT", td)  # flag 부재 → build_launcher_fn None
                setattr(DRV, "scan_once", fake_scan_once)
                setattr(DRV, "report_records", fake_report_records)
                rc = DRV.main()
            finally:
                setattr(DRV, "CANONICAL_ROOT", orig_root)
                setattr(DRV, "scan_once", orig_scan)
                setattr(DRV, "report_records", orig_report)
            self.assertEqual(rc, 0)
            # wake launch 0: real-wake flag OFF → launcher_fn None.
            self.assertIsNone(captured["launcher_fn"], "launcher_fn must be None (wake 0)")
            # 보고 결선: scan_once records 가 reporter 로 전달됨.
            self.assertEqual(len(captured["records"]), 1)

    def test_reporter_source_no_wake(self):
        src = (_ROOT / "dispatch" / "anu_pickup_reporter.py").read_text(encoding="utf-8")
        # reporter 는 launch_wake / cron prompt / relay 발사 import·호출 0.
        self.assertNotIn("launch_wake", src)
        self.assertNotIn("relay_to_anu", src)
        self.assertNotIn("--cron", src)


# ── report_one end-to-end + ⑤ dedupe + ⑥ canonical write 0 ───────────────────
class TestReportOne(unittest.TestCase):
    def _result_file(self, td, task_id, data):
        events = os.path.join(td, "memory", "events")
        os.makedirs(events, exist_ok=True)
        p = os.path.join(events, f"{task_id}.result.json")
        with open(p, "w", encoding="utf-8") as fh:
            json.dump(data, fh)
        return p

    def test_pass_sends_and_marks(self):
        with tempfile.TemporaryDirectory() as td:
            p = self._result_file(td, "task-2732",
                                  {"task_id": "task-2732", "sha256": "deadbeefcafe0000",
                                   "callback_schedule_created": False})
            sender = _FakeSender(ok=True)
            dec = R.report_one("CLOSEOUT_DONE", p, root=td, sender=sender, clock=_CLOCK)
            self.assertEqual(dec.decision, R.REPORT_PASS)
            self.assertTrue(dec.sent)
            self.assertEqual(len(sender.calls), 1)
            self.assertIn("task-2732: PASS", sender.calls[0][1])
            self.assertIn("deadbeefcafe", sender.calls[0][1])
            # ledger 기록됨
            self.assertTrue(R.already_reported("task-2732", root=td))

    def test_dedupe_no_double_report(self):
        with tempfile.TemporaryDirectory() as td:
            p = self._result_file(td, "task-2732",
                                  {"task_id": "task-2732"})
            sender = _FakeSender(ok=True)
            d1 = R.report_one("CLOSEOUT_DONE", p, root=td, sender=sender, clock=_CLOCK)
            d2 = R.report_one("CLOSEOUT_DONE", p, root=td, sender=sender, clock=_CLOCK)
            self.assertTrue(d1.sent)
            self.assertFalse(d2.sent)
            self.assertTrue(d2.deduped)
            self.assertEqual(len(sender.calls), 1, "중복 보고 0 — 전송 1회")

    def test_noop_not_sent(self):
        with tempfile.TemporaryDirectory() as td:
            sender = _FakeSender(ok=True)
            dec = R.report_one("NOOP_DISABLED", "", root=td, sender=sender)
            self.assertIsNone(dec.decision)
            self.assertEqual(len(sender.calls), 0)

    def test_sender_failure_not_marked(self):
        # 전송 실패 → ledger 미기록(다음 cycle 재시도 가능, 중복 0 유지).
        with tempfile.TemporaryDirectory() as td:
            p = self._result_file(td, "task-2732", {"task_id": "task-2732"})
            sender = _FakeSender(ok=False)
            dec = R.report_one("CLOSEOUT_DONE", p, root=td, sender=sender, clock=_CLOCK)
            self.assertFalse(dec.sent)
            self.assertFalse(R.already_reported("task-2732", root=td))

    def test_callback_missing_reported(self):
        with tempfile.TemporaryDirectory() as td:
            p = self._result_file(td, "task-99",
                                  {"task_id": "task-99", "callback_schedule_created": True})
            sender = _FakeSender(ok=True)
            dec = R.report_one("CLOSEOUT_DONE", p, root=td, sender=sender, clock=_CLOCK)
            self.assertEqual(dec.decision, R.REPORT_CALLBACK_MISSING)
            self.assertTrue(dec.sent)
            self.assertIn("CALLBACK_MISSING", sender.calls[0][1])

    def test_canonical_events_not_modified(self):
        # ⑥ 보고 경로가 memory/events 를 변형하지 않음(isolated state 만 기록).
        with tempfile.TemporaryDirectory() as td:
            events = os.path.join(td, "memory", "events")
            os.makedirs(events, exist_ok=True)
            before = set(os.listdir(events))
            records = [
                {"verdict": "CLOSEOUT_DONE", "result_path":
                 os.path.join(events, "task-1.result.json"), "sha256": "aa11bb22cc33dd44"},
                {"verdict": "QUARANTINE", "result_path":
                 os.path.join(events, "task-2.result.json")},
                {"verdict": "NOOP_DISABLED", "result_path": ""},
            ]
            sender = _FakeSender(ok=True)
            decs = R.report_records(records, root=td, sender=sender, clock=_CLOCK)
            after = set(os.listdir(events))
            self.assertEqual(before, after, "canonical events 변형 0")
            # 보고는 isolated p0b_state 에만
            self.assertTrue(os.path.isfile(os.path.join(td, R.REPORTER_LEDGER_REL)))
            sent = [d for d in decs if d.sent]
            self.assertEqual({d.decision for d in sent}, {R.REPORT_PASS, R.REPORT_HOLD})


# ── report_records 가 DriverRecord(dataclass) 도 처리 ─────────────────────────
class TestReportRecordsDataclass(unittest.TestCase):
    def test_driverrecord_objects(self):
        with tempfile.TemporaryDirectory() as td:
            rec = DRV.DriverRecord(
                ts="t", result_path=os.path.join(td, "task-7.result.json"),
                verdict=DRV.VERDICT_CLOSEOUT_DONE)
            sender = _FakeSender(ok=True)
            decs = R.report_records([rec], root=td, sender=sender, clock=_CLOCK)
            self.assertEqual(decs[0].decision, R.REPORT_PASS)
            self.assertTrue(decs[0].sent)


# ── (a) HIGH: _default_sender 성공 판정 의미론 ────────────────────────────────
class _FakeProc:
    """returncode 속성만 갖는 CompletedProcess 모사."""

    def __init__(self, returncode):
        self.returncode = returncode


class TestDefaultSenderSuccessSemantics(unittest.TestCase):
    """★ (a) HIGH: send 성공은 **실제 전송 성공(returncode==0 확인)** 인 경우에만 True.
    이전 ``getattr(res, "returncode", 0)`` 는 None/속성부재를 0 으로 오인 → 미전송을
    성공으로 기록하는 HIGH 결함이었다. send_report 를 monkeypatch 하여 의미론 고정."""

    def setUp(self):
        self._orig = R.send_report

    def tearDown(self):
        setattr(R, "send_report", self._orig)

    def test_returncode_zero_is_success(self):
        setattr(R, "send_report", lambda fp, *, chat_id: _FakeProc(0))
        self.assertTrue(R._default_sender("/tmp/x.report.txt", chat_id="c"))

    def test_returncode_nonzero_is_failure(self):
        setattr(R, "send_report", lambda fp, *, chat_id: _FakeProc(2))
        self.assertFalse(R._default_sender("/tmp/x.report.txt", chat_id="c"))

    def test_none_result_not_success(self):
        # send_report 가 None 반환(예: runner None) → 성공 간주 0.
        setattr(R, "send_report", lambda fp, *, chat_id: None)
        self.assertFalse(R._default_sender("/tmp/x.report.txt", chat_id="c"))

    def test_missing_returncode_attr_not_success(self):
        class _NoRC:
            pass

        setattr(R, "send_report", lambda fp, *, chat_id: _NoRC())
        self.assertFalse(R._default_sender("/tmp/x.report.txt", chat_id="c"))

    def test_non_int_returncode_not_success(self):
        # returncode 가 None/비정수(미확정) → 성공 간주 0.
        setattr(R, "send_report", lambda fp, *, chat_id: _FakeProc(None))
        self.assertFalse(R._default_sender("/tmp/x.report.txt", chat_id="c"))

    def test_exception_not_success(self):
        def _boom(fp, *, chat_id):
            raise RuntimeError("send blew up")

        setattr(R, "send_report", _boom)
        self.assertFalse(R._default_sender("/tmp/x.report.txt", chat_id="c"))


# ── ⑧ ANCHOR-3: ledger 성공 기록은 실제 send 성공 후에만 ─────────────────────
class TestLedgerOnlyOnRealSendSuccess(unittest.TestCase):
    """★ 검증 #8 / ANCHOR-3: dedupe ledger 성공 기록이 **실제 send 성공(returncode==0)**
    후에만 발생함을 default sender(주입 없이) 경로로 증명. None/예외/nonzero 반환 시
    ledger 미기록 → 다음 cycle 재시도 가능(중복 0)."""

    def setUp(self):
        self._orig = R.send_report

    def tearDown(self):
        setattr(R, "send_report", self._orig)

    def _result_file(self, td, task_id):
        events = os.path.join(td, "memory", "events")
        os.makedirs(events, exist_ok=True)
        p = os.path.join(events, f"{task_id}.result.json")
        with open(p, "w", encoding="utf-8") as fh:
            json.dump({"task_id": task_id}, fh)
        return p

    def test_real_send_success_marks_ledger(self):
        sent_files = []

        def _ok_send(fp, *, chat_id):
            sent_files.append(fp)
            return _FakeProc(0)

        with tempfile.TemporaryDirectory() as td:
            p = self._result_file(td, "task-2732")
            # default sender 경로(sender 미주입) — _default_sender → send_report.
            setattr(R, "send_report", _ok_send)
            dec = R.report_one("CLOSEOUT_DONE", p, root=td, clock=_CLOCK)
            self.assertTrue(dec.sent)
            self.assertEqual(len(sent_files), 1, "실제 send 1회 발생 후에만")
            self.assertTrue(R.already_reported("task-2732", root=td),
                            "실제 send 성공(returncode==0) 후에만 ledger 기록")

    def test_none_result_no_ledger_retriable(self):
        with tempfile.TemporaryDirectory() as td:
            p = self._result_file(td, "task-2732")
            # 미전송(계약 위반 반환: None) → 성공 간주 0.
            setattr(R, "send_report", lambda fp, *, chat_id: None)
            dec = R.report_one("CLOSEOUT_DONE", p, root=td, clock=_CLOCK)
            self.assertFalse(dec.sent, "None 반환은 성공 간주 0")
            self.assertFalse(R.already_reported("task-2732", root=td),
                             "ledger 미기록 — 다음 cycle 재시도 가능")

    def test_nonzero_returncode_no_ledger(self):
        with tempfile.TemporaryDirectory() as td:
            p = self._result_file(td, "task-2732")
            setattr(R, "send_report", lambda fp, *, chat_id: _FakeProc(1))
            dec = R.report_one("CLOSEOUT_DONE", p, root=td, clock=_CLOCK)
            self.assertFalse(dec.sent)
            self.assertFalse(R.already_reported("task-2732", root=td))

    def test_send_exception_no_ledger(self):
        def _boom(fp, *, chat_id):
            raise RuntimeError("network down")

        with tempfile.TemporaryDirectory() as td:
            p = self._result_file(td, "task-2732")
            setattr(R, "send_report", _boom)
            dec = R.report_one("CLOSEOUT_DONE", p, root=td, clock=_CLOCK)
            self.assertFalse(dec.sent)
            self.assertFalse(R.already_reported("task-2732", root=td))

    def test_no_send_no_dedupe_then_retry_succeeds(self):
        # 미전송(None) → 다음 cycle 에서 실제 성공 시 1회 전송·기록(누락 회피·중복 0).
        sent = []
        with tempfile.TemporaryDirectory() as td:
            p = self._result_file(td, "task-2732")
            setattr(R, "send_report", lambda fp, *, chat_id: None)
            d1 = R.report_one("CLOSEOUT_DONE", p, root=td, clock=_CLOCK)
            self.assertFalse(d1.sent)

            def _ok(fp, *, chat_id):
                sent.append(fp)
                return _FakeProc(0)

            setattr(R, "send_report", _ok)
            d2 = R.report_one("CLOSEOUT_DONE", p, root=td, clock=_CLOCK)
            self.assertTrue(d2.sent)
            self.assertEqual(len(sent), 1, "재시도 시 정확히 1회 전송")
            self.assertTrue(R.already_reported("task-2732", root=td))


# ── (c) MEDIUM: processed dir bounded scan (전체 무조건 listdir 회피) ──────────
class TestResultLoaderBoundedScan(unittest.TestCase):
    def test_direct_candidate_no_scan(self):
        # 원위치 존재 → 직접 적재(디렉토리 scan 불필요).
        with tempfile.TemporaryDirectory() as td:
            events = os.path.join(td, "memory", "events")
            os.makedirs(events, exist_ok=True)
            p = os.path.join(events, "task-5.result.json")
            with open(p, "w", encoding="utf-8") as fh:
                json.dump({"task_id": "task-5", "x": 1}, fh)
            data = R._default_result_loader(p, root=td)
            self.assertEqual(data.get("x"), 1)

    def test_processed_basename_direct_match(self):
        # 원위치 부재 → processed/<basename> 직접 후보로 적재(listdir 0).
        with tempfile.TemporaryDirectory() as td:
            pdir = os.path.join(td, "memory", "p0b_state", "processed")
            os.makedirs(pdir, exist_ok=True)
            base = "task-6.result.json"
            with open(os.path.join(pdir, base), "w", encoding="utf-8") as fh:
                json.dump({"task_id": "task-6", "moved": True}, fh)
            rp = os.path.join(td, "memory", "events", base)
            data = R._default_result_loader(rp, root=td)
            self.assertTrue(data.get("moved"))

    def test_processed_suffix_variant_bounded(self):
        # 충돌-회피 suffix(.<ms>) 변형은 bounded scan 으로 탐색. 노이즈 다수에도 동작.
        with tempfile.TemporaryDirectory() as td:
            pdir = os.path.join(td, "memory", "p0b_state", "processed")
            os.makedirs(pdir, exist_ok=True)
            base = "task-7.result.json"
            for i in range(60):  # 매칭 안 되는 노이즈
                with open(os.path.join(pdir, f"task-other-{i}.result.json"), "w") as fh:
                    fh.write("{}")
            with open(os.path.join(pdir, base + ".1700000000000"), "w",
                      encoding="utf-8") as fh:
                json.dump({"task_id": "task-7", "variant": True}, fh)
            rp = os.path.join(td, "memory", "events", base)
            data = R._default_result_loader(rp, root=td)
            self.assertTrue(data.get("variant"))

    def test_missing_returns_empty_dict(self):
        # 전부 부재 → {} (보수적: callback 판정 불가 → PASS 보존).
        with tempfile.TemporaryDirectory() as td:
            rp = os.path.join(td, "memory", "events", "task-404.result.json")
            self.assertEqual(R._default_result_loader(rp, root=td), {})
            self.assertEqual(R._default_result_loader("", root=td), {})

    def test_scan_limit_is_positive_int(self):
        self.assertIsInstance(R.PROCESSED_SCAN_LIMIT, int)
        self.assertGreater(R.PROCESSED_SCAN_LIMIT, 0)

    def test_no_unconditional_listdir_in_loader(self):
        # ★ (c): processed dir 전체 무조건 os.listdir() 호출 제거(소스 가드).
        #   실제 호출(``os.listdir(``)만 검사 — 주석/docstring 의 literal 언급은 무관.
        src = (_ROOT / "dispatch" / "anu_pickup_reporter.py").read_text(encoding="utf-8")
        self.assertNotIn("os.listdir(", src,
                         "reporter 는 무조건 listdir() 대신 직접후보+bounded scandir 사용")
        self.assertIn("os.scandir(", src, "bounded scandir 사용 확인")


# ── (b) MEDIUM: driver 보고 예외 silent pass 금지(관찰 가능) ──────────────────
class TestDriverReportingFailureObservable(unittest.TestCase):
    def test_driver_main_no_silent_pass_on_reporting_error(self):
        # report_records 가 예외를 던져도 driver 는 crash 0(return 0) 이되,
        # silent pass 금지: stderr 로 관찰 가능하게 기록. 재시도/wake/dispatch 0.
        import io
        import contextlib

        with tempfile.TemporaryDirectory() as td:
            def fake_scan_once(root, *, legacy_cutoff=False, launcher_fn="UNSET",
                               callback_launch_fn="UNSET", **kw):
                return [{"verdict": "CLOSEOUT_DONE",
                         "result_path": os.path.join(td, "task-1.result.json")}]

            def boom_report(records, **kw):
                raise RuntimeError("reporter exploded")

            orig_root = getattr(DRV, "CANONICAL_ROOT")
            orig_scan = getattr(DRV, "scan_once")
            orig_report = getattr(DRV, "report_records")
            buf = io.StringIO()
            try:
                setattr(DRV, "CANONICAL_ROOT", td)
                setattr(DRV, "scan_once", fake_scan_once)
                setattr(DRV, "report_records", boom_report)
                with contextlib.redirect_stderr(buf):
                    rc = DRV.main()
            finally:
                setattr(DRV, "CANONICAL_ROOT", orig_root)
                setattr(DRV, "scan_once", orig_scan)
                setattr(DRV, "report_records", orig_report)
            self.assertEqual(rc, 0, "보고 예외는 비치명 — driver crash 0")
            err = buf.getvalue()
            self.assertIn("reporting failure", err,
                          "silent pass 금지 — reporting failure 관찰 가능")
            self.assertIn("reporter exploded", err)

    def test_driver_source_no_silent_pass(self):
        src = (_ROOT / "dispatch" / "anu_pickup_driver.py").read_text(encoding="utf-8")
        # report_records 호출 직후 except 가 bare pass 가 아님(stderr 기록 존재).
        self.assertIn("reporting failure", src)
        # 재시도 폭주/wake/후속 dispatch 발사 0(보고 예외 핸들러에서).
        self.assertNotIn("launch_wake(", src.split("def main(")[-1])


if __name__ == "__main__":
    unittest.main()