"""anu_v2.tests.test_gemini_stale_prevention_runner_2545 — 회귀 13건 (회장 §명시 1:1).

회귀 케이스 (task-2545.md "3. 회귀 10건" + task-2545+2 회귀 박제 #11~13):
  01. same-PR safe (false_positive only)             — fixture 5 → SAME_PR_SAFE / SAME_PR_RESOLVED
  02. same-PR blocked replacement required           — fixture 1 → SAME_PR_BLOCKED_REPLACEMENT_REQUIRED → pivot 호출
  03. PR #86 사고 재현 → replacement                 — 4단 head 변경 시뮬, 1단계에서 SAME_PR_BLOCKED 차단 어설션
  04. PR #88 mixed thread 분리                        — fixture 2 → code-changing 2 / non-code 2 분리
  05. PR #76 empty commit 차단                        — fixture 3 → block_empty_commit_attempt() + audit jsonl 박제
  06. scope expansion → Critical 7종 #3              — proposed_fixes에 expected_files 외 path → kind=3
  07. replacement PR clean path                       — fixture 4 → opened + 3분 내 evidence → REPLACEMENT_PR_OPENED
  08. PR_OPEN_HEALTH_GATE evidence 미도착             — fixture 4 변형 → MISSED 분류, long_polling_invoked=False 강제
  09. md/report fallback 금지                         — runner 코드에 .md / qc-result magic string 0건 어설션
  10. interface contract                              — 7 method 시그니처 + class 이름 보존
  11. REPLACEMENT_PR_CONTRACT_FRAMING_INCONSISTENT 박제 — PR #93 사고 reproduce → Critical 7종 #6
  12. contract framing OK when replacement matches effective diff — task-2545+2 정상 케이스
  13. contract framing OK when original PR merged    — state-aware exemption (MERGED → 축소 정당)

본 회귀는 anu_v2/* 모듈만 import 한다 (one-way isolation).
"""

from __future__ import annotations

import inspect
import json
import re
import sys
from pathlib import Path
from typing import Any

import pytest

# workspace root → sys.path (anu_v2 패키지 절대 import)
WORKSPACE_ROOT = Path(__file__).resolve().parents[2]
if str(WORKSPACE_ROOT) not in sys.path:
    sys.path.insert(0, str(WORKSPACE_ROOT))

FIXTURES_DIR = Path(__file__).resolve().parent.parent / "fixtures"


def load(name: str) -> dict[str, Any]:
    return json.loads((FIXTURES_DIR / name).read_text(encoding="utf-8"))


# ─── runner import (defer) ──────────────────────────────────────────────────
# 불칸의 본체 작성과 병렬 진행이므로 import 실패 시 해당 회귀만 fail.
def _import_runner():
    from anu_v2.gemini_stale_prevention_runner import (  # type: ignore[import-not-found]  # noqa: E402
        GeminiStalePreventionRunner,
    )

    return GeminiStalePreventionRunner


# ─── shared mocks ───────────────────────────────────────────────────────────
def _make_fake_replacement_runner(
    replacement_pr_number: int = 1000,
    replacement_branch: str = "task/task-XXXX-replacement-1",
    replacement_head_sha: str = "newsha1234",
    record: list | None = None,
):
    def fake(*, original_pr, original_branch, original_head_sha, proposed_fixes, chat_id):
        if record is not None:
            record.append(
                {
                    "original_pr": original_pr,
                    "original_branch": original_branch,
                    "original_head_sha": original_head_sha,
                    "proposed_fixes": proposed_fixes,
                    "chat_id": chat_id,
                }
            )
        return {
            "replacement_pr_number": replacement_pr_number,
            "replacement_branch": replacement_branch,
            "replacement_head_sha": replacement_head_sha,
            "original_pr_preserved": True,
        }

    return fake


def _make_fake_health_gate(
    evidence_arrived: bool = True,
    elapsed_seconds: int = 60,
    classification: str = "PR_OPEN_GEMINI_TRIGGER_OK",
    long_polling_invoked: bool = False,
    record: list | None = None,
):
    def fake(replacement_pr, replacement_head_sha, grace_seconds):
        if record is not None:
            record.append(
                {
                    "replacement_pr": replacement_pr,
                    "replacement_head_sha": replacement_head_sha,
                    "grace_seconds": grace_seconds,
                }
            )
        return {
            "evidence_arrived": evidence_arrived,
            "elapsed_seconds": elapsed_seconds,
            "classification": classification,
            "long_polling_invoked": long_polling_invoked,
        }

    return fake


def _build_runner(
    *,
    replacement_pr_runner_callable=None,
    pr_open_health_gate_callable=None,
    audit_records: list | None = None,
    now: str = "2026-05-10T12:00:00Z",
):
    Runner = _import_runner()
    audit_records = audit_records if audit_records is not None else []

    def fake_audit(record):
        audit_records.append(record)

    return Runner(
        replacement_pr_runner_callable=replacement_pr_runner_callable
        or _make_fake_replacement_runner(),
        pr_open_health_gate_callable=pr_open_health_gate_callable
        or _make_fake_health_gate(),
        audit_writer=fake_audit,
        now_factory=lambda: now,
    )


# ════════════════════════════════════════════════════════════════════════════
# 회귀 #01 — same-PR safe (false_positive only)
# ════════════════════════════════════════════════════════════════════════════
def test_01_same_pr_safe_false_positive_only():
    fx = load("stale_prevention_false_positive_same_pr_resolve.json")
    triage = fx["input"]["triage_classifications"]

    runner = _build_runner()

    eval_result = runner.evaluate_push_safety(
        pr_number=fx["input"]["pr_number"],
        current_head_sha=fx["input"]["head_sha"],
        gemini_review_commit_id=fx["input"]["gemini_review_commit_id"],
        triage_classifications=triage,
    )
    assert eval_result["decision"] == "SAME_PR_SAFE", eval_result
    assert eval_result["code_changing_threads"] == [], eval_result
    assert sorted(eval_result["non_code_changing_threads"]) == [1, 2, 3], eval_result

    run_result = runner.run(
        pr_number=fx["input"]["pr_number"],
        current_head_sha=fx["input"]["head_sha"],
        gemini_review_commit_id=fx["input"]["gemini_review_commit_id"],
        triage_classifications=triage,
        original_branch="task/task-XXXX",
    )
    assert run_result["outcome"] == "SAME_PR_RESOLVED", run_result
    assert run_result["pivot_result"] is None, run_result
    assert run_result["health_gate_result"] is None, run_result
    assert run_result["critical_seven_classification"] is None, run_result


# ════════════════════════════════════════════════════════════════════════════
# 회귀 #02 — same-PR blocked replacement required (minor_fix_in_scope)
# ════════════════════════════════════════════════════════════════════════════
def test_02_same_pr_blocked_replacement_required():
    fx = load("stale_prevention_pr86_same_pr_fix.json")

    triage = [
        {
            "thread_id": 1,
            "classification": "minor_fix_in_scope",
            "proposed_fix": {"file": "anu_v2/some_module.py", "lines": [10, 20]},
        }
    ]

    pivot_calls: list = []
    fake_replacement = _make_fake_replacement_runner(record=pivot_calls)
    runner = _build_runner(replacement_pr_runner_callable=fake_replacement)

    eval_result = runner.evaluate_push_safety(
        pr_number=fx["pr_number"],
        current_head_sha=fx["head_sequence"][0]["sha"],
        gemini_review_commit_id=fx["head_sequence"][0]["sha"],
        triage_classifications=triage,
    )
    assert eval_result["decision"] == "SAME_PR_BLOCKED_REPLACEMENT_REQUIRED", eval_result
    assert eval_result["code_changing_threads"] == [1], eval_result
    assert eval_result["auto_retry_allowed"] is False, eval_result

    pivot_result = runner.pivot_to_replacement_pr(
        original_pr=fx["pr_number"],
        original_branch="task/task-2537",
        original_head_sha=fx["head_sequence"][0]["sha"],
        proposed_fixes=[{"file": "anu_v2/some_module.py", "lines": [10, 20]}],
    )
    assert pivot_result["replacement_pr_number"] == 1000, pivot_result
    assert pivot_result["original_pr_preserved"] is True, pivot_result
    assert len(pivot_calls) == 1, pivot_calls
    assert pivot_calls[0]["original_pr"] == fx["pr_number"], pivot_calls
    assert pivot_calls[0]["chat_id"] == 6937032012, pivot_calls


# ════════════════════════════════════════════════════════════════════════════
# 회귀 #03 — PR #86 사고 재현 → 첫 단계에서 SAME_PR_BLOCKED 차단됨을 어설션
# ════════════════════════════════════════════════════════════════════════════
def test_03_pr86_accident_reproduction_blocked_at_first_step():
    fx = load("stale_prevention_pr86_same_pr_fix.json")
    head_sequence = fx["head_sequence"]
    assert len(head_sequence) == 4, "PR #86 fixture는 4단 head sequence 박제"

    triage = [
        {
            "thread_id": 1,
            "classification": "minor_fix_in_scope",
            "proposed_fix": {"file": "anu_v2/example.py", "lines": [1, 2]},
        }
    ]

    pivot_calls: list = []
    runner = _build_runner(
        replacement_pr_runner_callable=_make_fake_replacement_runner(record=pivot_calls)
    )

    # 1단계 (initial_push, evidence missing) — Gemini 도착 후의 첫 fix 시도
    first_head = head_sequence[0]["sha"]
    # second_head (head_sequence[1]) 는 정책 적용 시 도달하지 않아야 하는 step이라
    # 명시적 변수 할당은 생략 (정책상 1→2 same-PR push 자체가 차단되기 때문)

    # Gemini evidence 도착 시점에 same-PR fix push 시도 (=시퀀스 1→2)
    eval_first = runner.evaluate_push_safety(
        pr_number=fx["pr_number"],
        current_head_sha=first_head,
        gemini_review_commit_id=first_head,  # evidence 도착 후
        triage_classifications=triage,
    )
    assert eval_first["decision"] == "SAME_PR_BLOCKED_REPLACEMENT_REQUIRED", eval_first

    # runner 정책 적용 시 head 변경은 0회 — replacement PR로 우회
    pivot = runner.pivot_to_replacement_pr(
        original_pr=fx["pr_number"],
        original_branch="task/task-2537",
        original_head_sha=first_head,
        proposed_fixes=[{"file": "anu_v2/example.py", "lines": [1, 2]}],
    )
    assert pivot["replacement_pr_number"] == 1000, pivot
    assert pivot["original_pr_preserved"] is True, pivot

    # second/third/fourth head 변경은 정책에 의해 발생하지 않음 — 즉 head 변화 카운트는 0
    # (실제 push가 없었음을 fake mock 호출 횟수로 입증)
    assert len(pivot_calls) == 1, pivot_calls
    # head_sequence의 2/3/4단 sha는 본 테스트에서 한 번도 head로 사용되지 않음
    used_heads = {first_head}
    blocked_heads = {h["sha"] for h in head_sequence[1:]}
    assert blocked_heads.isdisjoint(used_heads), "정책 적용 시 후속 head 변경 0"


# ════════════════════════════════════════════════════════════════════════════
# 회귀 #04 — PR #88 mixed thread 분리
# ════════════════════════════════════════════════════════════════════════════
def test_04_pr88_mixed_thread_separation():
    fx = load("stale_prevention_pr88_unresolved_push.json")

    triage = []
    for entry in fx["expected_decision_with_runner_per_thread"]:
        item = {
            "thread_id": entry["thread"],
            "classification": entry["classification"],
        }
        if entry["classification"] in ("minor_fix_in_scope", "real_bug_in_scope"):
            item["proposed_fix"] = {"file": "anu_v2/sample.py", "lines": [1]}
        triage.append(item)

    runner = _build_runner()
    eval_result = runner.evaluate_push_safety(
        pr_number=fx["pr_number"],
        current_head_sha=fx["current_head"],
        gemini_review_commit_id=fx["gemini_review_commit_id"],
        triage_classifications=triage,
    )

    assert eval_result["decision"] == "SAME_PR_BLOCKED_REPLACEMENT_REQUIRED", eval_result
    assert sorted(eval_result["code_changing_threads"]) == [1, 2], eval_result
    assert sorted(eval_result["non_code_changing_threads"]) == [3, 4], eval_result
    assert len(eval_result["code_changing_threads"]) == 2, eval_result
    assert len(eval_result["non_code_changing_threads"]) == 2, eval_result


# ════════════════════════════════════════════════════════════════════════════
# 회귀 #05 — PR #76 empty commit 차단
# ════════════════════════════════════════════════════════════════════════════
def test_05_pr76_empty_commit_blocked():
    fx = load("stale_prevention_pr76_empty_commit_fail.json")

    audit_records: list = []
    runner = _build_runner(audit_records=audit_records)

    result = runner.block_empty_commit_attempt(
        pr_number=fx["pr_number"],
        ts="2026-05-10T12:00:00Z",
    )

    assert result["blocked"] is True, result
    assert result["kind"] == "EMPTY_COMMIT_TRIGGER_ATTEMPT_BLOCKED", result
    assert "memory/orchestration-audit/empty_commit_block.jsonl" in result.get(
        "audit_jsonl_path", ""
    ), result
    assert result["ts"] == "2026-05-10T12:00:00Z", result

    # audit_writer로 박제됐는지
    assert len(audit_records) >= 1, audit_records
    last = audit_records[-1]
    assert last.get("kind") == "EMPTY_COMMIT_TRIGGER_ATTEMPT_BLOCKED", last
    assert last.get("pr_number") == fx["pr_number"], last

    # 실제 git/subprocess/os.system 호출이 코드에 import 조차 안 되었는지 grep
    runner_path = (
        Path(__file__).resolve().parent.parent / "gemini_stale_prevention_runner.py"
    )
    src = runner_path.read_text(encoding="utf-8")
    # subprocess는 절대 금지
    assert not re.search(r"^\s*import\s+subprocess", src, flags=re.MULTILINE), (
        "subprocess import 금지"
    )
    assert not re.search(r"^\s*from\s+subprocess\s+import", src, flags=re.MULTILINE), (
        "subprocess import 금지"
    )
    # os.system / popen 직접 호출 금지
    assert "os.system" not in src, "os.system 호출 금지"
    assert "os.popen" not in src, "os.popen 호출 금지"


# ════════════════════════════════════════════════════════════════════════════
# 회귀 #06 — scope expansion → Critical 7종 #3
# ════════════════════════════════════════════════════════════════════════════
def test_06_scope_expansion_critical_seven_kind_three():
    runner = _build_runner()

    expected_files = [
        "anu_v2/some_module.py",
        "anu_v2/tests/test_some_module.py",
    ]
    proposed_fixes = [
        {"file": "anu_v2/some_module.py", "lines": [1, 2]},
        {"file": "utils/legacy_helper.py", "lines": [10]},  # ← 외부 path
        {"file": "scripts/run_x.sh", "lines": [3]},  # ← 외부 path
    ]

    cls = runner.classify_scope_expansion_as_critical_three(
        pr_number=999,
        proposed_fixes=proposed_fixes,
        expected_files_original=expected_files,
    )

    assert cls["critical_seven_kind"] == 3, cls
    assert cls["kind_name"] == "EXPECTED_FILES_SCOPE_EXPANSION", cls
    assert cls["pr_number"] == 999, cls
    assert sorted(cls["proposed_outside_files"]) == sorted(
        ["utils/legacy_helper.py", "scripts/run_x.sh"]
    ), cls
    assert cls["report_to_chairman_required"] is True, cls
    assert cls["expected_files_original"] == expected_files, cls


# ════════════════════════════════════════════════════════════════════════════
# 회귀 #07 — replacement PR clean path
# ════════════════════════════════════════════════════════════════════════════
def test_07_replacement_pr_clean_path():
    fx = load("stale_prevention_replacement_pr_clean_path.json")
    inp = fx["input"]

    pivot_calls: list = []
    health_calls: list = []
    fake_replacement = _make_fake_replacement_runner(
        replacement_pr_number=fx["expected_replacement_pr"]["replacement_pr_number"],
        replacement_branch=fx["expected_replacement_pr"]["replacement_branch"],
        replacement_head_sha="newsha1234",
        record=pivot_calls,
    )
    fake_health = _make_fake_health_gate(
        evidence_arrived=True,
        elapsed_seconds=120,
        classification="PR_OPEN_GEMINI_TRIGGER_OK",
        long_polling_invoked=False,
        record=health_calls,
    )

    runner = _build_runner(
        replacement_pr_runner_callable=fake_replacement,
        pr_open_health_gate_callable=fake_health,
    )

    triage = [
        {
            "thread_id": t["thread_id"],
            "classification": t["classification"],
            "proposed_fix": {"file": "anu_v2/example.py", "lines": [1]},
        }
        for t in inp["minor_fix_threads"]
    ]

    run_result = runner.run(
        pr_number=inp["original_pr"],
        current_head_sha=inp["original_head"],
        gemini_review_commit_id=inp["gemini_review_commit_id"],
        triage_classifications=triage,
        original_branch="task/task-XXXX",
    )

    assert run_result["outcome"] == "REPLACEMENT_PR_OPENED", run_result
    assert run_result["pivot_result"] is not None, run_result
    assert (
        run_result["pivot_result"]["replacement_pr_number"]
        == fx["expected_replacement_pr"]["replacement_pr_number"]
    ), run_result
    assert run_result["pivot_result"]["original_pr_preserved"] is True, run_result
    assert run_result["health_gate_result"] is not None, run_result
    assert run_result["health_gate_result"]["evidence_arrived"] is True, run_result
    assert run_result["health_gate_result"]["long_polling_invoked"] is False, run_result
    assert (
        run_result["health_gate_result"]["classification"] == "PR_OPEN_GEMINI_TRIGGER_OK"
    ), run_result
    assert len(pivot_calls) == 1, pivot_calls
    assert len(health_calls) == 1, health_calls
    assert health_calls[0]["replacement_pr"] == 1000, health_calls
    assert health_calls[0]["grace_seconds"] == 180, health_calls


# ════════════════════════════════════════════════════════════════════════════
# 회귀 #08 — PR_OPEN_HEALTH_GATE evidence 미도착 → MISSED 분류, long polling 차단
# ════════════════════════════════════════════════════════════════════════════
def test_08_pr_open_health_gate_evidence_missed_no_long_polling():
    fx = load("stale_prevention_replacement_pr_clean_path.json")
    inp = fx["input"]

    fake_replacement = _make_fake_replacement_runner()
    # evidence 미도착 시나리오
    fake_health_missed = _make_fake_health_gate(
        evidence_arrived=False,
        elapsed_seconds=180,
        classification="PR_OPEN_GEMINI_TRIGGER_MISSED",
        long_polling_invoked=False,
    )

    runner = _build_runner(
        replacement_pr_runner_callable=fake_replacement,
        pr_open_health_gate_callable=fake_health_missed,
    )

    triage = [
        {
            "thread_id": 1,
            "classification": "minor_fix_in_scope",
            "proposed_fix": {"file": "anu_v2/example.py", "lines": [1]},
        }
    ]

    run_result = runner.run(
        pr_number=inp["original_pr"],
        current_head_sha=inp["original_head"],
        gemini_review_commit_id=inp["gemini_review_commit_id"],
        triage_classifications=triage,
        original_branch="task/task-XXXX",
    )

    assert run_result["health_gate_result"] is not None, run_result
    assert (
        run_result["health_gate_result"]["classification"] == "PR_OPEN_GEMINI_TRIGGER_MISSED"
    ), run_result
    assert run_result["health_gate_result"]["evidence_arrived"] is False, run_result
    assert run_result["health_gate_result"]["long_polling_invoked"] is False, run_result

    # 변형: long_polling_invoked=True 가 들어오면 runner는 거부해야 함
    fake_health_polling = _make_fake_health_gate(
        evidence_arrived=False,
        elapsed_seconds=600,
        classification="PR_OPEN_GEMINI_TRIGGER_MISSED",
        long_polling_invoked=True,  # ← 정책 위반
    )
    runner_violator = _build_runner(
        replacement_pr_runner_callable=fake_replacement,
        pr_open_health_gate_callable=fake_health_polling,
    )

    with pytest.raises((ValueError, AssertionError, RuntimeError)):
        runner_violator.run(
            pr_number=inp["original_pr"],
            current_head_sha=inp["original_head"],
            gemini_review_commit_id=inp["gemini_review_commit_id"],
            triage_classifications=triage,
            original_branch="task/task-XXXX",
        )


# ════════════════════════════════════════════════════════════════════════════
# 회귀 #09 — md/report fallback 금지 어설션
# ════════════════════════════════════════════════════════════════════════════
def _strip_docstrings_and_comments(src: str) -> str:
    """AST로 module/class/function docstring을 제거하고 # 주석을 제거하여 실행 코드만 반환."""
    import ast
    import io
    import tokenize

    # 1) AST로 docstring 노드 찾기
    tree = ast.parse(src)
    docstring_ranges: list[tuple[int, int]] = []  # (start_line, end_line) 1-based

    def visit(node):
        if isinstance(
            node, (ast.Module, ast.FunctionDef, ast.AsyncFunctionDef, ast.ClassDef)
        ):
            body = getattr(node, "body", None)
            if body and isinstance(body[0], ast.Expr) and isinstance(body[0].value, ast.Constant):
                if isinstance(body[0].value.value, str):
                    first = body[0]
                    docstring_ranges.append((first.lineno, first.end_lineno or first.lineno))
        for child in ast.iter_child_nodes(node):
            visit(child)

    visit(tree)

    lines = src.splitlines()
    keep = [True] * len(lines)
    for start, end in docstring_ranges:
        for i in range(start - 1, min(end, len(lines))):
            keep[i] = False

    code_only_lines = [ln for i, ln in enumerate(lines) if keep[i]]
    code_only = "\n".join(code_only_lines)

    # 2) tokenize로 주석 제거
    out = []
    try:
        tokens = tokenize.generate_tokens(io.StringIO(code_only).readline)
        for tok in tokens:
            if tok.type == tokenize.COMMENT:
                continue
            out.append(tok)
        # 단순 join 대신 untokenize
        result = tokenize.untokenize(out)
    except Exception:
        result = code_only
    return result


def test_09_no_md_report_fallback_in_runner():
    Runner = _import_runner()
    full_src = inspect.getsource(sys.modules[Runner.__module__])
    code_src = _strip_docstrings_and_comments(full_src)

    # 문자열 "qc-result", ".md", "Gemini PASS" magic match가 코드 path에 없는지
    # docstring/주석은 정책 설명용으로 허용, 실행 코드에는 0건이어야 함.
    md_hits = re.findall(r"\.md\b", code_src)
    assert md_hits == [], (
        f".md fallback path 금지 (코드 path에 found {len(md_hits)}): {md_hits[:5]}"
    )

    # qc-result 파일 직접 read 금지 (코드 path)
    assert "qc-result" not in code_src, (
        "qc-result 파일 참조 금지 (md/report fallback 차단)"
    )

    # Path(...).read_text on .md / report 파일 패턴 금지
    assert not re.search(r"read_text\([^)]*\.md", code_src), "read_text(...md) 호출 금지"

    # "Gemini PASS" 같은 magic string 매칭 path 금지
    assert "Gemini PASS" not in code_src, "Gemini PASS magic-string 매칭 금지"

    # bot /gemini review 코멘트 절대 금지 (성공 기준 #3) — 코드 path에 0건
    assert "/gemini review" not in code_src, "bot /gemini review 코멘트 금지 (코드 path)"

    # force / rebase / close-reopen 동사 절대 금지 (코드 path)
    forbidden = ["--force", "--force-with-lease", "git push -f", "force-push"]
    for kw in forbidden:
        assert kw not in code_src, f"force-push 동사 금지: {kw}"


# ════════════════════════════════════════════════════════════════════════════
# 회귀 #10 — interface contract (6 method + class 이름)
# ════════════════════════════════════════════════════════════════════════════
def test_10_interface_contract():
    Runner = _import_runner()

    assert Runner.__name__ == "GeminiStalePreventionRunner", Runner.__name__

    methods = [
        "evaluate_push_safety",
        "pivot_to_replacement_pr",
        "run_pr_open_health_gate",
        "block_empty_commit_attempt",
        "classify_scope_expansion_as_critical_three",
        "classify_replacement_contract_framing",
        "run",
    ]
    for m in methods:
        assert hasattr(Runner, m), f"missing method: {m}"
        assert callable(getattr(Runner, m)), f"not callable: {m}"

    # evaluate_push_safety signature
    sig = inspect.signature(Runner.evaluate_push_safety)
    expected_params = {
        "pr_number",
        "current_head_sha",
        "gemini_review_commit_id",
        "triage_classifications",
    }
    actual_params = set(sig.parameters.keys()) - {"self"}
    assert expected_params <= actual_params, (
        f"evaluate_push_safety params missing: expected {expected_params}, got {actual_params}"
    )

    # pivot_to_replacement_pr signature
    sig_pivot = inspect.signature(Runner.pivot_to_replacement_pr)
    expected_pivot = {
        "original_pr",
        "original_branch",
        "original_head_sha",
        "proposed_fixes",
    }
    actual_pivot = set(sig_pivot.parameters.keys()) - {"self"}
    assert expected_pivot <= actual_pivot, (
        f"pivot_to_replacement_pr params missing: expected {expected_pivot}, got {actual_pivot}"
    )

    # run_pr_open_health_gate signature
    sig_gate = inspect.signature(Runner.run_pr_open_health_gate)
    expected_gate = {"replacement_pr", "replacement_head_sha"}
    actual_gate = set(sig_gate.parameters.keys()) - {"self"}
    assert expected_gate <= actual_gate, (
        f"run_pr_open_health_gate params missing: expected {expected_gate}, got {actual_gate}"
    )

    # block_empty_commit_attempt signature
    sig_block = inspect.signature(Runner.block_empty_commit_attempt)
    expected_block = {"pr_number", "ts"}
    actual_block = set(sig_block.parameters.keys()) - {"self"}
    assert expected_block <= actual_block, (
        f"block_empty_commit_attempt params missing: expected {expected_block}, got {actual_block}"
    )

    # classify_scope_expansion_as_critical_three signature
    sig_scope = inspect.signature(Runner.classify_scope_expansion_as_critical_three)
    expected_scope = {"pr_number", "proposed_fixes", "expected_files_original"}
    actual_scope = set(sig_scope.parameters.keys()) - {"self"}
    assert expected_scope <= actual_scope, (
        f"classify_scope_expansion_as_critical_three params missing: "
        f"expected {expected_scope}, got {actual_scope}"
    )

    # run signature
    sig_run = inspect.signature(Runner.run)
    expected_run = {
        "pr_number",
        "current_head_sha",
        "gemini_review_commit_id",
        "triage_classifications",
        "original_branch",
    }
    actual_run = set(sig_run.parameters.keys()) - {"self"}
    assert expected_run <= actual_run, (
        f"run params missing: expected {expected_run}, got {actual_run}"
    )

    # GRACE_SECONDS_PR_OPEN_HEALTH 클래스 상수 (task-2544 정책과 동일 = 180)
    assert hasattr(Runner, "GRACE_SECONDS_PR_OPEN_HEALTH"), (
        "GRACE_SECONDS_PR_OPEN_HEALTH 상수 부재"
    )
    assert Runner.GRACE_SECONDS_PR_OPEN_HEALTH == 180, (
        f"grace = 180 (task-2544 동일), got {Runner.GRACE_SECONDS_PR_OPEN_HEALTH}"
    )

    # classify_replacement_contract_framing signature (task-2545+2 회귀 박제)
    sig_framing = inspect.signature(Runner.classify_replacement_contract_framing)
    expected_framing = {
        "replacement_task_id",
        "original_pr_number",
        "original_pr_state",
        "original_pr_merged_at",
        "replacement_expected_files",
        "origin_main_effective_diff_files",
    }
    actual_framing = set(sig_framing.parameters.keys()) - {"self"}
    assert expected_framing <= actual_framing, (
        f"classify_replacement_contract_framing params missing: "
        f"expected {expected_framing}, got {actual_framing}"
    )


# ════════════════════════════════════════════════════════════════════════════
# 회귀 #11 — REPLACEMENT_PR_CONTRACT_FRAMING_INCONSISTENT_WITH_ORIGIN_MAIN_STATE
#   회장 §명시 (2026-05-11, task-2545+2 corrected replacement 회귀 박제):
#     "original PR이 OPEN이면 replacement PR expected_files는 original PR이
#      main에 반영되었다고 가정해 축소하면 안 된다. replacement expected_files는
#      반드시 origin/main 실제 상태 기준 전체 effective diff로 산정한다."
#
#   실제 사고 (PR #93 task-2545+1):
#     - PR #92 state=OPEN, mergedAt=null (origin/main 미반영)
#     - origin/main 에 task-2545 산출물 0 file
#     - 그러나 task-2545+1 contract 가 expected_files=1 (runner.py only) 로 축소
#     - effective diff (7 files) 와 불일치 → REPLACEMENT_PR_CONTRACT_FRAMING_INCONSISTENT
# ════════════════════════════════════════════════════════════════════════════
def test_11_contract_framing_inconsistent_when_original_pr_open():
    """REPLACEMENT_PR_CONTRACT_FRAMING_INCONSISTENT_WITH_ORIGIN_MAIN_STATE 회귀 박제.

    PR #93 사고 reproduce: original PR OPEN + replacement expected_files 축소 →
    runner 가 contract_framing_inconsistent=True / Critical 7종 #6 보고 필요로 분류해야 한다.
    """
    from anu_v2.gemini_stale_prevention_runner import (  # noqa: E402
        CRITICAL_SEVEN_KIND_REPLACEMENT_CONTRACT_FRAMING,
        CRITICAL_SEVEN_KIND_REPLACEMENT_CONTRACT_FRAMING_NAME,
    )

    audit_records: list = []
    runner = _build_runner(audit_records=audit_records)

    # PR #93 사고 시나리오 (task-2545+1 contract framing 결함 reproduce)
    result = runner.classify_replacement_contract_framing(
        replacement_task_id="task-2545+1",
        original_pr_number=92,
        original_pr_state="OPEN",
        original_pr_merged_at=None,
        replacement_expected_files=[
            "anu_v2/gemini_stale_prevention_runner.py",
        ],  # 잘못된 framing — 1 file 만
        origin_main_effective_diff_files=[
            "anu_v2/gemini_stale_prevention_runner.py",
            "anu_v2/tests/test_gemini_stale_prevention_runner_2545.py",
            "anu_v2/fixtures/stale_prevention_false_positive_same_pr_resolve.json",
            "anu_v2/fixtures/stale_prevention_pr76_empty_commit_fail.json",
            "anu_v2/fixtures/stale_prevention_pr86_same_pr_fix.json",
            "anu_v2/fixtures/stale_prevention_pr88_unresolved_push.json",
            "anu_v2/fixtures/stale_prevention_replacement_pr_clean_path.json",
        ],  # 실제 effective diff = 7 files
    )

    assert result["contract_framing_inconsistent"] is True, (
        f"PR #93 사고 reproduce: contract_framing_inconsistent=True 기대, got {result}"
    )
    assert result["critical_seven_kind"] == CRITICAL_SEVEN_KIND_REPLACEMENT_CONTRACT_FRAMING
    assert result["critical_seven_kind"] == 6
    assert result["kind_name"] == CRITICAL_SEVEN_KIND_REPLACEMENT_CONTRACT_FRAMING_NAME
    assert result["kind_name"] == "REPLACEMENT_PR_CONTRACT_FRAMING_INCONSISTENT_WITH_ORIGIN_MAIN_STATE"
    assert result["original_pr"] == 92
    assert result["original_pr_state"] == "OPEN"
    assert result["original_pr_unmerged"] is True
    assert len(result["missing_from_replacement"]) == 6, (
        f"missing files = 6 (7 effective - 1 declared), got {result['missing_from_replacement']}"
    )
    assert result["report_to_chairman_required"] is True

    # audit jsonl 박제 어설션
    framing_audits = [
        r for r in audit_records
        if r.get("audit_kind") == "critical_seven_replacement_contract_framing"
    ]
    assert len(framing_audits) == 1
    assert framing_audits[0]["critical_seven_kind"] == 6
    assert framing_audits[0]["report_to_chairman_required"] is True
    assert framing_audits[0]["chat_id"] == 6937032012  # 회장 §명시 default chat


# ════════════════════════════════════════════════════════════════════════════
# 회귀 #12 — contract framing OK when replacement covers all effective diff
# ════════════════════════════════════════════════════════════════════════════
def test_12_contract_framing_ok_when_replacement_matches_effective_diff():
    """task-2545+2 corrected replacement 정상 케이스.

    replacement expected_files (7) == origin/main effective diff (7) →
    contract_framing_inconsistent=False / report_to_chairman_required=False.
    """
    audit_records: list = []
    runner = _build_runner(audit_records=audit_records)

    seven_files = [
        "anu_v2/gemini_stale_prevention_runner.py",
        "anu_v2/tests/test_gemini_stale_prevention_runner_2545.py",
        "anu_v2/fixtures/stale_prevention_false_positive_same_pr_resolve.json",
        "anu_v2/fixtures/stale_prevention_pr76_empty_commit_fail.json",
        "anu_v2/fixtures/stale_prevention_pr86_same_pr_fix.json",
        "anu_v2/fixtures/stale_prevention_pr88_unresolved_push.json",
        "anu_v2/fixtures/stale_prevention_replacement_pr_clean_path.json",
    ]
    result = runner.classify_replacement_contract_framing(
        replacement_task_id="task-2545+2",
        original_pr_number=92,
        original_pr_state="OPEN",
        original_pr_merged_at=None,
        replacement_expected_files=seven_files,
        origin_main_effective_diff_files=seven_files,
    )

    assert result["contract_framing_inconsistent"] is False
    assert result["report_to_chairman_required"] is False
    assert result["missing_from_replacement"] == []
    assert result["original_pr_unmerged"] is True  # PR OPEN 상태는 그대로 유지


# ════════════════════════════════════════════════════════════════════════════
# 회귀 #13 — contract framing OK when original PR merged (state-aware exemption)
# ════════════════════════════════════════════════════════════════════════════
def test_13_contract_framing_ok_when_original_pr_merged():
    """original PR 이 main 에 머지된 경우 expected_files 축소가 정당.

    예: original PR 머지 후 follow-up replacement 가 1 file 만 수정해도
    inconsistent 가 아니다. (origin/main 에는 이미 산출물이 반영되어 있으므로
    replacement 는 추가/수정분만 산정하면 됨)
    """
    audit_records: list = []
    runner = _build_runner(audit_records=audit_records)

    result = runner.classify_replacement_contract_framing(
        replacement_task_id="task-XXXX+1",
        original_pr_number=999,
        original_pr_state="MERGED",
        original_pr_merged_at="2026-05-10T12:00:00Z",
        replacement_expected_files=["anu_v2/some_module.py"],
        origin_main_effective_diff_files=[
            "anu_v2/some_module.py",
            "anu_v2/some_other.py",
        ],
    )

    # original PR merged → original_pr_unmerged=False → framing_inconsistent=False
    assert result["original_pr_unmerged"] is False
    assert result["contract_framing_inconsistent"] is False
    assert result["report_to_chairman_required"] is False