public/galaxy
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Restore agent tool panes and require explicit session resume

Browse Source 
Persist direct-provider tool calls, results, display details, and turn identities so saved conversations rebuild individual panes. Keep restored runs dormant until explicitly resumed and bind queued shell commands to their actual execution blocks.
This commit is contained in:
rkw6086
2026-09-20 17:11:41 -05:00
parent d7fbc09bcf
commit 35ff43de7f
21 changed files with 1011 additions and 88 deletions
Download Patch File Download Diff File Expand all files Collapse all files
app/src/terminal/input_tests.rs
+3
View File
@@ -2457,6 +2457,7 @@ fn test_histignorespace_support_in_zsh() {
(terminal.model.clone(), terminal.sessions_model().clone())
});
let block_id = model.lock().active_block_id().clone();
app.update(|ctx| {
update_command_history(
&ExecuteCommandEvent {
@@ -2467,6 +2468,7 @@ fn test_histignorespace_support_in_zsh() {
should_add_command_to_history: true,
source: CommandExecutionSource::User,
},
&block_id,
&model,
None,
&sessions,
@@ -2482,6 +2484,7 @@ fn test_histignorespace_support_in_zsh() {
should_add_command_to_history: true,
source: CommandExecutionSource::User,
},
&block_id,
&model,
None,
&sessions,
app/src/terminal/model/terminal_model.rs
+4 -1
View File
@@ -3022,7 +3022,10 @@ impl ansi::Handler for TerminalModel {
self.is_receiving_in_band_command_output = IsReceivingInBandCommandOutput::No;
}
IsReceivingInBandCommandOutput::No => {
log::warn!("Received 'end_in_band_command_output' while not expecting to read in-band command output.");
// Command dispatch calls this defensively even when no in-band output is active.
if from_osc_sequence {
log::warn!("Received 'end_in_band_command_output' while not expecting to read in-band command output.");
}
}
}
app/src/terminal/view.rs
+14 -11
View File
@@ -1763,6 +1763,7 @@ pub enum Event {
size_update: SizeUpdate,
},
ExecuteCommand(ExecuteCommandEvent),
CancelQueuedAgentCommand(AIAgentActionId),
BlockStarted {
is_for_in_band_command: bool,
},
@@ -7695,17 +7696,8 @@ impl TerminalView {
}
}
}
let block_id = model.active_block_id().clone();
drop(model);
self.cli_subagent_controller.update(ctx, |controller, _| {
controller.track_requested_command(
&block_id,
parent_conversation_id,
action_id,
);
});
ctx.emit(Event::ExecuteCommand(ExecuteCommandEvent {
command,
session_id,
@@ -7721,8 +7713,6 @@ impl TerminalView {
});
}
self.schedule_command_monitor_start(block_id, ctx);
if let Some(metadata) = workflow_telem_metadata {
send_telemetry_from_ctx!(TelemetryEvent::WorkflowExecuted(metadata), ctx);
}
@@ -7732,6 +7722,7 @@ impl TerminalView {
self.write_agent_bytes_to_pty(input.to_vec(), mode, ctx);
}
ShellCommandExecutorEvent::CancelExecution { action_id } => {
ctx.emit(Event::CancelQueuedAgentCommand(action_id.clone()));
// We need to manually invoke ctrl-c to terminate the running command because the
// user's ctrl-c was directed to the AIBlock instead of the command's shell block.
let is_exact_active_command = self
@@ -12185,6 +12176,15 @@ impl TerminalView {
return;
}
self.did_notify_long_running = false;
let is_agent_command = self
.model
.lock()
.block_list()
.block_with_id(block_id)
.is_some_and(|block| block.requested_command_action_id().is_some());
if is_agent_command {
self.schedule_command_monitor_start(block_id.clone(), ctx);
}
// Snapshot the prompt state as of when the command began executing.
// Commands may themselves affect the prompt (if running `git checkout`), for
@@ -26257,6 +26257,9 @@ impl PtyIntentEvent for Event {
}),
Event::Resize { size_update } => Some(PtyIntent::Resize(*size_update)),
Event::ExecuteCommand(event) => Some(PtyIntent::ExecuteCommand(event.clone())),
Event::CancelQueuedAgentCommand(action_id) => {
Some(PtyIntent::CancelQueuedAgentCommand(action_id.clone()))
}
Event::RunNativeShellCompletions {
buffer_text,
results_tx,
app/src/terminal/writeable_pty/command_history.rs
+5 -1
View File
@@ -5,19 +5,23 @@ use galaxyui::{AppContext, ModelHandle, SingletonEntity};
use parking_lot::FairMutex;
use crate::persistence::{ModelEvent, StartedCommandMetadata};
use crate::terminal::model::block::BlockId;
use crate::terminal::model::session::Sessions;
use crate::terminal::view::ExecuteCommandEvent;
use crate::terminal::{History, HistoryEntry, TerminalModel};
pub fn update_command_history(
event: &ExecuteCommandEvent,
block_id: &BlockId,
model: &Arc<FairMutex<TerminalModel>>,
model_event_sender: Option<&SyncSender<ModelEvent>>,
sessions: &ModelHandle<Sessions>,
ctx: &mut AppContext,
) {
let model = model.lock();
let active_block = model.block_list().active_block();
let Some(active_block) = model.block_list().block_with_id(block_id) else {
return;
};
let session_id = event.session_id;
let Some(session) = sessions.as_ref(ctx).get(session_id) else {
return;
app/src/terminal/writeable_pty/pty_controller.rs
+91 -38
View File
@@ -10,10 +10,11 @@ use parking_lot::FairMutex;
use thiserror::Error;
use super::Message;
use crate::ai::agent::AIAgentPtyWriteMode;
use crate::ai::agent::{AIAgentActionId, AIAgentPtyWriteMode};
use crate::terminal::input::CommandExecutionSource;
use crate::terminal::line_editor_status::{LineEditorStatus, LineEditorStatusEvent};
use crate::terminal::model::ansi::Handler;
use crate::terminal::model::block::BlockId;
use crate::terminal::model::completions::ShellCompletion;
use crate::terminal::model::escape_sequences;
use crate::terminal::model::session::{
@@ -21,7 +22,7 @@ use crate::terminal::model::session::{
};
use crate::terminal::model_events::{AnsiHandlerEvent, ModelEvent, ModelEventDispatcher};
use crate::terminal::shell::ShellType;
use crate::terminal::view::LINEFEED_REGEX;
use crate::terminal::view::{ExecuteCommandEvent, LINEFEED_REGEX};
#[cfg(not(target_family = "wasm"))]
use crate::terminal::writeable_pty::bootstrap_file::{permanent_bootstrap_file, TempBootstrapFile};
use crate::terminal::{bootstrap, SizeUpdate, TerminalModel};
@@ -46,6 +47,7 @@ enum PtyWrite {
in_band_command_id: Option<String>,
/// If 'some', the given callback is called right before the bytes are written to the PTY.
before_write_fn: Option<Box<dyn Fn() + Send + 'static>>,
execution: Option<Box<ExecuteCommandEvent>>,
},
Bytes {
/// The bytes to be written.
@@ -298,7 +300,17 @@ impl<T: EventLoopSender> PtyController<T> {
cancel_tx: Sender<InBandCommandCancelledEvent>,
ctx: &mut ModelContext<Self>,
) {
if self.is_user_command_executing {
if self.is_user_command_executing
|| self.pending_writes.iter().any(|write| {
matches!(
write,
PtyWrite::Command {
execution: Some(_),
..
}
)
})
{
// Send blocking should be okay b/c this is an unbound channel
if let Err(err) = block_on(cancel_tx.send(InBandCommandCancelledEvent { command_id })) {
log::warn!("Pty Controller failed to cancel in band command: {err:?}");
@@ -311,6 +323,7 @@ impl<T: EventLoopSender> PtyController<T> {
command: command.to_owned(),
shell_type,
in_band_command_id: Some(command_id),
execution: None,
before_write_fn: Some(Box::new(move || {
let mut terminal_model = terminal_model.lock();
terminal_model
@@ -510,47 +523,19 @@ impl<T: EventLoopSender> PtyController<T> {
/// sequences.
pub fn write_command(
&mut self,
command: &str,
event: &ExecuteCommandEvent,
shell_type: ShellType,
source: CommandExecutionSource,
ctx: &mut ModelContext<Self>,
) {
{
let mut model = self.terminal_model.lock();
// Explicitly start the block now that the command is executed.
match source {
CommandExecutionSource::AI { metadata } => {
model.start_command_execution_with_ai_metadata(metadata)
}
CommandExecutionSource::SharedSession {
participant_id,
ai_metadata,
..
} => model.start_command_execution_for_shared_session(participant_id, ai_metadata),
CommandExecutionSource::User | CommandExecutionSource::QueuedCommand => {
model.start_command_execution()
}
CommandExecutionSource::EnvVarCollection { metadata } => {
model.start_command_execution_from_env_var_collection(metadata)
}
}
// Ensure that the `TerminalModel` doesn't interpret any of the PTY output from the
// following commands as in-band command output. If the in-band command output is not
// currently being received by the `TerminalModel`, this is a no-op.
model.end_in_band_command_output(false);
}
self.pending_writes.clear();
self.is_user_command_executing = true;
// Send the write to the PTY event loop.
let write = PtyWrite::Command {
command: command.to_owned(),
command: event.command.clone(),
shell_type,
in_band_command_id: None,
before_write_fn: None,
execution: Some(Box::new(event.clone())),
};
if self.can_write_to_pty(ctx) {
// Cancel the async writer task and clear the async write queue.
@@ -561,6 +546,28 @@ impl<T: EventLoopSender> PtyController<T> {
}
}
/// Cancel only a command that has not yet been dispatched to the shell.
pub fn cancel_queued_agent_command(&mut self, action_id: &AIAgentActionId) {
self.pending_writes.retain(|write| {
let PtyWrite::Command {
execution: Some(event),
..
} = write
else {
return true;
};
let metadata = match &event.source {
CommandExecutionSource::AI { metadata } => Some(metadata),
CommandExecutionSource::SharedSession { ai_metadata, .. } => ai_metadata.as_ref(),
CommandExecutionSource::User
| CommandExecutionSource::QueuedCommand
| CommandExecutionSource::EnvVarCollection { .. } => None,
};
!metadata
.is_some_and(|metadata| metadata.requested_command_action_id() == Some(action_id))
});
}
/// Synchronously writes the EOT (End-of-Transmission) char to the PTY.
pub fn write_end_of_transmission_char(&mut self, ctx: &mut ModelContext<Self>) {
self.write_bytes(&[escape_sequences::C0::EOT][..], ctx);
@@ -630,11 +637,12 @@ impl<T: EventLoopSender> PtyController<T> {
/// If the write corresponds to a command, this also calls
/// [`LineEditorStatus::did_execute_command()`].
fn send_write_to_event_loop(&mut self, write: PtyWrite, ctx: &mut ModelContext<Self>) {
let (bytes_to_write, is_for_command, on_write_fn) = match write {
let (bytes_to_write, is_for_command, on_write_fn, execution) = match write {
PtyWrite::Command {
command,
shell_type,
before_write_fn: on_write_fn,
execution,
..
} => (
Cow::Owned(bytes_to_execute_command(
@@ -644,13 +652,14 @@ impl<T: EventLoopSender> PtyController<T> {
)),
true,
on_write_fn,
execution,
),
PtyWrite::AgentInput { bytes, mode } => {
let decorated_bytes =
mode.decorate_bytes(bytes.into_owned(), self.is_bracketed_paste_enabled);
(decorated_bytes.into(), false, None)
(decorated_bytes.into(), false, None, None)
}
PtyWrite::Bytes { bytes } => (bytes, false, None),
PtyWrite::Bytes { bytes } => (bytes, false, None, None),
PtyWrite::RunNativeShellCompletions(state) => {
self.in_flight_native_completions_state = Some(state);
@@ -658,7 +667,7 @@ impl<T: EventLoopSender> PtyController<T> {
// then wait for an OSC-based signal from the shell before we
// send the text that needs to be completed.
let bytes = vec![0x19_u8];
(bytes.into(), false, None)
(bytes.into(), false, None, None)
}
};
@@ -674,6 +683,41 @@ impl<T: EventLoopSender> PtyController<T> {
});
}
// A queued command must not claim the block still owned by the previous command
// or an in-band request. Attach its identity only once the shell accepts this write.
if let Some(event) = execution {
let block_id = {
let mut model = self.terminal_model.lock();
match &event.source {
CommandExecutionSource::AI { metadata } => {
model.start_command_execution_with_ai_metadata(metadata.clone());
}
CommandExecutionSource::SharedSession {
participant_id,
ai_metadata,
..
} => model.start_command_execution_for_shared_session(
participant_id.clone(),
ai_metadata.clone(),
),
CommandExecutionSource::User | CommandExecutionSource::QueuedCommand => {
model.start_command_execution();
}
CommandExecutionSource::EnvVarCollection { metadata } => {
model.start_command_execution_from_env_var_collection(metadata.clone());
}
}
model
.block_list_mut()
.active_block_mut()
.set_cloud_workflow_state(event.workflow_id);
model.end_in_band_command_output(false);
model.active_block_id().clone()
};
self.is_user_command_executing = true;
ctx.emit(PtyControllerEvent::CommandStarted { event, block_id });
}
if let Some(on_write_fn) = on_write_fn {
on_write_fn();
}
@@ -718,6 +762,11 @@ impl<T: EventLoopSender> PtyController<T> {
}
pub enum PtyControllerEvent {
/// The command has acquired its terminal block and is being written to the shell.
CommandStarted {
event: Box<ExecuteCommandEvent>,
block_id: BlockId,
},
/// Emitted when the event loop thread has exited.
PtyDisconnected,
}
@@ -790,6 +839,10 @@ fn wrap_bytes_in_bracketed_paste(bytes: impl IntoIterator<Item = u8>) -> impl It
.chain(escape_sequences::BRACKETED_PASTE_END.iter().copied())
}
#[cfg(test)]
#[path = "pty_controller_queue_tests.rs"]
mod queue_tests;
#[cfg(test)]
#[path = "pty_controller_command_bytes_tests.rs"]
mod command_bytes_tests;
app/src/terminal/writeable_pty/pty_controller_queue_tests.rs
+131
View File
@@ -0,0 +1,131 @@
use galaxyui::App;
use super::*;
use crate::ai::agent::conversation::AIConversationId;
use crate::ai::agent::AIAgentActionId;
use crate::terminal::model::block::AgentInteractionMetadata;
struct RecordingEventLoopSender(Arc<FairMutex<Vec<Message>>>);
impl EventLoopSender for RecordingEventLoopSender {
fn send(&self, message: Message) -> Result<(), EventLoopSendError> {
self.0.lock().push(message);
Ok(())
}
}
#[test]
fn queued_agent_command_claims_only_the_block_at_dispatch() {
App::test((), |mut app| async move {
let sessions = app.add_model(|_| Sessions::new_for_test());
let (_events_tx, events_rx) = async_channel::unbounded();
let dispatcher =
app.add_model(|ctx| ModelEventDispatcher::new(events_rx, sessions.clone(), ctx));
let line_editor =
app.add_model(|ctx| LineEditorStatus::new(dispatcher.clone(), sessions.clone(), ctx));
let (_commands_tx, commands_rx) = async_channel::unbounded();
let model = Arc::new(FairMutex::new(TerminalModel::mock(None, None)));
let writes = Arc::new(FairMutex::new(Vec::new()));
let controller = app.add_model(|ctx| {
PtyController::new(
RecordingEventLoopSender(writes.clone()),
dispatcher,
line_editor,
sessions,
commands_rx,
model.clone(),
ctx,
)
});
let conversation_id = AIConversationId::new();
let previous_action = AIAgentActionId::from("previous-command".to_string());
let next_action = AIAgentActionId::from("follow-up-command".to_string());
let previous_block = {
let mut model = model.lock();
model.simulate_long_running_block("echo previous", "previous output");
model
.block_list_mut()
.active_block_mut()
.set_agent_interaction_mode_for_requested_command(
previous_action.clone(),
None,
conversation_id,
);
model.active_block_id().clone()
};
controller.update(&mut app, |controller, ctx| {
controller.write_command(
&ExecuteCommandEvent {
command: "echo follow-up".into(),
session_id: 1.into(),
workflow_id: None,
workflow_command: None,
should_add_command_to_history: false,
source: CommandExecutionSource::AI {
metadata: AgentInteractionMetadata::new_hidden(
next_action.clone(),
conversation_id,
),
},
},
ShellType::Zsh,
ctx,
);
assert_eq!(controller.pending_writes.len(), 1);
controller.cancel_queued_agent_command(&previous_action);
assert_eq!(controller.pending_writes.len(), 1);
let write = controller.pending_writes.pop_front().unwrap();
let PtyWrite::Command {
execution: Some(event),
..
} = &write
else {
panic!("expected queued agent command");
};
let event = event.clone();
controller.pending_writes.push_back(write);
controller.cancel_queued_agent_command(&next_action);
assert!(controller.pending_writes.is_empty());
assert!(!controller.is_user_command_executing);
controller.write_command(&event, ShellType::Zsh, ctx);
});
assert!(writes.lock().is_empty());
{
let mut model = model.lock();
assert_eq!(
model
.block_list()
.active_block()
.requested_command_action_id(),
Some(&previous_action)
);
assert!(model
.block_list()
.block_for_ai_action_id(&next_action)
.is_none());
model.finish_block();
}
// The line editor becomes ready only after the preceding block has finished.
// Dispatch the queued write at that boundary, as the Active event normally does.
controller.update(&mut app, |controller, ctx| {
let write = controller.pending_writes.pop_front().unwrap();
controller.send_write_to_event_loop(write, ctx);
});
let model = model.lock();
let new_block = model
.block_list()
.block_for_ai_action_id(&next_action)
.unwrap();
assert_ne!(new_block.id(), &previous_block);
assert_eq!(new_block.ai_conversation_id(), Some(conversation_id));
assert_eq!(
model
.block_list()
.block_with_id(&previous_block)
.unwrap()
.requested_command_action_id(),
Some(&previous_action)
);
assert_eq!(writes.lock().len(), 1);
});
}
app/src/terminal/writeable_pty/terminal_manager_util.rs
+20 -14
View File
@@ -37,8 +37,8 @@ pub fn wire_up_pty_controller_with_surface<T: EventLoopSender, S: TerminalSurfac
{
let controller_weak_handle = pty_controller.downgrade();
let surface_weak_handle = surface.downgrade();
let model_clone = model.clone();
let command_sessions = sessions.clone();
ctx.subscribe_to_view(surface, move |_surface, event, ctx| {
// NOTE: we cannot simply use the strong reference (the model handle argument to this wire_up fn)
// because it'll create a reference cycle with the `subscribe_to_model` usage below. Instead,
@@ -79,7 +79,7 @@ pub fn wire_up_pty_controller_with_surface<T: EventLoopSender, S: TerminalSurfac
});
}
PtyIntent::ExecuteCommand(event) => {
let Some(shell_type) = sessions
let Some(shell_type) = command_sessions
.as_ref(ctx)
.get(event.session_id)
.map(|s| s.shell().shell_type())
@@ -88,20 +88,14 @@ pub fn wire_up_pty_controller_with_surface<T: EventLoopSender, S: TerminalSurfac
return;
};
model_clone.lock().block_list_mut().active_block_mut().set_cloud_workflow_state(event.workflow_id);
controller.update(ctx, |controller, ctx| {
controller.write_command(&event.command, shell_type, event.source.clone(), ctx)
controller.write_command(&event, shell_type, ctx)
});
}
PtyIntent::CancelQueuedAgentCommand(action_id) => {
controller.update(ctx, |controller, _| {
controller.cancel_queued_agent_command(&action_id);
});
if event.should_add_command_to_history {
update_command_history(
&event,
&model_clone,
model_event_sender.as_ref(),
&sessions,
ctx,
);
}
}
PtyIntent::RunNativeShellCompletions {
buffer_text,
@@ -120,6 +114,18 @@ pub fn wire_up_pty_controller_with_surface<T: EventLoopSender, S: TerminalSurfac
};
match event {
PtyControllerEvent::CommandStarted { event, block_id } => {
if event.should_add_command_to_history {
update_command_history(
event,
block_id,
&model,
model_event_sender.as_ref(),
&sessions,
ctx,
);
}
}
PtyControllerEvent::PtyDisconnected => surface.update(ctx, |_, _| {
model.lock().exit(ExitReason::PtyDisconnected);
}),
app/src/terminal/writeable_pty/terminal_surface.rs
+2 -1
View File
@@ -5,7 +5,7 @@ use async_channel::Sender;
use warpui::AppContext;
use warpui::{Entity, ViewContext};
use crate::ai::agent::AIAgentPtyWriteMode;
use crate::ai::agent::{AIAgentActionId, AIAgentPtyWriteMode};
#[cfg(unix)]
use crate::terminal::event::AfterBlockCompletedEvent;
use crate::terminal::model::completions::ShellCompletion;
@@ -30,6 +30,7 @@ pub enum PtyIntent {
},
Resize(SizeUpdate),
ExecuteCommand(ExecuteCommandEvent),
CancelQueuedAgentCommand(AIAgentActionId),
RunNativeShellCompletions {
buffer_text: String,
results_tx: Sender<Vec<ShellCompletion>>,