Refactor: rust modules (#633)

src/openhuman/about_app/mod.rs

@@ -5,43 +5,13 @@
 //! stable, beta, coming soon, or deprecated.
 
 mod catalog;
+mod ops;
 mod schemas;
 mod types;
 
-use crate::rpc::RpcOutcome;
-
 pub use catalog::{all_capabilities, capabilities_by_category, lookup, search};
+pub use ops::{list_capabilities, lookup_capability, search_capabilities};
 pub use schemas::{
     about_app_schemas, all_about_app_controller_schemas, all_about_app_registered_controllers,
 };
 pub use types::{Capability, CapabilityCategory, CapabilityStatus};
-
-pub fn list_capabilities(category: Option<CapabilityCategory>) -> RpcOutcome<Vec<Capability>> {
-    let capabilities = match category {
-        Some(category) => capabilities_by_category(category),
-        None => all_capabilities().to_vec(),
-    };
-    let log = format!(
-        "about_app.list returned {} capabilities",
-        capabilities.len()
-    );
-    RpcOutcome::single_log(capabilities, log)
-}
-
-pub fn lookup_capability(id: &str) -> Result<RpcOutcome<Capability>, String> {
-    let capability = lookup(id).ok_or_else(|| format!("unknown capability id '{}'", id.trim()))?;
-    Ok(RpcOutcome::single_log(
-        capability,
-        format!("about_app.lookup returned {}", capability.id),
-    ))
-}
-
-pub fn search_capabilities(query: &str) -> RpcOutcome<Vec<Capability>> {
-    let capabilities = search(query);
-    let log = format!(
-        "about_app.search returned {} capabilities for '{}'",
-        capabilities.len(),
-        query.trim()
-    );
-    RpcOutcome::single_log(capabilities, log)
-}

src/openhuman/about_app/ops.rs

@@ -0,0 +1,36 @@
+//! RPC entry points for the about_app capability catalog.
+
+use crate::rpc::RpcOutcome;
+
+use super::types::{Capability, CapabilityCategory};
+use super::{all_capabilities, capabilities_by_category, lookup, search};
+
+pub fn list_capabilities(category: Option<CapabilityCategory>) -> RpcOutcome<Vec<Capability>> {
+    let capabilities = match category {
+        Some(category) => capabilities_by_category(category),
+        None => all_capabilities().to_vec(),
+    };
+    let log = format!(
+        "about_app.list returned {} capabilities",
+        capabilities.len()
+    );
+    RpcOutcome::single_log(capabilities, log)
+}
+
+pub fn lookup_capability(id: &str) -> Result<RpcOutcome<Capability>, String> {
+    let capability = lookup(id).ok_or_else(|| format!("unknown capability id '{}'", id.trim()))?;
+    Ok(RpcOutcome::single_log(
+        capability,
+        format!("about_app.lookup returned {}", capability.id),
+    ))
+}
+
+pub fn search_capabilities(query: &str) -> RpcOutcome<Vec<Capability>> {
+    let capabilities = search(query);
+    let log = format!(
+        "about_app.search returned {} capabilities for '{}'",
+        capabilities.len(),
+        query.trim()
+    );
+    RpcOutcome::single_log(capabilities, log)
+}

src/openhuman/agent/agents/loader.rs

@@ -0,0 +1,348 @@
+//! Built-in agent definitions.
+//!
+//! Every built-in agent lives in its own subfolder here, with exactly
+//! two files:
+//!
+//! * `agent.toml`  — id, when_to_use, model, tool allowlist, sandbox,
+//!   iteration cap, and the `omit_*` flags. Parsed
+//!   directly into [`AgentDefinition`] via serde.
+//! * `prompt.md`   — the sub-agent's system prompt body.
+//!
+//! Adding a new built-in agent = creating a new subfolder with those two
+//! files and appending one entry to [`BUILTINS`] below. There are no
+//! match arms to update, no enum variants to add, and no `include_str!`
+//! paths scattered across the harness.
+//!
+//! ## Flow
+//!
+//! 1. [`load_builtins`] walks [`BUILTINS`].
+//! 2. For each entry, parses `agent.toml` into an [`AgentDefinition`].
+//! 3. Replaces the (unset) `system_prompt` with `PromptSource::Inline(prompt.md contents)`.
+//! 4. Stamps `source = DefinitionSource::Builtin`.
+//! 5. Returns the full `Vec<AgentDefinition>`, in the order listed in [`BUILTINS`].
+//!
+//! The synthetic `fork` definition is *not* listed here — it's a
+//! byte-stable replay of the parent and has no standalone prompt. It is
+//! added by [`super::harness::builtin_definitions::all`] on top of the
+//! loader output.
+//!
+//! Workspace-level overrides (`$OPENHUMAN_WORKSPACE/agents/*.toml`) are
+//! handled separately by [`super::harness::definition_loader`] and merged
+//! into the global registry, where they replace built-ins on `id`
+//! collision.
+
+use crate::openhuman::agent::harness::definition::{
+    AgentDefinition, DefinitionSource, PromptSource,
+};
+use anyhow::{Context, Result};
+
+/// A single built-in agent: its id plus the two files that define it.
+///
+/// Kept as a static slice (rather than e.g. `include_dir!`) so the
+/// compile-time file-existence check is explicit and grep-friendly.
+pub struct BuiltinAgent {
+    pub id: &'static str,
+    pub toml: &'static str,
+    pub prompt: &'static str,
+}
+
+/// Every built-in agent, in stable display order.
+///
+/// **This is the only list you touch when adding a new built-in agent.**
+pub const BUILTINS: &[BuiltinAgent] = &[
+    BuiltinAgent {
+        id: "orchestrator",
+        toml: include_str!("orchestrator/agent.toml"),
+        prompt: include_str!("orchestrator/prompt.md"),
+    },
+    BuiltinAgent {
+        id: "planner",
+        toml: include_str!("planner/agent.toml"),
+        prompt: include_str!("planner/prompt.md"),
+    },
+    BuiltinAgent {
+        id: "code_executor",
+        toml: include_str!("code_executor/agent.toml"),
+        prompt: include_str!("code_executor/prompt.md"),
+    },
+    BuiltinAgent {
+        id: "skills_agent",
+        toml: include_str!("skills_agent/agent.toml"),
+        prompt: include_str!("skills_agent/prompt.md"),
+    },
+    BuiltinAgent {
+        id: "tool_maker",
+        toml: include_str!("tool_maker/agent.toml"),
+        prompt: include_str!("tool_maker/prompt.md"),
+    },
+    BuiltinAgent {
+        id: "researcher",
+        toml: include_str!("researcher/agent.toml"),
+        prompt: include_str!("researcher/prompt.md"),
+    },
+    BuiltinAgent {
+        id: "critic",
+        toml: include_str!("critic/agent.toml"),
+        prompt: include_str!("critic/prompt.md"),
+    },
+    BuiltinAgent {
+        id: "archivist",
+        toml: include_str!("archivist/agent.toml"),
+        prompt: include_str!("archivist/prompt.md"),
+    },
+    BuiltinAgent {
+        id: "trigger_triage",
+        toml: include_str!("trigger_triage/agent.toml"),
+        prompt: include_str!("trigger_triage/prompt.md"),
+    },
+    BuiltinAgent {
+        id: "trigger_reactor",
+        toml: include_str!("trigger_reactor/agent.toml"),
+        prompt: include_str!("trigger_reactor/prompt.md"),
+    },
+    BuiltinAgent {
+        id: "morning_briefing",
+        toml: include_str!("morning_briefing/agent.toml"),
+        prompt: include_str!("morning_briefing/prompt.md"),
+    },
+    BuiltinAgent {
+        id: "welcome",
+        toml: include_str!("welcome/agent.toml"),
+        prompt: include_str!("welcome/prompt.md"),
+    },
+];
+
+/// Parse every entry in [`BUILTINS`] into an [`AgentDefinition`].
+///
+/// Errors out of the whole call on any parse failure — built-in TOML is
+/// baked into the binary and therefore must always be valid. Unit tests
+/// below keep that invariant honest.
+pub fn load_builtins() -> Result<Vec<AgentDefinition>> {
+    BUILTINS.iter().map(parse_builtin).collect()
+}
+
+/// Parse a single [`BuiltinAgent`] triple into a finished [`AgentDefinition`].
+fn parse_builtin(b: &BuiltinAgent) -> Result<AgentDefinition> {
+    // The TOML ships without `system_prompt` — serde falls back to
+    // `defaults::empty_inline_prompt` — and the loader injects the
+    // rendered sibling `prompt.md` immediately below.
+    let mut def: AgentDefinition = toml::from_str(b.toml)
+        .with_context(|| format!("parsing built-in agent `{}` TOML", b.id))?;
+
+    // Inject the prompt body and stamp the source.
+    def.system_prompt = PromptSource::Inline(b.prompt.to_string());
+    def.source = DefinitionSource::Builtin;
+
+    // Sanity check: file layout id must match declared TOML id. This
+    // catches copy-paste mistakes where someone forgets to update the
+    // `id` field after duplicating a folder.
+    anyhow::ensure!(
+        def.id == b.id,
+        "built-in agent folder `{}` declares mismatched TOML id `{}`",
+        b.id,
+        def.id
+    );
+
+    Ok(def)
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use crate::openhuman::agent::harness::definition::{ModelSpec, SandboxMode, ToolScope};
+
+    #[test]
+    fn all_builtins_parse() {
+        let defs = load_builtins().expect("built-in TOML must parse");
+        assert_eq!(defs.len(), BUILTINS.len());
+        assert_eq!(defs.len(), 12, "expected 12 built-in agents");
+    }
+
+    #[test]
+    fn trigger_reactor_has_agentic_hint_and_narrow_tools() {
+        let def = find("trigger_reactor");
+        assert!(matches!(def.model, ModelSpec::Hint(ref h) if h == "agentic"));
+        match &def.tools {
+            ToolScope::Named(tools) => {
+                assert!(
+                    tools.iter().any(|t| t == "memory_recall"),
+                    "trigger_reactor needs memory_recall"
+                );
+                assert!(
+                    tools.iter().any(|t| t == "memory_store"),
+                    "trigger_reactor needs memory_store"
+                );
+                assert!(
+                    tools.iter().any(|t| t == "spawn_subagent"),
+                    "trigger_reactor needs spawn_subagent for escalation"
+                );
+                // No shell / file_write — reactor does not execute code.
+                assert!(!tools.iter().any(|t| t == "shell"));
+                assert!(!tools.iter().any(|t| t == "file_write"));
+            }
+            ToolScope::Wildcard => panic!("trigger_reactor must have a Named tool scope"),
+        }
+        assert_eq!(def.sandbox_mode, SandboxMode::None);
+        assert_eq!(def.max_iterations, 6);
+        assert!(
+            !def.omit_memory_context,
+            "trigger_reactor needs global memory/context"
+        );
+    }
+
+    #[test]
+    fn trigger_triage_has_no_tools_and_pulls_memory_context() {
+        let def = find("trigger_triage");
+        match &def.tools {
+            ToolScope::Named(tools) => assert!(
+                tools.is_empty(),
+                "trigger_triage must have zero tools (got {tools:?})"
+            ),
+            ToolScope::Wildcard => panic!("trigger_triage must have a Named empty tool scope"),
+        }
+        assert!(
+            !def.omit_memory_context,
+            "trigger_triage needs global memory/context to reason about triggers"
+        );
+        assert!(def.omit_identity);
+        assert!(def.omit_safety_preamble);
+        assert!(def.omit_skills_catalog);
+        assert_eq!(def.sandbox_mode, SandboxMode::ReadOnly);
+        assert_eq!(def.max_iterations, 2);
+    }
+
+    #[test]
+    fn folder_ids_match_toml_ids() {
+        for b in BUILTINS {
+            let def = parse_builtin(b).expect("parse");
+            assert_eq!(def.id, b.id, "folder `{}` id mismatch", b.id);
+        }
+    }
+
+    #[test]
+    fn every_builtin_has_a_prompt_body() {
+        for def in load_builtins().unwrap() {
+            match &def.system_prompt {
+                PromptSource::Inline(body) => {
+                    assert!(!body.is_empty(), "{} has empty prompt", def.id);
+                }
+                PromptSource::File { .. } => {
+                    panic!("{} should use inline prompt, not File", def.id);
+                }
+            }
+        }
+    }
+
+    #[test]
+    fn every_builtin_is_stamped_builtin_source() {
+        for def in load_builtins().unwrap() {
+            assert_eq!(def.source, DefinitionSource::Builtin);
+        }
+    }
+
+    fn find(id: &str) -> AgentDefinition {
+        load_builtins()
+            .unwrap()
+            .into_iter()
+            .find(|d| d.id == id)
+            .unwrap_or_else(|| panic!("missing built-in {id}"))
+    }
+
+    #[test]
+    fn orchestrator_has_reasoning_hint_and_named_tools() {
+        let def = find("orchestrator");
+        assert!(matches!(def.model, ModelSpec::Hint(ref h) if h == "reasoning"));
+        match def.tools {
+            ToolScope::Named(tools) => {
+                assert!(tools.iter().any(|t| t == "spawn_subagent"));
+                assert!(!tools.iter().any(|t| t == "shell"));
+                assert!(!tools.iter().any(|t| t == "file_write"));
+            }
+            ToolScope::Wildcard => panic!("orchestrator must have named tool allowlist"),
+        }
+        assert_eq!(def.max_iterations, 15);
+    }
+
+    #[test]
+    fn code_executor_is_sandboxed_and_keeps_safety_preamble() {
+        let def = find("code_executor");
+        assert_eq!(def.sandbox_mode, SandboxMode::Sandboxed);
+        assert!(!def.omit_safety_preamble);
+        assert_eq!(def.max_iterations, 10);
+    }
+
+    #[test]
+    fn tool_maker_is_sandboxed_with_max_2_iterations() {
+        let def = find("tool_maker");
+        assert_eq!(def.sandbox_mode, SandboxMode::Sandboxed);
+        assert_eq!(def.max_iterations, 2);
+        assert!(!def.omit_safety_preamble);
+    }
+
+    #[test]
+    fn critic_is_read_only() {
+        let def = find("critic");
+        assert_eq!(def.sandbox_mode, SandboxMode::ReadOnly);
+        assert!(def.omit_safety_preamble);
+    }
+
+    #[test]
+    fn skills_agent_is_wildcard_with_skill_category_filter() {
+        let def = find("skills_agent");
+        assert!(matches!(def.tools, ToolScope::Wildcard));
+        assert_eq!(
+            def.category_filter,
+            Some(crate::openhuman::tools::ToolCategory::Skill)
+        );
+        assert!(!def.omit_safety_preamble);
+    }
+
+    #[test]
+    fn archivist_runs_in_background() {
+        let def = find("archivist");
+        assert!(def.background);
+        assert_eq!(def.max_iterations, 3);
+    }
+
+    #[test]
+    fn morning_briefing_is_read_only_with_skill_filter() {
+        let def = find("morning_briefing");
+        assert_eq!(def.sandbox_mode, SandboxMode::ReadOnly);
+        assert!(matches!(def.tools, ToolScope::Wildcard));
+        assert_eq!(
+            def.category_filter,
+            Some(crate::openhuman::tools::ToolCategory::Skill)
+        );
+        assert!(!def.omit_memory_context);
+        assert!(def.omit_identity);
+        assert!(def.omit_safety_preamble);
+        assert_eq!(def.max_iterations, 8);
+    }
+
+    #[test]
+    fn welcome_has_onboarding_and_memory_tools() {
+        let def = find("welcome");
+        assert_eq!(def.sandbox_mode, SandboxMode::ReadOnly);
+        match &def.tools {
+            ToolScope::Named(tools) => {
+                assert_eq!(tools.len(), 3, "welcome should have exactly three tools");
+                assert!(
+                    tools.iter().any(|t| t == "complete_onboarding"),
+                    "welcome needs complete_onboarding"
+                );
+                assert!(
+                    tools.iter().any(|t| t == "memory_recall"),
+                    "welcome needs memory_recall"
+                );
+                assert!(
+                    tools.iter().any(|t| t == "composio_authorize"),
+                    "welcome needs composio_authorize"
+                );
+            }
+            ToolScope::Wildcard => panic!("welcome must have a Named tool scope"),
+        }
+        assert!(!def.omit_memory_context);
+        assert!(def.omit_identity);
+        assert_eq!(def.max_iterations, 6);
+    }
+}

src/openhuman/agent/agents/mod.rs

@@ -1,348 +1,3 @@
-//! Built-in agent definitions.
-//!
-//! Every built-in agent lives in its own subfolder here, with exactly
-//! two files:
-//!
-//! * `agent.toml`  — id, when_to_use, model, tool allowlist, sandbox,
-//!   iteration cap, and the `omit_*` flags. Parsed
-//!   directly into [`AgentDefinition`] via serde.
-//! * `prompt.md`   — the sub-agent's system prompt body.
-//!
-//! Adding a new built-in agent = creating a new subfolder with those two
-//! files and appending one entry to [`BUILTINS`] below. There are no
-//! match arms to update, no enum variants to add, and no `include_str!`
-//! paths scattered across the harness.
-//!
-//! ## Flow
-//!
-//! 1. [`load_builtins`] walks [`BUILTINS`].
-//! 2. For each entry, parses `agent.toml` into an [`AgentDefinition`].
-//! 3. Replaces the (unset) `system_prompt` with `PromptSource::Inline(prompt.md contents)`.
-//! 4. Stamps `source = DefinitionSource::Builtin`.
-//! 5. Returns the full `Vec<AgentDefinition>`, in the order listed in [`BUILTINS`].
-//!
-//! The synthetic `fork` definition is *not* listed here — it's a
-//! byte-stable replay of the parent and has no standalone prompt. It is
-//! added by [`super::harness::builtin_definitions::all`] on top of the
-//! loader output.
-//!
-//! Workspace-level overrides (`$OPENHUMAN_WORKSPACE/agents/*.toml`) are
-//! handled separately by [`super::harness::definition_loader`] and merged
-//! into the global registry, where they replace built-ins on `id`
-//! collision.
+mod loader;
 
-use crate::openhuman::agent::harness::definition::{
-    AgentDefinition, DefinitionSource, PromptSource,
-};
-use anyhow::{Context, Result};
-
-/// A single built-in agent: its id plus the two files that define it.
-///
-/// Kept as a static slice (rather than e.g. `include_dir!`) so the
-/// compile-time file-existence check is explicit and grep-friendly.
-pub struct BuiltinAgent {
-    pub id: &'static str,
-    pub toml: &'static str,
-    pub prompt: &'static str,
-}
-
-/// Every built-in agent, in stable display order.
-///
-/// **This is the only list you touch when adding a new built-in agent.**
-pub const BUILTINS: &[BuiltinAgent] = &[
-    BuiltinAgent {
-        id: "orchestrator",
-        toml: include_str!("orchestrator/agent.toml"),
-        prompt: include_str!("orchestrator/prompt.md"),
-    },
-    BuiltinAgent {
-        id: "planner",
-        toml: include_str!("planner/agent.toml"),
-        prompt: include_str!("planner/prompt.md"),
-    },
-    BuiltinAgent {
-        id: "code_executor",
-        toml: include_str!("code_executor/agent.toml"),
-        prompt: include_str!("code_executor/prompt.md"),
-    },
-    BuiltinAgent {
-        id: "skills_agent",
-        toml: include_str!("skills_agent/agent.toml"),
-        prompt: include_str!("skills_agent/prompt.md"),
-    },
-    BuiltinAgent {
-        id: "tool_maker",
-        toml: include_str!("tool_maker/agent.toml"),
-        prompt: include_str!("tool_maker/prompt.md"),
-    },
-    BuiltinAgent {
-        id: "researcher",
-        toml: include_str!("researcher/agent.toml"),
-        prompt: include_str!("researcher/prompt.md"),
-    },
-    BuiltinAgent {
-        id: "critic",
-        toml: include_str!("critic/agent.toml"),
-        prompt: include_str!("critic/prompt.md"),
-    },
-    BuiltinAgent {
-        id: "archivist",
-        toml: include_str!("archivist/agent.toml"),
-        prompt: include_str!("archivist/prompt.md"),
-    },
-    BuiltinAgent {
-        id: "trigger_triage",
-        toml: include_str!("trigger_triage/agent.toml"),
-        prompt: include_str!("trigger_triage/prompt.md"),
-    },
-    BuiltinAgent {
-        id: "trigger_reactor",
-        toml: include_str!("trigger_reactor/agent.toml"),
-        prompt: include_str!("trigger_reactor/prompt.md"),
-    },
-    BuiltinAgent {
-        id: "morning_briefing",
-        toml: include_str!("morning_briefing/agent.toml"),
-        prompt: include_str!("morning_briefing/prompt.md"),
-    },
-    BuiltinAgent {
-        id: "welcome",
-        toml: include_str!("welcome/agent.toml"),
-        prompt: include_str!("welcome/prompt.md"),
-    },
-];
-
-/// Parse every entry in [`BUILTINS`] into an [`AgentDefinition`].
-///
-/// Errors out of the whole call on any parse failure — built-in TOML is
-/// baked into the binary and therefore must always be valid. Unit tests
-/// below keep that invariant honest.
-pub fn load_builtins() -> Result<Vec<AgentDefinition>> {
-    BUILTINS.iter().map(parse_builtin).collect()
-}
-
-/// Parse a single [`BuiltinAgent`] triple into a finished [`AgentDefinition`].
-fn parse_builtin(b: &BuiltinAgent) -> Result<AgentDefinition> {
-    // The TOML ships without `system_prompt` — serde falls back to
-    // `defaults::empty_inline_prompt` — and the loader injects the
-    // rendered sibling `prompt.md` immediately below.
-    let mut def: AgentDefinition = toml::from_str(b.toml)
-        .with_context(|| format!("parsing built-in agent `{}` TOML", b.id))?;
-
-    // Inject the prompt body and stamp the source.
-    def.system_prompt = PromptSource::Inline(b.prompt.to_string());
-    def.source = DefinitionSource::Builtin;
-
-    // Sanity check: file layout id must match declared TOML id. This
-    // catches copy-paste mistakes where someone forgets to update the
-    // `id` field after duplicating a folder.
-    anyhow::ensure!(
-        def.id == b.id,
-        "built-in agent folder `{}` declares mismatched TOML id `{}`",
-        b.id,
-        def.id
-    );
-
-    Ok(def)
-}
-
-#[cfg(test)]
-mod tests {
-    use super::*;
-    use crate::openhuman::agent::harness::definition::{ModelSpec, SandboxMode, ToolScope};
-
-    #[test]
-    fn all_builtins_parse() {
-        let defs = load_builtins().expect("built-in TOML must parse");
-        assert_eq!(defs.len(), BUILTINS.len());
-        assert_eq!(defs.len(), 12, "expected 12 built-in agents");
-    }
-
-    #[test]
-    fn trigger_reactor_has_agentic_hint_and_narrow_tools() {
-        let def = find("trigger_reactor");
-        assert!(matches!(def.model, ModelSpec::Hint(ref h) if h == "agentic"));
-        match &def.tools {
-            ToolScope::Named(tools) => {
-                assert!(
-                    tools.iter().any(|t| t == "memory_recall"),
-                    "trigger_reactor needs memory_recall"
-                );
-                assert!(
-                    tools.iter().any(|t| t == "memory_store"),
-                    "trigger_reactor needs memory_store"
-                );
-                assert!(
-                    tools.iter().any(|t| t == "spawn_subagent"),
-                    "trigger_reactor needs spawn_subagent for escalation"
-                );
-                // No shell / file_write — reactor does not execute code.
-                assert!(!tools.iter().any(|t| t == "shell"));
-                assert!(!tools.iter().any(|t| t == "file_write"));
-            }
-            ToolScope::Wildcard => panic!("trigger_reactor must have a Named tool scope"),
-        }
-        assert_eq!(def.sandbox_mode, SandboxMode::None);
-        assert_eq!(def.max_iterations, 6);
-        assert!(
-            !def.omit_memory_context,
-            "trigger_reactor needs global memory/context"
-        );
-    }
-
-    #[test]
-    fn trigger_triage_has_no_tools_and_pulls_memory_context() {
-        let def = find("trigger_triage");
-        match &def.tools {
-            ToolScope::Named(tools) => assert!(
-                tools.is_empty(),
-                "trigger_triage must have zero tools (got {tools:?})"
-            ),
-            ToolScope::Wildcard => panic!("trigger_triage must have a Named empty tool scope"),
-        }
-        assert!(
-            !def.omit_memory_context,
-            "trigger_triage needs global memory/context to reason about triggers"
-        );
-        assert!(def.omit_identity);
-        assert!(def.omit_safety_preamble);
-        assert!(def.omit_skills_catalog);
-        assert_eq!(def.sandbox_mode, SandboxMode::ReadOnly);
-        assert_eq!(def.max_iterations, 2);
-    }
-
-    #[test]
-    fn folder_ids_match_toml_ids() {
-        for b in BUILTINS {
-            let def = parse_builtin(b).expect("parse");
-            assert_eq!(def.id, b.id, "folder `{}` id mismatch", b.id);
-        }
-    }
-
-    #[test]
-    fn every_builtin_has_a_prompt_body() {
-        for def in load_builtins().unwrap() {
-            match &def.system_prompt {
-                PromptSource::Inline(body) => {
-                    assert!(!body.is_empty(), "{} has empty prompt", def.id);
-                }
-                PromptSource::File { .. } => {
-                    panic!("{} should use inline prompt, not File", def.id);
-                }
-            }
-        }
-    }
-
-    #[test]
-    fn every_builtin_is_stamped_builtin_source() {
-        for def in load_builtins().unwrap() {
-            assert_eq!(def.source, DefinitionSource::Builtin);
-        }
-    }
-
-    fn find(id: &str) -> AgentDefinition {
-        load_builtins()
-            .unwrap()
-            .into_iter()
-            .find(|d| d.id == id)
-            .unwrap_or_else(|| panic!("missing built-in {id}"))
-    }
-
-    #[test]
-    fn orchestrator_has_reasoning_hint_and_named_tools() {
-        let def = find("orchestrator");
-        assert!(matches!(def.model, ModelSpec::Hint(ref h) if h == "reasoning"));
-        match def.tools {
-            ToolScope::Named(tools) => {
-                assert!(tools.iter().any(|t| t == "spawn_subagent"));
-                assert!(!tools.iter().any(|t| t == "shell"));
-                assert!(!tools.iter().any(|t| t == "file_write"));
-            }
-            ToolScope::Wildcard => panic!("orchestrator must have named tool allowlist"),
-        }
-        assert_eq!(def.max_iterations, 15);
-    }
-
-    #[test]
-    fn code_executor_is_sandboxed_and_keeps_safety_preamble() {
-        let def = find("code_executor");
-        assert_eq!(def.sandbox_mode, SandboxMode::Sandboxed);
-        assert!(!def.omit_safety_preamble);
-        assert_eq!(def.max_iterations, 10);
-    }
-
-    #[test]
-    fn tool_maker_is_sandboxed_with_max_2_iterations() {
-        let def = find("tool_maker");
-        assert_eq!(def.sandbox_mode, SandboxMode::Sandboxed);
-        assert_eq!(def.max_iterations, 2);
-        assert!(!def.omit_safety_preamble);
-    }
-
-    #[test]
-    fn critic_is_read_only() {
-        let def = find("critic");
-        assert_eq!(def.sandbox_mode, SandboxMode::ReadOnly);
-        assert!(def.omit_safety_preamble);
-    }
-
-    #[test]
-    fn skills_agent_is_wildcard_with_skill_category_filter() {
-        let def = find("skills_agent");
-        assert!(matches!(def.tools, ToolScope::Wildcard));
-        assert_eq!(
-            def.category_filter,
-            Some(crate::openhuman::tools::ToolCategory::Skill)
-        );
-        assert!(!def.omit_safety_preamble);
-    }
-
-    #[test]
-    fn archivist_runs_in_background() {
-        let def = find("archivist");
-        assert!(def.background);
-        assert_eq!(def.max_iterations, 3);
-    }
-
-    #[test]
-    fn morning_briefing_is_read_only_with_skill_filter() {
-        let def = find("morning_briefing");
-        assert_eq!(def.sandbox_mode, SandboxMode::ReadOnly);
-        assert!(matches!(def.tools, ToolScope::Wildcard));
-        assert_eq!(
-            def.category_filter,
-            Some(crate::openhuman::tools::ToolCategory::Skill)
-        );
-        assert!(!def.omit_memory_context);
-        assert!(def.omit_identity);
-        assert!(def.omit_safety_preamble);
-        assert_eq!(def.max_iterations, 8);
-    }
-
-    #[test]
-    fn welcome_has_onboarding_and_memory_tools() {
-        let def = find("welcome");
-        assert_eq!(def.sandbox_mode, SandboxMode::ReadOnly);
-        match &def.tools {
-            ToolScope::Named(tools) => {
-                assert_eq!(tools.len(), 3, "welcome should have exactly three tools");
-                assert!(
-                    tools.iter().any(|t| t == "complete_onboarding"),
-                    "welcome needs complete_onboarding"
-                );
-                assert!(
-                    tools.iter().any(|t| t == "memory_recall"),
-                    "welcome needs memory_recall"
-                );
-                assert!(
-                    tools.iter().any(|t| t == "composio_authorize"),
-                    "welcome needs composio_authorize"
-                );
-            }
-            ToolScope::Wildcard => panic!("welcome must have a Named tool scope"),
-        }
-        assert!(!def.omit_memory_context);
-        assert!(def.omit_identity);
-        assert_eq!(def.max_iterations, 6);
-    }
-}
+pub use loader::{load_builtins, BuiltinAgent, BUILTINS};

src/openhuman/composio/providers/descriptions.rs

@@ -0,0 +1,43 @@
+//! Human-readable capability summaries for Composio toolkit slugs.
+
+/// Human-readable capability summary for a Composio toolkit slug.
+///
+/// Used by the prompt renderer to tell the orchestrator what each connected
+/// integration can do. Covers the most common toolkits; unknown slugs get
+/// a generic fallback so newly connected services still appear.
+pub fn toolkit_description(slug: &str) -> &'static str {
+    match slug {
+        "gmail" => {
+            "Send, read, draft, reply, forward, and search emails; manage labels and threads"
+        }
+        "notion" => "Create, read, update, and search notion pages and notion databases",
+        "github" => "Manage repositories, issues, pull requests on GitHub",
+        "slack" => "Send messages, read channels, manage threads, and post updates in Slack",
+        "discord" => "Send messages, manage channels, and interact with Discord servers",
+        "google_calendar" => "Create, update, and query calendar events; check availability",
+        "google_drive" => "Upload, download, search, and share files in Google Drive",
+        "google_docs" => "Create, read, and edit Google Docs documents",
+        "google_sheets" => "Read, write, and manage Google Sheets spreadsheets",
+        "outlook" => "Send, read, and manage emails in Microsoft Outlook",
+        "microsoft_teams" => "Send messages and manage channels in Microsoft Teams",
+        "linear" => "Create and manage issues, projects, and cycles in Linear",
+        "jira" => "Create and manage issues, projects, and sprints in Jira",
+        "trello" => "Create and manage cards, lists, and boards in Trello",
+        "asana" => "Create and manage tasks, projects, and sections in Asana",
+        "dropbox" => "Upload, download, and share files in Dropbox",
+        "twitter" => "Post tweets, read timelines, and manage Twitter interactions",
+        "spotify" => "Control playback, search music, and manage playlists on Spotify",
+        "telegram" => "Send and receive messages via Telegram",
+        "whatsapp" => "Send and receive messages via WhatsApp",
+        "twilio" => "Send SMS, make calls, and manage communications via Twilio",
+        "shopify" => "Manage products, orders, and customers in Shopify",
+        "stripe" => "Manage payments, subscriptions, and customers in Stripe",
+        "hubspot" => "Manage contacts, deals, and marketing in HubSpot",
+        "salesforce" => "Manage contacts, leads, and opportunities in Salesforce",
+        "airtable" => "Read and write records in Airtable bases",
+        "figma" => "Access and manage Figma design files and components",
+        "youtube" => "Search videos, manage playlists, and interact with YouTube",
+        "calendar" => "Create, update, and query calendar events",
+        _ => "Interact with this connected service via its available actions",
+    }
+}

src/openhuman/composio/providers/gmail/mod.rs

@@ -1,414 +1,6 @@
-//! Gmail provider — incremental sync with per-item persistence.
-//!
-//! On each sync pass:
-//!
-//!   1. Load persistent [`SyncState`] from the KV store.
-//!   2. Check the daily request budget — bail early if exhausted.
-//!   3. Fetch a page of recent messages via `GMAIL_FETCH_EMAILS`, adding
-//!      a date filter when a cursor exists so only newer mail is returned.
-//!   4. Deduplicate against `synced_ids` in the state.
-//!   5. Persist each **new** message as its own memory document (not a
-//!      single giant snapshot) so agent recall can find individual emails.
-//!   6. Paginate (up to budget) until no more results or all items in the
-//!      page are already synced.
-//!   7. Advance the cursor and save state.
-//!
-//! Daily budget (`DEFAULT_DAILY_REQUEST_LIMIT`, default 500) caps the
-//! number of `execute_tool` calls per calendar day, preventing runaway
-//! API usage during large initial backfills.
-
+mod provider;
 mod sync;
 #[cfg(test)]
 mod tests;
 
-use async_trait::async_trait;
-use serde_json::{json, Value};
-
-use super::sync_state::{extract_item_id, persist_single_item, SyncState};
-use super::{
-    pick_str, ComposioProvider, ProviderContext, ProviderUserProfile, SyncOutcome, SyncReason,
-};
-
-const ACTION_GET_PROFILE: &str = "GMAIL_GET_PROFILE";
-const ACTION_FETCH_EMAILS: &str = "GMAIL_FETCH_EMAILS";
-
-/// Page size per API call. Kept moderate so each call is fast and we
-/// get frequent checkpoints for the daily budget.
-const PAGE_SIZE: u32 = 25;
-
-/// Larger page size for the very first sync after OAuth so the user
-/// gets a meaningful initial snapshot.
-const INITIAL_PAGE_SIZE: u32 = 50;
-
-/// Maximum pages to fetch in a single sync pass (guards against infinite
-/// pagination loops). Combined with PAGE_SIZE this yields at most
-/// 500 items per sync pass, well within the daily budget.
-const MAX_PAGES_PER_SYNC: u32 = 20;
-
-/// Paths to try when extracting a message's unique ID from the Composio
-/// response envelope.
-const MESSAGE_ID_PATHS: &[&str] = &["id", "data.id", "messageId", "data.messageId"];
-
-/// Paths for extracting the internal date (epoch millis or date string)
-/// used as the sync cursor.
-const MESSAGE_DATE_PATHS: &[&str] = &[
-    "internalDate",
-    "data.internalDate",
-    "date",
-    "data.date",
-    "receivedAt",
-    "data.receivedAt",
-];
-
-pub struct GmailProvider;
-
-impl GmailProvider {
-    pub fn new() -> Self {
-        Self
-    }
-}
-
-impl Default for GmailProvider {
-    fn default() -> Self {
-        Self::new()
-    }
-}
-
-#[async_trait]
-impl ComposioProvider for GmailProvider {
-    fn toolkit_slug(&self) -> &'static str {
-        "gmail"
-    }
-
-    fn sync_interval_secs(&self) -> Option<u64> {
-        Some(15 * 60)
-    }
-
-    async fn fetch_user_profile(
-        &self,
-        ctx: &ProviderContext,
-    ) -> Result<ProviderUserProfile, String> {
-        tracing::debug!(
-            connection_id = ?ctx.connection_id,
-            "[composio:gmail] fetch_user_profile via {ACTION_GET_PROFILE}"
-        );
-
-        let resp = ctx
-            .client
-            .execute_tool(ACTION_GET_PROFILE, Some(json!({})))
-            .await
-            .map_err(|e| format!("[composio:gmail] {ACTION_GET_PROFILE} failed: {e:#}"))?;
-
-        if !resp.successful {
-            let err = resp
-                .error
-                .clone()
-                .unwrap_or_else(|| "provider reported failure".to_string());
-            return Err(format!("[composio:gmail] {ACTION_GET_PROFILE}: {err}"));
-        }
-
-        let data = &resp.data;
-        let email = pick_str(
-            data,
-            &[
-                "data.emailAddress",
-                "data.email",
-                "emailAddress",
-                "email",
-                "data.profile.emailAddress",
-            ],
-        );
-        let display_name = pick_str(
-            data,
-            &[
-                "data.name",
-                "data.profile.name",
-                "name",
-                "displayName",
-                "data.displayName",
-            ],
-        )
-        .or_else(|| email.clone());
-
-        let profile = ProviderUserProfile {
-            toolkit: "gmail".to_string(),
-            connection_id: ctx.connection_id.clone(),
-            display_name,
-            email,
-            username: None,
-            avatar_url: None,
-            extras: data.clone(),
-        };
-        let has_email = profile.email.is_some();
-        let email_domain = profile
-            .email
-            .as_deref()
-            .and_then(|e| e.split('@').nth(1))
-            .map(|d| d.to_string());
-        tracing::info!(
-            connection_id = ?profile.connection_id,
-            has_email,
-            email_domain = ?email_domain,
-            "[composio:gmail] fetched user profile"
-        );
-        Ok(profile)
-    }
-
-    async fn sync(&self, ctx: &ProviderContext, reason: SyncReason) -> Result<SyncOutcome, String> {
-        let started_at_ms = sync::now_ms();
-        let connection_id = ctx
-            .connection_id
-            .clone()
-            .unwrap_or_else(|| "default".to_string());
-
-        tracing::info!(
-            connection_id = %connection_id,
-            reason = reason.as_str(),
-            "[composio:gmail] incremental sync starting"
-        );
-
-        // ── Step 1: load persistent sync state ──────────────────────
-        let Some(memory) = ctx.memory_client() else {
-            return Err("[composio:gmail] memory client not ready".to_string());
-        };
-        let mut state = SyncState::load(&memory, "gmail", &connection_id).await?;
-
-        // ── Step 2: check daily budget ──────────────────────────────
-        if state.budget_exhausted() {
-            tracing::info!(
-                connection_id = %connection_id,
-                "[composio:gmail] daily request budget exhausted, skipping sync"
-            );
-            return Ok(SyncOutcome {
-                toolkit: "gmail".to_string(),
-                connection_id: Some(connection_id),
-                reason: reason.as_str().to_string(),
-                items_ingested: 0,
-                started_at_ms,
-                finished_at_ms: sync::now_ms(),
-                summary: "gmail sync skipped: daily budget exhausted".to_string(),
-                details: json!({ "budget_exhausted": true }),
-            });
-        }
-
-        // ── Step 3: paginated incremental fetch ─────────────────────
-        let page_size = match reason {
-            SyncReason::ConnectionCreated => INITIAL_PAGE_SIZE,
-            _ => PAGE_SIZE,
-        };
-
-        let mut total_fetched: usize = 0;
-        let mut total_persisted: usize = 0;
-        let mut newest_date: Option<String> = None;
-        let mut page_token: Option<String> = None;
-
-        for page_num in 0..MAX_PAGES_PER_SYNC {
-            if state.budget_exhausted() {
-                tracing::info!(
-                    page = page_num,
-                    "[composio:gmail] budget exhausted mid-sync, stopping pagination"
-                );
-                break;
-            }
-
-            // Build the Gmail query. If we have a cursor (date of last
-            // synced message), add `after:YYYY/MM/DD` so the API only
-            // returns newer mail.
-            let mut query = "in:inbox -in:spam -in:trash".to_string();
-            if let Some(ref cursor) = state.cursor {
-                if let Some(date_filter) = sync::cursor_to_gmail_after_filter(cursor) {
-                    query.push_str(&format!(" after:{date_filter}"));
-                    tracing::debug!(
-                        page = page_num,
-                        filter = %date_filter,
-                        "[composio:gmail] using date filter from cursor"
-                    );
-                }
-            }
-
-            let mut args = json!({
-                "max_results": page_size,
-                "query": query,
-            });
-            if let Some(ref token) = page_token {
-                args["page_token"] = json!(token);
-            }
-
-            let resp = ctx
-                .client
-                .execute_tool(ACTION_FETCH_EMAILS, Some(args))
-                .await
-                .map_err(|e| {
-                    format!("[composio:gmail] {ACTION_FETCH_EMAILS} page {page_num}: {e:#}")
-                })?;
-
-            state.record_requests(1);
-
-            if !resp.successful {
-                let err = resp
-                    .error
-                    .clone()
-                    .unwrap_or_else(|| "provider reported failure".to_string());
-                // Save state so budget accounting isn't lost.
-                let _ = state.save(&memory).await;
-                return Err(format!(
-                    "[composio:gmail] {ACTION_FETCH_EMAILS} page {page_num}: {err}"
-                ));
-            }
-
-            let messages = sync::extract_messages(&resp.data);
-            total_fetched += messages.len();
-
-            if messages.is_empty() {
-                tracing::debug!(
-                    page = page_num,
-                    "[composio:gmail] empty page, stopping pagination"
-                );
-                break;
-            }
-
-            // ── Step 4: deduplicate and persist per-item ────────────
-            let mut all_already_synced = true;
-            for msg in &messages {
-                let Some(msg_id) = extract_item_id(msg, MESSAGE_ID_PATHS) else {
-                    tracing::debug!("[composio:gmail] message missing ID, skipping");
-                    continue;
-                };
-
-                // Track the newest date we've seen for cursor advancement.
-                if let Some(date_val) = extract_item_id(msg, MESSAGE_DATE_PATHS) {
-                    if newest_date
-                        .as_ref()
-                        .map_or(true, |existing| date_val > *existing)
-                    {
-                        newest_date = Some(date_val);
-                    }
-                }
-
-                if state.is_synced(&msg_id) {
-                    continue;
-                }
-                all_already_synced = false;
-
-                // Build a human-readable title for this email document.
-                let subject = pick_str(
-                    msg,
-                    &[
-                        "subject",
-                        "data.subject",
-                        "payload.headers.Subject",
-                        "snippet",
-                    ],
-                )
-                .unwrap_or_else(|| format!("Email {msg_id}"));
-                let doc_id = format!("composio-gmail-msg-{msg_id}");
-                let title = format!("Gmail: {subject}");
-
-                match persist_single_item(
-                    &memory,
-                    "gmail",
-                    &doc_id,
-                    &title,
-                    msg,
-                    "gmail",
-                    ctx.connection_id.as_deref(),
-                )
-                .await
-                {
-                    Ok(_) => {
-                        state.mark_synced(&msg_id);
-                        total_persisted += 1;
-                    }
-                    Err(e) => {
-                        tracing::warn!(
-                            msg_id = %msg_id,
-                            error = %e,
-                            "[composio:gmail] failed to persist message (continuing)"
-                        );
-                    }
-                }
-            }
-
-            // If every message in this page was already synced, there's
-            // nothing new beyond this point — stop paginating.
-            if all_already_synced {
-                tracing::debug!(
-                    page = page_num,
-                    "[composio:gmail] all items in page already synced, stopping"
-                );
-                break;
-            }
-
-            // Check for next page token.
-            page_token = sync::extract_page_token(&resp.data);
-            if page_token.is_none() {
-                tracing::debug!(page = page_num, "[composio:gmail] no next page token, done");
-                break;
-            }
-        }
-
-        // ── Step 5: advance cursor and save state ───────────────────
-        if let Some(new_cursor) = newest_date {
-            state.advance_cursor(&new_cursor);
-        }
-        state.save(&memory).await?;
-
-        let finished_at_ms = sync::now_ms();
-        let summary = format!(
-            "gmail sync ({reason}): fetched {total_fetched}, persisted {total_persisted} new, \
-             budget remaining {remaining}",
-            reason = reason.as_str(),
-            remaining = state.budget_remaining(),
-        );
-        tracing::info!(
-            connection_id = %connection_id,
-            elapsed_ms = finished_at_ms.saturating_sub(started_at_ms),
-            total_fetched,
-            total_persisted,
-            budget_remaining = state.budget_remaining(),
-            "[composio:gmail] incremental sync complete"
-        );
-
-        Ok(SyncOutcome {
-            toolkit: "gmail".to_string(),
-            connection_id: Some(connection_id),
-            reason: reason.as_str().to_string(),
-            items_ingested: total_persisted,
-            started_at_ms,
-            finished_at_ms,
-            summary,
-            details: json!({
-                "messages_fetched": total_fetched,
-                "messages_persisted": total_persisted,
-                "budget_remaining": state.budget_remaining(),
-                "cursor": state.cursor,
-                "synced_ids_total": state.synced_ids.len(),
-            }),
-        })
-    }
-
-    async fn on_trigger(
-        &self,
-        ctx: &ProviderContext,
-        trigger: &str,
-        _payload: &Value,
-    ) -> Result<(), String> {
-        tracing::info!(
-            connection_id = ?ctx.connection_id,
-            trigger = %trigger,
-            "[composio:gmail] on_trigger"
-        );
-
-        if trigger.eq_ignore_ascii_case("GMAIL_NEW_GMAIL_MESSAGE")
-            || trigger.eq_ignore_ascii_case("GMAIL_NEW_MESSAGE")
-        {
-            if let Err(e) = self.sync(ctx, SyncReason::Manual).await {
-                tracing::warn!(
-                    error = %e,
-                    "[composio:gmail] trigger-driven sync failed (non-fatal)"
-                );
-            }
-        }
-        Ok(())
-    }
-}
+pub use provider::GmailProvider;

src/openhuman/composio/providers/gmail/provider.rs

@@ -0,0 +1,413 @@
+//! Gmail provider — incremental sync with per-item persistence.
+//!
+//! On each sync pass:
+//!
+//!   1. Load persistent [`SyncState`] from the KV store.
+//!   2. Check the daily request budget — bail early if exhausted.
+//!   3. Fetch a page of recent messages via `GMAIL_FETCH_EMAILS`, adding
+//!      a date filter when a cursor exists so only newer mail is returned.
+//!   4. Deduplicate against `synced_ids` in the state.
+//!   5. Persist each **new** message as its own memory document (not a
+//!      single giant snapshot) so agent recall can find individual emails.
+//!   6. Paginate (up to budget) until no more results or all items in the
+//!      page are already synced.
+//!   7. Advance the cursor and save state.
+//!
+//! Daily budget (`DEFAULT_DAILY_REQUEST_LIMIT`, default 500) caps the
+//! number of `execute_tool` calls per calendar day, preventing runaway
+//! API usage during large initial backfills.
+
+use async_trait::async_trait;
+use serde_json::{json, Value};
+
+use super::sync;
+use crate::openhuman::composio::providers::sync_state::{
+    extract_item_id, persist_single_item, SyncState,
+};
+use crate::openhuman::composio::providers::{
+    pick_str, ComposioProvider, ProviderContext, ProviderUserProfile, SyncOutcome, SyncReason,
+};
+
+const ACTION_GET_PROFILE: &str = "GMAIL_GET_PROFILE";
+const ACTION_FETCH_EMAILS: &str = "GMAIL_FETCH_EMAILS";
+
+/// Page size per API call. Kept moderate so each call is fast and we
+/// get frequent checkpoints for the daily budget.
+const PAGE_SIZE: u32 = 25;
+
+/// Larger page size for the very first sync after OAuth so the user
+/// gets a meaningful initial snapshot.
+const INITIAL_PAGE_SIZE: u32 = 50;
+
+/// Maximum pages to fetch in a single sync pass (guards against infinite
+/// pagination loops). Combined with PAGE_SIZE this yields at most
+/// 500 items per sync pass, well within the daily budget.
+const MAX_PAGES_PER_SYNC: u32 = 20;
+
+/// Paths to try when extracting a message's unique ID from the Composio
+/// response envelope.
+const MESSAGE_ID_PATHS: &[&str] = &["id", "data.id", "messageId", "data.messageId"];
+
+/// Paths for extracting the internal date (epoch millis or date string)
+/// used as the sync cursor.
+const MESSAGE_DATE_PATHS: &[&str] = &[
+    "internalDate",
+    "data.internalDate",
+    "date",
+    "data.date",
+    "receivedAt",
+    "data.receivedAt",
+];
+
+pub struct GmailProvider;
+
+impl GmailProvider {
+    pub fn new() -> Self {
+        Self
+    }
+}
+
+impl Default for GmailProvider {
+    fn default() -> Self {
+        Self::new()
+    }
+}
+
+#[async_trait]
+impl ComposioProvider for GmailProvider {
+    fn toolkit_slug(&self) -> &'static str {
+        "gmail"
+    }
+
+    fn sync_interval_secs(&self) -> Option<u64> {
+        Some(15 * 60)
+    }
+
+    async fn fetch_user_profile(
+        &self,
+        ctx: &ProviderContext,
+    ) -> Result<ProviderUserProfile, String> {
+        tracing::debug!(
+            connection_id = ?ctx.connection_id,
+            "[composio:gmail] fetch_user_profile via {ACTION_GET_PROFILE}"
+        );
+
+        let resp = ctx
+            .client
+            .execute_tool(ACTION_GET_PROFILE, Some(json!({})))
+            .await
+            .map_err(|e| format!("[composio:gmail] {ACTION_GET_PROFILE} failed: {e:#}"))?;
+
+        if !resp.successful {
+            let err = resp
+                .error
+                .clone()
+                .unwrap_or_else(|| "provider reported failure".to_string());
+            return Err(format!("[composio:gmail] {ACTION_GET_PROFILE}: {err}"));
+        }
+
+        let data = &resp.data;
+        let email = pick_str(
+            data,
+            &[
+                "data.emailAddress",
+                "data.email",
+                "emailAddress",
+                "email",
+                "data.profile.emailAddress",
+            ],
+        );
+        let display_name = pick_str(
+            data,
+            &[
+                "data.name",
+                "data.profile.name",
+                "name",
+                "displayName",
+                "data.displayName",
+            ],
+        )
+        .or_else(|| email.clone());
+
+        let profile = ProviderUserProfile {
+            toolkit: "gmail".to_string(),
+            connection_id: ctx.connection_id.clone(),
+            display_name,
+            email,
+            username: None,
+            avatar_url: None,
+            extras: data.clone(),
+        };
+        let has_email = profile.email.is_some();
+        let email_domain = profile
+            .email
+            .as_deref()
+            .and_then(|e| e.split('@').nth(1))
+            .map(|d| d.to_string());
+        tracing::info!(
+            connection_id = ?profile.connection_id,
+            has_email,
+            email_domain = ?email_domain,
+            "[composio:gmail] fetched user profile"
+        );
+        Ok(profile)
+    }
+
+    async fn sync(&self, ctx: &ProviderContext, reason: SyncReason) -> Result<SyncOutcome, String> {
+        let started_at_ms = sync::now_ms();
+        let connection_id = ctx
+            .connection_id
+            .clone()
+            .unwrap_or_else(|| "default".to_string());
+
+        tracing::info!(
+            connection_id = %connection_id,
+            reason = reason.as_str(),
+            "[composio:gmail] incremental sync starting"
+        );
+
+        // ── Step 1: load persistent sync state ──────────────────────
+        let Some(memory) = ctx.memory_client() else {
+            return Err("[composio:gmail] memory client not ready".to_string());
+        };
+        let mut state = SyncState::load(&memory, "gmail", &connection_id).await?;
+
+        // ── Step 2: check daily budget ──────────────────────────────
+        if state.budget_exhausted() {
+            tracing::info!(
+                connection_id = %connection_id,
+                "[composio:gmail] daily request budget exhausted, skipping sync"
+            );
+            return Ok(SyncOutcome {
+                toolkit: "gmail".to_string(),
+                connection_id: Some(connection_id),
+                reason: reason.as_str().to_string(),
+                items_ingested: 0,
+                started_at_ms,
+                finished_at_ms: sync::now_ms(),
+                summary: "gmail sync skipped: daily budget exhausted".to_string(),
+                details: json!({ "budget_exhausted": true }),
+            });
+        }
+
+        // ── Step 3: paginated incremental fetch ─────────────────────
+        let page_size = match reason {
+            SyncReason::ConnectionCreated => INITIAL_PAGE_SIZE,
+            _ => PAGE_SIZE,
+        };
+
+        let mut total_fetched: usize = 0;
+        let mut total_persisted: usize = 0;
+        let mut newest_date: Option<String> = None;
+        let mut page_token: Option<String> = None;
+
+        for page_num in 0..MAX_PAGES_PER_SYNC {
+            if state.budget_exhausted() {
+                tracing::info!(
+                    page = page_num,
+                    "[composio:gmail] budget exhausted mid-sync, stopping pagination"
+                );
+                break;
+            }
+
+            // Build the Gmail query. If we have a cursor (date of last
+            // synced message), add `after:YYYY/MM/DD` so the API only
+            // returns newer mail.
+            let mut query = "in:inbox -in:spam -in:trash".to_string();
+            if let Some(ref cursor) = state.cursor {
+                if let Some(date_filter) = sync::cursor_to_gmail_after_filter(cursor) {
+                    query.push_str(&format!(" after:{date_filter}"));
+                    tracing::debug!(
+                        page = page_num,
+                        filter = %date_filter,
+                        "[composio:gmail] using date filter from cursor"
+                    );
+                }
+            }
+
+            let mut args = json!({
+                "max_results": page_size,
+                "query": query,
+            });
+            if let Some(ref token) = page_token {
+                args["page_token"] = json!(token);
+            }
+
+            let resp = ctx
+                .client
+                .execute_tool(ACTION_FETCH_EMAILS, Some(args))
+                .await
+                .map_err(|e| {
+                    format!("[composio:gmail] {ACTION_FETCH_EMAILS} page {page_num}: {e:#}")
+                })?;
+
+            state.record_requests(1);
+
+            if !resp.successful {
+                let err = resp
+                    .error
+                    .clone()
+                    .unwrap_or_else(|| "provider reported failure".to_string());
+                // Save state so budget accounting isn't lost.
+                let _ = state.save(&memory).await;
+                return Err(format!(
+                    "[composio:gmail] {ACTION_FETCH_EMAILS} page {page_num}: {err}"
+                ));
+            }
+
+            let messages = sync::extract_messages(&resp.data);
+            total_fetched += messages.len();
+
+            if messages.is_empty() {
+                tracing::debug!(
+                    page = page_num,
+                    "[composio:gmail] empty page, stopping pagination"
+                );
+                break;
+            }
+
+            // ── Step 4: deduplicate and persist per-item ────────────
+            let mut all_already_synced = true;
+            for msg in &messages {
+                let Some(msg_id) = extract_item_id(msg, MESSAGE_ID_PATHS) else {
+                    tracing::debug!("[composio:gmail] message missing ID, skipping");
+                    continue;
+                };
+
+                // Track the newest date we've seen for cursor advancement.
+                if let Some(date_val) = extract_item_id(msg, MESSAGE_DATE_PATHS) {
+                    if newest_date
+                        .as_ref()
+                        .map_or(true, |existing| date_val > *existing)
+                    {
+                        newest_date = Some(date_val);
+                    }
+                }
+
+                if state.is_synced(&msg_id) {
+                    continue;
+                }
+                all_already_synced = false;
+
+                // Build a human-readable title for this email document.
+                let subject = pick_str(
+                    msg,
+                    &[
+                        "subject",
+                        "data.subject",
+                        "payload.headers.Subject",
+                        "snippet",
+                    ],
+                )
+                .unwrap_or_else(|| format!("Email {msg_id}"));
+                let doc_id = format!("composio-gmail-msg-{msg_id}");
+                let title = format!("Gmail: {subject}");
+
+                match persist_single_item(
+                    &memory,
+                    "gmail",
+                    &doc_id,
+                    &title,
+                    msg,
+                    "gmail",
+                    ctx.connection_id.as_deref(),
+                )
+                .await
+                {
+                    Ok(_) => {
+                        state.mark_synced(&msg_id);
+                        total_persisted += 1;
+                    }
+                    Err(e) => {
+                        tracing::warn!(
+                            msg_id = %msg_id,
+                            error = %e,
+                            "[composio:gmail] failed to persist message (continuing)"
+                        );
+                    }
+                }
+            }
+
+            // If every message in this page was already synced, there's
+            // nothing new beyond this point — stop paginating.
+            if all_already_synced {
+                tracing::debug!(
+                    page = page_num,
+                    "[composio:gmail] all items in page already synced, stopping"
+                );
+                break;
+            }
+
+            // Check for next page token.
+            page_token = sync::extract_page_token(&resp.data);
+            if page_token.is_none() {
+                tracing::debug!(page = page_num, "[composio:gmail] no next page token, done");
+                break;
+            }
+        }
+
+        // ── Step 5: advance cursor and save state ───────────────────
+        if let Some(new_cursor) = newest_date {
+            state.advance_cursor(&new_cursor);
+        }
+        state.save(&memory).await?;
+
+        let finished_at_ms = sync::now_ms();
+        let summary = format!(
+            "gmail sync ({reason}): fetched {total_fetched}, persisted {total_persisted} new, \
+             budget remaining {remaining}",
+            reason = reason.as_str(),
+            remaining = state.budget_remaining(),
+        );
+        tracing::info!(
+            connection_id = %connection_id,
+            elapsed_ms = finished_at_ms.saturating_sub(started_at_ms),
+            total_fetched,
+            total_persisted,
+            budget_remaining = state.budget_remaining(),
+            "[composio:gmail] incremental sync complete"
+        );
+
+        Ok(SyncOutcome {
+            toolkit: "gmail".to_string(),
+            connection_id: Some(connection_id),
+            reason: reason.as_str().to_string(),
+            items_ingested: total_persisted,
+            started_at_ms,
+            finished_at_ms,
+            summary,
+            details: json!({
+                "messages_fetched": total_fetched,
+                "messages_persisted": total_persisted,
+                "budget_remaining": state.budget_remaining(),
+                "cursor": state.cursor,
+                "synced_ids_total": state.synced_ids.len(),
+            }),
+        })
+    }
+
+    async fn on_trigger(
+        &self,
+        ctx: &ProviderContext,
+        trigger: &str,
+        _payload: &Value,
+    ) -> Result<(), String> {
+        tracing::info!(
+            connection_id = ?ctx.connection_id,
+            trigger = %trigger,
+            "[composio:gmail] on_trigger"
+        );
+
+        if trigger.eq_ignore_ascii_case("GMAIL_NEW_GMAIL_MESSAGE")
+            || trigger.eq_ignore_ascii_case("GMAIL_NEW_MESSAGE")
+        {
+            if let Err(e) = self.sync(ctx, SyncReason::Manual).await {
+                tracing::warn!(
+                    error = %e,
+                    "[composio:gmail] trigger-driven sync failed (non-fatal)"
+                );
+            }
+        }
+        Ok(())
+    }
+}

src/openhuman/composio/providers/helpers.rs

@@ -0,0 +1,33 @@
+//! Shared helpers for Composio provider implementations.
+
+/// Helper used by every provider's `fetch_user_profile` impl.
+///
+/// Walks a JSON object using a list of dotted-path candidates and
+/// returns the first non-empty string match. Keeps each provider's
+/// extraction code free of repetitive `as_object().and_then(...)`
+/// chains.
+pub(crate) fn pick_str(value: &serde_json::Value, paths: &[&str]) -> Option<String> {
+    for path in paths {
+        let mut cur = value;
+        let mut ok = true;
+        for segment in path.split('.') {
+            match cur.get(segment) {
+                Some(next) => cur = next,
+                None => {
+                    ok = false;
+                    break;
+                }
+            }
+        }
+        if !ok {
+            continue;
+        }
+        if let Some(s) = cur.as_str() {
+            let trimmed = s.trim();
+            if !trimmed.is_empty() {
+                return Some(trimmed.to_string());
+            }
+        }
+    }
+    None
+}

src/openhuman/composio/providers/mod.rs

@@ -32,13 +32,10 @@
 //! provider's implementation isolated, individually testable, and
 //! easy to add without touching the dispatch layer.
 
-use async_trait::async_trait;
-use serde::{Deserialize, Serialize};
-use std::sync::Arc;
-
-use crate::openhuman::config::Config;
-
-use super::client::{build_composio_client, ComposioClient};
+mod descriptions;
+pub(crate) mod helpers;
+mod traits;
+mod types;
 
 pub mod gmail;
 pub mod notion;
@@ -46,308 +43,13 @@ pub mod profile;
 pub mod registry;
 pub mod sync_state;
 
+pub use descriptions::toolkit_description;
+pub(crate) use helpers::pick_str;
 pub use registry::{
     all_providers, get_provider, init_default_providers, register_provider, ProviderArc,
 };
-
-/// Human-readable capability summary for a Composio toolkit slug.
-///
-/// Used by the prompt renderer to tell the orchestrator what each connected
-/// integration can do. Covers the most common toolkits; unknown slugs get
-/// a generic fallback so newly connected services still appear.
-pub fn toolkit_description(slug: &str) -> &'static str {
-    match slug {
-        "gmail" => {
-            "Send, read, draft, reply, forward, and search emails; manage labels and threads"
-        }
-        "notion" => "Create, read, update, and search notion pages and notion databases",
-        "github" => "Manage repositories, issues, pull requests on GitHub",
-        "slack" => "Send messages, read channels, manage threads, and post updates in Slack",
-        "discord" => "Send messages, manage channels, and interact with Discord servers",
-        "google_calendar" => "Create, update, and query calendar events; check availability",
-        "google_drive" => "Upload, download, search, and share files in Google Drive",
-        "google_docs" => "Create, read, and edit Google Docs documents",
-        "google_sheets" => "Read, write, and manage Google Sheets spreadsheets",
-        "outlook" => "Send, read, and manage emails in Microsoft Outlook",
-        "microsoft_teams" => "Send messages and manage channels in Microsoft Teams",
-        "linear" => "Create and manage issues, projects, and cycles in Linear",
-        "jira" => "Create and manage issues, projects, and sprints in Jira",
-        "trello" => "Create and manage cards, lists, and boards in Trello",
-        "asana" => "Create and manage tasks, projects, and sections in Asana",
-        "dropbox" => "Upload, download, and share files in Dropbox",
-        "twitter" => "Post tweets, read timelines, and manage Twitter interactions",
-        "spotify" => "Control playback, search music, and manage playlists on Spotify",
-        "telegram" => "Send and receive messages via Telegram",
-        "whatsapp" => "Send and receive messages via WhatsApp",
-        "twilio" => "Send SMS, make calls, and manage communications via Twilio",
-        "shopify" => "Manage products, orders, and customers in Shopify",
-        "stripe" => "Manage payments, subscriptions, and customers in Stripe",
-        "hubspot" => "Manage contacts, deals, and marketing in HubSpot",
-        "salesforce" => "Manage contacts, leads, and opportunities in Salesforce",
-        "airtable" => "Read and write records in Airtable bases",
-        "figma" => "Access and manage Figma design files and components",
-        "youtube" => "Search videos, manage playlists, and interact with YouTube",
-        "calendar" => "Create, update, and query calendar events",
-        _ => "Interact with this connected service via its available actions",
-    }
-}
-
-/// Reason a sync was triggered. Providers can use this to decide
-/// whether to do a full backfill or an incremental pull.
-#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
-#[serde(rename_all = "snake_case")]
-pub enum SyncReason {
-    /// First sync immediately after an OAuth handoff completes.
-    ConnectionCreated,
-    /// Periodic background sync from the scheduler.
-    Periodic,
-    /// Explicit user-driven sync from RPC / UI.
-    Manual,
-}
-
-impl SyncReason {
-    pub fn as_str(&self) -> &'static str {
-        match self {
-            SyncReason::ConnectionCreated => "connection_created",
-            SyncReason::Periodic => "periodic",
-            SyncReason::Manual => "manual",
-        }
-    }
-}
-
-/// Normalized user profile shape returned by every provider.
-///
-/// The shared fields (`display_name`, `email`, `username`, `avatar_url`)
-/// cover what the desktop UI actually needs to render a connected
-/// account card. Anything provider-specific (Gmail's `messagesTotal`,
-/// Notion's workspace ids, …) goes into [`extras`](Self::extras) so
-/// callers don't have to widen the shape every time a new toolkit
-/// lands.
-#[derive(Debug, Clone, Default, Serialize, Deserialize)]
-pub struct ProviderUserProfile {
-    pub toolkit: String,
-    pub connection_id: Option<String>,
-    pub display_name: Option<String>,
-    pub email: Option<String>,
-    pub username: Option<String>,
-    pub avatar_url: Option<String>,
-    /// Provider-specific extras (raw JSON object).
-    #[serde(default)]
-    pub extras: serde_json::Value,
-}
-
-/// Result of a provider sync run. Mostly used for logging + UI status.
-#[derive(Debug, Clone, Default, Serialize, Deserialize)]
-pub struct SyncOutcome {
-    pub toolkit: String,
-    pub connection_id: Option<String>,
-    pub reason: String,
-    pub items_ingested: usize,
-    pub started_at_ms: u64,
-    pub finished_at_ms: u64,
-    pub summary: String,
-    /// Provider-specific extras (raw JSON object).
-    #[serde(default)]
-    pub details: serde_json::Value,
-}
-
-impl SyncOutcome {
-    pub fn elapsed_ms(&self) -> u64 {
-        self.finished_at_ms.saturating_sub(self.started_at_ms)
-    }
-}
-
-/// Per-call context handed to provider methods.
-///
-/// `connection_id` is `None` when a method runs in a "no specific
-/// connection" mode (e.g. an across-the-board periodic sync that
-/// already iterated). For per-connection paths it is always populated.
-#[derive(Clone)]
-pub struct ProviderContext {
-    pub config: Arc<Config>,
-    pub client: ComposioClient,
-    pub toolkit: String,
-    pub connection_id: Option<String>,
-}
-
-impl ProviderContext {
-    /// Build a context from the current config + a toolkit slug.
-    ///
-    /// Returns `None` if a [`ComposioClient`] cannot be constructed
-    /// (no JWT yet — user not signed in). Callers should treat that
-    /// case as "skip silently" rather than as a hard error, mirroring
-    /// the existing op layer.
-    pub fn from_config(
-        config: Arc<Config>,
-        toolkit: impl Into<String>,
-        connection_id: Option<String>,
-    ) -> Option<Self> {
-        let client = build_composio_client(&config)?;
-        Some(Self {
-            config,
-            client,
-            toolkit: toolkit.into(),
-            connection_id,
-        })
-    }
-
-    /// Memory client handle if the global memory singleton is ready.
-    /// Used by providers that want to persist sync snapshots.
-    pub fn memory_client(&self) -> Option<crate::openhuman::memory::MemoryClientRef> {
-        crate::openhuman::memory::global::client_if_ready()
-    }
-}
-
-/// Native provider implementation for a specific Composio toolkit.
-///
-/// All methods are async and return `Result<_, String>` so the bus
-/// subscriber + RPC layer can forward errors as user-visible strings
-/// without `anyhow` round-tripping.
-#[async_trait]
-pub trait ComposioProvider: Send + Sync {
-    /// Toolkit slug (e.g. `"gmail"`). Must match the slug Composio /
-    /// the backend allowlist uses — the registry keys on this.
-    fn toolkit_slug(&self) -> &'static str;
-
-    /// Suggested periodic sync interval in seconds. Return `None` to
-    /// opt out of the periodic scheduler entirely (e.g. for write-only
-    /// providers like Slack send-message).
-    fn sync_interval_secs(&self) -> Option<u64> {
-        Some(15 * 60)
-    }
-
-    /// Fetch a normalized user profile for the current connection in
-    /// `ctx`. Most providers implement this by calling a provider
-    /// "get profile / about me" action via [`super::ops::composio_execute`].
-    async fn fetch_user_profile(
-        &self,
-        ctx: &ProviderContext,
-    ) -> Result<ProviderUserProfile, String>;
-
-    /// Run a sync pass for the current connection in `ctx`. Implementations
-    /// are responsible for persisting whatever they fetch (typically into
-    /// the memory layer via [`ProviderContext::memory_client`]).
-    async fn sync(&self, ctx: &ProviderContext, reason: SyncReason) -> Result<SyncOutcome, String>;
-
-    /// Hook fired when an OAuth handoff completes
-    /// ([`crate::core::event_bus::DomainEvent::ComposioConnectionCreated`]).
-    ///
-    /// Default impl: fetch the user profile, then run an initial sync.
-    /// Providers can override to add provider-specific bootstrapping
-    /// (e.g. registering Composio triggers, seeding labels, …).
-    async fn on_connection_created(&self, ctx: &ProviderContext) -> Result<(), String> {
-        let toolkit = self.toolkit_slug();
-        tracing::info!(
-            toolkit = %toolkit,
-            connection_id = ?ctx.connection_id,
-            "[composio:provider] on_connection_created → fetch_user_profile + initial sync"
-        );
-        match self.fetch_user_profile(ctx).await {
-            Ok(profile) => {
-                // PII discipline: do not log raw display_name or email.
-                // We log only presence indicators and the email domain
-                // (non-PII) so the trace is debuggable without leaking
-                // the user's identity. Provider-specific impls follow
-                // the same convention.
-                let has_display_name = profile.display_name.is_some();
-                let has_email = profile.email.is_some();
-                let email_domain = profile
-                    .email
-                    .as_deref()
-                    .and_then(|e| e.split('@').nth(1))
-                    .map(|d| d.to_string());
-                tracing::info!(
-                    toolkit = %toolkit,
-                    has_display_name,
-                    has_email,
-                    email_domain = ?email_domain,
-                    "[composio:provider] user profile fetched"
-                );
-
-                // Persist profile fields into the local user_profile
-                // facet table so display_name / email / avatar are
-                // available to the agent context and UI without a
-                // round-trip to the upstream provider.
-                let facets = profile::persist_provider_profile(&profile);
-                tracing::debug!(
-                    toolkit = %toolkit,
-                    facets_written = facets,
-                    "[composio:provider] profile facets persisted"
-                );
-            }
-            Err(e) => {
-                tracing::warn!(
-                    toolkit = %toolkit,
-                    error = %e,
-                    "[composio:provider] user profile fetch failed (continuing to sync)"
-                );
-            }
-        }
-        let outcome = self.sync(ctx, SyncReason::ConnectionCreated).await?;
-        tracing::info!(
-            toolkit = %toolkit,
-            items = outcome.items_ingested,
-            elapsed_ms = outcome.elapsed_ms(),
-            "[composio:provider] initial sync complete"
-        );
-        Ok(())
-    }
-
-    /// Hook fired when a Composio trigger webhook arrives for this
-    /// toolkit. `payload` is the raw provider payload as forwarded by
-    /// the backend. Implementations should be defensive — payload
-    /// shapes vary across triggers.
-    ///
-    /// Default impl: log and no-op. Most providers will want to
-    /// override this to react to specific triggers.
-    async fn on_trigger(
-        &self,
-        ctx: &ProviderContext,
-        trigger: &str,
-        payload: &serde_json::Value,
-    ) -> Result<(), String> {
-        tracing::debug!(
-            toolkit = %self.toolkit_slug(),
-            trigger = %trigger,
-            connection_id = ?ctx.connection_id,
-            payload_bytes = payload.to_string().len(),
-            "[composio:provider] on_trigger (default no-op)"
-        );
-        Ok(())
-    }
-}
-
-/// Helper used by every provider's `fetch_user_profile` impl.
-///
-/// Walks a JSON object using a list of dotted-path candidates and
-/// returns the first non-empty string match. Keeps each provider's
-/// extraction code free of repetitive `as_object().and_then(...)`
-/// chains.
-pub(crate) fn pick_str(value: &serde_json::Value, paths: &[&str]) -> Option<String> {
-    for path in paths {
-        let mut cur = value;
-        let mut ok = true;
-        for segment in path.split('.') {
-            match cur.get(segment) {
-                Some(next) => cur = next,
-                None => {
-                    ok = false;
-                    break;
-                }
-            }
-        }
-        if !ok {
-            continue;
-        }
-        if let Some(s) = cur.as_str() {
-            let trimmed = s.trim();
-            if !trimmed.is_empty() {
-                return Some(trimmed.to_string());
-            }
-        }
-    }
-    None
-}
+pub use traits::ComposioProvider;
+pub use types::{ProviderContext, ProviderUserProfile, SyncOutcome, SyncReason};
 
 #[cfg(test)]
 mod tests {

src/openhuman/composio/providers/notion/mod.rs

@@ -1,385 +1,6 @@
-//! Notion provider — incremental sync with per-item persistence.
-//!
-//! On each sync pass:
-//!
-//!   1. Load persistent [`SyncState`] from the KV store.
-//!   2. Check the daily request budget — bail early if exhausted.
-//!   3. Fetch a page of recently edited pages via `NOTION_FETCH_DATA`,
-//!      sorted by `last_edited_time` descending. When a cursor exists
-//!      we can stop as soon as we see pages older than the cursor.
-//!   4. Deduplicate against `synced_ids` in the state. Pages that have
-//!      been *edited* since their last sync are re-persisted (the cursor
-//!      is based on `last_edited_time`, so an edited page appears again).
-//!   5. Persist each **new or updated** page as its own memory document.
-//!   6. Paginate (up to budget) until no more results or all items in the
-//!      page are older than the cursor.
-//!   7. Advance the cursor and save state.
-
+mod provider;
 mod sync;
 #[cfg(test)]
 mod tests;
 
-use async_trait::async_trait;
-use serde_json::{json, Value};
-
-use super::sync_state::{extract_item_id, persist_single_item, SyncState};
-use super::{
-    pick_str, ComposioProvider, ProviderContext, ProviderUserProfile, SyncOutcome, SyncReason,
-};
-
-pub(crate) const ACTION_GET_ABOUT_ME: &str = "NOTION_GET_ABOUT_ME";
-pub(crate) const ACTION_FETCH_DATA: &str = "NOTION_FETCH_DATA";
-
-/// Page size per API call.
-const PAGE_SIZE: u32 = 25;
-
-/// Larger page size for initial sync after OAuth.
-const INITIAL_PAGE_SIZE: u32 = 50;
-
-/// Maximum pages per sync pass.
-const MAX_PAGES_PER_SYNC: u32 = 20;
-
-/// Paths for extracting a page's unique ID.
-const PAGE_ID_PATHS: &[&str] = &["id", "data.id", "pageId", "data.pageId"];
-
-/// Paths for extracting the `last_edited_time` used as sync cursor.
-const PAGE_EDITED_PATHS: &[&str] = &[
-    "last_edited_time",
-    "data.last_edited_time",
-    "lastEditedTime",
-    "data.lastEditedTime",
-];
-
-pub struct NotionProvider;
-
-impl NotionProvider {
-    pub fn new() -> Self {
-        Self
-    }
-}
-
-impl Default for NotionProvider {
-    fn default() -> Self {
-        Self::new()
-    }
-}
-
-#[async_trait]
-impl ComposioProvider for NotionProvider {
-    fn toolkit_slug(&self) -> &'static str {
-        "notion"
-    }
-
-    fn sync_interval_secs(&self) -> Option<u64> {
-        Some(30 * 60)
-    }
-
-    async fn fetch_user_profile(
-        &self,
-        ctx: &ProviderContext,
-    ) -> Result<ProviderUserProfile, String> {
-        tracing::debug!(
-            connection_id = ?ctx.connection_id,
-            "[composio:notion] fetch_user_profile via {ACTION_GET_ABOUT_ME}"
-        );
-
-        let resp = ctx
-            .client
-            .execute_tool(ACTION_GET_ABOUT_ME, Some(json!({})))
-            .await
-            .map_err(|e| format!("[composio:notion] {ACTION_GET_ABOUT_ME} failed: {e:#}"))?;
-
-        if !resp.successful {
-            let err = resp
-                .error
-                .clone()
-                .unwrap_or_else(|| "provider reported failure".to_string());
-            return Err(format!("[composio:notion] {ACTION_GET_ABOUT_ME}: {err}"));
-        }
-
-        let data = &resp.data;
-        let display_name = pick_str(
-            data,
-            &[
-                "data.name",
-                "data.user.name",
-                "name",
-                "data.bot.owner.user.name",
-            ],
-        );
-        let email = pick_str(
-            data,
-            &[
-                "data.person.email",
-                "data.user.person.email",
-                "person.email",
-                "email",
-            ],
-        );
-        let username = pick_str(
-            data,
-            &["data.bot.owner.user.id", "data.id", "id", "data.user.id"],
-        );
-        let avatar_url = pick_str(
-            data,
-            &["data.avatar_url", "data.user.avatar_url", "avatar_url"],
-        );
-
-        Ok(ProviderUserProfile {
-            toolkit: "notion".to_string(),
-            connection_id: ctx.connection_id.clone(),
-            display_name,
-            email,
-            username,
-            avatar_url,
-            extras: data.clone(),
-        })
-    }
-
-    async fn sync(&self, ctx: &ProviderContext, reason: SyncReason) -> Result<SyncOutcome, String> {
-        let started_at_ms = sync::now_ms();
-        let connection_id = ctx
-            .connection_id
-            .clone()
-            .unwrap_or_else(|| "default".to_string());
-
-        tracing::info!(
-            connection_id = %connection_id,
-            reason = reason.as_str(),
-            "[composio:notion] incremental sync starting"
-        );
-
-        // ── Step 1: load persistent sync state ──────────────────────
-        let Some(memory) = ctx.memory_client() else {
-            return Err("[composio:notion] memory client not ready".to_string());
-        };
-        let mut state = SyncState::load(&memory, "notion", &connection_id).await?;
-
-        // ── Step 2: check daily budget ──────────────────────────────
-        if state.budget_exhausted() {
-            tracing::info!(
-                connection_id = %connection_id,
-                "[composio:notion] daily request budget exhausted, skipping sync"
-            );
-            return Ok(SyncOutcome {
-                toolkit: "notion".to_string(),
-                connection_id: Some(connection_id),
-                reason: reason.as_str().to_string(),
-                items_ingested: 0,
-                started_at_ms,
-                finished_at_ms: sync::now_ms(),
-                summary: "notion sync skipped: daily budget exhausted".to_string(),
-                details: json!({ "budget_exhausted": true }),
-            });
-        }
-
-        // ── Step 3: paginated incremental fetch ─────────────────────
-        let page_size = match reason {
-            SyncReason::ConnectionCreated => INITIAL_PAGE_SIZE,
-            _ => PAGE_SIZE,
-        };
-
-        let mut total_fetched: usize = 0;
-        let mut total_persisted: usize = 0;
-        let mut newest_edited_time: Option<String> = None;
-        let mut notion_cursor: Option<String> = None;
-
-        for page_num in 0..MAX_PAGES_PER_SYNC {
-            if state.budget_exhausted() {
-                tracing::info!(
-                    page = page_num,
-                    "[composio:notion] budget exhausted mid-sync, stopping pagination"
-                );
-                break;
-            }
-
-            let mut args = json!({
-                "page_size": page_size,
-                "filter": { "value": "page", "property": "object" },
-                "sort": { "direction": "descending", "timestamp": "last_edited_time" }
-            });
-            if let Some(ref cursor) = notion_cursor {
-                args["start_cursor"] = json!(cursor);
-            }
-
-            let resp = ctx
-                .client
-                .execute_tool(ACTION_FETCH_DATA, Some(args))
-                .await
-                .map_err(|e| {
-                    format!("[composio:notion] {ACTION_FETCH_DATA} page {page_num}: {e:#}")
-                })?;
-
-            state.record_requests(1);
-
-            if !resp.successful {
-                let err = resp
-                    .error
-                    .clone()
-                    .unwrap_or_else(|| "provider reported failure".to_string());
-                let _ = state.save(&memory).await;
-                return Err(format!(
-                    "[composio:notion] {ACTION_FETCH_DATA} page {page_num}: {err}"
-                ));
-            }
-
-            let results = sync::extract_results(&resp.data);
-            total_fetched += results.len();
-
-            if results.is_empty() {
-                tracing::debug!(
-                    page = page_num,
-                    "[composio:notion] empty page, stopping pagination"
-                );
-                break;
-            }
-
-            // ── Step 4: deduplicate and persist per-item ────────────
-            let mut hit_cursor_boundary = false;
-            for page in &results {
-                let Some(page_id) = extract_item_id(page, PAGE_ID_PATHS) else {
-                    tracing::debug!("[composio:notion] page missing ID, skipping");
-                    continue;
-                };
-
-                let edited_time = extract_item_id(page, PAGE_EDITED_PATHS);
-
-                // Track the newest edited time for cursor advancement.
-                if let Some(ref et) = edited_time {
-                    if newest_edited_time
-                        .as_ref()
-                        .map_or(true, |existing| et > existing)
-                    {
-                        newest_edited_time = Some(et.clone());
-                    }
-                }
-
-                // For Notion, a page can be *edited* after we last synced
-                // it. We use a composite key of page_id + edited_time to
-                // detect this: if the page_id is in synced_ids but the
-                // edited_time is newer than the cursor, we re-sync it.
-                let sync_key = match &edited_time {
-                    Some(et) => format!("{page_id}@{et}"),
-                    None => page_id.clone(),
-                };
-
-                // If the page's edited time is older than our cursor,
-                // we've caught up — everything beyond is already synced.
-                if let (Some(ref cursor), Some(ref et)) = (&state.cursor, &edited_time) {
-                    if et <= cursor && state.is_synced(&sync_key) {
-                        hit_cursor_boundary = true;
-                        continue;
-                    }
-                }
-
-                if state.is_synced(&sync_key) {
-                    continue;
-                }
-
-                // Build a title from the page's properties.
-                let title_text = sync::extract_page_title(page)
-                    .unwrap_or_else(|| format!("Notion page {page_id}"));
-                let doc_id = format!("composio-notion-page-{page_id}");
-                let title = format!("Notion: {title_text}");
-
-                match persist_single_item(
-                    &memory,
-                    "notion",
-                    &doc_id,
-                    &title,
-                    page,
-                    "notion",
-                    ctx.connection_id.as_deref(),
-                )
-                .await
-                {
-                    Ok(_) => {
-                        state.mark_synced(&sync_key);
-                        total_persisted += 1;
-                    }
-                    Err(e) => {
-                        tracing::warn!(
-                            page_id = %page_id,
-                            error = %e,
-                            "[composio:notion] failed to persist page (continuing)"
-                        );
-                    }
-                }
-            }
-
-            if hit_cursor_boundary {
-                tracing::debug!(
-                    page = page_num,
-                    "[composio:notion] reached cursor boundary, stopping"
-                );
-                break;
-            }
-
-            // Check for next page cursor from Notion API.
-            notion_cursor = sync::extract_notion_cursor(&resp.data);
-            if notion_cursor.is_none() {
-                tracing::debug!(page = page_num, "[composio:notion] no next cursor, done");
-                break;
-            }
-        }
-
-        // ── Step 5: advance cursor and save state ───────────────────
-        if let Some(new_cursor) = newest_edited_time {
-            state.advance_cursor(&new_cursor);
-        }
-        state.save(&memory).await?;
-
-        let finished_at_ms = sync::now_ms();
-        let summary = format!(
-            "notion sync ({reason}): fetched {total_fetched}, persisted {total_persisted} new, \
-             budget remaining {remaining}",
-            reason = reason.as_str(),
-            remaining = state.budget_remaining(),
-        );
-        tracing::info!(
-            connection_id = %connection_id,
-            elapsed_ms = finished_at_ms.saturating_sub(started_at_ms),
-            total_fetched,
-            total_persisted,
-            budget_remaining = state.budget_remaining(),
-            "[composio:notion] incremental sync complete"
-        );
-
-        Ok(SyncOutcome {
-            toolkit: "notion".to_string(),
-            connection_id: Some(connection_id),
-            reason: reason.as_str().to_string(),
-            items_ingested: total_persisted,
-            started_at_ms,
-            finished_at_ms,
-            summary,
-            details: json!({
-                "results_fetched": total_fetched,
-                "results_persisted": total_persisted,
-                "budget_remaining": state.budget_remaining(),
-                "cursor": state.cursor,
-                "synced_ids_total": state.synced_ids.len(),
-            }),
-        })
-    }
-
-    async fn on_trigger(
-        &self,
-        ctx: &ProviderContext,
-        trigger: &str,
-        _payload: &Value,
-    ) -> Result<(), String> {
-        tracing::info!(
-            connection_id = ?ctx.connection_id,
-            trigger = %trigger,
-            "[composio:notion] on_trigger"
-        );
-        if let Err(e) = self.sync(ctx, SyncReason::Manual).await {
-            tracing::warn!(
-                error = %e,
-                "[composio:notion] trigger-driven sync failed (non-fatal)"
-            );
-        }
-        Ok(())
-    }
-}
+pub use provider::NotionProvider;

src/openhuman/composio/providers/notion/provider.rs

@@ -0,0 +1,384 @@
+//! Notion provider — incremental sync with per-item persistence.
+//!
+//! On each sync pass:
+//!
+//!   1. Load persistent [`SyncState`] from the KV store.
+//!   2. Check the daily request budget — bail early if exhausted.
+//!   3. Fetch a page of recently edited pages via `NOTION_FETCH_DATA`,
+//!      sorted by `last_edited_time` descending. When a cursor exists
+//!      we can stop as soon as we see pages older than the cursor.
+//!   4. Deduplicate against `synced_ids` in the state. Pages that have
+//!      been *edited* since their last sync are re-persisted (the cursor
+//!      is based on `last_edited_time`, so an edited page appears again).
+//!   5. Persist each **new or updated** page as its own memory document.
+//!   6. Paginate (up to budget) until no more results or all items in the
+//!      page are older than the cursor.
+//!   7. Advance the cursor and save state.
+
+use async_trait::async_trait;
+use serde_json::{json, Value};
+
+use super::sync;
+use crate::openhuman::composio::providers::sync_state::{
+    extract_item_id, persist_single_item, SyncState,
+};
+use crate::openhuman::composio::providers::{
+    pick_str, ComposioProvider, ProviderContext, ProviderUserProfile, SyncOutcome, SyncReason,
+};
+
+pub(crate) const ACTION_GET_ABOUT_ME: &str = "NOTION_GET_ABOUT_ME";
+pub(crate) const ACTION_FETCH_DATA: &str = "NOTION_FETCH_DATA";
+
+/// Page size per API call.
+const PAGE_SIZE: u32 = 25;
+
+/// Larger page size for initial sync after OAuth.
+const INITIAL_PAGE_SIZE: u32 = 50;
+
+/// Maximum pages per sync pass.
+const MAX_PAGES_PER_SYNC: u32 = 20;
+
+/// Paths for extracting a page's unique ID.
+const PAGE_ID_PATHS: &[&str] = &["id", "data.id", "pageId", "data.pageId"];
+
+/// Paths for extracting the `last_edited_time` used as sync cursor.
+const PAGE_EDITED_PATHS: &[&str] = &[
+    "last_edited_time",
+    "data.last_edited_time",
+    "lastEditedTime",
+    "data.lastEditedTime",
+];
+
+pub struct NotionProvider;
+
+impl NotionProvider {
+    pub fn new() -> Self {
+        Self
+    }
+}
+
+impl Default for NotionProvider {
+    fn default() -> Self {
+        Self::new()
+    }
+}
+
+#[async_trait]
+impl ComposioProvider for NotionProvider {
+    fn toolkit_slug(&self) -> &'static str {
+        "notion"
+    }
+
+    fn sync_interval_secs(&self) -> Option<u64> {
+        Some(30 * 60)
+    }
+
+    async fn fetch_user_profile(
+        &self,
+        ctx: &ProviderContext,
+    ) -> Result<ProviderUserProfile, String> {
+        tracing::debug!(
+            connection_id = ?ctx.connection_id,
+            "[composio:notion] fetch_user_profile via {ACTION_GET_ABOUT_ME}"
+        );
+
+        let resp = ctx
+            .client
+            .execute_tool(ACTION_GET_ABOUT_ME, Some(json!({})))
+            .await
+            .map_err(|e| format!("[composio:notion] {ACTION_GET_ABOUT_ME} failed: {e:#}"))?;
+
+        if !resp.successful {
+            let err = resp
+                .error
+                .clone()
+                .unwrap_or_else(|| "provider reported failure".to_string());
+            return Err(format!("[composio:notion] {ACTION_GET_ABOUT_ME}: {err}"));
+        }
+
+        let data = &resp.data;
+        let display_name = pick_str(
+            data,
+            &[
+                "data.name",
+                "data.user.name",
+                "name",
+                "data.bot.owner.user.name",
+            ],
+        );
+        let email = pick_str(
+            data,
+            &[
+                "data.person.email",
+                "data.user.person.email",
+                "person.email",
+                "email",
+            ],
+        );
+        let username = pick_str(
+            data,
+            &["data.bot.owner.user.id", "data.id", "id", "data.user.id"],
+        );
+        let avatar_url = pick_str(
+            data,
+            &["data.avatar_url", "data.user.avatar_url", "avatar_url"],
+        );
+
+        Ok(ProviderUserProfile {
+            toolkit: "notion".to_string(),
+            connection_id: ctx.connection_id.clone(),
+            display_name,
+            email,
+            username,
+            avatar_url,
+            extras: data.clone(),
+        })
+    }
+
+    async fn sync(&self, ctx: &ProviderContext, reason: SyncReason) -> Result<SyncOutcome, String> {
+        let started_at_ms = sync::now_ms();
+        let connection_id = ctx
+            .connection_id
+            .clone()
+            .unwrap_or_else(|| "default".to_string());
+
+        tracing::info!(
+            connection_id = %connection_id,
+            reason = reason.as_str(),
+            "[composio:notion] incremental sync starting"
+        );
+
+        // ── Step 1: load persistent sync state ──────────────────────
+        let Some(memory) = ctx.memory_client() else {
+            return Err("[composio:notion] memory client not ready".to_string());
+        };
+        let mut state = SyncState::load(&memory, "notion", &connection_id).await?;
+
+        // ── Step 2: check daily budget ──────────────────────────────
+        if state.budget_exhausted() {
+            tracing::info!(
+                connection_id = %connection_id,
+                "[composio:notion] daily request budget exhausted, skipping sync"
+            );
+            return Ok(SyncOutcome {
+                toolkit: "notion".to_string(),
+                connection_id: Some(connection_id),
+                reason: reason.as_str().to_string(),
+                items_ingested: 0,
+                started_at_ms,
+                finished_at_ms: sync::now_ms(),
+                summary: "notion sync skipped: daily budget exhausted".to_string(),
+                details: json!({ "budget_exhausted": true }),
+            });
+        }
+
+        // ── Step 3: paginated incremental fetch ─────────────────────
+        let page_size = match reason {
+            SyncReason::ConnectionCreated => INITIAL_PAGE_SIZE,
+            _ => PAGE_SIZE,
+        };
+
+        let mut total_fetched: usize = 0;
+        let mut total_persisted: usize = 0;
+        let mut newest_edited_time: Option<String> = None;
+        let mut notion_cursor: Option<String> = None;
+
+        for page_num in 0..MAX_PAGES_PER_SYNC {
+            if state.budget_exhausted() {
+                tracing::info!(
+                    page = page_num,
+                    "[composio:notion] budget exhausted mid-sync, stopping pagination"
+                );
+                break;
+            }
+
+            let mut args = json!({
+                "page_size": page_size,
+                "filter": { "value": "page", "property": "object" },
+                "sort": { "direction": "descending", "timestamp": "last_edited_time" }
+            });
+            if let Some(ref cursor) = notion_cursor {
+                args["start_cursor"] = json!(cursor);
+            }
+
+            let resp = ctx
+                .client
+                .execute_tool(ACTION_FETCH_DATA, Some(args))
+                .await
+                .map_err(|e| {
+                    format!("[composio:notion] {ACTION_FETCH_DATA} page {page_num}: {e:#}")
+                })?;
+
+            state.record_requests(1);
+
+            if !resp.successful {
+                let err = resp
+                    .error
+                    .clone()
+                    .unwrap_or_else(|| "provider reported failure".to_string());
+                let _ = state.save(&memory).await;
+                return Err(format!(
+                    "[composio:notion] {ACTION_FETCH_DATA} page {page_num}: {err}"
+                ));
+            }
+
+            let results = sync::extract_results(&resp.data);
+            total_fetched += results.len();
+
+            if results.is_empty() {
+                tracing::debug!(
+                    page = page_num,
+                    "[composio:notion] empty page, stopping pagination"
+                );
+                break;
+            }
+
+            // ── Step 4: deduplicate and persist per-item ────────────
+            let mut hit_cursor_boundary = false;
+            for page in &results {
+                let Some(page_id) = extract_item_id(page, PAGE_ID_PATHS) else {
+                    tracing::debug!("[composio:notion] page missing ID, skipping");
+                    continue;
+                };
+
+                let edited_time = extract_item_id(page, PAGE_EDITED_PATHS);
+
+                // Track the newest edited time for cursor advancement.
+                if let Some(ref et) = edited_time {
+                    if newest_edited_time
+                        .as_ref()
+                        .map_or(true, |existing| et > existing)
+                    {
+                        newest_edited_time = Some(et.clone());
+                    }
+                }
+
+                // For Notion, a page can be *edited* after we last synced
+                // it. We use a composite key of page_id + edited_time to
+                // detect this: if the page_id is in synced_ids but the
+                // edited_time is newer than the cursor, we re-sync it.
+                let sync_key = match &edited_time {
+                    Some(et) => format!("{page_id}@{et}"),
+                    None => page_id.clone(),
+                };
+
+                // If the page's edited time is older than our cursor,
+                // we've caught up — everything beyond is already synced.
+                if let (Some(ref cursor), Some(ref et)) = (&state.cursor, &edited_time) {
+                    if et <= cursor && state.is_synced(&sync_key) {
+                        hit_cursor_boundary = true;
+                        continue;
+                    }
+                }
+
+                if state.is_synced(&sync_key) {
+                    continue;
+                }
+
+                // Build a title from the page's properties.
+                let title_text = sync::extract_page_title(page)
+                    .unwrap_or_else(|| format!("Notion page {page_id}"));
+                let doc_id = format!("composio-notion-page-{page_id}");
+                let title = format!("Notion: {title_text}");
+
+                match persist_single_item(
+                    &memory,
+                    "notion",
+                    &doc_id,
+                    &title,
+                    page,
+                    "notion",
+                    ctx.connection_id.as_deref(),
+                )
+                .await
+                {
+                    Ok(_) => {
+                        state.mark_synced(&sync_key);
+                        total_persisted += 1;
+                    }
+                    Err(e) => {
+                        tracing::warn!(
+                            page_id = %page_id,
+                            error = %e,
+                            "[composio:notion] failed to persist page (continuing)"
+                        );
+                    }
+                }
+            }
+
+            if hit_cursor_boundary {
+                tracing::debug!(
+                    page = page_num,
+                    "[composio:notion] reached cursor boundary, stopping"
+                );
+                break;
+            }
+
+            // Check for next page cursor from Notion API.
+            notion_cursor = sync::extract_notion_cursor(&resp.data);
+            if notion_cursor.is_none() {
+                tracing::debug!(page = page_num, "[composio:notion] no next cursor, done");
+                break;
+            }
+        }
+
+        // ── Step 5: advance cursor and save state ───────────────────
+        if let Some(new_cursor) = newest_edited_time {
+            state.advance_cursor(&new_cursor);
+        }
+        state.save(&memory).await?;
+
+        let finished_at_ms = sync::now_ms();
+        let summary = format!(
+            "notion sync ({reason}): fetched {total_fetched}, persisted {total_persisted} new, \
+             budget remaining {remaining}",
+            reason = reason.as_str(),
+            remaining = state.budget_remaining(),
+        );
+        tracing::info!(
+            connection_id = %connection_id,
+            elapsed_ms = finished_at_ms.saturating_sub(started_at_ms),
+            total_fetched,
+            total_persisted,
+            budget_remaining = state.budget_remaining(),
+            "[composio:notion] incremental sync complete"
+        );
+
+        Ok(SyncOutcome {
+            toolkit: "notion".to_string(),
+            connection_id: Some(connection_id),
+            reason: reason.as_str().to_string(),
+            items_ingested: total_persisted,
+            started_at_ms,
+            finished_at_ms,
+            summary,
+            details: json!({
+                "results_fetched": total_fetched,
+                "results_persisted": total_persisted,
+                "budget_remaining": state.budget_remaining(),
+                "cursor": state.cursor,
+                "synced_ids_total": state.synced_ids.len(),
+            }),
+        })
+    }
+
+    async fn on_trigger(
+        &self,
+        ctx: &ProviderContext,
+        trigger: &str,
+        _payload: &Value,
+    ) -> Result<(), String> {
+        tracing::info!(
+            connection_id = ?ctx.connection_id,
+            trigger = %trigger,
+            "[composio:notion] on_trigger"
+        );
+        if let Err(e) = self.sync(ctx, SyncReason::Manual).await {
+            tracing::warn!(
+                error = %e,
+                "[composio:notion] trigger-driven sync failed (non-fatal)"
+            );
+        }
+        Ok(())
+    }
+}

src/openhuman/composio/providers/notion/sync.rs

@@ -3,7 +3,7 @@
 
 use serde_json::Value;
 
-use super::pick_str;
+use crate::openhuman::composio::providers::pick_str;
 
 /// Walk the Composio response envelope for Notion page results.
 pub(crate) fn extract_results(data: &Value) -> Vec<Value> {

src/openhuman/composio/providers/traits.rs

@@ -0,0 +1,124 @@
+//! The core provider trait for Composio toolkit implementations.
+
+use async_trait::async_trait;
+
+use super::types::{ProviderContext, ProviderUserProfile, SyncOutcome, SyncReason};
+
+/// Native provider implementation for a specific Composio toolkit.
+///
+/// All methods are async and return `Result<_, String>` so the bus
+/// subscriber + RPC layer can forward errors as user-visible strings
+/// without `anyhow` round-tripping.
+#[async_trait]
+pub trait ComposioProvider: Send + Sync {
+    /// Toolkit slug (e.g. `"gmail"`). Must match the slug Composio /
+    /// the backend allowlist uses — the registry keys on this.
+    fn toolkit_slug(&self) -> &'static str;
+
+    /// Suggested periodic sync interval in seconds. Return `None` to
+    /// opt out of the periodic scheduler entirely (e.g. for write-only
+    /// providers like Slack send-message).
+    fn sync_interval_secs(&self) -> Option<u64> {
+        Some(15 * 60)
+    }
+
+    /// Fetch a normalized user profile for the current connection in
+    /// `ctx`. Most providers implement this by calling a provider
+    /// "get profile / about me" action via [`super::super::ops::composio_execute`].
+    async fn fetch_user_profile(
+        &self,
+        ctx: &ProviderContext,
+    ) -> Result<ProviderUserProfile, String>;
+
+    /// Run a sync pass for the current connection in `ctx`. Implementations
+    /// are responsible for persisting whatever they fetch (typically into
+    /// the memory layer via [`ProviderContext::memory_client`]).
+    async fn sync(&self, ctx: &ProviderContext, reason: SyncReason) -> Result<SyncOutcome, String>;
+
+    /// Hook fired when an OAuth handoff completes
+    /// ([`crate::core::event_bus::DomainEvent::ComposioConnectionCreated`]).
+    ///
+    /// Default impl: fetch the user profile, then run an initial sync.
+    /// Providers can override to add provider-specific bootstrapping
+    /// (e.g. registering Composio triggers, seeding labels, …).
+    async fn on_connection_created(&self, ctx: &ProviderContext) -> Result<(), String> {
+        let toolkit = self.toolkit_slug();
+        tracing::info!(
+            toolkit = %toolkit,
+            connection_id = ?ctx.connection_id,
+            "[composio:provider] on_connection_created → fetch_user_profile + initial sync"
+        );
+        match self.fetch_user_profile(ctx).await {
+            Ok(profile) => {
+                // PII discipline: do not log raw display_name or email.
+                // We log only presence indicators and the email domain
+                // (non-PII) so the trace is debuggable without leaking
+                // the user's identity. Provider-specific impls follow
+                // the same convention.
+                let has_display_name = profile.display_name.is_some();
+                let has_email = profile.email.is_some();
+                let email_domain = profile
+                    .email
+                    .as_deref()
+                    .and_then(|e| e.split('@').nth(1))
+                    .map(|d| d.to_string());
+                tracing::info!(
+                    toolkit = %toolkit,
+                    has_display_name,
+                    has_email,
+                    email_domain = ?email_domain,
+                    "[composio:provider] user profile fetched"
+                );
+
+                // Persist profile fields into the local user_profile
+                // facet table so display_name / email / avatar are
+                // available to the agent context and UI without a
+                // round-trip to the upstream provider.
+                let facets = super::profile::persist_provider_profile(&profile);
+                tracing::debug!(
+                    toolkit = %toolkit,
+                    facets_written = facets,
+                    "[composio:provider] profile facets persisted"
+                );
+            }
+            Err(e) => {
+                tracing::warn!(
+                    toolkit = %toolkit,
+                    error = %e,
+                    "[composio:provider] user profile fetch failed (continuing to sync)"
+                );
+            }
+        }
+        let outcome = self.sync(ctx, SyncReason::ConnectionCreated).await?;
+        tracing::info!(
+            toolkit = %toolkit,
+            items = outcome.items_ingested,
+            elapsed_ms = outcome.elapsed_ms(),
+            "[composio:provider] initial sync complete"
+        );
+        Ok(())
+    }
+
+    /// Hook fired when a Composio trigger webhook arrives for this
+    /// toolkit. `payload` is the raw provider payload as forwarded by
+    /// the backend. Implementations should be defensive — payload
+    /// shapes vary across triggers.
+    ///
+    /// Default impl: log and no-op. Most providers will want to
+    /// override this to react to specific triggers.
+    async fn on_trigger(
+        &self,
+        ctx: &ProviderContext,
+        trigger: &str,
+        payload: &serde_json::Value,
+    ) -> Result<(), String> {
+        tracing::debug!(
+            toolkit = %self.toolkit_slug(),
+            trigger = %trigger,
+            connection_id = ?ctx.connection_id,
+            payload_bytes = payload.to_string().len(),
+            "[composio:provider] on_trigger (default no-op)"
+        );
+        Ok(())
+    }
+}

src/openhuman/composio/providers/types.rs

@@ -0,0 +1,113 @@
+//! Shared types for Composio provider implementations.
+
+use serde::{Deserialize, Serialize};
+use std::sync::Arc;
+
+use crate::openhuman::composio::client::{build_composio_client, ComposioClient};
+use crate::openhuman::config::Config;
+
+/// Reason a sync was triggered. Providers can use this to decide
+/// whether to do a full backfill or an incremental pull.
+#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
+#[serde(rename_all = "snake_case")]
+pub enum SyncReason {
+    /// First sync immediately after an OAuth handoff completes.
+    ConnectionCreated,
+    /// Periodic background sync from the scheduler.
+    Periodic,
+    /// Explicit user-driven sync from RPC / UI.
+    Manual,
+}
+
+impl SyncReason {
+    pub fn as_str(&self) -> &'static str {
+        match self {
+            SyncReason::ConnectionCreated => "connection_created",
+            SyncReason::Periodic => "periodic",
+            SyncReason::Manual => "manual",
+        }
+    }
+}
+
+/// Normalized user profile shape returned by every provider.
+///
+/// The shared fields (`display_name`, `email`, `username`, `avatar_url`)
+/// cover what the desktop UI actually needs to render a connected
+/// account card. Anything provider-specific (Gmail's `messagesTotal`,
+/// Notion's workspace ids, …) goes into [`extras`](Self::extras) so
+/// callers don't have to widen the shape every time a new toolkit
+/// lands.
+#[derive(Debug, Clone, Default, Serialize, Deserialize)]
+pub struct ProviderUserProfile {
+    pub toolkit: String,
+    pub connection_id: Option<String>,
+    pub display_name: Option<String>,
+    pub email: Option<String>,
+    pub username: Option<String>,
+    pub avatar_url: Option<String>,
+    /// Provider-specific extras (raw JSON object).
+    #[serde(default)]
+    pub extras: serde_json::Value,
+}
+
+/// Result of a provider sync run. Mostly used for logging + UI status.
+#[derive(Debug, Clone, Default, Serialize, Deserialize)]
+pub struct SyncOutcome {
+    pub toolkit: String,
+    pub connection_id: Option<String>,
+    pub reason: String,
+    pub items_ingested: usize,
+    pub started_at_ms: u64,
+    pub finished_at_ms: u64,
+    pub summary: String,
+    /// Provider-specific extras (raw JSON object).
+    #[serde(default)]
+    pub details: serde_json::Value,
+}
+
+impl SyncOutcome {
+    pub fn elapsed_ms(&self) -> u64 {
+        self.finished_at_ms.saturating_sub(self.started_at_ms)
+    }
+}
+
+/// Per-call context handed to provider methods.
+///
+/// `connection_id` is `None` when a method runs in a "no specific
+/// connection" mode (e.g. an across-the-board periodic sync that
+/// already iterated). For per-connection paths it is always populated.
+#[derive(Clone)]
+pub struct ProviderContext {
+    pub config: Arc<Config>,
+    pub client: ComposioClient,
+    pub toolkit: String,
+    pub connection_id: Option<String>,
+}
+
+impl ProviderContext {
+    /// Build a context from the current config + a toolkit slug.
+    ///
+    /// Returns `None` if a [`ComposioClient`] cannot be constructed
+    /// (no JWT yet — user not signed in). Callers should treat that
+    /// case as "skip silently" rather than as a hard error, mirroring
+    /// the existing op layer.
+    pub fn from_config(
+        config: Arc<Config>,
+        toolkit: impl Into<String>,
+        connection_id: Option<String>,
+    ) -> Option<Self> {
+        let client = build_composio_client(&config)?;
+        Some(Self {
+            config,
+            client,
+            toolkit: toolkit.into(),
+            connection_id,
+        })
+    }
+
+    /// Memory client handle if the global memory singleton is ready.
+    /// Used by providers that want to persist sync snapshots.
+    pub fn memory_client(&self) -> Option<crate::openhuman::memory::MemoryClientRef> {
+        crate::openhuman::memory::global::client_if_ready()
+    }
+}

src/openhuman/embeddings/factory.rs

@@ -0,0 +1,52 @@
+//! Factory functions for creating embedding providers.
+
+use std::sync::Arc;
+
+use super::provider_trait::EmbeddingProvider;
+use super::{NoopEmbedding, OllamaEmbedding, OpenAiEmbedding};
+
+/// Creates an embedding provider based on the specified name and configuration.
+///
+/// Supported provider names:
+/// - `"ollama"` → local Ollama server (default, preferred)
+/// - `"openai"` → OpenAI API
+/// - `"custom:<url>"` → OpenAI-compatible endpoint
+/// - `"none"` → no-op (keyword-only search, no embeddings)
+///
+/// Returns an error for unrecognised provider names so configuration
+/// mistakes surface immediately rather than silently degrading to
+/// keyword-only search.
+pub fn create_embedding_provider(
+    provider: &str,
+    api_key: Option<&str>,
+    model: &str,
+    dims: usize,
+) -> anyhow::Result<Box<dyn EmbeddingProvider>> {
+    match provider {
+        "ollama" => Ok(Box::new(OllamaEmbedding::new("", model, dims))),
+        "openai" => {
+            let key = api_key.unwrap_or("");
+            Ok(Box::new(OpenAiEmbedding::new(
+                "https://api.openai.com",
+                key,
+                model,
+                dims,
+            )))
+        }
+        name if name.starts_with("custom:") => {
+            let base_url = name.strip_prefix("custom:").unwrap_or("");
+            let key = api_key.unwrap_or("");
+            Ok(Box::new(OpenAiEmbedding::new(base_url, key, model, dims)))
+        }
+        "none" => Ok(Box::new(NoopEmbedding)),
+        unknown => Err(anyhow::anyhow!(
+            "unknown embedding provider: \"{unknown}\". \
+             Supported: \"ollama\", \"openai\", \"custom:<url>\", \"none\""
+        )),
+    }
+}
+
+/// Returns the default local embedding provider (Ollama-backed).
+pub fn default_local_embedding_provider() -> Arc<dyn EmbeddingProvider> {
+    Arc::new(OllamaEmbedding::default())
+}

src/openhuman/embeddings/mod.rs

@@ -7,89 +7,20 @@
 //! - **OpenAI**: Cloud-based embeddings via the OpenAI API or compatible endpoints.
 //! - **Noop**: A fallback provider for keyword-only search.
 
+mod factory;
 pub mod noop;
 pub mod ollama;
 pub mod openai;
+mod provider_trait;
 pub mod store;
 
-use std::sync::Arc;
-
-use async_trait::async_trait;
-
+pub use factory::{create_embedding_provider, default_local_embedding_provider};
 pub use noop::NoopEmbedding;
 pub use ollama::{OllamaEmbedding, DEFAULT_OLLAMA_DIMENSIONS, DEFAULT_OLLAMA_MODEL};
 pub use openai::OpenAiEmbedding;
+pub use provider_trait::EmbeddingProvider;
 pub use store::{bytes_to_vec, cosine_similarity, vec_to_bytes, SearchResult, VectorStore};
 
-/// Interface for embedding providers that convert text into numerical vectors.
-#[async_trait]
-pub trait EmbeddingProvider: Send + Sync {
-    /// Returns the name of the provider (e.g., "ollama", "openai").
-    fn name(&self) -> &str;
-
-    /// Returns the number of dimensions in the generated embeddings.
-    fn dimensions(&self) -> usize;
-
-    /// Generates embeddings for a batch of strings.
-    async fn embed(&self, texts: &[&str]) -> anyhow::Result<Vec<Vec<f32>>>;
-
-    /// Generates an embedding for a single string.
-    async fn embed_one(&self, text: &str) -> anyhow::Result<Vec<f32>> {
-        let mut results = self.embed(&[text]).await?;
-        results
-            .pop()
-            .ok_or_else(|| anyhow::anyhow!("Empty embedding result"))
-    }
-}
-
-// ── Factory ──────────────────────────────────────────────────
-
-/// Creates an embedding provider based on the specified name and configuration.
-///
-/// Supported provider names:
-/// - `"ollama"` → local Ollama server (default, preferred)
-/// - `"openai"` → OpenAI API
-/// - `"custom:<url>"` → OpenAI-compatible endpoint
-/// - `"none"` → no-op (keyword-only search, no embeddings)
-///
-/// Returns an error for unrecognised provider names so configuration
-/// mistakes surface immediately rather than silently degrading to
-/// keyword-only search.
-pub fn create_embedding_provider(
-    provider: &str,
-    api_key: Option<&str>,
-    model: &str,
-    dims: usize,
-) -> anyhow::Result<Box<dyn EmbeddingProvider>> {
-    match provider {
-        "ollama" => Ok(Box::new(OllamaEmbedding::new("", model, dims))),
-        "openai" => {
-            let key = api_key.unwrap_or("");
-            Ok(Box::new(OpenAiEmbedding::new(
-                "https://api.openai.com",
-                key,
-                model,
-                dims,
-            )))
-        }
-        name if name.starts_with("custom:") => {
-            let base_url = name.strip_prefix("custom:").unwrap_or("");
-            let key = api_key.unwrap_or("");
-            Ok(Box::new(OpenAiEmbedding::new(base_url, key, model, dims)))
-        }
-        "none" => Ok(Box::new(NoopEmbedding)),
-        unknown => Err(anyhow::anyhow!(
-            "unknown embedding provider: \"{unknown}\". \
-             Supported: \"ollama\", \"openai\", \"custom:<url>\", \"none\""
-        )),
-    }
-}
-
-/// Returns the default local embedding provider (Ollama-backed).
-pub fn default_local_embedding_provider() -> Arc<dyn EmbeddingProvider> {
-    Arc::new(OllamaEmbedding::default())
-}
-
 #[cfg(test)]
 mod tests {
     use super::*;

src/openhuman/embeddings/provider_trait.rs

@@ -0,0 +1,24 @@
+//! Interface for embedding providers that convert text into numerical vectors.
+
+use async_trait::async_trait;
+
+/// Interface for embedding providers that convert text into numerical vectors.
+#[async_trait]
+pub trait EmbeddingProvider: Send + Sync {
+    /// Returns the name of the provider (e.g., "ollama", "openai").
+    fn name(&self) -> &str;
+
+    /// Returns the number of dimensions in the generated embeddings.
+    fn dimensions(&self) -> usize;
+
+    /// Generates embeddings for a batch of strings.
+    async fn embed(&self, texts: &[&str]) -> anyhow::Result<Vec<Vec<f32>>>;
+
+    /// Generates an embedding for a single string.
+    async fn embed_one(&self, text: &str) -> anyhow::Result<Vec<f32>> {
+        let mut results = self.embed(&[text]).await?;
+        results
+            .pop()
+            .ok_or_else(|| anyhow::anyhow!("Empty embedding result"))
+    }
+}

src/openhuman/tools/impl/agent/dispatch.rs

@@ -0,0 +1,142 @@
+//! Subagent dispatch logic shared by all agent delegation tools.
+
+use crate::core::event_bus::{publish_global, DomainEvent};
+use crate::openhuman::agent::harness::definition::AgentDefinitionRegistry;
+use crate::openhuman::agent::harness::fork_context::current_parent;
+use crate::openhuman::agent::harness::subagent_runner::{run_subagent, SubagentRunOptions};
+use crate::openhuman::tools::traits::ToolResult;
+
+pub(crate) async fn dispatch_subagent(
+    agent_id: &str,
+    tool_name: &str,
+    prompt: &str,
+    skill_filter: Option<&str>,
+) -> anyhow::Result<ToolResult> {
+    let registry = match AgentDefinitionRegistry::global() {
+        Some(reg) => reg,
+        None => {
+            return Ok(ToolResult::error(
+                "Agent registry not initialised. This usually means the \
+                 core process started without calling \
+                 AgentDefinitionRegistry::init_global at startup.",
+            ));
+        }
+    };
+
+    let definition = match registry.get(agent_id) {
+        Some(def) => def,
+        None => {
+            return Ok(ToolResult::error(format!(
+                "{tool_name}: agent '{agent_id}' not found in registry"
+            )));
+        }
+    };
+
+    let parent_session = current_parent()
+        .map(|p| p.session_id.clone())
+        .unwrap_or_else(|| "standalone".into());
+    let task_id = format!("sub-{}", uuid::Uuid::new_v4());
+
+    publish_global(DomainEvent::SubagentSpawned {
+        parent_session: parent_session.clone(),
+        agent_id: definition.id.clone(),
+        mode: "typed".to_string(),
+        task_id: task_id.clone(),
+        prompt_chars: prompt.chars().count(),
+    });
+
+    log::info!(
+        "[agent] delegating to {} via {} (skill_filter={}) prompt_chars={}",
+        agent_id,
+        tool_name,
+        skill_filter.unwrap_or("<none>"),
+        prompt.chars().count()
+    );
+
+    // Propagate the per-call toolkit scope into the subagent runner so
+    // that `SkillDelegationTool`s can narrow `skills_agent` to a single
+    // Composio toolkit (e.g. `delegate_gmail` → skills_agent +
+    // toolkit="gmail"). Earlier code plumbed this through
+    // `skill_filter_override` (which matches `{skill}__` QuickJS-style
+    // names), but Composio actions are named `GMAIL_*` / `NOTION_*` —
+    // so the filter excluded every Composio tool instead of narrowing
+    // them. `toolkit_override` applies the correct `{TOOLKIT}_` prefix
+    // check, restricted to skill-category tools.
+    let options = SubagentRunOptions {
+        skill_filter_override: None,
+        category_filter_override: None,
+        toolkit_override: skill_filter.map(str::to_string),
+        context: None,
+        task_id: Some(task_id.clone()),
+    };
+
+    match run_subagent(definition, prompt, options).await {
+        Ok(outcome) => {
+            publish_global(DomainEvent::SubagentCompleted {
+                parent_session,
+                task_id: outcome.task_id.clone(),
+                agent_id: outcome.agent_id.clone(),
+                elapsed_ms: outcome.elapsed.as_millis() as u64,
+                output_chars: outcome.output.chars().count(),
+                iterations: outcome.iterations,
+            });
+            log::info!(
+                "[agent] {} completed via {} iterations={} output_chars={}",
+                agent_id,
+                tool_name,
+                outcome.iterations,
+                outcome.output.chars().count()
+            );
+            Ok(ToolResult::success(outcome.output))
+        }
+        Err(err) => {
+            let message = err.to_string();
+            publish_global(DomainEvent::SubagentFailed {
+                parent_session,
+                task_id,
+                agent_id: definition.id.clone(),
+                error: message.clone(),
+            });
+            Ok(ToolResult::error(format!("{tool_name} failed: {message}")))
+        }
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use crate::openhuman::tools::traits::Tool;
+
+    use super::super::AskClarificationTool;
+
+    #[test]
+    fn ask_clarification_tool_re_exported() {
+        let tool = AskClarificationTool::new();
+        assert_eq!(tool.name(), "ask_user_clarification");
+    }
+
+    #[tokio::test]
+    async fn dispatch_subagent_returns_tool_error_when_agent_unknown() {
+        // Exercises the graceful-failure paths of `dispatch_subagent`:
+        // without a global registry we get the "registry not initialised"
+        // branch, and with one (set by another test in the same binary)
+        // a bogus agent id hits the "agent not found" branch. Either way
+        // the function must return `Ok(ToolResult::error(..))` rather than
+        // panicking or returning `Err`.
+        let res = dispatch_subagent(
+            "__definitely_not_a_real_agent__",
+            "test_tool",
+            "irrelevant prompt",
+            None,
+        )
+        .await
+        .expect("dispatch_subagent should not return Err on these inputs");
+
+        assert!(res.is_error, "expected a tool-error ToolResult");
+        let out = res.output();
+        assert!(
+            out.contains("registry not initialised") || out.contains("not found in registry"),
+            "unexpected graceful-failure message: {out}"
+        );
+    }
+}

src/openhuman/tools/impl/agent/mod.rs

@@ -2,110 +2,11 @@ mod archetype_delegation;
 mod ask_clarification;
 pub(crate) mod complete_onboarding;
 mod delegate;
+mod dispatch;
 mod skill_delegation;
 mod spawn_subagent;
 
-use crate::core::event_bus::{publish_global, DomainEvent};
-use crate::openhuman::agent::harness::definition::AgentDefinitionRegistry;
-use crate::openhuman::agent::harness::fork_context::current_parent;
-use crate::openhuman::agent::harness::subagent_runner::{run_subagent, SubagentRunOptions};
-use crate::openhuman::tools::traits::ToolResult;
-
-pub(crate) async fn dispatch_subagent(
-    agent_id: &str,
-    tool_name: &str,
-    prompt: &str,
-    skill_filter: Option<&str>,
-) -> anyhow::Result<ToolResult> {
-    let registry = match AgentDefinitionRegistry::global() {
-        Some(reg) => reg,
-        None => {
-            return Ok(ToolResult::error(
-                "Agent registry not initialised. This usually means the \
-                 core process started without calling \
-                 AgentDefinitionRegistry::init_global at startup.",
-            ));
-        }
-    };
-
-    let definition = match registry.get(agent_id) {
-        Some(def) => def,
-        None => {
-            return Ok(ToolResult::error(format!(
-                "{tool_name}: agent '{agent_id}' not found in registry"
-            )));
-        }
-    };
-
-    let parent_session = current_parent()
-        .map(|p| p.session_id.clone())
-        .unwrap_or_else(|| "standalone".into());
-    let task_id = format!("sub-{}", uuid::Uuid::new_v4());
-
-    publish_global(DomainEvent::SubagentSpawned {
-        parent_session: parent_session.clone(),
-        agent_id: definition.id.clone(),
-        mode: "typed".to_string(),
-        task_id: task_id.clone(),
-        prompt_chars: prompt.chars().count(),
-    });
-
-    log::info!(
-        "[agent] delegating to {} via {} (skill_filter={}) prompt_chars={}",
-        agent_id,
-        tool_name,
-        skill_filter.unwrap_or("<none>"),
-        prompt.chars().count()
-    );
-
-    // Propagate the per-call toolkit scope into the subagent runner so
-    // that `SkillDelegationTool`s can narrow `skills_agent` to a single
-    // Composio toolkit (e.g. `delegate_gmail` → skills_agent +
-    // toolkit="gmail"). Earlier code plumbed this through
-    // `skill_filter_override` (which matches `{skill}__` QuickJS-style
-    // names), but Composio actions are named `GMAIL_*` / `NOTION_*` —
-    // so the filter excluded every Composio tool instead of narrowing
-    // them. `toolkit_override` applies the correct `{TOOLKIT}_` prefix
-    // check, restricted to skill-category tools.
-    let options = SubagentRunOptions {
-        skill_filter_override: None,
-        category_filter_override: None,
-        toolkit_override: skill_filter.map(str::to_string),
-        context: None,
-        task_id: Some(task_id.clone()),
-    };
-
-    match run_subagent(definition, prompt, options).await {
-        Ok(outcome) => {
-            publish_global(DomainEvent::SubagentCompleted {
-                parent_session,
-                task_id: outcome.task_id.clone(),
-                agent_id: outcome.agent_id.clone(),
-                elapsed_ms: outcome.elapsed.as_millis() as u64,
-                output_chars: outcome.output.chars().count(),
-                iterations: outcome.iterations,
-            });
-            log::info!(
-                "[agent] {} completed via {} iterations={} output_chars={}",
-                agent_id,
-                tool_name,
-                outcome.iterations,
-                outcome.output.chars().count()
-            );
-            Ok(ToolResult::success(outcome.output))
-        }
-        Err(err) => {
-            let message = err.to_string();
-            publish_global(DomainEvent::SubagentFailed {
-                parent_session,
-                task_id,
-                agent_id: definition.id.clone(),
-                error: message.clone(),
-            });
-            Ok(ToolResult::error(format!("{tool_name} failed: {message}")))
-        }
-    }
-}
+pub(crate) use dispatch::dispatch_subagent;
 
 pub use archetype_delegation::ArchetypeDelegationTool;
 pub use ask_clarification::AskClarificationTool;
@@ -113,40 +14,3 @@ pub use complete_onboarding::CompleteOnboardingTool;
 pub use delegate::DelegateTool;
 pub use skill_delegation::SkillDelegationTool;
 pub use spawn_subagent::SpawnSubagentTool;
-
-#[cfg(test)]
-mod tests {
-    use super::*;
-    use crate::openhuman::tools::traits::Tool;
-
-    #[test]
-    fn ask_clarification_tool_re_exported() {
-        let tool = AskClarificationTool::new();
-        assert_eq!(tool.name(), "ask_user_clarification");
-    }
-
-    #[tokio::test]
-    async fn dispatch_subagent_returns_tool_error_when_agent_unknown() {
-        // Exercises the graceful-failure paths of `dispatch_subagent`:
-        // without a global registry we get the "registry not initialised"
-        // branch, and with one (set by another test in the same binary)
-        // a bogus agent id hits the "agent not found" branch. Either way
-        // the function must return `Ok(ToolResult::error(..))` rather than
-        // panicking or returning `Err`.
-        let res = dispatch_subagent(
-            "__definitely_not_a_real_agent__",
-            "test_tool",
-            "irrelevant prompt",
-            None,
-        )
-        .await
-        .expect("dispatch_subagent should not return Err on these inputs");
-
-        assert!(res.is_error, "expected a tool-error ToolResult");
-        let out = res.output();
-        assert!(
-            out.contains("registry not initialised") || out.contains("not found in registry"),
-            "unexpected graceful-failure message: {out}"
-        );
-    }
-}