"""shot_continuity_author — 자세 정본·중의어 교정·carried state 저작 (2026-07-13).

s40/s41 레시피의 텍스트 안전망 3종(스틸·콘티 프롬프트에 주입될 연속성 문장)을
LLM 으로 저작한다. 정본: scratchpad/forest_exp/s37_rooftop_rebuild.py
stage_pose_canon / stage_pose_fix / stage_int_cont.

  1) pose_canon: 움직일 수 없는 인물(사망·의식불명)의 자세 정본 —
     등장 전 샷에 글자 그대로 재사용, 중의어(위치 lie 등) 회피.
  2) carried: 샷별 '이어받는 상태' — ★양방향 전파·씬 경계 차단은
     프롬프트 계약이 담당(코드 재구현 금지). 실험의 실내 한정(int_cont)을
     전 샷으로 일반화(계약 첫 줄만 수정, 규칙 본문 verbatim).
     ★v3(감사 1-B) 부터 **세 갈래**로 받는다 — objects / people[] /
     offscreen_effects. 누가 화면에 보이는지는 모델이 아니라 **코드가
     그 샷의 visible entity id 로** 정한다(`carried_text_for_shot`).
  3) pose_fix: 정본 샷의 movement/figures/state 문장 중의성 최소 교정.
     production v1 은 movement/figures 저작이 없으므로 빈 문자열로 전달
     (state 교정이 실효) — 도입 시 그대로 확장. v3 부터 교정본을
     `carried` 안 **제자리에** 넣는다(종전에는 조립이 두 값을 `or` 로
     골라 **일부러 비운 칸이 옛 값으로 되살아났다**).

코드 강제: pose_canon.shots 의 미지 태그 제거(+경고), carried 전 샷 누락
fail-fast, character_ko → EntityCanon short_id 정확 일치 매핑(불일치=None,
글자 부분매칭 금지).
"""
from __future__ import annotations

import logging
import threading
from typing import Any, Callable, Dict, List, Optional

from app.core.errors import AppError
from app.modules.prompt_loader import load_prompt, load_schema
from app.modules.pipeline.shot_ref_classify import tag_of

logger = logging.getLogger(__name__)

_MODULE = "shot_continuity"

PROMPT_VERSION_MAP = {
    "1": "1.202607132300",
    # v2 (E2E6 피드백 ④, 2026-07-16): pose_canon character_ko 를
    # EntityCanon 정본 명단(닫힌 집합)의 표기 그대로 저작하도록 계약 —
    # LLM 자유 표기("강민숙(민숙)")가 정확 일치 매핑을 깨서 short_id=None
    # 이 되던 identity 해소 공백 교정. 명단 밖 단역은 원문 표기 유지.
    "2": "2.202607161500",
    # v3 (감사 1-B, 2026-08-27): carried 를 **세 갈래**로 받는다 —
    # objects / people[] / offscreen_effects. 종전 한 칸(`carried_en`)은
    # 씬 단위로 저작돼 **그 샷에 없는 인물까지** 「여전히 있다」고 썼고,
    # 같은 프롬프트의 "they must be one of: 연희 — never anyone else" 와
    # 정면으로 맞섰다(실측: 인물 샷의 23~57%). 08-26 reve 변환 반려 사유
    # 「같은 두 남성이 여러 번 반복」이 그 결과다.
    #
    # ★**모델은 화면 안팎을 정하지 않는다.** 누구의 무슨 상태인지만
    #  말하고, 보이는지는 **코드가 그 샷의 visible entity id 로** 고른다
    #  (`carried_text_for_shot`). 자가신고를 계약으로 삼지 않는다.
    "3": "3.202608270900",
}

# v2+ 팩만 '등장인물 정본 명단' 입력 블록을 받는다 (v1 조립 byte-identical)
_NAME_ROSTER_PACKS = {"2", "3"}

# carried 를 세 갈래로 받는 팩 — 이 밖에서는 옛 한 칸 계약 그대로 돈다.
_PEOPLE_CONTRACT_PACKS = {"3"}

# 저작 산출에 함께 적는 계약 표시. ★selector 는 **코드의 현재 값**이라
# 옛 체크포인트를 새 계약으로 오독한다 — 갈래는 **데이터가** 말해야 한다
# (2026-08-27 Codex 판정).
CARRIED_CONTRACT_KEY = "carried_contract"
CARRIED_CONTRACT_SCOPED = "scoped_v1"


def resolve_prompt_version(selector: str) -> str:
    try:
        return PROMPT_VERSION_MAP[selector]
    except KeyError:
        raise ValueError(
            f"unknown shot_continuity prompt version selector "
            f"{selector!r} (known: {sorted(PROMPT_VERSION_MAP)})"
        )


def pack_has_people_contract(selector: str) -> bool:
    """그 팩이 carried 를 세 갈래로 내는가."""
    return selector in _PEOPLE_CONTRACT_PACKS


def row_has_people_contract(row: Optional[Dict[str, Any]]) -> bool:
    """**그 값이** 세 갈래로 저작된 것인가 — 판단 근거는 데이터다.

    ★팩 selector 로 판단하면 안 된다. selector 는 지금 도는 코드의 값이고
     체크포인트는 옛 판에서 만들어졌을 수 있다 — 그 둘이 어긋나면 옛
     한 칸을 새 계약처럼 읽어 **엉뚱한 값을 프롬프트에 싣는다**.
    """
    return bool(row) and (
        str((row or {}).get(CARRIED_CONTRACT_KEY) or "")
        == CARRIED_CONTRACT_SCOPED)


class LegacyCarriedContract(AppError):
    """옛 한 칸 carried 를 **새 그림을 사는 자리**에서 만났다.

    ★「옛 그림은 이미 나왔다」와 「옛 값이 앞으로 안 쓰인다」는 다르다
     (2026-08-27 Codex 판정). `scene_image` 나 콘티만 force 하면 옛 한
     칸으로 **새 그림을 다시 산다** — 그러면 1-B 가 그대로 재발한다.
     ★**제공자를 부르기 전에** 멈춘다. 소비하는 자리는 샷 하나를
     세우고 나머지를 계속한다 — 스텝을 통째로 죽이면 유일한 구제책이
     `shot_continuity` force 인데 그것은 하류 CP 를 지워 **에피소드의 모든
     이미지를 다시 산다.**

     ★`strict=False` 로 부르는 **프로덕션 경로는 아직 없다**
     (2026-08-27 자체 리뷰). 「열람은 안 막는다」를 실제로 하려면 그런
     경로를 만들어야 한다 — 지금은 감사 도구와 시험만 그 갈래를 쓴다.

    ★`AppError` 의 하위로 둔다 — 소비하는 자리가 넷이라 각자 감싸면
     한 곳이 빠지고, 그 경로만 사용자에게 다른 모양으로 나간다.
    """

    def __init__(self, message: str) -> None:
        super().__init__(
            code=f"step.contract_violation.{_MODULE}",
            message=message,
            status_code=409,
        )


class _CarriedAborted(Exception):
    """carried 팬아웃 내부 신호 — 다른 묶음이 이미 실패해 **보내지 않았다**.

    밖으로 새지 않는다: 부르는 자리가 삼키고 원래 예외를 올린다.
    """


def char_sids_of(ve_detail: "list") -> "set[str]":
    """VE 행에서 **인물 short_id 집합**을 뽑는다.

    ★`entity_type` 은 그 행이 이미 들고 있다 — 엔티티 지도를 한 번 더
     타지 않는다. 에피소드 링크로 만든 지도를 거치면 링크가 없는 인물이
     조용히 빠진다.
    """
    return {
        str(v.get("short_id"))
        for v in (ve_detail or [])
        if isinstance(v, dict) and v.get("short_id")
        and v.get("entity_type") == "character"
    }


def report_dropped_people(
    tag: str, row: Optional[Dict[str, Any]], visible_short_ids: "set[str]",
) -> None:
    """붙이지 못한 인물을 **로그로 남긴다** (2026-08-27 자체 리뷰).

    ★드롭 자체는 옳다 — 모르는 사람을 넣으면 이 판이 막으려던 것이
     그대로 돌아온다. 문제는 **조용한 것**이다. 셋을 구분할 수 없다:

       ㉠ 이 샷에 정말 사람이 없다
       ㉡ VE 가 아직 안 만들어졌다 (실측: 선택 샷 3,750개 중 124개가 빈다)
       ㉢ 신원이 안 맞는다 — 변형 인물이 대표적이다.
          DB 실측: `민숙`=C02 인데 `민숙 (변형 상태)`=C14 로 **다른 id** 다.
          그 샷 VE 에 변형만 있으면 저작이 낸 원본 이름은 안 맞는다.
          이름에서 접미사를 떼는 것은 글자 판단이라 안 한다 — 관계로
          잇는 것은 별건이다.
    """
    if not row_has_people_contract(row):
        return
    people = [p for p in (row or {}).get("people") or [] if isinstance(p, dict)]
    if not people:
        return
    unresolved = [p.get("character_ko") for p in people
                  if not str(p.get("character_short_id") or "").strip()]
    off = [p.get("character_ko") for p in people
           if str(p.get("character_short_id") or "").strip()
           and str(p.get("character_short_id")) not in visible_short_ids]
    if not (unresolved or off):
        return
    logger.info(
        "carried[%s]: 인물 %d명 중 안 붙인 것 — 신원 미해소 %s · "
        "그 샷 배정 밖 %s (배정=%s)",
        tag, len(people), unresolved or "없음", off or "없음",
        sorted(visible_short_ids) or "비어 있음",
    )


#: CARRIED 인물 문장을 **정본 표기에 결속**하는 계약 (2026-08-29, #104).
#  ★프롬프트 바이트가 바뀌므로 두 소비 경로의 **outer config hash** 에 접는다
#   (`SceneImagePipelineStep`·`ShotContiLightStep`). 샷 지문만 접으면 outer
#   가 clean-skip 해 이 값이 그림에 영영 안 닿는다 — 이 저장소에서 같은
#   자리를 이미 겪었다.
CARRIED_PERSON_PROJECTION_VERSION = "named_v1"


def _named_state(person: Dict[str, Any], state: str) -> str:
    """`state_en` 을 **그 사람의 정본 표기에 묶어** 한 문장으로 만든다.

    ## 왜 (2026-08-29 실측, S2sh6)

    저작은 사람마다 `character_ko`·`character_short_id` 를 **typed 로**
    주고, 이 모듈은 `visible_short_ids` 로 그것을 대조까지 한다. 그런데
    문장을 만들 때 그 id 를 버리고 `state_en` 만 이어 붙였다. 실물:

        objects_en          … The fallen cane remains beside the prone figure.
        people[0] 노인 C03   'He remains on his side … wearing a gray coat …'
        people[1] 민수 C01   'He wears navy coveralls … He kneels …'

    앞 문장이 노인을 세워 놓은 뒤 `He` 가 셋 이어지니 **둘 다 노인으로**
    읽힌다. 그 프롬프트로 그린 재생성본은 민수를 **노인의 회색 코트**로
    그렸다.

    ★**대명사를 고쳐 쓰지 않는다.** 저작 문안은 한 글자도 안 건드리고,
     앞에 **누구인지만** 붙인다 — 재작성은 뜻을 바꿀 수 있고, 이 저장소는
     의미 판단을 코드가 하지 않는다.

    ★이름이 비면 **종전 그대로** 문장만 낸다. 없는 이름을 지어내지 않는다.
    """
    name = str(person.get("character_ko") or "").strip()
    return f"{name}: {state}" if name else state


def carried_text_for_shot(
    row: Optional[Dict[str, Any]],
    *,
    visible_short_ids: "set[str]",
    bg_only: bool,
    strict: bool = True,
) -> str:
    """그 샷에 **실제로 나갈** CARRIED 본문을 만든다.

    조립 끝점이 넷이라(최종 스틸·마커 맵 둘·콘티) 같은 계약을 네 번
    손으로 쓰면 한 곳이 빠진다. **여기 한 곳만** 둔다.

    순서는 `objects → visible people → offscreen effects` 로 고정한다.

    Args:
        visible_short_ids: 그 샷에 **확정 배정된** 인물 short_id.
            ★`ve_ids_for_shot_ex` 의 씬 합집합 fallback(`exact=False`)을
             여기 넣으면 안 된다 — 그 추측은 PEOPLE 절의 배타 조항을
             뗄지 정하는 용도이지 **사람을 넣을 권한이 아니다.**
        bg_only: 무인 샷이면 사람 칸을 통째로 안 붙인다.
        strict: 옛 한 칸 값을 만났을 때 멈출지. 그림을 사는 경로는 참,
            열람·감사 경로는 거짓.
    """
    if not row:
        return ""
    if not row_has_people_contract(row):
        legacy = str(row.get("carried_en") or "").strip()
        if legacy and strict:
            raise LegacyCarriedContract(
                "carried 가 옛 한 칸 계약이다 — 이 값은 그 샷에 없는 "
                "인물을 말할 수 있어 새 그림에 쓸 수 없다(감사 1-B). "
                "`shot_continuity` 를 force 로 다시 돌릴 것."
            )
        return legacy

    parts: List[str] = []
    objects = str(row.get("objects_en") or "").strip()
    if objects:
        parts.append(objects)
    if not bg_only:
        for p in row.get("people") or []:
            if not isinstance(p, dict):
                continue
            sid = str(p.get("character_short_id") or "").strip()
            state = str(p.get("state_en") or "").strip()
            # ★못 맞춘 사람은 **안 넣는다.** 「모르겠으니 넣는다」로 가면
            #  이 판이 막으려던 것이 그대로 돌아온다.
            if sid and state and sid in visible_short_ids:
                parts.append(_named_state(p, state))
    effects = str(row.get("offscreen_effects_en") or "").strip()
    if effects:
        parts.append(effects)
    return " ".join(parts)


def carried_clause_for(
    carried_map: Optional[Dict[str, Any]],
    tag: str,
    pose_fix_row: Optional[Dict[str, Any]],
    *,
    visible_short_ids: "set[str]",
    bg_only: bool,
    strict: bool = True,
) -> str:
    """저장된 연속성에서 그 샷의 CARRIED 본문을 뽑는다.

    **소비하는 자리가 넷이라(최종 스틸·마커 맵 둘·콘티) 여기 한 곳만
    둔다.** 옛 계약의 `pose_fix` 병합까지 포함한다.

    ★옛 계약에서는 `fix.carried_en or carried[tag].carried_en` 로 두 값을
     골랐고, 그래서 **`pose_fix` 가 일부러 비운 칸이 옛 값으로
     되살아났다**(2026-08-27 Codex 지적). 새 계약에서는 `pose_fix` 가
     교정본을 `carried` 안 제자리에 넣으므로 읽을 곳이 한 군데다.
    """
    row = (carried_map or {}).get(tag) or {}
    if not row_has_people_contract(row):
        row = {"carried_en": ((pose_fix_row or {}).get("carried_en")
                              or row.get("carried_en") or "")}
    elif not bg_only:
        # ★네 소비처가 다 이 함수를 타므로 로그도 여기 한 곳이면 된다.
        report_dropped_people(tag, row, visible_short_ids)
    return carried_text_for_shot(
        row, visible_short_ids=visible_short_ids, bg_only=bg_only,
        strict=strict)


#: 전역 저작(pose_canon · carried)의 대기 상한. ★이 둘만 길게 준다.
#:
#: 왜 — 이 둘은 **씬 원문 전문 + 선택 샷 전부**를 한 번에 읽는다. 씬이 늘면
#:  입력도 출력도 같이 커진다. 파청(66씬 · 원문 5만자 · 선택 샷 194)에서
#:  전역 기본값 600초를 네 번 연속 넘겼다(2026-09-08 실측). 아마겟돈(7~9씬)
#:  에서는 안 났다 — 긴 대본에서만 드러나는 자리다.
#:
#: ★전역 `llm_timeout_text` 는 **안 올린다**. 올리면 모든 글 스텝이 오래
#:  기다려, 진짜로 걸린 호출도 그만큼 늦게 드러난다 (Codex 2026-09-09).
#: ★재시도는 0 — 42분짜리 시도를 네 번 되풀이하면 한 번 실패에 세 시간이
#:  날아간다. 한 번 재고, 안 되면 바로 드러나게 한다.
#: ★이것은 **응급 처치다.** 근본은 출력 책임을 나누는 것(입력은 그대로,
#:  샷 묶음마다 그 몫만 출력)이고 그 몫은 후속으로 남긴다.
GLOBAL_AUTHOR_TIMEOUT_SECONDS = 2400
GLOBAL_AUTHOR_RETRIES = 0

#: ★[2026-09-18 컨트리로드] carried 한 답에 몇 샷까지 맡기나.
#:  242샷을 한 답에 쓰게 했더니 **15분 1초 만에 제공자 게이트웨이가 504 로 끊었다**
#:  (우리 상한 2400초에는 닿기 전). 같은 주행의 pose_canon 은 100초 · 입력 71,998 /
#:  출력 4,834 토큰으로 멀쩡했다 — 즉 문제는 입력이 아니라 **한 답에 몰린 출력**이다.
#:  ★입력은 자르지 않는다: 씬 원문 전문과 **전체 시간순 샷 목록**을 묶음마다 그대로
#:  싣는다(팩 규칙 4 의 양방향 소급이 전역 문맥을 요구한다). 나누는 것은 답할 샷뿐이다.
CARRIED_CHUNK_SIZE = 30
#: 묶음을 같이 부르는 수 — entity_detail 과 같은 방식(스레드 풀 + 문맥 bind).
CARRIED_CHUNK_WORKERS = 3
#: ★나눈 묶음에만 주는 재시도 (2026-09-18). 전역 0 은 「42분짜리 한 방을 네 번
#:  되풀이하면 세 시간」이라 둔 값이다 — 묶음은 실측 166~192초라 셈이 다르다.
#:  실제로 묶음 둘이 끝난 뒤 제공자 500 한 번에 **완료분까지 통째로** 날아갔다
#:  (05:13 · InternalServerError). 일시 오류로 전부 잃지 않게 묶음만 다시 보낸다.
CARRIED_CHUNK_RETRIES = 2


def run_shot_continuity(
    *,
    shots: List[Dict[str, Any]],
    scene_texts: Dict[int, str],
    scene_headings: Dict[int, str],
    char_name_to_short: Dict[str, str],
    prompt_version: str = "1",
    call_structured_fn: Optional[Callable[..., Dict[str, Any]]] = None,
    project_config: Optional[Dict[str, Any]] = None,
    opik_metadata: Optional[Dict[str, Any]] = None,
) -> Dict[str, Any]:
    """연속성 3종 저작 → 체크포인트 payload.

    Returns:
        {"pose_canon": [{character_ko, character_short_id, pose_en, shots,
                          basis_ko}],
         "carried": {tag: {carried_en, basis_ko}},
         "pose_fix": {tag: {movement_en, figures_en, carried_en}}}
    """
    if call_structured_fn is None:
        from app.modules.llm.llm_client import call_structured

        call_structured_fn = call_structured

    resolved = resolve_prompt_version(prompt_version)
    # 판정 모델 명시 — 실험 정본=gpt(분석 계열), unknown-step silent
    # fallback(gemini-pro) 차단 (shot_ref_classify 와 동일 E2E 실측 대응)
    project_config = {
        **(project_config or {}),
        f"{_MODULE}_pose_canon": {"model": "gpt"},
        f"{_MODULE}_carried": {"model": "gpt"},
        f"{_MODULE}_pose_fix": {"model": "gpt"},
    }
    ordered = sorted(
        shots, key=lambda s: (int(s["scene_index"]), int(s["shot_index"]))
    )
    tags = [tag_of(int(s["scene_index"]), int(s["shot_index"])) for s in ordered]
    if not tags:
        return {"pose_canon": [], "carried": {}, "pose_fix": {}}
    desc_by_tag = {
        t: (s.get("description") or "") for t, s in zip(tags, ordered)
    }
    scene_ids = sorted({int(s["scene_index"]) for s in ordered})
    # 씬 원문 전문 — 절대 자르지 않는다
    scenes_txt = "\n\n".join(
        f"[씬 {si}] {scene_headings.get(si, '')}\n{scene_texts.get(si, '')}"
        for si in scene_ids
    )
    shots_txt = "\n".join(f"{t}: {desc_by_tag[t]}" for t in tags)

    # ── 1) pose_canon ────────────────────────────────────────────────
    pc_sys = load_prompt(_MODULE, "pose_canon_system", version=resolved)
    pc_schema = load_schema(
        _MODULE, "pose_canon_schema", None, version=resolved
    )
    # ★명단은 **이름을 값으로 받는 모든 산출**에 필요하다 (2026-08-27
    #  Codex BLOCK). v3 carried 도 `character_ko` 로 답하고 코드가 정확
    #  일치로 short_id 를 붙이는데, 명단을 안 보여 주면 표기가 흔들려
    #  매핑이 None 이 되고 **화면 안 인물의 유효한 상태가 조용히
    #  버려진다.** 팩 문안이 「입력에 명단이 제공된다」고 말하는 이상
    #  실제로 줘야 한다 — 안 주면 그 문장이 거짓말이다.
    # ★carried 쪽 명단은 **그 문안이 명단을 말하는 팩**에만 붙인다
    #  (2026-08-27 자체 리뷰). `_NAME_ROSTER_PACKS` 는 pose_canon 기준이라
    #  v2 도 들어 있는데, v2 의 carried 문안은 명단을 한 마디도 안 한다 —
    #  붙이면 모델이 쓸 줄 모르는 블록을 받고 **이미 발행된 v2 의 조립
    #  바이트가 달라진다.**
    carried_roster = ""
    roster_block = ""
    if prompt_version in _NAME_ROSTER_PACKS and char_name_to_short:
        roster_txt = "\n".join(
            f"- {name}" for name in sorted(char_name_to_short)
        )
        roster_block = f"\n\n등장인물 정본 명단:\n{roster_txt}"
        if pack_has_people_contract(prompt_version):
            carried_roster = roster_block

    pc_user = (f"씬 원문 전문:\n\n{scenes_txt}"
               f"\n\n샷 목록(시간순):\n{shots_txt}{roster_block}")
    pc_res = call_structured_fn(
        f"{_MODULE}_pose_canon",
        pc_sys,
        pc_user,
        pc_schema,
        project_config=project_config,
        schema_name=f"{_MODULE}_pose_canon",
        opik_metadata=opik_metadata,
        # ★전역 저작 — 긴 대본에서 기본 600초를 넘긴다. 위 상수 주석 참고.
        timeout=GLOBAL_AUTHOR_TIMEOUT_SECONDS,
        num_retries=GLOBAL_AUTHOR_RETRIES,
    )
    tag_set = set(tags)
    pose_canon: List[Dict[str, Any]] = []
    for it in pc_res.get("immobile") or []:
        known = [t for t in it.get("shots") or [] if t in tag_set]
        unknown = [t for t in it.get("shots") or [] if t not in tag_set]
        if unknown:
            logger.warning(
                "shot_continuity pose_canon: 미지 샷 제거 %s (character=%s)",
                unknown, it.get("character_ko"),
            )
        name = it.get("character_ko") or ""
        pose_canon.append(
            {
                "character_ko": name,
                # 정확 일치만 — 글자 부분매칭 금지 규칙
                "character_short_id": char_name_to_short.get(name),
                "pose_en": it.get("pose_en") or "",
                "shots": known,
                "basis_ko": it.get("basis_ko") or "",
            }
        )

    # ── 2) carried ───────────────────────────────────────────────────
    ca_sys = load_prompt(_MODULE, "carried_system", version=resolved)
    ca_schema = load_schema(_MODULE, "carried_schema", None, version=resolved)
    ca_base = (f"씬 원문 전문:\n\n{scenes_txt}"
               f"\n\n샷 목록(시간순 — 각각 carried state 판정):\n{shots_txt}"
               + carried_roster)
    # ★출력 책임만 묶음으로 나눈다 — 위 CARRIED_CHUNK_SIZE 주석 참고.
    ca_chunks = [tags[i:i + CARRIED_CHUNK_SIZE]
                 for i in range(0, len(tags), CARRIED_CHUNK_SIZE)]

    def _carried_call(chunk: List[str]) -> Dict[str, Any]:
        user = ca_base
        if len(ca_chunks) > 1:
            # 한 묶음이면 조립은 **옛날과 한 글자도 다르지 않다**.
            user += ("\n\n이번 답의 대상 샷 — 이 목록의 샷만 출력한다"
                     "(위 전체 목록은 시간순·소급 판단용 맥락):\n"
                     + "\n".join(f"{t}: {desc_by_tag[t]}" for t in chunk))
        return call_structured_fn(
            f"{_MODULE}_carried",
            ca_sys,
            user,
            ca_schema,
            project_config=project_config,
            schema_name=f"{_MODULE}_carried",
            opik_metadata=opik_metadata,
            # ★전역 저작 — 긴 대본에서 기본 600초를 넘긴다. 위 상수 주석 참고.
            timeout=GLOBAL_AUTHOR_TIMEOUT_SECONDS,
            # 나눈 묶음만 재시도를 준다 — 한 묶음이면 옛 계약(0) 그대로.
            num_retries=(CARRIED_CHUNK_RETRIES if len(ca_chunks) > 1
                         else GLOBAL_AUTHOR_RETRIES),
        )

    _carried_abort = threading.Event()

    def _send_carried_chunk(chunk: List[str]) -> Dict[str, Any]:
        """중단 표가 서 있으면 **보내기 전에** 선다 — 다음 묶음을 더 사지 않는다."""
        if _carried_abort.is_set():
            raise _CarriedAborted()
        try:
            return _carried_call(chunk)
        except BaseException:
            # ★실패한 일꾼이 **직접** 세운다 (Codex BLOCK 2026-09-18). 부모가 볼 때까지
            #  기다리면 빈 일꾼이 다음 묶음을 먼저 집어 **또 산다**.
            _carried_abort.set()
            raise

    ca_got: Dict[str, Any] = {}
    if len(ca_chunks) <= 1:
        for it in _carried_call(tags).get("items") or []:
            ca_got[it.get("shot")] = it
    else:
        from concurrent.futures import ThreadPoolExecutor, as_completed

        from app.core.research_call_budget import bind_current_research_budget
        from app.modules.llm.opik_trace import bind_current_trace
        # ★정지 표·예산·Opik trace 는 스레드마다 따로다 — 부모 스레드에서 감싼다
        #  (본보기: entity_steps.EntityDetailStep, Codex BLOCK 2026-09-17).
        send = bind_current_trace(bind_current_research_budget(_send_carried_chunk))
        with ThreadPoolExecutor(
            max_workers=min(CARRIED_CHUNK_WORKERS, len(ca_chunks))
        ) as pool:
            futs = {pool.submit(send, chunk): chunk for chunk in ca_chunks}
            first_exc: Optional[BaseException] = None
            # ★★**끝나는 순서대로** 본다 (Codex BLOCK 2026-09-18). 제출 순서로 기다리면
            #  뒤 묶음이 먼저 죽어도 못 보고, 빈 일꾼이 다음 묶음을 계속 **산다**.
            #  하나라도 실패하면 곧바로 중단 표를 세워 **보내기 전에** 세운다 —
            #  부모의 cancel 만으로는 일꾼이 다음 일을 먼저 집는 틈이 있다.
            for f in as_completed(futs):
                chunk = futs[f]
                try:
                    res = f.result()
                except _CarriedAborted:
                    continue          # 중단 표를 보고 안 보낸 묶음 — 원래 예외를 지키려 삼킨다
                except BaseException as exc:
                    if first_exc is None:
                        first_exc = exc
                        _carried_abort.set()
                        for g in futs:
                            g.cancel()
                    continue
                own = set(chunk)
                for it in res.get("items") or []:
                    # 묶음은 **제 샷만** 책임진다 — 맥락으로 딸려 온 답은 안 쓴다.
                    if it.get("shot") in own:
                        ca_got[it["shot"]] = it
            if first_exc is not None:
                raise first_exc
    missing = [t for t in tags if t not in ca_got]
    if missing:
        raise AppError(
            code=f"step.contract_violation.{_MODULE}",
            message=f"carried state 누락 샷: {missing}",
            status_code=422,
        )
    if pack_has_people_contract(prompt_version):
        # v3 — 사람을 **이름 달린 항목**으로 받고 코드가 short_id 를 붙인다.
        #  pose_canon 과 같은 계약이다(정확 일치, 글자 부분매칭 금지).
        carried = {}
        unresolved: List[str] = []
        for t in tags:
            people = []
            for p in ca_got[t].get("people") or []:
                if not isinstance(p, dict):
                    continue
                name = str(p.get("character_ko") or "").strip()
                sid = char_name_to_short.get(name)
                if not sid:
                    unresolved.append(f"{t}:{name}")
                people.append({
                    "character_ko": name,
                    # ★못 맞추면 None 을 그대로 남긴다 — **기록은 하되
                    #  주입하지 않는다**(`carried_text_for_shot`).
                    "character_short_id": sid,
                    "state_en": str(p.get("state_en") or "").strip(),
                    "basis_ko": str(p.get("basis_ko") or "").strip(),
                })
            carried[t] = {
                CARRIED_CONTRACT_KEY: CARRIED_CONTRACT_SCOPED,
                "objects_en": str(ca_got[t].get("objects_en") or "").strip(),
                "people": people,
                "offscreen_effects_en": str(
                    ca_got[t].get("offscreen_effects_en") or "").strip(),
                "basis_ko": str(ca_got[t].get("basis_ko") or "").strip(),
            }
        if unresolved:
            logger.warning(
                "shot_continuity carried: 정본 명단에 없는 이름 %d건 — "
                "기록만 하고 프롬프트에 넣지 않는다: %s",
                len(unresolved), unresolved[:20],
            )
    else:
        carried = {
            t: {
                "carried_en": ca_got[t].get("carried_en") or "",
                "basis_ko": ca_got[t].get("basis_ko") or "",
            }
            for t in tags
        }

    # ── 3) pose_fix (정본 샷만) ──────────────────────────────────────
    pf_sys = load_prompt(_MODULE, "pose_fix_system", version=resolved)
    pf_schema = load_schema(
        _MODULE, "pose_fix_schema", None, version=resolved
    )
    pose_fix: Dict[str, Dict[str, str]] = {}
    people_scoped = pack_has_people_contract(prompt_version)
    for it in pose_canon:
        for tag in it["shots"]:
            target: Optional[Dict[str, Any]] = None
            if people_scoped:
                # v3 — **그 인물 한 사람의 state 만** 교정한다. 종전에는
                #  샷 전체의 carried 한 덩이를 넘겨, 정본 인물이 둘인 샷
                #  에서 뒤엣것이 앞엣것을 덮었다.
                target = next(
                    (p for p in carried[tag].get("people") or []
                     if isinstance(p, dict)
                     and p.get("character_ko") == it["character_ko"]), None)
                if target is None:
                    continue     # 그 샷에 이 인물의 이어받는 상태가 없다
                subject = target["state_en"]
            else:
                subject = carried[tag]["carried_en"]
            user = (
                f"자세 정본({it['character_ko']}): {it['pose_en']}\n\n"
                f"샷 텍스트: {desc_by_tag[tag]}\n\n"
                # production v1: movement/figures 저작 부재 — 빈 문자열 전달
                f"movement: \n\n"
                f"figures: \n\n"
                + (f"state: {subject}" if people_scoped
                   else f"carried: {subject}")
            )
            res = call_structured_fn(
                f"{_MODULE}_pose_fix",  # 고정 tag — 모델 명시 매칭
                pf_sys,
                user,
                pf_schema,
                project_config=project_config,
                schema_name=f"{_MODULE}_pose_fix",
                opik_metadata=opik_metadata,
            )
            if target is not None:
                # ★교정본을 **제자리에** 넣는다. 종전에는 조립이
                #  `fix.carried_en or carried.carried_en` 으로 두 값을
                #  골랐는데, 그러면 **일부러 비운 칸이 옛 값으로
                #  되살아났다**(2026-08-27 Codex 지적).
                # ★**빈 값이 저작본을 지우게 두지 않는다** (2026-08-27
                #  자체 리뷰). `call_structured` 는 Tier 1 이 검열에 걸리면
                #  Tier 2 에서 입력을 정화해 다시 부른다 — 폭력 묘사가
                #  씻겨 나가면 모델이 `state_en: ""` 을 낼 수 있고, 키가
                #  있으니 schema 도 통과한다. 그대로 넣으면 그 인물의
                #  이어받는 상태가 **로그 한 줄 없이 사라진다.**
                #
                #  옛 계약에는 `fix.carried_en or base.carried_en` 이라는
                #  보호막이 있었다. 그것을 걷어내면서 이 자리를 만들었다.
                _fixed = str(res.get("state_en") or "").strip()
                if _fixed:
                    target["state_en"] = _fixed
                else:
                    logger.warning(
                        "shot_continuity pose_fix[%s/%s]: 교정본이 비어 "
                        "저작본을 유지한다", tag, it["character_ko"])
                pose_fix[tag] = {
                    "movement_en": res.get("movement_en") or "",
                    "figures_en": res.get("figures_en") or "",
                }
                continue
            pose_fix[tag] = {
                "movement_en": res.get("movement_en") or "",
                "figures_en": res.get("figures_en") or "",
                "carried_en": res.get("carried_en") or "",
            }

    return {"pose_canon": pose_canon, "carried": carried, "pose_fix": pose_fix}