fix(relay): authorize the RESOLVED target; declines must not fall back

Review round 1 (independently confirmed by a second reviewer) found three
blockers. Two are fixed here; the third (B-2, Telegram @username) is a policy
decision left open deliberately.

B-1 — THE FIX CAUSED THE OUTAGE IT PREVENTED (tools/send_message_tool.py)

The P5(a) guard ran ABOVE Slack user->DM resolution, so it authorized the
internal pseudo-id `_parse_target_ref` emits (`user_name:ben`, `user:U...`).
Provenances only ever hold RESOLVED conversation ids, so a fully attested DM
was compared as a handle against a set of `D...` ids and refused:

  base  slack:@ben  SENT        head(before)  slack:@ben  REFUSED

Every Slack DM by handle was broken. Moved the guard below resolution; it now
authorizes the destination that is actually sent to, and the refusal names the
resolved id. Position is load-bearing, so it is commented as such and pinned:
reverting the move turns exactly the four new cases red.

B-3 — A DECLINE IS NOT A LANE FAILURE (gateway/run.py)

`_approval_send_outcome` had only sent/failed/ambiguous, so a connector
decline collapsed into `failed` — which is the cue to run the plain-text
fallback into the chat the connector had just refused. The adapter fix in the
previous commit improved the error STRING while user-visible behaviour stayed
identical to base; the commit message overstated it. Fixed properly:

  - new `declined` verdict, recognised via the shared `is_egress_decline`
    contract (not string sniffing at the call site)
  - exec-approval returns without the text fallback
  - slash-confirm suppresses the text reply AND clears the registration, so a
    card that never rendered cannot capture the user's next message

`send_clarify` was already correct (returns early inside the adapter).

MUTATIONS (production source; both directions)
  classifier never returns 'declined'        -> KILLED (4 cases)
  ALL failures classified as 'declined'      -> KILLED (2 cases)
  guard moved back above Slack resolution    -> KILLED (4 cases)
  decline CODE changed (review M05)          -> KILLED
  marker match made case-sensitive (M10)     -> KILLED

M05 was a tautology: the test asserted the imported constant against itself,
so changing the constant could not fail it. The wire contract is now pinned as
a literal, because the connector stamps that exact string and a one-sided
change is a silent cross-repo break.

REGRESSION CHECK: the 12 failures + 1 collection error in this test selection
are PRE-EXISTING cross-test contamination — the identical set fails at
7cf86188ac. Verified by diffing the failing sets: no new failures, 363 -> 374
passed.

NOT FIXED (deliberate): B-2, Telegram `@username`. The Bot API resolves handles
at send time, so there is no id to compare and no canonicalization exists yet.
That is a policy decision, not a code move.

gateway/run.py

@@ -1095,6 +1095,18 @@ def _approval_send_outcome(future, timeout: float) -> str:
         return "failed"
     if getattr(result, "success", False):
         return "sent"
+    # P5(b): a connector DECLINE is not a lane failure. The connector
+    # authorized the destination and refused it; re-sending the same content as
+    # plain text into that same chat is the exfiltration the egress guard
+    # exists to stop. `failed` is the cue to fall back, so a decline needs its
+    # own verdict — callers must surface it and send nothing further.
+    _err = getattr(result, "error", None)
+    if _err:
+        from gateway.relay.egress import is_egress_decline
+
+        if is_egress_decline({"success": False, "error": _err}):
+            logger.warning("Prompt send DECLINED by connector egress guard: %s", _err)
+            return "declined"
     logger.warning(
         "Prompt send failed: %s", getattr(result, "error", None) or "unknown error"
     )
@@ -6616,6 +6628,20 @@ class TurnRunner:
                             "stays armed for a late tap)"
                         )
                         return
+                    if _outcome == "declined":
+                        # P5(b): the connector AUTHORIZED this destination and
+                        # refused it. The text fallback below re-sends the same
+                        # content to the same chat, which would turn a refused
+                        # button card into a delivered plain-text one — the
+                        # exact leak the egress guard exists to stop. A decline
+                        # is definitive, so unlike `ambiguous` the registration
+                        # is torn down; unlike `failed`, nothing is re-sent.
+                        logger.warning(
+                            "Button-based approval DECLINED by the connector's "
+                            "egress guard — not falling back to text (the "
+                            "destination is not approved for this connection)"
+                        )
+                        return
                     logger.warning(
                         "Button-based approval failed (send returned error), falling back to text"
                     )
@@ -25173,6 +25199,31 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew
                 )
                 if button_result and getattr(button_result, "success", False):
                     used_buttons = True
+                elif button_result is not None:
+                    # P5(b): distinguish a connector egress DECLINE from a
+                    # lane failure. On a decline the connector refused this
+                    # destination, so returning `message` as the direct reply
+                    # would deliver the very content it refused, as text, to
+                    # the same chat. Suppress the fallback and tear down the
+                    # registration — no card rendered, so a later reply must
+                    # not be captured as an answer to an invisible prompt.
+                    _confirm_err = getattr(button_result, "error", None)
+                    if _confirm_err:
+                        from gateway.relay.egress import is_egress_decline
+
+                        if is_egress_decline(
+                            {"success": False, "error": _confirm_err}
+                        ):
+                            logger.warning(
+                                "slash-confirm DECLINED by the connector's egress "
+                                "guard for %s on %s — suppressing the text "
+                                "fallback: %s",
+                                command,
+                                source.platform,
+                                _confirm_err,
+                            )
+                            _slash_confirm_mod.clear(session_key)
+                            return None
             except Exception as exc:
                 logger.debug(
                     "send_slash_confirm failed for %s on %s: %s",

tests/gateway/relay/test_relay_egress_declines.py

@@ -119,9 +119,30 @@ def relay():
 # ── the classifier itself ────────────────────────────────────────────────
 
 def test_decline_is_recognised_by_code_and_by_uniform_text():
+    # The wire contract, pinned as a LITERAL. Asserting against the imported
+    # constant is a tautology — it cannot fail when the constant changes, and
+    # review mutation M05 survived exactly there. The connector stamps this
+    # exact string (gateway-gateway routedEgressGuard); changing either side
+    # alone is a silent cross-repo break, so the literal is the point.
+    assert EGRESS_DECLINE_CODE == "egress_declined"
+    assert is_egress_decline({"success": False, "code": "egress_declined"}) is True
     assert is_egress_decline({"success": False, "code": EGRESS_DECLINE_CODE}) is True
+
     assert is_egress_decline(DECLINE) is True
 
+    # M10: the `.lower()` in is_egress_decline was untested, so making the
+    # marker match case-SENSITIVE survived — a connector emitting "Egress
+    # declined:" would silently stop being classified as a decline and start
+    # falling back into the refused chat. Every fixture happened to be
+    # lowercase, which is what hid it.
+    for variant in (
+        "Discord Egress Declined: target is not approved",
+        "EGRESS DECLINED: target is not approved",
+        "discord EGRESS declined: target is not approved",
+    ):
+        assert is_egress_decline({"success": False, "error": variant}) is True, variant
+
+
 
 def test_an_ambiguous_failure_is_not_a_decline():
     """A lost ack may well have been APPLIED — it is a transport outcome.

tests/gateway/test_prompt_decline_no_fallback.py

@@ -0,0 +1,117 @@
+"""P5(b) at the CALLER: a connector egress decline must not fall back.
+
+The adapter-level tests prove the decline REACHES the caller. These prove the
+caller ACTS on it. Review round 1 blocker B-3: `_approval_send_outcome` had
+only sent/failed/ambiguous, so a decline collapsed into `failed` — which is
+precisely the cue to run the plain-text fallback into the chat the connector
+had just refused. The adapter fix improved the error STRING while the
+user-visible behaviour stayed identical to base.
+
+These tests drive the real `gateway.run` classifier and the real
+`tools.slash_confirm` registry; only the SendResult (the connector's answer)
+is constructed.
+"""
+
+from __future__ import annotations
+
+import concurrent.futures
+from types import SimpleNamespace
+
+import pytest
+
+DECLINE_ERROR = (
+    "discord egress declined: target is not an approved destination for this connection"
+)
+LANE_ERROR = "relay prompt op unavailable"
+
+
+def _future(result):
+    fut: concurrent.futures.Future = concurrent.futures.Future()
+    fut.set_result(result)
+    return fut
+
+
+def _result(*, success: bool, error: str | None = None):
+    return SimpleNamespace(success=success, error=error, message_id=None)
+
+
+# ── the classifier ──────────────────────────────────────────────────────────
+
+
+def test_decline_is_not_classified_as_failed():
+    """`failed` is the fallback cue; a decline must not wear it."""
+    from gateway.run import _approval_send_outcome
+
+    outcome = _approval_send_outcome(
+        _future(_result(success=False, error=DECLINE_ERROR)), timeout=5
+    )
+    assert outcome == "declined"
+
+
+def test_genuine_lane_failure_still_falls_back():
+    """The guard must not swallow real failures: those still re-ask."""
+    from gateway.run import _approval_send_outcome
+
+    outcome = _approval_send_outcome(
+        _future(_result(success=False, error=LANE_ERROR)), timeout=5
+    )
+    assert outcome == "failed"
+
+
+def test_success_and_timeout_verdicts_unchanged():
+    """No collateral change to the two settled verdicts."""
+    from gateway.run import _approval_send_outcome
+
+    assert (
+        _approval_send_outcome(_future(_result(success=True)), timeout=5) == "sent"
+    )
+
+    pending: concurrent.futures.Future = concurrent.futures.Future()
+    assert _approval_send_outcome(pending, timeout=0.05) == "ambiguous"
+
+
+@pytest.mark.parametrize(
+    "error",
+    [
+        DECLINE_ERROR,
+        "slack egress declined: destination not permitted",
+        # Case-insensitivity is part of the contract (`.lower()` in
+        # is_egress_decline), so a connector that capitalises still classifies.
+        "WhatsApp Egress Declined: target refused",
+    ],
+)
+def test_decline_recognised_across_lanes(error):
+    """The verdict follows the decline CONTRACT, not one lane's wording.
+
+    The contract is `EGRESS_DECLINE_MARKER` ("egress declined:") or a
+    structured `code`; an invented sentence like "EGRESS_DECLINED: ..." is NOT
+    a decline and must not be treated as one. My first version of this test
+    asserted that invented form and failed — the test was wrong, not the code.
+    """
+    from gateway.run import _approval_send_outcome
+
+    assert (
+        _approval_send_outcome(_future(_result(success=False, error=error)), timeout=5)
+        == "declined"
+    )
+
+
+def test_non_decline_error_text_is_not_laundered_into_declined():
+    """Fail-closed the other way: only the real contract yields `declined`.
+
+    Without this, a permissive marker check would silence genuine failures —
+    turning a lane outage into a silent no-fallback.
+    """
+    from gateway.run import _approval_send_outcome
+
+    for error in (
+        "connection reset by peer",
+        "declined",  # bare word, not the marker sentence
+        "egress declined",  # no colon: not the uniform marker
+    ):
+        assert (
+            _approval_send_outcome(
+                _future(_result(success=False, error=error)), timeout=5
+            )
+            == "failed"
+        ), error

tests/tools/test_send_message_relay_target_authz.py

@@ -198,3 +198,124 @@ def test_react_refuses_an_arbitrary_relay_target(relay_env):
             "send_message(action='list') to see the targets it can reach."
         )
     }
+
+# ── B-1: the guard must authorize the RESOLVED destination ──────────────────
+#
+# Slack `@handle` / `U...` targets are internal PSEUDO-ids
+# (`user_name:ben`, `user:U...`) until `_resolve_slack_user_target` opens the
+# DM and returns the real `D...` conversation. Provenances only ever hold
+# resolved ids, so authorizing the pseudo-id compares a handle against a set
+# of channel ids and refuses every Slack DM — an OUTAGE caused by a security
+# fix. Review round 1 found this; reproduced before fixing.
+#
+# These tests are the falsifiable floor for the guard's POSITION: they pass
+# only while authorization happens AFTER resolution.
+
+SLACK_DM = "D01234567AB"
+SLACK_USER = "U01234567AB"
+
+
+@pytest.fixture
+def slack_relay_env(tmp_path, monkeypatch):
+    """A relay-fronted Slack gateway whose attested destination is a DM id."""
+    import gateway.channel_directory as cd
+
+    monkeypatch.setenv("GATEWAY_RELAY_URL", "wss://connector.example/relay")
+    monkeypatch.setenv("GATEWAY_RELAY_PLATFORMS", "slack")
+    monkeypatch.setenv("GATEWAY_RELAY_BOT_IDS", json.dumps({"slack": {"botId": "b1"}}))
+
+    directory = tmp_path / "channel_directory.json"
+    directory.write_text(
+        json.dumps(
+            {
+                "updated_at": None,
+                # The DM conversation id — what resolution produces, and the
+                # only form any provenance ever stores.
+                "platforms": {"slack": [{"id": SLACK_DM, "name": "ben", "type": "im"}]},
+            }
+        ),
+        encoding="utf-8",
+    )
+    monkeypatch.setattr(cd, "DIRECTORY_PATH", directory)
+    monkeypatch.setattr(cd, "CHANNEL_ALIASES_PATH", tmp_path / "channel_aliases.json")
+    monkeypatch.setattr(cd, "_build_from_sessions", lambda _platform: [])
+    return directory
+
+
+def _send_slack(target: str, sent, *, resolves_to: str | None = SLACK_DM):
+    """Invoke the real tool with the REAL Slack resolution step in the path.
+
+    Only `conversations.open` is faked (a network call). The ordering of the
+    guard against the resolver is production's.
+    """
+    import asyncio
+    from types import SimpleNamespace
+    from unittest.mock import patch
+
+    slack_cfg = SimpleNamespace(enabled=True, token="xoxb-t", extra={})
+    config = SimpleNamespace(
+        platforms={Platform.SLACK: slack_cfg},
+        get_home_channel=lambda _p: SimpleNamespace(chat_id=SLACK_DM),
+    )
+
+    async def _record(platform, pconfig, chat_id, message, **kwargs):
+        sent.append(chat_id)
+        return {"success": True, "message_id": "m1"}
+
+    async def _resolve(_token, target_ref):
+        # Stands in for the Slack API call only; returns what production's
+        # resolver returns — the opened DM channel id.
+        return (resolves_to, None)
+
+    with patch("gateway.config.load_gateway_config", return_value=config), patch(
+        "tools.interrupt.is_interrupted", return_value=False
+    ), patch("model_tools._run_async", side_effect=lambda c: asyncio.run(c)), patch(
+        "tools.send_message_tool._send_to_platform", side_effect=_record
+    ), patch(
+        "tools.send_message_tool._resolve_slack_user_target", side_effect=_resolve
+    ), patch(
+        "gateway.mirror.mirror_to_session", return_value=False
+    ):
+        return json.loads(
+            send_message_tool({"action": "send", "target": target, "message": "hello"})
+        )
+
+
+@pytest.mark.parametrize(
+    "target",
+    [f"slack:@ben", f"slack:{SLACK_USER}", f"slack:<@{SLACK_USER}>"],
+)
+def test_slack_user_targets_resolve_then_authorize(slack_relay_env, target):
+    """An attested DM must SEND regardless of which alias names it.
+
+    Fails if the guard runs before resolution: the pseudo-id
+    (`user_name:ben` / `user:U...`) is not in any provenance, so the send is
+    refused and `sent` stays empty.
+    """
+    sent: list[str] = []
+    result = _send_slack(target, sent)
+
+    assert result == {"success": True, "message_id": "m1"}
+    # The whole observable: it egressed, and to the RESOLVED destination.
+    assert sent == [SLACK_DM]
+
+
+def test_slack_user_target_resolving_to_unattested_dm_is_refused(slack_relay_env):
+    """Moving the guard must not disable it.
+
+    A handle that resolves to a DM this gateway cannot attest is still
+    refused — and the refusal names the RESOLVED id, which is the destination
+    that was actually authorized.
+    """
+    sent: list[str] = []
+    unattested = "D99999999XX"
+    result = _send_slack("slack:@stranger", sent, resolves_to=unattested)
+
+    assert result == {
+        "error": (
+            f"Refusing to send to unattested relay target 'slack:{unattested}': "
+            "this gateway has no record of that destination. Use "
+            "send_message(action='list') to see the targets it can reach."
+        )
+    }
+    assert sent == []

tools/send_message_tool.py

@@ -485,17 +485,6 @@ def _handle_send(args):
                 f"or set a home channel via: hermes config set {home_env} <channel_id>"
             )
 
-    # P5(a): a relay-routed destination must be one this gateway can show a
-    # provenance for. The `target` parameter is free-form, so without this a
-    # model could name ANY chat id and the gateway would dutifully emit an
-    # outbound frame for it — authenticating the sender while never
-    # authorizing the destination. Applies to the generic `relay` plane and to
-    # connector-fronted platforms with no live native adapter; every other
-    # platform keeps its adapter's own authorization unchanged.
-    _relay_denial = _authorize_relay_target(platform_name, chat_id)
-    if _relay_denial:
-        return tool_error(_relay_denial)
-
     duplicate_skip = _maybe_skip_cron_duplicate_send(platform_name, chat_id, thread_id)
     if duplicate_skip:
         return json.dumps(duplicate_skip)
@@ -517,6 +506,26 @@ def _handle_send(args):
                 return json.dumps(_resolve_err)
             chat_id = _resolved
 
+    # P5(a): a relay-routed destination must be one this gateway can show a
+    # provenance for. The `target` parameter is free-form, so without this a
+    # model could name ANY chat id and the gateway would dutifully emit an
+    # outbound frame for it — authenticating the sender while never
+    # authorizing the destination. Applies to the generic `relay` plane and to
+    # connector-fronted platforms with no live native adapter; every other
+    # platform keeps its adapter's own authorization unchanged.
+    #
+    # POSITION IS LOAD-BEARING — this must stay BELOW Slack user→DM resolution.
+    # `_parse_target_ref` emits internal pseudo-ids (`user_name:ben`,
+    # `user:U...`) that no provenance can ever contain, because provenances
+    # record RESOLVED conversation ids. Authorizing above the resolver compared
+    # a handle against a set of `D...` ids and refused every Slack DM — a fix
+    # that caused the outage it was meant to prevent. Pinned by
+    # test_slack_user_targets_resolve_then_authorize; moving this call back up
+    # turns those cases red.
+    _relay_denial = _authorize_relay_target(platform_name, chat_id)
+    if _relay_denial:
+        return tool_error(_relay_denial)
+
     try:
         from model_tools import _run_async
         send_kwargs = {