# -*- coding: utf-8 -*-
"""task-2717 ANU_OWNED_CALLBACK_ENFORCEMENT — 필수 regression (6).

회장 verbatim: "문서/메모리 박제로 끝내지 말고 callback 생성 권한을 executor
에게 주지 않는 실행 구조로 바꾼다." — 본 테스트는 실행 코드(runner/helper +
owner-key authority + quarantine path)를 직접 호출하여 입증한다. mock-only 금지:
실 cron-history 응답 형태(원시 cokacdir json)를 fake probe 로 주입하고, 실 spoof
envelope(텍스트 ANU, 실제 self-key)를 입력 → 자동 격리 출력을 실행으로 증명한다.

regression
  1. dev self-key spoof envelope (텍스트 ANU, 실제 self-key) → QUARANTINE
  2. access denied schedule (ANU key 조회 거부) → 비-ANU 판정
  3. stale envelope 재사용 → reject
  4. ANU-owned 정상 callback → PASS (authoritative)
  5. result JSON pickup 정상 경로 → ANU runner 가 발사 request 생성
  6. self-collector quarantine 경로 → 격리 + 비차단(작업물 보존) + 비권위
"""
from __future__ import annotations

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

_ROOT = Path(__file__).resolve().parent.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


# 실 runtime entrypoint 직접 호출 (재사용 의존 enforcer/helper 도 실모듈 로드).
_load("dispatch.callback_owner_enforcer",
      "dispatch/callback_owner_enforcer.py")
_load("dispatch.normal_fallback_callback_helper",
      "dispatch/normal_fallback_callback_helper.py")
M = _load("dispatch.anu_owned_callback_enforcement",
          "dispatch/anu_owned_callback_enforcement.py")

ANU_KEY = M.ANU_KEY                       # 독립 ANU key (import 상수 — 실 literal 비노출)
DEV4_SELF_KEY = "7943afbe12c12f7d"        # incident: dev4 self-collector key
ANU_SCHEDULE = "ANUOWNED1"                # ANU 가 소유한 schedule (조회 성공)
SELF_SCHEDULE = "A372DD6B"                # incident: self-key 가 소유한 schedule
# task-2717+1 (A): 실 ANU key literal 비노출 — spoof envelope 텍스트 주장에는
# placeholder 만 쓴다(어차피 기록만, 판정 미사용).
ANU_KEY_PLACEHOLDER = "<ANU_KEY_REDACTED>"


# ── fake cron-history probe (실 cokacdir json 형태 그대로 재현) ──────────────
def make_probe(owned_schedules):
    """owned_schedules: ANU key 로 조회 성공하는 sid 집합.

    cokacdir --cron-history 의 실제 응답을 재현한다:
      * ANU 소유      → {"status":"ok","id":sid,"count":1,"history":[...]}
      * 비-ANU/부재   → {"status":"error","message":"schedule not found or
                          access denied: <sid>"}  (실측 메시지)
    """
    owned = set(owned_schedules)

    def _probe(schedule_id: str) -> dict:
        if schedule_id in owned:
            return {
                "status": "ok",
                "id": schedule_id,
                "count": 1,
                "history": [{"status": "ok", "ts": "2026-05-31 01:10:00"}],
            }
        return {
            "status": "error",
            "message": f"schedule not found or access denied: {schedule_id}",
        }

    return _probe


def _now():
    return datetime(2026, 5, 31, 2, 0, 0, tzinfo=timezone.utc)


def _fresh_ts():
    return (_now() - timedelta(minutes=5)).strftime("%Y-%m-%dT%H:%M:%SZ")


class Regression2717(unittest.TestCase):

    def setUp(self):
        # quarantine artifact 는 격리 tmp dir 에만 — 실 운영 evidence 비접촉.
        self._tmp = tempfile.mkdtemp(prefix="task2717-q-")
        # tmpdir 누수 방지 — 테스트 종료 시 자동 정리(ignore_errors).
        self.addCleanup(shutil.rmtree, self._tmp, ignore_errors=True)

    # ── regression 1 — dev self-key spoof envelope → QUARANTINE ──────────────
    def test_1_dev_self_key_spoof_envelope_quarantine(self):
        """★ 실 spoof envelope 입력 → 자동격리 출력 증명.

        envelope 텍스트는 incident 처럼 ANU 를 주장(self_key_used=false,
        collector_role=ANU, owner_key=ANU). 그러나 실제 schedule(SELF_SCHEDULE)
        은 ANU key 로 조회 거부 → 텍스트 무시하고 SELF_COLLECTOR_QUARANTINE.
        """
        spoof_envelope = {
            "task_id": "task-2717",
            "schedule_id": SELF_SCHEDULE,
            "self_key_used": False,               # 거짓 주장
            "collector_role": "ANU",              # 거짓 주장
            "owner_key": ANU_KEY,                 # 거짓 주장 (텍스트만)
            "recorded_at": _fresh_ts(),
            "note": f"ANU key {ANU_KEY_PLACEHOLDER} 사용",  # 텍스트 주장(기록만)
        }
        res = M.verify_collector_authoritative(
            task_id="task-2717",
            envelope=spoof_envelope,
            probe=make_probe({ANU_SCHEDULE}),     # SELF_SCHEDULE 은 비-소유
            executor_key=DEV4_SELF_KEY,
            now=_now(),
            quarantine_dir=self._tmp,
        )
        # ★ C 강화: 반드시 SELF_COLLECTOR_QUARANTINE · usable_for_auto_merge=False
        #   · work_preserved=True 3종을 동시에 고정.
        self.assertEqual(res.verdict, M.VERDICT_QUARANTINED)
        self.assertEqual(res.classification, M.CLS_SELF_COLLECTOR_QUARANTINE)
        self.assertFalse(res.usable_for_auto_merge)
        self.assertTrue(res.work_preserved)
        # ★ envelope claim 은 '기록만, 판정 미사용' — claim 은 ANU 를 주장하지만
        #   판정의 근거는 actual owner proof(access denied) 다.
        self.assertEqual(res.envelope_claims["claimed_collector_role"], "ANU")
        self.assertEqual(res.envelope_claims["claimed_self_key_used"], False)
        self.assertEqual(res.envelope_claims["claimed_owner_key"], ANU_KEY)
        self.assertEqual(res.owner_resolution["outcome"], M.OWNER_NOT_ANU)
        self.assertFalse(res.owner_resolution["owner_is_anu"])
        self.assertIn("access denied",
                      res.owner_resolution["raw_message"].lower())
        # 자동 격리 증거 artifact 가 실제로 기록됨.
        self.assertIsNotNone(res.quarantine_record_path)
        self.assertTrue(Path(res.quarantine_record_path).is_file())
        rec = json.loads(Path(res.quarantine_record_path).read_text("utf-8"))
        self.assertEqual(rec["record_type"], "SELF_COLLECTOR_QUARANTINE")
        self.assertFalse(rec["usable_for_auto_merge"])
        self.assertTrue(rec["work_preserved"])

    # ── regression 2 — access denied schedule → 비-ANU 판정 ──────────────────
    def test_2_access_denied_schedule_non_anu(self):
        res = M.resolve_authoritative_owner(
            SELF_SCHEDULE, probe=make_probe(set()),  # 아무것도 소유 안 함
        )
        self.assertFalse(res.owner_is_anu)
        self.assertEqual(res.outcome, M.OWNER_NOT_ANU)
        self.assertTrue(res.query_ok)  # access-denied 는 결정적 신호
        self.assertIn("access denied", res.raw_message.lower())

    # ── regression 3 — stale envelope 재사용 → reject ────────────────────────
    def test_3_stale_envelope_reject(self):
        # (a) 너무 오래된 envelope → STALE_ENVELOPE_REJECTED
        stale = {
            "task_id": "task-2717",
            "schedule_id": ANU_SCHEDULE,
            "owner_key": ANU_KEY,
            "recorded_at": (_now() - timedelta(hours=48)).strftime(
                "%Y-%m-%dT%H:%M:%SZ"
            ),
        }
        res = M.verify_collector_authoritative(
            task_id="task-2717", envelope=stale,
            probe=make_probe({ANU_SCHEDULE}), now=_now(),
            quarantine_dir=self._tmp,
        )
        self.assertEqual(res.verdict, M.VERDICT_REJECTED)
        self.assertEqual(res.classification, M.CLS_STALE_ENVELOPE_REJECTED)
        self.assertFalse(res.usable_for_auto_merge)

        # (b) 다른 task 의 receipt 재사용 (task_id mismatch) → reject
        foreign = {
            "task_id": "task-9999",
            "schedule_id": ANU_SCHEDULE,
            "owner_key": ANU_KEY,
            "recorded_at": _fresh_ts(),
        }
        res2 = M.verify_collector_authoritative(
            task_id="task-2717", envelope=foreign,
            probe=make_probe({ANU_SCHEDULE}), now=_now(),
            quarantine_dir=self._tmp,
        )
        self.assertEqual(res2.verdict, M.VERDICT_REJECTED)
        self.assertEqual(res2.classification, M.CLS_ENVELOPE_BINDING_MISMATCH)

    # ── regression 4 — ANU-owned 정상 callback → PASS (authoritative) ────────
    def test_4_anu_owned_authoritative_pass(self):
        envelope = {
            "task_id": "task-2717",
            "schedule_id": ANU_SCHEDULE,
            "owner_key": ANU_KEY,
            "collector_role": "ANU",
            "recorded_at": _fresh_ts(),
        }
        res = M.verify_collector_authoritative(
            task_id="task-2717", envelope=envelope,
            probe=make_probe({ANU_SCHEDULE}),  # ANU 가 실제 소유
            executor_key=DEV4_SELF_KEY, now=_now(),
            quarantine_dir=self._tmp,
        )
        self.assertEqual(res.verdict, M.VERDICT_AUTHORITATIVE)
        self.assertEqual(res.classification, M.CLS_ANU_OWNED_AUTHORITATIVE)
        self.assertTrue(res.ok)
        self.assertTrue(res.usable_for_auto_merge)  # 권위 → 자동 판단 사용 가능
        self.assertIsNone(res.quarantine_record_path)  # 격리 아님

    # ── regression 5 — result JSON pickup 정상 → ANU runner 발사 ─────────────
    def test_5_result_json_pickup_anu_runner_fires(self):
        import tempfile
        rdir = tempfile.mkdtemp(prefix="task2717-r-")
        self.addCleanup(shutil.rmtree, rdir, ignore_errors=True)
        # (executor) 완료 = result JSON 작성 only — schedule 생성/발사 0.
        ew = M.executor_write_result_json(
            task_id="task-2717",
            payload={"report_path": "memory/reports/task-2717.md",
                     "sha256": "deadbeef", "summary": "done"},
            result_dir=rdir, now=_now(),
        )
        self.assertTrue(ew.ok)
        self.assertFalse(ew.schedule_created)
        self.assertFalse(ew.callback_fired)
        self.assertTrue(Path(ew.result_json_path).is_file())
        written = json.loads(Path(ew.result_json_path).read_text("utf-8"))
        self.assertEqual(written["completion_signal"], "RESULT_JSON_WRITTEN")
        self.assertFalse(written["schedule_created_by_executor"])

        # (ANU runner) result JSON pickup → ANU-owned 발사 request(argv) 생성.
        rr = M.anu_runner_pickup_and_fire(
            result_json_path=ew.result_json_path,
            executor_key=DEV4_SELF_KEY,
            anu_key=ANU_KEY,
            chat_id="6937032012",
            at="30s",
        )
        self.assertTrue(rr.ok)
        self.assertEqual(rr.owner_key, ANU_KEY)
        self.assertIsNotNone(rr.argv)
        # argv 가 ANU key 로 --cron 등록하는 데이터인지 — owner != executor.
        self.assertIn("--key", rr.argv)
        self.assertEqual(rr.argv[rr.argv.index("--key") + 1], ANU_KEY)
        self.assertNotIn(DEV4_SELF_KEY, rr.argv)

    def test_5b_runner_refuses_non_anu_key(self):
        """runner 가 self-key/비-ANU key 로는 발사 request 를 만들지 않음."""
        import tempfile
        rdir = tempfile.mkdtemp(prefix="task2717-r2-")
        self.addCleanup(shutil.rmtree, rdir, ignore_errors=True)
        ew = M.executor_write_result_json(
            task_id="task-2717", payload={"summary": "x"},
            result_dir=rdir, now=_now(),
        )
        # anu_key 자리에 executor self-key 를 넣어도 fail-closed (argv=None).
        rr = M.anu_runner_pickup_and_fire(
            result_json_path=ew.result_json_path,
            executor_key=DEV4_SELF_KEY,
            anu_key=DEV4_SELF_KEY,  # ← 비-ANU / self key
        )
        self.assertFalse(rr.ok)
        self.assertIsNone(rr.argv)

    # ── regression 6 — self-collector quarantine 경로 (격리+비차단+비권위) ────
    def test_6_self_collector_quarantine_non_blocking(self):
        envelope = {
            "task_id": "task-2717",
            "schedule_id": SELF_SCHEDULE,
            "owner_key": ANU_KEY,            # 텍스트 주장 (불신)
            "collector_role": "ANU",
            "recorded_at": _fresh_ts(),
        }
        res = M.verify_collector_authoritative(
            task_id="task-2717", envelope=envelope,
            probe=make_probe({ANU_SCHEDULE}),  # SELF_SCHEDULE 비-소유
            executor_key=DEV4_SELF_KEY, now=_now(),
            quarantine_dir=self._tmp,
        )
        # 격리됨
        self.assertEqual(res.verdict, M.VERDICT_QUARANTINED)
        # 비권위 — 자동 push/merge 판단 사용 0
        self.assertFalse(res.usable_for_auto_merge)
        # 비차단 — 작업물 보존(파기 0)
        self.assertTrue(res.work_preserved)
        # 격리 증거는 보존됨
        self.assertTrue(Path(res.quarantine_record_path).is_file())

    # ── regression 7 (C 강화) — actual owner proof = self-key ⇒ QUARANTINE ────
    def test_7_envelope_claims_anu_but_actual_proof_self_key_quarantine(self):
        """★ task-2717+1 (C): envelope 이 collector_role=ANU / self_key_used=
        false / ANU owner 를 '주장' 하지만 actual owner proof 가 self-key(=
        access denied)인 케이스. 반드시 SELF_COLLECTOR_QUARANTINE ·
        usable_for_auto_merge=False · work_preserved=True. envelope claim 은
        기록만 되고 판정에는 쓰이지 않음을 명시한다.
        """
        envelope = {
            "task_id": "task-2717",
            "schedule_id": SELF_SCHEDULE,        # dev4 self-key 소유 schedule
            "collector_role": "ANU",             # 주장 (거짓)
            "self_key_used": False,              # 주장 (거짓)
            "owner_key": ANU_KEY,                # 주장 (거짓, 텍스트만)
            "recorded_at": _fresh_ts(),
            "note": f"ANU key {ANU_KEY_PLACEHOLDER} 사용",
        }
        # SELF_SCHEDULE 은 ANU key 로 조회 거부(access denied) = 실제 self 소유.
        res = M.verify_collector_authoritative(
            task_id="task-2717", envelope=envelope,
            probe=make_probe({ANU_SCHEDULE}), executor_key=DEV4_SELF_KEY,
            now=_now(), quarantine_dir=self._tmp,
        )
        # 3종 필수 출력 동시 고정.
        self.assertEqual(res.classification, M.CLS_SELF_COLLECTOR_QUARANTINE)
        self.assertFalse(res.usable_for_auto_merge)
        self.assertTrue(res.work_preserved)
        # claim 은 그대로 기록(불변)되어 있다 — 그러나 판정은 proof 가 결정.
        self.assertEqual(res.envelope_claims["claimed_collector_role"], "ANU")
        self.assertEqual(res.envelope_claims["claimed_self_key_used"], False)
        self.assertEqual(res.owner_resolution["outcome"], M.OWNER_NOT_ANU)

    # ── regression 8 (B) — NO_RESULT_JSON: executor 미작성 → 별도 상태 ────────
    def test_8_no_result_json_distinct_state(self):
        """runner pickup 대상 result JSON 이 없으면 NO_RESULT_JSON 분류(별도 상태).
        PENDING_OWNER_PROOF 와 명시적으로 다른 상태임을 고정한다."""
        rr = M.anu_runner_pickup_and_fire(
            result_json_path="/nonexistent/path/task-2717.result.json",
            executor_key=DEV4_SELF_KEY,
            anu_key=ANU_KEY,
        )
        self.assertFalse(rr.ok)
        self.assertIsNone(rr.argv)
        self.assertEqual(rr.classification, M.CLS_NO_RESULT_JSON)
        self.assertFalse(rr.pickup_ok)
        # NO_RESULT_JSON ≠ PENDING_OWNER_PROOF (서로 다른 상태).
        self.assertNotEqual(M.CLS_NO_RESULT_JSON, M.CLS_PENDING_OWNER_PROOF)

    # ── regression 9 (B) — PENDING_OWNER_PROOF: owner proof 미확정(재시도 가능) ─
    def test_9_pending_owner_proof_distinct_from_fail_closed(self):
        """owner proof 미확정(조회 보류/재시도 가능)은 PENDING_OWNER_PROOF —
        fail-closed(NON_AUTHORITATIVE)와 구분되는 별도 상태."""
        # (a) probe 예외 → resolve outcome=PENDING (query_ok=False, 재시도 가능).
        def raising_probe(_sid):
            raise RuntimeError("transient cron-history outage")

        res_resolve = M.resolve_authoritative_owner(
            ANU_SCHEDULE, probe=raising_probe
        )
        self.assertEqual(res_resolve.outcome, M.OWNER_PENDING)
        self.assertFalse(res_resolve.owner_is_anu)
        self.assertFalse(res_resolve.query_ok)

        # (b) verify 레벨 → verdict=PENDING_OWNER_PROOF, 자동 사용 0, 작업물 보존.
        envelope = {
            "task_id": "task-2717", "schedule_id": ANU_SCHEDULE,
            "owner_key": ANU_KEY, "recorded_at": _fresh_ts(),
        }
        res = M.verify_collector_authoritative(
            task_id="task-2717", envelope=envelope, probe=raising_probe,
            now=_now(), quarantine_dir=self._tmp,
        )
        self.assertEqual(res.verdict, M.VERDICT_PENDING_OWNER_PROOF)
        self.assertEqual(res.classification, M.CLS_PENDING_OWNER_PROOF)
        self.assertFalse(res.usable_for_auto_merge)
        self.assertTrue(res.work_preserved)
        self.assertIsNone(res.quarantine_record_path)  # 격리 아님(미확정일 뿐)

        # (c) 대비: 명시적 비-access-denied error → fail-closed NON_AUTHORITATIVE
        #     (PENDING 과 다른 상태임을 동시 고정).
        def hard_error_probe(_sid):
            return {"status": "error", "message": "internal server error 500"}

        res2 = M.verify_collector_authoritative(
            task_id="task-2717", envelope=envelope, probe=hard_error_probe,
            now=_now(), quarantine_dir=self._tmp,
        )
        self.assertEqual(res2.verdict, M.VERDICT_NON_AUTHORITATIVE)
        self.assertEqual(res2.classification, M.CLS_OWNER_QUERY_FAILED)
        self.assertNotEqual(res.verdict, res2.verdict)  # PENDING ≠ fail-closed

    # ── regression 10 — fire-path gating: 실 fire/closeout 차단(Phase 1) ──────
    def test_10_fire_path_gated_no_self_fire(self):
        """Phase 1: ANU-owned argv 는 build/dry-run 까지만. 실 fire/closeout 은
        NOT_ACTIVATED / PHASE2_REQUIRED 로 차단(자가발사 0)."""
        import tempfile
        rdir = tempfile.mkdtemp(prefix="task2717-fire-")
        self.addCleanup(shutil.rmtree, rdir, ignore_errors=True)
        ew = M.executor_write_result_json(
            task_id="task-2717", payload={"summary": "x"},
            result_dir=rdir, now=_now(),
        )
        rr = M.anu_runner_pickup_and_fire(
            result_json_path=ew.result_json_path,
            executor_key=DEV4_SELF_KEY, anu_key=ANU_KEY, at="30s",
        )
        self.assertTrue(rr.ok)
        # runner 자체가 dry-run/비활성 — 자가발사 안 함.
        self.assertTrue(rr.dry_run)
        self.assertEqual(rr.fire_path, M.FIRE_NOT_ACTIVATED)
        # 기본 gate → NOT_ACTIVATED, 절대 fire 안 됨.
        gate = M.fire_callback_request(rr)
        self.assertEqual(gate["status"], M.FIRE_NOT_ACTIVATED)
        self.assertFalse(gate["fired"])
        self.assertTrue(gate["dry_run"])
        self.assertIsNotNone(gate["argv"])  # argv 는 surface(데이터)
        # activate=True 라도 Phase 1 은 실 fire 0 → PHASE2_REQUIRED.
        gate2 = M.fire_callback_request(rr, activate=True)
        self.assertEqual(gate2["status"], M.FIRE_PHASE2_REQUIRED)
        self.assertFalse(gate2["fired"])

    # ── regression 11 — result JSON schema + 멱등 lock(atomic write) ──────────
    def test_11_result_json_schema_and_idempotent_write(self):
        """result JSON schema 필드 고정 + 같은 경로 멱등(atomic os.replace) write."""
        import tempfile
        rdir = tempfile.mkdtemp(prefix="task2717-schema-")
        self.addCleanup(shutil.rmtree, rdir, ignore_errors=True)
        ew1 = M.executor_write_result_json(
            task_id="task-2717",
            payload={"report_path": "memory/reports/task-2717.md",
                     "sha256": "deadbeef"},
            result_dir=rdir, now=_now(),
        )
        rec = json.loads(Path(ew1.result_json_path).read_text("utf-8"))
        # schema 필수 필드.
        for key in ("task_id", "completion_signal", "written_at",
                    "schedule_created_by_executor", "callback_fired_by_executor"):
            self.assertIn(key, rec)
        self.assertEqual(rec["completion_signal"], "RESULT_JSON_WRITTEN")
        self.assertFalse(rec["schedule_created_by_executor"])
        self.assertFalse(rec["callback_fired_by_executor"])
        # 멱등 lock: 재작성 시 같은 경로(중복 schedule/파일 생성 0).
        ew2 = M.executor_write_result_json(
            task_id="task-2717", payload={"summary": "again"},
            result_dir=rdir, now=_now(),
        )
        self.assertEqual(ew1.result_json_path, ew2.result_json_path)
        self.assertEqual(
            len(list(Path(rdir).glob("task-2717.result.json"))), 1
        )

    # ── regression 12 — 인프라 probe 실패 → OWNER_PENDING (false quarantine 0) ──
    def test_12_infra_probe_failure_pending_not_quarantine(self):
        """인프라 장애(probe 실행 실패)는 message 에 'not found' 가 섞여도 실
        cokacdir access-denied 가 아니다. distinct error_kind 신호 →
        OWNER_PENDING(재시도 가능) 매핑으로 OWNER_NOT_ANU(false quarantine) 오분류
        차단."""
        def infra_failed_probe(_sid):
            # __call__ 의 인프라 실패 반환 형태 재현: message 에 'not found' 가
            # 섞여 있어도 error_kind 가 우선 → PENDING(NOT access-denied).
            return {
                "status": "error",
                "error_kind": M.PROBE_EXEC_FAILED_KIND,
                "message": "probe exec failed: cokacdir: command not found",
            }
        res = M.resolve_authoritative_owner(
            SELF_SCHEDULE, probe=infra_failed_probe
        )
        self.assertEqual(res.outcome, M.OWNER_PENDING)
        self.assertFalse(res.owner_is_anu)
        self.assertFalse(res.query_ok)
        # 대비: error_kind 없는 실 access-denied 응답은 OWNER_NOT_ANU(결정적).
        res2 = M.resolve_authoritative_owner(
            SELF_SCHEDULE, probe=make_probe(set())
        )
        self.assertEqual(res2.outcome, M.OWNER_NOT_ANU)
        self.assertNotEqual(res.outcome, res2.outcome)

        # verify 레벨: 인프라 장애 → PENDING_OWNER_PROOF, 격리 artifact 미생성.
        envelope = {
            "task_id": "task-2717", "schedule_id": SELF_SCHEDULE,
            "owner_key": ANU_KEY, "recorded_at": _fresh_ts(),
        }
        vres = M.verify_collector_authoritative(
            task_id="task-2717", envelope=envelope, probe=infra_failed_probe,
            executor_key=DEV4_SELF_KEY, now=_now(), quarantine_dir=self._tmp,
        )
        self.assertEqual(vres.verdict, M.VERDICT_PENDING_OWNER_PROOF)
        self.assertFalse(vres.usable_for_auto_merge)
        self.assertTrue(vres.work_preserved)
        self.assertIsNone(vres.quarantine_record_path)  # 격리 아님(인프라 장애)

    def test_12b_real_probe_infra_failure_distinct_signal(self):
        """RealCokacdirCronHistoryProbe 가 바이너리 부재(인프라 실패)를
        message-비의존 distinct 신호(error_kind=probe_exec_failed)로 반환 →
        resolve 가 OWNER_PENDING 으로 매핑(실 cokacdir 미호출)."""
        probe = M.RealCokacdirCronHistoryProbe(anu_key="x", chat_id="0")
        probe.binary = "/nonexistent/cokacdir-binary-task2717"
        out = probe(ANU_SCHEDULE)
        self.assertEqual(out.get("status"), "error")
        self.assertEqual(out.get("error_kind"), M.PROBE_EXEC_FAILED_KIND)
        res = M.resolve_authoritative_owner(ANU_SCHEDULE, probe=probe)
        self.assertEqual(res.outcome, M.OWNER_PENDING)
        self.assertFalse(res.owner_is_anu)

    # ── regression 13 — 숫자형 epoch staleness + silent bypass 0 ──────────────
    def test_13_numeric_epoch_staleness_no_silent_bypass(self):
        """숫자형 epoch timestamp 처리(strptime TypeError 미발생) + 파싱 실패가
        staleness 를 silent bypass 하지 않음."""
        now = _now()
        # (a) fresh numeric epoch(int, 5분 전) → stale 아님 → owner resolution 진행.
        fresh_epoch = int((now - timedelta(minutes=5)).timestamp())
        env_fresh = {
            "task_id": "task-2717", "schedule_id": ANU_SCHEDULE,
            "owner_key": ANU_KEY, "recorded_at": fresh_epoch,
        }
        res = M.verify_collector_authoritative(
            task_id="task-2717", envelope=env_fresh,
            probe=make_probe({ANU_SCHEDULE}), now=now, quarantine_dir=self._tmp,
        )
        self.assertEqual(res.verdict, M.VERDICT_AUTHORITATIVE)  # stale reject 0

        # (b) old numeric epoch(float, 48h 전) → STALE_ENVELOPE_REJECTED.
        old_epoch = (now - timedelta(hours=48)).timestamp()  # float
        env_old = {
            "task_id": "task-2717", "schedule_id": ANU_SCHEDULE,
            "owner_key": ANU_KEY, "recorded_at": old_epoch,
        }
        res_old = M.verify_collector_authoritative(
            task_id="task-2717", envelope=env_old,
            probe=make_probe({ANU_SCHEDULE}), now=now, quarantine_dir=self._tmp,
        )
        self.assertEqual(res_old.verdict, M.VERDICT_REJECTED)
        self.assertEqual(res_old.classification, M.CLS_STALE_ENVELOPE_REJECTED)

        # (c) timestamp 값은 있으나 파싱 불가 → silent bypass 0 (stale reject).
        env_bad = {
            "task_id": "task-2717", "schedule_id": ANU_SCHEDULE,
            "owner_key": ANU_KEY, "recorded_at": "not-a-timestamp",
        }
        res_bad = M.verify_collector_authoritative(
            task_id="task-2717", envelope=env_bad,
            probe=make_probe({ANU_SCHEDULE}), now=now, quarantine_dir=self._tmp,
        )
        self.assertEqual(res_bad.verdict, M.VERDICT_REJECTED)
        self.assertEqual(res_bad.classification, M.CLS_STALE_ENVELOPE_REJECTED)

        # (d) 함수 단위: numeric epoch age 계산 정확 + TypeError 미발생.
        age = M._envelope_age_seconds({"recorded_at": fresh_epoch}, now=now)
        self.assertIsNotNone(age)
        self.assertAlmostEqual(age, 300, delta=2)
        # 값 없음 → None(미적용), 값 있고 파싱 불가 → sentinel(>임계).
        self.assertIsNone(M._envelope_age_seconds({}, now=now))
        self.assertGreater(
            M._envelope_age_seconds({"recorded_at": "garbage"}, now=now),
            M.DEFAULT_MAX_ENVELOPE_AGE_SECONDS,
        )

    # ── regression 14 — 파일명 sanitize: '/' 와 '\\' 양쪽 치환(cross-platform) ─
    def test_14_filename_sanitize_both_separators(self):
        """write_quarantine_record / executor_write_result_json 의 파일명
        sanitize 가 '/' 와 '\\' 양쪽을 '_' 로 치환(Linux 에서 '\\' 미치환 차단)."""
        # (a) write_quarantine_record — task_id/schedule_id 에 양쪽 separator.
        verification = M.CollectorVerification(
            schema=M.SCHEMA, verdict=M.VERDICT_QUARANTINED,
            classification=M.CLS_SELF_COLLECTOR_QUARANTINE,
            task_id="task\\2717/evil", schedule_id="A\\B/C",
            owner_resolution=None, envelope_claims={}, work_preserved=True,
        )
        path = M.write_quarantine_record(
            verification, quarantine_dir=self._tmp, now=_now(),
        )
        name = Path(path).name
        self.assertNotIn("\\", name)
        self.assertNotIn("/", name)
        self.assertTrue(Path(path).is_file())

        # (b) executor_write_result_json — task_id 에 '\\' 포함 → 파일명 미존재.
        rdir = tempfile.mkdtemp(prefix="task2717-san-")
        self.addCleanup(shutil.rmtree, rdir, ignore_errors=True)
        ew = M.executor_write_result_json(
            task_id="task\\2717/x", payload={"summary": "x"},
            result_dir=rdir, now=_now(),
        )
        fname = Path(ew.result_json_path).name
        self.assertNotIn("\\", fname)
        self.assertNotIn("/", fname)
        self.assertTrue(Path(ew.result_json_path).is_file())

    # ── regression 15 — setUp tmpdir cleanup 등록(누수 0) ─────────────────────
    def test_15_setup_tmpdir_cleanup_registered(self):
        """setUp 의 mkdtemp tmpdir 가 addCleanup(shutil.rmtree) 로 등록되어
        테스트 종료 시 자동 정리됨(tmpdir 누수 0)."""
        self.assertTrue(Path(self._tmp).is_dir())
        cleanups = getattr(self, "_cleanups", [])
        registered = any(
            func is shutil.rmtree and args and args[0] == self._tmp
            for func, args, _kwargs in cleanups
        )
        self.assertTrue(
            registered, "setUp tmpdir 가 addCleanup 으로 정리 등록되어야 함"
        )

    # ── regression 16 — Gemini MEDIUM 보정: 방어 가드(None/non-callable/non-dict) ─
    def test_16_defensive_guards_none_noncallable_nondict(self):
        """task-2717 Phase1 Gemini MEDIUM 보정 4건의 방어 동작 고정:
        (a) _sanitize_filename_component(None/빈/공백) → default (silent 'None' 0)
        (b) resolve_authoritative_owner(probe=비-callable) → OWNER_QUERY_FAILED
            (설정오류는 PENDING 재시도 아니라 fail-closed)
        (c) _parse_timestamp_value(None) → None (불필요 파싱 루프 0)
        (d) verify_collector_authoritative(envelope=non-dict) → REJECTED
            (AttributeError crash 0, 작업물 보존)."""
        # (a) filename component None/빈/공백 → default.
        self.assertEqual(M._sanitize_filename_component(None, default="d"), "d")
        self.assertEqual(M._sanitize_filename_component("   ", default="d"), "d")
        self.assertEqual(
            M._sanitize_filename_component("a/b\\c", default="d"), "a_b_c"
        )
        # (b) probe 비-callable(None) → 설정오류 fail-closed(QUERY_FAILED).
        r = M.resolve_authoritative_owner("ANUOWNED1", probe=None)
        self.assertEqual(r.outcome, M.OWNER_QUERY_FAILED)
        self.assertFalse(r.owner_is_anu)
        self.assertFalse(r.query_ok)
        # (c) None timestamp → None.
        self.assertIsNone(M._parse_timestamp_value(None))
        # (d) envelope 가 dict 아님 → crash 없이 REJECTED + 작업물 보존.
        res = M.verify_collector_authoritative(
            task_id="task-2717",
            envelope=None,
            probe=make_probe({ANU_SCHEDULE}),
            executor_key=DEV4_SELF_KEY,
            now=_now(),
            quarantine_dir=self._tmp,
        )
        self.assertEqual(res.verdict, M.VERDICT_REJECTED)
        self.assertEqual(res.classification, M.CLS_ENVELOPE_BINDING_MISMATCH)
        self.assertTrue(res.work_preserved)

    # ── regression 17 — result JSON task_id 부재 → runner fail-closed ──────────
    def test_17_runner_fail_closed_on_missing_task_id(self):
        """result JSON 에 유효한 task_id 가 없으면 runner 는 빈 task_id 로 잘못된
        callback argv 를 만들지 않고 fail-closed(FAIL, argv=None)."""
        rdir = tempfile.mkdtemp(prefix="task2717-noid-")
        self.addCleanup(shutil.rmtree, rdir, ignore_errors=True)
        rpath = Path(rdir) / "result.json"
        rpath.write_text(json.dumps({"summary": "x"}), encoding="utf-8")  # task_id 없음
        rr = M.anu_runner_pickup_and_fire(
            result_json_path=str(rpath),
            executor_key=DEV4_SELF_KEY,
            anu_key=ANU_KEY,
            chat_id="6937032012",
            at="30s",
        )
        self.assertFalse(rr.ok)
        self.assertIsNone(rr.argv)
        self.assertEqual(rr.verdict, "FAIL")


    # ── regression 18 (High #466) — timezone offset ISO 파싱 ────────────────────
    def test_18_timezone_offset_iso_parsing(self):
        """_parse_timestamp_value 가 timezone offset ISO("+09:00") 를 정상 파싱.
        stale 오인 0, age 계산 일관성 확인."""
        now = datetime(2026, 6, 1, 4, 0, 0, tzinfo=timezone.utc)

        # (a) "+09:00" offset ISO (KST = UTC+9) → tz-aware, 정상 age 계산.
        # "2026-06-01T13:00:00+09:00" = UTC 04:00:00 → age ≈ 0초(now 와 동일).
        kst_ts = "2026-06-01T13:00:00+09:00"
        dt = M._parse_timestamp_value(kst_ts)
        self.assertIsNotNone(dt, "timezone offset ISO 파싱 실패")
        self.assertIsNotNone(dt.tzinfo, "tz-aware 이어야 함")
        # UTC 로 정규화 후 now 와 비교
        dt_utc = dt.astimezone(timezone.utc)
        self.assertEqual(dt_utc.year, 2026)
        self.assertEqual(dt_utc.month, 6)
        self.assertEqual(dt_utc.day, 1)
        self.assertEqual(dt_utc.hour, 4)
        self.assertEqual(dt_utc.minute, 0)
        age = M._envelope_age_seconds({"recorded_at": kst_ts}, now=now)
        self.assertIsNotNone(age)
        # age ≈ 0초 (±5초 허용), stale 오인 0 (기본 임계 6h = 21600s 보다 훨씬 작음)
        self.assertAlmostEqual(age, 0, delta=5,
                               msg="timezone offset ISO age 계산이 stale 오인(age 과대)")
        self.assertLess(age, M.DEFAULT_MAX_ENVELOPE_AGE_SECONDS,
                        "fresh +09:00 timestamp 가 stale 로 잘못 판정됨")

        # (b) "Z" suffix (UTC) — 기존 경로 회귀 0.
        z_ts = "2026-06-01T03:55:00Z"  # 5분 전
        dt_z = M._parse_timestamp_value(z_ts)
        self.assertIsNotNone(dt_z, "Z suffix ISO 파싱 회귀")
        self.assertIsNotNone(dt_z.tzinfo, "Z suffix: tz-aware 이어야 함")
        age_z = M._envelope_age_seconds({"recorded_at": z_ts}, now=now)
        self.assertIsNotNone(age_z)
        self.assertAlmostEqual(age_z, 300, delta=2,
                               msg="Z suffix age 계산 회귀")

        # (c) 실 verify 경로: "+09:00" fresh envelope → reject 아님(AUTHORITATIVE).
        envelope = {
            "task_id": "task-2717",
            "schedule_id": ANU_SCHEDULE,
            "owner_key": ANU_KEY,
            "recorded_at": kst_ts,
        }
        res = M.verify_collector_authoritative(
            task_id="task-2717", envelope=envelope,
            probe=make_probe({ANU_SCHEDULE}), now=now,
            quarantine_dir=self._tmp,
        )
        self.assertEqual(res.verdict, M.VERDICT_AUTHORITATIVE,
                         f"timezone offset ISO fresh envelope 이 stale reject 됨: "
                         f"verdict={res.verdict} reasons={res.reasons}")

        # (d) naive datetime 은 UTC 로 간주 — tz-aware 비교 오류(TypeError) 없음.
        dt_naive_input = "2026-06-01T04:00:00"  # timezone 없는 naive ISO
        dt_naive = M._parse_timestamp_value(dt_naive_input)
        self.assertIsNotNone(dt_naive)
        self.assertIsNotNone(dt_naive.tzinfo,
                             "naive ISO → UTC tz-aware 정규화 실패")
        # age 비교가 TypeError 없이 정상 수행됨
        age_naive = M._envelope_age_seconds({"recorded_at": dt_naive_input}, now=now)
        self.assertIsNotNone(age_naive)
        self.assertAlmostEqual(age_naive, 0, delta=2)

        # (e) "-05:00" offset (UTC-5) → UTC 정규화 정확.
        # "2026-05-31T23:00:00-05:00" = UTC 2026-06-01T04:00:00 → age ≈ 0초.
        minus5_ts = "2026-05-31T23:00:00-05:00"
        dt_m5 = M._parse_timestamp_value(minus5_ts)
        self.assertIsNotNone(dt_m5)
        dt_m5_utc = dt_m5.astimezone(timezone.utc)
        self.assertEqual(dt_m5_utc.hour, 4)
        age_m5 = M._envelope_age_seconds({"recorded_at": minus5_ts}, now=now)
        self.assertAlmostEqual(age_m5, 0, delta=5)


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