Add context-sourced command stdin

docs/public/reference/dot-language.mdx

@@ -246,6 +246,7 @@ audit [
 |---|---|---|
 | `script` | String | Shell command to execute |
 | `language` | String | `"shell"` (default) or `"python"` |
+| `stdin_source` | String | Flat runtime context key to pass to standard input. `context.NAME` first checks that exact key, then falls back to `NAME`. Strings are passed unchanged; other values use compact JSON. No newline is added. |
 | `output_schema` | String | Optional structured output validation. Accepts `routing`, `@path/to/schema.json`, or an inline JSON Schema object string. See [Structured output validation](#structured-output-validation). |
 
 ### Parallel (fan-out) nodes

docs/public/workflows/stages-and-nodes.mdx

@@ -100,12 +100,23 @@ Runs a shell script inside the configured sandbox and captures its output. The o
 
 ```dot
 test [label="Run Tests", shape=parallelogram, script="cargo test 2>&1 || true"]
+
+merge_results [
+    shape=parallelogram,
+    script="python3 scripts/merge.py",
+    stdin_source="context.parallel.results"
+]
 ```
 
 | Attribute | Description |
 |---|---|
 | `script` | The shell command to execute. Substitutes `{{ goal }}`, `{{ inputs.NAME }}`, and `{{ vars.NAME }}` — see [command node scripts](/workflows/variables#command-node-scripts) |
 | `language` | `"shell"` (default) or `"python"` |
+| `stdin_source` | Flat runtime context key to pass to the command's standard input. `context.NAME` first checks that exact key, then falls back to `NAME`. |
+
+For `stdin_source`, strings are passed unchanged. Other JSON values use compact
+JSON. Fabro does not add a newline. A missing source fails before the command
+starts.
 
 ### Human
 

docs/public/workflows/variables.mdx

@@ -175,6 +175,9 @@ Fabro keeps workflow structure static and renders workflow templates once:
 
 Templates are not supported in graph syntax, node IDs, edge structure, `import` paths, `@file` paths, child workflow paths, other file references, or any attribute besides `prompt` and `goal` — and `script`, which takes value substitution rather than templates.
 
+Command `stdin_source` values are literal context keys. Fabro resolves them at
+stage execution time, after upstream nodes have updated the workflow context.
+
 Fabro renders the graph `goal` first and stores the rendered value back onto the graph. Prompts that use `{{ goal }}` receive that rendered value.
 
 ## Undefined variables

lib/components/fabro-agent/src/lib.rs

@@ -58,9 +58,9 @@ pub use question_tools::{
     OPENAI_REQUEST_USER_INPUT_TOOL, register_question_tools,
 };
 pub use sandbox::{
-    CommandOutputCallback, DirEntry, ExecResult, ExecStreamingResult, GrepOptions, RefreshOutcome,
-    Sandbox, SandboxEvent, SandboxEventCallback, StderrCollector, StdioProcess, StdioProcessHandle,
-    format_lines_numbered, shell_quote,
+    CommandOutputCallback, DirEntry, ExecResult, ExecStreamingRequest, ExecStreamingResult,
+    GrepOptions, RefreshOutcome, Sandbox, SandboxEvent, SandboxEventCallback, StderrCollector,
+    StdioProcess, StdioProcessHandle, format_lines_numbered, shell_quote,
 };
 pub use session::{
     CompletionCoordinator, Session, SessionControlHandle, SessionInputTiming,

lib/components/fabro-agent/src/sandbox.rs

@@ -2,8 +2,8 @@
 // Re-export the delegate_sandbox! macro at crate root so existing
 // `crate::delegate_sandbox!` invocations continue to work.
 pub use fabro_sandbox::{
-    CommandOutputCallback, DirEntry, ExecResult, ExecStreamingResult, GrepOptions, RefreshOutcome,
-    Sandbox, SandboxEvent, SandboxEventCallback, SandboxFile, StderrCollector, StdioProcess,
-    StdioProcessHandle, StdioProcessTermination, WalkOptions, delegate_sandbox,
-    format_lines_numbered, shell_quote,
+    CommandOutputCallback, DirEntry, ExecResult, ExecStreamingRequest, ExecStreamingResult,
+    GrepOptions, RefreshOutcome, Sandbox, SandboxEvent, SandboxEventCallback, SandboxFile,
+    StderrCollector, StdioProcess, StdioProcessHandle, StdioProcessTermination, WalkOptions,
+    delegate_sandbox, format_lines_numbered, shell_quote,
 };

lib/components/fabro-agent/src/tools.rs

@@ -298,12 +298,11 @@ pub(crate) async fn execute_shell_command(
     );
     ctx.env
         .exec_command_streaming(
-            command,
-            Some(timeout_ms),
-            cwd,
-            tool_env.as_ref(),
-            Some(ctx.cancel.clone()),
-            None,
+            crate::ExecStreamingRequest::new(command)
+                .timeout_ms(Some(timeout_ms))
+                .working_dir(cwd)
+                .env_vars(tool_env.as_ref())
+                .cancel_token(Some(ctx.cancel.clone())),
         )
         .await
         .map_err(|e| format!("{SHELL_NO_PROCESS_RESULT}: {}", e.display_with_causes()))

lib/components/fabro-sandbox/src/daytona/mod.rs

@@ -33,9 +33,9 @@ use crate::sandbox::{
     REMOTE_WALK_TIMEOUT_MS, RefreshOutcome, optional_timeout, resolve_path, validate_bash_probe,
 };
 use crate::{
-    CommandOutputCallback, DirEntry, ExecResult, ExecStreamingResult, GrepOptions, Sandbox,
-    SandboxEvent, SandboxEventCallback, SandboxFile, StdioProcess, WalkOptions, managed_labels,
-    shell_quote,
+    CommandOutputCallback, DirEntry, ExecResult, ExecStreamingRequest, ExecStreamingResult,
+    GrepOptions, Sandbox, SandboxEvent, SandboxEventCallback, SandboxFile, StdioProcess,
+    WalkOptions, managed_labels, shell_quote,
 };
 
 /// Remediation shown when a Daytona sandbox has no usable Bash.
@@ -67,6 +67,8 @@ const DAYTONA_START_TIMEOUT: Duration = Duration::from_mins(1);
 /// Upper bound on explicit and Drop-triggered Daytona session deletion so a
 /// stalled REST call cannot block cancellation/timeout paths indefinitely.
 const DAYTONA_SESSION_CLOSE_TIMEOUT: Duration = Duration::from_secs(10);
+/// Upper bound for deleting one temporary command stdin file.
+const DAYTONA_STDIN_FILE_DELETE_TIMEOUT: Duration = Duration::from_secs(10);
 
 /// Permissions a Daytona API key needs for Fabro's snapshot and sandbox flow.
 pub const REQUIRED_DAYTONA_PERMISSIONS: &[Permissions] = &[
@@ -1799,19 +1801,31 @@ impl Sandbox for DaytonaSandbox {
 
     async fn exec_command_streaming(
         &self,
-        command: &str,
-        timeout_ms: Option<u64>,
-        working_dir: Option<&str>,
-        env_vars: Option<&HashMap<String, String>>,
-        cancel_token: Option<CancellationToken>,
-        output_callback: Option<CommandOutputCallback>,
+        request: ExecStreamingRequest<'_>,
     ) -> crate::Result<ExecStreamingResult> {
+        let ExecStreamingRequest {
+            command,
+            timeout_ms,
+            working_dir,
+            env_vars,
+            cancel_token,
+            stdin,
+            output_callback,
+        } = request;
         let sandbox = self.sandbox()?;
         let start = Instant::now();
         let cwd = working_dir.map_or_else(
             || self.working_directory().to_string(),
             |d| self.resolve_path(d),
         );
+        let stdin_file = match stdin {
+            Some(stdin) => Some(DaytonaStdinFile::create(sandbox, &stdin).await?),
+            None => None,
+        };
+        let command_with_stdin = stdin_file
+            .as_ref()
+            .map(|stdin_file| redirect_command_stdin(command, stdin_file.path()));
+        let command = command_with_stdin.as_deref().unwrap_or(command);
 
         let mut session = DaytonaSession::create(sandbox).await?;
 
@@ -1955,7 +1969,7 @@ impl Sandbox for DaytonaSandbox {
         let stdout = String::from_utf8_lossy(&stdout_seen.lock().await).into_owned();
         let stderr = String::from_utf8_lossy(&stderr_seen.lock().await).into_owned();
 
-        Ok(ExecStreamingResult {
+        let result = ExecStreamingResult {
             result: ExecResult {
                 stdout,
                 stderr,
@@ -1967,7 +1981,11 @@ impl Sandbox for DaytonaSandbox {
             },
             streams_separated,
             live_streaming: saw_live_chunk.load(Ordering::Relaxed),
-        })
+        };
+        if let Some(stdin_file) = stdin_file {
+            stdin_file.remove().await?;
+        }
+        Ok(result)
     }
 
     async fn spawn_stdio_process(
@@ -2132,6 +2150,109 @@ async fn finish_daytona_log_stream(
     }
 }
 
+/// A temporary Daytona file used to provide exact stdin bytes and EOF.
+///
+/// Daytona sessions accept input strings but do not expose a reliable EOF
+/// operation. A file redirection preserves arbitrary bytes and gives the
+/// command EOF without embedding workflow data in shell source.
+struct DaytonaStdinFile {
+    fs:   Option<daytona_sdk::FileSystemService>,
+    path: String,
+}
+
+impl DaytonaStdinFile {
+    async fn create(sandbox: &daytona_sdk::Sandbox, stdin: &[u8]) -> crate::Result<Self> {
+        let fs = sandbox
+            .fs()
+            .await
+            .map_err(|err| crate::Error::context("Failed to get Daytona file service", err))?;
+        let file = Self {
+            fs:   Some(fs),
+            path: format!("/tmp/fabro-command-stdin-{}", uuid::Uuid::new_v4()),
+        };
+        file.fs()
+            .upload_file_bytes(&file.path, stdin)
+            .await
+            .map_err(|err| crate::Error::context("Failed to upload Daytona command stdin", err))?;
+        Ok(file)
+    }
+
+    fn fs(&self) -> &daytona_sdk::FileSystemService {
+        self.fs
+            .as_ref()
+            .expect("DaytonaStdinFile used after removal")
+    }
+
+    fn path(&self) -> &str {
+        &self.path
+    }
+
+    async fn remove(mut self) -> crate::Result<()> {
+        let deletion = time::timeout(
+            DAYTONA_STDIN_FILE_DELETE_TIMEOUT,
+            self.fs().delete_file(&self.path, false),
+        )
+        .await;
+        match deletion {
+            Ok(Ok(())) => {
+                self.fs.take();
+                Ok(())
+            }
+            Ok(Err(err)) => Err(crate::Error::context(
+                "Failed to delete Daytona command stdin",
+                err,
+            )),
+            Err(_) => Err(crate::Error::message(format!(
+                "Timed out deleting Daytona command stdin after {}ms",
+                DAYTONA_STDIN_FILE_DELETE_TIMEOUT.as_millis()
+            ))),
+        }
+    }
+}
+
+impl Drop for DaytonaStdinFile {
+    fn drop(&mut self) {
+        let Some(fs) = self.fs.take() else {
+            return;
+        };
+        let path = std::mem::take(&mut self.path);
+        match Handle::try_current() {
+            Ok(handle) => {
+                handle.spawn(async move {
+                    match time::timeout(
+                        DAYTONA_STDIN_FILE_DELETE_TIMEOUT,
+                        fs.delete_file(&path, false),
+                    )
+                    .await
+                    {
+                        Ok(Ok(())) => {}
+                        Ok(Err(err)) => {
+                            tracing::warn!(
+                                error = %err,
+                                "Failed to delete Daytona command stdin from Drop"
+                            );
+                        }
+                        Err(_) => {
+                            tracing::warn!(
+                                timeout_ms =
+                                    u64::try_from(DAYTONA_STDIN_FILE_DELETE_TIMEOUT.as_millis())
+                                        .unwrap_or(u64::MAX),
+                                "Timed out deleting Daytona command stdin from Drop"
+                            );
+                        }
+                    }
+                });
+            }
+            Err(err) => {
+                tracing::warn!(
+                    error = %err,
+                    "Could not schedule Daytona command stdin cleanup"
+                );
+            }
+        }
+    }
+}
+
 /// RAII wrapper around a Daytona toolbox session.
 ///
 /// Holds the per-session [`ProcessService`] handle and the session id. Callers
@@ -2548,6 +2669,10 @@ fn wrap_bash_session_script(script: &str) -> String {
     format!("{REMOTE_BASH} -c {}", shell_quote(script))
 }
 
+fn redirect_command_stdin(command: &str, stdin_path: &str) -> String {
+    format!("(\n{command}\n) < {}", shell_quote(stdin_path))
+}
+
 /// Build a command for Daytona's streaming session transport.
 ///
 /// Both [`Sandbox::exec_command_streaming`] and the lifecycle probe call this
@@ -2567,7 +2692,7 @@ mod tests {
 
     use daytona_api_client::models::api_key_list::Permissions;
     use fabro_util::error::collect_chain;
-    use httpmock::Method::{GET, POST};
+    use httpmock::Method::{DELETE, GET, POST};
     use httpmock::{HttpMockResponse, MockServer};
 
     use super::*;
@@ -3237,6 +3362,80 @@ mod tests {
         auth.assert_async().await;
     }
 
+    #[tokio::test]
+    async fn daytona_stdin_file_uploads_exact_bytes_and_is_deleted() {
+        let server = MockServer::start_async().await;
+        let server_url = server.base_url();
+        let sandbox_response = server
+            .mock_async(|when, then| {
+                when.method(GET).path("/sandbox/sandbox-stdin");
+                then.status(200)
+                    .header("content-type", "application/json")
+                    .json_body(serde_json::json!({
+                        "id": "sandbox-stdin",
+                        "organizationId": "org-1",
+                        "name": "stdin-test",
+                        "user": "daytona",
+                        "env": {},
+                        "labels": {},
+                        "public": false,
+                        "networkBlockAll": false,
+                        "target": "us",
+                        "cpu": 2.0,
+                        "gpu": 0.0,
+                        "memory": 4.0,
+                        "disk": 20.0,
+                        "state": "started"
+                    }));
+            })
+            .await;
+        let toolbox_response = server
+            .mock_async(|when, then| {
+                when.method(GET)
+                    .path("/sandbox/sandbox-stdin/toolbox-proxy-url");
+                then.status(200)
+                    .header("content-type", "application/json")
+                    .json_body(serde_json::json!({"url": server_url}));
+            })
+            .await;
+        let upload = server
+            .mock_async(|when, then| {
+                when.method(POST)
+                    .path("/sandbox-stdin/files/upload")
+                    .body_includes("opaque\n$(not shell)\nlast");
+                then.status(200);
+            })
+            .await;
+        let delete = server
+            .mock_async(|when, then| {
+                when.method(DELETE)
+                    .path("/sandbox-stdin/files")
+                    .query_param("recursive", "false");
+                then.status(200);
+            })
+            .await;
+        let client = build_daytona_client_with(
+            Some("dtn_test".to_string()),
+            Some(server.base_url()),
+            None,
+            Some(fabro_test::test_http_client()),
+        )
+        .await
+        .expect("create Daytona client");
+        let sandbox = client.get("sandbox-stdin").await.expect("get mock sandbox");
+
+        let file = DaytonaStdinFile::create(&sandbox, b"opaque\n$(not shell)\nlast")
+            .await
+            .expect("upload stdin file");
+        assert!(file.path().starts_with("/tmp/fabro-command-stdin-"));
+        file.remove().await.expect("delete stdin file");
+
+        sandbox_response.assert_async().await;
+        toolbox_response.assert_async().await;
+        upload.assert_async().await;
+        delete.assert_async().await;
+    }
+
     /// Recover the inner command a wrapper carries, proving it survives the
     /// base64 transport byte-for-byte.
     fn decode_wrapped_command(wrapped: &str) -> String {
@@ -3364,6 +3563,37 @@ mod tests {
         assert!(!wrapped.contains("base64"));
     }
 
+    #[cfg(unix)]
+    #[test]
+    #[expect(
+        clippy::disallowed_methods,
+        reason = "test executes the generated stdin redirection to verify exact bytes and EOF"
+    )]
+    fn stdin_file_redirection_applies_to_the_whole_command() {
+        let dir = tempfile::tempdir().expect("create stdin transport temp dir");
+        let stdin_path = dir.path().join("stdin data");
+        let injection_path = dir.path().join("must-not-run");
+        let stdin = format!(
+            "first line\n$(touch {})\nlast line",
+            injection_path.display()
+        );
+        std::fs::write(&stdin_path, &stdin).expect("write stdin fixture");
+        let command = redirect_command_stdin(
+            "IFS= read -r first\nprintf '%s\\n' \"$first\"\ncat",
+            stdin_path.to_str().expect("temp path should be UTF-8"),
+        );
+
+        let output = std::process::Command::new(REMOTE_BASH)
+            .args(["-c", &command])
+            .env_remove(BASH_ENV_VAR)
+            .output()
+            .expect("execute redirected command");
+
+        assert!(output.status.success(), "{output:?}");
+        assert_eq!(String::from_utf8_lossy(&output.stdout), stdin);
+        assert!(!injection_path.exists());
+    }
+
     #[cfg(unix)]
     #[test]
     #[expect(

lib/components/fabro-sandbox/src/docker.rs

@@ -31,13 +31,13 @@ use crate::redact::redact_auth_url;
 use crate::sandbox::{
     self, BASH_ENV_VAR, BASH_PROBE_SCRIPT, BASH_PROBE_TIMEOUT_MS, REMOTE_BASH,
     REMOTE_WALK_TIMEOUT_MS, RefreshOutcome, StdioProcessControl, optional_timeout, resolve_path,
-    validate_bash_probe,
+    validate_bash_probe, write_process_stdin,
 };
 use crate::{
     CommandOutputCallback, DEFAULT_EXEC_OUTPUT_TAIL_BYTES, DirEntry, ExecResult,
-    ExecStreamingResult, GrepOptions, Sandbox, SandboxEvent, SandboxEventCallback, SandboxFile,
-    StderrCollector, StdioProcess, StdioProcessHandle, StdioProcessTermination, WalkOptions,
-    clone_retry, format_lines_numbered, shell_quote,
+    ExecStreamingRequest, ExecStreamingResult, GrepOptions, Sandbox, SandboxEvent,
+    SandboxEventCallback, SandboxFile, StderrCollector, StdioProcess, StdioProcessHandle,
+    StdioProcessTermination, WalkOptions, clone_retry, format_lines_numbered, shell_quote,
 };
 
 const DOCKER_BASH_REQUIREMENT: &str = "Docker sandboxes require /bin/bash for every command, with no `sh` fallback; use an \
@@ -402,12 +402,15 @@ impl DockerSandbox {
         cmd: Vec<String>,
         working_dir: Option<String>,
         env: Option<Vec<String>>,
+        stdin: Option<Vec<u8>>,
         output_callback: Option<CommandOutputCallback>,
     ) -> crate::Result<(Vec<u8>, Vec<u8>, i32)> {
         let exec_opts = CreateExecOptions {
             cmd: Some(cmd),
+            attach_stdin: Some(stdin.is_some()),
             attach_stdout: Some(true),
             attach_stderr: Some(true),
+            tty: Some(false),
             working_dir,
             env: env.map(|e| e.into_iter().collect()),
             ..Default::default()
@@ -417,7 +420,11 @@ impl DockerSandbox {
             &docker,
             &container_id,
             exec_opts,
-            None,
+            Some(StartExecOptions {
+                detach:          false,
+                tty:             false,
+                output_capacity: None,
+            }),
             "Failed to create exec",
             "Failed to start exec",
         )
@@ -426,7 +433,18 @@ impl DockerSandbox {
         let mut stdout = Vec::new();
         let mut stderr = Vec::new();
 
-        if let StartExecResults::Attached { mut output, .. } = start_result {
+        let StartExecResults::Attached { mut output, input } = start_result else {
+            return Err(crate::Error::message(
+                "Docker started streaming command without attached standard I/O",
+            ));
+        };
+        let write_stdin = async move {
+            if let Some(stdin) = stdin {
+                write_process_stdin(input, &stdin).await?;
+            }
+            crate::Result::Ok(())
+        };
+        let read_output = async {
             while let Some(chunk) = output.next().await {
                 match chunk {
                     Ok(LogOutput::StdOut { message }) => {
@@ -447,7 +465,9 @@ impl DockerSandbox {
                     }
                 }
             }
-        }
+            crate::Result::Ok(())
+        };
+        tokio::try_join!(write_stdin, read_output)?;
 
         let inspect = docker
             .inspect_exec(&exec_id)
@@ -525,6 +545,7 @@ impl DockerSandbox {
         working_dir: Option<&str>,
         env_vars: Option<&HashMap<String, String>>,
         cancel_token: Option<CancellationToken>,
+        stdin: Option<Vec<u8>>,
         output_callback: Option<CommandOutputCallback>,
     ) -> crate::Result<ExecStreamingResult> {
         let start = Instant::now();
@@ -551,6 +572,7 @@ impl DockerSandbox {
             cmd,
             Some(effective_dir.clone()),
             env,
+            stdin,
             output_callback,
         ));
 
@@ -796,6 +818,7 @@ impl DockerSandbox {
                             None,
                             None,
                             None,
+                            None,
                         )
                         .await
                         .map_err(|error| DockerCloneFailure {
@@ -1781,13 +1804,17 @@ impl Sandbox for DockerSandbox {
 
     async fn exec_command_streaming(
         &self,
-        command: &str,
-        timeout_ms: Option<u64>,
-        working_dir: Option<&str>,
-        env_vars: Option<&HashMap<String, String>>,
-        cancel_token: Option<CancellationToken>,
-        output_callback: Option<CommandOutputCallback>,
+        request: ExecStreamingRequest<'_>,
     ) -> crate::Result<ExecStreamingResult> {
+        let ExecStreamingRequest {
+            command,
+            timeout_ms,
+            working_dir,
+            env_vars,
+            cancel_token,
+            stdin,
+            output_callback,
+        } = request;
         let dir = working_dir.map(|path| self.resolve_container_path(path));
         self.docker_exec_shell_streaming(
             command,
@@ -1795,6 +1822,7 @@ impl Sandbox for DockerSandbox {
             dir.as_deref(),
             env_vars,
             cancel_token,
+            stdin,
             output_callback,
         )
         .await

lib/components/fabro-sandbox/src/lib.rs

@@ -52,9 +52,9 @@ pub use provider::{
 pub use reconnect::{reconnect, reconnect_for_run, reconnect_for_run_with_callback};
 pub use sandbox::{
     CommandOutputCallback, DEFAULT_EXEC_OUTPUT_TAIL_BYTES, DirEntry, ExecResult,
-    ExecStreamingResult, GitRunInfo, GitSetupIntent, GrepOptions, RefreshOutcome, Sandbox,
-    SandboxEvent, SandboxEventCallback, SandboxFile, StderrCollector, StdioProcess,
-    StdioProcessHandle, StdioProcessTermination, WalkOptions, format_lines_numbered,
+    ExecStreamingRequest, ExecStreamingResult, GitRunInfo, GitSetupIntent, GrepOptions,
+    RefreshOutcome, Sandbox, SandboxEvent, SandboxEventCallback, SandboxFile, StderrCollector,
+    StdioProcess, StdioProcessHandle, StdioProcessTermination, WalkOptions, format_lines_numbered,
     git_push_via_exec, redacted_output_tail, setup_git_via_exec, shell_quote,
 };
 pub use sandbox_spec::SandboxSpec;

lib/components/fabro-sandbox/src/local.rs

@@ -14,12 +14,13 @@ use tokio_util::sync::CancellationToken;
 
 use crate::sandbox::{
     BASH_ENV_VAR, BASH_PROBE_SCRIPT, BASH_PROBE_TIMEOUT_MS, StdioProcessControl, optional_timeout,
-    validate_bash_probe,
+    validate_bash_probe, write_process_stdin,
 };
 use crate::{
     CommandOutputCallback, DEFAULT_EXEC_OUTPUT_TAIL_BYTES, DirEntry, ExecResult,
-    ExecStreamingResult, GrepOptions, Sandbox, SandboxEvent, SandboxEventCallback, SandboxFile,
-    StderrCollector, StdioProcess, StdioProcessHandle, StdioProcessTermination, WalkOptions,
+    ExecStreamingRequest, ExecStreamingResult, GrepOptions, Sandbox, SandboxEvent,
+    SandboxEventCallback, SandboxFile, StderrCollector, StdioProcess, StdioProcessHandle,
+    StdioProcessTermination, WalkOptions,
 };
 
 /// Remediation shown when the worker has no usable Bash.
@@ -514,13 +515,17 @@ impl Sandbox for LocalSandbox {
 
     async fn exec_command_streaming(
         &self,
-        command: &str,
-        timeout_ms: Option<u64>,
-        working_dir: Option<&str>,
-        env_vars: Option<&std::collections::HashMap<String, String>>,
-        cancel_token: Option<CancellationToken>,
-        output_callback: Option<CommandOutputCallback>,
+        request: ExecStreamingRequest<'_>,
     ) -> crate::Result<ExecStreamingResult> {
+        let ExecStreamingRequest {
+            command,
+            timeout_ms,
+            working_dir,
+            env_vars,
+            cancel_token,
+            stdin,
+            output_callback,
+        } = request;
         let start = Instant::now();
 
         let filtered_env = filtered_env_vars(env_vars, ExplicitEnvPolicy::FilterSensitive);
@@ -537,6 +542,9 @@ impl Sandbox for LocalSandbox {
             .kill_on_drop(true)
             .stdout(std::process::Stdio::piped())
             .stderr(std::process::Stdio::piped());
+        if stdin.is_some() {
+            cmd.stdin(std::process::Stdio::piped());
+        }
 
         #[cfg(unix)]
         fabro_proc::pre_exec_setpgid(cmd.as_std_mut());
@@ -551,6 +559,17 @@ impl Sandbox for LocalSandbox {
 
         let stdout_pipe = child.stdout.take();
         let stderr_pipe = child.stderr.take();
+        let stdin_task = match stdin {
+            Some(stdin) => {
+                let stdin_pipe = child.stdin.take().ok_or_else(|| {
+                    crate::Error::message("Failed to open command standard input")
+                })?;
+                Some(tokio::spawn(async move {
+                    write_process_stdin(stdin_pipe, &stdin).await
+                }))
+            }
+            None => None,
+        };
         let stdout_callback = output_callback.clone();
         let stderr_callback = output_callback;
         let stdout_task = tokio::spawn(async move {
@@ -577,6 +596,11 @@ impl Sandbox for LocalSandbox {
         };
 
         let duration_ms = elapsed_ms(start);
+        if let Some(stdin_task) = stdin_task {
+            stdin_task
+                .await
+                .map_err(|e| crate::Error::context("stdin stream task failed", e))??;
+        }
         let stdout_bytes = stdout_task
             .await
             .map_err(|e| crate::Error::context("stdout stream task failed", e))??;
@@ -1344,12 +1368,9 @@ mod tests {
 
         let result = sandbox
             .exec_command_streaming(
-                BASH_ONLY_COMMAND,
-                Some(5000),
-                None,
-                None,
-                None,
-                Some(Arc::new(|_, _| Box::pin(async { Ok(()) }))),
+                ExecStreamingRequest::new(BASH_ONLY_COMMAND)
+                    .timeout_ms(Some(5000))
+                    .output_callback(Some(Arc::new(|_, _| Box::pin(async { Ok(()) })))),
             )
             .await
             .unwrap();
@@ -1364,6 +1385,27 @@ mod tests {
         std::fs::remove_dir_all(&dir).unwrap();
     }
 
+    #[tokio::test]
+    async fn exec_command_streaming_writes_exact_stdin_and_closes_it() {
+        let dir = temp_dir();
+        let sandbox = LocalSandbox::new(dir.clone());
+        let stdin = b"first line\n$(touch must-not-run)\nlast line".to_vec();
+
+        let result = sandbox
+            .exec_command_streaming(
+                ExecStreamingRequest::new("cat")
+                    .timeout_ms(Some(5000))
+                    .stdin(Some(stdin.clone())),
+            )
+            .await
+            .unwrap();
+
+        assert_eq!(result.result.exit_code, Some(0));
+        assert_eq!(result.result.stdout.as_bytes(), stdin);
+        assert!(!dir.join("must-not-run").exists());
+        std::fs::remove_dir_all(&dir).unwrap();
+    }
+
     #[tokio::test]
     async fn stdio_process_evaluates_bash_before_exec() {
         let dir = temp_dir();
@@ -1415,12 +1457,10 @@ mod tests {
 
         let streaming = sandbox
             .exec_command_streaming(
-                LOGIN_SHELL_REPORT,
-                Some(5000),
-                None,
-                Some(&env_vars),
-                None,
-                Some(Arc::new(|_, _| Box::pin(async { Ok(()) }))),
+                ExecStreamingRequest::new(LOGIN_SHELL_REPORT)
+                    .timeout_ms(Some(5000))
+                    .env_vars(Some(&env_vars))
+                    .output_callback(Some(Arc::new(|_, _| Box::pin(async { Ok(()) })))),
             )
             .await
             .unwrap();

lib/components/fabro-sandbox/src/sandbox.rs

@@ -11,7 +11,7 @@ use fabro_types::{CommandOutputStream, CommandTermination};
 use fabro_util::shell;
 use fabro_util::workspace_glob::WorkspaceGlob;
 use serde::{Deserialize, Serialize};
-use tokio::io::{AsyncRead, AsyncReadExt, AsyncWrite};
+use tokio::io::{AsyncRead, AsyncReadExt, AsyncWrite, AsyncWriteExt};
 use tokio::sync::Mutex as TokioMutex;
 use tokio::task::JoinHandle;
 use tokio::time;
@@ -184,23 +184,9 @@ macro_rules! delegate_sandbox {
 
             async fn exec_command_streaming(
                 &self,
-                command: &str,
-                timeout_ms: Option<u64>,
-                working_dir: Option<&str>,
-                env_vars: Option<&std::collections::HashMap<String, String>>,
-                cancel_token: Option<tokio_util::sync::CancellationToken>,
-                output_callback: Option<$crate::CommandOutputCallback>,
+                request: $crate::ExecStreamingRequest<'_>,
             ) -> $crate::Result<$crate::ExecStreamingResult> {
-                self.$field
-                    .exec_command_streaming(
-                        command,
-                        timeout_ms,
-                        working_dir,
-                        env_vars,
-                        cancel_token,
-                        output_callback,
-                    )
-                    .await
+                self.$field.exec_command_streaming(request).await
             }
 
             async fn spawn_stdio_process(
@@ -773,6 +759,96 @@ pub type CommandOutputCallback = Arc<
         + Sync,
 >;
 
+/// Inputs for a streaming command execution.
+///
+/// Standard input is owned so providers can move it into a writer task. This
+/// type does not implement `Debug` because standard input can contain
+/// sensitive workflow data.
+#[non_exhaustive]
+pub struct ExecStreamingRequest<'a> {
+    pub command:         &'a str,
+    pub timeout_ms:      Option<u64>,
+    pub working_dir:     Option<&'a str>,
+    pub env_vars:        Option<&'a HashMap<String, String>>,
+    pub cancel_token:    Option<CancellationToken>,
+    pub stdin:           Option<Vec<u8>>,
+    pub output_callback: Option<CommandOutputCallback>,
+}
+
+impl<'a> ExecStreamingRequest<'a> {
+    #[must_use]
+    pub fn new(command: &'a str) -> Self {
+        Self {
+            command,
+            timeout_ms: None,
+            working_dir: None,
+            env_vars: None,
+            cancel_token: None,
+            stdin: None,
+            output_callback: None,
+        }
+    }
+
+    #[must_use]
+    pub fn timeout_ms(mut self, timeout_ms: Option<u64>) -> Self {
+        self.timeout_ms = timeout_ms;
+        self
+    }
+
+    #[must_use]
+    pub fn working_dir(mut self, working_dir: Option<&'a str>) -> Self {
+        self.working_dir = working_dir;
+        self
+    }
+
+    #[must_use]
+    pub fn env_vars(mut self, env_vars: Option<&'a HashMap<String, String>>) -> Self {
+        self.env_vars = env_vars;
+        self
+    }
+
+    #[must_use]
+    pub fn cancel_token(mut self, cancel_token: Option<CancellationToken>) -> Self {
+        self.cancel_token = cancel_token;
+        self
+    }
+
+    #[must_use]
+    pub fn stdin(mut self, stdin: Option<Vec<u8>>) -> Self {
+        self.stdin = stdin;
+        self
+    }
+
+    #[must_use]
+    pub fn output_callback(mut self, output_callback: Option<CommandOutputCallback>) -> Self {
+        self.output_callback = output_callback;
+        self
+    }
+}
+
+pub(crate) async fn write_process_stdin<W>(mut writer: W, stdin: &[u8]) -> crate::Result<()>
+where
+    W: AsyncWrite + Unpin,
+{
+    if let Err(err) = writer.write_all(stdin).await {
+        if err.kind() != std::io::ErrorKind::BrokenPipe {
+            return Err(crate::Error::context(
+                "Failed to write command standard input",
+                err,
+            ));
+        }
+    }
+    if let Err(err) = writer.shutdown().await {
+        if err.kind() != std::io::ErrorKind::BrokenPipe {
+            return Err(crate::Error::context(
+                "Failed to close command standard input",
+                err,
+            ));
+        }
+    }
+    Ok(())
+}
+
 pub(crate) async fn replay_exec_result(
     result: ExecResult,
     streams_separated: bool,
@@ -1028,6 +1104,10 @@ pub trait Sandbox: Send + Sync {
     /// interpreter or shell options, so Bash-only syntax behaves identically
     /// through both.
     ///
+    /// When `request.stdin` is set, providers must write those exact bytes to
+    /// the process's standard input and then close it to deliver EOF. The bytes
+    /// must remain separate from command source and diagnostics.
+    ///
     /// **Production sandboxes must override this.** The default falls back to
     /// the non-streaming [`exec_command`](Self::exec_command) and replays its
     /// output through `output_callback` at the end when one is supplied,
@@ -1036,27 +1116,28 @@ pub trait Sandbox: Send + Sync {
     /// behavior for test mocks but silently drops live output for any real
     /// sandbox that wraps another — decorators in particular must forward to
     /// the inner sandbox's streaming implementation rather than relying on
-    /// this default.
+    /// this default. The fallback rejects `request.stdin` because
+    /// [`exec_command`](Self::exec_command) has no stdin channel.
     async fn exec_command_streaming(
         &self,
-        command: &str,
-        timeout_ms: Option<u64>,
-        working_dir: Option<&str>,
-        env_vars: Option<&std::collections::HashMap<String, String>>,
-        cancel_token: Option<CancellationToken>,
-        output_callback: Option<CommandOutputCallback>,
+        request: ExecStreamingRequest<'_>,
     ) -> crate::Result<ExecStreamingResult> {
-        let fallback_timeout_ms = timeout_ms.unwrap_or(u64::MAX);
+        if request.stdin.is_some() {
+            return Err(crate::Error::message(
+                "This sandbox does not support standard input for streaming commands",
+            ));
+        }
+        let fallback_timeout_ms = request.timeout_ms.unwrap_or(u64::MAX);
         let result = self
             .exec_command(
-                command,
+                request.command,
                 fallback_timeout_ms,
-                working_dir,
-                env_vars,
-                cancel_token,
+                request.working_dir,
+                request.env_vars,
+                request.cancel_token,
             )
             .await?;
-        replay_exec_result(result, true, output_callback.as_ref()).await
+        replay_exec_result(result, true, request.output_callback.as_ref()).await
     }
 
     /// Launch a long-lived process with bidirectional stdio attached.
@@ -1648,13 +1729,13 @@ mod tests {
     }
 
     #[test]
-    fn sandbox_tracing_events_do_not_log_raw_command_fields() {
+    fn sandbox_tracing_events_do_not_log_raw_command_or_stdin_fields() {
         let root = std::path::Path::new(env!("CARGO_MANIFEST_DIR")).join("src");
         let mut failures = Vec::new();
         scan_for_command_tracing(&root, &mut failures);
         assert!(
             failures.is_empty(),
-            "raw command/cmd tracing fields found:\n{}",
+            "raw command/cmd/stdin tracing fields found:\n{}",
             failures.join("\n")
         );
     }
@@ -1920,6 +2001,8 @@ mod tests {
                         || call.contains("command =")
                         || call.contains("cmd,")
                         || call.contains("cmd =")
+                        || call.contains("stdin,")
+                        || call.contains("stdin =")
                     {
                         failures.push(format!(
                             "{}: {}",

lib/components/fabro-sandbox/src/test_support.rs

@@ -12,9 +12,9 @@ use tokio_util::sync::CancellationToken;
 
 use crate::sandbox::{self, StdioProcessControl};
 use crate::{
-    DEFAULT_EXEC_OUTPUT_TAIL_BYTES, DirEntry, ExecResult, GrepOptions, Sandbox, SandboxEvent,
-    SandboxEventCallback, SandboxFile, StderrCollector, StdioProcess, StdioProcessHandle,
-    StdioProcessTermination, WalkOptions,
+    DEFAULT_EXEC_OUTPUT_TAIL_BYTES, DirEntry, ExecResult, ExecStreamingRequest, GrepOptions,
+    Sandbox, SandboxEvent, SandboxEventCallback, SandboxFile, StderrCollector, StdioProcess,
+    StdioProcessHandle, StdioProcessTermination, WalkOptions,
 };
 
 // --- MockSandbox ---
@@ -39,6 +39,8 @@ pub struct MockSandbox {
     pub captured_working_dirs: Mutex<Vec<Option<String>>>,
     /// Captures the `env_vars` argument from `exec_command` calls.
     pub captured_env_vars:     Mutex<Option<HashMap<String, String>>>,
+    /// Captures the bytes passed to a streaming command's standard input.
+    pub captured_stdin:        Mutex<Option<Vec<u8>>>,
     pub active:                AtomicBool,
     pub activate_error:        Option<String>,
     pub activate_calls:        Mutex<u32>,
@@ -159,6 +161,7 @@ impl Default for MockSandbox {
             captured_commands:     Mutex::new(Vec::new()),
             captured_working_dirs: Mutex::new(Vec::new()),
             captured_env_vars:     Mutex::new(None),
+            captured_stdin:        Mutex::new(None),
             active:                AtomicBool::new(true),
             activate_error:        None,
             activate_calls:        Mutex::new(0),
@@ -299,13 +302,21 @@ impl Sandbox for MockSandbox {
 
     async fn exec_command_streaming(
         &self,
-        command: &str,
-        timeout_ms: Option<u64>,
-        working_dir: Option<&str>,
-        env_vars: Option<&std::collections::HashMap<String, String>>,
-        cancel_token: Option<CancellationToken>,
-        output_callback: Option<crate::CommandOutputCallback>,
+        request: ExecStreamingRequest<'_>,
     ) -> crate::Result<crate::ExecStreamingResult> {
+        let ExecStreamingRequest {
+            command,
+            timeout_ms,
+            working_dir,
+            env_vars,
+            cancel_token,
+            stdin,
+            output_callback,
+        } = request;
+        *self
+            .captured_stdin
+            .lock()
+            .expect("captured_stdin lock poisoned") = stdin;
         let result = self
             .exec_command(
                 command,

lib/components/fabro-sandbox/tests/daytona_streaming_live.rs

@@ -110,6 +110,40 @@ mod daytona_streaming_live {
                 "exec_command_streaming should report the Bash-only result",
             )?;
 
+            let stdin = "first line\n$(touch /tmp/must-not-run)\nlast line";
+            let (stdin_result, _) = run_captured_with_stdin(
+                &sandbox,
+                "cat",
+                30_000,
+                None,
+                Some(stdin.as_bytes().to_vec()),
+            )
+            .await?;
+            ensure_eq(
+                &stdin_result.result.exit_code,
+                &Some(0),
+                "exec_command_streaming should close stdin with a successful EOF",
+            )?;
+            ensure_eq(
+                &stdin_result.result.stdout,
+                &stdin.to_string(),
+                "exec_command_streaming should preserve exact stdin bytes",
+            )?;
+            let stdin_cleanup = sandbox
+                .exec_command(
+                    "test ! -e /tmp/must-not-run && \
+                     ! compgen -G '/tmp/fabro-command-stdin-*' >/dev/null",
+                    30_000,
+                    None,
+                    None,
+                    None,
+                )
+                .await?;
+            ensure!(
+                stdin_cleanup.is_success(),
+                "Daytona stdin data must stay inert and its temporary file must be deleted: {stdin_cleanup:?}"
+            );
+
             Ok(())
         }
         .await;
@@ -330,12 +364,12 @@ mod daytona_streaming_live {
         let live_exec = tokio::spawn(async move {
             sandbox_for_exec
                 .exec_command_streaming(
-                    "printf 'live-out\\n'; printf 'live-err\\n' >&2; sleep 30",
-                    Some(60_000),
-                    None,
-                    None,
-                    Some(cancel_for_exec),
-                    Some(callback),
+                    fabro_sandbox::ExecStreamingRequest::new(
+                        "printf 'live-out\\n'; printf 'live-err\\n' >&2; sleep 30",
+                    )
+                    .timeout_ms(Some(60_000))
+                    .cancel_token(Some(cancel_for_exec))
+                    .output_callback(Some(callback)),
                 )
                 .await
         });
@@ -450,17 +484,26 @@ mod daytona_streaming_live {
         command: &str,
         timeout_ms: u64,
         cancel_token: Option<CancellationToken>,
+    ) -> Result<(ExecStreamingResult, Vec<CapturedChunk>)> {
+        run_captured_with_stdin(sandbox, command, timeout_ms, cancel_token, None).await
+    }
+
+    async fn run_captured_with_stdin(
+        sandbox: &DaytonaSandbox,
+        command: &str,
+        timeout_ms: u64,
+        cancel_token: Option<CancellationToken>,
+        stdin: Option<Vec<u8>>,
     ) -> Result<(ExecStreamingResult, Vec<CapturedChunk>)> {
         let chunks = Arc::new(Mutex::new(Vec::new()));
         let callback = capture_callback(Arc::clone(&chunks));
         let result = sandbox
             .exec_command_streaming(
-                command,
-                Some(timeout_ms),
-                None,
-                None,
-                cancel_token,
-                Some(callback),
+                fabro_sandbox::ExecStreamingRequest::new(command)
+                    .timeout_ms(Some(timeout_ms))
+                    .cancel_token(cancel_token)
+                    .stdin(stdin)
+                    .output_callback(Some(callback)),
             )
             .await?;
         let chunks = chunks.lock().await.clone();

lib/components/fabro-sandbox/tests/docker_streaming.rs

@@ -3,7 +3,9 @@
 use std::sync::Arc;
 
 use bollard::Docker;
-use fabro_sandbox::{CommandOutputCallback, DockerSandbox, DockerSandboxOptions, Sandbox};
+use fabro_sandbox::{
+    CommandOutputCallback, DockerSandbox, DockerSandboxOptions, ExecStreamingRequest, Sandbox,
+};
 use tokio::sync::Mutex;
 
 fn capture_bytes(chunks: Arc<Mutex<Vec<u8>>>) -> CommandOutputCallback {
@@ -48,14 +50,12 @@ async fn streaming_timeout_terminates_docker_exec_before_returning() {
     let chunks = Arc::new(Mutex::new(Vec::new()));
 
     let marker = "fabro_streaming_timeout_sentinel";
+    let command = format!("trap '' HUP TERM; echo start; sleep 5 # {marker}");
     let result = sandbox
         .exec_command_streaming(
-            &format!("trap '' HUP TERM; echo start; sleep 5 # {marker}"),
-            Some(200),
-            None,
-            None,
-            None,
-            Some(capture_bytes(Arc::clone(&chunks))),
+            ExecStreamingRequest::new(&command)
+                .timeout_ms(Some(200))
+                .output_callback(Some(capture_bytes(Arc::clone(&chunks)))),
         )
         .await
         .expect("streaming command should return a timeout result");
@@ -90,6 +90,67 @@ async fn streaming_timeout_terminates_docker_exec_before_returning() {
     );
 }
 
+#[tokio::test]
+#[ignore = "requires real Docker container lifecycle; run explicitly when changing Docker exec integration"]
+async fn streaming_command_receives_exact_stdin_and_eof() {
+    let image = "buildpack-deps:noble";
+    let Ok(docker) = Docker::connect_with_local_defaults() else {
+        return;
+    };
+    if docker.inspect_image(image).await.is_err() {
+        return;
+    }
+
+    let sandbox = DockerSandbox::new(
+        DockerSandboxOptions {
+            image: image.to_string(),
+            auto_pull: false,
+            skip_clone: true,
+            ..DockerSandboxOptions::default()
+        },
+        None,
+        None,
+        None,
+        None,
+    )
+    .expect("docker sandbox should construct");
+    sandbox
+        .initialize()
+        .await
+        .expect("docker sandbox should initialize");
+
+    let stdin = b"first line\n$(touch /tmp/must-not-run)\nlast line".to_vec();
+    let result = sandbox
+        .exec_command_streaming(
+            ExecStreamingRequest::new("cat")
+                .timeout_ms(Some(10_000))
+                .stdin(Some(stdin.clone())),
+        )
+        .await
+        .expect("streaming command should read stdin and finish at EOF");
+    let injection_probe = sandbox
+        .exec_command("test ! -e /tmp/must-not-run", 10_000, None, None, None)
+        .await
+        .expect("injection probe should run");
+
+    sandbox
+        .cleanup()
+        .await
+        .expect("docker cleanup should succeed");
+
+    assert!(
+        result.result.is_success(),
+        "stdin command failed: stdout={} stderr={}",
+        result.result.stdout,
+        result.result.stderr
+    );
+    assert_eq!(result.result.stdout.as_bytes(), stdin);
+    assert!(
+        injection_probe.is_success(),
+        "stdin bytes must not be evaluated as shell source"
+    );
+}
+
 #[tokio::test]
 #[ignore = "requires real Docker container lifecycle, image, network, and a public GitHub clone"]
 async fn cloned_docker_sandbox_uses_repos_checkout_and_workspace_symlink() {
@@ -213,12 +274,9 @@ async fn docker_runs_clean_bash_through_both_command_paths() {
     let chunks = Arc::new(Mutex::new(Vec::new()));
     let streaming = sandbox
         .exec_command_streaming(
-            command,
-            Some(10_000),
-            None,
-            None,
-            None,
-            Some(capture_bytes(Arc::clone(&chunks))),
+            ExecStreamingRequest::new(command)
+                .timeout_ms(Some(10_000))
+                .output_callback(Some(capture_bytes(Arc::clone(&chunks)))),
         )
         .await
         .expect("streaming command should run");

lib/components/fabro-validate/src/rules/inert_attribute.rs

@@ -17,6 +17,7 @@ pub(super) fn rule() -> Box<dyn LintRule> {
 const HANDLER_SPECIFIC_ATTRS: &[(&str, &[&str])] = &[
     ("script", &["command"]),
     ("language", &["command"]),
+    ("stdin_source", &["command", "tool"]),
     ("duration", &["wait"]),
     ("max_parallel", &["parallel"]),
     ("output_retries", &["agent", "prompt"]),
@@ -166,10 +167,12 @@ mod tests {
     #[test]
     fn accepts_attrs_on_their_own_handler_types() {
         let mut g = minimal_graph();
-        g.nodes.insert(
-            "run".to_string(),
-            node_with_attr("run", "parallelogram", "script", "echo hi"),
+        let mut run = node_with_attr("run", "parallelogram", "script", "echo hi");
+        run.attrs.insert(
+            "stdin_source".to_string(),
+            AttrValue::String("context.parallel.results".to_string()),
         );
+        g.nodes.insert("run".to_string(), run);
         g.nodes.insert(
             "audit".to_string(),
             node_with_attr("audit", "parallelogram", "output_schema", "routing"),

lib/components/fabro-validate/src/rules/mod.rs

@@ -25,6 +25,7 @@ mod script_absolute_cd;
 mod selection_valid;
 mod start_no_incoming;
 mod start_node;
+mod stdin_source_valid;
 mod stylesheet_model_known;
 mod stylesheet_syntax;
 mod terminal_node;
@@ -56,6 +57,7 @@ pub fn built_in_rules() -> Vec<Box<dyn LintRule>> {
         prompt_on_llm_nodes::rule(),
         freeform_edge_count::rule(),
         for_each_contract::rule(),
+        stdin_source_valid::rule(),
         direction_valid::rule(),
         reserved_keyword_node_id::rule(),
         all_conditional_edges::rule(),

lib/components/fabro-validate/src/rules/stdin_source_valid.rs

@@ -0,0 +1,91 @@
+use fabro_graphviz::graph::Graph;
+
+use crate::{Diagnostic, LintRule, Severity};
+
+pub(super) fn rule() -> Box<dyn LintRule> {
+    Box::new(Rule)
+}
+
+struct Rule;
+
+impl LintRule for Rule {
+    fn name(&self) -> &'static str {
+        "stdin_source_valid"
+    }
+
+    fn apply(&self, graph: &Graph) -> Vec<Diagnostic> {
+        graph
+            .nodes
+            .values()
+            .filter(|node| matches!(node.handler_type(), Some("command" | "tool")))
+            .filter(|node| node.attrs.contains_key("stdin_source"))
+            .filter(|node| {
+                node.stdin_source()
+                    .is_none_or(|source| source.trim().is_empty())
+            })
+            .map(|node| Diagnostic {
+                rule: self.name().to_string(),
+                severity: Severity::Error,
+                message: format!(
+                    "Command node '{}' has an empty or non-string 'stdin_source'",
+                    node.id
+                ),
+                node_id: Some(node.id.clone()),
+                edge: None,
+                fix: Some(
+                    "Set 'stdin_source' to a context key such as \"context.parallel.results\""
+                        .to_string(),
+                ),
+                ..Diagnostic::default()
+            })
+            .collect()
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use fabro_graphviz::graph::{AttrValue, Node};
+
+    use super::Rule;
+    use crate::rules::test_support::minimal_graph;
+    use crate::{LintRule, Severity};
+
+    fn command_node(value: AttrValue) -> Node {
+        let mut node = Node::new("merge");
+        node.attrs.insert(
+            "shape".to_string(),
+            AttrValue::String("parallelogram".to_string()),
+        );
+        node.attrs.insert("stdin_source".to_string(), value);
+        node
+    }
+
+    #[test]
+    fn accepts_non_empty_context_source() {
+        let mut graph = minimal_graph();
+        graph.nodes.insert(
+            "merge".to_string(),
+            command_node(AttrValue::String("context.parallel.results".to_string())),
+        );
+
+        assert!(Rule.apply(&graph).is_empty());
+    }
+
+    #[test]
+    fn rejects_empty_or_non_string_source() {
+        for value in [
+            AttrValue::String(String::new()),
+            AttrValue::String("   ".to_string()),
+            AttrValue::Integer(3),
+        ] {
+            let mut graph = minimal_graph();
+            graph.nodes.insert("merge".to_string(), command_node(value));
+
+            let diagnostics = Rule.apply(&graph);
+
+            assert_eq!(diagnostics.len(), 1);
+            assert_eq!(diagnostics[0].severity, Severity::Error);
+            assert!(diagnostics[0].message.contains("empty or non-string"));
+        }
+    }
+}

lib/components/fabro-workflow/src/handler/command.rs

@@ -1,15 +1,16 @@
 use std::path::Path;
 
 use async_trait::async_trait;
-use fabro_agent::CommandOutputCallback;
+use fabro_agent::{CommandOutputCallback, ExecStreamingRequest};
 use fabro_graphviz::graph::{Graph, Node};
 use fabro_types::{CommandTermination, StageTiming};
 use fabro_util::shell::shell_quote;
 
 use super::structured_output::{self, StructuredOutputError};
 use super::{EngineServices, Handler, NodeTimeoutPolicy};
+use crate::artifact;
 use crate::command_log::CommandLogRecorder;
-use crate::context::{Context, keys};
+use crate::context::{self, Context, keys};
 use crate::error::Error;
 use crate::event::{Event, StageScope};
 use crate::outcome::{Outcome, OutcomeExt};
@@ -31,6 +32,9 @@ impl Handler for CommandHandler {
         _run_dir: &Path,
         _services: &EngineServices,
     ) -> Result<Outcome, Error> {
+        if let Err(reason) = stdin_source(node) {
+            return Ok(Outcome::fail_deterministic(reason));
+        }
         let script = node
             .attrs
             .get("script")
@@ -77,6 +81,10 @@ impl Handler for CommandHandler {
             )));
         }
 
+        let stdin = match resolve_stdin(node, context, services).await {
+            Ok(stdin) => stdin,
+            Err(outcome) => return Ok(outcome),
+        };
         let output_schema = structured_output::parse_node_output_schema(node)?;
 
         let command = if language == "python" {
@@ -123,12 +131,12 @@ impl Handler for CommandHandler {
             .run
             .sandbox
             .exec_command_streaming(
-                &command,
-                Some(timeout_ms),
-                None,
-                env_vars,
-                Some(cancel_token.clone()),
-                Some(output_callback),
+                ExecStreamingRequest::new(&command)
+                    .timeout_ms(Some(timeout_ms))
+                    .env_vars(env_vars)
+                    .cancel_token(Some(cancel_token.clone()))
+                    .stdin(stdin)
+                    .output_callback(Some(output_callback)),
             )
             .await;
         cancel_token.cancel();
@@ -215,6 +223,56 @@ impl Handler for CommandHandler {
     }
 }
 
+fn stdin_source(node: &Node) -> std::result::Result<Option<&str>, String> {
+    if !node.attrs.contains_key("stdin_source") {
+        return Ok(None);
+    }
+    node.stdin_source()
+        .filter(|source| !source.trim().is_empty())
+        .map(Some)
+        .ok_or_else(|| {
+            format!(
+                "Command node '{}' requires 'stdin_source' to be a non-empty string",
+                node.id
+            )
+        })
+}
+
+async fn resolve_stdin(
+    node: &Node,
+    context: &Context,
+    services: &EngineServices,
+) -> std::result::Result<Option<Vec<u8>>, Outcome> {
+    let Some(source) = stdin_source(node).map_err(Outcome::fail_deterministic)? else {
+        return Ok(None);
+    };
+    let Some(value) = context::lookup_flat(context, source) else {
+        return Err(Outcome::fail_deterministic(format!(
+            "stdin_source '{source}' was not found in workflow context"
+        )));
+    };
+    let value = artifact::resolve_json_value(&value, &services.run.run_store)
+        .await
+        .map_err(|err| {
+            Outcome::fail_deterministic(format!(
+                "stdin_source '{source}' could not be resolved: {err}"
+            ))
+        })?;
+    let stdin = encode_stdin_value(value).map_err(|err| {
+        Outcome::fail_deterministic(format!(
+            "stdin_source '{source}' could not be serialized: {err}"
+        ))
+    })?;
+    Ok(Some(stdin))
+}
+
+fn encode_stdin_value(value: serde_json::Value) -> serde_json::Result<Vec<u8>> {
+    match value {
+        serde_json::Value::String(text) => Ok(text.into_bytes()),
+        value => serde_json::to_vec(&value),
+    }
+}
+
 fn schema_validation_failure_reason(
     script: &str,
     error: &StructuredOutputError,
@@ -268,6 +326,23 @@ mod tests {
     const PASSED_OUTPUT_SCHEMA: &str =
         r#"{"type":"object","required":["passed"],"properties":{"passed":{"type":"boolean"}}}"#;
 
+    #[test]
+    fn stdin_json_encoding_is_compact_and_strings_are_raw() {
+        for (value, expected) in [
+            (serde_json::json!("text"), b"text".as_slice()),
+            (serde_json::json!([1, 2]), br"[1,2]".as_slice()),
+            (
+                serde_json::json!({"ok": true}),
+                br#"{"ok":true}"#.as_slice(),
+            ),
+            (serde_json::json!(42), b"42".as_slice()),
+            (serde_json::json!(false), b"false".as_slice()),
+            (serde_json::Value::Null, b"null".as_slice()),
+        ] {
+            assert_eq!(encode_stdin_value(value).unwrap(), expected);
+        }
+    }
+
     #[derive(Default)]
     struct MemoryRunStoreBackend {
         blobs: Mutex<std::collections::HashMap<fabro_types::RunBlobId, Bytes>>,
@@ -1326,6 +1401,7 @@ mod tests {
         captured_command:      std::sync::Mutex<Option<String>>,
         captured_env_vars:     std::sync::Mutex<Option<std::collections::HashMap<String, String>>>,
         captured_cancel_token: std::sync::Mutex<Option<bool>>,
+        captured_stdin:        std::sync::Mutex<Option<Vec<u8>>>,
     }
 
     impl SpySandbox {
@@ -1336,6 +1412,7 @@ mod tests {
                 captured_command: std::sync::Mutex::new(None),
                 captured_env_vars: std::sync::Mutex::new(None),
                 captured_cancel_token: std::sync::Mutex::new(None),
+                captured_stdin: std::sync::Mutex::new(None),
             }
         }
 
@@ -1352,12 +1429,17 @@ mod tests {
                 captured_command:      std::sync::Mutex::new(None),
                 captured_env_vars:     std::sync::Mutex::new(None),
                 captured_cancel_token: std::sync::Mutex::new(None),
+                captured_stdin:        std::sync::Mutex::new(None),
             }
         }
 
         fn captured_command(&self) -> Option<String> {
             self.captured_command.lock().unwrap().clone()
         }
+
+        fn captured_stdin(&self) -> Option<Vec<u8>> {
+            self.captured_stdin.lock().unwrap().clone()
+        }
     }
 
     #[async_trait::async_trait]
@@ -1397,6 +1479,42 @@ mod tests {
             }
             Ok(self.exec_result.clone())
         }
+        async fn exec_command_streaming(
+            &self,
+            request: fabro_agent::sandbox::ExecStreamingRequest<'_>,
+        ) -> fabro_sandbox::Result<fabro_agent::sandbox::ExecStreamingResult> {
+            *self.captured_stdin.lock().unwrap() = request.stdin;
+            let result = self
+                .exec_command(
+                    request.command,
+                    request.timeout_ms.unwrap_or(u64::MAX),
+                    request.working_dir,
+                    request.env_vars,
+                    request.cancel_token,
+                )
+                .await?;
+            if let Some(callback) = request.output_callback.as_ref() {
+                if !result.stdout.is_empty() {
+                    callback(
+                        fabro_types::CommandOutputStream::Stdout,
+                        result.stdout.as_bytes().to_vec(),
+                    )
+                    .await?;
+                }
+                if !result.stderr.is_empty() {
+                    callback(
+                        fabro_types::CommandOutputStream::Stderr,
+                        result.stderr.as_bytes().to_vec(),
+                    )
+                    .await?;
+                }
+            }
+            Ok(fabro_agent::sandbox::ExecStreamingResult {
+                result,
+                streams_separated: true,
+                live_streaming: false,
+            })
+        }
         async fn grep(
             &self,
             _: &str,
@@ -1445,6 +1563,166 @@ mod tests {
         services
     }
 
+    #[tokio::test]
+    async fn stdin_source_serializes_parallel_results_as_compact_json() {
+        let spy = std::sync::Arc::new(SpySandbox::new(fabro_agent::sandbox::ExecResult {
+            stdout:      String::new(),
+            stderr:      String::new(),
+            exit_code:   Some(0),
+            termination: CommandTermination::Exited,
+            duration_ms: 5,
+        }));
+        let handler = CommandHandler;
+        let mut node = Node::new("merge");
+        node.attrs
+            .insert("script".to_string(), AttrValue::String("cat".to_string()));
+        node.attrs.insert(
+            "stdin_source".to_string(),
+            AttrValue::String("context.parallel.results".to_string()),
+        );
+        let parallel_results = serde_json::json!([
+            {
+                "branch": "one",
+                "response": "$(touch /tmp/must-not-run)\nsecond line"
+            },
+            {"branch": "two", "passed": true}
+        ]);
+        let context = Context::new();
+        context.set(keys::PARALLEL_RESULTS, parallel_results.clone());
+        let graph = Graph::new("test");
+        let run_dir = tempfile::tempdir().unwrap();
+        let services = make_spy_services(spy.clone());
+
+        let outcome = handler
+            .execute(&node, &context, &graph, run_dir.path(), &services)
+            .await
+            .unwrap();
+
+        assert_eq!(outcome.status, StageOutcome::Succeeded);
+        assert_eq!(
+            spy.captured_stdin(),
+            Some(serde_json::to_vec(&parallel_results).unwrap())
+        );
+        assert!(
+            !spy.captured_command()
+                .expect("command should run")
+                .contains("must-not-run"),
+            "stdin content must not be inserted into shell source"
+        );
+    }
+
+    #[tokio::test]
+    async fn stdin_source_passes_strings_without_adding_a_newline() {
+        let spy = std::sync::Arc::new(SpySandbox::new(fabro_agent::sandbox::ExecResult {
+            stdout:      String::new(),
+            stderr:      String::new(),
+            exit_code:   Some(0),
+            termination: CommandTermination::Exited,
+            duration_ms: 5,
+        }));
+        let handler = CommandHandler;
+        let mut node = Node::new("consume");
+        node.attrs
+            .insert("script".to_string(), AttrValue::String("cat".to_string()));
+        node.attrs.insert(
+            "stdin_source".to_string(),
+            AttrValue::String("context.input".to_string()),
+        );
+        let context = Context::new();
+        context.set("input", serde_json::json!("first\nlast"));
+        let graph = Graph::new("test");
+        let run_dir = tempfile::tempdir().unwrap();
+        let services = make_spy_services(spy.clone());
+
+        let outcome = handler
+            .execute(&node, &context, &graph, run_dir.path(), &services)
+            .await
+            .unwrap();
+
+        assert_eq!(outcome.status, StageOutcome::Succeeded);
+        assert_eq!(
+            spy.captured_stdin().as_deref(),
+            Some(b"first\nlast".as_slice())
+        );
+    }
+
+    #[tokio::test]
+    async fn missing_stdin_source_fails_before_starting_the_command() {
+        let spy = std::sync::Arc::new(SpySandbox::new(fabro_agent::sandbox::ExecResult {
+            stdout:      String::new(),
+            stderr:      String::new(),
+            exit_code:   Some(0),
+            termination: CommandTermination::Exited,
+            duration_ms: 5,
+        }));
+        let handler = CommandHandler;
+        let mut node = Node::new("consume");
+        node.attrs
+            .insert("script".to_string(), AttrValue::String("cat".to_string()));
+        node.attrs.insert(
+            "stdin_source".to_string(),
+            AttrValue::String("context.missing".to_string()),
+        );
+        let context = Context::new();
+        let graph = Graph::new("test");
+        let run_dir = tempfile::tempdir().unwrap();
+        let services = make_spy_services(spy.clone());
+
+        let outcome = handler
+            .execute(&node, &context, &graph, run_dir.path(), &services)
+            .await
+            .unwrap();
+
+        assert_eq!(
+            outcome.failure_category(),
+            Some(FailureCategory::Deterministic)
+        );
+        assert!(
+            outcome
+                .failure_reason()
+                .unwrap()
+                .contains("was not found in workflow context")
+        );
+        assert_eq!(spy.captured_command(), None);
+    }
+
+    #[tokio::test]
+    async fn simulation_validates_stdin_source_without_resolving_context() {
+        let handler = CommandHandler;
+        let mut valid = Node::new("valid");
+        valid
+            .attrs
+            .insert("script".to_string(), AttrValue::String("cat".to_string()));
+        valid.attrs.insert(
+            "stdin_source".to_string(),
+            AttrValue::String("context.not_available_in_dry_run".to_string()),
+        );
+        let mut invalid = valid.clone();
+        invalid.id = "invalid".to_string();
+        invalid
+            .attrs
+            .insert("stdin_source".to_string(), AttrValue::Integer(7));
+        let context = Context::new();
+        let graph = Graph::new("test");
+        let run_dir = tempfile::tempdir().unwrap();
+        let services = make_services();
+
+        let valid_outcome = handler
+            .simulate(&valid, &context, &graph, run_dir.path(), &services)
+            .await
+            .unwrap();
+        let invalid_outcome = handler
+            .simulate(&invalid, &context, &graph, run_dir.path(), &services)
+            .await
+            .unwrap();
+
+        assert_eq!(valid_outcome.status, StageOutcome::Succeeded);
+        assert_eq!(
+            invalid_outcome.failure_category(),
+            Some(FailureCategory::Deterministic)
+        );
+    }
+
     struct RefreshingMinter {
         calls: std::sync::atomic::AtomicUsize,
     }

lib/foundation/fabro-types/src/graph.rs

@@ -182,6 +182,11 @@ impl Node {
         self.str_attr("for_each")
     }
 
+    #[must_use]
+    pub fn stdin_source(&self) -> Option<&str> {
+        self.str_attr("stdin_source")
+    }
+
     #[must_use]
     pub fn output_schema(&self) -> Option<&str> {
         self.str_attr("output_schema")
@@ -606,6 +611,7 @@ mod tests {
         assert_eq!(node.node_type(), None);
         assert_eq!(node.prompt(), None);
         assert_eq!(node.for_each(), None);
+        assert_eq!(node.stdin_source(), None);
         assert_eq!(node.output_schema(), None);
         assert_eq!(node.output_retries(), 2);
         assert_eq!(node.max_retries(), None);
@@ -687,6 +693,17 @@ mod tests {
         assert_eq!(node.for_each(), Some("context.candidates"));
     }
 
+    #[test]
+    fn node_stdin_source_returns_context_source() {
+        let mut node = Node::new("merge");
+        node.attrs.insert(
+            "stdin_source".to_string(),
+            AttrValue::String("context.parallel.results".to_string()),
+        );
+
+        assert_eq!(node.stdin_source(), Some("context.parallel.results"));
+    }
+
     #[test]
     fn node_with_attrs() {
         let mut node = Node::new("plan");