chat/tests/test_memory_write.py

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