fix(gateway): judge delivery success against final content, not flag trust (#95382, #98552)

A record-less delivery flag (final_response_sent /
final_content_delivered set with no recorded turn-final payload) was
trusted blindly by delivered_final_matches (None -> legacy trust), so a
first-edit prefix or a truncated finalize suppressed the gateway's
corrective send — silent partial delivery.

- delivered_final_matches: record-less flags are now reconciled against
  the FINAL content via has_delivered_text; only the explicitly-marked
  ambiguous-timeout path (_delivery_ambiguous) keeps legacy trust.
- _try_fresh_final and the native-streaming optimistic finalize now
  record their delivered payload (the last record-less flag setters);
  the optimistic record rolls back on definitive dispatch failure.
- Discord adapter: dead-transport send failures (client gone, WS
  closed/reset) are classified as send_path_degraded (retryable) so the
  delivery-obligation ledger's reconnect sweep replays the stranded
  final response instead of losing it until a process restart.

Fixes #95382; closes the #98552 false-positive class.

gateway/stream_consumer.py

@@ -332,6 +332,12 @@ class GatewayStreamConsumer:
         # (#78541) — that combination was swallowing complete Telegram group
         # replies after an early/partial multi-message delivery.
         self._turn_split_delivery = False
+        # True when a full-final send timed out in a way that MAY have reached
+        # the platform (``_send_empty_fallback_final`` → "ambiguous").  The
+        # only case where a payload-less delivery flag keeps legacy trust in
+        # ``delivered_final_matches`` (#95382 tightening) — re-sending there
+        # risks a duplicate rather than recovering a loss.
+        self._delivery_ambiguous = False
         self._delivered_commentary_texts: list[str] = []
         # Retains the finalized visible text of each streaming segment so
         # ``has_delivered_text`` can still match after ``_reset_segment_state``
@@ -655,7 +661,22 @@ class GatewayStreamConsumer:
             if self._turn_split_delivery:
                 # #78541: refuse legacy trust for payload-less split delivery.
                 return False
-            return None
+            # #95382 / #98552 class fix: a delivery flag with NO recorded
+            # payload must still be judged against the FINAL content, not
+            # trusted blindly. Every internal flag-setting site records a
+            # payload; a record-less consumer whose visible/streamed text
+            # does not contain the completed response has demonstrably NOT
+            # delivered it (first-edit prefix, mid-stream truncation) — the
+            # flag alone must not suppress the corrective send.
+            if self.has_delivered_text(final_text):
+                return True
+            # The one legitimately ambiguous case keeps legacy trust: a
+            # timed-out full-final send may have reached the platform
+            # (``_send_empty_fallback_final`` → "ambiguous"), so re-sending
+            # risks a duplicate. That site marks itself explicitly.
+            if self._delivery_ambiguous:
+                return None
+            return False
         if self._delivered_final_text.strip() == target:
             return True
         # A segment break / commentary may have delivered the final text
@@ -864,6 +885,7 @@ class GatewayStreamConsumer:
         self._final_response_sent = False
         self._final_content_delivered = False
         self._delivered_final_text = None
+        self._delivery_ambiguous = False
         self._turn_split_delivery = False
         # Native draft streaming: bump the draft_id so the next text segment
         # animates as a fresh preview below the tool-progress bubbles, not
@@ -2196,6 +2218,7 @@ class GatewayStreamConsumer:
                     # client never received the response. Preserve duplicate
                     # suppression for that one uncertain outcome.
                     self._final_content_delivered = True
+                    self._delivery_ambiguous = True
                 else:
                     # A confirmed failure leaves the gateway free to perform
                     # its normal final send.
@@ -2934,6 +2957,10 @@ class GatewayStreamConsumer:
         self._last_sent_text = text
         if is_turn_final:
             self._final_response_sent = True
+            # Fresh send carried exactly ``text`` — record it so the gateway
+            # can reconcile the flag against the completed response
+            # (#71643/#95382 content-vs-flag contract).
+            self._record_turn_final_payload(text)
         return True
 
     async def _suppress_silence_marker(self) -> None:
@@ -2999,6 +3026,7 @@ class GatewayStreamConsumer:
         self._final_response_sent = False
         self._final_content_delivered = False
         self._delivered_final_text = None
+        self._delivery_ambiguous = False
         self._turn_split_delivery = False
         logger.info(
             "Suppressed streamed intentional-silence marker (chat=%s)",
@@ -3163,6 +3191,11 @@ class GatewayStreamConsumer:
             if _optimistic_finalize:
                 self._final_response_sent = True
                 self._final_content_delivered = True
+                # Record what this finalize frame carries so the gateway's
+                # content reconciliation (#71643/#95382) can judge the flag:
+                # a frame holding only a stale/partial snapshot must not
+                # suppress the corrective send of the complete response.
+                self._record_turn_final_payload(text)
 
             ok = False
             try:
@@ -3193,6 +3226,8 @@ class GatewayStreamConsumer:
             if _optimistic_finalize:
                 self._final_response_sent = False
                 self._final_content_delivered = False
+                # Roll back the recorded payload too — nothing was delivered.
+                self._delivered_final_text = None
 
             # Native streaming refused / failed — switch off so this and
             # subsequent frames take the edit/send fallback path below.

plugins/platforms/discord/adapter.py

@@ -142,6 +142,47 @@ import sys
 from pathlib import Path as _Path
 sys.path.insert(0, str(_Path(__file__).resolve().parents[3]))
 
+
+def _is_discord_transport_error(exc: BaseException) -> bool:
+    """Return True for connection-shaped send failures (dead/dropping WS).
+
+    These are the failures where the message demonstrably did NOT reach
+    Discord because the transport itself was down — the delivery-obligation
+    ledger can safely replay them after reconnect (#95382). HTTP-level
+    rejections (permissions, formatting, 4xx) are NOT transport errors and
+    must keep their original error string. Timeouts are excluded: a timed-out
+    send may have reached Discord, so replaying it risks a duplicate.
+    """
+    if isinstance(exc, asyncio.TimeoutError):
+        return False
+    if isinstance(exc, (ConnectionError, OSError)):
+        return True
+    if DISCORD_AVAILABLE and discord is not None:
+        _transport_types = tuple(
+            t
+            for t in (
+                getattr(discord, "ConnectionClosed", None),
+                getattr(discord, "GatewayNotFound", None),
+                getattr(discord, "DiscordServerError", None),
+            )
+            if isinstance(t, type)
+        )
+        if _transport_types and isinstance(exc, _transport_types):
+            return True
+    text = str(exc).lower()
+    return any(
+        marker in text
+        for marker in (
+            "websocket closed",
+            "connection reset",
+            "connection closed",
+            "session is closed",
+            "cannot write to closing transport",
+            "not connected",
+        )
+    )
+
+
 try:
     from .ffmpeg_utils import resolve_ffmpeg_executable
 except ImportError:
@@ -3451,7 +3492,15 @@ class DiscordAdapter(BasePlatformAdapter):
         created automatically.
         """
         if not self._client:
-            return SendResult(success=False, error="Not connected")
+            # Dead transport (client gone / gateway reconnecting): classify as
+            # send_path_degraded so the delivery-obligation ledger's reconnect
+            # sweep (_redeliver_failed_obligations_for_platform) can replay
+            # this final response once the adapter is live again — a generic
+            # "Not connected" error is not runtime-retryable and left the
+            # turn's output stranded until a full process restart (#95382).
+            return SendResult(
+                success=False, error="send_path_degraded", retryable=True
+            )
         if not (content or "").strip():
             logger.warning(
                 "[%s] Dropped empty message to chat=%s (caller bug). Call site:\n%s",
@@ -3582,7 +3631,16 @@ class DiscordAdapter(BasePlatformAdapter):
 
         except Exception as e:  # pragma: no cover - defensive logging
             logger.error("[%s] Failed to send Discord message: %s", self.name, e, exc_info=True)
-            result = SendResult(success=False, error=str(e))
+            if _is_discord_transport_error(e):
+                # Connection-shaped failure (WS drop / closed session): use
+                # the ledger's runtime-retryable marker so the reconnect
+                # sweep can replay this final response instead of stranding
+                # it until a process restart (#95382 silent partial loss).
+                result = SendResult(
+                    success=False, error="send_path_degraded", retryable=True
+                )
+            else:
+                result = SendResult(success=False, error=str(e))
             await asyncio.to_thread(
                 self._record_discord_response,
                 reply_to=reply_to,

tests/gateway/test_silent_partial_delivery_95382.py

@@ -0,0 +1,480 @@
+"""Regression coverage for #95382 / #98552 — silent partial delivery.
+
+#95382 (Discord): the WebSocket drops after the first streaming edit (which
+carried only a prefix). The consumer's delivery flags could suppress the
+gateway's normal final send even though no recorded payload proved the
+COMPLETE ``final_response`` ever reached the platform; and when the normal
+final send then failed on the dead transport, the failure was recorded with a
+non-retryable error string, so the delivery-obligation ledger's reconnect
+sweep never replayed it — the turn's output was silently lost until a full
+process restart.
+
+#98552 (Telegram): a finalize path that sets ``final_content_delivered=True``
+without recording what was actually delivered produced the same false
+positive on a 624-char message truncated at 333 chars.
+
+Class contract under test:
+
+1. ``delivered_final_matches`` judges a payload-less delivery flag against
+   the FINAL content (via ``has_delivered_text``) instead of returning the
+   legacy-trust ``None`` — only the explicitly-marked ambiguous-timeout path
+   keeps legacy trust.
+2. Every flag-setting site records its delivered payload (fresh-final and
+   the optimistic native finalize were the record-less holdouts).
+3. Discord transport-shaped send failures are classified as
+   ``send_path_degraded`` (retryable) so the ledger reconnect sweep can
+   replay the stranded final response.
+
+Boundary tests drive the REAL ``GatewayRunner._run_agent`` with a live
+``GatewayStreamConsumer`` (pattern from test_stale_finalize_suppression.py).
+"""
+
+import asyncio
+import importlib
+import sys
+import types
+from types import SimpleNamespace
+
+import pytest
+
+from gateway.config import Platform, PlatformConfig, StreamingConfig
+from gateway.platforms.base import BasePlatformAdapter, SendResult
+from gateway.session import SessionSource
+from gateway.stream_consumer import GatewayStreamConsumer, StreamConsumerConfig
+
+
+STREAMED_PREFIX = "Deploy summary: 713 items published (578 as of 08-26"
+MISSING_TAIL = ", another 135 over the past 4 days). All checks green."
+FULL_RESPONSE = STREAMED_PREFIX + MISSING_TAIL
+
+
+# ---------------------------------------------------------------------------
+# Unit coverage — delivered_final_matches tri-state tightening
+# ---------------------------------------------------------------------------
+
+
+def _make_consumer(adapter=None, **overrides):
+    adapter = adapter or SimpleNamespace(
+        MAX_MESSAGE_LENGTH=4096,
+        splits_long_messages=True,
+    )
+    consumer = GatewayStreamConsumer.__new__(GatewayStreamConsumer)
+    consumer.adapter = adapter
+    consumer.chat_id = "c1"
+    consumer.cfg = StreamConsumerConfig(cursor="▉")
+    consumer._final_response_sent = True
+    consumer._final_content_delivered = True
+    consumer._delivered_final_text = None
+    consumer._turn_split_delivery = False
+    consumer._delivery_ambiguous = False
+    consumer._delivered_commentary_texts = []
+    consumer._delivered_segment_texts = []
+    consumer._last_sent_text = ""
+    consumer._accumulated = ""
+    consumer._stream_ledger = ""
+    consumer._initial_reply_to_id = None
+    consumer.metadata = None
+    for key, value in overrides.items():
+        setattr(consumer, key, value)
+    return consumer
+
+
+class TestDeliveredFinalMatchesRecordless:
+    def test_recordless_flag_with_partial_visible_is_mismatch(self):
+        """#95382 core: flag set, no record, visible text is only a prefix —
+        the matcher must return False (recover), not None (legacy trust)."""
+        consumer = _make_consumer(_last_sent_text=STREAMED_PREFIX + "▉")
+        assert consumer.delivered_final_matches(FULL_RESPONSE) is False
+
+    def test_recordless_flag_with_no_visible_text_is_mismatch(self):
+        """Flag set but nothing visibly delivered at all — mismatch."""
+        consumer = _make_consumer()
+        assert consumer.delivered_final_matches(FULL_RESPONSE) is False
+
+    def test_recordless_flag_with_equal_visible_text_matches(self):
+        """Duplicate-suppression control: the visible text IS the final
+        answer — suppression must be retained (True)."""
+        consumer = _make_consumer(_last_sent_text=FULL_RESPONSE + "▉")
+        assert consumer.delivered_final_matches(FULL_RESPONSE) is True
+
+    def test_ambiguous_timeout_keeps_legacy_trust(self):
+        """The explicitly-marked ambiguous full-final timeout is the ONE
+        record-less case that keeps legacy trust (None) — re-sending there
+        risks a duplicate, not a recovery."""
+        consumer = _make_consumer(_delivery_ambiguous=True)
+        assert consumer.delivered_final_matches(FULL_RESPONSE) is None
+
+    def test_recorded_payload_still_wins_over_visible(self):
+        consumer = _make_consumer(
+            _delivered_final_text=FULL_RESPONSE,
+            _last_sent_text="something else entirely",
+        )
+        assert consumer.delivered_final_matches(FULL_RESPONSE) is True
+
+    def test_payloadless_split_still_refuses_trust(self):
+        """#78541 behavior preserved by the tightening."""
+        consumer = _make_consumer(_turn_split_delivery=True)
+        assert consumer.delivered_final_matches(FULL_RESPONSE) is False
+
+    def test_delivered_segment_text_matches(self):
+        """A segment-finalized delivery of the final text still suppresses."""
+        consumer = _make_consumer(
+            _delivered_segment_texts=[FULL_RESPONSE],
+        )
+        assert consumer.delivered_final_matches(FULL_RESPONSE) is True
+
+
+class TestFlagSettingSitesRecordPayload:
+    @pytest.mark.asyncio
+    async def test_fresh_final_records_delivered_payload(self):
+        """_try_fresh_final must record what it sent (#95382 holdout)."""
+
+        class FreshAdapter:
+            MAX_MESSAGE_LENGTH = 4096
+            splits_long_messages = True
+
+            def __init__(self):
+                self.sent = []
+
+            async def send(self, chat_id, content, reply_to=None, metadata=None):
+                self.sent.append(content)
+                return SendResult(success=True, message_id="m-1")
+
+        adapter = FreshAdapter()
+        consumer = _make_consumer(adapter)
+        consumer._final_response_sent = False
+        consumer._final_content_delivered = False
+        consumer._preview_message_ids = set()
+        consumer._message_id = "m-0"
+        consumer._message_created_ts = None
+        consumer.metadata = None
+        consumer._already_sent = False
+
+        ok = await consumer._try_fresh_final(STREAMED_PREFIX, is_turn_final=True)
+        assert ok is True
+        assert consumer._final_response_sent is True
+        # The recorded payload lets the gateway detect a stale fresh-final.
+        assert consumer._delivered_final_text is not None
+        assert STREAMED_PREFIX in consumer._delivered_final_text
+        assert consumer.delivered_final_matches(FULL_RESPONSE) is False
+        assert consumer.delivered_final_matches(STREAMED_PREFIX) is True
+
+
+# ---------------------------------------------------------------------------
+# Gateway-boundary regression — record-less flags must not swallow the reply
+# ---------------------------------------------------------------------------
+
+
+class CaptureAdapter(BasePlatformAdapter):
+    def __init__(self, platform=Platform.DISCORD):
+        super().__init__(PlatformConfig(enabled=True, token="***"), platform)
+        self.sent = []
+        self.edits = []
+        self._next_id = 0
+        self.fail_edits = False
+
+    async def connect(self, *, is_reconnect: bool = False) -> bool:
+        return True
+
+    async def disconnect(self) -> None:
+        return None
+
+    def _mint_id(self) -> str:
+        self._next_id += 1
+        return f"m-{self._next_id}"
+
+    async def send(self, chat_id, content, reply_to=None, metadata=None) -> SendResult:
+        self.sent.append({"chat_id": chat_id, "content": content})
+        return SendResult(success=True, message_id=self._mint_id())
+
+    async def edit_message(
+        self, chat_id, message_id, content, *, finalize: bool = False, metadata=None
+    ) -> SendResult:
+        if self.fail_edits:
+            return SendResult(success=False, error="websocket closed")
+        self.edits.append(
+            {"message_id": message_id, "content": content, "finalize": finalize}
+        )
+        return SendResult(success=True, message_id=message_id)
+
+    async def send_typing(self, chat_id, metadata=None) -> None:
+        return None
+
+    async def stop_typing(self, chat_id) -> None:
+        return None
+
+    async def get_chat_info(self, chat_id: str):
+        return {"id": chat_id}
+
+
+class PrefixOnlyAgent:
+    """Streams only a prefix; the completed response has a longer tail."""
+
+    def __init__(self, **kwargs):
+        self.stream_delta_callback = kwargs.get("stream_delta_callback")
+        self.tools = []
+
+    def run_conversation(self, message, conversation_history=None, task_id=None):
+        if self.stream_delta_callback:
+            self.stream_delta_callback(STREAMED_PREFIX)
+        return {
+            "final_response": FULL_RESPONSE,
+            "response_previewed": False,
+            "messages": [],
+            "api_calls": 1,
+        }
+
+
+class _RecordlessFlagConsumer(GatewayStreamConsumer):
+    """Sabotage subclass: models the #95382/#98552 incident state.
+
+    After a normal drain, claim final delivery via the flags but scrub the
+    recorded payload — the pre-fix gateway read matcher ``None`` as legacy
+    trust and suppressed the corrective send even though only the prefix was
+    ever visible.
+    """
+
+    async def run(self):
+        await super().run()
+        self._final_response_sent = True
+        self._final_content_delivered = True
+        self._turn_split_delivery = False
+        self._delivered_final_text = None
+        # Only the prefix was ever on screen.
+        self._last_sent_text = STREAMED_PREFIX
+        self._delivered_segment_texts = []
+        self._delivered_commentary_texts = []
+
+
+def _make_runner(adapter):
+    gateway_run = importlib.import_module("gateway.run")
+    runner = object.__new__(gateway_run.GatewayRunner)
+    runner.adapters = {adapter.platform: adapter}
+    runner._voice_mode = {}
+    runner._prefill_messages = []
+    runner._ephemeral_system_prompt = ""
+    runner._reasoning_config = None
+    runner._provider_routing = {}
+    runner._fallback_model = None
+    runner._session_db = None
+    runner._running_agents = {}
+    runner._session_run_generation = {}
+    runner.session_store = SimpleNamespace(_entries={}, _save=lambda: None)
+    runner.hooks = SimpleNamespace(loaded_hooks=False)
+    runner.config = SimpleNamespace(
+        thread_sessions_per_user=False,
+        group_sessions_per_user=False,
+        stt_enabled=False,
+        streaming=StreamingConfig.from_dict(
+            {"enabled": True, "edit_interval": 0.01, "buffer_threshold": 1}
+        ),
+    )
+    return runner
+
+
+async def _run_turn(monkeypatch, tmp_path, *, consumer_cls=None, session_id):
+    import yaml
+
+    (tmp_path / "config.yaml").write_text(
+        yaml.dump(
+            {
+                "display": {"tool_progress": "off", "interim_assistant_messages": False},
+                "streaming": {
+                    "enabled": True,
+                    "edit_interval": 0.01,
+                    "buffer_threshold": 1,
+                },
+            }
+        ),
+        encoding="utf-8",
+    )
+
+    fake_dotenv = types.ModuleType("dotenv")
+    fake_dotenv.load_dotenv = lambda *args, **kwargs: None
+    monkeypatch.setitem(sys.modules, "dotenv", fake_dotenv)
+
+    fake_run_agent = types.ModuleType("run_agent")
+    fake_run_agent.AIAgent = PrefixOnlyAgent
+    monkeypatch.setitem(sys.modules, "run_agent", fake_run_agent)
+
+    gateway_run = importlib.import_module("gateway.run")
+    if consumer_cls is not None:
+        stream_consumer_mod = importlib.import_module("gateway.stream_consumer")
+        monkeypatch.setattr(
+            stream_consumer_mod, "GatewayStreamConsumer", consumer_cls
+        )
+    monkeypatch.setattr(gateway_run, "_hermes_home", tmp_path)
+    monkeypatch.setattr(
+        gateway_run, "_resolve_runtime_agent_kwargs", lambda: {"api_key": "***"}
+    )
+
+    adapter = CaptureAdapter()
+    runner = _make_runner(adapter)
+    source = SessionSource(
+        platform=Platform.DISCORD, chat_id="1534932197436424204", chat_type="group"
+    )
+    result = await runner._run_agent(
+        message="deploy status?",
+        context_prompt="",
+        history=[],
+        source=source,
+        session_id=session_id,
+        session_key=f"agent:main:discord:group:{session_id}",
+    )
+    return adapter, result
+
+
+@pytest.mark.asyncio
+async def test_recordless_delivery_flag_does_not_suppress_complete_response(
+    monkeypatch, tmp_path
+):
+    """#95382 boundary: flags claim delivery, nothing recorded, only the
+    prefix visible — the complete response must NOT be suppressed."""
+    adapter, result = await _run_turn(
+        monkeypatch,
+        tmp_path,
+        consumer_cls=_RecordlessFlagConsumer,
+        session_id="sess-95382-recordless",
+    )
+    assert result["final_response"] == FULL_RESPONSE
+    # Pre-fix behavior: already_sent=True and the tail appears in NO platform
+    # call (silent partial delivery). Post-fix: either the gateway performed
+    # the reconciliation edit itself (full text on the wire), or it declined
+    # to claim delivery so the caller's normal final send delivers it.
+    all_payloads = [c["content"] for c in adapter.sent] + [
+        e["content"] for e in adapter.edits
+    ]
+    delivered_here = any(FULL_RESPONSE in p for p in all_payloads)
+    assert delivered_here or not result.get("already_sent"), (
+        "silent partial delivery: gateway claimed delivery but the complete "
+        f"response never reached the platform; payloads={all_payloads!r}"
+    )
+
+
+@pytest.mark.asyncio
+async def test_normal_streaming_turn_still_suppresses_exactly_once(
+    monkeypatch, tmp_path
+):
+    """Control: an honest streaming turn (finalize edit carries the full
+    response) must still suppress the duplicate normal send."""
+    adapter, result = await _run_turn(
+        monkeypatch, tmp_path, session_id="sess-95382-control"
+    )
+    assert result["final_response"] == FULL_RESPONSE
+    all_payloads = [c["content"] for c in adapter.sent] + [
+        e["content"] for e in adapter.edits
+    ]
+    assert any(FULL_RESPONSE in p for p in all_payloads)
+    full_sends = [c for c in adapter.sent if FULL_RESPONSE in c["content"]]
+    assert len(full_sends) <= 1, f"duplicate final delivery: {full_sends!r}"
+
+
+@pytest.mark.asyncio
+async def test_recordless_flag_with_dead_transport_leaves_normal_send(
+    monkeypatch, tmp_path
+):
+    """#95382 incident shape: the reconciliation edit ALSO fails (dead
+    transport). The gateway must NOT claim already_sent — the normal final
+    send (and, on failure there, the delivery ledger) owns recovery."""
+
+    class _DeadEditRecordlessConsumer(_RecordlessFlagConsumer):
+        async def run(self):
+            await super().run()
+            # Transport dies after the stream drained: every further edit
+            # fails, like a dropped Discord WebSocket.
+            self.adapter.fail_edits = True
+
+    adapter, result = await _run_turn(
+        monkeypatch,
+        tmp_path,
+        consumer_cls=_DeadEditRecordlessConsumer,
+        session_id="sess-95382-dead-transport",
+    )
+    assert result["final_response"] == FULL_RESPONSE
+    assert not result.get("already_sent"), (
+        "gateway claimed delivery although neither the stream nor the "
+        "reconciliation edit put the complete response on the wire"
+    )
+
+
+# ---------------------------------------------------------------------------
+# Discord transport classification + ledger reconnect replay (#95382 lane 2)
+# ---------------------------------------------------------------------------
+
+
+class TestDiscordTransportClassification:
+    def _adapter_module(self):
+        import plugins.platforms.discord.adapter as mod
+
+        return mod
+
+    def test_connection_error_is_transport(self):
+        mod = self._adapter_module()
+        assert mod._is_discord_transport_error(ConnectionError("websocket closed"))
+        assert mod._is_discord_transport_error(
+            RuntimeError("Session is closed")
+        )
+        assert mod._is_discord_transport_error(OSError(104, "Connection reset"))
+
+    def test_http_and_timeout_errors_are_not_transport(self):
+        mod = self._adapter_module()
+        assert not mod._is_discord_transport_error(
+            RuntimeError("error code: 50013: Missing Permissions")
+        )
+        assert not mod._is_discord_transport_error(asyncio.TimeoutError())
+
+    @pytest.mark.asyncio
+    async def test_send_without_client_reports_send_path_degraded(self):
+        mod = self._adapter_module()
+        adapter = mod.DiscordAdapter.__new__(mod.DiscordAdapter)
+        adapter._client = None
+        result = await mod.DiscordAdapter.send(adapter, "c1", "hello")
+        assert result.success is False
+        assert result.error == "send_path_degraded"
+        assert result.retryable is True
+
+
+class TestLedgerReplaysDegradedDiscordSend:
+    def test_reconnect_sweep_claims_degraded_discord_row(self, tmp_path, monkeypatch):
+        """End-to-end ledger check: a final response rejected with
+        ``send_path_degraded`` on Discord is claimed by the runtime
+        reconnect sweep; a generic 'Not connected' row (pre-fix error
+        string) is stranded. This is the exact silent-loss mechanism from
+        the #95382 field logs."""
+        monkeypatch.setenv("HERMES_HOME", str(tmp_path / ".hermes"))
+        import gateway.delivery_ledger as dl
+
+        importlib.reload(dl)
+
+        oid_degraded = dl.compute_obligation_id("sess-a", "msg-1", FULL_RESPONSE)
+        dl.record_obligation(
+            obligation_id=oid_degraded,
+            session_key="agent:main:discord:group:c1",
+            platform="discord",
+            chat_id="c1",
+            thread_id=None,
+            content=FULL_RESPONSE,
+        )
+        dl.mark_attempting(oid_degraded)
+        dl.mark_failed(oid_degraded, "send_path_degraded")
+
+        oid_generic = dl.compute_obligation_id("sess-b", "msg-2", FULL_RESPONSE)
+        dl.record_obligation(
+            obligation_id=oid_generic,
+            session_key="agent:main:discord:group:c2",
+            platform="discord",
+            chat_id="c2",
+            thread_id=None,
+            content=FULL_RESPONSE,
+        )
+        dl.mark_attempting(oid_generic)
+        dl.mark_failed(oid_generic, "Not connected")
+
+        claimed = dl.sweep_failed_for_runtime("discord")
+        claimed_ids = {row["obligation_id"] for row in claimed}
+        assert oid_degraded in claimed_ids, (
+            "send_path_degraded Discord row must be replayable after reconnect"
+        )
+        assert oid_generic not in claimed_ids, (
+            "non-transport errors must not be blindly replayed"
+        )

tests/gateway/test_stale_finalize_suppression.py

@@ -373,8 +373,22 @@ def _consumer():
 
 
 class TestDeliveredFinalMatches:
-    def test_no_record_returns_none(self):
+    def test_no_record_no_visible_text_returns_false(self):
+        """#95382 tightening: a record-less consumer with no visible match
+        for the final text is a demonstrable non-delivery, not legacy trust."""
         consumer = _consumer()
+        assert consumer.delivered_final_matches("anything") is False
+
+    def test_no_record_but_visible_final_returns_true(self):
+        """Ambiguous-dedup control: visible text equals the final answer."""
+        consumer = _consumer()
+        consumer._last_sent_text = FULL_RESPONSE
+        assert consumer.delivered_final_matches(FULL_RESPONSE) is True
+
+    def test_no_record_ambiguous_timeout_returns_none(self):
+        """The explicitly-marked ambiguous timeout keeps legacy trust."""
+        consumer = _consumer()
+        consumer._delivery_ambiguous = True
         assert consumer.delivered_final_matches("anything") is None
 
     def test_matching_record_returns_true(self):