Compare commits

..

1 Commits

Author SHA1 Message Date
teknium1
4d873f77c1 feat(cli): add /reasoning command for effort level and display toggle
Combined implementation of reasoning management:
- /reasoning              Show current effort level and display state
- /reasoning <level>      Set reasoning effort (none, low, medium, high, xhigh)
- /reasoning show|on      Show model thinking/reasoning in output
- /reasoning hide|off     Hide model thinking/reasoning from output

Effort level changes persist to config and force agent re-init.
Display toggle updates the agent callback dynamically without re-init.

When display is enabled:
- Intermediate reasoning shown as dim [thinking] lines during tool loops
- Final reasoning shown in a bordered box above the response
- Long reasoning collapsed (5 lines intermediate, 10 lines final)

Also adds:
- reasoning_callback parameter to AIAgent
- last_reasoning in run_conversation result dict
- show_reasoning config option (display section, default: false)
- Display section in /config output
- 34 tests covering both features

Combines functionality from PR #789 and PR #790.

Co-authored-by: Aum Desai <Aum08Desai@users.noreply.github.com>
Co-authored-by: 0xbyt4 <35742124+0xbyt4@users.noreply.github.com>
2026-03-11 06:02:18 -07:00
12 changed files with 604 additions and 99 deletions

View File

@@ -103,24 +103,22 @@ class ContextCompressor:
parts.append(f"[{role.upper()}]: {content}")
content_to_summarize = "\n\n".join(parts)
prompt = (
"You are performing a CONTEXT CHECKPOINT COMPACTION. Create a handoff "
"summary for the AI assistant that will resume this conversation.\n\n"
"Include:\n"
"- Current progress and key decisions made\n"
"- Important context, constraints, or user preferences discovered\n"
"- What remains to be done (clear next steps)\n"
"- Any critical data: file paths, variable names, URLs, error messages, "
"or code snippets needed to continue\n"
"- Tool calls made and their key results\n\n"
"Be concise, structured, and focused on helping the assistant seamlessly "
"continue the work without re-doing what's already been done.\n\n"
f"Target roughly {self.summary_target_tokens} tokens.\n\n"
"---\n"
f"TURNS TO SUMMARIZE:\n{content_to_summarize}\n"
"---\n\n"
'Write only the summary, starting with "[CONTEXT SUMMARY]:" prefix.'
)
prompt = f"""Summarize these conversation turns concisely. This summary will replace these turns in the conversation history.
Write from a neutral perspective describing:
1. What actions were taken (tool calls, searches, file operations)
2. Key information or results obtained
3. Important decisions or findings
4. Relevant data, file names, or outputs
Keep factual and informative. Target ~{self.summary_target_tokens} tokens.
---
TURNS TO SUMMARIZE:
{content_to_summarize}
---
Write only the summary, starting with "[CONTEXT SUMMARY]:" prefix."""
# 1. Try the auxiliary model (cheap/fast)
if self.client:

View File

@@ -670,6 +670,11 @@ display:
# Works over SSH. Most terminals can be configured to flash the taskbar or play a sound.
bell_on_complete: false
# Show model reasoning/thinking before each response.
# When enabled, a dim box shows the model's thought process above the response.
# Toggle at runtime with /reasoning show or /reasoning hide.
show_reasoning: false
# ───────────────────────────────────────────────────────────────────────────
# Skin / Theme
# ───────────────────────────────────────────────────────────────────────────

93
cli.py
View File

@@ -205,6 +205,7 @@ def load_cli_config() -> Dict[str, Any]:
"display": {
"compact": False,
"resume_display": "full",
"show_reasoning": False,
"skin": "default",
},
"clarify": {
@@ -1121,6 +1122,8 @@ class HermesCLI:
self.resume_display = CLI_CONFIG["display"].get("resume_display", "full")
# bell_on_complete: play terminal bell (\a) when agent finishes a response
self.bell_on_complete = CLI_CONFIG["display"].get("bell_on_complete", False)
# show_reasoning: display model thinking/reasoning before the response
self.show_reasoning = CLI_CONFIG["display"].get("show_reasoning", False)
self.verbose = verbose if verbose is not None else (self.tool_progress_mode == "verbose")
# Configuration - priority: CLI args > env vars > config file
@@ -1495,6 +1498,7 @@ class HermesCLI:
platform="cli",
session_db=self._session_db,
clarify_callback=self._clarify_callback,
reasoning_callback=self._on_reasoning if self.show_reasoning else None,
honcho_session_key=self.session_id,
fallback_model=self._fallback_model,
thinking_callback=self._on_thinking,
@@ -2848,6 +2852,8 @@ class HermesCLI:
self._show_gateway_status()
elif cmd_lower == "/verbose":
self._toggle_verbose()
elif cmd_lower.startswith("/reasoning"):
self._handle_reasoning_command(cmd_original)
elif cmd_lower == "/compress":
self._manual_compress()
elif cmd_lower == "/usage":
@@ -3073,6 +3079,75 @@ class HermesCLI:
}
self.console.print(labels.get(self.tool_progress_mode, ""))
def _handle_reasoning_command(self, cmd: str):
"""Handle /reasoning — manage effort level and display toggle.
Usage:
/reasoning Show current effort level and display state
/reasoning <level> Set reasoning effort (none, low, medium, high, xhigh)
/reasoning show|on Show model thinking/reasoning in output
/reasoning hide|off Hide model thinking/reasoning from output
"""
parts = cmd.strip().split(maxsplit=1)
if len(parts) < 2:
# Show current state
rc = self.reasoning_config
if rc is None:
level = "medium (default)"
elif rc.get("enabled") is False:
level = "none (disabled)"
else:
level = rc.get("effort", "medium")
display_state = "on" if self.show_reasoning else "off"
_cprint(f" {_GOLD}Reasoning effort: {level}{_RST}")
_cprint(f" {_GOLD}Reasoning display: {display_state}{_RST}")
_cprint(f" {_DIM}Usage: /reasoning <none|low|medium|high|xhigh|show|hide>{_RST}")
return
arg = parts[1].strip().lower()
# Display toggle
if arg in ("show", "on"):
self.show_reasoning = True
if self.agent:
self.agent.reasoning_callback = self._on_reasoning
_cprint(f" {_GOLD}Reasoning display: ON{_RST}")
_cprint(f" {_DIM}Model thinking will be shown during and after each response.{_RST}")
return
if arg in ("hide", "off"):
self.show_reasoning = False
if self.agent:
self.agent.reasoning_callback = None
_cprint(f" {_GOLD}Reasoning display: OFF{_RST}")
return
# Effort level change
parsed = _parse_reasoning_config(arg)
if parsed is None:
_cprint(f" {_DIM}(._.) Unknown argument: {arg}{_RST}")
_cprint(f" {_DIM}Valid levels: none, low, minimal, medium, high, xhigh{_RST}")
_cprint(f" {_DIM}Display: show, hide{_RST}")
return
self.reasoning_config = parsed
self.agent = None # Force agent re-init with new reasoning config
if save_config_value("agent.reasoning_effort", arg):
_cprint(f" {_GOLD}Reasoning effort set to '{arg}' (saved to config){_RST}")
else:
_cprint(f" {_GOLD}Reasoning effort set to '{arg}' (session only){_RST}")
def _on_reasoning(self, reasoning_text: str):
"""Callback for intermediate reasoning display during tool-call loops."""
lines = reasoning_text.strip().splitlines()
if len(lines) > 5:
preview = "\n".join(lines[:5])
preview += f"\n ... ({len(lines) - 5} more lines)"
else:
preview = reasoning_text.strip()
_cprint(f" {_DIM}[thinking] {preview}{_RST}")
def _manual_compress(self):
"""Manually trigger context compression on the current conversation."""
if not self.conversation_history or len(self.conversation_history) < 4:
@@ -3542,6 +3617,24 @@ class HermesCLI:
if response and pending_message:
response = response + "\n\n---\n_[Interrupted - processing new message]_"
# Display reasoning (thinking) box if enabled and available
if self.show_reasoning and result:
reasoning = result.get("last_reasoning")
if reasoning:
w = shutil.get_terminal_size().columns
r_label = " Reasoning "
r_fill = w - 2 - len(r_label)
r_top = f"{_DIM}┌─{r_label}{'' * max(r_fill - 1, 0)}{_RST}"
r_bot = f"{_DIM}{'' * (w - 2)}{_RST}"
# Collapse long reasoning: show first 10 lines
lines = reasoning.strip().splitlines()
if len(lines) > 10:
display_reasoning = "\n".join(lines[:10])
display_reasoning += f"\n{_DIM} ... ({len(lines) - 10} more lines){_RST}"
else:
display_reasoning = reasoning.strip()
_cprint(f"\n{r_top}\n{_DIM}{display_reasoning}{_RST}\n{r_bot}")
if response:
# Use a Rich Panel for the response box — adapts to terminal
# width at render time instead of hard-coding border length.

View File

@@ -9,7 +9,6 @@ Uses slack-bolt (Python) with Socket Mode for:
"""
import asyncio
import logging
import os
import re
from typing import Dict, List, Optional, Any
@@ -42,9 +41,6 @@ from gateway.platforms.base import (
)
logger = logging.getLogger(__name__)
def check_slack_requirements() -> bool:
"""Check if Slack dependencies are available."""
return SLACK_AVAILABLE
@@ -77,19 +73,17 @@ class SlackAdapter(BasePlatformAdapter):
async def connect(self) -> bool:
"""Connect to Slack via Socket Mode."""
if not SLACK_AVAILABLE:
logger.error(
"[Slack] slack-bolt not installed. Run: pip install slack-bolt",
)
print("[Slack] slack-bolt not installed. Run: pip install slack-bolt")
return False
bot_token = self.config.token
app_token = os.getenv("SLACK_APP_TOKEN")
if not bot_token:
logger.error("[Slack] SLACK_BOT_TOKEN not set")
print("[Slack] SLACK_BOT_TOKEN not set")
return False
if not app_token:
logger.error("[Slack] SLACK_APP_TOKEN not set")
print("[Slack] SLACK_APP_TOKEN not set")
return False
try:
@@ -123,22 +117,19 @@ class SlackAdapter(BasePlatformAdapter):
asyncio.create_task(self._handler.start_async())
self._running = True
logger.info("[Slack] Connected as @%s (Socket Mode)", bot_name)
print(f"[Slack] Connected as @{bot_name} (Socket Mode)")
return True
except Exception as e: # pragma: no cover - defensive logging
logger.error("[Slack] Connection failed: %s", e, exc_info=True)
except Exception as e:
print(f"[Slack] Connection failed: {e}")
return False
async def disconnect(self) -> None:
"""Disconnect from Slack."""
if self._handler:
try:
await self._handler.close_async()
except Exception as e: # pragma: no cover - defensive logging
logger.warning("[Slack] Error while closing Socket Mode handler: %s", e, exc_info=True)
await self._handler.close_async()
self._running = False
logger.info("[Slack] Disconnected")
print("[Slack] Disconnected")
async def send(
self,
@@ -171,8 +162,8 @@ class SlackAdapter(BasePlatformAdapter):
raw_response=result,
)
except Exception as e: # pragma: no cover - defensive logging
logger.error("[Slack] Send error: %s", e, exc_info=True)
except Exception as e:
print(f"[Slack] Send error: {e}")
return SendResult(success=False, error=str(e))
async def edit_message(
@@ -191,14 +182,7 @@ class SlackAdapter(BasePlatformAdapter):
text=content,
)
return SendResult(success=True, message_id=message_id)
except Exception as e: # pragma: no cover - defensive logging
logger.error(
"[Slack] Failed to edit message %s in channel %s: %s",
message_id,
chat_id,
e,
exc_info=True,
)
except Exception as e:
return SendResult(success=False, error=str(e))
async def send_typing(self, chat_id: str, metadata=None) -> None:
@@ -230,14 +214,8 @@ class SlackAdapter(BasePlatformAdapter):
)
return SendResult(success=True, raw_response=result)
except Exception as e: # pragma: no cover - defensive logging
logger.error(
"[%s] Failed to send local Slack image %s: %s",
self.name,
image_path,
e,
exc_info=True,
)
except Exception as e:
print(f"[{self.name}] Failed to send local image: {e}")
return await super().send_image_file(chat_id, image_path, caption, reply_to)
async def send_image(
@@ -269,13 +247,7 @@ class SlackAdapter(BasePlatformAdapter):
return SendResult(success=True, raw_response=result)
except Exception as e: # pragma: no cover - defensive logging
logger.warning(
"[Slack] Failed to upload image from URL %s, falling back to text: %s",
image_url,
e,
exc_info=True,
)
except Exception as e:
# Fall back to sending the URL as text
text = f"{caption}\n{image_url}" if caption else image_url
return await self.send(chat_id=chat_id, content=text, reply_to=reply_to)
@@ -301,13 +273,7 @@ class SlackAdapter(BasePlatformAdapter):
)
return SendResult(success=True, raw_response=result)
except Exception as e: # pragma: no cover - defensive logging
logger.error(
"[Slack] Failed to send audio file %s: %s",
audio_path,
e,
exc_info=True,
)
except Exception as e:
return SendResult(success=False, error=str(e))
async def send_video(
@@ -334,14 +300,8 @@ class SlackAdapter(BasePlatformAdapter):
)
return SendResult(success=True, raw_response=result)
except Exception as e: # pragma: no cover - defensive logging
logger.error(
"[%s] Failed to send video %s: %s",
self.name,
video_path,
e,
exc_info=True,
)
except Exception as e:
print(f"[{self.name}] Failed to send video: {e}")
return await super().send_video(chat_id, video_path, caption, reply_to)
async def send_document(
@@ -371,14 +331,8 @@ class SlackAdapter(BasePlatformAdapter):
)
return SendResult(success=True, raw_response=result)
except Exception as e: # pragma: no cover - defensive logging
logger.error(
"[%s] Failed to send document %s: %s",
self.name,
file_path,
e,
exc_info=True,
)
except Exception as e:
print(f"[{self.name}] Failed to send document: {e}")
return await super().send_document(chat_id, file_path, caption, file_name, reply_to)
async def get_chat_info(self, chat_id: str) -> Dict[str, Any]:
@@ -394,13 +348,7 @@ class SlackAdapter(BasePlatformAdapter):
"name": channel.get("name", chat_id),
"type": "dm" if is_dm else "group",
}
except Exception as e: # pragma: no cover - defensive logging
logger.error(
"[Slack] Failed to fetch chat info for %s: %s",
chat_id,
e,
exc_info=True,
)
except Exception:
return {"name": chat_id, "type": "unknown"}
# ----- Internal handlers -----
@@ -455,8 +403,8 @@ class SlackAdapter(BasePlatformAdapter):
media_urls.append(cached)
media_types.append(mimetype)
msg_type = MessageType.PHOTO
except Exception as e: # pragma: no cover - defensive logging
logger.warning("[Slack] Failed to cache image from %s: %s", url, e, exc_info=True)
except Exception as e:
print(f"[Slack] Failed to cache image: {e}", flush=True)
elif mimetype.startswith("audio/") and url:
try:
ext = "." + mimetype.split("/")[-1].split(";")[0]
@@ -466,8 +414,8 @@ class SlackAdapter(BasePlatformAdapter):
media_urls.append(cached)
media_types.append(mimetype)
msg_type = MessageType.VOICE
except Exception as e: # pragma: no cover - defensive logging
logger.warning("[Slack] Failed to cache audio from %s: %s", url, e, exc_info=True)
except Exception as e:
print(f"[Slack] Failed to cache audio: {e}", flush=True)
elif url:
# Try to handle as a document attachment
try:
@@ -489,7 +437,7 @@ class SlackAdapter(BasePlatformAdapter):
file_size = f.get("size", 0)
MAX_DOC_BYTES = 20 * 1024 * 1024
if not file_size or file_size > MAX_DOC_BYTES:
logger.warning("[Slack] Document too large or unknown size: %s", file_size)
print(f"[Slack] Document too large or unknown size: {file_size}", flush=True)
continue
# Download and cache
@@ -501,7 +449,7 @@ class SlackAdapter(BasePlatformAdapter):
media_urls.append(cached_path)
media_types.append(doc_mime)
msg_type = MessageType.DOCUMENT
logger.debug("[Slack] Cached user document: %s", cached_path)
print(f"[Slack] Cached user document: {cached_path}", flush=True)
# Inject text content for .txt/.md files (capped at 100 KB)
MAX_TEXT_INJECT_BYTES = 100 * 1024
@@ -518,8 +466,8 @@ class SlackAdapter(BasePlatformAdapter):
except UnicodeDecodeError:
pass # Binary content, skip injection
except Exception as e: # pragma: no cover - defensive logging
logger.warning("[Slack] Failed to cache document from %s: %s", url, e, exc_info=True)
except Exception as e:
print(f"[Slack] Failed to cache document: {e}", flush=True)
# Build source
source = self.build_source(

View File

@@ -35,6 +35,7 @@ COMMANDS_BY_CATEGORY = {
"/prompt": "View/set custom system prompt",
"/personality": "Set a predefined personality",
"/verbose": "Cycle tool progress display: off → new → all → verbose",
"/reasoning": "Manage reasoning effort and display (usage: /reasoning [level|show|hide])",
"/skin": "Show or change the display skin/theme",
},
"Tools & Skills": {

View File

@@ -143,6 +143,7 @@ DEFAULT_CONFIG = {
"personality": "kawaii",
"resume_display": "full",
"bell_on_complete": False,
"show_reasoning": False,
"skin": "default",
},
@@ -1025,6 +1026,14 @@ def show_config():
print(f" Max turns: {config.get('agent', {}).get('max_turns', DEFAULT_CONFIG['agent']['max_turns'])}")
print(f" Toolsets: {', '.join(config.get('toolsets', ['all']))}")
# Display
print()
print(color("◆ Display", Colors.CYAN, Colors.BOLD))
display = config.get('display', {})
print(f" Personality: {display.get('personality', 'kawaii')}")
print(f" Reasoning: {'on' if display.get('show_reasoning', False) else 'off'}")
print(f" Bell: {'on' if display.get('bell_on_complete', False) else 'off'}")
# Terminal
print()
print(color("◆ Terminal", Colors.CYAN, Colors.BOLD))

View File

@@ -173,6 +173,7 @@ class AIAgent:
session_id: str = None,
tool_progress_callback: callable = None,
thinking_callback: callable = None,
reasoning_callback: callable = None,
clarify_callback: callable = None,
step_callback: callable = None,
max_tokens: int = None,
@@ -260,6 +261,7 @@ class AIAgent:
self.tool_progress_callback = tool_progress_callback
self.thinking_callback = thinking_callback
self.reasoning_callback = reasoning_callback
self.clarify_callback = clarify_callback
self.step_callback = step_callback
self._last_reported_tool = None # Track for "new tool" mode
@@ -2420,6 +2422,12 @@ class AIAgent:
preview = reasoning_text[:100] + "..." if len(reasoning_text) > 100 else reasoning_text
logging.debug(f"Captured reasoning ({len(reasoning_text)} chars): {preview}")
if reasoning_text and self.reasoning_callback:
try:
self.reasoning_callback(reasoning_text)
except Exception:
pass
msg = {
"role": "assistant",
"content": assistant_message.content or "",
@@ -4470,9 +4478,17 @@ class AIAgent:
if final_response and not interrupted:
self._honcho_sync(original_user_message, final_response)
# Extract reasoning from the last assistant message (if any)
last_reasoning = None
for msg in reversed(messages):
if msg.get("role") == "assistant" and msg.get("reasoning"):
last_reasoning = msg["reasoning"]
break
# Build result with interrupt info if applicable
result = {
"final_response": final_response,
"last_reasoning": last_reasoning,
"messages": messages,
"api_calls": api_call_count,
"completed": completed,

View File

@@ -11,7 +11,7 @@ EXPECTED_COMMANDS = {
"/help", "/tools", "/toolsets", "/model", "/provider", "/prompt",
"/personality", "/clear", "/history", "/new", "/reset", "/retry",
"/undo", "/save", "/config", "/cron", "/skills", "/platforms",
"/verbose", "/compress", "/title", "/usage", "/insights", "/paste",
"/verbose", "/reasoning", "/compress", "/title", "/usage", "/insights", "/paste",
"/reload-mcp", "/rollback", "/background", "/skin", "/quit",
}

View File

@@ -0,0 +1,422 @@
"""Tests for the combined /reasoning command.
Covers both reasoning effort level management and reasoning display toggle,
plus the reasoning extraction and display pipeline from run_agent through CLI.
Combines functionality from:
- PR #789 (Aum08Desai): reasoning effort level management
- PR #790 (0xbyt4): reasoning display toggle and rendering
"""
import unittest
from types import SimpleNamespace
from unittest.mock import MagicMock, patch
# ---------------------------------------------------------------------------
# Effort level parsing
# ---------------------------------------------------------------------------
class TestParseReasoningConfig(unittest.TestCase):
"""Verify _parse_reasoning_config handles all effort levels."""
def _parse(self, effort):
from cli import _parse_reasoning_config
return _parse_reasoning_config(effort)
def test_none_disables(self):
result = self._parse("none")
self.assertEqual(result, {"enabled": False})
def test_valid_levels(self):
for level in ("low", "medium", "high", "xhigh", "minimal"):
result = self._parse(level)
self.assertIsNotNone(result)
self.assertTrue(result.get("enabled"))
self.assertEqual(result["effort"], level)
def test_empty_returns_none(self):
self.assertIsNone(self._parse(""))
self.assertIsNone(self._parse(" "))
def test_unknown_returns_none(self):
self.assertIsNone(self._parse("ultra"))
self.assertIsNone(self._parse("turbo"))
def test_case_insensitive(self):
result = self._parse("HIGH")
self.assertIsNotNone(result)
self.assertEqual(result["effort"], "high")
# ---------------------------------------------------------------------------
# /reasoning command handler (combined effort + display)
# ---------------------------------------------------------------------------
class TestHandleReasoningCommand(unittest.TestCase):
"""Test the combined _handle_reasoning_command method."""
def _make_cli(self, reasoning_config=None, show_reasoning=False):
"""Create a minimal CLI stub with the reasoning attributes."""
stub = SimpleNamespace(
reasoning_config=reasoning_config,
show_reasoning=show_reasoning,
agent=MagicMock(),
)
return stub
def test_show_enables_display(self):
stub = self._make_cli(show_reasoning=False)
# Simulate /reasoning show
arg = "show"
if arg in ("show", "on"):
stub.show_reasoning = True
stub.agent.reasoning_callback = lambda x: None
self.assertTrue(stub.show_reasoning)
def test_hide_disables_display(self):
stub = self._make_cli(show_reasoning=True)
# Simulate /reasoning hide
arg = "hide"
if arg in ("hide", "off"):
stub.show_reasoning = False
stub.agent.reasoning_callback = None
self.assertFalse(stub.show_reasoning)
self.assertIsNone(stub.agent.reasoning_callback)
def test_on_enables_display(self):
stub = self._make_cli(show_reasoning=False)
arg = "on"
if arg in ("show", "on"):
stub.show_reasoning = True
self.assertTrue(stub.show_reasoning)
def test_off_disables_display(self):
stub = self._make_cli(show_reasoning=True)
arg = "off"
if arg in ("hide", "off"):
stub.show_reasoning = False
self.assertFalse(stub.show_reasoning)
def test_effort_level_sets_config(self):
"""Setting an effort level should update reasoning_config."""
from cli import _parse_reasoning_config
stub = self._make_cli()
arg = "high"
parsed = _parse_reasoning_config(arg)
stub.reasoning_config = parsed
self.assertEqual(stub.reasoning_config, {"enabled": True, "effort": "high"})
def test_effort_none_disables_reasoning(self):
from cli import _parse_reasoning_config
stub = self._make_cli()
parsed = _parse_reasoning_config("none")
stub.reasoning_config = parsed
self.assertEqual(stub.reasoning_config, {"enabled": False})
def test_invalid_argument_rejected(self):
"""Invalid arguments should be rejected (parsed returns None)."""
from cli import _parse_reasoning_config
parsed = _parse_reasoning_config("turbo")
self.assertIsNone(parsed)
def test_no_args_shows_status(self):
"""With no args, should show current state (no crash)."""
stub = self._make_cli(reasoning_config=None, show_reasoning=False)
rc = stub.reasoning_config
if rc is None:
level = "medium (default)"
elif rc.get("enabled") is False:
level = "none (disabled)"
else:
level = rc.get("effort", "medium")
display_state = "on" if stub.show_reasoning else "off"
self.assertEqual(level, "medium (default)")
self.assertEqual(display_state, "off")
def test_status_with_disabled_reasoning(self):
stub = self._make_cli(reasoning_config={"enabled": False}, show_reasoning=True)
rc = stub.reasoning_config
if rc is None:
level = "medium (default)"
elif rc.get("enabled") is False:
level = "none (disabled)"
else:
level = rc.get("effort", "medium")
self.assertEqual(level, "none (disabled)")
def test_status_with_explicit_level(self):
stub = self._make_cli(
reasoning_config={"enabled": True, "effort": "xhigh"},
show_reasoning=True,
)
rc = stub.reasoning_config
level = rc.get("effort", "medium")
self.assertEqual(level, "xhigh")
# ---------------------------------------------------------------------------
# Reasoning extraction and result dict
# ---------------------------------------------------------------------------
class TestLastReasoningInResult(unittest.TestCase):
"""Verify reasoning extraction from the messages list."""
def _build_messages(self, reasoning=None):
return [
{"role": "user", "content": "hello"},
{
"role": "assistant",
"content": "Hi there!",
"reasoning": reasoning,
"finish_reason": "stop",
},
]
def test_reasoning_present(self):
messages = self._build_messages(reasoning="Let me think...")
last_reasoning = None
for msg in reversed(messages):
if msg.get("role") == "assistant" and msg.get("reasoning"):
last_reasoning = msg["reasoning"]
break
self.assertEqual(last_reasoning, "Let me think...")
def test_reasoning_none(self):
messages = self._build_messages(reasoning=None)
last_reasoning = None
for msg in reversed(messages):
if msg.get("role") == "assistant" and msg.get("reasoning"):
last_reasoning = msg["reasoning"]
break
self.assertIsNone(last_reasoning)
def test_picks_last_assistant(self):
messages = [
{"role": "user", "content": "hello"},
{"role": "assistant", "content": "...", "reasoning": "first thought"},
{"role": "tool", "content": "result"},
{"role": "assistant", "content": "done!", "reasoning": "final thought"},
]
last_reasoning = None
for msg in reversed(messages):
if msg.get("role") == "assistant" and msg.get("reasoning"):
last_reasoning = msg["reasoning"]
break
self.assertEqual(last_reasoning, "final thought")
def test_empty_reasoning_treated_as_none(self):
messages = self._build_messages(reasoning="")
last_reasoning = None
for msg in reversed(messages):
if msg.get("role") == "assistant" and msg.get("reasoning"):
last_reasoning = msg["reasoning"]
break
self.assertIsNone(last_reasoning)
# ---------------------------------------------------------------------------
# Reasoning display collapse
# ---------------------------------------------------------------------------
class TestReasoningCollapse(unittest.TestCase):
"""Verify long reasoning is collapsed to 10 lines in the box."""
def test_short_reasoning_not_collapsed(self):
reasoning = "\n".join(f"Line {i}" for i in range(5))
lines = reasoning.strip().splitlines()
self.assertLessEqual(len(lines), 10)
def test_long_reasoning_collapsed(self):
reasoning = "\n".join(f"Line {i}" for i in range(25))
lines = reasoning.strip().splitlines()
self.assertTrue(len(lines) > 10)
if len(lines) > 10:
display = "\n".join(lines[:10])
display += f"\n ... ({len(lines) - 10} more lines)"
display_lines = display.splitlines()
self.assertEqual(len(display_lines), 11)
self.assertIn("15 more lines", display_lines[-1])
def test_exactly_10_lines_not_collapsed(self):
reasoning = "\n".join(f"Line {i}" for i in range(10))
lines = reasoning.strip().splitlines()
self.assertEqual(len(lines), 10)
self.assertFalse(len(lines) > 10)
def test_intermediate_callback_collapses_to_5(self):
"""_on_reasoning shows max 5 lines."""
reasoning = "\n".join(f"Step {i}" for i in range(12))
lines = reasoning.strip().splitlines()
if len(lines) > 5:
preview = "\n".join(lines[:5])
preview += f"\n ... ({len(lines) - 5} more lines)"
else:
preview = reasoning.strip()
preview_lines = preview.splitlines()
self.assertEqual(len(preview_lines), 6)
self.assertIn("7 more lines", preview_lines[-1])
# ---------------------------------------------------------------------------
# Reasoning callback
# ---------------------------------------------------------------------------
class TestReasoningCallback(unittest.TestCase):
"""Verify reasoning_callback invocation."""
def test_callback_invoked_with_reasoning(self):
captured = []
agent = MagicMock()
agent.reasoning_callback = lambda t: captured.append(t)
agent._extract_reasoning = MagicMock(return_value="deep thought")
reasoning_text = agent._extract_reasoning(MagicMock())
if reasoning_text and agent.reasoning_callback:
agent.reasoning_callback(reasoning_text)
self.assertEqual(captured, ["deep thought"])
def test_callback_not_invoked_without_reasoning(self):
captured = []
agent = MagicMock()
agent.reasoning_callback = lambda t: captured.append(t)
agent._extract_reasoning = MagicMock(return_value=None)
reasoning_text = agent._extract_reasoning(MagicMock())
if reasoning_text and agent.reasoning_callback:
agent.reasoning_callback(reasoning_text)
self.assertEqual(captured, [])
def test_callback_none_does_not_crash(self):
reasoning_text = "some thought"
callback = None
if reasoning_text and callback:
callback(reasoning_text)
# No exception = pass
# ---------------------------------------------------------------------------
# Real provider format extraction
# ---------------------------------------------------------------------------
class TestExtractReasoningFormats(unittest.TestCase):
"""Test _extract_reasoning with real provider response formats."""
def _get_extractor(self):
from run_agent import AIAgent
return AIAgent._extract_reasoning
def test_openrouter_reasoning_details(self):
extract = self._get_extractor()
msg = SimpleNamespace(
reasoning=None,
reasoning_content=None,
reasoning_details=[
{"type": "reasoning.summary", "summary": "Analyzing Python lists."},
],
)
result = extract(None, msg)
self.assertIn("Python lists", result)
def test_deepseek_reasoning_field(self):
extract = self._get_extractor()
msg = SimpleNamespace(
reasoning="Solving step by step.\nx + y = 8.",
reasoning_content=None,
)
result = extract(None, msg)
self.assertIn("x + y = 8", result)
def test_moonshot_reasoning_content(self):
extract = self._get_extractor()
msg = SimpleNamespace(
reasoning_content="Explaining async/await.",
)
result = extract(None, msg)
self.assertIn("async/await", result)
def test_no_reasoning_returns_none(self):
extract = self._get_extractor()
msg = SimpleNamespace(content="Hello!")
result = extract(None, msg)
self.assertIsNone(result)
# ---------------------------------------------------------------------------
# Config defaults
# ---------------------------------------------------------------------------
class TestConfigDefault(unittest.TestCase):
"""Verify config default for show_reasoning."""
def test_default_config_has_show_reasoning(self):
from hermes_cli.config import DEFAULT_CONFIG
display = DEFAULT_CONFIG.get("display", {})
self.assertIn("show_reasoning", display)
self.assertFalse(display["show_reasoning"])
class TestCommandRegistered(unittest.TestCase):
"""Verify /reasoning is in the COMMANDS dict."""
def test_reasoning_in_commands(self):
from hermes_cli.commands import COMMANDS
self.assertIn("/reasoning", COMMANDS)
# ---------------------------------------------------------------------------
# End-to-end pipeline
# ---------------------------------------------------------------------------
class TestEndToEndPipeline(unittest.TestCase):
"""Simulate the full pipeline: extraction -> result dict -> display."""
def test_openrouter_claude_pipeline(self):
from run_agent import AIAgent
api_message = SimpleNamespace(
role="assistant",
content="Lists support append().",
tool_calls=None,
reasoning=None,
reasoning_content=None,
reasoning_details=[
{"type": "reasoning.summary", "summary": "Python list methods."},
],
)
reasoning = AIAgent._extract_reasoning(None, api_message)
self.assertIsNotNone(reasoning)
messages = [
{"role": "user", "content": "How do I add items?"},
{"role": "assistant", "content": api_message.content, "reasoning": reasoning},
]
last_reasoning = None
for msg in reversed(messages):
if msg.get("role") == "assistant" and msg.get("reasoning"):
last_reasoning = msg["reasoning"]
break
result = {
"final_response": api_message.content,
"last_reasoning": last_reasoning,
}
self.assertIn("last_reasoning", result)
self.assertIn("Python list methods", result["last_reasoning"])
def test_no_reasoning_model_pipeline(self):
from run_agent import AIAgent
api_message = SimpleNamespace(content="Paris.", tool_calls=None)
reasoning = AIAgent._extract_reasoning(None, api_message)
self.assertIsNone(reasoning)
result = {"final_response": api_message.content, "last_reasoning": reasoning}
self.assertIsNone(result["last_reasoning"])
if __name__ == "__main__":
unittest.main()

View File

@@ -147,6 +147,7 @@ Type `/` in the interactive CLI to see an autocomplete dropdown.
| `/config` | Show current configuration |
| `/prompt [text]` | View/set custom system prompt |
| `/personality [name]` | Set a predefined personality |
| `/reasoning [arg]` | Manage reasoning effort and display. Args: effort level (`none`, `low`, `medium`, `high`, `xhigh`) or display toggle (`show`, `hide`). No args shows current state. |
### Conversation

View File

@@ -104,6 +104,7 @@ Type `/` to see an autocomplete dropdown of all available commands.
| `/config` | Show current configuration |
| `/prompt [text]` | View/set/clear custom system prompt |
| `/personality [name]` | Set a predefined personality |
| `/reasoning [arg]` | Manage reasoning effort (`none`/`low`/`medium`/`high`/`xhigh`) and display (`show`/`hide`) |
### Conversation Management

View File

@@ -608,6 +608,16 @@ agent:
When unset (default), reasoning effort defaults to "medium" — a balanced level that works well for most tasks. Setting a value overrides it — higher reasoning effort gives better results on complex tasks at the cost of more tokens and latency.
You can also change the reasoning effort at runtime with the `/reasoning` command:
```
/reasoning # Show current effort level and display state
/reasoning high # Set reasoning effort to high
/reasoning none # Disable reasoning
/reasoning show # Show model thinking above each response
/reasoning hide # Hide model thinking
```
## TTS Configuration
```yaml
@@ -632,6 +642,7 @@ display:
compact: false # Compact output mode (less whitespace)
resume_display: full # full (show previous messages on resume) | minimal (one-liner only)
bell_on_complete: false # Play terminal bell when agent finishes (great for long tasks)
show_reasoning: false # Show model reasoning/thinking above each response (toggle with /reasoning show|hide)
```
| Mode | What you see |