"""anu_v2.owner_trigger_pat — Fine-grained OWNER PAT trigger-only 모듈 (task-2553).

회장 §명시 (2026-05-11 dec_1) "Fine-grained OWNER PAT trigger-only doctrine 예외
조건부 승인" 박제. OWNER PAT 일반 사용 허용 X — 오직 GitHub Gemini App 을 깨우기 위한
`/gemini review` 댓글 작성에만 한정한다.

설계 원칙 (회장 §명시 12 필수 + 15 금지):
  - one-way isolation: anu_v2/* 만 import. utils/dispatch/scripts/dashboard 의존성 0.
  - 외부 부수효과는 모두 주입 가능한 callable (gh_runner, audit_writer,
    decision_writer, clock) — 테스트 시 mock 으로 대체.
  - allowed gh args 정확히 `["api", "-X", "POST", "/repos/{owner}/{repo}/issues/
    {pr}/comments", "-f", "body=/gemini review"]` 만. 다른 endpoint / body 는
    정적 차단 (RuntimeError).
  - token raw 0: audit/log/error message 어디에도 토큰 값 노출 X.
    audit 에는 `token_present: bool` + `token_hash: sha256[:12]` 만.
  - default GH_TOKEN fallback 0: 토큰 누락 시 즉시 REJECT (silent skip 금지).
  - dedupe by (pr_number, head_sha) — update-branch 후 새 head 는 새 trigger 허용.

본 모듈 외 OWNER PAT 접근 시도 0 — `OWNER_GEMINI_TRIGGER_PAT` env 이름은 본 모듈
상수에서만 정의된다.
"""

from __future__ import annotations

import hashlib
import json
import os
import re
import subprocess
from dataclasses import asdict, dataclass
from datetime import datetime, timezone
from pathlib import Path
from typing import Any, Callable, Mapping, Sequence


# ─── 상수 (회장 §명시 박제) ──────────────────────────────────────────────────
#: OWNER fine-grained PAT 의 secret env 이름. BOT_GITHUB_TOKEN 과 **반드시 분리**.
OWNER_PAT_ENV_NAME: str = "OWNER_GEMINI_TRIGGER_PAT"

#: 허용되는 단 1 개의 comment body (strict equality). 다른 body 입력 시 fail-fast.
ALLOWED_COMMENT_BODY: str = "/gemini review"

#: Decision schema 상수.
DECISION_PASS: str = "PASS"
DECISION_REJECT: str = "REJECT"

#: Gemini evidence state 값 — current head 에 대해 evidence 미도착 상태.
EVIDENCE_MISSING_FOR_CURRENT_HEAD: str = "missing_for_current_head"

#: TriggerOutcome.outcome 상수.
OUTCOME_OK: str = "ok"
OUTCOME_REJECTED: str = "rejected"
OUTCOME_FAILED: str = "failed"
#: gh runner 호출 직전에 박제되는 선점 마커 (atomic dedupe). 회장 §명시 — 같은
#: dedupe_key 가 "pending" 또는 "ok" 상태로 박제되어 있으면 후속 trigger 는 차단.
OUTCOME_PENDING: str = "pending"

#: 보안 경계 위반 시 raise 되는 에러 메시지 (정적 차단).
ERR_BODY_NOT_ALLOWED: str = "BODY_NOT_ALLOWED"
ERR_ENDPOINT_NOT_ALLOWED: str = "ENDPOINT_NOT_ALLOWED"

#: audit actor 식별자 (어떤 모듈이 trigger 했는지 박제).
ACTOR_OWNER_TRIGGER_PAT: str = "anu_v2.owner_trigger_pat"

#: redaction 치환 문자열. 토큰 substring 노출 0 보장.
REDACTED_PLACEHOLDER: str = "***REDACTED***"


# ─── 외부 부수효과 콜백 시그니처 ─────────────────────────────────────────────
GhRunner = Callable[[Sequence[str], Mapping[str, str]], subprocess.CompletedProcess]
AuditWriter = Callable[[Mapping[str, Any]], None]
DecisionWriter = Callable[[Mapping[str, Any], Path], None]
Clock = Callable[[], str]


# ─── Data types ──────────────────────────────────────────────────────────────
@dataclass(frozen=True)
class OwnerTriggerDecision:
    """`owner_trigger_decision.json` schema (frozen, 박제 대상).

    회장 §명시: trigger 전 PASS/REJECT 판정을 본 dataclass 로 박제하고 JSON 으로 atomic
    write. 토큰 값은 본 record 어디에도 포함되지 않는다.
    """

    pr_number: int
    head_sha: str
    decision: str               # PASS / REJECT
    reason: str
    gemini_evidence_state: str  # 예: missing_for_current_head / present_for_current_head
    queue_position: int          # 0 == queue-head
    dedupe_key: str              # f"{pr_number}#{head_sha}"
    ts: str                      # iso8601


@dataclass(frozen=True)
class TriggerOutcome:
    """trigger_gemini_review() 반환 — 호출부가 머지 파이프라인을 진행할지 결정하는 contract."""

    outcome: str          # ok / rejected / failed
    reason: str
    decision_path: str    # decision JSON 박제 위치 (REJECT 도 박제됨)


# ─── Helpers ────────────────────────────────────────────────────────────────
def _now_iso() -> str:
    """결정성 ts (테스트는 clock callable 로 override)."""
    return datetime.now(timezone.utc).isoformat(timespec="seconds")


def make_dedupe_key(pr_number: int, head_sha: str) -> str:
    """dedupe key 생성 — `{pr}#{head_sha}`.

    update-branch 후 head_sha 가 바뀌면 자동으로 다른 dedupe key 가 되어 stale 처리.
    회장 §명시: dedupe 단위는 **PR/head** 1조.
    """
    return f"{pr_number}#{head_sha}"


def _hash_token(token: str) -> str:
    """token 의 sha256 prefix 12자 — token_present 검증 + audit 박제용.

    토큰 raw 가 audit/log 에 절대 들어가지 않게, hash 만 기록한다.
    """
    return hashlib.sha256(token.encode("utf-8")).hexdigest()[:12]


def _redact_token(text: str | None, token: str) -> str:
    """text 안의 token substring 을 모두 `***REDACTED***` 로 치환.

    회장 §명시 token redaction guard — stderr / audit dict / error message 등
    외부 출력 경로 어디에서도 token raw 가 노출되지 않게 한다.

    - token 이 빈 문자열이면 redaction 무의미 → 원본 그대로 반환.
    - text 가 None 이면 빈 문자열 반환 (직렬화 안전성).
    - substring 일치 (대소문자 구분) — fine-grained PAT 은 case-sensitive.
    """
    if text is None:
        return ""
    if not token:
        return str(text)
    return str(text).replace(token, REDACTED_PLACEHOLDER)


def load_owner_pat(token_env: str = OWNER_PAT_ENV_NAME) -> str:
    """OWNER PAT env 에서 로드 — **fail-fast**, default GH_TOKEN fallback 금지.

    회장 §15 금지: default GH_TOKEN fallback X. 토큰 누락 시 즉시 `RuntimeError`.
    호출부는 RuntimeError 를 catch 하여 REJECT outcome 으로 변환한다.

    토큰 값 자체는 본 함수 반환값으로만 전달되고 어떤 print/log 에도 노출되지 않는다.
    """
    token = os.environ.get(token_env, "").strip()
    if not token:
        # 에러 메시지에도 토큰 누출 0 — env 이름만 노출.
        raise RuntimeError(f"OWNER_PAT_MISSING: env {token_env} not set or empty")
    return token


def serialize_decision(decision: OwnerTriggerDecision) -> dict[str, Any]:
    """OwnerTriggerDecision → JSON 직렬화 가능 dict.

    frozen dataclass 의 asdict 결과를 그대로 반환. 토큰 필드 0 (구조적으로 차단).
    """
    return dict(asdict(decision))


def write_decision_json(decision_dict: Mapping[str, Any], path: Path) -> None:
    """decision JSON atomic write — `.tmp` → rename.

    중간 실패 시 부분 파일이 남지 않게 same-fs rename 원자성 활용.
    상위 디렉토리는 호출부가 보장하지만 안전을 위해 mkdir(parents, exist_ok).
    """
    target = Path(path)
    target.parent.mkdir(parents=True, exist_ok=True)
    tmp = target.with_suffix(target.suffix + ".tmp")
    payload = json.dumps(dict(decision_dict), ensure_ascii=False, sort_keys=True, indent=2)
    tmp.write_text(payload, encoding="utf-8")
    os.replace(tmp, target)


def _safe_dedupe_filename(dedupe_key: str) -> str:
    """dedupe_key 를 파일명으로 사용할 수 있게 path-safe 문자만 남긴다.

    `pr_number#head_sha` 형식이지만 `#` 등은 일부 FS 에서 회피하는 편이 안전. SHA 의
    16진수만이라도 그대로 보존되도록 `[A-Za-z0-9._-]` 외 문자는 `_` 로 치환.
    """
    return re.sub(r"[^A-Za-z0-9._-]", "_", dedupe_key)


def _dedupe_marker_path(audit_log_path: Path, dedupe_key: str) -> Path:
    """dedupe 선점 마커 파일 경로 — audit jsonl 와 같은 디렉토리에 박제.

    `<audit_dir>/.<audit_stem>.dedupe.<safe_key>.marker` 형식으로 audit 파일과
    동일 디렉토리에 위치시켜 같은 fs 의 atomic create 성질을 활용한다.
    """
    audit_path = Path(audit_log_path)
    safe_key = _safe_dedupe_filename(dedupe_key)
    return audit_path.parent / f".{audit_path.stem}.dedupe.{safe_key}.marker"


def _safe_remove_marker(marker_path: Path) -> None:
    """dedupe 선점 마커를 best-effort 로 제거 (실패해도 silent).

    회장 Codex G1 round 2 High — gh 실패 후 marker 가 남으면 같은 PR/head 의
    재시도가 영구 차단된다. trigger 실패 경로에서만 호출되어 marker 를 정리,
    다음 사이클이 자유롭게 재시도하도록 한다. 마커 미존재나 race 로 사라진
    경우는 무시 (FileNotFoundError) — 본 의도는 "있다면 지운다".

    OSError 외 예외는 trigger 흐름과 무관하므로 silent (예: permission denied
    시에도 trigger 실패 흐름은 계속). 보안 경계 위반 가능성 0 — 마커 정리는
    어떤 endpoint/token 도 건드리지 않는다.
    """
    try:
        marker_path.unlink()
    except FileNotFoundError:
        return
    except OSError:
        # 권한 / fs 에러 등 — silent. 다음 사이클 재시도는 jsonl audit 로 폴백 차단.
        return


def is_duplicate_trigger(audit_log_path: Path, dedupe_key: str) -> bool:
    """audit jsonl 또는 dedupe 마커 파일에서 같은 dedupe_key 선점 상태 검사.

    회장 §명시: PR/head 단위 1회 only. 새 head_sha 면 다른 dedupe_key 가 되어 자동 stale.

    TOCTOU 보강 (회장 Codex G1 Medium 3):
      - 1차 검사: jsonl 에 outcome == "ok"/"pending" 엔트리 존재 여부.
      - 2차 검사: 같은 디렉토리에 dedupe 마커 파일이 존재하는지 (gh runner 호출
        직전에 O_EXCL 로 박제되는 파일). 둘 중 하나라도 발견되면 중복으로 판정.

    설계:
      - jsonl append-only — 한 줄씩 파싱.
      - 깨진 라인은 skip (안전 fallback). 단, 정상 라인이 1개라도 매칭되면 True.
      - 파일 미존재 = 첫 trigger = False.
    """
    path = Path(audit_log_path)
    # 2차 — 마커 파일 우선 검사 (jsonl 보다 가벼움)
    marker_path = _dedupe_marker_path(path, dedupe_key)
    if marker_path.exists():
        return True
    if not path.exists():
        return False
    # jsonl append-only — 파일 객체 line-by-line 순회 (전량 메모리 적재 회피).
    # docstring "한 줄씩 파싱" 설계 의도와 정합. read_text+splitlines 시맨틱 1:1
    # (매칭 True / 깨진 라인 skip / 미존재·OSError False / 마커 우선) 보존.
    # decode-error parity: UnicodeDecodeError(ValueError 계열, OSError 아님)는
    # 양 형태 모두 포착하지 않고 전파 → 'except OSError' 만 유지(확대 금지).
    try:
        with path.open("r", encoding="utf-8") as f:
            for line in f:
                line = line.strip()
                if not line:
                    continue
                try:
                    record = json.loads(line)
                except json.JSONDecodeError:
                    continue
                if not isinstance(record, dict):
                    continue
                if record.get("dedupe_key") != dedupe_key:
                    continue
                outcome = record.get("outcome")
                if outcome == OUTCOME_OK or outcome == OUTCOME_PENDING:
                    return True
    except OSError:
        return False
    return False


# ─── 보안 경계 검증 (정적 차단) ───────────────────────────────────────────────
def _build_allowed_gh_args(owner: str, repo: str, pr_number: int) -> list[str]:
    """허용되는 단 1 가지 gh api 호출 인자 — 정확히 이것만 통과시킨다.

    회장 §명시: comment trigger 1 endpoint only. 다른 endpoint 호출 0.
    """
    endpoint = f"/repos/{owner}/{repo}/issues/{pr_number}/comments"
    return [
        "api",
        "-X", "POST",
        endpoint,
        "-f", f"body={ALLOWED_COMMENT_BODY}",
    ]


def _assert_args_allowlist(args: Sequence[str], owner: str, repo: str, pr_number: int) -> None:
    """gh args 가 허용 list 와 **정확히 일치**하는지 검증. 불일치 시 fail-fast.

    - endpoint path 가 다른 경우 → ENDPOINT_NOT_ALLOWED.
    - body 가 정확히 `/gemini review` 가 아닌 경우 → BODY_NOT_ALLOWED.
    - 다른 어떤 추가 인자도 허용되지 않는다 (예: --silent, --paginate 등).
    """
    expected = _build_allowed_gh_args(owner, repo, pr_number)
    args_list = list(args)
    if args_list == expected:
        return
    # 어떤 토큰이 다른지 분류하여 더 정확한 차단 사유 보고.
    # body= 필드는 따로 검사 (body strict equality).
    for token in args_list:
        if isinstance(token, str) and token.startswith("body=") and token != f"body={ALLOWED_COMMENT_BODY}":
            raise RuntimeError(ERR_BODY_NOT_ALLOWED)
    # path 가 허용 endpoint 와 다르면 endpoint 위반.
    raise RuntimeError(ERR_ENDPOINT_NOT_ALLOWED)


def assert_body_allowed(body: str) -> None:
    """comment body 가 strict equality `/gemini review` 인지 검증. fail-fast.

    회장 §명시 6 허용 1: comment body 정확히 `/gemini review` only.
    - 앞뒤 공백 / 대소문자 변형 / suffix 추가 등 모두 BODY_NOT_ALLOWED.
    """
    if body != ALLOWED_COMMENT_BODY:
        raise RuntimeError(ERR_BODY_NOT_ALLOWED)


def assert_endpoint_allowed(endpoint: str, owner: str, repo: str, pr_number: int) -> None:
    """endpoint path 가 허용된 issue comments POST 1 endpoint 인지 검증.

    회장 §명시 15 금지 1~5: merge/approve/push/close/reopen 호출 0. 본 함수가
    엔드포인트 정적 allowlist 의 단일 진입점.
    """
    expected_endpoint = f"/repos/{owner}/{repo}/issues/{pr_number}/comments"
    if endpoint != expected_endpoint:
        raise RuntimeError(ERR_ENDPOINT_NOT_ALLOWED)


# 회장 §명시 — 본 모듈에서 실제 호출 가능한 method name 정적 검사 대상 (regression grep).
# 호출부 / 코멘트 / docstring 에서 키워드 등장은 OK 이나 실제 gh api 호출 path 에는 X.
# (보안 어설션 grep 카운트는 docstring/주석 포함 카운트되므로 actual call 은 0 임을 별도 검증.)
_FORBIDDEN_ENDPOINT_FRAGMENTS: tuple[str, ...] = (
    "/merges",
    "/reviews",
    "/pulls/",
    "/branches/",
    "/git/refs",
)


def _validate_no_forbidden_fragments(args: Sequence[str]) -> None:
    """방어적 2차 차단: args 안에 금지 endpoint fragment 가 섞여 있는지 추가 검사.

    `_assert_args_allowlist` 가 이미 strict equality 로 차단하지만, 호출부가
    내부 헬퍼를 우회해 직접 gh_runner 를 호출하더라도 본 함수가 raise 하도록 한다.
    """
    for token in args:
        if not isinstance(token, str):
            continue
        for frag in _FORBIDDEN_ENDPOINT_FRAGMENTS:
            if frag in token:
                raise RuntimeError(ERR_ENDPOINT_NOT_ALLOWED)


# ─── 본체 ────────────────────────────────────────────────────────────────────
class OwnerTriggerPat:
    """Fine-grained OWNER PAT trigger-only comment writer.

    회장 §명시 (2026-05-11 dec_1) 1:1 강제:
      - OWNER PAT 은 `/gemini review` 댓글 작성 1 endpoint 에서만 사용.
      - merge/approve/push/close/reopen 호출 0 (정적 차단).
      - audit 에 token value 0, token_present + token_hash 만.
      - dedupe by (pr_number, head_sha) — update-branch 후 새 head 는 새 trigger 허용.

    외부 부수효과는 생성자 주입 callable (gh_runner / audit_writer / decision_writer /
    clock) 로 분리되어 테스트에서 mock 으로 교체 가능. 본 모듈은 anu_v2/* 외 어떤 외부
    모듈도 import 하지 않는다 (one-way isolation).
    """

    def __init__(
        self,
        *,
        gh_runner: GhRunner,
        audit_writer: AuditWriter,
        decision_writer: DecisionWriter,
        owner: str,
        repo: str,
        clock: Clock = _now_iso,
        token_env: str = OWNER_PAT_ENV_NAME,
    ) -> None:
        if not owner or not isinstance(owner, str):
            raise ValueError("owner must be non-empty str")
        if not repo or not isinstance(repo, str):
            raise ValueError("repo must be non-empty str")
        # owner/repo 에 path traversal 문자가 들어오면 endpoint 가 다른 repo 로 위장될
        # 수 있으므로 strict 검증. github org/repo 명명 규칙: 영문/숫자/`-`/`_`/`.`.
        if not re.fullmatch(r"[A-Za-z0-9._-]+", owner):
            raise ValueError("owner contains invalid characters")
        if not re.fullmatch(r"[A-Za-z0-9._-]+", repo):
            raise ValueError("repo contains invalid characters")
        # 회장 Codex G1 round 2 High — token_env override 화이트리스트 강제.
        # 운영 코드에서는 OWNER_PAT_ENV_NAME 고정값만 허용. BOT_GITHUB_TOKEN 등 다른
        # env 이름이 전달되면 즉시 ValueError → bot 토큰으로 `/gemini review` 작성하는
        # 우회 차단 (회장 §15 금지 15번 — BOT_GITHUB_TOKEN 대신 OWNER PAT 사용 금지의
        # 역방향 차단: OWNER PAT 자리에 BOT_GITHUB_TOKEN 도 동등하게 금지).
        if token_env != OWNER_PAT_ENV_NAME:
            raise ValueError(
                f"token_env must be {OWNER_PAT_ENV_NAME!r}; got {token_env!r}"
            )
        self._gh = gh_runner
        self._audit = audit_writer
        self._decision = decision_writer
        self._owner = owner
        self._repo = repo
        self._clock: Clock = clock
        self._token_env = token_env

    # ── 결정 박제 (PASS 또는 REJECT) ──────────────────────────────────────────
    def _make_decision(
        self,
        *,
        pr_number: int,
        head_sha: str,
        decision: str,
        reason: str,
        gemini_evidence_state: str,
        queue_position: int,
    ) -> OwnerTriggerDecision:
        return OwnerTriggerDecision(
            pr_number=pr_number,
            head_sha=head_sha,
            decision=decision,
            reason=reason,
            gemini_evidence_state=gemini_evidence_state,
            queue_position=queue_position,
            dedupe_key=make_dedupe_key(pr_number, head_sha),
            ts=self._clock(),
        )

    def _persist_decision(self, decision: OwnerTriggerDecision, path: Path) -> None:
        """decision_writer 콜백 호출 — REJECT 도 박제하여 회장 §명시 audit trail 유지."""
        self._decision(serialize_decision(decision), path)

    # ── audit jsonl 1건 append (token raw 0 보장) ────────────────────────────
    def _append_audit(
        self,
        *,
        pr_number: int,
        head_sha: str,
        outcome: str,
        reason: str,
        token_present: bool,
        token_hash: str,
        ts: str,
    ) -> None:
        record = {
            "ts": ts,
            "pr_number": pr_number,
            "head_sha": head_sha,
            "dedupe_key": make_dedupe_key(pr_number, head_sha),
            "outcome": outcome,
            "reason": reason,
            "actor": ACTOR_OWNER_TRIGGER_PAT,
            "token_present": bool(token_present),
            "token_hash": token_hash,
        }
        # 방어적 — record 어디에도 토큰 값이 들어가지 않음을 다시 확인할 수 있도록
        # 구조적으로 토큰 raw 가 들어갈 수 있는 필드를 두지 않았다 (정적 차단).
        self._audit(record)

    # ── trigger 메인 진입점 ───────────────────────────────────────────────────
    def trigger_gemini_review(
        self,
        pr_number: int,
        head_sha: str,
        *,
        queue_position: int,
        gemini_evidence_state: str,
        audit_log_path: Path,
        decision_json_path: Path,
    ) -> TriggerOutcome:
        """queue-head + evidence missing + dedupe + token 검증 후 `/gemini review` 댓글 1회 작성.

        회장 §명시 6 허용 1:1:
          1. body 정확히 `/gemini review` (내부 상수, 사용자 입력 받지 않음)
          2. queue_position == 0 (queue-head only)
          3. gemini_evidence_state == "missing_for_current_head"
          4. dedupe 검증 — (pr, head) 1회 only
          5. token 로딩 성공 (없으면 REJECT)
          6. owner_trigger_decision.json 박제 PASS 후에만 실제 호출

        모든 REJECT 케이스도 audit + decision 박제 — 추적성 유지.
        """
        # ── 1. body 검증 (내부 상수, strict equality) ──────────────────────────
        assert_body_allowed(ALLOWED_COMMENT_BODY)

        ts = self._clock()
        decision_path = Path(decision_json_path)
        audit_path = Path(audit_log_path)

        # ── 2. queue-head 검증 ────────────────────────────────────────────────
        if queue_position != 0:
            reason = f"not_queue_head:queue_position={queue_position}"
            decision = self._make_decision(
                pr_number=pr_number,
                head_sha=head_sha,
                decision=DECISION_REJECT,
                reason=reason,
                gemini_evidence_state=gemini_evidence_state,
                queue_position=queue_position,
            )
            self._persist_decision(decision, decision_path)
            self._append_audit(
                pr_number=pr_number,
                head_sha=head_sha,
                outcome=OUTCOME_REJECTED,
                reason=reason,
                token_present=False,
                token_hash="",
                ts=ts,
            )
            return TriggerOutcome(
                outcome=OUTCOME_REJECTED,
                reason=reason,
                decision_path=str(decision_path),
            )

        # ── 3. evidence state 검증 ────────────────────────────────────────────
        if gemini_evidence_state != EVIDENCE_MISSING_FOR_CURRENT_HEAD:
            reason = f"evidence_not_missing:state={gemini_evidence_state}"
            decision = self._make_decision(
                pr_number=pr_number,
                head_sha=head_sha,
                decision=DECISION_REJECT,
                reason=reason,
                gemini_evidence_state=gemini_evidence_state,
                queue_position=queue_position,
            )
            self._persist_decision(decision, decision_path)
            self._append_audit(
                pr_number=pr_number,
                head_sha=head_sha,
                outcome=OUTCOME_REJECTED,
                reason=reason,
                token_present=False,
                token_hash="",
                ts=ts,
            )
            return TriggerOutcome(
                outcome=OUTCOME_REJECTED,
                reason=reason,
                decision_path=str(decision_path),
            )

        # ── 4. dedupe 검증 ────────────────────────────────────────────────────
        dedupe_key = make_dedupe_key(pr_number, head_sha)
        if is_duplicate_trigger(audit_path, dedupe_key):
            reason = f"duplicate_trigger:dedupe_key={dedupe_key}"
            decision = self._make_decision(
                pr_number=pr_number,
                head_sha=head_sha,
                decision=DECISION_REJECT,
                reason=reason,
                gemini_evidence_state=gemini_evidence_state,
                queue_position=queue_position,
            )
            self._persist_decision(decision, decision_path)
            self._append_audit(
                pr_number=pr_number,
                head_sha=head_sha,
                outcome=OUTCOME_REJECTED,
                reason=reason,
                token_present=False,
                token_hash="",
                ts=ts,
            )
            return TriggerOutcome(
                outcome=OUTCOME_REJECTED,
                reason=reason,
                decision_path=str(decision_path),
            )

        # ── 5. token 로딩 (fail-fast) ────────────────────────────────────────
        try:
            token = load_owner_pat(self._token_env)
        except RuntimeError as exc:
            # exc 메시지에는 env 이름만 들어있고 token 값 X (load_owner_pat 보장).
            # 추가 방어로 redact 한 번 더 적용 (token 변수가 없으므로 no-op 안전).
            reason = f"token_missing:{exc}"
            decision = self._make_decision(
                pr_number=pr_number,
                head_sha=head_sha,
                decision=DECISION_REJECT,
                reason=reason,
                gemini_evidence_state=gemini_evidence_state,
                queue_position=queue_position,
            )
            self._persist_decision(decision, decision_path)
            self._append_audit(
                pr_number=pr_number,
                head_sha=head_sha,
                outcome=OUTCOME_REJECTED,
                reason=reason,
                token_present=False,
                token_hash="",
                ts=ts,
            )
            return TriggerOutcome(
                outcome=OUTCOME_REJECTED,
                reason=reason,
                decision_path=str(decision_path),
            )

        # 토큰 hash 만 audit 에 박제 (token raw 0 보장).
        token_hash = _hash_token(token)

        # ── 6. PASS 결정 박제 ────────────────────────────────────────────────
        pass_decision = self._make_decision(
            pr_number=pr_number,
            head_sha=head_sha,
            decision=DECISION_PASS,
            reason="all_gates_pass:queue_head+evidence_missing+dedupe_clean+token_loaded",
            gemini_evidence_state=gemini_evidence_state,
            queue_position=queue_position,
        )
        self._persist_decision(pass_decision, decision_path)

        # ── 6.5. atomic dedupe 선점 (회장 Codex G1 Medium 3) ─────────────────
        # gh runner 호출 직전에 dedupe_key 박제 마커 파일을 `O_CREAT|O_EXCL` 로
        # 생성해 TOCTOU 윈도우를 닫는다. 이미 존재하면 동시 trigger 가 진행 중이거나
        # 이전 사이클이 박제한 상태 → REJECT. 파일 자체는 jsonl audit 와 분리되어
        # 기존 audit 호출 카운트에 영향을 주지 않는다 (하위 호환).
        dedupe_marker_path = _dedupe_marker_path(audit_path, dedupe_key)
        try:
            fd = os.open(
                str(dedupe_marker_path),
                os.O_CREAT | os.O_EXCL | os.O_WRONLY,
                0o600,
            )
            try:
                os.write(
                    fd,
                    json.dumps(
                        {
                            "dedupe_key": dedupe_key,
                            "pr_number": pr_number,
                            "head_sha": head_sha,
                            "outcome": OUTCOME_PENDING,
                            "ts": ts,
                        },
                        ensure_ascii=False,
                        sort_keys=True,
                    ).encode("utf-8"),
                )
            finally:
                os.close(fd)
        except FileExistsError:
            # 같은 dedupe_key 마커가 이미 존재 → 다른 컨텍스트 / 이전 사이클이 박제.
            # 보수적으로 REJECT. is_duplicate_trigger jsonl 검사가 1차 차단이고
            # 본 분기는 동시성 race 의 2차 차단.
            reason = f"duplicate_trigger_marker:dedupe_key={dedupe_key}"
            self._append_audit(
                pr_number=pr_number,
                head_sha=head_sha,
                outcome=OUTCOME_REJECTED,
                reason=reason,
                token_present=True,
                token_hash=token_hash,
                ts=ts,
            )
            return TriggerOutcome(
                outcome=OUTCOME_REJECTED,
                reason=reason,
                decision_path=str(decision_path),
            )

        # ── 7. gh api POST issue comment 1회 호출 ────────────────────────────
        # endpoint + body 는 _build_allowed_gh_args 가 박제. 다른 args 합성 경로 X.
        args = _build_allowed_gh_args(self._owner, self._repo, pr_number)
        # 정적 차단: allowlist 일치 + 금지 fragment 미포함 (이중 검사).
        _assert_args_allowlist(args, self._owner, self._repo, pr_number)
        _validate_no_forbidden_fragments(args)

        # process-local env: 호스트 PATH 등은 유지하되 GH_TOKEN/GITHUB_TOKEN 은
        # OWNER PAT 으로 명시적 덮어쓰기. 다른 env 변수는 그대로 (PATH 등 실행 인프라).
        env = os.environ.copy()
        env["GH_TOKEN"] = token
        env["GITHUB_TOKEN"] = token

        try:
            cp = self._gh(args, env)
        except Exception as exc:  # noqa: BLE001 — 어떤 예외든 redact + audit
            # 회장 Codex G1 round 2 High — 실패 시 marker 정리 (재시도 가능).
            # 회장 §명시: "PR/head당 1회 only" 는 성공 1회. 실패한 trigger 는 같은
            # head 에서 재시도 가능해야 self-resume 가 동작한다.
            _safe_remove_marker(dedupe_marker_path)
            # 예외 메시지에 토큰이 누출되는 경우를 대비해 redact 적용.
            sanitized = _redact_token(f"{type(exc).__name__}: {exc}", token)
            self._append_audit(
                pr_number=pr_number,
                head_sha=head_sha,
                outcome=OUTCOME_FAILED,
                reason=f"gh_runner_exception:{sanitized}",
                token_present=True,
                token_hash=token_hash,
                ts=ts,
            )
            return TriggerOutcome(
                outcome=OUTCOME_FAILED,
                reason=f"gh_runner_exception:{sanitized}",
                decision_path=str(decision_path),
            )

        # gh stderr/stdout 에 토큰이 들어갈 가능성을 대비해 redact 적용 (안전 fallback).
        stderr_raw = _coerce_stream(getattr(cp, "stderr", None))
        stderr_redacted = _redact_token(stderr_raw, token)

        if cp.returncode != 0:
            # 회장 Codex G1 round 2 High — gh 실패 시 marker 정리 (재시도 가능).
            _safe_remove_marker(dedupe_marker_path)
            reason = f"gh_api_failed:rc={cp.returncode}:stderr={stderr_redacted[:256]}"
            self._append_audit(
                pr_number=pr_number,
                head_sha=head_sha,
                outcome=OUTCOME_FAILED,
                reason=reason,
                token_present=True,
                token_hash=token_hash,
                ts=ts,
            )
            return TriggerOutcome(
                outcome=OUTCOME_FAILED,
                reason=reason,
                decision_path=str(decision_path),
            )

        # ── 8. 성공 audit append ─────────────────────────────────────────────
        self._append_audit(
            pr_number=pr_number,
            head_sha=head_sha,
            outcome=OUTCOME_OK,
            reason="comment_posted:/gemini review",
            token_present=True,
            token_hash=token_hash,
            ts=ts,
        )
        return TriggerOutcome(
            outcome=OUTCOME_OK,
            reason="comment_posted",
            decision_path=str(decision_path),
        )


# ─── 보조 helpers ────────────────────────────────────────────────────────────
def _coerce_stream(value: Any) -> str:
    """subprocess stdout/stderr 정규화 (merge_queue_executor 동일 패턴).

    None → "", bytes → utf-8 decode, str → 그대로. 직렬화 시점에 bytes 가 dict 에
    박히지 않게 한다.
    """
    if value is None:
        return ""
    if isinstance(value, bytes):
        return value.decode("utf-8", errors="replace")
    return str(value)


def default_audit_writer(audit_log_path: Path) -> AuditWriter:
    """기본 audit writer factory — jsonl 1줄 append.

    호출부가 명시적으로 path 를 묶어 callable 을 만들 수 있도록 closure 반환.
    테스트에서는 list-collector callable 로 대체.
    """
    target = Path(audit_log_path)

    def _writer(record: Mapping[str, Any]) -> None:
        target.parent.mkdir(parents=True, exist_ok=True)
        line = json.dumps(dict(record), ensure_ascii=False, sort_keys=True)
        with target.open("a", encoding="utf-8") as fp:
            fp.write(line + "\n")

    return _writer