brain-of-reese/tests/unit/test_history.py

"""Unit: the client-history → model-messages mapper (phase 74, TODO L4).

``app.rag.prompts.history_to_messages`` is pure (no I/O) — every branch
is pinned here: the user/brain role mapping, the ``reasoning_content``
gating (prior thinking travels ONLY when non-empty — the preserve-
thinking wire convention, A4), the turn-count budget (newest kept,
oldest dropped), the char budget (``text`` + ``thinking`` accounted,
drop-WHOLE semantics — never cut mid-answer, A3), the budgets working
together, and the chronological (oldest → newest) order of the result.
"""
from __future__ import annotations

from typing import Any, Literal

from app.config import Settings
from app.rag.prompts import history_to_messages
from app.schemas import HistoryTurn


def _settings(**kwargs: Any) -> Settings:
    kwargs.setdefault("_env_file", None)
    return Settings(**kwargs)  # pyright: ignore[reportCallIssue]  (kwarg exists at runtime)


def _turn(
    who: Literal["user", "brain"], text: str, thinking: str | None = None
) -> HistoryTurn:
    return HistoryTurn(who=who, text=text, thinking=thinking)


# ---------- mapping ----------


def test_empty_history_yields_no_messages() -> None:
    """Absent client history (the pre-phase-74 request shape) → ``[]`` —
    the caller then builds the byte-identical two-message request."""
    assert history_to_messages([], _settings()) == []


def test_user_turn_maps_to_user_role() -> None:
    got = history_to_messages([_turn("user", "What port does Tailscale use?")], _settings())
    assert got == [{"role": "user", "content": "What port does Tailscale use?"}]


def test_brain_turn_without_thinking_maps_to_assistant_role() -> None:
    """No ``thinking`` key → a plain assistant message: NO
    ``reasoning_content`` key at all (A4 gating, ``None`` case)."""
    got = history_to_messages([_turn("brain", "Tailscale runs on 41641/udp.")], _settings())
    assert got == [{"role": "assistant", "content": "Tailscale runs on 41641/udp."}]
    assert "reasoning_content" not in got[0]


def test_brain_turn_with_thinking_carries_reasoning_content() -> None:
    """A prior thinking block travels as ``reasoning_content`` on the
    assistant message (A4 — the preserve-thinking wire convention the
    response side already reads)."""
    thinking = "Tailscale's wire protocol port is 41641/udp."
    got = history_to_messages(
        [_turn("brain", "Tailscale runs on 41641/udp.", thinking=thinking)],
        _settings(),
    )
    assert got == [
        {
            "role": "assistant",
            "content": "Tailscale runs on 41641/udp.",
            "reasoning_content": thinking,
        }
    ]


def test_brain_turn_with_empty_thinking_omits_reasoning_content() -> None:
    """``thinking=""`` is "empty" for the A4 gate — no
    ``reasoning_content`` key (an empty scratchpad carries nothing)."""
    got = history_to_messages(
        [_turn("brain", "Same answer.", thinking="")], _settings()
    )
    assert got == [{"role": "assistant", "content": "Same answer."}]
    assert "reasoning_content" not in got[0]


def test_result_is_chronological_oldest_to_newest() -> None:
    """The input is oldest-first; the output must be too — the newest
    turn ends up LAST, directly ahead of the current user message the
    caller appends."""
    turns = [
        _turn("user", "q1"),
        _turn("brain", "a1", thinking="t1"),
        _turn("user", "q2"),
        _turn("brain", "a2"),
        _turn("user", "q3"),
    ]
    got = history_to_messages(turns, _settings())
    assert [m["role"] for m in got] == ["user", "assistant", "user", "assistant", "user"]
    assert [m["content"] for m in got] == ["q1", "a1", "q2", "a2", "q3"]
    assert got[1]["reasoning_content"] == "t1"
    assert "reasoning_content" not in got[3]


# ---------- owner shape (phase 108, TODO L4: the 2-turn follow-up) ----------


def test_short_two_turn_history_kept_whole_and_chronological() -> None:
    """Phase 108 layer 1 (TODO L4 — the owner's exact shape): one user
    question + one brain answer is orders of magnitude under the DEFAULT
    budgets (40 turns / 24 000 chars — no overrides, the file's plain
    ``_settings()``) and must survive the trimmer WHOLE: both turns kept,
    chronological (Q1 then R1), roles mapped, no trim, no reordering. A
    server-side missing-first-turn on the owner's follow-up would have to
    be dropped here."""
    history = [
        _turn("user", "What is my name?"),
        _turn("brain", "Your name is Reese."),
    ]
    assert history_to_messages(history, _settings()) == [
        {"role": "user", "content": "What is my name?"},
        {"role": "assistant", "content": "Your name is Reese."},
    ]


def test_two_turn_history_thinking_mapping() -> None:
    """Phase 108 layer 1 (TODO L4), the A4 gate on the owner's shape: a
    non-empty prior ``thinking`` travels as ``reasoning_content`` on the
    assistant message; with ``thinking`` absent or empty the key is
    ABSENT (not an empty string) — the message equals the plain brain
    turn."""
    got = history_to_messages(
        [
            _turn("user", "What is my name?"),
            _turn("brain", "Your name is Reese.", thinking="The owner asked for their name."),
        ],
        _settings(),
    )
    assert got == [
        {"role": "user", "content": "What is my name?"},
        {
            "role": "assistant",
            "content": "Your name is Reese.",
            "reasoning_content": "The owner asked for their name.",
        },
    ]
    for thinking in (None, ""):  # absent and empty — both omit the key
        got = history_to_messages(
            [
                _turn("user", "What is my name?"),
                _turn("brain", "Your name is Reese.", thinking=thinking),
            ],
            _settings(),
        )
        assert got == [
            {"role": "user", "content": "What is my name?"},
            {"role": "assistant", "content": "Your name is Reese."},
        ]
        assert "reasoning_content" not in got[1]


# ---------- turn-count budget ----------


def test_turn_cap_keeps_newest_and_drops_oldest() -> None:
    """The newest ``history_max_turns`` turns are kept; the OLDEST are
    the ones dropped (newest-first walk, stop at the count cap)."""
    turns = [_turn("user", f"q{i}") for i in range(1, 6)]  # q1 … q5, oldest first
    got = history_to_messages(turns, _settings(history_max_turns=3, history_max_chars=10_000))
    assert [m["content"] for m in got] == ["q3", "q4", "q5"]


def test_default_turn_cap_is_40() -> None:
    """45 turns under the DEFAULT caps (40 turns / 24 000 chars, short
    texts so the char budget never binds) keep the newest 40."""
    turns = [_turn("user", f"question number {i}") for i in range(1, 46)]
    got = history_to_messages(turns, _settings())
    assert len(got) == 40
    assert got[0]["content"] == "question number 6"  # the five oldest are gone
    assert got[-1]["content"] == "question number 45"


# ---------- char budget ----------


def test_char_budget_counts_text_plus_thinking() -> None:
    """The per-turn size is ``len(text) + len(thinking or "")`` — prior
    thinking blocks count against the same budget as the answer text."""
    # Newest-first sizes: 5 + 100 (20+80) + 10; budget 110 keeps the
    # newest two (105) and drops the oldest (115 > 110).
    turns = [
        _turn("user", "a" * 10),  # oldest — dropped whole
        _turn("brain", "b" * 20, thinking="c" * 80),
        _turn("user", "d" * 5),  # newest
    ]
    got = history_to_messages(turns, _settings(history_max_turns=40, history_max_chars=110))
    assert len(got) == 2
    assert got[0]["content"] == "b" * 20
    assert got[0]["reasoning_content"] == "c" * 80
    assert got[1]["content"] == "d" * 5


def test_char_budget_exact_fit_is_kept() -> None:
    """Cumulative chars EQUAL to the cap fit (≤, not <) — the exact-fit
    turn is kept, and the older turn that would push past is dropped."""
    turns = [
        _turn("user", "a" * 10),  # oldest — 100+10=110 > 100, dropped
        _turn("brain", "b" * 100),  # newest — exactly the 100-char cap, kept
    ]
    got = history_to_messages(turns, _settings(history_max_turns=40, history_max_chars=100))
    assert [m["content"] for m in got] == ["b" * 100]


def test_overflowing_turn_is_dropped_whole_never_truncated() -> None:
    """A turn that would overflow the remaining budget is DROPPED WHOLE
    (A3) — its text appears nowhere in the result, not even partially,
    and the walk stops there (the kept history stays a contiguous
    newest window)."""
    big = "x" * 120  # alone it would overflow the 100-char budget
    turns = [
        _turn("user", "old question"),
        _turn("brain", "old answer"),
        _turn("brain", big),  # newest — does not fit at all
    ]
    got = history_to_messages(turns, _settings(history_max_turns=40, history_max_chars=100))
    assert got == []  # the newest does not fit → nothing is kept
    assert not any("x" in m["content"] for m in got)


def test_overflowing_middle_turn_stops_the_walk() -> None:
    """Newest-first: the newest fits, the NEXT (middle) turn would
    overflow → it is dropped whole AND the walk stops — the oldest turn
    is not sneaked in across the gap (no discontinuous history)."""
    turns = [
        _turn("user", "a" * 5),  # oldest — never even considered
        _turn("user", "b" * 51),  # middle — 60+51=111 > 100, dropped whole
        _turn("user", "c" * 60),  # newest — fits (60 ≤ 100)
    ]
    got = history_to_messages(turns, _settings(history_max_turns=40, history_max_chars=100))
    assert [m["content"] for m in got] == ["c" * 60]


def test_zero_char_budget_yields_no_history() -> None:
    """``history_max_chars=0`` is a budget that fits nothing — the
    kill-switch shape (no history, pre-phase-74 two-message request)."""
    turns = [_turn("user", "q1"), _turn("brain", "a1")]
    assert history_to_messages(turns, _settings(history_max_chars=0)) == []


def test_zero_turn_budget_yields_no_history() -> None:
    """``history_max_turns=0`` keeps no turns even though chars are free."""
    turns = [_turn("user", "q1"), _turn("brain", "a1")]
    assert history_to_messages(turns, _settings(history_max_turns=0)) == []


# ---------- budgets together ----------


def test_turn_cap_wins_when_chars_remain() -> None:
    """Both budgets in play: plenty of chars, a small turn cap — the
    count cap stops the walk first (newest 3 of 5 kept)."""
    turns = [_turn("user", f"q{i}") for i in range(1, 6)]
    got = history_to_messages(turns, _settings(history_max_turns=3, history_max_chars=10_000))
    assert len(got) == 3
    assert [m["content"] for m in got] == ["q3", "q4", "q5"]


def test_char_cap_wins_when_turns_remain() -> None:
    """Symmetrically: plenty of turn budget, a tight char cap — the char
    budget stops the walk (2 of 3 turns kept)."""
    turns = [
        _turn("user", "a" * 50),  # oldest — dropped (50+60=110 > 100)
        _turn("user", "b" * 60),
        _turn("user", "c" * 40),  # newest
    ]
    got = history_to_messages(turns, _settings(history_max_turns=40, history_max_chars=100))
    assert [m["content"] for m in got] == ["b" * 60, "c" * 40]