# -*- coding: utf-8 -*-
"""task-2744 — D-CANARY-2 legacy_skip preserve-move + schema-valid fixture (회귀 11).

대상: dispatch.anu_pickup_driver

문제(controlled active canary ABORTED_PARTIAL 실증):
  anu-pickup.path=PathExistsGlob(level-trigger). driver process_one 의 NOOP_LEGACY_SKIP
  (pre_activation_epoch)이 move 0 → epoch 이전 잔존 result 가 watched inbox
  (memory/events/p0b_inbox/task-*.result.json)에 계속 남음 → PathExistsGlob 계속 true →
  재트리거 폭주 → start-limit-hit.

수정 검증:
  ① pre_activation_epoch → legacy_skip preserve-move(watched glob 밖 memory/p0b_state/legacy_skip/).
     삭제 0·내용변경 0·"처리완료"(pickup) 아님.
  ② sha256 before/after 검증, mismatch 시 abort(원본 보존).
  ③ manifest append-only(9필드) — original_path·preserved_path·sha256_before·sha256_after·
     reason(pre_activation_epoch)·task_id·mtime·migrated_at·actor·reversible=true.
  ④ manifest 기준 rollback 가역.
  ⑤ (DD3) schema-valid controlled fixture builder/test 고정(completion_signal/collector_envelope).

절대 제약:
- ANU key literal 0. pickup_once/verify 실호출 0 — 전부 mock. 실제 wake/cron 0.
- tmp_path 만 사용. 실제 memory 디렉토리/운영 fixture 에 쓰지 않는다(재canary 0).
"""
from __future__ import annotations

import glob
import json
import os
import sys
import time
import types
from pathlib import Path

import pytest

# regression/conftest.py 가 worktree root 를 sys.path 에 보장하지만 단독 실행 대비 보강.
_ROOT = Path(__file__).resolve().parents[2]
if str(_ROOT) not in sys.path:
    sys.path.insert(0, str(_ROOT))

import importlib.util as _ilu  # noqa: E402

# 실제 dispatch 패키지를 파일 위치로 직접 로드(tests/dispatch 빈 패키지 오염 방지) — 2721 테스트와 동일.
_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)

from dispatch import anu_pickup_driver as drv  # noqa: E402

# schema-valid fixture builder 를 파일 경로로 로드(import-mode 무관 robust).
_BUILDER_PATH = Path(__file__).resolve().parent / "legacy_skip_fixture_builder_2744.py"
_bspec = _ilu.spec_from_file_location("legacy_skip_fixture_builder_2744", _BUILDER_PATH)
assert _bspec is not None and _bspec.loader is not None
builder = _ilu.module_from_spec(_bspec)
_bspec.loader.exec_module(builder)


# ─────────────────────────────────────────────────────────────────────────────
# 헬퍼
# ─────────────────────────────────────────────────────────────────────────────
_NO_SLEEP = lambda *a, **k: None  # noqa: E731

# 모든 mtime 을 이 기준 epoch 에 대해 배치한다. epoch 은 과거(now-100000)로 두어
# fresh 파일도 readiness(aged) 를 통과하게 한다.
_NOW = time.time()
_EPOCH = _NOW - 100_000.0            # activation_epoch (과거 기준점)
_LEGACY_MTIME = _EPOCH - 50_000.0    # epoch 이전 → pre_activation_epoch
_FRESH_MTIME = _NOW - 30.0           # epoch 이후 + aged(30s>2s) → 정상 path


def _make_dirs(tmp_path: Path) -> Path:
    (tmp_path / "memory" / "events" / "p0b_inbox").mkdir(parents=True, exist_ok=True)
    (tmp_path / "memory" / "state").mkdir(parents=True, exist_ok=True)
    (tmp_path / "memory" / "p0b_state").mkdir(parents=True, exist_ok=True)
    return tmp_path


def _inbox(root: Path) -> Path:
    return root / "memory" / "events" / "p0b_inbox"


def _legacy_skip_dir(root: Path) -> Path:
    return root / "memory" / "p0b_state" / "legacy_skip"


def _manifest_path(root: Path) -> Path:
    return root / drv.LEGACY_SKIP_MANIFEST_REL


def _write_result(root: Path, name: str, payload, *, mtime: float) -> Path:
    p = _inbox(root) / name
    if isinstance(payload, (dict, list)):
        p.write_text(json.dumps(payload, ensure_ascii=False), encoding="utf-8")
    else:
        p.write_text(str(payload), encoding="utf-8")
    os.utime(str(p), (mtime, mtime))
    return p


def _inbox_glob(root: Path) -> list:
    return sorted(glob.glob(str(_inbox(root) / drv.RESULT_GLOB)))


def make_pickup_mock(verdict: str = "CLOSEOUT_DONE"):
    calls: list = []

    def _pickup(*args, **kwargs):
        calls.append((args, kwargs))
        return types.SimpleNamespace(
            verdict=verdict, ok=True, argv=["x"], task_id="task-x",
            reasons=[], marker_path="/tmp/marker", sha256="abc",
            callback_launch_verdict=None,
        )

    _pickup.calls = calls  # type: ignore[attr-defined]
    return _pickup


def make_verify_mock(verdict: str = "AUTHORITATIVE"):
    calls: list = []

    def _verify(*args, **kwargs):
        calls.append((args, kwargs))
        return types.SimpleNamespace(
            verdict=verdict, ok=(verdict == "AUTHORITATIVE"),
            owner_resolution={"outcome": "ANU_OWNED", "query_ok": True},
            reasons=[],
        )

    _verify.calls = calls  # type: ignore[attr-defined]
    return _verify


def _process_legacy(root, path, **over):
    """preserve-move 활성 process_one 공통 호출(legacy_cutoff + legacy_preserve_move)."""
    kw = dict(
        root=str(root),
        pickup_fn=over.pop("pickup_fn", make_pickup_mock()),
        verify_fn=over.pop("verify_fn", make_verify_mock()),
        clock=lambda: __import__("datetime").datetime.now(drv.KST),
        legacy_cutoff=True,
        legacy_preserve_move=True,
        activation_epoch=_EPOCH,
        sleep_fn=_NO_SLEEP,
    )
    kw.update(over)
    return drv.process_one(str(path), **kw)


# ─────────────────────────────────────────────────────────────────────────────
# 1. pre_activation_epoch → owner-pickup 처리 0, legacy_skip preserve-move 발생
# ─────────────────────────────────────────────────────────────────────────────
def test_01_pre_epoch_preserve_move_no_pickup(tmp_path):
    root = _make_dirs(tmp_path)
    payload = builder.build_schema_valid_result(task_id="task-2309")
    p = _write_result(root, "task-2309.result.json", payload, mtime=_LEGACY_MTIME)
    pickup = make_pickup_mock()
    verify = make_verify_mock()
    rec = _process_legacy(root, p, pickup_fn=pickup, verify_fn=verify)
    assert rec.verdict == drv.VERDICT_LEGACY_SKIP_PRESERVED, rec.verdict
    assert rec.quarantine_reason == "pre_activation_epoch"
    assert rec.preserved_path and os.path.isfile(rec.preserved_path)
    # owner-pickup 처리 0: pickup/verify 미호출.
    assert pickup.calls == []
    assert verify.calls == []
    # quarantine 아님(verdict 명시 분류).
    assert rec.quarantined is False


# ─────────────────────────────────────────────────────────────────────────────
# 2. 이동 후 원본 watched inbox path 에 파일 없음
# ─────────────────────────────────────────────────────────────────────────────
def test_02_original_inbox_path_emptied(tmp_path):
    root = _make_dirs(tmp_path)
    p = _write_result(root, "task-2310.result.json",
                      builder.build_schema_valid_result(task_id="task-2310"),
                      mtime=_LEGACY_MTIME)
    rec = _process_legacy(root, p)
    assert rec.verdict == drv.VERDICT_LEGACY_SKIP_PRESERVED
    assert not os.path.exists(str(p)), "원본 watched inbox path 가 비워져야 함"
    assert _inbox_glob(root) == [], "watched glob 결과 0 — PathExistsGlob false 化"


# ─────────────────────────────────────────────────────────────────────────────
# 3. preserved file sha256 == original sha256 (내용 보존)
# ─────────────────────────────────────────────────────────────────────────────
def test_03_preserved_sha256_equals_original(tmp_path):
    root = _make_dirs(tmp_path)
    payload = builder.build_schema_valid_result(task_id="task-2311")
    p = _write_result(root, "task-2311.result.json", payload, mtime=_LEGACY_MTIME)
    original_sha = drv._sha256_file(str(p))
    rec = _process_legacy(root, p)
    assert rec.verdict == drv.VERDICT_LEGACY_SKIP_PRESERVED
    preserved_sha = drv._sha256_file(rec.preserved_path)
    assert preserved_sha == original_sha, "preserved sha256 == original sha256 (삭제0·내용변경0)"


# ─────────────────────────────────────────────────────────────────────────────
# 4. manifest jsonl append 확인
# ─────────────────────────────────────────────────────────────────────────────
def test_04_manifest_append(tmp_path):
    root = _make_dirs(tmp_path)
    p = _write_result(root, "task-2312.result.json",
                      builder.build_schema_valid_result(task_id="task-2312"),
                      mtime=_LEGACY_MTIME)
    rec = _process_legacy(root, p)
    assert rec.verdict == drv.VERDICT_LEGACY_SKIP_PRESERVED
    man = _manifest_path(root)
    assert man.is_file(), "legacy_skip_migration.jsonl 생성"
    lines = [l for l in man.read_text(encoding="utf-8").splitlines() if l.strip()]
    assert len(lines) == 1
    entry = json.loads(lines[0])
    assert entry["task_id"] == "task-2312"
    assert entry["reason"] == "pre_activation_epoch"


# ─────────────────────────────────────────────────────────────────────────────
# 5. manifest 기반 rollback 정보 전부(필수 9+필드) + 가역 동작
# ─────────────────────────────────────────────────────────────────────────────
def test_05_manifest_fields_and_rollback_reversible(tmp_path):
    root = _make_dirs(tmp_path)
    payload = builder.build_schema_valid_result(task_id="task-2313")
    p = _write_result(root, "task-2313.result.json", payload, mtime=_LEGACY_MTIME)
    original_sha = drv._sha256_file(str(p))
    rec = _process_legacy(root, p)
    assert rec.verdict == drv.VERDICT_LEGACY_SKIP_PRESERVED

    entry = json.loads(_manifest_path(root).read_text(encoding="utf-8").splitlines()[0])
    required = (
        "original_path", "preserved_path", "sha256_before", "sha256_after",
        "reason", "task_id", "mtime", "migrated_at", "actor", "reversible",
    )
    for k in required:
        assert k in entry, f"manifest 필수 필드 누락: {k}"
    assert entry["reversible"] is True
    assert entry["reason"] == "pre_activation_epoch"
    assert entry["actor"]
    assert entry["sha256_before"] == original_sha
    assert entry["sha256_after"] == original_sha
    assert entry["original_path"] == str(p)

    # rollback 가역: preserved → original 환원, sha256 재검증, preserved 제거.
    assert not os.path.exists(str(p))   # 환원 전 원본 없음
    ok, err = drv.rollback_legacy_skip(entry)
    assert ok, err
    assert os.path.isfile(str(p)), "rollback 으로 원본 watched 경로 복구"
    assert drv._sha256_file(str(p)) == original_sha
    assert not os.path.exists(entry["preserved_path"]), "환원 후 preserved 사본 제거(완전 가역)"


# ─────────────────────────────────────────────────────────────────────────────
# 6. sha256 mismatch → abort/explicit failure (원본 보존)
# ─────────────────────────────────────────────────────────────────────────────
def test_06_sha256_mismatch_aborts_preserves_original(tmp_path):
    root = _make_dirs(tmp_path)
    p = _write_result(root, "task-2314.result.json",
                      builder.build_schema_valid_result(task_id="task-2314"),
                      mtime=_LEGACY_MTIME)
    # before/after 가 다르게 보이도록 stateful hash_fn 주입(corruption 모사).
    seq = {"n": 0}

    def _bad_hash(_path):
        seq["n"] += 1
        return f"hash-{seq['n']}"   # 1회차(before) != 2회차(after) → mismatch

    rec = _process_legacy(root, p, legacy_hash_fn=_bad_hash)
    assert rec.verdict == drv.VERDICT_LEGACY_SKIP_PRESERVE_FAILED, rec.verdict
    assert "mismatch" in (rec.error or "")
    # 원본 보존 — watched 경로 유지.
    assert os.path.isfile(str(p)), "mismatch 시 원본 보존"
    # 사본/ manifest 미생성(orphan 0).
    assert not _manifest_path(root).exists() or \
        _manifest_path(root).read_text(encoding="utf-8").strip() == ""
    leftover = glob.glob(str(_legacy_skip_dir(root) / "task-*.result.json"))
    assert leftover == [], "mismatch 시 사본 정리"


# ─────────────────────────────────────────────────────────────────────────────
# 7. schema-valid controlled fixture 가 driver schema 통과 (DD3 보정)
# ─────────────────────────────────────────────────────────────────────────────
def test_07_schema_valid_fixture_passes_driver_schema(tmp_path):
    root = _make_dirs(tmp_path)
    payload = builder.build_schema_valid_result(task_id="task-2744-ok")
    # builder 자체 미러 검증.
    assert builder.is_schema_valid(payload) is True
    assert all(k in payload for k in builder.REQUIRED_SCHEMA_KEYS)
    assert isinstance(payload["completion_signal"], str) and payload["completion_signal"].strip()
    assert isinstance(payload["collector_envelope"], dict)

    # 실제 driver 게이트 통과: fresh(post-epoch) → 정상 path → pickup 도달(schema_fail 아님).
    p = _write_result(root, "task-2744-ok.result.json", payload, mtime=_FRESH_MTIME)
    pickup = make_pickup_mock("CLOSEOUT_DONE")
    verify = make_verify_mock("AUTHORITATIVE")
    rec = drv.process_one(
        str(p), root=str(root), pickup_fn=pickup, verify_fn=verify,
        legacy_cutoff=True, legacy_preserve_move=True, activation_epoch=_EPOCH,
        stable_sec=0.0, readiness_retries=1, sleep_fn=_NO_SLEEP,
    )
    assert rec.verdict != drv.VERDICT_QUARANTINE, f"schema-valid 인데 quarantine: {rec.error}"
    assert rec.verdict == drv.VERDICT_CLOSEOUT_DONE
    assert len(pickup.calls) == 1, "schema 통과 → pickup 도달"


# ─────────────────────────────────────────────────────────────────────────────
# 8. schema-invalid fixture → quarantine, watched inbox 에 남지 않아 폭주 0
# ─────────────────────────────────────────────────────────────────────────────
@pytest.mark.parametrize("kind", [
    "missing_completion_signal", "empty_completion_signal", "empty_task_id",
])
def test_08_schema_invalid_quarantined_not_in_inbox(tmp_path, kind):
    root = _make_dirs(tmp_path)
    payload = builder.build_schema_invalid_result(kind)
    assert builder.is_schema_valid(payload) is False
    p = _write_result(root, "task-2315.result.json", payload, mtime=_FRESH_MTIME)
    pickup = make_pickup_mock()
    rec = drv.process_one(
        str(p), root=str(root), pickup_fn=pickup, verify_fn=make_verify_mock(),
        legacy_cutoff=True, legacy_preserve_move=True, activation_epoch=_EPOCH,
        stable_sec=0.0, readiness_retries=1, sleep_fn=_NO_SLEEP,
    )
    assert rec.verdict == drv.VERDICT_QUARANTINE
    assert rec.quarantined is True
    # 폭주 0: quarantine 이동으로 watched inbox 가 비워짐.
    assert not os.path.exists(str(p))
    assert _inbox_glob(root) == []
    assert pickup.calls == [], "schema_fail → pickup 미도달"


# ─────────────────────────────────────────────────────────────────────────────
# 9. fresh(post-epoch) valid fixture 는 legacy_skip 으로 이동 0
# ─────────────────────────────────────────────────────────────────────────────
def test_09_fresh_valid_not_moved_to_legacy_skip(tmp_path):
    root = _make_dirs(tmp_path)
    payload = builder.build_schema_valid_result(task_id="task-2744-fresh")
    p = _write_result(root, "task-2744-fresh.result.json", payload, mtime=_FRESH_MTIME)
    pickup = make_pickup_mock("CLOSEOUT_DONE")
    rec = drv.process_one(
        str(p), root=str(root), pickup_fn=pickup, verify_fn=make_verify_mock(),
        legacy_cutoff=True, legacy_preserve_move=True, activation_epoch=_EPOCH,
        stable_sec=0.0, readiness_retries=1, sleep_fn=_NO_SLEEP,
    )
    assert rec.verdict != drv.VERDICT_LEGACY_SKIP_PRESERVED
    assert rec.verdict == drv.VERDICT_CLOSEOUT_DONE
    # legacy_skip 디렉토리로의 이동 0.
    assert not _legacy_skip_dir(root).exists() or \
        glob.glob(str(_legacy_skip_dir(root) / "task-*.result.json")) == []
    assert not _manifest_path(root).exists()


# ─────────────────────────────────────────────────────────────────────────────
# 10. PathExistsGlob 유지 상태에서도 legacy skip 누적 0 (처리 후 inbox 비워짐)
# ─────────────────────────────────────────────────────────────────────────────
def test_10_path_exists_glob_no_accumulation(tmp_path):
    root = _make_dirs(tmp_path)
    drv_enable = root / "memory" / "state" / "p0b_driver_enabled"
    drv_enable.write_text(drv.ACTIVATION_ENABLED + "\n", encoding="utf-8")
    names = [f"task-23{n}.result.json" for n in range(20, 29)]  # 9건(잔존 모사)
    for nm in names:
        p = _write_result(root, nm, builder.build_schema_valid_result(task_id=nm[:-12]),
                          mtime=_LEGACY_MTIME)
    assert len(_inbox_glob(root)) == 9

    pickup = make_pickup_mock()
    verify = make_verify_mock()
    records = drv.scan_once(
        str(root),
        pickup_fn=pickup, verify_fn=verify,
        legacy_cutoff=True, legacy_preserve_move=True, activation_epoch=_EPOCH,
        flag_reader=lambda: drv.ACTIVATION_ENABLED,
        write_evidence=False, sleep_fn=_NO_SLEEP,
    )
    processed = [r for r in records if r.result_path]
    assert len(processed) == 9
    assert all(r.verdict == drv.VERDICT_LEGACY_SKIP_PRESERVED for r in processed)
    # ★ 핵심: PathExistsGlob(level-trigger) 가 감시하는 watched inbox 가 완전히 비워짐.
    assert _inbox_glob(root) == [], "처리 후 watched inbox 비워짐 → 재트리거 0"
    # owner-pickup 처리 0(legacy 전부 preserve-move 단락).
    assert pickup.calls == []
    assert verify.calls == []
    # 2차 scan: 후보 0 → 처리 record 0(누적·폭주 0 증명).
    records2 = drv.scan_once(
        str(root),
        pickup_fn=pickup, verify_fn=verify,
        legacy_cutoff=True, legacy_preserve_move=True, activation_epoch=_EPOCH,
        flag_reader=lambda: drv.ACTIVATION_ENABLED,
        write_evidence=False, sleep_fn=_NO_SLEEP,
    )
    assert [r for r in records2 if r.result_path] == []
    assert pickup.calls == []
    # manifest 는 9건 append(가역 정보 보존).
    man_lines = [l for l in _manifest_path(root).read_text(encoding="utf-8").splitlines() if l.strip()]
    assert len(man_lines) == 9


# ─────────────────────────────────────────────────────────────────────────────
# 11. live fire 0 · cron 0 · ACTIVE=false
# ─────────────────────────────────────────────────────────────────────────────
def test_11_no_live_fire_no_cron_active_false(tmp_path):
    root = _make_dirs(tmp_path)
    # ACTIVE=false: activation flag 부재 → scan_once 즉시 NOOP_DISABLED, pickup 미호출.
    pickup = make_pickup_mock()
    records = drv.scan_once(
        str(root),
        pickup_fn=pickup, verify_fn=make_verify_mock(),
        legacy_cutoff=True, legacy_preserve_move=True, activation_epoch=_EPOCH,
        flag_reader=lambda: None,   # disabled
        write_evidence=False, sleep_fn=_NO_SLEEP,
    )
    assert len(records) == 1 and records[0].verdict == drv.VERDICT_NOOP_DISABLED
    assert pickup.calls == [], "ACTIVE=false → pickup 미호출(live fire 0)"

    # real-wake flag OFF(기본) → build_launcher_fn None → 실제 wake 결선 0.
    assert drv.build_launcher_fn(str(root), real_wake_reader=lambda: None) is None
    assert drv.build_launcher_fn(str(root), real_wake_reader=lambda: "disabled") is None

    # preserve-move path 자체도 launcher/relay/callback_launch 미주입(fire-capable 0):
    #   process_one 에 launcher_fn/relay_fn/callback_launch_fn 을 주입하지 않으므로
    #   preserve-move 는 어떤 fire executor 도 호출하지 않는다(구조적 fire 0).
    p = _write_result(root, "task-2316.result.json",
                      builder.build_schema_valid_result(task_id="task-2316"),
                      mtime=_LEGACY_MTIME)
    fired = {"launch": 0}

    def _spy_launch(*a, **k):
        fired["launch"] += 1
        return types.SimpleNamespace(decision="X")

    rec = _process_legacy(root, p, launcher_fn=_spy_launch, relay_fn=_spy_launch,
                          callback_launch_fn=None, owner_gate_fn=None)
    assert rec.verdict == drv.VERDICT_LEGACY_SKIP_PRESERVED
    assert fired["launch"] == 0, "legacy preserve-move 는 wake/relay/callback 발사 0(cron 0)"


# ─────────────────────────────────────────────────────────────────────────────
# 12. (Gemini medium) os.remove 지속 실패 → 사본 정리·원본 보존·무한 누적 0
# ─────────────────────────────────────────────────────────────────────────────
def test_12_remove_failure_cleans_copy_no_accumulation(tmp_path, monkeypatch):
    root = _make_dirs(tmp_path)
    p = _write_result(root, "task-2317.result.json",
                      builder.build_schema_valid_result(task_id="task-2317"),
                      mtime=_LEGACY_MTIME)
    orig_remove = os.remove

    def _fail_remove(target, *a, **k):
        # 원본(p)의 제거만 실패시키고, 사본 정리(_safe_unlink) 는 정상 동작하게 둔다.
        if os.path.abspath(str(target)) == os.path.abspath(str(p)):
            raise OSError("simulated permission denied")
        return orig_remove(target, *a, **k)

    monkeypatch.setattr(drv.os, "remove", _fail_remove)
    rec = _process_legacy(root, p)
    assert rec.verdict == drv.VERDICT_LEGACY_SKIP_PRESERVE_FAILED, rec.verdict
    assert "원본 제거 실패" in (rec.error or "")
    # 원본 보존(데이터 손실 0).
    assert os.path.isfile(str(p))
    # ★ 사본 정리 — legacy_skip 디렉토리에 누적 0.
    assert glob.glob(str(_legacy_skip_dir(root) / "task-*.result.json")) == []
    # manifest 미기록(orphan 0).
    assert not _manifest_path(root).exists() or \
        _manifest_path(root).read_text(encoding="utf-8").strip() == ""
    # 재시도(2회차)에도 동일 — 무한 누적 0.
    rec2 = _process_legacy(root, p)
    assert rec2.verdict == drv.VERDICT_LEGACY_SKIP_PRESERVE_FAILED
    assert glob.glob(str(_legacy_skip_dir(root) / "task-*.result.json")) == []