diff --git a/gateway/delivery.py b/gateway/delivery.py
index 6585cfbe1e87e..d35b60c10c4c2 100644
--- a/gateway/delivery.py
+++ b/gateway/delivery.py
@@ -251,6 +251,22 @@ def _cap_oversized_output(self, adapter: Any, content: str, job_id: str) -> str:
         logger.info("Cron output truncated (%d chars) — full output: %s", len(content), saved_path)
         return content[:max(0, MAX_PLATFORM_OUTPUT - len(footer))] + footer
 
+    async def _apply_delivery_guards(self, target: "DeliveryTarget", content: str) -> Optional[str]:
+        """Run outbound cron content through the shared output-guard pipeline.
+
+        Returns the (possibly rewritten) content, or ``None`` if a guard dropped the message.
+        Falls back to the legacy silence-only check if the pipeline can't be imported, so
+        delivery never breaks on a bad import."""
+        try:
+            from gateway.output_guards import apply_output_guards, GuardContext
+            ctx = GuardContext(platform=target.platform.value, chat_id=target.chat_id, is_final_response=False)
+            return await apply_output_guards(content, ctx)
+        except Exception:
+            logger.debug("output-guard pipeline failed in delivery; using legacy check", exc_info=True)
+            if _is_silence_narration(content):
+                return None
+            return content
+
     async def _deliver_to_platform(self, target: DeliveryTarget, content: str,
                                    metadata: Optional[Dict[str, Any]]) -> Dict[str, Any]:
         """Deliver content to a messaging platform."""
@@ -262,19 +278,28 @@ async def _deliver_to_platform(self, target: DeliveryTarget, content: str,
         adapter = transport.adapter
         content = self._cap_oversized_output(adapter, content, (metadata or {}).get("job_id", "unknown"))
 
-        # Substrate-level anti-loop guard: drop hallucinated "silence narration" (*(silent)*, 🔇, a bare ".")
-        # before it reaches any adapter — in bot-to-bot channels these mirror back and forth until a model
-        # crashes with "no content after all retries"; prompt rules drift across providers, so this single
-        # chokepoint covers every platform. Local/file delivery is never filtered (saved silence has no loop
-        # risk). Cron output is an ARTIFACT, not model chatter: a legitimately terse job ("...", a single 🔇)
-        # has no mirror loop, and dropping it while returning success is how a cron gets logged as delivered
-        # with nothing on the wire. Cron sends carry job_id in metadata; everything else is filtered.
+        # Substrate-level outbound guard pipeline. Historically this chokepoint only dropped
+        # hallucinated "silence narration" (*(silent)*, 🔇, a bare "."). It now threads content
+        # through the composable guard chain (gateway.output_guards) so secret redaction,
+        # provider-error rewriting, and opt-in guards (em-dash stripping, link verification) all
+        # apply to scheduled deliveries too — the same rules the agent's live replies get. The
+        # silence drop is one guard in that chain; the legacy single-check path (in
+        # _apply_delivery_guards' except branch) is kept as a fallback so a pipeline
+        # import/runtime error can never block a delivery. Local/file delivery is a separate path
+        # and is intentionally never filtered (saved silence has no loop risk). Cron output is an
+        # ARTIFACT, not model chatter: a legitimately terse job ("...", a single 🔇) has no mirror
+        # loop, and dropping it while returning success is how a cron gets logged as delivered
+        # with nothing on the wire — so cron sends (job_id in metadata) skip the silence guard.
         # See #77763.
         is_cron_artifact = "job_id" in (metadata or {})
-        if self._filter_silence_narration_enabled() and not is_cron_artifact and _is_silence_narration(content):
-            logger.warning("Dropped silence-narration outbound to %s (chat=%s): %r",
-                           target.platform.value, target.chat_id, content[:40])
-            return {"success": True, "filtered": "silence_narration", "delivered": False}
+        if self._filter_silence_narration_enabled() and not is_cron_artifact:
+            guarded = await self._apply_delivery_guards(target, content)
+            if guarded is None:
+                filtered_reason = "silence_narration" if _is_silence_narration(content) else "output_guard"
+                logger.warning("Dropped outbound to %s (chat=%s) by output guard [%s]: %r",
+                               target.platform.value, target.chat_id, filtered_reason, content[:40])
+                return {"success": True, "filtered": filtered_reason, "delivered": False}
+            content = guarded
 
         send_metadata = dict(metadata or {})
         home = self.config.get_home_channel(target.platform) if transport.is_relay else None
diff --git a/gateway/output_guards.py b/gateway/output_guards.py
new file mode 100644
index 0000000000000..52fc1ae5776ca
--- /dev/null
+++ b/gateway/output_guards.py
@@ -0,0 +1,360 @@
+"""
+Composable output-guard pipeline for outbound gateway messages.
+
+Every message the gateway sends to a user — an agent's final reply, a cron
+delivery, a status line — can be threaded through an ordered chain of small,
+independent validators before it leaves the process. Each validator inspects
+(and optionally rewrites or drops) the text. The pipeline is the reusable
+part; adding a new rule is one function plus one registry entry.
+
+This generalizes three checks that previously lived as one-off code paths:
+
+  * secret redaction         (``gateway/run.py:_redact_gateway_user_facing_secrets``)
+  * provider-error rewriting (``gateway/run.py:_looks_like_gateway_provider_error``)
+  * silence-narration drop   (``gateway/delivery.py:_is_silence_narration``)
+
+New guards drop in beside them:
+
+  * em-dash stripping        (opt-in; ``gateway.guards.strip_em_dashes``)
+  * link verification        (opt-in, async; ``gateway.guards.verify_links``)
+
+Design
+------
+A guard is a callable ``(text, ctx) -> GuardOutcome | None``. Returning
+``None`` means "no change". Returning a :class:`GuardOutcome` can rewrite the
+text (``text=...``) or suppress the message entirely (``drop=True``). Guards
+run in registration order and the (possibly rewritten) text threads from one
+to the next; a drop short-circuits the rest of the chain.
+
+Guards may be sync or async. :func:`apply_output_guards` awaits async guards;
+:func:`apply_output_guards_sync` runs only the sync ones (used on code paths
+that are not inside an event loop). Async-only guards (network I/O such as
+link checks) are simply skipped by the sync entry point.
+
+Guards are cheap to reason about because each one owns a single concern and
+sees the same :class:`GuardContext`. The pipeline never raises: a guard that
+throws is logged and skipped so a buggy rule can never block delivery.
+"""
+
+from __future__ import annotations
+
+import asyncio
+import inspect
+import logging
+import os
+import re
+from dataclasses import dataclass, field
+from typing import Any, Awaitable, Callable, Dict, List, Optional, Union
+
+logger = logging.getLogger(__name__)
+
+
+# =========================================================================
+# Data types
+# =========================================================================
+
+@dataclass
+class GuardContext:
+    """Everything a guard needs to decide what to do with a message.
+
+    Attributes:
+        platform:          Platform value string ("telegram", "discord", …).
+        chat_id:           Destination chat id, when known.
+        is_final_response: True for an agent's final reply (vs. a status line
+                           or cron delivery). Some guards only apply here.
+        metadata:          Free-form send metadata (thread id, job id, …).
+    """
+    platform: str = ""
+    chat_id: Optional[str] = None
+    is_final_response: bool = False
+    metadata: Dict[str, Any] = field(default_factory=dict)
+
+
+@dataclass
+class GuardOutcome:
+    """Result of a single guard.
+
+    ``text`` carries a rewrite (``None`` leaves the running text unchanged).
+    ``drop`` suppresses the whole message and short-circuits the chain.
+    ``reason`` is a short tag for logging/telemetry.
+    """
+    text: Optional[str] = None
+    drop: bool = False
+    reason: Optional[str] = None
+
+
+GuardResult = Optional[GuardOutcome]
+GuardFn = Callable[[str, GuardContext], Union[GuardResult, Awaitable[GuardResult]]]
+
+
+@dataclass
+class _Guard:
+    name: str
+    fn: GuardFn
+    is_async: bool
+
+
+# =========================================================================
+# Pipeline
+# =========================================================================
+
+class OutputGuardPipeline:
+    """An ordered, fault-isolated chain of output guards."""
+
+    def __init__(self) -> None:
+        self._guards: List[_Guard] = []
+
+    def register(self, name: str, fn: GuardFn) -> None:
+        """Append a guard. Order of registration is execution order."""
+        self._guards.append(
+            _Guard(name=name, fn=fn, is_async=inspect.iscoroutinefunction(fn))
+        )
+
+    def names(self) -> List[str]:
+        return [g.name for g in self._guards]
+
+    def _step(self, outcome: GuardResult, text: str, name: str) -> tuple[str, bool]:
+        """Fold one guard's outcome into the running text.
+
+        Returns ``(new_text, dropped)``.
+        """
+        if outcome is None:
+            return text, False
+        if outcome.drop:
+            logger.info("[output-guard] %s dropped message (%s)", name, outcome.reason or "")
+            return text, True
+        if outcome.text is not None and outcome.text != text:
+            logger.debug("[output-guard] %s rewrote message (%s)", name, outcome.reason or "")
+            return outcome.text, False
+        return text, False
+
+    def apply_sync(self, text: str, ctx: GuardContext) -> Optional[str]:
+        """Run the sync guards only. Returns the final text, or ``None`` to drop.
+
+        Async guards are skipped — use :meth:`apply` on event-loop paths that
+        need them (e.g. link verification).
+        """
+        current = str(text or "")
+        for g in self._guards:
+            if g.is_async:
+                continue
+            try:
+                outcome = g.fn(current, ctx)  # type: ignore[assignment]
+            except Exception as exc:  # never let a guard block delivery
+                logger.warning("[output-guard] %s raised (skipped): %s", g.name, exc)
+                continue
+            current, dropped = self._step(outcome, current, g.name)  # type: ignore[arg-type]
+            if dropped:
+                return None
+        return current
+
+    async def apply(self, text: str, ctx: GuardContext) -> Optional[str]:
+        """Run every guard (sync + async). Returns final text, or ``None`` to drop."""
+        current = str(text or "")
+        for g in self._guards:
+            try:
+                outcome = g.fn(current, ctx)
+                if g.is_async or inspect.isawaitable(outcome):
+                    outcome = await outcome  # type: ignore[assignment]
+            except Exception as exc:
+                logger.warning("[output-guard] %s raised (skipped): %s", g.name, exc)
+                continue
+            current, dropped = self._step(outcome, current, g.name)  # type: ignore[arg-type]
+            if dropped:
+                return None
+        return current
+
+
+# =========================================================================
+# Config helpers
+# =========================================================================
+
+def _guard_flag(key: str, default: bool) -> bool:
+    """Read a per-guard on/off flag.
+
+    Env ``HERMES_GUARD_<KEY>`` overrides config; otherwise
+    ``gateway.guards.<key>`` in config.yaml wins, falling back to ``default``.
+    """
+    env = os.getenv(f"HERMES_GUARD_{key.upper()}")
+    if env is not None:
+        return env.strip().lower() in ("1", "true", "yes", "on")
+    try:
+        from hermes_cli.config import load_config
+        cfg = load_config() or {}
+        guards = ((cfg.get("gateway", {}) or {}).get("guards", {}) or {})
+        val = guards.get(key)
+        if val is not None:
+            return bool(val)
+    except Exception:
+        pass
+    return default
+
+
+# =========================================================================
+# Built-in guards
+# =========================================================================
+# Each guard is a thin adapter around logic that already existed elsewhere in
+# the gateway, re-expressed as a pipeline stage. Keeping the heavy lifting in
+# the original modules (run.py, delivery.py) avoids duplicating regexes; the
+# guards here delegate to those helpers and decide drop-vs-rewrite.
+
+
+def _secret_redaction_guard(text: str, ctx: GuardContext) -> GuardResult:
+    """Redact anything that looks like a credential before it leaves."""
+    try:
+        from gateway.run import _redact_gateway_user_facing_secrets
+    except Exception:
+        return None
+    redacted = _redact_gateway_user_facing_secrets(text)
+    if redacted != text:
+        return GuardOutcome(text=redacted, reason="secret")
+    return None
+
+
+def _provider_error_guard(text: str, ctx: GuardContext) -> GuardResult:
+    """Rewrite raw provider/API error envelopes into a short safe reply.
+
+    Applies to every chat surface, not just Telegram. Upstream widened this
+    invariant from Telegram (#28533) to all chat platforms (#39293): a provider
+    error body can carry a leaked bearer token, request IDs, or policy text,
+    and none of it should reach a chat user on any platform. The caller
+    (``_sanitize_gateway_final_response``) has already excluded programmatic
+    surfaces via ``_GATEWAY_RAW_TEXT_PLATFORMS``, so anything reaching this
+    guard is a human-facing surface that should get the safe category instead
+    of the raw envelope.
+    """
+    try:
+        from gateway.run import (
+            _looks_like_gateway_provider_error,
+            _gateway_provider_error_reply,
+        )
+    except Exception:
+        return None
+    if _looks_like_gateway_provider_error(text):
+        return GuardOutcome(text=_gateway_provider_error_reply(text), reason="provider-error")
+    return None
+
+
+def _silence_narration_guard(text: str, ctx: GuardContext) -> GuardResult:
+    """Drop hallucinated silence tokens (*(silent)*, 🔇, a bare '.') entirely."""
+    try:
+        from gateway.delivery import _is_silence_narration
+    except Exception:
+        return None
+    if _is_silence_narration(text):
+        return GuardOutcome(drop=True, reason="silence-narration")
+    return None
+
+
+# --- em-dash stripping (opt-in) ------------------------------------------
+# A single user-facing style rule expressed as a guard: replace em/en dashes
+# with typographically safe substitutes. Off by default so it never surprises
+# other deployments; enable with gateway.guards.strip_em_dashes: true.
+
+_EM_DASH_RE = re.compile(r"\s*[\u2014\u2013]\s*")
+
+
+def _em_dash_guard(text: str, ctx: GuardContext) -> GuardResult:
+    if not _guard_flag("strip_em_dashes", default=False):
+        return None
+    if "\u2014" not in text and "\u2013" not in text:
+        return None
+    # Replace a dash flanked by spaces with ", " (clause break); a bare dash
+    # between words becomes a plain hyphen-free comma too. Collapse doubles.
+    rewritten = _EM_DASH_RE.sub(", ", text)
+    rewritten = re.sub(r",\s*,", ", ", rewritten)
+    if rewritten != text:
+        return GuardOutcome(text=rewritten, reason="em-dash")
+    return None
+
+
+# --- link verification (opt-in, async) -----------------------------------
+# Never emit a URL the gateway hasn't confirmed resolves. Off by default
+# (adds latency + network I/O); enable with gateway.guards.verify_links: true.
+
+_URL_RE = re.compile(r"https?://[^\s<>()\[\]\"']+")
+_LINK_TIMEOUT = 4.0
+
+
+async def _verify_links_guard(text: str, ctx: GuardContext) -> GuardResult:
+    if not _guard_flag("verify_links", default=False):
+        return None
+    urls = list(dict.fromkeys(_URL_RE.findall(text)))
+    if not urls:
+        return None
+    try:
+        import aiohttp
+    except Exception:
+        logger.debug("[output-guard] verify_links needs aiohttp; skipping")
+        return None
+
+    async def _resolves(session, url: str) -> bool:
+        for method in (session.head, session.get):
+            try:
+                async with method(url, allow_redirects=True,
+                                  timeout=aiohttp.ClientTimeout(total=_LINK_TIMEOUT)) as resp:
+                    if resp.status < 400:
+                        return True
+                    if resp.status in (403, 405) and method is session.head:
+                        continue  # some hosts reject HEAD; try GET
+                    return False
+            except Exception:
+                continue
+        return False
+
+    dead: List[str] = []
+    async with aiohttp.ClientSession() as session:
+        results = await asyncio.gather(*[_resolves(session, u) for u in urls])
+    dead = [u for u, ok in zip(urls, results) if not ok]
+    if not dead:
+        return None
+    # Strip the dead links inline and flag them, rather than dropping the
+    # whole message (the surrounding prose is usually still useful).
+    rewritten = text
+    for u in dead:
+        rewritten = rewritten.replace(u, "[link removed: did not resolve]")
+    logger.info("[output-guard] verify_links removed %d dead link(s)", len(dead))
+    return GuardOutcome(text=rewritten, reason="dead-links")
+
+
+# =========================================================================
+# Default pipeline (singleton)
+# =========================================================================
+
+_default_pipeline: Optional[OutputGuardPipeline] = None
+
+
+def get_default_pipeline() -> OutputGuardPipeline:
+    """Return the process-wide default pipeline, building it on first use.
+
+    Order matters: redact secrets first (so nothing downstream can leak a
+    key), then rewrite provider errors, then style rules (em-dash), then
+    network checks (links), and finally the silence drop last so a message
+    that got rewritten to empty-ish still gets the drop check.
+    """
+    global _default_pipeline
+    if _default_pipeline is None:
+        p = OutputGuardPipeline()
+        p.register("secret", _secret_redaction_guard)
+        p.register("provider-error", _provider_error_guard)
+        p.register("em-dash", _em_dash_guard)
+        p.register("verify-links", _verify_links_guard)
+        p.register("silence-narration", _silence_narration_guard)
+        _default_pipeline = p
+    return _default_pipeline
+
+
+def reset_default_pipeline() -> None:
+    """Drop the cached pipeline (tests toggle config between builds)."""
+    global _default_pipeline
+    _default_pipeline = None
+
+
+def apply_output_guards_sync(text: str, ctx: GuardContext) -> Optional[str]:
+    """Convenience wrapper: run sync guards of the default pipeline."""
+    return get_default_pipeline().apply_sync(text, ctx)
+
+
+async def apply_output_guards(text: str, ctx: GuardContext) -> Optional[str]:
+    """Convenience wrapper: run all guards of the default pipeline."""
+    return await get_default_pipeline().apply(text, ctx)
diff --git a/gateway/run.py b/gateway/run.py
index 8a1fecf6f259b..a112b71199652 100644
--- a/gateway/run.py
+++ b/gateway/run.py
@@ -624,7 +624,17 @@ def _looks_like_gateway_provider_error(text: str) -> bool:
 
 def _sanitize_gateway_final_response(platform: Any, text: str) -> str:
     """Sanitize final gateway replies for chat surfaces: concise, secret-redacted provider failure
-    categories instead of raw HTTP bodies, request IDs, leaked credentials, or policy text."""
+    categories instead of raw HTTP bodies, request IDs, leaked credentials, or policy text.
+
+    The sanitizing runs through the composable output-guard pipeline (``gateway.output_guards``),
+    which reproduces secret redaction and provider-error rewriting as ordered guards and lets
+    opt-in guards (em-dash stripping, link verification) drop in via ``gateway.guards.*`` config
+    without touching this call site. Only the synchronous guards run here (this helper is called
+    from both sync and async contexts); async guards such as link verification apply on the
+    delivery path instead. A pipeline "drop" is rare for a final response, so the original text
+    is kept rather than sending an empty reply; any pipeline failure falls back to the inline
+    legacy path so a bug in the guard layer can never mute the agent.
+    """
     if not text or _gateway_surface_passes_raw_text(platform):
         return text
 
@@ -644,10 +654,20 @@ def _sanitize_gateway_final_response(platform: Any, text: str) -> str:
     if str(text).strip().startswith(INTERRUPT_WAITING_FOR_MODEL_PREFIX):
         return ""
 
-    redacted = _redact_gateway_user_facing_secrets(str(text))
-    if _looks_like_gateway_provider_error(redacted):
-        return _gateway_provider_error_reply(redacted)
-    return redacted
+    try:
+        from gateway.output_guards import apply_output_guards_sync, GuardContext
+        ctx = GuardContext(
+            platform=_gateway_platform_value(platform),
+            is_final_response=True,
+        )
+        result = apply_output_guards_sync(str(text), ctx)
+        return result if result is not None else text
+    except Exception:
+        logger.debug("output-guard pipeline failed; using legacy sanitize", exc_info=True)
+        redacted = _redact_gateway_user_facing_secrets(str(text))
+        if _looks_like_gateway_provider_error(redacted):
+            return _gateway_provider_error_reply(redacted)
+        return redacted
 
 
 def _prepare_gateway_status_message(platform: Any, event_type: str, message: str) -> Optional[str]:
diff --git a/tests/gateway/test_output_guards.py b/tests/gateway/test_output_guards.py
new file mode 100644
index 0000000000000..ae5d150e484ec
--- /dev/null
+++ b/tests/gateway/test_output_guards.py
@@ -0,0 +1,171 @@
+"""Tests for the composable output-guard pipeline (gateway/output_guards.py)."""
+
+import asyncio
+import os
+
+import pytest
+
+from gateway.output_guards import (
+    GuardContext,
+    GuardOutcome,
+    OutputGuardPipeline,
+    apply_output_guards,
+    apply_output_guards_sync,
+    get_default_pipeline,
+    reset_default_pipeline,
+)
+
+
+@pytest.fixture(autouse=True)
+def _reset_pipeline():
+    """Isolate the process-wide guard pipeline + env flags for every test.
+
+    The pipeline is a module-level singleton and the guard flags are read from
+    env/config, so a test that enables an opt-in guard (em-dash stripping) can
+    otherwise leak into unrelated tests that later import the same singleton.
+    Clear env FIRST, then reset, on both setup and teardown so the ordering
+    can't leave a rebuilt-but-dirty pipeline behind.
+    """
+    def _clean():
+        for k in [k for k in os.environ if k.startswith("HERMES_GUARD_")]:
+            del os.environ[k]
+        reset_default_pipeline()
+
+    _clean()
+    yield
+    _clean()
+
+
+def _tg(**kw):
+    kw.setdefault("platform", "telegram")
+    kw.setdefault("is_final_response", True)
+    return GuardContext(**kw)
+
+
+# --- pipeline mechanics --------------------------------------------------
+
+def test_empty_pipeline_returns_text_unchanged():
+    p = OutputGuardPipeline()
+    ctx = _tg()
+    assert p.apply_sync("hello", ctx) == "hello"
+
+
+def test_rewrite_threads_through_chain():
+    p = OutputGuardPipeline()
+    p.register("upper", lambda t, c: GuardOutcome(text=t.upper()))
+    p.register("bang", lambda t, c: GuardOutcome(text=t + "!"))
+    assert p.apply_sync("hi", _tg()) == "HI!"
+
+
+def test_drop_short_circuits():
+    seen = []
+    p = OutputGuardPipeline()
+    p.register("drop", lambda t, c: GuardOutcome(drop=True, reason="x"))
+    p.register("after", lambda t, c: seen.append(t))
+    assert p.apply_sync("hi", _tg()) is None
+    assert seen == []  # second guard never ran
+
+
+def test_none_outcome_is_noop():
+    p = OutputGuardPipeline()
+    p.register("noop", lambda t, c: None)
+    assert p.apply_sync("hi", _tg()) == "hi"
+
+
+def test_guard_exception_is_isolated():
+    def boom(t, c):
+        raise RuntimeError("kaboom")
+
+    p = OutputGuardPipeline()
+    p.register("boom", boom)
+    p.register("ok", lambda t, c: GuardOutcome(text=t + "-ok"))
+    # The raising guard is skipped; the chain continues.
+    assert p.apply_sync("hi", _tg()) == "hi-ok"
+
+
+def test_sync_skips_async_guards():
+    async def aguard(t, c):
+        return GuardOutcome(text="async-ran")
+
+    p = OutputGuardPipeline()
+    p.register("async", aguard)
+    # apply_sync must not run async guards.
+    assert p.apply_sync("orig", _tg()) == "orig"
+
+
+def test_async_runs_all_guards():
+    async def aguard(t, c):
+        return GuardOutcome(text=t + "-async")
+
+    p = OutputGuardPipeline()
+    p.register("sync", lambda t, c: GuardOutcome(text=t + "-sync"))
+    p.register("async", aguard)
+    out = asyncio.run(p.apply("x", _tg()))
+    assert out == "x-sync-async"
+
+
+# --- built-in guards -----------------------------------------------------
+
+def test_default_pipeline_order():
+    names = get_default_pipeline().names()
+    assert names == [
+        "secret",
+        "provider-error",
+        "em-dash",
+        "verify-links",
+        "silence-narration",
+    ]
+
+
+def test_secret_redaction_all_platforms():
+    # Secret redaction is not telegram-gated. Assert the behaviour contract
+    # (the raw credential does not survive) rather than a specific mask
+    # marker: the authoritative redactor in agent.redact uses "***" while the
+    # gateway's belt-and-suspenders pass uses "[REDACTED]", and which one wins
+    # is an upstream implementation detail.
+    secret = "sk-" + "a" * 40
+    ctx = GuardContext(platform="discord", is_final_response=True)
+    out = apply_output_guards_sync(f"key {secret}", ctx)
+    assert out is not None
+    assert secret not in out
+
+
+def test_provider_error_rewrite_all_chat_surfaces():
+    # Upstream widened this from Telegram (#28533) to every chat surface
+    # (#39293): a raw provider envelope can carry a leaked token, so no chat
+    # platform should ever see it. Programmatic surfaces are excluded upstream
+    # of the pipeline by _GATEWAY_RAW_TEXT_PLATFORMS, not by this guard.
+    raw = "HTTP 401 incorrect api key"
+    for platform in ("telegram", "discord", "whatsapp", "slack", "signal", "matrix"):
+        out = apply_output_guards_sync(raw, GuardContext(platform=platform))
+        assert out is not None
+        assert "HTTP 401" not in out, f"{platform} leaked the raw provider envelope"
+        assert "authentication failed" in out.lower()
+
+
+def test_normal_text_untouched():
+    assert apply_output_guards_sync("Here is your answer.", _tg()) == "Here is your answer."
+
+
+def test_silence_narration_dropped():
+    out = asyncio.run(apply_output_guards("*(silent)*", _tg()))
+    assert out is None
+
+
+def test_em_dash_guard_opt_in():
+    text = "This is a test — with a dash."
+    # Off by default: unchanged.
+    assert apply_output_guards_sync(text, _tg()) == text
+    # On via env.
+    os.environ["HERMES_GUARD_STRIP_EM_DASHES"] = "1"
+    reset_default_pipeline()
+    out = apply_output_guards_sync(text, _tg())
+    assert "\u2014" not in out
+    assert "This is a test, with a dash." == out
+
+
+def test_em_dash_guard_handles_en_dash():
+    os.environ["HERMES_GUARD_STRIP_EM_DASHES"] = "1"
+    reset_default_pipeline()
+    out = apply_output_guards_sync("range 1\u20135 items", _tg())
+    assert "\u2013" not in out