"""Character turn pipeline (Section 8.1). For each player utterance directed at a character: 1. Build context (card + alibi + ledger + transcript), enforcing the knowledge boundary (strip locked/unknowable topics). 2. Draft a reply (character tier). 3. Run leak + consistency guards. 4. Pass -> emit, extract claims, append to ledger + transcript, inject tell. 5. Fail -> regenerate (bounded) with the guard's reason fed back. 6. Exhausted -> enter the crack state (the designed climax). """ from __future__ import annotations import random from dataclasses import dataclass, field from ..llm.client import LLMClient from ..llm.prompts import PromptRegistry from ..models import Alibi, CharacterCard, Claim, LedgerEntry, TranscriptEvent from ..worldio import World from .clues import ClueGraph from .crack import CrackMachine, CrackResult from .extractor import ClaimExtractor from .guard.consistency import ConsistencyGuard from .guard.leak import LeakGuard from .ledger import LedgerStore from .timeline import TimelineIndex @dataclass class TurnOutcome: text: str cracked: bool crack_behavior: str = "" tell: str = "" claims: list[Claim] = field(default_factory=list) guard_rejections: int = 0 discovered_clues: list[str] = field(default_factory=list) class CharacterPipeline: def __init__( self, *, world: World, client: LLMClient, prompts: PromptRegistry, ledger: LedgerStore, clue_graph: ClueGraph, regenerate_retries: int, ) -> None: self.world = world self.client = client self.prompts = prompts self.ledger = ledger self.clue_graph = clue_graph self.retries = regenerate_retries self.extractor = ClaimExtractor(client, prompts) self.leak_guard = LeakGuard(client, prompts) self.consistency_guard = ConsistencyGuard(client, prompts, ledger) self.crack_machine = CrackMachine(client, prompts) self.timeline = TimelineIndex(world.timeline) def _witnessed_facts(self, card: CharacterCard) -> list[dict[str, str]]: """Concrete timeline rows this character was present for or observed — the ground truth they can speak to (and choose to lie about).""" return [ {"time": s.time_slice, "location": s.location, "action": s.action} for s in self.timeline.witnessed_by(card.name) ] def _alibi_for(self, name: str) -> Alibi | None: for alibi in self.world.alibis.values(): if any(c.lower() == name.lower() for c in alibi.characters): return alibi return None def _unlocked_topics(self, discovered: set[str]) -> list[str]: return [ cid for cid in self.clue_graph.nodes if self.clue_graph.is_unlocked(cid, discovered) ] def run_turn( self, *, card: CharacterCard, player_message: str, turn: int, transcript_tail: list[TranscriptEvent], discovered: set[str], confront_count: int, already_cracked: bool, ) -> TurnOutcome: unlocked = self._unlocked_topics(discovered) alibi = self._alibi_for(card.name) # Engine-side crack trigger from surfaced pressure (Section 8.3). if already_cracked or self.crack_machine.should_crack( card, discovered, confront_count ): return self._do_crack( card, player_message, trigger=card.cracks_when or "pressure from discovered evidence", forced_by_corner=False, ) ledger_claims = self.ledger.get(card.name).claims transcript_text = "\n".join( f"{e.actor or e.kind}: {e.text}" for e in transcript_tail ) guard_feedback = "" rejections = 0 for _attempt in range(self.retries + 1): draft = self._draft( card, alibi, player_message, transcript_text, ledger_claims, unlocked, guard_feedback, ) leak = self.leak_guard.check( card=card, solution=self.world.solution, draft=draft, unlocked_topics=unlocked, ) if leak.leaks: rejections += 1 guard_feedback = f"Your previous draft leaked protected info: {leak.reason}" continue new_claims = self.extractor.extract( character=card.name, utterance=draft, turn=turn, truth_context=self._truth_context(card), ) cons = self.consistency_guard.check( character=card.name, draft=draft, new_claims=new_claims, ) if cons.contradicts: rejections += 1 guard_feedback = f"Your previous draft contradicted your record: {cons.reason}" continue # Passed both guards — commit. tell = self._maybe_tell(card, new_claims) final = draft + (f"\n\n*({tell})*" if tell else "") self.ledger.append( card.name, LedgerEntry(turn=turn, raw=draft, claims=new_claims) ) return TurnOutcome( text=final, cracked=False, tell=tell, claims=new_claims, guard_rejections=rejections, ) # Logical corner: no safe+consistent reply -> crack (Section 8.1 step 6). result = self._do_crack( card, player_message, trigger="cornered: no consistent answer remained", forced_by_corner=True, ) result.guard_rejections = rejections return result # -- internals ---------------------------------------------------------- def _draft( self, card: CharacterCard, alibi: Alibi | None, player_message: str, transcript_text: str, ledger_claims: list[Claim], unlocked: list[str], guard_feedback: str, ) -> str: prompt = self.prompts.render( "character/reply.md.j2", name=card.name, role=card.role, voice=card.voice, cover=card.cover, never_admit=card.never_admit, tells=card.tells, witnessed=card.knows.witnessed, witnessed_facts=self._witnessed_facts(card), topics_known=card.knows.topics_known, topics_unknowable=card.knows.topics_unknowable, unlocked_topics=unlocked, alibi=alibi.model_dump() if alibi else None, ledger_claims=[ {"topic": c.topic, "proposition": c.proposition} for c in ledger_claims ], transcript=transcript_text, player_message=player_message, guard_feedback=guard_feedback, ) return self.client.complete( tier="character", task="character_reply", user=prompt, ).text.strip() def _truth_context(self, card: CharacterCard) -> str: return ( f"CHARACTER TRUTH (engine-only): {card.truth}\n" f"SOLUTION: culprit={self.world.solution.culprit}; " f"means={self.world.solution.means}; motive={self.world.solution.motive}; " f"opportunity={self.world.solution.opportunity}" ) def _maybe_tell(self, card: CharacterCard, new_claims: list[Claim]) -> str: """Inject a tell when the reply asserts a known-false claim on a never_admit/protected topic (Section 8.6) — reliable, learnable.""" if not card.tells: return "" lies = [c for c in new_claims if c.engine_truth_value == "false"] if not lies: return "" return random.choice(card.tells) def _do_crack( self, card: CharacterCard, player_message: str, *, trigger: str, forced_by_corner: bool, ) -> TurnOutcome: res: CrackResult = self.crack_machine.crack( card=card, player_message=player_message, trigger=trigger, forced_by_corner=forced_by_corner, ) return TurnOutcome( text=res.text, cracked=True, crack_behavior=res.behavior, )