fix(cli): stop hard-wrapping streamed paragraphs; prefer OSC 52 over SSH

Streamed responses no longer insert real newlines at terminal width —
logical lines are emitted whole and the terminal soft-wraps them, so
highlight-copy rejoins the full line (emulators only keep linebreaks
the app actually printed). This is the CLI equivalent of the TUI's
selection copy, which reads logical source lines from its screen
buffer. TTFT perception is preserved by mirroring the unfinished
line's tail into the spinner status text instead of chunk-printing.

/copy now prefers OSC 52 when running over SSH (SSH_CONNECTION /
SSH_TTY / SSH_CLIENT) — native tools there write the REMOTE clipboard,
which is never what the user wants. The CLI's OSC 52 writer also gains
tmux/screen DCS passthrough wrapping, mirroring the TUI's
wrapForMultiplexer. Fixes #31528 for the CLI surface.

Sabotage-verified: restoring the old chunk emitter fails 3 of the new
tests (hard-wrap detection, spinner mirror, unbreakable-run split).

cli.py

@@ -2442,6 +2442,8 @@ _RST = "\033[0m"
 _STREAM_PAD = ""  # No indent for streamed response text — leading whitespace pollutes
 # terminal copy/paste (every selected line carried 4 spaces).  Matches the
 # response Panel's flush-left padding.
+_STREAM_PARTIAL_PREVIEW_LEN = 60  # tail of an unfinished logical line mirrored
+# into the spinner while streaming (TTFT perception without hard-wrapping)
 
 
 def _hex_to_ansi(hex_color: str, *, bold: bool = False) -> str:
@@ -6637,32 +6639,34 @@ class HermesCLI(CLIAgentSetupMixin, CLICommandsMixin, CLIBillingMixin):
                 line = _strip_markdown_syntax(line)
             _emit_one(line)
 
-        # Force-flush long partial lines so a response that opens with a
-        # long paragraph paints as tokens arrive instead of staying blank
-        # until the first newline (TTFT perception fix — the reasoning box
-        # has done this at 80 chars since day one; the response box never
-        # did). Wrap at the terminal's visible width so we only ever emit
-        # text that would have line-broken at that point anyway; the
-        # remainder stays buffered as the logical line's continuation.
-        # Table-shaped partials are exempt — they need the whole block for
-        # realignment (see the table side-buffer above).
+        # Long partial lines are emitted ONLY at real newlines — we no
+        # longer hard-wrap paragraphs at terminal width ourselves.  Each
+        # logical line lands in scrollback as one line; the TERMINAL
+        # soft-wraps it visually, and emulators (iTerm2/kitty/VTE/
+        # xterm.js/Windows Terminal) rejoin soft-wrapped rows on copy,
+        # so highlight-copy yields the original unwrapped text — same
+        # outcome as the TUI's selection copy.  (The pre-July-2026 chunk
+        # emitter baked real '\n's into every long paragraph, which is
+        # exactly what polluted copy/paste.)
+        #
+        # TTFT perception: while a long opening paragraph accumulates
+        # without a newline, mirror its tail into the status-bar spinner
+        # line so the user sees tokens arriving instead of a blank box.
         if (
             self._stream_buf
             and not self._in_stream_table
             and not self._stream_buf.lstrip().startswith("|")
+            and len(self._stream_buf) >= 80
         ):
-            wrap_w = max(40, _terminal_width_for_streaming())
-            while len(self._stream_buf) >= wrap_w:
-                cut = self._stream_buf.rfind(" ", 0, wrap_w)
-                if cut <= 0:
-                    cut = wrap_w  # single unbreakable run — hard wrap
-                chunk, self._stream_buf = (
-                    self._stream_buf[:cut],
-                    self._stream_buf[cut:].lstrip(" "),
-                )
-                if self.final_response_markdown == "strip":
-                    chunk = _strip_markdown_syntax(chunk)
-                _emit_one(chunk)
+            preview = self._stream_buf[-int(_STREAM_PARTIAL_PREVIEW_LEN):]
+            cut = preview.find(" ")
+            if 0 < cut < len(preview) - 1:
+                preview = preview[cut + 1:]
+            try:
+                self._spinner_text = f"… {preview}"
+                self._invalidate()
+            except Exception:
+                pass
 
     def _flush_stream(self) -> None:
         """Emit any remaining partial line from the stream buffer and close the box."""
@@ -7160,9 +7164,19 @@ class HermesCLI(CLIAgentSetupMixin, CLICommandsMixin, CLIBillingMixin):
 
 
     def _write_osc52_clipboard(self, text: str) -> None:
-        """Copy *text* to terminal clipboard via OSC 52."""
+        """Copy *text* to terminal clipboard via OSC 52.
+
+        Wrapped for tmux/screen passthrough (mirrors the TUI's
+        wrapForMultiplexer in ui-tui/src/lib/osc52.ts) — without the DCS
+        wrapper the multiplexer consumes the sequence and the copy is
+        silently lost.
+        """
         payload = base64.b64encode(text.encode("utf-8")).decode("ascii")
         seq = f"\x1b]52;c;{payload}\x07"
+        if os.environ.get("TMUX"):
+            seq = "\x1bPtmux;" + seq.replace("\x1b", "\x1b\x1b") + "\x1b\\"
+        elif os.environ.get("STY"):
+            seq = "\x1bP" + seq + "\x1b\\"
         out = getattr(self, "_app", None)
         output = getattr(out, "output", None) if out else None
         if output and hasattr(output, "write_raw"):

hermes_cli/cli_commands_mixin.py

@@ -559,7 +559,20 @@ class CLICommandsMixin:
             return
 
         try:
-            from hermes_cli.clipboard import write_clipboard_text
+            from hermes_cli.clipboard import (
+                is_remote_shell_session,
+                write_clipboard_text,
+            )
+            if is_remote_shell_session():
+                # Over SSH, native tools would write the REMOTE clipboard
+                # (or an X-forwarded one) — OSC 52 reaches the terminal
+                # the user is actually sitting at. Fixes #31528.
+                self._write_osc52_clipboard(text)
+                _cprint(
+                    f"  Copied assistant response #{idx + 1} via OSC 52 "
+                    "(terminal support required)"
+                )
+                return
             if write_clipboard_text(text):
                 _cprint(f"  Copied assistant response #{idx + 1} to clipboard")
                 return

hermes_cli/clipboard.py

@@ -84,6 +84,20 @@ def _write_clipboard_commands() -> list:
     return attempts
 
 
+def is_remote_shell_session(env=None) -> bool:
+    """True when running inside an SSH session.
+
+    Mirrors ui-tui/src/lib/terminalSetup.ts isRemoteShellSession().  Over
+    SSH, native clipboard tools write the REMOTE machine's clipboard (or
+    an X-forwarded one), which is almost never what the user wants —
+    OSC 52 reaches the LOCAL terminal emulator instead.
+    """
+    e = os.environ if env is None else env
+    return bool(
+        e.get("SSH_CONNECTION") or e.get("SSH_TTY") or e.get("SSH_CLIENT")
+    )
+
+
 def write_clipboard_text(text: str) -> bool:
     """Write *text* to the system clipboard via native platform tools.
 

tests/cli/test_cli_copy_command.py

@@ -65,12 +65,41 @@ def test_copy_falls_back_to_osc52_when_native_tools_fail():
     cli_obj.conversation_history = [{"role": "assistant", "content": "hello"}]
 
     with patch("hermes_cli.clipboard.write_clipboard_text", return_value=False), \
+         patch("hermes_cli.clipboard.is_remote_shell_session", return_value=False), \
          patch.object(cli_obj, "_write_osc52_clipboard") as mock_osc52:
         cli_obj.process_command("/copy")
 
     mock_osc52.assert_called_once_with("hello")
 
 
+def test_copy_prefers_osc52_in_ssh_sessions():
+    """Over SSH, native tools write the REMOTE clipboard — OSC 52 reaches
+    the local terminal instead (#31528)."""
+    cli_obj = _make_cli()
+    cli_obj.conversation_history = [{"role": "assistant", "content": "remote answer"}]
+
+    with patch("hermes_cli.clipboard.write_clipboard_text", return_value=True) as mock_native, \
+         patch("hermes_cli.clipboard.is_remote_shell_session", return_value=True), \
+         patch.object(cli_obj, "_write_osc52_clipboard") as mock_osc52:
+        cli_obj.process_command("/copy")
+
+    mock_osc52.assert_called_once_with("remote answer")
+    mock_native.assert_not_called()
+
+
+def test_copy_native_first_when_local():
+    cli_obj = _make_cli()
+    cli_obj.conversation_history = [{"role": "assistant", "content": "local answer"}]
+
+    with patch("hermes_cli.clipboard.write_clipboard_text", return_value=True) as mock_native, \
+         patch("hermes_cli.clipboard.is_remote_shell_session", return_value=False), \
+         patch.object(cli_obj, "_write_osc52_clipboard") as mock_osc52:
+        cli_obj.process_command("/copy")
+
+    mock_native.assert_called_once_with("local answer")
+    mock_osc52.assert_not_called()
+
+
 def test_copy_invalid_index_does_not_copy():
     cli_obj = _make_cli()
     cli_obj.conversation_history = [{"role": "assistant", "content": "only"}]

tests/cli/test_stream_partial_line_flush.py

@@ -1,10 +1,11 @@
-"""Streaming display force-flush: long partial lines must paint before the
-first newline arrives (TTFT-perception fix, July 2026).
+"""Streaming display: logical lines are emitted ONLY at real newlines.
 
-Previously ``_emit_stream_text`` only emitted on ``"\\n"``, so a response
-opening with a long paragraph stayed invisible until the model produced a
-newline — seconds of blank box on slow models. Now partial lines are
-force-flushed at terminal width (mirroring the reasoning box's 80-char rule).
+The July 2026 TTFT force-flush hard-wrapped long partial lines at
+terminal width, baking real '\\n's into every long paragraph — exactly
+what polluted highlight-copy/paste. Now paragraphs stay one logical
+line (the terminal soft-wraps them and rejoins on copy, matching the
+TUI's selection copy), and TTFT perception is served by mirroring the
+partial line's tail into the spinner status text instead.
 """
 import os
 import re
@@ -29,28 +30,50 @@ def cli_stub(monkeypatch):
     cli.final_response_markdown = "raw"
     cli.show_timestamps = False
     cli._reset_stream_state()
+    cli._spinner_text = ""
+    cli._invalidate = lambda *a, **kw: None
 
     emitted = []
     monkeypatch.setattr(climod, "_cprint", lambda s: emitted.append(s))
-    # Deterministic width regardless of the test runner's terminal
     monkeypatch.setattr(climod, "_terminal_width_for_streaming", lambda: 74)
     return cli, emitted
 
 
-class TestPartialLineForceFlush:
-    def test_long_paragraph_paints_before_first_newline(self, cli_stub):
+class TestLogicalLineStreaming:
+    def test_long_paragraph_not_hard_wrapped_before_newline(self, cli_stub):
         cli, emitted = cli_stub
         text = (
-            "This is a long opening paragraph that would previously sit "
-            "invisible in the buffer until the model finally produced a "
-            "newline character, which on a slow model could take seconds. "
+            "This is a long opening paragraph that previously got chopped "
+            "into terminal-width chunks with real newlines, which is what "
+            "made copy/paste come out full of broken lines. "
         ) * 3
         for i in range(0, len(text), 12):
             cli._stream_delta(text[i : i + 12])
-        # Box header + several wrapped lines painted with NO newline seen yet
-        assert len(emitted) > 3
+        # No newline seen yet → no content lines printed (box header only).
+        plain = _strip_ansi("\n".join(emitted))
+        assert "opening paragraph" not in plain
+        # The paragraph is still buffered as ONE logical line.
+        assert cli._stream_buf.startswith("This is a long opening")
 
-    def test_no_content_lost_across_wraps(self, cli_stub):
+    def test_partial_tail_mirrored_into_spinner(self, cli_stub):
+        cli, emitted = cli_stub
+        text = "A long paragraph streaming in without any newline " * 4
+        for i in range(0, len(text), 16):
+            cli._stream_delta(text[i : i + 16])
+        assert cli._spinner_text.startswith("…")
+        assert "newline" in cli._spinner_text
+
+    def test_logical_line_emitted_whole_at_newline(self, cli_stub):
+        cli, emitted = cli_stub
+        long_line = "word " * 60  # ~300 chars, far beyond terminal width
+        cli._stream_delta(long_line.rstrip() + "\n")
+        content = [
+            _strip_ansi(e) for e in emitted if "word" in _strip_ansi(e)
+        ]
+        assert len(content) == 1, "logical line was split across prints"
+        assert content[0] == long_line.rstrip()
+
+    def test_no_content_lost_across_stream(self, cli_stub):
         cli, emitted = cli_stub
         words = [f"word{i}" for i in range(120)]
         text = " ".join(words)
@@ -59,24 +82,22 @@ class TestPartialLineForceFlush:
         cli._flush_stream()
         plain = " ".join(_strip_ansi("\n".join(emitted)).split())
         for w in words:
-            assert w in plain, f"lost {w} at a wrap boundary"
+            assert w in plain, f"lost {w}"
 
     def test_short_partial_stays_buffered(self, cli_stub):
         cli, emitted = cli_stub
         cli._stream_delta("short line, no newline")
-        # Under wrap width: the box header may open, but the text itself
-        # stays buffered until a newline or the width threshold.
         plain = _strip_ansi("\n".join(emitted))
         assert "short line" not in plain
         assert cli._stream_buf == "short line, no newline"
 
-    def test_table_rows_not_force_flushed(self, cli_stub):
+    def test_table_rows_not_previewed_in_spinner(self, cli_stub):
         cli, emitted = cli_stub
-        # A long partial table row must stay buffered for block realignment
         row = "| " + " | ".join(f"cell{i}" for i in range(20)) + " |"
         cli._stream_delta(row)  # no newline
         plain = _strip_ansi("\n".join(emitted))
         assert "cell19" not in plain
+        assert cli._spinner_text == ""
 
     def test_newline_lines_still_emit_normally(self, cli_stub):
         cli, emitted = cli_stub
@@ -85,10 +106,14 @@ class TestPartialLineForceFlush:
         assert "line one" in plain
         assert "line two" in plain
 
-    def test_unbreakable_run_hard_wraps(self, cli_stub):
+    def test_unbreakable_run_stays_single_line(self, cli_stub):
         cli, emitted = cli_stub
         blob = "x" * 300  # no spaces
         cli._stream_delta(blob)
         cli._flush_stream()
         plain = _strip_ansi("\n".join(emitted))
         assert plain.count("x") == 300
+        content = [
+            _strip_ansi(e) for e in emitted if "x" in _strip_ansi(e)
+        ]
+        assert len(content) == 1, "unbreakable run was hard-wrapped"

tests/hermes_cli/test_clipboard_text_write.py

@@ -71,3 +71,54 @@ def test_wayland_prefers_wl_copy():
          patch.object(clip.subprocess, "run", return_value=_completed()) as run:
         assert clip.write_clipboard_text("x") is True
     assert run.call_args[0][0][0] == "wl-copy"
+
+
+def test_is_remote_shell_session_detects_ssh_env():
+    assert clip.is_remote_shell_session({"SSH_CONNECTION": "1.2.3.4 5 6.7.8.9 22"})
+    assert clip.is_remote_shell_session({"SSH_TTY": "/dev/pts/0"})
+    assert clip.is_remote_shell_session({"SSH_CLIENT": "1.2.3.4 5 22"})
+    assert not clip.is_remote_shell_session({})
+    assert not clip.is_remote_shell_session({"TERM": "xterm-256color"})
+
+
+class TestOsc52MultiplexerWrapping:
+    """CLI _write_osc52_clipboard must wrap for tmux/screen passthrough
+    (mirrors ui-tui/src/lib/osc52.ts wrapForMultiplexer)."""
+
+    def _capture_seq(self, env):
+        import io
+        from unittest.mock import patch as _patch
+        from cli import HermesCLI
+
+        cli_obj = HermesCLI.__new__(HermesCLI)
+        cli_obj._app = None
+        buf = io.StringIO()
+        with _patch.dict(clip.os.environ, env, clear=False), \
+             _patch("cli.sys.stdout", buf):
+            for var in ("TMUX", "STY"):
+                if var not in env:
+                    clip.os.environ.pop(var, None)
+            cli_obj._write_osc52_clipboard("hello")
+        return buf.getvalue()
+
+    def test_tmux_wraps_in_dcs_passthrough(self, monkeypatch):
+        monkeypatch.setenv("TMUX", "/tmp/tmux-123/default,1,0")
+        monkeypatch.delenv("STY", raising=False)
+        seq = self._capture_seq({"TMUX": "/tmp/tmux-123/default,1,0"})
+        assert seq.startswith("\x1bPtmux;")
+        assert "]52;c;" in seq
+        assert seq.endswith("\x1b\\")
+
+    def test_raw_osc52_outside_multiplexers(self, monkeypatch):
+        monkeypatch.delenv("TMUX", raising=False)
+        monkeypatch.delenv("STY", raising=False)
+        seq = self._capture_seq({})
+        assert seq.startswith("\x1b]52;c;")
+        assert seq.endswith("\x07")
+
+    def test_screen_wraps_in_dcs(self, monkeypatch):
+        monkeypatch.delenv("TMUX", raising=False)
+        monkeypatch.setenv("STY", "12345.pts-0.host")
+        seq = self._capture_seq({"STY": "12345.pts-0.host"})
+        assert seq.startswith("\x1bP\x1b]52;c;")
+        assert seq.endswith("\x1b\\")