"""Regenerate flow (T29). POST ``/chats//turns//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()