sessions/scripts/lint_python_thinning.py

#!/usr/bin/env python3
"""Boundary lint — Python thinning ban-list checker.

Wave 1.5 거버넌스 가드. PR/push diff에서 *추가된 라인*만 검사하므로
기존 코드의 grandfather 처리가 자동으로 된다.

Usage:
    scripts/lint_python_thinning.py [--base-ref REF] [--lint LINT [LINT ...]]
    scripts/lint_python_thinning.py --pr-body PATH   # Lint #6 only

활성 룰 (PR 0):
- #1 helper response parser 시그니처 ban (Python 측)
- #2.5 Track H2 retry/timeout 분산 ban (commands_*.py)
- #4 Rust ABI 영문 자연어 ban (Rust 측)
- #6 PR boundary-claim 헤더 검증

활성 룰 (PR 2):
- #3 Python python3 -c SSH 폴백 ban (sublime/sessions/, askpass 예외)

활성 룰 (PR 16c):
- #2 commands_*.py 신규 deque task queue ban (기존 _BACKGROUND_TASK_QUEUE,
     _MIRROR_TASK_QUEUE는 grandfather; callable dispatch는 Sublime UI
     thread 잔존 — rust-pragmatist 양보 영역).

후속 활성화 룰:
- #5 boundary inventory metasync (Wave 2.5에서 자동화)

normative 출처: planning/PYTHON_RUST_BOUNDARY.md (Wave 1.5 amend),
              planning/PYTHON_THINNING_PLAN.md §4.3.
"""

from __future__ import annotations

import argparse
import os
import re
import subprocess
import sys
from pathlib import Path
from typing import Iterable, List, Optional, Tuple

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


# ---------------------------------------------------------------------------
# 규칙 정의
# ---------------------------------------------------------------------------

# Lint #1 — Helper response parser 시그니처 ban (Python 측 sublime/sessions/)
# `_rust_ffi/`(또는 `_rust_ffi.py`)의 thin ctypes wrapper만 예외.
LINT_1_PARSER_SIGNATURES = re.compile(
    r"^\s*def\s+_?(parse_ruff|parse_pyright|parse_diagnostic|"
    r"parse_open_outcome|parse_request_outcome|parse_response_packet|"
    r"extract_handshake|payload_method_label)\b",
)
LINT_1_PATH_PATTERN = re.compile(r"^sublime/sessions/")
LINT_1_EXEMPT_PATH_PATTERN = re.compile(r"^sublime/sessions/_rust_ffi(/|\.py$)")

# Lint #2 — commands_*.py 신규 deque/Event task queue 신설 ban (PR 16c).
# commands.py 본체의 _BACKGROUND_TASK_QUEUE/_MIRROR_TASK_QUEUE는 grandfather
# (callable dispatch는 Sublime UI thread 잔존). Track H2 분리 모듈에서 새 큐가
# 생기면 fail.
LINT_2_QUEUE_PATTERNS = [
    re.compile(r"^_[A-Z_]*_TASK_QUEUE\s*=\s*deque\("),
    re.compile(r"^_[A-Z_]*_TASK_EVENT\s*=\s*threading\.Event\("),
]
LINT_2_PATH_PATTERN = re.compile(r"^sublime/sessions/commands_[^/]+\.py$")


# Lint #2.5 — Track H2 retry/timeout 분산 ban
# commands_*.py 분리 모듈에서 retry/timeout 원시 직접 사용 금지.
# (commands.py 본체는 이미 이런 코드를 보유 — diff 기반이라 자동 grandfather.)
LINT_2_5_RETRY_PATTERNS = [
    re.compile(r"\btime\.monotonic\s*\("),
    re.compile(r"\brequests\.exceptions\b"),
    re.compile(r"\btenacity\b"),
    re.compile(r"\bfor\s+\w+\s+in\s+range\s*\(\s*\w*retries?\b"),
    re.compile(r"\bbackoff\.\w+"),
]
LINT_2_5_PATH_PATTERN = re.compile(r"^sublime/sessions/commands_[^/]+\.py$")

# Lint #3 — Python `python3 -c` 원격 폴백 ban (boundary §17–19 Wave 1 closure)
# 원격에서 실행될 명령에 `python3 -c` literal이 새로 추가되는 것을 차단.
# 진짜 ban 의도: ssh 인자 또는 helper exec_once payload 안의 `python3 -c`.
# Diff 모드라 grandfather 자동: ssh_runner.py 로컬 askpass + marimo port pick은
# 기존 코드라 통과; 새 PR이 같은 패턴을 추가하면 fail.
LINT_3_REMOTE_PYTHON_C = [
    re.compile(r'["\']\s*python3\s+-c\s'),
    re.compile(r'["\']\s*python3["\']\s*,\s*["\']-c["\']'),
]
LINT_3_PATH_PATTERN = re.compile(r"^sublime/sessions/")
# askpass 모듈은 *로컬* python3 -c (Tk GUI dialog) 용도라 예외.
LINT_3_EXEMPT_PATH_PATTERN = re.compile(r"^sublime/sessions/(ssh_runner\.py)$")

# Lint #4 — Rust ABI 영문 자연어 ban (Rust 측 sessions_native ABI 함수)
# 식별자 코드만 반환해야 함. ABI 응답에 영문 자연어 문장(공백 + 3+ 어휘) 포함 금지.
# 휴리스틱: ABI 함수 본문 string literal "Word word word..." 패턴 grep.
LINT_4_NATURAL_LANGUAGE = re.compile(r'"[A-Z][a-z]+(?:\s+[a-z]+){2,}[\.,!?]?"')
LINT_4_PATH_PATTERN = re.compile(r"^rust/crates/sessions_native/src/")

# Lint #6 — PR boundary-claim 헤더 검증
# PR description에 다음 블록이 있어야 함:
#   boundary-claim:
#     removes: <list>
#     delete-count: <int>
#     ban-list: <list>
LINT_6_BOUNDARY_CLAIM = re.compile(
    r"^boundary-claim:\s*$\s*"
    r"(?:^\s+removes:\s*.*?\s*$\s*)?"
    r"(?:^\s+delete-count:\s*\d+\s*$\s*)?"
    r"(?:^\s+ban-list:\s*.*?\s*$\s*)?",
    re.MULTILINE,
)


# ---------------------------------------------------------------------------
# Diff 추출
# ---------------------------------------------------------------------------


def _git(args: List[str]) -> str:
    result = subprocess.run(
        ["git", *args],
        cwd=REPO_ROOT,
        capture_output=True,
        text=True,
        check=False,
    )
    return result.stdout


def _resolve_base_ref(explicit: Optional[str]) -> Optional[str]:
    if explicit:
        return explicit
    env_base = os.environ.get("LINT_THINNING_BASE_REF")
    if env_base:
        return env_base
    if os.environ.get("CI"):
        merge_base = _git(["merge-base", "HEAD", "origin/main"]).strip()
        if merge_base:
            return merge_base
    return None


def _added_lines(base_ref: Optional[str]) -> List[Tuple[Path, int, str]]:
    """Return (path, line_no_in_new_file, content) for every line added vs base.

    base_ref None이면 working tree 전체를 검사한다 (PR 0 활성화 시 sanity).
    """
    if base_ref is None:
        # 전수 검사 — grandfather 없음. PR 0에서는 호출하지 않는 게 정상.
        results: List[Tuple[Path, int, str]] = []
        for py in sorted(REPO_ROOT.glob("sublime/**/*.py")):
            rel = py.relative_to(REPO_ROOT)
            for n, line in enumerate(py.read_text(encoding="utf-8").splitlines(), 1):
                results.append((rel, n, line))
        for rs in sorted(REPO_ROOT.glob("rust/crates/**/*.rs")):
            rel = rs.relative_to(REPO_ROOT)
            for n, line in enumerate(rs.read_text(encoding="utf-8").splitlines(), 1):
                results.append((rel, n, line))
        return results

    raw = _git(["diff", "--unified=0", base_ref, "--", "sublime/", "rust/crates/"])
    added: List[Tuple[Path, int, str]] = []
    current_path: Optional[Path] = None
    new_line_no = 0
    for line in raw.splitlines():
        if line.startswith("+++ b/"):
            current_path = Path(line[len("+++ b/") :])
            continue
        if line.startswith("@@"):
            m = re.search(r"\+(\d+)", line)
            new_line_no = int(m.group(1)) - 1 if m else 0
            continue
        if line.startswith("+") and not line.startswith("+++") and current_path:
            new_line_no += 1
            added.append((current_path, new_line_no, line[1:]))
        elif not line.startswith("-") and current_path:
            new_line_no += 1
    return added


# ---------------------------------------------------------------------------
# Lint 실행
# ---------------------------------------------------------------------------


class Violation:
    __slots__ = ("lint_id", "path", "line_no", "content", "reason")

    def __init__(
        self,
        lint_id: str,
        path: Path,
        line_no: int,
        content: str,
        reason: str,
    ) -> None:
        self.lint_id = lint_id
        self.path = path
        self.line_no = line_no
        self.content = content
        self.reason = reason

    def __str__(self) -> str:
        return (
            f"[{self.lint_id}] {self.path}:{self.line_no}: {self.reason}\n"
            f"    {self.content.strip()}"
        )


def _check_lint_1(added: Iterable[Tuple[Path, int, str]]) -> List[Violation]:
    violations: List[Violation] = []
    for path, line_no, content in added:
        rel = str(path).replace("\\", "/")
        if not LINT_1_PATH_PATTERN.match(rel):
            continue
        if LINT_1_EXEMPT_PATH_PATTERN.match(rel):
            continue
        if LINT_1_PARSER_SIGNATURES.match(content):
            violations.append(
                Violation(
                    lint_id="#1",
                    path=path,
                    line_no=line_no,
                    content=content,
                    reason=(
                        "helper response parser 시그니처 신규 금지 — "
                        "Rust ABI 호출 + typed wrapper 1단계만 허용"
                    ),
                )
            )
    return violations


def _check_lint_2(added: Iterable[Tuple[Path, int, str]]) -> List[Violation]:
    violations: List[Violation] = []
    for path, line_no, content in added:
        rel = str(path).replace("\\", "/")
        if not LINT_2_PATH_PATTERN.match(rel):
            continue
        for pattern in LINT_2_QUEUE_PATTERNS:
            if pattern.search(content.lstrip()):
                violations.append(
                    Violation(
                        lint_id="#2",
                        path=path,
                        line_no=line_no,
                        content=content,
                        reason=(
                            "Track H2 분리 모듈에 새 deque/Event task queue 금지 "
                            "— 큐 state는 sessions_native::orchestrator"
                        ),
                    )
                )
                break
    return violations


def _check_lint_2_5(added: Iterable[Tuple[Path, int, str]]) -> List[Violation]:
    violations: List[Violation] = []
    for path, line_no, content in added:
        rel = str(path).replace("\\", "/")
        if not LINT_2_5_PATH_PATTERN.match(rel):
            continue
        for pattern in LINT_2_5_RETRY_PATTERNS:
            if pattern.search(content):
                violations.append(
                    Violation(
                        lint_id="#2.5",
                        path=path,
                        line_no=line_no,
                        content=content,
                        reason=(
                            "Track H2 분리 모듈에서 retry/timeout 원시 직접 사용 금지 "
                            "— _rust_ffi/bridge 호출 표면에 응집"
                        ),
                    )
                )
                break
    return violations


def _check_lint_3(added: Iterable[Tuple[Path, int, str]]) -> List[Violation]:
    violations: List[Violation] = []
    for path, line_no, content in added:
        rel = str(path).replace("\\", "/")
        if not LINT_3_PATH_PATTERN.match(rel):
            continue
        if LINT_3_EXEMPT_PATH_PATTERN.match(rel):
            continue
        for pattern in LINT_3_REMOTE_PYTHON_C:
            if pattern.search(content):
                violations.append(
                    Violation(
                        lint_id="#3",
                        path=path,
                        line_no=line_no,
                        content=content,
                        reason=(
                            "원격 명령에 `python3 -c` 폴백 신규 금지 "
                            "(boundary §17–19) — helper 채널 사용 필요"
                        ),
                    )
                )
                break
    return violations


def _check_lint_4(added: Iterable[Tuple[Path, int, str]]) -> List[Violation]:
    violations: List[Violation] = []
    for path, line_no, content in added:
        rel = str(path).replace("\\", "/")
        if not LINT_4_PATH_PATTERN.match(rel):
            continue
        if LINT_4_NATURAL_LANGUAGE.search(content):
            violations.append(
                Violation(
                    lint_id="#4",
                    path=path,
                    line_no=line_no,
                    content=content,
                    reason=(
                        "Rust ABI에 영문 자연어 문장 금지 — "
                        "식별자 코드(int, kebab-case)만 반환"
                    ),
                )
            )
    return violations


def _check_lint_6_pr_body(pr_body_path: Path) -> List[Violation]:
    if not pr_body_path.exists():
        return [
            Violation(
                lint_id="#6",
                path=pr_body_path,
                line_no=0,
                content="",
                reason=f"PR description 파일 없음: {pr_body_path}",
            )
        ]
    body = pr_body_path.read_text(encoding="utf-8")
    if not LINT_6_BOUNDARY_CLAIM.search(body):
        return [
            Violation(
                lint_id="#6",
                path=pr_body_path,
                line_no=0,
                content="(PR description)",
                reason=(
                    "PR description에 boundary-claim 블록이 필요함:\n"
                    "    boundary-claim:\n"
                    "      removes: <list of file:line ranges>\n"
                    "      delete-count: <int>\n"
                    "      ban-list: <activated lints, optional>\n"
                ),
            )
        ]
    return []


# ---------------------------------------------------------------------------
# CLI
# ---------------------------------------------------------------------------


ALL_LINTS = ("1", "2", "2.5", "3", "4", "6")


def main() -> int:
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument(
        "--base-ref",
        default=None,
        help="diff base; CI에서는 자동으로 origin/main과의 merge-base 사용",
    )
    parser.add_argument(
        "--lint",
        action="append",
        default=None,
        choices=ALL_LINTS,
        help="실행할 룰 (반복 가능, 기본은 활성 룰 전체)",
    )
    parser.add_argument(
        "--pr-body",
        type=Path,
        default=None,
        help="Lint #6: PR description 파일 경로",
    )
    parser.add_argument(
        "--all-files",
        action="store_true",
        help="diff 대신 전체 파일 검사 (PR 0 sanity 용도, grandfather 없음)",
    )
    args = parser.parse_args()

    selected = set(args.lint) if args.lint else set(ALL_LINTS)
    violations: List[Violation] = []

    if {"1", "2", "2.5", "3", "4"} & selected:
        base_ref = None if args.all_files else _resolve_base_ref(args.base_ref)
        added = _added_lines(base_ref)
        if "1" in selected:
            violations.extend(_check_lint_1(added))
        if "2" in selected:
            violations.extend(_check_lint_2(added))
        if "2.5" in selected:
            violations.extend(_check_lint_2_5(added))
        if "3" in selected:
            violations.extend(_check_lint_3(added))
        if "4" in selected:
            violations.extend(_check_lint_4(added))

    if "6" in selected:
        pr_body = args.pr_body
        if pr_body is None:
            env_path = os.environ.get("LINT_THINNING_PR_BODY")
            if env_path:
                pr_body = Path(env_path)
        if pr_body is not None:
            violations.extend(_check_lint_6_pr_body(pr_body))

    if violations:
        print("Boundary lint (Wave 1.5) — 위반 발견:", file=sys.stderr)
        for v in violations:
            print(str(v), file=sys.stderr)
        print(
            f"\n{len(violations)}건 위반. "
            "boundary 문서: planning/PYTHON_RUST_BOUNDARY.md",
            file=sys.stderr,
        )
        return 1

    return 0


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