# -*- coding: utf-8 -*-
"""dispatch.anu_pickup_driver — task-2721 P0-b user-level systemd path driver.

default DISABLED. activation flag(memory/state/p0b_driver_enabled == "enabled") 부재 시 전면 no-op.
scan 한정: 전용 inbox(memory/events/p0b_inbox/task-*.result.json) final 만(task-2733).
  memory/events 직속 legacy result.json 은 scan 대상 아님(무변경). tmp/partial/다른 marker·jsonl·md → NOOP_NOT_TARGET.
6조건 전부 PASS 시에만 P0-a pickup_once(lock-free) 호출. pickup_once 가 ANU-owned wake argv(dry-run)를 빌드.
실제 cron 발사 0(P0-a dry_run=True/FIRE_NOT_ACTIVATED). ANU key literal 0 — .env.keys 런타임 로드만.
"""
from __future__ import annotations

import functools
import glob
import hashlib
import hmac
import json
import os
import shutil
import tempfile
import time
from collections import OrderedDict
from dataclasses import dataclass
from datetime import datetime, timedelta, timezone
from typing import Optional

# 재사용 모듈 (절대 수정 금지)
from dispatch.anu_result_pickup_runner import pickup_once, _default_sealed_key_loader  # pyright: ignore[reportMissingImports]
from dispatch.anu_pickup_wake_launcher import launch_wake  # pyright: ignore[reportMissingImports]
from dispatch.anu_owned_callback_enforcement import (  # pyright: ignore[reportMissingImports]
    CANONICAL_ROOT,
    VERDICT_AUTHORITATIVE,
    verify_collector_authoritative,
)
# task-2731: ANU-owned callback launch 강제 팩토리(build_callback_launch_fn)는 main()
# 에서 **lazy import** 한다 — 모듈 레벨 의존을 피해 기존 file-spec 로더 테스트 결선을
# 깨뜨리지 않는다(코드베이스 lazy-import 패턴 일치). process_one/scan_once 는 주입된
# callback_launch_fn 을 전달만 하므로 import 불필요.
# task-2732: 수거 verdict → 회장 PASS/HOLD/CALLBACK_MISSING 보고(sendfile only, wake 0).
from dispatch.anu_pickup_reporter import report_records  # pyright: ignore[reportMissingImports]
# task-2753: Option B provenance-path (opt-in, default OFF). envelope 부재 result 의
#   provenance hard-fail 검증 + ANU collector_result(adopted_via="provenance") 생성.
from dispatch.anu_executor_provenance import (  # pyright: ignore[reportMissingImports]
    validate_executor_provenance,
)
from dispatch.anu_collector_result import (  # pyright: ignore[reportMissingImports]
    CollectorResult,
    OwnerProof,
    write_collector_result,
    collector_result_path,
    determine_agent_relay,
    CLOSEOUT_DONE_ACKED,
    COLLECTOR_ROLE_ANU,
    ADOPTED_VIA_PROVENANCE,
)
# task-2756: provenance ADOPTED → ANU-owned callback bridge (default OFF).
from dispatch.normal_fallback_callback_helper import build_anu_owned_callback_request  # pyright: ignore[reportMissingImports]

# ── 상수 ─────────────────────────────────────────────────────────────────────
DRIVER_NAME = "systemd-path"
ACTIVATION_FLAG_REL = "memory/state/p0b_driver_enabled"
EVENTS_DIR_REL = "memory/events"
# task-2733: 전용 inbox 디렉토리 — systemd .path 와 driver scan 이 함께 감시하는 신규 전용
#   경로. 팀봇은 신규 result.json 을 여기에만 작성하고, driver 는 처리 후 result 파일을
#   processed(inbox 밖)로 이동해 inbox 를 비운다. memory/events 직속 legacy 는 scan 대상
#   아님 → 잔존 legacy 폭주 0·재트리거 0, 기존 legacy 무변경(이동/삭제 0).
INBOX_DIR_REL = "memory/events/p0b_inbox"
QUARANTINE_DIR_REL = "memory/p0b_state/quarantine"
PROCESSED_DIR_REL = "memory/p0b_state/processed"
EVIDENCE_JSONL_REL = "memory/p0b_state/driver_runs.jsonl"
# task-2744: pre_activation_epoch legacy result 의 preserve-move 대상 디렉토리. ★ watched
#   glob(memory/events/p0b_inbox/task-*.result.json) **밖**이다 → PathExistsGlob(level-trigger)
#   를 유지한 채 watched inbox 를 비워 재트리거 폭주(start-limit-hit)를 구조적으로 제거한다.
#   삭제 0·내용변경 0: 원본을 이 디렉토리로 copy(sha256 before/after 검증) 후 watched 경로만
#   비운다(content 보존). append-only manifest 로 가역(rollback) 보장.
LEGACY_SKIP_DIR_REL = "memory/p0b_state/legacy_skip"
LEGACY_SKIP_MANIFEST_REL = "memory/p0b_state/legacy_skip/legacy_skip_migration.jsonl"
# preserve-move 가 적용되는 단 하나의 skip reason. epoch_absent/legacy_stat_fail 은
#   안전하게 분류 불가하므로 현행 NOOP_LEGACY_SKIP(move 0) 보수 유지.
LEGACY_SKIP_REASON = "pre_activation_epoch"
LEGACY_SKIP_ACTOR = "anu_pickup_driver"
RESULT_GLOB = "task-*.result.json"
MAX_FILES = 50  # scan_once 단일 cycle 처리 상한 (안전한 상한). 초과분은 다음 cycle.
ACTIVATION_EPOCH_REL = "memory/state/p0b_activation_epoch"  # driver 읽기 전용 마커

# ── task-2775: limited activation 3경계 (kill / N / T) 상수 ──────────────────
KILL_SWITCH_REL = "memory/state/p0b_kill"   # 존재 시 즉시 전면 NOOP (1-touch kill)
LIMITED_MAX_PICKUPS_N = 3                    # 활성창 실발사 누계 상한 (processed_ledger 기준)
LIMITED_WINDOW_MINUTES_T = 120              # 활성 시간창(분). epoch + T 경과 시 자동 disabled
# limited-bound gate verdict 라벨
VERDICT_NOOP_KILL_SWITCH = "NOOP_KILL_SWITCH"
VERDICT_NOOP_N_EXCEEDED = "NOOP_N_EXCEEDED_AUTO_DISABLED"
VERDICT_NOOP_T_EXPIRED = "NOOP_T_EXPIRED_AUTO_DISABLED"
VERDICT_NOOP_ACTIVE_EPOCH_MISSING = "NOOP_ACTIVE_EPOCH_MISSING_AUTO_DISABLED"  # task-2775+1: active인데 epoch 부재 → fail-closed
VERDICT_NOOP_ACTIVE_EPOCH_INVALID = "NOOP_ACTIVE_EPOCH_INVALID_AUTO_DISABLED"  # task-2775+7 thread5: elapsed(now/epoch) 손상 → fail-closed(auto-disabled)

# ── task-2760: P0B event strategy — live inbox + processed ledger ────────────────
# ★ "파일 존재" 감시 → "신규 이벤트(live inbox)" 감시 전환. PathExistsGlob 가 retained
#   legacy result.json 의 "존재"만으로 반복 발화(retrigger)하던 SYSTEMD_EVENT_STRATEGY_
#   CONFLICT 을 구조적으로 제거하기 위한 빌딩블록이다. 본 PR 은 driver 에 LIVE_INBOX 전용
#   scan 경로(scan_live_inbox_once)만 추가하며, 기존 INBOX_DIR_REL(systemd .path 정렬
#   p0b_inbox)·scan_once·process_one·executor writer 경로는 일절 변경하지 않는다(회귀 0).
#   systemd .path/.service 변경·production activation·ACTIVE=true·canary 실행은 이 PR
#   범위 밖(별도 회장 승인). 본 함수는 ACTIVE=false(activation flag 미설정) 시 전면 no-op.
LIVE_INBOX_DIR_REL = "memory/events/p0b_live_inbox"          # 신규 이벤트 전용 감시 디렉토리
LIVE_PROCESSED_DIR_REL = "memory/events/p0b_processed"       # 처리 성공 후 원본 이동(감시 밖)
LIVE_LEDGER_REL = "memory/events/p0b_processed_ledger.jsonl"  # append-only processed ledger(단일 done-marker)
LIVE_RUNNER_ID = "anu_pickup_driver"

# live-inbox event strategy verdict (라벨만 — raw key 0)
VERDICT_LIVE_PROCESSED = "LIVE_PROCESSED"               # gate-ready ANU pickup → processed move 성공 후 완료 ledger append
VERDICT_LIVE_SKIP_DEDUPE = "LIVE_SKIP_DEDUPE"           # (task_id, source_result_sha256) 이미 완료 ledger 존재 → 재처리 0
VERDICT_LIVE_PENDING_BUSY = "LIVE_PENDING_BUSY"         # busy → 지연(누락 아님), inbox 잔존(다음 idle cycle 재감지)
VERDICT_LIVE_PENDING_OWNER = "LIVE_PENDING_OWNER_PROOF"  # owner_proof gate 미통과(self/foreign/미주입) → fire 0, inbox 잔존
# ★ task-2760+2 move-first 불변식(HIGH 1): processed move 실패 → 완료 ledger 미기록·inbox 잔존(retry pending).
#   완료(ledger 존재 ⟺ move 성공) vs pending/failure 상태가 혼동되지 않게 분리한다. (ledger
#   무한증가/self-healing 류 강건화는 범위 분리 — DESIGN_BACKLOG. 본 PR 은 move-first 불변식만
#   보존하는 범위 축소 replacement 이다.)
VERDICT_LIVE_PENDING_MOVE = "LIVE_PENDING_MOVE"        # processed move 실패 → 완료 아님, inbox 잔존(다음 cycle 재시도)
# task-2755: P0-B scope-limited provenance-path 활성 flag (4조건 AND, default OFF).
PROVENANCE_PATH_FLAG_REL = "memory/state/p0b_provenance_path_enabled"
PROVENANCE_PATH_FLAG_ENABLED = "enabled"
PROVENANCE_SCOPE_ID_REL = "memory/state/p0b_provenance_scope_id"

# ── readiness grace window (write race 방어) ─────────────────────────────────
# writer flush 완료 전 부분 JSON 을 즉시 quarantine 하지 않기 위한 파라미터.
# 테스트에서 process_one kwargs(stable_sec/readiness_retries/readiness_interval/sleep_fn)로 주입 가능.
STABLE_SEC = 2.0            # mtime 이 now - STABLE_SEC 이내면 in-flight 가능성 → DEFER
STABILITY_RETRIES = 3      # size/mtime 안정성 재확인 stat 횟수 (총 window ≤ 2s)
STABILITY_INTERVAL_SEC = 0.2  # stat 샘플 간격(초)

# verdict
VERDICT_WAKE_BUILT = "WAKE_BUILT"
VERDICT_CLOSEOUT_DONE = "CLOSEOUT_DONE"   # task-2730: green-path 결정론 closeout(launcher 미호출)
VERDICT_PICKUP_SKIP = "PICKUP_SKIP"
VERDICT_QUARANTINE = "QUARANTINE"
VERDICT_FIRE_FAILED = "FIRE_FAILED"
VERDICT_NOOP_DISABLED = "NOOP_DISABLED"
VERDICT_NOOP_NOT_TARGET = "NOOP_NOT_TARGET"
VERDICT_NOOP_NOT_READY = "NOOP_NOT_READY"  # readiness 미충족 → DEFER (wake 0/quarantine 0)
VERDICT_NOOP_LEGACY_SKIP = "NOOP_LEGACY_SKIP"   # activation_epoch 이전 result → move 0/wake 0/quarantine 0
# task-2744: pre_activation_epoch legacy result 를 watched glob 밖으로 preserve-move 했음.
#   "처리 완료"(wake/closeout) 아님 — content 보존 이동(삭제0·내용변경0)·manifest 기록·가역.
VERDICT_LEGACY_SKIP_PRESERVED = "LEGACY_SKIP_PRESERVED"
# preserve-move 시도 중 sha256 mismatch/IO 실패 → 원본 보존(watched 경로 유지)·explicit failure.
VERDICT_LEGACY_SKIP_PRESERVE_FAILED = "LEGACY_SKIP_PRESERVE_FAILED"
VERDICT_NOOP_MAX_FILES_DEFER = "NOOP_MAX_FILES_DEFER"  # MAX_FILES 초과분 defer

# owner_key_class (literal 키 아님 — 분류 라벨만)
OKC_ANU = "ANU"
OKC_FOREIGN = "FOREIGN"
OKC_SELF = "SELF"
# task-2753: provenance-path 로 채택된 result 의 owner_key_class 라벨 (raw key 0).
OKC_PROVENANCE = "PROVENANCE_ADOPTED"

# task-2753: Option B provenance-path opt-in flag (default OFF).
#   flag ON 일 때만 envelope 부재 result 를 provenance hard-fail 검증한다.
#   flag OFF(기본) → envelope 부재 → 기존 owner_unprovable quarantine 100% 보존.
#   이 flag 는 activation flag(p0b_driver_enabled)/real-wake flag/ACTIVE flag 와
#   무관하다 — provenance-path 는 collector_result 생성만 하고 callback fire 0
#   (fire 는 여전히 frozen owner_proof_pickup_gate 경유). 켜도 live fire/cron 0.
PROVENANCE_PATH_ENABLED_DEFAULT = False
# provenance PASS → collector_result 생성 완료 verdict.
VERDICT_PROVENANCE_ADOPTED = "PROVENANCE_ADOPTED"

# task-2756: provenance ADOPTED → ANU-owned callback bridge 상태 라벨 (default OFF).
#   callback_launch_fn 미주입 → bridge 는 (None, None) 반환(fire 0, 무회귀).
BRIDGE_OWNER_GATE_BLOCKED = "BRIDGE_OWNER_GATE_BLOCKED"   # callback_launch_fn 주입 + owner_gate_fn 부재 → fail-closed
BRIDGE_DEDUPED = "BRIDGE_DEDUPED"                         # 동일 task_id+sha 재처리 → callback 0
BRIDGE_OWNER_GATE_FAIL = "BRIDGE_OWNER_GATE_FAIL"         # owner_gate_fn FAIL → fire 0
BRIDGE_CALLBACK_OWNER_FAIL = "BRIDGE_CALLBACK_OWNER_FAIL" # build_anu_owned_callback_request verdict!=PASS(self-key 등) → fire 0
BRIDGE_FIRED = "BRIDGE_CALLBACK_REQUEST_BUILT"            # 정상: request 생성 + callback_launch_fn(request) 호출
GATE_READY_STATUS = "OWNER_PICKUP_READY_NO_FIRE"

# task-2746: driver owner-pickup fire-state contract split (opt-in).
#   기본값 OFF → 기존 "collector_envelope dict 면 무조건 verify_collector_authoritative"
#   동작을 한 글자도 바꾸지 않고 보존(무회귀). ON 일 때만 fire-state 분기(경로 A/B)를 적용.
#   ★ 실제 driver main() 경로(= canary 실행 경로)에서만 ON 으로 주입한다(아래 main()).
#   이 flag 는 코드-레벨 verify 분기 토글일 뿐, activation flag(p0b_driver_enabled)/
#   real-wake flag/ACTIVE flag 와 무관하다 — 켜도 live fire/cron 0 (driver 자체가 disabled).
OWNER_PICKUP_CONTRACT_SPLIT_DEFAULT = False
# 경로 A(pre-fire owner-pickup required) 진입 시 owner_proof_info.l1_outcome/l2_verdict 라벨.
#   raw key 0 — 라벨 문자열만. verify_collector_authoritative 를 호출하지 않았음을 명시.
OWNER_PICKUP_PREFIRE_OUTCOME = "pre_fire_owner_pickup_required"
OWNER_PICKUP_PREFIRE_VERDICT = "OWNER_PICKUP_PREFIRE"
# 경로 A 인데 self/owner-key claim 을 동반한 모순 result → 즉시 격리(self-collector 차단 §7).
QREASON_OWNER_PICKUP_SELF_KEY_CLAIM = "owner_pickup_self_key_claim"

ACTIVATION_ENABLED = "enabled"
ACTIVATION_DISABLED = "disabled"

REAL_WAKE_FLAG_REL = "memory/state/p0b_real_wake_enabled"
REAL_WAKE_ENABLED = "enabled"

# ── task-2761: event strategy WIRED_CANDIDATE flag (driver.main 분기 전용, default OFF) ──
#   PR #243(main 진입) 의 scan_live_inbox_once 는 main 미연결(호출자 0) = WIRED=none.
#   본 flag 는 driver.main 에서 scan_live_inbox_once(live inbox) 경로로 분기할지를 토글한다.
#   ★ 2중 게이팅: is_activated(p0b_driver_enabled) AND read_event_strategy_enabled(본 flag)
#   둘 다 ON 일 때만 live 경로. 둘 중 하나라도 OFF(기본) → 기존 scan_once 경로 완전 무변경
#   (현행 동작 100% 보존). 본 flag 는 activation flag/real-wake flag/ACTIVE flag 와 무관한
#   "코드 연결(WIRED_CANDIDATE) 토글"일 뿐 — WIRED ≠ ACTIVE. systemd .path/.service·canary·
#   production activation·callback/result contract·ledger robustness/matching 축 무접촉.
EVENT_STRATEGY_FLAG_REL = "memory/state/p0b_event_strategy_enabled"
EVENT_STRATEGY_ENABLED = "enabled"

# 시간/KST
KST = timezone(timedelta(hours=9))

# pickup_once verdict 문자열 (P0-a 계약 + task-2730 closeout)
_PICKUP_WAKE_BUILT = "WAKE_BUILT"
_PICKUP_CLOSEOUT_DONE = "CLOSEOUT_DONE"   # task-2730: green-path 결정론 closeout
_PICKUP_SKIP_TERMINAL = "SKIP_TERMINAL"
_PICKUP_SKIP_DEDUPE = "SKIP_DEDUPE"
_PICKUP_QUARANTINE = "QUARANTINE"
_PICKUP_PENDING = "PENDING_OWNER_PROOF"

# dedupe ledger event 값 (wake-path / closeout-path 모두 dedupe 대상).
_LEDGER_DEDUPE_EVENTS = ("PICKUP_WAKE_BUILT", "PICKUP_CLOSEOUT_DONE")

# task-2756+2 fix⑤: callback_dedupe_seen 무한 증가 방지(장시간 데몬 메모리 가드).
#   장시간 daemon 호출자는 build_callback_dedupe_seen() 으로 만든 bounded set 을
#   callback_dedupe_seen 으로 주입해야 한다(명시적 contract). maxlen 초과 시 FIFO
#   (가장 오래된 키)로 evict → 무한 증가 0. plain set 주입도 호환되나 unbounded 이므로
#   데몬 경로에서는 반드시 bounded 사용.
CALLBACK_DEDUPE_MAXLEN = 10000


class _BoundedDedupeSet:
    """callback dedupe key 저장용 bounded set (FIFO eviction).

    plain set 과 동일 인터페이스(`key in obj`, `obj.add(key)`)를 제공하되,
    maxlen 초과 시 가장 오래된 키부터 제거하여 무한 증가를 막는다(task-2756+2 ⑤).
    """

    def __init__(self, maxlen: int = CALLBACK_DEDUPE_MAXLEN):
        self._maxlen = max(1, int(maxlen))
        self._d: "OrderedDict[str, None]" = OrderedDict()

    def __contains__(self, key) -> bool:
        return key in self._d

    def add(self, key) -> None:
        if key in self._d:
            self._d.move_to_end(key)  # 최근 사용 → 뒤로(LRU-ish)
            return
        self._d[key] = None
        while len(self._d) > self._maxlen:
            self._d.popitem(last=False)  # 가장 오래된 키 제거(FIFO)

    def __len__(self) -> int:
        return len(self._d)


def build_callback_dedupe_seen(maxlen: int = CALLBACK_DEDUPE_MAXLEN):
    """장시간 데몬용 bounded callback dedupe set 팩토리(task-2756+2 ⑤)."""
    return _BoundedDedupeSet(maxlen)


def _provenance_callback_bridge(
    *,
    result_path: str,
    task_id: str,
    source_result_sha256: str,
    terminal_status: str,
    callback_launch_fn,
    owner_gate_fn,
    callback_owner_key: str,
    callback_executor_key: str,
    callback_chat_id: str,
    callback_at: str,
    callback_request_builder,
    callback_dedupe_seen,
    callback_dispatch_cron_id: str = "",
    callback_normal_collector_cron_id: str = "",
    callback_fallback_cron_id: str = "",
    callback_no_fallback: bool = True,
):
    """provenance ADOPTED collector_result 직후 ANU-owned callback bridge.
    반환: (bridge_verdict: Optional[str], error: Optional[str]).
    callback_launch_fn 미주입 → (None, None) (default OFF, fire 0, 무회귀)."""
    # ① default OFF: fire executor 미주입 → bridge 비활성 (fire 0)
    if callback_launch_fn is None:
        return (None, None)
    # ② fail-closed: fire-capable 인데 owner_gate_fn 부재 → 우회 차단 (fire 0).
    #   ★ task-2756+2: 이는 구조적 미설정(transient 아님)이며 production main() 은 fire executor
    #   주입 시 owner_gate_fn 도 항상 함께 주입하므로 도달 불가(defense-in-depth). 재시도해도
    #   동일 cycle 에선 해소 불가하므로 result 를 stuck 시키지 않는다 — bridge_err=None 반환으로
    #   _move_processed 가 진행되게 한다(영구 stuck·collector_result 중복 write 방지). 관측은
    #   callback_launch_verdict=BRIDGE_OWNER_GATE_BLOCKED 로 유지(fire 0). 반면 callback launch
    #   예외·owner_gate 예외(transient)는 bridge_err 를 set 하여 result 보존(move 0)+재시도.
    if owner_gate_fn is None:
        return (BRIDGE_OWNER_GATE_BLOCKED, None)
    # ③ dedupe: task_id + source_result_sha256 기준 재처리 → callback 0
    dedupe_key = f"{task_id}:{source_result_sha256}"
    if callback_dedupe_seen is not None and dedupe_key in callback_dedupe_seen:
        return (BRIDGE_DEDUPED, None)
    # ④ owner_proof_pickup_gate (주입) PASS 시에만 진행.
    #   ★ task-2756+2 fix④(Med): owner_gate_fn 예외를 graceful capture → fail-closed
    #   (fire 0·driver crash 0). callback_launch_fn 과 동일 패턴. 예외 시 bridge_err 반환
    #   → process_one 이 result 를 inbox 보존(move 0)·다음 cycle 재시도(영구 누락 0).
    try:
        gate = owner_gate_fn(result_path, executor_key=callback_executor_key)
    except Exception as exc:  # noqa: BLE001 — owner_gate 예외 전파 차단(crash 0·fire 0)
        return (BRIDGE_OWNER_GATE_FAIL, f"owner_gate 예외: {exc}")
    gate_ready = bool(getattr(gate, "ready", False)) or (getattr(gate, "status", None) == GATE_READY_STATUS)
    if not gate_ready:
        return (BRIDGE_OWNER_GATE_FAIL, None)
    # ⑤ ANU-owned callback request build (envelope: task_id/terminal_status/source_sha/collector_role=ANU/owner_proof)
    _builder = callback_request_builder or build_anu_owned_callback_request
    prompt = json.dumps({
        "task_id": task_id,
        "terminal_status": terminal_status,
        "source_result_sha256": source_result_sha256,
        "collector_role": COLLECTOR_ROLE_ANU,
        "adopted_via": ADOPTED_VIA_PROVENANCE,
        "owner_proof": "OWNER_ANU",
    }, ensure_ascii=False)
    request = _builder(
        kind="normal",
        task_id=task_id,
        executor_key=callback_executor_key,
        owner_key=callback_owner_key,
        chat_id=callback_chat_id,
        prompt=prompt,
        at=callback_at,
        collector_role=COLLECTOR_ROLE_ANU,
        # ★ task-2756+1 Medium-1 fix: CallbackRequest.cron_id 필드 일관성.
        #   kind="normal" → cron_id = normal_collector_cron_id (helper 내부 패턴 정렬).
        #   미주입 시 request.cron_id 가 None 으로 남아 normal/fallback/dispatch cron_id
        #   필드 군과 drift 발생.
        cron_id=(callback_normal_collector_cron_id or None),
        dispatch_cron_id=callback_dispatch_cron_id,
        normal_collector_cron_id=(callback_normal_collector_cron_id or None),
        fallback_callback_cron_id=(callback_fallback_cron_id or None),
        no_fallback=callback_no_fallback,
    )
    # ⑥ self/executor key 등 → build 가 verdict FAIL (self-callback 차단) → fire 0
    if getattr(request, "verdict", None) != "PASS":
        return (BRIDGE_CALLBACK_OWNER_FAIL, None)
    # ⑦ PASS → fire executor 로 전달. dedupe 는 launch **성공(BRIDGE_FIRED) 후에만** 기록.
    #   ★ task-2756+2 fix②③: callback launch 예외 시 (None, error) 반환 → dedupe 미기록.
    #   process_one 이 bridge_err 를 받아 _move_processed 를 건너뛰므로(fix①) result 는
    #   inbox 에 보존되고 다음 cycle 에 재처리(재시도)된다 — callback 영구 누락 0.
    #   (성공 후 _move_processed 실패와 무관하게 dedupe 가 callback duplicate 0 을 보장.)
    try:
        callback_launch_fn(request)
    except Exception as exc:  # noqa: BLE001 — fire executor 예외 전파 차단(crash 0)
        return (None, f"callback launch 실패: {exc}")
    if callback_dedupe_seen is not None:
        callback_dedupe_seen.add(dedupe_key)
    return (BRIDGE_FIRED, None)


# ── DriverRecord dataclass ────────────────────────────────────────────────────
@dataclass
class DriverRecord:
    ts: str
    result_path: str
    verdict: str
    owner_key_class: Optional[str] = None
    quarantined: bool = False
    quarantine_reason: Optional[str] = None
    fire_cron_id: Optional[str] = None
    retry: int = 0
    error: Optional[str] = None
    driver: str = DRIVER_NAME
    activation: str = ACTIVATION_DISABLED
    # task-2731: pickup_once 가 강제한 ANU-owned callback launch verdict(라벨만·raw key 0).
    callback_launch_verdict: Optional[str] = None
    # task-2744: LEGACY_SKIP_PRESERVED 시 preserve-move 목적지(watched glob 밖). 그 외 None.
    preserved_path: Optional[str] = None

    def to_json(self) -> dict:
        return {
            "ts": self.ts,
            "result_path": self.result_path,
            "verdict": self.verdict,
            "owner_key_class": self.owner_key_class,
            "quarantined": self.quarantined,
            "quarantine_reason": self.quarantine_reason,
            "fire_cron_id": self.fire_cron_id,
            "retry": self.retry,
            "error": self.error,
            "driver": self.driver,
            "activation": self.activation,
            "callback_launch_verdict": self.callback_launch_verdict,
            "preserved_path": self.preserved_path,
        }


# ── activation ────────────────────────────────────────────────────────────────
def read_activation(root: str = CANONICAL_ROOT, *, flag_reader=None) -> str:
    """flag 파일 첫 줄 trim 값 반환. 부재/읽기실패 → "" (=> disabled).
    flag_reader: 테스트 주입용 callable() -> Optional[str] (None 이면 실제 파일 읽기)."""
    if flag_reader is not None:
        try:
            val = flag_reader()
        except Exception:  # noqa: BLE001 — reader 실패 → disabled(fail-closed)
            return ""
        if val is None:
            return ""
        return str(val).splitlines()[0].strip() if str(val).strip() else str(val).strip()
    flag_path = os.path.join(root, ACTIVATION_FLAG_REL)
    try:
        with open(flag_path, "r", encoding="utf-8") as fh:
            first = fh.readline()
    except (OSError, ValueError):
        return ""
    return first.strip()


def is_activated(root: str = CANONICAL_ROOT, *, flag_reader=None) -> bool:
    return read_activation(root, flag_reader=flag_reader) == ACTIVATION_ENABLED


def read_activation_epoch(root: str = CANONICAL_ROOT, *, epoch_reader=None) -> Optional[float]:
    """activation_epoch 마커(memory/state/p0b_activation_epoch) 첫 줄을 unix timestamp(float)로 읽음.
    부재/읽기실패/파싱실패 → None (호출자는 fail-closed 처리). epoch_reader: 테스트 주입 callable()->Optional[str|float]."""
    if epoch_reader is not None:
        try:
            val = epoch_reader()
        except Exception:  # noqa: BLE001
            return None
        if val is None:
            return None
        try:
            return float(str(val).splitlines()[0].strip())
        except (ValueError, IndexError):
            return None
    epoch_path = os.path.join(root, ACTIVATION_EPOCH_REL)
    try:
        with open(epoch_path, "r", encoding="utf-8") as fh:
            first = fh.readline()
    except (OSError, ValueError):
        return None
    try:
        return float(first.strip())
    except ValueError:
        return None


# ── task-2775: limited activation 3경계 헬퍼 ────────────────────────────────
def _atomic_write_flag_disabled(root: str = CANONICAL_ROOT, *, flag_writer=None) -> Optional[str]:
    """p0b_driver_enabled 를 'disabled' 로 atomic write. 성공 None, 실패 시 에러메시지(str).
    flag_writer 주입 시 그 callable(disabled_str)->Optional[str] 사용(테스트용)."""
    if flag_writer is not None:
        try:
            return flag_writer(ACTIVATION_DISABLED)
        except Exception as exc:  # noqa: BLE001
            return f"flag_writer 실패: {exc}"
    flag_path = os.path.join(root, ACTIVATION_FLAG_REL)
    d = os.path.dirname(flag_path)
    try:
        os.makedirs(d, exist_ok=True)
        fd, tmp = tempfile.mkstemp(dir=d, prefix=".p0b_flag_", suffix=".tmp")
    except OSError as exc:
        return f"atomic disabled write 준비 실패: {exc}"
    # ★ task-2775+6 (Gemini MEDIUM): rename 성공 전 어떤 예외(OSError·
    #   KeyboardInterrupt 등 BaseException 포함)에도 temp 파일이 잔존하지 않도록
    #   try/finally 로 cleanup 보장한다. cleanup 은 원래 예외를 삼키지 않는다(전파).
    replaced = False
    try:
        with os.fdopen(fd, "w", encoding="utf-8") as fh:
            fh.write(ACTIVATION_DISABLED + "\n")
            fh.flush()
            os.fsync(fh.fileno())
        # ★ task-2775+7 thread4 (Gemini MEDIUM): p0b_driver_enabled 는 secret 이 아니며
        #   값이 'enabled'/'disabled' 뿐이라 다른 프로세스(예: 모니터·게이트)가 read 할 수
        #   있어야 한다. mkstemp 는 0o600 으로 만들므로 replace 전 0o644 로 보정한다.
        #   ★ 본 권한 정책은 p0b_driver_enabled 한 파일에만 적용(일반화 금지).
        os.chmod(tmp, 0o644)
        os.replace(tmp, flag_path)   # 같은 fs atomic rename
        replaced = True
    except OSError as exc:
        return f"atomic disabled write 실패: {exc}"
    finally:
        if not replaced:
            try:
                os.unlink(tmp)
            except OSError:
                pass
    return None


def _parse_processed_at_to_unix(processed_at) -> Optional[float]:
    """ledger entry 의 processed_at(ISO8601 문자열, 예 '2026-06-18T15:36:29.955784+09:00')
    을 timezone-aware datetime 으로 파싱 → unix epoch(float) 반환 (task-2775+1).
    - tz offset(KST +09:00 등) 보존: datetime.timestamp() 가 offset 반영해 정확한 unix 반환.
    - naive(offset 미보존) 문자열은 보수적으로 제외(None).
    - 부재(None)/타입 불일치/빈 문자열/파싱 실패 → None (활성창 entry 로 오인 금지).
    """
    if not isinstance(processed_at, str) or not processed_at.strip():
        return None
    s = processed_at.strip()
    if s.endswith(("Z", "z")):  # 'Z'(UTC) → fromisoformat 호환 보정
        s = s[:-1] + "+00:00"
    try:
        dt = datetime.fromisoformat(s)
    except (ValueError, TypeError):
        return None
    if dt.tzinfo is None:
        return None  # tz 미보존 → 보수적 제외
    try:
        return dt.timestamp()
    except (OverflowError, OSError, ValueError):
        return None


def _reverse_line_iter(fh, *, chunk_size: int = 8192):
    """바이너리 파일 객체(fh)를 **파일 끝에서부터 chunk 단위로 seek/read** 하며
    완성된 line 을 newest-first(뒤에서부터)로 yield 한다 (task-2775+8).

    ★ 진짜 reverse: 전체 파일을 메모리에 올린 뒤 reversed 하지 않는다.
      - seek(SEEK_END) → chunk_size 바이트씩 **뒤로** 읽는다.
      - 각 chunk 안에서 완성된 line 만 newest-first 로 내보낸다. 한 chunk 의
        맨 앞 조각(parts[0])은 이전(더 과거) chunk 와 이어질 수 있어 보류한다.
      - consumer 가 일찍 break 하면 **더 과거 chunk 를 읽지 않는다**(O(필요분)).
        → active window(파일 끝) 만 읽고 멈추므로 historical entry 전수 스캔 0.
    - fh 는 'rb' 모드 파일 객체. 각 line 은 utf-8 decode(errors='ignore') str 로 yield.
      reversed() 는 **단일 chunk 내부 line 목록**(크기 ≤ chunk_size + 잔여)에만 적용 —
      전체 파일을 뒤집는 것이 아니다.
    - 빈 파일은 yield 0건, 빈 줄로 시작하는 파일은 그 첫 빈 라인도 정확히 yield(task-2775+9).
    """
    fh.seek(0, os.SEEK_END)
    pos = fh.tell()
    if pos == 0:
        return  # 빈 파일 → yield 0건 (첫 조각 오인 방지)
    buf = b""
    while pos > 0:
        read_size = chunk_size if pos >= chunk_size else pos
        pos -= read_size
        fh.seek(pos)
        chunk = fh.read(read_size)
        buf = chunk + buf
        parts = buf.split(b"\n")
        buf = parts[0]  # 맨 앞 미완성 조각 보류(다음 과거 chunk 와 합침)
        for part in reversed(parts[1:]):
            yield part.decode("utf-8", "ignore")
    # ★ task-2775+9 (Gemini MEDIUM driver:536): 루프가 1회 이상 실행됐다면 buf 는
    #   파일의 **첫 라인**이다. 빈 줄로 시작하는 파일도 그 첫 빈 라인을 누락하지 않도록
    #   buf 가 비어 있어도 항상 yield 한다(빈 파일은 위 early-return 으로 이미 분기됨).
    yield buf.decode("utf-8", "ignore")


def _count_active_window_pickups(
    root: str = CANONICAL_ROOT,
    *,
    activation_epoch: Optional[float] = None,
    ledger_count_fn=None,
    max_count: Optional[int] = None,
) -> int:
    """processed_ledger(LIVE_LEDGER_REL) 의 outcome == LIVE_PROCESSED 중
    **현재 limited activation window(activation_epoch 이후) 에 속한 entry만** 카운트 (task-2775+1).

    epoch scope 규칙:
      - activation_epoch(unix float) 가 주어지면 entry 의 processed_at(ISO8601, tz-aware)
        를 unix epoch(float)로 변환 후 processed_at_unix >= activation_epoch 인 entry만 카운트.
      - processed_at 부재 / 파싱 실패 / tz 미보존 entry 는 보수적 제외(활성창 entry 오인 금지).
      - activation_epoch 가 None 이면 **0 반환**(전체 ledger 누계 금지 — 재가동 즉시 false-OFF 방지).
        호출자(gate)가 'active인데 epoch 부재' fail-closed 정책을 별도 처리한다.
    스캔 방향: **newest-first**(reverse chunk iterator, task-2775+8) — active window(파일 끝)만 읽고 max_count 도달 시 historical 미스캔.
    파일 부재/전체 읽기실패 → 0. 개별 라인 파싱 실패 → 그 라인만 skip.
    ledger_count_fn 주입 시 그 callable()->int 사용(테스트용, epoch scope 우회).
    """
    if ledger_count_fn is not None:
        try:
            return int(ledger_count_fn())
        except Exception:  # noqa: BLE001
            return 0
    if activation_epoch is None:
        return 0
    path = os.path.join(root, LIVE_LEDGER_REL)
    count = 0
    try:
        # ★ task-2775+8 (Gemini MEDIUM driver:538): ledger 를 **newest-first**(파일
        #   끝에서 chunk 단위 reverse iterator)로 읽는다. active window(최신 entry)는
        #   파일 끝에 있으므로, 순방향(oldest-first) 스캔처럼 active window 도달 전
        #   historical entry 를 전부 거치지 않는다. max_count 도달 시 더 과거 chunk 를
        #   읽지 않아 O(N) 전수 스캔을 닫는다.
        #   ★ 전체 read 후 reversed 가 아니다 — _reverse_line_iter 가 끝에서부터 chunk
        #     단위로 seek/read 하며 필요한 line 만 yield 한다.
        with open(path, "rb") as fh:
            for line in _reverse_line_iter(fh):
                line = line.strip()
                if not line:
                    continue
                try:
                    obj = json.loads(line)
                except (ValueError, TypeError):
                    continue
                if not (isinstance(obj, dict) and obj.get("outcome") == VERDICT_LIVE_PROCESSED):
                    continue
                pa_unix = _parse_processed_at_to_unix(obj.get("processed_at"))
                if pa_unix is None:
                    continue  # 부재/파싱 실패/naive → 보수적 제외
                # ★ stop 조건은 count >= max_count 뿐. processed_at < activation_epoch 를
                #   만났다는 이유로 break 하지 않는다(단조성 가정 금지 → under-count 방지).
                #   newest-first 라도 N 미만이면 파일 시작까지 읽어 정확 count 를 보장한다.
                if pa_unix >= activation_epoch:
                    count += 1
                    if max_count is not None and count >= max_count:
                        break
    except OSError:
        return 0
    return count


def limited_activation_bound_gate(
    root: str = CANONICAL_ROOT,
    *,
    clock=None,
    now_fn=None,
    kill_exists_fn=None,
    ledger_count_fn=None,
    epoch_reader=None,
    is_activated_fn=None,
    driver_enabled_fn=None,
    flag_writer=None,
    n_max: int = LIMITED_MAX_PICKUPS_N,
    t_minutes: int = LIMITED_WINDOW_MINUTES_T,
    write_evidence: bool = False,
    evidence_path=None,
):
    """driver 진입 최상단 limited-bound gate. 확인 순서 = kill -> N -> T.
    하나라도 걸리면 DriverRecord(NOOP, fire 0, governor 미호출)를 반환한다.
    아무 것도 안 걸리면 None 반환(현행 경로 100% 보존, side-effect 0).

    - kill: KILL_SWITCH_REL 존재 -> 즉시 NOOP_KILL_SWITCH (flag write 없음).
    - N: 활성창 실발사 누계 >= n_max -> p0b_driver_enabled atomic 'disabled' write + NOOP_N_EXCEEDED.
    - T: epoch 존재 AND (now - epoch) >= t_minutes*60 -> atomic 'disabled' write + NOOP_T_EXPIRED.
    """
    clock = clock or (lambda: datetime.now(KST))
    now = (now_fn or time.time)

    # (a) kill switch — 최우선, flag write 없이 즉시 NOOP
    kill_exists = kill_exists_fn if kill_exists_fn is not None else (
        lambda: os.path.exists(os.path.join(root, KILL_SWITCH_REL))
    )
    try:
        tripped_kill = bool(kill_exists())
    except Exception:  # noqa: BLE001
        tripped_kill = False
    if tripped_kill:
        rec = DriverRecord(ts=_now_kst(clock), result_path="",
                           verdict=VERDICT_NOOP_KILL_SWITCH, activation=ACTIVATION_DISABLED)
        if write_evidence:
            _append_evidence(rec, root, evidence_path)
        return rec

    # ★ task-2775+6 (Gemini HIGH): driver(p0b_driver_enabled)가 이미 disabled/OFF면
    #   epoch read / N ledger 전체 scan(_count_active_window_pickups) /
    #   _atomic_write_flag_disabled(fsync) 재쓰기 전에 조기 short-circuit 한다.
    #   driver 가 disabled 로 전환된 뒤 main() 반복 호출마다 ledger 전체를 다시 읽고
    #   disabled flag 를 반복 fsync 쓰는 I/O 병목(Gemini HIGH)을 닫는다.
    #   ★ None 반환(pass-through): NOOP record 조기반환이 아님 — driver OFF 시 main 의
    #     기존 legacy scan_once 라우팅을 그대로 보존한다(default-OFF/legacy/test-2760 무손상).
    #   ★ driver_enabled_fn 미주입(legacy·default-OFF·test-2760 = None) → 본 분기 미발동,
    #     현행 N/T/kill 경로 100% 보존(side-effect 0).
    if driver_enabled_fn is not None:
        try:
            _driver_on = bool(driver_enabled_fn())
        except Exception:  # noqa: BLE001
            _driver_on = False
        if not _driver_on:
            return None

    # epoch 1회 읽기 (N/T 공통, epoch-scope 보정 — task-2775+1)
    epoch = read_activation_epoch(root, epoch_reader=epoch_reader)

    # active 판정 (epoch 부재 + active fail-closed 용).
    # ★ task-2775+1: is_activated_fn 미주입(production main 기본 호출) → active=False 로 보수 처리.
    #   본 fail-closed(epoch 부재 시 자동 disabled)는 호출자가 limited-activation active 를
    #   **명시적으로 주입(is_activated_fn)** 할 때만 발동한다. 이렇게 해야 기존 main()/legacy
    #   라우팅(is_activated=p0b_driver_enabled ON + epoch 부재 → scan_once; test-2760 등)을
    #   무손상으로 보존한다. over-count 위험은 이미 epoch-scope(_count_active_window_pickups
    #   activation_epoch=None → 0) 로 제거되었고, ACTIVE=false 인 production 에는 영향 0.
    #   (fail-closed 정책 자체는 본 게이트에 정의 + is_activated_fn 주입 테스트로 고정.)
    if is_activated_fn is not None:
        try:
            active = bool(is_activated_fn())
        except Exception:  # noqa: BLE001
            active = False
    else:
        active = False

    # (b) epoch 부재 정책:
    #   - active(명시 주입) + epoch 부재 → fail-closed(자동 disabled + NOOP_ACTIVE_EPOCH_MISSING).
    #     전체 ledger 누계/조용한 통과 금지.
    #   - not active + epoch 부재 → 기존 default-OFF 경로 보존: N/T 미평가 → None
    #     (main 의 is_activated → NOOP_DISABLED 경로로 이행, 무손상).
    if epoch is None:
        if active:
            err = _atomic_write_flag_disabled(root, flag_writer=flag_writer)
            rec = DriverRecord(ts=_now_kst(clock), result_path="",
                               verdict=VERDICT_NOOP_ACTIVE_EPOCH_MISSING,
                               activation=ACTIVATION_DISABLED, error=err)
            if write_evidence:
                _append_evidence(rec, root, evidence_path)
            return rec
        return None

    # (c) N — epoch-scope 활성창 누계 상한 도달 시 자동 disabled
    count = _count_active_window_pickups(
        root, activation_epoch=epoch, ledger_count_fn=ledger_count_fn, max_count=n_max)
    if count >= n_max:
        err = _atomic_write_flag_disabled(root, flag_writer=flag_writer)
        rec = DriverRecord(ts=_now_kst(clock), result_path="",
                           verdict=VERDICT_NOOP_N_EXCEEDED, activation=ACTIVATION_DISABLED,
                           error=err)
        if write_evidence:
            _append_evidence(rec, root, evidence_path)
        return rec

    # (d) T — 활성 시간창 만료 시 자동 disabled
    #   ★ task-2775+7 thread5 (Gemini MEDIUM): elapsed 계산 실패(now/epoch 손상)를
    #     elapsed=-1.0 로 통과(fail-OPEN)시키면 fire-capable limited 상태에서 T 검사를
    #     건너뛰어 안전모델을 위반한다. → fail-closed(auto-disabled)로 수렴시킨다.
    #     N 경로(count>=n_max)·active+epoch부재 경로와 동일 안전정책: fire 0, flag auto-disabled.
    try:
        elapsed = float(now()) - float(epoch)
    except (TypeError, ValueError, OverflowError):
        err = _atomic_write_flag_disabled(root, flag_writer=flag_writer)
        rec = DriverRecord(ts=_now_kst(clock), result_path="",
                           verdict=VERDICT_NOOP_ACTIVE_EPOCH_INVALID,
                           activation=ACTIVATION_DISABLED, error=err)
        if write_evidence:
            _append_evidence(rec, root, evidence_path)
        return rec
    if elapsed >= t_minutes * 60:
        err = _atomic_write_flag_disabled(root, flag_writer=flag_writer)
        rec = DriverRecord(ts=_now_kst(clock), result_path="",
                           verdict=VERDICT_NOOP_T_EXPIRED, activation=ACTIVATION_DISABLED,
                           error=err)
        if write_evidence:
            _append_evidence(rec, root, evidence_path)
        return rec

    return None


# ── real-wake activation ──────────────────────────────────────────────────────
def read_real_wake_enabled(root: str = CANONICAL_ROOT, *, flag_reader=None) -> bool:
    """real-wake activation flag(memory/state/p0b_real_wake_enabled) 첫 줄 trim == "enabled" 일 때만 True.
    부재/읽기실패/그 외 값 → False (fail-closed). flag_reader: 테스트 주입 callable()->Optional[str]."""
    if flag_reader is not None:
        try:
            val = flag_reader()
        except Exception:  # noqa: BLE001
            return False
        if val is None:
            return False
        return str(val).splitlines()[0].strip() == REAL_WAKE_ENABLED if str(val).strip() else False
    flag_path = os.path.join(root, REAL_WAKE_FLAG_REL)
    try:
        with open(flag_path, "r", encoding="utf-8") as fh:
            first = fh.readline()
    except (OSError, ValueError):
        return False
    return first.strip() == REAL_WAKE_ENABLED


# ── task-2761: event strategy WIRED_CANDIDATE flag reader ─────────────────────
def read_event_strategy_enabled(root: str = CANONICAL_ROOT, *, flag_reader=None) -> bool:
    """event strategy wiring flag(memory/state/p0b_event_strategy_enabled) 첫 줄 trim == "enabled" 일 때만 True.
    부재/읽기실패/그 외 값 → False (fail-closed, default OFF). flag_reader: 테스트 주입 callable()->Optional[str].

    ★ driver.main 의 live-inbox 분기 토글 전용. 이 flag 단독으로는 fire 0 — main 이 is_activated
    (p0b_driver_enabled) 와 AND(2중 게이팅) 로 묶고, live 경로(scan_live_inbox_once)도 내부에
    activation gate 를 유지한다. 이 함수는 flag 읽기만 — real pickup/move/callback 부작용 0."""
    if flag_reader is not None:
        try:
            val = flag_reader()
            if val is None:
                return False
            sval = str(val)  # ★ val 처리(str 변환 포함)도 try 안에서 — __str__ 예외 등 → fail-closed
            return sval.splitlines()[0].strip() == EVENT_STRATEGY_ENABLED if sval.strip() else False
        except Exception:  # noqa: BLE001 — reader/val 처리 실패 → disabled(fail-closed)
            return False
    flag_path = os.path.join(root, EVENT_STRATEGY_FLAG_REL)
    try:
        with open(flag_path, "r", encoding="utf-8") as fh:
            first = fh.readline()
    except (OSError, ValueError):
        return False
    return first.strip() == EVENT_STRATEGY_ENABLED


def read_provenance_path_scope_enabled(
    root: str = CANONICAL_ROOT,
    *,
    flag_reader=None,        # callable()->Optional[str] (조건1 flag 첫줄)
    epoch_reader=None,       # read_activation_epoch 에 위임 (조건2)
    scope_id_reader=None,    # callable()->Optional[str] (조건3 scope_id 첫줄)
    inbox_lister=None,       # callable()->list[str] (조건4 inbox target 목록)
) -> bool:
    """scope-limited provenance-path 활성 여부. 4개 AND 조건 — 하나라도 부재면 False(default OFF).
    broad/global ON 구조적 불가(4개 동시 ON 은 회장 승인 scope-limited pilot 에서만).

    조건1: p0b_provenance_path_enabled flag 첫줄 trim == "enabled"
    조건2: p0b_activation_epoch 마커 존재(read_activation_epoch is not None) = scope epoch
    조건3: p0b_provenance_scope_id 파일 첫줄 trim 비어있지 않음(scope_id)
    조건4: inbox(memory/events/p0b_inbox/task-*.result.json) target ≥ 1
    예외/읽기실패 → 보수적으로 False(fail-closed).
    """
    try:
        # ── 조건1: provenance-path flag ────────────────────────────────────────
        if flag_reader is not None:
            try:
                val = flag_reader()
            except Exception:  # noqa: BLE001
                return False
            if val is None:
                return False
            cond1 = str(val).splitlines()[0].strip() == PROVENANCE_PATH_FLAG_ENABLED if str(val).strip() else False
        else:
            flag_path = os.path.join(root, PROVENANCE_PATH_FLAG_REL)
            try:
                with open(flag_path, "r", encoding="utf-8") as fh:
                    first = fh.readline()
            except (OSError, ValueError):
                return False
            cond1 = first.strip() == PROVENANCE_PATH_FLAG_ENABLED
        if not cond1:
            return False

        # ── 조건2: p0b_activation_epoch 마커 존재 ──────────────────────────────
        epoch = read_activation_epoch(root, epoch_reader=epoch_reader)
        if epoch is None:
            return False

        # ── 조건3: p0b_provenance_scope_id 비어있지 않음 ───────────────────────
        if scope_id_reader is not None:
            try:
                sid_val = scope_id_reader()
            except Exception:  # noqa: BLE001
                return False
            if sid_val is None:
                return False
            cond3 = bool(str(sid_val).strip())
        else:
            sid_path = os.path.join(root, PROVENANCE_SCOPE_ID_REL)
            try:
                with open(sid_path, "r", encoding="utf-8") as fh:
                    sid_first = fh.readline()
            except (OSError, ValueError):
                return False
            cond3 = bool(sid_first.strip())
        if not cond3:
            return False

        # ── 조건4: inbox에 target result.json 이 1개 이상 ─────────────────────
        if inbox_lister is not None:
            try:
                targets = [p for p in list(inbox_lister()) if is_target(p)]
            except Exception:  # noqa: BLE001
                return False
        else:
            try:
                pattern = os.path.join(root, INBOX_DIR_REL, RESULT_GLOB)
                targets = [p for p in sorted(glob.glob(pattern)) if is_target(p)]
            except Exception:  # noqa: BLE001
                return False
        if not targets:
            return False

        return True
    except Exception:  # noqa: BLE001 — 어떤 예외든 False(fail-closed)
        return False


def build_launcher_fn(
    root: str = CANONICAL_ROOT,
    *,
    real_wake_reader=None,
    sealed_key_loader=None,
    launch_wake_fn=None,
):
    """real-wake flag on + sealed key 존재 시에만 sealed-key 기반 launcher_fn 반환.
    그 외(flag off / sealed key 부재) → None (현행 보존, launcher 함수 호출 0, real wake 0).

    fail-closed:
      - real-wake flag != "enabled" → None (sealed loader 조차 호출 안 함).
      - sealed key 부재(None/빈값) → None (real wake 0).
    verifier 는 sealed key 와의 상수시간 비교(hmac.compare_digest). 반환 launcher_fn 은
    functools.partial(launch_wake, dry_run=False, anu_key_verifier=verifier).
    """
    if not read_real_wake_enabled(root, flag_reader=real_wake_reader):
        return None
    loader = sealed_key_loader or _default_sealed_key_loader
    try:
        sealed_key = loader()
    except Exception:  # noqa: BLE001 — loader 실패 → fail-closed
        return None
    if not sealed_key:
        return None
    sealed_key_str = str(sealed_key)

    def _verifier(candidate) -> bool:
        try:
            return hmac.compare_digest(str(candidate), sealed_key_str)
        except Exception:  # noqa: BLE001
            return False

    lw = launch_wake_fn or launch_wake
    return functools.partial(lw, dry_run=False, anu_key_verifier=_verifier)


# ── target 판정 ────────────────────────────────────────────────────────────────
def is_target(path: str) -> bool:
    """final task-*.result.json 만 True. basename 이 'task-' 로 시작 + '.result.json' 으로 끝.
    '.result.json.tmp-...'/partial/.md/.jsonl/다른 marker → False."""
    if not path:
        return False
    base = os.path.basename(str(path))
    return base.startswith("task-") and base.endswith(".result.json")


# ── 내부 helper ────────────────────────────────────────────────────────────────
def _now_kst(clock) -> str:
    return clock().isoformat()


def _envelope_claim_class(envelope: dict) -> str:
    """envelope claim 으로 SELF/FOREIGN 라벨 추정 (판정 아님, 라벨링만)."""
    try:
        if bool(envelope.get("self_key_used")):
            return OKC_SELF
    except AttributeError:
        return OKC_FOREIGN
    return OKC_FOREIGN


def _collision_safe_dest(dest_dir: str, basename: str) -> str:
    """dest_dir/basename. 이미 존재하면 .{ms타임스탬프}[-n] suffix 로 충돌 회피."""
    dest = os.path.join(dest_dir, basename)
    if not os.path.exists(dest):
        return dest
    ts = str(int(time.time() * 1000))
    cand = os.path.join(dest_dir, f"{basename}.{ts}")
    n = 0
    while os.path.exists(cand):
        n += 1
        cand = os.path.join(dest_dir, f"{basename}.{ts}-{n}")
    return cand


def _move_processed(path: str, root: str, processed_dir: Optional[str] = None) -> Optional[str]:
    """terminal(WAKE_BUILT/PICKUP_SKIP) result 파일을 watched 밖 processed 디렉토리로
    atomic 이동. 성공 시 None, 실패 시 에러 메시지(str) 반환(fail-safe: 크래시 0).
    os.replace(같은 fs atomic) 우선, 실패 시 shutil.move fallback."""
    pdir = processed_dir or os.path.join(root, PROCESSED_DIR_REL)
    try:
        os.makedirs(pdir, exist_ok=True)
        dest = _collision_safe_dest(pdir, os.path.basename(path))
    except OSError as exc:
        return f"processed move 실패: {exc}"
    try:
        os.replace(path, dest)
        return None
    except OSError:
        try:
            shutil.move(path, dest)
            return None
        except (OSError, shutil.Error) as exc:
            return f"processed move 실패: {exc}"


def _quarantine_move(path: str, root: str, quarantine_dir: Optional[str]) -> Optional[str]:
    """result 파일을 quarantine 디렉토리로 이동. 실패 시 예외 메시지 반환(None=성공)."""
    qdir = quarantine_dir or os.path.join(root, QUARANTINE_DIR_REL)
    try:
        os.makedirs(qdir, exist_ok=True)
        dest = _collision_safe_dest(qdir, os.path.basename(path))
        shutil.move(path, dest)
        return None
    except (OSError, shutil.Error) as exc:
        return f"quarantine move 실패: {exc}"


def _append_evidence(record: DriverRecord, root: str, evidence_path: Optional[str]) -> None:
    """DriverRecord.to_json() 한 줄 JSON append. ANU key literal 절대 미기록."""
    ev = evidence_path or os.path.join(root, EVIDENCE_JSONL_REL)
    try:
        os.makedirs(os.path.dirname(ev), exist_ok=True)
        with open(ev, "a", encoding="utf-8") as fh:
            fh.write(json.dumps(record.to_json(), ensure_ascii=False) + "\n")
            fh.flush()
            os.fsync(fh.fileno())
    except OSError:
        pass  # evidence 기록 실패는 비치명 (fail-safe)


def _dedupe_hit(task_id: str, ledger_path: Optional[str], root: str) -> bool:
    """dedupe ledger 에 동일 task_id 의 PICKUP_WAKE_BUILT 항목 존재 여부."""
    ledger = ledger_path or os.path.join(
        root, "memory", "events", "callback_4tuple_index.jsonl"
    )
    if not os.path.isfile(ledger):
        return False
    try:
        with open(ledger, "r", encoding="utf-8") as fh:
            for line in fh:
                line = line.strip()
                if not line:
                    continue
                try:
                    entry = json.loads(line)
                except (ValueError, TypeError):
                    continue
                if (
                    isinstance(entry, dict)
                    and entry.get("event") in _LEDGER_DEDUPE_EVENTS
                    and entry.get("task_id") == task_id
                ):
                    return True
    except (OSError, ValueError):
        return False
    return False


def _legacy_cutoff_check(path, *, activation_epoch, stat_fn=None):
    """legacy cutoff 판정. skip 해야 하면 reason(str) 반환, 정상 진행이면 None.
    - activation_epoch is None (부재/불명확) → fail-open 금지 → skip("epoch_absent").
    - stat 실패 → skip("legacy_stat_fail") (보수적 skip — canonical 무수정).
    - result mtime < activation_epoch → skip("pre_activation_epoch") (legacy).
    - mtime >= activation_epoch → None (post-activation, 정상 decision path)."""
    if activation_epoch is None:
        return "epoch_absent"
    stat_fn = stat_fn or os.stat
    try:
        st = stat_fn(path)
    except OSError:
        return "legacy_stat_fail"
    if st.st_mtime < activation_epoch:
        return "pre_activation_epoch"
    return None


# ── task-2744: legacy_skip preserve-move (watched glob 밖) ───────────────────────
def _sha256_file(path: str, *, hash_fn=None) -> str:
    """파일 내용의 sha256 hexdigest. hash_fn(path)->str 주입 시 그것을 사용(테스트 corruption 모사)."""
    if hash_fn is not None:
        return hash_fn(path)
    h = hashlib.sha256()
    with open(path, "rb") as fh:
        for chunk in iter(lambda: fh.read(65536), b""):
            h.update(chunk)
    return h.hexdigest()


def _legacy_skip_task_id(path: str) -> str:
    """preserve-move manifest 용 task_id 를 basename 에서 추출(content 미신뢰 — legacy 는
    schema-invalid 일 수 있음). 'task-2309.result.json' → 'task-2309'. 형식 불명 시 basename 전체."""
    base = os.path.basename(str(path))
    if base.startswith("task-") and base.endswith(".result.json"):
        return base[: -len(".result.json")]
    return base


@dataclass
class LegacySkipMoveResult:
    """preserve-move 결과. ok=True 시 preserved_path/manifest_entry 유효, error=None.
    ok=False 시 원본 보존(watched 경로 유지)·preserved_path None·error(str) set."""
    ok: bool
    preserved_path: Optional[str] = None
    manifest_entry: Optional[dict] = None
    error: Optional[str] = None


def _append_legacy_skip_manifest(entry: dict, root: str, manifest_path: Optional[str]) -> Optional[str]:
    """legacy_skip_migration.jsonl 에 entry 한 줄 append-only(JSON). 성공 None, 실패 시 에러 str.
    ANU key literal 절대 미기록(원본 경로/sha/타임스탬프 라벨만)."""
    mpath = manifest_path or os.path.join(root, LEGACY_SKIP_MANIFEST_REL)
    try:
        os.makedirs(os.path.dirname(mpath), exist_ok=True)
        with open(mpath, "a", encoding="utf-8") as fh:
            fh.write(json.dumps(entry, ensure_ascii=False) + "\n")
            fh.flush()
            os.fsync(fh.fileno())
    except OSError as exc:
        return f"manifest append 실패: {exc}"
    return None


def _legacy_skip_preserve_move(
    path: str,
    root: str,
    *,
    reason: str,
    clock,
    legacy_skip_dir: Optional[str] = None,
    manifest_path: Optional[str] = None,
    actor: str = LEGACY_SKIP_ACTOR,
    task_id: Optional[str] = None,
    copy_fn=None,
    hash_fn=None,
    stat_fn=None,
) -> LegacySkipMoveResult:
    """pre_activation_epoch legacy result 를 watched glob 밖(LEGACY_SKIP_DIR_REL)으로 preserve-move.

    불변식:
      - 삭제 0 / 내용변경 0: 원본을 copy(shutil.copy2, mtime/metadata 보존)한 뒤 sha256
        before/after 가 **일치**할 때에만 원본 watched 경로를 비운다(content 보존 move).
      - sha256 mismatch / copy IO 실패 → **원본 보존**(watched 경로 유지)·preserved 사본 정리·
        manifest 미기록·ok=False (explicit failure). 데이터 손실/오염 0.
      - 원본 제거는 sha256 검증 통과 후 수행한다. 제거 실패(권한 등 지속 실패 가능) 시 사본을
        정리하고 원본을 보존(ok=False) → 매 cycle 새 사본/manifest 무한 누적(디스크 고갈) 0.
      - manifest(append-only)는 **원본 제거 성공 후** append 한다 → watched 를 떠난 파일만
        manifest 항목을 가진다(orphan 누적 0). append 실패(rare) 시 dest→original 복원으로
        가역 환원(manifest 없는 이동 금지).
    순서: sha_before → copy → sha_after(verify) → remove original → manifest append.
    """
    ldir = legacy_skip_dir or os.path.join(root, LEGACY_SKIP_DIR_REL)
    copy_fn = copy_fn or shutil.copy2
    tid = task_id or _legacy_skip_task_id(path)
    # 0) 원본 sha256 + mtime
    try:
        sha_before = _sha256_file(path, hash_fn=hash_fn)
    except OSError as exc:
        return LegacySkipMoveResult(ok=False, error=f"sha256(before) 실패: {exc}")
    stat_fn = stat_fn or os.stat
    try:
        mtime = float(stat_fn(path).st_mtime)
    except OSError as exc:
        return LegacySkipMoveResult(ok=False, error=f"stat 실패: {exc}")
    # 1) copy → dest (충돌 안전 경로)
    try:
        os.makedirs(ldir, exist_ok=True)
        dest = _collision_safe_dest(ldir, os.path.basename(path))
        copy_fn(path, dest)
    except (OSError, shutil.Error) as exc:
        return LegacySkipMoveResult(ok=False, error=f"copy 실패: {exc}")
    # 2) sha256(after) verify — mismatch 시 사본 정리·원본 보존
    try:
        sha_after = _sha256_file(dest, hash_fn=hash_fn)
    except OSError as exc:
        _safe_unlink(dest)
        return LegacySkipMoveResult(ok=False, error=f"sha256(after) 실패: {exc}")
    if sha_after != sha_before:
        _safe_unlink(dest)
        return LegacySkipMoveResult(
            ok=False,
            error=(f"sha256 mismatch: before={sha_before} after={sha_after} "
                   f"→ abort(원본 보존, 사본 정리)"),
        )
    entry = {
        "original_path": path,
        "preserved_path": dest,
        "sha256_before": sha_before,
        "sha256_after": sha_after,
        "reason": reason,
        "task_id": tid,
        "mtime": mtime,
        "migrated_at": clock().isoformat(),
        "actor": actor,
        "reversible": True,
    }
    # 3) 원본 제거(watched inbox 비움) — content 는 dest 에 이미 보존됨.
    #    ★ os.remove 가 지속 실패(권한 등)하면 매 cycle 마다 _collision_safe_dest 가 새 경로를
    #    만들어 사본/manifest 가 무한 누적(디스크 고갈)될 수 있다. 그러므로 제거 실패 시 **사본을
    #    정리**하고 원본을 보존(watched 잔존)한 채 ok=False 로 단락한다 → 누적 0·데이터 손실 0.
    try:
        os.remove(path)
    except OSError as exc:
        _safe_unlink(dest)
        return LegacySkipMoveResult(
            ok=False,
            error=f"원본 제거 실패(사본 정리됨, 원본 보존): {exc}",
        )
    # 4) manifest append (원본 제거 성공 후 — append-only). watched 를 떠난 파일만 기록되어
    #    orphan(원본 잔존 + manifest) 누적이 없다. append 실패(rare) 시 dest→original 복원으로
    #    가역 환원(원본을 inbox 로 되돌림)하여 manifest 없는 이동을 만들지 않는다.
    man_err = _append_legacy_skip_manifest(entry, root, manifest_path)
    if man_err:
        try:
            shutil.copy2(dest, path)   # 원본 복원(가역)
            _safe_unlink(dest)
        except (OSError, shutil.Error):
            # 복원 실패 시 사본은 보존(데이터 안전) — 가역 정보는 caller 에 노출.
            return LegacySkipMoveResult(ok=False, preserved_path=dest,
                                        error=f"{man_err}; 원본 복원 실패(사본 보존)")
        return LegacySkipMoveResult(ok=False, error=f"{man_err}; 원본 복원됨")
    return LegacySkipMoveResult(ok=True, preserved_path=dest, manifest_entry=entry, error=None)


def _safe_unlink(path: str) -> None:
    try:
        os.remove(path)
    except OSError:
        pass


def rollback_legacy_skip(entry: dict, *, hash_fn=None, copy_fn=None) -> tuple:
    """manifest entry 기준 preserve-move 가역 복구. preserved_path → original_path 로 환원.

    (ok: bool, error: Optional[str]) 반환. 가역성 검증:
      - preserved_path 의 현재 sha256 == manifest.sha256_after 이어야 한다(미일치 → abort).
      - original_path 가 이미 존재하면 abort(덮어쓰기 금지 — content 손실 방지).
      - copy 후 복구된 original 의 sha256 == sha256_before 재검증(미일치 → 복구본 정리·abort).
    성공 시 preserved 사본 제거(완전 환원). 실패 시 원본/사본 보존(데이터 손실 0)."""
    required = ("original_path", "preserved_path", "sha256_before", "sha256_after")
    for k in required:
        if not entry.get(k):
            return (False, f"manifest entry 필수 필드 누락: {k}")
    orig = entry["original_path"]
    preserved = entry["preserved_path"]
    copy_fn = copy_fn or shutil.copy2
    if not os.path.exists(preserved):
        return (False, f"preserved_path 부재: {preserved}")
    if os.path.exists(orig):
        return (False, f"original_path 이미 존재(덮어쓰기 금지): {orig}")
    try:
        cur = _sha256_file(preserved, hash_fn=hash_fn)
    except OSError as exc:
        return (False, f"preserved sha256 실패: {exc}")
    if cur != entry["sha256_after"]:
        return (False, f"preserved 무결성 손상: {cur} != manifest.sha256_after")
    try:
        os.makedirs(os.path.dirname(orig), exist_ok=True)
        copy_fn(preserved, orig)
    except (OSError, shutil.Error) as exc:
        return (False, f"복구 copy 실패: {exc}")
    try:
        restored = _sha256_file(orig, hash_fn=hash_fn)
    except OSError as exc:
        _safe_unlink(orig)
        return (False, f"복구본 sha256 실패: {exc}")
    if restored != entry["sha256_before"]:
        _safe_unlink(orig)
        return (False, f"복구본 무결성 미일치: {restored} != manifest.sha256_before")
    _safe_unlink(preserved)
    return (True, None)


def _check_readiness(
    path: str,
    *,
    clock,
    stable_sec: float = STABLE_SEC,
    retries: int = STABILITY_RETRIES,
    interval: float = STABILITY_INTERVAL_SEC,
    sleep_fn=None,
    stat_fn=None,
) -> tuple:
    """write race 방어 readiness 판정. (ready: bool, reason: str) 반환.

    판정 규칙:
    - 파일 stat 실패(존재X 등) → (False, "stat_fail"): DEFER, 다음 트리거 재평가.
    - mtime 이 now - stable_sec 이내(최근 생성/수정) → (False, "recent_mtime"): writer 미완 가능 → DEFER.
    - size/mtime 안정성: 짧은 간격으로 최대 retries 회 stat 하여 (size, mtime) 불변 확인.
      마지막 두 샘플이 다르면 (False, "unstable"): 아직 쓰는 중 → DEFER.
    - 위 모두 통과(aged + stable) → (True, "ready").
    단순 무한 sleep 없음: 총 window = (retries-1) * interval ≤ 약 0.6s.
    """
    stat_fn = stat_fn or os.stat
    sleep_fn = sleep_fn or time.sleep

    samples = []
    for attempt in range(max(1, retries)):
        try:
            st = stat_fn(path)
        except OSError:
            return (False, "stat_fail")
        samples.append((st.st_size, st.st_mtime))
        # 직전 샘플과 동일하면 안정화 → 조기 종료
        if len(samples) >= 2 and samples[-1] == samples[-2]:
            break
        if attempt < retries - 1:
            sleep_fn(interval)

    mtime = samples[-1][1]
    age = clock().timestamp() - mtime
    if age < stable_sec:
        return (False, "recent_mtime")
    if len(samples) >= 2 and samples[-1] != samples[-2]:
        return (False, "unstable")
    return (True, "ready")


# ── terminal marker 확인 후에만 move (task-2730 HIGH-2 remediation) ──────────
_MARKER_ATTR_MISSING = object()


def _marker_confirmed(res, pv: Optional[str]) -> bool:
    """terminal marker(.pickup.done) 가 확인되어 result 파일 이동(processed)이 안전한가.

    - res 에 marker_path 속성 부재(legacy/mock pickup_fn) → True (현행 이동 동작 보존).
    - SKIP_TERMINAL → 디스크에 marker 이미 존재(종결) → True.
    - 그 외(CLOSEOUT_DONE / WAKE_BUILT / SKIP_DEDUPE) → marker_path 가 set 이어야 True.
      marker write 실패(marker_path None) → 이동 보류 → 다음 cycle pickup_once recovery
      가 marker 멱등 보강 후 이동(ledger dedupe 로 wake/relay 재발사 0). closeout 의
      종결 sentinel(marker) 이 확정되기 전에는 watched 밖으로 이동하지 않는다."""
    mp = getattr(res, "marker_path", _MARKER_ATTR_MISSING)
    if mp is _MARKER_ATTR_MISSING:
        return True
    if pv == _PICKUP_SKIP_TERMINAL:
        return True
    return mp is not None


_MOVE_DEFERRED_ERR = ("terminal marker 미확정 — 이동 보류(다음 cycle pickup_once "
                      "recovery 로 marker 보강 후 이동; ledger dedupe 로 재발사 0).")


# ── process_one ────────────────────────────────────────────────────────────────
def process_one(
    path: str,
    *,
    root: str = CANONICAL_ROOT,
    pickup_fn=None,            # 기본 dispatch.anu_result_pickup_runner.pickup_once
    launcher_fn=None,          # 기본 None = 현행 보존(surface only). 주입 시 real wake 결선.
    callback_launch_fn=None,   # task-2731: 기본 None = 현행 보존. 주입 시 pickup_once closeout→ANU-owned callback launch 강제+audit.
    owner_gate_fn=None,        # task-2741-r1: 기본 None = 현행 보존. 주입 시 pickup_once 가 fire 전 owner_proof_pickup_gate 강제(우회 0). main() 이 bound gate 주입.
    spawn_governor_fn=None,    # task-2774+3: 기본 None = 현행 보존. pickup_once 로 그대로 전달(spawn governor 평가는 opt-in, flag OFF 면 builder None).
    relay_fn=None,             # task-2730: 기본 None = 현행 보존. 주입 시 relay-path(WAKE_BUILT) 2-tier terminal relay 결선.
    verify_fn=None,            # 기본 verify_collector_authoritative
    contract_split: bool = OWNER_PICKUP_CONTRACT_SPLIT_DEFAULT,  # task-2746: 기본 OFF=기존 무조건 verify 보존. ON=fire-state 경로 A/B 분기.
    probe=None,                # cron-history probe (owner proof 용). 기본 None
    clock=None,                # 기본 lambda: datetime.now(KST)
    executor_key: str = "",
    ledger_path: Optional[str] = None,
    quarantine_dir: Optional[str] = None,
    processed_dir: Optional[str] = None,
    write_evidence: bool = False,   # process_one 단독 호출 시 evidence append 여부
    evidence_path: Optional[str] = None,
    stable_sec: Optional[float] = None,
    readiness_retries: Optional[int] = None,
    readiness_interval: Optional[float] = None,
    sleep_fn=None,
    activation_epoch: Optional[float] = None,
    legacy_cutoff: bool = False,
    legacy_preserve_move: bool = False,   # task-2744: 기본 False = 현행(NOOP_LEGACY_SKIP move 0).
    legacy_skip_dir: Optional[str] = None,
    legacy_manifest_path: Optional[str] = None,
    legacy_copy_fn=None,
    legacy_hash_fn=None,
    stat_fn=None,
    provenance_path_enabled: bool = PROVENANCE_PATH_ENABLED_DEFAULT,  # opt-in, default OFF
    provenance_fn=None,              # default validate_executor_provenance (테스트 주입)
    collector_writer_fn=None,        # default write_collector_result (테스트 주입)
    capability_reader=None,          # provenance용 capability snapshot reader
    provenance_ledger_reader=None,   # provenance dedupe용 ledger reader
    forbidden_key_literals: tuple = (),  # provenance raw-key scan용
    # task-2756: provenance ADOPTED → ANU-owned callback bridge 파라미터 (모두 default OFF, 무회귀).
    callback_owner_key: str = "",       # 주입 ANU key. bridge fire 시 owner_key 로 사용.
    callback_executor_key: str = "",    # executor self-key (self-callback 차단 검증용).
    callback_chat_id: str = "",         # callback chat_id.
    callback_at: str = "",              # callback 스케줄 at.
    callback_request_builder=None,      # default None → 내부에서 build_anu_owned_callback_request 사용. 테스트는 mock 주입 가능.
    callback_dedupe_seen=None,          # default None. set 류 주입 시 dedupe 상태 공유. None 이면 dedupe 비활성.
    callback_dispatch_cron_id: str = "",          # task-2756: bridge build 4-tuple — dispatch cron id
    callback_normal_collector_cron_id: str = "",  # task-2756: bridge build 4-tuple — normal collector cron id
    callback_fallback_cron_id: str = "",          # task-2756: bridge build 4-tuple — fallback cron id (no_fallback 시 무시)
    callback_no_fallback: bool = True,            # task-2756: bridge build — recovery-only 기본 no_fallback=True(safety path)
) -> DriverRecord:
    """단일 result.json 처리. activation 은 호출자(scan_once)가 이미 보장 — 여기선 target/6조건만.

    task-2730: pickup verdict CLOSEOUT_DONE(green) → VERDICT_CLOSEOUT_DONE(launcher/relay
    미호출, move_processed). WAKE_BUILT(relay-path) → 기존 launcher 분기 + relay_fn 주입 시
    2-tier terminal relay. closeout write 는 pickup_once 소유(ANCHOR-A) — process_one 미소유."""
    pickup_fn = pickup_fn or pickup_once
    verify_fn = verify_fn or verify_collector_authoritative
    clock = clock or (lambda: datetime.now(KST))

    ts = _now_kst(clock)

    def _emit(rec: DriverRecord) -> DriverRecord:
        if write_evidence:
            _append_evidence(rec, root, evidence_path)
        return rec

    def _quarantine(reason: str, owner_class: Optional[str] = None,
                    extra_err: Optional[str] = None) -> DriverRecord:
        move_err = _quarantine_move(path, root, quarantine_dir)
        err = extra_err
        if move_err:
            err = (err + "; " + move_err) if err else move_err
        return _emit(DriverRecord(
            ts=ts,
            result_path=path,
            verdict=VERDICT_QUARANTINE,
            owner_key_class=owner_class,
            quarantined=True,
            quarantine_reason=reason,
            error=err,
            activation=ACTIVATION_ENABLED,
        ))

    # ── target 아니면 즉시 반환 ───────────────────────────────────────────────
    if not is_target(path):
        return _emit(DriverRecord(
            ts=ts,
            result_path=path,
            verdict=VERDICT_NOOP_NOT_TARGET,
            activation=ACTIVATION_ENABLED,
        ))

    # ── D2 legacy cutoff (activation_epoch 기준) ──────────────────────────────
    # legacy_cutoff 활성 시: activation_epoch 이전 생성 result 는 canonical 무수정 skip.
    #   epoch 부재/불명확 → fail-open 금지(epoch_absent skip). move 0·wake 0·quarantine 0.
    if legacy_cutoff:
        skip_reason = _legacy_cutoff_check(path, activation_epoch=activation_epoch, stat_fn=stat_fn)
        if skip_reason is not None:
            # ── task-2744: pre_activation_epoch → preserve-move (watched glob 밖) ──
            #   legacy_preserve_move 활성 + reason==pre_activation_epoch 일 때만 preserve-move.
            #   epoch_absent/legacy_stat_fail 은 안전 분류 불가 → 현행 NOOP_LEGACY_SKIP(move 0)
            #   보수 유지. preserve-move 는 watched inbox 를 비워(PathExistsGlob false 化)
            #   재트리거 폭주를 구조적으로 제거하되, 삭제0·내용변경0·sha256 검증·manifest 가역.
            if legacy_preserve_move and skip_reason == LEGACY_SKIP_REASON:
                mv = _legacy_skip_preserve_move(
                    path,
                    root,
                    reason=skip_reason,
                    clock=clock,
                    legacy_skip_dir=legacy_skip_dir,
                    manifest_path=legacy_manifest_path,
                    copy_fn=legacy_copy_fn,
                    hash_fn=legacy_hash_fn,
                    stat_fn=stat_fn,
                )
                if mv.ok:
                    return _emit(DriverRecord(
                        ts=ts,
                        result_path=path,
                        verdict=VERDICT_LEGACY_SKIP_PRESERVED,
                        quarantined=False,
                        quarantine_reason=skip_reason,
                        preserved_path=mv.preserved_path,
                        activation=ACTIVATION_ENABLED,
                    ))
                # preserve-move 실패(sha256 mismatch/IO) → 원본 보존·explicit failure.
                return _emit(DriverRecord(
                    ts=ts,
                    result_path=path,
                    verdict=VERDICT_LEGACY_SKIP_PRESERVE_FAILED,
                    quarantined=False,
                    quarantine_reason=skip_reason,
                    preserved_path=mv.preserved_path,
                    error=mv.error,
                    activation=ACTIVATION_ENABLED,
                ))
            return _emit(DriverRecord(
                ts=ts,
                result_path=path,
                verdict=VERDICT_NOOP_LEGACY_SKIP,
                quarantined=False,
                quarantine_reason=skip_reason,
                activation=ACTIVATION_ENABLED,
            ))

    # ── readiness grace window (write race 방어) ──────────────────────────────
    # ★ 부분 JSON 을 즉시 quarantine 하지 않는다. writer flush 미완 가능성을 먼저 배제.
    #   readiness 미충족(최근 mtime / size·mtime 불안정 / stat 실패) → NOOP_NOT_READY(DEFER):
    #   wake/quarantine 모두 금지. 다음 트리거/재시도에서 재평가.
    ready, ready_reason = _check_readiness(
        path,
        clock=clock,
        stable_sec=STABLE_SEC if stable_sec is None else stable_sec,
        retries=STABILITY_RETRIES if readiness_retries is None else readiness_retries,
        interval=STABILITY_INTERVAL_SEC if readiness_interval is None else readiness_interval,
        sleep_fn=sleep_fn,
        stat_fn=stat_fn,
    )
    if not ready:
        return _emit(DriverRecord(
            ts=ts,
            result_path=path,
            verdict=VERDICT_NOOP_NOT_READY,
            quarantined=False,
            quarantine_reason=ready_reason,
            activation=ACTIVATION_ENABLED,
        ))

    # ── 조건 1: size > 0 ──────────────────────────────────────────────────────
    try:
        size = os.path.getsize(path)
    except OSError as exc:
        return _quarantine("size0", extra_err=f"getsize 실패: {exc}")
    if size <= 0:
        return _quarantine("size0")

    # ── 조건 2: JSON parse (+ null byte truncation 방어) ──────────────────────
    try:
        with open(path, "rb") as fh:
            raw = fh.read()
    except OSError as exc:
        return _quarantine("parse_fail", extra_err=f"read: {exc}")
    # ★ null byte(\x00) 는 부분 write/truncation 흔적. grace 후(readiness 통과)에도
    #   잔존하면 quarantine. (grace 내 최근 mtime 은 위 readiness 에서 이미 DEFER.)
    if b"\x00" in raw:
        return _quarantine("null_byte")
    try:
        result = json.loads(raw.decode("utf-8"))
    except (ValueError, UnicodeDecodeError) as exc:
        return _quarantine("parse_fail", extra_err=f"parse: {exc}")

    # ── 조건 3: schema ────────────────────────────────────────────────────────
    if not isinstance(result, dict):
        return _quarantine("schema_fail")
    task_id = result.get("task_id")
    completion_signal = result.get("completion_signal")
    if not (isinstance(task_id, str) and task_id.strip()):
        return _quarantine("schema_fail")
    if not (isinstance(completion_signal, str) and completion_signal.strip()):
        return _quarantine("schema_fail")
    task_id = task_id.strip()
    # ★ path traversal 방어 (owner proof 전, defense-in-depth):
    # task_id 가 done_path=os.path.join(result_dir, f"{task_id}.pickup.done") 에
    # 직접 쓰이므로, 경로 탐색 문자 포함 시 schema_fail quarantine.
    if (
        os.path.basename(task_id) != task_id
        or ".." in task_id
        or "/" in task_id
        or "\\" in task_id
    ):
        return _quarantine("schema_fail")

    # ── 조건 4&5: owner proof + self/foreign 아님 ────────────────────────────
    envelope = result.get("collector_envelope")
    if not isinstance(envelope, dict):
        # ── task-2753 Option B: provenance-path (opt-in, default OFF) ──
        #   flag OFF → 기존 동작 100% 보존(envelope 부재 → owner_unprovable quarantine).
        if not provenance_path_enabled:
            return _quarantine("owner_unprovable", owner_class=OKC_FOREIGN)
        # flag ON → executor_result provenance hard-fail 검증.
        #   pass 시 ANU 가 collector_result(adopted_via="provenance") 생성.
        #   owner_proof_pickup_gate(callback fire 보호)는 미수정 — provenance-path 는
        #   collector_result 생성만 하고 callback fire 0 (fire 는 여전히 frozen gate 경유).
        _provenance_fn = provenance_fn or validate_executor_provenance
        pv = _provenance_fn(
            result, path,
            root=root,
            capability_reader=capability_reader,
            activation_epoch=activation_epoch,
            ledger_reader=provenance_ledger_reader,
            forbidden_key_literals=forbidden_key_literals,
        )
        if getattr(pv, "skip", False):
            # 조건9 stale epoch → NOOP_LEGACY_SKIP (move 0/wake 0/quarantine 0)
            return _emit(DriverRecord(
                ts=ts, result_path=path,
                verdict=VERDICT_NOOP_LEGACY_SKIP,
                quarantined=False,
                quarantine_reason=getattr(pv, "reason", "pre_activation_epoch"),
                activation=ACTIVATION_ENABLED,
            ))
        if getattr(pv, "duplicate", False):
            # 조건10 duplicate → idempotent no-op (재수거 0, collector_result 재작성 0)
            return _emit(DriverRecord(
                ts=ts, result_path=path,
                verdict=VERDICT_PICKUP_SKIP,
                owner_key_class=OKC_PROVENANCE,
                quarantined=False,
                activation=ACTIVATION_ENABLED,
            ))
        if not getattr(pv, "ok", False):
            # provenance 실패 = quarantine (untrusted→trusted 세탁 차단)
            return _quarantine(getattr(pv, "reason", "provenance_fail"),
                               owner_class=OKC_FOREIGN)
        # ── provenance PASS → ANU 가 collector_result 생성 (adopted_via="provenance") ──
        #   raw key 0: owner_proof 는 라벨만(OWNER_ANU). source_result_sha256 으로 원본 연결.
        _writer = collector_writer_fn or write_collector_result
        collector = CollectorResult(
            task_id=task_id,
            source_result_sha256=getattr(pv, "source_result_sha256", ""),
            verdict=VERDICT_PROVENANCE_ADOPTED,
            closeout_action=CLOSEOUT_DONE_ACKED,
            owner_proof=OwnerProof(l1_outcome="OWNER_ANU", l2_verdict="AUTHORITATIVE", query_ok=False),
            agent_relay=determine_agent_relay(result),
            adopted_via=ADOPTED_VIA_PROVENANCE,
            collector_role=COLLECTOR_ROLE_ANU,
            ts_kst=ts,
        )
        cr_path = collector_result_path(path, task_id)
        cr_err = None
        move_err = None
        try:
            _writer(collector, cr_path)
        except OSError as exc:
            cr_err = f"collector_result write 실패: {exc}"
        # ── task-2756: collector_result write 성공 시에만 callback bridge (write 실패→callback 0) ──
        # ★ task-2756+1 Medium-2 fix: callback fire 는 collector_result write 성공 후·
        #   _move_processed 전에 수행한다. move 실패 시 result 는 inbox 보존(다음 cycle
        #   재시도)되지만, dedupe(task_id+source_sha)가 재처리 시 BRIDGE_DEDUPED 로 막아
        #   callback duplicate 0 을 보장한다.
        bridge_verdict = None
        bridge_err = None
        if cr_err is None:
            bridge_verdict, bridge_err = _provenance_callback_bridge(
                result_path=path,
                task_id=task_id,
                source_result_sha256=getattr(pv, "source_result_sha256", ""),
                terminal_status=VERDICT_PROVENANCE_ADOPTED,
                callback_launch_fn=callback_launch_fn,
                owner_gate_fn=owner_gate_fn,
                callback_owner_key=callback_owner_key,
                callback_executor_key=callback_executor_key,
                callback_chat_id=callback_chat_id,
                callback_at=callback_at,
                callback_request_builder=callback_request_builder,
                callback_dedupe_seen=callback_dedupe_seen,
                callback_dispatch_cron_id=callback_dispatch_cron_id,
                callback_normal_collector_cron_id=callback_normal_collector_cron_id,
                callback_fallback_cron_id=callback_fallback_cron_id,
                callback_no_fallback=callback_no_fallback,
            )
        # ★ task-2753+1 finding③ + task-2756+2 fix①(HIGH): collector_result write 성공
        #   AND callback bridge 실패 없음(bridge_err is None) 일 때만 source 를 processed 로
        #   이동한다. callback 실패(bridge_err) 시 result 를 inbox 에 보존(move 0)하여 다음
        #   cycle 재처리 가능 → callback 영구 누락 0. (bridge_err 는 DriverRecord.error 로
        #   관측. callback 성공 시 _move_processed 실패해도 dedupe 가 duplicate 0 보장.)
        if cr_err is None and bridge_err is None:
            move_err = _move_processed(path, root, processed_dir)
        err = "; ".join(x for x in (cr_err, move_err, bridge_err) if x) or None
        return _emit(DriverRecord(
            ts=ts, result_path=path,
            verdict=VERDICT_PROVENANCE_ADOPTED,
            owner_key_class=OKC_PROVENANCE,
            quarantined=False,
            error=err,
            activation=ACTIVATION_ENABLED,
            callback_launch_verdict=bridge_verdict,
        ))

    schedule_id = envelope.get("schedule_id") or result.get("schedule_id")

    # ── task-2746: fire-state contract split (경로 A / 경로 B) ────────────────
    #   contract_split=False(기본) → 아래 분기를 건너뛰고 기존 무조건 verify 경로(B 본문)로
    #   직행한다(무회귀). contract_split=True 면 pre-fire owner-pickup result(경로 A)와
    #   already-fired/collector-claimed result(경로 B)를 분리한다.
    #
    #   경로 A 진입 조건(전부 만족 시에만):
    #     - result.owner_pickup_required is True
    #     - result.authoritative is not True            (authoritative claim 0)
    #     - schedule_created_by_executor / callback_fired_by_executor 둘 다 falsy (fired=False)
    #     - envelope 에 schedule_id claim 없음            (executor 가 schedule 안 만듦)
    #   → 이때 authoritative schedule proof 는 **아직 존재하지 않는 게 정상**이므로
    #     verify_collector_authoritative 를 요구하지 않고(호출 0) owner_proof_pickup_gate
    #     (sealed ANU)로 진행한다. schedule proof 는 이후 ANU-owned runner 의 fire 결과물.
    #
    #   self-collector 차단(§7): 경로 A 인데 owner_key_proof_present / envelope.self_key_used
    #     를 claim → 모순(pre-fire 인데 owner-key 보유 주장) → 즉시 quarantine.
    #
    #   그 외(fired=True · envelope.schedule_id claim · authoritative claim · 또는
    #     contract_split=False) → 경로 B = verify_collector_authoritative 필수.
    took_path_a = False
    if contract_split:
        fired = (
            bool(result.get("schedule_created_by_executor"))
            or bool(result.get("callback_fired_by_executor"))
        )
        envelope_claims_schedule = bool(envelope.get("schedule_id"))
        owner_key_claim = (
            bool(result.get("owner_key_proof_present"))
            or bool(envelope.get("self_key_used"))
        )
        pre_fire_owner_pickup = (
            result.get("owner_pickup_required") is True
            and result.get("authoritative") is not True
            and not fired
            and not envelope_claims_schedule
        )
        if pre_fire_owner_pickup:
            # self/owner-key claim on pre-fire = 위반 → 즉시 격리(B 로도 못 감).
            if owner_key_claim:
                return _quarantine(QREASON_OWNER_PICKUP_SELF_KEY_CLAIM,
                                   owner_class=OKC_SELF)
            took_path_a = True

    if took_path_a:
        # ── 경로 A: pre-fire owner-pickup required → verify 면제(호출 0) ────────
        #   owner_proof_pickup_gate(sealed ANU)는 이후 pickup_once 결선에서 강제된다
        #   (owner_gate_fn). 여기서는 ANU 라벨만 전달하고 verify_fn 을 호출하지 않는다.
        owner_class = OKC_ANU
        owner_proof_info = {
            "l1_outcome": OWNER_PICKUP_PREFIRE_OUTCOME,
            "l2_verdict": OWNER_PICKUP_PREFIRE_VERDICT,
            "schedule_id": "",      # pre-fire — schedule proof 아직 없음(정상)
            "query_ok": False,
        }
    else:
        # ── 경로 B: already-fired / collector-claimed → verify_collector_authoritative 필수 ──
        #   (contract_split=False 기본 동작도 이 경로 = 기존과 동일.)
        try:
            v = verify_fn(
                task_id=task_id,
                envelope=envelope,
                probe=probe,
                executor_key=executor_key,
                schedule_id=schedule_id,
                now=clock(),
            )
        except Exception as exc:  # noqa: BLE001 — verify 예외 → owner_proof_error 격리
            return _quarantine("owner_proof_error",
                               owner_class=_envelope_claim_class(envelope),
                               extra_err=f"verify_fn 예외: {exc}")

        if getattr(v, "verdict", None) != VERDICT_AUTHORITATIVE:
            return _quarantine("owner_proof_fail",
                               owner_class=_envelope_claim_class(envelope))

        owner_class = OKC_ANU

        # task-2730: driver 가 이미 수행한 owner-proof 라벨만 추출하여 pickup_once 에 전달한다.
        #   closeout write 는 pickup_once 소유(ANCHOR-A) — process_one 은 라벨 전달만(미소유).
        #   raw key 0: outcome/verdict/schedule_id/query_ok 라벨만(키/argv 0).
        _or = getattr(v, "owner_resolution", None)
        _or = _or if isinstance(_or, dict) else {}
        owner_proof_info = {
            "l1_outcome": str(_or.get("outcome") or ""),
            "l2_verdict": str(getattr(v, "verdict", "") or ""),
            "schedule_id": str(_or.get("schedule_id") or schedule_id or ""),
            "query_ok": bool(_or.get("query_ok", False)),
        }

    # ── 조건 6: terminal marker(done/acked) 만 조기 단락 ──────────────────────
    #   terminal marker 존재 = 확실히 종결됨 → 이동.
    #   ★ task-2730: ledger dedupe 는 여기서 단락하지 않는다. ledger 기록만 있고
    #   marker 가 없는 경우(crash after ledger before collector/marker)는 pickup_once
    #   의 recovery 경로(sha256 일치 dedupe + collector_result/marker 멱등 보강)로
    #   위임해야 closeout 이 완결된다. driver 가 여기서 move 해버리면 collector_result
    #   가 영영 미작성되고, _dedupe_hit 는 sha256 무시(task_id 만)라 신규 result 를
    #   오인 폐기할 수 있다 → pickup_once(authoritative dedupe)로 일원화.
    result_dir = os.path.dirname(path)
    done_path = os.path.join(result_dir, f"{task_id}.pickup.done")
    acked_path = os.path.join(result_dir, f"{task_id}.pickup.acked")
    if os.path.exists(done_path) or os.path.exists(acked_path):
        move_err = _move_processed(path, root, processed_dir)
        return _emit(DriverRecord(
            ts=ts,
            result_path=path,
            verdict=VERDICT_PICKUP_SKIP,
            owner_key_class=owner_class,
            quarantined=False,
            error=move_err,
            activation=ACTIVATION_ENABLED,
        ))

    # ── task-2741-r2 MANDATORY fail-closed (driver-level, fire 전 단락) ────────
    #   driver(systemd-path)는 프로덕션 fire-capable entrypoint 이다. fire executor
    #   (callback_launch_fn)가 주입됐는데 owner_gate_fn(owner_proof_pickup_gate)이 없으면
    #   = gate 우회 — pickup_fn 을 호출하지 않고 즉시 FIRE_FAILED 로 단락한다(callback_launch_fn
    #   미호출·wake argv 미구성·구조적 fire 0). main() 은 owner_gate_fn 을 항상 주입하므로
    #   프로덕션 경로는 도달하지 않는다(defense-in-depth — fire-capable path 는 owner_gate_fn
    #   없이 호출 불가). pickup_once 의 step 0 MANDATORY check 와 동일 보안계약(이중 backstop).
    if callback_launch_fn is not None and owner_gate_fn is None:
        return _emit(DriverRecord(
            ts=ts,
            result_path=path,
            verdict=VERDICT_FIRE_FAILED,
            owner_key_class=owner_class,
            error="MANDATORY fail-closed (task-2741-r2): fire-capable callback_launch_fn 이 "
                  "owner_gate_fn(owner_proof_pickup_gate) 없이 주입됨 — pickup_fn 미호출, "
                  "callback launch/wake 미수행(구조적 fire 0).",
            activation=ACTIVATION_ENABLED,
        ))

    # ── 6조건 전부 통과 → pickup_fn 호출 (P0-a, lock-free) ───────────────────
    try:
        res = pickup_fn(
            path,
            executor_key=executor_key,
            ledger_path=ledger_path,
            owner_proof=owner_proof_info,
            callback_launch_fn=callback_launch_fn,
            owner_gate_fn=owner_gate_fn,
            spawn_governor_fn=spawn_governor_fn,
        )
    except Exception as exc:  # noqa: BLE001 — pickup 예외 → FIRE_FAILED (파일 미이동)
        return _emit(DriverRecord(
            ts=ts,
            result_path=path,
            verdict=VERDICT_FIRE_FAILED,
            owner_key_class=owner_class,
            error=f"pickup_fn 예외: {exc}",
            activation=ACTIVATION_ENABLED,
        ))

    pv = getattr(res, "verdict", None)
    if pv == _PICKUP_CLOSEOUT_DONE:
        # ★ task-2730 green-path: pickup_once 가 결정론 closeout(collector_result+marker)
        #   를 이미 완료. launcher/relay 미호출(wake 0). marker 확정 시에만 processed 이동.
        if _marker_confirmed(res, pv):
            move_err = _move_processed(path, root, processed_dir)
        else:
            move_err = _MOVE_DEFERRED_ERR  # marker 미확정 → 이동 보류(다음 cycle recovery).
        return _emit(DriverRecord(
            ts=ts,
            result_path=path,
            verdict=VERDICT_CLOSEOUT_DONE,
            owner_key_class=OKC_ANU,
            error=move_err,
            activation=ACTIVATION_ENABLED,
            callback_launch_verdict=getattr(res, "callback_launch_verdict", None),
        ))
    if pv == _PICKUP_WAKE_BUILT:
        # ★ relay-path(agent_relay.required=true). launcher_fn/relay_fn 미주입(기본 None)
        #   → 현행 보존(surface only, 실행 0). 주입 시에만 결선:
        #   - relay_fn 주입 → 2-tier terminal relay(send_report+relay_to_anu) 위임.
        #   - 아니면 launcher_fn 주입 → 기존 real wake 결선. driver=decision-only.
        #   relay/wake 발사는 WAKE_BUILT 1회(ledger dedupe). marker 확정 시에만 이동.
        fire_cron_id = None
        launch_err = None
        relay_target = relay_fn if relay_fn is not None else launcher_fn
        if relay_target is not None:
            try:
                lr = relay_target(
                    getattr(res, "argv", None),
                    task_id=task_id,
                    sha256=getattr(res, "sha256", ""),
                )
                fire_cron_id = getattr(lr, "decision", None)  # 라벨만 (argv/key 0)
            except Exception as exc:  # noqa: BLE001 — launcher/relay 예외 → fail-safe(크래시 0)
                launch_err = f"launcher 예외: {exc}"
        if _marker_confirmed(res, pv):
            move_err = _move_processed(path, root, processed_dir)
        else:
            # marker 미확정 → 이동 보류. 발사는 이미 1회 완료(ledger). 다음 cycle 은
            # recovery(SKIP_DEDUPE, 재발사 0)로 marker 보강 후 이동.
            move_err = _MOVE_DEFERRED_ERR
        err = "; ".join(x for x in (launch_err, move_err) if x) or None
        return _emit(DriverRecord(
            ts=ts,
            result_path=path,
            verdict=VERDICT_WAKE_BUILT,
            owner_key_class=OKC_ANU,
            fire_cron_id=fire_cron_id,
            error=err,
            activation=ACTIVATION_ENABLED,
            callback_launch_verdict=getattr(res, "callback_launch_verdict", None),
        ))
    if pv in (_PICKUP_SKIP_TERMINAL, _PICKUP_SKIP_DEDUPE):
        if _marker_confirmed(res, pv):
            move_err = _move_processed(path, root, processed_dir)
        else:
            move_err = _MOVE_DEFERRED_ERR  # recovery 중 marker 보강 실패 → 다음 cycle 재시도.
        return _emit(DriverRecord(
            ts=ts,
            result_path=path,
            verdict=VERDICT_PICKUP_SKIP,
            owner_key_class=owner_class,
            error=move_err,
            activation=ACTIVATION_ENABLED,
        ))
    if pv in (_PICKUP_QUARANTINE, _PICKUP_PENDING):
        return _quarantine("pickup_" + str(pv).lower(), owner_class=owner_class)

    # 그 외 (FAIL / REJECT / SEALED_KEY_MISSING / NO_RESULT_JSON) → FIRE_FAILED.
    reasons = getattr(res, "reasons", None)
    err = "; ".join(reasons) if isinstance(reasons, list) and reasons else f"pickup verdict={pv}"
    return _emit(DriverRecord(
        ts=ts,
        result_path=path,
        verdict=VERDICT_FIRE_FAILED,
        owner_key_class=owner_class,
        error=err,
        activation=ACTIVATION_ENABLED,
    ))


# ── scan_once ───────────────────────────────────────────────────────────────────
def scan_once(
    root: str = CANONICAL_ROOT,
    *,
    pickup_fn=None,
    launcher_fn=None,          # 기본 None = 현행 보존. process_one 으로 그대로 전달.
    callback_launch_fn=None,   # task-2731: 기본 None = 현행 보존. process_one 으로 그대로 전달.
    owner_gate_fn=None,        # task-2741-r1: 기본 None = 현행 보존. process_one 으로 그대로 전달(fire 전 gate 강제).
    spawn_governor_fn=None,     # task-2774+3: 기본 None = 현행 보존. process_one 으로 그대로 전달.
    relay_fn=None,             # task-2730: 기본 None = 현행 보존. process_one 으로 그대로 전달.
    verify_fn=None,
    contract_split: bool = OWNER_PICKUP_CONTRACT_SPLIT_DEFAULT,  # task-2746: process_one 으로 그대로 전달(기본 OFF=무회귀).
    probe=None,
    clock=None,
    executor_key: str = "",
    ledger_path: Optional[str] = None,
    quarantine_dir: Optional[str] = None,
    processed_dir: Optional[str] = None,
    paths: Optional[list] = None,   # None 이면 glob(root/memory/events/p0b_inbox/task-*.result.json), 주어지면 그 목록
    flag_reader=None,
    write_evidence: bool = True,
    evidence_path: Optional[str] = None,
    stable_sec: Optional[float] = None,
    readiness_retries: Optional[int] = None,
    readiness_interval: Optional[float] = None,
    sleep_fn=None,
    max_files: Optional[int] = None,
    legacy_cutoff: bool = False,
    legacy_preserve_move: bool = False,   # task-2744: process_one 으로 그대로 전달.
    legacy_skip_dir: Optional[str] = None,
    legacy_manifest_path: Optional[str] = None,
    legacy_copy_fn=None,
    legacy_hash_fn=None,
    activation_epoch: Optional[float] = None,
    epoch_reader=None,
    stat_fn=None,
    provenance_path_enabled: bool = PROVENANCE_PATH_ENABLED_DEFAULT,  # task-2755 Wiring1: passthrough to process_one
    provenance_fn=None,              # task-2755 Wiring1: passthrough to process_one
    collector_writer_fn=None,        # task-2755 Wiring1: passthrough to process_one
    capability_reader=None,          # task-2755 Wiring1: passthrough to process_one
    provenance_ledger_reader=None,   # task-2755 Wiring1: passthrough to process_one
    forbidden_key_literals: tuple = (),  # task-2755 Wiring1: passthrough to process_one
) -> list:
    """진입점. (1) activation 재확인: disabled → [NOOP_DISABLED] 1건 + evidence + pickup 미호출.
    (2) enabled → 후보 경로 결정 → 각 path process_one → record 목록 반환. 각 record evidence append."""
    clock = clock or (lambda: datetime.now(KST))

    # (1) activation 재확인
    if not is_activated(root, flag_reader=flag_reader):
        rec = DriverRecord(
            ts=_now_kst(clock),
            result_path="",
            verdict=VERDICT_NOOP_DISABLED,
            activation=ACTIVATION_DISABLED,
        )
        if write_evidence:
            _append_evidence(rec, root, evidence_path)
        return [rec]

    # (2) enabled → 후보 경로 결정
    # ★ task-2733: 기본 scan 대상은 전용 inbox(INBOX_DIR_REL) 만이다. memory/events 직속의
    #   잔존 legacy result.json 은 glob 대상이 아니므로 scan/이동/삭제 0(무변경) — systemd
    #   .path 가 inbox 만 감시하는 것과 동일 경로로 정렬되어 잔존 legacy 폭주 0 을 보장한다.
    if paths is None:
        pattern = os.path.join(root, INBOX_DIR_REL, RESULT_GLOB)
        candidates = sorted(glob.glob(pattern))
    else:
        candidates = list(paths)

    # D2: legacy_cutoff 활성 시 activation_epoch 해석 (인자 우선, 없으면 마커 파일)
    epoch = activation_epoch
    if legacy_cutoff and epoch is None:
        epoch = read_activation_epoch(root, epoch_reader=epoch_reader)

    # D1: MAX_FILES 상한 — 초과분은 다음 cycle (전수 처리 금지, bounded)
    limit = MAX_FILES if max_files is None else max_files
    deferred = []
    if limit is not None and len(candidates) > limit:
        deferred = candidates[limit:]
        candidates = candidates[:limit]

    records = []
    for p in candidates:
        rec = process_one(
            p,
            root=root,
            pickup_fn=pickup_fn,
            launcher_fn=launcher_fn,
            callback_launch_fn=callback_launch_fn,
            owner_gate_fn=owner_gate_fn,
            spawn_governor_fn=spawn_governor_fn,
            relay_fn=relay_fn,
            verify_fn=verify_fn,
            contract_split=contract_split,
            probe=probe,
            clock=clock,
            executor_key=executor_key,
            ledger_path=ledger_path,
            quarantine_dir=quarantine_dir,
            processed_dir=processed_dir,
            write_evidence=write_evidence,
            evidence_path=evidence_path,
            stable_sec=stable_sec,
            readiness_retries=readiness_retries,
            readiness_interval=readiness_interval,
            sleep_fn=sleep_fn,
            activation_epoch=epoch,
            legacy_cutoff=legacy_cutoff,
            legacy_preserve_move=legacy_preserve_move,
            legacy_skip_dir=legacy_skip_dir,
            legacy_manifest_path=legacy_manifest_path,
            legacy_copy_fn=legacy_copy_fn,
            legacy_hash_fn=legacy_hash_fn,
            stat_fn=stat_fn,
            provenance_path_enabled=provenance_path_enabled,
            provenance_fn=provenance_fn,
            collector_writer_fn=collector_writer_fn,
            capability_reader=capability_reader,
            provenance_ledger_reader=provenance_ledger_reader,
            forbidden_key_literals=forbidden_key_literals,
        )
        records.append(rec)

    if deferred:
        defer_rec = DriverRecord(
            ts=_now_kst(clock),
            result_path="",
            verdict=VERDICT_NOOP_MAX_FILES_DEFER,
            quarantined=False,
            quarantine_reason=f"max_files_defer:{len(deferred)}",
            activation=ACTIVATION_ENABLED,
        )
        if write_evidence:
            _append_evidence(defer_rec, root, evidence_path)
        records.append(defer_rec)
    return records


# ── task-2760: scan_live_inbox_once (P0B event strategy 진입점) ──────────────────
def scan_live_inbox_once(
    root: str = CANONICAL_ROOT,
    *,
    owner_gate_fn=None,          # 기존 owner_proof gate 재사용(신규 구현 0). gate.ready/status 로 ANU owner 증명 판정.
    callback_launch_fn=None,     # 기본 None = fire 0. 주입 fn 자체에 ACTIVE flag gate 내장(OFF → 실발사 0).
    executor_key: str = "",      # driver 는 executor 가 아님(기본 ""). owner_gate_fn 으로 그대로 전달.
    busy_fn=None,                # 기본 None = busy 미고려. callable()->bool 주입 시 True → 지연(skip 아님).
    clock=None,
    flag_reader=None,
    paths: Optional[list] = None,    # None 이면 glob(root/LIVE_INBOX_DIR_REL/RESULT_GLOB)
    ledger_path: Optional[str] = None,
    processed_dir: Optional[str] = None,
    hash_fn=None,
    write_evidence: bool = False,
    evidence_path: Optional[str] = None,
) -> list:
    """task-2760 P0B event strategy — "신규 이벤트(live inbox)" 감시 진입점.

    기존 scan_once(systemd .path 정렬 INBOX_DIR_REL=p0b_inbox 경로)는 무변경. 본 함수는
    전용 LIVE_INBOX_DIR_REL 만 scan 하여 다음을 보장한다:
      (1) legacy memory/events/task-*.result.json 은 **다른 디렉토리**라 glob 대상 자체가
          아님 → 구조적 제외(retained legacy 의 "존재"로 인한 retrigger 0).
      (2) *.partial 등 비-final 파일은 is_target 으로 제외(dev 의 tmp→fsync→rename atomic
          write 중 부분쓰기 미수거).
      (3) (task_id, source_result_sha256) processed ledger(append-only jsonl) 로 idempotency
          — 동일 키 재감지 시 재처리 0(SKIP_DEDUPE).
      (4) busy 시 skip 이 아니라 **지연**: inbox 잔존 → 다음 idle cycle 재감지(누락 0).
      (5) self-collector 방지는 기존 owner_proof gate(owner_gate_fn) **재사용** — gate ready
          (ANU owner proof) 일 때만 callback_launch_fn fire. gate 미주입/미준비(self/foreign)
          → fire 0 + inbox 잔존(PENDING).
      (6) fire 성공 시 원본을 LIVE_PROCESSED_DIR_REL 로 이동(inbox 비움) **후**, move 성공
          시에만 완료 ledger append(단일 done-marker). ★ task-2760+2 move-first 불변식: 완료
          ledger 는 move 성공과 일관(완료 entry 존재 ⟺ move 성공). move 실패 시 완료 미기록·
          inbox 잔존(PENDING_MOVE, retry) → stuck-complete dedupe leak 0. (ledger 무한증가/
          self-healing 류 강건화는 본 PR 범위 밖 — DESIGN_BACKLOG. 단순 append-only 읽기/쓰기만 유지.)

    ACTIVE=false(activation flag 미설정) → 전면 no-op([NOOP_DISABLED]). 실 callback/wake/
    systemd 발사 0(callback_launch_fn 기본 None + activation gate). systemd enable/install·
    matching/allowlist 재구현·canary 실행은 본 함수 범위 밖이다."""
    clock = clock or (lambda: datetime.now(KST))
    ts = _now_kst(clock)

    def _emit(rec: DriverRecord) -> DriverRecord:
        if write_evidence:
            _append_evidence(rec, root, evidence_path)
        return rec

    # (1) activation gate — ACTIVE=false 면 전면 no-op (수거/이동/fire 0)
    if not is_activated(root, flag_reader=flag_reader):
        return [_emit(DriverRecord(
            ts=ts, result_path="", verdict=VERDICT_NOOP_DISABLED,
            activation=ACTIVATION_DISABLED,
        ))]

    # (2) 후보 = LIVE_INBOX 만. legacy memory/events 직속 result.json 은 다른 디렉토리라
    #     glob 패턴 자체가 매칭하지 않음(구조적 제외 — 이동/삭제 0).
    if paths is None:
        pattern = os.path.join(root, *LIVE_INBOX_DIR_REL.split("/"), RESULT_GLOB)
        candidates = sorted(glob.glob(pattern))
    else:
        candidates = list(paths)

    ledger = ledger_path or os.path.join(root, *LIVE_LEDGER_REL.split("/"))
    pdir = processed_dir or os.path.join(root, *LIVE_PROCESSED_DIR_REL.split("/"))

    # processed ledger 적재 → (task_id, source_result_sha256) seen set (재처리 0 판정 소스)
    seen = set()
    if os.path.isfile(ledger):
        try:
            with open(ledger, "r", encoding="utf-8") as fh:
                for line in fh:
                    line = line.strip()
                    if not line:
                        continue
                    try:
                        entry = json.loads(line)
                    except (ValueError, TypeError):
                        continue
                    if isinstance(entry, dict):
                        seen.add((entry.get("task_id"), entry.get("source_result_sha256")))
        except OSError:
            pass  # ledger 읽기 실패 → 보수적으로 빈 seen (재처리 위험은 move-after-success 가 완화)

    records = []
    for p in candidates:
        # (a) final task-*.result.json 만(.partial/tmp/기타 marker 제외 — atomic gate)
        if not is_target(p):
            records.append(_emit(DriverRecord(
                ts=ts, result_path=p, verdict=VERDICT_NOOP_NOT_TARGET,
                activation=ACTIVATION_ENABLED,
            )))
            continue
        task_id = _legacy_skip_task_id(p)  # basename → task-XXXX (content 미신뢰)
        try:
            sha = _sha256_file(p, hash_fn=hash_fn)
        except OSError as exc:
            records.append(_emit(DriverRecord(
                ts=ts, result_path=p, verdict=VERDICT_FIRE_FAILED,
                error=f"sha256 실패: {exc}", activation=ACTIVATION_ENABLED,
            )))
            continue
        # (b) idempotency: (task_id, sha256) 이미 처리 → 재처리 0 (move/fire/ledger 0)
        if (task_id, sha) in seen:
            records.append(_emit(DriverRecord(
                ts=ts, result_path=p, verdict=VERDICT_LIVE_SKIP_DEDUPE,
                activation=ACTIVATION_ENABLED,
            )))
            continue
        # (c) busy → skip 아님: 지연(inbox 잔존, move/fire/ledger 0). 다음 idle cycle 재감지.
        if busy_fn is not None:
            try:
                busy = bool(busy_fn())
            except Exception:  # noqa: BLE001 — busy probe 예외 → 보수적 지연(누락 0)
                busy = True
            if busy:
                records.append(_emit(DriverRecord(
                    ts=ts, result_path=p, verdict=VERDICT_LIVE_PENDING_BUSY,
                    activation=ACTIVATION_ENABLED,
                )))
                continue
        # (d) self-collector 방지: 기존 owner_proof gate 재사용. 미주입/미준비(self/foreign)
        #     → fire 0 + inbox 잔존(PENDING). ANU owner proof(ready) 일 때만 (e) 진행.
        gate_ready = False
        gate_err = None
        if owner_gate_fn is not None:
            try:
                gate = owner_gate_fn(p, executor_key=executor_key)
                gate_ready = bool(getattr(gate, "ready", False)) or (
                    getattr(gate, "status", None) == GATE_READY_STATUS)
            except Exception as exc:  # noqa: BLE001 — gate 예외 → fail-closed(fire 0·crash 0)
                gate_ready = False
                gate_err = f"owner_gate 예외: {exc}"
        if not gate_ready:
            records.append(_emit(DriverRecord(
                ts=ts, result_path=p, verdict=VERDICT_LIVE_PENDING_OWNER,
                owner_key_class=(OKC_SELF if owner_gate_fn is not None else None),
                error=gate_err, activation=ACTIVATION_ENABLED,
            )))
            continue
        # (e) gate ready(ANU) → fire(주입 시). ACTIVE=false 면 (1) 에서 이미 차단되어 도달
        #     불가하고, callback_launch_fn 에도 ACTIVE flag gate 내장(주입 fn) → 실발사 0.
        launch_verdict = None
        if callback_launch_fn is not None:
            try:
                callback_launch_fn(
                    p,
                    task_id=task_id,
                    sha256=sha,
                    executor_key=executor_key,
                )
                launch_verdict = BRIDGE_FIRED
            except Exception as exc:  # noqa: BLE001 — fire 예외 → result 보존(move/ledger 0), 재시도
                records.append(_emit(DriverRecord(
                    ts=ts, result_path=p, verdict=VERDICT_FIRE_FAILED,
                    owner_key_class=OKC_ANU, error=f"callback launch 실패: {exc}",
                    activation=ACTIVATION_ENABLED,
                )))
                continue
        # (f) fire 성공 → 원본 processed 이동 **선행**, move 성공 시에만 완료 ledger append.
        #     ★ task-2760+2 move-first 불변식: 완료 ledger 기록을 processed move 성공과 **일관**
        #     되게 한다(완료 entry 존재 ⟺ move 성공). ledger append 가 move 보다 선행하면 move
        #     실패 시 완료 entry 가 남아 다음 cycle 에 SKIP_DEDUPE 로 굳고 원본이 inbox 에 영구
        #     stuck(idempotency leak)된다. move-first 로 바꿔, move 실패 시 완료 ledger 미기록·
        #     seen 미등록·inbox 잔존(retry pending)으로 분리한다 → 동일 (task_id, sha256) 의
        #     완료 vs pending/failure 상태가 혼동되지 않는다.
        move_err = _move_processed(p, root, pdir)
        if move_err:
            # move 실패 → 완료 아님. 완료 ledger 미기록·seen 미등록 → stuck-complete dedupe-skip
            # 되지 않고 다음 cycle 재시도(pending). fire 는 이미 성공했으므로 재시도 시 재발사될
            # 수 있으나(move 실패는 예외 경로), 영구 stuck 보다 재시도가 안전하다.
            records.append(_emit(DriverRecord(
                ts=ts, result_path=p, verdict=VERDICT_LIVE_PENDING_MOVE,
                owner_key_class=OKC_ANU, error=move_err, activation=ACTIVATION_ENABLED,
                callback_launch_verdict=launch_verdict,
            )))
            continue
        # move 성공 → 완료 ledger append(단일 done-marker, 단순 append-only). append 실패는
        # 비치명: 원본은 이미 inbox 밖으로 이동되어 재감지/재처리 0(동일 내용 파일 재출현 시에만).
        entry = {
            "task_id": task_id,
            "source_result_sha256": sha,
            "processed_at": ts,
            "runner_id": LIVE_RUNNER_ID,
            "outcome": VERDICT_LIVE_PROCESSED,
        }
        ledger_err = None
        try:
            os.makedirs(os.path.dirname(ledger), exist_ok=True)
            with open(ledger, "a", encoding="utf-8") as fh:
                fh.write(json.dumps(entry, ensure_ascii=False) + "\n")
                fh.flush()
                os.fsync(fh.fileno())
        except OSError as exc:
            ledger_err = f"ledger append 실패: {exc}"
        seen.add((task_id, sha))  # 동일 cycle 내 동일 내용 파일 중복 감지도 재처리 0
        records.append(_emit(DriverRecord(
            ts=ts, result_path=p, verdict=VERDICT_LIVE_PROCESSED,
            owner_key_class=OKC_ANU, error=ledger_err, activation=ACTIVATION_ENABLED,
            callback_launch_verdict=launch_verdict,
        )))
    return records


# ── CLI ───────────────────────────────────────────────────────────────────────
def _build_governor_fn(root: str = CANONICAL_ROOT, *, flag_reader=None):
    """task-2774+3: activation-gated spawn governor builder.

    activation flag(p0b_driver_enabled) 부재/OFF → None 반환(현행 동작 100% 보존,
    governor 미주입). flag ON 일 때만 canonical governor(evaluate_spawn DEFAULT 경로)를
    바인딩한 callable 반환. 이 PR 은 flag 를 켜지 않는다(activation 0) — 기본 경로는 항상 None.
    """
    try:
        if not is_activated(root, flag_reader=flag_reader):
            return None
    except Exception:  # noqa: BLE001 — activation 체크 예외(OSError 등) → governor 미주입(None)으로 안전 보존
        return None
    from utils.spawn_safety_governor import evaluate_spawn  # pyright: ignore[reportMissingImports]  # lazy import (최소 의존)

    def _gov(candidate):
        return evaluate_spawn(candidate)  # governor DEFAULT(canonical) 경로 사용

    return _gov


def main(argv=None, *, activation_intended: bool = False, limited_runtime: bool = False) -> int:
    # ── task-2775+5: limited_runtime(production 무인 가동 전용 신호) 의미충돌 가드 ──
    #   ★ limited_runtime(epoch-bound 안전모델) 과 activation_intended(canary paths-scope
    #   안전모델)가 동시 True 면 두 안전모델이 충돌한다 → 임의로 한쪽을 우선하지 않고
    #   즉시 fail-closed STOP(fire 0·scan/pickup 0). (정책은 2775 bounds 테스트로 고정.)
    if limited_runtime and activation_intended:
        import sys
        print(
            "[anu_pickup_driver] limited_runtime + activation_intended 동시 True "
            "의미 충돌 → fail-closed STOP (fire 0, no scan/pickup/governor; exit 1)",
            file=sys.stderr,
        )
        return 1
    # ── task-2775: limited activation 3경계 게이트 (kill -> N -> T) ──────────────
    #   driver 진입 최상단, fire 가능 경로(launcher/callback/governor build·scan) 이전에
    #   limited-bound gate 1개. 어느 하나라도 걸리면 fire 0·governor 미호출로 즉시 단락.
    #   걸리지 않으면(kill 없음·N 미만·T 이내) None -> 아래 현행 경로 100% 보존.
    #   ★ task-2775+2 (B안): epoch 부재/손상 fail-closed 는 실제 limited/fire-capable
    #   activation(activation_intended=True) 경로에서만 is_activated 를 주입해 발동시킨다.
    #   activation_intended=False(production 기본·legacy scan_once·test-2760) → is_activated_fn
    #   미주입(None) → 현행 pass-through 보존(legacy·test-2760 무손상). epoch 정상 존재 시엔
    #   주입 여부와 무관하게 기존 N/T/kill 경로가 그대로 평가됨(무회귀).
    #   ★ is_activated 는 호출 시점의 모듈 CANONICAL_ROOT 를 명시적으로 읽도록 lambda 로
    #   감싼다(bare is_activated 의 default-arg 는 정의 시점 바인딩이라 root 변경을 못 따름).
    #   이는 아래 line 의 `is_activated(CANONICAL_ROOT)` 호출과 동일 root 의미로 일관된다.
    # ★ task-2775+5: bound gate 의 active 판정 fn 선택.
    #   - limited_runtime=True(운영 __main__ 경로): limited production 안전모델 =
    #     driver flag(p0b_driver_enabled) ON **AND** callback final gate
    #     (p0b_callback_launch_enabled) ON. 두 flag 모두 ON + epoch 부재/손상 →
    #     gate 가 fail-closed(auto-disabled·fire 0). canary preflight 는 타지 않는다.
    #   - activation_intended=True(canary): 기존대로 is_activated 만 주입(무변경).
    #   - 둘 다 False(legacy·test-2760 `drv.main()` 인자0): None → 현행 pass-through 보존.
    if limited_runtime:
        from dispatch.anu_callback_launch_audit import (  # pyright: ignore[reportMissingImports]
            read_callback_launch_enabled,
        )
        _bound_is_activated_fn = (
            lambda: is_activated(CANONICAL_ROOT)
            and read_callback_launch_enabled(CANONICAL_ROOT)
        )
        # ★ task-2775+6 (Gemini HIGH): driver flag(p0b_driver_enabled) 단독 reader 주입.
        #   driver 가 이미 disabled 면 gate 가 N/T ledger scan/disabled 재쓰기 전에 단락.
        _bound_driver_enabled_fn = lambda: is_activated(CANONICAL_ROOT)
    elif activation_intended:
        _bound_is_activated_fn = lambda: is_activated(CANONICAL_ROOT)
        _bound_driver_enabled_fn = lambda: is_activated(CANONICAL_ROOT)
    else:
        _bound_is_activated_fn = None
        _bound_driver_enabled_fn = None
    _bound = limited_activation_bound_gate(
        CANONICAL_ROOT,
        is_activated_fn=_bound_is_activated_fn,
        driver_enabled_fn=_bound_driver_enabled_fn,
    )
    if _bound is not None:
        import sys
        print(
            "[anu_pickup_driver] limited-activation bound gate NOOP "
            f"(fire 0, governor 미호출): {_bound.verdict}",
            file=sys.stderr,
        )
        return 0
    # ── task-2774+5: activation-intended preflight gate (G2 fail-closed + G3 backlog 보호) ──
    #   activation_intended=False(기본) → 미호출(현행 main 경로 100% 보존, 회귀 0).
    #   activation_intended=True 일 때만 scan_once 호출 **전** preflight 를 평가하고,
    #   stop=True 면 scan_once/process_one/pickup/fire 미진행(governor 미주입 live 진행 0).
    #   ★ 본 PR 은 flag/canary 를 켜지 않으므로 production 에서 activation_intended 는
    #   기본 False — 설령 True 라도 flag OFF → NOT_ACTIVATED, canary scope 미배선 →
    #   CANARY_PATHS_REQUIRED 로 fail-closed STOP(backlog 미소비·fire 0).
    if activation_intended:
        from dispatch.anu_activation_preflight import (  # pyright: ignore[reportMissingImports]
            activation_intended_preflight,
        )
        _preflight = activation_intended_preflight(
            CANONICAL_ROOT,
            paths=None,  # production 은 explicit canary scope 미배선 → CANARY_PATHS_REQUIRED fail-closed
            build_governor_fn=_build_governor_fn,
            is_activated_fn=is_activated,
            inbox_dir_rel=INBOX_DIR_REL,
        )
        if _preflight.get("stop"):
            import sys
            print(
                "[anu_pickup_driver] activation-intended preflight STOP "
                f"(no scan/pickup/fire, backlog 미소비): {_preflight.get('reason')!r}",
                file=sys.stderr,
            )
            return 0
    # ★ ANCHOR-1 (launch-0 posture 유지): launcher_fn 은 build_launcher_fn 결과로
    #   real-wake flag OFF(기본) → None → wake launch 0. callback_launch_fn 은 task-2731
    #   의 ANU-owned 강제 fn(ACTIVE flag gate 내장 — OFF 기본 → 전면 no-op, callback launch 0).
    #   본 task(P0B_MINIMUM reporting path)는 두 launch posture 를 변형하지 않고(둘 다 gate
    #   OFF → launch 0) 수거 verdict 의 **회장 보고만** 추가한다. (두 gate 활성화는 회장 별도 승인.)
    launcher_fn = build_launcher_fn(CANONICAL_ROOT)
    # task-2731: ANU-owned callback launch 강제 fn(ACTIVE flag gate 내장 — OFF 면 no-op).
    #   lazy import — 모듈 레벨 의존 회피(기존 테스트 결선 보존).
    from dispatch.anu_callback_launch_audit import (  # pyright: ignore[reportMissingImports]
        build_callback_launch_fn,
    )
    callback_launch_fn = build_callback_launch_fn(CANONICAL_ROOT)
    # task-2741-r1: ★ owner-proof gate 강제 결선 — driver(systemd-path)는 유일한 프로덕션
    #   fire-capable entrypoint 이므로, pickup_once 의 모든 fire(callback_launch_fn/wake
    #   argv/owner proof label) 앞에 owner_proof_pickup_gate 를 반드시 통과시킨다(우회 0).
    #   bound gate 는 owner-gate subcommand 와 동일한 owner_proof_pickup_gate 정책을 공유
    #   (sealed-key ANU proof·eligibility·freshness·dedupe). executor_key="" (driver 는
    #   executor 가 아님). no-fire 유지: gate PASS 여도 callback_launch_fn/launcher_fn 의
    #   ACTIVE flag gate 가 OFF(기본)면 실 launch 0. raw ANU key 0(loader 런타임 로드).
    from dispatch.anu_result_pickup_runner import (  # pyright: ignore[reportMissingImports]
        build_owner_gate_fn,
    )
    owner_gate_fn = build_owner_gate_fn(
        sealed_key_loader=_default_sealed_key_loader,
        executor_key="",
    )
    # task-2774+3: activation-gated spawn governor 결선. flag 부재/OFF(기본) → None →
    #   scan_once 가 spawn_governor_fn=None 으로 pickup_once 호출(현행 100% 보존, governor
    #   미주입). flag ON 시에만 canonical governor 주입. 이 PR 은 flag 변경 0(activation 0).
    spawn_governor_fn = _build_governor_fn(CANONICAL_ROOT)
    # task-2755 Wiring1: scope-gate 4조건 AND. 평상시 flag 부재 → False(OFF).
    # broad/global ON 구조적 불가(4조건 동시 ON = 회장 승인 scope-limited pilot 에서만).
    # ACTIVE=false 유지 — provenance_fn/capability_reader 는 None(기본) 으로 두어
    # process_one 이 production용 validate_executor_provenance + default reader 를 사용.
    # credential 경계 확장 0(forbidden_key_literals/capability_reader 미주입).
    # task-2761: event strategy WIRED_CANDIDATE 분기 (2중 게이팅, default OFF).
    #   is_activated(p0b_driver_enabled) AND read_event_strategy_enabled(p0b_event_strategy_enabled)
    #   둘 다 ON 일 때만 live inbox(scan_live_inbox_once) 경로로 분기한다. 둘 중 하나라도
    #   OFF(두 flag 모두 default OFF) → 아래 else 의 기존 scan_once 경로 그대로(현행 동작 100%
    #   보존, 회귀 0). ★ WIRED ≠ ACTIVE: live 경로(scan_live_inbox_once)도 내부에 activation
    #   gate 를 유지하고, callback_launch_fn 은 ACTIVE flag gate 내장(OFF 기본 → 실발사 0).
    #   기존 owner_proof gate(owner_gate_fn)/callback·result contract 재사용 — 신규 contract 0.
    #   systemd .path/.service·canary·production activation·ledger robustness/matching 무접촉.
    if is_activated(CANONICAL_ROOT) and read_event_strategy_enabled(CANONICAL_ROOT):
        # task-2776: ★ write_evidence=True 명시 전달 — main runtime(live inbox) 경로의
        #   LIVE_* verdict(LIVE_PENDING_OWNER_PROOF·LIVE_PROCESSED·LIVE_PENDING_BUSY·
        #   LIVE_SKIP_DEDUPE·LIVE_PENDING_MOVE 등)를 driver_runs.jsonl(EVIDENCE_JSONL_REL)
        #   에 영속한다. scan_live_inbox_once 의 write_evidence 기본값은 False 라, 미전달 시
        #   actual main runtime 경로에서 LIVE_* verdict evidence 가 0 줄 → OBSERVED 로 닫지
        #   못하는 gap 이 있었다(Phase 1 retry). else 의 scan_once(write_evidence=True 기본)
        #   경로와 **대칭**으로 일관시킨다. ★ report_records(reporter_outbox/ledger)는 별도
        #   파일이라 이중 기록 아님 — _emit→_append_evidence 가 EVIDENCE_JSONL_REL 에 정확히
        #   1줄 append(중복 0). ★ 영속만 추가 — fire/move/callback contract·activation gate
        #   무변경(callback_launch_fn 의 ACTIVE flag gate OFF 기본 → real fire 0 유지).
        records = scan_live_inbox_once(
            CANONICAL_ROOT,
            owner_gate_fn=owner_gate_fn,
            callback_launch_fn=callback_launch_fn,
            write_evidence=True,
        )
    else:
        provenance_path_enabled = read_provenance_path_scope_enabled(CANONICAL_ROOT)
        records = scan_once(
            CANONICAL_ROOT,
            legacy_cutoff=True,
            provenance_path_enabled=provenance_path_enabled,
            # ★ task-2744: pre_activation_epoch 잔존 result 를 watched glob 밖(LEGACY_SKIP_DIR_REL)
            #   으로 preserve-move(삭제0·내용변경0·sha256 검증·manifest 가역) → PathExistsGlob(level-
            #   trigger)를 유지한 채 watched inbox 를 비워 재트리거 폭주(start-limit-hit)를 구조적
            #   제거. ★ no live fire: driver 는 activation flag(p0b_driver_enabled) OFF 기본 →
            #   scan_once 가 NOOP_DISABLED 단락(아래 결선은 flag ON 시에만 도달). flag 변경 0.
            legacy_preserve_move=True,
            launcher_fn=launcher_fn,
            callback_launch_fn=callback_launch_fn,
            owner_gate_fn=owner_gate_fn,
            # ★ task-2746: driver main(= canary 실행) 경로에서만 fire-state contract split ON.
            #   pre-fire owner-pickup result(경로 A)는 authoritative schedule proof(= fire 결과물)를
            #   요구받지 않고 owner_proof_pickup_gate(sealed ANU)로 진행하고, already-fired/claim
            #   result(경로 B)는 verify_collector_authoritative 필수다. self/owner-key/schedule claim
            #   은 B 강제 또는 즉시 quarantine — canary 전용 우회 0·verify_fn 약화 0·fake schedule
            #   통과 0. ★ activation flag 와 무관(여전히 OFF 기본) → live fire/cron 0.
            contract_split=True,
            spawn_governor_fn=spawn_governor_fn,
        )
    # task-2732: 수거 verdict → 회장 chat 에 PASS/HOLD/CALLBACK_MISSING 자동 보고.
    #   sendfile only(wake/relay/cron prompt 0) · key literal 0 · task_id dedupe(1회).
    #   보고 실패는 비치명(수거/closeout 은 이미 완료) — fail-safe 로 driver crash 0.
    #   중복 보고 0: entrypoint 의 driver-level flock single-flight 로 동시 reporter 0
    #   (concurrent check-then-write race 미발생) + ledger task_id dedupe 이중 방어.
    #   ★ (b) MEDIUM fix: 보고 예외를 silent `pass` 로 삼키지 않는다. reporting failure 를
    #   stderr 로 관찰 가능하게 기록(수거/closeout 은 이미 완료 → driver crash 0 유지·fail-safe).
    #   자동 재시도 폭주·wake/callback launch·후속 dispatch 0 — 관찰만(다음 cycle 자연 재시도).
    try:
        report_records(records, root=CANONICAL_ROOT)
    except Exception as exc:  # noqa: BLE001
        import sys
        import traceback
        print(
            f"[anu_pickup_driver] reporting failure (non-fatal; no retry/wake/dispatch): {exc!r}",
            file=sys.stderr,
        )
        traceback.print_exc(file=sys.stderr)
    return 0


__all__ = [
    "DRIVER_NAME",
    "ACTIVATION_FLAG_REL",
    "EVENTS_DIR_REL",
    "INBOX_DIR_REL",
    "QUARANTINE_DIR_REL",
    "PROCESSED_DIR_REL",
    "EVIDENCE_JSONL_REL",
    "LEGACY_SKIP_DIR_REL",
    "LEGACY_SKIP_MANIFEST_REL",
    "LEGACY_SKIP_REASON",
    "LEGACY_SKIP_ACTOR",
    "RESULT_GLOB",
    "MAX_FILES",
    "ACTIVATION_EPOCH_REL",
    "KILL_SWITCH_REL",
    "LIMITED_MAX_PICKUPS_N",
    "LIMITED_WINDOW_MINUTES_T",
    "VERDICT_NOOP_KILL_SWITCH",
    "VERDICT_NOOP_N_EXCEEDED",
    "VERDICT_NOOP_T_EXPIRED",
    "limited_activation_bound_gate",
    "_atomic_write_flag_disabled",
    "_parse_processed_at_to_unix",
    "VERDICT_NOOP_ACTIVE_EPOCH_MISSING",
    "VERDICT_NOOP_ACTIVE_EPOCH_INVALID",
    "_reverse_line_iter",
    "_count_active_window_pickups",
    "LIVE_INBOX_DIR_REL",
    "LIVE_PROCESSED_DIR_REL",
    "LIVE_LEDGER_REL",
    "LIVE_RUNNER_ID",
    "VERDICT_LIVE_PROCESSED",
    "VERDICT_LIVE_SKIP_DEDUPE",
    "VERDICT_LIVE_PENDING_BUSY",
    "VERDICT_LIVE_PENDING_OWNER",
    "VERDICT_LIVE_PENDING_MOVE",
    "PROVENANCE_PATH_FLAG_REL",
    "PROVENANCE_PATH_FLAG_ENABLED",
    "PROVENANCE_SCOPE_ID_REL",
    "read_provenance_path_scope_enabled",
    "STABLE_SEC",
    "STABILITY_RETRIES",
    "STABILITY_INTERVAL_SEC",
    "VERDICT_WAKE_BUILT",
    "VERDICT_CLOSEOUT_DONE",
    "VERDICT_PICKUP_SKIP",
    "VERDICT_QUARANTINE",
    "VERDICT_FIRE_FAILED",
    "VERDICT_NOOP_DISABLED",
    "VERDICT_NOOP_NOT_TARGET",
    "VERDICT_NOOP_NOT_READY",
    "VERDICT_NOOP_LEGACY_SKIP",
    "VERDICT_LEGACY_SKIP_PRESERVED",
    "VERDICT_LEGACY_SKIP_PRESERVE_FAILED",
    "VERDICT_NOOP_MAX_FILES_DEFER",
    "OKC_ANU",
    "OKC_FOREIGN",
    "OKC_SELF",
    "OKC_PROVENANCE",
    "PROVENANCE_PATH_ENABLED_DEFAULT",
    "VERDICT_PROVENANCE_ADOPTED",
    "OWNER_PICKUP_CONTRACT_SPLIT_DEFAULT",
    "OWNER_PICKUP_PREFIRE_OUTCOME",
    "OWNER_PICKUP_PREFIRE_VERDICT",
    "QREASON_OWNER_PICKUP_SELF_KEY_CLAIM",
    "ACTIVATION_ENABLED",
    "ACTIVATION_DISABLED",
    "KST",
    "DriverRecord",
    "REAL_WAKE_FLAG_REL",
    "REAL_WAKE_ENABLED",
    "EVENT_STRATEGY_FLAG_REL",
    "EVENT_STRATEGY_ENABLED",
    "read_activation",
    "is_activated",
    "read_activation_epoch",
    "read_real_wake_enabled",
    "read_event_strategy_enabled",
    "build_launcher_fn",
    "is_target",
    "_legacy_cutoff_check",
    "_sha256_file",
    "_legacy_skip_task_id",
    "_legacy_skip_preserve_move",
    "LegacySkipMoveResult",
    "rollback_legacy_skip",
    "_check_readiness",
    "process_one",
    "scan_once",
    "scan_live_inbox_once",
    "report_records",
    "main",
]

if __name__ == "__main__":
    raise SystemExit(main(limited_runtime=True))