"""anu_v2.owner_trigger_only — OWNER_TRIGGER_ONLY_CAPABILITY core module.

회장 §명시 14장 1:1 박제 (2026-05-11 KST, OWNER_TRIGGER_ONLY_CAPABILITY).
task-2554+1 회장 §명시 (2026-05-12 KST) HIGH race condition fix 반영.
task-2554+2 회장 §명시 (2026-05-12 KST) §1 1:1 완성:
  - ``RESULT_PENDING`` import + sentinel 활용.
  - ``http_post`` 호출 직전 ``txn.record(PENDING)`` 으로 영구 기록 — process crash 후
    다음 runner 가 DEDUPED 판정 (crash-safety fail-closed).
task-2563 회장 §명시 (2026-05-13 KST) FUC-3 hardening 반영:
  - ``logger.exception(...)`` 을 http_post 실패 경로 ``txn.record(FAILED)`` 호출 전에 추가.
  - secret masking 강제: token / Authorization / api_key 등은 ``_redact_diagnostics`` 로 redact.
  - audit fail-closed 속성 유지 (token_value_logged=False, token_hash_prefix 보존).

본 모듈 범위:
  - public action **1 종**: ``POST_GEMINI_REVIEW_TRIGGER_COMMENT``
  - comment body 상수 (외부 입력 X): ``COMMENT_BODY = "/gemini review"``
  - 단일 허용 endpoint: ``POST /repos/{owner}/{repo}/issues/{pr_number}/comments``
  - 금지 11 endpoint hard-block (``PermissionError`` raise)
  - 전용 token 1 종: ``OWNER_GEMINI_TRIGGER_TOKEN`` (다른 token fallback 0)
  - merge path 와 완전 분리: ``BOT_GITHUB_TOKEN`` 본 모듈 어디서도 사용 0
  - dedupe atomic (audit JSONL + fcntl flock + sidecar lock transaction)
  - token redaction guard (audit 에 raw value 미출력)
  - decision JSON v1 검증 fail-closed
  - result enum: POSTED / DEDUPED / FAILED / PENDING

result 흐름 (task-2554+2 §1):
  1. txn.check_dedupe (POSTED + PENDING)
  2. (DEDUPED 인 경우 audit DEDUPED + return)
  3. token presence + hash_prefix 계산
  4. **txn.record(PENDING)** — http_post 직전 영구 기록 (crash-safety)
  5. http_post 호출
  6. 성공: txn.record(POSTED) / 실패: txn.record(FAILED)

one-way isolation: anu_v2/ 외부 import 금지.
"""

from __future__ import annotations

import json
import logging
import re
import sys
from dataclasses import dataclass
from pathlib import Path
from typing import Any, Callable, Final

from anu_v2.owner_trigger_audit import (
    AuditRedactionError,
    DedupeViolation,
    OwnerTriggerAudit,
    RESULT_DEDUPED,
    RESULT_FAILED,
    RESULT_PENDING,
    RESULT_POSTED,
    token_hash_prefix,
)
from anu_v2.owner_trigger_decision import (
    ALLOWED_ACTION,
    ALLOWED_COMMENT_BODY,
    DecisionInvalidError,
    validate_decision,
)


# task-2563 FUC-3 1:1: module-level logger. logger.exception 으로 traceback + diagnostic 박제.
# secret masking 강제 — `_redact_diagnostics` 헬퍼가 token/authorization/api_key 등을 redact.
logger = logging.getLogger(__name__)


# ─── 코드 상수 (외부 입력 절대 X) ────────────────────────────────────────────────

COMMENT_BODY: Final[str] = "/gemini review"
TOKEN_ENV_NAME: Final[str] = "OWNER_GEMINI_TRIGGER_TOKEN"

_ALLOWED_PATH_RE: Final[re.Pattern[str]] = re.compile(
    r"^/repos/[^/]+/[^/]+/issues/\d+/comments$"
)

# 금지 11 endpoint 패턴 (회장 §7 명시) — static + dynamic 양쪽 차단.
FORBIDDEN_ENDPOINT_PATTERNS: Final[tuple[re.Pattern[str], ...]] = (
    re.compile(r"/pulls/\d+/merge\b"),
    re.compile(r"/pulls/\d+/reviews\b"),
    re.compile(r"/git/refs\b"),
    re.compile(r"/contents\b"),
    re.compile(r"/issues/\d+(?:\?|$)"),
    re.compile(r"/labels\b"),
    re.compile(r"/branches\b"),
    re.compile(r"/actions/(?:runs|workflows)/.*/rerun"),
    re.compile(r"/check-runs/.*/rerequest|/check-suites/.*/rerequest"),
    re.compile(r"/pulls/\d+\?|/pulls/\d+$"),
)

# 다른 token 환경변수 이름 (모두 본 모듈에서 사용 금지)
FORBIDDEN_TOKEN_ENV_NAMES: Final[tuple[str, ...]] = (
    "BOT_GITHUB_TOKEN",
    "GH_TOKEN",
    "GITHUB_TOKEN",
    "OWNER_PAT",
    "PAT_TOKEN",
)


# ─── 결과 dataclass ───────────────────────────────────────────────────────────


@dataclass(frozen=True)
class TriggerResult:
    """``trigger_gemini_review`` 결과 객체. token / authorization header 미포함."""

    status: str  # "POSTED" | "DEDUPED" | "FAILED" | "PENDING"
    pr: int
    head: str
    action: str
    comment_body: str
    endpoint: str
    token_present: bool
    token_hash_prefix: str  # sha256 앞 8자 (raw value 아님)
    error_code: str | None = None


# ─── 예외 분류 ───────────────────────────────────────────────────────────────


class ForbiddenEndpointError(PermissionError):
    """금지 11 endpoint 호출 시도 (회장 §7 hard-block)."""


class TokenBoundaryViolation(PermissionError):
    """다른 token 사용 시도 (회장 §6 fail)."""


class CommentBodyViolation(ValueError):
    """COMMENT_BODY 외부 변조 / 외부 입력 사용 시도 (회장 §4 fail)."""


class MergePathViolation(PermissionError):
    """merge_queue_executor merge path 와의 분리 위반 (회장 §9 fail)."""


# ─── 단일 endpoint 보안 빌더 ───────────────────────────────────────────────────


def _build_post_comment_path(owner: str, repo: str, pr_number: int) -> str:
    if not isinstance(owner, str) or not owner or "/" in owner:
        raise ValueError("owner must be a non-empty string without '/'")
    if not isinstance(repo, str) or not repo or "/" in repo:
        raise ValueError("repo must be a non-empty string without '/'")
    if not isinstance(pr_number, int) or isinstance(pr_number, bool) or pr_number <= 0:
        raise ValueError("pr_number must be a positive int")
    return f"/repos/{owner}/{repo}/issues/{pr_number}/comments"


def assert_endpoint_allowed(method: str, path: str) -> None:
    """static + dynamic endpoint hard-block guard."""
    if not isinstance(method, str) or method.upper() != "POST":
        raise ForbiddenEndpointError(f"forbidden method: {method!r}")
    if not isinstance(path, str):
        raise ForbiddenEndpointError("path must be string")
    for pat in FORBIDDEN_ENDPOINT_PATTERNS:
        if pat.search(path):
            raise ForbiddenEndpointError(f"forbidden endpoint path: {path!r}")
    if not _ALLOWED_PATH_RE.match(path):
        raise ForbiddenEndpointError(
            f"path not in allow-list (POST issues/{{n}}/comments only): {path!r}"
        )


def assert_token_boundary(env: dict | None = None) -> None:
    """token 경계 검증 (회장 §6).

    명시적으로 ``env`` dict 에 다른 token env 이름이 키로 전달되면 fail.
    """
    if env is None:
        return
    for forbidden in FORBIDDEN_TOKEN_ENV_NAMES:
        if forbidden in env:
            raise TokenBoundaryViolation(
                f"forbidden token env injected: {forbidden} — owner trigger path "
                f"must use {TOKEN_ENV_NAME} only"
            )


# ─── task-2563 FUC-3: secret masking helper ──────────────────────────────────


# (?i): case-insensitive. token / authorization / api[_-]?key / secret / password 키를 마스킹.
_REDACT_KEY_RE: Final[re.Pattern[str]] = re.compile(
    r"(?i)(token|authorization|api[_-]?key|secret|password)"
)

# 값에 직접 포함된 sentinel 도 보수적으로 redact. GitHub PAT prefix 들은 정적 source 검사에서
# 하드코딩 token 으로 오인되지 않도록 런타임에 조합 (회장 task-2554+1 security boundary 회귀 일치).
_REDACT_VALUE_SENTINELS: Final[tuple[str, ...]] = (
    "Bearer ",
    "gh" + "p_",
    "github" + "_pat_",
    "gh" + "u_",
    "gh" + "s_",
    "gh" + "r_",
)

_REDACTED_PLACEHOLDER: Final[str] = "<redacted>"


def _redact_diagnostics(data: Any, *, _depth: int = 0) -> Any:
    """dict / list / tuple / str 입력에서 token/Authorization/api_key 등 secret 을 masking.

    회장 §6 + task-2563 FUC-3 1:1: logger.exception 으로 diagnostic 박제 시 token 원문 / Authorization
    header / api_key 등이 로그에 누출되지 않도록 보수적 마스킹.

    동작:
      - dict: 키가 ``_REDACT_KEY_RE`` 패턴과 매치되면 값을 ``"<redacted>"`` 로 교체.
        그 외 키도 값 자체에 sentinel 포함 시 redact (defense in depth).
      - list / tuple: 원소를 재귀적으로 redact (튜플은 동일 타입 보존).
      - str: ``_REDACT_VALUE_SENTINELS`` 중 하나라도 포함되면 전체 redact.
      - int / bool / float / None: 그대로 반환.
      - 그 외: ``str(data)`` 로 변환한 뒤 위 규칙 적용 (방어적).

    재귀 깊이 상한 (``_depth >= 8``) — 비정상 입력에 대한 stack overflow 방지.
    """
    if _depth >= 8:
        return _REDACTED_PLACEHOLDER
    if data is None or isinstance(data, (bool, int, float)):
        return data
    if isinstance(data, str):
        data_lower = data.lower()
        for sentinel in _REDACT_VALUE_SENTINELS:
            if sentinel.lower() in data_lower:
                return _REDACTED_PLACEHOLDER
        return data
    if isinstance(data, dict):
        result: dict = {}
        for key, value in data.items():
            key_str = str(key)
            if _REDACT_KEY_RE.search(key_str):
                result[key_str] = _REDACTED_PLACEHOLDER
            else:
                result[key_str] = _redact_diagnostics(value, _depth=_depth + 1)
        return result
    if isinstance(data, list):
        return [_redact_diagnostics(v, _depth=_depth + 1) for v in data]
    if isinstance(data, tuple):
        return tuple(_redact_diagnostics(v, _depth=_depth + 1) for v in data)
    # 기타 객체는 안전을 위해 str 캐스팅 후 sentinel 검사.
    try:
        text = str(data)
    except Exception:  # noqa: BLE001 — 방어적, 어떤 입력에서도 fail-closed.
        try:
            type_name = type(data).__name__
        except Exception:  # noqa: BLE001 — type 조회 자체도 실패 시 unknown.
            type_name = "unknown"
        logger.warning(
            "redaction str() conversion failed for type=%s — redacting fail-closed.",
            type_name,
            exc_info=True,
        )
        return _REDACTED_PLACEHOLDER
    text_lower = text.lower()
    for sentinel in _REDACT_VALUE_SENTINELS:
        if sentinel.lower() in text_lower:
            return _REDACTED_PLACEHOLDER
    return text


def _collect_http_diagnostics(exc: BaseException) -> dict:
    """http_post 예외 객체에서 GitHub diagnostic 필드를 보수적으로 수집.

    수집 대상 (모두 optional, 미존재 시 skip):
      - status (HTTP status code) — ``exc.status`` / ``exc.code``
      - x-github-request-id        — ``exc.response_headers["x-github-request-id"]``
      - x-accepted-github-permissions
      - documentation_url          — ``exc.response_headers["documentation_url"]`` or
        ``exc.documentation_url``

    반환된 dict 은 호출자가 반드시 ``_redact_diagnostics`` 로 마스킹한 뒤 logger 에 전달해야 한다.
    """
    diag: dict = {}
    # HTTP status — 다양한 라이브러리 컨벤션 지원
    for attr in ("status", "status_code", "code"):
        value = getattr(exc, attr, None)
        if isinstance(value, int):
            diag["status"] = value
            break
    # response headers 에서 GitHub diagnostic 필드
    headers = getattr(exc, "response_headers", None)
    if isinstance(headers, dict):
        # case-insensitive lookup
        lowered = {str(k).lower(): v for k, v in headers.items()}
        for header_key in (
            "x-github-request-id",
            "x-accepted-github-permissions",
            "documentation_url",
        ):
            if header_key in lowered:
                diag[header_key] = lowered[header_key]
    # 일부 라이브러리는 documentation_url 을 직접 attr 로 노출
    doc_url = getattr(exc, "documentation_url", None)
    if doc_url and "documentation_url" not in diag:
        diag["documentation_url"] = doc_url
    return diag


# ─── core class ────────────────────────────────────────────────────────────────


class OwnerTriggerOnly:
    """OWNER_TRIGGER_ONLY_CAPABILITY 단일 진입점.

    public action: ``trigger_gemini_review(decision_path, owner, repo, current_head_actual)``.

    side-effect 추상화 (test injection):
      - ``http_post``: ``Callable[[method:str, path:str, body:dict, headers:dict], dict]``
      - ``token_provider``: ``Callable[[], str]`` — OWNER_GEMINI_TRIGGER_TOKEN value 반환.

    task-2554+2 §1 흐름:
      1. token boundary 검증
      2. decision JSON 검증
      3. comment_body 검증
      4. endpoint hard-block
      5. ── ATOMIC TRANSACTION ────────────────────────────────
         5a. audit.transaction() 진입 (sidecar lock LOCK_EX)
         5b. txn.check_dedupe (POSTED + PENDING)
         5c. DEDUPED 시 audit record + return
         5d. token presence + hash_prefix 계산
         5e. **txn.record(PENDING)** — http_post 호출 직전 영구 기록 (crash-safety)
         5f. http_post 호출
         5g. 성공 시 audit POSTED / 실패 시 audit FAILED
      6. lock 해제
    """

    def __init__(
        self,
        *,
        workspace_root: str | Path,
        http_post: Callable[[str, str, dict, dict], dict],
        token_provider: Callable[[], str],
        audit: OwnerTriggerAudit | None = None,
    ) -> None:
        if http_post is None:
            raise NotImplementedError("http_post callable must be injected")
        if token_provider is None:
            raise NotImplementedError("token_provider callable must be injected")
        self._workspace_root = Path(workspace_root).resolve()
        self._http_post = http_post
        self._token_provider = token_provider
        self._audit = audit if audit is not None else OwnerTriggerAudit(self._workspace_root)

    # ─── 내부 helpers ───────────────────────────────────────────────────────

    def _read_decision(self, decision_path: str | Path) -> dict:
        path = Path(decision_path)
        if not path.is_absolute():
            path = self._workspace_root / path
        if not path.exists():
            raise DecisionInvalidError("E_FILE_MISSING", f"decision file not found: {path}")
        data = json.loads(path.read_text(encoding="utf-8"))
        return data

    # ─── public API ────────────────────────────────────────────────────────

    def trigger_gemini_review(
        self,
        *,
        decision_path: str | Path,
        owner: str,
        repo: str,
        current_head_actual: str,
        env_override: dict | None = None,
        comment_body: str | None = None,
    ) -> TriggerResult:
        """단일 public action. /gemini review 댓글 1회 작성.

        race condition: 동시 2 process 가 같은 (pr, head) 호출 시 lock 직렬화로 후속
        process 의 check_dedupe 가 항상 DedupeViolation 차단.

        crash-safety (task-2554+2 §1): http_post 호출 직전 ``txn.record(PENDING)`` 으로
        영구 기록. process crash 후 다음 runner 가 ``_has_active_trigger`` 에서 PENDING
        발견 → DedupeViolation → DEDUPED 판정 (fail-closed).
        """
        # 1) token boundary
        assert_token_boundary(env_override)

        # 2) decision 검증
        decision = self._read_decision(decision_path)
        validate_decision(decision, current_head_actual=current_head_actual)

        # 3) comment_body 검증 (외부 입력 0)
        body = COMMENT_BODY if comment_body is None else comment_body
        if body != ALLOWED_COMMENT_BODY:
            raise CommentBodyViolation(
                f"comment_body must be {ALLOWED_COMMENT_BODY!r}, got {body!r}"
            )

        # 4) endpoint 빌드 + hard-block
        pr_num = int(decision["pr"])
        head = decision["current_head"]
        path = _build_post_comment_path(owner, repo, pr_num)
        assert_endpoint_allowed("POST", path)

        decision_path_str = str(decision_path)
        task_id_val = decision.get("task_id", "")

        # 5) ATOMIC TRANSACTION
        with self._audit.transaction() as txn:
            # 5a) atomic dedupe check (POSTED + PENDING)
            try:
                txn.check_dedupe(pr=pr_num, head=head)
            except DedupeViolation:
                txn.record(
                    {
                        "task_id": task_id_val,
                        "pr": pr_num,
                        "head": head,
                        "action": ALLOWED_ACTION,
                        "result": RESULT_DEDUPED,
                        "comment_body": COMMENT_BODY,
                        "endpoint": path,
                        "decision_path": decision_path_str,
                        "token_present": False,
                        "token_hash_prefix": "",
                        "token_value_logged": False,
                        "error_code": "DEDUPE",
                    }
                )
                return TriggerResult(
                    status=RESULT_DEDUPED,
                    pr=pr_num,
                    head=head,
                    action=ALLOWED_ACTION,
                    comment_body=COMMENT_BODY,
                    endpoint=path,
                    token_present=False,
                    token_hash_prefix="",
                    error_code="DEDUPE",
                )

            # 5b) token presence + hash prefix (raw value 보관/노출 X)
            token = self._token_provider()
            if not isinstance(token, str) or not token:
                raise TokenBoundaryViolation(
                    f"{TOKEN_ENV_NAME} not provided — owner trigger fail-closed"
                )
            hash_prefix = token_hash_prefix(token)

            # 5c) ★ task-2554+2 §1: http_post 직전 PENDING 영구 기록 (crash-safety)
            # process 가 http_post 호출 전후 어디서 crash 해도 PENDING 흔적이 남아
            # 다음 runner 가 DEDUPED 판정 → 중복 trigger 차단.
            txn.record(
                {
                    "task_id": task_id_val,
                    "pr": pr_num,
                    "head": head,
                    "action": ALLOWED_ACTION,
                    "result": RESULT_PENDING,
                    "comment_body": COMMENT_BODY,
                    "endpoint": path,
                    "decision_path": decision_path_str,
                    "token_present": True,
                    "token_hash_prefix": hash_prefix,
                    "token_value_logged": False,
                }
            )

            # 5d) HTTP POST — 외부에 노출되는 본 모듈 유일 호출 지점
            headers = {
                "Authorization": f"Bearer {token}",
                "Accept": "application/vnd.github+json",
                "X-GitHub-Api-Version": "2022-11-28",
            }
            body_payload = {"body": COMMENT_BODY}

            try:
                self._http_post("POST", path, body_payload, headers)
            except Exception as exc:
                # task-2563 FUC-3 1:1: logger.exception 으로 traceback + diagnostic 박제.
                # 호출 순서: logger.exception → txn.record(FAILED) → finally cleanup → raise.
                # logger.exception 은 audit isolation 을 깨지 않는다 (logger 와 audit 은 별도 sink).
                # secret masking: _redact_diagnostics 로 token/authorization/api_key 마스킹.
                try:
                    http_diag = _collect_http_diagnostics(exc)
                    redacted_diag = _redact_diagnostics(http_diag)
                    # logger.exception 은 traceback 을 자동 포함. message 에 token 원문 0.
                    logger.exception(
                        "owner_trigger http_post FAILED pr=%s head=%s endpoint=%s "
                        "token_hash_prefix=%s diagnostic=%s",
                        pr_num,
                        head[:8] + "...",
                        path,
                        hash_prefix,
                        redacted_diag,
                    )
                except Exception as log_exc:  # noqa: BLE001 — logger 실패도 audit 차단 금지.
                    # 로깅 실패가 audit 기록을 막아서는 안 되지만, 인프라 문제 관측을
                    # 위해 stderr 최후 폴백 (주 실행 흐름 영향 0).
                    try:
                        print(
                            f"CRITICAL: owner_trigger logger failed pr={pr_num} "
                            f"head_prefix={head[:8]} log_exc={log_exc!r}",
                            file=sys.stderr,
                        )
                    except Exception:  # noqa: BLE001 — stderr 폴백 자체도 fail-closed.
                        pass

                # task-2554+1 medium #4: http_post 예외 시 audit FAILED 기록.
                # PENDING sentinel 은 이미 남아있으므로 다음 runner 가 fail-closed.
                try:
                    txn.record(
                        {
                            "task_id": task_id_val,
                            "pr": pr_num,
                            "head": head,
                            "action": ALLOWED_ACTION,
                            "result": RESULT_FAILED,
                            "comment_body": COMMENT_BODY,
                            "endpoint": path,
                            "decision_path": decision_path_str,
                            "token_present": True,
                            "token_hash_prefix": hash_prefix,
                            "token_value_logged": False,
                            "error_code": "HTTP_POST_FAIL",
                        }
                    )
                finally:
                    headers = {}
                    token = ""  # type: ignore[assignment]
                raise exc
            finally:
                headers = {}
                token = ""  # type: ignore[assignment]

            # 5e) audit POSTED append
            try:
                txn.record(
                    {
                        "task_id": task_id_val,
                        "pr": pr_num,
                        "head": head,
                        "action": ALLOWED_ACTION,
                        "result": RESULT_POSTED,
                        "comment_body": COMMENT_BODY,
                        "endpoint": path,
                        "decision_path": decision_path_str,
                        "token_present": True,
                        "token_hash_prefix": hash_prefix,
                        "token_value_logged": False,
                    }
                )
            except (DedupeViolation, AuditRedactionError):
                raise

            return TriggerResult(
                status=RESULT_POSTED,
                pr=pr_num,
                head=head,
                action=ALLOWED_ACTION,
                comment_body=COMMENT_BODY,
                endpoint=path,
                token_present=True,
                token_hash_prefix=hash_prefix,
                error_code=None,
            )

    # ─── merge path 분리 — 본 모듈에서 merge 시도 0 ──────────────────────────

    def merge(self, *args, **kwargs):  # pragma: no cover — 정적 정책 표현
        raise MergePathViolation(
            "owner_trigger_only must not perform merge. "
            "Use merge_queue_executor with BOT_GITHUB_TOKEN."
        )

    def approve(self, *args, **kwargs):  # pragma: no cover
        raise ForbiddenEndpointError("approve forbidden (review submit blocked)")

    def close(self, *args, **kwargs):  # pragma: no cover
        raise ForbiddenEndpointError("close forbidden")

    def reopen(self, *args, **kwargs):  # pragma: no cover
        raise ForbiddenEndpointError("reopen forbidden")


# ─── task-2556 §6: scheduler-initiated invocation 인터페이스 ──────────────────


def invoke_from_scheduler(
    runner: "OwnerTriggerOnly",
    *,
    decision_path: str | Path,
    owner: str,
    repo: str,
    current_head_actual: str,
) -> str:
    """ExecutorScheduler → OwnerTriggerOnly 호출 어댑터 (task-2556 §6).

    회장 §6 명시 1:1: ``owner_trigger_only.trigger_gemini_review()`` 를 scheduler 가
    자동 호출. 본 wrapper 는 runner 결과를 ``merge_queue_executor.orchestrate_owner_trigger_
    for_stale_pr`` 가 기대하는 string 형식 ("POSTED" / "DEDUPED" / "FAILED") 으로 정규화.

    설계 원칙:
      - 본 함수는 token 을 직접 만지지 않는다. ``runner`` 에 이미 ``token_provider`` 가
        주입되어 있어야 한다 (token boundary §11).
      - http_post 예외는 본 함수에서 catch — TokenBoundaryViolation / ForbiddenEndpointError /
        CommentBodyViolation 는 그대로 raise (fail-closed).
      - 일반 RuntimeError 는 "FAILED" 로 정규화 (transient 가능성, FAILED marker 가 caller
        에서 생성됨).

    Args:
      runner: OwnerTriggerOnly 인스턴스 (token_provider/http_post 이미 주입).
      decision_path: owner_trigger_decision.json 경로 (Path 또는 str).
      owner: GitHub owner.
      repo: GitHub repo.
      current_head_actual: 실측 PR head SHA (40-char hex).

    Returns:
      "POSTED" | "DEDUPED" | "FAILED" | "PENDING" — orchestrate_owner_trigger_for_stale_pr
      가 사용하는 enum.

    Raises:
      TokenBoundaryViolation: token 부재.
      ForbiddenEndpointError: 금지 endpoint 호출 시도.
      CommentBodyViolation: comment body 변조.
      MergePathViolation: merge path 침범.
      DecisionInvalidError: decision schema 위반.
    """
    from anu_v2.owner_trigger_decision import DecisionInvalidError as _DecisionInvalidError
    from anu_v2.owner_trigger_audit import DedupeViolation as _DedupeViolation

    if not isinstance(runner, OwnerTriggerOnly):
        raise TypeError("runner must be OwnerTriggerOnly instance")

    try:
        result = runner.trigger_gemini_review(
            decision_path=decision_path,
            owner=owner,
            repo=repo,
            current_head_actual=current_head_actual,
        )
    except (TokenBoundaryViolation, ForbiddenEndpointError,
            CommentBodyViolation, MergePathViolation, _DecisionInvalidError):
        # 보안 / schema 위반은 그대로 propagate — caller fail-closed.
        raise
    except _DedupeViolation:
        # transaction 안에서 dedupe 이미 발생 (다른 process) — DEDUPED 로 reporting.
        return RESULT_DEDUPED
    except Exception:
        # http_post 등 transient 실패 — FAILED 로 정규화 (audit 에는 이미 FAILED 기록됨).
        return RESULT_FAILED

    if result.status not in (RESULT_POSTED, RESULT_DEDUPED, RESULT_FAILED, RESULT_PENDING):
        # 미정의 status — fail-closed default
        return RESULT_FAILED
    return result.status


# ── task-2556 §11 정적 token boundary guard for scheduler 호출 ────────────────


def trigger_for_second_review(
    runner: "OwnerTriggerOnly",
    *,
    pr_number: int,
    head_sha: str,
    owner: str,
    repo: str,
    decision_path: "str | Path",
) -> TriggerResult:
    """second-review 자동 호출 진입점 (task-2565 §3.5 capability 재사용).

    내부적으로 기존 trigger_gemini_review를 호출. dedupe key는 pr_number+head_sha.

    Args:
      runner: OwnerTriggerOnly 인스턴스 (token_provider/http_post 이미 주입).
      pr_number: PR 번호.
      head_sha: 현재 head SHA.
      owner: GitHub owner.
      repo: GitHub repo.
      decision_path: owner_trigger_decision.json 경로.

    Returns:
      TriggerResult (status: POSTED|DEDUPED|FAILED|PENDING).
    """
    if not isinstance(runner, OwnerTriggerOnly):
        raise TypeError("runner must be OwnerTriggerOnly instance")
    return runner.trigger_gemini_review(
        decision_path=decision_path,
        owner=owner,
        repo=repo,
        current_head_actual=head_sha,
    )


def assert_scheduler_token_boundary(env: dict | None) -> None:
    """scheduler 진입 시 token boundary 검증 (task-2556 §11).

    scheduler 가 BOT_GITHUB_TOKEN / GH_TOKEN / OWNER_PAT 등을 들고
    owner_trigger 경로로 진입하려는 시도를 hard-block. 회장 §11 / §16 1:1.

    또한 ``OWNER_GEMINI_TRIGGER_TOKEN`` 이 env 에 명시되어 있지 않으면 fail-closed.
    """
    if env is None:
        raise TokenBoundaryViolation(
            "scheduler env must be provided for token boundary verification"
        )
    for forbidden in FORBIDDEN_TOKEN_ENV_NAMES:
        if forbidden in env:
            raise TokenBoundaryViolation(
                f"scheduler env contains forbidden token name {forbidden!r} — "
                f"owner_trigger path must use {TOKEN_ENV_NAME} only"
            )
    if TOKEN_ENV_NAME not in env:
        raise TokenBoundaryViolation(
            f"scheduler env missing {TOKEN_ENV_NAME} — owner_trigger fail-closed"
        )
    token_value = env.get(TOKEN_ENV_NAME)
    if not isinstance(token_value, str) or not token_value:
        raise TokenBoundaryViolation(
            f"{TOKEN_ENV_NAME} present but empty/non-string — fail-closed"
        )