# -*- coding: utf-8 -*-
"""task-2731 — finish-callback 강제 결선 regression (통합).

설계: finish_callback_enforcement_followup_task_packet_260610 +
      task2731_finish_callback_enforcement_prep_packet_260610.

ANCHOR:
  ① executor 가 callback 안 쏘고 죽어도 OS runner/collector closeout → ANU-owned
     callback launch 를 코드가 강제(executor 세션 독립 · dev bot prompt 의존 0 ·
     finish-task 끝단 단독 의존 0).
  ② callback = ANU key 강제 · self-key launch 거부 · audit jsonl 기록.
  ③ durability = ledger→collector_result→pickup.done, result 1건당 callback 1회 idempotent.

커버리지 (fixture 8 + acceptance):
  - fixture-dev1     : result.json+report 존재·.done/callback 미생성 → 수거·callback 강제.
  - fixture-dev2     : worktree mismatch + branch -r2 + task-lock 마찰 → result.json 만으로 수거·callback.
  - fixture-callback-absent : callback cron 0 → terminal state + callback launch.
  - fixture-self-key : executor self-key → 거부(ANU key 강제) + audit success=False.
  - fixture-branch-rN: branch task-2731-r2 안전 인식(suffix 보존·base 충돌 0).
  - fixture-worktree-mismatch : CODE_ROOT 직접 → preflight 탐지.
  - fixture-active-false : flag OFF → 전면 no-op(audit 0) + closeout 은 유지.
  - fixture-idempotent : 동일 result.json 2회 → callback 1회(audit dedupe + ledger dedupe).
  - audit jsonl 필드(task_id·owner_key_class·launch_ts·schedule_id·success·executor_session_independent).
  - raw key 0 (audit jsonl 에 ANU/executor key literal 미기록).
  - driver pass-through(callback_launch_verdict 기록).
  - systemd unit 파일/dry-run 검증(install 0).

네트워크 0 · 전부 mock/fixture · tmpdir(isolated) · canonical 0 touch.
ANU key literal 노출 0 — 모듈 상수 / fake loader 주입으로만.
"""
from __future__ import annotations

import importlib.util
import json
import os
import sys
import tempfile
import unittest
from datetime import datetime, timezone
from pathlib import Path

_ROOT = Path(__file__).resolve().parent.parent
if str(_ROOT) not in sys.path:
    sys.path.insert(0, str(_ROOT))


def _load(modname: str, relpath: str):
    if modname in sys.modules:
        return sys.modules[modname]
    spec = importlib.util.spec_from_file_location(modname, _ROOT / relpath)
    assert spec is not None and spec.loader is not None
    mod = importlib.util.module_from_spec(spec)
    sys.modules[modname] = mod
    spec.loader.exec_module(mod)
    return mod


# 의존 모듈 실 로드 (heavy dispatch/__init__ 우회 — file-spec 로드).
_load("dispatch.callback_owner_enforcer", "dispatch/callback_owner_enforcer.py")
_load("dispatch.normal_fallback_callback_helper",
      "dispatch/normal_fallback_callback_helper.py")
M_enf = _load("dispatch.anu_owned_callback_enforcement",
              "dispatch/anu_owned_callback_enforcement.py")
CR = _load("dispatch.anu_collector_result", "dispatch/anu_collector_result.py")
R = _load("dispatch.anu_result_pickup_runner", "dispatch/anu_result_pickup_runner.py")
A = _load("dispatch.anu_callback_launch_audit", "dispatch/anu_callback_launch_audit.py")
P = _load("dispatch.anu_finish_preflight", "dispatch/anu_finish_preflight.py")
_load("dispatch.anu_pickup_wake_launcher", "dispatch/anu_pickup_wake_launcher.py")
_load("dispatch.anu_terminal_relay", "dispatch/anu_terminal_relay.py")
DRV = _load("dispatch.anu_pickup_driver", "dispatch/anu_pickup_driver.py")

_ANU_KEY = M_enf.ANU_KEY            # 독립 ANU key (모듈 상수 — literal 노출 0)
_DEV_KEY = "7943afbe12c12f7d"       # executor(dev) self-key (테스트 전용 더미)
_NOW = datetime(2026, 6, 10, 5, 0, 0, tzinfo=timezone.utc)


def _clock():
    return _NOW


def _anu_loader():
    return _ANU_KEY


def _flag_enabled():
    return A.CALLBACK_LAUNCH_ENABLED


def _flag_disabled():
    return None


class _FakeRR:
    """anu_runner_pickup_and_fire 대체(hermetic). raw key literal 미포함 argv."""

    def __init__(self, ok=True, task_id="task-2731"):
        self.ok = ok
        self.verdict = "PASS" if ok else "FAIL"
        self.task_id = task_id
        self.argv = (["cokacdir", "--cron", "PROMPT", "--at", "60m"] if ok else None)
        self.reasons = [] if ok else ["owner enforcement 미통과(fake)."]


def _fake_fire(**kwargs):
    return _FakeRR(ok=True, task_id=str(kwargs.get("result_json_path", "")) or "task-2731")


def _write_result_json(path, task_id="task-2731", *, terminal_intent="completed",
                       callback_schedule_created=False, relay_hints=None,
                       schedule_id="SID-2731", with_envelope=True):
    payload = {
        "task_id": task_id,
        "completion_signal": "result_json_written",
        "terminal_intent": terminal_intent,
        "callback_schedule_created": callback_schedule_created,
    }
    if schedule_id:
        payload["schedule_id"] = schedule_id
    if with_envelope:
        payload["collector_envelope"] = {"schedule_id": schedule_id,
                                         "collector_role": "ANU"}
    if relay_hints is not None:
        payload["relay_hints"] = relay_hints
    with open(path, "w", encoding="utf-8") as fh:
        json.dump(payload, fh, ensure_ascii=False)
    return path


class FinishCallbackEnforcementTest(unittest.TestCase):
    def setUp(self):
        self._tmp = tempfile.mkdtemp(prefix="task2731_")
        self.root = self._tmp
        self.events = os.path.join(self.root, "memory", "events")
        os.makedirs(self.events, exist_ok=True)
        self.audit_path = os.path.join(self.root, "memory", "p0b_state",
                                       "callback_launch_audit.jsonl")
        self.ledger = os.path.join(self.root, "memory", "events",
                                   "callback_4tuple_index.jsonl")

    def tearDown(self):
        import shutil
        shutil.rmtree(self._tmp, ignore_errors=True)

    # ── 공통 헬퍼 ────────────────────────────────────────────────────────────
    def _launch_fn(self, *, flag_reader=None, fire=_fake_fire):
        return A.build_callback_launch_fn(
            self.root,
            audit_path=self.audit_path,
            flag_reader=flag_reader or _flag_enabled,
            anu_key_loader=_anu_loader,
            clock=_clock,
            runner_fire_fn=fire,
        )

    def _pickup(self, result_path, *, executor_key=_DEV_KEY, launch_fn=None,
                ledger_path=None):
        return R.pickup_once(
            result_path,
            sealed_key_loader=_anu_loader,
            executor_key=executor_key,
            ledger_path=ledger_path or self.ledger,
            callback_launch_fn=launch_fn,
        )

    def _audit_lines(self):
        if not os.path.isfile(self.audit_path):
            return []
        with open(self.audit_path, "r", encoding="utf-8") as fh:
            return [json.loads(l) for l in fh if l.strip()]

    # ── fixture-dev1 ─────────────────────────────────────────────────────────
    def test_fixture_dev1_result_and_report_no_done_no_callback(self):
        """dev1: result.json+report 존재·.done/callback 미생성 → 수거·callback 강제."""
        rp = os.path.join(self.events, "task-2731.result.json")
        _write_result_json(rp, "task-2731")
        # report 존재(executor 작성), 그러나 .done/callback 미생성 상태 재현.
        with open(os.path.join(self.root, "task-2731.report.md"), "w") as fh:
            fh.write("# report\n")
        self.assertFalse(os.path.exists(os.path.join(self.events, "task-2731.pickup.done")))

        res = self._pickup(rp, launch_fn=self._launch_fn())
        self.assertEqual(res.verdict, R.PICKUP_CLOSEOUT_DONE)
        # collector_result + terminal marker(pickup.done) 생성.
        self.assertTrue(os.path.exists(os.path.join(self.events, "task-2731.pickup.done")))
        self.assertTrue(os.path.isfile(res.closeout_path))
        # ANU-owned callback launch 강제됨(executor 미발사여도).
        self.assertEqual(res.callback_launch_verdict, A.LAUNCH_ENFORCED)
        lines = self._audit_lines()
        self.assertEqual(len(lines), 1)
        self.assertEqual(lines[0]["owner_key_class"], A.OKC_ANU)
        self.assertTrue(lines[0]["success"])
        self.assertTrue(lines[0]["executor_session_independent"])

    # ── fixture-dev2 (worktree mismatch + branch -r2 + task-lock 마찰) ────────
    def test_fixture_dev2_worktree_mismatch_branch_r2_tasklock(self):
        """dev2: worktree mismatch + branch -r2 + task-lock 마찰 → result.json 만으로 수거·callback."""
        pf = P.preflight(
            branch="task/task-2730-r2",
            actual_path="/home/jay/p0b-pickup-main",   # CODE_ROOT 직접(.worktrees 미사용)
            code_root="/home/jay/p0b-pickup-main",
            result_json_present=True,
            lock_contended=True,
            lock_path="/tmp/task-2730.lock",
        )
        # branch -r2 안전 인식: base 단독 충돌 회피용 full_task_ref 분리.
        self.assertEqual(pf.branch.base_task_id, "task-2730")
        self.assertEqual(pf.branch.full_task_ref, "task-2730-r2")
        self.assertTrue(pf.branch.is_revision)
        # worktree mismatch 탐지.
        self.assertTrue(pf.worktree.mismatch)
        # task-lock 마찰에도 수거 가능(lock 독립).
        self.assertTrue(pf.lock_bypass.bypass_ok)
        self.assertTrue(pf.lock_bypass.lock_contended)
        self.assertTrue(pf.collectible)

        # 실제 수거·callback: lock/worktree 마찰과 무관하게 runner 가 closeout+callback.
        rp = os.path.join(self.events, "task-2730-r2.result.json")
        _write_result_json(rp, "task-2730-r2")
        res = self._pickup(rp, launch_fn=self._launch_fn())
        self.assertEqual(res.verdict, R.PICKUP_CLOSEOUT_DONE)
        self.assertEqual(res.callback_launch_verdict, A.LAUNCH_ENFORCED)

    # ── fixture-callback-absent ──────────────────────────────────────────────
    def test_fixture_callback_absent(self):
        """callback cron 0(executor 미발사) → terminal state + ANU callback launch."""
        rp = os.path.join(self.events, "task-2731.result.json")
        _write_result_json(rp, "task-2731")
        res = self._pickup(rp, launch_fn=self._launch_fn())
        self.assertTrue(res.ok)
        self.assertEqual(res.callback_launch_verdict, A.LAUNCH_ENFORCED)
        self.assertEqual(len(self._audit_lines()), 1)

    # ── fixture-self-key (거부) ──────────────────────────────────────────────
    def test_fixture_self_key_refused(self):
        """executor self-key(ANU key)로 launch 시도 → 거부(ANU key 강제) + audit success=False."""
        rp = os.path.join(self.events, "task-2731.result.json")
        _write_result_json(rp, "task-2731")
        # executor_key == ANU key → self-key.
        res = self._pickup(rp, executor_key=_ANU_KEY, launch_fn=self._launch_fn())
        # closeout 자체는 완료(ANCHOR: callback 거부와 무관하게 terminal state 생성).
        self.assertTrue(res.ok)
        self.assertEqual(res.callback_launch_verdict, A.LAUNCH_SELF_KEY_REFUSED)
        lines = self._audit_lines()
        self.assertEqual(len(lines), 1)
        self.assertEqual(lines[0]["verdict"], A.LAUNCH_SELF_KEY_REFUSED)
        self.assertEqual(lines[0]["owner_key_class"], A.OKC_SELF)
        self.assertFalse(lines[0]["success"])

    def test_self_key_direct_launch_refused(self):
        """launch_callback_for_closeout 단독 self-key 거부 검증."""
        rp = os.path.join(self.events, "task-2731.result.json")
        _write_result_json(rp, "task-2731")
        r = A.launch_callback_for_closeout(
            rp, task_id="task-2731", sha256="abc", executor_key=_ANU_KEY,
            root=self.root, audit_path=self.audit_path,
            flag_reader=_flag_enabled, clock=_clock, anu_key_loader=_anu_loader,
            runner_fire_fn=_fake_fire,
        )
        self.assertEqual(r.verdict, A.LAUNCH_SELF_KEY_REFUSED)
        self.assertFalse(r.success)

    # ── fixture-branch-rN ────────────────────────────────────────────────────
    def test_fixture_branch_rN_safe_extraction(self):
        """branch task-2731-r2 / +1 / dev6 안전 인식(suffix 보존·base 충돌 0)."""
        self.assertEqual(P.parse_branch("task-2731-r2").full_task_ref, "task-2731-r2")
        self.assertEqual(P.parse_branch("task-2731-r2").base_task_id, "task-2731")
        self.assertTrue(P.parse_branch("task-2731-r2").is_revision)
        self.assertEqual(P.parse_branch("task/task-2731+1").full_task_ref, "task-2731+1")
        bp = P.parse_branch("task/task-2731-dev6")
        self.assertEqual(bp.base_task_id, "task-2731")
        self.assertEqual(bp.executor, "dev6")
        self.assertFalse(bp.is_revision)
        # base 단독 추출이 revision 과 다른 정체성을 가져야 함(dev1 state 충돌 0).
        self.assertNotEqual(
            P.parse_branch("task-2730-r2").full_task_ref,
            P.parse_branch("task-2730").full_task_ref,
        )
        # non-task ref.
        self.assertFalse(P.parse_branch("main").ok)
        self.assertFalse(P.parse_branch("").ok)

    # ── fixture-worktree-mismatch ────────────────────────────────────────────
    def test_fixture_worktree_mismatch(self):
        """CODE_ROOT 직접 → mismatch=True / 정상 worktree → mismatch=False."""
        bad = P.detect_worktree_mismatch(
            "/home/jay/p0b-pickup-main", task_ref="task-2731", executor="dev6",
            code_root="/home/jay/p0b-pickup-main")
        self.assertTrue(bad.mismatch)
        good = P.detect_worktree_mismatch(
            "/home/jay/p0b-pickup-main/.worktrees/task-2731-dev6",
            task_ref="task-2731", executor="dev6")
        self.assertFalse(good.mismatch)
        self.assertTrue(good.ok)

    # ── fixture-active-false (no-op) ─────────────────────────────────────────
    def test_fixture_active_false_noop(self):
        """flag OFF → callback launch 전면 no-op(audit 0)·closeout 은 유지."""
        rp = os.path.join(self.events, "task-2731.result.json")
        _write_result_json(rp, "task-2731")
        res = self._pickup(rp, launch_fn=self._launch_fn(flag_reader=_flag_disabled))
        # closeout(terminal state) 은 정상 완료.
        self.assertTrue(res.ok)
        self.assertTrue(os.path.exists(os.path.join(self.events, "task-2731.pickup.done")))
        # callback launch 는 no-op.
        self.assertEqual(res.callback_launch_verdict, A.LAUNCH_NOOP_DISABLED)
        # audit jsonl 미생성(전면 no-op).
        self.assertFalse(os.path.isfile(self.audit_path))

    def test_active_false_default_no_flag_file(self):
        """flag 파일 부재(기본) → no-op (read_callback_launch_enabled False)."""
        self.assertFalse(A.read_callback_launch_enabled(self.root))

    # ── fixture-idempotent (2회 → 1회) ───────────────────────────────────────
    def test_fixture_idempotent_two_calls_one_callback(self):
        """동일 result.json 2회 처리 → callback 1회(ledger dedupe + audit dedupe)."""
        rp = os.path.join(self.events, "task-2731.result.json")
        _write_result_json(rp, "task-2731")
        lf = self._launch_fn()
        res1 = self._pickup(rp, launch_fn=lf)
        self.assertEqual(res1.verdict, R.PICKUP_CLOSEOUT_DONE)
        self.assertEqual(res1.callback_launch_verdict, A.LAUNCH_ENFORCED)
        # marker 존재 → 2회차는 SKIP_TERMINAL (재발사 0).
        res2 = self._pickup(rp, launch_fn=lf)
        self.assertEqual(res2.verdict, R.PICKUP_SKIP_TERMINAL)
        # callback launch 는 1회만 강제됨.
        enforced = [l for l in self._audit_lines() if l["verdict"] == A.LAUNCH_ENFORCED]
        self.assertEqual(len(enforced), 1)

    def test_audit_dedupe_same_sha(self):
        """audit dedupe: 동일 (task_id, sha256) 직접 2회 launch → 2회차 SKIP_DEDUPE."""
        rp = os.path.join(self.events, "task-2731.result.json")
        _write_result_json(rp, "task-2731")
        kw = dict(task_id="task-2731", sha256="deadbeef", executor_key=_DEV_KEY,
                  root=self.root, audit_path=self.audit_path,
                  flag_reader=_flag_enabled, clock=_clock,
                  anu_key_loader=_anu_loader, runner_fire_fn=_fake_fire)
        r1 = A.launch_callback_for_closeout(rp, **kw)
        r2 = A.launch_callback_for_closeout(rp, **kw)
        self.assertEqual(r1.verdict, A.LAUNCH_ENFORCED)
        self.assertEqual(r2.verdict, A.LAUNCH_SKIP_DEDUPE)
        # ENFORCED audit 1줄만(SKIP_DEDUPE 는 audit 미기록).
        self.assertEqual(len([l for l in self._audit_lines()
                              if l["verdict"] == A.LAUNCH_ENFORCED]), 1)

    def test_transient_launch_failure_retried_no_permanent_skip(self):
        """전이 launch 실패 후 marker 존재해도 다음 reprocess 에서 재시도(영구 skip 0).

        Codex CRITICAL 회귀: marker 후 callback 전이 실패가 영구 비재시도되지 않는다.
        audit-dedupe 로 최종 ENFORCED 1회만(이중 callback 0)."""
        rp = os.path.join(self.events, "task-2731.result.json")
        _write_result_json(rp, "task-2731")

        calls = {"n": 0}

        def _flaky_launch(result_json_path, **kw):
            calls["n"] += 1
            verdict = A.LAUNCH_FAIL if calls["n"] == 1 else A.LAUNCH_ENFORCED
            res = A.CallbackLaunchResult(
                schema=A.CALLBACK_LAUNCH_AUDIT_SCHEMA, verdict=verdict,
                task_id=kw.get("task_id", ""), sha256=kw.get("sha256", ""),
                owner_key_class=A.OKC_ANU, success=(verdict == A.LAUNCH_ENFORCED),
                executor_session_independent=True, audit_path=self.audit_path)
            return res

        # pass1: closeout 완료(marker) + launch 전이 실패.
        res1 = self._pickup(rp, launch_fn=_flaky_launch)
        self.assertEqual(res1.verdict, R.PICKUP_CLOSEOUT_DONE)
        self.assertEqual(res1.callback_launch_verdict, A.LAUNCH_FAIL)
        self.assertTrue(os.path.exists(os.path.join(self.events, "task-2731.pickup.done")))
        # pass2: marker 존재(SKIP_TERMINAL) 이지만 callback 재시도 → ENFORCED.
        res2 = self._pickup(rp, launch_fn=_flaky_launch)
        self.assertEqual(res2.verdict, R.PICKUP_SKIP_TERMINAL)
        self.assertEqual(res2.callback_launch_verdict, A.LAUNCH_ENFORCED)
        # launch fn 이 2회 호출됨(전이 실패 → 재시도).
        self.assertEqual(calls["n"], 2)

    # ── audit jsonl 필드 + raw key 0 ─────────────────────────────────────────
    def test_audit_jsonl_fields_and_no_raw_key(self):
        """audit 필드 완비 + raw key(ANU/executor) literal 미기록."""
        rp = os.path.join(self.events, "task-2731.result.json")
        _write_result_json(rp, "task-2731", schedule_id="SID-XYZ")
        self._pickup(rp, launch_fn=self._launch_fn())
        lines = self._audit_lines()
        self.assertEqual(len(lines), 1)
        rec = lines[0]
        for key in ("task_id", "owner_key_class", "launch_ts", "schedule_id",
                    "success", "executor_session_independent"):
            self.assertIn(key, rec)
        self.assertEqual(rec["task_id"], "task-2731")
        self.assertEqual(rec["schedule_id"], "SID-XYZ")
        # raw key literal 미기록.
        blob = json.dumps(rec, ensure_ascii=False)
        self.assertNotIn(_ANU_KEY, blob)
        self.assertNotIn(_DEV_KEY, blob)

    # ── ANCHOR-1: executor 세션 독립 ─────────────────────────────────────────
    def test_executor_session_independent_always_true(self):
        """모든 launch 결과는 executor_session_independent=True (OS runner/collector 경로)."""
        rp = os.path.join(self.events, "task-2731.result.json")
        _write_result_json(rp, "task-2731")
        r = A.launch_callback_for_closeout(
            rp, task_id="task-2731", sha256="x", executor_key=_DEV_KEY,
            root=self.root, audit_path=self.audit_path, flag_reader=_flag_enabled,
            clock=_clock, anu_key_loader=_anu_loader, runner_fire_fn=_fake_fire)
        self.assertTrue(r.executor_session_independent)

    # ── relay-path 도 callback launch 강제 ───────────────────────────────────
    def test_relay_path_also_launches_callback(self):
        """relay_hints truthy(relay-path WAKE_BUILT) 도 callback launch 강제."""
        rp = os.path.join(self.events, "task-2731.result.json")
        _write_result_json(rp, "task-2731", relay_hints={"gemini_finding": True})
        # relay-path 는 wake argv 빌드(anu_runner_pickup_and_fire 실호출) — 실패해도
        # callback launch 강제는 별개. 여기선 closeout 진입 후 launch verdict 확인.
        res = self._pickup(rp, launch_fn=self._launch_fn())
        # relay-path 빌드 성공 시 WAKE_BUILT, 실패 시 FAIL — 어느 쪽이든 launch 시도됨은
        # WAKE_BUILT 일 때만(fresh closeout). WAKE_BUILT 면 callback 강제 검증.
        if res.verdict == R.PICKUP_WAKE_BUILT:
            self.assertEqual(res.callback_launch_verdict, A.LAUNCH_ENFORCED)

    # ── driver pass-through ──────────────────────────────────────────────────
    def test_driver_passes_callback_launch_fn(self):
        """driver.process_one 이 callback_launch_fn 을 pickup 에 전달 → record 에 verdict 기록."""
        rp = os.path.join(self.events, "task-2731.result.json")
        _write_result_json(rp, "task-2731")

        captured = {}

        def _spy_launch(_result_json_path, **kw):
            captured.update(kw)
            return A.CallbackLaunchResult(
                schema=A.CALLBACK_LAUNCH_AUDIT_SCHEMA,
                verdict=A.LAUNCH_ENFORCED, task_id=kw.get("task_id", ""),
                sha256=kw.get("sha256", ""), owner_key_class=A.OKC_ANU,
                success=True, executor_session_independent=True,
                audit_path=self.audit_path)

        # verify_fn 주입으로 AUTHORITATIVE 강제(owner-proof read-only 보존).
        class _V:
            verdict = M_enf.VERDICT_AUTHORITATIVE
            owner_resolution = {"outcome": "OWNER_ANU", "schedule_id": "SID-2731",
                                "query_ok": True}

        def _verify(**_kw):
            return _V()

        # clock 미주입(실 now) → readiness age ≈ 0 통과(고정 과거 clock 회피).
        rec = DRV.process_one(
            rp, root=self.root,
            callback_launch_fn=_spy_launch,
            verify_fn=_verify,
            executor_key=_DEV_KEY,
            ledger_path=self.ledger,
            stable_sec=0.0, readiness_retries=1,
        )
        # pickup_once 가 spy 를 호출하여 task_id/sha256/terminal_intent 전달했는지.
        self.assertIn("task_id", captured)
        self.assertEqual(captured["task_id"], "task-2731")
        self.assertIn("terminal_intent", captured)
        # DriverRecord 에 callback_launch_verdict 기록.
        self.assertEqual(rec.callback_launch_verdict, A.LAUNCH_ENFORCED)

    # ── terminal_intent contract ─────────────────────────────────────────────
    def test_terminal_intent_contract(self):
        """result-json terminal_intent + callback_schedule_created 코드 schema 읽기."""
        self.assertEqual(R.read_terminal_intent({}),
                         (R.TERMINAL_INTENT_DEFAULT, False))
        self.assertEqual(
            R.read_terminal_intent({"terminal_intent": "needs_relay",
                                    "callback_schedule_created": True}),
            ("needs_relay", True))
        rp = os.path.join(self.events, "task-2731.result.json")
        _write_result_json(rp, "task-2731", terminal_intent="needs_relay")
        res = self._pickup(rp, launch_fn=self._launch_fn())
        self.assertEqual(res.terminal_intent, "needs_relay")

    # ── durability order: ledger → collector_result → marker ─────────────────
    def test_durability_order_artifacts(self):
        """closeout 산출물: collector_result + pickup.done(marker) + ledger 모두 생성."""
        rp = os.path.join(self.events, "task-2731.result.json")
        _write_result_json(rp, "task-2731")
        res = self._pickup(rp, launch_fn=self._launch_fn())
        self.assertTrue(os.path.isfile(self.ledger))
        self.assertTrue(os.path.isfile(res.closeout_path))
        self.assertTrue(os.path.exists(os.path.join(self.events, "task-2731.pickup.done")))


class SystemdUnitDryRunTest(unittest.TestCase):
    """systemd unit 파일/dry-run 검증만(install/enable/start 0)."""

    def test_systemd_unit_files_present_and_disabled(self):
        deploy = _ROOT / "deploy" / "systemd"
        for name in ("anu-pickup.path", "anu-pickup.timer", "anu-pickup.service"):
            self.assertTrue((deploy / name).is_file(), f"{name} 부재")
        # path/timer 는 같은 service 를 호출하고 install 미전제(주석 명시).
        path_txt = (deploy / "anu-pickup.path").read_text(encoding="utf-8")
        self.assertIn("PathExistsGlob", path_txt)
        self.assertIn("anu-pickup.service", path_txt)
        timer_txt = (deploy / "anu-pickup.timer").read_text(encoding="utf-8")
        self.assertIn("OnUnitActiveSec", timer_txt)

    def test_no_systemd_install_side_effect(self):
        """본 테스트 스위트는 systemctl 호출 0 — 파일 검증만(install/enable/start 0)."""
        # 정적 검증: deploy 디렉토리에 .wants 심볼릭/enable 흔적 없음.
        deploy = _ROOT / "deploy" / "systemd"
        for child in deploy.iterdir():
            self.assertFalse(child.is_symlink(), f"예상치 못한 enable 심볼릭: {child}")


if __name__ == "__main__":
    unittest.main()
