chat/tests/test_phase4_integration.py

"""Phase 4 cross-feature integration tests (T97 follow-up).

Wave 8 / T101 will populate this file with the full Phase 4 retrieval +
embedding integration suite. For now this houses a single test pinning
the T97.5 wiring: the production turn route plumbs ``app=request.app``
all the way through ``record_turn_memory_for_present`` so the embedding
worker actually receives jobs in production. Without this fix-up the
plumbing added in T97 was dormant — every per-witness write took the
no-app branch and silently dropped the embed enqueue.

The test monkeypatches ``app.state.embedding_worker.enqueue`` to record
jobs (rather than draining the worker mid-test) so the assertion is
deterministic and free of asyncio-timing flakiness inside FastAPI's
TestClient. The bug we're guarding against is "did the call site pass
``app`` at all" — the worker's drain path is exercised in
:mod:`tests.test_embedding_worker`, so duplicating that here would add
no coverage.
"""

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


def _zero_state() -> str:
    return json.dumps(
        {"affinity_delta": 0, "trust_delta": 0, "knowledge_facts": []}
    )


def _override_llm(canned: list[str]) -> MockLLMClient:
    from chat.web.kickoff import get_llm_client

    mock = MockLLMClient(canned=list(canned))
    app.dependency_overrides[get_llm_client] = lambda: mock
    return mock


@pytest.fixture
def app_state_setup(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))
    with TestClient(app) as c:
        # The background worker is disabled so the canned-response queue
        # is consumed only by the request path. The embedding worker
        # stays "started" but its loop won't observe the captured
        # enqueues — we replace ``enqueue`` on the worker instance below.
        app.state.background_worker.enabled = False
        yield c
    app.dependency_overrides.clear()


def _seed(db_path: Path) -> None:
    """Mirror of ``tests/test_turn_flow.py::_seed`` — single bot + chat
    + edge + activities so the prompt assembler has something to render.
    """
    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"],
            },
        )
        for entity_id, verb in [("you", "talking"), ("bot_a", "listening")]:
            append_event(
                conn,
                kind="activity_change",
                payload={
                    "entity_id": entity_id,
                    "posture": "sitting",
                    "action": {
                        "verb": verb,
                        "interruptible": True,
                        "required_attention": "low",
                        "expected_duration": "ongoing",
                    },
                    "attention": "",
                    "holding": [],
                    "status": {},
                },
            )
        project(conn)


def test_post_turn_embeddings_indexed_via_worker_hook(
    app_state_setup, tmp_path
):
    """POST a turn; the route must pass ``app=request.app`` into
    ``record_turn_memory_for_present`` so the per-witness write enqueues
    an :class:`EmbeddingJob` on ``app.state.embedding_worker``.

    Without the T97.5 wiring this test fails: the call site previously
    omitted ``app=`` and the helper's ``app is None`` branch silently
    skipped every enqueue. We monkeypatch ``enqueue`` on the live
    embedding worker (rather than draining the queue mid-request) so the
    assertion does not depend on asyncio scheduling inside the
    TestClient — the bug is in the wiring, and the wiring is what we
    pin. The drain path is covered separately in
    :mod:`tests.test_embedding_worker`.
    """
    _seed(tmp_path / "test.db")

    canned_parse = json.dumps(
        {"segments": [{"kind": "dialogue", "text": "hello"}]}
    )
    _override_llm(
        [canned_parse, "Hi there.", _zero_state(), _zero_state()]
    )

    captured: list = []
    worker = app.state.embedding_worker
    original_enqueue = worker.enqueue
    worker.enqueue = captured.append  # type: ignore[assignment]
    try:
        response = app_state_setup.post(
            "/chats/chat_bot_a/turns", data={"prose": "hello"}
        )
        assert response.status_code == 204
    finally:
        worker.enqueue = original_enqueue  # type: ignore[assignment]
        app.dependency_overrides.clear()

    # Single-bot turn -> one ``memory_written`` -> one EmbeddingJob.
    # The job's ``memory_id`` should match the freshly-projected memory
    # row, and its ``text`` should carry the assistant's narrative text.
    assert len(captured) == 1
    job = captured[0]
    assert job.text == "Hi there."

    with open_db(tmp_path / "test.db") as conn:
        memory_ids = [
            r[0]
            for r in conn.execute(
                "SELECT id FROM memories WHERE owner_id = ?",
                ("bot_a",),
            ).fetchall()
        ]
    assert job.memory_id in memory_ids