dohertj2/chat
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity
Files
915d625d7f0ee482883d30e85020dc1e68e2e2b1
chat/tests/test_memory_write.py
T
Joseph Doherty 0d76a6b2d6 refactor: consolidate legacy record_turn_memory into unified API (T90.3) 
The Phase 1 single-bot ``record_turn_memory`` lingered next to the
unified ``record_turn_memory_for_present`` introduced in T84. Only test
fixtures still called the legacy entry point.

- Remove ``record_turn_memory`` from ``chat/services/memory_write.py``.
- Update the two test_memory_write.py callers to use
  ``record_turn_memory_for_present(..., guest_bot_id=None)``, which
  produces the same ``[you=1, host=1, guest=0]`` witness mask.

The unified API returns ``dict[bot_id, (event_id, memory_id)]``; tests
extract the host entry. No production callers were affected.
2026-04-27 02:25:07 -04:00

543 lines
18 KiB
Python
Raw Blame History

"""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_for_present
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:
# T90.3: legacy ``record_turn_memory`` was removed; the unified
# ``record_turn_memory_for_present`` with ``guest_bot_id=None``
# produces the same single-bot witness mask [1,1,0].
result = record_turn_memory_for_present(
conn,
chat_id="chat_bot_a",
host_bot_id="bot_a",
guest_bot_id=None,
narrative_text="BotA looks up. 'You're back late.'",
scene_id=None,
chat_clock_at="2026-04-26T20:00:00+00:00",
)
eid, mid = result["bot_a"]
assert eid > 0
assert mid is not None and mid > 0
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.
# T90.3: migrated from legacy ``record_turn_memory``.
result = record_turn_memory_for_present(
conn,
chat_id="chat_bot_a",
host_bot_id="bot_a",
guest_bot_id=None,
narrative_text="A simple memory.",
)
eid, mid = result["bot_a"]
assert eid > 0
assert mid is not None and mid > 0
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": []}
)
# T26 scene-close detection runs after the state-update pass. ``_seed_full``
# below doesn't open a scene so the classifier call is short-circuited in
# turns.py — but the canned slot stays in place to document the order.
canned_scene_close = json.dumps(
{"should_close": False, "reason": "no signal"}
)
from chat.web.kickoff import get_llm_client
mock = MockLLMClient(
canned=[
canned_parse,
canned_response,
canned_state_update,
canned_state_update,
canned_scene_close,
]
)
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
# ---------------------------------------------------------------------------
# T41: record_turn_memory_for_present — multi-witness helper.
# ---------------------------------------------------------------------------
def _seed_two_bots(db_path: Path) -> None:
"""Author host + guest bots and create a two-bot chat."""
with open_db(db_path) as conn:
for bot_id, name in (("bot_a", "BotA"), ("bot_b", "BotB")):
append_event(
conn,
kind="bot_authored",
payload={
"id": bot_id,
"name": name,
"persona": "...",
"voice_samples": [],
"traits": [],
"backstory": "",
"initial_relationship_to_you": "",
"kickoff_prose": "",
},
)
append_event(
conn,
kind="chat_created",
payload={
"id": "chat_ab",
"host_bot_id": "bot_a",
"guest_bot_id": "bot_b",
"initial_time": "2026-04-26T20:00:00+00:00",
"narrative_anchor": "Day 1",
"weather": "",
},
)
project(conn)
def test_record_for_present_no_guest_writes_single_memory_with_witness_1_1_0(tmp_path):
db = tmp_path / "t.db"
apply_migrations(db)
_seed_minimal(db)
with open_db(db) as conn:
result = record_turn_memory_for_present(
conn,
chat_id="chat_bot_a",
host_bot_id="bot_a",
guest_bot_id=None,
narrative_text="BotA glances out the window.",
scene_id=None,
chat_clock_at="2026-04-26T20:00:00+00:00",
)
# Returned dict has only the host key.
assert set(result.keys()) == {"bot_a"}
eid_h, mid_h = result["bot_a"]
assert eid_h > 0
assert mid_h is not None and mid_h > 0
rows = conn.execute(
"SELECT owner_id, witness_you, witness_host, witness_guest "
"FROM memories"
).fetchall()
assert len(rows) == 1
owner_id, w_you, w_host, w_guest = rows[0]
assert owner_id == "bot_a"
assert w_you == 1
assert w_host == 1
assert w_guest == 0
# Exactly one memory_written event was appended.
cur = conn.execute(
"SELECT COUNT(*) FROM event_log WHERE kind = 'memory_written'"
)
assert cur.fetchone()[0] == 1
def test_record_for_present_with_guest_writes_two_memories_with_witness_1_1_1(tmp_path):
db = tmp_path / "t.db"
apply_migrations(db)
_seed_two_bots(db)
with open_db(db) as conn:
result = record_turn_memory_for_present(
conn,
chat_id="chat_ab",
host_bot_id="bot_a",
guest_bot_id="bot_b",
narrative_text="BotA and BotB share a glance.",
scene_id=None,
chat_clock_at="2026-04-26T20:00:00+00:00",
)
# Returned dict has both keys.
assert set(result.keys()) == {"bot_a", "bot_b"}
eid_h, mid_h = result["bot_a"]
eid_g, mid_g = result["bot_b"]
assert eid_h > 0 and eid_g > 0
assert mid_h is not None and mid_h > 0
assert mid_g is not None and mid_g > 0
# Distinct event ids and memory ids.
assert eid_h != eid_g
assert mid_h != mid_g
rows = conn.execute(
"SELECT owner_id, witness_you, witness_host, witness_guest "
"FROM memories ORDER BY owner_id"
).fetchall()
assert len(rows) == 2
owners = {r[0] for r in rows}
assert owners == {"bot_a", "bot_b"}
# All rows should have witness mask [1, 1, 1].
for _owner, w_you, w_host, w_guest in rows:
assert w_you == 1
assert w_host == 1
assert w_guest == 1
# Two memory_written events were appended.
cur = conn.execute(
"SELECT COUNT(*) FROM event_log WHERE kind = 'memory_written'"
)
assert cur.fetchone()[0] == 2
def test_record_for_present_dict_keys_match(tmp_path):
db = tmp_path / "t.db"
apply_migrations(db)
_seed_two_bots(db)
with open_db(db) as conn:
# No guest: keys == {host_bot_id}.
result_no_guest = record_turn_memory_for_present(
conn,
chat_id="chat_ab",
host_bot_id="bot_a",
guest_bot_id=None,
narrative_text="Just BotA's POV.",
)
assert set(result_no_guest.keys()) == {"bot_a"}
# With guest: keys == {host_bot_id, guest_bot_id}.
result_with_guest = record_turn_memory_for_present(
conn,
chat_id="chat_ab",
host_bot_id="bot_a",
guest_bot_id="bot_b",
narrative_text="Both bots witness this.",
)
assert set(result_with_guest.keys()) == {"bot_a", "bot_b"}
# ---------------------------------------------------------------------------
# T84: unified record_turn_memory_for_present API with you_present kwarg.
# ---------------------------------------------------------------------------
def test_record_turn_memory_you_present_false_writes_meanwhile_witness_mask(tmp_path):
"""When ``you_present=False`` the witness mask should be
``[you=0, host=1, guest=1]`` for both bots — the meanwhile shape."""
db = tmp_path / "t.db"
apply_migrations(db)
_seed_two_bots(db)
with open_db(db) as conn:
result = record_turn_memory_for_present(
conn,
chat_id="chat_ab",
host_bot_id="bot_a",
guest_bot_id="bot_b",
narrative_text="BotA and BotB confer privately.",
scene_id=None,
chat_clock_at="2026-04-26T20:00:00+00:00",
you_present=False,
)
assert set(result.keys()) == {"bot_a", "bot_b"}
rows = conn.execute(
"SELECT owner_id, witness_you, witness_host, witness_guest "
"FROM memories ORDER BY owner_id"
).fetchall()
assert len(rows) == 2
for _owner, w_you, w_host, w_guest in rows:
assert w_you == 0
assert w_host == 1
assert w_guest == 1
# Two memory_written events were appended.
cur = conn.execute(
"SELECT COUNT(*) FROM event_log WHERE kind = 'memory_written'"
)
assert cur.fetchone()[0] == 2
def test_record_turn_memory_you_present_true_default_writes_normal_witness_mask(tmp_path):
"""Default ``you_present=True`` preserves Phase 2 behaviour:
``witness_you=1`` for the host POV row."""
db = tmp_path / "t.db"
apply_migrations(db)
_seed_minimal(db)
with open_db(db) as conn:
# No explicit you_present arg — should default to True.
result = record_turn_memory_for_present(
conn,
chat_id="chat_bot_a",
host_bot_id="bot_a",
guest_bot_id=None,
narrative_text="BotA hums to herself.",
)
assert set(result.keys()) == {"bot_a"}
row = conn.execute(
"SELECT witness_you, witness_host, witness_guest "
"FROM memories WHERE owner_id = 'bot_a'"
).fetchone()
assert row is not None
w_you, w_host, w_guest = row
assert w_you == 1
assert w_host == 1
assert w_guest == 0
def test_record_turn_memory_you_present_false_requires_guest(tmp_path):
"""Calling with ``you_present=False`` and no ``guest_bot_id`` is a
programming error — meanwhile scenes always have both bots."""
db = tmp_path / "t.db"
apply_migrations(db)
_seed_minimal(db)
with open_db(db) as conn:
with pytest.raises(ValueError, match="you_present=False requires guest_bot_id"):
record_turn_memory_for_present(
conn,
chat_id="chat_bot_a",
host_bot_id="bot_a",
guest_bot_id=None,
narrative_text="invalid",
you_present=False,
)
Reference in New Issue View Git Blame Copy Permalink