feat(sync): add Composio Outlook memory-sync pipeline

[YellowSnnowmann]

Summary

Adds OutlookSyncPipeline for the Composio outlook toolkit (Microsoft Outlook mail), message-shaped, modeled on the Gmail pipeline. Previously advertised but unsyncable (#106).

• Single action OUTLOOK_LIST_MESSAGES (verified against catalogs_productivity.rs). receivedDateTime cursor, filter depth window. Stable upsert key outlook:<id>, taint = external_sync, content-free logging.
• Registered through providers/mod.rs, composio/mod.rs, sync/mod.rs.

Disclosed assumption: the catalog pins the slug but not arg schemas, so the pagination token key (skip_token) and depth filter key are documented // NOTE: assumptions using Microsoft Graph's native parameter names. Pagination is driven off the response-side next-token (@odata.nextLink) read via multiple pointer fallbacks, so a wrong request-side key degrades gracefully to a single-page fetch — never an error or data corruption.

Pipeline body only (step 1 of 2); openhuman wiring closes the issue end to end.

API Or Behavior Changes

One new public SyncPipeline type exported from memory::sync. Additive.

Tests

• cargo fmt --check
• cargo clippy --all-targets -- -D warnings
• cargo build --all-targets
• cargo test (all-features green; new mock test drives 2-page pagination via the real token key, asserts stable document_id, cursor persistence, idempotent re-sync)

Documentation

Module/item docs including the arg-assumption notes. No external docs needed.

Part of #99 · tracker #106

Summary by CodeRabbit

• New Features

  • Added Microsoft Outlook synchronization with incremental message fetching.
  • Supports pagination, date filtering, configurable page and result limits, and duplicate prevention.
  • Makes Outlook synchronization available through the public synchronization interface.

• Bug Fixes

  • Preserves synchronization cursors and prevents duplicate imports across repeated runs.
  • Improves handling of authentication, pagination tokens, and external documents during Outlook imports.

• Tests

  • Added coverage for pagination, cursor persistence, idempotent synchronization, document handling, and authentication.

[coderabbitai]

No actionable comments were generated in the recent review. 🎉

ℹ️ Recent review info

⚙️ Run configuration

Configuration used: Organization UI

Review profile: CHILL

Plan: Pro

Run ID: 0fe0b540-4463-4b4a-81cc-32343fe4427e

📥 Commits

Reviewing files that changed from the base of the PR and between f869593 and b3de142.

📒 Files selected for processing (4)

• src/memory/sync/composio/mod.rs
• src/memory/sync/composio/providers/mod.rs
• src/memory/sync/mod.rs
• tests/composio_sync_mock.rs

🚧 Files skipped from review as they are similar to previous changes (4)

• src/memory/sync/mod.rs
• src/memory/sync/composio/providers/mod.rs
• src/memory/sync/composio/mod.rs
• tests/composio_sync_mock.rs

---

📝 Walkthrough

Walkthrough

This PR adds an incremental Outlook synchronization pipeline for Composio. It supports paginated message fetching, date filtering, cursor-based deduplication, skip-token normalization, document conversion, public exports, and mock-based integration tests.

Changes

Outlook Sync Pipeline

Layer / File(s)	Summary
Outlook pipeline core implementation src/memory/sync/composio/providers/outlook.rs	Defines OutlookSyncPipeline, configures incremental fetching, handles pagination and response extraction, deduplicates messages, and builds skill documents.
Public export wiring src/memory/sync/composio/providers/mod.rs, src/memory/sync/composio/mod.rs, src/memory/sync/mod.rs	Registers the Outlook provider and re-exports OutlookSyncPipeline through the synchronization module hierarchy.
Outlook sync mock tests tests/composio_sync_mock.rs	Tests paginated responses, authentication, document ingestion, cursor persistence, synced-ID persistence, and idempotent reruns.

Estimated code review effort: 3 (Moderate) | ~25 minutes

Possibly related PRs

• tinyhumansai/tinycortex#134: Adds a comparable paginated Composio pipeline with cursor handling and mock tests.
• tinyhumansai/tinycortex#135: Adds another Composio provider pipeline with corresponding registration and mock tests.
• tinyhumansai/tinycortex#137: Adds and exports a provider-specific Composio synchronization pipeline.

Sequence Diagram(s)

sequenceDiagram
participant Caller
participant OutlookSyncPipeline
participant ComposioClient
participant OutlookAPI
Caller->>OutlookSyncPipeline: tick()
OutlookSyncPipeline->>ComposioClient: Request messages with cursor and skip token
ComposioClient->>OutlookAPI: Fetch page
OutlookAPI-->>ComposioClient: Messages and next link
ComposioClient-->>OutlookSyncPipeline: Return response payload
OutlookSyncPipeline->>OutlookSyncPipeline: Deduplicate and build documents
OutlookSyncPipeline-->>Caller: Return synchronized documents and cursor

Loading

Poem

A rabbit sorts the Outlook flow,
Page by page, the messages go.
Cursors keep each record right,
Skip tokens guide the next hop's flight. 🐇

🚥 Pre-merge checks | ✅ 5

✅ Passed checks (5 passed)

Check name	Status	Explanation
Description Check	✅ Passed	Check skipped - CodeRabbit’s high-level summary is enabled.
Title check	✅ Passed	The title clearly and concisely identifies the new Composio Outlook memory-sync pipeline, which is the main change.
Docstring Coverage	✅ Passed	No functions found in the changed files to evaluate docstring coverage. Skipping docstring coverage check.
Linked Issues check	✅ Passed	Check skipped because no linked issues were found for this pull request.
Out of Scope Changes check	✅ Passed	Check skipped because no linked issues were found for this pull request.

---

_{Comment @coderabbitai help to get the list of available commands.}

[coderabbitai]

🧹 Nitpick comments (1)

src/memory/sync/composio/providers/outlook.rs (1)

30-44: 📐 Maintainability & Code Quality | 🔵 Trivial | ⚡ Quick win

Document the public constructors.

Add item documentation for new and with_limits. State that with_limits clamps each value below one to one.

Proposed documentation

 impl OutlookSyncPipeline {
+ /// Creates an Outlook synchronization pipeline for one Composio connection.
pub fn new(client: ComposioClient, connection_id: impl Into<String>) -> Self {
Self {
client,
connection_id: connection_id.into(),
max_pages: 10,
page_size: 25,
}
}
+ /// Sets the maximum page count and page size.+ ///+ /// Values below one are clamped to one.
pub fn with_limits(mut self, max_pages: usize, page_size: usize) -> Self {

As per coding guidelines, “Document public APIs, module contracts, and non-obvious behavior thoroughly, preferring module-level docs and item docs.”

🤖 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/outlook.rs` around lines 30 - 44, Add
item-level documentation to the public OutlookSyncPipeline constructors new and
with_limits, describing their purpose and parameters. Explicitly document that
with_limits clamps max_pages and page_size values below one to one, while
preserving the existing implementation.

Source: Coding guidelines

🤖 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.
Nitpick comments:
In `@src/memory/sync/composio/providers/outlook.rs`:
- Around line 30-44: Add item-level documentation to the public
OutlookSyncPipeline constructors new and with_limits, describing their purpose
and parameters. Explicitly document that with_limits clamps max_pages and
page_size values below one to one, while preserving the existing implementation.

---

ℹ️ Review info

⚙️ Run configuration

Configuration used: Organization UI

Review profile: CHILL

Plan: Pro

Run ID: 7a247310-4072-48a4-ba6f-2f835243621d

📥 Commits

Reviewing files that changed from the base of the PR and between 115e4bb and 0e23335.

📒 Files selected for processing (5)

• src/memory/sync/composio/mod.rs
• src/memory/sync/composio/providers/mod.rs
• src/memory/sync/composio/providers/outlook.rs
• src/memory/sync/mod.rs
• tests/composio_sync_mock.rs

[greptile-apps]

Greptile Summary

Adds OutlookSyncPipeline, an incremental Microsoft Outlook mail sync via Composio's OUTLOOK_LIST_MESSAGES action, modeled on the Gmail pipeline with a receivedDateTime cursor, OData $filter depth window, and normalize_skip_token to reduce full @odata.nextLink URLs to bare skip tokens before feeding them back as pagination args.

• New outlook.rs provider implementing IncrementalSource: stable outlook:<id> document key, receivedDateTime-only cursor (no fallback field to avoid filter/cursor mismatch), and graceful single-page degradation if Composio's arg key for the skip token differs from the assumed skip_token.
• normalize_skip_token correctly handles real Graph URL format (case-insensitive $skiptoken= search on ASCII-lowercased copy, original bytes used for slicing); test mock now uses a live-format nextLink URL so page 2 is only reachable via the extracted token.
• Registered additively through providers/mod.rs, composio/mod.rs, and sync/mod.rs; no existing exports changed.

Confidence Score: 5/5

Safe to merge — the change is purely additive, all existing exports and tests are untouched, and the new pipeline matches the patterns established by the Google Calendar and Gmail pipelines.

The normalize_skip_token logic is correct and is now directly exercised by a test that uses a realistic full Graph @odata.nextLink URL, so the key concern from the previous review thread is properly addressed. The cursor field is receivedDateTime-only, matching the OData filter, closing the prior field-mismatch issue. The filter+pagination-token co-sending pattern is consistent with the existing Google Calendar and Google Drive pipelines.

Files Needing Attention: No files require special attention.

Important Files Changed

Filename	Overview
src/memory/sync/composio/providers/outlook.rs	New OutlookSyncPipeline implementing IncrementalSource; normalize_skip_token correctly extracts bare skip tokens from full Graph @odata.nextLink URLs, cursor/filter field consistency is clean, and all methods follow established pipeline patterns.
tests/composio_sync_mock.rs	Adds outlook_paginates_persists_cursor_and_is_idempotent integration test; mock now uses a realistic full Graph @odata.nextLink URL, verifying normalize_skip_token extracts the bare token for the page-2 request.
src/memory/sync/composio/mod.rs	Additive re-export of OutlookSyncPipeline; no structural changes.
src/memory/sync/composio/providers/mod.rs	Additive module declaration and re-export; alphabetical insertion is correct.
src/memory/sync/mod.rs	Additive public re-export of OutlookSyncPipeline into the memory::sync namespace.

Sequence Diagram

sequenceDiagram
participant Caller
participant OutlookSyncPipeline
participant run_incremental_sync
participant ComposioClient
participant SyncState
Caller->>OutlookSyncPipeline: tick(config, context)
OutlookSyncPipeline->>run_incremental_sync: delegate
run_incremental_sync->>SyncState: "load(toolkit=outlook, connection_id)"
run_incremental_sync->>ComposioClient: "execute(OUTLOOK_LIST_MESSAGES, {top, orderby, filter?})"
ComposioClient-->>run_incremental_sync: "{data: {value:[...], @odata.nextLink: https://...?$skiptoken=XYZ}}"
run_incremental_sync->>OutlookSyncPipeline: extract_page(data)
Note over OutlookSyncPipeline: normalize_skip_token reduces full URL to bare XYZ
OutlookSyncPipeline-->>run_incremental_sync: "PageFetch{items, next=Some(XYZ)}"
run_incremental_sync->>ComposioClient: "execute(OUTLOOK_LIST_MESSAGES, {top, orderby, filter?, skip_token=XYZ})"
ComposioClient-->>run_incremental_sync: "{data: {value:[]}} no nextLink"
run_incremental_sync->>SyncState: advance_cursor(newest receivedDateTime)
run_incremental_sync->>SyncState: save()
run_incremental_sync-->>Caller: "SyncOutcome{records_ingested, actions_called}"

Loading

_{Reviews (3): Last reviewed commit: "Merge main into feat/composio-outlook-sy..." | Re-trigger Greptile}

[coderabbitai]

Caution

Some comments are outside the diff and can’t be posted inline due to platform limitations.

⚠️ Outside diff range comments (1)

src/memory/sync/composio/providers/outlook.rs (1)

91-112: 🗄️ Data Integrity & Integration | 🟠 Major | ⚡ Quick win

Use the OUTLOOK_LIST_MESSAGES schema for incremental sync.

• Send orderby as an array and use received_date_time_ge for the cursor or depth filter.
• Replace unsupported filter and skip_token arguments with page_token, passing next_page_token through unchanged.
• Update normalize_skip_token and the mock to model next_page_token and /arguments/page_token.

🤖 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/outlook.rs` around lines 91 - 112, The
Outlook incremental-sync request construction in `outlook.rs` must match the
`OUTLOOK_LIST_MESSAGES` schema: send `orderby` as an array, use
`received_date_time_ge` for cursor and depth filters, and replace `skip_token`
with `page_token`, passing `next_page_token` through unchanged. Update
`normalize_skip_token` and the mock in `tests/composio_sync_mock.rs` to model
`next_page_token` and `/arguments/page_token` consistently; apply these changes
at the listed Outlook provider and mock sites.

🤖 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.
Outside diff comments:
In `@src/memory/sync/composio/providers/outlook.rs`:
- Around line 91-112: The Outlook incremental-sync request construction in
`outlook.rs` must match the `OUTLOOK_LIST_MESSAGES` schema: send `orderby` as an
array, use `received_date_time_ge` for cursor and depth filters, and replace
`skip_token` with `page_token`, passing `next_page_token` through unchanged.
Update `normalize_skip_token` and the mock in `tests/composio_sync_mock.rs` to
model `next_page_token` and `/arguments/page_token` consistently; apply these
changes at the listed Outlook provider and mock sites.

---

ℹ️ Review info

⚙️ Run configuration

Configuration used: Organization UI

Review profile: CHILL

Plan: Pro

Run ID: cd44df11-331e-4fc5-b360-d6ac220ef17e

📥 Commits

Reviewing files that changed from the base of the PR and between 0e23335 and 2e71925.

📒 Files selected for processing (2)

• src/memory/sync/composio/providers/outlook.rs
• tests/composio_sync_mock.rs

[greptile-apps]

YellowSnnowmann has reached the 50-credit limit for trial accounts. To continue receiving code reviews, upgrade your plan.