public/galaxy
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity
Files
88c1ef9716c18b4ead8ca5672ba5b5920b685bdf
galaxy/app/src/ai/blocklist/action_model/execute.rs
T

1751 lines
72 KiB
Rust
Raw Blame History

pub(super) mod ask_user_question;
pub(super) mod call_mcp_tool;
pub(super) mod create_documents;
pub(super) mod edit_documents;
pub(super) mod fetch_conversation;
pub(super) mod file_glob;
pub(super) mod grep;
pub(super) mod notebooks;
pub(super) mod read_documents;
pub(super) mod read_files;
pub(super) mod read_mcp_resource;
pub(super) mod read_skill;
pub(super) mod request_computer_use;
pub(super) mod request_file_edits;
pub(super) mod run_agents;
pub(super) mod search_codebase;
pub(super) mod send_message;
pub(super) mod shell_command;
pub(super) mod start_agent;
pub(super) mod suggest_new_conversation;
pub(super) mod suggest_prompt;
pub(super) mod upload_artifact;
pub(super) mod use_computer;
pub(super) mod wait_for_events;
use std::any::Any;
use std::collections::{HashMap, HashSet};
use std::path::PathBuf;
use std::pin::Pin;
use std::sync::Arc;
use std::time::Duration;
use ai::agent::action_result::{InsertReviewCommentsResult, RequestCommandOutputResult};
pub use ask_user_question::AskUserQuestionExecutor;
pub(crate) use call_mcp_tool::coerce_integer_args;
use call_mcp_tool::CallMCPToolExecutor;
use create_documents::CreateDocumentsExecutor;
use edit_documents::EditDocumentsExecutor;
use fetch_conversation::FetchConversationExecutor;
use file_glob::FileGlobExecutor;
use futures::future::BoxFuture;
#[cfg(feature = "local_fs")]
use futures::AsyncReadExt;
use futures::FutureExt;
use galaxy_core::execution_mode::AppExecutionMode;
use galaxy_core::features::FeatureFlag;
#[cfg(feature = "local_fs")]
use galaxy_files::{FileModel, TextFileReadResult};
#[cfg(feature = "local_fs")]
use galaxy_util::file::FileLoadError;
#[cfg(feature = "local_fs")]
use galaxy_util::file_type::is_buffer_binary;
use galaxyui::r#async::{Spawnable, SpawnableOutput};
use galaxyui::{AppContext, Entity, EntityId, ModelContext, ModelHandle, SingletonEntity};
use grep::GrepExecutor;
use instant::Instant;
#[cfg(feature = "local_fs")]
use mime_guess::from_path;
use notebooks::NotebookExecutor;
use parking_lot::{FairMutex, Mutex};
use read_documents::ReadDocumentsExecutor;
pub(super) use read_files::ReadFilesExecutor;
use read_mcp_resource::ReadMCPResourceExecutor;
use read_skill::ReadSkillExecutor;
use request_computer_use::RequestComputerUseExecutor;
pub(crate) use request_file_edits::{apply_edits, FileReadResult, MalformedFinalLineProxyEvent};
pub use request_file_edits::{
EditAcceptAndContinueClickedEvent, EditAcceptClickedEvent, EditResolvedEvent, EditStats,
RequestFileEditsExecutor, RequestFileEditsFormatKind, RequestFileEditsTelemetryEvent,
};
#[cfg(test)]
pub use run_agents::{compose_run_agents_child_prompt, run_agents_to_start_agent_mode};
pub use run_agents::{RunAgentsExecutor, RunAgentsExecutorEvent, RunAgentsSpawningSnapshot};
pub use send_message::SendMessageToAgentExecutor;
use serde::{Deserialize, Serialize};
pub use shell_command::{ShellCommandExecutor, ShellCommandExecutorEvent};
pub use start_agent::{
StartAgentExecutor, StartAgentExecutorEvent, StartAgentRequest, StartAgentRequestId,
StartAgentWaitPolicy,
};
pub use suggest_new_conversation::NewConversationDecision;
use suggest_new_conversation::SuggestNewConversationExecutor;
pub use suggest_prompt::PromptSuggestionExecutor;
use upload_artifact::UploadArtifactExecutor;
use use_computer::UseComputerExecutor;
use wait_for_events::WaitForEventsExecutor;
use self::search_codebase::SearchCodebaseExecutor;
use crate::ai::agent::conversation::AIConversationId;
use crate::ai::agent::{
AIAgentAction, AIAgentActionId, AIAgentActionResult, AIAgentActionResultType,
AIAgentActionType, AIAgentActionTypeDiscriminants, CancellationReason, FileContext,
FileLocations, ServerOutputId,
};
use crate::ai::ambient_agents::AmbientAgentTaskId;
use crate::ai::get_relevant_files::controller::GetRelevantFilesController;
#[cfg(feature = "local_fs")]
use crate::ai::{agent::AnyFileContent, paths::host_native_absolute_path};
use crate::terminal::model::session::active_session::ActiveSession;
use crate::terminal::model::session::command_executor::shell_quote_arg;
use crate::terminal::model::session::{ExecuteCommandOptions, Session, SessionId};
use crate::terminal::model_events::ModelEventDispatcher;
use crate::terminal::shell::ShellType;
use crate::terminal::{ShellLaunchData, TerminalModel};
#[cfg(feature = "local_fs")]
use crate::util::image::{
is_supported_image_mime_type, process_image_for_agent, ProcessImageResult,
};
#[cfg(feature = "local_fs")]
use crate::util::openable_file_type::is_binary_file;
use crate::BlocklistAIHistoryModel;
const CHILD_AGENT_DELEGATION_DENIAL_REASON: &str =
"Child agents are leaf workers and cannot launch additional agents. Complete the assigned task directly or report the blocker to the lead agent.";
const CHILD_AGENT_LEAF_INSTRUCTIONS: &str = r#"You are a leaf worker launched by a lead agent.
- Complete the assigned task directly and stay within its stated scope.
- Do not launch, delegate to, or create additional agents.
- Report blockers and completion to the lead through the available coordination channel."#;
pub(super) fn child_agent_delegation_denial_reason(
conversation_id: AIConversationId,
ctx: &AppContext,
) -> Option<String> {
BlocklistAIHistoryModel::as_ref(ctx)
.conversation(&conversation_id)
.is_some_and(|conversation| conversation.is_child_agent_conversation())
.then(|| CHILD_AGENT_DELEGATION_DENIAL_REASON.to_string())
}
pub(super) fn compose_leaf_agent_prompt(task_prompt: &str) -> String {
format!("{CHILD_AGENT_LEAF_INSTRUCTIONS}\n\nAssigned task:\n{task_prompt}")
}
/// Types of actions that can be executed in parallel.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub(super) enum ParallelExecutionPolicy {
/// Read-only actions that only inspect local context and may be safely coalesced into the
/// same execution phase when the underlying runtime supports it.
ReadOnlyLocalContext,
}
/// Whether an action is running serially or in parallel with other actions.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub(super) enum RunningActionPhase {
/// A barrier action that must run by itself.
Serial,
/// A phase where several actions from the same compatibility group may be in flight together.
Parallel(ParallelExecutionPolicy),
}
const GIT_REPOSITORY_CACHE_TTL: Duration = Duration::from_secs(60);
type GitRepositoryCacheEntries = HashMap<(SessionId, String), (Instant, bool)>;
#[derive(Clone, Default)]
struct GitRepositoryCache {
entries: Arc<Mutex<GitRepositoryCacheEntries>>,
}
impl GitRepositoryCache {
fn get(&self, session_id: SessionId, path: &str) -> Option<bool> {
let key = (session_id, path.to_string());
let mut entries = self.entries.lock();
let (cached_at, is_repository) = entries.get(&key).copied()?;
if cached_at.elapsed() < GIT_REPOSITORY_CACHE_TTL {
Some(is_repository)
} else {
entries.remove(&key);
None
}
}
fn insert(&self, session_id: SessionId, path: String, is_repository: bool) {
let mut entries = self.entries.lock();
entries.retain(|_, (cached_at, _)| cached_at.elapsed() < GIT_REPOSITORY_CACHE_TTL);
entries.insert((session_id, path), (Instant::now(), is_repository));
}
}
#[derive(Debug, Clone, Copy)]
struct ExecuteActionInput<'a> {
action: &'a AIAgentAction,
conversation_id: AIConversationId,
}
#[derive(Debug, Clone, Copy)]
struct PreprocessActionInput<'a> {
action: &'a AIAgentAction,
conversation_id: AIConversationId,
}
type AsyncExecuteActionFn<T> = Pin<Box<dyn Spawnable<Output = T>>>;
type OnCompleteFn<T> = Box<dyn FnOnce(T, &mut AppContext) -> AIAgentActionResultType>;
enum ActionExecution<T: SpawnableOutput> {
Async {
execute_future: AsyncExecuteActionFn<T>,
on_complete: OnCompleteFn<T>,
},
Sync(AIAgentActionResultType),
NotReady,
InvalidAction,
}
impl<T: SpawnableOutput> ActionExecution<T> {
fn new_async(
execute_future: impl Spawnable<Output = T>,
on_complete: impl FnOnce(T, &mut AppContext) -> AIAgentActionResultType + 'static,
) -> Self {
Self::Async {
execute_future: Box::pin(execute_future),
on_complete: Box::new(on_complete),
}
}
}
/// A trait implemented by all types that implement [`Any`] and [`SpawnableOutput`].
trait AnySpawnableOutput: Any + SpawnableOutput {}
impl<T> AnySpawnableOutput for T where T: Any + SpawnableOutput {}
type AnyAsyncExecuteActionFn = Pin<Box<dyn Spawnable<Output = Box<dyn AnySpawnableOutput>>>>;
type AnyOnCompleteFn = Box<dyn FnOnce(Box<dyn Any>, &mut AppContext) -> AIAgentActionResultType>;
enum AnyActionExecution {
Async {
execute_future: AnyAsyncExecuteActionFn,
on_complete: AnyOnCompleteFn,
},
Sync(AIAgentActionResultType),
NotReady,
InvalidAction,
}
impl<T> From<ActionExecution<T>> for AnyActionExecution
where
T: Send + 'static,
{
fn from(value: ActionExecution<T>) -> Self {
match value {
ActionExecution::Async {
execute_future,
on_complete,
} => AnyActionExecution::Async {
execute_future: Box::pin(async move {
let result = execute_future.await;
Box::new(result) as Box<dyn AnySpawnableOutput>
}),
on_complete: Box::new(move |result, app| {
on_complete(*result.downcast::<T>().expect("Type is correct."), app)
}),
},
ActionExecution::Sync(result) => AnyActionExecution::Sync(result),
ActionExecution::NotReady => AnyActionExecution::NotReady,
ActionExecution::InvalidAction => AnyActionExecution::InvalidAction,
}
}
}
#[derive(Debug, Copy, Clone)]
pub enum NotExecutedReason {
NotReady,
NeedsConfirmation,
WaitingOnSharer,
}
/// Result type for `BlocklistAIActionExecutor::try_to_execute_action`.
#[derive(Debug)]
pub(super) enum TryExecuteResult {
ExecutedSync,
ExecutedAsync,
NotExecuted {
reason: NotExecutedReason,
action: Box<AIAgentAction>,
},
}
#[derive(Clone)]
struct AsyncExecutingAction {
action: AIAgentAction,
}
type AsyncExecutingActionKey = (AIConversationId, AIAgentActionId);
#[derive(Default)]
struct AsyncExecutingActions(
std::collections::HashMap<AsyncExecutingActionKey, AsyncExecutingAction>,
);
impl AsyncExecutingActions {
fn insert(&mut self, conversation_id: AIConversationId, running: AsyncExecutingAction) {
self.0
.insert((conversation_id, running.action.id.clone()), running);
}
fn get(
&self,
conversation_id: AIConversationId,
action_id: &AIAgentActionId,
) -> Option<&AsyncExecutingAction> {
self.0.get(&(conversation_id, action_id.clone()))
}
fn remove(
&mut self,
conversation_id: AIConversationId,
action_id: &AIAgentActionId,
) -> Option<AsyncExecutingAction> {
self.0.remove(&(conversation_id, action_id.clone()))
}
}
impl AsyncExecutingAction {
fn is_shell_command_action(&self) -> bool {
matches!(
self.action.action,
AIAgentActionType::RequestCommandOutput { .. }
| AIAgentActionType::WriteToLongRunningShellCommand { .. }
| AIAgentActionType::ReadShellCommandOutput { .. }
)
}
}
pub struct BlocklistAIActionExecutor {
shell_command_executor: ModelHandle<ShellCommandExecutor>,
read_files_executor: ModelHandle<ReadFilesExecutor>,
upload_artifact_executor: ModelHandle<UploadArtifactExecutor>,
search_codebase_executor: ModelHandle<SearchCodebaseExecutor>,
request_file_edits_executor: ModelHandle<RequestFileEditsExecutor>,
grep_executor: ModelHandle<GrepExecutor>,
file_glob_executor: ModelHandle<FileGlobExecutor>,
read_mcp_resource_executor: ModelHandle<ReadMCPResourceExecutor>,
call_mcp_tool_executor: ModelHandle<CallMCPToolExecutor>,
suggest_new_conversation_executor: ModelHandle<SuggestNewConversationExecutor>,
suggest_prompt_executor: ModelHandle<PromptSuggestionExecutor>,
read_documents_executor: ModelHandle<ReadDocumentsExecutor>,
edit_documents_executor: ModelHandle<EditDocumentsExecutor>,
create_documents_executor: ModelHandle<CreateDocumentsExecutor>,
notebook_executor: ModelHandle<NotebookExecutor>,
use_computer_executor: ModelHandle<UseComputerExecutor>,
request_computer_use_executor: ModelHandle<RequestComputerUseExecutor>,
read_skill_executor: ModelHandle<ReadSkillExecutor>,
fetch_conversation_executor: ModelHandle<FetchConversationExecutor>,
start_agent_executor: ModelHandle<StartAgentExecutor>,
run_agents_executor: ModelHandle<RunAgentsExecutor>,
send_message_executor: ModelHandle<SendMessageToAgentExecutor>,
ask_user_question_executor: ModelHandle<AskUserQuestionExecutor>,
wait_for_events_executor: ModelHandle<WaitForEventsExecutor>,
/// The actions currently executing asynchronously, scoped by conversation and action ID.
async_executing_actions: AsyncExecutingActions,
restored_action_ids: HashSet<AsyncExecutingActionKey>,
/// Reference to the terminal model for checking session sharing state.
terminal_model: Arc<FairMutex<TerminalModel>>,
}
impl BlocklistAIActionExecutor {
pub fn new(
terminal_model: Arc<FairMutex<TerminalModel>>,
active_session: ModelHandle<ActiveSession>,
model_event_dispatcher: &ModelHandle<ModelEventDispatcher>,
get_relevant_files_controller: ModelHandle<GetRelevantFilesController>,
terminal_view_id: EntityId,
ctx: &mut ModelContext<Self>,
) -> Self {
let read_files_executor =
ctx.add_model(|_| ReadFilesExecutor::new(active_session.clone(), terminal_view_id));
let upload_artifact_executor = ctx
.add_model(|_| UploadArtifactExecutor::new(active_session.clone(), terminal_view_id));
let search_codebase_executor = ctx.add_model(|ctx| {
SearchCodebaseExecutor::new(
active_session.clone(),
get_relevant_files_controller,
terminal_view_id,
ctx,
)
});
let shell_command_executor = ctx.add_model(|ctx| {
ShellCommandExecutor::new(
active_session.clone(),
terminal_model.clone(),
model_event_dispatcher,
terminal_view_id,
ctx,
)
});
let request_file_edits_executor = ctx.add_model(|ctx| {
RequestFileEditsExecutor::new(active_session.clone(), terminal_view_id, ctx)
});
let git_repository_cache = GitRepositoryCache::default();
let grep_executor = ctx.add_model(|_| {
GrepExecutor::new(
active_session.clone(),
terminal_view_id,
git_repository_cache.clone(),
)
});
let file_glob_executor = ctx.add_model(|_| {
FileGlobExecutor::new(
active_session.clone(),
terminal_view_id,
git_repository_cache,
)
});
let read_mcp_resource_executor = ctx
.add_model(|_| ReadMCPResourceExecutor::new(active_session.clone(), terminal_view_id));
let call_mcp_tool_executor =
ctx.add_model(|_| CallMCPToolExecutor::new(active_session.clone(), terminal_view_id));
let suggest_new_conversation_executor =
ctx.add_model(|_| SuggestNewConversationExecutor::new());
let suggest_prompt_executor = ctx.add_model(|_| PromptSuggestionExecutor::new());
let read_documents_executor = ctx.add_model(|_| ReadDocumentsExecutor::new());
let edit_documents_executor = ctx.add_model(|_| EditDocumentsExecutor::new());
let create_documents_executor = ctx
.add_model(|_| CreateDocumentsExecutor::new(active_session.clone(), terminal_view_id));
let notebook_executor = ctx.add_model(|_| NotebookExecutor::new());
let use_computer_executor = ctx.add_model(|_| UseComputerExecutor::new());
let request_computer_use_executor =
ctx.add_model(|_| RequestComputerUseExecutor::new(terminal_view_id));
let read_skill_executor = ctx.add_model(|_| ReadSkillExecutor::new(active_session.clone()));
let fetch_conversation_executor = ctx.add_model(|_| FetchConversationExecutor::new());
let start_agent_executor = ctx.add_model(StartAgentExecutor::new);
let run_agents_executor = ctx.add_model(|ctx| {
RunAgentsExecutor::new(start_agent_executor.clone(), terminal_view_id, ctx)
});
let send_message_executor = ctx.add_model(|_| SendMessageToAgentExecutor::new());
let ask_user_question_executor =
ctx.add_model(|_| AskUserQuestionExecutor::new(terminal_view_id));
let wait_for_events_executor =
ctx.add_model(|ctx| WaitForEventsExecutor::new(terminal_view_id, ctx));
Self {
shell_command_executor,
read_files_executor,
upload_artifact_executor,
search_codebase_executor,
request_file_edits_executor,
grep_executor,
file_glob_executor,
read_mcp_resource_executor,
call_mcp_tool_executor,
suggest_new_conversation_executor,
suggest_prompt_executor,
read_documents_executor,
edit_documents_executor,
create_documents_executor,
notebook_executor,
use_computer_executor,
request_computer_use_executor,
async_executing_actions: Default::default(),
restored_action_ids: Default::default(),
terminal_model,
read_skill_executor,
fetch_conversation_executor,
start_agent_executor,
run_agents_executor,
send_message_executor,
ask_user_question_executor,
wait_for_events_executor,
}
}
pub fn async_executing_action(
&self,
conversation_id: AIConversationId,
action_id: &AIAgentActionId,
) -> Option<&AIAgentAction> {
self.async_executing_actions
.get(conversation_id, action_id)
.map(|running| &running.action)
}
pub fn mark_restored_actions(
&mut self,
conversation_id: AIConversationId,
action_ids: &HashSet<AIAgentActionId>,
ctx: &mut ModelContext<Self>,
) {
self.restored_action_ids.extend(
action_ids
.iter()
.cloned()
.map(|action_id| (conversation_id, action_id)),
);
self.run_agents_executor.update(ctx, |executor, _| {
executor.mark_recovery_actions(conversation_id, action_ids);
});
}
pub(super) fn has_running_ask_user_question(&self, conversation_id: AIConversationId) -> bool {
self.async_executing_actions
.0
.iter()
.any(|((running_conversation_id, _), running)| {
*running_conversation_id == conversation_id
&& matches!(
running.action.action,
AIAgentActionType::AskUserQuestion { .. }
)
})
}
/// Returns the action_id of any running WaitForEvents action for the
/// given conversation. There is at most one (wait_for_events is
/// documented as exclusive within a turn).
pub(super) fn find_running_wait_for_events(
&self,
conversation_id: AIConversationId,
) -> Option<AIAgentActionId> {
self.async_executing_actions.0.iter().find_map(
|((running_conversation_id, action_id), running)| {
if *running_conversation_id == conversation_id
&& matches!(
running.action.action,
AIAgentActionType::WaitForEvents { .. }
)
{
Some(action_id.clone())
} else {
None
}
},
)
}
pub fn shell_command_executor(&self) -> &ModelHandle<ShellCommandExecutor> {
&self.shell_command_executor
}
pub fn request_file_edits_executor(&self) -> &ModelHandle<RequestFileEditsExecutor> {
&self.request_file_edits_executor
}
pub fn search_codebase_executor(&self) -> &ModelHandle<SearchCodebaseExecutor> {
&self.search_codebase_executor
}
pub fn suggest_new_conversation_executor(
&self,
) -> &ModelHandle<SuggestNewConversationExecutor> {
&self.suggest_new_conversation_executor
}
pub fn suggest_prompt_executor(&self) -> &ModelHandle<PromptSuggestionExecutor> {
&self.suggest_prompt_executor
}
pub fn start_agent_executor(&self) -> &ModelHandle<StartAgentExecutor> {
&self.start_agent_executor
}
pub fn run_agents_executor(&self) -> &ModelHandle<RunAgentsExecutor> {
&self.run_agents_executor
}
pub fn action_phase(&self, action: &AIAgentAction, ctx: &AppContext) -> RunningActionPhase {
match &action.action {
AIAgentActionType::ReadFiles(..)
| AIAgentActionType::SearchCodebase(..)
| AIAgentActionType::ReadSkill(_) => {
RunningActionPhase::Parallel(ParallelExecutionPolicy::ReadOnlyLocalContext)
}
AIAgentActionType::Grep { .. }
if self.grep_executor.as_ref(ctx).can_execute_in_parallel(ctx) =>
{
RunningActionPhase::Parallel(ParallelExecutionPolicy::ReadOnlyLocalContext)
}
AIAgentActionType::FileGlob { .. } | AIAgentActionType::FileGlobV2 { .. }
if self
.file_glob_executor
.as_ref(ctx)
.can_execute_in_parallel(ctx) =>
{
RunningActionPhase::Parallel(ParallelExecutionPolicy::ReadOnlyLocalContext)
}
AIAgentActionType::StartAgent { .. } => {
RunningActionPhase::Parallel(ParallelExecutionPolicy::ReadOnlyLocalContext)
}
_ => RunningActionPhase::Serial,
}
}
pub fn ask_user_question_executor(&self) -> &ModelHandle<AskUserQuestionExecutor> {
&self.ask_user_question_executor
}
pub fn set_ambient_agent_task_id(
&self,
id: Option<AmbientAgentTaskId>,
ctx: &mut ModelContext<Self>,
) {
self.send_message_executor.update(ctx, |executor, _| {
executor.set_ambient_agent_task_id(id);
});
self.request_computer_use_executor
.update(ctx, |executor, _| {
executor.set_ambient_agent_task_id(id);
});
}
pub fn preprocess_action(
&self,
action: &AIAgentAction,
conversation_id: AIConversationId,
ctx: &mut ModelContext<Self>,
) -> BoxFuture<'static, ()> {
// In view-only mode, we do not need to perform any preprocessing work.
if self.is_shared_session_viewer() {
return futures::future::ready(()).boxed();
}
let input = PreprocessActionInput {
action,
conversation_id,
};
match &action.action {
AIAgentActionType::RequestCommandOutput { .. }
| AIAgentActionType::WriteToLongRunningShellCommand { .. }
| AIAgentActionType::ReadShellCommandOutput { .. }
| AIAgentActionType::TransferShellCommandControlToUser { .. } => self
.shell_command_executor
.update(ctx, |executor, ctx| executor.preprocess_action(input, ctx)),
AIAgentActionType::ReadFiles(..) => self
.read_files_executor
.update(ctx, |executor, ctx| executor.preprocess_action(input, ctx)),
AIAgentActionType::UploadArtifact(..) => self
.upload_artifact_executor
.update(ctx, |executor, ctx| executor.preprocess_action(input, ctx)),
AIAgentActionType::SearchCodebase(..) => self
.search_codebase_executor
.update(ctx, |executor, ctx| executor.preprocess_action(input, ctx)),
AIAgentActionType::Grep { .. } => self
.grep_executor
.update(ctx, |executor, ctx| executor.preprocess_action(input, ctx)),
AIAgentActionType::FileGlob { .. } | AIAgentActionType::FileGlobV2 { .. } => self
.file_glob_executor
.update(ctx, |executor, ctx| executor.preprocess_action(input, ctx)),
AIAgentActionType::CallMCPTool { .. } => self
.call_mcp_tool_executor
.update(ctx, |executor, ctx| executor.preprocess_action(input, ctx)),
AIAgentActionType::ReadMCPResource { .. } => self
.read_mcp_resource_executor
.update(ctx, |executor, ctx| executor.preprocess_action(input, ctx)),
// Normally, requested file edits are not handled by the executor. However, when performing a task autonomously,
// the executor is responsible for auto-approving diffs.
AIAgentActionType::RequestFileEdits { .. } => self
.request_file_edits_executor
.update(ctx, |executor, ctx| executor.preprocess_action(input, ctx)),
AIAgentActionType::InitProject => futures::future::ready(()).boxed(),
AIAgentActionType::OpenCodeReview => futures::future::ready(()).boxed(),
AIAgentActionType::InsertCodeReviewComments { .. } => {
futures::future::ready(()).boxed()
}
AIAgentActionType::SuggestNewConversation { .. } => self
.suggest_new_conversation_executor
.update(ctx, |executor, ctx| executor.preprocess_action(input, ctx)),
AIAgentActionType::SuggestPrompt { .. } => self
.suggest_prompt_executor
.update(ctx, |executor, ctx| executor.preprocess_action(input, ctx)),
AIAgentActionType::ReadDocuments(_) => {
if action.tool_name.as_deref() == Some("notebook") {
self.notebook_executor
.update(ctx, |executor, ctx| executor.preprocess_action(input, ctx))
} else {
self.read_documents_executor
.update(ctx, |executor, ctx| executor.preprocess_action(input, ctx))
}
}
AIAgentActionType::EditDocuments(_) => {
if action.tool_name.as_deref() == Some("notebook") {
self.notebook_executor
.update(ctx, |executor, ctx| executor.preprocess_action(input, ctx))
} else {
self.edit_documents_executor
.update(ctx, |executor, ctx| executor.preprocess_action(input, ctx))
}
}
AIAgentActionType::CreateDocuments(_) => {
if action.tool_name.as_deref() == Some("notebook") {
self.notebook_executor
.update(ctx, |executor, ctx| executor.preprocess_action(input, ctx))
} else {
self.create_documents_executor
.update(ctx, |executor, ctx| executor.preprocess_action(input, ctx))
}
}
AIAgentActionType::UseComputer(_) => self
.use_computer_executor
.update(ctx, |executor, ctx| executor.preprocess_action(input, ctx)),
AIAgentActionType::RequestComputerUse(_) => self
.request_computer_use_executor
.update(ctx, |executor, ctx| executor.preprocess_action(input, ctx)),
AIAgentActionType::ReadSkill(_) => self
.read_skill_executor
.update(ctx, |executor, ctx| executor.preprocess_action(input, ctx)),
AIAgentActionType::FetchConversation { .. } => self
.fetch_conversation_executor
.update(ctx, |executor, ctx| executor.preprocess_action(input, ctx)),
AIAgentActionType::StartAgent { .. } => self
.start_agent_executor
.update(ctx, |executor, ctx| executor.preprocess_action(input, ctx)),
AIAgentActionType::SendMessageToAgent { .. } => self
.send_message_executor
.update(ctx, |executor, ctx| executor.preprocess_action(input, ctx)),
AIAgentActionType::AskUserQuestion { .. } => self
.ask_user_question_executor
.update(ctx, |executor, ctx| executor.preprocess_action(input, ctx)),
AIAgentActionType::RunAgents(_) => self
.run_agents_executor
.update(ctx, |executor, ctx| executor.preprocess_action(input, ctx)),
AIAgentActionType::WaitForEvents { .. } => self
.wait_for_events_executor
.update(ctx, |executor, ctx| executor.preprocess_action(input, ctx)),
}
}
/// Returns `None` if the action was executed (and thereby consumed).
///
/// If the executor cannot execute the action at this time, returns a result indicating why.
pub fn try_to_execute_action(
&mut self,
action: AIAgentAction,
conversation_id: AIConversationId,
is_user_initiated: bool,
ctx: &mut ModelContext<Self>,
) -> TryExecuteResult {
crate::ai::tool_diagnostics::tool_debug!(
"try_to_execute_action: action_id={:?}, type={:?}, is_user_initiated={}",
action.id,
std::mem::discriminant(&action.action),
is_user_initiated
);
// We should never actually execute actions in view-only mode.
if self.is_shared_session_viewer() {
crate::ai::tool_diagnostics::tool_debug!(
"try_to_execute_action: BLOCKED - shared session viewer mode"
);
return TryExecuteResult::NotExecuted {
reason: NotExecutedReason::WaitingOnSharer,
action: Box::new(action),
};
}
let input = ExecuteActionInput {
action: &action,
conversation_id,
};
let can_auto_execute = self.should_autoexecute(input, ctx);
let is_agent_autonomous = AppExecutionMode::as_ref(ctx).is_autonomous();
crate::ai::tool_diagnostics::tool_debug!(
"try_to_execute_action: can_auto_execute={}, is_agent_autonomous={}",
can_auto_execute,
is_agent_autonomous
);
// The agent cannot auto execute and either:
// - the agent is interactive, OR
// - the agent is autonomous and the action was not requesting command output
let needs_confirmation = !(is_user_initiated
|| can_auto_execute
|| (is_agent_autonomous && action.action.is_request_command_output()));
if needs_confirmation {
crate::ai::tool_diagnostics::tool_debug!(
"try_to_execute_action: NEEDS CONFIRMATION - action_id={:?}",
action.id
);
return TryExecuteResult::NotExecuted {
action: Box::new(action),
reason: NotExecutedReason::NeedsConfirmation,
};
} else if !is_user_initiated && !can_auto_execute && is_agent_autonomous {
// It must be the case that the autonomous agent is requesting a denylisted command.
if let AIAgentActionType::RequestCommandOutput { command, .. } = &action.action {
let action_id = action.id.clone();
let result = AIAgentActionResultType::RequestCommandOutput(
RequestCommandOutputResult::Denylisted {
command: command.clone(),
},
);
ctx.emit(BlocklistAIActionExecutorEvent::ExecutingAction {
action_id: action_id.clone(),
conversation_id,
});
ctx.emit(BlocklistAIActionExecutorEvent::FinishedAction {
result: Arc::new(AIAgentActionResult {
id: action_id,
task_id: action.task_id.clone(),
result,
}),
conversation_id,
cancellation_reason: None,
});
return TryExecuteResult::ExecutedSync;
}
}
crate::ai::tool_diagnostics::tool_debug!(
"try_to_execute_action: EXECUTING action_id={:?}, type={:?}",
action.id,
std::mem::discriminant(&action.action)
);
let action_key = (conversation_id, action.id.clone());
let is_restored = self.restored_action_ids.remove(&action_key);
let action_clone = action.clone();
let execution = match &action.action {
AIAgentActionType::RequestCommandOutput { .. }
| AIAgentActionType::WriteToLongRunningShellCommand { .. }
| AIAgentActionType::ReadShellCommandOutput { .. }
| AIAgentActionType::TransferShellCommandControlToUser { .. } => self
.shell_command_executor
.update(ctx, |executor, ctx| executor.execute(input, ctx))
.into(),
AIAgentActionType::InitProject => {
ctx.emit(BlocklistAIActionExecutorEvent::InitProject(action.id));
ActionExecution::<()>::Sync(AIAgentActionResultType::InitProject).into()
}
AIAgentActionType::OpenCodeReview => {
ctx.emit(BlocklistAIActionExecutorEvent::OpenCodeReview(action.id));
ActionExecution::<()>::Sync(AIAgentActionResultType::OpenCodeReview).into()
}
AIAgentActionType::InsertCodeReviewComments {
repo_path,
comments,
base_branch,
} => {
if FeatureFlag::PRCommentsSlashCommand.is_enabled() {
ctx.emit(BlocklistAIActionExecutorEvent::InsertCodeReviewComments {
action_id: action.id,
repo_path: repo_path.clone(),
comments: comments.clone(),
base_branch: base_branch.clone(),
});
}
ActionExecution::<()>::Sync(AIAgentActionResultType::InsertReviewComments(
InsertReviewCommentsResult::Success {
repo_path: repo_path.to_string_lossy().to_string(),
},
))
.into()
}
AIAgentActionType::ReadFiles(..) => self
.read_files_executor
.update(ctx, |executor, ctx| executor.execute(input, ctx))
.into(),
AIAgentActionType::UploadArtifact(..) => self
.upload_artifact_executor
.update(ctx, |executor, ctx| executor.execute(input, ctx)),
AIAgentActionType::SearchCodebase(..) => self
.search_codebase_executor
.update(ctx, |executor, ctx| executor.execute(input, ctx))
.into(),
AIAgentActionType::Grep { .. } => self
.grep_executor
.update(ctx, |executor, ctx| executor.execute(input, ctx))
.into(),
AIAgentActionType::FileGlob { .. } | AIAgentActionType::FileGlobV2 { .. } => self
.file_glob_executor
.update(ctx, |executor, ctx| executor.execute(input, ctx))
.into(),
AIAgentActionType::CallMCPTool { .. } => self
.call_mcp_tool_executor
.update(ctx, |executor, ctx| executor.execute(input, ctx))
.into(),
AIAgentActionType::ReadMCPResource { .. } => self
.read_mcp_resource_executor
.update(ctx, |executor, ctx| executor.execute(input, ctx))
.into(),
// Normally, requested file edits are not handled by the executor. However, when performing a task autonomously,
// the executor is responsible for auto-approving diffs.
AIAgentActionType::RequestFileEdits { .. } => self
.request_file_edits_executor
.update(ctx, |executor, ctx| executor.execute(input, ctx))
.into(),
AIAgentActionType::SuggestNewConversation { .. } => self
.suggest_new_conversation_executor
.update(ctx, |executor, ctx| executor.execute(input, ctx))
.into(),
AIAgentActionType::SuggestPrompt { .. } => self
.suggest_prompt_executor
.update(ctx, |executor, ctx| executor.execute(input, ctx))
.into(),
AIAgentActionType::ReadDocuments(_) => {
if action.tool_name.as_deref() == Some("notebook") {
self.notebook_executor
.update(ctx, |executor, ctx| executor.execute_read(input, ctx))
.into()
} else {
self.read_documents_executor
.update(ctx, |executor, ctx| executor.execute(input, ctx))
.into()
}
}
AIAgentActionType::EditDocuments(_) => {
if action.tool_name.as_deref() == Some("notebook") {
self.notebook_executor
.update(ctx, |executor, ctx| executor.execute_edit(input, ctx))
.into()
} else {
self.edit_documents_executor
.update(ctx, |executor, ctx| executor.execute(input, ctx))
.into()
}
}
AIAgentActionType::CreateDocuments(_) => {
if action.tool_name.as_deref() == Some("notebook") {
self.notebook_executor
.update(ctx, |executor, ctx| executor.execute_create(input, ctx))
.into()
} else {
self.create_documents_executor
.update(ctx, |executor, ctx| {
executor.execute(input, conversation_id, ctx)
})
.into()
}
}
AIAgentActionType::UseComputer(_) => self
.use_computer_executor
.update(ctx, |executor, ctx| executor.execute(input, ctx))
.into(),
AIAgentActionType::RequestComputerUse(_) => self
.request_computer_use_executor
.update(ctx, |executor, ctx| executor.execute(input, ctx))
.into(),
AIAgentActionType::ReadSkill(_) => self
.read_skill_executor
.update(ctx, |executor, ctx| executor.execute(input, ctx))
.into(),
AIAgentActionType::FetchConversation { .. } => self
.fetch_conversation_executor
.update(ctx, |executor, ctx| executor.execute(input, ctx))
.into(),
AIAgentActionType::StartAgent { .. } => self
.start_agent_executor
.update(ctx, |executor, ctx| executor.execute(input, ctx))
.into(),
AIAgentActionType::SendMessageToAgent { .. } => self
.send_message_executor
.update(ctx, |executor, ctx| executor.execute(input, ctx)),
AIAgentActionType::AskUserQuestion { .. } => self
.ask_user_question_executor
.update(ctx, |executor, ctx| executor.execute(input, ctx))
.into(),
AIAgentActionType::RunAgents(_) => self
.run_agents_executor
.update(ctx, |executor, ctx| executor.execute(input, ctx))
.into(),
AIAgentActionType::WaitForEvents { .. } => self
.wait_for_events_executor
.update(ctx, |executor, ctx| executor.execute(input, ctx))
.into(),
};
let action_id = action_clone.id.clone();
crate::ai::tool_diagnostics::tool_debug!(
"try_to_execute_action: execution result type={:?} for action_id={:?}",
match &execution {
AnyActionExecution::NotReady => "NotReady",
AnyActionExecution::InvalidAction => "InvalidAction",
AnyActionExecution::Async { .. } => "Async",
AnyActionExecution::Sync(_) => "Sync",
},
action_id
);
match execution {
AnyActionExecution::NotReady => {
crate::ai::tool_diagnostics::tool_debug!(
"try_to_execute_action: NOT READY - action_id={:?}",
action_id
);
TryExecuteResult::NotExecuted {
reason: NotExecutedReason::NotReady,
action: Box::new(action_clone),
}
}
AnyActionExecution::InvalidAction => {
log::error!(
"try_to_execute_action: invalid action, action_id={:?}",
action_id
);
debug_assert!(false, "Tried to execute AIAgentAction with wrong executor.");
TryExecuteResult::NotExecuted {
reason: NotExecutedReason::NotReady,
action: Box::new(action_clone),
}
}
AnyActionExecution::Async {
execute_future,
on_complete,
} => {
self.async_executing_actions.insert(
conversation_id,
AsyncExecutingAction {
action: action_clone,
},
);
if !is_restored {
ctx.emit(BlocklistAIActionExecutorEvent::ExecutingAction {
action_id: action_id.clone(),
conversation_id,
});
}
crate::ai::tool_diagnostics::tool_debug!(
"try_to_execute_action: spawning ASYNC execution for action_id={:?}",
action_id
);
ctx.spawn(execute_future, move |me, result, ctx| {
let Some(running) = me
.async_executing_actions
.remove(conversation_id, &action_id)
else {
log::warn!("try_to_execute_action: async action completed but not found in executing map, conversation_id={conversation_id}, action_id={action_id:?}");
return;
};
let result = on_complete(result, ctx);
crate::ai::tool_diagnostics::tool_debug!(
"try_to_execute_action: ASYNC action COMPLETED action_id={:?}, result_type={:?}",
action_id,
std::mem::discriminant(&result)
);
ctx.emit(BlocklistAIActionExecutorEvent::FinishedAction {
result: Arc::new(AIAgentActionResult {
id: action_id,
task_id: running.action.task_id,
result,
}),
conversation_id,
cancellation_reason: None,
});
});
TryExecuteResult::ExecutedAsync
}
AnyActionExecution::Sync(action_result) => {
if !is_restored {
ctx.emit(BlocklistAIActionExecutorEvent::ExecutingAction {
action_id: action_id.clone(),
conversation_id,
});
}
ctx.emit(BlocklistAIActionExecutorEvent::FinishedAction {
result: Arc::new(AIAgentActionResult {
id: action_id,
task_id: action.task_id,
result: action_result,
}),
conversation_id,
cancellation_reason: None,
});
TryExecuteResult::ExecutedSync
}
}
}
pub fn can_autoexecute_action(
&self,
action: &AIAgentAction,
conversation_id: AIConversationId,
ctx: &mut ModelContext<Self>,
) -> bool {
self.should_autoexecute(
ExecuteActionInput {
action,
conversation_id,
},
ctx,
)
}
pub fn cancel_running_async_action(
&mut self,
conversation_id: AIConversationId,
action_id: &AIAgentActionId,
reason: Option<CancellationReason>,
ctx: &mut ModelContext<Self>,
) {
// A viewer should not be able to cancel an action.
if self.is_shared_session_viewer() {
return;
}
if self
.async_executing_actions
.get(conversation_id, action_id)
.is_some_and(|running| {
matches!(
running.action.action,
AIAgentActionType::RequestCommandOutput { .. }
)
})
{
let termination_requested = self.shell_command_executor.update(ctx, |executor, ctx| {
executor.cancel_execution(action_id, ctx)
});
if termination_requested {
// Keep the action in flight until block completion proves the process stopped.
// Its normal async completion will report the actual terminal exit status.
return;
}
}
if let Some(running) = self
.async_executing_actions
.remove(conversation_id, action_id)
{
let action_kind = AIAgentActionTypeDiscriminants::from(&running.action.action);
crate::ai::tool_diagnostics::tool_debug!(
"Canceling running async action of type {action_kind:?} action_id={action_id:?}, reason={reason:?}"
);
if let Some(backtrace) = crate::ai::tool_diagnostics::capture_backtrace() {
log::debug!("Running action cancellation backtrace:\n{backtrace}");
}
if running.is_shell_command_action()
&& !matches!(
running.action.action,
AIAgentActionType::RequestCommandOutput { .. }
)
{
self.shell_command_executor.update(ctx, |executor, ctx| {
executor.cancel_execution(&running.action.id, ctx);
});
} else if matches!(running.action.action, AIAgentActionType::SearchCodebase(..)) {
self.search_codebase_executor.update(ctx, |executor, ctx| {
executor.cancel_execution(&running.action.id, ctx);
});
} else if matches!(running.action.action, AIAgentActionType::RunAgents(..)) {
self.run_agents_executor.update(ctx, |executor, ctx| {
executor.cancel_execution(conversation_id, &running.action.id, ctx);
});
} else if matches!(running.action.action, AIAgentActionType::StartAgent { .. }) {
self.start_agent_executor.update(ctx, |executor, _| {
executor.cancel_execution(conversation_id, &running.action.id);
});
} else if let AIAgentActionType::WaitForEvents { tool_call_id, .. } =
&running.action.action
{
// Drop the executor's pending entry; the shared cancel
// path emits FinishedAction(Cancelled).
let tool_call_id = tool_call_id.clone();
self.wait_for_events_executor.update(ctx, |executor, _| {
executor.cancel_execution(&tool_call_id);
});
}
ctx.emit(BlocklistAIActionExecutorEvent::FinishedAction {
result: Arc::new(AIAgentActionResult {
id: running.action.id.clone(),
task_id: running.action.task_id,
result: running.action.action.cancelled_result(),
}),
conversation_id,
cancellation_reason: reason,
});
}
}
pub fn cancel_all_running_async_actions_for_conversation(
&mut self,
conversation_id: AIConversationId,
reason: Option<CancellationReason>,
ctx: &mut ModelContext<Self>,
) {
let action_ids = self
.async_executing_actions
.0
.iter()
.filter_map(|((running_conversation_id, action_id), _)| {
(*running_conversation_id == conversation_id).then_some(action_id.clone())
})
.collect::<Vec<_>>();
for action_id in action_ids {
self.cancel_running_async_action(conversation_id, &action_id, reason, ctx);
}
}
/// Removes all per-conversation retained state from sub-executors.
/// Called when a conversation is removed or cleared from memory.
pub fn cleanup_conversation(
&mut self,
conversation_id: AIConversationId,
ctx: &mut ModelContext<Self>,
) {
self.restored_action_ids
.retain(|(conv_id, _)| *conv_id != conversation_id);
self.async_executing_actions
.0
.retain(|(conv_id, _), _| *conv_id != conversation_id);
self.run_agents_executor.update(ctx, |executor, _| {
executor.cleanup_conversation(conversation_id);
});
}
fn should_autoexecute(&self, input: ExecuteActionInput, ctx: &mut ModelContext<Self>) -> bool {
if self
.restored_action_ids
.contains(&(input.conversation_id, input.action.id.clone()))
|| cfg!(feature = "bedrock_smoke_test")
{
return true;
}
match input.action.action {
AIAgentActionType::RequestCommandOutput { .. }
| AIAgentActionType::WriteToLongRunningShellCommand { .. }
| AIAgentActionType::ReadShellCommandOutput { .. }
| AIAgentActionType::TransferShellCommandControlToUser { .. } => self
.shell_command_executor
.update(ctx, |executor, ctx| executor.should_autoexecute(input, ctx)),
AIAgentActionType::ReadFiles(_) => self
.read_files_executor
.update(ctx, |executor, ctx| executor.should_autoexecute(input, ctx)),
AIAgentActionType::UploadArtifact(_) => self
.upload_artifact_executor
.update(ctx, |executor, ctx| executor.should_autoexecute(input, ctx)),
AIAgentActionType::SearchCodebase(_) => self
.search_codebase_executor
.update(ctx, |executor, ctx| executor.should_autoexecute(input, ctx)),
AIAgentActionType::RequestFileEdits { .. } => self
.request_file_edits_executor
.update(ctx, |executor, ctx| executor.should_autoexecute(input, ctx)),
AIAgentActionType::Grep { .. } => self
.grep_executor
.update(ctx, |executor, ctx| executor.should_autoexecute(input, ctx)),
AIAgentActionType::FileGlob { .. } | AIAgentActionType::FileGlobV2 { .. } => self
.file_glob_executor
.update(ctx, |executor, ctx| executor.should_autoexecute(input, ctx)),
AIAgentActionType::CallMCPTool { .. } => self
.call_mcp_tool_executor
.update(ctx, |executor, ctx| executor.should_autoexecute(input, ctx)),
AIAgentActionType::ReadMCPResource { .. } => self
.read_mcp_resource_executor
.update(ctx, |executor, ctx| executor.should_autoexecute(input, ctx)),
AIAgentActionType::InitProject => true,
AIAgentActionType::OpenCodeReview => true,
AIAgentActionType::InsertCodeReviewComments { .. } => true,
AIAgentActionType::SuggestNewConversation { .. } => self
.suggest_new_conversation_executor
.update(ctx, |executor, ctx| executor.should_autoexecute(input, ctx)),
AIAgentActionType::SuggestPrompt { .. } => self
.suggest_prompt_executor
.update(ctx, |executor, ctx| executor.should_autoexecute(input, ctx)),
AIAgentActionType::ReadDocuments(_) => {
if input.action.tool_name.as_deref() == Some("notebook") {
self.notebook_executor
.update(ctx, |executor, ctx| executor.should_autoexecute(input, ctx))
} else {
self.read_documents_executor
.update(ctx, |executor, ctx| executor.should_autoexecute(input, ctx))
}
}
AIAgentActionType::EditDocuments(_) => {
if input.action.tool_name.as_deref() == Some("notebook") {
self.notebook_executor
.update(ctx, |executor, ctx| executor.should_autoexecute(input, ctx))
} else {
self.edit_documents_executor
.update(ctx, |executor, ctx| executor.should_autoexecute(input, ctx))
}
}
AIAgentActionType::CreateDocuments(_) => {
if input.action.tool_name.as_deref() == Some("notebook") {
self.notebook_executor
.update(ctx, |executor, ctx| executor.should_autoexecute(input, ctx))
} else {
self.create_documents_executor
.update(ctx, |executor, ctx| executor.should_autoexecute(input, ctx))
}
}
AIAgentActionType::UseComputer(_) => self
.use_computer_executor
.update(ctx, |executor, ctx| executor.should_autoexecute(input, ctx)),
AIAgentActionType::RequestComputerUse(_) => self
.request_computer_use_executor
.update(ctx, |executor, ctx| executor.should_autoexecute(input, ctx)),
AIAgentActionType::ReadSkill(_) => self
.read_skill_executor
.update(ctx, |executor, ctx| executor.should_autoexecute(input, ctx)),
AIAgentActionType::FetchConversation { .. } => self
.fetch_conversation_executor
.update(ctx, |executor, ctx| executor.should_autoexecute(input, ctx)),
AIAgentActionType::StartAgent { .. } => self
.start_agent_executor
.update(ctx, |executor, ctx| executor.should_autoexecute(input, ctx)),
AIAgentActionType::SendMessageToAgent { .. } => self
.send_message_executor
.update(ctx, |executor, ctx| executor.should_autoexecute(input, ctx)),
AIAgentActionType::AskUserQuestion { .. } => self
.ask_user_question_executor
.update(ctx, |executor, ctx| executor.should_autoexecute(input, ctx)),
AIAgentActionType::RunAgents(_) => self
.run_agents_executor
.update(ctx, |executor, ctx| executor.should_autoexecute(input, ctx)),
AIAgentActionType::WaitForEvents { .. } => self
.wait_for_events_executor
.update(ctx, |executor, ctx| executor.should_autoexecute(input, ctx)),
}
}
fn is_shared_session_viewer(&self) -> bool {
self.terminal_model.lock().is_shared_session_viewer()
}
}
impl Entity for BlocklistAIActionExecutor {
type Event = BlocklistAIActionExecutorEvent;
}
pub enum BlocklistAIActionExecutorEvent {
/// Emitted when an action begins execution.
ExecutingAction {
action_id: AIAgentActionId,
conversation_id: AIConversationId,
},
/// Emitted when an action has finished.
FinishedAction {
result: Arc<AIAgentActionResult>,
conversation_id: AIConversationId,
/// The reason for cancellation, if this action was cancelled.
cancellation_reason: Option<CancellationReason>,
},
InitProject(AIAgentActionId),
OpenCodeReview(AIAgentActionId),
InsertCodeReviewComments {
action_id: AIAgentActionId,
repo_path: PathBuf,
comments: Vec<ai::agent::action::InsertReviewComment>,
base_branch: Option<String>,
},
}
/// Per-file byte limit for [`read_local_file_context`]. Binary files larger
/// than this are skipped; text files are truncated at this limit.
#[cfg(feature = "local_fs")]
const MAX_FILE_READ_BYTES: usize = 1_000_000;
/// The results of a [`read_local_file_context`] call.
#[derive(Debug, Clone, Serialize, Deserialize, Eq, PartialEq)]
pub struct ReadFileContextResult {
/// [`FileContext`] data for all files that could be read.
pub file_contexts: Vec<FileContext>,
/// Expected absolute paths of requested files that did not exist or could
/// not be read (e.g. binary files that exceed the size limit).
pub missing_files: Vec<String>,
}
/// Reads the content of the given files at the given `FileLocations`.
///
/// If any files do not exist, they are included in the `missing_files` field of the result.
///
/// Binary files larger than the per-file byte limit are skipped and reported as oversized.
/// Text files are truncated at the per-file limit via line streaming.
/// If `max_file_bytes` is provided, it overrides the default per-file limit
/// ([`MAX_FILE_READ_BYTES`]). Pass `None` to use the default.
/// If `max_batch_bytes` is provided, the cumulative content of all files is capped at that
/// budget; once exceeded, remaining files are reported as oversized.
#[cfg_attr(not(feature = "local_fs"), allow(unused_variables))]
pub async fn read_local_file_context(
file_names: &[FileLocations],
current_working_directory: Option<String>,
shell: Option<ShellLaunchData>,
max_file_bytes: Option<usize>,
max_batch_bytes: Option<usize>,
) -> anyhow::Result<ReadFileContextResult> {
#[cfg(not(feature = "local_fs"))]
return Err(anyhow::anyhow!(
"Can't read files when not on a local filesystem"
));
#[cfg(feature = "local_fs")]
{
let mut result = ReadFileContextResult {
file_contexts: Vec::new(),
missing_files: Vec::new(),
};
let mut batch_bytes_remaining = max_batch_bytes;
for file in file_names {
let absolute_file_path = PathBuf::from(host_native_absolute_path(
&file.name,
&shell,
&current_working_directory,
));
let metadata = match async_fs::metadata(&absolute_file_path).await {
Ok(m) => m,
Err(e) if e.kind() == std::io::ErrorKind::NotFound => {
result
.missing_files
.push(absolute_file_path.to_string_lossy().to_string());
continue;
}
Err(e) => return Err(anyhow::anyhow!(e)),
};
let last_modified = metadata.modified().ok();
let file_size = metadata.len() as usize;
let path_str = absolute_file_path.to_string_lossy().to_string();
// Effective byte budget: the tighter of per-file and remaining batch.
let per_file_limit = max_file_bytes.unwrap_or(MAX_FILE_READ_BYTES);
let effective_max = match batch_bytes_remaining {
Some(remaining) => per_file_limit.min(remaining),
None => per_file_limit,
};
// Decide text vs binary before opening the file. Extension-based
// detection alone is wrong for extensionless text files (e.g. shell
// scripts named `bundle`, `run`), so for the ambiguous case we fall
// back to content-based inspection of the first chunk. The binary
// path below still acts as a safety net via
// `TextFileReadResult::NotText` if the text reader trips on invalid
// UTF-8.
if !should_read_as_binary(&absolute_file_path).await {
match FileModel::read_text_file(
&absolute_file_path,
effective_max,
&file.lines,
last_modified,
)
.await?
{
TextFileReadResult::Segments {
segments,
bytes_read,
} => {
if let Some(remaining) = &mut batch_bytes_remaining {
*remaining = remaining.saturating_sub(bytes_read);
}
result
.file_contexts
.extend(segments.into_iter().map(|seg| FileContext {
file_name: seg.file_name,
content: AnyFileContent::StringContent(seg.content),
line_range: seg.line_range,
last_modified: seg.last_modified,
line_count: seg.line_count,
}));
continue;
}
TextFileReadResult::NotText => {
// Fall through to binary path below.
}
}
}
// Binary path (either detected as binary, or text reading failed).
match read_binary_file_context(
&absolute_file_path,
effective_max,
file_size,
last_modified,
)
.await?
{
BinaryFileReadResult::Context {
file_context,
bytes_read,
} => {
if let Some(remaining) = &mut batch_bytes_remaining {
*remaining = remaining.saturating_sub(bytes_read);
}
result.file_contexts.push(file_context);
}
BinaryFileReadResult::Missing => result.missing_files.push(path_str),
}
}
Ok(result)
}
}
/// Returns `true` if the file at `path` should be read via the binary code
/// path in [`read_local_file_context`], `false` if it should be read as text.
///
/// Uses extension-based detection as a fast path and falls back to content
/// inspection (reading the first 1 KiB) for extensionless files whose names
/// don't match any known text or binary pattern. Without the content-based
/// fallback, extensionless text files (e.g. shell scripts named `bundle`)
/// would be incorrectly classified as binary and returned to the agent as
/// raw bytes instead of UTF-8 text.
#[cfg(feature = "local_fs")]
async fn should_read_as_binary(path: &std::path::Path) -> bool {
// Fast path: extension/filename clearly indicates text.
if !is_binary_file(path) {
return false;
}
// Fast path: file has a known binary extension (e.g. `.png`, `.exe`).
if path.extension().is_some() {
return true;
}
// Extensionless file with an unknown basename. Inspect the first chunk of
// the file to decide. Treat open/read errors as binary so the binary path
// takes over and reports a consistent error.
is_file_content_binary_async(path).await
}
/// Async sibling of [`galaxy_util::file_type::is_file_content_binary`]. Reads
/// the first 1 KiB of `path` asynchronously and returns `true` if the content
/// looks binary according to [`is_buffer_binary`]. Returns `true` on any I/O
/// error so callers default to the binary code path. Kept local to this
/// module so `galaxy_util` doesn't need to grow an `async_fs` dependency.
#[cfg(feature = "local_fs")]
async fn is_file_content_binary_async(path: &std::path::Path) -> bool {
const CHUNK_SIZE: usize = 1024;
let Ok(mut file) = async_fs::File::open(path).await else {
return true;
};
let mut buffer = [0u8; CHUNK_SIZE];
let Ok(n) = file.read(&mut buffer).await else {
return true;
};
is_buffer_binary(&buffer[..n])
}
#[cfg(feature = "local_fs")]
enum BinaryFileReadResult {
/// Successfully read as binary.
Context {
file_context: FileContext,
bytes_read: usize,
},
/// File doesn't exist, exceeds the size limit, or couldn't be processed.
Missing,
}
/// Reads a binary file, applying image processing when applicable.
#[cfg(feature = "local_fs")]
async fn read_binary_file_context(
path: &std::path::Path,
max_bytes: usize,
file_size: usize,
last_modified: Option<std::time::SystemTime>,
) -> anyhow::Result<BinaryFileReadResult> {
if file_size > max_bytes {
return Ok(BinaryFileReadResult::Missing);
}
let content = match read_file_as_binary(path).await {
Ok(content) => content,
Err(FileLoadError::DoesNotExist) => return Ok(BinaryFileReadResult::Missing),
Err(FileLoadError::IOError(e)) => return Err(anyhow::anyhow!(e)),
};
let mime_type = from_path(path).first_or_octet_stream().to_string();
let processed_content = if is_supported_image_mime_type(&mime_type) {
match process_image_for_agent(&content) {
ProcessImageResult::Success { data } => Some(data),
ProcessImageResult::TooLarge => {
log::warn!("Image file too large after processing: {}", path.display());
return Ok(BinaryFileReadResult::Missing);
}
ProcessImageResult::Error(err) => {
log::warn!("Error processing image file {}: {err:?}", path.display());
return Ok(BinaryFileReadResult::Missing);
}
}
} else {
None
};
let final_content = processed_content.unwrap_or(content);
if final_content.len() > max_bytes {
return Ok(BinaryFileReadResult::Missing);
}
let bytes_read = final_content.len();
Ok(BinaryFileReadResult::Context {
file_context: FileContext::new(
path.to_string_lossy().to_string(),
AnyFileContent::BinaryContent(final_content),
None,
last_modified,
),
bytes_read,
})
}
fn build_is_file_path_command(path: &str, shell_type: ShellType) -> String {
let escaped_path = shell_quote_arg(path, shell_type);
if shell_type == ShellType::PowerShell {
format!("if (Test-Path -PathType Leaf {escaped_path}) {{ exit 0 }} else {{ exit 1 }}")
} else {
format!("test -f {escaped_path}")
}
}
fn build_is_git_repository_command(absolute_path: &str, shell_type: ShellType) -> String {
format!(
"git -C {} rev-parse",
shell_quote_arg(absolute_path, shell_type)
)
}
/// Returns true if the given path is a regular file on the session's filesystem.
/// Uses native metadata for compatible local sessions and a shell probe otherwise.
async fn is_file_path(path: &str, session: &Session) -> bool {
#[cfg(feature = "local_fs")]
if session.is_local() && !session.is_wsl() && !session.is_msys2() {
return async_fs::metadata(path)
.await
.is_ok_and(|metadata| metadata.is_file());
}
let command = build_is_file_path_command(path, session.shell().shell_type());
session
.execute_command(&command, None, None, ExecuteCommandOptions::default())
.await
.map(|output| output.success())
.unwrap_or(false)
}
/// Returns true if git is installed and the given path is in a git repository.
async fn is_git_repository(absolute_path: &str, session: &Session) -> anyhow::Result<bool> {
let git_command = build_is_git_repository_command(absolute_path, session.shell().shell_type());
let command_output = session
.execute_command(
git_command.as_str(),
None,
None,
ExecuteCommandOptions::default(),
)
.await?;
Ok(command_output.success())
}
async fn is_git_repository_cached(
absolute_path: &str,
session: &Session,
cache: &GitRepositoryCache,
) -> anyhow::Result<bool> {
if let Some(is_repository) = cache.get(session.id(), absolute_path) {
return Ok(is_repository);
}
let is_repository = is_git_repository(absolute_path, session).await?;
cache.insert(session.id(), absolute_path.to_string(), is_repository);
Ok(is_repository)
}
fn get_server_output_id(
conversation_id: AIConversationId,
ctx: &mut AppContext,
) -> Option<ServerOutputId> {
BlocklistAIHistoryModel::as_ref(ctx)
.conversation(&conversation_id)?
.latest_exchange()?
.output_status
.server_output_id()
}
#[cfg(feature = "local_fs")]
async fn read_file_as_binary(file_path: &std::path::Path) -> Result<Vec<u8>, FileLoadError> {
if !FileModel::file_exists(file_path).await {
return Err(FileLoadError::DoesNotExist);
}
async_fs::read(file_path).await.map_err(FileLoadError::from)
}
#[cfg(test)]
mod async_executing_action_tests {
use super::*;
use crate::ai::agent::task::TaskId;
fn action(id: &str, task_id: &str) -> AIAgentAction {
AIAgentAction {
id: AIAgentActionId::from(id.to_owned()),
action: AIAgentActionType::InitProject,
task_id: TaskId::new(task_id.to_owned()),
requires_result: true,
tool_name: Some("init_project".to_owned()),
}
}
#[test]
fn duplicate_action_ids_can_execute_concurrently_in_different_conversations() {
let first_conversation = AIConversationId::new();
let second_conversation = AIConversationId::new();
let duplicate_id = AIAgentActionId::from("duplicate".to_owned());
let mut running = AsyncExecutingActions::default();
running.insert(
first_conversation,
AsyncExecutingAction {
action: action("duplicate", "first-task"),
},
);
running.insert(
second_conversation,
AsyncExecutingAction {
action: action("duplicate", "second-task"),
},
);
assert_eq!(running.0.len(), 2);
assert_eq!(
running
.get(first_conversation, &duplicate_id)
.unwrap()
.action
.task_id,
TaskId::new("first-task".to_owned())
);
assert_eq!(
running
.get(second_conversation, &duplicate_id)
.unwrap()
.action
.task_id,
TaskId::new("second-task".to_owned())
);
}
#[test]
fn duplicate_action_completion_and_cancellation_remove_only_the_matching_conversation() {
let first_conversation = AIConversationId::new();
let second_conversation = AIConversationId::new();
let duplicate_id = AIAgentActionId::from("duplicate".to_owned());
let mut running = AsyncExecutingActions::default();
running.insert(
first_conversation,
AsyncExecutingAction {
action: action("duplicate", "first-task"),
},
);
running.insert(
second_conversation,
AsyncExecutingAction {
action: action("duplicate", "second-task"),
},
);
let completed = running.remove(first_conversation, &duplicate_id).unwrap();
assert_eq!(
completed.action.task_id,
TaskId::new("first-task".to_owned())
);
assert!(running.get(second_conversation, &duplicate_id).is_some());
let cancelled = running.remove(second_conversation, &duplicate_id).unwrap();
assert_eq!(
cancelled.action.task_id,
TaskId::new("second-task".to_owned())
);
assert!(running.0.is_empty());
}
}
#[cfg(all(test, feature = "local_fs"))]
#[path = "execute_tests.rs"]
mod tests;
Reference in New Issue View Git Blame Copy Permalink