merge: T96 combined FTS + vector retrieval ranking via RRF

feat: combined FTS + vector retrieval ranking via RRF (T96)
merge: T95 delete-impact computation service
2026-04-27 02:44:03 -04:00 · 2026-04-27 02:42:38 -04:00 · 2026-04-27 02:37:28 -04:00 · 2026-04-27 02:37:28 -04:00 · 2026-04-27 02:36:30 -04:00 · 2026-04-27 02:35:58 -04:00
27 changed files with 3379 additions and 72 deletions
@@ -0,0 +1,14 @@
 -- Embeddings stored as JSON arrays (pure-Python cosine at query time).
 -- Phase 4.5+ may swap to sqlite-vec when the host Python supports
 -- loadable extensions; the schema is intentionally simple to make that
 -- migration straightforward.
 CREATE TABLE embeddings (
    memory_id INTEGER PRIMARY KEY,
    vector_json TEXT NOT NULL,   -- JSON array of floats, length = dim
    model TEXT NOT NULL,
    dim INTEGER NOT NULL,
    indexed_at TEXT NOT NULL DEFAULT (datetime('now')),
    FOREIGN KEY (memory_id) REFERENCES memories(id)
 );
 CREATE INDEX embeddings_model_idx ON embeddings(model);
@@ -0,0 +1,17 @@
 CREATE TABLE branches (
    id INTEGER PRIMARY KEY,
    name TEXT NOT NULL UNIQUE,
    origin_event_id INTEGER NOT NULL,
    head_event_id INTEGER NOT NULL,
    chat_id TEXT,
    created_at TEXT NOT NULL DEFAULT (datetime('now')),
    is_active INTEGER NOT NULL DEFAULT 0
 );
 -- Exactly one row may have is_active = 1 at any time.
 CREATE UNIQUE INDEX branches_active_idx ON branches(is_active) WHERE is_active = 1;
 -- Bootstrap the main branch. origin_event_id=0 + head_event_id=0 are
 -- placeholder seeds; the orchestrator updates head as new events land.
 INSERT INTO branches (name, origin_event_id, head_event_id, is_active)
 VALUES ('main', 0, 0, 1);
@@ -0,0 +1,107 @@
 """Branching service (T94, Phase 4).
 Wraps branches state with validation + event emission. Phase 4 ships
 the data model and creation/switching APIs; the read-side filter
 (event readers consulting is_active) is a Phase 4.5+ follow-up — for
 now branches are metadata-only and the existing event readers remain
 branch-agnostic. The drawer UI (T98) drives create/switch via these
 helpers.
 """
 from __future__ import annotations
 from sqlite3 import Connection
 from chat.eventlog.log import append_and_apply
 from chat.state.branches import get_branch, list_branches, active_branch  # noqa: F401
 def branch_from_event(
    conn: Connection,
    *,
    name: str,
    origin_event_id: int,
    chat_id: str | None = None,
 ) -> int:
    """Create a new named branch forking from origin_event_id.
    Emits a branch_created event. Returns the new branch's row id.
    Raises ValueError if name already exists or origin_event_id doesn't
    correspond to a real event."""
    if not name or not name.strip():
        raise ValueError("branch name must be non-empty")
    if get_branch(conn, name) is not None:
        raise ValueError(f"branch {name!r} already exists")
    # Validate origin_event_id is a real event id (or 0 for the bootstrap case
    # which only main uses).
    if origin_event_id < 0:
        raise ValueError(f"origin_event_id must be >= 0, got {origin_event_id}")
    if origin_event_id > 0:
        row = conn.execute(
            "SELECT 1 FROM event_log WHERE id = ?", (origin_event_id,)
        ).fetchone()
        if row is None:
            raise ValueError(
                f"origin_event_id {origin_event_id} does not exist in event_log"
            )
    append_and_apply(
        conn,
        kind="branch_created",
        payload={
            "name": name,
            "origin_event_id": origin_event_id,
            "head_event_id": origin_event_id,  # head starts at origin
            "chat_id": chat_id,
        },
    )
    branch = get_branch(conn, name)
    if branch is None:
        # Should be unreachable if append_and_apply worked.
        raise RuntimeError(f"branch {name!r} not found after creation")
    return branch["id"]
 def switch_active_branch(conn: Connection, *, name: str) -> None:
    """Make the named branch active. Emits branch_switched."""
    if get_branch(conn, name) is None:
        raise ValueError(f"branch {name!r} does not exist")
    append_and_apply(
        conn,
        kind="branch_switched",
        payload={"name": name},
    )
 def list_branches_with_metadata(
    conn: Connection, chat_id: str | None = None
 ) -> list[dict]:
    """List branches with computed event_count metadata.
    event_count = head_event_id - origin_event_id + 1 (when both are set)
                  OR head_event_id (when origin is 0, e.g., main branch)
                  OR 0 (when head <= origin, which is the bootstrap state)
    """
    branches = list_branches(conn, chat_id)
    enriched = []
    for b in branches:
        origin = b["origin_event_id"]
        head = b["head_event_id"]
        if head < origin:
            event_count = 0
        elif origin == 0:
            event_count = head
        else:
            event_count = head - origin + 1
        enriched.append({**b, "event_count": event_count})
    return enriched
 __all__ = [
    "branch_from_event",
    "switch_active_branch",
    "list_branches_with_metadata",
 ]
@@ -0,0 +1,75 @@
 """Cross-chat search service (T93, Phase 4).
 FTS5-based search across ALL owners and ALL chats. Used by the
 top-bar search UX (T100) for "where did I last see this character
 mention X?" queries. NO witness filter -- this is intentionally a
 power-user surface that surfaces memories across POVs.
 Mirrors the FTS5 access pattern of ``chat.state.memory.search_memories``
 but drops both the ``owner_id = ?`` and the per-witness predicates so a
 single query can sweep every chat in the database. The composite
 re-rank is also dropped: callers want raw BM25 ordering for the
 "highest match strength wins" semantics expected of a global search box.
 """
 from __future__ import annotations
 from sqlite3 import Connection
 def search_all_memories(
    conn: Connection,
    *,
    query: str,
    k: int = 20,
 ) -> list[dict]:
    """Search FTS5 across all owners and chats.
    Returns rows with ``{memory_id, owner_id, chat_id, scene_id,
    pov_summary, significance, ts, fts_rank}``, sorted by FTS5 BM25
    rank ascending (lower rank = stronger match, surfaced first).
    The ``memories`` table has no ``ts`` column; we expose ``created_at``
    (the projector-side row insertion timestamp) under that key so the
    UI does not have to know the storage name.
    An empty / whitespace-only ``query`` short-circuits to ``[]`` to
    avoid an FTS5 ``MATCH ''`` syntax error and to keep the top-bar
    "no input yet" state from triggering a full-table scan.
    """
    if not query or not query.strip():
        return []
    # FTS5 MATCH against the same ``memories_fts`` virtual table that
    # backs ``state.memory.search_memories``; the JOIN pulls metadata
    # from the content table because the FTS index only stores
    # ``pov_summary``. ORDER BY rank ASC because BM25 in FTS5 returns
    # negative scores where lower is better.
    rows = conn.execute(
        "SELECT m.id, m.owner_id, m.chat_id, m.scene_id, "
        "       m.pov_summary, m.significance, m.created_at, "
        "       memories_fts.rank "
        "FROM memories_fts "
        "JOIN memories m ON m.id = memories_fts.rowid "
        "WHERE memories_fts MATCH ? "
        "ORDER BY memories_fts.rank ASC "
        "LIMIT ?",
        (query.strip(), k),
    ).fetchall()
    return [
        {
            "memory_id": r[0],
            "owner_id": r[1],
            "chat_id": r[2],
            "scene_id": r[3],
            "pov_summary": r[4],
            "significance": r[5],
            "ts": r[6],
            "fts_rank": r[7],
        }
        for r in rows
    ]
 __all__ = ["search_all_memories"]
@@ -0,0 +1,147 @@
 """Delete-impact computation service (T95, Phase 4).
 Walks event_log forward from a target event_id and produces an ImpactReport
 describing what would be removed if rewind-to-target were invoked. Pure
 computation — does NOT mutate the database. Used by T98's drawer surgical-
 delete UI to render an 'are you sure?' modal before invoking the actual
 rewind path (chat/services/rewind.py).
 """
 from __future__ import annotations
 import json
 from sqlite3 import Connection
 from pydantic import BaseModel, Field
 class DeletedItem(BaseModel):
    kind: str
    description: str
    target_id: int | str | None = None
 class ImpactReport(BaseModel):
    target_event_id: int
    cascading: list[DeletedItem] = Field(default_factory=list)
    notes: list[str] = Field(default_factory=list)
 def _excerpt(text: str, n: int = 60) -> str:
    text = (text or "").strip().replace("\n", " ")
    return text if len(text) <= n else text[: n - 1] + "…"
 def compute_delete_impact(
    conn: Connection,
    *,
    target_event_id: int,
 ) -> ImpactReport:
    """Compute the cascading impact of rewinding to target_event_id."""
    # Verify target exists.
    target_row = conn.execute(
        "SELECT id, kind, payload_json FROM event_log WHERE id = ?",
        (target_event_id,),
    ).fetchone()
    if target_row is None:
        return ImpactReport(
            target_event_id=target_event_id,
            cascading=[],
            notes=[f"target event_id {target_event_id} not found"],
        )
    # Walk forward: every event with id >= target_event_id is in scope.
    rows = conn.execute(
        "SELECT id, kind, payload_json FROM event_log "
        "WHERE id >= ? ORDER BY id ASC",
        (target_event_id,),
    ).fetchall()
    cascading: list[DeletedItem] = []
    notes: list[str] = []
    scene_close_present = False
    regenerated_from = None
    for row_id, kind, payload_json in rows:
        try:
            payload = json.loads(payload_json) if payload_json else {}
        except (json.JSONDecodeError, TypeError):
            payload = {}
        if kind == "memory_written":
            cascading.append(
                DeletedItem(
                    kind=kind,
                    description=f"memory: {_excerpt(payload.get('pov_summary', ''))}",
                    target_id=payload.get("memory_id"),
                )
            )
        elif kind == "edge_update":
            src = payload.get("source_id", "?")
            tgt = payload.get("target_id", "?")
            cascading.append(
                DeletedItem(
                    kind=kind,
                    description=f"edge update: {src} -> {tgt}",
                    target_id=f"{src}->{tgt}",
                )
            )
        elif kind == "scene_closed":
            scene_close_present = True
            cascading.append(
                DeletedItem(
                    kind=kind,
                    description=f"scene close at {payload.get('closed_at', '?')}",
                    target_id=payload.get("scene_id"),
                )
            )
        elif kind in ("user_turn", "user_turn_edit", "assistant_turn"):
            speaker = payload.get("speaker_id") or ("you" if kind.startswith("user") else "?")
            prose = payload.get("prose") or payload.get("text") or ""
            cascading.append(
                DeletedItem(
                    kind=kind,
                    description=f"turn {row_id} ({speaker}: {_excerpt(prose, 50)})",
                    target_id=row_id,
                )
            )
            if regenerated_from is None and payload.get("regenerated_from"):
                regenerated_from = payload["regenerated_from"]
        elif kind == "manual_edit":
            target_kind = payload.get("target_kind", "?")
            cascading.append(
                DeletedItem(
                    kind=kind,
                    description=f"manual edit: {target_kind}",
                    target_id=payload.get("target_id"),
                )
            )
        else:
            cascading.append(
                DeletedItem(
                    kind=kind,
                    description=f"{kind} event",
                    target_id=row_id,
                )
            )
    # Notes / warnings.
    notes.append(f"{len(rows)} events would be discarded total")
    if scene_close_present:
        notes.append(
            "scene close events are in scope — closing-scene per-POV summaries "
            "and group_node updates will be reverted"
        )
    if regenerated_from is not None:
        notes.append(
            f"target turn was regenerated from event_id {regenerated_from}; "
            f"the original turn remains intact"
        )
    return ImpactReport(
        target_event_id=target_event_id,
        cascading=cascading,
        notes=notes,
    )
 __all__ = ["DeletedItem", "ImpactReport", "compute_delete_impact"]
@@ -0,0 +1,108 @@
 """Embedding generation service (T91, Phase 4).
 Wraps the embedding API call. For Phase 4's first cut we ship a
 deterministic local pseudo-embedding (hash-derived) so the vector
 retrieval pipeline can land without an external embedding endpoint
 or heavy local dependency. Phase 4.5+ swaps to a real model — the
 EmbeddingResult shape stays the same, only the generator changes.
 """
 from __future__ import annotations
 import hashlib
 import math
 import struct
 from pydantic import BaseModel
 from chat.llm.client import LLMClient
 DEFAULT_EMBEDDING_DIM = 384
 DEFAULT_EMBEDDING_MODEL = "pseudo-sha256-384"
 FALLBACK_EMBEDDING_MODEL = "fallback"
 class EmbeddingResult(BaseModel):
    vector: list[float]
    model: str
    dim: int
 def _pseudo_embed(text: str, dim: int = DEFAULT_EMBEDDING_DIM) -> list[float]:
    """Deterministic pseudo-embedding for Phase 4 first cut.
    Hashes the text with SHA-256, then expands by re-hashing each
    successive block with the previous block + a counter — this gives
    ``dim * 4`` bytes of fresh entropy per input rather than naively
    repeating the 32-byte digest (which would collapse the vector onto
    only 8 unique floats and make distinct inputs cosine-similar).
    Bytes are unpacked as little-endian int32s and rescaled to [-1, 1]
    so we sidestep the float32 NaN/denormal values that ``struct.unpack
    'f'`` would otherwise produce on raw hash bytes. The result is
    unit-normalized so cosine similarity reduces to a dot product.
    NOT semantically meaningful — just consistent for testing the
    pipeline. Phase 4.5 should swap to a real embedding model.
    """
    needed = dim * 4  # 4 bytes per int32
    seed = text.encode("utf-8")
    chunks: list[bytes] = []
    counter = 0
    while sum(len(c) for c in chunks) < needed:
        block = hashlib.sha256(seed + counter.to_bytes(4, "big")).digest()
        chunks.append(block)
        counter += 1
    full = b"".join(chunks)[:needed]
    ints = struct.unpack(f"<{dim}i", full)
    # Map int32 to roughly [-1, 1] — exact bound doesn't matter since we
    # normalize, but keeps values numerically tame.
    raw = [x / 2147483648.0 for x in ints]
    norm = math.sqrt(sum(x * x for x in raw)) or 1.0
    return [x / norm for x in raw]
 async def generate_embedding(
    client: LLMClient,
    *,
    text: str,
    model: str = DEFAULT_EMBEDDING_MODEL,
    dim: int = DEFAULT_EMBEDDING_DIM,
    timeout_s: float = 30.0,
 ) -> EmbeddingResult:
    """Generate an embedding for the given text.
    Phase 4 default uses a deterministic local pseudo-embedding. If
    the LLMClient grows an ``embed(...)`` method in Phase 4.5, this
    wrapper will route to it when ``model != "pseudo-sha256-384"``.
    Falls back to a zero vector with ``model="fallback"`` on any
    failure (callers detect the sentinel and skip indexing). For the
    pseudo path, failure is structurally impossible — it's pure local
    computation.
    """
    if not text or not text.strip():
        # Empty input — return fallback so caller doesn't index empty rows.
        return EmbeddingResult(
            vector=[0.0] * dim, model=FALLBACK_EMBEDDING_MODEL, dim=dim
        )
    if model == DEFAULT_EMBEDDING_MODEL:
        # Pure-local pseudo path — no LLMClient call.
        return EmbeddingResult(vector=_pseudo_embed(text, dim), model=model, dim=dim)
    # Future: real embedding via client.embed(...). Phase 4.5 work.
    # For Phase 4, any non-default model falls through to fallback.
    return EmbeddingResult(
        vector=[0.0] * dim, model=FALLBACK_EMBEDDING_MODEL, dim=dim
    )
 __all__ = [
    "DEFAULT_EMBEDDING_DIM",
    "DEFAULT_EMBEDDING_MODEL",
    "FALLBACK_EMBEDDING_MODEL",
    "EmbeddingResult",
    "generate_embedding",
 ]
@@ -22,62 +22,6 @@ from sqlite3 import Connection
 from chat.eventlog.log import append_and_apply
 def record_turn_memory(
    conn: Connection,
    *,
    chat_id: str,
    host_bot_id: str,
    narrative_text: str,
    scene_id: int | None = None,
    chat_clock_at: str | None = None,
    source: str = "direct",
    significance: int = 1,
 ) -> tuple[int, int | None]:
    """Append a ``memory_written`` event for the host bot's POV of this turn.
    Uses :func:`chat.eventlog.log.append_and_apply` (not raw
    :func:`append_event`) so the new memory row is projected immediately
    without re-running prior non-idempotent handlers (e.g. ``edge_update``
    deltas).
    Returns ``(event_id, memory_id)``. ``event_id`` is the row id of the
    just-appended ``memory_written`` event in ``event_log``. ``memory_id``
    is the autoincrement PK of the corresponding ``memories`` row — these
    are *different* numbers (event_log and memories use independent
    rowid sequences) so callers needing to update significance or pin
    state must use ``memory_id``. Falls back to ``None`` if the projected
    row can't be located, which shouldn't happen but keeps the return
    shape stable.
    """
    payload: dict = {
        "owner_id": host_bot_id,
        "chat_id": chat_id,
        "pov_summary": narrative_text,
        "witness_you": 1,
        "witness_host": 1,
        "witness_guest": 0,
        "source": source,
        "reliability": 1.0,
        "significance": significance,
        "pinned": 0,
        "auto_pinned": 0,
    }
    if scene_id is not None:
        payload["scene_id"] = scene_id
    if chat_clock_at is not None:
        payload["chat_clock_at"] = chat_clock_at
    event_id = append_and_apply(conn, kind="memory_written", payload=payload)
    row = conn.execute(
        "SELECT id FROM memories "
        "WHERE owner_id = ? AND chat_id = ? "
        "ORDER BY id DESC LIMIT 1",
        (host_bot_id, chat_id),
    ).fetchone()
    memory_id = row[0] if row else None
    return event_id, memory_id
 def _write_one_memory(
    conn: Connection,
    *,
@@ -182,9 +182,13 @@ async def regenerate_assistant_turn(
        (chat_id, original_assistant_event_id),
    ).fetchall()
    if unrolled_lifecycle:
        # T90.2: phrased as "at-or-after turn <id>" rather than "from
        # superseded turn" because regenerating an OLDER turn lists
        # intervening-turn transitions that legitimately stand on their
        # own — those weren't authored by the superseded turn itself.
        _log.warning(
-            "regenerate_assistant_turn: %d lifecycle transition(s) from "
+            "regenerate_assistant_turn: %d lifecycle transition(s) "
-            "superseded turn %s are NOT being rolled back (Phase 4 "
+            "at-or-after turn %s are NOT being rolled back (Phase 4 "
            "follow-up). Affected event ids: %s",
            len(unrolled_lifecycle),
            original_assistant_event_id,
@@ -54,14 +54,21 @@ def read_recent_dialogue(
    regenerate to drop the original assistant_turn from its prompt
    context window before that row has been marked superseded (the
    supersede UPDATE lands at the end so the new event_id is known).
    T90.1: the chat_id filter is pushed into SQL via ``json_extract`` so
    ``LIMIT N`` always returns N rows scoped to the requested chat. The
    previous implementation filtered chat_id post-fetch in Python, which
    let foreign-chat rows fill the LIMIT and yield fewer than N relevant
    rows in busy multi-chat databases.
    """
    if exclude_event_id is None:
        cur = conn.execute(
            "SELECT id, kind, payload_json FROM event_log "
            "WHERE kind IN ('user_turn', 'user_turn_edit', 'assistant_turn') "
            "  AND superseded_by IS NULL AND hidden = 0 "
            "  AND json_extract(payload_json, '$.chat_id') = ? "
            "ORDER BY id DESC LIMIT ?",
-            (limit,),
+            (chat_id, limit),
        )
    else:
        cur = conn.execute(
@@ -69,15 +76,14 @@ def read_recent_dialogue(
            "WHERE kind IN ('user_turn', 'user_turn_edit', 'assistant_turn') "
            "  AND id != ? "
            "  AND superseded_by IS NULL AND hidden = 0 "
            "  AND json_extract(payload_json, '$.chat_id') = ? "
            "ORDER BY id DESC LIMIT ?",
-            (exclude_event_id, limit),
+            (exclude_event_id, chat_id, limit),
        )
    rows = list(reversed(cur.fetchall()))
    out: list[dict] = []
    for row_id, kind, payload_json in rows:
        p = json.loads(payload_json)
        if p.get("chat_id") != chat_id:
            continue
        if kind in ("user_turn", "user_turn_edit"):
            out.append(
                {
@@ -0,0 +1,79 @@
 """Vector search service (T92, Phase 4).
 Pure-Python cosine similarity over the embeddings table. Phase 4 ships
 this without sqlite-vec because the host Python build doesn't support
 loadable extensions. For single-user scale (< few thousand memories
 per owner), iterating in Python is sub-millisecond.
 Phase 4.5+ may swap to sqlite-vec when the host Python supports
 enable_load_extension; the public API stays stable.
 """
 from __future__ import annotations
 import math
 from sqlite3 import Connection
 from chat.state.embeddings import list_embeddings_for_owner
 _VALID_WITNESS_ROLES = {"you", "host", "guest"}
 def _cosine_similarity(a: list[float], b: list[float]) -> float:
    """Cosine similarity. Assumes both vectors are non-zero."""
    if len(a) != len(b):
        return 0.0
    dot = sum(x * y for x, y in zip(a, b))
    norm_a = math.sqrt(sum(x * x for x in a)) or 1.0
    norm_b = math.sqrt(sum(x * x for x in b)) or 1.0
    return dot / (norm_a * norm_b)
 def vector_search(
    conn: Connection,
    *,
    owner_id: str,
    witness_role: str,    # "you" | "host" | "guest"
    query_vector: list[float],
    k: int = 4,
 ) -> list[dict]:
    """Return top-K memories by cosine similarity to query_vector,
    witness-filtered for the viewer's POV. Returns rows with
    {memory_id, pov_summary, significance, score} sorted by score
    DESC. Empty list if no embeddings indexed for this owner.
    """
    if witness_role not in _VALID_WITNESS_ROLES:
        raise ValueError(
            f"witness_role must be one of {_VALID_WITNESS_ROLES}, got {witness_role!r}"
        )
    rows = list_embeddings_for_owner(conn, owner_id)
    if not rows:
        return []
    # Witness-filter by the requesting role.
    witness_key = f"witness_{witness_role}"
    filtered = [r for r in rows if r.get(witness_key) == 1]
    if not filtered:
        return []
    scored: list[tuple[float, dict]] = []
    for row in filtered:
        score = _cosine_similarity(query_vector, row["vector"])
        scored.append(
            (
                score,
                {
                    "memory_id": row["memory_id"],
                    "pov_summary": row["pov_summary"],
                    "significance": row["significance"],
                    "score": score,
                },
            )
        )
    scored.sort(key=lambda t: t[0], reverse=True)
    return [item for _, item in scored[:k]]
 __all__ = ["vector_search"]
@@ -0,0 +1,133 @@
 """Branches projector + readers (T89, Phase 4).
 A branch is a named fork of the event log. The 'main' branch is bootstrapped
 by migration 0013 with is_active=1. Subsequent branches reference an
 origin_event_id (the event they forked from). Phase 4 enables creation
 and switching; the read-side filter (event readers consulting is_active)
 is a Phase 4.5 follow-up — for now branches are metadata-only and the
 existing event readers remain branch-agnostic.
 """
 from __future__ import annotations
 from sqlite3 import Connection
 from chat.eventlog.projector import on
 from chat.eventlog.log import Event
@on("branch_created")
 def _apply_branch_created(conn: Connection, e: Event) -> None:
    """Insert a new branch row with is_active=0. Idempotent via INSERT OR IGNORE."""
    p = e.payload
    conn.execute(
        "INSERT OR IGNORE INTO branches "
        "(name, origin_event_id, head_event_id, chat_id, is_active) "
        "VALUES (?, ?, ?, ?, 0)",
        (
            p["name"],
            int(p["origin_event_id"]),
            int(p.get("head_event_id", p["origin_event_id"])),
            p.get("chat_id"),
        ),
    )
@on("branch_switched")
 def _apply_branch_switched(conn: Connection, e: Event) -> None:
    """Set is_active=1 on the named branch and is_active=0 on all others.
    Atomic via two UPDATEs ordered to avoid the unique-active-index race.
    """
    p = e.payload
    name = p["name"]
    # Clear ALL is_active flags first (avoids the unique-index trip).
    conn.execute("UPDATE branches SET is_active = 0 WHERE is_active = 1")
    conn.execute(
        "UPDATE branches SET is_active = 1 WHERE name = ?",
        (name,),
    )
@on("branch_head_updated")
 def _apply_branch_head_updated(conn: Connection, e: Event) -> None:
    """Update head_event_id on the named branch."""
    p = e.payload
    conn.execute(
        "UPDATE branches SET head_event_id = ? WHERE name = ?",
        (int(p["head_event_id"]), p["name"]),
    )
 def get_branch(conn: Connection, name: str) -> dict | None:
    row = conn.execute(
        "SELECT id, name, origin_event_id, head_event_id, chat_id, "
        "       created_at, is_active "
        "FROM branches WHERE name = ?",
        (name,),
    ).fetchone()
    if not row:
        return None
    return {
        "id": row[0],
        "name": row[1],
        "origin_event_id": row[2],
        "head_event_id": row[3],
        "chat_id": row[4],
        "created_at": row[5],
        "is_active": bool(row[6]),
    }
 def list_branches(conn: Connection, chat_id: str | None = None) -> list[dict]:
    if chat_id is None:
        rows = conn.execute(
            "SELECT id, name, origin_event_id, head_event_id, chat_id, "
            "       created_at, is_active "
            "FROM branches ORDER BY id ASC"
        ).fetchall()
    else:
        rows = conn.execute(
            "SELECT id, name, origin_event_id, head_event_id, chat_id, "
            "       created_at, is_active "
            "FROM branches WHERE chat_id = ? OR chat_id IS NULL "
            "ORDER BY id ASC",
            (chat_id,),
        ).fetchall()
    return [
        {
            "id": r[0],
            "name": r[1],
            "origin_event_id": r[2],
            "head_event_id": r[3],
            "chat_id": r[4],
            "created_at": r[5],
            "is_active": bool(r[6]),
        }
        for r in rows
    ]
 def active_branch(conn: Connection) -> dict | None:
    row = conn.execute(
        "SELECT id, name, origin_event_id, head_event_id, chat_id, "
        "       created_at, is_active "
        "FROM branches WHERE is_active = 1"
    ).fetchone()
    if not row:
        return None
    return {
        "id": row[0],
        "name": row[1],
        "origin_event_id": row[2],
        "head_event_id": row[3],
        "chat_id": row[4],
        "created_at": row[5],
        "is_active": bool(row[6]),
    }
 __all__ = [
    "get_branch",
    "list_branches",
    "active_branch",
 ]
@@ -0,0 +1,105 @@
 """Embeddings projector + readers (T88, Phase 4).
 Embeddings are stored as JSON-serialized float arrays in a regular
 SQLite table. Cosine similarity is computed in Python at query time
 (see chat/services/vector_search.py / T92). This deliberately avoids
 the sqlite-vec extension dependency — the host Python build doesn't
 support enable_load_extension. Phase 4.5+ may revisit if memory counts
 grow beyond pure-Python feasibility (~few thousand per query).
 """
 from __future__ import annotations
 import json
 from sqlite3 import Connection
 from chat.eventlog.projector import on
 from chat.eventlog.log import Event
@on("embedding_indexed")
 def _apply_embedding_indexed(conn: Connection, e: Event) -> None:
    """Insert or replace the embedding for a memory.
    Idempotent: re-projection or re-indexing replaces the prior vector.
    """
    p = e.payload
    vector = p["vector"]
    conn.execute(
        "INSERT OR REPLACE INTO embeddings "
        "(memory_id, vector_json, model, dim, indexed_at) "
        "VALUES (?, ?, ?, ?, datetime('now'))",
        (
            int(p["memory_id"]),
            json.dumps(list(vector)),
            p["model"],
            int(p.get("dim") or len(vector)),
        ),
    )
@on("embedding_deindexed")
 def _apply_embedding_deindexed(conn: Connection, e: Event) -> None:
    """Remove the embedding for a memory (used by reset cascade)."""
    p = e.payload
    conn.execute(
        "DELETE FROM embeddings WHERE memory_id = ?",
        (int(p["memory_id"]),),
    )
 def get_embedding(conn: Connection, memory_id: int) -> dict | None:
    row = conn.execute(
        "SELECT memory_id, vector_json, model, dim, indexed_at "
        "FROM embeddings WHERE memory_id = ?",
        (memory_id,),
    ).fetchone()
    if not row:
        return None
    return {
        "memory_id": row[0],
        "vector": json.loads(row[1]),
        "model": row[2],
        "dim": row[3],
        "indexed_at": row[4],
    }
 def list_embeddings_for_owner(conn: Connection, owner_id: str) -> list[dict]:
    """Return all embeddings for memories owned by ``owner_id``.
    Used by vector search at query time (T92). The join carries the
    fields the cosine ranker needs to assemble result rows without a
    second round-trip: the POV summary text, significance, and witness
    flags. The ``memories`` table has no separate ``text`` column —
    ``pov_summary`` is the canonical narrative text per
    ``chat/services/memory_write.py``.
    """
    rows = conn.execute(
        "SELECT e.memory_id, e.vector_json, e.model, e.dim, "
        "       m.pov_summary, m.significance, "
        "       m.witness_you, m.witness_host, m.witness_guest "
        "FROM embeddings e "
        "JOIN memories m ON m.id = e.memory_id "
        "WHERE m.owner_id = ?",
        (owner_id,),
    ).fetchall()
    return [
        {
            "memory_id": r[0],
            "vector": json.loads(r[1]),
            "model": r[2],
            "dim": r[3],
            "pov_summary": r[4],
            "significance": r[5],
            "witness_you": r[6],
            "witness_host": r[7],
            "witness_guest": r[8],
        }
        for r in rows
    ]
 __all__ = [
    "get_embedding",
    "list_embeddings_for_owner",
 ]
@@ -102,6 +102,15 @@ _RECENCY_WEIGHT = 0.5
 # a higher-is-better score by a positive constant per the spec wording.
 SIGNIFICANCE_RANK_BIAS = 0.5
 # T96 (Phase 4): reciprocal-rank-fusion constant used when ``search_memories``
 # is given a ``query_vector`` and must merge FTS + vector candidate lists. The
 # value 60 is the canonical RRF constant from Cormack et al. ("Reciprocal Rank
 # Fusion outperforms Condorcet and Individual Rank Learning Methods", SIGIR
 # 2009): large enough to dampen the head of either ranking so that a strong
 # top-1 in ranking A doesn't crowd out a moderate top-3 in ranking B, but
 # small enough that the position-1/position-2 gap still matters.
 RRF_CONST = 60
 def search_memories(
    conn: Connection,
@@ -109,6 +118,8 @@ def search_memories(
    witness_role: str,
    query: str,
    k: int = 4,
    *,
    query_vector: list[float] | None = None,
 ) -> list[dict]:
    """FTS5 search over pov_summary, scoped by owner and witness role.
@@ -135,6 +146,23 @@ def search_memories(
    * **Python-side** — a composite re-rank with ``_SIGNIFICANCE_WEIGHT``
      reinforces the ordering after candidate retrieval, alongside the
      recency boost above.
    PHASE 4 EXTENSION (T96): when ``query_vector`` is provided, fuses FTS and
    vector hits via reciprocal-rank fusion (RRF):
        fusion_score = 1/(RRF_CONST + fts_rank) + 1/(RRF_CONST + vec_rank)
    where ``fts_rank`` and ``vec_rank`` are the 0-indexed positions of the
    memory in each candidate list. Each candidate gets the sum of its
    reciprocal ranks across both rankings; memories appearing in only one
    ranking still get a partial score (the other term is dropped). Both
    candidate lists are over-fetched at ``k * 2`` so a memory dominant in
    only one channel has a fair chance to surface. The Python-side
    significance + recency re-rank is then applied as a final pass to
    break ties in favour of more important / more recent memories.
    When ``query_vector`` is None: FTS-only behaviour unchanged — all
    Phase 1-3.5 callers see the same row shape and ordering as before.
    """
    if witness_role not in _VALID_WITNESS_ROLES:
        raise ValueError(
@@ -148,7 +176,10 @@ def search_memories(
    select_list = ", ".join(f"m.{c}" for c in cols)
    # Over-fetch from FTS so the Python-side re-rank has room to reorder
    # results that BM25 alone would have demoted past the top-k boundary.
-    over_fetch = max(k * 4, 20)
+    # When fusing with a vector ranking, we still over-fetch (k*2 from each
    # channel) so memories that are weak in FTS but strong in vector — and
    # vice versa — make it into the merge pool.
    over_fetch = max(k * 2, 20) if query_vector is not None else max(k * 4, 20)
    sql = (
        f"SELECT {select_list}, memories_fts.rank AS fts_rank "
        "FROM memories_fts "
@@ -165,11 +196,37 @@ def search_memories(
    )
    cur = conn.execute(sql, (owner_id, query, SIGNIFICANCE_RANK_BIAS, over_fetch))
    rows = cur.fetchall()
    # FTS-only path: preserve pre-T96 behaviour exactly.
    if query_vector is None:
        if not rows:
            return []
        return _composite_rerank(conn, cols, rows, owner_id, k)
-    # Recency normalises against the current max id for this owner so the
+    # Fused path: combine FTS candidates with vector candidates via RRF.
-    # boost magnitude is bounded regardless of dataset size.
+    return _rrf_fuse_and_rerank(
        conn,
        cols=cols,
        fts_rows=rows,
        owner_id=owner_id,
        witness_role=witness_role,
        query_vector=query_vector,
        k=k,
    )
 def _composite_rerank(
    conn: Connection,
    cols: list[str],
    rows: list[tuple],
    owner_id: str,
    k: int,
 ) -> list[dict]:
    """Apply the significance + recency composite re-rank to FTS rows.
    Extracted from ``search_memories`` so the no-vector path stays a single
    call and the fused path can re-use the same boost formulae after RRF.
    """
    max_id_row = conn.execute(
        "SELECT MAX(id) FROM memories WHERE owner_id = ?", (owner_id,)
    ).fetchone()
@@ -187,3 +244,129 @@ def search_memories(
    enriched.sort(key=lambda x: x["composite_score"])
    return enriched[:k]
 def _rrf_fuse_and_rerank(
    conn: Connection,
    *,
    cols: list[str],
    fts_rows: list[tuple],
    owner_id: str,
    witness_role: str,
    query_vector: list[float],
    k: int,
 ) -> list[dict]:
    """Merge FTS + vector candidates via reciprocal-rank fusion, then apply
    the existing significance + recency boost as a final tie-breaker.
    RRF formula (Cormack et al. 2009)::
        fusion_score = sum over rankings r of  1 / (RRF_CONST + rank_r)
    where ``rank_r`` is the 0-indexed position of the memory in ranking r.
    "Missing from a ranking" is handled by SKIPPING the term for that
    ranking — i.e. that channel contributes 0 to the sum, which preserves
    the fairness property: a memory that only appears in one ranking is
    not penalised relative to itself, just relative to memories that
    appeared in both. This matches the canonical RRF presentation.
    The final composite score subtracted from the *negated* fusion score
    is::
        composite = -fusion - sig_boost - recency_boost
    Sorted ascending, smaller-is-better — the same ordering convention as
    the FTS-only path so the Python-side significance + recency boosts
    apply as a clean tie-breaker without inverting any sign.
    """
    # Lazy import to avoid a hard module-level cycle: vector_search reads
    # from chat.state.embeddings, which is itself a sibling of this module.
    from chat.services.vector_search import vector_search
    fts_rank_by_id: dict[int, int] = {}
    fts_row_by_id: dict[int, tuple] = {}
    id_idx = cols.index("id")
    for rank, row in enumerate(fts_rows):
        memory_id = row[id_idx]
        fts_rank_by_id[memory_id] = rank
        fts_row_by_id[memory_id] = row
    # Over-fetch the vector channel symmetrically so each channel gets a
    # fair shot at surfacing its strongest candidates.
    vec_over_fetch = max(k * 2, 20)
    vec_hits = vector_search(
        conn,
        owner_id=owner_id,
        witness_role=witness_role,
        query_vector=query_vector,
        k=vec_over_fetch,
    )
    vec_rank_by_id: dict[int, int] = {
        hit["memory_id"]: rank for rank, hit in enumerate(vec_hits)
    }
    # If the vector channel returned nothing (no embeddings indexed), the
    # fused path collapses cleanly to the FTS-only path. No error, no
    # surprise zero-hit return.
    if not vec_rank_by_id and not fts_row_by_id:
        return []
    if not vec_rank_by_id:
        return _composite_rerank(conn, cols, fts_rows, owner_id, k)
    # For any vector-only hits we don't have a full memory row for yet,
    # fetch them in a single round-trip. The FTS row carries an ``fts_rank``
    # column at the end; vector-only rows get ``None`` there.
    missing_ids = [mid for mid in vec_rank_by_id if mid not in fts_row_by_id]
    select_list = ", ".join(cols)
    if missing_ids:
        placeholders = ",".join("?" * len(missing_ids))
        cur = conn.execute(
            f"SELECT {select_list} FROM memories WHERE id IN ({placeholders})",
            missing_ids,
        )
        for row in cur.fetchall():
            # Pad with a None for the trailing ``fts_rank`` slot so the row
            # shape matches FTS rows downstream.
            fts_row_by_id[row[id_idx]] = tuple(row) + (None,)
    # Compute fusion score per candidate. Missing-from-ranking terms are
    # simply omitted from the sum.
    all_ids = set(fts_rank_by_id) | set(vec_rank_by_id)
    fusion_by_id: dict[int, float] = {}
    for mid in all_ids:
        score = 0.0
        if mid in fts_rank_by_id:
            score += 1.0 / (RRF_CONST + fts_rank_by_id[mid])
        if mid in vec_rank_by_id:
            score += 1.0 / (RRF_CONST + vec_rank_by_id[mid])
        fusion_by_id[mid] = score
    # Final composite re-rank: significance + recency boosts on top of the
    # negated fusion score so the sort direction matches the FTS-only path.
    max_id_row = conn.execute(
        "SELECT MAX(id) FROM memories WHERE owner_id = ?", (owner_id,)
    ).fetchone()
    max_id = max_id_row[0] if max_id_row and max_id_row[0] else 1
    result_cols = cols + ["fts_rank"]
    enriched: list[dict] = []
    for mid in all_ids:
        row = fts_row_by_id.get(mid)
        if row is None:
            # Defensive: a vector hit with no memory row would be a logic
            # bug (vector_search joins memories), so just skip it rather
            # than crash the whole search.
            continue
        d = dict(zip(result_cols, row))
        sig_boost = _SIGNIFICANCE_WEIGHT * (d.get("significance") or 0)
        recency_boost = _RECENCY_WEIGHT * ((d.get("id") or 0) / max_id)
        fusion = fusion_by_id[mid]
        # Sort ascending, smaller-is-better → negate fusion so a larger
        # fusion score yields a smaller composite. Significance and recency
        # boosts then act as tie-breakers exactly like the FTS-only path.
        d["fusion_score"] = fusion
        d["composite_score"] = -fusion - sig_boost - recency_boost
        enriched.append(d)
    enriched.sort(key=lambda x: x["composite_score"])
    return enriched[:k]
@@ -0,0 +1,832 @@
 # Roleplay Engine — Phase 4 Implementation Plan
 > **For Claude:** REQUIRED SUB-SKILL: Use `superpowers-extended-cc:executing-plans` to implement this plan task-by-task. Use the parallel-dispatch pattern documented under "Parallel-Execution Strategy" for parallel waves.
 **Goal:** Land Phase 4 polish per requirements doc §13 + §14: vector retrieval, branching UI, drawer-edit on every field, backup tooling, significance review UI, surgical delete with cascade preview, hide-from-view soft delete, plus cross-chat search and the small Phase 3.6 carry-over fixes.
 **Architecture:** Builds on Phase 3.5's stable base. Two new tables (`embeddings`, `branches`) and one external dependency (sqlite-vec extension). Embedding generation runs as a deferred async job — NOT inline with turns — so the play loop stays fast even when the embedding endpoint is slow. Branching is data-model-only at first (events + selectors); UI grafts on top. Surgical delete + cascade preview reuses the existing rewind-and-supersede plumbing. Cross-chat search piggybacks on the existing FTS5 + (now) vector retrieval.
 **Tech Stack:**
 - **NEW dependency: `sqlite-vec`** (or `sqlite-vss` — Phase 4 picks; recommended `sqlite-vec` for simpler load semantics and active maintenance). Add to `pyproject.toml`.
 - **Embedding model selection** is part of T91 spec. Recommended default: a small model on Featherless (e.g., `BAAI/bge-small-en-v1.5` if available) or a local CPU-friendly model via `sentence-transformers`. Document choice in CLAUDE.md.
 - Same as Phase 3 otherwise (Python 3.11+, FastAPI, HTMX, SQLite).
 **Source-of-truth references:**
 - Phase 4 scope: requirements doc §13 "Phase 4 — polish" + §14 "Open / Deferred Decisions".
 - Behavioral details: §6 (prompt assembly + retrieval), §10 (rewind / regenerate / reset), §11 (compression + significance), §12 (snapshots).
 - Conventions: [`CLAUDE.md`](../../CLAUDE.md) §"Behavioral defaults" + §"Phase 3 status" + §"Phase 3.5 status".
 - Phase 3.5 cleanup plan (style, file-bundling pattern): [2026-04-26-v3.5-phase3.5-cleanup.md](2026-04-26-v3.5-phase3.5-cleanup.md).
 ---
 ## Pre-flight
 **Branch:** create `phase-4` from the latest `main` after Phase 3.5 has merged (it has — main is at `1b66a28`):
 ```bash
 git checkout main && git pull && git checkout -b phase-4
 ```
 **Schema baseline:** Phase 3.5 leaves the DB at version 11. Phase 4 adds two migrations: `0012_embeddings.sql` and `0013_branches.sql`. Final schema version: 13.
 **External dependency setup (BEFORE T88 dispatch):**
 The controlling agent should add `sqlite-vec` to `pyproject.toml` and run `pip install -e .` (or equivalent) so all worktrees pick up the new dependency. Confirm `sqlite_vec` imports cleanly:
 ```bash
 python -c "import sqlite_vec; print(sqlite_vec.__version__)"
 ```
 If `sqlite_vec` isn't on PyPI when this plan executes, fall back to `sqlite-vss` and adapt T88/T92 accordingly. Both expose vector-search SQL via a loadable extension.
 **Pinned non-negotiables (carried forward):**
 - State changes go through the event log. Use `append_and_apply(conn, kind, payload)` for the live path; `apply_event` only after a fresh `append_event` returning the new id.
 - Witness filter every memory read at SQL level (hard `WHERE` constraint; never a soft signal).
 - Per-POV scene summaries — never write omniscient narration.
 - TDD: every task starts with a failing test (or a regression test pinning existing contract before refactor).
 - One commit per task minimum. Tasks that bundle multiple sub-features SHOULD split commits internally.
 **Verification before claiming done:** Use `superpowers-extended-cc:verification-before-completion` — run the test command, paste actual output. Don't assume green.
 ---
 ## Phase 3.6 carry-overs folded in
 Three small items from Phase 3.6 backlog are bundled into Phase 4's Wave 1 trivial-fixes task (T90):
 1. `read_recent_dialogue` chat-id pushdown into SQL (T80 review nit)
 2. Lifecycle warning wording in regenerate (T83.4 — "at-or-after turn X" tightening)
 3. Legacy single-bot `record_turn_memory` consolidation (T84 review nit)
 Three items remain DEFERRED beyond Phase 4 (Phase 4.5 if needed):
 - Scene-close-on-cancel UX revisit (no action unless real play surfaces a regression).
 - Cross-feature canned-queue brittleness (structured fixture builder for tests — not blocking).
 - Full lifecycle-rollback in regenerate (warning log already shipped in T83.4; proper rollback needs schema-level back-references, deferred indefinitely).
 ---
 ## Parallel-Execution Strategy
 Same pattern as Phase 3.5. Eight waves: parallel within each wave (file-disjoint), serial across waves.
 ### How to dispatch a wave in parallel
 Use the **Agent tool with `isolation: "worktree"`** so each subagent gets its own git worktree. (If the controlling session's working directory is **not** the chat repo, create worktrees manually with `git worktree add .worktrees/<wave>-<task> -b <wave>/<task> phase-4` from inside the chat repo.)
 Dispatch all tasks in a wave in a single message:
 ```
 Agent({ description: "Wave 1 — T88 embeddings table", prompt: "...", isolation: "worktree" })
 Agent({ description: "Wave 1 — T89 branches table", ... })
 Agent({ description: "Wave 1 — T90 phase 3.6 carry-overs", ... })
 ```
 ### After a wave completes
 1. Each subagent returns its worktree path and commit SHA(s).
 2. **Run a spec + code-quality reviewer subagent on each completed task.** Combined review acceptable for trivial tasks (T90 carry-overs); separate spec + quality reviewers for vector-retrieval tasks (T91, T92, T96, T97) since the integration surface is wider.
 3. **Merge the wave into `phase-4`** in any order (file-disjointness guarantees no conflict). Use `--no-ff`.
 4. **Run the full test suite** on the merged `phase-4`. If red, the wave's mutual-independence assumption was violated — bisect, fix, re-merge.
 5. **Push `phase-4`** to gitea.
 6. Optionally clean up worktrees.
 ### Conflict prevention checklist
 For each parallel wave, verify the **Files** sections of all tasks have **no overlapping paths**. Hot files in this plan: `chat/web/drawer.py` + `chat/templates/_drawer.html` (T98 only — bundled), `chat/state/memory.py` (T96 only), `chat/services/memory_write.py` (T90 + T97 — sequential), `chat/web/turns.py` (T98 only via delete affordance — sequential after T96).
 ### Why each wave is parallel-safe
 | Wave | Tasks | Hot files touched | Disjoint? |
 |------|-------|-------------------|-----------|
 | 1 | T88, T89, T90 | new migrations + new state modules; T90 touches `turn_common.py` + `regenerate.py` + `memory_write.py` (additive only) | ✅ |
 | 2 | T91, T92, T93 | new service modules (embeddings, vector_search, cross_chat_search) | ✅ |
 | 3 | T94, T95 | new service modules (branching, delete_impact) | ✅ |
 | 4 | T96 | `chat/state/memory.py` (combined retrieval ranking) | (single task) |
 | 5 | T97 | `chat/services/memory_write.py` + new backfill script | (single task) |
 | 6 | T98 | `chat/web/drawer.py` + `chat/templates/_drawer.html` (drawer Phase 4 bundle) | (single task) |
 | 7 | T99, T100 | new files: `chat/web/snapshots.py` + `chat/templates/snapshots.html` (T99); `chat/web/search.py` + `chat/templates/search.html` + small chat.html top-bar addition (T100) | ✅ (disjoint) |
 | 8 | T101, T102 | new test file (T101); CLAUDE.md + design doc (T102) | ✅ |
 ---
 ## Task overview
 ```
 Wave 1 ─┬─ T88: embeddings table + projector handlers
        ├─ T89: branches table + projector handlers
        └─ T90: Phase 3.6 carry-overs trio (chat-id SQL pushdown + lifecycle wording + legacy-fn consolidation)
 Wave 2 ─┬─ T91: embedding generation service (Featherless or local)
        ├─ T92: vector search service via sqlite-vec
        └─ T93: cross-chat search service (FTS over all owners)
 Wave 3 ─┬─ T94: branch_from_event service (event-log fork, branch metadata)
        └─ T95: delete-impact computation service (cascade preview)
 Wave 4 ─── T96: combined FTS + vector retrieval ranking in search_memories
 Wave 5 ─── T97: memory_write enqueues embedding job + backfill script for existing memories
 Wave 6 ─── T98: drawer Phase 4 bundle — branching UI + significance review + hide-from-view + surgical delete + remaining v1 edits
 Wave 7 ─┬─ T99: snapshot UX (manual trigger, retention display, restore-from-snapshot UI)
        └─ T100: cross-chat search UX (top-bar input + search results page)
 Wave 8 ─┬─ T101: cross-feature integration tests (vector × branching × delete × snapshot × search)
        └─ T102: Phase 4 documentation update
 ```
 Critical path: 8 sequential merge points. Total tasks: 15. Parallelism: Waves 1, 2, 3, 7, 8 dispatch concurrently (3-way and 2-way). Waves 4, 5, 6 are single-task by hot-file constraint.
 ---
 ## Wave 1 — Schema foundation + Phase 3.6 carry-overs (parallel)
 ### Task 88: Embeddings table + projector handlers
 **Files:**
 - Create: `chat/db/migrations/0012_embeddings.sql`
 - Create: `chat/state/embeddings.py`
 - Create: `tests/test_embeddings_state.py`
 - Modify: `pyproject.toml` (add `sqlite-vec` dependency — controlling agent should pre-install before dispatch; the worktree commits the dependency declaration)
 **Spec:**
 Adds the `embeddings` table that stores per-memory embedding vectors for vector retrieval. Uses `sqlite-vec` virtual-table syntax for cosine-similarity search. Schema:
 ```sql
 -- Load sqlite-vec extension at connection time (handled in chat/db/connection.py).
 -- Embeddings are stored as blobs in a vec0 virtual table for fast similarity search.
 CREATE VIRTUAL TABLE embeddings USING vec0(
    memory_id INTEGER PRIMARY KEY,
    embedding FLOAT[384]   -- 384-dim default; adjust per chosen model
 );
 -- Sidecar table for non-vector metadata (model used, dim, indexed_at).
 CREATE TABLE embeddings_meta (
    memory_id INTEGER PRIMARY KEY,
    model TEXT NOT NULL,
    dim INTEGER NOT NULL,
    indexed_at TEXT NOT NULL DEFAULT (datetime('now')),
    FOREIGN KEY (memory_id) REFERENCES memories(id)
 );
 ```
 (If `sqlite-vss` is chosen instead, replace `vec0` with `vss0` and adapt the dim declaration. Both have similar Python loading semantics.)
 **`chat/state/embeddings.py`:**
 - `@on("embedding_indexed")` payload `{memory_id, model, dim, vector: list[float]}`. Inserts into both `embeddings` and `embeddings_meta`. Idempotent via `INSERT OR REPLACE` (re-indexing a memory replaces the prior vector).
 - `@on("embedding_deindexed")` payload `{memory_id}`. Deletes from both tables. Used when a memory is purged via reset/cascade.
 - Reader `get_embedding_meta(conn, memory_id) -> dict | None` returns the meta row.
 The `chat/db/connection.py` `open_db` helper needs to load the sqlite-vec extension on each connection. Add:
 ```python
 import sqlite_vec
 # Inside open_db, after connection is opened:
 conn.enable_load_extension(True)
 sqlite_vec.load(conn)
 conn.enable_load_extension(False)
 ```
 This is a small modification to `connection.py`. Include it in T88's diff.
 **Tests:** 3 minimum.
 1. `test_embedding_indexed_inserts_row`: append `bot_authored`, `chat_created`, `memory_written` (creates a memory), then `embedding_indexed` with `vector=[0.1] * 384`. Project. Assert `embeddings_meta` row exists for that memory_id with the right model.
 2. `test_embedding_deindexed_removes_row`: same setup; index then de-index; assert row is gone.
 3. `test_vector_similarity_search_returns_nearest`: index two memories with distinct vectors; query for nearest neighbor of one vector; assert correct memory_id returned. Uses `sqlite-vec`'s `MATCH '...'` syntax (verify against actual sqlite-vec docs; adapt if needed).
 If running tests requires sqlite-vec to be loaded, the test fixture may need to skip / xfail when the extension isn't installed. Use `pytest.importorskip("sqlite_vec")` at the top of the test file.
 **Commit:** `feat: embeddings table + projector handlers via sqlite-vec (T88)`.
 **Notes:**
 - Schema version after migration alone: 12. T89 adds 0013, taking final to 13. The schema_version assertion in `tests/test_world.py` updates to 13 in the wave-merge step.
 - The `connection.py` change is small but cross-cutting — affects every `open_db` call. Verify the existing 343 tests still pass after the change.
 ---
 ### Task 89: Branches table + projector handlers
 **Files:**
 - Create: `chat/db/migrations/0013_branches.sql`
 - Create: `chat/state/branches.py`
 - Create: `tests/test_branches_state.py`
 **Spec:**
 Adds the `branches` table that records named alternate event-log forks. A branch is metadata: a name, an `origin_event_id` (the event we forked from), and a `head_event_id` (the latest event in this branch). The event log itself is unchanged — the branch table just **labels** linear ranges of event ids.
 ```sql
 CREATE TABLE branches (
    id INTEGER PRIMARY KEY,
    name TEXT NOT NULL UNIQUE,
    origin_event_id INTEGER NOT NULL,
    head_event_id INTEGER NOT NULL,
    chat_id TEXT,
    created_at TEXT NOT NULL DEFAULT (datetime('now')),
    is_active INTEGER NOT NULL DEFAULT 0
 );
 -- Exactly one row may have is_active = 1 at any time.
 CREATE UNIQUE INDEX branches_active_idx ON branches(is_active) WHERE is_active = 1;
 ```
 The "main" branch is implicit and bootstrapped by the migration: `INSERT INTO branches (name, origin_event_id, head_event_id, is_active) VALUES ('main', 0, 0, 1);`. Subsequent branches reference an `origin_event_id` (the event that the branch forked from).
 `chat/state/branches.py`:
 - `@on("branch_created")` payload `{name, origin_event_id, chat_id?, head_event_id}`. Inserts a new row with `is_active=0`. Idempotent re-insertion via `INSERT OR IGNORE`.
 - `@on("branch_switched")` payload `{name}`. Sets `is_active=1` on the named branch and `is_active=0` on all others. Atomic via a single UPDATE.
 - `@on("branch_head_updated")` payload `{name, head_event_id}`. Updates `head_event_id` on the named branch. Used by the orchestrator when new events extend the branch.
 - Readers: `get_branch(conn, name)`, `list_branches(conn, chat_id=None)`, `active_branch(conn)`.
 **Tests:** 3 minimum.
 1. `test_branch_created_inserts_row`: append `branch_created` with name="experiment", origin_event_id=42; project; assert `get_branch(conn, "experiment")` returns the row.
 2. `test_branch_switched_atomic`: seed two branches; switch from one to the other; assert exactly one is active.
 3. `test_main_branch_bootstrapped_by_migration`: open a fresh DB, apply migrations; assert `active_branch(conn)["name"] == "main"`.
 **Commit:** `feat: branches table + projector handlers (T89)`.
 **Notes:**
 - Schema version after this migration alone: 13. Combined with T88: 13 (since T88 was 12, T89 stacks). Wave-merge bumps `tests/test_world.py` schema_version assertion to 13.
 - This task does NOT yet teach the orchestrator to consult `is_active` — the existing event_log queries assume a single timeline. T98 (drawer branching UI) will enable user-driven switches, but the actual "follow only the active branch" filter on event reads is a follow-up (Phase 4.5 nit; document in T102 docs sweep).
 ---
 ### Task 90: Phase 3.6 carry-overs trio
 **Files:**
 - Modify: `chat/services/turn_common.py` (push chat_id filter into SQL)
 - Modify: `chat/services/regenerate.py` (lifecycle warning wording tightening)
 - Modify: `chat/services/memory_write.py` (consolidate legacy `record_turn_memory` into the unified API or delete it)
 - Modify: `tests/test_turn_common.py`, `tests/test_regenerate.py`, `tests/test_memory_write.py`
 **Spec:** Three small Phase 3.6 carry-over fixes bundled because each is 1-line + 1-test.
 #### 90.1 — `read_recent_dialogue` chat-id SQL pushdown
 Per T80 review nit. Currently `read_recent_dialogue` filters chat_id post-fetch in Python. Push into SQL for tighter LIMIT semantics:
 ```sql
 SELECT id, kind, payload_json
 FROM event_log
 WHERE kind IN ('user_turn', 'user_turn_edit', 'assistant_turn')
  AND superseded_by IS NULL
  AND hidden = 0
  AND json_extract(payload_json, '$.chat_id') = ?
 ORDER BY id DESC
 LIMIT ?
 ```
 Then the post-fetch loop becomes a simple reverse + slice — no chat_id check needed.
 **Test added:** `test_read_recent_dialogue_limit_respects_chat_scope` — seed two chats with 60 turns each; query chat_a with `limit=50`; assert returned rows are exactly 50 chat_a rows (not 50 cross-chat rows that filter down to <50 after Python).
 **Commit:** `perf: read_recent_dialogue pushes chat-id filter into SQL (T90.1)`.
 #### 90.2 — Lifecycle warning wording tightening
 Per T83.4 review nit. Current warning lists "lifecycle transitions from superseded turn are NOT being rolled back". When user regenerates an OLDER turn (T29 supports this), the warning lists intervening-turn transitions that legitimately stand. Tighten wording to "lifecycle transitions at-or-after turn X" so operators reading logs aren't misled.
 Change is one log message string. Test asserts the new wording appears.
 **Commit:** `chore: clarify regenerate lifecycle warning wording (T90.2)`.
 #### 90.3 — Legacy `record_turn_memory` consolidation
 Per T84 review nit. The original Phase 1 single-bot `record_turn_memory` function still exists alongside the unified `record_turn_memory_for_present`. Either:
 - (a) Remove the legacy function entirely; update any remaining callers to use the unified API.
 - (b) Convert it to a thin wrapper for backward compat.
 Pick (a) if there are zero remaining callers; (b) if any callers exist. Read the codebase to confirm. The mock-data seed scripts may still use the legacy fn.
 **Commit:** `refactor: consolidate legacy record_turn_memory into unified API (T90.3)`.
 **TDD process for T90:**
 1. Read all 3 affected files + their tests.
 2. Implement 90.1 with test; commit.
 3. Implement 90.2 with test; commit.
 4. Implement 90.3 with test; commit.
 5. Run full suite — should be 343 + 3 = 346 (or +2 if 90.3 had no behavioral change).
 ---
 ## Wave 2 — Embedding & search services (parallel)
 Three new service modules. Fully file-disjoint.
 ### Task 91: Embedding generation service
 **Files:**
 - Create: `chat/services/embeddings.py`
 - Create: `tests/test_embeddings.py`
 **Spec:** Wraps the embedding API call. Signature:
 ```python
 class EmbeddingResult(BaseModel):
    vector: list[float]
    model: str
    dim: int
 async def generate_embedding(
    client: LLMClient,    # or a separate embedding-specific client
    *,
    text: str,
    model: str,
    timeout_s: float = 30.0,
 ) -> EmbeddingResult:
    """Generate an embedding vector for the given text. Falls back to a
    zero-vector with model='fallback' on failure (so callers get a deterministic
    sentinel they can detect and skip indexing)."""
 ```
 **Implementation:** call the embedding endpoint (Featherless OpenAI-compatible `/v1/embeddings`, or a local `sentence-transformers` model). Add a new method `client.embed(text, model)` to `LLMClient` Protocol (and to `MockLLMClient` and `FeatherlessClient`).
 **Embedding model choice:**
 Default to a small CPU-friendly model accessible through the existing Featherless setup:
 - If Featherless has `BAAI/bge-small-en-v1.5` or similar 384-dim model: use that.
 - If not: fall back to local `sentence-transformers/all-MiniLM-L6-v2` (384-dim, runs CPU). Add `sentence-transformers` to `pyproject.toml`.
 - Document choice in CLAUDE.md (T102 docs sweep).
 The 384 dim is hardcoded in T88's migration. If a different model with different dim is chosen, update T88's schema accordingly BEFORE T88 dispatches.
 **Tests:** 3 minimum.
 1. `test_generate_embedding_returns_vector_of_correct_dim`: mock embedding response with a 384-element vector; assert returned `vector` length is 384.
 2. `test_generate_embedding_returns_correct_model_metadata`: assert `result.model` matches the input.
 3. `test_generate_embedding_falls_back_on_failure`: mock the client to raise; assert the result is a 384-element zero vector with `model="fallback"`.
 **Commit:** `feat: embedding generation service (T91)`.
 ---
 ### Task 92: Vector search service via sqlite-vec
 **Files:**
 - Create: `chat/services/vector_search.py`
 - Create: `tests/test_vector_search.py`
 **Spec:** Wraps sqlite-vec's `MATCH` syntax for cosine-similarity search over the `embeddings` virtual table. Witness-filter aware (joins through `memories` table for the witness check).
 ```python
 def vector_search(
    conn,
    *,
    owner_id: str,
    witness_role: str,    # "you" | "host" | "guest"
    query_vector: list[float],
    k: int = 4,
 ) -> list[dict]:
    """Return top-K memories by cosine similarity to query_vector,
    witness-filtered for the requesting bot's POV. Returns same row
    shape as state.memory.search_memories for combined-ranking
    compatibility."""
 ```
 SQL pattern (sqlite-vec):
 ```sql
 SELECT m.id, m.text, m.pov_summary, m.significance, e.distance
 FROM embeddings e
 JOIN memories m ON m.id = e.memory_id
 WHERE e.embedding MATCH ?
  AND k = ?
  AND m.owner_id = ?
  AND m.witness_<role> = 1
 ORDER BY e.distance ASC
 LIMIT ?
 ```
 (Adapt to actual sqlite-vec syntax — use `vec0` MATCH semantics. The `witness_<role>` interpolation needs the same allowlist guard pattern as Phase 2.5 T72.3.)
 **Tests:** 3 minimum.
 1. `test_vector_search_returns_nearest_neighbors`: index 5 memories with synthetic vectors; query for nearest 3; assert correct order.
 2. `test_vector_search_respects_witness_filter`: index a memory with witness `[1, 1, 0]`; query with `witness_role="guest"`; assert empty result.
 3. `test_vector_search_respects_owner_filter`: index memories for two owners; assert query for owner_a doesn't return owner_b's memories.
 **Commit:** `feat: vector search service via sqlite-vec (T92)`.
 ---
 ### Task 93: Cross-chat search service
 **Files:**
 - Create: `chat/services/cross_chat_search.py`
 - Create: `tests/test_cross_chat_search.py`
 **Spec:** FTS5-based search across ALL chats and all owners (admin-style search; no witness filter). For "where did I last see this person mention X?" queries.
 ```python
 def search_all_memories(
    conn,
    *,
    query: str,
    k: int = 20,
 ) -> list[dict]:
    """Search FTS across all owners and chats. Returns rows with
    {memory_id, owner_id, chat_id, text, pov_summary, scene_id,
    significance, ts}. Sorted by FTS rank."""
 ```
 This is intentionally NOT witness-filtered — it's a power-user search surface. The UI (T100) prompts the user to acknowledge they're seeing memories across POVs.
 **Tests:** 3 minimum.
 1. `test_search_all_memories_returns_matches_across_owners`: seed 2 owners with overlapping keyword; search; assert both owner's matches appear.
 2. `test_search_all_memories_orders_by_fts_rank`: seed memories with varying FTS-match strength; assert order.
 3. `test_search_all_memories_respects_k_limit`.
 **Commit:** `feat: cross-chat search service (FTS5 over all owners) (T93)`.
 ---
 ## Wave 3 — Branching + delete services (parallel)
 Two new service modules. Fully file-disjoint.
 ### Task 94: branch_from_event service
 **Files:**
 - Create: `chat/services/branching.py`
 - Create: `tests/test_branching.py`
 **Spec:**
 ```python
 def branch_from_event(
    conn,
    *,
    name: str,
    origin_event_id: int,
    chat_id: str | None = None,
 ) -> int:
    """Create a new named branch forking from origin_event_id.
    Emits a branch_created event. Returns the new branch's row id.
    Raises ValueError if name already exists."""
 def switch_active_branch(conn, *, name: str) -> None:
    """Make the named branch active. Emits branch_switched. Subsequent
    event reads should consult is_active to filter."""
 def list_branches_with_metadata(conn, chat_id: str | None = None) -> list[dict]:
    """List branches with: name, origin_event_id, head_event_id, is_active,
    event_count (number of events between origin and head, inclusive),
    created_at."""
 ```
 Tests cover: basic create, duplicate-name raises, switch updates `is_active` exclusively, list returns metadata.
 **Commit:** `feat: branching service (T94)`.
 ---
 ### Task 95: Delete-impact computation service
 **Files:**
 - Create: `chat/services/delete_impact.py`
 - Create: `tests/test_delete_impact.py`
 **Spec:** Computes the cascade impact of deleting a single event_log row (or a turn group: user_turn + assistant_turn + interjection if any). Returns a structured `ImpactReport` for the UI to render.
 ```python
 class DeletedItem(BaseModel):
    kind: str           # "memory" | "edge_update" | "scene_close" | etc.
    description: str    # human-readable
    target_id: int | str | None
 class ImpactReport(BaseModel):
    target_event_id: int
    cascading: list[DeletedItem]
    notes: list[str]    # warnings, e.g. "this turn opened scene_X which has 3 subsequent turns"
 def compute_delete_impact(conn, *, target_event_id: int) -> ImpactReport:
    """Walk the event log forward from target_event_id and identify
    everything that depends on this event: child memory_written events,
    edge_update events with this turn as source, scene_closed events
    triggered by this turn, etc. Also identify subsequent turns that
    REFERENCE this event (regenerated_from chains, etc.).
    Does NOT mutate the database. Pure computation for preview."""
 ```
 The actual delete (truncate + supersede) is the existing rewind path from Phase 1 T31. T95 just builds the preview.
 **Tests:** 4 minimum.
 1. `test_impact_for_simple_turn_lists_memory_and_edges`: seed a chat with a turn that wrote 1 memory + 2 edge_updates. Compute impact. Assert the 3 items appear in `cascading`.
 2. `test_impact_for_scene_opening_turn_warns_about_subsequent_turns`: seed a turn that opened a scene + 5 subsequent turns. Assert `notes` mentions the dependency.
 3. `test_impact_for_regenerated_turn_lists_supersede_chain`: seed a turn that's been regenerated (has `superseded_by`). Compute impact for the original. Assert the chain appears.
 4. `test_impact_does_not_mutate_database`: snapshot event_log before + after; assert byte-identical.
 **Commit:** `feat: delete-impact computation service (T95)`.
 ---
 ## Wave 4 — Combined retrieval ranking (single)
 ### Task 96: Combined FTS + vector retrieval ranking
 **Files:**
 - Modify: `chat/state/memory.py` — extend `search_memories` to optionally include vector hits
 - Modify: `tests/test_memory_search.py` — add 4 tests
 **Spec:**
 `search_memories` currently does FTS5 + Python-side significance/recency re-rank. Phase 4 adds:
 - An optional `query_vector: list[float] | None = None` kwarg.
 - When `query_vector` is provided, run `vector_search` (T92) for top-K-vector candidates.
 - Merge with FTS top-K candidates via reciprocal-rank fusion (RRF) or a simpler sum-of-ranks scheme — implementer's choice. Document the merge formula.
 - Final result is top-K from the fused set, with the existing significance + recency boosts applied as a final pass.
 When `query_vector` is None: existing behavior unchanged. Phase 1/2/3 callers that don't pass `query_vector` see no change.
 **Implementation note:** the embedding for the query (the speaker's recent context) must be generated by the caller (Wave 5 T97 wires the prompt-assembly pipeline to call `generate_embedding` on the dialogue tail). T96 only handles the search side — assumes the vector is pre-computed.
 **Tests:** 4 added.
 1. `test_search_memories_without_query_vector_uses_fts_only`: regression — call without `query_vector`; assert the existing FTS+rerank behavior.
 2. `test_search_memories_with_query_vector_includes_vector_hits`: index 5 memories where 1 is FTS-only-matching, 1 is vector-only-matching, 3 are unrelated. Pass both `query=...` and `query_vector=...`. Assert both the FTS hit and the vector hit appear in results.
 3. `test_search_memories_fusion_significance_bias_still_applies`: confirm the existing significance bias rerank still works on top of fused results.
 4. `test_search_memories_fusion_handles_empty_vector_results`: pass a vector for a memory that has no embeddings indexed; assert FTS-only results still come back.
 **Commit:** `feat: combined FTS + vector retrieval ranking (T96)`.
 ---
 ## Wave 5 — Memory write hook + backfill (single)
 ### Task 97: Embedding generation hook + backfill script
 **Files:**
 - Modify: `chat/services/memory_write.py` — after each `memory_written` event, enqueue a background embedding job
 - Create: `chat/services/embedding_worker.py` — async worker that consumes the queue and emits `embedding_indexed` events
 - Create: `scripts/backfill_embeddings.py` — one-time script that walks all existing memories and embeds them
 - Modify: `chat/app.py` — wire the embedding worker into the lifespan startup
 - Modify: `tests/test_memory_write.py` — add 2 tests for the enqueue hook
 - Create: `tests/test_embedding_worker.py` — 3 tests for the worker drain logic
 **Spec:**
 After each successful `memory_written` event, enqueue an embedding job. The worker dequeues and:
 1. Reads the memory text (via `get_memory(conn, memory_id)`).
 2. Calls `generate_embedding(client, text=memory.text, model=settings.embedding_model)`.
 3. Appends `embedding_indexed` event with the result. (Skip if `result.model == "fallback"` — leave the memory un-indexed; will retry later via backfill.)
 The worker pattern mirrors Phase 1's `chat/services/significance.py` SignificanceWorker. Reuse its queue + lifecycle pattern.
 **Backfill script:**
 ```bash
 .venv/bin/python scripts/backfill_embeddings.py [--limit N] [--dry-run]
 ```
 Walks all memories where no `embeddings_meta` row exists. For each, generates an embedding and emits `embedding_indexed`. Useful for the initial migration after Phase 4 lands AND for periodic re-runs if an embedding model changes.
 **Tests:**
 `tests/test_memory_write.py`:
 1. `test_record_turn_memory_enqueues_embedding_job`: monkeypatch the worker's enqueue method; record_turn_memory_for_present; assert the worker received a job per memory.
 `tests/test_embedding_worker.py`:
 1. `test_worker_drains_jobs_and_emits_indexed_events`: enqueue 3 jobs with mock embeddings; run worker; assert 3 `embedding_indexed` events landed.
 2. `test_worker_skips_fallback_results`: mock the embedding service to return a fallback result; assert NO `embedding_indexed` event landed for that job.
 3. `test_worker_handles_concurrent_jobs_serially`: pin the Featherless 2-conn cap behavior (worker calls embed sequentially under the existing semaphore).
 **Commit (split):**
 - `feat: embedding worker drains queue and emits embedding_indexed events (T97.1)`
 - `feat: memory_write enqueues embedding job after each memory_written (T97.2)`
 - `feat: backfill_embeddings script for existing memories (T97.3)`
 **Verification gates:**
 - All Phase 1/2/3/3.5 memory tests still pass (regression critical).
 - New tests pass.
 - Manual smoke: run `scripts/backfill_embeddings.py --dry-run` against a seeded DB and verify expected count.
 ---
 ## Wave 6 — Drawer Phase 4 bundle (single task)
 ### Task 98: Drawer Phase 4 features
 **Files:**
 - Modify: `chat/web/drawer.py` (add many new POST routes and GET extensions)
 - Modify: `chat/templates/_drawer.html` (add 5 new sections)
 - Create: `tests/test_drawer_phase4.py`
 **Spec:** Drawer affordances for 5 Phase 4 features. Single task by hot-file constraint; split into 5 commits internally.
 #### 98.1 — Branching UI
 GET drawer extension: `list_branches_with_metadata(conn)` → render in a "Branches" section (active branch highlighted + count of events).
 POST routes:
 - `/drawer/branch/create` — form `{name, origin_event_id}` → `branch_from_event` service.
 - `/drawer/branch/switch` — form `{name}` → `switch_active_branch`.
 - `/drawer/branch/from-turn/{event_id}` — convenience: branch from a specific turn (used by per-turn UI affordance).
 #### 98.2 — Significance review panel
 GET extension: significance distribution per chat (`SELECT significance, COUNT(*) GROUP BY significance`) → render histogram.
 POST route:
 - `/drawer/memory/significance/{memory_id}` — form `{new_value}` (already supported via T22 `manual_edit` `target_kind=memory_significance`); just add the UI form.
 Bulk re-rate is a Phase 4.5 polish — not in scope here. Just per-memory edit + distribution display.
 #### 98.3 — Hide-from-view toggle
 POST route:
 - `/drawer/turn/hide/{event_id}` — form `{hidden: bool}` → emits a `manual_edit` with `target_kind="turn_hidden"`.
 NEW `manual_edit` projector branch for `turn_hidden`: sets `event_log.hidden = ?` for the target event. Reuses the existing `hidden` column.
 UI affordance: per-turn checkbox in the chat surface or drawer (per-turn list with hide toggle).
 #### 98.4 — Surgical delete with cascade preview
 GET extension:
 - `/drawer/turn/delete-preview/{event_id}` → returns the `ImpactReport` (T95) rendered as a modal.
 POST route:
 - `/drawer/turn/delete/{event_id}` — invokes the rewind-and-truncate path (Phase 1 T31's `rewind_to_turn`) restricted to the target turn group.
 Important: this reuses the existing pre-rewind snapshot path so the action is undoable.
 #### 98.5 — Remaining v1 edits
 Audit: are any v1 fields STILL not editable from the drawer? Phase 2.5 T72.1 added edge_trust/edge_summary/memory_pov_summary/edge_knowledge_facts. T72.3 added witness flags. Anything left?
 Likely candidates: scene `narrative_anchor`, scene `weather`, container `properties` JSON. Add edit forms for any that surface during the audit. If none, this sub-fix is a no-op.
 **Tests:** 8+ in `tests/test_drawer_phase4.py` (one per sub-feature × happy path; plus 1 for the cascade-preview rendering).
 **Commits (5):**
 - `feat: drawer branching UI (T98.1)`
 - `feat: drawer significance review panel (T98.2)`
 - `feat: drawer hide-from-view toggle + manual_edit turn_hidden branch (T98.3)`
 - `feat: drawer surgical delete with cascade preview (T98.4)`
 - `feat: drawer remaining v1 field edits (T98.5)` (or "no-op audit" if nothing left)
 ---
 ## Wave 7 — Snapshot + cross-chat search UX (parallel)
 ### Task 99: Snapshot UX
 **Files:**
 - Create: `chat/web/snapshots.py` (new route module)
 - Create: `chat/templates/snapshots.html` (snapshot list page)
 - Modify: `chat/templates/layout.html` (add "Snapshots" nav link)
 - Create: `tests/test_snapshot_ux.py`
 **Spec:** Surface the existing snapshot infrastructure (Phase 1 T20 wrote snapshots; Phase 4 makes them visible).
 GET `/snapshots` — list all snapshots (periodic + pre-rewind) with metadata: kind, created_at, event_log_size, file_size_bytes.
 POST `/snapshots/take` — manually trigger a snapshot now.
 POST `/snapshots/restore/{snapshot_id}` — restore from snapshot (with hard confirmation).
 GET `/snapshots/{snapshot_id}/preview` — show what's in the snapshot vs. current state.
 **Tests:** 4 minimum (list, take, restore, preview).
 **Commit:** `feat: snapshot UX (manual trigger, list, restore) (T99)`.
 ---
 ### Task 100: Cross-chat search UX
 **Files:**
 - Create: `chat/web/search.py` (new route module)
 - Create: `chat/templates/search.html` (search results page)
 - Modify: `chat/templates/layout.html` (add top-bar search input)
 - Create: `tests/test_search_ux.py`
 **Spec:** Top-bar search box submits to `/search?q=...`. Results page shows up to 50 matches across all chats and all owners (uses T93's `search_all_memories`). Each result shows: chat name, owner bot name, scene context, memory text excerpt with FTS highlight, "Open chat at this turn" link.
 **Tests:** 3 minimum.
 1. Search returns results from multiple chats.
 2. Empty query returns empty result set.
 3. Result links navigate to the right chat anchor.
 **Commit:** `feat: cross-chat search UX (top-bar input + results page) (T100)`.
 ---
 ## Wave 8 — Polish (parallel)
 ### Task 101: Cross-feature integration tests
 **Files:**
 - Create: `tests/test_phase4_integration.py`
 **Spec:** End-to-end multi-feature flows. 5 tests minimum.
 1. **Vector retrieval feedback loop**: write a memory → embedding worker indexes it → search retrieves it via vector path.
 2. **Branch + diverge**: create branch B from turn 10 → switch to B → play 3 new turns → switch back to main → assert main's turn 11+ are still intact.
 3. **Surgical delete**: compute impact for a turn → confirm → assert event log truncated correctly + pre-rewind snapshot saved.
 4. **Hide + retrieval**: hide a turn → assert it doesn't appear in `read_recent_dialogue` (existing `hidden = 0` filter) → unhide → assert it reappears.
 5. **Cross-chat search**: write memories in 3 chats → search for keyword present in all 3 → assert all 3 appear in results.
 **Commit:** `test: phase 4 cross-feature integration coverage (T101)`.
 ---
 ### Task 102: Phase 4 documentation update
 **Files:**
 - Modify: `CLAUDE.md` (add "Phase 4 status" section; update behavioral defaults; add "Phase 4.5 / 5 backlog" with carry-overs)
 - Modify: `docs/plans/2026-04-26-v1-requirements-design.md` (annotate §13 Phase 4 as **Status: shipped 2026-04-27**)
 **Spec:**
 Mirror the Phase 3 / 3.5 status sections. Document:
 - **Vector retrieval**: sqlite-vec virtual table, embedding worker async pipeline, combined FTS + vector ranking via RRF.
 - **Branching**: forks the event log; UI in drawer; `is_active` flag plus orchestrator filter (caveat — see backlog if filter not yet wired into all readers).
 - **Drawer-edit on every field**: branching, significance review, hide-from-view, surgical delete with preview, plus any audit findings.
 - **Backup tooling**: snapshots panel surfaces existing infra.
 - **Significance review UI**: distribution + per-memory edit.
 - **Surgical delete + cascade preview**: piggybacks on rewind path; impact report from T95.
 - **Hide-from-view soft delete**: `manual_edit` `turn_hidden` branch.
 - **Cross-chat search**: top-bar + results page over T93's service.
 **Phase 4.5 / 5 backlog candidates** (reflect any discovered during execution):
 - Branching read-side filter — if T89's `is_active` isn't yet consulted by every event reader, this is the work to do.
 - Bulk significance re-rate (per T98.2 deferral).
 - Snapshot retention policy UI controls (per Phase 1 T19 deferred).
 - Auto-pin override UI (per Phase 2 design).
 - Embedding model swap migration tooling (when changing embedding model, need to re-embed everything).
 - Vector index optimization (HNSW vs flat — Phase 5 if needed).
 - Carry-overs that remained deferred from Phase 3.6: scene-close-on-cancel UX revisit, canned-queue brittleness fixture builder, full lifecycle rollback in regenerate.
 **Commit:** `docs: phase 4 status, behavioral defaults, deferred items (T102)`.
 ---
 ## Wrap-up
 After Wave 8 lands:
 1. **Run full suite** on `phase-4`: should be ~390+ tests passing (343 from Phase 3.5 + ~50 new).
 2. **Manual smoke** (recommended before opening the PR):
   - Run `scripts/backfill_embeddings.py` against a seeded DB to verify vector indexing works.
   - Search for a phrase that's substring-distinct but semantically similar to a memory; verify vector path returns it (FTS would miss).
   - Create a branch from an old turn; switch; play a few turns; switch back.
   - Trigger surgical delete on a turn; verify the impact preview matches what actually gets removed.
   - Hide a turn; verify it disappears from the chat surface; unhide.
   - Use top-bar search to find a phrase; verify cross-chat results appear.
   - Click the "Snapshots" nav link; trigger a manual snapshot; verify it appears.
 3. **Push `phase-4`** to gitea.
 4. **Open PR** `phase-4 → main`.
 ---
 ## Notes for the controller running this plan
 - **External dependency**: `sqlite-vec` (or `sqlite-vss`) MUST be added to `pyproject.toml` and installed BEFORE Wave 1 dispatches. The migration in T88 expects the extension to be loadable.
 - **Embedding model choice**: pin in T91 spec before dispatch. The 384 dim is hardcoded in T88's migration; if a different dim is used, update T88 first.
 - **After each parallel wave**, run a code-review subagent. Combined spec+quality acceptable for trivial tasks (T90 carry-overs); separate spec + quality reviewers for vector-retrieval and integration tasks (T91, T96, T97, T98, T101) — surface area is larger.
 - **Don't dispatch Wave 5 until Wave 4 merged green.** T97 (memory_write enqueue) calls into the embedding-aware worker; the worker uses T91's `generate_embedding`. Both must be merged into `phase-4` first.
 - **Don't dispatch Wave 6 until Wave 5 merged green.** T98 (drawer) wires UI affordances over services from earlier waves.
 - **Token-spend rough estimate**: Phase 4 should be ~70-80% the size of Phase 3 (similar scope, larger per-task because vector + branching are non-trivial). Per-task spend similar to Phase 3's larger tasks (T59, T64).
 - **DO NOT break existing v1/v2/v3/v3.5 surface contracts.** Every test file that was green at the start of Phase 4 must stay green at the end. The cross-feature integration tests from Phase 3 (`tests/test_phase3_integration.py`) are particularly load-bearing.
@@ -0,0 +1,22 @@
 {
  "planPath": "docs/plans/2026-04-27-v4-phase4-implementation.md",
  "tasks": [
    {"id": 88, "subject": "T88: embeddings table + projector handlers (sqlite-vec)", "status": "pending", "wave": 1, "parallelGroup": "wave-1"},
    {"id": 89, "subject": "T89: branches table + projector handlers", "status": "pending", "wave": 1, "parallelGroup": "wave-1"},
    {"id": 90, "subject": "T90: phase 3.6 carry-overs (chat-id pushdown + lifecycle wording + legacy fn consolidation)", "status": "pending", "wave": 1, "parallelGroup": "wave-1"},
    {"id": 91, "subject": "T91: embedding generation service", "status": "pending", "wave": 2, "parallelGroup": "wave-2", "blockedBy": [88]},
    {"id": 92, "subject": "T92: vector search service via sqlite-vec", "status": "pending", "wave": 2, "parallelGroup": "wave-2", "blockedBy": [88]},
    {"id": 93, "subject": "T93: cross-chat search service (FTS5 over all owners)", "status": "pending", "wave": 2, "parallelGroup": "wave-2"},
    {"id": 94, "subject": "T94: branch_from_event service", "status": "pending", "wave": 3, "parallelGroup": "wave-3", "blockedBy": [89]},
    {"id": 95, "subject": "T95: delete-impact computation service", "status": "pending", "wave": 3, "parallelGroup": "wave-3"},
    {"id": 96, "subject": "T96: combined FTS + vector retrieval ranking in search_memories", "status": "pending", "wave": 4, "parallelGroup": null, "blockedBy": [91, 92]},
    {"id": 97, "subject": "T97: memory_write enqueues embedding job + backfill script", "status": "pending", "wave": 5, "parallelGroup": null, "blockedBy": [91, 96]},
    {"id": 98, "subject": "T98: drawer Phase 4 bundle (branching + sig review + hide + surgical delete + remaining edits)", "status": "pending", "wave": 6, "parallelGroup": null, "blockedBy": [94, 95, 97]},
    {"id": 99, "subject": "T99: snapshot UX (manual trigger + list + restore + preview)", "status": "pending", "wave": 7, "parallelGroup": "wave-7"},
    {"id": 100, "subject": "T100: cross-chat search UX (top-bar + results page)", "status": "pending", "wave": 7, "parallelGroup": "wave-7", "blockedBy": [93]},
    {"id": 101, "subject": "T101: cross-feature integration tests (vector × branching × delete × snapshot × search)", "status": "pending", "wave": 8, "parallelGroup": "wave-8", "blockedBy": [98, 99, 100]},
    {"id": 102, "subject": "T102: Phase 4 documentation update", "status": "pending", "wave": 8, "parallelGroup": "wave-8", "blockedBy": [98, 99, 100]}
  ],
  "lastUpdated": "2026-04-27T00:00:00Z",
  "notes": "15 tasks across 8 waves. Adds vector retrieval (sqlite-vec), branching UI, drawer-edit on every field, backup tooling, significance review UI, surgical delete with cascade preview, hide-from-view, and cross-chat search. Phase 3.6 carry-overs (3 small fixes) bundled into T90. External dependency: sqlite-vec must be installed BEFORE Wave 1 dispatch. Embedding model choice (default: 384-dim small model) pinned in T91 spec before dispatch — schema 0012 hardcodes 384 dim. Two new schema migrations (0012 embeddings, 0013 branches), final schema version 13. Uses task ids T88-T102."
 }
@@ -0,0 +1,141 @@
 from __future__ import annotations
 from chat.db.connection import open_db
 from chat.db.migrate import apply_migrations
 from chat.eventlog.log import append_event
 from chat.eventlog.projector import project
 import chat.state.branches  # registers handlers
 from chat.state.branches import active_branch, get_branch, list_branches
 def test_main_branch_bootstrapped_by_migration(tmp_path):
    db = tmp_path / "t.db"
    apply_migrations(db)
    with open_db(db) as conn:
        active = active_branch(conn)
        assert active is not None
        assert active["name"] == "main"
        assert active["is_active"] is True
        assert active["origin_event_id"] == 0
        assert active["head_event_id"] == 0
 def test_branch_created_inserts_row(tmp_path):
    db = tmp_path / "t.db"
    apply_migrations(db)
    with open_db(db) as conn:
        append_event(
            conn,
            kind="branch_created",
            payload={
                "name": "experiment",
                "origin_event_id": 42,
                "chat_id": "chat_a",
            },
        )
        project(conn)
        b = get_branch(conn, "experiment")
        assert b is not None
        assert b["name"] == "experiment"
        assert b["origin_event_id"] == 42
        # head defaults to origin when not specified
        assert b["head_event_id"] == 42
        assert b["chat_id"] == "chat_a"
        assert b["is_active"] is False
        # main remains active
        active = active_branch(conn)
        assert active is not None
        assert active["name"] == "main"
 def test_branch_switched_atomic(tmp_path):
    db = tmp_path / "t.db"
    apply_migrations(db)
    with open_db(db) as conn:
        append_event(
            conn,
            kind="branch_created",
            payload={
                "name": "experiment",
                "origin_event_id": 5,
                "chat_id": "chat_a",
            },
        )
        append_event(
            conn,
            kind="branch_switched",
            payload={"name": "experiment"},
        )
        project(conn)
        active = active_branch(conn)
        assert active is not None
        assert active["name"] == "experiment"
        main = get_branch(conn, "main")
        assert main is not None
        assert main["is_active"] is False
        # switch back
        append_event(
            conn,
            kind="branch_switched",
            payload={"name": "main"},
        )
        project(conn)
        active2 = active_branch(conn)
        assert active2 is not None
        assert active2["name"] == "main"
        experiment = get_branch(conn, "experiment")
        assert experiment is not None
        assert experiment["is_active"] is False
 def test_branch_head_updated_changes_head(tmp_path):
    db = tmp_path / "t.db"
    apply_migrations(db)
    with open_db(db) as conn:
        append_event(
            conn,
            kind="branch_created",
            payload={
                "name": "experiment",
                "origin_event_id": 10,
                "head_event_id": 10,
                "chat_id": "chat_a",
            },
        )
        append_event(
            conn,
            kind="branch_head_updated",
            payload={"name": "experiment", "head_event_id": 20},
        )
        project(conn)
        b = get_branch(conn, "experiment")
        assert b is not None
        assert b["head_event_id"] == 20
 def test_list_branches_returns_all(tmp_path):
    db = tmp_path / "t.db"
    apply_migrations(db)
    with open_db(db) as conn:
        append_event(
            conn,
            kind="branch_created",
            payload={
                "name": "experiment",
                "origin_event_id": 1,
                "chat_id": "chat_a",
            },
        )
        project(conn)
        names = [b["name"] for b in list_branches(conn)]
        assert "main" in names
        assert "experiment" in names
@@ -0,0 +1,131 @@
 """Tests for the branching service (T94, Phase 4)."""
 from __future__ import annotations
 import pytest
 from chat.db.connection import open_db
 from chat.db.migrate import apply_migrations
 from chat.eventlog.log import append_and_apply
 import chat.state.branches  # noqa: F401  registers handlers
 from chat.services.branching import (
    branch_from_event,
    list_branches_with_metadata,
    switch_active_branch,
 )
 from chat.state.branches import active_branch, get_branch
 def _seed_event(conn) -> int:
    """Append a benign event so we have a real event_log row to fork from.
    ``user_turn`` is a transcript-only kind with no registered projector
    handler, so ``append_and_apply`` is a clean no-op on the projector
    side regardless of what other handlers are imported by the suite.
    """
    return append_and_apply(
        conn,
        kind="user_turn",
        payload={"chat_id": "c1", "text": "hi"},
    )
 def test_branch_from_event_creates_branch_via_event(tmp_path):
    db = tmp_path / "t.db"
    apply_migrations(db)
    with open_db(db) as conn:
        seed_id = _seed_event(conn)
        new_id = branch_from_event(
            conn,
            name="experiment",
            origin_event_id=seed_id,
            chat_id="c1",
        )
        assert isinstance(new_id, int) and new_id > 0
        b = get_branch(conn, "experiment")
        assert b is not None
        assert b["id"] == new_id
        assert b["origin_event_id"] == seed_id
        assert b["head_event_id"] == seed_id
        assert b["chat_id"] == "c1"
        assert b["is_active"] is False
        # branch_created event landed in event_log
        row = conn.execute(
            "SELECT COUNT(*) FROM event_log WHERE kind = 'branch_created'"
        ).fetchone()
        assert row[0] == 1
 def test_branch_from_event_duplicate_name_raises(tmp_path):
    db = tmp_path / "t.db"
    apply_migrations(db)
    with open_db(db) as conn:
        seed_id = _seed_event(conn)
        branch_from_event(conn, name="dup", origin_event_id=seed_id)
        with pytest.raises(ValueError, match="already exists"):
            branch_from_event(conn, name="dup", origin_event_id=seed_id)
 def test_branch_from_event_invalid_origin_raises(tmp_path):
    db = tmp_path / "t.db"
    apply_migrations(db)
    with open_db(db) as conn:
        with pytest.raises(ValueError, match="does not exist"):
            branch_from_event(conn, name="ghost", origin_event_id=99999)
 def test_switch_active_branch_changes_active(tmp_path):
    db = tmp_path / "t.db"
    apply_migrations(db)
    with open_db(db) as conn:
        seed_id = _seed_event(conn)
        branch_from_event(conn, name="experiment", origin_event_id=seed_id)
        switch_active_branch(conn, name="experiment")
        active = active_branch(conn)
        assert active is not None
        assert active["name"] == "experiment"
        # Switch back to main.
        switch_active_branch(conn, name="main")
        active2 = active_branch(conn)
        assert active2 is not None
        assert active2["name"] == "main"
 def test_switch_active_branch_unknown_name_raises(tmp_path):
    db = tmp_path / "t.db"
    apply_migrations(db)
    with open_db(db) as conn:
        with pytest.raises(ValueError, match="does not exist"):
            switch_active_branch(conn, name="nope")
 def test_list_branches_with_metadata_includes_event_count(tmp_path):
    db = tmp_path / "t.db"
    apply_migrations(db)
    with open_db(db) as conn:
        # Seed enough events to cover origin=10 and head=15.
        for _ in range(15):
            _seed_event(conn)
        # Create the branch at origin=10, then bump its head to 15.
        branch_from_event(conn, name="exp", origin_event_id=10)
        append_and_apply(
            conn,
            kind="branch_head_updated",
            payload={"name": "exp", "head_event_id": 15},
        )
        rows = {b["name"]: b for b in list_branches_with_metadata(conn)}
        # main: bootstrap state — origin=0, head=0 — event_count == 0.
        assert rows["main"]["event_count"] == 0
        # exp: origin=10, head=15 — event_count == 6 (inclusive).
        assert rows["exp"]["origin_event_id"] == 10
        assert rows["exp"]["head_event_id"] == 15
        assert rows["exp"]["event_count"] == 6
@@ -0,0 +1,155 @@
 """T93 (Phase 4): cross-chat FTS5 search across all owners and chats.
 Verifies that ``chat.services.cross_chat_search.search_all_memories``:
 * surfaces matches across multiple owner_ids (the per-owner restriction
  used by ``state.memory.search_memories`` is intentionally absent),
 * applies no witness filter (admin/power-user surface),
 * orders results by FTS5 BM25 rank (lower = stronger match, surfaced
  first), and
 * honours the ``k`` LIMIT and the empty-query fast-path.
 """
 from __future__ import annotations
 from chat.db.connection import open_db
 from chat.db.migrate import apply_migrations
 from chat.eventlog.log import append_event
 from chat.eventlog.projector import project
 from chat.services.cross_chat_search import search_all_memories
 import chat.state.memory  # noqa: F401  (registers memory_written handler)
 def _seed(db, *, memory_specs):
    """Apply migrations + project a list of memory_written events."""
    apply_migrations(db)
    with open_db(db) as conn:
        for spec in memory_specs:
            payload = {
                "owner_id": spec.get("owner_id", "bot_a"),
                "chat_id": spec.get("chat_id", "chat_bot_a"),
                "pov_summary": spec["pov_summary"],
                "witness_you": spec.get("witness_you", 1),
                "witness_host": spec.get("witness_host", 1),
                "witness_guest": spec.get("witness_guest", 0),
                "source": "direct",
                "reliability": 1.0,
                "significance": spec.get("significance", 1),
                "pinned": 0,
                "auto_pinned": 0,
            }
            append_event(conn, kind="memory_written", payload=payload)
        project(conn)
 def test_search_all_memories_returns_matches_across_owners(tmp_path):
    """Cross-owner: a single query must surface memories from every owner.
    The per-owner ``owner_id = ?`` predicate that ``search_memories`` uses
    is intentionally absent here, so a "rabbit" memory under ``bot_a`` and
    one under ``bot_b`` should both come back from a single call.
    """
    db = tmp_path / "t.db"
    _seed(
        db,
        memory_specs=[
            {
                "owner_id": "bot_a",
                "chat_id": "chat_bot_a",
                "pov_summary": "the rabbit darted into the brambles",
            },
            {
                "owner_id": "bot_b",
                "chat_id": "chat_bot_b",
                "pov_summary": "a white rabbit watched from the hedge",
            },
            # Distractor: must not appear for "rabbit".
            {
                "owner_id": "bot_a",
                "chat_id": "chat_bot_a",
                "pov_summary": "the kettle whistled",
            },
        ],
    )
    with open_db(db) as conn:
        out = search_all_memories(conn, query="rabbit")
        owners = {row["owner_id"] for row in out}
        assert owners == {"bot_a", "bot_b"}
        assert len(out) == 2
        # Returned shape contract.
        for row in out:
            assert set(row.keys()) >= {
                "memory_id",
                "owner_id",
                "chat_id",
                "scene_id",
                "pov_summary",
                "significance",
                "ts",
                "fts_rank",
            }
 def test_search_all_memories_orders_by_fts_rank(tmp_path):
    """Stronger BM25 match must come first (rank ASC = lower is better)."""
    db = tmp_path / "t.db"
    _seed(
        db,
        memory_specs=[
            # Single occurrence -> weaker BM25 score.
            {
                "owner_id": "bot_a",
                "chat_id": "chat_bot_a",
                "pov_summary": "a rabbit appeared",
            },
            # Triple occurrence in a short row -> stronger BM25 score.
            {
                "owner_id": "bot_b",
                "chat_id": "chat_bot_b",
                "pov_summary": "rabbit rabbit rabbit",
            },
        ],
    )
    with open_db(db) as conn:
        out = search_all_memories(conn, query="rabbit", k=5)
        assert len(out) == 2
        # Stronger match first; fts_rank monotonically non-decreasing
        # (lower-is-better, so ASC).
        assert out[0]["pov_summary"] == "rabbit rabbit rabbit"
        assert out[0]["fts_rank"] <= out[1]["fts_rank"]
 def test_search_all_memories_respects_k_limit(tmp_path):
    """LIMIT ? must cap result count even when more matches exist."""
    db = tmp_path / "t.db"
    _seed(
        db,
        memory_specs=[
            {
                "owner_id": f"bot_{i}",
                "chat_id": f"chat_{i}",
                "pov_summary": f"rabbit sighting number {i}",
            }
            for i in range(10)
        ],
    )
    with open_db(db) as conn:
        out = search_all_memories(conn, query="rabbit", k=3)
        assert len(out) == 3
 def test_search_all_memories_empty_query_returns_empty(tmp_path):
    """Empty / whitespace-only query must short-circuit to []."""
    db = tmp_path / "t.db"
    _seed(
        db,
        memory_specs=[
            {
                "owner_id": "bot_a",
                "chat_id": "chat_bot_a",
                "pov_summary": "the rabbit darted into the brambles",
            },
        ],
    )
    with open_db(db) as conn:
        assert search_all_memories(conn, query="") == []
        assert search_all_memories(conn, query="   ") == []
@@ -0,0 +1,248 @@
 """Tests for Task 95 — delete-impact computation service (Phase 4).
 `compute_delete_impact` walks event_log forward from a target event_id and
 produces an :class:`ImpactReport` describing what would be removed if
 rewind-to-target were invoked. It is a pure preview — no database mutation.
 T98's drawer surgical-delete UI uses this to render an "are you sure?"
 modal before invoking the actual rewind path.
 """
 from __future__ import annotations
 from chat.db.connection import open_db
 from chat.db.migrate import apply_migrations
 from chat.eventlog.log import append_event
 from chat.services.delete_impact import compute_delete_impact
 def _seed_chat(conn) -> tuple[int, int]:
    """Append minimal bot + chat events; return their event ids."""
    bot_id = append_event(
        conn,
        kind="bot_authored",
        payload={
            "id": "bot_a",
            "name": "BotA",
            "persona": "...",
            "voice_samples": [],
            "traits": [],
            "backstory": "",
            "initial_relationship_to_you": "",
            "kickoff_prose": "",
        },
    )
    chat_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": "",
        },
    )
    return bot_id, chat_id
 def test_impact_for_simple_turn_lists_memory_and_edges(tmp_path):
    db = tmp_path / "t.db"
    apply_migrations(db)
    with open_db(db) as conn:
        _seed_chat(conn)
        user_id = append_event(
            conn,
            kind="user_turn",
            payload={
                "chat_id": "chat_bot_a",
                "prose": "hey there friend",
                "segments": [],
            },
        )
        append_event(
            conn,
            kind="assistant_turn",
            payload={
                "chat_id": "chat_bot_a",
                "speaker_id": "bot_a",
                "text": "Hi! Good to see you.",
                "truncated": False,
                "user_turn_id": user_id,
            },
        )
        append_event(
            conn,
            kind="memory_written",
            payload={
                "owner_id": "bot_a",
                "chat_id": "chat_bot_a",
                "pov_summary": "You greeted me warmly today.",
                "witness_you": 1,
                "witness_host": 1,
                "witness_guest": 0,
                "source": "turn",
                "reliability": 1.0,
                "significance": 1,
                "pinned": 0,
                "auto_pinned": 0,
            },
        )
        append_event(
            conn,
            kind="edge_update",
            payload={
                "source_id": "you",
                "target_id": "bot_a",
                "affinity_delta": 0.1,
            },
        )
        report = compute_delete_impact(conn, target_event_id=user_id)
    assert report.target_event_id == user_id
    kinds = [item.kind for item in report.cascading]
    # Walk from user_turn forward — user_turn, assistant_turn,
    # memory_written, edge_update should all be in scope, in order.
    assert kinds == [
        "user_turn",
        "assistant_turn",
        "memory_written",
        "edge_update",
    ]
    # Memory description includes the pov_summary excerpt.
    mem_item = report.cascading[2]
    assert "memory:" in mem_item.description
    assert "greeted" in mem_item.description
    # Edge description includes both endpoints.
    edge_item = report.cascading[3]
    assert "you" in edge_item.description
    assert "bot_a" in edge_item.description
    assert edge_item.target_id == "you->bot_a"
    # Notes mentions total count.
    assert any("4 events" in n for n in report.notes)
 def test_impact_for_scene_opening_turn_warns_about_subsequent(tmp_path):
    db = tmp_path / "t.db"
    apply_migrations(db)
    with open_db(db) as conn:
        _seed_chat(conn)
        early_id = append_event(
            conn,
            kind="user_turn",
            payload={"chat_id": "chat_bot_a", "prose": "the start", "segments": []},
        )
        append_event(
            conn,
            kind="assistant_turn",
            payload={
                "chat_id": "chat_bot_a",
                "speaker_id": "bot_a",
                "text": "ok",
                "truncated": False,
                "user_turn_id": early_id,
            },
        )
        append_event(
            conn,
            kind="scene_closed",
            payload={
                "scene_id": 1,
                "closed_at": "2026-04-26T21:00:00+00:00",
                "significance": 2,
            },
        )
        report = compute_delete_impact(conn, target_event_id=early_id)
    # Scene-close warning fires when one is in scope.
    assert any("scene close" in n.lower() for n in report.notes)
    # The scene_closed event also appears as a cascading item.
    assert any(item.kind == "scene_closed" for item in report.cascading)
 def test_impact_for_missing_event_returns_empty_with_note(tmp_path):
    db = tmp_path / "t.db"
    apply_migrations(db)
    with open_db(db) as conn:
        _seed_chat(conn)
        report = compute_delete_impact(conn, target_event_id=999_999)
    assert report.cascading == []
    assert any("not found" in n for n in report.notes)
 def test_impact_does_not_mutate_database(tmp_path):
    db = tmp_path / "t.db"
    apply_migrations(db)
    with open_db(db) as conn:
        _seed_chat(conn)
        user_id = append_event(
            conn,
            kind="user_turn",
            payload={"chat_id": "chat_bot_a", "prose": "hi", "segments": []},
        )
        append_event(
            conn,
            kind="assistant_turn",
            payload={
                "chat_id": "chat_bot_a",
                "speaker_id": "bot_a",
                "text": "hello",
                "truncated": False,
                "user_turn_id": user_id,
            },
        )
        # Snapshot all event_log rows as a tuple-of-tuples.
        before = conn.execute(
            "SELECT id, branch_id, ts, kind, payload_json, superseded_by, "
            "hidden FROM event_log ORDER BY id"
        ).fetchall()
        compute_delete_impact(conn, target_event_id=user_id)
        after = conn.execute(
            "SELECT id, branch_id, ts, kind, payload_json, superseded_by, "
            "hidden FROM event_log ORDER BY id"
        ).fetchall()
    # Byte-identical: nothing inserted, deleted, or updated.
    assert before == after
 def test_impact_includes_regenerated_from_warning(tmp_path):
    db = tmp_path / "t.db"
    apply_migrations(db)
    with open_db(db) as conn:
        _seed_chat(conn)
        original_id = append_event(
            conn,
            kind="assistant_turn",
            payload={
                "chat_id": "chat_bot_a",
                "speaker_id": "bot_a",
                "text": "first try",
                "truncated": False,
                "user_turn_id": 0,
            },
        )
        regen_id = append_event(
            conn,
            kind="assistant_turn",
            payload={
                "chat_id": "chat_bot_a",
                "speaker_id": "bot_a",
                "text": "second try",
                "truncated": False,
                "user_turn_id": 0,
                "regenerated_from": original_id,
            },
        )
        report = compute_delete_impact(conn, target_event_id=regen_id)
    # The regenerated_from note carries the original event id so the user
    # knows the original turn isn't lost.
    assert any("regenerated from" in n for n in report.notes)
    assert any(str(original_id) in n for n in report.notes)
@@ -0,0 +1,91 @@
 """Tests for the embedding generation service (T91, Phase 4).
 Phase 4's first cut ships a deterministic local pseudo-embedding so the
 vector retrieval pipeline can land without an external embeddings API
 or a heavy local model dependency. These tests pin the contract:
 * the result has the right shape (vector length, ``dim`` metadata),
 * the default ``model`` string is reported back unchanged,
 * output is byte-identical for the same input (deterministic),
 * distinct inputs produce distinct vectors (so cosine actually
  discriminates),
 * empty / whitespace-only input collapses to the ``"fallback"`` sentinel
  with a zero vector — callers detect this and skip indexing,
 * the vector is unit-normalized so cosine similarity behaves.
 The pseudo path doesn't touch the LLMClient, so we pass an empty
 ``MockLLMClient`` — any accidental call into it would raise
 ``IndexError`` and surface as a regression.
 """
 from __future__ import annotations
 import math
 import pytest
 from chat.llm.mock import MockLLMClient
 from chat.services.embeddings import (
    DEFAULT_EMBEDDING_DIM,
    DEFAULT_EMBEDDING_MODEL,
    FALLBACK_EMBEDDING_MODEL,
    EmbeddingResult,
    generate_embedding,
 )
 def _client() -> MockLLMClient:
    # Pseudo path never calls the client — empty canned list ensures any
    # accidental call raises and surfaces the regression loudly.
    return MockLLMClient(canned=[])
@pytest.mark.asyncio
 async def test_generate_embedding_returns_vector_of_correct_dim():
    result = await generate_embedding(_client(), text="hello")
    assert isinstance(result, EmbeddingResult)
    assert isinstance(result.vector, list)
    assert len(result.vector) == DEFAULT_EMBEDDING_DIM == 384
    assert result.dim == 384
    assert all(isinstance(x, float) for x in result.vector)
@pytest.mark.asyncio
 async def test_generate_embedding_returns_correct_model_metadata():
    result = await generate_embedding(_client(), text="hello")
    assert result.model == DEFAULT_EMBEDDING_MODEL == "pseudo-sha256-384"
@pytest.mark.asyncio
 async def test_generate_embedding_is_deterministic():
    a = await generate_embedding(_client(), text="hello world")
    b = await generate_embedding(_client(), text="hello world")
    assert a.vector == b.vector
@pytest.mark.asyncio
 async def test_generate_embedding_distinct_text_produces_distinct_vectors():
    a = await generate_embedding(_client(), text="hello world")
    b = await generate_embedding(_client(), text="totally different content")
    assert a.vector != b.vector
    # Sanity-check cosine similarity — both vectors are unit-normalized,
    # so this reduces to a plain dot product.
    cosine = sum(x * y for x, y in zip(a.vector, b.vector))
    assert cosine < 0.99
@pytest.mark.asyncio
 async def test_generate_embedding_empty_text_returns_fallback():
    for empty in ("", "   ", "\n\t"):
        result = await generate_embedding(_client(), text=empty)
        assert result.model == FALLBACK_EMBEDDING_MODEL == "fallback"
        assert result.dim == DEFAULT_EMBEDDING_DIM
        assert len(result.vector) == DEFAULT_EMBEDDING_DIM
        assert all(x == 0.0 for x in result.vector)
@pytest.mark.asyncio
 async def test_generate_embedding_unit_normalized():
    result = await generate_embedding(_client(), text="some non-empty text")
    norm_sq = sum(x * x for x in result.vector)
    assert math.isclose(norm_sq, 1.0, abs_tol=1e-6)
@@ -0,0 +1,218 @@
 from __future__ import annotations
 from chat.db.connection import open_db
 from chat.db.migrate import apply_migrations
 from chat.eventlog.log import append_event
 from chat.eventlog.projector import project
 import chat.state.memory  # registers memory_written handler
 import chat.state.embeddings  # registers embedding handlers
 from chat.state.embeddings import get_embedding, list_embeddings_for_owner
 def _base_memory(**overrides):
    payload = {
        "owner_id": "bot_a",
        "chat_id": "chat_bot_a",
        "scene_id": 1,
        "pov_summary": "She laughed at his joke about owls.",
        "witness_you": 1,
        "witness_host": 1,
        "witness_guest": 0,
        "chat_clock_at": "2026-04-26T10:00:00",
        "source": "direct",
        "reliability": 1.0,
        "significance": 1,
        "pinned": 0,
        "auto_pinned": 0,
    }
    payload.update(overrides)
    return payload
 def _vec(n: int = 384, base: float = 0.1) -> list[float]:
    """Return a length-n float vector with predictable values for assertions."""
    return [round(base + i * 0.001, 6) for i in range(n)]
 def test_embedding_indexed_inserts_row(tmp_path):
    db = tmp_path / "t.db"
    apply_migrations(db)
    with open_db(db) as conn:
        append_event(conn, kind="memory_written", payload=_base_memory())
        project(conn)
        memory_id = conn.execute("SELECT id FROM memories").fetchone()[0]
        vector = _vec(384, base=0.1)
        append_event(
            conn,
            kind="embedding_indexed",
            payload={
                "memory_id": memory_id,
                "vector": vector,
                "model": "test-model",
                "dim": 384,
            },
        )
        project(conn)
        emb = get_embedding(conn, memory_id)
        assert emb is not None
        assert emb["memory_id"] == memory_id
        assert emb["vector"] == vector
        assert emb["model"] == "test-model"
        assert emb["dim"] == 384
        assert emb["indexed_at"] is not None
 def test_embedding_deindexed_removes_row(tmp_path):
    db = tmp_path / "t.db"
    apply_migrations(db)
    with open_db(db) as conn:
        append_event(conn, kind="memory_written", payload=_base_memory())
        project(conn)
        memory_id = conn.execute("SELECT id FROM memories").fetchone()[0]
        append_event(
            conn,
            kind="embedding_indexed",
            payload={
                "memory_id": memory_id,
                "vector": _vec(),
                "model": "test-model",
                "dim": 384,
            },
        )
        project(conn)
        assert get_embedding(conn, memory_id) is not None
        append_event(
            conn,
            kind="embedding_deindexed",
            payload={"memory_id": memory_id},
        )
        project(conn)
        assert get_embedding(conn, memory_id) is None
 def test_embedding_indexed_replaces_existing(tmp_path):
    db = tmp_path / "t.db"
    apply_migrations(db)
    with open_db(db) as conn:
        append_event(conn, kind="memory_written", payload=_base_memory())
        project(conn)
        memory_id = conn.execute("SELECT id FROM memories").fetchone()[0]
        vec_a = _vec(384, base=0.1)
        vec_b = _vec(384, base=0.5)
        append_event(
            conn,
            kind="embedding_indexed",
            payload={
                "memory_id": memory_id,
                "vector": vec_a,
                "model": "test-model",
                "dim": 384,
            },
        )
        project(conn)
        first = get_embedding(conn, memory_id)
        assert first is not None
        assert first["vector"] == vec_a
        append_event(
            conn,
            kind="embedding_indexed",
            payload={
                "memory_id": memory_id,
                "vector": vec_b,
                "model": "test-model",
                "dim": 384,
            },
        )
        project(conn)
        second = get_embedding(conn, memory_id)
        assert second is not None
        assert second["vector"] == vec_b
        # Still exactly one row for this memory.
        count = conn.execute(
            "SELECT COUNT(*) FROM embeddings WHERE memory_id = ?", (memory_id,)
        ).fetchone()[0]
        assert count == 1
 def test_list_embeddings_for_owner_returns_joined_rows(tmp_path):
    db = tmp_path / "t.db"
    apply_migrations(db)
    with open_db(db) as conn:
        # Two memories for bot_a, one for bot_b.
        append_event(
            conn,
            kind="memory_written",
            payload=_base_memory(
                owner_id="bot_a",
                pov_summary="Alpha memory.",
                significance=2,
            ),
        )
        append_event(
            conn,
            kind="memory_written",
            payload=_base_memory(
                owner_id="bot_a",
                pov_summary="Beta memory.",
                significance=3,
            ),
        )
        append_event(
            conn,
            kind="memory_written",
            payload=_base_memory(
                owner_id="bot_b",
                pov_summary="Gamma memory.",
                significance=1,
            ),
        )
        project(conn)
        rows = conn.execute(
            "SELECT id, owner_id FROM memories ORDER BY id"
        ).fetchall()
        # Index every memory with a distinct vector so we can check ordering.
        for i, (mid, _owner) in enumerate(rows):
            append_event(
                conn,
                kind="embedding_indexed",
                payload={
                    "memory_id": mid,
                    "vector": _vec(384, base=0.1 * (i + 1)),
                    "model": "test-model",
                    "dim": 384,
                },
            )
        project(conn)
        a_rows = list_embeddings_for_owner(conn, "bot_a")
        assert len(a_rows) == 2
        summaries = {r["pov_summary"] for r in a_rows}
        assert summaries == {"Alpha memory.", "Beta memory."}
        sigs = {r["significance"] for r in a_rows}
        assert sigs == {2, 3}
        for r in a_rows:
            assert r["model"] == "test-model"
            assert r["dim"] == 384
            assert isinstance(r["vector"], list)
            assert len(r["vector"]) == 384
            assert r["witness_you"] == 1
            assert r["witness_host"] == 1
            assert r["witness_guest"] == 0
        b_rows = list_embeddings_for_owner(conn, "bot_b")
        assert len(b_rows) == 1
        assert b_rows[0]["pov_summary"] == "Gamma memory."
 def test_get_embedding_returns_none_when_missing(tmp_path):
    db = tmp_path / "t.db"
    apply_migrations(db)
    with open_db(db) as conn:
        assert get_embedding(conn, 999) is None
@@ -16,6 +16,7 @@ from chat.eventlog.log import append_event
 from chat.eventlog.projector import project
 from chat.state.memory import search_memories
 import chat.state.memory  # noqa: F401  (registers memory_written handler)
 import chat.state.embeddings  # noqa: F401  (registers embedding_indexed handler)
 def _seed(db, *, memory_specs):
@@ -159,3 +160,216 @@ def test_significance_bias_is_constant_module_level():
    # Must be non-negative -- a negative bias would invert the desired
    # "higher significance ranks higher" semantics.
    assert SIGNIFICANCE_RANK_BIAS >= 0
 # ---------------------------------------------------------------------------
 # T96 (Phase 4): combined FTS + vector retrieval ranking via reciprocal-rank
 # fusion. The fused path activates only when ``query_vector`` is provided —
 # the no-vector path (above) is unchanged.
 # ---------------------------------------------------------------------------
 def _one_hot(dim: int, idx: int) -> list[float]:
    v = [0.0] * dim
    v[idx] = 1.0
    return v
 def _seed_memories_with_optional_embeddings(db, *, memory_specs):
    """Like ``_seed`` but also projects ``embedding_indexed`` events for any
    spec carrying a ``vector`` key.
    Memory rows are assigned ids in the order their ``memory_written`` events
    were appended (the ``memories.id`` column is an autoincrementing primary
    key), so we predict ``memory_id = i + 1`` per spec and append both kinds
    of events back-to-back BEFORE projecting. Projecting only once keeps the
    INSERT-based ``memory_written`` handler from duplicating rows.
    """
    apply_migrations(db)
    with open_db(db) as conn:
        # First pass: append every memory_written event in order. The DB
        # assigns autoincrementing ids 1..N matching the order of these
        # events, so we can pair vectors to memories by index.
        for spec in memory_specs:
            payload = {
                "owner_id": spec.get("owner_id", "bot_a"),
                "chat_id": spec.get("chat_id", "chat_bot_a"),
                "pov_summary": spec["pov_summary"],
                "witness_you": spec.get("witness_you", 1),
                "witness_host": spec.get("witness_host", 1),
                "witness_guest": spec.get("witness_guest", 0),
                "source": "direct",
                "reliability": 1.0,
                "significance": spec.get("significance", 1),
                "pinned": 0,
                "auto_pinned": 0,
            }
            append_event(conn, kind="memory_written", payload=payload)
        # Second pass: append embedding_indexed events for any spec that
        # supplied a vector, using the predicted memory id.
        for i, spec in enumerate(memory_specs, start=1):
            if "vector" not in spec:
                continue
            vec = spec["vector"]
            append_event(
                conn,
                kind="embedding_indexed",
                payload={
                    "memory_id": i,
                    "vector": list(vec),
                    "model": "test-model",
                    "dim": len(vec),
                },
            )
        # Single projection — avoids the memory_written handler INSERTing
        # the same row twice on a re-projection.
        project(conn)
 def test_search_memories_without_query_vector_uses_fts_only(tmp_path):
    """Regression: omitting ``query_vector`` keeps the existing FTS-only path.
    Identical seed to ``test_search_higher_significance_ranks_above_lower``
    but pinned to the no-vector code path explicitly (no kwarg passed).
    """
    db = tmp_path / "t.db"
    _seed(
        db,
        memory_specs=[
            {"pov_summary": "small promise"},
            {"pov_summary": "huge promise"},
            {"pov_summary": "tiny promise", "significance": 3},
        ],
    )
    with open_db(db) as conn:
        out = search_memories(conn, "bot_a", "host", "promise", k=3)
        assert len(out) == 3
        # The composite re-rank surfaces the high-significance row first.
        assert out[0]["pov_summary"] == "tiny promise"
        # Sanity: the row shape still carries ``fts_rank`` + ``composite_score``
        # like the FTS-only path always has.
        assert "fts_rank" in out[0]
        assert "composite_score" in out[0]
 def test_search_memories_with_query_vector_includes_vector_hits(tmp_path):
    """RRF fuses FTS hits with vector hits — both kinds surface in the result.
    Memory 1 only matches FTS (keyword "rabbit", embedding far from query).
    Memory 2 only matches the vector (embedding identical to query, no
    keyword overlap). Memories 3-5 are unrelated. The fused top-K must
    contain BOTH memory 1 and memory 2.
    """
    db = tmp_path / "t.db"
    dim = 8
    # Query vector = one-hot at index 0. Memory 2 mirrors it exactly. The
    # FTS-only memory (memory 1) has NO embedding so it cannot leak into
    # the vector ranking; the filler memories (3-5) likewise have no
    # embeddings, so the vector ranking returns memory 2 alone.
    query_vec = _one_hot(dim, 0)
    _seed_memories_with_optional_embeddings(
        db,
        memory_specs=[
            # Memory 1: FTS-only match. No embedding indexed.
            {"pov_summary": "rabbit hopped over the fence"},
            # Memory 2: vector-only match. No keyword overlap with "rabbit".
            {
                "pov_summary": "completely unrelated narrative line",
                "vector": _one_hot(dim, 0),
            },
            # Memories 3-5: filler, irrelevant to both channels.
            {"pov_summary": "lighthouse keeper polished the lens"},
            {"pov_summary": "they discussed cartography for hours"},
            {"pov_summary": "she taught him semaphore signals"},
        ],
    )
    with open_db(db) as conn:
        out = search_memories(
            conn,
            "bot_a",
            "host",
            "rabbit",
            k=4,
            query_vector=query_vec,
        )
        summaries = [r["pov_summary"] for r in out]
        # FTS-only candidate (memory 1) made it through.
        assert "rabbit hopped over the fence" in summaries
        # Vector-only candidate (memory 2) also made it through despite
        # having no keyword overlap with the query string.
        assert "completely unrelated narrative line" in summaries
 def test_search_memories_fusion_significance_bias_still_applies(tmp_path):
    """With two RRF-tied candidates, the higher-significance one ranks first.
    Two memories share the keyword "promise" AND share an identical
    embedding to the query — so their FTS rank and vector rank are both
    ties. RRF gives them the same fusion score. The Python-side
    significance + recency boost must break the tie in favour of the
    higher-significance memory.
    """
    db = tmp_path / "t.db"
    dim = 4
    shared_vec = _one_hot(dim, 0)
    _seed_memories_with_optional_embeddings(
        db,
        memory_specs=[
            {
                "pov_summary": "she made a promise",
                "significance": 0,
                "vector": list(shared_vec),
            },
            {
                "pov_summary": "she made a promise",
                "significance": 3,
                "vector": list(shared_vec),
            },
        ],
    )
    with open_db(db) as conn:
        out = search_memories(
            conn,
            "bot_a",
            "host",
            "promise",
            k=2,
            query_vector=list(shared_vec),
        )
        assert len(out) == 2
        # Higher significance breaks the RRF tie.
        assert out[0]["significance"] == 3
        assert out[1]["significance"] == 0
 def test_search_memories_fusion_handles_empty_vector_results(tmp_path):
    """Vector path returning [] (no embeddings indexed) must not break FTS.
    No ``embedding_indexed`` events are projected, so ``vector_search``
    returns an empty list. The function should still return the FTS hits
    as if ``query_vector`` had not been supplied.
    """
    db = tmp_path / "t.db"
    _seed(
        db,
        memory_specs=[
            {"pov_summary": "the vault held an old promise"},
            {"pov_summary": "another promise was kept that night"},
        ],
    )
    with open_db(db) as conn:
        out = search_memories(
            conn,
            "bot_a",
            "host",
            "promise",
            k=4,
            query_vector=[0.0] * 384,  # No embeddings exist for this owner.
        )
        # Both FTS hits still come back — no error from the empty vector path.
        assert len(out) == 2
        summaries = {r["pov_summary"] for r in out}
        assert summaries == {
            "the vault held an old promise",
            "another promise was kept that night",
        }
@@ -22,7 +22,7 @@ from chat.db.migrate import apply_migrations
 from chat.eventlog.log import append_event
 from chat.eventlog.projector import project
 from chat.llm.mock import MockLLMClient
-from chat.services.memory_write import record_turn_memory, record_turn_memory_for_present
+from chat.services.memory_write import record_turn_memory_for_present
 import chat.state.entities  # noqa: F401  - register handlers
 import chat.state.memory  # noqa: F401
 import chat.state.world  # noqa: F401
@@ -64,14 +64,19 @@ def test_record_turn_memory_writes_event_and_projects(tmp_path):
    apply_migrations(db)
    _seed_minimal(db)
    with open_db(db) as conn:
-        eid, mid = record_turn_memory(
+        # T90.3: legacy ``record_turn_memory`` was removed; the unified
        # ``record_turn_memory_for_present`` with ``guest_bot_id=None``
        # produces the same single-bot witness mask [1,1,0].
        result = record_turn_memory_for_present(
            conn,
            chat_id="chat_bot_a",
            host_bot_id="bot_a",
            guest_bot_id=None,
            narrative_text="BotA looks up. 'You're back late.'",
            scene_id=None,
            chat_clock_at="2026-04-26T20:00:00+00:00",
        )
        eid, mid = result["bot_a"]
        assert eid > 0
        assert mid is not None and mid > 0
@@ -111,12 +116,15 @@ def test_record_turn_memory_omits_optional_fields(tmp_path):
    _seed_minimal(db)
    with open_db(db) as conn:
        # Call without scene_id/chat_clock_at — should default to None.
-        eid, mid = record_turn_memory(
+        # T90.3: migrated from legacy ``record_turn_memory``.
        result = record_turn_memory_for_present(
            conn,
            chat_id="chat_bot_a",
            host_bot_id="bot_a",
            guest_bot_id=None,
            narrative_text="A simple memory.",
        )
        eid, mid = result["bot_a"]
        assert eid > 0
        assert mid is not None and mid > 0
@@ -757,6 +757,13 @@ def test_regenerate_with_prior_lifecycle_logs_warning(tmp_path, monkeypatch, cap
    # row's id.
    assert str(at_id) in msg
    assert str(completed_id) in msg
    # T90.2: wording was tightened from "from superseded turn" to
    # "at-or-after turn <id>" — when regenerating an OLDER turn, the
    # listed transitions may include legitimate intervening-turn ones
    # that stand on their own. The new phrasing avoids implying the
    # warning's target turn directly authored every listed transition.
    assert "at-or-after turn" in msg
    assert "from superseded turn" not in msg
 def test_regenerate_sibling_lookup_scoped_to_chat(tmp_path, monkeypatch):
@@ -186,6 +186,82 @@ def test_read_recent_dialogue_filters_superseded_and_other_chats(tmp_path):
    assert ut_id is not None
 def test_read_recent_dialogue_limit_respects_chat_scope(tmp_path):
    """T90.1: ``read_recent_dialogue`` must push the chat_id filter into
    SQL so that ``LIMIT N`` returns N rows scoped to the requested chat —
    not N globally-recent rows that may then be filtered down to fewer in
    Python.
    Setup: two chats with 60 turns each, interleaved. With the old
    post-fetch filter, ``LIMIT 50`` would pull 50 globally-recent rows
    (most or all from chat_b — the most recent inserts) and then drop
    chat_b ones via the Python check, yielding far fewer than 50 chat_a
    rows. After the SQL pushdown, ``LIMIT 50`` should return exactly 50
    chat_a rows.
    """
    db = tmp_path / "test.db"
    apply_migrations(db)
    with open_db(db) as conn:
        for chat_id, host_bot in (("chat_a", "bot_a"), ("chat_b", "bot_b")):
            append_event(
                conn,
                kind="bot_authored",
                payload={
                    "id": host_bot,
                    "name": host_bot,
                    "persona": "...",
                    "voice_samples": [],
                    "traits": [],
                    "backstory": "",
                    "initial_relationship_to_you": "",
                    "kickoff_prose": "",
                },
            )
            append_event(
                conn,
                kind="chat_created",
                payload={
                    "id": chat_id,
                    "host_bot_id": host_bot,
                    "initial_time": "2026-04-26T20:00:00+00:00",
                    "narrative_anchor": "Day 1",
                    "weather": "",
                },
            )
        # Interleave 60 user_turn rows in each chat — chat_b's go in last
        # so they dominate the global tail.
        for i in range(60):
            append_event(
                conn,
                kind="user_turn",
                payload={
                    "chat_id": "chat_a",
                    "prose": f"a-{i}",
                    "segments": [],
                },
            )
        for i in range(60):
            append_event(
                conn,
                kind="user_turn",
                payload={
                    "chat_id": "chat_b",
                    "prose": f"b-{i}",
                    "segments": [],
                },
            )
        project(conn)
        out = read_recent_dialogue(conn, "chat_a", limit=50)
    # All returned rows should belong to chat_a (texts a-* only).
    assert len(out) == 50
    for entry in out:
        assert entry["text"].startswith("a-"), (
            f"foreign chat row leaked: {entry!r}"
        )
 def test_gather_prior_edges_fills_missing_with_default(tmp_path):
    """``gather_prior_edges`` returns one entry per directed pair across
    ``present_ids``. Missing rows fall back to the schema default
@@ -0,0 +1,242 @@
 from __future__ import annotations
 import pytest
 from chat.db.connection import open_db
 from chat.db.migrate import apply_migrations
 from chat.eventlog.log import append_event
 from chat.eventlog.projector import project
 import chat.state.memory  # registers memory_written handler
 import chat.state.embeddings  # registers embedding handlers
 from chat.services.vector_search import vector_search
 def _base_memory(**overrides):
    payload = {
        "owner_id": "bot_a",
        "chat_id": "chat_bot_a",
        "scene_id": 1,
        "pov_summary": "She laughed at his joke about owls.",
        "witness_you": 1,
        "witness_host": 1,
        "witness_guest": 0,
        "chat_clock_at": "2026-04-26T10:00:00",
        "source": "direct",
        "reliability": 1.0,
        "significance": 1,
        "pinned": 0,
        "auto_pinned": 0,
    }
    payload.update(overrides)
    return payload
 def _one_hot(dim: int, idx: int) -> list[float]:
    """Return a one-hot vector of length ``dim`` with 1.0 at ``idx``."""
    v = [0.0] * dim
    v[idx] = 1.0
    return v
 def _seed_memory_with_embedding(
    conn,
    *,
    owner_id: str,
    pov_summary: str,
    vector: list[float],
    significance: int = 1,
    witness_you: int = 1,
    witness_host: int = 1,
    witness_guest: int = 0,
    model: str = "test-model",
 ) -> int:
    append_event(
        conn,
        kind="memory_written",
        payload=_base_memory(
            owner_id=owner_id,
            pov_summary=pov_summary,
            significance=significance,
            witness_you=witness_you,
            witness_host=witness_host,
            witness_guest=witness_guest,
        ),
    )
    project(conn)
    memory_id = conn.execute(
        "SELECT id FROM memories WHERE pov_summary = ?", (pov_summary,)
    ).fetchone()[0]
    append_event(
        conn,
        kind="embedding_indexed",
        payload={
            "memory_id": memory_id,
            "vector": vector,
            "model": model,
            "dim": len(vector),
        },
    )
    project(conn)
    return memory_id
 def test_vector_search_returns_nearest_neighbors(tmp_path):
    db = tmp_path / "t.db"
    apply_migrations(db)
    with open_db(db) as conn:
        dim = 8
        ids = []
        for i in range(5):
            mid = _seed_memory_with_embedding(
                conn,
                owner_id="bot_a",
                pov_summary=f"Memory {i}.",
                vector=_one_hot(dim, i),
            )
            ids.append(mid)
        # Query close to memory index 3 (one-hot at position 3, plus tiny noise).
        query = _one_hot(dim, 3)
        query[2] = 0.01
        results = vector_search(
            conn,
            owner_id="bot_a",
            witness_role="you",
            query_vector=query,
            k=3,
        )
        assert len(results) == 3
        # Top-1 must be memory at index 3.
        assert results[0]["memory_id"] == ids[3]
        assert results[0]["pov_summary"] == "Memory 3."
        # Score for the near-perfect match should be very close to 1.0.
        assert results[0]["score"] > 0.99
        # Results sorted by score DESC.
        scores = [r["score"] for r in results]
        assert scores == sorted(scores, reverse=True)
        # Second place should be memory index 2 (the small noise component).
        assert results[1]["memory_id"] == ids[2]
 def test_vector_search_respects_witness_filter(tmp_path):
    db = tmp_path / "t.db"
    apply_migrations(db)
    with open_db(db) as conn:
        dim = 4
        # Memory visible to you=1, host=1, guest=0.
        _seed_memory_with_embedding(
            conn,
            owner_id="bot_a",
            pov_summary="Restricted.",
            vector=_one_hot(dim, 0),
            witness_you=1,
            witness_host=1,
            witness_guest=0,
        )
        # Guest sees nothing.
        guest_results = vector_search(
            conn,
            owner_id="bot_a",
            witness_role="guest",
            query_vector=_one_hot(dim, 0),
            k=4,
        )
        assert guest_results == []
        # Host sees the memory.
        host_results = vector_search(
            conn,
            owner_id="bot_a",
            witness_role="host",
            query_vector=_one_hot(dim, 0),
            k=4,
        )
        assert len(host_results) == 1
        assert host_results[0]["pov_summary"] == "Restricted."
        # You also see it.
        you_results = vector_search(
            conn,
            owner_id="bot_a",
            witness_role="you",
            query_vector=_one_hot(dim, 0),
            k=4,
        )
        assert len(you_results) == 1
 def test_vector_search_respects_owner_filter(tmp_path):
    db = tmp_path / "t.db"
    apply_migrations(db)
    with open_db(db) as conn:
        dim = 4
        _seed_memory_with_embedding(
            conn,
            owner_id="bot_a",
            pov_summary="Owner A memory.",
            vector=_one_hot(dim, 0),
        )
        _seed_memory_with_embedding(
            conn,
            owner_id="bot_b",
            pov_summary="Owner B memory.",
            vector=_one_hot(dim, 0),
        )
        a_results = vector_search(
            conn,
            owner_id="bot_a",
            witness_role="you",
            query_vector=_one_hot(dim, 0),
            k=10,
        )
        assert len(a_results) == 1
        assert a_results[0]["pov_summary"] == "Owner A memory."
        b_results = vector_search(
            conn,
            owner_id="bot_b",
            witness_role="you",
            query_vector=_one_hot(dim, 0),
            k=10,
        )
        assert len(b_results) == 1
        assert b_results[0]["pov_summary"] == "Owner B memory."
 def test_vector_search_invalid_witness_role_raises(tmp_path):
    db = tmp_path / "t.db"
    apply_migrations(db)
    with open_db(db) as conn:
        with pytest.raises(ValueError, match="witness_role"):
            vector_search(
                conn,
                owner_id="bot_a",
                witness_role="invalid",
                query_vector=[1.0, 0.0, 0.0],
                k=4,
            )
 def test_vector_search_empty_when_no_embeddings_indexed(tmp_path):
    db = tmp_path / "t.db"
    apply_migrations(db)
    with open_db(db) as conn:
        # Seed a memory but don't index an embedding for it.
        append_event(
            conn,
            kind="memory_written",
            payload=_base_memory(owner_id="bot_a", pov_summary="No embedding here."),
        )
        project(conn)
        results = vector_search(
            conn,
            owner_id="bot_a",
            witness_role="you",
            query_vector=[1.0, 0.0, 0.0, 0.0],
            k=4,
        )
        assert results == []
@@ -324,11 +324,11 @@ def test_get_scene_returns_none_for_missing(tmp_path):
        assert active_scene(conn, "chat_missing") is None
-def test_schema_version_after_migration_is_11(tmp_path):
+def test_schema_version_after_migration_is_13(tmp_path):
    db = tmp_path / "t.db"
    apply_migrations(db)
    with open_db(db) as conn:
        row = conn.execute(
            "SELECT value FROM meta WHERE key = 'schema_version'"
        ).fetchone()
-        assert int(row[0]) == 11
+        assert int(row[0]) == 13
Author	SHA1	Message	Date
Joseph Doherty	50448b72f8	merge: T96 combined FTS + vector retrieval ranking via RRF	2026-04-27 02:44:03 -04:00
Joseph Doherty	b8b4aed6d9	feat: combined FTS + vector retrieval ranking via RRF (T96)	2026-04-27 02:42:38 -04:00
Joseph Doherty	5ff107574c	merge: T95 delete-impact computation service	2026-04-27 02:37:28 -04:00
Joseph Doherty	915d625d7f	merge: T94 branching service	2026-04-27 02:37:28 -04:00
Joseph Doherty	28e13d416f	feat: delete-impact computation service (preview without mutation) (T95)	2026-04-27 02:36:30 -04:00
Joseph Doherty	296e8fdddd	feat: branching service (branch_from_event + switch + metadata) (T94)	2026-04-27 02:35:58 -04:00
Joseph Doherty	013b563f21	merge: T93 cross-chat search service	2026-04-27 02:32:53 -04:00
Joseph Doherty	62d5cdd826	merge: T92 pure-Python cosine vector search service	2026-04-27 02:32:53 -04:00
Joseph Doherty	a25c166174	merge: T91 embedding generation service (pseudo-embedding)	2026-04-27 02:32:53 -04:00
Joseph Doherty	8f66e1123a	feat: cross-chat search service (T93)	2026-04-27 02:31:31 -04:00
Joseph Doherty	caa17b4174	feat: embedding generation service (Phase 4 pseudo-embedding) (T91)	2026-04-27 02:31:07 -04:00
Joseph Doherty	c7cb0eb01e	feat: pure-Python cosine vector search service (T92)	2026-04-27 02:31:06 -04:00
Joseph Doherty	1d6768e980	test: bump schema_version assertion to 13 (0012 embeddings + 0013 branches)	2026-04-27 02:28:11 -04:00
Joseph Doherty	8b086d4bb8	merge: T90 phase 3.6 carry-overs trio	2026-04-27 02:27:48 -04:00
Joseph Doherty	6c7ac8f69f	merge: T89 branches table + projector handlers	2026-04-27 02:27:48 -04:00
Joseph Doherty	fe34d4f4c0	merge: T88 embeddings table + projector handlers	2026-04-27 02:27:48 -04:00
Joseph Doherty	0d76a6b2d6	refactor: consolidate legacy record_turn_memory into unified API (T90.3) The Phase 1 single-bot ``record_turn_memory`` lingered next to the unified ``record_turn_memory_for_present`` introduced in T84. Only test fixtures still called the legacy entry point. - Remove ``record_turn_memory`` from ``chat/services/memory_write.py``. - Update the two test_memory_write.py callers to use ``record_turn_memory_for_present(..., guest_bot_id=None)``, which produces the same ``[you=1, host=1, guest=0]`` witness mask. The unified API returns ``dict[bot_id, (event_id, memory_id)]``; tests extract the host entry. No production callers were affected.	2026-04-27 02:25:07 -04:00
Joseph Doherty	cc71fb4d01	chore: clarify regenerate lifecycle warning wording (T90.2) The warning said "lifecycle transitions from superseded turn ARE NOT being rolled back". When regenerating an OLDER turn, the listed transitions can include intervening-turn ones that legitimately stand on their own — they weren't authored by the superseded turn itself. Reword to "lifecycle transitions at-or-after turn <id>" so operators reading logs aren't misled into thinking every listed event id was emitted by the target turn. Cosmetic change to a single log message. Test: extends test_regenerate_with_prior_lifecycle_logs_warning to assert the new phrasing is present and the old phrasing is gone.	2026-04-27 02:23:55 -04:00
Joseph Doherty	c06a32767b	perf: read_recent_dialogue pushes chat-id filter into SQL (T90.1) The previous implementation pulled the last N rows in SQL across all chats and dropped foreign-chat rows in Python. With LIMIT N this could return far fewer than N relevant rows when other chats had recent activity. Push the chat_id filter into SQL via json_extract so LIMIT N always returns N rows scoped to the requested chat. Test: seeds two chats with 60 turns each interleaved; queries chat_a with limit=50; asserts exactly 50 chat_a rows returned (was 0 prior to the fix because chat_b's rows dominated the global tail).	2026-04-27 02:23:15 -04:00
Joseph Doherty	0ba374b790	feat: embeddings table + projector handlers (pure-Python cosine, T88)	2026-04-27 02:22:32 -04:00
Joseph Doherty	77f1636086	feat: branches table + projector handlers (T89)	2026-04-27 02:22:27 -04:00
Joseph Doherty	bffd9a2f38	docs: add Phase 4 implementation plan (vector retrieval + branching + polish) 15 tasks across 8 waves landing the Phase 4 deliverables per requirements doc §13 + §14: - Vector retrieval via sqlite-vec (new external dependency) - Branching UI (event log forks) - Drawer-edit on every field (significance review, hide-from-view, surgical delete with cascade preview, branching affordances) - Backup tooling (snapshot UX surface) - Cross-chat search Plus the 3 Phase 3.6 carry-over fixes (T90 bundle). Wave structure: - W1 (parallel 3-way): schema foundation + carry-overs - W2 (parallel 3-way): embedding/search services - W3 (parallel 2-way): branching + delete services - W4 (single): combined retrieval ranking - W5 (single): memory write hook + backfill - W6 (single): drawer Phase 4 bundle (5 sub-features) - W7 (parallel 2-way): snapshot UX + cross-chat search UX - W8 (parallel 2-way): integration tests + docs External dependency: sqlite-vec must be installed BEFORE Wave 1. Embedding model choice (384-dim default) pinned in T91 before dispatch since the migration hardcodes the dimension. Schema baseline: 11 -> 13 (adds 0012_embeddings.sql + 0013_branches.sql). Task ids T88-T102 to avoid collision with prior phases.	2026-04-27 02:03:08 -04:00
dohertj2	1b66a2821c	Merge pull request 'Phase 3.5 cleanup: 17-item backlog burndown' (#5 ) from phase-3.5 into main	2026-04-27 01:56:28 -04:00