mirror of
https://github.com/NousResearch/hermes-agent.git
synced 2026-07-05 12:42:30 +00:00
6e4e5967f7
Cut over the agent half of Shape A (D-Q1.5a/b.1/c) to front a SET of platforms on
one relay WS:
- relay_platform_identities() parses GATEWAY_RELAY_PLATFORMS (list) +
GATEWAY_RELAY_BOT_IDS (JSON keyed map {platform:{botId,username?}}). Cut over
from the scalar GATEWAY_RELAY_PLATFORM/_BOT_ID (no fallback, D-Q1.5c).
- self_provision_relay() loops one /relay/provision per platform under one
gatewayId+secret, partial-failure-tolerant.
- WebSocketRelayTransport takes the identity SET, sends one hello per identity
(connector accumulates the advertised set), and stamps the per-frame
OutboundFrame.platform + its matching advertised botId on outbound.
- RelayAdapter remembers each chat's underlying source.platform (mirroring the
existing guild/dm scope capture) and tags the reply's egress platform.
- send_relay_policy() declares one relevance policy per fronted platform (the
connector keys policy by (tenant,platform,instanceId)).
Single-platform deploys are byte-identical on the wire (1-element list, no per-frame
tag -> connector session-default fallback). typecheck/ruff clean; relay unit 221 pass
(+10 new); all 15 cross-repo E2E drivers green vs connector origin/main.
709 lines
35 KiB
Python
709 lines
35 KiB
Python
"""Production WebSocket RelayTransport — the gateway's live link to the connector.
|
|
|
|
The gateway dials OUT to the connector's relay endpoint over a WebSocket and
|
|
speaks the newline-delimited JSON frame protocol defined in the connector repo
|
|
(``gateway-gateway`` ``src/relay/protocol.ts``) and mirrored in
|
|
``docs/relay-connector-contract.md``:
|
|
|
|
gateway -> connector : hello, outbound, interrupt
|
|
connector -> gateway : descriptor, inbound, outbound_result, interrupt_inbound
|
|
|
|
Frames:
|
|
hello {type, platform, botId}
|
|
descriptor {type, descriptor} (handshake reply)
|
|
inbound {type, event, bufferId?} (a normalized MessageEvent)
|
|
outbound {type, requestId, action} (send/edit/typing/follow_up)
|
|
outbound_result {type, requestId, result}
|
|
interrupt {type, session_key, reason?} (gateway egresses /stop)
|
|
interrupt_inbound{type, session_key, chat_id} (connector -> owning gateway)
|
|
|
|
This is the concrete transport behind the ``RelayTransport`` Protocol; the
|
|
``RelayAdapter`` delegates all wire I/O to it. Outbound calls block on a
|
|
per-request future keyed by ``requestId`` until the matching ``outbound_result``
|
|
arrives. A background reader task pumps inbound frames to the registered handler
|
|
and resolves pending outbound futures.
|
|
|
|
EXPERIMENTAL: the frame schema may change without a deprecation cycle until at
|
|
least two Class-1 platforms validate it.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import asyncio
|
|
import json
|
|
import logging
|
|
import uuid
|
|
from dataclasses import dataclass
|
|
from typing import Any, Dict, Optional
|
|
|
|
from gateway.platforms.base import MessageEvent, MessageType
|
|
from gateway.session import SessionSource
|
|
from gateway.relay.descriptor import CapabilityDescriptor
|
|
from gateway.relay.transport import InboundHandler
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
try: # lazy/optional dep — mirrors gateway/platforms/feishu.py
|
|
import websockets
|
|
except ImportError: # pragma: no cover - exercised only when the extra is absent
|
|
websockets = None # type: ignore[assignment]
|
|
|
|
WEBSOCKETS_AVAILABLE = websockets is not None
|
|
|
|
# How long to wait for the handshake descriptor and for each outbound result.
|
|
_HANDSHAKE_TIMEOUT_S = 30.0
|
|
_OUTBOUND_TIMEOUT_S = 30.0
|
|
|
|
# Phase 7 Unit 7d-B: the application close code the connector sends when it
|
|
# rejects/revokes a gateway's WS upgrade auth (mirrors the connector's
|
|
# `4401` "unauthorized" close — a private-use code, not a standard WS code).
|
|
# A 4401 received AFTER a successful handshake means the per-gateway secret was
|
|
# revoked (opt-out / deprovision), which the transport treats as terminal.
|
|
_RELAY_UNAUTHORIZED_CLOSE_CODE = 4401
|
|
|
|
|
|
def _ws_dial_url(url: str) -> str:
|
|
"""Normalize a connector URL to the ``ws(s)://…/relay`` dial target.
|
|
|
|
The relay URL is configured once (``GATEWAY_RELAY_URL`` / ``gateway.relay_url``)
|
|
as the connector's BASE URL (e.g. ``https://connector.example``) and shared by
|
|
both the provision POST (which needs ``http(s)://…/relay/provision`` — see
|
|
``_provision_url``) and the WS dial (which needs ``ws(s)://…/relay``, the path
|
|
the connector mounts its ``WebSocketServer`` on). Two normalizations, both
|
|
load-bearing:
|
|
|
|
- scheme: ``https -> wss``, ``http -> ws`` (``websockets.connect`` raises
|
|
"scheme isn't ws or wss" on an http(s) URL).
|
|
- path: ensure it ends in ``/relay`` (the connector returns HTTP 400 on an
|
|
upgrade to any other path, since the WS server is mounted at ``/relay``).
|
|
|
|
Idempotent: an already-``ws(s)://…/relay`` URL is returned unchanged, so a URL
|
|
configured WITH the scheme and/or ``/relay`` still works.
|
|
"""
|
|
raw = (url or "").strip()
|
|
if raw.startswith("https://"):
|
|
raw = "wss://" + raw[len("https://"):]
|
|
elif raw.startswith("http://"):
|
|
raw = "ws://" + raw[len("http://"):]
|
|
raw = raw.rstrip("/")
|
|
if not raw.endswith("/relay"):
|
|
raw = f"{raw}/relay"
|
|
return raw
|
|
|
|
|
|
def _event_from_wire(raw: Dict[str, Any]) -> MessageEvent:
|
|
"""Rebuild a MessageEvent from the connector's normalized inbound payload.
|
|
|
|
The connector emits SessionSource as the snake_case wire form (§3); map it
|
|
back onto the gateway dataclasses. Unknown message types fall back to TEXT.
|
|
"""
|
|
src = raw.get("source", {}) or {}
|
|
from gateway.config import Platform
|
|
|
|
platform = src.get("platform", "relay")
|
|
try:
|
|
platform_enum = Platform(platform)
|
|
except ValueError:
|
|
platform_enum = Platform.RELAY
|
|
|
|
source = SessionSource(
|
|
platform=platform_enum,
|
|
chat_id=src.get("chat_id", ""),
|
|
chat_type=src.get("chat_type", "dm"),
|
|
chat_name=src.get("chat_name"),
|
|
user_id=src.get("user_id"),
|
|
user_name=src.get("user_name"),
|
|
thread_id=src.get("thread_id"),
|
|
chat_topic=src.get("chat_topic"),
|
|
user_id_alt=src.get("user_id_alt"),
|
|
chat_id_alt=src.get("chat_id_alt"),
|
|
guild_id=src.get("guild_id"),
|
|
parent_chat_id=src.get("parent_chat_id"),
|
|
message_id=src.get("message_id"),
|
|
# Authentic upstream-trust signal: this event arrived over the
|
|
# per-instance-authenticated relay WS, so the connector already resolved
|
|
# it to this instance's owner-bound author. ``platform`` is the
|
|
# UNDERLYING platform (e.g. discord), not ``relay`` — authz keys the
|
|
# upstream-trust decision off THIS flag, not off ``platform`` (which
|
|
# would miss because the relay adapter is registered under
|
|
# ``Platform.RELAY``). Stamped here, never read off the wire.
|
|
delivered_via_upstream_relay=True,
|
|
)
|
|
try:
|
|
msg_type = MessageType(raw.get("message_type", "text"))
|
|
except ValueError:
|
|
msg_type = MessageType.TEXT
|
|
|
|
return MessageEvent(
|
|
text=raw.get("text", ""),
|
|
message_type=msg_type,
|
|
source=source,
|
|
message_id=raw.get("message_id"),
|
|
reply_to_message_id=raw.get("reply_to_message_id"),
|
|
media_urls=raw.get("media_urls") or [],
|
|
)
|
|
|
|
|
|
@dataclass
|
|
class PassthroughForward:
|
|
"""A connector-forwarded passthrough-plane request (Phase 5 §5.1).
|
|
|
|
The connector answered the provider's latency-critical ACK at its edge, then
|
|
forwarded the real (already-sanitized) request to this gateway over the WS.
|
|
``body`` is the exact decoded bytes the connector forwarded (the wire carries
|
|
it base64-encoded for byte parity). ``headers`` preserve arrival order.
|
|
"""
|
|
|
|
platform: str
|
|
bot_id: str
|
|
method: str
|
|
path: str
|
|
headers: list[tuple[str, str]]
|
|
body: bytes
|
|
|
|
|
|
def _passthrough_from_wire(raw: Dict[str, Any]) -> PassthroughForward:
|
|
"""Rebuild a PassthroughForward from the connector's wire frame.
|
|
|
|
Mirrors the connector's ``PassthroughForward`` (relay/protocol.ts): the body
|
|
is base64-decoded back to the exact bytes the connector forwarded, so the
|
|
gateway re-processes byte-identical content (the connector is the trust
|
|
boundary; it already verified at the edge).
|
|
"""
|
|
import base64
|
|
|
|
body_b64 = raw.get("bodyB64", "") or ""
|
|
try:
|
|
body = base64.b64decode(body_b64)
|
|
except Exception: # noqa: BLE001 - a malformed body must not crash the reader
|
|
body = b""
|
|
headers_raw = raw.get("headers", []) or []
|
|
headers: list[tuple[str, str]] = []
|
|
for pair in headers_raw:
|
|
if isinstance(pair, (list, tuple)) and len(pair) == 2:
|
|
headers.append((str(pair[0]), str(pair[1])))
|
|
return PassthroughForward(
|
|
platform=str(raw.get("platform", "")),
|
|
bot_id=str(raw.get("botId", "")),
|
|
method=str(raw.get("method", "")),
|
|
path=str(raw.get("path", "")),
|
|
headers=headers,
|
|
body=body,
|
|
)
|
|
|
|
|
|
class WebSocketRelayTransport:
|
|
"""RelayTransport over a WebSocket connection the gateway dials to the connector."""
|
|
|
|
def __init__(
|
|
self,
|
|
url: str,
|
|
platform: str,
|
|
bot_id: str,
|
|
*,
|
|
identities: Optional[list[tuple[str, str]]] = None,
|
|
connect_timeout_s: float = _HANDSHAKE_TIMEOUT_S,
|
|
outbound_timeout_s: float = _OUTBOUND_TIMEOUT_S,
|
|
gateway_id: Optional[str] = None,
|
|
upgrade_secret: Optional[str] = None,
|
|
reconnect: bool = False,
|
|
reconnect_backoff_s: float = 1.0,
|
|
reconnect_max_backoff_s: float = 30.0,
|
|
) -> None:
|
|
if not WEBSOCKETS_AVAILABLE:
|
|
raise RuntimeError(
|
|
"WebSocketRelayTransport requires the 'websockets' package "
|
|
"(install the messaging extra)."
|
|
)
|
|
self._url = _ws_dial_url(url)
|
|
self._platform = platform
|
|
self._bot_id = bot_id
|
|
# Phase 1.5 (Shape A): the full SET of (platform, bot_id) this gateway
|
|
# fronts on this one WS. The handshake sends one `hello` per identity so
|
|
# the connector accumulates them into its advertised set (gateway-gateway
|
|
# D-Q1.5b.1); the first identity (platform/bot_id above) is the default an
|
|
# untagged outbound falls back to. Defaults to the single (platform, bot_id)
|
|
# so existing single-platform callers are unchanged.
|
|
self._identities = list(identities) if identities else [(platform, bot_id)]
|
|
self._connect_timeout_s = connect_timeout_s
|
|
self._outbound_timeout_s = outbound_timeout_s
|
|
# Connection auth (Phase 2): when a per-gateway secret is configured the
|
|
# gateway presents an HMAC bearer on the WS upgrade so the connector can
|
|
# authenticate it (reject 4401 otherwise). gateway_id identifies the
|
|
# enrolled instance — the connector peeks it to index its secret verify
|
|
# list, then verifies the signature. Absent -> unauthenticated upgrade
|
|
# (dev/test, or a connector that doesn't enforce auth).
|
|
self._gateway_id = gateway_id
|
|
self._upgrade_secret = upgrade_secret
|
|
|
|
# Phase 5 §5.3: a NET-NEW reconnect supervisor. The base transport's
|
|
# _read_loop just ends on socket close ("reconnection is caller policy");
|
|
# with reconnect=True the transport re-dials + re-handshakes after an
|
|
# UNEXPECTED close (not a deliberate disconnect()), so a gateway that went
|
|
# idle/suspended re-establishes its socket — which makes the connector
|
|
# drain that instance's buffered-only delivery-leg backlog (onResume) on
|
|
# the new handshake. Off by default so existing tests + the stub are
|
|
# unaffected; register_relay_adapter turns it on in production.
|
|
self._reconnect = reconnect
|
|
self._reconnect_backoff_s = reconnect_backoff_s
|
|
self._reconnect_max_backoff_s = reconnect_max_backoff_s
|
|
self._supervisor: Optional[asyncio.Task[None]] = None
|
|
# scale-to-zero §Phase 0 (D12/F14): a DORMANT close is distinct from both
|
|
# disconnect() (terminal: cancels the supervisor) and an unexpected close
|
|
# (re-dials immediately). go_dormant() sets this True, then closes the
|
|
# socket WITHOUT setting _closing — so _read_loop's fall-through still
|
|
# kicks the reconnect supervisor (the wake path stays armed), but the
|
|
# supervisor waits on the longer dormant cadence instead of the fast
|
|
# reconnect backoff, so it does not fight the platform's suspend window.
|
|
# On resume (process unfrozen) the pending wait completes, the re-dial
|
|
# succeeds, and the connector drains this instance's buffered backlog on
|
|
# the new handshake. Cleared on a successful re-dial (_dial_and_start).
|
|
self._dormant = False
|
|
# The re-dial poll cadence while dormant. A suspended machine's event
|
|
# loop is frozen, so this timer only advances once the machine is awake;
|
|
# it just needs to be short enough that a freshly-woken machine re-dials
|
|
# promptly (the connector's wake poke is what triggers the platform
|
|
# autostart in the first place — §3.4(5)).
|
|
self._dormant_redial_s = 1.0
|
|
|
|
self._ws: Any = None
|
|
self._reader: Optional[asyncio.Task[None]] = None
|
|
self._inbound: Optional[InboundHandler] = None
|
|
self._descriptor: Optional[CapabilityDescriptor] = None
|
|
self._descriptor_ready: asyncio.Future[CapabilityDescriptor] | None = None
|
|
# requestId -> future awaiting the matching outbound_result.
|
|
self._pending: Dict[str, asyncio.Future[Dict[str, Any]]] = {}
|
|
# Phase 5 §5.3: future awaiting the connector's going_idle_ack.
|
|
self._going_idle_ack: asyncio.Future[None] | None = None
|
|
self._closing = False
|
|
# Phase 7 Unit 7d-B: a 4401 (unauthorized) close AFTER we have already
|
|
# handshaked successfully at least once means the connector REVOKED this
|
|
# gateway's per-gateway secret — i.e. the operator opted this instance
|
|
# OUT of the relay (Unit 7b deprovision). That is TERMINAL: the secret is
|
|
# gone, so re-dialing just spins against a dead credential forever
|
|
# (the "retrying 4401" the dashboard showed). We stop reconnecting and
|
|
# surface it as a clean, non-retryable "disabled" state. A 4401 BEFORE
|
|
# any successful handshake stays retryable — that's a cold-start /
|
|
# not-yet-provisioned race, not a revocation.
|
|
self._handshake_succeeded = False
|
|
self._auth_revoked = False
|
|
|
|
# ── lifecycle ────────────────────────────────────────────────────────
|
|
async def connect(self) -> bool:
|
|
await self._dial_and_start()
|
|
return True
|
|
|
|
async def _dial_and_start(self) -> None:
|
|
"""Open the socket, start the reader, send hello. Used by connect() and
|
|
by the reconnect supervisor on a re-dial."""
|
|
loop = asyncio.get_running_loop()
|
|
self._descriptor_ready = loop.create_future()
|
|
# A fresh handshake is coming; clear any stale descriptor so handshake()
|
|
# awaits the new one (matters on a re-dial).
|
|
self._descriptor = None
|
|
# scale-to-zero (D12): a successful (re-)dial ends any dormant state — we
|
|
# are live again, so a subsequent UNEXPECTED close should reconnect on the
|
|
# normal fast backoff, not the dormant cadence.
|
|
self._dormant = False
|
|
headers = self._upgrade_headers()
|
|
if headers:
|
|
self._ws = await websockets.connect(self._url, additional_headers=headers) # type: ignore[union-attr]
|
|
else:
|
|
self._ws = await websockets.connect(self._url) # type: ignore[union-attr]
|
|
self._reader = asyncio.create_task(self._read_loop(), name="relay-ws-reader")
|
|
# Send one hello PER fronted identity (Phase 1.5 Shape A). The connector
|
|
# accumulates them into its advertised set (the first sets the session
|
|
# default; each adds to the egress-allowed set). A single-platform gateway
|
|
# sends exactly one hello — byte-identical to before. The descriptor for
|
|
# the FIRST identity resolves handshake(); later descriptors are absorbed.
|
|
for platform, bot_id in self._identities:
|
|
await self._send({"type": "hello", "platform": platform, "botId": bot_id})
|
|
|
|
def _upgrade_headers(self) -> Dict[str, str]:
|
|
"""Auth headers for the WS upgrade, or {} when no secret is configured.
|
|
|
|
Presents ``Authorization: Bearer *** where the token is a signed
|
|
bearer built with the per-gateway secret (``gateway/relay/auth.py``
|
|
``make_upgrade_token``), keyed by ``gateway_id`` so the connector can
|
|
index its verify list. The connector rejects the upgrade (close 4401)
|
|
when this is missing/invalid/revoked; an unauthenticated connector
|
|
ignores it.
|
|
"""
|
|
if not (self._upgrade_secret and self._gateway_id):
|
|
return {}
|
|
from gateway.relay.auth import make_upgrade_token
|
|
|
|
token = make_upgrade_token(self._gateway_id, self._upgrade_secret)
|
|
return {"Authorization": f"Bearer {token}"}
|
|
|
|
async def disconnect(self) -> None:
|
|
self._closing = True
|
|
if self._supervisor is not None:
|
|
self._supervisor.cancel()
|
|
try:
|
|
await self._supervisor
|
|
except (asyncio.CancelledError, Exception): # noqa: BLE001 - best-effort teardown
|
|
pass
|
|
self._supervisor = None
|
|
if self._reader is not None:
|
|
self._reader.cancel()
|
|
try:
|
|
await self._reader
|
|
except (asyncio.CancelledError, Exception): # noqa: BLE001 - best-effort teardown
|
|
pass
|
|
self._reader = None
|
|
if self._ws is not None:
|
|
try:
|
|
await self._ws.close()
|
|
except Exception: # noqa: BLE001
|
|
pass
|
|
self._ws = None
|
|
# Fail any in-flight outbound waiters so callers don't hang.
|
|
for fut in self._pending.values():
|
|
if not fut.done():
|
|
fut.set_exception(RuntimeError("relay transport closed"))
|
|
self._pending.clear()
|
|
if self._going_idle_ack is not None and not self._going_idle_ack.done():
|
|
self._going_idle_ack.set_exception(RuntimeError("relay transport closed"))
|
|
|
|
async def handshake(self) -> CapabilityDescriptor:
|
|
if self._descriptor is not None:
|
|
return self._descriptor
|
|
if self._descriptor_ready is None:
|
|
raise RuntimeError("handshake() called before connect()")
|
|
return await asyncio.wait_for(self._descriptor_ready, timeout=self._connect_timeout_s)
|
|
|
|
@property
|
|
def auth_revoked(self) -> bool:
|
|
"""True once the connector closed the socket with 4401 AFTER a prior
|
|
successful handshake — i.e. the per-gateway secret was revoked (the
|
|
operator opted this instance out of the relay). Terminal: the transport
|
|
stops reconnecting, and the adapter surfaces a clean "disabled" state."""
|
|
return self._auth_revoked
|
|
|
|
def set_inbound_handler(self, handler: InboundHandler) -> None:
|
|
self._inbound = handler
|
|
|
|
# ── outbound ─────────────────────────────────────────────────────────
|
|
async def send_outbound(
|
|
self, action: Dict[str, Any], *, platform: Optional[str] = None
|
|
) -> Dict[str, Any]:
|
|
return await self._request_response(action, platform=platform)
|
|
|
|
async def send_follow_up(
|
|
self, action: Dict[str, Any], *, platform: Optional[str] = None
|
|
) -> Dict[str, Any]:
|
|
# follow_up rides the same outbound frame; the connector dispatches by
|
|
# action.op. Kept as a distinct method to satisfy the transport Protocol
|
|
# and to make the A2 call site explicit.
|
|
return await self._request_response(action, platform=platform)
|
|
|
|
def _bot_id_for(self, platform: Optional[str]) -> Optional[str]:
|
|
"""The bot_id this transport advertised at hello for ``platform`` (Phase 1.5).
|
|
|
|
The connector validates a per-frame egress target against the SET of
|
|
``platform:botId`` pairs it accumulated from the N hellos, so a per-frame
|
|
``platform`` must ride with its MATCHING ``botId`` (the session default
|
|
botId belongs to the first identity and would mis-key for a second
|
|
platform). Resolved from the identity set this transport was built with.
|
|
None when the platform isn't one we front (the connector then rejects it
|
|
with a structured failure — never a wrong-credential send)."""
|
|
if not platform:
|
|
return None
|
|
for p, b in self._identities:
|
|
if p == platform:
|
|
return b
|
|
return None
|
|
|
|
async def get_chat_info(self, chat_id: str) -> Dict[str, Any]:
|
|
result = await self._request_response(
|
|
{"op": "get_chat_info", "chat_id": chat_id}, frame_type="outbound"
|
|
)
|
|
# The connector answers chat-info inside the outbound_result envelope.
|
|
info = result.get("chat_info") or result
|
|
return {"name": info.get("name", chat_id), "type": info.get("type", "dm")}
|
|
|
|
async def send_interrupt(self, session_key: str, reason: Optional[str] = None) -> None:
|
|
await self._send({"type": "interrupt", "session_key": session_key, "reason": reason})
|
|
|
|
# ── going-idle / buffered-flip (Phase 5 §5.3) ────────────────────────
|
|
async def go_idle(self, timeout_s: float = 10.0) -> bool:
|
|
"""Ask the connector to flip this instance's destination to buffered-only.
|
|
|
|
Sends ``going_idle`` and awaits the connector's ``going_idle_ack`` — the
|
|
connector-AUTHORITATIVE confirmation that live delivery has stopped and
|
|
subsequent inbound buffers durably (Q-5.3c). Returns True on ack, False on
|
|
timeout / not-connected (the caller proceeds to close anyway — at worst a
|
|
live event races a closing socket exactly as before §5.3, no regression).
|
|
|
|
The gateway stays serving (the read loop keeps handling inbound) until the
|
|
ack, so an event landing in the flip window is delivered live, not lost.
|
|
"""
|
|
if self._ws is None:
|
|
return False
|
|
loop = asyncio.get_running_loop()
|
|
self._going_idle_ack = loop.create_future()
|
|
try:
|
|
await self._send({"type": "going_idle"})
|
|
await asyncio.wait_for(self._going_idle_ack, timeout=timeout_s)
|
|
return True
|
|
except (asyncio.TimeoutError, Exception): # noqa: BLE001 - ack is best-effort
|
|
return False
|
|
finally:
|
|
self._going_idle_ack = None
|
|
|
|
async def go_dormant(self, timeout_s: float = 10.0) -> bool:
|
|
"""Quiesce this transport for a scale-to-zero suspend (D12 / Phase 0).
|
|
|
|
Distinct from BOTH ``disconnect()`` and an unexpected close (F14):
|
|
- ``disconnect()`` sets ``_closing=True`` and CANCELS the reconnect
|
|
supervisor — terminal, "shutting down for good." A machine suspended
|
|
after that never re-dials on wake, so its buffered backlog strands.
|
|
- An unexpected close re-dials IMMEDIATELY (fast backoff) — the socket
|
|
never stays down, so the platform proxy never sees the connection go
|
|
away and never suspends the machine.
|
|
|
|
``go_dormant()`` is the third mode the suspend behaviour needs:
|
|
1. ``go_idle()`` → the connector flips this instance to buffered-only
|
|
and acks (so inbound that arrives while we sleep buffers durably and
|
|
replays on the next handshake).
|
|
2. Close the socket so the platform proxy sees load drop to zero (the
|
|
precondition for Fly ``autostop:"suspend"``) — but WITHOUT setting
|
|
``_closing``. The reader's normal end-of-socket fall-through still
|
|
arms the reconnect supervisor, so the wake path stays live; the
|
|
``_dormant`` flag just makes that supervisor poll on the dormant
|
|
cadence rather than fight the suspend window.
|
|
|
|
On resume (process unfrozen) the supervisor's pending wait completes, the
|
|
re-dial succeeds, and the connector drains the buffered backlog on the new
|
|
handshake. Returns the ``go_idle`` ack result (True on ack); the dormancy
|
|
close happens regardless (a missed ack at worst races one live event onto
|
|
a closing socket, exactly as §5.3 already tolerates).
|
|
|
|
No-op-safe: a transport that never connected (``_ws is None``) just
|
|
returns False without closing.
|
|
"""
|
|
if self._ws is None:
|
|
return False
|
|
acked = await self.go_idle(timeout_s=timeout_s)
|
|
# Mark dormant BEFORE closing so the supervisor (armed by the reader's
|
|
# fall-through) takes the dormant cadence, and a racing live event can't
|
|
# flip us back to a fast reconnect.
|
|
self._dormant = True
|
|
try:
|
|
await self._ws.close()
|
|
except Exception: # noqa: BLE001 - best-effort; the reader still ends + arms reconnect
|
|
logger.debug("relay go_dormant: ws.close() raised", exc_info=True)
|
|
return acked
|
|
|
|
async def _send_inbound_ack(self, buffer_id: str) -> None:
|
|
"""Acknowledge durable receipt of a buffered inbound delivery (§5.3).
|
|
|
|
Sent after the adapter has durably taken a buffered inbound event the
|
|
connector replayed on reconnect; the connector acks the buffer entry only
|
|
after this, giving drain-without-dup on the delivery leg.
|
|
"""
|
|
try:
|
|
await self._send({"type": "inbound_ack", "bufferId": buffer_id})
|
|
except Exception: # noqa: BLE001 - a failed ack just redelivers the entry next time
|
|
logger.debug("relay: inbound_ack send failed for %s", buffer_id)
|
|
|
|
async def _request_response(
|
|
self,
|
|
action: Dict[str, Any],
|
|
frame_type: str = "outbound",
|
|
*,
|
|
platform: Optional[str] = None,
|
|
) -> Dict[str, Any]:
|
|
if self._ws is None:
|
|
return {"success": False, "error": "relay transport not connected"}
|
|
request_id = uuid.uuid4().hex
|
|
loop = asyncio.get_running_loop()
|
|
fut: asyncio.Future[Dict[str, Any]] = loop.create_future()
|
|
self._pending[request_id] = fut
|
|
frame: Dict[str, Any] = {"type": frame_type, "requestId": request_id, "action": action}
|
|
# Phase 1.5: tag the per-frame egress platform on the OutboundFrame
|
|
# envelope (gateway-gateway D-Q1.5b.1), with its MATCHING advertised botId
|
|
# so the connector's `${platform}:${botId}` advertised-set check passes.
|
|
# Only set when a concrete platform was resolved for this chat so a
|
|
# single-platform gateway emits the exact frame shape as before (the
|
|
# connector falls back to the session's default platform when absent).
|
|
if platform:
|
|
frame["platform"] = platform
|
|
bot_id = self._bot_id_for(platform)
|
|
if bot_id:
|
|
frame["botId"] = bot_id
|
|
try:
|
|
await self._send(frame)
|
|
return await asyncio.wait_for(fut, timeout=self._outbound_timeout_s)
|
|
except asyncio.TimeoutError:
|
|
return {"success": False, "error": "relay outbound timed out"}
|
|
finally:
|
|
self._pending.pop(request_id, None)
|
|
|
|
# ── wire I/O ─────────────────────────────────────────────────────────
|
|
async def _send(self, frame: Dict[str, Any]) -> None:
|
|
if self._ws is None:
|
|
raise RuntimeError("relay transport not connected")
|
|
await self._ws.send(json.dumps(frame) + "\n")
|
|
|
|
async def _read_loop(self) -> None:
|
|
assert self._ws is not None
|
|
buf = ""
|
|
try:
|
|
async for chunk in self._ws:
|
|
buf += chunk if isinstance(chunk, str) else chunk.decode("utf-8")
|
|
# Newline-delimited frames; keep any trailing partial line.
|
|
*lines, buf = buf.split("\n")
|
|
for line in lines:
|
|
if line.strip():
|
|
await self._handle_frame(line)
|
|
except asyncio.CancelledError:
|
|
raise
|
|
except Exception as exc: # noqa: BLE001 - log + let the task end; reconnection handled below
|
|
# Phase 7 Unit 7d-B: detect a 4401 (unauthorized) close. After a prior
|
|
# successful handshake this is a REVOCATION (opt-out / deprovision) —
|
|
# the per-gateway secret is gone, so reconnecting is futile. Latch a
|
|
# terminal "auth revoked" state and DON'T re-dial. Before any
|
|
# successful handshake a 4401 stays retryable (cold-start race).
|
|
if self._close_code_of(exc) == _RELAY_UNAUTHORIZED_CLOSE_CODE and self._handshake_succeeded:
|
|
self._auth_revoked = True
|
|
if not self._closing:
|
|
logger.warning(
|
|
"relay ws closed 4401 (unauthorized) after a successful handshake — "
|
|
"treating as a revoked relay credential (opt-out); not reconnecting"
|
|
)
|
|
elif not self._closing:
|
|
logger.warning("relay ws read loop ended: %s", exc)
|
|
# Phase 5 §5.3: the socket closed. If reconnect is enabled and this was
|
|
# NOT a deliberate disconnect(), kick the reconnect supervisor so the
|
|
# gateway re-dials + re-handshakes (which triggers the connector's
|
|
# buffered-flip drain on the new handshake). Self-scheduling: the reader
|
|
# ends here, the supervisor re-dials and starts a fresh reader.
|
|
# Phase 7 Unit 7d-B: a revoked credential (terminal 4401) is the one case
|
|
# we deliberately do NOT reconnect — the secret is dead until the
|
|
# instance is recreated, so spinning would just reproduce the failure.
|
|
if (
|
|
self._reconnect
|
|
and not self._closing
|
|
and not self._auth_revoked
|
|
and (self._supervisor is None or self._supervisor.done())
|
|
):
|
|
self._supervisor = asyncio.create_task(
|
|
self._reconnect_loop(), name="relay-ws-reconnect"
|
|
)
|
|
|
|
@staticmethod
|
|
def _close_code_of(exc: BaseException) -> Optional[int]:
|
|
"""Best-effort extraction of a WebSocket close code from a raised
|
|
exception. websockets' ConnectionClosed* expose the peer's Close frame
|
|
via `.rcvd`/`.sent` (preferred; `.code` is deprecated in websockets 13+).
|
|
Returns None when unknown."""
|
|
for attr in ("rcvd", "sent"):
|
|
frame = getattr(exc, attr, None)
|
|
fcode = getattr(frame, "code", None)
|
|
if isinstance(fcode, int):
|
|
return fcode
|
|
code = getattr(exc, "code", None)
|
|
return code if isinstance(code, int) else None
|
|
|
|
async def _reconnect_loop(self) -> None:
|
|
"""Re-dial the connector with capped exponential backoff until reconnected
|
|
or disconnect() is called. NET-NEW for §5.3: a re-established socket makes
|
|
the connector replay this instance's buffered-only backlog on the new
|
|
handshake (the delivery-leg onResume). Never raises out (a re-dial failure
|
|
just retries); ends when a dial succeeds (its reader takes over) or closing.
|
|
|
|
scale-to-zero (D12): when the close was a deliberate go_dormant() rather
|
|
than an unexpected drop, start from the dormant poll cadence. On a
|
|
suspended machine the event loop is frozen, so this sleep only advances
|
|
once the machine is awake — it just needs to be short enough that a
|
|
freshly-woken machine re-dials promptly. A successful _dial_and_start()
|
|
clears _dormant, so any LATER unexpected drop reconnects on the normal
|
|
fast backoff."""
|
|
backoff = self._dormant_redial_s if self._dormant else self._reconnect_backoff_s
|
|
while not self._closing:
|
|
try:
|
|
await asyncio.sleep(backoff)
|
|
except asyncio.CancelledError:
|
|
raise
|
|
if self._closing:
|
|
return
|
|
try:
|
|
await self._dial_and_start()
|
|
logger.info("relay ws reconnected")
|
|
return # the fresh reader is running; supervisor's job is done
|
|
except asyncio.CancelledError:
|
|
raise
|
|
except Exception as exc: # noqa: BLE001 - keep retrying on dial failure
|
|
logger.warning("relay ws reconnect failed: %s", exc)
|
|
backoff = min(backoff * 2, self._reconnect_max_backoff_s)
|
|
|
|
async def _handle_frame(self, line: str) -> None:
|
|
try:
|
|
frame = json.loads(line)
|
|
except json.JSONDecodeError:
|
|
logger.warning("relay: skipping malformed frame")
|
|
return
|
|
ftype = frame.get("type")
|
|
if ftype == "descriptor":
|
|
descriptor = CapabilityDescriptor.from_json(json.dumps(frame.get("descriptor", {})))
|
|
self._descriptor = descriptor
|
|
# Phase 7 Unit 7d-B: a received descriptor means the WS upgrade auth
|
|
# passed and the connector accepted us — record that we've handshaked
|
|
# at least once, so a LATER 4401 close is read as a revocation
|
|
# (opt-out), not a cold-start race.
|
|
self._handshake_succeeded = True
|
|
if self._descriptor_ready is not None and not self._descriptor_ready.done():
|
|
self._descriptor_ready.set_result(descriptor)
|
|
elif ftype == "inbound":
|
|
if self._inbound is not None:
|
|
event = _event_from_wire(frame.get("event", {}))
|
|
await self._inbound(event)
|
|
# Phase 5 §5.3: a buffered delivery (replayed on reconnect) carries
|
|
# a bufferId; ack it after the handler has durably taken it so the
|
|
# connector advances its delivery-leg buffer cursor (no dup). A live
|
|
# delivery has no bufferId — nothing to ack.
|
|
buffer_id = frame.get("bufferId")
|
|
if buffer_id:
|
|
await self._send_inbound_ack(str(buffer_id))
|
|
elif ftype == "going_idle_ack":
|
|
# Phase 5 §5.3: the connector confirmed our destination is now
|
|
# buffered-only; resolve the waiter go_idle() is blocked on.
|
|
if self._going_idle_ack is not None and not self._going_idle_ack.done():
|
|
self._going_idle_ack.set_result(None)
|
|
elif ftype == "outbound_result":
|
|
fut = self._pending.get(frame.get("requestId", ""))
|
|
if fut is not None and not fut.done():
|
|
fut.set_result(frame.get("result", {}))
|
|
elif ftype == "interrupt_inbound":
|
|
# Bridged into the adapter's interrupt path by the runner wiring.
|
|
handler = getattr(self, "_interrupt_inbound_handler", None)
|
|
if handler is not None:
|
|
await handler(frame.get("session_key", ""), frame.get("chat_id", ""))
|
|
elif ftype == "passthrough_forward":
|
|
# Phase 5 §5.1: a forwarded passthrough-plane request (Discord
|
|
# interaction, Twilio, …) the connector already edge-ACKed. It rides
|
|
# the SAME outbound WS as inbound messages so a hosted gateway needs
|
|
# no public inbound port. Dispatch to the adapter's handler; the
|
|
# bufferId (when present, §5.3 buffered flip) is passed for ack.
|
|
handler = getattr(self, "_passthrough_handler", None)
|
|
if handler is not None:
|
|
fwd = _passthrough_from_wire(frame.get("forward", {}))
|
|
await handler(fwd, frame.get("bufferId"))
|
|
else:
|
|
# hello/outbound/interrupt are gateway->connector; ignore if echoed.
|
|
pass
|
|
|
|
def set_interrupt_inbound_handler(self, handler: Any) -> None:
|
|
"""Register the callback for connector->gateway interrupt_inbound frames."""
|
|
self._interrupt_inbound_handler = handler
|
|
|
|
def set_passthrough_handler(self, handler: Any) -> None:
|
|
"""Register the callback for connector->gateway passthrough_forward frames.
|
|
|
|
Mirrors set_interrupt_inbound_handler: the runner/adapter wires this so a
|
|
forwarded passthrough request (Phase 5 §5.1) reaches the adapter over the
|
|
same outbound WS the gateway already holds. ``handler(forward, buffer_id)``.
|
|
"""
|
|
self._passthrough_handler = handler
|