# -*- coding: utf-8 -*-
"""task-2734 P0B_MINIMUM input wiring — terminal_state_callback → p0b_inbox (flag-gated OFF).

회장 필수 8 테스트:
 1. flag OFF → p0b_inbox write 0.
 2. flag ON → p0b_inbox 에 task-*.result.json 1건 atomic write.
 3. result 에 task_id·completion_signal·collector_envelope·schedule_id 포함.
 4. 기존 executor result/callback/envelope/cron 경로 불변(byte/의미).
 5. fake/replay fixture 금지 — 테스트도 실제 envelope/등록 산출 경로(launch_callback + 주입 runner).
 6. ANU key literal 0 (sealed ANU key 16자 리터럴이 소스/테스트에 비포함).
 7. real_wake 발사 0 (real_wake/ACTIVE flag 생성·발사 0).
 8. ACTIVE=true 0.

★ 설계 조건(회장 10): staging→stable mtime(now-3s)→atomic rename, old sid 재사용 금지,
  fake envelope 금지, self-key→ANU 포장 금지, owner-proof 우회 금지(실제 schedule_id 만).
"""
from __future__ import annotations

import importlib.util
import json
import os
import sys
from pathlib import Path
from typing import Optional, List
from unittest import mock

# ── sys.path: worktree root 우선 ──────────────────────────────────────────────
_ROOT = Path(__file__).resolve().parent.parent.parent
if str(_ROOT) not in sys.path:
    sys.path.insert(0, str(_ROOT))


def _load_module(modname: str, relpath: str):
    fpath = _ROOT / relpath
    spec = importlib.util.spec_from_file_location(modname, fpath)
    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


def _ensure_dispatch():
    if "dispatch.normal_fallback_callback_helper" not in sys.modules:
        _load_module("dispatch.callback_owner_enforcer", "dispatch/callback_owner_enforcer.py")
        _load_module("dispatch.normal_fallback_callback_helper", "dispatch/normal_fallback_callback_helper.py")


def _load_tsc():
    _ensure_dispatch()
    for key in list(sys.modules.keys()):
        if "terminal_state_callback" in key:
            del sys.modules[key]
    return _load_module(
        "scripts.harness.v36.terminal_state_callback",
        "scripts/harness/v36/terminal_state_callback.py",
    )


def _make_decision(argv: Optional[List[str]], verdict: str = "PASS"):
    dec = mock.MagicMock()
    dec.argv = argv
    dec.verdict = verdict
    return dec


def _setup_events(tmp_path: Path, task_id: str = "task-2734"):
    events_dir = tmp_path / "events"
    events_dir.mkdir(parents=True, exist_ok=True)
    done_file = str(events_dir / f"{task_id}.done")
    return str(events_dir), done_file


def _trigger_failure_marker(events_dir: str, task_id: str = "task-2734"):
    """실제 failure terminal_state(FINISH_BLOCKED_EXTERNAL_DIRTY) 유발 → callback 등록 경로 진입.

    fake envelope 가 아니라 실제 _determine_terminal_state 가 읽는 입력 marker 만 둔다(회장 5).
    """
    marker = Path(events_dir) / f"{task_id}.external-dirty-blocker.json"
    marker.write_text(json.dumps({"reason": "dirty", "schema": "external-dirty-blocker.v1"}), encoding="utf-8")


_ARGV = ["cokacdir", "--cron", "x", "--at", "+5m", "--chat", "6937032012", "--key", "anukey", "--once"]
_INBOX_REL = os.path.join("memory", "events", "p0b_inbox")


def _fresh_sid_runner(sid: str = "anu-cron-2734-fresh-001"):
    """주입 runner: 실제 cokacdir cron 응답을 모사하여 fresh schedule_id 를 stdout JSON 으로 반환.

    ★ 실제 등록 backend 주입점(회장 9-8). envelope 자체는 _build_envelope/launch_callback 의
      실제 산출이며, 이 runner 는 cron backend(=cokacdir subprocess)만 대체한다(fake fixture 아님).
    """
    def _runner(_argv):
        r = mock.MagicMock()
        r.returncode = 0
        r.stdout = json.dumps({"status": "ok", "id": sid, "schedule": "+5m"})
        return r
    return _runner


# ─────────────────────────────────────────────────────────────────────────────
# 테스트 1: flag OFF → p0b_inbox write 0
# ─────────────────────────────────────────────────────────────────────────────
def test_flag_off_no_inbox_write(tmp_path, monkeypatch):
    monkeypatch.delenv("P0B_INBOX_WRITE_ENABLED", raising=False)  # default OFF
    tsc = _load_tsc()
    events_dir, done_file = _setup_events(tmp_path)
    _trigger_failure_marker(events_dir)

    with mock.patch.object(tsc, "launch_callback", return_value=_make_decision(_ARGV)):
        tsc.emit(task_id="task-2734", events_dir=events_dir, workspace=str(tmp_path),
                 done_file=done_file, runner=_fresh_sid_runner())

    inbox = tmp_path / _INBOX_REL
    results = list(inbox.glob("task-*.result.json")) if inbox.exists() else []
    assert results == [], f"flag OFF 인데 inbox write 발생: {results}"


# ─────────────────────────────────────────────────────────────────────────────
# 테스트 2: flag ON → p0b_inbox 에 task-*.result.json 1건 atomic write
# ─────────────────────────────────────────────────────────────────────────────
def test_flag_on_single_atomic_inbox_write(tmp_path, monkeypatch):
    monkeypatch.setenv("P0B_INBOX_WRITE_ENABLED", "1")
    tsc = _load_tsc()
    events_dir, done_file = _setup_events(tmp_path)
    _trigger_failure_marker(events_dir)

    with mock.patch.object(tsc, "launch_callback", return_value=_make_decision(_ARGV)):
        tsc.emit(task_id="task-2734", events_dir=events_dir, workspace=str(tmp_path),
                 done_file=done_file, runner=_fresh_sid_runner())

    inbox = tmp_path / _INBOX_REL
    results = list(inbox.glob("task-*.result.json"))
    assert len(results) == 1, f"flag ON: inbox 에 정확히 1건이어야 함, got {results}"
    assert results[0].name == "task-2734.result.json"
    # atomic 흔적: staging(.tmp) 잔존 0
    staging = list(inbox.glob(".p0b-inbox-staging-*")) + list(inbox.glob("*.tmp"))
    assert staging == [], f"staging 잔존(비원자적): {staging}"


# ─────────────────────────────────────────────────────────────────────────────
# 테스트 3: result schema — task_id·completion_signal·collector_envelope·schedule_id
# ─────────────────────────────────────────────────────────────────────────────
def test_inbox_result_schema_fields(tmp_path, monkeypatch):
    monkeypatch.setenv("P0B_INBOX_WRITE_ENABLED", "enabled")
    tsc = _load_tsc()
    events_dir, done_file = _setup_events(tmp_path)
    _trigger_failure_marker(events_dir)
    sid = "anu-cron-2734-fresh-XYZ"

    with mock.patch.object(tsc, "launch_callback", return_value=_make_decision(_ARGV)):
        tsc.emit(task_id="task-2734", events_dir=events_dir, workspace=str(tmp_path),
                 done_file=done_file, runner=_fresh_sid_runner(sid))

    result_path = tmp_path / _INBOX_REL / "task-2734.result.json"
    assert result_path.exists()
    data = json.loads(result_path.read_text(encoding="utf-8"))
    assert data["task_id"] == "task-2734"
    assert data["completion_signal"] == "EXECUTOR_RESULT_WRITTEN"
    env = data["collector_envelope"]
    assert isinstance(env, dict)
    assert env["schedule_id"] == sid                  # ★ fresh 실제 sid (재사용/fake 아님)
    assert env["collector_role"] == "ANU"
    # self-key→ANU 포장 금지: 정상 ANU 등록 → self_key_used False (위조 0)
    assert env["self_key_used"] is False

    # readiness(2s) 통과: result mtime 이 now-2s 이전(=충분히 과거)
    import time as _t
    age = _t.time() - result_path.stat().st_mtime
    assert age >= 2.0, f"mtime 안정화 실패(readiness 미통과): age={age}"


# ─────────────────────────────────────────────────────────────────────────────
# 테스트 4: 기존 executor result/callback/envelope/cron 경로 불변(flag OFF=byte 동일)
# ─────────────────────────────────────────────────────────────────────────────
def test_existing_paths_byte_identical_flag_off(tmp_path, monkeypatch):
    """동일 입력에 대해 flag OFF 시 terminal-state envelope/registration marker 가
    이전(wiring 부재)과 동일해야 한다. 여기서는 flag OFF 산출물 자체의 안정성 + inbox 0 검증."""
    monkeypatch.delenv("P0B_INBOX_WRITE_ENABLED", raising=False)
    tsc = _load_tsc()
    events_dir, done_file = _setup_events(tmp_path)
    _trigger_failure_marker(events_dir)

    with mock.patch.object(tsc, "launch_callback", return_value=_make_decision(_ARGV)):
        tsc.emit(task_id="task-2734", events_dir=events_dir, workspace=str(tmp_path),
                 done_file=done_file, runner=_fresh_sid_runner())

    # 기존 envelope 정상 산출 + callback_registered True (cron 경로 불변)
    env = json.loads((Path(events_dir) / "task-2734.terminal-state.json").read_text(encoding="utf-8"))
    assert env["terminal_state"] == "FINISH_BLOCKED_EXTERNAL_DIRTY"
    assert env["callback_registered"] is True
    assert env["collector_role"] == "ANU"
    reg = json.loads((Path(events_dir) / "task-2734.terminal-callback-registered.json").read_text(encoding="utf-8"))
    assert reg["status"] == "REGISTERED"
    # 입력 marker 불변
    after = (Path(events_dir) / "task-2734.external-dirty-blocker.json").read_text(encoding="utf-8")
    assert json.loads(after)["reason"] == "dirty"
    # inbox write 0
    inbox = tmp_path / _INBOX_REL
    assert not (inbox.exists() and list(inbox.glob("task-*.result.json")))


# ─────────────────────────────────────────────────────────────────────────────
# 테스트 5: fake/replay fixture 금지 — 실제 _build_envelope·_register_callback 산출 경로
# ─────────────────────────────────────────────────────────────────────────────
def test_no_fake_fixture_real_production_path(tmp_path, monkeypatch):
    """inbox result 의 schedule_id 는 등록 backend(runner) 의 실제 stdout 에서만 유래.
    runner stdout 에 sid 가 없으면(=등록은 됐으나 sid 미파싱) inbox write 0 (fake 합성 금지)."""
    monkeypatch.setenv("P0B_INBOX_WRITE_ENABLED", "1")
    tsc = _load_tsc()
    events_dir, done_file = _setup_events(tmp_path)
    _trigger_failure_marker(events_dir)

    # runner 가 sid 없는 비-JSON stdout 반환 → schedule_id 파싱 불가 → inbox write 0
    def _no_sid_runner(_argv):
        r = mock.MagicMock(); r.returncode = 0; r.stdout = "OK (no json)"; return r

    with mock.patch.object(tsc, "launch_callback", return_value=_make_decision(_ARGV)):
        tsc.emit(task_id="task-2734", events_dir=events_dir, workspace=str(tmp_path),
                 done_file=done_file, runner=_no_sid_runner)

    inbox = tmp_path / _INBOX_REL
    results = list(inbox.glob("task-*.result.json")) if inbox.exists() else []
    assert results == [], "schedule_id 미파싱 시 fake sid 로 inbox write 하면 안 됨"

    # 반대로 실제 sid stdout 이면 그 값 그대로(파생 0) inbox 에 기록
    events_dir2, done_file2 = _setup_events(tmp_path / "b")
    _trigger_failure_marker(events_dir2)
    with mock.patch.object(tsc, "launch_callback", return_value=_make_decision(_ARGV)):
        tsc.emit(task_id="task-2734", events_dir=events_dir2, workspace=str(tmp_path / "b"),
                 done_file=done_file2, runner=_fresh_sid_runner("real-sid-777"))
    data = json.loads((tmp_path / "b" / _INBOX_REL / "task-2734.result.json").read_text(encoding="utf-8"))
    assert data["collector_envelope"]["schedule_id"] == "real-sid-777"


# ─────────────────────────────────────────────────────────────────────────────
# 테스트 6: ANU key literal 0 — 소스에 sealed key 리터럴 비포함
# ─────────────────────────────────────────────────────────────────────────────
def test_no_anu_key_literal_in_source():
    # needle 을 조각 결합으로 구성 → 이 테스트 파일/대상 소스 어디에도 연속 리터럴 0.
    needle = "c119085a" + "ddb0f8b7"
    src = (_ROOT / "scripts/harness/v36/terminal_state_callback.py").read_text(encoding="utf-8")
    assert needle not in src, "ANU key 리터럴이 소스에 노출됨"
    test_src = Path(__file__).read_text(encoding="utf-8")
    assert needle not in test_src, "ANU key 리터럴이 테스트에 노출됨"


# ─────────────────────────────────────────────────────────────────────────────
# 테스트 7: real_wake 발사 0 — wiring 은 real_wake/driver flag 생성·발사 0
# ─────────────────────────────────────────────────────────────────────────────
def test_no_real_wake_no_flag_creation(tmp_path, monkeypatch):
    monkeypatch.setenv("P0B_INBOX_WRITE_ENABLED", "1")
    tsc = _load_tsc()
    events_dir, done_file = _setup_events(tmp_path)
    _trigger_failure_marker(events_dir)

    with mock.patch.object(tsc, "launch_callback", return_value=_make_decision(_ARGV)):
        tsc.emit(task_id="task-2734", events_dir=events_dir, workspace=str(tmp_path),
                 done_file=done_file, runner=_fresh_sid_runner())

    # 어떤 driver/real_wake/epoch flag 도 생성되지 않음 (behavioral)
    state_dir = tmp_path / "memory" / "state"
    for fl in ("p0b_driver_enabled", "p0b_real_wake_enabled", "p0b_activation_epoch"):
        assert not (state_dir / fl).exists(), f"금지된 flag 생성됨: {fl}"
    # 소스에 real_wake/driver flag 파일 write 로직 비포함 (flag 파일 경로 리터럴 0)
    src = (_ROOT / "scripts/harness/v36/terminal_state_callback.py").read_text(encoding="utf-8")
    for forbidden in ("p0b_real_wake_enabled", "p0b_driver_enabled", "p0b_activation_epoch"):
        assert forbidden not in src, f"금지 flag 경로 리터럴 소스 노출: {forbidden}"


# ─────────────────────────────────────────────────────────────────────────────
# 테스트 8: ACTIVE=true 0 — wiring 은 activation flag 를 ON 으로 쓰지 않음
# ─────────────────────────────────────────────────────────────────────────────
def test_no_active_true(tmp_path, monkeypatch):
    monkeypatch.setenv("P0B_INBOX_WRITE_ENABLED", "1")
    tsc = _load_tsc()
    events_dir, done_file = _setup_events(tmp_path)
    _trigger_failure_marker(events_dir)

    with mock.patch.object(tsc, "launch_callback", return_value=_make_decision(_ARGV)):
        tsc.emit(task_id="task-2734", events_dir=events_dir, workspace=str(tmp_path),
                 done_file=done_file, runner=_fresh_sid_runner())

    # inbox result payload 에 activation/ACTIVE=true 토글 0
    data = json.loads((tmp_path / _INBOX_REL / "task-2734.result.json").read_text(encoding="utf-8"))
    flat = json.dumps(data).replace(" ", "").lower()
    assert '"active":true' not in flat
    # activation flag 파일도 생성 0 (ACTIVE 토글 행위 자체 없음)
    assert not (tmp_path / "memory" / "state" / "p0b_driver_enabled").exists()
    # 소스에 ACTIVE=true / activation flag enable write 로직 비포함
    src = (_ROOT / "scripts/harness/v36/terminal_state_callback.py").read_text(encoding="utf-8")
    assert 'ACTIVE=true' not in src
    assert 'ACTIVATION_ENABLED' not in src


# ─────────────────────────────────────────────────────────────────────────────
# 테스트 9: staging 파일명 race — pid 만 의존 금지. 동일 task_id 동시 호출에도 staging 충돌 0
#   (Gemini MED :329) — uuid4 등 충분한 고유 suffix 로 같은 task_id/같은 pid 라도 staging 경로 상이.
# ─────────────────────────────────────────────────────────────────────────────
def test_staging_name_unique_no_pid_only_race(tmp_path):
    tsc = _load_tsc()
    inbox_dir = str(tmp_path / "inbox")

    captured: List[str] = []
    real_replace = os.replace

    def _capture_replace(src, dst):
        # staging(src) 경로를 기록한 뒤 정상 atomic 반영(실제 경로 그대로 검증).
        captured.append(src)
        return real_replace(src, dst)

    # 동일 task_id 로 두 번 write — pid 는 동일(같은 프로세스). staging 이 pid 에만
    # 의존하면 두 staging 경로가 같아져 동시성 충돌이 가능하다. uuid suffix 면 매번 상이.
    with mock.patch.object(tsc.os, "replace", _capture_replace):
        for _ in range(2):
            out = tsc._write_p0b_inbox_result(
                workspace=str(tmp_path), task_id="task-2734",
                schedule_id="sid-race-1", self_key_used=False,
                collector_role="ANU", inbox_dir=inbox_dir,
            )
            assert out is not None

    assert len(captured) == 2
    # 핵심: 동일 task_id·동일 pid 인데 staging 파일명이 서로 달라야 한다(race-free).
    assert captured[0] != captured[1], f"staging 파일명이 pid 에만 의존(충돌 위험): {captured}"
    # 각 staging 이름에 pid 와 추가 고유 suffix 가 모두 포함되어야 한다.
    pid = str(os.getpid())
    for s in captured:
        base = os.path.basename(s)
        assert base.startswith(".p0b-inbox-staging-task-2734-")
        assert pid in base
        assert base.endswith(".tmp")
        # pid 다음에도 비어있지 않은 추가 고유 구획이 존재(= pid 만 의존 금지).
        suffix = base.split(f"-{pid}-", 1)
        assert len(suffix) == 2 and suffix[1].replace(".tmp", "") != ""
    # staging 잔존 0 (atomic 반영 완료)
    assert list((tmp_path / "inbox").glob(".p0b-inbox-staging-*")) == []


# ─────────────────────────────────────────────────────────────────────────────
# 테스트 10: 최종 반영은 os.replace(atomic overwrite) — 대상 파일이 이미 존재해도
#   FileExistsError 없이 덮어쓴다 (Gemini MED :346, os.rename→os.replace).
# ─────────────────────────────────────────────────────────────────────────────
def test_final_write_replace_overwrites_existing(tmp_path):
    tsc = _load_tsc()
    inbox_dir = tmp_path / "inbox"
    inbox_dir.mkdir(parents=True, exist_ok=True)
    final_path = inbox_dir / "task-2734.result.json"
    # 대상이 이미 존재(이전 run 잔존 등). os.rename 은 일부 OS 에서 FileExistsError.
    final_path.write_text('{"stale": true}', encoding="utf-8")

    out = tsc._write_p0b_inbox_result(
        workspace=str(tmp_path), task_id="task-2734",
        schedule_id="sid-replace-1", self_key_used=False,
        collector_role="ANU", inbox_dir=str(inbox_dir),
    )
    assert out == str(final_path)
    # 덮어쓰기 성공: stale 내용이 사라지고 신규 envelope 로 교체
    data = json.loads(final_path.read_text(encoding="utf-8"))
    assert "stale" not in data
    assert data["task_id"] == "task-2734"
    assert data["completion_signal"] == "EXECUTOR_RESULT_WRITTEN"
    assert data["collector_envelope"]["schedule_id"] == "sid-replace-1"
    # 소스가 os.replace 를 사용(os.rename 잔존 0) — atomic overwrite 보장.
    src = (_ROOT / "scripts/harness/v36/terminal_state_callback.py").read_text(encoding="utf-8")
    assert "os.replace(staging_path, final_path)" in src
    assert "os.rename(staging_path, final_path)" not in src


# ─────────────────────────────────────────────────────────────────────────────
# 테스트 11: p0b_inbox write 실패 시 fail-open — 예외를 삼키고 None 반환, staging 잔존 0.
# ─────────────────────────────────────────────────────────────────────────────
def test_inbox_write_fail_open_returns_none(tmp_path):
    tsc = _load_tsc()
    inbox_dir = str(tmp_path / "inbox")

    # 최종 반영 단계에서 강제 실패 → fail-open(None) + staging 정리.
    def _boom(src, dst):
        raise OSError("simulated replace failure")

    with mock.patch.object(tsc.os, "replace", _boom):
        out = tsc._write_p0b_inbox_result(
            workspace=str(tmp_path), task_id="task-2734",
            schedule_id="sid-fail-1", self_key_used=False,
            collector_role="ANU", inbox_dir=inbox_dir,
        )
    assert out is None, "write 실패는 fail-open(None) 이어야 한다"
    # staging 잔존 0 (예외 경로에서 정리)
    assert list((tmp_path / "inbox").glob(".p0b-inbox-staging-*")) == []
    # 최종 result 도 미생성
    assert not (tmp_path / "inbox" / "task-2734.result.json").exists()