mirror of
https://github.com/NousResearch/hermes-agent.git
synced 2026-07-31 19:16:29 +00:00
fix(gateway): relay TTS attachments + semantic auto-thread rename on the title turn (#74482)
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.
This commit is contained in:
Ben Barclay
GitHub
parent
8da8a7887d
commit
d26983e485
5 changed files with 305 additions and 15 deletions
|
|
@ -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.
|
||||
|
|
|
|||
|
|
@ -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,
|
||||
|
|
|
|||
|
|
@ -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",
|
||||
|
|
|
|||
|
|
@ -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")
|
||||
|
|
|
|||
113
tests/gateway/test_history_media_current_turn.py
Normal file
113
tests/gateway/test_history_media_current_turn.py
Normal file
|
|
@ -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)
|
||||
Loading…
Add table
Add a link
Reference in a new issue