common/hermes
2
0
Fork 0
Code Issues Pull Requests Actions 1 Packages Projects Releases Wiki Activity

feat(gateway): live-stream /update output + interactive prompt buttons (#5180)

Browse Source 
* feat(gateway): live-stream /update output + forward interactive prompts

Adds real-time output streaming and interactive prompt forwarding for
the gateway /update command, so users on Telegram/Discord/etc see the
full update progress and can respond to prompts (stash restore, config
migration) without needing terminal access.

Changes:

hermes_cli/main.py:
- Add --gateway flag to 'hermes update' argparse
- Add _gateway_prompt() file-based IPC function that writes
  .update_prompt.json and polls for .update_response
- Modify _restore_stashed_changes() to accept optional input_fn
  parameter for gateway mode prompt forwarding
- cmd_update() uses _gateway_prompt when --gateway is set, enabling
  interactive stash restore and config migration prompts

gateway/run.py:
- _handle_update_command: spawn with --gateway flag and
  PYTHONUNBUFFERED=1 for real-time output flushing
- Store session_key in .update_pending.json for cross-restart
  session matching
- Add _update_prompt_pending dict to track sessions awaiting
  update prompt responses
- Replace _watch_for_update_completion with _watch_update_progress:
  streams output chunks every ~4s, detects .update_prompt.json and
  forwards prompts to the user, handles completion/failure/timeout
- Add update prompt interception in _handle_message: when a prompt
  is pending, the user's next message is written to .update_response
  instead of being processed normally
- Preserve _send_update_notification as legacy fallback for
  post-restart cases where adapter isn't available yet

File-based IPC protocol:
- .update_prompt.json: written by update process with prompt text,
  default value, and unique ID
- .update_response: written by gateway with user's answer
- .update_output.txt: existing, now streamed in real-time
- .update_exit_code: existing completion marker

Tests: 16 new tests covering _gateway_prompt IPC, output streaming,
prompt detection/forwarding, message interception, and cleanup.

* feat: interactive buttons for update prompts (Telegram + Discord)

Telegram: Inline keyboard with ✓ Yes / ✗ No buttons. Clicking a button
answers the callback query, edits the message to show the choice, and
writes .update_response directly. CallbackQueryHandler registered on
the update_prompt: prefix.

Discord: UpdatePromptView (discord.ui.View) with green Yes / red No
buttons. Follows the ExecApprovalView pattern — auth check, embed color
update, disabled-after-click. Writes .update_response on click.

All platforms: /approve and /deny (and /yes, /no) now work as shorthand
for yes/no when an update prompt is pending. The text fallback message
instructs users to use these commands. Raw message interception still
works as a fallback for non-command responses.

Gateway watcher checks adapter for send_update_prompt method (class-level
check to avoid MagicMock false positives) and falls back to text prompt
with /approve instructions when unavailable.

* fix: block /update on non-messaging platforms (API, webhooks, ACP)

Add _UPDATE_ALLOWED_PLATFORMS frozenset that explicitly lists messaging
platforms where /update is permitted. API server, webhook, and ACP
platforms get a clear error directing them to run hermes update from
the terminal instead.

ACP and API server already don't reach _handle_message (separate
codepaths), and webhooks have distinct session keys that can't collide
with messaging sessions. This guard is belt-and-suspenders.
This commit is contained in:
Teknium
2026-04-05 00:28:58 -07:00
committed by GitHub
parent 441ec48802
commit 0c54da8aaf
6 changed files with 996 additions and 18 deletions
Download Patch File Download Diff File Expand all files Collapse all files

110
gateway/platforms/discord.py
View File

@@ -1932,6 +1932,37 @@ class DiscordAdapter(BasePlatformAdapter):
except Exception as e:
return SendResult(success=False, error=str(e))
async def send_update_prompt(
self, chat_id: str, prompt: str, default: str = "",
session_key: str = "",
) -> SendResult:
"""Send an interactive button-based update prompt (Yes / No).
Used by the gateway ``/update`` watcher when ``hermes update --gateway``
needs user input (stash restore, config migration).
"""
if not self._client or not DISCORD_AVAILABLE:
return SendResult(success=False, error="Not connected")
try:
channel = self._client.get_channel(int(chat_id))
if not channel:
channel = await self._client.fetch_channel(int(chat_id))
default_hint = f" (default: {default})" if default else ""
embed = discord.Embed(
title="⚕ Update Needs Your Input",
description=f"{prompt}{default_hint}",
color=discord.Color.gold(),
)
view = UpdatePromptView(
session_key=session_key,
allowed_user_ids=self._allowed_user_ids,
)
msg = await channel.send(embed=embed, view=view)
return SendResult(success=True, message_id=str(msg.id))
except Exception as e:
return SendResult(success=False, error=str(e))
def _get_parent_channel_id(self, channel: Any) -> Optional[str]:
"""Return the parent channel ID for a Discord thread-like channel, if present."""
parent = getattr(channel, "parent", None)
@@ -2344,3 +2375,82 @@ if DISCORD_AVAILABLE:
self.resolved = True
for child in self.children:
child.disabled = True
class UpdatePromptView(discord.ui.View):
"""Interactive Yes/No buttons for ``hermes update`` prompts.
Clicking a button writes the answer to ``.update_response`` so the
detached update process can pick it up. Only authorized users can
click. Times out after 5 minutes (the update process also has a
5-minute timeout on its side).
"""
def __init__(self, session_key: str, allowed_user_ids: set):
super().__init__(timeout=300)
self.session_key = session_key
self.allowed_user_ids = allowed_user_ids
self.resolved = False
def _check_auth(self, interaction: discord.Interaction) -> bool:
if not self.allowed_user_ids:
return True
return str(interaction.user.id) in self.allowed_user_ids
async def _respond(
self, interaction: discord.Interaction, answer: str,
color: discord.Color, label: str,
):
if self.resolved:
await interaction.response.send_message(
"Already answered~", ephemeral=True
)
return
if not self._check_auth(interaction):
await interaction.response.send_message(
"You're not authorized~", ephemeral=True
)
return
self.resolved = True
# Update embed
embed = interaction.message.embeds[0] if interaction.message.embeds else None
if embed:
embed.color = color
embed.set_footer(text=f"{label} by {interaction.user.display_name}")
for child in self.children:
child.disabled = True
await interaction.response.edit_message(embed=embed, view=self)
# Write response file
try:
from hermes_constants import get_hermes_home
home = get_hermes_home()
response_path = home / ".update_response"
tmp = response_path.with_suffix(".tmp")
tmp.write_text(answer)
tmp.replace(response_path)
logger.info(
"Discord update prompt answered '%s' by %s",
answer, interaction.user.display_name,
)
except Exception as exc:
logger.error("Failed to write update response: %s", exc)
@discord.ui.button(label="Yes", style=discord.ButtonStyle.green, emoji="✓")
async def yes_btn(
self, interaction: discord.Interaction, button: discord.ui.Button
):
await self._respond(interaction, "y", discord.Color.green(), "Yes")
@discord.ui.button(label="No", style=discord.ButtonStyle.red, emoji="✗")
async def no_btn(
self, interaction: discord.Interaction, button: discord.ui.Button
):
await self._respond(interaction, "n", discord.Color.red(), "No")
async def on_timeout(self):
self.resolved = True
for child in self.children:
child.disabled = True

74
gateway/platforms/telegram.py
View File

@@ -17,10 +17,11 @@ from typing import Dict, List, Optional, Any
logger = logging.getLogger(__name__)
try:
from telegram import Update, Bot, Message
from telegram import Update, Bot, Message, InlineKeyboardButton, InlineKeyboardMarkup
from telegram.ext import (
Application,
CommandHandler,
CallbackQueryHandler,
MessageHandler as TelegramMessageHandler,
ContextTypes,
filters,
@@ -33,8 +34,11 @@ except ImportError:
Update = Any
Bot = Any
Message = Any
InlineKeyboardButton = Any
InlineKeyboardMarkup = Any
Application = Any
CommandHandler = Any
CallbackQueryHandler = Any
TelegramMessageHandler = Any
HTTPXRequest = Any
filters = None
@@ -543,6 +547,8 @@ class TelegramAdapter(BasePlatformAdapter):
filters.PHOTO | filters.VIDEO | filters.AUDIO | filters.VOICE | filters.Document.ALL | filters.Sticker.ALL,
self._handle_media_message
))
# Handle inline keyboard button callbacks (update prompts)
self._app.add_handler(CallbackQueryHandler(self._handle_callback_query))
# Start polling — retry initialize() for transient TLS resets
try:
@@ -950,6 +956,72 @@ class TelegramAdapter(BasePlatformAdapter):
)
return SendResult(success=False, error=str(e))
async def send_update_prompt(
self, chat_id: str, prompt: str, default: str = "",
session_key: str = "",
) -> SendResult:
"""Send an inline-keyboard update prompt (Yes / No buttons).
Used by the gateway ``/update`` watcher when ``hermes update --gateway``
needs user input (stash restore, config migration).
"""
if not self._bot:
return SendResult(success=False, error="Not connected")
try:
default_hint = f" (default: {default})" if default else ""
text = f"⚕ *Update needs your input:*\n\n{prompt}{default_hint}"
keyboard = InlineKeyboardMarkup([
[
InlineKeyboardButton("✓ Yes", callback_data="update_prompt:y"),
InlineKeyboardButton("✗ No", callback_data="update_prompt:n"),
]
])
msg = await self._bot.send_message(
chat_id=int(chat_id),
text=text,
parse_mode=ParseMode.MARKDOWN,
reply_markup=keyboard,
)
return SendResult(success=True, message_id=str(msg.message_id))
except Exception as e:
logger.warning("[%s] send_update_prompt failed: %s", self.name, e)
return SendResult(success=False, error=str(e))
async def _handle_callback_query(
self, update: "Update", context: "ContextTypes.DEFAULT_TYPE"
) -> None:
"""Handle inline keyboard button clicks (update prompts)."""
query = update.callback_query
if not query or not query.data:
return
data = query.data
if not data.startswith("update_prompt:"):
return
answer = data.split(":", 1)[1] # "y" or "n"
await query.answer(text=f"Sent '{answer}' to the update process.")
# Edit the message to show the choice and remove buttons
label = "Yes" if answer == "y" else "No"
try:
await query.edit_message_text(
text=f"⚕ Update prompt answered: *{label}*",
parse_mode=ParseMode.MARKDOWN,
reply_markup=None,
)
except Exception:
pass # non-fatal if edit fails
# Write the response file
try:
from hermes_constants import get_hermes_home
home = get_hermes_home()
response_path = home / ".update_response"
tmp = response_path.with_suffix(".tmp")
tmp.write_text(answer)
tmp.replace(response_path)
logger.info("Telegram update prompt answered '%s' by user %s",
answer, getattr(query.from_user, "id", "unknown"))
except Exception as exc:
logger.error("Failed to write update response from callback: %s", exc)
async def send_voice(
self,
chat_id: str,

249
gateway/run.py
View File

@@ -517,6 +517,10 @@ class GatewayRunner:
# Key: Platform enum, Value: {"config": platform_config, "attempts": int, "next_retry": float}
self._failed_platforms: Dict[Platform, Dict[str, Any]] = {}
# Track pending /update prompt responses per session.
# Key: session_key, Value: True when a prompt is waiting for user input.
self._update_prompt_pending: Dict[str, bool] = {}
# Persistent Honcho managers keyed by gateway session key.
# This preserves write_frequency="session" semantics across short-lived
# per-message AIAgent instances.
@@ -1737,6 +1741,35 @@ class GatewayRunner:
self.pairing_store._record_rate_limit(platform_name, source.user_id)
return None
# Intercept messages that are responses to a pending /update prompt.
# The update process (detached) wrote .update_prompt.json; the watcher
# forwarded it to the user; now the user's reply goes back via
# .update_response so the update process can continue.
_quick_key = self._session_key_for_source(source)
_update_prompts = getattr(self, "_update_prompt_pending", {})
if _update_prompts.get(_quick_key):
raw = (event.text or "").strip()
# Accept /approve and /deny as shorthand for yes/no
cmd = event.get_command()
if cmd in ("approve", "yes"):
response_text = "y"
elif cmd in ("deny", "no"):
response_text = "n"
else:
response_text = raw
if response_text:
response_path = _hermes_home / ".update_response"
try:
tmp = response_path.with_suffix(".tmp")
tmp.write_text(response_text)
tmp.replace(response_path)
except OSError as e:
logger.warning("Failed to write update response: %s", e)
return f"✗ Failed to send response to update process: {e}"
_update_prompts.pop(_quick_key, None)
label = response_text if len(response_text) <= 20 else response_text[:20] + "…"
return f"✓ Sent `{label}` to the update process."
# PRIORITY handling when an agent is already running for this session.
# Default behavior is to interrupt immediately so user text/stop messages
# are handled with minimal latency.
@@ -1744,7 +1777,6 @@ class GatewayRunner:
# Special case: Telegram/photo bursts often arrive as multiple near-
# simultaneous updates. Do NOT interrupt for photo-only follow-ups here;
# let the adapter-level batching/queueing logic absorb them.
_quick_key = self._session_key_for_source(source)
# Staleness eviction: if an entry has been in _running_agents for
# longer than the agent timeout, it's a leaked lock from a hung or
@@ -4929,6 +4961,15 @@ class GatewayRunner:
logger.info("User denied %d dangerous command(s) via /deny", count)
return f"❌ Command{'s' if count > 1 else ''} denied{count_msg}."
# Platforms where /update is allowed. ACP, API server, and webhooks are
# programmatic interfaces that should not trigger system updates.
_UPDATE_ALLOWED_PLATFORMS = frozenset({
Platform.TELEGRAM, Platform.DISCORD, Platform.SLACK, Platform.WHATSAPP,
Platform.SIGNAL, Platform.MATTERMOST, Platform.MATRIX,
Platform.HOMEASSISTANT, Platform.EMAIL, Platform.SMS, Platform.DINGTALK,
Platform.FEISHU, Platform.WECOM, Platform.LOCAL,
})
async def _handle_update_command(self, event: MessageEvent) -> str:
"""Handle /update command — update Hermes Agent to the latest version.
@@ -4943,6 +4984,11 @@ class GatewayRunner:
from datetime import datetime
from hermes_cli.config import is_managed, format_managed_message
# Block non-messaging platforms (API server, webhooks, ACP)
platform = event.source.platform
if platform not in self._UPDATE_ALLOWED_PLATFORMS:
return "✗ /update is only available from messaging platforms. Run `hermes update` from the terminal."
if is_managed():
return f"✗ {format_managed_message('update Hermes Agent')}"
@@ -4964,10 +5010,12 @@ class GatewayRunner:
pending_path = _hermes_home / ".update_pending.json"
output_path = _hermes_home / ".update_output.txt"
exit_code_path = _hermes_home / ".update_exit_code"
session_key = self._session_key_for_source(event.source)
pending = {
"platform": event.source.platform.value,
"chat_id": event.source.chat_id,
"user_id": event.source.user_id,
"session_key": session_key,
"timestamp": datetime.now().isoformat(),
}
_tmp_pending = pending_path.with_suffix(".tmp")
@@ -4975,12 +5023,18 @@ class GatewayRunner:
_tmp_pending.replace(pending_path)
exit_code_path.unlink(missing_ok=True)
# Spawn `hermes update` detached so it survives gateway restart.
# Spawn `hermes update --gateway` detached so it survives gateway restart.
# --gateway enables file-based IPC for interactive prompts (stash
# restore, config migration) so the gateway can forward them to the
# user instead of silently skipping them.
# Use setsid for portable session detach (works under system services
# where systemd-run --user fails due to missing D-Bus session).
# PYTHONUNBUFFERED ensures output is flushed line-by-line so the
# gateway can stream it to the messenger in near-real-time.
hermes_cmd_str = " ".join(shlex.quote(part) for part in hermes_cmd)
update_cmd = (
f"{hermes_cmd_str} update > {shlex.quote(str(output_path))} 2>&1; "
f"PYTHONUNBUFFERED=1 {hermes_cmd_str} update --gateway"
f" > {shlex.quote(str(output_path))} 2>&1; "
f"status=$?; printf '%s' \"$status\" > {shlex.quote(str(exit_code_path))}"
)
try:
@@ -5007,7 +5061,7 @@ class GatewayRunner:
return f"✗ Failed to start update: {e}"
self._schedule_update_notification_watch()
return "⚕ Starting Hermes update… I'll notify you when it's done."
return "⚕ Starting Hermes update… I'll stream progress here."
def _schedule_update_notification_watch(self) -> None:
"""Ensure a background task is watching for update completion."""
@@ -5017,39 +5071,210 @@ class GatewayRunner:
try:
self._update_notification_task = asyncio.create_task(
self._watch_for_update_completion()
self._watch_update_progress()
)
except RuntimeError:
logger.debug("Skipping update notification watcher: no running event loop")
async def _watch_for_update_completion(
async def _watch_update_progress(
self,
poll_interval: float = 2.0,
stream_interval: float = 4.0,
timeout: float = 1800.0,
) -> None:
"""Wait for ``hermes update`` to finish, then send its notification."""
"""Watch ``hermes update --gateway``, streaming output + forwarding prompts.
Polls ``.update_output.txt`` for new content and sends chunks to the
user periodically. Detects ``.update_prompt.json`` (written by the
update process when it needs user input) and forwards the prompt to
the messenger. The user's next message is intercepted by
``_handle_message`` and written to ``.update_response``.
"""
import json
import re as _re
pending_path = _hermes_home / ".update_pending.json"
claimed_path = _hermes_home / ".update_pending.claimed.json"
output_path = _hermes_home / ".update_output.txt"
exit_code_path = _hermes_home / ".update_exit_code"
prompt_path = _hermes_home / ".update_prompt.json"
loop = asyncio.get_running_loop()
deadline = loop.time() + timeout
while (pending_path.exists() or claimed_path.exists()) and loop.time() < deadline:
if exit_code_path.exists():
# Resolve the adapter and chat_id for sending messages
adapter = None
chat_id = None
session_key = None
for path in (claimed_path, pending_path):
if path.exists():
try:
pending = json.loads(path.read_text())
platform_str = pending.get("platform")
chat_id = pending.get("chat_id")
session_key = pending.get("session_key")
if platform_str and chat_id:
platform = Platform(platform_str)
adapter = self.adapters.get(platform)
# Fallback session key if not stored (old pending files)
if not session_key:
session_key = f"{platform_str}:{chat_id}"
break
except Exception:
pass
if not adapter or not chat_id:
logger.warning("Update watcher: cannot resolve adapter/chat_id, falling back to completion-only")
# Fall back to old behavior: wait for exit code and send final notification
while (pending_path.exists() or claimed_path.exists()) and loop.time() < deadline:
if exit_code_path.exists():
await self._send_update_notification()
return
await asyncio.sleep(poll_interval)
if (pending_path.exists() or claimed_path.exists()) and not exit_code_path.exists():
exit_code_path.write_text("124")
await self._send_update_notification()
return
def _strip_ansi(text: str) -> str:
return _re.sub(r'\x1b\[[0-9;]*[A-Za-z]', '', text)
bytes_sent = 0
last_stream_time = loop.time()
buffer = ""
async def _flush_buffer() -> None:
"""Send buffered output to the user."""
nonlocal buffer, last_stream_time
if not buffer.strip():
buffer = ""
return
# Chunk to fit message limits (Telegram: 4096, others: generous)
clean = _strip_ansi(buffer).strip()
buffer = ""
last_stream_time = loop.time()
if not clean:
return
# Split into chunks if too long
max_chunk = 3500
chunks = [clean[i:i + max_chunk] for i in range(0, len(clean), max_chunk)]
for chunk in chunks:
try:
await adapter.send(chat_id, f"```\n{chunk}\n```")
except Exception as e:
logger.debug("Update stream send failed: %s", e)
while loop.time() < deadline:
# Check for completion
if exit_code_path.exists():
# Read any remaining output
if output_path.exists():
try:
content = output_path.read_text()
if len(content) > bytes_sent:
buffer += content[bytes_sent:]
bytes_sent = len(content)
except OSError:
pass
await _flush_buffer()
# Send final status
try:
exit_code_raw = exit_code_path.read_text().strip() or "1"
exit_code = int(exit_code_raw)
if exit_code == 0:
await adapter.send(chat_id, "✅ Hermes update finished.")
else:
await adapter.send(chat_id, "❌ Hermes update failed (exit code {}).".format(exit_code))
logger.info("Update finished (exit=%s), notified %s", exit_code, session_key)
except Exception as e:
logger.warning("Update final notification failed: %s", e)
# Cleanup
for p in (pending_path, claimed_path, output_path,
exit_code_path, prompt_path):
p.unlink(missing_ok=True)
(_hermes_home / ".update_response").unlink(missing_ok=True)
self._update_prompt_pending.pop(session_key, None)
return
# Check for new output
if output_path.exists():
try:
content = output_path.read_text()
if len(content) > bytes_sent:
buffer += content[bytes_sent:]
bytes_sent = len(content)
except OSError:
pass
# Flush buffer periodically
if buffer.strip() and (loop.time() - last_stream_time) >= stream_interval:
await _flush_buffer()
# Check for prompts
if prompt_path.exists() and session_key:
try:
prompt_data = json.loads(prompt_path.read_text())
prompt_text = prompt_data.get("prompt", "")
default = prompt_data.get("default", "")
if prompt_text:
# Flush any buffered output first so the user sees
# context before the prompt
await _flush_buffer()
# Try platform-native buttons first (Discord, Telegram)
sent_buttons = False
if getattr(type(adapter), "send_update_prompt", None) is not None:
try:
await adapter.send_update_prompt(
chat_id=chat_id,
prompt=prompt_text,
default=default,
session_key=session_key,
)
sent_buttons = True
except Exception as btn_err:
logger.debug("Button-based update prompt failed: %s", btn_err)
if not sent_buttons:
default_hint = f" (default: {default})" if default else ""
await adapter.send(
chat_id,
f"⚕ **Update needs your input:**\n\n"
f"{prompt_text}{default_hint}\n\n"
f"Reply `/approve` (yes) or `/deny` (no), "
f"or type your answer directly."
)
self._update_prompt_pending[session_key] = True
logger.info("Forwarded update prompt to %s: %s", session_key, prompt_text[:80])
except (json.JSONDecodeError, OSError) as e:
logger.debug("Failed to read update prompt: %s", e)
await asyncio.sleep(poll_interval)
if (pending_path.exists() or claimed_path.exists()) and not exit_code_path.exists():
logger.warning("Update watcher timed out waiting for completion marker")
# Timeout
if not exit_code_path.exists():
logger.warning("Update watcher timed out after %.0fs", timeout)
exit_code_path.write_text("124")
await self._send_update_notification()
await _flush_buffer()
try:
await adapter.send(chat_id, "❌ Hermes update timed out after 30 minutes.")
except Exception:
pass
for p in (pending_path, claimed_path, output_path,
exit_code_path, prompt_path):
p.unlink(missing_ok=True)
(_hermes_home / ".update_response").unlink(missing_ok=True)
self._update_prompt_pending.pop(session_key, None)
async def _send_update_notification(self) -> bool:
"""If an update finished, notify the user.
Returns False when the update is still running so a caller can retry
later. Returns True after a definitive send/skip decision.
This is the legacy notification path used when the streaming watcher
cannot resolve the adapter (e.g. after a gateway restart where the
platform hasn't reconnected yet).
"""
import json
import re as _re