"""
Vignette 엔진 게이트웨이 — 로컬 claude -p(Opus 4.8) 상주 멀티턴 풀.

컨테이너 밖(호스트)에서 실행한다. api 컨테이너가 ENGINE_URL(예: http://host.docker.internal:9099)로 HTTP 호출.
회기당 1 EngineSession = claude -p 1 프로세스 상주. 페르소나 시스템프롬프트를 고정하고 발화마다 stdin 주입.
턴 간 컨텍스트 유지 + prompt caching 재사용(2026-06-25 실증, reference_claude_p_resident_engine).

실행:
  cd apps/api && uvicorn engine_gateway.gateway:app --host 0.0.0.0 --port 9099
"""
import asyncio
import json
import os
import time
import uuid
from dataclasses import dataclass
from typing import Any, Optional

from fastapi import FastAPI, HTTPException
from fastapi.responses import JSONResponse, StreamingResponse
from pydantic import BaseModel

from app.contracts.engine_gateway import (
    AIRole,
    ENGINE_GATEWAY_SSE_DONE,
    ENGINE_GATEWAY_SSE_ERROR,
    ENGINE_GATEWAY_SSE_TOKEN,
    EngineCapabilitiesResponse,
    EngineMessage as GwMessage,
    EngineProvider,
    GenerateResponse,
    GenerateRequest as GwGenerateReq,
    ReasoningEffort,
    StreamDoneEvent,
    StreamErrorEvent,
    StreamTokenEvent,
    normalize_engine_gateway_model,
    sse_frame,
)
from engine_gateway.provider_registry import (
    ProviderError,
    discover_capabilities,
    generate_with_provider,
    stream_with_provider,
)

CLAUDE_BIN = os.environ.get("CLAUDE_BIN", "claude")
DEFAULT_MODEL = os.environ.get("ENGINE_MODEL", "")  # 비우면 CLI 기본(Opus 4.8)
FALLBACK_MODEL = os.environ.get("ENGINE_FALLBACK_MODEL", "")
DEFAULT_BUDGET = float(os.environ.get("SESSION_BUDGET_USD", "5.0"))
READY_TTL_SECONDS = float(os.environ.get("ENGINE_READY_TTL_SECONDS", "30"))
READY_TIMEOUT_SECONDS = float(os.environ.get("ENGINE_READY_TIMEOUT_SECONDS", "20"))
READY_BUDGET_USD = float(os.environ.get("ENGINE_READY_BUDGET_USD", "0.5"))
# 단발 생성(/v1/generate) 턴 타임아웃 — 페르소나 초안 생성 같은 대형 구조화 출력은
# 120초를 넘길 수 있어 설정 가능하게 한다(2026-07-15). 호출부(app ENGINE_TIMEOUT)와 정합 필요.
GENERATE_TURN_TIMEOUT_SECONDS = float(os.environ.get("ENGINE_GENERATE_TIMEOUT_SECONDS", "300"))
SESSION_IDLE_TTL_SECONDS = float(os.environ.get("ENGINE_SESSION_IDLE_TTL_SECONDS", "3600"))
MAX_RESIDENT_SESSIONS = max(1, int(os.environ.get("ENGINE_MAX_RESIDENT_SESSIONS", "24")))
GATEWAY_PROVIDER = "claude_cli"
GATEWAY_FALLBACK_MODEL_NAME = "claude-opus-4-8"


@dataclass(frozen=True, slots=True)
class GatewayPromptParts:
    system_prompt: str
    user_payload: str
    current_user_payload: str


BASE_ARGS = [
    "-p",
    "--input-format", "stream-json",
    "--output-format", "stream-json",
    "--include-partial-messages",
    "--verbose",
    # 상담 축어록은 게이트웨이 프로세스 수명 안에서만 유지한다. Claude CLI의 로컬
    # 세션 파일로 이중 저장하지 않아 개인정보 노출과 매 턴 디스크 I/O를 줄인다.
    "--no-session-persistence",
    "--dangerously-skip-permissions",
    # 페르소나 격리: cwd/env/git status/메모리(CLAUDE.md) 등 per-machine 섹션을 시스템프롬프트에서
    # 제거 → 내담자 AI가 자신이 개발 환경(Claude Code/Vignette repo) 안에 있음을 알아채 캐릭터를
    # 깨는 것을 차단. (시스템프롬프트는 아래에서 --system-prompt 로 페르소나만 '교체' 주입.)
    "--exclude-dynamic-system-prompt-sections",
]

class EngineSession:
    """claude -p 상주 프로세스 1개 = 상담 회기 1개."""

    def __init__(
        self,
        system_prompt: str | None = None,
        budget: float = DEFAULT_BUDGET,
        model: str | None = None,
        reasoning_effort: ReasoningEffort | None = None,
    ):
        self.id = uuid.uuid4().hex
        self.system_prompt = system_prompt
        self.budget = budget
        self.model = normalize_engine_gateway_model(model)
        self.reasoning_effort = reasoning_effort
        self.proc: asyncio.subprocess.Process | None = None
        self.lock = asyncio.Lock()  # 한 회기 안의 턴은 직렬(상담 왕복)
        self.cost_usd = 0.0
        self.turns = 0
        self.last_used_at = time.monotonic()

    async def start(self) -> None:
        args = [CLAUDE_BIN, *BASE_ARGS, "--max-budget-usd", str(self.budget)]
        model = self.model or DEFAULT_MODEL
        if model:
            args += ["--model", model]
        if self.reasoning_effort:
            args += ["--effort", self.reasoning_effort]
        if FALLBACK_MODEL:
            args += ["--fallback-model", FALLBACK_MODEL]
        if self.system_prompt:
            # APPEND(기본 코딩 어시스턴트 프롬프트에 덧붙임) 대신 REPLACE → 페르소나가 유일한
            # 정체성. 기본 Claude Code 시스템프롬프트가 남으면 내담자가 어시스턴트로 작동/캐릭터 붕괴.
            args += ["--system-prompt", self.system_prompt]
        self.proc = await asyncio.create_subprocess_exec(
            *args,
            stdin=asyncio.subprocess.PIPE,
            stdout=asyncio.subprocess.PIPE,
            stderr=asyncio.subprocess.PIPE,
        )

    async def turn(self, content: str, timeout: float = 120.0) -> dict:
        if self.proc is None or self.proc.returncode is not None:
            raise RuntimeError("engine session not running")
        async with self.lock:
            msg = json.dumps(
                {"type": "user", "message": {"role": "user", "content": content}},
                ensure_ascii=False,
            )
            self.proc.stdin.write((msg + "\n").encode("utf-8"))
            await self.proc.stdin.drain()

            text_parts: list[str] = []

            async def _read_until_result() -> dict:
                # 한 턴: system(init/hook) 라인 무시 → assistant 텍스트 누적 → result 에서 종료
                while True:
                    line = await self.proc.stdout.readline()
                    if not line:
                        return {}
                    try:
                        obj = json.loads(line)
                    except json.JSONDecodeError:
                        continue
                    t = obj.get("type")
                    if t == "assistant":
                        for c in obj.get("message", {}).get("content", []):
                            if c.get("type") == "text":
                                text_parts.append(c["text"])
                    elif t == "result":
                        return obj

            result = await asyncio.wait_for(_read_until_result(), timeout=timeout)
            self.last_used_at = time.monotonic()
            self.cost_usd = result.get("total_cost_usd", self.cost_usd)
            self.turns += 1
            error_detail = (
                result.get("error")
                or result.get("result")
                or (result.get("errors") or [None])[0]
                or "".join(text_parts)
                or None
            )
            return {
                "text": "".join(text_parts),
                "cost_usd": self.cost_usd,
                "turns": self.turns,
                "is_error": result.get("is_error", False),
                "error": error_detail,
            }

    async def turn_stream(self, content: str, timeout: float = 600.0):
        """한 턴을 *증분 텍스트 청크*로 흘려보낸다 (SSE /v1/stream 용).

        기존 turn() 은 result 까지 모아 한 번에 text 를 반환한다. 스트림 경로는
        assistant 텍스트 델타가 도착하는 즉시 yield 해야 토큰 스트리밍 UX 가 산다.
        마지막에 {"__done__": ...} 메타(누적 cost/turns)를 한 번 yield 한다.

        주의: claude -p stream-json 의 assistant 메시지는 청크가 전체 누적본일 수도,
        델타일 수도 있어 게이트웨이가 *이미 보낸 접두 길이*를 추적해 신규분만 흘린다.
        """
        if self.proc is None or self.proc.returncode is not None:
            raise RuntimeError("engine session not running")
        async with self.lock:
            msg = json.dumps(
                {"type": "user", "message": {"role": "user", "content": content}},
                ensure_ascii=False,
            )
            self.proc.stdin.write((msg + "\n").encode("utf-8"))
            await self.proc.stdin.drain()

            emitted = ""  # 이미 흘려보낸 누적 텍스트 (델타 산출 기준)

            async def _next_line():
                return await asyncio.wait_for(self.proc.stdout.readline(), timeout=timeout)

            while True:
                line = await _next_line()
                if not line:
                    break
                try:
                    obj = json.loads(line)
                except json.JSONDecodeError:
                    continue
                t = obj.get("type")
                if t == "stream_event":
                    stream_event = obj.get("event") or {}
                    delta_payload = stream_event.get("delta") or {}
                    if (
                        stream_event.get("type") == "content_block_delta"
                        and delta_payload.get("type") == "text_delta"
                    ):
                        delta = str(delta_payload.get("text") or "")
                        if delta:
                            emitted += delta
                            yield {"type": "delta", "text": delta}
                elif t == "assistant":
                    # 이번 메시지의 텍스트 전체를 재구성
                    full = "".join(
                        c.get("text", "")
                        for c in obj.get("message", {}).get("content", [])
                        if c.get("type") == "text"
                    )
                    if full.startswith(emitted):
                        delta = full[len(emitted):]
                    else:
                        # 누적이 아니라 메시지 분절 → 통째로 추가분
                        delta = full
                    if delta:
                        emitted += delta if full.startswith(emitted) else full
                        yield {"type": "delta", "text": delta}
                elif t == "result":
                    self.last_used_at = time.monotonic()
                    self.cost_usd = obj.get("total_cost_usd", self.cost_usd)
                    self.turns += 1
                    error_detail = (
                        obj.get("error")
                        or obj.get("result")
                        or (obj.get("errors") or [None])[0]
                        or emitted
                        or None
                    )
                    yield {
                        "type": "done",
                        "text": emitted,
                        "cost_usd": self.cost_usd,
                        "turns": self.turns,
                        "is_error": obj.get("is_error", False),
                        "error": error_detail,
                    }
                    return

    async def close(self) -> None:
        if self.proc and self.proc.returncode is None:
            try:
                self.proc.stdin.close()
            except Exception:
                pass
            try:
                await asyncio.wait_for(self.proc.wait(), timeout=5)
            except Exception:
                self.proc.kill()


SESSIONS: dict[str, EngineSession] = {}
_SESSION_RESOLVE_LOCK = asyncio.Lock()
_READY_CACHE: dict[tuple[str, str, str], dict[str, Any]] = {}
_READY_LOCK = asyncio.Lock()
app = FastAPI(title="Vignette Engine Gateway")


class CreateReq(BaseModel):
    system_prompt: str | None = None
    budget_usd: float | None = None


class TurnReq(BaseModel):
    content: str


@app.get("/health")
async def health():
    return {
        "ok": True,
        "engine": "claude_cli",
        "model": DEFAULT_MODEL or "gateway-default",
        "sessions": len(SESSIONS),
    }


def _ready_response(
    entry: dict[str, Any],
    *,
    provider: EngineProvider,
    model: str | None,
    reasoning_effort: ReasoningEffort | None,
    cached: bool,
    age_seconds: float = 0.0,
) -> JSONResponse:
    ok = bool(entry.get("ok"))
    return JSONResponse(
        {
            "ok": ok,
            "engine": provider,
            "model": model or (
                DEFAULT_MODEL or "default(opus-4-8)"
                if provider == "claude_cli"
                else "provider-default"
            ),
            "reasoning_effort": reasoning_effort,
            "sessions": len(SESSIONS),
            "detail": entry.get("detail"),
            "age_seconds": round(max(0.0, age_seconds), 3),
            "cached": cached,
        },
        status_code=200 if ok else 503,
    )


@app.get("/ready")
async def ready(
    force: bool = False,
    provider: EngineProvider = "claude_cli",
    model: str | None = None,
    reasoning_effort: ReasoningEffort | None = None,
):
    """Prove that claude -p can complete a real generation.

    /health is shallow process liveness. This endpoint catches the installed-but-
    not-authenticated CLI state before a learner reaches POST /sessions/:id/turn.
    """
    cache_key = (provider, model or "", reasoning_effort or "")
    entry = _READY_CACHE.get(
        cache_key, {"checked_at": 0.0, "ok": False, "detail": "not checked"}
    )
    age = time.monotonic() - float(entry.get("checked_at", 0.0) or 0.0)
    if not force and age < READY_TTL_SECONDS:
        return _ready_response(
            entry,
            provider=provider,
            model=model,
            reasoning_effort=reasoning_effort,
            cached=True,
            age_seconds=age,
        )

    async with _READY_LOCK:
        entry = _READY_CACHE.get(
            cache_key, {"checked_at": 0.0, "ok": False, "detail": "not checked"}
        )
        age = time.monotonic() - float(entry.get("checked_at", 0.0) or 0.0)
        if not force and age < READY_TTL_SECONDS:
            return _ready_response(
                entry,
                provider=provider,
                model=model,
                reasoning_effort=reasoning_effort,
                cached=True,
                age_seconds=age,
            )

        ok = False
        detail = "unknown readiness failure"
        if provider == "claude_cli":
            probe = EngineSession(
                system_prompt="You are a readiness probe. Reply with exactly OK.",
                budget=READY_BUDGET_USD,
                model=model,
                reasoning_effort=reasoning_effort,
            )
            try:
                await probe.start()
                result = await probe.turn(
                    "Reply with exactly OK.", timeout=READY_TIMEOUT_SECONDS
                )
                if result.get("is_error"):
                    detail = str(result.get("error") or "engine returned an error")
                else:
                    text = str(result.get("text") or "").strip()
                    ok = bool(text)
                    detail = text or "empty engine response"
            except Exception as exc:
                detail = str(exc)
            finally:
                await probe.close()
        else:
            request = GwGenerateReq(
                provider=provider,
                model=model,
                reasoning_effort=reasoning_effort,
                max_tokens=16,
                temperature=0,
                messages=[GwMessage(role="user", content="Reply with exactly OK.")],
            )
            try:
                result = await generate_with_provider(
                    request,
                    system_prompt="You are a readiness probe. Reply with exactly OK.",
                    user_payload="Reply with exactly OK.",
                )
                ok = bool(result.text.strip())
                detail = result.text.strip() or "empty engine response"
            except Exception as exc:
                detail = str(exc)

        entry = {"checked_at": time.monotonic(), "ok": ok, "detail": detail}
        _READY_CACHE[cache_key] = entry
        return _ready_response(
            entry,
            provider=provider,
            model=model,
            reasoning_effort=reasoning_effort,
            cached=False,
        )


@app.get("/v1/capabilities", response_model=EngineCapabilitiesResponse)
async def v1_capabilities(provider: EngineProvider, force: bool = False):
    return await discover_capabilities(provider, force=force)


@app.post("/session")
async def create_session(req: CreateReq):
    s = EngineSession(system_prompt=req.system_prompt, budget=req.budget_usd or DEFAULT_BUDGET)
    await s.start()
    SESSIONS[s.id] = s
    return {"session_id": s.id}


@app.post("/session/{sid}/turn")
async def turn(sid: str, req: TurnReq):
    s = SESSIONS.get(sid)
    if not s:
        raise HTTPException(404, "session not found")
    return await s.turn(req.content)


@app.delete("/session/{sid}")
async def close_session(sid: str):
    s = SESSIONS.pop(sid, None)
    if s:
        await s.close()
    return {"closed": bool(s)}


# ════════════════════════════════════════════════════════════════════════════
# stateless 어댑터 — apps/api/app/engine_client.py 의 계약과 정합
#   POST /v1/generate  : EngineMessage[] → 1턴 생성 → GenerateResponse
#   POST /v1/stream    : 같은 입력 → SSE(text/event-stream) 토큰 스트림
# 상주 풀(/session 계열)은 그대로 유지하고, 단발/스트림은 아래가 흡수한다.
# session_id 가 오면 풀을 재사용해 멀티턴 prompt caching 이점을 살린다.
# ════════════════════════════════════════════════════════════════════════════

def _split_messages(messages: list[GwMessage], *, ai_role: AIRole | None = None) -> GatewayPromptParts:
    """EngineMessage[] → named prompt parts for the current gateway turn."""
    system_parts: list[str] = []
    turn_control_parts: list[str] = []
    non_system: list[GwMessage] = []
    for m in messages:
        if m.role == "system":
            if ai_role == "client" and not m.cache:
                turn_control_parts.append(m.content)
            else:
                system_parts.append(m.content)
        else:
            non_system.append(m)

    last_user_index: int | None = None
    for index, m in enumerate(non_system):
        if m.role == "user":
            last_user_index = index

    last_user = ""
    if last_user_index is not None:
        last_user = non_system[last_user_index].content

    user_payload = last_user
    current_user_payload = last_user
    if ai_role == "client" and last_user_index is not None:
        control = "\n\n".join(p for p in turn_control_parts if p.strip())
        current_sections: list[str] = []
        if control:
            current_sections.append("[현재 턴 상태와 연기 지시]\n" + control)
        current_sections.append("[이번 상담자 발화]\n" + last_user)
        current_user_payload = "\n\n".join(current_sections)

        history_parts: list[str] = []
        for m in non_system[:last_user_index]:
            content = m.content.strip()
            if not content:
                continue
            speaker = "상담자" if m.role == "user" else "내담자"
            history_parts.append(f"{speaker}: {content}")
        history_sections = list(current_sections[:-1])
        if history_parts:
            history_sections.append("[직전 대화]\n" + "\n".join(history_parts))
        history_sections.append(current_sections[-1])
        user_payload = "\n\n".join(history_sections)

    system_prompt = "\n\n".join(p for p in system_parts if p.strip())
    return GatewayPromptParts(
        system_prompt=system_prompt,
        user_payload=user_payload,
        current_user_payload=current_user_payload,
    )


def _inject_schema(system_prompt: str, schema: Optional[dict[str, Any]]) -> str:
    """structured_schema 가 오면 system 프롬프트에 JSON 스키마 준수 지시를 주입.

    게이트웨이 단 strict 강제는 후속(claude -p 는 response_format 미지원). 현실적 차선.
    """
    if not schema:
        return system_prompt
    directive = (
        "\n\n[출력 형식 — 반드시 준수]\n"
        "아래 JSON 스키마에 *정확히* 맞는 단일 JSON 객체만 출력한다. "
        "설명·마크다운 코드펜스 없이 JSON 본문만.\n"
        f"{json.dumps(schema, ensure_ascii=False)}"
    )
    return (system_prompt + directive) if system_prompt else directive.lstrip()


def _response_model_name(session: EngineSession) -> str:
    return session.model or DEFAULT_MODEL or GATEWAY_FALLBACK_MODEL_NAME


async def _resolve_session(req: GwGenerateReq, system_prompt: str) -> tuple[EngineSession, bool]:
    """내담자 회기는 session_id 에 바인딩하고, 나머지는 단발 세션으로 실행한다.

    반환: (session, ephemeral). ephemeral=True 면 호출부가 응답 후 close 한다.
    """
    requested_model = normalize_engine_gateway_model(req.model)
    persistent_key = req.session_id if req.session_id and req.ai_role == "client" else None
    if persistent_key:
        async with _SESSION_RESOLVE_LOCK:
            await _prune_resident_sessions(exclude={persistent_key})
            existing = SESSIONS.get(persistent_key)
            if existing is not None:
                running = existing.proc is not None and existing.proc.returncode is None
                same_model = requested_model is None or (existing.model or DEFAULT_MODEL) == requested_model
                same_effort = req.reasoning_effort is None or existing.reasoning_effort == req.reasoning_effort
                if running and same_model and same_effort:
                    existing.last_used_at = time.monotonic()
                    return existing, False
                SESSIONS.pop(persistent_key, None)
                await existing.close()

            session = EngineSession(
                system_prompt=system_prompt or None,
                budget=DEFAULT_BUDGET,
                model=requested_model,
                reasoning_effort=req.reasoning_effort,
            )
            await session.start()
            SESSIONS[persistent_key] = session
            return session, False

    # 평가·관리자 생성처럼 페르소나 회기와 정체성을 섞으면 안 되는 호출은 1회성 세션이다.
    s = EngineSession(
        system_prompt=system_prompt or None,
        budget=DEFAULT_BUDGET,
        model=requested_model,
        reasoning_effort=req.reasoning_effort,
    )
    await s.start()
    return s, True


async def _prune_resident_sessions(*, exclude: set[str] | None = None) -> None:
    """죽었거나 오래 유휴인 회기와 상한 초과 회기를 안전하게 정리한다."""
    protected = exclude or set()
    now = time.monotonic()
    stale_keys = [
        key
        for key, session in SESSIONS.items()
        if key not in protected
        and not session.lock.locked()
        and (
            session.proc is None
            or session.proc.returncode is not None
            or now - session.last_used_at >= SESSION_IDLE_TTL_SECONDS
        )
    ]
    for key in stale_keys:
        session = SESSIONS.pop(key, None)
        if session is not None:
            await session.close()

    overflow = len(SESSIONS) - MAX_RESIDENT_SESSIONS + 1
    if overflow <= 0:
        return
    candidates = sorted(
        (
            (key, session)
            for key, session in SESSIONS.items()
            if key not in protected and not session.lock.locked()
        ),
        key=lambda item: item[1].last_used_at,
    )
    for key, session in candidates[:overflow]:
        SESSIONS.pop(key, None)
        await session.close()


def _session_turn_payload(session: EngineSession, prompt_parts: GatewayPromptParts) -> str:
    """상주 프로세스는 자체 대화기록을 가지므로 재사용 턴에는 L6를 중복 주입하지 않는다."""
    if session.turns > 0:
        return prompt_parts.current_user_payload
    return prompt_parts.user_payload


@app.post("/v1/generate")
async def v1_generate(req: GwGenerateReq):
    """단발 생성 (평가 deep-loop, 회기종료 압축 등). GenerateResponse 호환 dict 반환."""
    prompt_parts = _split_messages(req.messages, ai_role=req.ai_role)
    system_prompt = _inject_schema(prompt_parts.system_prompt, req.structured_schema)
    if not prompt_parts.user_payload:
        raise HTTPException(400, "no user message in payload")

    provider = req.provider or "claude_cli"
    if provider != "claude_cli":
        try:
            result = await generate_with_provider(
                req,
                system_prompt=system_prompt,
                user_payload=prompt_parts.user_payload,
            )
        except ProviderError as exc:
            raise HTTPException(502, f"engine provider error: {exc}") from exc
        return GenerateResponse(
            text=result.text,
            model=result.model,
            provider=result.provider,
            tokens_in=result.tokens_in,
            tokens_out=result.tokens_out,
            cost_usd=result.cost_usd,
            inference_geo=result.inference_geo,
            structured=result.structured,
        ).model_dump()

    s, ephemeral = await _resolve_session(req, system_prompt)
    try:
        result = await s.turn(
            _session_turn_payload(s, prompt_parts),
            timeout=GENERATE_TURN_TIMEOUT_SECONDS,
        )
    finally:
        if ephemeral:
            await s.close()
    if result.get("is_error"):
        raise HTTPException(502, f"engine turn error: {result.get('error')}")

    structured: Optional[dict] = None
    text = result.get("text", "")
    if req.structured_schema:
        try:
            structured = json.loads(text)
        except (json.JSONDecodeError, TypeError):
            structured = None  # 파싱 실패는 호출부가 text 로 폴백
    return GenerateResponse(
        text=text,
        model=_response_model_name(s),
        provider=GATEWAY_PROVIDER,
        tokens_in=0,
        tokens_out=0,
        cost_usd=result.get("cost_usd", 0.0),
        inference_geo="us",
        structured=structured,
    ).model_dump()


@app.post("/v1/stream")
async def v1_stream(req: GwGenerateReq):
    """SSE 토큰 스트림. data: 라인으로 텍스트 델타를 흘리고 done/error 프레이밍."""
    prompt_parts = _split_messages(req.messages, ai_role=req.ai_role)
    system_prompt = _inject_schema(prompt_parts.system_prompt, req.structured_schema)
    if not prompt_parts.user_payload:
        raise HTTPException(400, "no user message in payload")

    provider = req.provider or "claude_cli"
    if provider != "claude_cli":
        async def _provider_sse():
            try:
                async for event in stream_with_provider(
                    req,
                    system_prompt=system_prompt,
                    user_payload=prompt_parts.user_payload,
                ):
                    if event.type == "delta" and event.text:
                        yield sse_frame(
                            ENGINE_GATEWAY_SSE_TOKEN,
                            StreamTokenEvent(text=event.text),
                        )
                    elif event.type == "done" and event.result is not None:
                        result = event.result
                        yield sse_frame(
                            ENGINE_GATEWAY_SSE_DONE,
                            StreamDoneEvent(
                                provider=result.provider,
                                model=result.model,
                                tokens_in=result.tokens_in,
                                tokens_out=result.tokens_out,
                                cost_usd=result.cost_usd,
                                turns=1,
                            ),
                        )
            except ProviderError as exc:
                yield sse_frame(
                    ENGINE_GATEWAY_SSE_ERROR,
                    StreamErrorEvent(detail=f"engine provider error: {exc}"),
                )

        return StreamingResponse(
            _provider_sse(),
            media_type="text/event-stream",
            headers={"Cache-Control": "no-cache", "X-Accel-Buffering": "no"},
        )

    s, ephemeral = await _resolve_session(req, system_prompt)

    async def _sse():
        try:
            async for evt in s.turn_stream(
                _session_turn_payload(s, prompt_parts),
                timeout=600.0,
            ):
                if evt.get("type") == "delta":
                    yield sse_frame(
                        ENGINE_GATEWAY_SSE_TOKEN,
                        StreamTokenEvent(text=evt["text"]),
                    )
                elif evt.get("type") == "done":
                    if evt.get("is_error"):
                        yield sse_frame(
                            ENGINE_GATEWAY_SSE_ERROR,
                            StreamErrorEvent(detail=str(evt.get("error"))),
                        )
                    else:
                        yield sse_frame(
                            ENGINE_GATEWAY_SSE_DONE,
                            StreamDoneEvent(
                                provider=GATEWAY_PROVIDER,
                                model=_response_model_name(s),
                                tokens_in=0,
                                tokens_out=0,
                                cost_usd=evt.get("cost_usd", 0.0),
                                turns=evt.get("turns", 0),
                            ),
                        )
        except Exception as e:  # 전송 도중 실패도 SSE 프레임으로 알린다
            yield sse_frame(
                ENGINE_GATEWAY_SSE_ERROR,
                StreamErrorEvent(detail=str(e)),
            )
        finally:
            if ephemeral:
                await s.close()

    return StreamingResponse(
        _sse(),
        media_type="text/event-stream",
        headers={"Cache-Control": "no-cache", "X-Accel-Buffering": "no"},
    )