Joseph Doherty
111 lines
4.1 KiB
Python
111 lines
4.1 KiB
Python
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", "")),
|
|
)
|
|
|
|
|
|
@on("bot_reset")
|
|
def _apply_bot_reset(conn: Connection, e: Event) -> None:
|
|
"""Purge per-bot runtime state while preserving the bot's identity row.
|
|
|
|
Wipes chats hosted by this bot (with cascading chat-scoped tables),
|
|
memories owned by this bot, edges involving this bot, and the bot's own
|
|
activity row. The ``bots`` row itself is preserved so identity,
|
|
initial-relationship, and kickoff prose remain authored.
|
|
"""
|
|
bot_id = e.payload["bot_id"]
|
|
|
|
chat_ids = [
|
|
row[0]
|
|
for row in conn.execute(
|
|
"SELECT id FROM chats WHERE host_bot_id = ?", (bot_id,)
|
|
).fetchall()
|
|
]
|
|
# T69: purge orphaned "you" activity rows pointing at containers in this
|
|
# bot's chats BEFORE the containers/chats themselves are deleted, otherwise
|
|
# the subqueries find nothing and the FK constraint on activity.container_id
|
|
# blocks the container delete.
|
|
conn.execute(
|
|
"DELETE FROM activity WHERE entity_id = 'you' "
|
|
"AND container_id IN (SELECT id FROM containers WHERE chat_id IN ("
|
|
" SELECT id FROM chats WHERE host_bot_id = ?"
|
|
"))",
|
|
(bot_id,),
|
|
)
|
|
for chat_id in chat_ids:
|
|
conn.execute("DELETE FROM scenes WHERE chat_id = ?", (chat_id,))
|
|
conn.execute("DELETE FROM containers WHERE chat_id = ?", (chat_id,))
|
|
conn.execute("DELETE FROM chat_state WHERE chat_id = ?", (chat_id,))
|
|
conn.execute("DELETE FROM chats WHERE id = ?", (chat_id,))
|
|
|
|
# Activity for this bot's entity row (independent of chat_id since the
|
|
# ``activity`` table is keyed on entity_id).
|
|
conn.execute("DELETE FROM activity WHERE entity_id = ?", (bot_id,))
|
|
|
|
# Memories authored by this bot.
|
|
conn.execute("DELETE FROM memories WHERE owner_id = ?", (bot_id,))
|
|
|
|
# Edges in either direction involving this bot.
|
|
conn.execute(
|
|
"DELETE FROM edges WHERE source_id = ? OR target_id = ?",
|
|
(bot_id, bot_id),
|
|
)
|
|
|
|
# Phase 2 cascade: clear guest references in other bots' chats so the host
|
|
# doesn't see a stale guest_bot_id pointing at this (now-purged) bot.
|
|
conn.execute(
|
|
"UPDATE chats SET guest_bot_id = NULL WHERE guest_bot_id = ?",
|
|
(bot_id,),
|
|
)
|
|
# NOTE: bots row itself is preserved (identity, kickoff_prose intact).
|
|
|
|
|
|
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]}
|