@samitouri / QOSami_Agents_Zero-2 / commits / 1bbbfa44

Improve Telegram integration command and streaming UX

Anmol Malik committed May 26, 2026 at 20:18 UTC 1bbbfa44acd87b9702176dfb88be410bcd861554
16 files changed +1405 -90
helpers/integration_commands.py
+302 -12
@@ -1,6 +1,7 @@
1 from __future__ import annotations
2
3 import re
4 +from dataclasses import dataclass
5 from typing import TYPE_CHECKING
6
7 from helpers import message_queue as mq
@@ -14,7 +15,83 @@ if TYPE_CHECKING:
15
16
17 _CLEAR_VALUES = {"", "default", "none", "clear", "off"}
17 -_SUPPORTED_COMMANDS = {"/send", "/queue", "/project", "/config", "/preset"}
18 +
19 +
20 +@dataclass(frozen=True)
21 +class IntegrationCommandDef:
22 + name: str
23 + description: str
24 + category: str
25 + aliases: tuple[str, ...] = ()
26 + args_hint: str = ""
27 + menu: bool = True
28 +
29 +
30 +COMMAND_REGISTRY: tuple[IntegrationCommandDef, ...] = (
31 + IntegrationCommandDef("commands", "Show all integration commands.", "Info", aliases=("help",)),
32 + IntegrationCommandDef(
33 + "status",
34 + "Show this chat's project, model, agent, and queue state.",
35 + "Info",
36 + ),
37 + IntegrationCommandDef("new", "Start a fresh chat context.", "Session"),
38 + IntegrationCommandDef("clear", "Reset the current chat context.", "Session", aliases=("reset",)),
39 + IntegrationCommandDef(
40 + "queue",
41 + "Show or manage queued messages.",
42 + "Session",
43 + args_hint="[send|clear]",
44 + ),
45 + IntegrationCommandDef("send", "Send queued messages now.", "Session", aliases=("push",)),
46 + IntegrationCommandDef(
47 + "steer",
48 + "Intervene in the currently running task.",
49 + "Session",
50 + args_hint="<message>",
51 + ),
52 + IntegrationCommandDef("pause", "Pause the active run.", "Session"),
53 + IntegrationCommandDef("resume", "Resume a paused run.", "Session"),
54 + IntegrationCommandDef("nudge", "Nudge the active run.", "Session"),
55 + IntegrationCommandDef(
56 + "stream",
57 + "Enable or disable Telegram response streaming.",
58 + "Configuration",
59 + args_hint="[on|off]",
60 + ),
61 + IntegrationCommandDef(
62 + "tools",
63 + "Show or hide Telegram tool progress.",
64 + "Configuration",
65 + args_hint="[on|off]",
66 + ),
67 + IntegrationCommandDef(
68 + "project",
69 + "Show or switch the active project.",
70 + "Configuration",
71 + args_hint="[name|none]",
72 + ),
73 + IntegrationCommandDef(
74 + "model",
75 + "Show or switch the chat model preset.",
76 + "Configuration",
77 + aliases=("config", "preset"),
78 + args_hint="[preset|default]",
79 + ),
80 + IntegrationCommandDef(
81 + "agent",
82 + "Show or switch the agent profile.",
83 + "Configuration",
84 + aliases=("profile",),
85 + args_hint="[profile]",
86 + ),
87 +)
88 +
89 +
90 +_COMMAND_LOOKUP = {
91 + f"/{name}": command
92 + for command in COMMAND_REGISTRY
93 + for name in (command.name, *command.aliases)
94 +}
95
96
97 def extract_command_line(text: str) -> str:
@@ -32,11 +109,53 @@ def parse_command(text: str) -> tuple[str, str] | None:
109 return None
110
111 command, _, args = line.partition(" ")
35 - command = command.strip().lower()
36 - if command not in _SUPPORTED_COMMANDS:
112 + command = _normalize_command_token(command)
113 + resolved = resolve_command(command)
114 + if not resolved:
115 return None
116
39 - return command, args.strip()
117 + return f"/{resolved.name}", args.strip()
118 +
119 +
120 +def resolve_command(command: str) -> IntegrationCommandDef | None:
121 + normalized = _normalize_command_token(command)
122 + if not normalized.startswith("/"):
123 + normalized = f"/{normalized}"
124 + return _COMMAND_LOOKUP.get(normalized)
125 +
126 +
127 +def telegram_menu_commands() -> list[tuple[str, str]]:
128 + return [
129 + (command.name, _telegram_description(command))
130 + for command in COMMAND_REGISTRY
131 + if command.menu
132 + ]
133 +
134 +
135 +def command_names(include_aliases: bool = True) -> list[str]:
136 + names: list[str] = []
137 + for command in COMMAND_REGISTRY:
138 + names.append(command.name)
139 + if include_aliases:
140 + names.extend(command.aliases)
141 + return names
142 +
143 +
144 +def help_text(*, full: bool = False) -> str:
145 + commands = COMMAND_REGISTRY if full else tuple(c for c in COMMAND_REGISTRY if c.menu)
146 + lines = ["Available commands:"]
147 + for command in commands:
148 + args = f" {command.args_hint}" if command.args_hint else ""
149 + alias_text = ""
150 + if command.aliases:
151 + alias_text = f" (alias: {', '.join('/' + alias for alias in command.aliases)})"
152 + lines.append(f"/{command.name}{args} - {command.description}{alias_text}")
153 + return "\n".join(lines)
154 +
155 +
156 +def unknown_command_text(command: str) -> str:
157 + token = _normalize_command_token(command).split(" ", 1)[0]
158 + return f"Unknown command: {token}\n\n{help_text(full=True)}"
159
160
161 def try_handle_command(context: "AgentContext", text: str) -> str | None:
@@ -45,17 +164,47 @@ def try_handle_command(context: "AgentContext", text: str) -> str | None:
164 return None
165
166 command, args = parsed
167 + if command == "/commands":
168 + return help_text(full=True)
169 + if command == "/status":
170 + return _handle_status(context)
171 + if command in {"/new", "/clear"}:
172 + return _handle_clear(context, new_chat=(command == "/new"))
173 if command == "/send":
174 return _handle_queue(context, "send")
175 if command == "/queue":
176 return _handle_queue(context, args)
177 + if command == "/steer":
178 + return _handle_steer(context, args)
179 + if command == "/pause":
180 + return _handle_pause(context)
181 + if command == "/resume":
182 + return _handle_resume(context)
183 + if command == "/nudge":
184 + return _handle_nudge(context)
185 + if command == "/stream":
186 + return _handle_toggle(context, args, "telegram_stream_enabled", "Response streaming")
187 + if command == "/tools":
188 + return _handle_toggle(context, args, "telegram_tools_enabled", "Tool progress")
189 if command == "/project":
190 return _handle_project(context, args)
54 - if command in {"/config", "/preset"}:
55 - return _handle_config(context, args)
191 + if command == "/model":
192 + return _handle_model(context, args)
193 + if command == "/agent":
194 + return _handle_agent(context, args)
195 return None
196
197
198 +def _normalize_command_token(command: str) -> str:
199 + normalized = command.strip().lower()
200 + if not normalized:
201 + return ""
202 + token, *rest = normalized.split(" ", 1)
203 + if "@" in token:
204 + token = token.split("@", 1)[0]
205 + return f"{token} {rest[0]}".strip() if rest else token
206 +
207 +
208 def _handle_queue(context: "AgentContext", args: str) -> str:
209 queue = mq.get_queue(context)
210 count = len(queue)
@@ -68,8 +217,13 @@ def _handle_queue(context: "AgentContext", args: str) -> str:
217 "Use /send or /queue send to send everything as one batch."
218 )
219
220 + if action in {"clear", "reset"}:
221 + mq.remove(context)
222 + mark_dirty_for_context(context.id, reason="integration_commands.queue_clear")
223 + return "Queue cleared."
224 +
225 if action not in {"send", "all"}:
72 - return "Unknown queue action. Use /queue send to flush the queue."
226 + return "Unknown queue action. Use /queue send to flush or /queue clear to clear."
227
228 if count == 0:
229 return "Queue is empty."
@@ -80,6 +234,73 @@ def _handle_queue(context: "AgentContext", args: str) -> str:
234 return f"Sent {sent_count} queued {noun} as one batch."
235
236
237 +def _handle_status(context: "AgentContext") -> str:
238 + project_name = context.get_data("project") or "none"
239 + override = context.get_data("chat_model_override")
240 + agent_profile = getattr(context.agent0.config, "profile", "default")
241 + running = "running" if context.is_running() else "idle"
242 + if getattr(context, "paused", False):
243 + running = "paused"
244 + queue_count = len(mq.get_queue(context))
245 + return (
246 + f"Status: {running}\n"
247 + f"Project: {project_name}\n"
248 + f"Model: {_describe_override(override)}\n"
249 + f"Agent: {agent_profile}\n"
250 + f"Queued messages: {queue_count}"
251 + )
252 +
253 +
254 +def _handle_clear(context: "AgentContext", *, new_chat: bool) -> str:
255 + context.reset()
256 + mq.remove(context)
257 + save_tmp_chat(context)
258 + reason = "integration_commands.new" if new_chat else "integration_commands.clear"
259 + mark_dirty_for_context(context.id, reason=reason)
260 + return "Started a fresh chat." if new_chat else "Chat cleared."
261 +
262 +
263 +def _handle_steer(context: "AgentContext", args: str) -> str:
264 + message = args.strip()
265 + if not message:
266 + return "Usage: /steer <message>"
267 + from agent import UserMessage
268 +
269 + context.communicate(UserMessage(message=message))
270 + if context.is_running():
271 + return "Steering message sent to the active run."
272 + return "Message sent."
273 +
274 +
275 +def _handle_pause(context: "AgentContext") -> str:
276 + if not context.is_running():
277 + return "No active run is currently running."
278 + context.paused = True
279 + return "Agent paused."
280 +
281 +
282 +def _handle_resume(context: "AgentContext") -> str:
283 + context.paused = False
284 + return "Agent resumed."
285 +
286 +
287 +def _handle_nudge(context: "AgentContext") -> str:
288 + context.nudge()
289 + return "Agent nudged."
290 +
291 +
292 +def _handle_toggle(context: "AgentContext", args: str, key: str, label: str) -> str:
293 + value = _parse_toggle(args)
294 + current = _get_toggle(context, key)
295 + if value is None:
296 + state = "on" if current else "off"
297 + return f"{label}: {state}. Use /{key.split('_')[1]} on or /{key.split('_')[1]} off."
298 + context.set_data(key, value)
299 + save_tmp_chat(context)
300 + mark_dirty_for_context(context.id, reason=f"integration_commands.{key}")
301 + return f"{label} {'enabled' if value else 'disabled'}."
302 +
303 +
304 def _handle_project(context: "AgentContext", args: str) -> str:
305 items = projects.get_active_projects_list() or []
306 current_name = context.get_data("project") or ""
@@ -115,7 +336,7 @@ def _handle_project(context: "AgentContext", args: str) -> str:
336 return f"Switched project to {match.get('title') or match['name']}."
337
338
118 -def _handle_config(context: "AgentContext", args: str) -> str:
339 +def _handle_model(context: "AgentContext", args: str) -> str:
340 allowed = model_config.is_chat_override_allowed(context.agent0)
341 presets = [preset for preset in model_config.get_presets() if preset.get("name")]
342 current_override = context.get_data("chat_model_override")
@@ -123,12 +344,12 @@ def _handle_config(context: "AgentContext", args: str) -> str:
344 if not args:
345 current_label = _describe_override(current_override)
346 available = ", ".join(preset["name"] for preset in presets) or "none"
126 - suffix = "Use /config <name> to switch, or /config default to clear it."
347 + suffix = "Use /model <name> to switch, or /model default to clear it."
348 if not allowed:
349 suffix = "Per-chat config switching is disabled in Model Configuration."
350 return (
130 - f"Current config: {current_label}\n"
131 - f"Available configs: {available}\n"
351 + f"Current model: {current_label}\n"
352 + f"Available presets: {available}\n"
353 f"{suffix}"
354 )
355
@@ -159,7 +380,49 @@ def _handle_config(context: "AgentContext", args: str) -> str:
380 context.set_data("chat_model_override", {"preset_name": preset_name})
381 save_tmp_chat(context)
382 mark_dirty_for_context(context.id, reason="integration_commands.config_set")
162 - return f"Switched config to {preset_name}."
383 + return f"Switched model preset to {preset_name}."
384 +
385 +
386 +def _handle_agent(context: "AgentContext", args: str) -> str:
387 + from agent import Agent
388 + from helpers import subagents
389 + from initialize import initialize_agent
390 +
391 + items = subagents.get_all_agents_list()
392 + current = getattr(context.agent0.config, "profile", "default")
393 + if not args:
394 + available = ", ".join(_format_agent_entry(item) for item in items) or "none"
395 + return (
396 + f"Current agent: {current}\n"
397 + f"Available agents: {available}\n"
398 + "Use /agent <profile> to switch after the current run finishes."
399 + )
400 +
401 + if context.is_running():
402 + return "Agent profile can be changed after the current run finishes."
403 +
404 + desired = _strip_quotes(args)
405 + match, ambiguous = _match_named_item(items, desired, keys=("key", "label"))
406 + if ambiguous:
407 + names = ", ".join(_format_agent_entry(item) for item in ambiguous)
408 + return f"Agent profile is ambiguous. Matches: {names}"
409 + if not match:
410 + available = ", ".join(_format_agent_entry(item) for item in items) or "none"
411 + return f"Agent profile '{desired}' was not found. Available agents: {available}"
412 +
413 + profile = str(match["key"])
414 + if profile == current:
415 + return f"Already using agent {match.get('label') or profile}."
416 +
417 + config = initialize_agent(override_settings={"agent_profile": profile})
418 + context.config = config
419 + agent = context.agent0
420 + while agent:
421 + agent.config = config
422 + agent = agent.get_data(Agent.DATA_NAME_SUBORDINATE)
423 + save_tmp_chat(context)
424 + mark_dirty_for_context(context.id, reason="integration_commands.agent_set")
425 + return f"Switched agent to {match.get('label') or profile}."
426
427
428 def _format_project_entry(item: dict) -> str:
@@ -170,6 +433,19 @@ def _format_project_entry(item: dict) -> str:
433 return name or title
434
435
436 +def _format_agent_entry(item: dict) -> str:
437 + key = str(item.get("key", "") or "").strip()
438 + label = str(item.get("label", "") or "").strip()
439 + if label and label.lower() != key.lower():
440 + return f"{label} ({key})"
441 + return key or label
442 +
443 +
444 +def _telegram_description(command: IntegrationCommandDef) -> str:
445 + description = command.description.strip()
446 + return description[:255] if len(description) > 255 else description
447 +
448 +
449 def _describe_project(items: list[dict], current_name: str) -> str:
450 if not current_name:
451 return "none"
@@ -201,6 +477,20 @@ def _normalize_lookup(value: str) -> str:
477 return lowered.strip()
478
479
480 +def _get_toggle(context: "AgentContext", key: str) -> bool:
481 + value = context.get_data(key)
482 + return True if value is None else bool(value)
483 +
484 +
485 +def _parse_toggle(args: str) -> bool | None:
486 + value = _normalize_lookup(args)
487 + if value in {"on", "enable", "enabled", "yes", "true", "1"}:
488 + return True
489 + if value in {"off", "disable", "disabled", "no", "false", "0"}:
490 + return False
491 + return None
492 +
493 +
494 def _match_named_item(
495 items: list[dict],
496 desired: str,
plugins/_telegram_integration/README.md
+20 -4
@@ -16,10 +16,16 @@ This plugin connects one or more Telegram bots to Agent Zero. Each bot runs inde
16 - Supports both long-polling and webhook delivery modes.
17 - **Per-user chat sessions**
18 - Each Telegram user gets a dedicated `AgentContext`, persisted across restarts via a JSON state file.
19 - - `/start` creates a context; `/clear` resets it.
19 + - `/start` creates a context; `/new` starts fresh; `/clear` resets the current context.
20 + - `/commands` shows the shared integration command menu (`/help` is an alias).
21 + - `/status` shows project, model, agent profile, and queue state.
22 - `/project <name>` switches the active project for the current chat.
21 - - `/config <preset>` switches the active model preset for the current chat.
22 - - `/send` or `/queue send` flushes the queued messages for the current chat.
23 + - `/model <preset>` switches the active model preset for the current chat (`/config` remains an alias).
24 + - `/agent <profile>` switches the active agent profile when the current run is idle.
25 + - `/model`, `/project`, and `/agent` show Telegram inline keyboard pickers when used without arguments.
26 + - `/stream` toggles live response streaming; `/tools` toggles tool progress messages.
27 + - `/send` or `/queue send` flushes queued messages for the current chat.
28 + - `/steer <message>` sends an intervention to the active run.
29 - **Group support**
30 - Three modes: `mention` (respond only when @mentioned or replied to), `all` (respond to every message), `off` (private only).
31 - Optional welcome message for new members.
@@ -27,6 +33,8 @@ This plugin connects one or more Telegram bots to Agent Zero. Each bot runs inde
33 - Extracts text, captions, locations, contacts, stickers, and attachment indicators.
34 - Downloads photos, documents, audio, voice, and video into `usr/uploads/` with configurable auto-cleanup.
35 - **Reply delivery**
36 + - Streams tool progress and response text through real Telegram messages updated with `editMessageText`.
37 + - Tool progress and the AI response are separate messages; only the AI response replies to the user message.
38 - `tool_execute_after` intercepts the `response` tool — sends inline progress for `break_loop=false`.
39 - `process_chain_end` auto-sends the final response, with retry logic on failure.
40 - **Formatting**
@@ -36,8 +44,11 @@ This plugin connects one or more Telegram bots to Agent Zero. Each bot runs inde
44 - Agent can attach a `keyboard` array to the `response` tool; button presses feed back as user messages.
45 - **Typing indicator**
46 - Persistent "typing…" action while the agent is processing, cancelled on reply.
47 +- **Busy-run queue**
48 + - Normal user messages received while a run is active are queued automatically.
49 + - `/steer <message>` is still delivered immediately as an intervention.
50 - **Notifications**
40 - - Optional WebUI notifications for incoming Telegram messages and `/clear` events.
51 + - Optional WebUI notifications for incoming Telegram messages.
52 - **Access control**
53 - Per-bot allow-list by Telegram user ID or @username. Empty list = open to all.
54 - **Project binding**
@@ -50,8 +61,13 @@ This plugin connects one or more Telegram bots to Agent Zero. Each bot runs inde
61 - `helpers/handler.py` — Central message routing, context lifecycle, user auth, attachment download, reply sending, typing indicator.
62 - `helpers/bot_manager.py` — Bot creation, polling/webhook lifecycle, bot registry.
63 - `helpers/telegram_client.py` — Low-level Telegram API wrapper: send text/file/photo, Markdown→HTML converter, keyboard builder, message splitting.
64 + - `helpers/command_ui.py` — Telegram inline keyboard command pickers and callback handling.
65 + - `helpers/draft_stream.py` — Editable Telegram message streaming for tool progress and response previews.
66 - **Extensions**
67 - `extensions/python/job_loop/_10_telegram_bot.py` — Bot lifecycle manager, starts/stops bots on each tick.
68 + - `extensions/python/message_loop_start/_45_telegram_draft_start.py` — Initializes Telegram response streaming.
69 + - `extensions/python/tool_execute_before/_45_telegram_draft_tool.py` — Streams tool-start progress.
70 + - `extensions/python/response_stream/_45_telegram_draft_response.py` — Streams final response previews.
71 - `extensions/python/system_prompt/_20_telegram_context.py` — Injects Telegram-specific system prompt.
72 - `extensions/python/tool_execute_after/_50_telegram_response.py` — Intercepts `response` tool for inline delivery.
73 - `extensions/python/process_chain_end/_55_telegram_reply.py` — Auto-sends final reply with retry.
plugins/_telegram_integration/extensions/python/job_loop/_10_telegram_bot.py
+2 -3
@@ -31,13 +31,13 @@ class TelegramBotManager(Extension):
31 get_all_bots,
32 create_bot,
33 cache_bot_info,
34 + register_bot_commands,
35 start_polling,
36 setup_webhook,
37 stop_bot,
38 )
39 from plugins._telegram_integration.helpers.handler import (
40 handle_start,
40 - handle_clear,
41 handle_message,
42 handle_callback_query,
43 handle_new_members,
@@ -73,7 +73,6 @@ class TelegramBotManager(Extension):
73 try:
74 # Create handler closures that capture bot_name and config
75 _on_start = partial(_make_handler(handle_start), bot_name=name, bot_cfg=bot_cfg)
76 - _on_clear = partial(_make_handler(handle_clear), bot_name=name, bot_cfg=bot_cfg)
76 _on_message = partial(_make_handler(handle_message), bot_name=name, bot_cfg=bot_cfg)
77 _on_callback = partial(_make_handler(handle_callback_query), bot_name=name, bot_cfg=bot_cfg)
78 _on_new_members = partial(_make_handler(handle_new_members), bot_name=name, bot_cfg=bot_cfg)
@@ -83,7 +82,6 @@ class TelegramBotManager(Extension):
82 token=bot_cfg["token"],
83 on_message=_on_message,
84 on_command_start=_on_start,
86 - on_command_clear=_on_clear,
85 on_command_control=_on_message,
86 on_callback_query=_on_callback,
87 on_new_members=_on_new_members,
@@ -91,6 +89,7 @@ class TelegramBotManager(Extension):
89 )
90
91 await cache_bot_info(instance)
92 + await register_bot_commands(instance)
93
94 mode = bot_cfg.get("mode", "polling")
95 if mode == "webhook":
plugins/_telegram_integration/extensions/python/message_loop_start/_45_telegram_draft_start.py new
+17
@@ -0,0 +1,17 @@
1 +from agent import LoopData
2 +from helpers.extension import Extension
3 +from plugins._telegram_integration.helpers.constants import CTX_TG_BOT
4 +
5 +
6 +class TelegramDraftStart(Extension):
7 +
8 + async def execute(self, loop_data: LoopData = LoopData(), **kwargs):
9 + if not self.agent or self.agent.number != 0:
10 + return
11 + context = self.agent.context
12 + if not context.data.get(CTX_TG_BOT):
13 + return
14 +
15 + from plugins._telegram_integration.helpers import draft_stream
16 +
17 + await draft_stream.start(context)
plugins/_telegram_integration/extensions/python/process_chain_end/_55_telegram_reply.py
+3
@@ -41,6 +41,9 @@ class TelegramAutoReply(Extension):
41 typing_stop = context.data.pop(CTX_TG_TYPING_STOP, None)
42 if typing_stop:
43 typing_stop.set()
44 + from plugins._telegram_integration.helpers import draft_stream
45 +
46 + draft_stream.clear(context)
47 context.data.pop(CTX_TG_REPLY_TO, None)
48
49 async def _send_reply(
plugins/_telegram_integration/extensions/python/response_stream/_45_telegram_draft_response.py new
+30
@@ -0,0 +1,30 @@
1 +from agent import LoopData
2 +from helpers.extension import Extension
3 +from plugins._telegram_integration.helpers.constants import CTX_TG_BOT
4 +
5 +
6 +class TelegramDraftResponse(Extension):
7 +
8 + async def execute(
9 + self,
10 + loop_data: LoopData = LoopData(),
11 + parsed: dict | None = None,
12 + **kwargs,
13 + ):
14 + if not self.agent or self.agent.number != 0:
15 + return
16 + context = self.agent.context
17 + if not context.data.get(CTX_TG_BOT):
18 + return
19 +
20 + parsed = parsed or {}
21 + if parsed.get("tool_name") != "response":
22 + return
23 + tool_args = parsed.get("tool_args") or {}
24 + text = tool_args.get("text") or tool_args.get("message") or ""
25 + if not text:
26 + return
27 +
28 + from plugins._telegram_integration.helpers import draft_stream
29 +
30 + await draft_stream.update_response(context, str(text))
plugins/_telegram_integration/extensions/python/system_prompt/_20_telegram_context.py
+17
@@ -1,5 +1,6 @@
1 from helpers.extension import Extension
2 from agent import LoopData
3 +from helpers import integration_commands
4 from plugins._telegram_integration.helpers.constants import CTX_TG_BOT, CTX_TG_BOT_CFG
5
6
@@ -18,6 +19,7 @@ class TelegramContextPrompt(Extension):
19 system_prompt.append(
20 self.agent.read_prompt("fw.telegram.system_context_reply.md")
21 )
22 + system_prompt.append(_telegram_commands_prompt())
23
24 # Inject per-bot agent instructions (once in system prompt)
25 bot_cfg = self.agent.context.data.get(CTX_TG_BOT_CFG, {})
@@ -29,3 +31,18 @@ class TelegramContextPrompt(Extension):
31 instructions=instructions,
32 )
33 )
34 +
35 +
36 +def _telegram_commands_prompt() -> str:
37 + lines = [
38 + "Telegram slash commands are handled by the integration before you see the message.",
39 + "Do not invent command help, unknown-command replies, or pretend to execute slash commands.",
40 + "If the user asks what commands exist, refer them to /commands.",
41 + "Current integration commands:",
42 + ]
43 + for name in integration_commands.command_names(include_aliases=False):
44 + definition = integration_commands.resolve_command(name)
45 + if definition:
46 + args = f" {definition.args_hint}" if definition.args_hint else ""
47 + lines.append(f"- /{definition.name}{args}: {definition.description}")
48 + return "\n".join(lines)
plugins/_telegram_integration/extensions/python/tool_execute_after/_50_telegram_response.py
+6 -2
@@ -13,14 +13,18 @@ class TelegramResponseIntercept(Extension):
13 async def execute(
14 self, tool_name: str = "", response: Response | None = None, **kwargs,
15 ):
16 - if tool_name != "response":
17 - return
16 if not self.agent:
17 return
18 context = self.agent.context
19 if not context.data.get(CTX_TG_BOT):
20 return
21
22 + from plugins._telegram_integration.helpers import draft_stream
23 +
24 + if tool_name != "response":
25 + await draft_stream.add_tool_done(context, tool_name, ok=response is not None)
26 + return
27 +
28 tool = self.agent.loop_data.current_tool
29 if not tool:
30 return
plugins/_telegram_integration/extensions/python/tool_execute_before/_45_telegram_draft_tool.py new
+18
@@ -0,0 +1,18 @@
1 +from helpers.extension import Extension
2 +from plugins._telegram_integration.helpers.constants import CTX_TG_BOT
3 +
4 +
5 +class TelegramDraftToolStart(Extension):
6 +
7 + async def execute(self, tool_name: str = "", tool_args: dict | None = None, **kwargs):
8 + if not self.agent or self.agent.number != 0:
9 + return
10 + if (tool_name or "").strip().lower() == "response":
11 + return
12 + context = self.agent.context
13 + if not context.data.get(CTX_TG_BOT):
14 + return
15 +
16 + from plugins._telegram_integration.helpers import draft_stream
17 +
18 + await draft_stream.add_tool_start(context, tool_name, tool_args or {})
plugins/_telegram_integration/helpers/bot_manager.py
+18 -4
@@ -6,7 +6,8 @@ from aiogram import Bot, Dispatcher, Router, F
6 from aiogram.client.default import DefaultBotProperties
7 from aiogram.enums import ParseMode, ChatType, ContentType
8 from aiogram.filters import Command, CommandStart
9 -from aiogram.types import Message
9 +from aiogram.types import BotCommand, Message
10 +from helpers import integration_commands
11
12 from helpers.errors import format_error
13 from helpers.print_style import PrintStyle
@@ -44,7 +45,6 @@ def create_bot(
45 token: str,
46 on_message: Callable[..., Awaitable],
47 on_command_start: Callable[..., Awaitable],
47 - on_command_clear: Callable[..., Awaitable],
48 on_command_control: Callable[..., Awaitable] | None = None,
49 on_callback_query: Callable[..., Awaitable] | None = None,
50 on_new_members: Callable[..., Awaitable] | None = None,
@@ -56,11 +56,10 @@ def create_bot(
56
57 # Register command handlers
58 router.message.register(on_command_start, CommandStart())
59 - router.message.register(on_command_clear, Command("clear"))
59 if on_command_control:
60 router.message.register(
61 on_command_control,
63 - Command(commands=["project", "config", "preset", "queue", "send"]),
62 + Command(commands=integration_commands.command_names()),
63 )
64
65 if on_callback_query:
@@ -93,6 +92,21 @@ def create_bot(
92 return instance
93
94
95 +async def register_bot_commands(instance: BotInstance) -> None:
96 + """Register Telegram's native / command menu from the shared integration registry."""
97 + commands = [
98 + BotCommand(command=name, description=description)
99 + for name, description in integration_commands.telegram_menu_commands()
100 + ]
101 + if not commands:
102 + return
103 + try:
104 + await instance.bot.set_my_commands(commands)
105 + PrintStyle.info(f"Telegram ({instance.name}): registered {len(commands)} bot commands")
106 + except Exception as e:
107 + PrintStyle.error(f"Telegram ({instance.name}): failed to register bot commands: {format_error(e)}")
108 +
109 +
110 async def cache_bot_info(instance: BotInstance):
111 """Fetch and cache bot info. Call after create_bot."""
112 if not instance.bot_info:
plugins/_telegram_integration/helpers/command_ui.py new
+401
@@ -0,0 +1,401 @@
1 +from __future__ import annotations
2 +
3 +from dataclasses import dataclass
4 +
5 +from agent import AgentContext
6 +from helpers import projects, subagents
7 +from helpers import integration_commands
8 +from helpers.persist_chat import save_tmp_chat
9 +from helpers.state_monitor_integration import mark_dirty_for_context
10 +from plugins._model_config.helpers import model_config
11 +from plugins._telegram_integration.helpers import telegram_client as tc
12 +from plugins._telegram_integration.helpers.constants import (
13 + CTX_TG_STREAM_ENABLED,
14 + CTX_TG_TOOLS_ENABLED,
15 +)
16 +
17 +PAGE_SIZE = 8
18 +CALLBACK_PREFIX = "tg"
19 +
20 +
21 +@dataclass(frozen=True)
22 +class PickerItem:
23 + key: str
24 + label: str
25 +
26 +
27 +async def handle_command(
28 + context: AgentContext,
29 + token: str,
30 + chat_id: int,
31 + reply_to_message_id: int | None,
32 + text: str,
33 +) -> bool:
34 + parsed = integration_commands.parse_command(text or "")
35 + if not parsed:
36 + return False
37 + command, args = parsed
38 + if command in {"/model", "/config", "/preset"} and not args:
39 + await send_model_picker(context, token, chat_id, reply_to_message_id, 0)
40 + return True
41 + if command == "/project" and not args:
42 + await send_project_picker(context, token, chat_id, reply_to_message_id, 0)
43 + return True
44 + if command in {"/agent", "/profile"} and not args:
45 + await send_agent_picker(context, token, chat_id, reply_to_message_id, 0)
46 + return True
47 + if command == "/stream":
48 + await send_toggle_picker(
49 + context,
50 + token,
51 + chat_id,
52 + reply_to_message_id,
53 + CTX_TG_STREAM_ENABLED,
54 + "Response streaming",
55 + args,
56 + )
57 + return True
58 + if command == "/tools":
59 + await send_toggle_picker(
60 + context,
61 + token,
62 + chat_id,
63 + reply_to_message_id,
64 + CTX_TG_TOOLS_ENABLED,
65 + "Tool progress",
66 + args,
67 + )
68 + return True
69 + return False
70 +
71 +
72 +async def handle_callback(
73 + context: AgentContext,
74 + token: str,
75 + chat_id: int,
76 + message_id: int,
77 + data: str,
78 +) -> bool:
79 + parts = (data or "").split(":")
80 + if len(parts) < 3 or parts[0] != CALLBACK_PREFIX:
81 + return False
82 + kind, action = parts[1], parts[2]
83 + value = parts[3] if len(parts) > 3 else ""
84 + if action == "noop":
85 + return True
86 + if action == "page":
87 + page = _safe_int(value)
88 + if kind == "model":
89 + await edit_model_picker(context, token, chat_id, message_id, page)
90 + elif kind == "project":
91 + await edit_project_picker(context, token, chat_id, message_id, page)
92 + elif kind == "agent":
93 + await edit_agent_picker(context, token, chat_id, message_id, page)
94 + return True
95 + if kind == "model" and action in {"set", "clear"}:
96 + await _select_model(context, _safe_int(value), clear=(action == "clear"))
97 + await edit_model_picker(context, token, chat_id, message_id, 0, selected=True)
98 + return True
99 + if kind == "project" and action in {"set", "clear"}:
100 + await _select_project(context, _safe_int(value), clear=(action == "clear"))
101 + await edit_project_picker(context, token, chat_id, message_id, 0, selected=True)
102 + return True
103 + if kind == "agent" and action == "set":
104 + await _select_agent(context, _safe_int(value))
105 + await edit_agent_picker(context, token, chat_id, message_id, 0, selected=True)
106 + return True
107 + if kind in {"stream", "tools"} and action in {"on", "off"}:
108 + key = CTX_TG_STREAM_ENABLED if kind == "stream" else CTX_TG_TOOLS_ENABLED
109 + label = "Response streaming" if kind == "stream" else "Tool progress"
110 + context.set_data(key, action == "on")
111 + save_tmp_chat(context)
112 + mark_dirty_for_context(context.id, reason=f"telegram.{kind}_toggle")
113 + await edit_toggle_picker(context, token, chat_id, message_id, key, label)
114 + return True
115 + return True
116 +
117 +
118 +async def send_model_picker(
119 + context: AgentContext,
120 + token: str,
121 + chat_id: int,
122 + reply_to_message_id: int | None,
123 + page: int,
124 +) -> None:
125 + text, markup = _model_view(context, page)
126 + await tc.raw_send_text(token, chat_id, text, reply_to_message_id, "HTML", markup)
127 +
128 +
129 +async def edit_model_picker(
130 + context: AgentContext,
131 + token: str,
132 + chat_id: int,
133 + message_id: int,
134 + page: int,
135 + *,
136 + selected: bool = False,
137 +) -> None:
138 + text, markup = _model_view(context, page, selected=selected)
139 + await tc.raw_edit_text(token, chat_id, message_id, text, "HTML", markup)
140 +
141 +
142 +async def send_project_picker(
143 + context: AgentContext,
144 + token: str,
145 + chat_id: int,
146 + reply_to_message_id: int | None,
147 + page: int,
148 +) -> None:
149 + text, markup = _project_view(context, page)
150 + await tc.raw_send_text(token, chat_id, text, reply_to_message_id, "HTML", markup)
151 +
152 +
153 +async def edit_project_picker(
154 + context: AgentContext,
155 + token: str,
156 + chat_id: int,
157 + message_id: int,
158 + page: int,
159 + *,
160 + selected: bool = False,
161 +) -> None:
162 + text, markup = _project_view(context, page, selected=selected)
163 + await tc.raw_edit_text(token, chat_id, message_id, text, "HTML", markup)
164 +
165 +
166 +async def send_agent_picker(
167 + context: AgentContext,
168 + token: str,
169 + chat_id: int,
170 + reply_to_message_id: int | None,
171 + page: int,
172 +) -> None:
173 + text, markup = _agent_view(context, page)
174 + await tc.raw_send_text(token, chat_id, text, reply_to_message_id, "HTML", markup)
175 +
176 +
177 +async def edit_agent_picker(
178 + context: AgentContext,
179 + token: str,
180 + chat_id: int,
181 + message_id: int,
182 + page: int,
183 + *,
184 + selected: bool = False,
185 +) -> None:
186 + text, markup = _agent_view(context, page, selected=selected)
187 + await tc.raw_edit_text(token, chat_id, message_id, text, "HTML", markup)
188 +
189 +
190 +async def send_toggle_picker(
191 + context: AgentContext,
192 + token: str,
193 + chat_id: int,
194 + reply_to_message_id: int | None,
195 + key: str,
196 + label: str,
197 + args: str = "",
198 +) -> None:
199 + desired = _parse_toggle(args)
200 + if desired is not None:
201 + context.set_data(key, desired)
202 + save_tmp_chat(context)
203 + mark_dirty_for_context(context.id, reason=f"telegram.{key}")
204 + text, markup = _toggle_view(context, key, label)
205 + await tc.raw_send_text(token, chat_id, text, reply_to_message_id, "HTML", markup)
206 +
207 +
208 +async def edit_toggle_picker(
209 + context: AgentContext,
210 + token: str,
211 + chat_id: int,
212 + message_id: int,
213 + key: str,
214 + label: str,
215 +) -> None:
216 + text, markup = _toggle_view(context, key, label)
217 + await tc.raw_edit_text(token, chat_id, message_id, text, "HTML", markup)
218 +
219 +
220 +def _model_view(
221 + context: AgentContext,
222 + page: int,
223 + *,
224 + selected: bool = False,
225 +) -> tuple[str, dict | None]:
226 + presets = [
227 + PickerItem(str(preset.get("name", "")), str(preset.get("name", "")))
228 + for preset in model_config.get_presets()
229 + if isinstance(preset, dict) and preset.get("name")
230 + ]
231 + current = context.get_data("chat_model_override")
232 + current_name = current.get("preset_name") if isinstance(current, dict) else ""
233 + status = f"Current model: <b>{_html(current_name or 'Default')}</b>"
234 + if not model_config.is_chat_override_allowed(context.agent0):
235 + return status + "\nPer-chat model switching is disabled.", None
236 + if selected:
237 + status = "Model updated.\n" + status
238 + rows = _paged_buttons("model", presets, page, current_name)
239 + rows.append([{"text": "Default", "callback_data": "tg:model:clear"}])
240 + return status, {"inline_keyboard": rows}
241 +
242 +
243 +def _project_view(
244 + context: AgentContext,
245 + page: int,
246 + *,
247 + selected: bool = False,
248 +) -> tuple[str, dict | None]:
249 + items = [
250 + PickerItem(str(item.get("name", "")), str(item.get("title") or item.get("name") or ""))
251 + for item in projects.get_active_projects_list() or []
252 + if item.get("name")
253 + ]
254 + current = context.get_data("project") or ""
255 + status = f"Current project: <b>{_html(_label_for(items, current) or 'none')}</b>"
256 + if selected:
257 + status = "Project updated.\n" + status
258 + rows = _paged_buttons("project", items, page, current)
259 + rows.append([{"text": "No project", "callback_data": "tg:project:clear"}])
260 + return status, {"inline_keyboard": rows}
261 +
262 +
263 +def _agent_view(
264 + context: AgentContext,
265 + page: int,
266 + *,
267 + selected: bool = False,
268 +) -> tuple[str, dict | None]:
269 + items = [
270 + PickerItem(str(item.get("key", "")), str(item.get("label") or item.get("key") or ""))
271 + for item in subagents.get_all_agents_list()
272 + if item.get("key")
273 + ]
274 + current = getattr(context.agent0.config, "profile", "") or "agent0"
275 + status = f"Current agent: <b>{_html(_label_for(items, current) or current)}</b>"
276 + if context.is_running():
277 + status += "\nAgent profile can be changed after the current run finishes."
278 + elif selected:
279 + status = "Agent updated.\n" + status
280 + rows = _paged_buttons("agent", items, page, current, disabled=context.is_running())
281 + if not rows:
282 + return status + "\nNo agent profiles were found.", None
283 + return status, {"inline_keyboard": rows}
284 +
285 +
286 +def _toggle_view(context: AgentContext, key: str, label: str) -> tuple[str, dict]:
287 + enabled = _toggle_enabled(context, key)
288 + kind = "stream" if key == CTX_TG_STREAM_ENABLED else "tools"
289 + state = "enabled" if enabled else "disabled"
290 + text = f"{_html(label)}: <b>{state}</b>"
291 + rows = [[
292 + {"text": ("On" if enabled else "Turn on"), "callback_data": f"tg:{kind}:on"},
293 + {"text": ("Off" if not enabled else "Turn off"), "callback_data": f"tg:{kind}:off"},
294 + ]]
295 + return text, {"inline_keyboard": rows}
296 +
297 +
298 +async def _select_model(context: AgentContext, index: int, *, clear: bool = False) -> None:
299 + if not model_config.is_chat_override_allowed(context.agent0):
300 + return
301 + if clear:
302 + context.set_data("chat_model_override", None)
303 + else:
304 + presets = [preset for preset in model_config.get_presets() if preset.get("name")]
305 + if index < 0 or index >= len(presets):
306 + return
307 + context.set_data("chat_model_override", {"preset_name": presets[index]["name"]})
308 + save_tmp_chat(context)
309 + mark_dirty_for_context(context.id, reason="telegram.model_select")
310 +
311 +
312 +async def _select_project(context: AgentContext, index: int, *, clear: bool = False) -> None:
313 + if clear:
314 + projects.deactivate_project(context.id)
315 + return
316 + items = [item for item in projects.get_active_projects_list() or [] if item.get("name")]
317 + if index < 0 or index >= len(items):
318 + return
319 + projects.activate_project(context.id, str(items[index]["name"]))
320 +
321 +
322 +async def _select_agent(context: AgentContext, index: int) -> None:
323 + if context.is_running():
324 + return
325 + from agent import Agent
326 + from initialize import initialize_agent
327 +
328 + items = [item for item in subagents.get_all_agents_list() if item.get("key")]
329 + if index < 0 or index >= len(items):
330 + return
331 + profile = str(items[index]["key"])
332 + config = initialize_agent(override_settings={"agent_profile": profile})
333 + context.config = config
334 + agent = context.agent0
335 + while agent:
336 + agent.config = config
337 + agent = agent.get_data(Agent.DATA_NAME_SUBORDINATE)
338 + save_tmp_chat(context)
339 + mark_dirty_for_context(context.id, reason="telegram.agent_select")
340 +
341 +
342 +def _paged_buttons(
343 + kind: str,
344 + items: list[PickerItem],
345 + page: int,
346 + current: str,
347 + *,
348 + disabled: bool = False,
349 +) -> list[list[dict[str, str]]]:
350 + page = max(0, page)
351 + total_pages = max(1, (len(items) + PAGE_SIZE - 1) // PAGE_SIZE)
352 + page = min(page, total_pages - 1)
353 + start = page * PAGE_SIZE
354 + rows: list[list[dict[str, str]]] = []
355 + for offset, item in enumerate(items[start:start + PAGE_SIZE], start=start):
356 + marker = "• " if item.key == current else ""
357 + action = "noop" if disabled else "set"
358 + rows.append([{
359 + "text": f"{marker}{item.label}"[:64],
360 + "callback_data": f"tg:{kind}:{action}:{offset}",
361 + }])
362 + nav: list[dict[str, str]] = []
363 + if page > 0:
364 + nav.append({"text": "Prev", "callback_data": f"tg:{kind}:page:{page - 1}"})
365 + if page < total_pages - 1:
366 + nav.append({"text": "Next", "callback_data": f"tg:{kind}:page:{page + 1}"})
367 + if nav:
368 + rows.append(nav)
369 + return rows
370 +
371 +
372 +def _toggle_enabled(context: AgentContext, key: str) -> bool:
373 + value = context.get_data(key)
374 + return True if value is None else bool(value)
375 +
376 +
377 +def _parse_toggle(args: str) -> bool | None:
378 + value = (args or "").strip().lower()
379 + if value in {"on", "enable", "enabled", "yes", "true", "1"}:
380 + return True
381 + if value in {"off", "disable", "disabled", "no", "false", "0"}:
382 + return False
383 + return None
384 +
385 +
386 +def _safe_int(value: str) -> int:
387 + try:
388 + return int(value)
389 + except (TypeError, ValueError):
390 + return 0
391 +
392 +
393 +def _label_for(items: list[PickerItem], key: str) -> str:
394 + for item in items:
395 + if item.key == key:
396 + return item.label
397 + return key
398 +
399 +
400 +def _html(value: str) -> str:
401 + return str(value).replace("&", "&amp;").replace("<", "&lt;").replace(">", "&gt;")
plugins/_telegram_integration/helpers/constants.py
+8
@@ -6,11 +6,19 @@ STATE_FILE = "usr/plugins/_telegram_integration/state.json"
6 CTX_TG_BOT = "telegram_bot"
7 CTX_TG_BOT_CFG = "telegram_bot_cfg"
8 CTX_TG_CHAT_ID = "telegram_chat_id"
9 +CTX_TG_CHAT_TYPE = "telegram_chat_type"
10 CTX_TG_USER_ID = "telegram_user_id"
11 CTX_TG_USERNAME = "telegram_username"
12 CTX_TG_TYPING_STOP = "_telegram_typing_stop"
13 CTX_TG_REPLY_TO = "_telegram_reply_to_message_id"
14 +CTX_TG_STREAM_ENABLED = "telegram_stream_enabled"
15 +CTX_TG_TOOLS_ENABLED = "telegram_tools_enabled"
16
17 # Transient
18 CTX_TG_ATTACHMENTS = "_telegram_response_attachments"
19 CTX_TG_KEYBOARD = "_telegram_response_keyboard"
20 +CTX_TG_PROGRESS_MESSAGE_ID = "_telegram_progress_message_id"
21 +CTX_TG_PROGRESS_LINES = "_telegram_progress_lines"
22 +CTX_TG_RESPONSE_MESSAGE_ID = "_telegram_response_message_id"
23 +CTX_TG_RESPONSE_TEXT = "_telegram_response_text"
24 +CTX_TG_RESPONSE_LAST_UPDATE = "_telegram_response_last_update"
plugins/_telegram_integration/helpers/draft_stream.py new
+361
@@ -0,0 +1,361 @@
1 +from __future__ import annotations
2 +
3 +import re
4 +import time
5 +
6 +from agent import AgentContext
7 +from helpers.print_style import PrintStyle
8 +from plugins._telegram_integration.helpers import telegram_client as tc
9 +from plugins._telegram_integration.helpers.bot_manager import get_bot
10 +from plugins._telegram_integration.helpers.constants import (
11 + CTX_TG_BOT,
12 + CTX_TG_CHAT_ID,
13 + CTX_TG_PROGRESS_LINES,
14 + CTX_TG_PROGRESS_MESSAGE_ID,
15 + CTX_TG_REPLY_TO,
16 + CTX_TG_RESPONSE_LAST_UPDATE,
17 + CTX_TG_RESPONSE_MESSAGE_ID,
18 + CTX_TG_RESPONSE_TEXT,
19 + CTX_TG_STREAM_ENABLED,
20 + CTX_TG_TOOLS_ENABLED,
21 +)
22 +
23 +MAX_STREAM_CHARS: int = 3900
24 +MIN_RESPONSE_UPDATE_SECONDS: float = 1.0
25 +MAX_PROGRESS_LINES: int = 12
26 +
27 +TOOL_EMOJIS: dict[str, str] = {
28 + "browser": "🌐",
29 + "code": "⌨️",
30 + "code_execution_tool": "⌨️",
31 + "duckduckgo_search": "🔎",
32 + "file": "📄",
33 + "knowledge_tool": "📚",
34 + "memory": "🧠",
35 + "read_file": "📖",
36 + "search": "🔎",
37 + "search_engine": "🔎",
38 + "search_files": "🔎",
39 + "skill": "📚",
40 + "skill_view": "📚",
41 +}
42 +
43 +
44 +async def start(context: AgentContext) -> None:
45 + # Do not pre-create a placeholder Telegram message. Wait until we have
46 + # real assistant text so the stream starts with meaningful content.
47 + if not _stream_enabled(context):
48 + return
49 +
50 +
51 +async def add_tool_start(
52 + context: AgentContext,
53 + tool_name: str,
54 + args: dict | None = None,
55 +) -> None:
56 + if not _tools_enabled(context):
57 + return
58 + label = _tool_progress_label(tool_name, args or {})
59 + _append_progress_line(context, f"{_tool_emoji(tool_name)} {label}")
60 + await _send_progress(context)
61 +
62 +
63 +async def add_tool_done(context: AgentContext, tool_name: str, ok: bool = True) -> None:
64 + if not _tools_enabled(context):
65 + return
66 + if ok:
67 + return
68 + _mark_tool_failed(context, tool_name)
69 + await _send_progress(context)
70 +
71 +
72 +async def update_response(context: AgentContext, response_text: str) -> None:
73 + if not _stream_enabled(context):
74 + return
75 + cleaned = _visible_response_text(response_text)
76 + if not cleaned:
77 + return
78 + context.data[CTX_TG_RESPONSE_TEXT] = response_text or ""
79 + now = time.time()
80 + last = float(context.data.get(CTX_TG_RESPONSE_LAST_UPDATE) or 0.0)
81 + if now - last < MIN_RESPONSE_UPDATE_SECONDS:
82 + return
83 + await _update_response_message(context, response_text or "")
84 + context.data[CTX_TG_RESPONSE_LAST_UPDATE] = now
85 +
86 +
87 +async def finalize_response(
88 + context: AgentContext,
89 + response_text: str,
90 + keyboard: list[list[dict]] | None = None,
91 +) -> bool:
92 + message_id = context.data.get(CTX_TG_RESPONSE_MESSAGE_ID)
93 + if not message_id:
94 + return False
95 + ok = await _update_response_message(
96 + context,
97 + response_text or context.data.get(CTX_TG_RESPONSE_TEXT) or "",
98 + keyboard=keyboard,
99 + force=True,
100 + )
101 + if ok:
102 + context.data.pop(CTX_TG_RESPONSE_MESSAGE_ID, None)
103 + context.data.pop(CTX_TG_RESPONSE_TEXT, None)
104 + context.data.pop(CTX_TG_RESPONSE_LAST_UPDATE, None)
105 + return ok
106 +
107 +
108 +def clear(context: AgentContext) -> None:
109 + for key in (
110 + CTX_TG_PROGRESS_LINES,
111 + CTX_TG_PROGRESS_MESSAGE_ID,
112 + CTX_TG_RESPONSE_MESSAGE_ID,
113 + CTX_TG_RESPONSE_TEXT,
114 + CTX_TG_RESPONSE_LAST_UPDATE,
115 + ):
116 + context.data.pop(key, None)
117 +
118 +
119 +def _stream_enabled(context: AgentContext) -> bool:
120 + value = context.get_data(CTX_TG_STREAM_ENABLED)
121 + return True if value is None else bool(value)
122 +
123 +
124 +def _tools_enabled(context: AgentContext) -> bool:
125 + value = context.get_data(CTX_TG_TOOLS_ENABLED)
126 + return True if value is None else bool(value)
127 +
128 +
129 +async def _send_progress(context: AgentContext) -> None:
130 + bot = _bot_instance(context)
131 + chat_id = context.data.get(CTX_TG_CHAT_ID)
132 + if not bot or not chat_id:
133 + return
134 + text = "\n".join(context.data.get(CTX_TG_PROGRESS_LINES) or [])
135 + if not text:
136 + return
137 + message_id = context.data.get(CTX_TG_PROGRESS_MESSAGE_ID)
138 + try:
139 + if message_id:
140 + await tc.raw_edit_text(bot.bot.token, int(chat_id), int(message_id), text, parse_mode=None)
141 + return
142 + sent_id = await tc.raw_send_text(
143 + bot.bot.token,
144 + int(chat_id),
145 + text,
146 + parse_mode=None,
147 + )
148 + if sent_id:
149 + context.data[CTX_TG_PROGRESS_MESSAGE_ID] = sent_id
150 + except Exception as e:
151 + PrintStyle.debug(f"Telegram progress update failed: {e}")
152 +
153 +
154 +async def _ensure_response_message(context: AgentContext, text: str) -> int | None:
155 + message_id = context.data.get(CTX_TG_RESPONSE_MESSAGE_ID)
156 + if message_id:
157 + return int(message_id)
158 + html = _format_response(text)
159 + if not html:
160 + return None
161 + bot = _bot_instance(context)
162 + chat_id = context.data.get(CTX_TG_CHAT_ID)
163 + if not bot or not chat_id:
164 + return None
165 + sent_id = await tc.raw_send_text(
166 + bot.bot.token,
167 + int(chat_id),
168 + html,
169 + reply_to_message_id=_reply_to(context),
170 + parse_mode="HTML",
171 + )
172 + if sent_id:
173 + context.data[CTX_TG_RESPONSE_MESSAGE_ID] = sent_id
174 + context.data[CTX_TG_RESPONSE_LAST_UPDATE] = time.time()
175 + return sent_id
176 +
177 +
178 +async def _update_response_message(
179 + context: AgentContext,
180 + text: str,
181 + *,
182 + keyboard: list[list[dict]] | None = None,
183 + force: bool = False,
184 +) -> bool:
185 + message_id = await _ensure_response_message(context, text)
186 + bot = _bot_instance(context)
187 + chat_id = context.data.get(CTX_TG_CHAT_ID)
188 + if not message_id or not bot or not chat_id:
189 + return False
190 + markup = _keyboard_markup(keyboard)
191 + html = _format_response(text)
192 + try:
193 + ok = await tc.raw_edit_text(
194 + bot.bot.token,
195 + int(chat_id),
196 + int(message_id),
197 + html,
198 + parse_mode="HTML",
199 + reply_markup=markup,
200 + )
201 + if ok or force:
202 + return ok
203 + return False
204 + except Exception as e:
205 + PrintStyle.debug(f"Telegram response update failed: {e}")
206 + return False
207 +
208 +
209 +def _append_progress_line(context: AgentContext, line: str) -> None:
210 + lines = list(context.data.get(CTX_TG_PROGRESS_LINES) or [])
211 + if lines and lines[-1] == line:
212 + return
213 + lines.append(line)
214 + context.data[CTX_TG_PROGRESS_LINES] = lines[-MAX_PROGRESS_LINES:]
215 +
216 +
217 +def _mark_tool_failed(context: AgentContext, tool_name: str) -> None:
218 + lines = list(context.data.get(CTX_TG_PROGRESS_LINES) or [])
219 + if not lines:
220 + return
221 +
222 + label = _tool_label(tool_name)
223 + for index in range(len(lines) - 1, -1, -1):
224 + line = lines[index]
225 + if _line_matches_tool(line, label):
226 + _, _, suffix = line.partition(" ")
227 + lines[index] = f"❌ {suffix or label}"
228 + context.data[CTX_TG_PROGRESS_LINES] = lines[-MAX_PROGRESS_LINES:]
229 + return
230 +
231 + lines.append(f"❌ {label}")
232 + context.data[CTX_TG_PROGRESS_LINES] = lines[-MAX_PROGRESS_LINES:]
233 +
234 +
235 +def _bot_instance(context: AgentContext):
236 + bot_name = context.data.get(CTX_TG_BOT)
237 + return get_bot(bot_name) if bot_name else None
238 +
239 +
240 +def _reply_to(context: AgentContext) -> int | None:
241 + value = context.data.get(CTX_TG_REPLY_TO)
242 + try:
243 + return int(value) if value else None
244 + except (TypeError, ValueError):
245 + return None
246 +
247 +
248 +def _tool_label(tool_name: str) -> str:
249 + value = (tool_name or "tool").replace("_", " ").replace("-", " ").strip()
250 + return value or "tool"
251 +
252 +
253 +def _line_matches_tool(line: str, label: str) -> bool:
254 + _, _, suffix = line.partition(" ")
255 + normalized_suffix = suffix.strip().lower()
256 + normalized_label = label.strip().lower()
257 + return normalized_suffix == normalized_label or normalized_suffix.startswith(f"{normalized_label}:")
258 +
259 +
260 +def _tool_progress_label(tool_name: str, args: dict) -> str:
261 + label = _tool_label(tool_name)
262 + detail = _tool_detail(tool_name, args)
263 + return f"{label}: {detail}" if detail else label
264 +
265 +
266 +def _tool_detail(tool_name: str, args: dict) -> str:
267 + if not isinstance(args, dict) or not args:
268 + return ""
269 +
270 + normalized = (tool_name or "").strip().lower()
271 + if "search" in normalized:
272 + return _first_arg(args, ("query", "q", "search", "term", "keywords", "pattern"))
273 + if "browser" in normalized:
274 + return _first_arg(args, ("url", "link", "query", "action"))
275 + if "file" in normalized:
276 + return _first_arg(args, ("path", "file", "filename", "query", "pattern"))
277 + if "skill" in normalized:
278 + return _first_arg(args, ("skill", "name", "query", "path"))
279 +
280 + return _first_arg(args, ("action", "method", "operation"))
281 +
282 +
283 +def _first_arg(args: dict, keys: tuple[str, ...]) -> str:
284 + for key in keys:
285 + value = args.get(key)
286 + if value is None:
287 + continue
288 + text = _compact_detail(value)
289 + if text:
290 + return text
291 + return ""
292 +
293 +
294 +def _compact_detail(value: object) -> str:
295 + if isinstance(value, (list, tuple)):
296 + parts = [_compact_detail(item) for item in value[:2]]
297 + text = ", ".join(part for part in parts if part)
298 + if len(value) > 2:
299 + text = f"{text}, ..."
300 + elif isinstance(value, dict):
301 + return ""
302 + else:
303 + text = str(value).strip()
304 +
305 + text = re.sub(r"\s+", " ", text).strip().strip("\"'")
306 + if len(text) > 80:
307 + text = f"{text[:77].rstrip()}..."
308 + return text
309 +
310 +
311 +def _tool_emoji(tool_name: str) -> str:
312 + normalized = (tool_name or "").strip().lower()
313 + for key, emoji in TOOL_EMOJIS.items():
314 + if key in normalized:
315 + return emoji
316 + return "🛠️"
317 +
318 +
319 +def _format_response(text: str) -> str:
320 + value = _visible_response_text(text)
321 + if not value:
322 + return ""
323 + return tc.md_to_telegram_html(value)
324 +
325 +
326 +def _strip_incomplete_tool_markup(text: str) -> str:
327 + value = text.lstrip()
328 + value = re.sub(r"^<[^>\n]{0,80}$", "", value)
329 + value = re.sub(r"^```(?:json|xml)?\s*$", "", value, flags=re.IGNORECASE)
330 + return value
331 +
332 +
333 +def _visible_response_text(text: str) -> str:
334 + value = _trim(text or "")
335 + return _strip_incomplete_tool_markup(value)
336 +
337 +
338 +def _trim(text: str) -> str:
339 + if len(text) <= MAX_STREAM_CHARS:
340 + return text
341 + return text[-MAX_STREAM_CHARS:]
342 +
343 +
344 +def _keyboard_markup(keyboard: list[list[dict]] | None) -> dict | None:
345 + if not keyboard:
346 + return None
347 + rows: list[list[dict[str, str]]] = []
348 + for row in keyboard:
349 + out_row: list[dict[str, str]] = []
350 + for button in row:
351 + text = str(button.get("text") or "")[:64]
352 + if not text:
353 + continue
354 + if button.get("url"):
355 + out_row.append({"text": text, "url": str(button["url"])})
356 + else:
357 + data = str(button.get("callback_data", text))
358 + out_row.append({"text": text, "callback_data": data[:64]})
359 + if out_row:
360 + rows.append(out_row)
361 + return {"inline_keyboard": rows} if rows else None
plugins/_telegram_integration/helpers/handler.py
+100 -65
@@ -21,6 +21,7 @@ from helpers.errors import format_error
21 from initialize import initialize_agent
22
23 from plugins._telegram_integration.helpers import telegram_client as tc
24 +from plugins._telegram_integration.helpers import command_ui
25 from plugins._telegram_integration.helpers.bot_manager import get_bot
26 from plugins._telegram_integration.helpers.constants import (
27 PLUGIN_NAME,
@@ -29,6 +30,7 @@ from plugins._telegram_integration.helpers.constants import (
30 CTX_TG_BOT,
31 CTX_TG_BOT_CFG,
32 CTX_TG_CHAT_ID,
33 + CTX_TG_CHAT_TYPE,
34 CTX_TG_USER_ID,
35 CTX_TG_USERNAME,
36 CTX_TG_TYPING_STOP,
@@ -141,55 +143,15 @@ async def handle_start(message: TgMessage, bot_name: str, bot_cfg: dict):
143 f"\U0001f44b Hello {user.first_name}! I'm connected to Agent Zero.\n\n"
144 "Send me a message and I'll process it.\n"
145 "Use /clear to reset the conversation.\n"
144 - "Use /project, /config, or /send to control the current chat.",
146 + "Use /project, /model, /agent, or /send to control the current chat.",
147 parse_mode=None,
148 + reply_to_message_id=message.message_id,
149 )
150
151 # Ensure a chat context exists
152 await _get_or_create_context(bot_name, bot_cfg, message)
153
154
152 -async def handle_clear(message: TgMessage, bot_name: str, bot_cfg: dict):
153 - """Handle /clear command — reset user's chat context."""
154 - user = message.from_user
155 - if not user:
156 - return
157 -
158 - if not _is_allowed(bot_cfg, user.id, user.username):
159 - return
160 -
161 - key = _map_key(bot_name, user.id, message.chat.id)
162 -
163 - with _chat_map_lock:
164 - state = _load_state()
165 - ctx_id = state.get("chats", {}).get(key)
166 - if ctx_id:
167 - ctx = AgentContext.get(ctx_id)
168 - if ctx:
169 - ctx.reset()
170 - PrintStyle.info(f"Telegram ({bot_name}): cleared chat for user {user.id}")
171 -
172 - instance = get_bot(bot_name)
173 - if instance:
174 - await _send_with_temp_bot(
175 - instance.bot.token, message.chat.id,
176 - "Chat cleared. Send a new message to start fresh.",
177 - parse_mode=None,
178 - )
179 -
180 - # Send notification
181 - if bot_cfg.get("notify_messages", False):
182 - username_str = f"@{user.username}" if user.username else str(user.id)
183 - NotificationManager.send_notification(
184 - type=NotificationType.INFO,
185 - priority=NotificationPriority.NORMAL,
186 - title="Telegram: chat cleared",
187 - message=f"{username_str} cleared their chat via /clear",
188 - display_time=5,
189 - group="telegram",
190 - )
191 -
192 -
155 async def handle_message(message: TgMessage, bot_name: str, bot_cfg: dict):
156 """Handle incoming user message."""
157 user = message.from_user
@@ -212,26 +174,38 @@ async def handle_message(message: TgMessage, bot_name: str, bot_cfg: dict):
174 parse_mode=None,
175 )
176 return
177 + context.data[CTX_TG_CHAT_TYPE] = str(message.chat.type or "")
178 + context.data[CTX_TG_REPLY_TO] = message.message_id
179 +
180 + if await command_ui.handle_command(
181 + context,
182 + instance.bot.token,
183 + message.chat.id,
184 + message.message_id,
185 + text,
186 + ):
187 + return
188
189 command_reply = integration_commands.try_handle_command(context, text)
190 if command_reply is not None:
218 - await _send_with_temp_bot(instance.bot.token, message.chat.id, command_reply, parse_mode=None)
191 + await _send_with_temp_bot(
192 + instance.bot.token,
193 + message.chat.id,
194 + command_reply,
195 + parse_mode=None,
196 + reply_to_message_id=message.message_id,
197 + )
198 + return
199 + if integration_commands.extract_command_line(text).startswith("/"):
200 + command = integration_commands.extract_command_line(text).split(" ", 1)[0]
201 + await _send_with_temp_bot(
202 + instance.bot.token,
203 + message.chat.id,
204 + integration_commands.unknown_command_text(command),
205 + parse_mode=None,
206 + reply_to_message_id=message.message_id,
207 + )
208 return
220 -
221 - # Start persistent typing indicator (thread-based, works across event loops)
222 - typing_stop = _start_typing(instance.bot.token, message.chat.id)
223 -
224 - # Store stop event so send_telegram_reply can cancel typing
225 - context.data[CTX_TG_TYPING_STOP] = typing_stop
226 -
227 - # In group chats, if user replied to the bot's message, reply to the user's message
228 - reply_to_id = None
229 - if message.chat.type != "private" and instance.bot_info:
230 - if (message.reply_to_message
231 - and message.reply_to_message.from_user
232 - and message.reply_to_message.from_user.id == instance.bot_info.id):
233 - reply_to_id = message.message_id
234 - context.data[CTX_TG_REPLY_TO] = reply_to_id
209
210 # Use temp bot for downloads (cross-event-loop safe)
211 async with _temp_bot(instance.bot.token) as dl_bot:
@@ -245,6 +219,24 @@ async def handle_message(message: TgMessage, bot_name: str, bot_cfg: dict):
219 body=text,
220 )
221
222 + if context.is_running():
223 + item = mq.add(context, user_msg, attachments)
224 + save_tmp_chat(context)
225 + await _send_with_temp_bot(
226 + instance.bot.token,
227 + message.chat.id,
228 + f"Queued message #{item.get('seq', len(mq.get_queue(context)))}. Use /send to flush queued work, or /steer <message> to interrupt the active run.",
229 + parse_mode=None,
230 + reply_to_message_id=message.message_id,
231 + )
232 + return
233 +
234 + # Start persistent typing indicator (thread-based, works across event loops)
235 + typing_stop = _start_typing(instance.bot.token, message.chat.id)
236 +
237 + # Store stop event so send_telegram_reply can cancel typing
238 + context.data[CTX_TG_TYPING_STOP] = typing_stop
239 +
240 msg_id = str(uuid.uuid4())
241 mq.log_user_message(context, user_msg, attachments, message_id=msg_id, source=" (telegram)")
242 context.communicate(UserMessage(
@@ -287,16 +279,40 @@ async def handle_callback_query(query: CallbackQuery, bot_name: str, bot_cfg: di
279 return
280
281 context = await _get_or_create_context_from_user(
290 - bot_name, bot_cfg, user.id, user.username, query.message.chat.id,
282 + bot_name, bot_cfg, user.id, user.username, query.message.chat.id, str(query.message.chat.type or ""),
283 )
284 if not context:
285 return
286 + context.data[CTX_TG_REPLY_TO] = query.message.message_id
287 +
288 + instance = get_bot(bot_name)
289 + if instance:
290 + try:
291 + if await command_ui.handle_callback(
292 + context,
293 + instance.bot.token,
294 + query.message.chat.id,
295 + query.message.message_id,
296 + text,
297 + ):
298 + return
299 + except Exception as e:
300 + PrintStyle.error(f"Telegram callback failed: {format_error(e)}")
301 + if text.startswith("tg:"):
302 + return
303 + if text.startswith("tg:"):
304 + return
305
306 command_reply = integration_commands.try_handle_command(context, text)
307 if command_reply is not None:
297 - instance = get_bot(bot_name)
308 if instance:
299 - await _send_with_temp_bot(instance.bot.token, query.message.chat.id, command_reply, parse_mode=None)
309 + await _send_with_temp_bot(
310 + instance.bot.token,
311 + query.message.chat.id,
312 + command_reply,
313 + parse_mode=None,
314 + reply_to_message_id=query.message.message_id,
315 + )
316 return
317
318 agent = context.agent0
@@ -347,7 +363,7 @@ async def _get_or_create_context(
363 if not user:
364 return None
365 return await _get_or_create_context_from_user(
350 - bot_name, bot_cfg, user.id, user.username, message.chat.id,
366 + bot_name, bot_cfg, user.id, user.username, message.chat.id, str(message.chat.type or ""),
367 )
368
369
@@ -357,6 +373,7 @@ async def _get_or_create_context_from_user(
373 user_id: int,
374 username: str | None,
375 chat_id: int,
376 + chat_type: str = "",
377 ) -> AgentContext | None:
378 key = _map_key(bot_name, user_id, chat_id)
379
@@ -369,6 +386,7 @@ async def _get_or_create_context_from_user(
386 if ctx_id:
387 ctx = AgentContext.get(ctx_id)
388 if ctx:
389 + ctx.data[CTX_TG_CHAT_TYPE] = chat_type or ctx.data.get(CTX_TG_CHAT_TYPE, "")
390 return ctx
391 # Context was garbage collected, remove stale mapping
392 chats.pop(key, None)
@@ -382,6 +400,7 @@ async def _get_or_create_context_from_user(
400 ctx.data[CTX_TG_BOT] = bot_name
401 ctx.data[CTX_TG_BOT_CFG] = bot_cfg
402 ctx.data[CTX_TG_CHAT_ID] = chat_id
403 + ctx.data[CTX_TG_CHAT_TYPE] = chat_type
404 ctx.data[CTX_TG_USER_ID] = user_id
405 ctx.data[CTX_TG_USERNAME] = username or ""
406
@@ -512,7 +531,11 @@ async def send_telegram_reply(
531
532 if response_text:
533 html_text = tc.md_to_telegram_html(response_text)
515 - if keyboard:
534 + from plugins._telegram_integration.helpers import draft_stream
535 +
536 + if await draft_stream.finalize_response(context, response_text, keyboard):
537 + pass
538 + elif keyboard:
539 await tc.send_text_with_keyboard(reply_bot, chat_id, html_text, keyboard, reply_to_message_id=reply_to)
540 else:
541 await tc.send_text(reply_bot, chat_id, html_text, reply_to_message_id=reply_to)
@@ -537,10 +560,22 @@ async def _temp_bot(token: str, **kwargs):
560 await bot.session.close()
561
562
540 -async def _send_with_temp_bot(token: str, chat_id: int, text: str, parse_mode: str | None = None):
563 +async def _send_with_temp_bot(
564 + token: str,
565 + chat_id: int,
566 + text: str,
567 + parse_mode: str | None = None,
568 + reply_to_message_id: int | None = None,
569 +):
570 """Send text using a temporary Bot to avoid cross-event-loop session issues."""
571 async with _temp_bot(token) as bot:
543 - await tc.send_text(bot, chat_id, text, parse_mode=parse_mode)
572 + await tc.send_text(
573 + bot,
574 + chat_id,
575 + text,
576 + reply_to_message_id=reply_to_message_id,
577 + parse_mode=parse_mode,
578 + )
579
580
581 def _start_typing(token: str, chat_id: int) -> threading.Event:
plugins/_telegram_integration/helpers/telegram_client.py
+84
@@ -1,6 +1,7 @@
1 import os
2 import re
3
4 +import aiohttp
5 from aiogram import Bot
6 from aiogram.exceptions import TelegramBadRequest
7 from aiogram.types import (
@@ -17,6 +18,7 @@ _UNSET = object() # sentinel: "not provided" (lets Bot default apply)
18 # Text messages
19
20 MAX_MESSAGE_LENGTH: int = 4096 # Telegram message length limit
21 +TELEGRAM_API_BASE: str = "https://api.telegram.org"
22
23
24 async def send_text(
@@ -171,6 +173,88 @@ async def send_typing(bot: Bot, chat_id: int):
173 except Exception:
174 pass
175
176 +
177 +async def raw_send_text(
178 + token: str,
179 + chat_id: int,
180 + text: str,
181 + reply_to_message_id: int | None = None,
182 + parse_mode: str | None = "HTML",
183 + reply_markup: dict | None = None,
184 +) -> int | None:
185 + payload: dict[str, object] = {
186 + "chat_id": chat_id,
187 + "text": text[:MAX_MESSAGE_LENGTH],
188 + }
189 + if parse_mode:
190 + payload["parse_mode"] = parse_mode
191 + if reply_to_message_id:
192 + payload["reply_parameters"] = {"message_id": int(reply_to_message_id)}
193 + if reply_markup:
194 + payload["reply_markup"] = reply_markup
195 + data = await _raw_post(token, "sendMessage", payload)
196 + result = data.get("result") if isinstance(data, dict) else None
197 + if isinstance(result, dict):
198 + return result.get("message_id")
199 + return None
200 +
201 +
202 +async def raw_edit_text(
203 + token: str,
204 + chat_id: int,
205 + message_id: int,
206 + text: str,
207 + parse_mode: str | None = "HTML",
208 + reply_markup: dict | None = None,
209 +) -> bool:
210 + payload: dict[str, object] = {
211 + "chat_id": chat_id,
212 + "message_id": message_id,
213 + "text": text[:MAX_MESSAGE_LENGTH],
214 + }
215 + if parse_mode:
216 + payload["parse_mode"] = parse_mode
217 + if reply_markup:
218 + payload["reply_markup"] = reply_markup
219 + data = await _raw_post(token, "editMessageText", payload)
220 + if not isinstance(data, dict):
221 + return False
222 + if data.get("ok"):
223 + return True
224 + description = str(data.get("description") or "").lower()
225 + return "message is not modified" in description
226 +
227 +
228 +async def raw_edit_reply_markup(
229 + token: str,
230 + chat_id: int,
231 + message_id: int,
232 + reply_markup: dict | None = None,
233 +) -> bool:
234 + payload: dict[str, object] = {
235 + "chat_id": chat_id,
236 + "message_id": message_id,
237 + }
238 + if reply_markup:
239 + payload["reply_markup"] = reply_markup
240 + data = await _raw_post(token, "editMessageReplyMarkup", payload)
241 + return bool(isinstance(data, dict) and data.get("ok"))
242 +
243 +
244 +async def _raw_post(token: str, method: str, payload: dict[str, object]) -> dict:
245 + url = f"{TELEGRAM_API_BASE}/bot{token}/{method}"
246 + try:
247 + timeout = aiohttp.ClientTimeout(total=10)
248 + async with aiohttp.ClientSession(timeout=timeout) as session:
249 + async with session.post(url, json=payload) as response:
250 + data = await response.json(content_type=None)
251 + if response.status != 200 or not data.get("ok"):
252 + PrintStyle.debug(f"Telegram {method} failed: {data}")
253 + return data if isinstance(data, dict) else {}
254 + except Exception as e:
255 + PrintStyle.debug(f"Telegram {method} failed: {format_error(e)}")
256 + return {}
257 +
258 # File download
259
260 async def download_file(
plugins/_whatsapp_integration/helpers/handler.py
+18
@@ -207,6 +207,24 @@ async def _route_to_chat(
207 msg_id = str(uuid.uuid4())
208 media_urls = msg.get("mediaUrls", [])
209 attachments = await _save_incoming_media(media_urls) if media_urls else []
210 +
211 + if context.is_running():
212 + item = mq.add(context, user_msg, attachments)
213 + save_tmp_chat(context)
214 + port = int((plugins.get_plugin_config(PLUGIN_NAME) or {}).get("bridge_port", 3100))
215 + base_url = bridge_manager.get_bridge_url(port)
216 + chat_id = context.data.get(CTX_WA_CHAT_ID, "") or msg.get("chatId", "")
217 + reply_to = msg.get("messageId", "") if context.data.get(CTX_WA_IS_GROUP) else ""
218 + await wa_client.send_message(
219 + base_url,
220 + chat_id,
221 + f"Queued message #{item.get('seq', len(mq.get_queue(context)))}. Use /send to flush queued work, or /steer <message> to interrupt the active run.",
222 + reply_to=reply_to,
223 + )
224 + await wa_client.send_typing(base_url, chat_id, paused=True)
225 + context.data[CTX_WA_TYPING_ACTIVE] = False
226 + return
227 +
228 mq.log_user_message(
229 context, user_msg, attachments, message_id=msg_id, source=" (whatsapp)",
230 )