Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
13 changes: 11 additions & 2 deletions gateway/run_turn.py
Original file line number Diff line number Diff line change
Expand Up @@ -3633,9 +3633,18 @@ async def _run_agent_mark_streamed_delivery(self, response: Any, turn_ctx: TurnC
ok=("Edited streamed message %s for session %s to include plugin-transformed content.", _sc.message_id, _sk),
fail_result=None, fail_exc="Failed to edit streamed message for session %s: %s",
)
elif _sc is not None and not (_streamed or _previewed or _content_delivered):
# No stream delivery signal was set, so the normal final send is the only answer delivery.
# A consumer's mere existence cannot duplicate a reply and must not raise a delivery alarm.
logger.info(
"Stream consumer completed without user-visible final delivery for session %s: "
"normal final send required (streamed=%s previewed=%s content_delivered=%s "
"transformed=%s final_len=%d).",
_sk, _streamed, _previewed, _content_delivered, _transformed, len(_final),
)
elif _sc is not None:
# DUPLICATE-RISK DIAGNOSTIC: a stream consumer existed but suppression did NOT fire; log the
# decision inputs ("signal never set" vs "ack-pending race").
# A delivery signal existed but could not prove the complete final was rendered. Keep this
# as a warning for the genuine acknowledgement/mismatch diagnostic family.
logger.warning(
"Normal final-send NOT suppressed despite active stream consumer for session %s: "
"streamed=%s previewed=%s content_delivered=%s transformed=%s final_len=%d — "
Expand Down
35 changes: 35 additions & 0 deletions tests/gateway/test_suppression_contract_matrix.py
Original file line number Diff line number Diff line change
Expand Up @@ -26,11 +26,14 @@
"""

import asyncio
import logging
from types import SimpleNamespace

import pytest

from gateway.config import Platform, PlatformConfig
from gateway.platforms.base import BasePlatformAdapter, SendResult
from gateway.session import SessionSource
from gateway.stream_consumer import GatewayStreamConsumer, StreamConsumerConfig

CURSOR = " ▉"
Expand Down Expand Up @@ -232,6 +235,38 @@ async def test_suppression_requires_the_complete_answer_on_the_wire(
)


@pytest.mark.asyncio
async def test_no_delivery_signal_is_not_reported_as_a_duplicate_risk(caplog):
"""A consumer that delivered no final content requires the normal final send.

Its existence alone cannot produce a duplicate: both delivery signals are
false. The diagnostic should retain that fact without raising a duplicate
alert for the health-check monitor.
"""
from gateway.run import GatewayRunner

consumer = SimpleNamespace(
final_response_sent=False,
final_content_delivered=False,
delivered_final_matches=lambda final_text: False,
)
turn_ctx = SimpleNamespace(
stream_consumer_holder=[consumer],
source=SessionSource(platform=Platform.TELEGRAM, chat_id="chat-1"),
session_key="agent:main:telegram:dm:chat-1",
)
response = {"final_response": "Complete final answer."}

with caplog.at_level(logging.INFO, logger="gateway.run"):
await GatewayRunner._run_agent_mark_streamed_delivery(
object.__new__(GatewayRunner), response, turn_ctx
)

assert response.get("already_sent") is not True
assert "possible duplicate send" not in caplog.text
assert "normal final send required" in caplog.text


@pytest.mark.parametrize(
"edit_behaviour,send_behaviour,prefers_fresh_final,interrupt",
list(_scenarios(lying_edits=True)),
Expand Down