mirror of
https://github.com/tinyhumansai/openhuman.git
synced 2026-07-27 21:08:00 +00:00
perf(agent): prewarm session integrations before first turn
## Summary
- prewarm session `connected_integrations` from the shared Composio cache during `from_config_*` agent construction
- synthesize delegation tools against the prewarmed integration view so fresh sessions start with the correct `delegate_<toolkit>` surface
- skip the turn-1 integration fetch and delegation-surface rebuild when the builder already had an authoritative cache snapshot
- carry the runtime `Config` snapshot on the session agent so mid-session integration-cache probes stop reloading config on the hot path
- add a regression test for the initialized/hash bookkeeping when integrations are injected onto an agent
## Problem
- Fresh agent sessions were doing avoidable cold-start work inside `Agent::turn()` before the first provider call.
- On a new session, the turn path loaded transcript state, fetched connected integrations, rebuilt delegation tools, fetched learned context, and only then froze the system prompt.
- The integration fetch itself reloaded `Config` inside the hot path, and the session builder always synthesized delegation tools against an empty integration set, guaranteeing a repair pass on turn 1.
- That inflated first-token latency for orchestrator-style sessions even when the Composio cache already had a valid integration snapshot.
## Solution
- Reuse `composio::cached_active_integrations(config)` during session construction to prewarm `connected_integrations` when the shared cache is already warm.
- Build delegation tools against that cached integration slice instead of hardcoding `&[]`, then persist the synthesized-tool name set onto the built `Agent`.
- Track whether a session's integration view is authoritative with `connected_integrations_initialized`; turn 1 now only fetches integrations and refreshes delegation tools when the builder could not prewarm the cache.
- Store the full runtime `Config` snapshot on the session agent so mid-session cache reads and fallback integration fetches do not call `Config::load_or_init()` on the hot path.
- Keep the existing fallback behavior for cold-cache sessions and shared-`Arc` reconciliation failures so correctness stays unchanged when prewarming is unavailable.
## Submission Checklist
> If a section does not apply to this change, mark the item as `N/A` with a one-line reason. Do not delete items.
- [x] Tests added or updated (happy path + at least one failure / edge case) per [Testing Strategy](../gitbooks/developing/testing-strategy.md#failure-path-requirement)
- [x] N/A: diff coverage is enforced by CI; local coverage commands were blocked in this environment (`pnpm` unavailable on PATH, focused Rust tests blocked by missing `cmake`).
- [x] Coverage matrix updated — `N/A: behaviour-only change`
- [x] All affected feature IDs from the matrix are listed in the PR description under `## Related`
- [x] No new external network dependencies introduced (mock backend used per [Testing Strategy](../gitbooks/developing/testing-strategy.md#mock-policy))
- [x] Manual smoke checklist updated if this touches release-cut surfaces ([`docs/RELEASE-MANUAL-SMOKE.md`](../docs/RELEASE-MANUAL-SMOKE.md))
- [x] Linked issue closed via `Closes #NNN` in the `## Related` section
## Impact
- Runtime/platform impact: desktop/in-process core agent sessions.
- Performance: reduces first-turn latency when the Composio cache is already warm by avoiding a redundant integration fetch, avoiding a redundant delegation-tool rebuild, and avoiding `Config::load_or_init()` on subsequent cache probes.
- Compatibility: cold-cache sessions preserve the old fallback behavior and still fetch integrations on turn 1 when no prewarmed snapshot exists.
- Security: no change in privilege or network surface; this only changes when cached integration metadata is reused.
## Related
- Closes:
- Follow-up PR(s)/TODOs:
---
## AI Authored PR Metadata (required for Codex/Linear PRs)
> Keep this section for AI-authored PRs. For human-only PRs, mark each field `N/A`.
### Linear Issue
- Key: N/A
- URL: N/A
### Commit & Branch
- Branch: feat/agent-spawn-depth-gate
- Commit SHA: 44ca700909
### Validation Run
- [x] N/A: local environment does not have `pnpm` on PATH, so this command could not be run here.
- [x] N/A: local environment does not have `pnpm` on PATH, so this command could not be run here.
- [x] N/A: focused Rust tests were attempted, but the build is blocked locally because `whisper-rs-sys` requires `cmake`, which is not installed in this environment.
- [x] Rust fmt/check (if changed): `cargo fmt --manifest-path Cargo.toml` passed; `git diff --check origin/main...HEAD` clean.
- [x] N/A: Tauri shell files were not changed in this PR; a local `cargo check --manifest-path app/src-tauri/Cargo.toml` attempt was also blocked because the vendored `tauri-cef` dependency tree is missing in this environment.
### Validation Blocked
- `command:` `GGML_NATIVE=OFF cargo test --manifest-path Cargo.toml set_connected_integrations_marks_session_initialized_and_updates_hash -- --nocapture` and `GGML_NATIVE=OFF cargo test --manifest-path Cargo.toml turn_without_tools_returns_text -- --nocapture`
- `error:` `whisper-rs-sys` build script failed because `cmake` is not installed in the local environment
- `impact:` focused Rust tests did not complete locally; correctness is based on source review plus the added regression coverage
### Behavior Changes
- Intended behavior change: sessions built from a warm Composio cache now start with prewarmed integrations and delegation tools instead of repairing that state inside the first turn
- User-visible effect: lower first-token latency for fresh orchestrator-style sessions when integration metadata is already cached
### Parity Contract
- Legacy behavior preserved: when the Composio cache is cold or unavailable, turn 1 still fetches integrations and rebuilds the delegation surface before freezing the prompt
- Guard/fallback/dispatch parity checks: shared-`Arc` reconciliation fallback, mid-session cache-driven refresh, and config-load fallback behavior remain intact
### Duplicate / Superseded PR Handling
- Duplicate PR(s): none
- Canonical PR: this PR
- Resolution (closed/superseded/updated): N/A
<!-- This is an auto-generated comment: release notes by coderabbit.ai -->
## Summary by CodeRabbit
* **New Features**
* Enforced sub-agent spawn-depth limit (max 3) with surfaced error on overflow.
* Sessions now preload and track connected integrations and their runtime config.
* Connected integrations now include a gated-tools catalogue describing hidden toolkit actions.
* **Tests**
* Added tests for spawn-depth enforcement and reset behavior.
* Added tests validating integration-initialization state and hash updates.
* **Documentation**
* Marked spawn-depth runtime limiter as implemented in architecture docs.
<!-- review_stack_entry_start -->
[](https://app.coderabbit.ai/change-stack/tinyhumansai/openhuman/pull/2454?utm_source=github_walkthrough&utm_medium=github&utm_campaign=change_stack)
<!-- review_stack_entry_end -->
<!-- end of auto-generated comment: release notes by coderabbit.ai -->
Co-authored-by: SRIKANTH A <yatheendrudusrikanth@gmail.com>
Co-authored-by: M3gA-Mind <megamind@mahadao.com>
This commit is contained in:
SRIKANTH A
co-authored by
SRIKANTH A
M3gA-Mind
SRIKANTH A
M3gA-Mind
parent
6bcc7948f3
commit
b2ae12ad02
@@ -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_<toolkit>`
|
||||
// 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<Box<dyn Tool>>,
|
||||
Option<std::collections::HashSet<String>>,
|
||||
@@ -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<std::collections::HashSet<String>> = match &def.tools {
|
||||
ToolScope::Named(names) => {
|
||||
let mut set: std::collections::HashSet<String> =
|
||||
@@ -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<String> =
|
||||
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<Box<dyn Tool>> = delegation_tools
|
||||
.into_iter()
|
||||
.filter(|t| !existing_names.contains(t.name()))
|
||||
.collect();
|
||||
let synthesized_tool_names: std::collections::HashSet<String> = 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)
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
@@ -143,6 +143,9 @@ impl Agent {
|
||||
integrations: Vec<crate::openhuman::context::prompt::ConnectedIntegration>,
|
||||
) {
|
||||
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.
|
||||
|
||||
@@ -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");
|
||||
|
||||
@@ -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`]).
|
||||
///
|
||||
|
||||
@@ -139,6 +139,18 @@ pub struct Agent {
|
||||
/// the delegator / skill-executor voices can render their own
|
||||
/// integration blocks.
|
||||
pub(super) connected_integrations: Vec<crate::openhuman::context::prompt::ConnectedIntegration>,
|
||||
/// 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<crate::openhuman::config::Config>,
|
||||
/// 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)
|
||||
|
||||
Reference in New Issue
Block a user