test: bump schema_version assertion to 13 (0012 embeddings + 0013 branches)

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.

chat/db/migrations/0012_embeddings.sql

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

chat/db/migrations/0013_branches.sql

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

chat/services/memory_write.py

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

chat/services/regenerate.py

@@ -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 "
-            "superseded turn %s are NOT being rolled back (Phase 4 "
+            "regenerate_assistant_turn: %d lifecycle transition(s) "
+            "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,

chat/services/turn_common.py

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

chat/state/branches.py

@@ -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",
+]

chat/state/embeddings.py

@@ -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",
+]

tests/test_branches_state.py

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

tests/test_embeddings_state.py

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

tests/test_memory_write.py

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

tests/test_regenerate.py

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

tests/test_turn_common.py

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

tests/test_world.py

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