feat: enhance subagent context with abort flag and implement tool call timeout

2026-07-13 04:59:16 +07:00
parent 3b711bbf3b
commit 2856dd78b8
9 changed files with 272 additions and 131 deletions
@@ -54,65 +54,26 @@ pub fn spawn_bash_job(command: String) -> BashJob {
    let (pid_tx, pid_rx) = mpsc::channel::<u32>();
    let cmd = command.clone();
    let id_for_log = id.clone();
+    let thread_id = id.clone();

-    thread::spawn(move || {
-        let mut child = match Command::new("sh")
-            .arg("-c")
-            .arg(&cmd)
-            .stdout(Stdio::piped())
-            .stderr(Stdio::piped())
-            .spawn()
-        {
-            Ok(c) => c,
-            Err(e) => {
-                let _ = output_tx.try_send(format!("__error:{}", e));
-                let _ = output_tx.try_send("__exit:-1".to_string());
-                return;
-            }
-        };
-
-        // Send the child PID back to the caller so bash_kill can terminate it
-        let _ = pid_tx.send(child.id());
-
-        // Drain stderr on a separate thread to prevent deadlock when
-        // the child produces more than ~64 KB of stderr after closing
-        // stdout (the pipe buffer fills and the child blocks on write,
-        // while the parent thread waits for the child to exit).
-        let stderr_tx = output_tx.clone();
-        let _stderr_drain = child.stderr.take().map(|stderr| {
-            std::thread::spawn(move || {
-                let reader = std::io::BufReader::new(stderr);
-                // stderr is intentionally discarded to prevent output-line
-                // quota pressure from error diagnostics.
-                for _line in reader.lines().map_while(Result::ok) {
-                    // Discard stderr lines to prevent pipe buffer deadlock.
-                }
-                drop(stderr_tx);
-            })
+    // Spawn a named thread for easier debugging. If Builder::spawn fails
+    // (e.g. OS resource limit), fall back to unnameable thread::spawn.
+    let thread_name = format!("bgbash-{}", &thread_id[..8.min(thread_id.len())]);
+    if thread::Builder::new().name(thread_name).spawn({
+        // Clone everything the closure captures so we can also pass it
+        // to the fallback thread without moving.
+        let cmd = cmd.clone();
+        let output_tx = output_tx.clone();
+        let pid_tx = pid_tx.clone();
+        let id_for_log = id_for_log.clone();
+        move || spawn_bash_thread_body(cmd, output_tx, pid_tx, id_for_log)
+    }).is_err()
+    {
+        tracing::warn!("[bgbash:{}] failed to spawn named thread, using unnamed fallback", id_for_log);
+        thread::spawn(move || {
+            spawn_bash_thread_body(cmd, output_tx, pid_tx, id_for_log)
        });
-
-        if let Some(stdout) = child.stdout.take() {
-            let reader = std::io::BufReader::new(stdout);
-            for line in reader.lines().map_while(Result::ok) {
-                // Use try_send so if the channel buffer is full (producer
-                // faster than consumer), old lines are silently dropped
-                // rather than growing memory without bound.
-                if output_tx.try_send(line).is_err() {
-                    // Buffer full — consumer is not draining fast enough.
-                    // Stop reading to apply backpressure; remaining output
-                    // is lost but the process will eventually drain.
-                    tracing::debug!(
-                        "[bgbash:{}] output buffer full ({} lines), discarding remaining output",
-                        id_for_log, MAX_OUTPUT_LINES,
-                    );
-                    break;
-                }
-            }
-        }
-        let status = child.wait();
-        let code = status.ok().and_then(|s| s.code());
-        let _ = output_tx.try_send(format!("__exit:{}", code.unwrap_or(-1)));
-    });
+    }

    let child_pid = pid_rx.recv().unwrap_or(0);

@@ -124,6 +85,71 @@ pub fn spawn_bash_job(command: String) -> BashJob {
    }
 }

+/// Core bash-thread logic extracted into a free function so it can be
+/// spawned from both the named Builder and the unnamed fallback without
+/// double-moving the closure.
+fn spawn_bash_thread_body(
+    cmd: String,
+    output_tx: std::sync::mpsc::SyncSender<String>,
+    pid_tx: std::sync::mpsc::Sender<u32>,
+    id_for_log: String,
+) {
+    let mut child = match Command::new("sh")
+        .arg("-c")
+        .arg(&cmd)
+        .stdout(Stdio::piped())
+        .stderr(Stdio::piped())
+        .spawn()
+    {
+        Ok(c) => c,
+        Err(e) => {
+            let _ = output_tx.try_send(format!("__error:{}", e));
+            let _ = output_tx.try_send("__exit:-1".to_string());
+            return;
+        }
+    };
+
+    // Send the child PID back to the caller so bash_kill can terminate it
+    let _ = pid_tx.send(child.id());
+
+    // Drain stderr on a separate thread to prevent deadlock when
+    // the child produces more than ~64 KB of stderr after closing
+    // stdout (the pipe buffer fills and the child blocks on write,
+    // while the parent thread waits for the child to exit).
+    // Stderr lines are now prefixed with "[stderr] " and sent through
+    // the output channel so users can see error diagnostics from
+    // background jobs.
+    let stderr_tx = output_tx.clone();
+    let _stderr_drain = child.stderr.take().map(|stderr| {
+        std::thread::spawn(move || {
+            let reader = std::io::BufReader::new(stderr);
+            for line in reader.lines().map_while(Result::ok) {
+                if stderr_tx.try_send(format!("[stderr] {}", line)).is_err() {
+                    tracing::debug!("[bgbash] stderr buffer full, discarding remaining stderr");
+                    break;
+                }
+            }
+            drop(stderr_tx);
+        })
+    });
+
+    if let Some(stdout) = child.stdout.take() {
+        let reader = std::io::BufReader::new(stdout);
+        for line in reader.lines().map_while(Result::ok) {
+            if output_tx.try_send(line).is_err() {
+                tracing::debug!(
+                    "[bgbash:{}] output buffer full ({} lines), discarding remaining output",
+                    id_for_log, MAX_OUTPUT_LINES,
+                );
+                break;
+            }
+        }
+    }
+    let status = child.wait();
+    let code = status.ok().and_then(|s| s.code());
+    let _ = output_tx.try_send(format!("__exit:{}", code.unwrap_or(-1)));
+}
+
 impl BashJob {
    /// Non-blocking poll for the next output line from the job's channel.
    ///