test_reasoning_command.py
1 """Tests for the combined /reasoning command. 2 3 Covers both reasoning effort level management and reasoning display toggle, 4 plus the reasoning extraction and display pipeline from run_agent through CLI. 5 6 Combines functionality from: 7 - PR #789 (Aum08Desai): reasoning effort level management 8 - PR #790 (0xbyt4): reasoning display toggle and rendering 9 """ 10 11 import unittest 12 from types import SimpleNamespace 13 from unittest.mock import MagicMock, patch 14 import re 15 16 17 # --------------------------------------------------------------------------- 18 # Effort level parsing 19 # --------------------------------------------------------------------------- 20 21 class TestParseReasoningConfig(unittest.TestCase): 22 """Verify _parse_reasoning_config handles all effort levels.""" 23 24 def _parse(self, effort): 25 from cli import _parse_reasoning_config 26 return _parse_reasoning_config(effort) 27 28 def test_none_disables(self): 29 result = self._parse("none") 30 self.assertEqual(result, {"enabled": False}) 31 32 def test_valid_levels(self): 33 for level in ("low", "medium", "high", "xhigh", "minimal"): 34 result = self._parse(level) 35 self.assertIsNotNone(result) 36 self.assertTrue(result.get("enabled")) 37 self.assertEqual(result["effort"], level) 38 39 def test_empty_returns_none(self): 40 self.assertIsNone(self._parse("")) 41 self.assertIsNone(self._parse(" ")) 42 43 def test_unknown_returns_none(self): 44 self.assertIsNone(self._parse("ultra")) 45 self.assertIsNone(self._parse("turbo")) 46 47 def test_case_insensitive(self): 48 result = self._parse("HIGH") 49 self.assertIsNotNone(result) 50 self.assertEqual(result["effort"], "high") 51 52 53 # --------------------------------------------------------------------------- 54 # /reasoning command handler (combined effort + display) 55 # --------------------------------------------------------------------------- 56 57 class TestHandleReasoningCommand(unittest.TestCase): 58 """Test the combined _handle_reasoning_command method.""" 59 60 def _make_cli(self, reasoning_config=None, show_reasoning=False): 61 """Create a minimal CLI stub with the reasoning attributes.""" 62 stub = SimpleNamespace( 63 reasoning_config=reasoning_config, 64 show_reasoning=show_reasoning, 65 agent=MagicMock(), 66 ) 67 return stub 68 69 def test_show_enables_display(self): 70 stub = self._make_cli(show_reasoning=False) 71 # Simulate /reasoning show 72 arg = "show" 73 if arg in ("show", "on"): 74 stub.show_reasoning = True 75 stub.agent.reasoning_callback = lambda x: None 76 self.assertTrue(stub.show_reasoning) 77 78 def test_hide_disables_display(self): 79 stub = self._make_cli(show_reasoning=True) 80 # Simulate /reasoning hide 81 arg = "hide" 82 if arg in ("hide", "off"): 83 stub.show_reasoning = False 84 stub.agent.reasoning_callback = None 85 self.assertFalse(stub.show_reasoning) 86 self.assertIsNone(stub.agent.reasoning_callback) 87 88 def test_on_enables_display(self): 89 stub = self._make_cli(show_reasoning=False) 90 arg = "on" 91 if arg in ("show", "on"): 92 stub.show_reasoning = True 93 self.assertTrue(stub.show_reasoning) 94 95 def test_off_disables_display(self): 96 stub = self._make_cli(show_reasoning=True) 97 arg = "off" 98 if arg in ("hide", "off"): 99 stub.show_reasoning = False 100 self.assertFalse(stub.show_reasoning) 101 102 def test_effort_level_sets_config(self): 103 """Setting an effort level should update reasoning_config.""" 104 from cli import _parse_reasoning_config 105 stub = self._make_cli() 106 arg = "high" 107 parsed = _parse_reasoning_config(arg) 108 stub.reasoning_config = parsed 109 self.assertEqual(stub.reasoning_config, {"enabled": True, "effort": "high"}) 110 111 def test_effort_none_disables_reasoning(self): 112 from cli import _parse_reasoning_config 113 stub = self._make_cli() 114 parsed = _parse_reasoning_config("none") 115 stub.reasoning_config = parsed 116 self.assertEqual(stub.reasoning_config, {"enabled": False}) 117 118 def test_invalid_argument_rejected(self): 119 """Invalid arguments should be rejected (parsed returns None).""" 120 from cli import _parse_reasoning_config 121 parsed = _parse_reasoning_config("turbo") 122 self.assertIsNone(parsed) 123 124 def test_no_args_shows_status(self): 125 """With no args, should show current state (no crash).""" 126 stub = self._make_cli(reasoning_config=None, show_reasoning=False) 127 rc = stub.reasoning_config 128 if rc is None: 129 level = "medium (default)" 130 elif rc.get("enabled") is False: 131 level = "none (disabled)" 132 else: 133 level = rc.get("effort", "medium") 134 display_state = "on" if stub.show_reasoning else "off" 135 self.assertEqual(level, "medium (default)") 136 self.assertEqual(display_state, "off") 137 138 def test_status_with_disabled_reasoning(self): 139 stub = self._make_cli(reasoning_config={"enabled": False}, show_reasoning=True) 140 rc = stub.reasoning_config 141 if rc is None: 142 level = "medium (default)" 143 elif rc.get("enabled") is False: 144 level = "none (disabled)" 145 else: 146 level = rc.get("effort", "medium") 147 self.assertEqual(level, "none (disabled)") 148 149 def test_status_with_explicit_level(self): 150 stub = self._make_cli( 151 reasoning_config={"enabled": True, "effort": "xhigh"}, 152 show_reasoning=True, 153 ) 154 rc = stub.reasoning_config 155 level = rc.get("effort", "medium") 156 self.assertEqual(level, "xhigh") 157 158 159 # --------------------------------------------------------------------------- 160 # Reasoning extraction and result dict 161 # --------------------------------------------------------------------------- 162 163 class TestLastReasoningInResult(unittest.TestCase): 164 """Verify reasoning extraction from the messages list.""" 165 166 def _build_messages(self, reasoning=None): 167 return [ 168 {"role": "user", "content": "hello"}, 169 { 170 "role": "assistant", 171 "content": "Hi there!", 172 "reasoning": reasoning, 173 "finish_reason": "stop", 174 }, 175 ] 176 177 def test_reasoning_present(self): 178 messages = self._build_messages(reasoning="Let me think...") 179 last_reasoning = None 180 for msg in reversed(messages): 181 if msg.get("role") == "assistant" and msg.get("reasoning"): 182 last_reasoning = msg["reasoning"] 183 break 184 self.assertEqual(last_reasoning, "Let me think...") 185 186 def test_reasoning_none(self): 187 messages = self._build_messages(reasoning=None) 188 last_reasoning = None 189 for msg in reversed(messages): 190 if msg.get("role") == "assistant" and msg.get("reasoning"): 191 last_reasoning = msg["reasoning"] 192 break 193 self.assertIsNone(last_reasoning) 194 195 def test_picks_last_assistant(self): 196 messages = [ 197 {"role": "user", "content": "hello"}, 198 {"role": "assistant", "content": "...", "reasoning": "first thought"}, 199 {"role": "tool", "content": "result"}, 200 {"role": "assistant", "content": "done!", "reasoning": "final thought"}, 201 ] 202 last_reasoning = None 203 for msg in reversed(messages): 204 if msg.get("role") == "assistant" and msg.get("reasoning"): 205 last_reasoning = msg["reasoning"] 206 break 207 self.assertEqual(last_reasoning, "final thought") 208 209 def test_empty_reasoning_treated_as_none(self): 210 messages = self._build_messages(reasoning="") 211 last_reasoning = None 212 for msg in reversed(messages): 213 if msg.get("role") == "assistant" and msg.get("reasoning"): 214 last_reasoning = msg["reasoning"] 215 break 216 self.assertIsNone(last_reasoning) 217 218 219 # --------------------------------------------------------------------------- 220 # Reasoning display collapse 221 # --------------------------------------------------------------------------- 222 223 class TestReasoningCollapse(unittest.TestCase): 224 """Verify long reasoning is collapsed to 10 lines in the box.""" 225 226 def test_short_reasoning_not_collapsed(self): 227 reasoning = "\n".join(f"Line {i}" for i in range(5)) 228 lines = reasoning.strip().splitlines() 229 self.assertLessEqual(len(lines), 10) 230 231 def test_long_reasoning_collapsed(self): 232 reasoning = "\n".join(f"Line {i}" for i in range(25)) 233 lines = reasoning.strip().splitlines() 234 self.assertTrue(len(lines) > 10) 235 if len(lines) > 10: 236 display = "\n".join(lines[:10]) 237 display += f"\n ... ({len(lines) - 10} more lines)" 238 display_lines = display.splitlines() 239 self.assertEqual(len(display_lines), 11) 240 self.assertIn("15 more lines", display_lines[-1]) 241 242 def test_exactly_10_lines_not_collapsed(self): 243 reasoning = "\n".join(f"Line {i}" for i in range(10)) 244 lines = reasoning.strip().splitlines() 245 self.assertEqual(len(lines), 10) 246 self.assertFalse(len(lines) > 10) 247 248 def test_intermediate_callback_collapses_to_5(self): 249 """_on_reasoning shows max 5 lines.""" 250 reasoning = "\n".join(f"Step {i}" for i in range(12)) 251 lines = reasoning.strip().splitlines() 252 if len(lines) > 5: 253 preview = "\n".join(lines[:5]) 254 preview += f"\n ... ({len(lines) - 5} more lines)" 255 else: 256 preview = reasoning.strip() 257 preview_lines = preview.splitlines() 258 self.assertEqual(len(preview_lines), 6) 259 self.assertIn("7 more lines", preview_lines[-1]) 260 261 262 # --------------------------------------------------------------------------- 263 # Reasoning callback 264 # --------------------------------------------------------------------------- 265 266 class TestReasoningCallback(unittest.TestCase): 267 """Verify reasoning_callback invocation.""" 268 269 def test_callback_invoked_with_reasoning(self): 270 captured = [] 271 agent = MagicMock() 272 agent.reasoning_callback = lambda t: captured.append(t) 273 agent._extract_reasoning = MagicMock(return_value="deep thought") 274 275 reasoning_text = agent._extract_reasoning(MagicMock()) 276 if reasoning_text and agent.reasoning_callback: 277 agent.reasoning_callback(reasoning_text) 278 self.assertEqual(captured, ["deep thought"]) 279 280 def test_callback_not_invoked_without_reasoning(self): 281 captured = [] 282 agent = MagicMock() 283 agent.reasoning_callback = lambda t: captured.append(t) 284 agent._extract_reasoning = MagicMock(return_value=None) 285 286 reasoning_text = agent._extract_reasoning(MagicMock()) 287 if reasoning_text and agent.reasoning_callback: 288 agent.reasoning_callback(reasoning_text) 289 self.assertEqual(captured, []) 290 291 def test_callback_none_does_not_crash(self): 292 reasoning_text = "some thought" 293 callback = None 294 if reasoning_text and callback: 295 callback(reasoning_text) 296 # No exception = pass 297 298 299 class TestReasoningPreviewBuffering(unittest.TestCase): 300 def _make_cli(self): 301 from cli import HermesCLI 302 303 cli = HermesCLI.__new__(HermesCLI) 304 cli.verbose = True 305 cli._spinner_text = "" 306 cli._reasoning_preview_buf = "" 307 cli._invalidate = lambda *args, **kwargs: None 308 return cli 309 310 @patch("cli._cprint") 311 def test_streamed_reasoning_chunks_wait_for_boundary(self, mock_cprint): 312 cli = self._make_cli() 313 314 cli._on_reasoning("Let") 315 cli._on_reasoning(" me") 316 cli._on_reasoning(" think") 317 318 self.assertEqual(mock_cprint.call_count, 0) 319 320 cli._on_reasoning(" about this.\n") 321 322 self.assertEqual(mock_cprint.call_count, 1) 323 rendered = mock_cprint.call_args[0][0] 324 self.assertIn("[thinking] Let me think about this.", rendered) 325 326 @patch("cli._cprint") 327 def test_pending_reasoning_flushes_when_thinking_stops(self, mock_cprint): 328 cli = self._make_cli() 329 330 cli._on_reasoning("see") 331 cli._on_reasoning(" how") 332 cli._on_reasoning(" this") 333 cli._on_reasoning(" plays") 334 cli._on_reasoning(" out") 335 336 self.assertEqual(mock_cprint.call_count, 0) 337 338 cli._on_thinking("") 339 340 self.assertEqual(mock_cprint.call_count, 1) 341 rendered = mock_cprint.call_args[0][0] 342 self.assertIn("[thinking] see how this plays out", rendered) 343 344 @patch("cli._cprint") 345 @patch("cli.shutil.get_terminal_size", return_value=SimpleNamespace(columns=50)) 346 def test_reasoning_preview_compacts_newlines_and_wraps_to_terminal(self, _mock_term, mock_cprint): 347 cli = self._make_cli() 348 349 cli._emit_reasoning_preview( 350 "First line\nstill same thought\n\n\nSecond paragraph with more detail here." 351 ) 352 353 rendered = mock_cprint.call_args[0][0] 354 plain = re.sub(r"\x1b\[[0-9;]*m", "", rendered) 355 normalized = " ".join(plain.split()) 356 self.assertIn("[thinking] First line still same thought", plain) 357 self.assertIn("Second paragraph with more detail here.", normalized) 358 self.assertNotIn("\n\n\n", plain) 359 360 @patch("cli.shutil.get_terminal_size", return_value=SimpleNamespace(columns=60)) 361 def test_reasoning_flush_threshold_tracks_terminal_width(self, _mock_term): 362 cli = self._make_cli() 363 364 cli._reasoning_preview_buf = "a" * 30 365 cli._flush_reasoning_preview(force=False) 366 self.assertEqual(cli._reasoning_preview_buf, "a" * 30) 367 368 369 class TestReasoningDisplayModeSelection(unittest.TestCase): 370 def _make_cli(self, *, show_reasoning=False, streaming_enabled=False, verbose=False): 371 from cli import HermesCLI 372 373 cli = HermesCLI.__new__(HermesCLI) 374 cli.show_reasoning = show_reasoning 375 cli.streaming_enabled = streaming_enabled 376 cli.verbose = verbose 377 cli._stream_reasoning_delta = lambda text: ("stream", text) 378 cli._on_reasoning = lambda text: ("preview", text) 379 return cli 380 381 def test_show_reasoning_non_streaming_uses_final_box_only(self): 382 cli = self._make_cli(show_reasoning=True, streaming_enabled=False, verbose=False) 383 384 self.assertIsNone(cli._current_reasoning_callback()) 385 386 def test_show_reasoning_streaming_uses_live_reasoning_box(self): 387 cli = self._make_cli(show_reasoning=True, streaming_enabled=True, verbose=False) 388 389 callback = cli._current_reasoning_callback() 390 self.assertIsNotNone(callback) 391 self.assertEqual(callback("x"), ("stream", "x")) 392 393 def test_verbose_without_show_reasoning_uses_preview_callback(self): 394 cli = self._make_cli(show_reasoning=False, streaming_enabled=False, verbose=True) 395 396 callback = cli._current_reasoning_callback() 397 self.assertIsNotNone(callback) 398 self.assertEqual(callback("x"), ("preview", "x")) 399 400 401 # --------------------------------------------------------------------------- 402 # Real provider format extraction 403 # --------------------------------------------------------------------------- 404 405 class TestExtractReasoningFormats(unittest.TestCase): 406 """Test _extract_reasoning with real provider response formats.""" 407 408 def _get_extractor(self): 409 from run_agent import AIAgent 410 return AIAgent._extract_reasoning 411 412 def test_openrouter_reasoning_details(self): 413 extract = self._get_extractor() 414 msg = SimpleNamespace( 415 reasoning=None, 416 reasoning_content=None, 417 reasoning_details=[ 418 {"type": "reasoning.summary", "summary": "Analyzing Python lists."}, 419 ], 420 ) 421 result = extract(None, msg) 422 self.assertIn("Python lists", result) 423 424 def test_deepseek_reasoning_field(self): 425 extract = self._get_extractor() 426 msg = SimpleNamespace( 427 reasoning="Solving step by step.\nx + y = 8.", 428 reasoning_content=None, 429 ) 430 result = extract(None, msg) 431 self.assertIn("x + y = 8", result) 432 433 def test_moonshot_reasoning_content(self): 434 extract = self._get_extractor() 435 msg = SimpleNamespace( 436 reasoning_content="Explaining async/await.", 437 ) 438 result = extract(None, msg) 439 self.assertIn("async/await", result) 440 441 def test_no_reasoning_returns_none(self): 442 extract = self._get_extractor() 443 msg = SimpleNamespace(content="Hello!") 444 result = extract(None, msg) 445 self.assertIsNone(result) 446 447 448 # --------------------------------------------------------------------------- 449 # Inline <think> block extraction fallback 450 # --------------------------------------------------------------------------- 451 452 class TestInlineThinkBlockExtraction(unittest.TestCase): 453 """Test _build_assistant_message extracts inline <think> blocks as reasoning 454 when no structured API-level reasoning fields are present.""" 455 456 def _build_msg(self, content, reasoning=None, reasoning_content=None, reasoning_details=None, tool_calls=None): 457 """Create a mock API response message.""" 458 msg = SimpleNamespace(content=content, tool_calls=tool_calls) 459 if reasoning is not None: 460 msg.reasoning = reasoning 461 if reasoning_content is not None: 462 msg.reasoning_content = reasoning_content 463 if reasoning_details is not None: 464 msg.reasoning_details = reasoning_details 465 return msg 466 467 def _make_agent(self): 468 """Create a minimal agent with _build_assistant_message.""" 469 from run_agent import AIAgent 470 agent = MagicMock(spec=AIAgent) 471 agent._build_assistant_message = AIAgent._build_assistant_message.__get__(agent) 472 agent._extract_reasoning = AIAgent._extract_reasoning.__get__(agent) 473 agent.verbose_logging = False 474 agent.reasoning_callback = None 475 agent.stream_delta_callback = None # non-streaming by default 476 agent._stream_callback = None # non-streaming by default 477 return agent 478 479 def test_single_think_block_extracted(self): 480 agent = self._make_agent() 481 api_msg = self._build_msg("<think>Let me calculate 2+2=4.</think>The answer is 4.") 482 result = agent._build_assistant_message(api_msg, "stop") 483 self.assertEqual(result["reasoning"], "Let me calculate 2+2=4.") 484 485 def test_multiple_think_blocks_extracted(self): 486 agent = self._make_agent() 487 api_msg = self._build_msg("<think>First thought.</think>Some text<think>Second thought.</think>More text") 488 result = agent._build_assistant_message(api_msg, "stop") 489 self.assertIn("First thought.", result["reasoning"]) 490 self.assertIn("Second thought.", result["reasoning"]) 491 492 def test_no_think_blocks_no_reasoning(self): 493 agent = self._make_agent() 494 api_msg = self._build_msg("Just a plain response.") 495 result = agent._build_assistant_message(api_msg, "stop") 496 # No structured reasoning AND no inline think blocks → None 497 self.assertIsNone(result["reasoning"]) 498 499 def test_structured_reasoning_takes_priority(self): 500 """When structured API reasoning exists, inline think blocks should NOT override.""" 501 agent = self._make_agent() 502 api_msg = self._build_msg( 503 "<think>Inline thought.</think>Response text.", 504 reasoning="Structured reasoning from API.", 505 ) 506 result = agent._build_assistant_message(api_msg, "stop") 507 self.assertEqual(result["reasoning"], "Structured reasoning from API.") 508 509 def test_empty_think_block_ignored(self): 510 agent = self._make_agent() 511 api_msg = self._build_msg("<think></think>Hello!") 512 result = agent._build_assistant_message(api_msg, "stop") 513 # Empty think block should not produce reasoning 514 self.assertIsNone(result["reasoning"]) 515 516 def test_multiline_think_block(self): 517 agent = self._make_agent() 518 api_msg = self._build_msg("<think>\nStep 1: Analyze.\nStep 2: Solve.\n</think>Done.") 519 result = agent._build_assistant_message(api_msg, "stop") 520 self.assertIn("Step 1: Analyze.", result["reasoning"]) 521 self.assertIn("Step 2: Solve.", result["reasoning"]) 522 523 def test_callback_fires_for_inline_think(self): 524 """Reasoning callback should fire when reasoning is extracted from inline think blocks.""" 525 agent = self._make_agent() 526 captured = [] 527 agent.reasoning_callback = lambda t: captured.append(t) 528 api_msg = self._build_msg("<think>Deep analysis here.</think>Answer.") 529 agent._build_assistant_message(api_msg, "stop") 530 self.assertEqual(len(captured), 1) 531 self.assertIn("Deep analysis", captured[0]) 532 533 534 # --------------------------------------------------------------------------- 535 # Config defaults 536 # --------------------------------------------------------------------------- 537 538 class TestConfigDefault(unittest.TestCase): 539 """Verify config default for show_reasoning.""" 540 541 def test_default_config_has_show_reasoning(self): 542 from hermes_cli.config import DEFAULT_CONFIG 543 display = DEFAULT_CONFIG.get("display", {}) 544 self.assertIn("show_reasoning", display) 545 self.assertFalse(display["show_reasoning"]) 546 547 548 class TestCommandRegistered(unittest.TestCase): 549 """Verify /reasoning is in the COMMANDS dict.""" 550 551 def test_reasoning_in_commands(self): 552 from hermes_cli.commands import COMMANDS 553 self.assertIn("/reasoning", COMMANDS) 554 555 556 # --------------------------------------------------------------------------- 557 # End-to-end pipeline 558 # --------------------------------------------------------------------------- 559 560 class TestEndToEndPipeline(unittest.TestCase): 561 """Simulate the full pipeline: extraction -> result dict -> display.""" 562 563 def test_openrouter_claude_pipeline(self): 564 from run_agent import AIAgent 565 566 api_message = SimpleNamespace( 567 role="assistant", 568 content="Lists support append().", 569 tool_calls=None, 570 reasoning=None, 571 reasoning_content=None, 572 reasoning_details=[ 573 {"type": "reasoning.summary", "summary": "Python list methods."}, 574 ], 575 ) 576 577 reasoning = AIAgent._extract_reasoning(None, api_message) 578 self.assertIsNotNone(reasoning) 579 580 messages = [ 581 {"role": "user", "content": "How do I add items?"}, 582 {"role": "assistant", "content": api_message.content, "reasoning": reasoning}, 583 ] 584 585 last_reasoning = None 586 for msg in reversed(messages): 587 if msg.get("role") == "assistant" and msg.get("reasoning"): 588 last_reasoning = msg["reasoning"] 589 break 590 591 result = { 592 "final_response": api_message.content, 593 "last_reasoning": last_reasoning, 594 } 595 596 self.assertIn("last_reasoning", result) 597 self.assertIn("Python list methods", result["last_reasoning"]) 598 599 def test_no_reasoning_model_pipeline(self): 600 from run_agent import AIAgent 601 602 api_message = SimpleNamespace(content="Paris.", tool_calls=None) 603 reasoning = AIAgent._extract_reasoning(None, api_message) 604 self.assertIsNone(reasoning) 605 606 result = {"final_response": api_message.content, "last_reasoning": reasoning} 607 self.assertIsNone(result["last_reasoning"]) 608 609 610 # --------------------------------------------------------------------------- 611 # Duplicate reasoning box prevention (Bug fix: 3 boxes for 1 reasoning) 612 # --------------------------------------------------------------------------- 613 614 class TestReasoningDeltasFiredFlag(unittest.TestCase): 615 """_build_assistant_message should not re-fire reasoning_callback when 616 reasoning was already streamed via _fire_reasoning_delta.""" 617 618 def _make_agent(self): 619 from run_agent import AIAgent 620 agent = AIAgent.__new__(AIAgent) 621 agent.reasoning_callback = None 622 agent.stream_delta_callback = None 623 agent._stream_callback = None 624 agent.verbose_logging = False 625 return agent 626 627 def test_fire_reasoning_delta_calls_callback(self): 628 agent = self._make_agent() 629 captured = [] 630 agent.reasoning_callback = lambda t: captured.append(t) 631 agent._fire_reasoning_delta("thinking...") 632 self.assertEqual(captured, ["thinking..."]) 633 634 def test_build_assistant_message_skips_callback_when_already_streamed(self): 635 """When streaming already fired reasoning deltas, the post-stream 636 _build_assistant_message should NOT re-fire the callback.""" 637 agent = self._make_agent() 638 captured = [] 639 agent.reasoning_callback = lambda t: captured.append(t) 640 agent.stream_delta_callback = lambda t: None # streaming is active 641 642 # Simulate streaming having already fired reasoning 643 644 msg = SimpleNamespace( 645 content="I'll merge that.", 646 tool_calls=None, 647 reasoning_content="Let me merge the PR.", 648 reasoning=None, 649 reasoning_details=None, 650 ) 651 agent._build_assistant_message(msg, "stop") 652 653 # Callback should NOT have been fired again 654 self.assertEqual(captured, []) 655 656 def test_build_assistant_message_skips_callback_when_streaming_active(self): 657 """When streaming is active, callback should NEVER fire from 658 _build_assistant_message — reasoning was already displayed during the 659 stream (either via reasoning_content deltas or content tag extraction). 660 Any missed reasoning is caught by the CLI post-response fallback.""" 661 agent = self._make_agent() 662 captured = [] 663 agent.reasoning_callback = lambda t: captured.append(t) 664 agent.stream_delta_callback = lambda t: None # streaming active 665 666 # Reasoning came through content tags, not reasoning_content deltas. 667 # Callback should not fire since streaming is active. 668 669 msg = SimpleNamespace( 670 content="I'll merge that.", 671 tool_calls=None, 672 reasoning_content="Let me merge the PR.", 673 reasoning=None, 674 reasoning_details=None, 675 ) 676 agent._build_assistant_message(msg, "stop") 677 678 # Callback should NOT fire — streaming is active 679 self.assertEqual(captured, []) 680 681 def test_build_assistant_message_fires_callback_without_streaming(self): 682 """When no streaming is active, callback always fires for structured 683 reasoning.""" 684 agent = self._make_agent() 685 captured = [] 686 agent.reasoning_callback = lambda t: captured.append(t) 687 # No streaming 688 agent.stream_delta_callback = None 689 690 msg = SimpleNamespace( 691 content="I'll merge that.", 692 tool_calls=None, 693 reasoning_content="Let me merge the PR.", 694 reasoning=None, 695 reasoning_details=None, 696 ) 697 agent._build_assistant_message(msg, "stop") 698 699 self.assertEqual(captured, ["Let me merge the PR."]) 700 701 702 class TestReasoningShownThisTurnFlag(unittest.TestCase): 703 """Post-response reasoning display should be suppressed when reasoning 704 was already shown during streaming in a tool-calling loop.""" 705 706 def _make_cli(self): 707 from cli import HermesCLI 708 cli = HermesCLI.__new__(HermesCLI) 709 cli.show_reasoning = True 710 cli.streaming_enabled = True 711 cli._stream_box_opened = False 712 cli._reasoning_box_opened = False 713 cli._reasoning_stream_started = False 714 cli._reasoning_shown_this_turn = False 715 cli._reasoning_buf = "" 716 cli._stream_buf = "" 717 cli._stream_started = False 718 cli._stream_text_ansi = "" 719 cli._stream_prefilt = "" 720 cli._in_reasoning_block = False 721 cli._reasoning_preview_buf = "" 722 return cli 723 724 @patch("cli._cprint") 725 def test_streaming_reasoning_sets_turn_flag(self, mock_cprint): 726 cli = self._make_cli() 727 self.assertFalse(cli._reasoning_shown_this_turn) 728 cli._stream_reasoning_delta("Thinking about it...") 729 self.assertTrue(cli._reasoning_shown_this_turn) 730 731 @patch("cli._cprint") 732 def test_turn_flag_survives_reset_stream_state(self, mock_cprint): 733 """_reasoning_shown_this_turn must NOT be cleared by 734 _reset_stream_state (called at intermediate turn boundaries).""" 735 cli = self._make_cli() 736 cli._stream_reasoning_delta("Thinking...") 737 self.assertTrue(cli._reasoning_shown_this_turn) 738 739 # Simulate intermediate turn boundary (tool call) 740 cli._reset_stream_state() 741 742 # Flag must persist 743 self.assertTrue(cli._reasoning_shown_this_turn) 744 745 @patch("cli._cprint") 746 def test_turn_flag_cleared_before_new_turn(self, mock_cprint): 747 """The turn flag should be reset at the start of a new user turn. 748 This happens outside _reset_stream_state, at the call site.""" 749 cli = self._make_cli() 750 cli._reasoning_shown_this_turn = True 751 752 # Simulate new user turn setup 753 cli._reset_stream_state() 754 cli._reasoning_shown_this_turn = False # done by process_input 755 756 self.assertFalse(cli._reasoning_shown_this_turn) 757 758 759 if __name__ == "__main__": 760 unittest.main()