fix(session): resumed threads keep their integration tools and tool_search

AGENTS.md

@@ -389,6 +389,13 @@ 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.
 - **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.

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."),

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;

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<PrefixSnapshot> {
-    let leading: Vec<Message> = history
-        .iter()
-        .take_while(|message| matches!(message, Message::System(_)))
-        .cloned()
-        .collect();
-    (!leading.is_empty()).then(|| PrefixSnapshot::new(leading))
-}

crates/openhuman-core/src/agent/session_host/prelude_integrations.rs

@@ -0,0 +1,223 @@
+//! 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 std::sync::Arc;
+
+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::super::recorded_tools::recorded_integration_actions(recorded.specs());
+        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, 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!(
+                "[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::<std::collections::HashSet<_>>();
+        let mut mutable = self
+            .mutable
+            .lock()
+            .unwrap_or_else(std::sync::PoisonError::into_inner);
+        mutable.connected_integrations = connected;
+        // 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
+            .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();
+        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) {
+                Some(current) => Some((current, true)),
+                None => load_connected_integrations(config).await,
+            };
+            if let Some((current, authoritative)) = 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();
+                mutable
+                    .announced_integrations
+                    .retain(|slug| current_slugs.contains(slug));
+                mutable
+                    .pending_integration_announcement
+                    .retain(|slug| current_slugs.contains(slug));
+                for slug in &current_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;
+                mutable.connected_integrations_authoritative = authoritative;
+            }
+        }
+        let connected_mcp = crate::mcp::registry::connections::connected_overview()
+            .await
+            .into_iter()
+            .map(|server| server.qualified_name)
+            .collect::<Vec<_>>();
+        let mut mutable = self
+            .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)
+            {
+                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(&current) {
+            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;
+    }
+}
+
+/// 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<(Vec<crate::agent::prompts::ConnectedIntegration>, bool)> {
+    use crate::integrations::composio::FetchConnectedIntegrationsStatus;
+    match crate::integrations::composio::fetch_connected_integrations_status(config).await {
+        FetchConnectedIntegrationsStatus::Authoritative(connected) => Some((connected, true)),
+        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.map(|connected| (connected, false))
+        }
+    }
+}