#!/usr/bin/env python3
"""
test_stash_lifecycle_failstop.py
task: task-2571+1 (Gemini HIGH carry-over) — spec §3.2.2.3 fail-stop 회귀

검증 목표 (Gemini HIGH finding 회귀 차단):
- spec §3.2.2.3 "실패 시 stop (다음 stash 건드리지 않음)" 강제
- scripts/finish-task.sh 의 stash-lifecycle dispatch heredoc 내부에서 git stash
  pop/drop 이 실패한 경우, 루프가 BREAK 되어 다음 stash 를 건드리지 않아야 한다.
- 박제된 audit log (memory/events/stash-lifecycle-action.<ts>.json) 가 다음을
  포함해야 한다:
    * top-level fail_stop == True
    * 실패한 decision 의 action == "failed"
    * 실패한 decision 의 stderr 필드 (non-empty)
    * decisions 배열 길이 == 실패까지 처리된 stash 수 (전체 stash 수보다 적음)

작성자: 하누만 (개발4팀 QA)
관련 PR: #123 (Gemini HIGH BLOCK 사유)
관련 fix: 카르티케야 (scripts/finish-task.sh)
"""

import json
import os
import shutil
import subprocess
import sys
import tempfile
from pathlib import Path

import pytest

WORKTREE_ROOT = Path(__file__).resolve().parents[2]
FINISH_TASK_SH = WORKTREE_ROOT / "scripts" / "finish-task.sh"
STASH_AUDIT_PY = WORKTREE_ROOT / "scripts" / "stash_audit.py"


# ---------------------------------------------------------------------------
# Helper: 환경 / 임시 repo
# ---------------------------------------------------------------------------

def _git_env() -> dict:
    """git 명령 실행용 환경변수 (테스트용 author/committer 고정)."""
    env = os.environ.copy()
    env["GIT_AUTHOR_NAME"] = "test"
    env["GIT_AUTHOR_EMAIL"] = "test@example.com"
    env["GIT_COMMITTER_NAME"] = "test"
    env["GIT_COMMITTER_EMAIL"] = "test@example.com"
    return env


def _init_temp_workspace() -> str:
    """
    격리된 임시 git repo 초기화 + scripts/stash_audit.py 심볼릭 링크 + memory/events 생성.
    finish-task.sh 의 lifecycle dispatch heredoc 가 기대하는 워크스페이스 레이아웃 구성.
    """
    d = tempfile.mkdtemp(prefix="stash-failstop-test-")
    env = _git_env()
    subprocess.run(["git", "init", "-q", "-b", "main"], cwd=d, check=True, env=env)
    subprocess.run(["git", "config", "user.email", "test@example.com"], cwd=d, check=True, env=env)
    subprocess.run(["git", "config", "user.name", "test"], cwd=d, check=True, env=env)
    (Path(d) / "base.txt").write_text("base initial\n")
    subprocess.run(["git", "add", "base.txt"], cwd=d, check=True, env=env)
    subprocess.run(["git", "commit", "-q", "-m", "init"], cwd=d, check=True, env=env)

    # scripts/stash_audit.py 를 심볼릭 링크로 노출 (heredoc 가 f"{workspace}/scripts/stash_audit.py" 호출).
    scripts_dir = Path(d) / "scripts"
    scripts_dir.mkdir(parents=True, exist_ok=True)
    (scripts_dir / "stash_audit.py").symlink_to(STASH_AUDIT_PY)

    # memory/events 디렉토리 미리 생성 (heredoc 가 박제하는 위치).
    (Path(d) / "memory" / "events").mkdir(parents=True, exist_ok=True)

    return d


def _stash_push_modify(repo_dir: str, message: str, filename: str, content: str) -> None:
    """
    파일을 dirty 상태로 만든 뒤 stash push.
    동일 filename 을 반복 호출하면 stash 충돌을 의도적으로 유발 가능.
    """
    env = _git_env()
    fpath = Path(repo_dir) / filename
    fpath.write_text(content)
    subprocess.run(["git", "add", filename], cwd=repo_dir, check=True, env=env)
    subprocess.run(["git", "stash", "push", "-m", message], cwd=repo_dir, check=True, env=env)


def _stash_count(repo_dir: str) -> int:
    env = _git_env()
    r = subprocess.run(
        ["git", "stash", "list"], cwd=repo_dir, capture_output=True, text=True, env=env,
    )
    if r.returncode != 0 or not r.stdout.strip():
        return 0
    return len([line for line in r.stdout.splitlines() if line.strip()])


# ---------------------------------------------------------------------------
# Helper: scripts/finish-task.sh 의 stash-lifecycle dispatch heredoc 추출
# ---------------------------------------------------------------------------

# heredoc 시작/종료 마커 — finish-task.sh 내부에 동일 PYEOF 가 4개 존재하므로,
# stash lifecycle dispatch 컨텍스트를 명시적으로 식별한다.
LIFECYCLE_DISPATCH_MARKER = "task-2571: stash lifecycle dispatch"


def _extract_lifecycle_dispatch_heredoc(finish_task_sh: Path) -> str:
    """
    finish-task.sh 에서 stash-lifecycle dispatch heredoc 본문을 추출한다.
    카르티케야가 finish-task.sh 를 수정해도 마커 기반이라 lock-in 되지 않음.

    추출 규칙:
        1. LIFECYCLE_DISPATCH_MARKER 가 포함된 라인의 위치를 찾음.
        2. 그 이후 첫 <<'PYEOF' 라인을 heredoc 시작점으로 한다.
        3. 다음 ^PYEOF$ 라인을 heredoc 종료점으로 한다.
    """
    text = finish_task_sh.read_text(encoding="utf-8")
    lines = text.splitlines()

    marker_idx = None
    for i, line in enumerate(lines):
        if LIFECYCLE_DISPATCH_MARKER in line:
            marker_idx = i
            break
    assert marker_idx is not None, (
        f"{finish_task_sh} 에서 마커 미발견: {LIFECYCLE_DISPATCH_MARKER!r}"
    )

    # heredoc 시작 라인 (마커 이후 첫 <<'PYEOF')
    start_idx = None
    for i in range(marker_idx, len(lines)):
        if "<<'PYEOF'" in lines[i]:
            start_idx = i + 1
            break
    assert start_idx is not None, (
        f"마커({marker_idx}) 이후 <<'PYEOF' 미발견"
    )

    # heredoc 종료 라인 (다음 ^PYEOF$)
    end_idx = None
    for i in range(start_idx, len(lines)):
        if lines[i].strip() == "PYEOF":
            end_idx = i
            break
    assert end_idx is not None, (
        f"PYEOF 종료 토큰 미발견 (start={start_idx})"
    )

    body = "\n".join(lines[start_idx:end_idx]) + "\n"
    return body


# ---------------------------------------------------------------------------
# Fixture: 2개 pre-task stash + 동일 파일 dirty working tree
# ---------------------------------------------------------------------------

# 동일 파일에 대해 stash A → stash B → 현재 dirty 시나리오:
#   * stash@{1} = stash A (먼저 push, 이후 stash B push 로 인해 index 1 로 밀림)
#   * stash@{0} = stash B (가장 최근 push)
#   * 현재 working tree: 동일 파일이 또 다른 dirty content
#
# 역순 처리 (높은 index 부터 pop) 시:
#   1) stash@{1} (A) pop 시도 → working tree 에 동일 파일 dirty 가 있어 conflict → fail
#   2) spec §3.2.2.3 fail-stop: 다음 stash@{0} (B) 는 건드리지 않아야 함
#   3) 박제 audit log 에 fail_stop=True, decisions[0].action == "failed", stderr 비어있지 않음

@pytest.fixture()
def two_pretask_stashes_with_dirty_conflict():
    """
    pre-task 분류 stash 2건 + working tree dirty conflict 시나리오.
    yield: (workspace_dir, expected_stash_count_before)
    """
    repo_dir = _init_temp_workspace()

    # stash A: 동일 파일 c.txt 변경 후 push
    _stash_push_modify(
        repo_dir,
        message="WIP: pre-task-2571-A first stash",
        filename="c.txt",
        content="stash_A_content\n",
    )
    # stash B: 동일 파일 c.txt 변경 후 push (stash A 는 index 1 로 밀림)
    _stash_push_modify(
        repo_dir,
        message="WIP: pre-task-2571-B second stash",
        filename="c.txt",
        content="stash_B_content\n",
    )

    # working tree 에 conflict 유발용 dirty change (commit/stash 하지 않음)
    (Path(repo_dir) / "c.txt").write_text("dirty_uncommitted_change\n")

    yield repo_dir, 2

    shutil.rmtree(repo_dir, ignore_errors=True)


# ---------------------------------------------------------------------------
# Helper: heredoc 추출본을 임시 .py 로 작성 + 실행
# ---------------------------------------------------------------------------

def _run_lifecycle_dispatch(workspace: str, task_id: str, approve: bool) -> tuple[int, str, str]:
    """
    finish-task.sh 에서 추출한 lifecycle dispatch heredoc 본문을 임시 .py 로 쓴 뒤
    동일 argv 시그니처로 실행한다. 박제 audit log 는 workspace/memory/events/ 에 생성된다.

    Returns: (exit_code, stdout, stderr)
    """
    body = _extract_lifecycle_dispatch_heredoc(FINISH_TASK_SH)

    tmp_py = tempfile.NamedTemporaryFile(
        mode="w", suffix=".py", delete=False, encoding="utf-8",
    )
    try:
        tmp_py.write(body)
        tmp_py.close()

        # heredoc 의 argv 시그니처 (finish-task.sh 1167~1174 행 참조):
        #   sys.argv[1] = task_id
        #   sys.argv[2] = approve ("0" or "1")
        #   sys.argv[3] = indices_raw
        #   sys.argv[4] = debug ("0" or "1")
        #   sys.argv[5] = workspace
        #   sys.argv[6] = done_file
        #   sys.argv[7] = audit_timeout_sec  (카르티케야 fix 이후 추가됨; 미존재 시 IndexError 가능)
        done_file = str(Path(workspace) / "memory" / "events" / f"{task_id}.done")
        argv = [
            sys.executable,
            tmp_py.name,
            task_id,
            "1" if approve else "0",
            "",                # indices_raw 비움
            "1",               # debug on (stderr 출력)
            workspace,
            done_file,
            "30",              # audit_timeout_sec (카르티케야 fix 추가분; 없으면 무시됨)
        ]
        proc = subprocess.run(argv, capture_output=True, text=True, env=_git_env())
        return proc.returncode, proc.stdout, proc.stderr
    finally:
        try:
            os.unlink(tmp_py.name)
        except OSError:
            pass


def _latest_audit_log(workspace: str) -> dict:
    """
    workspace/memory/events/stash-lifecycle-action.*.json 중 가장 최근 파일을 로드.
    """
    events_dir = Path(workspace) / "memory" / "events"
    candidates = sorted(events_dir.glob("stash-lifecycle-action.*.json"))
    assert candidates, (
        f"audit log 미발견: {events_dir} 내 stash-lifecycle-action.*.json 없음.\n"
        f"실제 디렉토리: {list(events_dir.iterdir()) if events_dir.exists() else 'NOT EXISTS'}"
    )
    with open(candidates[-1], "r", encoding="utf-8") as f:
        return json.load(f)


# ---------------------------------------------------------------------------
# 테스트
# ---------------------------------------------------------------------------

def test_failstop_heredoc_extraction_works():
    """
    finish-task.sh 에서 stash-lifecycle dispatch heredoc 본문을 마커 기반으로 추출 가능해야 한다.
    추출본에 핵심 토큰들이 포함되어야 한다.
    """
    body = _extract_lifecycle_dispatch_heredoc(FINISH_TASK_SH)
    # spec §2 분류 처리 토큰
    assert "pre-task" in body, "heredoc 추출본에 pre-task 분기 없음"
    # 박제 토큰 (spec §3.2.2.2)
    assert "stash-lifecycle-action" in body, "heredoc 추출본에 audit log 박제 토큰 없음"
    # destructive 호출 토큰 — fail-stop 의 대상이 되는 경로
    assert "stash" in body and "pop" in body, (
        "heredoc 추출본에 stash pop 호출 없음"
    )


def _assert_dispatch_rc_acceptable(rc: int, stderr: str) -> None:
    """
    dispatch heredoc 의 exit 코드 검증.
    spec §3.2.2.3 fail-stop 시 sys.exit(1) 가 정상 동작이므로 rc ∈ {0, 1} 모두 허용.
    그 외 (예: IndexError → 2, 기타 예외 → 1 with traceback) 는 실패로 간주.
    fail_stop 시그널은 stderr 에서 식별: "FAIL-STOP" 또는 audit log 자체 존재 여부로 검증.
    """
    if rc not in (0, 1):
        raise AssertionError(
            f"lifecycle dispatch 비정상 종료: exit={rc}\nstderr={stderr}"
        )


def test_failstop_audit_log_is_produced(two_pretask_stashes_with_dirty_conflict):
    """
    pre-task 2건 시드 + dirty conflict 환경에서 lifecycle dispatch 를 approve=1 로 실행하면
    workspace/memory/events/ 에 audit log 가 박제되어야 한다.
    fail-stop 시 sys.exit(1) 동작 (spec §3.2.2.3 의 시그널 전파) 도 정상으로 간주.
    """
    workspace, _ = two_pretask_stashes_with_dirty_conflict
    rc, _, stderr = _run_lifecycle_dispatch(workspace, task_id="task-failstop-test", approve=True)
    _assert_dispatch_rc_acceptable(rc, stderr)

    log = _latest_audit_log(workspace)
    assert "decisions" in log, f"audit log 에 decisions 필드 없음: {list(log.keys())}"


def test_failstop_top_level_fail_stop_is_true(two_pretask_stashes_with_dirty_conflict):
    """
    spec §3.2.2.3: 실패 시 fail_stop=True 가 audit log top-level 에 박제되어야 한다.
    카르티케야의 fix 적용 전에는 이 assertion 이 RED 가 된다 (예상된 동작).
    """
    workspace, _ = two_pretask_stashes_with_dirty_conflict
    rc, _, stderr = _run_lifecycle_dispatch(workspace, task_id="task-failstop-test", approve=True)
    _assert_dispatch_rc_acceptable(rc, stderr)

    log = _latest_audit_log(workspace)
    assert log.get("fail_stop") is True, (
        f"spec §3.2.2.3: top-level fail_stop=True 기대, 실제 audit log:\n"
        f"{json.dumps(log, ensure_ascii=False, indent=2)}"
    )


def test_failstop_failed_decision_has_action_failed(two_pretask_stashes_with_dirty_conflict):
    """
    spec §3.2.2.3: 실패한 stash 의 decision.action == "failed" (기존 'skipped' 가 아님).
    Gemini HIGH 의 핵심 회귀 포인트 — 'skipped' 로 표기 시 fail-stop 식별 불가.
    """
    workspace, _ = two_pretask_stashes_with_dirty_conflict
    rc, _, stderr = _run_lifecycle_dispatch(workspace, task_id="task-failstop-test", approve=True)
    _assert_dispatch_rc_acceptable(rc, stderr)

    log = _latest_audit_log(workspace)
    decisions = log.get("decisions", [])
    failed_decisions = [d for d in decisions if d.get("action") == "failed"]
    assert len(failed_decisions) >= 1, (
        f"action=='failed' decision 1건 이상 기대, 실제: {len(failed_decisions)}\n"
        f"decisions: {json.dumps(decisions, ensure_ascii=False, indent=2)}"
    )


def test_failstop_failed_decision_has_nonempty_stderr(two_pretask_stashes_with_dirty_conflict):
    """
    spec §3.2.2.3: 실패 decision 은 stderr 필드에 git 의 실제 오류 메시지를 박제해야 한다.
    회장결정 박제 doctrine — 사후 진단 가능성 확보.
    """
    workspace, _ = two_pretask_stashes_with_dirty_conflict
    rc, _, stderr = _run_lifecycle_dispatch(workspace, task_id="task-failstop-test", approve=True)
    _assert_dispatch_rc_acceptable(rc, stderr)

    log = _latest_audit_log(workspace)
    failed_decisions = [d for d in log.get("decisions", []) if d.get("action") == "failed"]
    assert failed_decisions, "전제: action=='failed' decision 존재 (test_failstop_failed_decision_has_action_failed 참조)"

    for fd in failed_decisions:
        stderr_value = fd.get("stderr", "")
        assert isinstance(stderr_value, str) and stderr_value.strip(), (
            f"failed decision 에 stderr 필드(non-empty) 기대, 실제: {fd!r}"
        )


def test_failstop_loop_breaks_no_further_decisions(two_pretask_stashes_with_dirty_conflict):
    """
    spec §3.2.2.3 핵심: 실패 발생 시 루프가 BREAK 되어야 하며, 그 이후 stash 는
    decisions 배열에 등장하지 않아야 한다.

    이 시나리오:
        - stash 2건 시드 (양쪽 모두 pre-task)
        - 역순 처리 (높은 index 부터): stash@{1} 가 먼저 처리됨
        - stash@{1} pop 은 dirty conflict 로 실패
        - spec 통과 시: decisions 길이 == 1 (stash@{1} 만 등장; stash@{0} 미처리)
        - spec 위반(현행 버그) 시: decisions 길이 == 2 (stash@{0} 도 'skipped' 로 등장)
    """
    workspace, expected_count = two_pretask_stashes_with_dirty_conflict
    assert _stash_count(workspace) == expected_count, (
        f"전제: stash {expected_count}건 시드 기대, 실제: {_stash_count(workspace)}"
    )

    rc, _, stderr = _run_lifecycle_dispatch(workspace, task_id="task-failstop-test", approve=True)
    _assert_dispatch_rc_acceptable(rc, stderr)

    log = _latest_audit_log(workspace)
    decisions = log.get("decisions", [])
    assert log.get("fail_stop") is True, (
        f"전제: fail_stop=True (test_failstop_top_level_fail_stop_is_true 참조)"
    )

    # 루프 break 검증 — failed 이후 어떤 decision 도 추가되지 않아야 함.
    # decisions 리스트의 마지막 항목이 failed 여야 한다 (failed → break 시퀀스).
    assert decisions, "decisions 비어있음"
    last = decisions[-1]
    assert last.get("action") == "failed", (
        f"spec §3.2.2.3 fail-stop 위반: failed decision 이후 다른 decision 추가됨.\n"
        f"마지막 decision.action 기대: 'failed', 실제: {last.get('action')!r}\n"
        f"전체 decisions: {json.dumps(decisions, ensure_ascii=False, indent=2)}"
    )

    # 추가 강화: 처리된 decisions 수가 전체 stash 수보다 적어야 한다 (loop 가 끝까지 돌지 않았다는 증거).
    # stash 2건 중 stash@{1} 만 처리되어야 하므로 decisions 수는 1.
    assert len(decisions) < expected_count, (
        f"spec §3.2.2.3 fail-stop 위반: 전체 stash {expected_count}건 모두 처리됨 "
        f"(decisions 수: {len(decisions)}). 실패 시 break 되지 않았다는 증거.\n"
        f"decisions: {json.dumps(decisions, ensure_ascii=False, indent=2)}"
    )


def test_failstop_remaining_stash_preserved(two_pretask_stashes_with_dirty_conflict):
    """
    spec §3.2.2.3: 실패 후 break 되었으므로 처리되지 않은 stash 는 그대로 남아있어야 한다.
    stash 2건 중 stash@{1} pop 실패 시:
        - pop 실패 → stash@{1} 도 그대로 남음 (git stash pop 의 keep-on-conflict 동작)
        - stash@{0} 은 break 로 인해 미처리 → 그대로 남음
    => 최종 stash 수 == 2 (변동 없음)
    """
    workspace, expected_count = two_pretask_stashes_with_dirty_conflict
    rc, _, stderr = _run_lifecycle_dispatch(workspace, task_id="task-failstop-test", approve=True)
    _assert_dispatch_rc_acceptable(rc, stderr)

    after = _stash_count(workspace)
    assert after == expected_count, (
        f"spec §3.2.2.3 위반 가능성: 실패 후 stash 수 변동.\n"
        f"기대: {expected_count} (stash@{{1}} pop 실패 keep + stash@{{0}} break 미처리),\n"
        f"실제: {after}"
    )


# ---------------------------------------------------------------------------
# Codex G1 review MEDIUM finding 회귀 차단 (정적 검증)
# ---------------------------------------------------------------------------
# MEDIUM 권고 (Codex G1):
#   - heredoc 내부 `timeout=30` 리터럴을 STASH_AUDIT_TIMEOUT_SEC 상수로 교체했지만,
#     상수 연결이 끊겨도 (literal 30 으로 회귀해도) 잡아내는 회귀 테스트가 없다.
#   - 정적 grep 기반 fast assertion 으로 회귀 방지 (subprocess/git 불필요).

def test_medium_no_timeout_30_literal_in_lifecycle_heredoc():
    """
    Codex G1 MEDIUM: lifecycle dispatch heredoc 본문에 `timeout=30` 리터럴이
    재등장하지 않아야 한다. 상수 (STASH_AUDIT_TIMEOUT_SEC → sys.argv[7]) 연결을 사용해야 함.
    """
    body = _extract_lifecycle_dispatch_heredoc(FINISH_TASK_SH)
    assert "timeout=30" not in body, (
        "Codex G1 MEDIUM 회귀: heredoc 본문에 `timeout=30` 리터럴 재등장.\n"
        "STASH_AUDIT_TIMEOUT_SEC 상수 연결이 끊겼을 가능성. sys.argv[7] 경로 확인 필요."
    )


def test_medium_audit_timeout_uses_sys_argv_7():
    """
    Codex G1 MEDIUM: heredoc 본문이 `audit_timeout = int(sys.argv[7])` 로 상수를 받고,
    `timeout=audit_timeout` 으로 subprocess 에 전달해야 한다. 두 토큰 모두 존재해야 상수 연결 증명.
    """
    body = _extract_lifecycle_dispatch_heredoc(FINISH_TASK_SH)
    assert "audit_timeout = int(sys.argv[7])" in body, (
        "Codex G1 MEDIUM 회귀: heredoc 에 `audit_timeout = int(sys.argv[7])` 미존재.\n"
        "bash → python 상수 전달 경로가 끊김."
    )
    assert "timeout=audit_timeout" in body, (
        "Codex G1 MEDIUM 회귀: heredoc 에 `timeout=audit_timeout` 미존재.\n"
        "subprocess timeout 인자에 상수가 반영되지 않음."
    )


def test_medium_constant_passed_as_argv_in_bash():
    """
    Codex G1 MEDIUM: scripts/finish-task.sh 의 dispatch 호출 블록에서
    `"$STASH_AUDIT_TIMEOUT_SEC"` 가 `"$DONE_FILE"` 와 `<<'PYEOF'` 사이에 위치해야 한다.
    이는 bash 측 상수 → python sys.argv[7] 매핑을 정적으로 검증한다.
    """
    text = FINISH_TASK_SH.read_text(encoding="utf-8")

    # lifecycle dispatch 컨텍스트 마커 이후 첫 호출 블록만 검증 (false positive 방지).
    marker_pos = text.find(LIFECYCLE_DISPATCH_MARKER)
    assert marker_pos != -1, f"마커 미발견: {LIFECYCLE_DISPATCH_MARKER!r}"

    region = text[marker_pos:]
    done_pos = region.find('"$DONE_FILE"')
    pyeof_pos = region.find("<<'PYEOF'")
    assert done_pos != -1, "dispatch 블록에 \"$DONE_FILE\" 인자 미발견"
    assert pyeof_pos != -1, "dispatch 블록에 <<'PYEOF' 미발견"
    assert done_pos < pyeof_pos, (
        f"\"$DONE_FILE\" (pos={done_pos}) 가 <<'PYEOF' (pos={pyeof_pos}) 보다 뒤에 위치."
    )

    between = region[done_pos:pyeof_pos]
    assert '"$STASH_AUDIT_TIMEOUT_SEC"' in between, (
        "Codex G1 MEDIUM 회귀: dispatch 블록 [\"$DONE_FILE\" ... <<'PYEOF'] 사이에\n"
        "`\"$STASH_AUDIT_TIMEOUT_SEC\"` 인자가 없음. bash → python sys.argv[7] 연결 끊김.\n"
        f"실제 사이 구간:\n{between}"
    )