From 3cfb7dee251c6c000c00fc0f31a69958304efaff Mon Sep 17 00:00:00 2001
From: Kris <2893855659@qq.com>
Date: Wed, 13 May 2026 16:05:25 +0800
Subject: [PATCH] =?UTF-8?q?refactor(bluebubbles):=20=E9=87=8D=E6=9E=84?=
=?UTF-8?q?=E5=B9=B6=E6=96=B0=E5=A2=9E=E5=A4=9A=E9=A1=B9=E8=93=9D=E6=B3=A1?=
=?UTF-8?q?=E6=B3=A1=E9=80=82=E9=85=8D=E5=99=A8=E5=8A=9F=E8=83=BD?=
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit
本次提交对蓝泡泡适配器进行了多维度改进：
1. 新增缓存超限自动清理逻辑，防止内存溢出
2. 增强流式打字同步功能，增加超时自动停止逻辑
3. 重构markdown渲染逻辑，修复格式匹配问题并支持更多语法
4. 新增配置校验工具函数，增加streaming_mode和auth_strategy的合法性校验
5. 新增多项环境配置参数，完善配置项
6. 重构初始化流程，新增多个工具类实例
7. 优化TTS禁用逻辑，新增aiohttp依赖检查
8. 重构流式消息发送逻辑，整合打字指示器逻辑
9. 新增健康状态检查的问题检测功能
---
.../channels/adapters/bluebubbles/adapter.py | 126 +++++++++++-------
.../channels/adapters/bluebubbles/config.py | 65 ++++++++-
.../adapters/bluebubbles/monitor_debounce.py | 4 +
.../channels/adapters/bluebubbles/renderer.py | 60 +++++++--
.../bluebubbles/streaming_enhanced.py | 19 ++-
5 files changed, 217 insertions(+), 57 deletions(-)
diff --git a/backend/package/yuxi/channels/adapters/bluebubbles/adapter.py b/backend/package/yuxi/channels/adapters/bluebubbles/adapter.py
index 7cda2d49..c8e70def 100644
--- a/backend/package/yuxi/channels/adapters/bluebubbles/adapter.py
+++ b/backend/package/yuxi/channels/adapters/bluebubbles/adapter.py
@@ -9,11 +9,15 @@ from typing import Any
from yuxi.channels.adapters.bluebubbles.accounts import parse_accounts_config
from yuxi.channels.adapters.bluebubbles.action_gate import ActionGate
-from yuxi.channels.adapters.bluebubbles.catchup import CatchupCursorStore, CatchupSummary, run_full_catchup
+from yuxi.channels.adapters.bluebubbles.catchup import CatchupCursorStore, run_full_catchup
from yuxi.channels.adapters.bluebubbles.client import BlueBubblesClient
from yuxi.channels.adapters.bluebubbles.contact_resolver import resolve_contact
+from yuxi.channels.adapters.bluebubbles.conversation_bindings import ConversationBindings
+from yuxi.channels.adapters.bluebubbles.conversation_id import ConversationID
+from yuxi.channels.adapters.bluebubbles.conversation_route import ConversationRoute
from yuxi.channels.adapters.bluebubbles.directory import list_chats, list_contacts, search_chats, search_contacts
from yuxi.channels.adapters.bluebubbles.exceptions import BlueBubblesConnectionError, BlueBubblesHTTPError
+from yuxi.channels.adapters.bluebubbles.exec_approval import ExecApproval
from yuxi.channels.adapters.bluebubbles.group_actions import (
add_participant,
leave_chat,
@@ -22,9 +26,9 @@ from yuxi.channels.adapters.bluebubbles.group_actions import (
set_group_icon,
)
from yuxi.channels.adapters.bluebubbles.inbound_dedupe import InboundDedupe
+from yuxi.channels.adapters.bluebubbles.message_effects import send_with_effect
from yuxi.channels.adapters.bluebubbles.monitor import BlueBubblesMonitor, InboundReorderBuffer
from yuxi.channels.adapters.bluebubbles.monitor_debounce import MonitorDebounce
-from yuxi.channels.adapters.bluebubbles.message_effects import send_with_effect
from yuxi.channels.adapters.bluebubbles.probe import (
is_macos_26_or_higher,
probe_imessage_login,
@@ -49,6 +53,9 @@ from yuxi.channels.adapters.bluebubbles.send import (
)
from yuxi.channels.adapters.bluebubbles.sent_cache import SentMessageCache
from yuxi.channels.adapters.bluebubbles.session import resolve_chat_type
+from yuxi.channels.adapters.bluebubbles.session_route import SessionRoute
+from yuxi.channels.adapters.bluebubbles.status_issues import StatusIssues
+from yuxi.channels.adapters.bluebubbles.streaming_enhanced import StreamingTypingSync
from yuxi.channels.adapters.bluebubbles.targets import resolve_chat_guid_with_auto_create
from yuxi.channels.adapters.bluebubbles.voice import check_tts_available
from yuxi.channels.adapters.bluebubbles.webhook_ingress import WebhookIngress
@@ -132,6 +139,13 @@ class BlueBubblesAdapter(BaseChannelAdapter):
group_actions = {"rename_group", "set_group_icon", "add_participant", "remove_participant", "leave_group"}
if not group_actions.issubset(set(available)):
base.group_management = False
+
+ tts_available = check_tts_available()
+ if not tts_available["espeak"]:
+ base.tts.voice.enabled = False
+ elif not tts_available["ffmpeg"] and not tts_available["pydub"]:
+ base.tts.voice.enabled = False
+
return base
meta = ChannelMeta(id="bluebubbles", label="BlueBubbles")
@@ -168,9 +182,13 @@ class BlueBubblesAdapter(BaseChannelAdapter):
self._health_monitor_config: dict = config.get("health_monitor", {}) if config else {}
self._network_config: dict = config.get("network", {}) if config else {}
self._allow_private_network: bool = (
- self._network_config.get("dangerously_allow_private_network", False)
- or config.get("allow_private_network", False)
- ) if config else False
+ (
+ self._network_config.get("dangerously_allow_private_network", False)
+ or config.get("allow_private_network", False)
+ )
+ if config
+ else False
+ )
self._media_local_roots: list[str] = config.get("media_local_roots", []) if config else []
self._require_mention: bool = config.get("require_mention", False) if config else False
self._message_order_check: bool = config.get("message_order_check", True) if config else True
@@ -187,14 +205,18 @@ class BlueBubblesAdapter(BaseChannelAdapter):
self._catchup_task: asyncio.Task | None = None
self._catchup_cursor_store: CatchupCursorStore | None = None
self._sent_cache = SentMessageCache()
+ self._typing_sync = StreamingTypingSync()
+ self._conversation_bindings = ConversationBindings()
+ self._conversation_id = ConversationID()
+ self._conversation_route = ConversationRoute()
+ self._session_route = SessionRoute()
+ self._exec_approval = ExecApproval()
+ self._status_issues = StatusIssues()
- tts_available = check_tts_available()
- if not tts_available["espeak"]:
- self.capabilities.tts.voice.enabled = False
- logger.info("[BlueBubbles] TTS disabled: espeak not found")
- elif not tts_available["ffmpeg"] and not tts_available["pydub"]:
- self.capabilities.tts.voice.enabled = False
- logger.info("[BlueBubbles] TTS disabled: no transcoder (ffmpeg or pydub) found")
+ try:
+ import aiohttp # noqa: F401
+ except ImportError:
+ logger.info("[BlueBubbles] AI Vision unavailable: aiohttp not installed")
self._text_chunk_limit: int = config.get("text_chunk_limit", 4096) if config else 4096
self._chunk_mode: str = config.get("chunk_mode", "newline") if config else "newline"
@@ -286,6 +308,7 @@ class BlueBubblesAdapter(BaseChannelAdapter):
)
self._secret_contract.audit("password", "connect", {"server_url": self._server_url})
await self._client.__aenter__()
+ self._typing_sync.set_client(self._client)
health = await self.health_check()
if health.status == "unhealthy":
@@ -382,22 +405,40 @@ class BlueBubblesAdapter(BaseChannelAdapter):
await account_client.__aenter__()
self._account_clients[account_id] = account_client
logger.info("[BlueBubbles] Connected account: %s", account_id)
+
+ account_monitor = BlueBubblesMonitor(
+ server_url=account_cfg.server_url,
+ password=account_cfg.password,
+ on_event=self._handle_ws_event,
+ on_disconnect=lambda aid=account_id: logger.info("[BlueBubbles] Account WS disconnected: %s", aid),
+ on_reconnect=lambda aid=account_id: logger.info("[BlueBubbles] Account WS reconnected: %s", aid),
+ )
+ await account_monitor.start()
+ self._account_monitors[account_id] = account_monitor
+ logger.info("[BlueBubbles] Account WS monitor started: %s", account_id)
except Exception:
logger.warning("[BlueBubbles] Failed to connect account: %s", account_id)
async def disconnect(self) -> None:
+ await self._typing_sync.stop_all()
if self._catchup_task and not self._catchup_task.done():
self._catchup_task.cancel()
self._catchup_task = None
self._catchup_cursor_store = None
+ for account_id, account_monitor in self._account_monitors.items():
+ try:
+ await account_monitor.stop()
+ except Exception:
+ logger.debug("[BlueBubbles] Error stopping account monitor: %s", account_id)
+ self._account_monitors.clear()
+
for account_id, account_client in self._account_clients.items():
try:
await account_client.__aexit__()
except Exception:
logger.debug("[BlueBubbles] Error disconnecting account: %s", account_id)
self._account_clients.clear()
- self._account_monitors.clear()
if self._monitor:
await self._monitor.stop()
@@ -633,8 +674,9 @@ class BlueBubblesAdapter(BaseChannelAdapter):
except Exception as e:
return DeliveryResult(success=False, error=str(e))
- async def send_message_with_effect(self, chat_id: str, content: str, effect: str,
- reply_to: str | None = None) -> DeliveryResult:
+ async def send_message_with_effect(
+ self, chat_id: str, content: str, effect: str, reply_to: str | None = None
+ ) -> DeliveryResult:
if not self._client:
return DeliveryResult(success=False, error="Client not initialized")
try:
@@ -675,43 +717,31 @@ class BlueBubblesAdapter(BaseChannelAdapter):
return await self.send(response)
key = f"{chat_id}:{msg_id}"
- coalesce_enabled = self._block_streaming_coalesce.get("enabled", False)
- max_flush_ms = self._block_streaming_coalesce.get("max_flush_interval_ms", 500)
- if coalesce_enabled:
- self._stream_buffers[key] = self._stream_buffers.get(key, "") + chunk
-
- now = time.monotonic()
- last = self._last_edit_at.get(key, 0)
- if not finished and now - last < max_flush_ms / 1000.0:
- return DeliveryResult(success=True, message_id=msg_id)
-
- combined = self._stream_buffers.pop(key, "")
- prev_sent = self._stream_sent.get(key, "")
- full_content = prev_sent + combined
- self._stream_sent[key] = full_content
-
- lane_content = self._lane_sent.get(key, "")
- display_content = self._compose_stream_content(full_content, lane_content, finished)
-
- try:
- await update_activity(self._client, chat_id, msg_id, display_content)
- self._last_edit_at[key] = now
- if finished:
- self._cleanup_stream_key(key)
- return DeliveryResult(success=True, message_id=msg_id)
- except Exception as e:
- logger.warning(f"[BlueBubbles] Stream chunk update failed: {e}")
- return DeliveryResult(success=False, error=str(e))
+ if self._streaming_typing_indicator:
+ if finished:
+ await self._typing_sync.stop_typing(chat_id)
+ else:
+ await self._typing_sync.start_typing(chat_id)
self._stream_buffers[key] = self._stream_buffers.get(key, "") + chunk
now = time.monotonic()
last = self._last_edit_at.get(key, 0)
- interval_s = self._edit_interval_ms / 1000.0
- if not finished and now - last < interval_s:
- return DeliveryResult(success=True, message_id=msg_id)
+ coalesce_enabled = self._block_streaming_coalesce.get("enabled", False)
+ if coalesce_enabled:
+ max_flush_ms = self._block_streaming_coalesce.get("max_flush_interval_ms", 500)
+ if not finished and now - last < max_flush_ms / 1000.0:
+ return DeliveryResult(success=True, message_id=msg_id)
+ else:
+ interval_s = self._edit_interval_ms / 1000.0
+ if not finished and now - last < interval_s:
+ return DeliveryResult(success=True, message_id=msg_id)
+
+ return await self._flush_stream_chunk(key, chat_id, msg_id, finished)
+
+ async def _flush_stream_chunk(self, key: str, chat_id: str, msg_id: str, finished: bool) -> DeliveryResult:
combined = self._stream_buffers.pop(key, "")
prev_sent = self._stream_sent.get(key, "")
full_content = prev_sent + combined
@@ -722,7 +752,7 @@ class BlueBubblesAdapter(BaseChannelAdapter):
try:
await update_activity(self._client, chat_id, msg_id, display_content)
- self._last_edit_at[key] = now
+ self._last_edit_at[key] = time.monotonic()
if finished:
self._cleanup_stream_key(key)
return DeliveryResult(success=True, message_id=msg_id)
@@ -1111,6 +1141,10 @@ class BlueBubblesAdapter(BaseChannelAdapter):
hm_enabled = self._health_monitor_config.get("enabled", True)
health.metadata["health_monitor"] = {"enabled": hm_enabled}
+ ws_connected = self._monitor is not None and getattr(self._monitor, "_connected", False)
+ self._status_issues.check_connection(str(self._status.value), ws_connected)
+ health.metadata["issues"] = self._status_issues.get_issues()
+
return health
except Exception as e:
return HealthStatus(
diff --git a/backend/package/yuxi/channels/adapters/bluebubbles/config.py b/backend/package/yuxi/channels/adapters/bluebubbles/config.py
index b73e9320..242f994d 100644
--- a/backend/package/yuxi/channels/adapters/bluebubbles/config.py
+++ b/backend/package/yuxi/channels/adapters/bluebubbles/config.py
@@ -106,6 +106,13 @@ class BlueBubblesConfig(BaseModel):
auth_header_name: str = "X-BB-Password"
tts_prefer_caf: bool = True
+ def __repr__(self) -> str:
+ d = self.model_dump()
+ if d.get("password"):
+ d["password"] = "***"
+ fields = ", ".join(f"{k}={v!r}" for k, v in d.items())
+ return f"{self.__class__.__name__}({fields})"
+
@field_validator("allow_private_network", mode="before")
@classmethod
def _deprecate_allow_private_network(cls, v, info) -> Any:
@@ -116,6 +123,19 @@ class BlueBubblesConfig(BaseModel):
)
return v
+ @classmethod
+ def _validate_literal(cls, value: str, valid_values: list[str], field_name: str, default: str) -> str:
+ if value not in valid_values:
+ logger.warning(
+ "[BlueBubbles] Invalid value '%s' for '%s', falling back to '%s'. Valid values: %s",
+ value,
+ field_name,
+ default,
+ valid_values,
+ )
+ return default
+ return value
+
@classmethod
def from_env(cls) -> BlueBubblesConfig:
dm_raw = os.getenv("BLUEBUBBLES_DM_ALLOW_FROM", "")
@@ -137,7 +157,12 @@ class BlueBubblesConfig(BaseModel):
media_max_mb=int(os.getenv("BLUEBUBBLES_MEDIA_MAX_MB", "100")),
enable_stickers=os.getenv("BLUEBUBBLES_ENABLE_STICKERS", "true").lower() == "true",
media_local_roots=[x.strip() for x in media_roots_raw.split(",") if x.strip()] if media_roots_raw else [],
- streaming_mode=os.getenv("BLUEBUBBLES_STREAMING_MODE", "partial"),
+ streaming_mode=cls._validate_literal(
+ os.getenv("BLUEBUBBLES_STREAMING_MODE", "partial"),
+ ["off", "partial", "block"],
+ "streaming_mode",
+ "partial",
+ ),
edit_interval_ms=int(os.getenv("BLUEBUBBLES_EDIT_INTERVAL_MS", "500")),
tapback_enabled=os.getenv("BLUEBUBBLES_TAPBACK_ENABLED", "true").lower() == "true",
require_mention=os.getenv("BLUEBUBBLES_REQUIRE_MENTION", "false").lower() == "true",
@@ -147,6 +172,42 @@ class BlueBubblesConfig(BaseModel):
allow_private_network=os.getenv("BLUEBUBBLES_ALLOW_PRIVATE_NETWORK", "false").lower() == "true",
webhook_path=os.getenv("BLUEBUBBLES_WEBHOOK_PATH", "/bluebubbles/webhook"),
webhook_secret=os.getenv("BLUEBUBBLES_WEBHOOK_SECRET", ""),
- auth_strategy=os.getenv("BLUEBUBBLES_AUTH_STRATEGY", "header"),
+ auth_strategy=cls._validate_literal(
+ os.getenv("BLUEBUBBLES_AUTH_STRATEGY", "header"),
+ ["header", "query", "both"],
+ "auth_strategy",
+ "header",
+ ),
tts_prefer_caf=os.getenv("BLUEBUBBLES_TTS_PREFER_CAF", "true").lower() == "true",
+ streaming_typing_indicator=os.getenv("BLUEBUBBLES_STREAMING_TYPING", "true").lower() == "true",
+ message_order_check=os.getenv("BLUEBUBBLES_MESSAGE_ORDER_CHECK", "true").lower() == "true",
+ max_cache_entries=int(os.getenv("BLUEBUBBLES_MAX_CACHE_ENTRIES", "2048")),
+ dm_history_limit=int(os.getenv("BLUEBUBBLES_DM_HISTORY_LIMIT", "100")),
+ coalesce_same_sender_dms=os.getenv("BLUEBUBBLES_COALESCE_DMS", "false").lower() == "true",
+ enrich_group_participants_from_contacts=os.getenv("BLUEBUBBLES_ENRICH_GROUP", "true").lower() == "true",
+ config_writes=os.getenv("BLUEBUBBLES_CONFIG_WRITES", "true").lower() == "true",
+ block_streaming_coalesce=BlockStreamingCoalesceConfig(
+ enabled=os.getenv("BLUEBUBBLES_COALESCE_ENABLED", "false").lower() == "true",
+ max_flush_interval_ms=int(os.getenv("BLUEBUBBLES_COALESCE_MAX_FLUSH_MS", "500")),
+ ),
+ network=NetworkConfig(
+ dangerously_allow_private_network=os.getenv(
+ "BLUEBUBBLES_NETWORK_DANGEROUSLY_ALLOW_PRIVATE", "false"
+ ).lower()
+ == "true",
+ ),
+ markdown=MarkdownConfig(
+ enabled=os.getenv("BLUEBUBBLES_MARKDOWN_ENABLED", "true").lower() == "true",
+ code_blocks=os.getenv("BLUEBUBBLES_MARKDOWN_CODE_BLOCKS", "true").lower() == "true",
+ ),
+ catchup=CatchupConfig(
+ enabled=os.getenv("BLUEBUBBLES_CATCHUP_ENABLED", "true").lower() == "true",
+ max_age_minutes=int(os.getenv("BLUEBUBBLES_CATCHUP_MAX_AGE_MINUTES", "120")),
+ per_run_limit=int(os.getenv("BLUEBUBBLES_CATCHUP_PER_RUN_LIMIT", "50")),
+ first_run_lookback_minutes=int(os.getenv("BLUEBUBBLES_CATCHUP_FIRST_LOOKBACK", "30")),
+ max_failure_retries=int(os.getenv("BLUEBUBBLES_CATCHUP_MAX_FAILURES", "10")),
+ ),
+ health_monitor=HealthMonitorConfig(
+ enabled=os.getenv("BLUEBUBBLES_HEALTH_MONITOR_ENABLED", "true").lower() == "true",
+ ),
)
diff --git a/backend/package/yuxi/channels/adapters/bluebubbles/monitor_debounce.py b/backend/package/yuxi/channels/adapters/bluebubbles/monitor_debounce.py
index ea637fdd..b470ffa6 100644
--- a/backend/package/yuxi/channels/adapters/bluebubbles/monitor_debounce.py
+++ b/backend/package/yuxi/channels/adapters/bluebubbles/monitor_debounce.py
@@ -24,3 +24,7 @@ class MonitorDebounce:
expired = [k for k, v in self._cache.items() if now - v > self.WINDOW_SECONDS]
for k in expired:
del self._cache[k]
+ if len(self._cache) > self.MAX_ENTRIES:
+ oldest = sorted(self._cache.items(), key=lambda x: x[1])[: len(self._cache) - self.MAX_ENTRIES]
+ for k, _ in oldest:
+ del self._cache[k]
diff --git a/backend/package/yuxi/channels/adapters/bluebubbles/renderer.py b/backend/package/yuxi/channels/adapters/bluebubbles/renderer.py
index 237480a5..b8edb918 100644
--- a/backend/package/yuxi/channels/adapters/bluebubbles/renderer.py
+++ b/backend/package/yuxi/channels/adapters/bluebubbles/renderer.py
@@ -77,8 +77,6 @@ def render_markdown_to_imessage(text: str) -> str:
if not text:
return ""
- import re
-
codes: dict[str, str] = {}
def _save_code(m: re.Match) -> str:
@@ -88,12 +86,58 @@ def render_markdown_to_imessage(text: str) -> str:
text = re.sub(r"`[^`]+`", _save_code, text)
- text = re.sub(r"\*\*(.+?)\*\*", r"\1", text)
- text = re.sub(r"__(.+?)__", r"\1", text)
-
- text = re.sub(r"(?\1", text)
+ tokens = _tokenize_markdown(text)
+ result = _render_tokens(tokens)
for key, val in codes.items():
- text = text.replace(key, val)
+ result = result.replace(key, val)
- return text
+ return result
+
+
+def _tokenize_markdown(text: str) -> list[tuple[str, str]]:
+ TOKEN_RE = re.compile(r"(\*\*|__|(? pos:
+ tokens.append(("text", text[pos : m.start()]))
+ tokens.append(("marker", m.group()))
+ pos = m.end()
+ if pos < len(text):
+ tokens.append(("text", text[pos:]))
+ return tokens
+
+
+def _render_tokens(tokens: list[tuple[str, str]]) -> str:
+ stack: list[tuple[str, str]] = []
+ result: list[str] = []
+
+ MARKER_MAP = {
+ "**": ("", ""),
+ "__": ("", ""),
+ "*": ("", ""),
+ "~": ("", ""),
+ "~~": ("", ""),
+ }
+
+ for kind, value in tokens:
+ if kind == "text":
+ result.append(value)
+ elif kind == "marker":
+ if value in MARKER_MAP:
+ if stack and stack[-1][0] == value:
+ stack.pop()
+ result.append(MARKER_MAP[value][1])
+ else:
+ stack.append((value, MARKER_MAP[value][0]))
+ result.append(MARKER_MAP[value][0])
+ else:
+ result.append(value)
+
+ for marker, _ in reversed(stack):
+ mapped = MARKER_MAP.get(marker)
+ if mapped:
+ result.append(mapped[1])
+
+ return "".join(result)
diff --git a/backend/package/yuxi/channels/adapters/bluebubbles/streaming_enhanced.py b/backend/package/yuxi/channels/adapters/bluebubbles/streaming_enhanced.py
index 3d0d2666..5f0be416 100644
--- a/backend/package/yuxi/channels/adapters/bluebubbles/streaming_enhanced.py
+++ b/backend/package/yuxi/channels/adapters/bluebubbles/streaming_enhanced.py
@@ -12,13 +12,20 @@ from yuxi.utils.logging_config import logger
class StreamingTypingSync:
TYPING_REFRESH_INTERVAL = 8.0
+ DEFAULT_STREAM_TIMEOUT = 60.0
- def __init__(self, client: BlueBubblesClient | None = None):
+ def __init__(
+ self,
+ client: BlueBubblesClient | None = None,
+ stream_timeout: float = DEFAULT_STREAM_TIMEOUT,
+ ):
self._client = client
self._active_chats: set[str] = set()
self._last_typing: dict[str, float] = {}
self._keepalive_task: asyncio.Task | None = None
self._running = False
+ self._stream_timeout = stream_timeout
+ self._stream_started: dict[str, float] = {}
def set_client(self, client: BlueBubblesClient) -> None:
self._client = client
@@ -28,6 +35,7 @@ class StreamingTypingSync:
return
self._active_chats.add(chat_id)
+ self._stream_started.setdefault(chat_id, time.monotonic())
try:
await send_typing_indicator(self._client, chat_id, display=True)
self._last_typing[chat_id] = time.monotonic()
@@ -40,6 +48,7 @@ class StreamingTypingSync:
async def stop_typing(self, chat_id: str) -> None:
self._active_chats.discard(chat_id)
+ self._stream_started.pop(chat_id, None)
if not self._client:
return
try:
@@ -74,6 +83,14 @@ class StreamingTypingSync:
except Exception:
logger.debug(f"[BlueBubbles] Typing keepalive failed for {chat_id}")
+ stream_started = self._stream_started.get(chat_id)
+ if stream_started and (now - stream_started) > self._stream_timeout:
+ logger.warning(
+ f"[BlueBubbles] Stream timeout for {chat_id}, "
+ f"({now - stream_started:.0f}s), auto-stopping typing"
+ )
+ await self.stop_typing(chat_id)
+
async def send_stream_with_typing(
adapter: Any,