diff --git a/cli.py b/cli.py index 2017e6b9f26..0333d2be4ec 100644 --- a/cli.py +++ b/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"): diff --git a/hermes_cli/cli_commands_mixin.py b/hermes_cli/cli_commands_mixin.py index 102dee19679..622104380f4 100644 --- a/hermes_cli/cli_commands_mixin.py +++ b/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 diff --git a/hermes_cli/clipboard.py b/hermes_cli/clipboard.py index db87421194a..3881502e5df 100644 --- a/hermes_cli/clipboard.py +++ b/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. diff --git a/tests/cli/test_cli_copy_command.py b/tests/cli/test_cli_copy_command.py index 965551b496e..460414dc701 100644 --- a/tests/cli/test_cli_copy_command.py +++ b/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"}] diff --git a/tests/cli/test_stream_partial_line_flush.py b/tests/cli/test_stream_partial_line_flush.py index 7d5e96a60a2..2fc745d3e33 100644 --- a/tests/cli/test_stream_partial_line_flush.py +++ b/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" diff --git a/tests/hermes_cli/test_clipboard_text_write.py b/tests/hermes_cli/test_clipboard_text_write.py index 8939e0e0584..252e684bb90 100644 --- a/tests/hermes_cli/test_clipboard_text_write.py +++ b/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\\")