feat(gateway): token-less follow_up outbound op (A2 capability action)

The relay outbound surface had send/edit/typing but no way to act on a
SHARED-identity capability (e.g. a Discord interaction follow-up token,
~15min) that the connector captured + stripped at the edge. Under A2 that
credential never reaches the gateway, so the gateway can't just 'send with
the token' — it needs a semantic op naming the session it's already in.

Adds the follow_up op end to end on the gateway side:
- RelayTransport.send_follow_up(action): protocol method. Action carries
  op='follow_up' + session_key + kind + content (+ metadata) and NO token.
- RelayAdapter.send_follow_up(session_key, kind, content, metadata): builds
  that action and returns a SendResult. The connector resolves the real
  capability (its resolveOutboundCapability), enforces the tenant match so
  tenant B can't wield tenant A's capability, and egresses; success=False
  when the capability is absent/expired/mismatched (nothing to retry — a
  leaked gateway holds zero capability material).
- StubConnector records follow_ups + a canned next_follow_up_result.

Tests: round-trips without a token; the wire action carries only session
refs (no credential value field — the 'kind' string is a type ref, not the
secret); failure surfaces when the connector can't resolve; no-transport
fails cleanly. 55 passed. §4 doc entry follows in the contract-rewrite commit.

gateway/relay/adapter.py

@@ -126,3 +126,35 @@ class RelayAdapter(BasePlatformAdapter):
         if self._transport is None:
             return {"name": chat_id, "type": "dm"}
         return await self._transport.get_chat_info(chat_id)
+
+    async def send_follow_up(
+        self,
+        session_key: str,
+        kind: str,
+        content: str,
+        metadata: Optional[Dict[str, Any]] = None,
+    ) -> SendResult:
+        """Send via a shared-identity capability bound to a session (A2 outbound).
+
+        The gateway never holds the credential: it names the session it is
+        already in plus the capability ``kind``, and the connector resolves the
+        real value from its vault and egresses (enforcing the tenant match). Used
+        e.g. to post a Discord interaction follow-up as the shared bot without
+        the token ever reaching the gateway. See RelayTransport.send_follow_up.
+        """
+        if self._transport is None:
+            return SendResult(success=False, error="no transport")
+        result = await self._transport.send_follow_up(
+            {
+                "op": "follow_up",
+                "session_key": session_key,
+                "kind": kind,
+                "content": content,
+                "metadata": metadata or {},
+            }
+        )
+        return SendResult(
+            success=bool(result.get("success")),
+            message_id=result.get("message_id"),
+            error=result.get("error"),
+        )

gateway/relay/transport.py

@@ -73,3 +73,29 @@ class RelayTransport(Protocol):
         (handled in Task 1.4).
         """
         ...
+
+    async def send_follow_up(self, action: Dict[str, Any]) -> Dict[str, Any]:
+        """Act on a shared-identity capability bound to a session (A2 outbound).
+
+        Some platforms hand the connector a credential that acts on the SHARED
+        bot identity (e.g. a Discord interaction follow-up token, valid ~15min).
+        Under A2 that credential NEVER reaches the gateway — the connector
+        stripped it at the edge and bound it in its capability vault keyed by
+        the session. To use it, the gateway issues a SEMANTIC action against the
+        session it is already in; it never names or holds a token.
+
+        The action dict carries:
+          ``op``          == ``"follow_up"``
+          ``session_key`` the session whose bound capability to wield
+          ``kind``        the capability kind (e.g. ``"discord.interaction_token"``)
+          ``content``     the message content to send via that capability
+          ``metadata?``   optional extras
+
+        The connector resolves the real capability (``resolveOutboundCapability``
+        on its side), enforces the tenant match (tenant B can never wield tenant
+        A's capability), and egresses. Returns ``{success, message_id?, error?}``;
+        ``success`` is False when the capability is absent/expired or the tenant
+        doesn't match — the gateway then has nothing to retry with (by design: a
+        leaked gateway holds zero capability material).
+        """
+        ...

tests/gateway/relay/stub_connector.py

@@ -29,9 +29,14 @@ class StubConnector:
         self.connected = False
         self.sent: List[Dict[str, Any]] = []
         self.interrupts: List[Dict[str, Any]] = []
+        self.follow_ups: List[Dict[str, Any]] = []
         self.chat_info: Dict[str, Dict[str, Any]] = {}
         # Canned result for the next send_outbound (override per-test).
         self.next_send_result: Dict[str, Any] = {"success": True, "message_id": "m1"}
+        # Canned result for the next send_follow_up (override per-test). Default
+        # mimics a resolved capability egress; set success=False to simulate an
+        # absent/expired capability or a tenant mismatch on the connector side.
+        self.next_follow_up_result: Dict[str, Any] = {"success": True, "message_id": "f1"}
 
     async def connect(self) -> bool:
         self.connected = True
@@ -58,6 +63,10 @@ class StubConnector:
     async def send_interrupt(self, session_key: str, reason: Optional[str] = None) -> None:
         self.interrupts.append({"session_key": session_key, "reason": reason})
 
+    async def send_follow_up(self, action: Dict[str, Any]) -> Dict[str, Any]:
+        self.follow_ups.append(action)
+        return dict(self.next_follow_up_result)
+
     # ── test driver ──────────────────────────────────────────────────────
     async def push_inbound(self, event: MessageEvent) -> None:
         """Simulate the connector delivering a normalized inbound event."""

tests/gateway/relay/test_relay_follow_up.py

@@ -0,0 +1,117 @@
+"""A2 outbound capability action: the token-less ``follow_up`` op.
+
+Proves the gateway can act on a shared-identity capability (e.g. a Discord
+interaction follow-up token) WITHOUT ever holding the credential: it names the
+session it is in plus the capability ``kind``, and the connector resolves the
+real value from its vault and egresses. See gateway/relay/transport.py
+(send_follow_up) and docs/relay-connector-contract.md §4.
+
+The gateway side is what's exercised here (against the stub connector); the
+connector's resolve + tenant-match enforcement lives in the connector repo
+(resolveOutboundCapability). The key gateway-side guarantees:
+  - the wire action carries NO token (only session_key + kind + content),
+  - success/failure surfaces from the connector's resolve result,
+  - a failed resolve (absent/expired/tenant mismatch) returns success=False
+    with nothing for the gateway to retry with.
+"""
+
+from __future__ import annotations
+
+import pytest
+
+from gateway.config import PlatformConfig
+from gateway.relay.adapter import RelayAdapter
+from gateway.relay.descriptor import CONTRACT_VERSION, CapabilityDescriptor
+
+from tests.gateway.relay.stub_connector import StubConnector
+
+
+def _discord_descriptor() -> CapabilityDescriptor:
+    return CapabilityDescriptor(
+        contract_version=CONTRACT_VERSION,
+        platform="discord",
+        label="Discord",
+        max_message_length=2000,
+        supports_draft_streaming=False,
+        supports_edit=True,
+        supports_threads=True,
+        markdown_dialect="discord",
+        len_unit="chars",
+    )
+
+
+@pytest.fixture
+def wired():
+    stub = StubConnector(_discord_descriptor())
+    adapter = RelayAdapter(PlatformConfig(), _discord_descriptor(), transport=stub)
+    return adapter, stub
+
+
+@pytest.mark.asyncio
+async def test_follow_up_round_trips_without_a_token(wired):
+    adapter, stub = wired
+    await adapter.connect()
+    stub.next_follow_up_result = {"success": True, "message_id": "fu-7"}
+
+    result = await adapter.send_follow_up(
+        session_key="agent:main:discord:group:chanA:userX",
+        kind="discord.interaction_token",
+        content="here is your follow-up",
+    )
+
+    assert result.success is True
+    assert result.message_id == "fu-7"
+    assert len(stub.follow_ups) == 1
+    action = stub.follow_ups[0]
+    assert action["op"] == "follow_up"
+    assert action["session_key"] == "agent:main:discord:group:chanA:userX"
+    assert action["kind"] == "discord.interaction_token"
+    assert action["content"] == "here is your follow-up"
+
+
+@pytest.mark.asyncio
+async def test_follow_up_wire_action_carries_no_credential(wired):
+    """The action dict must carry only session refs — no credential VALUE.
+
+    Note the capability ``kind`` legitimately names the credential type
+    (e.g. ``"discord.interaction_token"``) — that's a reference, not the secret.
+    The guarantee is structural: the action has exactly the token-less semantic
+    fields, and no field holds an actual credential value.
+    """
+    adapter, stub = wired
+    await adapter.connect()
+    await adapter.send_follow_up(
+        session_key="sess-1", kind="discord.interaction_token", content="x", metadata={"a": 1}
+    )
+    action = stub.follow_ups[0]
+    # Exactly the token-less semantic fields (+ metadata); no value/secret field.
+    assert set(action.keys()) == {"op", "session_key", "kind", "content", "metadata"}
+    # No field NAMES a credential carrier (the kind string is a type ref, allowed).
+    assert "value" not in action
+    assert "token" not in action
+    assert "secret" not in action
+    assert "credential" not in action
+
+
+@pytest.mark.asyncio
+async def test_follow_up_failure_surfaces_when_capability_unresolvable(wired):
+    """Connector couldn't resolve (absent/expired/tenant mismatch) -> success=False."""
+    adapter, stub = wired
+    await adapter.connect()
+    stub.next_follow_up_result = {"success": False, "error": "capability absent or tenant mismatch"}
+
+    result = await adapter.send_follow_up(
+        session_key="sess-1", kind="discord.interaction_token", content="x"
+    )
+
+    assert result.success is False
+    assert result.message_id is None
+    assert "tenant mismatch" in (result.error or "")
+
+
+@pytest.mark.asyncio
+async def test_follow_up_without_transport_fails_cleanly():
+    adapter = RelayAdapter(PlatformConfig(), _discord_descriptor(), transport=None)
+    result = await adapter.send_follow_up(session_key="s", kind="k", content="c")
+    assert result.success is False
+    assert result.error == "no transport"