# -*- coding: utf-8 -*-
"""tests/regression/test_sealed_key_and_launcher_wiring_2729p13.py

task-2729+13 Lv.3 보안 민감 회귀 테스트 — sealed key + launcher_fn 결선 검증.
헤임달 (개발2팀 테스터) 작성.

실행 방법:
    PYTHONPATH=/home/jay/p0b-pickup-main python3 -m pytest \
        tests/regression/test_sealed_key_and_launcher_wiring_2729p13.py -v

주의:
- 모든 경로는 tmp_path (pytest fixture) 하위. canonical(/home/jay/workspace) 절대 접근 금지.
- 실제 subprocess 발생 금지 — autouse fixture 로 sabotage 전역 적용(항목15).
- 실제 systemd 활성화는 단위테스트 범위 외 (주석으로 명시).
"""
from __future__ import annotations

import functools
import importlib.util as _ilu
import json
import os
import sys
import time
from datetime import datetime, timezone
from pathlib import Path
from typing import Optional
from unittest.mock import MagicMock

import pytest

# ── sys.path / sys.modules 보강 (test_anu_pickup_driver_2721.py 동일 패턴) ──────
# regression/conftest.py 가 worktree root 를 sys.path[0] 에 보장하지만,
# 단독 실행/순서 변동 및 tests/ 하위 dispatch 패키지 충돌 방지.
_ROOT = Path(__file__).resolve().parents[2]
if str(_ROOT) not in sys.path:
    sys.path.insert(0, str(_ROOT))
elif sys.path[0] != str(_ROOT):
    sys.path.remove(str(_ROOT))
    sys.path.insert(0, str(_ROOT))

# tests/dispatch/__init__.py 존재 시 잘못된 패키지가 로드될 수 있으므로
# 워크트리 root 의 실제 dispatch 패키지를 직접 로드해 sys.modules 에 고정.
_real_init = _ROOT / "dispatch" / "__init__.py"
_cached = sys.modules.get("dispatch")
if _cached is None or (getattr(_cached, "__file__", "") or "") != str(_real_init):
    for _k in [k for k in list(sys.modules) if k == "dispatch" or k.startswith("dispatch.")]:
        del sys.modules[_k]
    _spec = _ilu.spec_from_file_location(
        "dispatch", _real_init, submodule_search_locations=[str(_ROOT / "dispatch")]
    )
    assert _spec is not None and _spec.loader is not None
    _pkg = _ilu.module_from_spec(_spec)
    sys.modules["dispatch"] = _pkg
    _spec.loader.exec_module(_pkg)

# ── import 대상 ───────────────────────────────────────────────────────────────
from dispatch.anu_pickup_driver import (
    REAL_WAKE_FLAG_REL,
    VERDICT_FIRE_FAILED,
    VERDICT_NOOP_LEGACY_SKIP,
    VERDICT_PICKUP_SKIP,
    VERDICT_QUARANTINE,
    VERDICT_WAKE_BUILT,
    build_launcher_fn,
    process_one,
    read_real_wake_enabled,
    scan_once,
)
from dispatch.anu_pickup_wake_launcher import (
    DECISION_FAIL_CLOSED_NON_ANU_KEY,
    DECISION_FAIL_CLOSED_NO_ARGV,
    DECISION_LAUNCHED,
    LaunchRecord,
    launch_wake,
)
from dispatch.anu_result_pickup_runner import (
    PICKUP_SEALED_KEY_MISSING,
    PICKUP_SKIP_DEDUPE,
    PICKUP_SKIP_TERMINAL,
    PICKUP_WAKE_BUILT,
    PickupResult,
    pickup_once,
)

# ── 공통 상수 ─────────────────────────────────────────────────────────────────
_DUMMY_KEY = "SEALEDKEY"         # 명백한 테스트 더미 키 (절대 실 키 아님)
_DUMMY_KEY2 = "ANUKEY123"        # 명백한 테스트 더미 키 (절대 실 키 아님)
_WRONG_KEY = "WRONGKEY_FAKE"     # 틀린 키 (검증 실패용 더미)

# ── autouse fixture: subprocess sabotage (항목 15) ───────────────────────────
# 이 테스트 모듈 전체에서 실제 subprocess/os.system 호출 시 RuntimeError 발생.
# canonical 환경/systemd/cokacdir 등 외부 프로세스가 절대 실행되지 않음을 보장.

@pytest.fixture(autouse=True)
def _sabotage_subprocess(monkeypatch):
    """항목15: subprocess.run/Popen/call/os.system 을 RuntimeError 로 대체.
    어떤 테스트도 실제 subprocess 를 타지 않음을 강제 보장."""

    def _boom(*args, **kwargs):
        raise RuntimeError(
            "[테스트 sabotage] 실제 subprocess 호출 감지 — 테스트에서 subprocess 금지."
        )

    import subprocess as _sp
    monkeypatch.setattr(_sp, "run", _boom)
    monkeypatch.setattr(_sp, "Popen", _boom)
    monkeypatch.setattr(_sp, "call", _boom)
    monkeypatch.setattr(os, "system", _boom)
    yield


# ── 공통 헬퍼 ─────────────────────────────────────────────────────────────────

def _make_result_json(tmp_dir: str, task_id: str = "task-tst-001") -> str:
    """tmp_dir 하위에 유효한 result.json 을 작성하고 경로 반환.
    process_one 통과용: collector_envelope(dict) 포함, verify_fn stub 주입 전제.
    collector_envelope.task_id 도 동일하게 세팅하여 binding 검증 통과."""
    content = {
        "task_id": task_id,
        "completion_signal": "DONE",
        "collector_envelope": {
            "task_id": task_id,
            "schedule_id": "stub-schedule-id",
        },
    }
    path = os.path.join(tmp_dir, f"{task_id}.result.json")
    with open(path, "w", encoding="utf-8") as fh:
        json.dump(content, fh)
    return path


def _make_full_result_json(tmp_dir: str, task_id: str = "task-tst-full-001") -> str:
    """collector_envelope 포함 result.json (pickup_once 직접 호출용)."""
    content = {
        "task_id": task_id,
        "completion_signal": "DONE",
        # envelope 없음 → gh_probe=None 경로 사용(pickup_once step 5 skip)
    }
    path = os.path.join(tmp_dir, f"{task_id}.result.json")
    with open(path, "w", encoding="utf-8") as fh:
        json.dump(content, fh)
    return path


def _authoritative_verify_fn(**kwargs):
    """process_one 용 verify_fn stub — 항상 AUTHORITATIVE 반환."""
    from dispatch.anu_owned_callback_enforcement import (
        VERDICT_AUTHORITATIVE,
        CollectorVerification,
    )
    return CollectorVerification(
        schema="stub",
        verdict=VERDICT_AUTHORITATIVE,
        classification="ANU_OWNED_AUTHORITATIVE",
        task_id=str(kwargs.get("task_id", "")),
        schedule_id="stub-schedule",
        owner_resolution=None,
        envelope_claims={},
        work_preserved=True,
        reasons=["stub authoritative"],
    )


def _aged_stat(path, _orig=os.stat):
    """파일 stat 을 반환하되 st_mtime 을 충분히 과거로 덮어쓴 StatResult 반환.
    write-race 방어(readiness) 를 bypass 하기 위해 aged mtime 주입."""
    st = _orig(path)
    # os.stat_result 는 불변이므로 os.stat_result 로 재구성 불가 — StatResult 흉내 객체 사용
    class _FakeStat:
        def __init__(self, real):
            for attr in dir(real):
                if attr.startswith("st_"):
                    setattr(self, attr, getattr(real, attr))
            self.st_mtime = time.time() - 3600.0  # 1시간 전
    return _FakeStat(st)


def _make_wake_built_pickup_result(
    result_path: str,
    task_id: str = "task-tst-wake-001",
    key: str = _DUMMY_KEY,
) -> PickupResult:
    """WAKE_BUILT PickupResult 를 직접 생성 (mock pickup_fn 용)."""
    import hashlib
    sha = hashlib.sha256(b"dummy").hexdigest()
    argv = ["--cron", "some-cron-id", "--key", key]
    return PickupResult(
        schema="dispatch.anu_result_pickup_runner.v1",
        verdict=PICKUP_WAKE_BUILT,
        task_id=task_id,
        result_json_path=result_path,
        sha256=sha,
        wake_built=True,
        argv=argv,
        classification="",
        reasons=["stub"],
        marker_path=None,
    )


# ════════════════════════════════════════════════════════════════════════════
# 항목 1: sealed key present/absent 각각
# ════════════════════════════════════════════════════════════════════════════

def test_01_sealed_key_absent_returns_none(tmp_path):
    """sealed_key_loader=lambda: None + flag on → build_launcher_fn 이 None 반환.
    real wake 가 활성화되어 있어도 key 없으면 launcher 생성 0."""
    result = build_launcher_fn(
        str(tmp_path),
        real_wake_reader=lambda: "enabled",
        sealed_key_loader=lambda: None,
    )
    assert result is None, "key 부재 시 None 반환해야 함"


def test_01_sealed_key_present_returns_callable(tmp_path):
    """sealed_key_loader=lambda: 'ANUKEY123' + flag on → callable(partial) 반환."""
    result = build_launcher_fn(
        str(tmp_path),
        real_wake_reader=lambda: "enabled",
        sealed_key_loader=lambda: _DUMMY_KEY2,
    )
    assert callable(result), "key 존재 + flag on 시 callable 반환해야 함"
    # functools.partial 인지 확인
    assert isinstance(result, functools.partial), "반환값이 functools.partial 이어야 함"


# ════════════════════════════════════════════════════════════════════════════
# 항목 2: key absent → wake 0 fail-closed (pickup_once 직접 호출)
# ════════════════════════════════════════════════════════════════════════════

def test_02_key_absent_pickup_once_sealed_key_missing(tmp_path):
    """pickup_once(sealed_key_loader=lambda: None) → SEALED_KEY_MISSING, argv None, wake_built False.
    gh_probe=None 경로 사용(collector verify skip). terminal marker/dedupe 없음."""
    result_path = _make_full_result_json(str(tmp_path), task_id="task-02-keyabsent")
    # ledger 를 tmp_path 하위로 격리
    ledger = str(tmp_path / "ledger.jsonl")

    res = pickup_once(
        result_path,
        gh_probe=None,
        sealed_key_loader=lambda: None,
        ledger_path=ledger,
    )
    assert res.verdict == PICKUP_SEALED_KEY_MISSING, (
        f"key 부재 시 SEALED_KEY_MISSING 이어야 함, got={res.verdict}"
    )
    assert res.argv is None, "argv 는 None 이어야 함"
    assert res.wake_built is False, "wake_built 는 False 이어야 함"


# ════════════════════════════════════════════════════════════════════════════
# 항목 3: key present → verifier 동작 + LaunchRecord 에 argv/key 미포함
# ════════════════════════════════════════════════════════════════════════════

def test_03_verifier_correct_key_passes(tmp_path):
    """build_launcher_fn 이 생성한 verifier: 올바른 key → True, 틀린 key → False.
    hmac.compare_digest 상수시간 비교 동작 확인."""
    launcher = build_launcher_fn(
        str(tmp_path),
        real_wake_reader=lambda: "enabled",
        sealed_key_loader=lambda: _DUMMY_KEY,
    )
    assert launcher is not None, "launcher 가 None 이면 안 됨"
    # partial 의 keywords 에서 verifier 추출
    verifier = launcher.keywords.get("anu_key_verifier")
    assert callable(verifier), "verifier 가 callable 이어야 함"
    assert verifier(_DUMMY_KEY) is True, "올바른 key 는 True"
    assert verifier(_WRONG_KEY) is False, "틀린 key 는 False"


def test_03_launch_record_no_argv_no_key_literal(tmp_path):
    """LaunchRecord.to_json() 에 argv/key literal 미포함, argv_len(정수)만.
    dry_run=True 경로로 실행 없이 검증."""
    ledger = str(tmp_path / "launch_ledger.jsonl")
    audit = str(tmp_path / "audit.jsonl")
    argv = ["--cron", "cron-id-stub", "--key", _DUMMY_KEY]

    lr = launch_wake(
        argv,
        task_id="task-03-rawkey",
        sha256="abc123",
        dry_run=True,
        launch_ledger_path=ledger,
        audit_path=audit,
    )
    d = lr.to_json()
    assert "argv" not in d, "to_json 에 argv 필드 있으면 안 됨"
    assert _DUMMY_KEY not in json.dumps(d), "to_json 결과에 key literal 있으면 안 됨"
    assert "argv_len" in d, "argv_len 필드는 있어야 함"
    assert isinstance(d["argv_len"], int), "argv_len 은 정수여야 함"
    assert d["argv_len"] == 4, "argv 길이 4 이어야 함"


# ════════════════════════════════════════════════════════════════════════════
# 항목 4: real-wake flag off → launcher_fn=None, sealed_key_loader 호출 0
# ════════════════════════════════════════════════════════════════════════════

def test_04_real_wake_flag_off_returns_none(tmp_path):
    """real_wake_reader=lambda: None → build_launcher_fn 이 None 반환."""
    result = build_launcher_fn(
        str(tmp_path),
        real_wake_reader=lambda: None,
        sealed_key_loader=lambda: _DUMMY_KEY,
    )
    assert result is None, "flag off 시 None 이어야 함"


def test_04_sealed_key_loader_not_called_when_flag_off(tmp_path):
    """flag off 시 sealed_key_loader 가 한 번도 호출되지 않음을 spy 로 확인."""
    call_count = {"n": 0}

    def _spy_loader():
        call_count["n"] += 1
        return _DUMMY_KEY

    build_launcher_fn(
        str(tmp_path),
        real_wake_reader=lambda: None,
        sealed_key_loader=_spy_loader,
    )
    assert call_count["n"] == 0, (
        f"flag off 시 sealed_key_loader 호출 0 이어야 함, got={call_count['n']}"
    )


def test_04_real_wake_reader_returns_wrong_value_returns_none(tmp_path):
    """real_wake_reader 가 'enabled' 가 아닌 값 반환 시 None."""
    for val in ["disabled", "off", "true", "1", "ENABLED", ""]:
        result = build_launcher_fn(
            str(tmp_path),
            real_wake_reader=lambda v=val: v,
            sealed_key_loader=lambda: _DUMMY_KEY,
        )
        assert result is None, f"값={val!r} 에서 None 이어야 함"


# ════════════════════════════════════════════════════════════════════════════
# 항목 5: real-wake flag on + mock launcher → mock 으로만, 실제 subprocess 0
# ════════════════════════════════════════════════════════════════════════════

def test_05_mock_launcher_called_once_with_dry_run_false(tmp_path):
    """build_launcher_fn(..., launch_wake_fn=mock) → process_one 주입 → mock 1회 호출.
    dry_run=False, anu_key_verifier 포함 확인. 실제 subprocess 미발생."""
    mock_launch = MagicMock()
    mock_launch.return_value = LaunchRecord(
        ts="2026-01-01T00:00:00Z",
        task_id="task-05-mock",
        sha256="sha256stub",
        decision=DECISION_LAUNCHED,
        dry_run=False,
        argv_len=4,
    )

    launcher = build_launcher_fn(
        str(tmp_path),
        real_wake_reader=lambda: "enabled",
        sealed_key_loader=lambda: _DUMMY_KEY,
        launch_wake_fn=mock_launch,
    )
    assert launcher is not None, "launcher 가 None 이면 안 됨"

    # 가짜 pickup_fn: WAKE_BUILT + 유효 argv + sha256 반환
    task_id = "task-05-mock"
    result_path = _make_result_json(str(tmp_path), task_id=task_id)

    fake_res = _make_wake_built_pickup_result(result_path, task_id=task_id)

    def _fake_pickup_fn(path, **kwargs):
        return fake_res

    # process_one 에 launcher_fn 주입
    os.makedirs(str(tmp_path / "memory" / "p0b_state" / "processed"), exist_ok=True)
    rec = process_one(
        result_path,
        root=str(tmp_path),
        pickup_fn=_fake_pickup_fn,
        launcher_fn=launcher,
        verify_fn=_authoritative_verify_fn,
        stable_sec=0.0,
        readiness_retries=1,
        stat_fn=_aged_stat,
    )

    # launcher_fn 이 1회 호출되었는지 확인
    assert mock_launch.call_count == 1, (
        f"mock launch_wake_fn 호출 횟수 1 이어야 함, got={mock_launch.call_count}"
    )
    call_kwargs = mock_launch.call_args[1]
    assert call_kwargs.get("dry_run") is False, "dry_run=False 이어야 함"
    assert callable(call_kwargs.get("anu_key_verifier")), "anu_key_verifier 가 callable 이어야 함"


def test_05_real_launch_wake_with_mock_subprocess_runner(tmp_path):
    """실제 launch_wake 를 mock subprocess_runner 와 함께 호출.
    dry_run=False + 올바른 argv(--cron/--key+값) + verifier 통과 → DECISION_LAUNCHED.
    mock runner 1회 호출, returncode 반영. subprocess.run 미호출(sabotage fixture 보장)."""
    ledger = str(tmp_path / "launch_ledger.jsonl")
    task_id = "task-05-realwake"
    sha256 = "sha256stubval"
    argv = ["--cron", "cron-stub-id", "--key", _DUMMY_KEY]

    mock_runner = MagicMock(return_value=0)

    def _verifier(candidate: str) -> bool:
        import hmac
        return hmac.compare_digest(candidate, _DUMMY_KEY)

    lr = launch_wake(
        argv,
        task_id=task_id,
        sha256=sha256,
        dry_run=False,
        launch_ledger_path=ledger,
        anu_key_verifier=_verifier,
        subprocess_runner=mock_runner,
    )

    assert lr.decision == DECISION_LAUNCHED, f"DECISION_LAUNCHED 이어야 함, got={lr.decision}"
    assert mock_runner.call_count == 1, "mock runner 1회 호출되어야 함"
    assert lr.returncode == 0, "returncode 0 이어야 함"
    # argv_len 정수 확인
    assert isinstance(lr.argv_len, int), "argv_len 은 정수여야 함"


# ════════════════════════════════════════════════════════════════════════════
# 항목 6: legacy 140 skip invariant
# ════════════════════════════════════════════════════════════════════════════

def test_06_legacy_cutoff_aged_result_noop_skip(tmp_path):
    """legacy_cutoff=True + activation_epoch=미래 → VERDICT_NOOP_LEGACY_SKIP.
    파일 이동 0(quarantine 0, processed 0), launcher 호출 0."""
    task_id = "task-06-legacy"
    result_path = _make_result_json(str(tmp_path), task_id=task_id)
    mock_launcher = MagicMock()

    # activation_epoch 을 충분히 미래로 설정 → 현재 파일의 mtime < epoch → legacy skip
    future_epoch = time.time() + 86400 * 365  # 1년 후

    qdir = str(tmp_path / "quarantine")
    pdir = str(tmp_path / "processed")

    rec = process_one(
        result_path,
        root=str(tmp_path),
        launcher_fn=mock_launcher,
        verify_fn=_authoritative_verify_fn,
        stable_sec=0.0,
        readiness_retries=1,
        stat_fn=_aged_stat,
        legacy_cutoff=True,
        activation_epoch=future_epoch,
        quarantine_dir=qdir,
        processed_dir=pdir,
    )

    assert rec.verdict == VERDICT_NOOP_LEGACY_SKIP, (
        f"VERDICT_NOOP_LEGACY_SKIP 이어야 함, got={rec.verdict}"
    )
    assert rec.quarantined is False, "quarantined 는 False 이어야 함"
    assert mock_launcher.call_count == 0, "launcher 호출 0 이어야 함"
    # 파일 이동 0: quarantine/processed 디렉토리가 생성되지 않거나 파일이 없어야 함
    if os.path.isdir(qdir):
        assert not os.listdir(qdir), "quarantine 디렉토리가 비어있어야 함"
    if os.path.isdir(pdir):
        assert not os.listdir(pdir), "processed 디렉토리가 비어있어야 함"
    assert os.path.exists(result_path), "원본 result 파일이 그대로 있어야 함"


# ════════════════════════════════════════════════════════════════════════════
# 항목 7: duplicate 0 (dedupe ledger)
# ════════════════════════════════════════════════════════════════════════════

def test_07_dedupe_pickup_once_skip_dedupe(tmp_path):
    """pickup_once: dedupe ledger 에 동일 (task_id, sha256) PICKUP_WAKE_BUILT 존재 시
    SKIP_DEDUPE 반환, wake 0."""
    task_id = "task-07-dedupe"
    result_path = _make_full_result_json(str(tmp_path), task_id=task_id)
    ledger = str(tmp_path / "ledger.jsonl")

    # 먼저 WAKE_BUILT 기록 생성
    # pickup_once 를 한 번 호출하여 ledger 에 기록 (sealed key 주입)
    # anu_runner_pickup_and_fire 는 실 키 검증이 있으므로 직접 ledger 에 쓰기
    import hashlib
    with open(result_path, "rb") as fh:
        raw = fh.read()
    sha256 = hashlib.sha256(raw).hexdigest()

    entry = json.dumps({
        "schema": "dispatch.anu_result_pickup_runner.dedupe.v1",
        "event": "PICKUP_WAKE_BUILT",
        "task_id": task_id,
        "sha256": sha256,
        "ts": "2026-01-01T00:00:00+00:00",
    })
    os.makedirs(os.path.dirname(ledger), exist_ok=True)
    with open(ledger, "w", encoding="utf-8") as fh:
        fh.write(entry + "\n")

    # 두 번째 pickup_once 호출 → SKIP_DEDUPE
    res = pickup_once(
        result_path,
        gh_probe=None,
        sealed_key_loader=lambda: _DUMMY_KEY,
        ledger_path=ledger,
    )
    assert res.verdict == PICKUP_SKIP_DEDUPE, (
        f"SKIP_DEDUPE 이어야 함, got={res.verdict}"
    )
    assert res.wake_built is False, "wake_built 는 False 이어야 함"


def test_07_dedupe_process_one_pickup_skip(tmp_path):
    """process_one: dedupe ledger 에 동일 task_id PICKUP_WAKE_BUILT 존재 시
    VERDICT_PICKUP_SKIP 반환, launcher 호출 0."""
    task_id = "task-07-drv-dedupe"
    result_path = _make_result_json(str(tmp_path), task_id=task_id)
    ledger = str(tmp_path / "cb4tuple.jsonl")

    # driver 의 _dedupe_hit 는 event==PICKUP_WAKE_BUILT + task_id 로 확인
    entry = json.dumps({
        "event": "PICKUP_WAKE_BUILT",
        "task_id": task_id,
        "sha256": "anysha",
    })
    os.makedirs(os.path.dirname(ledger), exist_ok=True)
    with open(ledger, "w", encoding="utf-8") as fh:
        fh.write(entry + "\n")

    mock_launcher = MagicMock()
    rec = process_one(
        result_path,
        root=str(tmp_path),
        launcher_fn=mock_launcher,
        verify_fn=_authoritative_verify_fn,
        stable_sec=0.0,
        readiness_retries=1,
        stat_fn=_aged_stat,
        ledger_path=ledger,
    )
    assert rec.verdict == VERDICT_PICKUP_SKIP, (
        f"VERDICT_PICKUP_SKIP 이어야 함, got={rec.verdict}"
    )
    assert mock_launcher.call_count == 0, "launcher 호출 0 이어야 함"


# ════════════════════════════════════════════════════════════════════════════
# 항목 8: terminal marker no-op
# ════════════════════════════════════════════════════════════════════════════

def test_08_terminal_marker_done_skip(tmp_path):
    """pickup_once: {task_id}.pickup.done 존재 시 SKIP_TERMINAL, launcher 호출 0."""
    task_id = "task-08-done"
    result_path = _make_full_result_json(str(tmp_path), task_id=task_id)
    ledger = str(tmp_path / "ledger.jsonl")

    # done marker 생성
    done_path = os.path.join(str(tmp_path), f"{task_id}.pickup.done")
    with open(done_path, "w") as fh:
        fh.write("{}")

    res = pickup_once(
        result_path,
        gh_probe=None,
        sealed_key_loader=lambda: _DUMMY_KEY,
        ledger_path=ledger,
    )
    assert res.verdict == PICKUP_SKIP_TERMINAL, (
        f"SKIP_TERMINAL 이어야 함, got={res.verdict}"
    )
    assert res.wake_built is False, "wake_built 는 False 이어야 함"


def test_08_terminal_marker_process_one_pickup_skip(tmp_path):
    """process_one: done marker 존재 시 VERDICT_PICKUP_SKIP, launcher 호출 0."""
    task_id = "task-08-drv-done"
    result_path = _make_result_json(str(tmp_path), task_id=task_id)

    # done marker 생성 (result.json 과 같은 디렉토리)
    done_path = os.path.join(str(tmp_path), f"{task_id}.pickup.done")
    with open(done_path, "w") as fh:
        fh.write("{}")

    mock_launcher = MagicMock()
    rec = process_one(
        result_path,
        root=str(tmp_path),
        launcher_fn=mock_launcher,
        verify_fn=_authoritative_verify_fn,
        stable_sec=0.0,
        readiness_retries=1,
        stat_fn=_aged_stat,
    )
    assert rec.verdict == VERDICT_PICKUP_SKIP, (
        f"VERDICT_PICKUP_SKIP 이어야 함, got={rec.verdict}"
    )
    assert mock_launcher.call_count == 0, "launcher 호출 0 이어야 함"


# ════════════════════════════════════════════════════════════════════════════
# 항목 9: ledger/marker failure fail-safe
# ════════════════════════════════════════════════════════════════════════════

def test_09_ledger_write_blocked_no_crash(tmp_path):
    """pickup_once 에서 ledger_path 를 쓰기 불가 경로로 줘도 크래시 0.
    ledger 디렉토리 자리에 파일을 만들어 쓰기 막기."""
    task_id = "task-09-ledgerfail"
    result_path = _make_full_result_json(str(tmp_path), task_id=task_id)

    # ledger 경로의 부모 디렉토리 자리에 파일을 만들어 os.makedirs 실패 유도
    blocker_dir = str(tmp_path / "blocked_dir")
    with open(blocker_dir, "w") as fh:
        fh.write("I am a file, not a directory")
    blocked_ledger = blocker_dir + "/ledger.jsonl"

    # sealed_key_loader 를 실제 반환하는 stub 으로 주입하되
    # anu_runner_pickup_and_fire 가 실제 키 검증으로 실패 가능 → PICKUP_FAIL 허용
    # 중요한 것은 크래시 0
    try:
        res = pickup_once(
            result_path,
            gh_probe=None,
            sealed_key_loader=lambda: _DUMMY_KEY,
            ledger_path=blocked_ledger,
        )
        # 크래시 없이 어떤 verdict 이든 반환되면 PASS
        assert isinstance(res, PickupResult), "PickupResult 인스턴스 반환되어야 함"
    except Exception as exc:
        pytest.fail(f"크래시 발생 (fail-safe 위반): {exc}")


def test_09_process_one_move_fail_no_crash(tmp_path):
    """process_one 의 _move_processed 실패 시 DriverRecord.error 에 기록되고 크래시 0.
    processed_dir 자리를 파일로 막기."""
    task_id = "task-09-movefail"
    result_path = _make_result_json(str(tmp_path), task_id=task_id)

    # processed_dir 자리에 파일 생성 → makedirs 실패 유도
    blocker = str(tmp_path / "processed_blocker")
    with open(blocker, "w") as fh:
        fh.write("blocker")

    fake_res = _make_wake_built_pickup_result(result_path, task_id=task_id)

    def _fake_pickup_fn(path, **kwargs):
        return fake_res

    try:
        rec = process_one(
            result_path,
            root=str(tmp_path),
            pickup_fn=_fake_pickup_fn,
            launcher_fn=None,
            verify_fn=_authoritative_verify_fn,
            stable_sec=0.0,
            readiness_retries=1,
            stat_fn=_aged_stat,
            processed_dir=blocker,  # 파일 경로를 디렉토리로 써서 실패 유도
        )
        # 크래시 없이 반환되어야 함
        assert rec is not None, "DriverRecord 반환되어야 함"
        # verdict 는 WAKE_BUILT (move 실패는 error 에 기록)
        assert rec.verdict == VERDICT_WAKE_BUILT, (
            f"VERDICT_WAKE_BUILT 이어야 함, got={rec.verdict}"
        )
    except Exception as exc:
        pytest.fail(f"크래시 발생 (fail-safe 위반): {exc}")


# ════════════════════════════════════════════════════════════════════════════
# 항목 10: raw key 0 (LaunchRecord/DriverRecord to_json 에 키/argv 미포함)
# ════════════════════════════════════════════════════════════════════════════

def test_10_launch_record_to_json_no_key_no_argv(tmp_path):
    """LaunchRecord.to_json() 에 key literal(SEALEDKEY) 미포함 + argv 필드 미포함.
    argv_len(정수)만 포함."""
    ledger = str(tmp_path / "launch_ledger.jsonl")
    argv = ["--cron", "cron-10", "--key", _DUMMY_KEY]
    lr = launch_wake(
        argv,
        task_id="task-10-rawkey",
        sha256="sha10",
        dry_run=True,
        launch_ledger_path=ledger,
    )
    d = lr.to_json()
    serialized = json.dumps(d)
    assert _DUMMY_KEY not in serialized, "LaunchRecord.to_json() 에 key literal 있으면 안 됨"
    assert "argv" not in d, "LaunchRecord.to_json() 에 argv 필드 있으면 안 됨"
    assert d.get("argv_len") == 4, "argv_len 은 4 이어야 함"


def test_10_driver_record_to_json_no_key_no_argv(tmp_path):
    """DriverRecord.to_json() 에 키/argv 필드 미포함.
    launcher_fn 주입해서 fire_cron_id 만 기록됨을 확인."""
    task_id = "task-10-drvrec"
    result_path = _make_result_json(str(tmp_path), task_id=task_id)

    fake_lr = LaunchRecord(
        ts="2026-01-01T00:00:00Z",
        task_id=task_id,
        sha256="sha10drv",
        decision=DECISION_LAUNCHED,
        dry_run=False,
        argv_len=4,
    )
    mock_launcher = MagicMock(return_value=fake_lr)

    fake_res = _make_wake_built_pickup_result(result_path, task_id=task_id)

    def _fake_pickup_fn(path, **kwargs):
        return fake_res

    rec = process_one(
        result_path,
        root=str(tmp_path),
        pickup_fn=_fake_pickup_fn,
        launcher_fn=mock_launcher,
        verify_fn=_authoritative_verify_fn,
        stable_sec=0.0,
        readiness_retries=1,
        stat_fn=_aged_stat,
    )
    d = rec.to_json()
    serialized = json.dumps(d)
    assert _DUMMY_KEY not in serialized, "DriverRecord.to_json() 에 key literal 있으면 안 됨"
    assert "argv" not in d, "DriverRecord.to_json() 에 argv 필드 있으면 안 됨"
    # fire_cron_id 는 decision 라벨만 (DECISION_LAUNCHED 문자열)
    assert d.get("fire_cron_id") == DECISION_LAUNCHED, "fire_cron_id 는 decision 라벨이어야 함"


# ════════════════════════════════════════════════════════════════════════════
# 항목 11: ACTIVE=false 불변 (flag 파일 부재 → None/False)
# ════════════════════════════════════════════════════════════════════════════

def test_11_build_launcher_fn_no_flag_file_returns_none(tmp_path):
    """temp root 에 p0b_real_wake_enabled 파일 없음 → build_launcher_fn → None.
    프로덕션 기본 동작(ACTIVE=false) 불변."""
    # tmp_path 에 아무 flag 파일도 없음
    result = build_launcher_fn(str(tmp_path))
    assert result is None, "flag 파일 부재 시 build_launcher_fn 은 None이어야 함"


def test_11_read_real_wake_enabled_no_flag_returns_false(tmp_path):
    """read_real_wake_enabled(<temp root, flag 부재>) → False."""
    result = read_real_wake_enabled(str(tmp_path))
    assert result is False, "flag 파일 부재 시 False 이어야 함"


def test_11_read_real_wake_enabled_various_wrong_values(tmp_path):
    """flag 파일이 있어도 'enabled' 가 아닌 값 → False."""
    flag_path = tmp_path / REAL_WAKE_FLAG_REL
    flag_path.parent.mkdir(parents=True, exist_ok=True)

    for val in ["disabled", "off", "true", "1", "ENABLED", "  ", "yes"]:
        flag_path.write_text(val + "\n")
        result = read_real_wake_enabled(str(tmp_path))
        assert result is False, f"값={val!r} 에서 False 이어야 함"


def test_11_read_real_wake_enabled_correct_value_true(tmp_path):
    """flag 파일 첫 줄 'enabled' → True."""
    flag_path = tmp_path / REAL_WAKE_FLAG_REL
    flag_path.parent.mkdir(parents=True, exist_ok=True)
    flag_path.write_text("enabled\n")
    result = read_real_wake_enabled(str(tmp_path))
    assert result is True, "flag='enabled' 시 True 이어야 함"


# ════════════════════════════════════════════════════════════════════════════
# 항목 12+13: systemd not enabled / activation_epoch absent 불변
# ════════════════════════════════════════════════════════════════════════════

def test_12_13_temp_root_no_state_files(tmp_path):
    """temp root 에 p0b_activation_epoch, p0b_real_wake_enabled, p0b_driver_enabled
    파일이 생성되지 않음을 assert.
    (실제 systemd 활성화는 단위테스트 범위 외 — 이 테스트에서 파일 생성 0.)

    scan_once/build_launcher_fn 호출 후에도 temp root 에 해당 파일 부재 확인."""
    # 어떤 테스트 픽스처도 이 파일들을 만들지 않음을 확인
    epoch_path = tmp_path / "memory" / "state" / "p0b_activation_epoch"
    wake_flag_path = tmp_path / "memory" / "state" / "p0b_real_wake_enabled"
    driver_flag_path = tmp_path / "memory" / "state" / "p0b_driver_enabled"

    assert not epoch_path.exists(), "p0b_activation_epoch 파일이 없어야 함"
    assert not wake_flag_path.exists(), "p0b_real_wake_enabled 파일이 없어야 함"
    assert not driver_flag_path.exists(), "p0b_driver_enabled 파일이 없어야 함"

    # scan_once 호출 후에도 없어야 함 (disabled → NOOP, 파일 생성 0)
    scan_once(
        str(tmp_path),
        flag_reader=lambda: None,  # disabled
        paths=[],
        write_evidence=False,
    )
    assert not epoch_path.exists(), "scan_once 후에도 p0b_activation_epoch 없어야 함"
    assert not wake_flag_path.exists(), "scan_once 후에도 p0b_real_wake_enabled 없어야 함"
    assert not driver_flag_path.exists(), "scan_once 후에도 p0b_driver_enabled 없어야 함"

    # build_launcher_fn 호출 후에도 없어야 함
    build_launcher_fn(str(tmp_path))
    assert not epoch_path.exists(), "build_launcher_fn 후에도 p0b_activation_epoch 없어야 함"
    assert not wake_flag_path.exists(), "build_launcher_fn 후에도 p0b_real_wake_enabled 없어야 함"
    assert not driver_flag_path.exists(), "build_launcher_fn 후에도 p0b_driver_enabled 없어야 함"


# ════════════════════════════════════════════════════════════════════════════
# 항목 14: canonical delta 0 (모든 경로 tmp_path)
# ════════════════════════════════════════════════════════════════════════════

def test_14_no_canonical_root_writes(tmp_path):
    """canonical root(/home/jay/workspace) 에 절대 쓰지 않음.
    모든 root 인자는 tmp_path 하위.
    (방어적 검증: 이 테스트에서 /home/jay/workspace 아래 어떤 파일도 생성/수정 금지.)

    주의: canonical root 존재 여부 확인은 읽기 전용(stat) 만 수행.
    이 테스트는 canonical root 의 mtime 을 확인하여 수정 여부를 검증한다.
    canonical root 가 없으면 skip."""
    CANONICAL_ROOT = "/home/jay/workspace"

    # canonical root 존재 여부 확인 (없으면 skip)
    if not os.path.isdir(CANONICAL_ROOT):
        pytest.skip("canonical root 부재 — delta 검증 skip")

    # 검증용: canonical root mtime 기록
    canonical_mtime_before = os.stat(CANONICAL_ROOT).st_mtime

    # 모든 작업은 tmp_path 로 격리
    task_id = "task-14-canon"
    result_path = _make_result_json(str(tmp_path), task_id=task_id)

    scan_once(
        str(tmp_path),  # root = tmp_path (canonical 아님)
        flag_reader=lambda: None,
        paths=[result_path],
        write_evidence=False,
    )

    build_launcher_fn(
        str(tmp_path),  # root = tmp_path (canonical 아님)
        real_wake_reader=lambda: None,
    )

    # canonical root 의 mtime 이 변경되지 않았는지 확인
    canonical_mtime_after = os.stat(CANONICAL_ROOT).st_mtime
    assert canonical_mtime_before == canonical_mtime_after, (
        "canonical root 가 수정되었습니다 — 절대 쓰기 금지 위반"
    )


# ════════════════════════════════════════════════════════════════════════════
# 항목 15: subprocess sabotage 상태에서 PASS
# ════════════════════════════════════════════════════════════════════════════

def test_15_subprocess_sabotage_no_real_spawn(tmp_path):
    """autouse fixture(_sabotage_subprocess) 가 subprocess.run/Popen/call/os.system 을
    RuntimeError 로 대체한 상태에서, 핵심 함수들이 PASS 함을 명시적으로 확인.

    이 테스트가 PASS 이면:
    - build_launcher_fn, read_real_wake_enabled, scan_once, process_one(pickup skip 경로)
      어느 것도 실제 subprocess 를 호출하지 않음이 증명됨.
    - launch_wake(dry_run=True) + mock subprocess_runner 경로도 실제 subprocess 미발생.
    """
    # 1) read_real_wake_enabled — subprocess 불필요
    result = read_real_wake_enabled(str(tmp_path))
    assert result is False

    # 2) build_launcher_fn — subprocess 불필요
    launcher = build_launcher_fn(str(tmp_path))
    assert launcher is None

    # 3) scan_once(disabled) — subprocess 불필요
    records = scan_once(
        str(tmp_path),
        flag_reader=lambda: None,
        paths=[],
        write_evidence=False,
    )
    assert records, "records 가 비어있으면 안 됨"

    # 4) launch_wake dry_run=True — subprocess 불필요
    ledger = str(tmp_path / "launch_ledger15.jsonl")
    argv = ["--cron", "cron-15", "--key", _DUMMY_KEY]
    lr = launch_wake(
        argv,
        task_id="task-15-sabotage",
        sha256="sha15",
        dry_run=True,
        launch_ledger_path=ledger,
    )
    assert lr is not None

    # 5) launch_wake dry_run=False + mock subprocess_runner
    #    (실제 subprocess.run 은 sabotage fixture 에 의해 차단됨)
    mock_runner = MagicMock(return_value=0)
    lr2 = launch_wake(
        argv,
        task_id="task-15-sabotage-real",
        sha256="sha15real",
        dry_run=False,
        launch_ledger_path=ledger,
        subprocess_runner=mock_runner,
    )
    assert lr2.decision == DECISION_LAUNCHED
    assert mock_runner.call_count == 1, "mock runner 1회 호출"

    # 전체 PASS → 실제 subprocess 미발생 확인 완료