8bb1d15da4
Automated cleanup via pyflakes + autoflake with manual review.
Changes:
- Removed unused stdlib imports (os, sys, json, pathlib.Path, etc.)
- Removed unused typing imports (List, Dict, Any, Optional, Tuple, Set, etc.)
- Removed unused internal imports (hermes_cli.auth, hermes_cli.config, etc.)
- Fixed cli.py: removed 8 shadowed banner imports (imported from hermes_cli.banner
then immediately redefined locally — only build_welcome_banner is actually used)
- Added noqa comments to imports that appear unused but serve a purpose:
- Re-exports (gateway/session.py SessionResetPolicy, tools/terminal_tool.py
is_interrupted/_interrupt_event)
- SDK presence checks in try/except (daytona, fal_client, discord)
- Test mock targets (auxiliary_client.py Path, mcp_config.py get_hermes_home)
Zero behavioral changes. Full test suite passes (6162/6162, 2 pre-existing
streaming test failures unrelated to this change).
78 lines
2.5 KiB
Python
78 lines
2.5 KiB
Python
"""ACP permission bridging — maps ACP approval requests to hermes approval callbacks."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import asyncio
|
|
import logging
|
|
from concurrent.futures import TimeoutError as FutureTimeout
|
|
from typing import Callable
|
|
|
|
from acp.schema import (
|
|
AllowedOutcome,
|
|
PermissionOption,
|
|
)
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
# Maps ACP PermissionOptionKind -> hermes approval result strings
|
|
_KIND_TO_HERMES = {
|
|
"allow_once": "once",
|
|
"allow_always": "always",
|
|
"reject_once": "deny",
|
|
"reject_always": "deny",
|
|
}
|
|
|
|
|
|
def make_approval_callback(
|
|
request_permission_fn: Callable,
|
|
loop: asyncio.AbstractEventLoop,
|
|
session_id: str,
|
|
timeout: float = 60.0,
|
|
) -> Callable[[str, str], str]:
|
|
"""
|
|
Return a hermes-compatible ``approval_callback(command, description) -> str``
|
|
that bridges to the ACP client's ``request_permission`` call.
|
|
|
|
Args:
|
|
request_permission_fn: The ACP connection's ``request_permission`` coroutine.
|
|
loop: The event loop on which the ACP connection lives.
|
|
session_id: Current ACP session id.
|
|
timeout: Seconds to wait for a response before auto-denying.
|
|
"""
|
|
|
|
def _callback(command: str, description: str) -> str:
|
|
options = [
|
|
PermissionOption(option_id="allow_once", kind="allow_once", name="Allow once"),
|
|
PermissionOption(option_id="allow_always", kind="allow_always", name="Allow always"),
|
|
PermissionOption(option_id="deny", kind="reject_once", name="Deny"),
|
|
]
|
|
import acp as _acp
|
|
|
|
tool_call = _acp.start_tool_call("perm-check", command, kind="execute")
|
|
|
|
coro = request_permission_fn(
|
|
session_id=session_id,
|
|
tool_call=tool_call,
|
|
options=options,
|
|
)
|
|
|
|
try:
|
|
future = asyncio.run_coroutine_threadsafe(coro, loop)
|
|
response = future.result(timeout=timeout)
|
|
except (FutureTimeout, Exception) as exc:
|
|
logger.warning("Permission request timed out or failed: %s", exc)
|
|
return "deny"
|
|
|
|
outcome = response.outcome
|
|
if isinstance(outcome, AllowedOutcome):
|
|
option_id = outcome.option_id
|
|
# Look up the kind from our options list
|
|
for opt in options:
|
|
if opt.option_id == option_id:
|
|
return _KIND_TO_HERMES.get(opt.kind, "deny")
|
|
return "once" # fallback for unknown option_id
|
|
else:
|
|
return "deny"
|
|
|
|
return _callback
|