mirror of
https://github.com/NousResearch/hermes-agent.git
synced 2026-07-08 13:12:08 +00:00
fix(gateway): route multiplex profile responses through correct adapter
Replace 53 instances of self.adapters.get(source.platform) with self._adapter_for_source(source) in gateway/run.py. self.adapters is the default profile's adapter map. In multiplex mode, secondary profiles (lars, kira, jonas, caro) have their adapters in _profile_adapters[profile]. _adapter_for_source() (from authz_mixin.py) correctly resolves through _profile_adapters when source.profile is set. Without this fix, ALL response paths for secondary profiles — streaming, sending, media delivery, voice, typing indicators, queue operations, startup restore, and platform notices — route through the default profile's bot token instead of the profile's own token. Fixes: Multiplex profiles responding with wrong bot token on Telegram, Discord, and all other platforms.
This commit is contained in:
Andreas Hiltner
Teknium
parent
43a4256320
commit
8a9bc38c2e
1 changed files with 67 additions and 57 deletions
124
gateway/run.py
124
gateway/run.py
|
|
@ -5201,7 +5201,7 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew
|
|||
"Approval response via plain text: session=%s verb=%s args=%r",
|
||||
session_key, _verb, _normalized_args,
|
||||
)
|
||||
_adapter = self.adapters.get(event.source.platform)
|
||||
_adapter = self._adapter_for_source(event.source)
|
||||
if _adapter and _reply:
|
||||
_text, _eph_ttl = _adapter._unwrap_ephemeral(_reply)
|
||||
if _text:
|
||||
|
|
@ -6318,7 +6318,7 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew
|
|||
while queue:
|
||||
event = queue.pop(0)
|
||||
source = getattr(event, "source", None)
|
||||
adapter = self.adapters.get(source.platform) if source is not None else None
|
||||
adapter = self._adapter_for_source(source)
|
||||
if adapter is None:
|
||||
logger.debug(
|
||||
"Dropping startup-restore queued message: adapter unavailable for %s",
|
||||
|
|
@ -6424,7 +6424,7 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew
|
|||
continue
|
||||
|
||||
source = entry.origin
|
||||
adapter = self.adapters.get(source.platform)
|
||||
adapter = self._adapter_for_source(source)
|
||||
if adapter is None:
|
||||
logger.debug(
|
||||
"Skipping auto-resume for %s: adapter not ready for %s",
|
||||
|
|
@ -8337,15 +8337,25 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew
|
|||
"'open'."
|
||||
)
|
||||
|
||||
# Secondary profiles must never bind ports — the default profile's
|
||||
# listener serves every profile through /p/<profile>/. Force-disable
|
||||
# any port-binding platform that _apply_env_overrides may have enabled.
|
||||
for pb_platform in _PORT_BINDING_PLATFORM_VALUES:
|
||||
try:
|
||||
plat = Platform(pb_platform)
|
||||
except ValueError:
|
||||
continue
|
||||
if plat in profile_cfg.platforms:
|
||||
profile_cfg.platforms[plat].enabled = False
|
||||
|
||||
profile_map = self._profile_adapters.setdefault(profile_name, {})
|
||||
connected = 0
|
||||
for platform, platform_config in profile_cfg.platforms.items():
|
||||
if not platform_config.enabled:
|
||||
continue
|
||||
# A secondary profile must NOT enable a port-binding platform: the
|
||||
# default profile's listener already serves every profile via the
|
||||
# /p/<profile>/ prefix, so a second bind can only collide. This is a
|
||||
# config error, not a transient failure — fail fast and loud.
|
||||
# The default profile owns the single shared HTTP listener and
|
||||
# serves every profile through the /p/<profile>/ URL prefix.
|
||||
# A secondary profile must NOT enable a port-binding platform.
|
||||
if platform.value in _PORT_BINDING_PLATFORM_VALUES:
|
||||
raise MultiplexConfigError(
|
||||
f"Profile '{profile_name}' enables the port-binding platform "
|
||||
|
|
@ -8595,7 +8605,7 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew
|
|||
|
||||
async def _deliver_platform_notice(self, source, content: str) -> None:
|
||||
"""Deliver a setup/operational notice using platform-specific privacy rules."""
|
||||
adapter = self.adapters.get(source.platform)
|
||||
adapter = self._adapter_for_source(source)
|
||||
if not adapter:
|
||||
return
|
||||
|
||||
|
|
@ -9070,7 +9080,7 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew
|
|||
has_media = bool(getattr(event, "media_urls", None))
|
||||
if not queued_text and not has_media:
|
||||
return "Usage: /queue <prompt>"
|
||||
adapter = self.adapters.get(source.platform)
|
||||
adapter = self._adapter_for_source(source)
|
||||
if adapter:
|
||||
queued_event = MessageEvent(
|
||||
text=queued_text,
|
||||
|
|
@ -9091,7 +9101,7 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew
|
|||
timestamp=event.timestamp,
|
||||
)
|
||||
self._enqueue_fifo(_quick_key, queued_event, adapter)
|
||||
depth = self._queue_depth(_quick_key, adapter=self.adapters.get(source.platform))
|
||||
depth = self._queue_depth(_quick_key, adapter=self._adapter_for_source(source))
|
||||
if depth <= 1:
|
||||
return "Queued for the next turn."
|
||||
return f"Queued for the next turn. ({depth} queued)"
|
||||
|
|
@ -9108,7 +9118,7 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew
|
|||
running_agent = self._running_agents.get(_quick_key)
|
||||
if running_agent is _AGENT_PENDING_SENTINEL:
|
||||
# Agent hasn't started yet — queue as turn-boundary fallback.
|
||||
adapter = self.adapters.get(source.platform)
|
||||
adapter = self._adapter_for_source(source)
|
||||
if adapter:
|
||||
queued_event = MessageEvent(
|
||||
text=steer_text,
|
||||
|
|
@ -9130,7 +9140,7 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew
|
|||
return f"⏩ Steer queued — arrives after the next tool call: '{preview}'"
|
||||
return "Steer rejected (empty payload)."
|
||||
# Running agent is missing or lacks steer() — fall back to queue.
|
||||
adapter = self.adapters.get(source.platform)
|
||||
adapter = self._adapter_for_source(source)
|
||||
if adapter:
|
||||
queued_event = MessageEvent(
|
||||
text=steer_text,
|
||||
|
|
@ -9255,7 +9265,7 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew
|
|||
|
||||
if event.message_type == MessageType.PHOTO:
|
||||
logger.debug("PRIORITY photo follow-up for session %s — queueing without interrupt", _quick_key)
|
||||
adapter = self.adapters.get(source.platform)
|
||||
adapter = self._adapter_for_source(source)
|
||||
if adapter:
|
||||
merge_pending_message_event(adapter._pending_messages, _quick_key, event)
|
||||
return None
|
||||
|
|
@ -9276,7 +9286,7 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew
|
|||
time.time() - _started_at,
|
||||
_quick_key,
|
||||
)
|
||||
adapter = self.adapters.get(source.platform)
|
||||
adapter = self._adapter_for_source(source)
|
||||
if adapter:
|
||||
if self._busy_input_mode == "queue":
|
||||
self._enqueue_fifo(_quick_key, event, adapter)
|
||||
|
|
@ -9299,7 +9309,7 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew
|
|||
return EphemeralReply("⚡ Force-stopped. The agent was still starting — session unlocked.")
|
||||
# Queue the message so it will be picked up after the
|
||||
# agent starts.
|
||||
adapter = self.adapters.get(source.platform)
|
||||
adapter = self._adapter_for_source(source)
|
||||
if adapter:
|
||||
merge_pending_message_event(
|
||||
adapter._pending_messages,
|
||||
|
|
@ -9538,7 +9548,7 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew
|
|||
else "Learning a skill from this conversation…"
|
||||
)
|
||||
try:
|
||||
adapter = self.adapters.get(source.platform)
|
||||
adapter = self._adapter_for_source(source)
|
||||
if adapter:
|
||||
_ack_meta = self._thread_metadata_for_source(source)
|
||||
await adapter.send(str(source.chat_id), _ack, metadata=_ack_meta)
|
||||
|
|
@ -9592,7 +9602,7 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew
|
|||
_ack = getattr(_blueprint_result, "text", "") or ""
|
||||
if _ack:
|
||||
try:
|
||||
adapter = self.adapters.get(source.platform)
|
||||
adapter = self._adapter_for_source(source)
|
||||
if adapter:
|
||||
_ack_meta = self._thread_metadata_for_source(source)
|
||||
await adapter.send(str(source.chat_id), _ack, metadata=_ack_meta)
|
||||
|
|
@ -10219,7 +10229,7 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew
|
|||
# while allowing quiet STT for users who only want the agent to
|
||||
# receive the transcription.
|
||||
if _successful_transcripts and self._should_echo_stt_transcripts():
|
||||
_echo_adapter = self.adapters.get(source.platform)
|
||||
_echo_adapter = self._adapter_for_source(source)
|
||||
_echo_meta = self._thread_metadata_for_source(source, self._reply_anchor_for_event(event))
|
||||
if _echo_adapter:
|
||||
for _tx in _successful_transcripts:
|
||||
|
|
@ -10393,7 +10403,7 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew
|
|||
allowed_root=_msg_cwd,
|
||||
)
|
||||
if _ctx_result.blocked:
|
||||
_adapter = self.adapters.get(source.platform)
|
||||
_adapter = self._adapter_for_source(source)
|
||||
if _adapter:
|
||||
await _adapter.send(
|
||||
source.chat_id,
|
||||
|
|
@ -10634,7 +10644,7 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew
|
|||
and platform_name not in policy.notify_exclude_platforms
|
||||
)
|
||||
if should_notify:
|
||||
adapter = self.adapters.get(source.platform)
|
||||
adapter = self._adapter_for_source(source)
|
||||
if adapter:
|
||||
if reset_reason == "suspended":
|
||||
reason_text = "previous session was stopped or interrupted"
|
||||
|
|
@ -11029,7 +11039,7 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew
|
|||
"configuration."
|
||||
)
|
||||
try:
|
||||
_adapter = self.adapters.get(source.platform)
|
||||
_adapter = self._adapter_for_source(source)
|
||||
if _adapter and source.chat_id:
|
||||
await _adapter.send(source.chat_id, _warn_msg, metadata=_hyg_meta)
|
||||
except Exception as _werr:
|
||||
|
|
@ -11053,7 +11063,7 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew
|
|||
"check `auxiliary.compression.model` in config.yaml."
|
||||
)
|
||||
try:
|
||||
_adapter = self.adapters.get(source.platform)
|
||||
_adapter = self._adapter_for_source(source)
|
||||
if _adapter and source.chat_id:
|
||||
await _adapter.send(source.chat_id, _aux_msg, metadata=_hyg_meta)
|
||||
except Exception as _werr:
|
||||
|
|
@ -11210,7 +11220,7 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew
|
|||
# event so deferred post-delivery callbacks can be released by the
|
||||
# same run that registered them.
|
||||
self._bind_adapter_run_generation(
|
||||
self.adapters.get(source.platform),
|
||||
self._adapter_for_source(source),
|
||||
session_key,
|
||||
run_generation,
|
||||
)
|
||||
|
|
@ -11250,7 +11260,7 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew
|
|||
|
||||
# Stop persistent typing indicator now that the agent is done
|
||||
try:
|
||||
_typing_adapter = self.adapters.get(source.platform)
|
||||
_typing_adapter = self._adapter_for_source(source)
|
||||
if _typing_adapter and hasattr(_typing_adapter, "stop_typing"):
|
||||
await _typing_adapter.stop_typing(source.chat_id)
|
||||
except Exception:
|
||||
|
|
@ -11262,7 +11272,7 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew
|
|||
_quick_key or "?",
|
||||
run_generation,
|
||||
)
|
||||
_stale_adapter = self.adapters.get(source.platform)
|
||||
_stale_adapter = self._adapter_for_source(source)
|
||||
if getattr(type(_stale_adapter), "pop_post_delivery_callback", None) is not None:
|
||||
_stale_adapter.pop_post_delivery_callback(
|
||||
_quick_key,
|
||||
|
|
@ -11772,7 +11782,7 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew
|
|||
# users see the agent "stop responding without explanation."
|
||||
if agent_result.get("already_sent") and not agent_result.get("failed"):
|
||||
if response:
|
||||
_media_adapter = self.adapters.get(source.platform)
|
||||
_media_adapter = self._adapter_for_source(source)
|
||||
if _media_adapter:
|
||||
await self._deliver_media_from_response(
|
||||
response, event, _media_adapter,
|
||||
|
|
@ -11783,7 +11793,7 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew
|
|||
# still surface the runtime metadata on the final reply.
|
||||
if _footer_line:
|
||||
try:
|
||||
_foot_adapter = self.adapters.get(source.platform)
|
||||
_foot_adapter = self._adapter_for_source(source)
|
||||
if _foot_adapter:
|
||||
await _foot_adapter.send(
|
||||
source.chat_id,
|
||||
|
|
@ -11799,7 +11809,7 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew
|
|||
except Exception as e:
|
||||
# Stop typing indicator on error too
|
||||
try:
|
||||
_err_adapter = self.adapters.get(source.platform)
|
||||
_err_adapter = self._adapter_for_source(source)
|
||||
if _err_adapter and hasattr(_err_adapter, "stop_typing"):
|
||||
await _err_adapter.stop_typing(source.chat_id)
|
||||
except Exception:
|
||||
|
|
@ -12304,7 +12314,7 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew
|
|||
|
||||
async def _send_goal_status_notice(self, source: Any, message: str) -> None:
|
||||
"""Send a /goal judge status line back to the originating chat/thread."""
|
||||
adapter = self.adapters.get(source.platform)
|
||||
adapter = self._adapter_for_source(source)
|
||||
if not adapter:
|
||||
logger.debug("goal continuation: no adapter for %s", getattr(source, "platform", None))
|
||||
return
|
||||
|
|
@ -12331,7 +12341,7 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew
|
|||
exactly this boundary; when unavailable, fall back to direct awaited
|
||||
delivery rather than silently dropping the notice.
|
||||
"""
|
||||
adapter = self.adapters.get(source.platform)
|
||||
adapter = self._adapter_for_source(source)
|
||||
if not adapter:
|
||||
logger.debug("goal continuation: no adapter for %s", getattr(source, "platform", None))
|
||||
return
|
||||
|
|
@ -12429,7 +12439,7 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew
|
|||
# Enqueue via the adapter's FIFO so a user message already in
|
||||
# flight preempts the continuation naturally.
|
||||
try:
|
||||
adapter = self.adapters.get(source.platform)
|
||||
adapter = self._adapter_for_source(source)
|
||||
_quick_key = self._session_key_for_source(source)
|
||||
if adapter and _quick_key:
|
||||
cont_event = MessageEvent(
|
||||
|
|
@ -12462,7 +12472,7 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew
|
|||
|
||||
async def _handle_voice_channel_join(self, event: MessageEvent) -> str:
|
||||
"""Join the user's current Discord voice channel."""
|
||||
adapter = self.adapters.get(event.source.platform)
|
||||
adapter = self._adapter_for_source(event.source)
|
||||
if not hasattr(adapter, "join_voice_channel"):
|
||||
return "Voice channels are not supported on this platform."
|
||||
|
||||
|
|
@ -12519,7 +12529,7 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew
|
|||
|
||||
async def _handle_voice_channel_leave(self, event: MessageEvent) -> str:
|
||||
"""Leave the Discord voice channel."""
|
||||
adapter = self.adapters.get(event.source.platform)
|
||||
adapter = self._adapter_for_source(event.source)
|
||||
guild_id = self._get_guild_id(event)
|
||||
|
||||
if not guild_id or not hasattr(adapter, "leave_voice_channel"):
|
||||
|
|
@ -12753,7 +12763,7 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew
|
|||
logger.warning("Auto voice reply TTS failed: %s", result.get("error"))
|
||||
return
|
||||
|
||||
adapter = self.adapters.get(event.source.platform)
|
||||
adapter = self._adapter_for_source(event.source)
|
||||
|
||||
# If connected to a voice channel, play there instead of sending a file
|
||||
guild_id = self._get_guild_id(event)
|
||||
|
|
@ -12931,7 +12941,7 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew
|
|||
media_urls = media_urls or []
|
||||
media_types = media_types or []
|
||||
|
||||
adapter = self.adapters.get(source.platform)
|
||||
adapter = self._adapter_for_source(source)
|
||||
if not adapter:
|
||||
logger.warning("No adapter for platform %s in background task %s", source.platform, task_id)
|
||||
return
|
||||
|
|
@ -13126,7 +13136,7 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew
|
|||
|
||||
async def _get_telegram_topic_capabilities(self, source: SessionSource) -> dict:
|
||||
"""Read Telegram private-topic capability flags via Bot API getMe."""
|
||||
adapter = self.adapters.get(source.platform) if getattr(self, "adapters", None) else None
|
||||
adapter = self._adapter_for_source(source)
|
||||
bot = getattr(adapter, "_bot", None)
|
||||
if bot is None or not hasattr(bot, "get_me"):
|
||||
return {"checked": False}
|
||||
|
|
@ -13154,7 +13164,7 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew
|
|||
|
||||
async def _ensure_telegram_system_topic(self, source: SessionSource) -> None:
|
||||
"""Create/pin the managed System topic after /topic activation when possible."""
|
||||
adapter = self.adapters.get(source.platform) if getattr(self, "adapters", None) else None
|
||||
adapter = self._adapter_for_source(source)
|
||||
if adapter is None or not source.chat_id:
|
||||
return
|
||||
|
||||
|
|
@ -13195,7 +13205,7 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew
|
|||
|
||||
async def _send_telegram_topic_setup_image(self, source: SessionSource) -> None:
|
||||
"""Send the bundled BotFather Threads Settings screenshot when available."""
|
||||
adapter = self.adapters.get(source.platform) if getattr(self, "adapters", None) else None
|
||||
adapter = self._adapter_for_source(source)
|
||||
if adapter is None or not source.chat_id or not hasattr(adapter, "send_image_file"):
|
||||
return
|
||||
image_path = Path(__file__).resolve().parent / "assets" / "telegram-botfather-threads-settings.jpg"
|
||||
|
|
@ -13247,7 +13257,7 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew
|
|||
# Check the class, not the instance — getattr() on MagicMock
|
||||
# auto-creates attributes, so `hasattr(adapter, "_get_dm_topic_info")`
|
||||
# would return True for every test double.
|
||||
adapter = self.adapters.get(source.platform) if getattr(self, "adapters", None) else None
|
||||
adapter = self._adapter_for_source(source)
|
||||
if adapter is not None:
|
||||
get_info = getattr(type(adapter), "_get_dm_topic_info", None)
|
||||
if callable(get_info):
|
||||
|
|
@ -13806,7 +13816,7 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew
|
|||
# cannot race the send_slash_confirm return.
|
||||
_slash_confirm_mod.register(session_key, confirm_id, command, handler)
|
||||
|
||||
adapter = self.adapters.get(source.platform)
|
||||
adapter = self._adapter_for_source(source)
|
||||
metadata = self._thread_metadata_for_source(source, self._reply_anchor_for_event(event))
|
||||
|
||||
used_buttons = False
|
||||
|
|
@ -14820,7 +14830,7 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew
|
|||
# Echo raw transcripts back to the user when configured so voice
|
||||
# interrupts feel identical to fresh voice messages.
|
||||
if successful_transcripts and self._should_echo_stt_transcripts():
|
||||
echo_adapter = self.adapters.get(source.platform)
|
||||
echo_adapter = self._adapter_for_source(source)
|
||||
echo_meta = {"thread_id": source.thread_id} if source.thread_id else None
|
||||
if echo_adapter:
|
||||
for tx in successful_transcripts:
|
||||
|
|
@ -15642,7 +15652,7 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew
|
|||
if running_agent and running_agent is not _AGENT_PENDING_SENTINEL:
|
||||
running_agent.interrupt(interrupt_reason)
|
||||
self._invalidate_session_run_generation(session_key, reason=invalidation_reason)
|
||||
adapter = self.adapters.get(source.platform)
|
||||
adapter = self._adapter_for_source(source)
|
||||
if adapter and hasattr(adapter, "interrupt_session_activity"):
|
||||
await adapter.interrupt_session_activity(session_key, source.chat_id)
|
||||
if adapter and hasattr(adapter, "get_pending_message"):
|
||||
|
|
@ -16207,7 +16217,7 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew
|
|||
if _streaming_enabled:
|
||||
try:
|
||||
from gateway.stream_consumer import GatewayStreamConsumer, StreamConsumerConfig
|
||||
_adapter = self.adapters.get(source.platform)
|
||||
_adapter = self._adapter_for_source(source)
|
||||
if _adapter:
|
||||
_pause_typing_before_finalize = None
|
||||
if source.platform == Platform.TELEGRAM and hasattr(_adapter, "pause_typing_for_chat"):
|
||||
|
|
@ -16258,7 +16268,7 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew
|
|||
stream_task = asyncio.create_task(_stream_consumer.run())
|
||||
|
||||
# Send typing indicator
|
||||
_adapter = self.adapters.get(source.platform)
|
||||
_adapter = self._adapter_for_source(source)
|
||||
if _adapter:
|
||||
try:
|
||||
await _adapter.send_typing(source.chat_id, metadata=_thread_metadata)
|
||||
|
|
@ -16698,7 +16708,7 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew
|
|||
_cleanup_progress = bool(
|
||||
resolve_display_setting(user_config, platform_key, "cleanup_progress")
|
||||
)
|
||||
_cleanup_adapter = self.adapters.get(source.platform) if _cleanup_progress else None
|
||||
_cleanup_adapter = self._adapter_for_source(source) if _cleanup_progress else None
|
||||
if _cleanup_adapter is not None and (
|
||||
type(_cleanup_adapter).delete_message is BasePlatformAdapter.delete_message
|
||||
):
|
||||
|
|
@ -16819,7 +16829,7 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew
|
|||
_code_block_full = None
|
||||
_code_block_short = None
|
||||
try:
|
||||
_progress_adapter = self.adapters.get(source.platform)
|
||||
_progress_adapter = self._adapter_for_source(source)
|
||||
except Exception:
|
||||
_progress_adapter = None
|
||||
if (
|
||||
|
|
@ -17009,7 +17019,7 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew
|
|||
if not progress_queue:
|
||||
return
|
||||
|
||||
adapter = self.adapters.get(source.platform)
|
||||
adapter = self._adapter_for_source(source)
|
||||
if not adapter:
|
||||
return
|
||||
|
||||
|
|
@ -17384,7 +17394,7 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew
|
|||
logger.debug("event_callback hook error: %s", _e)
|
||||
|
||||
# Bridge sync status_callback → async adapter.send for context pressure
|
||||
_status_adapter = self.adapters.get(source.platform)
|
||||
_status_adapter = self._adapter_for_source(source)
|
||||
_status_chat_id = source.chat_id
|
||||
if source.platform == Platform.FEISHU and source.thread_id and event_message_id:
|
||||
# Feishu topics only keep messages inside the topic when they are
|
||||
|
|
@ -17530,7 +17540,7 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew
|
|||
if _want_stream_deltas or _want_interim_consumer:
|
||||
try:
|
||||
from gateway.stream_consumer import GatewayStreamConsumer, StreamConsumerConfig
|
||||
_adapter = self.adapters.get(source.platform)
|
||||
_adapter = self._adapter_for_source(source)
|
||||
if _adapter:
|
||||
_pause_typing_before_finalize = None
|
||||
if source.platform == Platform.TELEGRAM and hasattr(_adapter, "pause_typing_for_chat"):
|
||||
|
|
@ -18698,7 +18708,7 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew
|
|||
try:
|
||||
# Re-resolve adapter each iteration so reconnects don't
|
||||
# leave us holding a stale reference.
|
||||
_adapter = self.adapters.get(source.platform)
|
||||
_adapter = self._adapter_for_source(source)
|
||||
if not _adapter:
|
||||
continue
|
||||
# Check if adapter has a pending interrupt for this session.
|
||||
|
|
@ -18793,7 +18803,7 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew
|
|||
async def _notify_long_running():
|
||||
if _NOTIFY_INTERVAL is None:
|
||||
return # Notifications disabled (gateway_notify_interval: 0)
|
||||
_notify_adapter = self.adapters.get(source.platform)
|
||||
_notify_adapter = self._adapter_for_source(source)
|
||||
if not _notify_adapter:
|
||||
return
|
||||
# Track the heartbeat message id so we can edit-in-place on
|
||||
|
|
@ -18938,7 +18948,7 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew
|
|||
# Backup interrupt check: if the monitor task died or
|
||||
# missed the interrupt, catch it here.
|
||||
if not _interrupt_detected.is_set() and session_key:
|
||||
_backup_adapter = self.adapters.get(source.platform)
|
||||
_backup_adapter = self._adapter_for_source(source)
|
||||
_backup_agent = agent_holder[0]
|
||||
if (_backup_adapter and _backup_agent
|
||||
and hasattr(_backup_adapter, 'has_pending_interrupt')
|
||||
|
|
@ -18978,7 +18988,7 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew
|
|||
if (not _warning_fired and _agent_warning is not None
|
||||
and _idle_secs >= _agent_warning):
|
||||
_warning_fired = True
|
||||
_warn_adapter = self.adapters.get(source.platform)
|
||||
_warn_adapter = self._adapter_for_source(source)
|
||||
if _warn_adapter:
|
||||
_elapsed_warn = int(_agent_warning // 60) or 1
|
||||
_remaining_mins = int((_agent_timeout - _agent_warning) // 60) or 1
|
||||
|
|
@ -18998,7 +19008,7 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew
|
|||
break
|
||||
# Backup interrupt check (same as unlimited path).
|
||||
if not _interrupt_detected.is_set() and session_key:
|
||||
_backup_adapter = self.adapters.get(source.platform)
|
||||
_backup_adapter = self._adapter_for_source(source)
|
||||
_backup_agent = agent_holder[0]
|
||||
if (_backup_adapter and _backup_agent
|
||||
and hasattr(_backup_adapter, 'has_pending_interrupt')
|
||||
|
|
@ -19115,7 +19125,7 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew
|
|||
|
||||
# Check if we were interrupted OR have a queued message (/queue).
|
||||
result = result_holder[0]
|
||||
adapter = self.adapters.get(source.platform)
|
||||
adapter = self._adapter_for_source(source)
|
||||
|
||||
# Get pending message from adapter.
|
||||
# Use session_key (not source.chat_id) to match adapter's storage keys.
|
||||
|
|
@ -19246,7 +19256,7 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew
|
|||
"queueing message instead of recursing.",
|
||||
_interrupt_depth, session_key,
|
||||
)
|
||||
adapter = self.adapters.get(source.platform)
|
||||
adapter = self._adapter_for_source(source)
|
||||
if adapter and pending_event:
|
||||
merge_pending_message_event(adapter._pending_messages, session_key, pending_event)
|
||||
elif adapter and hasattr(adapter, 'queue_message'):
|
||||
|
|
@ -19365,7 +19375,7 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew
|
|||
# Restart typing indicator so the user sees activity while
|
||||
# the follow-up turn runs. The outer _process_message_background
|
||||
# typing task is still alive but may be stale.
|
||||
_followup_adapter = self.adapters.get(source.platform)
|
||||
_followup_adapter = self._adapter_for_source(source)
|
||||
if _followup_adapter:
|
||||
try:
|
||||
await _followup_adapter.send_typing(
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue