3 Commits

Author SHA1 Message Date
Joseph Doherty 82be8b3f51 feat: bot reset with hard confirm and event-driven purge 2026-04-26 14:07:56 -04:00
Joseph Doherty 46062973c2 feat: regenerate with edit-then-regenerate inline UX 2026-04-26 14:04:02 -04:00
Joseph Doherty aa0563b4fa feat: rewind with impact preview, pre-rewind snapshot, undo toast 2026-04-26 13:58:20 -04:00
11 changed files with 1338 additions and 2 deletions
+282
View File
@@ -0,0 +1,282 @@
"""Regenerate flow (T29).
The user clicks "Regenerate" on the latest ``assistant_turn``. The UI
puts the prior ``user_turn`` into inline edit mode and submits to
:func:`regenerate_assistant_turn` either:
- with **no edit** — we re-run the narrative against the original user
prose and append a fresh ``assistant_turn`` superseding the old one;
- with **edited prose** — we additionally append a ``user_turn_edit``
event capturing the new prose, mark the original ``user_turn`` as
superseded by the edit, then run the narrative against the edited
prose.
Per Requirements §10.2 superseded events are *kept in the log* — the
display layer hides them. This is what makes rewinding to before a
regenerate cheap: we just clear ``superseded_by`` on the old row.
The supersede update is one of the rare "direct DB write" exceptions
documented in the plan: we manipulate metadata fields on the canonical
event_log row itself rather than projecting through a handler.
Phase 1 simplifications (per the plan's "bound it" guidance):
- Significance pass is *not* re-run on regenerate. The original score
remains attached to the prior memory. The state-update pass *is* re-run
so affinity/trust/knowledge reflect the new output.
- The route does not broadcast a fresh ``turn_html`` SSE event; T34
polishes UI swaps. The user refreshes the page to see the new turn.
"""
from __future__ import annotations
import json
from sqlite3 import Connection
from chat.config import Settings
from chat.eventlog.log import append_and_apply, append_event
from chat.services.memory_write import record_turn_memory
from chat.services.prompt import assemble_narrative_prompt
from chat.services.state_update import compute_state_update
from chat.state.edges import get_edge
from chat.state.entities import get_bot, get_you
from chat.state.world import active_scene, get_chat
from chat.web.pubsub import publish
async def regenerate_assistant_turn(
conn: Connection,
client,
*,
settings: Settings,
chat_id: str,
original_assistant_event_id: int,
edited_user_prose: str | None = None,
) -> str:
"""Regenerate the assistant turn linked to ``original_assistant_event_id``.
When ``edited_user_prose`` is provided the original user_turn is also
superseded by a fresh ``user_turn_edit`` event capturing the new
prose. Returns the new assistant text.
Raises :class:`ValueError` when the chat or the assistant_turn event
cannot be found — the FastAPI route translates this to 404.
"""
chat = get_chat(conn, chat_id)
if chat is None:
raise ValueError("chat not found")
host_bot_id = chat["host_bot_id"]
host_bot = get_bot(conn, host_bot_id) or {
"id": host_bot_id,
"name": "bot",
"persona": "",
}
# 1. Locate the original assistant_turn event.
row = conn.execute(
"SELECT payload_json FROM event_log "
"WHERE id = ? AND kind = 'assistant_turn'",
(original_assistant_event_id,),
).fetchone()
if row is None:
raise ValueError("assistant_turn event not found")
original_assistant_payload = json.loads(row[0])
original_user_turn_id = original_assistant_payload.get("user_turn_id")
# 2. Determine the prose for the new prompt and (when edited) capture
# the user_turn_edit event up front so the new event ids exist before
# we link them from the assistant_turn payload.
new_user_event_id: int | None = None
if edited_user_prose is not None:
new_user_event_id = append_event(
conn,
kind="user_turn_edit",
payload={
"chat_id": chat_id,
"prose": edited_user_prose,
"supersedes_user_turn_id": original_user_turn_id,
},
)
if original_user_turn_id is not None:
conn.execute(
"UPDATE event_log SET superseded_by = ? WHERE id = ?",
(new_user_event_id, original_user_turn_id),
)
prose_for_prompt = edited_user_prose
else:
original_user_row = conn.execute(
"SELECT payload_json FROM event_log WHERE id = ?",
(original_user_turn_id,),
).fetchone() if original_user_turn_id is not None else None
if original_user_row is not None:
prose_for_prompt = json.loads(original_user_row[0]).get("prose", "")
else:
prose_for_prompt = ""
# 3. Build the recent-dialogue slice. Exclude the original
# assistant_turn explicitly (we haven't superseded it yet — that
# update lands at the end so the new event_id is known) and use the
# standard ``superseded_by IS NULL AND hidden = 0`` filter so any
# prior regenerates also drop out.
you_entity = get_you(conn) or {"name": "you", "persona": ""}
you_name = you_entity.get("name", "you")
cur = conn.execute(
"SELECT id, kind, payload_json FROM event_log "
"WHERE kind IN ('user_turn', 'user_turn_edit', 'assistant_turn') "
" AND id != ? "
" AND superseded_by IS NULL AND hidden = 0 "
"ORDER BY id DESC LIMIT 20",
(original_assistant_event_id,),
)
rows = list(reversed(cur.fetchall()))
recent: list[dict] = []
for _eid, kind, payload_json in rows:
p = json.loads(payload_json)
if p.get("chat_id") != chat_id:
continue
if kind in ("user_turn", "user_turn_edit"):
recent.append({"speaker": you_name, "text": p.get("prose", "")})
else:
recent.append(
{"speaker": host_bot.get("name", "bot"), "text": p.get("text", "")}
)
# 4. Assemble the narrative prompt. ``recent`` already excludes the
# current user prose, which we pass through ``user_turn_prose``.
messages = assemble_narrative_prompt(
conn,
chat_id=chat_id,
speaker_bot_id=host_bot_id,
user_turn_prose=prose_for_prompt or None,
recent_dialogue=recent,
budget_soft=settings.narrative_budget_soft,
budget_hard=settings.narrative_budget_hard,
)
# 5. Stream the new narrative.
accumulated: list[str] = []
async for chunk in client.stream(
messages, model=settings.narrative_model
):
accumulated.append(chunk)
await publish(
chat_id,
{"event": "token", "text": chunk, "speaker_id": host_bot_id},
)
new_text = "".join(accumulated)
# 6. Append the new assistant_turn event. ``user_turn_id`` points at
# the edit event when one was created, otherwise the original. The
# ``regenerated_from`` field is the back-pointer the UI uses for an
# "originally said …" affordance.
new_assistant_event_id = append_event(
conn,
kind="assistant_turn",
payload={
"chat_id": chat_id,
"speaker_id": host_bot_id,
"text": new_text,
"truncated": False,
"user_turn_id": (
new_user_event_id
if new_user_event_id is not None
else original_user_turn_id
),
"regenerated_from": original_assistant_event_id,
},
)
# 7. Mark the original assistant_turn as superseded by the new one.
conn.execute(
"UPDATE event_log SET superseded_by = ? WHERE id = ?",
(new_assistant_event_id, original_assistant_event_id),
)
# 8. Re-run downstream classifier passes (memory write + state update
# for both directed edges). Significance is intentionally skipped on
# regenerate (the prior score remains attached to the prior memory).
scene = active_scene(conn, chat_id)
record_turn_memory(
conn,
chat_id=chat_id,
host_bot_id=host_bot_id,
narrative_text=new_text,
scene_id=scene["id"] if scene else None,
chat_clock_at=chat.get("time"),
)
last_at = chat.get("time")
recent_for_update = recent + [
{"speaker": host_bot.get("name", "bot"), "text": new_text}
]
edge_b2y = get_edge(conn, host_bot_id, "you") or {
"affinity": 50,
"trust": 50,
"summary": "",
}
update_b2y = await compute_state_update(
client,
model=settings.classifier_model,
source_id=host_bot_id,
target_id="you",
source_name=host_bot.get("name", "bot"),
source_persona=host_bot.get("persona", "") or "",
target_name=you_name,
prior_affinity=edge_b2y["affinity"],
prior_trust=edge_b2y["trust"],
prior_summary=edge_b2y.get("summary", "") or "",
recent_dialogue=recent_for_update,
)
append_and_apply(
conn,
kind="edge_update",
payload={
"source_id": host_bot_id,
"target_id": "you",
"chat_id": chat_id,
"affinity_delta": update_b2y.affinity_delta,
"trust_delta": update_b2y.trust_delta,
"knowledge_facts": update_b2y.knowledge_facts,
"last_interaction_at": last_at,
"last_interaction_chat_id": chat_id,
},
)
edge_y2b = get_edge(conn, "you", host_bot_id) or {
"affinity": 50,
"trust": 50,
"summary": "",
}
update_y2b = await compute_state_update(
client,
model=settings.classifier_model,
source_id="you",
target_id=host_bot_id,
source_name=you_name,
source_persona=you_entity.get("persona", "") or "",
target_name=host_bot.get("name", "bot"),
prior_affinity=edge_y2b["affinity"],
prior_trust=edge_y2b["trust"],
prior_summary=edge_y2b.get("summary", "") or "",
recent_dialogue=recent_for_update,
)
append_and_apply(
conn,
kind="edge_update",
payload={
"source_id": "you",
"target_id": host_bot_id,
"chat_id": chat_id,
"affinity_delta": update_y2b.affinity_delta,
"trust_delta": update_y2b.trust_delta,
"knowledge_facts": update_y2b.knowledge_facts,
"last_interaction_at": last_at,
"last_interaction_chat_id": chat_id,
},
)
return new_text
__all__ = ["regenerate_assistant_turn"]
+23
View File
@@ -0,0 +1,23 @@
from __future__ import annotations
from sqlite3 import Connection
from chat.eventlog.log import append_and_apply
from chat.state.entities import get_bot
def reset_bot(conn: Connection, bot_id: str, *, confirm_name: str) -> None:
"""Reset a bot's runtime state via a ``bot_reset`` event.
Validates that ``confirm_name`` matches the bot's stored ``name``
exactly (case-sensitive, no trim). Raises:
- ``ValueError("bot {bot_id} not found")`` when the bot is missing.
- ``ValueError("confirm_name does not match bot name")`` on mismatch.
"""
bot = get_bot(conn, bot_id)
if bot is None:
raise ValueError(f"bot {bot_id} not found")
if confirm_name != bot["name"]:
raise ValueError("confirm_name does not match bot name")
append_and_apply(conn, kind="bot_reset", payload={"bot_id": bot_id})
+112
View File
@@ -0,0 +1,112 @@
"""Rewind service — truncate the event log past a chosen turn and re-project.
Per Requirements §10.1 and Plan Task 28, "rewind to here" must:
1. Take a snapshot of the current state so the user can recover (handed
off to :mod:`chat.services.snapshot`).
2. Truncate the event log past ``after_event_id`` — physical DELETE for
v1 simplicity; the spec says rewind should be a hard truncation, not
the soft ``hidden=1`` mechanism used by edits/regenerate.
3. Clear projected tables and re-project from the truncated log so live
state matches "what the world looked like at turn N". Without the
re-projection, projected tables would carry forward stale rows from
rewound events (e.g. an ``edge_update`` that bumped affinity past the
rewind point would still show in ``edges``).
Re-projection is a full replay rather than a "revert delta" because most
projector handlers are idempotent inserts, but the edge handler is a
delta-shaped accumulator — there's no clean way to invert a single
``edge_update`` against ``edges.affinity`` without replay. Wiping +
replaying is straightforward and correct.
"""
from __future__ import annotations
from pathlib import Path
from sqlite3 import Connection
from chat.db.connection import open_db
from chat.eventlog.projector import project
from chat.services.snapshot import take_snapshot
def compute_rewind_preview(
conn: Connection, after_event_id: int
) -> dict:
"""Return counts of each event kind that would be removed by rewinding.
Used by the preview modal so the user sees the impact (e.g. "this
will remove 8 events: 4 user_turn, 4 assistant_turn") before
confirming. Counts include hidden/superseded rows — they're still
physically deleted.
"""
cur = conn.execute(
"SELECT kind, COUNT(*) FROM event_log WHERE id > ? GROUP BY kind "
"ORDER BY kind",
(after_event_id,),
)
counts = {kind: count for kind, count in cur.fetchall()}
total = sum(counts.values())
return {
"after_event_id": after_event_id,
"total_events": total,
"by_kind": counts,
}
def execute_rewind(
*, db_path: Path, data_dir: Path, after_event_id: int
) -> Path:
"""Take a snapshot, truncate, and re-project. Returns the snapshot path.
The snapshot is taken inside the same connection scope as the
truncate + reproject so all three commit together — if any step
fails the connection's commit-on-exit is bypassed by the exception
and the database stays untouched. The snapshot file is on disk
regardless, which is the desired behaviour: even if the truncate
aborts, the user has a recovery point.
"""
with open_db(db_path) as conn:
# 1. Snapshot first — we want this on disk before any destructive
# operation runs.
snapshot_path = take_snapshot(
conn, data_dir=data_dir, kind="rewind"
)
# 2. Truncate the event log past the chosen id. Foreign keys are
# ON, but ``event_log.superseded_by`` self-references and the
# rows we're deleting are the only ones that could point
# forward — there's nothing to cascade.
conn.execute(
"DELETE FROM event_log WHERE id > ?", (after_event_id,)
)
# 3. Clear projected tables in topological order so FK ON DELETE
# constraints don't fire on referenced rows. ``activity`` and
# ``scenes`` reference ``containers``; ``chat_state`` references
# ``chats`` by id-convention only (no FK declared). ``memories``,
# ``edges``, ``bots``, ``you_entity``, and ``classifier_failures``
# have no incoming FKs from other projected tables.
#
# ``executescript`` is intentionally avoided so foreign_keys=ON
# stays in effect for each statement — executescript would
# implicitly commit and reset some pragmas on certain SQLite
# builds.
conn.execute("DELETE FROM memories")
conn.execute("DELETE FROM activity")
conn.execute("DELETE FROM scenes")
conn.execute("DELETE FROM containers")
conn.execute("DELETE FROM chat_state")
conn.execute("DELETE FROM chats")
conn.execute("DELETE FROM edges")
conn.execute("DELETE FROM bots")
conn.execute("DELETE FROM you_entity")
conn.execute("DELETE FROM classifier_failures")
# 4. Re-project from the truncated event log. Handler registry
# is module-level state populated by importing chat.state.* —
# callers (the route, tests) need to have those modules
# imported for this to do anything useful.
project(conn)
return snapshot_path
+100
View File
@@ -0,0 +1,100 @@
"""Snapshot service — write a JSON dump of all projected tables to disk.
Used by the rewind flow (Requirements §10.1, T28) so the user can recover a
pre-rewind state if the rewind was a mistake. Stored under
``data/snapshots/{kind}/`` with a UTC timestamp filename.
The dump captures both the event log (so the original event sequence is
preserved verbatim) and every projected table (so a future restore could
either re-load tables directly or re-project from the saved event log).
The FTS shadow table ``memories_fts`` is intentionally skipped — it's a
virtual table maintained by the ``memories_ai/au/ad`` triggers, so it would
rebuild itself on a memories re-load. Snapshotting it would also fail
``PRAGMA table_info`` cleanly since FTS5 reports its columns differently.
"""
from __future__ import annotations
import json
from datetime import datetime, timezone
from pathlib import Path
from sqlite3 import Connection
# Order doesn't affect correctness for snapshotting (we read, not write),
# but listing tables explicitly keeps the snapshot stable across schema
# evolution: a new table won't silently change the dump shape until it's
# added here.
PROJECTED_TABLES = [
"bots",
"you_entity",
"edges",
"memories",
"memories_fts",
"chats",
"chat_state",
"containers",
"scenes",
"activity",
"classifier_failures",
]
def take_snapshot(
conn: Connection, *, data_dir: Path, kind: str = "rewind"
) -> Path:
"""Write a JSON dump of the event log and projected tables.
Returns the path to the written snapshot file. Creates parent
directories as needed. Filename is a UTC timestamp in
``YYYYMMDDTHHMMSSZ`` form so chronological listing matches creation
order.
"""
snapshot_dir = data_dir / "snapshots" / kind
snapshot_dir.mkdir(parents=True, exist_ok=True)
timestamp = datetime.now(timezone.utc).strftime("%Y%m%dT%H%M%SZ")
path = snapshot_dir / f"{timestamp}.json"
dump: dict[str, list] = {}
# Event log: pull every column we care about. ``ts`` and the
# superseded/hidden flags are needed to faithfully reconstruct the
# log on restore.
cur = conn.execute(
"SELECT id, branch_id, ts, kind, payload_json, superseded_by, hidden "
"FROM event_log ORDER BY id"
)
dump["event_log"] = [
{
"id": r[0],
"branch_id": r[1],
"ts": r[2],
"kind": r[3],
"payload_json": r[4],
"superseded_by": r[5],
"hidden": r[6],
}
for r in cur.fetchall()
]
for table in PROJECTED_TABLES:
if table == "memories_fts":
# Virtual FTS5 table — rebuilt by triggers on insert, no need
# to snapshot it (and ``PRAGMA table_info`` reports its
# columns differently).
continue
cur = conn.execute(f"PRAGMA table_info({table})")
cols = [c[1] for c in cur.fetchall()]
if not cols:
# Table not present in this schema version — leave an empty
# list rather than raising, so older snapshots can survive.
dump[table] = []
continue
cur = conn.execute(f"SELECT {', '.join(cols)} FROM {table}")
dump[table] = [dict(zip(cols, row)) for row in cur.fetchall()]
# ``default=str`` covers Path-like or datetime values that might
# sneak through if a column ever stored them; the projected tables
# all use TEXT so this is mostly defensive.
path.write_text(json.dumps(dump, default=str))
return path
+40
View File
@@ -31,6 +31,46 @@ def _apply_you_authored(conn: Connection, e: Event) -> None:
)
@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()
]
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),
)
# NOTE: bots row itself is preserved (identity, kickoff_prose intact).
# NOTE: "you" activity (entity_id="you") may linger from a deleted chat;
# acceptable for v1 — Phase 1.5 cleanup if needed.
def get_bot(conn: Connection, bot_id: str) -> dict | None:
row = conn.execute("SELECT * FROM bots WHERE id = ?", (bot_id,)).fetchone()
if not row:
+12 -1
View File
@@ -8,7 +8,18 @@
{% if bots %}
<ul class="bot-list">
{% for bot in bots %}
<li><a href="/bots/{{ bot.id }}">{{ bot.name }}</a></li>
<li>
<a href="/bots/{{ bot.id }}">{{ bot.name }}</a>
<details class="bot-row-reset">
<summary>Reset</summary>
<form method="post" action="/bots/{{ bot.id }}/reset" class="inline-edit">
<label>Type "{{ bot.name }}" to confirm:
<input type="text" name="confirm_name" required>
</label>
<button type="submit">Reset bot</button>
</form>
</details>
</li>
{% endfor %}
</ul>
{% else %}
+19
View File
@@ -118,3 +118,22 @@ async def bot_create(
append_event(conn, kind="bot_authored", payload=payload)
project(conn)
return RedirectResponse(url=f"/bots/{payload['id']}/kickoff", status_code=303)
@router.post("/bots/{bot_id}/reset")
async def reset_bot_route(
bot_id: str,
request: Request,
confirm_name: str = Form(""),
conn=Depends(get_conn),
):
from chat.services.reset import reset_bot
try:
reset_bot(conn, bot_id, confirm_name=confirm_name)
except ValueError as e:
msg = str(e).lower()
if "not found" in msg:
raise HTTPException(status_code=404, detail=str(e))
raise HTTPException(status_code=400, detail=str(e))
return RedirectResponse(url="/bots", status_code=303)
+128 -1
View File
@@ -36,12 +36,13 @@ import html
import json
from fastapi import APIRouter, Depends, Form, HTTPException, Request
from fastapi.responses import Response
from fastapi.responses import HTMLResponse, RedirectResponse, Response
from chat.eventlog.log import append_and_apply, append_event
from chat.services.background import SignificanceJob
from chat.services.memory_write import record_turn_memory
from chat.services.prompt import assemble_narrative_prompt
from chat.services.rewind import compute_rewind_preview, execute_rewind
from chat.services.scene_close import detect_scene_close
from chat.services.scene_summarize import apply_scene_close_summary
from chat.services.state_update import compute_state_update
@@ -409,3 +410,129 @@ async def post_turn(
raise asyncio.CancelledError
return Response(status_code=204)
# ---------------------------------------------------------------------------
# Rewind routes (Task 28).
#
# Two endpoints: a GET that renders the impact-preview modal, and a POST
# that actually executes the rewind. The execution path opens its own
# database connection because the route's ``conn`` is closed when the
# dependency-injection scope exits — passing it to ``execute_rewind``
# would dangle.
# ---------------------------------------------------------------------------
@router.get(
"/chats/{chat_id}/rewind/preview/{event_id}",
response_class=HTMLResponse,
)
async def rewind_preview(
chat_id: str,
event_id: int,
request: Request,
conn=Depends(get_conn),
):
"""Render the rewind impact-preview modal as a small HTML fragment.
The HTMX form inside the fragment posts to the execute endpoint
below. v1 keeps the markup minimal — Task 35 polishes the modal.
"""
chat = get_chat(conn, chat_id)
if chat is None:
raise HTTPException(status_code=404, detail=f"chat not found: {chat_id}")
preview = compute_rewind_preview(conn, event_id)
items = "".join(
f"<li>{count} × {html.escape(kind)}</li>"
for kind, count in preview["by_kind"].items()
)
body = (
"<div class='rewind-modal'>"
f"<h3>Rewind to event {event_id}?</h3>"
f"<p>This will remove {preview['total_events']} events:</p>"
f"<ul>{items}</ul>"
f"<form hx-post='/chats/{html.escape(chat_id)}/rewind/{event_id}' "
"hx-target='body' hx-swap='innerHTML'>"
"<button type='submit'>Confirm Rewind</button>"
"</form>"
"</div>"
)
return HTMLResponse(body)
# ---------------------------------------------------------------------------
# Regenerate route (Task 29).
#
# A POST that re-streams the most recent assistant turn. The prior
# ``assistant_turn`` event is kept in the log but flagged
# ``superseded_by`` so the timeline filter in :func:`_read_recent_dialogue`
# hides it. When the user supplies ``prose`` the original ``user_turn``
# is also superseded by a fresh ``user_turn_edit`` event capturing the
# edit. Significance is *not* re-run on regenerate (per plan §11.1) but
# state-update + memory writes are.
# ---------------------------------------------------------------------------
@router.post("/chats/{chat_id}/turns/{event_id}/regenerate")
async def regenerate_turn(
chat_id: str,
event_id: int,
request: Request,
prose: str | None = Form(None),
conn=Depends(get_conn),
client=Depends(get_llm_client),
):
"""Regenerate the assistant turn referenced by ``event_id``.
``prose`` is optional. When provided (and non-empty) we capture a
``user_turn_edit`` event before re-streaming. Returns 204 on
success, 404 when the chat or assistant_turn event is missing. The
SSE channel emits per-token events as the new text arrives.
"""
chat = get_chat(conn, chat_id)
if chat is None:
raise HTTPException(status_code=404, detail=f"chat not found: {chat_id}")
settings = request.app.state.settings
# Local import keeps the module import graph flat (the service
# imports from ``state`` / ``services`` siblings already).
from chat.services.regenerate import regenerate_assistant_turn
edited_prose = prose if prose else None
try:
await regenerate_assistant_turn(
conn,
client,
settings=settings,
chat_id=chat_id,
original_assistant_event_id=event_id,
edited_user_prose=edited_prose,
)
except ValueError as e:
raise HTTPException(status_code=404, detail=str(e))
return Response(status_code=204)
@router.post("/chats/{chat_id}/rewind/{event_id}")
async def rewind_execute(
chat_id: str,
event_id: int,
request: Request,
conn=Depends(get_conn),
):
"""Execute the rewind: snapshot, truncate event_log, re-project.
Note: ``conn`` is only used to validate the chat exists. The actual
rewind opens its own connection inside ``execute_rewind`` because
we need it to commit independently and survive the route's
dependency teardown.
"""
chat = get_chat(conn, chat_id)
if chat is None:
raise HTTPException(status_code=404, detail=f"chat not found: {chat_id}")
settings = request.app.state.settings
execute_rewind(
db_path=settings.db_path,
data_dir=settings.data_dir,
after_event_id=event_id,
)
return RedirectResponse(url=f"/chats/{chat_id}", status_code=303)
+273
View File
@@ -0,0 +1,273 @@
"""Regenerate flow (T29).
POST ``/chats/<chat_id>/turns/<event_id>/regenerate`` re-streams the
assistant turn, supersedes the prior ``assistant_turn`` event, and — when
prose is supplied — captures a ``user_turn_edit`` event that supersedes
the original ``user_turn``.
These tests cover the functional core required by the plan:
- Without edit: a new ``assistant_turn`` is appended; the original is
marked ``superseded_by`` the new one.
- With edit: a ``user_turn_edit`` event is appended; the original
``user_turn`` is also marked ``superseded_by``.
- Missing event id returns 404.
"""
from __future__ import annotations
import json
import pytest
from fastapi.testclient import TestClient
from chat.app import app
from chat.db.connection import open_db
from chat.eventlog.log import append_event
from chat.eventlog.projector import project
from chat.llm.mock import MockLLMClient
@pytest.fixture
def client(tmp_path, monkeypatch):
cfg = tmp_path / "config.toml"
cfg.write_text('featherless_api_key = "test"\n')
monkeypatch.setenv("CHAT_CONFIG_PATH", str(cfg))
db = tmp_path / "test.db"
monkeypatch.setenv("CHAT_DB_PATH", str(db))
with TestClient(app) as c:
# Disable lifespan-managed background worker (would otherwise try
# to score significance through Featherless with the test key).
if hasattr(app.state, "background_worker"):
app.state.background_worker.enabled = False
yield c
app.dependency_overrides.clear()
def _seed_with_one_turn(db_path):
"""Seed bot, chat, edges/activity, and ONE round of user_turn + assistant_turn.
Returns ``(user_turn_event_id, assistant_turn_event_id)``.
"""
with open_db(db_path) as conn:
append_event(
conn,
kind="bot_authored",
payload={
"id": "bot_a",
"name": "BotA",
"persona": "thoughtful",
"voice_samples": [],
"traits": [],
"backstory": "",
"initial_relationship_to_you": "",
"kickoff_prose": "",
},
)
append_event(
conn,
kind="chat_created",
payload={
"id": "chat_bot_a",
"host_bot_id": "bot_a",
"initial_time": "2026-04-26T20:00:00+00:00",
"narrative_anchor": "Day 1",
"weather": "",
},
)
append_event(
conn,
kind="edge_update",
payload={
"source_id": "bot_a",
"target_id": "you",
"chat_id": "chat_bot_a",
},
)
append_event(
conn,
kind="edge_update",
payload={
"source_id": "you",
"target_id": "bot_a",
"chat_id": "chat_bot_a",
},
)
append_event(
conn,
kind="activity_change",
payload={
"entity_id": "you",
"posture": "sitting",
"action": {"verb": "talking"},
"attention": "",
"holding": [],
"status": {},
},
)
append_event(
conn,
kind="activity_change",
payload={
"entity_id": "bot_a",
"posture": "sitting",
"action": {"verb": "listening"},
"attention": "",
"holding": [],
"status": {},
},
)
# First round: user_turn + assistant_turn.
ut_id = append_event(
conn,
kind="user_turn",
payload={
"chat_id": "chat_bot_a",
"prose": "hello",
"segments": [],
},
)
at_id = append_event(
conn,
kind="assistant_turn",
payload={
"chat_id": "chat_bot_a",
"speaker_id": "bot_a",
"text": "Original response.",
"truncated": False,
"user_turn_id": ut_id,
},
)
project(conn)
return ut_id, at_id
def test_regenerate_without_edit_creates_new_assistant_turn(client, tmp_path):
"""Reissuing the regenerate POST with no prose should:
- Stream a new ``assistant_turn`` carrying ``regenerated_from`` and
the canned narrative text.
- Mark the original ``assistant_turn`` row as ``superseded_by`` the
new one.
"""
ut_id, at_id = _seed_with_one_turn(tmp_path / "test.db")
narrative_canned = "New response."
state_canned = json.dumps(
{"affinity_delta": 0, "trust_delta": 0, "knowledge_facts": []}
)
canned = [narrative_canned, state_canned, state_canned]
from chat.web.kickoff import get_llm_client
app.dependency_overrides[get_llm_client] = lambda: MockLLMClient(
canned=list(canned)
)
try:
response = client.post(
f"/chats/chat_bot_a/turns/{at_id}/regenerate", data={}
)
assert response.status_code == 204
finally:
app.dependency_overrides.clear()
with open_db(tmp_path / "test.db") as conn:
# Original assistant_turn is now superseded.
row = conn.execute(
"SELECT superseded_by FROM event_log WHERE id = ?", (at_id,)
).fetchone()
assert row[0] is not None
# A new assistant_turn exists, links back to the original, and
# carries the canned narrative text.
cur = conn.execute(
"SELECT id, payload_json FROM event_log "
"WHERE kind = 'assistant_turn' AND id != ? "
"AND superseded_by IS NULL",
(at_id,),
).fetchall()
assert len(cur) == 1
new_id, new_payload_json = cur[0]
new_payload = json.loads(new_payload_json)
assert new_payload["text"] == "New response."
assert new_payload["regenerated_from"] == at_id
# The original assistant_turn's superseded_by points at the new one.
assert row[0] == new_id
# The original user_turn is NOT touched when no prose was supplied.
ut_row = conn.execute(
"SELECT superseded_by FROM event_log WHERE id = ?", (ut_id,)
).fetchone()
assert ut_row[0] is None
def test_regenerate_with_edit_appends_user_turn_edit(client, tmp_path):
"""Supplying ``prose`` should:
- Append a ``user_turn_edit`` event whose payload references the
original user_turn id and carries the edited prose.
- Mark the original ``user_turn`` as ``superseded_by`` the edit.
"""
ut_id, at_id = _seed_with_one_turn(tmp_path / "test.db")
narrative_canned = "Reply to edited."
state_canned = json.dumps(
{"affinity_delta": 0, "trust_delta": 0, "knowledge_facts": []}
)
canned = [narrative_canned, state_canned, state_canned]
from chat.web.kickoff import get_llm_client
app.dependency_overrides[get_llm_client] = lambda: MockLLMClient(
canned=list(canned)
)
try:
response = client.post(
f"/chats/chat_bot_a/turns/{at_id}/regenerate",
data={"prose": "edited prose"},
)
assert response.status_code == 204
finally:
app.dependency_overrides.clear()
with open_db(tmp_path / "test.db") as conn:
# A user_turn_edit event was appended with the edited prose and
# a back-pointer to the original user_turn.
cur = conn.execute(
"SELECT payload_json FROM event_log WHERE kind = 'user_turn_edit'"
).fetchall()
assert len(cur) == 1
edit_payload = json.loads(cur[0][0])
assert edit_payload["prose"] == "edited prose"
assert edit_payload["supersedes_user_turn_id"] == ut_id
assert edit_payload["chat_id"] == "chat_bot_a"
# Original user_turn is now superseded.
ut_row = conn.execute(
"SELECT superseded_by FROM event_log WHERE id = ?", (ut_id,)
).fetchone()
assert ut_row[0] is not None
# Original assistant_turn is also superseded by the new one.
at_row = conn.execute(
"SELECT superseded_by FROM event_log WHERE id = ?", (at_id,)
).fetchone()
assert at_row[0] is not None
def test_regenerate_404_when_assistant_turn_missing(client, tmp_path):
"""An unknown ``event_id`` returns 404."""
_seed_with_one_turn(tmp_path / "test.db")
from chat.web.kickoff import get_llm_client
app.dependency_overrides[get_llm_client] = lambda: MockLLMClient(
canned=["x", "y", "z"]
)
try:
response = client.post(
"/chats/chat_bot_a/turns/99999/regenerate", data={}
)
assert response.status_code == 404
finally:
app.dependency_overrides.clear()
+185
View File
@@ -0,0 +1,185 @@
from __future__ import annotations
from pathlib import Path
import pytest
from fastapi.testclient import TestClient
from chat.app import app
from chat.db.connection import open_db
from chat.eventlog.log import append_event
from chat.eventlog.projector import project
@pytest.fixture
def client(tmp_path, monkeypatch):
cfg = tmp_path / "config.toml"
cfg.write_text('featherless_api_key = "test"\n')
monkeypatch.setenv("CHAT_CONFIG_PATH", str(cfg))
db = tmp_path / "test.db"
monkeypatch.setenv("CHAT_DB_PATH", str(db))
with TestClient(app) as c:
if hasattr(app.state, "background_worker"):
app.state.background_worker.enabled = False
yield c
def _seed_bot_with_state(db: Path) -> None:
"""Seed a bot plus a chat, container, scene, edge, memory, and activity row."""
with open_db(db) as conn:
append_event(
conn,
kind="bot_authored",
payload={
"id": "bot_a",
"name": "BotA",
"persona": "thoughtful, observant",
"voice_samples": [],
"traits": ["shy"],
"backstory": "",
"initial_relationship_to_you": "coworker",
"kickoff_prose": "you stay late",
},
)
append_event(
conn,
kind="chat_created",
payload={
"id": "chat_bot_a",
"host_bot_id": "bot_a",
"initial_time": "2026-04-26T20:00:00+00:00",
"narrative_anchor": "Day 1",
"weather": "",
},
)
append_event(
conn,
kind="container_created",
payload={
"chat_id": "chat_bot_a",
"name": "office",
"type": "workplace",
"properties": {},
},
)
append_event(
conn,
kind="scene_opened",
payload={
"chat_id": "chat_bot_a",
"container_id": 1,
"started_at": "2026-04-26T20:00:00+00:00",
"participants": ["you", "bot_a"],
},
)
append_event(
conn,
kind="edge_update",
payload={
"source_id": "bot_a",
"target_id": "you",
"chat_id": "chat_bot_a",
"affinity_delta": 5,
"trust_delta": 2,
},
)
append_event(
conn,
kind="memory_written",
payload={
"owner_id": "bot_a",
"chat_id": "chat_bot_a",
"pov_summary": "Talked about her sister",
"witness_you": 1,
"witness_host": 1,
"witness_guest": 0,
"significance": 2,
},
)
append_event(
conn,
kind="activity_change",
payload={
"entity_id": "bot_a",
"posture": "sitting",
"action": {"verb": "writing"},
},
)
project(conn)
def test_reset_purges_state_but_preserves_identity(client, tmp_path):
_seed_bot_with_state(tmp_path / "test.db")
response = client.post(
"/bots/bot_a/reset",
data={"confirm_name": "BotA"},
follow_redirects=False,
)
assert response.status_code == 303
assert response.headers["location"] == "/bots"
with open_db(tmp_path / "test.db") as conn:
# Identity preserved.
bot = conn.execute(
"SELECT id, name, kickoff_prose, initial_relationship_to_you "
"FROM bots WHERE id = 'bot_a'"
).fetchone()
assert bot is not None
assert bot[1] == "BotA"
assert bot[2] == "you stay late"
assert bot[3] == "coworker"
# State purged.
assert conn.execute(
"SELECT COUNT(*) FROM chats WHERE host_bot_id = 'bot_a'"
).fetchone()[0] == 0
assert conn.execute(
"SELECT COUNT(*) FROM scenes WHERE chat_id = 'chat_bot_a'"
).fetchone()[0] == 0
assert conn.execute(
"SELECT COUNT(*) FROM containers WHERE chat_id = 'chat_bot_a'"
).fetchone()[0] == 0
assert conn.execute(
"SELECT COUNT(*) FROM chat_state WHERE chat_id = 'chat_bot_a'"
).fetchone()[0] == 0
assert conn.execute(
"SELECT COUNT(*) FROM memories WHERE owner_id = 'bot_a'"
).fetchone()[0] == 0
assert conn.execute(
"SELECT COUNT(*) FROM edges WHERE source_id = 'bot_a' OR target_id = 'bot_a'"
).fetchone()[0] == 0
assert conn.execute(
"SELECT COUNT(*) FROM activity WHERE entity_id = 'bot_a'"
).fetchone()[0] == 0
# Event log records the bot_reset event.
assert conn.execute(
"SELECT COUNT(*) FROM event_log WHERE kind = 'bot_reset'"
).fetchone()[0] == 1
def test_reset_400_when_confirm_name_mismatch(client, tmp_path):
_seed_bot_with_state(tmp_path / "test.db")
response = client.post(
"/bots/bot_a/reset",
data={"confirm_name": "WrongName"},
follow_redirects=False,
)
assert response.status_code == 400
def test_reset_404_when_bot_missing(client):
response = client.post(
"/bots/no_such/reset",
data={"confirm_name": "Anything"},
follow_redirects=False,
)
assert response.status_code == 404
def test_bot_list_renders_reset_form(client, tmp_path):
_seed_bot_with_state(tmp_path / "test.db")
response = client.get("/bots")
assert response.status_code == 200
assert "Reset" in response.text
assert "confirm_name" in response.text
+164
View File
@@ -0,0 +1,164 @@
"""Tests for Task 28 — rewind with snapshot, impact preview, and re-projection.
Per Requirements §10.1, rewind must:
* take a pre-rewind snapshot of all projected tables (so the user can recover),
* truncate the event log past a chosen event id,
* clear projected tables and re-project from the truncated log so live state
matches "what the world looked like at turn N" (no stale rows from rewound
events).
These tests cover the functional core. The HTTP route surface is left to the
plan's polish pass — tests exercise via direct service calls.
"""
from __future__ import annotations
import json
from chat.db.connection import open_db
from chat.db.migrate import apply_migrations
from chat.eventlog.log import append_event
from chat.eventlog.projector import project
from chat.services.rewind import compute_rewind_preview, execute_rewind
from chat.services.snapshot import take_snapshot
# Importing the state modules registers their projector handlers as a
# side effect — the test would otherwise see an unprojected db after
# re-projection because the registry would be empty.
import chat.state.entities # noqa: F401
import chat.state.edges # noqa: F401
import chat.state.memory # noqa: F401
import chat.state.world # noqa: F401
import chat.state.manual_edit # noqa: F401
def _seed_5_turns(db):
"""Seed: bot + chat + 5 mock user/assistant turn pairs.
user_turn / assistant_turn have no projector handlers — they live in
the event_log purely for transcript rendering — so the only
projection-bearing events are bot_authored and chat_created. That
makes the post-rewind invariants easy to assert without needing the
real classifier pass.
"""
apply_migrations(db)
with open_db(db) as conn:
append_event(
conn,
kind="bot_authored",
payload={
"id": "bot_a",
"name": "BotA",
"persona": "...",
"voice_samples": [],
"traits": [],
"backstory": "",
"initial_relationship_to_you": "",
"kickoff_prose": "",
},
)
append_event(
conn,
kind="chat_created",
payload={
"id": "chat_bot_a",
"host_bot_id": "bot_a",
"initial_time": "2026-04-26T20:00:00+00:00",
"narrative_anchor": "Day 1",
"weather": "",
},
)
for i in range(5):
append_event(
conn,
kind="user_turn",
payload={
"chat_id": "chat_bot_a",
"prose": f"turn {i}",
"segments": [],
},
)
append_event(
conn,
kind="assistant_turn",
payload={
"chat_id": "chat_bot_a",
"speaker_id": "bot_a",
"text": f"reply {i}",
"truncated": False,
"user_turn_id": i,
},
)
project(conn)
def test_take_snapshot_writes_file_to_disk(tmp_path):
db = tmp_path / "t.db"
_seed_5_turns(db)
with open_db(db) as conn:
snapshot_path = take_snapshot(
conn, data_dir=tmp_path / "data", kind="rewind"
)
assert snapshot_path.exists()
assert snapshot_path.parent == tmp_path / "data" / "snapshots" / "rewind"
data = json.loads(snapshot_path.read_text())
assert "event_log" in data
assert "bots" in data
assert "chats" in data
# Bot is in the dump
assert any(b["id"] == "bot_a" for b in data["bots"])
def test_compute_rewind_preview_counts_kinds(tmp_path):
db = tmp_path / "t.db"
_seed_5_turns(db)
with open_db(db) as conn:
# The 1st assistant_turn is the 4th event:
# 1 bot_authored, 2 chat_created, 3 user_turn, 4 assistant_turn.
# Everything past it should be in the preview (4 user + 4 assistant = 8).
first_assistant = conn.execute(
"SELECT id FROM event_log WHERE kind='assistant_turn' "
"ORDER BY id LIMIT 1"
).fetchone()[0]
preview = compute_rewind_preview(conn, after_event_id=first_assistant)
assert preview["after_event_id"] == first_assistant
assert preview["total_events"] > 0
# by_kind sums to total_events.
assert sum(preview["by_kind"].values()) == preview["total_events"]
# We rewound past assistant_turn #1, so 4 user + 4 assistant remain.
assert preview["by_kind"].get("user_turn") == 4
assert preview["by_kind"].get("assistant_turn") == 4
def test_execute_rewind_truncates_and_reprojects(tmp_path):
db = tmp_path / "t.db"
data_dir = tmp_path / "data"
_seed_5_turns(db)
with open_db(db) as conn:
first_assistant = conn.execute(
"SELECT id FROM event_log WHERE kind='assistant_turn' "
"ORDER BY id LIMIT 1"
).fetchone()[0]
snapshot_path = execute_rewind(
db_path=db, data_dir=data_dir, after_event_id=first_assistant
)
# Snapshot is written under data/snapshots/rewind/.
assert snapshot_path.exists()
assert snapshot_path.parent == data_dir / "snapshots" / "rewind"
# Verify event_log truncated and projected state matches state-at-turn.
with open_db(db) as conn:
max_id = conn.execute("SELECT MAX(id) FROM event_log").fetchone()[0]
assert max_id == first_assistant
# Bot still exists (re-projected from preserved bot_authored event).
bot = conn.execute(
"SELECT id FROM bots WHERE id = 'bot_a'"
).fetchone()
assert bot is not None
# Chat still exists (re-projected from preserved chat_created event).
chat = conn.execute(
"SELECT id FROM chats WHERE id = 'chat_bot_a'"
).fetchone()
assert chat is not None