feat(relay): multi-platform-per-agent — list identity, N-hello, per-frame egress (Phase 1.5)

gateway/relay/__init__.py

@@ -44,15 +44,81 @@ def relay_url() -> Optional[str]:
     return None
 
 
+def relay_platform_identities() -> list[tuple[str, str]]:
+    """The (platform, bot_id) pairs this gateway fronts over the relay (Phase 1.5).
+
+    Shape A (multi-platform-per-agent, D-Q1.5c — CUT OVER, no scalar fallback):
+    one gateway fronts a SET of platforms on one WS connection. The set is the
+    env-stamped deploy config:
+
+      - ``GATEWAY_RELAY_PLATFORMS`` — comma-sep list (e.g. ``discord,telegram``).
+      - ``GATEWAY_RELAY_BOT_IDS`` — JSON keyed map
+        ``{"discord": {"botId": "..."}, "telegram": {"botId": "...", "username": "..."}}``.
+
+    Returns the ordered list of ``(platform, bot_id)`` pairs (the FIRST is the
+    default the handshake/descriptor falls back to). The connector accepts N
+    hellos accumulating into its advertised set; outbound frames discriminate
+    per-frame on the platform (gateway-gateway D-Q1.5b.1). A platform present in
+    the list but absent from the ids map resolves with an empty bot_id (the
+    connector rejects an unprovisioned platform with a structured failure).
+
+    Defaults to ``[("relay", "")]`` when nothing is configured (the generic
+    single-plane fallback for a connector that didn't stamp a platform set).
+    """
+    platforms_raw = os.environ.get("GATEWAY_RELAY_PLATFORMS", "").strip()
+    platforms = [p.strip() for p in platforms_raw.split(",") if p.strip()]
+    if not platforms:
+        return [("relay", "")]
+    ids = _relay_bot_ids_map()
+    out: list[tuple[str, str]] = []
+    for platform in platforms:
+        entry = ids.get(platform) or {}
+        bot_id = str(entry.get("botId", "")).strip() if isinstance(entry, dict) else ""
+        out.append((platform, bot_id))
+    return out
+
+
+def _relay_bot_ids_map() -> dict:
+    """Parse ``GATEWAY_RELAY_BOT_IDS`` (JSON keyed map). Never raises — a malformed
+    map yields ``{}`` so a bad config degrades to empty bot ids (the connector
+    rejects an unprovisioned platform) rather than crashing boot."""
+    import json
+    import logging
+
+    raw = os.environ.get("GATEWAY_RELAY_BOT_IDS", "").strip()
+    if not raw:
+        return {}
+    try:
+        parsed = json.loads(raw)
+        return parsed if isinstance(parsed, dict) else {}
+    except Exception:  # noqa: BLE001 - a bad map must not crash boot
+        logging.getLogger("gateway.relay").warning(
+            "GATEWAY_RELAY_BOT_IDS is not valid JSON; treating as empty"
+        )
+        return {}
+
+
+def relay_bot_username(platform: str) -> Optional[str]:
+    """The bot's deep-link username/handle for a platform (e.g. Telegram's
+    ``@handle`` for ``t.me/<handle>``), read from the per-platform entry in
+    ``GATEWAY_RELAY_BOT_IDS``. None when absent (most platforms don't need one).
+    """
+    entry = _relay_bot_ids_map().get(platform)
+    if isinstance(entry, dict):
+        username = entry.get("username")
+        if username:
+            return str(username).lstrip("@")
+    return None
+
+
 def relay_platform_identity() -> tuple[str, str]:
-    """Platform + bot id this gateway fronts over the relay (for the handshake hello).
+    """The PRIMARY (platform, bot_id) — the first identity in the configured set.
 
-    Defaults to ``("relay", "")``; overridable via ``GATEWAY_RELAY_PLATFORM`` /
-    ``GATEWAY_RELAY_BOT_ID`` so one connector can front several platforms.
+    Kept for call sites that need a single representative identity (the default
+    descriptor platform, the policy projection's primary). The full set is
+    ``relay_platform_identities()``. Defaults to ``("relay", "")``.
     """
-    platform = os.environ.get("GATEWAY_RELAY_PLATFORM", "relay").strip() or "relay"
-    bot_id = os.environ.get("GATEWAY_RELAY_BOT_ID", "").strip()
-    return platform, bot_id
+    return relay_platform_identities()[0]
 
 
 def relay_connection_auth() -> tuple[Optional[str], Optional[str]]:
@@ -217,14 +283,16 @@ def _policy_url(relay_dial_url: str) -> str:
     return f"{raw}/relay/policy"
 
 
-def relay_relevance_policy() -> Optional[dict]:
-    """Project this gateway's RELEVANCE config into the connector's generic vocabulary.
+def relay_relevance_policy(platform: Optional[str] = None) -> Optional[dict]:
+    """Project a fronted platform's RELEVANCE config into the connector's generic vocabulary.
 
     The connector's relevance gate (Phase 6 Unit ζ) reasons over a
     platform-agnostic policy — ``requireAddress`` / ``freeResponseScopes`` /
     ``allowOtherBots`` — NOT over Discord/Telegram words. This is the gateway
     side of that contract: it reads the agent's existing relevance knobs and
-    emits the generic shape the connector stores per-instance.
+    emits the generic shape the connector stores per-instance (Phase 1.5: the
+    connector keys the policy by ``(tenant, platform, instanceId)``, so each
+    fronted platform gets its own row — pass its name here).
 
     Mapping (the connector vocabulary ← the gateway's existing config):
       - ``requireAddress``     ← the platform's ``require_mention`` (the agent
@@ -237,11 +305,13 @@ def relay_relevance_policy() -> Optional[dict]:
 
     Read from the relay platform's config block (the platform the connector
     fronts, e.g. ``discord:``), falling back to the bridged top-level keys, then
-    the ``{PLATFORM}_*`` env. Returns the generic dict, or None when relay isn't
+    the ``{PLATFORM}_*`` env. ``platform`` defaults to the PRIMARY fronted
+    platform (back-compat). Returns the generic dict, or None when relay isn't
     configured or the platform exposes no relevance knobs (⇒ the connector's
     quiet default already matches, so there's nothing to declare).
     """
-    platform, _bot_id = relay_platform_identity()
+    if platform is None:
+        platform, _bot_id = relay_platform_identity()
     if not platform or platform == "relay":
         # No concrete fronted platform resolved ⇒ nothing platform-specific to project.
         return None
@@ -428,7 +498,7 @@ def self_provision_relay() -> bool:
         logger.warning("relay self-provision skipped: could not resolve Nous token (%s)", exc)
         return False
 
-    platform, bot_id = relay_platform_identity()
+    identities = relay_platform_identities()
     # gatewayId default mirrors the enroll CLI's hostname-based slug.
     import socket
 
@@ -442,35 +512,61 @@ def self_provision_relay() -> bool:
     instance_id = relay_instance_id()
     wake_url = relay_wake_url()
 
-    try:
-        result = _post_provision(
-            provision_url=_provision_url(dial_url),
-            access_token=access_token,
-            gateway_id=gateway_id,
-            platform=platform,
-            bot_id=bot_id,
-            gateway_endpoint=endpoint,
-            route_keys=route_keys,
-            instance_id=instance_id,
-            wake_url=wake_url,
+    # Phase 1.5 (D-Q1.5c): provision EACH fronted platform under the SAME
+    # gatewayId + the SAME (platform-less) per-gateway secret. The connector's
+    # secret record is (gatewayId -> tenant) only; platform/botId live on the
+    # per-platform route rows (relayProvision.ts:124/148), so N provision POSTs
+    # with one gatewayId add N platforms' routes under one secret. The loop is
+    # PARTIAL-FAILURE-TOLERANT: a platform that fails to provision is logged and
+    # skipped (it just isn't fronted) — the others still come up. The FIRST
+    # successful provision sets the in-process creds; later platforms re-provision
+    # against the same gatewayId (idempotent on the secret, additive on routes).
+    provisioned: list[str] = []
+    result: dict = {}
+    for platform, bot_id in identities:
+        try:
+            result = _post_provision(
+                provision_url=_provision_url(dial_url),
+                access_token=access_token,
+                gateway_id=gateway_id,
+                platform=platform,
+                bot_id=bot_id,
+                gateway_endpoint=endpoint,
+                route_keys=route_keys,
+                instance_id=instance_id,
+                wake_url=wake_url,
+            )
+        except RuntimeError as exc:
+            logger.warning(
+                "relay self-provision failed for platform=%s (%s); continuing with the rest",
+                platform,
+                exc,
+            )
+            continue
+        provisioned.append(platform)
+        # Set creds in-process on the FIRST success so register_relay_adapter()
+        # reads them from os.environ (the per-gateway secret authenticates the
+        # outbound WS upgrade). Subsequent platforms share the same gatewayId +
+        # secret (the connector returns the same record for the same gatewayId).
+        # Never logged.
+        if "GATEWAY_RELAY_SECRET" not in os.environ or not os.environ.get("GATEWAY_RELAY_SECRET"):
+            os.environ["GATEWAY_RELAY_ID"] = str(result.get("gatewayId") or gateway_id)
+            os.environ["GATEWAY_RELAY_SECRET"] = str(result.get("secret") or "")
+            os.environ["GATEWAY_RELAY_DELIVERY_KEY"] = str(result.get("deliveryKey") or "")
+
+    if not provisioned:
+        logger.warning(
+            "relay self-provision failed for ALL platforms (%s); gateway will boot without relay auth",
+            ",".join(p for p, _ in identities),
         )
-    except RuntimeError as exc:
-        logger.warning("relay self-provision failed (%s); gateway will boot without relay auth", exc)
         return False
 
-    # Set creds in-process so register_relay_adapter() reads them from os.environ
-    # (the per-gateway secret authenticates the outbound WS upgrade). The delivery
-    # key is still issued by the connector and persisted for forward-compat, but
-    # inbound now rides the WS (no HTTP receiver), so it is not consumed here.
-    # Never logged.
-    os.environ["GATEWAY_RELAY_ID"] = str(result.get("gatewayId") or gateway_id)
-    os.environ["GATEWAY_RELAY_SECRET"] = str(result.get("secret") or "")
-    os.environ["GATEWAY_RELAY_DELIVERY_KEY"] = str(result.get("deliveryKey") or "")
     tenant = str(result.get("tenant") or "")
     logger.info(
-        "relay self-provisioned (gateway_id=%s tenant=%s routes=%d inbound=%s instance=%s wake=%s)",
-        os.environ["GATEWAY_RELAY_ID"],
+        "relay self-provisioned (gateway_id=%s tenant=%s platforms=%s routes=%d inbound=%s instance=%s wake=%s)",
+        os.environ.get("GATEWAY_RELAY_ID", gateway_id),
         tenant or "?",
+        ",".join(provisioned),
         len(route_keys),
         "yes" if endpoint else "outbound-only",
         instance_id or "unbound",
@@ -546,33 +642,50 @@ def send_relay_policy() -> bool:
         # be unauthenticated too, so there's no instance to attach a policy to).
         return False
 
-    policy = relay_relevance_policy()
-    if policy is None:
-        # Nothing non-default to declare ⇒ the connector's quiet default already
-        # matches; don't write a redundant row.
-        logger.info("relay policy: no non-default relevance config to declare; using connector default")
-        return False
-
+    # Phase 1.5: declare a policy PER fronted platform — the connector keys the
+    # relevance policy by (tenant, platform, instanceId), so each platform this
+    # gateway fronts gets its own row. Per-platform, partial-tolerant: a platform
+    # with nothing non-default to declare is skipped; a failed POST for one
+    # platform doesn't block the others. A single-platform gateway declares one
+    # policy exactly as before.
     try:
         from gateway.relay.auth import make_upgrade_token
 
         token = make_upgrade_token(gateway_id, secret)
-        status = _post_policy(policy_url=_policy_url(dial_url), token=token, policy=policy)
-    except Exception as exc:  # noqa: BLE001 - boot must survive a policy-declare failure
-        logger.warning("relay policy declaration failed (%s); connector keeps prior/default policy", exc)
+    except Exception as exc:  # noqa: BLE001 - boot must survive a token-build failure
+        logger.warning("relay policy declaration failed to build token (%s); connector keeps prior policy", exc)
         return False
 
-    if status == 200:
-        logger.info(
-            "relay policy declared (platform=%s require_address=%s free_scopes=%d allow_bots=%s)",
-            policy.get("platform"),
-            policy.get("requireAddress"),
-            len(policy.get("freeResponseScopes") or []),
-            policy.get("allowOtherBots"),
-        )
-        return True
-    logger.warning("relay policy declaration returned HTTP %s; connector keeps prior/default policy", status)
-    return False
+    any_declared = False
+    for platform, _bot_id in relay_platform_identities():
+        policy = relay_relevance_policy(platform)
+        if policy is None:
+            # Nothing non-default to declare for this platform ⇒ the connector's
+            # quiet default already matches; don't write a redundant row.
+            continue
+        try:
+            status = _post_policy(policy_url=_policy_url(dial_url), token=token, policy=policy)
+        except Exception as exc:  # noqa: BLE001 - boot must survive a policy-declare failure
+            logger.warning(
+                "relay policy declaration failed for platform=%s (%s); continuing", platform, exc
+            )
+            continue
+        if status == 200:
+            any_declared = True
+            logger.info(
+                "relay policy declared (platform=%s require_address=%s free_scopes=%d allow_bots=%s)",
+                policy.get("platform"),
+                policy.get("requireAddress"),
+                len(policy.get("freeResponseScopes") or []),
+                policy.get("allowOtherBots"),
+            )
+        else:
+            logger.warning(
+                "relay policy declaration for platform=%s returned HTTP %s; connector keeps prior/default policy",
+                platform,
+                status,
+            )
+    return any_declared
 
 
 def register_relay_adapter(force: bool = False, url: Optional[str] = None) -> bool:
@@ -621,6 +734,12 @@ def _factory(config):
                 resolved_url,
                 platform,
                 bot_id,
+                # Phase 1.5: the full SET of (platform, bot_id) this gateway fronts.
+                # The transport sends one hello per identity (the connector
+                # accumulates them) and resolves the per-frame egress botId from
+                # this set. A single-platform deploy passes a 1-element list, so
+                # behaviour is byte-identical to before.
+                identities=relay_platform_identities(),
                 gateway_id=gateway_id,
                 upgrade_secret=upgrade_secret,
                 # Phase 5 §5.3: re-dial + re-handshake after an unexpected socket

gateway/relay/adapter.py

@@ -71,6 +71,15 @@ def __init__(
         # recipient's author binding; we re-attach this user_id as
         # metadata.user_id on the outbound action so it can. See _capture_scope.
         self._dm_user_by_chat: Dict[str, str] = {}
+        # chat_id -> the UNDERLYING platform (e.g. "discord", "telegram") this
+        # chat belongs to (Phase 1.5 multi-platform-per-agent). One relay adapter
+        # fronts N platforms on one WS; an outbound reply must egress through the
+        # platform the inbound came from. We remember it per chat_id from the
+        # inbound event's source.platform and stamp it on the OutboundFrame so the
+        # connector dispatches to the right sender. Empty for a single-platform
+        # gateway (the connector falls back to its session default). See
+        # _capture_scope / send.
+        self._platform_by_chat: Dict[str, str] = {}
         self.supports_code_blocks = descriptor.markdown_dialect not in ("", "plain")
         # Phase 7 Unit 7d-B: watches the transport for a terminal auth revocation
         # (a 4401 close after a successful handshake = the operator opted this
@@ -226,6 +235,17 @@ def _capture_scope(self, event) -> None:
             chat = getattr(src, "chat_id", None)
             if not chat:
                 return
+            # Phase 1.5: remember the underlying platform for this chat so the
+            # reply egresses through the right sender (one relay adapter fronts N
+            # platforms). source.platform is a Platform enum (e.g. Platform.DISCORD,
+            # mapped from the connector's "discord" by ws_transport _frame_to_event);
+            # record its string VALUE, skipping the generic RELAY fallback (a
+            # single-platform connector that didn't tag a concrete platform — the
+            # connector's session default handles egress then).
+            platform = getattr(src, "platform", None)
+            platform_value = getattr(platform, "value", platform)
+            if platform_value and platform_value != "relay":
+                self._platform_by_chat[str(chat)] = str(platform_value)
             guild = getattr(src, "guild_id", None)
             if guild:
                 self._scope_by_chat[str(chat)] = str(guild)
@@ -265,6 +285,17 @@ def _with_scope(self, chat_id: str, metadata: Optional[Dict[str, Any]]) -> Dict[
                 meta["user_id"] = dm_user
         return meta
 
+    def _platform_is_fronted(self, platform: str) -> bool:
+        """Whether ``platform`` is one of the platforms this gateway fronts over
+        the relay (Phase 1.5). Reads the transport's advertised identity set; used
+        to decide whether a follow-up's platform-prefixed `kind` names a real
+        fronted platform worth tagging on the frame (vs. leaving egress to the
+        session default). Safe when the transport is absent or single-identity."""
+        ids = getattr(self._transport, "_identities", None)
+        if not ids:
+            return False
+        return any(p == platform for p, _ in ids)
+
     async def on_interrupt(self, session_key: str, chat_id: str) -> None:
         """Bridge a connector-delivered /stop into the adapter's interrupt path.
 
@@ -439,7 +470,8 @@ async def send(
                 "content": content,
                 "reply_to": reply_to,
                 "metadata": self._with_scope(chat_id, metadata),
-            }
+            },
+            platform=self._platform_by_chat.get(str(chat_id)),
         )
         return SendResult(
             success=bool(result.get("success")),
@@ -470,14 +502,25 @@ async def send_follow_up(
         """
         if self._transport is None:
             return SendResult(success=False, error="no transport")
+        # Phase 1.5: the capability `kind` is platform-prefixed (e.g.
+        # "discord.interaction_token"), so derive the egress platform from it when
+        # it names one we front — that tags the OutboundFrame so a multi-platform
+        # gateway routes the follow-up through the right sender. Falls back to the
+        # session default (connector-side) when the prefix isn't a fronted platform.
+        follow_up_platform = None
+        if kind and "." in kind:
+            prefix = kind.split(".", 1)[0]
+            if self._platform_is_fronted(prefix):
+                follow_up_platform = prefix
         result = await self._transport.send_follow_up(
             {
                 "op": "follow_up",
                 "session_key": session_key,
                 "kind": kind,
                 "content": content,
                 "metadata": metadata or {},
-            }
+            },
+            platform=follow_up_platform,
         )
         return SendResult(
             success=bool(result.get("success")),