97de4b4ef8
Conflict resolutions and semantic fixups: - utils.py / hermes_yaml.py: main widened ruamel's round-trip emitter so a long double-quoted scalar is never folded after an escaped backslash. pm-clean builds every rt emitter through hermes_yaml.roundtrip_yaml(), so the width lives there (ROUNDTRIP_YAML_WIDTH moves with it); xai_retirement imports it from hermes_yaml. - hermes_cli/banner.py: keep pm-clean's removal of the banner update check. Main's GIT_NO_LAZY_FETCH fix for it applies to its replacement, source_check: every read-only probe (source_git_env) now refuses promisor lazy fetches, and the partial-clone test targets that probe (red without the flag). - .github/workflows/tests.yml: keep setup-pm; main's uv pin bump does not apply. Main's WAL-capable SQLite gates are kept, run against $HERMES_PYTHON (the PM-pinned interpreter, SQLite 3.53.1). The e2e step takes main's --include-integration invocation. - apps/desktop: package.json has no build block here, so main's macOS locale-marker restore joins the darwin branch of the existing after-pack.mjs, and its test loads the hook from electron-builder.config.cjs and imports PlatformPackager from app-builder-lib's root (electron-builder 27 exports no ./out paths). The win32 row is dropped: this hook sanitizes and signs PE trees on win32 by design. - reconciliation.ts: main's rowId hydration (#119326) was merged into the first of pm-clean's split helpers only; the resolver is now one helper both halves use. - en.ts: both sides' keys kept. tests/tools/test_lazy_deps.py stays deleted. - Tests main added with `import yaml` use hermes_yaml, like the rest of the tree.
311 lines
15 KiB
Python
311 lines
15 KiB
Python
"""Headless slash commands for ACP sessions (``/help``, ``/model``, ``/compress`` ...)."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import contextvars
|
|
import logging
|
|
from collections import Counter
|
|
from typing import Any
|
|
|
|
from acp.schema import AvailableCommand, AvailableCommandsUpdate, UnstructuredCommandInput
|
|
|
|
from acp_adapter.session import SessionState, _expand_acp_enabled_toolsets
|
|
|
|
logger = logging.getLogger("acp_adapter.server")
|
|
|
|
|
|
def _estimate_tokens(history: list, agent: Any, system_prompt: str | None = None, tools: Any = None) -> int:
|
|
"""Rough request-token estimate over history + system prompt + tool schemas."""
|
|
from agent.model_metadata import estimate_request_tokens_rough
|
|
|
|
if system_prompt is None:
|
|
system_prompt = getattr(agent, "_cached_system_prompt", "") or ""
|
|
if tools is None:
|
|
tools = getattr(agent, "tools", None) or None
|
|
return estimate_request_tokens_rough(history, system_prompt=system_prompt, tools=tools)
|
|
|
|
|
|
def _queue_prompt(state: SessionState, text: str) -> int:
|
|
with state.runtime_lock:
|
|
state.queued_prompts.append(text)
|
|
return len(state.queued_prompts)
|
|
|
|
|
|
# Commands that mutate shared turn state must not run beside a live turn or beside each
|
|
# other: slash dispatch happens on a worker thread while the turn iterates state.history and
|
|
# reads agent._session_db, so clearing, rebinding, or swapping state.agent underneath
|
|
# run_conversation tears the running turn. Gateway parity: all three are idle-only there.
|
|
# The command_op flag is held for the whole handler so a turn cannot claim the session in
|
|
# the check-then-act window (the /compress LLM call and /model agent rebuild take seconds).
|
|
_MID_TURN_BLOCKED_COMMANDS = frozenset({"reset", "compress", "model"})
|
|
|
|
|
|
class SlashCommandsMixin:
|
|
"""Slash-command surface for ``HermesACPAgent``; relies on ``_conn``, ``_send``, ``_schedule_soon``,
|
|
``session_manager`` and ``_switch_model`` from the host class."""
|
|
|
|
# name -> (help text, advertised description, input hint)
|
|
_COMMANDS: dict[str, tuple[str, str, str | None]] = {
|
|
"help": ("Show available commands", "List available commands", None),
|
|
"model": (
|
|
"Show or change current model",
|
|
"Show current model and provider, or switch models",
|
|
"model name to switch to",
|
|
),
|
|
"tools": ("List available tools", "List available tools with descriptions", None),
|
|
"context": ("Show conversation context info", "Show conversation message counts by role", None),
|
|
"reset": ("Clear conversation history", "Clear conversation history", None),
|
|
"compress": ("Compress conversation context", "Compress conversation context", None),
|
|
"steer": (
|
|
"Inject guidance into the currently running agent turn",
|
|
"Inject guidance into the currently running agent turn",
|
|
"guidance for the active turn",
|
|
),
|
|
"queue": (
|
|
"Queue a prompt to run after the current turn finishes",
|
|
"Queue a prompt to run after the current turn finishes",
|
|
"prompt to run next",
|
|
),
|
|
"version": ("Show Hermes version", "Show Hermes version", None),
|
|
}
|
|
|
|
|
|
@classmethod
|
|
def _available_commands(cls) -> list[AvailableCommand]:
|
|
return [
|
|
AvailableCommand(name=name, description=desc, input=UnstructuredCommandInput(hint=hint) if hint else None)
|
|
for name, (_help, desc, hint) in cls._COMMANDS.items()
|
|
]
|
|
|
|
async def _send_available_commands_update(self, session_id: str) -> None:
|
|
"""Advertise supported slash commands to the connected ACP client."""
|
|
if not self._conn:
|
|
return
|
|
update = AvailableCommandsUpdate(
|
|
session_update="available_commands_update", available_commands=self._available_commands()
|
|
)
|
|
await self._send(session_id, update, fail_msg="Failed to advertise ACP slash commands for session %s")
|
|
|
|
def _schedule_available_commands_update(self, session_id: str) -> None:
|
|
self._schedule_soon(lambda: self._send_available_commands_update(session_id))
|
|
|
|
def _handle_slash_command(self, text: str, state: SessionState) -> str | None:
|
|
"""Dispatch a slash command; ``None`` for unknown ones so they fall through to the LLM."""
|
|
parts = text.split(maxsplit=1)
|
|
cmd = parts[0].lstrip("/").lower()
|
|
args = parts[1].strip() if len(parts) > 1 else ""
|
|
|
|
if cmd not in self._COMMANDS:
|
|
return None
|
|
mutating = cmd in _MID_TURN_BLOCKED_COMMANDS
|
|
if mutating:
|
|
with state.runtime_lock:
|
|
if state.is_running or state.command_op:
|
|
return (f"⏳ Session is busy; /{cmd} only works while the session is "
|
|
"idle. Wait for the current response or cancel first.")
|
|
state.command_op = True
|
|
handler = getattr(self, f"_cmd_{cmd}")
|
|
|
|
# Handlers run outside the per-turn cwd-pinning context. ``/compress``
|
|
# and ``/model`` REBUILD the system prompt, so unpinned they'd bake the Hermes install tree
|
|
# into the persisted cached prompt. Pin inside a fresh context: no leak, no teardown.
|
|
def _dispatch() -> str | None:
|
|
try:
|
|
from agent.runtime_cwd import set_session_cwd
|
|
|
|
set_session_cwd(state.cwd)
|
|
except Exception:
|
|
logger.debug("Could not pin ACP session cwd for slash command", exc_info=True)
|
|
return handler(args, state)
|
|
|
|
try:
|
|
return contextvars.copy_context().run(_dispatch)
|
|
except Exception as e:
|
|
logger.error("Slash command /%s error: %s", cmd, e, exc_info=True)
|
|
return f"Error executing /{cmd}: {e}"
|
|
finally:
|
|
if mutating:
|
|
with state.runtime_lock:
|
|
state.command_op = False
|
|
|
|
def _cmd_help(self, args: str, state: SessionState) -> str:
|
|
lines = ["Available commands:", ""]
|
|
lines.extend(f" /{cmd:10s} {desc}" for cmd, (desc, _adv, _hint) in self._COMMANDS.items())
|
|
lines.extend(["", "Unrecognized /commands are sent to the model as normal messages."])
|
|
return "\n".join(lines)
|
|
|
|
def _cmd_model(self, args: str, state: SessionState) -> str:
|
|
if not args:
|
|
model = state.model or getattr(state.agent, "model", "unknown")
|
|
provider = getattr(state.agent, "provider", None) or "auto"
|
|
return f"Current model: {model}\nProvider: {provider}"
|
|
|
|
current_provider, target_provider, new_model = self._switch_model(state, args)
|
|
provider_label = getattr(state.agent, "provider", None) or target_provider or current_provider or "openrouter"
|
|
logger.info("Session %s: model switched to %s", state.session_id, new_model)
|
|
return f"Model switched to: {new_model}\nProvider: {provider_label}"
|
|
|
|
def _cmd_tools(self, args: str, state: SessionState) -> str:
|
|
try:
|
|
from model_tools import get_tool_definitions
|
|
from types import SimpleNamespace
|
|
from agent.memory_manager import inject_memory_provider_tools
|
|
|
|
toolsets = _expand_acp_enabled_toolsets(getattr(state.agent, "enabled_toolsets", None))
|
|
tools = get_tool_definitions(
|
|
enabled_toolsets=toolsets,
|
|
disabled_toolsets=getattr(state.agent, "disabled_toolsets", None),
|
|
quiet_mode=True,
|
|
)
|
|
tool_view = SimpleNamespace(
|
|
tools=list(tools or []),
|
|
valid_tool_names={t.get("function", {}).get("name") for t in tools or [] if isinstance(t, dict)},
|
|
enabled_toolsets=toolsets,
|
|
disabled_toolsets=getattr(state.agent, "disabled_toolsets", None),
|
|
_memory_manager=getattr(state.agent, "_memory_manager", None),
|
|
)
|
|
inject_memory_provider_tools(tool_view)
|
|
tools = tool_view.tools
|
|
if not tools:
|
|
return "No tools available."
|
|
lines = [f"Available tools ({len(tools)}):"]
|
|
for t in tools:
|
|
name = (t.get("function") or {}).get("name", "?")
|
|
desc = (t.get("function") or {}).get("description", "")
|
|
if len(desc) > 80:
|
|
desc = desc[:77] + "..."
|
|
lines.append(f" {name}: {desc}")
|
|
return "\n".join(lines)
|
|
except Exception as e:
|
|
return f"Could not list tools: {e}"
|
|
|
|
def _cmd_context(self, args: str, state: SessionState) -> str:
|
|
"""Show ACP session context pressure and compression guidance."""
|
|
n_messages = len(state.history)
|
|
roles = Counter(msg.get("role", "unknown") for msg in state.history)
|
|
|
|
agent = state.agent
|
|
model = state.model or getattr(agent, "model", "")
|
|
provider = getattr(agent, "provider", None) or "auto"
|
|
compressor = getattr(agent, "context_compressor", None)
|
|
context_length = int(getattr(compressor, "context_length", 0) or 0)
|
|
threshold_tokens = int(getattr(compressor, "threshold_tokens", 0) or 0)
|
|
|
|
try:
|
|
approx_tokens = _estimate_tokens(state.history, agent)
|
|
except Exception:
|
|
logger.debug("Could not estimate ACP context usage", exc_info=True)
|
|
approx_tokens = 0
|
|
|
|
if threshold_tokens <= 0 and context_length > 0:
|
|
threshold_tokens = int(context_length * 0.80)
|
|
|
|
lines = [
|
|
f"Conversation: {n_messages} messages" if n_messages else "Conversation is empty (no messages yet).",
|
|
f" user: {roles.get('user', 0)}, assistant: {roles.get('assistant', 0)}, "
|
|
f"tool: {roles.get('tool', 0)}, system: {roles.get('system', 0)}",
|
|
]
|
|
if model:
|
|
lines.append(f"Model: {model}")
|
|
lines.append(f"Provider: {provider}")
|
|
|
|
if approx_tokens > 0 and context_length > 0:
|
|
usage_pct = (approx_tokens / context_length) * 100
|
|
lines.append(f"Context usage: ~{approx_tokens:,} / {context_length:,} tokens ({usage_pct:.1f}%)")
|
|
elif approx_tokens > 0:
|
|
lines.append(f"Context usage: ~{approx_tokens:,} tokens")
|
|
|
|
if threshold_tokens > 0 and approx_tokens > 0:
|
|
threshold_pct = (threshold_tokens / context_length) * 100 if context_length > 0 else 0
|
|
pct_note = f", {threshold_pct:.0f}%" if threshold_pct else ""
|
|
if approx_tokens >= threshold_tokens:
|
|
lines.append(f"Compression: due now (threshold ~{threshold_tokens:,}{pct_note}). Run /compress.")
|
|
else:
|
|
remaining = max(threshold_tokens - approx_tokens, 0)
|
|
lines.append(f"Compression: ~{remaining:,} tokens until threshold (~{threshold_tokens:,}{pct_note}).")
|
|
elif threshold_tokens > 0:
|
|
lines.append(f"Compression threshold: ~{threshold_tokens:,} tokens")
|
|
|
|
lines.append(
|
|
"Auto-compaction is disabled (compression.enabled: false); /compress still compresses manually."
|
|
if getattr(agent, "compression_enabled", True) is False
|
|
else "Tip: run /compress to compress manually before the threshold."
|
|
)
|
|
return "\n".join(lines)
|
|
|
|
def _cmd_reset(self, args: str, state: SessionState) -> str:
|
|
state.history.clear()
|
|
try:
|
|
reset_session_state = getattr(state.agent, "reset_session_state", None)
|
|
if callable(reset_session_state):
|
|
reset_session_state()
|
|
except Exception:
|
|
logger.warning("ACP session state reset failed for %s", state.session_id, exc_info=True)
|
|
return "Conversation history cleared. Agent session state reset failed; see logs."
|
|
finally:
|
|
self.session_manager.save_session(state.session_id)
|
|
return "Conversation history cleared."
|
|
|
|
def _cmd_compress(self, args: str, state: SessionState) -> str:
|
|
"""``/compress [here [N] | <focus>] [--preview] [--aggressive]`` through the shared core."""
|
|
from agent.conversation_compression import finalize_context_engine_compression_notification
|
|
from agent.conversation_compression_manual import (
|
|
AGGRESSIVE_UNSUPPORTED, compress_now, parse_compress_args, render_compress_result)
|
|
|
|
if not state.history:
|
|
return "Nothing to compress — conversation is empty."
|
|
agent = state.agent
|
|
# No compression_enabled gate: it only disables *automatic* compaction (CLI/gateway parity).
|
|
if not hasattr(agent, "_compress_context"):
|
|
return "Context compression not available for this agent."
|
|
request = parse_compress_args(args)
|
|
if request.aggressive:
|
|
return AGGRESSIVE_UNSUPPORTED
|
|
original_session_db = getattr(agent, "_session_db", None)
|
|
try:
|
|
# Stable ACP session id: suppress _compress_context's SQLite session split.
|
|
agent._session_db = None
|
|
result = compress_now(
|
|
agent, state.history, request, system_message=getattr(agent, "_cached_system_prompt", "") or "",
|
|
task_id=state.session_id)
|
|
except Exception as e:
|
|
return f"Compression failed: {e}"
|
|
finally:
|
|
agent._session_db = original_session_db
|
|
if result.status != "compressed":
|
|
return "\n".join(render_compress_result(result))
|
|
state.history = result.after_messages
|
|
self.session_manager.save_session(state.session_id)
|
|
finalize_context_engine_compression_notification(agent, committed=True)
|
|
return (
|
|
f"Context compressed: {len(result.before_messages)} -> {len(state.history)} messages\n"
|
|
f"~{result.before_tokens:,} -> ~{result.after_tokens:,} tokens"
|
|
)
|
|
|
|
def _cmd_steer(self, args: str, state: SessionState) -> str:
|
|
steer_text = args.strip()
|
|
if not steer_text:
|
|
return "Usage: /steer <guidance>"
|
|
|
|
if state.is_running and hasattr(state.agent, "steer"):
|
|
try:
|
|
if state.agent.steer(steer_text):
|
|
preview = steer_text[:80] + ("..." if len(steer_text) > 80 else "")
|
|
return f"⏩ Steer queued for the active turn: {preview}"
|
|
except Exception as exc:
|
|
logger.warning("ACP steer failed for session %s: %s", state.session_id, exc)
|
|
return f"⚠️ Steer failed: {exc}"
|
|
|
|
return f"No active turn — queued for the next turn. ({_queue_prompt(state, steer_text)} queued)"
|
|
|
|
def _cmd_queue(self, args: str, state: SessionState) -> str:
|
|
queued_text = args.strip()
|
|
if not queued_text:
|
|
return "Usage: /queue <prompt>"
|
|
return f"Queued for the next turn. ({_queue_prompt(state, queued_text)} queued)"
|
|
|
|
def _cmd_version(self, args: str, state: SessionState) -> str:
|
|
from hermes_cli.version_info import get_version_info
|
|
|
|
return f"Hermes Agent v{get_version_info().derived_version}"
|