Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
181 changes: 170 additions & 11 deletions gateway/platforms/base.py
Original file line number Diff line number Diff line change
Expand Up @@ -416,7 +416,7 @@ def is_host_excluded_by_no_proxy(hostname: str, no_proxy_value: str | None = Non
from dataclasses import dataclass, field
from datetime import datetime
from pathlib import Path
from typing import Dict, List, Optional, Any, Callable, Awaitable, Tuple
from typing import Dict, List, Optional, Any, Callable, Awaitable, Tuple, Union
from enum import Enum

from pathlib import Path as _Path
Expand Down Expand Up @@ -981,7 +981,7 @@ def coerce_plaintext_gateway_command(event: "MessageEvent") -> None:
return


@dataclass
@dataclass
class SendResult:
"""Result of sending a message."""
success: bool
Expand All @@ -991,6 +991,45 @@ class SendResult:
retryable: bool = False # True for transient connection errors — base will retry automatically


class EphemeralReply(str):
"""System-notice reply that auto-deletes after a TTL.

Slash-command handlers in ``gateway/run.py`` can return this wrapper
instead of a plain string to request that the reply message be deleted
after ``ttl_seconds`` on platforms that support ``delete_message``.

Subclassing ``str`` keeps the wrapper transparent to anything that
treats handler return values as text (existing tests use ``in`` /
``startswith`` / equality; the ``_process_message_background`` pipeline
extracts attachments from the string content). ``isinstance(r,
EphemeralReply)`` still distinguishes ephemeral replies from plain
strings so the send path can schedule deletion.

Platforms that don't override :meth:`BasePlatformAdapter.delete_message`
silently ignore the TTL — the message is sent normally and left in
place. When ``ttl_seconds`` is ``None``, the pipeline uses the
configured ``display.ephemeral_system_ttl`` default. A default of ``0``
disables auto-deletion globally, preserving prior behavior.
"""

ttl_seconds: Optional[int]

def __new__(cls, text: str, ttl_seconds: Optional[int] = None):
instance = super().__new__(cls, text)
instance.ttl_seconds = ttl_seconds
return instance

@property
def text(self) -> str:
"""Return the underlying text.

Provided for call sites that want an explicit string conversion,
though ``str(reply)`` and using ``reply`` directly where a string
is expected both work identically.
"""
return str.__str__(self)


def merge_pending_message_event(
pending_messages: Dict[str, MessageEvent],
session_key: str,
Expand Down Expand Up @@ -1073,8 +1112,10 @@ def merge_pending_message_event(
)


# Type for message handlers
MessageHandler = Callable[[MessageEvent], Awaitable[Optional[str]]]
# Type for message handlers. Handlers may return a plain string (normal
# reply), an ``EphemeralReply`` to opt the reply into auto-deletion, or
# ``None`` when the response was already delivered (e.g. via streaming).
MessageHandler = Callable[[MessageEvent], Awaitable[Optional[Union[str, "EphemeralReply"]]]]


def resolve_channel_prompt(
Expand Down Expand Up @@ -1459,6 +1500,64 @@ async def delete_message(
"""
return False

def _get_ephemeral_system_ttl_default(self) -> int:
"""Read ``display.ephemeral_system_ttl`` from config.

Returns the TTL in seconds to use when an :class:`EphemeralReply`
does not specify one explicitly. ``0`` (the default) disables
auto-deletion. Non-fatal if config is unreadable.
"""
try:
from hermes_cli.config import load_config as _load_config
except Exception:
return 0
try:
cfg = _load_config()
except Exception:
return 0
display = cfg.get("display", {}) if isinstance(cfg, dict) else {}
if not isinstance(display, dict):
return 0
raw = display.get("ephemeral_system_ttl", 0)
try:
return int(raw)
except (TypeError, ValueError):
return 0

def _schedule_ephemeral_delete(
self,
chat_id: str,
message_id: str,
ttl_seconds: int,
) -> None:
"""Spawn a detached task that deletes ``message_id`` after ``ttl_seconds``.

Best-effort — failures (gateway restart, permission denied, message
too old for Telegram's 48h window) are swallowed at debug level.
Does not block the caller.
"""

async def _run_delete() -> None:
try:
await asyncio.sleep(max(1, int(ttl_seconds)))
await self.delete_message(chat_id=chat_id, message_id=message_id)
except asyncio.CancelledError:
raise
except Exception as e:
logger.debug(
"[%s] Ephemeral delete failed for %s/%s: %s",
self.name, chat_id, message_id, e,
)

coro = _run_delete()
try:
asyncio.create_task(coro)
except RuntimeError:
# No running loop (e.g. unit tests that never reach the async
# path). Close the coroutine cleanly so Python doesn't warn
# about it never being awaited, then drop silently.
coro.close()

async def send_slash_confirm(
self,
chat_id: str,
Expand Down Expand Up @@ -2048,6 +2147,28 @@ def _is_timeout_error(error: Optional[str]) -> bool:
lowered = error.lower()
return "timed out" in lowered or "readtimeout" in lowered or "writetimeout" in lowered

def _unwrap_ephemeral(self, response: Any) -> Tuple[Optional[str], int]:
"""Unwrap a handler response into (text, ttl_seconds).

Accepts a plain string, ``None``, or an :class:`EphemeralReply`.
Returns ``(text, ttl)`` where ``ttl > 0`` means the caller should
schedule a deletion via :meth:`_schedule_ephemeral_delete` after
the send succeeds. ``ttl`` is forced to 0 when the adapter
doesn't override :meth:`delete_message` so non-supporting
platforms silently degrade to normal sends.
"""
if isinstance(response, EphemeralReply):
ttl = response.ttl_seconds
if ttl is None:
try:
ttl = int(self._get_ephemeral_system_ttl_default())
except Exception:
ttl = 0
if ttl and ttl > 0 and type(self).delete_message is BasePlatformAdapter.delete_message:
ttl = 0
return response.text, int(ttl or 0)
return response, 0

async def _send_with_retry(
self,
chat_id: str,
Expand Down Expand Up @@ -2355,13 +2476,20 @@ async def _dispatch_active_session_command(
release_guard=False,
discard_pending=False,
)
if response:
await self._send_with_retry(
_text, _eph_ttl = self._unwrap_ephemeral(response)
if _text:
_r = await self._send_with_retry(
chat_id=event.source.chat_id,
content=response,
content=_text,
reply_to=event.message_id,
metadata=thread_meta,
)
if _eph_ttl > 0 and _r.success and _r.message_id:
self._schedule_ephemeral_delete(
chat_id=event.source.chat_id,
message_id=_r.message_id,
ttl_seconds=_eph_ttl,
)
except Exception:
# On failure, restore the original guard if one still exists so
# we don't leave the session in a half-reset state.
Expand Down Expand Up @@ -2441,13 +2569,20 @@ async def handle_message(self, event: MessageEvent) -> None:
try:
_thread_meta = {"thread_id": event.source.thread_id} if event.source.thread_id else None
response = await self._message_handler(event)
if response:
await self._send_with_retry(
_text, _eph_ttl = self._unwrap_ephemeral(response)
if _text:
_r = await self._send_with_retry(
chat_id=event.source.chat_id,
content=response,
content=_text,
reply_to=event.message_id,
metadata=_thread_meta,
)
if _eph_ttl > 0 and _r.success and _r.message_id:
self._schedule_ephemeral_delete(
chat_id=event.source.chat_id,
message_id=_r.message_id,
ttl_seconds=_eph_ttl,
)
except Exception as e:
logger.error("[%s] Command '/%s' dispatch failed: %s", self.name, cmd, e, exc_info=True)
return
Expand Down Expand Up @@ -2553,7 +2688,16 @@ async def _stop_typing_task() -> None:

# Call the handler (this can take a while with tool calls)
response = await self._message_handler(event)


# Slash-command handlers may return an EphemeralReply sentinel to
# request that their reply message auto-delete after a TTL (used
# for system notices like "✨ New session started!" that the user
# doesn't need to keep in the thread). Unwrap here so all the
# downstream extract_media / text-processing logic sees a plain
# string, and remember the TTL + platform capability so the
# post-send block can schedule the deletion.
response, _ephemeral_ttl = self._unwrap_ephemeral(response)

# Send response if any. A None/empty response is normal when
# streaming already delivered the text (already_sent=True) or
# when the message was queued behind an active agent. Log at
Expand Down Expand Up @@ -2642,6 +2786,21 @@ async def _stop_typing_task() -> None:
)
_record_delivery(result)

# Schedule auto-deletion of system-notice replies.
# Detached so the handler returns immediately; errors
# (permission denied, message too old) are swallowed.
if (
_ephemeral_ttl
and _ephemeral_ttl > 0
and result.success
and result.message_id
):
self._schedule_ephemeral_delete(
chat_id=event.source.chat_id,
message_id=result.message_id,
ttl_seconds=_ephemeral_ttl,
)

# Human-like pacing delay between text and media
human_delay = self._get_human_delay()

Expand Down
31 changes: 16 additions & 15 deletions gateway/run.py
Original file line number Diff line number Diff line change
Expand Up @@ -29,7 +29,7 @@
from contextvars import copy_context
from pathlib import Path
from datetime import datetime
from typing import Dict, Optional, Any, List
from typing import Dict, Optional, Any, List, Union

# account_usage imports the OpenAI SDK chain (~230 ms). Only needed by
# /usage; we still import it at module top in the gateway because test
Expand Down Expand Up @@ -454,6 +454,7 @@ def _ensure_ssl_certs() -> None:
from gateway.delivery import DeliveryRouter
from gateway.platforms.base import (
BasePlatformAdapter,
EphemeralReply,
MessageEvent,
MessageType,
merge_pending_message_event,
Expand Down Expand Up @@ -4472,7 +4473,7 @@ async def _handle_message(self, event: MessageEvent) -> Optional[str]:
invalidation_reason="stop_command",
)
logger.info("STOP for session %s — agent interrupted, session lock released", _quick_key)
return "⚡ Stopped. You can continue this session."
return EphemeralReply("⚡ Stopped. You can continue this session.")

# /reset and /new must bypass the running-agent guard so they
# actually dispatch as commands instead of being queued as user
Expand Down Expand Up @@ -4677,7 +4678,7 @@ async def _handle_message(self, event: MessageEvent) -> Optional[str]:
# Force-clean the sentinel so the session is unlocked.
self._release_running_agent_state(_quick_key)
logger.info("HARD STOP (pending) for session %s — sentinel cleared", _quick_key)
return "⚡ Force-stopped. The agent was still starting — session unlocked."
return EphemeralReply("⚡ Force-stopped. The agent was still starting — session unlocked.")
# Queue the message so it will be picked up after the
# agent starts.
adapter = self.adapters.get(source.platform)
Expand Down Expand Up @@ -6353,7 +6354,7 @@ def _format_session_info(self) -> str:

return "\n".join(lines)

async def _handle_reset_command(self, event: MessageEvent) -> str:
async def _handle_reset_command(self, event: MessageEvent) -> Union[str, EphemeralReply]:
"""Handle /new or /reset command."""
source = event.source

Expand Down Expand Up @@ -6464,8 +6465,8 @@ async def _handle_reset_command(self, event: MessageEvent) -> str:
_tip_line = ""

if session_info:
return f"{header}\n\n{session_info}{_tip_line}"
return f"{header}{_tip_line}"
return EphemeralReply(f"{header}\n\n{session_info}{_tip_line}")
return EphemeralReply(f"{header}{_tip_line}")

async def _handle_profile_command(self, event: MessageEvent) -> str:
"""Handle /profile — show active profile name and home directory."""
Expand Down Expand Up @@ -6713,7 +6714,7 @@ async def _handle_agents_command(self, event: MessageEvent) -> str:

return "\n".join(lines)

async def _handle_stop_command(self, event: MessageEvent) -> str:
async def _handle_stop_command(self, event: MessageEvent) -> Union[str, EphemeralReply]:
"""Handle /stop command - interrupt a running agent.

When an agent is truly hung (blocked thread that never checks
Expand All @@ -6738,7 +6739,7 @@ async def _handle_stop_command(self, event: MessageEvent) -> str:
invalidation_reason="stop_command_pending",
)
logger.info("STOP (pending) for session %s — sentinel cleared", session_key)
return "⚡ Stopped. The agent hadn't started yet — you can continue this session."
return EphemeralReply("⚡ Stopped. The agent hadn't started yet — you can continue this session.")
if agent:
# Force-clean the session lock so a truly hung agent doesn't
# keep it locked forever.
Expand All @@ -6748,11 +6749,11 @@ async def _handle_stop_command(self, event: MessageEvent) -> str:
interrupt_reason=_INTERRUPT_REASON_STOP,
invalidation_reason="stop_command_handler",
)
return "⚡ Stopped. You can continue this session."
return EphemeralReply("⚡ Stopped. You can continue this session.")
else:
return "No active task to stop."

async def _handle_restart_command(self, event: MessageEvent) -> str:
async def _handle_restart_command(self, event: MessageEvent) -> Union[str, EphemeralReply]:
"""Handle /restart command - drain active work, then restart the gateway."""
# Defensive idempotency check: if the previous gateway process
# recorded this same /restart (same platform + update_id) and the new
Expand All @@ -6778,7 +6779,7 @@ async def _handle_restart_command(self, event: MessageEvent) -> str:
count = self._running_agent_count()
if count:
return f"⏳ Draining {count} active agent(s) before restart..."
return "⏳ Gateway restart already in progress..."
return EphemeralReply("⏳ Gateway restart already in progress...")

# Save the requester's routing info so the new gateway process can
# notify them once it comes back online.
Expand Down Expand Up @@ -6830,7 +6831,7 @@ async def _handle_restart_command(self, event: MessageEvent) -> str:
self.request_restart(detached=True, via_service=False)
if active_agents:
return f"⏳ Draining {active_agents} active agent(s) before restart..."
return "♻ Restarting gateway. If you aren't notified within 60 seconds, restart from the console with `hermes gateway restart`."
return EphemeralReply("♻ Restarting gateway. If you aren't notified within 60 seconds, restart from the console with `hermes gateway restart`.")

def _is_stale_restart_redelivery(self, event: MessageEvent) -> bool:
"""Return True if this /restart is a Telegram re-delivery we already handled.
Expand Down Expand Up @@ -8321,7 +8322,7 @@ def _save_config_key(key_path: str, value):
return f"⚡ ✓ Priority Processing: **{label}** (saved to config)\n_(takes effect on next message)_"
return f"⚡ ✓ Priority Processing: **{label}** (this session only)"

async def _handle_yolo_command(self, event: MessageEvent) -> str:
async def _handle_yolo_command(self, event: MessageEvent) -> Union[str, EphemeralReply]:
"""Handle /yolo — toggle dangerous command approval bypass for this session only."""
from tools.approval import (
disable_session_yolo,
Expand All @@ -8333,10 +8334,10 @@ async def _handle_yolo_command(self, event: MessageEvent) -> str:
current = is_session_yolo_enabled(session_key)
if current:
disable_session_yolo(session_key)
return "⚠️ YOLO mode **OFF** for this session — dangerous commands will require approval."
return EphemeralReply("⚠️ YOLO mode **OFF** for this session — dangerous commands will require approval.")
else:
enable_session_yolo(session_key)
return "⚡ YOLO mode **ON** for this session — all commands auto-approved. Use with caution."
return EphemeralReply("⚡ YOLO mode **ON** for this session — all commands auto-approved. Use with caution.")

async def _handle_verbose_command(self, event: MessageEvent) -> str:
"""Handle /verbose command — cycle tool progress display mode.
Expand Down
Loading
Loading