From 2ed91ca39d80680cfcac40293d0750c77d915afe Mon Sep 17 00:00:00 2001 From: kshitijk4poor <82637225+kshitijk4poor@users.noreply.github.com> Date: Sat, 26 Sep 2026 06:27:11 +0530 Subject: [PATCH] fix(gateway): a turn that answers an addressed message keeps the silence fallback reply_expected was read from the event that opened the turn, so an addressed message the same turn ended up answering could still end on a bare silence marker and vanish: - a queued chain paired the terminal turn's display kind with the opener's flag; the recursive run now receives the pending event's flag, persists it, and returns it as queued_terminal_reply_expected beside queued_terminal_display_kind for the outer shaping to read; - pending-message merges (merge_pending_message_event, text batching, busy debounce) and an active-turn redirect folded the new message in but kept the old flag; MessageEvent.absorb_reply_expected now folds it: an addressed message wins, then an unknown one. Also: reply_expected is persisted only when the adapter set it, so rows on other platforms carry no null key; the turn_params pop and getattr reads go (turn_params already flows into TurnContext); silence_allowed is a module import; crash recovery keeps the diagnostic-mute check on machinery turns and applies silence_allowed only to the silence verdict; the DEBUG line no longer fires for machinery turns. --- gateway/platforms/base.py | 4 +++ gateway/platforms/event.py | 5 ++++ gateway/response_filters.py | 6 ++++ gateway/run_busy.py | 3 ++ gateway/run_startup.py | 19 ++++++------ gateway/run_turn.py | 28 +++++++++-------- ...est_discord_triggering_note_persistence.py | 1 + tests/gateway/test_gateway_silence_tokens.py | 30 +++++++++++++++++-- tests/gateway/test_queued_final_ledger.py | 4 +-- 9 files changed, 74 insertions(+), 26 deletions(-) diff --git a/gateway/platforms/base.py b/gateway/platforms/base.py index 1e6dfb7217..6c544cf9e6 100644 --- a/gateway/platforms/base.py +++ b/gateway/platforms/base.py @@ -1765,6 +1765,7 @@ def merge_pending_message_event(pending_messages: Dict[str, MessageEvent], sessi existing.media_text_inlined.extend(incoming_inline_flags) if event.text: existing.text = BasePlatformAdapter._merge_caption(existing.text, event.text) + existing.absorb_reply_expected(event) if existing_is_photo or incoming_is_photo: existing.message_type = MessageType.PHOTO elif existing_type == MessageType.TEXT and event.message_type != MessageType.TEXT: @@ -1779,6 +1780,7 @@ def merge_pending_message_event(pending_messages: Dict[str, MessageEvent], sessi if merge_text and both_text: if event.text: existing.text = _append_text(existing.text, event.text) + existing.absorb_reply_expected(event) return pending_messages[session_key] = event @@ -2515,6 +2517,7 @@ class BasePlatformAdapter(ABC): if event.media_urls: existing.media_urls.extend(event.media_urls) existing.media_types.extend(event.media_types) + existing.absorb_reply_expected(event) existing._last_chunk_len = len(event.text or "") # type: ignore[attr-defined] prior_task = self._pending_text_batch_tasks.get(key) if prior_task and not prior_task.done(): @@ -3784,6 +3787,7 @@ class BasePlatformAdapter(ABC): else: if event.text: state.event.text = _append_text(state.event.text, event.text) + state.event.absorb_reply_expected(event) latest_message_id = getattr(event, "message_id", None) latest_anchor = latest_message_id or getattr(event, "reply_to_message_id", None) if latest_message_id is not None: diff --git a/gateway/platforms/event.py b/gateway/platforms/event.py index 3ed9c74289..9293e21cb9 100644 --- a/gateway/platforms/event.py +++ b/gateway/platforms/event.py @@ -97,6 +97,11 @@ class MessageEvent: # Run-owned final presentation snapshot; never deserialized from ingress metadata. _notification_reply_muted: Optional[bool] = field(default=None, init=False, repr=False, compare=False) + def absorb_reply_expected(self, other: "MessageEvent") -> None: + """One turn now answers *other* too: an addressed message wins, then an unknown one.""" + if self.reply_expected is not True and other.reply_expected is not False: + self.reply_expected = other.reply_expected + def is_command(self) -> bool: """Check if this is a command message (e.g., /new, /reset).""" return self.allow_gateway_control and (self.text or "").lstrip().startswith("/") diff --git a/gateway/response_filters.py b/gateway/response_filters.py index dabc03e9f6..6faf9b7308 100644 --- a/gateway/response_filters.py +++ b/gateway/response_filters.py @@ -128,6 +128,12 @@ def silence_allowed(display_kind: Any, reply_expected: Optional[bool] = None) -> return is_machinery_display_kind(display_kind) or reply_expected is False +def reply_expected_metadata(reply_expected: Optional[bool]) -> dict: + """The persisted user row's ``reply_expected`` key, only when the adapter knew; crash recovery + reads it back to judge a silence marker as the live turn did.""" + return {} if reply_expected is None else {"reply_expected": reply_expected} + + def is_partial_silence_marker(text: Any) -> bool: """True while streamed ``text`` could still resolve to a silence marker. diff --git a/gateway/run_busy.py b/gateway/run_busy.py index 66ae8d9cbe..523aa43474 100644 --- a/gateway/run_busy.py +++ b/gateway/run_busy.py @@ -644,9 +644,12 @@ class GatewayBusySessionMixin: if turn.event is not None and turn.event is not event: turn.event.reply_anchor_override = anchor turn.event.ledger_message_id = inbound_id + turn.event.absorb_reply_expected(event) if turn.ctx is not None: turn.ctx.event_message_id = anchor turn.ctx.inbound_message_id = inbound_id + if turn.event is not None: + turn.ctx.reply_expected = turn.event.reply_expected return True async def _interrupt_running_agent_for_busy_event(self, event: MessageEvent, adapter, running_agent) -> None: diff --git a/gateway/run_startup.py b/gateway/run_startup.py index a5c17c9c88..03f113734e 100644 --- a/gateway/run_startup.py +++ b/gateway/run_startup.py @@ -760,10 +760,13 @@ class GatewayStartupMixin: def _crash_left_reply(self, history: list, started: float, origin) -> Optional[str]: """What a crash-left turn owes, judged as live delivery would have: ``None`` when it never persisted a final reply after *started*; ``""`` when nothing would have been presented (a - silence marker on a machinery turn, a muted diagnostic wake); else the text to send, with a - human turn's bare silence marker replaced by the same notice the live path sends.""" + silence marker on a machinery turn or on a turn the adapter reported as not addressed to the + bot, a muted diagnostic wake); else the text to send, with any other bare silence marker + replaced by the same notice the live path sends.""" from gateway.platforms.base import _strip_media_directives - from gateway.response_filters import is_intentional_silence_response, silence_allowed + from gateway.response_filters import ( + is_intentional_silence_response, is_machinery_display_kind, silence_allowed, + ) from gateway.run import _sanitize_gateway_final_response from gateway.run_turn import _UNEXPECTED_SILENCE_REPLY from gateway.warning_notifications import diagnostic_turn_muted @@ -774,11 +777,7 @@ class GatewayStartupMixin: or (coerce_epoch(last.get("timestamp")) or 0) < started): return None prompt = next((m for m in reversed(visible) if m.get("role") == "user"), {}) - machinery = silence_allowed( - prompt.get("display_kind"), - (prompt.get("display_metadata") or {}).get("reply_expected"), - ) - if machinery: + if is_machinery_display_kind(prompt.get("display_kind")): try: # the owning profile's display policy, as the adapter reads it at delivery scope = self._media_delivery_scope_for_source(origin) except Exception: @@ -788,7 +787,9 @@ class GatewayStartupMixin: if diagnostic_turn_muted(prompt.get("display_metadata"), origin.platform): return "" if is_intentional_silence_response(last["content"]): - return "" if machinery else _UNEXPECTED_SILENCE_REPLY + silent_ok = silence_allowed( + prompt.get("display_kind"), (prompt.get("display_metadata") or {}).get("reply_expected")) + return "" if silent_ok else _UNEXPECTED_SILENCE_REPLY return _strip_media_directives(_sanitize_gateway_final_response(origin.platform, last["content"])).strip() or None @staticmethod diff --git a/gateway/run_turn.py b/gateway/run_turn.py index a03ab0ea8d..81e28dba54 100644 --- a/gateway/run_turn.py +++ b/gateway/run_turn.py @@ -25,7 +25,9 @@ from gateway.config import Platform from gateway.media_repair import repair_explicit_computer_use_media_paths from gateway.platforms.base import BasePlatformAdapter, ProcessingOutcome from gateway.platforms.event import MessageEvent -from gateway.response_filters import display_kind_for_event, is_machinery_display_kind +from gateway.response_filters import ( + display_kind_for_event, is_machinery_display_kind, reply_expected_metadata, silence_allowed, +) from gateway.warning_notifications import diagnostic_metadata, diagnostic_turn_muted, diagnostic_wake_muted from gateway.session import ( SessionSource, _session_key_namespace, build_channel_continuity_note, @@ -1527,7 +1529,6 @@ class GatewayTurnMixin: logging, resume-pending clear, empty-response normalization, and identity-guarded post-compression session_id propagation. Returns ``(response, _intentional_silence, agent_messages)``.""" - from gateway.response_filters import silence_allowed from gateway.run import ( _is_gateway_hidden_reasoning_incomplete_turn, _normalize_empty_agent_response, _sanitize_gateway_final_response, _should_clear_resume_pending_after_turn, @@ -1541,14 +1542,15 @@ class GatewayTurnMixin: # A queued (/queue) chain's TERMINAL turn owns the silence verdict, not the event that # opened the chain: an internal follow-up may go silent, a human one must not. _silence_kind = agent_result.get("queued_terminal_display_kind", persist_user_display_kind) - if _intentional_silence and not silence_allowed(_silence_kind, reply_expected): + _silence_reply_expected = agent_result.get("queued_terminal_reply_expected", reply_expected) + if _intentional_silence and not silence_allowed(_silence_kind, _silence_reply_expected): logger.warning( "silence marker rejected on a user turn: platform=%s chat=%s", _platform_name, source.chat_id or "unknown", ) _intentional_silence = False response = _UNEXPECTED_SILENCE_REPLY - elif _intentional_silence and reply_expected is False: + elif _intentional_silence and not is_machinery_display_kind(_silence_kind): logger.debug( "silence marker suppressed on an unaddressed turn: platform=%s chat=%s", _platform_name, source.chat_id or "unknown", @@ -2209,11 +2211,10 @@ class GatewayTurnMixin: persist_user_message=prepared.persist_user_message, persist_user_timestamp=prepared.persist_user_timestamp, persist_user_display_kind=prepared.persist_user_display_kind, - reply_expected=getattr(event, "reply_expected", None), + reply_expected=event.reply_expected, persist_user_display_metadata={ "gateway_input_owner": prepared.persistence_owner, - "reply_expected": getattr(event, "reply_expected", None), - **diagnostic_metadata(event)}, + **reply_expected_metadata(event.reply_expected), **diagnostic_metadata(event)}, message_type=event.message_type, scheduled_heartbeat=bool(getattr(event, "_heartbeat_session_id", None)), ) @@ -2242,7 +2243,7 @@ class GatewayTurnMixin: agent_result, source, history, session_entry, session_key, _quick_key, run_generation, _run_start_session_id, _platform_name, _msg_start_time, persist_user_display_kind=prepared.persist_user_display_kind, - reply_expected=getattr(event, "reply_expected", None), + reply_expected=event.reply_expected, ) response = self._hmwa_prepend_reasoning(agent_result, response, source, _intentional_silence) _footer_line = self._hmwa_runtime_footer_line(agent_result, source, _turn_seconds) @@ -3104,7 +3105,7 @@ class GatewayTurnMixin: # The one-slot progress/holder containers shared with the callbacks are TurnContext defaults. turn_ctx = TurnContext( - source=source, reply_expected=turn_params.pop("reply_expected", None), message=message, AIAgent=AIAgent, session_key=session_key, + source=source, message=message, AIAgent=AIAgent, session_key=session_key, run_generation=run_generation, _cleanup_progress=_cleanup_progress, _run_still_current=self._run_still_current_fn(session_key, run_generation), progress_queue=queue.Queue() if disp.needs_progress_queue else None, @@ -3737,8 +3738,7 @@ class GatewayTurnMixin: ) # Same silence predicate as the normal path, else this branch leaks the literal marker. if self._is_intentional_silence(_delivery_result, first_response): - from gateway.response_filters import silence_allowed - if silence_allowed(turn_ctx.persist_user_display_kind, getattr(turn_ctx, "reply_expected", None)): + if silence_allowed(turn_ctx.persist_user_display_kind, turn_ctx.reply_expected): logger.info( "Queued follow-up for session %s: suppressing intentional silence marker before continuing.", session_key or "?", @@ -3841,6 +3841,7 @@ class GatewayTurnMixin: # Queued Discord turns carry the same routing note as first turns; persist the authored text. next_persist_message = None next_display_kind = display_kind_for_event(pending_event) + next_reply_expected = pending_event.reply_expected if pending_event is not None else None # See #60671. if pending_event is not None: next_source = getattr(pending_event, "source", None) or source @@ -3917,7 +3918,9 @@ class GatewayTurnMixin: channel_prompt=next_channel_prompt, message_type=next_message_type, persist_user_message=next_persist_message, persist_user_display_kind=next_display_kind, - persist_user_display_metadata=diagnostic_metadata(pending_event) or None, + reply_expected=next_reply_expected, + persist_user_display_metadata={ + **reply_expected_metadata(next_reply_expected), **diagnostic_metadata(pending_event)} or None, ) except asyncio.CancelledError: await _run_followup_processing_hook( @@ -3941,6 +3944,7 @@ class GatewayTurnMixin: **merged, "queued_terminal_inbound_id": next_inbound_id, "queued_terminal_display_kind": next_display_kind, + "queued_terminal_reply_expected": next_reply_expected, "queued_terminal_notification_category": ( (pending_event.metadata or {}).get("notification_category", "result") if pending_event is not None and pending_event.internal else "result"), diff --git a/tests/gateway/test_discord_triggering_note_persistence.py b/tests/gateway/test_discord_triggering_note_persistence.py index 403d2a35a9..078f17978f 100644 --- a/tests/gateway/test_discord_triggering_note_persistence.py +++ b/tests/gateway/test_discord_triggering_note_persistence.py @@ -76,6 +76,7 @@ async def test_queued_followup_persists_authored_text(): ) pending_event = SimpleNamespace( source=source, message_id="6002", channel_prompt=None, message_type=None, internal=False, metadata={}, + reply_expected=None, ) await GatewayRunner._run_agent_queued_followup( diff --git a/tests/gateway/test_gateway_silence_tokens.py b/tests/gateway/test_gateway_silence_tokens.py index 900e219e48..6238f2f5fc 100644 --- a/tests/gateway/test_gateway_silence_tokens.py +++ b/tests/gateway/test_gateway_silence_tokens.py @@ -195,6 +195,7 @@ async def test_queued_human_turn_also_gets_the_visible_fallback(): stream_consumer_holder=[None], mute_notification_reply=False, persist_user_display_kind=None, + reply_expected=None, source=_source(), _status_thread_metadata=None, event_message_id=None, @@ -227,20 +228,25 @@ async def test_queued_terminal_turn_owns_the_silence_verdict(monkeypatch, tmp_pa run_generation=1, _interrupt_depth=0, history=[], _status_thread_metadata=None, context_prompt=None, result_holder=[None]) pending_event = SimpleNamespace(source=_source(), message_id="43", channel_prompt=None, - message_type=None, internal=True, metadata={}) + message_type=None, internal=True, metadata={}, reply_expected=True) merged = await gateway_run.GatewayRunner._run_agent_queued_followup( runner, turn_ctx, adapter=None, pending="hi again", pending_event=pending_event, response="resp", result={"interrupted": True, "messages": []}, stream_task=None) - assert runner._run_agent.await_args.kwargs["persist_user_display_kind"] == "internal_notification" + followup = runner._run_agent.await_args.kwargs + assert followup["persist_user_display_kind"] == "internal_notification" + assert followup["reply_expected"] is True + assert followup["persist_user_display_metadata"]["reply_expected"] is True assert merged["queued_terminal_display_kind"] == "internal_notification" + assert merged["queued_terminal_reply_expected"] is True - def _result(terminal_kind): + def _result(terminal_kind, terminal_reply_expected=None): return { "final_response": "[SILENT]", "tools": [], "history_offset": 0, "last_prompt_tokens": 0, "api_calls": 1, "failed": False, "queued_terminal_inbound_id": "43", "queued_terminal_display_kind": terminal_kind, + "queued_terminal_reply_expected": terminal_reply_expected, "messages": [{"role": "user", "content": "x"}, {"role": "assistant", "content": "[SILENT]"}], } @@ -255,6 +261,24 @@ async def test_queued_terminal_turn_owns_the_silence_verdict(monkeypatch, tmp_pa response = await runner._handle_message_with_agent( _event(internal=True), _source(), "agent:main:telegram:group:-1001:12345", 1) assert response and not is_intentional_silence_response(response) + # Unaddressed opener, addressed terminal turn: visible fallback. + runner = _runner(monkeypatch, tmp_path) + runner._run_agent = AsyncMock(return_value=_result(None, True)) + response = await runner._handle_message_with_agent( + _event(reply_expected=False), _source(), "agent:main:telegram:group:-1001:12345", 1) + assert response and not is_intentional_silence_response(response) + + +@pytest.mark.parametrize("opener, absorbed, merged", [ + (False, True, True), (False, None, None), (True, False, True), (False, False, False), +]) +def test_one_turn_answering_several_messages_is_addressed_if_any_was(opener, absorbed, merged): + """A merged pending message answers both texts, so an addressed one keeps the fallback.""" + from gateway.platforms.base import merge_pending_message_event + + pending = {"k": _event(reply_expected=opener)} + merge_pending_message_event(pending, "k", _event(reply_expected=absorbed), merge_text=True) + assert pending["k"].reply_expected is merged @pytest.mark.asyncio diff --git a/tests/gateway/test_queued_final_ledger.py b/tests/gateway/test_queued_final_ledger.py index 0cf20b6a45..764c733a49 100644 --- a/tests/gateway/test_queued_final_ledger.py +++ b/tests/gateway/test_queued_final_ledger.py @@ -211,7 +211,7 @@ def _chain_runner_and_ctx(followup_return): context_prompt=None, result_holder=[None]) pending_event = SimpleNamespace( source=topic, message_id="6002", channel_prompt=None, message_type=None, - internal=False, metadata={}) + internal=False, metadata={}, reply_expected=None) return GatewayRunner, runner, turn_ctx, pending_event @@ -239,7 +239,7 @@ async def test_a_chained_queued_turn_carries_its_own_inbound_id(): context_prompt=None, result_holder=[None]) pending_event = SimpleNamespace( source=topic, message_id="6002", channel_prompt=None, message_type=None, - internal=False, metadata={}) + internal=False, metadata={}, reply_expected=None) await GatewayRunner._run_agent_queued_followup( runner, turn_ctx, adapter=None, pending="hi again", pending_event=pending_event,