hermes-admin/hermes-agent
1
0
Fork 0
Code Issues Pull Requests Actions 28 Packages Projects Releases Wiki Activity

refactor(acp): session/events/edit_approval/permissions/provenance/entry/auth — hoist lazy hermes_constants imports, fold branches, pack literals

Browse Source
This commit is contained in:
Teknium
2026-09-02 19:45:20 -07:00
parent a522df03ed
commit 3a7ce79912
7 changed files with 42 additions and 86 deletions
Download Patch File Download Diff File Expand all files Collapse all files

18
acp_adapter/auth.py
View File

@@ -19,12 +19,11 @@ def detect_provider() -> Optional[str]:
from hermes_cli.runtime_provider import resolve_runtime_provider
runtime = resolve_runtime_provider()
api_key, provider = runtime.get("api_key"), runtime.get("provider")
if not isinstance(provider, str) or not provider.strip():
return None
if (isinstance(api_key, str) and api_key.strip()) or callable(api_key):
if isinstance(provider, str) and provider.strip() and (
(isinstance(api_key, str) and api_key.strip()) or callable(api_key)):
return provider.strip().lower()
except Exception:
return None
pass
return None
@@ -45,13 +44,8 @@ def build_auth_methods() -> list[Any]:
description=f"Authenticate Hermes using the currently configured {provider} runtime credentials.",
))
methods.append(TerminalAuthMethod(
id=TERMINAL_SETUP_AUTH_METHOD_ID,
name="Configure Hermes provider",
description=(
"Open Hermes' interactive model/provider setup in a terminal. "
"Use this when Hermes has not been configured on this machine yet."
),
type="terminal",
args=["--setup"],
id=TERMINAL_SETUP_AUTH_METHOD_ID, name="Configure Hermes provider", type="terminal", args=["--setup"],
description=("Open Hermes' interactive model/provider setup in a terminal. "
"Use this when Hermes has not been configured on this machine yet."),
))
return methods

19
acp_adapter/edit_approval.py
View File

@@ -58,11 +58,11 @@ def reset_edit_approval_requester(token: Token) -> None:
def _read_text_if_exists(path: str) -> str | None:
p = Path(path).expanduser()
if not p.exists():
return None
if not p.is_file():
if p.is_file():
return p.read_text(encoding="utf-8", errors="replace")
if p.exists():
raise OSError(f"Cannot edit non-file path: {path}")
return p.read_text(encoding="utf-8", errors="replace")
return None
def _required_path(arguments: dict[str, Any]) -> str:
@@ -92,8 +92,7 @@ def _proposal_for_patch_replace(arguments: dict[str, Any]) -> EditProposal:
from tools.fuzzy_match import fuzzy_find_and_replace
new_text, match_count, _strategy, error = fuzzy_find_and_replace(
old_text, str(old_string), str(new_string), bool(arguments.get("replace_all", False)),
)
old_text, str(old_string), str(new_string), bool(arguments.get("replace_all", False)))
if error or match_count == 0:
raise ValueError(error or f"Could not find match for old_string in {path}")
return EditProposal("patch", path, old_text, new_text, dict(arguments))
@@ -154,10 +153,8 @@ def should_auto_approve_edit(proposal: EditProposal, policy: str, cwd: str | Non
if policy == AUTO_APPROVE_WORKSPACE:
# tempfile.gettempdir() is the real temp root on every platform
# (``/private/tmp`` on macOS since resolve() follows the symlink).
if path.is_relative_to(Path(tempfile.gettempdir()).resolve(strict=False)):
return True
if cwd:
return path.is_relative_to(Path(cwd).expanduser().resolve(strict=False))
return path.is_relative_to(Path(tempfile.gettempdir()).resolve(strict=False)) or (
bool(cwd) and path.is_relative_to(Path(cwd).expanduser().resolve(strict=False)))
return False
@@ -225,8 +222,6 @@ def make_acp_edit_approval_requester(
PermissionOption(option_id="deny", kind="reject_once", name="Deny")],
timeout=timeout, what="Edit approval request",
)
if response is None:
return False
outcome = getattr(response, "outcome", None)
return getattr(outcome, "outcome", None) == "selected" and getattr(outcome, "option_id", None) == "allow_once"

11
acp_adapter/entry.py
View File

@@ -52,8 +52,7 @@ class _BenignProbeMethodFilter(logging.Filter):
if not isinstance(exc, RequestError) or getattr(exc, "code", None) != -32601:
return True
data = getattr(exc, "data", None)
method = data.get("method") if isinstance(data, dict) else None
return method not in _BENIGN_PROBE_METHODS
return not (isinstance(data, dict) and data.get("method") in _BENIGN_PROBE_METHODS)
def _setup_logging() -> None:
@@ -61,9 +60,8 @@ def _setup_logging() -> None:
from agent.redact import RedactingFormatter
handler = logging.StreamHandler(sys.stderr)
handler.setFormatter(RedactingFormatter(
"%(asctime)s [%(levelname)s] %(name)s: %(message)s", datefmt="%Y-%m-%d %H:%M:%S",
))
handler.setFormatter(RedactingFormatter("%(asctime)s [%(levelname)s] %(name)s: %(message)s",
datefmt="%Y-%m-%d %H:%M:%S"))
handler.addFilter(_BenignProbeMethodFilter())
root = logging.getLogger()
root.handlers.clear()
@@ -166,8 +164,7 @@ def main(argv: list[str] | None = None) -> None:
if getattr(args, flag):
return action()
if args.setup_browser:
rc = _run_setup_browser(assume_yes=args.assume_yes)
if rc != 0:
if rc := _run_setup_browser(assume_yes=args.assume_yes):
sys.exit(rc)
return

8
acp_adapter/events.py
View File

@@ -146,7 +146,7 @@ def make_step_cb(
"""Create a ``step_callback(api_call_count: int, prev_tools: list)`` for AIAgent."""
def _step(api_call_count: int, prev_tools: Any = None) -> None:
if not prev_tools or not isinstance(prev_tools, list):
if not isinstance(prev_tools, list):
return
for tool_info in prev_tools:
tool_name = result = function_args = None
@@ -168,10 +168,8 @@ def make_step_cb(
tc_id, tool_name, result=str(result) if result is not None else None,
function_args=function_args or meta.get("args"), snapshot=meta.get("snapshot"),
))
if tool_name == "todo":
plan_update = _build_plan_update_from_todo_result(result)
if plan_update is not None:
_send_update(conn, session_id, loop, plan_update)
if tool_name == "todo" and (plan_update := _build_plan_update_from_todo_result(result)) is not None:
_send_update(conn, session_id, loop, plan_update)
if not queue:
tool_call_ids.pop(tool_name, None)

7
acp_adapter/permissions.py
View File

@@ -25,9 +25,9 @@ def _permission_option_supports_kind(kind: str) -> bool:
"""Return whether the installed ACP SDK accepts a permission option kind."""
try:
PermissionOption(option_id="__probe__", kind=kind, name="probe")
return True
except Exception:
return False
return True
def _build_permission_options(
@@ -105,9 +105,8 @@ def make_approval_callback(request_permission_fn: Callable, loop: asyncio.Abstra
def _callback(command: str, description: str, *, allow_permanent: bool = True,
allow_session: bool = True, smart_denied: bool = False, **_: object) -> str:
options = _build_permission_options(
allow_permanent=allow_permanent, allow_session=allow_session, smart_denied=smart_denied,
)
options = _build_permission_options(allow_permanent=allow_permanent, allow_session=allow_session,
smart_denied=smart_denied)
response, timed_out = await_permission(
request_permission_fn, loop, session_id, tool_call=_build_permission_tool_call(command, description),
options=options, timeout=timeout, what="Permission request",

18
acp_adapter/provenance.py
View File

@@ -46,9 +46,7 @@ def build_session_provenance(
# Walk parents to the lineage root. Only compression-split parents
# (parent.end_reason == 'compression') count toward depth — delegate/branch
# children share the parent_session_id column but are not compaction boundaries.
root_id = current_hermes_session_id
compression_depth = 0
cursor_parent = parent_id
root_id, compression_depth, cursor_parent = current_hermes_session_id, 0, parent_id
seen = {current_hermes_session_id}
for _ in range(_MAX_WALK):
if not cursor_parent or cursor_parent in seen:
@@ -65,12 +63,9 @@ def build_session_provenance(
is_continuation = bool(parent_id) and _is_compression_end(_get_row(db, parent_id))
provenance: Dict[str, Any] = {
"acpSessionId": acp_session_id,
"currentHermesSessionId": current_hermes_session_id,
"rootHermesSessionId": root_id,
"parentHermesSessionId": parent_id,
"sessionKind": "continuation" if is_continuation else "root",
"compressionDepth": compression_depth,
"acpSessionId": acp_session_id, "currentHermesSessionId": current_hermes_session_id,
"rootHermesSessionId": root_id, "parentHermesSessionId": parent_id,
"sessionKind": "continuation" if is_continuation else "root", "compressionDepth": compression_depth,
}
if previous_hermes_session_id:
provenance["previousHermesSessionId"] = previous_hermes_session_id
@@ -86,7 +81,6 @@ def session_provenance_meta(
db: Any, acp_session_id: str, current_hermes_session_id: str, *, previous_hermes_session_id: Optional[str] = None,
) -> Optional[Dict[str, Any]]:
"""Return a ready ``_meta`` payload: ``{"hermes": {"sessionProvenance": ...}}``."""
prov = build_session_provenance(
db, acp_session_id, current_hermes_session_id, previous_hermes_session_id=previous_hermes_session_id,
)
prov = build_session_provenance(db, acp_session_id, current_hermes_session_id,
previous_hermes_session_id=previous_hermes_session_id)
return None if prov is None else {"hermes": {"sessionProvenance": prov}}

47
acp_adapter/session.py
View File

@@ -6,7 +6,7 @@ survive process restarts and appear in ``session_search``; ``load_session`` /
"""
from __future__ import annotations
from hermes_constants import get_hermes_home
from hermes_constants import get_hermes_home, translate_cwd_for_wsl_backend, windows_path_to_wsl
import copy
import json
@@ -19,7 +19,6 @@ import time
import uuid
from datetime import datetime, timezone
from dataclasses import dataclass, field
from threading import Lock
from typing import Any, Dict, List, Optional
logger = logging.getLogger(__name__)
@@ -28,8 +27,6 @@ logger = logging.getLogger(__name__)
def _translate_acp_cwd(cwd: str) -> str:
"""Translate Windows ACP cwd values (``E:\\Projects``, ``\\\\wsl.localhost\\``) to POSIX form
when Hermes runs in WSL so agents, tools, and persisted sessions agree; no-op elsewhere."""
from hermes_constants import translate_cwd_for_wsl_backend
return translate_cwd_for_wsl_backend(str(cwd))
@@ -37,8 +34,6 @@ def _normalize_cwd_for_compare(cwd: str | None) -> str:
expanded = os.path.expanduser(str(cwd or ".").strip() or ".")
# Windows drive paths -> WSL mount form so history filters match across hosts.
from hermes_constants import windows_path_to_wsl
translated = windows_path_to_wsl(expanded)
if translated is not None:
expanded = translated
@@ -84,7 +79,6 @@ def _updated_at_sort_key(value: Any) -> float:
def _acp_stderr_print(*args, **kwargs) -> None:
"""Route incidental AIAgent output to stderr; ACP reserves stdout for JSON-RPC."""
kwargs = dict(kwargs)
kwargs.setdefault("file", sys.stderr)
print(*args, **kwargs)
@@ -120,14 +114,8 @@ def _parse_model_config(mc: Any) -> dict:
def _session_info(sid: str, cwd: str, model: Any, history_len: int, title: Any, preview: Any,
updated_at: Any) -> Dict[str, Any]:
return {
"session_id": sid,
"cwd": cwd,
"model": model,
"history_len": history_len,
"title": _build_session_title(title, preview, cwd),
"updated_at": _format_updated_at(updated_at),
}
return {"session_id": sid, "cwd": cwd, "model": model, "history_len": history_len,
"title": _build_session_title(title, preview, cwd), "updated_at": _format_updated_at(updated_at)}
def _first_user_preview(history: List[Dict[str, Any]], default: str) -> str:
@@ -147,7 +135,7 @@ class SessionState:
cancel_event: Any = None # threading.Event
is_running: bool = False
queued_prompts: List[str] = field(default_factory=list)
runtime_lock: Any = field(default_factory=Lock)
runtime_lock: Any = field(default_factory=threading.Lock)
current_prompt_text: str = ""
interrupted_prompt_text: str = ""
@@ -162,7 +150,7 @@ class SessionManager:
"""``agent_factory``: AIAgent-like factory (tests); default builds a real AIAgent from
the runtime provider config. ``db``: SessionDB; default lazily opens ``~/.hermes/state.db``."""
self._sessions: Dict[str, SessionState] = {}
self._lock = Lock()
self._lock = threading.Lock()
self._agent_factory = agent_factory
self._db_instance = db # None → lazy-init on first use
@@ -313,8 +301,7 @@ class SessionManager:
# active=0 rows; replace_messages() would DELETE those (and, after a compression
# id rotation, clobber the ended parent transcript). Skip it in that case.
agent = state.agent
agent_db = getattr(agent, "_session_db", None)
if agent_db is not None and agent_db is db and bool(getattr(agent, "_session_db_created", False)):
if getattr(agent, "_session_db", None) is db and getattr(agent, "_session_db_created", False):
return
# A non-owning agent (model switch, /restore: fresh agent, _session_db_created=False)
# may still sit on archived rows, so replace ONLY the active=1 set: on a fresh
@@ -351,10 +338,9 @@ class SessionManager:
try:
agent = self._make_agent(
session_id=session_id, cwd=cwd, model=model,
session_id=session_id, cwd=cwd, model=model, api_mode=meta.get("api_mode") or None,
requested_provider=meta.get("provider") or row.get("billing_provider"),
base_url=meta.get("base_url") or row.get("billing_base_url"),
api_mode=meta.get("api_mode") or None)
base_url=meta.get("base_url") or row.get("billing_base_url"))
except Exception:
logger.warning("Failed to recreate agent for ACP session %s", session_id, exc_info=True)
return None
@@ -366,8 +352,7 @@ class SessionManager:
# ---- internal -----------------------------------------------------------
def _make_agent(self, *, session_id: str, cwd: str, model: str | None = None,
requested_provider: str | None = None, base_url: str | None = None,
api_mode: str | None = None):
requested_provider: str | None = None, base_url: str | None = None, api_mode: str | None = None):
if self._agent_factory is not None:
return self._agent_factory()
@@ -388,22 +373,16 @@ class SessionManager:
if not isinstance(cfg, dict) or cfg.get("enabled", True) is not False
]
kwargs = {
"platform": "acp",
"platform": "acp", "quiet_mode": True, "session_id": session_id, "session_db": self._get_db(),
"enabled_toolsets": _expand_acp_enabled_toolsets(["hermes-acp"], mcp_server_names=configured_mcp_servers),
"quiet_mode": True,
"session_id": session_id,
"session_db": self._get_db(),
"model": model or default_model,
}
try:
runtime = resolve_runtime_provider(requested=requested_provider or config_provider)
kwargs.update({
"provider": runtime.get("provider"),
"api_mode": api_mode or runtime.get("api_mode"),
"base_url": base_url or runtime.get("base_url"),
"api_key": runtime.get("api_key"),
"command": runtime.get("command"),
"args": list(runtime.get("args") or []),
"provider": runtime.get("provider"), "api_mode": api_mode or runtime.get("api_mode"),
"base_url": base_url or runtime.get("base_url"), "api_key": runtime.get("api_key"),
"command": runtime.get("command"), "args": list(runtime.get("args") or []),
})
except Exception:
logger.debug("ACP session falling back to default provider resolution", exc_info=True)