visualnovel/state.py

"""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"