"""Tests for ChatPanel per-message widget rendering and the visible-window cap.
Each message is its own ``_ChatItemView`` widget. On a per-token stream only the
streaming bubble's widget re-renders (signature-gated); everything else is
untouched. The window cap bounds how many item widgets stay mounted regardless
of total history length, and a truncation hint sits at the top when older
messages are dropped.
"""
from __future__ import annotations
from pathlib import Path
import pytest
from clients.terminal.tui.tui_app import NaviCodeTui
@pytest.fixture(autouse=True)
def tmp_state_dir(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> Path:
from clients.terminal import config
import clients.terminal.tui.settings as settings_module
original = config.settings.state_dir
config.settings.state_dir = tmp_path
settings_module._tui_settings = None
yield tmp_path
config.settings.state_dir = original
settings_module._tui_settings = None
@pytest.fixture(autouse=True)
def mock_tui_api(monkeypatch: pytest.MonkeyPatch) -> None:
import clients.terminal.api as api_module
monkeypatch.setattr(api_module, "create_session", lambda *a, **k: {"session_id": "s", "profile_id": "navi_code"})
monkeypatch.setattr(api_module, "get_session", lambda sid: {"session_id": sid, "profile_id": "navi_code"})
monkeypatch.setattr(api_module, "list_sessions", lambda: [])
monkeypatch.setattr(api_module, "get_profile_model", lambda pid: "configured-model")
def _wrap_render(chat) -> list:
"""Wrap the panel's renderer registry.render to count calls."""
calls: list[dict] = []
original_render = chat._registry.render
def _record(msg, *args, **kwargs):
calls.append(msg)
return original_render(msg, *args, **kwargs)
chat._registry.render = _record # type: ignore[method-assign]
return calls
def _item_widgets(chat) -> list:
"""The panel's mounted item widgets, in display (insertion) order."""
return list(chat._widgets.values())
@pytest.mark.anyio
async def test_per_token_refresh_renders_only_the_streaming_item() -> None:
"""A stream_delta changes only the assistant bubble's signature, so the
registry is called exactly once per token — every other visible item
widget is left untouched instead of being rebuilt/re-parsed."""
async with NaviCodeTui(new_session=True).run_test() as pilot:
await pilot.pause()
chat = pilot.app.query_one("ChatPanel")
calls = _wrap_render(chat)
chat.add_user_message("hi") # renders the user bubble once
chat.handle_ws_event({"type": "stream_start"})
chat.handle_ws_event({"type": "stream_delta", "delta": "a"}) # creates + renders assistant
baseline = len(calls)
# Two more tokens — only the streaming assistant widget should re-render.
chat.handle_ws_event({"type": "stream_delta", "delta": "b"})
after_one = len(calls)
chat.handle_ws_event({"type": "stream_delta", "delta": "c"})
after_two = len(calls)
assert after_one - baseline == 1
assert after_two - baseline == 2
@pytest.mark.anyio
async def test_per_token_refresh_keeps_other_widgets_stable(monkeypatch) -> None:
"""Re-rendering the streaming widget must NOT touch the other item widgets:
the same widget objects persist across tokens (no re-mount of older items)."""
async with NaviCodeTui(new_session=True).run_test() as pilot:
await pilot.pause()
chat = pilot.app.query_one("ChatPanel")
monkeypatch.setattr(chat, "_max_visible_items", lambda: 50)
chat.clear()
chat.add_user_message("hi")
chat.handle_ws_event({"type": "stream_start"})
chat.handle_ws_event({"type": "stream_delta", "delta": "a"})
await pilot.pause()
widgets_before = _item_widgets(chat)
user_widget = widgets_before[0]
assistant_widget = widgets_before[-1]
assert user_widget._item.kind == "user_message"
assert assistant_widget._item.kind == "assistant_message"
# Two more tokens — the user widget must be the same object; only the
# assistant widget re-renders.
chat.handle_ws_event({"type": "stream_delta", "delta": "b"})
chat.handle_ws_event({"type": "stream_delta", "delta": "c"})
await pilot.pause()
widgets_after = _item_widgets(chat)
assert widgets_after[0] is user_widget, "user bubble widget must not be re-mounted"
assert widgets_after[-1] is assistant_widget, "assistant widget updated in place, not re-mounted"
assert assistant_widget._item.content == "abc"
@pytest.mark.anyio
async def test_window_cap_mounts_only_last_n_with_hint(monkeypatch) -> None:
"""Beyond the cap, only the last N item widgets stay mounted and a hint is shown."""
async with NaviCodeTui(new_session=True).run_test() as pilot:
await pilot.pause()
chat = pilot.app.query_one("ChatPanel")
monkeypatch.setattr(chat, "_max_visible_items", lambda: 3)
chat.clear() # drop the startup cwd/connected status items
for i in range(5):
chat.add_user_message(f"msg {i}")
await pilot.pause()
widgets = _item_widgets(chat)
# Only the last 3 items are mounted.
assert len(widgets) == 3
assert [w._item.content for w in widgets] == ["msg 2", "msg 3", "msg 4"]
# Truncation hint is visible and reports the 2 dropped oldest items.
assert chat._hidden_count == 2
assert chat._hint_view.styles.display != "none"
@pytest.mark.anyio
async def test_no_hint_when_within_limit(monkeypatch) -> None:
async with NaviCodeTui(new_session=True).run_test() as pilot:
await pilot.pause()
chat = pilot.app.query_one("ChatPanel")
monkeypatch.setattr(chat, "_max_visible_items", lambda: 10)
chat.clear() # drop the startup cwd/connected status items
for i in range(3):
chat.add_user_message(f"msg {i}")
await pilot.pause()
assert len(_item_widgets(chat)) == 3
assert chat._hidden_count == 0
assert chat._hint_view.styles.display == "none"
@pytest.mark.anyio
async def test_clear_unmounts_all_item_widgets() -> None:
async with NaviCodeTui(new_session=True).run_test() as pilot:
await pilot.pause()
chat = pilot.app.query_one("ChatPanel")
chat.add_user_message("hi")
chat.handle_ws_event({"type": "stream_start"})
chat.handle_ws_event({"type": "stream_delta", "delta": "a"})
await pilot.pause()
assert chat._widgets # item widgets are mounted
chat.clear()
await pilot.pause()
assert chat._widgets == {}
assert chat._model.items == []
@pytest.mark.anyio
async def test_removed_item_is_unmounted() -> None:
"""stream_end purges empty assistant bubbles; their widget must be removed
so only live items stay mounted."""
async with NaviCodeTui(new_session=True).run_test() as pilot:
await pilot.pause()
chat = pilot.app.query_one("ChatPanel")
chat.handle_ws_event({"type": "stream_start"})
chat.handle_ws_event({"type": "stream_delta", "delta": ""}) # empty bubble
await pilot.pause()
empty_ids = {id(it) for it in chat._model.items if it.kind == "assistant_message"}
assert empty_ids # the empty bubble exists before stream_end
chat.handle_ws_event({"type": "stream_end", "content": ""})
await pilot.pause()
mounted_ids = set(chat._widgets.keys())
assert not (mounted_ids & empty_ids)
# And the empty bubble is no longer in the model.
assert not any(it.kind == "assistant_message" for it in chat._model.items)
# ─── load_history (session resume) ─────────────────────────────────────────
def _hist_msg(**kw) -> dict:
"""A persisted Message dict as GET /sessions/{id} returns it."""
base = {"is_display": True}
base.update(kw)
return base
# ─── turn_meta (request duration) ───────────────────────────────────────────
def test_chat_model_stream_end_appends_turn_meta() -> None:
"""stream_end records the whole-turn duration as a metadata item."""
from clients.terminal.tui.chat_model import ChatModel
model = ChatModel()
model.handle_ws_event({"type": "stream_delta", "delta": "done"})
model.handle_ws_event({"type": "stream_end", "content": "done", "elapsed_seconds": 136})
meta = [it for it in model.items if it.kind == "turn_meta"]
assert len(meta) == 1
assert meta[0].meta["elapsed_seconds"] == 136
# It sits below the assistant answer.
assert model.items[-1].kind == "turn_meta"
def test_chat_model_stream_end_without_elapsed_still_appends() -> None:
"""If the backend omits elapsed_seconds, the metadata item still lands (renders —)."""
from clients.terminal.tui.chat_model import ChatModel
model = ChatModel()
model.handle_ws_event({"type": "stream_end", "content": ""})
meta = [it for it in model.items if it.kind == "turn_meta"]
assert len(meta) == 1
assert meta[0].meta["elapsed_seconds"] is None
def test_chat_model_stream_stopped_does_not_append_turn_meta() -> None:
"""A user-interrupted turn records a status line, not a duration metadata line."""
from clients.terminal.tui.chat_model import ChatModel
model = ChatModel()
model.handle_ws_event({"type": "stream_stopped"})
assert not any(it.kind == "turn_meta" for it in model.items)
assert any(it.kind == "status" for it in model.items)
def test_chat_model_load_history_maps_roles() -> None:
from clients.terminal.tui.chat_model import ChatModel
messages = [
_hist_msg(role="user", content="hello"),
_hist_msg(role="assistant", thinking="hmm", content=""),
_hist_msg(
role="assistant",
content="",
tool_calls=[{"id": "1", "name": "filesystem", "arguments": {"path": "/x"}}],
),
_hist_msg(role="tool", name="filesystem", content="listed: a, b"),
_hist_msg(role="assistant", content="X has a and b."),
]
model = ChatModel()
model.load_history(messages)
kinds = [it.kind for it in model.items]
assert kinds == [
"user_message",
"thinking_block",
"tool_started",
"tool_call",
"assistant_message",
]
assert model.items[0].content == "hello"
assert model.items[1].content == "hmm"
assert model.items[2].meta["tool"] == "filesystem"
assert model.items[2].meta["args"] == {"path": "/x"}
assert model.items[3].meta["result"] == "listed: a, b"
assert model.items[4].content == "X has a and b."
def test_chat_model_load_history_skips_non_display() -> None:
"""Context-only user message, summaries, compression events are not shown."""
from clients.terminal.tui.chat_model import ChatModel
messages = [
_hist_msg(role="user", content="display me", is_display=True),
_hist_msg(role="user", content="context only", is_display=False),
_hist_msg(role="assistant", content="summary", is_display=False, is_summary=True),
_hist_msg(role="assistant", content="real answer"),
]
model = ChatModel()
model.load_history(messages)
contents = [(it.kind, it.content) for it in model.items]
assert contents == [("user_message", "display me"), ("assistant_message", "real answer")]
def test_chat_model_load_history_plan_block() -> None:
from clients.terminal.tui.chat_model import ChatModel
messages = [
_hist_msg(role="assistant", content="1. step\n2. step", is_plan=True),
]
model = ChatModel()
model.load_history(messages)
assert model.items[0].kind == "plan_ready"
assert "step" in model.items[0].content
def test_chat_model_load_history_clears_existing() -> None:
"""Loading history replaces any items already in the model (session switch)."""
from clients.terminal.tui.chat_model import ChatModel
model = ChatModel()
model.add_user_message("old")
assert len(model.items) == 1
model.load_history([_hist_msg(role="user", content="new")])
assert len(model.items) == 1
assert model.items[0].content == "new"
@pytest.mark.anyio
async def test_chat_panel_load_history_renders(monkeypatch) -> None:
from rich.console import Console
async with NaviCodeTui(new_session=True).run_test() as pilot:
await pilot.pause()
chat = pilot.app.query_one("ChatPanel")
chat.clear()
chat.load_history(
[
_hist_msg(role="user", content="hi there"),
_hist_msg(role="assistant", content="hello back"),
]
)
await pilot.pause()
widgets = _item_widgets(chat)
# Two item widgets mounted, no truncation hint.
assert len(widgets) == 2
assert chat._hidden_count == 0
# Items are mounted in order: user first, then assistant.
assert [w._item.kind for w in widgets] == ["user_message", "assistant_message"]
# Each widget carries the rendered bubble for its item; re-rendering the
# item through the panel's own registry reproduces what's on screen.
from clients.terminal.tui.widgets.chat_panel import _item_msg
console = Console(record=True, width=80, force_terminal=True, color_system=None)
for widget in widgets:
console.print(chat._registry.render(_item_msg(widget._item)))
rendered = console.export_text(clear=True)
assert "hi there" in rendered
assert "hello back" in rendered
