mirrors/hermes-agent
1
0
Fork 0
mirror of https://github.com/NousResearch/hermes-agent.git synced 2026-07-31 19:16:29 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 6

fix(relay): apply the Slack thread anchor on the media lane too

Browse source 
The DM thread-anchor contract was resolved only in send(). _send_media() —
backing send_image, send_image_file, send_voice, send_video and send_document
— passed reply_to straight to the frame and never touched metadata, so
attachments egressing through the same connector-side Slack sender got both
failure shapes this branch set out to remove:

  flat mode   → reply_to survives, the image threads UNDER the user's DM
                message (the original reported symptom)
  thread mode → no metadata.thread_id, and threadTs() never reads reply_to,
                so the image lands in the home channel instead of the
                per-message thread

Both are reachable: gateway/run.py delivers agent artifacts through
send_voice/send_document.

Extract the three steps that must always happen together (mode gate, mirrored
reply_to_message_id strip, metadata promotion) into
_apply_slack_thread_anchor and route BOTH lanes through it, so text and media
cannot drift again. The media lane copies caller metadata rather than mutating
it — these helpers are called in loops with a shared mapping.

Also fold send_typing/stop_typing's duplicated status-anchor blocks into
_with_status_thread_anchor. They had already drifted (stop_typing omitted the
platform check) and the clear must target the thread the heartbeat set or the
status line sticks until Slack's own timeout.

Tests: media lane pinned in both modes plus the channel and
no-caller-mutation cases; verified as real by reverting the fix and watching
them fail.
This commit is contained in:
Ben Barclay 2026-07-29 12:03:00 -07:00
parent dd866eef34
commit 33833e232e
2 changed files with 191 additions and 55 deletions
Download patch file Download diff file Expand all files Collapse all files

148
gateway/relay/adapter.py
View file

@ -904,32 +904,12 @@ class RelayAdapter(BasePlatformAdapter):
if self._transport is None:
return SendResult(success=False, error="no transport")
# Native _resolve_thread_ts parity: a Slack DM reply must post flat at
# the DM root, not threaded under the triggering message. Drop the
# synthetic self-anchor from BOTH the top-level reply_to and the mirrored
# metadata.reply_to_message_id so the connector can't thread on either.
effective_reply_to = self._resolve_reply_to_for_send(
# the DM root, not threaded under the triggering message. One shared
# helper resolves the anchor for EVERY egress lane (see
# _apply_slack_thread_anchor) so the text and media lanes cannot drift.
effective_reply_to = self._apply_slack_thread_anchor(
chat_id, reply_to, send_metadata
)
if effective_reply_to is None and reply_to is not None:
send_metadata.pop("reply_to_message_id", None)
# The connector's Slack sender THREADS ON METADATA ONLY —
# threadTs() reads metadata.thread_id/thread_ts and never looks at
# the frame's reply_to. A send whose only threading signal is
# reply_to (base.py's final-reply and fallback lanes build metadata
# from source.thread_id = None for a top-level DM) would post to the
# home channel even though _resolve_reply_to_for_send kept the
# anchor. Promote the surviving anchor into metadata.thread_id so
# the wire carries it where the connector actually reads it. Only
# when the mode gate kept the anchor (thread-per-message / real
# thread) — flat mode already nulled effective_reply_to above.
if (
effective_reply_to is not None
and self._platform_by_chat.get(str(chat_id)) == Platform.SLACK.value
and not (
send_metadata.get("thread_id") or send_metadata.get("thread_ts")
)
):
send_metadata["thread_id"] = str(effective_reply_to)
result = await self._transport.send_outbound(
{
"op": "send",
@ -1010,6 +990,80 @@ class RelayAdapter(BasePlatformAdapter):
# Flat mode: synthetic DM self-anchor — post flat at the DM root.
return None
def _apply_slack_thread_anchor(
self,
chat_id: str,
reply_to: Optional[str],
metadata: Dict[str, Any],
*,
mirror_key: str = "reply_to_message_id",
) -> Optional[str]:
"""Resolve the outbound Slack thread anchor for ONE egress frame.
The single choke point every send lane goes through — text (``send``)
and media (``send_media``) alike. It does three things that must always
happen together, and previously only happened on the text lane:
1. Mode gate: ``_resolve_reply_to_for_send`` drops the synthetic DM
self-anchor in flat mode, keeps it in thread-per-message mode.
2. Mirror strip: when the anchor is dropped, remove the mirrored
``metadata.reply_to_message_id`` too, so the connector cannot
thread on the copy we forgot about.
3. Anchor promotion: the connector's Slack sender THREADS ON METADATA
ONLY — ``threadTs()`` reads ``metadata.thread_id``/``thread_ts``
and never looks at the frame's ``reply_to``. A surviving anchor is
promoted into ``metadata.thread_id`` or the message silently lands
in the home channel instead of the per-message thread.
``metadata`` is mutated in place; the effective ``reply_to`` is
returned. Non-Slack and non-DM chats are untouched by (1), and (3) is
Slack-only, so other fronted platforms keep their existing behaviour.
"""
effective_reply_to = self._resolve_reply_to_for_send(
chat_id, reply_to, metadata
)
if effective_reply_to is None and reply_to is not None:
metadata.pop(mirror_key, None)
if (
effective_reply_to is not None
and self._platform_by_chat.get(str(chat_id)) == Platform.SLACK.value
and not (metadata.get("thread_id") or metadata.get("thread_ts"))
):
metadata["thread_id"] = str(effective_reply_to)
return effective_reply_to
def _with_status_thread_anchor(
self, chat_id: str, metadata: Optional[Dict[str, Any]]
) -> Dict[str, Any]:
"""Copy ``metadata`` with the typing/status thread anchor applied.
Slack's status line is THREAD-scoped: the connector's typing case
no-ops without a thread anchor, and the typing lane's metadata
(base.py ``_thread_metadata_for_source``) carries none for a top-level
DM (``source.thread_id`` is None). Synthesize it from the per-chat
inbound-ts cache, exactly as native ``send_typing`` resolves
``thread_ts`` from ``metadata.message_id``.
Unconditional across both modes: in flat mode the send lane strips its
own anchors (see ``_apply_slack_thread_anchor``), so the status anchor
cannot leak into reply placement, and ``setStatus`` clears without
leaving a message artifact.
Shared by ``send_typing`` and ``stop_typing`` — the clear MUST target
the same thread the heartbeat set, or the status line sticks until
Slack's own timeout. Keeping one implementation is what guarantees it.
"""
md = dict(metadata or {})
if (
not (md.get("thread_id") or md.get("thread_ts"))
and self._platform_by_chat.get(str(chat_id)) == Platform.SLACK.value
and self._chat_type_by_chat.get(str(chat_id)) == "dm"
):
anchor = self._last_inbound_ts_by_chat.get(str(chat_id))
if anchor:
md["thread_id"] = anchor
return md
async def edit_message(
self,
chat_id: str,
@ -1079,22 +1133,7 @@ class RelayAdapter(BasePlatformAdapter):
# like native send_typing resolves thread_ts from metadata.message_id.
# Flat mode (reply_in_thread=false) keeps the no-anchor no-op: there
# is no thread and must not be one (#18859).
md = dict(metadata or {})
if (
not (md.get("thread_id") or md.get("thread_ts"))
and self._platform_by_chat.get(str(chat_id)) == Platform.SLACK.value
and self._chat_type_by_chat.get(str(chat_id)) == "dm"
):
# Thread mode: status targets the per-message thread.
# Flat mode: the status STILL anchors to the triggering ts —
# setStatus renders in the footer space and clears without a
# message artifact, and flat sends strip their anchors
# so reply placement cannot inherit it. Unconditional: liveliness
# is not a preference, it ships in whatever form the mode
# supports (no speculative opt-out knob).
anchor = self._last_inbound_ts_by_chat.get(str(chat_id))
if anchor:
md["thread_id"] = anchor
md = self._with_status_thread_anchor(chat_id, metadata)
# Rich status parity: run.py's live-status lane stashes the
# current per-tool phrase via set_status_text() (base class store).
# Carry it as the typing frame's content so the connector's Slack
@ -1143,18 +1182,10 @@ class RelayAdapter(BasePlatformAdapter):
platform = self._platform_by_chat.get(str(chat_id))
if platform != Platform.SLACK.value:
return
# Clear must target the SAME thread the heartbeat set: apply
# the identical synthetic-anchor rule as send_typing, or the clear
# frame no-ops threadless and the status line sticks until Slack's
# own timeout.
md = dict(metadata or {})
if (
not (md.get("thread_id") or md.get("thread_ts"))
and self._chat_type_by_chat.get(str(chat_id)) == "dm"
):
anchor = self._last_inbound_ts_by_chat.get(str(chat_id))
if anchor:
md["thread_id"] = anchor
# Clear must target the SAME thread the heartbeat set, or the clear
# frame no-ops threadless and the status line sticks until Slack's own
# timeout. Shared helper with send_typing so the two guards cannot drift.
md = self._with_status_thread_anchor(chat_id, metadata)
try:
await self._transport.send_outbound(
{
@ -1286,14 +1317,23 @@ class RelayAdapter(BasePlatformAdapter):
if not uploaded:
return None
source_url = uploaded
# Same Slack thread-anchor contract as the text lane (send). Media
# frames egress through the connector's Slack sender too, so an
# unresolved anchor threads an image under the user's DM message in
# flat mode, and loses the per-message thread entirely in thread mode
# (threadTs() reads metadata only). Route through the shared helper.
media_metadata: Dict[str, Any] = dict(metadata or {})
effective_reply_to = self._apply_slack_thread_anchor(
chat_id, reply_to, media_metadata
)
action: Dict[str, Any] = {
"op": "send_media",
"chat_id": chat_id,
"media_kind": media_kind,
"source_url": source_url,
"content": caption or "",
"reply_to": reply_to,
"metadata": self._with_scope(chat_id, metadata),
"reply_to": effective_reply_to,
"metadata": self._with_scope(chat_id, media_metadata),
}
if filename:
action["filename"] = filename

98
tests/gateway/relay/test_relay_slack_dm_streaming.py
View file

@ -87,7 +87,7 @@ async def test_slack_dm_reply_keeps_anchor_in_thread_per_message_mode():
assert frame["reply_to"] == "1700.0001", (
"thread-per-message: the triggering ts anchors the final reply"
)
# QA-7: the connector's Slack sender threads on metadata.thread_id ONLY
# The connector's Slack sender threads on metadata.thread_id ONLY
# (threadTs() never reads the frame's reply_to), so the surviving anchor
# must be promoted into metadata for the send to actually thread.
assert (frame["metadata"] or {}).get("thread_id") == "1700.0001"
@ -263,3 +263,99 @@ async def test_slack_dm_stream_consumer_threads_in_thread_per_message_mode():
assert consumer.message_id and consumer.message_id != "__no_edit__"
edit_ids = {f["message_id"] for f in stub.sent if f["op"] == "edit"}
assert edit_ids <= {stub.next_send_result["message_id"]}
# ---------------------------------------------------------------------------
# The media lane obeys the SAME thread-anchor contract as the text lane.
#
# send() and _send_media() both egress through the connector's Slack sender,
# so an anchor resolved on only one of them threads an image under the user's
# DM message in flat mode, and loses the per-message thread in thread mode
# (threadTs() reads metadata only). Both lanes route through
# _apply_slack_thread_anchor; these pin that they stay in agreement.
# ---------------------------------------------------------------------------
def _media_wire(chat_id: str, chat_type: str):
"""Like _wire, but the descriptor advertises the send_media op."""
desc = _slack_desc(supported_ops=("send", "edit", "typing", "send_media"))
stub = StubConnector(desc)
adapter = RelayAdapter(PlatformConfig(), desc, transport=stub)
src = SessionSource(
platform=Platform.SLACK,
chat_id=chat_id,
chat_type=chat_type,
user_id="U1",
)
adapter._capture_scope(
MessageEvent(text="hi", source=src, message_type=MessageType.TEXT)
)
return adapter, stub
@pytest.mark.asyncio
async def test_slack_dm_media_keeps_and_promotes_anchor_in_thread_mode():
"""Thread-per-message: an image must land in the per-message thread. The
connector threads on metadata.thread_id only, so the surviving anchor has
to be promoted there — a bare reply_to would post to the home channel."""
adapter, stub = _media_wire("D1", "dm")
await adapter.send_image("D1", "https://example.com/x.png", reply_to="1700.0001")
frame = [f for f in stub.sent if f["op"] == "send_media"][-1]
assert frame["reply_to"] == "1700.0001"
assert (frame["metadata"] or {}).get("thread_id") == "1700.0001", (
"media frame must carry the anchor where the connector reads it"
)
@pytest.mark.asyncio
async def test_slack_dm_media_drops_synthetic_anchor_in_flat_mode():
"""Flat mode: the media frame drops the synthetic self-anchor exactly as
the text lane does, so the image posts flat at the DM root instead of
threading under the user's message, and invents no thread (#18859)."""
adapter, stub = _media_wire("D1", "dm")
adapter.config.extra = {"reply_in_thread": False}
await adapter.send_image("D1", "https://example.com/x.png", reply_to="1700.0001")
frame = [f for f in stub.sent if f["op"] == "send_media"][-1]
assert frame["reply_to"] is None
assert "thread_id" not in (frame["metadata"] or {})
assert "thread_ts" not in (frame["metadata"] or {})
@pytest.mark.asyncio
async def test_slack_channel_media_anchor_untouched():
"""Non-DM chats are outside the synthetic-anchor rule: a channel media
send keeps its reply_to unchanged."""
adapter, stub = _media_wire("C1", "channel")
await adapter.send_image("C1", "https://example.com/x.png", reply_to="1700.0009")
frame = [f for f in stub.sent if f["op"] == "send_media"][-1]
assert frame["reply_to"] == "1700.0009"
@pytest.mark.asyncio
async def test_media_caller_metadata_not_mutated():
"""The anchor promotion must not leak into the caller's dict — media
helpers are called in loops with a shared metadata mapping."""
adapter, stub = _media_wire("D1", "dm")
caller_md = {"user_id": "U1"}
await adapter.send_image(
"D1", "https://example.com/x.png", reply_to="1700.0001", metadata=caller_md
)
assert caller_md == {"user_id": "U1"}, "caller metadata was mutated in place"
# ---------------------------------------------------------------------------
# The status clear targets the same thread the heartbeat set.
# ---------------------------------------------------------------------------
@pytest.mark.asyncio
async def test_typing_and_clear_share_one_status_anchor():
"""send_typing and stop_typing resolve the anchor through one helper: a
clear that no-ops threadless leaves the status line stuck until Slack's
own timeout."""
adapter, stub = _wire("D1", "dm")
adapter._last_inbound_ts_by_chat["D1"] = "1700.0001"
await adapter.send_typing("D1")
await adapter.stop_typing("D1")
typing_frames = [f for f in stub.sent if f["op"] == "typing"]
assert len(typing_frames) == 2
anchors = [(f["metadata"] or {}).get("thread_id") for f in typing_frames]
assert anchors == ["1700.0001", "1700.0001"], (
"the clear must target the thread the heartbeat set"
)