"""LiteLLM + Opik 통합 LLM 클라이언트.

모든 텍스트 LLM 호출을 LiteLLM Router를 통해 수행하고,
Opik callback으로 자동 추적한다.

기존 GeminiTextClient / OpenAIClient / gemini_key_pool / llm_logger를 대체.
이미지 생성(gemini_image_client)은 별도 유지.
"""

import json
import logging
import os
import threading
from pathlib import Path
from typing import Any, Callable, Dict, List, NamedTuple, Optional

import jsonschema
import litellm
from litellm import Router

logger = logging.getLogger(__name__)


def _is_local_schema_validate_enabled() -> bool:
    """local jsonschema 검증 toggle 평가 (problems.md #13, review I1).

    우선순위: ``LLM_LOCAL_SCHEMA_VALIDATE`` ENV → ``settings.llm_local_schema
    _validate`` (Pydantic). 둘 다 미설정 시 default True. provider strict mode
    (LiteLLM ``response_format=json_schema strict=True``) 가 일부 provider 에서
    일부 필드만 enforce 할 수 있으므로 JSON 파싱 후 ``jsonschema.validate`` 로
    한 번 더 shape 검증한다. 운영 중 schema 배포 시 즉시 disable 가능.
    """
    raw = os.environ.get("LLM_LOCAL_SCHEMA_VALIDATE")
    if raw is not None:
        return raw.strip().lower() in ("1", "true", "yes", "on")
    try:
        from app.core.config import settings
        return bool(getattr(settings, "llm_local_schema_validate", True))
    except Exception:
        return True


# structured output 이 tool-call 래퍼를 흘릴 때 최상위에 붙는 키.
# 2026-08-06 실측: Anthropic 경로가 간헐적으로 `{"parameters": {…}}` 로 한 겹
# 감싸 돌려준다(같은 프롬프트 4회 중 1회). 안쪽은 스키마와 정확히 일치한다.
# 벗기지 않으면 `_validate_local_schema` 가 root 에서 실패하고 call_structured
# 가 Tier 2 로 내려가는데, Tier 2 는 **입력을 sanitize 한 다른 프롬프트**로
# 다시 묻는다 — 판정에 조용한 계약 변경이 들어간다(223쌍 재판정에서 92건).
_TOOL_ENVELOPE_KEYS = frozenset({"parameters", "arguments", "input"})


def _unwrap_tool_envelope(payload: Any, schema: Dict[str, Any]) -> Any:
    """tool-call 래퍼 한 겹을 벗긴다 — 스키마가 그 키를 원하지 않을 때만.

    보수적으로만 벗긴다: ①최상위가 dict 이고 키가 **정확히 하나** ②그 키가
    알려진 래퍼 이름 ③스키마의 properties 에 그 키가 **없다** ④안쪽이 dict.
    넷을 다 만족하면 스키마가 그 키를 담을 수 없으므로 래퍼가 확실하다.
    하나라도 어긋나면 원본을 그대로 돌려준다.
    """
    if not isinstance(payload, dict) or len(payload) != 1:
        return payload
    (key,), (inner,) = payload.keys(), payload.values()
    if key not in _TOOL_ENVELOPE_KEYS or not isinstance(inner, dict):
        return payload
    props = schema.get("properties") if isinstance(schema, dict) else None
    if isinstance(props, dict) and key in props:
        return payload
    logger.info("structured output: '%s' 래퍼 한 겹 벗김", key)
    return inner


def _validate_local_schema(
    payload: Dict[str, Any],
    response_schema: Dict[str, Any],
    *,
    step: str,
    schema_name: str,
) -> None:
    """``jsonschema.validate`` wrapper — 실패 시 ``SchemaValidationError`` raise.

    ENV ``LLM_LOCAL_SCHEMA_VALIDATE=false`` 면 no-op. ``response_schema`` 가 빈 dict
    이거나 비-dict 면 검증 skip (legacy/free-form caller 보호). ``ValidationError``
    의 path/message 를 보존하여 Tier 2/3 로직과 logger 가 root cause 진단 가능.
    """
    from app.modules.llm.safety import InvalidSchemaError, SchemaValidationError

    if not _is_local_schema_validate_enabled():
        return
    if not isinstance(response_schema, dict) or not response_schema:
        return
    try:
        jsonschema.validate(instance=payload, schema=response_schema)
    except jsonschema.ValidationError as exc:
        path = "/".join(str(p) for p in exc.absolute_path) or "<root>"
        raise SchemaValidationError(
            f"Local schema validation failed for step={step} schema={schema_name} "
            f"at path={path}: {exc.message}"
        ) from exc
    except jsonschema.SchemaError as exc:
        # caller 가 invalid jsonschema 전달 — fallback 진행해도 동일 결과이므로
        # InvalidSchemaError(ValueError) 로 래핑하여 _NON_SAFETY 분류 + ValueError
        # 호환성 양쪽 보존 (problems.md #13 B2).
        raise InvalidSchemaError(
            f"Invalid jsonschema spec for step={step} (schema_id={schema_name}): "
            f"{exc.message}"
        ) from exc

# ── Thread-local Opik context (StepRunner에서 설정, call_* 에서 자동 병합) ──
_thread_local = threading.local()


def set_opik_context(meta: Optional[Dict] = None) -> None:
    """현재 스레드에 Opik metadata context 설정. call_* 호출 시 자동 병합."""
    _thread_local.opik_meta = meta


def _get_thread_opik_meta() -> Optional[Dict]:
    return getattr(_thread_local, "opik_meta", None)


def _build_opik_metadata(step: str, opik_metadata: Optional[Dict] = None) -> Dict:
    """Opik metadata 구성 — thread-local context 자동 병합.

    우선순위: 명시적 opik_metadata > thread-local context > step tag만
    """
    metadata = {"opik": {"tags": [step]}}

    # 1) thread-local context (StepRunner에서 설정)
    thread_meta = _get_thread_opik_meta()
    if thread_meta:
        extra_tags = thread_meta.get("tags", [])
        merged = {k: v for k, v in thread_meta.items() if k != "tags"}
        metadata["opik"].update(merged)
        if extra_tags:
            metadata["opik"]["tags"].extend(extra_tags)

    # 2) 명시적 파라미터 (덮어쓰기) — 원본 dict 변경 안 함
    if opik_metadata:
        extra_tags = opik_metadata.get("tags", [])
        merged = {k: v for k, v in opik_metadata.items() if k != "tags"}
        metadata["opik"].update(merged)
        if extra_tags:
            metadata["opik"]["tags"].extend(extra_tags)

    return metadata

# ── Opik 콜백 설정 ──
_opik_initialized = False
# 전역 lazy init 보호 (멀티 worker에서 _init_opik/_get_router 동시 호출 방지).
# RLock — _get_router 내부에서 _init_opik 호출 시 재진입 가능해야 함.
_init_lock = threading.RLock()


# 감싼 함수임을 표시하는 이름 — 전역 변수 대신 함수 속성에 남긴다. 모듈이
# 다시 읽히면 전역은 초기화되지만 설치본 모듈에 꽂아 둔 함수는 그대로라,
# 속성으로 봐야 이중 래핑을 실제로 막는다.
_OPIK_CACHE_WRAP_MARK = "_theroad_opik_cache_fields_wrapper"

# 캐시 항목 배선이 실제로 걸렸는가. litellm 구조가 바뀌어 배선이 무력화되면
# 토큰 기록은 계속 남고 캐시 칸만 비어, 겉으로는 "캐시가 안 걸린 주행"과
# 구분이 안 된다. 그래서 상태를 남기고 콜백 켬 로그에 함께 적는다.
_opik_cache_wiring_ok = False


def _cache_fields_of(usage) -> Dict[str, int]:
    """usage 에서 캐시로 재사용된 입력 토큰을 꺼낸다 (없으면 빈 dict).

    두 자리를 본다 — `prompt_tokens_details.cached_tokens` (OpenAI 형식,
    litellm 이 Gemini 의 cachedContentTokenCount 도 여기 채운다) 와
    `cache_read_input_tokens` (Anthropic 형식). 값이 없으면 키를 만들지
    않는다: 없는 것을 0 으로 적으면 적중률 0% 라는 거짓 기록이 남는다.
    """
    def _int_or_none(value):
        if isinstance(value, bool) or not isinstance(value, int):
            return None
        return value

    fields: Dict[str, int] = {}

    details = getattr(usage, "prompt_tokens_details", None)
    cached = None
    if isinstance(details, dict):
        cached = details.get("cached_tokens")
    elif details is not None:
        cached = getattr(details, "cached_tokens", None)
    cached = _int_or_none(cached)
    if cached is not None:
        fields["cached_tokens"] = cached

    read = _int_or_none(getattr(usage, "cache_read_input_tokens", None))
    if read is not None:
        fields["cache_read_input_tokens"] = read

    return fields


def _wrap_opik_usage_object() -> None:
    """litellm 의 Opik 콜백이 버리는 캐시 항목을 usage 에 되살린다.

    `litellm/integrations/opik/utils.py` 의 `create_usage_object` 는 세 값
    (completion/prompt/total)만 담는다. 캐시로 재사용된 입력 토큰은 거기서
    사라져 Opik span 에 남지 않는다 — 값은 제공사가 주는데 기록만 없다.
    그 상태로는 프롬프트 조립 순서를 바꿔도 효과를 잴 수 없다.

    설치본 파일은 고치지 않는다(재설치 때 날아간다). 호출 지점이
    `utils.create_usage_object(...)` 로 모듈 속성을 매번 찾으므로 속성만
    바꿔 두면 걸린다. 원 함수는 `__wrapped__` 로 들고 있고, 표식을 보고
    두 번 감싸지 않는다.

    배선이 안 걸리면 **ERROR 로 남기고** `_opik_cache_wiring_ok` 를 False 로
    둔다. 이 실패는 조용하면 안 된다 — 토큰 기록은 그대로 남고 캐시 칸만
    비어서, 나중에 보면 "캐시가 안 걸린 주행"과 구분이 안 된다. 다만 호출
    자체는 죽이지 않는다(기록 보강이 본 작업을 막지 않는다는 기존 관례).
    """
    global _opik_cache_wiring_ok
    _opik_cache_wiring_ok = False

    try:
        from litellm.integrations.opik import utils as opik_utils
    except Exception as exc:  # litellm 버전이 바뀌어 경로가 없어진 경우
        logger.error(
            "Opik usage 캐시 항목 배선 실패 — litellm opik utils 없음: %s. "
            "캐시 적중은 기록되지 않는다(토큰 세 값은 남는다).", exc)
        return

    original = getattr(opik_utils, "create_usage_object", None)
    if original is None:
        logger.error(
            "Opik usage 캐시 항목 배선 실패 — create_usage_object 없음 "
            "(litellm 구조 변경). 캐시 적중은 기록되지 않는다"
            "(토큰 세 값은 남는다).")
        return
    if getattr(original, _OPIK_CACHE_WRAP_MARK, False):
        _opik_cache_wiring_ok = True
        return  # 이미 감쌌다

    def create_usage_object(usage):
        usage_dict = original(usage)
        try:
            usage_dict.update(_cache_fields_of(usage))
        except Exception as exc:
            # 기록 보강이 호출을 죽이면 안 된다 — 세 값은 그대로 남긴다.
            logger.warning("Opik usage 캐시 항목 추출 실패: %s", exc)
        return usage_dict

    setattr(create_usage_object, _OPIK_CACHE_WRAP_MARK, True)
    create_usage_object.__wrapped__ = original
    try:
        opik_utils.create_usage_object = create_usage_object
    except Exception as exc:
        logger.error(
            "Opik usage 캐시 항목 배선 실패 — 함수 교체 불가: %s. "
            "캐시 적중은 기록되지 않는다(토큰 세 값은 남는다).", exc)
        return
    _opik_cache_wiring_ok = True
    logger.info("Opik usage 에 캐시 항목 기록 켬 (create_usage_object 래핑)")


def _init_opik():
    global _opik_initialized
    if _opik_initialized:
        return
    with _init_lock:
        if _opik_initialized:  # double-checked locking
            return
        from app.core.config import settings
        # 셀프 호스팅(2026-08-14): 키 없이 주소만으로도 켠다 — SDK 는
        # 환경 변수만 읽으므로 .env 값을 여기서 올린다.
        if settings.opik_api_key or settings.opik_url_override:
            if settings.opik_api_key:
                os.environ["OPIK_API_KEY"] = settings.opik_api_key
            if settings.opik_url_override:
                os.environ["OPIK_URL_OVERRIDE"] = settings.opik_url_override
            os.environ["OPIK_WORKSPACE"] = settings.opik_workspace
            os.environ["OPIK_PROJECT_NAME"] = settings.opik_project_name
            # 콜백을 켜기 전에 감싼다 — 켠 뒤에 감싸면 그 사이 호출이
            # 캐시 항목 없이 기록된다.
            _wrap_opik_usage_object()
            litellm.callbacks = ["opik"]
            logger.info(
                "Opik callback enabled (project: %s, url: %s, cache_wiring=%s)",
                settings.opik_project_name,
                settings.opik_url_override or "cloud",
                "ok" if _opik_cache_wiring_ok else "failed")
        _opik_initialized = True


# ── LiteLLM Router 초기화 ──
_router: Optional[Router] = None


class _RouterBinding(NamedTuple):
    """Router 와 **그것을 지을 때 쓴 슬롯**을 하나로 묶는다.

    [2026-08-01 A5 후속, Codex 재확인 BLOCKING] Router 는 어느 슬롯 키로
    지어졌는지 기록하지 않았고, ``_completion`` 은 호출 직전 전역 활성 슬롯을
    다시 읽어 그것을 자기 슬롯으로 삼았다. 그래서 이런 창이 열린다:

        ① 한 요청이 primary 키로 지은 Router 를 이미 들고 있다
        ② 다른 요청이 전역을 secondary 로 옮긴다
        ③ ①이 그 stale Router 로 호출한다 → 실제로는 primary 로 나간다
        ④ primary billing 실패가 "secondary 실패"로 보고된다
        ⑤ 마지막 슬롯 소진 판정 — **보조 키를 한 번도 안 써 보고 죽는다**

    슬롯과 Router 를 짝으로 다루면 ③에서 실패한 슬롯을 정확히 지목한다.
    """

    slot: Optional[str]
    router: Router


_binding: Optional[_RouterBinding] = None


def _load_gemini_keys() -> List[str]:
    """backend/.env 에서 GEMINI_API_KEY* 수집."""
    env_file = Path(__file__).resolve().parent.parent.parent.parent / ".env"
    keys = []
    if not env_file.exists():
        return keys
    for line in env_file.read_text(encoding="utf-8").splitlines():
        line = line.strip()
        if not line or line.startswith("#") or "=" not in line:
            continue
        k, _, v = line.partition("=")
        k = k.strip()
        v = v.strip().strip('"').strip("'")
        if k.startswith("GEMINI_API_KEY") and v:
            if v not in keys:
                keys.append(v)
    return keys


def _completion(binding: "_RouterBinding", model: str, kwargs: Dict[str, Any]):
    """Router 호출 — OpenAI **키 수준** 실패면 다음 키 슬롯으로 재시도.

    전환 대상은 billing hard limit / quota 소진 / 401·403 뿐이다. 그 외
    실패(타임아웃·5xx·단순 429·스키마 위반)는 그대로 올려 기존 재시도·
    Tier fallback 경로가 처리하게 둔다. 전환이 일어나면 Router 가 무효화
    되므로 `_get_router_binding()` 로 새 키가 박힌 짝을 다시 받는다.

    ★``binding`` 을 받는다 (2026-08-01 A5 후속) — 전역 활성 슬롯을 호출 직전에
    다시 읽으면, stale Router 를 든 요청이 **다른 요청이 옮겨 놓은 슬롯**을
    자기 것으로 착각한다. 그러면 primary 실패가 secondary 실패로 보고되어
    보조 키를 한 번도 안 써 보고 죽는다. 실패한 슬롯은 그 Router 를 지은
    슬롯이지, 지금 전역이 가리키는 슬롯이 아니다.
    """
    from app.core import openai_keys

    attempts = max(1, openai_keys.slot_count())
    last: Optional[BaseException] = None
    for _ in range(attempts):
        try:
            return binding.router.completion(**kwargs)
        except Exception as exc:  # noqa: BLE001
            last = exc
            if not openai_keys.failover_on(
                exc, where=f"router.completion[{model}]",
                attempted_slot=binding.slot,
            ):
                raise
            binding = _get_router_binding()
    assert last is not None
    raise last


def _invalidate_router() -> None:
    """다음 호출에서 Router 를 다시 짓게 한다.

    OpenAI 키 슬롯이 바뀌면 deployment 에 박힌 api_key 가 옛 키라 그대로
    두면 전환이 무의미하다. 대입은 GIL 하에서 원자적이고, 실제 재빌드는
    `_get_router_binding` 의 double-checked locking 이 처리한다 — 여기서 `_init_lock`
    을 잡으면 호출 중 전환과 엮여 교착이 될 수 있다.
    """
    global _router, _binding
    _router = None
    _binding = None
    logger.info("LiteLLM Router 무효화 — OpenAI 키 슬롯 전환 반영")


def _get_router_binding() -> "_RouterBinding":
    """Router 와 그 생성 슬롯을 **짝으로** 받는다.

    ★캐시 판정을 "객체가 있는가"가 아니라 **"그 Router 를 지은 슬롯이 아직
    현재 슬롯인가"** 로 한다. 전환 훅(`_invalidate_router`)은 `_active_index`
    가 바뀐 **뒤 락 밖에서** 돌기 때문에, 그 사이에 다른 스레드가 옛 Router 를
    그대로 받아 갈 수 있다. 슬롯을 대조하면 그 창이 닫힌다.
    """
    from app.core import openai_keys

    b = _binding
    if b is not None and b.slot == openai_keys.active_slot():
        return b

    with _init_lock:
        b = _binding
        if b is not None and b.slot == openai_keys.active_slot():
            return b  # double-checked locking
        _init_opik()
        _build_router()
        built = _binding
        if built is None:
            # 계약 위반이다 — 조용히 넘기면 슬롯 없는 Router 가 돌아다닌다.
            raise RuntimeError("LiteLLM Router binding 이 생성되지 않았다")
        return built


def router_completion(*, model: str, **kwargs: Any):
    """Router 를 거치는 **유일한** 호출 경로 — 키 슬롯 전환이 항상 걸린다.

    [2026-08-01 A5 마무리] 이전에는 `_get_router()` 로 원시 Router 를 꺼내
    `.completion(...)` 을 직접 부르는 소비자가 넷 있었고, 그 경로에서는 1차 키가
    billing 으로 죽어도 **보조 키를 한 번도 안 써 보고** 예외가 그대로 올라갔다
    (직접 재현: active_slot 이 primary 그대로, Router 빌드 1회).

    "원시 Router 를 꺼내지 마라"는 규칙으로 두면 또 생긴다 — 실제로 네 번
    생겼다. 그래서 `_get_router()` 자체를 없애고 이 함수만 남긴다.
    """
    return _completion(_get_router_binding(), model, {"model": model, **kwargs})


def _build_router() -> Router:
    """_get_router의 lock 내부 빌드 로직 — _init_lock를 이미 잡고 있어야 한다."""
    global _router, _binding

    from app.core.config import settings

    model_list = []

    # Gemini 키 풀 — 모든 Gemini 텍스트 모델에 동일 키 풀 적용
    gemini_keys = _load_gemini_keys()
    if not gemini_keys and settings.gemini_api_key:
        gemini_keys = [settings.gemini_api_key]

    gemini_models = [
        ("gemini-pro", settings.gemini_text_model),
        ("gemini-flash", settings.gemini_flash_model),
        ("gemini-lite", settings.gemini_lite_model),
        # gpt-mini: 2026-07-11 Gemini 원복(사용자 goal — 시각 저작 열화 실측).
        ("gpt-mini", settings.gemini_flash_model),
    ]

    for alias, model_id in gemini_models:
        for i, key in enumerate(gemini_keys):
            model_list.append({
                "model_name": alias,
                "litellm_params": {
                    "model": f"gemini/{model_id}",
                    "api_key": key,
                },
                "model_info": {"id": f"{alias}-key{i}"},
            })

    # OpenAI — GPT 5.5 / 5.5 mini / 5.5 nano + 레거시 4.1
    # 주의: gpt-5.5* 는 litellm 내장 model registry에 없을 수 있어
    # provider를 자동 추론하지 못한다. `openai/` 접두사를 명시해 라우팅 강제.
    # OpenAI 키는 슬롯 브로커가 정한다 (2026-07-30) — 1차 키가 키 수준
    # 실패를 내면 보조 슬롯으로 전환되고, 그때 Router 를 무효화해 새 키로
    # 다시 짓는다. 슬롯이 하나뿐이면 기존과 동일하다.
    from app.core import openai_keys

    openai_keys.register_switch_hook(_invalidate_router)
    # ★슬롯과 키를 한 번에 읽는다 (A5) — 따로 읽으면 그사이 전환이 둘을
    # 어긋나게 해 "primary 라고 적힌 secondary 키 Router" 가 생긴다.
    openai_slot, openai_key = openai_keys.active_slot_and_key()
    if openai_key:
        # gpt-5* family는 temperature=1만 허용 — drop_params=True를 deployment에 직접
        # 박아 호출 측 temperature를 자동 drop. 글로벌 litellm.drop_params만으로는
        # router 호출 경로에서 일관되게 적용되지 않는 버전이 있어 deployment level에서 강제.
        _OPENAI_DROP = {"drop_params": True}
        model_list.append({
            "model_name": "gpt",
            "litellm_params": {
                "model": f"openai/{settings.openai_model}",
                "api_key": openai_key,
                **_OPENAI_DROP,
            },
        })
        # 2026-07-11 Gemini 원복: gpt-mini 는 위 gemini_models(flash 매핑)로
        # 복귀. gpt-terra/gpt-luna alias 는 override 안전망으로 유지.
        for alt_model, alt_alias in [
            ("gpt-5.6-terra", "gpt-terra"),
            ("gpt-5.6-sol", "gpt-luna"),
            ("gpt-5.4-nano", "gpt-nano"),
            ("gpt-4.1", "gpt-4.1"),
            ("gpt-4.1-mini", "gpt-4.1-mini"),
        ]:
            model_list.append({
                "model_name": alt_alias,
                "litellm_params": {
                    "model": f"openai/{alt_model}",
                    "api_key": openai_key,
                    **_OPENAI_DROP,
                },
            })

    # Anthropic — 후보 선정 판정 전용 (2026-08-06 사용자 지시).
    # 키가 없으면 alias 를 등록하지 않는다 → 선정 판정은 `gemini-pro` 로
    # 남는다(fail-open). 등록될 때만 `claude-opus` 가 실재한다.
    if settings.anthropic_api_key:
        for _alias, _mdl in (
            ("claude-opus", settings.anthropic_judge_model),
            # 비교 실험용 — 판정 배선에는 쓰지 않는다.
            ("claude-fable", settings.anthropic_fable_model),
        ):
            model_list.append({
                "model_name": _alias,
                "litellm_params": {
                    "model": f"anthropic/{_mdl}",
                    "api_key": settings.anthropic_api_key,
                    # Claude 5 계열은 sampling 파라미터를 거부(400)한다 —
                    # `_NO_TEMPERATURE_ALIASES` 와 이중 방어.
                    "drop_params": True,
                },
            })

    # gpt-5* family는 temperature=1만 허용 등 모델별 spec 차이가 있다.
    # 글로벌 drop_params=True 설정으로 unsupported param을 자동 drop해
    # 호출 측 코드를 단순화. (Router.__init__ 키워드는 미지원이라 글로벌로.)
    import litellm as _litellm
    _litellm.drop_params = True

    _router = Router(
        model_list=model_list,
        routing_strategy="simple-shuffle",
        num_retries=settings.llm_max_retries,
        timeout=settings.llm_timeout_text,
        retry_after=5,  # 429 후 5초 대기
    )

    # 지은 슬롯을 Router 와 묶어 둔다 — 이후 실패는 이 슬롯의 실패다.
    _binding = _RouterBinding(slot=openai_slot, router=_router)

    logger.info(
        "LiteLLM Router initialized: %d deployments (%d Gemini keys, "
        "OpenAI=%s slot=%s/%d)",
        len(model_list), len(gemini_keys), bool(openai_key),
        openai_slot or "없음", openai_keys.slot_count(),
    )

    return _router


# ── 파이프라인 단계 → 모델 매핑 (problems.md #5 통합) ──
#
# Single source of truth: ``app.core.step_manifest.STEP_MANIFEST``.
# 이 파일의 ``PIPELINE_STEPS`` 는 manifest 의 default_model/provider/category/
# label 을 derived view 로 build 한 후, manifest 에 없는 sub_step / v2 legacy
# extension 을 합친다. 이전에는 manifest 와 별도 dict 가 drift 가능했고
# (``planning_doc_analysis`` 의 alias mismatch), runtime 은 PIPELINE_STEPS 만
# 봐서 manifest 의 default_model 필드가 dead 였다.
#
# Extension table (manifest 미등록):
#   - scene_image_pipeline 의 sub_step 8종 (prompt_translation 등)
#   - v2 legacy step 7종 (entity_extract / entity_style / scene_dependency /
#     location_consistency / outlook_merge / webbook_gen / style_rules /
#     entity_detail_batch)
# 이들은 step_runner 흐름과 별개로 LLM call routing 이 필요한 sub-routine.

# manifest 미등록 step 의 model alias 매핑 (extension table).
_PIPELINE_STEP_EXTENSIONS: Dict[str, Dict[str, str]] = {
    # scene_image_pipeline sub_steps
    "prompt_translation":   {"label": "T2I 프롬프트 번역",   "default": "gpt-mini",      "category": "image_sub"},
    "scene_t2i_gen":        {"label": "T2I 이미지 생성",     "default": "gemini-image",  "category": "image_sub"},
    "scene_t2i_validation": {"label": "이미지 검증",         "default": "gpt",           "category": "image_sub"},
    "prompt_sanitize":      {"label": "프롬프트 안전화",      "default": "gpt",           "category": "image_sub"},
    "angle_recommend":      {"label": "앵글 추천",           "default": "gpt",           "category": "image_sub"},
    "fal_angle_apply":      {"label": "fal.ai 앵글 적용",    "default": "fal-ai",        "category": "image_sub"},
    "final_select":         {"label": "최종 선택",           "default": "gpt",           "category": "image_sub"},
    "t2i_translation":      {"label": "T2I 편집 번역",       "default": "gpt-mini",      "category": "image_sub"},

    # G3.2 judge — scene_detail post-parse owned validation
    # round 4 MINOR 1 / round 5 M2: 등록 키는 'default' (NOT 'default_model').
    # 1-call no retry (round 4 Q1=B / round 5 M1) — caller `run_owned_judge` 가
    # violations 발견 시 contract_violation status 마킹만 수행.
    "scene_detail_owned_judge":
        {"label": "owned 객체 redraw 검사", "default": "gpt-mini", "category": "analysis_sub"},

    # FINDING 7 (e2e-bughunt-v1) — owned redraw violation 1-call repair.
    # owned judge 와 동일 tier (analysis_sub). caller `_attempt_owned_redraw_repair`
    # 가 max 1 attempt 관리 + 결과를 기존 validator 로 재검증.
    "scene_detail_owned_repair":
        {"label": "owned 객체 redraw 수정", "default": "gpt-mini", "category": "analysis_sub"},

    # W-M (2026-07-03) — 야외 같은 장소 그룹 plate 생성 순서 결정(단계화 체인).
    # caller = background_render 그룹 체인 lane. 검증(순열+depends_on)/재시도/
    # fallback 은 location_aerial.decide_group_plate_order 가 담당.
    "background_plate_order":
        {"label": "야외 plate 생성 순서", "default": "gpt-mini", "category": "image_sub"},

    # v2 legacy (기존 코드 호환용 — manifest 제외, runtime 라벨 유지)
    "entity_extract":       {"label": "요소 추출 (3턴)",      "default": "gpt",           "category": "analysis"},
    "entity_style":         {"label": "요소 추출 — 스타일+이름", "default": "gpt",        "category": "analysis"},
    "entity_detail_batch":  {"label": "요소 추출 — 상세",     "default": "gpt",           "category": "analysis"},
    "webbook_gen":          {"label": "웹북 패키지 생성",     "default": "gpt",           "category": "auxiliary"},
    "style_rules":          {"label": "스타일 규칙 생성",      "default": "gpt",           "category": "analysis"},
    "outlook_merge":        {"label": "아웃룩 병합",         "default": "gpt",           "category": "analysis"},
    "location_consistency": {"label": "Location 외형 고정",   "default": "gemini-pro",    "category": "analysis"},
}


def _build_pipeline_steps() -> Dict[str, Dict[str, str]]:
    """STEP_MANIFEST + extension 을 합쳐 ``PIPELINE_STEPS`` view 를 build.

    manifest 의 ``default_model`` / ``provider`` / ``category`` / ``label`` 을
    꺼내 기존 PIPELINE_STEPS schema (``label`` / ``default`` / ``category``) 로
    변환. 동일 step 이 양쪽에 있으면 manifest 우선 (problems.md #5: manifest
    is single source).
    """
    from app.core.step_manifest import STEP_MANIFEST

    out: Dict[str, Dict[str, str]] = {}
    for sid, info in STEP_MANIFEST.items():
        out[sid] = {
            "label": info.get("label", sid),
            "default": info.get("default_model", "gemini-pro"),
            "category": info.get("category", "analysis"),
        }
    # extension 은 manifest 에 없는 step 만 추가 (manifest 가 single source).
    # overlap 시 logger.warning + extension 무시 — 회귀 가드 보강 (review M1).
    for sid, info in _PIPELINE_STEP_EXTENSIONS.items():
        if sid in out:
            logger.warning(
                "_PIPELINE_STEP_EXTENSIONS overlap with STEP_MANIFEST: %s — "
                "manifest wins (extension entry is dead and should be removed).",
                sid,
            )
            continue
        out[sid] = dict(info)
    return out


PIPELINE_STEPS = _build_pipeline_steps()

# 사용 가능한 모델 별칭 (UI용)
AVAILABLE_MODELS = [
    {"alias": "gpt",          "label": "GPT-5.6 Sol",          "provider": "openai"},
    {"alias": "gpt-terra",    "label": "GPT-5.6 Terra",        "provider": "openai"},
    {"alias": "gpt-luna",     "label": "GPT-5.6 Sol (구 Luna)", "provider": "openai"},
    {"alias": "gpt-mini",     "label": "Gemini 3.5 Flash",     "provider": "gemini"},
    {"alias": "gpt-nano",     "label": "GPT-5.4 Nano",         "provider": "openai"},
    {"alias": "gemini-pro",   "label": "Gemini 3.1 Pro",       "provider": "gemini"},
    {"alias": "gemini-flash", "label": "Gemini 3.5 Flash",     "provider": "gemini"},
    {"alias": "gemini-lite",  "label": "Gemini 3.5 Flash",     "provider": "gemini"},
    {"alias": "gpt-4.1",      "label": "GPT-4.1",              "provider": "openai"},
    {"alias": "gpt-4.1-mini", "label": "GPT-4.1 Mini",         "provider": "openai"},
    # ANTHROPIC_API_KEY 가 있을 때만 Router 에 실재한다 (_build_router).
    {"alias": "claude-opus",  "label": "Claude Opus 5",        "provider": "anthropic"},
    {"alias": "claude-fable", "label": "Claude Fable 5",       "provider": "anthropic"},
]


_UNKNOWN_STEP_WARNED: set[str] = set()


def _resolve_model(step: str, project_config: Optional[Dict] = None) -> str:
    """단계 + 프로젝트 설정 → LiteLLM Router 모델 별칭 반환.

    PIPELINE_STEPS (manifest+extension) 에 없는 step 은 ``gemini-pro`` 로 silent
    fallback. typo 등을 surface 하기 위해 process 당 1회 logger.warning emit
    (review I2). project_config 가 명시한 step 은 fallback 대상 아님.
    """
    if project_config and step in project_config:
        return project_config[step].get("model", PIPELINE_STEPS.get(step, {}).get("default", "gemini-pro"))
    info = PIPELINE_STEPS.get(step)
    if info is None and step not in _UNKNOWN_STEP_WARNED:
        _UNKNOWN_STEP_WARNED.add(step)
        logger.warning(
            "_resolve_model: unknown step '%s' — falling back to 'gemini-pro'. "
            "If intentional, register in STEP_MANIFEST or _PIPELINE_STEP_EXTENSIONS.",
            step,
        )
    return (info or {}).get("default", "gemini-pro")


# temperature를 강제로 제거해야 하는 모델 alias.
# - gpt-5* family: temperature=1만 허용 (그 외 값은 OpenAI/litellm이 거부)
# - gemini-3 family (pro/flash/lite): temperature<1.0이면 "infinite loops, degraded reasoning,
#   failure on complex tasks" warning. 빈 응답으로 schema 검증 실패하는 사례 발생
#   (예: scene_camera_flow S25에서 gemini-pro가 empty content 반환).
#   LiteLLM 경고 그대로 따라 temperature=1.0 default 사용.
# - claude-opus (Claude Opus 5): `temperature`/`top_p`/`top_k` 를 **거부**한다
#   (400 invalid_request_error). call_structured 의 기본 temperature=0.2 가
#   그대로 나가면 전 호출이 실패하므로 제거 대상 필수.
_NO_TEMPERATURE_ALIASES = frozenset({
    "gpt", "gpt-terra", "gpt-luna", "gpt-mini", "gpt-nano",
    "gemini-pro", "gemini-flash", "gemini-lite",
    "claude-opus", "claude-fable",
})  # gpt-mini 는 Gemini 매핑이지만 temperature 제거 대상 유지 (양쪽 무해)


def _sanitize_kwargs_for_model(model: str, kwargs: Dict[str, Any]) -> None:
    """모델별로 전달 불가 파라미터를 in-place 제거.

    drop_params 글로벌/deployment 설정이 일관되게 적용되지 않는 버전이 있어 명시 제거.
    """
    if model in _NO_TEMPERATURE_ALIASES:
        kwargs.pop("temperature", None)


# Gemini 모델 alias 집합 — safety_settings 강제 적용 대상.
# gpt-mini: 2026-07-11 Gemini 원복 — flash 매핑 복귀로 재편입.
_GEMINI_ALIASES = frozenset({"gemini-pro", "gemini-flash", "gemini-lite", "gpt-mini"})

# Gemini safety filter BLOCK_NONE (전 카테고리) — 픽션 시나리오(영화/드라마) 분석용.
# 시나리오는 폭력/사망/성적 묘사 등 fictional 서사 요소를 포함하므로
# 기본 safety threshold(BLOCK_MEDIUM_AND_ABOVE)에서 content_filter trip 빈발.
# entity_character_list (PID 0bb48ebf, 2026-05-01) 5회 연속 finish_reason='content_filter' 사례.
_GEMINI_SAFETY_SETTINGS_OFF = [
    {"category": "HARM_CATEGORY_HARASSMENT",        "threshold": "BLOCK_NONE"},
    {"category": "HARM_CATEGORY_HATE_SPEECH",       "threshold": "BLOCK_NONE"},
    {"category": "HARM_CATEGORY_SEXUALLY_EXPLICIT", "threshold": "BLOCK_NONE"},
    {"category": "HARM_CATEGORY_DANGEROUS_CONTENT", "threshold": "BLOCK_NONE"},
]


def _apply_gemini_safety(model: str, kwargs: Dict[str, Any]) -> None:
    """Gemini 모델 호출에 safety_settings BLOCK_NONE 강제 — 픽션 콘텐츠 분석 차단 회피."""
    if model in _GEMINI_ALIASES:
        kwargs["safety_settings"] = _GEMINI_SAFETY_SETTINGS_OFF


# OpenAI 계열 model alias (strict structured output 호출 대상).
# 2026-07-11 Gemini 원복: gpt-mini 제외(Gemini flash 매핑 복귀).
_OPENAI_ALIASES = frozenset({"gpt", "gpt-terra", "gpt-luna", "gpt-nano"})

# OpenAI strict structured output (response_format json_schema) 이 거부하는
# jsonschema keyword. FINDING 10: scene_detail detail_schema 의
# t2i_variations[].reference_phrase_kinds.uniqueItems 가 GPT fallback 경로에서
# BadRequestError("'uniqueItems' is not permitted") 를 유발. scope = uniqueItems 한정.
_OPENAI_UNSUPPORTED_SCHEMA_KEYS = frozenset({"uniqueItems"})


def _strip_schema_keys(node: Any, keys: frozenset) -> Any:
    """node 트리의 모든 중첩 dict 에서 keys 를 제거한 deep copy 반환.

    원본 node 는 mutate 하지 않는다 — dict/list 를 새로 재구성한다.
    """
    if isinstance(node, dict):
        return {
            k: _strip_schema_keys(v, keys)
            for k, v in node.items()
            if k not in keys
        }
    if isinstance(node, list):
        return [_strip_schema_keys(v, keys) for v in node]
    return node


def _is_openai_strict_compatible(node: Any) -> bool:
    """OpenAI strict 의 known object-contract 호환성 재귀 검사 (완전 판정 아님).

    검사 범위는 이번에 실측된 object 계약(required=전 property 키 +
    additionalProperties:false)에 한정 — strict 의 전체 keyword subset 을
    판정하지 않는다 (uniqueItems 는 별도 strip). properties={} 빈 object
    검사는 후속 hardening 항목 (Codex 비차단 권고 2026-07-10).

    strict=True 는 모든 object 노드에 대해 ①`required` 가 properties 의 전
    키를 포함하고 ②`additionalProperties: false` 명시를 요구한다. Gemini
    시절 저작 팩 스키마(optional 필드 관용)는 이 규칙을 어겨 BadRequestError
    ("'required' is required to be supplied and to be an array including
    every key in properties") 가 난다 — GPT-5.6 이관 2회차 E2E shot_extract
    30/30 실측 (2026-07-10).
    """
    if isinstance(node, dict):
        props = node.get("properties")
        if isinstance(props, dict) and props:
            req = node.get("required")
            if not isinstance(req, list) or set(req) != set(props.keys()):
                return False
            if node.get("additionalProperties") is not False:
                return False
        return all(_is_openai_strict_compatible(v) for v in node.values())
    if isinstance(node, list):
        return all(_is_openai_strict_compatible(v) for v in node)
    return True


_STRICT_DOWNGRADE_LOGGED: set[str] = set()


def _sanitize_response_format_for_model(model: str, kwargs: Dict[str, Any]) -> None:
    """OpenAI 계열 model 호출 시 response_format schema 의 OpenAI 비호환
    jsonschema keyword 를 제거 (FINDING 10 — provider-boundary fix).

    OpenAI strict structured output 은 `uniqueItems` 를 거부한다. Gemini 경로는
    허용하므로 정상 동작하지만 GPT fallback 경로에서 BadRequestError 가 난다.
    `kwargs["response_format"]["json_schema"]["schema"]` 를 deep-copy + strip 한
    새 객체로 교체한다 — caller 의 원본 response_schema 는 mutate 하지 않으므로
    `_validate_local_schema` 의 local 검증은 원본 schema(uniqueItems 포함)로
    그대로 수행된다. 비-OpenAI(Gemini 등) model 은 no-op.

    GPT-5.6 이관 (2026-07-10): strict 비호환 스키마(Gemini 팩 유래 — optional
    필드/additionalProperties 미명시)는 strict=False 로 강등한다. 스키마
    준수는 `_validate_local_schema` + call_structured retry 가 전 tier 에서
    이미 보증(Gemini 경로와 동일한 enforcement 모델). strict 호환 스키마
    (기존 gpt 스텝 팩)는 strict=True 그대로 — byte-identical. 팩 수정 없이
    provider boundary 에서 해소 (프롬프트 팩 덮어쓰기 금지 준수).
    """
    if model not in _OPENAI_ALIASES:
        return
    response_format = kwargs.get("response_format")
    if not isinstance(response_format, dict):
        return
    json_schema = response_format.get("json_schema")
    if not isinstance(json_schema, dict) or "schema" not in json_schema:
        return
    json_schema["schema"] = _strip_schema_keys(
        json_schema["schema"], _OPENAI_UNSUPPORTED_SCHEMA_KEYS,
    )
    if json_schema.get("strict") and not _is_openai_strict_compatible(
            json_schema["schema"]):
        json_schema["strict"] = False
        name = str(json_schema.get("name", "?"))
        if name not in _STRICT_DOWNGRADE_LOGGED:
            _STRICT_DOWNGRADE_LOGGED.add(name)
            logger.info(
                "response_format '%s': OpenAI strict 비호환 스키마(Gemini 팩 "
                "유래) — strict=False 강등, 준수는 local validation+retry 가 "
                "보증 (process 당 1회 로그)", name,
            )


# ── 통합 호출 함수 ──

def _add_fallback_tag(metadata: Dict[str, Any], tag: str) -> Dict[str, Any]:
    """Opik metadata에 fallback tag(sanitized/gpt_fallback)를 추가한 새 dict 반환.

    원본 dict 변경 없음. opik.tags가 없으면 새로 만든다.
    """
    new_meta = dict(metadata) if metadata else {}
    opik_meta = dict(new_meta.get("opik") or {})
    tags = list(opik_meta.get("tags") or [])
    if tag not in tags:
        tags.append(tag)
    opik_meta["tags"] = tags
    new_meta["opik"] = opik_meta
    return new_meta


def call_structured(
    step: str,
    system_prompt: str,
    user_prompt: "str | list",
    response_schema: Dict[str, Any],
    project_config: Optional[Dict] = None,
    schema_name: str = "response",
    opik_metadata: Optional[Dict] = None,
    temperature: float = 0.2,
    max_tokens: Optional[int] = None,
    *,
    enable_fallback: bool = True,
    validate_response: Optional[Callable[[Dict[str, Any]], bool]] = None,
) -> Dict[str, Any]:
    """Structured JSON output 호출 — LiteLLM Router 경유.

    모든 provider에 대해 response_format으로 JSON schema 강제.
    LiteLLM이 provider별 변환 자동 처리.
    user_prompt: str 또는 multimodal content list (PDF/이미지 포함 시).

    3-tier fallback (enable_fallback=True 시 자동, default):
      Tier 1: 기본 모델 (gemini-pro 등 step 기본).
      Tier 2: sanitize_for_safety + SAFETY_SYSTEM_SUFFIX prepend.
      Tier 3: GPT 강제 (project_config[step] = {"model": "gpt"}).

    enable_fallback=False (legacy 호환): 1차 호출만 수행, 실패 시 즉시 raise.

    `validate_response` (P2-3): Tier 1/2 응답이 valid JSON이지만 의미상 빈 결과
    (예: 빈 list)일 때 caller가 fallback을 강제 트리거할 수 있는 callback.
    callback이 False 반환 → `EmptySemanticResponseError` raise → safety 분류 →
    Tier 2/3 진행. None이면 schema 검증만 통과해도 즉시 반환 (기본 동작).
    이 callback 은 Tier 3에는 적용 안 함 (마지막 시도 보호).

    Local jsonschema 검증 (problems.md #13): ``_do_call`` 안에서 ``jsonschema.
    validate`` 가 **모든 tier 에서** 동작하여 provider strict mode 의 enforcement
    약화를 보완한다. ``validate_response`` callback (의미적 빈 결과) 와 별개의
    레이어 — Tier 3 도 schema 위반 시 ``SchemaValidationError`` 가 caller 까지
    전파된다. ENV ``LLM_LOCAL_SCHEMA_VALIDATE=false`` 로 즉시 disable 가능.

    fallback 트리거 분류 (`is_safety_related_error`):
      - 콘텐츠 안전/모더레이션 신호 (PROHIBITED/content_filter/empty/...): 진행
      - rate-limit/timeout/auth 등 명시적 transient: 즉시 raise (비용 보호)
    """
    from app.modules.llm.safety import (
        SAFETY_SYSTEM_SUFFIX,
        EmptySemanticResponseError,
        is_safety_related_error,
        sanitize_for_safety,
    )

    def _do_call(
        sys_p: str,
        user_p: "str | list",
        cfg: Optional[Dict],
        suffix_tag: str = "",
    ) -> Dict[str, Any]:
        binding = _get_router_binding()
        model = _resolve_model(step, cfg)
        base_metadata = _build_opik_metadata(step, opik_metadata)
        metadata = _add_fallback_tag(base_metadata, suffix_tag) if suffix_tag else base_metadata

        kwargs: Dict[str, Any] = {
            "model": model,
            "messages": [
                {"role": "system", "content": sys_p},
                {"role": "user", "content": user_p},
            ],
            "response_format": {
                "type": "json_schema",
                "json_schema": {
                    "name": schema_name,
                    "schema": response_schema,
                    "strict": True,
                },
            },
            "temperature": temperature,
            "metadata": metadata,
        }
        from app.core.config import settings
        kwargs["max_tokens"] = max_tokens if max_tokens is not None else settings.llm_max_output_tokens

        _sanitize_kwargs_for_model(model, kwargs)
        _sanitize_response_format_for_model(model, kwargs)
        _apply_gemini_safety(model, kwargs)
        response = _completion(binding, model, kwargs)

        content = response.choices[0].message.content
        if not content:
            raise RuntimeError(f"LLM returned empty response for step={step}, model={model}")

        payload = _unwrap_tool_envelope(json.loads(content), response_schema)
        _validate_local_schema(
            payload, response_schema, step=step, schema_name=schema_name,
        )
        return payload

    def _validate_or_raise(result: Dict[str, Any], tier_label: str) -> Dict[str, Any]:
        """validate_response callback 적용. False 반환 시 EmptySemanticResponseError raise."""
        if validate_response is not None and not validate_response(result):
            raise EmptySemanticResponseError(
                f"validate_response failed for step={step} tier={tier_label} — "
                f"semantic empty result"
            )
        return result

    # Tier 1: 기본
    try:
        result = _do_call(system_prompt, user_prompt, project_config)
        return _validate_or_raise(result, "tier1")
    except Exception as exc_t1:
        if not enable_fallback or not is_safety_related_error(exc_t1):
            raise
        logger.warning(
            "call_structured[%s] Tier 1 failed (%s), trying sanitized input",
            step, exc_t1,
        )

    # Tier 2: sanitize + 영화 프레이밍 system suffix
    sanitized_user = sanitize_for_safety(user_prompt)
    safe_system = system_prompt + SAFETY_SYSTEM_SUFFIX
    try:
        result = _do_call(safe_system, sanitized_user, project_config, suffix_tag="sanitized")
        return _validate_or_raise(result, "tier2")
    except Exception as exc_t2:
        logger.warning(
            "call_structured[%s] Tier 2 sanitized failed (%s), trying GPT fallback",
            step, exc_t2,
        )

    # Tier 3: GPT fallback (Gemini → GPT, sanitize 유지)
    # validate_response (의미적 빈 결과 callback) 는 Tier 3에 적용하지 않음 —
    # 마지막 시도이므로 결과를 보존. 단, _do_call 안의 local jsonschema validation
    # 은 Tier 3 에서도 동작하므로 schema 위반은 SchemaValidationError 로 raise.
    gpt_config = dict(project_config) if project_config else {}
    gpt_config[step] = {"model": "gpt"}
    return _do_call(safe_system, sanitized_user, gpt_config, suffix_tag="gpt_fallback")


def call_text(
    step: str,
    system_prompt: str,
    user_prompt: str,
    project_config: Optional[Dict] = None,
    opik_metadata: Optional[Dict] = None,
    temperature: float = 0.2,
    *,
    enable_fallback: bool = True,
) -> str:
    """Free-text 호출 — LiteLLM Router 경유.

    3-tier fallback (call_structured와 동일 정책)을 자동 적용한다.

    `enable_fallback=False` (legacy 호환):
      - Tier 1만 수행. 빈 응답은 `""` 반환 (raise 안 함) — pre-fallback 시기 동작 보존.
      - 그 외 예외는 그대로 raise.

    `enable_fallback=True` (default):
      - 빈 응답을 RuntimeError로 변환 → safety 분류 → Tier 2/3 진행.
      - rate-limit/timeout 등 transient는 즉시 raise (비용 보호).
    """
    from app.modules.llm.safety import (
        SAFETY_SYSTEM_SUFFIX,
        is_safety_related_error,
        sanitize_for_safety,
    )

    def _do_call(sys_p: str, user_p: str, cfg: Optional[Dict], suffix_tag: str = "") -> str:
        """완료된 응답 content 반환. 빈 응답이라도 raise 없이 그대로 (caller가 처리)."""
        binding = _get_router_binding()
        model = _resolve_model(step, cfg)
        base_metadata = _build_opik_metadata(step, opik_metadata)
        metadata = _add_fallback_tag(base_metadata, suffix_tag) if suffix_tag else base_metadata

        from app.core.config import settings
        text_kwargs: Dict[str, Any] = {
            "model": model,
            "messages": [
                {"role": "system", "content": sys_p},
                {"role": "user", "content": user_p},
            ],
            "temperature": temperature,
            "max_tokens": settings.llm_max_output_tokens,
            "metadata": metadata,
        }
        _sanitize_kwargs_for_model(model, text_kwargs)
        _apply_gemini_safety(model, text_kwargs)
        response = _completion(binding, model, text_kwargs)
        return response.choices[0].message.content or ""

    # Tier 1: 기본
    try:
        content = _do_call(system_prompt, user_prompt, project_config)
    except Exception as exc_t1:
        if not enable_fallback or not is_safety_related_error(exc_t1):
            raise
        logger.warning(
            "call_text[%s] Tier 1 raised (%s), trying sanitized input",
            step, exc_t1,
        )
    else:
        if content:
            return content
        # 빈 응답: enable_fallback=False면 legacy 동작 (빈 문자열) 보존.
        if not enable_fallback:
            return ""
        # enable_fallback=True: Tier 2 진행 (safety 가능성)
        logger.warning(
            "call_text[%s] Tier 1 returned empty, trying sanitized input",
            step,
        )

    # Tier 2: sanitize + 영화 프레이밍 system suffix
    sanitized_user = sanitize_for_safety(user_prompt)
    safe_system = system_prompt + SAFETY_SYSTEM_SUFFIX
    try:
        content = _do_call(safe_system, sanitized_user, project_config, suffix_tag="sanitized")
    except Exception as exc_t2:
        logger.warning(
            "call_text[%s] Tier 2 sanitized failed (%s), trying GPT fallback",
            step, exc_t2,
        )
    else:
        if content:
            return content
        logger.warning(
            "call_text[%s] Tier 2 returned empty, trying GPT fallback",
            step,
        )

    # Tier 3: GPT fallback
    gpt_config = dict(project_config) if project_config else {}
    gpt_config[step] = {"model": "gpt"}
    content = _do_call(safe_system, sanitized_user, gpt_config, suffix_tag="gpt_fallback")
    if not content:
        # 마지막 시도까지 빈 응답 — 명시적으로 raise (caller가 인지)
        raise RuntimeError(
            f"LLM returned empty text response for step={step} after all 3 tiers"
        )
    return content


def call_multiturn(
    step: str,
    messages: List[Dict[str, str]],
    response_schema: Optional[Dict[str, Any]] = None,
    project_config: Optional[Dict] = None,
    opik_metadata: Optional[Dict] = None,
    schema_name: str = "response",
    temperature: float = 0.2,
    *,
    enable_fallback: bool = True,
) -> Any:
    """멀티턴 대화 호출 — entity_extractor Turn 0→1 등에 사용.

    messages: [{"role": "system", "content": ...}, {"role": "user", "content": ...}, ...]
    response_schema: 있으면 structured, 없으면 free text.

    3-tier fallback (call_structured와 동일 정책)을 자동 적용한다.
    Tier 2/3에서 system 메시지에 SAFETY_SYSTEM_SUFFIX append +
    user/assistant content 일괄 sanitize.

    Tier 1 예외 분류 (`is_safety_related_error`):
      - 콘텐츠 안전 신호: Tier 2/3 진행
      - 명시적 transient (timeout/rate-limit 등): 즉시 raise (비용 보호)
    """
    from app.modules.llm.safety import (
        SAFETY_SYSTEM_SUFFIX,
        is_safety_related_error,
        sanitize_messages,
    )

    def _do_call(msgs: List[Dict[str, str]], cfg: Optional[Dict], suffix_tag: str = "") -> Any:
        binding = _get_router_binding()
        model = _resolve_model(step, cfg)
        base_metadata = _build_opik_metadata(step, opik_metadata)
        metadata = _add_fallback_tag(base_metadata, suffix_tag) if suffix_tag else base_metadata

        kwargs: Dict[str, Any] = {
            "model": model,
            "messages": msgs,
            "temperature": temperature,
            "metadata": metadata,
        }
        _sanitize_kwargs_for_model(model, kwargs)
        _apply_gemini_safety(model, kwargs)

        if response_schema:
            kwargs["response_format"] = {
                "type": "json_schema",
                "json_schema": {
                    "name": schema_name,
                    "schema": response_schema,
                    "strict": True,
                },
            }
            _sanitize_response_format_for_model(model, kwargs)

        response = _completion(binding, model, kwargs)
        content = response.choices[0].message.content or ""

        if response_schema:
            if not content:
                raise RuntimeError(
                    f"LLM returned empty multiturn response for step={step}, model={model}"
                )
            payload = _unwrap_tool_envelope(
                json.loads(content), response_schema)
            _validate_local_schema(
                payload, response_schema, step=step, schema_name=schema_name,
            )
            return payload
        if not content:
            raise RuntimeError(
                f"LLM returned empty multiturn text for step={step}, model={model}"
            )
        return content

    try:
        return _do_call(messages, project_config)
    except Exception as exc_t1:
        if not enable_fallback or not is_safety_related_error(exc_t1):
            raise
        logger.warning(
            "call_multiturn[%s] Tier 1 failed (%s), trying sanitized input",
            step, exc_t1,
        )

    # Tier 2: system 메시지 첫 항목에 SAFETY_SYSTEM_SUFFIX append + content 전체 sanitize.
    safe_messages = sanitize_messages(messages)
    if safe_messages and isinstance(safe_messages[0], dict) and safe_messages[0].get("role") == "system":
        first = dict(safe_messages[0])
        sys_content = first.get("content", "")
        if isinstance(sys_content, str):
            first["content"] = sys_content + SAFETY_SYSTEM_SUFFIX
        safe_messages = [first] + safe_messages[1:]

    try:
        return _do_call(safe_messages, project_config, suffix_tag="sanitized")
    except Exception as exc_t2:
        logger.warning(
            "call_multiturn[%s] Tier 2 sanitized failed (%s), trying GPT fallback",
            step, exc_t2,
        )

    gpt_config = dict(project_config) if project_config else {}
    gpt_config[step] = {"model": "gpt"}
    return _do_call(safe_messages, gpt_config, suffix_tag="gpt_fallback")