c13ea774e6
Runtime identity resolved through hermes_cli.__version__ (a static 0.0.0 on source installs, rewritten by release stamping) leaked v0.0.0 into About, /api/health, User-Agents, and plugin compat, and source updates showed "couldn't reach update server" because identity and channel authority disagreed with the checkout. Now: get_version_info() resolves install stamp -> live git -> unknown, never pyproject metadata, never a package constant. Source checkouts derive identity from their reachable release tag; the completion tail of every successful install/update/historical takeover atomically rewrites install-stamp.json with that identity; a stale source stamp whose commit no longer matches HEAD defers to live git. ACP/TUI use derived_version for display and base_version for protocol fields; all ~44 runtime __version__ consumers migrated; hermes_cli.__version__ and generated _version.py are gone; release stamping only touches the native manifests external builders consume (nix/tauri/cargo) and passes release identity straight into write_install_stamp.py; pyproject.toml stays inert 0.0.0. Desktop no longer synthesizes a competing install-stamp.json: the checkout owns its stamp, and desktop-bootstrap classification keys on the bootstrap-complete marker. verify-bootstrap-version-stamp.py now cross-checks the checkout's stamp (baseVersion + commit == HEAD). Validation: 31-file focused suite green (version identity, stamping, adoption, providers, gateway, acp/tui runtime identity, api server via extras env, release graph); desktop tsc + 25 vitest green; real-repo probe: base=unknown derived=git.0635606.dirty source=git on this checkout; clean-env imports resolve entirely from this tree; windows footgun + compat-pointer scans clean.
1126 lines
58 KiB
Python
1126 lines
58 KiB
Python
"""ACP agent server — exposes Hermes Agent via the Agent Client Protocol."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import asyncio
|
|
from datetime import datetime, timezone
|
|
import contextlib
|
|
import contextvars
|
|
import logging
|
|
import os
|
|
import threading
|
|
from collections import defaultdict, deque
|
|
from concurrent.futures import ThreadPoolExecutor
|
|
from dataclasses import dataclass
|
|
from typing import Any, Callable, Deque, Optional
|
|
|
|
import acp
|
|
from acp.schema import (
|
|
AgentCapabilities, AgentMessageChunk, AuthenticateResponse, ClientCapabilities, ForkSessionResponse,
|
|
Implementation, InitializeResponse, ListSessionsResponse, LoadSessionResponse, McpServerHttp, McpServerSse,
|
|
McpServerStdio, ModelInfo, NewSessionResponse, PromptCapabilities, PromptResponse, ResumeSessionResponse,
|
|
SessionCapabilities, SessionForkCapabilities, SessionInfo, SessionInfoUpdate, SessionListCapabilities,
|
|
SessionMode, SessionModeState, SessionModelState, SessionResumeCapabilities, SetSessionConfigOptionResponse,
|
|
SetSessionModeResponse, SetSessionModelResponse, TextContentBlock, Usage, UsageUpdate, UserMessageChunk,
|
|
)
|
|
|
|
from acp_adapter.auth import TERMINAL_SETUP_AUTH_METHOD_ID, build_auth_methods, detect_provider
|
|
from acp_adapter.commands import SlashCommandsMixin, _estimate_tokens
|
|
from acp_adapter.content import PromptBlock, _content_blocks_to_openai_user_content, _extract_text
|
|
from acp_adapter.events import (
|
|
AssistantMessageIdAllocator, _build_plan_update_from_todo_result, _send_update, flush_open_tool_calls,
|
|
make_message_cb, make_step_cb, make_thinking_cb, make_tool_progress_cb,
|
|
)
|
|
from acp_adapter.model_catalog import build_model_state, encode_model_choice
|
|
from acp_adapter.permissions import make_approval_callback
|
|
from acp_adapter.provenance import session_provenance_meta
|
|
from acp_adapter.session import SessionManager, SessionState, _expand_acp_enabled_toolsets
|
|
from acp_adapter.tools import build_tool_complete, build_tool_start, coerce_tool_args
|
|
from agent.context_compressor import (COMPRESSED_SUMMARY_METADATA_KEY, ContextCompressor)
|
|
from agent.interrupt_compat import request_hard_interrupt
|
|
from tools.approval_context import reset_hermes_interactive_context, set_hermes_interactive_context
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
# Runs the synchronous AIAgent off the event loop.
|
|
_executor = ThreadPoolExecutor(max_workers=4, thread_name_prefix="acp-agent")
|
|
|
|
# ListSessionsRequest has no client-side limit; clients paginate via `cursor`/`next_cursor`.
|
|
_LIST_SESSIONS_PAGE_SIZE = 50
|
|
|
|
|
|
def _flatten_history_text(value: Any) -> str:
|
|
"""Persisted content/reasoning (str, or list of ``{"text"}`` / ``{"type": "text", "content"}``
|
|
parts) -> one stripped string; whitespace-only collapses to ``""`` ("nothing to emit")."""
|
|
if isinstance(value, str):
|
|
return value.strip()
|
|
if isinstance(value, list):
|
|
parts: list[str] = []
|
|
for item in value:
|
|
if isinstance(item, dict):
|
|
text = item.get("text")
|
|
if isinstance(text, str):
|
|
parts.append(text)
|
|
elif item.get("type") == "text" and isinstance(item.get("content"), str):
|
|
parts.append(item["content"])
|
|
elif isinstance(item, str):
|
|
parts.append(item)
|
|
return "\n".join(part.strip() for part in parts if part and part.strip()).strip()
|
|
return ""
|
|
|
|
|
|
def _history_reasoning_text(message: dict[str, Any]) -> str:
|
|
"""First non-empty of ``reasoning_content`` and ``reasoning`` — both live keys, for
|
|
different transports (not old-vs-new)."""
|
|
for key in ("reasoning_content", "reasoning"):
|
|
text = _flatten_history_text(message.get(key))
|
|
if text:
|
|
return text
|
|
return ""
|
|
|
|
|
|
def _history_summary_meta(message: dict[str, Any], text: str) -> dict[str, Any] | None:
|
|
"""``_meta`` for a replayed compaction summary, else None.
|
|
|
|
Summaries persist as ordinary messages, standalone (either role) or merged into the first
|
|
preserved tail message. Two keys so clients can't hide real content: ``compactionSummary``
|
|
(whole chunk; safe to collapse) vs ``containsCompactionSummary`` (real content + summary).
|
|
Uses the in-process flag, falling back to content classification for DB-reloaded sessions."""
|
|
kind = ContextCompressor.classify_summary_content(text)
|
|
if kind is None and message.get(COMPRESSED_SUMMARY_METADATA_KEY):
|
|
# Flagged but unclassified (prefix drift): the flag only marks summaries -> standalone.
|
|
kind = "standalone"
|
|
if kind == "standalone":
|
|
return {"hermes": {"compactionSummary": True}}
|
|
if kind == "merged":
|
|
return {"hermes": {"containsCompactionSummary": True}}
|
|
return None
|
|
|
|
|
|
# role -> (chunk class, session_update tag) for history replay.
|
|
_HISTORY_CHUNK_TYPES = {
|
|
"user": (UserMessageChunk, "user_message_chunk"), "assistant": (AgentMessageChunk, "agent_message_chunk")
|
|
}
|
|
|
|
|
|
def _history_tool_call_name_args(tool_call: dict[str, Any]) -> tuple[str, dict[str, Any]]:
|
|
"""Extract function name/arguments from an OpenAI-style tool_call."""
|
|
function = tool_call.get("function") if isinstance(tool_call.get("function"), dict) else {}
|
|
name = str(function.get("name") or tool_call.get("name") or "unknown_tool")
|
|
raw_args = function.get("arguments") or tool_call.get("arguments") or tool_call.get("args") or {}
|
|
return name, coerce_tool_args(raw_args)
|
|
|
|
|
|
def _history_message_chunk(role: str, message: dict[str, Any]) -> UserMessageChunk | AgentMessageChunk | None:
|
|
text = _flatten_history_text(message.get("content"))
|
|
if not text:
|
|
return None
|
|
cls, session_update = _HISTORY_CHUNK_TYPES[role]
|
|
return cls(
|
|
session_update=session_update, content=TextContentBlock(type="text", text=text),
|
|
field_meta=_history_summary_meta(message, text),
|
|
)
|
|
|
|
|
|
def _history_replay_updates(history: list[dict[str, Any]]):
|
|
"""Yield ACP session updates that reconstruct a persisted transcript, in order: user/assistant
|
|
text (with compaction ``_meta``), assistant thoughts, and tool-call start/complete pairs
|
|
(``todo`` results also re-emit the plan)."""
|
|
active_tool_calls: dict[str, tuple[str, dict[str, Any]]] = {}
|
|
for message in history:
|
|
role = str(message.get("role") or "")
|
|
if role == "user":
|
|
if (chunk := _history_message_chunk(role, message)) is not None:
|
|
yield chunk
|
|
elif role == "assistant":
|
|
thought = _history_reasoning_text(message)
|
|
if thought:
|
|
yield acp.update_agent_thought_text(thought)
|
|
if (chunk := _history_message_chunk(role, message)) is not None:
|
|
yield chunk
|
|
tool_calls = message.get("tool_calls")
|
|
for tool_call in tool_calls if isinstance(tool_calls, list) else ():
|
|
if not isinstance(tool_call, dict):
|
|
continue
|
|
tool_call_id = str(
|
|
tool_call.get("id") or tool_call.get("call_id") or tool_call.get("tool_call_id") or ""
|
|
).strip()
|
|
if not tool_call_id:
|
|
continue
|
|
tool_name, args = _history_tool_call_name_args(tool_call)
|
|
active_tool_calls[tool_call_id] = (tool_name, args)
|
|
yield build_tool_start(tool_call_id, tool_name, args)
|
|
elif role == "tool":
|
|
tool_call_id = str(message.get("tool_call_id") or "").strip()
|
|
tool_name = str(message.get("tool_name") or "").strip()
|
|
function_args: dict[str, Any] | None = None
|
|
if tool_call_id in active_tool_calls:
|
|
tool_name, function_args = active_tool_calls.pop(tool_call_id)
|
|
if not tool_call_id or not tool_name:
|
|
continue
|
|
result = message.get("content")
|
|
result_text = result if isinstance(result, str) else None
|
|
yield build_tool_complete(tool_call_id, tool_name, result=result_text, function_args=function_args)
|
|
if tool_name == "todo":
|
|
plan_update = _build_plan_update_from_todo_result(result_text)
|
|
if plan_update is not None:
|
|
yield plan_update
|
|
|
|
|
|
def _mcp_server_config(server: McpServerStdio | McpServerHttp | McpServerSse) -> dict:
|
|
if isinstance(server, McpServerStdio):
|
|
return {"command": server.command, "args": list(server.args), "env": {i.name: i.value for i in server.env}}
|
|
return {"url": server.url, "headers": {i.name: i.value for i in server.headers}}
|
|
|
|
|
|
def _restore_env(key: str, value: str | None) -> None:
|
|
if value is None:
|
|
os.environ.pop(key, None)
|
|
else:
|
|
os.environ[key] = value
|
|
|
|
|
|
def _bind_guarded(stack: contextlib.ExitStack, label: str, setup: Callable[[], Callable[[], None]]) -> None:
|
|
"""Run ``setup`` (returns its teardown) and register the teardown; failures in either half only
|
|
log — the turn must still run without the binding."""
|
|
try:
|
|
teardown = setup()
|
|
except Exception:
|
|
logger.debug("Could not set ACP %s", label, exc_info=True)
|
|
return
|
|
|
|
def _teardown() -> None:
|
|
try:
|
|
teardown()
|
|
except Exception:
|
|
logger.debug("Could not restore ACP %s", label, exc_info=True)
|
|
|
|
stack.callback(_teardown)
|
|
|
|
|
|
def _attach_interrupted_prompt(interrupted_prompt: str, guidance: str) -> str:
|
|
return f"{interrupted_prompt}\n\nUser correction/guidance after interrupt: {guidance}"
|
|
|
|
|
|
def _take_interrupted_prompt(state: SessionState) -> tuple[bool, str]:
|
|
"""``(idle, interrupted_prompt)``; consumes the cancelled prompt only when the session is idle."""
|
|
with state.runtime_lock:
|
|
if state.is_running:
|
|
return False, ""
|
|
text, state.interrupted_prompt_text = state.interrupted_prompt_text, ""
|
|
return True, text
|
|
|
|
|
|
class ModelRejected(ValueError):
|
|
"""``switch_model`` refused the requested model (no provider can serve it)."""
|
|
|
|
|
|
@dataclass
|
|
class _TurnCallbacks:
|
|
"""Per-turn ACP streaming callbacks; all None when no client is connected."""
|
|
|
|
tool_progress_cb: Any = None
|
|
reasoning_cb: Any = None
|
|
step_cb: Any = None
|
|
stream_delta_cb: Any = None
|
|
approval_cb: Any = None
|
|
edit_approval_requester: Any = None
|
|
streamed: bool = False
|
|
tool_call_ids: Any = None
|
|
tool_call_meta: Any = None
|
|
|
|
|
|
class HermesACPAgent(SlashCommandsMixin, acp.Agent):
|
|
"""ACP Agent implementation wrapping Hermes AIAgent."""
|
|
|
|
_EDIT_APPROVAL_POLICY_CONFIG_ID = "edit_approval_policy"
|
|
_EDIT_APPROVAL_POLICY_DEFAULT = "ask"
|
|
_MODE_DEFAULT = "default"
|
|
# mode id -> (edit approval policy, display name, description)
|
|
_MODES: dict[str, tuple[str, str, str]] = {
|
|
"default": ("ask", "Default", "Ask before edits."),
|
|
"accept_edits": (
|
|
"workspace_session",
|
|
"Accept Edits",
|
|
"Auto-allow workspace and temp-dir edits; still asks for sensitive paths.",
|
|
),
|
|
"dont_ask": (
|
|
"session", "Don't Ask", "Auto-allow file edits for this session except sensitive paths."
|
|
),
|
|
}
|
|
_MODE_TO_EDIT_APPROVAL_POLICY = {mode: spec[0] for mode, spec in _MODES.items()}
|
|
_EDIT_APPROVAL_POLICY_TO_MODE = {spec[0]: mode for mode, spec in _MODES.items()}
|
|
|
|
def __init__(self, session_manager: SessionManager | None = None):
|
|
super().__init__()
|
|
self.session_manager = session_manager or SessionManager()
|
|
self._conn: Optional[acp.Client] = None
|
|
|
|
# ---- Connection lifecycle -----------------------------------------------
|
|
|
|
def on_connect(self, conn: acp.Client) -> None:
|
|
"""Store the client connection for sending session updates."""
|
|
self._conn = conn
|
|
logger.info("ACP client connected")
|
|
|
|
async def _send(self, session_id: str, update: Any, *, fail_msg: str, level: int = logging.WARNING) -> bool:
|
|
"""``session_update`` that logs instead of raising; False on failure."""
|
|
try:
|
|
await self._conn.session_update(session_id=session_id, update=update)
|
|
return True
|
|
except Exception:
|
|
logger.log(level, fail_msg, session_id, exc_info=True)
|
|
return False
|
|
|
|
def _schedule_soon(self, make_coro: Callable[[], Any]) -> None:
|
|
"""Run a notification coroutine right after the current response is queued."""
|
|
if not self._conn:
|
|
return
|
|
loop = asyncio.get_running_loop()
|
|
loop.call_soon(asyncio.create_task, make_coro())
|
|
|
|
def _session_modes(self, state: SessionState) -> SessionModeState:
|
|
"""Edit-approval policy as ACP modes. Zed renders ``config_options`` in the model
|
|
picker's slot; modes (as Claude/Codex use) coexist with the picker."""
|
|
current = str(getattr(state, "mode", "") or self._MODE_DEFAULT)
|
|
if current not in self._MODES:
|
|
current = self._MODE_DEFAULT
|
|
return SessionModeState(
|
|
current_mode_id=current,
|
|
available_modes=[SessionMode(id=m, name=n, description=d) for m, (_p, n, d) in self._MODES.items()],
|
|
)
|
|
|
|
def _edit_approval_policy_for_state(self, state: SessionState) -> tuple[str, str | None]:
|
|
mode = str(getattr(state, "mode", "") or self._MODE_DEFAULT)
|
|
policy = self._MODE_TO_EDIT_APPROVAL_POLICY.get(mode, self._EDIT_APPROVAL_POLICY_DEFAULT)
|
|
return policy, state.cwd
|
|
|
|
def _build_model_state(self, state: SessionState) -> SessionModelState | None:
|
|
"""Authenticated providers + models, from the shared Hermes inventory (same substrate
|
|
as ``hermes model``/TUI/dashboard) so the selector isn't just the current curated list."""
|
|
model = str(state.model or getattr(state.agent, "model", "") or "").strip()
|
|
provider = getattr(state.agent, "provider", None) or detect_provider() or "openrouter"
|
|
try:
|
|
picker = build_model_state(model, provider, str(getattr(state.agent, "base_url", "") or ""))
|
|
if picker is not None:
|
|
return picker
|
|
except Exception:
|
|
logger.debug("Could not build ACP model state", exc_info=True)
|
|
|
|
if not model:
|
|
return None
|
|
choice = encode_model_choice(provider, model)
|
|
return SessionModelState(available_models=[ModelInfo(model_id=choice, name=model)], current_model_id=choice)
|
|
|
|
def _switch_model(
|
|
self, state: SessionState, raw_model: str, *, keep_endpoint: bool = False
|
|
) -> tuple[str | None, str, str]:
|
|
"""Rebuild the session agent on a new model -> (old provider, new provider, model).
|
|
|
|
Resolution goes through ``hermes_cli.model_switch.switch_model`` seeded with the live
|
|
agent route — the same catalog/alias/credential validation as CLI/gateway/TUI ``/model``
|
|
— so ACP never hands the session a model no provider can serve. ``provider:model`` picker
|
|
ids become ``--provider``. ACP never persists. ``keep_endpoint`` carries base_url/api_mode
|
|
over when the provider is unchanged."""
|
|
from hermes_cli.config import get_compatible_custom_providers, load_config
|
|
from hermes_cli.model_switch import switch_model
|
|
from hermes_cli.models import parse_model_input
|
|
|
|
current_provider = getattr(state.agent, "provider", None)
|
|
explicit_provider, model_input = parse_model_input(raw_model, "")
|
|
cfg = load_config()
|
|
result = switch_model(
|
|
raw_input=model_input, explicit_provider=explicit_provider,
|
|
current_provider=current_provider or "openrouter", current_model=str(state.model or ""),
|
|
current_base_url=str(getattr(state.agent, "base_url", "") or ""),
|
|
current_api_key=str(getattr(state.agent, "api_key", "") or ""),
|
|
user_providers=cfg.get("providers") if isinstance(cfg.get("providers"), dict) else {},
|
|
custom_providers=get_compatible_custom_providers(cfg))
|
|
if not result.success:
|
|
raise ModelRejected(result.error_message or f"Cannot switch to {raw_model}")
|
|
target_provider, new_model = result.target_provider, result.new_model
|
|
endpoint: dict[str, Any] = {}
|
|
if keep_endpoint and not (current_provider and target_provider != current_provider):
|
|
endpoint = {
|
|
"base_url": getattr(state.agent, "base_url", None), "api_mode": getattr(state.agent, "api_mode", None)
|
|
}
|
|
# ACP-provided MCP servers live only on the running agent's toolsets (``_register_session_mcp_servers``);
|
|
# a rebuild that re-derived them from config would silently drop every session MCP tool (#42719).
|
|
agent = self.session_manager._make_agent(
|
|
session_id=state.session_id, cwd=state.cwd, model=new_model,
|
|
requested_provider=target_provider, **endpoint,
|
|
enabled_toolsets=getattr(state.agent, "enabled_toolsets", None),
|
|
disabled_toolsets=getattr(state.agent, "disabled_toolsets", None),
|
|
)
|
|
# Assign only after the rebuild succeeded so a failed switch leaves the session on its
|
|
# working model instead of a model/agent mismatch that persists via save_session.
|
|
state.agent, state.model = agent, new_model
|
|
self.session_manager.save_session(state.session_id)
|
|
return current_provider, target_provider, new_model
|
|
|
|
@staticmethod
|
|
def _build_usage_update(state: SessionState) -> UsageUpdate | None:
|
|
"""``usage_update`` for Zed's context indicator: ``size`` = context window, ``used`` =
|
|
estimated request pressure (system prompt + history + tool schemas)."""
|
|
compressor = getattr(state.agent, "context_compressor", None)
|
|
size = int(getattr(compressor, "context_length", 0) or 0)
|
|
if size <= 0:
|
|
return None
|
|
try:
|
|
used = _estimate_tokens(state.history, state.agent)
|
|
except Exception:
|
|
logger.debug("Could not estimate ACP native context usage", exc_info=True)
|
|
used = int(getattr(compressor, "last_prompt_tokens", 0) or 0)
|
|
return UsageUpdate(session_update="usage_update", size=max(size, 0), used=max(used, 0))
|
|
|
|
async def _send_usage_update(self, state: SessionState) -> None:
|
|
if self._conn and (update := self._build_usage_update(state)) is not None:
|
|
await self._send(state.session_id, update, fail_msg="Failed to send ACP usage update for session %s")
|
|
|
|
def _provenance_meta(
|
|
self, acp_session_id: str, current_hermes_session_id: str, previous_hermes_session_id: Optional[str] = None
|
|
) -> Optional[dict]:
|
|
"""Best-effort ``_meta.hermes.sessionProvenance`` for an ACP session."""
|
|
try:
|
|
return session_provenance_meta(
|
|
self.session_manager._get_db(), acp_session_id, current_hermes_session_id,
|
|
previous_hermes_session_id=previous_hermes_session_id,
|
|
)
|
|
except Exception:
|
|
logger.debug("Could not build ACP session provenance for %s", acp_session_id, exc_info=True)
|
|
return None
|
|
|
|
async def _send_session_info_update(
|
|
self, session_id: str, *,
|
|
current_hermes_session_id: Optional[str] = None, previous_hermes_session_id: Optional[str] = None,
|
|
) -> None:
|
|
"""Session metadata update; pass ``previous_hermes_session_id`` when the internal head
|
|
rotated (compression split) so provenance flags the reason."""
|
|
if not self._conn:
|
|
return
|
|
try:
|
|
row = self.session_manager._get_db().get_session(session_id)
|
|
except Exception:
|
|
logger.debug("Could not read ACP session info for %s", session_id, exc_info=True)
|
|
return
|
|
if not row:
|
|
return
|
|
title = row.get("title")
|
|
# `sessions` has no `updated_at`; "now" is right since this fires when the title changed.
|
|
update = SessionInfoUpdate(
|
|
session_update="session_info_update",
|
|
title=title if isinstance(title, str) and title.strip() else None,
|
|
updated_at=datetime.now(timezone.utc).isoformat(),
|
|
field_meta=self._provenance_meta(
|
|
session_id, current_hermes_session_id or session_id, previous_hermes_session_id
|
|
),
|
|
)
|
|
await self._send(
|
|
session_id, update, fail_msg="Could not send ACP session info update for %s", level=logging.DEBUG
|
|
)
|
|
|
|
async def _register_session_mcp_servers(
|
|
self, state: SessionState, mcp_servers: list[McpServerStdio | McpServerHttp | McpServerSse] | None
|
|
) -> None:
|
|
"""Register ACP-provided MCP servers and refresh the agent tool surface."""
|
|
if not mcp_servers:
|
|
return
|
|
try:
|
|
from agent.runtime_cwd import set_session_cwd
|
|
from tools.mcp_tool_discovery import register_mcp_servers
|
|
|
|
configs = {s.name: _mcp_server_config(s) for s in mcp_servers}
|
|
|
|
def _register_pinned() -> None:
|
|
# new_session/load_session run outside the per-turn cwd pin; the session's logical cwd is
|
|
# the default stdio child cwd (tools/mcp_tool_transport.py::_run_stdio), so pin it here.
|
|
set_session_cwd(state.cwd)
|
|
register_mcp_servers(configs)
|
|
|
|
await asyncio.to_thread(_register_pinned) # to_thread already runs in a copied context
|
|
except Exception:
|
|
logger.warning("Session %s: failed to register ACP MCP servers", state.session_id, exc_info=True)
|
|
return
|
|
try:
|
|
from model_tools import get_tool_definitions
|
|
from agent.memory_manager import inject_memory_provider_tools
|
|
|
|
agent = state.agent
|
|
agent.enabled_toolsets = _expand_acp_enabled_toolsets(
|
|
getattr(agent, "enabled_toolsets", None) or ["hermes-acp"],
|
|
mcp_server_names=[s.name for s in mcp_servers],
|
|
)
|
|
agent.tools = get_tool_definitions(
|
|
enabled_toolsets=agent.enabled_toolsets,
|
|
disabled_toolsets=getattr(agent, "disabled_toolsets", None), quiet_mode=True,
|
|
)
|
|
agent.valid_tool_names = {tool["function"]["name"] for tool in agent.tools or []}
|
|
inject_memory_provider_tools(agent)
|
|
if callable(invalidate := getattr(agent, "_invalidate_system_prompt", None)):
|
|
invalidate()
|
|
logger.info(
|
|
"Session %s: refreshed tool surface after ACP MCP registration (%d tools)",
|
|
state.session_id, len(agent.tools or []),
|
|
)
|
|
except Exception:
|
|
logger.warning(
|
|
"Session %s: failed to refresh tool surface after ACP MCP registration", state.session_id, exc_info=True,
|
|
)
|
|
|
|
def _schedule_mcp_late_refresh(self, state: SessionState) -> None:
|
|
"""Refresh the tool snapshot when background MCP discovery lands after agent build
|
|
(``_make_agent`` only joins ~1.5s). Waits up to 30s off the critical path, then rebuilds
|
|
via ``refresh_agent_mcp_tools`` (same as ``/reload-mcp``).
|
|
|
|
Cache safety: only pre-first-turn (nothing cached yet); afterwards the snapshot stays
|
|
frozen and late servers land via the between-turns prologue refresh
|
|
(``agent/turn_context.py``). No-op if discovery finished, join timed out, registry
|
|
unchanged, or session closed."""
|
|
try:
|
|
from hermes_cli.mcp_startup import mcp_discovery_in_flight
|
|
except Exception:
|
|
return
|
|
if not mcp_discovery_in_flight():
|
|
return
|
|
agent, session_id = state.agent, state.session_id
|
|
|
|
def _wait_then_refresh() -> None:
|
|
try:
|
|
from hermes_cli.mcp_startup import join_mcp_discovery
|
|
|
|
if not join_mcp_discovery(timeout=30.0):
|
|
return
|
|
|
|
# In-memory only: ``get_session()`` would restore from DB and build a new AIAgent.
|
|
with self.session_manager._lock:
|
|
current = self.session_manager._sessions.get(session_id)
|
|
if current is None or current.agent is not agent:
|
|
return
|
|
|
|
# ``prompt()`` flips ``is_running`` under ``runtime_lock`` before dispatching, so
|
|
# holding it here closes the window where a refresh would swap ``tools=`` mid-turn.
|
|
with current.runtime_lock:
|
|
if current.is_running:
|
|
return
|
|
if any(int(getattr(agent, k, 0) or 0) > 0 for k in ("_user_turn_count", "_api_call_count")):
|
|
return
|
|
|
|
from tools.mcp_tool_agent import refresh_agent_mcp_tools
|
|
|
|
added = refresh_agent_mcp_tools(agent, quiet_mode=True)
|
|
if added:
|
|
logger.info(
|
|
"Session %s: late MCP refresh added %d tools: %s",
|
|
session_id, len(added), ", ".join(sorted(added)),
|
|
)
|
|
except Exception:
|
|
logger.debug("Session %s: late MCP refresh failed", session_id, exc_info=True)
|
|
|
|
threading.Thread(target=_wait_then_refresh, name=f"acp-mcp-late-refresh-{session_id}", daemon=True).start()
|
|
|
|
# ---- ACP lifecycle ------------------------------------------------------
|
|
|
|
async def initialize(
|
|
self, protocol_version: int | None = None, client_capabilities: ClientCapabilities | None = None,
|
|
client_info: Implementation | None = None, **kwargs: Any,
|
|
) -> InitializeResponse:
|
|
from hermes_cli.version_info import get_version_info
|
|
|
|
auth_methods = build_auth_methods()
|
|
logger.info(
|
|
"Initialize from %s (protocol v%s)", client_info.name if client_info else "unknown",
|
|
protocol_version if isinstance(protocol_version, int) else acp.PROTOCOL_VERSION,
|
|
)
|
|
|
|
return InitializeResponse(
|
|
protocol_version=acp.PROTOCOL_VERSION,
|
|
agent_info=Implementation(name="hermes-agent", version=get_version_info().base_version),
|
|
agent_capabilities=AgentCapabilities(
|
|
load_session=True,
|
|
prompt_capabilities=PromptCapabilities(image=True),
|
|
session_capabilities=SessionCapabilities(
|
|
fork=SessionForkCapabilities(), list=SessionListCapabilities(), resume=SessionResumeCapabilities(),
|
|
),
|
|
),
|
|
auth_methods=auth_methods,
|
|
)
|
|
|
|
async def authenticate(self, method_id: str, **kwargs: Any) -> AuthenticateResponse | None:
|
|
# Only acknowledge the method_id advertised in initialize().
|
|
if not isinstance(method_id, str):
|
|
return None
|
|
normalized_method = method_id.strip().lower()
|
|
provider = detect_provider()
|
|
|
|
if normalized_method == TERMINAL_SETUP_AUTH_METHOD_ID:
|
|
# Terminal auth runs setup out-of-band; succeed only once credentials exist.
|
|
return AuthenticateResponse() if provider else None
|
|
|
|
if not provider or normalized_method != provider:
|
|
return None
|
|
return AuthenticateResponse()
|
|
|
|
# ---- Session management -------------------------------------------------
|
|
|
|
async def _replay_session_history(self, state: SessionState) -> None:
|
|
"""Replay history as user/assistant/thought chunks plus reconstructed tool-call
|
|
start/complete events so the editor shows the transcript, not a clean thread."""
|
|
if not self._conn or not state.history:
|
|
return
|
|
for update in _history_replay_updates(state.history):
|
|
if not await self._send(state.session_id, update, fail_msg="Failed to replay ACP history for session %s"):
|
|
return
|
|
|
|
async def _session_response_fields(self, state: SessionState, replay_verb: str | None = None) -> dict[str, Any]:
|
|
"""``models``/``modes``/``field_meta`` for session responses, after an optional history replay;
|
|
schedules command advertisement + usage refresh.
|
|
|
|
Per ACP spec, load/resume must stream history via ``session/update`` BEFORE responding
|
|
(Codex/Claude Code/OpenCode/Zed rely on this; deferring via ``call_soon`` broke them).
|
|
Best-effort: a corrupt message must not turn the load into an error."""
|
|
if replay_verb:
|
|
try:
|
|
# Per ACP spec, `session/load` must stream the prior conversation back to the client via
|
|
# `session/update` notifications BEFORE responding, so the client receives the full
|
|
# transcript within the load request's lifetime. Awaiting the replay here matches Codex /
|
|
# Claude Code / OpenCode / Pi and the Zed client (which registers the session-update routing
|
|
# entry before awaiting the loadSession RPC specifically so in-call history replay updates
|
|
# can find the thread). Deferring this via `loop.call_soon` (as we did briefly in May 2026)
|
|
# broke every spec-compliant ACP client that measures notifications synchronously against
|
|
# the load response — see #12285 follow-up.
|
|
await self._replay_session_history(state)
|
|
except Exception:
|
|
logger.warning(
|
|
f"ACP history replay raised during session/{replay_verb} for %s — "
|
|
f"{replay_verb} will still succeed, partial transcript may be missing",
|
|
state.session_id, exc_info=True,
|
|
)
|
|
self._schedule_available_commands_update(state.session_id)
|
|
self._schedule_soon(lambda: self._send_usage_update(state))
|
|
return {
|
|
"models": self._build_model_state(state),
|
|
"modes": self._session_modes(state),
|
|
"field_meta": self._provenance_meta(state.session_id, getattr(state.agent, "session_id", state.session_id)),
|
|
}
|
|
|
|
async def _attach_session_mcp(self, state: SessionState, mcp_servers: list | None, log: str, *log_args) -> None:
|
|
await self._register_session_mcp_servers(state, mcp_servers)
|
|
self._schedule_mcp_late_refresh(state)
|
|
logger.info(log, *log_args)
|
|
|
|
async def new_session(self, cwd: str, mcp_servers: list | None = None, **kwargs: Any) -> NewSessionResponse:
|
|
# Agent construction (config, memory-provider import, SessionDB) is slow and fully
|
|
# blocking; inline it froze the loop serving every JSON-RPC request (#58083).
|
|
state = await asyncio.to_thread(self.session_manager.create_session, cwd=cwd)
|
|
await self._attach_session_mcp(state, mcp_servers, "New session %s (cwd=%s)", state.session_id, cwd)
|
|
return NewSessionResponse(session_id=state.session_id, **await self._session_response_fields(state))
|
|
|
|
async def load_session(
|
|
self, cwd: str, session_id: str, mcp_servers: list | None = None, **kwargs: Any
|
|
) -> LoadSessionResponse | None:
|
|
state = await asyncio.to_thread(self.session_manager.update_cwd, session_id, cwd)
|
|
if state is None:
|
|
logger.warning("load_session: session %s not found", session_id)
|
|
return None
|
|
await self._attach_session_mcp(state, mcp_servers, "Loaded session %s", session_id)
|
|
return LoadSessionResponse(**await self._session_response_fields(state, "load"))
|
|
|
|
async def resume_session(
|
|
self, cwd: str, session_id: str, mcp_servers: list | None = None, **kwargs: Any
|
|
) -> ResumeSessionResponse:
|
|
state = await asyncio.to_thread(self.session_manager.update_cwd, session_id, cwd)
|
|
if state is None:
|
|
logger.warning("resume_session: session %s not found, creating new", session_id)
|
|
state = await asyncio.to_thread(self.session_manager.create_session, cwd=cwd)
|
|
await self._attach_session_mcp(state, mcp_servers, "Resumed session %s", state.session_id)
|
|
return ResumeSessionResponse(**await self._session_response_fields(state, "resume"))
|
|
|
|
async def cancel(self, session_id: str, **kwargs: Any) -> None:
|
|
# get_session restores a not-in-memory id from the DB (full AIAgent build) and waits
|
|
# on the restore lock — off the loop, like new/load/resume/fork (#58083).
|
|
state = await asyncio.to_thread(self.session_manager.get_session, session_id)
|
|
if not (state and state.cancel_event):
|
|
return
|
|
with state.runtime_lock:
|
|
if state.is_running and state.current_prompt_text:
|
|
state.interrupted_prompt_text = state.current_prompt_text
|
|
# Cancel + hard-stop under the lock so no other prompt mistakes this turn for
|
|
# redirectable work.
|
|
state.cancel_event.set()
|
|
try:
|
|
if state.agent:
|
|
request_hard_interrupt(state.agent)
|
|
# Background delegations are detached from the turn's fan-out; they end with the cancel.
|
|
from tools.async_delegation import interrupt_for_session
|
|
interrupt_for_session(parent_session_id=str(getattr(state.agent, "session_id", "") or ""),
|
|
reason="acp_cancel")
|
|
except Exception:
|
|
logger.debug("Failed to interrupt ACP session %s", session_id, exc_info=True)
|
|
logger.info("Cancelled session %s", session_id)
|
|
|
|
async def fork_session(
|
|
self, cwd: str, session_id: str, mcp_servers: list | None = None, **kwargs: Any
|
|
) -> ForkSessionResponse:
|
|
state = await asyncio.to_thread(self.session_manager.fork_session, session_id, cwd=cwd)
|
|
if state is None:
|
|
logger.info("Forked session %s -> %s", session_id, "")
|
|
return ForkSessionResponse(session_id="")
|
|
await self._register_session_mcp_servers(state, mcp_servers)
|
|
logger.info("Forked session %s -> %s", session_id, state.session_id)
|
|
self._schedule_available_commands_update(state.session_id)
|
|
return ForkSessionResponse(
|
|
session_id=state.session_id, models=self._build_model_state(state), modes=self._session_modes(state)
|
|
)
|
|
|
|
async def list_sessions(
|
|
self, cursor: str | None = None, cwd: str | None = None, **kwargs: Any
|
|
) -> ListSessionsResponse:
|
|
"""``cursor`` is a ``session_id`` returned as ``next_cursor``; results resume after it
|
|
(unknown cursor -> empty page, never the full list). Pages cap at the fixed size."""
|
|
infos = self.session_manager.list_sessions(cwd=cwd)
|
|
|
|
if cursor:
|
|
for idx, s in enumerate(infos):
|
|
if s["session_id"] == cursor:
|
|
infos = infos[idx + 1:]
|
|
break
|
|
else:
|
|
infos = []
|
|
|
|
has_more = len(infos) > _LIST_SESSIONS_PAGE_SIZE
|
|
sessions = [
|
|
SessionInfo(
|
|
session_id=s["session_id"], cwd=s["cwd"], title=s.get("title"),
|
|
updated_at=None if s.get("updated_at") is None else str(s["updated_at"]),
|
|
)
|
|
for s in infos[:_LIST_SESSIONS_PAGE_SIZE]
|
|
]
|
|
next_cursor = sessions[-1].session_id if has_more and sessions else None
|
|
return ListSessionsResponse(sessions=sessions, next_cursor=next_cursor)
|
|
|
|
# ---- Prompt (core) ------------------------------------------------------
|
|
|
|
def _rewrite_prompt_for_interrupt(
|
|
self, state: SessionState, user_text: str, user_content: Any, text_only: bool
|
|
) -> tuple[str, Any]:
|
|
"""Idle ``/steer`` has nothing to inject into (gateway parity): if a prompt was just
|
|
cancelled, replay it with the steer text as explicit correction; otherwise run the steer
|
|
payload as a plain prompt rather than silently queueing it as if ``/queue`` was typed.
|
|
Plain text after a cancel likewise keeps the cancelled request attached ("stop and
|
|
send" clients) so deictic follow-ups have a target."""
|
|
if not (text_only and isinstance(user_content, str)):
|
|
return user_text, user_content
|
|
|
|
if user_text.startswith("/steer"):
|
|
split = user_text.split(maxsplit=1)
|
|
steer_text = split[1].strip() if len(split) > 1 else ""
|
|
if not steer_text:
|
|
return user_text, user_content
|
|
idle, interrupted_prompt = _take_interrupted_prompt(state)
|
|
if interrupted_prompt:
|
|
return (_attach_interrupted_prompt(interrupted_prompt, steer_text),) * 2
|
|
return (steer_text, steer_text) if idle else (user_text, user_content)
|
|
if not user_text.startswith("/") and (interrupted_prompt := _take_interrupted_prompt(state)[1]):
|
|
return (_attach_interrupted_prompt(interrupted_prompt, user_text),) * 2
|
|
return user_text, user_content
|
|
|
|
def _claim_turn_or_queue(
|
|
self, state: SessionState, session_id: str, user_text: str, user_content: Any, text_only: bool
|
|
) -> str | None:
|
|
"""Mark the session running; if a turn is active, redirect it (text-only, supported
|
|
runtime) or queue it. Returns the client message when absorbed, else None."""
|
|
with state.runtime_lock:
|
|
if not state.is_running and not state.command_op:
|
|
state.is_running = True
|
|
state.current_prompt_text = user_text or "[Image attachment]"
|
|
return None
|
|
# Redirect steers a live turn; a state-mutating command (command_op) has none.
|
|
if state.is_running and text_only and isinstance(user_content, str) and hasattr(state.agent, "redirect") and (
|
|
getattr(state.agent, "_supports_active_turn_redirect", False) is True
|
|
):
|
|
try:
|
|
if state.agent.redirect(user_content):
|
|
return "Redirected the active turn with your correction."
|
|
except Exception:
|
|
logger.debug("ACP active-turn redirect failed for %s", session_id, exc_info=True)
|
|
state.queued_prompts.append(user_text or "[Image attachment]")
|
|
return f"Queued for the next turn. ({len(state.queued_prompts)} queued)"
|
|
|
|
def _run_agent_turn(
|
|
self, *, state: SessionState, session_id: str, user_text: str, user_content: Any, conn: Any,
|
|
loop: asyncio.AbstractEventLoop, approval_cb: Any, edit_approval_requester: Any,
|
|
) -> dict:
|
|
"""Executor-thread body of one turn, run inside ``contextvars.copy_context()`` so
|
|
ContextVar writes are isolated from concurrent sessions.
|
|
|
|
Approval routing is thread-local, so it MUST be bound here, not on the loop thread.
|
|
Interactive routing is a ``tools.approval`` contextvar, not ``HERMES_INTERACTIVE`` in
|
|
os.environ, so concurrent workers can't race a global flag onto the non-interactive
|
|
auto-approve path (GHSA-96vc-wcxf-jjff)."""
|
|
agent = state.agent
|
|
with contextlib.ExitStack() as stack:
|
|
# HERMES_SESSION_KEY scopes per-session caches (interactive sudo password) to this
|
|
# session, not the reused thread. ``cwd`` pins what the system prompt reports as the
|
|
# working directory — otherwise it advertises the Hermes workspace while tools are
|
|
# rooted at the client's project and edits land outside it. ``cron_session=""`` masks
|
|
# any leaked process-global HERMES_CRON_SESSION.
|
|
def _session_context() -> Callable[[], None]:
|
|
from gateway.session_context import clear_session_vars, set_session_vars
|
|
|
|
tokens = set_session_vars(
|
|
session_key=session_id, session_id=session_id, cwd=state.cwd, cron_session="",
|
|
)
|
|
return lambda: clear_session_vars(tokens)
|
|
|
|
def _approval() -> Callable[[], None]:
|
|
from tools import terminal_tool
|
|
|
|
previous = terminal_tool._get_approval_callback()
|
|
terminal_tool.set_approval_callback(approval_cb)
|
|
return lambda: terminal_tool.set_approval_callback(previous)
|
|
|
|
def _edit_approval() -> Callable[[], None]:
|
|
from acp_adapter.edit_approval import reset_edit_approval_requester, set_edit_approval_requester
|
|
|
|
token = set_edit_approval_requester(edit_approval_requester)
|
|
return lambda: reset_edit_approval_requester(token)
|
|
|
|
_bind_guarded(stack, "session context", _session_context)
|
|
if approval_cb:
|
|
_bind_guarded(stack, "approval callback", _approval)
|
|
if edit_approval_requester:
|
|
_bind_guarded(stack, "edit approval requester", _edit_approval)
|
|
stack.callback(reset_hermes_interactive_context, set_hermes_interactive_context(True))
|
|
# Tools tag side-effects with the ACP session (``kanban_create``); save/restore it.
|
|
stack.callback(_restore_env, "HERMES_SESSION_ID", os.environ.get("HERMES_SESSION_ID"))
|
|
os.environ["HERMES_SESSION_ID"] = session_id
|
|
|
|
# Auto-titling fires in the turn prologue; push the title now as a session-info update.
|
|
def _notify_title_update(_title: str, _source: str) -> None:
|
|
if conn:
|
|
loop.call_soon_threadsafe(asyncio.create_task, self._send_session_info_update(session_id))
|
|
|
|
agent._on_session_title = _notify_title_update
|
|
try:
|
|
return agent.run_conversation(
|
|
user_message=user_content, conversation_history=state.history, task_id=session_id,
|
|
persist_user_message=user_text or "[Image attachment]",
|
|
)
|
|
except Exception as e:
|
|
logger.exception("Agent error in session %s", session_id)
|
|
return {"final_response": f"Error: {e}", "messages": state.history}
|
|
|
|
async def prompt(self, prompt: list[PromptBlock], session_id: str, **kwargs: Any) -> PromptResponse:
|
|
"""Run Hermes on the user's prompt and stream events back to the editor."""
|
|
state = await asyncio.to_thread(self.session_manager.get_session, session_id)
|
|
if state is None:
|
|
logger.error("prompt: session %s not found", session_id)
|
|
return PromptResponse(stop_reason="refusal")
|
|
|
|
user_text = _extract_text(prompt).strip()
|
|
user_content = _content_blocks_to_openai_user_content(prompt)
|
|
text_only_prompt = all(isinstance(block, TextContentBlock) for block in prompt)
|
|
if not user_text and not (isinstance(user_content, list) and user_content):
|
|
return PromptResponse(stop_reason="end_turn")
|
|
|
|
user_text, user_content = self._rewrite_prompt_for_interrupt(state, user_text, user_content, text_only_prompt)
|
|
|
|
# Slash commands are text-only; a prompt with media goes to the agent even if it starts with "/".
|
|
if text_only_prompt and isinstance(user_content, str) and user_text.startswith("/"):
|
|
# Off the loop: /model validates through switch_model (network I/O) and /compress
|
|
# calls the LLM; handlers are sync and hold no loop-bound state.
|
|
response_text = await asyncio.to_thread(self._handle_slash_command, user_text, state)
|
|
if response_text is not None:
|
|
if self._conn:
|
|
await self._conn.session_update(session_id, acp.update_agent_message_text(response_text))
|
|
await self._send_usage_update(state)
|
|
# A mutating command held command_op; prompts that arrived mid-op are queued.
|
|
await self._drain_queued_prompts(state, session_id, self._conn)
|
|
return PromptResponse(stop_reason="end_turn")
|
|
|
|
absorbed = self._claim_turn_or_queue(state, session_id, user_text, user_content, text_only_prompt)
|
|
if absorbed is not None:
|
|
if self._conn:
|
|
await self._conn.session_update(session_id, acp.update_agent_message_text(absorbed))
|
|
return PromptResponse(stop_reason="end_turn")
|
|
|
|
logger.info("Prompt on session %s: %s", session_id, user_text[:100])
|
|
conn, loop = self._conn, asyncio.get_running_loop()
|
|
if state.cancel_event:
|
|
state.cancel_event.clear()
|
|
cbs = self._wire_turn_callbacks(state, session_id, conn, loop)
|
|
|
|
def _run_agent() -> dict:
|
|
try:
|
|
return self._run_agent_turn(
|
|
state=state, session_id=session_id, user_text=user_text, user_content=user_content, conn=conn,
|
|
loop=loop, approval_cb=cbs.approval_cb, edit_approval_requester=cbs.edit_approval_requester,
|
|
)
|
|
finally:
|
|
# Still on the executor thread: ``_send_update`` blocks on the loop, so flushing
|
|
# here (not after ``run_in_executor``) lands the update before the response.
|
|
self._flush_turn_tool_calls(cbs, session_id, conn, loop)
|
|
|
|
try:
|
|
# ACP `session_id` is the stable handle; agent.session_id is the internal head that
|
|
# compression may rotate — snapshot it to detect rotation after the turn.
|
|
pre_turn_hermes_id = getattr(state.agent, "session_id", None)
|
|
# Fresh context copy: concurrent sessions on the shared executor must not share ContextVars.
|
|
ctx = contextvars.copy_context()
|
|
result = await loop.run_in_executor(_executor, ctx.run, _run_agent)
|
|
except Exception:
|
|
logger.exception("Executor error for session %s", session_id)
|
|
with state.runtime_lock:
|
|
state.is_running = False
|
|
state.current_prompt_text = ""
|
|
return PromptResponse(stop_reason="end_turn")
|
|
|
|
return await self._finish_turn(state, session_id, conn, result, pre_turn_hermes_id, cbs.streamed)
|
|
|
|
def _flush_turn_tool_calls(
|
|
self, cbs: _TurnCallbacks, session_id: str, conn: Any, loop: asyncio.AbstractEventLoop
|
|
) -> None:
|
|
"""Close any tool call the turn left open, so no bubble spins after the turn ends."""
|
|
if not conn or cbs.tool_call_ids is None:
|
|
return
|
|
try:
|
|
flush_open_tool_calls(conn, session_id, loop, cbs.tool_call_ids, cbs.tool_call_meta or {})
|
|
except Exception:
|
|
logger.debug("Could not flush open ACP tool calls for %s", session_id, exc_info=True)
|
|
|
|
def _wire_turn_callbacks(
|
|
self, state: SessionState, session_id: str, conn: Any, loop: asyncio.AbstractEventLoop
|
|
) -> _TurnCallbacks:
|
|
"""Install the ACP streaming callbacks on the session agent for one turn."""
|
|
cbs = _TurnCallbacks()
|
|
if conn:
|
|
tool_call_ids: dict[str, Deque[str]] = defaultdict(deque)
|
|
tool_call_meta: dict[str, dict[str, Any]] = {}
|
|
cbs.tool_call_ids, cbs.tool_call_meta = tool_call_ids, tool_call_meta
|
|
# Shared with the step callback so a runtime that projects
|
|
# ``tool.completed`` closes each call once, not twice.
|
|
turn_state: dict[str, Any] = {}
|
|
policy_getter = lambda: self._edit_approval_policy_for_state(state) # noqa: E731
|
|
cbs.tool_progress_cb = make_tool_progress_cb(
|
|
conn, session_id, loop, tool_call_ids, tool_call_meta, edit_approval_policy_getter=policy_getter,
|
|
turn_state=turn_state,
|
|
)
|
|
# Per-session allocator: a new turn must never reuse a previous turn's
|
|
# assistant messageId (ACP clients replace the bubble with that id).
|
|
if state.message_ids is None:
|
|
state.message_ids = AssistantMessageIdAllocator()
|
|
state.message_ids.close() # new turn -> next chunk opens a fresh id
|
|
cbs.reasoning_cb = make_thinking_cb(conn, session_id, loop, state.message_ids)
|
|
cbs.step_cb = make_step_cb(conn, session_id, loop, tool_call_ids, tool_call_meta, turn_state)
|
|
message_cb = make_message_cb(conn, session_id, loop, state.message_ids)
|
|
|
|
def stream_delta_cb(text: str) -> None:
|
|
cbs.streamed = cbs.streamed or bool(text)
|
|
message_cb(text)
|
|
|
|
cbs.stream_delta_cb = stream_delta_cb
|
|
# Closes the synthetic permission-request bubble once the user has answered.
|
|
send_update = lambda update: _send_update(conn, session_id, loop, update) # noqa: E731
|
|
cbs.approval_cb = make_approval_callback(conn.request_permission, loop, session_id, send_update=send_update)
|
|
try:
|
|
from acp_adapter.edit_approval import make_acp_edit_approval_requester
|
|
|
|
cbs.edit_approval_requester = make_acp_edit_approval_requester(
|
|
conn.request_permission, loop, session_id, auto_approve_getter=policy_getter,
|
|
send_update=send_update,
|
|
)
|
|
except Exception:
|
|
logger.debug("Could not create ACP edit approval requester", exc_info=True)
|
|
|
|
agent = state.agent
|
|
agent.tool_progress_callback = cbs.tool_progress_cb
|
|
# Thought panes get provider reasoning only — no local status updates, no fake accordion.
|
|
agent.thinking_callback = None
|
|
agent.reasoning_callback, agent.step_callback = cbs.reasoning_cb, cbs.step_cb
|
|
agent.stream_delta_callback = cbs.stream_delta_cb
|
|
return cbs
|
|
|
|
async def _finish_turn(
|
|
self, state: SessionState, session_id: str, conn: Any, result: dict, pre_turn_hermes_id: Any,
|
|
streamed_message: bool,
|
|
) -> PromptResponse:
|
|
"""Persist, emit provenance/final text, drain queued prompts, report usage."""
|
|
try:
|
|
# Key presence, not truthiness: ``messages=[]`` is a legitimate cleared transcript (#10844);
|
|
# only a result without the key leaves the history untouched.
|
|
if "messages" in result and isinstance(result["messages"], list):
|
|
state.history = result["messages"]
|
|
self.session_manager.save_session(session_id)
|
|
|
|
# Head rotated (compression split): emit provenance so clients can render the boundary.
|
|
post_turn_hermes_id = getattr(state.agent, "session_id", None)
|
|
if conn and post_turn_hermes_id and pre_turn_hermes_id and post_turn_hermes_id != pre_turn_hermes_id:
|
|
try:
|
|
await self._send_session_info_update(
|
|
session_id, current_hermes_session_id=post_turn_hermes_id,
|
|
previous_hermes_session_id=pre_turn_hermes_id,
|
|
)
|
|
except Exception:
|
|
logger.debug("Could not emit ACP provenance update after rotation for %s", session_id, exc_info=True)
|
|
|
|
final_response = result.get("final_response") or "" # None on an interrupted turn
|
|
cancelled = bool(state.cancel_event and state.cancel_event.is_set())
|
|
# The local "waiting for model" interrupt status is metadata, not prose; stop_reason carries it.
|
|
from agent.conversation_loop import INTERRUPT_WAITING_FOR_MODEL_PREFIX
|
|
|
|
interrupted = bool(result.get("interrupted")) or cancelled
|
|
suppress = interrupted and final_response.startswith(INTERRUPT_WAITING_FOR_MODEL_PREFIX)
|
|
# Send the final text unless already streamed — or if a plugin hook transformed it after.
|
|
if final_response and conn and not suppress and (not streamed_message or result.get("response_transformed")):
|
|
update = acp.update_agent_message_text(final_response)
|
|
if state.message_ids is not None:
|
|
# A plugin-rewritten reply replaces the streamed bubble (same id); an
|
|
# unstreamed final response opens its own.
|
|
if streamed_message and result.get("response_transformed"):
|
|
update.message_id = state.message_ids.last() or state.message_ids.current()
|
|
else:
|
|
update.message_id = state.message_ids.current()
|
|
state.message_ids.close()
|
|
await conn.session_update(session_id, update)
|
|
|
|
finally:
|
|
# Go idle before draining so recursive prompt() calls can acquire the session.
|
|
with state.runtime_lock:
|
|
state.is_running = False
|
|
state.current_prompt_text = ""
|
|
await self._drain_queued_prompts(state, session_id, conn)
|
|
|
|
usage = None
|
|
if any(result.get(k) is not None for k in ("prompt_tokens", "completion_tokens", "total_tokens")):
|
|
usage = Usage(
|
|
input_tokens=result.get("prompt_tokens", 0), output_tokens=result.get("completion_tokens", 0),
|
|
total_tokens=result.get("total_tokens", 0), thought_tokens=result.get("reasoning_tokens"),
|
|
cached_read_tokens=result.get("cache_read_tokens"),
|
|
)
|
|
await self._send_usage_update(state)
|
|
return PromptResponse(stop_reason="cancelled" if cancelled else "end_turn", usage=usage)
|
|
|
|
async def _drain_queued_prompts(self, state: SessionState, session_id: str, conn: Any) -> None:
|
|
"""Run queued prompts while the session is idle. Reached from ``_finish_turn`` and
|
|
from the slash path after a state-mutating command releases ``command_op``."""
|
|
while True:
|
|
with state.runtime_lock:
|
|
if state.is_running or state.command_op or not state.queued_prompts:
|
|
return
|
|
next_prompt = state.queued_prompts.pop(0)
|
|
if conn:
|
|
await conn.session_update(session_id, acp.update_user_message_text(next_prompt))
|
|
await self.prompt(prompt=[TextContentBlock(type="text", text=next_prompt)], session_id=session_id)
|
|
|
|
# ---- Session settings (ACP protocol methods) -----------------------------
|
|
|
|
async def set_session_model(self, model_id: str, session_id: str, **kwargs: Any) -> SetSessionModelResponse | None:
|
|
"""Switch the model for a session (called by ACP protocol)."""
|
|
state = await asyncio.to_thread(self.session_manager.get_session, session_id)
|
|
if state is None:
|
|
logger.warning("Session %s: model switch requested for missing session", session_id)
|
|
return None
|
|
# The picker swaps state.agent wholesale; mid-turn that strands the running agent and
|
|
# makes _finish_turn emit a spurious compression-rotation update. Same exclusion as
|
|
# the /model slash command.
|
|
with state.runtime_lock:
|
|
if state.is_running or state.command_op:
|
|
raise acp.RequestError(-32603, "Session is busy; switch models while the session is idle")
|
|
state.command_op = True
|
|
try:
|
|
# switch_model() does synchronous network I/O (models.dev, custom-endpoint probes,
|
|
# ~10 s cold) — off the loop, like the gateway, so other ACP sessions keep flowing.
|
|
try:
|
|
_old, requested_provider, resolved_model = await asyncio.to_thread(
|
|
self._switch_model, state, model_id, keep_endpoint=True)
|
|
except ModelRejected as exc:
|
|
# A model no provider can serve is a bad ``modelId`` param (-32602), not an agent
|
|
# internal error (-32603): the client attributes it to the request, not to Hermes (#72439).
|
|
# Only the switch_model rejection maps here; a ValueError from the rebuild itself
|
|
# (disabled provider, context window below the floor) stays on the -32603 path.
|
|
from acp.exceptions import RequestError
|
|
raise RequestError.invalid_params({"details": str(exc)}) from exc
|
|
finally:
|
|
with state.runtime_lock:
|
|
state.command_op = False
|
|
# Drain AFTER this response is queued, never inside it: a prompt that arrived
|
|
# mid-switch would otherwise run a whole turn before the client sees the
|
|
# (possibly failed) switch result.
|
|
self._schedule_soon(lambda: self._drain_queued_prompts(state, session_id, self._conn))
|
|
logger.info(
|
|
"Session %s: model switched to %s via provider %s", session_id, resolved_model, requested_provider
|
|
)
|
|
return SetSessionModelResponse()
|
|
|
|
async def set_session_mode(self, mode_id: str, session_id: str, **kwargs: Any) -> SetSessionModeResponse | None:
|
|
"""Persist the editor-requested mode so ACP clients do not fail on mode switches."""
|
|
state = await asyncio.to_thread(self.session_manager.get_session, session_id)
|
|
if state is None:
|
|
logger.warning("Session %s: mode switch requested for missing session", session_id)
|
|
return None
|
|
normalized_mode = str(mode_id or "").strip()
|
|
if normalized_mode not in self._MODES:
|
|
normalized_mode = self._MODE_DEFAULT
|
|
state.mode = normalized_mode
|
|
self.session_manager.save_session(session_id)
|
|
logger.info("Session %s: mode switched to %s", session_id, normalized_mode)
|
|
return SetSessionModeResponse()
|
|
|
|
async def set_config_option(
|
|
self, config_id: str, session_id: str, value: str, **kwargs: Any
|
|
) -> SetSessionConfigOptionResponse | None:
|
|
"""Accept ACP config option updates even when Hermes has no typed ACP config surface yet."""
|
|
state = await asyncio.to_thread(self.session_manager.get_session, session_id)
|
|
if state is None:
|
|
logger.warning("Session %s: config update requested for missing session", session_id)
|
|
return None
|
|
|
|
if str(config_id) == self._EDIT_APPROVAL_POLICY_CONFIG_ID:
|
|
state.mode = self._EDIT_APPROVAL_POLICY_TO_MODE.get(str(value), self._MODE_DEFAULT)
|
|
else:
|
|
options = getattr(state, "config_options", None)
|
|
if not isinstance(options, dict):
|
|
options = {}
|
|
options[str(config_id)] = value
|
|
state.config_options = options
|
|
self.session_manager.save_session(session_id)
|
|
logger.info("Session %s: config option %s updated", session_id, config_id)
|
|
return SetSessionConfigOptionResponse(config_options=[])
|
|
|
|
|
|
# ---- BEGIN PLUGIN-COMPAT (revert-scheduled; see COMPAT_MANIFEST.md) ----
|
|
# Names external plugins imported from this module before the Sep 2026 decomposition.
|
|
# Internal code MUST NOT use these (scripts/check_compat_pointers.py fails CI if it does).
|
|
# The whole block is removed by reverting the commit that added it.
|
|
from acp.schema import AgentThoughtChunk # noqa: F401,E402
|
|
from acp.schema import AudioContentBlock # noqa: F401,E402
|
|
from acp.schema import AvailableCommand # noqa: F401,E402
|
|
from acp.schema import AvailableCommandsUpdate # noqa: F401,E402
|
|
from acp.schema import BlobResourceContents # noqa: F401,E402
|
|
from acp.schema import EmbeddedResourceContentBlock # noqa: F401,E402
|
|
from acp.schema import ImageContentBlock # noqa: F401,E402
|
|
from pathlib import Path # noqa: F401,E402
|
|
from acp.schema import ResourceContentBlock # noqa: F401,E402
|
|
from acp.schema import TextResourceContents # noqa: F401,E402
|
|
from acp.schema import UnstructuredCommandInput # noqa: F401,E402
|
|
import base64 # noqa: F401,E402
|
|
import json # noqa: F401,E402
|
|
from urllib.parse import unquote # noqa: F401,E402
|
|
from urllib.parse import urlparse # noqa: F401,E402
|
|
|
|
|
|
_PLUGIN_COMPAT_LAZY = {
|
|
'ACP_MAX_MODELS_PER_PROVIDER': ('acp_adapter.model_catalog', 'ACP_MAX_MODELS_PER_PROVIDER'),
|
|
}
|
|
|
|
|
|
def __getattr__(name): # PEP 562 — lazy so no import cycles
|
|
target = _PLUGIN_COMPAT_LAZY.get(name)
|
|
if target is None:
|
|
raise AttributeError(f"module {__name__!r} has no attribute {name!r}")
|
|
import importlib
|
|
from hermes_cli.plugin_compat import warn_once
|
|
warn_once(__name__, name, *target)
|
|
return getattr(importlib.import_module(target[0]), target[1])
|
|
# ---- END PLUGIN-COMPAT ----
|