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

Merge pull request #72103 from victor-kyriazakos/feat/relay-slack-blockkit-native-parity

Browse source 
fix(relay): Slack DM-root prompts + flat-DM edit-streaming (native _resolve_thread_ts parity)
This commit is contained in:
Teknium 2026-07-29 15:13:42 -07:00 • committed by GitHub
commit fc551f9691
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
5 changed files with 1415 additions and 47 deletions
Download patch file Download diff file Expand all files Collapse all files

51
docs/relay-connector-contract.md
View file

@ -711,7 +711,56 @@ per-gateway secret and the same host as `/relay/provision`.
---
## 8. Versioning policy
## 8. Gateway-side platform behavior controls (enterprise)
Enterprise deployments configure fronted-platform behavior on the GATEWAY
side, under `platforms.relay.extra.<platform>` — a supported subset of that
platform's native options. The native platform block (e.g. `platforms.slack`)
is not read on the relay lane; the connector receives the *outcome* of these
controls as frame metadata (§4) and executes mechanically — it holds no
platform behavior policy of its own.
```yaml
platforms:
relay:
extra:
slack:
reply_in_thread: true # default
```
Resolution: nested `extra.<platform>` object wins → legacy flat key on
`extra` honored as fallback → default. Source of truth:
`RelayAdapter._effective_reply_in_thread` (`gateway/relay/adapter.py`).
Values coerce exactly as the native Slack adapter's do — `1/true/yes/on`
(case-insensitive, whitespace-trimmed) are ON, anything else is OFF — so a
YAML-quoted `"false"` turns a knob off rather than being read as a truthy
string.
Current controls (Slack):
| Key | Default | Effect |
| --- | --- | --- |
| `reply_in_thread` | `true` | `true`: thread-per-message — each top-level DM message anchors its own thread (status, progress, prompts, final reply all carry that `metadata.thread_id`). `false`: flat rolling DM — send-lane frames carry NO thread anchor (stripped, not omitted), one shared session per DM. |
| `dm_top_level_threads_as_sessions` | `true` | Native-parity escape hatch (mirrors `platforms.slack.extra.dm_top_level_threads_as_sessions`). `true`: in thread-per-message mode each top-level DM message keys its own session, so concurrent messages run in parallel. `false`: threaded reply placement is kept but the session stamp is skipped — one rolling DM session (legacy steer/queue posture). No effect in flat mode, which always keeps the single rolling session. |
Typing/status frames always carry the triggering-ts anchor when one is known
(liveliness is unconditional, both modes): Slack's status line is
thread-scoped, and in flat mode the send-side anchor strip guarantees the
status anchor can never leak into reply placement. Semantics of the native
key: see `website/docs/user-guide/messaging/slack.md`.
Thread-anchor resolution applies to EVERY send lane — text (`send`) and media
(`send_media`) alike — through one choke point
(`RelayAdapter._apply_slack_thread_anchor`). Media frames egress via the same
connector-side Slack sender, which threads on `metadata.thread_id` only, so an
attachment resolves its anchor identically to a text reply: promoted into
metadata in thread-per-message mode, stripped in flat mode.
Changes take effect on gateway restart; no connector involvement.
---
## 9. Versioning policy
- `contract_version` is an int; bump **only** for additive changes during the
experimental phase (new optional fields, new `op`s).

503
gateway/relay/adapter.py
View file

@ -72,6 +72,20 @@ 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 -> 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
# reply_to=<triggering message ts> as its edit anchor, but the connector
# maps a raw reply_to to a Slack thread_ts — so a plain DM reply would be
# threaded UNDER the user's message (and lose progressive edit streaming)
# instead of posting flat at the DM root. Native SlackAdapter drops that
# synthetic reply_to in _resolve_thread_ts; the relay lane needs the same
# disambiguation, and it needs the chat_type to know a chat is a DM.
self._chat_type_by_chat: Dict[str, str] = {}
# chat_id -> last triggering message ts (Slack). The typing/status
# lane's synthetic thread anchor in thread-per-message mode;
# see _capture_scope and send_typing.
self._last_inbound_ts_by_chat: Dict[str, str] = {}
# chat_id -> the UNDERLYING platform (e.g. "discord", "telegram") this
# chat belongs to (Phase 1.5 multi-platform-per-agent). One relay adapter
# fronts N platforms on one WS; an outbound reply must egress through the
@ -123,6 +137,27 @@ class RelayAdapter(BasePlatformAdapter):
def message_len_fn(self) -> Callable[[str], int]:
return _LEN_FNS.get(self.descriptor.len_unit, len)
@property
def supports_status_text(self) -> bool: # type: ignore[override]
"""Whether the fronted platform renders a TEXT status line.
Native parity (rich status text): Slack's typing surface is the
assistant status line ("Finding answers…" next to the bot name), a
text-rendering indicator. When the relay fronts Slack, advertise it so
run.py's live-status lane feeds per-tool phrases via
``set_status_text()`` — exactly the wiring the native SlackAdapter
gets (``supports_status_text = True``). Other fronted platforms keep
textless typing bubbles and must NOT receive phrase traffic.
Property (not class attr) because ONE RelayAdapter class fronts many
platforms; the answer depends on the handshaked descriptor. On a
multi-platform relay this scalar reflects the PRIMARY identity's
platform (same convention as the scalar ``descriptor``); per-chat
egress paths that need chat-accurate capabilities use
``_descriptor_for_chat`` below.
"""
return self.descriptor.platform == Platform.SLACK.value
# ── per-chat capability resolution (Phase 1.5 multi-platform) ─────────
def _descriptor_for_chat(self, chat_id: str) -> CapabilityDescriptor:
"""The capability descriptor governing a specific chat.
@ -279,6 +314,7 @@ class RelayAdapter(BasePlatformAdapter):
async def _on_inbound(self, event) -> None:
"""Bridge a connector-delivered MessageEvent into the normal adapter path."""
self._capture_scope(event)
self._stamp_slack_session_thread(event)
# Phase 3: a structured prompt answer resolves its waiting primitive
# (approval/confirm/clarify) and is CONSUMED — it must not also
# dispatch as a chat message. Unknown/expired prompt ids fall through
@ -288,6 +324,112 @@ class RelayAdapter(BasePlatformAdapter):
await self._localize_inbound_media(event)
await self.handle_message(event)
def _relay_slack_extra(self) -> Dict[str, Any]:
"""The Slack-behavior subset of the RELAY platform config.
Enterprise knob shape (Hermes-config directed, relay-namespaced):
platforms:
relay:
extra:
slack: # supported subset of native Slack fields
reply_in_thread: true
The native ``platforms.slack`` block keeps meaning "native adapter
settings"; relay-fronted Slack reads its subset here. Legacy fallback:
a flat key on the relay extra (``extra.reply_in_thread``) still wins
when no ``slack`` object exists, preserving current staging configs.
"""
extra = getattr(self.config, "extra", None) or {}
sub = extra.get("slack")
return sub if isinstance(sub, dict) else extra
@staticmethod
def _coerce_flag(raw: Any, default: bool) -> bool:
"""Coerce an operator-supplied boolean exactly as native Slack does.
Native SlackAdapter reads its behavior flags with
``str(raw).strip().lower() in {"1","true","yes","on"}``, so a
YAML-quoted ``"false"`` — a shape operators write routinely — turns
the flag OFF. A bare ``bool()`` would read that same string as True
(non-empty string), silently ignoring the off switch. These knobs are
documented as native-parity mirrors, so they must coerce identically
or the parity claim only holds for unquoted YAML booleans.
"""
if raw is None:
return default
if isinstance(raw, bool):
return raw
return str(raw).strip().lower() in {"1", "true", "yes", "on"}
def _effective_reply_in_thread(self) -> bool:
"""Resolve the thread-per-message vs flat-DM mode for fronted Slack."""
try:
return self._coerce_flag(
self._relay_slack_extra().get("reply_in_thread"), True
)
except Exception: # noqa: BLE001 - config shape is operator-owned
return True
def _dm_top_level_threads_as_sessions(self) -> bool:
"""Native-parity escape hatch: per-message DM sessions on/off.
Mirrors native SlackAdapter._dm_top_level_threads_as_sessions
(platforms.slack.extra.dm_top_level_threads_as_sessions). Default
True: in thread-per-message mode each top-level DM message keys its
own session (parallel turns). Set
platforms.relay.extra.slack.dm_top_level_threads_as_sessions: false
to keep threaded reply PLACEMENT but ONE rolling DM session — the
legacy steer/queue posture, decoupled from reply_in_thread.
"""
try:
return self._coerce_flag(
self._relay_slack_extra().get("dm_top_level_threads_as_sessions"),
True,
)
except Exception: # noqa: BLE001 - config shape is operator-owned
return True
def _stamp_slack_session_thread(self, event) -> None:
"""Native session-keying parity for fronted Slack DMs.
Native SlackAdapter's inbound handler stamps ``thread_ts =
event.thread_ts or ts`` — every TOP-LEVEL message carries its own ts
as ``source.thread_id``, so build_session_key appends it and each
top-level message gets a FRESH session (per-message threads ⇒
per-message sessions; a 2nd message runs parallel instead of steering
the in-flight turn). The connector normalizes a top-level message
with thread_id=null, so without this stamp every top-level DM
collapses into ONE session key and message 2 pre-empts message 1
("Redirected current run", 2026-07-27 report).
Only in thread-per-message mode: flat mode keeps the shared rolling
DM session on purpose (steer/queue there is the intended UX). Never
overwrites a real thread_id (an in-thread reply must keep resolving
to its thread's session).
"""
try:
src = getattr(event, "source", None)
if not src:
return
platform = getattr(src, "platform", None)
if getattr(platform, "value", platform) != Platform.SLACK.value:
return
if getattr(src, "thread_id", None):
return # real thread — its session key is already correct
message_id = getattr(event, "message_id", None) or getattr(
src, "message_id", None
)
if not message_id:
return
if not self._effective_reply_in_thread():
return
if not self._dm_top_level_threads_as_sessions():
return # opt-out: threaded replies, one rolling session
src.thread_id = str(message_id)
except Exception: # noqa: BLE001 - session stamping must never break inbound
logger.debug("slack session-thread stamp failed", exc_info=True)
async def _localize_inbound_media(self, event) -> None:
"""Download connector re-hosted attachments to local temp paths.
@ -380,10 +522,32 @@ class RelayAdapter(BasePlatformAdapter):
scope = getattr(src, "scope_id", None)
if scope:
self._scope_by_chat[str(chat)] = str(scope)
# Remember the chat_type so send() can suppress the synthetic-DM
# thread anchor on Slack (native _resolve_thread_ts parity). send()
# only receives a chat_id, so it needs this per-chat cache to know a
# chat is a DM.
chat_type = getattr(src, "chat_type", None)
if chat_type:
self._chat_type_by_chat[str(chat)] = str(chat_type)
# Triggering message ts: the typing/status lane's metadata
# (base.py _thread_metadata_for_source) carries NO thread anchor
# for a top-level DM, but in thread-per-message mode the status
# must target the per-message thread (its root = this ts). Cache
# it per chat so send_typing can synthesize the anchor, mirroring
# native send_typing's _resolve_thread_ts(metadata.message_id).
# NOTE: message_id lives on the EVENT (MessageEvent), not the
# source — fall back to source for defensive coverage.
message_id = getattr(event, "message_id", None) or getattr(
src, "message_id", None
)
if message_id:
self._last_inbound_ts_by_chat[str(chat)] = str(message_id)
except Exception: # noqa: BLE001 - scope tracking must never break inbound
pass
def _with_scope(self, chat_id: str, metadata: Optional[Dict[str, Any]]) -> Dict[str, Any]:
def _with_scope(
self, chat_id: str, metadata: Optional[Dict[str, Any]]
) -> Dict[str, Any]:
"""Ensure the outbound metadata carries the discriminator(s) the connector's
egress guard needs to resolve the owning tenant.
@ -542,16 +706,26 @@ class RelayAdapter(BasePlatformAdapter):
else:
text = ""
member = payload.get("member") or {}
user = (member.get("user") if isinstance(member, dict) else None) or payload.get("user") or {}
user = (
(member.get("user") if isinstance(member, dict) else None)
or payload.get("user")
or {}
)
channel_id = str(payload.get("channel_id") or "")
guild_id = payload.get("guild_id") # real Discord interaction wire field
source = SessionSource(
platform=Platform.RELAY,
chat_id=channel_id,
chat_type="channel" if guild_id else "dm",
user_id=str(user.get("id")) if isinstance(user, dict) and user.get("id") else None,
user_name=str(user.get("username")) if isinstance(user, dict) and user.get("username") else None,
scope_id=str(guild_id) if guild_id else None, # Discord guild → generic scope slot
user_id=str(user.get("id"))
if isinstance(user, dict) and user.get("id")
else None,
user_name=str(user.get("username"))
if isinstance(user, dict) and user.get("username")
else None,
scope_id=str(guild_id)
if guild_id
else None, # Discord guild → generic scope slot
message_id=str(payload.get("id")) if payload.get("id") else None,
)
event = MessageEvent(text=text, message_type=message_type, source=source)
@ -567,7 +741,9 @@ class RelayAdapter(BasePlatformAdapter):
prompt_id, option_id = decoded
msg = payload.get("message") or {}
prompt_message_id = (
str(msg.get("id")) if isinstance(msg, dict) and msg.get("id") else None
str(msg.get("id"))
if isinstance(msg, dict) and msg.get("id")
else None
)
event.prompt_response = {
"prompt_id": prompt_id,
@ -620,7 +796,9 @@ class RelayAdapter(BasePlatformAdapter):
sub_name = str(opt.get("name") or "").strip()
if sub_name:
parts.append(sub_name)
parts.extend(RelayAdapter._render_interaction_options(opt.get("options")))
parts.extend(
RelayAdapter._render_interaction_options(opt.get("options"))
)
else:
value = opt.get("value")
if value is not None and str(value).strip():
@ -744,12 +922,19 @@ 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. 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
)
result = await self._transport.send_outbound(
{
"op": "send",
"chat_id": chat_id,
"content": content,
"reply_to": reply_to,
"reply_to": effective_reply_to,
"metadata": self._with_scope(chat_id, send_metadata),
},
platform=self._platform_by_chat.get(str(chat_id)),
@ -760,6 +945,144 @@ class RelayAdapter(BasePlatformAdapter):
error=result.get("error"),
)
def _resolve_reply_to_for_send(
self,
chat_id: str,
reply_to: Optional[str],
metadata: Optional[Dict[str, Any]],
) -> Optional[str]:
"""Suppress the synthetic-DM thread anchor for a Slack DM reply.
A DM turn's streaming reply is sent with ``reply_to`` = the triggering
message's ts (the stream consumer's ``initial_reply_to_id``, used as the
edit anchor and, on threading platforms, the reply target). The
connector's slackRestSender maps a raw ``reply_to`` to a Slack
``thread_ts``, so a plain DM reply would be posted THREADED under the
user's message instead of flat at the DM root — and a threaded first
send loses the progressive edit-streaming the user sees in a real
thread (the reported symptom: DM/home replies arrive flat, no
progressive edits).
Native Slack Hermes already suppresses this synthetic DM thread anchor:
``SlackAdapter._resolve_thread_ts`` returns ``None`` for a top-level /
DM message when ``reply_in_thread`` is off. The relay lane has no such
disambiguation, so we reproduce it here. run.py already encodes the
real-thread decision in ``metadata["thread_id"]`` (it is set only when
progress threading is active — a real thread, or channel autoThread);
for a DM with no real thread that key is absent. So the rule is:
Slack DM + no real ``thread_id`` in metadata ⇒ drop ``reply_to``.
This posts the reply flat at the DM root and lets the consumer edit its
own first-send ts — streaming works exactly as in a thread. It does NOT:
* reintroduce a synthetic DM thread_id (#18859 / the /sethome
landmine) — it removes an anchor, never adds one;
* regress real-thread streaming — a real thread carries a distinct
``thread_id`` in metadata, so the guard leaves ``reply_to`` alone;
* regress channel autoThread — a channel/group top-level reply carries
``thread_id`` (the message's own ts) in metadata when threading is
on, so it is left alone; and a non-DM chat is never matched here.
"""
if reply_to is None:
return None
if self._platform_by_chat.get(str(chat_id)) != Platform.SLACK.value:
return reply_to
if self._chat_type_by_chat.get(str(chat_id)) != "dm":
return reply_to
md = metadata or {}
if md.get("thread_id") or md.get("thread_ts"):
# A real thread was resolved by run.py — honour it.
return reply_to
# Mode gate (native _resolve_thread_ts parity). The final-reply lane
# (gateway/platforms/base.py) builds metadata from source.thread_id
# ONLY — for a top-level DM that is None, so in thread-per-message
# mode the triggering-ts reply_to here is the final reply's ONLY
# threading signal (run.py's synthetic root feeds just the
# progress/status lane). Dropping it unconditionally exiled the final
# message to the DM root while progress stayed threaded (2026-07-27
# report, same class as the prompt-placement bug). Native SlackAdapter only
# suppresses the anchor when reply_in_thread=false; mirror that.
reply_in_thread = self._effective_reply_in_thread()
if reply_in_thread:
# Thread-per-message: the triggering ts is the thread anchor.
return reply_to
# 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,
@ -817,13 +1140,39 @@ class RelayAdapter(BasePlatformAdapter):
"""
if self._transport is None:
return
# Thread anchor for the status surface. Slack's status line
# ("is thinking…" in the thread's replies footer — works with plain
# chat:write, confirmed on native no-assistant bots) is THREAD-only:
# the connector's typing case no-ops without a thread_ts. But the
# typing lane's metadata (base.py _thread_metadata_for_source) has no
# anchor for a top-level DM — source.thread_id is None — so every
# heartbeat was silently dropped. In thread-per-message mode the
# turn's thread root IS the triggering message ts (run.py's synthetic
# root); synthesize it here from the per-chat inbound cache, exactly
# 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 = 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
# sender renders it on assistant.threads.setStatus — the same phrase
# the native adapter shows ("is running pytest…", "Finding answers…").
# Absent (None/empty) => omit content; the connector falls back to its
# default "is typing…" heartbeat, preserving pre-phrase behaviour on
# every platform. Never send empty-string content here: on Slack that
# is the explicit CLEAR request reserved for stop_typing.
frame: Dict[str, Any] = {
"op": "typing",
"chat_id": chat_id,
"metadata": self._with_scope(chat_id, md),
}
phrase = getattr(self, "_status_text", {}).get(str(chat_id))
if phrase:
frame["content"] = str(phrase)
try:
await self._transport.send_outbound(
{
"op": "typing",
"chat_id": chat_id,
"metadata": self._with_scope(chat_id, metadata),
},
frame,
platform=self._platform_by_chat.get(str(chat_id)),
)
except Exception: # noqa: BLE001 - typing is cosmetic, never breaks a turn
@ -852,13 +1201,17 @@ 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, 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(
{
"op": "typing",
"chat_id": chat_id,
"content": "",
"metadata": self._with_scope(chat_id, metadata),
"metadata": self._with_scope(chat_id, md),
},
platform=platform,
)
@ -983,14 +1336,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
@ -1065,8 +1427,12 @@ class RelayAdapter(BasePlatformAdapter):
if result is not None:
return result
return await super().send_image_file(
chat_id, image_path, caption=caption, reply_to=reply_to,
metadata=metadata, **kwargs,
chat_id,
image_path,
caption=caption,
reply_to=reply_to,
metadata=metadata,
**kwargs,
)
async def send_voice(
@ -1091,8 +1457,12 @@ class RelayAdapter(BasePlatformAdapter):
if result is not None:
return result
return await super().send_voice(
chat_id, audio_path, caption=caption, reply_to=reply_to,
metadata=metadata, **kwargs,
chat_id,
audio_path,
caption=caption,
reply_to=reply_to,
metadata=metadata,
**kwargs,
)
async def send_video(
@ -1117,8 +1487,12 @@ class RelayAdapter(BasePlatformAdapter):
if result is not None:
return result
return await super().send_video(
chat_id, video_path, caption=caption, reply_to=reply_to,
metadata=metadata, **kwargs,
chat_id,
video_path,
caption=caption,
reply_to=reply_to,
metadata=metadata,
**kwargs,
)
async def send_document(
@ -1145,13 +1519,20 @@ class RelayAdapter(BasePlatformAdapter):
if result is not None:
return result
return await super().send_document(
chat_id, file_path, caption=caption, file_name=file_name,
reply_to=reply_to, metadata=metadata, **kwargs,
chat_id,
file_path,
caption=caption,
file_name=file_name,
reply_to=reply_to,
metadata=metadata,
**kwargs,
)
# ── Phase 3 interactive: prompt + react ──────────────────────────────
def _mint_prompt(self, kind: str, state: Dict[str, Any], timeout_s: float = 3600.0) -> str:
def _mint_prompt(
self, kind: str, state: Dict[str, Any], timeout_s: float = 3600.0
) -> str:
"""Register a pending prompt and return its 8-hex id.
``state`` carries what the resolver needs when the answer comes back
@ -1171,7 +1552,9 @@ class RelayAdapter(BasePlatformAdapter):
# Opportunistic sweep so abandoned prompts can't accumulate: drop
# anything already expired (cheap — dict is small by construction).
now = time.time()
for stale in [k for k, v in self._pending_prompts.items() if v.get("expires_at", 0) < now]:
for stale in [
k for k, v in self._pending_prompts.items() if v.get("expires_at", 0) < now
]:
self._pending_prompts.pop(stale, None)
return prompt_id
@ -1207,6 +1590,20 @@ class RelayAdapter(BasePlatformAdapter):
"""
if self._transport is None or not self.descriptor.supports_op("prompt"):
return None
# An interactive prompt (approval / clarify / slash-confirm) is emitted
# mid-turn in reply to the triggering inbound event, so `metadata` carries
# that event's thread context (run.py _thread_metadata_for_source stamps
# metadata.thread_id — for a Slack DM the triggering message's own ts,
# used only as a session-keying fallback). Forwarding it makes the
# connector thread the prompt card UNDER the triggering message instead
# of posting it flat at the DM root (the reported bug). Native Slack
# Hermes suppresses this synthetic DM thread anchor; drop it here for the
# same Slack-DM-with-no-real-thread case, matching _resolve_reply_to_for_send.
# Prompt metadata is forwarded VERBATIM. The threading mode is decided
# in exactly one place — run.py's _resolve_progress_thread_id (flat mode
# suppresses the synthetic self-anchor there; thread mode stamps the
# turn's thread). Boundary pinned by test_run_py_suppresses_self_anchor*.
prompt_metadata = metadata
action: Dict[str, Any] = {
"op": "prompt",
"chat_id": chat_id,
@ -1214,8 +1611,10 @@ class RelayAdapter(BasePlatformAdapter):
"prompt_kind": prompt_kind,
"prompt_id": prompt_id,
"options": options,
"reply_to": reply_to,
"metadata": self._with_scope(chat_id, metadata),
"reply_to": self._resolve_reply_to_for_send(
chat_id, reply_to, prompt_metadata
),
"metadata": self._with_scope(chat_id, prompt_metadata),
}
if timeout_s is not None:
action["timeout_s"] = int(timeout_s)
@ -1260,12 +1659,15 @@ class RelayAdapter(BasePlatformAdapter):
button→text fallback takes over (same contract as a native adapter's
failed button send).
"""
options: list = [{"id": "once", "label": "✅ Allow Once", "style": "success"}]
options: list = [{"id": "once", "label": "Allow Once", "style": "primary"}]
if not smart_denied and allow_session:
options.append({"id": "session", "label": "✅ Session", "style": "primary"})
options.append({"id": "session", "label": "Allow Session"})
if allow_permanent:
options.append({"id": "always", "label": "✅ Always", "style": "primary"})
options.append({"id": "deny", "label": "❌ Deny", "style": "danger"})
options.append({
"id": "always",
"label": "Always Allow",
})
options.append({"id": "deny", "label": "Deny", "style": "danger"})
cmd_preview = command if len(command) <= 1500 else command[:1500] + "..."
text = (
@ -1274,7 +1676,9 @@ class RelayAdapter(BasePlatformAdapter):
f"Reason: {description}"
)
if smart_denied:
text += "\n\n**Smart DENY:** owner override applies to this one operation only."
text += (
"\n\n**Smart DENY:** owner override applies to this one operation only."
)
prompt_id = self._mint_prompt(
"exec_approval",
@ -1310,9 +1714,9 @@ class RelayAdapter(BasePlatformAdapter):
gateway's text-intercept flow when the prompt lane is unavailable.
"""
options = [
{"id": "once", "label": "✅ Approve Once", "style": "success"},
{"id": "always", "label": "🔒 Always Approve", "style": "primary"},
{"id": "cancel", "label": "❌ Cancel", "style": "danger"},
{"id": "once", "label": "Approve Once", "style": "primary"},
{"id": "always", "label": "Always Approve"},
{"id": "cancel", "label": "Cancel", "style": "danger"},
]
text = f"**{title}**\n\n{message}" if title else message
prompt_id = self._mint_prompt(
@ -1422,7 +1826,11 @@ class RelayAdapter(BasePlatformAdapter):
if kind == "exec_approval":
from tools.approval import resolve_gateway_approval
choice = option_id if option_id in {"once", "session", "always", "deny"} else "deny"
choice = (
option_id
if option_id in {"once", "session", "always", "deny"}
else "deny"
)
count = resolve_gateway_approval(session_key, choice)
label = {
"once": "✅ Approved once",
@ -1435,13 +1843,17 @@ class RelayAdapter(BasePlatformAdapter):
# Acknowledge in-channel (the connector's prompt message can't
# be edited cross-platform yet — edit support varies; a short
# confirmation preserves the audit trail the native edit gives).
await self.send(chat_id, label, metadata=self._prompt_reply_metadata(event))
await self.send(
chat_id, label, metadata=self._prompt_reply_metadata(event)
)
if count:
self.resume_typing_for_chat(chat_id)
elif kind == "slash_confirm":
from tools import slash_confirm as slash_confirm_mod
choice = option_id if option_id in {"once", "always", "cancel"} else "cancel"
choice = (
option_id if option_id in {"once", "always", "cancel"} else "cancel"
)
result_text = await slash_confirm_mod.resolve(
session_key, str(state.get("confirm_id") or ""), choice
)
@ -1450,13 +1862,20 @@ class RelayAdapter(BasePlatformAdapter):
"always": "🔒 Always approve",
"cancel": "❌ Cancelled",
}.get(choice, "Resolved")
await self.send(chat_id, label, metadata=self._prompt_reply_metadata(event))
await self.send(
chat_id, label, metadata=self._prompt_reply_metadata(event)
)
if result_text:
await self.send(
chat_id, str(result_text), metadata=self._prompt_reply_metadata(event)
chat_id,
str(result_text),
metadata=self._prompt_reply_metadata(event),
)
elif kind == "clarify":
from tools.clarify_gateway import mark_awaiting_text, resolve_gateway_clarify
from tools.clarify_gateway import (
mark_awaiting_text,
resolve_gateway_clarify,
)
clarify_id = str(state.get("clarify_id") or "")
if option_id == "other":

19
gateway/run.py
View file

@ -21924,11 +21924,22 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew
_slack_adapter_for_progress = self._adapter_for_source(source)
if _slack_adapter_for_progress is not None:
try:
_progress_reply_in_thread = bool(
_slack_adapter_for_progress.config.extra.get(
"reply_in_thread", True
)
# Relay lane: the adapter owns mode resolution (nested
# platforms.relay.extra.slack subset with flat-key
# fallback). Native lane: read the flat extra as before.
_mode_fn = getattr(
_slack_adapter_for_progress,
"_effective_reply_in_thread",
None,
)
if callable(_mode_fn):
_progress_reply_in_thread = bool(_mode_fn())
else:
_progress_reply_in_thread = bool(
_slack_adapter_for_progress.config.extra.get(
"reply_in_thread", True
)
)
except Exception:
_progress_reply_in_thread = True
_progress_thread_id = _resolve_progress_thread_id(

403
tests/gateway/relay/test_relay_slack_dm_streaming.py Normal file
View file

@ -0,0 +1,403 @@
"""Slack relay: edit-based streaming of the reply must fire in a DM.
Reported symptom (live): agent responses stream progressively (edit-based) in a
Slack THREAD but arrive FLAT (single message, no progressive edits) in a Slack
DM/home over the relay.
Root cause: a DM turn's streaming reply is sent with
``reply_to = <triggering message ts>`` (the stream consumer's
``initial_reply_to_id`` — its edit anchor). The connector's slackRestSender maps
a raw ``reply_to`` to a Slack ``thread_ts``, so a plain DM reply gets threaded
UNDER the user's message instead of posting flat at the DM root, and a threaded
first send loses the progressive edit streaming the user sees in a real thread.
Native ``SlackAdapter._resolve_thread_ts`` already suppresses this synthetic DM
thread anchor; the relay lane had no such disambiguation.
These are behaviour-contract tests: they assert how the outbound frame relates to
the chat type + thread metadata (the invariant the connector depends on), not a
snapshot. They drive the REAL ``RelayAdapter`` + ``GatewayStreamConsumer`` +
``StubConnector`` end to end.
"""
from __future__ import annotations
import pytest
from gateway.config import Platform, PlatformConfig
from gateway.platforms.base import MessageEvent, MessageType
from gateway.relay.adapter import RelayAdapter
from gateway.relay.descriptor import CONTRACT_VERSION, CapabilityDescriptor
from gateway.session import SessionSource
from gateway.stream_consumer import GatewayStreamConsumer, StreamConsumerConfig
from tests.gateway.relay.stub_connector import StubConnector
def _slack_desc(**kw) -> CapabilityDescriptor:
base = dict(
contract_version=CONTRACT_VERSION,
platform="slack",
label="Slack",
max_message_length=4000,
supports_draft_streaming=False,
supports_edit=True,
supports_threads=True,
markdown_dialect="mrkdwn",
len_unit="chars",
emoji="\U0001f4ac",
platform_hint="",
pii_safe=False,
)
base.update(kw)
return CapabilityDescriptor(**base)
def _wire(chat_id: str, chat_type: str, *, user_id="U1", scope_id=None):
"""A RelayAdapter fronting Slack, with inbound scope captured for chat_id."""
stub = StubConnector(_slack_desc())
adapter = RelayAdapter(PlatformConfig(), _slack_desc(), transport=stub)
src = SessionSource(
platform=Platform.SLACK,
chat_id=chat_id,
chat_type=chat_type,
user_id=user_id,
scope_id=scope_id,
)
adapter._capture_scope(
MessageEvent(text="hi", source=src, message_type=MessageType.TEXT)
)
return adapter, stub
# ---------------------------------------------------------------------------
# The pure disambiguation contract (RelayAdapter.send)
# ---------------------------------------------------------------------------
@pytest.mark.asyncio
async def test_slack_dm_reply_keeps_anchor_in_thread_per_message_mode():
"""Default mode (reply_in_thread=True, thread-per-message): the triggering
ts reply_to is the final reply's ONLY threading signal (base.py builds
metadata from source.thread_id, which is None for a top-level DM) — it
must be KEPT so the final message lands in the per-message thread with
the progress bubbles (2026-07-27 mixed-placement report)."""
adapter, stub = _wire("D1", "dm")
await adapter.send("D1", "the answer", reply_to="1700.0001")
assert len(stub.sent) == 1
frame = stub.sent[0]
assert frame["op"] == "send"
assert frame["reply_to"] == "1700.0001", (
"thread-per-message: the triggering ts anchors the final reply"
)
# 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"
@pytest.mark.asyncio
async def test_slack_dm_reply_drops_synthetic_anchor_in_flat_mode():
"""Flat mode (reply_in_thread=False): the synthetic self-anchor is dropped
so the reply posts flat at the DM root (native _resolve_thread_ts parity)
and no synthetic thread is invented (#18859)."""
adapter, stub = _wire("D1", "dm")
adapter.config.extra = {"reply_in_thread": False}
await adapter.send("D1", "the answer", reply_to="1700.0001")
frame = stub.sent[0]
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_dm_reply_with_real_thread_keeps_anchor():
"""A DM turn that IS inside a real thread (metadata carries a distinct
thread_id) must keep threading — the guard only drops the synthetic anchor."""
adapter, stub = _wire("D1", "dm")
await adapter.send(
"D1", "in thread", reply_to="1700.0002", metadata={"thread_id": "1699.9000"}
)
frame = stub.sent[0]
assert frame["reply_to"] == "1700.0002"
assert frame["metadata"]["thread_id"] == "1699.9000"
@pytest.mark.asyncio
async def test_slack_channel_top_level_reply_keeps_autothread_anchor():
"""A channel top-level reply carries thread_id (its own ts) in metadata when
autoThread is on; the DM-only guard must not touch it."""
adapter, stub = _wire("C1", "channel", scope_id="T1")
await adapter.send(
"C1", "channel reply", reply_to="1700.0003", metadata={"thread_id": "1700.0003"}
)
frame = stub.sent[0]
assert frame["reply_to"] == "1700.0003"
assert frame["metadata"]["thread_id"] == "1700.0003"
@pytest.mark.asyncio
async def test_non_slack_dm_reply_unchanged():
"""The disambiguation is Slack-scoped: a non-Slack relay chat keeps reply_to
(its connector owns its own threading semantics)."""
stub = StubConnector(_slack_desc(platform="discord"))
adapter = RelayAdapter(
PlatformConfig(), _slack_desc(platform="discord"), transport=stub
)
src = SessionSource(
platform=Platform.DISCORD, chat_id="dc1", chat_type="dm", user_id="U1"
)
adapter._capture_scope(
MessageEvent(text="hi", source=src, message_type=MessageType.TEXT)
)
await adapter.send("dc1", "hi", reply_to="msg-9")
assert stub.sent[0]["reply_to"] == "msg-9"
# ---------------------------------------------------------------------------
# End-to-end: the stream consumer keeps edit-streaming in a DM
# ---------------------------------------------------------------------------
async def _drive_stream(adapter, chat_id, *, metadata, initial_reply_to_id, chat_type):
cfg = StreamConsumerConfig(
edit_interval=0.0,
buffer_threshold=1,
transport="edit",
chat_type=chat_type,
)
consumer = GatewayStreamConsumer(
adapter=adapter,
chat_id=chat_id,
config=cfg,
metadata=metadata,
initial_reply_to_id=initial_reply_to_id,
)
# Feed progressive deltas, then finalize — mirrors the live delta callback.
for chunk in ("Hel", "lo ", "world", ". Done."):
consumer.on_delta(chunk)
consumer.finish()
await consumer.run()
return consumer
@pytest.mark.asyncio
async def test_slack_dm_stream_consumer_edits_own_ts_not_flat():
"""A Slack DM turn (chat_type='dm', no thread, metadata None) still builds a
stream consumer that keeps edit support and emits progressive EDITs of the
reply message — the flat-DM regression contract.
The connector returns a real message_id for the flat first send, so edit
support must stay on and at least one edit op must be emitted (progressive
streaming), identical to a thread. No synthetic thread is created.
Runs in EXPLICIT flat mode (reply_in_thread=False) — that is the mode this
contract belongs to; the default thread-per-message path is covered by
test_slack_dm_stream_consumer_threads_in_thread_per_message_mode."""
adapter, stub = _wire("D1", "dm")
adapter.config.extra = {"reply_in_thread": False}
consumer = await _drive_stream(
adapter,
"D1",
metadata=None, # DM: _status_thread_metadata is None in run.py
initial_reply_to_id="1700.0001", # the triggering message ts
chat_type="dm",
)
ops = [f["op"] for f in stub.sent]
# First a flat send; edit support stays on so progressive edits CAN flow
# (exact intermediate-frame timing is covered by the stream_consumer unit
# suite — here we assert the DM regression contract: streaming is not
# self-disabled and every edit targets the reply's own ts).
assert ops[0] == "send"
# Edit support survived: message_id set, not the __no_edit__ sentinel.
assert consumer.message_id and consumer.message_id != "__no_edit__"
assert consumer._edit_supported is True
first_send = stub.sent[0]
# The reply posts FLAT at the DM root — no synthetic thread anchor.
assert first_send["reply_to"] is None
assert "thread_id" not in (first_send["metadata"] or {})
assert "thread_ts" not in (first_send["metadata"] or {})
# reply_to_message_id (the mirrored self-anchor) is stripped too.
assert "reply_to_message_id" not in (first_send["metadata"] or {})
# Any edits that flowed target the same first-send ts (editing its own
# message), never a synthetic thread.
edit_ids = {f["message_id"] for f in stub.sent if f["op"] == "edit"}
assert edit_ids <= {stub.next_send_result["message_id"]}
@pytest.mark.asyncio
async def test_slack_thread_stream_consumer_still_threads_and_streams():
"""Regression guard: a Slack THREAD turn keeps its real thread_id AND streams
(the DM fix must not change the thread path)."""
adapter, stub = _wire("C1", "channel", scope_id="T1")
consumer = await _drive_stream(
adapter,
"C1",
metadata={"thread_id": "1699.9000"},
initial_reply_to_id="1700.0002",
chat_type="channel",
)
ops = [f["op"] for f in stub.sent]
assert ops[0] == "send"
assert consumer._edit_supported is True
first_send = stub.sent[0]
# Thread preserved: the real thread_id rides along and reply_to is kept.
assert first_send["metadata"]["thread_id"] == "1699.9000"
assert first_send["reply_to"] == "1700.0002"
@pytest.mark.asyncio
async def test_slack_dm_stream_consumer_threads_in_thread_per_message_mode():
"""Default mode: the DM stream's first send keeps the triggering-ts anchor
so the streamed final reply lands in the per-message thread; edits still
target the reply's own ts."""
adapter, stub = _wire("D1", "dm")
consumer = await _drive_stream(
adapter,
"D1",
metadata=None,
initial_reply_to_id="1700.0001",
chat_type="dm",
)
first_send = stub.sent[0]
assert first_send["op"] == "send"
assert first_send["reply_to"] == "1700.0001"
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"
# ---------------------------------------------------------------------------
# Operator flags coerce exactly as the native Slack adapter's do.
# ---------------------------------------------------------------------------
@pytest.mark.parametrize(
"raw,expected",
[
(False, False),
("false", False),
("False", False),
(" no ", False),
("off", False),
("0", False),
(True, True),
("true", True),
("yes", True),
("on", True),
("1", True),
],
)
def test_relay_slack_flags_coerce_like_native(raw, expected):
"""A YAML-quoted "false" must turn these knobs OFF, matching native's
str().strip().lower() predicate. A bare bool() would read any non-empty
string as True and silently ignore the operator's off switch."""
adapter, _stub = _wire("D1", "dm")
adapter.config.extra = {
"slack": {
"reply_in_thread": raw,
"dm_top_level_threads_as_sessions": raw,
}
}
assert adapter._effective_reply_in_thread() is expected
assert adapter._dm_top_level_threads_as_sessions() is expected
def test_relay_slack_flags_default_true_when_absent():
"""Both knobs default ON when the operator sets nothing."""
adapter, _stub = _wire("D1", "dm")
adapter.config.extra = {}
assert adapter._effective_reply_in_thread() is True
assert adapter._dm_top_level_threads_as_sessions() is True
# ---------------------------------------------------------------------------
# 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"
)

486
tests/gateway/relay/test_relay_slack_prompt_dm_root.py Normal file
View file

@ -0,0 +1,486 @@
"""Slack relay: interactive prompts follow the turn's thread stamp.
The threading MODE (flat DM vs thread-per-message) is decided in exactly ONE
place: run.py's ``_resolve_progress_thread_id``, which reads
``platforms.slack.extra.reply_in_thread`` and encodes the verdict into the
outbound ``metadata`` stamp:
* flat mode -> the synthetic self-anchor is suppressed in run.py, so prompt
metadata arrives with NO ``thread_id`` and the card posts at the DM root;
* thread-per-message (default) -> ``metadata.thread_id`` is stamped for the
whole turn; on the FIRST turn it legitimately equals the triggering
message's ts (the synthetic root IS the thread).
The prompt lane must TRUST that stamp, like ``_resolve_reply_to_for_send``
does. Re-deriving the mode here (the old unconditional
``thread_id == message_id`` strip) exiled the approval card and its
resolved-state swap to the DM root while progress bubbles honoured the thread
(the 2026-07-27 mixed-placement report).
These are behaviour-contract tests: they assert how the outbound ``prompt``
frame relates to the inherited thread metadata (the invariant the connector
depends on), not a snapshot. They drive the REAL ``RelayAdapter`` +
``StubConnector`` end to end.
"""
from __future__ import annotations
import pytest
from gateway.config import Platform, PlatformConfig
from gateway.platforms.base import MessageEvent, MessageType
from gateway.relay.adapter import RelayAdapter
from gateway.relay.descriptor import CONTRACT_VERSION, CapabilityDescriptor
from gateway.session import SessionSource
from tests.gateway.relay.stub_connector import StubConnector
FULL_OPS = ("send", "edit", "typing", "get_chat_info", "send_media", "prompt", "react")
def _slack_desc(**kw) -> CapabilityDescriptor:
base = dict(
contract_version=CONTRACT_VERSION,
platform="slack",
label="Slack",
max_message_length=4000,
supports_draft_streaming=False,
supports_edit=True,
supports_threads=True,
markdown_dialect="mrkdwn",
len_unit="chars",
supported_ops=FULL_OPS,
)
base.update(kw)
return CapabilityDescriptor(**base)
def _wire(
chat_id: str,
chat_type: str,
*,
user_id="U1",
scope_id=None,
platform=Platform.SLACK,
):
"""A RelayAdapter fronting Slack, with inbound scope + chat_type captured."""
stub = StubConnector(_slack_desc())
adapter = RelayAdapter(PlatformConfig(), _slack_desc(), transport=stub)
src = SessionSource(
platform=platform,
chat_id=chat_id,
chat_type=chat_type,
user_id=user_id,
scope_id=scope_id,
)
adapter._capture_scope(
MessageEvent(text="hi", source=src, message_type=MessageType.TEXT)
)
return adapter, stub
def _last_prompt(stub) -> dict:
prompts = [f for f in stub.sent if f["op"] == "prompt"]
assert prompts, "expected a prompt op on the wire"
return prompts[-1]
# ---------------------------------------------------------------------------
# Flat mode: run.py stamps NO thread_id -> the card posts at the DM root.
# ---------------------------------------------------------------------------
@pytest.mark.asyncio
async def test_exec_approval_flat_mode_posts_at_dm_root():
"""Flat-DM turn (reply_in_thread=false): run.py suppressed the synthetic
anchor upstream, so prompt metadata has no thread_id and none appears on
the wire — the card posts at the DM root."""
adapter, stub = _wire("D1", "dm", scope_id="T1")
md = {"message_id": "1700000000.000100", "scope_id": "T1"}
result = await adapter.send_exec_approval(
"D1", "rm -rf /tmp/x", "sess:1", description="deletes files", metadata=md
)
assert result.success is True
frame = _last_prompt(stub)
meta = frame["metadata"] or {}
assert "thread_id" not in meta
assert "thread_ts" not in meta
# reply_to on the outbound action stays unset — a root-level post.
assert frame["reply_to"] is None
# Tenant scope is preserved untouched (egress routing must not break).
assert meta.get("scope_id") == "T1"
# ---------------------------------------------------------------------------
# Thread-per-message mode, end-to-end placement contract: run.py stamps the
# turn's thread (first turn: the triggering message's own ts) and the adapter
# forwards prompt metadata UNTOUCHED — no re-derivation, no strip. Mixed
# placement (progress threaded, card at root) was the 2026-07-27 regression.
# ---------------------------------------------------------------------------
@pytest.mark.asyncio
async def test_exec_approval_forwards_run_py_thread_stamp_untouched():
"""The adapter must forward run.py's thread stamp verbatim: the approval
card posts INTO the stamped thread. Any adapter-side re-derivation or
strip exiled the card to the home channel (2026-07-27 report)."""
adapter, stub = _wire("D1", "dm", scope_id="T1")
md = {
"thread_id": "1700000000.000100",
"message_id": "1700000000.000100",
"scope_id": "T1",
}
result = await adapter.send_exec_approval(
"D1", "rm -rf /tmp/x", "sess:1", description="deletes files", metadata=md
)
assert result.success is True
frame = _last_prompt(stub)
meta = frame["metadata"] or {}
assert meta.get("thread_id") == "1700000000.000100", (
"first-turn self-anchor is the thread root; the prompt must honour it"
)
assert meta.get("scope_id") == "T1"
@pytest.mark.asyncio
async def test_clarify_forwards_run_py_thread_stamp_untouched():
adapter, stub = _wire("D1", "dm", scope_id="T1")
md = {
"thread_id": "1700000000.000200",
"message_id": "1700000000.000200",
"scope_id": "T1",
}
result = await adapter.send_clarify(
"D1", "Which env?", ["prod", "staging"], "cl-1", "sess:1", metadata=md
)
assert result.success is True
frame = _last_prompt(stub)
meta = frame["metadata"] or {}
assert meta.get("thread_id") == "1700000000.000200"
assert meta.get("scope_id") == "T1"
@pytest.mark.asyncio
async def test_slash_confirm_forwards_run_py_thread_stamp_untouched():
"""The forward-untouched rule covers every prompt surface (single
_send_prompt choke point)."""
adapter, stub = _wire("D1", "dm")
md = {"thread_id": "1700000000.000300", "message_id": "1700000000.000300"}
await adapter.send_slash_confirm(
"D1", "Reload MCP", "invalidates cache", "s", "cf-1", metadata=md
)
frame = _last_prompt(stub)
assert (frame["metadata"] or {}).get("thread_id") == "1700000000.000300"
# ---------------------------------------------------------------------------
# Regression guards: a REAL thread and non-DM / non-Slack chats are untouched
# ---------------------------------------------------------------------------
@pytest.mark.asyncio
async def test_exec_approval_in_real_thread_keeps_thread_id():
"""A DM prompt raised inside a REAL thread (thread_id distinct from the
triggering message ts) stays in that thread."""
adapter, stub = _wire("D1", "dm", scope_id="T1")
md = {
"thread_id": "1699000000.999000",
"message_id": "1700000000.000100",
"scope_id": "T1",
}
await adapter.send_exec_approval("D1", "cmd", "s", metadata=md)
frame = _last_prompt(stub)
assert frame["metadata"]["thread_id"] == "1699000000.999000"
@pytest.mark.asyncio
async def test_channel_approval_keeps_thread_id():
"""A Slack CHANNEL prompt keeps its thread_id (autoThread / real thread)."""
adapter, stub = _wire("C1", "channel", scope_id="T1")
md = {
"thread_id": "1700000000.000400",
"message_id": "1700000000.000400",
"scope_id": "T1",
}
await adapter.send_exec_approval("C1", "cmd", "s", metadata=md)
frame = _last_prompt(stub)
assert frame["metadata"]["thread_id"] == "1700000000.000400"
@pytest.mark.asyncio
async def test_non_slack_dm_approval_keeps_thread_id():
"""A non-Slack relay DM keeps thread_id (its connector owns its own
threading semantics)."""
adapter, stub = _wire("dc1", "dm", platform=Platform.DISCORD)
md = {"thread_id": "9000", "message_id": "9000"}
await adapter.send_exec_approval("dc1", "cmd", "s", metadata=md)
frame = _last_prompt(stub)
assert frame["metadata"]["thread_id"] == "9000"
# ---------------------------------------------------------------------------
# Rich status: the relay advertises Slack's text status line and carries
# the live per-tool phrase on the typing frame (native set_status_text parity).
# ---------------------------------------------------------------------------
@pytest.mark.asyncio
async def test_slack_relay_advertises_status_text():
adapter, _stub = _wire("D1", "dm")
assert adapter.supports_status_text is True
@pytest.mark.asyncio
async def test_non_slack_relay_does_not_advertise_status_text():
stub = StubConnector(_slack_desc(platform="discord"))
adapter = RelayAdapter(
PlatformConfig(), _slack_desc(platform="discord"), transport=stub
)
assert adapter.supports_status_text is False
@pytest.mark.asyncio
async def test_typing_carries_live_status_phrase():
"""set_status_text() -> the next typing frame carries the phrase as
content; clearing it (None) reverts to a content-less heartbeat frame
(never an empty string, which is Slack's explicit clear)."""
adapter, stub = _wire("D1", "dm", scope_id="T1")
adapter.set_status_text("D1", "is running pytest…")
await adapter.send_typing("D1", metadata={"scope_id": "T1"})
typing = [f for f in stub.sent if f["op"] == "typing"]
assert typing and typing[-1].get("content") == "is running pytest…"
adapter.set_status_text("D1", None)
await adapter.send_typing("D1", metadata={"scope_id": "T1"})
typing = [f for f in stub.sent if f["op"] == "typing"]
assert "content" not in typing[-1], (
"cleared phrase must omit content (empty string means CLEAR on Slack)"
)
# ---------------------------------------------------------------------------
# Status thread anchor: typing frames synthesize the per-message thread
# root in thread-per-message mode (the status line is thread-only on Slack).
# ---------------------------------------------------------------------------
def _wire_with_ts(chat_id, chat_type, message_id, **kw):
adapter, stub = _wire(chat_id, chat_type, **kw)
src = SessionSource(
platform=Platform.SLACK, chat_id=chat_id, chat_type=chat_type,
user_id="U1", scope_id=kw.get("scope_id"),
)
ev = MessageEvent(
text="hi", source=src, message_type=MessageType.TEXT, message_id=message_id
)
adapter._capture_scope(ev)
return adapter, stub
@pytest.mark.asyncio
async def test_typing_synthesizes_thread_anchor_in_thread_mode():
"""Top-level DM turn, thread-per-message mode: the typing frame gains the
triggering ts as thread_id so the connector's setStatus targets the
per-message thread instead of no-oping threadless."""
adapter, stub = _wire_with_ts("D1", "dm", "1700.0042")
await adapter.send_typing("D1", metadata=None)
typing = [f for f in stub.sent if f["op"] == "typing"]
assert typing and typing[-1]["metadata"].get("thread_id") == "1700.0042"
@pytest.mark.asyncio
async def test_typing_flat_mode_status_anchors_to_trigger_ts_by_default():
"""Flat-DM liveliness: the STATUS still anchors to the triggering ts
(renders in the footer space, no message artifact) while replies stay
flat — the send lane strips its anchors, so placement cannot inherit this."""
adapter, stub = _wire_with_ts("D1", "dm", "1700.0042")
adapter.config.extra = {"reply_in_thread": False}
await adapter.send_typing("D1", metadata=None)
typing = [f for f in stub.sent if f["op"] == "typing"]
assert typing and typing[-1]["metadata"].get("thread_id") == "1700.0042"
@pytest.mark.asyncio
async def test_typing_anchors_unconditionally_in_both_modes():
"""Liveliness is not a preference: the status anchors whenever an inbound
ts exists, regardless of reply_in_thread. Placement safety comes from the
send-side anchor strip, not from suppressing the status."""
for extra in ({}, {"slack": {"reply_in_thread": False}}):
adapter, stub = _wire_with_ts("D1", "dm", "1700.0042")
adapter.config.extra = extra
await adapter.send_typing("D1", metadata=None)
typing = [f for f in stub.sent if f["op"] == "typing"]
assert typing and typing[-1]["metadata"].get("thread_id") == "1700.0042"
@pytest.mark.asyncio
async def test_flat_mode_sends_stay_flat_with_status_anchor_active():
"""The liveliness anchor must NOT leak into reply placement: sends in
flat mode still strip the synthetic anchor (send-lane contract)."""
adapter, stub = _wire_with_ts("D1", "dm", "1700.0042")
adapter.config.extra = {"reply_in_thread": False}
await adapter.send_typing("D1", metadata=None)
await adapter.send("D1", "the answer", reply_to="1700.0042")
frame = [f for f in stub.sent if f["op"] == "send"][-1]
assert frame["reply_to"] is None
assert "thread_id" not in (frame["metadata"] or {})
@pytest.mark.asyncio
async def test_typing_honours_real_thread_anchor():
"""Metadata that already names a thread wins over the synthetic cache."""
adapter, stub = _wire_with_ts("D1", "dm", "1700.0042")
await adapter.send_typing("D1", metadata={"thread_id": "1699.9000"})
typing = [f for f in stub.sent if f["op"] == "typing"]
assert typing[-1]["metadata"]["thread_id"] == "1699.9000"
@pytest.mark.asyncio
async def test_stop_typing_clear_targets_same_synthesized_thread():
"""The clear frame targets the same synthesized thread as the heartbeat
(else the status line sticks)."""
adapter, stub = _wire_with_ts("D1", "dm", "1700.0042")
await adapter.send_typing("D1", metadata=None)
await adapter.stop_typing("D1", metadata=None)
clears = [
f for f in stub.sent if f["op"] == "typing" and f.get("content") == ""
]
assert clears and clears[-1]["metadata"].get("thread_id") == "1700.0042"
# ---------------------------------------------------------------------------
# Session keying: a top-level Slack DM message gets its own ts stamped as
# source.thread_id (native inbound parity) so each message keys a FRESH
# session in thread-per-message mode; flat mode and real threads untouched.
# ---------------------------------------------------------------------------
def _inbound_event(chat_id="D1", message_id="1700.0100", thread_id=None):
src = SessionSource(
platform=Platform.SLACK, chat_id=chat_id, chat_type="dm",
user_id="U1", scope_id="T1", thread_id=thread_id,
)
return MessageEvent(
text="hi", source=src, message_type=MessageType.TEXT,
message_id=message_id,
)
def test_top_level_dm_gets_session_thread_stamp():
adapter, _ = _wire("D1", "dm")
ev = _inbound_event(message_id="1700.0100")
adapter._stamp_slack_session_thread(ev)
assert ev.source.thread_id == "1700.0100"
def test_two_top_level_messages_key_distinct_sessions():
from gateway.session import build_session_key
adapter, _ = _wire("D1", "dm")
e1 = _inbound_event(message_id="1700.0100")
e2 = _inbound_event(message_id="1700.0200")
adapter._stamp_slack_session_thread(e1)
adapter._stamp_slack_session_thread(e2)
k1 = build_session_key(e1.source)
k2 = build_session_key(e2.source)
assert k1 != k2, "each top-level message must be its own session"
def test_real_thread_reply_keeps_its_thread_session():
adapter, _ = _wire("D1", "dm")
ev = _inbound_event(message_id="1700.0300", thread_id="1700.0100")
adapter._stamp_slack_session_thread(ev)
assert ev.source.thread_id == "1700.0100", (
"an in-thread reply must keep resolving to its thread's session"
)
def test_flat_mode_keeps_shared_dm_session():
adapter, _ = _wire("D1", "dm")
adapter.config.extra = {"reply_in_thread": False}
ev = _inbound_event(message_id="1700.0400")
adapter._stamp_slack_session_thread(ev)
assert ev.source.thread_id is None, (
"flat mode: shared rolling DM session (steer/queue) is intended UX"
)
def test_nested_relay_slack_config_subset_wins():
"""Enterprise knob shape: platforms.relay.extra.slack.reply_in_thread."""
adapter, _ = _wire("D1", "dm")
adapter.config.extra = {"slack": {"reply_in_thread": False}}
assert adapter._effective_reply_in_thread() is False
adapter.config.extra = {"slack": {"reply_in_thread": True}}
assert adapter._effective_reply_in_thread() is True
# Legacy flat key still honoured when no nested object exists.
adapter.config.extra = {"reply_in_thread": False}
assert adapter._effective_reply_in_thread() is False
# Default: thread-per-message.
adapter.config.extra = {}
assert adapter._effective_reply_in_thread() is True
# ---------------------------------------------------------------------------
# Cross-module boundary pin (review 2026-07-28): the adapter deliberately has
# NO prompt-side strip — flat-mode placement depends entirely on run.py's
# _resolve_progress_thread_id suppressing the synthetic self-anchor upstream.
# If that suppression regresses, prompt cards silently thread again. These
# tests pin the boundary in BOTH modes so the coupling is load-bearing.
# ---------------------------------------------------------------------------
def test_run_py_suppresses_self_anchor_in_flat_mode():
from gateway.run import _resolve_progress_thread_id
# Flat mode + synthetic self-anchor (thread_id == own message id) => None:
# prompt/progress metadata arrives at the adapter with NO thread anchor.
assert (
_resolve_progress_thread_id(
"slack", "1700.001", "1700.001", reply_in_thread=False
)
is None
)
# Flat mode + REAL thread (ids differ) => the real thread survives.
assert (
_resolve_progress_thread_id(
"slack", "1699.000", "1700.001", reply_in_thread=False
)
== "1699.000"
)
def test_run_py_keeps_self_anchor_in_thread_mode():
from gateway.run import _resolve_progress_thread_id
# Thread-per-message mode: the first-turn self-anchor IS the thread root
# and must flow through to the adapter unchanged.
assert (
_resolve_progress_thread_id(
"slack", "1700.001", "1700.001", reply_in_thread=True
)
== "1700.001"
)
# No source thread at all: Slack synthesizes the root from the message id.
assert (
_resolve_progress_thread_id("slack", None, "1700.001", reply_in_thread=True)
== "1700.001"
)
# ---------------------------------------------------------------------------
# Native parity escape hatch: platforms.relay.extra.slack.
# dm_top_level_threads_as_sessions=false keeps threaded replies but ONE
# rolling DM session (mirrors native SlackAdapter._dm_top_level_threads_as_sessions).
# Without the knob, reply_in_thread alone couples placement AND session
# keying — a posture native operators can express and relay ones could not.
# ---------------------------------------------------------------------------
@pytest.mark.asyncio
async def test_session_stamp_opt_out_keeps_rolling_dm_session():
adapter, stub = _wire("D1", "dm")
adapter.config.extra = {
"slack": {
"reply_in_thread": True,
"dm_top_level_threads_as_sessions": False,
}
}
event = _inbound_event("D1", message_id="1700.0001", thread_id=None)
adapter._stamp_slack_session_thread(event)
assert getattr(event.source, "thread_id", None) is None, (
"opt-out: top-level DM must NOT be stamped — one rolling session"
)
@pytest.mark.asyncio
async def test_session_stamp_default_remains_per_message():
adapter, stub = _wire("D1", "dm")
adapter.config.extra = {"slack": {"reply_in_thread": True}}
event = _inbound_event("D1", message_id="1700.0002", thread_id=None)
adapter._stamp_slack_session_thread(event)
assert getattr(event.source, "thread_id", None) == "1700.0002", (
"default (native parity): per-message sessions stay on"
)