chat/tests/test_turn_common.py

"""Shared turn helpers (T83.2).

``chat.services.turn_common`` extracts two snippets that were duplicated
between ``chat.web.turns`` and ``chat.services.regenerate``: the recent
user-side / assistant_turn read, and the directed-pair edge gather for
the multi-pair state-update pass. These tests pin the helpers' behavior
independently of either call site.
"""

from __future__ import annotations

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.services.turn_common import gather_prior_edges, read_recent_dialogue


def _seed_basic_chat(db_path):
    """Seed bot + chat + a couple of edges + one round of user/assistant
    turns. Returns ``(user_turn_id, assistant_turn_id)``.
    """
    apply_migrations(db_path)
    with open_db(db_path) as conn:
        append_event(
            conn,
            kind="bot_authored",
            payload={
                "id": "bot_a",
                "name": "BotA",
                "persona": "thoughtful",
                "voice_samples": [],
                "traits": [],
                "backstory": "",
                "initial_relationship_to_you": "",
                "kickoff_prose": "",
            },
        )
        append_event(
            conn,
            kind="chat_created",
            payload={
                "id": "chat_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_a",
                "affinity_delta": 7,
                "trust_delta": 3,
            },
        )
        append_event(
            conn,
            kind="edge_update",
            payload={
                "source_id": "you",
                "target_id": "bot_a",
                "chat_id": "chat_a",
                "affinity_delta": 2,
                "trust_delta": 1,
            },
        )
        ut_id = append_event(
            conn,
            kind="user_turn",
            payload={
                "chat_id": "chat_a",
                "prose": "hello",
                "segments": [],
            },
        )
        at_id = append_event(
            conn,
            kind="assistant_turn",
            payload={
                "chat_id": "chat_a",
                "speaker_id": "bot_a",
                "text": "Original.",
                "truncated": False,
                "user_turn_id": ut_id,
            },
        )
        project(conn)
    return ut_id, at_id


def test_read_recent_dialogue_returns_chronological_pairs(tmp_path):
    """``read_recent_dialogue`` returns oldest-first ``{speaker, text}``
    entries scoped to the requested chat. Speaker is "you" for user-side
    rows and the assistant_turn's ``speaker_id`` for bot rows.
    """
    db = tmp_path / "test.db"
    _seed_basic_chat(db)

    with open_db(db) as conn:
        out = read_recent_dialogue(conn, "chat_a", limit=10)

    # Each entry now carries the source ``event_log.id`` as ``event_id``
    # (T86 follow-up) so the chat-detail Jinja loop can stamp
    # ``id="turn-<n>"`` on each rendered turn DIV — needed by the
    # ``turn_html_replace`` SSE handler for in-place regenerate swaps.
    speakers = [(e["speaker"], e["text"]) for e in out]
    assert speakers == [
        ("you", "hello"),
        ("bot_a", "Original."),
    ]
    assert all("event_id" in e and isinstance(e["event_id"], int) for e in out)


def test_read_recent_dialogue_filters_superseded_and_other_chats(tmp_path):
    """Superseded rows drop out (regenerate-aware). Rows scoped to a
    different chat are also filtered. ``exclude_event_id`` excludes a
    specific row even when it isn't superseded yet (regenerate uses this
    to drop the original assistant_turn before the supersede UPDATE
    lands).
    """
    db = tmp_path / "test.db"
    ut_id, at_id = _seed_basic_chat(db)

    with open_db(db) as conn:
        # Append a second user/assistant pair.
        ut_id2 = append_event(
            conn,
            kind="user_turn",
            payload={
                "chat_id": "chat_a",
                "prose": "how are you",
                "segments": [],
            },
        )
        at_id2 = append_event(
            conn,
            kind="assistant_turn",
            payload={
                "chat_id": "chat_a",
                "speaker_id": "bot_a",
                "text": "Second.",
                "truncated": False,
                "user_turn_id": ut_id2,
            },
        )
        # And a row scoped to a different chat — must NOT appear.
        append_event(
            conn,
            kind="user_turn",
            payload={
                "chat_id": "other_chat",
                "prose": "should be filtered",
                "segments": [],
            },
        )
        # Mark the first assistant_turn as superseded — must drop out.
        conn.execute(
            "UPDATE event_log SET superseded_by = ? WHERE id = ?",
            (at_id2, at_id),
        )

        out = read_recent_dialogue(conn, "chat_a", limit=10)
        # First (superseded) assistant turn dropped; "other_chat" rows
        # filtered; first user_turn still present.
        speakers = [(e["speaker"], e["text"]) for e in out]
        assert speakers == [
            ("you", "hello"),
            ("you", "how are you"),
            ("bot_a", "Second."),
        ]

        # exclude_event_id drops at_id2 even though it's not superseded.
        out2 = read_recent_dialogue(
            conn, "chat_a", limit=10, exclude_event_id=at_id2
        )
        speakers2 = [(e["speaker"], e["text"]) for e in out2]
        assert ("bot_a", "Second.") not in speakers2
        assert ("you", "how are you") in speakers2

    # Ensure ut_id is still part of the dataset (sanity for the seed).
    assert ut_id is not None


def test_gather_prior_edges_fills_missing_with_default(tmp_path):
    """``gather_prior_edges`` returns one entry per directed pair across
    ``present_ids``. Missing rows fall back to the schema default
    50/50 baseline; existing rows carry their stored values.
    """
    db = tmp_path / "test.db"
    _seed_basic_chat(db)

    with open_db(db) as conn:
        out = gather_prior_edges(conn, ["bot_a", "you"])

    # 2 entities -> 2 directed pairs (a->b and b->a, no self-pairs).
    assert set(out.keys()) == {("bot_a", "you"), ("you", "bot_a")}
    bot_to_you = out[("bot_a", "you")]
    you_to_bot = out[("you", "bot_a")]
    # Both edges seeded with deltas — they must reflect the projected
    # affinity/trust (not the default 50/50).
    assert bot_to_you["affinity"] == 57  # 50 + 7
    assert bot_to_you["trust"] == 53  # 50 + 3
    assert you_to_bot["affinity"] == 52
    assert you_to_bot["trust"] == 51

    # A pair with no row yet falls back to 50/50.
    with open_db(db) as conn:
        out_with_missing = gather_prior_edges(
            conn, ["bot_a", "you", "ghost_bot"]
        )
    # 3 entities -> 6 directed pairs.
    assert len(out_with_missing) == 6
    fallback = out_with_missing[("bot_a", "ghost_bot")]
    assert fallback["affinity"] == 50
    assert fallback["trust"] == 50
    assert fallback["summary"] == ""