"""카드 진행 상태머신 — 서사 진행의 최종 결정권 (결정적 코드, LLM 없음). - entry_beat부터 canon 카드 페이징, next 링크 전이 - 선택지: axis 반영 + 분기(next/preset_branch) 결정 - 쇠약도: #메커니즘/쇠약도 카드 진입 시 단조 증가 - 챕터 출구(E07 등) 도달 시 ending_rules 조건식을 코드로 평가 (LLM 판정 금지) - 심층 게이트(trust_high) 지문은 신뢰 레벨에 따라 노출 결정 """ from __future__ import annotations import re from dataclasses import dataclass, field from typing import Optional from engine.repositories.content import ContentRepository from engine.schemas.beatcard import BeatCard, Choice, NarrationBlock from engine.schemas.chapter import EndingRule from engine.services.state import HiddenState # ending_rules condition에 허용되는 문법: 변수/정수/비교/and (그 외는 검증 실패) RE_CONDITION_TOKEN = re.compile( r"^\s*(trust_score|doubt_score)\s*(>=|<=|>|<|==)\s*(\d+)\s*$") def eval_condition(condition: str, state: HiddenState) -> bool: """조건식 평가 — eval() 금지, 허용 문법만 파싱.""" if condition.strip() == "default": return True values = {"trust_score": state.trust_score, "doubt_score": state.doubt_score} for clause in condition.split(" and "): m = RE_CONDITION_TOKEN.match(clause) if not m: raise ValueError(f"허용되지 않는 조건식: {condition!r}") left, op, right = values[m.group(1)], m.group(2), int(m.group(3)) ok = {"<": left < right, "<=": left <= right, ">": left > right, ">=": left >= right, "==": left == right}[op] if not ok: return False return True @dataclass class StepResult: """한 걸음의 출력 — 프론트/CLI가 그대로 렌더링한다.""" card: Optional[BeatCard] = None narration: list[NarrationBlock] = field(default_factory=list) # 게이트 필터 적용분 choices: list[Choice] = field(default_factory=list) is_pocket: bool = False ending: Optional[EndingRule] = None # 챕터 종료 시에만 class StoryEngine: def __init__(self, repo: ContentRepository, state: HiddenState | None = None): self.repo = repo self.state = state or HiddenState() self.current_id: Optional[str] = None self.preset_branch: Optional[str] = None # 상류 선택이 예약한 하류 분기 self.visited: list[str] = [] self.ending: Optional[EndingRule] = None # ── 영속화 (스냅샷/복원) ──────────────────────────────── def snapshot(self) -> dict: """JSON 직렬화 가능한 전체 스냅샷 (story + hidden).""" return { "story": { "current_id": self.current_id, "visited": list(self.visited), "preset_branch": self.preset_branch, "ending_id": self.ending.id if self.ending else None, }, "hidden": self.state.snapshot(), } @classmethod def restore(cls, repo: ContentRepository, payload: dict) -> "StoryEngine": """스냅샷 복원 — _enter()를 거치지 않는다 (쇠약도/축 부수효과 재적용 금지).""" story = payload["story"] engine = cls(repo, state=HiddenState.from_snapshot(payload["hidden"])) engine.current_id = story["current_id"] engine.visited = list(story["visited"]) engine.preset_branch = story.get("preset_branch") ending_id = story.get("ending_id") if ending_id: engine.ending = next( r for r in repo.chapter.ending_rules if r.id == ending_id) return engine def current_step(self) -> StepResult: """현재 위치의 StepResult 재구성 (렌더 전용 — 상태 무변경).""" if self.ending: return StepResult(ending=self.ending) card = self.repo.get_card(self.current_id) return StepResult(card=card, narration=self.visible_narration(card), choices=list(card.choices), is_pocket=card.is_pocket) # ── 진행 ──────────────────────────────────────────────── def start(self) -> StepResult: return self._enter(self.repo.chapter.entry_beat) def _enter(self, card_id: str) -> StepResult: card = self.repo.get_card(card_id) self.current_id = card_id self.visited.append(card_id) # 앎 확립 (엔진 A): 챕터 진입 카드에서 1회 if card_id == self.repo.chapter.entry_beat: self.state.establish_awareness() # 쇠약도 (신뢰와 독립 — 핵심 아이러니) if card.frailty: self.state.advance_frailty() # 카드 자체 축 — 캐논 이벤트 (예: E06-10 '첫 진짜 의심'은 모든 경로에 적용). # 단, 선택지들이 축을 갖고 있으면 선택이 결정하므로 중복 적용하지 않는다 if card.axis and not any(ch.axis for ch in card.choices): self.state.apply_axis(card.axis, source=card_id) return StepResult(card=card, narration=self.visible_narration(card), choices=list(card.choices), is_pocket=card.is_pocket) def advance(self) -> StepResult: """선택지 없는 카드에서 다음으로 (책장 넘기기).""" card = self.repo.get_card(self.current_id) if card.choices: raise RuntimeError(f"{card.id}: 선택지가 있는 카드 — choose()를 쓸 것") return self._goto(card.next, card) def choose(self, index: int) -> tuple[Choice, StepResult]: """선택지 반영 — axis 적용 + 분기 결정.""" card = self.repo.get_card(self.current_id) if not card.choices: raise RuntimeError(f"{card.id}: 선택지 없는 카드") choice = card.choices[index] self.state.apply_axis(choice.axis, source=f"{card.id}[{index}]") if choice.preset_branch: self.preset_branch = choice.preset_branch # 분기 카드: preset이 예약돼 있으면 그것이 우선 (상류 커밋 존중) target = choice.next if card.next_options: if self.preset_branch in card.next_options: target = self.preset_branch self.preset_branch = None elif target is None: raise RuntimeError(f"{card.id}: 분기 카드인데 선택지에 next 없음") return choice, self._goto(target or card.next, card) def _goto(self, target: Optional[str], from_card: BeatCard) -> StepResult: if target is None: # 나가는 링크 없음 — 콘텐츠 검증에서 경고된 상태. 챕터 종료로 처리 return self._finish() resolved = self.repo.resolve_ref(target) if resolved is None: # E07 등 챕터 범위 밖 → 챕터 종료 return self._finish() return self._enter(resolved) # ── 엔딩 (결정적 코드) ────────────────────────────────── def _finish(self) -> StepResult: rules = sorted(self.repo.chapter.ending_rules, key=lambda r: r.priority) for rule in rules: if eval_condition(rule.condition, self.state): self.ending = rule return StepResult(ending=rule) raise RuntimeError("default 엔딩 규칙 없음 — 콘텐츠 결함") # ── 지문 게이트 (엔진 D: 심층 해금) — Pager 등 렌더 계층도 사용 ── def visible_narration(self, card: BeatCard) -> list[NarrationBlock]: return [b for b in card.narration if b.gate == "all" or (b.gate == "trust_high" and self.state.deep_unlock)]