From 506dc5b225bd5e2bb81f09b22cb5b80fedb6e4c1 Mon Sep 17 00:00:00 2001 From: Ryan Ward Date: Mon, 14 Sep 2026 16:17:16 -0500 Subject: [PATCH] Reduce indexing overhead and repair agent search and child recovery --- Cargo.lock | 2 - Cargo.toml | 3 + app/src/ai/agent/conversation.rs | 52 +- .../action_model/execute/search_codebase.rs | 66 ++- .../execute/search_codebase_context.rs | 121 +++++ .../execute/search_codebase_context_tests.rs | 68 +++ .../action_model/execute/send_message.rs | 68 ++- .../action_model/execute/start_agent.rs | 5 + app/src/ai/blocklist/controller.rs | 32 +- app/src/ai/blocklist/orchestration_events.rs | 168 ++++++- .../blocklist/orchestration_events_tests.rs | 36 ++ app/src/ai/get_relevant_files/controller.rs | 3 +- app/src/ai/openai/response_translator.rs | 1 + app/src/ai/provider/request_translator.rs | 31 +- .../ai/provider/request_translator_tests.rs | 25 + app/src/ai/provider/response_translator.rs | 1 + app/src/ai/remote_logging.rs | 13 +- app/src/ai/runtime/rig_request.rs | 2 +- app/src/pane_group/mod.rs | 19 + app/src/pane_group/mod_tests.rs | 62 ++- app/src/workspace/view.rs | 69 +++ crates/ai/src/agent/action_result/convert.rs | 5 + crates/ai/src/agent/action_result/mod.rs | 3 + crates/ai/src/index/file_outline/mod.rs | 1 + crates/ai/src/index/file_outline/native.rs | 15 +- .../ai/src/index/local_project_index/mod.rs | 447 ++++++++++++++---- .../ai/src/index/local_project_index/tests.rs | 127 +++++ crates/command-signatures-v2/build.rs | 6 +- crates/galaxy_agent_core/src/provider_run.rs | 55 +-- .../src/provider_run_tests.rs | 39 ++ crates/warp_search_core/src/searcher.rs | 57 ++- crates/warp_search_core/src/searcher_tests.rs | 33 ++ vendor/block/Cargo.toml | 20 + vendor/block/GALAXY_PATCH.md | 11 + vendor/block/LICENSE | 21 + vendor/block/README.md | 42 ++ vendor/block/src/lib.rs | 399 ++++++++++++++++ vendor/block/src/test_utils.rs | 31 ++ 38 files changed, 1941 insertions(+), 218 deletions(-) create mode 100644 app/src/ai/blocklist/action_model/execute/search_codebase_context.rs create mode 100644 app/src/ai/blocklist/action_model/execute/search_codebase_context_tests.rs create mode 100644 vendor/block/Cargo.toml create mode 100644 vendor/block/GALAXY_PATCH.md create mode 100644 vendor/block/LICENSE create mode 100644 vendor/block/README.md create mode 100644 vendor/block/src/lib.rs create mode 100644 vendor/block/src/test_utils.rs diff --git a/Cargo.lock b/Cargo.lock index 7f337f4e..3fcc501e 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -2281,8 +2281,6 @@ dependencies = [ [[package]] name = "block" version = "0.1.6" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "0d8c1fef690941d3e7788d328517591fecc684c084084702d6ff1641e993699a" [[package]] name = "block-buffer" diff --git a/Cargo.toml b/Cargo.toml index c4a4b981..62428508 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -4,6 +4,7 @@ members = [ "app", ] resolver = "2" +exclude = ["vendor/block"] # Include only the application and core libraries used by the shipped local # client. Integration-only and server-only crates remain outside default builds. @@ -532,6 +533,8 @@ inherits = "dev" opt-level = "s" [patch.crates-io] +# Keep the legacy Blocks ABI while fixing its uninhabited external static. +block = { path = "vendor/block" } # Pinned to the merge commit of servo/core-foundation-rs#746, which fixes a # double-retain bug in `CTFontCollection::get_descriptors` that leaks the # CoreText font-descriptor NSArray on every call. The fix is not yet in any diff --git a/app/src/ai/agent/conversation.rs b/app/src/ai/agent/conversation.rs index 06b708da..ca8b3ba6 100644 --- a/app/src/ai/agent/conversation.rs +++ b/app/src/ai/agent/conversation.rs @@ -818,57 +818,7 @@ impl AIConversation { &mut self, messages: Vec, ) { - use crate::ai::provider::convert::{ContentPart, MessageContent}; - - // Cap the archive to prevent unbounded growth. The archive is only used by - // `recall_tool_history` which already truncates individual results to 50K chars, - // so retaining the most recent entries is sufficient for lookup. - const MAX_TOOL_RESULT_ARCHIVE_ENTRIES: usize = 400; - - let mut pending_tool_uses: Vec = - Vec::new(); - - for msg in messages { - match &msg.content { - MessageContent::ToolUse { .. } => pending_tool_uses.push(msg), - MessageContent::ToolResult { .. } => { - if !pending_tool_uses.is_empty() { - let tool_use = pending_tool_uses.remove(pending_tool_uses.len() - 1); - self.tool_result_archive.push(tool_use); - } - self.tool_result_archive.push(msg); - } - MessageContent::MultiPart(parts) => { - for part in parts { - match part { - ContentPart::ToolUse { .. } => { - pending_tool_uses.push(msg.clone()); - } - ContentPart::ToolResult { .. } => { - if !pending_tool_uses.is_empty() { - let tool_use = - pending_tool_uses.remove(pending_tool_uses.len() - 1); - self.tool_result_archive.push(tool_use); - } - self.tool_result_archive.push(msg.clone()); - } - _ => {} - } - } - } - _ => {} - } - } - - for tool_use in pending_tool_uses { - self.tool_result_archive.push(tool_use); - } - - // Evict oldest entries if the archive exceeds the cap. - if self.tool_result_archive.len() > MAX_TOOL_RESULT_ARCHIVE_ENTRIES { - let excess = self.tool_result_archive.len() - MAX_TOOL_RESULT_ARCHIVE_ENTRIES; - self.tool_result_archive.drain(0..excess); - } + galaxy_agent_core::archive_tool_results(&mut self.tool_result_archive, messages); } pub fn append_to_bedrock_history( diff --git a/app/src/ai/blocklist/action_model/execute/search_codebase.rs b/app/src/ai/blocklist/action_model/execute/search_codebase.rs index c31a264e..bcc05c8a 100644 --- a/app/src/ai/blocklist/action_model/execute/search_codebase.rs +++ b/app/src/ai/blocklist/action_model/execute/search_codebase.rs @@ -27,12 +27,21 @@ use crate::features::FeatureFlag; use crate::terminal::model::session::active_session::ActiveSession; use crate::{send_telemetry_from_ctx, TelemetryEvent}; +#[path = "search_codebase_context.rs"] +mod context; +use context::{bound_search_result, MAX_SEARCH_FILES, MAX_SEARCH_SCAN_BYTES}; + +struct PendingSearch { + sender: oneshot::Sender, + query: String, +} + pub struct SearchCodebaseExecutor { active_session: ModelHandle, get_relevant_files_controller: ModelHandle, /// Per-action response channels for searches that are still waiting on /// `GetRelevantFilesController`. - active_searches: HashMap>, + active_searches: HashMap, /// Cached repo roots derived during preprocessing so permission checks and execution can agree /// on which repository the action actually targets. root_repo_paths: HashMap, @@ -59,9 +68,7 @@ impl SearchCodebaseExecutor { let Some(result_tx) = me.active_searches.remove(action_id) else { return; }; - if let Err(e) = result_tx.send(result.clone()) { - log::warn!("Failed to send search codebase results to receiver {e:?}."); - } + Self::send_bounded_result(result_tx, result.clone(), ctx); } GetRelevantFilesControllerEvent::Success { result: GetRelevantFilesControllerResult::Locations(fragments), @@ -70,6 +77,8 @@ impl SearchCodebaseExecutor { let action_id = event.action_id().clone(); let locations = fragments .iter() + .sorted_by_key(|location| location.path()) + .take(MAX_SEARCH_FILES) .map(|location| location.into()) .collect_vec(); let current_working_directory = me @@ -84,13 +93,15 @@ impl SearchCodebaseExecutor { &locations, current_working_directory, shell, - None, - None, + Some(MAX_SEARCH_SCAN_BYTES), + Some(MAX_SEARCH_SCAN_BYTES * MAX_SEARCH_FILES), ) .await { Ok(result) => { - if !result.missing_files.is_empty() { + if result.file_contexts.is_empty() + && !result.missing_files.is_empty() + { let missing_files = result.missing_files.join(", "); SearchCodebaseResult::Failed { message: format!( @@ -110,15 +121,11 @@ impl SearchCodebaseExecutor { }, } }, - move |me, result, _| { + move |me, result, ctx| { let Some(result_tx) = me.active_searches.remove(&action_id) else { return; }; - if let Err(e) = result_tx.send(result) { - log::warn!( - "Failed to send search codebase results to receiver {e:?}." - ); - } + Self::send_bounded_result(result_tx, result, ctx); }, ); } @@ -126,7 +133,7 @@ impl SearchCodebaseExecutor { let Some(result_tx) = me.active_searches.remove(action_id) else { return; }; - if let Err(e) = result_tx.send(SearchCodebaseResult::Failed { + if let Err(e) = result_tx.sender.send(SearchCodebaseResult::Failed { message: "The search failed. Try another way to locate the relevant files." .to_owned(), reason: SearchCodebaseFailureReason::GetRelevantFilesError, @@ -146,6 +153,21 @@ impl SearchCodebaseExecutor { } } + fn send_bounded_result( + pending: PendingSearch, + result: SearchCodebaseResult, + ctx: &mut ModelContext, + ) { + ctx.spawn( + async move { (pending.sender, bound_search_result(result, &pending.query)) }, + |_, (sender, result), _| { + if let Err(error) = sender.send(result) { + log::debug!("Search result receiver was dropped: {error:?}"); + } + }, + ); + } + pub(super) fn should_autoexecute( &self, input: ExecuteActionInput, @@ -245,7 +267,13 @@ impl SearchCodebaseExecutor { }); let (result_tx, result_rx) = oneshot::channel(); - self.active_searches.insert(id.clone(), result_tx); + self.active_searches.insert( + id.clone(), + PendingSearch { + sender: result_tx, + query: query.clone(), + }, + ); match self .get_relevant_files_controller @@ -358,7 +386,13 @@ impl SearchCodebaseExecutor { }); let (result_tx, result_rx) = oneshot::channel(); - self.active_searches.insert(id.clone(), result_tx); + self.active_searches.insert( + id.clone(), + PendingSearch { + sender: result_tx, + query: query.clone(), + }, + ); // Start the actual search. match self diff --git a/app/src/ai/blocklist/action_model/execute/search_codebase_context.rs b/app/src/ai/blocklist/action_model/execute/search_codebase_context.rs new file mode 100644 index 00000000..56ce7c84 --- /dev/null +++ b/app/src/ai/blocklist/action_model/execute/search_codebase_context.rs @@ -0,0 +1,121 @@ +//! Bounded discovery excerpts. Explicit `read_files` calls provide full follow-up context. +use std::collections::HashSet; + +pub(super) use ai::agent::action_result::MAX_SEARCH_CODEBASE_FILES as MAX_SEARCH_FILES; + +use crate::ai::agent::{AnyFileContent, FileContext, SearchCodebaseResult}; +pub(super) const MAX_SEARCH_SCAN_BYTES: usize = 256_000; +const MAX_EXCERPT_BYTES: usize = 4_000; +const MAX_EXCERPT_LINES: usize = 40; + +pub(super) fn bound_search_result( + result: SearchCodebaseResult, + query: &str, +) -> SearchCodebaseResult { + let SearchCodebaseResult::Success { files } = result else { + return result; + }; + let terms: Vec<_> = query + .split(|character: char| !character.is_alphanumeric() && character != '_') + .filter(|term| term.len() >= 3) + .map(str::to_lowercase) + .filter(|term| { + !matches!( + term.as_str(), + "the" + | "and" + | "for" + | "with" + | "from" + | "that" + | "this" + | "find" + | "show" + | "all" + | "code" + | "files" + | "where" + | "what" + | "when" + | "which" + | "into" + ) + }) + .take(32) + .collect(); + let mut seen = HashSet::new(); + let files = files + .into_iter() + .filter(|file| seen.insert(file.file_name.clone())) + .filter_map(|file| excerpt(file, &terms)) + .take(MAX_SEARCH_FILES) + .collect(); + SearchCodebaseResult::Success { files } +} + +fn excerpt(mut file: FileContext, terms: &[String]) -> Option { + let AnyFileContent::StringContent(content) = &file.content else { + return None; + }; + // Also bound remote payloads before scoring. Never split a UTF-8 code point. + let mut scan_end = content.len().min(MAX_SEARCH_SCAN_BYTES); + while !content.is_char_boundary(scan_end) { + scan_end -= 1; + } + let lines: Vec<_> = content[..scan_end].split_inclusive('\n').collect(); + if lines.is_empty() { + return None; + } + let mut best_line = 0; + let mut best_score = 0; + for (index, line) in lines.iter().enumerate() { + let lower = line.to_lowercase(); + let score: usize = terms + .iter() + .filter(|term| lower.contains(term.as_str())) + .map(String::len) + .sum(); + if score > best_score { + best_score = score; + best_line = index; + } + } + let mut start = best_line; + let mut preceding_bytes = 0; + while start > best_line.saturating_sub(5) { + let length = lines[start - 1].len(); + if preceding_bytes + length > MAX_EXCERPT_BYTES / 3 { + break; + } + preceding_bytes += length; + start -= 1; + } + let mut text = String::new(); + let mut line_count: usize = 0; + for line in lines.iter().skip(start).take(MAX_EXCERPT_LINES) { + let remaining = MAX_EXCERPT_BYTES - text.len(); + if remaining == 0 { + break; + } + let mut end = line.len().min(remaining); + while !line.is_char_boundary(end) { + end -= 1; + } + if end == 0 { + break; + } + text.push_str(&line[..end]); + line_count += 1; + if end < line.len() { + break; + } + } + let first_line = file.line_range.as_ref().map_or(1, |range| range.start) + start; + file.line_range = Some(first_line..first_line + line_count.saturating_sub(1)); + file.content = AnyFileContent::StringContent(text); + Some(file) +} + +#[cfg(test)] +#[path = "search_codebase_context_tests.rs"] +mod tests; diff --git a/app/src/ai/blocklist/action_model/execute/search_codebase_context_tests.rs b/app/src/ai/blocklist/action_model/execute/search_codebase_context_tests.rs new file mode 100644 index 00000000..8a12b607 --- /dev/null +++ b/app/src/ai/blocklist/action_model/execute/search_codebase_context_tests.rs @@ -0,0 +1,68 @@ +use super::*; + +fn file(name: &str, text: String) -> FileContext { + FileContext::new( + name.to_string(), + AnyFileContent::StringContent(text), + None, + None, + ) +} + +#[test] +fn search_returns_bounded_unique_files_with_valid_utf8() { + let files = (0..20) + .flat_map(|index| { + let file = file(&format!("{index}.rs"), "€".repeat(20_000)); + [file.clone(), file] + }) + .collect(); + let SearchCodebaseResult::Success { files } = + bound_search_result(SearchCodebaseResult::Success { files }, "find models") + else { + panic!("expected success"); + }; + assert_eq!(files.len(), MAX_SEARCH_FILES); + let mut bytes = 0; + for file in files { + let AnyFileContent::StringContent(text) = file.content else { + panic!("expected text"); + }; + assert!(text.len() <= MAX_EXCERPT_BYTES); + assert_eq!(text.len() % 3, 0); + bytes += text.len(); + } + assert!(bytes <= MAX_SEARCH_FILES * MAX_EXCERPT_BYTES); +} + +#[test] +fn excerpt_finds_a_later_match_and_preserves_source_line_numbers() { + let text = format!( + "{}fn construct_chat_model() {{}}\n{}", + "// unrelated\n".repeat(200), + "// trailing\n".repeat(100) + ); + let mut input = file("models.rs", text); + input.line_range = Some(101..401); + let output = excerpt(input, &["construct_chat_model".into()]).unwrap(); + assert_eq!(output.line_range, Some(296..335)); + let AnyFileContent::StringContent(text) = output.content else { + panic!("expected text"); + }; + assert!(text.contains("construct_chat_model")); + assert_eq!(text.lines().count(), MAX_EXCERPT_LINES); +} + +#[test] +fn long_preceding_line_does_not_hide_the_relevant_match() { + let input = file( + "models.rs", + format!("{}\nfn target_model() {{}}\n", "x".repeat(10_000)), + ); + let output = excerpt(input, &["target_model".into()]).unwrap(); + assert_eq!(output.line_range, Some(2..2)); + let AnyFileContent::StringContent(text) = output.content else { + panic!("expected text"); + }; + assert!(text.contains("target_model")); +} diff --git a/app/src/ai/blocklist/action_model/execute/send_message.rs b/app/src/ai/blocklist/action_model/execute/send_message.rs index 66a14556..a5b22b9a 100644 --- a/app/src/ai/blocklist/action_model/execute/send_message.rs +++ b/app/src/ai/blocklist/action_model/execute/send_message.rs @@ -1,3 +1,4 @@ +use std::collections::HashMap; #[cfg(not(target_family = "wasm"))] use std::time::Duration; @@ -13,12 +14,13 @@ use warpui::r#async::Timer; use warpui::{AppContext, Entity, ModelContext, SingletonEntity}; use super::{ActionExecution, AnyActionExecution, ExecuteActionInput, PreprocessActionInput}; -use crate::ai::agent::conversation::AIConversationId; +use crate::ai::agent::conversation::{AIConversationId, ConversationStatus}; use crate::ai::agent::{ AIAgentAction, AIAgentActionResultType, AIAgentActionType, SendMessageToAgentResult, }; use crate::ai::ambient_agents::AmbientAgentTaskId; use crate::ai::blocklist::history_model::BlocklistAIHistoryModel; +use crate::ai::blocklist::orchestration_events::OrchestrationEventService; use crate::ai::blocklist::telemetry::{ BlocklistOrchestrationTelemetryEvent, TeamAgentCommunicationFailedEvent, TeamAgentCommunicationFailureReason, TeamAgentCommunicationKind, @@ -32,6 +34,7 @@ const SEND_AGENT_MESSAGE_TIMEOUT: Duration = Duration::from_secs(15); pub struct SendMessageToAgentExecutor { ambient_agent_task_id: Option, + local_recovery_attempts: HashMap, } #[derive(Debug, Clone, Copy, PartialEq, Eq)] @@ -121,6 +124,7 @@ impl SendMessageToAgentExecutor { pub fn new() -> Self { Self { ambient_agent_task_id: None, + local_recovery_attempts: HashMap::new(), } } @@ -159,6 +163,68 @@ impl SendMessageToAgentExecutor { let subject = subject.clone(); let message_body = message.clone(); + // Resolve only an actual child of this sender; local routing never trusts an arbitrary + // conversation ID as authority to resume another task. + let history = BlocklistAIHistoryModel::as_ref(ctx); + let local_children = history.child_conversations_of(conversation_id); + let local_child = if addresses.len() == 1 { + local_children.iter().find(|child| { + !child.is_remote_child() + && (child.id().to_string() == addresses[0] + || child.orchestration_agent_id().as_deref() == Some(addresses[0].as_str())) + }) + } else { + None + }; + if let Some(child) = local_child { + let child_id = child.id(); + let error = if message_body.trim().is_empty() { + Some("A corrective instruction or follow-up message is required.") + } else if matches!(child.status(), ConversationStatus::Cancelled) { + Some("The child was cancelled. Ask the user to resume it; do not override cancellation.") + } else if matches!(child.status(), ConversationStatus::Blocked { .. }) { + Some("The child is blocked on permission or user input. Resolve that block rather than restarting it.") + } else if matches!(child.status(), ConversationStatus::Error) + && self + .local_recovery_attempts + .get(&child_id) + .copied() + .unwrap_or(0) + >= 2 + { + Some("This child has exhausted two orchestrator recovery attempts. Report the blocker to the user.") + } else { + None + }; + if let Some(error) = error { + return ActionExecution::<()>::Sync(AIAgentActionResultType::SendMessageToAgent( + SendMessageToAgentResult::Error(error.to_string()), + )) + .into(); + } + if matches!(child.status(), ConversationStatus::Error) { + *self.local_recovery_attempts.entry(child_id).or_default() += 1; + } + let message_id = OrchestrationEventService::handle(ctx).update(ctx, |service, ctx| { + service.send_local_follow_up(conversation_id, child_id, message_body, ctx) + }); + return ActionExecution::<()>::Sync(AIAgentActionResultType::SendMessageToAgent( + SendMessageToAgentResult::Success { message_id }, + )) + .into(); + } + if history + .conversation(&conversation_id) + .is_some_and(|parent| parent.run_id().is_none()) + || addresses.iter().any(|address| { + history.all_live_conversations().iter().any(|(_, target)| { + !target.is_remote_child() && target.id().to_string() == *address + }) + }) + { + return ActionExecution::<()>::Sync(AIAgentActionResultType::SendMessageToAgent(SendMessageToAgentResult::Error("The target must be one existing child of this orchestrator. Use its agent_id from run_agents; do not message unrelated conversations.".to_string()))).into(); + } + let (sender_run_id, task_id, task_resolution) = sender_run_id_and_task_id_for_send(conversation_id, self.ambient_agent_task_id, ctx); let log_addresses = addresses.clone(); diff --git a/app/src/ai/blocklist/action_model/execute/start_agent.rs b/app/src/ai/blocklist/action_model/execute/start_agent.rs index 96107a9a..6cae7bb8 100644 --- a/app/src/ai/blocklist/action_model/execute/start_agent.rs +++ b/app/src/ai/blocklist/action_model/execute/start_agent.rs @@ -313,6 +313,11 @@ impl StartAgentExecutor { let Some(pending) = self.pending.remove(&request_id) else { return; }; + let error_msg = if matches!(pending.wait_policy, StartAgentWaitPolicy::Completion) { + format!("Child agent {child_conversation_id} failed: {error_msg}. For a recoverable error (never a user cancellation), use send_message_to_agent with this agent_id and a corrective instruction to continue the same child; do not create a duplicate.") + } else { + error_msg + }; let _ = pending.sender.try_send(StartAgentOutcome::Error(error_msg)); // Only startup acknowledgements may clean up a conversation that never // initialized. Direct-provider completion waits preserve the terminal diff --git a/app/src/ai/blocklist/controller.rs b/app/src/ai/blocklist/controller.rs index b1662372..d347c435 100644 --- a/app/src/ai/blocklist/controller.rs +++ b/app/src/ai/blocklist/controller.rs @@ -3948,6 +3948,11 @@ impl BlocklistAIController { }); } + if has_piggybacked_events && result.is_ok() { + OrchestrationEventService::handle(ctx).update(ctx, |svc, _| { + svc.confirm_local_messages(conversation_id); + }); + } self.pending_passive_follow_ups.remove(&conversation_id); } @@ -4102,8 +4107,13 @@ impl BlocklistAIController { let is_ready_status = matches!( conversation.status(), ConversationStatus::Success | ConversationStatus::WaitingForEvents, - ); - if !owns || has_active_stream || !is_ready_status { + ) || (matches!(conversation.status(), ConversationStatus::Error) + && OrchestrationEventService::as_ref(ctx).has_pending_local_message(conversation_id)); + // A direct-provider wait can yield to a queued successor. send_request_input + // checkpoints that successor and cancels the old generation before resuming. + let provider_wait = self.active_provider_runs.contains_key(&conversation_id) + && matches!(conversation.status(), ConversationStatus::WaitingForEvents); + if !owns || (has_active_stream && !provider_wait) || !is_ready_status { log::info!( "Pending events are not ready: conversation_id={conversation_id:?} owns_conversation={owns} has_active_stream={has_active_stream} status={:?}", conversation.status() @@ -4316,6 +4326,17 @@ impl BlocklistAIController { return; }; + // A provider wait owns a durable run. Terminalize that generation before + // queuing the message turn, just as for a same-conversation follow-up. + if self.active_provider_runs.contains_key(&conversation_id) { + self.cancel_active_provider_run( + conversation_id, + CancellationReason::FollowUpSubmitted { + is_for_same_conversation: true, + }, + ctx, + ); + } // The resume request supersedes any in-flight wait_for_events. self.action_model.update(ctx, |action_model, ctx| { action_model.cancel_wait_for_events_for_conversation(conversation_id, ctx); @@ -4349,6 +4370,10 @@ impl BlocklistAIController { OrchestrationEventService::handle(ctx).update(ctx, |svc, ctx| { svc.requeue_awaiting_events(conversation_id, ctx); }); + } else { + OrchestrationEventService::handle(ctx).update(ctx, |svc, _| { + svc.confirm_local_messages(conversation_id); + }); } } @@ -8372,6 +8397,9 @@ impl BlocklistAIController { }); self.maybe_refresh_ai_overages(ctx); self.start_next_queued_provider_run(conversation_id, ctx); + // Status notifications may arrive before the stream releases its slot. + // Retry queued child messages now that terminal cleanup is complete. + self.handle_pending_events_ready(conversation_id, ctx); } fn start_next_queued_provider_run( diff --git a/app/src/ai/blocklist/orchestration_events.rs b/app/src/ai/blocklist/orchestration_events.rs index e64a47a4..71cee50f 100644 --- a/app/src/ai/blocklist/orchestration_events.rs +++ b/app/src/ai/blocklist/orchestration_events.rs @@ -47,6 +47,12 @@ impl LifecycleEventDetailStage { /// Type-specific queued data. #[derive(Debug, Clone)] pub enum PendingEventDetail { + /// A locally authorized parent/child message; never sent through the hosted messaging API. + LocalMessage { + message_id: String, + subject: String, + message_body: String, + }, Message { message_id: String, addresses: Vec, @@ -90,6 +96,7 @@ pub struct OrchestrationEventService { pending_events: HashMap>, awaiting_server_echo_events: HashMap>, conversation_statuses: HashMap, + restarted_children: HashSet, } impl OrchestrationEventService { @@ -106,6 +113,7 @@ impl OrchestrationEventService { pending_events: HashMap::new(), awaiting_server_echo_events: HashMap::new(), conversation_statuses: HashMap::new(), + restarted_children: HashSet::new(), } } @@ -148,6 +156,7 @@ impl OrchestrationEventService { self.pending_events.remove(conversation_id); self.awaiting_server_echo_events.remove(conversation_id); self.conversation_statuses.remove(conversation_id); + self.restarted_children.remove(conversation_id); } _ => {} } @@ -184,8 +193,12 @@ impl OrchestrationEventService { conversation.status().clone() }; - self.conversation_statuses + let previous = self + .conversation_statuses .insert(conversation_id, current_status.clone()); + if !is_restored && previous.as_ref() != Some(¤t_status) { + self.report_restarted_child(conversation_id, previous.as_ref(), ¤t_status, ctx); + } let has_pending = self .pending_events .get(&conversation_id) @@ -196,16 +209,148 @@ impl OrchestrationEventService { // the stream finishes — either to `Success` or to // `WaitingForEvents` via a `wait_for_events` yield. if !is_restored - && matches!( + && (matches!( ¤t_status, ConversationStatus::Success | ConversationStatus::WaitingForEvents - ) + ) || (matches!(current_status, ConversationStatus::Error) + && self.has_pending_local_message(conversation_id))) && has_pending { ctx.emit(OrchestrationEventServiceEvent::EventsReady { conversation_id }); } } + pub fn has_pending_local_message(&self, conversation_id: AIConversationId) -> bool { + self.pending_events + .get(&conversation_id) + .is_some_and(|events| { + events + .iter() + .any(|event| matches!(event.detail, PendingEventDetail::LocalMessage { .. })) + }) + } + + /// Local providers commit the input directly and do not emit a hosted message echo. + pub fn confirm_local_messages(&mut self, conversation_id: AIConversationId) { + if let Some(events) = self.awaiting_server_echo_events.get_mut(&conversation_id) { + events.retain(|event| !matches!(event.detail, PendingEventDetail::LocalMessage { .. })); + if events.is_empty() { + self.awaiting_server_echo_events.remove(&conversation_id); + } + } + } + + pub fn send_local_follow_up( + &mut self, + parent: AIConversationId, + child: AIConversationId, + message: String, + ctx: &mut ModelContext, + ) -> String { + self.restarted_children.insert(child); + self.enqueue_local_message( + child, + parent, + "Follow-up from your orchestrator", + message, + ctx, + ) + } + + fn enqueue_local_message( + &mut self, + target: AIConversationId, + source: AIConversationId, + subject: &str, + message_body: String, + ctx: &mut ModelContext, + ) -> String { + let message_id = Uuid::new_v4().to_string(); + self.enqueue_event_batch( + target, + vec![PendingEvent { + event_id: message_id.clone(), + source_agent_id: source.to_string(), + attempt_count: 0, + detail: PendingEventDetail::LocalMessage { + message_id: message_id.clone(), + subject: subject.to_string(), + message_body, + }, + }], + ctx, + ); + message_id + } + + fn report_restarted_child( + &mut self, + child: AIConversationId, + previous: Option<&ConversationStatus>, + status: &ConversationStatus, + ctx: &mut ModelContext, + ) { + let history = BlocklistAIHistoryModel::as_ref(ctx); + let Some(conversation) = history.conversation(&child) else { + return; + }; + if conversation.is_remote_child() { + return; + } + let Some(parent) = conversation.parent_conversation_id() else { + return; + }; + let restarting = matches!( + previous, + Some( + ConversationStatus::Error + | ConversationStatus::Cancelled + | ConversationStatus::Success + ) + ) && matches!( + status, + ConversationStatus::InProgress | ConversationStatus::TransientError + ); + if restarting { + self.restarted_children.insert(child); + } + if !self.restarted_children.contains(&child) { + return; + } + let name = conversation.agent_name().unwrap_or("child agent"); + let message = if restarting { + format!("Child {name} ({child}) has restarted. Its earlier terminal result is no longer the final outcome. Keep tracking this child; do not spawn a replacement. A completion update will follow.") + } else if matches!(status, ConversationStatus::Success) { + let output = conversation + .all_exchanges() + .into_iter() + .rev() + .filter_map(|exchange| exchange.output_status.output()) + .map(|output| output.get().format_for_copy(None)) + .find(|text| !text.is_empty()) + .unwrap_or_default(); + let mut end = output.len().min(16_000); + while !output.is_char_boundary(end) { + end -= 1; + } + format!("Restarted child {name} ({child}) completed successfully. Incorporate this result into your task, even if an earlier attempt failed.\n{}{}", &output[..end], if end < output.len() { "\n[Output excerpt truncated; inspect the child conversation for more.]" } else { "" }) + } else if matches!( + status, + ConversationStatus::Error | ConversationStatus::Cancelled + ) { + format!("Restarted child {name} ({child}) ended with {status:?}: {}. Diagnose the failure before retrying; do not loop or override a user cancellation.", conversation.status_error_message().unwrap_or_default()) + } else { + return; + }; + if matches!( + status, + ConversationStatus::Success | ConversationStatus::Error | ConversationStatus::Cancelled + ) { + self.restarted_children.remove(&child); + } + self.enqueue_local_message(parent, child, "Child agent progress update", message, ctx); + } + fn enqueue_lifecycle_event( &mut self, target_conversation_id: AIConversationId, @@ -305,6 +450,20 @@ impl OrchestrationEventService { let mut server_bound_events = Vec::new(); for event in deliverable { match &event.detail { + PendingEventDetail::LocalMessage { + message_id, + subject, + message_body, + } => { + messages.push(ReceivedMessageInput { + message_id: message_id.clone(), + sender_agent_id: event.source_agent_id.clone(), + addresses: vec![conversation_id.to_string()], + subject: subject.clone(), + message_body: message_body.clone(), + }); + server_bound_events.push(event); + } PendingEventDetail::Message { message_id, addresses, @@ -625,7 +784,8 @@ fn did_event_round_trip_through_server( echoed_lifecycle_event_ids: &HashSet<&str>, ) -> bool { match &pending_event.detail { - PendingEventDetail::Message { message_id, .. } => { + PendingEventDetail::Message { message_id, .. } + | PendingEventDetail::LocalMessage { message_id, .. } => { echoed_message_ids.contains(message_id.as_str()) } PendingEventDetail::Lifecycle { event } => { diff --git a/app/src/ai/blocklist/orchestration_events_tests.rs b/app/src/ai/blocklist/orchestration_events_tests.rs index e9ffddfb..60c8cf62 100644 --- a/app/src/ai/blocklist/orchestration_events_tests.rs +++ b/app/src/ai/blocklist/orchestration_events_tests.rs @@ -355,3 +355,39 @@ fn test_has_pending_events_tracks_any_event_kind() { service.pending_events.remove(&conversation_id); assert!(!service.has_pending_events(conversation_id)); } + +#[test] +fn local_follow_up_is_delivered_once_and_acknowledged_without_server_echo() { + let conversation_id = AIConversationId::new(); + let mut service = OrchestrationEventService::new_without_subscriptions(); + service.pending_events.insert( + conversation_id, + vec![ + PendingEvent { + event_id: "local-1".to_string(), + source_agent_id: "parent".to_string(), + attempt_count: 0, + detail: PendingEventDetail::LocalMessage { + message_id: "local-1".to_string(), + subject: "Correction".to_string(), + message_body: "Continue the existing task.".to_string(), + }, + }, + message_pending_event("hosted-1"), + ], + ); + assert!(service.has_pending_local_message(conversation_id)); + let inputs = service.drain_and_convert_events(conversation_id); + let AIAgentInput::MessagesReceivedFromAgents { messages } = &inputs[0] else { + panic!("Expected child follow-up input"); + }; + assert_eq!(messages[0].sender_agent_id, "parent"); + assert_eq!(messages[0].message_body, "Continue the existing task."); + assert_eq!(messages[0].addresses, vec![conversation_id.to_string()]); + assert!(service.drain_and_convert_events(conversation_id).is_empty()); + service.confirm_local_messages(conversation_id); + let awaiting = &service.awaiting_server_echo_events[&conversation_id]; + assert_eq!(awaiting.len(), 1); + assert_eq!(awaiting[0].event_id, "hosted-1"); + assert!(!service.has_pending_local_message(conversation_id)); +} diff --git a/app/src/ai/get_relevant_files/controller.rs b/app/src/ai/get_relevant_files/controller.rs index befa8cd1..8fa4e4f2 100644 --- a/app/src/ai/get_relevant_files/controller.rs +++ b/app/src/ai/get_relevant_files/controller.rs @@ -21,7 +21,7 @@ use instant::Instant; #[cfg(not(target_family = "wasm"))] use crate::ai::agent::SearchCodebaseFailureReason; -use crate::ai::agent::{AIAgentActionId, SearchCodebaseResult}; +use crate::ai::agent::{AIAgentActionId, SearchCodebaseResult, MAX_SEARCH_CODEBASE_FILES}; use crate::ai::blocklist::SessionContext; #[cfg(any(not(feature = "local_fs"), target_family = "wasm"))] use crate::ai::get_relevant_files::api::{FileContext as FileContextRequest, GetRelevantFiles}; @@ -362,6 +362,7 @@ impl GetRelevantFilesController { .search(directory, &query, partial_path_segments.map(Vec::as_slice)) .map_err(|error| LocalIndexUnavailable(error.to_string()))? .into_iter() + .take(MAX_SEARCH_CODEBASE_FILES) .map(|hit| CodeContextLocation::WholeFile(hit.path)) .collect(); let file_paths = locations diff --git a/app/src/ai/openai/response_translator.rs b/app/src/ai/openai/response_translator.rs index f074956b..18f491d4 100644 --- a/app/src/ai/openai/response_translator.rs +++ b/app/src/ai/openai/response_translator.rs @@ -686,6 +686,7 @@ const KNOWN_TOOLS: &[&str] = &[ "edit_documents", "run_agents", "wait_for_events", + "send_message_to_agent", "start_agent", "ask_user_question", "read_skill", diff --git a/app/src/ai/provider/request_translator.rs b/app/src/ai/provider/request_translator.rs index 558f3f49..ba4aec33 100644 --- a/app/src/ai/provider/request_translator.rs +++ b/app/src/ai/provider/request_translator.rs @@ -1560,7 +1560,7 @@ pub fn extract_system_prompt( let has_tool = |name: &str| tool_names.iter().any(|tool_name| tool_name == name); if has_tool("search_codebase") { prompt.push_str( - "- For source-code discovery, semantic questions, or finding an unfamiliar implementation, MUST use `search_codebase` first. Never use `grep` to discover or search source code when `search_codebase` is available; reserve `grep` for exact known text in non-code files. Use `file_glob` only to locate filenames and `read_files` for focused follow-up context.\n", + "- For source-code discovery, semantic questions, or finding an unfamiliar implementation, MUST use `search_codebase` first. Never use `grep` to discover or search source code when `search_codebase` is available; reserve `grep` for exact known text in non-code files. Use `file_glob` only to locate filenames and `read_files` for focused follow-up context. Start with one concise search query using distinctive terms; avoid parallel rephrasings over the same scope. Search returns at most eight bounded excerpts, not an exhaustive audit. Read the relevant files next and only broaden or narrow the search when the results leave a specific gap.\n", ); } if [ @@ -1701,7 +1701,9 @@ pub(crate) fn tool_name_is_supported(name: &str, supported: &HashSet has(ToolType::ReadDocuments), "create_plan" | "create_notebook" => has(ToolType::CreateDocuments), "edit_plan" | "edit_notebook" => has(ToolType::EditDocuments), - "run_agents" | "start_agent" => has(ToolType::Subagent) || has(ToolType::StartAgentV2), + "run_agents" | "start_agent" | "send_message_to_agent" | "wait_for_events" => { + has(ToolType::Subagent) || has(ToolType::StartAgentV2) + } "ask_user_question" => has(ToolType::AskUserQuestion), "read_skill" => has(ToolType::ReadSkill), "fetch_conversation" => has(ToolType::FetchConversation), @@ -2030,9 +2032,15 @@ pub fn default_tool_definitions() -> Vec { "required": ["name", "prompt"] }), }, - // send_message_to_agent removed — child agents in local/Bedrock mode - // run autonomously and cannot receive messages. Exposing this tool - // causes the model to poll in a loop. + ToolDefinition { + name: "send_message_to_agent".to_string(), + description: "Continue an existing child agent with a corrective instruction or follow-up. Use the child agent_id returned by run_agents, including IDs in failure messages. A failed local child resumes with its existing context; at most two recovery attempts are allowed. Diagnose the error first, avoid blind retries, never override cancellation or permission blocks, and never spawn a duplicate replacement. Delivery is not completion: use wait_for_events for the child's updated result.".to_string(), + input_schema: serde_json::json!({ + "type": "object", "properties": { + "agent_id": {"type": "string"}, "message": {"type": "string"} + }, "required": ["agent_id", "message"] + }), + }, ToolDefinition { name: "ask_user_question".to_string(), description: "Ask the user a question when you need clarification or a decision. Present clear options when possible. Use sparingly — prefer making reasonable assumptions.".to_string(), @@ -2276,6 +2284,19 @@ fn extract_tool_result_content(result: &api::request::input::ToolCallResult) -> None => ("MCP tool completed.".to_string(), false), } } + api::request::input::tool_call_result::Result::SendMessageToAgent(result) => { + match &result.result { + Some(api::send_message_to_agent_result::Result::Success(success)) => ( + format!("Follow-up queued (message {}). Delivery is not completion; use wait_for_events for the child's updated result.", success.message_id), + false, + ), + Some(api::send_message_to_agent_result::Result::Error(error)) => ( + format!("Agent message failed: {}", error.message), + true, + ), + None => ("Agent message returned no delivery result.".to_string(), true), + } + } api::request::input::tool_call_result::Result::SearchCodebase(search_result) => { match &search_result.result { Some(api::search_codebase_result::Result::Success(success)) => ( diff --git a/app/src/ai/provider/request_translator_tests.rs b/app/src/ai/provider/request_translator_tests.rs index e3796ce4..1134289a 100644 --- a/app/src/ai/provider/request_translator_tests.rs +++ b/app/src/ai/provider/request_translator_tests.rs @@ -101,7 +101,9 @@ fn advertised_tools_follow_client_capabilities_and_include_local_subagents() { "run_shell_command", "read_files", "run_agents", + "wait_for_events", "start_agent", + "send_message_to_agent", "recall_tool_history" ] ); @@ -680,3 +682,26 @@ fn injected_user_query_persists_images_for_session_restore() { if data == b"\x89PNG\r\n\x1a\npersisted" && mime_type == "image/png" )); } + +#[test] +fn leaf_capabilities_do_not_advertise_child_recovery_or_delegation() { + let request = api::Request { + settings: Some(api::request::Settings { + supported_tools: vec![api::ToolType::ReadFiles.into()], + ..Default::default() + }), + ..Default::default() + }; + let names = extract_tools(&request) + .into_iter() + .map(|tool| tool.name) + .collect::>(); + for name in [ + "run_agents", + "start_agent", + "send_message_to_agent", + "wait_for_events", + ] { + assert!(!names.iter().any(|advertised| advertised == name)); + } +} diff --git a/app/src/ai/provider/response_translator.rs b/app/src/ai/provider/response_translator.rs index 92521ce3..9c9d561d 100644 --- a/app/src/ai/provider/response_translator.rs +++ b/app/src/ai/provider/response_translator.rs @@ -1619,6 +1619,7 @@ const KNOWN_TOOLS: &[&str] = &[ "edit_documents", "run_agents", "wait_for_events", + "send_message_to_agent", "start_agent", "ask_user_question", "read_skill", diff --git a/app/src/ai/remote_logging.rs b/app/src/ai/remote_logging.rs index 4fb30521..1cc888a9 100644 --- a/app/src/ai/remote_logging.rs +++ b/app/src/ai/remote_logging.rs @@ -4,6 +4,7 @@ //! exclusively by local settings and should only receive operational metadata: //! provider/model IDs, lifecycle states, counts, timings, and sanitized errors. +use std::sync::LazyLock; use std::time::Duration; #[cfg(all(feature = "remote_logs_local_file", not(target_family = "wasm")))] use std::{ @@ -25,6 +26,13 @@ use crate::AISettings; const DEFAULT_LOGS_PATH: &str = "/api/logs"; const REMOTE_LOG_SERVICE: &str = "galaxy-ai"; const REMOTE_LOG_TIMEOUT: Duration = Duration::from_secs(5); +// Authentication and endpoint selection remain scoped to each request. +static REMOTE_LOG_CLIENT: LazyLock> = LazyLock::new(|| { + reqwest::Client::builder() + .timeout(REMOTE_LOG_TIMEOUT) + .build() + .map_err(|error| format!("could not create HTTP client: {error}")) +}); const MAX_ERROR_CHARS: usize = 500; #[derive(Clone, Copy, Debug, PartialEq, Eq)] @@ -225,10 +233,7 @@ fn app_context() -> Value { } async fn send_remote_log(config: RemoteLogConfig, payload: RemoteLogPayload) -> Result<(), String> { - let client = reqwest::Client::builder() - .timeout(REMOTE_LOG_TIMEOUT) - .build() - .map_err(|error| format!("could not create HTTP client: {error}"))?; + let client = REMOTE_LOG_CLIENT.as_ref().map_err(Clone::clone)?; let response = client .post(&config.endpoint) diff --git a/app/src/ai/runtime/rig_request.rs b/app/src/ai/runtime/rig_request.rs index 0dadef9f..f49435b7 100644 --- a/app/src/ai/runtime/rig_request.rs +++ b/app/src/ai/runtime/rig_request.rs @@ -924,7 +924,7 @@ fn build_system_prompt( let has_search_codebase = tools.iter().any(|tool| tool.name == "search_codebase"); if has_search_codebase { prompt.push_str( - "For source-code discovery, semantic questions, or finding an unfamiliar implementation, MUST use `search_codebase` first. Never use `grep` to discover or search source code when `search_codebase` is available; reserve `grep` for exact known text in non-code files. Use `file_glob` only to locate filenames and `read_files` for focused follow-up context.\n", + "For source-code discovery, semantic questions, or finding an unfamiliar implementation, MUST use `search_codebase` first. Never use `grep` to discover or search source code when `search_codebase` is available; reserve `grep` for exact known text in non-code files. Use `file_glob` only to locate filenames and `read_files` for focused follow-up context. Start with one concise search query using distinctive terms; avoid parallel rephrasings over the same scope. Search returns at most eight bounded excerpts, not an exhaustive audit. Read the relevant files next and only broaden or narrow the search when the results leave a specific gap.\n", ); } if tools.iter().any(|tool| tool.name == "run_shell_command") { diff --git a/app/src/pane_group/mod.rs b/app/src/pane_group/mod.rs index 2b023eb3..f91c1181 100644 --- a/app/src/pane_group/mod.rs +++ b/app/src/pane_group/mod.rs @@ -6949,6 +6949,25 @@ impl PaneGroup { None } + /// Close only the child presentation; retain its live view and history for later inspection. + pub fn hide_completed_child_agent( + &mut self, + conversation_id: AIConversationId, + ctx: &mut ViewContext, + ) { + let Some(pane_id) = self.child_agent_panes.get(&conversation_id).copied() else { + return; + }; + if !self.panes.is_pane_in_tree(pane_id) + && self.panes.original_pane_for_replacement(pane_id).is_none() + { + return; + } + if let Some(pane_content) = self.take_child_agent_pane_for_split_off(conversation_id, ctx) { + self.re_adopt_child_agent_pane(pane_content, conversation_id, ctx); + } + } + /// Make the pane that owns `conversation_id` the visible one in the /// focused pane's slot via temporary replacement. The previous occupant /// is restored on revert (back-button, ESC, pill-click, close, split-off). diff --git a/app/src/pane_group/mod_tests.rs b/app/src/pane_group/mod_tests.rs index 1e0f6458..27e0fe24 100644 --- a/app/src/pane_group/mod_tests.rs +++ b/app/src/pane_group/mod_tests.rs @@ -196,6 +196,7 @@ fn initialize_app(app: &mut App) { app.add_singleton_model(voice_input::VoiceInput::new); #[cfg(feature = "local_fs")] app.add_singleton_model(RepoMetadataModel::new); + app.add_singleton_model(|_| ProjectContextModel::default()); #[cfg(all(feature = "local_fs", not(target_family = "wasm")))] app.add_singleton_model(ai::index::local_project_index::LocalProjectIndexManager::new); app.add_singleton_model(SkillManager::new); @@ -209,7 +210,6 @@ fn initialize_app(app: &mut App) { CodebaseIndexManager::new_for_test(ServerApiProvider::as_ref(ctx).get(), ctx) }); app.add_singleton_model(|ctx| PersistedWorkspace::new(vec![], HashMap::new(), None, ctx)); - app.add_singleton_model(|_| ProjectContextModel::default()); app.add_singleton_model(|ctx| crate::ai::agent_tips::AITipModel::new_for_agent_tips(ctx)); app.add_singleton_model(|_| RestoredAgentConversations::new(vec![])); app.add_singleton_model(OneTimeModalModel::new); @@ -3388,3 +3388,63 @@ fn decide_remote_child_hydration_empty_token_falls_back() { ); } } + +#[test] +fn test_auto_hiding_completed_child_preserves_parent_focus_and_reopen() { + App::test((), |mut app| async move { + initialize_app(&mut app); + let pane_group = mock_pane_group(&mut app, Default::default()); + + pane_group.update(&mut app, |panes, ctx| { + let parent_pane_id = get_newly_created_pane_id(panes, &[]); + let parent_conversation_id = start_parent_conversation(panes, parent_pane_id, ctx); + let child = create_hidden_child_agent_conversation( + panes, + HiddenChildAgentConversationRequest { + parent_pane_id, + name: "Agent 1".to_string(), + parent_conversation_id, + orchestration_harness: None, + env_vars: HashMap::new(), + task_context: None, + is_shared_session_creator: IsSharedSessionCreator::No, + }, + ctx, + ) + .expect("fresh hidden child conversation should be created"); + let child_pane_id = panes + .child_agent_panes + .get(&child.conversation_id) + .copied() + .expect("fresh hidden child pane should be tracked"); + + assert_eq!( + panes.unhide_child_agent_pane_for_split_off(child.conversation_id, ctx), + Some(child_pane_id) + ); + assert!(panes.panes.is_pane_in_tree(child_pane_id)); + + panes.hide_completed_child_agent(child.conversation_id, ctx); + + assert_eq!(panes.focused_pane_id(ctx), parent_pane_id); + assert!(panes.has_pane_id(child_pane_id)); + assert_eq!( + panes.child_agent_panes.get(&child.conversation_id), + Some(&child_pane_id) + ); + assert!(!panes.panes.is_pane_in_tree(child_pane_id)); + + assert_eq!( + panes.unhide_child_agent_pane_for_split_off(child.conversation_id, ctx), + Some(child_pane_id) + ); + assert_eq!(panes.focused_pane_id(ctx), child_pane_id); + assert!(panes.panes.is_pane_in_tree(child_pane_id)); + assert!(panes + .panes + .pane_ids() + .into_iter() + .all(|pane_id| panes.has_pane_id(pane_id))); + }); + }); +} diff --git a/app/src/workspace/view.rs b/app/src/workspace/view.rs index f67ec8f1..874f92e6 100644 --- a/app/src/workspace/view.rs +++ b/app/src/workspace/view.rs @@ -3580,6 +3580,23 @@ impl Workspace { event: &BlocklistAIHistoryEvent, ctx: &mut ViewContext, ) { + if let BlocklistAIHistoryEvent::UpdatedConversationStatus { + conversation_id, + update, + .. + } = event + { + if !matches!( + update, + crate::ai::blocklist::history_model::ConversationStatusUpdate::Restored + ) { + let conversation_id = *conversation_id; + // Defer pane movement until completion subscribers have captured the result. + ctx.spawn(async {}, move |workspace, (), ctx| { + workspace.close_completed_child_views(conversation_id, ctx); + }); + } + } #[cfg(target_family = "wasm")] if self .current_workspace_state @@ -3624,6 +3641,58 @@ impl Workspace { } } + fn close_completed_child_views( + &mut self, + conversation_id: AIConversationId, + ctx: &mut ViewContext, + ) { + let history = BlocklistAIHistoryModel::as_ref(ctx); + let Some(conversation) = history.conversation(&conversation_id) else { + return; + }; + if conversation.parent_conversation_id().is_none() + || !matches!( + conversation.status(), + crate::ai::agent::conversation::ConversationStatus::Success + ) + { + return; + } + let Some(owner) = history.terminal_surface_id_for_conversation(&conversation_id) else { + return; + }; + // Do not close a pane the user has repurposed for a different conversation. + if history + .active_conversation(owner) + .is_none_or(|active| active.id() != conversation_id) + { + return; + } + let detached_tabs: Vec<_> = self + .tabs + .iter() + .enumerate() + .filter_map(|(index, tab)| { + let group = tab.pane_group.as_ref(ctx); + group + .child_agent_origin() + .is_some_and(|origin| origin.conversation_id == conversation_id) + .then_some(index) + }) + .collect(); + for index in detached_tabs.into_iter().rev() { + if self.tabs.len() > 1 && self.try_re_adopt_split_off_child_agent_tab(index, ctx) { + self.remove_tab(index, false, false, ctx); + } + } + let groups: Vec<_> = self.tabs.iter().map(|tab| tab.pane_group.clone()).collect(); + for group in groups { + group.update(ctx, |group, ctx| { + group.hide_completed_child_agent(conversation_id, ctx) + }); + } + } + fn workspace_contains_terminal_view( &self, terminal_view_id: EntityId, diff --git a/crates/ai/src/agent/action_result/convert.rs b/crates/ai/src/agent/action_result/convert.rs index f6f7de8a..5d62abf2 100644 --- a/crates/ai/src/agent/action_result/convert.rs +++ b/crates/ai/src/agent/action_result/convert.rs @@ -306,6 +306,11 @@ impl TryFrom for api::request::input::tool_call_result::Re files: files .into_iter() .flat_map(Into::>::into) + .map(|mut file| { + let range = file.line_range.as_ref().map(|range| format!(" (lines {}-{})", range.start, range.end)).unwrap_or_default(); + file.content = format!("Bounded search excerpt{range}; more matches/content may be omitted. Use read_files for focused follow-up.\n{}", file.content); + file + }) .collect(), }, )), diff --git a/crates/ai/src/agent/action_result/mod.rs b/crates/ai/src/agent/action_result/mod.rs index 2e7d8c94..feeaf30b 100644 --- a/crates/ai/src/agent/action_result/mod.rs +++ b/crates/ai/src/agent/action_result/mod.rs @@ -878,6 +878,9 @@ pub enum SearchCodebaseFailureReason { MissingCurrentWorkingDirectory, } +/// Maximum number of files returned by a codebase discovery search. +pub const MAX_SEARCH_CODEBASE_FILES: usize = 8; + #[derive(Debug, Clone, Eq, PartialEq)] pub enum SearchCodebaseResult { Success { diff --git a/crates/ai/src/index/file_outline/mod.rs b/crates/ai/src/index/file_outline/mod.rs index 39a6380e..5aafc178 100644 --- a/crates/ai/src/index/file_outline/mod.rs +++ b/crates/ai/src/index/file_outline/mod.rs @@ -4,6 +4,7 @@ cfg_if::cfg_if! { if #[cfg(not(target_arch = "wasm32"))] { mod native; pub use native::build_outline; + pub(crate) use native::parse_file_outline_content; } } diff --git a/crates/ai/src/index/file_outline/native.rs b/crates/ai/src/index/file_outline/native.rs index 630b69fe..52c20ac5 100644 --- a/crates/ai/src/index/file_outline/native.rs +++ b/crates/ai/src/index/file_outline/native.rs @@ -232,19 +232,26 @@ fn parse_file_outline(path: &Path) -> anyhow::Result { if !is_file_parsable(path)? { return Err(anyhow!("File exceeds max file size limit for parsing")); } + let content = fs::read_to_string(path)?; + parse_file_outline_content(path, &content) +} + +pub(crate) fn parse_file_outline_content( + path: &Path, + content: &str, +) -> anyhow::Result { let standardized_path = StandardizedPath::try_from_local(path)?; let Some(language) = languages::language_by_filename(&standardized_path) else { return Err(anyhow!("Language unsupported for file {:?}", path)); }; - let content = fs::read_to_string(path)?; let mut parser = Parser::new(); parser.set_language(&language.grammar)?; - let Some(tree) = parser.parse(&content, None) else { + let Some(tree) = parser.parse(content, None) else { return Err(anyhow!("Couldn't parse AST")); }; let symbols = language.symbols_query.as_ref().map(|query| { - get_symbols(query, &tree, &content) + get_symbols(query, &tree, content) .into_iter() .map(|(fn_name, type_prefix, comments, line_number)| Symbol { name: fn_name.to_owned(), @@ -285,7 +292,7 @@ fn parse_file_outline(path: &Path) -> anyhow::Result { fn get_symbols<'a>( query: &'a Query, tree: &Tree, - file_content: &'a String, + file_content: &'a str, ) -> Vec<(&'a str, Option<&'a str>, Vec<&'a str>, usize)> { struct PendingComment<'a> { lines: Vec<&'a str>, diff --git a/crates/ai/src/index/local_project_index/mod.rs b/crates/ai/src/index/local_project_index/mod.rs index 4f0f34b9..073fed6e 100644 --- a/crates/ai/src/index/local_project_index/mod.rs +++ b/crates/ai/src/index/local_project_index/mod.rs @@ -7,6 +7,7 @@ use std::collections::{HashMap, HashSet}; use std::fs::{self, File, OpenOptions}; +use std::io::Read; use std::path::{Path, PathBuf}; use std::time::Duration; @@ -16,13 +17,15 @@ use galaxy_util::standardized_path::StandardizedPath; use galaxyui_core::r#async::Timer; use galaxyui_core::{Entity, ModelContext, SingletonEntity}; use itertools::Itertools; -use repo_metadata::{RepoMetadataEvent, RepositoryIdentifier}; +use repo_metadata::entry::{BudgetExceededBehavior, IgnoredPathStrategy}; +use repo_metadata::file_tree_update::RepoNodeMetadata; +use repo_metadata::{Entry, MetadataUpdateType, RepoMetadataEvent, RepositoryIdentifier}; use serde::{Deserialize, Serialize}; use sha2::{Digest, Sha256}; use warp_search_core::define_search_schema; use warp_search_core::searcher::{SimpleFullTextSearcher, DEFAULT_MEMORY_BUDGET}; -use crate::index::build_outline; +use crate::index::file_outline::parse_file_outline_content; const INDEX_SCHEMA_VERSION: u32 = 1; const MAX_INDEX_FILES: usize = 5_000; @@ -31,6 +34,7 @@ const MAX_INDEXED_BODY_BYTES: usize = 256_000; const INDEX_DIRECTORY_NAME: &str = "local_project_indices"; const CURRENT_FILE_NAME: &str = "CURRENT"; const METADATA_FILE_NAME: &str = "metadata.json"; +const UPDATE_DEBOUNCE: Duration = Duration::from_millis(300); const INDEX_LOCK_RETRY_INTERVAL: Duration = Duration::from_millis(50); // Field weights intentionally prioritize symbols and paths over implementation text. @@ -108,10 +112,18 @@ impl LocalProjectIndex { } } +/// Paths are re-read at execution time, so repeated saves and rename sequences coalesce. +#[derive(Default)] +struct PendingUpdate { + full_rebuild: bool, + paths: HashSet, +} + pub struct LocalProjectIndexManager { indices: HashMap, statuses: HashMap, - pending_rebuilds: HashSet, + pending_updates: HashMap, + scheduled_updates: HashSet, rebuild_epochs: HashMap, storage_root: PathBuf, } @@ -125,15 +137,13 @@ impl SingletonEntity for LocalProjectIndexManager {} impl LocalProjectIndexManager { pub fn new(ctx: &mut ModelContext) -> Self { let repo_metadata = repo_metadata::RepoMetadataModel::handle(ctx); - let manager = Self::new_at(state_dir().join(INDEX_DIRECTORY_NAME), ctx); + let mut manager = Self::new_at(state_dir().join(INDEX_DIRECTORY_NAME), ctx); ctx.subscribe_to_model(&repo_metadata, |manager, _, event, ctx| { manager.handle_repo_metadata_event(event, ctx); }); - // Re-register restored roots with repository metadata so filesystem updates continue to - // refresh the local index after an app restart. RepositoryUpdated is intentionally ignored - // below; the restored generation is already usable and only file-tree updates need a - // rebuild. + // Re-register watchers and reconcile changes made while the app was closed. The restored + // snapshot remains searchable during this one-time rebuild. let restored_roots: Vec = manager.statuses.keys().cloned().collect(); for root_path in restored_roots { let Ok(standardized_root) = StandardizedPath::from_local_canonicalized(&root_path) @@ -143,6 +153,7 @@ impl LocalProjectIndexManager { let _ = repo_metadata.update(ctx, |model, ctx| { model.index_local_directory_path(&standardized_root, ctx) }); + manager.start_rebuild(root_path, ctx); } manager } @@ -152,7 +163,8 @@ impl LocalProjectIndexManager { let mut manager = Self { indices: HashMap::new(), statuses: HashMap::new(), - pending_rebuilds: HashSet::new(), + pending_updates: HashMap::new(), + scheduled_updates: HashSet::new(), rebuild_epochs: HashMap::new(), storage_root, }; @@ -179,7 +191,10 @@ impl LocalProjectIndexManager { self.statuses.get(&root_path), Some(LocalIndexStatus::Indexing) ) { - self.pending_rebuilds.insert(root_path); + self.pending_updates + .entry(root_path) + .or_default() + .full_rebuild = true; return Ok(()); } @@ -196,6 +211,47 @@ impl LocalProjectIndexManager { } fn start_rebuild(&mut self, root_path: PathBuf, ctx: &mut ModelContext) { + self.start_update( + root_path, + PendingUpdate { + full_rebuild: true, + paths: HashSet::new(), + }, + ctx, + ); + } + + fn schedule_update(&mut self, root_path: PathBuf, ctx: &mut ModelContext) { + if matches!( + self.statuses.get(&root_path), + Some(LocalIndexStatus::Indexing) + ) || !self.pending_updates.contains_key(&root_path) + || !self.scheduled_updates.insert(root_path.clone()) + { + return; + } + let epoch = self.rebuild_epochs.get(&root_path).copied(); + ctx.spawn( + async { Timer::after(UPDATE_DEBOUNCE).await }, + move |manager, _, ctx| { + if manager.rebuild_epochs.get(&root_path).copied() != epoch { + return; + } + manager.scheduled_updates.remove(&root_path); + if let Some(update) = manager.pending_updates.remove(&root_path) { + manager.start_update(root_path, update, ctx); + } + }, + ); + } + + fn start_update( + &mut self, + root_path: PathBuf, + update: PendingUpdate, + ctx: &mut ModelContext, + ) { + self.scheduled_updates.remove(&root_path); let rebuild_epoch = { let epoch = self.rebuild_epochs.entry(root_path.clone()).or_default(); *epoch += 1; @@ -210,12 +266,34 @@ impl LocalProjectIndexManager { let storage_root = self.storage_root.clone(); ctx.spawn( - async move { build_persisted_index(root_path, storage_root).await }, + async move { + if update.full_rebuild { + build_persisted_index(root_path, storage_root).await + } else { + match update_persisted_index( + root_path.clone(), + storage_root.clone(), + update.paths, + ) + .await + { + Ok(result) => Ok(result), + Err(error) => { + log::warn!( + "Incremental local index update failed; rebuilding: {error:#}" + ); + build_persisted_index(root_path, storage_root).await + } + } + } + }, move |manager, result, ctx| match result { Ok((root_path, built_index, file_count)) => { if manager.rebuild_epochs.get(&root_path).copied() != Some(rebuild_epoch) { drop(built_index.searcher); - let _ = fs::remove_dir_all(&built_index.generation_directory); + if built_index.new_generation { + let _ = fs::remove_dir_all(&built_index.generation_directory); + } return; } if let Err(error) = publish_generation( @@ -241,12 +319,10 @@ impl LocalProjectIndexManager { root_path: root_path.clone(), status, }); - let should_rebuild = manager.pending_rebuilds.remove(&root_path); + cleanup_old_generations(&manager.storage_root, &root_path); drop(built_index.repository_lock); - if should_rebuild { - manager.start_rebuild(root_path, ctx); - } + manager.schedule_update(root_path, ctx); } Err((root_path, error)) => { if manager.rebuild_epochs.get(&root_path).copied() != Some(rebuild_epoch) { @@ -289,10 +365,7 @@ impl LocalProjectIndexManager { status, }); } - let should_rebuild = self.pending_rebuilds.remove(&root_path); - if should_rebuild { - self.start_rebuild(root_path, ctx); - } + self.schedule_update(root_path, ctx); } /// Returns the most specific indexed root containing `path`. @@ -362,15 +435,65 @@ impl LocalProjectIndexManager { event: &RepoMetadataEvent, ctx: &mut ModelContext, ) { - let local_root = match event { + match event { RepoMetadataEvent::FileTreeEntryUpdated { id: RepositoryIdentifier::Local(path), - .. - } => path.to_local_path(), + update_type, + } => { + let Some(root) = path.to_local_path() else { + return; + }; + let root = dunce::canonicalize(&root).unwrap_or(root); + if !self.statuses.contains_key(&root) { + return; + } + let pending = self.pending_updates.entry(root.clone()).or_default(); + match update_type { + MetadataUpdateType::FullReplace => pending.full_rebuild = true, + MetadataUpdateType::IncrementalUpdate(update) if !pending.full_rebuild => { + pending.paths.extend( + update + .remove_entries + .iter() + .filter_map(StandardizedPath::to_local_path), + ); + for entry in &update.update_entries { + for node in &entry.subtree_metadata { + let path = match node { + RepoNodeMetadata::File(file) => &file.path, + RepoNodeMetadata::Directory(directory) => &directory.path, + }; + if let Some(path) = path.to_local_path() { + pending.paths.insert(path); + } + } + } + pending.paths.retain(|path| { + path.starts_with(&root) + && !path.components().any(|part| part.as_os_str() == ".git") + }); + if pending.paths.len() > MAX_INDEX_FILES + || pending.paths.iter().any(|path| { + path.file_name().is_some_and(|name| name == ".gitignore") + }) + { + pending.full_rebuild = true; + } + } + MetadataUpdateType::IncrementalUpdate(_) => {} + } + if pending.full_rebuild { + pending.paths.clear(); + } + if !pending.full_rebuild && pending.paths.is_empty() { + self.pending_updates.remove(&root); + return; + } + self.schedule_update(root, ctx); + } // FileTreeUpdated is emitted before repository mutations are applied. The - // post-application FileTreeEntryUpdated event below is the rebuild trigger; using - // both would start duplicate full rebuilds for every watcher batch. - RepoMetadataEvent::FileTreeUpdated { .. } => None, + // post-application delta is the update trigger; using both duplicates each batch. + RepoMetadataEvent::FileTreeUpdated { .. } => (), RepoMetadataEvent::RepositoryRemoved { id: RepositoryIdentifier::Local(path), } => { @@ -380,7 +503,6 @@ impl LocalProjectIndexManager { self.remove_index(&root_path, ctx); ctx.emit(LocalProjectIndexEvent::IndexRemoved { root_path }); } - return; } RepoMetadataEvent::RepositoryUpdated { .. } | RepoMetadataEvent::RepositoryRemoved { @@ -392,22 +514,7 @@ impl LocalProjectIndexManager { } | RepoMetadataEvent::UpdatingRepositoryFailed { .. } | RepoMetadataEvent::StandingQueryResultsUpdated { .. } - | RepoMetadataEvent::IncrementalUpdateReady { .. } => None, - }; - - let Some(local_root) = local_root else { - return; - }; - let canonical_root = dunce::canonicalize(&local_root).unwrap_or(local_root); - if self.statuses.contains_key(&canonical_root) { - if matches!( - self.statuses.get(&canonical_root), - Some(LocalIndexStatus::Indexing) - ) { - self.pending_rebuilds.insert(canonical_root); - } else { - self.start_rebuild(canonical_root, ctx); - } + | RepoMetadataEvent::IncrementalUpdateReady { .. } => (), } } @@ -415,7 +522,8 @@ impl LocalProjectIndexManager { let root_path = dunce::canonicalize(root_path).unwrap_or_else(|_| root_path.to_path_buf()); self.indices.remove(&root_path); self.statuses.remove(&root_path); - self.pending_rebuilds.remove(&root_path); + self.pending_updates.remove(&root_path); + self.scheduled_updates.remove(&root_path); let epoch = self.rebuild_epochs.entry(root_path.clone()).or_default(); *epoch += 1; let storage_root = self.storage_root.clone(); @@ -486,20 +594,19 @@ impl LocalProjectIndexManager { &generation_path, ) { Ok(searcher) => { + let Ok(file_count) = searcher.get_all_doc_ids().map(|ids| ids.len()) else { + continue; + }; self.indices.insert( root_path.clone(), LocalProjectIndex { root_path: root_path.clone(), searcher, - file_count: metadata.file_count, - }, - ); - self.statuses.insert( - root_path, - LocalIndexStatus::Ready { - file_count: metadata.file_count, + file_count, }, ); + self.statuses + .insert(root_path, LocalIndexStatus::Ready { file_count }); } Err(error) => { log::warn!( @@ -518,6 +625,7 @@ struct BuiltLocalProjectIndex { generation: String, generation_directory: PathBuf, repository_lock: File, + new_generation: bool, } async fn build_persisted_index( @@ -531,6 +639,7 @@ async fn build_persisted_index( // one process cannot delete another process's in-progress Tantivy generation. let repository_lock = acquire_repository_lock(&storage_root, &root_path).await?; let documents = build_documents(&root_path).await?; + let file_count = documents.len(); let index_directory = repository_storage_directory(&storage_root, &root_path); fs::create_dir_all(index_directory.join("generations"))?; let generation = format!("generation-{}", uuid::Uuid::new_v4()); @@ -552,7 +661,7 @@ async fn build_persisted_index( return Err(error); } }; - if let Err(error) = searcher.build_index(build_search_documents(&documents)) { + if let Err(error) = searcher.build_index(build_search_documents(documents)) { let _ = fs::remove_dir_all(&temporary_directory); return Err(error); } @@ -571,7 +680,7 @@ async fn build_persisted_index( root_path: root_path.to_string_lossy().into_owned(), schema_version: INDEX_SCHEMA_VERSION, generation: generation.clone(), - file_count: documents.len(), + file_count, }; let metadata_json = serde_json::to_vec_pretty(&metadata)?; // Store metadata inside the generation so an interrupted refresh cannot replace the @@ -585,8 +694,9 @@ async fn build_persisted_index( generation, generation_directory, repository_lock, + new_generation: true, }, - documents.len(), + file_count, )) } .await; @@ -602,11 +712,10 @@ struct LocalProjectDocument { } fn build_search_documents( - documents: &[LocalProjectDocument], -) -> impl Iterator + '_ { + documents: Vec, +) -> impl Iterator { documents - .iter() - .cloned() + .into_iter() .map(|document| LocalProjectSearchDocument { symbol: document.symbol, path: document.file_path.clone(), @@ -617,42 +726,204 @@ fn build_search_documents( } async fn build_documents(root_path: &Path) -> Result> { - let root_path = dunce::canonicalize(root_path).with_context(|| { - format!( - "Failed to canonicalize project root {}", - root_path.display() - ) - })?; - let outline = build_outline(&root_path, Some(MAX_INDEX_FILES)).await?; + let root_path = dunce::canonicalize(root_path)?; + documents_under_path(&root_path, &root_path, MAX_INDEX_FILES) +} + +/// Enumerate only the affected subtree, preserving the outline's existing ignore policy. +fn documents_under_path( + root: &Path, + path: &Path, + limit: usize, +) -> Result> { + if !path.starts_with(root) + || limit == 0 + || path.components().any(|part| part.as_os_str() == ".git") + { + return Ok(Vec::new()); + } + let metadata = match fs::symlink_metadata(path) { + Ok(metadata) => metadata, + Err(error) if error.kind() == std::io::ErrorKind::NotFound => return Ok(Vec::new()), + Err(error) => return Err(error.into()), + }; + if metadata.file_type().is_symlink() { + return Ok(Vec::new()); + } + // A watcher path below a symlink must follow the same exclusion policy as a full scan. + for ancestor in path + .ancestors() + .skip(1) + .take_while(|ancestor| ancestor.starts_with(root) && *ancestor != root) + { + if fs::symlink_metadata(ancestor)?.file_type().is_symlink() { + return Ok(Vec::new()); + } + } + let mut ignores = repo_metadata::gitignores_for_directory(root); + if let Some(parent) = path.parent() { + let mut ancestors: Vec<_> = parent + .ancestors() + .take_while(|ancestor| ancestor.starts_with(root) && *ancestor != root) + .collect(); + ancestors.reverse(); + for ancestor in ancestors { + let ignore_path = ancestor.join(".gitignore"); + if ignore_path.is_file() { + ignores.push(ignore::gitignore::Gitignore::new(ignore_path).0); + } + } + } + if repo_metadata::matches_gitignores(path, metadata.is_dir(), &ignores, true) { + return Ok(Vec::new()); + } + let mut files = Vec::new(); + let mut budget = limit; + Entry::build_tree( + path, + &mut files, + &mut ignores, + Some(&mut budget), + 200, + 0, + &IgnoredPathStrategy::Exclude, + BudgetExceededBehavior::StopAndLazyLoad, + )?; let mut documents = Vec::new(); - for file in outline.to_file_symbols(None) { - let relative_path = PathBuf::from(&file.path); - let absolute_path = root_path.join(&relative_path); - let Ok(bytes) = fs::read(&absolute_path) else { - continue; - }; - if bytes.is_empty() || bytes.len() > MAX_INDEXED_FILE_BYTES { - continue; + for file in files { + if let Some(document) = build_document(&file.path.to_local_path_lossy())? { + documents.push(document); } - let Ok(content) = String::from_utf8(bytes) else { - continue; - }; - let body = content.chars().take(MAX_INDEXED_BODY_BYTES).collect(); - let extension = absolute_path - .extension() - .and_then(|extension| extension.to_str()) - .unwrap_or_default(); - let metadata = format!("{extension} {}", file.symbols); - documents.push(LocalProjectDocument { - file_path: absolute_path.to_string_lossy().into_owned(), - symbol: file.symbols, - metadata, - body, - }); } Ok(documents) } +fn build_document(path: &Path) -> Result> { + let metadata = match fs::symlink_metadata(path) { + Ok(metadata) => metadata, + Err(error) + if matches!( + error.kind(), + std::io::ErrorKind::NotFound | std::io::ErrorKind::PermissionDenied + ) => + { + return Ok(None) + } + Err(error) => return Err(error.into()), + }; + if !metadata.is_file() || metadata.len() > MAX_INDEXED_FILE_BYTES as u64 { + return Ok(None); + } + let file = match File::open(path) { + Ok(file) => file, + Err(error) + if matches!( + error.kind(), + std::io::ErrorKind::NotFound | std::io::ErrorKind::PermissionDenied + ) => + { + return Ok(None) + } + Err(error) => return Err(error.into()), + }; + // The read limit also covers files growing between stat and read. + let mut bytes = Vec::new(); + file.take(MAX_INDEXED_FILE_BYTES as u64 + 1) + .read_to_end(&mut bytes)?; + if bytes.is_empty() || bytes.len() > MAX_INDEXED_FILE_BYTES { + return Ok(None); + } + let Ok(mut content) = String::from_utf8(bytes) else { + return Ok(None); + }; + let symbol = parse_file_outline_content(path, &content) + .unwrap_or_default() + .to_string() + .unwrap_or_default(); + let mut end = content.len().min(MAX_INDEXED_BODY_BYTES); + while !content.is_char_boundary(end) { + end -= 1; + } + content.truncate(end); + content.shrink_to_fit(); + let extension = path + .extension() + .and_then(|extension| extension.to_str()) + .unwrap_or_default(); + Ok(Some(LocalProjectDocument { + file_path: path.to_string_lossy().into_owned(), + metadata: format!("{extension} {symbol}"), + symbol, + body: content, + })) +} + +async fn update_persisted_index( + root_path: PathBuf, + storage_root: PathBuf, + paths: HashSet, +) -> Result<(PathBuf, BuiltLocalProjectIndex, usize)> { + let repository_lock = acquire_repository_lock(&storage_root, &root_path).await?; + let directory = repository_storage_directory(&storage_root, &root_path); + let metadata = read_metadata(&directory)?; + anyhow::ensure!( + metadata.schema_version == INDEX_SCHEMA_VERSION, + "Local index schema changed" + ); + let generation_directory = directory.join("generations").join(&metadata.generation); + let searcher = SimpleFullTextSearcher::open_in_dir( + &LOCAL_PROJECT_INDEX_SCHEMA, + DEFAULT_MEMORY_BUDGET, + &generation_directory, + )?; + let existing = searcher.get_all_doc_ids()?; + // Collapse directory/child events so no subtree is parsed more than once per batch. + let mut paths: Vec<_> = paths + .into_iter() + .filter(|path| path.starts_with(&root_path)) + .collect(); + paths.sort(); + let mut roots: Vec = Vec::new(); + for path in paths { + if !roots.iter().any(|root| path.starts_with(root)) { + roots.push(path); + } + } + let existing_count = existing.len(); + let deleted: Vec<_> = existing + .into_iter() + .filter(|id| { + roots + .iter() + .any(|path| Path::new(&id.file_path).starts_with(path)) + }) + .collect(); + let retained = existing_count - deleted.len(); + let mut documents = Vec::new(); + for path in roots { + documents.extend(documents_under_path( + &root_path, + &path, + MAX_INDEX_FILES.saturating_sub(retained + documents.len()), + )?); + } + if !deleted.is_empty() || !documents.is_empty() { + searcher.update_documents(deleted, build_search_documents(documents))?; + } + let file_count = searcher.get_all_doc_ids()?.len(); + Ok(( + root_path, + BuiltLocalProjectIndex { + searcher, + generation: metadata.generation, + generation_directory, + repository_lock, + new_generation: false, + }, + file_count, + )) +} + fn format_storage_directory_name(root_path: &Path) -> String { let mut hasher = Sha256::new(); hasher.update(root_path.to_string_lossy().as_bytes()); diff --git a/crates/ai/src/index/local_project_index/tests.rs b/crates/ai/src/index/local_project_index/tests.rs index 06b5084e..5995d82f 100644 --- a/crates/ai/src/index/local_project_index/tests.rs +++ b/crates/ai/src/index/local_project_index/tests.rs @@ -86,3 +86,130 @@ fn partial_paths_filter_local_search_results() { assert_eq!(matches.len(), 1); assert!(matches[0].path.ends_with("src/lib.rs")); } + +#[test] +fn incremental_commit_replaces_and_removes_files_without_reparsing_unchanged_files() { + let root = TempDir::new().unwrap(); + let storage = TempDir::new().unwrap(); + let root = dunce::canonicalize(root.path()).unwrap(); + fs::write(root.join("changed.rs"), "fn old_symbol() {}\n").unwrap(); + fs::write(root.join("untouched.rs"), "fn retained_symbol() {}\n").unwrap(); + fs::write(root.join("removed.rs"), "fn removed_symbol() {}\n").unwrap(); + let (_, initial, _) = futures::executor::block_on(build_persisted_index( + root.clone(), + storage.path().to_path_buf(), + )) + .unwrap(); + publish_generation(storage.path(), &root, &initial.generation).unwrap(); + // Keep the old reader open, as UI searches do during an update. + drop(initial.repository_lock); + fs::write(root.join("changed.rs"), "fn replacement_symbol() {}\n").unwrap(); + fs::write(root.join("untouched.rs"), "fn must_not_be_read() {}\n").unwrap(); + fs::remove_file(root.join("removed.rs")).unwrap(); + fs::write(root.join("added.rs"), "fn added_symbol() {}\n").unwrap(); + let paths = ["changed.rs", "removed.rs", "added.rs"] + .map(|path| root.join(path)) + .into_iter() + .collect(); + let (_, updated, count) = futures::executor::block_on(update_persisted_index( + root.clone(), + storage.path().to_path_buf(), + paths, + )) + .unwrap(); + assert_eq!(count, 3); + assert_eq!(updated.generation, initial.generation); + assert_eq!(updated.searcher.search_id("old_symbol").unwrap().len(), 0); + assert_eq!( + updated.searcher.search_id("removed_symbol").unwrap().len(), + 0 + ); + for symbol in ["replacement_symbol", "retained_symbol", "added_symbol"] { + assert_eq!(updated.searcher.search_id(symbol).unwrap().len(), 1); + } + assert_eq!( + updated + .searcher + .search_id("must_not_be_read") + .unwrap() + .len(), + 0 + ); + assert_eq!(initial.searcher.search_id("old_symbol").unwrap().len(), 1); + // A second writer can open while both readers are still alive; persisted contents survive reopen. + drop(updated.repository_lock); + let (_, reopened, count) = futures::executor::block_on(update_persisted_index( + root.clone(), + storage.path().to_path_buf(), + HashSet::from([root.join("added.rs")]), + )) + .unwrap(); + assert_eq!(count, 3); + assert_eq!( + reopened + .searcher + .search_id("replacement_symbol") + .unwrap() + .len(), + 1 + ); +} + +#[test] +fn indexed_body_is_capped_at_a_utf8_boundary_and_oversized_files_are_skipped() { + let root = TempDir::new().unwrap(); + let path = root.path().join("unicode.txt"); + let content = "€".repeat(MAX_INDEXED_BODY_BYTES); + fs::write(&path, &content).unwrap(); + let document = build_document(&path).unwrap().unwrap(); + assert_eq!( + document.body.len(), + MAX_INDEXED_BODY_BYTES - MAX_INDEXED_BODY_BYTES % 3 + ); + assert!(content.starts_with(&document.body)); + let oversized = root.path().join("oversized.rs"); + File::create(&oversized) + .unwrap() + .set_len(MAX_INDEXED_FILE_BYTES as u64 + 1) + .unwrap(); + assert!(build_document(&oversized).unwrap().is_none()); + let binary = root.path().join("binary.dat"); + fs::write(&binary, [255, 254]).unwrap(); + assert!(build_document(&binary).unwrap().is_none()); +} + +#[test] +fn subtree_updates_respect_inherited_ignores_and_remove_renamed_paths() { + let root = TempDir::new().unwrap(); + let storage = TempDir::new().unwrap(); + let root = dunce::canonicalize(root.path()).unwrap(); + fs::create_dir(root.join("old")).unwrap(); + fs::write(root.join(".gitignore"), "ignored.rs\n").unwrap(); + fs::write(root.join("old/keep.rs"), "fn kept_symbol() {}\n").unwrap(); + fs::write(root.join("old/ignored.rs"), "fn excluded_symbol() {}\n").unwrap(); + let (_, initial, _) = futures::executor::block_on(build_persisted_index( + root.clone(), + storage.path().to_path_buf(), + )) + .unwrap(); + publish_generation(storage.path(), &root, &initial.generation).unwrap(); + drop(initial); + fs::rename(root.join("old"), root.join("new")).unwrap(); + let paths = HashSet::from([root.join("old"), root.join("new"), root.join("new/keep.rs")]); + let (_, updated, _) = futures::executor::block_on(update_persisted_index( + root.clone(), + storage.path().to_path_buf(), + paths, + )) + .unwrap(); + let hits = updated.searcher.search_id("kept_symbol").unwrap(); + assert_eq!(hits.len(), 1); + assert_eq!( + Path::new(&hits[0].values.file_path), + root.join("new/keep.rs") + ); + assert_eq!( + updated.searcher.search_id("excluded_symbol").unwrap().len(), + 0 + ); +} diff --git a/crates/command-signatures-v2/build.rs b/crates/command-signatures-v2/build.rs index 6027b5d0..f4516273 100644 --- a/crates/command-signatures-v2/build.rs +++ b/crates/command-signatures-v2/build.rs @@ -8,6 +8,8 @@ fn main() -> anyhow::Result<()> { println!("cargo:rerun-if-changed=js/build"); println!("cargo:rerun-if-changed=js/package.json"); println!("cargo:rerun-if-changed=js/tsconfig.json"); + println!("cargo:rerun-if-changed=js/yarn.lock"); + println!("cargo:rerun-if-changed=js/.yarnrc.yml"); if let Err(e) = build_command_signatures() { if !Path::new(format!("{}/js/build", env!("CARGO_MANIFEST_DIR")).as_str()).exists() { @@ -31,8 +33,8 @@ If you continue to encounter issues, ensure you don't have conflicting Node inst } fn build_command_signatures() -> anyhow::Result<()> { - match Command::new("yarn") - .arg("build") + match Command::new("corepack") + .args(["yarn", "build"]) .current_dir(format!("{}/js", env!("CARGO_MANIFEST_DIR"))) .output() { diff --git a/crates/galaxy_agent_core/src/provider_run.rs b/crates/galaxy_agent_core/src/provider_run.rs index 0346c289..5c1ee3ba 100644 --- a/crates/galaxy_agent_core/src/provider_run.rs +++ b/crates/galaxy_agent_core/src/provider_run.rs @@ -495,6 +495,7 @@ impl ProviderRun { pub fn replace_tool_result_archive(&mut self, archive: Vec) { self.tool_result_archive = archive; + archive_tool_results(&mut self.tool_result_archive, Vec::new()); } /// Replaces a summarized prefix while the run is parked between model calls. @@ -782,6 +783,7 @@ impl ProviderRun { &mut self, recoverable_call_ids: &HashSet, ) -> Result { + archive_tool_results(&mut self.tool_result_archive, Vec::new()); if matches!(self.state, ProviderRunState::AwaitingModel { .. }) { self.state = ProviderRunState::Failed { failure: ProviderRunFailure { @@ -1740,43 +1742,42 @@ fn add_usage(total: &mut Usage, turn: &Usage) { .saturating_add(turn.cache_creation_input_tokens); } -fn archive_tool_results( +/// Retain tool history once per call/result ID, including when repairing older archives. +/// A multipart message is appended once, never once per tool contained in it. +pub fn archive_tool_results( archive: &mut Vec, messages: Vec, ) { const MAX_TOOL_RESULT_ARCHIVE_ENTRIES: usize = 400; - - let mut pending_tool_uses = Vec::new(); - for message in messages { - match &message.content { - MessageContent::ToolUse { .. } => pending_tool_uses.push(message), - MessageContent::ToolResult { .. } => { - if let Some(tool_use) = pending_tool_uses.pop() { - archive.push(tool_use); - } - archive.push(message); + let mut seen_uses = HashSet::new(); + let mut seen_results = HashSet::new(); + let previous = std::mem::take(archive); + for mut message in previous.into_iter().chain(messages) { + let keep = match &mut message.content { + MessageContent::ToolUse { tool_use_id, .. } => seen_uses.insert(tool_use_id.clone()), + MessageContent::ToolResult { tool_use_id, .. } => { + seen_results.insert(tool_use_id.clone()) } MessageContent::MultiPart(parts) => { - for part in parts { - match part { - ContentPart::ToolUse { .. } => pending_tool_uses.push(message.clone()), - ContentPart::ToolResult { .. } => { - if let Some(tool_use) = pending_tool_uses.pop() { - archive.push(tool_use); - } - archive.push(message.clone()); - } - ContentPart::Text(_) - | ContentPart::Reasoning { .. } - | ContentPart::Image { .. } => {} + parts.retain(|part| match part { + ContentPart::ToolUse { tool_use_id, .. } => { + seen_uses.insert(tool_use_id.clone()) } - } + ContentPart::ToolResult { tool_use_id, .. } => { + seen_results.insert(tool_use_id.clone()) + } + ContentPart::Text(_) + | ContentPart::Reasoning { .. } + | ContentPart::Image { .. } => false, + }); + !parts.is_empty() } - MessageContent::Text(_) => {} + MessageContent::Text(_) => false, + }; + if keep { + archive.push(message); } } - archive.extend(pending_tool_uses); - if archive.len() > MAX_TOOL_RESULT_ARCHIVE_ENTRIES { archive.drain(0..archive.len() - MAX_TOOL_RESULT_ARCHIVE_ENTRIES); } diff --git a/crates/galaxy_agent_core/src/provider_run_tests.rs b/crates/galaxy_agent_core/src/provider_run_tests.rs index 368d7c23..447080b7 100644 --- a/crates/galaxy_agent_core/src/provider_run_tests.rs +++ b/crates/galaxy_agent_core/src/provider_run_tests.rs @@ -1426,3 +1426,42 @@ fn serialized_numeric_turn_budgets_remain_explicit_limits() { assert_eq!(limits.max_model_turns, Some(100)); assert_eq!(ProviderRunLimits::default().max_model_turns, None); } + +#[test] +fn multipart_archive_is_not_multiplied_and_legacy_duplicates_are_repaired() { + let mut run = run(); + let batch = accept_tool_turn( + &mut run, + tool_turn( + (0..4) + .map(|id| tool_call(&format!("search-{id}"), "search_codebase")) + .collect(), + &["search_codebase"], + ), + ); + for id in 0..4 { + run.complete_tool( + &batch.work_id, + successful_result(&format!("search-{id}"), &format!("file contents {id}")), + ) + .unwrap(); + } + run.commit_tool_batch(&batch.work_id).unwrap(); + let transcript = run.transcript().to_vec(); + let mut archive = Vec::new(); + archive_tool_results(&mut archive, transcript.clone()); + assert_eq!(archive.len(), 2); + for message in &archive { + let MessageContent::MultiPart(parts) = &message.content else { + panic!("expected multipart tools"); + }; + assert_eq!(parts.len(), 4); + } + let expected = archive.clone(); + let duplicated: Vec<_> = (0..4).flat_map(|_| expected.clone()).collect(); + run.replace_tool_result_archive(duplicated); + assert_eq!(run.tool_result_archive(), expected); + // A repeated compaction callback must not grow the archive either. + archive_tool_results(&mut archive, transcript); + assert_eq!(archive, expected); +} diff --git a/crates/warp_search_core/src/searcher.rs b/crates/warp_search_core/src/searcher.rs index 2d8fde40..ccd34353 100644 --- a/crates/warp_search_core/src/searcher.rs +++ b/crates/warp_search_core/src/searcher.rs @@ -470,19 +470,12 @@ impl SearcherWriterWrapper { let guard = handle_copy.read(); let memory_budget = memory_budget.max(MIN_MEMORY_BUDGET); let search_index = guard.search_index.clone(); - let num_threads = std::cmp::min( - available_parallelism().map(|ap| ap.get()).unwrap_or(1), - MAX_THREADS_PER_INDEX_WRITER, - ); - let writer = search_index - .writer_with_num_threads(num_threads, memory_budget) - .ok(); SearcherWriterWrapper { search_index, memory_budget, composite_key_field, - writer, + writer: None, reader_handle, id_field_names: guard.id_fields.keys().cloned().collect(), } @@ -627,7 +620,13 @@ impl SearcherWriterWrapper { fn writer(&mut self) -> anyhow::Result<&mut IndexWriter> { if self.writer.is_none() { - let writer = self.search_index.writer(self.memory_budget)?; + let num_threads = std::cmp::min( + available_parallelism().map(|ap| ap.get()).unwrap_or(1), + MAX_THREADS_PER_INDEX_WRITER.min(self.memory_budget / MIN_MEMORY_BUDGET), + ); + let writer = self + .search_index + .writer_with_num_threads(num_threads, self.memory_budget)?; self.writer = Some(writer); } @@ -785,6 +784,46 @@ impl SimpleFullTextSearcher { ) } + /// Atomically apply deletions and replacements in one commit, then release the writer. + /// Read-only handles remain usable without holding a disk writer lock or worker threads. + pub fn update_documents( + &self, + deleted: impl IntoIterator, + documents: impl IntoIterator, + ) -> anyhow::Result<()> { + let mut writer = self.writer.lock(); + let result = writer.execute_operations( + deleted + .into_iter() + .map(|entry| SearcherEvent::DocumentDeleted(entry.into_identifying_entry())) + .chain( + documents + .into_iter() + .map(|entry| SearcherEvent::DocumentInserted(entry.into_document_entry())), + ), + ); + let rollback = if result.is_err() { + writer + .writer + .as_mut() + .map(|active_writer| active_writer.rollback()) + .transpose() + } else { + Ok(None) + }; + // Let committed segment merges finish before releasing the disk lock. Dropping the + // writer immediately cancels merges and would accumulate small segments on every save. + let merges = writer + .writer + .take() + .map(|active_writer| active_writer.wait_merging_threads()) + .transpose(); + rollback?; + result?; + merges?; + Ok(()) + } + #[allow(unused)] /// Inserts a document into the index. It will overwrite any existing document with the same composite key hash. pub fn insert_document(&self, entry: C::SearchDocEntry) -> anyhow::Result<()> { diff --git a/crates/warp_search_core/src/searcher_tests.rs b/crates/warp_search_core/src/searcher_tests.rs index f99211a9..36a70169 100644 --- a/crates/warp_search_core/src/searcher_tests.rs +++ b/crates/warp_search_core/src/searcher_tests.rs @@ -310,3 +310,36 @@ fn test_searcher_async() { "the index should be cleared and contain no documents" ); } + +#[test] +fn minimum_memory_budget_supports_repeated_atomic_updates() { + define_search_schema!( + schema_name: BUDGET_SCHEMA, + config_name: BudgetSchema, + search_doc: BudgetDocument, + identifying_doc: BudgetId, + search_fields: [name: 1.0], + id_fields: [id: u64] + ); + let searcher = BUDGET_SCHEMA.create_searcher(MIN_MEMORY_BUDGET); + searcher + .update_documents( + Vec::::new(), + [BudgetDocument { + id: 1, + name: "initial".into(), + }], + ) + .unwrap(); + searcher + .update_documents( + [BudgetId { id: 1 }], + [BudgetDocument { + id: 2, + name: "replacement".into(), + }], + ) + .unwrap(); + assert_eq!(searcher.search_id("initial").unwrap().len(), 0); + assert_eq!(searcher.search_id("replacement").unwrap().len(), 1); +} diff --git a/vendor/block/Cargo.toml b/vendor/block/Cargo.toml new file mode 100644 index 00000000..5e97b676 --- /dev/null +++ b/vendor/block/Cargo.toml @@ -0,0 +1,20 @@ +[package] +name = "block" +version = "0.1.6" +authors = ["Steven Sheldon"] + +description = "Rust interface for Apple's C language extension of blocks." +keywords = ["blocks", "osx", "ios", "objective-c"] +readme = "README.md" +repository = "http://github.com/SSheldon/rust-block" +documentation = "http://ssheldon.github.io/rust-objc/block/" +license = "MIT" + +exclude = [ + ".gitignore", + ".travis.yml", + "travis_install.sh", + "travis_test.sh", + "tests-ios/**", +] + diff --git a/vendor/block/GALAXY_PATCH.md b/vendor/block/GALAXY_PATCH.md new file mode 100644 index 00000000..8c73940a --- /dev/null +++ b/vendor/block/GALAXY_PATCH.md @@ -0,0 +1,11 @@ +# Local compatibility patch + +Source: crates.io `block` 0.1.6 by Steven Sheldon, licensed MIT. + +Replaces the uninhabited opaque `Class` enum with an opaque C-layout struct. +Only pointers to this type cross the ABI; no storage is read or constructed in Rust. +This fixes Rust's `uninhabited_static` warning without suppressing the lint. +The previously implicit C ABI is also written explicitly on extern declarations. + +The upstream manifest references an unpublished local test utility; that dev-only +entry is omitted. The library source and README otherwise match the published crate. diff --git a/vendor/block/LICENSE b/vendor/block/LICENSE new file mode 100644 index 00000000..2bd6bdee --- /dev/null +++ b/vendor/block/LICENSE @@ -0,0 +1,21 @@ +MIT License + +Copyright (c) Steven Sheldon + +Permission is hereby granted, free of charge, to any person obtaining a copy +of this software and associated documentation files (the "Software"), to deal +in the Software without restriction, including without limitation the rights +to use, copy, modify, merge, publish, distribute, sublicense, and/or sell +copies of the Software, and to permit persons to whom the Software is +furnished to do so, subject to the following conditions: + +The above copyright notice and this permission notice shall be included in +all copies or substantial portions of the Software. + +THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR +IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, +FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE +AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER +LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, +OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN +THE SOFTWARE. diff --git a/vendor/block/README.md b/vendor/block/README.md new file mode 100644 index 00000000..78c65d08 --- /dev/null +++ b/vendor/block/README.md @@ -0,0 +1,42 @@ +Rust interface for Apple's C language extension of blocks. + +For more information on the specifics of the block implementation, see +Clang's documentation: http://clang.llvm.org/docs/Block-ABI-Apple.html + +## Invoking blocks + +The `Block` struct is used for invoking blocks from Objective-C. For example, +consider this Objective-C function: + +``` objc +int32_t sum(int32_t (^block)(int32_t, int32_t)) { + return block(5, 8); +} +``` + +We could write it in Rust as the following: + +``` rust +unsafe fn sum(block: &Block<(i32, i32), i32>) -> i32 { + block.call((5, 8)) +} +``` + +Note the extra parentheses in the `call` method, since the arguments must be +passed as a tuple. + +## Creating blocks + +Creating a block to pass to Objective-C can be done with the `ConcreteBlock` +struct. For example, to create a block that adds two `i32`s, we could write: + +``` rust +let block = ConcreteBlock::new(|a: i32, b: i32| a + b); +let block = block.copy(); +assert!(unsafe { block.call((5, 8)) } == 13); +``` + +It is important to copy your block to the heap (with the `copy` method) before +passing it to Objective-C; this is because our `ConcreteBlock` is only meant +to be copied once, and we can enforce this in Rust, but if Objective-C code +were to copy it twice we could have a double free. diff --git a/vendor/block/src/lib.rs b/vendor/block/src/lib.rs new file mode 100644 index 00000000..74baa0a9 --- /dev/null +++ b/vendor/block/src/lib.rs @@ -0,0 +1,399 @@ +/*! +A Rust interface for Objective-C blocks. + +For more information on the specifics of the block implementation, see +Clang's documentation: http://clang.llvm.org/docs/Block-ABI-Apple.html + +# Invoking blocks + +The `Block` struct is used for invoking blocks from Objective-C. For example, +consider this Objective-C function: + +``` objc +int32_t sum(int32_t (^block)(int32_t, int32_t)) { + return block(5, 8); +} +``` + +We could write it in Rust as the following: + +``` +# use block::Block; +unsafe fn sum(block: &Block<(i32, i32), i32>) -> i32 { + block.call((5, 8)) +} +``` + +Note the extra parentheses in the `call` method, since the arguments must be +passed as a tuple. + +# Creating blocks + +Creating a block to pass to Objective-C can be done with the `ConcreteBlock` +struct. For example, to create a block that adds two `i32`s, we could write: + +``` +# use block::ConcreteBlock; +let block = ConcreteBlock::new(|a: i32, b: i32| a + b); +let block = block.copy(); +assert!(unsafe { block.call((5, 8)) } == 13); +``` + +It is important to copy your block to the heap (with the `copy` method) before +passing it to Objective-C; this is because our `ConcreteBlock` is only meant +to be copied once, and we can enforce this in Rust, but if Objective-C code +were to copy it twice we could have a double free. +*/ + +#[cfg(test)] +mod test_utils; + +use std::marker::PhantomData; +use std::mem; +use std::ops::{Deref, DerefMut}; +use std::os::raw::{c_int, c_ulong, c_void}; +use std::ptr; + +// Opaque Objective-C runtime storage has an address; an empty enum would make +// the external static uninhabited (rust-lang/rust#74840). +#[repr(C)] +struct Class { _private: [u8; 0] } + +#[cfg_attr(any(target_os = "macos", target_os = "ios"), + link(name = "System", kind = "dylib"))] +#[cfg_attr(not(any(target_os = "macos", target_os = "ios")), + link(name = "BlocksRuntime", kind = "dylib"))] +extern "C" { + static _NSConcreteStackBlock: Class; + + fn _Block_copy(block: *const c_void) -> *mut c_void; + fn _Block_release(block: *const c_void); +} + +/// Types that may be used as the arguments to an Objective-C block. +pub trait BlockArguments: Sized { + /// Calls the given `Block` with self as the arguments. + /// + /// Unsafe because `block` must point to a valid `Block` and this invokes + /// foreign code whose safety the compiler cannot verify. + unsafe fn call_block(self, block: *mut Block) -> R; +} + +macro_rules! block_args_impl { + ($($a:ident : $t:ident),*) => ( + impl<$($t),*> BlockArguments for ($($t,)*) { + unsafe fn call_block(self, block: *mut Block) -> R { + let invoke: unsafe extern "C" fn(*mut Block $(, $t)*) -> R = { + let base = block as *mut BlockBase; + mem::transmute((*base).invoke) + }; + let ($($a,)*) = self; + invoke(block $(, $a)*) + } + } + ); +} + +block_args_impl!(); +block_args_impl!(a: A); +block_args_impl!(a: A, b: B); +block_args_impl!(a: A, b: B, c: C); +block_args_impl!(a: A, b: B, c: C, d: D); +block_args_impl!(a: A, b: B, c: C, d: D, e: E); +block_args_impl!(a: A, b: B, c: C, d: D, e: E, f: F); +block_args_impl!(a: A, b: B, c: C, d: D, e: E, f: F, g: G); +block_args_impl!(a: A, b: B, c: C, d: D, e: E, f: F, g: G, h: H); +block_args_impl!(a: A, b: B, c: C, d: D, e: E, f: F, g: G, h: H, i: I); +block_args_impl!(a: A, b: B, c: C, d: D, e: E, f: F, g: G, h: H, i: I, j: J); +block_args_impl!(a: A, b: B, c: C, d: D, e: E, f: F, g: G, h: H, i: I, j: J, k: K); +block_args_impl!(a: A, b: B, c: C, d: D, e: E, f: F, g: G, h: H, i: I, j: J, k: K, l: L); + +#[repr(C)] +struct BlockBase { + isa: *const Class, + flags: c_int, + _reserved: c_int, + invoke: unsafe extern "C" fn(*mut Block, ...) -> R, +} + +/// An Objective-C block that takes arguments of `A` when called and +/// returns a value of `R`. +#[repr(C)] +pub struct Block { + _base: PhantomData>, +} + +impl Block where A: BlockArguments { + /// Call self with the given arguments. + /// + /// Unsafe because this invokes foreign code that the caller must verify + /// doesn't violate any of Rust's safety rules. For example, if this block + /// is shared with multiple references, the caller must ensure that calling + /// it will not cause a data race. + pub unsafe fn call(&self, args: A) -> R { + args.call_block(self as *const _ as *mut _) + } +} + +/// A reference-counted Objective-C block. +pub struct RcBlock { + ptr: *mut Block, +} + +impl RcBlock { + /// Construct an `RcBlock` for the given block without copying it. + /// The caller must ensure the block has a +1 reference count. + /// + /// Unsafe because `ptr` must point to a valid `Block` and must have a +1 + /// reference count or it will be overreleased when the `RcBlock` is + /// dropped. + pub unsafe fn new(ptr: *mut Block) -> Self { + RcBlock { ptr: ptr } + } + + /// Constructs an `RcBlock` by copying the given block. + /// + /// Unsafe because `ptr` must point to a valid `Block`. + pub unsafe fn copy(ptr: *mut Block) -> Self { + let ptr = _Block_copy(ptr as *const c_void) as *mut Block; + RcBlock { ptr: ptr } + } +} + +impl Clone for RcBlock { + fn clone(&self) -> RcBlock { + unsafe { + RcBlock::copy(self.ptr) + } + } +} + +impl Deref for RcBlock { + type Target = Block; + + fn deref(&self) -> &Block { + unsafe { &*self.ptr } + } +} + +impl Drop for RcBlock { + fn drop(&mut self) { + unsafe { + _Block_release(self.ptr as *const c_void); + } + } +} + +/// Types that may be converted into a `ConcreteBlock`. +pub trait IntoConcreteBlock: Sized where A: BlockArguments { + /// The return type of the resulting `ConcreteBlock`. + type Ret; + + /// Consumes self to create a `ConcreteBlock`. + fn into_concrete_block(self) -> ConcreteBlock; +} + +macro_rules! concrete_block_impl { + ($f:ident) => ( + concrete_block_impl!($f,); + ); + ($f:ident, $($a:ident : $t:ident),*) => ( + impl<$($t,)* R, X> IntoConcreteBlock<($($t,)*)> for X + where X: Fn($($t,)*) -> R { + type Ret = R; + + fn into_concrete_block(self) -> ConcreteBlock<($($t,)*), R, X> { + unsafe extern "C" fn $f<$($t,)* R, X>( + block_ptr: *mut ConcreteBlock<($($t,)*), R, X> + $(, $a: $t)*) -> R + where X: Fn($($t,)*) -> R { + let block = &*block_ptr; + (block.closure)($($a),*) + } + + let f: unsafe extern "C" fn(*mut ConcreteBlock<($($t,)*), R, X> $(, $a: $t)*) -> R = $f; + unsafe { + ConcreteBlock::with_invoke(mem::transmute(f), self) + } + } + } + ); +} + +concrete_block_impl!(concrete_block_invoke_args0); +concrete_block_impl!(concrete_block_invoke_args1, a: A); +concrete_block_impl!(concrete_block_invoke_args2, a: A, b: B); +concrete_block_impl!(concrete_block_invoke_args3, a: A, b: B, c: C); +concrete_block_impl!(concrete_block_invoke_args4, a: A, b: B, c: C, d: D); +concrete_block_impl!(concrete_block_invoke_args5, a: A, b: B, c: C, d: D, e: E); +concrete_block_impl!(concrete_block_invoke_args6, a: A, b: B, c: C, d: D, e: E, f: F); +concrete_block_impl!(concrete_block_invoke_args7, a: A, b: B, c: C, d: D, e: E, f: F, g: G); +concrete_block_impl!(concrete_block_invoke_args8, a: A, b: B, c: C, d: D, e: E, f: F, g: G, h: H); +concrete_block_impl!(concrete_block_invoke_args9, a: A, b: B, c: C, d: D, e: E, f: F, g: G, h: H, i: I); +concrete_block_impl!(concrete_block_invoke_args10, a: A, b: B, c: C, d: D, e: E, f: F, g: G, h: H, i: I, j: J); +concrete_block_impl!(concrete_block_invoke_args11, a: A, b: B, c: C, d: D, e: E, f: F, g: G, h: H, i: I, j: J, k: K); +concrete_block_impl!(concrete_block_invoke_args12, a: A, b: B, c: C, d: D, e: E, f: F, g: G, h: H, i: I, j: J, k: K, l: L); + +/// An Objective-C block whose size is known at compile time and may be +/// constructed on the stack. +#[repr(C)] +pub struct ConcreteBlock { + base: BlockBase, + descriptor: Box>>, + closure: F, +} + +impl ConcreteBlock + where A: BlockArguments, F: IntoConcreteBlock { + /// Constructs a `ConcreteBlock` with the given closure. + /// When the block is called, it will return the value that results from + /// calling the closure. + pub fn new(closure: F) -> Self { + closure.into_concrete_block() + } +} + +impl ConcreteBlock { + /// Constructs a `ConcreteBlock` with the given invoke function and closure. + /// Unsafe because the caller must ensure the invoke function takes the + /// correct arguments. + unsafe fn with_invoke(invoke: unsafe extern "C" fn(*mut Self, ...) -> R, + closure: F) -> Self { + ConcreteBlock { + base: BlockBase { + isa: &_NSConcreteStackBlock, + // 1 << 25 = BLOCK_HAS_COPY_DISPOSE + flags: 1 << 25, + _reserved: 0, + invoke: mem::transmute(invoke), + }, + descriptor: Box::new(BlockDescriptor::new()), + closure: closure, + } + } +} + +impl ConcreteBlock where F: 'static { + /// Copy self onto the heap as an `RcBlock`. + pub fn copy(self) -> RcBlock { + unsafe { + let mut block = self; + let copied = RcBlock::copy(&mut *block); + // At this point, our copy helper has been run so the block will + // be moved to the heap and we can forget the original block + // because the heap block will drop in our dispose helper. + mem::forget(block); + copied + } + } +} + +impl Clone for ConcreteBlock where F: Clone { + fn clone(&self) -> Self { + unsafe { + ConcreteBlock::with_invoke(mem::transmute(self.base.invoke), + self.closure.clone()) + } + } +} + +impl Deref for ConcreteBlock { + type Target = Block; + + fn deref(&self) -> &Block { + unsafe { &*(&self.base as *const _ as *const Block) } + } +} + +impl DerefMut for ConcreteBlock { + fn deref_mut(&mut self) -> &mut Block { + unsafe { &mut *(&mut self.base as *mut _ as *mut Block) } + } +} + +unsafe extern "C" fn block_context_dispose(block: &mut B) { + // Read the block onto the stack and let it drop + ptr::read(block); +} + +unsafe extern "C" fn block_context_copy(_dst: &mut B, _src: &B) { + // The runtime memmoves the src block into the dst block, nothing to do +} + +#[repr(C)] +struct BlockDescriptor { + _reserved: c_ulong, + block_size: c_ulong, + copy_helper: unsafe extern "C" fn(&mut B, &B), + dispose_helper: unsafe extern "C" fn(&mut B), +} + +impl BlockDescriptor { + fn new() -> BlockDescriptor { + BlockDescriptor { + _reserved: 0, + block_size: mem::size_of::() as c_ulong, + copy_helper: block_context_copy::, + dispose_helper: block_context_dispose::, + } + } +} + +#[cfg(test)] +mod tests { + use test_utils::*; + use super::{ConcreteBlock, RcBlock}; + + #[test] + fn test_call_block() { + let block = get_int_block_with(13); + unsafe { + assert!(block.call(()) == 13); + } + } + + #[test] + fn test_call_block_args() { + let block = get_add_block_with(13); + unsafe { + assert!(block.call((2,)) == 15); + } + } + + #[test] + fn test_create_block() { + let block = ConcreteBlock::new(|| 13); + let result = invoke_int_block(&block); + assert!(result == 13); + } + + #[test] + fn test_create_block_args() { + let block = ConcreteBlock::new(|a: i32| a + 5); + let result = invoke_add_block(&block, 6); + assert!(result == 11); + } + + #[test] + fn test_concrete_block_copy() { + let s = "Hello!".to_string(); + let expected_len = s.len() as i32; + let block = ConcreteBlock::new(move || s.len() as i32); + assert!(invoke_int_block(&block) == expected_len); + + let copied = block.copy(); + assert!(invoke_int_block(&copied) == expected_len); + } + + #[test] + fn test_concrete_block_stack_copy() { + fn make_block() -> RcBlock<(), i32> { + let x = 7; + let block = ConcreteBlock::new(move || x); + block.copy() + } + + let block = make_block(); + assert!(invoke_int_block(&block) == 7); + } +} diff --git a/vendor/block/src/test_utils.rs b/vendor/block/src/test_utils.rs new file mode 100644 index 00000000..940a3d22 --- /dev/null +++ b/vendor/block/src/test_utils.rs @@ -0,0 +1,31 @@ +extern crate objc_test_utils; + +use {Block, RcBlock}; + +pub fn get_int_block_with(i: i32) -> RcBlock<(), i32> { + unsafe { + let ptr = objc_test_utils::get_int_block_with(i); + RcBlock::new(ptr as *mut _) + } +} + +pub fn get_add_block_with(i: i32) -> RcBlock<(i32,), i32> { + unsafe { + let ptr = objc_test_utils::get_add_block_with(i); + RcBlock::new(ptr as *mut _) + } +} + +pub fn invoke_int_block(block: &Block<(), i32>) -> i32 { + let ptr = block as *const _; + unsafe { + objc_test_utils::invoke_int_block(ptr as *mut _) + } +} + +pub fn invoke_add_block(block: &Block<(i32,), i32>, a: i32) -> i32 { + let ptr = block as *const _; + unsafe { + objc_test_utils::invoke_add_block(ptr as *mut _, a) + } +}