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