File size: 7,215 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 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 202 203 204 205 206 207 208 | """
Runtime state snapshot for a session.
Pulls together current node + character state + progress metrics
in one immutable struct that the router / evaluator can reason
about without N additional SQLite reads.
``build_runtime_state(session_id)`` is the one read-heavy function
— it loads all the needed rows in one batch, then subsequent logic
is pure.
"""
from __future__ import annotations
import json
import sqlite3
from dataclasses import dataclass, field
from typing import Any, Dict, List, Optional
from .. import store
@dataclass(frozen=True)
class RuntimeState:
"""Full runtime snapshot for one session."""
session_id: str
experience_id: str
current_node_id: str
language: str
consent_version: str
personalization: Dict[str, Any]
character_mood: str
affinity_score: float
outfit_state: Dict[str, Any]
recent_flags: List[str]
progress: Dict[str, Dict[str, float]] # scheme → {metric_key: value}
uses_by_action: Dict[str, int] # action_id → uses-so-far-this-session
def _load_progress(con: sqlite3.Connection, session_id: str) -> Dict[str, Dict[str, float]]:
rows = con.execute(
"SELECT scheme, metric_key, metric_value FROM ix_session_progress "
"WHERE session_id = ?",
(session_id,),
).fetchall()
out: Dict[str, Dict[str, float]] = {}
for r in rows:
out.setdefault(str(r[0]), {})[str(r[1])] = float(r[2])
return out
def _load_uses_by_action(con: sqlite3.Connection, session_id: str) -> Dict[str, int]:
rows = con.execute(
"SELECT action_id, COUNT(*) FROM ix_session_events "
"WHERE session_id = ? AND action_id != '' "
"GROUP BY action_id",
(session_id,),
).fetchall()
return {str(r[0]): int(r[1]) for r in rows}
def build_runtime_state(session_id: str) -> Optional[RuntimeState]:
"""Single read pass that assembles a RuntimeState.
Returns None if the session doesn't exist.
"""
store.ensure_schema()
with store._conn() as con:
srow = con.execute(
"SELECT * FROM ix_sessions WHERE id = ?", (session_id,),
).fetchone()
if not srow:
return None
char_row = con.execute(
"SELECT * FROM ix_character_state WHERE session_id = ?",
(session_id,),
).fetchone()
progress = _load_progress(con, session_id)
uses = _load_uses_by_action(con, session_id)
# JSON fields
def _jparse(val: Any, default: Any) -> Any:
if val is None or val == "":
return default
if isinstance(val, (dict, list)):
return val
try:
return json.loads(val)
except (TypeError, ValueError):
return default
personalization = _jparse(srow["personalization"], {})
if char_row:
mood = str(char_row["mood"] or "neutral")
affinity = float(char_row["affinity_score"] or 0.5)
outfit = _jparse(char_row["outfit_state"], {})
flags = _jparse(char_row["recent_flags"], [])
else:
mood = "neutral"
affinity = 0.5
outfit = {}
flags = []
return RuntimeState(
session_id=str(srow["id"]),
experience_id=str(srow["experience_id"]),
current_node_id=str(srow["current_node_id"] or ""),
language=str(srow["language"] or "en"),
consent_version=str(srow["consent_version"] or ""),
personalization=personalization,
character_mood=mood,
affinity_score=affinity,
outfit_state=outfit,
recent_flags=flags,
progress=progress,
uses_by_action=uses,
)
# ─────────────────────────────────────────────────────────────────
# Mutators (used by router on hot path)
# ─────────────────────────────────────────────────────────────────
def upsert_character_state(
session_id: str, persona_id: str,
*,
mood: Optional[str] = None,
affinity_score: Optional[float] = None,
outfit_state: Optional[Dict[str, Any]] = None,
recent_flags: Optional[List[str]] = None,
language: Optional[str] = None,
) -> None:
"""Insert/update the character-state row for a session."""
store.ensure_schema()
with store._conn() as con:
existing = con.execute(
"SELECT id FROM ix_character_state WHERE session_id = ?",
(session_id,),
).fetchone()
if existing:
sets: List[str] = []
params: List[Any] = []
if mood is not None:
sets.append("mood = ?")
params.append(mood)
if affinity_score is not None:
sets.append("affinity_score = ?")
params.append(float(affinity_score))
if outfit_state is not None:
sets.append("outfit_state = ?")
params.append(store._dump_json(outfit_state))
if recent_flags is not None:
sets.append("recent_flags = ?")
params.append(store._dump_json(recent_flags))
if language is not None:
sets.append("language = ?")
params.append(language)
if not sets:
return
sets.append("updated_at = ?")
params.append(store.now_iso())
params.append(session_id)
con.execute(
f"UPDATE ix_character_state SET {', '.join(sets)} WHERE session_id = ?",
tuple(params),
)
else:
con.execute(
"""
INSERT INTO ix_character_state (
id, session_id, persona_id, mood, affinity_score,
outfit_state, recent_flags, language
) VALUES (?, ?, ?, ?, ?, ?, ?, ?)
""",
(
store.new_id("ixc"),
session_id,
persona_id,
mood or "neutral",
float(affinity_score if affinity_score is not None else 0.5),
store._dump_json(outfit_state or {}),
store._dump_json(recent_flags or []),
language or "en",
),
)
con.commit()
def upsert_progress(
session_id: str, scheme: str, metric_key: str, new_value: float,
) -> None:
"""Set a progression metric. Replaces the value (not additive)."""
store.ensure_schema()
with store._conn() as con:
con.execute(
"""
INSERT INTO ix_session_progress (id, session_id, scheme, metric_key, metric_value)
VALUES (?, ?, ?, ?, ?)
ON CONFLICT(session_id, scheme, metric_key) DO UPDATE
SET metric_value = excluded.metric_value,
updated_at = CURRENT_TIMESTAMP
""",
(store.new_id("ixp"), session_id, scheme, metric_key, float(new_value)),
)
con.commit()
|