wehub-resource-sync
296 lines
12 KiB
Python
296 lines
12 KiB
Python
"""Hermes-native tool-approval mirror (TUI → web elicitation).
|
||
|
||
The native ``hermes`` TUI gates commands it flags as dangerous with an
|
||
in-terminal approval prompt (its own ``tools/approval.py`` gate). That prompt
|
||
lives only in the TUI; to also surface it in the Omnigent web UI (so a user can
|
||
approve from the chat view, not only the embedded terminal), the runner watches
|
||
the Hermes pane:
|
||
|
||
1. poll ``capture-pane`` and detect the approval PANEL — the interactive TUI
|
||
renders a prompt_toolkit panel titled ``⚠️ Dangerous Command`` with NUMBERED
|
||
choices (``❯ 1. Allow once`` … ``4. Deny``), NOT the legacy ``Choice
|
||
[o/s/a/D]:`` ``input()`` prompt (that path is fail-closed while prompt_toolkit
|
||
owns the terminal). Verified against hermes-agent ``cli.py``
|
||
``_get_approval_display_fragments`` + the number-key bindings,
|
||
2. POST it to the server's generic ``native-permission-request`` hook, which
|
||
publishes ``response.elicitation_request`` and parks for the web verdict,
|
||
3. on the verdict, send the choice's DIGIT key (e.g. ``1`` = Allow once, ``4`` =
|
||
Deny) into the pane — Hermes' number-key binding selects AND confirms in one
|
||
press,
|
||
4. if the panel instead disappears on its own (answered in the embedded
|
||
terminal), POST ``external_elicitation_resolved`` so the parked web card
|
||
clears.
|
||
|
||
This does NOT suppress Hermes' own gate — its panel stays the source of truth
|
||
and the fallback if pane detection ever fails (the user can still pick in the
|
||
terminal). Mirrors :mod:`omnigent.cursor_native_permissions`. NB: Hermes only
|
||
prompts for commands it flags *dangerous* (and may auto-approve low-risk ones via
|
||
smart-approval), so non-dangerous tools won't raise a card.
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import asyncio
|
||
import hashlib
|
||
import logging
|
||
import re
|
||
from dataclasses import dataclass
|
||
from pathlib import Path
|
||
|
||
import httpx
|
||
|
||
from omnigent.hermes_native_bridge import capture_hermes_pane, send_hermes_pane_keys
|
||
|
||
_logger = logging.getLogger(__name__)
|
||
|
||
_POLL_INTERVAL_S = 0.3
|
||
# The hook parks server-side until a human answers; allow a day so the runner's
|
||
# POST never abandons a live prompt.
|
||
_POST_TIMEOUT_S = 86400.0
|
||
|
||
# Hermes' interactive TUI renders the dangerous-command gate as a prompt_toolkit
|
||
# PANEL (cli.py ``_get_approval_display_fragments``), NOT the legacy ``input()``
|
||
# ``Choice [o/s/a/D]:`` prompt — that path is fail-closed while prompt_toolkit
|
||
# owns the terminal. The panel is titled ``⚠️ Dangerous Command`` and lists
|
||
# NUMBERED choices (``❯ 1. Allow once`` … ``4. Deny``); pressing the number both
|
||
# selects and confirms (cli.py number-key bindings call _handle_approval_selection).
|
||
# So we detect the panel by title + numbered choices and answer with the digit.
|
||
_TITLE_RE = re.compile(r"Dangerous Command", re.IGNORECASE)
|
||
# A numbered choice row, e.g. "❯ 1. Allow once" / " 4. Deny" (box borders ignored).
|
||
_CHOICE_RE = re.compile(
|
||
r"(?P<num>\d)\.\s*(?P<label>Allow once|Allow for this session|"
|
||
r"Add to permanent allowlist|Deny)",
|
||
re.IGNORECASE,
|
||
)
|
||
|
||
|
||
@dataclass(frozen=True)
|
||
class HermesApprovalPrompt:
|
||
"""A parsed Hermes dangerous-command approval panel.
|
||
|
||
:param command: Best-effort command/description preview from the panel.
|
||
:param message: Human-readable card message.
|
||
:param preview: Compact preview for the card.
|
||
:param accept_key: digit key that selects+confirms "Allow once" (e.g. ``"1"``).
|
||
:param decline_key: digit key that selects+confirms "Deny" (e.g. ``"4"`` with
|
||
the permanent-allowlist option, else ``"3"``).
|
||
:param block_hash: Stable hash of the preview (kept for debugging/preview).
|
||
"""
|
||
|
||
command: str
|
||
message: str
|
||
preview: str
|
||
accept_key: str
|
||
decline_key: str
|
||
block_hash: str
|
||
|
||
|
||
def hermes_permission_elicitation_id(session_id: str, token: str) -> str:
|
||
"""Return the deterministic Omnigent elicitation id for a Hermes prompt.
|
||
|
||
*token* identifies one approval episode (a per-session counter), not the
|
||
scraped content, so a re-render never spawns a duplicate card.
|
||
"""
|
||
return f"elicit_hermes_{session_id}_{token}"
|
||
|
||
|
||
def _strip_border(line: str) -> str:
|
||
"""Strip the panel's box-drawing borders/padding from *line*."""
|
||
return line.strip().strip("│").strip()
|
||
|
||
|
||
def parse_hermes_approval_prompt(pane: str) -> HermesApprovalPrompt | None:
|
||
"""Parse a Hermes ``⚠️ Dangerous Command`` approval panel from pane text.
|
||
|
||
Requires the panel title AND both an "Allow once" and a "Deny" numbered
|
||
choice (so a title lingering without the live choice list is not re-detected),
|
||
and reads the digit key for each from the panel itself — robust to whether the
|
||
permanent-allowlist option is offered (Deny is ``4`` with it, ``3`` without).
|
||
|
||
:param pane: Visible pane text from ``capture-pane -p``.
|
||
:returns: The parsed prompt, or ``None`` when no live panel is visible.
|
||
"""
|
||
if not pane or not _TITLE_RE.search(pane):
|
||
return None
|
||
lines = pane.splitlines()
|
||
label_to_key: dict[str, str] = {}
|
||
first_choice_idx: int | None = None
|
||
for i, line in enumerate(lines):
|
||
match = _CHOICE_RE.search(line)
|
||
if match:
|
||
label_to_key.setdefault(match.group("label").lower(), match.group("num"))
|
||
if first_choice_idx is None:
|
||
first_choice_idx = i
|
||
accept_key = label_to_key.get("allow once")
|
||
decline_key = label_to_key.get("deny")
|
||
if accept_key is None or decline_key is None or first_choice_idx is None:
|
||
return None
|
||
|
||
# Best-effort command/description preview: the panel lines between the title
|
||
# and the first choice, minus borders/blank/title lines.
|
||
title_idx = next((i for i, line in enumerate(lines) if _TITLE_RE.search(line)), 0)
|
||
preview_parts: list[str] = []
|
||
for line in lines[title_idx + 1 : first_choice_idx]:
|
||
text = _strip_border(line)
|
||
if text and not _TITLE_RE.search(text):
|
||
preview_parts.append(text)
|
||
preview = " ".join(preview_parts)[:1024]
|
||
block_hash = hashlib.sha256(preview.encode("utf-8")).hexdigest()[:16]
|
||
return HermesApprovalPrompt(
|
||
command=preview,
|
||
message="Hermes flagged a dangerous command. Run it?",
|
||
preview=preview or "dangerous command",
|
||
accept_key=accept_key,
|
||
decline_key=decline_key,
|
||
block_hash=block_hash,
|
||
)
|
||
|
||
|
||
async def supervise_hermes_approval_mirror(
|
||
*,
|
||
base_url: str,
|
||
headers: dict[str, str],
|
||
session_id: str,
|
||
bridge_dir: Path,
|
||
auth: httpx.Auth | None = None,
|
||
poll_interval_s: float = _POLL_INTERVAL_S,
|
||
) -> None:
|
||
"""Poll the Hermes pane and mirror its approval prompts to web elicitations.
|
||
|
||
Runs for the session's lifetime (cancelled on teardown). At most one prompt
|
||
is active at a time: a new block spawns a task that parks on the server and,
|
||
on the web verdict, sends the keystroke; a block that vanishes while still
|
||
parked means the user answered in the TUI, so the parked card is released.
|
||
|
||
:param base_url: Server base URL.
|
||
:param headers: Auth/routing headers for the runner's requests.
|
||
:param session_id: Omnigent conversation id.
|
||
:param bridge_dir: The hermes-native bridge dir holding ``tmux.json``.
|
||
:param auth: Optional httpx auth for the runner's requests.
|
||
:param poll_interval_s: Pane poll cadence in seconds.
|
||
"""
|
||
active: dict[str, object] | None = None
|
||
episode = 0
|
||
timeout = httpx.Timeout(_POST_TIMEOUT_S, connect=10.0)
|
||
async with httpx.AsyncClient(
|
||
base_url=base_url, headers=headers, auth=auth, timeout=timeout
|
||
) as client:
|
||
while True:
|
||
try:
|
||
pane = await asyncio.to_thread(capture_hermes_pane, bridge_dir)
|
||
prompt = parse_hermes_approval_prompt(pane) if pane else None
|
||
if prompt is not None:
|
||
# Rising edge only: ONE card per visible-prompt episode (do not
|
||
# re-mint while the prompt stays up).
|
||
if active is None:
|
||
episode += 1
|
||
elicitation_id = hermes_permission_elicitation_id(session_id, str(episode))
|
||
task = asyncio.create_task(
|
||
_run_one_approval(
|
||
client,
|
||
session_id=session_id,
|
||
bridge_dir=bridge_dir,
|
||
prompt=prompt,
|
||
elicitation_id=elicitation_id,
|
||
),
|
||
name=f"hermes-approval-{episode}",
|
||
)
|
||
active = {"elicitation_id": elicitation_id, "task": task}
|
||
elif active is not None:
|
||
# Falling edge: prompt vanished. Release the card if still
|
||
# parked (answered in the TUI); no-op if answered via the web.
|
||
task = active["task"]
|
||
if isinstance(task, asyncio.Task) and not task.done():
|
||
await _post_external_elicitation_resolved(
|
||
client, session_id, str(active["elicitation_id"])
|
||
)
|
||
active = None
|
||
except asyncio.CancelledError:
|
||
raise
|
||
except Exception:
|
||
_logger.exception(
|
||
"hermes approval mirror poll failed; session=%s bridge_dir=%s",
|
||
session_id,
|
||
bridge_dir,
|
||
)
|
||
await asyncio.sleep(poll_interval_s)
|
||
|
||
|
||
async def _run_one_approval(
|
||
client: httpx.AsyncClient,
|
||
*,
|
||
session_id: str,
|
||
bridge_dir: Path,
|
||
prompt: HermesApprovalPrompt,
|
||
elicitation_id: str,
|
||
) -> None:
|
||
"""Park one Hermes prompt on the server and send the verdict keystroke."""
|
||
payload = {
|
||
"elicitation_id": elicitation_id,
|
||
"agent": "Hermes",
|
||
"policy_name": "hermes_native_permission",
|
||
"operation_type": "shell",
|
||
"message": prompt.message,
|
||
"content_preview": prompt.preview,
|
||
}
|
||
try:
|
||
response = await client.post(
|
||
f"/v1/sessions/{session_id}/hooks/native-permission-request",
|
||
json=payload,
|
||
)
|
||
except httpx.HTTPError:
|
||
_logger.exception("hermes permission hook POST failed; session=%s", session_id)
|
||
return
|
||
if response.status_code >= 400:
|
||
_logger.warning(
|
||
"hermes permission hook rejected: status=%s body=%s",
|
||
response.status_code,
|
||
response.text[:512],
|
||
)
|
||
return
|
||
if not response.content:
|
||
# Empty 2xx → resolved elsewhere (TUI answered) or timeout: no keystroke.
|
||
return
|
||
try:
|
||
result = response.json()
|
||
except ValueError:
|
||
_logger.warning("hermes permission hook returned non-JSON: %s", response.text[:512])
|
||
return
|
||
action = result.get("action") if isinstance(result, dict) else None
|
||
key = None
|
||
if action == "accept":
|
||
key = prompt.accept_key
|
||
elif action in {"decline", "cancel"}:
|
||
key = prompt.decline_key
|
||
if key is None:
|
||
return
|
||
try:
|
||
await asyncio.to_thread(send_hermes_pane_keys, bridge_dir, key)
|
||
except RuntimeError:
|
||
_logger.exception(
|
||
"failed to send hermes approval keystroke %r; session=%s", key, session_id
|
||
)
|
||
|
||
|
||
async def _post_external_elicitation_resolved(
|
||
client: httpx.AsyncClient, session_id: str, elicitation_id: str
|
||
) -> None:
|
||
"""Tell the server the native TUI answered a pending Hermes prompt."""
|
||
try:
|
||
response = await client.post(
|
||
f"/v1/sessions/{session_id}/events",
|
||
json={
|
||
"type": "external_elicitation_resolved",
|
||
"data": {"elicitation_id": elicitation_id},
|
||
},
|
||
timeout=10.0,
|
||
)
|
||
if response.status_code >= 400:
|
||
_logger.warning(
|
||
"hermes external_elicitation_resolved rejected: status=%s body=%s",
|
||
response.status_code,
|
||
response.text[:512],
|
||
)
|
||
except httpx.HTTPError:
|
||
_logger.exception("hermes external_elicitation_resolved POST failed")
|