feat: bot and you entity schemas with projector handlers

chat/state/entities.py

@@ -0,0 +1,54 @@
+from __future__ import annotations
+import json
+from sqlite3 import Connection
+from chat.eventlog.projector import on
+from chat.eventlog.log import Event
+
+
+@on("bot_authored")
+def _apply_bot_authored(conn: Connection, e: Event) -> None:
+    p = e.payload
+    conn.execute(
+        "INSERT OR REPLACE INTO bots "
+        "(id, name, persona, voice_samples_json, traits_json, backstory, "
+        " initial_relationship_to_you, kickoff_prose) "
+        "VALUES (?, ?, ?, ?, ?, ?, ?, ?)",
+        (p["id"], p["name"], p["persona"],
+         json.dumps(p.get("voice_samples", [])),
+         json.dumps(p.get("traits", [])),
+         p.get("backstory", ""),
+         p.get("initial_relationship_to_you", ""),
+         p.get("kickoff_prose", "")),
+    )
+
+
+@on("you_authored")
+def _apply_you_authored(conn: Connection, e: Event) -> None:
+    p = e.payload
+    conn.execute(
+        "INSERT OR REPLACE INTO you_entity (id, name, pronouns, persona) VALUES (1, ?, ?, ?)",
+        (p["name"], p.get("pronouns", ""), p.get("persona", "")),
+    )
+
+
+def get_bot(conn: Connection, bot_id: str) -> dict | None:
+    row = conn.execute("SELECT * FROM bots WHERE id = ?", (bot_id,)).fetchone()
+    if not row:
+        return None
+    cols = [c[1] for c in conn.execute("PRAGMA table_info(bots)").fetchall()]
+    d = dict(zip(cols, row))
+    d["voice_samples"] = json.loads(d.pop("voice_samples_json"))
+    d["traits"] = json.loads(d.pop("traits_json"))
+    return d
+
+
+def list_bots(conn: Connection) -> list[dict]:
+    cur = conn.execute("SELECT id, name FROM bots ORDER BY name")
+    return [{"id": r[0], "name": r[1]} for r in cur]
+
+
+def get_you(conn: Connection) -> dict | None:
+    row = conn.execute("SELECT name, pronouns, persona FROM you_entity WHERE id = 1").fetchone()
+    if not row:
+        return None
+    return {"name": row[0], "pronouns": row[1], "persona": row[2]}