From c5e7f6140d814be57f970e4273782e467b41b0ce Mon Sep 17 00:00:00 2001 From: Nocturnal Date: Mon, 24 Aug 2026 17:53:02 +0700 Subject: [PATCH] Add a standing Gmail search filter that scopes background sync MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit A host can now pass an optional Gmail search query (e.g. 'label:brain') that the Gmail sync pipeline ANDs onto every page fetch, so background ingestion only stores matching messages while staying incremental: - ComposioMode (api host config): new gmail_sync_query field for the host to populate from its own config. - ComposioSyncConfig.gmail_query, plumbed through composio_config() and build_composio_pipeline(). - GmailSyncPipeline::with_filter(): a standing clause composed with the incremental after: / sync_depth_days clause — unlike with_query(), which keeps its replace semantics for backfills. Lets a privacy-conscious user admit mail to memory by labeling it in Gmail instead of syncing the entire inbox window. Co-Authored-By: Claude Fable 5 Claude-Session: https://claude.ai/code/session_01UMNxXS5ucxpzNoHnuhyQPu --- crates/tinymemory-api/src/host/config.rs | 4 + .../src/sync/pipelines/composio/gmail.rs | 34 ++++++- .../sync/pipelines/composio/gmail_tests.rs | 90 ++++++++++++++++++- .../src/sync/pipelines/host.rs | 17 +++- .../src/sync/pipelines/traits.rs | 4 + .../composio_gmail_non_tinycortex_e2e.rs | 1 + 6 files changed, 145 insertions(+), 5 deletions(-) diff --git a/crates/tinymemory-api/src/host/config.rs b/crates/tinymemory-api/src/host/config.rs index 040afb8c..b8f2f551 100644 --- a/crates/tinymemory-api/src/host/config.rs +++ b/crates/tinymemory-api/src/host/config.rs @@ -61,6 +61,10 @@ pub struct ComposioMode { pub api_key: Option, /// Whether the LLM triage turn is switched off for all triggers. pub triage_disabled: bool, + /// Optional Gmail search query scoping the background Gmail sync to + /// matching messages only (full Gmail search syntax, e.g. `label:brain`). + /// `None`/empty = the whole inbox window. On-demand access is unaffected. + pub gmail_sync_query: Option, } impl std::fmt::Debug for ComposioMode { diff --git a/crates/tinymemory-core/src/sync/pipelines/composio/gmail.rs b/crates/tinymemory-core/src/sync/pipelines/composio/gmail.rs index de04774f..21a4cb6d 100644 --- a/crates/tinymemory-core/src/sync/pipelines/composio/gmail.rs +++ b/crates/tinymemory-core/src/sync/pipelines/composio/gmail.rs @@ -26,6 +26,11 @@ pub struct GmailSyncPipeline { max_pages: usize, page_size: usize, query_override: Option, + /// Standing Gmail search filter (e.g. `label:brain`) ANDed onto every + /// fetch, *including* the incremental `after:` clause — unlike + /// [`Self::with_query`], which replaces the incremental clause outright + /// (backfill semantics). + filter: Option, } impl GmailSyncPipeline { @@ -55,6 +60,7 @@ impl GmailSyncPipeline { // needing more throughput can raise it via `with_limits`. page_size: 25, query_override: None, + filter: None, } } @@ -68,6 +74,16 @@ impl GmailSyncPipeline { self.query_override = Some(query.into()); self } + + /// Set a standing Gmail search filter (e.g. `label:brain`). Every page + /// fetch ANDs it with the incremental clause (`after:` / + /// `sync_depth_days`), so background sync stays incremental while only + /// matching messages are ingested. Contrast [`Self::with_query`], which + /// *replaces* the incremental clause (backfill semantics). + pub fn with_filter(mut self, filter: impl Into) -> Self { + self.filter = Some(filter.into()); + self + } } #[async_trait] @@ -144,19 +160,31 @@ impl IncrementalSource for GmailSyncPipeline { if let Some(token) = page { arguments["page_token"] = serde_json::json!(token); } + // Gmail search ANDs space-separated clauses, so the standing filter + // (`label:brain`) composes with whichever incremental clause applies. + let mut clauses: Vec = Vec::new(); + if let Some(filter) = self.filter.as_deref() { + let filter = filter.trim(); + if !filter.is_empty() { + clauses.push(filter.to_string()); + } + } if let Some(query) = self.query_override.as_deref() { - arguments["query"] = Value::String(query.into()); + clauses.push(query.to_string()); } else if let Some(cursor) = state.cursor.as_deref() { - arguments["query"] = serde_json::json!(format!( + clauses.push(format!( "after:{}", cursor_to_seconds(cursor).unwrap_or_default() )); } else if let Some(days) = config.sync_depth_days { - arguments["query"] = serde_json::json!(format!( + clauses.push(format!( "after:{}", (chrono::Utc::now() - chrono::Duration::days(days as i64)).timestamp() )); } + if !clauses.is_empty() { + arguments["query"] = Value::String(clauses.join(" ")); + } arguments } diff --git a/crates/tinymemory-core/src/sync/pipelines/composio/gmail_tests.rs b/crates/tinymemory-core/src/sync/pipelines/composio/gmail_tests.rs index d4de7681..e33ca480 100644 --- a/crates/tinymemory-core/src/sync/pipelines/composio/gmail_tests.rs +++ b/crates/tinymemory-core/src/sync/pipelines/composio/gmail_tests.rs @@ -1,8 +1,96 @@ -//! Tests for the Gmail message → canonical Markdown adapter. +//! Tests for the Gmail message → canonical Markdown adapter and the +//! query-clause composition in [`GmailSyncPipeline::arguments`]. + +use std::sync::Arc; use serde_json::json; +use super::GmailSyncPipeline; use super::{canonical_markdown, message_body, message_recipients, message_sent_at}; +use crate::sync::pipelines::composio::client::{ActionExecutor, ExecuteResponse}; +use crate::sync::pipelines::composio::gmail::SyncState; +use crate::sync::pipelines::composio::orchestrator::{IncrementalSource, SyncScope}; +use crate::sync::pipelines::traits::PipelineConfig; + +/// Executor that must never run — `arguments` is pure argument-building. +struct NeverExecutor; + +#[async_trait::async_trait] +impl ActionExecutor for NeverExecutor { + async fn execute( + &self, + _action: &str, + _arguments: serde_json::Value, + _connection_id: Option<&str>, + ) -> anyhow::Result { + unreachable!("arguments() must not execute anything") + } +} + +fn query_of( + pipeline: &GmailSyncPipeline, + state: &SyncState, + config: &PipelineConfig, +) -> Option { + let args = pipeline.arguments(&SyncScope::flat(), config, state, None); + args.get("query") + .and_then(|q| q.as_str()) + .map(str::to_string) +} + +/// The standing filter ANDs with the incremental `after:` clause — +/// scoped sync stays incremental instead of re-querying the whole label. +#[test] +fn filter_composes_with_the_incremental_cursor_clause() { + let pipeline = GmailSyncPipeline::with_executor(Arc::new(NeverExecutor), "conn-1") + .with_filter("label:brain"); + let mut state = SyncState::new("gmail", "conn-1"); + state.cursor = Some("2026-05-02T09:15:00Z".into()); + + let query = query_of(&pipeline, &state, &PipelineConfig::default()).expect("query set"); + assert!(query.starts_with("label:brain after:"), "got: {query}"); +} + +/// Filter alone (no cursor, no depth cap): the query is exactly the filter. +#[test] +fn filter_alone_scopes_the_first_sync() { + let pipeline = GmailSyncPipeline::with_executor(Arc::new(NeverExecutor), "conn-1") + .with_filter("label:brain"); + let state = SyncState::new("gmail", "conn-1"); + + let query = query_of(&pipeline, &state, &PipelineConfig::default()).expect("query set"); + assert_eq!(query, "label:brain"); +} + +/// No filter, no cursor, no depth: no query argument at all (pre-existing +/// behaviour, must not regress to an empty-string query). +#[test] +fn no_clauses_means_no_query_argument() { + let pipeline = GmailSyncPipeline::with_executor(Arc::new(NeverExecutor), "conn-1"); + let state = SyncState::new("gmail", "conn-1"); + + assert_eq!( + query_of(&pipeline, &state, &PipelineConfig::default()), + None + ); +} + +/// `with_query` (backfill) still *replaces* the incremental clause, and a +/// standing filter composes in front of it. +#[test] +fn query_override_still_replaces_cursor_and_composes_with_filter() { + let pipeline = GmailSyncPipeline::with_executor(Arc::new(NeverExecutor), "conn-1") + .with_filter("label:brain") + .with_query("newer_than:3d"); + let mut state = SyncState::new("gmail", "conn-1"); + state.cursor = Some("2026-05-02T09:15:00Z".into()); + + let query = query_of(&pipeline, &state, &PipelineConfig::default()).expect("query set"); + assert_eq!( + query, "label:brain newer_than:3d", + "override wins over cursor" + ); +} /// One message in the shape the Gmail response reshaper emits: a slim envelope /// whose body is pre-rendered into `markdown`. diff --git a/crates/tinymemory-core/src/sync/pipelines/host.rs b/crates/tinymemory-core/src/sync/pipelines/host.rs index dbc91e91..91971a74 100644 --- a/crates/tinymemory-core/src/sync/pipelines/host.rs +++ b/crates/tinymemory-core/src/sync/pipelines/host.rs @@ -257,6 +257,7 @@ pub fn composio_config(config: &Config) -> Result { api_key: Some(SecretString::new(api_key)), bearer_token: None, entity_id: Some(config.composio().entity_id.clone()), + gmail_query: config.composio().gmail_sync_query.clone(), }) } else { let bearer = config @@ -268,6 +269,7 @@ pub fn composio_config(config: &Config) -> Result { api_key: None, bearer_token: Some(SecretString::new(bearer)), entity_id: Some(config.composio().entity_id.clone()), + gmail_query: config.composio().gmail_sync_query.clone(), }) } } @@ -299,9 +301,22 @@ fn build_composio_pipeline( if !syncable_composio_toolkits().contains(&slug.as_str()) { return Err(format!("memory sync does not support toolkit '{toolkit}'")); } + // Pull the Gmail scope filter out before the client consumes the config. + let gmail_filter = composio + .gmail_query + .as_deref() + .map(str::trim) + .filter(|q| !q.is_empty()) + .map(str::to_string); let client = ComposioClient::new(composio); Ok(match slug.as_str() { - "gmail" => Arc::new(GmailSyncPipeline::new(client, connection_id)), + "gmail" => { + let mut pipeline = GmailSyncPipeline::new(client, connection_id); + if let Some(filter) = gmail_filter { + pipeline = pipeline.with_filter(filter); + } + Arc::new(pipeline) + } "github" => Arc::new(GitHubSyncPipeline::new(client, connection_id)), "notion" => Arc::new(NotionSyncPipeline::new(client, connection_id)), "linear" => Arc::new(LinearSyncPipeline::new(client, connection_id)), diff --git a/crates/tinymemory-core/src/sync/pipelines/traits.rs b/crates/tinymemory-core/src/sync/pipelines/traits.rs index a3b76d30..598ddde7 100644 --- a/crates/tinymemory-core/src/sync/pipelines/traits.rs +++ b/crates/tinymemory-core/src/sync/pipelines/traits.rs @@ -102,6 +102,10 @@ pub struct ComposioSyncConfig { pub api_key: Option, pub bearer_token: Option, pub entity_id: Option, + /// Optional Gmail search query the Gmail pipeline ANDs onto every page + /// fetch (e.g. `label:brain`) so background sync only ingests matching + /// messages. `None` = whole inbox window. + pub gmail_query: Option, } /// A string whose `Debug` never prints the value. diff --git a/crates/tinymemory-core/tests/composio_gmail_non_tinycortex_e2e.rs b/crates/tinymemory-core/tests/composio_gmail_non_tinycortex_e2e.rs index 4488a62a..dcde9030 100644 --- a/crates/tinymemory-core/tests/composio_gmail_non_tinycortex_e2e.rs +++ b/crates/tinymemory-core/tests/composio_gmail_non_tinycortex_e2e.rs @@ -174,6 +174,7 @@ async fn composio_gmail_sync_completes_against_the_namespace_driver() { api_key: Some(SecretString::new("test-key")), bearer_token: None, entity_id: Some("entity-1".into()), + gmail_query: None, }; let pipeline = Arc::new(GmailSyncPipeline::new( ComposioClient::new(composio),