Fix Windows ConPTY worker interrupts

Author: t-kalinowski

src/ipc/transport.rs

@@ -157,13 +157,16 @@ impl IpcServer {
     }
 
     #[cfg(target_family = "windows")]
-    pub fn connect(
+    pub fn connect<ChildExited>(
         self,
         handle: IpcHandle,
         handlers: IpcHandlers,
-        child: &mut std::process::Child,
+        child_exited: ChildExited,
         max_wait: Duration,
-    ) -> io::Result<()> {
+    ) -> io::Result<()>
+    where
+        ChildExited: FnMut() -> io::Result<bool>,
+    {
         let Some(server_pipe_to_worker) = self.server_pipe_to_worker else {
             return Err(io::Error::other(
                 "missing ipc named pipe handle (to-worker)",
@@ -175,9 +178,14 @@ impl IpcServer {
             ));
         };
         let start = Instant::now();
-        connect_named_pipe_with_process_retry(&server_pipe_to_worker, child, max_wait)?;
+        let mut child_exited = child_exited;
+        connect_named_pipe_with_process_retry(&server_pipe_to_worker, &mut child_exited, max_wait)?;
         let remaining = max_wait.saturating_sub(start.elapsed());
-        connect_named_pipe_with_process_retry(&server_pipe_from_worker, child, remaining)?;
+        connect_named_pipe_with_process_retry(
+            &server_pipe_from_worker,
+            &mut child_exited,
+            remaining,
+        )?;
         let conn = ServerIpcConnection::new(
             IpcTransport {
                 reader: Box::new(server_pipe_from_worker),
@@ -507,12 +515,12 @@ fn join_connector_with_grace(connector: thread::JoinHandle<()>, max_wait: Durati
 #[cfg(target_family = "windows")]
 fn connect_named_pipe_with_process_retry(
     server_pipe: &File,
-    child: &mut std::process::Child,
+    child_exited: &mut impl FnMut() -> io::Result<bool>,
     max_wait: Duration,
 ) -> io::Result<()> {
     connect_named_pipe_with_process_retry_impl(
         |timeout| connect_named_pipe(server_pipe, timeout),
-        || child.try_wait().map(|status| status.is_some()),
+        child_exited,
         max_wait,
     )
 }

src/python_session.rs

@@ -203,6 +203,8 @@ fn finish_active_request_at_next_read() {
     let mut guard = state.inner.lock().unwrap();
     guard.waiting_for_input = false;
     if let Some(active) = guard.active_request.as_mut() {
+        #[cfg(windows)]
+        active.queued_lines.clear();
         active.line_count = active.consumed_lines.saturating_add(1);
         active.fallback_prompt = None;
         active.skip_next_hook = false;

src/server/response.rs

@@ -1739,6 +1739,24 @@ fn omission_event_line_len() -> usize {
     build_events_log_server_line(OUTPUT_BUNDLE_OMITTED_NOTICE).len()
 }
 
+fn terminal_mode_toggle_only(text: &str) -> bool {
+    let mut rest = text;
+    let mut saw_toggle = false;
+    loop {
+        if let Some(next) = rest.strip_prefix("\u{1b}[?9001h") {
+            rest = next;
+            saw_toggle = true;
+            continue;
+        }
+        if let Some(next) = rest.strip_prefix("\u{1b}[?1004h") {
+            rest = next;
+            saw_toggle = true;
+            continue;
+        }
+        return saw_toggle && rest.trim_matches(['\r', '\n']).is_empty();
+    }
+}
+
 /// Normalizes one worker reply into renderable items while preserving the split between
 /// worker-originated transcript text and inline-only server notices.
 fn prepare_reply_material(reply: WorkerReply, detached_prefix_item_count: usize) -> ReplyMaterial {
@@ -1772,6 +1790,13 @@ fn prepare_reply_material(reply: WorkerReply, detached_prefix_item_count: usize)
                 } else {
                     text
                 };
+                if matches!(
+                    (origin, stream),
+                    (ContentOrigin::Worker, TextStream::Stdout)
+                ) && terminal_mode_toggle_only(&text)
+                {
+                    continue;
+                }
                 if text.is_empty() {
                     continue;
                 }
@@ -2797,6 +2822,54 @@ mod tests {
         assert_eq!(normalize_error_prompt(text.clone(), true), text);
     }
 
+    #[test]
+    fn standalone_terminal_mode_toggles_are_dropped_from_inline_reply() {
+        let mut state = ResponseState::new().expect("response state should initialize");
+        let result = state.finalize_worker_result(
+            Ok(worker_reply(
+                vec![
+                    WorkerContent::worker_stdout("\u{1b}[?9001h\u{1b}[?1004h"),
+                    WorkerContent::worker_stdout("2\n"),
+                ],
+                None,
+            )),
+            false,
+            TimeoutBundleReuse::None,
+            0,
+        );
+
+        assert_eq!(result_text(&result), "2\n");
+    }
+
+    #[test]
+    fn standalone_terminal_mode_toggles_are_dropped_from_output_bundle() {
+        let mut state = ResponseState::new().expect("response state should initialize");
+        let visible = "x".repeat(super::INLINE_TEXT_HARD_SPILL_THRESHOLD + 200);
+        let result = state.finalize_worker_result(
+            Ok(worker_reply(
+                vec![
+                    WorkerContent::worker_stdout("\u{1b}[?9001h\u{1b}[?1004h"),
+                    WorkerContent::worker_stdout(visible.clone()),
+                ],
+                None,
+            )),
+            false,
+            TimeoutBundleReuse::None,
+            0,
+        );
+
+        let text = result_text(&result);
+        let transcript_path = disclosed_path(&text, "transcript.txt")
+            .unwrap_or_else(|| panic!("expected transcript path, got: {text:?}"));
+        let transcript = fs::read_to_string(&transcript_path)
+            .unwrap_or_else(|err| panic!("expected transcript to be readable: {err}"));
+        assert_eq!(transcript, visible);
+        assert!(
+            !text.contains("\u{1b}[?9001h") && !text.contains("\u{1b}[?1004h"),
+            "did not expect terminal mode toggles inline: {text:?}"
+        );
+    }
+
     #[test]
     fn events_log_text_rows_preserve_partial_line_state_across_images() {
         let mut store = OutputStore::new().expect("output store should initialize");

src/windows_conpty.rs

@@ -179,10 +179,7 @@ pub fn run_conpty_command_with_env_map(
         upsert_env_case_insensitive(&mut env_map, WINDOWS_CONPTY_REQUEST_ENV, "1");
         upsert_env_case_insensitive(&mut env_map, WINDOWS_CONPTY_ATTACHED_ENV, "1");
         crate::diagnostics::startup_log("windows-conpty: conpty created");
-        let output_read = conpty
-            .output_read
-            .try_clone()
-            .map_err(|err| format!("failed to clone ConPTY output handle: {err}"))?;
+        let output_read = conpty.take_output_reader()?;
         let output_forwarder = spawn_conpty_output_forwarder(output_read);
         crate::diagnostics::startup_log("windows-conpty: spawning child");
         let proc_info = spawn_conpty_process(command, cwd, &env_map, conpty.hpc)?;
@@ -270,22 +267,31 @@ pub unsafe fn spawn_conpty_process_as_user(
     let mut conpty = Conpty::new()?;
     upsert_env_case_insensitive(env_map, WINDOWS_CONPTY_REQUEST_ENV, "1");
     upsert_env_case_insensitive(env_map, WINDOWS_CONPTY_ATTACHED_ENV, "1");
-    let output_read = conpty
-        .output_read
-        .try_clone()
-        .map_err(|err| format!("failed to clone ConPTY output handle: {err}"))?;
+    let output_read = conpty.take_output_reader()?;
     let output_forwarder = spawn_conpty_output_forwarder(output_read);
     let proc_info = spawn_conpty_process_with_token(token, command, cwd, env_map, conpty.hpc)?;
     conpty.close_child_side_handles();
     Ok((proc_info, conpty, output_forwarder))
 }
 
+pub unsafe fn spawn_conpty_process_direct(
+    command: &[String],
+    cwd: Option<&Path>,
+    env_map: &mut HashMap<String, String>,
+) -> Result<(PROCESS_INFORMATION, Conpty), String> {
+    let mut conpty = Conpty::new()?;
+    upsert_env_case_insensitive(env_map, WINDOWS_CONPTY_REQUEST_ENV, "1");
+    upsert_env_case_insensitive(env_map, WINDOWS_CONPTY_ATTACHED_ENV, "1");
+    let proc_info = spawn_conpty_process(command, cwd, env_map, conpty.hpc)?;
+    conpty.close_child_side_handles();
+    Ok((proc_info, conpty))
+}
+
 pub struct Conpty {
     hpc: HPCON,
     input_read: Option<File>,
-    #[allow(dead_code)]
-    input_write: File,
-    output_read: File,
+    input_write: Option<File>,
+    output_read: Option<File>,
     output_write: Option<File>,
 }
 
@@ -342,12 +348,24 @@ impl Conpty {
         Ok(Self {
             hpc,
             input_read: Some(File::from_raw_handle(input_read as _)),
-            input_write: File::from_raw_handle(input_write as _),
-            output_read: File::from_raw_handle(output_read as _),
+            input_write: Some(File::from_raw_handle(input_write as _)),
+            output_read: Some(File::from_raw_handle(output_read as _)),
             output_write: Some(File::from_raw_handle(output_write as _)),
         })
     }
 
+    pub fn take_input_writer(&mut self) -> Result<File, String> {
+        self.input_write
+            .take()
+            .ok_or_else(|| "ConPTY input writer already taken".to_string())
+    }
+
+    pub fn take_output_reader(&mut self) -> Result<File, String> {
+        self.output_read
+            .take()
+            .ok_or_else(|| "ConPTY output reader already taken".to_string())
+    }
+
     fn close_child_side_handles(&mut self) {
         self.input_read.take();
         self.output_write.take();