From 8633a398e653eee47b375c963996c78a8a14aa48 Mon Sep 17 00:00:00 2001 From: JJ Liebig Date: Wed, 2 Sep 2026 15:34:03 +0400 Subject: fix: reliably submit delayed agent prompts (#3506) --- src/api/wait.rs | 70 +++--- src/app/api.rs | 8 +- src/app/api/agents.rs | 170 +++++++++----- src/cli/spec.rs | 2 +- src/pane.rs | 40 ++-- src/pty/actor.rs | 321 +++++++++++++++++++++---- src/pty/actor/unix.rs | 607 +++++++++++++++++++++++++++++++++++------------- src/server/headless.rs | 6 + src/terminal/runtime.rs | 9 +- 9 files changed, 917 insertions(+), 316 deletions(-) (limited to 'src') diff --git a/src/api/wait.rs b/src/api/wait.rs index 0c47fbff..9403c4f2 100644 --- a/src/api/wait.rs +++ b/src/api/wait.rs @@ -193,7 +193,6 @@ pub(super) fn prompt_agent( ))); }; - let last_event_sequence = event_hub.current_sequence(); let before_prompt = match agent_get(&request_id, ¶ms.target, api_tx) { Ok(agent) => agent, Err(response) => { @@ -202,7 +201,11 @@ pub(super) fn prompt_agent( .map_err(std::io::Error::other); } }; + let prompt_started_working = + before_prompt.agent_status == crate::api::schema::AgentStatus::Working; let target = params.target.clone(); + let wait_started = std::time::Instant::now(); + let last_event_sequence = event_hub.current_sequence(); let prompt_response = dispatch_to_app_with_timeout( Request { id: request_id.clone(), @@ -223,39 +226,38 @@ pub(super) fn prompt_agent( return agent_wait_not_running(request_id).map(Some); } - let wait_started = std::time::Instant::now(); + let prompt_activity_observed = prompt_started_working + || matches!( + prompted.agent_status, + crate::api::schema::AgentStatus::Working | crate::api::schema::AgentStatus::Blocked + ); let prompt_state_change_seq = prompted.state_change_seq; let until = agent_wait_statuses(wait.until); let mut initial = prompted; - let mut after_state_change_seq = Some(prompt_state_change_seq); - - if initial.agent_status != crate::api::schema::AgentStatus::Working { - let effect_timeout_ms = wait - .timeout_ms - .map_or(AGENT_PROMPT_EFFECT_TIMEOUT_MS, |timeout_ms| { - timeout_ms.min(AGENT_PROMPT_EFFECT_TIMEOUT_MS) - }); - let timeout_kind = if wait - .timeout_ms - .is_some_and(|timeout_ms| timeout_ms <= AGENT_PROMPT_EFFECT_TIMEOUT_MS) - { - AgentWaitTimeoutKind::Status - } else { - AgentWaitTimeoutKind::PromptStalled { - baseline: prompt_state_change_seq, - timeout_ms: effect_timeout_ms, + + if !prompt_activity_observed { + let remaining_timeout_ms = remaining_timeout_ms(wait.timeout_ms, wait_started); + let (effect_timeout_ms, timeout_kind) = match remaining_timeout_ms { + Some(timeout_ms) if timeout_ms <= AGENT_PROMPT_EFFECT_TIMEOUT_MS => { + (timeout_ms, AgentWaitTimeoutKind::Status) } + _ => ( + AGENT_PROMPT_EFFECT_TIMEOUT_MS, + AgentWaitTimeoutKind::PromptStalled { + timeout_ms: AGENT_PROMPT_EFFECT_TIMEOUT_MS, + }, + ), }; let Some(outcome) = wait_for_resolved_agent( request_id.clone(), ResolvedAgentWait { target: target.clone(), - until: all_agent_statuses(), + until: prompt_activity_statuses(), timeout_ms: Some(effect_timeout_ms), initial, last_event_sequence, - after_state_change_seq, - accept_transient_status: false, + after_state_change_seq: Some(prompt_state_change_seq), + accept_transient_status: true, timeout_kind, }, stream, @@ -270,10 +272,9 @@ pub(super) fn prompt_agent( AgentWaitOutcome::Matched(agent) => *agent, AgentWaitOutcome::Response(response) => return Ok(Some(response)), }; - after_state_change_seq = None; - if agent_wait_matches(&initial, &until, None) { - return agent_prompt_success(request_id, initial).map(Some); - } + } + if agent_wait_matches(&initial, &until, None) { + return agent_prompt_success(request_id, initial).map(Some); } let Some(outcome) = wait_for_resolved_agent( @@ -286,7 +287,7 @@ pub(super) fn prompt_agent( // Replay from before submission so terminal lifecycle events consumed by // the activity gate still terminate this settled-state wait. last_event_sequence, - after_state_change_seq, + after_state_change_seq: None, accept_transient_status: false, timeout_kind: AgentWaitTimeoutKind::Status, }, @@ -337,7 +338,7 @@ struct ResolvedAgentWait { #[derive(Clone, Copy)] enum AgentWaitTimeoutKind { Status, - PromptStalled { baseline: u64, timeout_ms: u64 }, + PromptStalled { timeout_ms: u64 }, } enum AgentWaitOutcome { @@ -497,14 +498,10 @@ fn wait_for_resolved_agent( } } -fn all_agent_statuses() -> Vec { - // Keep this exhaustive: every status is evidence that the sequence advanced. +fn prompt_activity_statuses() -> Vec { vec![ - crate::api::schema::AgentStatus::Idle, crate::api::schema::AgentStatus::Working, crate::api::schema::AgentStatus::Blocked, - crate::api::schema::AgentStatus::Done, - crate::api::schema::AgentStatus::Unknown, ] } @@ -617,15 +614,12 @@ fn agent_wait_timeout( AgentWaitTimeoutKind::Status => { ("timeout", "timed out waiting for agent status".to_string()) } - AgentWaitTimeoutKind::PromptStalled { - baseline, - timeout_ms, - } => { + AgentWaitTimeoutKind::PromptStalled { timeout_ms } => { let status = format!("{:?}", current.agent_status).to_ascii_lowercase(); ( "agent_prompt_stalled", format!( - "agent prompt produced no observed state change within {timeout_ms} ms; status is {status} and state_change_seq remained {baseline}" + "agent prompt produced no observed working or blocked state within {timeout_ms} ms; current status is {status}" ), ) } diff --git a/src/app/api.rs b/src/app/api.rs index e6f6b6ec..0c3b9f98 100644 --- a/src/app/api.rs +++ b/src/app/api.rs @@ -945,7 +945,13 @@ impl App { return self.handle_agent_view_clear(request.id, params) } Method::AgentStart(params) => return self.handle_agent_start(request.id, params), - Method::AgentPrompt(params) => return self.handle_agent_prompt(request.id, params), + Method::AgentPrompt(_) => { + return responses::encode_error( + request.id, + "invalid_request", + "agent.prompt is handled asynchronously by the app runtime", + ); + } Method::AgentWait(_) => { return responses::encode_error( request.id, diff --git a/src/app/api/agents.rs b/src/app/api/agents.rs index 65a65863..3e4e83a8 100644 --- a/src/app/api/agents.rs +++ b/src/app/api/agents.rs @@ -59,13 +59,54 @@ impl App { encode_success(id, ResponseResult::AgentStarted { agent, argv }) } - pub(super) fn handle_agent_prompt(&mut self, id: String, params: AgentPromptParams) -> String { + pub(crate) fn handle_deferred_agent_api_request( + &mut self, + request: crate::api::schema::Request, + respond_to: std::sync::mpsc::Sender, + ) -> bool { + let crate::api::schema::Method::AgentPrompt(params) = request.method else { + return false; + }; + match self.queue_agent_prompt(request.id, params) { + Ok((id, agent, completion)) => { + std::thread::spawn(move || { + let response = match completion.recv() { + Ok(Ok(())) => encode_success(id, ResponseResult::AgentPrompted { agent }), + Ok(Err(err)) => encode_error(id, "agent_prompt_failed", err.to_string()), + Err(_) => encode_error(id, "agent_prompt_failed", "pty actor closed"), + }; + let _ = respond_to.send(response); + }); + } + Err(response) => { + let _ = respond_to.send(response); + } + } + true + } + + fn queue_agent_prompt( + &mut self, + id: String, + params: AgentPromptParams, + ) -> Result< + ( + String, + crate::api::schema::AgentInfo, + std::sync::mpsc::Receiver>, + ), + String, + > { if params.text.is_empty() { - return encode_error(id, "empty_agent_prompt", "agent prompt must not be empty"); + return Err(encode_error( + id, + "empty_agent_prompt", + "agent prompt must not be empty", + )); } let resolved = match self.resolve_agent_target(¶ms.target) { Ok(resolved) => resolved, - Err(err) => return encode_error_body(id, self.agent_target_error_body(err)), + Err(err) => return Err(encode_error_body(id, self.agent_target_error_body(err))), }; let Some(terminal_id) = self .state @@ -74,60 +115,65 @@ impl App { .and_then(|workspace| workspace.terminal_id(resolved.pane_id)) .cloned() else { - return agent_not_found(id, ¶ms.target); + return Err(agent_not_found(id, ¶ms.target)); }; let Some(terminal) = self.state.terminals.get(&terminal_id) else { - return agent_not_found(id, ¶ms.target); + return Err(agent_not_found(id, ¶ms.target)); }; if terminal.state == crate::detect::AgentState::Blocked { - return encode_error( + return Err(encode_error( id, "agent_blocked", format!( "agent {} is blocked and requires interactive input", params.target ), - ); + )); } let Some(expected_agent) = terminal.effective_known_agent() else { - return agent_not_ready(id, ¶ms.target); + return Err(agent_not_ready(id, ¶ms.target)); }; if terminal.managed_agent_launch_pending() { - return agent_not_ready(id, ¶ms.target); + return Err(agent_not_ready(id, ¶ms.target)); } let Some(runtime) = self.lookup_runtime_sender(resolved.ws_idx, resolved.pane_id) else { - return agent_not_found(id, ¶ms.target); + return Err(agent_not_found(id, ¶ms.target)); }; if !super::super::agents::runtime_hosts_agent(runtime, expected_agent) { - return encode_error( + return Err(encode_error( id, "agent_not_ready", format!( "agent {} is no longer the pane foreground process", params.target ), - ); + )); } if expected_agent == crate::detect::Agent::GithubCopilot { // Copilot ignores synthetic Enter after focus loss until it receives focus gained. let focus = match crate::ghostty::encode_focus(crate::ghostty::FocusEvent::Gained) { Ok(focus) => focus, - Err(err) => return encode_error(id, "agent_prompt_failed", err.to_string()), + Err(err) => { + return Err(encode_error(id, "agent_prompt_failed", err.to_string())); + } }; if let Err(err) = runtime.try_send_bytes(Bytes::from(focus)) { - return encode_error(id, "agent_prompt_failed", err.to_string()); + return Err(encode_error(id, "agent_prompt_failed", err.to_string())); } } let (text, enter) = crate::app::api_helpers::encode_api_submission_parts(runtime, ¶ms.text); - if let Err(err) = runtime.try_send_bytes(Bytes::from(text)) { - return encode_error(id, "agent_prompt_failed", err.to_string()); - } - runtime.send_bytes_after(Bytes::from(enter), AGENT_PROMPT_SUBMIT_DELAY); let Some(agent) = self.agent_info(resolved.ws_idx, resolved.pane_id) else { - return agent_not_found(id, ¶ms.target); + return Err(agent_not_found(id, ¶ms.target)); }; - encode_success(id, ResponseResult::AgentPrompted { agent }) + let completion = runtime + .queue_user_input_submission( + Bytes::from(text), + Bytes::from(enter), + AGENT_PROMPT_SUBMIT_DELAY, + ) + .map_err(|err| encode_error(id.clone(), "agent_prompt_failed", err.to_string()))?; + Ok((id, agent, completion)) } pub(super) fn handle_agent_read( @@ -332,6 +378,28 @@ mod tests { app } + fn start_deferred_agent_prompt( + app: &mut App, + id: &str, + params: AgentPromptParams, + ) -> std::sync::mpsc::Receiver { + let (respond_to, response_rx) = std::sync::mpsc::channel(); + assert!(app.handle_deferred_agent_api_request( + crate::api::schema::Request { + id: id.into(), + method: crate::api::schema::Method::AgentPrompt(params), + }, + respond_to, + )); + response_rx + } + + fn run_deferred_agent_prompt(app: &mut App, id: &str, params: AgentPromptParams) -> String { + start_deferred_agent_prompt(app, id, params) + .recv_timeout(Duration::from_secs(1)) + .expect("agent prompt responds after submission") + } + #[tokio::test] async fn agent_prompt_sends_text_then_delays_enter() { let mut app = app_with_agent(); @@ -344,21 +412,26 @@ mod tests { terminal.set_detected_state(Some(Agent::OpenCode), AgentState::Working); let (runtime, mut rx) = crate::terminal::TerminalRuntime::test_with_channel_and_scrollback_bytes( - 80, 24, 0, b"", 1, + 80, 24, 0, b"", 2, ); runtime.test_process_pty_bytes(b"\x1b[?2004h"); app.state.insert_test_runtime(pane_id, runtime); let public_pane_id = app.public_pane_id(0, pane_id).unwrap(); let bracketed_started = std::time::Instant::now(); - let response = app.handle_agent_prompt( - "req".into(), + let response_rx = start_deferred_agent_prompt( + &mut app, + "req", AgentPromptParams { target: public_pane_id, text: "A != B".into(), wait: None, }, ); + assert!(response_rx.try_recv().is_err()); + let response = response_rx + .recv_timeout(Duration::from_secs(1)) + .expect("agent prompt responds after submission"); let success: SuccessResponse = serde_json::from_str(&response).unwrap(); let ResponseResult::AgentPrompted { agent, .. } = success.result else { panic!("expected prompted response"); @@ -368,22 +441,16 @@ mod tests { rx.try_recv().unwrap(), Bytes::from_static(b"\x1b[200~A != B\x1b[201~") ); - assert!(rx.try_recv().is_err()); - assert_eq!( - tokio::time::timeout(Duration::from_secs(1), rx.recv()) - .await - .unwrap() - .unwrap(), - Bytes::from_static(b"\r") - ); + assert_eq!(rx.try_recv().unwrap(), Bytes::from_static(b"\r")); assert!(bracketed_started.elapsed() >= AGENT_PROMPT_SUBMIT_DELAY); app.lookup_runtime_sender(0, pane_id) .unwrap() .test_process_pty_bytes(b"\x1b[?2004l"); let raw_started = std::time::Instant::now(); - let raw = app.handle_agent_prompt( - "req-raw".into(), + let raw = run_deferred_agent_prompt( + &mut app, + "req-raw", AgentPromptParams { target: "reviewer".into(), text: "A != B".into(), @@ -393,18 +460,12 @@ mod tests { let raw: SuccessResponse = serde_json::from_str(&raw).unwrap(); assert!(matches!(raw.result, ResponseResult::AgentPrompted { .. })); assert_eq!(rx.try_recv().unwrap(), Bytes::from_static(b"A != B")); - assert!(rx.try_recv().is_err()); - assert_eq!( - tokio::time::timeout(Duration::from_secs(1), rx.recv()) - .await - .unwrap() - .unwrap(), - Bytes::from_static(b"\r") - ); + assert_eq!(rx.try_recv().unwrap(), Bytes::from_static(b"\r")); assert!(raw_started.elapsed() >= AGENT_PROMPT_SUBMIT_DELAY); - let rejected = app.handle_agent_prompt( - "req-label".into(), + let rejected = run_deferred_agent_prompt( + &mut app, + "req-label", AgentPromptParams { target: "opencode".into(), text: "wrong target".into(), @@ -429,8 +490,9 @@ mod tests { let (runtime, mut rx) = crate::terminal::TerminalRuntime::test_with_channel(80, 24); app.state.insert_test_runtime(pane_id, runtime); - let response = app.handle_agent_prompt( - "req".into(), + let response = run_deferred_agent_prompt( + &mut app, + "req", AgentPromptParams { target: "reviewer".into(), text: "unrelated prompt".into(), @@ -468,8 +530,9 @@ mod tests { runtime.test_process_pty_bytes(b"\x1b[?2004h"); app.state.insert_test_runtime(pane_id, runtime); - let response = app.handle_agent_prompt( - "req".into(), + let response = run_deferred_agent_prompt( + &mut app, + "req", AgentPromptParams { target: "reviewer".into(), text: "A != B".into(), @@ -486,13 +549,7 @@ mod tests { rx.try_recv().unwrap(), Bytes::from_static(b"\x1b[200~A != B\x1b[201~") ); - assert_eq!( - tokio::time::timeout(Duration::from_secs(1), rx.recv()) - .await - .unwrap() - .unwrap(), - Bytes::from_static(b"\r") - ); + assert_eq!(rx.try_recv().unwrap(), Bytes::from_static(b"\r")); } #[tokio::test] @@ -552,8 +609,9 @@ mod tests { let (runtime, mut rx) = crate::terminal::TerminalRuntime::test_with_channel(80, 24); app.state.insert_test_runtime(pane_id, runtime); - let response = app.handle_agent_prompt( - "req-pending".into(), + let response = run_deferred_agent_prompt( + &mut app, + "req-pending", AgentPromptParams { target: "reviewer".into(), text: "A != B".into(), diff --git a/src/cli/spec.rs b/src/cli/spec.rs index bcfa6c35..cf23e320 100644 --- a/src/cli/spec.rs +++ b/src/cli/spec.rs @@ -367,7 +367,7 @@ fn agent_command() -> Command { .help("Fail after this many milliseconds"), ) .after_help( - "If the agent is already blocked, submission is rejected with agent_blocked before any input is sent. When an accepted submission starts from another non-working state, --wait first requires an observed state change within 5000ms; otherwise it returns agent_prompt_stalled. A shorter --timeout returns timeout instead. It then matches idle, done, or blocked by default, or any exact --until state. It does not track turns: if the agent is already working, that active turn's completion may match. Without --timeout, the settled-state wait is indefinite.", + "If the agent is already blocked, submission is rejected with agent_blocked before any input is sent. When an accepted submission starts from another non-working state, --wait requires an observed working or blocked state within 5000ms; otherwise it returns agent_prompt_stalled. A caller timeout that expires first returns timeout. It then matches idle, done, or blocked by default, or any exact --until state. It does not track turns: if the agent is already working, that active turn's completion may match.", ), ) .subcommand( diff --git a/src/pane.rs b/src/pane.rs index 48928034..c71665f5 100644 --- a/src/pane.rs +++ b/src/pane.rs @@ -1181,24 +1181,29 @@ impl PaneRuntimeIo { } } - fn send_bytes_after(&self, bytes: Bytes, delay: std::time::Duration) { + fn queue_user_input_submission( + &self, + text: Bytes, + enter: Bytes, + delay: std::time::Duration, + ) -> std::io::Result>> { match self { - PaneRuntimeIo::Actor(actor) => { - let actor = actor.clone(); - tokio::spawn(async move { - tokio::time::sleep(delay).await; - if let Err(err) = actor.write_user_input(bytes).await { - warn!(error = %err, "failed to send delayed PTY input"); - } - }); - } + PaneRuntimeIo::Actor(actor) => actor.queue_user_input_submission(text, enter, delay), #[cfg(test)] PaneRuntimeIo::TestChannel { sender, .. } => { let sender = sender.clone(); - tokio::spawn(async move { - tokio::time::sleep(delay).await; - let _ = sender.send(bytes).await; + let (reply_tx, reply_rx) = std::sync::mpsc::channel(); + std::thread::spawn(move || { + let result = sender + .try_send(text) + .map_err(std::io::Error::other) + .and_then(|()| { + std::thread::sleep(delay); + sender.try_send(enter).map_err(std::io::Error::other) + }); + let _ = reply_tx.send(result); }); + Ok(reply_rx) } } } @@ -2856,8 +2861,13 @@ impl PaneRuntime { self.io.try_send_bytes(bytes) } - pub fn send_bytes_after(&self, bytes: Bytes, delay: std::time::Duration) { - self.io.send_bytes_after(bytes, delay); + pub fn queue_user_input_submission( + &self, + text: Bytes, + enter: Bytes, + delay: std::time::Duration, + ) -> std::io::Result>> { + self.io.queue_user_input_submission(text, enter, delay) } pub fn try_send_paste(&self, text: String) -> Result<(), mpsc::error::TrySendError> { diff --git a/src/pty/actor.rs b/src/pty/actor.rs index 6fde5051..39b1d29f 100644 --- a/src/pty/actor.rs +++ b/src/pty/actor.rs @@ -43,6 +43,24 @@ mod windows { pub on_reader_exit: Option, } + enum PtyIoDataCommand { + WriteUserInput(Bytes), + SubmitUserInput { + text: Bytes, + enter: Bytes, + delay: Duration, + reply: std_mpsc::Sender>, + }, + } + + enum PtyIoWriteCommand { + Write(Bytes), + SubmissionPart { + bytes: Bytes, + reply: std_mpsc::Sender>, + }, + } + enum PtyIoControlCommand { Resize(PtyResizeRequest), Shutdown, @@ -50,40 +68,77 @@ mod windows { #[derive(Clone)] pub(crate) struct PtyIoActorHandle { - data_tx: mpsc::Sender, + data_tx: mpsc::Sender, control_tx: std_mpsc::Sender, - write_tx: std_mpsc::Sender, + write_tx: std_mpsc::Sender, response_order: Arc>, accepting: Arc>, } impl PtyIoActorHandle { - pub(crate) async fn write_user_input( + pub(crate) fn try_write_user_input( &self, bytes: Bytes, - ) -> Result<(), mpsc::error::SendError> { + ) -> Result<(), mpsc::error::TrySendError> { if !*self .accepting .lock() .unwrap_or_else(|poisoned| poisoned.into_inner()) { - return Err(mpsc::error::SendError(bytes)); + return Err(mpsc::error::TrySendError::Closed(bytes)); } - self.data_tx.send(bytes).await + self.data_tx + .try_send(PtyIoDataCommand::WriteUserInput(bytes)) + .map_err(|err| match err { + mpsc::error::TrySendError::Full(command) => { + let PtyIoDataCommand::WriteUserInput(bytes) = command else { + unreachable!("queued write returned another command") + }; + mpsc::error::TrySendError::Full(bytes) + } + mpsc::error::TrySendError::Closed(command) => { + let PtyIoDataCommand::WriteUserInput(bytes) = command else { + unreachable!("queued write returned another command") + }; + mpsc::error::TrySendError::Closed(bytes) + } + }) } - pub(crate) fn try_write_user_input( + pub(crate) fn queue_user_input_submission( &self, - bytes: Bytes, - ) -> Result<(), mpsc::error::TrySendError> { - if !*self + text: Bytes, + enter: Bytes, + delay: Duration, + ) -> std::io::Result>> { + let accepting = self .accepting .lock() - .unwrap_or_else(|poisoned| poisoned.into_inner()) - { - return Err(mpsc::error::TrySendError::Closed(bytes)); + .unwrap_or_else(|poisoned| poisoned.into_inner()); + if !*accepting { + return Err(std::io::Error::new( + std::io::ErrorKind::BrokenPipe, + "pty actor closed", + )); } - self.data_tx.try_send(bytes) + let (reply_tx, reply_rx) = std_mpsc::channel(); + self.data_tx + .try_send(PtyIoDataCommand::SubmitUserInput { + text, + enter, + delay, + reply: reply_tx, + }) + .map_err(|err| match err { + mpsc::error::TrySendError::Full(_) => std::io::Error::new( + std::io::ErrorKind::WouldBlock, + "pty input queue is full", + ), + mpsc::error::TrySendError::Closed(_) => { + std::io::Error::new(std::io::ErrorKind::BrokenPipe, "pty actor closed") + } + })?; + Ok(reply_rx) } pub(crate) fn write_terminal_response(&self, response: impl FnOnce() -> Option) { @@ -92,7 +147,7 @@ mod windows { .lock() .unwrap_or_else(|poisoned| poisoned.into_inner()); if let Some(bytes) = response().filter(|bytes| !bytes.is_empty()) { - let _ = self.write_tx.send(bytes); + let _ = self.write_tx.send(PtyIoWriteCommand::Write(bytes)); } } @@ -143,29 +198,22 @@ mod windows { let mut writer = master .take_writer() .map_err(|err| std::io::Error::other(err.to_string()))?; - let (data_tx, mut data_rx) = mpsc::channel::(1024); + let (data_tx, mut data_rx) = mpsc::channel::(1024); let (control_tx, control_rx) = std_mpsc::channel::(); - let (write_tx, write_rx) = std_mpsc::channel::(); + let (write_tx, write_rx) = std_mpsc::channel::(); let response_order = Arc::new(Mutex::new(())); let accepting = Arc::new(Mutex::new(!initially_quiesced)); std::thread::spawn(move || { - for bytes in write_rx { - if writer.write_all(&bytes).is_err() || writer.flush().is_err() { - break; - } - } + run_writer(&mut writer, write_rx); debug!(pane_id, "windows pty writer thread exiting"); }); { let write_tx = write_tx.clone(); + let accepting = Arc::clone(&accepting); std::thread::spawn(move || { - while let Some(bytes) = data_rx.blocking_recv() { - if write_tx.send(bytes).is_err() { - break; - } - } + run_input_forwarder(&mut data_rx, write_tx, accepting); debug!(pane_id, "windows pty input thread exiting"); }); } @@ -183,11 +231,9 @@ mod windows { .lock() .unwrap_or_else(|poisoned| poisoned.into_inner()); let result = on_read(&buf[..n]); - if result - .terminal_responses - .into_iter() - .any(|response| write_tx.send(response).is_err()) - { + if result.terminal_responses.into_iter().any(|response| { + write_tx.send(PtyIoWriteCommand::Write(response)).is_err() + }) { break; } } @@ -219,11 +265,9 @@ mod windows { }) { warn!(pane_id, err = %err, "windows pty resize failed"); } - if request - .terminal_responses - .into_iter() - .any(|response| write_tx.send(response).is_err()) - { + if request.terminal_responses.into_iter().any(|response| { + write_tx.send(PtyIoWriteCommand::Write(response)).is_err() + }) { break; } } @@ -244,8 +288,207 @@ mod windows { } } - #[allow(dead_code)] - fn _assert_duration_send(_: Duration) {} + fn run_writer(writer: &mut impl Write, write_rx: std_mpsc::Receiver) { + for command in write_rx { + let result = match command { + PtyIoWriteCommand::Write(bytes) => write_and_flush(writer, &bytes), + PtyIoWriteCommand::SubmissionPart { bytes, reply } => { + let result = write_and_flush(writer, &bytes); + let failed = result.is_err(); + let _ = reply.send(result); + if failed { + break; + } + continue; + } + }; + if result.is_err() { + break; + } + } + } + + fn run_input_forwarder( + data_rx: &mut mpsc::Receiver, + write_tx: std_mpsc::Sender, + accepting: Arc>, + ) { + while let Some(command) = data_rx.blocking_recv() { + match command { + PtyIoDataCommand::WriteUserInput(bytes) => { + if write_tx.send(PtyIoWriteCommand::Write(bytes)).is_err() { + break; + } + } + PtyIoDataCommand::SubmitUserInput { + text, + enter, + delay, + reply, + } => { + let result = write_submission_part(&write_tx, text).and_then(|()| { + std::thread::sleep(delay); + let accepting = accepting + .lock() + .unwrap_or_else(|poisoned| poisoned.into_inner()); + if !*accepting { + return Err(pty_actor_closed()); + } + write_submission_part(&write_tx, enter) + }); + let failed = result.is_err(); + let _ = reply.send(result); + if failed { + break; + } + } + } + } + } + + fn write_submission_part( + write_tx: &std_mpsc::Sender, + bytes: Bytes, + ) -> std::io::Result<()> { + let (reply, completion) = std_mpsc::channel(); + write_tx + .send(PtyIoWriteCommand::SubmissionPart { bytes, reply }) + .map_err(|_| pty_actor_closed())?; + completion + .recv() + .unwrap_or_else(|_| Err(pty_actor_closed())) + } + + fn pty_actor_closed() -> std::io::Error { + std::io::Error::new(std::io::ErrorKind::BrokenPipe, "pty actor closed") + } + + fn write_and_flush(writer: &mut impl Write, bytes: &[u8]) -> std::io::Result<()> { + writer.write_all(bytes)?; + writer.flush() + } + + #[cfg(test)] + mod tests { + use super::*; + use std::time::Instant; + + struct RecordingWriter { + writes: Vec<(Vec, Instant)>, + flushes: Vec, + fail_after: Option, + flushed: std_mpsc::Sender<()>, + } + + impl Write for RecordingWriter { + fn write(&mut self, bytes: &[u8]) -> std::io::Result { + if self.fail_after == Some(self.writes.len()) { + return Err(std::io::Error::new( + std::io::ErrorKind::BrokenPipe, + "writer closed", + )); + } + self.writes.push((bytes.to_vec(), Instant::now())); + Ok(bytes.len()) + } + + fn flush(&mut self) -> std::io::Result<()> { + self.flushes.push(Instant::now()); + let _ = self.flushed.send(()); + Ok(()) + } + } + + fn run_recorded_submission( + fail_after: Option, + delay: Duration, + during_delay: impl FnOnce(&std_mpsc::Sender, &Arc>), + ) -> (RecordingWriter, std::io::Result<()>) { + let (flushed_tx, flushed_rx) = std_mpsc::channel(); + let mut writer = RecordingWriter { + writes: Vec::new(), + flushes: Vec::new(), + fail_after, + flushed: flushed_tx, + }; + let (data_tx, mut data_rx) = mpsc::channel(2); + let (write_tx, write_rx) = std_mpsc::channel(); + let (reply_tx, reply_rx) = std_mpsc::channel(); + let accepting = Arc::new(Mutex::new(true)); + data_tx + .try_send(PtyIoDataCommand::SubmitUserInput { + text: Bytes::from_static(b"prompt"), + enter: Bytes::from_static(b"\r"), + delay, + reply: reply_tx, + }) + .unwrap(); + data_tx + .try_send(PtyIoDataCommand::WriteUserInput(Bytes::from_static( + b"user", + ))) + .unwrap(); + let writer_thread = std::thread::spawn(move || { + run_writer(&mut writer, write_rx); + writer + }); + let input_write_tx = write_tx.clone(); + let input_accepting = Arc::clone(&accepting); + let input_thread = std::thread::spawn(move || { + run_input_forwarder(&mut data_rx, input_write_tx, input_accepting) + }); + flushed_rx.recv().expect("prompt was flushed"); + during_delay(&write_tx, &accepting); + let result = reply_rx.recv().expect("writer reports submission"); + drop(data_tx); + input_thread.join().expect("input thread joins"); + drop(write_tx); + (writer_thread.join().expect("writer thread joins"), result) + } + + #[test] + fn submission_sequences_user_input_but_allows_terminal_responses() { + let delay = Duration::from_millis(30); + let (writer, result) = run_recorded_submission(None, delay, |write_tx, _| { + write_tx + .send(PtyIoWriteCommand::Write(Bytes::from_static(b"response"))) + .unwrap(); + }); + result.expect("submission succeeds"); + + assert_eq!(writer.writes[0].0, b"prompt"); + assert_eq!(writer.writes[1].0, b"response"); + assert_eq!(writer.writes[2].0, b"\r"); + assert_eq!(writer.writes[3].0, b"user"); + assert!(writer.writes[2].1.duration_since(writer.flushes[0]) >= delay); + } + + #[test] + fn submission_returns_enter_write_failure() { + let (_writer, result) = run_recorded_submission(Some(1), Duration::ZERO, |_, _| {}); + let err = result.expect_err("enter failure reaches caller"); + + assert_eq!(err.kind(), std::io::ErrorKind::BrokenPipe); + } + + #[test] + fn shutdown_during_submission_delay_cancels_enter() { + let (writer, result) = + run_recorded_submission(None, Duration::from_millis(30), |_, accepting| { + *accepting.lock().unwrap() = false; + }); + let err = result.expect_err("shutdown cancels enter"); + assert_eq!(err.kind(), std::io::ErrorKind::BrokenPipe); + assert_eq!( + writer + .writes + .iter() + .map(|write| write.0.as_slice()) + .collect::>(), + vec![b"prompt"] + ); + } + } } #[cfg(windows)] diff --git a/src/pty/actor/unix.rs b/src/pty/actor/unix.rs index 599f02cb..d7a8bb78 100644 --- a/src/pty/actor/unix.rs +++ b/src/pty/actor/unix.rs @@ -73,6 +73,12 @@ pub(crate) struct PtyIoActorConfig { enum PtyIoDataCommand { WriteUserInput(Bytes), + SubmitUserInput { + text: Bytes, + enter: Bytes, + delay: Duration, + reply: std_mpsc::Sender>, + }, } enum PtyIoControlCommand { @@ -100,37 +106,6 @@ struct UserWriteGate { } impl PtyIoActorHandle { - pub(crate) async fn write_user_input( - &self, - bytes: Bytes, - ) -> Result<(), mpsc::error::SendError> { - { - let user_writes = self - .user_writes - .lock() - .unwrap_or_else(|poisoned| poisoned.into_inner()); - if !user_writes.accepting { - return Err(mpsc::error::SendError(bytes)); - } - } - - let permit = match self.data_tx.reserve().await { - Ok(permit) => permit, - Err(_) => return Err(mpsc::error::SendError(bytes)), - }; - - let user_writes = self - .user_writes - .lock() - .unwrap_or_else(|poisoned| poisoned.into_inner()); - if !user_writes.accepting { - return Err(mpsc::error::SendError(bytes)); - } - permit.send(PtyIoDataCommand::WriteUserInput(bytes)); - self.wake_actor(); - Ok(()) - } - pub(crate) fn try_write_user_input( &self, bytes: Bytes, @@ -150,15 +125,57 @@ impl PtyIoActorHandle { self.wake_actor(); Ok(()) } - Err(mpsc::error::TrySendError::Full(PtyIoDataCommand::WriteUserInput(bytes))) => { + Err(mpsc::error::TrySendError::Full(command)) => { + let PtyIoDataCommand::WriteUserInput(bytes) = command else { + unreachable!("queued write returned another command") + }; Err(mpsc::error::TrySendError::Full(bytes)) } - Err(mpsc::error::TrySendError::Closed(PtyIoDataCommand::WriteUserInput(bytes))) => { + Err(mpsc::error::TrySendError::Closed(command)) => { + let PtyIoDataCommand::WriteUserInput(bytes) = command else { + unreachable!("queued write returned another command") + }; Err(mpsc::error::TrySendError::Closed(bytes)) } } } + pub(crate) fn queue_user_input_submission( + &self, + text: Bytes, + enter: Bytes, + delay: Duration, + ) -> std::io::Result>> { + let user_writes = self + .user_writes + .lock() + .unwrap_or_else(|poisoned| poisoned.into_inner()); + if !user_writes.accepting { + return Err(std::io::Error::new( + std::io::ErrorKind::BrokenPipe, + "pty actor closed", + )); + } + let (reply_tx, reply_rx) = std_mpsc::channel(); + self.data_tx + .try_send(PtyIoDataCommand::SubmitUserInput { + text, + enter, + delay, + reply: reply_tx, + }) + .map_err(|err| match err { + mpsc::error::TrySendError::Full(_) => { + std::io::Error::new(std::io::ErrorKind::WouldBlock, "pty input queue is full") + } + mpsc::error::TrySendError::Closed(_) => { + std::io::Error::new(std::io::ErrorKind::BrokenPipe, "pty actor closed") + } + })?; + self.wake_actor(); + Ok(reply_rx) + } + pub(crate) fn write_terminal_response(&self, response: impl FnOnce() -> Option) { let _order = self .response_order @@ -232,6 +249,12 @@ impl PtyIoActorHandle { .user_writes .lock() .unwrap_or_else(|poisoned| poisoned.into_inner()); + if !user_writes.accepting { + return Err(std::io::Error::new( + std::io::ErrorKind::WouldBlock, + "PTY handoff is already in progress", + )); + } user_writes.accepting = false; if self .control_tx @@ -391,6 +414,8 @@ impl PtyIoActor { }, pending_writes: VecDeque::new(), current_write_offset: 0, + active_submission: None, + pending_handoff: None, wake_read_fd: wake_pipe.read_fd, controls, response_order, @@ -421,8 +446,10 @@ struct PtyIoActorRunner { data_rx: mpsc::Receiver, control_rx: std_mpsc::Receiver, state: ActorState, - pending_writes: VecDeque, + pending_writes: VecDeque, current_write_offset: usize, + active_submission: Option, + pending_handoff: Option>>, wake_read_fd: OwnedFd, controls: Arc>, response_order: Arc>, @@ -431,10 +458,47 @@ struct PtyIoActorRunner { poll_observer: Option>, } +struct ActiveSubmission { + enter: Bytes, + delay: Duration, + phase: SubmissionPhase, + reply: std_mpsc::Sender>, +} + +#[derive(Debug, PartialEq, Eq)] +struct PendingWrite { + bytes: Bytes, + boundary: Option, +} + +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +enum SubmissionBoundary { + Text, + Enter, +} + +enum SubmissionPhase { + WritingText, + WaitingUntil(Instant), + WritingEnter, +} + impl PtyIoActorRunner { fn enqueue_write(&mut self, bytes: Bytes) { if !bytes.is_empty() { - self.pending_writes.push_back(bytes); + self.pending_writes.push_back(PendingWrite { + bytes, + boundary: None, + }); + } + } + + fn enqueue_submission_write(&mut self, bytes: Bytes, boundary: SubmissionBoundary) { + if !bytes.is_empty() { + self.pending_writes.push_back(PendingWrite { + bytes, + boundary: Some(boundary), + }); } } @@ -449,7 +513,18 @@ impl PtyIoActorRunner { self.apply_pending_controls(); if !self.pending_writes.is_empty() { - self.flush_pending_writes_once(); + match self.flush_pending_writes_once() { + Ok(Some(boundary)) => self.complete_submission_boundary(boundary), + Ok(None) => {} + Err(err) => { + self.fail_active_submission(err); + break; + } + } + } + self.schedule_submission_enter(); + if self.active_submission.is_none() && self.pending_handoff.is_some() { + continue; } if let Some(poll_observer) = &self.poll_observer { @@ -461,7 +536,7 @@ impl PtyIoActorRunner { self.wake_read_fd.as_raw_fd(), self.state == ActorState::Running, !self.pending_writes.is_empty(), - ACTOR_IDLE_POLL_MS, + self.poll_timeout_ms(), ) { Ok(readiness) => { if readiness.wake_ready { @@ -478,7 +553,14 @@ impl PtyIoActorRunner { break; } if readiness.pty_write_ready && !self.pending_writes.is_empty() { - self.flush_pending_writes_once(); + match self.flush_pending_writes_once() { + Ok(Some(boundary)) => self.complete_submission_boundary(boundary), + Ok(None) => {} + Err(err) => { + self.fail_active_submission(err); + break; + } + } } } Err(err) => { @@ -488,6 +570,10 @@ impl PtyIoActorRunner { } } + self.fail_active_submission(std::io::Error::new( + std::io::ErrorKind::BrokenPipe, + "PTY actor closed during input submission", + )); if let Some(on_reader_exit) = self.on_reader_exit.take() { on_reader_exit(); } @@ -498,6 +584,13 @@ impl PtyIoActorRunner { if self.drain_control_commands() { return true; } + if self.active_submission.is_some() { + return false; + } + if let Some(reply) = self.pending_handoff.take() { + self.defer_or_begin_handoff(reply); + return false; + } self.drain_data_commands() } @@ -530,6 +623,9 @@ impl PtyIoActorRunner { should_exit = true; break; } + if self.active_submission.is_some() { + break; + } } Err(DataTryRecvError::Empty) => break, Err(DataTryRecvError::Disconnected) => { @@ -548,6 +644,32 @@ impl PtyIoActorRunner { self.enqueue_write(bytes); } } + PtyIoDataCommand::SubmitUserInput { + text, + enter, + delay, + reply, + } => { + if self.state == ActorState::Running { + let phase = if text.is_empty() { + SubmissionPhase::WaitingUntil(Instant::now() + delay) + } else { + self.enqueue_submission_write(text, SubmissionBoundary::Text); + SubmissionPhase::WritingText + }; + self.active_submission = Some(ActiveSubmission { + enter, + delay, + phase, + reply, + }); + } else { + let _ = reply.send(Err(std::io::Error::new( + std::io::ErrorKind::BrokenPipe, + "pty actor is not accepting input", + ))); + } + } } false } @@ -555,8 +677,7 @@ impl PtyIoActorRunner { fn handle_control_command(&mut self, command: PtyIoControlCommand) -> bool { match command { PtyIoControlCommand::BeginHandoff(reply) => { - let result = self.begin_handoff(); - let _ = reply.send(result); + self.defer_or_begin_handoff(reply); } PtyIoControlCommand::DuplicateForHandoff(reply) => { let result = if self.state == ActorState::Quiesced { @@ -574,6 +695,7 @@ impl PtyIoActorRunner { let _ = reply.send(result); } PtyIoControlCommand::RollbackHandoff(reply) => { + self.pending_handoff.take(); let result = if self.state == ActorState::Released { Err(std::io::Error::new( std::io::ErrorKind::BrokenPipe, @@ -596,8 +718,26 @@ impl PtyIoActorRunner { false } + fn defer_or_begin_handoff(&mut self, reply: std_mpsc::Sender>) { + if self.active_submission.is_none() { + self.drain_pre_quiesce_commands(); + } + if self.active_submission.is_some() { + self.pending_handoff = Some(reply); + } else { + let result = self.begin_handoff(); + let _ = reply.send(result); + } + } + fn begin_handoff(&mut self) -> std::io::Result<()> { self.drain_pre_quiesce_commands(); + if self.active_submission.is_some() { + return Err(std::io::Error::new( + std::io::ErrorKind::WouldBlock, + "PTY input submission is still in progress", + )); + } self.apply_pending_controls(); if self.state == ActorState::Released { return Err(std::io::Error::new( @@ -606,7 +746,7 @@ impl PtyIoActorRunner { )); } let deadline = Instant::now() + HANDOFF_DRAIN_TIMEOUT; - self.flush_pending_writes_once(); + let _ = self.flush_pending_writes_once()?; while !self.pending_writes.is_empty() { let remaining = deadline.saturating_duration_since(Instant::now()); if remaining.is_zero() { @@ -633,7 +773,7 @@ impl PtyIoActorRunner { )); } if readiness.pty_write_ready { - self.flush_pending_writes_once(); + let _ = self.flush_pending_writes_once()?; } } self.state = ActorState::Quiesced; @@ -641,9 +781,12 @@ impl PtyIoActorRunner { } fn drain_pre_quiesce_commands(&mut self) { - while let Ok(PtyIoDataCommand::WriteUserInput(bytes)) = self.data_rx.try_recv() { - if self.state != ActorState::Released { - self.enqueue_write(bytes); + while let Ok(command) = self.data_rx.try_recv() { + if self.handle_data_command(command) { + break; + } + if self.active_submission.is_some() { + break; } } } @@ -717,32 +860,100 @@ impl PtyIoActorRunner { } } - fn flush_pending_writes_once(&mut self) { - while let Some(bytes) = self.pending_writes.front() { - let chunk = &bytes[self.current_write_offset..]; + fn complete_submission_boundary(&mut self, boundary: SubmissionBoundary) { + match boundary { + SubmissionBoundary::Text => { + let Some(submission) = self.active_submission.as_mut() else { + return; + }; + debug_assert!(matches!(submission.phase, SubmissionPhase::WritingText)); + submission.phase = SubmissionPhase::WaitingUntil(Instant::now() + submission.delay); + } + SubmissionBoundary::Enter => { + let Some(submission) = self.active_submission.take() else { + return; + }; + debug_assert!(matches!(submission.phase, SubmissionPhase::WritingEnter)); + let _ = submission.reply.send(Ok(())); + } + } + } + + fn schedule_submission_enter(&mut self) { + let Some(ActiveSubmission { + enter, + phase: SubmissionPhase::WaitingUntil(deadline), + .. + }) = self.active_submission.as_ref() + else { + return; + }; + if Instant::now() >= *deadline { + let enter = enter.clone(); + if enter.is_empty() { + let submission = self.active_submission.take().unwrap(); + let _ = submission.reply.send(Ok(())); + } else { + self.active_submission.as_mut().unwrap().phase = SubmissionPhase::WritingEnter; + self.enqueue_submission_write(enter, SubmissionBoundary::Enter); + } + } + } + + fn poll_timeout_ms(&self) -> i32 { + let Some(ActiveSubmission { + phase: SubmissionPhase::WaitingUntil(deadline), + .. + }) = self.active_submission.as_ref() + else { + return ACTOR_IDLE_POLL_MS; + }; + deadline + .saturating_duration_since(Instant::now()) + .as_millis() + .max(1) + .min(ACTOR_IDLE_POLL_MS as u128) as i32 + } + + fn fail_active_submission(&mut self, err: std::io::Error) { + if let Some(submission) = self.active_submission.take() { + let _ = submission.reply.send(Err(err)); + } + } + + fn flush_pending_writes_once(&mut self) -> std::io::Result> { + while let Some(write) = self.pending_writes.front() { + let chunk = &write.bytes[self.current_write_offset..]; match self.file.write(chunk) { Ok(0) => { - warn!(pane = self.pane_id, "PTY actor write returned zero bytes"); - return; + return Err(std::io::Error::new( + std::io::ErrorKind::WriteZero, + "PTY actor write returned zero bytes", + )); } Ok(written) => { self.current_write_offset += written; - if self.current_write_offset >= bytes.len() { - self.pending_writes.pop_front(); + if self.current_write_offset >= write.bytes.len() { + let completed = self.pending_writes.pop_front().unwrap(); self.current_write_offset = 0; + if let Some(boundary) = completed.boundary { + self.file.flush()?; + return Ok(Some(boundary)); + } } } - Err(err) if err.kind() == std::io::ErrorKind::WouldBlock => return, - Err(err) if err.kind() == std::io::ErrorKind::Interrupted => return, + Err(err) if err.kind() == std::io::ErrorKind::WouldBlock => return Ok(None), + Err(err) if err.kind() == std::io::ErrorKind::Interrupted => return Ok(None), Err(err) => { warn!(pane = self.pane_id, err = %err, "PTY actor write failed"); self.pending_writes.clear(); self.current_write_offset = 0; - return; + return Err(err); } } } - let _ = self.file.flush(); + self.file.flush()?; + Ok(None) } fn resize(&self, resize: PtyResize) { @@ -879,6 +1090,8 @@ mod tests { state: ActorState::Running, pending_writes: VecDeque::new(), current_write_offset: 0, + active_submission: None, + pending_handoff: None, wake_read_fd: wake_pipe.read_fd, controls: Arc::new(Mutex::new(SharedPtyControls::default())), response_order: Arc::new(Mutex::new(())), @@ -898,6 +1111,22 @@ mod tests { assert!(runner.pending_writes.is_empty()); } + #[test] + fn submission_boundary_does_not_wait_for_following_protocol_write() { + let (mut runner, _peer) = actor_runner_for_unit_test(); + runner.enqueue_submission_write(Bytes::from_static(b"prompt"), SubmissionBoundary::Text); + runner.enqueue_write(Bytes::from_static(b"response")); + + assert_eq!( + runner.flush_pending_writes_once().unwrap(), + Some(SubmissionBoundary::Text) + ); + assert_eq!( + runner.pending_writes[0].bytes, + Bytes::from_static(b"response") + ); + } + #[test] fn actor_writes_user_input_to_owned_fd() { let (handle, mut peer, _read_rx) = actor_with_socket_pair(false); @@ -912,6 +1141,127 @@ mod tests { handle.shutdown(); } + #[test] + fn actor_delays_enter_from_completed_prompt_write() { + let (handle, mut peer, _read_rx) = actor_with_socket_pair(false); + let text = Bytes::from(vec![b'x'; 4 * 1024 * 1024]); + let text_len = text.len(); + let delay = Duration::from_millis(200); + let reader = std::thread::spawn(move || { + std::thread::sleep(delay); + let mut received = vec![0; text_len]; + peer.read_exact(&mut received) + .expect("peer receives prompt"); + let prompt_completed = Instant::now(); + let mut enter = [0; 1]; + peer.read_exact(&mut enter).expect("peer receives enter"); + let enter_received = Instant::now(); + let mut user = [0; 4]; + peer.read_exact(&mut user) + .expect("peer receives queued input"); + (prompt_completed, enter_received, enter, user) + }); + + let completion = handle + .queue_user_input_submission(text, Bytes::from_static(b"\r"), delay) + .expect("submission queues"); + handle + .try_write_user_input(Bytes::from_static(b"user")) + .expect("ordinary input queues behind submission"); + completion + .recv() + .expect("actor reports submission") + .expect("submission completes"); + let (prompt_completed, enter_received, enter, user) = reader.join().expect("reader joins"); + + assert_eq!(enter, *b"\r"); + assert_eq!(user, *b"user"); + assert!(enter_received.duration_since(prompt_completed) >= delay / 2); + + let err = match handle.queue_user_input_submission( + Bytes::from_static(b"prompt"), + Bytes::from_static(b"\r"), + Duration::ZERO, + ) { + Ok(completion) => completion + .recv() + .expect("actor reports submission") + .expect_err("closed PTY rejects submission"), + Err(err) => err, + }; + + assert!(matches!( + err.kind(), + std::io::ErrorKind::BrokenPipe + | std::io::ErrorKind::ConnectionReset + | std::io::ErrorKind::WriteZero + )); + } + + #[test] + fn actor_completes_empty_submission_parts() { + let (handle, mut peer, _read_rx) = actor_with_socket_pair(false); + peer.set_read_timeout(Some(Duration::from_secs(1))) + .expect("peer timeout"); + + let completion = handle + .queue_user_input_submission(Bytes::new(), Bytes::from_static(b"\r"), Duration::ZERO) + .expect("empty prompt submission queues"); + let mut enter = [0; 1]; + peer.read_exact(&mut enter) + .expect("peer receives enter for empty prompt"); + assert_eq!(enter, *b"\r"); + completion + .recv_timeout(Duration::from_secs(1)) + .expect("actor reports empty prompt submission") + .expect("empty prompt submission completes"); + + let completion = handle + .queue_user_input_submission( + Bytes::from_static(b"prompt"), + Bytes::new(), + Duration::from_millis(40), + ) + .expect("empty enter submission queues"); + let handoff_handle = handle.clone(); + let handoff = + std::thread::spawn(move || handoff_handle.begin_handoff(Duration::from_millis(250))); + let mut prompt = [0; 6]; + peer.read_exact(&mut prompt) + .expect("peer receives prompt before empty enter"); + assert_eq!(&prompt, b"prompt"); + completion + .recv_timeout(Duration::from_secs(1)) + .expect("actor reports empty enter submission") + .expect("empty enter submission completes"); + handoff + .join() + .expect("handoff thread joins") + .expect("handoff resumes without an idle poll after submission"); + handle.shutdown(); + } + + #[test] + fn actor_reports_peer_closure_during_submission_delay() { + let (handle, mut peer, _read_rx) = actor_with_socket_pair(false); + let completion = handle + .queue_user_input_submission( + Bytes::from_static(b"prompt"), + Bytes::from_static(b"\r"), + Duration::from_secs(1), + ) + .expect("submission queues"); + let mut prompt = [0; 6]; + peer.read_exact(&mut prompt).expect("peer receives prompt"); + drop(peer); + + let err = completion + .recv_timeout(Duration::from_secs(1)) + .expect("actor reports peer closure") + .expect_err("peer closure fails the active submission"); + assert_eq!(err.kind(), std::io::ErrorKind::BrokenPipe); + } + #[test] fn actor_wakes_idle_poll_for_user_input() { let (poll_tx, poll_rx) = std_mpsc::channel(); @@ -976,9 +1326,9 @@ mod tests { .expect("actor spawn"); let marker = Bytes::from_static(b"queued-input"); - handle - .try_write_user_input(marker.clone()) - .expect("write command accepted"); + let completion = handle + .queue_user_input_submission(marker.clone(), Bytes::from_static(b"\r"), Duration::ZERO) + .expect("submission accepted"); const OUTPUT_LEN: usize = 128 * 1024; let mut peer_writer = peer.try_clone().expect("clone peer writer"); @@ -1004,11 +1354,27 @@ mod tests { assert_eq!(output_len, OUTPUT_LEN); output_writer.join().expect("output writer joins"); - let mut received_input = vec![0; prefilled + marker.len()]; + let handoff_handle = handle.clone(); + let handoff = + std::thread::spawn(move || handoff_handle.begin_handoff(Duration::from_secs(1))); + + let mut received_input = vec![0; prefilled + marker.len() + 1]; peer.read_exact(&mut received_input) .expect("peer receives prefill and queued input"); assert!(received_input[..prefilled].iter().all(|byte| *byte == 0xAA)); - assert_eq!(&received_input[prefilled..], marker.as_ref()); + assert_eq!( + &received_input[prefilled..prefilled + marker.len()], + marker.as_ref() + ); + assert_eq!(received_input.last(), Some(&b'\r')); + completion + .recv_timeout(Duration::from_secs(1)) + .expect("actor reports submission") + .expect("submission completes"); + handoff + .join() + .expect("handoff thread joins") + .expect("handoff waits for submission"); handle.shutdown(); } @@ -1080,6 +1446,10 @@ mod tests { handle .begin_handoff(Duration::from_secs(1)) .expect("handoff quiesced"); + let err = handle + .begin_handoff(Duration::from_secs(1)) + .expect_err("concurrent handoff rejected"); + assert_eq!(err.kind(), std::io::ErrorKind::WouldBlock); assert!(handle .try_write_user_input(Bytes::from_static(b"blocked")) .is_err()); @@ -1205,6 +1575,8 @@ mod tests { state: ActorState::Running, pending_writes: VecDeque::new(), current_write_offset: 0, + active_submission: None, + pending_handoff: None, wake_read_fd: wake_pipe.read_fd, controls: Arc::clone(&controls), response_order: Arc::clone(&response_order), @@ -1251,8 +1623,14 @@ mod tests { assert_eq!( runner.pending_writes, VecDeque::from([ - Bytes::from_static(b"live-light"), - Bytes::from_static(b"query-light"), + PendingWrite { + bytes: Bytes::from_static(b"live-light"), + boundary: None, + }, + PendingWrite { + bytes: Bytes::from_static(b"query-light"), + boundary: None, + }, ]) ); } @@ -1271,107 +1649,6 @@ mod tests { handle.shutdown(); } - #[tokio::test] - async fn async_user_input_waits_for_queue_capacity() { - let (data_tx, mut data_rx) = mpsc::channel(1); - let (control_tx, _control_rx) = std_mpsc::channel(); - data_tx - .try_send(PtyIoDataCommand::WriteUserInput(Bytes::from_static( - b"fill", - ))) - .expect("fill data queue"); - let (wake, _wake_read_fd) = test_wake_pair(); - let handle = PtyIoActorHandle { - data_tx, - control_tx, - wake, - user_writes: Arc::new(Mutex::new(UserWriteGate { accepting: true })), - controls: Arc::new(Mutex::new(SharedPtyControls::default())), - response_order: Arc::new(Mutex::new(())), - }; - - let write = tokio::spawn(async move { - handle - .write_user_input(Bytes::from_static(b"wait-for-capacity")) - .await - }); - tokio::time::sleep(Duration::from_millis(50)).await; - assert!( - !write.is_finished(), - "async input should wait for queue capacity" - ); - - assert!(matches!( - data_rx.recv().await, - Some(PtyIoDataCommand::WriteUserInput(_)) - )); - write - .await - .expect("write task joins") - .expect("write succeeds after capacity opens"); - match data_rx.recv().await { - Some(PtyIoDataCommand::WriteUserInput(bytes)) => { - assert_eq!(bytes, Bytes::from_static(b"wait-for-capacity")); - } - _ => panic!("expected queued user input"), - } - } - - #[tokio::test] - async fn async_user_input_waiting_for_capacity_is_rejected_after_handoff_begins() { - let (data_tx, mut data_rx) = mpsc::channel(1); - let (control_tx, control_rx) = std_mpsc::channel(); - data_tx - .try_send(PtyIoDataCommand::WriteUserInput(Bytes::from_static( - b"fill", - ))) - .expect("fill data queue"); - let (wake, _wake_read_fd) = test_wake_pair(); - let handle = PtyIoActorHandle { - data_tx, - control_tx, - wake, - user_writes: Arc::new(Mutex::new(UserWriteGate { accepting: true })), - controls: Arc::new(Mutex::new(SharedPtyControls::default())), - response_order: Arc::new(Mutex::new(())), - }; - let write_handle = handle.clone(); - let write = tokio::spawn(async move { - write_handle - .write_user_input(Bytes::from_static(b"after-handoff-start")) - .await - }); - tokio::time::sleep(Duration::from_millis(50)).await; - - let handoff = std::thread::spawn(move || handle.begin_handoff(Duration::from_secs(1))); - match control_rx - .recv_timeout(Duration::from_secs(1)) - .expect("handoff control command") - { - PtyIoControlCommand::BeginHandoff(reply) => { - reply.send(Ok(())).expect("handoff waiter alive"); - } - _ => panic!("expected begin handoff command"), - } - handoff - .join() - .expect("handoff thread joins") - .expect("handoff succeeds"); - assert!(matches!( - data_rx.recv().await, - Some(PtyIoDataCommand::WriteUserInput(_)) - )); - - let err = write.await.expect("write task joins").expect_err( - "write waiting for capacity must be rejected after handoff closes the input gate", - ); - assert_eq!(err.0, Bytes::from_static(b"after-handoff-start")); - match tokio::time::timeout(Duration::from_millis(50), data_rx.recv()).await { - Err(_) | Ok(None) => {} - Ok(Some(_)) => panic!("rejected write must not be queued"), - } - } - #[test] fn handoff_control_is_not_blocked_by_full_data_queue() { let (data_tx, _data_rx) = mpsc::channel(1); @@ -1431,6 +1708,8 @@ mod tests { state: ActorState::Running, pending_writes: VecDeque::new(), current_write_offset: 0, + active_submission: None, + pending_handoff: None, wake_read_fd: fd::create_wake_pipe().expect("wake pipe").read_fd, controls: Arc::new(Mutex::new(SharedPtyControls::default())), response_order: Arc::new(Mutex::new(())), diff --git a/src/server/headless.rs b/src/server/headless.rs index c77462ab..1efa2d1a 100644 --- a/src/server/headless.rs +++ b/src/server/headless.rs @@ -3037,6 +3037,12 @@ impl HeadlessServer { return changed; } let alt_screen_read_spec = self.alt_screen_read_spec(&msg.request); + if matches!(&msg.request.method, api::schema::Method::AgentPrompt(_)) { + let deferred_changed = self + .app + .handle_deferred_agent_api_request(msg.request, msg.respond_to); + return changed | deferred_changed; + } if matches!( &msg.request.method, api::schema::Method::WorktreeCreate(_) | api::schema::Method::WorktreeRemove(_) diff --git a/src/terminal/runtime.rs b/src/terminal/runtime.rs index 025983a3..66484ea9 100644 --- a/src/terminal/runtime.rs +++ b/src/terminal/runtime.rs @@ -449,8 +449,13 @@ impl TerminalRuntime { self.0.try_send_bytes(bytes) } - pub fn send_bytes_after(&self, bytes: Bytes, delay: std::time::Duration) { - self.0.send_bytes_after(bytes, delay); + pub fn queue_user_input_submission( + &self, + text: Bytes, + enter: Bytes, + delay: std::time::Duration, + ) -> std::io::Result>> { + self.0.queue_user_input_submission(text, enter, delay) } pub fn try_send_paste(&self, text: String) -> Result<(), mpsc::error::TrySendError> { -- cgit v1.2.3