feat(plugins): add standalone_sender_fn for out-of-process cron delivery

Plugin platforms (IRC, Teams, Google Chat) currently fail with
`No live adapter for platform '<name>'` when a `deliver=<plugin>` cron
job runs in a separate process from the gateway, even though the
platforms are eligible cron targets via `cron_deliver_env_var` (added
in #21306). Built-in platforms (Telegram, Discord, Slack, etc.) use
direct REST helpers in `tools/send_message_tool.py` so cron can deliver
without holding the gateway in the same process; plugin platforms
historically depended on `_gateway_runner_ref()` which returns `None`
out of process.

This change adds an optional `standalone_sender_fn` field to
`PlatformEntry` so plugins can register an ephemeral send path that
opens its own connection, sends, and closes without needing the live
adapter. The dispatch site in `_send_via_adapter` falls through to the
hook when the gateway runner is unavailable, with a descriptive error
when neither path applies. The hook is optional, so existing plugins
are unaffected.

Reference migrations land in the same change for IRC, Teams, and
Google Chat, exercising the hook across stdlib (asyncio + IRC protocol),
Bot Framework OAuth client_credentials, and Google service-account
flows respectively.

Security hardening on the new code paths:
* IRC: control-character stripping on chat_id and message body to
  block CRLF command injection; bounded nick-collision retries; JOIN
  before PRIVMSG so channels with the default `+n` mode accept the
  delivery.
* Teams: TEAMS_SERVICE_URL validated against an allowlist of known
  Bot Framework hosts (`smba.trafficmanager.net`,
  `smba.infra.gov.teams.microsoft.us`) to block SSRF; chat_id and
  tenant_id constrained to the documented Bot Framework character set;
  per-request timeouts so a slow STS endpoint cannot starve the
  activity POST.
* Google Chat: chat_id and thread_id validated against strict
  resource-name regexes; service-account refresh wrapped in
  `asyncio.wait_for` so a hung token endpoint cannot stall the
  scheduler.

Test coverage: 20 new tests covering happy path, missing-config errors,
network failure modes, and each defensive validation. Existing tests
unchanged. `bash scripts/run_tests.sh tests/tools/test_send_message_tool.py
tests/gateway/test_irc_adapter.py tests/gateway/test_teams.py
tests/gateway/test_google_chat.py` reports 341 passed, 0 regressions.

Documentation: new "Out-of-process cron delivery" section in
website/docs/developer-guide/adding-platform-adapters.md and an entry
in gateway/platforms/ADDING_A_PLATFORM.md naming the hook.

tests/gateway/test_google_chat.py

@@ -2696,3 +2696,173 @@ class TestCronSchedulerRegistry:
         from cron.scheduler import _resolve_home_env_var
 
         assert _resolve_home_env_var("google_chat") == "GOOGLE_CHAT_HOME_CHANNEL"
+
+
+# ── _standalone_send (out-of-process cron delivery) ──────────────────────
+
+
+class _FakeAiohttpResponse:
+    def __init__(self, status: int, payload, text_body: str = ""):
+        self.status = status
+        self._payload = payload
+        self._text = text_body or (str(payload) if payload is not None else "")
+
+    async def json(self):
+        return self._payload
+
+    async def text(self):
+        return self._text
+
+    async def __aenter__(self):
+        return self
+
+    async def __aexit__(self, exc_type, exc, tb):
+        return None
+
+
+class _FakeAiohttpSession:
+    def __init__(self, scripts):
+        self._scripts = list(scripts)
+        self.calls: list[tuple[str, dict]] = []
+
+    async def __aenter__(self):
+        return self
+
+    async def __aexit__(self, exc_type, exc, tb):
+        return None
+
+    def post(self, url, **kwargs):
+        self.calls.append((url, kwargs))
+        if not self._scripts:
+            raise AssertionError(f"No scripted response for POST {url}")
+        return self._scripts.pop(0)
+
+
+def _install_fake_aiohttp(monkeypatch, session):
+    fake_aiohttp = types.SimpleNamespace(
+        ClientSession=lambda timeout=None: session,
+        ClientTimeout=lambda total=None: None,
+    )
+    monkeypatch.setitem(sys.modules, "aiohttp", fake_aiohttp)
+
+
+def _install_fake_google_auth_transport(monkeypatch):
+    fake_request_module = types.SimpleNamespace(Request=lambda: object())
+    monkeypatch.setitem(sys.modules, "google.auth.transport", types.SimpleNamespace(requests=fake_request_module))
+    monkeypatch.setitem(sys.modules, "google.auth.transport.requests", fake_request_module)
+
+
+class TestGoogleChatStandaloneSend:
+
+    @pytest.mark.asyncio
+    async def test_standalone_send_refreshes_token_and_posts_message(
+        self, monkeypatch, tmp_path
+    ):
+        sa_file = tmp_path / "sa.json"
+        sa_file.write_text(json.dumps({
+            "type": "service_account",
+            "client_email": "bot@example.iam.gserviceaccount.com",
+            "private_key": "fake",
+            "token_uri": "https://example/token",
+        }))
+        monkeypatch.setenv("GOOGLE_CHAT_SERVICE_ACCOUNT_JSON", str(sa_file))
+
+        fake_creds = MagicMock()
+        fake_creds.token = "the-token"
+        fake_creds.refresh = MagicMock(return_value=None)
+
+        original = _gc_mod.service_account.Credentials.from_service_account_info
+        _gc_mod.service_account.Credentials.from_service_account_info = MagicMock(
+            return_value=fake_creds
+        )
+        try:
+            _install_fake_google_auth_transport(monkeypatch)
+            send_resp = _FakeAiohttpResponse(200, {"name": "spaces/AAA/messages/MMM"})
+            session = _FakeAiohttpSession([send_resp])
+            _install_fake_aiohttp(monkeypatch, session)
+
+            result = await _gc_mod._standalone_send(
+                PlatformConfig(enabled=True, extra={}),
+                "spaces/AAAA-BBBB",
+                "hello cron",
+            )
+        finally:
+            _gc_mod.service_account.Credentials.from_service_account_info = original
+
+        assert result == {
+            "success": True,
+            "message_id": "spaces/AAA/messages/MMM",
+        }
+        fake_creds.refresh.assert_called_once()
+        assert len(session.calls) == 1
+        url, kwargs = session.calls[0]
+        assert url == "https://chat.googleapis.com/v1/spaces/AAAA-BBBB/messages"
+        assert kwargs["headers"]["Authorization"] == "Bearer the-token"
+        assert kwargs["json"] == {"text": "hello cron"}
+
+    @pytest.mark.asyncio
+    async def test_standalone_send_returns_error_on_invalid_chat_id(self, monkeypatch):
+        monkeypatch.delenv("GOOGLE_CHAT_SERVICE_ACCOUNT_JSON", raising=False)
+        result = await _gc_mod._standalone_send(
+            PlatformConfig(enabled=True, extra={}),
+            "not-a-resource-name",
+            "hi",
+        )
+        assert "error" in result
+        assert "spaces/" in result["error"] or "users/" in result["error"]
+
+    @pytest.mark.asyncio
+    async def test_standalone_send_propagates_api_failure(self, monkeypatch, tmp_path):
+        sa_file = tmp_path / "sa.json"
+        sa_file.write_text(json.dumps({
+            "type": "service_account",
+            "client_email": "bot@example.iam.gserviceaccount.com",
+            "private_key": "fake",
+            "token_uri": "https://example/token",
+        }))
+        monkeypatch.setenv("GOOGLE_CHAT_SERVICE_ACCOUNT_JSON", str(sa_file))
+
+        fake_creds = MagicMock()
+        fake_creds.token = "the-token"
+        fake_creds.refresh = MagicMock(return_value=None)
+
+        original = _gc_mod.service_account.Credentials.from_service_account_info
+        _gc_mod.service_account.Credentials.from_service_account_info = MagicMock(
+            return_value=fake_creds
+        )
+        try:
+            _install_fake_google_auth_transport(monkeypatch)
+            send_resp = _FakeAiohttpResponse(
+                403,
+                {"error": {"code": 403, "message": "forbidden"}},
+                text_body='{"error":{"code":403,"message":"forbidden"}}',
+            )
+            session = _FakeAiohttpSession([send_resp])
+            _install_fake_aiohttp(monkeypatch, session)
+
+            result = await _gc_mod._standalone_send(
+                PlatformConfig(enabled=True, extra={}),
+                "spaces/AAAA-BBBB",
+                "hi",
+            )
+        finally:
+            _gc_mod.service_account.Credentials.from_service_account_info = original
+
+        assert "error" in result
+        assert "403" in result["error"]
+
+    @pytest.mark.asyncio
+    async def test_standalone_send_rejects_chat_id_with_path_traversal(self, monkeypatch):
+        monkeypatch.delenv("GOOGLE_CHAT_SERVICE_ACCOUNT_JSON", raising=False)
+
+        # Attempt to inject extra path segments after the prefix passes the
+        # startswith check.  The strict regex must reject this.
+        result = await _gc_mod._standalone_send(
+            PlatformConfig(enabled=True, extra={}),
+            "spaces/AAAA/messages?messageReplyOption=REPLY_MESSAGE_FALLBACK_TO_NEW_THREAD",
+            "hi",
+        )
+
+        assert "error" in result
+        # The error names the expected resource shape so plugin authors can self-correct
+        assert "spaces/" in result["error"] or "users/" in result["error"]

tests/gateway/test_irc_adapter.py

@@ -20,6 +20,7 @@ IRCAdapter = _irc_mod.IRCAdapter
 check_requirements = _irc_mod.check_requirements
 validate_config = _irc_mod.validate_config
 register = _irc_mod.register
+_standalone_send = _irc_mod._standalone_send
 
 
 class TestIRCProtocolHelpers:
@@ -500,3 +501,224 @@ class TestIRCPluginRegistration:
         ctx.register_platform.assert_called_once()
         call_kwargs = ctx.register_platform.call_args
         assert call_kwargs[1]["name"] == "irc" or call_kwargs[0][0] == "irc" if call_kwargs[0] else call_kwargs[1]["name"] == "irc"
+
+
+# ── _standalone_send (out-of-process cron delivery) ──────────────────────
+
+
+class _FakeIRCConnection:
+    """A scripted reader/writer pair used to simulate an IRC server.
+
+    Construct with the lines the server should respond with (already
+    framed by ``\\r\\n``).  Captures every line written by the client so
+    tests can assert NICK/USER/PRIVMSG/QUIT order.
+    """
+
+    def __init__(self, scripted_lines):
+        self.writes: list[bytes] = []
+        self._closed = False
+        self._scripted = list(scripted_lines)
+        self._buffer = b""
+
+    # writer side ────────────────────────────────────────────────────
+    def write(self, data: bytes) -> None:
+        self.writes.append(data)
+
+    async def drain(self) -> None:
+        return None
+
+    def close(self) -> None:
+        self._closed = True
+
+    async def wait_closed(self) -> None:
+        return None
+
+    def is_closing(self) -> bool:
+        return self._closed
+
+    # reader side ────────────────────────────────────────────────────
+    async def readuntil(self, separator: bytes = b"\r\n") -> bytes:
+        if not self._scripted:
+            raise asyncio.IncompleteReadError(b"", None)
+        line = self._scripted.pop(0)
+        if not line.endswith(b"\r\n"):
+            line = line + b"\r\n"
+        return line
+
+    async def read(self, n: int = -1) -> bytes:
+        return b""
+
+
+class TestIRCStandaloneSend:
+
+    @pytest.mark.asyncio
+    async def test_standalone_send_completes_handshake_and_sends_privmsg(self, monkeypatch):
+        from gateway.config import PlatformConfig
+
+        monkeypatch.setenv("IRC_SERVER", "irc.test.net")
+        monkeypatch.setenv("IRC_CHANNEL", "#cron")
+        monkeypatch.setenv("IRC_NICKNAME", "hermesbot")
+        monkeypatch.setenv("IRC_USE_TLS", "false")
+
+        # Server greets us with 001 RPL_WELCOME, then nothing for QUIT drain.
+        conn = _FakeIRCConnection([b":server 001 hermesbot-cron :Welcome"])
+
+        async def _fake_open(host, port, **kwargs):
+            return conn, conn  # reader and writer share the same fake
+
+        monkeypatch.setattr(_irc_mod.asyncio, "open_connection", _fake_open)
+
+        result = await _standalone_send(
+            PlatformConfig(enabled=True, extra={}),
+            "#cron",
+            "hello from cron",
+        )
+
+        assert result["success"] is True
+        assert "message_id" in result
+
+        sent_lines = b"".join(conn.writes).decode("utf-8").splitlines()
+        # NICK uses the cron-suffixed identity to avoid colliding with the
+        # long-running gateway adapter that may already hold the nickname.
+        assert any(line.startswith("NICK hermesbot-cron") for line in sent_lines)
+        assert any(line.startswith("USER hermesbot-cron 0 * :Hermes Agent (cron)")
+                   for line in sent_lines)
+        assert any(line == "PRIVMSG #cron :hello from cron" for line in sent_lines)
+        assert any(line.startswith("QUIT ") for line in sent_lines)
+
+    @pytest.mark.asyncio
+    async def test_standalone_send_returns_error_when_unconfigured(self, monkeypatch):
+        from gateway.config import PlatformConfig
+
+        for var in ("IRC_SERVER", "IRC_CHANNEL"):
+            monkeypatch.delenv(var, raising=False)
+
+        result = await _standalone_send(
+            PlatformConfig(enabled=True, extra={}),
+            "",
+            "hi",
+        )
+
+        assert "error" in result
+        assert "IRC_SERVER" in result["error"] or "IRC_CHANNEL" in result["error"]
+
+    @pytest.mark.asyncio
+    async def test_standalone_send_returns_error_on_registration_timeout(self, monkeypatch):
+        from gateway.config import PlatformConfig
+
+        monkeypatch.setenv("IRC_SERVER", "irc.test.net")
+        monkeypatch.setenv("IRC_CHANNEL", "#cron")
+        monkeypatch.setenv("IRC_NICKNAME", "hermesbot")
+        monkeypatch.setenv("IRC_USE_TLS", "false")
+
+        # No 001 response: the readuntil call returns IncompleteReadError so
+        # the registration loop times out via the asyncio wait_for inside.
+        conn = _FakeIRCConnection([])
+
+        async def _fake_open(host, port, **kwargs):
+            return conn, conn
+
+        monkeypatch.setattr(_irc_mod.asyncio, "open_connection", _fake_open)
+
+        # Patch wait_for to raise TimeoutError immediately so the test is fast
+        async def _fast_timeout(coro, timeout):
+            try:
+                return await coro
+            except asyncio.IncompleteReadError:
+                raise asyncio.TimeoutError()
+
+        monkeypatch.setattr(_irc_mod.asyncio, "wait_for", _fast_timeout)
+
+        result = await _standalone_send(
+            PlatformConfig(enabled=True, extra={}),
+            "#cron",
+            "hi",
+        )
+
+        assert "error" in result
+        assert "registration" in result["error"].lower() or "timeout" in result["error"].lower()
+
+    @pytest.mark.asyncio
+    async def test_standalone_send_rejects_crlf_in_chat_id(self, monkeypatch):
+        from gateway.config import PlatformConfig
+
+        monkeypatch.setenv("IRC_SERVER", "irc.test.net")
+        monkeypatch.setenv("IRC_CHANNEL", "#cron")
+        monkeypatch.setenv("IRC_NICKNAME", "hermesbot")
+        monkeypatch.setenv("IRC_USE_TLS", "false")
+
+        # Attempt to inject a second IRC command via CRLF in chat_id
+        result = await _standalone_send(
+            PlatformConfig(enabled=True, extra={}),
+            "#cron\r\nKICK #cron hermesbot",
+            "hi",
+        )
+
+        assert "error" in result
+        assert "illegal IRC characters" in result["error"]
+
+    @pytest.mark.asyncio
+    async def test_standalone_send_strips_crlf_from_message_body(self, monkeypatch):
+        from gateway.config import PlatformConfig
+
+        monkeypatch.setenv("IRC_SERVER", "irc.test.net")
+        monkeypatch.setenv("IRC_CHANNEL", "#cron")
+        monkeypatch.setenv("IRC_NICKNAME", "hermesbot")
+        monkeypatch.setenv("IRC_USE_TLS", "false")
+
+        conn = _FakeIRCConnection([b":server 001 hermesbot-cron :Welcome"])
+
+        async def _fake_open(host, port, **kwargs):
+            return conn, conn
+
+        monkeypatch.setattr(_irc_mod.asyncio, "open_connection", _fake_open)
+
+        # A bare \r in message content tries to inject a NICK command.
+        # Our control-char stripper must blank \r so the line stays one PRIVMSG.
+        result = await _standalone_send(
+            PlatformConfig(enabled=True, extra={}),
+            "#cron",
+            "hello\rNICK eviltwin",
+        )
+
+        sent_lines = b"".join(conn.writes).decode("utf-8").splitlines()
+        # No injected NICK command after the legitimate registration NICK
+        nick_lines = [line for line in sent_lines if line.startswith("NICK ")]
+        # Only the original registration NICK should be present (no injected one)
+        assert all(line.startswith("NICK hermesbot-cron") for line in nick_lines)
+        # The PRIVMSG should contain "hello NICK eviltwin" as one line (with \r blanked)
+        assert any("PRIVMSG #cron :hello NICK eviltwin" in line for line in sent_lines)
+
+    @pytest.mark.asyncio
+    async def test_standalone_send_joins_channel_before_privmsg(self, monkeypatch):
+        from gateway.config import PlatformConfig
+
+        monkeypatch.setenv("IRC_SERVER", "irc.test.net")
+        monkeypatch.setenv("IRC_CHANNEL", "#cron")
+        monkeypatch.setenv("IRC_NICKNAME", "hermesbot")
+        monkeypatch.setenv("IRC_USE_TLS", "false")
+
+        # Register, then accept JOIN with 366 RPL_ENDOFNAMES, then PRIVMSG.
+        conn = _FakeIRCConnection([
+            b":server 001 hermesbot-cron :Welcome",
+            b":server 366 hermesbot-cron #cron :End of /NAMES list.",
+        ])
+
+        async def _fake_open(host, port, **kwargs):
+            return conn, conn
+
+        monkeypatch.setattr(_irc_mod.asyncio, "open_connection", _fake_open)
+
+        result = await _standalone_send(
+            PlatformConfig(enabled=True, extra={}),
+            "#cron",
+            "hello",
+        )
+
+        assert result["success"] is True
+        sent_lines = b"".join(conn.writes).decode("utf-8").splitlines()
+        join_idx = next((i for i, line in enumerate(sent_lines) if line.startswith("JOIN #cron")), None)
+        privmsg_idx = next((i for i, line in enumerate(sent_lines) if line.startswith("PRIVMSG #cron")), None)
+        assert join_idx is not None, "JOIN must be sent for channel targets"
+        assert privmsg_idx is not None
+        assert join_idx < privmsg_idx, "JOIN must precede PRIVMSG"

tests/gateway/test_teams.py

@@ -703,3 +703,177 @@ class TestTeamsMessageHandling:
         await adapter._on_message(ctx)
 
         assert adapter.handle_message.await_count == 1
+
+
+# ── _standalone_send (out-of-process cron delivery) ──────────────────────
+
+
+class _FakeAiohttpResponse:
+    def __init__(self, status: int, payload, text_body: str = ""):
+        self.status = status
+        self._payload = payload
+        self._text = text_body or (str(payload) if payload is not None else "")
+
+    async def json(self):
+        return self._payload
+
+    async def text(self):
+        return self._text
+
+    async def __aenter__(self):
+        return self
+
+    async def __aexit__(self, exc_type, exc, tb):
+        return None
+
+
+class _FakeAiohttpSession:
+    """Scripted aiohttp.ClientSession with a queue of responses so tests
+    can assert calls in order."""
+
+    def __init__(self, scripts):
+        self._scripts = list(scripts)
+        self.calls: list[tuple[str, dict]] = []
+
+    async def __aenter__(self):
+        return self
+
+    async def __aexit__(self, exc_type, exc, tb):
+        return None
+
+    def post(self, url, **kwargs):
+        self.calls.append((url, kwargs))
+        if not self._scripts:
+            raise AssertionError(f"No scripted response for POST {url}")
+        return self._scripts.pop(0)
+
+
+def _install_fake_aiohttp(monkeypatch, session):
+    """Replace ``aiohttp`` in ``sys.modules`` so ``import aiohttp as _aiohttp``
+    inside ``_standalone_send`` picks up our fake."""
+    fake_aiohttp = types.SimpleNamespace(
+        ClientSession=lambda timeout=None: session,
+        ClientTimeout=lambda total=None: None,
+    )
+    monkeypatch.setitem(sys.modules, "aiohttp", fake_aiohttp)
+
+
+class TestTeamsStandaloneSend:
+
+    @pytest.mark.asyncio
+    async def test_standalone_send_acquires_token_and_posts_activity(self, monkeypatch):
+        monkeypatch.setenv("TEAMS_CLIENT_ID", "client-id")
+        monkeypatch.setenv("TEAMS_CLIENT_SECRET", "secret")
+        monkeypatch.setenv("TEAMS_TENANT_ID", "tenant")
+        monkeypatch.delenv("TEAMS_SERVICE_URL", raising=False)
+
+        token_resp = _FakeAiohttpResponse(200, {"access_token": "the-token"})
+        activity_resp = _FakeAiohttpResponse(200, {"id": "msg-99"})
+        session = _FakeAiohttpSession([token_resp, activity_resp])
+        _install_fake_aiohttp(monkeypatch, session)
+
+        result = await _teams_mod._standalone_send(
+            PlatformConfig(enabled=True, extra={}),
+            "19:abc@thread.skype",
+            "hello cron",
+        )
+
+        assert result == {"success": True, "message_id": "msg-99"}
+        assert len(session.calls) == 2
+
+        token_url, token_kwargs = session.calls[0]
+        assert "login.microsoftonline.com/tenant/oauth2/v2.0/token" in token_url
+        assert token_kwargs["data"]["client_id"] == "client-id"
+        assert token_kwargs["data"]["client_secret"] == "secret"
+        assert token_kwargs["data"]["scope"] == "https://api.botframework.com/.default"
+
+        activity_url, activity_kwargs = session.calls[1]
+        # Default service URL when TEAMS_SERVICE_URL is unset
+        assert "smba.trafficmanager.net" in activity_url
+        assert "/v3/conversations/19:abc@thread.skype/activities" in activity_url
+        assert activity_kwargs["headers"]["Authorization"] == "Bearer the-token"
+        assert activity_kwargs["json"]["text"] == "hello cron"
+        assert activity_kwargs["json"]["type"] == "message"
+
+    @pytest.mark.asyncio
+    async def test_standalone_send_returns_error_when_unconfigured(self, monkeypatch):
+        for var in ("TEAMS_CLIENT_ID", "TEAMS_CLIENT_SECRET", "TEAMS_TENANT_ID"):
+            monkeypatch.delenv(var, raising=False)
+
+        result = await _teams_mod._standalone_send(
+            PlatformConfig(enabled=True, extra={}),
+            "19:abc@thread.skype",
+            "hi",
+        )
+
+        assert "error" in result
+        assert "TEAMS_CLIENT_ID" in result["error"]
+
+    @pytest.mark.asyncio
+    async def test_standalone_send_propagates_token_failure(self, monkeypatch):
+        monkeypatch.setenv("TEAMS_CLIENT_ID", "client-id")
+        monkeypatch.setenv("TEAMS_CLIENT_SECRET", "secret")
+        monkeypatch.setenv("TEAMS_TENANT_ID", "tenant")
+
+        token_resp = _FakeAiohttpResponse(
+            401,
+            {"error": "unauthorized_client"},
+            text_body='{"error":"unauthorized_client"}',
+        )
+        session = _FakeAiohttpSession([token_resp])
+        _install_fake_aiohttp(monkeypatch, session)
+
+        result = await _teams_mod._standalone_send(
+            PlatformConfig(enabled=True, extra={}),
+            "19:abc@thread.skype",
+            "hi",
+        )
+
+        assert "error" in result
+        assert "401" in result["error"]
+        assert "token" in result["error"].lower()
+
+    @pytest.mark.asyncio
+    async def test_standalone_send_rejects_off_allowlist_service_url(self, monkeypatch):
+        monkeypatch.setenv("TEAMS_CLIENT_ID", "client-id")
+        monkeypatch.setenv("TEAMS_CLIENT_SECRET", "secret")
+        monkeypatch.setenv("TEAMS_TENANT_ID", "tenant")
+        # SSRF attempt: point us at an attacker-controlled host
+        monkeypatch.setenv("TEAMS_SERVICE_URL", "https://attacker.example.com/teams/")
+
+        # If the allowlist check fails to fire, the fake session will assert
+        # because no scripts are queued; a passing test means we returned
+        # before any HTTP call.
+        session = _FakeAiohttpSession([])
+        _install_fake_aiohttp(monkeypatch, session)
+
+        result = await _teams_mod._standalone_send(
+            PlatformConfig(enabled=True, extra={}),
+            "19:abc@thread.skype",
+            "hi",
+        )
+
+        assert "error" in result
+        assert "allowlist" in result["error"].lower()
+        assert len(session.calls) == 0, "must not call any HTTP endpoint with a tampered service URL"
+
+    @pytest.mark.asyncio
+    async def test_standalone_send_rejects_chat_id_with_path_traversal(self, monkeypatch):
+        monkeypatch.setenv("TEAMS_CLIENT_ID", "client-id")
+        monkeypatch.setenv("TEAMS_CLIENT_SECRET", "secret")
+        monkeypatch.setenv("TEAMS_TENANT_ID", "tenant")
+        monkeypatch.delenv("TEAMS_SERVICE_URL", raising=False)
+
+        session = _FakeAiohttpSession([])
+        _install_fake_aiohttp(monkeypatch, session)
+
+        # Attempt to break out of /v3/conversations/<id>/activities via a `/`
+        result = await _teams_mod._standalone_send(
+            PlatformConfig(enabled=True, extra={}),
+            "19:abc/activities/19:other@thread.skype",
+            "hi",
+        )
+
+        assert "error" in result
+        assert "Bot Framework conversation ID" in result["error"]
+        assert len(session.calls) == 0

tests/tools/test_send_message_tool.py

@@ -2052,3 +2052,180 @@ class TestSendSignalChunking:
         # Only the existing file made it into the RPC
         params = fake.calls[0]["payload"]["params"]
         assert len(params["attachments"]) == 1
+
+
+# ── _send_via_adapter standalone fallback ────────────────────────────────
+
+
+class _FakePlatform:
+    """Stand-in for the gateway.config.Platform enum.  Holds the .value
+    attribute consulted by ``_send_via_adapter`` for registry lookups."""
+
+    def __init__(self, value):
+        self.value = value
+
+
+class TestSendViaAdapterStandaloneFallback:
+    """Coverage for the out-of-process plugin-platform send path.
+
+    When the gateway runner is not in this process (e.g. ``hermes cron``
+    runs separately from ``hermes gateway``), ``_send_via_adapter`` should
+    fall through to the plugin's ``standalone_sender_fn`` registered on
+    its ``PlatformEntry``.  Without the hook, the existing error string
+    is returned (with a more helpful tail).
+    """
+
+    @staticmethod
+    def _make_entry(send_fn):
+        from gateway.platform_registry import PlatformEntry
+
+        return PlatformEntry(
+            name="fakeplatform",
+            label="Fake",
+            adapter_factory=lambda cfg: None,
+            check_fn=lambda: True,
+            standalone_sender_fn=send_fn,
+        )
+
+    @pytest.mark.asyncio
+    async def test_standalone_sender_fn_called_when_no_adapter(self, monkeypatch):
+        """Registry has hook, runner ref returns None: the hook is awaited."""
+        from tools.send_message_tool import _send_via_adapter
+        from gateway.platform_registry import platform_registry
+
+        recorded = {}
+
+        async def fake_send(pconfig, chat_id, message, **kwargs):
+            recorded["pconfig"] = pconfig
+            recorded["chat_id"] = chat_id
+            recorded["message"] = message
+            recorded["kwargs"] = kwargs
+            return {"success": True, "message_id": "msg-42"}
+
+        platform_registry.register(self._make_entry(fake_send))
+        try:
+            monkeypatch.setattr("gateway.run._gateway_runner_ref", lambda: None)
+
+            pconfig = SimpleNamespace(extra={})
+            result = await _send_via_adapter(
+                _FakePlatform("fakeplatform"),
+                pconfig,
+                "room/123",
+                "hello cron",
+            )
+        finally:
+            platform_registry.unregister("fakeplatform")
+
+        assert result == {"success": True, "message_id": "msg-42"}
+        assert recorded["chat_id"] == "room/123"
+        assert recorded["message"] == "hello cron"
+        assert recorded["pconfig"] is pconfig
+
+    @pytest.mark.asyncio
+    async def test_standalone_sender_fn_kwargs_forwarded(self, monkeypatch):
+        """thread_id, media_files, and force_document all reach the hook."""
+        from tools.send_message_tool import _send_via_adapter
+        from gateway.platform_registry import platform_registry
+
+        recorded = {}
+
+        async def fake_send(pconfig, chat_id, message, *, thread_id=None,
+                            media_files=None, force_document=False):
+            recorded["thread_id"] = thread_id
+            recorded["media_files"] = media_files
+            recorded["force_document"] = force_document
+            return {"success": True, "message_id": "x"}
+
+        platform_registry.register(self._make_entry(fake_send))
+        try:
+            monkeypatch.setattr("gateway.run._gateway_runner_ref", lambda: None)
+
+            await _send_via_adapter(
+                _FakePlatform("fakeplatform"),
+                SimpleNamespace(extra={}),
+                "chat-1",
+                "hi",
+                thread_id="thread-7",
+                media_files=["/tmp/a.png"],
+                force_document=True,
+            )
+        finally:
+            platform_registry.unregister("fakeplatform")
+
+        assert recorded["thread_id"] == "thread-7"
+        assert recorded["media_files"] == ["/tmp/a.png"]
+        assert recorded["force_document"] is True
+
+    @pytest.mark.asyncio
+    async def test_standalone_sender_fn_absent_returns_helpful_error(self, monkeypatch):
+        """Registry entry has no hook: the fall-through error explains both
+        options (gateway-running and standalone hook)."""
+        from tools.send_message_tool import _send_via_adapter
+        from gateway.platform_registry import platform_registry
+
+        platform_registry.register(self._make_entry(None))
+        try:
+            monkeypatch.setattr("gateway.run._gateway_runner_ref", lambda: None)
+
+            result = await _send_via_adapter(
+                _FakePlatform("fakeplatform"),
+                SimpleNamespace(extra={}),
+                "chat-1",
+                "hi",
+            )
+        finally:
+            platform_registry.unregister("fakeplatform")
+
+        assert "error" in result
+        assert "fakeplatform" in result["error"]
+        assert "standalone_sender_fn" in result["error"]
+
+    @pytest.mark.asyncio
+    async def test_standalone_sender_fn_raises_is_caught_and_formatted(self, monkeypatch):
+        """Hook raises: error dict has 'Plugin standalone send failed: ...'"""
+        from tools.send_message_tool import _send_via_adapter
+        from gateway.platform_registry import platform_registry
+
+        async def boom(pconfig, chat_id, message, **kwargs):
+            raise ValueError("boom!")
+
+        platform_registry.register(self._make_entry(boom))
+        try:
+            monkeypatch.setattr("gateway.run._gateway_runner_ref", lambda: None)
+
+            result = await _send_via_adapter(
+                _FakePlatform("fakeplatform"),
+                SimpleNamespace(extra={}),
+                "chat-1",
+                "hi",
+            )
+        finally:
+            platform_registry.unregister("fakeplatform")
+
+        assert result == {"error": "Plugin standalone send failed: boom!"}
+
+    @pytest.mark.asyncio
+    async def test_standalone_sender_fn_return_shape_passed_through(self, monkeypatch):
+        """Hook returns success dict: passed through unchanged."""
+        from tools.send_message_tool import _send_via_adapter
+        from gateway.platform_registry import platform_registry
+
+        async def fake_send(pconfig, chat_id, message, **kwargs):
+            return {"success": True, "message_id": "abc-123", "extra_field": "preserved"}
+
+        platform_registry.register(self._make_entry(fake_send))
+        try:
+            monkeypatch.setattr("gateway.run._gateway_runner_ref", lambda: None)
+
+            result = await _send_via_adapter(
+                _FakePlatform("fakeplatform"),
+                SimpleNamespace(extra={}),
+                "chat-1",
+                "hi",
+            )
+        finally:
+            platform_registry.unregister("fakeplatform")
+
+        assert result["success"] is True
+        assert result["message_id"] == "abc-123"
+        assert result["extra_field"] == "preserved"