Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
19 changes: 15 additions & 4 deletions hermes_cli/cli_stream_mixin.py
Original file line number Diff line number Diff line change
Expand Up @@ -266,11 +266,20 @@ def _stream_reasoning_delta(self, text: str) -> None:
# Emit complete lines; force-flush long partial lines so reasoning is visible in
# real-time even without newlines.
while "\n" in self._reasoning_buf:
line, self._reasoning_buf = self._reasoning_buf.split("\n", 1)
line, rest = self._reasoning_buf.split("\n", 1)
if not line and not rest.strip("\n"):
break # hold trailing newlines: thinking text ends with "\n\n" before the footer
self._reasoning_buf = rest
if not line.strip() and getattr(self, "_reasoning_partial_flushed", False):
# This newline only ends a force-flushed partial line that is already printed.
self._reasoning_partial_flushed = False
continue
self._reasoning_partial_flushed = False
_cprint(f"{_DIM}{line}{_RST}")
if len(self._reasoning_buf) > 80:
if "\n" not in self._reasoning_buf and len(self._reasoning_buf) > 80:
_cprint(f"{_DIM}{self._reasoning_buf}{_RST}")
self._reasoning_buf = ""
self._reasoning_partial_flushed = True

def _agent_status_print(self, *args, **kwargs) -> None:
"""``agent._print_fn`` for the interactive CLI: agent status lines (subagent completion ``✓ [set n · i/N]``,
Expand All @@ -296,10 +305,11 @@ def _close_reasoning_box(self) -> None:
from cli import _DIM, _RST, _cprint
if not getattr(self, "_reasoning_box_opened", False):
return
buf = getattr(self, "_reasoning_buf", "")
buf = getattr(self, "_reasoning_buf", "").rstrip("\n")
if buf:
_cprint(f"{_DIM}{buf}{_RST}")
self._reasoning_buf = ""
self._reasoning_buf = ""
self._reasoning_partial_flushed = False
w = self._scrollback_box_width()
_cprint(f"{_DIM}└{'─' * (w - 2)}┘{_RST}")
self._reasoning_box_opened = False
Expand Down Expand Up @@ -536,6 +546,7 @@ def _reset_stream_state(self) -> None:
self._stream_last_was_newline = True
self._reasoning_box_opened = False
self._reasoning_buf = ""
self._reasoning_partial_flushed = False
self._reasoning_preview_buf = ""
# A batch cancelled/errored before any tool.started would otherwise mute the next turn's line.
self.__dict__.pop("_tool_gen_announced", None)
Expand Down
52 changes: 52 additions & 0 deletions tests/hermes_cli/test_reasoning_box_blank_lines.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,52 @@
"""The streamed reasoning box must not end with a blank line.

Anthropic thinking text ends with ``\\n\\n``; the CLI used to print each newline as a line,
leaving an empty dim row above the box footer. The newline that ends a force-flushed long
line also printed as a blank row. Paragraph breaks inside the reasoning must still render.
"""

from unittest.mock import patch

import pytest


def _make_cli():
from cli import HermesCLI

cli = HermesCLI.__new__(HermesCLI)
cli.show_reasoning = True
cli.streaming_enabled = True
cli._stream_box_opened = False
cli._stream_box_live = False
cli._reasoning_box_opened = False
cli._reasoning_buf = ""
cli._held_status_lines = []
cli._scrollback_box_width = lambda width=None: 40
return cli


def _render(chunks):
from cli import _DIM, _RST

cli = _make_cli()
with patch("cli._cprint") as cprint:
for chunk in chunks:
cli._stream_reasoning_delta(chunk)
cli._close_reasoning_box()
lines = [call.args[0] for call in cprint.call_args_list]
# Drop the header and footer; strip the dim wrapper from body lines.
return [line.removeprefix(_DIM).removesuffix(_RST) for line in lines[1:-1]]


@pytest.mark.parametrize("chunks", [
["First thought.\n\nSecond thought.\n\n"],
["First thought.\n", "\nSecond thought.", "\n", "\n"],
])
def test_trailing_blank_lines_are_not_rendered(chunks):
assert _render(chunks) == ["First thought.", "", "Second thought."]


@pytest.mark.parametrize("continuation", ["\n\nNext.\n\n", " \n\nNext.\n\n"])
def test_force_flushed_long_line_does_not_add_a_blank_line(continuation):
long_line = "x" * 90
assert _render([long_line, continuation]) == [long_line, "", "Next."]