Spaces:
Running
Running
File size: 5,007 Bytes
414dc55 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 | """Pure reducers that fold a turn + adjudication into a new GameState."""
from __future__ import annotations
from ..schemas.case import CaseFile
from ..schemas.interrogation import InterrogationTurn
from .director import Adjudication
from .game_state import Exchange, GameState, NotebookEntry, NotebookKind, SuspectState
def _fact_statement(case: CaseFile, fact_id: str) -> str:
for fact in case.facts:
if fact.fact_id == fact_id:
return fact.statement
return fact_id
def _suspect_name(case: CaseFile, sus_id: str) -> str:
try:
return case.suspect(sus_id).name
except KeyError:
return sus_id
def _next_suspect_state(
prev: SuspectState,
*,
turn_index: int,
question: str,
turn: InterrogationTurn,
adj: Adjudication,
presented_clue_id: str | None,
) -> SuspectState:
evidence_shown = prev.evidence_shown
if presented_clue_id is not None:
evidence_shown = evidence_shown | {presented_clue_id}
broken = prev.broken_lie_ids
contradictions = prev.contradictions
if adj.is_contradiction:
if adj.broken_lie_id is not None:
broken = broken | {adj.broken_lie_id}
contradictions = (*contradictions, f"Alibi contradicted on turn {turn_index}.")
return prev.model_copy(
update={
"turns": prev.turns + 1,
"stress": adj.stress_after,
"rapport": adj.rapport_after,
"transcript": (*prev.transcript, Exchange(turn_index=turn_index, question=question, answer=turn.spoken)),
"revealed_fact_ids": prev.revealed_fact_ids | set(adj.revealed_fact_ids),
"evidence_shown": evidence_shown,
"broken_lie_ids": broken,
"contradictions": contradictions,
}
)
def _notebook_entries(
case: CaseFile, sus_id: str, turn_index: int, adj: Adjudication
) -> tuple[NotebookEntry, ...]:
entries: list[NotebookEntry] = []
if adj.is_contradiction:
entries.append(
NotebookEntry(
kind=NotebookKind.CONTRADICTION,
text=f"{_suspect_name(case, sus_id)}'s alibi cracked under evidence.",
turn_index=turn_index,
suspect_id=sus_id,
)
)
for fid in adj.revealed_fact_ids:
entries.append(
NotebookEntry(
kind=NotebookKind.LEAD,
text=_fact_statement(case, fid),
turn_index=turn_index,
suspect_id=sus_id,
)
)
return tuple(entries)
def apply_turn(
state: GameState,
case: CaseFile,
sus_id: str,
question: str,
turn: InterrogationTurn,
adj: Adjudication,
presented_clue_id: str | None = None,
) -> GameState:
"""Return a new GameState with this turn folded in. Never mutates ``state``."""
turn_index = state.turn_count
prev = state.state_for(sus_id)
next_state = _next_suspect_state(
prev,
turn_index=turn_index,
question=question,
turn=turn,
adj=adj,
presented_clue_id=presented_clue_id,
)
new_states = {**state.suspect_states, sus_id: next_state}
notebook = state.notebook + _notebook_entries(case, sus_id, turn_index, adj)
return state.model_copy(
update={
"suspect_states": new_states,
"current_suspect_id": sus_id,
"turn_count": turn_index + 1,
"notebook": notebook,
}
)
def add_player_note(state: GameState, text: str) -> GameState:
"""Append a free-text note the player typed into the notebook (never mutates state)."""
text = text.strip()
if not text:
return state
note = NotebookEntry(kind=NotebookKind.NOTE, text=text[:240], turn_index=state.turn_count)
return state.model_copy(update={"notebook": (*state.notebook, note)})
def discover_clues(state: GameState, clue_ids: tuple[str, ...], case: CaseFile) -> GameState:
"""Add searched/forensic/document clues to the discovered set with notebook notes.
Unknown clue ids are ignored so a bad id can never enter the discovered set."""
seen = set(state.discovered_clue_ids)
collected: list[str] = []
for cid in clue_ids:
if cid in seen or not _clue_exists(case, cid):
continue
seen.add(cid)
collected.append(cid)
newly = tuple(collected)
if not newly:
return state
notes = tuple(
NotebookEntry(
kind=NotebookKind.CLUE,
text=case.clue(cid).name,
turn_index=state.turn_count,
clue_id=cid,
)
for cid in newly
)
return state.model_copy(
update={
"discovered_clue_ids": state.discovered_clue_ids | set(newly),
"notebook": state.notebook + notes,
}
)
def _clue_exists(case: CaseFile, clue_id: str) -> bool:
try:
case.clue(clue_id)
return True
except KeyError:
return False
|