File size: 5,554 Bytes
921d377 | 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 163 164 165 166 167 168 169 170 171 172 173 | """
Scene memory — rolling context for the real-time scene planner.
PLAY-1/8. Pure, synchronous snapshot builder — no LLM calls here.
PLAY-2's scene planner will read from this module on every turn
and call ``set_synopsis`` when it decides the rolling one-liner
needs a refresh.
Persistence model:
- Character state (mood / affinity / outfit) comes from
``ix_character_state``.
- Transcript turns come from ``ix_session_turns``.
- The rolling synopsis lives in a module-level cache keyed by
session id — the same in-process pattern used by
``interaction/cooldown.py``. Short live-play sessions don't
need DB-backed continuity; if that changes, promoting this
dict to an additive DB column is a single-patch swap that
keeps the public API stable.
"""
from __future__ import annotations
from dataclasses import dataclass
from typing import Any, Dict, List, Optional, Tuple
from .. import repo, store
_DEFAULT_RECENT_N = 6
_DEFAULT_REFRESH_EVERY = 6
@dataclass(frozen=True)
class TurnSnapshot:
"""One row from ``ix_session_turns``, prompt-friendly."""
role: str # 'viewer' | 'character' | 'system'
text: str
action_id: str = ""
node_id: str = ""
created_at: str = ""
@dataclass(frozen=True)
class SceneMemory:
"""Everything the scene planner needs, in one bundle.
Frozen so the planner can't accidentally mutate the snapshot
between turns.
"""
session_id: str
experience_id: str
persona_id: str
current_node_id: str
mood: str
affinity_score: float
outfit_state: Dict[str, Any]
recent_turns: List[TurnSnapshot]
total_turns: int
synopsis: str
turns_since_synopsis: int
# ``(synopsis_text, total_turns_at_time_of_set)``
_SYNOPSIS_CACHE: Dict[str, Tuple[str, int]] = {}
def build_scene_memory(
session_id: str, *, recent_n: int = _DEFAULT_RECENT_N,
) -> Optional[SceneMemory]:
"""Assemble a snapshot or return ``None`` if the session is gone."""
sess = repo.get_session(session_id)
if not sess:
return None
cs = _fetch_character_state(session_id)
turns = _fetch_recent_turns(session_id, recent_n)
total = _count_turns(session_id)
synopsis_text, at_count = _SYNOPSIS_CACHE.get(session_id, ("", 0))
turns_since = max(0, total - at_count)
mood = str((cs.get("mood") if cs else None) or "neutral")
affinity = float((cs.get("affinity_score") if cs else None) or 0.5)
outfit = dict((cs.get("outfit_state") if cs else None) or {})
persona = str((cs.get("persona_id") if cs else None) or "")
return SceneMemory(
session_id=session_id,
experience_id=sess.experience_id,
persona_id=persona,
current_node_id=sess.current_node_id or "",
mood=mood,
affinity_score=affinity,
outfit_state=outfit,
recent_turns=turns,
total_turns=total,
synopsis=synopsis_text,
turns_since_synopsis=turns_since,
)
def set_synopsis(session_id: str, synopsis: str, at_turn_count: int) -> None:
"""Store a freshly-generated synopsis for this session."""
_SYNOPSIS_CACHE[session_id] = (synopsis, int(at_turn_count))
def should_refresh_synopsis(
memory: SceneMemory, *, every: int = _DEFAULT_REFRESH_EVERY,
) -> bool:
"""Tell the caller whether enough new turns have accumulated.
Bootstrap: the very first refresh fires at half the cadence so
the planner gets an early anchor instead of a completely
context-less first few scenes.
"""
if every <= 0:
return False
if not memory.synopsis:
return memory.total_turns >= max(1, every // 2)
return memory.turns_since_synopsis >= every
def reset_session(session_id: str) -> None:
"""Drop the synopsis cache entry for a session (useful in tests)."""
_SYNOPSIS_CACHE.pop(session_id, None)
# ── Internals ───────────────────────────────────────────────────
def _fetch_character_state(session_id: str) -> Optional[Dict[str, Any]]:
store.ensure_schema()
with store._conn() as con:
row = con.execute(
"SELECT * FROM ix_character_state WHERE session_id = ?", (session_id,),
).fetchone()
if not row:
return None
return store.row_to_dict(row, json_fields=("outfit_state", "recent_flags"))
def _fetch_recent_turns(session_id: str, n: int) -> List[TurnSnapshot]:
store.ensure_schema()
with store._conn() as con:
rows = con.execute(
"SELECT turn_role, text, action_id, node_id, created_at "
"FROM ix_session_turns WHERE session_id = ? "
"ORDER BY created_at DESC, rowid DESC LIMIT ?",
(session_id, max(1, min(n, 50))),
).fetchall()
out: List[TurnSnapshot] = []
for r in rows:
d = dict(r)
out.append(TurnSnapshot(
role=str(d.get("turn_role") or ""),
text=str(d.get("text") or ""),
action_id=str(d.get("action_id") or ""),
node_id=str(d.get("node_id") or ""),
created_at=str(d.get("created_at") or ""),
))
out.reverse() # chronological for prompt building
return out
def _count_turns(session_id: str) -> int:
store.ensure_schema()
with store._conn() as con:
row = con.execute(
"SELECT COUNT(*) AS c FROM ix_session_turns WHERE session_id = ?",
(session_id,),
).fetchone()
return int(row["c"] if row else 0)
|