"""utils/replacement_pr_runner.py — task-2510 5 모듈 #2.
회장 명시: contaminated PR 자동 감지 + origin/main 기준 clean branch에서
expected_files만 이식하여 replacement PR 자동 생성.
원 PR은 close/delete 없이 보존.
"""
from __future__ import annotations

import argparse
import json
import logging
import os
import re
import subprocess
import sys
from dataclasses import asdict
from datetime import datetime, timezone
from pathlib import Path
from typing import Any, Callable, Optional, TYPE_CHECKING, TypeAlias

WORKSPACE = Path(os.environ.get("WORKSPACE_ROOT", "/home/jay/workspace"))

# CLI 직접 실행 시 패키지 루트를 sys.path에 추가
_HERE = Path(__file__).resolve().parent.parent  # utils/ → worktree root
if str(_HERE) not in sys.path:
    sys.path.insert(0, str(_HERE))

from utils.automation_contracts import (  # noqa: E402  # pyright: ignore[reportMissingImports]
    ReplacementResult,
    CriticalEscalationType,
    EscalationPacket,
)

# ─── circular import 회피 (task-2516) ────────────────────────────────────────
# merge_queue_executor.py가 top-level에서 ReplacementPRRunner를 import하기 때문에
# 본 모듈이 top-level에서 merge_queue_executor를 import하면 cycle이 발생한다.
# wrapper 함수로 lazy 위임하여 default wiring path(_WIRING_AVAILABLE=True)를 활성화한다.

TaskSpec: TypeAlias = Any  # runtime placeholder — 실제 클래스는 type hint 용도로만 참조됨
if TYPE_CHECKING:
    from utils.merge_queue_executor import TaskSpec  # type: ignore[no-redef]  # noqa: F401  # pyright: ignore[reportMissingImports]

def compare_effective_diff(*args, **kwargs):
    from utils import merge_queue_executor as _mqe  # pyright: ignore[reportMissingImports]
    return _mqe.compare_effective_diff(*args, **kwargs)


def detect_forbidden_paths(*args, **kwargs):
    from utils import merge_queue_executor as _mqe  # pyright: ignore[reportMissingImports]
    return _mqe.detect_forbidden_paths(*args, **kwargs)


def assert_no_forbidden_git_flags(*args, **kwargs):
    from utils import merge_queue_executor as _mqe  # pyright: ignore[reportMissingImports]
    return _mqe.assert_no_forbidden_git_flags(*args, **kwargs)


def load_task_spec(*args, **kwargs):
    from utils import merge_queue_executor as _mqe  # pyright: ignore[reportMissingImports]
    return _mqe.load_task_spec(*args, **kwargs)

logger = logging.getLogger(__name__)
RunnerType = Callable[..., subprocess.CompletedProcess]


def _default_runner(args, cwd=None, timeout=60):
    return subprocess.run(args, cwd=cwd or str(WORKSPACE), capture_output=True, text=True, timeout=timeout)


# ─── §0.5 safe repo_dir invariant (task-2784 P1) ────────────────────────────
def _resolve_safe_repo_dir(repo_dir: Optional[str], *, allow_workspace: bool = False) -> str:
    """replacement write/mutate 경로가 canonical WORKSPACE에 직접 write하지 못하게
    막는 safe repo_dir invariant. 통과 시 **git top-level 기준 safe repo root(str)** 반환,
    위반 시 RuntimeError raise.

    - repo_dir is None:
        · allow_workspace=False(automation/merge_queue/P1 기본) → RuntimeError("MISSING_REPO_DIR") (fail-closed)
        · ★ allow_workspace=True(명시적 CLI/manual opt-in 한정) → str(WORKSPACE.resolve()) 허용.
          이 예외는 **CLI 수동 모드 전용** — automation/merge_queue/P1 경로의 WORKSPACE fallback 부활 아님.
    - allow_workspace=True (repo_dir 지정 시) → Path(repo_dir).expanduser().resolve() 정규화 후 반환.
    - 입력 경로는 expanduser().resolve()로 정규화(~ 홈경로도 실제 경로로 확장).
    - Path(repo_dir).expanduser().resolve() == WORKSPACE → RuntimeError("WORKSPACE_WRITE_FORBIDDEN")
    - git top-level 기준 판정(핵심):
        git -C <repo_dir> rev-parse --show-toplevel 결과가 canonical WORKSPACE와 같으면
        → RuntimeError("WORKSPACE_WRITE_FORBIDDEN")
        · canonical 하위 일반 디렉터리(top-level=WORKSPACE) → fail
        · 별도 git worktree(top-level=worktree root) → 허용(경로가 WORKSPACE 하위여도)
    - ★ git rev-parse 실패/비-git dir → **fail-closed(RuntimeError)**. 보수적 허용 금지
      (판정 실패 시 fail-closed 원칙 — canonical write를 확실히 차단).
    - 통과 시 입력 경로가 아니라 **git top-level(정규화된 safe repo root)**을 반환하여
      이후 self.repo_dir 재할당 경로(create_clean_replacement_branch/open_replacement_pr 등)가
      raw/비정규화 경로를 다시 쓰지 않게 한다.
    """
    if repo_dir is None:
        # ★ 명시적 CLI/manual allow_workspace 예외 한정: repo_dir 미지정 CLI 수동 실행 허용.
        # automation/merge_queue/P1 경로(allow_workspace=False)는 여전히 MISSING_REPO_DIR fail-closed.
        if allow_workspace:
            return str(WORKSPACE.resolve())
        raise RuntimeError("MISSING_REPO_DIR")
    if allow_workspace:
        return str(Path(repo_dir).expanduser().resolve())
    workspace_resolved = WORKSPACE.resolve()
    resolved_input = Path(repo_dir).expanduser().resolve()
    if resolved_input == workspace_resolved:
        raise RuntimeError("WORKSPACE_WRITE_FORBIDDEN")
    # git top-level 판정 (실패/비-git dir = fail-closed)
    try:
        proc = subprocess.run(
            ["git", "-C", str(resolved_input), "rev-parse", "--show-toplevel"],
            capture_output=True, text=True, errors="replace", timeout=30,
        )
    except Exception as exc:
        raise RuntimeError(f"REPO_DIR_GIT_PROBE_FAILED:{exc}")
    if proc.returncode != 0:
        raise RuntimeError("REPO_DIR_NOT_GIT_REPO")
    toplevel = (proc.stdout or "").strip()
    if not toplevel:
        raise RuntimeError("REPO_DIR_TOPLEVEL_EMPTY")
    toplevel_resolved = Path(toplevel).expanduser().resolve()
    if toplevel_resolved == workspace_resolved:
        raise RuntimeError("WORKSPACE_WRITE_FORBIDDEN")
    return str(toplevel_resolved)


# ─── §1 cherry-pick 정적 차단 ────────────────────────────────────────────────
def assert_no_cherry_pick(args: list[str]) -> None:
    """args에 'cherry-pick' 토큰이 들어가면 RuntimeError(CHERRY_PICK_FORBIDDEN)."""
    if "cherry-pick" in args:
        raise RuntimeError("CHERRY_PICK_FORBIDDEN")


# ─── §2 PR metadata 수집 ────────────────────────────────────────────────────
def fetch_pr_metadata(pr_number: int, runner: RunnerType) -> dict:
    """gh pr view --json headRefName,headRefOid,baseRefName,files,title.
    return {"head_ref": ..., "head_sha": ..., "base_ref": ..., "files": [...], "task_id": ..., "title": ..., "number": pr_number}
    실패 시 RuntimeError.
    task_id는 title 또는 head_ref에서 [task-NNNN] 패턴 추출.
    """
    args = ["gh", "pr", "view", str(pr_number), "--json",
            "headRefName,headRefOid,baseRefName,files,title"]
    result = runner(args)
    if result.returncode != 0:
        raise RuntimeError(f"FETCH_PR_METADATA_FAILED: pr={pr_number} stderr={result.stderr!r}")
    data = json.loads(result.stdout or "{}")
    head_ref = data.get("headRefName", "")
    title = data.get("title", "")
    files = [f.get("path", "") for f in (data.get("files") or []) if f.get("path")]
    # task_id 추출
    m = re.search(r"task-\d+(?:\+\d+)?", title) or re.search(r"task-\d+(?:\+\d+)?", head_ref)
    task_id = m.group(0) if m else "unknown"
    return {
        "head_ref": head_ref,
        "head_sha": data.get("headRefOid", ""),
        "base_ref": data.get("baseRefName", "main"),
        "files": files,
        "task_id": task_id,
        "title": title,
        "number": pr_number,
    }


# ─── §3 effective diff 산출 ─────────────────────────────────────────────────
def compute_effective_diff(pr_meta: dict, runner: RunnerType) -> list[str]:
    """gh pr view에서 받은 files 우선 사용, 비어있으면 git diff origin/main...PR_HEAD --name-only.
    """
    if pr_meta.get("files"):
        return list(pr_meta["files"])
    head_sha = pr_meta.get("head_sha", "")
    if not head_sha:
        return []
    args = ["git", "diff", "--name-only", f"origin/{pr_meta.get('base_ref', 'main')}...{head_sha}"]
    assert_no_forbidden_git_flags(args)
    result = runner(args)
    if result.returncode != 0:
        return []
    return [line.strip() for line in (result.stdout or "").splitlines() if line.strip()]


# ─── §4 contaminated 판정 ──────────────────────────────────────────────────
def detect_contamination(
    effective_files: list[str],
    expected_files: list[str],
    extra_forbidden_patterns: Optional[list] = None,
) -> dict:
    """반환 dict: {"contaminated": bool, "forbidden_paths": [...], "extra": [...], "missing": [...]}"""
    extra_pats = None
    if extra_forbidden_patterns:
        extra_pats = [p if hasattr(p, "search") else re.compile(p) for p in extra_forbidden_patterns]
    forbidden = detect_forbidden_paths(effective_files, expected_files, extra_patterns=extra_pats)
    equal, extra, missing = compare_effective_diff(effective_files, expected_files)
    contaminated = bool(forbidden) or not equal
    return {"contaminated": contaminated, "forbidden_paths": forbidden, "extra": extra, "missing": missing}


# ─── §5 clean replacement branch 생성 ───────────────────────────────────────
def create_clean_replacement_branch(
    task_id: str,
    runner: RunnerType,
    *,
    timestamp: Optional[str] = None,
    repo_dir: Optional[str] = None,
) -> str:
    """origin/main 기준 신규 brace `task/<task_id>-replacement-<timestamp>` 생성.
    fetch origin main으로 stale base 회피. force/rebase/cherry-pick 금지.
    """
    if timestamp is None:
        timestamp = datetime.now(timezone.utc).strftime("%Y%m%d%H%M%S")
    branch = f"task/{task_id}-replacement-{timestamp}"

    # ★ Codex high #1: stale origin/main 회피를 위해 fetch 먼저 수행
    fetch_args = ["git", "fetch", "origin", "main", "--quiet"]
    assert_no_forbidden_git_flags(fetch_args)
    assert_no_cherry_pick(fetch_args)
    fetch_result = runner(fetch_args, cwd=repo_dir)
    if fetch_result.returncode != 0:
        raise RuntimeError(
            f"FETCH_ORIGIN_MAIN_FAILED: stderr={fetch_result.stderr!r}"
        )

    args = ["git", "checkout", "-b", branch, "origin/main"]
    assert_no_forbidden_git_flags(args)
    assert_no_cherry_pick(args)
    result = runner(args, cwd=repo_dir)
    if result.returncode != 0:
        raise RuntimeError(f"CREATE_CLEAN_BRANCH_FAILED: branch={branch} stderr={result.stderr!r}")
    return branch


# ─── §6 expected_files 이식 (cherry-pick 금지) ─────────────────────────────
def transplant_expected_files(
    expected_files: list[str],
    source_head: str,
    runner: RunnerType,
    *,
    repo_dir: Optional[str] = None,
    allow_workspace: bool = False,
) -> list[str]:
    """각 파일 git show <source_head>:<path> → write → git add. cherry-pick 정적 차단.
    파일 시스템에 실제로 write하므로 호출자는 임시 작업 dir(tmp_path 등)을 repo_dir로 줘야 source 손상 방지.
    """
    transplanted: list[str] = []
    cwd = _resolve_safe_repo_dir(repo_dir, allow_workspace=allow_workspace)
    repo_root = Path(cwd).resolve()
    for filepath in expected_files:
        show_args = ["git", "show", f"{source_head}:{filepath}"]
        assert_no_forbidden_git_flags(show_args)
        assert_no_cherry_pick(show_args)
        sr = runner(show_args, cwd=cwd)
        if sr.returncode != 0:
            raise RuntimeError(f"GIT_SHOW_FAILED: file={filepath} sha={source_head} stderr={sr.stderr!r}")
        target = Path(cwd) / filepath
        # ── Guard 1: target은 반드시 repo root(cwd) 안으로 resolve되어야 함 (밖이면 fail-closed) ──
        resolved_target = target.resolve()
        try:
            resolved_target.relative_to(repo_root)
        except ValueError:
            raise RuntimeError(
                f"TARGET_OUTSIDE_REPO_ROOT: file={filepath} resolved={resolved_target} repo_root={repo_root}"
            )
        # A안(task-2783+1): git show rc0 + empty stdout = 정상 empty blob → write 허용.
        # EMPTY_OVERWRITE_REFUSED guard 제거(정상 empty-file PR 과차단 false-positive 해소).
        # git show 실패(returncode!=0)는 위 GIT_SHOW_FAILED에서 이미 fail-closed로 방어됨.
        content = sr.stdout or ""
        target.parent.mkdir(parents=True, exist_ok=True)
        # Guard 2 (task-2783+2): repo 안으로 확인된 target이 symlink면 링크 따라가기(victim 덮어쓰기) 방지 —
        # symlink 자체를 unlink 후 regular file로 write. repo 밖 가리키는 symlink는 위 Guard1에서 이미 fail(미삭제).
        if target.is_symlink():
            target.unlink()
        target.write_text(content, encoding="utf-8")
        add_args = ["git", "add", filepath]
        assert_no_forbidden_git_flags(add_args)
        assert_no_cherry_pick(add_args)
        ar = runner(add_args, cwd=cwd)
        if ar.returncode != 0:
            raise RuntimeError(f"GIT_ADD_FAILED: file={filepath} stderr={ar.stderr!r}")
        transplanted.append(filepath)
    return transplanted


# ─── §7 commit (push X) ────────────────────────────────────────────────────
def commit_local(
    task_id: str,
    runner: RunnerType,
    *,
    repo_dir: Optional[str] = None,
    allow_workspace: bool = False,
):
    """git commit -m only (push 하지 않음). force 금지. 실패 시 RuntimeError."""
    cwd = _resolve_safe_repo_dir(repo_dir, allow_workspace=allow_workspace)
    commit_msg = f"[{task_id}] replacement: expected_files only (auto-generated)"
    cargs = ["git", "commit", "-m", commit_msg]
    assert_no_forbidden_git_flags(cargs)
    assert_no_cherry_pick(cargs)
    cr = runner(cargs, cwd=cwd)
    if cr.returncode != 0:
        raise RuntimeError(f"COMMIT_FAILED: stderr={cr.stderr!r}")
    return cr  # CompletedProcess of commit


# ─── §7b push branch ───────────────────────────────────────────────────────
def push_branch(
    branch: str,
    runner: RunnerType,
    *,
    repo_dir: Optional[str] = None,
    allow_workspace: bool = False,
):
    """git push origin <branch>. force 금지. 실패 시 RuntimeError."""
    cwd = _resolve_safe_repo_dir(repo_dir, allow_workspace=allow_workspace)
    pargs = ["git", "push", "origin", branch]
    assert_no_forbidden_git_flags(pargs)
    assert_no_cherry_pick(pargs)
    pr = runner(pargs, cwd=cwd)
    if pr.returncode != 0:
        raise RuntimeError(f"PUSH_FAILED: stderr={pr.stderr!r}")
    return pr


# ─── §7c commit_and_push (하위 호환) ──────────────────────────────────────
def commit_and_push(
    task_id: str,
    branch: str,
    runner: RunnerType,
    *,
    repo_dir: Optional[str] = None,
    push: bool = True,
):
    """git commit -m + (선택적) git push. force 금지. 실패 시 RuntimeError."""
    cr = commit_local(task_id, runner, repo_dir=repo_dir)
    if push:
        push_branch(branch, runner, repo_dir=repo_dir)
    return cr  # CompletedProcess of commit


# ─── §8 로컬 diff 사전 검증 (push 전) ────────────────────────────────────
def precheck_local_replacement_diff(
    branch: str,
    expected_files: list[str],
    runner: RunnerType,
    *,
    repo_dir: Optional[str] = None,
    allow_workspace: bool = False,
) -> tuple[bool, list[str], list[str]]:
    """commit 후 push 전, 로컬 git diff 기반 사전 검증.

    git show --stat HEAD 또는 git diff --name-only origin/main...HEAD 호출.
    return (valid, extra, missing).
    valid=True이면 expected_files와 일치하고 forbidden path 없음.
    """
    cwd = _resolve_safe_repo_dir(repo_dir, allow_workspace=allow_workspace)
    # git diff --name-only origin/main...HEAD
    diff_args = ["git", "diff", "--name-only", "origin/main...HEAD"]
    assert_no_forbidden_git_flags(diff_args)
    assert_no_cherry_pick(diff_args)
    result = runner(diff_args, cwd=cwd)
    if result.returncode != 0:
        # fallback: git show --stat HEAD
        show_args = ["git", "show", "--stat", "--name-only", "--format=", "HEAD"]
        assert_no_forbidden_git_flags(show_args)
        assert_no_cherry_pick(show_args)
        result = runner(show_args, cwd=cwd)
        if result.returncode != 0:
            return False, [], list(expected_files)
    local_files = [line.strip() for line in (result.stdout or "").splitlines() if line.strip()]
    equal, extra, missing = compare_effective_diff(local_files, expected_files)
    # forbidden path 검사
    forbidden = detect_forbidden_paths(local_files, expected_files)
    if forbidden:
        return False, extra + forbidden, missing
    return equal, extra, missing


# ─── §9 dirty tree 사전 검사 (High #2) ────────────────────────────────────
def assert_clean_working_tree(runner: RunnerType, *, repo_dir: Optional[str] = None, allow_workspace: bool = False) -> None:
    """git status --porcelain 출력이 비어있지 않으면 RuntimeError(DIRTY_WORKING_TREE)."""
    cwd = _resolve_safe_repo_dir(repo_dir, allow_workspace=allow_workspace)
    args = ["git", "status", "--porcelain"]
    assert_no_forbidden_git_flags(args)
    result = runner(args, cwd=cwd)
    if result.returncode != 0:
        raise RuntimeError(f"DIRTY_WORKING_TREE: git status failed stderr={result.stderr!r}")
    if (result.stdout or "").strip():
        raise RuntimeError("DIRTY_WORKING_TREE: cannot proceed with replacement")


# ─── §10 replacement PR open ───────────────────────────────────────────────
def open_replacement_pr(
    task_id: str,
    branch: str,
    source_pr: int,
    runner: RunnerType,
    *,
    repo_dir: Optional[str] = None,
) -> int:
    """gh pr create. base=main, head=branch. body에 source PR 링크. return PR number.
    repo_dir은 다른 git/gh 호출과 일관되게 worktree/temp repo에 라우팅하기 위해 필요.
    """
    title = f"[{task_id}] replacement (auto for #{source_pr})"
    body = f"Auto-generated replacement for #{source_pr}. Original PR preserved (no close/delete)."
    args = ["gh", "pr", "create", "--base", "main", "--head", branch, "--title", title, "--body", body]
    assert_no_forbidden_git_flags(args)
    assert_no_cherry_pick(args)
    result = runner(args, cwd=repo_dir)
    if result.returncode != 0:
        raise RuntimeError(f"OPEN_REPLACEMENT_PR_FAILED: stderr={result.stderr!r}")
    # gh pr create는 PR URL을 stdout에 출력. 끝에서 숫자 추출.
    out = (result.stdout or "").strip()
    # gh pr create stdout은 https://github.com/<owner>/<repo>/pull/<N> 형식 — /pull/<N> 만 신뢰.
    # last-digits fallback은 task id/version과 충돌 위험이 있어 제거 (Gemini high #1).
    m = re.search(r"/pull/(\d+)\b", out)
    if not m:
        raise RuntimeError(f"OPEN_REPLACEMENT_PR_NO_PR_NUMBER: stdout={out!r}")
    return int(m.group(1))


# ─── §11 원 PR 보존 (close 절대 금지) ─────────────────────────────────────
def post_replaced_comment(source_pr: int, replacement_pr: int, runner: RunnerType):
    """gh pr comment <source_pr> -b "[REPLACED] by #<replacement_pr>". close/delete 호출 금지.
    args에 'close', 'delete', 'edit --state closed' 들어가면 즉시 RuntimeError.
    """
    body = f"[REPLACED] by #{replacement_pr} — automated by replacement_pr_runner (task-2510). Original PR preserved."
    args = ["gh", "pr", "comment", str(source_pr), "-b", body]
    # 정적 검증: close/delete/edit 토큰이 args에 들어가지 않도록
    forbidden_tokens = {"close", "delete", "delete-branch"}
    if any(t in args for t in forbidden_tokens):
        raise RuntimeError(f"ORIGINAL_PR_PRESERVE_FORBIDDEN_OP: args={args}")
    assert_no_forbidden_git_flags(args)
    assert_no_cherry_pick(args)
    result = runner(args)
    if result.returncode != 0:
        raise RuntimeError(f"POST_REPLACED_COMMENT_FAILED: stderr={result.stderr!r}")
    return result


# ─── §12 replacement diff 사후 검증 ──────────────────────────────────────
def validate_replacement_diff(
    replacement_pr: int,
    expected_files: list[str],
    runner: RunnerType,
) -> tuple[bool, list[str], list[str]]:
    """gh pr view <replacement_pr> --json files → compare_effective_diff.
    return (valid, extra, missing).
    """
    args = ["gh", "pr", "view", str(replacement_pr), "--json", "files"]
    result = runner(args)
    if result.returncode != 0:
        raise RuntimeError(f"VALIDATE_REPLACEMENT_DIFF_FAILED: pr={replacement_pr} stderr={result.stderr!r}")
    data = json.loads(result.stdout or "{}")
    effective = [f["path"] for f in (data.get("files") or []) if f.get("path")]
    return compare_effective_diff(effective, expected_files)


# ─── §13 EscalationPacket helper ────────────────────────────────────────
def build_escalation_packet(
    task_id: str,
    pr_number: int,
    escalation_type: CriticalEscalationType,
    reason: str,
    evidence: dict,
) -> EscalationPacket:
    """본 task에서는 EscalationPacket dataclass 인스턴스만 생성. 실제 보고는 task-2513 영역."""
    return EscalationPacket(
        task_id=task_id,
        pr_number=pr_number,
        escalation_type=escalation_type,
        reason=reason,
        why_auto_cannot_continue=f"replacement_pr_runner cannot continue without chair: {reason}",
        safe_options=[
            "회장 수동 검토 후 재개",
            "원 PR 보존 + clean branch 수동 생성",
            "expected_files 재정의",
        ],
        recommended_option="회장 수동 검토 후 재개",
        evidence=evidence,
    )


# ─── §14 main runner 클래스 ────────────────────────────────────────────────
class ReplacementPRRunner:
    """contaminated PR을 받아 replacement PR을 자동 생성하는 main entry."""

    def __init__(
        self,
        runner: Optional[RunnerType] = None,
        *,
        dry_run: bool = False,
        allow_workspace: bool = False,
        repo_dir: Optional[str] = None,
        extra_forbidden_patterns: Optional[list] = None,
        timestamp_provider: Optional[Callable[[], str]] = None,
    ):
        self.runner = runner or _default_runner
        self.dry_run = dry_run
        self.allow_workspace = allow_workspace
        self.repo_dir = repo_dir
        self.extra_forbidden_patterns = extra_forbidden_patterns
        self._ts_provider = timestamp_provider or (
            lambda: datetime.now(timezone.utc).strftime("%Y%m%d%H%M%S")
        )
        self.last_escalation_packet: Optional[EscalationPacket] = None

    def _record_escalation(
        self,
        task_id: str,
        pr_number: int,
        escalation_type: CriticalEscalationType,
        reason: str,
        evidence: dict,
    ) -> None:
        """실패 경로에서 EscalationPacket을 생성하여 last_escalation_packet에 보관."""
        try:
            self.last_escalation_packet = build_escalation_packet(
                task_id=task_id,
                pr_number=pr_number,
                escalation_type=escalation_type,
                reason=reason,
                evidence=evidence,
            )
        except Exception as e:
            logger.warning("build_escalation_packet failed: %s", e)

    def execute(self, pr_number: int, task_spec: Optional[TaskSpec] = None) -> ReplacementResult:
        self.last_escalation_packet = None  # 매 실행마다 초기화

        # Step 1: PR metadata
        if self.dry_run:
            pr_meta = self._dry_run_pr_meta(pr_number, task_spec)
        else:
            try:
                pr_meta = fetch_pr_metadata(pr_number, self.runner)
            except Exception:
                task_id = (task_spec.task_id if task_spec else "unknown") or "unknown"
                self._record_escalation(
                    task_id=task_id,
                    pr_number=pr_number,
                    escalation_type=CriticalEscalationType.REPLACEMENT_PR_AUTO_CREATION_FAILED_FOR_CONTAMINATED_DIFF,
                    reason="FETCH_PR_METADATA_FAILED",
                    evidence={"pr_number": pr_number},
                )
                # Gemini medium: 원 PR에 어떤 변경도 가하지 않았으므로 preserved=True
                return ReplacementResult(
                    source_pr=pr_number, replacement_pr=None, original_pr_preserved=True,
                    expected_files=[], effective_diff_files=[], forbidden_paths=[],
                    success=False,
                    failure_reason=CriticalEscalationType.REPLACEMENT_PR_AUTO_CREATION_FAILED_FOR_CONTAMINATED_DIFF.value,
                )
        # Step 2: task_spec / expected_files
        if task_spec is not None:
            expected_files = list(task_spec.expected_files)
            task_id = task_spec.task_id
        else:
            task_id = pr_meta.get("task_id") or "unknown"
            expected_files = []
        # Step 3: effective diff
        if self.dry_run:
            effective_files = list(pr_meta.get("files", []))
        else:
            try:
                effective_files = compute_effective_diff(pr_meta, self.runner)
            except Exception:
                self._record_escalation(
                    task_id=task_id,
                    pr_number=pr_number,
                    escalation_type=CriticalEscalationType.REPLACEMENT_PR_AUTO_CREATION_FAILED_FOR_CONTAMINATED_DIFF,
                    reason="COMPUTE_EFFECTIVE_DIFF_FAILED",
                    evidence={"pr_number": pr_number, "task_id": task_id},
                )
                # Gemini medium: 원 PR에 어떤 변경도 가하지 않았으므로 preserved=True
                return ReplacementResult(
                    source_pr=pr_number, replacement_pr=None, original_pr_preserved=True,
                    expected_files=expected_files, effective_diff_files=[], forbidden_paths=[],
                    success=False,
                    failure_reason=CriticalEscalationType.REPLACEMENT_PR_AUTO_CREATION_FAILED_FOR_CONTAMINATED_DIFF.value,
                )
        # Step 4: contamination
        c = detect_contamination(effective_files, expected_files, extra_forbidden_patterns=self.extra_forbidden_patterns)
        # Step 5: forbidden path
        if c["forbidden_paths"]:
            self._record_escalation(
                task_id=task_id,
                pr_number=pr_number,
                escalation_type=CriticalEscalationType.FORBIDDEN_PATH_INTRUSION,
                reason="FORBIDDEN_PATH_DETECTED",
                evidence={"forbidden_paths": c["forbidden_paths"], "task_id": task_id},
            )
            return ReplacementResult(
                source_pr=pr_number, replacement_pr=None, original_pr_preserved=True,
                expected_files=expected_files, effective_diff_files=effective_files,
                forbidden_paths=c["forbidden_paths"],
                success=False,
                failure_reason=CriticalEscalationType.FORBIDDEN_PATH_INTRUSION.value,
            )
        # Step 6: clean → no-op
        if not c["contaminated"]:
            return ReplacementResult(
                source_pr=pr_number, replacement_pr=None, original_pr_preserved=True,
                expected_files=expected_files, effective_diff_files=effective_files,
                forbidden_paths=[],
                success=True, failure_reason=None,
            )
        # Step 7: contaminated → replacement
        if self.dry_run:
            return self._dry_run_replacement(
                pr_number=pr_number, task_id=task_id,
                expected_files=expected_files, effective_files=effective_files,
                forbidden_paths=c["forbidden_paths"],
            )

        # ★ task-2784 P1: safe repo_dir invariant 선검사 — canonical WORKSPACE 직접 write 차단.
        # invariant 위반(MISSING_REPO_DIR / WORKSPACE_WRITE_FORBIDDEN / REPO_DIR_NOT_GIT_REPO)은
        # execute() 밖으로 raise 전파되어야 merge_queue W2의 except fallthrough가 잡아 기존 동작(회귀 0)을
        # 보존한다. (return success=False 금지)
        # 반환된 safe git top-level을 self.repo_dir에 재할당하여 helper를 직접 호출하지 않는 후속 경로
        # (create_clean_replacement_branch / open_replacement_pr 등)도 정규화된 repo root를 쓰게 한다.
        self.repo_dir = _resolve_safe_repo_dir(self.repo_dir, allow_workspace=self.allow_workspace)

        # 실제 모드 — High #2: pre-flight dirty tree check
        try:
            assert_clean_working_tree(self.runner, repo_dir=self.repo_dir, allow_workspace=self.allow_workspace)
        except RuntimeError as e:
            self._record_escalation(
                task_id=task_id,
                pr_number=pr_number,
                escalation_type=CriticalEscalationType.REPLACEMENT_PR_AUTO_CREATION_FAILED_FOR_CONTAMINATED_DIFF,
                reason=str(e),
                evidence={"dirty_tree": True, "task_id": task_id},
            )
            return ReplacementResult(
                source_pr=pr_number, replacement_pr=None, original_pr_preserved=True,
                expected_files=expected_files, effective_diff_files=effective_files,
                forbidden_paths=[],
                success=False,
                failure_reason=CriticalEscalationType.REPLACEMENT_PR_AUTO_CREATION_FAILED_FOR_CONTAMINATED_DIFF.value,
            )

        # High #1: commit 후 push 전 사전 검증 흐름
        branch = None
        try:
            branch = create_clean_replacement_branch(task_id, self.runner, timestamp=self._ts_provider(), repo_dir=self.repo_dir)
            transplant_expected_files(expected_files, pr_meta["head_sha"], self.runner, repo_dir=self.repo_dir, allow_workspace=self.allow_workspace)
            # commit (push X)
            commit_local(task_id, self.runner, repo_dir=self.repo_dir, allow_workspace=self.allow_workspace)
        except Exception as e:
            self._record_escalation(
                task_id=task_id,
                pr_number=pr_number,
                escalation_type=CriticalEscalationType.REPLACEMENT_PR_AUTO_CREATION_FAILED_FOR_CONTAMINATED_DIFF,
                reason=str(e),
                evidence={"branch": branch, "task_id": task_id},
            )
            return ReplacementResult(
                source_pr=pr_number, replacement_pr=None, original_pr_preserved=True,
                expected_files=expected_files, effective_diff_files=effective_files,
                forbidden_paths=[],
                success=False,
                failure_reason=CriticalEscalationType.REPLACEMENT_PR_AUTO_CREATION_FAILED_FOR_CONTAMINATED_DIFF.value,
            )

        # ★ NEW: precheck_local_replacement_diff — push 전 사전 검증
        try:
            precheck_valid, precheck_extra, precheck_missing = precheck_local_replacement_diff(
                branch, expected_files, self.runner, repo_dir=self.repo_dir, allow_workspace=self.allow_workspace
            )
        except Exception:
            precheck_valid, precheck_extra, precheck_missing = False, [], []

        if not precheck_valid:
            # push 하지 않았으므로 local branch만 삭제
            try:
                del_args = ["git", "branch", "-d", branch]
                assert_no_forbidden_git_flags(del_args)
                self.runner(del_args, cwd=self.repo_dir)
            except Exception:
                pass
            self._record_escalation(
                task_id=task_id,
                pr_number=pr_number,
                escalation_type=CriticalEscalationType.REPLACEMENT_PR_FAILED,
                reason="PRECHECK_LOCAL_DIFF_MISMATCH",
                evidence={
                    "branch": branch,
                    "extra": precheck_extra,
                    "missing": precheck_missing,
                    "task_id": task_id,
                },
            )
            return ReplacementResult(
                source_pr=pr_number, replacement_pr=None, original_pr_preserved=True,
                expected_files=expected_files, effective_diff_files=effective_files,
                forbidden_paths=[],
                success=False,
                failure_reason=CriticalEscalationType.REPLACEMENT_PR_FAILED.value,
            )

        # 사전 검증 통과 → push + PR open
        try:
            push_branch(branch, self.runner, repo_dir=self.repo_dir, allow_workspace=self.allow_workspace)
            replacement_pr = open_replacement_pr(task_id, branch, pr_number, self.runner, repo_dir=self.repo_dir)
        except Exception as e:
            self._record_escalation(
                task_id=task_id,
                pr_number=pr_number,
                escalation_type=CriticalEscalationType.REPLACEMENT_PR_AUTO_CREATION_FAILED_FOR_CONTAMINATED_DIFF,
                reason=str(e),
                evidence={"branch": branch, "task_id": task_id},
            )
            return ReplacementResult(
                source_pr=pr_number, replacement_pr=None, original_pr_preserved=True,
                expected_files=expected_files, effective_diff_files=effective_files,
                forbidden_paths=[],
                success=False,
                failure_reason=CriticalEscalationType.REPLACEMENT_PR_AUTO_CREATION_FAILED_FOR_CONTAMINATED_DIFF.value,
            )

        # Step 8: validate replacement diff (사후 검증)
        try:
            valid_result = validate_replacement_diff(replacement_pr, expected_files, self.runner)
            valid = valid_result if isinstance(valid_result, bool) else valid_result[0]
        except Exception:
            valid = False
        if not valid:
            self._record_escalation(
                task_id=task_id,
                pr_number=pr_number,
                escalation_type=CriticalEscalationType.REPLACEMENT_PR_FAILED,
                reason="VALIDATE_REPLACEMENT_DIFF_FAILED",
                evidence={"replacement_pr": replacement_pr, "task_id": task_id},
            )
            return ReplacementResult(
                source_pr=pr_number, replacement_pr=replacement_pr, original_pr_preserved=True,
                expected_files=expected_files, effective_diff_files=effective_files,
                forbidden_paths=[],
                success=False,
                failure_reason=CriticalEscalationType.REPLACEMENT_PR_FAILED.value,
            )
        # Step 9: 원 PR 보존 코멘트
        try:
            post_replaced_comment(pr_number, replacement_pr, self.runner)
        except Exception as e:
            self._record_escalation(
                task_id=task_id,
                pr_number=pr_number,
                escalation_type=CriticalEscalationType.REPLACEMENT_PR_FAILED,
                reason=str(e),
                evidence={"replacement_pr": replacement_pr, "task_id": task_id},
            )
            # Gemini medium: post_replaced_comment 실패해도 원 PR은 close/delete되지 않음 → preserved=True
            return ReplacementResult(
                source_pr=pr_number, replacement_pr=replacement_pr, original_pr_preserved=True,
                expected_files=expected_files, effective_diff_files=effective_files,
                forbidden_paths=[],
                success=False,
                failure_reason=CriticalEscalationType.REPLACEMENT_PR_FAILED.value,
            )
        return ReplacementResult(
            source_pr=pr_number, replacement_pr=replacement_pr, original_pr_preserved=True,
            expected_files=expected_files, effective_diff_files=effective_files,
            forbidden_paths=[],
            success=True, failure_reason=None,
        )

    def _dry_run_pr_meta(self, pr_number, task_spec):
        task_id = task_spec.task_id if task_spec else "unknown"
        return {
            "head_ref": f"task/{task_id}-dry-run",
            "head_sha": "dry-run-sha-0000000",
            "base_ref": "main",
            "files": list(task_spec.expected_files) if task_spec else [],
            "task_id": task_id,
            "title": f"[DRY-RUN] {task_id}",
            "number": pr_number,
        }

    def _dry_run_replacement(self, *, pr_number, task_id, expected_files, effective_files, forbidden_paths):
        timestamp = self._ts_provider()
        simulated_branch = f"task/{task_id}-replacement-{timestamp}"
        logger.info("[DRY-RUN] Would create branch=%r and replacement PR for PR #%d", simulated_branch, pr_number)
        return ReplacementResult(
            source_pr=pr_number, replacement_pr=None, original_pr_preserved=True,
            expected_files=expected_files, effective_diff_files=effective_files,
            forbidden_paths=forbidden_paths,
            success=True, failure_reason=None,
        )


# ─── §15 CLI ──────────────────────────────────────────────────────────────
def main(argv: Optional[list[str]] = None) -> int:
    parser = argparse.ArgumentParser(description="task-2510 replacement_pr_runner CLI")
    parser.add_argument("--pr", type=int, required=True)
    parser.add_argument("--dry-run", action="store_true")
    parser.add_argument("--allow-workspace", action="store_true", help="CLI 수동 모드에서만 canonical WORKSPACE write 허용(automation 금지)")
    parser.add_argument("--task-file", type=str, default=None)
    args = parser.parse_args(argv)
    runner = ReplacementPRRunner(dry_run=args.dry_run, allow_workspace=args.allow_workspace)
    spec = None
    if args.task_file:
        spec = load_task_spec(Path(args.task_file))
    result = runner.execute(args.pr, task_spec=spec)
    print(json.dumps(asdict(result), default=str, ensure_ascii=False, indent=2))
    if runner.last_escalation_packet:
        print(json.dumps({"escalation_packet": asdict(runner.last_escalation_packet)}, default=str), file=sys.stderr)
    return 0 if result.success else 2


if __name__ == "__main__":
    sys.exit(main())