"""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_recovers_tool_args_and_metadata() -> None:
"""Resume must reproduce the live tool_call payload: the tool message in
storage carries only the result, so load_history recovers the call args via
the assistant message's tool_call_id and forwards the stored metadata.
Without this the filesystem renderer sees args={} → action=None and falls
back to a plain (un-highlighted) result — the resume "no syntax highlighting"
symptom."""
from clients.terminal.tui.chat_model import ChatModel
messages = [
_hist_msg(
role="assistant",
content="",
tool_calls=[{"id": "call-7", "name": "filesystem", "arguments": {"action": "edit", "path": "/a.py"}}],
),
_hist_msg(
role="tool",
name="filesystem",
tool_call_id="call-7",
content="replaced",
metadata={"diff": "- 2│ old\n+ 2│ new"},
),
]
model = ChatModel()
model.load_history(messages)
tool_call = [it for it in model.items if it.kind == "tool_call"][0]
assert tool_call.meta["tool"] == "filesystem"
assert tool_call.meta["args"] == {"action": "edit", "path": "/a.py"}
assert tool_call.meta["result"] == "replaced"
assert tool_call.meta["metadata"] == {"diff": "- 2│ old\n+ 2│ new"}
def test_chat_model_load_history_tool_args_fallback_when_id_missing() -> None:
"""If the tool message has no tool_call_id (or it doesn't match a stored
call), args fall back to {} rather than raising — the renderer then degrades
to the plain result, which is acceptable for legacy/foreign sessions."""
from clients.terminal.tui.chat_model import ChatModel
messages = [
_hist_msg(role="tool", name="filesystem", content="listed"),
]
model = ChatModel()
model.load_history(messages)
tool_call = [it for it in model.items if it.kind == "tool_call"][0]
assert tool_call.meta["args"] == {}
assert tool_call.meta["metadata"] == {}
def test_chat_model_load_history_emits_assistant_text_before_tool_started() -> None:
"""When an assistant message carries BOTH text and tool_calls, the text
bubble must land ABOVE the tool cards — matching the live stream where
stream_delta precedes tool_started. The previous code emitted tool_started
first, stranding the answer under its own tool calls on resume."""
from clients.terminal.tui.chat_model import ChatModel
messages = [
_hist_msg(
role="assistant",
content="Let me check the file.",
tool_calls=[{"id": "1", "name": "filesystem", "arguments": {"action": "read"}}],
),
_hist_msg(role="tool", name="filesystem", tool_call_id="1", content="ok"),
]
model = ChatModel()
model.load_history(messages)
kinds = [it.kind for it in model.items]
assert kinds == ["assistant_message", "tool_started", "tool_call"]
assert model.items[0].content == "Let me check the file."
def test_chat_model_load_history_appends_turn_meta_for_elapsed() -> None:
"""The whole-turn duration is stamped on the assistant message that closed
the run; load_history reproduces the single turn_meta line stream_end
appends. Intermediate tool-tour assistant messages (no elapsed_seconds) do
not produce extra turn_meta items."""
from clients.terminal.tui.chat_model import ChatModel
messages = [
_hist_msg(
role="assistant",
content="",
tool_calls=[{"id": "1", "name": "filesystem", "arguments": {}}],
),
_hist_msg(role="tool", name="filesystem", tool_call_id="1", content="ok"),
_hist_msg(role="assistant", content="done", elapsed_seconds=42.5),
]
model = ChatModel()
model.load_history(messages)
metas = [it for it in model.items if it.kind == "turn_meta"]
assert len(metas) == 1
assert metas[0].meta["elapsed_seconds"] == 42.5
# turn_meta sits below the final answer, as in the live stream.
assert model.items[-1].kind == "turn_meta"
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"
def test_chat_model_context_compressed_creates_summary_item() -> None:
"""A live context_compressed event surfaces the summary text as a distinct
chat item (for both forced /compact and mid-turn auto-compress), carrying
the before/after counts so the renderer can head the block with them."""
from clients.terminal.tui.chat_model import ChatModel
model = ChatModel()
item = model.handle_ws_event(
{
"type": "context_compressed",
"summary": "Earlier the user asked to build a TUI client.",
"messages_before": 42,
"messages_after": 8,
"context_tokens": 1200,
"max_context_tokens": 4096,
}
)
assert item is not None
assert item.kind == "context_summary"
assert item.content == "Earlier the user asked to build a TUI client."
assert item.meta["messages_before"] == 42
assert item.meta["messages_after"] == 8
assert model.items[-1] is item
@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
@pytest.mark.anyio
async def test_sync_keeps_scroll_position_when_user_scrolled_up() -> None:
"""Stick-to-bottom: when the user has scrolled up to read earlier messages,
incoming stream content must not yank the view back to the bottom on every
sync. Auto-follow resumes only after the user scrolls back to the bottom.
"""
async with NaviCodeTui(new_session=True).run_test() as pilot:
await pilot.pause()
chat = pilot.app.query_one("ChatPanel")
chat.clear()
# Fill the chat so it overflows (vertical scroll range > 0).
for i in range(30):
chat.add_user_message(f"line {i} " + "x" * 40)
chat.handle_ws_event({"type": "stream_start"})
chat.handle_ws_event({"type": "stream_delta", "delta": "y" * 40})
chat.handle_ws_event({"type": "stream_end"})
await pilot.pause()
assert chat.max_scroll_y > 0
# Freshly filled: stuck to the bottom.
assert chat.scroll_y == chat.max_scroll_y
# User scrolls up to read earlier messages.
chat.scroll_to(y=0, animate=False)
await pilot.pause()
scrolled_y = chat.scroll_y
assert scrolled_y < chat.max_scroll_y
# New content arrives while the user is scrolled up: position must be
# preserved — _sync must not call scroll_end.
chat.handle_ws_event({"type": "stream_start"})
chat.handle_ws_event({"type": "stream_delta", "delta": "new " * 20})
await pilot.pause()
assert chat.scroll_y == scrolled_y
# User scrolls back to the bottom: auto-follow resumes on the next sync.
chat.scroll_end(animate=False)
await pilot.pause()
assert chat.scroll_y == chat.max_scroll_y
chat.handle_ws_event({"type": "stream_delta", "delta": "more " * 20})
await pilot.pause()
assert chat.scroll_y == chat.max_scroll_y
# ─── text selection (copy without bubble chrome) ──────────────────────────────
def _rendered_lines_of(widget) -> list[str]:
"""The exact screen lines a _ChatItemView draws (the plain text of the
selection-aware Content its render() builds, line layout identical to the
screen), for asserting selection coordinates."""
return widget._build_content().plain.split("\n")
@pytest.mark.anyio
async def test_get_selection_returns_clean_message_content() -> None:
"""A whole-message selection returns the bubble's text without the rounded
border or the "You" title — the base Widget.get_selection returns None for
Panel renderables, so without the override drag-select copies nothing."""
from textual.geometry import Offset
from textual.selection import Selection
async with NaviCodeTui(new_session=True).run_test() as pilot:
await pilot.pause()
chat = pilot.app.query_one("ChatPanel")
chat.clear()
chat.add_user_message("Hello world")
await pilot.pause()
widget = list(chat._widgets.values())[0]
text, _ = widget.get_selection(Selection(None, None))
assert text == "Hello world"
# No panel chrome leaks into the copy.
assert "╭" not in text and "│" not in text and "You" not in text
@pytest.mark.anyio
async def test_get_selection_partial_line_aligns_to_content() -> None:
"""A partial-line selection maps the mouse column onto the content column
(skipping the ``│ `` gutter), so the copied substring is exactly the
highlighted characters — not offset by the border."""
from textual.geometry import Offset
from textual.selection import Selection
async with NaviCodeTui(new_session=True).run_test() as pilot:
await pilot.pause()
chat = pilot.app.query_one("ChatPanel")
chat.clear()
chat.add_user_message("Hello world")
await pilot.pause()
widget = list(chat._widgets.values())[0]
lines = _rendered_lines_of(widget)
# Layout sanity: top border, one content line, bottom border; content
# starts two columns in (after ``│ ``).
assert lines[0].lstrip().startswith("╭")
assert lines[1].startswith("│ Hello world")
assert lines[-1].lstrip().startswith("╰")
# Select columns 2..7 on the content line → "Hello" (H at col 2).
text, _ = widget.get_selection(Selection(Offset(x=2, y=1), Offset(x=7, y=1)))
assert text == "Hello"
@pytest.mark.anyio
async def test_get_selection_across_two_content_lines() -> None:
"""A selection spanning two content rows returns both rows' content joined,
without borders between them."""
from textual.geometry import Offset
from textual.selection import Selection
async with NaviCodeTui(new_session=True).run_test() as pilot:
await pilot.pause()
chat = pilot.app.query_one("ChatPanel")
chat.clear()
chat.add_user_message("alpha\nbeta")
await pilot.pause()
widget = list(chat._widgets.values())[0]
lines = _rendered_lines_of(widget)
content_rows = [ln for ln in lines if ln.startswith("│")]
assert len(content_rows) >= 2
text, _ = widget.get_selection(Selection(Offset(x=2, y=1), Offset(x=4, y=2)))
assert text == "alpha\nbe"
@pytest.mark.anyio
async def test_get_selection_clamps_top_border_to_content_start() -> None:
"""Starting the drag on the title/border line does not pull chrome into the
copy — it clamps to the start of the first content line."""
from textual.geometry import Offset
from textual.selection import Selection
async with NaviCodeTui(new_session=True).run_test() as pilot:
await pilot.pause()
chat = pilot.app.query_one("ChatPanel")
chat.clear()
chat.add_user_message("Hello world")
await pilot.pause()
widget = list(chat._widgets.values())[0]
text, _ = widget.get_selection(Selection(Offset(x=0, y=0), Offset(x=7, y=1)))
assert text == "Hello"
assert "You" not in text
@pytest.mark.anyio
async def test_get_selection_plain_text_widget_has_no_chrome() -> None:
"""Non-panel items (status line, turn_meta) render plain Text — selection
returns the text verbatim with no column shift."""
from textual.geometry import Offset
from textual.selection import Selection
async with NaviCodeTui(new_session=True).run_test() as pilot:
await pilot.pause()
chat = pilot.app.query_one("ChatPanel")
chat.clear()
chat.handle_ws_event({"type": "status", "content": "working"})
await pilot.pause()
widget = list(chat._widgets.values())[0]
assert widget._item.kind == "status"
lines = _rendered_lines_of(widget)
assert lines == ["• working"]
text, _ = widget.get_selection(Selection(None, None))
assert text == "• working"
@pytest.mark.anyio
async def test_get_selection_filesystem_tool_output_is_selectable() -> None:
"""Filesystem tool output renders a plain Group/Text (no outer panel), so a
selection over the result returns the content — the resume "can't copy tool
output" symptom alongside the missing highlights."""
from textual.geometry import Offset
from textual.selection import Selection
async with NaviCodeTui(new_session=True).run_test() as pilot:
await pilot.pause()
chat = pilot.app.query_one("ChatPanel")
chat.clear()
chat.handle_ws_event(
{
"type": "tool_call",
"tool": "filesystem",
"args": {"action": "read", "path": "/a.py"},
"result": "1: line one\n2: line two",
"success": True,
"metadata": {},
}
)
await pilot.pause()
widget = list(chat._widgets.values())[0]
assert widget._item.kind == "tool_call"
text, _ = widget.get_selection(Selection(None, None))
assert "line one" in text
assert "line two" in text
# Plain renderable — no panel borders anywhere.
assert "│" not in text and "╭" not in text
@pytest.mark.anyio
async def test_screen_copy_text_strips_chrome_across_messages() -> None:
"""End-to-end: Screen.get_selected_text (what Ctrl+C copies) walks every
selected widget's get_selection, so selecting across a user bubble and an
assistant bubble yields both bodies joined — no borders, no titles."""
from textual.geometry import Offset
async with NaviCodeTui(new_session=True).run_test() as pilot:
await pilot.pause()
chat = pilot.app.query_one("ChatPanel")
chat.clear()
chat.add_user_message("question")
chat.handle_ws_event({"type": "stream_start"})
chat.handle_ws_event({"type": "stream_delta", "delta": "answer"})
chat.handle_ws_event({"type": "stream_end", "content": "answer"})
await pilot.pause()
screen = pilot.app.screen
screen.text_select_all()
await pilot.pause()
text = screen.get_selected_text()
assert "question" in text
assert "answer" in text
assert "╭" not in text and "│" not in text
# ─── reading mode (borderless render for raw-terminal copy) ──────────────────
@pytest.mark.anyio
async def test_reading_mode_strips_bubble_borders_in_place() -> None:
"""set_reading_mode(True) re-renders every visible item without Panel chrome
(and back with it on False), in place — so a raw Shift+drag + Ctrl+Shift+C in
a terminal without OSC 52 copies clean text. Markdown answers keep raw
fences (no code-block Panels)."""
async with NaviCodeTui(new_session=True).run_test(size=(100, 24)) as pilot:
await pilot.pause()
chat = pilot.app.query_one("ChatPanel")
chat.clear()
chat.add_user_message("Hello world")
chat.handle_ws_event({"type": "stream_start"})
chat.handle_ws_event({"type": "stream_delta", "delta": "Answer with ```py\nx=1\n``` block"})
chat.handle_ws_event({"type": "stream_end", "content": "x"})
await pilot.pause()
widgets = list(chat._widgets.values())
w_user = next(w for w in widgets if w._item.kind == "user_message")
w_ans = next(w for w in widgets if w._item.kind == "assistant_message")
# Normal mode: panels with borders.
assert "╭" in w_user._build_content().plain
assert "╭" in w_ans._build_content().plain
# Enter reading mode: borders gone, content intact, scroll preserved.
scroll_before = chat.scroll_y
chat.set_reading_mode(True)
await pilot.pause()
user_plain = w_user._build_content().plain
ans_plain = w_ans._build_content().plain
assert "╭" not in user_plain and "│" not in user_plain
assert "Hello world" in user_plain
# Raw markdown fences preserved (no code-block Panel borders).
assert "```py" in ans_plain and "╭" not in ans_plain
assert chat.scroll_y == scroll_before
# Exit reading mode: borders return.
chat.set_reading_mode(False)
await pilot.pause()
assert "╭" in w_user._build_content().plain
assert "╭" in w_ans._build_content().plain
@pytest.mark.anyio
async def test_new_items_in_reading_mode_are_borderless() -> None:
"""Items streamed in while reading mode is on mount without bubble borders."""
async with NaviCodeTui(new_session=True).run_test(size=(100, 24)) as pilot:
await pilot.pause()
chat = pilot.app.query_one("ChatPanel")
chat.clear()
chat.set_reading_mode(True)
await pilot.pause()
chat.add_user_message("late message")
await pilot.pause()
w = list(chat._widgets.values())[0]
plain = w._build_content().plain
assert "╭" not in plain and "late message" in plain
