4 Commits

Author SHA1 Message Date
Joseph Doherty 3995a8671b feat: FTS5 memory retrieval with witness filter and ranking boosts 2026-04-26 13:30:40 -04:00
Joseph Doherty eb4cdf9cbb feat: async significance pass with auto-pin on score 3 2026-04-26 13:27:25 -04:00
Joseph Doherty a45dabb6ae feat: per-turn memory writes with witness flags 2026-04-26 13:20:43 -04:00
Joseph Doherty e8d24a0875 feat: post-turn state-update pass per present entity 2026-04-26 13:17:07 -04:00
13 changed files with 1639 additions and 13 deletions
+21 -1
View File
@@ -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)
+39
View File
@@ -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 "
+173
View File
@@ -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,
},
)
+78
View File
@@ -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
+75
View File
@@ -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))
+144
View File
@@ -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,
)
+75 -5
View File
@@ -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)))
return out
d = dict(zip(result_cols, row))
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]
+123 -6
View File
@@ -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
ready-to-swap HTML fragment so HTMX's SSE extension can append it to
the timeline without a page reload.
text and ``truncated=False``. Then run a post-turn state-update pass
(Requirements §3.4): one classifier call per directed edge between
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.entities import get_bot
from chat.state.edges import get_edge
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(
+127
View File
@@ -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)
+290
View File
@@ -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
+237
View File
@@ -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
+237
View File
@@ -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
+20 -1
View File
@@ -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