From d26983e4856ee54424fd9daa20df7b4747298eec Mon Sep 17 00:00:00 2001 From: Ben Barclay Date: Wed, 29 Jul 2026 18:00:18 -0700 Subject: [PATCH] fix(gateway): relay TTS attachments + semantic auto-thread rename on the title turn (#74482) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Two relay-lane bugs from live Discord staging testing (2026-07-29): 1. TTS audio never attached over relay (any platform, any lane). _history_media_paths_for_session excluded only the trailing assistant entry from the persisted transcript when building the delivered-media dedup set. The agent persists rows as it produces them, so THIS turn's text_to_speech tool result (media_tag JSON) was already in the transcript at delivery time — the fresh TTS path deduped against itself and extract_media's attachment was silently stripped (response_delivery_dropped for a MEDIA-tag-only reply; fly logs show the exact signature). Fix: exclude everything from the last USER message onward (the current turn); prior-turn dedup unchanged. Affects every platform adapter (native + relay) on the non-streaming delivery path — the streaming path passes explicit history and was unaffected. 2. Connector-auto-created threads never got the LLM session-title rename. The title fires on the FIRST exchange, whose source is the PARENT channel event — the thread didn't exist at ingest, so the Phase 4 auto-thread markers can't be present and _is_discord_auto_thread_lane never matches on the relay title turn (initial titles worked; semantic renames never happened; staging telemetry shows zero thread_rename ops ever sent). Fix: consume the connector's new send-result feedback (paired gateway-gateway PR — contract §SendResult thread_id/auto_thread_name, additive): RelayAdapter.send() caches (thread_id, initial_name) per chat (bounded 256), run.py's title-callback registration + rename lane read it back and pass initial_name as only_if_current_name so the human-rename-wins guard holds on the relay lane too. Native marker path unchanged; connectors that don't stamp the fields degrade to exactly the old behavior. Tests: 3 new (send-result feedback capture, absence, bound) in test_relay_threads.py; 3 new in test_history_media_current_turn.py (current-turn TTS not deduped, prior-turn still deduped, no-user-row fallback). Relay suite 144 passed. --- gateway/platforms/base.py | 28 ++++- gateway/relay/adapter.py | 36 +++++- gateway/run.py | 81 +++++++++++-- tests/gateway/relay/test_relay_threads.py | 62 ++++++++++ .../test_history_media_current_turn.py | 113 ++++++++++++++++++ 5 files changed, 305 insertions(+), 15 deletions(-) create mode 100644 tests/gateway/test_history_media_current_turn.py diff --git a/gateway/platforms/base.py b/gateway/platforms/base.py index 0dee81351f1..98588648db0 100644 --- a/gateway/platforms/base.py +++ b/gateway/platforms/base.py @@ -3418,14 +3418,30 @@ class BasePlatformAdapter(ABC): return None if not transcript: return None - # Exclude the current turn's assistant message, which has already been - # persisted by the time we reach delivery but must not be treated as - # "history" for dedup purposes. + # Exclude the CURRENT TURN entirely — everything from the last user + # message onward. The agent persists rows as it produces them, so by + # delivery time the transcript already contains this turn's tool + # results and assistant reply. The old form dropped only the most + # recent assistant entry, which left THIS turn's tool results in + # "history": a text_to_speech result carrying media_tag then put its + # own path into the dedup set and delivery silently stripped the + # attachment (staging repro 2026-07-29 — `response_delivery_dropped` + # for a MEDIA-tag-only reply; user saw TTS produce no audio). history = list(transcript) - for msg in reversed(history): - if msg.get("role") == "assistant": - history.remove(msg) + last_user_idx = None + for i in range(len(history) - 1, -1, -1): + if history[i].get("role") == "user": + last_user_idx = i break + if last_user_idx is not None: + history = history[:last_user_idx] + else: + # No user row (unusual store shape): keep the old safe behavior of + # at least excluding the trailing assistant reply. + for msg in reversed(history): + if msg.get("role") == "assistant": + history.remove(msg) + break if not history: return None # Avoid circular import: gateway.run already imports this module. diff --git a/gateway/relay/adapter.py b/gateway/relay/adapter.py index f4170d16444..534ffea914f 100644 --- a/gateway/relay/adapter.py +++ b/gateway/relay/adapter.py @@ -20,7 +20,7 @@ from __future__ import annotations import asyncio import logging -from typing import Any, Callable, Dict, Optional, cast +from typing import Any, Callable, Dict, Optional, Tuple, cast from gateway.config import Platform, PlatformConfig from gateway.platforms.base import BasePlatformAdapter, MessageEvent, SendResult @@ -72,6 +72,11 @@ class RelayAdapter(BasePlatformAdapter): # recipient's author binding; we re-attach this user_id as # metadata.user_id on the outbound action so it can. See _capture_scope. self._dm_user_by_chat: Dict[str, str] = {} + # chat_id -> (thread_id, initial_name) of the auto-thread the CONNECTOR + # created for our most recent send into that chat (auto-thread routing + # feedback off SendResult — see send()). Consumed by the gateway's + # semantic thread-rename lane; bounded like the sibling caches. + self._auto_thread_by_chat: Dict[str, Tuple[str, str]] = {} # chat_id -> chat_type (e.g. "dm", "channel", "group") learned from the # inbound event. Used to reproduce native Slack's synthetic-DM-thread # suppression on the relay lane: a DM streaming reply carries @@ -939,12 +944,41 @@ class RelayAdapter(BasePlatformAdapter): }, platform=self._platform_by_chat.get(str(chat_id)), ) + # Auto-thread routing feedback (contract §SendResult): when the + # connector's auto-thread egress policy routed this send into a + # thread it just created, the result carries thread_id (+ the + # initial name). The conversation was keyed on the PARENT channel — + # the thread didn't exist at ingest — so this is the only place the + # gateway learns where the reply landed. The semantic-rename lane + # (auto session title) reads it via auto_thread_info_for_chat. + try: + _at_thread = result.get("thread_id") + _at_name = result.get("auto_thread_name") + if _at_thread and _at_name: + self._auto_thread_by_chat[str(chat_id)] = ( + str(_at_thread), + str(_at_name), + ) + if len(self._auto_thread_by_chat) > 256: + self._auto_thread_by_chat.pop( + next(iter(self._auto_thread_by_chat)), None + ) + except Exception: # noqa: BLE001 - feedback capture must never break send + pass return SendResult( success=bool(result.get("success")), message_id=result.get("message_id"), error=result.get("error"), ) + def auto_thread_info_for_chat( + self, chat_id: str + ) -> Optional[Tuple[str, str]]: + """(thread_id, initial_name) of the auto-thread the connector created + for the most recent send into *chat_id*, if any. Consumed by the + gateway's semantic thread-rename lane (auto session title).""" + return self._auto_thread_by_chat.get(str(chat_id)) + def _resolve_reply_to_for_send( self, chat_id: str, diff --git a/gateway/run.py b/gateway/run.py index 97c5d4a6030..b33f759bd46 100644 --- a/gateway/run.py +++ b/gateway/run.py @@ -44,7 +44,7 @@ from collections import OrderedDict from contextvars import copy_context from pathlib import Path from datetime import datetime -from typing import Awaitable, Callable, Dict, Optional, Any, List, Union, cast +from typing import Awaitable, Callable, Dict, Optional, Any, List, Tuple, Union, cast from agent.async_utils import consume_detached_task_result, safe_schedule_threadsafe from agent.conversation_compression import ( @@ -5328,7 +5328,9 @@ class TurnRunner: effective_session_id, title, ) - elif self._runner._is_discord_auto_thread_lane(ctx.source): + elif self._runner._is_discord_auto_thread_lane(ctx.source) or ( + self._runner._relay_auto_thread_info(ctx.source) is not None + ): maybe_auto_title_kwargs["title_callback"] = lambda title: self._runner._schedule_discord_semantic_thread_rename( ctx.source, effective_session_id, @@ -18807,6 +18809,43 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew and bool(getattr(source, "auto_thread_initial_name", None)) ) + def _relay_auto_thread_info( + self, source: SessionSource + ) -> Optional[Tuple[str, str]]: + """(thread_id, initial_name) when the RELAY connector auto-threaded our + reply to this source's chat — the title-turn sibling of + _is_discord_auto_thread_lane. + + The marker-based check above only lights up for events ARRIVING IN an + auto-created thread (turn 2+). The auto-title fires on the FIRST + exchange, whose source is the PARENT channel event — the thread did + not exist at ingest, so no markers can be present and the native lane + check never matches on the relay title turn (staging repro + 2026-07-29: initial titles fine, semantic renames never happened). + The connector reports where the reply actually landed on the send + result (contract §SendResult thread_id/auto_thread_name); the relay + adapter caches it per chat and this reads it back. + """ + if source.platform != Platform.DISCORD or not source.chat_id: + return None + if not getattr(source, "delivered_via_upstream_relay", False): + return None + adapter = self._adapter_for_source(source) + info_fn = getattr(adapter, "auto_thread_info_for_chat", None) + if not callable(info_fn): + return None + try: + info = info_fn(str(source.chat_id)) + if ( + isinstance(info, tuple) + and len(info) == 2 + and all(isinstance(x, str) for x in info) + ): + return cast(Tuple[str, str], info) + return None + except Exception: + return None + def _sanitize_discord_thread_title(self, title: str) -> str: """Return a Discord-safe semantic thread title from a session title. @@ -18826,9 +18865,19 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew source: SessionSource, session_id: str, title: str, + relay_info: Optional[Tuple[str, str]] = None, ) -> None: - """Best-effort semantic rename of a newly auto-created Discord thread.""" - if not await asyncio.to_thread(self._is_discord_auto_thread_lane, source): + """Best-effort semantic rename of a newly auto-created Discord thread. + + ``relay_info`` is the (thread_id, initial_name) pair from the relay + connector's send-result feedback — supplied on the title turn, where + the source is the parent-channel event and carries no auto-thread + markers (see _relay_auto_thread_info). When absent, the native + marker-based lane supplies thread identity from the source itself. + """ + if relay_info is None and not await asyncio.to_thread( + self._is_discord_auto_thread_lane, source + ): return adapter = self._adapter_for_source(source) if getattr(self, "adapters", None) else None if adapter is None: @@ -18836,12 +18885,18 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew rename_thread = getattr(adapter, "rename_thread", None) if rename_thread is None: return + target_thread_id = relay_info[0] if relay_info else str(source.thread_id) + guard_name = ( + relay_info[1] + if relay_info + else getattr(source, "auto_thread_initial_name", None) + ) thread_name = self._sanitize_discord_thread_title(title) try: await rename_thread( - str(source.thread_id), + target_thread_id, thread_name, - only_if_current_name=getattr(source, "auto_thread_initial_name", None), + only_if_current_name=guard_name, ) except Exception: logger.debug("Failed to rename Discord auto-thread for generated session title", exc_info=True) @@ -18853,8 +18908,16 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew title: str, ) -> None: """Schedule Discord auto-thread rename from the auto-title background thread.""" - if not title or not self._is_discord_auto_thread_lane(source): + relay_info = None + if not title: return + if not self._is_discord_auto_thread_lane(source): + # Relay title turn: the source is the PARENT channel event (the + # thread didn't exist at ingest, so no auto-thread markers). The + # connector's send-result feedback tells us where the reply landed. + relay_info = self._relay_auto_thread_info(source) + if relay_info is None: + return try: loop = asyncio.get_running_loop() except RuntimeError: @@ -18866,7 +18929,9 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew except Exception: copied_source = source future = safe_schedule_threadsafe( - self._rename_discord_auto_thread_for_session_title(copied_source, session_id, title), + self._rename_discord_auto_thread_for_session_title( + copied_source, session_id, title, relay_info=relay_info + ), loop, logger=logger, log_message="Discord semantic thread rename failed to schedule", diff --git a/tests/gateway/relay/test_relay_threads.py b/tests/gateway/relay/test_relay_threads.py index 6b9ebc0e0b7..e511fffdc94 100644 --- a/tests/gateway/relay/test_relay_threads.py +++ b/tests/gateway/relay/test_relay_threads.py @@ -150,3 +150,65 @@ def test_event_from_wire_reply_to_absent_and_partial(): # ── hello command manifest ─────────────────────────────────────────────── + + +# ── auto-thread routing feedback (send-result thread_id) ───────────────── + + +@pytest.mark.asyncio +async def test_send_captures_auto_thread_feedback(): + """A send result carrying thread_id + auto_thread_name (the connector's + auto-thread egress policy routed the reply into a thread it created) + populates auto_thread_info_for_chat for the semantic-rename lane.""" + adapter, stub = _adapter() + + async def send_outbound(action, *, platform=None): + stub.sent.append(action) + return { + "success": True, + "message_id": "m1", + "thread_id": "th-auto-1", + "auto_thread_name": "What is a duck", + } + + stub.send_outbound = send_outbound # type: ignore[method-assign] + result = await adapter.send("chan1", "quack") + assert result.success + assert adapter.auto_thread_info_for_chat("chan1") == ( + "th-auto-1", + "What is a duck", + ) + # Plain results (no auto-thread) leave no feedback for other chats. + assert adapter.auto_thread_info_for_chat("chan-other") is None + + +@pytest.mark.asyncio +async def test_send_without_thread_feedback_leaves_no_info(): + adapter, stub = _adapter() + + async def send_outbound(action, *, platform=None): + return {"success": True, "message_id": "m2"} + + stub.send_outbound = send_outbound # type: ignore[method-assign] + await adapter.send("chan2", "hello") + assert adapter.auto_thread_info_for_chat("chan2") is None + + +@pytest.mark.asyncio +async def test_auto_thread_feedback_is_bounded(): + adapter, stub = _adapter() + + async def send_outbound(action, *, platform=None): + return { + "success": True, + "message_id": "m", + "thread_id": f"th-{action['chat_id']}", + "auto_thread_name": "n", + } + + stub.send_outbound = send_outbound # type: ignore[method-assign] + for i in range(300): + await adapter.send(f"c{i}", "x") + assert len(adapter._auto_thread_by_chat) <= 256 + # Newest entries survive the bound. + assert adapter.auto_thread_info_for_chat("c299") == ("th-c299", "n") diff --git a/tests/gateway/test_history_media_current_turn.py b/tests/gateway/test_history_media_current_turn.py new file mode 100644 index 00000000000..750e59b0974 --- /dev/null +++ b/tests/gateway/test_history_media_current_turn.py @@ -0,0 +1,113 @@ +"""Regression: current-turn TTS media must not be dedup-stripped (#B, staging 2026-07-29). + +_history_media_paths_for_session feeds the delivery-time dedup that stops the +model re-sending files it already delivered in PRIOR turns. The agent persists +transcript rows as it produces them, so by delivery time the CURRENT turn's +tool results (e.g. text_to_speech's MEDIA-tagged JSON) are already in the +persisted transcript. The old implementation excluded only the trailing +assistant entry — leaving this turn's tool rows in "history", so a fresh TTS +path deduped against ITSELF and the attachment was silently stripped +(`response_delivery_dropped`: user saw TTS succeed but no audio arrive). + +Contract: everything from the LAST USER MESSAGE onward is the current turn and +must be excluded from the dedup set; genuinely-prior turns must stay in it. +""" + +from __future__ import annotations + +from typing import Any, Dict, List, Optional + +from gateway.config import PlatformConfig, Platform +from gateway.platforms.base import BasePlatformAdapter, SendResult + + +class _StubStore: + def __init__(self, transcript: List[Dict[str, Any]]) -> None: + self._transcript = transcript + + def load_transcript(self, session_id: str) -> List[Dict[str, Any]]: + return list(self._transcript) + + +class _StubAdapter(BasePlatformAdapter): + """Minimal concrete adapter (BasePlatformAdapter is abstract).""" + + def __init__(self, transcript: List[Dict[str, Any]]) -> None: + super().__init__(PlatformConfig(), Platform.API_SERVER) + self._session_store = _StubStore(transcript) + + async def start(self) -> None: # pragma: no cover - unused + pass + + async def stop(self) -> None: # pragma: no cover - unused + pass + + async def connect(self, *, is_reconnect: bool = False) -> bool: # pragma: no cover - unused + return True + + async def disconnect(self) -> None: # pragma: no cover - unused + pass + + async def get_chat_info(self, chat_id): # pragma: no cover - unused + return {} + + async def send(self, chat_id, content, reply_to=None, metadata=None) -> SendResult: # pragma: no cover - unused + return SendResult(success=True) + + +def _tts_tool_row(path: str) -> Dict[str, Any]: + return { + "role": "tool", + "content": ( + '{"success": true, "file_path": "%s", "media_tag": "MEDIA:%s"}' + % (path, path) + ), + } + + +def test_current_turn_tts_media_not_treated_as_history(): + """This turn's TTS output (persisted before delivery) must NOT be deduped.""" + current = "/opt/data/cache/audio/tts_now.mp3" + transcript = [ + {"role": "user", "content": "hi"}, + {"role": "assistant", "content": "hello"}, + # ── current turn ── + {"role": "user", "content": "say it out loud"}, + {"role": "assistant", "content": "", "tool_calls": [{}]}, + _tts_tool_row(current), + {"role": "assistant", "content": f"MEDIA:{current}"}, + ] + adapter = _StubAdapter(transcript) + paths: Optional[set] = adapter._history_media_paths_for_session("k") + assert not paths or current not in paths + + +def test_prior_turn_media_still_deduped(): + """A file delivered in a PRIOR turn stays in the dedup set.""" + old = "/opt/data/cache/audio/tts_old.mp3" + transcript = [ + {"role": "user", "content": "speak"}, + {"role": "assistant", "content": "", "tool_calls": [{}]}, + _tts_tool_row(old), + {"role": "assistant", "content": f"MEDIA:{old}"}, + # ── current turn ── + {"role": "user", "content": "thanks"}, + {"role": "assistant", "content": "you're welcome"}, + ] + adapter = _StubAdapter(transcript) + paths = adapter._history_media_paths_for_session("k") + assert paths and old in paths + + +def test_no_user_row_falls_back_to_trailing_assistant_exclusion(): + """Unusual store shape (no user rows): keep the old safe behavior.""" + old = "/opt/data/cache/audio/tts_only.mp3" + transcript = [ + _tts_tool_row(old), + {"role": "assistant", "content": f"MEDIA:{old}"}, + ] + adapter = _StubAdapter(transcript) + # Only guarantee: it does not crash and returns a set-or-None; the tool + # row (not excludable without a user anchor) may keep the path present. + result = adapter._history_media_paths_for_session("k") + assert result is None or isinstance(result, set)