# -*- coding: utf-8 -*-
"""dispatch.anu_result_pickup_runner — task-2720 P0-a callback pickup runner.

wired 후보 / active=P0-b 대기, 자가발사 0.

이 모듈은 executor 가 작성한 result.json 을 pickup 하여 ANU-owned callback
발사 request(argv)를 생성한다. crontab/systemd/inotify 설치·감시 코드 0.
실 발사는 권한 있는 ANU 세션이 수행한다 (P0-b driver 가 pickup_once 를 호출).

재사용 (중복 신설 0):
  - executor_write_result_json  : dispatch.anu_owned_callback_enforcement
  - anu_runner_pickup_and_fire  : dispatch.anu_owned_callback_enforcement
  - verify_collector_authoritative: dispatch.anu_owned_callback_enforcement
  - 상수/verdicts               : 동일 모듈

Layer A / NO-CRON: cron register/remove 0, subprocess 0, merge 0.

task-2741 (OS-level owner-proof gate, no-fire): 본 모듈 하단에 sealed-key 기반
owner-proof gate(owner_proof_pickup_gate / attempt_owner_fire)를 추가했다. 실
cokacdir --cron 발사(real-fire)·cron register/remove·systemd·merge 0 — argv 구성
까지만. 유일한 subprocess 는 freshness 판정용 **읽기 전용 git rev-parse HEAD**
(_default_head_probe, 주입 가능)이며 cokacdir/cron 호출이 아니다.

task-2741-r2 (gate MANDATORY fail-closed — opt-in 제거): fire-capable callback
executor(``callback_launch_fn``)가 주입됐는데 ``owner_gate_fn`` 이 없으면(r1 의
opt-in 우회) pickup_once 는 어떤 작업보다 **먼저 즉시 fail-closed**(OWNER_GATE_BLOCKED)
한다 — callback_launch_fn 호출 0·wake/callback argv 구성 0·owner proof label 기록 0
(ANU/dev key 무관·구조적 fire 0). 외부 fire-capable API(pickup_once)는 gate mandatory.
fire-없는 라이브러리 primitive(wake argv 빌더 ``anu_runner_pickup_and_fire`` 는 argv
구성만 — 실 발사 0)는 그대로 유지된다(callback_launch_fn 미주입 경로). production
fire-capable entrypoint(driver main·CLI pickup·owner-gate)는 모두 owner_gate_fn 을
주입하므로 wake argv 까지도 동일 gate 정책으로 보호된다(우회 0).
"""
from __future__ import annotations

import hashlib
import json
import os
from dataclasses import dataclass, field
from datetime import datetime, timezone
from typing import Callable, List, Optional, Sequence, cast

# 재사용 모듈 (절대 수정 금지)
from dispatch.anu_owned_callback_enforcement import (  # pyright: ignore[reportMissingImports]
    CANONICAL_ROOT,
    VERDICT_NON_AUTHORITATIVE,
    VERDICT_PENDING_OWNER_PROOF,
    VERDICT_QUARANTINED,
    VERDICT_REJECTED,
    anu_runner_pickup_and_fire,
    verify_collector_authoritative,
)
# task-2741: sealed-key 기반 owner-proof 의 단일 신뢰원천 = is_anu_key 구조적 검증.
#   ANU key set(DEFAULT_ANU_KEYS) literal 은 callback_owner_enforcer(무수정)에만 존재
#   하므로 본 모듈에는 raw ANU key literal 0 (import 만 한다).
from dispatch.callback_owner_enforcer import (  # pyright: ignore[reportMissingImports]
    DEFAULT_ANU_KEYS,
    is_anu_key,
)
# task-2730: collector_result schema + atomic writer + agent_relay 판정 (신규 helper).
from dispatch.anu_collector_result import (  # pyright: ignore[reportMissingImports]
    CLOSEOUT_DONE_ACKED,
    CLOSEOUT_RELAY_PENDING,
    AgentRelay,
    CollectorResult,
    OwnerProof,
    _fsync_dir,
    collector_result_path,
    determine_agent_relay,
    write_collector_result,
)

# ── 상수 ─────────────────────────────────────────────────────────────────────
PICKUP_SCHEMA = "dispatch.anu_result_pickup_runner.v1"

# verdict 상수
PICKUP_WAKE_BUILT = "WAKE_BUILT"
PICKUP_CLOSEOUT_DONE = "CLOSEOUT_DONE"   # task-2730: green-path 결정론 closeout(wake 0)
PICKUP_NO_RESULT_JSON = "NO_RESULT_JSON"
PICKUP_FAIL = "FAIL"
PICKUP_SKIP_TERMINAL = "SKIP_TERMINAL"
PICKUP_SKIP_DEDUPE = "SKIP_DEDUPE"
PICKUP_QUARANTINE = "QUARANTINE"
PICKUP_PENDING = "PENDING_OWNER_PROOF"
PICKUP_REJECT = "REJECT"
PICKUP_SEALED_KEY_MISSING = "SEALED_KEY_MISSING"
PICKUP_LEDGER_WRITE_FAILED = "LEDGER_WRITE_FAILED"
# task-2730: collector_result write 실패 → marker 미작성·retry (collector before marker 불변).
PICKUP_COLLECTOR_WRITE_FAILED = "COLLECTOR_WRITE_FAILED"
# task-2741-r1: owner-proof gate 미PASS 로 fire-capable 경로가 차단된 fail-closed verdict.
#   owner_gate_fn 주입(프로덕션 fire-capable entrypoint)에서만 도달 — fire 0·argv 0·label 0.
PICKUP_OWNER_GATE_BLOCKED = "OWNER_GATE_BLOCKED"
# task-2774: spawn safety governor 가 차단(BLOCK/QUEUE/TRIP)한 fire-capable 경로 verdict.
PICKUP_SPAWN_BLOCKED = "SPAWN_GOVERNOR_BLOCKED"

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

# dedupe ledger 기본 경로
DEFAULT_LEDGER_PATH = os.path.join(
    CANONICAL_ROOT, "memory", "events", "callback_4tuple_index.jsonl"
)

# ANU key 환경변수명 (sealed-key 로더용)
ENV_ANU_KEY = "COKACDIR_KEY_ANU"

# task-2731: result-json contract — terminal_intent + callback_schedule_created.
#   dev bot 은 result.json 만 작성하고 callback schedule 을 **자가 생성하지 않는다**
#   (callback_schedule_created 는 false 가 contract). terminal_intent 는 executor 가
#   완료 의도를 명시하는 자유 문자열(예: "completed", "needs_relay"). 둘 다 코드
#   schema 로만 읽고 devbot prompt 수정 0.
TERMINAL_INTENT_DEFAULT = "completed"


def read_terminal_intent(result: object) -> tuple:
    """result.json 의 terminal_intent + callback_schedule_created 를 코드 schema 로 읽음.

    반환: (terminal_intent: str, callback_schedule_created: bool).
      - terminal_intent 부재/비문자열 → TERMINAL_INTENT_DEFAULT.
      - callback_schedule_created truthy → True(contract violation 라벨 — dev bot 이
        callback schedule 을 자가 생성했음을 의미. 코드 강제 경로는 이와 무관하게
        ANU-owned launch 를 수행하되, 라벨로 surface 한다)."""
    if not isinstance(result, dict):
        return TERMINAL_INTENT_DEFAULT, False
    ti = result.get("terminal_intent")
    ti_stripped = ti.strip() if isinstance(ti, str) else ''
    intent = ti_stripped or TERMINAL_INTENT_DEFAULT
    created = bool(result.get("callback_schedule_created", False))
    return intent, created


# ── sealed-key 로더 ───────────────────────────────────────────────────────────
def _default_sealed_key_loader() -> Optional[str]:
    """ANU key 는 .env.keys 의 COKACDIR_KEY_ANU 환경변수로만 로드.
    코드/argv/prompt 어디에도 키 literal 금지."""
    try:
        from utils.env_loader import load_env_keys  # pyright: ignore[reportMissingImports]
        load_env_keys()  # 기본 경로 /home/jay/workspace/.env.keys
    except Exception:  # noqa: BLE001 — 로더 자체 실패는 fail-closed(None)
        pass
    val = (os.environ.get(ENV_ANU_KEY) or "").strip()
    return val or None


# ── PickupResult dataclass ───────────────────────────────────────────────────
@dataclass
class PickupResult:
    """pickup_once 의 반환값."""

    schema: str
    verdict: str
    task_id: str
    result_json_path: str
    sha256: str
    wake_built: bool
    argv: Optional[List[str]]
    classification: str
    reasons: List[str] = field(default_factory=list)
    marker_path: Optional[str] = None
    # task-2730: closeout 산출물 정보 (driver/테스트 관측용 — raw key 0).
    closeout_action: Optional[str] = None
    closeout_path: Optional[str] = None
    agent_relay_required: bool = False
    agent_relay_reason: str = ""
    # task-2731: result-json contract terminal_intent + callback launch 강제 관측값.
    terminal_intent: str = ""
    callback_schedule_created: bool = False
    callback_launch_verdict: Optional[str] = None
    callback_audit_path: Optional[str] = None

    @property
    def ok(self) -> bool:
        # WAKE_BUILT(relay-path) 와 CLOSEOUT_DONE(green-path) 모두 정상 종료.
        return self.verdict in (PICKUP_WAKE_BUILT, PICKUP_CLOSEOUT_DONE)

    def to_json(self) -> dict:
        return {
            "schema": self.schema,
            "verdict": self.verdict,
            "task_id": self.task_id,
            "result_json_path": self.result_json_path,
            "sha256": self.sha256,
            "wake_built": self.wake_built,
            "argv": list(self.argv) if self.argv is not None else None,
            "classification": self.classification,
            "reasons": list(self.reasons),
            "marker_path": self.marker_path,
            "closeout_action": self.closeout_action,
            "closeout_path": self.closeout_path,
            "agent_relay_required": self.agent_relay_required,
            "agent_relay_reason": self.agent_relay_reason,
            "terminal_intent": self.terminal_intent,
            "callback_schedule_created": self.callback_schedule_created,
            "callback_launch_verdict": self.callback_launch_verdict,
            "callback_audit_path": self.callback_audit_path,
        }


def _fail(verdict: str, *, task_id: str = "", result_json_path: str = "",
          sha256: str = "", classification: str = "",
          reasons: Optional[List[str]] = None) -> PickupResult:
    """no-op(wake=0) PickupResult 편의 생성."""
    return PickupResult(
        schema=PICKUP_SCHEMA,
        verdict=verdict,
        task_id=task_id,
        result_json_path=result_json_path,
        sha256=sha256,
        wake_built=False,
        argv=None,
        classification=classification,
        reasons=list(reasons or []),
        marker_path=None,
    )


# ── owner_proof 라벨 빌더 (raw key 0) ─────────────────────────────────────────
def _owner_proof_from_verification(v, schedule_id) -> OwnerProof:
    """pickup_once 자체 verify(CollectorVerification) → OwnerProof 라벨 (raw key 0)."""
    res = v.owner_resolution if isinstance(getattr(v, "owner_resolution", None), dict) else {}
    return OwnerProof(
        l1_outcome=str(res.get("outcome") or ""),
        l2_verdict=str(getattr(v, "verdict", "") or ""),
        schedule_id=str(res.get("schedule_id") or schedule_id or ""),
        query_ok=bool(res.get("query_ok", False)),
    )


def _owner_proof_from_param(owner_proof: dict, envelope) -> OwnerProof:
    """driver(process_one)가 전달한 owner-proof 라벨 dict → OwnerProof (raw key 0).
    driver 가 closeout 을 소유하지 않고(ANCHOR-A), 이미 계산한 owner-proof 라벨만 전달한다."""
    sid = str(owner_proof.get("schedule_id") or "")
    if not sid and isinstance(envelope, dict):
        sid = str(envelope.get("schedule_id") or "")
    return OwnerProof(
        l1_outcome=str(owner_proof.get("l1_outcome") or ""),
        l2_verdict=str(owner_proof.get("l2_verdict") or ""),
        schedule_id=sid,
        query_ok=bool(owner_proof.get("query_ok", False)),
    )


# ── pickup_once (공개 API) ────────────────────────────────────────────────────
def pickup_once(
    result_json_path: str,
    *,
    gh_probe: Optional[Callable[[str], dict]] = None,
    clock: Optional[Callable[[], datetime]] = None,
    sealed_key_loader: Optional[Callable[[], Optional[str]]] = None,
    executor_key: str = "",
    ledger_path: Optional[str] = None,
    anu_keys: Optional[object] = None,
    owner_proof: Optional[dict] = None,
    callback_launch_fn: Optional[Callable[..., object]] = None,
    owner_gate_fn: Optional[Callable[[str], object]] = None,
    spawn_governor_fn: Optional[Callable[..., dict]] = None,
) -> PickupResult:
    """result.json pickup → deterministic closeout (또는 relay-path WAKE_BUILT).

    task-2730 closeout 통합 (corrected_spec_v2 §3·§4 / ANCHOR-A·B):
      - closeout write OWNER = 이 함수(pickup_once) + helper anu_collector_result.
        process_one(driver)은 closeout 미소유 — verdict 분기만.
      - durability order(ANCHOR-B 불변): **ledger(dedupe·fsync) → collector_result
        (atomic os.replace) → done marker(terminal sentinel·마지막)**.
      - agent_relay 판정 = result.json `relay_hints` producer-contract 4 필드만
        (determine_agent_relay, 추론 0).
        · required=false(green) → collector_result(done_acked) + marker →
          **CLOSEOUT_DONE 반환(wake argv 미사용)**.
        · required=true(예외) → 기존 wake argv 경로 → collector_result(relay_pending)
          + marker → **WAKE_BUILT 반환**(driver 가 relay 분기).

    task-2741-r2 (owner-proof gate MANDATORY fail-closed — opt-in 제거):
      - ★ fire-capable callback executor(callback_launch_fn)가 주입됐는데 owner_gate_fn 이
        없으면(=r1 의 opt-in 우회) **어떤 작업보다 먼저 즉시 fail-closed**: result.json 읽기
        전 단계에서 OWNER_GATE_BLOCKED 반환 — callback_launch_fn 호출 0·wake/callback argv
        구성 0·owner proof label 기록 0·closeout write 0(ANU/dev key 무관·구조적 fire 0).
      - owner_gate_fn 주입 시 모든 fire(callback_launch_fn·wake/callback argv·owner proof
        label) 보다 먼저 owner_proof_pickup_gate 를 반드시 통과해야 한다(우회 0). 미PASS →
        OWNER_GATE_BLOCKED fail-closed(fire 0).
      - callback_launch_fn 미주입(=fire executor 부재) = fire-없는 라이브러리 primitive
        경로(wake argv 빌더는 argv 구성만·실 발사 0) — gate 없이 closeout/argv 구성 허용.
        production fire-capable entrypoint(driver main·CLI pickup)는 owner_gate_fn 을
        항상 주입하므로 이 경로에서도 wake argv 가 동일 gate 로 보호된다.

    처리 순서 (fail-closed):
      0. ★ MANDATORY: callback_launch_fn 주입 + owner_gate_fn 부재 → 즉시 OWNER_GATE_BLOCKED
      1. result.json 읽기 / 파싱            2. task_id 검증 + sha256
      2.5 owner-proof gate 강제(owner_gate_fn 주입 시·fire 전 단락)
      3. terminal-marker no-op              4. dedupe ledger 확인(already_in_ledger 플래그)
      5. collector verify(조건부, gh_probe) 6. sealed-key proof
      6.5 agent_relay 판정(relay_hints)     7. wake argv 생성(relay-path·fresh 만)
      8. ledger append(ANCHOR-B 1순위)      9. collector_result write(ANCHOR-B 2순위)
      10. done marker write(ANCHOR-B 3순위·마지막)  11. verdict 반환
    """
    # ── 0. ★ task-2741-r2 MANDATORY fail-closed (opt-in 제거) ──────────────────
    #   fire-capable callback executor(callback_launch_fn)가 주입됐는데 owner-proof gate
    #   (owner_gate_fn)이 없으면 = r1 의 opt-in 우회(ANU 재현: ANU/dev key 무관 callback 1회
    #   호출). 이 경우 result.json 을 읽기도 전에 즉시 fail-closed 한다: callback_launch_fn
    #   미호출 · wake/callback argv 미구성 · owner proof label 미기록 · ledger/collector_result/
    #   marker write 0(구조적 fire 0). 외부 fire-capable API 는 gate mandatory — 문서/관례가
    #   아닌 실행 경로 강제. (callback_launch_fn 미주입 = fire executor 부재 = primitive 보존.)
    if callback_launch_fn is not None and owner_gate_fn is None:
        return _fail(
            PICKUP_OWNER_GATE_BLOCKED,
            result_json_path=str(result_json_path or ""),
            reasons=[
                "MANDATORY fail-closed (task-2741-r2): fire-capable callback_launch_fn 이 "
                "owner_gate_fn(owner_proof_pickup_gate) 없이 주입됨 — opt-in 우회 차단. "
                "callback_launch_fn 미호출 · wake/callback argv 미구성 · owner proof label "
                "미기록 · closeout write 0 (ANU/dev key 무관·구조적 fire 0).",
            ],
        )

    _clock = clock or (lambda: datetime.now(timezone.utc))
    _ledger = ledger_path or DEFAULT_LEDGER_PATH

    # ── 1. result.json 읽기 ───────────────────────────────────────────────────
    if not result_json_path or not os.path.isfile(result_json_path):
        return _fail(PICKUP_NO_RESULT_JSON,
                     result_json_path=str(result_json_path or ""),
                     reasons=["result.json 부재 — executor 미작성 또는 경로 오류."])
    try:
        with open(result_json_path, "rb") as fh:
            raw_bytes = fh.read()
        result: object = json.loads(raw_bytes.decode("utf-8"))
    except (OSError, ValueError) as exc:
        return _fail(PICKUP_FAIL,
                     result_json_path=result_json_path,
                     reasons=[f"result.json 읽기/파싱 실패: {exc}"])
    if not isinstance(result, dict):
        return _fail(PICKUP_FAIL,
                     result_json_path=result_json_path,
                     reasons=["result.json 이 object(dict) 아님."])

    # ── 2. task_id 검증 + sha256 계산 ─────────────────────────────────────────
    task_id = str(result.get("task_id") or "").strip()
    if not task_id:
        return _fail(PICKUP_REJECT,
                     result_json_path=result_json_path,
                     reasons=["result.json task_id 없음/빈값 — 잘못된 argv 차단(fail-closed)."])
    sha256 = hashlib.sha256(raw_bytes).hexdigest()

    # task-2731: result-json contract — terminal_intent + callback_schedule_created.
    terminal_intent, callback_schedule_created = read_terminal_intent(result)

    # ── 2.45 task-2774: spawn safety governor (owner-proof gate 직전 차단형) ─────
    #   opt-in: spawn_governor_fn 이 명시 주입된 경우에만 평가한다(기본 None →
    #   기존 체인/회귀 무손상). ALLOW 만 owner-proof 로 진행. BLOCK/QUEUE/TRIP →
    #   spawn 0(callback_launch_fn 미호출 · owner gate 미호출 · argv 미구성) + 반환.
    #   governor 호출/오류는 차단형이므로 fail-closed BLOCK (보수적).
    if spawn_governor_fn is not None:
        _gov_candidate = {
            "task_id": task_id,
            "head_sha": str(result.get("head_sha") or ""),
            "terminal_state": str(result.get("terminal_state") or terminal_intent or ""),
        }
        try:
            _gov_decision = spawn_governor_fn(_gov_candidate)
            _gov_verdict = (
                (_gov_decision or {}).get("decision")
                if isinstance(_gov_decision, dict) else None
            )
            _gov_reason = (
                (_gov_decision or {}).get("reason")
                if isinstance(_gov_decision, dict) else "GOVERNOR_BAD_RETURN"
            )
        except Exception as _gov_exc:  # noqa: BLE001 — 차단형: 오류 시 fail-closed BLOCK
            _gov_verdict = None
            _gov_reason = f"GOVERNOR_ERROR:{_gov_exc}"
        if _gov_verdict != "ALLOW":
            return _fail(
                PICKUP_SPAWN_BLOCKED,
                task_id=task_id,
                result_json_path=result_json_path,
                sha256=sha256,
                reasons=[
                    "spawn_safety_governor 차단 — fire-capable 경로 spawn 0(우회 0). "
                    f"decision={_gov_verdict or 'UNAVAILABLE'} reason={_gov_reason}. "
                    "callback_launch_fn 미호출 · owner gate 미호출 · argv 미구성.",
                ],
            )

    # ── 2.5 task-2741-r1: owner-proof gate 강제 결선 (fire-capable 우회 0) ──────
    #   owner_gate_fn 주입 시(프로덕션 fire-capable entrypoint = anu_pickup_driver.main
    #   →scan_once→process_one, CLI pickup) 어떤 fire 보다 **먼저** owner_proof_pickup_gate
    #   를 반드시 통과해야 한다. gate 미PASS(OWNER_PICKUP_READY_NO_FIRE 아님) → 즉시
    #   OWNER_GATE_BLOCKED fail-closed 로 단락: callback_launch_fn 미호출 · wake/callback
    #   argv 미구성 · owner proof label 미기록 · ledger/collector_result/marker write 0.
    #   ★ owner_gate_fn 미주입(기본 None) 이면서 여기 도달했다는 것은 step 0 의 MANDATORY
    #   check 를 통과했다는 뜻 = callback_launch_fn 도 None(fire executor 부재 = fire-없는
    #   라이브러리 primitive 직접 호출, 프로덕션 entrypoint 아님) — 현행 보존. owner-gate
    #   subcommand 와 동일한 owner_proof_pickup_gate 정책을 공유한다(pickup/process_one 우회
    #   경로 0). gate PASS 라벨은 sealed-key 구조 검증에서만 온다(raw ANU key literal 0).
    if owner_gate_fn is not None:
        try:
            _gate_res = owner_gate_fn(result_json_path)
        except Exception:  # noqa: BLE001 — gate 호출 예외는 fail-closed(미PASS 취급)
            _gate_res = None
        _gate_status = getattr(_gate_res, "status", None)
        _gate_ready = (
            bool(getattr(_gate_res, "ready", False))
            or _gate_status == GATE_OWNER_PICKUP_READY_NO_FIRE
        )
        if not _gate_ready:
            return _fail(
                PICKUP_OWNER_GATE_BLOCKED,
                task_id=task_id,
                result_json_path=result_json_path,
                sha256=sha256,
                reasons=[
                    "owner_proof_pickup_gate 미PASS — fire-capable 경로 차단(우회 0). "
                    f"gate_status={_gate_status or 'UNAVAILABLE'}. callback_launch_fn 미호출 · "
                    "wake/callback argv 미구성 · owner proof label 미기록 · closeout write 0.",
                ],
            )

    # task-2731: ANU-owned callback launch 강제 helper (audit-dedupe 멱등).
    #   ★ 모든 reprocess 경로(fresh / recovery / SKIP_TERMINAL)에서 시도한다 —
    #   marker 후 launch 가 전이 실패(LAUNCH_FAIL/예외)해도 다음 reprocess 에서
    #   retry 되어 영구 skip 트랩을 제거한다. at-most-once 는 callback launch 모듈의
    #   audit jsonl dedupe(task_id, sha256)가 보장한다(이중 callback 0).
    def _enforce_callback(relay_flag: bool) -> tuple:
        if callback_launch_fn is None:
            return None, None
        _sid = result.get("schedule_id")
        _env = result.get("collector_envelope")
        if isinstance(_env, dict):
            _sid = _env.get("schedule_id") or _sid
        try:
            clr = callback_launch_fn(
                result_json_path,
                task_id=task_id,
                sha256=sha256,
                executor_key=executor_key,
                schedule_id=str(_sid or ""),
                terminal_intent=terminal_intent,
                relay_required=relay_flag,
            )
            return getattr(clr, "verdict", None), getattr(clr, "audit_path", None)
        except Exception:  # noqa: BLE001 — launch 예외는 closeout 을 막지 않음(fail-safe)
            #   고정 라벨만 — 예외 문자열 persist 0(raw-key hygiene).
            return "LAUNCH_EXCEPTION", None

    result_dir = os.path.dirname(result_json_path)

    # ── 3. terminal-marker no-op (marker = 종결 sentinel — 최우선 단락) ────────
    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):
        # marker 존재 = 종결. 단 callback launch 는 audit-dedupe 멱등으로 재시도하여
        # 이전 pass 의 전이 실패를 복구한다(ENFORCED audit 있으면 SKIP_DEDUPE no-op).
        cb_verdict, cb_audit = _enforce_callback(False)
        return PickupResult(
            schema=PICKUP_SCHEMA,
            verdict=PICKUP_SKIP_TERMINAL,
            task_id=task_id,
            result_json_path=result_json_path,
            sha256=sha256,
            wake_built=False,
            argv=None,
            classification="",
            reasons=["terminal marker 존재 — 이미 처리됨 (no-op). callback launch 는 "
                     "audit-dedupe 멱등 재시도(이전 전이 실패 복구)."],
            marker_path=None,
            terminal_intent=terminal_intent,
            callback_schedule_created=callback_schedule_created,
            callback_launch_verdict=cb_verdict,
            callback_audit_path=cb_audit,
        )

    # ── 4. dedupe ledger 확인 → already_in_ledger 플래그 (early return 0) ──────
    #   marker 부재(위에서 단락) + ledger 존재 = ledger append 후 crash → recovery 경로.
    #   wake/relay 재발사 0(dedupe), 단 collector_result/marker 는 멱등 재작성한다.
    already_in_ledger = False
    if os.path.isfile(_ledger):
        try:
            with open(_ledger, "r", encoding="utf-8") as lfh:
                for line in lfh:
                    line = line.strip()
                    if not line:
                        continue
                    try:
                        entry = json.loads(line)
                    except (ValueError, TypeError):
                        continue  # 파싱 실패 라인 skip (fail-safe)
                    if (
                        isinstance(entry, dict)
                        and entry.get("event") in LEDGER_DEDUPE_EVENTS
                        and entry.get("task_id") == task_id
                        and entry.get("sha256") == sha256
                    ):
                        already_in_ledger = True
                        break
        except OSError:
            pass  # ledger 읽기 실패는 skip (fail-safe, 진행)

    # ── 5. collector verify (조건부, gh_probe 주입 시) ───────────────────────
    envelope = result.get("collector_envelope")
    schedule_id = None
    _owner_proof = OwnerProof()
    if isinstance(envelope, dict) and gh_probe is not None:
        schedule_id = (
            envelope.get("schedule_id")
            or result.get("schedule_id")
            or None
        )
        v = verify_collector_authoritative(
            task_id=task_id,
            envelope=envelope,
            probe=gh_probe,
            executor_key=executor_key,
            schedule_id=schedule_id,
            now=_clock(),
        )
        if v.verdict == VERDICT_QUARANTINED:
            return _fail(PICKUP_QUARANTINE,
                         task_id=task_id,
                         result_json_path=result_json_path,
                         sha256=sha256,
                         classification=v.classification,
                         reasons=[
                             "collector verify: QUARANTINED (self-collector) — 작업물 보존, wake 0."
                         ] + list(v.reasons))
        if v.verdict == VERDICT_PENDING_OWNER_PROOF:
            return _fail(PICKUP_PENDING,
                         task_id=task_id,
                         result_json_path=result_json_path,
                         sha256=sha256,
                         classification=v.classification,
                         reasons=[
                             "collector verify: PENDING_OWNER_PROOF (재시도 가능) — wake 0."
                         ] + list(v.reasons))
        if v.verdict in (VERDICT_NON_AUTHORITATIVE, VERDICT_REJECTED):
            return _fail(PICKUP_FAIL,
                         task_id=task_id,
                         result_json_path=result_json_path,
                         sha256=sha256,
                         classification=v.classification,
                         reasons=[
                             f"collector verify: {v.verdict} — fail-closed, wake 0."
                         ] + list(v.reasons))
        # v.ok (AUTHORITATIVE) → owner_proof 라벨 채취 후 통과.
        _owner_proof = _owner_proof_from_verification(v, schedule_id)
    elif isinstance(owner_proof, dict):
        # driver(process_one)가 이미 owner-proof 를 수행하고 라벨만 전달한 경로.
        _owner_proof = _owner_proof_from_param(owner_proof, envelope)
    elif isinstance(envelope, dict):
        _owner_proof = OwnerProof(schedule_id=str(envelope.get("schedule_id") or ""))

    # ── 6. sealed-key proof (fresh closeout gate) ────────────────────────────
    #   sealed-key 는 fresh wake argv 생성에만 필요하다(relay-path). fresh closeout
    #   진입 게이트로 요구하되, **recovery(already_in_ledger) 경로는 면제**한다:
    #   recovery 는 신규 wake/relay 0(argv 미생성)이고 collector_result/marker 멱등
    #   보강만 수행하므로 key 가 불필요하다. recovery 를 key 가용성에 묶으면 crash-
    #   after-ledger 가 key 부재 cycle 에서 영구 stranded 된다(closeout 미완결).
    _loader = sealed_key_loader if sealed_key_loader is not None else _default_sealed_key_loader
    anu_key = _loader()
    if not already_in_ledger and not anu_key:
        return _fail(PICKUP_SEALED_KEY_MISSING,
                     task_id=task_id,
                     result_json_path=result_json_path,
                     sha256=sha256,
                     reasons=[
                         "ANU sealed key 미로드 — COKACDIR_KEY_ANU 환경변수 미설정 또는 "
                         ".env.keys 부재. fail-closed (wake 0). (recovery 경로는 면제.)"
                     ])

    # ── 6.5 agent_relay 판정 (relay_hints producer contract, 추론 0) ──────────
    relay = determine_agent_relay(result)
    relay_required = relay.required

    # ── 7. wake argv 생성 (relay-path 만; green=wake 미사용 / recovery=재발사 0) ─
    argv: List[str] = []
    if relay_required and not already_in_ledger:
        fire_kwargs: dict = dict(
            result_json_path=result_json_path,
            executor_key=executor_key,
            anu_key=anu_key,
        )
        if anu_keys is not None:
            fire_kwargs["anu_keys"] = anu_keys
        r = anu_runner_pickup_and_fire(**fire_kwargs)
        if not r.ok:
            return _fail(PICKUP_FAIL,
                         task_id=task_id,
                         result_json_path=result_json_path,
                         sha256=sha256,
                         classification=getattr(r, "classification", ""),
                         reasons=[
                             "anu_runner_pickup_and_fire FAIL (executor self-key refuse 포함)."
                         ] + list(r.reasons))
        argv = list(r.argv) if r.argv is not None else []

    closeout_action = CLOSEOUT_RELAY_PENDING if relay_required else CLOSEOUT_DONE_ACKED
    closeout_verdict = PICKUP_WAKE_BUILT if relay_required else PICKUP_CLOSEOUT_DONE

    # ── 8. dedupe ledger 기록 (ANCHOR-B 1순위 — fresh 만; recovery 는 이미 기록) ─
    if not already_in_ledger:
        ledger_event = LEDGER_EVENT_WAKE if relay_required else LEDGER_EVENT_CLOSEOUT
        ledger_entry = json.dumps({
            "schema": "dispatch.anu_result_pickup_runner.dedupe.v1",
            "event": ledger_event,
            "task_id": task_id,
            "sha256": sha256,
            "ts": _clock().isoformat(),
        }, ensure_ascii=False)
        try:
            os.makedirs(os.path.dirname(_ledger) or ".", exist_ok=True)
            with open(_ledger, "a", encoding="utf-8") as lfh:
                lfh.write(ledger_entry + "\n")
                lfh.flush()
                os.fsync(lfh.fileno())
        except OSError as exc:
            # ledger write 실패 → dedupe 기록 실패 시 duplicate wake/relay 방지 위해
            # fail-closed (wake 0). collector_result/marker 도 미작성. 다음 cycle 재시도.
            return _fail(PICKUP_LEDGER_WRITE_FAILED,
                         task_id=task_id,
                         result_json_path=result_json_path,
                         sha256=sha256,
                         reasons=[
                             f"LEDGER_WRITE_FAILED: dedupe ledger 기록 실패({exc}) — "
                             "duplicate wake 방지 위해 fail-closed (wake 0). 다음 cycle 재시도."
                         ])

    # ── 9. collector_result write (ANCHOR-B 2순위 — atomic os.replace) ────────
    collector = CollectorResult(
        task_id=task_id,
        source_result_sha256=sha256,
        verdict=closeout_verdict,
        closeout_action=closeout_action,
        owner_proof=_owner_proof,
        agent_relay=AgentRelay(required=relay_required, reason=relay.reason),
        dedupe=already_in_ledger,
        ts_kst=_clock().isoformat(),
    )
    cr_path = collector_result_path(result_json_path, task_id)
    try:
        write_collector_result(collector, cr_path)
    except OSError as exc:
        # collector_result before marker 불변 보존: collector_result write 실패 시
        # marker 미작성 → 다음 cycle 재시도(ledger dedupe 로 wake 재발사 0).
        return _fail(PICKUP_COLLECTOR_WRITE_FAILED,
                     task_id=task_id,
                     result_json_path=result_json_path,
                     sha256=sha256,
                     reasons=[
                         f"COLLECTOR_WRITE_FAILED: collector_result write 실패({exc}) — "
                         "marker 미작성, 다음 cycle 재시도(ledger dedupe 로 wake 재발사 0)."
                     ])

    # ── 10. done marker write (ANCHOR-B 3순위·마지막 — terminal sentinel) ─────
    done_content = json.dumps({
        "schema": PICKUP_SCHEMA,
        "event": LEDGER_EVENT_WAKE if relay_required else LEDGER_EVENT_CLOSEOUT,
        "task_id": task_id,
        "sha256": sha256,
        "closeout_action": closeout_action,
        "ts": _clock().isoformat(),
    }, ensure_ascii=False)
    done_path_out: Optional[str] = None
    marker_write_failed_reason: Optional[str] = None
    tmp_path = f"{done_path}.tmp-{os.getpid()}"
    try:
        with open(tmp_path, "w", encoding="utf-8") as dfh:
            dfh.write(done_content)
            dfh.flush()
            os.fsync(dfh.fileno())
        os.replace(tmp_path, done_path)
        _fsync_dir(result_dir or ".")  # marker(rename) crash-durable화 — terminal sentinel 유실 방지.
        done_path_out = done_path
    except OSError as exc:
        try:
            os.unlink(tmp_path)
        except OSError:
            pass
        # marker 실패 관측. ledger+collector_result 는 기록됨 → dedupe 로 duplicate
        # wake 차단(wake 1회 유지). verdict 유지하되 reasons 에 기록.
        marker_write_failed_reason = (
            f"MARKER_WRITE_FAILED: done marker 기록 실패({exc}) — "
            "ledger dedupe 로 duplicate wake 차단(wake 1회 유지)."
        )

    # ── 10.5 ANU-owned callback launch 코드강제 (task-2731 ANCHOR-1) ───────────
    #   collector_result + marker 가 기록된 closeout 직후 호출한다.
    #   executor 세션 생존 무관 · finish-task 끝단 무관 · dev bot prompt 의존 0.
    #   callback_launch_fn 미주입(기본 None) → 현행 보존(surface only) + ACTIVE=false.
    #   ★ fresh / recovery(already_in_ledger) 양쪽에서 시도 — marker 후 전이 실패가
    #   영구 skip 되지 않게 한다. at-most-once 는 audit jsonl dedupe(task_id, sha256)가
    #   보장(이중 callback 0). 예외는 closeout 을 막지 않는다(fail-safe).
    callback_launch_verdict, callback_audit_path = _enforce_callback(relay_required)

    # ── 11. verdict 반환 ───────────────────────────────────────────────────────
    if already_in_ledger:
        # recovery: ledger 기록은 이미 있음(wake/relay 1회 완료). collector_result/
        # marker 만 멱등 보강 — 신규 wake/relay 0 → SKIP_DEDUPE.
        final_reasons = [
            "dedupe: (task_id, sha256) ledger 기록 존재 — wake/relay 재발사 0. "
            "collector_result/marker 멱등 보강(crash recovery)."
        ]
        if marker_write_failed_reason:
            final_reasons.append(marker_write_failed_reason)
        return PickupResult(
            schema=PICKUP_SCHEMA,
            verdict=PICKUP_SKIP_DEDUPE,
            task_id=task_id,
            result_json_path=result_json_path,
            sha256=sha256,
            wake_built=False,
            argv=None,
            classification="",
            reasons=final_reasons,
            marker_path=done_path_out,
            closeout_action=closeout_action,
            closeout_path=cr_path,
            agent_relay_required=relay_required,
            agent_relay_reason=relay.reason,
            terminal_intent=terminal_intent,
            callback_schedule_created=callback_schedule_created,
            callback_launch_verdict=callback_launch_verdict,
            callback_audit_path=callback_audit_path,
        )

    if relay_required:
        final_reasons = [
            "agent_relay.required=true (relay_hints) → ANU-owned callback 발사 argv 생성 "
            "(relay-path). 실 발사/2-tier relay 는 권한 있는 ANU 세션이 수행 (자가발사 0)."
        ]
        if marker_write_failed_reason:
            final_reasons.append(marker_write_failed_reason)
        return PickupResult(
            schema=PICKUP_SCHEMA,
            verdict=PICKUP_WAKE_BUILT,
            task_id=task_id,
            result_json_path=result_json_path,
            sha256=sha256,
            wake_built=True,
            argv=argv,
            classification="",
            reasons=final_reasons,
            marker_path=done_path_out,
            closeout_action=closeout_action,
            closeout_path=cr_path,
            agent_relay_required=True,
            agent_relay_reason=relay.reason,
            terminal_intent=terminal_intent,
            callback_schedule_created=callback_schedule_created,
            callback_launch_verdict=callback_launch_verdict,
            callback_audit_path=callback_audit_path,
        )

    # green-path: 결정론 closeout 완료 (wake 0).
    final_reasons = [
        "green-path: agent_relay.required=false → deterministic closeout "
        "(collector_result done_acked + marker). wake argv 미사용 (wake 0)."
    ]
    if marker_write_failed_reason:
        final_reasons.append(marker_write_failed_reason)
    return PickupResult(
        schema=PICKUP_SCHEMA,
        verdict=PICKUP_CLOSEOUT_DONE,
        task_id=task_id,
        result_json_path=result_json_path,
        sha256=sha256,
        wake_built=False,
        argv=None,
        classification="",
        reasons=final_reasons,
        marker_path=done_path_out,
        closeout_action=closeout_action,
        closeout_path=cr_path,
        agent_relay_required=False,
        agent_relay_reason=relay.reason,
        terminal_intent=terminal_intent,
        callback_schedule_created=callback_schedule_created,
        callback_launch_verdict=callback_launch_verdict,
        callback_audit_path=callback_audit_path,
    )


# ═══════════════════════════════════════════════════════════════════════════════
# task-2741 — OS-level owner-proof gate (옵션 B, no-fire)
# ═══════════════════════════════════════════════════════════════════════════════
# OS-level ANU-owned pickup runner 가 cokacdir 발사 전에 통과해야 하는 sealed-key
# 기반 owner-proof gate. 본 경로는 pre_tool_use_v36 hook(rule_7) **밖**의 OS-level
# 경로이므로, 본 gate 가 rule_7 과 동등한 self-collector 차단을 자체적으로 보장한다
# (hook 의존 없는 이중 backstop).
#
# ★ owner proof 의 단일 신뢰원천 = runtime sealed-key loader 가 로드한 key 가 실제
#   독립 ANU key 인지의 구조적 검증(is_anu_key) + executor key 와 불일치. env 문자열·
#   collector_role=ANU 텍스트·executor 가 result.json 에 쓴 owner field 는 owner proof
#   로 **신뢰하지 않는다**(eligibility 판정에만 사용). proof 기록은 boolean/label 만 —
#   raw ANU key literal 0.
#
# ★ no-fire: 본 task 는 argv 구성까지만. 실 cokacdir --cron 호출(real-fire)·ACTIVE=true·
#   systemd/timer enable·callback schedule 직접 생성 0. real-fire 시도는 회장 승인
#   (approved)+ACTIVE(active) 가 모두 참이어야 하나 둘 다 기본 False 이고 본 task 에서
#   세팅하지 않으므로 항상 FIRE_BLOCKED_NO_APPROVAL 로 단락된다(구조적 fire 0).

# ── gate terminal status (회장 5 enum) ────────────────────────────────────────
GATE_OWNER_PICKUP_READY_NO_FIRE = "OWNER_PICKUP_READY_NO_FIRE"
GATE_DEDUPED = "DEDUPED"
GATE_RESULT_NOT_PICKUP_ELIGIBLE = "RESULT_NOT_PICKUP_ELIGIBLE"
GATE_FAIL_CLOSED_NON_ANU_KEY = "FAIL_CLOSED_NON_ANU_KEY"
GATE_FIRE_BLOCKED_NO_APPROVAL = "FIRE_BLOCKED_NO_APPROVAL"

GATE_TERMINAL_STATUSES = (
    GATE_OWNER_PICKUP_READY_NO_FIRE,
    GATE_DEDUPED,
    GATE_RESULT_NOT_PICKUP_ELIGIBLE,
    GATE_FAIL_CLOSED_NON_ANU_KEY,
    GATE_FIRE_BLOCKED_NO_APPROVAL,
)

# ── task-2742: controlled real-fire 1회 결선 terminal status (approved one-shot) ──
#   attempt_owner_fire 의 approved one-shot 실행 경로가 도달하는 추가 terminal status.
#   ★ GATE_TERMINAL_STATUSES(gate 자체 5 enum)는 의도적으로 확장하지 않는다 — gate
#   (owner_proof_pickup_gate)는 여전히 5 enum 만 반환하며, 아래 4 status 는 fire 경로
#   (attempt_owner_fire) 전용이다. 전체 9 enum 은 GATE_ALL_TERMINAL_STATUSES 로 합친다.
GATE_FIRE_ERROR = "FIRE_ERROR"
GATE_PARTIAL_FIRE_NO_SPAWN = "PARTIAL_FIRE_NO_SPAWN"
GATE_PARTIAL_FIRE_NO_CALLBACK_EVIDENCE = "PARTIAL_FIRE_NO_CALLBACK_EVIDENCE"
GATE_REALFIRE_1SHOT_SUCCESS = "REALFIRE_1SHOT_SUCCESS"

GATE_ONESHOT_FIRE_STATUSES = (
    GATE_FIRE_ERROR,
    GATE_PARTIAL_FIRE_NO_SPAWN,
    GATE_PARTIAL_FIRE_NO_CALLBACK_EVIDENCE,
    GATE_REALFIRE_1SHOT_SUCCESS,
)

# ── task-2774+8: D-2 fire wiring — approved 1-shot 전용 wrapper terminal status ──
#   run_approved_oneshot_fire(승인된 1-shot real-fire 의 유일한 진입점)가 도달하는 추가
#   terminal status 2종. ★ GATE_TERMINAL_STATUSES(gate 5 enum)·GATE_ONESHOT_FIRE_STATUSES
#   (fire 경로 4 enum)·GATE_ALL_TERMINAL_STATUSES(attempt_owner_fire 9 enum 계약)는 모두
#   의도적으로 **확장하지 않는다** — 아래 2종은 wrapper 전용이며 attempt_owner_fire 의 enum
#   계약을 바꾸지 않는다(2742/2743 회귀 0). 별도 tuple 로만 노출한다.
#
#   ★ NOOP_DISABLED          : ACTIVE gate(callback_launch flag, 함수인자 flag_reader) OFF →
#                              fire_fn 미주입·미호출·실 발사 0(전역 flag 파일 생성/읽기 0).
#   ★ FIRE_BLOCKED_NO_FIRE_FN : ACTIVE gate ON 이나 실발사 fire_fn 명시 주입 부재 → fail-closed.
#                              default 실발사 fire_fn 0(예상치 못한 경로 fire 차단)·fire_fn 미호출.
GATE_NOOP_DISABLED = "NOOP_DISABLED"
GATE_FIRE_BLOCKED_NO_FIRE_FN = "FIRE_BLOCKED_NO_FIRE_FN"
#   ★ INVALID_FLAG_READER (task-2774+11): flag_reader 가 None 아닌 non-callable(예: True,
#     object())이면 silent NOOP_DISABLED 로 삼키지 않고 **명시 fail-closed** 한다. silent
#     misconfiguration 추적 가능(fail-fast diagnostic). fire_fn 미호출·실 발사 0·argv None.
GATE_INVALID_FLAG_READER = "INVALID_FLAG_READER"

GATE_ONESHOT_WRAPPER_STATUSES = (
    GATE_NOOP_DISABLED,
    GATE_FIRE_BLOCKED_NO_FIRE_FN,
    GATE_INVALID_FLAG_READER,
)

# ── task-2743: late spawn reconciliation verdict (CONTROLLED_REALFIRE_1SHOT 사건 fix) ──
#   근본 결함: cron 등록 직후 spawn 미확인을 즉시 PARTIAL_FIRE_NO_SPAWN(terminal)으로
#   조기 확정 → late spawn(실발사)을 false negative 로 오판. fix: cron 등록 성공과 spawn
#   확인을 분리하고, 등록 성공·spawn 미확인은 **non-terminal** PENDING 으로 기록한 뒤,
#   stateless reconcile 로 late spawn evidence 확인 시 success 로 재평가한다.
#
#   ★ CRON_REGISTERED_PENDING_SPAWN 은 **non-terminal** — "현 시점 evidence 로 미확정"
#     이라는 기록일 뿐이다. 봇이 polling 으로 spawn 을 기다리는 상태가 아니다(회장
#     doctrine: session-bound polling loop 도입 금지). 따라서 GATE_*_TERMINAL_STATUSES
#     어디에도 포함하지 않는다.
GATE_CRON_REGISTERED_PENDING_SPAWN = "CRON_REGISTERED_PENDING_SPAWN"

#   ★ reconcile 가 late spawn evidence(전부 충족)를 확인했을 때의 terminal success.
#     LATE_SPAWN_CONFIRMED(late spawn 자체 확인) ≡ REALFIRE_1SHOT_SUCCESS_CONFIRMED
#     (controlled real-fire 1shot 성공 확정). 둘 다 terminal success-equivalent.
GATE_LATE_SPAWN_CONFIRMED = "LATE_SPAWN_CONFIRMED"
GATE_REALFIRE_1SHOT_SUCCESS_CONFIRMED = "REALFIRE_1SHOT_SUCCESS_CONFIRMED"

#   기존 PARTIAL evidence 를 **삭제하지 않고**(보존) reconcile 가 supersede 했음을
#   기록하는 marker status(append-only ledger event 의 status 값). 기존 PARTIAL 줄은
#   그대로 두고 별도 줄로 superseded marker 만 append 한다(삭제 0).
GATE_SUPERSEDED_BY_LATE_SPAWN = "superseded_by_late_spawn_confirmation"

# reconcile 가 도달하는 terminal success status 집합(stateless 재평가 결과).
GATE_RECONCILE_SUCCESS_STATUSES = (
    GATE_LATE_SPAWN_CONFIRMED,
    GATE_REALFIRE_1SHOT_SUCCESS_CONFIRMED,
)

# reconcile append-only ledger event 명(기존 OWNER_PICKUP_FIRED marker 와 구분). 새 fire
# 0 — reconcile 은 read-only verdict + 아래 두 event 의 append-only 기록만 남긴다.
GATE_LEDGER_EVENT_RECONCILED = "OWNER_PICKUP_LATE_SPAWN_RECONCILED"   # confirm marker
GATE_LEDGER_EVENT_SUPERSEDED = "OWNER_PICKUP_PARTIAL_SUPERSEDED"      # 기존 PARTIAL 보존 표식

# 회장 9 terminal status 전체 집합(gate 5 + one-shot fire 4). ★ fire 경로(attempt_owner_fire)
#   가 도달하는 terminal 집합으로 **불변** — reconcile success 2종과 non-terminal PENDING 은
#   여기 포함하지 않는다(task-2743 가 fire 경로 enum 계약을 바꾸지 않음).
GATE_ALL_TERMINAL_STATUSES = GATE_TERMINAL_STATUSES + GATE_ONESHOT_FIRE_STATUSES

# task-2743 — fire 9 terminal + reconcile success 2종(전체 verdict 의 terminal-equivalent
#   집합). CRON_REGISTERED_PENDING_SPAWN 은 non-terminal 이므로 여전히 제외.
GATE_ALL_VERDICT_TERMINAL_STATUSES = (
    GATE_ALL_TERMINAL_STATUSES + GATE_RECONCILE_SUCCESS_STATUSES
)

GATE_SCHEMA = "dispatch.anu_result_pickup_runner.owner_proof_gate.v1"

# proof label (raw key 0)
GATE_OWNER_PROOF_ANU = "ANU"

# dedupe ledger (task_id + head_sha marker) — gate 전용. no-fire 이므로 본 gate 는
# 읽기만 하고 marker 를 쓰지 않는다(marker write 는 승인된 real-fire 경로 소유).
DEFAULT_OWNER_GATE_LEDGER_PATH = os.path.join(
    CANONICAL_ROOT, "memory", "events", "owner_pickup_gate_index.jsonl"
)
GATE_LEDGER_EVENT = "OWNER_PICKUP_FIRED"

# argv placeholder (raw key 0): --key 자리에는 sealed-key placeholder 만. 실 key 는
# 회장 승인+ACTIVE 인 real-fire 경로가 sealed-key loader 로 마지막 순간 주입한다(본
# task 도달 0). at placeholder 도 마찬가지로 발사 시점에 결정.
OWNER_PICKUP_KEY_PLACEHOLDER = "<ANU_SEALED_KEY>"
OWNER_PICKUP_AT_PLACEHOLDER = "<AT_ON_APPROVED_FIRE>"

# ── task-2742: approved one-shot 고정 대상 (대상 result 1건에만 유효) ───────────
#   approved one-shot 승인은 아래 task_id + head_sha + absolute path 가 **모두 일치**
#   하는 단 1건에만 적용된다. 그 외(다른 task_id/head/path)에는 승인이 적용되지
#   않아 FIRE_BLOCKED_NO_APPROVAL 로 단락된다(승인 scope 한정). 테스트는 target 인자로
#   tmp 경로를 주입해 격리한다 — production 기본값은 아래 고정 fixture 이다.
ONESHOT_TARGET_TASK_ID = "task-realfire-pilot-260614"
# ★ task-2742-r3: head_sha 하드코드 상수 **제거** (squash merge catch-22 해소).
#   이전(r2)에는 ONESHOT_TARGET_HEAD_SHA = "4012bc50…" 고정 상수로 target head 를
#   못박았으나, task-2742 가 squash merge 되면 canonical HEAD 가 새 mergeCommit 으로
#   바뀌어 상수 head_sha 가 영구 stale → live real-fire 가 항상 FIRE_BLOCKED 되는
#   구조적 catch-22 였다. r3 부터 target head 는 **런타임 canonical HEAD read-only
#   probe** 로 동적 판정한다(attempt_owner_fire ① 참조). task_id·absolute path 는 고정.
ONESHOT_TARGET_REL_PATH = os.path.join(
    "memory", "events", "p0b_inbox", "task-realfire-pilot-260614.result.json"
)
ONESHOT_TARGET_ABS_PATH = os.path.join(CANONICAL_ROOT, ONESHOT_TARGET_REL_PATH)

# real-fire argv 의 --at 값(approved one-shot). 실 cron 1회(--once) 발사 시각 spec.
#   ★ 본 task 는 실 발사 0(fire_fn mock) — 이 값은 argv 형태 구성에만 쓰인다.
ONESHOT_AT_SPEC = "2m"
# real-fire argv 의 cron prompt tag(no-fire gate 와 구분).
ONESHOT_FIRE_TAG = "[task-2742][realfire-1shot]"

# ── task-2742-r4: owner pickup wake 발사 대상 chat (--chat 결선) ─────────────────
#   ★ 근본 결함(live real-fire PARTIAL_FIRE_NO_SPAWN): fire argv 에 `--chat <chat_id>`
#   가 누락되면 cokacdir --cron 이 대상 chat 미상으로 등록되지 않아 spawn 0 →
#   PARTIAL_FIRE_NO_SPAWN 이 된다. argv builder 에 --chat 을 결선해 real_argv·
#   redacted_argv 둘 다 --chat 을 싣고, chat_id 누락/None/empty 는 fail-closed(fire 0)
#   로 단락한다.
#   --chat 값은 ANU callback/envelope 가 도달할 **대상 chat** 기준이다.
#   ★ 격리: 아래 값은 **이번 controlled pilot 전용** 하드코드 상수다(단일 pilot chat).
#   production 다중 chat 운영에서는 호출자 주입/런타임 config 로 받아야 한다.
# TODO(production): controlled pilot 전용 하드코드 chat_id → 호출자 주입/런타임 config
#   분리(다중 chat·테넌트 대응). 현재는 단일 pilot chat 한정 결선.
ONESHOT_PILOT_CHAT_ID = "6937032012"


# ── OwnerProofGateResult dataclass ────────────────────────────────────────────
@dataclass
class OwnerProofGateResult:
    """owner_proof_pickup_gate 의 반환값. raw key 0 — boolean/label 만."""

    schema: str
    status: str            # GATE_TERMINAL_STATUSES 중 하나
    task_id: str
    result_json_path: str
    head_sha: str
    argv: Optional[List[str]]
    fire_blocked: bool     # real-fire 차단 여부 (본 task 항상 True)
    active: bool           # ACTIVE 상태 (본 task 항상 False)
    owner_key_proof_present: bool  # sealed-key ANU 검증 성립 시에만 True(label)
    schedule_owner_proof: str      # 성립 시 "ANU"(label), 아니면 ""
    eligibility: dict              # 관측된 eligibility boolean snapshot
    reasons: List[str] = field(default_factory=list)
    dedupe_marker_present: bool = False

    @property
    def ready(self) -> bool:
        return self.status == GATE_OWNER_PICKUP_READY_NO_FIRE

    def to_json(self) -> dict:
        return {
            "schema": self.schema,
            "status": self.status,
            "task_id": self.task_id,
            "result_json_path": self.result_json_path,
            "head_sha": self.head_sha,
            "argv": list(self.argv) if self.argv is not None else None,
            "fire_blocked": self.fire_blocked,
            "active": self.active,
            "owner_key_proof_present": self.owner_key_proof_present,
            "schedule_owner_proof": self.schedule_owner_proof,
            "eligibility": dict(self.eligibility),
            "reasons": list(self.reasons),
            "dedupe_marker_present": self.dedupe_marker_present,
        }


def _gate_fail(status: str, *, task_id: str = "", result_json_path: str = "",
               head_sha: str = "", eligibility: Optional[dict] = None,
               reasons: Optional[List[str]] = None,
               owner_key_proof_present: bool = False,
               schedule_owner_proof: str = "",
               dedupe_marker_present: bool = False,
               argv: Optional[List[str]] = None) -> OwnerProofGateResult:
    """no-fire(argv=None 기본) OwnerProofGateResult 편의 생성."""
    return OwnerProofGateResult(
        schema=GATE_SCHEMA, status=status, task_id=task_id,
        result_json_path=result_json_path, head_sha=head_sha,
        argv=argv, fire_blocked=True, active=False,
        owner_key_proof_present=owner_key_proof_present,
        schedule_owner_proof=schedule_owner_proof,
        eligibility=dict(eligibility or {}), reasons=list(reasons or []),
        dedupe_marker_present=dedupe_marker_present,
    )


# ── eligibility (result field 는 eligibility 에만 — owner proof 신뢰 0) ─────────
def _read_gate_eligibility(result: dict) -> dict:
    """result.json field 를 eligibility 판정용으로만 읽는다(owner proof 로 신뢰 0).

    canonical executor result(task-2740 A) 의 owner-pickup 적격 shape:
      - authoritative == False               (executor self-authority 주장 금지)
      - owner_pickup_required == True         (owner pickup 명시 요구)
      - owner_key_proof_present == False      (executor 가 key proof 주장 금지)
      - schedule_created_by_executor == False (executor self-schedule 금지)
      - callback_fired_by_executor == False   (executor self-collector 발사 금지)
    """
    return {
        "authoritative": bool(result.get("authoritative", False)),
        "owner_pickup_required": bool(result.get("owner_pickup_required", False)),
        "owner_key_proof_present": bool(result.get("owner_key_proof_present", False)),
        "schedule_created_by_executor": bool(result.get("schedule_created_by_executor", False)),
        "callback_fired_by_executor": bool(result.get("callback_fired_by_executor", False)),
    }


def _gate_eligibility_ok(elig: dict) -> tuple:
    """eligibility 통과 여부 + 위반 사유(fail-closed). rule_7 동등 self-collector 차단."""
    reasons: List[str] = []
    if elig["authoritative"]:
        reasons.append("authoritative=true — executor self-authority 주장(self-collector). 적격 아님.")
    if not elig["owner_pickup_required"]:
        reasons.append("owner_pickup_required!=true — owner pickup 미요구. 적격 아님.")
    if elig["owner_key_proof_present"]:
        reasons.append("owner_key_proof_present=true(result field) — executor key proof 주장 신뢰 금지. 적격 아님.")
    if elig["schedule_created_by_executor"]:
        reasons.append("schedule_created_by_executor=true — executor self-schedule. 적격 아님.")
    if elig["callback_fired_by_executor"]:
        reasons.append("callback_fired_by_executor=true — executor self-collector 발사. 적격 아님.")
    return (not reasons), reasons


# ── freshness / dedupe / argv helpers ─────────────────────────────────────────
def _default_head_probe() -> str:
    """canonical main head sha 조회(freshness 판정용·읽기 전용 git rev-parse).
    cokacdir/cron 호출 0. 실패 시 빈 문자열(fail-closed)."""
    try:
        import subprocess  # noqa: PLC0415 — read-only git rev-parse(발사/cron 0)
        out = subprocess.run(
            ["git", "-C", CANONICAL_ROOT, "rev-parse", "HEAD"],
            capture_output=True, text=True, timeout=10, check=False,
        )
        return (out.stdout or "").strip()
    except Exception:  # noqa: BLE001
        return ""


def _owner_gate_marker_present(ledger_path: str, task_id: str, head_sha: str) -> bool:
    """(task_id, head_sha) dedupe marker 존재 여부(읽기 전용). no-fire gate 는 marker
    를 쓰지 않는다 — marker write 는 승인된 real-fire 경로가 소유한다."""
    if not ledger_path or not os.path.isfile(ledger_path):
        return False
    try:
        with open(ledger_path, "r", encoding="utf-8") as lfh:
            for line in lfh:
                line = line.strip()
                if not line:
                    continue
                try:
                    entry = json.loads(line)
                except (ValueError, TypeError):
                    continue
                if (isinstance(entry, dict)
                        and entry.get("event") == GATE_LEDGER_EVENT
                        and entry.get("task_id") == task_id
                        and entry.get("head_sha") == head_sha):
                    return True
    except OSError:
        return False
    return False


def _write_owner_gate_marker(ledger_path: str, *, task_id: str, head_sha: str,
                             status: str, ts: str) -> bool:
    """approved one-shot real-fire 직후 재발사 방지 marker append (raw key 0).

    no-fire gate 는 marker 를 쓰지 않는다 — marker write 는 **승인된 real-fire 경로
    (attempt_owner_fire approved one-shot)만 소유**한다. marker 는 (task_id, head_sha)
    dedupe 키 + status + ts 만 담는다(raw ANU key/argv key 0). 실패 시 False 반환(관측).
    """
    entry = json.dumps({
        "schema": "dispatch.anu_result_pickup_runner.owner_gate.fire.v1",
        "event": GATE_LEDGER_EVENT,
        "task_id": task_id,
        "head_sha": head_sha,
        "status": status,
        "ts": ts,
    }, ensure_ascii=False)
    try:
        os.makedirs(os.path.dirname(ledger_path) or ".", exist_ok=True)
        with open(ledger_path, "a", encoding="utf-8") as lfh:
            lfh.write(entry + "\n")
            lfh.flush()
            os.fsync(lfh.fileno())
        return True
    except OSError:
        return False


def _read_fire_outcome(raw: object) -> tuple:
    """주입형 fire_fn 의 반환값 →
    (spawned, callback_evidence, cron_registered, schedule_id) 정규화.

    fire_fn 은 dict({"spawned":bool,"callback_evidence":bool,"cron_registered":bool,
    "schedule_id":str}) 또는 동명 attribute 를 가진 객체(또는 .returncode)를 반환할 수
    있다. fail-closed: 필드 부재/None 은 보수적 기본값.

      - spawned 부재 시 returncode==0 으로 유추(있을 때만), 아니면 False.
      - callback_evidence 부재 시 False(증거 없음으로 보수적 판정).

    ★ task-2743: **cron 등록 성공과 spawn 확인을 분리**한다(근본 fix). cokacdir --cron 이
      대상 chat 에 schedule 을 등록(register)하는 사건과, 등록된 schedule 이 owner session
      을 실제 spawn 하는 사건은 시간적으로 분리되며(late spawn), 후자는 등록 직후에는
      아직 관측되지 않을 수 있다. 따라서:
      - cron_registered 부재 시 보수적 유추: schedule_id 가 있거나 이미 spawned 가 True
        이면 등록은 성립한 것으로 본다(spawn 했다면 등록은 선행 사실). 둘 다 없으면
        False(등록 자체 실패로 보수 판정 → PARTIAL_FIRE_NO_SPAWN).
      - schedule_id 부재 시 빈 문자열.
      이 분리로 "cron 등록 자체 실패(PARTIAL_FIRE_NO_SPAWN)" 와 "등록 성공·spawn evidence
      pending(CRON_REGISTERED_PENDING_SPAWN, non-terminal)" 을 구분한다.
    """
    if raw is None:
        return False, False, False, ""

    def _get(key: str):
        if isinstance(raw, dict):
            return raw.get(key)
        return getattr(raw, key, None)

    spawned = _get("spawned")
    if spawned is None:
        rc = _get("returncode")
        spawned = (rc == 0) if rc is not None else False
    spawned = bool(spawned)

    cb = bool(_get("callback_evidence"))

    schedule_id = str(_get("schedule_id") or "").strip()

    cron_registered = _get("cron_registered")
    if cron_registered is None:
        # 명시값 부재 → 보수적 유추: spawn 했거나 schedule_id 발급됐으면 등록 성립.
        cron_registered = spawned or bool(schedule_id)
    cron_registered = bool(cron_registered)

    return spawned, cb, cron_registered, schedule_id


@dataclass(frozen=True)
class OneShotTarget:
    """approved one-shot 승인이 적용되는 고정 대상(task_id + absolute path).

    ★ task-2742-r3: head_sha 는 **target 상수가 아니다**. fire 판정 시 런타임 canonical
      HEAD read-only probe 로 동적 판정한다(squash merge catch-22 해소). 따라서 본
      dataclass 에는 head_sha 필드가 없다 — 승인 scope 는 task_id 정확 일치 + absolute
      path 정확 일치 2개로 고정하고, 3번째 검증(result.head_sha == current canonical HEAD
      full 40자 정확 일치)은 attempt_owner_fire 가 런타임 probe 로 수행한다."""

    task_id: str
    result_json_path: str  # absolute


# production 기본 고정 대상(task md §3-2). 테스트는 target 인자로 tmp 경로를 주입.
#   head_sha 는 런타임 canonical HEAD probe 로 판정하므로 target 에 싣지 않는다(r3).
ONESHOT_TARGET = OneShotTarget(
    task_id=ONESHOT_TARGET_TASK_ID,
    result_json_path=ONESHOT_TARGET_ABS_PATH,
)


def _build_owner_pickup_argv(*, task_id: str, head_sha: str,
                             result_json_path: str,
                             chat_id: str = ONESHOT_PILOT_CHAT_ID,
                             key_value: str = OWNER_PICKUP_KEY_PLACEHOLDER,
                             at_value: str = OWNER_PICKUP_AT_PLACEHOLDER,
                             tag: str = "[task-2741][no-fire]") -> List[str]:
    """cokacdir --cron 1회(--once) argv 구성.

    no-fire gate(owner_proof_pickup_gate)는 기본값(placeholder key/at + no-fire tag)으로
    호출하므로 argv 구성만·발사 0이다. task-2742 approved one-shot real-fire 경로는
    key_value=실 ANU key·at_value=실 spec·tag=realfire 로 호출해 **실 argv** 를 만들어
    주입형 fire_fn 에만 전달한다(persist/return 되는 argv 는 placeholder 로 redact).

    ★ task-2742-r4 --chat 결선: argv 에 `--chat <chat_id>` 를 싣는다(real_argv·
      redacted_argv 둘 다). chat_id 누락/None/empty 면 cokacdir --cron 이 대상 chat
      미상으로 등록되지 않아 spawn 0(PARTIAL_FIRE_NO_SPAWN) 가 되므로, 빈 chat_id 는
      **fail-closed(ValueError)** 로 거부해 argv 자체를 만들지 않는다. 호출자(fire 경로)
      가 사전에 검증·단락하지만 본 함수도 방어적으로 거부한다(빈 --chat argv 0).
      --chat 은 redaction 대상이 아니다(key 가 아니라 대상 chat 식별자).

    ★ raw ANU key 0: 본 함수 자체에 key literal 0. 기본 --key 는 sealed-key placeholder.
      실 key 는 호출자가 sealed-key loader 로 마지막 순간 로드해 key_value 로만 넘긴다.
      argv 는 ANU-owned pickup wake 발사 형태일 뿐 executor 가 자가 생성하는 callback
      schedule 이 아니다(callback schedule 직접 생성 0).
    """
    _chat = "" if chat_id is None else str(chat_id).strip()
    if not _chat:
        raise ValueError(
            "owner pickup argv fail-closed — --chat chat_id None/빈 문자열. 대상 chat "
            "미상이면 cokacdir --cron 미등록(spawn 0) → argv 미구성(빈 --chat 0)."
        )
    return [
        "cokacdir", "--cron",
        f"{tag} ANU owner pickup wake: {task_id}@{head_sha[:12]}",
        "--at", at_value,
        "--chat", _chat,
        "--key", key_value,
        "--once",
        "--owner-pickup-result", result_json_path,
    ]


# ── owner_proof_pickup_gate (공개 API) ────────────────────────────────────────
def owner_proof_pickup_gate(
    result_json_path: str,
    *,
    sealed_key_loader: Optional[Callable[[], Optional[str]]] = None,
    executor_key: str = "",
    anu_keys: Optional[Sequence[str]] = None,
    head_probe: Optional[Callable[[], str]] = None,
    dedupe_ledger_path: Optional[str] = None,
) -> OwnerProofGateResult:
    """OS-level owner-proof gate (fail-closed 6단계, argv 구성만·발사 0).

    단계:
      ① 명시 result.json 1건(glob 전체 스캔 0)
      ② eligibility(result field — owner proof 신뢰 0): authoritative=false·
         owner_pickup_required=true·owner_key_proof_present=false·
         schedule_created_by_executor=false·callback_fired_by_executor=false.
         위반 → RESULT_NOT_PICKUP_ELIGIBLE (rule_7 동등 self-collector 차단).
      ③ sealed-key loader 검증(runtime, raw 0): 로드 key 가 비어있지 않고 독립 ANU
         key set 에 속하며 executor key 와 다를 때만 owner proof 성립. 아니면
         (부재/dev self-key/executor key/비-ANU/위조 env) → FAIL_CLOSED_NON_ANU_KEY.
      ④ freshness: result.head_sha != canonical head(stale/superseded) →
         RESULT_NOT_PICKUP_ELIGIBLE.
      ⑤ dedupe: (task_id, head_sha) marker 존재 → DEDUPED (fire 0).
      ⑥ 통과 → argv 구성만 → OWNER_PICKUP_READY_NO_FIRE (실 cron 발사 0).
    """
    _anu_keys = tuple(anu_keys) if anu_keys is not None else tuple(DEFAULT_ANU_KEYS)
    _ledger = dedupe_ledger_path or DEFAULT_OWNER_GATE_LEDGER_PATH

    # ── ① 명시 result.json 1건 (glob 전체 스캔 0) ─────────────────────────────
    if not result_json_path or not isinstance(result_json_path, str):
        return _gate_fail(GATE_RESULT_NOT_PICKUP_ELIGIBLE,
                          reasons=["명시 result.json 경로 인자 부재 — gate 는 단일 명시 "
                                   "인자만 처리(glob 전체 스캔 0)."])
    if not os.path.isfile(result_json_path):
        return _gate_fail(GATE_RESULT_NOT_PICKUP_ELIGIBLE,
                          result_json_path=result_json_path,
                          reasons=["명시 result.json 부재 — executor 미작성/경로 오류. fail-closed."])
    try:
        with open(result_json_path, "rb") as fh:
            raw_bytes = fh.read()
        result: object = json.loads(raw_bytes.decode("utf-8"))
    except (OSError, ValueError) as exc:
        return _gate_fail(GATE_RESULT_NOT_PICKUP_ELIGIBLE,
                          result_json_path=result_json_path,
                          reasons=[f"result.json 읽기/파싱 실패: {exc} — fail-closed."])
    if not isinstance(result, dict):
        return _gate_fail(GATE_RESULT_NOT_PICKUP_ELIGIBLE,
                          result_json_path=result_json_path,
                          reasons=["result.json 이 object(dict) 아님 — fail-closed."])

    task_id = str(result.get("task_id") or "").strip()
    head_sha = str(result.get("head_sha") or "").strip()
    if not task_id:
        return _gate_fail(GATE_RESULT_NOT_PICKUP_ELIGIBLE,
                          result_json_path=result_json_path, head_sha=head_sha,
                          reasons=["result.json task_id 없음/빈값 — fail-closed."])

    # ── ② eligibility (result field 는 eligibility 에만, owner proof 신뢰 0) ────
    elig = _read_gate_eligibility(result)
    elig_ok, elig_reasons = _gate_eligibility_ok(elig)
    if not elig_ok:
        return _gate_fail(GATE_RESULT_NOT_PICKUP_ELIGIBLE,
                          task_id=task_id, result_json_path=result_json_path,
                          head_sha=head_sha, eligibility=elig,
                          reasons=["eligibility fail-closed (rule_7 동등 self-collector 차단)."]
                          + elig_reasons)

    # ── ③ sealed-key loader 검증 (runtime, raw 0) + ANU key 검증 ───────────────
    #   owner proof 의 단일 신뢰원천. env 문자열/result field 무관 — 로더가 로드한 key
    #   가 (a) 비어있지 않고 (b) 독립 ANU key set 에 속하며 (c) executor key 와 다를
    #   때만 proof 성립. 아니면 FAIL_CLOSED_NON_ANU_KEY.
    _loader = sealed_key_loader if sealed_key_loader is not None else _default_sealed_key_loader
    try:
        runtime_key = _loader()
    except Exception:  # noqa: BLE001 — loader 실패는 fail-closed
        runtime_key = None
    if not runtime_key:
        return _gate_fail(GATE_FAIL_CLOSED_NON_ANU_KEY,
                          task_id=task_id, result_json_path=result_json_path,
                          head_sha=head_sha, eligibility=elig,
                          reasons=["sealed ANU key 미로드(부재) — owner proof 불성립. fail-closed."])
    if not is_anu_key(runtime_key, _anu_keys):
        # dev self-key / executor key / 비-ANU key (위조 env 포함) — proof 불성립.
        return _gate_fail(GATE_FAIL_CLOSED_NON_ANU_KEY,
                          task_id=task_id, result_json_path=result_json_path,
                          head_sha=head_sha, eligibility=elig,
                          reasons=["로드 key 가 독립 ANU key 아님(dev self-key/executor key/"
                                   "비-ANU/위조 env). owner proof 불성립. fail-closed."])
    if executor_key and runtime_key == executor_key:
        # 런타임 key == executor key → executor 가 gate 를 self 실행. self-collector.
        return _gate_fail(GATE_FAIL_CLOSED_NON_ANU_KEY,
                          task_id=task_id, result_json_path=result_json_path,
                          head_sha=head_sha, eligibility=elig,
                          reasons=["로드 key == executor key — executor self-collector 실행. fail-closed."])
    # owner proof 성립 — boolean/label 만 기록(raw key 0).
    owner_key_proof_present = True
    schedule_owner_proof = GATE_OWNER_PROOF_ANU

    # ── ④ freshness (stale/superseded head mismatch → 적격 아님) ──────────────
    _probe = head_probe if head_probe is not None else _default_head_probe
    try:
        current_head = str(_probe() or "").strip()
    except Exception:  # noqa: BLE001
        current_head = ""
    if not head_sha or not current_head or head_sha != current_head:
        return _gate_fail(GATE_RESULT_NOT_PICKUP_ELIGIBLE,
                          task_id=task_id, result_json_path=result_json_path,
                          head_sha=head_sha, eligibility=elig,
                          owner_key_proof_present=owner_key_proof_present,
                          schedule_owner_proof=schedule_owner_proof,
                          reasons=[
                              "freshness fail-closed — result head_sha 가 canonical head 와 "
                              "불일치(stale/superseded) 또는 head 미상. 적격 아님.",
                              f"result.head_sha={head_sha or '∅'} / canonical_head={current_head or '∅'}",
                          ])

    # ── ⑤ dedupe (task_id + head_sha marker → DEDUPED, fire 0) ────────────────
    if _owner_gate_marker_present(_ledger, task_id, head_sha):
        return _gate_fail(GATE_DEDUPED,
                          task_id=task_id, result_json_path=result_json_path,
                          head_sha=head_sha, eligibility=elig,
                          owner_key_proof_present=owner_key_proof_present,
                          schedule_owner_proof=schedule_owner_proof,
                          dedupe_marker_present=True,
                          reasons=["dedupe: (task_id, head_sha) marker 존재 — 이미 처리됨. fire 0."])

    # ── ⑥ argv 구성만 (실 cron 발사 0) → OWNER_PICKUP_READY_NO_FIRE ────────────
    argv = _build_owner_pickup_argv(task_id=task_id, head_sha=head_sha,
                                    result_json_path=result_json_path)
    return OwnerProofGateResult(
        schema=GATE_SCHEMA,
        status=GATE_OWNER_PICKUP_READY_NO_FIRE,
        task_id=task_id,
        result_json_path=result_json_path,
        head_sha=head_sha,
        argv=argv,
        fire_blocked=True,   # no-fire: argv 구성만, 실 발사 0(승인 전).
        active=False,
        owner_key_proof_present=owner_key_proof_present,
        schedule_owner_proof=schedule_owner_proof,
        eligibility=elig,
        reasons=[
            "gate PASS — sealed-key ANU owner proof 성립·eligible·fresh·non-dup. argv "
            "구성 완료(실 cokacdir --cron 발사 0). real-fire 는 회장 승인+ACTIVE 필요.",
        ],
        dedupe_marker_present=False,
    )


def build_owner_gate_fn(
    *,
    sealed_key_loader: Optional[Callable[[], Optional[str]]] = None,
    executor_key: str = "",
    anu_keys: Optional[Sequence[str]] = None,
    head_probe: Optional[Callable[[], str]] = None,
    dedupe_ledger_path: Optional[str] = None,
) -> Callable[..., OwnerProofGateResult]:
    """fire-capable entrypoint(anu_pickup_driver.main/scan_once/process_one, CLI pickup)가
    pickup_once 에 주입하는, 설정이 bound 된 owner_proof_pickup_gate 호출자.

    pickup_once 는 owner_gate_fn(result_json_path) 한 가지 형태로만 호출하므로(우회 0),
    sealed_key_loader/executor_key/anu_keys/head_probe/dedupe_ledger_path 설정은 본 빌더가
    bound 한다. 반환 callable 은 owner-gate subcommand 와 **동일한 owner_proof_pickup_gate**
    를 호출한다(pickup/process_one/owner-gate 동일 gate 정책 공유). raw ANU key 0.

    scan/bridge 경로는 owner_gate_fn(result_json_path, executor_key=...) 로 call-time
    executor_key 를 전달할 수 있으며 _gate 가 이를 수용해 self-collector 방어에 반영
    (비어있으면 bound 값 폴백)."""
    # factory-bound executor_key 를 캡처(아래 _gate 의 동명 kwarg 가 가리지 않도록).
    bound_executor_key = executor_key
    def _gate(result_json_path: str, *, executor_key: Optional[str] = None) -> OwnerProofGateResult:
        # call-time executor_key 정합: 비어있지 않은 call-time 값이 들어오면 우선 사용(self-collector
        # 방어 입력 — 무시로 약화 금지). 없거나 빈 문자열이면 factory-bound 값으로 폴백.
        # owner_proof_pickup_gate ③단계(executor_key and runtime_key == executor_key → FAIL_CLOSED)
        # 정책 보존. driver 경로는 bound·call 둘 다 "" → effective "" → 동작 동일(회귀 0).
        effective_executor_key = executor_key if (isinstance(executor_key, str) and executor_key) else bound_executor_key
        return owner_proof_pickup_gate(
            result_json_path,
            sealed_key_loader=sealed_key_loader,
            executor_key=effective_executor_key,
            anu_keys=anu_keys,
            head_probe=head_probe,
            dedupe_ledger_path=dedupe_ledger_path,
        )
    return _gate


def _fire_blocked_no_approval(gate_result: OwnerProofGateResult, *,
                              active: bool,
                              extra_reasons: Optional[List[str]] = None,
                              ) -> OwnerProofGateResult:
    """FIRE_BLOCKED_NO_APPROVAL 결과 생성(argv 는 gate 의 placeholder argv 그대로·실 발사 0)."""
    return OwnerProofGateResult(
        schema=gate_result.schema,
        status=GATE_FIRE_BLOCKED_NO_APPROVAL,
        task_id=gate_result.task_id,
        result_json_path=gate_result.result_json_path,
        head_sha=gate_result.head_sha,
        argv=gate_result.argv,
        fire_blocked=True,
        active=bool(active),
        owner_key_proof_present=gate_result.owner_key_proof_present,
        schedule_owner_proof=gate_result.schedule_owner_proof,
        eligibility=dict(gate_result.eligibility or {}),
        reasons=list(gate_result.reasons or []) + list(extra_reasons or []),
        dedupe_marker_present=gate_result.dedupe_marker_present,
    )


def attempt_owner_fire(
    gate_result: OwnerProofGateResult,
    *,
    approved: bool = False,
    active: bool = False,
    approved_one_shot: bool = False,
    fire_fn: Optional[Callable[[Sequence[str]], object]] = None,
    sealed_key_loader: Optional[Callable[[], Optional[str]]] = None,
    executor_key: str = "",
    anu_keys: Optional[Sequence[str]] = None,
    head_probe: Optional[Callable[[], str]] = None,
    dedupe_ledger_path: Optional[str] = None,
    target: OneShotTarget = ONESHOT_TARGET,
    at_spec: str = ONESHOT_AT_SPEC,
    chat_id: str = ONESHOT_PILOT_CHAT_ID,
    clock: Optional[Callable[[], datetime]] = None,
) -> OwnerProofGateResult:
    """real-fire 시도 진입점.

    두 경로로 분기한다:

    A) **persistent ACTIVE 경로(레거시·무결선)**: approved_one_shot 미지정 시. 회장 승인
       (approved)+persistent ACTIVE(active) 가 모두 참이어도 persistent driver 실 발사
       결선은 본 모듈에 **부재**하다(구조적 fire 0) — fail-closed RuntimeError. 그 외는
       FIRE_BLOCKED_NO_APPROVAL(no-fire). persistent ACTIVE 는 본 task 에서 전환하지
       않는다(p0b_driver_enabled/systemd 무관·ACTIVE=false 유지).

    B) **task-2742 approved one-shot 경로(controlled real-fire 1회 결선)**:
       approved_one_shot=True 시. persistent ACTIVE 와 **완전히 분리**된 1회성 승인이다
       (active 인자/ACTIVE 상태 무관·세팅 0). 아래를 모두 만족할 때만 주입형 fire_fn 으로
       cokacdir --cron 1회를 **실제 생성**할 수 있다:
         1. 대상 result 1건 일치 (task-2742-r3 runtime canonical HEAD) — 3개 검증:
            (a) gate_result.task_id == target.task_id **정확 일치**,
            (b) gate_result absolute path == target absolute path **정확 일치**,
            (c) gate_result.head_sha(=result.head_sha) == **런타임 canonical HEAD probe
                full 40자 정확 일치**(== only). head_sha 는 더 이상 target 상수가 아니라
                head_probe(기본 read-only `git rev-parse HEAD`)로 동적 판정한다 — squash
                merge 로 canonical HEAD 가 바뀌어도 stale 상수 catch-22 가 사라진다.
                ★ prefix/startswith/short SHA 완화 0 — len==40 + 문자열 동일성 둘 다 요구.
            · canonical HEAD probe 실패(빈 문자열/40자 아님) → fail-closed(fire 0,
              RESULT_NOT_PICKUP_ELIGIBLE). (a)/(b) 불일치 → FIRE_BLOCKED_NO_APPROVAL.
              (c) 불일치 → RESULT_NOT_PICKUP_ELIGIBLE(stale/superseded).
         2. gate READY 재확인 — owner_proof_pickup_gate 재실행 후 READY 아니면 해당
            status 전파(stale→RESULT_NOT_PICKUP_ELIGIBLE / dup→DEDUPED / 위조·부재→
            FAIL_CLOSED_NON_ANU_KEY).
         3. duplicate marker 부재 재확인 — 존재 시 DEDUPED.
       이후 sealed-key loader 로 마지막 순간 ANU key 를 로드해 **실 argv** 를 구성하고
       fire_fn(real_argv)을 **정확히 1회** 호출한다. 성공/부분실패/에러 후 재발사 방지
       marker + terminal status 를 기록한다.

       ★ argv result path 견고화(task-2742-r2): argv 의 --owner-pickup-result 자리는
         호출자가 넘긴 re_gate.result_json_path(relative 가능) 가 아니라 scope 검사에서
         이미 검증된 **absolute t_path_abs** 를 사용한다 — cokacdir 가 다른 CWD 에서
         실행돼도 경로 resolution 이 견고하다. real_argv·redacted_argv 둘 다 동일한
         t_path_abs 기반이다. result_json_path(target/gate_result)가 None/빈 문자열이면
         abspath 계산 전에 명시적 fail-closed 한다(argv 미구성·fire_fn 미호출).

       ★ 실 live cron fire 0(본 task): subprocess 호출부는 fire_fn 으로 **주입**되며,
         본 모듈은 default fire_fn 을 두지 않는다(fire_fn 미주입 → 실 발사 0·FIRE_ERROR).
         tests 는 mock fire_fn 으로 1회 호출·argv 형태를 검증한다. 실 pilot 발사는
         별도 회장 승인에서만 실 fire_fn 을 주입한다.

       ★ raw ANU key 0: 실 key 는 fire_fn 에 넘기는 ephemeral real_argv 에만 존재하고,
         return/persist 되는 OwnerProofGateResult.argv 와 marker 에는 placeholder 로
         redact 한다. executor/dev key 사용 0(sealed loader 의 독립 ANU key 만).
    """
    # ── 경로 A: 레거시 persistent ACTIVE (무결선·no-fire) ───────────────────────
    if not approved_one_shot:
        if not (approved and active):
            return _fire_blocked_no_approval(
                gate_result, active=active,
                extra_reasons=[
                    "real-fire BLOCKED — 회장 승인(approved)+ACTIVE(active) 미충족 또는 "
                    "approved_one_shot 미지정. actual cokacdir --cron 호출 0 (no-fire).",
                ])
        # persistent ACTIVE 실 발사 결선 부재(본 task 무결선) — fail-closed raise.
        raise RuntimeError(
            "owner pickup persistent-ACTIVE real-fire 결선 부재 — 회장 승인 후 별도 task. "
            "본 task 는 approved one-shot(approved_one_shot=True) 경로만 결선한다."
        )

    # ── 경로 B: approved one-shot controlled real-fire (task-2742) ──────────────
    _clock = clock or (lambda: datetime.now(timezone.utc))
    _ledger = dedupe_ledger_path or DEFAULT_OWNER_GATE_LEDGER_PATH

    # ①-pre. result_json_path None/빈 문자열 명시적 fail-closed (argv 견고화) ───────
    #   ★ argv 의 --owner-pickup-result 자리는 **검증된 absolute path** 만 받는다. 빈/None
    #   경로를 그대로 os.path.abspath 에 넘기면 abspath("")=CWD 로 둔갑해 (a) scope 검사가
    #   양쪽 CWD 로 spurious 일치하거나 (b) 잘못된 CWD 경로가 argv 에 실릴 수 있다. 따라서
    #   target.result_json_path / gate_result.result_json_path 가 None/빈 문자열(공백 only
    #   포함)이면 abspath 계산 **전에** 즉시 fail-closed 한다(fire 0·argv 미구성).
    if not target.result_json_path or not str(target.result_json_path).strip():
        return _fire_blocked_no_approval(
            gate_result, active=False,
            extra_reasons=[
                "approved one-shot fail-closed — target.result_json_path None/빈 문자열. "
                "검증된 absolute path 부재 → argv 미구성·fire_fn 미호출(fire 0).",
            ])
    if not gate_result.result_json_path or not str(gate_result.result_json_path).strip():
        return _fire_blocked_no_approval(
            gate_result, active=False,
            extra_reasons=[
                "approved one-shot fail-closed — gate_result.result_json_path None/빈 문자열. "
                "검증된 absolute path 부재 → argv 미구성·fire_fn 미호출(fire 0).",
            ])

    # ①-pre2. --chat 결선 fail-closed(task-2742-r4): chat_id None/빈 문자열(공백 only 포함)
    #   이면 cokacdir --cron 이 대상 chat 미상으로 등록되지 않아 spawn 0(PARTIAL_FIRE_NO_SPAWN)
    #   가 된다. 따라서 abspath/key 로드 **전에** 즉시 fail-closed 한다(argv 미구성·fire_fn
    #   미호출·fire 0). 실 live real-fire PARTIAL_FIRE_NO_SPAWN 의 근본 fix.
    _chat_id = "" if chat_id is None else str(chat_id).strip()
    if not _chat_id:
        return _fire_blocked_no_approval(
            gate_result, active=False,
            extra_reasons=[
                "approved one-shot fail-closed — 발사 대상 --chat chat_id None/빈 문자열. "
                "대상 chat 미상이면 cokacdir --cron 미등록(spawn 0) → argv 미구성·fire_fn "
                "미호출(fire 0). actual cokacdir --cron 호출 0.",
            ])

    # ① 대상 result 1건 일치 (task-2742-r3 runtime canonical HEAD) ─────────────────
    #   3개 검증: (a) task_id 정확 일치 + (b) absolute path 정확 일치 + (c) result.head_sha
    #   == 런타임 canonical HEAD probe full 40자 정확 일치(== only). head_sha 는 더 이상
    #   target 상수가 아니라 head_probe(기본 read-only git rev-parse HEAD)로 동적 판정한다
    #   — squash merge 로 canonical HEAD 가 바뀌어도 stale 상수 catch-22 가 사라진다.
    #   ★ prefix/startswith/short SHA 완화 0: len==40 + 문자열 동일성 둘 다 요구.
    _probe = head_probe if head_probe is not None else _default_head_probe
    try:
        current_head = str(_probe() or "").strip()
    except Exception:  # noqa: BLE001 — probe 예외는 fail-closed
        current_head = ""
    # canonical HEAD probe 실패(빈 문자열 / full 40자 아님) → fail-closed(fire 0).
    if not current_head or len(current_head) != 40:
        return _gate_fail(
            GATE_RESULT_NOT_PICKUP_ELIGIBLE,
            task_id=str(gate_result.task_id or ""),
            result_json_path=gate_result.result_json_path,
            head_sha=str(gate_result.head_sha or ""),
            eligibility=gate_result.eligibility,
            owner_key_proof_present=gate_result.owner_key_proof_present,
            schedule_owner_proof=gate_result.schedule_owner_proof,
            reasons=[
                "approved one-shot fail-closed — canonical HEAD probe 실패/비정상(full 40자 "
                f"아님: probe len={len(current_head)}). target head 동적 판정 불가 → fire 0. "
                "actual cokacdir --cron 호출 0.",
            ])

    g_task = str(gate_result.task_id or "")
    g_head = str(gate_result.head_sha or "")
    g_path_abs = os.path.abspath(gate_result.result_json_path)
    t_path_abs = os.path.abspath(target.result_json_path)

    # (a)(b): task_id + absolute path 정확 일치(scope 한정). 불일치 → FIRE_BLOCKED_NO_APPROVAL.
    if not (g_task == target.task_id and g_path_abs == t_path_abs):
        return _fire_blocked_no_approval(
            gate_result, active=False,
            extra_reasons=[
                "approved one-shot scope 불일치 — 승인은 고정 대상 1건(task_id 정확 일치 + "
                "absolute path 정확 일치)에만 유효. 대상 아님 → FIRE_BLOCKED_NO_APPROVAL. "
                f"got(task_id={g_task or '∅'}, path={g_path_abs}) / "
                f"target(task_id={target.task_id}, path={t_path_abs}). "
                "actual cokacdir --cron 호출 0.",
            ])

    # (c): result.head_sha == current canonical HEAD full 40자 정확 일치(== only).
    #   ★ prefix/startswith/short SHA 완화 0 — len==40 + 동일성. 불일치(stale/superseded/
    #   short SHA) → RESULT_NOT_PICKUP_ELIGIBLE(fire 0).
    if not (len(g_head) == 40 and g_head == current_head):
        return _gate_fail(
            GATE_RESULT_NOT_PICKUP_ELIGIBLE,
            task_id=g_task,
            result_json_path=gate_result.result_json_path,
            head_sha=g_head,
            eligibility=gate_result.eligibility,
            owner_key_proof_present=gate_result.owner_key_proof_present,
            schedule_owner_proof=gate_result.schedule_owner_proof,
            reasons=[
                "approved one-shot head 불일치 — result.head_sha 가 current canonical HEAD "
                "full 40자와 정확 일치하지 않음(stale/superseded/short SHA·prefix). "
                "== only·prefix 완화 0 → RESULT_NOT_PICKUP_ELIGIBLE(fire 0). "
                f"result.head_sha={g_head or '∅'}(len={len(g_head)}) / "
                f"canonical_head={current_head}(len={len(current_head)}). "
                "actual cokacdir --cron 호출 0.",
            ])

    # ★ 검증된 canonical HEAD(=g_head==current_head, full 40자). 이후 marker/argv 의
    #   head 식별자는 target 상수가 아니라 이 런타임 검증값을 사용한다.
    verified_head = current_head

    # ② gate READY 재확인 — owner_proof_pickup_gate 재실행(발사 직전 최신 상태).
    re_gate = owner_proof_pickup_gate(
        gate_result.result_json_path,
        sealed_key_loader=sealed_key_loader,
        executor_key=executor_key,
        anu_keys=anu_keys,
        head_probe=head_probe,
        dedupe_ledger_path=_ledger,
    )
    if re_gate.status != GATE_OWNER_PICKUP_READY_NO_FIRE:
        # gate 미PASS status(RESULT_NOT_PICKUP_ELIGIBLE/DEDUPED/FAIL_CLOSED_NON_ANU_KEY)
        # 를 그대로 전파한다 — fire 0(actual cokacdir --cron 호출 0).
        return OwnerProofGateResult(
            schema=re_gate.schema,
            status=re_gate.status,
            task_id=re_gate.task_id,
            result_json_path=re_gate.result_json_path,
            head_sha=re_gate.head_sha,
            argv=re_gate.argv,
            fire_blocked=True,
            active=False,
            owner_key_proof_present=re_gate.owner_key_proof_present,
            schedule_owner_proof=re_gate.schedule_owner_proof,
            eligibility=dict(re_gate.eligibility),
            reasons=list(re_gate.reasons) + [
                "approved one-shot 발사 직전 gate READY 재확인 실패 — 위 status 전파. "
                "actual cokacdir --cron 호출 0.",
            ],
            dedupe_marker_present=re_gate.dedupe_marker_present,
        )

    # ③ duplicate marker 부재 재확인(belt-and-suspenders; gate 도 dedupe 검사).
    #   ★ dedupe 키의 head 식별자는 런타임 검증값 verified_head(=current canonical HEAD).
    if _owner_gate_marker_present(_ledger, target.task_id, verified_head):
        return OwnerProofGateResult(
            schema=re_gate.schema, status=GATE_DEDUPED,
            task_id=target.task_id, result_json_path=re_gate.result_json_path,
            head_sha=verified_head, argv=None, fire_blocked=True, active=False,
            owner_key_proof_present=re_gate.owner_key_proof_present,
            schedule_owner_proof=re_gate.schedule_owner_proof,
            eligibility=dict(re_gate.eligibility),
            reasons=["approved one-shot 발사 직전 dup marker 재확인 — (task_id, head_sha) "
                     "marker 존재. 재발사 방지 → DEDUPED. actual cokacdir --cron 호출 0."],
            dedupe_marker_present=True,
        )

    # ④ 실 ANU key 로드(마지막 순간) — fire_fn real_argv 에만 ephemeral 사용(persist 0).
    _loader = sealed_key_loader if sealed_key_loader is not None else _default_sealed_key_loader
    try:
        anu_key = _loader()
    except Exception:  # noqa: BLE001 — loader 실패는 fail-closed
        anu_key = None
    if not anu_key:
        # gate ③ 가 이미 검증했으나 재로드 실패 방어 — FAIL_CLOSED(fire 0).
        return OwnerProofGateResult(
            schema=re_gate.schema, status=GATE_FAIL_CLOSED_NON_ANU_KEY,
            task_id=target.task_id, result_json_path=re_gate.result_json_path,
            head_sha=verified_head, argv=None, fire_blocked=True, active=False,
            owner_key_proof_present=False, schedule_owner_proof="",
            eligibility=dict(re_gate.eligibility),
            reasons=["approved one-shot 발사 직전 sealed ANU key 재로드 실패 — fail-closed. "
                     "actual cokacdir --cron 호출 0."],
            dedupe_marker_present=False,
        )

    # ⑤ argv 구성 — real_argv(실 key·fire_fn 전용)와 redacted_argv(placeholder·persist용).
    #   ★ result path 는 re_gate.result_json_path(호출자가 넘긴 relative 가능) 대신, 위
    #   scope 검사에서 이미 검증된 **absolute t_path_abs** 를 사용한다(cokacdir 가 다른 CWD
    #   에서 실행돼도 경로 resolution 견고). real_argv·redacted_argv 둘 다 동일한 t_path_abs
    #   기반이며, t_path_abs 는 step ① 에서 g_path_abs 와 정확 일치함이 검증된 값이다.
    real_argv = _build_owner_pickup_argv(
        task_id=target.task_id, head_sha=verified_head,
        result_json_path=t_path_abs, chat_id=_chat_id,
        key_value=anu_key, at_value=at_spec, tag=ONESHOT_FIRE_TAG,
    )
    redacted_argv = _build_owner_pickup_argv(
        task_id=target.task_id, head_sha=verified_head,
        result_json_path=t_path_abs, chat_id=_chat_id,
        key_value=OWNER_PICKUP_KEY_PLACEHOLDER, at_value=at_spec, tag=ONESHOT_FIRE_TAG,
    )

    # ⑥ fire_fn 미주입 → 실 발사 경로 부재(본 task 기본). 실 발사 0·marker 미작성
    #    (legit pilot 의 fire_fn 주입을 막지 않음) → FIRE_ERROR.
    if fire_fn is None:
        return OwnerProofGateResult(
            schema=re_gate.schema, status=GATE_FIRE_ERROR,
            task_id=target.task_id, result_json_path=re_gate.result_json_path,
            head_sha=verified_head, argv=redacted_argv, fire_blocked=True, active=False,
            owner_key_proof_present=re_gate.owner_key_proof_present,
            schedule_owner_proof=re_gate.schedule_owner_proof,
            eligibility=dict(re_gate.eligibility),
            reasons=["approved one-shot READY 이나 주입형 fire_fn 부재 — 실 발사 0(FIRE_ERROR). "
                     "재발사 marker 미작성(별도 승인 pilot 의 fire_fn 주입 보존)."],
            dedupe_marker_present=False,
        )

    # ⑦ fire_fn(real_argv) 정확히 1회 호출(주입형 subprocess). 예외 → FIRE_ERROR.
    fire_exception = False
    try:
        fire_raw = fire_fn(real_argv)
    except Exception:  # noqa: BLE001 — fire_fn 예외는 FIRE_ERROR(고정 라벨·raw key persist 0)
        fire_raw = None
        fire_exception = True

    if fire_exception:
        status = GATE_FIRE_ERROR
        fire_reason = ("approved one-shot fire_fn 호출 중 예외 — FIRE_ERROR. fire_fn 1회 "
                       "호출됨(uncertain). 재발사 방지 marker 기록(double-fire 0).")
    else:
        # ★ task-2743: cron 등록 성공과 spawn 확인을 **분리** 판정한다(근본 fix).
        #   기존 결함: spawned False 면 즉시 PARTIAL_FIRE_NO_SPAWN(terminal)으로 조기
        #   확정 → late spawn(등록 직후엔 미관측이나 이후 실발사)을 false negative 로
        #   오판했다. fix 분기:
        #     (1) cron 미등록 → PARTIAL_FIRE_NO_SPAWN (cron 등록 자체 실패, terminal).
        #     (2) cron 등록 성공·spawn 미확인 → CRON_REGISTERED_PENDING_SPAWN
        #         (**non-terminal**, polling loop 아님 — 현 시점 evidence 로 미확정 기록).
        #         이후 reconcile_late_spawn 이 late spawn evidence 확인 시 success 로
        #         **append-only stateless 재평가**한다(새 fire 0).
        #     (3) 등록·spawn 성공이나 callback evidence 부재 → PARTIAL_FIRE_NO_CALLBACK_EVIDENCE.
        #     (4) 전부 충족 → REALFIRE_1SHOT_SUCCESS.
        spawned, callback_evidence, cron_registered, schedule_id = _read_fire_outcome(fire_raw)
        if not cron_registered:
            status = GATE_PARTIAL_FIRE_NO_SPAWN
            fire_reason = ("approved one-shot fire_fn 호출됨 — cron 등록 자체 실패(cron 미생성). "
                           "PARTIAL_FIRE_NO_SPAWN(terminal). 재발사 방지 marker 기록(자동 재시도 0).")
        elif not spawned:
            status = GATE_CRON_REGISTERED_PENDING_SPAWN
            fire_reason = ("approved one-shot fire_fn 호출됨 — cron 등록 성공이나 현 시점 spawn "
                           "미확인. CRON_REGISTERED_PENDING_SPAWN(**non-terminal**, polling loop "
                           "아님). late spawn evidence 확인 시 reconcile_late_spawn 이 append-only "
                           "stateless 재평가(새 fire 0). 재발사 방지 marker 기록(자동 재시도 0). "
                           f"schedule_id={schedule_id or '∅'}.")
        elif not callback_evidence:
            status = GATE_PARTIAL_FIRE_NO_CALLBACK_EVIDENCE
            fire_reason = ("approved one-shot fire_fn 호출됨 — spawn 성공이나 callback evidence "
                           "부재. PARTIAL_FIRE_NO_CALLBACK_EVIDENCE. 재발사 방지 marker 기록.")
        else:
            status = GATE_REALFIRE_1SHOT_SUCCESS
            fire_reason = ("approved one-shot real-fire 1회 성공 — cokacdir --cron 1회 생성 + "
                           "callback evidence 확인. REALFIRE_1SHOT_SUCCESS. 재발사 방지 marker 기록.")

    # ⑧ 재발사 방지 marker 기록(fire_fn 호출 후 모든 상태 — double-fire 차단). raw key 0.
    marker_ok = _write_owner_gate_marker(
        _ledger, task_id=target.task_id, head_sha=verified_head,
        status=status, ts=_clock().isoformat(),
    )
    reasons = [fire_reason]
    if not marker_ok:
        reasons.append("MARKER_WRITE_FAILED — 재발사 방지 marker 기록 실패(관측). dedupe "
                       "ledger write 불가. 후속 호출은 자동 재시도 0(승인 1회성).")

    fired_ok = (status == GATE_REALFIRE_1SHOT_SUCCESS)
    return OwnerProofGateResult(
        schema=re_gate.schema,
        status=status,
        task_id=target.task_id,
        result_json_path=re_gate.result_json_path,
        head_sha=verified_head,
        argv=redacted_argv,           # ★ persist 는 redacted(placeholder key) — raw key 0.
        fire_blocked=not fired_ok,    # 성공 시에만 fire_blocked=False(실 발사함).
        active=False,                 # ★ persistent ACTIVE 전환 0 — 항상 False.
        owner_key_proof_present=re_gate.owner_key_proof_present,
        schedule_owner_proof=re_gate.schedule_owner_proof,
        eligibility=dict(re_gate.eligibility),
        reasons=reasons,
        dedupe_marker_present=marker_ok,
    )


# ═══════════════════════════════════════════════════════════════════════════════
# task-2774+8 — D-2 fire wiring (실발사 가능하게 하는 안전 결선, guard-native)
# ═══════════════════════════════════════════════════════════════════════════════
# 배경(D-2-A audit 확정): attempt_owner_fire 의 approved one-shot 경로는 골격만 있고
#   production 실발사 fire_fn default 결선이 0이라(fire_fn 미주입 → FIRE_ERROR) 코드 변경
#   없이 real fire 가 불가했다. 또한 ONESHOT_TARGET 이 고정 하드코딩이라 자동 default 발사
#   벡터가 잠재했다. 본 wiring 은 다음을 닫는다:
#     (1) 실 cokacdir --cron --once 1회 subprocess fire_fn 을 **명시 factory**
#         (build_oneshot_cokacdir_fire_fn) 로 제공한다(production fire_fn 결선).
#     (2) 승인된 1-shot 전용 entrypoint(run_approved_oneshot_fire)에서만 그 fire_fn 과
#         승인 target 을 **명시 주입**하도록 결선한다(default 실발사 fire_fn·하드코딩 target
#         자동발사 0).
#
# ★ 본 PR 은 real fire 를 쏘지 않는다(결선만). production 어디에서도 본 entrypoint/factory 를
#   자동 호출하지 않으며(runner/pickup/driver 무결선·attempt_owner_fire production call site 0),
#   ACTIVE gate(callback_launch flag) 는 OFF, 테스트는 mock fire_fn·mock flag_reader 만
#   사용한다 → 실 cokacdir --cron 호출 0. 실 발사는 PR merge 후 별도 D-2 승인에서만.
# ★ default 실발사 fire_fn 금지(쐐기1): attempt_owner_fire·run_approved_oneshot_fire 어디에도
#   real fire_fn 을 default 로 두지 않는다. real fire_fn 은 호출자가
#   build_oneshot_cokacdir_fire_fn() 으로 **명시 생성**해 run_approved_oneshot_fire(fire_fn=...)
#   으로만 주입한다(예상치 못한 경로 fire 0).
# ★ raw key hygiene(쐐기8): 실 ANU key 의 로드/argv 구성/redaction 은 attempt_owner_fire 가
#   소유한다. 본 섹션의 fire_fn 은 attempt_owner_fire 가 건넨 real_argv 를 subprocess 로
#   전달만 하며, argv/stdout/예외를 절대 log/persist/return 하지 않는다(고정 라벨만).


# task-2774+9 — cron 등록 성공 판정 fail-closed 상태명(real fire 직전 코드).
#   registration response 의 id 가 실제 존재+비어있지 않을 때만 성공으로 기록한다. 미등록을
#   성공으로 기록하면 dedupe/duplicate fire/teardown/callback 판단이 흐려지므로 보수 차단.
FIRE_BLOCKED_NO_CRON_ID = "FIRE_BLOCKED_NO_CRON_ID"      # status=ok 이나 id 부재/빈값/None
FIRE_REGISTRATION_FAILED = "FIRE_REGISTRATION_FAILED"    # stdout 파싱 실패/비정상 응답


def _parse_cokacdir_cron_stdout(stdout: object) -> dict:
    """cokacdir --cron --once 의 JSON stdout → fire outcome dict(raw key 미포함).

    cokacdir 성공 출력 형태: {"status":"ok","id":"<schedule_id>", ...}. id 가 실제 발급된
    경우(비어있지 않은 문자열)에만 cron 등록 성공으로 본다. spawn/callback evidence 는 등록
    시점엔 미관측이므로 보수적 False — late spawn 은 reconcile_late_spawn 이 별도 evidence 로
    재평가한다(새 fire 0).
    ★ fail-closed(task-2774+9): status=="ok" 이라도 id 가 없거나 빈 문자열/None 이면
      registered=False 로 둔다(FIRE_BLOCKED_NO_CRON_ID). real fire 직전 미등록을 성공으로
      기록하지 않는다. 기존 id 있는 성공 경로는 그대로 유지.
    ★ stdout 전체를 reason/return 에 담지 않는다(raw key hygiene) — schedule_id label 만.
    """
    schedule_id = ""
    registered = False
    fire_status = ""
    try:
        import json  # noqa: PLC0415 — stdout JSON 파싱(발사/cron 0)
        text = stdout if stdout is not None else ""
        if isinstance(text, bytes):
            text = text.decode("utf-8", "replace")
        last = ""
        for line in str(text).splitlines():
            if line.strip():
                last = line.strip()
        if last:
            data = json.loads(last)
            if isinstance(data, dict) and data.get("status") == "ok":
                # fail-closed: id 가 실제 존재하고 비어있지 않을 때만 registered=True.
                raw_id = data.get("id")
                if raw_id is not None:
                    schedule_id = str(raw_id).strip()
                    if schedule_id:
                        registered = True
                if not registered:
                    # id 부재/빈문자열/None → 미등록(fire 성공 기록 금지).
                    fire_status = FIRE_BLOCKED_NO_CRON_ID
            else:
                # status!=ok 등 비정상 응답 → fail-closed.
                fire_status = FIRE_REGISTRATION_FAILED
        else:
            fire_status = FIRE_REGISTRATION_FAILED
    except Exception:  # noqa: BLE001 — 파싱 실패는 보수적 미등록(고정 라벨·stdout 미노출)
        schedule_id = ""
        registered = False
        fire_status = FIRE_REGISTRATION_FAILED
    outcome = {
        "spawned": False,            # 등록 시점 spawn 미관측(late spawn → reconcile)
        "cron_registered": registered,
        "callback_evidence": False,  # 등록 시점 callback 미관측
        "schedule_id": schedule_id,
    }
    if not registered and fire_status:
        # 고정 라벨만(raw key/stdout 미포함). _read_fire_outcome 는 미사용 키 무시.
        outcome["error"] = fire_status
    return outcome


def build_oneshot_cokacdir_fire_fn(
    *,
    timeout_sec: int = 60,
    runner: Optional[Callable[[Sequence[str]], object]] = None,
) -> Callable[[Sequence[str]], dict]:
    """승인된 1-shot real-fire 용 production fire_fn 을 생성해 반환한다(결선만).

    반환된 fire_fn 은 attempt_owner_fire 가 구성한 real_argv(실 ANU key 포함·cokacdir
    --cron --once 1회 형태)를 받아 **정확히 1회** subprocess 를 실행하고, _read_fire_outcome
    가 해석할 dict(spawned/cron_registered/callback_evidence/schedule_id)를 반환한다.

    ★ 본 PR 은 본 factory 의 반환 fire_fn 을 production 에서 호출하지 않는다(결선만). 실 발사는
      PR merge 후 별도 D-2 승인에서 run_approved_oneshot_fire(fire_fn=build_oneshot_cokacdir_fire_fn())
      형태로만 일어난다.
    ★ raw key hygiene(쐐기8): argv 는 실 ANU key 를 담는다 — argv/stdout/예외를 절대
      log/persist/return 하지 않는다. 실패·예외 reason 은 고정 라벨만 반환한다(raw key 0).
    ★ recurring 금지(쐐기6): argv 에 `--cron` 과 `--once` 가 모두 있어야만 실행한다. `--once`
      부재 시 recurring 위험으로 보고 실행하지 않고 fail-closed(cron_registered=False).
    ★ runner 인자: 테스트가 실 subprocess 대신 mock 을 주입하기 위한 격리 hook 이다(실
      cokacdir 호출 0 보장). production 은 runner=None(실 subprocess).
    """

    def _fire(argv: Sequence[str]) -> dict:
        argv_list = [str(a) for a in (argv or [])]
        # recurring 금지 방어(쐐기6): --cron + --once 동시 존재만 허용(없으면 실행 0·fail-closed).
        if "--cron" not in argv_list or "--once" not in argv_list:
            return {
                "spawned": False,
                "cron_registered": False,
                "callback_evidence": False,
                "schedule_id": "",
                "error": "ONESHOT_FIRE_ARGV_INVALID",  # 고정 라벨(raw key 0·argv 미포함)
            }
        try:
            if runner is not None:
                raw = runner(argv_list)
            else:
                import subprocess  # noqa: PLC0415 — approved 1-shot real-fire subprocess
                # text=True 제거(task-2774+10): stdout 을 raw bytes 로 캡처해 디코딩을
                # _parse_cokacdir_cron_stdout(bytes + errors="replace" fail-closed)에 위임한다.
                # 시스템 기본 인코딩 의존 UnicodeDecodeError 로 fire_fn 이 깨지지 않게(플랫폼 독립).
                proc = subprocess.run(
                    argv_list, capture_output=True,
                    timeout=timeout_sec, check=False,
                )
                raw = _parse_cokacdir_cron_stdout(proc.stdout)
        except Exception:  # noqa: BLE001 — 예외는 고정 라벨 fail-closed(raw key/argv 미노출)
            return {
                "spawned": False,
                "cron_registered": False,
                "callback_evidence": False,
                "schedule_id": "",
                "error": "ONESHOT_FIRE_SUBPROCESS_ERROR",  # 고정 라벨
            }
        if not isinstance(raw, dict):
            # runner/parser 가 dict 가 아니면 보수적 fail-closed(고정 라벨).
            return {
                "spawned": False, "cron_registered": False,
                "callback_evidence": False, "schedule_id": "",
                "error": "ONESHOT_FIRE_OUTCOME_INVALID",
            }
        return raw

    return _fire


def _oneshot_wrapper_block(gate_result: OwnerProofGateResult, *, status: str,
                          reason: str) -> OwnerProofGateResult:
    """run_approved_oneshot_fire 의 wrapper-gate 차단 결과(no-fire·raw key 0·argv None)."""
    return OwnerProofGateResult(
        schema=gate_result.schema,
        status=status,
        task_id=gate_result.task_id,
        result_json_path=gate_result.result_json_path,
        head_sha=gate_result.head_sha,
        argv=None,                    # ★ 차단 — argv 미구성(raw key 0).
        fire_blocked=True,
        active=False,                 # ★ persistent ACTIVE 전환 0(별개 1-shot gate).
        owner_key_proof_present=gate_result.owner_key_proof_present,
        schedule_owner_proof=gate_result.schedule_owner_proof,
        eligibility=dict(gate_result.eligibility or {}),
        reasons=list(gate_result.reasons or []) + [reason],
        dedupe_marker_present=gate_result.dedupe_marker_present,
    )


def run_approved_oneshot_fire(
    gate_result: OwnerProofGateResult,
    *,
    flag_reader: Optional[Callable[[], bool]] = None,
    fire_fn: Optional[Callable[[Sequence[str]], object]] = None,
    target: Optional[OneShotTarget] = None,
    sealed_key_loader: Optional[Callable[[], Optional[str]]] = None,
    executor_key: str = "",
    anu_keys: Optional[Sequence[str]] = None,
    head_probe: Optional[Callable[[], str]] = None,
    dedupe_ledger_path: Optional[str] = None,
    at_spec: str = ONESHOT_AT_SPEC,
    chat_id: str = ONESHOT_PILOT_CHAT_ID,
    clock: Optional[Callable[[], datetime]] = None,
) -> OwnerProofGateResult:
    """승인된 1-shot real-fire 전용 entrypoint (결선만 — 본 PR real fire 0).

    이 함수는 D-2 real fire 1-shot 의 **유일한** 진입점이다. 일반 runner/pickup/driver
    경로는 본 함수를 자동으로 타지 않는다(production 무결선). 아래 순서로 fail-closed 한다:

      1. ACTIVE gate(callback_launch flag) = **함수인자 flag_reader** 로만 판정한다(전역
         flag 파일 생성/읽기 0·p0b_driver_enabled activation 과 별개·혼동 금지). flag_reader
         부재·False·예외 → NOOP_DISABLED(fire_fn 미호출·실 발사 0).
      2. 실발사 fire_fn 명시 주입 부재(fire_fn None) → FIRE_BLOCKED_NO_FIRE_FN(fail-closed).
         ★ default 실발사 fire_fn 0 — real fire_fn 은 호출자가 build_oneshot_cokacdir_fire_fn()
         으로 명시 생성해 fire_fn= 로 주입해야 한다(예상치 못한 경로 fire 차단).
      3. 승인 target 명시 주입 부재(target None) → FIRE_BLOCKED_NO_APPROVAL(승인 대상 1건
         미지정). ★ ONESHOT_TARGET 하드코딩을 default 로 자동발사하지 않는다 — 승인된 1-shot
         target 은 항상 호출자가 명시 주입한다(하드코딩 자동발사 차단).
      4. 위 통과 시에만 attempt_owner_fire(approved_one_shot=True, fire_fn=<주입>, target=<주입>)
         에 위임한다. target 3조건 정확 일치(task_id + abs path + head_sha==canonical HEAD
         40자)·gate READY 재확인·dedupe 1회 소진은 모두 attempt_owner_fire 가 수행한다(무수정
         재사용). 따라서 owner-proof gate 미통과·target 미일치·dedupe 시 fire_fn 미호출이
         그대로 보존된다.

    ★ raw key 0: 본 함수는 key 를 직접 다루지 않는다. 실 key 로드/argv 구성/redaction 은
      attempt_owner_fire 가 소유하며, 반환 result 의 argv 는 redacted(placeholder)다.
    """
    # 1. ACTIVE gate(callback_launch flag) — 함수인자 flag_reader 로만 판정(전역 파일 0).
    #    flag_reader 가 None 아닌 non-callable(예: True/object())이면 silent NOOP 금지 →
    #    명시 fail-closed(INVALID_FLAG_READER). silent misconfiguration 원인 추적 가능.
    #    ★ object 로 받아 검사 — 시그니처는 Optional[Callable] 이나 호출자가 잘못된 객체를
    #      줄 수 있어 런타임 방어한다(정적 narrow 우회).
    _reader_obj = cast(object, flag_reader)
    if _reader_obj is not None and not callable(_reader_obj):
        return _oneshot_wrapper_block(
            gate_result, status=GATE_INVALID_FLAG_READER,
            reason=("approved 1-shot ACTIVE gate flag_reader 가 None 아닌 non-callable "
                    f"(type={type(_reader_obj).__name__}) — INVALID_FLAG_READER(fail-closed). "
                    "silent NOOP_DISABLED 로 삼키지 않음(원인 추적 가능). fire_fn 미호출·실 발사 0."),
        )
    _active = False
    if callable(flag_reader):
        try:
            _active = bool(flag_reader())
        except Exception:  # noqa: BLE001 — flag_reader 예외는 보수적 OFF(fail-closed)
            _active = False
    if not _active:
        return _oneshot_wrapper_block(
            gate_result, status=GATE_NOOP_DISABLED,
            reason=("approved 1-shot ACTIVE gate(callback_launch flag, 함수인자 flag_reader) "
                    "OFF/부재 — NOOP_DISABLED. fire_fn 미주입·미호출·실 발사 0(전역 flag 파일 "
                    "생성/읽기 0·p0b_driver_enabled 별개)."),
        )
    # 2. 실발사 fire_fn 명시 주입 부재 → FIRE_BLOCKED_NO_FIRE_FN(fail-closed·default 0).
    if fire_fn is None:
        return _oneshot_wrapper_block(
            gate_result, status=GATE_FIRE_BLOCKED_NO_FIRE_FN,
            reason=("approved 1-shot ACTIVE 이나 실발사 fire_fn 명시 주입 부재 — "
                    "FIRE_BLOCKED_NO_FIRE_FN(fail-closed). default 실발사 fire_fn 0 → fire_fn "
                    "미호출·실 발사 0. real fire_fn 은 build_oneshot_cokacdir_fire_fn() 으로 "
                    "명시 생성·주입해야 한다."),
        )
    # 3. 승인 target 명시 주입 부재 → FIRE_BLOCKED_NO_APPROVAL(하드코딩 자동발사 차단).
    if target is None:
        return _fire_blocked_no_approval(
            gate_result, active=False,
            extra_reasons=[
                "approved 1-shot 승인 target 명시 주입 부재(target None) — FIRE_BLOCKED_NO_APPROVAL. "
                "ONESHOT_TARGET 하드코딩 default 자동발사 차단(승인 대상 1건은 호출자 명시 주입). "
                "actual cokacdir --cron 호출 0.",
            ])
    # 4. 모든 wrapper-gate 통과 → attempt_owner_fire 위임(target 3조건·gate READY·dedupe 1회 소진).
    return attempt_owner_fire(
        gate_result,
        approved_one_shot=True,
        fire_fn=fire_fn,
        sealed_key_loader=sealed_key_loader,
        executor_key=executor_key,
        anu_keys=anu_keys,
        head_probe=head_probe,
        dedupe_ledger_path=dedupe_ledger_path,
        target=target,
        at_spec=at_spec,
        chat_id=chat_id,
        clock=clock,
    )


# ── task-2743: late spawn stateless reconciliation (no-new-fire·append-only) ──────
@dataclass(frozen=True)
class LateSpawnEvidence:
    """stateless reconcile 입력 evidence — 부수효과 없이 idempotent 판정에 쓰인다.

    reconcile_late_spawn 은 아래 evidence 만으로 verdict 를 재도출한다(외부 상태 조회
    0). 누가 언제 호출하든 같은 evidence 면 같은 verdict 다.

    필드(회장 §4 최소 항목):
      schedule_id          관측된 schedule id (cron-history/cron-list 에서 확인)
      expected_schedule_id 등록 시 발급된 schedule id — schedule_id 와 정확 일치해야
                           한다(wrong schedule_id false-positive 차단)
      cron_registered      cron 등록 성공 여부(등록 실패 false-positive 차단)
      cron_history_present cron-history 실행 기록 존재(spawn 발생의 1차 증거)
      workspace_dir_present spawned owner session 의 workspace dir 존재(spawn 2차 증거)
      cron_list_consumed   once schedule 이 cron-list 에서 소비(미존재)됨 — 1회성 소비 확인
      spawned_head_sha     spawned owner session 의 HEAD **full 40자**
      fixture_head_sha     fixture/result 의 head_sha **full 40자**
      owner_proof          schedule owner proof label — "ANU" 이어야 한다
      executor_self_key    dev/self key 사용 evidence(true → success reconcile 금지)
    """

    schedule_id: str
    expected_schedule_id: str
    cron_registered: bool
    cron_history_present: bool
    workspace_dir_present: bool
    cron_list_consumed: bool
    spawned_head_sha: str
    fixture_head_sha: str
    owner_proof: str = ""
    executor_self_key: bool = False


# reconcile 가 stateless 재평가의 대상으로 받아들이는 입력 status(pending 또는 보존된
# 기존 PARTIAL). 그 외 status(이미 성공/DEDUPED/FIRE_ERROR 등)는 reconcile 가 success 를
# 제조하지 않고 입력을 그대로 반환한다(임의 상태에서 success 날조 0).
GATE_RECONCILABLE_INPUT_STATUSES = (
    GATE_CRON_REGISTERED_PENDING_SPAWN,
    GATE_PARTIAL_FIRE_NO_SPAWN,
)


def _reconcile_marker_present(ledger_path: str, task_id: str, head_sha: str,
                              event: str) -> bool:
    """(task_id, head_sha) 에 대해 지정 reconcile event marker 가 존재하는지(읽기 전용).

    idempotency/dedupe 용 — 이미 confirm marker 가 있으면 재append 0(중복 기록 방지)."""
    if not ledger_path or not os.path.isfile(ledger_path):
        return False
    try:
        with open(ledger_path, "r", encoding="utf-8") as lfh:
            for line in lfh:
                line = line.strip()
                if not line:
                    continue
                try:
                    entry = json.loads(line)
                except (ValueError, TypeError):
                    continue
                if (isinstance(entry, dict)
                        and entry.get("event") == event
                        and entry.get("task_id") == task_id
                        and entry.get("head_sha") == head_sha):
                    return True
    except OSError:
        return False
    return False


def _write_reconcile_marker(ledger_path: str, *, event: str, task_id: str,
                            head_sha: str, status: str, schedule_id: str,
                            ts: str) -> bool:
    """reconcile confirm/superseded marker append (append-only·raw key 0).

    기존 줄(PARTIAL/PENDING marker 포함)을 **삭제/수정하지 않는다** — 새 줄만 append.
    실패 시 False(관측). 새 cron fire 0(이 함수는 ledger 기록만)."""
    entry = json.dumps({
        "schema": "dispatch.anu_result_pickup_runner.owner_gate.reconcile.v1",
        "event": event,
        "task_id": task_id,
        "head_sha": head_sha,
        "status": status,
        "schedule_id": schedule_id,
        "ts": ts,
    }, ensure_ascii=False)
    try:
        os.makedirs(os.path.dirname(ledger_path) or ".", exist_ok=True)
        with open(ledger_path, "a", encoding="utf-8") as lfh:
            lfh.write(entry + "\n")
            lfh.flush()
            os.fsync(lfh.fileno())
        return True
    except OSError:
        return False


def _evaluate_late_spawn_success(evidence: LateSpawnEvidence,
                                 anu_proof_label: str) -> tuple:
    """late spawn success 8조건을 stateless 평가 → (ok: bool, fail_reasons: List[str]).

    부수효과 0 — evidence 만으로 판정한다(idempotent). false-positive 6종을 명시적으로
    차단한다(stale head / prefix·short SHA only / wrong schedule_id / dev-self key /
    등록 실패 / — dup marker 는 호출부 reconcile 에서 dedupe 처리).
    """
    fails: List[str] = []
    sh = str(evidence.spawned_head_sha or "")
    fh = str(evidence.fixture_head_sha or "")

    # (조건1) cron 등록 성공 — 등록 실패는 success reconcile 금지(false-positive #5).
    if not evidence.cron_registered:
        fails.append("cron 등록 실패(cron_registered=false) — 등록 실패는 success reconcile "
                     "금지. PENDING 유지(false-positive #5 차단).")

    # (조건2) schedule owner proof == ANU + dev/self key 아님(false-positive #4).
    if str(evidence.owner_proof or "") != anu_proof_label:
        fails.append(f"owner proof != {anu_proof_label}(got={evidence.owner_proof or '∅'}) — "
                     "owner proof ANU 아님. success reconcile 금지.")
    if evidence.executor_self_key:
        fails.append("dev/self key evidence(executor_self_key=true) — ANU owner proof 아님. "
                     "success reconcile 금지(false-positive #4 차단).")

    # (조건3) schedule_id 정확 일치 — wrong/empty schedule_id 차단(false-positive #3).
    if not evidence.schedule_id or not evidence.expected_schedule_id:
        fails.append("schedule_id/expected_schedule_id 부재 — 소비된 once schedule 식별 불가. "
                     "success reconcile 금지.")
    elif evidence.schedule_id != evidence.expected_schedule_id:
        fails.append(f"wrong schedule_id(got={evidence.schedule_id} != "
                     f"expected={evidence.expected_schedule_id}) — 다른 schedule evidence. "
                     "success reconcile 금지(false-positive #3 차단).")

    # (조건3-cont) cron 소비 확인(once schedule 미존재 = 소비됨).
    if not evidence.cron_list_consumed:
        fails.append("cron-list 에서 once schedule 미소비 — cron 소비 미확인. success reconcile 금지.")

    # (조건4) ANU owner session spawn 확인 — cron-history 실행 기록 + workspace dir 존재.
    if not (evidence.cron_history_present and evidence.workspace_dir_present):
        fails.append("spawn evidence 부족 — cron-history 실행 기록/workspace dir 미확인. "
                     "spawn 미확정 → PENDING 유지(success reconcile 금지).")

    # (조건5) spawned HEAD == fixture head_sha **full 40자 정확 일치**(== only, prefix 완화 0).
    #   stale head / prefix·short SHA only 차단(false-positive #1·#2).
    if not (len(sh) == 40 and len(fh) == 40 and sh == fh):
        fails.append("spawned HEAD != fixture head_sha full 40자 정확 일치 아님("
                     f"spawned len={len(sh)}, fixture len={len(fh)}, equal={sh == fh}) — "
                     "stale head/prefix·short SHA 차단. == only·prefix 완화 0. success reconcile "
                     "금지(false-positive #1·#2 차단).")

    return (not fails), fails


def reconcile_late_spawn(
    pending: OwnerProofGateResult,
    evidence: LateSpawnEvidence,
    *,
    dedupe_ledger_path: Optional[str] = None,
    clock: Optional[Callable[[], datetime]] = None,
    anu_proof_label: str = GATE_OWNER_PROOF_ANU,
) -> OwnerProofGateResult:
    """late spawn stateless reconciliation — pending → 재평가(부수효과 최소·idempotent).

    ★ 회장 doctrine 코드 레벨 보장:
      · **stateless 재평가**: evidence(LateSpawnEvidence)만으로 verdict 를 재도출한다.
        누가 언제 호출하든 같은 evidence 면 같은 verdict(idempotent).
      · **새 cron fire 절대 0**: fire_fn 호출/argv 구성/cokacdir --cron 호출 0. read-only
        verdict + append-only ledger record(confirm/superseded)만 남긴다.
      · **append-only·삭제 0**: 기존 PARTIAL/PENDING marker 줄을 삭제/수정하지 않는다.
        기존 PARTIAL 은 superseded_by_late_spawn_confirmation marker 를 **별도 줄로**
        append 해 보존한다(삭제 0).
      · **자동 재시도와 분리**: reconcile 은 spawn 을 기다리는 polling loop 가 아니다.
        현 evidence 로 미충족이면 PENDING(non-terminal) 그대로 반환한다(재발사 0).
      · **dedupe 유지**: 이미 confirm marker 가 있으면 재append 0(중복 기록 방지)·재발사 0.

    success 8조건(전부 충족 시에만 LATE_SPAWN_CONFIRMED/REALFIRE_1SHOT_SUCCESS_CONFIRMED):
      등록 성공 + owner proof ANU + cron 소비 + spawn 확인 + spawned HEAD==fixture
      head_sha full 40자 정확 일치(prefix 완화 0) + evidence append + 추가 발사 0 +
      ACTIVE=false. false-positive 6종(stale/prefix-short/wrong sid/dev-self key/등록실패/
      dup)은 success reconcile 금지.
    """
    _clock = clock or (lambda: datetime.now(timezone.utc))
    _ledger = dedupe_ledger_path or DEFAULT_OWNER_GATE_LEDGER_PATH
    task_id = str(pending.task_id or "")
    head_sha = str(pending.head_sha or "")

    def _carry(status: str, *, fire_blocked: bool, dedupe_marker_present: bool,
               extra_reasons: List[str]) -> OwnerProofGateResult:
        return OwnerProofGateResult(
            schema=pending.schema,
            status=status,
            task_id=pending.task_id,
            result_json_path=pending.result_json_path,
            head_sha=pending.head_sha,
            argv=list(pending.argv) if pending.argv is not None else None,
            fire_blocked=fire_blocked,
            active=False,                 # ★ ACTIVE 전환 0 — 항상 False.
            owner_key_proof_present=pending.owner_key_proof_present,
            schedule_owner_proof=pending.schedule_owner_proof,
            eligibility=dict(pending.eligibility),
            reasons=list(pending.reasons) + list(extra_reasons),
            dedupe_marker_present=dedupe_marker_present,
        )

    # ── 입력 status 검증: pending/보존된 PARTIAL 만 재평가 대상. 그 외는 success 날조 0. ──
    if pending.status not in GATE_RECONCILABLE_INPUT_STATUSES:
        return _carry(
            pending.status, fire_blocked=pending.fire_blocked,
            dedupe_marker_present=pending.dedupe_marker_present,
            extra_reasons=[
                f"reconcile no-op — 입력 status={pending.status} 는 재평가 대상 아님("
                f"{GATE_RECONCILABLE_INPUT_STATUSES} 만 reconcile). success 날조 0·새 fire 0.",
            ])

    # ── idempotency/dedupe: 이미 confirm marker 존재 → 재append 0·재발사 0. ──
    if _reconcile_marker_present(_ledger, task_id, head_sha, GATE_LEDGER_EVENT_RECONCILED):
        return _carry(
            GATE_REALFIRE_1SHOT_SUCCESS_CONFIRMED, fire_blocked=False,
            dedupe_marker_present=True,
            extra_reasons=[
                "reconcile dedupe — (task_id, head_sha) confirm marker 이미 존재. 재append 0·"
                "새 fire 0(idempotent). LATE_SPAWN_CONFIRMED 유지.",
            ])

    # ── stateless success 8조건 평가(부수효과 0). ──
    ok, fail_reasons = _evaluate_late_spawn_success(evidence, anu_proof_label)
    if not ok:
        # 미충족 → PENDING(non-terminal) 유지. ledger write 0(read-only verdict)·새 fire 0.
        return _carry(
            GATE_CRON_REGISTERED_PENDING_SPAWN, fire_blocked=True,
            dedupe_marker_present=pending.dedupe_marker_present,
            extra_reasons=[
                "late spawn reconcile 미충족 — 현 evidence 로 success 미확정. PENDING(non-"
                "terminal) 유지·새 fire 0·자동 재시도 0.",
            ] + fail_reasons)

    # ── 전부 충족 → append-only reconcile. 기존 PARTIAL 보존(superseded marker)·confirm marker. ──
    ts = _clock().isoformat()
    reasons: List[str] = []

    # 기존 PARTIAL marker 가 있으면 삭제 0 — superseded marker 를 **별도 줄로** append.
    partial_present = _owner_gate_marker_present(_ledger, task_id, head_sha)
    superseded_ok = True
    if partial_present and not _reconcile_marker_present(
            _ledger, task_id, head_sha, GATE_LEDGER_EVENT_SUPERSEDED):
        superseded_ok = _write_reconcile_marker(
            _ledger, event=GATE_LEDGER_EVENT_SUPERSEDED, task_id=task_id,
            head_sha=head_sha, status=GATE_SUPERSEDED_BY_LATE_SPAWN,
            schedule_id=evidence.schedule_id, ts=ts,
        )
        reasons.append("기존 PARTIAL/PENDING evidence 보존 — superseded_by_late_spawn_"
                       "confirmation marker 를 별도 줄로 append(삭제 0).")

    # confirm marker append(late spawn 확정). event status=LATE_SPAWN_CONFIRMED.
    confirm_ok = _write_reconcile_marker(
        _ledger, event=GATE_LEDGER_EVENT_RECONCILED, task_id=task_id,
        head_sha=head_sha, status=GATE_LATE_SPAWN_CONFIRMED,
        schedule_id=evidence.schedule_id, ts=ts,
    )
    reasons.insert(0,
        "late spawn evidence 전부 충족(등록 성공·owner proof ANU·cron 소비·spawn 확인·"
        "spawned HEAD==fixture head_sha full 40자 정확 일치) → REALFIRE_1SHOT_SUCCESS_"
        "CONFIRMED(≡ LATE_SPAWN_CONFIRMED). stateless 재평가·새 fire 0·append-only.")
    if not (confirm_ok and superseded_ok):
        reasons.append("RECONCILE_MARKER_WRITE_FAILED — append-only marker 기록 일부 실패(관측). "
                       "verdict 는 evidence 기반 idempotent(다음 호출도 같은 evidence면 동일).")

    return _carry(
        GATE_REALFIRE_1SHOT_SUCCESS_CONFIRMED, fire_blocked=False,
        dedupe_marker_present=confirm_ok, extra_reasons=reasons)


# ── CLI ───────────────────────────────────────────────────────────────────────
def main(argv: Optional[List[str]] = None) -> int:
    import argparse

    ap = argparse.ArgumentParser(prog="dispatch.anu_result_pickup_runner")
    sub = ap.add_subparsers(dest="cmd", required=True)

    pu = sub.add_parser("pickup")
    pu.add_argument("--result-json-path", required=True)
    pu.add_argument("--executor-key", default="")

    # task-2741: OS-level owner-proof gate (no-fire). argv 구성만 출력, 발사 0.
    og = sub.add_parser("owner-gate")
    og.add_argument("--result-json-path", required=True)
    og.add_argument("--executor-key", default="")

    a = ap.parse_args(argv)

    if a.cmd == "pickup":
        # task-2741-r1: CLI pickup 도 owner-gate subcommand 와 동일 gate 정책 공유 —
        #   fire 전 owner_proof_pickup_gate 강제(우회 0). 미PASS → OWNER_GATE_BLOCKED.
        res = pickup_once(
            a.result_json_path,
            executor_key=a.executor_key,
            owner_gate_fn=build_owner_gate_fn(executor_key=a.executor_key),
        )
        print(json.dumps(res.to_json(), ensure_ascii=False))
        return 0 if res.ok else 2

    if a.cmd == "owner-gate":
        gres = owner_proof_pickup_gate(
            a.result_json_path,
            executor_key=a.executor_key,
        )
        print(json.dumps(gres.to_json(), ensure_ascii=False))
        return 0 if gres.ready else 2

    return 2


__all__ = [
    "PICKUP_SCHEMA",
    "PICKUP_WAKE_BUILT",
    "PICKUP_CLOSEOUT_DONE",
    "PICKUP_NO_RESULT_JSON",
    "PICKUP_FAIL",
    "PICKUP_SKIP_TERMINAL",
    "PICKUP_SKIP_DEDUPE",
    "PICKUP_QUARANTINE",
    "PICKUP_PENDING",
    "PICKUP_REJECT",
    "PICKUP_SEALED_KEY_MISSING",
    "PICKUP_LEDGER_WRITE_FAILED",
    "PICKUP_COLLECTOR_WRITE_FAILED",
    "PICKUP_OWNER_GATE_BLOCKED",
    "PICKUP_SPAWN_BLOCKED",
    "LEDGER_EVENT_WAKE",
    "LEDGER_EVENT_CLOSEOUT",
    "LEDGER_DEDUPE_EVENTS",
    "DEFAULT_LEDGER_PATH",
    "ENV_ANU_KEY",
    "TERMINAL_INTENT_DEFAULT",
    "read_terminal_intent",
    "PickupResult",
    "pickup_once",
    # task-2741 — OS-level owner-proof gate (no-fire)
    "GATE_OWNER_PICKUP_READY_NO_FIRE",
    "GATE_DEDUPED",
    "GATE_RESULT_NOT_PICKUP_ELIGIBLE",
    "GATE_FAIL_CLOSED_NON_ANU_KEY",
    "GATE_FIRE_BLOCKED_NO_APPROVAL",
    "GATE_TERMINAL_STATUSES",
    # task-2742 — approved one-shot real-fire terminal statuses (gate 5 외 4)
    "GATE_FIRE_ERROR",
    "GATE_PARTIAL_FIRE_NO_SPAWN",
    "GATE_PARTIAL_FIRE_NO_CALLBACK_EVIDENCE",
    "GATE_REALFIRE_1SHOT_SUCCESS",
    "GATE_ONESHOT_FIRE_STATUSES",
    "GATE_ALL_TERMINAL_STATUSES",
    # task-2774+8 — D-2 fire wiring: approved 1-shot wrapper terminal status + entrypoint
    "GATE_NOOP_DISABLED",
    "GATE_FIRE_BLOCKED_NO_FIRE_FN",
    "GATE_INVALID_FLAG_READER",
    "GATE_ONESHOT_WRAPPER_STATUSES",
    "build_oneshot_cokacdir_fire_fn",
    "run_approved_oneshot_fire",
    # task-2743 — late spawn reconciliation verdict (non-terminal pending + confirm)
    "GATE_CRON_REGISTERED_PENDING_SPAWN",
    "GATE_LATE_SPAWN_CONFIRMED",
    "GATE_REALFIRE_1SHOT_SUCCESS_CONFIRMED",
    "GATE_SUPERSEDED_BY_LATE_SPAWN",
    "GATE_RECONCILE_SUCCESS_STATUSES",
    "GATE_ALL_VERDICT_TERMINAL_STATUSES",
    "GATE_RECONCILABLE_INPUT_STATUSES",
    "GATE_LEDGER_EVENT_RECONCILED",
    "GATE_LEDGER_EVENT_SUPERSEDED",
    "LateSpawnEvidence",
    "reconcile_late_spawn",
    "GATE_SCHEMA",
    "GATE_OWNER_PROOF_ANU",
    "DEFAULT_OWNER_GATE_LEDGER_PATH",
    "GATE_LEDGER_EVENT",
    "OWNER_PICKUP_KEY_PLACEHOLDER",
    "OWNER_PICKUP_AT_PLACEHOLDER",
    # task-2742 — approved one-shot 고정 대상 + argv spec
    #   (r3: ONESHOT_TARGET_HEAD_SHA 하드코드 상수 제거 — 런타임 canonical HEAD 동적 판정)
    "ONESHOT_TARGET_TASK_ID",
    "ONESHOT_TARGET_REL_PATH",
    "ONESHOT_TARGET_ABS_PATH",
    "ONESHOT_AT_SPEC",
    "ONESHOT_FIRE_TAG",
    "ONESHOT_TARGET",
    "OneShotTarget",
    "OwnerProofGateResult",
    "owner_proof_pickup_gate",
    "build_owner_gate_fn",
    "attempt_owner_fire",
    "main",
]

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