# -*- coding: utf-8 -*-
"""task-2775 limited activation guard micro-PR 회귀 테스트.

헤임달(QA) — limited_activation_bound_gate 3경계(kill / N / T) 경계 조건 전수 검증.
대상 심볼: dispatch.anu_pickup_driver

테스트 케이스 목록 (21개):
  1.  test_kill_switch_immediate_noop_no_flag_write
  2.  test_kill_switch_via_injected_fn
  3.  test_n_exceeded_writes_disabled_atomic            [epoch-scope 갱신]
  4.  test_n_below_threshold_passes
  5.  test_n_count_ignores_non_processed_outcomes       [epoch-scope 갱신]
  6.  test_t_expired_writes_disabled
  7.  test_t_within_window_passes
  8.  test_no_epoch_no_trip
  9.  test_order_kill_precedes_n
  10. test_atomic_disabled_write_is_complete
  11. test_flag_writer_injection_used                   [epoch-scope 갱신]
  12. test_clean_state_returns_none
  13. test_epoch_scope_before_epoch_not_counted         [신규]
  14. test_epoch_scope_after_epoch_counted_and_trips    [신규]
  15. test_reactivation_safe_old_window_ignored         [신규]
  16. test_processed_at_absent_excluded                 [신규]
  17. test_processed_at_unparseable_excluded            [신규]
  18. test_timezone_aware_kst_offset_preserved          [신규]
  19. test_epoch_absent_active_fail_closed               [신규]
  20. test_epoch_absent_inactive_returns_none            [신규]
  21. test_main_t_expired_smoke                          [신규]
"""
from __future__ import annotations

import importlib.util as _ilu
import json
import os
import sys
import time
from datetime import datetime, timedelta, timezone
from pathlib import Path

import pytest

# ── import 부트스트랩 ──────────────────────────────────────────────────────────
# conftest.py 가 worktree root 를 sys.path[0] 에 보장하지만,
# 단독 실행/순서 변동 대비로 한 번 더 보강한다.
_ROOT = Path(__file__).resolve().parents[2]
if str(_ROOT) not in sys.path:
    sys.path.insert(0, str(_ROOT))

# tests/dispatch (테스트용 빈 패키지)가 실제 dispatch 패키지를 가리는 것을 방지:
# 실제 dispatch 패키지를 파일 위치로 직접 로드해 sys.modules 에 고정한다.
_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

ACTIVATION_DISABLED = _drv.ACTIVATION_DISABLED
ACTIVATION_FLAG_REL = _drv.ACTIVATION_FLAG_REL
ACTIVATION_EPOCH_REL = _drv.ACTIVATION_EPOCH_REL
KILL_SWITCH_REL = _drv.KILL_SWITCH_REL
LIVE_LEDGER_REL = _drv.LIVE_LEDGER_REL
LIMITED_MAX_PICKUPS_N = _drv.LIMITED_MAX_PICKUPS_N
VERDICT_LIVE_PROCESSED = _drv.VERDICT_LIVE_PROCESSED
VERDICT_NOOP_KILL_SWITCH = _drv.VERDICT_NOOP_KILL_SWITCH
VERDICT_NOOP_N_EXCEEDED = _drv.VERDICT_NOOP_N_EXCEEDED
VERDICT_NOOP_T_EXPIRED = _drv.VERDICT_NOOP_T_EXPIRED
VERDICT_NOOP_ACTIVE_EPOCH_MISSING = _drv.VERDICT_NOOP_ACTIVE_EPOCH_MISSING
VERDICT_NOOP_ACTIVE_EPOCH_INVALID = _drv.VERDICT_NOOP_ACTIVE_EPOCH_INVALID
_atomic_write_flag_disabled = _drv._atomic_write_flag_disabled
_count_active_window_pickups = _drv._count_active_window_pickups
_reverse_line_iter = _drv._reverse_line_iter
_parse_processed_at_to_unix = _drv._parse_processed_at_to_unix
limited_activation_bound_gate = _drv.limited_activation_bound_gate

# KST timezone (+09:00)
_KST = timezone(timedelta(hours=9))


# ── 공통 헬퍼 ─────────────────────────────────────────────────────────────────

def _make_ledger(tmp_path, *, live_count: int, other_count: int = 0) -> str:
    """LIVE_LEDGER_REL 경로에 ledger jsonl 파일 생성.

    live_count 만큼 outcome==LIVE_PROCESSED 항목을,
    other_count 만큼 outcome==OTHER_OUTCOME 항목을 추가한다.
    processed_at 필드 없음 — 기존 테스트(4, 8, 9, 12) 호환 유지.
    """
    ledger_path = tmp_path / LIVE_LEDGER_REL
    ledger_path.parent.mkdir(parents=True, exist_ok=True)
    with open(ledger_path, "w", encoding="utf-8") as fh:
        for i in range(live_count):
            fh.write(json.dumps({"outcome": VERDICT_LIVE_PROCESSED, "task_id": f"task-{i}"}) + "\n")
        for i in range(other_count):
            fh.write(json.dumps({"outcome": "OTHER_OUTCOME", "task_id": f"other-{i}"}) + "\n")
    return str(ledger_path)


def _make_ledger_scoped(
    tmp_path,
    *,
    after_epoch_count: int,
    before_epoch_count: int = 0,
    epoch_unix: float,
    other_count: int = 0,
) -> str:
    """epoch-scope 대응 ledger 생성 헬퍼.

    - after_epoch_count 만큼: outcome==LIVE_PROCESSED + processed_at(KST, epoch_unix + 1초 이후)
    - before_epoch_count 만큼: outcome==LIVE_PROCESSED + processed_at(KST, epoch_unix - 1초 이전)
    - other_count 만큼: outcome==OTHER_OUTCOME (processed_at 포함)
    """
    ledger_path = tmp_path / LIVE_LEDGER_REL
    ledger_path.parent.mkdir(parents=True, exist_ok=True)

    def _to_kst_iso(unix_val: float) -> str:
        dt = datetime.fromtimestamp(unix_val, tz=_KST)
        return dt.isoformat()

    with open(ledger_path, "w", encoding="utf-8") as fh:
        # epoch 이후 LIVE_PROCESSED (카운트 대상)
        for i in range(after_epoch_count):
            ts_unix = epoch_unix + 1.0 + i
            fh.write(json.dumps({
                "outcome": VERDICT_LIVE_PROCESSED,
                "task_id": f"task-after-{i}",
                "processed_at": _to_kst_iso(ts_unix),
            }) + "\n")
        # epoch 이전 LIVE_PROCESSED (카운트 비대상)
        for i in range(before_epoch_count):
            ts_unix = epoch_unix - 1.0 - i
            fh.write(json.dumps({
                "outcome": VERDICT_LIVE_PROCESSED,
                "task_id": f"task-before-{i}",
                "processed_at": _to_kst_iso(ts_unix),
            }) + "\n")
        # 기타 outcome
        for i in range(other_count):
            ts_unix = epoch_unix + 1.0 + i
            fh.write(json.dumps({
                "outcome": "OTHER_OUTCOME",
                "task_id": f"other-{i}",
                "processed_at": _to_kst_iso(ts_unix),
            }) + "\n")
    return str(ledger_path)


def _flag_path(tmp_path) -> str:
    return str(tmp_path / ACTIVATION_FLAG_REL)


def _kill_path(tmp_path) -> str:
    return str(tmp_path / KILL_SWITCH_REL)


# ── 테스트 1: kill switch — 즉시 NOOP, flag write 없음 ─────────────────────────
def test_kill_switch_immediate_noop_no_flag_write(tmp_path):
    """tmp_path 에 p0b_kill 터치 → NOOP_KILL_SWITCH, p0b_driver_enabled 미생성."""
    kill_file = tmp_path / KILL_SWITCH_REL
    kill_file.parent.mkdir(parents=True, exist_ok=True)
    kill_file.touch()

    rec = limited_activation_bound_gate(str(tmp_path))

    assert rec is not None, "kill switch 존재 시 None 이 아닌 DriverRecord 반환이어야 함"
    assert rec.verdict == VERDICT_NOOP_KILL_SWITCH, (
        f"kill switch verdict 불일치: {rec.verdict!r}"
    )

    # flag write 0 검증: p0b_driver_enabled 파일 생성되지 않아야 함
    flag_file = tmp_path / ACTIVATION_FLAG_REL
    assert not flag_file.exists(), (
        "NOOP_KILL_SWITCH 경로에서 p0b_driver_enabled 를 생성하면 안 됨 (flag write 0)"
    )


# ── 테스트 2: kill_exists_fn 주입으로 kill switch 트립 ────────────────────────
def test_kill_switch_via_injected_fn(tmp_path):
    """kill_exists_fn=lambda:True 주입 → NOOP_KILL_SWITCH."""
    rec = limited_activation_bound_gate(str(tmp_path), kill_exists_fn=lambda: True)

    assert rec is not None
    assert rec.verdict == VERDICT_NOOP_KILL_SWITCH, (
        f"injected kill_exists_fn=True → NOOP_KILL_SWITCH 기대, 실제: {rec.verdict!r}"
    )


# ── 테스트 3: N 초과 — disabled atomic write 검증 (epoch-scope 갱신) ───────────
def test_n_exceeded_writes_disabled_atomic(tmp_path):
    """epoch_reader 주입 + ledger epoch 이후 LIVE_PROCESSED 3건 → NOOP_N_EXCEEDED + disabled write.

    epoch-scope 경로를 실제로 타야 함: epoch_reader 주입, ledger에 processed_at 포함,
    ledger_count_fn 미사용.
    """
    epoch_unix = time.time() - 30 * 60  # 30분 전 epoch (T=120 이내)

    _make_ledger_scoped(
        tmp_path,
        after_epoch_count=LIMITED_MAX_PICKUPS_N,  # epoch 이후 3건
        epoch_unix=epoch_unix,
    )

    rec = limited_activation_bound_gate(
        str(tmp_path),
        epoch_reader=lambda: str(epoch_unix),
        now_fn=lambda: time.time(),
    )

    assert rec is not None
    assert rec.verdict == VERDICT_NOOP_N_EXCEEDED, (
        f"N 초과 verdict 불일치: {rec.verdict!r}"
    )

    flag_file = tmp_path / ACTIVATION_FLAG_REL
    assert flag_file.exists(), "N 초과 시 p0b_driver_enabled 파일이 생성되어야 함"
    first_line = flag_file.read_text(encoding="utf-8").splitlines()[0].strip()
    assert first_line == ACTIVATION_DISABLED, (
        f"p0b_driver_enabled 첫줄 == 'disabled' 기대, 실제: {first_line!r}"
    )


# ── 테스트 4: N 미만 — None 반환, side-effect 없음 ────────────────────────────
def test_n_below_threshold_passes(tmp_path):
    """ledger LIVE_PROCESSED 2줄 (<3) → None 반환, p0b_driver_enabled 미생성."""
    _make_ledger(tmp_path, live_count=LIMITED_MAX_PICKUPS_N - 1)

    rec = limited_activation_bound_gate(str(tmp_path))

    assert rec is None, f"N 미만이면 None 반환 기대, 실제: {rec!r}"

    flag_file = tmp_path / ACTIVATION_FLAG_REL
    assert not flag_file.exists(), "N 미만 시 p0b_driver_enabled 생성되어서는 안 됨 (side-effect 0)"


# ── 테스트 5: outcome != LIVE_PROCESSED 무시, LIVE_PROCESSED 1개만 카운트 (epoch-scope 갱신) ──
def test_n_count_ignores_non_processed_outcomes(tmp_path):
    """epoch 이후 LIVE_PROCESSED 1건 + 다른 outcome 5건 → 누계 1 → None 반환."""
    epoch_unix = time.time() - 30 * 60  # 30분 전 epoch

    _make_ledger_scoped(
        tmp_path,
        after_epoch_count=1,      # LIVE_PROCESSED 1건
        epoch_unix=epoch_unix,
        other_count=5,            # OTHER_OUTCOME 5건
    )

    rec = limited_activation_bound_gate(
        str(tmp_path),
        epoch_reader=lambda: str(epoch_unix),
        now_fn=lambda: time.time(),
    )

    assert rec is None, (
        f"LIVE_PROCESSED 1개(< n_max=3)면 None 기대, 실제: {rec!r}"
    )


# ── 테스트 6: T 만료 — disabled write 검증 ────────────────────────────────────
def test_t_expired_writes_disabled(tmp_path):
    """epoch = now - 121분 → NOOP_T_EXPIRED, disabled write 확인."""
    now_unix = time.time()
    epoch_str = str(now_unix - 121 * 60)

    rec = limited_activation_bound_gate(
        str(tmp_path),
        epoch_reader=lambda: epoch_str,
        now_fn=lambda: now_unix,
    )

    assert rec is not None
    assert rec.verdict == VERDICT_NOOP_T_EXPIRED, (
        f"T 만료 verdict 불일치: {rec.verdict!r}"
    )

    flag_file = tmp_path / ACTIVATION_FLAG_REL
    assert flag_file.exists(), "T 만료 시 p0b_driver_enabled 파일이 생성되어야 함"
    first_line = flag_file.read_text(encoding="utf-8").splitlines()[0].strip()
    assert first_line == ACTIVATION_DISABLED, (
        f"T 만료 후 p0b_driver_enabled 첫줄 == 'disabled' 기대, 실제: {first_line!r}"
    )


# ── 테스트 7: T 이내 (1시간 전) — None, side-effect 없음 ─────────────────────
def test_t_within_window_passes(tmp_path):
    """epoch = now - 60분 (T=120 이내) → None, side-effect 0."""
    now_unix = time.time()
    epoch_str = str(now_unix - 60 * 60)  # 1시간 전, T=120 이내

    rec = limited_activation_bound_gate(
        str(tmp_path),
        epoch_reader=lambda: epoch_str,
        now_fn=lambda: now_unix,
    )

    assert rec is None, f"T 이내면 None 기대, 실제: {rec!r}"

    flag_file = tmp_path / ACTIVATION_FLAG_REL
    assert not flag_file.exists(), "T 이내 시 p0b_driver_enabled 생성되어서는 안 됨 (side-effect 0)"


# ── 테스트 8: epoch None → T 미발동 → None ────────────────────────────────────
def test_no_epoch_no_trip(tmp_path):
    """epoch_reader=lambda:None + is_activated_fn=lambda:False → None 반환."""
    rec = limited_activation_bound_gate(
        str(tmp_path),
        epoch_reader=lambda: None,
        is_activated_fn=lambda: False,
    )

    assert rec is None, f"epoch 없고 not active → None 기대, 실제: {rec!r}"


# ── 테스트 9: kill 우선순위 (kill AND N 동시 초과) ────────────────────────────
def test_order_kill_precedes_n(tmp_path):
    """kill 존재 AND ledger 3줄(N 초과) 동시 → NOOP_KILL_SWITCH(kill 우선), flag write 0."""
    # kill switch 파일 생성
    kill_file = tmp_path / KILL_SWITCH_REL
    kill_file.parent.mkdir(parents=True, exist_ok=True)
    kill_file.touch()

    # ledger N 초과 상태 (processed_at 없는 entry — kill 이 먼저라 N 평가 안 됨)
    _make_ledger(tmp_path, live_count=LIMITED_MAX_PICKUPS_N)

    rec = limited_activation_bound_gate(str(tmp_path))

    assert rec is not None
    assert rec.verdict == VERDICT_NOOP_KILL_SWITCH, (
        f"kill+N 동시 → kill 우선(NOOP_KILL_SWITCH) 기대, 실제: {rec.verdict!r}"
    )

    # kill 경로에서는 flag write 없음
    flag_file = tmp_path / ACTIVATION_FLAG_REL
    assert not flag_file.exists(), (
        "NOOP_KILL_SWITCH(kill 우선) 경로에서 p0b_driver_enabled 생성되어서는 안 됨"
    )


# ── 테스트 10: _atomic_write_flag_disabled 직접 호출 — 완전한 파일 쓰기 ────────
def test_atomic_disabled_write_is_complete(tmp_path):
    """_atomic_write_flag_disabled 직접 호출 → 반환 None(성공), 내용 == 'disabled\\n'."""
    err = _atomic_write_flag_disabled(str(tmp_path))

    assert err is None, f"atomic write 실패(err != None): {err!r}"

    flag_file = tmp_path / ACTIVATION_FLAG_REL
    assert flag_file.exists(), "p0b_driver_enabled 파일이 생성되어야 함"

    content = flag_file.read_text(encoding="utf-8")
    assert content == ACTIVATION_DISABLED + "\n", (
        f"파일 내용 == 'disabled\\n' 기대, 실제: {content!r}"
    )


# ── 테스트 11: flag_writer 주입 — N 경로에서 주입 fn이 사용됨 (epoch-scope 갱신) ─
def test_flag_writer_injection_used(tmp_path):
    """epoch_reader 주입 + epoch 이후 LIVE_PROCESSED 3건 + flag_writer 주입 →
    N 경로에서 flag_writer 1회 호출 + NOOP_N_EXCEEDED.

    실제 epoch-scope 경로를 타야 함(ledger_count_fn 미사용).
    """
    calls: list[str] = []

    def _mock_flag_writer(val: str):
        calls.append(val)
        return None  # 성공 신호

    epoch_unix = time.time() - 30 * 60  # 30분 전 epoch (T=120 이내)

    _make_ledger_scoped(
        tmp_path,
        after_epoch_count=LIMITED_MAX_PICKUPS_N,  # epoch 이후 3건
        epoch_unix=epoch_unix,
    )

    rec = limited_activation_bound_gate(
        str(tmp_path),
        epoch_reader=lambda: str(epoch_unix),
        now_fn=lambda: time.time(),
        flag_writer=_mock_flag_writer,
    )

    assert rec is not None
    assert rec.verdict == VERDICT_NOOP_N_EXCEEDED, (
        f"N 초과 + flag_writer 주입 → NOOP_N_EXCEEDED 기대, 실제: {rec.verdict!r}"
    )

    assert len(calls) == 1, f"flag_writer 1회 호출 기대, 실제 호출 수: {len(calls)}"
    assert calls[0] == ACTIVATION_DISABLED, (
        f"flag_writer 인자 == ACTIVATION_DISABLED('{ACTIVATION_DISABLED}') 기대, 실제: {calls[0]!r}"
    )


# ── 테스트 12: 완전히 빈 상태 — None 반환 (default-OFF 회귀 가드) ───────────────
def test_clean_state_returns_none(tmp_path):
    """빈 tmp_path(kill 없음, ledger 없음, epoch 없음) → None 반환.

    이것이 핵심 회귀 가드: 아무 경계도 없으면 현행 경로(default-OFF/B 1-shot) 100% 보존.
    """
    rec = limited_activation_bound_gate(str(tmp_path))

    assert rec is None, (
        f"완전히 빈 상태(kill 없음·ledger 없음·epoch 없음)이면 None 반환 기대, 실제: {rec!r}"
    )

    # side-effect 전혀 없음 확인
    flag_file = tmp_path / ACTIVATION_FLAG_REL
    assert not flag_file.exists(), "빈 상태에서 p0b_driver_enabled 생성되어서는 안 됨 (side-effect 0)"


# ── 테스트 13: epoch 이전 entry는 카운트 0 ────────────────────────────────────
def test_epoch_scope_before_epoch_not_counted(tmp_path):
    """_count_active_window_pickups 직접 호출.
    ledger에 processed_at이 epoch E 이전인 LIVE_PROCESSED 3건만 → 반환 0.
    """
    epoch_unix = time.time()

    _make_ledger_scoped(
        tmp_path,
        after_epoch_count=0,
        before_epoch_count=3,  # epoch 이전 3건만
        epoch_unix=epoch_unix,
    )

    count = _count_active_window_pickups(str(tmp_path), activation_epoch=epoch_unix)
    assert count == 0, (
        f"epoch 이전 LIVE_PROCESSED 3건은 카운트 0 기대, 실제: {count}"
    )


# ── 테스트 14: epoch 이후 N건 → gate NOOP_N_EXCEEDED + disabled write ─────────
def test_epoch_scope_after_epoch_counted_and_trips(tmp_path):
    """gate에 epoch_reader=E + epoch 이후 LIVE_PROCESSED 3건 →
    NOOP_N_EXCEEDED + disabled flag write + rec.result_path == "".
    """
    epoch_unix = time.time() - 30 * 60  # 30분 전 (T 이내)

    _make_ledger_scoped(
        tmp_path,
        after_epoch_count=LIMITED_MAX_PICKUPS_N,
        epoch_unix=epoch_unix,
    )

    rec = limited_activation_bound_gate(
        str(tmp_path),
        epoch_reader=lambda: str(epoch_unix),
        now_fn=lambda: time.time(),
    )

    assert rec is not None
    assert rec.verdict == VERDICT_NOOP_N_EXCEEDED, (
        f"epoch 이후 N건 초과 → NOOP_N_EXCEEDED 기대, 실제: {rec.verdict!r}"
    )

    flag_file = tmp_path / ACTIVATION_FLAG_REL
    assert flag_file.exists(), "N 초과 시 disabled flag 생성 기대"
    first_line = flag_file.read_text(encoding="utf-8").splitlines()[0].strip()
    assert first_line == ACTIVATION_DISABLED, (
        f"p0b_driver_enabled 첫줄 == 'disabled' 기대, 실제: {first_line!r}"
    )

    # gate NOOP 경로: result_path == "" (fire 0)
    assert rec.result_path == "", (
        f"gate NOOP 경로: rec.result_path == '' 기대, 실제: {rec.result_path!r}"
    )


# ── 테스트 15: 재가동 안전 — 이전 창 entry는 무시 ────────────────────────────
def test_reactivation_safe_old_window_ignored(tmp_path):
    """ledger에 이전 창(과거 epoch 이전) LIVE_PROCESSED 3건 + 새 epoch_reader=E2 주입,
    E2 이후 entry 0건 → gate 반환 None (재가동 false-OFF 없음).

    재가동 안전 핵심 케이스: 이전 창 누계가 새 epoch 기준으로 무시됨.
    """
    old_epoch_unix = time.time() - 200 * 60   # 200분 전 (옛 epoch)
    new_epoch_unix = time.time() - 30 * 60    # 30분 전 (새 epoch, T=120 이내)

    # 이전 창 기준 epoch 이후(=old_epoch + 1초 이후)이지만 새 epoch(new_epoch) 이전인 3건
    ledger_path = tmp_path / LIVE_LEDGER_REL
    ledger_path.parent.mkdir(parents=True, exist_ok=True)

    def _to_kst_iso(unix_val: float) -> str:
        dt = datetime.fromtimestamp(unix_val, tz=_KST)
        return dt.isoformat()

    with open(ledger_path, "w", encoding="utf-8") as fh:
        for i in range(3):
            # old_epoch 이후이지만 new_epoch 이전 → new epoch 기준으로는 카운트 비대상
            ts_unix = old_epoch_unix + 10.0 + i  # 여전히 new_epoch 보다 훨씬 이전
            fh.write(json.dumps({
                "outcome": VERDICT_LIVE_PROCESSED,
                "task_id": f"old-task-{i}",
                "processed_at": _to_kst_iso(ts_unix),
            }) + "\n")

    # new_epoch 이후 entry 0건 → N < n_max → gate None
    rec = limited_activation_bound_gate(
        str(tmp_path),
        epoch_reader=lambda: str(new_epoch_unix),
        now_fn=lambda: time.time(),
    )

    assert rec is None, (
        f"새 epoch 이후 entry 0건 → None(재가동 안전) 기대, 실제: {rec!r}"
    )

    flag_file = tmp_path / ACTIVATION_FLAG_REL
    assert not flag_file.exists(), "재가동 안전: 이전 창 entry로 인한 false-OFF 금지 (flag 미생성)"


# ── 테스트 16: processed_at 부재 entry 제외 ──────────────────────────────────
def test_processed_at_absent_excluded(tmp_path):
    """_count_active_window_pickups 직접 호출.
    processed_at 필드 없는 LIVE_PROCESSED 3건 + activation_epoch=E → 반환 0.
    """
    epoch_unix = time.time() - 30 * 60

    ledger_path = tmp_path / LIVE_LEDGER_REL
    ledger_path.parent.mkdir(parents=True, exist_ok=True)
    with open(ledger_path, "w", encoding="utf-8") as fh:
        for i in range(3):
            # processed_at 필드 없음
            fh.write(json.dumps({
                "outcome": VERDICT_LIVE_PROCESSED,
                "task_id": f"task-{i}",
            }) + "\n")

    count = _count_active_window_pickups(str(tmp_path), activation_epoch=epoch_unix)
    assert count == 0, (
        f"processed_at 부재 LIVE_PROCESSED → 카운트 0 기대, 실제: {count}"
    )


# ── 테스트 17: processed_at 파싱 실패 entry 제외 ─────────────────────────────
def test_processed_at_unparseable_excluded(tmp_path):
    """processed_at 이 불량 값인 LIVE_PROCESSED 3건 + activation_epoch=E → 반환 0.
    _parse_processed_at_to_unix 도 각 입력에 None 반환 검증.
    """
    epoch_unix = time.time() - 30 * 60

    bad_values = ["not-a-date", "", 12345]  # str/빈str/int

    ledger_path = tmp_path / LIVE_LEDGER_REL
    ledger_path.parent.mkdir(parents=True, exist_ok=True)
    with open(ledger_path, "w", encoding="utf-8") as fh:
        for i, bv in enumerate(bad_values):
            fh.write(json.dumps({
                "outcome": VERDICT_LIVE_PROCESSED,
                "task_id": f"task-bad-{i}",
                "processed_at": bv,
            }) + "\n")

    count = _count_active_window_pickups(str(tmp_path), activation_epoch=epoch_unix)
    assert count == 0, (
        f"불량 processed_at LIVE_PROCESSED → 카운트 0 기대, 실제: {count}"
    )

    # _parse_processed_at_to_unix 개별 검증
    assert _parse_processed_at_to_unix("not-a-date") is None, \
        "'not-a-date' 파싱 → None 기대"
    assert _parse_processed_at_to_unix("") is None, \
        "빈 문자열 → None 기대"
    assert _parse_processed_at_to_unix(12345) is None, \
        "int 입력 → None 기대"


# ── 테스트 18: KST +09:00 timezone 보존 검증 ──────────────────────────────────
def test_timezone_aware_kst_offset_preserved(tmp_path):
    """_parse_processed_at_to_unix("2026-06-26T10:00:00+09:00") 와
    동일 시각 UTC "2026-06-26T01:00:00+00:00" 의 unix float 가 같음 검증.
    epoch 를 그 unix 와 동일하게 두면 >= 비교로 1건 카운트됨.
    """
    kst_str = "2026-06-26T10:00:00+09:00"
    utc_str = "2026-06-26T01:00:00+00:00"

    unix_kst = _parse_processed_at_to_unix(kst_str)
    unix_utc = _parse_processed_at_to_unix(utc_str)

    assert unix_kst is not None, f"KST 문자열 파싱 실패: {kst_str!r}"
    assert unix_utc is not None, f"UTC 문자열 파싱 실패: {utc_str!r}"
    assert unix_kst == unix_utc, (
        f"KST/UTC 동일 시각의 unix epoch 가 달라야 하지 않음: kst={unix_kst} utc={unix_utc}"
    )

    # epoch == unix_kst 로 설정하면 >= 비교로 해당 entry 1건 카운트 확인
    epoch_unix = unix_kst

    ledger_path = tmp_path / LIVE_LEDGER_REL
    ledger_path.parent.mkdir(parents=True, exist_ok=True)
    with open(ledger_path, "w", encoding="utf-8") as fh:
        fh.write(json.dumps({
            "outcome": VERDICT_LIVE_PROCESSED,
            "task_id": "task-tz-test",
            "processed_at": kst_str,
        }) + "\n")

    count = _count_active_window_pickups(str(tmp_path), activation_epoch=epoch_unix)
    assert count == 1, (
        f"processed_at == epoch (>= 비교) → 카운트 1 기대, 실제: {count}"
    )


# ── 테스트 19: epoch 부재 + active → fail-closed ──────────────────────────────
def test_epoch_absent_active_fail_closed(tmp_path):
    """gate에 epoch_reader=lambda:None + is_activated_fn=lambda:True →
    VERDICT_NOOP_ACTIVE_EPOCH_MISSING + disabled flag write 확인.
    """
    rec = limited_activation_bound_gate(
        str(tmp_path),
        epoch_reader=lambda: None,
        is_activated_fn=lambda: True,
    )

    assert rec is not None, "epoch 부재 + active → DriverRecord 반환 기대 (None 아님)"
    assert rec.verdict == VERDICT_NOOP_ACTIVE_EPOCH_MISSING, (
        f"epoch 부재 + active → NOOP_ACTIVE_EPOCH_MISSING 기대, 실제: {rec.verdict!r}"
    )

    flag_file = tmp_path / ACTIVATION_FLAG_REL
    assert flag_file.exists(), "fail-closed: disabled flag 생성 기대"
    first_line = flag_file.read_text(encoding="utf-8").splitlines()[0].strip()
    assert first_line == ACTIVATION_DISABLED, (
        f"disabled flag 첫줄 == 'disabled' 기대, 실제: {first_line!r}"
    )


# ── 테스트 20: epoch 부재 + not active → None (default-OFF 보존) ──────────────
def test_epoch_absent_inactive_returns_none(tmp_path):
    """gate에 epoch_reader=lambda:None + is_activated_fn=lambda:False →
    None 반환, flag 미생성 (default-OFF 보존).
    """
    rec = limited_activation_bound_gate(
        str(tmp_path),
        epoch_reader=lambda: None,
        is_activated_fn=lambda: False,
    )

    assert rec is None, (
        f"epoch 부재 + not active → None 기대(default-OFF), 실제: {rec!r}"
    )

    flag_file = tmp_path / ACTIVATION_FLAG_REL
    assert not flag_file.exists(), "default-OFF: flag 미생성 기대 (side-effect 0)"


# ── 테스트 21: main() T 만료 스모크 테스트 ────────────────────────────────────
def test_main_t_expired_smoke(tmp_path, monkeypatch):
    """main() 경로 L1 스모크: monkeypatch로 CANONICAL_ROOT = tmp_path.
    epoch 파일을 now-121분으로 생성(active flag 없음 → is_activated=False이지만 epoch 존재 → T 평가).
    rc = _drv.main([]) → rc == 0 + p0b_driver_enabled 첫줄 == "disabled" (T 만료 자동 disabled).
    """
    monkeypatch.setattr(_drv, "CANONICAL_ROOT", str(tmp_path))

    # activation epoch 파일 생성 (now - 121분)
    now_unix = time.time()
    epoch_unix = now_unix - 121 * 60

    epoch_file = tmp_path / _drv.ACTIVATION_EPOCH_REL
    epoch_file.parent.mkdir(parents=True, exist_ok=True)
    epoch_file.write_text(str(epoch_unix), encoding="utf-8")

    # active flag 는 생성하지 않음 → is_activated=False
    # → gate: epoch 존재이므로 epoch 부재 정책 분기 건너뜀,
    #         N은 epoch-scope count=0 (ledger 없음) → N 미초과,
    #         T는 (now - epoch) = 121분 >= 120분 → T 만료 → disabled + NOOP_T_EXPIRED

    rc = _drv.main([])
    assert rc == 0, f"main() 반환 코드 0 기대, 실제: {rc}"

    flag_file = tmp_path / _drv.ACTIVATION_FLAG_REL
    assert flag_file.exists(), "T 만료 → p0b_driver_enabled 생성 기대"
    first_line = flag_file.read_text(encoding="utf-8").splitlines()[0].strip()
    assert first_line == ACTIVATION_DISABLED, (
        f"p0b_driver_enabled 첫줄 == 'disabled' 기대, 실제: {first_line!r}"
    )


# ── 테스트 22: production main() — activation_intended=True + active + epoch 부재 → fail-closed ──
def test_main_active_intended_epoch_absent_fail_closed(tmp_path, monkeypatch):
    """production main() 실제 경로 (B안 핵심):
    activation_intended=True + p0b_driver_enabled='enabled'(active) + epoch 부재 →
    NOOP_ACTIVE_EPOCH_MISSING fail-closed: flag 'disabled' 자동전환 + rc==0 +
    scan_once/scan_live_inbox_once/launcher/governor 미호출(fire 0, governor/launcher build 이전 단락).
    """
    monkeypatch.setattr(_drv, "CANONICAL_ROOT", str(tmp_path))

    # active flag = 'enabled' 생성, epoch 파일은 생성하지 않음 (부재)
    flag_file = tmp_path / _drv.ACTIVATION_FLAG_REL
    flag_file.parent.mkdir(parents=True, exist_ok=True)
    flag_file.write_text(_drv.ACTIVATION_ENABLED + "\n", encoding="utf-8")

    # fire 경로 호출 0 검증용 스파이: 호출되면 즉시 실패시키도록 기록
    calls: dict[str, int] = {"scan_once": 0, "scan_live": 0, "launcher": 0, "governor": 0}

    def _spy_scan_once(*a, **k):
        calls["scan_once"] += 1
        return []

    def _spy_scan_live(*a, **k):
        calls["scan_live"] += 1
        return []

    def _spy_launcher(*a, **k):
        calls["launcher"] += 1
        return None

    def _spy_governor(*a, **k):
        calls["governor"] += 1
        return None

    monkeypatch.setattr(_drv, "scan_once", _spy_scan_once)
    monkeypatch.setattr(_drv, "scan_live_inbox_once", _spy_scan_live)
    monkeypatch.setattr(_drv, "build_launcher_fn", _spy_launcher)
    monkeypatch.setattr(_drv, "_build_governor_fn", _spy_governor)

    rc = _drv.main([], activation_intended=True)

    assert rc == 0, f"main() 반환 코드 0 기대, 실제: {rc}"

    # fail-closed: flag 'disabled' 자동전환
    first_line = flag_file.read_text(encoding="utf-8").splitlines()[0].strip()
    assert first_line == ACTIVATION_DISABLED, (
        f"active+epoch부재 fail-closed: p0b_driver_enabled 'disabled' 기대, 실제: {first_line!r}"
    )

    # governor/launcher build 이전 단락 → fire 경로 전부 미호출
    assert calls["scan_once"] == 0, f"scan_once 미호출 기대(fire 0), 실제 호출 {calls['scan_once']}회"
    assert calls["scan_live"] == 0, f"scan_live_inbox_once 미호출 기대(fire 0), 실제 {calls['scan_live']}회"
    assert calls["launcher"] == 0, f"launcher build 미호출 기대(단락), 실제 {calls['launcher']}회"
    assert calls["governor"] == 0, f"governor build 미호출 기대(단락), 실제 {calls['governor']}회"


# ── 테스트 23: production main() — activation_intended=True + active + epoch 손상 → fail-closed ──
def test_main_active_intended_epoch_corrupt_fail_closed(tmp_path, monkeypatch):
    """production main() 실제 경로:
    activation_intended=True + active + epoch 파일이 파싱불가(손상) →
    read_activation_epoch None → epoch 부재와 동일 fail-closed: flag 'disabled' + rc==0 + fire 0.
    """
    monkeypatch.setattr(_drv, "CANONICAL_ROOT", str(tmp_path))

    flag_file = tmp_path / _drv.ACTIVATION_FLAG_REL
    flag_file.parent.mkdir(parents=True, exist_ok=True)
    flag_file.write_text(_drv.ACTIVATION_ENABLED + "\n", encoding="utf-8")

    # 손상된 epoch 파일 (float 파싱 불가)
    epoch_file = tmp_path / _drv.ACTIVATION_EPOCH_REL
    epoch_file.parent.mkdir(parents=True, exist_ok=True)
    epoch_file.write_text("CORRUPT-NOT-A-FLOAT\n", encoding="utf-8")

    calls = {"scan_once": 0, "scan_live": 0}

    def _spy_scan_once(*a, **k):
        calls["scan_once"] += 1
        return []

    def _spy_scan_live(*a, **k):
        calls["scan_live"] += 1
        return []

    monkeypatch.setattr(_drv, "scan_once", _spy_scan_once)
    monkeypatch.setattr(_drv, "scan_live_inbox_once", _spy_scan_live)

    rc = _drv.main([], activation_intended=True)

    assert rc == 0, f"main() 반환 코드 0 기대, 실제: {rc}"
    first_line = flag_file.read_text(encoding="utf-8").splitlines()[0].strip()
    assert first_line == ACTIVATION_DISABLED, (
        f"active+epoch손상 fail-closed: 'disabled' 기대, 실제: {first_line!r}"
    )
    assert calls["scan_once"] == 0 and calls["scan_live"] == 0, (
        f"epoch 손상 fail-closed → scan 미호출(fire 0) 기대, 실제: {calls}"
    )


# ── 테스트 24: legacy 무손상 — activation_intended=False + active + epoch 부재 → fail-closed 아님 ──
def test_main_legacy_active_epoch_absent_not_fail_closed(tmp_path, monkeypatch):
    """legacy scan_once / test-2760 라우팅 무손상 검증:
    activation_intended=False(기본) + active(enabled) + epoch 부재 →
    fail-closed로 바뀌지 않음: p0b_driver_enabled 'enabled' 유지(disabled 전환 금지) +
    기존 scan_once 경로 도달(legacy 라우팅 보존).
    """
    monkeypatch.setattr(_drv, "CANONICAL_ROOT", str(tmp_path))

    flag_file = tmp_path / _drv.ACTIVATION_FLAG_REL
    flag_file.parent.mkdir(parents=True, exist_ok=True)
    flag_file.write_text(_drv.ACTIVATION_ENABLED + "\n", encoding="utf-8")
    # epoch 파일 없음 (부재)

    reached = {"scan_once": 0}

    def _spy_scan_once(*a, **k):
        reached["scan_once"] += 1
        return []

    monkeypatch.setattr(_drv, "scan_once", _spy_scan_once)

    # activation_intended 기본값(False)으로 호출 — legacy/test-2760 경로
    rc = _drv.main([])

    assert rc == 0, f"main() 반환 코드 0 기대, 실제: {rc}"

    # 무손상: flag 가 'disabled'로 전환되지 않고 'enabled' 유지
    first_line = flag_file.read_text(encoding="utf-8").splitlines()[0].strip()
    assert first_line == _drv.ACTIVATION_ENABLED, (
        f"legacy 무손상: 'enabled' 유지 기대(disabled 전환 금지), 실제: {first_line!r}"
    )

    # legacy scan_once 경로 도달 (fail-closed 단락 아님)
    assert reached["scan_once"] == 1, (
        f"legacy 경로: scan_once 도달 기대, 실제 호출 {reached['scan_once']}회"
    )


# ── 테스트 25: __main__ 블록이 main(limited_runtime=True)로 호출함을 정적 검증 ──
def test_main_limited_runtime_main_block_passes_true():
    """driver 소스에서 __main__ 진입점이 main(limited_runtime=True)로 호출하는지 정적 검증.

    inspect.getsource 를 사용해 소스 텍스트를 가져온 뒤 두 가지 사실을 확인한다:
    1. 'main(limited_runtime=True)' 가 소스에 존재함.
    2. 'if __name__ == "__main__":' 블록이 소스에 존재함 — python3 -m dispatch.anu_pickup_driver
       진입점과 동일한 경로임을 보증한다.
    """
    import inspect
    src = inspect.getsource(_drv)
    assert "main(limited_runtime=True)" in src, (
        "__main__ 블록이 main(limited_runtime=True)를 호출하지 않음 "
        "— __main__ 진입점 배선 이상"
    )
    assert 'if __name__ == "__main__":' in src, (
        'if __name__ == "__main__": 블록이 driver 소스에 없음'
    )


# ── 테스트 26: limited_runtime=True + 두 flag ON + epoch 부재 → auto-disabled ──
def test_main_limited_runtime_both_flags_on_epoch_absent_auto_disabled(tmp_path, monkeypatch):
    """limited_runtime=True, driver flag='enabled', callback flag='enabled', epoch 파일 부재.

    bound gate 가 두 flag 모두 ON(active=True) 상태에서 epoch 부재를 탐지하면
    NOOP_ACTIVE_EPOCH_MISSING(fail-closed) → driver flag 자동 'disabled' 전환 + rc==0.
    scan_once / scan_live_inbox_once / build_launcher_fn / _build_governor_fn 미호출.
    """
    from dispatch.anu_callback_launch_audit import (
        CALLBACK_LAUNCH_FLAG_REL,
        CALLBACK_LAUNCH_ENABLED,
    )

    monkeypatch.setattr(_drv, "CANONICAL_ROOT", str(tmp_path))

    # driver flag = 'enabled'
    flag_file = tmp_path / _drv.ACTIVATION_FLAG_REL
    flag_file.parent.mkdir(parents=True, exist_ok=True)
    flag_file.write_text(_drv.ACTIVATION_ENABLED + "\n", encoding="utf-8")

    # callback flag = 'enabled'
    cb_flag_file = tmp_path / CALLBACK_LAUNCH_FLAG_REL
    cb_flag_file.parent.mkdir(parents=True, exist_ok=True)
    cb_flag_file.write_text(CALLBACK_LAUNCH_ENABLED + "\n", encoding="utf-8")

    # epoch 파일 미생성 (부재)

    calls: dict[str, int] = {"scan_once": 0, "scan_live": 0, "launcher": 0, "governor": 0}

    def _spy_scan_once(*a, **k):
        calls["scan_once"] += 1
        return []

    def _spy_scan_live(*a, **k):
        calls["scan_live"] += 1
        return []

    def _spy_launcher(*a, **k):
        calls["launcher"] += 1
        return None

    def _spy_governor(*a, **k):
        calls["governor"] += 1
        return None

    monkeypatch.setattr(_drv, "scan_once", _spy_scan_once)
    monkeypatch.setattr(_drv, "scan_live_inbox_once", _spy_scan_live)
    monkeypatch.setattr(_drv, "build_launcher_fn", _spy_launcher)
    monkeypatch.setattr(_drv, "_build_governor_fn", _spy_governor)

    rc = _drv.main([], limited_runtime=True)

    assert rc == 0, f"main() 반환 코드 0 기대, 실제: {rc}"

    # fail-closed: driver flag 'disabled' 자동전환 확인
    first_line = flag_file.read_text(encoding="utf-8").splitlines()[0].strip()
    assert first_line == ACTIVATION_DISABLED, (
        f"두 flag ON + epoch 부재 → auto-disabled 기대, 실제: {first_line!r}"
    )

    # 모든 fire 경로 미호출
    assert calls["scan_once"] == 0, f"scan_once 미호출 기대, 실제 {calls['scan_once']}회"
    assert calls["scan_live"] == 0, f"scan_live 미호출 기대, 실제 {calls['scan_live']}회"
    assert calls["launcher"] == 0, f"launcher 미호출 기대, 실제 {calls['launcher']}회"
    assert calls["governor"] == 0, f"governor 미호출 기대, 실제 {calls['governor']}회"


# ── 테스트 27: limited_runtime=True + 두 flag ON + epoch 손상 → auto-disabled ──
def test_main_limited_runtime_both_flags_on_epoch_corrupt_auto_disabled(tmp_path, monkeypatch):
    """limited_runtime=True, driver flag='enabled', callback flag='enabled', epoch 파일 손상.

    epoch 파일이 float 파싱 불가 문자열일 경우 read_activation_epoch가 None 반환 →
    epoch 부재와 동일 fail-closed 경로: driver flag 'disabled' 전환 + rc==0.
    scan_once / scan_live_inbox_once 미호출.
    """
    from dispatch.anu_callback_launch_audit import (
        CALLBACK_LAUNCH_FLAG_REL,
        CALLBACK_LAUNCH_ENABLED,
    )

    monkeypatch.setattr(_drv, "CANONICAL_ROOT", str(tmp_path))

    # driver flag = 'enabled'
    flag_file = tmp_path / _drv.ACTIVATION_FLAG_REL
    flag_file.parent.mkdir(parents=True, exist_ok=True)
    flag_file.write_text(_drv.ACTIVATION_ENABLED + "\n", encoding="utf-8")

    # callback flag = 'enabled'
    cb_flag_file = tmp_path / CALLBACK_LAUNCH_FLAG_REL
    cb_flag_file.parent.mkdir(parents=True, exist_ok=True)
    cb_flag_file.write_text(CALLBACK_LAUNCH_ENABLED + "\n", encoding="utf-8")

    # 손상된 epoch 파일
    epoch_file = tmp_path / _drv.ACTIVATION_EPOCH_REL
    epoch_file.parent.mkdir(parents=True, exist_ok=True)
    epoch_file.write_text("CORRUPT-NOT-A-FLOAT\n", encoding="utf-8")

    calls: dict[str, int] = {"scan_once": 0, "scan_live": 0}

    def _spy_scan_once(*a, **k):
        calls["scan_once"] += 1
        return []

    def _spy_scan_live(*a, **k):
        calls["scan_live"] += 1
        return []

    monkeypatch.setattr(_drv, "scan_once", _spy_scan_once)
    monkeypatch.setattr(_drv, "scan_live_inbox_once", _spy_scan_live)

    rc = _drv.main([], limited_runtime=True)

    assert rc == 0, f"main() 반환 코드 0 기대, 실제: {rc}"

    # fail-closed: driver flag 'disabled' 자동전환 확인
    first_line = flag_file.read_text(encoding="utf-8").splitlines()[0].strip()
    assert first_line == ACTIVATION_DISABLED, (
        f"두 flag ON + epoch 손상 → auto-disabled 기대, 실제: {first_line!r}"
    )

    assert calls["scan_once"] == 0, f"scan_once 미호출 기대, 실제 {calls['scan_once']}회"
    assert calls["scan_live"] == 0, f"scan_live 미호출 기대, 실제 {calls['scan_live']}회"


# ── 테스트 28: limited_runtime=True + callback flag OFF → legacy 무손상 ──
def test_main_limited_runtime_callback_flag_off_legacy_intact(tmp_path, monkeypatch):
    """limited_runtime=True, driver flag='enabled', callback flag 미생성(OFF), epoch 부재.

    callback flag OFF → active=False → gate None → legacy scan_once 경로 도달(무손상).
    driver flag 'disabled' 전환 없음(fail-closed 아님).
    scan_once 1회 도달 확인.
    """
    monkeypatch.setattr(_drv, "CANONICAL_ROOT", str(tmp_path))

    # driver flag = 'enabled'
    flag_file = tmp_path / _drv.ACTIVATION_FLAG_REL
    flag_file.parent.mkdir(parents=True, exist_ok=True)
    flag_file.write_text(_drv.ACTIVATION_ENABLED + "\n", encoding="utf-8")

    # callback flag 미생성 (OFF)

    calls: dict[str, int] = {"scan_once": 0}

    def _spy_scan_once(*a, **k):
        calls["scan_once"] += 1
        return []

    monkeypatch.setattr(_drv, "scan_once", _spy_scan_once)

    rc = _drv.main([], limited_runtime=True)

    assert rc == 0, f"main() 반환 코드 0 기대, 실제: {rc}"

    # 'disabled' 전환 없음 — 'enabled' 유지
    first_line = flag_file.read_text(encoding="utf-8").splitlines()[0].strip()
    assert first_line == _drv.ACTIVATION_ENABLED, (
        f"callback OFF → legacy 무손상: 'enabled' 유지 기대, 실제: {first_line!r}"
    )

    # legacy scan_once 도달 확인
    assert calls["scan_once"] == 1, (
        f"callback OFF → scan_once 1회 기대, 실제 {calls['scan_once']}회"
    )


# ── 테스트 29: limited_runtime=True + driver flag OFF → legacy 무손상 ──
def test_main_limited_runtime_driver_flag_off_default_off_intact(tmp_path, monkeypatch):
    """limited_runtime=True, driver flag 미생성(OFF), callback flag='enabled', epoch 부재.

    driver flag OFF → active=False → gate None → legacy scan_once 경로로 진행(무손상).
    driver flag 파일 자동생성/자동전환 없음.
    scan_once 1회 도달.
    """
    from dispatch.anu_callback_launch_audit import (
        CALLBACK_LAUNCH_FLAG_REL,
        CALLBACK_LAUNCH_ENABLED,
    )

    monkeypatch.setattr(_drv, "CANONICAL_ROOT", str(tmp_path))

    # driver flag 미생성 (OFF)

    # callback flag = 'enabled'
    cb_flag_file = tmp_path / CALLBACK_LAUNCH_FLAG_REL
    cb_flag_file.parent.mkdir(parents=True, exist_ok=True)
    cb_flag_file.write_text(CALLBACK_LAUNCH_ENABLED + "\n", encoding="utf-8")

    calls: dict[str, int] = {"scan_once": 0}

    def _spy_scan_once(*a, **k):
        calls["scan_once"] += 1
        return []

    monkeypatch.setattr(_drv, "scan_once", _spy_scan_once)

    rc = _drv.main([], limited_runtime=True)

    assert rc == 0, f"main() 반환 코드 0 기대, 실제: {rc}"

    # driver flag 파일 미생성 유지
    assert not (tmp_path / _drv.ACTIVATION_FLAG_REL).exists(), (
        "driver flag OFF → flag 파일 자동생성 금지(side-effect 0)"
    )

    # legacy scan_once 도달
    assert calls["scan_once"] == 1, (
        f"driver flag OFF → scan_once 1회 기대, 실제 {calls['scan_once']}회"
    )


# ── 테스트 30: limited_runtime=True + 두 flag ON + 정상 epoch → preflight 미호출 + scan 진행 ──
def test_main_limited_runtime_normal_epoch_no_preflight_scan_proceeds(tmp_path, monkeypatch):
    """limited_runtime=True, driver flag='enabled', callback flag='enabled', 정상 epoch(now-60초).

    두 flag ON + 정상 epoch(활성창 이내) → gate None(N/T/kill 미트립).
    canary preflight(activation_intended_preflight) 미호출.
    scan_once 1회 도달. driver flag 'enabled' 유지(auto-disabled 아님).
    """
    from dispatch.anu_callback_launch_audit import (
        CALLBACK_LAUNCH_FLAG_REL,
        CALLBACK_LAUNCH_ENABLED,
    )
    import dispatch.anu_activation_preflight as _pf

    monkeypatch.setattr(_drv, "CANONICAL_ROOT", str(tmp_path))

    # driver flag = 'enabled'
    flag_file = tmp_path / _drv.ACTIVATION_FLAG_REL
    flag_file.parent.mkdir(parents=True, exist_ok=True)
    flag_file.write_text(_drv.ACTIVATION_ENABLED + "\n", encoding="utf-8")

    # callback flag = 'enabled'
    cb_flag_file = tmp_path / CALLBACK_LAUNCH_FLAG_REL
    cb_flag_file.parent.mkdir(parents=True, exist_ok=True)
    cb_flag_file.write_text(CALLBACK_LAUNCH_ENABLED + "\n", encoding="utf-8")

    # 정상 epoch (now - 60초, 활성창 120분 이내)
    now_unix = time.time()
    epoch_unix = now_unix - 60.0
    epoch_file = tmp_path / _drv.ACTIVATION_EPOCH_REL
    epoch_file.parent.mkdir(parents=True, exist_ok=True)
    epoch_file.write_text(str(epoch_unix), encoding="utf-8")

    calls: dict[str, int] = {"scan_once": 0, "preflight": 0}

    def _spy_scan_once(*a, **k):
        calls["scan_once"] += 1
        return []

    def _spy_preflight(*a, **k):
        calls["preflight"] += 1
        return {"stop": False}

    monkeypatch.setattr(_drv, "scan_once", _spy_scan_once)
    monkeypatch.setattr(_pf, "activation_intended_preflight", _spy_preflight)

    rc = _drv.main([], limited_runtime=True)

    assert rc == 0, f"main() 반환 코드 0 기대, 실제: {rc}"

    # scan_once 1회 도달 (gate None → legacy 경로 진행)
    assert calls["scan_once"] == 1, (
        f"정상 epoch → scan_once 1회 기대, 실제 {calls['scan_once']}회"
    )

    # canary preflight 미호출 (limited_runtime=True는 activation_intended=False)
    assert calls["preflight"] == 0, (
        f"limited_runtime → preflight 미호출 기대, 실제 {calls['preflight']}회"
    )

    # driver flag 'enabled' 유지 (auto-disabled 아님)
    first_line = flag_file.read_text(encoding="utf-8").splitlines()[0].strip()
    assert first_line == _drv.ACTIVATION_ENABLED, (
        f"정상 epoch → 'enabled' 유지 기대, 실제: {first_line!r}"
    )


# ── 테스트 31: limited_runtime=True + activation_intended=True 동시 → fail-closed STOP ──
def test_main_limited_runtime_and_activation_intended_conflict_stop(tmp_path, monkeypatch):
    """limited_runtime=True + activation_intended=True 동시 True → 의미충돌 fail-closed STOP.

    즉시 return 1(non-zero, exit code 충돌 감지용), 모든 fire 경로(scan_once/scan_live/launcher/governor/preflight) 미호출.
    driver flag write 없음(가드는 flag 변경 안 함).
    """
    from dispatch.anu_callback_launch_audit import (
        CALLBACK_LAUNCH_FLAG_REL,
        CALLBACK_LAUNCH_ENABLED,
    )
    import dispatch.anu_activation_preflight as _pf

    monkeypatch.setattr(_drv, "CANONICAL_ROOT", str(tmp_path))

    # driver flag = 'enabled'
    flag_file = tmp_path / _drv.ACTIVATION_FLAG_REL
    flag_file.parent.mkdir(parents=True, exist_ok=True)
    flag_file.write_text(_drv.ACTIVATION_ENABLED + "\n", encoding="utf-8")

    # callback flag = 'enabled'
    cb_flag_file = tmp_path / CALLBACK_LAUNCH_FLAG_REL
    cb_flag_file.parent.mkdir(parents=True, exist_ok=True)
    cb_flag_file.write_text(CALLBACK_LAUNCH_ENABLED + "\n", encoding="utf-8")

    # 정상 epoch
    now_unix = time.time()
    epoch_unix = now_unix - 60.0
    epoch_file = tmp_path / _drv.ACTIVATION_EPOCH_REL
    epoch_file.parent.mkdir(parents=True, exist_ok=True)
    epoch_file.write_text(str(epoch_unix), encoding="utf-8")

    calls: dict[str, int] = {
        "scan_once": 0, "scan_live": 0, "launcher": 0, "governor": 0, "preflight": 0,
    }

    def _spy_scan_once(*a, **k):
        calls["scan_once"] += 1
        return []

    def _spy_scan_live(*a, **k):
        calls["scan_live"] += 1
        return []

    def _spy_launcher(*a, **k):
        calls["launcher"] += 1
        return None

    def _spy_governor(*a, **k):
        calls["governor"] += 1
        return None

    def _spy_preflight(*a, **k):
        calls["preflight"] += 1
        return {"stop": False}

    monkeypatch.setattr(_drv, "scan_once", _spy_scan_once)
    monkeypatch.setattr(_drv, "scan_live_inbox_once", _spy_scan_live)
    monkeypatch.setattr(_drv, "build_launcher_fn", _spy_launcher)
    monkeypatch.setattr(_drv, "_build_governor_fn", _spy_governor)
    monkeypatch.setattr(_pf, "activation_intended_preflight", _spy_preflight)

    rc = _drv.main([], limited_runtime=True, activation_intended=True)

    assert rc == 1, f"의미충돌 fail-closed STOP → 반환 코드 1(non-zero) 기대, 실제: {rc}"

    # 모든 fire/scan/governor/preflight 경로 미호출
    assert calls["scan_once"] == 0, f"scan_once 미호출 기대, 실제 {calls['scan_once']}회"
    assert calls["scan_live"] == 0, f"scan_live 미호출 기대, 실제 {calls['scan_live']}회"
    assert calls["launcher"] == 0, f"launcher 미호출 기대, 실제 {calls['launcher']}회"
    assert calls["governor"] == 0, f"governor 미호출 기대, 실제 {calls['governor']}회"
    assert calls["preflight"] == 0, f"preflight 미호출 기대, 실제 {calls['preflight']}회"

    # driver flag 'enabled' 유지 (가드는 flag write 안 함)
    first_line = flag_file.read_text(encoding="utf-8").splitlines()[0].strip()
    assert first_line == _drv.ACTIVATION_ENABLED, (
        f"충돌 가드: flag write 없음(enabled 유지) 기대, 실제: {first_line!r}"
    )


# ── 테스트 32: main() 인자 없음(limited_runtime 기본 False) → legacy 무손상 ──
def test_main_no_args_limited_runtime_false_legacy_preserved(tmp_path, monkeypatch):
    """main() 인자 완전 0 — limited_runtime=False(기본), activation_intended=False(기본).

    driver flag='enabled', callback flag 미생성, epoch 부재.
    legacy scan_once 경로 도달(fail-closed 아님). driver flag 'enabled' 유지.
    이는 drv.main() 인자0 호출이 기존 레거시 동작을 100% 보존함을 검증한다.
    """
    monkeypatch.setattr(_drv, "CANONICAL_ROOT", str(tmp_path))

    # driver flag = 'enabled'
    flag_file = tmp_path / _drv.ACTIVATION_FLAG_REL
    flag_file.parent.mkdir(parents=True, exist_ok=True)
    flag_file.write_text(_drv.ACTIVATION_ENABLED + "\n", encoding="utf-8")

    # callback flag 미생성, epoch 파일 미생성

    calls: dict[str, int] = {"scan_once": 0}

    def _spy_scan_once(*a, **k):
        calls["scan_once"] += 1
        return []

    monkeypatch.setattr(_drv, "scan_once", _spy_scan_once)

    rc = _drv.main()

    assert rc == 0, f"main() 반환 코드 0 기대, 실제: {rc}"

    # 'disabled' 전환 없음 — 'enabled' 유지
    first_line = flag_file.read_text(encoding="utf-8").splitlines()[0].strip()
    assert first_line == _drv.ACTIVATION_ENABLED, (
        f"legacy 무손상: 'enabled' 유지 기대, 실제: {first_line!r}"
    )

    # legacy scan_once 1회 도달
    assert calls["scan_once"] == 1, (
        f"legacy 경로: scan_once 1회 기대, 실제 {calls['scan_once']}회"
    )


# ── 테스트 33: limited_runtime=True + kill switch 존재 → scan_once 0회(kill 최우선) ──
def test_main_limited_runtime_kill_switch_precedes(tmp_path, monkeypatch):
    """limited_runtime=True, driver flag='enabled', callback flag='enabled', 정상 epoch.
    kill switch 파일 존재 → kill 이 최우선 NOOP → scan_once 0회.
    kill 은 flag write 없음 → driver flag 'enabled' 유지.
    """
    from dispatch.anu_callback_launch_audit import (
        CALLBACK_LAUNCH_FLAG_REL,
        CALLBACK_LAUNCH_ENABLED,
    )

    monkeypatch.setattr(_drv, "CANONICAL_ROOT", str(tmp_path))

    # driver flag = 'enabled'
    flag_file = tmp_path / _drv.ACTIVATION_FLAG_REL
    flag_file.parent.mkdir(parents=True, exist_ok=True)
    flag_file.write_text(_drv.ACTIVATION_ENABLED + "\n", encoding="utf-8")

    # callback flag = 'enabled'
    cb_flag_file = tmp_path / CALLBACK_LAUNCH_FLAG_REL
    cb_flag_file.parent.mkdir(parents=True, exist_ok=True)
    cb_flag_file.write_text(CALLBACK_LAUNCH_ENABLED + "\n", encoding="utf-8")

    # 정상 epoch (now - 60초)
    now_unix = time.time()
    epoch_unix = now_unix - 60.0
    epoch_file = tmp_path / _drv.ACTIVATION_EPOCH_REL
    epoch_file.parent.mkdir(parents=True, exist_ok=True)
    epoch_file.write_text(str(epoch_unix), encoding="utf-8")

    # kill switch 파일 생성
    kill_file = tmp_path / _drv.KILL_SWITCH_REL
    kill_file.parent.mkdir(parents=True, exist_ok=True)
    kill_file.write_text("", encoding="utf-8")

    calls: dict[str, int] = {"scan_once": 0}

    def _spy_scan_once(*a, **k):
        calls["scan_once"] += 1
        return []

    monkeypatch.setattr(_drv, "scan_once", _spy_scan_once)

    rc = _drv.main([], limited_runtime=True)

    assert rc == 0, f"main() 반환 코드 0 기대, 실제: {rc}"

    # kill 최우선 → scan_once 0회
    assert calls["scan_once"] == 0, (
        f"kill switch → scan_once 0회 기대, 실제 {calls['scan_once']}회"
    )

    # kill 은 flag write 없음 → 'enabled' 유지
    first_line = flag_file.read_text(encoding="utf-8").splitlines()[0].strip()
    assert first_line == _drv.ACTIVATION_ENABLED, (
        f"kill switch: flag write 없음(enabled 유지) 기대, 실제: {first_line!r}"
    )


# ── task-2775+6 (Gemini HIGH): driver disabled short-circuit ──────────────────
def test_driver_disabled_short_circuit_skips_ledger_and_epoch(tmp_path):
    """driver_enabled_fn=lambda:False → gate 가 epoch read / ledger count 전에 None 단락.
    ledger_count_fn·epoch_reader 스파이가 호출되지 않고 rec is None (Gemini HIGH I/O 병목 차단)."""
    ledger_calls = {"n": 0}
    epoch_calls = {"n": 0}

    def _spy_ledger():
        ledger_calls["n"] += 1
        return 99999  # 호출됐다면 N trip 했을 값

    def _spy_epoch():
        epoch_calls["n"] += 1
        return time.time()

    rec = limited_activation_bound_gate(
        str(tmp_path),
        driver_enabled_fn=lambda: False,
        ledger_count_fn=_spy_ledger,
        epoch_reader=_spy_epoch,
    )
    assert rec is None, f"driver disabled → None pass-through 기대, 실제: {rec!r}"
    assert ledger_calls["n"] == 0, f"ledger count 미호출 기대, 실제 {ledger_calls['n']}회"
    assert epoch_calls["n"] == 0, f"epoch read 미호출 기대, 실제 {epoch_calls['n']}회"


def test_driver_disabled_short_circuit_no_repeated_disabled_write(tmp_path):
    """driver disabled + epoch present + ledger over threshold 라도 flag_writer(disabled write)
    가 호출되지 않는다. 3회 반복 호출해도 disabled write 0 (반복 fsync 차단)."""
    writer_calls = {"n": 0}

    def _spy_writer(val):
        writer_calls["n"] += 1
        return None

    for _ in range(3):
        rec = limited_activation_bound_gate(
            str(tmp_path),
            driver_enabled_fn=lambda: False,
            ledger_count_fn=lambda: 999,
            epoch_reader=lambda: time.time() - 10,
            flag_writer=_spy_writer,
        )
        assert rec is None, f"driver disabled → None 기대, 실제: {rec!r}"
    assert writer_calls["n"] == 0, (
        f"driver disabled → disabled write 0회 기대, 실제 {writer_calls['n']}회"
    )


def test_driver_enabled_still_evaluates_n_path(tmp_path):
    """무회귀 안전망: driver_enabled_fn=lambda:True → short-circuit 미발동 →
    N 경로 정상 평가되어 NOOP_N_EXCEEDED + disabled write 1회."""
    writer_calls = {"n": 0}

    def _spy_writer(val):
        writer_calls["n"] += 1
        return None

    rec = limited_activation_bound_gate(
        str(tmp_path),
        driver_enabled_fn=lambda: True,
        ledger_count_fn=lambda: 999,
        epoch_reader=lambda: time.time() - 10,
        flag_writer=_spy_writer,
    )
    assert rec is not None, "driver enabled → N trip 기대, 실제 None"
    assert rec.verdict == _drv.VERDICT_NOOP_N_EXCEEDED, (
        f"NOOP_N_EXCEEDED 기대, 실제: {rec.verdict!r}"
    )
    assert writer_calls["n"] == 1, f"disabled write 1회 기대, 실제 {writer_calls['n']}회"


# ── task-2775+6 (Gemini MEDIUM): _atomic_write_flag_disabled temp cleanup + 전파 ──
def test_atomic_disabled_write_keyboardinterrupt_cleanup_and_propagates(tmp_path, monkeypatch):
    """os.replace 가 KeyboardInterrupt(BaseException) 를 던져도 temp 파일이 잔존하지 않고,
    원래 예외(KeyboardInterrupt)는 삼켜지지 않고 그대로 전파된다 (try/finally cleanup)."""
    import glob as _glob

    flag_dir = tmp_path / os.path.dirname(_drv.ACTIVATION_FLAG_REL)

    def _boom(*a, **k):
        raise KeyboardInterrupt("simulated interrupt during replace")

    monkeypatch.setattr(_drv.os, "replace", _boom)
    with pytest.raises(KeyboardInterrupt):
        _drv._atomic_write_flag_disabled(str(tmp_path))
    leftover = _glob.glob(str(flag_dir / ".p0b_flag_*.tmp"))
    assert leftover == [], f"KeyboardInterrupt 후 temp 파일 잔존: {leftover}"


def test_atomic_disabled_write_runtimeerror_cleanup_and_propagates(tmp_path, monkeypatch):
    """os.replace 가 OSError 가 아닌 RuntimeError 를 던져도 temp cleanup 보장 + 예외 전파."""
    import glob as _glob

    flag_dir = tmp_path / os.path.dirname(_drv.ACTIVATION_FLAG_REL)

    def _boom(*a, **k):
        raise RuntimeError("non-OSError during replace")

    monkeypatch.setattr(_drv.os, "replace", _boom)
    with pytest.raises(RuntimeError):
        _drv._atomic_write_flag_disabled(str(tmp_path))
    leftover = _glob.glob(str(flag_dir / ".p0b_flag_*.tmp"))
    assert leftover == [], f"RuntimeError 후 temp 파일 잔존: {leftover}"


# ── task-2775+7 (Gemini fresh-head): thread3 early-stop / thread4 0644 / thread5 fail-closed ──

def _write_after_epoch_ledger(tmp_path, *, count: int, epoch_unix: float) -> str:
    """epoch 이후 LIVE_PROCESSED entry를 count개 기록(전부 카운트 대상)."""
    ledger_path = tmp_path / LIVE_LEDGER_REL
    ledger_path.parent.mkdir(parents=True, exist_ok=True)
    with open(ledger_path, "w", encoding="utf-8") as fh:
        for i in range(count):
            dt = datetime.fromtimestamp(epoch_unix + 1.0 + i, tz=_KST)
            fh.write(json.dumps({
                "outcome": VERDICT_LIVE_PROCESSED,
                "task_id": f"task-after-{i}",
                "processed_at": dt.isoformat(),
            }) + "\n")
    return str(ledger_path)


def test_count_early_stop_caps_at_max_count(tmp_path):
    """thread3: after-epoch entry가 max_count보다 많아도 결과는 max_count로 cap."""
    epoch_unix = time.time() - 100
    _write_after_epoch_ledger(tmp_path, count=10, epoch_unix=epoch_unix)
    # max_count=3 → cap 3 (전수 스캔이면 10)
    capped = _count_active_window_pickups(str(tmp_path), activation_epoch=epoch_unix, max_count=3)
    assert capped == 3
    # max_count=None → 기존 전수 스캔(하위호환)
    full = _count_active_window_pickups(str(tmp_path), activation_epoch=epoch_unix)
    assert full == 10


def test_count_early_stop_actually_stops_scanning(tmp_path):
    """thread3: max_count 도달 시 ledger 끝까지 읽지 않고 stop(코드 근거: parse 호출 횟수)."""
    epoch_unix = time.time() - 100
    _write_after_epoch_ledger(tmp_path, count=50, epoch_unix=epoch_unix)
    calls = {"n": 0}
    real_parse = _drv._parse_processed_at_to_unix

    def _counting_parse(value):
        calls["n"] += 1
        return real_parse(value)

    orig = _drv._parse_processed_at_to_unix
    _drv._parse_processed_at_to_unix = _counting_parse
    try:
        result = _count_active_window_pickups(str(tmp_path), activation_epoch=epoch_unix, max_count=3)
    finally:
        _drv._parse_processed_at_to_unix = orig
    # 3번째 after-epoch entry에서 break → parse 정확히 3회만 호출(50개 전수 아님)
    assert result == 3
    assert calls["n"] == 3


def test_count_no_epoch_before_break_under_count(tmp_path):
    """thread3: epoch 이전 entry가 파일 앞에 와도 break하지 않음(under-count 0)."""
    epoch_unix = time.time() - 100
    ledger_path = tmp_path / LIVE_LEDGER_REL
    ledger_path.parent.mkdir(parents=True, exist_ok=True)
    with open(ledger_path, "w", encoding="utf-8") as fh:
        # epoch 이전 entry 5개를 먼저 기록(이걸로 break하면 안 됨)
        for i in range(5):
            dt = datetime.fromtimestamp(epoch_unix - 10.0 - i, tz=_KST)
            fh.write(json.dumps({
                "outcome": VERDICT_LIVE_PROCESSED,
                "task_id": f"task-before-{i}",
                "processed_at": dt.isoformat(),
            }) + "\n")
        # 그 뒤 epoch 이후 entry 3개
        for i in range(3):
            dt = datetime.fromtimestamp(epoch_unix + 1.0 + i, tz=_KST)
            fh.write(json.dumps({
                "outcome": VERDICT_LIVE_PROCESSED,
                "task_id": f"task-after-{i}",
                "processed_at": dt.isoformat(),
            }) + "\n")
    # max_count 충분히 큼 → after-epoch 3개 정확히 카운트(epoch 이전 entry로 조기 break 안 함)
    count = _count_active_window_pickups(str(tmp_path), activation_epoch=epoch_unix, max_count=10)
    assert count == 3


def test_count_under_max_returns_exact(tmp_path):
    """thread3: N 미만이면 max_count 지정에도 정확한 전수 결과(under-count 금지)."""
    epoch_unix = time.time() - 100
    _write_after_epoch_ledger(tmp_path, count=2, epoch_unix=epoch_unix)
    count = _count_active_window_pickups(str(tmp_path), activation_epoch=epoch_unix, max_count=3)
    assert count == 2


def test_gate_n_path_uses_early_stop_and_trips(tmp_path):
    """thread3: 게이트가 max_count=n_max로 호출해도 N 도달 시 NOOP_N_EXCEEDED 무회귀."""
    epoch_unix = time.time() - 10
    _write_after_epoch_ledger(tmp_path, count=LIMITED_MAX_PICKUPS_N + 5, epoch_unix=epoch_unix)
    rec = limited_activation_bound_gate(
        str(tmp_path),
        now_fn=lambda: epoch_unix + 5,           # T 미만(만료 아님)
        epoch_reader=lambda: str(epoch_unix),
    )
    assert rec is not None
    assert rec.verdict == VERDICT_NOOP_N_EXCEEDED
    assert rec.activation == ACTIVATION_DISABLED


def test_atomic_disabled_write_mode_is_0644(tmp_path):
    """thread4: _atomic_write_flag_disabled 최종 flag 파일 권한 0o644."""
    err = _atomic_write_flag_disabled(str(tmp_path))
    assert err is None
    flag_path = _flag_path(tmp_path)
    assert os.path.exists(flag_path)
    mode = os.stat(flag_path).st_mode & 0o777
    assert mode == 0o644, f"기대 0o644, 실제 {oct(mode)}"
    with open(flag_path, encoding="utf-8") as fh:
        assert fh.read() == ACTIVATION_DISABLED + "\n"


def test_t_elapsed_typeerror_fail_closed(tmp_path):
    """thread5: now() 비숫자(TypeError) → fail-OPEN 금지, auto-disabled + NOOP_ACTIVE_EPOCH_INVALID."""
    epoch_unix = time.time() - 10
    rec = limited_activation_bound_gate(
        str(tmp_path),
        now_fn=lambda: None,                     # float(None) → TypeError
        epoch_reader=lambda: str(epoch_unix),
    )
    assert rec is not None, "fail-open(None pass-through) 금지"
    assert rec.verdict == VERDICT_NOOP_ACTIVE_EPOCH_INVALID
    assert rec.activation == ACTIVATION_DISABLED
    # flag auto-disabled 기록 확인
    flag_path = _flag_path(tmp_path)
    assert os.path.exists(flag_path)
    with open(flag_path, encoding="utf-8") as fh:
        assert fh.read() == ACTIVATION_DISABLED + "\n"


def test_t_elapsed_valueerror_fail_closed(tmp_path):
    """thread5: now() 파싱불가 문자열(ValueError) → fail-closed."""
    epoch_unix = time.time() - 10
    rec = limited_activation_bound_gate(
        str(tmp_path),
        now_fn=lambda: "not-a-number",           # float("not-a-number") → ValueError
        epoch_reader=lambda: str(epoch_unix),
    )
    assert rec is not None
    assert rec.verdict == VERDICT_NOOP_ACTIVE_EPOCH_INVALID
    assert rec.activation == ACTIVATION_DISABLED


def test_three_paths_converge_same_safety_policy(tmp_path):
    """thread3/5 일관성: N초과·active+epoch부재·elapsed손상 세 경로 모두
    동일 안전정책(activation=DISABLED + NOOP verdict + fire 0)으로 수렴."""
    # 경로1: N 초과
    p1 = tmp_path / "p1"
    epoch1 = time.time() - 10
    _write_after_epoch_ledger(p1, count=LIMITED_MAX_PICKUPS_N + 2, epoch_unix=epoch1)
    rec_n = limited_activation_bound_gate(
        str(p1), now_fn=lambda: epoch1 + 5, epoch_reader=lambda: str(epoch1))
    # 경로2: active + epoch 부재
    p2 = tmp_path / "p2"
    p2.mkdir(parents=True, exist_ok=True)
    rec_epoch_missing = limited_activation_bound_gate(
        str(p2), epoch_reader=lambda: None, is_activated_fn=lambda: True)
    # 경로3: elapsed 손상
    p3 = tmp_path / "p3"
    p3.mkdir(parents=True, exist_ok=True)
    epoch3 = time.time() - 10
    rec_invalid = limited_activation_bound_gate(
        str(p3), now_fn=lambda: None, epoch_reader=lambda: str(epoch3))
    for rec in (rec_n, rec_epoch_missing, rec_invalid):
        assert rec is not None
        assert rec.activation == ACTIVATION_DISABLED
        assert rec.verdict.startswith("NOOP_")
        assert rec.result_path == ""   # fire 0(결과 발사 경로 없음)
    assert rec_n.verdict == VERDICT_NOOP_N_EXCEEDED
    assert rec_epoch_missing.verdict == VERDICT_NOOP_ACTIVE_EPOCH_MISSING
    assert rec_invalid.verdict == VERDICT_NOOP_ACTIVE_EPOCH_INVALID


# ── task-2775+8: newest-first reverse chunk scan 검증 ─────────────────────────

def _write_historical_front_then_active_tail(tmp_path, *, before_count, after_count, epoch_unix):
    """파일 **앞쪽**에 historical(before-epoch) entry 대량 + **끝쪽**에 active(after-epoch) entry.
    실제 ledger append 순서(과거→현재)를 재현한다."""
    ledger_path = tmp_path / LIVE_LEDGER_REL
    ledger_path.parent.mkdir(parents=True, exist_ok=True)
    with open(ledger_path, "w", encoding="utf-8") as fh:
        for i in range(before_count):
            dt = datetime.fromtimestamp(epoch_unix - 100.0 - i, tz=_KST)
            fh.write(json.dumps({
                "outcome": VERDICT_LIVE_PROCESSED,
                "task_id": f"hist-{i}",
                "processed_at": dt.isoformat(),
            }) + "\n")
        for i in range(after_count):
            dt = datetime.fromtimestamp(epoch_unix + 1.0 + i, tz=_KST)
            fh.write(json.dumps({
                "outcome": VERDICT_LIVE_PROCESSED,
                "task_id": f"active-{i}",
                "processed_at": dt.isoformat(),
            }) + "\n")
    return str(ledger_path)


def test_reverse_iter_yields_lines_newest_first(tmp_path):
    """_reverse_line_iter 가 line 을 파일 끝→앞(newest-first) 순서로 yield."""
    p = tmp_path / "lines.txt"
    p.write_text("L0\nL1\nL2\nL3\nL4\n", encoding="utf-8")
    with open(p, "rb") as fh:
        got = [ln for ln in _reverse_line_iter(fh) if ln != ""]
    assert got == ["L4", "L3", "L2", "L1", "L0"]


def test_reverse_iter_handles_line_split_across_chunks(tmp_path):
    """tiny chunk_size 로 line 이 chunk 경계를 넘어도 정확히 복원(newest-first)."""
    p = tmp_path / "lines.txt"
    lines = [f"line-number-{i:03d}-padding" for i in range(20)]
    p.write_text("\n".join(lines) + "\n", encoding="utf-8")
    with open(p, "rb") as fh:
        got = [ln for ln in _reverse_line_iter(fh, chunk_size=4) if ln != ""]
    assert got == list(reversed(lines))


def test_reverse_iter_no_trailing_newline(tmp_path):
    """파일이 newline 으로 끝나지 않아도 마지막 line 누락 없음."""
    p = tmp_path / "lines.txt"
    p.write_text("A\nB\nC", encoding="utf-8")  # trailing newline 없음
    with open(p, "rb") as fh:
        got = [ln for ln in _reverse_line_iter(fh) if ln != ""]
    assert got == ["C", "B", "A"]


def test_count_reverse_skips_historical_front_when_max_reached(tmp_path):
    """★ 핵심: 파일 앞 historical 1000건 + 끝 active 3건. max_count=3 도달 시
    historical entry 를 **읽지 않는다**(parse 호출 횟수로 입증)."""
    epoch_unix = time.time() - 100
    _write_historical_front_then_active_tail(
        tmp_path, before_count=1000, after_count=3, epoch_unix=epoch_unix)
    calls = {"n": 0}
    real_parse = _drv._parse_processed_at_to_unix

    def _counting_parse(value):
        calls["n"] += 1
        return real_parse(value)

    orig = _drv._parse_processed_at_to_unix
    _drv._parse_processed_at_to_unix = _counting_parse
    try:
        result = _count_active_window_pickups(
            str(tmp_path), activation_epoch=epoch_unix, max_count=3)
    finally:
        _drv._parse_processed_at_to_unix = orig
    assert result == 3
    # active 3건만 parse → 호출 3회. 순방향이면 1000+ 회였을 것.
    assert calls["n"] == 3, f"historical front 미스캔 기대(parse 3회), 실제 {calls['n']}회"


def test_count_reverse_bounded_bytes_not_whole_file_read(tmp_path):
    """★ 전체 readlines() 가 아님 입증: 파일 앞 historical 2000건 + 끝 active 3건일 때
    max_count=3 도달까지 읽은 **바이트 수가 파일 전체보다 훨씬 작다**."""
    epoch_unix = time.time() - 100
    ledger = _write_historical_front_then_active_tail(
        tmp_path, before_count=2000, after_count=3, epoch_unix=epoch_unix)
    total_size = os.path.getsize(ledger)

    class _ByteCountingFile:
        def __init__(self, fh):
            self._fh = fh
            self.bytes_read = 0
        def read(self, n=-1):
            data = self._fh.read(n)
            self.bytes_read += len(data)
            return data
        def seek(self, *a, **k):
            return self._fh.seek(*a, **k)
        def tell(self):
            return self._fh.tell()

    # _reverse_line_iter 를 직접 구동: active 3건만 소비하고 멈춘다.
    collected = []
    with open(ledger, "rb") as raw:
        cf = _ByteCountingFile(raw)
        for ln in _reverse_line_iter(cf, chunk_size=4096):
            ln = ln.strip()
            if not ln:
                continue
            collected.append(ln)
            if len(collected) >= 3:
                break
    assert len(collected) == 3
    # 끝에서 3줄만 읽었으므로 읽은 바이트는 전체의 작은 일부여야 한다.
    assert cf.bytes_read < total_size, "전체 파일을 읽지 않아야 함(readlines 금지)"
    assert cf.bytes_read <= 4096 * 3, f"tail 일부만 읽어야 함, 실제 {cf.bytes_read}B / 전체 {total_size}B"


def test_count_reverse_under_max_reads_to_start_exact(tmp_path):
    """★ N 미만: 앞 historical 500건 + 끝 active 2건, max_count=3.
    active 2건을 정확히 카운트(under-count 0) — 파일 시작까지 읽어야 함."""
    epoch_unix = time.time() - 100
    _write_historical_front_then_active_tail(
        tmp_path, before_count=500, after_count=2, epoch_unix=epoch_unix)
    count = _count_active_window_pickups(
        str(tmp_path), activation_epoch=epoch_unix, max_count=3)
    assert count == 2


def test_count_reverse_mixed_epoch_no_regression(tmp_path):
    """epoch 이전/이후 mixed ledger 에서 reverse scan 도 epoch-scope 정확."""
    epoch_unix = time.time() - 100
    ledger_path = tmp_path / LIVE_LEDGER_REL
    ledger_path.parent.mkdir(parents=True, exist_ok=True)
    with open(ledger_path, "w", encoding="utf-8") as fh:
        # interleave: before, after, before, after, after
        specs = [
            (epoch_unix - 5, "b0"), (epoch_unix + 1, "a0"),
            (epoch_unix - 3, "b1"), (epoch_unix + 2, "a1"), (epoch_unix + 3, "a2"),
        ]
        for ts, tid in specs:
            dt = datetime.fromtimestamp(ts, tz=_KST)
            fh.write(json.dumps({
                "outcome": VERDICT_LIVE_PROCESSED,
                "task_id": tid,
                "processed_at": dt.isoformat(),
            }) + "\n")
    # after-epoch 3건(a0,a1,a2) 정확 카운트
    count = _count_active_window_pickups(str(tmp_path), activation_epoch=epoch_unix, max_count=10)
    assert count == 3


def test_count_reverse_excludes_missing_and_unparseable(tmp_path):
    """reverse scan 에서도 processed_at 부재/파싱실패/naive 제외 정책 무회귀."""
    epoch_unix = time.time() - 100
    ledger_path = tmp_path / LIVE_LEDGER_REL
    ledger_path.parent.mkdir(parents=True, exist_ok=True)
    with open(ledger_path, "w", encoding="utf-8") as fh:
        # 정상 after-epoch 2건
        for i in range(2):
            dt = datetime.fromtimestamp(epoch_unix + 1.0 + i, tz=_KST)
            fh.write(json.dumps({"outcome": VERDICT_LIVE_PROCESSED, "task_id": f"ok-{i}",
                                 "processed_at": dt.isoformat()}) + "\n")
        # processed_at 부재
        fh.write(json.dumps({"outcome": VERDICT_LIVE_PROCESSED, "task_id": "no-pa"}) + "\n")
        # 파싱 불가
        fh.write(json.dumps({"outcome": VERDICT_LIVE_PROCESSED, "task_id": "bad-pa",
                             "processed_at": "not-a-timestamp"}) + "\n")
        # naive(tz 미보존)
        naive = datetime.fromtimestamp(epoch_unix + 5.0).replace(tzinfo=None)
        fh.write(json.dumps({"outcome": VERDICT_LIVE_PROCESSED, "task_id": "naive-pa",
                             "processed_at": naive.isoformat()}) + "\n")
    count = _count_active_window_pickups(str(tmp_path), activation_epoch=epoch_unix, max_count=10)
    assert count == 2, "부재/파싱실패/naive 는 제외(정상 2건만)"


# ── task-2775+9: _reverse_line_iter blank-line correctness 검증 ──────────────


def test_reverse_iter_leading_blank_line_not_dropped(tmp_path):
    """파일이 빈 줄로 시작할 때 첫 빈 라인 누락 없음(핵심 회귀)."""
    p = tmp_path / "leading_blank.txt"
    p.write_text("\nfoo\n", encoding="utf-8")
    with open(p, "rb") as fh:
        got = [ln for ln in _reverse_line_iter(fh)]
    assert got == ["", "foo", ""], f"기대 ['', 'foo', ''], 실제 {got!r}"


def test_reverse_iter_leading_consecutive_blank_lines(tmp_path):
    """연속 빈 줄로 시작하는 파일 — 전부 누락 없이 yield."""
    data = "\n\nb\n"
    p = tmp_path / "consec_blank.txt"
    p.write_text(data, encoding="utf-8")
    expected = list(reversed(data.split("\n")))
    with open(p, "rb") as fh:
        got = [ln for ln in _reverse_line_iter(fh)]
    assert got == expected, f"기대 {expected!r}, 실제 {got!r}"


def test_reverse_iter_empty_file_yields_nothing(tmp_path):
    """빈 파일(0바이트) 에서 _reverse_line_iter 는 아무것도 yield 하지 않아야 한다."""
    p = tmp_path / "empty.txt"
    p.write_bytes(b"")
    with open(p, "rb") as fh:
        got = list(_reverse_line_iter(fh))
    assert got == [], f"빈 파일 → [] 기대, 실제 {got!r}"


def test_reverse_iter_single_newline_file(tmp_path):
    """파일 내용이 newline 1바이트만 있을 때 ['', ''] yield."""
    p = tmp_path / "single_nl.txt"
    p.write_text("\n", encoding="utf-8")
    with open(p, "rb") as fh:
        got = list(_reverse_line_iter(fh))
    assert got == ["", ""], f"'\\n' 파일 → ['', ''] 기대, 실제 {got!r}"


def test_reverse_iter_trailing_newline_no_regression(tmp_path):
    """trailing newline 이 있는 일반 파일 — reversed(split) 와 완전 일치(무회귀)."""
    data = "A\nB\n"
    p = tmp_path / "trailing_nl.txt"
    p.write_text(data, encoding="utf-8")
    expected = list(reversed(data.split("\n")))
    with open(p, "rb") as fh:
        got = list(_reverse_line_iter(fh))
    assert got == expected, f"기대 {expected!r}, 실제 {got!r}"


def test_reverse_iter_blank_line_at_chunk_boundary(tmp_path):
    """chunk 경계에 빈 줄이 걸려도 정확히 복원 — chunk_size=4 로 강제."""
    data = "aaaa\n\nbbbb\n"
    p = tmp_path / "chunk_boundary.txt"
    p.write_text(data, encoding="utf-8")
    expected = list(reversed(data.split("\n")))
    with open(p, "rb") as fh:
        got = list(_reverse_line_iter(fh, chunk_size=4))
    assert got == expected, f"기대 {expected!r}, 실제 {got!r}"


def test_reverse_iter_matches_forward_split_general(tmp_path):
    """다양한 입력 × chunk_size 에서 reversed(data.split('\\n')) 와 항상 일치."""
    cases = [
        "\nfoo\n",
        "\n\n\n",
        "a\nb\nc",
        "x\n\ny\n\n",
    ]
    chunk_sizes = [1, 3, 8192]
    for data in cases:
        expected = list(reversed(data.split("\n")))
        p = tmp_path / "general.txt"
        p.write_text(data, encoding="utf-8")
        for cs in chunk_sizes:
            with open(p, "rb") as fh:
                got = list(_reverse_line_iter(fh, chunk_size=cs))
            assert got == expected, (
                f"data={data!r}, chunk_size={cs}: 기대 {expected!r}, 실제 {got!r}"
            )
    # 빈 파일은 별도 처리: [] 기대
    empty_p = tmp_path / "empty_general.txt"
    empty_p.write_bytes(b"")
    for cs in chunk_sizes:
        with open(empty_p, "rb") as fh:
            got = list(_reverse_line_iter(fh, chunk_size=cs))
        assert got == [], f"빈 파일, chunk_size={cs}: [] 기대, 실제 {got!r}"


def test_count_active_window_with_leading_blank_ledger_lines(tmp_path):
    """ledger 파일이 빈 줄로 시작해도 _count_active_window_pickups 카운트 정확."""
    epoch_unix = time.time() - 100
    ledger_path = tmp_path / LIVE_LEDGER_REL
    ledger_path.parent.mkdir(parents=True, exist_ok=True)
    with open(ledger_path, "w", encoding="utf-8") as fh:
        # 파일 맨 앞에 빈 줄 2개
        fh.write("\n")
        fh.write("\n")
        # after-epoch entry 2건 (카운트 대상)
        for i in range(2):
            dt = datetime.fromtimestamp(epoch_unix + 1.0 + i, tz=_KST)
            fh.write(json.dumps({
                "outcome": VERDICT_LIVE_PROCESSED,
                "task_id": f"after-{i}",
                "processed_at": dt.isoformat(),
            }) + "\n")
    count = _count_active_window_pickups(
        str(tmp_path), activation_epoch=epoch_unix, max_count=10)
    assert count == 2, (
        f"빈 줄 시작 ledger 에서도 after-epoch 2건 카운트 기대, 실제 {count}"
    )