"""anu_v2.tests.test_owner_trigger_pat_phase3_integration_2553 — Phase 3 통합.

회장 §명시 12 필수 + 11/12번 (task-2553):
  - merge_queue_executor 가 queue-head PR 의 Gemini evidence missing 시 주입된
    owner_trigger 인스턴스를 호출하여 `/gemini review` 댓글 1회 작성.
  - Gemini fresh evidence (commit_id == current_head) 도착 후에는 BOT_GITHUB_TOKEN
    경로 머지로 진행 (OWNER PAT 은 머지에 사용 X — 격리).
  - self-resume (OWNER manual 개입 0).

회귀 박제:
  1. evidence missing + queue-head + owner_trigger 주입 → OWNER_TRIGGER_REQUESTED.
  2. evidence 도착 후 → AUTO_MERGE_ALLOWED (4 gate PASS).
  3. owner_trigger=None → 기존 GEMINI_UNRESOLVED_BLOCK 그대로 (회귀 방지).
  4. queue_predecessors_open > 0 → trigger 호출 0 (queue-head only).
  5. owner_trigger 호출 ↔ BOT_GITHUB_TOKEN 머지 경로 분리 (env 검증).

mock 패턴은 `test_merge_queue_executor_2531.py` + `test_owner_trigger_pat_phase2_2553.py`
와 동일.
"""

from __future__ import annotations

import json
import subprocess
import sys
from pathlib import Path
from typing import Any, Mapping, Sequence

import pytest

# workspace root → sys.path
WORKSPACE_ROOT = Path(__file__).resolve().parents[2]
if str(WORKSPACE_ROOT) not in sys.path:
    sys.path.insert(0, str(WORKSPACE_ROOT))

from anu_v2.merge_queue_executor import (  # noqa: E402
    AUTO_MERGE_ALLOWED,
    AUTO_MERGE_SUCCESS,
    GEMINI_COMPLETED,
    GEMINI_SCOPE_EXPANSION,
    GEMINI_UNRESOLVED,
    GEMINI_UNRESOLVED_BLOCK,
    NON_CRITICAL_AUTO_RESOLVED,
    OWNER_TRIGGER_REQUESTED,
    STALE_EVIDENCE_BLOCK,
    MergeQueueExecutor,
    PRMeta,
)
from anu_v2.owner_trigger_pat import (  # noqa: E402
    ALLOWED_COMMENT_BODY,
    OWNER_PAT_ENV_NAME,
    OwnerTriggerPat,
    write_decision_json,
)


# ─── helpers (test_merge_queue_executor_2531.py 동일 패턴) ───────────────────
def _cp(returncode: int = 0, stdout: str = "", stderr: str = "") -> subprocess.CompletedProcess:
    return subprocess.CompletedProcess(args=[], returncode=returncode, stdout=stdout, stderr=stderr)


_FAKE_OWNER_TOKEN = "github_pat_FAKE_OWNER_PHASE3"
_FAKE_BOT_TOKEN = "ghs_BOT_PHASE3"
_FIXED_TS = "2026-05-11T12:00:00+00:00"


def _make_pr(
    *,
    number: int = 81,
    head_sha: str = "headsha-current",
    head_ref: str = "task/task-2553-dev5",
    base_ref: str = "main",
    changed_files: Sequence[str] = (
        "anu_v2/__init__.py",
        "anu_v2/owner_trigger_pat.py",
        "anu_v2/merge_queue_executor.py",
        "anu_v2/tests/__init__.py",
        "anu_v2/tests/test_owner_trigger_pat_phase3_integration_2553.py",
    ),
    ci_required_all_success: bool = True,
    gemini_status: str = GEMINI_UNRESOLVED,
    merge_state_status: str = "CLEAN",
    queue_predecessors_open: int = 0,
    gemini_commit_id: str = "",
) -> PRMeta:
    return PRMeta(
        number=number,
        head_sha=head_sha,
        head_ref=head_ref,
        base_ref=base_ref,
        changed_files=tuple(changed_files),
        ci_required_all_success=ci_required_all_success,
        gemini_status=gemini_status,
        merge_state_status=merge_state_status,
        queue_predecessors_open=queue_predecessors_open,
        gemini_commit_id=gemini_commit_id,
    )


def _write_task_md(tmp_path: Path) -> Path:
    body = [
        "# task-2553",
        "",
        "```yaml",
        "expected_files:",
        '  - "anu_v2/__init__.py"',
        '  - "anu_v2/owner_trigger_pat.py"',
        '  - "anu_v2/merge_queue_executor.py"',
        '  - "anu_v2/tests/__init__.py"',
        '  - "anu_v2/tests/test_owner_trigger_pat_phase3_integration_2553.py"',
        "cherry_pick_allowed: false",
        "```",
    ]
    p = tmp_path / "task-2553.md"
    p.write_text("\n".join(body), encoding="utf-8")
    return p


# ─── fixtures ───────────────────────────────────────────────────────────────
@pytest.fixture
def gh_owner_calls() -> list[dict[str, Any]]:
    """OWNER PAT 으로 호출된 gh args/env 수집."""
    return []


@pytest.fixture
def gh_bot_calls() -> list[dict[str, Any]]:
    """BOT 토큰 머지 경로 gh args/env 수집."""
    return []


@pytest.fixture
def audit_calls() -> list[dict[str, Any]]:
    return []


@pytest.fixture
def trigger_audit_calls() -> list[dict[str, Any]]:
    return []


@pytest.fixture
def decision_calls() -> list[dict[str, Any]]:
    return []


@pytest.fixture
def owner_trigger(
    gh_owner_calls: list[dict[str, Any]],
    trigger_audit_calls: list[dict[str, Any]],
    decision_calls: list[dict[str, Any]],
) -> OwnerTriggerPat:
    """OWNER PAT trigger 인스턴스 — OWNER 토큰 env 만 읽음."""

    def gh_runner(args: Sequence[str], env: Mapping[str, str]) -> subprocess.CompletedProcess:
        gh_owner_calls.append({"args": list(args), "env": dict(env or {})})
        return _cp()

    def audit_writer(record: Mapping[str, Any]) -> None:
        trigger_audit_calls.append(dict(record))

    def decision_writer(payload: Mapping[str, Any], path: Path) -> None:
        decision_calls.append({"payload": dict(payload), "path": str(path)})
        write_decision_json(payload, path)

    return OwnerTriggerPat(
        gh_runner=gh_runner,
        audit_writer=audit_writer,
        decision_writer=decision_writer,
        owner="jeon-jonghyuk",
        repo="taskctl-anu",
        clock=lambda: _FIXED_TS,
    )


@pytest.fixture
def executor(
    tmp_path: Path,
    gh_bot_calls: list[dict[str, Any]],
    audit_calls: list[dict[str, Any]],
) -> MergeQueueExecutor:
    """머지 큐 executor — BOT 토큰 머지 경로 mock."""

    def gh_runner(args: Sequence[str], env: Mapping[str, str] | None) -> subprocess.CompletedProcess:
        gh_bot_calls.append({"args": list(args), "env": dict(env or {})})
        return _cp()

    def git_runner(*_args: Any) -> subprocess.CompletedProcess:
        return _cp()

    def pytest_runner(_args: Sequence[str]) -> int:
        return 0

    def audit_writer(record: Mapping[str, Any]) -> None:
        audit_calls.append(dict(record))

    return MergeQueueExecutor(
        gh_runner=gh_runner,
        git_runner=git_runner,
        pytest_runner=pytest_runner,
        audit_writer=audit_writer,
        task_md_root=tmp_path,
    )


# ─── 1. evidence missing + queue-head + trigger 주입 → OWNER_TRIGGER_REQUESTED ─
def test_evaluate_with_owner_trigger_missing_evidence_triggers_pat(
    tmp_path: Path,
    executor: MergeQueueExecutor,
    owner_trigger: OwnerTriggerPat,
    gh_owner_calls: list[dict[str, Any]],
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    """evidence missing + queue-head + owner_trigger 주입 → OWNER PAT 으로 `/gemini review`
    댓글 1회 작성 + OWNER_TRIGGER_REQUESTED 반환."""
    _write_task_md(tmp_path)
    monkeypatch.setenv(OWNER_PAT_ENV_NAME, _FAKE_OWNER_TOKEN)
    audit_path = tmp_path / "owner_trigger_audit.jsonl"
    decision_path = tmp_path / "owner_trigger_decision.json"

    pr = _make_pr(
        head_sha="locked-sha",
        gemini_status=GEMINI_UNRESOLVED,
        queue_predecessors_open=0,
    )
    out = executor.evaluate_with_owner_trigger(
        pr=pr,
        head_sha_at_lock="locked-sha",
        owner_trigger=owner_trigger,
        audit_log_path=audit_path,
        decision_json_path=decision_path,
    )
    assert out.decision == OWNER_TRIGGER_REQUESTED
    assert out.reason == "owner_trigger_pat_comment_posted"
    # 회장 §명시: prior_decision 박제 (다음 사이클에서 재검증 단서)
    assert out.extra["prior_decision"] in (
        GEMINI_UNRESOLVED_BLOCK,
        NON_CRITICAL_AUTO_RESOLVED,
    )

    # OWNER PAT 으로 `/gemini review` 댓글 정확히 1회
    assert len(gh_owner_calls) == 1
    call = gh_owner_calls[0]
    assert call["args"] == [
        "api", "-X", "POST",
        f"/repos/jeon-jonghyuk/taskctl-anu/issues/{pr.number}/comments",
        "-f", f"body={ALLOWED_COMMENT_BODY}",
    ]
    assert call["env"]["GH_TOKEN"] == _FAKE_OWNER_TOKEN

    # decision JSON 박제됨 (회장 §명시 audit trail)
    assert decision_path.exists()
    decision_data = json.loads(decision_path.read_text(encoding="utf-8"))
    assert decision_data["decision"] == "PASS"
    assert decision_data["pr_number"] == pr.number


# ─── 2. evidence 도착 후 → AUTO_MERGE_ALLOWED (4 gate PASS) ──────────────────
def test_evaluate_with_owner_trigger_evidence_present_proceeds_to_merge(
    tmp_path: Path,
    executor: MergeQueueExecutor,
    owner_trigger: OwnerTriggerPat,
    gh_owner_calls: list[dict[str, Any]],
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    """Gemini evidence 도착 (GEMINI_COMPLETED) 후 → 4 gate 모두 PASS → AUTO_MERGE_ALLOWED.
    OWNER trigger 는 호출되지 않아야 함 (이미 evidence 있음)."""
    _write_task_md(tmp_path)
    monkeypatch.setenv(OWNER_PAT_ENV_NAME, _FAKE_OWNER_TOKEN)
    audit_path = tmp_path / "audit.jsonl"
    decision_path = tmp_path / "decision.json"

    pr = _make_pr(
        head_sha="locked-sha",
        gemini_status=GEMINI_COMPLETED,
        queue_predecessors_open=0,
    )
    out = executor.evaluate_with_owner_trigger(
        pr=pr,
        head_sha_at_lock="locked-sha",
        owner_trigger=owner_trigger,
        audit_log_path=audit_path,
        decision_json_path=decision_path,
    )
    assert out.decision == AUTO_MERGE_ALLOWED
    assert out.passed
    # OWNER trigger 호출 0 (evidence 이미 도착)
    assert len(gh_owner_calls) == 0
    # decision JSON 도 박제 안 됨 (trigger 진입 자체가 X)
    assert not decision_path.exists()


# ─── 3. owner_trigger=None → 기존 GEMINI_UNRESOLVED_BLOCK 그대로 (회귀 방지) ─
def test_evaluate_with_owner_trigger_no_injection_falls_back_to_unresolved(
    tmp_path: Path,
    executor: MergeQueueExecutor,
    gh_owner_calls: list[dict[str, Any]],
) -> None:
    """owner_trigger 미주입 → 기존 evaluate() 동작 유지 (회귀 방지).

    GEMINI_UNRESOLVED_BLOCK 은 비critical 이므로 auto_handle_non_critical 에서
    NON_CRITICAL_AUTO_RESOLVED 로 변환되어 반환된다. 핵심은 OWNER_TRIGGER_REQUESTED
    가 절대 나오면 안 된다는 점.
    """
    _write_task_md(tmp_path)
    audit_path = tmp_path / "audit.jsonl"
    decision_path = tmp_path / "decision.json"

    pr = _make_pr(
        head_sha="locked-sha",
        gemini_status=GEMINI_UNRESOLVED,
        queue_predecessors_open=0,
    )
    out = executor.evaluate_with_owner_trigger(
        pr=pr,
        head_sha_at_lock="locked-sha",
        owner_trigger=None,  # 의존성 미주입
        audit_log_path=audit_path,
        decision_json_path=decision_path,
    )
    # OWNER trigger 신호 없음
    assert out.decision != OWNER_TRIGGER_REQUESTED
    # 기존 evaluate() 결과 그대로 (비critical 자동 처리 경로)
    assert out.decision == NON_CRITICAL_AUTO_RESOLVED
    # OWNER PAT gh 호출 0 (owner_trigger 없으므로 당연)
    assert len(gh_owner_calls) == 0
    # decision JSON 박제 X
    assert not decision_path.exists()


# ─── 4. non queue-head → trigger 호출 0 (회장 §명시 queue-head only) ─────────
def test_evaluate_with_owner_trigger_non_queue_head_skips_trigger(
    tmp_path: Path,
    executor: MergeQueueExecutor,
    owner_trigger: OwnerTriggerPat,
    gh_owner_calls: list[dict[str, Any]],
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    """queue_predecessors_open > 0 → trigger 호출 0. mock counter 로 검증."""
    _write_task_md(tmp_path)
    monkeypatch.setenv(OWNER_PAT_ENV_NAME, _FAKE_OWNER_TOKEN)
    audit_path = tmp_path / "audit.jsonl"
    decision_path = tmp_path / "decision.json"

    pr = _make_pr(
        head_sha="locked-sha",
        gemini_status=GEMINI_UNRESOLVED,
        queue_predecessors_open=2,  # not queue-head
    )
    out = executor.evaluate_with_owner_trigger(
        pr=pr,
        head_sha_at_lock="locked-sha",
        owner_trigger=owner_trigger,
        audit_log_path=audit_path,
        decision_json_path=decision_path,
    )
    # OWNER_TRIGGER_REQUESTED 절대 안 나옴
    assert out.decision != OWNER_TRIGGER_REQUESTED
    # OWNER PAT gh 호출 0 (queue-head 가 아니므로 trigger 진입 자체 X)
    assert len(gh_owner_calls) == 0
    # decision JSON 박제 X
    assert not decision_path.exists()


# ─── 5. OWNER PAT 호출이 BOT_GITHUB_TOKEN 머지 경로와 격리되는지 ─────────────
def test_evaluate_with_owner_trigger_owner_pat_isolated_from_bot_merge(
    tmp_path: Path,
    executor: MergeQueueExecutor,
    owner_trigger: OwnerTriggerPat,
    gh_owner_calls: list[dict[str, Any]],
    gh_bot_calls: list[dict[str, Any]],
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    """OWNER PAT trigger 호출 시 OWNER 토큰만 사용되고, BOT 토큰 머지 경로는 분리된 채
    BOT_GITHUB_TOKEN 으로 호출된다 (env mock 으로 검증).

    회장 §명시 — 두 토큰은 서로 다른 경로/env 로 격리:
      - OWNER PAT (`OWNER_GEMINI_TRIGGER_PAT`) → `/gemini review` comment only.
      - BOT_GITHUB_TOKEN → squash merge.
    """
    _write_task_md(tmp_path)
    # 양쪽 토큰 모두 환경에 존재 — 격리가 깨지지 않는지 검증
    monkeypatch.setenv(OWNER_PAT_ENV_NAME, _FAKE_OWNER_TOKEN)
    monkeypatch.setenv("BOT_GITHUB_TOKEN", _FAKE_BOT_TOKEN)
    monkeypatch.setenv("PATH", "/usr/bin")

    audit_path = tmp_path / "audit.jsonl"
    decision_path = tmp_path / "decision.json"

    # 1단계 — evidence missing 시 OWNER PAT trigger
    pr_missing = _make_pr(
        head_sha="locked-sha",
        gemini_status=GEMINI_UNRESOLVED,
        queue_predecessors_open=0,
    )
    out1 = executor.evaluate_with_owner_trigger(
        pr=pr_missing,
        head_sha_at_lock="locked-sha",
        owner_trigger=owner_trigger,
        audit_log_path=audit_path,
        decision_json_path=decision_path,
    )
    assert out1.decision == OWNER_TRIGGER_REQUESTED
    # OWNER 측 호출은 OWNER 토큰만
    assert len(gh_owner_calls) == 1
    owner_env = gh_owner_calls[0]["env"]
    assert owner_env["GH_TOKEN"] == _FAKE_OWNER_TOKEN
    assert owner_env["GITHUB_TOKEN"] == _FAKE_OWNER_TOKEN
    # OWNER 호출 env 에 BOT 토큰이 누설되지 않음
    assert _FAKE_BOT_TOKEN not in owner_env.get("GH_TOKEN", "")
    assert _FAKE_BOT_TOKEN not in owner_env.get("GITHUB_TOKEN", "")
    # BOT 머지 경로는 아직 호출되지 않음 (evidence 미도착 단계)
    assert len(gh_bot_calls) == 0

    # 2단계 — evidence 도착 후 BOT 토큰 머지 실행
    pr_evidence = _make_pr(
        head_sha="locked-sha",
        gemini_status=GEMINI_COMPLETED,
        queue_predecessors_open=0,
    )
    merge_out = executor.execute_bot_squash_merge(pr_evidence, head_sha_at_lock="locked-sha")
    assert merge_out.decision == AUTO_MERGE_SUCCESS
    # BOT 머지 측 호출은 BOT 토큰만
    assert len(gh_bot_calls) == 1
    bot_env = gh_bot_calls[0]["env"]
    assert bot_env["GH_TOKEN"] == _FAKE_BOT_TOKEN
    assert bot_env["GITHUB_TOKEN"] == _FAKE_BOT_TOKEN
    # BOT 머지 env 에 OWNER PAT 이 누설되지 않음 (BOT_GITHUB_TOKEN 으로 덮어쓰기 검증)
    assert _FAKE_OWNER_TOKEN not in bot_env.get("GH_TOKEN", "")
    assert _FAKE_OWNER_TOKEN not in bot_env.get("GITHUB_TOKEN", "")
    # OWNER 호출 카운터는 그대로 (BOT 호출이 OWNER 호출을 추가로 발생시키지 않음)
    assert len(gh_owner_calls) == 1


# ─── 보너스: trigger 실패 (rejected) 시 evaluate 결과 그대로 ─────────────────
def test_evaluate_with_owner_trigger_rejected_falls_back_to_outcome(
    tmp_path: Path,
    executor: MergeQueueExecutor,
    owner_trigger: OwnerTriggerPat,
    gh_owner_calls: list[dict[str, Any]],
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    """OWNER PAT env 미설정 → trigger 가 REJECT → 기존 evaluate 결과 그대로 (재시도 대기).

    OWNER_TRIGGER_REQUESTED 는 trigger outcome == "ok" 일 때만 반환된다.
    """
    _write_task_md(tmp_path)
    # OWNER PAT env 명시적으로 제거 → trigger 단계에서 REJECT 발생
    monkeypatch.delenv(OWNER_PAT_ENV_NAME, raising=False)
    audit_path = tmp_path / "audit.jsonl"
    decision_path = tmp_path / "decision.json"

    pr = _make_pr(
        head_sha="locked-sha",
        gemini_status=GEMINI_UNRESOLVED,
        queue_predecessors_open=0,
    )
    out = executor.evaluate_with_owner_trigger(
        pr=pr,
        head_sha_at_lock="locked-sha",
        owner_trigger=owner_trigger,
        audit_log_path=audit_path,
        decision_json_path=decision_path,
    )
    # OWNER_TRIGGER_REQUESTED 절대 안 나옴 (trigger 가 REJECT 했으므로)
    assert out.decision != OWNER_TRIGGER_REQUESTED
    # gh 호출은 0 (REJECT 가 gh 호출 전에 발생)
    assert len(gh_owner_calls) == 0


# ─── G1 Critical 1: stale evidence (commit_id mismatch) → STALE_EVIDENCE_BLOCK ─
def test_stale_evidence_blocks_merge(
    tmp_path: Path,
    executor: MergeQueueExecutor,
    owner_trigger: OwnerTriggerPat,
    gh_owner_calls: list[dict[str, Any]],
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    """회장 §명시 11번 — Gemini evidence commit_id 가 현재 head_sha 와 다르면 stale.

    gemini_status == GEMINI_COMPLETED 라도 gemini_commit_id != head_sha 면
    STALE_EVIDENCE_BLOCK (비critical, 다음 사이클 재검증 가능). 호출부가 BOT 머지
    경로로 잘못 진입하지 않게 4 gate 단계에서 차단된다.
    """
    _write_task_md(tmp_path)
    monkeypatch.setenv(OWNER_PAT_ENV_NAME, _FAKE_OWNER_TOKEN)
    audit_path = tmp_path / "audit.jsonl"
    decision_path = tmp_path / "decision.json"

    pr = _make_pr(
        head_sha="newhead-after-update-branch",
        gemini_status=GEMINI_COMPLETED,
        gemini_commit_id="oldhead-before-update-branch",   # stale evidence
        queue_predecessors_open=0,
    )
    out = executor.check_ci_gemini_clean_sha_lock(
        pr,
        head_sha_at_lock="newhead-after-update-branch",
    )
    assert out.decision == STALE_EVIDENCE_BLOCK
    assert out.reason == "gemini_evidence_stale_commit_id_mismatch"
    assert out.extra["evidence_commit_id"] == "oldhead-before-update-branch"
    assert out.extra["current_head"] == "newhead-after-update-branch"
    # 비critical — auto_handle 경로로 처리되어야 함
    assert not out.is_critical


def test_stale_evidence_via_evaluate_returns_non_critical(
    tmp_path: Path,
    executor: MergeQueueExecutor,
    owner_trigger: OwnerTriggerPat,
    gh_owner_calls: list[dict[str, Any]],
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    """evaluate() 전체 파이프라인에서도 stale evidence 가 안전하게 차단되고 OWNER
    trigger 가 호출되지 않는지 검증."""
    _write_task_md(tmp_path)
    monkeypatch.setenv(OWNER_PAT_ENV_NAME, _FAKE_OWNER_TOKEN)
    audit_path = tmp_path / "audit.jsonl"
    decision_path = tmp_path / "decision.json"

    pr = _make_pr(
        head_sha="newhead",
        gemini_status=GEMINI_COMPLETED,
        gemini_commit_id="oldhead",   # stale
        queue_predecessors_open=0,
    )
    out = executor.evaluate_with_owner_trigger(
        pr=pr,
        head_sha_at_lock="newhead",
        owner_trigger=owner_trigger,
        audit_log_path=audit_path,
        decision_json_path=decision_path,
    )
    # 4 gate 단계에서 차단되었으므로 AUTO_MERGE_ALLOWED 아님
    assert out.decision != AUTO_MERGE_ALLOWED
    # GEMINI_COMPLETED 이므로 trigger 분기 진입 X (Gemini complete short-circuit)
    assert len(gh_owner_calls) == 0


def test_gemini_commit_id_match_passes_gate(
    tmp_path: Path,
    executor: MergeQueueExecutor,
) -> None:
    """gemini_commit_id == head_sha 면 정상 통과 (회귀 방지)."""
    pr = _make_pr(
        head_sha="samehead-001",
        gemini_status=GEMINI_COMPLETED,
        gemini_commit_id="samehead-001",
        queue_predecessors_open=0,
    )
    out = executor.check_ci_gemini_clean_sha_lock(
        pr,
        head_sha_at_lock="samehead-001",
    )
    assert out.decision == AUTO_MERGE_ALLOWED
    assert out.reason == "all_gates_clean"


def test_gemini_commit_id_empty_skips_stale_check(
    tmp_path: Path,
    executor: MergeQueueExecutor,
) -> None:
    """gemini_commit_id 가 빈 문자열 (정보 미제공) 이면 stale 검사 skip — 하위 호환."""
    pr = _make_pr(
        head_sha="anyhead",
        gemini_status=GEMINI_COMPLETED,
        gemini_commit_id="",   # 기본값
        queue_predecessors_open=0,
    )
    out = executor.check_ci_gemini_clean_sha_lock(
        pr,
        head_sha_at_lock="anyhead",
    )
    assert out.decision == AUTO_MERGE_ALLOWED


# ─── G1 High 2: HEAD SHA lock 검증 순서 (trigger 호출 전) ─────────────────────
def test_head_sha_changed_blocks_trigger(
    tmp_path: Path,
    executor: MergeQueueExecutor,
    owner_trigger: OwnerTriggerPat,
    gh_owner_calls: list[dict[str, Any]],
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    """회장 §명시 11번 — current head 미확정 (`pr.head_sha != head_sha_at_lock`) 상태에서는
    OWNER PAT trigger 발사 금지. update-branch 직후 새 head 가 생긴 경우 다음 사이클
    재검증으로 미룬다.
    """
    _write_task_md(tmp_path)
    monkeypatch.setenv(OWNER_PAT_ENV_NAME, _FAKE_OWNER_TOKEN)
    audit_path = tmp_path / "audit.jsonl"
    decision_path = tmp_path / "decision.json"

    pr = _make_pr(
        head_sha="newhead-after-rebase",
        gemini_status=GEMINI_UNRESOLVED,
        queue_predecessors_open=0,
    )
    out = executor.evaluate_with_owner_trigger(
        pr=pr,
        head_sha_at_lock="oldhead-locked",   # lock 시점 SHA 와 현재 head 가 다름
        owner_trigger=owner_trigger,
        audit_log_path=audit_path,
        decision_json_path=decision_path,
    )
    # OWNER_TRIGGER_REQUESTED 절대 안 나옴 (current head 미확정)
    assert out.decision != OWNER_TRIGGER_REQUESTED
    # gh 호출 0 — trigger 발사 차단
    assert len(gh_owner_calls) == 0
    # decision JSON 박제 X (trigger 진입 전에 차단)
    assert not decision_path.exists()


def test_head_sha_match_still_triggers(
    tmp_path: Path,
    executor: MergeQueueExecutor,
    owner_trigger: OwnerTriggerPat,
    gh_owner_calls: list[dict[str, Any]],
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    """head_sha == head_sha_at_lock 이면 정상 trigger 진행 (회귀 방지)."""
    _write_task_md(tmp_path)
    monkeypatch.setenv(OWNER_PAT_ENV_NAME, _FAKE_OWNER_TOKEN)
    audit_path = tmp_path / "audit.jsonl"
    decision_path = tmp_path / "decision.json"

    pr = _make_pr(
        head_sha="stable-head",
        gemini_status=GEMINI_UNRESOLVED,
        queue_predecessors_open=0,
    )
    out = executor.evaluate_with_owner_trigger(
        pr=pr,
        head_sha_at_lock="stable-head",
        owner_trigger=owner_trigger,
        audit_log_path=audit_path,
        decision_json_path=decision_path,
    )
    assert out.decision == OWNER_TRIGGER_REQUESTED
    assert len(gh_owner_calls) == 1


# ─── 회장 Codex G1 round 2 fix 검증 ──────────────────────────────────────────
def test_g1_r2_ci_failure_does_not_trigger(
    tmp_path: Path,
    executor: MergeQueueExecutor,
    owner_trigger: OwnerTriggerPat,
    gh_owner_calls: list[dict[str, Any]],
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    """회장 Codex G1 round 2 Critical — CI 실패 상황에서 OWNER trigger 발사 0.

    `ci_required_all_success=False` → `evaluate()` 가 CI_FAILURE_BLOCK 반환 →
    `evaluate_with_owner_trigger()` 가 `_gemini_unresolved_allowed=False` 로
    판정 → trigger 호출 0. 회장 §명시 "evidence missing 일 때만" 강제.
    """
    _write_task_md(tmp_path)
    monkeypatch.setenv(OWNER_PAT_ENV_NAME, _FAKE_OWNER_TOKEN)
    audit_path = tmp_path / "audit.jsonl"
    decision_path = tmp_path / "decision.json"

    pr = _make_pr(
        head_sha="headsha-ci-fail",
        ci_required_all_success=False,
        gemini_status=GEMINI_UNRESOLVED,
        queue_predecessors_open=0,
    )
    out = executor.evaluate_with_owner_trigger(
        pr=pr,
        head_sha_at_lock="headsha-ci-fail",
        owner_trigger=owner_trigger,
        audit_log_path=audit_path,
        decision_json_path=decision_path,
    )
    # CI 실패 → CI_FAILURE_BLOCK (비critical) → auto_handle_non_critical →
    # NON_CRITICAL_AUTO_RESOLVED. 어떤 경우든 OWNER_TRIGGER_REQUESTED 아님.
    assert out.decision != OWNER_TRIGGER_REQUESTED
    # OWNER PAT 으로 댓글 발사 0 — 회장 §명시 보안 경계 강제
    assert len(gh_owner_calls) == 0


def test_g1_r2_gemini_scope_expansion_does_not_trigger(
    tmp_path: Path,
    executor: MergeQueueExecutor,
    owner_trigger: OwnerTriggerPat,
    gh_owner_calls: list[dict[str, Any]],
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    """SCOPE_EXPANSION (critical) 상황에서도 OWNER trigger 발사 0.

    회장 §명시: critical 차단 상황은 `/gemini review` 트리거 대상 X.
    """
    _write_task_md(tmp_path)
    monkeypatch.setenv(OWNER_PAT_ENV_NAME, _FAKE_OWNER_TOKEN)
    audit_path = tmp_path / "audit.jsonl"
    decision_path = tmp_path / "decision.json"

    pr = _make_pr(
        head_sha="headsha-scope",
        gemini_status=GEMINI_SCOPE_EXPANSION,
        queue_predecessors_open=0,
    )
    out = executor.evaluate_with_owner_trigger(
        pr=pr,
        head_sha_at_lock="headsha-scope",
        owner_trigger=owner_trigger,
        audit_log_path=audit_path,
        decision_json_path=decision_path,
    )
    # SCOPE_EXPANSION → BLOCKED_WITH_REASON (critical) — trigger 발사 X
    assert out.decision != OWNER_TRIGGER_REQUESTED
    assert len(gh_owner_calls) == 0


# NOTE: DIFF_CONTAMINATION + gemini=UNRESOLVED 조합은 evaluate() gate 순서상
# 자연스레 unreachable (gemini check 가 diff check 보다 먼저 fail) — 본 시나리오는
# evaluate() 가 GEMINI_UNRESOLVED_BLOCK 을 먼저 반환하여 trigger 가 발사되더라도
# 다음 사이클에 diff gate 가 차단한다. self-correcting 동작.


def test_g1_r2_token_env_override_rejected_at_constructor(
    tmp_path: Path,
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    """회장 Codex G1 round 2 High — token_env override 화이트리스트 강제.

    BOT_GITHUB_TOKEN 등 OWNER_GEMINI_TRIGGER_PAT 외 env 이름을 token_env 로
    넘기면 `__init__` 단계에서 즉시 ValueError. doctrine 위반 시나리오 API 차단.
    """
    import subprocess

    def gh_runner(args: Sequence[str], env: Mapping[str, str]) -> subprocess.CompletedProcess:
        return _cp()

    def audit_writer(record: Mapping[str, Any]) -> None:
        return

    def decision_writer(payload: Mapping[str, Any], path: Path) -> None:
        return

    with pytest.raises(ValueError, match=OWNER_PAT_ENV_NAME):
        OwnerTriggerPat(
            gh_runner=gh_runner,
            audit_writer=audit_writer,
            decision_writer=decision_writer,
            owner="jeon-jonghyuk",
            repo="taskctl-anu",
            token_env="BOT_GITHUB_TOKEN",
        )


def test_g1_r2_marker_cleanup_on_gh_failure_allows_retry(
    tmp_path: Path,
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    """회장 Codex G1 round 2 High — gh 실패 후 marker 정리 → 재시도 가능.

    gh runner 가 rc!=0 으로 실패하면 marker 가 자동 정리되어, 같은 PR/head 에
    대한 다음 사이클 trigger 가 dedupe 차단 없이 진행 가능해야 한다.
    self-resume 요구 박제.
    """
    import subprocess

    monkeypatch.setenv(OWNER_PAT_ENV_NAME, _FAKE_OWNER_TOKEN)
    audit_path = tmp_path / "audit.jsonl"
    decision_path = tmp_path / "decision.json"

    call_count = {"n": 0}
    gh_outputs: list[subprocess.CompletedProcess] = []

    def gh_runner(args: Sequence[str], env: Mapping[str, str]) -> subprocess.CompletedProcess:
        call_count["n"] += 1
        # 첫 호출: rc=1 (실패), 두 번째: rc=0 (성공)
        if call_count["n"] == 1:
            out = subprocess.CompletedProcess(args=[], returncode=1, stdout="", stderr="temporary")
        else:
            out = subprocess.CompletedProcess(args=[], returncode=0, stdout="ok", stderr="")
        gh_outputs.append(out)
        return out

    audit_calls: list[dict[str, Any]] = []

    def audit_writer(record: Mapping[str, Any]) -> None:
        audit_calls.append(dict(record))

    def decision_writer(payload: Mapping[str, Any], path: Path) -> None:
        write_decision_json(payload, path)

    trigger = OwnerTriggerPat(
        gh_runner=gh_runner,
        audit_writer=audit_writer,
        decision_writer=decision_writer,
        owner="jeon-jonghyuk",
        repo="taskctl-anu",
        clock=lambda: _FIXED_TS,
    )

    # 1차 trigger: gh 실패 → marker 제거 → outcome="failed"
    out1 = trigger.trigger_gemini_review(
        pr_number=99,
        head_sha="retry-sha",
        queue_position=0,
        gemini_evidence_state="missing_for_current_head",
        audit_log_path=audit_path,
        decision_json_path=decision_path,
    )
    assert out1.outcome == "failed"

    # 2차 trigger (같은 PR/head) — dedupe 차단 없이 재시도 가능
    out2 = trigger.trigger_gemini_review(
        pr_number=99,
        head_sha="retry-sha",
        queue_position=0,
        gemini_evidence_state="missing_for_current_head",
        audit_log_path=audit_path,
        decision_json_path=decision_path,
    )
    assert out2.outcome == "ok"
    assert call_count["n"] == 2  # gh runner 2회 호출 (재시도 성공)