"""entity_detail 은 요소를 **잘라서** 묻는다 (2026-09-17 컨트리로드 실측).

## 무엇이 결함이었나

241개를 한 호출에 실었더니 gpt-6-astra 가 **1개만 답하고 `stop`** 했다
(출력 689토큰). 빠진 240개를 다시 보낸 재시도도 1개. 결과는 `partial 2/241`
로 **경고 한 줄 없이** 넘어가, 뒤 단계 59개가 빈 인물 정보 위에서 돌 뻔했다.
같은 입력에 목록만 60개·30개로 자르니 60/60 · 30/30 이 돌아왔다.

## 잠그는 것

① 묶음마다 50개 이하 · 모든 요소가 **정확히 한 묶음**에 실린다
② 대본 전문은 묶음마다 **자르지 않고** 실린다 (CLAUDE.md 절대 규칙)
③ 재시도에도 세계관·기획서 인물 절이 실리고, 특이 외형이 버려지지 않는다
④ 한 묶음이 터져도 다른 묶음의 답은 산다
⑤ 끝내 빈 요소가 있으면 경고가 남는다
⑥ 표식 칸은 **그 호출에 실린 표식 중 하나**만 받는다 — 표식 없는 줄이 섞이면 빈 값도

## ⑥ 은 왜

50개로 잘라도 장소 50·소품 41 묶음이 **1개만 쓰고 끝났다**(첫 호출·재시도 모두).
답의 맨 끝 칸인 표식을 쓰다가 글자가 뭉개지고('P41ள', 'L64 "location"') 목록을
닫아 버린다. 같은 입력에서 원래 스키마는 6번 중 5번 1개로 끝났고, 표식을 그 묶음의
표식으로 묶은 스키마는 4번 모두 전부(50/50 · 41/41) 왔고 표식↔이름 짝도 전부 맞았다.
"""
from __future__ import annotations

import logging
import re

import app.core.steps.entity_steps as es
from app.core.steps.entity_steps import EntityDetailStep, _DETAIL_CHUNK_SIZE
from app.modules.pipeline.entity_extractor_v3 import _load_schema as _real_load_schema

FULLTEXT = "대본 전문 " + ("가나다라마바사 " * 400)
PLAN = "\n\n## 기획서 인물 참고 정보\n<planning_doc_reference>\n- **테스트인물**: 설명\n</planning_doc_reference>"


def _step(monkeypatch, tmp_path, *, n, answer, calls):
    from app.core.config import settings
    monkeypatch.setattr(settings, "projects_dir", str(tmp_path / "projects"))
    st = EntityDetailStep.__new__(EntityDetailStep)
    st.project_id, st.episode_id = "p", "e"
    st.project_config = {}
    st.db = None
    st.build_opik_metadata = lambda *a, **k: {"tags": list(a[0]) if a else []}
    st._load_cleaned_text = lambda: FULLTEXT
    chars = [{"name": f"인물{i:03d}", "short_id": f"C{i:03d}"} for i in range(1, n + 1)]
    st._load_prev_checkpoint = lambda sid: (
        {"status": "completed", "data": {"filtered_entities": {"characters": chars}}}
        if sid == "entity_filter" else
        {"status": "completed", "data": {"era": "근미래", "region": "인천", "rules": []}}
        if sid == "visual_world_rules" else None)
    st._journal = lambda *a, **k: None

    class _P:
        has_planning_doc = True
        is_first_episode = True
        def inject_if_available(self, section, header=""):
            return PLAN if section == "characters_text" else ""
    monkeypatch.setattr("app.core.planning_doc_context.get_planning_context",
                        lambda pid, eid, db=None: _P())

    def fake_call(**kw):
        calls.append(kw)
        return answer(kw, len(calls))
    monkeypatch.setattr(es, "call_structured", fake_call, raising=False)
    monkeypatch.setattr("app.modules.pipeline.entity_extractor_v3._load_prompt",
                        lambda name, **k: f"[목록]\n{k.get('entity_list','')}\n[시나리오]\n{k.get('fulltext','')}")
    monkeypatch.setattr("app.modules.pipeline.entity_extractor_v3._load_schema", lambda name: {"type": "object"})
    monkeypatch.setattr(st, "_config_hash", lambda: "h", raising=False)
    return st


def _rows(prompt):
    return re.findall(r"- \[(C\d+)\] (인물\d+) \((\w+)\)", prompt)


def _echo(kw, _i):
    return {"entities": [
        {"entity_id": sid, "name": nm, "entity_type": t, "description": f"d-{sid}",
         "visual_traits": ["v"],
         "distinctive_visual_traits": [{"trait": f"t-{sid}", "source_quote": "q"}]}
        for sid, nm, t in _rows(kw["user_prompt"])]}


def _first_line_only_then_all(kw, i):
    """첫 호출은 첫 줄만 답한다(실측 모양) — 재시도가 나머지를 받는다."""
    rows = _rows(kw["user_prompt"])
    if kw["schema_name"] == "entity_detail_batch":
        rows = rows[:1]
    return _echo({"user_prompt": "\n".join(f"- [{s}] {n} ({t})" for s, n, t in rows)}, i)


def test_1_every_entity_rides_exactly_one_chunk_of_at_most_50(monkeypatch, tmp_path):
    calls = []
    st = _step(monkeypatch, tmp_path, n=121, answer=_echo, calls=calls)
    out = st._execute()
    assert out["completed_count"] == 121 and out["failed_count"] == 0
    first_pass = [c for c in calls if c["schema_name"] == "entity_detail_batch"]
    sizes = [len(_rows(c["user_prompt"])) for c in first_pass]
    assert all(s <= _DETAIL_CHUNK_SIZE for s in sizes), sizes
    ids = [sid for c in first_pass for sid, _, _ in _rows(c["user_prompt"])]
    assert sorted(ids) == sorted(f"C{i:03d}" for i in range(1, 122)), "빠지거나 겹친 요소가 있다"
    assert len(first_pass) == 3


def test_2_full_scenario_text_rides_every_chunk_untruncated(monkeypatch, tmp_path):
    calls = []
    st = _step(monkeypatch, tmp_path, n=120, answer=_echo, calls=calls)
    st._execute()
    for c in calls:
        assert FULLTEXT in c["user_prompt"], "대본 전문이 잘렸거나 빠졌다"


def test_3_retry_keeps_world_planning_and_distinctive(monkeypatch, tmp_path):
    """첫 답은 첫 줄만(실측 모양) → 재시도가 나머지를 받는다."""
    calls = []
    st = _step(monkeypatch, tmp_path, n=60, answer=_first_line_only_then_all, calls=calls)
    out = st._execute()
    assert out["completed_count"] == 60
    retries = [c for c in calls if c["schema_name"] == "entity_detail_retry"]
    assert retries, "재시도가 안 불렸다"
    for c in retries:
        assert PLAN in c["user_prompt"], "재시도에 기획서 인물 절이 빠졌다"
        assert "[세계관]" in c["user_prompt"], "재시도에 세계관이 빠졌다"
    details = out["data"]["entity_details"]
    assert details["C060"]["distinctive_visual_traits"], "재시도 답의 특이 외형이 버려졌다"


def test_4_one_chunk_blowing_up_keeps_the_others(monkeypatch, tmp_path):
    calls = []

    def second_chunk_explodes(kw, i):
        rows = _rows(kw["user_prompt"])
        if any(sid == "C051" for sid, _, _ in rows):
            raise RuntimeError("boom")
        return _echo(kw, i)

    st = _step(monkeypatch, tmp_path, n=120, answer=second_chunk_explodes, calls=calls)
    out = st._execute()
    assert out["completed_count"] == 70, out["completed_count"]   # 1묶음 50 + 3묶음 20
    assert out["failed_count"] == 50


def test_5_leftover_empties_are_logged_not_silent(monkeypatch, tmp_path, caplog):
    calls = []
    st = _step(monkeypatch, tmp_path, n=10, answer=lambda kw, i: {"entities": []}, calls=calls)
    with caplog.at_level(logging.WARNING, logger=es.logger.name):
        out = st._execute()
    assert out["completed_count"] == 0
    assert any("끝내 비었다" in r.getMessage() for r in caplog.records), "빈 요소가 로그에 안 남았다"


def test_6_workers_carry_stop_check_budget_and_trace(monkeypatch, tmp_path):
    """★정지 표·텍스트 예산·Opik trace 는 스레드마다 따로다 (Codex BLOCK 2026-09-17).

    묶음이 둘 이상이면 스레드 풀로 가는데, 그냥 넘기면 worker 안의 호출은
    셋 다 못 본다 — 정지가 안 들리고 예산이 안 세어지고 계층이 끊긴다.
    """
    import threading

    from app.core import image_call_budget as icb
    from app.core import research_call_budget as rcb
    from app.modules.llm import opik_trace as ot

    stop_sentinel = object()
    budget = rcb.ResearchCallBudget.__new__(rcb.ResearchCallBudget)
    trace_sentinel = object()
    seen = []

    def spy(kw, i):
        seen.append((threading.get_ident(), icb.get_current_stop_check(),
                     rcb.get_current_budget(), ot.current_trace()))
        return _echo(kw, i)

    calls = []
    st = _step(monkeypatch, tmp_path, n=120, answer=spy, calls=calls)
    main = threading.get_ident()
    prev_stop, prev_budget = icb.get_current_stop_check(), rcb.get_current_budget()
    icb.install_stop_check(stop_sentinel)
    rcb.install_budget(budget)
    token = ot._trace_ctx.set(trace_sentinel)
    try:
        st._execute()
    finally:
        ot._trace_ctx.reset(token)
        icb.install_stop_check(prev_stop)
        if prev_budget is not None:
            rcb.install_budget(prev_budget)
        else:
            rcb.uninstall_budget()
    assert any(tid != main for tid, *_ in seen), "스레드 풀 갈래를 안 탔다 — 시험이 무의미"
    for tid, stop, bud, tr in seen:
        assert stop is stop_sentinel, "worker 가 정지 표를 못 본다"
        assert bud is budget, "worker 가 텍스트 예산을 못 본다"
        assert tr is trace_sentinel, "worker 가 Opik trace 를 못 본다"


def _prod_like(sends, *, after=None):
    """프로덕션 `_completion` 이 보내기 전에 하는 두 가지를 **진짜 함수로** 한다.

    ① 이 스레드의 정지 확인  ② 조사 예산 예약(팔을 들었을 때만).
    둘을 지나야 「보냈다」로 센다 — 대역이 문맥을 무시하면 이 시험은 무의미하다.
    """
    from app.core.research_call_budget import reserve_current_research_call
    from app.modules.llm import llm_client

    def answer(kw, i):
        chk = llm_client._current_stop_check()
        if chk is not None:
            chk()
        reserve_current_research_call(source="test.entity_detail")
        sends.append(kw["schema_name"])
        if after:
            after(kw)
        return _echo(kw, i)
    return answer


def test_7_cancel_before_start_sends_nothing_and_propagates(monkeypatch, tmp_path):
    import pytest

    from app.core import image_call_budget as icb
    from app.core.errors import AppError

    def cancelled():
        raise AppError(code="step.cancelled", message="정지", status_code=409)

    sends, calls = [], []
    st = _step(monkeypatch, tmp_path, n=120, answer=_prod_like(sends), calls=calls)
    prev = icb.get_current_stop_check()
    icb.install_stop_check(cancelled)
    try:
        with pytest.raises(AppError) as ei:
            st._execute()
    finally:
        icb.install_stop_check(prev)
    assert ei.value.code == "step.cancelled", "정지가 묶음 실패로 삼켜졌다"
    assert sends == [], f"정지 뒤에 보냈다: {sends}"


def test_8_cancel_mid_run_is_not_retried(monkeypatch, tmp_path):
    """정지가 도중에 올라오면 **삼키지 않고** 올린다 — 재시도·다른 묶음이 더 안 보낸다.

    정지 확인은 **첫 한 번만 통과**하고 그다음부터 선다. 그래서 스레드 순서와
    무관하게 전송은 정확히 1회여야 한다(옛 코드는 둘째 묶음의 정지를 삼키고
    재시도로 넘어가 예외 없이 partial 을 돌려준다).
    """
    import threading

    import pytest

    from app.core import image_call_budget as icb
    from app.core.errors import AppError

    lock, passed = threading.Lock(), [0]

    def first_check_only():
        with lock:
            passed[0] += 1
            if passed[0] > 1:
                raise AppError(code="step.cancelled", message="정지", status_code=409)

    sends, calls = [], []
    monkeypatch.setattr(es, "_DETAIL_CHUNK_SIZE", 20)
    st = _step(monkeypatch, tmp_path, n=40, answer=_prod_like(sends), calls=calls)
    prev = icb.get_current_stop_check()
    icb.install_stop_check(first_check_only)
    try:
        with pytest.raises(AppError) as ei:
            st._execute()
    finally:
        icb.install_stop_check(prev)
    assert ei.value.code == "step.cancelled"
    assert sends == ["entity_detail_batch"], f"정지 뒤에도 보냈다: {sends}"


def test_9_research_budget_cap_is_honoured_inside_workers(monkeypatch, tmp_path):
    import pytest

    from app.core import research_call_budget as rcb

    sends, calls = [], []
    st = _step(monkeypatch, tmp_path, n=120, answer=_prod_like(sends), calls=calls)
    prev = rcb.get_current_budget()
    rcb.install_budget(rcb.ResearchCallBudget(cap=0))
    try:
        with rcb.research_calls_armed():
            with pytest.raises(rcb.ResearchCallBudgetExceeded):
                st._execute()
    finally:
        if prev is not None:
            rcb.install_budget(prev)
        else:
            rcb.uninstall_budget()
    assert sends == [], f"상한 0 인데 worker 가 보냈다: {sends}"


def _id_enum(kw):
    return kw["response_schema"]["properties"]["entities"]["items"]["properties"]["entity_id"].get("enum")


def test_10_entity_id_accepts_only_the_ids_sent_in_that_call(monkeypatch, tmp_path):
    """첫 호출·재시도 모두 표식 칸이 **그 호출의 목록**과 같다 (팩 스키마 그대로 · 묶음끼리 안 섞인다)."""
    calls = []
    st = _step(monkeypatch, tmp_path, n=60, answer=_first_line_only_then_all, calls=calls)
    monkeypatch.setattr("app.modules.pipeline.entity_extractor_v3._load_schema", _real_load_schema)
    out = st._execute()
    assert out["completed_count"] == 60
    assert {c["schema_name"] for c in calls} == {"entity_detail_batch", "entity_detail_retry"}
    for c in calls:
        assert _id_enum(c) == sorted(sid for sid, _, _ in _rows(c["user_prompt"])), c["schema_name"]


def test_11_a_row_without_id_keeps_empty_allowed(monkeypatch, tmp_path):
    """표식 없는 줄이 섞이면 빈 값도 받는다 — 막으면 그 줄의 답이 남의 표식을 달고 덮는다."""
    calls = []
    st = _step(monkeypatch, tmp_path, n=1, answer=_echo, calls=calls)
    monkeypatch.setattr("app.modules.pipeline.entity_extractor_v3._load_schema", _real_load_schema)
    st.project_config = {"grounding_mode": "legacy"}
    chars = [{"name": "인물001", "short_id": "C001"}, {"name": "표식없는인물"}]
    st._load_prev_checkpoint = lambda sid: (
        {"status": "completed", "data": {"filtered_entities": {"characters": chars}}}
        if sid == "entity_filter" else None)
    st._execute()
    assert _id_enum(calls[0]) == ["C001", ""]