feat(engine): claude -p 상주 멀티턴 엔진 게이트웨이 + 실동작 검증

- engine_gateway/gateway.py: 회기당 claude -p 상주 프로세스(stream-json), 턴 직렬, budget 제한
- 세션 생성/턴/종료 HTTP API(FastAPI, :9099)
- 검증: 멀티턴 컨텍스트 유지 + prompt caching 재사용(턴2 +$0.07) 실동작 확인

apps/api/engine_gateway/gateway.py

@@ -0,0 +1,153 @@
+"""
+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 uuid
+
+from fastapi import FastAPI, HTTPException
+from pydantic import BaseModel
+
+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"))
+
+BASE_ARGS = [
+    "-p",
+    "--input-format", "stream-json",
+    "--output-format", "stream-json",
+    "--verbose",
+    "--dangerously-skip-permissions",
+]
+
+
+class EngineSession:
+    """claude -p 상주 프로세스 1개 = 상담 회기 1개."""
+
+    def __init__(self, system_prompt: str | None = None, budget: float = DEFAULT_BUDGET):
+        self.id = uuid.uuid4().hex
+        self.system_prompt = system_prompt
+        self.budget = budget
+        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)]
+        if DEFAULT_MODEL:
+            args += ["--model", DEFAULT_MODEL]
+        if FALLBACK_MODEL:
+            args += ["--fallback-model", FALLBACK_MODEL]
+        if self.system_prompt:
+            args += ["--append-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
+            return {
+                "text": "".join(text_parts),
+                "cost_usd": self.cost_usd,
+                "turns": self.turns,
+                "is_error": result.get("is_error", False),
+                "error": (result.get("errors") or [None])[0],
+            }
+
+    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] = {}
+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)}
+
+
+@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)}