#!/usr/bin/env python3 """Process-local ThinkPool Hermes ACP tool policy. Never import this through the user profile. The bridge starts this file with Hermes' installed venv interpreter and passes a validated role policy in env. """ import json import os import re import sys from urllib.parse import urlsplit, urlunsplit POLICY_ENV = "THINKPOOL_HERMES_ACP_POLICY" CODING_TOOLS = frozenset({ "web_search", "web_extract", "terminal", "process", "read_file", "write_file", "patch", "search_files", "vision_analyze", "skills_list", "skill_view", "skill_manage", "browser_navigate", "browser_snapshot", "browser_click", "browser_type", "browser_scroll", "browser_back", "browser_press", "browser_get_images", "browser_vision", "browser_console", "browser_cdp", "browser_dialog", "todo", "memory", "execute_code", }) READ_ONLY_TOOLS = frozenset({"read_file", "search_files"}) ESSENTIAL_CODING_TOOLS = frozenset({ "terminal", "process", "read_file", "write_file", "patch", "search_files", }) def die(message): print("ThinkPool Hermes ACP policy error: " + message, file=sys.stderr) raise SystemExit(78) def policy(): raw = os.environ.get(POLICY_ENV) try: value = json.loads(raw) except Exception: die("missing or malformed policy") if not isinstance(value, dict) or value.get("version") != 1: die("unsupported policy") role = value.get("role") builtin = value.get("builtinTools") required_builtin = value.get("requiredBuiltinTools") tools = value.get("mcpTools") if role not in {"ordinary", "plan", "builder", "conductor", "reviewer", "manual-review"}: die("unknown role") if value.get("mcpServer") != "thinkpool" or not isinstance(builtin, list) or not isinstance(required_builtin, list) or not isinstance(tools, list): die("invalid tool policy") if not all(isinstance(x, str) and x and x.replace("_", "").isalnum() for x in builtin + required_builtin + tools): die("invalid tool name") forbidden = {"delegate_task", "session_search"} if forbidden.intersection(builtin) or forbidden.intersection(required_builtin) or forbidden.intersection(tools): die("delegation and session search are forbidden") required_mcp = { "ordinary": {"read_terminal"}, "plan": set(), "builder": {"mark_flow_done"}, "conductor": {"submit_flow_plan"}, "reviewer": {"submit_flow_review", "run_review_check", "read_review_file"}, "manual-review": {"run_review_check", "read_review_file"}, } if not required_mcp[role].issubset(tools): die("incomplete role MCP policy") restricted = role in {"plan", "conductor", "reviewer", "manual-review"} allowed_builtin = READ_ONLY_TOOLS if restricted else CODING_TOOLS required = READ_ONLY_TOOLS if restricted else ESSENTIAL_CODING_TOOLS if set(builtin) != allowed_builtin: die("restricted role requires exact read-only builtins" if restricted else "coding role requires the full approved builtin allowlist") if set(required_builtin) != required: die("invalid required builtin policy") return role, tuple(dict.fromkeys(builtin)), tuple(dict.fromkeys(required_builtin)), tuple(dict.fromkeys(tools)) ROLE, BUILTIN, REQUIRED_BUILTIN, MCP_TOOLS = policy() ALLOWED = frozenset(BUILTIN) | frozenset("mcp__thinkpool__" + x for x in MCP_TOOLS) | frozenset("mcp_thinkpool_" + x for x in MCP_TOOLS) def name_of(schema): if not isinstance(schema, dict): return "" fn = schema.get("function") return fn.get("name", "") if isinstance(fn, dict) else schema.get("name", "") def filter_schemas(items): return [item for item in (items or []) if name_of(item) in ALLOWED] def assert_exact_inventory(agent): """Reject an ACP lifecycle that lost the bridge-owned MCP surface. Hermes treats registration errors as non-fatal. That is acceptable for a standalone CLI, but never for a ThinkPool role: the bridge must not let a reconstructed agent accept a prompt with a partial policy. """ tools = list(getattr(agent, "tools", []) or []) names = {name_of(item) for item in tools} # Allowed is deliberately broader than required for ordinary/builder: # browser/provider/vision integrations are availability-gated upstream. missing_builtin = set(REQUIRED_BUILTIN) - names missing_mcp = [name for name in MCP_TOOLS if not ({"mcp__thinkpool__" + name, "mcp_thinkpool_" + name} & names)] extras = names - ALLOWED if missing_builtin or missing_mcp or extras: details = [] if missing_builtin: details.append("missing builtins " + ", ".join(sorted(missing_builtin))) if missing_mcp: details.append("missing MCP " + ", ".join(sorted(missing_mcp))) if extras: details.append("forbidden extras " + ", ".join(sorted(extras))) raise RuntimeError("ThinkPool exact inventory is incomplete: " + "; ".join(details)) agent.tools = tools agent.valid_tool_names = names # Patch before importing ACP server. Each new/resumed/reset ACP process reloads # this exact policy; no mutable shell alias or profile config participates. import toolsets toolsets.TOOLSETS["hermes-acp"] = { "description": "ThinkPool process-local ACP policy", "tools": list(BUILTIN), "includes": [] } # entry.main() normally discovers profile-configured MCP servers before it # creates the ACP server. ThinkPool ACP never inherits those profile servers; # the only permitted registration is the bridge's per-session `thinkpool` one. import tools.mcp_tool tools.mcp_tool.discover_mcp_tools = lambda *args, **kwargs: [] import model_tools _get_definitions = model_tools.get_tool_definitions def constrained_definitions(*args, **kwargs): return filter_schemas(_get_definitions(*args, **kwargs)) model_tools.get_tool_definitions = constrained_definitions import agent.memory_manager _inject_memory = agent.memory_manager.inject_memory_provider_tools def constrained_memory(agent): result = _inject_memory(agent) if hasattr(agent, "tools"): agent.tools = filter_schemas(agent.tools) agent.valid_tool_names = {name_of(x) for x in agent.tools} return result agent.memory_manager.inject_memory_provider_tools = constrained_memory import acp_adapter.session _expand = acp_adapter.session._expand_acp_enabled_toolsets def constrained_expand(toolsets_arg=None, mcp_server_names=None): requested = list(toolsets_arg or ["hermes-acp"]) if any(name not in {"hermes-acp", "mcp-thinkpool"} for name in requested): raise RuntimeError("ThinkPool ACP only permits hermes-acp and mcp-thinkpool toolsets") names = list(mcp_server_names or []) if any(name != "thinkpool" for name in names): raise RuntimeError("ThinkPool ACP only permits dynamic MCP server thinkpool") # Hermes 0.18.2 calls this from /tools with an already-expanded # ["hermes-acp", "mcp-thinkpool"] list and no mcp_server_names. Preserve # that exact legal expansion; otherwise /tools silently omits ThinkPool. return ["hermes-acp"] + (["mcp-thinkpool"] if names or "mcp-thinkpool" in requested else []) acp_adapter.session._expand_acp_enabled_toolsets = constrained_expand import acp_adapter.server # ThinkPool-only commands live in this process patch rather than in Hermes' # profile. They are intentionally narrow: no identity, shared configuration, # account tokens, or lifecycle/admin controls enter the room surface. _TP_REASONING_CONFIG_ID = "thinkpool_reasoning_effort" _TP_REASONING_LEVELS = frozenset({"none", "low", "medium", "high", "xhigh", "max"}) _TP_COMMANDS = ( {"name": "credits", "description": "Show safe Nous credit balance and top-up handoff"}, {"name": "status", "description": "Show session, model, context, version, and reasoning status"}, {"name": "reasoning", "description": "Set session-only reasoning effort", "input_hint": "low, medium, high, xhigh, max, none, or reset"}, ) _native_compact = getattr(acp_adapter.server.HermesACPAgent, "_cmd_compact", None) def thinkpool_compact(self, args, state): """Run a genuine manual compact and never label an unchanged list success.""" if not state.history: return "Nothing to compress — conversation is empty." try: agent = state.agent if not getattr(agent, "compression_enabled", True): return "Context compression is disabled for this agent." if not hasattr(agent, "_compress_context"): return "Context compression not available for this agent." from agent.model_metadata import estimate_request_tokens_rough original_history = state.history original_count = len(original_history) system_prompt = getattr(agent, "_cached_system_prompt", "") or "" tools = getattr(agent, "tools", None) or None original_tokens = estimate_request_tokens_rough(original_history, system_prompt=system_prompt, tools=tools) original_session_db = getattr(agent, "_session_db", None) try: # ACP sessions keep one native identity. Manual compact must bypass # an automatic-summary cooldown, matching Hermes' own documented # `/compress` contract, while disabling session rotation here. agent._session_db = None compressed, _ = agent._compress_context( original_history, system_prompt, approx_tokens=original_tokens, task_id=state.session_id, force=True, ) finally: agent._session_db = original_session_db if compressed is original_history: return "Compression made no progress — context unchanged." new_system_prompt = getattr(agent, "_cached_system_prompt", "") or system_prompt new_tools = getattr(agent, "tools", None) or tools new_tokens = estimate_request_tokens_rough(compressed, system_prompt=new_system_prompt, tools=new_tools) # A shorter message list can still be a larger model request when the # generated summary exceeds the discarded turns. That is not useful # compaction. Restore the exact pre-attempt prompt/history and do not # persist a boundary unless estimated request pressure actually falls. if new_tokens >= original_tokens: agent._cached_system_prompt = system_prompt return ( "Compression made no progress — context unchanged " f"(~{original_tokens:,} -> ~{new_tokens:,} estimated tokens)." ) state.history = compressed self.session_manager.save_session(state.session_id) return ( f"Context compressed: {original_count} -> {len(compressed)} messages\n" f"~{original_tokens:,} -> ~{new_tokens:,} tokens" ) except Exception as error: return f"Compression failed: {error}" if _native_compact: acp_adapter.server.HermesACPAgent._cmd_compact = thinkpool_compact def _reasoning_config(value): value = str(value or "").strip().lower() if value == "reset": return None if value not in _TP_REASONING_LEVELS: raise ValueError("Usage: /reasoning [low|medium|high|xhigh|max|none|reset]") return {"enabled": value != "none", **({"effort": value} if value != "none" else {})} def _apply_reasoning(state): config = getattr(state, "reasoning_config", None) # AIAgent consumes reasoning_config; setting it on the fresh process-local # agent is the important part. The state copy is carried across set_model. state.agent.reasoning_config = dict(config) if isinstance(config, dict) else None state.agent._reasoning_config = dict(config) if isinstance(config, dict) else None def _safe_topup_url(value): try: parsed = urlsplit(str(value or "")) host = (parsed.hostname or "").lower() if (parsed.scheme != "https" or host != "portal.nousresearch.com" or parsed.username is not None or parsed.password is not None or parsed.port not in {None, 443}): return None # Never relay account-derived paths, userinfo, query, fragment, or a # provider-selected subdomain. Org-pinned paths identify the account; # the generic billing page is the only safe room handoff. return urlunsplit(("https", "portal.nousresearch.com", "/billing", "", "")) except Exception: return None def _credits_command(): try: from agent.account_usage import build_credits_view view = build_credits_view(markdown=True) except Exception: return "Credits are unavailable right now; no balance was inferred." if view is None or not getattr(view, "logged_in", False): return "Credits are unavailable because this Hermes account is not signed in." lines = ["Nous credits"] # Never echo provider-rendered lines whole. Reconstruct only the exact # numeric balance shapes produced by Hermes' current account core; an # otherwise-valid prefix with an identity/credential suffix must fail the # full match rather than smuggling that suffix into the room transcript. balance = re.compile(r"^(Subscription credits|Top-up credits|Total usable|Rollover):\s*\$([0-9][0-9,]*(?:\.[0-9]{2})?)$") for line in list(getattr(view, "balance_lines", []) or []): rendered = str(line).strip() match = balance.fullmatch(rendered) if match: lines.append(f"{match.group(1)}: ${match.group(2)}") topup = _safe_topup_url(getattr(view, "topup_url", None)) if topup: lines.extend(["", "Top up: " + topup]) if len(lines) == 1: lines.append("Balance details are unavailable; no value was inferred.") return "\n".join(lines) _available_commands = getattr(acp_adapter.server.HermesACPAgent, "_available_commands", None) @classmethod def thinkpool_available_commands(cls): # Keep the upstream catalog canonical, then append exactly our process-local # commands. This drives ACP updates and /help from one registry. base = list(_available_commands.__func__(cls)) if _available_commands else [] try: from acp.schema import AvailableCommand, UnstructuredCommandInput known = {getattr(item, "name", "") for item in base} for spec in _TP_COMMANDS: if spec["name"] not in known: hint = spec.get("input_hint") base.append(AvailableCommand(name=spec["name"], description=spec["description"], input=UnstructuredCommandInput(hint=hint) if hint else None)) except Exception: # In fixture/minimal ACP environments a dict still proves the registry # behavior without making bootstrap startup fail. base.extend(spec for spec in _TP_COMMANDS if spec["name"] not in {getattr(item, "name", item.get("name", "") if isinstance(item, dict) else "") for item in base}) return base if _available_commands: acp_adapter.server.HermesACPAgent._available_commands = thinkpool_available_commands _slash = getattr(acp_adapter.server.HermesACPAgent, "_handle_slash_command", None) def thinkpool_slash(self, text, state): parts = str(text or "").split(maxsplit=1) command = parts[0].lstrip("/").lower() if parts else "" args = parts[1].strip() if len(parts) > 1 else "" if command == "credits": return _credits_command() if command == "reasoning": if not args: cfg = getattr(state, "reasoning_config", None) level = "reset/default" if not cfg else ("none" if cfg.get("enabled") is False else cfg.get("effort", "medium")) return "Session reasoning: " + level try: state.reasoning_config = _reasoning_config(args) except ValueError as error: return str(error) _apply_reasoning(state) self.session_manager.save_session(state.session_id) return "Session reasoning " + ("reset to model default" if args.lower() == "reset" else "set to " + args.lower()) if command == "status": context = self._cmd_context("", state) model = self._cmd_model("", state) version = self._cmd_version("", state) cfg = getattr(state, "reasoning_config", None) effort = "default" if not cfg else ("none" if cfg.get("enabled") is False else cfg.get("effort", "medium")) return "\n".join([model, "Session: " + str(state.session_id), "Reasoning: " + effort, version, context]) return _slash(self, text, state) if _slash else None if _slash: acp_adapter.server.HermesACPAgent._handle_slash_command = thinkpool_slash _help = getattr(acp_adapter.server.HermesACPAgent, "_cmd_help", None) def thinkpool_help(self, args, state): try: lines = ["Available commands:", ""] for item in self._available_commands(): name = getattr(item, "name", "") description = getattr(item, "description", "") if isinstance(item, dict): name, description = item.get("name", ""), item.get("description", "") lines.append(f" /{name:10s} {description}") return "\n".join(lines) except Exception: return _help(self, args, state) if _help else "Available commands unavailable." if _help: acp_adapter.server.HermesACPAgent._cmd_help = thinkpool_help _set_config = getattr(acp_adapter.server.HermesACPAgent, "set_config_option", None) async def thinkpool_set_config(self, config_id, session_id, value, **kwargs): if str(config_id) != _TP_REASONING_CONFIG_ID: return await _set_config(self, config_id, session_id, value, **kwargs) if _set_config else None state = self.session_manager.get_session(session_id) if state is None: return None try: state.reasoning_config = _reasoning_config(value) except ValueError: return None _apply_reasoning(state) self.session_manager.save_session(session_id) try: from acp.schema import SetSessionConfigOptionResponse return SetSessionConfigOptionResponse(config_options=[]) except Exception: # Minimal fixture/older ACP environments still need a JSON-serializable # response; a bare object wedges the JSON-RPC encoder after the config # was already applied and makes the bridge time out dishonestly. return {"configOptions": []} acp_adapter.server.HermesACPAgent.set_config_option = thinkpool_set_config # Hermes 0.18.2 emits the real result through ``tool.completed`` immediately, # but its ACP callback ignores that event and waits for the next model-step # summary. Parallel Read/search calls are not always present in that summary, # leaving clients with a ToolCallStart and no ToolCallUpdate forever. Complete # from the authoritative execution callback and consume the same per-name FIFO; # the later step callback then sees an empty queue and cannot double-emit. If a # future Hermes release handles completion itself, it consumes the queue before # this wrapper runs and this compatibility branch becomes a no-op. from collections import deque import acp_adapter.events import acp_adapter.tools _tool_progress_factory = acp_adapter.server.make_tool_progress_cb def completion_aware_tool_progress_factory( conn, session_id, loop, tool_call_ids, tool_call_meta, edit_approval_policy_getter=None, ): upstream = _tool_progress_factory( conn, session_id, loop, tool_call_ids, tool_call_meta, edit_approval_policy_getter=edit_approval_policy_getter, ) def progress(event_type, name=None, preview=None, args=None, **kwargs): upstream(event_type, name, preview, args, **kwargs) if event_type != "tool.completed" or not name: return queue = tool_call_ids.get(name) if isinstance(queue, str): queue = deque([queue]) tool_call_ids[name] = queue if not queue: return tool_call_id = queue.popleft() meta = tool_call_meta.pop(tool_call_id, {}) result = kwargs.get("result") update = acp_adapter.tools.build_tool_complete( tool_call_id, name, result=str(result) if result is not None else None, function_args=meta.get("args"), snapshot=meta.get("snapshot"), ) acp_adapter.events._send_update(conn, session_id, loop, update) if not queue: tool_call_ids.pop(name, None) return progress acp_adapter.server.make_tool_progress_cb = completion_aware_tool_progress_factory _register = acp_adapter.server.HermesACPAgent._register_session_mcp_servers async def constrained_register(self, state, mcp_servers): if any(getattr(server, "name", None) != "thinkpool" for server in (mcp_servers or [])): raise RuntimeError("ThinkPool ACP only permits dynamic MCP server thinkpool") if mcp_servers: # SessionState is process-local. Keep only the validated descriptors so # a subsequent set_model reconstruction can re-register the same MCP. state._thinkpool_mcp_servers = tuple(mcp_servers) await _register(self, state, mcp_servers) _apply_reasoning(state) assert_exact_inventory(state.agent) acp_adapter.server.HermesACPAgent._register_session_mcp_servers = constrained_register # Hermes 0.18.2's session/set_model creates a fresh state.agent. Upstream # does not re-run ACP MCP registration, so the new agent can expose only its # built-ins while the request still returns success. Keep the old state until # the replacement has re-registered and passed the same exact policy check. _set_model = acp_adapter.server.HermesACPAgent.set_session_model async def constrained_set_model(self, model_id, session_id, **kwargs): state = self.session_manager.get_session(session_id) if state is None: return await _set_model(self, model_id, session_id, **kwargs) old_agent, old_model = state.agent, getattr(state, "model", None) try: result = await _set_model(self, model_id, session_id, **kwargs) if result is None: raise RuntimeError("Hermes did not acknowledge model switch") servers = getattr(state, "_thinkpool_mcp_servers", ()) if MCP_TOOLS and not servers: raise RuntimeError("ThinkPool MCP registration is unavailable after model switch") await constrained_register(self, state, list(servers)) _apply_reasoning(state) assert_exact_inventory(state.agent) self.session_manager.save_session(session_id) return result except Exception: # Fail closed and restore the prior usable agent/model. A later bridge # /tools probe is the external acknowledgement before UI persistence. state.agent, state.model = old_agent, old_model try: servers = getattr(state, "_thinkpool_mcp_servers", ()) if servers: await constrained_register(self, state, list(servers)) self.session_manager.save_session(session_id) except Exception: pass raise acp_adapter.server.HermesACPAgent.set_session_model = constrained_set_model from acp_adapter.entry import main main()