Add a standing Gmail search filter that scopes background sync

crates/tinymemory-api/src/host/config.rs

@@ -61,6 +61,10 @@ pub struct ComposioMode {
     pub api_key: Option<String>,
     /// 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<String>,
 }
 
 impl std::fmt::Debug for ComposioMode {

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<String>,
+    /// Standing Gmail search filter (e.g. `label:brain`) ANDed onto every
+    /// fetch, *including* the incremental `after:<cursor>` clause — unlike
+    /// [`Self::with_query`], which replaces the incremental clause outright
+    /// (backfill semantics).
+    filter: Option<String>,
 }
 
 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:<cursor>` /
+    /// `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<String>) -> 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<String> = 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
     }
 

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<ExecuteResponse> {
+        unreachable!("arguments() must not execute anything")
+    }
+}
+
+fn query_of(
+    pipeline: &GmailSyncPipeline,
+    state: &SyncState,
+    config: &PipelineConfig,
+) -> Option<String> {
+    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:<cursor>` 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`.

crates/tinymemory-core/src/sync/pipelines/host.rs

@@ -257,6 +257,7 @@ pub fn composio_config(config: &Config) -> Result<ComposioSyncConfig, String> {
             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<ComposioSyncConfig, String> {
             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)),

crates/tinymemory-core/src/sync/pipelines/traits.rs

@@ -102,6 +102,10 @@ pub struct ComposioSyncConfig {
     pub api_key: Option<SecretString>,
     pub bearer_token: Option<SecretString>,
     pub entity_id: Option<String>,
+    /// 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<String>,
 }
 
 /// A string whose `Debug` never prints the value.

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),