1 · The Gateway Layer

gateway/: 99 files, including run.py, a single 1.55 MB file — the largest module in the whole project. This chapter is about how one always-on process lets 22 chat apps reach the same agent.

1.1 The Problem It Solves

Start with what happens without a gateway. Say you want the agent to support two platforms, Slack and Telegram:

The gateway is the layer that pulls all of that shared logic out. It is a background process that never shuts down, connected to every platform at once, and it is responsible for:

• Normalizing the wildly different message formats of each platform into a uniform event object
• Session routing — deciding which conversation a message belongs to
• User authorization
• Dispatching slash commands
• Driving the clock for scheduled tasks
• Managing the cache of agent instances (the same session reuses the same instance)
• Sending replies out according to each platform's rules

1.2 The Platform Adapter Abstract Base Class

gateway/platforms/base.py, 333 KB. This is the “wall-socket standard” — it specifies what any platform must provide in order to plug in.

The core idea: turn the differences into “capability query methods”

class BasePlatformAdapter(ABC):

    # —— message length limits ——
    def max_message_length_for_chat(self, chat_id: str) -> int
    def message_len_fn(self) -> Callable[[str], int]
    def message_len_fn_for_chat(self, chat_id: str) -> Callable[[str], int]

    # —— streaming capabilities ——
    def supports_draft_streaming(self, ...) -> bool
    def prefers_fresh_final_streaming(self, ...) -> bool
    def streaming_overflow_limit(self) -> Optional[int]
    async def send_draft(self, ...)

    # —— permission model ——
    def enforces_own_access_policy(self) -> bool
    def authorization_is_upstream(self) -> bool

    # —— rendering ——
    def render_message_event(self, event, sink) -> None
    def format_tool_event(self, event, *, mode: str = "all", ...) -> str
    def format_tool_preview(self, preview: "ToolPreview") -> str
    def set_status_text(self, chat_id: str, text: Optional[str]) -> None

hermes-agent/gateway/platforms/base.py

# Style A: branch on the platform in the main flow
if platform == 'slack':
    max_len = 40000
elif platform == 'discord':
    max_len = 2000
elif platform == 'telegram':
    max_len = 4096
elif platform == 'sms':
    max_len = 160
# ... 22 branches, and checks like this scattered across dozens of places

# Style B: ask the adapter
max_len = adapter.max_message_length_for_chat(chat_id)

How big are the differences being abstracted away?

1.3 Normalizing Message Events

Every platform has a completely different message format. The gateway unifies them into one event object:

class MessageType(Enum): ...
class ProcessingOutcome(Enum): ...

class MessageEvent:
    ...
    def is_command(self) -> bool                 # is this a slash command?
    def get_command(self) -> Optional[str]       # command name
    def get_command_args(self) -> str            # command arguments

class CachedMedia:
    def context_note(self) -> str                # context note for an attachment

class TextDebounceState: ...                     # ★ text debouncing
class SendResult: ...

class EphemeralReply(str):                       # ★ a temporary reply with an expiry
    def __new__(cls, text: str, ttl_seconds: Optional[int] = None): ...
    def text(self) -> str: ...

Two classes worth calling out

TextDebounceState handles this scenario:

EphemeralReply is a class that inherits from str and carries an expiry time. It is for messages that “should disappear once said” — status hints like “Thinking…”. On platforms that support it, these messages delete themselves after a while and don't clutter the chat history.

Inheriting from str is a very practical design: every piece of code that already handles strings keeps working unchanged; only the code that cares about expiry checks whether it is an EphemeralReply.

1.4 The Defensive Mechanisms in the Gateway Main Loop

The function names in run.py are, on their own, a checklist of “what goes wrong for a long-running process”:

① Hygiene cooldown: stop retrying failed compaction

def _hygiene_cooldown_for_failure(...)
def _reset_hygiene_failure_streak(gateway, session_key: str) -> None
def hygiene_compaction_recovered(...)
def _record_hygiene_cooldown(...)

“Hygiene” here means the automatic context tidying done in the background. If compaction keeps failing for a session, give it a cooldown period rather than trying again and again — otherwise it burns resources continuously and fails continuously.

The name _reset_hygiene_failure_streak tells you it tracks a count of consecutive failures, reset to zero on the first success. This is the same idea as the idempotency lock in Chapter 3.

② Recognizing transient network errors

def _is_transient_network_error(exc: BaseException) -> bool

A long-running process has to tell “the network hiccuped” apart from “something is actually broken.” The former should retry silently; the latter should alert. If you don't distinguish them, users drown in meaningless alerts about network blips and eventually mute all alerts.

③ Redacting secrets from user-visible text

def _redact_gateway_user_facing_secrets(text: str) -> str
def _redact_approval_command(cmd: "str | None") -> str

The gateway sends some internal information into the chat window (error messages, approval requests). That text may carry API keys, tokens, and passwords. And a chat window is visible to many people, searchable, and archived. So it must be redacted before it goes out.

Note that there are two redaction functions: one for general text, one specifically for “the command awaiting approval.” Secrets in a command take different shapes (in an environment variable assignment, in an argument, in a pipeline).

④ Making provider errors user-friendly

def _gateway_provider_error_reply(text: str) -> str
def _looks_like_gateway_provider_error(text: str) -> bool
def _sanitize_gateway_final_response(platform: Any, text: str) -> str

Error messages from model services are usually written for developers (stack traces, internal error codes, request IDs). Sent straight into a chat window, they mean nothing to the user. So they have to be recognized and translated into plain language.

⑤ Restart guard

gateway/restart.py
gateway/restart_loop_guard.py     ← ★ restart storm protection

A restart storm is the classic failure of a long-running process: the process crashes → auto-restarts → crashes again on startup → restarts again → forever. Dozens of restarts per second, logs flooded, CPU pinned.

The usual guard: record recent restart counts and intervals, and if there are too many restarts in a short window, stop auto-restarting and stay down — so a human takes a look.

⑥ Memory monitoring and graceful drain

gateway/memory_monitor.py           memory usage monitoring
gateway/agent_cache_pressure.py     agent cache pressure
gateway/drain_control.py            ★ graceful drain
gateway/disk_status.py              disk status

Agent cache pressure: the gateway caches agent instances (the same session reuses the same one). But every instance holds the full message history — a few dozen active sessions can eat several GB of memory. So the pressure has to be monitored, and inactive instances evicted when necessary.

Graceful drain: when the gateway needs to shut down, you can't just kill it — messages currently being processed would be lost. The right way is “stop accepting new messages, finish what's in hand, then exit.”

⑦ Delivery ledger

gateway/delivery.py
gateway/delivery_ledger.py     ← ★ delivery ledger
gateway/rich_sent_store.py
gateway/message_timestamps.py
gateway/dead_targets.py        ← dead targets (a deleted channel, say)

The delivery ledger records “which reply has already been sent where.” It guards against duplicate delivery: on a network timeout you don't know whether the message went out, and a retry could make the user see it twice. With a ledger, you can tell.

Dead targets covers the cases where the channel the agent wants to reply to has been deleted, the bot has been kicked from the group, or the user has blocked it. These deliveries will fail forever, and they must be recognized and retries stopped, or the retry queue grows without bound.

1.5 Status Messages and Progress Feedback

def _status_template_to_regex(template: str) -> str
def _gateway_compression_progress_notices_enabled() -> bool
def _prepare_gateway_status_message(platform, event_type: str, message: str) -> Optional[str]
async def _send_or_update_status_coro(adapter, chat_id, status_key, content, metadata)
def render_notice_line(notice) -> str

During a long task the agent needs to tell the user “I'm still working.” But doing that inside a chat app is delicate:

• Too many status messages → spam
• None → the user thinks it died
• Updating by editing the same message → only works on platforms that support editing

The function _status_template_to_regex deserves a note: it converts a status-message template into a regular expression. The likely purpose is “recognize which messages in the chat history are status messages I sent earlier,” so they can be updated or deleted — because the message IDs returned by the platform API may already be lost, and matching on content is the only way to find them.

1.6 Interruption and Recovery

def _is_fresh_gateway_interruption(...)
def build_resume_recovery_note(...)
def _prepare_resume_pending_message(...)
def _build_replay_entry(...)
def _startup_restore_drain_timeout_secs() -> float
def _auto_continue_freshness_window() -> float

This group of functions handles: after the gateway restarts, what happens to the sessions that were “halfway through”?

1.7 Built-in Gateway Hooks

gateway/hooks.py
gateway/builtin_hooks/

The gateway layer has its own hook system, letting plugins insert logic at the key points where messages enter and leave. Together with the plugin system in Chapter 9, this gives you the ability to “extend gateway behavior without changing core code.”

1.8 What This Layer Costs