feat: FTS5 memory retrieval with witness filter and ranking boosts

chat/app.py

@@ -7,6 +7,7 @@ from fastapi.staticfiles import StaticFiles
 
 from chat.config import load_settings
 from chat.db.migrate import apply_migrations
+from chat.services.background import BackgroundWorker
 
 # Trigger handler registration:
 import chat.state.entities  # noqa: F401
@@ -29,7 +30,26 @@ async def lifespan(app: FastAPI):
     settings.db_path.parent.mkdir(parents=True, exist_ok=True)
     apply_migrations(settings.db_path)
     app.state.settings = settings
-    yield
+
+    # Background worker for the async significance pass (T22). Each job
+    # constructs a fresh FeatherlessClient via the factory; tests can
+    # disable enqueue by toggling ``app.state.background_worker.enabled``.
+    def _factory():
+        from chat.llm.featherless import FeatherlessClient
+
+        return FeatherlessClient(
+            api_key=settings.featherless_api_key,
+            base_url=settings.featherless_base_url,
+        )
+
+    worker = BackgroundWorker(settings, llm_client_factory=_factory)
+    await worker.start()
+    app.state.background_worker = worker
+
+    try:
+        yield
+    finally:
+        await worker.stop()
 
 
 app = FastAPI(title="chat", lifespan=lifespan)

chat/eventlog/log.py

@@ -24,6 +24,45 @@ def append_event(conn: Connection, *, kind: str, payload: dict[str, Any], branch
     return cur.lastrowid
 
 
+def append_and_apply(
+    conn: Connection,
+    *,
+    kind: str,
+    payload: dict[str, Any],
+    branch_id: int = 1,
+) -> int:
+    """Append an event AND immediately apply just that event's handler.
+
+    Calling :func:`chat.eventlog.projector.project` after an append
+    re-runs every prior event, which is fine for idempotent inserts but
+    catastrophic for delta-shaped events like ``edge_update`` whose
+    handler is *not* replay-safe (each pass would re-add the same
+    ``affinity_delta``). This helper runs only the brand-new event
+    through the registered handler, leaving prior state untouched.
+
+    No-ops cleanly when ``kind`` has no registered handler — useful for
+    transcript-only events like ``user_turn`` / ``assistant_turn`` where
+    callers may swap ``append_event`` for ``append_and_apply`` without
+    side effects.
+    """
+    # Local import to avoid a circular dependency at module import: the
+    # projector imports from .log to define ``Event``.
+    from chat.eventlog.projector import apply_event
+
+    eid = append_event(conn, kind=kind, payload=payload, branch_id=branch_id)
+    event = Event(
+        id=eid,
+        branch_id=branch_id,
+        ts="",
+        kind=kind,
+        payload=payload,
+        superseded_by=None,
+        hidden=False,
+    )
+    apply_event(conn, event)
+    return eid
+
+
 def read_events(conn: Connection, branch_id: int = 1, after_id: int = 0) -> Iterator[Event]:
     cur = conn.execute(
         "SELECT id, branch_id, ts, kind, payload_json, superseded_by, hidden "

chat/services/background.py

@@ -0,0 +1,173 @@
+"""Async background worker for post-turn jobs (T22).
+
+The turn flow records a ``memory_written`` event synchronously on the
+request path so the timeline updates immediately. Significance scoring is
+a separate classifier round-trip that we don't want to block on, so the
+turn handler enqueues a :class:`SignificanceJob` here and the worker
+drains the queue out-of-band.
+
+A single :class:`BackgroundWorker` is started/stopped via FastAPI lifespan
+in :mod:`chat.app`. The worker owns its own ``asyncio.Queue`` and runs
+exactly one task that pulls jobs off the queue, calls
+:func:`chat.services.significance.compute_significance`, and writes
+``memory_significance_set`` (and on score 3, ``memory_pin_changed``)
+events. Each job opens its own DB connection — workers and request
+handlers don't share connections.
+
+Failures inside ``_process`` are logged and swallowed: a flaky classifier
+shouldn't take down the worker. Tests can disable enqueue() by setting
+``BackgroundWorker.enabled = False`` (e.g. in the existing turn-flow
+fixture, which doesn't have a usable LLM key for the lifespan-managed
+factory).
+"""
+
+from __future__ import annotations
+
+import asyncio
+import logging
+from dataclasses import dataclass
+from typing import Callable
+
+from chat.config import Settings
+from chat.db.connection import open_db
+from chat.eventlog.log import append_and_apply
+from chat.llm.client import LLMClient
+from chat.services.significance import compute_significance
+
+log = logging.getLogger(__name__)
+
+
+@dataclass
+class SignificanceJob:
+    """One unit of work for the background worker.
+
+    ``host_bot_id`` is the memory's owner — used both for the auto-pin
+    soft cap query and as the eventual scope for the soft-cap eviction.
+    """
+
+    memory_id: int
+    narrative_text: str
+    prior_dialogue: list[dict]
+    host_bot_id: str
+
+
+class BackgroundWorker:
+    """asyncio.Queue-backed single-worker task.
+
+    Started on app startup; ``stop()`` enqueues a sentinel and awaits the
+    task so any in-flight job has a chance to finish. Pending jobs after
+    the sentinel are dropped on shutdown — Phase 1 simplification.
+    """
+
+    def __init__(
+        self,
+        settings: Settings,
+        llm_client_factory: Callable[[], LLMClient],
+        *,
+        enabled: bool = True,
+    ) -> None:
+        self._settings = settings
+        self._llm_client_factory = llm_client_factory
+        self._queue: asyncio.Queue[SignificanceJob | None] = asyncio.Queue()
+        self._task: asyncio.Task | None = None
+        self.enabled = enabled
+
+    async def start(self) -> None:
+        if self._task is not None:
+            return
+        self._task = asyncio.create_task(self._run())
+
+    async def stop(self) -> None:
+        if self._task is None:
+            return
+        await self._queue.put(None)  # sentinel
+        await self._task
+        self._task = None
+
+    def enqueue(self, job: SignificanceJob) -> None:
+        if not self.enabled:
+            return
+        self._queue.put_nowait(job)
+
+    async def _run(self) -> None:
+        while True:
+            job = await self._queue.get()
+            if job is None:
+                return
+            try:
+                await self._process(job)
+            except Exception as exc:  # noqa: BLE001 — worker must not die
+                log.exception("significance job failed: %s", exc)
+
+    async def _process(self, job: SignificanceJob) -> None:
+        client = self._llm_client_factory()
+        score = await compute_significance(
+            client,
+            model=self._settings.classifier_model,
+            narrative_text=job.narrative_text,
+            prior_dialogue=job.prior_dialogue,
+        )
+        with open_db(self._settings.db_path) as conn:
+            append_and_apply(
+                conn,
+                kind="memory_significance_set",
+                payload={
+                    "memory_id": job.memory_id,
+                    "significance": score,
+                },
+            )
+            if score >= 3:
+                _auto_pin_with_cap(
+                    conn,
+                    owner_id=job.host_bot_id,
+                    memory_id=job.memory_id,
+                )
+
+
+def _auto_pin_with_cap(
+    conn,
+    *,
+    owner_id: str,
+    memory_id: int,
+    cap: int = 8,
+) -> None:
+    """Auto-pin ``memory_id`` and evict the oldest auto-pin if over ``cap``.
+
+    Per §8.5: pivotal turns are auto-pinned, with a soft cap of 8 pins per
+    bot. When the cap is exceeded the oldest auto-pin is unpinned (manual
+    pins are never auto-evicted — we filter on ``auto_pinned = 1``).
+    """
+    append_and_apply(
+        conn,
+        kind="memory_pin_changed",
+        payload={
+            "memory_id": memory_id,
+            "pinned": 1,
+            "auto_pinned": 1,
+        },
+    )
+    cur = conn.execute(
+        "SELECT COUNT(*) FROM memories WHERE owner_id = ? AND pinned = 1",
+        (owner_id,),
+    )
+    count = cur.fetchone()[0]
+    if count <= cap:
+        return
+    cur = conn.execute(
+        "SELECT id FROM memories "
+        "WHERE owner_id = ? AND pinned = 1 AND auto_pinned = 1 AND id != ? "
+        "ORDER BY created_at ASC, id ASC LIMIT 1",
+        (owner_id, memory_id),
+    )
+    row = cur.fetchone()
+    if row is None:
+        return
+    append_and_apply(
+        conn,
+        kind="memory_pin_changed",
+        payload={
+            "memory_id": row[0],
+            "pinned": 0,
+            "auto_pinned": 0,
+        },
+    )

chat/services/memory_write.py

@@ -0,0 +1,78 @@
+"""Per-turn memory writes (T21).
+
+After ``assistant_turn`` lands, the turn flow records a ``memory_written``
+event for each present POV owner. Phase 1 single-bot turns only have the
+host bot as a memory-store owner — ``you`` doesn't have a memory store in
+v1 — so we write exactly one row per turn.
+
+Phase 1 simplifications (per plan §11.1, T27 will refine):
+
+- ``pov_summary`` is the assistant's raw narrative text. T27 rewrites at
+  scene close into per-POV summary form.
+- ``significance`` defaults to ``1`` (Notable). T22's async significance
+  pass overwrites via a follow-up event.
+- Witness flags are hard-coded ``[you=1, host=1, guest=0]``. Phase 2 will
+  derive them from ``chat.guest_bot_id`` once a guest can be present.
+"""
+
+from __future__ import annotations
+
+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

chat/services/significance.py

@@ -0,0 +1,75 @@
+"""Turn-level significance scorer (T22).
+
+Per Requirements §11.1, each turn is scored on a 0-3 scale:
+
+- 0 = Routine: small talk, ordinary action.
+- 1 = Notable: a specific detail or beat worth remembering.
+- 2 = Significant: a scene-level moment, real disagreement, confided secret.
+- 3 = Pivotal: a relationship-altering event (first kiss, betrayal, "I love
+  you").
+
+The scorer is conservative: pivotal (3) requires a clear signal because the
+auto-pin rule (§8.5) gives those memories permanent shelf space. The
+classifier returns a strict-JSON ``SignificanceVerdict``; a malformed or
+refusal-shaped response falls back to ``score=1`` (Notable) — a safe
+middle-of-the-road default that won't trigger auto-pin.
+"""
+
+from __future__ import annotations
+
+from pydantic import BaseModel, Field
+
+from chat.llm.classify import classify
+from chat.llm.client import LLMClient
+
+
+class SignificanceVerdict(BaseModel):
+    score: int = Field(ge=0, le=3)
+    reason: str = ""
+
+
+_SYSTEM = """You score the significance of a roleplay turn 0-3:
+0 = Routine: small talk, ordinary action.
+1 = Notable: a specific detail or beat worth remembering.
+2 = Significant: a scene-level moment, real disagreement, confided secret.
+3 = Pivotal: a relationship-altering event (first kiss, betrayal, "I love you").
+
+Be conservative — pivotal (3) requires a clear signal. Reply with JSON: {"score": int 0-3, "reason": str}."""
+
+
+async def compute_significance(
+    client: LLMClient,
+    *,
+    model: str,
+    narrative_text: str,
+    prior_dialogue: list[dict],
+    timeout_s: float = 10.0,
+) -> int:
+    """Score the significance of ``narrative_text`` (the just-written turn).
+
+    ``prior_dialogue`` is a list of ``{"speaker", "text"}`` dicts ordered
+    oldest-first; the last 6 entries are stitched into the user prompt as
+    context so the classifier can recognize escalation. Returns an int in
+    ``[0, 3]`` — clamped defensively in case the classifier slips a value
+    past the schema validator.
+    """
+    user_prompt = "PRIOR DIALOGUE:\n"
+    for turn in prior_dialogue[-6:]:
+        speaker = turn.get("speaker", "?")
+        text = turn.get("text", "")
+        user_prompt += f"{speaker}: {text}\n"
+    user_prompt += (
+        f"\nNEW TURN:\n{narrative_text}\n\n"
+        "Score the significance of the NEW TURN."
+    )
+
+    result = await classify(
+        client,
+        model=model,
+        system=_SYSTEM,
+        user=user_prompt,
+        schema=SignificanceVerdict,
+        default=SignificanceVerdict(score=1, reason="fallback"),
+        timeout_s=timeout_s,
+    )
+    return max(0, min(3, result.score))

chat/services/state_update.py

@@ -0,0 +1,144 @@
+"""Post-turn state-update pass.
+
+Per Requirements §3.4, after every utterance we run a classifier on each
+present entity (silent witnesses included) to extract directed-edge
+deltas — what changed in *source*'s view of *target*. The classifier
+returns three signals:
+
+- ``affinity_delta`` — signed change in how warmly source feels (typical
+  range -3..+3; the edge handler clamps the running total to 0..100).
+- ``trust_delta`` — signed change in source's trust of target (same
+  shape).
+- ``knowledge_facts`` — concrete things source learned about target
+  during this exchange. Stored verbatim and appended to ``edge.knowledge``.
+
+The wrapper deliberately uses :func:`chat.llm.classify.classify` with a
+``default=StateUpdate()`` so a flapping classifier never blocks the turn
+flow — at worst the edge sits unchanged and the next turn tries again
+(§3.3 "graceful degradation" rule).
+"""
+
+from __future__ import annotations
+
+from pydantic import BaseModel, Field
+
+from chat.llm.classify import classify
+from chat.llm.client import LLMClient
+
+
+class StateUpdate(BaseModel):
+    """One directed-edge update from a single classifier call.
+
+    Defaults are deliberately a no-op (zero deltas, empty facts) so a
+    failing classifier produces a benign event rather than a disruption.
+    """
+
+    affinity_delta: int = 0
+    trust_delta: int = 0
+    knowledge_facts: list[str] = Field(default_factory=list)
+
+
+_SYSTEM_PROMPT = (
+    "You are reading a recent slice of dialogue from a roleplay scene. "
+    "You assess how SOURCE's view of TARGET shifted based on what was "
+    "said — including silent witnessing (SOURCE may not have spoken).\n\n"
+    "Output a JSON object with exactly three fields:\n"
+    "- affinity_delta: signed integer in [-3, 3]. How much warmer "
+    "(positive) or cooler (negative) SOURCE now feels toward TARGET.\n"
+    "- trust_delta: signed integer in [-3, 3]. How much more (positive) "
+    "or less (negative) SOURCE now trusts TARGET.\n"
+    "- knowledge_facts: list of short strings. New, concrete facts "
+    "SOURCE learned about TARGET in this exchange. Use TARGET's actual "
+    "stated content; do not infer or interpret. Empty list is fine.\n\n"
+    "Be conservative. Most turns produce small deltas (-1, 0, +1). "
+    "Reserve +/-2 or +/-3 for moments that materially shift the "
+    "relationship. Knowledge_facts should be specific things stated in "
+    "dialogue (e.g. \"works at the bakery\"), not interpretations "
+    "(\"seems lonely\")."
+)
+
+
+def _format_dialogue(recent_dialogue: list[dict]) -> str:
+    """Render the recent-dialogue slice as plain ``Speaker: text`` lines."""
+    if not recent_dialogue:
+        return "(no dialogue yet)"
+    lines = []
+    for turn in recent_dialogue:
+        speaker = turn.get("speaker", "?")
+        text = turn.get("text", "")
+        lines.append(f"{speaker}: {text}")
+    return "\n".join(lines)
+
+
+def _build_user_prompt(
+    *,
+    source_name: str,
+    source_persona: str,
+    target_name: str,
+    prior_affinity: int,
+    prior_trust: int,
+    prior_summary: str,
+    recent_dialogue: list[dict],
+) -> str:
+    return (
+        f"SOURCE: {source_name}\n"
+        f"SOURCE_PERSONA: {source_persona or '(none)'}\n"
+        f"TARGET: {target_name}\n"
+        f"PRIOR_AFFINITY (0-100): {prior_affinity}\n"
+        f"PRIOR_TRUST (0-100): {prior_trust}\n"
+        f"PRIOR_SUMMARY: {prior_summary or '(none)'}\n\n"
+        f"RECENT_DIALOGUE:\n{_format_dialogue(recent_dialogue)}\n\n"
+        "How did SOURCE's view of TARGET shift? Respond with JSON only."
+    )
+
+
+async def compute_state_update(
+    client: LLMClient,
+    *,
+    model: str,
+    source_id: str,
+    target_id: str,
+    source_name: str,
+    source_persona: str,
+    target_name: str,
+    prior_affinity: int,
+    prior_trust: int,
+    prior_summary: str,
+    recent_dialogue: list[dict],
+    timeout_s: float = 10.0,
+) -> StateUpdate:
+    """Run a classifier pass and return the directed-edge update.
+
+    On classifier failure (after retry) returns the schema default — a
+    no-op ``StateUpdate`` — so the turn flow can keep moving. The
+    ``source_id`` / ``target_id`` arguments are accepted for symmetry
+    with the caller (T20's POST flow uses them when emitting the
+    ``edge_update`` event); they're not currently embedded in the
+    prompt because the classifier reasons about names, not opaque ids.
+    """
+    # ``source_id``/``target_id`` are kept on the signature even though
+    # the prompt only quotes the names: callers in turns.py thread the
+    # ids straight from this function's args into the appended event.
+    del source_id, target_id  # silence unused-arg lint cleanly
+
+    user_prompt = _build_user_prompt(
+        source_name=source_name,
+        source_persona=source_persona,
+        target_name=target_name,
+        prior_affinity=prior_affinity,
+        prior_trust=prior_trust,
+        prior_summary=prior_summary,
+        recent_dialogue=recent_dialogue,
+    )
+
+    return await classify(
+        client,
+        model=model,
+        system=_SYSTEM_PROMPT,
+        user=user_prompt,
+        schema=StateUpdate,
+        default=StateUpdate(),
+        timeout_s=timeout_s,
+    )
+
+

chat/state/memory.py

@@ -38,6 +38,35 @@ def _apply_memory_written(conn: Connection, e: Event) -> None:
     )
 
 
+@on("memory_significance_set")
+def _apply_memory_significance_set(conn: Connection, e: Event) -> None:
+    """Update an existing memory's significance score (T22).
+
+    Emitted by the async significance worker after it scores the turn.
+    """
+    p = e.payload
+    conn.execute(
+        "UPDATE memories SET significance = ? WHERE id = ?",
+        (int(p["significance"]), int(p["memory_id"])),
+    )
+
+
+@on("memory_pin_changed")
+def _apply_memory_pin_changed(conn: Connection, e: Event) -> None:
+    """Toggle a memory's pin state (T22, §8.5).
+
+    Used both for auto-pinning a pivotal turn and for evicting the oldest
+    auto-pin when the per-owner soft cap is exceeded. Manual pins use the
+    same handler; the ``auto_pinned`` flag distinguishes them so the
+    eviction query can leave manual pins alone.
+    """
+    p = e.payload
+    conn.execute(
+        "UPDATE memories SET pinned = ?, auto_pinned = ? WHERE id = ?",
+        (int(p["pinned"]), int(p["auto_pinned"]), int(p["memory_id"])),
+    )
+
+
 def get_memory(conn: Connection, memory_id: int) -> dict | None:
     row = conn.execute(
         "SELECT * FROM memories WHERE id = ?", (memory_id,)
@@ -58,6 +87,14 @@ def get_pinned(conn: Connection, owner_id: str) -> list[dict]:
     return [dict(zip(cols, row)) for row in rows]
 
 
+# Composite-score weights used by ``search_memories`` (T23, §8 retrieval).
+# FTS5 BM25 ``rank`` is *more negative* for better matches, so subtracting a
+# positive boost from it drives stronger candidates further down (i.e. earlier
+# in an ascending sort). Hardcoded for v1 — tunable in a later pass.
+_SIGNIFICANCE_WEIGHT = 0.3
+_RECENCY_WEIGHT = 0.5
+
+
 def search_memories(
     conn: Connection,
     owner_id: str,
@@ -68,16 +105,32 @@ def search_memories(
     """FTS5 search over pov_summary, scoped by owner and witness role.
 
     witness_role must be one of {"you", "host", "guest"} per the witness flags
-    on each memory row. Returns up to k rows ordered by FTS5 bm25 rank.
+    on each memory row. Returns up to ``k`` rows ranked by a composite score
+    that combines the FTS5 BM25 rank with two boosts (§8 retrieval rules):
+
+    * **significance boost** — ``0.3 * significance`` (0..3 per §11.1).
+    * **recency boost** — ``0.5 * (id / max_id)``, using the row id as a
+      monotonic recency proxy. Newer memories therefore tilt above older ones
+      when the BM25 rank and significance are otherwise tied.
+
+    BM25 returns negative scores (lower = better). Both boosts are subtracted
+    so that stronger candidates yield smaller composite scores; the result is
+    sorted ascending and truncated to ``k``. The unmodified ``fts_rank`` and a
+    debug-friendly ``composite_score`` are kept on each returned dict.
     """
     if witness_role not in _VALID_WITNESS_ROLES:
         raise ValueError(
             f"witness_role must be one of {sorted(_VALID_WITNESS_ROLES)}, "
             f"got {witness_role!r}"
         )
+    if not query.strip():
+        return []
     witness_col = f"witness_{witness_role}"
     cols = [c[1] for c in conn.execute("PRAGMA table_info(memories)").fetchall()]
     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)
     sql = (
         f"SELECT {select_list}, memories_fts.rank AS fts_rank "
         "FROM memories_fts "
@@ -87,10 +140,27 @@ def search_memories(
         "ORDER BY memories_fts.rank "
         "LIMIT ?"
     )
-    cur = conn.execute(sql, (owner_id, query, k))
+    cur = conn.execute(sql, (owner_id, query, over_fetch))
     rows = cur.fetchall()
-    out: list[dict] = []
+    if not rows:
+        return []
+
+    # Recency normalises against the current max id for this owner so the
+    # boost magnitude is bounded regardless of dataset size.
+    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 row in rows:
-        out.append(dict(zip(result_cols, row)))
+        d = dict(zip(result_cols, row))
-    return out
+        fts_rank = d.get("fts_rank") or 0.0
+        sig_boost = _SIGNIFICANCE_WEIGHT * (d.get("significance") or 0)
+        recency_boost = _RECENCY_WEIGHT * ((d.get("id") or 0) / max_id)
+        d["composite_score"] = fts_rank - sig_boost - recency_boost
+        enriched.append(d)
+
+    enriched.sort(key=lambda x: x["composite_score"])
+    return enriched[:k]

chat/web/turns.py

@@ -15,9 +15,12 @@ The turn flow strings together the pieces built in T17 (turn parser), T18
    channel as a ``token`` event so any subscribed browser tab sees them
    arrive in real time.
 6. On stream complete, append an ``assistant_turn`` event with the full
-   text and ``truncated=False``. Also publish a ``turn_html`` event with a
+   text and ``truncated=False``. Then run a post-turn state-update pass
-   ready-to-swap HTML fragment so HTMX's SSE extension can append it to
+   (Requirements §3.4): one classifier call per directed edge between
-   the timeline without a page reload.
+   present entities, each producing an ``edge_update`` event with
+   affinity/trust/knowledge deltas. Finally publish a ``turn_html``
+   event with a ready-to-swap HTML fragment so HTMX's SSE extension can
+   append it to the timeline without a page reload.
 7. Return ``204 No Content`` — the SSE channel is the real conveyor of
    state, not the POST response body.
 
@@ -35,11 +38,15 @@ import json
 from fastapi import APIRouter, Depends, Form, HTTPException, Request
 from fastapi.responses import Response
 
-from chat.eventlog.log import append_event
+from chat.eventlog.log import append_and_apply, append_event
+from chat.services.background import SignificanceJob
+from chat.services.memory_write import record_turn_memory
 from chat.services.prompt import assemble_narrative_prompt
+from chat.services.state_update import compute_state_update
 from chat.services.turn_parse import ParsedTurn, parse_turn
-from chat.state.world import get_chat
+from chat.state.edges import get_edge
-from chat.state.entities import get_bot
+from chat.state.entities import get_bot, get_you
+from chat.state.world import active_scene, get_chat
 from chat.web.bots import get_conn
 from chat.web.kickoff import get_llm_client
 from chat.web.pubsub import publish
@@ -214,6 +221,116 @@ async def post_turn(
         },
     )
 
+    # 6a. Per-turn memory write (Plan §11.1, T21). Phase 1 single-bot:
+    # only the host bot has a memory store, witness flags are
+    # ``[you=1, host=1, guest=0]``, and ``pov_summary`` is the raw
+    # narrative text (T27 will rewrite at scene close). Significance
+    # defaults to 1; T22's async classifier pass will overwrite it.
+    scene = active_scene(conn, chat_id)
+    _event_id, memory_id = record_turn_memory(
+        conn,
+        chat_id=chat_id,
+        host_bot_id=host_bot["id"],
+        narrative_text=full_text,
+        scene_id=scene["id"] if scene else None,
+        chat_clock_at=chat.get("time"),
+    )
+
+    # 6b. Post-turn state-update pass (Requirements §3.4). For Phase 1
+    # the only present entities are ``you`` and ``host_bot`` so we run
+    # two classifier calls — one per directed edge — and append the
+    # resulting ``edge_update`` events. The recent-dialogue slice is
+    # re-read here so the pass sees the just-appended assistant turn.
+    # We use ``append_and_apply`` (vs append + project) because the
+    # edge_update handler is *not* replay-safe: re-projecting prior
+    # events would re-apply their deltas on top of the live row.
+    recent_for_update = _read_recent_dialogue(conn, chat_id, limit=10)
+    you_entity = get_you(conn) or {"name": "you", "persona": ""}
+    last_at = chat.get("time")
+
+    edge_b2y = get_edge(conn, host_bot["id"], "you") or {
+        "affinity": 50,
+        "trust": 50,
+        "summary": "",
+    }
+    update_b2y = await compute_state_update(
+        client,
+        model=settings.classifier_model,
+        source_id=host_bot["id"],
+        target_id="you",
+        source_name=host_bot["name"],
+        source_persona=host_bot.get("persona", ""),
+        target_name=you_entity.get("name", "you"),
+        prior_affinity=edge_b2y["affinity"],
+        prior_trust=edge_b2y["trust"],
+        prior_summary=edge_b2y.get("summary", "") or "",
+        recent_dialogue=recent_for_update,
+    )
+    append_and_apply(
+        conn,
+        kind="edge_update",
+        payload={
+            "source_id": host_bot["id"],
+            "target_id": "you",
+            "chat_id": chat_id,
+            "affinity_delta": update_b2y.affinity_delta,
+            "trust_delta": update_b2y.trust_delta,
+            "knowledge_facts": update_b2y.knowledge_facts,
+            "last_interaction_at": last_at,
+            "last_interaction_chat_id": chat_id,
+        },
+    )
+
+    edge_y2b = get_edge(conn, "you", host_bot["id"]) or {
+        "affinity": 50,
+        "trust": 50,
+        "summary": "",
+    }
+    update_y2b = await compute_state_update(
+        client,
+        model=settings.classifier_model,
+        source_id="you",
+        target_id=host_bot["id"],
+        source_name=you_entity.get("name", "you"),
+        source_persona=you_entity.get("persona", "") or "",
+        target_name=host_bot["name"],
+        prior_affinity=edge_y2b["affinity"],
+        prior_trust=edge_y2b["trust"],
+        prior_summary=edge_y2b.get("summary", "") or "",
+        recent_dialogue=recent_for_update,
+    )
+    append_and_apply(
+        conn,
+        kind="edge_update",
+        payload={
+            "source_id": "you",
+            "target_id": host_bot["id"],
+            "chat_id": chat_id,
+            "affinity_delta": update_y2b.affinity_delta,
+            "trust_delta": update_y2b.trust_delta,
+            "knowledge_facts": update_y2b.knowledge_facts,
+            "last_interaction_at": last_at,
+            "last_interaction_chat_id": chat_id,
+        },
+    )
+
+    # 6c. Enqueue the async significance pass (Plan §11.1, T22). The
+    # worker scores the just-written memory 0-3, updates significance,
+    # and auto-pins on score 3 with the §8.5 soft-cap eviction rule.
+    # Enqueued before the broadcast so it's outstanding by the time the
+    # client sees ``turn_html`` — but the worker is async, so the user
+    # never blocks on it.
+    worker = getattr(request.app.state, "background_worker", None)
+    if worker is not None and memory_id is not None:
+        worker.enqueue(
+            SignificanceJob(
+                memory_id=memory_id,
+                narrative_text=full_text,
+                prior_dialogue=recent_for_update,
+                host_bot_id=host_bot["id"],
+            )
+        )
+
     # 7. Broadcast a JSON completion event (for JS consumers) and an HTML
     # fragment event (for HTMX SSE swap-into-timeline).
     await publish(

tests/test_memory_search.py

@@ -0,0 +1,127 @@
+"""Task 23: FTS5 memory retrieval with witness filter and ranking boosts.
+
+Verifies that ``search_memories`` applies recency + significance boosts on top
+of the FTS5 BM25 rank so that newer / more significant memories surface above
+older / less significant ones for the same match. Existing T8 behaviour
+(witness filter, k limit, FTS match, role validation) is exercised again here
+to lock the contract.
+"""
+
+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
+from chat.state.memory import search_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.
+
+    memory_specs: list of dicts. Required key: ``pov_summary``. Optional keys
+    override the defaults below.
+    """
+    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_filters_by_witness_bit(tmp_path):
+    db = tmp_path / "t.db"
+    _seed(
+        db,
+        memory_specs=[
+            {
+                "pov_summary": "BotA mentioned her sister",
+                "witness_you": 1,
+                "witness_host": 1,
+                "witness_guest": 0,
+            },
+        ],
+    )
+    with open_db(db) as conn:
+        # Witnessed by host -> returned.
+        out = search_memories(conn, "bot_a", "host", "sister", k=4)
+        assert len(out) == 1
+        # NOT witnessed by guest -> filtered out.
+        out = search_memories(conn, "bot_a", "guest", "sister", k=4)
+        assert out == []
+
+
+def test_search_higher_significance_ranks_above_lower(tmp_path):
+    db = tmp_path / "t.db"
+    _seed(
+        db,
+        memory_specs=[
+            # Both match "promise"; the third row carries significance 3 and
+            # should outrank the first two, which carry the default of 1.
+            {"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
+        assert out[0]["pov_summary"] == "tiny promise"
+        assert out[0]["significance"] == 3
+
+
+def test_search_newer_memory_ranks_above_older_when_same_match(tmp_path):
+    db = tmp_path / "t.db"
+    _seed(
+        db,
+        memory_specs=[
+            {"pov_summary": "BotA said hello"},
+            {"pov_summary": "BotA said hello again"},
+        ],
+    )
+    with open_db(db) as conn:
+        out = search_memories(conn, "bot_a", "host", "hello", k=2)
+        assert len(out) == 2
+        # Newer (higher id, "again") wins on the recency boost when the BM25
+        # rank and significance are otherwise comparable.
+        assert out[0]["pov_summary"] == "BotA said hello again"
+
+
+def test_search_respects_k_limit(tmp_path):
+    db = tmp_path / "t.db"
+    _seed(
+        db,
+        memory_specs=[
+            {"pov_summary": "the cat sat"},
+            {"pov_summary": "the cat ran"},
+            {"pov_summary": "the cat slept"},
+            {"pov_summary": "the cat ate"},
+            {"pov_summary": "the cat purred"},
+        ],
+    )
+    with open_db(db) as conn:
+        out = search_memories(conn, "bot_a", "host", "cat", k=2)
+        assert len(out) == 2
+
+
+def test_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):
+            search_memories(conn, "bot_a", "invalid_role", "anything", k=4)

tests/test_memory_write.py

@@ -0,0 +1,290 @@
+"""Per-turn memory writes (T21).
+
+After ``assistant_turn`` lands the turn flow records a ``memory_written``
+event for each present POV owner. Phase 1 single-bot turns only have the
+host bot as a memory-store owner — ``you`` doesn't have a memory store in
+v1 — so we write exactly one row per turn with witness flags
+``[you=1, host=1, guest=0]``. The ``pov_summary`` is the assistant's raw
+narrative text; T27 rewrites at scene close into per-POV summary form.
+"""
+
+from __future__ import annotations
+
+import json
+from pathlib import Path
+
+import pytest
+from fastapi.testclient import TestClient
+
+from chat.app import app
+from chat.db.connection import open_db
+from chat.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
+import chat.state.entities  # noqa: F401  - register handlers
+import chat.state.memory  # noqa: F401
+import chat.state.world  # noqa: F401
+
+
+def _seed_minimal(db_path: Path) -> None:
+    """Author a bot and create a chat — bare minimum for memory writes."""
+    with open_db(db_path) as conn:
+        append_event(
+            conn,
+            kind="bot_authored",
+            payload={
+                "id": "bot_a",
+                "name": "BotA",
+                "persona": "...",
+                "voice_samples": [],
+                "traits": [],
+                "backstory": "",
+                "initial_relationship_to_you": "",
+                "kickoff_prose": "",
+            },
+        )
+        append_event(
+            conn,
+            kind="chat_created",
+            payload={
+                "id": "chat_bot_a",
+                "host_bot_id": "bot_a",
+                "initial_time": "2026-04-26T20:00:00+00:00",
+                "narrative_anchor": "Day 1",
+                "weather": "",
+            },
+        )
+        project(conn)
+
+
+def test_record_turn_memory_writes_event_and_projects(tmp_path):
+    db = tmp_path / "t.db"
+    apply_migrations(db)
+    _seed_minimal(db)
+    with open_db(db) as conn:
+        eid, mid = record_turn_memory(
+            conn,
+            chat_id="chat_bot_a",
+            host_bot_id="bot_a",
+            narrative_text="BotA looks up. 'You're back late.'",
+            scene_id=None,
+            chat_clock_at="2026-04-26T20:00:00+00:00",
+        )
+        assert eid > 0
+        assert mid is not None and mid > 0
+
+        rows = conn.execute(
+            "SELECT id, owner_id, chat_id, pov_summary, "
+            "witness_you, witness_host, witness_guest, "
+            "source, reliability, significance, pinned, auto_pinned, "
+            "chat_clock_at "
+            "FROM memories WHERE owner_id = ?",
+            ("bot_a",),
+        ).fetchall()
+        assert len(rows) == 1
+        m = rows[0]
+        assert m[1] == "bot_a"
+        assert m[2] == "chat_bot_a"
+        assert "looks up" in m[3]
+        assert m[4] == 1  # witness_you
+        assert m[5] == 1  # witness_host
+        assert m[6] == 0  # witness_guest
+        assert m[7] == "direct"
+        assert m[8] == 1.0  # reliability default
+        assert m[9] == 1  # significance default
+        assert m[10] == 0  # pinned default
+        assert m[11] == 0  # auto_pinned default
+        assert m[12] == "2026-04-26T20:00:00+00:00"
+
+        # And the underlying event_log row is the canonical source.
+        cur = conn.execute(
+            "SELECT COUNT(*) FROM event_log WHERE kind = 'memory_written'"
+        )
+        assert cur.fetchone()[0] == 1
+
+
+def test_record_turn_memory_omits_optional_fields(tmp_path):
+    db = tmp_path / "t.db"
+    apply_migrations(db)
+    _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(
+            conn,
+            chat_id="chat_bot_a",
+            host_bot_id="bot_a",
+            narrative_text="A simple memory.",
+        )
+        assert eid > 0
+        assert mid is not None and mid > 0
+
+        row = conn.execute(
+            "SELECT scene_id, chat_clock_at, source, reliability, "
+            "significance, pinned, auto_pinned "
+            "FROM memories WHERE owner_id = 'bot_a'"
+        ).fetchone()
+        assert row is not None
+        scene_id, chat_clock_at, source, reliability, significance, pinned, auto_pinned = row
+        assert scene_id is None
+        assert chat_clock_at is None
+        assert source == "direct"
+        assert reliability == 1.0
+        assert significance == 1
+        assert pinned == 0
+        assert auto_pinned == 0
+
+
+# ---------------------------------------------------------------------------
+# Integration: POST /chats/<id>/turns produces a memory_written event.
+# ---------------------------------------------------------------------------
+
+
+@pytest.fixture
+def client(tmp_path, monkeypatch):
+    cfg = tmp_path / "config.toml"
+    cfg.write_text('featherless_api_key = "test"\n')
+    monkeypatch.setenv("CHAT_CONFIG_PATH", str(cfg))
+    db = tmp_path / "test.db"
+    monkeypatch.setenv("CHAT_DB_PATH", str(db))
+
+    canned_parse = json.dumps(
+        {"segments": [{"kind": "dialogue", "text": "hello"}]}
+    )
+    canned_response = "BotA nods. 'Hi there.'"
+    canned_state_update = json.dumps(
+        {"affinity_delta": 0, "trust_delta": 0, "knowledge_facts": []}
+    )
+
+    from chat.web.kickoff import get_llm_client
+
+    mock = MockLLMClient(
+        canned=[
+            canned_parse,
+            canned_response,
+            canned_state_update,
+            canned_state_update,
+        ]
+    )
+    app.dependency_overrides[get_llm_client] = lambda: mock
+
+    with TestClient(app) as c:
+        # Disable the lifespan-managed background worker — it would try
+        # to call Featherless with the test API key. The unit tests in
+        # test_significance.py exercise the worker directly with a mock
+        # factory; here we only care about the synchronous turn flow.
+        app.state.background_worker.enabled = False
+        c.mock_llm = mock  # type: ignore[attr-defined]
+        yield c
+
+    app.dependency_overrides.clear()
+
+
+def _seed_full(db_path: Path) -> None:
+    """Seed enough state for a full turn flow (matches test_turn_flow.py)."""
+    with open_db(db_path) as conn:
+        append_event(
+            conn,
+            kind="bot_authored",
+            payload={
+                "id": "bot_a",
+                "name": "BotA",
+                "persona": "thoughtful, observant",
+                "voice_samples": [],
+                "traits": [],
+                "backstory": "",
+                "initial_relationship_to_you": "",
+                "kickoff_prose": "...",
+            },
+        )
+        append_event(
+            conn,
+            kind="chat_created",
+            payload={
+                "id": "chat_bot_a",
+                "host_bot_id": "bot_a",
+                "initial_time": "2026-04-26T20:00:00+00:00",
+                "narrative_anchor": "Day 1",
+                "weather": "",
+            },
+        )
+        append_event(
+            conn,
+            kind="edge_update",
+            payload={
+                "source_id": "bot_a",
+                "target_id": "you",
+                "chat_id": "chat_bot_a",
+                "knowledge_facts": ["coworker"],
+            },
+        )
+        append_event(
+            conn,
+            kind="activity_change",
+            payload={
+                "entity_id": "you",
+                "posture": "sitting",
+                "action": {
+                    "verb": "talking",
+                    "interruptible": True,
+                    "required_attention": "low",
+                    "expected_duration": "ongoing",
+                },
+                "attention": "",
+                "holding": [],
+                "status": {},
+            },
+        )
+        append_event(
+            conn,
+            kind="activity_change",
+            payload={
+                "entity_id": "bot_a",
+                "posture": "sitting",
+                "action": {
+                    "verb": "listening",
+                    "interruptible": True,
+                    "required_attention": "low",
+                    "expected_duration": "ongoing",
+                },
+                "attention": "",
+                "holding": [],
+                "status": {},
+            },
+        )
+        project(conn)
+
+
+def test_post_turn_writes_memory_for_host_bot(client, tmp_path):
+    """After a POST turn, exactly one memory_written event is appended and
+    a corresponding memory row is projected for the host bot's POV."""
+    _seed_full(tmp_path / "test.db")
+    response = client.post(
+        "/chats/chat_bot_a/turns", data={"prose": "hello"}
+    )
+    assert response.status_code == 204
+
+    with open_db(tmp_path / "test.db") as conn:
+        cur = conn.execute(
+            "SELECT COUNT(*) FROM event_log WHERE kind = 'memory_written'"
+        )
+        assert cur.fetchone()[0] == 1
+
+        cur = conn.execute(
+            "SELECT owner_id, chat_id, pov_summary, "
+            "witness_you, witness_host, witness_guest, source, significance "
+            "FROM memories"
+        )
+        rows = cur.fetchall()
+        assert len(rows) == 1
+        owner_id, chat_id, pov_summary, w_you, w_host, w_guest, source, sig = rows[0]
+        assert owner_id == "bot_a"
+        assert chat_id == "chat_bot_a"
+        # pov_summary is the assistant's narrative text (Phase 1 simplification).
+        assert pov_summary == "BotA nods. 'Hi there.'"
+        assert w_you == 1
+        assert w_host == 1
+        assert w_guest == 0
+        assert source == "direct"
+        assert sig == 1

tests/test_significance.py

@@ -0,0 +1,237 @@
+"""Async significance pass with auto-pin on score 3 (T22).
+
+After ``assistant_turn`` lands the turn flow enqueues a SignificanceJob on
+a background asyncio worker. The worker calls a classifier (per §11.1,
+score 0-3) and writes a ``memory_significance_set`` event. On score 3 the
+memory is auto-pinned and a soft cap of 8 pins per owner is enforced —
+when the cap is exceeded the oldest auto-pin (excluding the just-pinned
+row) is unpinned via another ``memory_pin_changed`` event.
+"""
+
+from __future__ import annotations
+
+import asyncio
+import json
+
+import pytest
+
+from chat.config import load_settings
+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.llm.mock import MockLLMClient
+from chat.services.background import BackgroundWorker, SignificanceJob
+from chat.services.significance import compute_significance
+
+# Trigger handler registration for projection.
+import chat.state.entities  # noqa: F401
+import chat.state.memory  # noqa: F401
+import chat.state.world  # noqa: F401
+
+
+async def test_compute_significance_parses_score():
+    canned = json.dumps({"score": 2, "reason": "notable"})
+    mock = MockLLMClient(canned=[canned])
+    score = await compute_significance(
+        mock,
+        model="x",
+        narrative_text="...",
+        prior_dialogue=[],
+    )
+    assert score == 2
+
+
+async def test_compute_significance_default_on_failure():
+    # Both attempts return non-JSON text; the classify wrapper falls back
+    # to the SignificanceVerdict default (score=1, "fallback").
+    mock = MockLLMClient(canned=["nope", "still nope"])
+    score = await compute_significance(
+        mock,
+        model="x",
+        narrative_text="...",
+        prior_dialogue=[],
+    )
+    assert score == 1
+
+
+async def test_background_worker_processes_job_and_updates_significance(
+    tmp_path, monkeypatch
+):
+    cfg = tmp_path / "config.toml"
+    cfg.write_text('featherless_api_key = "test"\n')
+    monkeypatch.setenv("CHAT_CONFIG_PATH", str(cfg))
+    db = tmp_path / "test.db"
+    monkeypatch.setenv("CHAT_DB_PATH", str(db))
+    apply_migrations(db)
+    settings = load_settings()
+
+    # Seed bot, chat, memory.
+    with open_db(db) as conn:
+        append_event(
+            conn,
+            kind="bot_authored",
+            payload={
+                "id": "bot_a",
+                "name": "BotA",
+                "persona": "...",
+                "voice_samples": [],
+                "traits": [],
+                "backstory": "",
+                "initial_relationship_to_you": "",
+                "kickoff_prose": "",
+            },
+        )
+        append_event(
+            conn,
+            kind="chat_created",
+            payload={
+                "id": "chat_bot_a",
+                "host_bot_id": "bot_a",
+                "initial_time": "2026-04-26T20:00:00+00:00",
+                "narrative_anchor": "Day 1",
+                "weather": "",
+            },
+        )
+        append_event(
+            conn,
+            kind="memory_written",
+            payload={
+                "owner_id": "bot_a",
+                "chat_id": "chat_bot_a",
+                "pov_summary": "Some scene",
+                "witness_you": 1,
+                "witness_host": 1,
+                "witness_guest": 0,
+                "source": "direct",
+                "reliability": 1.0,
+                "significance": 1,
+                "pinned": 0,
+                "auto_pinned": 0,
+            },
+        )
+        project(conn)
+        memory_id = conn.execute(
+            "SELECT id FROM memories WHERE owner_id = 'bot_a'"
+        ).fetchone()[0]
+
+    # Worker with mock LLM that returns score=3 (pivotal).
+    canned = [json.dumps({"score": 3, "reason": "pivotal"})]
+    factory = lambda: MockLLMClient(canned=list(canned))
+    worker = BackgroundWorker(settings, llm_client_factory=factory)
+    await worker.start()
+    worker.enqueue(
+        SignificanceJob(
+            memory_id=memory_id,
+            narrative_text="...",
+            prior_dialogue=[],
+            host_bot_id="bot_a",
+        )
+    )
+    # Drain via stop sentinel — guarantees the prior job completed.
+    await worker.stop()
+
+    # Verify significance updated AND memory auto-pinned.
+    with open_db(db) as conn:
+        row = conn.execute(
+            "SELECT significance, pinned, auto_pinned FROM memories "
+            "WHERE id = ?",
+            (memory_id,),
+        ).fetchone()
+        assert row[0] == 3
+        assert row[1] == 1  # pinned
+        assert row[2] == 1  # auto_pinned
+
+
+async def test_auto_pin_evicts_oldest_when_over_cap(tmp_path, monkeypatch):
+    """Pin 9 memories with score 3; verify only 8 are pinned at the end."""
+    cfg = tmp_path / "config.toml"
+    cfg.write_text('featherless_api_key = "test"\n')
+    monkeypatch.setenv("CHAT_CONFIG_PATH", str(cfg))
+    db = tmp_path / "test.db"
+    monkeypatch.setenv("CHAT_DB_PATH", str(db))
+    apply_migrations(db)
+    settings = load_settings()
+
+    with open_db(db) as conn:
+        append_event(
+            conn,
+            kind="bot_authored",
+            payload={
+                "id": "bot_a",
+                "name": "BotA",
+                "persona": "...",
+                "voice_samples": [],
+                "traits": [],
+                "backstory": "",
+                "initial_relationship_to_you": "",
+                "kickoff_prose": "",
+            },
+        )
+        append_event(
+            conn,
+            kind="chat_created",
+            payload={
+                "id": "chat_bot_a",
+                "host_bot_id": "bot_a",
+                "initial_time": "2026-04-26T20:00:00+00:00",
+                "narrative_anchor": "Day 1",
+                "weather": "",
+            },
+        )
+        for i in range(9):
+            append_event(
+                conn,
+                kind="memory_written",
+                payload={
+                    "owner_id": "bot_a",
+                    "chat_id": "chat_bot_a",
+                    "pov_summary": f"memory {i}",
+                    "witness_you": 1,
+                    "witness_host": 1,
+                    "witness_guest": 0,
+                    "source": "direct",
+                    "reliability": 1.0,
+                    "significance": 1,
+                    "pinned": 0,
+                    "auto_pinned": 0,
+                },
+            )
+        project(conn)
+        memory_ids = [
+            r[0]
+            for r in conn.execute(
+                "SELECT id FROM memories WHERE owner_id = 'bot_a' ORDER BY id"
+            ).fetchall()
+        ]
+
+    # Each job runs through its own MockLLMClient with one canned response.
+    factory = lambda: MockLLMClient(
+        canned=[json.dumps({"score": 3, "reason": "pivotal"})]
+    )
+    worker = BackgroundWorker(settings, llm_client_factory=factory)
+    await worker.start()
+    for mid in memory_ids:
+        worker.enqueue(
+            SignificanceJob(
+                memory_id=mid,
+                narrative_text="...",
+                prior_dialogue=[],
+                host_bot_id="bot_a",
+            )
+        )
+    await worker.stop()
+
+    with open_db(db) as conn:
+        pinned_count = conn.execute(
+            "SELECT COUNT(*) FROM memories "
+            "WHERE owner_id = 'bot_a' AND pinned = 1"
+        ).fetchone()[0]
+        assert pinned_count == 8
+
+        # The oldest should have been evicted.
+        first_id = memory_ids[0]
+        first_pinned = conn.execute(
+            "SELECT pinned FROM memories WHERE id = ?", (first_id,)
+        ).fetchone()[0]
+        assert first_pinned == 0

tests/test_state_update.py

@@ -0,0 +1,237 @@
+"""Post-turn state-update pass (T20).
+
+Per Requirements §3.4, after each utterance we run a classifier on every
+present entity (silent witnesses too) to extract directed-edge deltas
+(``affinity_delta``, ``trust_delta``, ``knowledge_facts``). The deltas
+land as ``edge_update`` events and project into the ``edges`` table.
+
+These tests cover:
+- The unit-level :func:`compute_state_update` happy path: classifier
+  returns valid JSON, the wrapper returns a populated ``StateUpdate``.
+- The unit-level fallback path: classifier fails twice, the wrapper
+  returns a no-op ``StateUpdate`` (zeros + empty facts) per §3.3.
+- The integration path: a successful POST appends two ``edge_update``
+  events (one per direction) after the ``assistant_turn`` and the edge
+  projections reflect the deltas.
+"""
+
+from __future__ import annotations
+
+import json
+from pathlib import Path
+
+import pytest
+from fastapi.testclient import TestClient
+
+from chat.app import app
+from chat.db.connection import open_db
+from chat.eventlog.log import append_event
+from chat.eventlog.projector import project
+from chat.llm.mock import MockLLMClient
+from chat.services.state_update import StateUpdate, compute_state_update
+
+
+@pytest.mark.asyncio
+async def test_compute_state_update_parses_classifier_output():
+    canned = json.dumps(
+        {"affinity_delta": 2, "trust_delta": 1, "knowledge_facts": ["likes coffee"]}
+    )
+    mock = MockLLMClient(canned=[canned])
+    result = await compute_state_update(
+        mock,
+        model="x",
+        source_id="bot_a",
+        target_id="you",
+        source_name="BotA",
+        source_persona="thoughtful",
+        target_name="Me",
+        prior_affinity=50,
+        prior_trust=50,
+        prior_summary="",
+        recent_dialogue=[
+            {"speaker": "you", "text": "hi"},
+            {"speaker": "BotA", "text": "Hello!"},
+        ],
+    )
+    assert isinstance(result, StateUpdate)
+    assert result.affinity_delta == 2
+    assert result.trust_delta == 1
+    assert result.knowledge_facts == ["likes coffee"]
+
+
+@pytest.mark.asyncio
+async def test_compute_state_update_returns_default_on_failure():
+    """Two malformed classifier responses -> default StateUpdate (zeros)."""
+    mock = MockLLMClient(canned=["nope", "still nope"])
+    result = await compute_state_update(
+        mock,
+        model="x",
+        source_id="bot_a",
+        target_id="you",
+        source_name="BotA",
+        source_persona="",
+        target_name="Me",
+        prior_affinity=50,
+        prior_trust=50,
+        prior_summary="",
+        recent_dialogue=[],
+    )
+    assert result.affinity_delta == 0
+    assert result.trust_delta == 0
+    assert result.knowledge_facts == []
+
+
+# --- integration test --------------------------------------------------------
+
+
+@pytest.fixture
+def client(tmp_path, monkeypatch):
+    cfg = tmp_path / "config.toml"
+    cfg.write_text('featherless_api_key = "test"\n')
+    monkeypatch.setenv("CHAT_CONFIG_PATH", str(cfg))
+    db = tmp_path / "test.db"
+    monkeypatch.setenv("CHAT_DB_PATH", str(db))
+
+    canned_parse = json.dumps(
+        {"segments": [{"kind": "dialogue", "text": "hello"}]}
+    )
+    canned_response = "Hi there."
+    canned_state_b2y = json.dumps(
+        {"affinity_delta": 2, "trust_delta": 1, "knowledge_facts": ["greets warmly"]}
+    )
+    canned_state_y2b = json.dumps(
+        {"affinity_delta": 3, "trust_delta": 0, "knowledge_facts": []}
+    )
+
+    from chat.web.kickoff import get_llm_client
+
+    mock = MockLLMClient(
+        canned=[canned_parse, canned_response, canned_state_b2y, canned_state_y2b]
+    )
+    app.dependency_overrides[get_llm_client] = lambda: mock
+
+    with TestClient(app) as c:
+        c.mock_llm = mock  # type: ignore[attr-defined]
+        yield c
+
+    app.dependency_overrides.clear()
+
+
+def _seed(db_path: Path) -> None:
+    """Author a bot, create a chat, and seed enough state for prompt assembly."""
+    with open_db(db_path) as conn:
+        append_event(
+            conn,
+            kind="bot_authored",
+            payload={
+                "id": "bot_a",
+                "name": "BotA",
+                "persona": "thoughtful, observant",
+                "voice_samples": [],
+                "traits": [],
+                "backstory": "",
+                "initial_relationship_to_you": "",
+                "kickoff_prose": "...",
+            },
+        )
+        append_event(
+            conn,
+            kind="chat_created",
+            payload={
+                "id": "chat_bot_a",
+                "host_bot_id": "bot_a",
+                "initial_time": "2026-04-26T20:00:00+00:00",
+                "narrative_anchor": "Day 1",
+                "weather": "",
+            },
+        )
+        append_event(
+            conn,
+            kind="edge_update",
+            payload={
+                "source_id": "bot_a",
+                "target_id": "you",
+                "chat_id": "chat_bot_a",
+                "knowledge_facts": ["coworker"],
+            },
+        )
+        append_event(
+            conn,
+            kind="activity_change",
+            payload={
+                "entity_id": "you",
+                "posture": "sitting",
+                "action": {
+                    "verb": "talking",
+                    "interruptible": True,
+                    "required_attention": "low",
+                    "expected_duration": "ongoing",
+                },
+                "attention": "",
+                "holding": [],
+                "status": {},
+            },
+        )
+        append_event(
+            conn,
+            kind="activity_change",
+            payload={
+                "entity_id": "bot_a",
+                "posture": "sitting",
+                "action": {
+                    "verb": "listening",
+                    "interruptible": True,
+                    "required_attention": "low",
+                    "expected_duration": "ongoing",
+                },
+                "attention": "",
+                "holding": [],
+                "status": {},
+            },
+        )
+        project(conn)
+
+
+def test_post_turn_appends_edge_updates_and_applies_deltas(client, tmp_path):
+    """After a turn, edge_update events fire for both directions and project."""
+    db_path = tmp_path / "test.db"
+    _seed(db_path)
+
+    response = client.post("/chats/chat_bot_a/turns", data={"prose": "hello"})
+    assert response.status_code == 204
+
+    with open_db(db_path) as conn:
+        # Two new edge_update events should land *after* the assistant_turn.
+        cur = conn.execute(
+            "SELECT kind, payload_json FROM event_log "
+            "WHERE kind = 'edge_update' "
+            "AND id > (SELECT MAX(id) FROM event_log WHERE kind = 'assistant_turn') "
+            "ORDER BY id"
+        )
+        rows = cur.fetchall()
+        assert len(rows) == 2
+        kinds = [r[0] for r in rows]
+        assert kinds == ["edge_update", "edge_update"]
+
+        # Inspect the two payloads — one per direction.
+        payloads = [json.loads(r[1]) for r in rows]
+        directions = {(p["source_id"], p["target_id"]) for p in payloads}
+        assert ("bot_a", "you") in directions
+        assert ("you", "bot_a") in directions
+
+        # Edge bot_a -> you: seeded affinity=50, plus delta 2 -> 52.
+        from chat.state.edges import get_edge
+
+        edge_b2y = get_edge(conn, "bot_a", "you")
+        assert edge_b2y is not None
+        assert edge_b2y["affinity"] == 52
+        assert edge_b2y["trust"] == 51
+        # Existing fact preserved, new fact appended.
+        assert "coworker" in edge_b2y["knowledge"]
+        assert "greets warmly" in edge_b2y["knowledge"]
+
+        # Edge you -> bot_a: defaults (50/50) plus delta +3 affinity -> 53.
+        edge_y2b = get_edge(conn, "you", "bot_a")
+        assert edge_y2b is not None
+        assert edge_y2b["affinity"] == 53
+        assert edge_y2b["trust"] == 50

tests/test_turn_flow.py

@@ -36,14 +36,33 @@ def client(tmp_path, monkeypatch):
         {"segments": [{"kind": "dialogue", "text": "hello"}]}
     )
     canned_response = "Hi there."
+    # Two state-update classifier calls fire after the assistant_turn
+    # (one per directed edge: bot->you, you->bot). We feed them benign
+    # zero-delta JSON so the existing assertions about ``user_turn`` /
+    # ``assistant_turn`` are unaffected.
+    canned_state_update = json.dumps(
+        {"affinity_delta": 0, "trust_delta": 0, "knowledge_facts": []}
+    )
 
     # Import here so env vars are visible to the dependency lookup.
     from chat.web.kickoff import get_llm_client
 
-    mock = MockLLMClient(canned=[canned_parse, canned_response])
+    mock = MockLLMClient(
+        canned=[
+            canned_parse,
+            canned_response,
+            canned_state_update,
+            canned_state_update,
+        ]
+    )
     app.dependency_overrides[get_llm_client] = lambda: mock
 
     with TestClient(app) as c:
+        # Disable the lifespan-managed background worker — it would
+        # otherwise try to score significance through Featherless with
+        # a fake test API key. Worker behavior is exercised directly in
+        # tests/test_significance.py with a mock LLM factory.
+        app.state.background_worker.enabled = False
         c.mock_llm = mock  # type: ignore[attr-defined]
         yield c