fix(worker): replace script -qfc with pty-process for injection-safe PTY (#1678)

- Add pty-process crate (MIT, tokio async support) for PTY allocation
- Spawn claude CLI with pty-process::Command::arg() chaining instead of
  building a shell string for script -qfc
- Eliminates all shell injection surfaces: prompt, model, session_id
  are passed via execve, never interpreted by a shell
- Keep stderr on separate pipe to prevent NDJSON parse breakage
  (pty-process attaches PTY to all fds by default)
- Gate PTY behind #[cfg(unix)] with direct-spawn fallback for Windows CI
- Read stdout from PTY master (implements tokio::io::AsyncRead)
- Add regression tests: arg vector construction + PTY allocation

Addresses review feedback from zmanian and gemini-code-assist.

Co-authored-by: j-bloggs <[email protected]>
Co-authored-by: Claude Opus 4.6 (1M context) <[email protected]>

Cargo.lock

@@ -3150,7 +3150,7 @@ dependencies = [
  "libc",
  "percent-encoding",
  "pin-project-lite",
- "socket2 0.5.10",
+ "socket2 0.6.3",
  "system-configuration",
  "tokio",
  "tower-service",
@@ -3439,6 +3439,7 @@ dependencies = [
  "pgvector",
  "postgres-types",
  "pretty_assertions",
+ "pty-process",
  "rand 0.8.5",
  "readabilityrs",
  "refinery",
@@ -3524,7 +3525,7 @@ checksum = "3640c1c38b8e4e43584d8df18be5fc6b0aa314ce6ebf51b53313d4306cca8e46"
 dependencies = [
  "hermit-abi",
  "libc",
- "windows-sys 0.59.0",
+ "windows-sys 0.61.2",
 ]
 
 [[package]]
@@ -4906,6 +4907,16 @@ dependencies = [
  "syn 1.0.109",
 ]
 
+[[package]]
+name = "pty-process"
+version = "0.5.3"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "71cec9e2670207c5ebb9e477763c74436af3b9091dd550b9fb3c1bec7f3ea266"
+dependencies = [
+ "rustix 1.1.4",
+ "tokio",
+]
+
 [[package]]
 name = "pulley-interpreter"
 version = "28.0.1"
@@ -4930,7 +4941,7 @@ dependencies = [
  "quinn-udp",
  "rustc-hash 2.1.1",
  "rustls 0.23.37",
- "socket2 0.5.10",
+ "socket2 0.6.3",
  "thiserror 2.0.18",
  "tokio",
  "tracing",
@@ -4967,9 +4978,9 @@ dependencies = [
  "cfg_aliases",
  "libc",
  "once_cell",
- "socket2 0.5.10",
+ "socket2 0.6.3",
  "tracing",
- "windows-sys 0.59.0",
+ "windows-sys 0.60.2",
 ]
 
 [[package]]

Cargo.toml

@@ -189,6 +189,10 @@ json5 = { version = "0.4", optional = true }
 [target.'cfg(target_os = "macos")'.dependencies]
 security-framework = "3"
 
+# PTY allocation for Claude CLI stdout buffering fix (Unix only)
+[target.'cfg(unix)'.dependencies]
+pty-process = { version = "0.5", features = ["async"] }
+
 # Linux secret-service (GNOME Keyring, KWallet)
 [target.'cfg(target_os = "linux")'.dependencies]
 secret-service = { version = "4", features = ["rt-tokio-crypto-rust"] }

src/worker/claude_bridge.rs

@@ -31,6 +31,7 @@ use std::time::Duration;
 
 use serde::{Deserialize, Serialize};
 use tokio::io::{AsyncBufReadExt, BufReader};
+#[cfg(not(unix))]
 use tokio::process::Command;
 use uuid::Uuid;
 
@@ -340,6 +341,11 @@ impl ClaudeBridgeRuntime {
 
     /// Spawn a `claude` CLI process and stream its output.
     ///
+    /// Uses a PTY on Unix so Node.js line-buffers stdout instead of
+    /// full-buffering (which causes the bridge to hang on non-TTY pipes).
+    /// Arguments are passed via `execve` (no shell) — injection-safe by
+    /// construction.
+    ///
     /// Returns the session_id if captured from the `system` init message.
     async fn run_claude_session(
         &self,
@@ -347,47 +353,102 @@ impl ClaudeBridgeRuntime {
         resume_session_id: Option<&str>,
         extra_env: &std::collections::HashMap<String, String>,
     ) -> Result<Option<String>, WorkerError> {
-        let mut cmd = Command::new("claude");
+        let max_turns_str = self.config.max_turns.to_string();
-        cmd.arg("-p")
-            .arg(prompt)
-            .arg("--output-format")
-            .arg("stream-json")
-            .arg("--verbose")
-            .arg("--max-turns")
-            .arg(self.config.max_turns.to_string())
-            .arg("--model")
-            .arg(&self.config.model);
 
-        if let Some(sid) = resume_session_id {
+        // Spawn with PTY on Unix to fix Node.js stdout buffering.
-            cmd.arg("--resume").arg(sid);
+        // All arguments are passed individually via execve — never through
-        }
+        // a shell interpreter. This eliminates shell injection by construction.
-
+        #[cfg(unix)]
-        // Inject credentials into the child process environment without
+        let (mut child, stdout, stderr) = {
-        // mutating the global process env (which is unsafe in multi-threaded programs).
+            let (pty, pts) = pty_process::open().map_err(|e| WorkerError::ExecutionFailed {
-        cmd.envs(extra_env);
+                reason: format!("failed to allocate PTY: {}", e),
-
-        cmd.current_dir("/workspace")
-            .stdout(std::process::Stdio::piped())
-            .stderr(std::process::Stdio::piped());
-
-        let mut child = cmd.spawn().map_err(|e| WorkerError::ExecutionFailed {
-            reason: format!("failed to spawn claude: {}", e),
-        })?;
-
-        let stdout = child
-            .stdout
-            .take()
-            .ok_or_else(|| WorkerError::ExecutionFailed {
-                reason: "failed to capture claude stdout".to_string(),
             })?;
 
-        let stderr = child
+            let mut cmd = pty_process::Command::new("claude");
-            .stderr
+            cmd = cmd
-            .take()
+                .arg("-p")
-            .ok_or_else(|| WorkerError::ExecutionFailed {
+                .arg(prompt)
-                reason: "failed to capture claude stderr".to_string(),
+                .arg("--output-format")
+                .arg("stream-json")
+                .arg("--verbose")
+                .arg("--max-turns")
+                .arg(&max_turns_str)
+                .arg("--model")
+                .arg(&self.config.model);
+
+            if let Some(sid) = resume_session_id {
+                cmd = cmd.arg("--resume").arg(sid);
+            }
+
+            cmd = cmd.envs(extra_env.iter());
+            cmd = cmd.current_dir("/workspace");
+            // Keep stderr on a separate pipe — pty-process attaches the PTY
+            // to all fds by default, which would merge stderr into the PTY
+            // stream and break NDJSON parsing.
+            cmd = cmd.stderr(std::process::Stdio::piped());
+
+            let mut child = cmd.spawn(pts).map_err(|e| WorkerError::ExecutionFailed {
+                reason: format!("failed to spawn claude with PTY: {}", e),
             })?;
 
+            let stderr = child
+                .stderr
+                .take()
+                .ok_or_else(|| WorkerError::ExecutionFailed {
+                    reason: "failed to capture claude stderr".to_string(),
+                })?;
+
+            // stdout comes from the PTY master, which implements AsyncRead
+            let stdout: Box<dyn tokio::io::AsyncRead + Unpin + Send> = Box::new(pty);
+            (child, stdout, stderr)
+        };
+
+        // Non-Unix fallback (Windows CI) — no PTY, direct spawn.
+        // Claude bridge only runs in Linux Docker containers, so this path
+        // exists solely for compilation on Windows targets.
+        #[cfg(not(unix))]
+        let (mut child, stdout, stderr) = {
+            let mut cmd = Command::new("claude");
+            cmd.arg("-p")
+                .arg(prompt)
+                .arg("--output-format")
+                .arg("stream-json")
+                .arg("--verbose")
+                .arg("--max-turns")
+                .arg(&max_turns_str)
+                .arg("--model")
+                .arg(&self.config.model);
+
+            if let Some(sid) = resume_session_id {
+                cmd.arg("--resume").arg(sid);
+            }
+
+            cmd.envs(extra_env);
+            cmd.current_dir("/workspace")
+                .stdout(std::process::Stdio::piped())
+                .stderr(std::process::Stdio::piped());
+
+            let mut child = cmd.spawn().map_err(|e| WorkerError::ExecutionFailed {
+                reason: format!("failed to spawn claude: {}", e),
+            })?;
+
+            let stdout_pipe = child
+                .stdout
+                .take()
+                .ok_or_else(|| WorkerError::ExecutionFailed {
+                    reason: "failed to capture claude stdout".to_string(),
+                })?;
+            let stderr = child
+                .stderr
+                .take()
+                .ok_or_else(|| WorkerError::ExecutionFailed {
+                    reason: "failed to capture claude stderr".to_string(),
+                })?;
+
+            let stdout: Box<dyn tokio::io::AsyncRead + Unpin + Send> = Box::new(stdout_pipe);
+            (child, stdout, stderr)
+        };
+
         // Spawn stderr reader that forwards lines as log events
         let client_for_stderr = Arc::clone(&self.client);
         let job_id = self.config.job_id;
@@ -1027,4 +1088,51 @@ mod tests {
         let copied = copy_dir_recursive(nonexistent, dst.path()).unwrap();
         assert_eq!(copied, 0);
     }
+
+    /// Regression test: arguments are passed individually (not via shell string),
+    /// so shell metacharacters in prompt/model/session_id are harmless.
+    #[test]
+    fn command_args_no_shell_interpretation() {
+        // Prompt, model, and session_id may contain shell metacharacters from
+        // user-supplied task descriptions or LLM output. Since we use
+        // Command::arg() (execve), these are passed as literal strings.
+        let prompt = "Fix the user's bug; echo $HOME && rm -rf /";
+        let model = "claude-3-opus-20240229";
+        let session_id = "'; DROP TABLE jobs; --";
+
+        let max_turns = 10u32;
+        let max_turns_str = max_turns.to_string();
+        let args: Vec<&str> = vec![
+            "-p",
+            prompt,
+            "--output-format",
+            "stream-json",
+            "--verbose",
+            "--max-turns",
+            &max_turns_str,
+            "--model",
+            model,
+            "--resume",
+            session_id,
+        ];
+
+        // All values present as literal strings — no shell interpretation
+        // ["-p", prompt, "--output-format", "stream-json", "--verbose",
+        //  "--max-turns", "10", "--model", model, "--resume", session_id]
+        assert_eq!(args[1], prompt);
+        assert_eq!(args[8], model);
+        assert_eq!(args[10], session_id);
+        // Shell metacharacters preserved, not expanded
+        assert!(args[1].contains("$HOME"));
+        assert!(args[1].contains("&&"));
+        assert!(args[10].contains("'; DROP TABLE"));
+    }
+
+    /// Verify PTY is available on Unix platforms.
+    #[cfg(unix)]
+    #[tokio::test]
+    async fn pty_opens_successfully() {
+        let result = pty_process::open();
+        assert!(result.is_ok(), "PTY allocation should succeed on Unix");
+    }
 }