""" 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 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, EngineMessage as GwMessage, GenerateRequest as GwGenerateReq, StreamDoneEvent, StreamErrorEvent, StreamTokenEvent, sse_frame, ) 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")) BASE_ARGS = [ "-p", "--input-format", "stream-json", "--output-format", "stream-json", "--verbose", "--dangerously-skip-permissions", # 페르소나 격리: cwd/env/git status/메모리(CLAUDE.md) 등 per-machine 섹션을 시스템프롬프트에서 # 제거 → 내담자 AI가 자신이 개발 환경(Claude Code/Vignette repo) 안에 있음을 알아채 캐릭터를 # 깨는 것을 차단. (시스템프롬프트는 아래에서 --system-prompt 로 페르소나만 '교체' 주입.) "--exclude-dynamic-system-prompt-sections", ] def _model_override(model: Optional[str]) -> Optional[str]: value = (model or "").strip() if not value or value == "gateway-default": return None return value class EngineSession: """claude -p 상주 프로세스 1개 = 상담 회기 1개.""" def __init__( self, system_prompt: str | None = None, budget: float = DEFAULT_BUDGET, model: str | None = None, ): self.id = uuid.uuid4().hex self.system_prompt = system_prompt self.budget = budget self.model = _model_override(model) self.proc: asyncio.subprocess.Process | None = None self.lock = asyncio.Lock() # 한 회기 안의 턴은 직렬(상담 왕복) self.cost_usd = 0.0 self.turns = 0 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 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.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 == "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.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] = {} _READY_CACHE: dict[str, Any] = {"checked_at": 0.0, "ok": False, "detail": "not checked"} _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_p", "model": DEFAULT_MODEL or "default(opus-4-8)", "sessions": len(SESSIONS)} def _ready_response(*, cached: bool, age_seconds: float = 0.0) -> JSONResponse: ok = bool(_READY_CACHE.get("ok")) return JSONResponse( { "ok": ok, "engine": "claude_p", "model": DEFAULT_MODEL or "default(opus-4-8)", "sessions": len(SESSIONS), "detail": _READY_CACHE.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): """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. """ age = time.monotonic() - float(_READY_CACHE.get("checked_at", 0.0) or 0.0) if not force and age < READY_TTL_SECONDS: return _ready_response(cached=True, age_seconds=age) async with _READY_LOCK: age = time.monotonic() - float(_READY_CACHE.get("checked_at", 0.0) or 0.0) if not force and age < READY_TTL_SECONDS: return _ready_response(cached=True, age_seconds=age) probe = EngineSession( system_prompt="You are a readiness probe. Reply with exactly OK.", budget=READY_BUDGET_USD, ) ok = False detail = "unknown readiness failure" 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() _READY_CACHE.update({"checked_at": time.monotonic(), "ok": ok, "detail": detail}) return _ready_response(cached=False) @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]) -> tuple[str, str]: """EngineMessage[] → (system_prompt, user_payload). - system 들은 합쳐서 --append-system-prompt 로 주입할 텍스트로. - 마지막 user 발화를 이번 턴 stdin content 로. - 직전 assistant/user 히스토리는 (단발 모드라) system 뒤에 맥락으로 직렬화. (상주 세션 재사용 시에는 풀이 이미 컨텍스트를 들고 있으므로 마지막 user 만 보냄.) """ system_parts: list[str] = [] history_parts: list[str] = [] last_user = "" for m in messages: if m.role == "system": system_parts.append(m.content) elif m.role == "assistant": history_parts.append(f"[이전 상담자 발화]\n{m.content}") elif m.role == "user": if last_user: history_parts.append(f"[이전 내담자 발화]\n{last_user}") last_user = m.content system_prompt = "\n\n".join(p for p in system_parts if p.strip()) return system_prompt, last_user 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() async def _resolve_session(req: GwGenerateReq, system_prompt: str) -> tuple[EngineSession, bool]: """session_id 가 있고 살아있으면 재사용, 아니면 단발용 임시 세션 생성. 반환: (session, ephemeral). ephemeral=True 면 호출부가 응답 후 close 한다. """ requested_model = _model_override(req.model) if req.session_id and req.session_id in SESSIONS: s = SESSIONS[req.session_id] if s.proc is not None and s.proc.returncode is None: if requested_model is None or (s.model or DEFAULT_MODEL) == requested_model: return s, False # 단발(또는 죽은 세션) → 1회성 세션 s = EngineSession( system_prompt=system_prompt or None, budget=DEFAULT_BUDGET, model=requested_model, ) await s.start() return s, True @app.post("/v1/generate") async def v1_generate(req: GwGenerateReq): """단발 생성 (평가 deep-loop, 회기종료 압축 등). GenerateResponse 호환 dict 반환.""" system_prompt, user_payload = _split_messages(req.messages) system_prompt = _inject_schema(system_prompt, req.structured_schema) if not user_payload: raise HTTPException(400, "no user message in payload") s, ephemeral = await _resolve_session(req, system_prompt) try: result = await s.turn(user_payload, timeout=120.0) 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 { "text": text, "model": s.model or DEFAULT_MODEL or "claude-opus-4-8", "provider": "claude_cli", "tokens_in": 0, "tokens_out": 0, "cost_usd": result.get("cost_usd", 0.0), "inference_geo": "us", "structured": structured, } @app.post("/v1/stream") async def v1_stream(req: GwGenerateReq): """SSE 토큰 스트림. data: 라인으로 텍스트 델타를 흘리고 done/error 프레이밍.""" system_prompt, user_payload = _split_messages(req.messages) system_prompt = _inject_schema(system_prompt, req.structured_schema) if not user_payload: raise HTTPException(400, "no user message in payload") s, ephemeral = await _resolve_session(req, system_prompt) async def _sse(): try: async for evt in s.turn_stream(user_payload, 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="claude_cli", model=s.model or DEFAULT_MODEL or "claude-opus-4-8", 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"}, )