diff --git a/src/openhuman/agent/harness/session/builder.rs b/src/openhuman/agent/harness/session/builder.rs index 6a441accf..6508954f5 100644 --- a/src/openhuman/agent/harness/session/builder.rs +++ b/src/openhuman/agent/harness/session/builder.rs @@ -582,6 +582,8 @@ impl AgentBuilder { context, on_progress: None, connected_integrations: Vec::new(), + connected_integrations_initialized: false, + integration_runtime_config: None, // Default to `true` (omit) so legacy / custom agents built // without a definition stay lean. Opt-in agents thread their // `omit_profile = false` through the builder. @@ -1219,6 +1221,13 @@ impl Agent { None }; + // Best-effort prewarm from the shared Composio cache. This avoids + // building the session with a knowingly stale `&[]` integration view + // and then paying a repair pass on turn 1 just to recover the real + // delegation surface. + let prewarmed_integrations = crate::openhuman::composio::cached_active_integrations(config); + let prewarmed_integrations_slice = prewarmed_integrations.as_deref().unwrap_or(&[]); + // Resolve the per-agent delegation tool set and visible-tool // whitelist from the target definition (when we have one) or // fall back to the orchestrator's synthesis path. @@ -1237,12 +1246,12 @@ impl Agent { // filter. // // This builder is synchronous and sits on the CLI / REPL / - // Tauri-web code path. It does not have access to the async - // Composio fetcher, so we pass an empty slice of connected - // integrations here — the skill-wildcard expansion therefore - // produces zero delegation tools. That is correct behaviour: - // callers that need live integration expansion go through the - // bus-based `channels::runtime::dispatch` path instead. + // Tauri-web code path. It still opportunistically reuses the + // process-wide Composio cache when one is already warm, which + // lets the session start with the right `delegate_` + // surface and prompt block without paying a turn-1 fetch. On a + // cold cache we still fall back to the empty slice and let the + // first turn repair the session state if needed. let (delegation_tools, filter_from_scope): ( Vec>, Option>, @@ -1251,7 +1260,11 @@ impl Agent { crate::openhuman::agent::harness::definition::AgentDefinitionRegistry::global(), ) { (Some(def), Some(reg)) => { - let synthed = tools::orchestrator_tools::collect_orchestrator_tools(def, reg, &[]); + let synthed = tools::orchestrator_tools::collect_orchestrator_tools( + def, + reg, + prewarmed_integrations_slice, + ); let filter: Option> = match &def.tools { ToolScope::Named(names) => { let mut set: std::collections::HashSet = @@ -1271,9 +1284,11 @@ impl Agent { // callers that invoke the old `from_config` on a // pre-startup or test registry state. let synthed = match reg.get("orchestrator") { - Some(orch_def) => { - tools::orchestrator_tools::collect_orchestrator_tools(orch_def, reg, &[]) - } + Some(orch_def) => tools::orchestrator_tools::collect_orchestrator_tools( + orch_def, + reg, + prewarmed_integrations_slice, + ), None => { log::debug!( "[agent::builder] orchestrator definition not in registry — \ @@ -1343,11 +1358,15 @@ impl Agent { // cheap to guard against). let existing_names: std::collections::HashSet = tools.iter().map(|t| t.name().to_string()).collect(); - tools.extend( - delegation_tools - .into_iter() - .filter(|t| !existing_names.contains(t.name())), - ); + let inserted_delegation_tools: Vec> = delegation_tools + .into_iter() + .filter(|t| !existing_names.contains(t.name())) + .collect(); + let synthesized_tool_names: std::collections::HashSet = inserted_delegation_tools + .iter() + .map(|t| t.name().to_string()) + .collect(); + tools.extend(inserted_delegation_tools); // Pre-fetch Critical + High priority tool-scoped memory rules so they // pin into the (compression-resistant) system prompt for the whole @@ -1554,7 +1573,15 @@ impl Agent { } builder = builder.archivist_hook(archivist_hook_arc); builder = builder.unified_compaction_enabled(config.learning.unified_compaction_enabled); - builder.build() + let mut agent = builder.build()?; + let connected_integrations_initialized = prewarmed_integrations.is_some(); + agent.connected_integrations = prewarmed_integrations.unwrap_or_default(); + agent.connected_integrations_initialized = connected_integrations_initialized; + agent.integration_runtime_config = Some(config.clone()); + agent.last_seen_integrations_hash = + crate::openhuman::composio::connected_set_hash(&agent.connected_integrations); + agent.synthesized_tool_names = synthesized_tool_names; + Ok(agent) } } diff --git a/src/openhuman/agent/harness/session/runtime.rs b/src/openhuman/agent/harness/session/runtime.rs index ae93a72f4..3515906fe 100644 --- a/src/openhuman/agent/harness/session/runtime.rs +++ b/src/openhuman/agent/harness/session/runtime.rs @@ -143,6 +143,9 @@ impl Agent { integrations: Vec, ) { self.connected_integrations = integrations; + self.connected_integrations_initialized = true; + self.last_seen_integrations_hash = + crate::openhuman::composio::connected_set_hash(&self.connected_integrations); } /// The agent's runtime config snapshot. diff --git a/src/openhuman/agent/harness/session/tests.rs b/src/openhuman/agent/harness/session/tests.rs index 4bc89b887..fc33de6a8 100644 --- a/src/openhuman/agent/harness/session/tests.rs +++ b/src/openhuman/agent/harness/session/tests.rs @@ -242,6 +242,33 @@ fn agent_builder_falls_back_to_main_when_definition_name_unset() { ); } +#[test] +fn set_connected_integrations_marks_session_initialized_and_updates_hash() { + let mut agent = build_minimal_agent_with_definition_name(Some("orchestrator")); + assert!( + !agent.connected_integrations_initialized, + "fresh builder-built agents should start with placeholder integration state" + ); + + agent.set_connected_integrations(vec![ + crate::openhuman::context::prompt::ConnectedIntegration { + toolkit: "gmail".into(), + description: "Email".into(), + tools: vec![], + gated_tools: vec![], + connected: true, + }, + ]); + + assert!(agent.connected_integrations_initialized); + assert_eq!(agent.connected_integrations().len(), 1); + assert_eq!(agent.connected_integrations()[0].toolkit, "gmail"); + assert_eq!( + agent.last_seen_integrations_hash, + crate::openhuman::composio::connected_set_hash(agent.connected_integrations()) + ); +} + #[tokio::test] async fn turn_without_tools_returns_text() { let workspace = tempfile::TempDir::new().expect("temp workspace"); diff --git a/src/openhuman/agent/harness/session/turn.rs b/src/openhuman/agent/harness/session/turn.rs index eb5c39dbb..3040cd63e 100644 --- a/src/openhuman/agent/harness/session/turn.rs +++ b/src/openhuman/agent/harness/session/turn.rs @@ -95,19 +95,17 @@ impl Agent { // stored prompt verbatim to preserve the KV-cache prefix the // inference backend has already tokenised. Fetching it later // would just burn memory-store reads on data we throw away. - self.fetch_connected_integrations().await; - // The synchronous builder couldn't synthesise `delegate_*` - // tools for connected Composio toolkits (no async runtime - // handle for the Composio fetcher), so it baked in `&[]`. - // Now that integrations are live, inject the matching - // delegation tools so the orchestrator's prompt + tool-spec - // list actually expose `delegate_gmail`, `delegate_notion`, - // etc. The shared-Arc failure path returns `false`, but on - // turn 1 the Arc is uniquely owned (no sub-agent has run - // yet); a `false` return here would indicate a programmer - // error and the warn-level log inside the helper already - // surfaces it, so we ignore the return value here. - let _ = self.refresh_delegation_tools(); + if !self.connected_integrations_initialized { + self.fetch_connected_integrations().await; + // Sessions born without a cached Composio view still need + // a one-shot delegation-surface reconcile before the system + // prompt is frozen. The shared-Arc failure path returns + // `false`, but on turn 1 the Arc should still be uniquely + // owned; a `false` return here indicates a programmer error + // and the warn-level log inside the helper already surfaces + // it, so we keep the existing best-effort contract. + let _ = self.refresh_delegation_tools(); + } let learned = self.fetch_learned_context().await; let rendered_prompt = self.build_system_prompt(learned)?; log::info!("[agent] system prompt built — initialising conversation history"); @@ -176,15 +174,13 @@ impl Agent { // [`crate::openhuman::composio::cached_active_integrations`] // helper — never trigger a backend fetch ourselves, never // block on a writer. + // Session agents built through `from_config_*` carry their + // runtime `Config` snapshot directly, so this read avoids the + // old `Config::load_or_init()` round-trip on every turn. // - // We need a `Config` to key into `INTEGRATIONS_CACHE`. The - // `Config::load_or_init()` call is cached internally so this - // is cheap on the hot path. On config-load failure we skip - // the refresh — no signal we can safely act on, same as - // when the cache itself is empty. - if let Ok(cfg) = crate::openhuman::config::Config::load_or_init().await { + if let Some(cfg) = self.integration_runtime_config.as_ref() { if let Some(cache_view) = - crate::openhuman::composio::cached_active_integrations(&cfg) + crate::openhuman::composio::cached_active_integrations(cfg) { let new_hash = crate::openhuman::composio::connected_set_hash(&cache_view); if new_hash != self.last_seen_integrations_hash { @@ -205,6 +201,7 @@ impl Agent { std::mem::replace(&mut self.connected_integrations, cache_view); if self.refresh_delegation_tools() { self.last_seen_integrations_hash = new_hash; + self.connected_integrations_initialized = true; } else { // Reconcile aborted (shared Arc) — restore // the previous integration list so the @@ -1669,17 +1666,21 @@ impl Agent { /// `composio/tools.rs`, and the spawn-time per-action tool build /// path in `subagent_runner/ops.rs`. pub async fn fetch_connected_integrations(&mut self) { - let config = match crate::openhuman::config::Config::load_or_init().await { - Ok(c) => c, - Err(e) => { - log::debug!( - "[agent] skipping connected integrations fetch: config load failed: {e}" - ); - return; - } + let config = match self.integration_runtime_config.clone() { + Some(config) => config, + None => match crate::openhuman::config::Config::load_or_init().await { + Ok(config) => config, + Err(e) => { + log::debug!( + "[agent] skipping connected integrations fetch: config load failed: {e}" + ); + return; + } + }, }; self.connected_integrations = crate::openhuman::composio::fetch_connected_integrations(&config).await; + self.connected_integrations_initialized = true; } /// Re-synthesise `delegate_*` tools for the orchestrator's `subagents` @@ -1709,10 +1710,10 @@ impl Agent { /// guarantees every final entry is either a non-synthesised /// direct tool or a member of the fresh `synthed` set. /// - /// **When to call**: on turn 1 (after [`Agent::fetch_connected_integrations`] - /// populates `self.connected_integrations` for the first time) and - /// on any subsequent turn where the connection set has changed - /// since the last reconcile (detected via + /// **When to call**: on turn 1 only when the session was built + /// without a prewarmed Composio cache snapshot, and on any + /// subsequent turn where the connection set has changed since the + /// last reconcile (detected via /// [`Self::last_seen_integrations_hash`] vs. /// [`crate::openhuman::composio::cached_active_integrations`]). /// diff --git a/src/openhuman/agent/harness/session/types.rs b/src/openhuman/agent/harness/session/types.rs index a0a450d09..bfa18f5a9 100644 --- a/src/openhuman/agent/harness/session/types.rs +++ b/src/openhuman/agent/harness/session/types.rs @@ -139,6 +139,18 @@ pub struct Agent { /// the delegator / skill-executor voices can render their own /// integration blocks. pub(super) connected_integrations: Vec, + /// Whether `connected_integrations` is an authoritative session-start + /// snapshot (prewarmed from the shared Composio cache or fetched + /// explicitly) versus the default empty placeholder installed by + /// `AgentBuilder::build`. Turn 1 uses this to decide whether it must + /// still pay the cold-start fetch cost before freezing the system prompt. + pub(super) connected_integrations_initialized: bool, + /// Full runtime config snapshot for integration-cache reads and the + /// best-effort fallback fetch path. Session agents built from + /// `Config` carry this directly so the turn loop does not need to + /// re-run `Config::load_or_init()` on the hot path just to key into + /// the Composio cache. + pub(super) integration_runtime_config: Option, /// Mirrors the agent definition's `omit_profile` flag. Threaded into /// [`PromptContext::include_profile`] in `turn::build_system_prompt` /// so only user-facing agents (welcome, orchestrator, triggers)