Telegram channel integration - #4
Open
lefoulkrod wants to merge 15 commits into
Open
lefoulkrod wants to merge 15 commits into
lefoulkrod wants to merge 15 commits into
Conversation
End-to-end Telegram support routed through a credential-isolated
broker process. The bot token never enters the main app; user-ID
allowlist enforcement happens in the broker; the channel and the
notifications path both call into the same broker via
broker_client.call().
Architecture:
1. sdk/turn/_executor.py + sdk/__init__.py — TurnExecutor and
Conversation as SDK primitives with explicit injection points
(TurnPersistence protocol, SystemPromptBuilder callable,
preloaded_skills sequence). No imports from agents.AgentProfile,
conversations, tools.memory, or tools.virtual_computer; channels
build their own Agent and supply persistence + a memory-aware
prompt builder.
2. conversations._turn_persistence.DiskTurnPersistence — adapter that
exposes the on-disk store through the TurnPersistence protocol.
3. tools.memory.memory_prompt_block — formatted memory block that
channels prepend to the system prompt; called fresh per turn so
memory updates are visible immediately.
4. channels/ — new top-level package. channels/telegram/_runner.py
pulls Telegram updates via broker_client.call("next_updates", ...)
and sends outbound text via broker_client.call("send_message", ...).
Long-poll loop with exponential backoff on transient errors and
long backoff on auth/not-connected.
5. integrations/brokers/telegram_broker/ — credential-isolated
broker. Owns the aiogram.Bot, runs an UpdatePump that long-polls
Telegram and filters each update against the allowlist before
enqueueing. RPC verbs: get_me, next_updates(timeout_ms),
send_message. send_document declared in the requirement table
but not implemented (host-path binding pending).
6. integrations.supervisor._catalog — telegram entry with the
Capability.TELEGRAM (new) and env_injection mapping for token and
allowed_user_ids.
7. tasks/_notifier.py — drops the direct httpx Telegram API code in
favor of broker_client.call("send_message", ...). One broker
serves both the bidirectional channel and the notifier.
8. Integrations wizard — providers.js gains a "telegram" entry with
authFlow=bot_token. BotTokenSteps.jsx hosts the new three-step
wizard (explainer → credentials → verifying). AddIntegrationModal
routes the new flow and posts auth_blob {token, allowed_user_ids}
with a client-supplied user_suffix derived from a sanitized
instance name. server/_integrations_routes.handle_add_integration
now accepts a client-supplied user_suffix, falling back to email
derivation when absent.
Security model:
- Token lives only in the broker subprocess. The supervisor passes
it via env on spawn; the broker wipes it from os.environ after
capture.
- Empty user-ID allowlist fails closed at broker boot (exit
GENERIC_ERROR) — refuses to start rather than accept all senders.
- Unauthorized messages are silently dropped — replying confirms
the bot is live and burns outbound rate limits — with aggregate
per-sender drop counts logged on a 60-second window.
- The channel-side allowlist is gone entirely; the broker is the
single chokepoint.
Tests:
- 80 new unit tests across TurnExecutor injection paths, formatter
split branches, ConversationMap semantics, _parse_allowed_user_ids
for the broker, UpdatePump filter + drop logging, VerbDispatcher
permissions + verb behavior. 1209 pass total.
Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
…gration-v2 # Conflicts: # sdk/tools/_spawn_agent.py # server/message_handler.py
Channels boot once at app start, but the Telegram integration is added later via the wizard. Previous shape did a one-shot discovery at boot and required an app restart to pick up a freshly-added integration. Replace the one-shot with an event-driven wait: - TelegramChannel exposes ``notify_integration_added(slug)``. Sets an internal asyncio.Event when slug == "telegram"; ignores everything else so other integrations don't thrash the channel. - The supervise task tries discovery once. If nothing is found, it parks on the event. No polling. - server/_integrations_routes.handle_add_integration calls ``notify_integration_added`` after a successful add. The channel wakes, re-runs discovery, finds the new integration, probes the broker, and transitions into the pull loop. All without an app restart. - stop() also sets the event so a shutdown while waiting unblocks cleanly. Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
Two unhandled exception classes were crashing the Telegram pull task when an integration was removed and re-added (broker process killed + socket unlinked, then respawned with a fresh socket): - broker_client._rpc_one_shot didn't catch the asyncio.IncompleteReadError fired when the broker dies mid-call, nor the FileNotFoundError / ConnectionRefusedError that fires when the socket has been unlinked. Wrap both in IntegrationError so callers' normal retry-with-backoff paths handle them. - TelegramChannel._pull_loop gains an except Exception catch-all that logs with traceback and backs off, so any future surprise exception doesn't silently end the pull task and leave the bot unreachable until app restart. Symptom that surfaced this: removing the Telegram integration while the channel was inside a long-poll left the pull task dead even after re-adding the integration spawned a fresh broker on the same socket. The fix lets the loop catch the disconnect, backoff, and reconnect on the next iteration. Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
Real-bot test traceback: AttributeError: 'AgentEvent' object has no attribute 'type' The Telegram channel's event-dispatch loop and the spawn_agent event-accumulation loop both reached for ``event.type`` to switch on event kind. AgentEvent is just the envelope — type lives on the discriminated ``payload`` (``event.payload.type``). Why tests didn't catch it: the unit tests construct AgentEvent with real payload subclasses, but the dispatch code paths weren't exercised end-to-end against a real event stream until a Telegram turn actually ran in the manual-test container. Unit coverage for "channel dispatches on event kind" would have caught this. Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
The broker downloads photo/document attachments eagerly to the shared
"downloads" host-path role (same place email attachments and browser
saves land). The channel sees a local path on each update and feeds the
agent an augmented user message that names the files — same shape SSE
attachments use, so the agent has one mental model.
Broker:
- catalog entry gains a host_paths binding: role="downloads",
env_var="DOWNLOADS_DIR", mode="write".
- __main__ reads DOWNLOADS_DIR and threads it through to UpdatePump.
- _updates._message_to_dict forwards messages with text, caption, or
any supported attachment (was: text-only).
- _extract_attachment_meta picks the largest photo size and any
document, returns {kind, file_id, file_name, mime_type, ...}.
- Filenames sanitized — path separators stripped, unsafe chars
collapsed to "_", length-capped at 120.
- UpdatePump._handle_update is now async. Downloads each attachment
via bot.get_file + bot.download into DOWNLOADS_DIR before enqueueing.
Download failures drop the attachment from the wire shape but keep
the message (caption + successful peers still reach the agent).
Channel:
- _dispatch_message tolerates attachments-without-text and
text-with-attachments. Builds the agent-facing user_content via a
new _build_user_content helper that mirrors SSE's
_augment_message_with_attachments shape.
- Existing turn flow unchanged; the augmented text goes straight into
TurnExecutor.execute(user_content=...).
Out of scope (future):
- Voice notes / video / stickers / animations
- File size limits (Telegram caps at 20MB via getFile)
- Cleanup of accumulated files in the downloads dir
- Caption-only message with no attachments uses the text-only path
(already worked)
Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
Two Telegram messages arriving in one broker batch both passed the "is a turn already running?" guard because is_turn_active() reads a ContextVar that doesn't flip True until the turn task actually enters turn_scope. Between asyncio.create_task and the task running, the var is unset, so the dispatcher's second iteration in the same await window saw "no turn running" and spawned a duplicate. Replace the ContextVar check with a per-chat in-flight tasks dict the channel owns directly. Membership is updated synchronously before any await yields, so the second dispatch sees the first task immediately. The /stop and the channel's own use of is_turn_active() are unchanged (those want the "actually executing in turn_scope" semantics). Reproed by sending a file with a caption — Telegram delivered the media and the text as separate updates in one getUpdates batch, both of which went on to spawn turns for the same chat. Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
Inline-keyboard picker driven entirely by the existing on-disk conversation store. No new state, no new broker verb. The same mechanism the web UI's conversation list reads. Channel-side: - New /list [query] command. Lists conversations belonging to this chat (default ``telegram_<chat_id>`` plus any /new-spawned uuid variants), most-recent first, capped at 10. Optional query is a case-insensitive substring match against title and first message. - Tapping a row fires a callback with a new "conv:" prefix; the channel binds this chat to the picked conversation via a new ConversationMap.set() method, drops the in-memory cache so the next turn rehydrates from disk, and confirms. - _belongs_to_chat uses a strict-prefix check (``telegram_<id>`` or ``telegram_<id>_…``) to avoid the false-positive a naive startswith hits when one chat_id is a prefix of another. - _resume_conversation validates the picked id actually belongs to this chat — a malicious callback payload can't trick the channel into binding to someone else's conversation. - /help updated to mention /list. Backed by: - New ``ConversationMap.set(chat_id, conv_id)`` — bind without resetting. - ``conversations.list_conversations()`` (existing) — single source of truth for the chat's conversation history. Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
User expectation was that /list surfaces every past conversation regardless of origin — the conversation store is the single source of truth and matches the web UI's listing behavior. Removed the per-chat prefix filter from /list and from the resume safety check. Callback data is bot-controlled (only valid conversation ids ever get put on buttons) so no per-chat prefix check is needed on the resume path. The disk-existence check still catches a stale callback whose conversation was deleted between list-and-tap. Dropped the now-unused _belongs_to_chat helper and its tests. Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
…gration-v2 # Conflicts: # server/aiohttp_app.py # server/message_handler.py
…Telegram channel Both call sites kept their own in-memory map of Conversation objects, but the implementations had drifted: the SSE handler had a proper LRU with a 25-entry cap and active-turn-skip eviction, while the Telegram channel kept a plain unbounded dict that grew for the lifetime of the process. This was a latent memory leak — every chat the bot interacted with stayed cached forever — and an open invitation for a third channel to copy whichever shape it found first. New ``conversations.ConversationCache``: - Owns the LRU OrderedDict, the max-size cap, disk-hydrate on miss, and the active-turn-skip eviction logic. - ``is_active`` and ``on_evict`` are injected callbacks so the cache doesn't know about ``sdk.turn.is_turn_active`` or the SSE handler's per-conversation Playwright release directly. - API: ``get(id) -> (Conversation, is_new)``, ``resume(id) -> Conversation | None`` (force-reload from disk), ``pop(id)``, ``clear()``, plus ``__contains__`` and ``__iter__`` for test-friendly access. Wiring: - ``server/message_handler.py`` constructs a module-level cache with ``is_active=is_turn_active`` and an ``on_evict`` that releases the evicted conversation's browser context — preserving the eviction-time Playwright cleanup the inline implementation had. - ``channels/telegram/_runner.py`` constructs its own instance with ``is_active=is_turn_active`` (no ``on_evict``, no per-conv browser). This is where the bounded-cache behavior is genuinely new. Module structure: - Pulled the ``Conversation`` dataclass out of ``sdk/turn/_executor.py`` into ``sdk/turn/_conversation.py`` so importers below the SDK (the new cache) can grab the dataclass without pulling in ``TurnExecutor`` and its transitive deps — that triggered a circular import. Tests: - Moved all cache-behavior tests out of ``test_message_handler.py`` into ``tests/unit/conversations/test_cache.py`` and rewrote them against the cache class directly. Net +1 test (added an explicit ``on_evict`` coverage case). - ``test_message_handler.py`` shrinks to a smoke test for ``resume_conversation``'s dict-return shape. Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
- channels/telegram/_formatter.py: replace dead `escape_markdown` with `to_markdownv2_chunks`; agent output runs through telegramify-markdown (convert → split_markdownv2) and is sent with parse_mode="MarkdownV2" so **bold**, code blocks, lists, etc. render natively instead of as literal asterisks. Broker `send_message` verb gains optional parse_mode. - channels/telegram/_runner.py → _channel.py rename, with import + aiohttp app-key updates (`telegram_bot_runner` → `telegram_channel`). - TELEGRAM_INTEGRATION_ID / TELEGRAM_CHAT_ID env vars become settings (`telegram_notifier_integration_id`, `telegram_notifier_chat_id`) editable in Settings → System → Notifications; migration 006 seeds empty defaults on existing installs. Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
Per PR review on sdk/turn/_executor.py: TurnExecutor was orchestrating
setup that callers already had context for (skill loading, persistence
wiring, first-turn hooks, system-prompt building). Push those out to
the caller; Conversation grows the per-conversation state that actually
needs to survive across turns.
- Conversation: add agent_state field; TYPE_CHECKING import keeps it a
dependency-free leaf.
- AgentState.create(skill_names=…): async factory used by every
callsite to assemble (core tools + named skills), log+skip on missing.
- TurnExecutor.execute(): kwargs slim to conversation/agent/user_content
+ a few span-metadata params. Raises if conversation.agent_state is
None. Stops loading skills, saving events, saving skills, firing
on_new_conversation, and evaluating build_system_prompt.
- TurnPersistence Protocol moves to sdk/turn/_persistence.py;
SystemPromptBuilder dropped (caller composes the instruction directly).
- Callers updated:
- server/message_handler.py and channels/telegram/_channel.py hydrate
agent_state on cache miss, prefix the agent instruction with
memory_prompt_block(), then save events / save skills / fire
on_new_conversation themselves after the turn.
- tasks/_executor.py: _build_agent returns (Agent, AgentState);
state attaches to the Conversation it constructs.
- sdk/tools/_spawn_agent.py: attaches the already-built state to the
spawned sub-agent's Conversation.
Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
…gration-v2 # Conflicts: # integrations/permissions.py # integrations/supervisor/_catalog.py # server/_integrations_routes.py # server/ui/src/components/integrations/add-wizard/AddIntegrationModal.jsx # server/ui/src/components/integrations/add-wizard/providers.js
A pick-up note for next session: where the PR lives (omnideck#4), what's landed on the branch, the known deferred items (no streaming reply, parallel sub-agent browser contention, live chat-ID discovery, markdown edge cases), and how to resume. Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
1 participant
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Summary
/listsearches and resumes any past conversation;/profilepicks an agent profile via inline keyboard.get_me,next_updates,send_message,send_document,send_chat_action,answer_callback_query,edit_message_text,delete_message.ConversationCacheextracted from the SSE handler so the Telegram channel and the web channel share the same bounded-LRU hydrate/evict policy (skips eviction of conversations whose turn is in flight).TurnExecutorslimmed to a turn-driver primitive:Conversationgrows anagent_statefield, the executor stops loading skills / saving events / firing first-turn hooks, and channels do that lifecycle wiring inline. The cleanup covers the web channel, Telegram channel,TaskExecutor, andspawn_agent.TELEGRAM_INTEGRATION_ID/TELEGRAM_CHAT_IDenv vars tosettings.jsonkeys editable in Settings → System → Notifications (integration dropdown filtered to registered Telegram integrations, chat ID number input). Migration 006 seeds empty defaults on existing installs.telegramify-markdownso**bold**, fenced code, and lists render natively in Telegram instead of as literal asterisks.Test plan
just unit(1429 tests)runningstate🤔 Thinking…→ tool labels →✍️ Writing…) and that the final reply renders MarkdownV2 cleanly/listshows recent conversations (including web ones); tap one to resume/profileopens the inline-keyboard picker; selecting a profile persists for the chat🤖 Generated with Claude Code