@samitouri / QOSami_Agents_Zero-1 / commits / af60a6e2

feat(telegram): message formatting rewrite, cross-event-loop fixes & hot mode switching

keyboardstaff committed Mar 21, 2026 at 09:02 UTC af60a6e29193420fa22395787b57ce24c1f618f6

4 files changed +267 -35

plugins/_telegram_integration/extensions/python/job_loop/_10_telegram_bot.py

+7 -4

@@ -52,10 +52,13 @@ class TelegramBotManager(Extension):

52 continue

53 if name in running:

54 inst = running[name]

55 - # Already running: polling (task alive) or webhook (active)

56 - if (inst.task and not inst.task.done()) or inst.webhook_active:

57 - continue

58 - # Instance exists but is dead — stop and recreate

55 + current_mode = bot_cfg.get("mode", "polling")

56 + running_mode = "webhook" if inst.webhook_active else "polling"

57 + if current_mode == running_mode:

58 + # Same mode and still alive → skip

59 + if (inst.task and not inst.task.done()) or inst.webhook_active:

60 + continue

61 + # Mode changed or instance died — stop and recreate

62 await stop_bot(name)

64 try:

plugins/_telegram_integration/helpers/bot_manager.py

+1 -1

@@ -47,7 +47,7 @@ def create_bot(

47 on_callback_query: Callable[..., Awaitable] | None = None,

48 group_mode: str = "mention",

49 ) -> BotInstance:

50 - bot = Bot(token=token, default=DefaultBotProperties(parse_mode=ParseMode.MARKDOWN))

50 + bot = Bot(token=token, default=DefaultBotProperties(parse_mode=ParseMode.HTML))

51 dp = Dispatcher()

52 router = Router()

plugins/_telegram_integration/helpers/handler.py

+87 -25

@@ -4,10 +4,13 @@ import threading

4 import time

5 import uuid

7 +from aiogram import Bot

8 +from aiogram.client.default import DefaultBotProperties

9 +from aiogram.enums import ParseMode

10 from aiogram.types import Message as TgMessage, CallbackQuery

12 from agent import AgentContext, UserMessage

10 -from helpers import plugins, files

13 +from helpers import plugins, files, projects

14 from helpers import message_queue as mq

15 from helpers.notification import NotificationManager, NotificationType, NotificationPriority

16 from helpers.persist_chat import save_tmp_chat

@@ -28,6 +31,7 @@ CTX_TG_BOT = "telegram_bot"

31 CTX_TG_CHAT_ID = "telegram_chat_id"

32 CTX_TG_USER_ID = "telegram_user_id"

33 CTX_TG_USERNAME = "telegram_username"

34 +CTX_TG_TYPING_STOP = "_telegram_typing_stop"

36 # Transient

37 CTX_TG_ATTACHMENTS = "_telegram_response_attachments"

@@ -122,17 +126,19 @@ async def handle_start(message: TgMessage, bot_name: str, bot_cfg: dict):

126 return

127

128 if not _is_allowed(bot_cfg, user.id, user.username):

125 - await message.reply("⛔ You are not authorized to use this bot.")

129 + await message.reply("You are not authorized to use this bot.")

130 return

131

132 instance = get_bot(bot_name)

133 if not instance:

134 return

135

132 - await message.reply(

133 - f"👋 Hello {user.first_name}! I'm connected to Agent Zero.\n\n"

136 + await _send_with_temp_bot(

137 + instance.bot.token, message.chat.id,

138 + f"\U0001f44b Hello {user.first_name}! I'm connected to Agent Zero.\n\n"

139 "Send me a message and I'll process it.\n"

135 - "Use /clear to reset the conversation."

140 + "Use /clear to reset the conversation.",

141 + parse_mode=None,

142 )

143

144 # Ensure a chat context exists

@@ -161,9 +167,9 @@ async def handle_clear(message: TgMessage, bot_name: str, bot_cfg: dict):

167

168 instance = get_bot(bot_name)

169 if instance:

164 - await tc.send_text(

165 - instance.bot, message.chat.id,

166 - "🗑 Chat cleared. Send a new message to start fresh.",

170 + await _send_with_temp_bot(

171 + instance.bot.token, message.chat.id,

172 + "Chat cleared. Send a new message to start fresh.",

173 parse_mode=None,

174 )

175

@@ -192,22 +198,35 @@ async def handle_message(message: TgMessage, bot_name: str, bot_cfg: dict):

198 if not instance:

199 return

200

195 - # Send typing indicator

196 - await tc.send_typing(instance.bot, message.chat.id)

201 + # Start persistent typing indicator (thread-based, works across event loops)

202 + typing_stop = _start_typing(instance.bot.token, message.chat.id)

203

204 # Get or create agent context

205 context = await _get_or_create_context(bot_name, bot_cfg, message)

206 if not context:

201 - await tc.send_text(

202 - instance.bot, message.chat.id,

203 - "❌ Failed to create chat session.",

207 + typing_stop.set()

208 + await _send_with_temp_bot(

209 + instance.bot.token, message.chat.id,

210 + "Failed to create chat session.",

211 parse_mode=None,

212 )

213 return

214

215 + # Store stop event so send_telegram_reply can cancel typing

216 + context.data[CTX_TG_TYPING_STOP] = typing_stop

217 +

218 # Build user message text

219 text = _extract_message_content(message)

210 - attachments = await _download_attachments(instance.bot, message, bot_name=bot_name)

220 +

221 + # Use temp bot for downloads (cross-event-loop safe)

222 + dl_bot = Bot(token=instance.bot.token)

223 + try:

224 + attachments = await _download_attachments(dl_bot, message, bot_name=bot_name)

225 + finally:

226 + try:

227 + await dl_bot.session.close()

228 + except Exception:

229 + pass

230

231 # Build user message with prompt

232 agent = context.agent0

@@ -332,7 +351,6 @@ async def _get_or_create_context_from_user(

351

352 project = _get_project(bot_cfg, user_id)

353 if project:

335 - from helpers import projects

354 projects.activate_project(ctx.id, project)

355

356 chats[key] = ctx.id

@@ -445,10 +463,6 @@ async def send_telegram_reply(

463 keyboard: list[list[dict]] | None = None,

464 ) -> str | None:

465 """Send reply to Telegram user. Returns error string or None on success."""

448 - from aiogram import Bot

449 - from aiogram.client.default import DefaultBotProperties

450 - from aiogram.enums import ParseMode

451 -

466 bot_name = context.data.get(CTX_TG_BOT)

467 if not bot_name:

468 return "No Telegram bot configured on context"

@@ -461,11 +475,14 @@ async def send_telegram_reply(

475 if not chat_id:

476 return "No chat_id on context"

477

464 - # Create a temporary Bot bound to the current event loop to avoid

465 - # cross-event-loop issues with the shared instance's aiohttp session.

478 + # Cancel persistent typing indicator

479 + typing_stop = context.data.pop(CTX_TG_TYPING_STOP, None)

480 + if typing_stop:

481 + typing_stop.set()

482 +

483 reply_bot = Bot(

484 token=instance.bot.token,

468 - default=DefaultBotProperties(parse_mode=ParseMode.MARKDOWN),

485 + default=DefaultBotProperties(parse_mode=ParseMode.HTML),

486 )

487 try:

488 # Send attachments first

@@ -476,14 +493,15 @@ async def send_telegram_reply(

493 else:

494 await tc.send_file(reply_bot, chat_id, path)

495

479 - # Send text (with or without keyboard)

496 + # Send text (with or without keyboard), convert Markdown → HTML

497 if response_text:

498 + html_text = tc.md_to_telegram_html(response_text)

499 if keyboard:

500 await tc.send_text_with_keyboard(

483 - reply_bot, chat_id, response_text, keyboard,

501 + reply_bot, chat_id, html_text, keyboard,

502 )

503 else:

486 - await tc.send_text(reply_bot, chat_id, response_text)

504 + await tc.send_text(reply_bot, chat_id, html_text)

505

506 return None

507

@@ -499,6 +517,50 @@ async def send_telegram_reply(

517

518 # Helpers

519

520 +async def _send_with_temp_bot(token: str, chat_id: int, text: str, parse_mode: str | None = None):

521 + """Send text using a temporary Bot to avoid cross-event-loop session issues."""

522 + bot = Bot(token=token)

523 + try:

524 + await tc.send_text(bot, chat_id, text, parse_mode=parse_mode)

525 + finally:

526 + try:

527 + await bot.session.close()

528 + except Exception:

529 + pass

530 +

531 +

532 +def _start_typing(token: str, chat_id: int) -> threading.Event:

533 + """Spawn a daemon thread that sends typing every 4s. Returns a stop Event."""

534 + stop = threading.Event()

535 +

536 + def _run():

537 + import asyncio

538 +

539 + async def _loop():

540 + bot = Bot(token=token)

541 + try:

542 + while not stop.is_set():

543 + await tc.send_typing(bot, chat_id)

544 + # Sleep 4s total, checking stop every 0.5s

545 + for _ in range(8):

546 + if stop.is_set():

547 + return

548 + await asyncio.sleep(0.5)

549 + except Exception:

550 + pass

551 + finally:

552 + try:

553 + await bot.session.close()

554 + except Exception:

555 + pass

556 +

557 + asyncio.run(_loop())

558 +

559 + t = threading.Thread(target=_run, daemon=True)

560 + t.start()

561 + return stop

562 +

563 +

564 def _format_user(user) -> str:

565 name = user.first_name or ""

566 if user.last_name:

plugins/_telegram_integration/helpers/telegram_client.py

+172 -5

@@ -1,4 +1,5 @@

1 import os

2 +import re

4 from aiogram import Bot

5 from aiogram.exceptions import TelegramBadRequest

@@ -11,6 +12,8 @@ from aiogram.types import (

12 from helpers.errors import format_error

13 from helpers.print_style import PrintStyle

15 +_UNSET = object() # sentinel: "not provided" (lets Bot default apply)

16 +

17 # Text messages

19 MAX_MESSAGE_LENGTH: int = 4096 # Telegram message length limit

@@ -21,26 +24,34 @@ async def send_text(

24 chat_id: int,

25 text: str,

26 reply_to_message_id: int | None = None,

24 - parse_mode: str | None = None,

27 + parse_mode: object = _UNSET,

28 ) -> int | None:

26 - """Send text message, splitting if too long. Returns last message_id or None on error."""

29 + """Send text message, splitting if too long. Returns last message_id or None on error.

30 +

31 + parse_mode behaviour:

32 + - _UNSET (default): omitted from send_message → Bot's DefaultBotProperties applies.

33 + - None: explicitly no formatting.

34 + - "HTML"/"Markdown"/etc.: that specific mode.

35 + """

36 try:

37 chunks = _split_text(text, MAX_MESSAGE_LENGTH)

38 last_msg_id = None

39 + pm_kwargs: dict = {} if parse_mode is _UNSET else {"parse_mode": parse_mode}

40 for chunk in chunks:

41 try:

42 msg = await bot.send_message(

43 chat_id=chat_id,

44 text=chunk,

45 reply_to_message_id=reply_to_message_id,

36 - parse_mode=parse_mode,

46 + **pm_kwargs,

47 )

48 last_msg_id = msg.message_id

49 except TelegramBadRequest:

40 - # Retry without markdown if parse fails

50 + # Retry as plain text, stripping HTML tags

51 + plain = re.sub(r"<[^>]+>", "", chunk)

52 msg = await bot.send_message(

53 chat_id=chat_id,

43 - text=chunk,

54 + text=plain,

55 reply_to_message_id=reply_to_message_id,

56 parse_mode=None,

57 )

@@ -133,15 +144,18 @@ async def send_text_with_keyboard(

144 text: str,

145 buttons: list[list[dict]],

146 reply_to_message_id: int | None = None,

147 + parse_mode: object = _UNSET,

148 ) -> int | None:

149 """Send text with inline keyboard buttons."""

150 try:

151 keyboard = build_inline_keyboard(buttons)

152 + pm_kwargs: dict = {} if parse_mode is _UNSET else {"parse_mode": parse_mode}

153 msg = await bot.send_message(

154 chat_id=chat_id,

155 text=text,

156 reply_markup=keyboard,

157 reply_to_message_id=reply_to_message_id,

158 + **pm_kwargs,

159 )

160 return msg.message_id

161 except Exception as e:

@@ -201,3 +215,156 @@ _IMAGE_EXTENSIONS = {".jpg", ".jpeg", ".png", ".gif", ".webp", ".bmp"}

215 def is_image_file(path: str) -> bool:

216 _, ext = os.path.splitext(path.lower())

217 return ext in _IMAGE_EXTENSIONS

218 +

219 +

220 +def md_to_telegram_html(text: str) -> str:

221 + """Convert standard Markdown to Telegram-compatible HTML.

222 +

223 + Handles: fenced code blocks, inline code, bold, italic, strikethrough,

224 + links, headings, blockquotes, tables, and nested lists.

225 + """

226 + stash: list[str] = []

227 +

228 + def _put(html: str) -> str:

229 + stash.append(html)

230 + return f"\x00B{len(stash) - 1}\x00"

231 +

232 + def _esc(t: str) -> str:

233 + return t.replace("&", "&").replace("<", "<").replace(">", ">")

234 +

235 + # Stash fenced code blocks

236 + def _code_block(m: re.Match) -> str:

237 + lang, code = m.group(1) or "", _esc(m.group(2))

238 + tag = f'<pre><code class="language-{lang}">{code}</code></pre>' if lang else f"<pre>{code}</pre>"

239 + return _put(tag)

240 +

241 + text = re.sub(r"```(\w*)\n(.*?)```", _code_block, text, flags=re.DOTALL)

242 +

243 + # Stash inline code

244 + text = re.sub(r"`([^`]+)`", lambda m: _put(f"<code>{_esc(m.group(1))}</code>"), text)

245 +

246 + # Stash Markdown tables as list-style text

247 + text = _stash_tables(text, _put, _esc)

248 +

249 + # Escape remaining HTML chars

250 + text = _esc(text)

251 +

252 + # Inline formatting

253 + text = re.sub(r"\*\*(.+?)\*\*", r"\1", text)

254 + text = re.sub(r"__(.+?)__", r"\1", text)

255 + text = re.sub(r"(?<!\w)\*([^*]+?)\*(?!\w)", r"\1", text)

256 + text = re.sub(r"(?<!\w)_([^_]+?)_(?!\w)", r"\1", text)

257 + text = re.sub(r"~~(.+?)~~", r"<s>\1</s>", text)

258 + text = re.sub(r"\[([^\]]+)\]$([^)]+)$", r'<a href="\2">\1</a>', text)

259 + text = re.sub(r"^#{1,6}\s+(.+)$", r"\1", text, flags=re.MULTILINE)

260 +

261 + # Blockquotes: > text → <blockquote>

262 + text = _convert_blockquotes(text)

263 +

264 + # Lists (nested + ordered)

265 + text = _convert_lists(text)

266 +

267 + # Restore all stashed blocks

268 + for i, block in enumerate(stash):

269 + text = text.replace(f"\x00B{i}\x00", block)

270 + return text

271 +

272 +

273 +# ── Helpers for md_to_telegram_html ──

274 +

275 +_TABLE_ROW = re.compile(r"^\|(.+)\|$")

276 +_TABLE_SEP = re.compile(r"^[\s|:-]+$")

277 +

278 +

279 +def _stash_tables(text: str, put_fn, esc_fn) -> str:

280 + """Convert Markdown tables to key-value list format and stash them.

281 +

282 + | Name | Age |

283 + |-------|-----|

284 + | Alice | 30 | → • Name: Alice, Age: 30

285 + | Bob | 25 | • Name: Bob, Age: 25

286 + """

287 + lines = text.split("\n")

288 + out: list[str] = []

289 + headers: list[str] = []

290 + data_rows: list[list[str]] = []

291 +

292 + def _flush():

293 + if not headers and not data_rows:

294 + return

295 + if headers and data_rows:

296 + items: list[str] = []

297 + for row in data_rows:

298 + pairs = ", ".join(

299 + f"{headers[i]}: {row[i]}" if i < len(row) else headers[i]

300 + for i in range(len(headers))

301 + )

302 + items.append(f"\u2022 {pairs}")

303 + out.append(put_fn(esc_fn("\n".join(items))))

304 + elif data_rows:

305 + # No header row — just bullet each row

306 + for row in data_rows:

307 + out.append(put_fn(esc_fn(f"\u2022 {', '.join(row)}")))

308 + headers.clear()

309 + data_rows.clear()

310 +

311 + for line in lines:

312 + m = _TABLE_ROW.match(line.strip())

313 + if m:

314 + if _TABLE_SEP.match(line.strip()):

315 + continue

316 + cells = [c.strip() for c in m.group(1).split("|")]

317 + if not headers:

318 + headers.extend(cells)

319 + else:

320 + data_rows.append(cells)

321 + else:

322 + _flush()

323 + out.append(line)

324 + _flush()

325 + return "\n".join(out)

326 +

327 +

328 +_LIST_ITEM = re.compile(r"^( *)([-*+]|\d+\.)\s+(.*)$")

329 +_BULLETS = ["\u2022", "\u25e6", "\u25aa"] # • ◦ ▪

330 +

331 +

332 +def _convert_lists(text: str) -> str:

333 + """Convert Markdown list items to indented bullet / numbered lines."""

334 + lines = text.split("\n")

335 + out: list[str] = []

336 + for line in lines:

337 + m = _LIST_ITEM.match(line)

338 + if m:

339 + depth = len(m.group(1)) // 2

340 + marker, content = m.group(2), m.group(3)

341 + px = " " * depth

342 + if marker.rstrip(".").isdigit():

343 + out.append(f"{px}{marker} {content}")

344 + else:

345 + out.append(f"{px}{_BULLETS[min(depth, len(_BULLETS) - 1)]} {content}")

346 + else:

347 + out.append(line)

348 + return "\n".join(out)

349 +

350 +

351 +def _convert_blockquotes(text: str) -> str:

352 + """Convert Markdown blockquotes (> text) to <blockquote> tags."""

353 + lines = text.split("\n")

354 + out: list[str] = []

355 + quote_buf: list[str] = []

356 +

357 + def _flush():

358 + if quote_buf:

359 + out.append("<blockquote>" + "\n".join(quote_buf) + "</blockquote>")

360 + quote_buf.clear()

361 +

362 + for line in lines:

363 + m = re.match(r"^>\s?(.*)", line) # > is already HTML-escaped at this point

364 + if m:

365 + quote_buf.append(m.group(1))

366 + else:

367 + _flush()

368 + out.append(line)

369 + _flush()

370 + return "\n".join(out)