"""기획서 project-level 분석 service — race / source_hash / atomic / consumer 가드.

리뷰어 NEEDS_REVISION (3 round) 의 핵심 종합 회귀:
  - L1 1차 status guard
  - L2 manifest payload 에 job_id + source_hash
  - L3 staging file 패턴 (manifest.{job_id}.json → status 재확인 → promote)
  - L4 reader source_hash 검증 + cleanup
  - get_planning_context 가 검증된 reader 위임 (raw json 우회 금지)
  - load_status corrupt JSON surface as error
  - project_has_planning_doc empty/stale manifest false-positive 차단
"""

from __future__ import annotations

import json
import threading
import time
from pathlib import Path
from unittest.mock import MagicMock, patch

import pytest

from app.core.config import settings
from app.services import planning_doc_analysis_service as svc


# ── helpers ──────────────────────────────────────────────────────────


@pytest.fixture
def project_dir(tmp_path, monkeypatch):
    """settings.projects_dir 를 tmp 로 격리. PID 와 base path 반환."""
    monkeypatch.setattr(settings, "projects_dir", str(tmp_path))
    pid = "00000000-0000-0000-0000-000000000abc"
    (tmp_path / pid / "assets").mkdir(parents=True)
    yield pid, tmp_path / pid


def _mock_db_with_text(text: str = ""):
    """SessionLocal 을 MagicMock 으로 바꾸는 patcher contextmanager."""
    fake_proj = MagicMock()
    fake_proj.planning_doc_text = text
    sess = MagicMock()
    sess.query.return_value.filter.return_value.first.return_value = fake_proj
    return patch("app.core.database.SessionLocal", return_value=sess)


# ── source_hash ──────────────────────────────────────────────────────


def test_source_hash_deterministic_same_input(project_dir):
    pid, _ = project_dir
    with _mock_db_with_text("hello world"):
        a = svc.compute_source_hash(pid, db=None)
        b = svc.compute_source_hash(pid, db=None)
    assert a == b


def test_source_hash_changes_when_pdf_changes(project_dir):
    pid, base = project_dir
    pdf = base / "assets" / "planning_doc.pdf"
    with _mock_db_with_text(""):
        h_no_pdf = svc.compute_source_hash(pid, db=None)
        pdf.write_bytes(b"%PDF-1.4 v1")
        h_v1 = svc.compute_source_hash(pid, db=None)
        pdf.write_bytes(b"%PDF-1.4 v2-different-bytes")
        h_v2 = svc.compute_source_hash(pid, db=None)
    assert h_no_pdf != h_v1 != h_v2
    assert h_v1 != h_v2


def test_source_hash_changes_when_text_changes(project_dir):
    pid, _ = project_dir
    with _mock_db_with_text("text-A"):
        h_a = svc.compute_source_hash(pid, db=None)
    with _mock_db_with_text("text-B-different"):
        h_b = svc.compute_source_hash(pid, db=None)
    assert h_a != h_b


# ── L1 + L3 staging promote ──────────────────────────────────────────


def test_write_manifest_drops_when_status_job_id_mismatches_at_l1(project_dir):
    pid, _ = project_dir
    svc._write_status(pid, {"state": "queued", "job_id": "current_B"})
    out = svc._write_manifest(
        pid, {"x": 1}, expected_job_id="stale_A", source_hash="h",
    )
    assert out is None
    # manifest.json 은 그대로 없음 (write 자체 안 일어남)
    assert not (svc.get_project_checkpoint_dir(pid) / "manifest.json").exists()


def test_write_manifest_promotes_via_staging_when_status_matches(project_dir):
    pid, _ = project_dir
    svc._write_status(pid, {"state": "queued", "job_id": "B"})
    out = svc._write_manifest(
        pid, {"available_sections": ["characters"], "characters": [{"name": "X"}]},
        expected_job_id="B", source_hash="h_B",
    )
    assert out is not None
    cp = svc.get_project_checkpoint_dir(pid)
    assert (cp / "manifest.json").exists()
    # staging 파일은 promote 후 사라짐
    assert not (cp / "manifest.B.json").exists()
    raw = json.loads((cp / "manifest.json").read_text())
    assert raw["job_id"] == "B"
    assert raw["source_hash"] == "h_B"


def test_write_manifest_l3_does_not_clobber_other_jobs_manifest(project_dir):
    """A 가 staging 쓴 후 status 가 B 로 바뀌면, A 는 자기 staging 만 정리하고
    이미 존재하는 manifest.json 은 절대 건드리지 않아야 함 (race-safe).
    """
    pid, _ = project_dir
    cp = svc.get_project_checkpoint_dir(pid)

    # 시나리오: B 가 이미 manifest.json 을 promote 한 상태
    svc._write_status(pid, {"state": "queued", "job_id": "B"})
    svc._write_manifest(
        pid, {"job": "B-data", "available_sections": [], "characters": []},
        expected_job_id="B", source_hash="hB",
    )
    assert (cp / "manifest.json").exists()
    b_payload = json.loads((cp / "manifest.json").read_text())

    # A 는 status guard 에서 1차 drop. write_manifest 가 manifest.json 을
    # 건드리지 않음.
    out = svc._write_manifest(
        pid, {"job": "A-data"}, expected_job_id="A", source_hash="hA",
    )
    assert out is None
    # B 의 manifest 가 그대로 보존
    assert json.loads((cp / "manifest.json").read_text()) == b_payload
    # A 의 staging 도 안 만들어짐 (1차 drop 단계라서)
    assert not (cp / "manifest.A.json").exists()


def test_write_manifest_promote_drop_only_removes_own_staging(project_dir):
    """A 가 staging 까지 갔는데 promote 직전 status 가 바뀌면, A 는 자기
    staging 만 unlink — manifest.json (다른 job 의 SOT) 은 절대 안 지움.
    """
    pid, _ = project_dir
    cp = svc.get_project_checkpoint_dir(pid)

    # B 가 먼저 promote
    svc._write_status(pid, {"state": "queued", "job_id": "B"})
    svc._write_manifest(
        pid, {"job": "B-data", "available_sections": ["characters"],
              "characters": [{"name": "y"}]},
        expected_job_id="B", source_hash="hB",
    )
    assert (cp / "manifest.json").exists()
    b_payload = json.loads((cp / "manifest.json").read_text())

    # A 의 입장에서 race: L1 통과 후 status 가 바뀌는 시나리오를 흉내냄.
    # L1 가드를 일시적으로 통과시키기 위해 status 를 잠시 A 로 둠 → staging
    # 후 B 로 다시 도장.
    svc._write_status(pid, {"state": "queued", "job_id": "A"})
    original_atomic = svc._atomic_write_json

    def racing_atomic(path, payload):
        original_atomic(path, payload)
        # staging 직후 race 발생: status 를 B 로 되돌림
        if path.name.startswith("manifest.A"):
            svc._write_status(pid, {"state": "queued", "job_id": "B"})

    with patch.object(svc, "_atomic_write_json", side_effect=racing_atomic):
        out = svc._write_manifest(
            pid, {"job": "A-data"}, expected_job_id="A", source_hash="hA",
        )
    assert out is None
    # A 의 staging 은 청소됨
    assert not (cp / "manifest.A.json").exists()
    # B 의 manifest 는 절대 안 지워졌어야 함 (가장 핵심 — 이전 race bug)
    assert (cp / "manifest.json").exists()
    assert json.loads((cp / "manifest.json").read_text()) == b_payload


# ── L4 reader source_hash 가드 ───────────────────────────────────────


def test_load_project_checkpoint_rejects_stale_source_hash_without_unlink(project_dir):
    """reader 는 reject 만 — cleanup unlink 는 race 위험으로 의도적 제거.

    리뷰어 BLOCKING #2 회귀: hash 계산 도중 fresh manifest 가 promote 되어도
    reader 가 unlink 로 fresh 를 죽이지 않도록. 다음 dispatch promote 또는
    clear_project_checkpoint 가 stale 자연 정리.
    """
    pid, base = project_dir
    pdf = base / "assets" / "planning_doc.pdf"
    pdf.write_bytes(b"%PDF-1.4 src-A")
    with _mock_db_with_text("text-A"):
        svc._write_status(pid, {"state": "queued", "job_id": "X"})
        h = svc.compute_source_hash(pid, db=None)
        svc._write_manifest(
            pid, {"available_sections": ["characters"], "characters": [{"name": "A"}]},
            expected_job_id="X", source_hash=h,
        )
        # source 가 바뀜 — reader 가 reject 해야 함
        pdf.write_bytes(b"%PDF-1.4 src-B-completely-different")
        loaded = svc.load_project_checkpoint(pid, db=None)
    assert loaded is None
    # cleanup unlink 는 의도적으로 수행 안 함 — race-safe.
    assert (svc.get_project_checkpoint_dir(pid) / "manifest.json").exists(), (
        "reader 는 reject 만 — cleanup unlink 가 fresh manifest 를 죽이는 "
        "race 차단 (다음 promote / clear 가 자연 정리)"
    )


def test_load_project_checkpoint_accepts_when_source_hash_matches(project_dir):
    pid, base = project_dir
    (base / "assets" / "planning_doc.pdf").write_bytes(b"%PDF-1.4 stable")
    with _mock_db_with_text("stable-text"):
        svc._write_status(pid, {"state": "queued", "job_id": "Y"})
        h = svc.compute_source_hash(pid, db=None)
        svc._write_manifest(
            pid, {"available_sections": ["characters"],
                  "characters": [{"name": "Z"}]},
            expected_job_id="Y", source_hash=h,
        )
        loaded = svc.load_project_checkpoint(pid, db=None)
    assert loaded is not None
    assert loaded["characters"][0]["name"] == "Z"


def test_load_project_checkpoint_accepts_legacy_no_source_hash(project_dir):
    """옛 schema (source_hash 없는 manifest) 는 fail-open — 한 번 read 후
    새 분석 트리거 시 새 schema 로 덮어씀."""
    pid, _ = project_dir
    cp = svc.get_project_checkpoint_dir(pid)
    cp.mkdir(parents=True, exist_ok=True)
    (cp / "manifest.json").write_text(json.dumps({
        "data": {"available_sections": ["world_setting"], "world_setting": "old"},
        "generated_at": "2026-01-01T00:00:00",
    }))
    with _mock_db_with_text(""):
        loaded = svc.load_project_checkpoint(pid, db=None)
    assert loaded is not None
    assert loaded["world_setting"] == "old"


# ── load_status corrupt JSON surface ────────────────────────────────


def test_load_status_corrupt_json_surfaces_as_error(project_dir):
    pid, _ = project_dir
    cp = svc.get_project_checkpoint_dir(pid)
    cp.mkdir(parents=True)
    (cp / "status.json").write_text("{not valid json at all")
    s = svc.load_status(pid)
    assert s["state"] == "error"
    assert s["error"] == "status_unreadable"


def test_load_status_missing_returns_idle(project_dir):
    pid, _ = project_dir
    s = svc.load_status(pid)
    assert s == {"state": "idle"}


# ── project_has_planning_doc — empty/stale false-positive 차단 ──────


def test_project_has_planning_doc_false_when_only_empty_manifest(project_dir):
    pid, _ = project_dir
    cp = svc.get_project_checkpoint_dir(pid)
    cp.mkdir(parents=True)
    # source_hash 일치 (no source) + 빈 결과
    with _mock_db_with_text(""):
        empty_h = svc.compute_source_hash(pid, db=None)
    (cp / "manifest.json").write_text(json.dumps({
        "data": {"available_sections": [], "characters": []},
        "job_id": "Z", "source_hash": empty_h, "generated_at": "x",
    }))
    with _mock_db_with_text(""):
        assert svc.project_has_planning_doc(pid, db=None) is False


def test_project_has_planning_doc_true_when_manifest_has_content(project_dir):
    pid, base = project_dir
    (base / "assets" / "planning_doc.pdf").write_bytes(b"%PDF")
    with _mock_db_with_text("doc"):
        h = svc.compute_source_hash(pid, db=None)
        svc._write_status(pid, {"state": "queued", "job_id": "Z"})
        svc._write_manifest(
            pid, {"available_sections": ["characters"],
                  "characters": [{"name": "X"}]},
            expected_job_id="Z", source_hash=h,
        )
        assert svc.project_has_planning_doc(pid, db=None) is True


def test_project_has_planning_doc_true_when_only_pdf(project_dir):
    pid, base = project_dir
    (base / "assets" / "planning_doc.pdf").write_bytes(b"%PDF only")
    with _mock_db_with_text(""):
        assert svc.project_has_planning_doc(pid, db=None) is True


# ── atomic write — no .tmp leftover ─────────────────────────────────


def test_atomic_write_no_tmp_leftover(project_dir):
    pid, _ = project_dir
    cp = svc.get_project_checkpoint_dir(pid)
    cp.mkdir(parents=True)
    svc._atomic_write_json(cp / "manifest.json", {"a": 1})
    leftovers = [p for p in cp.iterdir() if ".tmp" in p.name]
    assert leftovers == []
    assert (cp / "manifest.json").exists()


def test_clear_project_checkpoint_removes_staging_files(project_dir):
    """clear 가 manifest+status 외에 staging file (manifest.{job_id}.json) 도 청소."""
    pid, _ = project_dir
    cp = svc.get_project_checkpoint_dir(pid)
    cp.mkdir(parents=True)
    (cp / "manifest.json").write_text("{}")
    (cp / "status.json").write_text("{}")
    (cp / "manifest.A.json").write_text("{}")
    (cp / "manifest.B.json").write_text("{}")
    svc.clear_project_checkpoint(pid)
    assert not (cp / "manifest.json").exists()
    assert not (cp / "status.json").exists()
    assert not (cp / "manifest.A.json").exists()
    assert not (cp / "manifest.B.json").exists()


# ── get_planning_context 위임 가드 ───────────────────────────────────


def test_get_planning_context_uses_validated_reader(project_dir):
    """L4 우회 차단 회귀 — get_planning_context 는 raw read 로 stale manifest
    를 prompt 에 주입하지 말아야 함."""
    from app.core.planning_doc_context import get_planning_context

    pid, base = project_dir
    pdf = base / "assets" / "planning_doc.pdf"
    pdf.write_bytes(b"%PDF-1.4 src-A")

    with _mock_db_with_text("text-A"):
        svc._write_status(pid, {"state": "queued", "job_id": "X"})
        h = svc.compute_source_hash(pid, db=None)
        svc._write_manifest(
            pid, {"available_sections": ["world_setting"],
                  "characters": [], "world_setting": "from-source-A",
                  "tone_mood": "", "story_arc": "", "visual_concepts": "",
                  "key_relationships": []},
            expected_job_id="X", source_hash=h,
        )

        # 정상 read — A 의 데이터 주입
        ctx = get_planning_context(pid, episode_id="any-ep")
        assert ctx.has_planning_doc is True
        assert ctx.world_setting == "from-source-A"

        # source 가 바뀜 — get_planning_context 가 stale 을 reject 해야
        pdf.write_bytes(b"%PDF-1.4 src-B-totally-different-content")
        ctx2 = get_planning_context(pid, episode_id="any-ep")
        assert ctx2.has_planning_doc is False, (
            "get_planning_context must use load_project_checkpoint() — "
            "raw reader bypass would resurrect stale manifest into prompt"
        )
        assert ctx2.world_setting == ""


# ── thread-level e2e race — 핵심 BLOCKING 회귀 ──────────────────────


def test_project_lock_serializes_concurrent_writes(project_dir):
    """_project_lock 안에서 모든 status write 가 직렬화 — 동시에 수많은
    thread 가 status 를 갱신해도 마지막 thread 의 값이 SOT 로 남아야.
    """
    pid, _ = project_dir
    N = 20
    finished_order = []

    def writer(idx):
        with svc._project_lock(pid):
            time.sleep(0.005)  # critical section 안에서 일부러 시간 끌기
            svc._atomic_write_json(
                svc.get_project_checkpoint_dir(pid) / "status.json",
                {"state": "queued", "job_id": f"job_{idx:02d}"},
            )
            finished_order.append(idx)

    threads = [threading.Thread(target=writer, args=(i,)) for i in range(N)]
    for t in threads:
        t.start()
    for t in threads:
        t.join()

    # 모든 thread 가 직렬화되어 finished_order 에 N 개 모두 — 손실 없음
    assert len(finished_order) == N
    final_status = svc.load_status(pid)
    final_job = final_status["job_id"]
    last_winner = f"job_{finished_order[-1]:02d}"
    assert final_job == last_winner, (
        f"final status job_id={final_job} should be last writer={last_winner}"
    )


def test_status_write_race_dispatch_and_running_status(project_dir, monkeypatch):
    """리뷰어 BLOCKING #1 회귀 — A 가 running status 쓰기 직전 B 가 dispatch
    해도, lock 으로 직렬화되어 status 가 뒤집히지 않아야.

    lock 도입 전에는: A check OK → B dispatch (status=B) → A write (status=A,
    덮어씀) → B 의 background thread 가 stale 로 판단되어 결과 drop.

    lock 도입 후: A 의 check+write 가 atomic, B 는 lock 못 잡아 대기.
    A 가 끝나면 B 가 lock 잡고 dispatch → status=B 로 도장 → 이후 A 의
    background thread 가 "stale" 로 자연 bail-out (정상 동작).
    """
    pid, _ = project_dir
    # 분석 thread 없이 _write_status 동작만 검증 — race 시뮬:
    # T0: status={job:A}
    svc._write_status(pid, {"state": "queued", "job_id": "A"})

    # A 가 lock 잡고 running 마킹 시도하는 동안 B dispatch 도 lock 잡으려 시도.
    # lock 으로 직렬화 → 마지막 lock holder 의 도장이 SOT.
    barrier = threading.Barrier(2)
    a_done = threading.Event()

    def thread_a():
        with svc._project_lock(pid):
            barrier.wait()  # B 가 동시에 시작했음을 보장
            time.sleep(0.05)  # B 가 lock 대기 중
            if svc._is_current_job(pid, "A"):
                svc._atomic_write_json(
                    svc.get_project_checkpoint_dir(pid) / "status.json",
                    {"state": "running", "job_id": "A"},
                )
        a_done.set()

    def thread_b_dispatch():
        barrier.wait()
        a_done.wait(timeout=2.0)  # A 가 끝날 때까지 대기 보장
        with svc._project_lock(pid):
            svc._atomic_write_json(
                svc.get_project_checkpoint_dir(pid) / "status.json",
                {"state": "queued", "job_id": "B"},
            )

    ta = threading.Thread(target=thread_a)
    tb = threading.Thread(target=thread_b_dispatch)
    ta.start(); tb.start()
    ta.join(); tb.join()

    final = svc.load_status(pid)
    # 마지막 writer 는 B (dispatch). status.job_id == B.
    assert final["job_id"] == "B"


def test_reader_does_not_unlink_during_concurrent_promote(project_dir):
    """리뷰어 BLOCKING #2 e2e — reader 가 hash 계산 도중 fresh manifest 가
    promote 되어도 unlink 하지 않음.

    lock 도입 전: reader 가 raw 읽고 → compute_hash 시간 → 그 사이 새 promote
    → reader 가 cp.unlink() → fresh 손실.

    cleanup 제거 후: reader 는 reject 만 → fresh manifest 보존. caller 는
    None 받지만 다음 호출 (혹은 promote 직후 호출) 에서 새 manifest 정상
    read.
    """
    pid, base = project_dir
    pdf = base / "assets" / "planning_doc.pdf"
    pdf.write_bytes(b"%PDF-1.4 src-A")

    fresh_payload = {
        "data": {"available_sections": ["characters"],
                 "characters": [{"name": "FRESH"}]},
        "job_id": "B",
        "source_hash": None,  # 계산 후 채움
        "generated_at": "2026-01-01",
    }

    # A 의 stale manifest 를 disk 에 둠
    with _mock_db_with_text("text-A"):
        h_a = svc.compute_source_hash(pid, db=None)
    svc._write_status(pid, {"state": "queued", "job_id": "A"})
    svc._write_manifest(
        pid, {"available_sections": ["characters"], "characters": [{"name": "STALE"}]},
        expected_job_id="A", source_hash=h_a,
    )

    # source 변경 (reader 가 hash mismatch 감지하도록)
    pdf.write_bytes(b"%PDF-1.4 src-B-different")

    cp = svc.get_project_checkpoint_dir(pid) / "manifest.json"
    assert cp.exists()
    with _mock_db_with_text("text-A"):  # text 그대로, PDF 만 바뀜
        loaded = svc.load_project_checkpoint(pid, db=None)
    # reader 는 reject (None)
    assert loaded is None
    # 그러나 fresh-replacement 시뮬: 다른 thread 가 그 사이 새 manifest 를
    # promote 했다고 가정. cleanup 이 unlink 한다면 이 fresh 가 죽음.
    # cleanup 제거되었으므로 manifest 파일은 그대로 (또는 다음 promote 가 덮음).
    assert cp.exists(), (
        "reader cleanup 이 제거되어 fresh-promote 시나리오 안전. "
        "다음 dispatch promote 가 덮어씀."
    )


def test_run_sync_phase1_exception_marks_status_error(project_dir, monkeypatch):
    """리뷰어 IMPORTANT — Phase 1 (PDF read/DB query/config load) 예외가
    status 에 반영되어야 함. outer try/except 가 status=error 마킹.

    이전 구조에서는 Phase 1 예외 시 finally 가 db.close 만 하고 raise →
    status 가 running/queued 에 stuck. outer except 추가로 보장.
    """
    pid, base = project_dir
    pdf = base / "assets" / "planning_doc.pdf"
    pdf.write_bytes(b"%PDF-1.4 dummy")

    fake_proj = MagicMock()
    fake_proj.planning_doc_text = "x" * 200

    # compute_source_hash 가 Phase 1 안에서 폭발 (DB query / config 는 통과).
    boom = RuntimeError("phase1 source_hash boom")
    monkeypatch.setattr(svc, "compute_source_hash", MagicMock(side_effect=boom))
    monkeypatch.setattr(
        "app.services.analysis_dispatch_service.load_project_llm_config",
        lambda *a, **k: {},
    )

    sess = MagicMock()
    sess.query.return_value.filter.return_value.first.return_value = fake_proj
    monkeypatch.setattr("app.core.database.SessionLocal", MagicMock(return_value=sess))

    # Phase 1 안에 도달하려면 status job_id 가 일치해야
    job_id = "phase1err"
    svc._write_status(pid, {"state": "queued", "job_id": job_id})

    with pytest.raises(RuntimeError, match="phase1 source_hash boom"):
        svc.run_planning_doc_analysis_sync(
            pid, job_id=job_id, queued_at="t0",
        )

    final = svc.load_status(pid)
    assert final["state"] == "error", (
        f"Phase 1 예외 시 status 가 error 로 마킹되어야 함, got {final}"
    )
    assert final["job_id"] == job_id
    assert "phase1 source_hash boom" in final.get("error", "")


def test_run_sync_phase2_text_fallback_exception_marks_status_error(project_dir, monkeypatch):
    """Phase 2 (lock 밖 LLM call) text fallback 예외도 outer except 로
    status=error 마킹되어야 함. PDF multimodal 1차 실패 + text 2차 raise.
    """
    pid, base = project_dir
    # text-only path (PDF 없음) 로 가도록
    fake_proj = MagicMock()
    fake_proj.planning_doc_text = "x" * 200

    monkeypatch.setattr(
        "app.services.analysis_dispatch_service.load_project_llm_config",
        lambda *a, **k: {},
    )
    sess = MagicMock()
    sess.query.return_value.filter.return_value.first.return_value = fake_proj
    monkeypatch.setattr("app.core.database.SessionLocal", MagicMock(return_value=sess))
    monkeypatch.setattr(
        "app.modules.llm.llm_client.call_structured",
        MagicMock(side_effect=RuntimeError("LLM boom")),
    )

    job_id = "phase2err"
    svc._write_status(pid, {"state": "queued", "job_id": job_id})
    with pytest.raises(RuntimeError, match="LLM boom"):
        svc.run_planning_doc_analysis_sync(
            pid, job_id=job_id, queued_at="t0",
        )
    final = svc.load_status(pid)
    assert final["state"] == "error"
    assert final["job_id"] == job_id
    assert "LLM boom" in final.get("error", "")


def test_outer_except_respects_stale_job_guard(project_dir, monkeypatch):
    """outer except 의 _write_status 가 stale 시 guard 로 drop — 늦게 끝난
    A thread 가 B 의 status 를 자기 error 로 덮지 않음.
    """
    pid, base = project_dir
    fake_proj = MagicMock(); fake_proj.planning_doc_text = "x"*200

    monkeypatch.setattr(
        "app.services.analysis_dispatch_service.load_project_llm_config",
        lambda *a, **k: {},
    )
    sess = MagicMock()
    sess.query.return_value.filter.return_value.first.return_value = fake_proj
    monkeypatch.setattr("app.core.database.SessionLocal", MagicMock(return_value=sess))

    # text path. call_structured 가 LLM 호출 시점에 그 사이 다른 dispatch (B)
    # 가 status 를 도장찍었다고 가정. A 가 raise 하고 outer except 가
    # _write_status(expected_job_id="A") 호출 → stale → drop.
    job_id_a = "A_stale"
    job_id_b = "B_fresh"

    def call_then_swap(**kw):
        # B 가 새 dispatch — status 도장 변경
        svc._write_status(pid, {"state": "queued", "job_id": job_id_b})
        raise RuntimeError("A LLM boom")

    monkeypatch.setattr(
        "app.modules.llm.llm_client.call_structured",
        call_then_swap,
    )

    # A 가 dispatch — 자기 status 도장
    svc._write_status(pid, {"state": "queued", "job_id": job_id_a})
    with pytest.raises(RuntimeError, match="A LLM boom"):
        svc.run_planning_doc_analysis_sync(pid, job_id=job_id_a, queued_at="t0")

    final = svc.load_status(pid)
    # B 의 도장이 보존되어야 — A 의 outer except 가 stale 로 drop
    assert final["job_id"] == job_id_b, (
        f"B 의 status 가 보존되어야 — A 의 stale error 가 덮으면 안 됨. got={final}"
    )
    assert final["state"] == "queued"


def test_thread_race_late_finisher_does_not_resurrect_stale_sot(project_dir, monkeypatch):
    """A 가 분석 늦게 끝나도 B 의 SOT 를 덮지 못하고, manifest.job_id == B."""
    from app.modules.llm import llm_client
    from app.services import analysis_dispatch_service

    pid, base = project_dir
    (base / "assets" / "planning_doc.pdf").write_bytes(b"%PDF-1.4 src-A")

    fake_result = {
        "characters": [{"name": "X", "description": "y"}],
        "world_setting": "w", "tone_mood": "t", "story_arc": "s",
        "visual_concepts": "v", "key_relationships": [],
        "available_sections": [],
    }

    def slow_call(**kw):
        time.sleep(0.4)
        return fake_result

    monkeypatch.setattr(llm_client, "call_structured", slow_call)
    monkeypatch.setattr(
        analysis_dispatch_service, "load_project_llm_config",
        lambda *a, **k: {},
    )

    with _mock_db_with_text("x" * 200):
        job_a = svc.dispatch_planning_doc_analysis_async(pid)
        time.sleep(0.05)
        # source 갈아엎기 + 새 dispatch
        svc.clear_project_checkpoint(pid)
        (base / "assets" / "planning_doc.pdf").write_bytes(b"%PDF-1.4 src-B")
        job_b = svc.dispatch_planning_doc_analysis_async(pid)
        time.sleep(1.0)
        cp = svc.get_project_checkpoint_dir(pid) / "manifest.json"
        assert cp.exists(), "B should have promoted manifest"
        raw = json.loads(cp.read_text())
        assert raw["job_id"] == job_b, (
            f"manifest.job_id={raw['job_id']} should be B={job_b} (not A={job_a})"
        )
        # stale staging file 은 모두 청소되어 있어야 함
        leftover_stagings = list(
            svc.get_project_checkpoint_dir(pid).glob("manifest.*.json")
        )
        assert leftover_stagings == [], f"stale staging: {leftover_stagings}"


# ── 성공 경로 끝까지 (2026-09-18, Codex BLOCK A·B) ────────────────────


def test_run_sync_success_path_writes_locations_and_done(project_dir, monkeypatch):
    """★업로드 성공 경로를 **끝까지** 태운다.

    2026-09-18 에 장소 칸을 더하면서 `available_sections` 계산을 스텝과 공용
    함수로 합쳤는데, 두 번 죽었다 — 둘 다 **분석을 정상으로 마친 뒤**였다:

      A. import 를 부르는 함수(`run_planning_doc_analysis_sync`)에 적어 두고
         실제로 쓰는 함수(`_run_planning_doc_analysis_phases`)에서 NameError
      B. `computed` 대입을 지웠는데 상태 기록·로그가 그대로 그 이름을 씀

    「소스에 이 문자열이 있나」로는 둘 다 통과한다. 그래서 LLM·DB·파일 경계만
    대역으로 두고 **함수가 끝까지 돌아 manifest 와 status 를 남기는지**를 본다.
    """
    pid, base = project_dir
    (base / "assets" / "planning_doc.pdf").write_bytes(b"%PDF-1.4 dummy")

    analysed = {
        "characters": [{"name": "갑", "description": "설명"}],
        "locations": [{"name": "어디", "description": "좁다",
                       "visual_traits": "젖은 콘크리트"}],
        "world_setting": "어떤 시대", "tone_mood": "", "story_arc": "",
        "visual_concepts": "", "key_relationships": [],
        "available_sections": [],
    }
    monkeypatch.setattr(
        "app.modules.llm.llm_client.call_structured",
        lambda **kw: dict(analysed))
    monkeypatch.setattr(
        "app.services.analysis_dispatch_service.load_project_llm_config",
        lambda *a, **k: {})

    job_id = "okpath"
    svc._write_status(pid, {"state": "queued", "job_id": job_id})
    with _mock_db_with_text("x" * 200):
        result = svc.run_planning_doc_analysis_sync(
            pid, job_id=job_id, queued_at="t0")

    assert "locations" in result["available_sections"], (
        f"장소가 available 에 안 실렸다: {result['available_sections']}")

    cp = svc.get_project_checkpoint_dir(pid) / "manifest.json"
    assert cp.exists(), "성공 경로인데 manifest 가 없다"
    saved = json.loads(cp.read_text(encoding="utf-8"))
    data = saved.get("data", saved)
    assert data["locations"][0]["visual_traits"] == "젖은 콘크리트"
    assert "locations" in data["available_sections"]

    status = svc.load_status(pid)
    assert status["state"] == "done", f"상태가 done 이 아니다: {status}"
    assert status["sections"] == result["available_sections"], (
        "상태에 적힌 절 목록이 결과와 다르다")


def test_run_sync_success_path_reaches_the_reader(project_dir, monkeypatch):
    """저장한 것이 **소비자(get_planning_context)**까지 닿는지 — 끝점."""
    pid, base = project_dir
    (base / "assets" / "planning_doc.pdf").write_bytes(b"%PDF-1.4 dummy")
    monkeypatch.setattr(
        "app.modules.llm.llm_client.call_structured",
        lambda **kw: {
            "characters": [], "locations": [
                {"name": "어디", "description": "좁다",
                 "visual_traits": "젖은 콘크리트"}],
            "world_setting": "", "tone_mood": "", "story_arc": "",
            "visual_concepts": "", "key_relationships": [],
            "available_sections": []})
    monkeypatch.setattr(
        "app.services.analysis_dispatch_service.load_project_llm_config",
        lambda *a, **k: {})

    job_id = "reader"
    svc._write_status(pid, {"state": "queued", "job_id": job_id})
    with _mock_db_with_text("x" * 200):
        svc.run_planning_doc_analysis_sync(pid, job_id=job_id, queued_at="t0")

        from app.core.planning_doc_context import get_planning_context
        ctx = get_planning_context(pid, "any-episode", db=None)

    assert ctx.locations_text, "저장은 됐는데 읽는 쪽에 장소가 안 온다"
    assert "젖은 콘크리트" in ctx.locations_text
    assert "장소" in ctx.inject_if_available(
        "locations_text", "## 기획서 장소 참고 정보")