Skip to content

fix(mattermost): harden thread routing and delivery hygiene #41640

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

Closed
WolframRavenwolf wants to merge 1 commit into from
Closed

fix(mattermost): harden thread routing and delivery hygiene #41640

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
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
49 changes: 25 additions & 24 deletions gateway/platforms/base.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -77,6 +77,13 @@ def _thread_metadata_for_source(source, reply_to_message_id: str | None = None)
return metadata


def _mark_notify_metadata(metadata: dict | None) -> dict:
"""Clone metadata and mark a user-visible reply as notify-worthy."""
notify_metadata = dict(metadata) if metadata else {}
notify_metadata["notify"] = True
return notify_metadata


def _reply_anchor_for_event(event) -> str | None:
"""Return reply_to id for platforms that need reply semantics.

Expand Down Expand Up @@ -3889,7 +3896,7 @@ async def _dispatch_active_session_command(
chat_id=event.source.chat_id,
content=_text,
reply_to=_reply_anchor_for_event(event),
metadata=thread_meta,
metadata=_mark_notify_metadata(thread_meta),
)
if _eph_ttl > 0 and _r.success and _r.message_id:
self._schedule_ephemeral_delete(
Expand Down Expand Up @@ -3995,7 +4002,7 @@ async def handle_message(self, event: MessageEvent) -> None:
chat_id=event.source.chat_id,
content=_text,
reply_to=_reply_anchor_for_event(event),
metadata=_thread_meta,
metadata=_mark_notify_metadata(_thread_meta),
)
if _eph_ttl > 0 and _r.success and _r.message_id:
self._schedule_ephemeral_delete(
Expand Down Expand Up @@ -4045,7 +4052,7 @@ async def handle_message(self, event: MessageEvent) -> None:
chat_id=event.source.chat_id,
content=_text,
reply_to=_reply_anchor_for_event(event),
metadata=_thread_meta,
metadata=_mark_notify_metadata(_thread_meta),
)
if _eph_ttl > 0 and _r.success and _r.message_id:
self._schedule_ephemeral_delete(
Expand Down Expand Up @@ -4268,6 +4275,12 @@ async def _stop_typing_task() -> None:
)
text_content = _recovered

# Final user-visible content (text, TTS, media, files) gets
# the existing notify=True marker. Clone once so typing/status
# metadata stays unmarked and progress bubbles remain
# thread-strict.
_final_thread_metadata = _mark_notify_metadata(_thread_metadata)

# Auto-TTS: if voice message, generate audio FIRST (before sending text)
# Gated via ``_should_auto_tts_for_chat``: fires when the chat has
# an explicit ``/voice on|tts`` opt-in OR when ``voice.auto_tts`` is
Expand Down Expand Up @@ -4307,7 +4320,7 @@ async def _stop_typing_task() -> None:
chat_id=event.source.chat_id,
audio_path=_tts_path,
caption=telegram_tts_caption,
metadata=_thread_metadata,
metadata=_final_thread_metadata,
)
_tts_caption_delivered = bool(
telegram_tts_caption and getattr(tts_result, "success", False)
Expand All @@ -4322,23 +4335,11 @@ async def _stop_typing_task() -> None:
if text_content and not _tts_caption_delivered:
logger.info("[%s] Sending response (%d chars) to %s", self.name, len(text_content), event.source.chat_id)
_reply_anchor = _reply_anchor_for_event(event)
# Mark final response messages for notification delivery.
# Platform adapters that support per-message notification
# control (e.g. Telegram's disable_notification) use this
# flag to override silent-mode and ensure the final
# response triggers a push notification.
# Clone to avoid mutating the metadata shared with the
# typing-indicator task (which must remain unmarked).
if _thread_metadata is not None:
_thread_metadata = dict(_thread_metadata)
_thread_metadata["notify"] = True
else:
_thread_metadata = {"notify": True}
result = await self._send_with_retry(
chat_id=event.source.chat_id,
content=text_content,
reply_to=_reply_anchor,
metadata=_thread_metadata,
metadata=_final_thread_metadata,
)
_record_delivery(result)

Expand Down Expand Up @@ -4367,7 +4368,7 @@ async def _stop_typing_task() -> None:
await self.send_multiple_images(
chat_id=event.source.chat_id,
images=images,
metadata=_thread_metadata,
metadata=_final_thread_metadata,
human_delay=human_delay,
)
except Exception as batch_err:
Expand Down Expand Up @@ -4409,7 +4410,7 @@ async def _stop_typing_task() -> None:
await self.send_multiple_images(
chat_id=event.source.chat_id,
images=_batch,
metadata=_thread_metadata,
metadata=_final_thread_metadata,
human_delay=human_delay,
)
except Exception as batch_err:
Expand All @@ -4424,19 +4425,19 @@ async def _stop_typing_task() -> None:
media_result = await self.send_voice(
chat_id=event.source.chat_id,
audio_path=media_path,
metadata=_thread_metadata,
metadata=_final_thread_metadata,
)
elif ext in _VIDEO_EXTS:
media_result = await self.send_video(
chat_id=event.source.chat_id,
video_path=media_path,
metadata=_thread_metadata,
metadata=_final_thread_metadata,
)
else:
media_result = await self.send_document(
chat_id=event.source.chat_id,
file_path=media_path,
metadata=_thread_metadata,
metadata=_final_thread_metadata,
)

if not media_result.success:
Expand All @@ -4454,13 +4455,13 @@ async def _stop_typing_task() -> None:
await self.send_video(
chat_id=event.source.chat_id,
video_path=file_path,
metadata=_thread_metadata,
metadata=_final_thread_metadata,
)
else:
await self.send_document(
chat_id=event.source.chat_id,
file_path=file_path,
metadata=_thread_metadata,
metadata=_final_thread_metadata,
)
except Exception as file_err:
logger.error("[%s] Error sending local file %s: %s", self.name, file_path, file_err)
Expand Down
122 changes: 108 additions & 14 deletions gateway/run.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -413,6 +413,57 @@ def _resolve_progress_thread_id(platform: Any, source_thread_id: Any, event_mess
return None


def _has_platform_display_override(user_config: dict, platform_key: str, setting: str) -> bool:
"""Return True when display.platforms.<platform> explicitly sets setting."""
display = user_config.get("display") if isinstance(user_config, dict) else None
if not isinstance(display, dict):
return False
platforms = display.get("platforms")
if not isinstance(platforms, dict):
return False
platform_cfg = platforms.get(platform_key)
return isinstance(platform_cfg, dict) and setting in platform_cfg


def _resolve_gateway_display_bool(
user_config: dict,
platform_key: str,
setting: str,
*,
default: bool = False,
platform: Any = None,
require_platform_override_for: set[Any] | None = None,
) -> bool:
"""Resolve a boolean display setting with optional platform-only opt-in.

Some display features expose assistant scratch text rather than deliberate
user-facing output. For high-noise threaded chat surfaces such as
Mattermost, a global opt-in is too broad: they must be enabled with an
explicit display.platforms.<platform>.<setting> override.
"""
current_platform = _gateway_platform_value(platform or platform_key)
platform_only = {
_gateway_platform_value(candidate)
for candidate in (require_platform_override_for or set())
}
if (
current_platform in platform_only
and not _has_platform_display_override(user_config, platform_key, setting)
):
return False

from gateway.display_config import resolve_display_setting

value = resolve_display_setting(user_config, platform_key, setting, default)
if isinstance(value, bool):
return value
if isinstance(value, str):
return value.strip().lower() in {"true", "yes", "1", "on"}
if value is None:
return bool(default)
return bool(value)


def _telegramize_command_mentions(text: str, platform: Any) -> str:
"""Rewrite slash-command mentions to Telegram-valid command names.

Expand Down Expand Up @@ -8989,17 +9040,24 @@ async def _handle_message_with_agent(self, event, source, _quick_key: str, run_g
source, session_entry, reason="agent-result-compression",
)

# Prepend reasoning/thinking if display is enabled (per-platform)
# Prepend reasoning/thinking if display is enabled (per-platform).
# Mattermost requires explicit per-platform opt-in because this is
# scratch text, not ordinary final-answer content.
try:
from gateway.display_config import resolve_display_setting as _rds
_show_reasoning_effective = _rds(
_show_reasoning_effective = _resolve_gateway_display_bool(
_load_gateway_config(),
_platform_config_key(source.platform),
"show_reasoning",
getattr(self, "_show_reasoning", False),
default=bool(getattr(self, "_show_reasoning", False)),
platform=source.platform,
require_platform_override_for={Platform.MATTERMOST},
)
except Exception:
_show_reasoning_effective = getattr(self, "_show_reasoning", False)
_show_reasoning_effective = (
False
if source.platform == Platform.MATTERMOST
else getattr(self, "_show_reasoning", False)
)
if _show_reasoning_effective and response and not _intentional_silence:
last_reasoning = agent_result.get("last_reasoning")
if last_reasoning:
Expand Down Expand Up @@ -13633,18 +13691,32 @@ def _run_still_current() -> bool:
# in chat platforms while opting into concise mid-turn updates.
interim_assistant_messages_enabled = (
source.platform != Platform.WEBHOOK
and bool(
resolve_display_setting(
user_config,
platform_key,
"interim_assistant_messages",
True,
)
and _resolve_gateway_display_bool(
user_config,
platform_key,
"interim_assistant_messages",
default=True,
platform=source.platform,
require_platform_override_for={Platform.MATTERMOST},
)
)

# thinking_progress is independent — if enabled, we need the progress
# queue even when tool_progress is off (thinking relay uses same infra).
# Mattermost requires a per-platform opt-in: global scratch-text display
# is too easy to leak into busy public threads.
_thinking_enabled = _resolve_gateway_display_bool(
user_config,
platform_key,
"thinking_progress",
default=False,
platform=source.platform,
require_platform_override_for={Platform.MATTERMOST},
)
needs_progress_queue = tool_progress_enabled or _thinking_enabled


# Queue for progress messages (thread-safe)
progress_queue = queue.Queue() if tool_progress_enabled else None
progress_queue = queue.Queue() if needs_progress_queue else None
last_tool = [None] # Mutable container for tracking in closure
last_progress_msg = [None] # Track last message for dedup
repeat_count = [0] # How many times the same message repeated
Expand Down Expand Up @@ -13750,6 +13822,24 @@ def progress_callback(event_type: str, tool_name: str = None, preview: str = Non
logger.debug("tool-progress onboarding hint failed: %s", _hint_err)
return

# "_thinking" is assistant scratch text between tool calls. It
# is never ordinary tool progress: only relay it when the platform
# explicitly opted into thinking_progress. Handle both legacy
# callback shapes: ("_thinking", text) and
# ("reasoning.available", "_thinking", text, ...).
if event_type == "_thinking" or tool_name == "_thinking":
if not _thinking_enabled:
return
thinking_text = preview if tool_name == "_thinking" else tool_name
msg = f"💬 {thinking_text}" if thinking_text else None
if msg:
progress_queue.put(msg)
return

# If tool_progress is off, only _thinking passes through (above).
# Regular tool calls are suppressed.
if not tool_progress_enabled:
return

# Only act on tool.started events (ignore tool.completed, reasoning.available, etc.)
if event_type not in {"tool.started",}:
Expand Down Expand Up @@ -14773,6 +14863,10 @@ def _clarify_callback_sync(question: str, choices) -> str:

agent.clarify_callback = _clarify_callback_sync

# Show assistant thinking between tool calls — independent of
# tool_progress mode. Mattermost needs an explicit per-platform
# opt-in so global scratch-text display does not leak into threads.
agent.thinking_progress = _thinking_enabled
# Store agent reference for interrupt support
agent_holder[0] = agent
# Capture the full tool definitions for transcript logging
Expand Down
Loading
Loading