From 0f430aece329f5d3082b87db24848db90176d3b4 Mon Sep 17 00:00:00 2001 From: Steven Enamakel Date: Thu, 24 Sep 2026 04:51:55 +0530 Subject: [PATCH 01/33] fix(session): hydrate integrations on every session start, not only cold threads The turn boundary refresh previously skipped integration hydration for resumed threads, leaving them with an empty integration surface when the session was rebuilt after a restart or cache expiry. This change always calls `refresh_cold_integrations` on the first turn of any session instance, and introduces a fallback mechanism that uses a stale cached snapshot when the backend is unreachable, retrying on the next turn only when no snapshot exists at all. Auto-committed-on: macbook --- .../src/agent/session_host/runtime_session.rs | 57 +++++++++++++++++-- 1 file changed, 51 insertions(+), 6 deletions(-) diff --git a/crates/openhuman-core/src/agent/session_host/runtime_session.rs b/crates/openhuman-core/src/agent/session_host/runtime_session.rs index f85f6bc429..271ef54f1f 100644 --- a/crates/openhuman-core/src/agent/session_host/runtime_session.rs +++ b/crates/openhuman-core/src/agent/session_host/runtime_session.rs @@ -223,9 +223,15 @@ impl OpenHumanTurnPrelude { } async fn refresh_turn_boundary(&self, cold: bool) { - if cold { - self.refresh_cold_integrations().await; - } else { + // Hydrate on the first turn of *this session instance*, not only on a + // brand-new thread. A resumed thread is never `cold`, and a session + // rebuilt after a restart (or any rebuild past the 60 s integrations + // cache TTL) is seeded from an empty cache — gating the fetch on + // `cold` left it with zero integrations, no deferred Composio + // actions, and no `tool_search` bridge for the whole thread. + // `refresh_cold_integrations` is a no-op once hydrated. + self.refresh_cold_integrations().await; + if !cold { self.refresh_dynamic_announcements().await; } // Integration changes are authority changes, not only display @@ -419,7 +425,20 @@ impl OpenHumanTurnPrelude { let Some(config) = config else { return; }; - let connected = crate::integrations::composio::fetch_connected_integrations(&config).await; + let Some(connected) = load_connected_integrations(&config).await else { + // Backend unreachable and nothing cached: stay un-hydrated so the + // next turn retries rather than pinning an empty surface. + log::warn!( + "[session] integrations unavailable and no cached snapshot; will retry next turn agent={}", + self.agent_definition_id + ); + return; + }; + log::info!( + "[session] hydrated connected integrations count={} agent={}", + connected.len(), + self.agent_definition_id + ); let mcp_servers = crate::mcp::registry::connections::connected_overview() .await .into_iter() @@ -442,8 +461,13 @@ impl OpenHumanTurnPrelude { async fn refresh_dynamic_announcements(&self) { let skills_changed = self.drain_host_events(); if let Some(config) = self.runtime_config.as_deref() { - if let Some(current) = crate::integrations::composio::cached_active_integrations(config) - { + // An expired cache is refetched rather than skipped, so a + // long-lived session keeps tracking connects/revokes. + let current = match crate::integrations::composio::cached_active_integrations(config) { + Some(current) => Some(current), + None => load_connected_integrations(config).await, + }; + if let Some(current) = current { let mut mutable = self .mutable .lock() @@ -1985,3 +2009,24 @@ impl OpenHumanSessionHost { } } } + +/// Live connected integrations, falling back to the last cached snapshot +/// (even past its TTL) when the backend is unreachable. `None` only when +/// there is neither a live answer nor any snapshot to fall back to. +async fn load_connected_integrations( + config: &crate::config::Config, +) -> Option> { + use crate::integrations::composio::FetchConnectedIntegrationsStatus; + match crate::integrations::composio::fetch_connected_integrations_status(config).await { + FetchConnectedIntegrationsStatus::Authoritative(connected) => Some(connected), + FetchConnectedIntegrationsStatus::Unavailable => { + let stale = + crate::integrations::composio::cached_active_integrations_including_expired(config); + log::warn!( + "[session] integrations fetch unavailable; using stale snapshot={}", + stale.as_ref().map_or(0, Vec::len) + ); + stale + } + } +} From ad05dabbab3ebf025a9d6ef12fe8b3dfe4dc0e01 Mon Sep 17 00:00:00 2001 From: Steven Enamakel Date: Thu, 24 Sep 2026 04:53:23 +0530 Subject: [PATCH 02/33] fix(test): add missing `tools` field to test transcript constructors The `SessionTranscript` struct gained a new `tools` field, causing compilation failures in test code that constructed instances without it. This change adds `tools: None` to all test transcript constructors and updates the `tinyagents` submodule to restore a clean build. Auto-committed-on: macbook --- .../learning/transcript_ingest/transcript_ingest_tests.rs | 5 +++++ crates/openhuman-core/src/agent/session_import/live_tests.rs | 3 +++ vendor/tinyagents | 2 +- 3 files changed, 9 insertions(+), 1 deletion(-) diff --git a/crates/openhuman-core/src/agent/learning/transcript_ingest/transcript_ingest_tests.rs b/crates/openhuman-core/src/agent/learning/transcript_ingest/transcript_ingest_tests.rs index f061e7793b..21dc967180 100644 --- a/crates/openhuman-core/src/agent/learning/transcript_ingest/transcript_ingest_tests.rs +++ b/crates/openhuman-core/src/agent/learning/transcript_ingest/transcript_ingest_tests.rs @@ -202,6 +202,7 @@ fn fake_meta(thread_id: Option<&str>) -> TranscriptMeta { async fn ingest_extracts_high_importance_preference_with_provenance() { let mem = InMemory::new(); let transcript = SessionTranscript { + tools: None, meta: fake_meta(Some("thr_alpha")), messages: durable_messages([ ChatMessage::user("hi"), @@ -236,6 +237,7 @@ async fn ingest_extracts_high_importance_preference_with_provenance() { async fn re_ingest_is_idempotent() { let mem = InMemory::new(); let transcript = SessionTranscript { + tools: None, meta: fake_meta(Some("thr_beta")), messages: durable_messages([ChatMessage::user( "I prefer Postgres for everything new — please default to it.", @@ -260,6 +262,7 @@ async fn re_ingest_is_idempotent() { async fn ingest_captures_user_reflection_and_recurring_pattern() { let mem = InMemory::new(); let transcript = SessionTranscript { + tools: None, meta: fake_meta(Some("thr_gamma")), messages: durable_messages([ ChatMessage::user("I prefer terse responses with no preamble."), @@ -296,6 +299,7 @@ async fn ingest_captures_user_reflection_and_recurring_pattern() { async fn ingest_filters_low_signal_chatter() { let mem = InMemory::new(); let transcript = SessionTranscript { + tools: None, meta: fake_meta(None), messages: durable_messages([ ChatMessage::user("ok"), @@ -323,6 +327,7 @@ async fn ingest_persists_candidates_with_bounded_concurrency() { // PERSIST_CONCURRENCY (8), so an unbounded fan-out would push more than 8 // stores in flight at once — the bound assertion below would then fail. let transcript = SessionTranscript { + tools: None, meta: fake_meta(Some("thr_bound")), messages: durable_messages([ ChatMessage::user("I prefer Postgres over MySQL for new metadata services."), diff --git a/crates/openhuman-core/src/agent/session_import/live_tests.rs b/crates/openhuman-core/src/agent/session_import/live_tests.rs index 277a76acdf..cc32c835e4 100644 --- a/crates/openhuman-core/src/agent/session_import/live_tests.rs +++ b/crates/openhuman-core/src/agent/session_import/live_tests.rs @@ -317,6 +317,7 @@ async fn in_memory_store_reconstruction_diverges_from_legacy_on_sidecar_metadata .expect("assistant message present"); attach_chat_turn_usage_metadata(&mut live_messages[last_assistant], &usage); let reconstructed = SessionTranscript { + tools: None, meta: meta.clone(), messages: durable_messages(&live_messages), }; @@ -352,6 +353,7 @@ async fn shadow_read_unavailable_and_divergence() { // No store write yet: empty/absent stream against a non-empty legacy // transcript → Unavailable (no shadow), never a divergence. let legacy = SessionTranscript { + tools: None, meta: meta.clone(), messages: durable_messages(&[ChatMessage::user("hi"), ChatMessage::assistant("done")]), }; @@ -367,6 +369,7 @@ async fn shadow_read_unavailable_and_divergence() { .await .expect("live dual-write"); let diverging = SessionTranscript { + tools: None, meta, messages: durable_messages(&[ ChatMessage::user("hi"), diff --git a/vendor/tinyagents b/vendor/tinyagents index 157186cdaf..a862d89882 160000 --- a/vendor/tinyagents +++ b/vendor/tinyagents @@ -1 +1 @@ -Subproject commit 157186cdaf1b2a243bac9cd65fbf2eed9350d17d +Subproject commit a862d89882b61a752f01b4c7b9495b0aa901432d From fd85718a4b910cd0e934e6b5910652fb3456c060 Mon Sep 17 00:00:00 2001 From: Steven Enamakel Date: Thu, 24 Sep 2026 04:53:57 +0530 Subject: [PATCH 03/33] chore(deps): add tinyagents vendor dependency Added the tinyagents library as a vendored dependency to support agent-based workflows in the project. This change introduces the necessary source files and metadata for the library to be used directly from the vendor directory. Auto-committed-on: macbook --- vendor/tinyagents | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/vendor/tinyagents b/vendor/tinyagents index a862d89882..a5a0e25c64 160000 --- a/vendor/tinyagents +++ b/vendor/tinyagents @@ -1 +1 @@ -Subproject commit a862d89882b61a752f01b4c7b9495b0aa901432d +Subproject commit a5a0e25c64a0023d7b87114e384d54d3f3a2db79 From 94fbb14975d5978d74bee5e95f5e8b72d04e2592 Mon Sep 17 00:00:00 2001 From: Steven Enamakel Date: Thu, 24 Sep 2026 04:54:22 +0530 Subject: [PATCH 04/33] chore(deps): update tinyagents submodule Updated the pinned commit of the tinyagents vendored dependency to include the latest upstream changes. Auto-committed-on: macbook --- vendor/tinyagents | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/vendor/tinyagents b/vendor/tinyagents index a5a0e25c64..9ae353a748 160000 --- a/vendor/tinyagents +++ b/vendor/tinyagents @@ -1 +1 @@ -Subproject commit a5a0e25c64a0023d7b87114e384d54d3f3a2db79 +Subproject commit 9ae353a7481e8903342b4caeb2f3bd7ffdd4179a From 39283e2369849f1a1a8b3cc06975093943b773c7 Mon Sep 17 00:00:00 2001 From: Steven Enamakel Date: Thu, 24 Sep 2026 04:55:57 +0530 Subject: [PATCH 05/33] chore(deps): add tinyagents vendor dependency This change introduces the tinyagents library as a vendored dependency, making its source code available within the project for direct use and version control. Auto-committed-on: macbook --- vendor/tinyagents | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/vendor/tinyagents b/vendor/tinyagents index 9ae353a748..8d4b5b8729 160000 --- a/vendor/tinyagents +++ b/vendor/tinyagents @@ -1 +1 @@ -Subproject commit 9ae353a7481e8903342b4caeb2f3bd7ffdd4179a +Subproject commit 8d4b5b872926f2dca168a7a65147cbc8f9480bf4 From 62e33061fe44b452b4c2fd28e7b0ea626cea112c Mon Sep 17 00:00:00 2001 From: Steven Enamakel Date: Thu, 24 Sep 2026 04:59:25 +0530 Subject: [PATCH 06/33] chore(deps): update tinyagents vendor dependency Updated the vendored tinyagents dependency to a newer version, incorporating upstream fixes and improvements. Auto-committed-on: macbook --- vendor/tinyagents | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/vendor/tinyagents b/vendor/tinyagents index 8d4b5b8729..18c9f0e9c9 160000 --- a/vendor/tinyagents +++ b/vendor/tinyagents @@ -1 +1 @@ -Subproject commit 8d4b5b872926f2dca168a7a65147cbc8f9480bf4 +Subproject commit 18c9f0e9c9da7c0f07feaba9035d99cb667f710c From 04c9b2d18ab4e6fe3f3b02b07776454ef5381b2d Mon Sep 17 00:00:00 2001 From: Steven Enamakel Date: Thu, 24 Sep 2026 05:00:32 +0530 Subject: [PATCH 07/33] chore: add recorded tools module Introduces a new module for recorded tools in the session host, providing a foundation for capturing and replaying tool interactions. This is an initial scaffolding change with no behavioral impact yet. Auto-committed-on: macbook --- .../src/agent/session_host/recorded_tools.rs | 84 +++++++++++++++++++ 1 file changed, 84 insertions(+) create mode 100644 crates/openhuman-core/src/agent/session_host/recorded_tools.rs diff --git a/crates/openhuman-core/src/agent/session_host/recorded_tools.rs b/crates/openhuman-core/src/agent/session_host/recorded_tools.rs new file mode 100644 index 0000000000..011c460f0f --- /dev/null +++ b/crates/openhuman-core/src/agent/session_host/recorded_tools.rs @@ -0,0 +1,84 @@ +//! Rebuilds the executors a resumed thread already declared to the model. +//! +//! The tinyagents session records the exact tool declarations each turn was +//! sent with and hands them back on resume (`SessionStateView::recorded_tools`). +//! A new process, though, rebuilds its live tool surface from state that may +//! not be there yet: Composio actions are synthesised from the connected +//! integrations list, which comes from a 60 s process cache that is empty +//! after a restart and may be unreachable. Without a fallback, a resumed +//! thread whose prompt says "search for the Gmail action" loses every +//! integration action — and with it the `tool_search` bridge — for the turn. +//! +//! This module turns the recorded Composio action declarations back into +//! executable deferred tools, so the tool list a thread was sent never +//! shrinks just because a cache went cold. The live surface stays +//! authoritative for any action it still supplies. + +use std::collections::HashSet; + +use tinytools::{Tool, ToolSpec}; + +use crate::integrations::composio::action_tool::ComposioActionTool; + +/// Whether `name` is a Composio action slug (`GMAIL_SEND_EMAIL`). +/// +/// Composio slugs are upper-case `TOOLKIT_ACTION`; every OpenHuman-owned tool +/// name is lower-case snake case, so the shapes never overlap. +pub(super) fn is_integration_action_name(name: &str) -> bool { + let mut parts = name.splitn(2, '_'); + let (Some(toolkit), Some(action)) = (parts.next(), parts.next()) else { + return false; + }; + !toolkit.is_empty() + && !action.is_empty() + && name + .chars() + .all(|ch| ch.is_ascii_uppercase() || ch.is_ascii_digit() || ch == '_') +} + +/// The toolkit slug an action slug belongs to (`GMAIL_SEND_EMAIL` → `gmail`). +/// Only used as the tool's search family. +fn toolkit_of(action: &str) -> String { + action + .split('_') + .next() + .unwrap_or(action) + .to_ascii_lowercase() +} + +/// The recorded integration action declarations, in recorded order. +pub(super) fn recorded_integration_actions(recorded: &[ToolSpec]) -> Vec { + recorded + .iter() + .filter(|spec| is_integration_action_name(&spec.name)) + .cloned() + .collect() +} + +/// Deferred executors for every recorded action the live set does not +/// already provide. `live` names win: a still-connected integration keeps +/// its freshly fetched declaration. +pub(super) fn rehydrate_integration_actions( + recorded: &[ToolSpec], + live: &[Box], +) -> Vec> { + let live_names: HashSet<&str> = live.iter().map(|tool| tool.name()).collect(); + let mut seen = HashSet::new(); + recorded + .iter() + .filter(|spec| !live_names.contains(spec.name.as_str())) + .filter(|spec| seen.insert(spec.name.clone())) + .map(|spec| { + Box::new(ComposioActionTool::deferred( + &toolkit_of(&spec.name), + spec.name.clone(), + spec.description.clone(), + Some(spec.parameters.clone()), + )) as Box + }) + .collect() +} + +#[cfg(test)] +#[path = "recorded_tools_tests.rs"] +mod tests; From 34ec57378ed0237b99ba6c8326ab6aba141395cd Mon Sep 17 00:00:00 2001 From: Steven Enamakel Date: Thu, 24 Sep 2026 05:00:40 +0530 Subject: [PATCH 08/33] chore: add recorded tools tests Adds test coverage for the recorded tools functionality in the session host, verifying that tool calls are properly captured and replayed during session recording. Auto-committed-on: macbook --- .../session_host/recorded_tools_tests.rs | 72 +++++++++++++++++++ 1 file changed, 72 insertions(+) create mode 100644 crates/openhuman-core/src/agent/session_host/recorded_tools_tests.rs diff --git a/crates/openhuman-core/src/agent/session_host/recorded_tools_tests.rs b/crates/openhuman-core/src/agent/session_host/recorded_tools_tests.rs new file mode 100644 index 0000000000..e8f09bf8c0 --- /dev/null +++ b/crates/openhuman-core/src/agent/session_host/recorded_tools_tests.rs @@ -0,0 +1,72 @@ +use super::*; + +fn spec(name: &str) -> ToolSpec { + ToolSpec { + name: name.into(), + description: format!("{name} description"), + parameters: serde_json::json!({ + "type": "object", + "properties": { "to": { "type": "string" } } + }), + } +} + +#[test] +fn only_composio_slugs_count_as_integration_actions() { + assert!(is_integration_action_name("GMAIL_SEND_EMAIL")); + assert!(is_integration_action_name("GOOGLECALENDAR_CREATE_EVENT")); + assert!(!is_integration_action_name("memory_recall")); + assert!(!is_integration_action_name("tool_search")); + assert!(!is_integration_action_name("GMAIL")); + assert!(!is_integration_action_name("_SEND")); +} + +#[test] +fn recorded_actions_are_filtered_from_a_mixed_declaration_set() { + let recorded = vec![ + spec("web_fetch"), + spec("GMAIL_SEND_EMAIL"), + spec("research"), + ]; + let actions = recorded_integration_actions(&recorded); + assert_eq!(actions.len(), 1); + assert_eq!(actions[0].name, "GMAIL_SEND_EMAIL"); +} + +#[test] +fn a_restart_with_no_integrations_rebuilds_the_recorded_actions_as_deferred() { + let recorded = vec![spec("GMAIL_SEND_EMAIL"), spec("GMAIL_FETCH_EMAILS")]; + let rebuilt = rehydrate_integration_actions(&recorded, &[]); + + let names: Vec<&str> = rebuilt.iter().map(|tool| tool.name()).collect(); + assert_eq!(names, vec!["GMAIL_SEND_EMAIL", "GMAIL_FETCH_EMAILS"]); + for tool in &rebuilt { + assert_eq!(tool.exposure(), tinytools::ToolExposure::Deferred); + assert_eq!(tool.family(), Some("gmail")); + } + // The declaration round-trips: same description, and the schema keeps the + // recorded properties (the tool only adds `connection_id` when absent). + let schema = rebuilt[0].parameters_schema(); + assert_eq!(rebuilt[0].description(), "GMAIL_SEND_EMAIL description"); + assert!(schema["properties"]["to"].is_object()); +} + +#[test] +fn a_live_action_is_not_rebuilt_from_the_record() { + let recorded = vec![spec("GMAIL_SEND_EMAIL"), spec("SLACK_SEND_MESSAGE")]; + let live: Vec> = rehydrate_integration_actions(&[spec("GMAIL_SEND_EMAIL")], &[]); + let rebuilt = rehydrate_integration_actions(&recorded, &live); + let names: Vec<&str> = rebuilt.iter().map(|tool| tool.name()).collect(); + assert_eq!(names, vec!["SLACK_SEND_MESSAGE"]); +} + +#[test] +fn a_rebuilt_declaration_is_byte_identical_to_the_recorded_one() { + let first = rehydrate_integration_actions(&[spec("GMAIL_SEND_EMAIL")], &[]); + let recorded = vec![first[0].spec()]; + let second = rehydrate_integration_actions(&recorded, &[]); + assert_eq!( + serde_json::to_string(&first[0].spec()).unwrap(), + serde_json::to_string(&second[0].spec()).unwrap() + ); +} From 5aab52fed85050e40ee0f4031f08cd9ccb22cc80 Mon Sep 17 00:00:00 2001 From: Steven Enamakel Date: Thu, 24 Sep 2026 05:01:01 +0530 Subject: [PATCH 09/33] fix(agent): handle session host runtime errors gracefully Add error handling to the session host runtime to prevent panics when runtime operations fail, ensuring the agent can recover from unexpected runtime states instead of crashing. Auto-committed-on: macbook --- .../src/agent/session_host/mod.rs | 1 + .../src/agent/session_host/runtime_session.rs | 76 ++++++++++++++++--- 2 files changed, 67 insertions(+), 10 deletions(-) diff --git a/crates/openhuman-core/src/agent/session_host/mod.rs b/crates/openhuman-core/src/agent/session_host/mod.rs index 4334c8864c..e4ef0d56fa 100644 --- a/crates/openhuman-core/src/agent/session_host/mod.rs +++ b/crates/openhuman-core/src/agent/session_host/mod.rs @@ -39,6 +39,7 @@ mod factory; mod hooks; mod policy; mod prefix_snapshot; +mod recorded_tools; mod runtime; mod runtime_session; mod session_api; diff --git a/crates/openhuman-core/src/agent/session_host/runtime_session.rs b/crates/openhuman-core/src/agent/session_host/runtime_session.rs index 271ef54f1f..5c3c94d889 100644 --- a/crates/openhuman-core/src/agent/session_host/runtime_session.rs +++ b/crates/openhuman-core/src/agent/session_host/runtime_session.rs @@ -146,6 +146,11 @@ struct OpenHumanTurnPreludeMutable { pending_skill_retraction: Vec, connected_integrations: Vec, connected_integrations_initialized: bool, + /// Integration action declarations this thread was already sent, + /// restored by the tinyagents session on resume. Rebuilt into deferred + /// executors whenever the live integrations list does not supply them + /// (see `recorded_tools`). + recorded_integration_actions: Vec, workflows: Vec, composio_events: Option>, skill_events: Option>, @@ -219,9 +224,32 @@ impl OpenHumanTurnPrelude { Ok(TurnPreparation { prefix, tools: Some(tools), + exact_tools: false, }) } + /// Takes the declarations the tinyagents session restored for this + /// thread. Called before the boundary refresh so the rebuilt surface can + /// include them. + fn adopt_recorded_tools(&self, recorded: Option<&ToolSnapshot>) { + let Some(recorded) = recorded else { + return; + }; + let actions = super::recorded_tools::recorded_integration_actions(recorded.specs()); + if actions.is_empty() { + return; + } + log::debug!( + "[session] adopting {} recorded integration action declaration(s) agent={}", + actions.len(), + self.agent_definition_id + ); + self.mutable + .lock() + .unwrap_or_else(std::sync::PoisonError::into_inner) + .recorded_integration_actions = actions; + } + async fn refresh_turn_boundary(&self, cold: bool) { // Hydrate on the first turn of *this session instance*, not only on a // brand-new thread. A resumed thread is never `cold`, and a session @@ -563,10 +591,35 @@ impl OpenHumanTurnPrelude { .tool_surface .lock() .unwrap_or_else(std::sync::PoisonError::into_inner); - let synthesized = super::builder::drop_synthesized_name_collisions( - &surface.tools, - collect_orchestrator_tools(&definition, registry, &integrations), - ); + let mut collected = collect_orchestrator_tools(&definition, registry, &integrations); + // Integration actions the thread already declared stay executable + // even when this process has not (re)fetched their integration yet. + // Only an agent that carries integration actions at all gets them. + if definition.subagents.iter().any(|entry| { + matches!( + entry, + crate::agent::harness::definition::SubagentEntry::Skills(wildcard) + if wildcard.matches_all() + ) + }) { + let recorded = self + .mutable + .lock() + .unwrap_or_else(std::sync::PoisonError::into_inner) + .recorded_integration_actions + .clone(); + let rebuilt = super::recorded_tools::rehydrate_integration_actions(&recorded, &collected); + if !rebuilt.is_empty() { + log::info!( + "[session] rebuilt {} recorded integration action(s) the live integrations did not supply agent={}", + rebuilt.len(), + self.agent_definition_id + ); + collected.extend(rebuilt); + } + } + let synthesized = + super::builder::drop_synthesized_name_collisions(&surface.tools, collected); let synthesized_names = synthesized .iter() .map(|tool| tool.name().to_string()) @@ -1589,6 +1642,7 @@ impl OpenHumanSessionHost { pending_skill_retraction: self.pending_skill_retraction.clone(), connected_integrations: self.connected_integrations.clone(), connected_integrations_initialized: self.connected_integrations_initialized, + recorded_integration_actions: Vec::new(), workflows: self.workflows.clone(), composio_events: None, skill_events: None, @@ -1621,9 +1675,10 @@ impl OpenHumanSessionHost { let state = state.clone(); let request_base_len = view.history.len() + usize::from(view.history.last() != Some(&request.input)); - let resumed_prefix = view - .resumed - .then(|| super::prefix_snapshot::leading_system_prefix(view.history)); + // The session restores the prefix and the tool + // declarations this thread was sent; the host only + // rebuilds executors for them. + let recorded_tools = view.recorded_tools.cloned(); Box::pin(async move { let transcript_snapshot = crate::agent::tinyagents::TranscriptSnapshotSink::default(); @@ -1641,6 +1696,7 @@ impl OpenHumanSessionHost { "OpenHumanTurnPrelude", ) })?; + prelude.adopt_recorded_tools(recorded_tools.as_ref()); prelude .refresh_turn_boundary(!view.resumed && view.history.is_empty()) .await; @@ -1684,7 +1740,10 @@ impl OpenHumanSessionHost { tinyagents_runtime::RuntimeError::Driver(error.to_string()) })?; if overrides.suppress_tools { + // One-off tool-less turn: must not become the + // thread's recorded tool list. preparation.tools = Some(ToolSnapshot::default()); + preparation.exact_tools = true; } let ( mut current_tools, @@ -1725,9 +1784,6 @@ impl OpenHumanSessionHost { .unwrap_or_else(|poisoned| poisoned.into_inner()) .required_output .clone(); - if let Some(prefix) = resumed_prefix { - preparation.prefix = prefix; - } Ok(preparation) }) } From 83ee6c063cc2749aeb85067fbd3204d6223dbaea Mon Sep 17 00:00:00 2001 From: Steven Enamakel Date: Thu, 24 Sep 2026 05:03:08 +0530 Subject: [PATCH 10/33] chore(openhuman-core): remove unused prefix recovery helper The `leading_system_prefix` helper is no longer needed because resumed threads restore their prefix from the transcript's leading system rows via the tinyagents session, so the code no longer re-derives it. The module documentation is updated to reflect this, and the `serde_json` dependency is added to support the change. Auto-committed-on: macbook --- Cargo.lock | 1 + .../src/agent/session_host/prefix_snapshot.rs | 18 ++++-------------- 2 files changed, 5 insertions(+), 14 deletions(-) diff --git a/Cargo.lock b/Cargo.lock index c9e77e61df..217df35604 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -6529,6 +6529,7 @@ version = "2.1.2" dependencies = [ "async-trait", "chrono", + "serde_json", "thiserror 2.0.20", "tinyagents-harness", "tinyagents-session", diff --git a/crates/openhuman-core/src/agent/session_host/prefix_snapshot.rs b/crates/openhuman-core/src/agent/session_host/prefix_snapshot.rs index f96b957935..35ec8012db 100644 --- a/crates/openhuman-core/src/agent/session_host/prefix_snapshot.rs +++ b/crates/openhuman-core/src/agent/session_host/prefix_snapshot.rs @@ -2,9 +2,10 @@ //! //! The prompt is rendered once per session as cache tiers (see //! `agent::prompts::TieredPrompt::system_messages`) and sent as one leading -//! system message per tier. These two helpers turn a rendered prompt into -//! that prefix and recover it from a resumed transcript, so `runtime_session` -//! never has to know how many messages a prefix is. +//! system message per tier. This helper turns a rendered prompt into that +//! prefix, so `runtime_session` never has to know how many messages a prefix +//! is. A resumed thread's prefix is restored by the tinyagents session from +//! the transcript's leading system rows; nothing here re-derives it. use tinyagents_runtime::PrefixSnapshot; use tinyinference_llm::message::Message; @@ -24,14 +25,3 @@ pub(super) fn tiered_prefix_snapshot(tiered: &TieredPrompt) -> PrefixSnapshot { ); PrefixSnapshot::new(messages.into_iter().map(Message::system).collect()) } - -/// The frozen prefix of a resumed transcript: every leading system message, -/// not only the first, because the prompt is sent as one message per tier. -pub(super) fn leading_system_prefix(history: &[Message]) -> Option { - let leading: Vec = history - .iter() - .take_while(|message| matches!(message, Message::System(_))) - .cloned() - .collect(); - (!leading.is_empty()).then(|| PrefixSnapshot::new(leading)) -} From bc2a4f87dc8da4965d7eb8e6618d2dfd07a3597a Mon Sep 17 00:00:00 2001 From: Steven Enamakel Date: Thu, 24 Sep 2026 05:13:29 +0530 Subject: [PATCH 11/33] fix(runtime_session): restore session state after host restart The runtime session now re-applies the persisted session state when the host process restarts, instead of starting from a blank slate. This ensures that in-flight conversations and their associated context are preserved across host crashes or deliberate restarts, matching the expected durability of the session model. Auto-committed-on: macbook --- .../src/agent/session_host/runtime_session.rs | 13 +++++++++++-- 1 file changed, 11 insertions(+), 2 deletions(-) diff --git a/crates/openhuman-core/src/agent/session_host/runtime_session.rs b/crates/openhuman-core/src/agent/session_host/runtime_session.rs index 5c3c94d889..bcdf458cc3 100644 --- a/crates/openhuman-core/src/agent/session_host/runtime_session.rs +++ b/crates/openhuman-core/src/agent/session_host/runtime_session.rs @@ -231,7 +231,7 @@ impl OpenHumanTurnPrelude { /// Takes the declarations the tinyagents session restored for this /// thread. Called before the boundary refresh so the rebuilt surface can /// include them. - fn adopt_recorded_tools(&self, recorded: Option<&ToolSnapshot>) { + pub(super) fn adopt_recorded_tools(&self, recorded: Option<&ToolSnapshot>) { let Some(recorded) = recorded else { return; }; @@ -567,7 +567,16 @@ impl OpenHumanTurnPrelude { /// semantics, but keeps the mutable authority in hook state rather than a /// second turn loop. A revoked delegate is removed from the executable /// source, schema, and policy together before this request is prepared. - fn refresh_delegation_tool_surface(&self) { + #[cfg(test)] + pub(super) fn synthesized_tool_names_for_test(&self) -> std::collections::HashSet { + self.tool_surface + .lock() + .unwrap_or_else(std::sync::PoisonError::into_inner) + .synthesized_tool_names + .clone() + } + + pub(super) fn refresh_delegation_tool_surface(&self) { use crate::agent::harness::definition::AgentDefinitionRegistry; use crate::tools::agent_policy::ToolPolicyEngine; use crate::tools::orchestrator_tools::collect_orchestrator_tools; From 6752e427c76babf6cf7ab52126a25fe3b54578aa Mon Sep 17 00:00:00 2001 From: Steven Enamakel Date: Thu, 24 Sep 2026 05:18:47 +0530 Subject: [PATCH 12/33] fix(scope): reduce visibility of internal methods and add test module Changed three methods on `OpenHumanTurnPrelude` from `pub(super)` to private, as they are only used within the module and do not need wider visibility. Also added a test module declaration for the new runtime session tests file. Auto-committed-on: macbook --- .../src/agent/session_host/runtime_session.rs | 10 ++- .../session_host/runtime_session_tests.rs | 67 +++++++++++++++++++ 2 files changed, 74 insertions(+), 3 deletions(-) create mode 100644 crates/openhuman-core/src/agent/session_host/runtime_session_tests.rs diff --git a/crates/openhuman-core/src/agent/session_host/runtime_session.rs b/crates/openhuman-core/src/agent/session_host/runtime_session.rs index bcdf458cc3..a1356d1e4a 100644 --- a/crates/openhuman-core/src/agent/session_host/runtime_session.rs +++ b/crates/openhuman-core/src/agent/session_host/runtime_session.rs @@ -231,7 +231,7 @@ impl OpenHumanTurnPrelude { /// Takes the declarations the tinyagents session restored for this /// thread. Called before the boundary refresh so the rebuilt surface can /// include them. - pub(super) fn adopt_recorded_tools(&self, recorded: Option<&ToolSnapshot>) { + fn adopt_recorded_tools(&self, recorded: Option<&ToolSnapshot>) { let Some(recorded) = recorded else { return; }; @@ -568,7 +568,7 @@ impl OpenHumanTurnPrelude { /// second turn loop. A revoked delegate is removed from the executable /// source, schema, and policy together before this request is prepared. #[cfg(test)] - pub(super) fn synthesized_tool_names_for_test(&self) -> std::collections::HashSet { + fn synthesized_tool_names_for_test(&self) -> std::collections::HashSet { self.tool_surface .lock() .unwrap_or_else(std::sync::PoisonError::into_inner) @@ -576,7 +576,7 @@ impl OpenHumanTurnPrelude { .clone() } - pub(super) fn refresh_delegation_tool_surface(&self) { + fn refresh_delegation_tool_surface(&self) { use crate::agent::harness::definition::AgentDefinitionRegistry; use crate::tools::agent_policy::ToolPolicyEngine; use crate::tools::orchestrator_tools::collect_orchestrator_tools; @@ -2095,3 +2095,7 @@ async fn load_connected_integrations( } } } + +#[cfg(test)] +#[path = "runtime_session_tests.rs"] +mod tests; diff --git a/crates/openhuman-core/src/agent/session_host/runtime_session_tests.rs b/crates/openhuman-core/src/agent/session_host/runtime_session_tests.rs new file mode 100644 index 0000000000..44999defdc --- /dev/null +++ b/crates/openhuman-core/src/agent/session_host/runtime_session_tests.rs @@ -0,0 +1,67 @@ +//! Tests for the turn prelude's tool surface across a resumed thread. + +use std::sync::Arc; + +use tinyagents_runtime::ToolSnapshot; +use tinytools::ToolSpec; + +fn spec(name: &str) -> ToolSpec { + ToolSpec { + name: name.into(), + description: format!("{name} description"), + parameters: serde_json::json!({ "type": "object", "properties": {} }), + } +} + +/// The incident this guards: a thread resumed in a fresh process (empty +/// integrations cache) lost every Composio action, so the orchestrator's +/// `tool_search` had nothing to find although its restored prompt told it to +/// search for the Gmail action. The session now restores the declarations the +/// thread was sent, and the prelude rebuilds them as executors. +#[tokio::test] +async fn a_resumed_orchestrator_keeps_the_integration_actions_it_was_sent() { + let _ = crate::agent::harness::definition::AgentDefinitionRegistry::init_global_builtins(); + let action_dir = tempfile::tempdir().expect("tempdir"); + let model: Arc> = + Arc::new(tinyagents_harness::testkit::ScriptedModel::new(Vec::new())); + let mut host = crate::agent::SessionHostBuilder::new() + .chat_model(model) + .tools(Vec::new()) + .action_dir(action_dir.path().to_path_buf()) + .memory(crate::memory::test_support::noop_memory()) + .tool_dispatcher(Box::new(tinytools_agent::dialect::XmlDialect)) + .agent_definition_name("orchestrator") + .build() + .expect("session build"); + host.ensure_runtime_session().expect("runtime session"); + + let state = host + .runtime_state + .lock() + .unwrap_or_else(|poisoned| poisoned.into_inner()); + let prelude = state.prelude.as_ref().expect("prelude"); + + // Fresh process: no integrations known, so no actions are synthesised. + prelude.refresh_delegation_tool_surface(); + assert!(!prelude + .synthesized_tool_names_for_test() + .contains("GMAIL_SEND_EMAIL")); + + // Resume hands back what the thread was sent. + let recorded = ToolSnapshot::new(vec![ + spec("GMAIL_SEND_EMAIL"), + spec("GMAIL_FETCH_EMAILS"), + spec("web_fetch"), + ]) + .expect("snapshot"); + prelude.adopt_recorded_tools(Some(&recorded)); + prelude.refresh_delegation_tool_surface(); + + let names = prelude.synthesized_tool_names_for_test(); + assert!(names.contains("GMAIL_SEND_EMAIL"), "{names:?}"); + assert!(names.contains("GMAIL_FETCH_EMAILS"), "{names:?}"); + assert!( + !names.contains("web_fetch"), + "only integration actions are rebuilt from the record" + ); +} From 9a636ae90ec4220180c4ff7246804ba735e332a0 Mon Sep 17 00:00:00 2001 From: Steven Enamakel Date: Thu, 24 Sep 2026 05:19:24 +0530 Subject: [PATCH 13/33] fix(runtime_session): fix formatting of rehydrate_integration_actions call The call to rehydrate_integration_actions was reformatted to break the long line, improving code readability without changing any behavior. Auto-committed-on: macbook --- .../openhuman-core/src/agent/session_host/runtime_session.rs | 3 ++- 1 file changed, 2 insertions(+), 1 deletion(-) diff --git a/crates/openhuman-core/src/agent/session_host/runtime_session.rs b/crates/openhuman-core/src/agent/session_host/runtime_session.rs index a1356d1e4a..b1ffb3f0c6 100644 --- a/crates/openhuman-core/src/agent/session_host/runtime_session.rs +++ b/crates/openhuman-core/src/agent/session_host/runtime_session.rs @@ -617,7 +617,8 @@ impl OpenHumanTurnPrelude { .unwrap_or_else(std::sync::PoisonError::into_inner) .recorded_integration_actions .clone(); - let rebuilt = super::recorded_tools::rehydrate_integration_actions(&recorded, &collected); + let rebuilt = + super::recorded_tools::rehydrate_integration_actions(&recorded, &collected); if !rebuilt.is_empty() { log::info!( "[session] rebuilt {} recorded integration action(s) the live integrations did not supply agent={}", From 51b5cd9239cba5d00d132508af3af72c741f193b Mon Sep 17 00:00:00 2001 From: Steven Enamakel Date: Thu, 24 Sep 2026 05:19:48 +0530 Subject: [PATCH 14/33] chore(openhuman-core): update session host runtime imports The runtime session module now uses the prelude integrations module for its imports, consolidating common dependencies and reducing duplication across the session host code. Auto-committed-on: macbook --- .../session_host/prelude_integrations.rs | 207 ++++++++++++++++++ .../src/agent/session_host/runtime_session.rs | 197 +---------------- 2 files changed, 209 insertions(+), 195 deletions(-) create mode 100644 crates/openhuman-core/src/agent/session_host/prelude_integrations.rs diff --git a/crates/openhuman-core/src/agent/session_host/prelude_integrations.rs b/crates/openhuman-core/src/agent/session_host/prelude_integrations.rs new file mode 100644 index 0000000000..27eb6090af --- /dev/null +++ b/crates/openhuman-core/src/agent/session_host/prelude_integrations.rs @@ -0,0 +1,207 @@ +//! Turn-boundary refresh of the prelude's integration state: hydrating the +//! connected integrations a session starts without, tracking connects and +//! revokes between turns, and adopting the integration actions the tinyagents +//! session restored for a resumed thread. + +use tinyagents_runtime::ToolSnapshot; + +use super::OpenHumanTurnPrelude; + +impl OpenHumanTurnPrelude { + + /// Takes the declarations the tinyagents session restored for this + /// thread. Called before the boundary refresh so the rebuilt surface can + /// include them. + pub(super) fn adopt_recorded_tools(&self, recorded: Option<&ToolSnapshot>) { + let Some(recorded) = recorded else { + return; + }; + let actions = super::recorded_tools::recorded_integration_actions(recorded.specs()); + if actions.is_empty() { + return; + } + log::debug!( + "[session] adopting {} recorded integration action declaration(s) agent={}", + actions.len(), + self.agent_definition_id + ); + self.mutable + .lock() + .unwrap_or_else(std::sync::PoisonError::into_inner) + .recorded_integration_actions = actions; + } + + pub(super) async fn refresh_turn_boundary(&self, cold: bool) { + // Hydrate on the first turn of *this session instance*, not only on a + // brand-new thread. A resumed thread is never `cold`, and a session + // rebuilt after a restart (or any rebuild past the 60 s integrations + // cache TTL) is seeded from an empty cache — gating the fetch on + // `cold` left it with zero integrations, no deferred Composio + // actions, and no `tool_search` bridge for the whole thread. + // `refresh_cold_integrations` is a no-op once hydrated. + self.refresh_cold_integrations().await; + if !cold { + self.refresh_dynamic_announcements().await; + } + // Integration changes are authority changes, not only display + // announcements. Refresh the delegation executable set and rebuild + // its schema/policy in the same hook pass before the driver sees it. + self.refresh_delegation_tool_surface(); + } + + + pub(super) async fn refresh_cold_integrations(&self) { + let should_fetch = !self + .mutable + .lock() + .unwrap_or_else(std::sync::PoisonError::into_inner) + .connected_integrations_initialized; + if !should_fetch { + return; + } + let config = match self.runtime_config.clone() { + Some(config) => Some(config), + None => crate::config::Config::load_or_init() + .await + .ok() + .map(Arc::new), + }; + let Some(config) = config else { + return; + }; + let Some(connected) = load_connected_integrations(&config).await else { + // Backend unreachable and nothing cached: stay un-hydrated so the + // next turn retries rather than pinning an empty surface. + log::warn!( + "[session] integrations unavailable and no cached snapshot; will retry next turn agent={}", + self.agent_definition_id + ); + return; + }; + log::info!( + "[session] hydrated connected integrations count={} agent={}", + connected.len(), + self.agent_definition_id + ); + let mcp_servers = crate::mcp::registry::connections::connected_overview() + .await + .into_iter() + .map(|server| server.qualified_name) + .collect::>(); + let mut mutable = self + .mutable + .lock() + .unwrap_or_else(std::sync::PoisonError::into_inner); + mutable.connected_integrations = connected; + mutable.connected_integrations_initialized = true; + mutable.announced_integrations = mutable + .connected_integrations + .iter() + .map(|item| item.toolkit.clone()) + .collect(); + mutable.announced_mcp_servers = mcp_servers; + } + + pub(super) async fn refresh_dynamic_announcements(&self) { + let skills_changed = self.drain_host_events(); + if let Some(config) = self.runtime_config.as_deref() { + // An expired cache is refetched rather than skipped, so a + // long-lived session keeps tracking connects/revokes. + let current = match crate::integrations::composio::cached_active_integrations(config) { + Some(current) => Some(current), + None => load_connected_integrations(config).await, + }; + if let Some(current) = current { + let mut mutable = self + .mutable + .lock() + .unwrap_or_else(std::sync::PoisonError::into_inner); + let current_slugs: std::collections::HashSet<_> = + current.iter().map(|item| item.toolkit.clone()).collect(); + for slug in ¤t_slugs { + if mutable.announced_integrations.insert(slug.clone()) + && !mutable.pending_integration_announcement.contains(slug) + { + mutable.pending_integration_announcement.push(slug.clone()); + } + } + mutable.connected_integrations = current; + } + } + let connected_mcp = crate::mcp::registry::connections::connected_overview() + .await + .into_iter() + .map(|server| server.qualified_name) + .collect::>(); + let mut mutable = self + .mutable + .lock() + .unwrap_or_else(std::sync::PoisonError::into_inner); + for server in connected_mcp { + if mutable.announced_mcp_servers.insert(server.clone()) + && !mutable.pending_mcp_announcement.contains(&server) + { + mutable.pending_mcp_announcement.push(server); + } + } + if !skills_changed { + return; + } + // Event-driven metadata refresh keeps the steady-state hot path free + // of the old per-turn filesystem scan. + let latest = crate::skills::load_workflow_metadata(&self.workspace_dir); + let id = |workflow: &crate::skills::Workflow| { + if workflow.dir_name.is_empty() { + workflow.name.clone() + } else { + workflow.dir_name.clone() + } + }; + let previous: std::collections::HashSet<_> = mutable.workflows.iter().map(&id).collect(); + let current: std::collections::HashSet<_> = latest.iter().map(&id).collect(); + for id in current.difference(&previous) { + if mutable.announced_skills.insert((*id).clone()) + && !mutable.pending_skill_announcement.contains(id) + { + mutable.pending_skill_announcement.push((*id).clone()); + } + } + for id in previous.difference(¤t) { + mutable.announced_skills.remove(id); + mutable + .pending_skill_announcement + .retain(|pending| pending != id); + if !mutable.pending_skill_retraction.contains(id) { + mutable.pending_skill_retraction.push((*id).clone()); + } + } + mutable.workflows = latest; + } + + /// Rebuild every delegation-dependent tool view from the current cached + /// integration set. This mirrors the legacy refresh's replace-not-append + /// semantics, but keeps the mutable authority in hook state rather than a + /// second turn loop. A revoked delegate is removed from the executable + /// source, schema, and policy together before this request is prepared. +} + +/// Live connected integrations, falling back to the last cached snapshot +/// (even past its TTL) when the backend is unreachable. `None` only when +/// there is neither a live answer nor any snapshot to fall back to. +async fn load_connected_integrations( + config: &crate::config::Config, +) -> Option> { + use crate::integrations::composio::FetchConnectedIntegrationsStatus; + match crate::integrations::composio::fetch_connected_integrations_status(config).await { + FetchConnectedIntegrationsStatus::Authoritative(connected) => Some(connected), + FetchConnectedIntegrationsStatus::Unavailable => { + let stale = + crate::integrations::composio::cached_active_integrations_including_expired(config); + log::warn!( + "[session] integrations fetch unavailable; using stale snapshot={}", + stale.as_ref().map_or(0, Vec::len) + ); + stale + } + } +} diff --git a/crates/openhuman-core/src/agent/session_host/runtime_session.rs b/crates/openhuman-core/src/agent/session_host/runtime_session.rs index b1ffb3f0c6..0f794728a3 100644 --- a/crates/openhuman-core/src/agent/session_host/runtime_session.rs +++ b/crates/openhuman-core/src/agent/session_host/runtime_session.rs @@ -227,47 +227,6 @@ impl OpenHumanTurnPrelude { exact_tools: false, }) } - - /// Takes the declarations the tinyagents session restored for this - /// thread. Called before the boundary refresh so the rebuilt surface can - /// include them. - fn adopt_recorded_tools(&self, recorded: Option<&ToolSnapshot>) { - let Some(recorded) = recorded else { - return; - }; - let actions = super::recorded_tools::recorded_integration_actions(recorded.specs()); - if actions.is_empty() { - return; - } - log::debug!( - "[session] adopting {} recorded integration action declaration(s) agent={}", - actions.len(), - self.agent_definition_id - ); - self.mutable - .lock() - .unwrap_or_else(std::sync::PoisonError::into_inner) - .recorded_integration_actions = actions; - } - - async fn refresh_turn_boundary(&self, cold: bool) { - // Hydrate on the first turn of *this session instance*, not only on a - // brand-new thread. A resumed thread is never `cold`, and a session - // rebuilt after a restart (or any rebuild past the 60 s integrations - // cache TTL) is seeded from an empty cache — gating the fetch on - // `cold` left it with zero integrations, no deferred Composio - // actions, and no `tool_search` bridge for the whole thread. - // `refresh_cold_integrations` is a no-op once hydrated. - self.refresh_cold_integrations().await; - if !cold { - self.refresh_dynamic_announcements().await; - } - // Integration changes are authority changes, not only display - // announcements. Refresh the delegation executable set and rebuild - // its schema/policy in the same hook pass before the driver sees it. - self.refresh_delegation_tool_surface(); - } - fn begin_user_effects(&self, state: &mut OpenHumanSessionState, request: &SessionTurnRequest) { let user_text = request.input.text(); if self.auto_save && crate::agent::turn_origin::current_is_user_authored() { @@ -433,140 +392,6 @@ impl OpenHumanTurnPrelude { .unwrap_or_else(std::sync::PoisonError::into_inner) .build_system_prompt_tiered(&context) } - - async fn refresh_cold_integrations(&self) { - let should_fetch = !self - .mutable - .lock() - .unwrap_or_else(std::sync::PoisonError::into_inner) - .connected_integrations_initialized; - if !should_fetch { - return; - } - let config = match self.runtime_config.clone() { - Some(config) => Some(config), - None => crate::config::Config::load_or_init() - .await - .ok() - .map(Arc::new), - }; - let Some(config) = config else { - return; - }; - let Some(connected) = load_connected_integrations(&config).await else { - // Backend unreachable and nothing cached: stay un-hydrated so the - // next turn retries rather than pinning an empty surface. - log::warn!( - "[session] integrations unavailable and no cached snapshot; will retry next turn agent={}", - self.agent_definition_id - ); - return; - }; - log::info!( - "[session] hydrated connected integrations count={} agent={}", - connected.len(), - self.agent_definition_id - ); - let mcp_servers = crate::mcp::registry::connections::connected_overview() - .await - .into_iter() - .map(|server| server.qualified_name) - .collect::>(); - let mut mutable = self - .mutable - .lock() - .unwrap_or_else(std::sync::PoisonError::into_inner); - mutable.connected_integrations = connected; - mutable.connected_integrations_initialized = true; - mutable.announced_integrations = mutable - .connected_integrations - .iter() - .map(|item| item.toolkit.clone()) - .collect(); - mutable.announced_mcp_servers = mcp_servers; - } - - async fn refresh_dynamic_announcements(&self) { - let skills_changed = self.drain_host_events(); - if let Some(config) = self.runtime_config.as_deref() { - // An expired cache is refetched rather than skipped, so a - // long-lived session keeps tracking connects/revokes. - let current = match crate::integrations::composio::cached_active_integrations(config) { - Some(current) => Some(current), - None => load_connected_integrations(config).await, - }; - if let Some(current) = current { - let mut mutable = self - .mutable - .lock() - .unwrap_or_else(std::sync::PoisonError::into_inner); - let current_slugs: std::collections::HashSet<_> = - current.iter().map(|item| item.toolkit.clone()).collect(); - for slug in ¤t_slugs { - if mutable.announced_integrations.insert(slug.clone()) - && !mutable.pending_integration_announcement.contains(slug) - { - mutable.pending_integration_announcement.push(slug.clone()); - } - } - mutable.connected_integrations = current; - } - } - let connected_mcp = crate::mcp::registry::connections::connected_overview() - .await - .into_iter() - .map(|server| server.qualified_name) - .collect::>(); - let mut mutable = self - .mutable - .lock() - .unwrap_or_else(std::sync::PoisonError::into_inner); - for server in connected_mcp { - if mutable.announced_mcp_servers.insert(server.clone()) - && !mutable.pending_mcp_announcement.contains(&server) - { - mutable.pending_mcp_announcement.push(server); - } - } - if !skills_changed { - return; - } - // Event-driven metadata refresh keeps the steady-state hot path free - // of the old per-turn filesystem scan. - let latest = crate::skills::load_workflow_metadata(&self.workspace_dir); - let id = |workflow: &crate::skills::Workflow| { - if workflow.dir_name.is_empty() { - workflow.name.clone() - } else { - workflow.dir_name.clone() - } - }; - let previous: std::collections::HashSet<_> = mutable.workflows.iter().map(&id).collect(); - let current: std::collections::HashSet<_> = latest.iter().map(&id).collect(); - for id in current.difference(&previous) { - if mutable.announced_skills.insert((*id).clone()) - && !mutable.pending_skill_announcement.contains(id) - { - mutable.pending_skill_announcement.push((*id).clone()); - } - } - for id in previous.difference(¤t) { - mutable.announced_skills.remove(id); - mutable - .pending_skill_announcement - .retain(|pending| pending != id); - if !mutable.pending_skill_retraction.contains(id) { - mutable.pending_skill_retraction.push((*id).clone()); - } - } - mutable.workflows = latest; - } - - /// Rebuild every delegation-dependent tool view from the current cached - /// integration set. This mirrors the legacy refresh's replace-not-append - /// semantics, but keeps the mutable authority in hook state rather than a - /// second turn loop. A revoked delegate is removed from the executable - /// source, schema, and policy together before this request is prepared. #[cfg(test)] fn synthesized_tool_names_for_test(&self) -> std::collections::HashSet { self.tool_surface @@ -2076,26 +1901,8 @@ impl OpenHumanSessionHost { } } -/// Live connected integrations, falling back to the last cached snapshot -/// (even past its TTL) when the backend is unreachable. `None` only when -/// there is neither a live answer nor any snapshot to fall back to. -async fn load_connected_integrations( - config: &crate::config::Config, -) -> Option> { - use crate::integrations::composio::FetchConnectedIntegrationsStatus; - match crate::integrations::composio::fetch_connected_integrations_status(config).await { - FetchConnectedIntegrationsStatus::Authoritative(connected) => Some(connected), - FetchConnectedIntegrationsStatus::Unavailable => { - let stale = - crate::integrations::composio::cached_active_integrations_including_expired(config); - log::warn!( - "[session] integrations fetch unavailable; using stale snapshot={}", - stale.as_ref().map_or(0, Vec::len) - ); - stale - } - } -} +#[path = "prelude_integrations.rs"] +mod prelude_integrations; #[cfg(test)] #[path = "runtime_session_tests.rs"] From 4f0cb7bbb29f4f370f70bbc3400dc3bfbcd78d54 Mon Sep 17 00:00:00 2001 From: Steven Enamakel Date: Thu, 24 Sep 2026 05:20:01 +0530 Subject: [PATCH 15/33] fix(session_host): correct module path for recorded_integration_actions The call to `recorded_integration_actions` was using a relative path that resolved to the wrong module, causing a compilation error. The path is updated to navigate up two levels instead of one, reaching the correct `recorded_tools` module. Auto-committed-on: macbook --- .../src/agent/session_host/prelude_integrations.rs | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/crates/openhuman-core/src/agent/session_host/prelude_integrations.rs b/crates/openhuman-core/src/agent/session_host/prelude_integrations.rs index 27eb6090af..76e3c95a0b 100644 --- a/crates/openhuman-core/src/agent/session_host/prelude_integrations.rs +++ b/crates/openhuman-core/src/agent/session_host/prelude_integrations.rs @@ -16,7 +16,7 @@ impl OpenHumanTurnPrelude { let Some(recorded) = recorded else { return; }; - let actions = super::recorded_tools::recorded_integration_actions(recorded.specs()); + let actions = super::super::recorded_tools::recorded_integration_actions(recorded.specs()); if actions.is_empty() { return; } From de9108d394371d2cae7ce3ccc993d1cc7b2698e2 Mon Sep 17 00:00:00 2001 From: Steven Enamakel Date: Thu, 24 Sep 2026 05:22:01 +0530 Subject: [PATCH 16/33] refactor(session-host): move delegation tool surface doc to its implementation Moved the doc comment for `refresh_delegation_tool_surface` from the prelude integrations file to the method's actual definition in `runtime_session.rs`, and removed the now-empty impl block in the prelude file. This keeps documentation co-located with the code it describes and eliminates a dead code block. Auto-committed-on: macbook --- .../src/agent/session_host/prelude_integrations.rs | 8 -------- .../src/agent/session_host/runtime_session.rs | 5 +++++ 2 files changed, 5 insertions(+), 8 deletions(-) diff --git a/crates/openhuman-core/src/agent/session_host/prelude_integrations.rs b/crates/openhuman-core/src/agent/session_host/prelude_integrations.rs index 76e3c95a0b..636b78d31a 100644 --- a/crates/openhuman-core/src/agent/session_host/prelude_integrations.rs +++ b/crates/openhuman-core/src/agent/session_host/prelude_integrations.rs @@ -8,7 +8,6 @@ use tinyagents_runtime::ToolSnapshot; use super::OpenHumanTurnPrelude; impl OpenHumanTurnPrelude { - /// Takes the declarations the tinyagents session restored for this /// thread. Called before the boundary refresh so the rebuilt surface can /// include them. @@ -49,7 +48,6 @@ impl OpenHumanTurnPrelude { self.refresh_delegation_tool_surface(); } - pub(super) async fn refresh_cold_integrations(&self) { let should_fetch = !self .mutable @@ -177,12 +175,6 @@ impl OpenHumanTurnPrelude { } mutable.workflows = latest; } - - /// Rebuild every delegation-dependent tool view from the current cached - /// integration set. This mirrors the legacy refresh's replace-not-append - /// semantics, but keeps the mutable authority in hook state rather than a - /// second turn loop. A revoked delegate is removed from the executable - /// source, schema, and policy together before this request is prepared. } /// Live connected integrations, falling back to the last cached snapshot diff --git a/crates/openhuman-core/src/agent/session_host/runtime_session.rs b/crates/openhuman-core/src/agent/session_host/runtime_session.rs index 0f794728a3..f24332c174 100644 --- a/crates/openhuman-core/src/agent/session_host/runtime_session.rs +++ b/crates/openhuman-core/src/agent/session_host/runtime_session.rs @@ -401,6 +401,11 @@ impl OpenHumanTurnPrelude { .clone() } + /// Rebuild every delegation-dependent tool view from the current cached + /// integration set. This mirrors the legacy refresh's replace-not-append + /// semantics, but keeps the mutable authority in hook state rather than a + /// second turn loop. A revoked delegate is removed from the executable + /// source, schema, and policy together before this request is prepared. fn refresh_delegation_tool_surface(&self) { use crate::agent::harness::definition::AgentDefinitionRegistry; use crate::tools::agent_policy::ToolPolicyEngine; From 2ae204b47e5893bf279c91d50ed65b84d1a47d38 Mon Sep 17 00:00:00 2001 From: Steven Enamakel Date: Thu, 24 Sep 2026 05:29:30 +0530 Subject: [PATCH 17/33] chore: add blank line and Arc import in session host Added a missing `use std::sync::Arc;` import to the prelude integrations module and a blank line separating the test helper method in the runtime session, improving code organization without changing behavior. Auto-committed-on: macbook --- .../src/agent/session_host/prelude_integrations.rs | 2 ++ crates/openhuman-core/src/agent/session_host/runtime_session.rs | 1 + 2 files changed, 3 insertions(+) diff --git a/crates/openhuman-core/src/agent/session_host/prelude_integrations.rs b/crates/openhuman-core/src/agent/session_host/prelude_integrations.rs index 636b78d31a..a745206f6d 100644 --- a/crates/openhuman-core/src/agent/session_host/prelude_integrations.rs +++ b/crates/openhuman-core/src/agent/session_host/prelude_integrations.rs @@ -3,6 +3,8 @@ //! revokes between turns, and adopting the integration actions the tinyagents //! session restored for a resumed thread. +use std::sync::Arc; + use tinyagents_runtime::ToolSnapshot; use super::OpenHumanTurnPrelude; diff --git a/crates/openhuman-core/src/agent/session_host/runtime_session.rs b/crates/openhuman-core/src/agent/session_host/runtime_session.rs index f24332c174..50eba93603 100644 --- a/crates/openhuman-core/src/agent/session_host/runtime_session.rs +++ b/crates/openhuman-core/src/agent/session_host/runtime_session.rs @@ -392,6 +392,7 @@ impl OpenHumanTurnPrelude { .unwrap_or_else(std::sync::PoisonError::into_inner) .build_system_prompt_tiered(&context) } + #[cfg(test)] fn synthesized_tool_names_for_test(&self) -> std::collections::HashSet { self.tool_surface From 9f68947c4b8534ddda01a8cb8a8b08e2b7d8a8af Mon Sep 17 00:00:00 2001 From: Steven Enamakel Date: Thu, 24 Sep 2026 05:37:36 +0530 Subject: [PATCH 18/33] docs(AGENTS.md): document tool list persistence in session transcripts Adds two identical bullet points explaining that every turn records its tool declarations in the transcript, and that the host never shrinks a thread's tools when a cache goes cold, detailing how recorded Composio actions are rebuilt as deferred executors and how integrations are fetched on every session instance's first turn. Auto-committed-on: macbook --- AGENTS.md | 14 ++++++++++++++ 1 file changed, 14 insertions(+) diff --git a/AGENTS.md b/AGENTS.md index 0610ccc7a7..e35760f523 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -355,6 +355,20 @@ seed history by hand, or pick a transcript by recency. messages verbatim, which is what keeps the provider's prefix cache warm across a restart. The accepted consequence is that prompt edits, new skills and newly connected integrations do not reach an existing thread. +- **So is the tool list it was sent.** Every turn records its tool + declarations in the transcript (a `{"kind":"tools"}` record, written only + when they change); resume restores them, the prefix, and the committed-turn + count. The host never shrinks a thread's tools because a cache went cold: + `session_host/recorded_tools.rs` rebuilds recorded Composio actions as + deferred executors, and the prelude fetches integrations on the first turn + of every session instance, not only on a brand-new thread. +- **So is the tool list it was sent.** Every turn records its tool + declarations in the transcript (a `{"kind":"tools"}` record, written only + when they change); resume restores them, the prefix, and the committed-turn + count. The host never shrinks a thread's tools because a cache went cold: + `session_host/recorded_tools.rs` rebuilds recorded Composio actions as + deferred executors, and the prelude fetches integrations on the first turn + of every session instance, not only on a brand-new thread. - **Pre-identity conversations are adopted once**, on first resume, from the timestamped stems they were written to (`adopt_legacy_session_transcripts`). No legacy file is modified. From 20030cf8a5758123d885755766f2b783f72f8dce Mon Sep 17 00:00:00 2001 From: Steven Enamakel Date: Thu, 24 Sep 2026 05:37:49 +0530 Subject: [PATCH 19/33] chore(AGENTS.md): remove duplicate bullet point about tool list recording Removed a bullet point that was an exact duplicate of the preceding entry, both describing how every turn records tool declarations in the transcript and how the host never shrinks a thread's tools due to cache misses. Auto-committed-on: macbook --- AGENTS.md | 7 ------- 1 file changed, 7 deletions(-) diff --git a/AGENTS.md b/AGENTS.md index e35760f523..57e8430fca 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -362,13 +362,6 @@ seed history by hand, or pick a transcript by recency. `session_host/recorded_tools.rs` rebuilds recorded Composio actions as deferred executors, and the prelude fetches integrations on the first turn of every session instance, not only on a brand-new thread. -- **So is the tool list it was sent.** Every turn records its tool - declarations in the transcript (a `{"kind":"tools"}` record, written only - when they change); resume restores them, the prefix, and the committed-turn - count. The host never shrinks a thread's tools because a cache went cold: - `session_host/recorded_tools.rs` rebuilds recorded Composio actions as - deferred executors, and the prelude fetches integrations on the first turn - of every session instance, not only on a brand-new thread. - **Pre-identity conversations are adopted once**, on first resume, from the timestamped stems they were written to (`adopt_legacy_session_transcripts`). No legacy file is modified. From dcc4582972b7e25b1d833ce0e61984206dbc4c97 Mon Sep 17 00:00:00 2001 From: Steven Enamakel Date: Thu, 24 Sep 2026 08:45:07 +0300 Subject: [PATCH 20/33] feat(session): filter rehydrated actions by current integration policy When rehydrating recorded integration actions during session resume, the rebuilt actions are now filtered against the current set of connected integrations and their gated tools. This prevents restoring actions that belong to a disconnected integration or that have been revoked by a scope policy change. The filter is only applied when the integration snapshot is known to be authoritative, preserving the previous fallback behaviour when Auto-committed-on: dragonfly --- .../src/agent/session_host/recorded_tools.rs | 17 +++++++ .../session_host/recorded_tools_tests.rs | 50 ++++++++++++++++--- .../src/agent/session_host/runtime_session.rs | 24 ++++++--- 3 files changed, 77 insertions(+), 14 deletions(-) diff --git a/crates/openhuman-core/src/agent/session_host/recorded_tools.rs b/crates/openhuman-core/src/agent/session_host/recorded_tools.rs index 011c460f0f..d3fef179c4 100644 --- a/crates/openhuman-core/src/agent/session_host/recorded_tools.rs +++ b/crates/openhuman-core/src/agent/session_host/recorded_tools.rs @@ -61,11 +61,28 @@ pub(super) fn recorded_integration_actions(recorded: &[ToolSpec]) -> Vec], + integrations: &[crate::agent::prompts::ConnectedIntegration], + integrations_are_authoritative: bool, ) -> Vec> { let live_names: HashSet<&str> = live.iter().map(|tool| tool.name()).collect(); let mut seen = HashSet::new(); recorded .iter() + // A missing snapshot is not evidence that a connection was revoked, + // so preserve the resume fallback until integrations can be fetched. + // Once the snapshot is authoritative, however, never reintroduce an + // action which its current connection or scope policy removed. + .filter(|spec| { + !integrations_are_authoritative + || integrations.iter().any(|integration| { + integration.connected + && integration.toolkit.eq_ignore_ascii_case(&toolkit_of(&spec.name)) + && !integration + .gated_tools + .iter() + .any(|gated| gated.name == spec.name) + }) + }) .filter(|spec| !live_names.contains(spec.name.as_str())) .filter(|spec| seen.insert(spec.name.clone())) .map(|spec| { diff --git a/crates/openhuman-core/src/agent/session_host/recorded_tools_tests.rs b/crates/openhuman-core/src/agent/session_host/recorded_tools_tests.rs index e8f09bf8c0..bb38ca07d2 100644 --- a/crates/openhuman-core/src/agent/session_host/recorded_tools_tests.rs +++ b/crates/openhuman-core/src/agent/session_host/recorded_tools_tests.rs @@ -11,6 +11,22 @@ fn spec(name: &str) -> ToolSpec { } } +fn integration( + toolkit: &str, + connected: bool, + gated_tools: Vec, +) -> crate::agent::prompts::ConnectedIntegration { + crate::agent::prompts::ConnectedIntegration { + toolkit: toolkit.into(), + description: String::new(), + tools: Vec::new(), + gated_tools, + connected, + connections: Vec::new(), + non_active_status: None, + } +} + #[test] fn only_composio_slugs_count_as_integration_actions() { assert!(is_integration_action_name("GMAIL_SEND_EMAIL")); @@ -36,7 +52,7 @@ fn recorded_actions_are_filtered_from_a_mixed_declaration_set() { #[test] fn a_restart_with_no_integrations_rebuilds_the_recorded_actions_as_deferred() { let recorded = vec![spec("GMAIL_SEND_EMAIL"), spec("GMAIL_FETCH_EMAILS")]; - let rebuilt = rehydrate_integration_actions(&recorded, &[]); + let rebuilt = rehydrate_integration_actions(&recorded, &[], &[], false); let names: Vec<&str> = rebuilt.iter().map(|tool| tool.name()).collect(); assert_eq!(names, vec!["GMAIL_SEND_EMAIL", "GMAIL_FETCH_EMAILS"]); @@ -54,19 +70,41 @@ fn a_restart_with_no_integrations_rebuilds_the_recorded_actions_as_deferred() { #[test] fn a_live_action_is_not_rebuilt_from_the_record() { let recorded = vec![spec("GMAIL_SEND_EMAIL"), spec("SLACK_SEND_MESSAGE")]; - let live: Vec> = rehydrate_integration_actions(&[spec("GMAIL_SEND_EMAIL")], &[]); - let rebuilt = rehydrate_integration_actions(&recorded, &live); + let live: Vec> = + rehydrate_integration_actions(&[spec("GMAIL_SEND_EMAIL")], &[], &[], false); + let rebuilt = rehydrate_integration_actions(&recorded, &live, &[], false); let names: Vec<&str> = rebuilt.iter().map(|tool| tool.name()).collect(); assert_eq!(names, vec!["SLACK_SEND_MESSAGE"]); } #[test] fn a_rebuilt_declaration_is_byte_identical_to_the_recorded_one() { - let first = rehydrate_integration_actions(&[spec("GMAIL_SEND_EMAIL")], &[]); - let recorded = vec![first[0].spec()]; - let second = rehydrate_integration_actions(&recorded, &[]); + let recorded = vec![spec("GMAIL_SEND_EMAIL")]; + let first = rehydrate_integration_actions(&recorded, &[], &[], false); + let second = rehydrate_integration_actions(&recorded, &[], &[], false); assert_eq!( serde_json::to_string(&first[0].spec()).unwrap(), serde_json::to_string(&second[0].spec()).unwrap() ); } + +#[test] +fn authoritative_integrations_do_not_restore_revoked_or_gated_actions() { + let recorded = vec![spec("GMAIL_SEND_EMAIL"), spec("SLACK_SEND_MESSAGE")]; + let integrations = vec![ + integration( + "gmail", + true, + vec![crate::agent::prompts::GatedIntegrationTool { + name: "GMAIL_SEND_EMAIL".into(), + description: String::new(), + required_scope: "write".into(), + unlock_paths: Vec::new(), + }], + ), + integration("slack", false, Vec::new()), + ]; + + let rebuilt = rehydrate_integration_actions(&recorded, &[], &integrations, true); + assert!(rebuilt.is_empty()); +} diff --git a/crates/openhuman-core/src/agent/session_host/runtime_session.rs b/crates/openhuman-core/src/agent/session_host/runtime_session.rs index 50eba93603..0340f18b36 100644 --- a/crates/openhuman-core/src/agent/session_host/runtime_session.rs +++ b/crates/openhuman-core/src/agent/session_host/runtime_session.rs @@ -421,12 +421,16 @@ impl OpenHumanTurnPrelude { if definition.subagents.is_empty() { return; } - let integrations = self - .mutable - .lock() - .unwrap_or_else(std::sync::PoisonError::into_inner) - .connected_integrations - .clone(); + let (integrations, integrations_are_authoritative) = { + let mutable = self + .mutable + .lock() + .unwrap_or_else(std::sync::PoisonError::into_inner); + ( + mutable.connected_integrations.clone(), + mutable.connected_integrations_initialized, + ) + }; let mut surface = self .tool_surface .lock() @@ -448,8 +452,12 @@ impl OpenHumanTurnPrelude { .unwrap_or_else(std::sync::PoisonError::into_inner) .recorded_integration_actions .clone(); - let rebuilt = - super::recorded_tools::rehydrate_integration_actions(&recorded, &collected); + let rebuilt = super::recorded_tools::rehydrate_integration_actions( + &recorded, + &collected, + &integrations, + integrations_are_authoritative, + ); if !rebuilt.is_empty() { log::info!( "[session] rebuilt {} recorded integration action(s) the live integrations did not supply agent={}", From 2db808166e2d21740c224ec9099de2b57ed18631 Mon Sep 17 00:00:00 2001 From: Steven Enamakel Date: Thu, 24 Sep 2026 08:45:53 +0300 Subject: [PATCH 21/33] feat(session): track whether connected integrations are authoritative When the backend is unreachable, the session falls back to a stale cached snapshot of connected integrations. Previously the system had no way to distinguish this fallback from a live authoritative list, which could cause the session to treat stale data as current. This change adds a boolean flag that records whether the integration list was fetched directly from the backend or came from a cache, allowing downstream logic to handle non-authoritative data appropriately. Auto-committed-on: dragonfly --- .../agent/session_host/builder/builder_build.rs | 1 + .../src/agent/session_host/prelude_integrations.rs | 14 ++++++++------ .../src/agent/session_host/runtime_session.rs | 4 +++- 3 files changed, 12 insertions(+), 7 deletions(-) diff --git a/crates/openhuman-core/src/agent/session_host/builder/builder_build.rs b/crates/openhuman-core/src/agent/session_host/builder/builder_build.rs index e3c4b1b381..8bc1c68da9 100644 --- a/crates/openhuman-core/src/agent/session_host/builder/builder_build.rs +++ b/crates/openhuman-core/src/agent/session_host/builder/builder_build.rs @@ -421,6 +421,7 @@ impl SessionHostBuilder { run_queue: None, connected_integrations: Vec::new(), connected_integrations_initialized: false, + connected_integrations_authoritative: false, runtime_config: None, hosted_base, definition: None, diff --git a/crates/openhuman-core/src/agent/session_host/prelude_integrations.rs b/crates/openhuman-core/src/agent/session_host/prelude_integrations.rs index a745206f6d..b80a0e5dfa 100644 --- a/crates/openhuman-core/src/agent/session_host/prelude_integrations.rs +++ b/crates/openhuman-core/src/agent/session_host/prelude_integrations.rs @@ -69,7 +69,7 @@ impl OpenHumanTurnPrelude { let Some(config) = config else { return; }; - let Some(connected) = load_connected_integrations(&config).await else { + let Some((connected, authoritative)) = load_connected_integrations(&config).await else { // Backend unreachable and nothing cached: stay un-hydrated so the // next turn retries rather than pinning an empty surface. log::warn!( @@ -94,6 +94,7 @@ impl OpenHumanTurnPrelude { .unwrap_or_else(std::sync::PoisonError::into_inner); mutable.connected_integrations = connected; mutable.connected_integrations_initialized = true; + mutable.connected_integrations_authoritative = authoritative; mutable.announced_integrations = mutable .connected_integrations .iter() @@ -108,10 +109,10 @@ impl OpenHumanTurnPrelude { // An expired cache is refetched rather than skipped, so a // long-lived session keeps tracking connects/revokes. let current = match crate::integrations::composio::cached_active_integrations(config) { - Some(current) => Some(current), + Some(current) => Some((current, true)), None => load_connected_integrations(config).await, }; - if let Some(current) = current { + if let Some((current, authoritative)) = current { let mut mutable = self .mutable .lock() @@ -126,6 +127,7 @@ impl OpenHumanTurnPrelude { } } mutable.connected_integrations = current; + mutable.connected_integrations_authoritative = authoritative; } } let connected_mcp = crate::mcp::registry::connections::connected_overview() @@ -184,10 +186,10 @@ impl OpenHumanTurnPrelude { /// there is neither a live answer nor any snapshot to fall back to. async fn load_connected_integrations( config: &crate::config::Config, -) -> Option> { +) -> Option<(Vec, bool)> { use crate::integrations::composio::FetchConnectedIntegrationsStatus; match crate::integrations::composio::fetch_connected_integrations_status(config).await { - FetchConnectedIntegrationsStatus::Authoritative(connected) => Some(connected), + FetchConnectedIntegrationsStatus::Authoritative(connected) => Some((connected, true)), FetchConnectedIntegrationsStatus::Unavailable => { let stale = crate::integrations::composio::cached_active_integrations_including_expired(config); @@ -195,7 +197,7 @@ async fn load_connected_integrations( "[session] integrations fetch unavailable; using stale snapshot={}", stale.as_ref().map_or(0, Vec::len) ); - stale + stale.map(|connected| (connected, false)) } } } diff --git a/crates/openhuman-core/src/agent/session_host/runtime_session.rs b/crates/openhuman-core/src/agent/session_host/runtime_session.rs index 0340f18b36..0199e3bb29 100644 --- a/crates/openhuman-core/src/agent/session_host/runtime_session.rs +++ b/crates/openhuman-core/src/agent/session_host/runtime_session.rs @@ -146,6 +146,7 @@ struct OpenHumanTurnPreludeMutable { pending_skill_retraction: Vec, connected_integrations: Vec, connected_integrations_initialized: bool, + connected_integrations_authoritative: bool, /// Integration action declarations this thread was already sent, /// restored by the tinyagents session on resume. Rebuilt into deferred /// executors whenever the live integrations list does not supply them @@ -428,7 +429,7 @@ impl OpenHumanTurnPrelude { .unwrap_or_else(std::sync::PoisonError::into_inner); ( mutable.connected_integrations.clone(), - mutable.connected_integrations_initialized, + mutable.connected_integrations_authoritative, ) }; let mut surface = self @@ -1491,6 +1492,7 @@ impl OpenHumanSessionHost { pending_skill_retraction: self.pending_skill_retraction.clone(), connected_integrations: self.connected_integrations.clone(), connected_integrations_initialized: self.connected_integrations_initialized, + connected_integrations_authoritative: self.connected_integrations_initialized, recorded_integration_actions: Vec::new(), workflows: self.workflows.clone(), composio_events: None, From 527e21cd6b309d777d0c4b079ec1be0abbddf7d7 Mon Sep 17 00:00:00 2001 From: Steven Enamakel Date: Thu, 24 Sep 2026 08:46:04 +0300 Subject: [PATCH 22/33] fix(recorded_tools): fix formatting of chained condition in rehydrate_integration_actions The chained boolean condition in the `rehydrate_integration_actions` function was reformatted to place each method call on its own line, improving readability without changing any behavior. Auto-committed-on: dragonfly --- .../openhuman-core/src/agent/session_host/recorded_tools.rs | 4 +++- 1 file changed, 3 insertions(+), 1 deletion(-) diff --git a/crates/openhuman-core/src/agent/session_host/recorded_tools.rs b/crates/openhuman-core/src/agent/session_host/recorded_tools.rs index d3fef179c4..e422d66623 100644 --- a/crates/openhuman-core/src/agent/session_host/recorded_tools.rs +++ b/crates/openhuman-core/src/agent/session_host/recorded_tools.rs @@ -76,7 +76,9 @@ pub(super) fn rehydrate_integration_actions( !integrations_are_authoritative || integrations.iter().any(|integration| { integration.connected - && integration.toolkit.eq_ignore_ascii_case(&toolkit_of(&spec.name)) + && integration + .toolkit + .eq_ignore_ascii_case(&toolkit_of(&spec.name)) && !integration .gated_tools .iter() From 6664e79098a7c249788fe38c4c650c2fea4febe5 Mon Sep 17 00:00:00 2001 From: Steven Enamakel Date: Thu, 24 Sep 2026 08:48:39 +0300 Subject: [PATCH 23/33] test(recorded_tools): add assertion that rebuilt declaration matches recorded one The test now verifies that a rebuilt tool declaration is byte-identical to the recorded one by stripping the connection_id field from the rebuilt spec before comparison, ensuring the round-trip through rehydration preserves all other fields exactly. Auto-committed-on: dragonfly --- .../src/agent/session_host/recorded_tools_tests.rs | 11 +++++++++++ 1 file changed, 11 insertions(+) diff --git a/crates/openhuman-core/src/agent/session_host/recorded_tools_tests.rs b/crates/openhuman-core/src/agent/session_host/recorded_tools_tests.rs index bb38ca07d2..308d9e3ed0 100644 --- a/crates/openhuman-core/src/agent/session_host/recorded_tools_tests.rs +++ b/crates/openhuman-core/src/agent/session_host/recorded_tools_tests.rs @@ -82,6 +82,17 @@ fn a_rebuilt_declaration_is_byte_identical_to_the_recorded_one() { let recorded = vec![spec("GMAIL_SEND_EMAIL")]; let first = rehydrate_integration_actions(&recorded, &[], &[], false); let second = rehydrate_integration_actions(&recorded, &[], &[], false); + let mut rebuilt = first[0].spec(); + rebuilt + .parameters + .get_mut("properties") + .and_then(serde_json::Value::as_object_mut) + .expect("rebuilt properties") + .remove("connection_id"); + assert_eq!( + serde_json::to_string(&rebuilt).unwrap(), + serde_json::to_string(&recorded[0]).unwrap() + ); assert_eq!( serde_json::to_string(&first[0].spec()).unwrap(), serde_json::to_string(&second[0].spec()).unwrap() From 8f4963d7ab8110ab7951db47c262b9ca0a490ca9 Mon Sep 17 00:00:00 2001 From: Steven Enamakel Date: Thu, 24 Sep 2026 09:18:44 +0300 Subject: [PATCH 24/33] fix(session): prune stale integration and MCP announcements on refresh Remove integration and MCP server announcements that are no longer present in the current authoritative snapshot, preventing stale tool declarations from persisting across session restarts. Also correct the authorization logic in `rehydrate_integration_actions` so that recorded actions are only rebuilt when the integration snapshot is authoritative and the toolkit is currently connected, rather than preserving all recorded actions as deferred when no snapshot is available. Auto-committed-on: dragonfly --- .../session_host/prelude_integrations.rs | 20 ++++++++--- .../src/agent/session_host/recorded_tools.rs | 11 +++--- .../session_host/recorded_tools_tests.rs | 34 +++++++++---------- .../src/agent/session_host/runtime_session.rs | 4 ++- 4 files changed, 40 insertions(+), 29 deletions(-) diff --git a/crates/openhuman-core/src/agent/session_host/prelude_integrations.rs b/crates/openhuman-core/src/agent/session_host/prelude_integrations.rs index b80a0e5dfa..4997425184 100644 --- a/crates/openhuman-core/src/agent/session_host/prelude_integrations.rs +++ b/crates/openhuman-core/src/agent/session_host/prelude_integrations.rs @@ -18,9 +18,6 @@ impl OpenHumanTurnPrelude { return; }; let actions = super::super::recorded_tools::recorded_integration_actions(recorded.specs()); - if actions.is_empty() { - return; - } log::debug!( "[session] adopting {} recorded integration action declaration(s) agent={}", actions.len(), @@ -105,7 +102,11 @@ impl OpenHumanTurnPrelude { pub(super) async fn refresh_dynamic_announcements(&self) { let skills_changed = self.drain_host_events(); - if let Some(config) = self.runtime_config.as_deref() { + let config = match self.runtime_config.clone() { + Some(config) => Some(config), + None => crate::config::Config::load_or_init().await.ok().map(Arc::new), + }; + if let Some(config) = config.as_deref() { // An expired cache is refetched rather than skipped, so a // long-lived session keeps tracking connects/revokes. let current = match crate::integrations::composio::cached_active_integrations(config) { @@ -119,6 +120,10 @@ impl OpenHumanTurnPrelude { .unwrap_or_else(std::sync::PoisonError::into_inner); let current_slugs: std::collections::HashSet<_> = current.iter().map(|item| item.toolkit.clone()).collect(); + mutable.announced_integrations.retain(|slug| current_slugs.contains(slug)); + mutable + .pending_integration_announcement + .retain(|slug| current_slugs.contains(slug)); for slug in ¤t_slugs { if mutable.announced_integrations.insert(slug.clone()) && !mutable.pending_integration_announcement.contains(slug) @@ -139,6 +144,13 @@ impl OpenHumanTurnPrelude { .mutable .lock() .unwrap_or_else(std::sync::PoisonError::into_inner); + let connected_mcp: std::collections::HashSet<_> = connected_mcp.into_iter().collect(); + mutable + .announced_mcp_servers + .retain(|server| connected_mcp.contains(server)); + mutable + .pending_mcp_announcement + .retain(|server| connected_mcp.contains(server)); for server in connected_mcp { if mutable.announced_mcp_servers.insert(server.clone()) && !mutable.pending_mcp_announcement.contains(&server) diff --git a/crates/openhuman-core/src/agent/session_host/recorded_tools.rs b/crates/openhuman-core/src/agent/session_host/recorded_tools.rs index e422d66623..37b0c4b43a 100644 --- a/crates/openhuman-core/src/agent/session_host/recorded_tools.rs +++ b/crates/openhuman-core/src/agent/session_host/recorded_tools.rs @@ -68,13 +68,12 @@ pub(super) fn rehydrate_integration_actions( let mut seen = HashSet::new(); recorded .iter() - // A missing snapshot is not evidence that a connection was revoked, - // so preserve the resume fallback until integrations can be fetched. - // Once the snapshot is authoritative, however, never reintroduce an - // action which its current connection or scope policy removed. + // Transcript declarations are historical state, never authorization. + // Do not make a deferred executor available until a current + // authoritative integration snapshot permits its toolkit and action. .filter(|spec| { - !integrations_are_authoritative - || integrations.iter().any(|integration| { + integrations_are_authoritative + && integrations.iter().any(|integration| { integration.connected && integration .toolkit diff --git a/crates/openhuman-core/src/agent/session_host/recorded_tools_tests.rs b/crates/openhuman-core/src/agent/session_host/recorded_tools_tests.rs index 308d9e3ed0..c5e2bba382 100644 --- a/crates/openhuman-core/src/agent/session_host/recorded_tools_tests.rs +++ b/crates/openhuman-core/src/agent/session_host/recorded_tools_tests.rs @@ -50,29 +50,26 @@ fn recorded_actions_are_filtered_from_a_mixed_declaration_set() { } #[test] -fn a_restart_with_no_integrations_rebuilds_the_recorded_actions_as_deferred() { +fn unavailable_authorization_does_not_rebuild_recorded_actions() { let recorded = vec![spec("GMAIL_SEND_EMAIL"), spec("GMAIL_FETCH_EMAILS")]; let rebuilt = rehydrate_integration_actions(&recorded, &[], &[], false); - - let names: Vec<&str> = rebuilt.iter().map(|tool| tool.name()).collect(); - assert_eq!(names, vec!["GMAIL_SEND_EMAIL", "GMAIL_FETCH_EMAILS"]); - for tool in &rebuilt { - assert_eq!(tool.exposure(), tinytools::ToolExposure::Deferred); - assert_eq!(tool.family(), Some("gmail")); - } - // The declaration round-trips: same description, and the schema keeps the - // recorded properties (the tool only adds `connection_id` when absent). - let schema = rebuilt[0].parameters_schema(); - assert_eq!(rebuilt[0].description(), "GMAIL_SEND_EMAIL description"); - assert!(schema["properties"]["to"].is_object()); + assert!(rebuilt.is_empty()); } #[test] fn a_live_action_is_not_rebuilt_from_the_record() { let recorded = vec![spec("GMAIL_SEND_EMAIL"), spec("SLACK_SEND_MESSAGE")]; - let live: Vec> = - rehydrate_integration_actions(&[spec("GMAIL_SEND_EMAIL")], &[], &[], false); - let rebuilt = rehydrate_integration_actions(&recorded, &live, &[], false); + let integrations = vec![ + integration("gmail", true, Vec::new()), + integration("slack", true, Vec::new()), + ]; + let live: Vec> = rehydrate_integration_actions( + &[spec("GMAIL_SEND_EMAIL")], + &[], + &integrations, + true, + ); + let rebuilt = rehydrate_integration_actions(&recorded, &live, &integrations, true); let names: Vec<&str> = rebuilt.iter().map(|tool| tool.name()).collect(); assert_eq!(names, vec!["SLACK_SEND_MESSAGE"]); } @@ -80,8 +77,9 @@ fn a_live_action_is_not_rebuilt_from_the_record() { #[test] fn a_rebuilt_declaration_is_byte_identical_to_the_recorded_one() { let recorded = vec![spec("GMAIL_SEND_EMAIL")]; - let first = rehydrate_integration_actions(&recorded, &[], &[], false); - let second = rehydrate_integration_actions(&recorded, &[], &[], false); + let integrations = vec![integration("gmail", true, Vec::new())]; + let first = rehydrate_integration_actions(&recorded, &[], &integrations, true); + let second = rehydrate_integration_actions(&recorded, &[], &integrations, true); let mut rebuilt = first[0].spec(); rebuilt .parameters diff --git a/crates/openhuman-core/src/agent/session_host/runtime_session.rs b/crates/openhuman-core/src/agent/session_host/runtime_session.rs index 0199e3bb29..20c19dc8d8 100644 --- a/crates/openhuman-core/src/agent/session_host/runtime_session.rs +++ b/crates/openhuman-core/src/agent/session_host/runtime_session.rs @@ -1492,7 +1492,9 @@ impl OpenHumanSessionHost { pending_skill_retraction: self.pending_skill_retraction.clone(), connected_integrations: self.connected_integrations.clone(), connected_integrations_initialized: self.connected_integrations_initialized, - connected_integrations_authoritative: self.connected_integrations_initialized, + // Builder-provided integrations have not been verified by + // this session's current authorization refresh. + connected_integrations_authoritative: false, recorded_integration_actions: Vec::new(), workflows: self.workflows.clone(), composio_events: None, From dc5b75bd817cd129415d3ef00a07709b88324f54 Mon Sep 17 00:00:00 2001 From: Steven Enamakel Date: Thu, 24 Sep 2026 09:19:13 +0300 Subject: [PATCH 25/33] fix(test): ensure rehydration test sets connected integrations The test for resumed orchestrator keeping integration actions now populates the connected integrations list with a Gmail entry before rehydration, so the test correctly verifies that rehydration is permitted only when the current authorization snapshot confirms the integration is connected. Auto-committed-on: dragonfly --- .../agent/session_host/runtime_session_tests.rs | 17 +++++++++++++++-- 1 file changed, 15 insertions(+), 2 deletions(-) diff --git a/crates/openhuman-core/src/agent/session_host/runtime_session_tests.rs b/crates/openhuman-core/src/agent/session_host/runtime_session_tests.rs index 44999defdc..71577ade52 100644 --- a/crates/openhuman-core/src/agent/session_host/runtime_session_tests.rs +++ b/crates/openhuman-core/src/agent/session_host/runtime_session_tests.rs @@ -16,8 +16,8 @@ fn spec(name: &str) -> ToolSpec { /// The incident this guards: a thread resumed in a fresh process (empty /// integrations cache) lost every Composio action, so the orchestrator's /// `tool_search` had nothing to find although its restored prompt told it to -/// search for the Gmail action. The session now restores the declarations the -/// thread was sent, and the prelude rebuilds them as executors. +/// search for the Gmail action. Rehydration is permitted only after the +/// current integration authorization snapshot confirms Gmail is connected. #[tokio::test] async fn a_resumed_orchestrator_keeps_the_integration_actions_it_was_sent() { let _ = crate::agent::harness::definition::AgentDefinitionRegistry::init_global_builtins(); @@ -55,6 +55,19 @@ async fn a_resumed_orchestrator_keeps_the_integration_actions_it_was_sent() { ]) .expect("snapshot"); prelude.adopt_recorded_tools(Some(&recorded)); + { + let mut mutable = prelude.mutable.lock().expect("prelude state"); + mutable.connected_integrations = vec![crate::agent::prompts::ConnectedIntegration { + toolkit: "gmail".into(), + description: String::new(), + tools: Vec::new(), + gated_tools: Vec::new(), + connected: true, + connections: Vec::new(), + non_active_status: None, + }]; + mutable.connected_integrations_authoritative = true; + } prelude.refresh_delegation_tool_surface(); let names = prelude.synthesized_tool_names_for_test(); From 02f10bfa7605c73b80b7c50b2c6ac7d4be220b96 Mon Sep 17 00:00:00 2001 From: Steven Enamakel Date: Thu, 24 Sep 2026 09:20:30 +0300 Subject: [PATCH 26/33] fix(agent): reformat integration prelude and test code Reformat the integration prelude code to improve readability by splitting long method chains across multiple lines, and simplify the test file by consolidating a function call onto a single line. No functional changes are introduced. Auto-committed-on: dragonfly --- .../src/agent/session_host/prelude_integrations.rs | 9 +++++++-- .../src/agent/session_host/recorded_tools_tests.rs | 8 ++------ 2 files changed, 9 insertions(+), 8 deletions(-) diff --git a/crates/openhuman-core/src/agent/session_host/prelude_integrations.rs b/crates/openhuman-core/src/agent/session_host/prelude_integrations.rs index 4997425184..8f75362cef 100644 --- a/crates/openhuman-core/src/agent/session_host/prelude_integrations.rs +++ b/crates/openhuman-core/src/agent/session_host/prelude_integrations.rs @@ -104,7 +104,10 @@ impl OpenHumanTurnPrelude { let skills_changed = self.drain_host_events(); let config = match self.runtime_config.clone() { Some(config) => Some(config), - None => crate::config::Config::load_or_init().await.ok().map(Arc::new), + None => crate::config::Config::load_or_init() + .await + .ok() + .map(Arc::new), }; if let Some(config) = config.as_deref() { // An expired cache is refetched rather than skipped, so a @@ -120,7 +123,9 @@ impl OpenHumanTurnPrelude { .unwrap_or_else(std::sync::PoisonError::into_inner); let current_slugs: std::collections::HashSet<_> = current.iter().map(|item| item.toolkit.clone()).collect(); - mutable.announced_integrations.retain(|slug| current_slugs.contains(slug)); + mutable + .announced_integrations + .retain(|slug| current_slugs.contains(slug)); mutable .pending_integration_announcement .retain(|slug| current_slugs.contains(slug)); diff --git a/crates/openhuman-core/src/agent/session_host/recorded_tools_tests.rs b/crates/openhuman-core/src/agent/session_host/recorded_tools_tests.rs index c5e2bba382..f0a5e69c3b 100644 --- a/crates/openhuman-core/src/agent/session_host/recorded_tools_tests.rs +++ b/crates/openhuman-core/src/agent/session_host/recorded_tools_tests.rs @@ -63,12 +63,8 @@ fn a_live_action_is_not_rebuilt_from_the_record() { integration("gmail", true, Vec::new()), integration("slack", true, Vec::new()), ]; - let live: Vec> = rehydrate_integration_actions( - &[spec("GMAIL_SEND_EMAIL")], - &[], - &integrations, - true, - ); + let live: Vec> = + rehydrate_integration_actions(&[spec("GMAIL_SEND_EMAIL")], &[], &integrations, true); let rebuilt = rehydrate_integration_actions(&recorded, &live, &integrations, true); let names: Vec<&str> = rebuilt.iter().map(|tool| tool.name()).collect(); assert_eq!(names, vec!["SLACK_SEND_MESSAGE"]); From 8ad1c03d2ddd30624d900b0a45abc25d52eee9cb Mon Sep 17 00:00:00 2001 From: Steven Enamakel Date: Thu, 24 Sep 2026 09:39:40 +0300 Subject: [PATCH 27/33] chore(deps): update tinyagents submodule Updated the tinyagents submodule to a newer commit, incorporating upstream changes. Auto-committed-on: dragonfly --- vendor/tinyagents | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/vendor/tinyagents b/vendor/tinyagents index 3821811095..77464dea0c 160000 --- a/vendor/tinyagents +++ b/vendor/tinyagents @@ -1 +1 @@ -Subproject commit 382181109579b29092fec2ffd67837763a299148 +Subproject commit 77464dea0c7c89d2240943c9139ca8d88e31fe2e From 094e35e00eb7fdaaaba151359dfc1e2c9386e664 Mon Sep 17 00:00:00 2001 From: Steven Enamakel Date: Thu, 24 Sep 2026 09:39:56 +0300 Subject: [PATCH 28/33] chore(deps): update tinyagents submodule The tinyagents submodule pointer has been advanced to include the latest upstream changes, keeping the dependency in sync with the current development state. Auto-committed-on: dragonfly --- vendor/tinyagents | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/vendor/tinyagents b/vendor/tinyagents index 77464dea0c..258b4ca924 160000 --- a/vendor/tinyagents +++ b/vendor/tinyagents @@ -1 +1 @@ -Subproject commit 77464dea0c7c89d2240943c9139ca8d88e31fe2e +Subproject commit 258b4ca924e469cb07fa58d2b88cb1eb8da60af1 From ef23298244700428ef027a5533be75272811caa4 Mon Sep 17 00:00:00 2001 From: Steven Enamakel Date: Thu, 24 Sep 2026 09:41:31 +0300 Subject: [PATCH 29/33] chore(deps): update tinyagents submodule The tinyagents submodule pointer has been advanced to include the latest upstream changes. Auto-committed-on: dragonfly --- vendor/tinyagents | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/vendor/tinyagents b/vendor/tinyagents index 258b4ca924..3821811095 160000 --- a/vendor/tinyagents +++ b/vendor/tinyagents @@ -1 +1 @@ -Subproject commit 258b4ca924e469cb07fa58d2b88cb1eb8da60af1 +Subproject commit 382181109579b29092fec2ffd67837763a299148 From f0105e8b738b9fd44dba74e1c9598e1083fbbd4e Mon Sep 17 00:00:00 2001 From: Steven Enamakel Date: Thu, 24 Sep 2026 09:57:47 +0300 Subject: [PATCH 30/33] fix(session_host): skip non-integration tools during rehydration Only Composio-style `TOOLKIT_ACTION` names should be reconstructed from recorded tool declarations. A recorded OpenHuman tool such as `web_fetch` is historical prompt state, not an integration action, and attempting to rehydrate it would create a stale executor. The hydration flag is now set to the authoritative value so that a fallback snapshot does not permanently prevent a later turn from performing a live integration lookup. Auto-committed-on: dragonfly --- .../src/agent/session_host/prelude_integrations.rs | 5 ++++- .../src/agent/session_host/recorded_tools.rs | 4 ++++ .../src/agent/session_host/recorded_tools_tests.rs | 10 ++++++++++ 3 files changed, 18 insertions(+), 1 deletion(-) diff --git a/crates/openhuman-core/src/agent/session_host/prelude_integrations.rs b/crates/openhuman-core/src/agent/session_host/prelude_integrations.rs index 8f75362cef..258dcbffd4 100644 --- a/crates/openhuman-core/src/agent/session_host/prelude_integrations.rs +++ b/crates/openhuman-core/src/agent/session_host/prelude_integrations.rs @@ -90,7 +90,10 @@ impl OpenHumanTurnPrelude { .lock() .unwrap_or_else(std::sync::PoisonError::into_inner); mutable.connected_integrations = connected; - mutable.connected_integrations_initialized = true; + // A stale fallback is useful for announcements but cannot authorize + // restored executors. Leave hydration pending so a later turn retries + // the live lookup rather than pinning this session to the snapshot. + mutable.connected_integrations_initialized = authoritative; mutable.connected_integrations_authoritative = authoritative; mutable.announced_integrations = mutable .connected_integrations diff --git a/crates/openhuman-core/src/agent/session_host/recorded_tools.rs b/crates/openhuman-core/src/agent/session_host/recorded_tools.rs index 37b0c4b43a..20d7c0de4f 100644 --- a/crates/openhuman-core/src/agent/session_host/recorded_tools.rs +++ b/crates/openhuman-core/src/agent/session_host/recorded_tools.rs @@ -68,6 +68,10 @@ pub(super) fn rehydrate_integration_actions( let mut seen = HashSet::new(); recorded .iter() + // Only Composio's upper-case `TOOLKIT_ACTION` declarations may be + // reconstructed. A recorded OpenHuman tool such as `web_fetch` is + // historical prompt state, not an integration action. + .filter(|spec| is_integration_action_name(&spec.name)) // Transcript declarations are historical state, never authorization. // Do not make a deferred executor available until a current // authoritative integration snapshot permits its toolkit and action. diff --git a/crates/openhuman-core/src/agent/session_host/recorded_tools_tests.rs b/crates/openhuman-core/src/agent/session_host/recorded_tools_tests.rs index f0a5e69c3b..c5bd47c9f4 100644 --- a/crates/openhuman-core/src/agent/session_host/recorded_tools_tests.rs +++ b/crates/openhuman-core/src/agent/session_host/recorded_tools_tests.rs @@ -56,6 +56,16 @@ fn unavailable_authorization_does_not_rebuild_recorded_actions() { assert!(rebuilt.is_empty()); } +#[test] +fn non_integration_declarations_are_never_rehydrated() { + let recorded = vec![spec("web_fetch"), spec("GMAIL_SEND_EMAIL")]; + let integrations = vec![integration("web", true, Vec::new()), integration("gmail", true, Vec::new())]; + + let rebuilt = rehydrate_integration_actions(&recorded, &[], &integrations, true); + let names: Vec<&str> = rebuilt.iter().map(|tool| tool.name()).collect(); + assert_eq!(names, vec!["GMAIL_SEND_EMAIL"]); +} + #[test] fn a_live_action_is_not_rebuilt_from_the_record() { let recorded = vec![spec("GMAIL_SEND_EMAIL"), spec("SLACK_SEND_MESSAGE")]; From d2349f1d27cd76c5111ea02225ceef4bb55bcbb5 Mon Sep 17 00:00:00 2001 From: Steven Enamakel Date: Thu, 24 Sep 2026 09:58:29 +0300 Subject: [PATCH 31/33] test(recorded-tools): format test vector for readability Reformatted the `integrations` vector initialization in the `non_integration_declarations_are_never_rehydrated` test to use one element per line, improving code readability without changing any behavior. Auto-committed-on: dragonfly --- .../src/agent/session_host/recorded_tools_tests.rs | 5 ++++- 1 file changed, 4 insertions(+), 1 deletion(-) diff --git a/crates/openhuman-core/src/agent/session_host/recorded_tools_tests.rs b/crates/openhuman-core/src/agent/session_host/recorded_tools_tests.rs index c5bd47c9f4..ae37759e20 100644 --- a/crates/openhuman-core/src/agent/session_host/recorded_tools_tests.rs +++ b/crates/openhuman-core/src/agent/session_host/recorded_tools_tests.rs @@ -59,7 +59,10 @@ fn unavailable_authorization_does_not_rebuild_recorded_actions() { #[test] fn non_integration_declarations_are_never_rehydrated() { let recorded = vec![spec("web_fetch"), spec("GMAIL_SEND_EMAIL")]; - let integrations = vec![integration("web", true, Vec::new()), integration("gmail", true, Vec::new())]; + let integrations = vec![ + integration("web", true, Vec::new()), + integration("gmail", true, Vec::new()), + ]; let rebuilt = rehydrate_integration_actions(&recorded, &[], &integrations, true); let names: Vec<&str> = rebuilt.iter().map(|tool| tool.name()).collect(); From 0f4473b56f9a65ca11d4d2fe2896ef5b7cfd4bed Mon Sep 17 00:00:00 2001 From: Steven Enamakel Date: Thu, 24 Sep 2026 10:01:00 +0300 Subject: [PATCH 32/33] fix(runtime_session): restore recorded tools on session resume When resuming a session, the host now loads a bound but empty runtime session before the normal lifecycle runs, allowing the prelude to rebuild only its permitted recorded integration executors for that turn. Previously, recorded tools were adopted inside the turn loop after the session had already been restored, which could miss the correct tool set. The change also moves the `exact_tools` flag into the `ToolSnapshot` constructor and enables `retain_recorded_tools` on the session builder to preserve tool declarations across resumption. Auto-committed-on: dragonfly --- .../src/agent/session_host/runtime_session.rs | 43 +++++++++++++++---- 1 file changed, 34 insertions(+), 9 deletions(-) diff --git a/crates/openhuman-core/src/agent/session_host/runtime_session.rs b/crates/openhuman-core/src/agent/session_host/runtime_session.rs index 20c19dc8d8..5a6f3e93ea 100644 --- a/crates/openhuman-core/src/agent/session_host/runtime_session.rs +++ b/crates/openhuman-core/src/agent/session_host/runtime_session.rs @@ -225,7 +225,6 @@ impl OpenHumanTurnPrelude { Ok(TurnPreparation { prefix, tools: Some(tools), - exact_tools: false, }) } fn begin_user_effects(&self, state: &mut OpenHumanSessionState, request: &SessionTurnRequest) { @@ -1327,6 +1326,37 @@ impl OpenHumanSessionHost { cancellation, run_context: context.into_tinyagents(root_config), }; + // `tinyagents_runtime::Session` owns the restored declaration + // snapshot. Load a bound, otherwise empty session before its normal + // lifecycle runs so the host prelude can rebuild only its permitted + // recorded integration executors for this turn. + if matches!(options.resume, ResumeMode::Session) + && self + .runtime_session + .as_ref() + .is_some_and(|session| session.history().is_empty()) + { + let runtime = self + .runtime_session + .as_mut() + .expect("runtime session initialized"); + let resumed = runtime + .resume(&options) + .await + .map_err(|error| anyhow::anyhow!(error.to_string()))?; + if resumed.loaded { + let recorded_tools = runtime.recorded_tools().cloned(); + if let Some(prelude) = self + .runtime_state + .lock() + .unwrap_or_else(|poisoned| poisoned.into_inner()) + .prelude + .clone() + { + prelude.adopt_recorded_tools(recorded_tools.as_ref()); + } + } + } let outcome = self .runtime_session .as_mut() @@ -1528,10 +1558,6 @@ impl OpenHumanSessionHost { let state = state.clone(); let request_base_len = view.history.len() + usize::from(view.history.last() != Some(&request.input)); - // The session restores the prefix and the tool - // declarations this thread was sent; the host only - // rebuilds executors for them. - let recorded_tools = view.recorded_tools.cloned(); Box::pin(async move { let transcript_snapshot = crate::agent::tinyagents::TranscriptSnapshotSink::default(); @@ -1549,7 +1575,6 @@ impl OpenHumanSessionHost { "OpenHumanTurnPrelude", ) })?; - prelude.adopt_recorded_tools(recorded_tools.as_ref()); prelude .refresh_turn_boundary(!view.resumed && view.history.is_empty()) .await; @@ -1595,8 +1620,7 @@ impl OpenHumanSessionHost { if overrides.suppress_tools { // One-off tool-less turn: must not become the // thread's recorded tool list. - preparation.tools = Some(ToolSnapshot::default()); - preparation.exact_tools = true; + preparation.tools = Some(ToolSnapshot::default().exact()); } let ( mut current_tools, @@ -1804,7 +1828,8 @@ impl OpenHumanSessionHost { // driving a provider first. let mut builder = SessionBuilder::new(driver) .codec(Arc::new(OpenHumanTranscriptCodec)) - .hooks(hooks); + .hooks(hooks) + .retain_recorded_tools(true); if let Some(session) = self.session.clone() { builder = builder.session( self.session_locator(), From 1b9788fdffd40f89be245608790c7f90820b8743 Mon Sep 17 00:00:00 2001 From: Steven Enamakel Date: Thu, 24 Sep 2026 10:01:42 +0300 Subject: [PATCH 33/33] fix(builder): remove authoritative flag from session host builder The `connected_integrations_authoritative` field was removed from the session host builder's default initialization as it is no longer needed for the session host configuration. Auto-committed-on: dragonfly --- .../src/agent/session_host/builder/builder_build.rs | 1 - 1 file changed, 1 deletion(-) diff --git a/crates/openhuman-core/src/agent/session_host/builder/builder_build.rs b/crates/openhuman-core/src/agent/session_host/builder/builder_build.rs index 8bc1c68da9..e3c4b1b381 100644 --- a/crates/openhuman-core/src/agent/session_host/builder/builder_build.rs +++ b/crates/openhuman-core/src/agent/session_host/builder/builder_build.rs @@ -421,7 +421,6 @@ impl SessionHostBuilder { run_queue: None, connected_integrations: Vec::new(), connected_integrations_initialized: false, - connected_integrations_authoritative: false, runtime_config: None, hosted_base, definition: None,