dohertj2/chat
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

merge: T98 drawer Phase 4 bundle (branching + sig review + hide + delete + remaining edits)

Browse Source
This commit is contained in:
Joseph Doherty
2026-04-27 03:38:15 -04:00
parent 7899c50b6c 4546bc0d9c
commit 3dbe1a01ff
4 changed files with 1092 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files
chat/state/manual_edit.py
+38
View File
@@ -30,6 +30,20 @@ T72.3 adds a per-flag witness toggle:
``{"flag": "you"|"host"|"guest", "value": 0|1}`` and ``prior_value``
mirrors the same shape so an inverse edit can restore the flag.
T98.3 adds a hide-from-view toggle:
- ``turn_hidden`` — flip ``event_log.hidden`` on a single turn row.
Hidden turns are filtered by ``read_recent_dialogue`` (see
:mod:`chat.services.turn_common`) so they vanish from the prompt
without being deleted from the log. ``target_id`` is the integer
``event_log.id`` of the turn; ``new_value`` is ``{"hidden": 0|1}``
and ``prior_value`` mirrors the shape so an inverse edit restores it.
T98.5 finishes the v1 drawer surface with two chat-scope text edits:
- ``chat_narrative_anchor`` and ``chat_weather`` — string overwrites of
the matching ``chat_state`` columns. ``target_id`` is the chat id
(``chats.id``); ``new_value`` is the new string and ``prior_value``
carries the previous content for §6.4 reversibility.
Pin toggles intentionally use the existing ``memory_pin_changed`` event
(registered in :mod:`chat.state.memory`) rather than ``manual_edit`` so
the projection writes both ``pinned`` and ``auto_pinned`` atomically.
@@ -138,5 +152,29 @@ def _apply_manual_edit(conn: Connection, e: Event) -> None:
f"UPDATE memories SET witness_{flag} = ? WHERE id = ?",
(1 if int(new_value["value"]) else 0, int(target_id)),
)
elif kind == "turn_hidden":
# T98.3: hide-from-view toggle on a turn (event_log row). Sets
# ``event_log.hidden`` so :func:`read_recent_dialogue` (which
# filters ``hidden = 0``) drops the row from the prompt window
# without deleting it from the log. ``new_value`` is
# ``{"hidden": 0|1}``.
hidden_int = 1 if int(new_value.get("hidden", 0)) else 0
conn.execute(
"UPDATE event_log SET hidden = ? WHERE id = ?",
(hidden_int, int(target_id)),
)
elif kind == "chat_narrative_anchor":
# T98.5: string overwrite of ``chat_state.narrative_anchor`` for
# the chat keyed by ``target_id``.
conn.execute(
"UPDATE chat_state SET narrative_anchor = ? WHERE chat_id = ?",
(str(new_value), str(target_id)),
)
elif kind == "chat_weather":
# T98.5: string overwrite of ``chat_state.weather``.
conn.execute(
"UPDATE chat_state SET weather = ? WHERE chat_id = ?",
(str(new_value), str(target_id)),
)
# Unknown target_kind: silently no-op for v1. Future kinds (activity
# fields, etc.) extend the dispatch above.
chat/templates/_drawer.html
+135
View File
@@ -16,6 +16,26 @@
<p class="muted">No active container.</p>
{% endif %}
<p>Time: {{ chat.time }}</p>
<form class="inline-edit"
hx-post="/chats/{{ chat.id }}/drawer/chat/narrative-anchor"
hx-target="#drawer" hx-swap="innerHTML">
<label>
Narrative anchor:
<input type="text" name="new_value" maxlength="500"
value="{{ chat.narrative_anchor or '' }}">
</label>
<button type="submit">Save</button>
</form>
<form class="inline-edit"
hx-post="/chats/{{ chat.id }}/drawer/chat/weather"
hx-target="#drawer" hx-swap="innerHTML">
<label>
Weather:
<input type="text" name="new_value" maxlength="500"
value="{{ chat.weather or '' }}">
</label>
<button type="submit">Save</button>
</form>
{% if scene %}
<form class="inline-edit"
hx-post="/chats/{{ chat.id }}/drawer/scene/close"
@@ -414,6 +434,121 @@
{% endif %}
</section>
<section class="drawer-section">
<h3>Branches</h3>
{% if branches %}
<ul class="branch-list">
{% for b in branches %}
<li class="branch-row{% if b.is_active %} branch-active{% endif %}">
<strong>{{ b.name }}</strong>
{% if b.is_active %}<span class="muted"> (active)</span>{% endif %}
<span class="muted"> &middot; {{ b.event_count }} events</span>
{% if not b.is_active %}
<form class="inline-edit"
hx-post="/chats/{{ chat.id }}/drawer/branch/switch"
hx-target="#drawer" hx-swap="innerHTML">
<input type="hidden" name="name" value="{{ b.name }}">
<button type="submit">Switch</button>
</form>
{% endif %}
</li>
{% endfor %}
</ul>
{% else %}
<p class="muted">No branches yet.</p>
{% endif %}
<details>
<summary>Create branch</summary>
<form class="inline-edit"
hx-post="/chats/{{ chat.id }}/drawer/branch/create"
hx-target="#drawer" hx-swap="innerHTML">
<label>
Name:
<input type="text" name="name" required
placeholder="e.g. experiment_a">
</label>
<label>
Origin event id:
<input type="number" name="origin_event_id" required min="0">
</label>
<button type="submit">Create</button>
</form>
</details>
</section>
<section class="drawer-section">
<h3>Recent turns</h3>
{% if recent_turns %}
<ul class="recent-turns-list">
{% for t in recent_turns %}
<li class="turn-row{% if t.hidden %} turn-hidden{% endif %}">
<span class="muted">#{{ t.event_id }} {{ t.kind }}</span>
<strong>{{ t.speaker }}:</strong>
{{ t.excerpt }}{% if t.excerpt|length >= 120 %}…{% endif %}
<form class="inline-edit"
hx-post="/chats/{{ chat.id }}/drawer/turn/hide/{{ t.event_id }}"
hx-target="#drawer" hx-swap="innerHTML">
<input type="hidden" name="hidden" value="{{ 0 if t.hidden else 1 }}">
<label>
<input type="checkbox" {% if t.hidden %}checked{% endif %}
onchange="this.form.requestSubmit()">
hide from view
</label>
</form>
</li>
{% endfor %}
</ul>
{% else %}
<p class="muted">No turns yet.</p>
{% endif %}
</section>
<section class="drawer-section">
<h3>Significance review</h3>
{% set total_mem = significance_distribution.values()|sum %}
{% if total_mem %}
<ul class="significance-distribution">
{% for level in [0, 1, 2, 3] %}
{% set count = significance_distribution[level] %}
{% set marker = ['·','•','★','★★'][level] %}
{% set pct = (100 * count / total_mem)|round(0, 'floor')|int if total_mem else 0 %}
<li class="sig-bar sig-{{ level }}">
<span class="sig-label">{{ marker }} ({{ level }})</span>
<span class="sig-bar-fill" style="width: {{ pct }}%"></span>
<span class="sig-count">{{ count }}</span>
</li>
{% endfor %}
</ul>
{% else %}
<p class="muted">No memories yet.</p>
{% endif %}
{% if recent_memories %}
<details>
<summary>Edit significance (recent memories)</summary>
<ul class="significance-edit-list">
{% for m in recent_memories %}
<li>
<span class="sig sig-{{ m.significance }}">{{ ['·','•','★','★★'][m.significance|default(0)] }}</span>
{{ m.pov_summary[:80] }}{% if m.pov_summary|length > 80 %}…{% endif %}
<form class="inline-edit"
hx-post="/chats/{{ chat.id }}/drawer/memory/{{ m.id }}/significance"
hx-target="#drawer" hx-swap="innerHTML">
<label>
Significance:
<input type="range" name="significance" min="0" max="3"
value="{{ m.significance|default(0) }}"
oninput="this.nextElementSibling.value = this.value">
<output>{{ m.significance|default(0) }}</output>
</label>
<button type="submit">Save</button>
</form>
</li>
{% endfor %}
</ul>
</details>
{% endif %}
</section>
<section class="drawer-section">
<h3>Pinned memories ({{ pinned|length }} / {{ pin_cap }})</h3>
{% if pinned %}
chat/web/drawer.py
+396
View File
@@ -36,7 +36,14 @@ from fastapi.responses import HTMLResponse
from fastapi.templating import Jinja2Templates
from chat.eventlog.log import append_and_apply
from chat.services.branching import (
branch_from_event,
list_branches_with_metadata,
switch_active_branch,
)
from chat.services.delete_impact import compute_delete_impact
from chat.services.relationship_seed import seed_inter_bot_edges
from chat.services.rewind import execute_rewind
from chat.services.scene_summarize import apply_scene_close_summary
from chat.state.edges import get_edge
from chat.state.entities import get_bot, get_you, list_bots
@@ -169,6 +176,63 @@ async def drawer(chat_id: str, request: Request, conn=Depends(get_conn)):
active_events = list_active_events(conn, chat_id)
open_threads = list_open_threads(conn, chat_id)
# T98.3: recent turns (user_turn / assistant_turn) for the hide-from-view
# panel. Includes ``hidden`` rows so the user can un-hide them — the
# filter on the read side (read_recent_dialogue) is what drops hidden
# rows from the prompt; the drawer panel always shows everything.
turn_rows = conn.execute(
"""
SELECT id, kind, payload_json, hidden
FROM event_log
WHERE kind IN ('user_turn', 'assistant_turn', 'user_turn_edit')
AND superseded_by IS NULL
ORDER BY id DESC
LIMIT ?
""",
(RECENT_LIMIT,),
).fetchall()
recent_turns: list[dict] = []
for row in turn_rows:
try:
payload = json.loads(row[2]) if row[2] else {}
except (json.JSONDecodeError, TypeError):
payload = {}
if payload.get("chat_id") != chat_id:
continue
text = payload.get("prose") or payload.get("text") or ""
speaker = payload.get("speaker_id") or (
"you" if row[1].startswith("user") else "?"
)
recent_turns.append(
{
"event_id": int(row[0]),
"kind": row[1],
"speaker": speaker,
"excerpt": (text or "").replace("\n", " ")[:120],
"hidden": bool(row[3]),
}
)
# T98.1: branch metadata (every chat sees the global branch list — branches
# may be chat-scoped or global, so :func:`list_branches_with_metadata`
# returns both flavours and the template highlights the active one).
branches = list_branches_with_metadata(conn, chat_id)
# T98.2: significance distribution across this chat's memories. Powers
# the "Significance review" panel — a small histogram letting authors
# spot lopsided buckets (e.g. nothing significant=3 yet) and triage by
# editing individual memory significance values.
sig_rows = conn.execute(
"SELECT significance, COUNT(*) FROM memories "
"WHERE chat_id = ? GROUP BY significance ORDER BY significance",
(chat_id,),
).fetchall()
significance_distribution = {int(r[0]): int(r[1]) for r in sig_rows}
# Ensure every bucket 0..3 is present so the bar-chart template can
# render a stable axis even when a level has zero rows.
for level in (0, 1, 2, 3):
significance_distribution.setdefault(level, 0)
return TEMPLATES.TemplateResponse(
request,
"_drawer.html",
@@ -196,6 +260,9 @@ async def drawer(chat_id: str, request: Request, conn=Depends(get_conn)):
"pin_cap": PIN_CAP,
"active_events": active_events,
"open_threads": open_threads,
"branches": branches,
"significance_distribution": significance_distribution,
"recent_turns": recent_turns,
},
)
@@ -1080,3 +1147,332 @@ async def close_thread(
},
)
return await drawer(chat_id, request, conn)
# --- T98.1 branching UI --------------------------------------------------
#
# Three POST endpoints wired to the Phase 4 :mod:`chat.services.branching`
# helpers. The drawer's "Branches" panel exposes:
#
# * Create from a free-form ``origin_event_id``.
# * Switch the active branch by name.
# * Convenience "branch from this turn" against a per-turn event_id (the
# chat surface stamps ``id="turn-<event_id>"`` on every turn so users can
# pick the right one without copying ids by hand).
#
# All three return the refreshed drawer partial; failures from the service
# layer (duplicate name, unknown branch, invalid origin) surface as 400 so
# HTMX displays the inline error.
@router.post(
"/chats/{chat_id}/drawer/branch/create",
response_class=HTMLResponse,
)
async def create_branch(
chat_id: str,
request: Request,
name: str = Form(...),
origin_event_id: int = Form(...),
conn=Depends(get_conn),
):
chat = get_chat(conn, chat_id)
if chat is None:
raise HTTPException(status_code=404, detail=f"chat not found: {chat_id}")
try:
branch_from_event(
conn,
name=name,
origin_event_id=int(origin_event_id),
chat_id=chat_id,
)
except ValueError as exc:
raise HTTPException(status_code=400, detail=str(exc))
return await drawer(chat_id, request, conn)
@router.post(
"/chats/{chat_id}/drawer/branch/switch",
response_class=HTMLResponse,
)
async def switch_branch(
chat_id: str,
request: Request,
name: str = Form(...),
conn=Depends(get_conn),
):
chat = get_chat(conn, chat_id)
if chat is None:
raise HTTPException(status_code=404, detail=f"chat not found: {chat_id}")
try:
switch_active_branch(conn, name=name)
except ValueError as exc:
raise HTTPException(status_code=400, detail=str(exc))
return await drawer(chat_id, request, conn)
@router.get(
"/chats/{chat_id}/drawer/turn/delete-preview/{event_id}",
response_class=HTMLResponse,
)
async def delete_preview(
chat_id: str,
event_id: int,
request: Request,
conn=Depends(get_conn),
):
"""Render an :class:`ImpactReport` for ``event_id`` as a small modal.
Read-only — :func:`compute_delete_impact` does not mutate the
database. The modal contains a confirmation form posting to
:func:`delete_turn` below; HTMX swaps the fragment into a modal
target on the chat page.
"""
chat = get_chat(conn, chat_id)
if chat is None:
raise HTTPException(status_code=404, detail=f"chat not found: {chat_id}")
report = compute_delete_impact(conn, target_event_id=int(event_id))
# Build the modal HTML directly — the impact report is small and
# reusing the drawer template would require a fragment include just
# for this surface. Mirrors the rewind-preview style in
# :func:`chat.web.turns.rewind_preview`.
items_html = "".join(
f"<li><strong>{item.kind}</strong>: {item.description}</li>"
for item in report.cascading
)
notes_html = "".join(f"<li>{note}</li>" for note in report.notes)
body = (
"<div class='delete-impact-modal'>"
f"<h3>Delete event {report.target_event_id}?</h3>"
f"<p>This will discard {len(report.cascading)} events. Cascade:</p>"
f"<ul class='delete-impact-cascade'>{items_html or '<li>none</li>'}</ul>"
f"<ul class='delete-impact-notes'>{notes_html}</ul>"
f"<form hx-post='/chats/{chat_id}/drawer/turn/delete/{report.target_event_id}' "
"hx-target='#drawer' hx-swap='innerHTML'>"
"<button type='submit'>Confirm delete</button>"
"</form>"
"</div>"
)
return HTMLResponse(body)
@router.post(
"/chats/{chat_id}/drawer/turn/delete/{event_id}",
response_class=HTMLResponse,
)
async def delete_turn(
chat_id: str,
event_id: int,
request: Request,
conn=Depends(get_conn),
):
"""Delete a turn (and everything after) by invoking the existing rewind path.
The :func:`chat.services.rewind.execute_rewind` API takes
``after_event_id``: it removes events with id strictly greater than
that argument. To make ``event_id`` itself disappear we pass
``after_event_id = event_id - 1`` — a thin adapter, not a
re-implementation of rewind.
A snapshot is taken before truncation (inside ``execute_rewind``)
so the user can recover via the snapshot index.
"""
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=int(event_id) - 1,
)
# ``conn`` is now stale (the rewind opened its own connection and
# truncated/reprojected). Re-render the drawer through a fresh open
# so the partial reflects the truncated state.
from chat.db.connection import open_db
with open_db(settings.db_path) as fresh:
return await drawer(chat_id, request, fresh)
@router.post(
"/chats/{chat_id}/drawer/turn/hide/{event_id}",
response_class=HTMLResponse,
)
async def hide_turn(
chat_id: str,
event_id: int,
request: Request,
hidden: int = Form(...),
conn=Depends(get_conn),
):
"""Toggle ``event_log.hidden`` on a turn via the ``turn_hidden``
``manual_edit`` projector branch.
The route validates the target is an actual turn-shaped row in this
chat (so a stray click on the chat panel can't hide a system event)
and snapshots the prior ``hidden`` value for §6.4 reversibility.
"""
chat = get_chat(conn, chat_id)
if chat is None:
raise HTTPException(status_code=404, detail=f"chat not found: {chat_id}")
row = conn.execute(
"SELECT kind, payload_json, hidden FROM event_log WHERE id = ?",
(int(event_id),),
).fetchone()
if row is None:
raise HTTPException(
status_code=404, detail=f"event not found: {event_id}"
)
if row[0] not in ("user_turn", "assistant_turn", "user_turn_edit"):
raise HTTPException(
status_code=400,
detail=f"event {event_id} is not a turn (kind={row[0]})",
)
try:
payload = json.loads(row[1]) if row[1] else {}
except (json.JSONDecodeError, TypeError):
payload = {}
if payload.get("chat_id") != chat_id:
raise HTTPException(
status_code=404,
detail=f"event {event_id} not in chat {chat_id}",
)
prior_hidden = 1 if int(row[2]) else 0
new_hidden = 1 if int(hidden) else 0
append_and_apply(
conn,
kind="manual_edit",
payload={
"target_kind": "turn_hidden",
"target_id": int(event_id),
"prior_value": {"hidden": prior_hidden},
"new_value": {"hidden": new_hidden},
},
)
return await drawer(chat_id, request, conn)
# --- T98.5 chat narrative anchor + weather ----------------------------
#
# Audit (T98.5) found two §6.4 fields without drawer affordances despite
# both being prose strings stored on ``chat_state``: ``narrative_anchor``
# (the "Day 1" / "morning of the gala" hint above the chat clock) and
# ``weather``. Both land via the existing ``manual_edit`` projector with
# new branches added in :mod:`chat.state.manual_edit`. The container
# ``properties_json`` blob is more invasive — bounded JSON edits aren't
# wired through manual_edit and the drawer never surfaces multiple
# containers at once, so it stays out of v1.
CHAT_NARRATIVE_ANCHOR_MAX = 500
CHAT_WEATHER_MAX = 500
@router.post(
"/chats/{chat_id}/drawer/chat/narrative-anchor",
response_class=HTMLResponse,
)
async def edit_chat_narrative_anchor(
chat_id: str,
request: Request,
new_value: str = Form(...),
conn=Depends(get_conn),
):
chat = get_chat(conn, chat_id)
if chat is None:
raise HTTPException(status_code=404, detail=f"chat not found: {chat_id}")
if len(new_value) > CHAT_NARRATIVE_ANCHOR_MAX:
raise HTTPException(
status_code=400,
detail=(
f"narrative_anchor exceeds {CHAT_NARRATIVE_ANCHOR_MAX} chars "
f"(got {len(new_value)})"
),
)
prior = chat.get("narrative_anchor") or ""
append_and_apply(
conn,
kind="manual_edit",
payload={
"target_kind": "chat_narrative_anchor",
"target_id": chat_id,
"prior_value": prior,
"new_value": new_value,
},
)
return await drawer(chat_id, request, conn)
@router.post(
"/chats/{chat_id}/drawer/chat/weather",
response_class=HTMLResponse,
)
async def edit_chat_weather(
chat_id: str,
request: Request,
new_value: str = Form(...),
conn=Depends(get_conn),
):
chat = get_chat(conn, chat_id)
if chat is None:
raise HTTPException(status_code=404, detail=f"chat not found: {chat_id}")
if len(new_value) > CHAT_WEATHER_MAX:
raise HTTPException(
status_code=400,
detail=(
f"weather exceeds {CHAT_WEATHER_MAX} chars "
f"(got {len(new_value)})"
),
)
prior = chat.get("weather") or ""
append_and_apply(
conn,
kind="manual_edit",
payload={
"target_kind": "chat_weather",
"target_id": chat_id,
"prior_value": prior,
"new_value": new_value,
},
)
return await drawer(chat_id, request, conn)
@router.post(
"/chats/{chat_id}/drawer/branch/from-turn/{event_id}",
response_class=HTMLResponse,
)
async def branch_from_turn(
chat_id: str,
event_id: int,
request: Request,
name: str = Form(...),
conn=Depends(get_conn),
):
"""Convenience: branch from a specific turn event.
Identical to :func:`create_branch` except ``origin_event_id`` is
encoded in the URL — the chat surface renders one such form per turn
so users can fork mid-conversation without authoring an event id by
hand.
"""
chat = get_chat(conn, chat_id)
if chat is None:
raise HTTPException(status_code=404, detail=f"chat not found: {chat_id}")
try:
branch_from_event(
conn,
name=name,
origin_event_id=int(event_id),
chat_id=chat_id,
)
except ValueError as exc:
raise HTTPException(status_code=400, detail=str(exc))
return await drawer(chat_id, request, conn)
tests/test_drawer_phase4.py
+523
View File
@@ -0,0 +1,523 @@
"""T98 (Phase 4): drawer phase-4 bundle.
Five sub-features extending the chat drawer:
* T98.1 — branching UI (create / switch / from-turn).
* T98.2 — significance-review panel (distribution + significance edits).
* T98.3 — hide-from-view toggle (per-turn, via ``manual_edit`` projector
branch ``turn_hidden``).
* T98.4 — surgical delete with cascade preview (preview modal +
rewind execution against a target turn).
* T98.5 — remaining v1 edits (chat narrative_anchor + weather).
Tests follow the T59 pattern in ``tests/test_drawer_events_threads_skip.py``
— a TestClient against the real FastAPI app with a per-test temp DB.
"""
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_and_apply, 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 _bot_payload(bot_id: str, name: str) -> dict:
return {
"id": bot_id,
"name": name,
"persona": "...",
"voice_samples": [],
"traits": [],
"backstory": "",
"initial_relationship_to_you": "",
"kickoff_prose": "",
}
def _seed_chat(db: Path, *, with_scene: bool = True) -> int:
"""Seed a chat hosted by ``bot_a``; return the latest event id (chat_created)."""
with open_db(db) as conn:
append_event(conn, kind="bot_authored", payload=_bot_payload("bot_a", "BotA"))
append_event(
conn,
kind="you_authored",
payload={"name": "Me", "pronouns": "they/them", "persona": ""},
)
chat_event_id = 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": "",
},
)
if with_scene:
append_event(
conn,
kind="scene_opened",
payload={
"chat_id": "chat_bot_a",
"container_id": None,
"started_at": "2026-04-26T20:00:00+00:00",
"participants": ["you", "bot_a"],
},
)
project(conn)
return chat_event_id
# ---------------------------------------------------------------------------
# T98.1 — branching UI.
# ---------------------------------------------------------------------------
def test_t98_1_create_branch_emits_branch_created_and_renders(client, tmp_path):
db = tmp_path / "test.db"
seed_id = _seed_chat(db)
response = client.post(
"/chats/chat_bot_a/drawer/branch/create",
data={"name": "experiment_a", "origin_event_id": str(seed_id)},
)
assert response.status_code == 200
with open_db(db) as conn:
rows = conn.execute(
"SELECT COUNT(*) FROM event_log WHERE kind = 'branch_created'"
).fetchone()
assert rows[0] == 1
from chat.state.branches import get_branch
b = get_branch(conn, "experiment_a")
assert b is not None
assert b["origin_event_id"] == seed_id
assert b["chat_id"] == "chat_bot_a"
# Drawer partial lists the new branch.
body = response.text
assert "<h3>Branches</h3>" in body
assert "experiment_a" in body
def test_t98_1_switch_branch_marks_active_and_unknown_400s(client, tmp_path):
db = tmp_path / "test.db"
seed_id = _seed_chat(db)
# Create branch directly via the service so this test focuses on switch.
with open_db(db) as conn:
from chat.services.branching import branch_from_event
branch_from_event(
conn, name="experiment_b", origin_event_id=seed_id, chat_id="chat_bot_a"
)
response = client.post(
"/chats/chat_bot_a/drawer/branch/switch",
data={"name": "experiment_b"},
)
assert response.status_code == 200
with open_db(db) as conn:
from chat.state.branches import active_branch
active = active_branch(conn)
assert active is not None
assert active["name"] == "experiment_b"
# Unknown branch -> 400.
bad = client.post(
"/chats/chat_bot_a/drawer/branch/switch",
data={"name": "ghost_branch"},
)
assert bad.status_code == 400
def test_t98_1_branch_from_turn_emits_branch_created(client, tmp_path):
db = tmp_path / "test.db"
seed_id = _seed_chat(db)
# Append an extra turn so we can branch from it specifically.
with open_db(db) as conn:
turn_id = append_event(
conn,
kind="user_turn",
payload={"chat_id": "chat_bot_a", "prose": "hi", "segments": []},
)
response = client.post(
f"/chats/chat_bot_a/drawer/branch/from-turn/{turn_id}",
data={"name": "fork_at_turn"},
)
assert response.status_code == 200
with open_db(db) as conn:
from chat.state.branches import get_branch
b = get_branch(conn, "fork_at_turn")
assert b is not None
assert b["origin_event_id"] == turn_id
assert b["chat_id"] == "chat_bot_a"
# Duplicate name -> 400 from service ValueError.
dup = client.post(
f"/chats/chat_bot_a/drawer/branch/from-turn/{turn_id}",
data={"name": "fork_at_turn"},
)
assert dup.status_code == 400
assert seed_id < turn_id # sanity: turn is after chat_created
# ---------------------------------------------------------------------------
# T98.2 — significance review panel.
# ---------------------------------------------------------------------------
def _seed_memories_for_significance(db: Path) -> list[int]:
"""Seed three memories with significance levels 0, 1, 2. Returns ids.
Uses ``append_and_apply`` (vs ``append_event`` + a final ``project``)
so each row is applied exactly once — the chat row was already
materialised by ``_seed_chat`` and a re-projection would conflict
on ``chats.id`` UNIQUE.
"""
ids: list[int] = []
with open_db(db) as conn:
for sig in (0, 1, 2):
append_and_apply(
conn,
kind="memory_written",
payload={
"owner_id": "bot_a",
"chat_id": "chat_bot_a",
"pov_summary": f"memory at significance {sig}",
"witness_you": 1,
"witness_host": 1,
"witness_guest": 0,
"significance": sig,
},
)
rows = conn.execute(
"SELECT id FROM memories WHERE chat_id = 'chat_bot_a' "
"ORDER BY id ASC"
).fetchall()
ids = [int(r[0]) for r in rows]
return ids
def test_t98_2_distribution_renders_per_significance_bucket(client, tmp_path):
db = tmp_path / "test.db"
_seed_chat(db)
_seed_memories_for_significance(db)
response = client.get("/chats/chat_bot_a/drawer")
assert response.status_code == 200
body = response.text
# Section heading + bar entries for each significance level.
assert "<h3>Significance review</h3>" in body
# All four buckets appear by their canonical label even when count=0.
assert ">★★ (3)<" in body or "(3)" in body
# The distribution markup names each level explicitly.
for level in (0, 1, 2, 3):
assert f"sig-bar sig-{level}" in body
# Three seeded memories (sigs 0, 1, 2) — each has a count = 1 bar.
# We don't pin exact text formatting, just verify the per-level bars
# are present.
def test_t98_2_edit_significance_via_existing_route_lands_manual_edit(
client, tmp_path
):
db = tmp_path / "test.db"
_seed_chat(db)
ids = _seed_memories_for_significance(db)
target_id = ids[0] # initially significance=0
response = client.post(
f"/chats/chat_bot_a/drawer/memory/{target_id}/significance",
data={"significance": "3"},
)
assert response.status_code == 200
with open_db(db) as conn:
# Significance updated in the projected table.
row = conn.execute(
"SELECT significance FROM memories WHERE id = ?", (target_id,)
).fetchone()
assert int(row[0]) == 3
# manual_edit landed in the event log with the prior snapshot.
import json as _json
log_rows = conn.execute(
"SELECT payload_json FROM event_log "
"WHERE kind = 'manual_edit' ORDER BY id DESC LIMIT 1"
).fetchone()
payload = _json.loads(log_rows[0])
assert payload["target_kind"] == "memory_significance"
assert int(payload["target_id"]) == target_id
assert payload["prior_value"] == 0
assert payload["new_value"] == 3
# ---------------------------------------------------------------------------
# T98.3 — hide-from-view toggle.
# ---------------------------------------------------------------------------
def _seed_turns(db: Path) -> tuple[int, int]:
"""Append one user_turn + one assistant_turn; return their event ids."""
with open_db(db) as conn:
user_id = append_and_apply(
conn,
kind="user_turn",
payload={
"chat_id": "chat_bot_a",
"prose": "How are you doing today?",
"segments": [],
},
)
bot_id = append_and_apply(
conn,
kind="assistant_turn",
payload={
"chat_id": "chat_bot_a",
"speaker_id": "bot_a",
"text": "Quite well, thanks for asking!",
"truncated": False,
"user_turn_id": user_id,
},
)
return user_id, bot_id
def test_t98_3_hide_turn_flips_event_log_hidden_via_manual_edit(
client, tmp_path
):
db = tmp_path / "test.db"
_seed_chat(db)
user_id, bot_id = _seed_turns(db)
response = client.post(
f"/chats/chat_bot_a/drawer/turn/hide/{user_id}",
data={"hidden": "1"},
)
assert response.status_code == 200
with open_db(db) as conn:
# event_log.hidden flipped to 1.
row = conn.execute(
"SELECT hidden FROM event_log WHERE id = ?", (user_id,)
).fetchone()
assert int(row[0]) == 1
# manual_edit landed with the prior snapshot.
import json as _json
log = conn.execute(
"SELECT payload_json FROM event_log "
"WHERE kind = 'manual_edit' ORDER BY id DESC LIMIT 1"
).fetchone()
payload = _json.loads(log[0])
assert payload["target_kind"] == "turn_hidden"
assert int(payload["target_id"]) == user_id
assert payload["prior_value"] == {"hidden": 0}
assert payload["new_value"] == {"hidden": 1}
def test_t98_3_hidden_turn_disappears_from_read_recent_dialogue(
client, tmp_path
):
"""Hiding a turn must drop it from the prompt-window read.
``read_recent_dialogue`` (chat.services.turn_common) filters
``hidden = 0`` server-side, so flipping the flag via the drawer
route must surface immediately.
"""
db = tmp_path / "test.db"
_seed_chat(db)
user_id, bot_id = _seed_turns(db)
# Sanity baseline — both turns visible before the hide.
with open_db(db) as conn:
from chat.services.turn_common import read_recent_dialogue
before = read_recent_dialogue(conn, "chat_bot_a", limit=10)
before_ids = [t["event_id"] for t in before]
assert user_id in before_ids
assert bot_id in before_ids
# Hide the user turn via the drawer route.
response = client.post(
f"/chats/chat_bot_a/drawer/turn/hide/{user_id}",
data={"hidden": "1"},
)
assert response.status_code == 200
with open_db(db) as conn:
from chat.services.turn_common import read_recent_dialogue
after = read_recent_dialogue(conn, "chat_bot_a", limit=10)
after_ids = [t["event_id"] for t in after]
assert user_id not in after_ids
assert bot_id in after_ids # the unhidden bot turn still surfaces
# ---------------------------------------------------------------------------
# T98.4 — surgical delete with cascade preview.
# ---------------------------------------------------------------------------
def test_t98_4_delete_preview_returns_impact_report_html(client, tmp_path):
db = tmp_path / "test.db"
_seed_chat(db)
user_id, bot_id = _seed_turns(db)
response = client.get(
f"/chats/chat_bot_a/drawer/turn/delete-preview/{user_id}"
)
assert response.status_code == 200
body = response.text
# Modal markup with the event id and the cascade list.
assert "delete-impact-modal" in body
assert f"Delete event {user_id}?" in body
assert "delete-impact-cascade" in body
# Both turns ride along in the cascade — user_turn at user_id, then
# the assistant_turn at bot_id (>= user_id).
assert "user_turn" in body
assert "assistant_turn" in body
# Confirm-form posts to the delete route.
assert f"/drawer/turn/delete/{user_id}" in body
def test_t98_4_delete_invokes_rewind_and_drops_cascade(client, tmp_path):
db = tmp_path / "test.db"
_seed_chat(db)
user_id, bot_id = _seed_turns(db)
# Append a third turn after the assistant_turn so we can verify the
# cascade catches everything from user_id forward.
with open_db(db) as conn:
extra_id = append_and_apply(
conn,
kind="user_turn",
payload={
"chat_id": "chat_bot_a",
"prose": "follow-up",
"segments": [],
},
)
# Sanity: all three turn rows exist.
with open_db(db) as conn:
turn_count = conn.execute(
"SELECT COUNT(*) FROM event_log "
"WHERE kind IN ('user_turn', 'assistant_turn')"
).fetchone()[0]
assert turn_count == 3
# Delete from user_id forward.
response = client.post(f"/chats/chat_bot_a/drawer/turn/delete/{user_id}")
assert response.status_code == 200
# All three turns are gone — the rewind truncated the log past
# user_id - 1, removing user_id, bot_id, and extra_id.
with open_db(db) as conn:
turn_count = conn.execute(
"SELECT COUNT(*) FROM event_log "
"WHERE kind IN ('user_turn', 'assistant_turn')"
).fetchone()[0]
assert turn_count == 0
for ev_id in (user_id, bot_id, extra_id):
row = conn.execute(
"SELECT 1 FROM event_log WHERE id = ?", (ev_id,)
).fetchone()
assert row is None, f"event {ev_id} should have been deleted"
# ---------------------------------------------------------------------------
# T98.5 — remaining v1 edits (chat narrative anchor + weather).
# ---------------------------------------------------------------------------
def test_t98_5_edit_chat_narrative_anchor_emits_manual_edit(client, tmp_path):
db = tmp_path / "test.db"
_seed_chat(db)
response = client.post(
"/chats/chat_bot_a/drawer/chat/narrative-anchor",
data={"new_value": "Late evening, after dinner"},
)
assert response.status_code == 200
with open_db(db) as conn:
row = conn.execute(
"SELECT narrative_anchor FROM chat_state WHERE chat_id = ?",
("chat_bot_a",),
).fetchone()
assert row[0] == "Late evening, after dinner"
import json as _json
log = conn.execute(
"SELECT payload_json FROM event_log "
"WHERE kind = 'manual_edit' ORDER BY id DESC LIMIT 1"
).fetchone()
payload = _json.loads(log[0])
assert payload["target_kind"] == "chat_narrative_anchor"
assert payload["target_id"] == "chat_bot_a"
assert payload["prior_value"] == "Day 1"
assert payload["new_value"] == "Late evening, after dinner"
def test_t98_5_edit_chat_weather_emits_manual_edit(client, tmp_path):
db = tmp_path / "test.db"
_seed_chat(db)
response = client.post(
"/chats/chat_bot_a/drawer/chat/weather",
data={"new_value": "thunderstorm rolling in"},
)
assert response.status_code == 200
with open_db(db) as conn:
row = conn.execute(
"SELECT weather FROM chat_state WHERE chat_id = ?",
("chat_bot_a",),
).fetchone()
assert row[0] == "thunderstorm rolling in"
import json as _json
log = conn.execute(
"SELECT payload_json FROM event_log "
"WHERE kind = 'manual_edit' ORDER BY id DESC LIMIT 1"
).fetchone()
payload = _json.loads(log[0])
assert payload["target_kind"] == "chat_weather"
assert payload["target_id"] == "chat_bot_a"
assert payload["prior_value"] == ""
assert payload["new_value"] == "thunderstorm rolling in"