feat(cli,gateway): /new accepts optional session name argument

Allow users to start a fresh session and immediately set its title by
passing a name to /new (or /reset):

    /new Refactor auth module

Changes:
- hermes_cli/commands.py: add args_hint='[name]' to /new command
- cli.py: parse title argument in process_command(), pass to new_session()
- cli.py: new_session() accepts title=None, sets title via SessionDB
- gateway/run.py: _handle_reset_command() parses title, sets on new entry
- gateway/session.py: reset_session() accepts optional display_name
- tests: add test_new_session_with_title, test_reset_command_with_title,
  test_new_command_in_help_output

All 36 affected tests pass.
This commit is contained in:
Exx 2026-05-04 06:20:19 +00:00 committed by Teknium
parent 055fde40e0
commit f720751d79
6 changed files with 138 additions and 9 deletions

18
cli.py
View file

@ -4932,7 +4932,7 @@ class HermesCLI:
except Exception: except Exception:
pass pass
def new_session(self, silent=False): def new_session(self, silent=False, title=None):
"""Start a fresh session with a new session ID and cleared agent state.""" """Start a fresh session with a new session ID and cleared agent state."""
if self.agent and self.conversation_history: if self.agent and self.conversation_history:
# Trigger memory extraction on the old session before session_id rotates. # Trigger memory extraction on the old session before session_id rotates.
@ -4987,6 +4987,15 @@ class HermesCLI:
self.agent._session_db_created = True self.agent._session_db_created = True
except Exception: except Exception:
pass pass
if title and self._session_db:
try:
from hermes_state import SessionDB
sanitized = SessionDB.sanitize_title(title)
if sanitized:
self._session_db.set_session_title(self.session_id, sanitized)
self._pending_title = None
except Exception:
pass
# Notify memory providers that session_id rotated to a fresh # Notify memory providers that session_id rotated to a fresh
# conversation. reset=True signals providers to flush accumulated # conversation. reset=True signals providers to flush accumulated
# per-session state (_session_turns, _turn_counter, _document_id). # per-session state (_session_turns, _turn_counter, _document_id).
@ -5006,6 +5015,9 @@ class HermesCLI:
self._notify_session_boundary("on_session_reset") self._notify_session_boundary("on_session_reset")
if not silent: if not silent:
if title:
print(f"(^_^)v New session started: {title}")
else:
print("(^_^)v New session started!") print("(^_^)v New session started!")
def _handle_resume_command(self, cmd_original: str) -> None: def _handle_resume_command(self, cmd_original: str) -> None:
@ -6418,7 +6430,9 @@ class HermesCLI:
else: else:
_cprint(" Session database not available.") _cprint(" Session database not available.")
elif canonical == "new": elif canonical == "new":
self.new_session() parts = cmd_original.split(maxsplit=1)
title = parts[1].strip() if len(parts) > 1 else None
self.new_session(title=title)
elif canonical == "resume": elif canonical == "resume":
self._handle_resume_command(cmd_original) self._handle_resume_command(cmd_original)
elif canonical == "model": elif canonical == "model":

View file

@ -6896,6 +6896,17 @@ class GatewayRunner:
new_entry = self.session_store.get_or_create_session(source, force_new=True) new_entry = self.session_store.get_or_create_session(source, force_new=True)
header = "✨ New session started!" header = "✨ New session started!"
# Set session title if provided with /new <title>
_title_arg = event.get_command_args().strip()
if _title_arg and self._session_db and new_entry:
try:
from hermes_state import SessionDB
sanitized = SessionDB.sanitize_title(_title_arg)
if sanitized:
self._session_db.set_session_title(new_entry.session_id, sanitized)
except Exception:
pass
# Fire plugin on_session_reset hook (new session guaranteed to exist) # Fire plugin on_session_reset hook (new session guaranteed to exist)
try: try:
from hermes_cli.plugins import invoke_hook as _invoke_hook from hermes_cli.plugins import invoke_hook as _invoke_hook

View file

@ -1121,7 +1121,7 @@ class SessionStore:
self._save() self._save()
return count return count
def reset_session(self, session_key: str) -> Optional[SessionEntry]: def reset_session(self, session_key: str, display_name: Optional[str] = None) -> Optional[SessionEntry]:
"""Force reset a session, creating a new session ID.""" """Force reset a session, creating a new session ID."""
db_end_session_id = None db_end_session_id = None
db_create_kwargs = None db_create_kwargs = None
@ -1145,7 +1145,7 @@ class SessionStore:
created_at=now, created_at=now,
updated_at=now, updated_at=now,
origin=old_entry.origin, origin=old_entry.origin,
display_name=old_entry.display_name, display_name=display_name if display_name is not None else old_entry.display_name,
platform=old_entry.platform, platform=old_entry.platform,
chat_type=old_entry.chat_type, chat_type=old_entry.chat_type,
is_fresh_reset=True, is_fresh_reset=True,

View file

@ -64,7 +64,7 @@ class CommandDef:
COMMAND_REGISTRY: list[CommandDef] = [ COMMAND_REGISTRY: list[CommandDef] = [
# Session # Session
CommandDef("new", "Start a new session (fresh session ID + history)", "Session", CommandDef("new", "Start a new session (fresh session ID + history)", "Session",
aliases=("reset",)), aliases=("reset",), args_hint="[name]"),
CommandDef("clear", "Clear screen and start a new session", "Session", CommandDef("clear", "Clear screen and start a new session", "Session",
cli_only=True), cli_only=True),
CommandDef("redraw", "Force a full UI repaint (recovers from terminal drift)", "Session", CommandDef("redraw", "Force a full UI repaint (recovers from terminal drift)", "Session",

View file

@ -5,7 +5,7 @@ from __future__ import annotations
import importlib import importlib
import os import os
import sys import sys
from datetime import timedelta from datetime import datetime, timedelta
from unittest.mock import MagicMock, patch from unittest.mock import MagicMock, patch
from hermes_state import SessionDB from hermes_state import SessionDB
@ -219,3 +219,22 @@ def test_new_session_resets_token_counters(tmp_path):
assert comp.last_total_tokens == 0 assert comp.last_total_tokens == 0
assert comp.compression_count == 0 assert comp.compression_count == 0
assert comp._context_probed is False assert comp._context_probed is False
def test_new_session_with_title(capsys):
"""new_session(title=...) creates a session and sets the title."""
cli = _make_cli()
cli._session_db = MagicMock()
cli.agent = _FakeAgent("old_session_id", datetime.now())
cli.conversation_history = []
cli.new_session(title="My Test Session")
# Assert set_session_title was called with the new session ID and sanitized title
cli._session_db.set_session_title.assert_called_once()
call_args = cli._session_db.set_session_title.call_args
assert call_args[0][0] == cli.session_id
assert call_args[0][1] == "My Test Session"
captured = capsys.readouterr()
assert "My Test Session" in captured.out

View file

@ -5,11 +5,12 @@ across all gateway messenger platforms.
""" """
import os import os
from unittest.mock import MagicMock, patch from types import SimpleNamespace
from unittest.mock import AsyncMock, MagicMock, patch
import pytest import pytest
from gateway.config import Platform from gateway.config import GatewayConfig, Platform, PlatformConfig
from gateway.platforms.base import MessageEvent from gateway.platforms.base import MessageEvent
from gateway.session import SessionSource from gateway.session import SessionSource
@ -206,3 +207,87 @@ class TestTitleInHelp:
import inspect import inspect
source = inspect.getsource(GatewayRunner._handle_message) source = inspect.getsource(GatewayRunner._handle_message)
assert '"title"' in source assert '"title"' in source
# ---------------------------------------------------------------------------
# /new with title
# ---------------------------------------------------------------------------
class TestResetCommandWithTitle:
"""Tests for GatewayRunner._handle_reset_command with a title argument."""
@pytest.mark.asyncio
async def test_reset_command_with_title(self):
"""Sending /new <title> resets session and sets the title."""
from datetime import datetime
from gateway.run import GatewayRunner
from gateway.session import SessionEntry, SessionSource, build_session_key
runner = object.__new__(GatewayRunner)
runner.config = GatewayConfig(
platforms={Platform.TELEGRAM: PlatformConfig(enabled=True, token="***")}
)
adapter = MagicMock()
adapter.send = AsyncMock()
runner.adapters = {Platform.TELEGRAM: adapter}
runner._voice_mode = {}
runner.hooks = SimpleNamespace(emit=AsyncMock(), loaded_hooks=False)
runner._session_model_overrides = {}
runner._pending_model_notes = {}
runner._background_tasks = set()
source = SessionSource(
platform=Platform.TELEGRAM,
user_id="12345",
chat_id="67890",
user_name="testuser",
)
session_key = build_session_key(source)
new_session_entry = SessionEntry(
session_key=session_key,
session_id="sess-new",
created_at=datetime.now(),
updated_at=datetime.now(),
platform=Platform.TELEGRAM,
chat_type="dm",
)
runner.session_store = MagicMock()
runner.session_store.get_or_create_session.return_value = new_session_entry
runner.session_store.reset_session.return_value = new_session_entry
runner.session_store._entries = {session_key: new_session_entry}
runner.session_store._generate_session_key.return_value = session_key
runner._running_agents = {}
runner._pending_messages = {}
runner._pending_approvals = {}
runner._session_db = MagicMock()
runner._agent_cache = {}
runner._agent_cache_lock = None
runner._is_user_authorized = lambda _source: True
runner._format_session_info = lambda: ""
event = _make_event(text="/new Custom Name")
result = await runner._handle_reset_command(event)
runner.session_store.reset_session.assert_called_once()
runner._session_db.set_session_title.assert_called_once_with(
"sess-new", "Custom Name"
)
# ---------------------------------------------------------------------------
# /new in help output
# ---------------------------------------------------------------------------
class TestNewInHelp:
"""Verify /new appears in help text with the [name] args hint."""
def test_new_command_in_help_output(self):
"""The gateway help output includes /new with the [name] hint."""
from hermes_cli.commands import gateway_help_lines
lines = gateway_help_lines()
new_line = next((line for line in lines if line.startswith("`/new ")), None)
assert new_line is not None
assert "[name]" in new_line