"""Phase 별 금지 행위(publish/deploy/force-push)의 post-hoc 자동 스캐너.

배경: 금지 행위는 `prompts/profiles/forbidden-actions.json` (SSOT) 에 선언되고
launch 경계(PHASE_FORBIDDEN_ACTIONS)와 profile 본문으로 렌더되지만, 지금까지 lead 의
수동 self-review 로만 감사됐다(코드 강제 없음). 이 모듈은 run 종료 후 세션 transcript 의
Bash tool_use 명령을 phase deny-list 와 대조해 위반을 기계적으로 잡아낸다.

스캔 규칙(false-positive 방지가 핵심, validate_session_conformance 와 동일 seam):
- `type == "assistant"` 레코드만, `isSidechain` 제외.
- run 윈도우(resolve_run_window)로 스코핑 — 같은 세션 jsonl 에 섞인 직전 run 의
  명령을 증거로 오인하지 않는다.
- Bash tool_use 의 `input.command` 만 본다 — text 블록(설명/금지어 언급)이나 다른
  tool 의 인자는 보지 않으므로 "npm publish 하지 말 것" 같은 산문은 잡히지 않는다.

한계 1: claude-code 세션 jsonl 만 스캔한다. codex/antigravity lead·worker 는 동일 형식의
tool_use transcript 를 ~/.claude/projects 에 남기지 않으므로 이 스캐너의 사정권 밖이며,
해당 런타임의 금지 행위는 여전히 수동 감사에 의존한다.
한계 2: deny-list 는 heredoc 본문과 검색 명령 절을 걷어낸(strip_heredoc_bodies,
strip_search_clauses) 명령 문자열 전체를 검색하므로, 금지 토큰이 인자/메시지에
등장하는 경우(예: `git commit -m "... npm publish ..."`)는 여전히 위반으로 잡힐 수
있다. `cd <path> && <cmd>` 형태 때문에 명령 head 앵커링은 불가하므로 이 잔여 오탐은
감수한다. 검색 절을 예외로 둔 것은 그 오탐이 우연이 아니라 계약이 시키는 감사에서
반드시 발생하기 때문이다.
"""
from __future__ import annotations

import json
import os
import re
import sys
from pathlib import Path

_UNIVERSAL = [
    ("npm publish", re.compile(r"\bnpm\s+publish\b")),
    ("cargo publish", re.compile(r"\bcargo\s+publish\b")),
    ("pip publish", re.compile(r"\bpip\s+publish\b")),
    ("twine upload", re.compile(r"\btwine\s+upload\b")),
    ("gh release", re.compile(r"\bgh\s+release\s+(create|edit)\b")),
    ("docker push", re.compile(r"\bdocker\s+push\b")),
    ("terraform apply", re.compile(r"\bterraform\s+apply\b")),
    ("kubectl apply", re.compile(r"\bkubectl\s+apply\b")),
    # 강제 push 만 잡는다: force 플래그가 같은 명령 절에 있어야 하므로 git push 와
    # 플래그 사이의 gap 이 shell 구분자(`&&`/`||`/`;`/`|`/개행)를 넘지 못하게 막아
    # `git push origin && rm -rf …` 의 `-rf` 같은 후속 명령 토큰을 오탐하지 않는다.
    # 단문자 플래그 클러스터는 `f` 가 어디에 있든(`-qf`/`-fq`) 매칭한다.
    (
        "git push --force",
        re.compile(r"\bgit\s+push\b[^\n;&|]*(?:--force(-with-lease)?\b|\s-[A-Za-z]*f[A-Za-z]*\b)"),
    ),
]
_NON_HANDOFF_PUSH = ("git push", re.compile(r"\bgit\s+push\b"))

_HEREDOC_START = re.compile(r"<<-?\s*(['\"]?)([A-Za-z_][A-Za-z0-9_]*)\1")


def strip_heredoc_bodies(command: str) -> str:
    """heredoc 본문을 걷어낸 명령 문자열.

    heredoc 안의 토큰은 실행되는 명령이 아니라 파일에 **기록되는 데이터**다.
    implementation-planning 은 배포 명령을 서술하는 계획서를 산출물로 내고 그
    문서를 `python3 - <<'PY'` / `cat > … <<EOF` 로 패치하므로, 본문을 그대로
    스캔하면 정상 경로에서 매 run 오탐이 난다. heredoc 시작 줄과 종료 구분자는
    남겨 실제 명령 위치의 토큰은 계속 잡는다.

    종료 구분자가 뒤에 없으면 아무것도 걷어내지 않는다 — `python3 -c "1 << shift"`
    같은 시프트 연산이 heredoc 으로 오인돼 뒤따르는 진짜 명령을 숨기지 않도록,
    본문 제거는 짝이 맞는 구분자를 실제로 찾았을 때만 한다.
    """
    lines = command.split("\n")
    kept: list[str] = []
    index = 0
    while index < len(lines):
        line = lines[index]
        kept.append(line)
        index += 1
        start = _HEREDOC_START.search(line)
        if not start:
            continue
        delimiter = start.group(2)
        end = next(
            (i for i in range(index, len(lines)) if lines[i].strip() == delimiter),
            None,
        )
        if end is None:
            continue
        kept.append(lines[end])
        index = end + 1
    return "\n".join(kept)


_SEARCH_HEADS = frozenset({"grep", "egrep", "fgrep", "rg", "ag", "ack"})


def _shell_clauses(command: str) -> list[str]:
    """쉘 구분자로 자른 절 목록. 인용 안의 구분자는 자르지 않는다.

    검색 패턴은 거의 언제나 `'git push|npm publish'` 처럼 인용된 교대(alternation)
    이므로, 인용을 무시하고 자르면 패턴이 절 경계로 쪼개져 뒷조각이 그대로 명령처럼
    남는다.
    """
    clauses: list[str] = []
    buffer: list[str] = []
    quote = ""
    index = 0
    while index < len(command):
        char = command[index]
        if quote:
            buffer.append(char)
            if char == quote:
                quote = ""
            index += 1
            continue
        if char in "'\"":
            quote = char
            buffer.append(char)
            index += 1
            continue
        if command.startswith(("&&", "||"), index):
            clauses.append("".join(buffer))
            buffer = []
            index += 2
            continue
        if char in ";|\n":
            clauses.append("".join(buffer))
            buffer = []
            index += 1
            continue
        buffer.append(char)
        index += 1
    clauses.append("".join(buffer))
    return clauses


def strip_search_clauses(command: str) -> str:
    """검색 명령 절을 걷어낸 명령 문자열.

    `grep -E 'git push|npm publish' <logs>` 의 패턴은 실행되는 명령이 아니라
    **찾는 대상**이다. 그리고 그 감사는 우연이 아니라 계약이 시킨다 —
    `_implementation-deliverable.md` 의 자체 리뷰 항목이 리드에게 세션
    트랜스크립트에서 배포 명령을 스캔하라고 요구한다. 걷어내지 않으면 계약을
    이행한 런이 매번 그 이행 때문에 실패했다.

    절 단위로 자르므로 `cd <path> && grep …` 에서도 grep 절만 빠지고, 같은 줄
    뒤에 이어지는 실제 명령은 계속 스캔된다.
    """
    kept: list[str] = []
    for clause in _shell_clauses(command):
        tokens = clause.split()
        if not tokens:
            continue
        head = tokens[0].rsplit("/", 1)[-1]
        if head in _SEARCH_HEADS:
            continue
        if head == "git" and len(tokens) > 1 and tokens[1] == "grep":
            continue
        kept.append(clause)
    return "\n".join(kept)


def forbidden_patterns_for(task_type: str) -> list[tuple[str, re.Pattern]]:
    """task_type(=phase profile) 의 deny-list. 모든 phase 가 _UNIVERSAL 을 받고,
    release-handoff 를 제외한 모든 phase 는 bare `git push` 도 금지한다
    (release-handoff 는 feature 브랜치 push 가 정당하므로 제외)."""
    patterns = list(_UNIVERSAL)
    if task_type != "release-handoff":
        patterns.append(_NON_HANDOFF_PUSH)
    return patterns


def _ensure_token_usage_importable() -> None:
    """okstra_token_usage 패키지를 레이아웃별(repo/scripts, runtime/python,
    OKSTRA_PYTHONPATH)로 해소 — validate_session_conformance 와 동일 후보."""
    here = Path(__file__).resolve().parent
    candidates = [here.parent / "scripts", here.parent / "python"]
    env_pp = os.environ.get("OKSTRA_PYTHONPATH", "").strip()
    if env_pp:
        candidates.append(Path(env_pp))
    for candidate in candidates:
        if candidate.is_dir() and (candidate / "okstra_token_usage").is_dir():
            if str(candidate) not in sys.path:
                sys.path.insert(0, str(candidate))
            break


def _scan_one(
    path: Path,
    since: str | None,
    until: str | None,
    patterns: list[tuple[str, re.Pattern]],
) -> list[tuple[str, str]]:
    """jsonl 한 파일에서 deny-list 에 걸린 (label, command) 쌍을 추출한다."""
    from okstra_token_usage.paths import ts_in_window

    hits: list[tuple[str, str]] = []
    try:
        fh = path.open(encoding="utf-8")
    except OSError:
        return hits
    with fh:
        for raw in fh:
            try:
                rec = json.loads(raw)
            except (json.JSONDecodeError, UnicodeDecodeError):
                continue
            if rec.get("type") != "assistant" or rec.get("isSidechain"):
                continue
            ts = rec.get("timestamp") or ""
            if ts and not ts_in_window(ts, since, until):
                continue
            for block in (rec.get("message") or {}).get("content") or []:
                if not isinstance(block, dict):
                    continue
                if block.get("type") != "tool_use" or block.get("name") != "Bash":
                    continue
                command = (block.get("input") or {}).get("command") or ""
                executed = strip_search_clauses(strip_heredoc_bodies(command))
                for label, pattern in patterns:
                    if pattern.search(executed):
                        hits.append((label, command))
                        break  # one violation per command — most-specific pattern wins
    return hits


def scan_forbidden_actions(
    *,
    team_state: dict,
    team_state_path: Path,
    project_root: Path,
    task_type: str,
    claude_projects_dir: Path | None = None,
) -> list[str]:
    """Return human-readable violation strings (empty = clean)."""
    _ensure_token_usage_importable()
    from okstra_token_usage.claude import find_claude_team_sessions
    from okstra_token_usage.collect import (
        resolve_run_window,
        resolve_team_needles_with_source,
    )

    since, until = resolve_run_window(team_state_path, team_state, relax_start=False)
    lead_sid = (team_state.get("lead") or {}).get("sessionId") or ""
    team_needles, _needle_source = resolve_team_needles_with_source(
        team_state, project_root, since, until, projects_root=claude_projects_dir
    )
    sessions = find_claude_team_sessions(
        project_root,
        team_needles,
        lead_sid,
        projects_root=claude_projects_dir,
    )
    patterns = forbidden_patterns_for(task_type)
    violations: list[str] = []
    for sid, path in sorted(sessions.items()):
        for label, command in _scan_one(path, since, until, patterns):
            violations.append(
                f"{label} 명령이 세션 {sid} 에서 실행됨: {command.strip()[:120]}"
            )
    return violations