feat(sync): add Composio telegram memory-sync pipeline - #111
CodeGhost21 wants to merge 1 commit into
Conversation
Adds `TelegramSyncPipeline` (`SyncPipeline` + `IncrementalSource`) so the `telegram` toolkit syncs end to end instead of failing at the selector's `_` arm with `does not support toolkit 'telegram'`. Modeled on the Slack pipeline (message-shaped, list-scopes-then-fetch): chats are discovered from a single `TELEGRAM_GET_UPDATES` poll and each chat's history is paged with `TELEGRAM_GET_CHAT_HISTORY`. Dedupe is stable — `document_id = telegram:<chat_id>:<message_id>`, `metadata.taint = "external_sync"`, and a per-chat `metadata.path_scope` collection scope — so re-syncing upserts rather than duplicating. Debug logging is content-free (counts and identifiers only; never message text or names). Registered through providers/mod.rs, composio/mod.rs, and sync/mod.rs alongside the six existing pipelines. Tests: toolkit()/action(), extract_page() offset paging on a sample Composio payload, dedup_key(), and a stable document() id/taint/scope. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
📝 WalkthroughWalkthroughAdds an incremental Composio Telegram synchronization pipeline that discovers chats, paginates message history, creates stable scoped documents, and exposes the pipeline through the memory sync module. ChangesTelegram synchronization
Estimated code review effort: 3 (Moderate) | ~25 minutes Sequence Diagram(s)sequenceDiagram
participant SyncPipeline
participant TelegramSyncPipeline
participant ComposioClient
SyncPipeline->>TelegramSyncPipeline: tick()
TelegramSyncPipeline->>ComposioClient: Run TELEGRAM_GET_UPDATES
ComposioClient-->>TelegramSyncPipeline: Active chats
TelegramSyncPipeline->>ComposioClient: Run TELEGRAM_GET_CHAT_HISTORY with offset and page size
ComposioClient-->>TelegramSyncPipeline: Message page
TelegramSyncPipeline-->>SyncPipeline: Stable scoped message documents
Possibly related issues
Suggested reviewers: Poem
🚥 Pre-merge checks | ✅ 5✅ Passed checks (5 passed)
Comment |
![coderabbitai[bot]](https://avatars.githubusercontent.com/in/347564?s=60&v=4){kind=small}
There was a problem hiding this comment.
Actionable comments posted: 1
🧹 Nitpick comments (2)
src/memory/sync/composio/providers/telegram.rs (2)
64-87: 📐 Maintainability & Code Quality | 🔵 Trivial | ⚡ Quick winAdd doc comments to the public struct and constructors.
TelegramSyncPipelineand its publicnew/with_limitsmethods have no///doc comments. Per path instructions forsrc/**/*.rs, public APIs should be documented thoroughly, preferring item docs; only the module-level doc references this type indirectly.As per path instructions,
src/**/*.rs: "Document public APIs, module contracts, and non-obvious behavior thoroughly, preferring module-level docs and item docs."📝 Suggested doc additions
+/// Incremental Composio sync pipeline for Telegram: discovers active chats via +/// `TELEGRAM_GET_UPDATES` and pages through each chat's history with +/// `TELEGRAM_GET_CHAT_HISTORY`. pub struct TelegramSyncPipeline { client: ComposioClient, connection_id: String, max_pages: usize, page_size: usize, } impl TelegramSyncPipeline { + /// Builds a pipeline with default limits (`max_pages: 10`, `page_size: 100`). pub fn new(client: ComposioClient, connection_id: impl Into<String>) -> Self { ... } + /// Overrides the default pagination limits; `page_size` is clamped to Telegram's + /// 1-100 range. pub fn with_limits(mut self, max_pages: usize, page_size: usize) -> Self { ... } }🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@src/memory/sync/composio/providers/telegram.rs` around lines 64 - 87, Add thorough Rust doc comments to the public TelegramSyncPipeline struct and its public new and with_limits methods. Document the pipeline’s purpose, constructor inputs and defaults, and explain that with_limits clamps max_pages to at least 1 and page_size to the supported 1–100 range.Source: Path instructions
280-382: 📐 Maintainability & Code Quality | 🔵 Trivial | ⚡ Quick winMove tests into a
telegram_tests.rssibling.The
#[cfg(test)] mod testsblock is mixed into the implementation file. Per path instructions forsrc/**/*.rs, tests should live in a per-file sibling (e.g.,telegram_tests.rs) rather than inline.As per path instructions,
src/**/*.rs: "Keep tests in per-file<name>_tests.rssiblings, such asstore.rsandstore_tests.rs, rather than mixing tests into implementation files."🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@src/memory/sync/composio/providers/telegram.rs` around lines 280 - 382, Move the inline #[cfg(test)] mod tests block from the Telegram implementation into a sibling telegram_tests.rs module, preserving all existing test cases and helpers. Wire the sibling into the implementation’s test configuration using the project’s established per-file test-module pattern, and retain access to the TelegramSyncPipeline behavior under test.Source: Path instructions
🤖 Prompt for all review comments with AI agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
Inline comments:
In `@src/memory/sync/composio/providers/telegram.rs`:
- Around line 139-180: The scopes method must persist and advance the Telegram
updates offset while discovering chats. Read the stored highest update_id from
SyncState, include offset as last_update_id + 1 in the ACTION_UPDATES request,
track the maximum update_id returned by the response, and save it back to
SyncState so subsequent polls progress beyond the first 100 updates.
---
Nitpick comments:
In `@src/memory/sync/composio/providers/telegram.rs`:
- Around line 64-87: Add thorough Rust doc comments to the public
TelegramSyncPipeline struct and its public new and with_limits methods. Document
the pipeline’s purpose, constructor inputs and defaults, and explain that
with_limits clamps max_pages to at least 1 and page_size to the supported 1–100
range.
- Around line 280-382: Move the inline #[cfg(test)] mod tests block from the
Telegram implementation into a sibling telegram_tests.rs module, preserving all
existing test cases and helpers. Wire the sibling into the implementation’s test
configuration using the project’s established per-file test-module pattern, and
retain access to the TelegramSyncPipeline behavior under test.
🪄 Autofix (Beta)
Fix all unresolved CodeRabbit comments on this PR:
- Push a commit to this branch (recommended)
- Create a new PR with the fixes
ℹ️ Review info
⚙️ Run configuration
Configuration used: Organization UI
Review profile: CHILL
Plan: Pro
Run ID: 06df63be-ae51-4bf2-bea4-1866dc903ca5
📒 Files selected for processing (4)
src/memory/sync/composio/mod.rssrc/memory/sync/composio/providers/mod.rssrc/memory/sync/composio/providers/telegram.rssrc/memory/sync/mod.rs
| async fn scopes( | ||
| &self, | ||
| executor: &dyn ActionExecutor, | ||
| connection_id: &str, | ||
| state: &mut SyncState, | ||
| ) -> anyhow::Result<Vec<SyncScope>> { | ||
| let response = checked_execute( | ||
| executor, | ||
| ACTION_UPDATES, | ||
| serde_json::json!({ "limit": 100 }), | ||
| connection_id, | ||
| state, | ||
| ) | ||
| .await?; | ||
|
|
||
| // Distinct chats, ordered by id for deterministic scope iteration. | ||
| let mut chats: BTreeMap<String, String> = BTreeMap::new(); | ||
| for update in first_array(&response.data, UPDATE_POINTERS) { | ||
| let Some(message) = MESSAGE_KEYS.iter().find_map(|key| update.get(*key)) else { | ||
| continue; | ||
| }; | ||
| let Some(chat) = message.get("chat") else { | ||
| continue; | ||
| }; | ||
| let Some(chat_id) = pick_str(chat, &["id"]) else { | ||
| continue; | ||
| }; | ||
| let label = chat_label(chat, &chat_id); | ||
| chats.entry(chat_id).or_insert(label); | ||
| } | ||
|
|
||
| tracing::debug!( | ||
| toolkit = "telegram", | ||
| connection_id, | ||
| chats = chats.len(), | ||
| "[sync:telegram] discovered chats" | ||
| ); | ||
| Ok(chats | ||
| .into_iter() | ||
| .map(|(id, label)| SyncScope::named(id, label)) | ||
| .collect()) | ||
| } |
There was a problem hiding this comment.
🎯 Functional Correctness | 🟠 Major | 🏗️ Heavy lift
🧩 Analysis chain
🌐 Web query:
Telegram Bot API getUpdates offset semantics recalculate after each call
💡 Result:
In the Telegram Bot API, the offset parameter does not automatically recalculate or increment on its own; the bot developer must explicitly manage it to confirm the receipt of updates [1][2]. To avoid receiving the same updates repeatedly, your application must keep track of the update_id and explicitly send an updated offset in subsequent calls to the getUpdates method [2][3]. The correct practice is to set the offset to the update_id of the last processed update plus one (offset = last_update_id + 1) [2][4][5]. An update is considered "confirmed" by the Telegram server as soon as the getUpdates method is called with an offset value that is higher than that update's update_id [1][6]. Once confirmed, those updates are removed from the queue and will no longer be returned by the server [2][7]. Failure to update this offset will result in the server continuously resending the oldest unconfirmed updates [2][3].
Citations:
- 1: https://core.telegram.org/bots/API
- 2: https://core.telegram.org/bots/faq
- 3: https://stackoverflow.com/questions/37098352/telegram-update-consuming
- 4: https://stackoverflow.com/questions/74364235/how-to-properly-use-getupdates-method-from-telegram-api
- 5: https://stackoverflow.com/questions/69469695/telegram-bot-getupdates-api-php
- 6: https://docs.rs/telegram-bot-api/latest/telegram_bot_api/methods/struct.GetUpdates.html
- 7: https://tg-bot-sdk.website/api/methods/get-updates/
🏁 Script executed:
#!/bin/bash
set -euo pipefail
printf '\n== File outline ==\n'
ast-grep outline src/memory/sync/composio/providers/telegram.rs --view expanded || true
printf '\n== Relevant symbol search ==\n'
rg -n "getUpdates|ACTION_UPDATES|offset|SyncState|update_id|scopes\\(" src/memory/sync/composio/providers/telegram.rs src/memory/sync -S || true
printf '\n== Nearby file sections ==\n'
nl -ba src/memory/sync/composio/providers/telegram.rs | sed -n '1,260p'Repository: tinyhumansai/tinycortex
Length of output: 9713
🏁 Script executed:
#!/bin/bash
set -euo pipefail
printf '\n== telegram provider excerpt ==\n'
sed -n '139,260p' src/memory/sync/composio/providers/telegram.rs
printf '\n== SyncState definition ==\n'
sed -n '1,240p' src/memory/sync/state.rs
printf '\n== Orchestrator / cursor handling around scopes ==\n'
sed -n '1,260p' src/memory/sync/composio/orchestrator.rsRepository: tinyhumansai/tinycortex
Length of output: 19046
🏁 Script executed:
#!/bin/bash
set -euo pipefail
rg -n "last_seen_id|advance_scope_cursor|per_scope_cursors\\(" src/memory/sync -SRepository: tinyhumansai/tinycortex
Length of output: 1537
🏁 Script executed:
#!/bin/bash
set -euo pipefail
sed -n '1,150p' src/memory/sync/composio/providers/telegram.rs
printf '\n== tests around Telegram scope/page behavior ==\n'
sed -n '300,420p' src/memory/sync/composio/providers/telegram.rsRepository: tinyhumansai/tinycortex
Length of output: 8166
Persist and advance the Telegram updates offset in chat discovery. scopes() always calls TELEGRAM_GET_UPDATES with only limit: 100, so it keeps re-reading the oldest unconfirmed updates and can miss newer chats once the backlog grows past 100. Store the highest seen update_id in SyncState and send offset = last_update_id + 1 on the next poll.
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@src/memory/sync/composio/providers/telegram.rs` around lines 139 - 180, The
scopes method must persist and advance the Telegram updates offset while
discovering chats. Read the stored highest update_id from SyncState, include
offset as last_update_id + 1 in the ACTION_UPDATES request, track the maximum
update_id returned by the response, and save it back to SyncState so subsequent
polls progress beyond the first 100 updates.
1 participant
Summary
Adds
TelegramSyncPipeline, implementing bothSyncPipelineandIncrementalSource, so the Composiotelegramtoolkit syncs end to end. Before this, connecting Telegram reportedACTIVEand then failed at sync withtinycortex sync does not support toolkit 'telegram'(the selector's_arm) — a source that looked healthy and ingested nothing.Pipeline
Message-shaped, modeled on
SlackSyncPipeline(the closest reference: chat messages, list-scopes-then-fetch-per-scope):TELEGRAM_GET_UPDATESpoll; distinctchat.ids become sync scopes (there is no bot API to enumerate a user's chats).TELEGRAM_GET_CHAT_HISTORY(offset-based paging,limitcapped at 100/request).document_id = telegram:<chat_id>:<message_id>,metadata.taint = "external_sync", and a per-chatmetadata.path_scope = telegram:chat:<chat_id>collection scope, so re-syncing upserts instead of duplicating (applies the lesson from fix(memory): use stable document_id as sync upsert key (fixes #4947 Bug 2 secret-guard sync failure) openhuman#4953; does not reintroduce per-run ids).tolerate_scope_errorsfences one bad chat from aborting the account;stop_on_empty_pendingdrives incremental convergence via the retained dedupe set (the chat-history action has no server-side "since" filter).Registered through
providers/mod.rs,composio/mod.rs, andsync/mod.rsexactly like the six existing pipelines.Tests run
cargo fmt --all+cargo test --features sync— all green (1270+ lib tests;sync::composiosuite 21 passed). New unit tests covertoolkit()/action(),extract_page()offset paging on a sample Composio JSON payload,dedup_key(), and a stabledocument()id/taint/scope.Part of #79 (tinycortex pipeline body; openhuman wiring is the follow-up step) — the openhuman side (bump the
vendor/tinycortexpointer, add thesync.rsselector arm, register theComposioProvidersomemory_sources.supported_toolkitsadvertises the slug) is a separate change and is intentionally not included here.🤖 Generated with Claude Code
Summary by CodeRabbit