"""The ONLY place that mutates `GameState`. `apply_directives` validates and applies a `DirectorOutput` and returns a list of string "effects" (e.g. "scene_changed", "new_character:lantern_moth") so the engine knows what to repaint. The LLM proposes; this disposes. Also renders `GameState` -> the `.md` dream-memory views (a derived, human-readable mirror — NOT the source of truth). Parsing `.md` back is stubbed: a single-session game never needs it. """ from __future__ import annotations import re from pathlib import Path from . import config from .schemas import Character, DirectorOutput, GameState, Scene _CLAMP = (-100, 100) # --------------------------------------------------------------------------- # # Apply directives # --------------------------------------------------------------------------- # def apply_directives(state: GameState, out: DirectorOutput) -> list[str]: d = out.directives effects: list[str] = [] # milestone50 one-shot: a "pending" flag was injected into the context exactly once # (by memory.assemble_context, which runs before this) — retire it now. for key, val in state.flags.items(): if key.startswith("milestone50_") and val == "pending": state.flags[key] = "done" # new character arrives — handled FIRST so an arriving newcomer can be this turn's # speaker (their mood/relationship below would otherwise be silently skipped) if d.new_character: nc = d.new_character cid = _slug(nc.id) or _slug(nc.name) # always sanitize — LLM ids can contain invalid chars if cid not in state.characters: state.characters[cid] = Character( id=cid, name=nc.name, one_line=nc.one_line, appearance=nc.appearance, voice=nc.voice, traits=nc.traits, goals=nc.goals, sprite_seed=_seed_for(state.seed, cid), tts_voice_description=nc.voice, # frozen at creation; never updated ) if cid not in state.scene.present: state.scene.present.append(cid) # honour the on-stage cap if len(state.scene.present) > config.MAX_PRESENT: state.scene.present = state.scene.present[-config.MAX_PRESENT :] effects.append(f"new_character:{cid}") # speaker mood -> drives sprite variant if out.speaker in state.characters: state.characters[out.speaker].mood = out.emotion # accumulate relationship for the speaker, clamped if d.relationship_delta: ch = state.characters[out.speaker] ch.relationship = max(_CLAMP[0], min(_CLAMP[1], ch.relationship + d.relationship_delta)) # check for newly unlocked traits / secret goal effects += _check_discoveries(ch) # arm the milestone-50 special moment once per character (key absence check # so a "done" flag never re-arms) if ch.relationship >= 50 and f"milestone50_{ch.id}" not in state.flags: state.flags[f"milestone50_{ch.id}"] = "pending" effects.append(f"milestone50:{ch.id}") # the speaker learned something about the player this turn if d.remember_fact and out.speaker in state.characters: fact = d.remember_fact.strip() facts = state.characters[out.speaker].known_facts if fact and fact not in facts: facts.append(fact) del facts[:-8] # cap at 8, drop oldest effects.append(f"fact:{out.speaker}") # scene change if d.scene_change: sc = d.scene_change state.scene = Scene( id=_slug(sc.place), place=sc.place, description=sc.description, mood=sc.mood, present=state.scene.present, # keep the people on stage across a move by default backdrop_seed=_seed_for(state.seed, sc.place), ) effects.append("scene_changed") # character leaves if d.exit_character and d.exit_character in state.scene.present: state.scene.present.remove(d.exit_character) effects.append(f"exit_character:{d.exit_character}") # NPC↔NPC bond updates for bond in d.npc_relation_deltas: src, tgt = bond.source, bond.target if src in state.characters and tgt in state.characters: ch = state.characters[src] old_val = ch.npc_relations.get(tgt, 0) new_val = max(_CLAMP[0], min(_CLAMP[1], old_val + bond.delta)) ch.npc_relations[tgt] = new_val if bond.note: ch.npc_relation_notes[tgt] = bond.note effects.append(f"npc_bond:{src}:{tgt}:{new_val}") # music track switch _VALID_TRACKS = {"calm", "romantic", "dramatic", "mystery", "sad", "joyful"} if d.set_music and d.set_music in _VALID_TRACKS: state.flags["current_music"] = d.set_music effects.append(f"music:{d.set_music}") # arbitrary world flags if d.set_flags: state.flags.update(d.set_flags) effects.append("flags") # pacing — a beat lasts at least 4 turns; the model otherwise spams advance_beat # and sprints opening -> resolution in a handful of turns if d.advance_beat: started = int(state.flags.get("beat_started_turn", "0") or 0) if state.turn_index - started >= 4: state.beat = _next_beat(state.beat) state.flags["beat_started_turn"] = str(state.turn_index) # feeds the pacing nudge effects.append("beat") # ending if d.ending: state.beat = "ended" state.flags["ending_kind"] = d.ending.kind state.flags["ending_text"] = d.ending.text effects.append("ended") return effects # --------------------------------------------------------------------------- # # .md derived views (rendered after each turn for debugging + the demo) # --------------------------------------------------------------------------- # def save_memory(state: GameState) -> None: (config.MEMORY_DIR / "world_state.md").write_text(render_world_md(state), encoding="utf-8") chars_dir = config.MEMORY_DIR / "characters" chars_dir.mkdir(exist_ok=True) for ch in state.characters.values(): (chars_dir / f"{ch.id}.md").write_text(render_character_md(state, ch), encoding="utf-8") def render_world_md(state: GameState) -> str: tpl = _template("world_state.md") index = "\n".join( f"- {c.name} — {c.one_line} — {c.relationship}/100" for c in state.characters.values() ) recent = "\n".join( f"- **{t.speaker}** ({t.emotion}): {t.dialogue} ⟵ *“{t.player}”*" for t in state.recent_turns[-config.RECENT_TURNS_K :] ) flags = "\n".join(f"- {k}: {v}" for k, v in state.flags.items()) return _fill( tpl, { "seed": state.seed, "vibe": state.vibe, "beat": state.beat, "turn_index": state.turn_index, "style_guide": state.style_guide, "scene.place": state.scene.place, "scene.description": state.scene.description, "scene.mood": state.scene.mood, "scene.backdrop_seed": state.scene.backdrop_seed, "scene.present": ", ".join(state.scene.present), "summary": state.summary or "_(the tale has only just begun)_", "flags": flags or "_(none)_", "character_index": index or "_(no one yet)_", "recent_turns": recent or "_(nothing spoken yet)_", }, ) def render_character_md(state: GameState, ch: Character) -> str: tpl = _template("character_sheet.md") return _fill( tpl, { "id": ch.id, "name": ch.name, "one_line": ch.one_line, "traits": ", ".join(ch.traits), "voice": ch.voice, "goals": ch.goals, "appearance": ch.appearance, "mood": ch.mood, "relationship": ch.relationship, "sprite_seed": ch.sprite_seed, "known_facts": "\n- ".join(ch.known_facts) if ch.known_facts else "_(unknown)_", "recent_beats": "\n- ".join(ch.recent_beats) if ch.recent_beats else "_(none yet)_", }, ) def load_from_md(path: Path) -> GameState: # pragma: no cover raise NotImplementedError( "Single-session game: state lives in memory. Implement only if you wire " "HF persistent storage and want resume-on-restart." ) # --------------------------------------------------------------------------- # # helpers # --------------------------------------------------------------------------- # _BEATS = ["opening", "rising", "turn", "resolution", "ended"] def _next_beat(beat: str) -> str: i = _BEATS.index(beat) if beat in _BEATS else 0 return _BEATS[min(i + 1, len(_BEATS) - 2)] # never auto-advance into "ended" _TRAIT_THRESHOLDS = [20, 40, 60] # relationship needed to unlock trait 0, 1, 2 _GOAL_THRESHOLD = 80 # relationship needed to unlock the secret goal def _check_discoveries(ch: Character) -> list[str]: # type: ignore[name-defined] """Unlock traits/goal based on current relationship. Returns notification effects.""" new_effects: list[str] = [] for i, trait in enumerate(ch.traits): threshold = _TRAIT_THRESHOLDS[i] if i < len(_TRAIT_THRESHOLDS) else (i + 1) * 20 if ch.relationship >= threshold and i not in ch.discovered_traits: ch.discovered_traits.append(i) new_effects.append(f"unlock_trait:{ch.id}:{trait}") if ch.relationship >= _GOAL_THRESHOLD and not ch.goal_unlocked and ch.goals: ch.goal_unlocked = True new_effects.append(f"unlock_goal:{ch.id}") return new_effects def _slug(text: str) -> str: return "".join(c if c.isalnum() else "_" for c in text.lower()).strip("_")[:40] or "x" def _seed_for(world_seed: int, key: str) -> int: return (world_seed * 2654435761 + sum(ord(c) for c in key) * 40503) % (2**31) def _template(name: str) -> str: return (config.TEMPLATES_DIR / name).read_text(encoding="utf-8") def _fill(tpl: str, values: dict) -> str: out = tpl for k, v in values.items(): out = out.replace("{{" + k + "}}", str(v)) # drop the template's own HTML author-comments (incl. the {{double_braces}} legend) out = re.sub(r"", "", out, flags=re.DOTALL) return out.strip() + "\n"