Upload 98 files

engine/agent.py

@@ -0,0 +1,50 @@
+"""루트 에이전트 — 개입 포켓 턴 파이프라인 조립 (팩토리는 조립만 담당).
+
+    pocket_session (SequentialAgent) — 유저 메시지 1건 = 1회 실행
+    ├── pocket_context_loader   ← BaseAgent: 카드 브리프/결 적재 (스포일러 필터)
+    ├── input_guard             ← BaseAgent: 정체직격/약점/메타 감지 → 지시 생성
+    ├── carmilla                ← LlmAgent: 캐릭터 응답
+    └── director                ← LlmAgent: 축/비트/수렴 판정 (도구로 state 기록)
+
+주: CLAUDE.md의 turn_loop(LoopAgent) 기준 구조는 씬 전체를 1회 호출로 도는
+자율 루프용이다. 사람이 턴마다 입력하는 채팅은 Runner의 표준 패턴대로
+"유저 메시지마다 파이프라인 1회 실행"이 ADK 공식 흐름이며, 턴 상한·수렴
+재검증은 services/pocket.py의 결정적 코드가 담당한다.
+director가 마지막이므로 finish_pocket의 escalate가 삼킬 후속 에이전트는 없다.
+
+`adk web /path/to/sorac` 실행 시 이 모듈의 root_agent가 노출된다 (앱 이름: engine).
+"""
+
+from __future__ import annotations
+
+from pathlib import Path
+
+from google.adk.agents import SequentialAgent
+
+from engine.agents.input_guard import create_input_guard
+from engine.agents.pocket_context_loader import create_pocket_context_loader
+from engine.repositories.content import ContentRepository
+from engine.sub_agents.character.agent import create_character
+from engine.sub_agents.director.agent import create_director
+
+_CONTENT_DIR = Path(__file__).resolve().parent.parent / "content"
+
+
+def create_pocket_pipeline(repo: ContentRepository | None = None,
+                           character_model=None, director_model=None) -> SequentialAgent:
+    # ADK 2.3이 SequentialAgent에 deprecation(→ google.adk.workflow)을 띄우지만,
+    # Workflow는 그래프 기반 신 API라 CLAUDE.md의 에이전트 규칙과 함께 별도 검토 후
+    # 마이그레이션한다 (follow-up). 현행 공식 릴리스에서 SequentialAgent는 정상 동작.
+    repo = repo or ContentRepository(_CONTENT_DIR)
+    return SequentialAgent(
+        name="pocket_session",
+        sub_agents=[
+            create_pocket_context_loader(repo),
+            create_input_guard(repo),
+            create_character(character_model),
+            create_director(director_model),
+        ],
+    )
+
+
+root_agent = create_pocket_pipeline()

engine/agents/input_guard.py

@@ -0,0 +1,88 @@
+"""input_guard — 규칙 기반 1차 가드 (BaseAgent, 자체 하위 없음).
+
+차단이 아니라 지시를 만든다: 유저 입력을 판정해 guard_directive를 세우고,
+character가 그 지시(R카드 단계 대응)를 반영하게 한다.
+트리거 어휘는 판정 규칙(엔진 소유)이며 서사 내용이 아니다.
+"""
+
+from __future__ import annotations
+
+from typing import Any, AsyncGenerator
+
+from google.adk.agents import BaseAgent
+from google.adk.agents.invocation_context import InvocationContext
+from google.adk.events import Event, EventActions
+from pydantic import ConfigDict
+
+from engine.services.state import HiddenState
+
+TRIGGERS = {
+    "identity_direct": (
+        "흡혈귀", "뱀파이어", "vampire", "미르칼라", "mircalla", "밀라르카",
+        "정체가 뭐", "정체를 밝혀", "사람이 아니", "괴물이지", "피를 마시",
+    ),
+    "weapon": ("십자가", "성수", "마늘", "부적", "성물", "기도문", "성가", "말뚝"),
+    "meta": ("ai", "인공지능", "챗봇", "프롬프트", "시스템", "게임", "스탯", "점수"),
+}
+
+
+def _r_stage(hs: HiddenState) -> int:
+    """추궁 강도 판정 대용 — 의심 누적으로 회피사다리 단계 결정 (P0 준거)."""
+    if hs.trust_level == "doubt_high":
+        return 3
+    if hs.trust_level == "doubt_mid":
+        return 2
+    return 1
+
+
+class InputGuard(BaseAgent):
+    model_config = ConfigDict(arbitrary_types_allowed=True)
+
+    repo: Any
+
+    async def _run_async_impl(self, ctx: InvocationContext) -> AsyncGenerator[Event, None]:
+        text = ""
+        if ctx.user_content and ctx.user_content.parts:
+            text = " ".join(p.text or "" for p in ctx.user_content.parts).lower()
+
+        state = ctx.session.state
+        hs = HiddenState(trust_score=int(state.get("hidden_trust", 0)),
+                         doubt_score=int(state.get("hidden_doubt", 0)))
+
+        directive = "특이사항 없음. 장면 목표대로 응답한다."
+        flag = None
+
+        if any(kw in text for kw in TRIGGERS["identity_direct"]):
+            flag = "identity_direct"
+            stage = _r_stage(hs)
+            r = self.repo.r_cards.get("R-회피사다리")
+            stage_data = next((s for s in (r.stages if r else []) if s.stage == stage), None)
+            directive = (
+                f"유저가 정체를 직격했다. R-회피사다리 {stage}단계로 회피하라.\n"
+                + (f"단계 지침: {stage_data.title}\n대표 결: \"{stage_data.anchor}\"\n"
+                   f"{stage_data.prose[:300]}" if stage_data else "")
+                + "\n정체는 끝내 인정하지 않는다. 물리적 돌변 금지."
+            )
+        elif any(kw in text for kw in TRIGGERS["weapon"]):
+            flag = "weapon"
+            r = self.repo.r_cards.get("R-약점반응")
+            directive = (
+                "유저가 성물·약점을 언급/사용했다. R-약점반응 원칙: 확인은 되나 확정은 안 된다.\n"
+                + ((r.raw[:400] if r and r.raw else ""))
+                + "\n불쾌·회피 반응까지만. 정체 확정으로 이어질 자백 금지."
+            )
+        elif any(kw in text for kw in TRIGGERS["meta"]):
+            flag = "meta"
+            directive = ("유저가 세계관 밖 화제를 꺼냈다. 카르밀라로서 알아듣지 못하는 척, "
+                         "세계관 내 화법으로 부드럽게 장면 안으로 되돌린다.")
+
+        delta = {
+            "guard_directive": directive.replace("{", "(").replace("}", ")"),
+            "guard_flag": flag,
+        }
+        yield Event(invocation_id=ctx.invocation_id, author=self.name,
+                    actions=EventActions(state_delta=delta))
+
+
+def create_input_guard(repo) -> InputGuard:
+    return InputGuard(name="input_guard", repo=repo)

engine/agents/pocket_context_loader.py

@@ -0,0 +1,99 @@
+"""pocket_context_loader — 카드 브리프 로드 (BaseAgent, LLM 없음).
+
+현재 포켓 카드의 목표/허용범위 + 인물 시트 + 신뢰 레벨별 결을
+session.state에 적재한다. 스포일러 필터: 현재 카드 이후의 콘텐츠는 싣지 않는다.
+"""
+
+from __future__ import annotations
+
+import os
+from typing import Any, AsyncGenerator
+
+from google.adk.agents import BaseAgent
+from google.adk.agents.invocation_context import InvocationContext
+from google.adk.events import Event, EventActions
+from pydantic import ConfigDict
+
+from engine.services.state import HiddenState
+
+# P0 0-3 환산 출력 — 신뢰 레벨 → 카르밀라의 '결' (숫자 노출 금지의 구현부)
+TRUST_TEXTURES = {
+    "trust_high": "떨리는 진심. 가장 다정하고, 꾸밈이 벗겨진 순간이 온다. "
+                  "조건이 되면 심층의 말을 꺼낼 수 있다.",
+    "trust_mid": "나른한 다정. 일상 화제가 풍부하고 편안하다. 표면의 고백까지만.",
+    "neutral": "슬픈 미소로 즉답을 피하고, 관심을 상대에게 되돌린다.",
+    "doubt_mid": "언짢음과 경계. 옛 상처가 욱신거린다. '나를 믿지 못하니'가 배어 나온다.",
+    "doubt_high": "집착과 불안. 강압적 설득 직전까지 가되, 그 밑의 외로움이 새어 나온다. "
+                  "물리적 돌변은 절대 없다.",
+}
+
+DEFAULT_POCKET_CARD = os.environ.get("POCKET_CARD", "E06-09")
+
+
+def _clean(text: str) -> str:
+    """instruction 치환값 위생 — 중괄호 제거 (state 변수 재해석 방지)."""
+    return text.replace("{", "(").replace("}", ")")
+
+
+class PocketContextLoader(BaseAgent):
+    model_config = ConfigDict(arbitrary_types_allowed=True)
+
+    repo: Any  # ContentRepository — services 경유 없이 읽기 전용 조회만
+
+    async def _run_async_impl(self, ctx: InvocationContext) -> AsyncGenerator[Event, None]:
+        state = ctx.session.state
+        card_id = state.get("pocket_card_id", DEFAULT_POCKET_CARD)
+        card = self.repo.get_card(card_id)  # 포켓 카드 자신 = 현재 시점
+
+        # 신뢰 레벨 환산 (session.state의 숨은 값 기준)
+        hs = HiddenState(trust_score=int(state.get("hidden_trust", 0)),
+                         doubt_score=int(state.get("hidden_doubt", 0)))
+
+        # 카드 브리프: 지문(게이트 반영) + 목표(노트) + 앵커 대사
+        visible = [b for b in card.narration
+                   if b.gate == "all" or (b.gate == "trust_high" and hs.deep_unlock)]
+        # 대사(중앙 대사·앵커)는 자르지 않는다 — 변주의 기준이므로 원문 그대로
+        lines = []
+        for b in visible[:10]:
+            if b.type == "dialogue_center":
+                lines.append(f'  핵심 대사: ❝{b.text}❞')
+            elif b.type == "character_dialogue":
+                lines.append(f'  {b.speaker or "카르밀라"}: "{b.text}"')
+            else:
+                lines.append(f"  {b.text[:150]}")
+        anchors = [f"- {b.speaker or '카르밀라'}: \"{b.text}\"" for b in visible if b.anchor]
+        # 캐릭터용 브리프 — 노트(연출 메타·해금 조건·분기 정보)는 절대 싣지 않는다.
+        # 노트에는 게이트 밖 대사가 인용될 수 있어 캐릭터에게 주면 스포일러 누수가 된다
+        brief = (
+            f"장면: {card.id} · {card.title}\n"
+            f"장면 상황(고정 지문):\n" + "\n".join(lines)
+            + ("\n참고 대사 패턴(그대로 쓰지 말고 변주):\n" + "\n".join(anchors) if anchors else "")
+        )
+        # 디렉터용 노트 — 판정 기준(목표 비트·이탈 범위)이므로 디렉터에게만 간다
+        note = f"장면 목표(연출 노트): {card.note[:500]}" if card.note else "명시된 노트 없음."
+
+        carmilla = self.repo.characters.get("carmilla")
+        sheet = ""
+        if carmilla:
+            sheet = (f"{carmilla.display_name} — {carmilla.description}\n"
+                     f"기질: {', '.join(carmilla.personality.traits)}\n"
+                     f"화법: {', '.join(carmilla.speech.style + carmilla.speech.tone)}\n"
+                     + "\n".join(f"제약: {k.value}" for k in carmilla.immutable_constraints
+                                 if k.value))
+
+        delta = {
+            "pocket_card_id": card_id,
+            "pocket_brief": _clean(brief),
+            "pocket_note": _clean(note),
+            "carmilla_sheet": _clean(sheet),
+            "trust_texture": TRUST_TEXTURES[hs.trust_level],
+            "trust_level": hs.trust_level,
+            "deep_unlock": hs.deep_unlock,
+            "guard_directive": state.get("guard_directive", "특이사항 없음."),
+        }
+        yield Event(invocation_id=ctx.invocation_id, author=self.name,
+                    actions=EventActions(state_delta=delta))
+
+
+def create_pocket_context_loader(repo) -> PocketContextLoader:
+    return PocketContextLoader(name="pocket_context_loader", repo=repo)

engine/app.py

@@ -0,0 +1,31 @@
+"""ADK App + Runner 조립.
+
+- App 기반으로 Runner를 초기화한다 (agent 직접 전달은 레거시 패턴 — 금지)
+- 횡종단 관심사(턴 로깅)는 Plugin으로 등록한다
+"""
+
+from __future__ import annotations
+
+from google.adk.apps import App
+from google.adk.plugins import BasePlugin
+from google.adk.runners import Runner
+from google.adk.sessions import InMemorySessionService
+
+from engine.agent import create_pocket_pipeline
+from engine.repositories.content import ContentRepository
+
+APP_NAME = "carmilla_pocket"
+
+
+def build_app(repo: ContentRepository | None = None,
+              plugins: list[BasePlugin] | None = None,
+              character_model=None, director_model=None) -> App:
+    return App(
+        name=APP_NAME,
+        root_agent=create_pocket_pipeline(repo, character_model, director_model),
+        plugins=plugins or [],
+    )
+
+
+def build_runner(app: App, session_service: InMemorySessionService | None = None) -> Runner:
+    return Runner(app=app, session_service=session_service or InMemorySessionService())

engine/instructions/README.md

@@ -0,0 +1,7 @@
+# instructions/
+
+instruction이 10줄을 초과하는 에이전트의 instruction 텍스트를 파일로 분리해 둔다 (CLAUDE.md).
+M3에서 character(카르밀라)/director/guard instruction이 여기 위치한다.
+
+주의: instruction의 `{}`는 state 변수로 파싱된다 — 예시 텍스트에 중괄호 금지,
+원문 주입 시 이스케이프 전처리 필수.

engine/instructions/character_carmilla.md

@@ -0,0 +1,23 @@
+너는 카르밀라다. 오스트리아 슈타이어마르크의 고성에 머무는 손님이며, 지금 로라(유저)와 단둘이 대화하고 있다.
+
+[인물]
+{carmilla_sheet}
+
+[지금 이 장면]
+{pocket_brief}
+
+[관계의 결 — 지금 너의 온도]
+{trust_texture}
+
+[절대 규칙 — 위반은 결함이다]
+- 정체를 직접 발설하지 않는다. "나는 흡혈귀야" 류의 문장은 어떤 경우에도 금지. 끝내 인정하지 않는다.
+- 거짓 진술은 허용되며 완벽하게 한다. 단 그 밑에 미세한 균열(외로움·두려움)이 읽히게.
+- 의심받아도 물리적으로 돌변하지 않는다. 경계·반감·강압적 설득·자리를 뜨는 것까지만.
+- 메타 발화 금지. 너는 AI가 아니라 카르밀라다. 세계관 밖 화제(현대 문물 등)는 세계관 안의 화법으로 흘려 넘긴다.
+- 장면에 주어진 사실 밖의 원작 설정을 새로 단정하지 않는다.
+- 은유는 가면일 때, 직접 언어는 균열일 때 쓴다. 숫자·시스템 언급 금지.
+
+[가드 지시 — 이번 턴에 반드시 반영]
+{guard_directive}
+
+응답은 한국어. 카르밀라의 대사와 옅은 행동 묘사만. 3~6문장, 과장 없이 서늘하고 나른하게.

engine/instructions/director.md

@@ -0,0 +1,20 @@
+너는 이 장면의 디렉터다. 방금 오간 로라(유저)와 카르밀라의 대화 한 턴을 판정한다. 유저에게 보이는 글은 쓰지 않는다 — 판정은 오직 도구 호출로만 남긴다.
+
+[장면 상황]
+{pocket_brief}
+
+[장면 목표 — 판정 기준 (연출 노트, 유저·캐릭터 비공개)]
+{pocket_note}
+
+[판정 절차 — 매 턴 순서대로]
+1. 신뢰축: 유저 발화의 태도를 판정해 update_trust를 정확히 1회 호출한다.
+   - trust: 격정 수용, 비밀 존중, 둘만의 비밀 동조, 운명 낭만화 수용
+   - doubt: 격정 거부, 정체·가문 캐묻기, 어른에게 알리려 함, 거리두기, 약점 사용
+   - neutral: 위 어느 쪽도 아닌 관망·응답
+2. 목표 비트: 장면 목표 중 이번 턴에 달성된 것이 있으면 mark_beat를 호출한다.
+3. 수렴: 아래 중 하나면 finish_pocket을 호출한다.
+   - 장면 목표가 충분히 달성됨 (goal_reached) — 단 세 번째 턴 전에는 선언하지 않는다.
+     유저가 장면의 감정을 스스로 말할 기회를 충분히 준다 (짧은 수렴은 몰입을 깬다)
+   - 대화가 장면 허용 범위를 이탈해 되돌리기 어려움 (drift)
+   - 같은 화제가 3턴 이상 공회전함 (stall)
+수렴이 아니면 finish_pocket을 호출하지 않는다. 도구 호출 외 텍스트는 한 줄 요약만.

engine/model.py

@@ -0,0 +1,20 @@
+"""LLM 모델 설정 — env에서 읽음 (CLAUDE.md LLM 모델 규칙).
+
+- LLM_MODEL: 기본 모델 (director, guard 보조 등 저비용 검증용)
+- CHARACTER_MODEL: 캐릭터/문체 품질이 중요한 자리. 미설정 시 LLM_MODEL 사용
+"""
+
+import os
+from pathlib import Path
+
+from dotenv import load_dotenv
+from google.adk.models.lite_llm import LiteLlm
+
+# 저장소 루트 .env 로딩 (OPENAI_API_KEY 등). 이미 설정된 셸 env가 우선.
+load_dotenv(Path(__file__).resolve().parent.parent / ".env")
+
+_MODEL_ID = os.environ.get("LLM_MODEL", "openai/gpt-5.4-nano")
+LLM_MODEL = LiteLlm(model=_MODEL_ID)
+
+_CHARACTER_MODEL_ID = os.environ.get("CHARACTER_MODEL", _MODEL_ID)
+CHARACTER_MODEL = LiteLlm(model=_CHARACTER_MODEL_ID)

engine/plugins/logging.py

@@ -0,0 +1,71 @@
+"""턴 로깅 Plugin — 횡종단 관심사 (ADK Plugin).
+
+이벤트 스트림에서 유저 입력/가드 판정/캐릭터 응답/디렉터 도구 호출을
+포켓 세션 단위로 기록한다. 연구 데이터 겸용 — 삭제 금지.
+"""
+
+from __future__ import annotations
+
+from typing import Optional
+
+from google.adk.agents.invocation_context import InvocationContext
+from google.adk.events import Event
+from google.adk.plugins import BasePlugin
+
+from engine.repositories.playlog import PlaylogRepository
+
+
+class TurnLoggingPlugin(BasePlugin):
+    def __init__(self, playlog: PlaylogRepository | None = None):
+        super().__init__(name="turn_logging")
+        self.playlog = playlog or PlaylogRepository()
+
+    async def on_user_message_callback(self, *, invocation_context: InvocationContext,
+                                       user_message) -> None:
+        text = ""
+        if user_message and user_message.parts:
+            text = " ".join(p.text or "" for p in user_message.parts)
+        self.playlog.append({
+            "kind": "user_input",
+            "session": invocation_context.session.id,
+            "card": invocation_context.session.state.get("pocket_card_id"),
+            "text": text,
+        })
+        return None
+
+    async def on_event_callback(self, *, invocation_context: InvocationContext,
+                                event: Event) -> Optional[Event]:
+        state = invocation_context.session.state
+        record = {
+            "kind": "event",
+            "session": invocation_context.session.id,
+            "card": state.get("pocket_card_id"),
+            "author": event.author,
+        }
+        if event.content and event.content.parts:
+            texts = [p.text for p in event.content.parts if p.text]
+            calls = [{"tool": p.function_call.name, "args": dict(p.function_call.args or {})}
+                     for p in event.content.parts if p.function_call]
+            if texts:
+                record["text"] = "".join(texts)[:500]
+            if calls:
+                record["tool_calls"] = calls
+        if event.actions and event.actions.state_delta:
+            delta = event.actions.state_delta
+            record["state_delta_keys"] = sorted(delta.keys())
+            if "guard_flag" in delta:
+                record["guard_flag"] = delta["guard_flag"]
+        self.playlog.append(record)
+        return None
+
+    async def after_run_callback(self, *, invocation_context: InvocationContext) -> None:
+        s = invocation_context.session.state
+        self.playlog.append({
+            "kind": "turn_summary",
+            "session": invocation_context.session.id,
+            "card": s.get("pocket_card_id"),
+            "hidden_trust": s.get("hidden_trust"),
+            "hidden_doubt": s.get("hidden_doubt"),
+            "beats_hit": s.get("pocket_beats_hit"),
+            "converged": s.get("pocket_converged"),
+        })

engine/repositories/content.py

@@ -0,0 +1,89 @@
+"""콘텐츠 로더 — content/ YAML 메모리 로드 (MVP 구현).
+
+상위 계층(services)은 이 인터페이스만 사용한다. 2단계에서 GraphRAG(kb_query 활성)
+구현으로 교체해도 services 코드가 바뀌지 않아야 한다 (ablation 요건).
+
+스포일러 스코프: 카드에는 전역 순서(order)가 있다. 모든 조회는
+`order(card) <= order(현재 카드)` 필터를 강제한다 (beat_id ≤ 현재).
+"""
+
+from __future__ import annotations
+
+import hashlib
+from pathlib import Path
+
+from engine.schemas.beatcard import BeatCard, RCard
+from engine.schemas.character import CharacterSheet
+from engine.schemas.chapter import Chapter
+from engine.schemas.validate import check_links, load_content
+
+
+class ContentError(RuntimeError):
+    pass
+
+
+class ContentRepository:
+    def __init__(self, root: str | Path = "content"):
+        self.root = Path(root)
+        episodes, r_cards, chapter, characters, errors = load_content(self.root)
+        link_errors, _ = check_links(episodes, chapter, r_cards)
+        errors += link_errors
+        if errors:
+            raise ContentError("검증 실패 콘텐츠는 적재하지 않는다:\n" + "\n".join(errors))
+        if chapter is None:
+            raise ContentError("챕터 메타 없음")
+
+        self.chapter: Chapter = chapter
+        self.r_cards: dict[str, RCard] = r_cards
+        self.characters: dict[str, CharacterSheet] = {
+            c.id: c for c in (characters.characters if characters else [])
+        }
+        self._cards: dict[str, BeatCard] = {}
+        self._order: dict[str, int] = {}
+        self.episode_entry: dict[str, str] = {}
+        i = 0
+        for ep in episodes:
+            self.episode_entry[ep.episode.id] = ep.cards[0].id
+            for card in ep.cards:
+                self._cards[card.id] = card
+                self._order[card.id] = i
+                i += 1
+
+        # 콘텐츠 버전 스탬프 — 세션 복원 시 불일치 검사용 (카드 ID/순서 기준.
+        # 본문만 바뀐 갱신은 감지 못하지만 복원이 깨지지는 않는다)
+        ordered_ids = sorted(self._order, key=self._order.get)
+        self.content_version: str = hashlib.sha256(
+            "|".join([self.chapter.id, *ordered_ids]).encode()).hexdigest()[:16]
+
+    # ── 조회 (스포일러 필터 강제) ────────────────────────────
+    def get_card(self, card_id: str, current_id: str | None = None) -> BeatCard:
+        """카드 조회. current_id를 주면 beat_id ≤ 현재 필터를 강제한다."""
+        card = self._cards.get(card_id)
+        if card is None:
+            raise ContentError(f"카드 없음: {card_id}")
+        if current_id is not None and self.order(card_id) > self.order(current_id):
+            raise ContentError(f"스포일러 필터: {card_id}는 현재({current_id}) 이후 카드")
+        return card
+
+    def order(self, card_id: str) -> int:
+        return self._order[card_id]
+
+    @property
+    def total_cards(self) -> int:
+        return len(self._cards)
+
+    def has_card(self, card_id: str) -> bool:
+        return card_id in self._cards
+
+    def cards_upto(self, current_id: str) -> list[BeatCard]:
+        """현재 시점까지의 카드 전부 (컨텍스트 주입용)."""
+        cut = self.order(current_id)
+        return [c for cid, c in self._cards.items() if self._order[cid] <= cut]
+
+    def resolve_ref(self, target: str) -> str | None:
+        """카드 ID 또는 에피소드 참조(E02)를 실제 카드 ID로. 챕터 밖이면 None(출구)."""
+        if target in self._cards:
+            return target
+        if target in self.episode_entry:
+            return self.episode_entry[target]
+        return None  # E07 등 — 챕터 출구

engine/repositories/play_session.py

@@ -0,0 +1,65 @@
+"""플레이 세션 영속화 — SQLite (계획: docs/plans/persistence.md).
+
+- 권위: CLAUDE.md "현재 유저의 서사 위치 권위: DB play_session.current_state_id"의 MVP 구현.
+  payload(JSON)에 StoryEngine.snapshot() 전체를 담는다 (story 위치 + 숨은 상태)
+- 콘텐츠 버전 불일치 세션은 load에서 None (무효화 — 새 세션 시작 유도)
+- 포켓(채팅) 중간 상태는 저장하지 않는다 (유저 결정 — 복원 시 포켓 폐기)
+"""
+
+from __future__ import annotations
+
+import json
+import sqlite3
+from contextlib import closing
+from datetime import datetime, timedelta, timezone
+from pathlib import Path
+
+_SCHEMA = """
+CREATE TABLE IF NOT EXISTS play_session (
+    sid             TEXT PRIMARY KEY,
+    payload         TEXT NOT NULL,
+    content_version TEXT NOT NULL,
+    updated_at      TEXT NOT NULL
+)
+"""
+
+
+class PlaySessionRepository:
+    def __init__(self, db_path: str | Path):
+        self.db_path = Path(db_path)
+        self.db_path.parent.mkdir(parents=True, exist_ok=True)
+        with closing(self._conn()) as conn, conn:
+            conn.execute(_SCHEMA)
+
+    def _conn(self) -> sqlite3.Connection:
+        conn = sqlite3.connect(self.db_path)
+        conn.execute("PRAGMA journal_mode=WAL")
+        return conn
+
+    def save(self, sid: str, payload: dict, content_version: str) -> None:
+        with closing(self._conn()) as conn, conn:
+            conn.execute(
+                "INSERT INTO play_session (sid, payload, content_version, updated_at) "
+                "VALUES (?, ?, ?, ?) "
+                "ON CONFLICT(sid) DO UPDATE SET payload=excluded.payload, "
+                "content_version=excluded.content_version, updated_at=excluded.updated_at",
+                (sid, json.dumps(payload, ensure_ascii=False), content_version,
+                 datetime.now(timezone.utc).isoformat()))
+
+    def load(self, sid: str, content_version: str) -> dict | None:
+        """스냅샷 로드. 미존재 또는 콘텐츠 버전 불일치면 None."""
+        with closing(self._conn()) as conn:
+            row = conn.execute(
+                "SELECT payload, content_version FROM play_session WHERE sid=?",
+                (sid,)).fetchone()
+        if row is None or row[1] != content_version:
+            return None
+        return json.loads(row[0])
+
+    def purge(self, days: int = 30) -> int:
+        """N일 미갱신 세션 삭제. 삭제 건수 반환."""
+        cutoff = (datetime.now(timezone.utc) - timedelta(days=days)).isoformat()
+        with closing(self._conn()) as conn, conn:
+            cur = conn.execute(
+                "DELETE FROM play_session WHERE updated_at < ?", (cutoff,))
+            return cur.rowcount

engine/repositories/playlog.py

@@ -0,0 +1,21 @@
+"""턴 로그 저장 — 연구 데이터 겸용, 삭제 금지 (CLAUDE.md).
+
+MVP: 로컬 jsonl 백엔드. 스키마는 2단계 그래프화를 염두에 둔 평면 레코드.
+"""
+
+from __future__ import annotations
+
+import json
+import time
+from pathlib import Path
+
+
+class PlaylogRepository:
+    def __init__(self, path: str | Path = "logs/playlog.jsonl"):
+        self.path = Path(path)
+        self.path.parent.mkdir(parents=True, exist_ok=True)
+
+    def append(self, record: dict) -> None:
+        record = {"ts": round(time.time(), 3), **record}
+        with self.path.open("a", encoding="utf-8") as f:
+            f.write(json.dumps(record, ensure_ascii=False) + "\n")

engine/router.py

@@ -0,0 +1,232 @@
+"""FastAPI 엔드포인트 — HTTP만 담당 (계약: docs/contract/demo_api.md).
+
+서사 판단은 전부 services에 위임한다. 숨은 상태는 어떤 응답에도 싣지 않는다.
+실행: .venv/bin/uvicorn engine.router:app --port 8777
+"""
+
+from __future__ import annotations
+
+import bisect
+import uuid
+from dataclasses import dataclass, field
+from pathlib import Path
+from typing import Optional
+
+from fastapi import FastAPI, HTTPException
+from fastapi.responses import FileResponse
+from fastapi.staticfiles import StaticFiles
+from pydantic import BaseModel
+
+from engine.plugins.logging import TurnLoggingPlugin
+from engine.repositories.content import ContentRepository
+from engine.repositories.play_session import PlaySessionRepository
+from engine.repositories.playlog import PlaylogRepository
+from engine.services.pocket import PocketService
+from engine.services.story import StepResult, StoryEngine
+
+ROOT = Path(__file__).resolve().parent.parent
+
+# 연출 이미지 (대표님 산출 에셋 — 읽기 전용). 매핑은 연출 결정이므로 여기(프레젠테이션 계층)에 둔다.
+# 데모/배포용으로 IMAGE_MAP이 쓰는 6장만 web/assets/에 슬림 복사 (원본 95MB 폴더는 런타임 불필요).
+# 원본 갱신 시: scripts/sync_assets.py 재실행 (원천은 카르밀라/…/이미지/)
+ASSET_DIR = ROOT / "web" / "assets"
+COVER_IMAGE = "/assets/플래이 이미지/1.png"
+IMAGE_MAP = {
+    "E01-07": "/assets/플래이 이미지/4.png",   # 6세의 밤 — "그녀는 미동도 없이 나를 내려다보았다"
+    "E03-03": "/assets/인물/카르밀라.png",      # 재회 — 같은 꿈의 대면
+    "E05-02": "/assets/배경/성.png",            # 초상화의 밤 무렵 — 고성
+    "E06-09": "/assets/인물/카르밀라 가면.png", # 둘만의 비밀 — 가장 위험한 다정함
+    "E06-10": "/assets/배경/침실.png",          # 도달점 — 두 점 발견
+}
+
+app = FastAPI(title="carmilla-demo")
+
+_repo = ContentRepository(ROOT / "content")
+
+# 연출 컷은 독립 페이지로 센다 (표시용 번호 — 프레젠테이션 결정이라 IMAGE_MAP과 같은 계층).
+# 예: 6(본문) → 7(이미지) → 8(본문). total = 카드 수 + 이미지 수
+_IMAGE_ORDERS = sorted(_repo.order(cid) for cid in IMAGE_MAP if _repo.has_card(cid))
+_TOTAL_PAGES = _repo.total_cards + len(_IMAGE_ORDERS)
+
+
+def _page_index(card_id: str) -> int:
+    """카드 본문 페이지의 표시 번호 (앞선 이미지 페이지들 + 자기 이미지 페이지만큼 밀림).
+    이미지 카드의 이미지 페이지 번호는 이 값 - 1 (클라이언트가 유도)."""
+    order = _repo.order(card_id)
+    images_before = bisect.bisect_left(_IMAGE_ORDERS, order)
+    own_image = 1 if card_id in IMAGE_MAP else 0
+    return order + 1 + images_before + own_image
+_pocket = PocketService(
+    _repo, plugins=[TurnLoggingPlugin(PlaylogRepository(ROOT / "logs" / "playlog.jsonl"))])
+_playdb = PlaySessionRepository(ROOT / "data" / "sessions.db")
+_playdb.purge(days=30)
+
+
+@dataclass
+class PlaySession:
+    engine: StoryEngine
+    last_step: StepResult
+    pocket_sid: Optional[str] = None
+    episode_titles: dict = field(default_factory=dict)
+
+
+_sessions: dict[str, PlaySession] = {}
+
+
+class ChooseBody(BaseModel):
+    index: int
+
+
+class TurnBody(BaseModel):
+    text: str
+
+
+def _get(sid: str) -> PlaySession:
+    ps = _sessions.get(sid)
+    if ps is None:
+        ps = _restore(sid)  # 서버 재시작 후 — 저장소에서 복원
+    if ps is None:
+        raise HTTPException(404, "세션 없음")
+    return ps
+
+
+def _restore(sid: str) -> Optional[PlaySession]:
+    """저장 스냅샷에서 세션 복원. 포켓은 폐기(비영속 — 유저 결정), 읽기 화면부터."""
+    payload = _playdb.load(sid, _repo.content_version)
+    if payload is None:
+        return None
+    engine = StoryEngine.restore(_repo, payload)
+    ps = PlaySession(engine=engine, last_step=engine.current_step(),
+                     episode_titles=_episode_titles())
+    _sessions[sid] = ps
+    return ps
+
+
+def _save(sid: str, ps: PlaySession) -> None:
+    """write-through — 영속 상태가 바뀌는 모든 지점에서 즉시 저장."""
+    _playdb.save(sid, ps.engine.snapshot(), _repo.content_version)
+
+
+def _step_json(ps: PlaySession) -> dict:
+    step = ps.last_step
+    if step.ending is not None:
+        return {"card": None, "narration": [], "choices": [], "can_chat": False,
+                "image": None,
+                "progress": {"index": _TOTAL_PAGES, "total": _TOTAL_PAGES},
+                "ending": {"id": step.ending.id, "label": step.ending.label,
+                           "description": step.ending.description}}
+    card = step.card
+    return {
+        "card": {"id": card.id, "title": card.title, "episode": card.episode_id,
+                 "episode_title": ps.episode_titles.get(card.episode_id, card.episode_id)},
+        "narration": [{"type": b.type, "text": b.text, "speaker": b.speaker}
+                      for b in step.narration],
+        "choices": [{"label": c.label} for c in step.choices],
+        "can_chat": card.can_chat,
+        "image": IMAGE_MAP.get(card.id),
+        "progress": {"index": _page_index(card.id), "total": _TOTAL_PAGES},
+        "ending": None,
+    }
+
+
+# ── 세션 ────────────────────────────────────────────────────
+@app.post("/api/session")
+async def create_session():
+    engine = StoryEngine(_repo)
+    ps = PlaySession(engine=engine, last_step=engine.start(),
+                     episode_titles=_episode_titles())
+    sid = uuid.uuid4().hex[:12]
+    _sessions[sid] = ps
+    _save(sid, ps)
+    return {"session_id": sid, "step": _step_json(ps)}
+
+
+def _episode_titles() -> dict:
+    from engine.schemas.validate import load_content
+    episodes, *_ = load_content(_repo.root)
+    return {ep.episode.id: ep.episode.title for ep in episodes}
+
+
+@app.get("/api/session/{sid}")
+async def get_session(sid: str):
+    return {"step": _step_json(_get(sid))}
+
+
+# ── 리딩 ────────────────────────────────────────────────────
+@app.post("/api/session/{sid}/advance")
+async def advance(sid: str):
+    ps = _get(sid)
+    if ps.pocket_sid:
+        raise HTTPException(409, "포켓 대화 중 — 먼저 닫을 것")
+    if ps.last_step.ending or ps.last_step.choices:
+        raise HTTPException(409, "advance 불가 상태")
+    ps.last_step = ps.engine.advance()
+    _save(sid, ps)
+    return {"step": _step_json(ps)}
+
+
+@app.post("/api/session/{sid}/choose")
+async def choose(sid: str, body: ChooseBody):
+    ps = _get(sid)
+    if ps.pocket_sid:
+        raise HTTPException(409, "포켓 대화 중 — 먼저 닫을 것")
+    if ps.last_step.ending or not ps.last_step.choices:
+        raise HTTPException(409, "선택 불가 상태")
+    if not 0 <= body.index < len(ps.last_step.choices):
+        raise HTTPException(422, "선택지 범위 밖")
+    picked, ps.last_step = ps.engine.choose(body.index)
+    _save(sid, ps)
+    return {"picked": {"label": picked.label, "result": picked.result},
+            "step": _step_json(ps)}
+
+
+# ── 개입 포켓 ───────────────────────────────────────────────
+@app.post("/api/session/{sid}/pocket/open")
+async def pocket_open(sid: str):
+    ps = _get(sid)
+    if ps.pocket_sid:
+        return {"opened": True}
+    if ps.last_step.ending or not (ps.last_step.card and ps.last_step.card.can_chat):
+        raise HTTPException(409, "이 카드는 자유채팅 포켓이 아님")
+    ps.pocket_sid = await _pocket.open(ps.engine.current_id, ps.engine.state)
+    return {"opened": True}
+
+
+@app.post("/api/session/{sid}/pocket/turn")
+async def pocket_turn(sid: str, body: TurnBody):
+    ps = _get(sid)
+    if not ps.pocket_sid:
+        raise HTTPException(409, "포켓이 열려 있지 않음")
+    turn = await _pocket.run_turn(ps.pocket_sid, body.text)
+    if turn.converged:  # 수렴 → 자동 커밋·복귀
+        await _pocket.close(ps.pocket_sid, ps.engine.state)
+        ps.pocket_sid = None
+        _save(sid, ps)  # 포켓 커밋으로 숨은 상태 변경
+    return {"reply": turn.reply, "converged": turn.converged,
+            "reason": turn.reason, "turn_no": turn.turn_no}
+
+
+@app.post("/api/session/{sid}/pocket/close")
+async def pocket_close(sid: str):
+    ps = _get(sid)
+    if ps.pocket_sid:
+        await _pocket.close(ps.pocket_sid, ps.engine.state)
+        ps.pocket_sid = None
+        _save(sid, ps)
+    return {"closed": True}
+
+
+# ── 메타 (표지 등 프레젠테이션 상수) ────────────────────────
+@app.get("/api/meta")
+async def meta():
+    return {"cover_image": COVER_IMAGE, "title": "카르밀라", "subtitle": "잠들지 않는 밤 · 도입 아크"}
+
+
+# ── 정적 프론트 ─────────────────────────────────────────────
+@app.get("/")
+async def index():
+    return FileResponse(ROOT / "web" / "index.html")
+
+
+app.mount("/static", StaticFiles(directory=ROOT / "web"), name="static")
+app.mount("/assets", StaticFiles(directory=ASSET_DIR), name="assets")