diff --git a/src/openhuman/about_app/mod.rs b/src/openhuman/about_app/mod.rs index 7328066ed..fa548b3a6 100644 --- a/src/openhuman/about_app/mod.rs +++ b/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) -> RpcOutcome> { - 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, 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> { - let capabilities = search(query); - let log = format!( - "about_app.search returned {} capabilities for '{}'", - capabilities.len(), - query.trim() - ); - RpcOutcome::single_log(capabilities, log) -} diff --git a/src/openhuman/about_app/ops.rs b/src/openhuman/about_app/ops.rs new file mode 100644 index 000000000..6084d3bfe --- /dev/null +++ b/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) -> RpcOutcome> { + 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, 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> { + let capabilities = search(query); + let log = format!( + "about_app.search returned {} capabilities for '{}'", + capabilities.len(), + query.trim() + ); + RpcOutcome::single_log(capabilities, log) +} diff --git a/src/openhuman/agent/agents/loader.rs b/src/openhuman/agent/agents/loader.rs new file mode 100644 index 000000000..47769c0d4 --- /dev/null +++ b/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`, 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> { + BUILTINS.iter().map(parse_builtin).collect() +} + +/// Parse a single [`BuiltinAgent`] triple into a finished [`AgentDefinition`]. +fn parse_builtin(b: &BuiltinAgent) -> Result { + // 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); + } +} diff --git a/src/openhuman/agent/agents/mod.rs b/src/openhuman/agent/agents/mod.rs index 47769c0d4..ca9184dcb 100644 --- a/src/openhuman/agent/agents/mod.rs +++ b/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`, 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> { - BUILTINS.iter().map(parse_builtin).collect() -} - -/// Parse a single [`BuiltinAgent`] triple into a finished [`AgentDefinition`]. -fn parse_builtin(b: &BuiltinAgent) -> Result { - // 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}; diff --git a/src/openhuman/composio/providers/descriptions.rs b/src/openhuman/composio/providers/descriptions.rs new file mode 100644 index 000000000..7db61a47a --- /dev/null +++ b/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", + } +} diff --git a/src/openhuman/composio/providers/gmail/mod.rs b/src/openhuman/composio/providers/gmail/mod.rs index f46f9d634..23f03dac2 100644 --- a/src/openhuman/composio/providers/gmail/mod.rs +++ b/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 { - Some(15 * 60) - } - - async fn fetch_user_profile( - &self, - ctx: &ProviderContext, - ) -> Result { - 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 { - 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 = None; - let mut page_token: Option = 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; diff --git a/src/openhuman/composio/providers/gmail/provider.rs b/src/openhuman/composio/providers/gmail/provider.rs new file mode 100644 index 000000000..05c333a17 --- /dev/null +++ b/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 { + Some(15 * 60) + } + + async fn fetch_user_profile( + &self, + ctx: &ProviderContext, + ) -> Result { + 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 { + 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 = None; + let mut page_token: Option = 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(()) + } +} diff --git a/src/openhuman/composio/providers/helpers.rs b/src/openhuman/composio/providers/helpers.rs new file mode 100644 index 000000000..a4b096019 --- /dev/null +++ b/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 { + 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 +} diff --git a/src/openhuman/composio/providers/mod.rs b/src/openhuman/composio/providers/mod.rs index 6f8f5c6b2..4bd5bacd4 100644 --- a/src/openhuman/composio/providers/mod.rs +++ b/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, - pub display_name: Option, - pub email: Option, - pub username: Option, - pub avatar_url: Option, - /// 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, - 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, - pub client: ComposioClient, - pub toolkit: String, - pub connection_id: Option, -} - -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, - toolkit: impl Into, - connection_id: Option, - ) -> Option { - 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::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 { - 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; - - /// 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; - - /// 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 { - 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 { diff --git a/src/openhuman/composio/providers/notion/mod.rs b/src/openhuman/composio/providers/notion/mod.rs index da1b122c1..0a94909c0 100644 --- a/src/openhuman/composio/providers/notion/mod.rs +++ b/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 { - Some(30 * 60) - } - - async fn fetch_user_profile( - &self, - ctx: &ProviderContext, - ) -> Result { - 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 { - 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 = None; - let mut notion_cursor: Option = 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; diff --git a/src/openhuman/composio/providers/notion/provider.rs b/src/openhuman/composio/providers/notion/provider.rs new file mode 100644 index 000000000..f11a7df5f --- /dev/null +++ b/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 { + Some(30 * 60) + } + + async fn fetch_user_profile( + &self, + ctx: &ProviderContext, + ) -> Result { + 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 { + 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 = None; + let mut notion_cursor: Option = 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(()) + } +} diff --git a/src/openhuman/composio/providers/notion/sync.rs b/src/openhuman/composio/providers/notion/sync.rs index e8c01e180..4eb494212 100644 --- a/src/openhuman/composio/providers/notion/sync.rs +++ b/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 { diff --git a/src/openhuman/composio/providers/traits.rs b/src/openhuman/composio/providers/traits.rs new file mode 100644 index 000000000..c1aec19e4 --- /dev/null +++ b/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 { + 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; + + /// 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; + + /// 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(()) + } +} diff --git a/src/openhuman/composio/providers/types.rs b/src/openhuman/composio/providers/types.rs new file mode 100644 index 000000000..179bf5fdb --- /dev/null +++ b/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, + pub display_name: Option, + pub email: Option, + pub username: Option, + pub avatar_url: Option, + /// 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, + 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, + pub client: ComposioClient, + pub toolkit: String, + pub connection_id: Option, +} + +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, + toolkit: impl Into, + connection_id: Option, + ) -> Option { + 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::global::client_if_ready() + } +} diff --git a/src/openhuman/embeddings/factory.rs b/src/openhuman/embeddings/factory.rs new file mode 100644 index 000000000..dba0ef33d --- /dev/null +++ b/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:"` → 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> { + 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:\", \"none\"" + )), + } +} + +/// Returns the default local embedding provider (Ollama-backed). +pub fn default_local_embedding_provider() -> Arc { + Arc::new(OllamaEmbedding::default()) +} diff --git a/src/openhuman/embeddings/mod.rs b/src/openhuman/embeddings/mod.rs index fd36129f8..0d4b61d71 100644 --- a/src/openhuman/embeddings/mod.rs +++ b/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>>; - - /// Generates an embedding for a single string. - async fn embed_one(&self, text: &str) -> anyhow::Result> { - 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:"` → 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> { - 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:\", \"none\"" - )), - } -} - -/// Returns the default local embedding provider (Ollama-backed). -pub fn default_local_embedding_provider() -> Arc { - Arc::new(OllamaEmbedding::default()) -} - #[cfg(test)] mod tests { use super::*; diff --git a/src/openhuman/embeddings/provider_trait.rs b/src/openhuman/embeddings/provider_trait.rs new file mode 100644 index 000000000..2927c7184 --- /dev/null +++ b/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>>; + + /// Generates an embedding for a single string. + async fn embed_one(&self, text: &str) -> anyhow::Result> { + let mut results = self.embed(&[text]).await?; + results + .pop() + .ok_or_else(|| anyhow::anyhow!("Empty embedding result")) + } +} diff --git a/src/openhuman/tools/impl/agent/dispatch.rs b/src/openhuman/tools/impl/agent/dispatch.rs new file mode 100644 index 000000000..0bc7d2618 --- /dev/null +++ b/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 { + 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(""), + 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}" + ); + } +} diff --git a/src/openhuman/tools/impl/agent/mod.rs b/src/openhuman/tools/impl/agent/mod.rs index 26b619d05..f9f22fc12 100644 --- a/src/openhuman/tools/impl/agent/mod.rs +++ b/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 { - 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(""), - 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}" - ); - } -}