From 1c582d212da4a6a1e4d75853d0ee1d72227b050c Mon Sep 17 00:00:00 2001 From: Sami Shukri Date: Mon, 29 Jun 2026 16:57:16 -0600 Subject: [PATCH] perf(chat): run 'chat search' over one plugin host (38 spawns -> 1) chat search was O(conversations): the conversation_store Plugin client spawns + reaps the plugin host (and its PG pool) on EVERY rpc, and search did list + load_messages-per-conversation, so a search spawned the plugin N+1 times (~30s for 37 tiny conversations -> portal 502). Add search_async: spawn the host once, handshake once, run list + every load_messages over the live host, reap once. The File store keeps the cheap in-process scan. v0.6.24. --- Cargo.lock | 2 +- crates/orchestrator-cli/Cargo.toml | 2 +- .../services/runtime/runtime_chat/client.rs | 98 +++++++++++++++++++ .../src/services/runtime/runtime_chat/mod.rs | 4 +- 4 files changed, 102 insertions(+), 4 deletions(-) diff --git a/Cargo.lock b/Cargo.lock index 59208a55..d7168736 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -2764,7 +2764,7 @@ checksum = "04744f49eae99ab78e0d5c0b603ab218f515ea8cfe5a456d7629ad883a3b6e7d" [[package]] name = "orchestrator-cli" -version = "0.6.23" +version = "0.6.24" dependencies = [ "animus-actor", "animus-control-protocol", diff --git a/crates/orchestrator-cli/Cargo.toml b/crates/orchestrator-cli/Cargo.toml index dbc23dda..8e228809 100644 --- a/crates/orchestrator-cli/Cargo.toml +++ b/crates/orchestrator-cli/Cargo.toml @@ -1,6 +1,6 @@ [package] name = "orchestrator-cli" -version = "0.6.23" +version = "0.6.24" edition = "2021" license = "Elastic-2.0" default-run = "animus" diff --git a/crates/orchestrator-cli/src/services/runtime/runtime_chat/client.rs b/crates/orchestrator-cli/src/services/runtime/runtime_chat/client.rs index 80cffbd2..1b7163c2 100644 --- a/crates/orchestrator-cli/src/services/runtime/runtime_chat/client.rs +++ b/crates/orchestrator-cli/src/services/runtime/runtime_chat/client.rs @@ -71,6 +71,25 @@ impl ConversationStoreClient { } } + /// Search conversations for `query`, newest-first, up to `limit` matches. + /// The File store reads in-process, so the generic per-conversation scan is + /// cheap. The Plugin store spawns + reaps the host on EVERY rpc, so the + /// scan's N+1 (`list` + `load_messages` per conversation) would spawn the + /// plugin N+1 times — instead we run the whole scan over ONE spawned host + /// (see [`PluginConversationStore::search_async`]). + pub(crate) fn search( + &self, + query: &str, + case_insensitive: bool, + limit: usize, + as_user: Option<&str>, + ) -> Result> { + match self { + Self::File(_) => super::search_conversations(self, query, case_insensitive, limit, as_user), + Self::Plugin(plugin) => run_blocking(plugin.search_async(query, case_insensitive, limit, as_user))?, + } + } + /// Build a `File`-backed client rooted at an explicit directory. Test-only /// escape hatch mirroring [`FileConversationStore::with_root_for_test`] so /// query-layer helpers that take a `ConversationStoreClient` can be @@ -393,6 +412,85 @@ impl PluginConversationStore { .with_context(|| format!("{method} on conversation_store plugin {}", self.plugin.name))?; serde_json::from_value(value).with_context(|| format!("decoding {method} response")) } + + /// Run a whole `chat search` over ONE spawned plugin host. The per-rpc + /// `call` path spawns+reaps the host every call, so the search's N+1 (a + /// `list` then one `load_messages` per conversation) would spawn the plugin + /// N+1 times. Here we spawn once, handshake once, run every rpc over the live host, and + /// reap once — turning ~N spawns into a single one. + async fn search_async( + &self, + query: &str, + case_insensitive: bool, + limit: usize, + as_user: Option<&str>, + ) -> Result> { + if query.is_empty() || limit == 0 { + return Ok(Vec::new()); + } + let options = PluginSpawnOptions::for_manifest( + self.plugin.name.clone(), + &self.plugin.manifest.env_required, + std::iter::empty::(), + None, + ) + .with_working_dir(self.project_root.clone()); + let host = PluginHost::spawn_with_options(&self.plugin.path, &[], options) + .await + .with_context(|| format!("spawning conversation_store plugin {}", self.plugin.name))?; + // Reap the host (and its DB pool) no matter how the scan ends — a + // persistent stdio plugin never EOFs, so dropping the handle leaks it. + let scan = async { + host.handshake() + .await + .with_context(|| format!("handshake with conversation_store plugin {}", self.plugin.name))?; + let list_params = serde_json::to_value(proto::ConversationListRequest { + scope: self.scope(), + as_user: as_user.map(ToOwned::to_owned), + })?; + let list_val = host + .request_typed_with_timeout(proto::METHOD_CONVERSATION_LIST, Some(list_params), RPC_TIMEOUT) + .await + .with_context(|| format!("conversation/list on {}", self.plugin.name))?; + let list_resp: proto::ConversationListResponse = serde_json::from_value(list_val)?; + let mut out: Vec = Vec::new(); + for psummary in list_resp.conversations { + if out.len() >= limit { + break; + } + let summary = from_proto_summary(psummary); + let lm_params = serde_json::to_value(proto::ConversationLoadMessagesRequest { + scope: self.scope(), + id: summary.id.clone(), + as_user: self.acting_user(), + })?; + let lm_val = host + .request_typed_with_timeout(proto::METHOD_CONVERSATION_LOAD_MESSAGES, Some(lm_params), RPC_TIMEOUT) + .await + .with_context(|| format!("conversation/load_messages on {}", self.plugin.name))?; + let lm_resp: proto::ConversationLoadMessagesResponse = serde_json::from_value(lm_val)?; + for pm in lm_resp.messages { + if out.len() >= limit { + break; + } + let m = from_proto_message(pm)?; + if let Some(snippet) = super::snippet_around(&m.content, query, case_insensitive) { + out.push(super::SearchMatch { + conversation_id: summary.id.clone(), + title: summary.title.clone(), + role: super::role_str(m.role), + seq: m.seq, + snippet, + }); + } + } + } + Ok::<_, anyhow::Error>(out) + } + .await; + let _ = host.shutdown().await; + scan + } } /// Bridge an async future into the sync `ConversationStore` trait. Works diff --git a/crates/orchestrator-cli/src/services/runtime/runtime_chat/mod.rs b/crates/orchestrator-cli/src/services/runtime/runtime_chat/mod.rs index 26b8d125..36c5a086 100644 --- a/crates/orchestrator-cli/src/services/runtime/runtime_chat/mod.rs +++ b/crates/orchestrator-cli/src/services/runtime/runtime_chat/mod.rs @@ -40,7 +40,7 @@ pub(crate) async fn handle_chat(command: ChatCommand, project_root: &str, json: } #[derive(Debug, Serialize, PartialEq)] -struct SearchMatch { +pub(crate) struct SearchMatch { conversation_id: String, #[serde(skip_serializing_if = "Option::is_none")] title: Option, @@ -142,7 +142,7 @@ fn search_conversations( fn handle_chat_search(args: ChatSearchArgs, project_root: &str, json: bool) -> Result<()> { let store = ConversationStoreClient::for_project_as_user(Path::new(project_root), args.as_user.as_deref())?; - let matches = search_conversations(&store, &args.query, !args.case_sensitive, args.limit, args.as_user.as_deref())?; + let matches = store.search(&args.query, !args.case_sensitive, args.limit, args.as_user.as_deref())?; print_value(matches, json) }