fix(cli): a resize mid-stream repaints each transcript line once

The viewport-sized replay (#95375) still duplicated rows when the width
changed while a reply streamed: the review's e2e cell showed the warmup
reply twice after two widens.

- A widen leaves the transcript alone. It wraps nothing into extra rows,
  so prompt_toolkit's own stale-cursor erase still covers the chrome, and
  no replay size is right for both terminals that keep rows in place
  (xterm) and reflowing ones. The 3J rebuild opt-in still rebuilds.
- Output is recorded in _OUTPUT_HISTORY when it is painted, not when a
  worker requests it: a replay no longer prints lines still queued for
  the app loop (they printed again right after).
- Each recorded line carries the terminal width it was painted at, and
  the replay budget counts it at that width: a line painted after the
  terminal shrank but before the debounced handler ran wraps into more
  rows than the old width says, which pushed the budget into scrollback.
- The room above the chrome is the renderer's last paint (it can be
  taller than the preferred height), not a re-fit of the future chrome.
- A line taller than the room keeps its bottom rows (_ansi_drop_cells)
  instead of vanishing.
- The two blank separator lines of every turn go through _cprint, so
  the replay budget sees them (print() through patch_stdout bypassed it).

Tests: widen skips the replay; worker prints enter the history when
painted, tagged with the width; the tail fit counts painted widths and
keeps a tall line's bottom; the width-change test asserts the fit.

cli.py

@@ -110,6 +110,8 @@ from hermes_cli.cli_render import (  # noqa: F401,E402
     _maybe_remap_for_light_mode,
     _output_history_recording,
     _output_tail_fitting,
+    _painted_columns,
+    _PaintedLine,
     _panel_box_width,
     _post_stream_transform_output,
     _prepend_note_to_message,
@@ -668,12 +670,16 @@ def _replay_output_history(fit=None) -> None:
                     continue
                 if isinstance(lines, str):
                     lines = lines.splitlines()
-            rendered_lines.extend(str(line) for line in lines)
+            rendered_lines.extend(line if isinstance(line, str) else str(line) for line in lines)
         if fit is not None:
             rendered_lines = _output_tail_fitting(rendered_lines, *fit)
         if rendered_lines:
             # One payload: per-line pt prints each force a sync redraw (a waterfall of old output).
             _pt_print(_PT_ANSI("\n".join(rendered_lines)))
+            width = _painted_columns()
+            for line in rendered_lines:  # repainted: they wrap at today's width from now on
+                if isinstance(line, _PaintedLine):
+                    line.width = width
     except Exception:
         pass
     finally:

hermes_cli/cli_chat_turn_mixin.py

@@ -93,7 +93,7 @@ class CLIChatTurnMixin:
             message = str(message)  # UI metadata is on the staged row, never in model content.
 
         ChatConsole().print(f"[{_accent_hex()}]{'─' * 40}[/]")
-        print(flush=True)
+        _cprint("")
 
         from agent.notification_presentation import notification_config_snapshot, notification_policy_snapshot
         with notification_policy_snapshot(agent, "cli", notification_config_snapshot()):

hermes_cli/cli_render.py

@@ -491,21 +491,79 @@ def _record_output_history_entry(entry) -> None:
         _cli()._OUTPUT_HISTORY.append(entry)
 
 
-def _record_output_history(text: str) -> None:
+class _PaintedLine(str):
+    """A recorded output line tagged with the terminal ``width`` it was painted at: a terminal
+    that does not reflow keeps the rows it wrapped into then, whatever the width is now."""
+    width = None
+
+
+def _painted_columns():
+    """The width the terminal soft-wraps a print at right now, or ``None``."""
+    try:
+        from prompt_toolkit.application import get_app_or_none
+        app = get_app_or_none()
+        if app is not None:
+            return app.output.get_size().columns
+    except Exception:
+        pass
+    try:
+        return os.get_terminal_size(sys.__stdout__.fileno()).columns
+    except (AttributeError, OSError, ValueError):
+        return None
+
+
+def _record_output_history(text: str, *, force: bool = False) -> None:
+    """Record ``text`` as painted now. ``force`` skips the recording check when the caller
+    made it at print-request time (the print itself was deferred to the app loop)."""
     from cli import _output_history_recording
-    if _output_history_recording():
-        _cli()._OUTPUT_HISTORY.extend(str(text).replace("\r", "").rstrip("\n").splitlines())
+    if force or _output_history_recording():
+        width = _painted_columns()
+        lines = []
+        # One entry per printed line: ``_pt_print`` ends every text with a newline, so "" and a
+        # trailing "\n" are blank rows on screen and must count in the replay's row budget.
+        for line in str(text).replace("\r", "").split("\n"):
+            line = _PaintedLine(line)
+            line.width = width
+            lines.append(line)
+        _cli()._OUTPUT_HISTORY.extend(lines)
+
+
+_ANSI_SEQUENCE_RE = re.compile(r"\x1b(?:\[[0-?]*[ -/]*[@-~]|\][^\x07\x1b]*(?:\x07|\x1b\\)|[@-Z\\-_])")
+
+
+def _ansi_drop_cells(line: str, cells: int) -> str:
+    """``line`` without its first ``cells`` visible cells; escape sequences are kept for their styling."""
+    from prompt_toolkit.utils import get_cwidth
+    out, pos = [], 0
+    for match in [*_ANSI_SEQUENCE_RE.finditer(line), None]:
+        end = match.start() if match else len(line)
+        for ch in line[pos:end]:
+            if cells > 0:
+                cells -= get_cwidth(ch)
+            else:
+                out.append(ch)
+        if match:
+            out.append(match.group())
+            pos = match.end()
+    return "".join(out)
 
 
 def _output_tail_fitting(lines: list[str], max_rows: int, columns: int) -> list[str]:
-    """Newest ``lines`` whose soft-wrapped height at ``columns`` fits in ``max_rows``."""
+    """Newest ``lines`` filling ``max_rows`` rows, each soft-wrapped at the width it was painted
+    at (``columns`` when unknown). The oldest one may only partly fit: its bottom rows are
+    kept, the rows above them are already in scrollback."""
     from prompt_toolkit.formatted_text import ANSI, fragment_list_width, to_formatted_text
     kept, used = [], 0
     for line in reversed(lines):
-        width = fragment_list_width(to_formatted_text(ANSI(line)))
-        used += max(1, -(-width // columns)) if columns > 0 else 1
-        if used > max_rows:
+        if used >= max_rows:
             break
+        cols = getattr(line, "width", None) or columns
+        width = fragment_list_width(to_formatted_text(ANSI(line)))
+        height = max(1, -(-width // cols)) if cols > 0 else 1
+        if used + height > max_rows:
+            kept.append(_ansi_drop_cells(line, (height - (max_rows - used)) * cols))
+            break
+        used += height
         kept.append(line)
     kept.reverse()
     return kept
@@ -528,13 +586,24 @@ def _cprint(text: str):
     From a background thread while an Application runs, a direct print races the input
     redraw and gets buried, so those go through ``run_in_terminal`` via ``call_soon_threadsafe``.
     """
-    from cli import _PT_ANSI, _pt_print, _pt_print_ansi, _record_output_history
-    _record_output_history(text)
+    from cli import _PT_ANSI, _output_history_recording, _pt_print, _pt_print_ansi, _record_output_history
+    recording = _output_history_recording()
+
+    def _painted(paint):
+        # Recorded when painted, not when requested: a redraw replaying the history must
+        # neither print rows still queued for the loop nor size them at a stale width.
+        def _paint():
+            if recording:
+                _record_output_history(text, force=True)
+            paint()
+        return _paint
+    paint_pt = _painted(lambda: _pt_print(_PT_ANSI(text)))
+    paint_fallback = _painted(lambda: _pt_print_ansi(text))
 
     try:
         from prompt_toolkit.application import get_app_or_none, run_in_terminal
     except Exception:
-        _pt_print(_PT_ANSI(text))
+        paint_pt()
         return
 
     try:
@@ -543,7 +612,7 @@ def _cprint(text: str):
         app = None
 
     if app is None or not getattr(app, "_is_running", False):
-        _pt_print_ansi(text)
+        paint_fallback()
         return
 
     import asyncio as _asyncio
@@ -561,7 +630,7 @@ def _cprint(text: str):
     except Exception:
         current_loop = None
     if loop is None or (current_loop is loop and loop.is_running()):
-        _pt_print(_PT_ANSI(text))
+        paint_pt()
         return
 
     def _schedule():
@@ -570,14 +639,14 @@ def _cprint(text: str):
         # Never fall back to a bare print on error: the sync path already printed.
         with suppress(Exception):
             import inspect as _inspect
-            coro = run_in_terminal(lambda: _pt_print(_PT_ANSI(text)))
+            coro = run_in_terminal(paint_pt)
             if coro is not None and (_inspect.isawaitable(coro) or _inspect.iscoroutine(coro)):
                 _asyncio.ensure_future(coro)
 
     try:
         loop.call_soon_threadsafe(_schedule)
     except Exception:
-        _pt_print_ansi(text)
+        paint_fallback()
 
 
 def _prepend_note_to_message(message, note: str):

hermes_cli/cli_terminal_mixin.py

@@ -186,15 +186,16 @@ class CLITerminalMixin:
             pass
         self._force_full_redraw()
 
-    def _clear_prompt_toolkit_screen(self, app, *, rebuild_scrollback: bool = False):
+    def _clear_prompt_toolkit_screen(self, app, *, rebuild_scrollback: bool = False, painted_width=None):
         """Clear the terminal and reset prompt_toolkit renderer state.
 
-        Returns ``(rows, columns)`` of transcript room above the prompt chrome for the
-        replay that follows, or ``None`` when the whole history should be replayed
-        (scrollback wiped by CSI 3J, or the clear failed). Without 3J the older transcript
-        stays in scrollback, so the replay may only refill the viewport (#95375); the
-        viewport is erased row by row because CSI 2J makes scroll-on-clear terminals
-        (tmux, VTE) copy the whole screen into scrollback first, stacking a duplicate.
+        Returns ``(rows, columns)`` for the replay that follows: the viewport rows above the
+        prompt chrome and ``painted_width``, the width the transcript on screen was painted
+        at (default: the current one). ``None`` means replay the whole history (scrollback
+        wiped by CSI 3J, or the clear failed). Without 3J the older transcript stays in
+        scrollback, so the replay may only repaint what the viewport held (#95375); the
+        viewport is erased row by row because CSI 2J makes scroll-on-clear terminals (tmux,
+        VTE) copy the whole screen into scrollback first, stacking a duplicate.
         """
         if getattr(self, "_terminal_io_broken", False):
             return None
@@ -210,12 +211,10 @@ class CLITerminalMixin:
                 except Exception:
                     pass
             else:
-                size = out.get_size()
-                for row in range(size.rows):
+                fit = self._transcript_room(app, painted_width)
+                for row in range(out.get_size().rows):
                     out.cursor_goto(row, 0)
                     out.erase_end_of_line()
-                chrome = app.layout.container.preferred_height(size.columns, size.rows).preferred
-                fit = (max(0, size.rows - chrome), size.columns)
             out.cursor_goto(0, 0)
             out.flush()
             # Drop cached screen + cursor state so the next _redraw() starts from a
@@ -229,6 +228,21 @@ class CLITerminalMixin:
             pass
         return None
 
+    @staticmethod
+    def _transcript_room(app, columns=None):
+        """``(rows, columns)``: viewport rows above the prompt chrome, measured at ``columns``
+        (default: the current width).
+
+        The chrome height is the renderer's last paint: prompt_toolkit may draw the app
+        taller than its preferred height (it fills the rows below the cursor it measured).
+        """
+        renderer = app.renderer
+        size = renderer.output.get_size()
+        screen = renderer._last_screen
+        drawn = (screen.height if screen is not None
+                 else app.layout.container.preferred_height(size.columns, size.rows).preferred)
+        return max(0, size.rows - max(renderer._min_available_height, drawn)), columns or size.columns
+
     def _recover_after_resize(self, app, original_on_resize) -> None:
         """Recover a resized classic CLI without desynchronizing cursor state.
 
@@ -239,8 +253,13 @@ class CLITerminalMixin:
         already-painted rows into scrollback first, so a fresh bar looks duplicated
         (#19280, #22976). Suppression cannot erase the already-reflowed OLD bar
         (``renderer.erase()`` uses ``_cursor_pos.y`` cached at the OLD width), so on an
-        OBSERVED width change we wipe the viewport (banner-safe; 3J only via
-        ``display.cli_rebuild_scrollback_on_redraw``) and replay the transcript first.
+        OBSERVED column shrink we wipe the viewport (banner-safe) and replay what it held
+        first. A widen re-wraps nothing into extra rows, so the stale-cursor erase still
+        covers the chrome and the transcript is left as the terminal shows it: a replay there
+        duplicates rows on terminals that keep them in place (xterm) and cannot be sized
+        right for both those and reflowing ones (#95375). With
+        ``display.cli_rebuild_scrollback_on_redraw`` (3J + whole-history replay) every width
+        change rebuilds.
         Same-width SIGWINCH (tmux attach, GNOME tab bar, focus) and the first signal
         without a seeded baseline are left alone — 2J+replay against preserved scrollback
         duplicates ``_OUTPUT_HISTORY`` (#65293). tmux-attach's stale previous_screen is
@@ -252,23 +271,23 @@ class CLITerminalMixin:
         if getattr(getattr(self, '_subagent_monitor', None), 'opening', False):
             return
         from cli import _replay_output_history
+        self._status_bar_suppressed_after_resize = True
         try:
             new_width = self._get_tui_terminal_width()
         except Exception:
             new_width = None
         prev_width = getattr(self, "_last_resize_width", None)
         width_changed = new_width is not None and prev_width is not None and new_width != prev_width
-        if width_changed:
-            # Budget the replay against the full chrome: the bar/rules hidden while the
-            # reflow settles come back and would push the top replayed rows into scrollback.
-            self._status_bar_suppressed_after_resize = False
+        rebuild = self._redraw_rebuilds_scrollback()
+        if width_changed and (rebuild or new_width < prev_width):
             try:
+                # Rows are counted at the width they were painted at (a terminal that does
+                # not reflow still shows them, truncated); the OLD width sizes the rest.
                 fit = self._clear_prompt_toolkit_screen(
-                    app, rebuild_scrollback=self._redraw_rebuilds_scrollback())
+                    app, rebuild_scrollback=rebuild, painted_width=prev_width)
                 _replay_output_history(fit)
             except Exception:
                 pass
-        self._status_bar_suppressed_after_resize = True
         if new_width is not None:
             self._last_resize_width = new_width
         if width_changed:

hermes_cli/cli_tui_runtime_mixin.py

@@ -108,7 +108,7 @@ class CLITuiRuntimeMixin:
 
         if isinstance(user_input, str) and _PASTE_REF_RE.search(user_input):
             user_input = self._expand_paste_references(user_input)
-        print()
+        _cprint("")
         self._print_user_message_preview(notification_preview or user_input)
 
         if submit_images:

tests/hermes_cli/test_cli_force_redraw.py

@@ -24,12 +24,17 @@ def bare_cli():
     return cli
 
 
-def _fake_app(*, rows, columns, chrome):
-    """MagicMock app whose output reports a real size and whose layout is ``chrome`` rows tall."""
+def _fake_app(*, rows, columns, chrome, painted=None):
+    """MagicMock app whose output reports a real size and whose layout is ``chrome`` rows tall;
+    ``painted`` is the height of the renderer's last paint (``None``: nothing painted yet)."""
+    from types import SimpleNamespace
+
     from prompt_toolkit.data_structures import Size
 
     app = MagicMock()
     app.renderer.output.get_size.return_value = Size(rows=rows, columns=columns)
+    app.renderer._last_screen = None if painted is None else SimpleNamespace(height=painted)
+    app.renderer._min_available_height = 0
     app.layout.container.preferred_height.return_value.preferred = chrome
     return app
 
@@ -44,16 +49,20 @@ class TestForceFullRedraw:
 
 
     def test_resize_recovery_clears_viewport_on_width_change(self, bare_cli, monkeypatch):
-        """A WIDTH change must wipe the visible viewport and replay.
+        """A column shrink must wipe the visible viewport and replay what it held.
 
         On column shrink the terminal reflows the old full-width chrome into
         extra rows that prompt_toolkit's stale-cursor erase cannot reach,
         leaving a duplicated status bar (#19280/#5474 class). We route through
         the same recovery as Ctrl+L: erase the viewport + replay transcript.
-        It must be banner-safe — CSI 3J (write_raw) must NOT fire.
+        It must be banner-safe — CSI 3J (write_raw) must NOT fire. The replay
+        refills exactly the rows the transcript had on screen: the terminal
+        height minus the chrome as last painted (here taller than its preferred
+        height), at the width those rows were painted at (#95375).
         """
-        app = _fake_app(rows=30, columns=90, chrome=5)
+        app = _fake_app(rows=30, columns=90, chrome=5, painted=7)
         events = []
+        fits = []
         app.renderer.output.erase_end_of_line.side_effect = lambda: events.append("erase")
         app.renderer.output.write_raw.side_effect = lambda *_: events.append("scrollback_wipe")
         original_on_resize = lambda: events.append("original_resize")
@@ -62,7 +71,8 @@ class TestForceFullRedraw:
         bare_cli._last_resize_width = 200
         monkeypatch.setattr(bare_cli, "_get_tui_terminal_width", lambda: 90)
         monkeypatch.setattr(bare_cli, "_schedule_status_bar_unsuppress", lambda *_: None)
-        monkeypatch.setattr(cli_mod, "_replay_output_history", lambda *_: events.append("replay"))
+        monkeypatch.setattr(
+            cli_mod, "_replay_output_history", lambda fit=None: (events.append("replay"), fits.append(fit)))
         monkeypatch.setattr(
             cli_mod,
             "CLI_CONFIG",
@@ -75,6 +85,7 @@ class TestForceFullRedraw:
         assert "erase" in events
         assert "replay" in events
         assert events.index("erase") < events.index("original_resize")
+        assert fits == [(30 - 7, 200)]
         # Banner-safe: scrollback (CSI 3J) must never be wiped on a resize.
         assert "scrollback_wipe" not in events
         # New width recorded for the next comparison.
@@ -144,6 +155,26 @@ class TestForceFullRedraw:
         assert events[:3] == ["erase", "scrollback_wipe", "replay"]
         assert events.index("scrollback_wipe") < events.index("original_resize")
 
+    def test_widen_leaves_the_transcript_in_place(self, bare_cli, monkeypatch):
+        """#95375: a widen wraps nothing into extra rows, so prompt_toolkit's own erase still
+        covers the chrome. A replay would print again rows the terminal still shows."""
+        app = _fake_app(rows=24, columns=132, chrome=5, painted=5)
+        events = []
+        app.renderer.output.erase_end_of_line.side_effect = lambda: events.append("erase")
+        app.renderer.output.erase_screen.side_effect = lambda: events.append("erase")
+        original_on_resize = lambda: events.append("original_resize")
+
+        bare_cli._last_resize_width = 116
+        monkeypatch.setattr(bare_cli, "_get_tui_terminal_width", lambda: 132)
+        monkeypatch.setattr(bare_cli, "_schedule_status_bar_unsuppress", lambda *_: None)
+        monkeypatch.setattr(cli_mod, "_replay_output_history", lambda *_: events.append("replay"))
+        monkeypatch.setattr(cli_mod, "CLI_CONFIG", {"display": {"cli_rebuild_scrollback_on_redraw": False}})
+
+        bare_cli._recover_after_resize(app, original_on_resize)
+
+        assert events == ["original_resize"]
+        assert bare_cli._last_resize_width == 132
+
     def test_same_width_sigwinch_is_left_untouched(self, bare_cli, monkeypatch):
         """Same-width SIGWINCH (tmux attach, benign focus/tab signals) must not
         clear the viewport or replay: a 2J without replay erases the visible
@@ -429,6 +460,20 @@ class TestReplayFitsViewport:
         assert printed[0].split("\n") == [f"history line {i}" for i in range(192, 200)] + ["x" * 150]
         assert len(cli_mod._OUTPUT_HISTORY) == 200  # the buffer itself is untouched
 
+    def test_tail_counts_rows_at_the_painted_width_and_keeps_a_tall_lines_bottom(self):
+        """A line keeps the rows it wrapped into when painted (a terminal that does not reflow
+        never re-wraps it), and a line taller than the room keeps its bottom rows instead of
+        vanishing: the rows above them are what already scrolled into scrollback."""
+        tall = "".join(chr(ord("a") + i) * 10 for i in range(6))  # 60 cells = 6 rows at 10 cols
+        assert cli_mod._output_tail_fitting(["older", tall], 2, 10) == ["eeeeeeeeeeffffffffff"]
+        styled = f"\x1b[1m{tall}\x1b[0m"
+        assert cli_mod._output_tail_fitting([styled], 2, 10) == ["\x1b[1meeeeeeeeeeffffffffff\x1b[0m"]
+
+        from hermes_cli.cli_render import _PaintedLine
+        narrow = _PaintedLine("n" * 150)
+        narrow.width = 50  # painted before the terminal widened to 100: 3 rows, not 2
+        assert cli_mod._output_tail_fitting(["older", narrow, "last"], 4, 100) == [narrow, "last"]
+
 
 class TestFocusRegainRedraw:
     """Focus-in (CSI I) routes through the same recovery as Ctrl+L, rate-limited.

tests/hermes_cli/test_cprint_bg_thread.py

@@ -142,6 +142,40 @@ def test_cprint_swallows_prompt_toolkit_import_error(monkeypatch):
 
 
 
+def test_cprint_from_a_worker_is_recorded_when_painted(monkeypatch):
+    """#95375: a worker's print is queued for the app loop. A resize replay that runs before
+    it must not find it in the history (it would print it, then the queued print again), and
+    the row is tagged with the width the terminal wraps it at when it is painted."""
+    cli._configure_output_history(True, 10)
+    painted = []
+    monkeypatch.setattr(cli, "_pt_print", lambda x: painted.append(x))
+    monkeypatch.setattr(cli, "_PT_ANSI", lambda t: t)
+    queued = []
+
+    class FakeLoop:
+        def is_running(self):
+            return True
+
+        def call_soon_threadsafe(self, cb, *args):
+            queued.append(cb)
+
+    fake_app = SimpleNamespace(
+        _is_running=True, loop=FakeLoop(),
+        output=SimpleNamespace(get_size=lambda: SimpleNamespace(columns=77)))
+    fake_pt_app = types.ModuleType("prompt_toolkit.application")
+    fake_pt_app.get_app_or_none = lambda: fake_app
+    fake_pt_app.run_in_terminal = lambda fn, **kw: fn()
+    monkeypatch.setitem(sys.modules, "prompt_toolkit.application", fake_pt_app)
+
+    cli._cprint("streamed line")  # not on the app loop: no running loop in this thread
+
+    assert painted == [] and list(cli._OUTPUT_HISTORY) == []
+    queued.pop()()
+    assert painted == ["streamed line"]
+    assert list(cli._OUTPUT_HISTORY) == ["streamed line"]
+    assert cli._OUTPUT_HISTORY[0].width == 77
+
+
 def test_replay_output_history_rerenders_callable_entries(monkeypatch):
     cli._configure_output_history(True, 10)
     widths_seen = []