Skip to content

fix(gateway): prevent blank auto-resume messages #23314

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
kaiyisg wants to merge 3 commits into
base: main
Choose a base branch
from
Open

fix(gateway): prevent blank auto-resume messages #23314

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
a7603e5
fix(gateway): prevent blank auto-resume messages
kaiyisg May 10, 2026
d043eeb
fix(slack): route replies from mentioned thread parents
kaiyisg May 13, 2026
d63f381
fix(slack): dedupe mentioned socket envelopes
kaiyisg May 14, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
190 changes: 181 additions & 9 deletions gateway/platforms/slack.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -290,6 +290,7 @@ def __init__(self, config: PlatformConfig):
self._bot_user_id: Optional[str] = None
self._user_name_cache: Dict[str, str] = {} # user_id → display name
self._socket_mode_task: Optional[asyncio.Task] = None
self._raw_socket_mode_task: Optional[asyncio.Task] = None
# Multi-workspace support
self._team_clients: Dict[str, Any] = {} # team_id → WebClient
self._team_bot_user_ids: Dict[str, str] = {} # team_id → bot_user_id
Expand Down Expand Up @@ -660,12 +661,22 @@ async def handle_hermes_command(ack, command):
):
self._app.action(_action_id)(self._handle_slash_confirm_action)

# Start Socket Mode handler in background
raw_socket_mode_enabled = self._raw_socket_mode_fallback_enabled()

# Start Socket Mode handler in background. The raw fallback is an
# optional second reader for deployments where Bolt's background
# Socket Mode task goes stale or misses an envelope. Event dedup in
# _handle_slack_message suppresses duplicate replies.
self._handler = AsyncSocketModeHandler(self._app, app_token, proxy=proxy_url)
_apply_slack_proxy(self._handler.client, proxy_url)
self._socket_mode_task = asyncio.create_task(self._handler.start_async())

self._running = True
if raw_socket_mode_enabled:
self._raw_socket_mode_task = asyncio.create_task(
self._raw_socket_mode_fallback(app_token, proxy_url)
)
logger.warning("[Slack] Raw Socket Mode fallback enabled")
logger.info(
"[Slack] Socket Mode connected (%d workspace(s))",
len(self._team_clients),
Expand All @@ -681,6 +692,16 @@ async def handle_hermes_command(ack, command):

async def disconnect(self) -> None:
"""Disconnect from Slack."""
if self._raw_socket_mode_task:
self._raw_socket_mode_task.cancel()
try:
await self._raw_socket_mode_task
except asyncio.CancelledError:
pass
except Exception as e: # pragma: no cover - defensive logging
logger.warning("[Slack] Raw Socket Mode fallback close error: %s", e, exc_info=True)
finally:
self._raw_socket_mode_task = None
if self._handler:
try:
await self._handler.close_async()
Expand All @@ -692,6 +713,106 @@ async def disconnect(self) -> None:

logger.info("[Slack] Disconnected")

def _raw_socket_mode_fallback_enabled(self) -> bool:
configured = self.config.extra.get("socket_raw_fallback")
if configured is None:
configured = os.getenv("SLACK_SOCKET_RAW_FALLBACK", "")
return str(configured).strip().lower() in {"1", "true", "yes", "on"}

async def _raw_socket_mode_fallback(self, app_token: str, proxy_url: Optional[str]) -> None:
"""Read Slack Socket Mode directly when Bolt's background task misses events."""
reconnect_delay = 5
while self._running:
try:
async with aiohttp.ClientSession() as session:
async with session.post(
"https://slack.com/api/apps.connections.open",
headers={"Authorization": f"Bearer {app_token}"},
proxy=proxy_url,
timeout=aiohttp.ClientTimeout(total=20),
) as response:
payload = await response.json()
if not payload.get("ok") or not payload.get("url"):
logger.warning(
"[Slack] Raw Socket Mode fallback connection failed: %s",
payload.get("error") or response.status,
)
await asyncio.sleep(reconnect_delay)
continue

async with session.ws_connect(
payload["url"],
heartbeat=20,
proxy=proxy_url,
) as websocket:
logger.warning("[Slack] Raw Socket Mode fallback connected")
async for message in websocket:
if not self._running:
break
if message.type == aiohttp.WSMsgType.TEXT:
try:
envelope = json.loads(message.data)
except json.JSONDecodeError:
continue
envelope_id = envelope.get("envelope_id")
if envelope_id:
await websocket.send_json({"envelope_id": envelope_id})
event_payload = envelope.get("payload") or {}
event = event_payload.get("event") or {}
if event_payload.get("team_id") and not event.get("team"):
event["team"] = event_payload.get("team_id")
event_type = event.get("type")
channel_id = event.get("channel") or ""
event_text = event.get("text") or ""
bot_mentioned = (
self._bot_user_id
and f"<@{self._bot_user_id}>" in event_text
)
should_dispatch = (
event_type == "app_mention"
or (
event_type == "message"
and (
channel_id.startswith("D")
or bot_mentioned
)
)
)
if should_dispatch:
if (
event_type in {"message", "app_mention"}
and not channel_id.startswith("D")
and event.get("ts")
and (bot_mentioned or event_type == "app_mention")
):
event["_hermes_dedup_key"] = f"mentioned:{event.get('ts')}"
logger.info(
"[Slack] Raw Socket Mode fallback dispatch: %s channel=%s ts=%s",
event_type,
channel_id,
event.get("ts"),
)
event_for_handler = dict(event)
task = asyncio.create_task(self._handle_slack_message(event_for_handler))
task.add_done_callback(self._log_raw_fallback_dispatch_result)
elif message.type in {aiohttp.WSMsgType.CLOSED, aiohttp.WSMsgType.ERROR}:
break
except asyncio.CancelledError:
raise
except Exception as e: # pragma: no cover - network watchdog
if self._running:
logger.warning("[Slack] Raw Socket Mode fallback error: %s", e, exc_info=True)
if self._running:
await asyncio.sleep(reconnect_delay)

def _log_raw_fallback_dispatch_result(self, task: asyncio.Task) -> None:
try:
task.result()
except asyncio.CancelledError:
pass
except Exception as e: # pragma: no cover - diagnostic callback
logger.warning("[Slack] Raw Socket Mode fallback dispatch failed: %s", e, exc_info=True)

def _get_client(self, chat_id: str) -> Any:
"""Return the workspace-specific WebClient for a channel."""
team_id = self._channel_team.get(chat_id)
Expand Down Expand Up @@ -1710,8 +1831,28 @@ async def _handle_assistant_thread_lifecycle_event(self, event: dict) -> None:

async def _handle_slack_message(self, event: dict) -> None:
"""Handle an incoming Slack message event."""
# Dedup: Slack Socket Mode can redeliver events after reconnects (#4777)
event_ts = event.get("ts", "")
# Dedup: Slack Socket Mode can redeliver events after reconnects (#4777).
# Slack can emit both `message` and `app_mention` envelopes for the same
# channel @mention. Normalize mentioned channel posts to one key so the
# agent only gets one turn even when Bolt or a raw Socket Mode fallback
# delivers both envelopes.
event_ts = event.get("_hermes_dedup_key")
if not event_ts:
raw_ts = event.get("ts", "")
raw_text = event.get("text", "")
raw_channel = event.get("channel", "")
raw_mentioned = bool(
raw_ts
and not raw_channel.startswith("D")
and (
event.get("type") == "app_mention"
or (
self._bot_user_id
and f"<@{self._bot_user_id}>" in raw_text
)
)
)
event_ts = f"mentioned:{raw_ts}" if raw_mentioned else raw_ts
if event_ts and self._dedup.is_duplicate(event_ts):
return

Expand Down Expand Up @@ -1915,7 +2056,34 @@ async def _handle_slack_message(self, event: dict) -> None:
user_id=user_id,
)
)
if not reply_to_bot_thread and not in_mentioned_thread and not has_session:
parent_mentions_bot = False
if (
is_thread_reply
and event_thread_ts
and not reply_to_bot_thread
and not in_mentioned_thread
and not has_session
):
parent_text = await self._fetch_thread_parent_text(
channel_id=channel_id,
thread_ts=event_thread_ts,
team_id=team_id,
strip_bot_mention=False,
)
parent_mentions_bot = bool(bot_uid and f"<@{bot_uid}>" in parent_text)
if parent_mentions_bot:
self._mentioned_threads.add(event_thread_ts)
if len(self._mentioned_threads) > self._MENTIONED_THREADS_MAX:
to_remove = list(self._mentioned_threads)[:self._MENTIONED_THREADS_MAX // 2]
for t in to_remove:
self._mentioned_threads.discard(t)

if (
not reply_to_bot_thread
and not in_mentioned_thread
and not has_session
and not parent_mentions_bot
):
return

if is_mentioned:
Expand All @@ -1925,8 +2093,8 @@ async def _handle_slack_message(self, event: dict) -> None:
# Skipped in strict mode: strict_mention=true bots must be
# re-mentioned every turn, so remembering the thread would
# defeat the feature (and re-enable agent-to-agent ack loops).
if event_thread_ts and not self._slack_strict_mention():
self._mentioned_threads.add(event_thread_ts)
if thread_ts and not self._slack_strict_mention():
self._mentioned_threads.add(thread_ts)
if len(self._mentioned_threads) > self._MENTIONED_THREADS_MAX:
to_remove = list(self._mentioned_threads)[:self._MENTIONED_THREADS_MAX // 2]
for t in to_remove:
Expand Down Expand Up @@ -2638,9 +2806,13 @@ async def _fetch_thread_context(
return ""

async def _fetch_thread_parent_text(
self, channel_id: str, thread_ts: str, team_id: str = "",
self,
channel_id: str,
thread_ts: str,
team_id: str = "",
strip_bot_mention: bool = True,
) -> str:
"""Return the raw text of the thread parent message (for reply_to_text).
"""Return the text of the thread parent message.

Uses the same per-thread cache as :meth:`_fetch_thread_context` to avoid
hitting ``conversations.replies`` twice. Falls back to a cheap single-
Expand Down Expand Up @@ -2671,7 +2843,7 @@ async def _fetch_thread_parent_text(
return ""
bot_uid = self._team_bot_user_ids.get(team_id, self._bot_user_id)
text = (parent.get("text") or "").strip()
if bot_uid:
if strip_bot_mention and bot_uid:
text = text.replace(f"<@{bot_uid}>", "").strip()
return text
except Exception as exc: # pragma: no cover - defensive
Expand Down
34 changes: 27 additions & 7 deletions gateway/run.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -3132,9 +3132,10 @@ def _schedule_resume_pending_sessions(self) -> int:
``resume_pending`` already preserves the transcript AND the existing
``_is_resume_pending`` branch in ``_handle_message_with_agent``
injects a reason-aware recovery system note on the next turn. This
method closes the UX gap by synthesizing that next turn once
adapters are back online — the event text is empty so the existing
injection path owns the wording and we never double up.
method closes the UX gap by synthesizing that next turn once adapters
are back online. When the interrupted user message was captured before
shutdown, replay it; otherwise send a nonblank continuation instruction
so the model does not treat recovery as a user-sent empty message.

Adapters that are not yet ready (adapter missing from
``self.adapters``) are skipped silently; their sessions stay
Expand Down Expand Up @@ -3173,11 +3174,18 @@ def _schedule_resume_pending_sessions(self) -> int:
)
continue

# Empty-text internal event — the _is_resume_pending branch in
# _handle_message_with_agent prepends the proper reason-aware
# system note before the turn runs.
resume_text = (getattr(entry, "in_flight_user_message", None) or "").strip()
if not resume_text:
resume_text = (
"[Internal gateway auto-resume: continue the interrupted "
"turn from the existing conversation history. This is not "
"a user-sent blank message. Do not tell the user their "
"message came through empty or ask them to repeat the same "
"command.]"
)

event = MessageEvent(
text="",
text=resume_text,
message_type=MessageType.TEXT,
source=source,
internal=True,
Expand Down Expand Up @@ -7102,6 +7110,11 @@ async def _handle_message_with_agent(self, event, source, _quick_key: str, run_g
)
if message_text is None:
return
if session_key:
try:
self.session_store.mark_in_flight(session_key, message_text)
except Exception as _e:
logger.debug("mark_in_flight failed for %s: %s", session_key, _e)

# Bind this gateway run generation to the adapter's active-session
# event so deferred post-delivery callbacks can be released by the
Expand Down Expand Up @@ -7199,6 +7212,13 @@ async def _handle_message_with_agent(self, event, source, _quick_key: str, run_g
"clear_resume_pending failed for %s: %s",
session_key, _e,
)
try:
self.session_store.clear_in_flight(session_key)
except Exception as _e:
logger.debug(
"clear_in_flight failed for %s: %s",
session_key, _e,
)

# Normalize empty responses: surface errors, partial failures, and
# the case where agent did work but returned no text. Fix for #18765.
Expand Down
Loading