#!/usr/bin/env python3
"""lock_in_verify.py — Lock-in 1 First-line 가드 자동 검증 (CI 강제용).

task-2439에서 도입한 "First-line guard" 패턴이 머지 경로 함수의 *첫 statement*로
유지되는지 AST 기반으로 검사한다. 위반 시 exit_code=1로 CI를 차단한다.

검사 대상:
  1) scripts/anu_confirm_bot/main.py::_execute_approve
       - 첫 문장: cancelled 마커 Path() 할당
       - 그 다음: cancelled .exists() 가드 + early return
       - guard.sh subprocess.run(...)이 *모든* gh pr merge 호출보다 line-number 기준 먼저
  2) scripts/auto_merge.py::AutoMerger.execute_merge
       - 동일 패턴, merge 시그너처는 worktree_manager finish 호출

task-3016 (2026-08-25): 8건 PR이 taskctl 정본 경로를 통째로 우회해
`GH_TOKEN=... gh api -X PUT repos/<owner>/<repo>/pulls/<n>/merge -f merge_method=squash`
로 직접 머지된 사고 후속. 위 First-line 가드는 특정 2개 함수 *내부* 구조만
검사하며, "그 2개 함수 밖에서 머지 호출을 새로 심는" 것 자체는 잡지 못한다.
이를 보강하기 위해 census_merge_invocation_sites() 를 추가한다 — scripts/**,
utils/** 전수에서 "gh pr merge" / "gh api ... PUT <repo>/pulls/<n>/merge" 호출 지점을 찾고,
scripts/taskctl.py::cmd_merge 를 제외한 모든 지점을 위반으로 보고한다.

★ ACTIVE=false 원칙: 이 census 는 기본적으로 위반을 발견해도 stderr 경고만
출력하고 exit 0 이다. `LOCK_IN_MERGE_CENSUS_ACTIVE=1` 환경변수가 설정된
경우에만 위반 시 exit 1 로 CI를 차단한다. 기존 2개 First-line 가드 검사의
동작(CHECKS 루프)은 이 변경으로 전혀 바뀌지 않는다.
"""

from __future__ import annotations

import argparse
import ast
import os
import sys
from pathlib import Path

WORKSPACE = Path(__file__).resolve().parent.parent

# task-3016: 머지 실행 지점 전수조사(census) allowlist.
# "file::func" 형태만 정본으로 인정한다. 그 외 위치에서 발견되는 gh pr merge /
# gh api PUT .../merge 호출은 전부 위반으로 보고된다.
MERGE_CENSUS_ALLOWLIST = {
    ("scripts/taskctl.py", "cmd_merge"),
}

CHECKS = [
    {
        # task-2449 Fix 5: gh pr merge 직접 호출 → taskctl merge 라우팅으로 변경.
        # First-line 가드(cancelled / guard.sh)는 보존되며, 실제 머지 subprocess는
        # taskctl.py를 호출한다. token은 변수명/리터럴에서 동시에 잡히도록 ("taskctl", "merge").
        "file": "scripts/anu_confirm_bot/main.py",
        "func": "_execute_approve",
        "merge_signature_tokens": ("taskctl", "merge"),
    },
    {
        "file": "scripts/auto_merge.py",
        "func": "execute_merge",
        "merge_signature_tokens": ("worktree_manager", "finish"),
    },
]


def _find_function(tree: ast.AST, name: str) -> ast.FunctionDef | None:
    for node in ast.walk(tree):
        if isinstance(node, (ast.FunctionDef, ast.AsyncFunctionDef)) and node.name == name:
            return node  # type: ignore[return-value]
    return None


def _strip_docstring(body: list[ast.stmt]) -> list[ast.stmt]:
    if body and isinstance(body[0], ast.Expr) and isinstance(body[0].value, ast.Constant) and isinstance(body[0].value.value, str):
        return body[1:]
    return body


def _is_cancelled_path_assign(stmt: ast.stmt) -> bool:
    if not isinstance(stmt, ast.Assign):
        return False
    if not stmt.targets or not isinstance(stmt.targets[0], ast.Name):
        return False
    if "cancelled" not in stmt.targets[0].id.lower():
        return False
    src = ast.unparse(stmt.value)
    return ".cancelled" in src and "memory" in src and "events" in src


def _is_cancelled_exists_guard(stmt: ast.stmt) -> bool:
    if not isinstance(stmt, ast.If):
        return False
    test_src = ast.unparse(stmt.test)
    if "cancelled" not in test_src.lower() or ".exists()" not in test_src:
        return False
    if not stmt.body:
        return False
    first = stmt.body[0]
    return isinstance(first, (ast.Return, ast.Raise))


def _is_guard_sh_subprocess(call: ast.Call, guard_var_names: set[str]) -> bool:
    func_src = ast.unparse(call.func)
    if func_src != "subprocess.run":
        return False
    src = ast.unparse(call)
    if "guard.sh" in src:
        return True
    return any(name in src for name in guard_var_names)


def _collect_guard_var_names(func: ast.FunctionDef) -> set[str]:
    names: set[str] = set()
    for node in ast.walk(func):
        if isinstance(node, ast.Assign) and len(node.targets) == 1 and isinstance(node.targets[0], ast.Name):
            value_src = ast.unparse(node.value)
            if "guard.sh" in value_src:
                names.add(node.targets[0].id)
    return names


def _list_contains_tokens(node: ast.AST, tokens: tuple[str, ...]) -> bool:
    """Find any list literal or call whose string-literal contents include all tokens."""
    if isinstance(node, (ast.List, ast.Tuple)):
        joined = " ".join(
            elt.value for elt in node.elts if isinstance(elt, ast.Constant) and isinstance(elt.value, str)
        )
        if all(t in joined for t in tokens):
            return True
    if isinstance(node, ast.Call):
        for arg in node.args:
            if _list_contains_tokens(arg, tokens):
                return True
    return False


def _find_first_guard_sh_lineno(func: ast.FunctionDef) -> int | None:
    guard_vars = _collect_guard_var_names(func)
    earliest: int | None = None
    for node in ast.walk(func):
        if isinstance(node, ast.Call) and _is_guard_sh_subprocess(node, guard_vars):
            if earliest is None or node.lineno < earliest:
                earliest = node.lineno
    return earliest


def _find_merge_call_linenos(func: ast.FunctionDef, tokens: tuple[str, ...]) -> list[int]:
    """Locate each subprocess.run(cmd, ...) where cmd resolves to a list containing tokens.

    Tokens may be present either as direct string literals in the list, or via
    Path()-assigned variables whose value source-text contains the token. We
    collect both kinds and treat any list element matching either as a hit.
    """
    token_var_names: set[str] = set()
    for node in ast.walk(func):
        if isinstance(node, ast.Assign) and len(node.targets) == 1 and isinstance(node.targets[0], ast.Name):
            value_src = ast.unparse(node.value)
            if all(t in value_src for t in tokens):
                token_var_names.add(node.targets[0].id)

    def list_matches(node: ast.AST) -> bool:
        if isinstance(node, (ast.List, ast.Tuple)):
            joined_strs: list[str] = []
            joined_names: list[str] = []
            for elt in node.elts:
                if isinstance(elt, ast.Constant) and isinstance(elt.value, str):
                    joined_strs.append(elt.value)
                else:
                    joined_names.append(ast.unparse(elt))
            joined = " ".join(joined_strs)
            joined_var = " ".join(joined_names)
            combined = joined + " " + joined_var
            if all(t in combined for t in tokens):
                return True
            if any(name in joined_var for name in token_var_names):
                return True
        return False

    list_assignments: dict[str, int] = {}
    merge_lines: list[int] = []
    for node in ast.walk(func):
        if isinstance(node, ast.Assign):
            value = node.value
            if isinstance(value, (ast.List, ast.Tuple)) and len(node.targets) == 1 and isinstance(node.targets[0], ast.Name):
                if list_matches(value):
                    list_assignments[node.targets[0].id] = node.lineno
        if isinstance(node, ast.Call):
            if ast.unparse(node.func) != "subprocess.run" or not node.args:
                continue
            first_arg = node.args[0]
            if isinstance(first_arg, (ast.List, ast.Tuple)) and list_matches(first_arg):
                merge_lines.append(node.lineno)
            elif isinstance(first_arg, ast.Name) and first_arg.id in list_assignments:
                merge_lines.append(node.lineno)
    return sorted(merge_lines)


def verify_function(file_path: Path, func_name: str, merge_tokens: tuple[str, ...]) -> list[str]:
    errors: list[str] = []
    if not file_path.exists():
        return [f"{file_path}: file not found"]
    src = file_path.read_text(encoding="utf-8")
    tree = ast.parse(src)
    func = _find_function(tree, func_name)
    if func is None:
        return [f"{file_path}::{func_name}: function not found"]
    body = _strip_docstring(list(func.body))
    if len(body) < 3:
        return [f"{file_path}::{func_name}: body too short ({len(body)} stmts)"]
    if not _is_cancelled_path_assign(body[0]):
        errors.append(f"{file_path}::{func_name}: 첫 statement가 cancelled 마커 Path 할당이 아님 — got {ast.unparse(body[0])[:120]}")
    if not _is_cancelled_exists_guard(body[1]):
        errors.append(f"{file_path}::{func_name}: 두 번째 statement가 cancelled.exists() 가드+return/raise 가 아님 — got {ast.unparse(body[1])[:120]}")
    guard_lineno = _find_first_guard_sh_lineno(func)
    if guard_lineno is None:
        errors.append(f"{file_path}::{func_name}: guard.sh subprocess.run 호출 없음")
    merge_linenos = _find_merge_call_linenos(func, merge_tokens)
    if not merge_linenos:
        errors.append(f"{file_path}::{func_name}: merge subprocess({merge_tokens}) 호출 없음")
    if guard_lineno is not None and merge_linenos and guard_lineno >= merge_linenos[0]:
        errors.append(
            f"{file_path}::{func_name}: guard.sh line({guard_lineno})이 merge subprocess line({merge_linenos[0]})보다 늦음 — Lock-in 위반"
        )
    return errors


# ---------------------------------------------------------------------------
# task-3016: 머지 실행 지점 전수조사(census)
# ---------------------------------------------------------------------------

_CENSUS_TARGET_DIRS = ("scripts", "utils")
_CENSUS_SELF_PATH = Path(__file__).resolve()


def _census_skip_file(rel_parts: tuple[str, ...], filename: str) -> bool:
    """census 스캔에서 제외할 파일 판정 — 자기 자신 + 테스트 파일."""
    if "__tests__" in rel_parts:
        return True
    if filename.startswith("test_") or filename.endswith("_test.py"):
        return True
    return False


def _iter_census_files(workspace: Path) -> list[Path]:
    files: list[Path] = []
    for base in _CENSUS_TARGET_DIRS:
        root = workspace / base
        if not root.exists():
            continue
        for p in sorted(root.rglob("*.py")):
            if p.resolve() == _CENSUS_SELF_PATH:
                continue
            rel = p.relative_to(workspace)
            if _census_skip_file(rel.parts, p.name):
                continue
            files.append(p)
    return files


def _elt_to_partial_str(elt: ast.AST) -> str | None:
    """리스트 원소 하나를 (부분) 문자열로 환원.

    일반 문자열 리터럴은 그대로, f-string(JoinedStr)은 상수 부분만 이어붙여
    부분 문자열로 취급한다 — 예: f"repos/x/pulls/{n}/merge" → "repos/x/pulls//merge".
    보간되는 값(FormattedValue)은 알 수 없으므로 빈 문자열로 취급하되, 앞뒤에
    붙는 리터럴 텍스트(예: "/merge" 접미부)는 여전히 토큰 매칭에 쓸 수 있다.
    """
    if isinstance(elt, ast.Constant) and isinstance(elt.value, str):
        return elt.value
    if isinstance(elt, ast.JoinedStr):
        parts: list[str] = []
        has_const = False
        for seg in elt.values:
            if isinstance(seg, ast.Constant) and isinstance(seg.value, str):
                parts.append(seg.value)
                has_const = True
            else:
                parts.append("")
        return "".join(parts) if has_const else None
    return None


def _resolve_str_list(node: ast.AST, list_vars: dict[str, list[str | None]]) -> list[str | None] | None:
    """리스트/튜플 리터럴이거나 그런 리터럴이 할당된 변수명이면 문자열 원소 목록을 반환."""
    if isinstance(node, (ast.List, ast.Tuple)):
        return [_elt_to_partial_str(elt) for elt in node.elts]
    if isinstance(node, ast.Name) and node.id in list_vars:
        return list_vars[node.id]
    return None


def _argv_has_gh_pr_merge(strs: list[str]) -> bool:
    return "gh" in strs and "pr" in strs and "merge" in strs


def _argv_has_gh_api_put_merge(strs: list[str]) -> bool:
    if "gh" not in strs or "api" not in strs:
        return False
    has_x_flag = ("-X" in strs) or ("--method" in strs)
    has_put_value = "PUT" in strs
    has_merge_path = any(s and "/merge" in s for s in strs)
    return has_x_flag and has_put_value and has_merge_path


_EXECUTION_CALL_HINTS = ("run", "call", "exec", "invoke", "popen")


def _looks_like_execution_call(node: ast.Call) -> bool:
    """이 Call 이 실제로 프로세스를 실행하는 호출처럼 보이는지 판정.

    subprocess.run / _run / runner(...) / check_call 등은 잡되, 같은 argv
    리스트를 검사만 하는 assert_no_forbidden_git_flags(args) 같은 헬퍼는
    걸러내 census 위반 목록의 중복 잡음을 줄인다.
    """
    func_src = ast.unparse(node.func).lower()
    return any(hint in func_src for hint in _EXECUTION_CALL_HINTS)


def _classify_call_kind(node: ast.Call, list_vars: dict[str, list[str | None]]) -> str | None:
    """Call 노드의 인자에서 gh pr merge / gh api PUT <repo>/pulls/<n>/merge 패턴을 찾는다."""
    if not _looks_like_execution_call(node):
        return None
    candidates: list[ast.AST] = list(node.args)
    for kw in node.keywords:
        if kw.arg in ("args", "cmd") and kw.value is not None:
            candidates.append(kw.value)
    for arg in candidates:
        str_list = _resolve_str_list(arg, list_vars)
        if str_list is not None:
            strs = [s for s in str_list if s]
            if _argv_has_gh_pr_merge(strs):
                return "gh_pr_merge"
            if _argv_has_gh_api_put_merge(strs):
                return "gh_api_put_merge"
        elif isinstance(arg, ast.Constant) and isinstance(arg.value, str):
            s = arg.value
            if "gh pr merge" in s:
                return "gh_pr_merge"
            if "gh api" in s and "PUT" in s and "/merge" in s:
                return "gh_api_put_merge"
    return None


class _MergeCallCensusVisitor(ast.NodeVisitor):
    """파일 하나를 훑으며 함수 스코프를 추적, gh 머지 호출 지점을 수집."""

    def __init__(self, rel_path: str) -> None:
        self.rel_path = rel_path
        self._func_stack: list[str] = []
        self._list_vars: dict[str, list[str | None]] = {}
        self.hits: list[dict] = []

    def _current_func(self) -> str:
        return self._func_stack[-1] if self._func_stack else "<module>"

    def visit_FunctionDef(self, node: ast.FunctionDef) -> None:  # noqa: N802
        self._func_stack.append(node.name)
        self.generic_visit(node)
        self._func_stack.pop()

    def visit_AsyncFunctionDef(self, node: ast.AsyncFunctionDef) -> None:  # noqa: N802
        self._func_stack.append(node.name)
        self.generic_visit(node)
        self._func_stack.pop()

    def visit_Assign(self, node: ast.Assign) -> None:  # noqa: N802
        if len(node.targets) == 1 and isinstance(node.targets[0], ast.Name) and isinstance(node.value, (ast.List, ast.Tuple)):
            elems = [_elt_to_partial_str(elt) for elt in node.value.elts]
            self._list_vars[node.targets[0].id] = elems
        self.generic_visit(node)

    def visit_Call(self, node: ast.Call) -> None:  # noqa: N802
        kind = _classify_call_kind(node, self._list_vars)
        if kind is not None:
            self.hits.append(
                {
                    "file": self.rel_path,
                    "line": node.lineno,
                    "func": self._current_func(),
                    "kind": kind,
                }
            )
        self.generic_visit(node)


def census_merge_invocation_sites(workspace: Path) -> list[dict]:
    """scripts/**/*.py, utils/**/*.py 전수에서 gh pr merge 실행 지점을 찾는다.

    allowlist(scripts/taskctl.py::cmd_merge)를 제외한 모든 발견은 위반으로
    반환한다. 반환 항목: {"file", "line", "func", "kind"}.
    """
    workspace = workspace.resolve()
    violations: list[dict] = []
    for path in _iter_census_files(workspace):
        rel_path = str(path.relative_to(workspace))
        try:
            src = path.read_text(encoding="utf-8")
            tree = ast.parse(src, filename=rel_path)
        except (SyntaxError, UnicodeDecodeError, OSError):
            continue
        visitor = _MergeCallCensusVisitor(rel_path)
        visitor.visit(tree)
        for hit in visitor.hits:
            if (hit["file"], hit["func"]) in MERGE_CENSUS_ALLOWLIST:
                continue
            violations.append(hit)
    return violations


# ---------------------------------------------------------------------------
# task-3016: 순수 판정 함수 — 부작용 없음, 래퍼/훅에서 재사용 가능
# ---------------------------------------------------------------------------

_WRITE_METHODS = ("PUT", "POST", "PATCH", "DELETE")


def classify_merge_invocation(argv: list[str], env: dict) -> dict:
    """argv 가 머지 호출인지 판정하고, 정본 경로(taskctl) 여부를 검사한다.

    - gh pr merge / gh api -X PUT <repo>/pulls/<n>/merge 가 아니면 is_merge=False, verdict=ALLOW
    - TASKCTL_INVOKED == "1" (엄격 일치) → ALLOW
    - TASKCTL_BYPASS == "1" (엄격 일치) → BYPASS_AUDITED (차단하지 않음 — 회장 비상구.
      감사기록 필요를 알리는 표시일 뿐 여기서 감사기록을 강제하지는 않는다 —
      그건 lifecycle_guards.check_bypass_audit()의 책임이다)
    - 그 외 → BLOCK_OFF_PATH

    ★ "정상 포맷의 위조값"(TASKCTL_INVOKED="0"/"true"/" 1"/"1 " 등)은 엄격
    문자열 일치("1")가 아니므로 자동으로 BLOCK_OFF_PATH 가 된다.
    """
    strs = [a for a in argv if isinstance(a, str)]
    is_merge = _argv_has_gh_pr_merge(strs) or _argv_has_gh_api_put_merge(strs)
    if not is_merge:
        return {"is_merge": False, "verdict": "ALLOW", "reason": "머지 호출 아님 — 게이트 미적용"}

    if env.get("TASKCTL_INVOKED") == "1":
        return {"is_merge": True, "verdict": "ALLOW", "reason": "TASKCTL_INVOKED=1 — taskctl 정본 경로"}
    if env.get("TASKCTL_BYPASS") == "1":
        return {
            "is_merge": True,
            "verdict": "BYPASS_AUDITED",
            "reason": "TASKCTL_BYPASS=1 — 회장 비상구, 차단하지 않음(감사기록 필요)",
        }
    return {
        "is_merge": True,
        "verdict": "BLOCK_OFF_PATH",
        "reason": "TASKCTL_INVOKED/TASKCTL_BYPASS 둘 다 없음 — taskctl 정본 경로 우회",
    }


def detect_token_fallback(argv: list[str], env: dict) -> dict:
    """쓰기성 gh 호출인데 GH_TOKEN/GITHUB_TOKEN 이 없으면 회장 개인 PAT 폴백 위험.

    - ALLOW_OWNER_PAT=1 이면(env) 회장 본인의 의도적 사용으로 보고 통과시킨다
      (verdict=ALLOWED_EXPLICIT) — 이 플래그가 있으면 토큰 유무와 무관하게 통과.
    - 읽기 호출(GET 등, 쓰기 플래그 없음)은 is_write=False, verdict=OK.
    """
    strs = [a for a in argv if isinstance(a, str)]
    has_write_method = any(
        strs[i] in ("-X", "--method") and i + 1 < len(strs) and strs[i + 1] in _WRITE_METHODS
        for i in range(len(strs))
    )
    is_pr_merge = _argv_has_gh_pr_merge(strs)
    is_push = "push" in strs and ("git" in strs or "push" in strs)
    is_write = has_write_method or is_pr_merge or is_push

    if not is_write:
        return {"is_write": False, "verdict": "OK", "reason": "읽기 전용 gh 호출 — 토큰 폴백 위험 없음"}

    if env.get("ALLOW_OWNER_PAT") == "1":
        return {
            "is_write": True,
            "verdict": "ALLOWED_EXPLICIT",
            "reason": "ALLOW_OWNER_PAT=1 — 회장 본인의 명시적 PAT 사용, 허용",
        }

    gh_token = env.get("GH_TOKEN") or ""
    github_token = env.get("GITHUB_TOKEN") or ""
    if gh_token.strip() or github_token.strip():
        return {"is_write": True, "verdict": "OK", "reason": "GH_TOKEN/GITHUB_TOKEN 존재 — 토큰 폴백 아님"}

    return {
        "is_write": True,
        "verdict": "OWNER_PAT_FALLBACK_RISK",
        "reason": "쓰기성 gh 호출인데 GH_TOKEN/GITHUB_TOKEN 둘 다 없음 — 회장 개인 PAT(gh 기본 인증) 폴백 위험",
    }


def _print_merge_census(violations: list[dict]) -> None:
    if not violations:
        print("MERGE-CENSUS  위반 없음 — 모든 gh pr merge/gh api PUT <repo>/pulls/<n>/merge 호출이 allowlist 내부")
        return
    print(f"MERGE-CENSUS  위반 {len(violations)}건 발견:", file=sys.stderr)
    for v in violations:
        print(f"  - {v['file']}:{v['line']} ({v['func']}) [{v['kind']}]", file=sys.stderr)


def main() -> int:
    parser = argparse.ArgumentParser(description="Lock-in First-line 가드 검증")
    parser.add_argument("--workspace", default=str(WORKSPACE))
    parser.add_argument("--quiet", action="store_true")
    parser.add_argument(
        "--merge-census",
        action="store_true",
        help="task-3016: gh pr merge / gh api PUT <repo>/pulls/<n>/merge 전수조사 실행 "
        "(기본 ACTIVE=false — LOCK_IN_MERGE_CENSUS_ACTIVE=1 이어야 위반 시 exit 1)",
    )
    args = parser.parse_args()
    ws = Path(args.workspace).resolve()

    if args.merge_census:
        violations = census_merge_invocation_sites(ws)
        if not args.quiet:
            _print_merge_census(violations)
        active = os.environ.get("LOCK_IN_MERGE_CENSUS_ACTIVE") == "1"
        if violations and active:
            return 1
        if violations and not active and not args.quiet:
            print(
                "MERGE-CENSUS  경고만 출력 (LOCK_IN_MERGE_CENSUS_ACTIVE 미설정 — exit 0 유지)",
                file=sys.stderr,
            )
        return 0

    all_errors: list[str] = []
    for spec in CHECKS:
        target = ws / spec["file"]
        errs = verify_function(target, spec["func"], spec["merge_signature_tokens"])
        if errs:
            all_errors.extend(errs)
        elif not args.quiet:
            print(f"PASS  {spec['file']}::{spec['func']}")
    if all_errors:
        print("FAIL  Lock-in First-line 가드 위반:", file=sys.stderr)
        for e in all_errors:
            print(f"  - {e}", file=sys.stderr)
        return 1
    if not args.quiet:
        print("PASS  Lock-in First-line 가드 모든 함수 통과")
    return 0


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