File size: 5,958 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 | """
Aggregate session events into compact analytics summaries.
Two scopes:
session_summary one session — turn count, per-action
use counts, final node, mood/affinity end
state, per-scheme progress.
experience_summary all sessions of one experience —
completion rate, total turns, popular
actions, action block rate.
No percentile / histogram work here — that belongs in the studio
dashboard if / when it needs it. These summaries are the minimum
viable signal for the ship-readiness review.
"""
from __future__ import annotations
import json
from collections import Counter
from dataclasses import dataclass
from typing import Any, Dict, List, Optional
from .. import store
@dataclass(frozen=True)
class SessionSummary:
"""Per-session analytic snapshot."""
session_id: str
experience_id: str
turns: int
events: int
action_uses: Dict[str, int]
decisions: Dict[str, int] # 'allow' | 'block' → count
intents: Dict[str, int]
final_node_id: str
final_mood: str
final_affinity: float
progress: Dict[str, Dict[str, float]]
completed: bool
@dataclass(frozen=True)
class ExperienceSummary:
"""Aggregate across every session of one experience."""
experience_id: str
session_count: int
completed_sessions: int
completion_rate: float
total_turns: int
total_events: int
popular_actions: List[Dict[str, Any]] # [{'action_id': ..., 'uses': int}, ...]
block_rate: float # fraction of turns blocked by policy
def _parse_payload(raw: Any) -> Dict[str, Any]:
if not raw:
return {}
if isinstance(raw, dict):
return raw
try:
return json.loads(raw)
except (TypeError, ValueError):
return {}
def session_summary(session_id: str) -> Optional[SessionSummary]:
"""Compute the summary for one session. Returns None if missing."""
store.ensure_schema()
with store._conn() as con:
sess = con.execute(
"SELECT * FROM ix_sessions WHERE id = ?", (session_id,),
).fetchone()
if not sess:
return None
turns = con.execute(
"SELECT COUNT(*) FROM ix_session_turns WHERE session_id = ?", (session_id,),
).fetchone()[0]
events = con.execute(
"SELECT event_kind, payload, action_id FROM ix_session_events WHERE session_id = ?",
(session_id,),
).fetchall()
cs = con.execute(
"SELECT * FROM ix_character_state WHERE session_id = ?", (session_id,),
).fetchone()
progress_rows = con.execute(
"SELECT scheme, metric_key, metric_value FROM ix_session_progress "
"WHERE session_id = ?",
(session_id,),
).fetchall()
action_uses: Counter = Counter()
decisions: Counter = Counter()
intents: Counter = Counter()
for ev in events:
if ev["event_kind"] != "turn_resolved":
continue
if ev["action_id"]:
action_uses[ev["action_id"]] += 1
payload = _parse_payload(ev["payload"])
if "decision" in payload:
decisions[str(payload["decision"])] += 1
if "intent_code" in payload and payload["intent_code"]:
intents[str(payload["intent_code"])] += 1
progress: Dict[str, Dict[str, float]] = {}
for row in progress_rows:
progress.setdefault(row["scheme"], {})[row["metric_key"]] = float(row["metric_value"])
mood = cs["mood"] if cs else "neutral"
aff = float(cs["affinity_score"]) if cs else 0.5
return SessionSummary(
session_id=session_id,
experience_id=str(sess["experience_id"]),
turns=int(turns),
events=len(events),
action_uses=dict(action_uses),
decisions=dict(decisions),
intents=dict(intents),
final_node_id=str(sess["current_node_id"] or ""),
final_mood=str(mood),
final_affinity=aff,
progress=progress,
completed=sess["completed_at"] is not None,
)
def experience_summary(experience_id: str) -> ExperienceSummary:
"""Aggregate across every session of ``experience_id``."""
store.ensure_schema()
with store._conn() as con:
sessions = con.execute(
"SELECT id, completed_at FROM ix_sessions WHERE experience_id = ?",
(experience_id,),
).fetchall()
total_turns = con.execute(
"SELECT COUNT(*) FROM ix_session_turns t "
"JOIN ix_sessions s ON t.session_id = s.id "
"WHERE s.experience_id = ?",
(experience_id,),
).fetchone()[0]
events = con.execute(
"SELECT e.payload, e.action_id FROM ix_session_events e "
"JOIN ix_sessions s ON e.session_id = s.id "
"WHERE s.experience_id = ? AND e.event_kind = 'turn_resolved'",
(experience_id,),
).fetchall()
sc = len(sessions)
completed = sum(1 for s in sessions if s["completed_at"] is not None)
action_uses: Counter = Counter()
decisions: Counter = Counter()
for ev in events:
if ev["action_id"]:
action_uses[ev["action_id"]] += 1
payload = _parse_payload(ev["payload"])
if "decision" in payload:
decisions[str(payload["decision"])] += 1
popular = [{"action_id": aid, "uses": n} for aid, n in action_uses.most_common(10)]
total_decisions = sum(decisions.values())
block_rate = 0.0
if total_decisions > 0:
block_rate = decisions.get("block", 0) / total_decisions
return ExperienceSummary(
experience_id=experience_id,
session_count=sc,
completed_sessions=completed,
completion_rate=(completed / sc) if sc > 0 else 0.0,
total_turns=int(total_turns),
total_events=len(events),
popular_actions=popular,
block_rate=block_rate,
)
|