From fd998120c12146ccf20ed9e2d3d400a322fb83f2 Mon Sep 17 00:00:00 2001 From: teknium1 <127238744+teknium1@users.noreply.github.com> Date: Tue, 1 Sep 2026 10:10:24 -0700 Subject: [PATCH] fix(gateway): judge delivery success against final content, not flag trust (#95382, #98552) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 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 | 37 +- plugins/platforms/discord/adapter.py | 62 ++- .../test_silent_partial_delivery_95382.py | 480 ++++++++++++++++++ .../test_stale_finalize_suppression.py | 16 +- 4 files changed, 591 insertions(+), 4 deletions(-) create mode 100644 tests/gateway/test_silent_partial_delivery_95382.py diff --git a/gateway/stream_consumer.py b/gateway/stream_consumer.py index 22aac6e8cc..0cc69e1722 100644 --- a/gateway/stream_consumer.py +++ b/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. diff --git a/plugins/platforms/discord/adapter.py b/plugins/platforms/discord/adapter.py index 79268849df..a98b016bfc 100644 --- a/plugins/platforms/discord/adapter.py +++ b/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, diff --git a/tests/gateway/test_silent_partial_delivery_95382.py b/tests/gateway/test_silent_partial_delivery_95382.py new file mode 100644 index 0000000000..ed602ae7c5 --- /dev/null +++ b/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" + ) diff --git a/tests/gateway/test_stale_finalize_suppression.py b/tests/gateway/test_stale_finalize_suppression.py index 15d591c8c9..bc833a1fc3 100644 --- a/tests/gateway/test_stale_finalize_suppression.py +++ b/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):