hyungi_document_server/app/api/eid_chat.py

"""이드 채팅 표면 — POST /api/eid/chat (eid-chat 트랙).

확정 결정:
  - D-1 경로 = /api/eid/chat (main.py prefix=/api/eid + 본 라우터 POST /chat)
  - D-2 mode 닫힌 어휘: daily / deep — 둘 다 mac-mini-default (맥북 백지화 2026-06-11,
    맥미니 Qwen 27B 단일 호스트. deep = ReAct 자동검색 모드 구분). 클라는 mode 만 보냄 —
    claude-cloud / auto 금지 (Literal 로 422 차단). 게이트 = alias 기준 자동 적용(무게이트 폐지).
  - D-3 독립 /chat 라우트 (frontend) — 본 모듈은 백엔드 API 만.
  - D-5 LLM 호출 = EidAIClient.call_stream 한 곳 (이드 egress 봉쇄 불변식 #5,
    RouterBackend 직접 호출 금지).
  - D-6 rules.md 부재 = 503 substrate_degraded fail-closed — 다른 표면의 degraded 배너
    컨벤션(compose._rules)과 달리 채팅은 진행 자체를 거부.

응답 = router SSE 라인 단위 중계 (text/event-stream — call_stream 이 model 필드를 mode
어휘로 치환·usage 제거, 프레이밍 보존. 본 모듈은 무변형 relay). 스트림 시작 전
backend 실패는 /api/search/ask 와 동일 shape 의 503 + error_reason 매핑(자동 fallback 0).
로그는 메타 1줄(mode·턴수·status)만 — 대화 본문 로깅 0.
"""

from __future__ import annotations

import asyncio
import json
from collections.abc import AsyncIterator
from typing import Annotated, Literal

import httpx
from fastapi import APIRouter, Depends
from fastapi.responses import JSONResponse, StreamingResponse
from pydantic import BaseModel, Field, field_validator, model_validator
from sqlalchemy.ext.asyncio import AsyncSession

from core.auth import get_current_user
from core.database import get_session
from core.utils import setup_logger
from eid import compose as eid_compose
from eid.ai import EidAIClient
from models.user import User
from services.llm.backends import BackendUnavailable, _router_url, get_backend
from services.search import llm_gate
from services.search.react_loop import agentic_ask_loop

logger = setup_logger("eid_chat")

router = APIRouter()

# ── ds-eid-ask-absorb P1: deep 모드 = ReAct 자동검색 (맥미니 Qwen 27B, 2026-06-11~) ──
# 비생성 reachability probe — router 도달만 확인(coarse). 27B(맥북) 자체 미가용은
# 첫 generate_with_tools 호출의 BackendUnavailable → mid-stream error envelope 로 커버
# (plan: probe 정밀도 불필요, TOCTOU 는 in-stream error 가 처리). ~2s 타임아웃·생성 슬롯 비점유.
_DEEP_PROBE_TIMEOUT = httpx.Timeout(connect=2.0, read=2.0, write=2.0, pool=2.0)
# heartbeat: ReAct 다회 tool call 시 수십초 무출력 → 프록시 idle timeout 차단.
# `{"phase":"ping"}` no-op 이벤트 (프론트 envelope 파서가 자연 스킵 — `: ping` comment 는
# POST SSE fetch 파서가 처리 보장 안 됨).
_HEARTBEAT_INTERVAL_S = 10.0


async def _probe_router_reachable() -> bool:
    """router(:8890) /v1/models GET — 도달 확인(비생성). 실패/비200 = 미가용."""
    url = f"{_router_url().rstrip('/')}/v1/models"
    try:
        async with httpx.AsyncClient(timeout=_DEEP_PROBE_TIMEOUT) as client:
            resp = await client.get(url)
            return resp.status_code == 200
    except Exception:
        return False


def _sse(obj: dict) -> bytes:
    """SSE 이벤트 1건 — data: <json>\\n\\n. final_answer 는 OpenAI 호환 choices.delta.content
    로, sources/phase 는 별 envelope 키로(프론트가 분기). model/usage 머신 메타 미포함."""
    return b"data: " + json.dumps(obj, ensure_ascii=False).encode("utf-8") + b"\n\n"


class ChatMessage(BaseModel):
    """채팅 턴 1건. role=system 은 Literal 밖 → 422 (system 합본은 서버 compose 만 주입)."""

    role: Literal["user", "assistant"]
    content: str = Field(min_length=1, max_length=8000)


# 대화 총량 cap (전 메시지 content 합) — per-message 8000·40턴 제한과 별도의 총량 상한
_TOTAL_CONTENT_CAP = 32000


class ChatRequest(BaseModel):
    """POST /api/eid/chat body. mode 는 닫힌 어휘(D-2), messages 는 1~40턴 + 총량 32000자."""

    mode: Literal["daily", "deep"]
    messages: list[ChatMessage] = Field(min_length=1, max_length=40)

    @field_validator("messages")
    @classmethod
    def _last_turn_is_user(cls, v: list[ChatMessage]) -> list[ChatMessage]:
        if v and v[-1].role != "user":
            raise ValueError("마지막 메시지는 role=user 여야 합니다")
        return v

    @model_validator(mode="after")
    def _total_content_cap(self) -> "ChatRequest":
        if sum(len(m.content) for m in self.messages) > _TOTAL_CONTENT_CAP:
            raise ValueError(
                "대화 총량 초과 — 새 대화로 시작하거나 입력을 줄여주세요 "
                f"(전체 메시지 합 {_TOTAL_CONTENT_CAP}자 제한)"
            )
        return self


@router.get("/status")
async def eid_status(
    user: Annotated[User, Depends(get_current_user)],
):
    """이드 backend 점유 상태 스냅샷 — GET /api/eid/status (UI 의 "대기 vs 고장" 구분용).

    daily(맥미니 MLX) 의 DS 프로세스 내부 llm_gate 점유만 본다 — 외부 소비자
    (맥미니 자체 derived-worker·Hermes 등)의 endpoint 점유는 미포착.
    따라서 busy=true 는 확실(지금 줄이 있다), false 는 근사(외부 점유 가능성 잔존).

    가벼움 보장: DB 0 / LLM 0 / 본문 로깅 0 — 폴링 대상으로 안전.
    자동 fallback 판단 근거로 쓰지 않는다 (모드 전환 = 명시 버튼만, 정책).
    """
    snap = llm_gate.gate_status()
    inflight = bool(snap["inflight"])
    waiters = int(snap["waiters"])
    return {
        "daily": {
            "busy": inflight or waiters > 0,
            "inflight": inflight,
            "waiters": waiters,
        }
    }


def _backend_unavailable_response(body: ChatRequest, reason: str, backend_name: str) -> JSONResponse:
    """스트림 시작 전 27B 미가용 → ask 컨벤션과 동일 shape 503 (자동 fallback 0)."""
    logger.warning(
        "eid_chat backend_unavailable mode=%s turns=%d status=503 reason=%s",
        body.mode, len(body.messages), reason,
    )
    return JSONResponse(
        status_code=503,
        content={
            "error": "backend_unavailable",
            "error_reason": reason,
            "backend_requested": backend_name,
            "detail": (
                "심층 엔진(검색)이 일시적으로 응답할 수 없습니다. "
                "잠시 후 다시 시도하거나 일상 모드로 물어보세요."
            ),
        },
    )


async def _eid_chat_deep(body: ChatRequest, session: AsyncSession) -> StreamingResponse | JSONResponse:
    """deep 모드 = ReAct 자동검색. ReAct(`tool_choice=auto`)가 검색 여부를 LLM 자율 판단 —
    검색 불요 질문은 early-exit 으로 대화 답변. substrate(persona+rules+react_ask task)는
    agentic_ask_loop 내부 compose("react_ask") 가 주입(evidence-first 자동 상속).

    멀티턴 = 1단계는 마지막 user 메시지 단독 처리(agentic_ask_loop 가 query: str — history
    미지원). 후속 질문 대명사 해소는 2단계 백로그.
    """
    # ① 첫 SSE 바이트(=HTTP 200 확정) 전 비생성 probe — router 도달 실패 시 503 (재매핑 가능 구간)
    if not await _probe_router_reachable():
        return _backend_unavailable_response(body, "router_unreachable", "mac-mini-default")

    query = body.messages[-1].content  # 메시지 단독 처리 (마지막 user 턴)
    backend = get_backend("mac-mini-default")

    async def _stream() -> AsyncIterator[bytes]:
        # ② phase:searching 방출 = HTTP 200 확정. 이후 미가용은 503 불가 → in-stream error.
        yield _sse({"phase": "searching"})
        task = asyncio.create_task(agentic_ask_loop(session, query, backend=backend))
        try:
            # heartbeat: task 미완 동안 ~10s 마다 ping (shield 로 wait_for 취소가 task 안 죽임)
            while not task.done():
                try:
                    await asyncio.wait_for(asyncio.shield(task), timeout=_HEARTBEAT_INTERVAL_S)
                except asyncio.TimeoutError:
                    yield _sse({"phase": "ping"})
            result = task.result()  # BackendUnavailable 은 여기서 raise (mid-stream)
            # final_answer = OpenAI 호환 1청크(프론트 기존 content 누적 경로 재사용)
            yield _sse({"choices": [{"delta": {"content": result.final_answer}}]})
            # 근거 = 별 envelope (citation 번호 없음 — 프론트가 순서 기반). partial = 근거 부족 표식
            yield _sse({"eid_sources": result.sources, "partial": result.partial})
            yield b"data: [DONE]\n\n"
            logger.info(
                "eid_chat deep ok turns=%d sources=%d partial=%s iters=%d",
                len(body.messages), len(result.sources), result.partial, result.iterations,
            )
        except BackendUnavailable as exc:
            # mid-stream 미가용(검색 중 AC 분리·뚜껑 닫힘) — 200 이미 송신, in-stream error envelope.
            # error 뒤 [DONE] = 프론트 sawDone 로 '중단' 오경보 방지(명시 error notice 유지).
            logger.warning(
                "eid_chat deep mid-stream unavailable turns=%d reason=%s",
                len(body.messages), exc.reason,
            )
            yield _sse({"phase": "error", "error_reason": exc.reason})
            yield b"data: [DONE]\n\n"
        except asyncio.CancelledError:
            raise  # 클라 disconnect — finally 가 task 정리
        except Exception:
            logger.exception("eid_chat deep stream failed turns=%d", len(body.messages))
            yield _sse({"phase": "error", "error_reason": "deep_failed"})
            yield b"data: [DONE]\n\n"
        finally:
            # 클라 disconnect 시 ReAct task 고아화 방지 — cancel + await(전파 완료 보장).
            # 안 하면 27B 가 닫힌 연결 위해 수분 점유, router 동시성상 다음 검색 대기.
            if not task.done():
                task.cancel()
            try:
                await task
            except (asyncio.CancelledError, Exception):
                pass

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


@router.post("/chat")
async def eid_chat(
    body: ChatRequest,
    user: Annotated[User, Depends(get_current_user)],
    session: Annotated[AsyncSession, Depends(get_session)],
):
    """이드 채팅 — daily = router SSE pass-through(대화) / deep = ReAct 자동검색(근거).

    503 경로 (모두 자동 fallback 없음):
      - substrate_degraded: rules.md 부재 (D-6 fail-closed, 채팅 진행 거부)
      - backend_unavailable: 스트림 시작 전 backend 실패 (daily/deep 공통, ask 컨벤션 shape)
    """
    # D-6: rules 부재 = fail-closed. 채팅은 안전·정책 가드 없이 진행하지 않는다(배너 X).
    if not eid_compose.rules_present():
        logger.error(
            "eid_chat substrate_degraded mode=%s turns=%d status=503 — rules.md 부재, 채팅 거부",
            body.mode, len(body.messages),
        )
        return JSONResponse(
            status_code=503,
            content={
                "detail": (
                    "이드 substrate 가 degraded 상태입니다 (운영 규칙 rules.md 부재). "
                    "복구 전까지 채팅을 진행하지 않습니다."
                ),
                "error_reason": "substrate_degraded",
            },
        )

    # deep = ReAct 자동검색 (별 흐름 — probe + 동기 ReAct → SSE 변환)
    if body.mode == "deep":
        return await _eid_chat_deep(body, session)

    # daily = 순수 대화 SSE pass-through (기존)
    system = eid_compose.compose("eid_chat", task="")
    client = EidAIClient()
    stream = client.call_stream(
        body.mode, [m.model_dump() for m in body.messages], system,
    )

    # async generator 는 첫 __anext__ 에서야 실제 요청 전송 — 스트림 시작 전 실패(연결/4xx/5xx)
    # 를 503 으로 매핑하기 위해 첫 chunk 를 여기서 먼저 당긴다.
    try:
        first = await anext(stream, None)
    except BackendUnavailable as exc:
        logger.warning(
            "eid_chat backend_unavailable mode=%s turns=%d status=503 reason=%s",
            body.mode, len(body.messages), exc.reason,
        )
        await client.close()
        return JSONResponse(
            status_code=503,
            content={
                "error": "backend_unavailable",
                "error_reason": exc.reason,
                "backend_requested": exc.backend_name,
                "detail": (
                    "선택한 모드의 backend 가 일시적으로 응답할 수 없습니다. "
                    "잠시 후 다시 시도하거나 mode 를 바꿔 호출하세요."
                ),
            },
        )
    except BaseException:
        await client.close()
        raise

    # 메타 로그 1줄 — 본문 로깅 0 (대화 내용은 어디에도 남기지 않는다)
    logger.info(
        "eid_chat stream mode=%s turns=%d status=200", body.mode, len(body.messages)
    )

    async def _passthrough():
        # call_stream 방출분 무변형 relay (정화는 call_stream 라인 단위 한 곳). 취소·
        # disconnect 포함 finally 에서 generator aclose → AsyncExitStack 이 upstream 정리.
        try:
            try:
                if first is not None:
                    yield first
                async for chunk in stream:
                    yield chunk
            except (BackendUnavailable, httpx.HTTPError) as exc:
                # 스트림 시작 후 절단 — status 200 은 이미 송신돼 재매핑 불가. 메타 로그
                # 1줄만 남기고 조용히 종료(traceback 전파 0) — 프론트는 [DONE] 부재로 처리.
                logger.warning(
                    "eid_chat stream aborted mode=%s turns=%d reason=%s",
                    body.mode, len(body.messages),
                    getattr(exc, "reason", type(exc).__name__),
                )
                return
        finally:
            # stream.aclose() 가 예외여도 client.close() 는 보장 (중첩 finally)
            try:
                await stream.aclose()
            finally:
                await client.close()

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