Repository navigation
refactor(memory_sync): generic Composio sync orchestrator + convert Notion (Refs #3333) - #3852
Conversation
…otion (Refs tinyhumansai#3333) First increment of tinyhumansai#3333: replace each provider's bespoke ~600-1000-line sync() loop with one generic run_sync orchestrator over a slim IncrementalSource primitive trait, and convert Notion as the proof. The orchestrator owns the control flow every provider copy-pasted (and which is how the tinyhumansai#3304 page-granular cap bug landed in five providers): budget check -> preamble -> per-scope page loop bounded by ItemCap::max_pages -> sync_depth_days window -> dedup -> ItemCap::clamp_batch (the precise EXACT-N cap) -> ingest -> record -> cursor advance/hold -> save state -> SyncOutcome. Providers implement only primitives: preamble (resolve identity / list scopes), fetch_page, item_dedup_key, item_sort_ts, depth_floor, ingest. SyncScope unifies flat (gmail/github/notion/linear: one implicit scope) and nested (clickup workspaces, slack channels: one scope per container) shapes behind one scope loop, so the nested providers slot in without a control-flow change. Notion now delegates sync() to run_sync(&NotionSource); its dedup key ({page_id}@{edited_time}), RFC3339 depth floor, budget-counted NOTION_GET_PAGE_MARKDOWN body fetch, and bounded-concurrency ingest move verbatim into notion/source.rs — identical cursor/dedup/budget/window behavior. gmail/github/linear/clickup/slack keep their existing sync(). Tests: generic orchestrator unit tests via a test-only FakeSource (max_items caps to EXACTLY N, no-cap full page, sync_depth_days filters, nested scopes share one cap, budget-exhaustion skip, pure select_pending dedup/cursor); Notion's existing cap e2e now rides the shared path; new Notion sync_depth_days e2e. Co-Authored-By: Claude <noreply@anthropic.com>
|
Warning Review limit reached
More reviews will be available in 28 minutes and 9 seconds. Learn how PR review limits work. Your organization has used up its prepaid credits, and credit purchases are no longer available. Enable the review add-on in the billing tab to keep reviews running — you're only billed for reviews past your plan's rate limits ($0.25/file). ⌛ How to resolve this issue?After more reviews become available, a review can be triggered using the To avoid repeated limits, reduce automatic review volume by pausing incremental auto-reviews earlier, using label-based review opt-in, excluding WIP or generated PR titles, or requesting reviews manually when the PR is ready. If your team needs uninterrupted high-volume reviews, an organization admin can enable usage-based credits. 🚦 How do rate limits work?CodeRabbit enforces per-developer PR review limits for each organization. Most developers receive the normal plan refill rate. For paid Pro and Pro+ PR reviews, CodeRabbit uses adaptive limits for sustained high-volume activity. When a developer's recent PR review activity reaches the 95th percentile or higher among CodeRabbit users, the refill rate gradually slows as usage increases. The highest same-day bursts are limited more strictly. Please see our Fair Usage Limits Policy for further information. ℹ️ Review info⚙️ Run configurationConfiguration used: Organization UI Review profile: CHILL Plan: Pro Run ID: 📒 Files selected for processing (2)
📝 WalkthroughWalkthroughIntroduces a generic ChangesIncremental sync orchestrator and Notion adapter
Sequence Diagram(s)sequenceDiagram
participant NotionProvider
participant run_sync as run_sync (orchestrator)
participant NotionSource
participant ComposioAPI as Composio NOTION_FETCH_DATA
participant MarkdownAPI as Composio NOTION_GET_PAGE_MARKDOWN
participant MemoryTree as ingest_page_into_memory_tree
NotionProvider->>run_sync: run_notion_sync(ctx, reason)
run_sync->>run_sync: load SyncState, check budget
run_sync->>NotionSource: preamble(ctx, state)
NotionSource-->>run_sync: [flat SyncScope]
loop page loop
run_sync->>NotionSource: fetch_page(scope, cursor, reason)
NotionSource->>ComposioAPI: NOTION_FETCH_DATA (page_size, filter, sort, cursor)
ComposioAPI-->>NotionSource: items + next_cursor
NotionSource-->>run_sync: PageFetch
run_sync->>run_sync: select_pending (dedup, depth filter, max_items clamp)
run_sync->>NotionSource: ingest(scope, state, SyncItems)
NotionSource->>MarkdownAPI: ACTION_GET_PAGE_MARKDOWN (budget-capped, bounded concurrency)
MarkdownAPI-->>NotionSource: markdown bodies
NotionSource->>MemoryTree: ingest_pending_buffered (bounded concurrency)
MemoryTree-->>NotionSource: ok / err per page
NotionSource-->>run_sync: IngestOutcome
run_sync->>run_sync: advance or hold cursor, check cap
end
run_sync->>run_sync: persist SyncState + last_sync_at_ms
run_sync-->>NotionProvider: SyncOutcome
Estimated code review effort🎯 4 (Complex) | ⏱️ ~60 minutes Possibly related PRs
Suggested labels
🚥 Pre-merge checks | ✅ 5✅ Passed checks (5 passed)
✏️ Tip: You can configure your own custom pre-merge checks in the settings. Comment |
![chatgpt-codex-connector[bot]](https://avatars.githubusercontent.com/in/1144995?s=60&v=4){kind=small}
There was a problem hiding this comment.
💡 Codex Review
Here are some automated review suggestions for this pull request.
Reviewed commit: 31ff6270d9
ℹ️ About Codex in GitHub
Codex has been enabled to automatically review pull requests in this repo. Reviews are triggered when you
- Open a pull request for review
- Mark a draft as ready
- Comment "@codex review".
If Codex has suggestions, it will comment; otherwise it will react with 👍.
When you sign up for Codex through ChatGPT, Codex can also answer questions or update the PR, like "@codex address that feedback".
| .error | ||
| .clone() | ||
| .unwrap_or_else(|| "provider reported failure".to_string()); | ||
| return Err(format!("[composio:notion] {ACTION_FETCH_DATA}: {err}")); |
There was a problem hiding this comment.
When NOTION_FETCH_DATA completes but returns successful == false, this Err reaches the orchestrator before state.record_requests(1) runs, so the request is saved without consuming the daily Composio budget. The previous Notion loop, and the other provider loops I checked, record a completed response before checking successful, which prevents a broken/unauthorized connection from repeatedly making billable failed calls without ever hitting the per-day cap. Consider recording completed page requests before converting provider-reported failures into sync errors, or returning enough fetch metadata for the orchestrator to count them.
Useful? React with 👍 / 👎.
There was a problem hiding this comment.
Good catch — fixed in 34bdc7c. fetch_page now takes &mut SyncState and records the page request against the daily budget for any completed round-trip (including successful == false) before converting it to a sync error; a transport error (no round-trip) is still not counted. This restores parity with the per-provider loops. Added a unit test (provider_reported_page_failure_still_consumes_budget) pinning that a completed-but-failed page consumes exactly one budget request.
coderabbitai
Bot
added
memory
…mble/fetch errors) Add three orchestrator unit tests via FakeSource so the diff-cover gate sees the save-and-propagate / short-circuit branches: empty scopes -> skip outcome, preamble error -> propagate (after persisting spent budget), fetch_page error -> propagate. Co-Authored-By: Claude <noreply@anthropic.com>
![coderabbitai[bot]](https://avatars.githubusercontent.com/in/347564?s=60&v=4){kind=small}
There was a problem hiding this comment.
Actionable comments posted: 4
🧹 Nitpick comments (1)
src/openhuman/memory_sync/composio/providers/orchestrator.rs (1)
1-1: ⚡ Quick winMove the new orchestrator into a directory module.
orchestrator.rsis new Rust core functionality at the provider module root. Please move it to a dedicated module directory such asproviders/orchestrator/mod.rsso follow-up provider adapters do not continue adding root-level modules.As per coding guidelines, “In Rust core, create new functionality in dedicated subdirectories (
openhuman/<domain>/mod.rs+ siblings), never as root-level*.rsfiles.”🤖 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/openhuman/memory_sync/composio/providers/orchestrator.rs` at line 1, Move the orchestrator.rs file from the providers directory root into a dedicated orchestrator subdirectory following Rust module conventions. Create a new directory named orchestrator within the providers folder, move the orchestrator.rs file into it and rename it to mod.rs (so the path becomes providers/orchestrator/mod.rs). Update the module declaration in the parent providers module file (likely providers/mod.rs) to declare orchestrator as a module using mod orchestrator; instead of directly including the orchestrator.rs file, ensuring all public items remain properly exported for use by other modules.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.
Inline comments:
In `@src/openhuman/memory_sync/composio/providers/notion/source.rs`:
- Around line 107-140: The fetch_page function and related functions in the
157-283 range lack debug logging for diagnostics on the migrated Notion sync
flow. Add debug logs at the entry point of fetch_page (including non-sensitive
context fields like cursor_present, page_size, and connection_id), at successful
completion, and at error points. Similarly, add debug logs to body_fetch and
ingest functions for their entry, exit, and state transitions with non-sensitive
diagnostic fields like pending, fetch_count, persisted, and had_failures. Use
stable, grep-friendly log message prefixes for all new logs and ensure no
secrets or PII are logged, following the project's log/tracing guidelines.
In `@src/openhuman/memory_sync/composio/providers/orchestrator.rs`:
- Around line 52-54: The `SyncScope::label` field is being logged in tracing
statements as `scope = %scope.label` across multiple locations in the
orchestrator loop, but this can leak PII since labels can contain Slack channel
or ClickUp workspace names. Replace all instances of `scope = %scope.label` in
the tracing fields (at the specified line ranges: 396-400, 422-426, 471-475,
482-486, and the location in the diff) with an opaque identifier such as
`scope_id` or a hashed value instead of the human-readable label. Ensure that
the `label` field is never included in generic tracing output to comply with the
guideline of not logging PII.
- Around line 408-418: The state.record_requests(1) call only increments the
request counter after a successful fetch_page() call, allowing failed page fetch
attempts to bypass the daily request budget. Move the state.record_requests(1)
call so that it executes for both successful and failed fetch_page() attempts,
ensuring every request attempt (whether it succeeds or fails) is counted against
the daily budget limit to prevent repeated failed requests from hammering the
upstream provider.
- Around line 431-447: The depth floor cutoff is applied after the dedup logic
in select_pending, which means items below the floor that are already synced or
missing dedup keys won't be caught in the truncate check, allowing pagination to
continue fetching older pages outside sync_depth_days. Move the depth floor
filtering to occur before the select_pending call by filtering fetch.items to
remove any items with sort_ts below the floor before passing them to
select_pending, ensuring pagination stops appropriately regardless of dedup
status.
---
Nitpick comments:
In `@src/openhuman/memory_sync/composio/providers/orchestrator.rs`:
- Line 1: Move the orchestrator.rs file from the providers directory root into a
dedicated orchestrator subdirectory following Rust module conventions. Create a
new directory named orchestrator within the providers folder, move the
orchestrator.rs file into it and rename it to mod.rs (so the path becomes
providers/orchestrator/mod.rs). Update the module declaration in the parent
providers module file (likely providers/mod.rs) to declare orchestrator as a
module using mod orchestrator; instead of directly including the orchestrator.rs
file, ensuring all public items remain properly exported for use by other
modules.
🪄 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: 0b314985-1820-4818-8382-4fc1bdd84add
📒 Files selected for processing (6)
src/openhuman/memory_sync/composio/providers/mod.rssrc/openhuman/memory_sync/composio/providers/notion/mod.rssrc/openhuman/memory_sync/composio/providers/notion/provider.rssrc/openhuman/memory_sync/composio/providers/notion/source.rssrc/openhuman/memory_sync/composio/providers/orchestrator.rstests/memory_sync_providers_raw_coverage_e2e.rs
| async fn fetch_page( | ||
| &self, | ||
| ctx: &ProviderContext, | ||
| _scope: &SyncScope, | ||
| cursor: Option<&str>, | ||
| reason: SyncReason, | ||
| ) -> Result<PageFetch, String> { | ||
| let mut args = json!({ | ||
| "page_size": self.page_size(reason), | ||
| "filter": { "value": "page", "property": "object" }, | ||
| "sort": { "direction": "descending", "timestamp": "last_edited_time" } | ||
| }); | ||
| if let Some(cursor) = cursor { | ||
| args["start_cursor"] = json!(cursor); | ||
| } | ||
|
|
||
| let resp = ctx | ||
| .execute(ACTION_FETCH_DATA, Some(args)) | ||
| .await | ||
| .map_err(|e| format!("[composio:notion] {ACTION_FETCH_DATA}: {e:#}"))?; | ||
|
|
||
| if !resp.successful { | ||
| let err = resp | ||
| .error | ||
| .clone() | ||
| .unwrap_or_else(|| "provider reported failure".to_string()); | ||
| return Err(format!("[composio:notion] {ACTION_FETCH_DATA}: {err}")); | ||
| } | ||
|
|
||
| Ok(PageFetch { | ||
| items: sync::extract_results(&resp.data), | ||
| next: sync::extract_notion_cursor(&resp.data), | ||
| }) | ||
| } |
There was a problem hiding this comment.
Add required debug diagnostics around the new Notion sync flow.
The migrated fetch_page/body-fetch/ingest path logs failures, but not entry/exit, successful external calls, or state transitions with correlation fields. Add debug/trace logs for page fetch start/complete, body fetch start/complete, ingest start/complete, and include non-sensitive fields like connection_id, cursor_present, page_size, pending, fetch_count, persisted, and had_failures.
Suggested diagnostic shape
async fn fetch_page(
&self,
ctx: &ProviderContext,
_scope: &SyncScope,
cursor: Option<&str>,
reason: SyncReason,
) -> Result<PageFetch, String> {
+ let connection_id = ctx.connection_id.as_deref().unwrap_or("default");
+ let page_size = self.page_size(reason);
+ tracing::debug!(
+ connection_id,
+ cursor_present = cursor.is_some(),
+ page_size,
+ "[composio:notion] fetch_page start"
+ );
let mut args = json!({
- "page_size": self.page_size(reason),
+ "page_size": page_size,
"filter": { "value": "page", "property": "object" },
"sort": { "direction": "descending", "timestamp": "last_edited_time" }
});
@@
- Ok(PageFetch {
- items: sync::extract_results(&resp.data),
- next: sync::extract_notion_cursor(&resp.data),
- })
+ let items = sync::extract_results(&resp.data);
+ let next = sync::extract_notion_cursor(&resp.data);
+ tracing::debug!(
+ connection_id,
+ items = items.len(),
+ has_next = next.is_some(),
+ "[composio:notion] fetch_page complete"
+ );
+ Ok(PageFetch { items, next })
} if fetch_count > 0 {
@@
.map(|page_id| {
- let ctx = &ctx;
+ let ctx = ctx;
+ let connection_id = connection_id;
async move {
+ tracing::debug!(
+ connection_id,
+ page_id = %page_id,
+ "[composio:notion] GET_PAGE_MARKDOWN start"
+ );
match ctx
@@
Ok(resp) if resp.successful => {
let markdown = sync::extract_page_markdown(&resp.data);
+ tracing::debug!(
+ connection_id,
+ page_id = %page_id,
+ has_markdown = markdown.is_some(),
+ "[composio:notion] GET_PAGE_MARKDOWN complete"
+ );
if markdown.is_none() {As per coding guidelines, “Include debug logging with verbose diagnostics on all new/changed flows: log entry/exit, branches, external calls, retries/timeouts, state transitions, and errors with stable grep-friendly prefixes” and “Use Rust log/tracing at debug/trace levels for diagnostics; never log secrets or full PII”.
Also applies to: 157-283
🤖 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/openhuman/memory_sync/composio/providers/notion/source.rs` around lines
107 - 140, The fetch_page function and related functions in the 157-283 range
lack debug logging for diagnostics on the migrated Notion sync flow. Add debug
logs at the entry point of fetch_page (including non-sensitive context fields
like cursor_present, page_size, and connection_id), at successful completion,
and at error points. Similarly, add debug logs to body_fetch and ingest
functions for their entry, exit, and state transitions with non-sensitive
diagnostic fields like pending, fetch_count, persisted, and had_failures. Use
stable, grep-friendly log message prefixes for all new logs and ensure no
secrets or PII are logged, following the project's log/tracing guidelines.
Source: Coding guidelines
| /// Human-readable label for logs. Never logged at a level that would leak | ||
| /// PII — ids/labels here are container identifiers, not user content. | ||
| pub label: String, |
There was a problem hiding this comment.
Don’t emit human-readable scope labels in generic logs.
SyncScope::label can become a Slack channel or ClickUp workspace name, and the orchestrator logs it on branch paths. Use an opaque scope_id, scope index, or hashed value instead, and keep label out of generic tracing fields.
Suggested adjustment
- /// Human-readable label for logs. Never logged at a level that would leak
- /// PII — ids/labels here are container identifiers, not user content.
+ /// Human-readable label for provider-local messages. Do not emit this from
+ /// generic orchestrator logs; nested providers may supply user-defined names.
pub label: String,- scope = %scope.label,
+ scope_id = %scope.id,
+ scope_kind = if scope.id.is_empty() { "flat" } else { "nested" },Apply the same replacement to the other scope = %scope.label tracing fields in this loop.
As per coding guidelines, “Use Rust log/tracing at debug/trace levels for diagnostics; never log secrets or full PII.”
Also applies to: 396-400, 422-426, 471-475, 482-486
🤖 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/openhuman/memory_sync/composio/providers/orchestrator.rs` around lines 52
- 54, The `SyncScope::label` field is being logged in tracing statements as
`scope = %scope.label` across multiple locations in the orchestrator loop, but
this can leak PII since labels can contain Slack channel or ClickUp workspace
names. Replace all instances of `scope = %scope.label` in the tracing fields (at
the specified line ranges: 396-400, 422-426, 471-475, 482-486, and the location
in the diff) with an opaque identifier such as `scope_id` or a hashed value
instead of the human-readable label. Ensure that the `label` field is never
included in generic tracing output to comply with the guideline of not logging
PII.
Source: Coding guidelines
| // Dedup + cursor-boundary detection + newest-ts tracking. | ||
| let (mut pending, mut hit_cursor_boundary) = | ||
| select_pending(source, &fetch.items, &state, &mut newest_ts); | ||
|
|
||
| // sync_depth_days: `pending` is in descending-timestamp order, so | ||
| // truncate at the first item below the floor and stop paginating. | ||
| if let Some(ref floor) = depth_floor { | ||
| if let Some(cut) = pending.iter().position(|it| { | ||
| it.sort_ts | ||
| .as_deref() | ||
| .map(|t| t < floor.as_str()) | ||
| .unwrap_or(false) | ||
| }) { | ||
| pending.truncate(cut); | ||
| hit_cursor_boundary = true; | ||
| } | ||
| } |
There was a problem hiding this comment.
Apply the depth cutoff before dedup skips items.
The cutoff scans only pending, so a below-floor raw item that was already synced or missing a dedup key will not stop pagination. That can keep fetching older pages outside sync_depth_days, and those below-floor rows can update newest_ts before being dropped.
Suggested fix
- // Dedup + cursor-boundary detection + newest-ts tracking.
- let (mut pending, mut hit_cursor_boundary) =
- select_pending(source, &fetch.items, &state, &mut newest_ts);
-
- // sync_depth_days: `pending` is in descending-timestamp order, so
- // truncate at the first item below the floor and stop paginating.
- if let Some(ref floor) = depth_floor {
- if let Some(cut) = pending.iter().position(|it| {
- it.sort_ts
- .as_deref()
- .map(|t| t < floor.as_str())
- .unwrap_or(false)
- }) {
- pending.truncate(cut);
- hit_cursor_boundary = true;
- }
- }
+ // sync_depth_days: raw provider items are descending by timestamp,
+ // so cut the page before dedup can hide the boundary item.
+ let mut hit_depth_boundary = false;
+ let page_items = if let Some(ref floor) = depth_floor {
+ if let Some(cut) = fetch.items.iter().position(|item| {
+ source
+ .item_sort_ts(item)
+ .as_deref()
+ .map(|t| t < floor.as_str())
+ .unwrap_or(false)
+ }) {
+ hit_depth_boundary = true;
+ &fetch.items[..cut]
+ } else {
+ fetch.items.as_slice()
+ }
+ } else {
+ fetch.items.as_slice()
+ };
+
+ // Dedup + cursor-boundary detection + newest-ts tracking.
+ let (mut pending, mut hit_cursor_boundary) =
+ select_pending(source, page_items, &state, &mut newest_ts);
+ if hit_depth_boundary {
+ hit_cursor_boundary = true;
+ }🤖 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/openhuman/memory_sync/composio/providers/orchestrator.rs` around lines
431 - 447, The depth floor cutoff is applied after the dedup logic in
select_pending, which means items below the floor that are already synced or
missing dedup keys won't be caught in the truncate check, allowing pagination to
continue fetching older pages outside sync_depth_days. Move the depth floor
filtering to occur before the select_pending call by filtering fetch.items to
remove any items with sort_ts below the floor before passing them to
select_pending, ensuring pagination stops appropriately regardless of dedup
status.
Addresses Codex review (P2): in the orchestrator, fetch_page now records the page request against the daily budget for any *completed* round-trip — incl. one the provider reports as successful==false — before converting it to a sync error. A transport error (no round-trip) is still not counted. This restores parity with the per-provider loops, where a broken/unauthorized connection could not make unlimited billable failed page calls without hitting the per-day cap. fetch_page gains a &mut SyncState param (the primitive records the calls it makes, like ingest's body fetch already does); the orchestrator no longer records per page. Adds a unit test pinning that a provider-reported page failure consumes exactly one budget request. Co-Authored-By: Claude <noreply@anthropic.com>
…chestrator (Refs tinyhumansai#3333, stacked on tinyhumansai#3852) (tinyhumansai#3860)\n\n
Summary
run_sync) over a slimIncrementalSourceprimitive trait, so every Composio provider gets themax_itemscap, thesync_depth_dayswindow, dedup, cursor advance, and the daily budget enforced in one place.ItemCap(PR feat(memory-sources): default sync sources on with conservative caps, per-source settings & All In #3304) centralised the cap math; this centralises the control flow.run_sync, with behavioral parity (cursor, dedup, budget,sync_depth_days, thread/account handling identical).SyncScopeabstraction unifies flat (gmail/github/notion/linear: one implicit scope) and nested (clickup workspaces, slack channels: one scope per container) shapes behind a single scope loop — proven extensible even though only a flat provider converts now.sync(); no per-provider cap/window/pagination code is deleted yet.Problem
#3333: each provider hand-rolls the same ~600–1000-line
sync()loop (loadSyncState→ budget check → preamble → page loop bounded byItemCap::max_pages→select_pendingdedup →sync_depth_daystruncate →clamp_batch→ bounded-concurrency ingest →mark_synced→ cursor advance/hold → save →SyncOutcome). The duplication is a correctness liability (#3304) and blocks new toolkits from inheriting caps+window for free.Solution
IncrementalSourceprimitives (provider-specific only):preamble(resolve identity / list scopes),fetch_page(scope, cursor) -> (items, next),item_dedup_key,item_sort_ts,depth_floor,ingest(items), plushold_cursor_on_ingest_failure.run_syncowns: budget pre-check →preamble→ per-scope page loop bounded byItemCap::max_pages→ genericselect_pending(dedup + cursor-boundary + newest-ts tracking) →sync_depth_daystruncate →ItemCap::clamp_batch(the precise EXACT-N cap that fixes #3304) →ingest→record→is_reached()stop → cursor advance/hold → save state →SyncOutcome.Scopes cover flat AND nested in one loop:
SyncScope { id, label }. Flat providers returnvec![SyncScope::flat()]; nested providers resolve containers inpreambleand return one scope each.for scope { for page {…} }is identical for both, so clickup (workspaces→tasks) and slack (channels→history, with the user-list backfill as apreambleside-effect) slot in without a control-flow change.Notion parity:
NotionProvider::syncdelegates torun_sync(&NotionSource). The{page_id}@{edited_time}dedup key, RFC3339 depth floor, budget-countedNOTION_GET_PAGE_MARKDOWNbody fetch, and bounded-concurrency ingest move verbatim intonotion/source.rs.fetch_user_profile/fetch_tasks/list_databases/on_triggerare unchanged.Submission Checklist
FakeSource(max_items caps to EXACTLY N, no-cap full page,sync_depth_daysfilters, nested scopes share one cap, budget-exhaustion skip, pureselect_pendingdedup/cursor); Notion's existing cap e2e now rides the shared path; new Notionsync_depth_dayse2e.fetch_page→ orchestrator →ingest/body-fetch end-to-end). Ran focusedcargo testforcomposio::providers::orchestrator/notionand--test memory_sync_providers_raw_coverage_e2elocally — all green.N/A: behaviour-only refactor(no new user-facing feature row; cap/window behaviour unchanged).## Related—N/A: behaviour-only refactor.N/A: internal sync refactor, no release-cut surface.Closes #NNN: this is the first increment of a multi-PR migration, so it references (not closes) Generic Composio sync orchestrator so every toolkit enforces max_items and sync_depth_days #3333. See migration checklist below.Impact
last_sync_at_mson Notion syncs (previously unset; Notion never read it — no behavior change).Migration checklist (deferred to follow-up PRs, Refs #3333)
IncrementalSource.IncrementalSource.depth_flooroverride,hold_cursor_on_ingest_failure = false.preamblehook).run_sync.composio/mod.rs:149.Related
AI Authored PR Metadata (required for Codex/Linear PRs)
Linear Issue
Commit & Branch
Validation Run
pnpm --filter openhuman-app format:check—N/A: no frontend changespnpm typecheck—N/A: no frontend changescargo test --lib -- composio::providers::orchestrator composio::providers::notion(54 passed) +cargo test --test memory_sync_providers_raw_coverage_e2e -- --test-threads=1(13 passed)cargo fmt --checkclean;cargo check --manifest-path Cargo.tomlcleanN/A: no Tauri/src-tauri changesValidation Blocked
command:N/Aerror:N/Aimpact:N/ABehavior Changes
Parity Contract
sync_depth_days/body-fetch/ingest identical; existing Notion e2e green on the new path.SyncStatebefore propagating a page error (matches per-provider loops); cursor held on cap-truncation and on ingest failure (Notion's delete-first safety).Duplicate / Superseded PR Handling
Summary by CodeRabbit
Release Notes
Refactor
Tests