From 5d72f9a53820de34018b554d45a2c73629190855 Mon Sep 17 00:00:00 2001 From: Bryan Helmkamp Date: Wed, 29 Jul 2026 10:25:51 -0400 Subject: [PATCH] Add context-sourced command stdin --- docs/public/reference/dot-language.mdx | 1 + docs/public/workflows/stages-and-nodes.mdx | 11 + docs/public/workflows/variables.mdx | 3 + lib/components/fabro-agent/src/lib.rs | 6 +- lib/components/fabro-agent/src/sandbox.rs | 8 +- lib/components/fabro-agent/src/tools.rs | 11 +- .../fabro-sandbox/src/daytona/mod.rs | 254 ++++++++++++++- lib/components/fabro-sandbox/src/docker.rs | 54 +++- lib/components/fabro-sandbox/src/lib.rs | 6 +- lib/components/fabro-sandbox/src/local.rs | 82 +++-- lib/components/fabro-sandbox/src/sandbox.rs | 147 +++++++-- .../fabro-sandbox/src/test_support.rs | 29 +- .../tests/daytona_streaming_live.rs | 67 +++- .../fabro-sandbox/tests/docker_streaming.rs | 84 ++++- .../src/rules/inert_attribute.rs | 9 +- .../fabro-validate/src/rules/mod.rs | 2 + .../src/rules/stdin_source_valid.rs | 91 ++++++ .../fabro-workflow/src/handler/command.rs | 294 +++++++++++++++++- lib/foundation/fabro-types/src/graph.rs | 17 + 19 files changed, 1037 insertions(+), 139 deletions(-) create mode 100644 lib/components/fabro-validate/src/rules/stdin_source_valid.rs diff --git a/docs/public/reference/dot-language.mdx b/docs/public/reference/dot-language.mdx index edc9b921e..4c24f640a 100644 --- a/docs/public/reference/dot-language.mdx +++ b/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 diff --git a/docs/public/workflows/stages-and-nodes.mdx b/docs/public/workflows/stages-and-nodes.mdx index 39fde175f..6e48db6a9 100644 --- a/docs/public/workflows/stages-and-nodes.mdx +++ b/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 diff --git a/docs/public/workflows/variables.mdx b/docs/public/workflows/variables.mdx index 782ee743f..6093a6766 100644 --- a/docs/public/workflows/variables.mdx +++ b/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 diff --git a/lib/components/fabro-agent/src/lib.rs b/lib/components/fabro-agent/src/lib.rs index 96552968c..c9d7f2e3d 100644 --- a/lib/components/fabro-agent/src/lib.rs +++ b/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, diff --git a/lib/components/fabro-agent/src/sandbox.rs b/lib/components/fabro-agent/src/sandbox.rs index 3b007e75b..47fbf894c 100644 --- a/lib/components/fabro-agent/src/sandbox.rs +++ b/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, }; diff --git a/lib/components/fabro-agent/src/tools.rs b/lib/components/fabro-agent/src/tools.rs index 7009e1a8a..bcd00ea4e 100644 --- a/lib/components/fabro-agent/src/tools.rs +++ b/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())) diff --git a/lib/components/fabro-sandbox/src/daytona/mod.rs b/lib/components/fabro-sandbox/src/daytona/mod.rs index 36b94f9ee..73f371c10 100644 --- a/lib/components/fabro-sandbox/src/daytona/mod.rs +++ b/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, - working_dir: Option<&str>, - env_vars: Option<&HashMap>, - cancel_token: Option, - output_callback: Option, + request: ExecStreamingRequest<'_>, ) -> crate::Result { + 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, + path: String, +} + +impl DaytonaStdinFile { + async fn create(sandbox: &daytona_sdk::Sandbox, stdin: &[u8]) -> crate::Result { + 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( diff --git a/lib/components/fabro-sandbox/src/docker.rs b/lib/components/fabro-sandbox/src/docker.rs index f6cb40891..e55fedbfc 100644 --- a/lib/components/fabro-sandbox/src/docker.rs +++ b/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, working_dir: Option, env: Option>, + stdin: Option>, output_callback: Option, ) -> crate::Result<(Vec, Vec, 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>, cancel_token: Option, + stdin: Option>, output_callback: Option, ) -> crate::Result { 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, - working_dir: Option<&str>, - env_vars: Option<&HashMap>, - cancel_token: Option, - output_callback: Option, + request: ExecStreamingRequest<'_>, ) -> crate::Result { + 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 diff --git a/lib/components/fabro-sandbox/src/lib.rs b/lib/components/fabro-sandbox/src/lib.rs index 06285ae37..3a5e91ab2 100644 --- a/lib/components/fabro-sandbox/src/lib.rs +++ b/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; diff --git a/lib/components/fabro-sandbox/src/local.rs b/lib/components/fabro-sandbox/src/local.rs index 937abdc64..cea90a861 100644 --- a/lib/components/fabro-sandbox/src/local.rs +++ b/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, - working_dir: Option<&str>, - env_vars: Option<&std::collections::HashMap>, - cancel_token: Option, - output_callback: Option, + request: ExecStreamingRequest<'_>, ) -> crate::Result { + 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(); diff --git a/lib/components/fabro-sandbox/src/sandbox.rs b/lib/components/fabro-sandbox/src/sandbox.rs index b29ccea39..8d2109714 100644 --- a/lib/components/fabro-sandbox/src/sandbox.rs +++ b/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, - working_dir: Option<&str>, - env_vars: Option<&std::collections::HashMap>, - cancel_token: Option, - 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, + pub working_dir: Option<&'a str>, + pub env_vars: Option<&'a HashMap>, + pub cancel_token: Option, + pub stdin: Option>, + pub output_callback: Option, +} + +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) -> 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>) -> Self { + self.env_vars = env_vars; + self + } + + #[must_use] + pub fn cancel_token(mut self, cancel_token: Option) -> Self { + self.cancel_token = cancel_token; + self + } + + #[must_use] + pub fn stdin(mut self, stdin: Option>) -> Self { + self.stdin = stdin; + self + } + + #[must_use] + pub fn output_callback(mut self, output_callback: Option) -> Self { + self.output_callback = output_callback; + self + } +} + +pub(crate) async fn write_process_stdin(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, - working_dir: Option<&str>, - env_vars: Option<&std::collections::HashMap>, - cancel_token: Option, - output_callback: Option, + request: ExecStreamingRequest<'_>, ) -> crate::Result { - 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!( "{}: {}", diff --git a/lib/components/fabro-sandbox/src/test_support.rs b/lib/components/fabro-sandbox/src/test_support.rs index 65ff4af25..7364dc8b0 100644 --- a/lib/components/fabro-sandbox/src/test_support.rs +++ b/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>>, /// Captures the `env_vars` argument from `exec_command` calls. pub captured_env_vars: Mutex>>, + /// Captures the bytes passed to a streaming command's standard input. + pub captured_stdin: Mutex>>, pub active: AtomicBool, pub activate_error: Option, pub activate_calls: Mutex, @@ -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, - working_dir: Option<&str>, - env_vars: Option<&std::collections::HashMap>, - cancel_token: Option, - output_callback: Option, + request: ExecStreamingRequest<'_>, ) -> crate::Result { + 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, diff --git a/lib/components/fabro-sandbox/tests/daytona_streaming_live.rs b/lib/components/fabro-sandbox/tests/daytona_streaming_live.rs index c0c42ebd5..fc7d6665b 100644 --- a/lib/components/fabro-sandbox/tests/daytona_streaming_live.rs +++ b/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, + ) -> Result<(ExecStreamingResult, Vec)> { + 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, + stdin: Option>, ) -> Result<(ExecStreamingResult, Vec)> { 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(); diff --git a/lib/components/fabro-sandbox/tests/docker_streaming.rs b/lib/components/fabro-sandbox/tests/docker_streaming.rs index f4833b1e7..a0cfa7548 100644 --- a/lib/components/fabro-sandbox/tests/docker_streaming.rs +++ b/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>>) -> 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"); diff --git a/lib/components/fabro-validate/src/rules/inert_attribute.rs b/lib/components/fabro-validate/src/rules/inert_attribute.rs index ac2629422..af84956de 100644 --- a/lib/components/fabro-validate/src/rules/inert_attribute.rs +++ b/lib/components/fabro-validate/src/rules/inert_attribute.rs @@ -17,6 +17,7 @@ pub(super) fn rule() -> Box { 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"), diff --git a/lib/components/fabro-validate/src/rules/mod.rs b/lib/components/fabro-validate/src/rules/mod.rs index b2085c9e7..24776f6be 100644 --- a/lib/components/fabro-validate/src/rules/mod.rs +++ b/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> { 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(), diff --git a/lib/components/fabro-validate/src/rules/stdin_source_valid.rs b/lib/components/fabro-validate/src/rules/stdin_source_valid.rs new file mode 100644 index 000000000..b3c59f937 --- /dev/null +++ b/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 { + Box::new(Rule) +} + +struct Rule; + +impl LintRule for Rule { + fn name(&self) -> &'static str { + "stdin_source_valid" + } + + fn apply(&self, graph: &Graph) -> Vec { + 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")); + } + } +} diff --git a/lib/components/fabro-workflow/src/handler/command.rs b/lib/components/fabro-workflow/src/handler/command.rs index 74a32f0ac..be0389c76 100644 --- a/lib/components/fabro-workflow/src/handler/command.rs +++ b/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 { + 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, 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>, 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> { + 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>, @@ -1326,6 +1401,7 @@ mod tests { captured_command: std::sync::Mutex>, captured_env_vars: std::sync::Mutex>>, captured_cancel_token: std::sync::Mutex>, + captured_stdin: std::sync::Mutex>>, } 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 { self.captured_command.lock().unwrap().clone() } + + fn captured_stdin(&self) -> Option> { + 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 { + *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(¶llel_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, } diff --git a/lib/foundation/fabro-types/src/graph.rs b/lib/foundation/fabro-types/src/graph.rs index fdc746f08..9a7bcf93c 100644 --- a/lib/foundation/fabro-types/src/graph.rs +++ b/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");