merge: T102 phase 4 documentation update

Wires T93's `search_all_memories` service into a small read-only HTML
surface so users can find a memory across every chat in the database.

* `chat/web/search.py` (new): GET `/search?q=...` runs the FTS service
  with k=50, hydrates each row with bot name + scene timestamp, and
  renders `search.html`. Empty `q` short-circuits to no results so the
  top-bar form can submit even with an empty input.
* `chat/templates/search.html` (new): empty-state placeholder, results
  list with chat-level "Open chat" links (`/chats/{chat_id}` — memories
  don't carry an event_id today, so no per-turn anchor).
* `chat/templates/layout.html`: append a small `<form>` to the rail
  nav, additive only.
* `chat/app.py`: register `search_router` (additive import + include).
* `tests/test_search_ux.py`: 3 tests — multi-chat results, empty-query
  placeholder, chat link.

CLAUDE.md

@@ -287,3 +287,88 @@ New follow-ups discovered during Phase 3.5 reviews and execution. None are block
 - **Scene-close-on-cancel UX revisit** (Phase 2.5 carry-over): T74.3 pinned the existing behavior; revisit if real play-testing surfaces a regression.
 - **Cross-feature canned-queue brittleness**: meanwhile-scene close test required a canned response for T65's digest call after T64+T65 merge. Future close-path additions will keep extending the queue. Consider a structured fixture builder rather than positional canned arrays. NOT addressed in Phase 3.5.
 - **Lifecycle-transition rollback in regenerate**: T83.4 added a warning log; actual rollback (with proper schema linkage from lifecycle event back to producing turn) is Phase 4 work.
+
+## Phase 4 status
+
+Phase 4 polish shipped end-to-end across 15 tasks (T88–T102). Vector retrieval is functional via pure-Python cosine over a JSON-blob embeddings table (sqlite-vec deferred — host Python lacks loadable extensions). Branching is data-model + drawer UI. Surgical delete with cascade preview, hide-from-view soft delete, significance review panel, snapshot UX, and cross-chat search all surface from the drawer or top-bar. Test count grew from 343 (Phase 3.5) to ~413 (+70 new tests).
+
+- **Wave 1 — schema + Phase 3.6 carry-overs (parallel)**:
+  - **T88** `embeddings` table + projector handlers (pure-Python cosine, JSON-blob storage; sqlite-vec deferred).
+  - **T89** `branches` table + handlers (main bootstrapped; `is_active` flag; partial unique index).
+  - **T90** Phase 3.6 carry-overs trio — `read_recent_dialogue` chat-id SQL pushdown, lifecycle warning wording tightening, legacy `record_turn_memory` removed.
+- **Wave 2 — services (parallel)**:
+  - **T91** embedding generation service (Phase 4 ships a deterministic SHA-256-derived pseudo-embedding; real model swap is Phase 4.5+).
+  - **T92** vector search service via pure-Python cosine.
+  - **T93** cross-chat search service (FTS5 across all owners, no witness filter — admin-style).
+- **Wave 3 — services (parallel)**:
+  - **T94** branching service (`branch_from_event`, `switch_active_branch`, `list_branches_with_metadata`).
+  - **T95** delete-impact computation service (cascade preview, no DB mutation).
+- **Wave 4 — combined retrieval (single)**:
+  - **T96** combined FTS + vector retrieval ranking via reciprocal-rank fusion (RRF, `RRF_CONST=60`); existing significance/recency boost applied as final pass.
+- **Wave 5 — memory write hook + backfill (single)**:
+  - **T97** `EmbeddingWorker` drains queue and emits `embedding_indexed` events; `memory_write` enqueues per `memory_written`; `backfill_embeddings` script for existing memories; ALL 4 production call sites wired (turns, regenerate, meanwhile, drawer).
+- **Wave 6 — drawer Phase 4 bundle (single, 5 sub-features)**:
+  - **T98.1** branching UI (Branches panel + 3 routes).
+  - **T98.2** significance review panel (distribution bar chart + per-memory edit).
+  - **T98.3** hide-from-view toggle + `turn_hidden` `manual_edit` branch.
+  - **T98.4** surgical delete with cascade preview (reuses existing rewind path; pre-rewind snapshot preserved).
+  - **T98.5** remaining v1 edits — `narrative_anchor` + weather drawer affordances + 2 new `manual_edit` branches.
+- **Wave 7 — UX surfaces (parallel)**:
+  - **T99** snapshot UX (manual trigger, list, restore with hard-confirm, preview).
+  - **T100** cross-chat search UX (top-bar form + results page).
+- **Wave 8 — polish (parallel)**:
+  - **T101** cross-feature integration tests (5 multi-feature scenarios).
+  - **T102** documentation (this section).
+
+### Phase 4.5 / 5 backlog
+
+New follow-ups discovered during Phase 4 reviews and execution. None are blocking; pick up at any time.
+
+#### From T88 review
+
+- **`embeddings` FK lacks `ON DELETE CASCADE`**: deindex events are the only deletion path; if memories ever get deleted directly (raw SQL), embedding rows orphan. Defensible since projector model uses explicit deindex events, but worth a comment or `ON DELETE CASCADE` addition.
+
+#### From T89 review
+
+- **`list_branches(chat_id=...)` filter leaks global branches** (`chat_id IS NULL`) into every chat scope. Intentional? Document.
+- **Branch-switch to nonexistent silently leaves zero active branches** — log a warning when this would happen.
+
+#### From T91 review
+
+- **Real embedding model swap**: Phase 4 ships pseudo-embedding (deterministic SHA-256 hash). Phase 4.5+ should swap to a real model (Featherless `bge-small-en-v1.5` if available; or local `sentence-transformers/all-MiniLM-L6-v2`). The 384-dim is hardcoded in `0012_embeddings.sql`; if dim changes, migrate first.
+- **`timeout_s` unused on pseudo path** — fine, but log when non-default model falls through to fallback so misconfigured callers don't silently degrade.
+
+#### From T96 review
+
+- **Duplicate `MAX(id)` lookup** between `_composite_rerank` and the fused-path tail — DRY follow-up.
+- **`fts_rank=None` for vector-only rows** — document downstream contract.
+
+#### From T98 review
+
+- **`event_id <= 0` guard in `delete_turn`** — currently silently rewinds everything if `event_id` is 0. Add `if event_id <= 0: 400`.
+- **`html.escape()` on `compute_delete_impact` output rendered into the modal** — defense in depth (currently model-controlled strings, but if event payload fields ever appear in descriptions, autoescape needed).
+- **Extract delete-impact modal HTML to a Jinja partial** — testability + autoescape inheritance.
+
+#### From T99 review
+
+- **Hoist `datetime`/`timezone` imports to module level** in `chat/web/snapshots.py`.
+- **`kind` defaulting in restore/preview** — reject missing `kind` rather than silent 404.
+- **`created_at` from file mtime** vs filename-encoded timestamp — small drift if files copied; document.
+
+#### From T100 review
+
+- **Hardcoded `k=50`** — extract to module constant.
+- **N+1 lookups (`get_bot`/`get_chat`/`get_scene` per row)** — fine at `k=50`, revisit if `k` grows.
+- **FTS highlighting via `snippet()`** — Phase 4 skipped this; UX nice-to-have.
+- **Result links chat-level only** — `memories` table has no `event_id` column; deep-linking to specific turn requires schema addition.
+
+#### Deferred items
+
+- **sqlite-vec swap** when host Python supports `enable_load_extension`.
+- **Real embedding model** with proper semantic similarity.
+- **Branching read-side filter**: T89 ships data-model + UI but event readers don't yet consult `is_active`. Each branch is metadata-only labeled ranges. Consult-on-read is Phase 4.5+ work.
+- **Bulk significance re-rate** in drawer (T98.2 deferred — only per-memory edit shipped).
+- **Vector index optimization** (HNSW) — only relevant if memory counts grow past pure-Python feasibility.
+- **`scene-close-on-cancel` UX revisit** (Phase 2.5 carry-over).
+- **Cross-feature canned-queue brittleness fixture builder** (Phase 3 carry-over).
+- **Full lifecycle-rollback in regenerate** — Phase 3.5 T83.4 shipped a warning log; proper rollback needs schema-level back-references (`triggered_by_assistant_turn_id` payload field).

chat/app.py

@@ -32,7 +32,9 @@ from chat.web.drawer import router as drawer_router
 from chat.web.kickoff import router as kickoff_router
 from chat.web.middleware import FirstRunRedirectMiddleware
 from chat.web.nav import router as nav_router
+from chat.web.search import router as search_router
 from chat.web.settings import router as settings_router
+from chat.web.snapshots import router as snapshots_router
 from chat.web.sse import router as sse_router
 from chat.web.turns import router as turns_router
 
@@ -137,9 +139,11 @@ async def http_exception_handler(request: Request, exc: StarletteHTTPException):
 app.include_router(bots_router)
 app.include_router(kickoff_router)
 app.include_router(settings_router)
+app.include_router(snapshots_router)
 app.include_router(nav_router)
 app.include_router(chat_router)
 app.include_router(drawer_router)
+app.include_router(search_router)
 app.include_router(sse_router)
 app.include_router(turns_router)
 

chat/state/manual_edit.py

@@ -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

@@ -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/templates/layout.html

@@ -5,8 +5,16 @@
   <ul>
     <li><a href="/chats" class="{% if active_nav == 'chats' %}active{% endif %}">Chats</a></li>
     <li><a href="/bots" class="{% if active_nav == 'bots' %}active{% endif %}">Bots</a></li>
+    <li><a href="/snapshots" class="{% if active_nav == 'snapshots' %}active{% endif %}">Snapshots</a></li>
     <li><a href="/settings" class="{% if active_nav == 'settings' %}active{% endif %}">Settings</a></li>
   </ul>
+  {# T100: cross-chat search box. GET /search so the URL is shareable
+     and back-button friendly; the results page itself re-renders this
+     form with the query pre-filled. #}
+  <form class="rail-search" action="/search" method="get" role="search">
+    <input type="search" name="q" placeholder="Search" aria-label="Search memories">
+    <button type="submit">Go</button>
+  </form>
 </nav>
 <main class="content">
   {% block content %}{% endblock %}

chat/templates/search.html

@@ -0,0 +1,37 @@
+{% extends "layout.html" %}
+{% block title %}Search - chat{% endblock %}
+{% block content %}
+<header class="page-header">
+  <h1>Search</h1>
+</header>
+
+<form class="search-page-form" action="/search" method="get">
+  <input type="text" name="q" value="{{ query|default('', true) }}"
+         placeholder="Search memories across all chats" autofocus>
+  <button type="submit">Search</button>
+</form>
+
+{% if not query %}
+  {# Empty-state placeholder: the top-bar form submits to /search even
+     with no input, so this page must render cleanly with no query. #}
+  <p class="muted search-empty">Enter a query to search memories across all chats.</p>
+{% elif not results %}
+  <p class="muted">No matches for &ldquo;{{ query }}&rdquo;.</p>
+{% else %}
+  <ul class="search-results">
+    {% for r in results %}
+    <li class="search-result">
+      <a class="search-result-link" href="/chats/{{ r.chat_id }}">
+        <div class="search-result-meta muted">
+          <strong>{{ r.owner_name }}</strong>
+          <span>&middot; {{ r.chat_id }}</span>
+          {% if r.chat_name %}<span>&middot; {{ r.chat_name }}</span>{% endif %}
+          {% if r.scene_label %}<span>&middot; scene {{ r.scene_label }}</span>{% endif %}
+        </div>
+        <div class="search-result-summary">{{ r.pov_summary }}</div>
+      </a>
+    </li>
+    {% endfor %}
+  </ul>
+{% endif %}
+{% endblock %}

chat/templates/snapshots.html

@@ -0,0 +1,66 @@
+{% extends "layout.html" %}
+{% block title %}Snapshots - chat{% endblock %}
+{% block content %}
+<header class="page-header">
+    <h1>Snapshots</h1>
+    <form method="post" action="/snapshots/take" class="inline-edit">
+        <button type="submit">Take snapshot now</button>
+    </form>
+</header>
+
+{% if preview %}
+<section class="snapshot-preview">
+    <h2>Preview: {{ preview.snapshot_id }}</h2>
+    <dl>
+        <dt>kind</dt><dd>{{ preview.kind }}</dd>
+        <dt>filename</dt><dd>{{ preview.filename }}</dd>
+        <dt>file size (bytes)</dt><dd>{{ preview.file_size_bytes }}</dd>
+        <dt>snapshot last_event_id</dt><dd>{{ preview.last_event_id }}</dd>
+        <dt>current event_log max id</dt><dd>{{ preview.current_event_log_max_id }}</dd>
+        <dt>events since snapshot</dt><dd>{{ preview.event_delta }}</dd>
+        <dt>events stored in snapshot</dt><dd>{{ preview.event_log_rows_in_snapshot }}</dd>
+    </dl>
+</section>
+{% endif %}
+
+{% if snapshots %}
+<table class="snapshot-list">
+    <thead>
+        <tr>
+            <th>ID</th>
+            <th>Kind</th>
+            <th>Created (UTC)</th>
+            <th>Size (bytes)</th>
+            <th>last_event_id</th>
+            <th>Actions</th>
+        </tr>
+    </thead>
+    <tbody>
+        {% for snap in snapshots %}
+        <tr>
+            <td>{{ snap.snapshot_id }}</td>
+            <td>{{ snap.kind }}</td>
+            <td>{{ snap.created_at }}</td>
+            <td>{{ snap.file_size_bytes }}</td>
+            <td>{{ snap.last_event_id if snap.last_event_id is not none else '?' }}</td>
+            <td>
+                <a href="/snapshots/{{ snap.snapshot_id }}/preview?kind={{ snap.kind }}">Preview</a>
+                <details class="snapshot-row-restore">
+                    <summary>Restore</summary>
+                    <form method="post" action="/snapshots/restore/{{ snap.snapshot_id }}" class="inline-edit">
+                        <input type="hidden" name="kind" value="{{ snap.kind }}">
+                        <label>Type "{{ snap.snapshot_id }}" to confirm:
+                            <input type="text" name="confirm_id" required>
+                        </label>
+                        <button type="submit">Restore from this snapshot</button>
+                    </form>
+                </details>
+            </td>
+        </tr>
+        {% endfor %}
+    </tbody>
+</table>
+{% else %}
+<p class="muted">No snapshots yet. Use "Take snapshot now" to create one.</p>
+{% endif %}
+{% endblock %}

chat/web/drawer.py

@@ -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)

chat/web/search.py

@@ -0,0 +1,92 @@
+"""T100 (Phase 4): cross-chat search UX route.
+
+Wraps T93's :func:`chat.services.cross_chat_search.search_all_memories`
+in a small read-only HTML surface so the top-bar search input has
+somewhere to land. The route does no filtering of its own beyond the
+empty-query fast-path that T93 already implements; ranking, owner
+scope, and witness scope all live in the service layer.
+
+For each match we hydrate just enough metadata to render a row:
+* the owner bot's display name (so users see "BOTA" not "bot_a"),
+* the originating ``chat_id`` (the link target — there's no per-turn
+  anchor today because memories don't carry an ``event_id`` column,
+  so we deep-link to the chat as a whole),
+* the originating scene title when one exists,
+* and the ``pov_summary`` itself.
+
+We deliberately keep this module synchronous and template-only — no
+HTMX swaps, no JSON API — because the search box is a "leave the
+current chat to look something up" surface, not an inline drawer.
+"""
+
+from __future__ import annotations
+
+from pathlib import Path
+
+from fastapi import APIRouter, Depends, Request
+from fastapi.responses import HTMLResponse
+from fastapi.templating import Jinja2Templates
+
+from chat.services.cross_chat_search import search_all_memories
+from chat.state.entities import get_bot
+from chat.state.world import get_chat, get_scene
+from chat.web.bots import get_conn
+
+TEMPLATES = Jinja2Templates(
+    directory=str(Path(__file__).resolve().parent.parent / "templates")
+)
+
+router = APIRouter()
+
+
+@router.get("/search", response_class=HTMLResponse)
+async def search(request: Request, q: str = "", conn=Depends(get_conn)):
+    """Render ``search.html`` with up to 50 cross-chat FTS matches.
+
+    ``q`` is intentionally allowed to be empty — that path renders the
+    page's "enter a query" placeholder rather than a 400, because the
+    top-bar form submits to this URL even with an empty input. T93's
+    service short-circuits whitespace-only queries to ``[]`` so there
+    is no FTS5 ``MATCH ''`` syntax error to guard against here.
+    """
+    raw_results = search_all_memories(conn, query=q, k=50) if q else []
+
+    # Hydrate display fields per row. We do this in the route (not the
+    # service) so the service stays a pure FTS shim that other UIs
+    # can reuse.
+    results = []
+    for row in raw_results:
+        bot = get_bot(conn, row["owner_id"])
+        chat = get_chat(conn, row["chat_id"])
+        scene = get_scene(conn, row["scene_id"]) if row["scene_id"] else None
+        results.append(
+            {
+                "memory_id": row["memory_id"],
+                "owner_id": row["owner_id"],
+                "owner_name": bot["name"] if bot else row["owner_id"],
+                "chat_id": row["chat_id"],
+                "chat_name": (
+                    chat.get("narrative_anchor") if chat else None
+                ),
+                "scene_id": row["scene_id"],
+                # Scenes have no ``title`` column today; surface the
+                # ``started_at`` timestamp as a human-friendly label
+                # when a scene is set, otherwise leave it blank.
+                "scene_label": (
+                    scene.get("started_at") if scene else None
+                ),
+                "pov_summary": row["pov_summary"],
+                "significance": row["significance"],
+                "ts": row["ts"],
+            }
+        )
+
+    return TEMPLATES.TemplateResponse(
+        request,
+        "search.html",
+        {
+            "query": q,
+            "results": results,
+            "active_nav": "search",
+        },
+    )

chat/web/snapshots.py

@@ -0,0 +1,190 @@
+"""Snapshot UX routes (T99).
+
+Surfaces the existing snapshot service (``chat/services/snapshot.py``)
+through HTML so the user can see, take, restore, and preview snapshots
+without dropping to a shell.
+
+Routes:
+
+* ``GET  /snapshots``                    list all snapshots (both kinds)
+* ``POST /snapshots/take``               take a periodic snapshot now
+* ``POST /snapshots/restore/{id}``       restore (requires matching ``confirm_id``)
+* ``GET  /snapshots/{id}/preview``       show metadata + delta vs current
+
+The ``snapshot_id`` is the filename stem (the UTC timestamp written by
+:func:`chat.services.snapshot.take_snapshot`) — there's no separate UUID,
+and the timestamp filename is already unique per snapshot kind. Both
+periodic and rewind snapshots share the same id space lookup-wise, so
+the restore + preview routes accept ``kind`` as a form/query param to
+disambiguate.
+"""
+
+from __future__ import annotations
+
+import json
+from pathlib import Path
+
+from fastapi import APIRouter, Depends, Form, HTTPException, Request
+from fastapi.responses import HTMLResponse, RedirectResponse
+from fastapi.templating import Jinja2Templates
+
+from chat.services.snapshot import (
+    restore_from_snapshot,
+    take_snapshot,
+)
+from chat.web.bots import get_conn
+
+TEMPLATES = Jinja2Templates(
+    directory=str(Path(__file__).resolve().parent.parent / "templates")
+)
+
+router = APIRouter()
+
+SNAPSHOT_KINDS = ("periodic", "rewind")
+
+
+def _list_all_snapshots(data_dir: Path) -> list[dict]:
+    """Walk ``data/snapshots/{kind}/`` for both kinds and collect metadata.
+
+    Each entry exposes the fields the template needs: ``snapshot_id``
+    (filename stem), ``kind``, ``created_at`` (file mtime as ISO), the
+    on-disk ``file_size_bytes``, and the snapshot's stored
+    ``last_event_id`` (parsed from the JSON body — small enough that
+    listing isn't a performance concern for the handful of files we keep).
+    """
+    from datetime import datetime, timezone
+
+    rows: list[dict] = []
+    for kind in SNAPSHOT_KINDS:
+        snap_dir = data_dir / "snapshots" / kind
+        if not snap_dir.exists():
+            continue
+        for path in sorted(snap_dir.glob("*.json")):
+            try:
+                dump = json.loads(path.read_text())
+                last_event_id = dump.get("last_event_id", 0)
+            except (OSError, json.JSONDecodeError):
+                # Corrupt or unreadable files still get listed so the
+                # user can see and delete them; just don't crash here.
+                last_event_id = None
+            stat = path.stat()
+            rows.append(
+                {
+                    "snapshot_id": path.stem,
+                    "kind": kind,
+                    "created_at": datetime.fromtimestamp(
+                        stat.st_mtime, tz=timezone.utc
+                    ).isoformat(),
+                    "file_size_bytes": stat.st_size,
+                    "last_event_id": last_event_id,
+                    "filename": path.name,
+                }
+            )
+    # Newest first for display.
+    rows.sort(key=lambda r: r["created_at"], reverse=True)
+    return rows
+
+
+def _resolve_snapshot_path(
+    data_dir: Path, snapshot_id: str, kind: str
+) -> Path:
+    """Map an ``(id, kind)`` pair to the on-disk file, or 404."""
+    if kind not in SNAPSHOT_KINDS:
+        raise HTTPException(status_code=400, detail=f"unknown kind: {kind}")
+    path = data_dir / "snapshots" / kind / f"{snapshot_id}.json"
+    if not path.exists():
+        raise HTTPException(status_code=404, detail="snapshot not found")
+    return path
+
+
+@router.get("/snapshots", response_class=HTMLResponse)
+async def snapshots_list(request: Request):
+    settings = request.app.state.settings
+    rows = _list_all_snapshots(settings.data_dir)
+    return TEMPLATES.TemplateResponse(
+        request,
+        "snapshots.html",
+        {"snapshots": rows, "active_nav": "snapshots"},
+    )
+
+
+@router.post("/snapshots/take")
+async def snapshots_take(request: Request, conn=Depends(get_conn)):
+    """Take a periodic snapshot now.
+
+    We use ``kind="periodic"`` for manual snapshots since they're
+    user-initiated checkpoints, not pre-rewind safety dumps. They count
+    against the 5-snapshot retention but that's fine — manual ones are
+    the most recent so they're the last to be pruned.
+    """
+    settings = request.app.state.settings
+    take_snapshot(conn, data_dir=settings.data_dir, kind="periodic")
+    return RedirectResponse(url="/snapshots", status_code=303)
+
+
+@router.post("/snapshots/restore/{snapshot_id}")
+async def snapshots_restore(
+    snapshot_id: str,
+    request: Request,
+    confirm_id: str = Form(""),
+    kind: str = Form("periodic"),
+    conn=Depends(get_conn),
+):
+    """Hard-confirm restore: ``confirm_id`` must equal the path id.
+
+    Mismatched confirm → 400 (without touching the DB). On match, the
+    existing :func:`restore_from_snapshot` clears projected tables and
+    re-loads them from the dump.
+    """
+    if confirm_id != snapshot_id:
+        raise HTTPException(
+            status_code=400,
+            detail="confirm_id does not match snapshot id",
+        )
+    settings = request.app.state.settings
+    path = _resolve_snapshot_path(settings.data_dir, snapshot_id, kind)
+    restore_from_snapshot(conn, path)
+    return RedirectResponse(url="/snapshots", status_code=303)
+
+
+@router.get("/snapshots/{snapshot_id}/preview", response_class=HTMLResponse)
+async def snapshots_preview(
+    snapshot_id: str,
+    request: Request,
+    kind: str = "periodic",
+    conn=Depends(get_conn),
+):
+    """Show snapshot metadata + a basic delta against the current event log.
+
+    Phase 4 keeps this simple: the snapshot's ``last_event_id`` plus the
+    current ``MAX(event_log.id)`` is enough to tell the user how far the
+    log has moved on. A richer per-table diff is a Phase 4.5+ concern.
+    """
+    settings = request.app.state.settings
+    path = _resolve_snapshot_path(settings.data_dir, snapshot_id, kind)
+    dump = json.loads(path.read_text())
+    last_event_id = dump.get("last_event_id", 0)
+
+    cur = conn.execute("SELECT MAX(id) FROM event_log")
+    row = cur.fetchone()
+    current_max_id = row[0] if row[0] is not None else 0
+
+    stat = path.stat()
+    return TEMPLATES.TemplateResponse(
+        request,
+        "snapshots.html",
+        {
+            "snapshots": _list_all_snapshots(settings.data_dir),
+            "active_nav": "snapshots",
+            "preview": {
+                "snapshot_id": snapshot_id,
+                "kind": kind,
+                "filename": path.name,
+                "file_size_bytes": stat.st_size,
+                "last_event_id": last_event_id,
+                "current_event_log_max_id": current_max_id,
+                "event_delta": current_max_id - last_event_id,
+                "event_log_rows_in_snapshot": len(dump.get("event_log", [])),
+            },
+        },
+    )

docs/plans/2026-04-26-v1-requirements-design.md

@@ -520,6 +520,8 @@ Written per witness when a scene closes. Different details, different interpreta
 
 ### Phase 4 — polish
 
+**Status: shipped 2026-04-27** (T88–T102, 15 tasks across 8 waves; +70 tests). See "Phase 4 status" in CLAUDE.md for the per-task breakdown. Vector retrieval shipped via pure-Python cosine over a JSON-blob embeddings table (sqlite-vec deferred — host Python lacks loadable extensions); branching is data-model + drawer UI; significance review, hide-from-view soft delete, surgical delete with cascade preview, snapshot UX, and cross-chat search all surface from the drawer or top-bar.
+
 - Vector retrieval (sqlite-vss or sqlite-vec).
 - Branching UI.
 - Drawer-edit on every field.

tests/test_drawer_phase4.py

@@ -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"

tests/test_phase4_integration.py

@@ -1,20 +1,38 @@
-"""Phase 4 cross-feature integration tests (T97 follow-up).
+"""Phase 4 cross-feature integration tests (T97 follow-up + T101).
 
-Wave 8 / T101 will populate this file with the full Phase 4 retrieval +
-embedding integration suite. For now this houses a single test pinning
-the T97.5 wiring: the production turn route plumbs ``app=request.app``
-all the way through ``record_turn_memory_for_present`` so the embedding
-worker actually receives jobs in production. Without this fix-up the
-plumbing added in T97 was dormant — every per-witness write took the
-no-app branch and silently dropped the embed enqueue.
+Cross-feature flows for the Phase 4 retrieval + branching + drawer
+features. Each test drives multiple Phase 4 surfaces end-to-end and
+asserts both event_log and projected-state outcomes.
 
-The test monkeypatches ``app.state.embedding_worker.enqueue`` to record
-jobs (rather than draining the worker mid-test) so the assertion is
-deterministic and free of asyncio-timing flakiness inside FastAPI's
-TestClient. The bug we're guarding against is "did the call site pass
-``app`` at all" — the worker's drain path is exercised in
-:mod:`tests.test_embedding_worker`, so duplicating that here would add
-no coverage.
+Test inventory:
+
+* ``test_post_turn_embeddings_indexed_via_worker_hook`` (T97.5) —
+  pins the production turn route's ``app=request.app`` plumbing so
+  the embedding worker actually receives jobs.
+
+T101 additions (the "Phase 4 cross-feature integration" suite):
+
+1. ``test_vector_retrieval_feedback_loop`` — write a memory, drain
+   the embedding worker, assert the vector path retrieves it.
+2. ``test_branch_diverge_main_intact`` — create a branch from a
+   mid-log turn, switch, append more events, switch back and assert
+   the original log past the branch point is still present (Phase 4
+   branching is metadata-only — no read-side filter yet).
+3. ``test_surgical_delete_truncates_log_and_writes_snapshot`` —
+   compute impact, confirm via the drawer route, assert the log was
+   truncated and a pre-rewind snapshot landed on disk.
+4. ``test_hide_then_unhide_round_trip_through_read_recent_dialogue``
+   — flip ``hidden`` via the drawer route both directions and assert
+   ``read_recent_dialogue`` honours the flag in real time.
+5. ``test_cross_chat_search_surfaces_memories_in_three_chats`` —
+   write memories in 3 chats, hit ``/search?q=...`` and assert all
+   three appear.
+
+The T97.5 test monkeypatches ``app.state.embedding_worker.enqueue`` to
+record jobs (rather than draining the worker) because the bug it pins
+is "did the call site pass ``app`` at all". T101 test 1 takes the
+opposite tack: it drives the worker for real to verify the entire
+write -> index -> retrieve loop.
 """
 
 from __future__ import annotations
@@ -27,7 +45,7 @@ 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.log import append_and_apply, append_event
 from chat.eventlog.projector import project
 from chat.llm.mock import MockLLMClient
 
@@ -178,3 +196,696 @@ def test_post_turn_embeddings_indexed_via_worker_hook(
             ).fetchall()
         ]
     assert job.memory_id in memory_ids
+
+
+# ---------------------------------------------------------------------------
+# T101 — Phase 4 cross-feature integration suite.
+# ---------------------------------------------------------------------------
+#
+# Helpers + the five required scenarios. Each test drives multiple Phase 4
+# features so a regression in any one of them fails an integration check.
+
+
+def _seed_minimal_chat(db_path: Path, chat_id: str = "chat_bot_a") -> None:
+    """Seed bot_a, you, a chat, edges, and activities — same shape as
+    ``tests/test_phase3_integration.py::_seed_single_bot_chat`` but
+    parameterised on chat_id so the cross-chat search test can stamp
+    several chats in the same database without renaming bots.
+
+    Uses ``append_and_apply`` rather than ``append_event`` + a final
+    ``project`` so successive calls (e.g. one per chat in the
+    cross-chat-search test) don't try to re-project the cumulative
+    log and trip the ``chats.id`` UNIQUE constraint on the prior
+    chat's row.
+    """
+    with open_db(db_path) as conn:
+        existing_bot = conn.execute(
+            "SELECT 1 FROM bots WHERE id = 'bot_a'"
+        ).fetchone()
+        if existing_bot is None:
+            append_and_apply(
+                conn,
+                kind="bot_authored",
+                payload={
+                    "id": "bot_a",
+                    "name": "BotA",
+                    "persona": "thoughtful",
+                    "voice_samples": [],
+                    "traits": [],
+                    "backstory": "",
+                    "initial_relationship_to_you": "",
+                    "kickoff_prose": "...",
+                },
+            )
+            append_and_apply(
+                conn,
+                kind="you_authored",
+                payload={
+                    "name": "Me",
+                    "pronouns": "they/them",
+                    "persona": "",
+                },
+            )
+        append_and_apply(
+            conn,
+            kind="chat_created",
+            payload={
+                "id": chat_id,
+                "host_bot_id": "bot_a",
+                "initial_time": "2026-04-26T20:00:00+00:00",
+                "narrative_anchor": "Day 1",
+                "weather": "",
+            },
+        )
+        append_and_apply(
+            conn,
+            kind="edge_update",
+            payload={
+                "source_id": "bot_a",
+                "target_id": "you",
+                "chat_id": chat_id,
+                "knowledge_facts": [],
+            },
+        )
+        # Activities are unique per (entity_id) — only seed them on the
+        # first call (when the bot row is also fresh).
+        if existing_bot is None:
+            for entity_id, verb in [
+                ("you", "talking"),
+                ("bot_a", "listening"),
+            ]:
+                append_and_apply(
+                    conn,
+                    kind="activity_change",
+                    payload={
+                        "entity_id": entity_id,
+                        "posture": "sitting",
+                        "action": {
+                            "verb": verb,
+                            "interruptible": True,
+                            "required_attention": "low",
+                            "expected_duration": "ongoing",
+                        },
+                        "attention": "",
+                        "holding": [],
+                        "status": {},
+                    },
+                )
+
+
+# ---------------------------------------------------------------------------
+# 1. Vector retrieval feedback loop.
+# ---------------------------------------------------------------------------
+
+
+async def test_vector_retrieval_feedback_loop(tmp_path):
+    """End-to-end: write a memory through
+    :func:`record_turn_memory_for_present` so an :class:`EmbeddingJob`
+    lands on a worker, drain the worker, then call
+    :func:`vector_search` with the SAME pseudo-embedding function and
+    assert the just-written memory is the top hit.
+
+    Why this test does NOT use the TestClient fixture: the live
+    ``app.state.embedding_worker`` is created inside the FastAPI
+    lifespan's event loop. ``await``-ing on it from pytest-asyncio's
+    loop trips ``"got Future attached to a different loop"``. We
+    instead spin up a fresh :class:`EmbeddingWorker` in the test
+    loop, exactly mirroring ``tests/test_embedding_worker.py``'s
+    pattern. The T97.5 test above pins the wiring between the live
+    HTTP route and the live app worker; this test pins the
+    write -> index -> retrieve loop with no transport in scope.
+
+    Cross-feature gaps this test catches:
+    * Memory write enqueues to the worker but the worker never
+      drains (e.g. ``_run`` deadlock or sentinel mishandled).
+    * Worker uses a different embedding function than
+      ``vector_search`` at query time, producing different vectors
+      and breaking cosine retrieval.
+    * ``embeddings`` projector handler is not registered (e.g.
+      import ordering bug) so the event fires but the table stays
+      empty.
+    """
+    from types import SimpleNamespace
+
+    from chat.db.migrate import apply_migrations
+    from chat.services.embedding_worker import EmbeddingWorker
+    from chat.services.embeddings import generate_embedding
+    from chat.services.memory_write import record_turn_memory_for_present
+    from chat.services.vector_search import vector_search
+
+    # Trigger projector handler registration. ``record_turn_memory_for_present``
+    # imports memory_write which imports the worker module, but the
+    # projector handlers live in ``chat.state.*`` modules and are
+    # registered as a side effect of import.
+    import chat.state.embeddings  # noqa: F401
+    import chat.state.entities  # noqa: F401
+    import chat.state.memory  # noqa: F401
+    import chat.state.world  # noqa: F401
+
+    db = tmp_path / "test.db"
+    apply_migrations(db)
+    _seed_minimal_chat(db)
+
+    # Spin up our own worker in the test event loop. ``client=None``
+    # is fine for the pseudo-embedding path — the local hash function
+    # does not require an LLM client.
+    worker = EmbeddingWorker(
+        conn_factory=lambda: open_db(db),
+        client=None,
+    )
+    await worker.start()
+
+    # Stub ``app`` — only ``app.state.embedding_worker`` is read by
+    # ``_write_one_memory``. SimpleNamespace gives us a stand-in that
+    # exposes ``state.embedding_worker`` without the full FastAPI app.
+    fake_app = SimpleNamespace(state=SimpleNamespace(embedding_worker=worker))
+
+    distinctive_text = "Maya watched the gondola lights drift across the lagoon."
+    with open_db(db) as conn:
+        record_turn_memory_for_present(
+            conn,
+            chat_id="chat_bot_a",
+            host_bot_id="bot_a",
+            guest_bot_id=None,
+            narrative_text=distinctive_text,
+            app=fake_app,
+        )
+
+    # Drain the worker via the sentinel. After this returns the
+    # ``embedding_indexed`` event has been projected.
+    await worker.stop()
+
+    # Generate a query embedding using the same function the worker
+    # used. The pseudo-embedding is deterministic so a query equal to
+    # the indexed text produces the identical vector and a cosine
+    # similarity of 1.0.
+    query_result = await generate_embedding(client=None, text=distinctive_text)
+
+    with open_db(db) as conn:
+        emb_count = conn.execute(
+            "SELECT COUNT(*) FROM embeddings"
+        ).fetchone()[0]
+        assert emb_count == 1, (
+            "embedding worker did not project an embedding_indexed event"
+        )
+
+        hits = vector_search(
+            conn,
+            owner_id="bot_a",
+            witness_role="host",  # bot_a is host, witness_host=1 by default
+            query_vector=query_result.vector,
+            k=4,
+        )
+        assert len(hits) == 1
+        top = hits[0]
+        assert top["pov_summary"] == distinctive_text
+        # Self-match: cosine of identical vectors is 1.0.
+        assert top["score"] == pytest.approx(1.0, abs=1e-9)
+
+
+# ---------------------------------------------------------------------------
+# 2. Branch + diverge: main's post-branch tail stays intact (Phase 4
+#    branches are metadata-only).
+# ---------------------------------------------------------------------------
+
+
+def test_branch_diverge_main_intact(app_state_setup, tmp_path):
+    """Append turns 1-12 on main, branch from turn 10's event_id, switch
+    to the new branch, append 3 more "play" turns, switch back to main,
+    assert the original turn 11+ events are untouched.
+
+    Phase 4's branches table is metadata-only — the read-side filter
+    isn't wired yet, so all events live in one log regardless of which
+    branch is "active". This test pins that contract: switching does
+    not mutate or hide existing events on either branch.
+
+    Canned LLM queue: none. ``user_turn`` / ``assistant_turn`` are
+    transcript-only kinds with no projector handler that needs an
+    LLM call, and ``branch_created`` / ``branch_switched`` are pure
+    state events. We use ``append_and_apply`` directly rather than
+    driving the HTTP turn route, which would require a 6-slot canned
+    queue per turn (parse + narrative + 2 state-updates + scene-close
+    + memory) for 15 turns total = 90 slots of plumbing irrelevant to
+    the branch contract.
+    """
+    from chat.services.branching import branch_from_event, switch_active_branch
+    from chat.state.branches import active_branch
+
+    db = tmp_path / "test.db"
+    _seed_minimal_chat(db)
+
+    # Append 12 user_turn / assistant_turn pairs on main. We collect
+    # the assistant_turn id at index 10 (1-based: "turn 10") so the
+    # branch fork point is unambiguous.
+    main_turn_ids: list[int] = []
+    with open_db(db) as conn:
+        for i in range(1, 13):
+            user_id = append_and_apply(
+                conn,
+                kind="user_turn",
+                payload={
+                    "chat_id": "chat_bot_a",
+                    "prose": f"main turn {i}",
+                    "segments": [],
+                },
+            )
+            asst_id = append_and_apply(
+                conn,
+                kind="assistant_turn",
+                payload={
+                    "chat_id": "chat_bot_a",
+                    "speaker_id": "bot_a",
+                    "text": f"main reply {i}",
+                    "truncated": False,
+                    "user_turn_id": user_id,
+                },
+            )
+            main_turn_ids.append(asst_id)
+        turn_10_id = main_turn_ids[9]
+
+        # Snapshot the post-turn-10 main tail (turns 11, 12 + their
+        # user_turn predecessors) so we can byte-compare after the
+        # round-trip.
+        main_tail_before = conn.execute(
+            "SELECT id, kind, payload_json, hidden, superseded_by "
+            "FROM event_log WHERE id > ? ORDER BY id",
+            (turn_10_id,),
+        ).fetchall()
+        assert len(main_tail_before) == 4  # 2 user + 2 assistant past turn 10
+
+        # Branch from turn 10. Phase 4's helper validates the origin
+        # event id exists and emits ``branch_created``.
+        branch_from_event(
+            conn,
+            name="experiment",
+            origin_event_id=turn_10_id,
+            chat_id="chat_bot_a",
+        )
+        switch_active_branch(conn, name="experiment")
+        active = active_branch(conn)
+        assert active is not None and active["name"] == "experiment"
+
+        # Play 3 turns on the experiment branch.
+        for i in range(1, 4):
+            user_id = append_and_apply(
+                conn,
+                kind="user_turn",
+                payload={
+                    "chat_id": "chat_bot_a",
+                    "prose": f"experiment turn {i}",
+                    "segments": [],
+                },
+            )
+            append_and_apply(
+                conn,
+                kind="assistant_turn",
+                payload={
+                    "chat_id": "chat_bot_a",
+                    "speaker_id": "bot_a",
+                    "text": f"experiment reply {i}",
+                    "truncated": False,
+                    "user_turn_id": user_id,
+                },
+            )
+
+        # Switch back to main.
+        switch_active_branch(conn, name="main")
+        active2 = active_branch(conn)
+        assert active2 is not None and active2["name"] == "main"
+
+        # Main's original tail past turn 10 is byte-identical: the
+        # branching events (branch_created, branch_switched x2) and the
+        # 3 experiment turns sit AFTER the original tail in event_log
+        # order, never overwriting it.
+        main_tail_after = conn.execute(
+            "SELECT id, kind, payload_json, hidden, superseded_by "
+            "FROM event_log "
+            "WHERE id > ? AND id <= ? ORDER BY id",
+            (turn_10_id, main_turn_ids[-1]),
+        ).fetchall()
+        assert main_tail_after == main_tail_before
+
+        # The 6 experiment events (3 user + 3 assistant) all live in
+        # the same log past the original main tail. Verify their
+        # prose payloads to disambiguate from main's content.
+        diverged = conn.execute(
+            "SELECT kind, json_extract(payload_json, '$.prose'), "
+            "  json_extract(payload_json, '$.text') "
+            "FROM event_log WHERE id > ? "
+            "  AND kind IN ('user_turn', 'assistant_turn') ORDER BY id",
+            (main_turn_ids[-1],),
+        ).fetchall()
+        assert len(diverged) == 6
+        prose_or_text = [(row[1] or row[2]) for row in diverged]
+        # Sequence: user1, asst1, user2, asst2, user3, asst3.
+        assert "experiment turn 1" in prose_or_text
+        assert "experiment reply 1" in prose_or_text
+        assert "experiment turn 3" in prose_or_text
+        assert "experiment reply 3" in prose_or_text
+
+
+# ---------------------------------------------------------------------------
+# 3. Surgical delete: impact preview -> confirm -> log truncated +
+#    pre-rewind snapshot saved.
+# ---------------------------------------------------------------------------
+
+
+def test_surgical_delete_truncates_log_and_writes_snapshot(
+    app_state_setup, tmp_path
+):
+    """Compute the delete-impact for a turn (read-only preview), then
+    confirm via the POST drawer route. Assert:
+
+    * The preview returns 200 + cascade markup.
+    * The event_log is physically truncated past ``target_id - 1``.
+    * A snapshot file lands under ``<data_dir>/snapshots/rewind/``.
+    * The pre-rewind snapshot's ``last_event_id`` matches the high
+      water mark BEFORE the truncate (so recovery can replay back to
+      pre-delete state).
+
+    Snapshot location: T97.5's ``data_dir`` derives from the db's
+    parent directory when ``CHAT_DATA_DIR`` is unset. The fixture
+    sets ``CHAT_DB_PATH = tmp_path / "test.db"`` so the snapshot
+    parent is ``tmp_path / "snapshots" / "rewind"``.
+
+    No canned LLM queue — the preview is pure SQL and the rewind path
+    is also pure SQL (delete + reproject). The drawer routes don't
+    invoke the LLM.
+    """
+    import json as _json
+
+    db = tmp_path / "test.db"
+    _seed_minimal_chat(db)
+
+    # Append a small fixed turn sequence we can predict the cascade for.
+    with open_db(db) as conn:
+        first_user = append_and_apply(
+            conn,
+            kind="user_turn",
+            payload={
+                "chat_id": "chat_bot_a",
+                "prose": "first message",
+                "segments": [],
+            },
+        )
+        append_and_apply(
+            conn,
+            kind="assistant_turn",
+            payload={
+                "chat_id": "chat_bot_a",
+                "speaker_id": "bot_a",
+                "text": "first reply",
+                "truncated": False,
+                "user_turn_id": first_user,
+            },
+        )
+        target_user = append_and_apply(
+            conn,
+            kind="user_turn",
+            payload={
+                "chat_id": "chat_bot_a",
+                "prose": "this turn will be deleted",
+                "segments": [],
+            },
+        )
+        target_asst = append_and_apply(
+            conn,
+            kind="assistant_turn",
+            payload={
+                "chat_id": "chat_bot_a",
+                "speaker_id": "bot_a",
+                "text": "and so will this reply",
+                "truncated": False,
+                "user_turn_id": target_user,
+            },
+        )
+        # One trailing event past the target so we can verify the
+        # cascade catches >1 event.
+        trailing = append_and_apply(
+            conn,
+            kind="user_turn",
+            payload={
+                "chat_id": "chat_bot_a",
+                "prose": "trailing context",
+                "segments": [],
+            },
+        )
+        max_id_before = conn.execute(
+            "SELECT MAX(id) FROM event_log"
+        ).fetchone()[0]
+
+    # ---- Preview: GET delete-preview returns 200 + the cascade list. ----
+    preview = app_state_setup.get(
+        f"/chats/chat_bot_a/drawer/turn/delete-preview/{target_user}"
+    )
+    assert preview.status_code == 200
+    body = preview.text
+    assert "delete-impact-modal" in body
+    assert f"Delete event {target_user}?" in body
+    assert "user_turn" in body
+    assert "assistant_turn" in body
+    # Confirm form points at the delete route.
+    assert f"/drawer/turn/delete/{target_user}" in body
+
+    # ---- Confirm: POST delete drops user, assistant, AND trailing. ----
+    confirm = app_state_setup.post(
+        f"/chats/chat_bot_a/drawer/turn/delete/{target_user}"
+    )
+    assert confirm.status_code == 200
+
+    # ---- Event log truncated past target_user - 1. ----
+    with open_db(db) as conn:
+        max_id_after = conn.execute(
+            "SELECT MAX(id) FROM event_log"
+        ).fetchone()[0]
+        # delete_turn passes ``after_event_id = target_user - 1`` so
+        # everything from target_user forward is gone.
+        assert max_id_after == target_user - 1
+        for ev_id in (target_user, target_asst, trailing):
+            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"
+
+    # ---- Pre-rewind snapshot landed on disk. ----
+    snapshot_dir = tmp_path / "snapshots" / "rewind"
+    assert snapshot_dir.exists(), (
+        f"snapshot dir not created: {snapshot_dir}"
+    )
+    snapshots = sorted(snapshot_dir.glob("*.json"))
+    assert len(snapshots) >= 1, (
+        f"no rewind snapshot written under {snapshot_dir}"
+    )
+    # Most-recent snapshot's last_event_id == pre-truncate high water
+    # mark, so a "restore" path could fully reverse the delete.
+    latest_snapshot = snapshots[-1]
+    snap_data = _json.loads(latest_snapshot.read_text())
+    assert snap_data["last_event_id"] == max_id_before
+
+
+# ---------------------------------------------------------------------------
+# 4. Hide + retrieval: drawer hide drops a turn from read_recent_dialogue,
+#    unhide restores it.
+# ---------------------------------------------------------------------------
+
+
+def test_hide_then_unhide_round_trip_through_read_recent_dialogue(
+    app_state_setup, tmp_path
+):
+    """Drive a hide -> read -> unhide -> read cycle through the drawer
+    HTTP route and assert ``read_recent_dialogue`` flips visibility
+    each step. T98.3 wires the route; T55 / turn_common owns the
+    ``hidden = 0`` filter.
+
+    Cross-feature: the drawer HTTP handler emits a ``manual_edit``
+    event with branch ``turn_hidden``, the manual_edit projector
+    flips ``event_log.hidden``, and the prompt-window reader filters
+    on that column. Three layers — any one breaking would fail this
+    test.
+
+    No canned LLM queue — hide/unhide are pure SQL routes.
+    """
+    from chat.services.turn_common import read_recent_dialogue
+
+    db = tmp_path / "test.db"
+    _seed_minimal_chat(db)
+
+    with open_db(db) as conn:
+        user_a = append_and_apply(
+            conn,
+            kind="user_turn",
+            payload={
+                "chat_id": "chat_bot_a",
+                "prose": "first user line",
+                "segments": [],
+            },
+        )
+        asst_a = append_and_apply(
+            conn,
+            kind="assistant_turn",
+            payload={
+                "chat_id": "chat_bot_a",
+                "speaker_id": "bot_a",
+                "text": "first reply",
+                "truncated": False,
+                "user_turn_id": user_a,
+            },
+        )
+        user_b = append_and_apply(
+            conn,
+            kind="user_turn",
+            payload={
+                "chat_id": "chat_bot_a",
+                "prose": "second user line",
+                "segments": [],
+            },
+        )
+        asst_b = append_and_apply(
+            conn,
+            kind="assistant_turn",
+            payload={
+                "chat_id": "chat_bot_a",
+                "speaker_id": "bot_a",
+                "text": "second reply",
+                "truncated": False,
+                "user_turn_id": user_b,
+            },
+        )
+
+        # Baseline: all 4 turns visible.
+        baseline = read_recent_dialogue(conn, "chat_bot_a", limit=10)
+        baseline_ids = {t["event_id"] for t in baseline}
+        assert {user_a, asst_a, user_b, asst_b} <= baseline_ids
+
+    # ---- Hide user_b via the drawer route. ----
+    hide_resp = app_state_setup.post(
+        f"/chats/chat_bot_a/drawer/turn/hide/{user_b}",
+        data={"hidden": "1"},
+    )
+    assert hide_resp.status_code == 200
+
+    with open_db(db) as conn:
+        # event_log.hidden flipped.
+        row = conn.execute(
+            "SELECT hidden FROM event_log WHERE id = ?", (user_b,)
+        ).fetchone()
+        assert int(row[0]) == 1
+
+        # read_recent_dialogue drops user_b but keeps the others.
+        after_hide = read_recent_dialogue(conn, "chat_bot_a", limit=10)
+        after_hide_ids = {t["event_id"] for t in after_hide}
+        assert user_b not in after_hide_ids
+        # The other 3 turns still surface.
+        assert {user_a, asst_a, asst_b} <= after_hide_ids
+
+    # ---- Unhide via the SAME route with hidden=0. ----
+    unhide_resp = app_state_setup.post(
+        f"/chats/chat_bot_a/drawer/turn/hide/{user_b}",
+        data={"hidden": "0"},
+    )
+    assert unhide_resp.status_code == 200
+
+    with open_db(db) as conn:
+        row = conn.execute(
+            "SELECT hidden FROM event_log WHERE id = ?", (user_b,)
+        ).fetchone()
+        assert int(row[0]) == 0
+
+        # read_recent_dialogue restores user_b.
+        after_unhide = read_recent_dialogue(conn, "chat_bot_a", limit=10)
+        after_unhide_ids = {t["event_id"] for t in after_unhide}
+        assert {user_a, asst_a, user_b, asst_b} <= after_unhide_ids
+
+        # Two manual_edit events landed (one per toggle), each with the
+        # turn_hidden branch tag.
+        edits = conn.execute(
+            "SELECT payload_json FROM event_log "
+            "WHERE kind = 'manual_edit' "
+            "  AND json_extract(payload_json, '$.target_kind') = 'turn_hidden' "
+            "ORDER BY id"
+        ).fetchall()
+        assert len(edits) == 2
+
+
+# ---------------------------------------------------------------------------
+# 5. Cross-chat search: memories across 3 chats all surface from /search.
+# ---------------------------------------------------------------------------
+
+
+def test_cross_chat_search_surfaces_memories_in_three_chats(
+    app_state_setup, tmp_path
+):
+    """Seed 3 chats each owned by bot_a (so the bot row exists for the
+    search route's display-name hydration), write a distinctive
+    memory in each, then GET ``/search?q=<distinctive>`` and assert
+    every chat appears as a result row.
+
+    Cross-feature: T93's :func:`search_all_memories` (no per-owner
+    filter) + T100's HTML route (display-name hydration via
+    ``get_bot``/``get_chat``). The route's empty-query short-circuit
+    is incidentally exercised by the request setup but isn't the
+    focus.
+
+    No canned LLM queue — memory_written events are projected directly
+    via ``append_and_apply`` and the search route is pure SQL +
+    template rendering.
+    """
+    db = tmp_path / "test.db"
+    # Three chats, all hosted by bot_a so bot_a is the owner of all
+    # three memories. _seed_minimal_chat skips the bot/you bootstrap
+    # after the first call so the cumulative seed is consistent.
+    chat_ids = ["chat_bot_a", "chat_bot_a_2", "chat_bot_a_3"]
+    for chat_id in chat_ids:
+        _seed_minimal_chat(db, chat_id=chat_id)
+
+    # Distinctive token — "wisteria" appears nowhere else in the seed.
+    distinctive = "wisteria"
+    with open_db(db) as conn:
+        for idx, chat_id in enumerate(chat_ids):
+            append_and_apply(
+                conn,
+                kind="memory_written",
+                payload={
+                    "owner_id": "bot_a",
+                    "chat_id": chat_id,
+                    "pov_summary": (
+                        f"the {distinctive} bloomed by the gate (chat {idx})"
+                    ),
+                    "witness_you": 1,
+                    "witness_host": 1,
+                    "witness_guest": 0,
+                    "source": "direct",
+                    "reliability": 1.0,
+                    "significance": 1,
+                    "pinned": 0,
+                    "auto_pinned": 0,
+                },
+            )
+
+    # ---- GET /search?q=wisteria -> all 3 chats appear as result rows. ----
+    response = app_state_setup.get(f"/search?q={distinctive}")
+    assert response.status_code == 200
+    body = response.text
+
+    # Each chat_id appears in a result link href, e.g.
+    # ``href="/chats/chat_bot_a"``. The template renders one
+    # ``<a class="search-result-link" href="/chats/{chat_id}">`` per
+    # row, so a substring match per chat is sufficient.
+    for chat_id in chat_ids:
+        assert f'href="/chats/{chat_id}"' in body, (
+            f"chat {chat_id} missing from /search results: {body!r}"
+        )
+    # The owner display name (BotA) renders for each row — verify >= 3
+    # occurrences so we know all 3 result rows hydrated, not just 1.
+    assert body.count("BotA") >= 3
+
+    # ---- Sanity: distractor query yields no results. ----
+    distractor_response = app_state_setup.get(
+        "/search?q=nonexistentterm12345"
+    )
+    assert distractor_response.status_code == 200
+    distractor_body = distractor_response.text
+    # The "no matches" empty-state copy fires.
+    assert "No matches" in distractor_body
+    for chat_id in chat_ids:
+        assert f'href="/chats/{chat_id}"' not in distractor_body

tests/test_search_ux.py

@@ -0,0 +1,135 @@
+"""T100 (Phase 4): cross-chat search UX (top-bar + results page).
+
+Verifies the FastAPI ``/search`` route that wraps T93's
+``search_all_memories`` service:
+
+* ``/search?q=...`` returns 200 + an HTML page that lists matches drawn
+  from MULTIPLE chats (not just the current one) and links each result
+  back to ``/chats/{chat_id}``.
+* ``/search`` with no query renders the page in its empty state with a
+  "enter a query" placeholder and no result rows (avoids hitting the
+  FTS index with an invalid empty MATCH).
+* Result links navigate to the originating chat so users can pick up
+  the thread where the memory came from.
+"""
+
+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
+import chat.state.memory  # noqa: F401  (registers memory_written handler)
+
+
+@pytest.fixture
+def client(tmp_path, monkeypatch):
+    config_path = tmp_path / "config.toml"
+    config_path.write_text('featherless_api_key = "test"\n')
+    monkeypatch.setenv("CHAT_CONFIG_PATH", str(config_path))
+    monkeypatch.setenv("CHAT_DB_PATH", str(tmp_path / "test.db"))
+    with TestClient(app) as c:
+        yield c
+
+
+def _seed_two_chats_with_memories(db_path: Path) -> None:
+    """Seed: a ``you_entity``, two bots, two chats, and one ``rabbit``
+    memory per chat. Two-chat seeding lets the cross-chat assertion
+    actually distinguish "both chats appear" from "only the current
+    one does"."""
+    with open_db(db_path) as conn:
+        append_event(
+            conn,
+            kind="you_authored",
+            payload={"name": "Me", "pronouns": "", "persona": ""},
+        )
+        for bot_id, chat_id in (("bot_a", "chat_a"), ("bot_b", "chat_b")):
+            append_event(
+                conn,
+                kind="bot_authored",
+                payload={
+                    "id": bot_id,
+                    "name": bot_id.upper(),
+                    "persona": "thoughtful",
+                    "voice_samples": [],
+                    "traits": [],
+                    "backstory": "",
+                    "initial_relationship_to_you": "friend",
+                    "kickoff_prose": "kickoff",
+                },
+            )
+            append_event(
+                conn,
+                kind="chat_created",
+                payload={
+                    "id": chat_id,
+                    "host_bot_id": bot_id,
+                    "initial_time": "2026-04-26T20:00:00+00:00",
+                    "narrative_anchor": "Day 1",
+                    "weather": "",
+                },
+            )
+            append_event(
+                conn,
+                kind="memory_written",
+                payload={
+                    "owner_id": bot_id,
+                    "chat_id": chat_id,
+                    "pov_summary": f"the rabbit darted across {chat_id}",
+                    "witness_you": 1,
+                    "witness_host": 1,
+                    "witness_guest": 0,
+                    "source": "direct",
+                    "reliability": 1.0,
+                    "significance": 1,
+                    "pinned": 0,
+                    "auto_pinned": 0,
+                },
+            )
+        project(conn)
+
+
+def test_search_returns_results_from_multiple_chats(client, tmp_path):
+    """A single ``/search?q=rabbit`` must surface matches from BOTH
+    chats — the whole point of the cross-chat search box is that it
+    isn't owner-scoped."""
+    _seed_two_chats_with_memories(tmp_path / "test.db")
+    resp = client.get("/search?q=rabbit")
+    assert resp.status_code == 200
+    body = resp.text
+    # Both chats' memory snippets must appear in the rendered page.
+    assert "chat_a" in body
+    assert "chat_b" in body
+    assert "rabbit" in body.lower()
+
+
+def test_empty_query_renders_placeholder_not_results(client, tmp_path):
+    """``/search`` with no query renders the page in its empty state.
+
+    The placeholder copy is a contract with the user — they should see
+    "enter a query" rather than an empty result list that looks like a
+    no-match. Also: the FTS short-circuit means there are no result
+    rows to leak into the body."""
+    _seed_two_chats_with_memories(tmp_path / "test.db")
+    resp = client.get("/search")
+    assert resp.status_code == 200
+    body = resp.text.lower()
+    assert "enter a query" in body
+    # Seeded "rabbit" memories must NOT appear: empty query => no results.
+    assert "the rabbit darted" not in resp.text
+
+
+def test_result_links_navigate_to_chat(client, tmp_path):
+    """Each result links back to its originating chat so the user can
+    reopen the thread where the memory was first witnessed."""
+    _seed_two_chats_with_memories(tmp_path / "test.db")
+    resp = client.get("/search?q=rabbit")
+    assert resp.status_code == 200
+    # The link target is chat-level (memories don't carry an event_id
+    # column today, so we don't deep-link to a specific turn).
+    assert 'href="/chats/chat_a"' in resp.text

tests/test_snapshot_ux.py

@@ -0,0 +1,182 @@
+"""Tests for Task 99 — snapshot UX (manual trigger + list + restore + preview).
+
+Phase 4 surfaces the existing snapshot infrastructure (Phase 1 T20 / T31)
+through HTML routes so the user can:
+
+* see what snapshots exist,
+* take one on demand,
+* restore one with a hard confirm,
+* peek at metadata before restoring.
+
+The underlying service API lives in ``chat/services/snapshot.py`` and is
+already exercised by ``test_snapshot.py``; here we only verify the web
+surface wires the existing functions correctly.
+"""
+
+from __future__ import annotations
+
+import json
+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
+
+
+def _bot_payload(bot_id: str, name: str) -> dict:
+    return {
+        "id": bot_id,
+        "name": name,
+        "persona": "fancy",
+        "voice_samples": ["sample"],
+        "traits": ["shy"],
+        "backstory": "",
+        "initial_relationship_to_you": "coworker",
+        "kickoff_prose": "",
+    }
+
+
+@pytest.fixture
+def client(tmp_path, monkeypatch):
+    """A TestClient whose db + data_dir live under ``tmp_path``.
+
+    ``load_settings`` derives ``data_dir`` from ``CHAT_DB_PATH``'s parent
+    when ``CHAT_DATA_DIR`` is unset (see ``chat/config.py``), so this also
+    isolates the ``data/snapshots/`` tree to ``tmp_path``.
+    """
+    config_path = tmp_path / "config.toml"
+    config_path.write_text('featherless_api_key = "test"\n')
+    monkeypatch.setenv("CHAT_CONFIG_PATH", str(config_path))
+    monkeypatch.setenv("CHAT_DB_PATH", str(tmp_path / "test.db"))
+
+    with TestClient(app) as c:
+        c.tmp_path = tmp_path  # type: ignore[attr-defined]
+        yield c
+
+
+def _seed_bot(db_path: Path, bot_id: str = "bot_a", name: str = "BotA") -> None:
+    with open_db(db_path) as conn:
+        append_event(conn, kind="bot_authored", payload=_bot_payload(bot_id, name))
+        project(conn)
+
+
+def _take_snapshot_via_service(
+    db_path: Path, data_dir: Path, kind: str = "periodic"
+) -> Path:
+    from chat.services.snapshot import take_snapshot
+
+    with open_db(db_path) as conn:
+        return take_snapshot(conn, data_dir=data_dir, kind=kind)
+
+
+def test_list_snapshots_renders_page(client, tmp_path):
+    _seed_bot(tmp_path / "test.db", "bot_a", "BotA")
+    # Take two snapshots through the service so the listing has rows.
+    p1 = _take_snapshot_via_service(tmp_path / "test.db", tmp_path, kind="periodic")
+    p2 = _take_snapshot_via_service(tmp_path / "test.db", tmp_path, kind="rewind")
+
+    response = client.get("/snapshots")
+    assert response.status_code == 200
+    body = response.text
+    # Both filenames should appear in the listing.
+    assert p1.stem in body
+    assert p2.stem in body
+    # Both kinds should be visible.
+    assert "periodic" in body
+    assert "rewind" in body
+
+
+def test_take_snapshot_creates_new(client, tmp_path):
+    _seed_bot(tmp_path / "test.db", "bot_a", "BotA")
+    snapshot_dir = tmp_path / "snapshots" / "periodic"
+
+    before = (
+        len(list(snapshot_dir.glob("*.json"))) if snapshot_dir.exists() else 0
+    )
+    response = client.post("/snapshots/take", follow_redirects=False)
+    assert response.status_code == 303
+    assert response.headers["location"] == "/snapshots"
+
+    after = len(list(snapshot_dir.glob("*.json")))
+    assert after == before + 1
+
+
+def test_restore_snapshot_with_correct_confirm(client, tmp_path):
+    db_path = tmp_path / "test.db"
+    _seed_bot(db_path, "bot_a", "BotA")
+    snapshot_path = _take_snapshot_via_service(
+        db_path, tmp_path, kind="periodic"
+    )
+    snapshot_id = snapshot_path.stem  # filename without extension
+
+    # Mutate the DB after the snapshot was taken — restoring should erase
+    # the new bot.
+    with open_db(db_path) as conn:
+        append_event(
+            conn, kind="bot_authored", payload=_bot_payload("bot_b", "BotB")
+        )
+        project(conn)
+        bots_before = conn.execute(
+            "SELECT id FROM bots ORDER BY id"
+        ).fetchall()
+    assert {r[0] for r in bots_before} == {"bot_a", "bot_b"}
+
+    response = client.post(
+        f"/snapshots/restore/{snapshot_id}",
+        data={"confirm_id": snapshot_id, "kind": "periodic"},
+        follow_redirects=False,
+    )
+    assert response.status_code == 303
+
+    with open_db(db_path) as conn:
+        bots_after = conn.execute(
+            "SELECT id FROM bots ORDER BY id"
+        ).fetchall()
+    # The post-snapshot bot should be gone.
+    assert {r[0] for r in bots_after} == {"bot_a"}
+
+
+def test_restore_snapshot_wrong_confirm_400(client, tmp_path):
+    db_path = tmp_path / "test.db"
+    _seed_bot(db_path, "bot_a", "BotA")
+    snapshot_path = _take_snapshot_via_service(
+        db_path, tmp_path, kind="periodic"
+    )
+    snapshot_id = snapshot_path.stem
+
+    response = client.post(
+        f"/snapshots/restore/{snapshot_id}",
+        data={"confirm_id": "not_the_right_id", "kind": "periodic"},
+        follow_redirects=False,
+    )
+    assert response.status_code == 400
+
+
+def test_preview_renders_metadata(client, tmp_path):
+    db_path = tmp_path / "test.db"
+    _seed_bot(db_path, "bot_a", "BotA")
+    snapshot_path = _take_snapshot_via_service(
+        db_path, tmp_path, kind="periodic"
+    )
+    snapshot_id = snapshot_path.stem
+
+    # Append more events post-snapshot so the delta is non-zero.
+    with open_db(db_path) as conn:
+        append_event(
+            conn, kind="bot_authored", payload=_bot_payload("bot_b", "BotB")
+        )
+        project(conn)
+
+    response = client.get(
+        f"/snapshots/{snapshot_id}/preview", params={"kind": "periodic"}
+    )
+    assert response.status_code == 200
+    body = response.text
+    assert snapshot_id in body
+    # Snapshot's last_event_id and current event_log size should appear.
+    dump = json.loads(snapshot_path.read_text())
+    assert str(dump["last_event_id"]) in body