# -*- coding: utf-8 -*-
"""tests.regression.test_owner_gemini_trigger_router — task-2641 Track A.

회장 verbatim 12 (2026-05-23) 1:1 정합 — router state machine + return enum 회귀.

본 회귀는 Layer A / NO-CRON: subprocess / cokacdir / merge / cron / live gh 호출 0.
freshness_checker / invoke_scheduler / github_api / permission diagnostics provider
모두 callable inject (mock).
"""
from __future__ import annotations

from typing import Any

import pytest

from anu_v2.gemini_evidence_freshness_checker import (
    FreshnessResult,
    RESULT_FRESH,
    RESULT_STALE,
)
from anu_v2.owner_gemini_trigger_router import (
    DEFAULT_FRESH_REVIEW_TIMEOUT_S,
    NUDGE_HARD_LIMIT_PER_PR_HEAD,
    OwnerGeminiTriggerRouter,
    ROUTER_RESULT_CHAIR_UI_FALLBACK_REQUIRED,
    ROUTER_RESULT_FRESH,
    ROUTER_RESULT_GEMINI_EXTERNAL_TRIGGER_STALE,
    ROUTER_RESULT_NOT_GEMINI_TRIGGER,
    ROUTER_RESULT_NUDGE_DEDUPED,
    ROUTER_RESULT_NUDGE_PERMISSION_DENIED,
    ROUTER_RESULT_NUDGE_POSTED,
    RouterContractError,
)
from anu_v2.owner_gemini_trigger_router_audit import (
    AUDIT_SCHEMA,
    OwnerGeminiTriggerRouterAudit,
)
from anu_v2.owner_trigger_audit import (
    RESULT_DEDUPED,
    RESULT_FAILED,
    RESULT_POSTED,
)


HEAD_A = "a" * 40
HEAD_B = "b" * 40
HEAD_C = "c" * 40


# ─── helpers ──────────────────────────────────────────────────────────────────


def _make_freshness(
    *,
    status: str,
    pr_number: int = 143,
    head: str = HEAD_A,
    gemini_commit_id: str | None = None,
    inspected: int = 1,
    reason: str = "test",
) -> FreshnessResult:
    return FreshnessResult(
        status=status,
        pr_number=pr_number,
        current_head_sha=head,
        gemini_commit_id_observed=gemini_commit_id,
        reviews_inspected=inspected,
        reason=reason,
    )


def _stub_freshness(result: FreshnessResult):
    def _checker(**_kwargs):
        return result

    return _checker


def _stub_freshness_queue(*results: FreshnessResult):
    """multi-call freshness — 첫 호출과 post-nudge 재호출 분리."""
    state = {"i": 0}

    def _checker(**_kwargs):
        idx = state["i"]
        state["i"] = idx + 1
        return results[min(idx, len(results) - 1)]

    return _checker


def _stub_invoke(status: str):
    captured = []

    def _invoke(**kwargs):
        captured.append(kwargs)
        return status

    _invoke.captured = captured  # type: ignore[attr-defined]
    return _invoke


def _stub_invoke_raises(exc: Exception):
    def _invoke(**_kwargs):
        raise exc

    return _invoke


def _no_diag():
    return None


def _stub_diag(payload: dict | None):
    def _provider():
        return payload

    return _provider


def _dummy_github_api(method: str, path: str) -> Any:  # noqa: ARG001 — unused
    return []


def _make_router(
    tmp_path,
    *,
    freshness_checker,
    invoke_scheduler,
    permission_diagnostics_provider=None,
    token_provider=None,
    github_api=_dummy_github_api,
) -> OwnerGeminiTriggerRouter:
    audit = OwnerGeminiTriggerRouterAudit(tmp_path)
    return OwnerGeminiTriggerRouter(
        workspace_root=tmp_path,
        freshness_checker=freshness_checker,
        invoke_scheduler=invoke_scheduler,
        permission_diagnostics_provider=(
            permission_diagnostics_provider or _no_diag
        ),
        token_provider=token_provider,
        audit=audit,
        github_api=github_api,
    )


# ─── 1. PR Review oversight (회장 verbatim §1/§2/§12 — task-2640 사고 박제) ───


def test_pr_review_empty_body_blocks_misroute(tmp_path):
    router = _make_router(
        tmp_path,
        freshness_checker=_stub_freshness(_make_freshness(status=RESULT_FRESH)),
        invoke_scheduler=_stub_invoke(RESULT_POSTED),
    )
    observed = {
        "id": 4350214991,
        "kind": "review",
        "state": "COMMENTED",
        "body": "",
        "user": {"login": "Jeon-Jonghyuk"},
    }
    result = router.route_for_pr(
        pr_number=143,
        current_head_sha=HEAD_A,
        owner="o",
        repo="r",
        observed_comment=observed,
        decision_path=None,
    )
    assert result.final_state == ROUTER_RESULT_NOT_GEMINI_TRIGGER
    assert result.nudge_attempted is False
    assert "PR Review" in result.reason
    assert "task-2640" in result.reason


def test_issue_comment_exact_body_does_not_short_circuit(tmp_path):
    """issue_comment "/gemini review" 는 short-circuit 0 → freshness 평가 진행."""
    router = _make_router(
        tmp_path,
        freshness_checker=_stub_freshness(
            _make_freshness(status=RESULT_FRESH, gemini_commit_id=HEAD_A)
        ),
        invoke_scheduler=_stub_invoke(RESULT_POSTED),
    )
    observed = {
        "kind": "issue_comment",
        "body": "/gemini review",
        "user": {"login": "Jeon-Jonghyuk"},
    }
    result = router.route_for_pr(
        pr_number=143,
        current_head_sha=HEAD_A,
        owner="o",
        repo="r",
        observed_comment=observed,
        decision_path=None,
    )
    assert result.final_state == ROUTER_RESULT_FRESH
    assert result.nudge_attempted is False


# ─── 2. FRESH 통과 (spec §3.2-1) ──────────────────────────────────────────────


def test_freshness_fresh_passes_without_nudge(tmp_path):
    router = _make_router(
        tmp_path,
        freshness_checker=_stub_freshness(
            _make_freshness(status=RESULT_FRESH, gemini_commit_id=HEAD_A)
        ),
        invoke_scheduler=_stub_invoke(RESULT_POSTED),
    )
    result = router.route_for_pr(
        pr_number=143,
        current_head_sha=HEAD_A,
        owner="o",
        repo="r",
        decision_path=None,
    )
    assert result.final_state == ROUTER_RESULT_FRESH
    assert result.nudge_attempted is False
    assert result.gemini_commit_id_observed == HEAD_A


# ─── 3. STALE → POSTED → fresh review 도착 (spec §3.2 POSTED-FRESH) ───────────


def test_stale_then_posted_then_fresh_review_arrives(tmp_path):
    invoke = _stub_invoke(RESULT_POSTED)
    router = _make_router(
        tmp_path,
        freshness_checker=_stub_freshness_queue(
            _make_freshness(status=RESULT_STALE, gemini_commit_id=HEAD_B),
            _make_freshness(status=RESULT_FRESH, gemini_commit_id=HEAD_A),
        ),
        invoke_scheduler=invoke,
    )
    result = router.route_for_pr(
        pr_number=143,
        current_head_sha=HEAD_A,
        owner="o",
        repo="r",
        decision_path="memory/decisions/d.json",
        fresh_review_arrived_post_nudge=True,
    )
    assert result.final_state == ROUTER_RESULT_FRESH
    assert result.nudge_attempted is True
    assert result.nudge_result == RESULT_POSTED
    # caller invoked with normalized lowercase head
    assert len(invoke.captured) == 1  # type: ignore[attr-defined]
    assert invoke.captured[0]["current_head_actual"] == HEAD_A  # type: ignore[attr-defined]


# ─── 4. STALE → POSTED → fresh review 미도착 (spec §3.2 POSTED-STALE) ─────────


def test_stale_then_posted_no_fresh_review_classify_stale(tmp_path):
    router = _make_router(
        tmp_path,
        freshness_checker=_stub_freshness_queue(
            _make_freshness(status=RESULT_STALE, gemini_commit_id=HEAD_B),
            _make_freshness(status=RESULT_STALE, gemini_commit_id=HEAD_B),
        ),
        invoke_scheduler=_stub_invoke(RESULT_POSTED),
    )
    result = router.route_for_pr(
        pr_number=143,
        current_head_sha=HEAD_A,
        owner="o",
        repo="r",
        decision_path="memory/decisions/d.json",
        fresh_review_arrived_post_nudge=True,
    )
    assert result.final_state == ROUTER_RESULT_GEMINI_EXTERNAL_TRIGGER_STALE
    assert result.nudge_attempted is True
    assert result.nudge_result == RESULT_POSTED


def test_stale_then_posted_without_post_nudge_signal_returns_nudge_posted(tmp_path):
    router = _make_router(
        tmp_path,
        freshness_checker=_stub_freshness(
            _make_freshness(status=RESULT_STALE, gemini_commit_id=HEAD_B)
        ),
        invoke_scheduler=_stub_invoke(RESULT_POSTED),
    )
    result = router.route_for_pr(
        pr_number=143,
        current_head_sha=HEAD_A,
        owner="o",
        repo="r",
        decision_path="memory/decisions/d.json",
        fresh_review_arrived_post_nudge=False,
    )
    assert result.final_state == ROUTER_RESULT_NUDGE_POSTED
    assert result.nudge_attempted is True


# ─── 5. nudge limit dedupe (회장 verbatim §9) ─────────────────────────────────


def test_nudge_limit_dedupe_blocks_second_call(tmp_path):
    """audit 에 prior nudge_attempted=true 가 1건 있으면 2회째는 DEDUPED."""
    audit = OwnerGeminiTriggerRouterAudit(tmp_path)
    # prior record
    audit.append(
        {
            "task_id": "prior",
            "pr_number": 143,
            "current_head_sha": HEAD_A,
            "freshness_state": "STALE",
            "gemini_commit_id_observed": HEAD_B,
            "nudge_attempted": True,
            "nudge_result": RESULT_POSTED,
            "permission_header_diagnostics": None,
            "token_present": True,
            "token_hash_prefix": "0123456789ab",
            "token_value_logged": False,
            "final_state": "NUDGE_POSTED",
            "reason": "prior",
        }
    )
    invoke = _stub_invoke(RESULT_POSTED)
    router = OwnerGeminiTriggerRouter(
        workspace_root=tmp_path,
        freshness_checker=_stub_freshness(
            _make_freshness(status=RESULT_STALE, gemini_commit_id=HEAD_B)
        ),
        invoke_scheduler=invoke,
        permission_diagnostics_provider=_no_diag,
        audit=audit,
        github_api=_dummy_github_api,
    )
    result = router.route_for_pr(
        pr_number=143,
        current_head_sha=HEAD_A,
        owner="o",
        repo="r",
        decision_path="memory/decisions/d.json",
    )
    assert result.final_state == ROUTER_RESULT_NUDGE_DEDUPED
    assert result.nudge_attempted is False
    assert result.nudge_result == RESULT_DEDUPED
    # invoke_scheduler MUST NOT have been called
    assert invoke.captured == []  # type: ignore[attr-defined]


def test_nudge_hard_limit_default_is_one():
    assert NUDGE_HARD_LIMIT_PER_PR_HEAD == 1


def test_default_fresh_review_timeout_s_positive():
    assert DEFAULT_FRESH_REVIEW_TIMEOUT_S > 0


# ─── 6. 403 permission diagnostics → NUDGE_PERMISSION_DENIED ─────────────────


def test_failed_invoke_with_403_headers_classifies_permission_denied(tmp_path):
    headers_403 = {
        "X-Accepted-GitHub-Permissions": "issues=write; pull_requests=write",
        "X-RateLimit-Remaining": "4321",
        "Authorization": "Bearer ghp_LEAK_NEVER",
    }
    router = _make_router(
        tmp_path,
        freshness_checker=_stub_freshness(
            _make_freshness(status=RESULT_STALE, gemini_commit_id=HEAD_B)
        ),
        invoke_scheduler=_stub_invoke(RESULT_FAILED),
        permission_diagnostics_provider=_stub_diag(headers_403),
    )
    result = router.route_for_pr(
        pr_number=107,
        current_head_sha=HEAD_A,
        owner="o",
        repo="r",
        decision_path="memory/decisions/d.json",
    )
    assert result.final_state == ROUTER_RESULT_NUDGE_PERMISSION_DENIED
    assert result.nudge_attempted is True
    assert result.nudge_result == RESULT_FAILED
    diag = result.permission_header_diagnostics
    assert isinstance(diag, dict)
    assert "x-accepted-github-permissions" in diag
    assert "x-ratelimit-remaining" in diag
    # raw token absent from diagnostics
    assert "authorization" not in diag
    import json as _json

    serialised = _json.dumps(diag)
    assert "ghp_" not in serialised
    assert "Bearer " not in serialised


def test_invoke_raises_with_403_headers_classifies_permission_denied(tmp_path):
    """invoke_scheduler 가 직접 예외를 던져도 diagnostics provider 가 403 회수하면 분류."""
    headers_403 = {
        "X-Accepted-GitHub-Permissions": "issues=write",
        "X-OAuth-Scopes": "public_repo",
    }
    router = _make_router(
        tmp_path,
        freshness_checker=_stub_freshness(
            _make_freshness(status=RESULT_STALE, gemini_commit_id=HEAD_B)
        ),
        invoke_scheduler=_stub_invoke_raises(RuntimeError("403")),
        permission_diagnostics_provider=_stub_diag(headers_403),
    )
    result = router.route_for_pr(
        pr_number=107,
        current_head_sha=HEAD_A,
        owner="o",
        repo="r",
        decision_path="memory/decisions/d.json",
    )
    assert result.final_state == ROUTER_RESULT_NUDGE_PERMISSION_DENIED


def test_failed_invoke_without_403_headers_falls_back_to_chair_ui(tmp_path):
    """diagnostics provider 가 None 반환 → CHAIR_UI_FALLBACK_REQUIRED 최후 수단."""
    router = _make_router(
        tmp_path,
        freshness_checker=_stub_freshness(
            _make_freshness(status=RESULT_STALE, gemini_commit_id=HEAD_B)
        ),
        invoke_scheduler=_stub_invoke(RESULT_FAILED),
        permission_diagnostics_provider=_no_diag,
    )
    result = router.route_for_pr(
        pr_number=143,
        current_head_sha=HEAD_A,
        owner="o",
        repo="r",
        decision_path="memory/decisions/d.json",
    )
    assert result.final_state == ROUTER_RESULT_CHAIR_UI_FALLBACK_REQUIRED


def test_invoke_raises_without_diagnostics_falls_back_to_chair_ui(tmp_path):
    router = _make_router(
        tmp_path,
        freshness_checker=_stub_freshness(
            _make_freshness(status=RESULT_STALE, gemini_commit_id=HEAD_B)
        ),
        invoke_scheduler=_stub_invoke_raises(RuntimeError("token gone")),
        permission_diagnostics_provider=_no_diag,
    )
    result = router.route_for_pr(
        pr_number=143,
        current_head_sha=HEAD_A,
        owner="o",
        repo="r",
        decision_path="memory/decisions/d.json",
    )
    assert result.final_state == ROUTER_RESULT_CHAIR_UI_FALLBACK_REQUIRED


# ─── 7. invoke 결과 DEDUPED (owner_trigger_only 의 dedupe 보조 가드) ──────────


def test_invoke_returns_deduped_classifies_router_deduped(tmp_path):
    router = _make_router(
        tmp_path,
        freshness_checker=_stub_freshness(
            _make_freshness(status=RESULT_STALE, gemini_commit_id=HEAD_B)
        ),
        invoke_scheduler=_stub_invoke(RESULT_DEDUPED),
    )
    result = router.route_for_pr(
        pr_number=143,
        current_head_sha=HEAD_A,
        owner="o",
        repo="r",
        decision_path="memory/decisions/d.json",
    )
    assert result.final_state == ROUTER_RESULT_NUDGE_DEDUPED


# ─── 8. STALE with no decision_path raises contract error ─────────────────────


def test_stale_without_decision_path_raises_contract_error(tmp_path):
    router = _make_router(
        tmp_path,
        freshness_checker=_stub_freshness(
            _make_freshness(status=RESULT_STALE, gemini_commit_id=HEAD_B)
        ),
        invoke_scheduler=_stub_invoke(RESULT_POSTED),
    )
    with pytest.raises(RouterContractError):
        router.route_for_pr(
            pr_number=143,
            current_head_sha=HEAD_A,
            owner="o",
            repo="r",
            decision_path=None,
        )


# ─── 9. input validation (positive int / 40-char hex / non-empty strings) ────


@pytest.mark.parametrize(
    "kwargs",
    [
        {"pr_number": 0},
        {"pr_number": -1},
        {"pr_number": True},
        {"current_head_sha": "short"},
        {"current_head_sha": "g" * 40},  # invalid hex
        {"owner": ""},
        {"owner": "with/slash"},
        {"repo": ""},
        {"repo": "x/y"},
    ],
)
def test_input_validation_rejects_bad_kwargs(tmp_path, kwargs):
    router = _make_router(
        tmp_path,
        freshness_checker=_stub_freshness(_make_freshness(status=RESULT_FRESH)),
        invoke_scheduler=_stub_invoke(RESULT_POSTED),
    )
    base = {
        "pr_number": 143,
        "current_head_sha": HEAD_A,
        "owner": "o",
        "repo": "r",
        "decision_path": "x",
    }
    base.update(kwargs)
    with pytest.raises(RouterContractError):
        router.route_for_pr(**base)


# ─── 10. audit JSONL final_state recorded ────────────────────────────────────


def test_audit_jsonl_records_final_state_and_redacted_diagnostics(tmp_path):
    import json as _json

    audit = OwnerGeminiTriggerRouterAudit(tmp_path)
    router = OwnerGeminiTriggerRouter(
        workspace_root=tmp_path,
        freshness_checker=_stub_freshness(
            _make_freshness(status=RESULT_STALE, gemini_commit_id=HEAD_B)
        ),
        invoke_scheduler=_stub_invoke(RESULT_FAILED),
        permission_diagnostics_provider=_stub_diag(
            {
                "X-Accepted-GitHub-Permissions": "issues=write",
                "X-Accepted-OAuth-Scopes": "repo",
                "X-RateLimit-Remaining": "1",
                "Authorization": "Bearer ghp_NEVER",
            }
        ),
        audit=audit,
        github_api=_dummy_github_api,
    )
    result = router.route_for_pr(
        pr_number=143,
        current_head_sha=HEAD_A,
        owner="o",
        repo="r",
        decision_path="d",
    )
    assert result.final_state == ROUTER_RESULT_NUDGE_PERMISSION_DENIED
    text = audit.path.read_text(encoding="utf-8").strip()
    row = _json.loads(text.splitlines()[-1])
    assert row["schema"] == AUDIT_SCHEMA
    assert row["final_state"] == ROUTER_RESULT_NUDGE_PERMISSION_DENIED
    assert row["nudge_attempted"] is True
    assert row["nudge_result"] == RESULT_FAILED
    assert row["token_value_logged"] is False
    # diag whitelisted
    diag = row["permission_header_diagnostics"]
    assert "x-accepted-github-permissions" in diag
    assert "authorization" not in diag


# ─── 11. token_hash_prefix is recorded but raw token never logged ────────────


def test_token_provider_records_only_hash_prefix(tmp_path):
    secret_token = "ghp_super_secret_token_value_xyz"
    router = _make_router(
        tmp_path,
        freshness_checker=_stub_freshness(_make_freshness(status=RESULT_FRESH)),
        invoke_scheduler=_stub_invoke(RESULT_POSTED),
        token_provider=lambda: secret_token,
    )
    result = router.route_for_pr(
        pr_number=143,
        current_head_sha=HEAD_A,
        owner="o",
        repo="r",
    )
    assert result.token_present is True
    assert len(result.token_hash_prefix) == 12
    assert secret_token not in result.token_hash_prefix
    # audit file 에도 raw token 미포함
    import json as _json

    text = router._audit.path.read_text(encoding="utf-8")  # noqa: SLF001
    assert secret_token not in text
    row = _json.loads(text.strip().splitlines()[-1])
    assert row["token_hash_prefix"] == result.token_hash_prefix
    assert row["token_value_logged"] is False


def test_token_provider_failure_yields_token_present_false(tmp_path):
    def _bad_token():
        raise RuntimeError("token unavailable")

    router = _make_router(
        tmp_path,
        freshness_checker=_stub_freshness(_make_freshness(status=RESULT_FRESH)),
        invoke_scheduler=_stub_invoke(RESULT_POSTED),
        token_provider=_bad_token,
    )
    result = router.route_for_pr(
        pr_number=143,
        current_head_sha=HEAD_A,
        owner="o",
        repo="r",
    )
    assert result.token_present is False
    assert result.token_hash_prefix == ""


# ─── 12. router_result enum exposed ──────────────────────────────────────────


def test_router_result_enum_distinct_states():
    states = {
        ROUTER_RESULT_FRESH,
        ROUTER_RESULT_NUDGE_POSTED,
        ROUTER_RESULT_NUDGE_DEDUPED,
        ROUTER_RESULT_GEMINI_EXTERNAL_TRIGGER_STALE,
        ROUTER_RESULT_CHAIR_UI_FALLBACK_REQUIRED,
        ROUTER_RESULT_NUDGE_PERMISSION_DENIED,
        ROUTER_RESULT_NOT_GEMINI_TRIGGER,
    }
    assert len(states) == 7