From 0eda219376ec7130a0342fb5930c609c71e317cc Mon Sep 17 00:00:00 2001 From: Bryan Helmkamp Date: Wed, 29 Jul 2026 11:14:48 -0400 Subject: [PATCH] Simplify stdin_source plumbing after review - ExecStreamingRequest: drop #[non_exhaustive] and the six Option-taking builder setters; call sites use struct literals over ::new(), matching GrepOptions/WalkOptions, and providers can destructure exhaustively - Docker: pass ExecStreamingRequest through docker_exec_shell_streaming instead of seven positional args; revert the no-op StartExecOptions - Daytona: stdin temp-file cleanup is now best-effort (mirrors DaytonaSession::close) so a failed delete cannot fail a completed command or double-delete from Drop; upload overlaps session creation; one shared DAYTONA_CLEANUP_TIMEOUT - write_process_stdin tolerates ConnectionReset/ConnectionAborted so a command that stops reading stdin does not fail on TCP Docker daemons - Local sandbox aborts the stdin writer after process exit instead of joining unbounded - Cap stdin_source payloads at 10 MiB, mirroring the for_each bound - Add Node::context_key_attr() tri-state so the handler and lint rule share one definition of a valid context-key attribute - inert_attribute canonicalizes handler types via StageHandler, fixing false warnings for command attrs on tool nodes - Share resolve_flat_context_value between command stdin and for_each; resolve_json_value takes Value by value, removing a deep clone - Reuse MockSandbox in command handler stdin tests instead of extending SpySandbox with a hand-rolled streaming override Co-Authored-By: Claude Fable 5 --- docs/public/workflows/stages-and-nodes.mdx | 4 +- lib/components/fabro-agent/src/tools.rs | 14 +- .../fabro-sandbox/src/daytona/mod.rs | 125 ++++++------ lib/components/fabro-sandbox/src/docker.rs | 63 +++--- lib/components/fabro-sandbox/src/local.rs | 50 +++-- lib/components/fabro-sandbox/src/sandbox.rs | 59 ++---- .../tests/daytona_streaming_live.rs | 26 +-- .../fabro-sandbox/tests/docker_streaming.rs | 30 +-- .../src/rules/inert_attribute.rs | 58 ++++-- .../src/rules/stdin_source_valid.rs | 16 +- lib/components/fabro-workflow/src/artifact.rs | 36 +++- .../fabro-workflow/src/handler/command.rs | 186 +++++++----------- .../fabro-workflow/src/handler/parallel.rs | 33 ++-- lib/foundation/fabro-types/src/graph.rs | 43 +++- lib/foundation/fabro-types/src/lib.rs | 4 +- 15 files changed, 372 insertions(+), 375 deletions(-) diff --git a/docs/public/workflows/stages-and-nodes.mdx b/docs/public/workflows/stages-and-nodes.mdx index 6e48db6a9..cdb29884d 100644 --- a/docs/public/workflows/stages-and-nodes.mdx +++ b/docs/public/workflows/stages-and-nodes.mdx @@ -115,8 +115,8 @@ merge_results [ | `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. +JSON. Fabro does not add a newline. A missing source, or a value larger than +10 MiB, fails before the command starts. ### Human diff --git a/lib/components/fabro-agent/src/tools.rs b/lib/components/fabro-agent/src/tools.rs index bcd00ea4e..f892c971c 100644 --- a/lib/components/fabro-agent/src/tools.rs +++ b/lib/components/fabro-agent/src/tools.rs @@ -297,13 +297,13 @@ pub(crate) async fn execute_shell_command( "Injecting sandbox env vars into tool execution" ); ctx.env - .exec_command_streaming( - crate::ExecStreamingRequest::new(command) - .timeout_ms(Some(timeout_ms)) - .working_dir(cwd) - .env_vars(tool_env.as_ref()) - .cancel_token(Some(ctx.cancel.clone())), - ) + .exec_command_streaming(crate::ExecStreamingRequest { + timeout_ms: Some(timeout_ms), + working_dir: cwd, + env_vars: tool_env.as_ref(), + cancel_token: Some(ctx.cancel.clone()), + ..crate::ExecStreamingRequest::new(command) + }) .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 73f371c10..c157e7d52 100644 --- a/lib/components/fabro-sandbox/src/daytona/mod.rs +++ b/lib/components/fabro-sandbox/src/daytona/mod.rs @@ -64,11 +64,10 @@ pub(crate) const DAYTONA_DASHBOARD_SANDBOXES_URL: &str = const FABRO_SANDBOX_USER_AGENT: &str = concat!("fabro-sandbox/", env!("CARGO_PKG_VERSION")); const DAYTONA_PROBE_TIMEOUT: Duration = Duration::from_secs(20); 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); +/// Upper bound on explicit and Drop-triggered Daytona cleanup calls (session +/// deletion, temporary stdin files) so a stalled REST call cannot block +/// cancellation/timeout paths indefinitely. +const DAYTONA_CLEANUP_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] = &[ @@ -1818,17 +1817,19 @@ impl Sandbox for DaytonaSandbox { || 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 stdin_upload = async { + match stdin { + Some(stdin) => DaytonaStdinFile::create(sandbox, &stdin).await.map(Some), + None => Ok(None), + } }; + let (mut stdin_file, mut session) = + tokio::try_join!(stdin_upload, DaytonaSession::create(sandbox))?; let command_with_stdin = stdin_file .as_ref() - .map(|stdin_file| redirect_command_stdin(command, stdin_file.path())); + .map(|stdin_file| stdin_file.redirect(command)); let command = command_with_stdin.as_deref().unwrap_or(command); - let mut session = DaytonaSession::create(sandbox).await?; - let session_command = build_bash_session_command(command, &cwd, env_vars); let session_exec = match session.execute(&session_command, true, true).await { Ok(result) => result, @@ -1982,8 +1983,8 @@ 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?; + if let Some(stdin_file) = stdin_file.as_mut() { + stdin_file.close().await; } Ok(result) } @@ -2166,46 +2167,47 @@ impl DaytonaStdinFile { .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) + let path = format!( + "/tmp/fabro-command-stdin-{:016x}", + rand::rng().random::() + ); + fs.upload_file_bytes(&path, stdin) .await .map_err(|err| crate::Error::context("Failed to upload Daytona command stdin", err))?; - Ok(file) + Ok(Self { fs: Some(fs), path }) } - fn fs(&self) -> &daytona_sdk::FileSystemService { - self.fs - .as_ref() - .expect("DaytonaStdinFile used after removal") + /// Wrap `command` so it reads this file as its standard input. + fn redirect(&self, command: &str) -> String { + redirect_command_stdin(command, &self.path) } - fn path(&self) -> &str { - &self.path - } + /// Idempotent, best-effort deletion bounded by + /// [`DAYTONA_CLEANUP_TIMEOUT`]. Failures are logged rather than surfaced + /// so cleanup can never fail a command that already completed. + async fn close(&mut self) { + let Some(fs) = self.fs.as_ref() else { + return; + }; + let deletion = + time::timeout(DAYTONA_CLEANUP_TIMEOUT, fs.delete_file(&self.path, false)).await; - async fn remove(mut self) -> crate::Result<()> { - let deletion = time::timeout( - DAYTONA_STDIN_FILE_DELETE_TIMEOUT, - self.fs().delete_file(&self.path, false), - ) - .await; + // Keep the service owned until the delete future completes. If this + // method is cancelled at the await above, Drop still has everything it + // needs to retry cleanup. + self.fs.take(); match deletion { - Ok(Ok(())) => { - self.fs.take(); - Ok(()) + Ok(Ok(())) => {} + Ok(Err(err)) => { + tracing::warn!(error = %err, "Failed to delete Daytona command stdin"); + } + Err(_) => { + tracing::warn!( + timeout_ms = + u64::try_from(DAYTONA_CLEANUP_TIMEOUT.as_millis()).unwrap_or(u64::MAX), + "Timed out deleting Daytona command stdin" + ); } - 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() - ))), } } } @@ -2219,11 +2221,7 @@ impl Drop for DaytonaStdinFile { match Handle::try_current() { Ok(handle) => { handle.spawn(async move { - match time::timeout( - DAYTONA_STDIN_FILE_DELETE_TIMEOUT, - fs.delete_file(&path, false), - ) - .await + match time::timeout(DAYTONA_CLEANUP_TIMEOUT, fs.delete_file(&path, false)).await { Ok(Ok(())) => {} Ok(Err(err)) => { @@ -2234,9 +2232,8 @@ impl Drop for DaytonaStdinFile { } Err(_) => { tracing::warn!( - timeout_ms = - u64::try_from(DAYTONA_STDIN_FILE_DELETE_TIMEOUT.as_millis()) - .unwrap_or(u64::MAX), + timeout_ms = u64::try_from(DAYTONA_CLEANUP_TIMEOUT.as_millis()) + .unwrap_or(u64::MAX), "Timed out deleting Daytona command stdin from Drop" ); } @@ -2321,14 +2318,14 @@ impl DaytonaSession { /// Idempotent: a second call after the process service is consumed is a /// no-op. /// - /// `delete_session` is bounded by [`DAYTONA_SESSION_CLOSE_TIMEOUT`] so a + /// `delete_session` is bounded by [`DAYTONA_CLEANUP_TIMEOUT`] so a /// stalled Daytona REST call cannot block cancellation paths indefinitely. async fn close(&mut self, reason: &'static str) { let Some(svc) = self.process_svc.as_ref() else { return; }; let deletion = time::timeout( - DAYTONA_SESSION_CLOSE_TIMEOUT, + DAYTONA_CLEANUP_TIMEOUT, svc.delete_session(&self.session_id), ) .await; @@ -2351,7 +2348,7 @@ impl DaytonaSession { tracing::warn!( session_id = %self.session_id, reason, - timeout_ms = u64::try_from(DAYTONA_SESSION_CLOSE_TIMEOUT.as_millis()) + timeout_ms = u64::try_from(DAYTONA_CLEANUP_TIMEOUT.as_millis()) .unwrap_or(u64::MAX), "timed out deleting Daytona session" ); @@ -2369,11 +2366,8 @@ impl Drop for DaytonaSession { match Handle::try_current() { Ok(handle) => { handle.spawn(async move { - match time::timeout( - DAYTONA_SESSION_CLOSE_TIMEOUT, - svc.delete_session(&session_id), - ) - .await + match time::timeout(DAYTONA_CLEANUP_TIMEOUT, svc.delete_session(&session_id)) + .await { Ok(Ok(())) => {} Ok(Err(err)) => { @@ -2386,9 +2380,8 @@ impl Drop for DaytonaSession { Err(_) => { tracing::warn!( session_id, - timeout_ms = - u64::try_from(DAYTONA_SESSION_CLOSE_TIMEOUT.as_millis()) - .unwrap_or(u64::MAX), + timeout_ms = u64::try_from(DAYTONA_CLEANUP_TIMEOUT.as_millis()) + .unwrap_or(u64::MAX), "Daytona session leaked; timed out deleting from Drop" ); } @@ -3424,11 +3417,11 @@ mod tests { .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") + let mut 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"); + assert!(file.path.starts_with("/tmp/fabro-command-stdin-")); + file.close().await; sandbox_response.assert_async().await; toolbox_response.assert_async().await; diff --git a/lib/components/fabro-sandbox/src/docker.rs b/lib/components/fabro-sandbox/src/docker.rs index e55fedbfc..d69dcca4a 100644 --- a/lib/components/fabro-sandbox/src/docker.rs +++ b/lib/components/fabro-sandbox/src/docker.rs @@ -420,11 +420,7 @@ impl DockerSandbox { &docker, &container_id, exec_opts, - Some(StartExecOptions { - detach: false, - tty: false, - output_capacity: None, - }), + None, "Failed to create exec", "Failed to start exec", ) @@ -540,14 +536,17 @@ impl DockerSandbox { async fn docker_exec_shell_streaming( &self, - command: &str, - timeout_ms: Option, - working_dir: Option<&str>, - env_vars: Option<&HashMap>, - cancel_token: Option, - stdin: 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 effective_dir = working_dir .unwrap_or_else(|| self.working_directory()) @@ -811,15 +810,11 @@ impl DockerSandbox { }); } let result = self - .docker_exec_shell_streaming( - command, - Some(timeout_ms), - Some("/"), - None, - None, - None, - None, - ) + .docker_exec_shell_streaming(ExecStreamingRequest { + timeout_ms: Some(timeout_ms), + working_dir: Some("/"), + ..ExecStreamingRequest::new(command) + }) .await .map_err(|error| DockerCloneFailure { error: crate::Error::context( @@ -1806,25 +1801,13 @@ impl Sandbox for DockerSandbox { &self, 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, - timeout_ms, - dir.as_deref(), - env_vars, - cancel_token, - stdin, - output_callback, - ) + let dir = request + .working_dir + .map(|path| self.resolve_container_path(path)); + self.docker_exec_shell_streaming(ExecStreamingRequest { + working_dir: dir.as_deref(), + ..request + }) .await } diff --git a/lib/components/fabro-sandbox/src/local.rs b/lib/components/fabro-sandbox/src/local.rs index cea90a861..49737780d 100644 --- a/lib/components/fabro-sandbox/src/local.rs +++ b/lib/components/fabro-sandbox/src/local.rs @@ -597,9 +597,21 @@ 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))??; + // The process is gone, so unwritten stdin bytes are unwanted. + // Abort instead of joining unbounded: a backgrounded grandchild + // that inherited the pipe could otherwise block the writer + // forever. + stdin_task.abort(); + match stdin_task.await { + Ok(result) => result?, + Err(join_error) if join_error.is_cancelled() => {} + Err(join_error) => { + return Err(crate::Error::context( + "stdin stream task failed", + join_error, + )); + } + } } let stdout_bytes = stdout_task .await @@ -1367,11 +1379,11 @@ mod tests { let sandbox = LocalSandbox::new(dir.clone()); let result = sandbox - .exec_command_streaming( - ExecStreamingRequest::new(BASH_ONLY_COMMAND) - .timeout_ms(Some(5000)) - .output_callback(Some(Arc::new(|_, _| Box::pin(async { Ok(()) })))), - ) + .exec_command_streaming(ExecStreamingRequest { + timeout_ms: Some(5000), + output_callback: Some(Arc::new(|_, _| Box::pin(async { Ok(()) }))), + ..ExecStreamingRequest::new(BASH_ONLY_COMMAND) + }) .await .unwrap(); @@ -1392,11 +1404,11 @@ mod tests { 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())), - ) + .exec_command_streaming(ExecStreamingRequest { + timeout_ms: Some(5000), + stdin: Some(stdin.clone()), + ..ExecStreamingRequest::new("cat") + }) .await .unwrap(); @@ -1456,12 +1468,12 @@ mod tests { assert_eq!(non_streaming.stdout.trim(), "nonlogin"); let streaming = sandbox - .exec_command_streaming( - ExecStreamingRequest::new(LOGIN_SHELL_REPORT) - .timeout_ms(Some(5000)) - .env_vars(Some(&env_vars)) - .output_callback(Some(Arc::new(|_, _| Box::pin(async { Ok(()) })))), - ) + .exec_command_streaming(ExecStreamingRequest { + timeout_ms: Some(5000), + env_vars: Some(&env_vars), + output_callback: Some(Arc::new(|_, _| Box::pin(async { Ok(()) }))), + ..ExecStreamingRequest::new(LOGIN_SHELL_REPORT) + }) .await .unwrap(); assert_eq!(streaming.result.stdout.trim(), "nonlogin"); diff --git a/lib/components/fabro-sandbox/src/sandbox.rs b/lib/components/fabro-sandbox/src/sandbox.rs index 8d2109714..6e6e2b336 100644 --- a/lib/components/fabro-sandbox/src/sandbox.rs +++ b/lib/components/fabro-sandbox/src/sandbox.rs @@ -761,10 +761,14 @@ pub type CommandOutputCallback = Arc< /// Inputs for a streaming command execution. /// +/// Construct with a struct literal over [`ExecStreamingRequest::new`]: +/// `ExecStreamingRequest { stdin, ..ExecStreamingRequest::new(command) }`. +/// Providers should destructure exhaustively so a new field is a compile +/// error rather than silently ignored input. +/// /// 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, @@ -788,50 +792,27 @@ impl<'a> ExecStreamingRequest<'a> { 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, { + // A command that stops reading its input (`head -1`, an early exit) is + // not an error; its exit code is the authoritative result. Local pipes + // surface that as `BrokenPipe`, remote transports (a TCP Docker daemon) + // as `ConnectionReset`/`ConnectionAborted`. + fn command_stopped_reading(err: &std::io::Error) -> bool { + matches!( + err.kind(), + std::io::ErrorKind::BrokenPipe + | std::io::ErrorKind::ConnectionReset + | std::io::ErrorKind::ConnectionAborted + ) + } + if let Err(err) = writer.write_all(stdin).await { - if err.kind() != std::io::ErrorKind::BrokenPipe { + if !command_stopped_reading(&err) { return Err(crate::Error::context( "Failed to write command standard input", err, @@ -839,7 +820,7 @@ where } } if let Err(err) = writer.shutdown().await { - if err.kind() != std::io::ErrorKind::BrokenPipe { + if !command_stopped_reading(&err) { return Err(crate::Error::context( "Failed to close command standard input", err, diff --git a/lib/components/fabro-sandbox/tests/daytona_streaming_live.rs b/lib/components/fabro-sandbox/tests/daytona_streaming_live.rs index fc7d6665b..85e146041 100644 --- a/lib/components/fabro-sandbox/tests/daytona_streaming_live.rs +++ b/lib/components/fabro-sandbox/tests/daytona_streaming_live.rs @@ -363,14 +363,14 @@ mod daytona_streaming_live { let live_exec = tokio::spawn(async move { sandbox_for_exec - .exec_command_streaming( - fabro_sandbox::ExecStreamingRequest::new( + .exec_command_streaming(fabro_sandbox::ExecStreamingRequest { + timeout_ms: Some(60_000), + cancel_token: Some(cancel_for_exec), + output_callback: 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 }); @@ -498,13 +498,13 @@ mod daytona_streaming_live { let chunks = Arc::new(Mutex::new(Vec::new())); let callback = capture_callback(Arc::clone(&chunks)); let result = sandbox - .exec_command_streaming( - fabro_sandbox::ExecStreamingRequest::new(command) - .timeout_ms(Some(timeout_ms)) - .cancel_token(cancel_token) - .stdin(stdin) - .output_callback(Some(callback)), - ) + .exec_command_streaming(fabro_sandbox::ExecStreamingRequest { + timeout_ms: Some(timeout_ms), + cancel_token, + stdin, + output_callback: Some(callback), + ..fabro_sandbox::ExecStreamingRequest::new(command) + }) .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 a0cfa7548..6274c638c 100644 --- a/lib/components/fabro-sandbox/tests/docker_streaming.rs +++ b/lib/components/fabro-sandbox/tests/docker_streaming.rs @@ -52,11 +52,11 @@ async fn streaming_timeout_terminates_docker_exec_before_returning() { let marker = "fabro_streaming_timeout_sentinel"; let command = format!("trap '' HUP TERM; echo start; sleep 5 # {marker}"); let result = sandbox - .exec_command_streaming( - ExecStreamingRequest::new(&command) - .timeout_ms(Some(200)) - .output_callback(Some(capture_bytes(Arc::clone(&chunks)))), - ) + .exec_command_streaming(ExecStreamingRequest { + timeout_ms: Some(200), + output_callback: Some(capture_bytes(Arc::clone(&chunks))), + ..ExecStreamingRequest::new(&command) + }) .await .expect("streaming command should return a timeout result"); @@ -121,11 +121,11 @@ async fn streaming_command_receives_exact_stdin_and_eof() { 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())), - ) + .exec_command_streaming(ExecStreamingRequest { + timeout_ms: Some(10_000), + stdin: Some(stdin.clone()), + ..ExecStreamingRequest::new("cat") + }) .await .expect("streaming command should read stdin and finish at EOF"); let injection_probe = sandbox @@ -273,11 +273,11 @@ async fn docker_runs_clean_bash_through_both_command_paths() { let chunks = Arc::new(Mutex::new(Vec::new())); let streaming = sandbox - .exec_command_streaming( - ExecStreamingRequest::new(command) - .timeout_ms(Some(10_000)) - .output_callback(Some(capture_bytes(Arc::clone(&chunks)))), - ) + .exec_command_streaming(ExecStreamingRequest { + timeout_ms: Some(10_000), + output_callback: Some(capture_bytes(Arc::clone(&chunks))), + ..ExecStreamingRequest::new(command) + }) .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 af84956de..e9b8cc7a5 100644 --- a/lib/components/fabro-validate/src/rules/inert_attribute.rs +++ b/lib/components/fabro-validate/src/rules/inert_attribute.rs @@ -1,4 +1,5 @@ use fabro_graphviz::graph::{self, Graph}; +use fabro_types::StageHandler; use crate::{Diagnostic, LintRule, Severity}; @@ -6,24 +7,39 @@ pub(super) fn rule() -> Box { Box::new(Rule) } -/// Attributes that only specific handler types read, paired with the handler -/// types that consume them. On every other node type the attribute is inert: -/// accepted by the parser and read by nothing at runtime. +/// Attributes that only specific handlers read, paired with the handlers that +/// consume them. On every other node type the attribute is inert: accepted by +/// the parser and read by nothing at runtime. +/// +/// Node types are compared after canonicalization through +/// [`StageHandler::from_handler_type`], so alias types (`tool` runs the +/// command handler) accept the same attributes as their canonical form. /// /// Attributes read by several handlers (`timeout`), resolved for every node /// (`fidelity`, `retry_policy`, `max_visits`, `goal_gate`), or injectable via /// model stylesheets (`model`, `provider`, `reasoning_effort`, `speed`, /// `backend`) are deliberately not listed. -const HANDLER_SPECIFIC_ATTRS: &[(&str, &[&str])] = &[ - ("script", &["command"]), - ("language", &["command"]), - ("stdin_source", &["command", "tool"]), - ("duration", &["wait"]), - ("max_parallel", &["parallel"]), - ("output_retries", &["agent", "prompt"]), - ("output_schema", &["agent", "prompt", "command"]), - ("prompt", &["agent", "prompt", "parallel.fan_in"]), - ("review_target", &["human"]), +const HANDLER_SPECIFIC_ATTRS: &[(&str, &[StageHandler])] = &[ + ("script", &[StageHandler::Command]), + ("language", &[StageHandler::Command]), + ("stdin_source", &[StageHandler::Command]), + ("duration", &[StageHandler::Wait]), + ("max_parallel", &[StageHandler::Parallel]), + ("output_retries", &[ + StageHandler::Agent, + StageHandler::Prompt, + ]), + ("output_schema", &[ + StageHandler::Agent, + StageHandler::Prompt, + StageHandler::Command, + ]), + ("prompt", &[ + StageHandler::Agent, + StageHandler::Prompt, + StageHandler::ParallelFanIn, + ]), + ("review_target", &[StageHandler::Human]), ]; struct Rule; @@ -38,12 +54,13 @@ impl LintRule for Rule { for node in graph.nodes.values() { // An unknown shape or type is covered by the type_known rule; a // node this rule cannot classify is skipped rather than guessed at. - let Some(handler) = node.handler_type() else { + let Some(raw_type) = node.handler_type() else { continue; }; - if !graph::is_known_handler_type(handler) { + if !graph::is_known_handler_type(raw_type) { continue; } + let handler = StageHandler::from_handler_type(Some(raw_type)); for (attr, consumers) in HANDLER_SPECIFIC_ATTRS { if !node.attrs.contains_key(*attr) { continue; @@ -51,19 +68,22 @@ impl LintRule for Rule { if consumers.contains(&handler) { continue; } + let consumer_names = consumers + .iter() + .map(ToString::to_string) + .collect::>() + .join(", "); diagnostics.push(Diagnostic { rule: self.name().to_string(), severity: Severity::Warning, message: format!( - "Node '{}' (type '{handler}') sets '{attr}', which is only read by {} nodes and has no effect here", + "Node '{}' (type '{raw_type}') sets '{attr}', which is only read by {consumer_names} nodes and has no effect here", node.id, - consumers.join(", "), ), node_id: Some(node.id.clone()), edge: None, fix: Some(format!( - "Remove '{attr}' or change the node to a type that reads it ({})", - consumers.join(", "), + "Remove '{attr}' or change the node to a type that reads it ({consumer_names})", )), ..Diagnostic::default() }); diff --git a/lib/components/fabro-validate/src/rules/stdin_source_valid.rs b/lib/components/fabro-validate/src/rules/stdin_source_valid.rs index b3c59f937..e9f62f3a3 100644 --- a/lib/components/fabro-validate/src/rules/stdin_source_valid.rs +++ b/lib/components/fabro-validate/src/rules/stdin_source_valid.rs @@ -1,4 +1,5 @@ -use fabro_graphviz::graph::Graph; +use fabro_graphviz::graph::{ContextKeyAttr, Graph}; +use fabro_types::StageHandler; use crate::{Diagnostic, LintRule, Severity}; @@ -17,17 +18,20 @@ impl LintRule for Rule { 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()) + StageHandler::from_handler_type(node.handler_type()) == StageHandler::Command + }) + .filter(|node| { + matches!( + node.context_key_attr("stdin_source"), + ContextKeyAttr::Invalid + ) }) .map(|node| Diagnostic { rule: self.name().to_string(), severity: Severity::Error, message: format!( - "Command node '{}' has an empty or non-string 'stdin_source'", + "Node '{}' has an empty or non-string 'stdin_source'", node.id ), node_id: Some(node.id.clone()), diff --git a/lib/components/fabro-workflow/src/artifact.rs b/lib/components/fabro-workflow/src/artifact.rs index 1581911b0..517f5a435 100644 --- a/lib/components/fabro-workflow/src/artifact.rs +++ b/lib/components/fabro-workflow/src/artifact.rs @@ -234,14 +234,12 @@ pub async fn resolve_text_or_blob_ref(value: &Value, run_store: &RunStoreHandle) /// Managed `file://` references are normalized through their content-addressed /// blob id instead of reading an execution-local path. Ordinary strings and /// ordinary file references remain unchanged for the caller to validate. -pub(crate) async fn resolve_json_value(value: &Value, run_store: &RunStoreHandle) -> Result { - let Some(reference) = value.as_str() else { - return Ok(value.clone()); - }; - let Some(blob_id) = +pub(crate) async fn resolve_json_value(value: Value, run_store: &RunStoreHandle) -> Result { + let blob_id = value.as_str().and_then(|reference| { parse_blob_ref(reference).or_else(|| parse_managed_blob_file_ref(reference)) - else { - return Ok(value.clone()); + }); + let Some(blob_id) = blob_id else { + return Ok(value); }; let bytes = read_required_blob(&blob_id, run_store).await?; @@ -249,6 +247,22 @@ pub(crate) async fn resolve_json_value(value: &Value, run_store: &RunStoreHandle .map_err(|err| Error::engine_with_source("artifact blob was not valid JSON", err)) } +/// Resolve a flat workflow context key (`context.NAME` or `NAME`) to a +/// hydrated JSON value. +/// +/// Returns `Ok(None)` when the key is absent from the context, and `Err` when +/// the value exists but its blob reference could not be hydrated. +pub(crate) async fn resolve_flat_context_value( + context: &Context, + key: &str, + run_store: &RunStoreHandle, +) -> Result> { + let Some(value) = context::lookup_flat(context, key) else { + return Ok(None); + }; + resolve_json_value(value, run_store).await.map(Some) +} + pub async fn resolve_text_or_blob_ref_str( current: &str, run_store: &RunStoreHandle, @@ -584,14 +598,14 @@ mod tests { let handle = run_store.clone().into(); assert_eq!( - resolve_json_value(&serde_json::json!(format_blob_ref(&blob_id)), &handle) + resolve_json_value(serde_json::json!(format_blob_ref(&blob_id)), &handle) .await .unwrap(), value ); assert_eq!( resolve_json_value( - &serde_json::json!(format!("file:///sandbox/.fabro/blobs/{blob_id}.json")), + serde_json::json!(format!("file:///sandbox/.fabro/blobs/{blob_id}.json")), &handle, ) .await @@ -606,7 +620,9 @@ mod tests { let value = serde_json::json!([1, 2, 3]); assert_eq!( - resolve_json_value(&value, &run_store.into()).await.unwrap(), + resolve_json_value(value.clone(), &run_store.into()) + .await + .unwrap(), value ); } diff --git a/lib/components/fabro-workflow/src/handler/command.rs b/lib/components/fabro-workflow/src/handler/command.rs index be0389c76..05604d3e6 100644 --- a/lib/components/fabro-workflow/src/handler/command.rs +++ b/lib/components/fabro-workflow/src/handler/command.rs @@ -2,7 +2,7 @@ use std::path::Path; use async_trait::async_trait; use fabro_agent::{CommandOutputCallback, ExecStreamingRequest}; -use fabro_graphviz::graph::{Graph, Node}; +use fabro_graphviz::graph::{ContextKeyAttr, Graph, Node}; use fabro_types::{CommandTermination, StageTiming}; use fabro_util::shell::shell_quote; @@ -10,7 +10,7 @@ use super::structured_output::{self, StructuredOutputError}; use super::{EngineServices, Handler, NodeTimeoutPolicy}; use crate::artifact; use crate::command_log::CommandLogRecorder; -use crate::context::{self, Context, keys}; +use crate::context::{Context, keys}; use crate::error::Error; use crate::event::{Event, StageScope}; use crate::outcome::{Outcome, OutcomeExt}; @@ -32,7 +32,7 @@ impl Handler for CommandHandler { _run_dir: &Path, _services: &EngineServices, ) -> Result { - if let Err(reason) = stdin_source(node) { + if let Err(reason) = validated_stdin_source(node) { return Ok(Outcome::fail_deterministic(reason)); } let script = node @@ -130,14 +130,14 @@ impl Handler for CommandHandler { let result = services .run .sandbox - .exec_command_streaming( - 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)), - ) + .exec_command_streaming(ExecStreamingRequest { + timeout_ms: Some(timeout_ms), + env_vars, + cancel_token: Some(cancel_token.clone()), + stdin, + output_callback: Some(output_callback), + ..ExecStreamingRequest::new(&command) + }) .await; cancel_token.cancel(); let streaming = match result { @@ -223,46 +223,58 @@ impl Handler for CommandHandler { } } -fn stdin_source(node: &Node) -> std::result::Result, String> { - if !node.attrs.contains_key("stdin_source") { - return Ok(None); +/// Ceiling on encoded stdin bytes. `stdin_source` values are runtime data — +/// often model-produced — so their size is not something a workflow author +/// reviewed; this bounds peak memory and remote uploads the same way +/// `MAX_FOR_EACH_ITEMS` bounds `for_each` fan-out. +const MAX_STDIN_BYTES: usize = 10 * 1024 * 1024; + +fn validated_stdin_source(node: &Node) -> Result, String> { + match node.context_key_attr("stdin_source") { + ContextKeyAttr::Absent => Ok(None), + ContextKeyAttr::Invalid => Err(format!( + "Command node '{}' requires 'stdin_source' to be a non-empty string", + node.id + )), + ContextKeyAttr::Present(source) => Ok(Some(source)), } - 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 { +) -> Result>, Outcome> { + let Some(source) = validated_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) + let value = match artifact::resolve_flat_context_value(context, source, &services.run.run_store) .await - .map_err(|err| { - Outcome::fail_deterministic(format!( + { + Ok(Some(value)) => value, + Ok(None) => { + return Err(Outcome::fail_deterministic(format!( + "stdin_source '{source}' was not found in workflow context" + ))); + } + Err(err) => { + return 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}" )) })?; + if stdin.len() > MAX_STDIN_BYTES { + return Err(Outcome::fail_deterministic(format!( + "stdin_source '{source}' resolved to {} bytes, above the limit of {MAX_STDIN_BYTES}. \ + Reduce the value in the node that produces it, or pass it through a file instead.", + stdin.len() + ))); + } Ok(Some(stdin)) } @@ -313,6 +325,7 @@ mod tests { use bytes::Bytes; use fabro_graphviz::graph::AttrValue; + use fabro_sandbox::test_support::MockSandbox; use fabro_store::{Database, RunDatabase, StageId}; use fabro_types::{Graph, RunProjection, RunSpec, WorkflowSettings, fixtures, test_support}; use object_store::memory::InMemory; @@ -830,7 +843,7 @@ mod tests { let context = Context::new(); let graph = Graph::new("test"); let run_dir = tempfile::tempdir().unwrap(); - let mut services = make_spy_services(spy.clone()); + let mut services = make_sandbox_services(spy.clone()); let event_names = Arc::new(std::sync::Mutex::new(Vec::new())); let captured_event_names = Arc::clone(&event_names); let emitter = Arc::new(crate::event::Emitter::new(fixtures::RUN_1)); @@ -1401,7 +1414,6 @@ 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 { @@ -1412,7 +1424,6 @@ 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), } } @@ -1429,17 +1440,12 @@ 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] @@ -1479,42 +1485,6 @@ 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, @@ -1557,7 +1527,7 @@ mod tests { } } - fn make_spy_services(sandbox: std::sync::Arc) -> EngineServices { + fn make_sandbox_services(sandbox: std::sync::Arc) -> EngineServices { let mut services = make_services(); services.run = services.run.with_sandbox(sandbox); services @@ -1565,13 +1535,7 @@ mod tests { #[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 mock = std::sync::Arc::new(MockSandbox::default()); let handler = CommandHandler; let mut node = Node::new("merge"); node.attrs @@ -1591,7 +1555,7 @@ mod tests { 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 services = make_sandbox_services(mock.clone()); let outcome = handler .execute(&node, &context, &graph, run_dir.path(), &services) @@ -1600,11 +1564,15 @@ mod tests { assert_eq!(outcome.status, StageOutcome::Succeeded); assert_eq!( - spy.captured_stdin(), + *mock.captured_stdin.lock().unwrap(), Some(serde_json::to_vec(¶llel_results).unwrap()) ); assert!( - !spy.captured_command() + !mock + .captured_command + .lock() + .unwrap() + .clone() .expect("command should run") .contains("must-not-run"), "stdin content must not be inserted into shell source" @@ -1613,13 +1581,7 @@ mod tests { #[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 mock = std::sync::Arc::new(MockSandbox::default()); let handler = CommandHandler; let mut node = Node::new("consume"); node.attrs @@ -1632,7 +1594,7 @@ mod tests { 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 services = make_sandbox_services(mock.clone()); let outcome = handler .execute(&node, &context, &graph, run_dir.path(), &services) @@ -1641,20 +1603,14 @@ mod tests { assert_eq!(outcome.status, StageOutcome::Succeeded); assert_eq!( - spy.captured_stdin().as_deref(), + mock.captured_stdin.lock().unwrap().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 mock = std::sync::Arc::new(MockSandbox::default()); let handler = CommandHandler; let mut node = Node::new("consume"); node.attrs @@ -1666,7 +1622,7 @@ mod tests { let context = Context::new(); let graph = Graph::new("test"); let run_dir = tempfile::tempdir().unwrap(); - let services = make_spy_services(spy.clone()); + let services = make_sandbox_services(mock.clone()); let outcome = handler .execute(&node, &context, &graph, run_dir.path(), &services) @@ -1683,7 +1639,7 @@ mod tests { .unwrap() .contains("was not found in workflow context") ); - assert_eq!(spy.captured_command(), None); + assert_eq!(*mock.captured_command.lock().unwrap(), None); } #[tokio::test] @@ -1758,7 +1714,7 @@ mod tests { let graph = Graph::new("test"); let run_dir = tempfile::tempdir().unwrap(); - let services = make_spy_services(spy.clone()); + let services = make_sandbox_services(spy.clone()); let outcome = handler .execute(&node, &context, &graph, run_dir.path(), &services) .await @@ -1808,7 +1764,7 @@ mod tests { &context, &graph, run_dir.path(), - &make_spy_services(spy.clone()), + &make_sandbox_services(spy.clone()), ) .await .unwrap(); @@ -1839,7 +1795,7 @@ mod tests { let graph = Graph::new("test"); let run_dir = tempfile::tempdir().unwrap(); - let mut services = make_spy_services(spy.clone()); + let mut services = make_sandbox_services(spy.clone()); services .base_env .insert("MY_VAR".to_string(), "my_value".to_string()); @@ -1868,7 +1824,7 @@ mod tests { let minter = std::sync::Arc::new(RefreshingMinter { calls: std::sync::atomic::AtomicUsize::new(0), }); - let mut services = make_spy_services(spy.clone()); + let mut services = make_sandbox_services(spy.clone()); services.github_token = Some(std::sync::Arc::new( crate::github_token_source::GitHubTokenSource::mintable(minter.clone()), )); @@ -1929,7 +1885,7 @@ mod tests { let graph = Graph::new("test"); let run_dir = tempfile::tempdir().unwrap(); - let mut services = make_spy_services(spy.clone()); + let mut services = make_sandbox_services(spy.clone()); services.run = services .run .with_cancel_token(tokio_util::sync::CancellationToken::new()); @@ -1968,7 +1924,7 @@ mod tests { &context, &graph, run_dir.path(), - &make_spy_services(spy), + &make_sandbox_services(spy), ) .await .unwrap_err(); @@ -2055,7 +2011,7 @@ mod tests { let context = Context::new(); let graph = Graph::new("test"); let run_dir = tempfile::tempdir().unwrap(); - let services = make_spy_services(std::sync::Arc::new(SpySandbox::fail("No such file"))); + let services = make_sandbox_services(std::sync::Arc::new(SpySandbox::fail("No such file"))); let err = handler .execute(&node, &context, &graph, run_dir.path(), &services) diff --git a/lib/components/fabro-workflow/src/handler/parallel.rs b/lib/components/fabro-workflow/src/handler/parallel.rs index 3e8a9c9ba..19a538cc6 100644 --- a/lib/components/fabro-workflow/src/handler/parallel.rs +++ b/lib/components/fabro-workflow/src/handler/parallel.rs @@ -16,9 +16,7 @@ use tokio::time::sleep; use uuid::Uuid; use super::{EngineServices, Handler}; -use crate::context::{ - self, Context, ParallelBranchPreamble, WorkflowContext, context_diff_public, keys, -}; +use crate::context::{Context, ParallelBranchPreamble, WorkflowContext, context_diff_public, keys}; use crate::error::Error; use crate::event::{Emitter, Event, RunNoticeCode, RunNoticeLevel, StageScope}; use crate::hook_context::set_hook_node; @@ -215,24 +213,25 @@ async fn build_branch_plan( // data, so an absent or unusable source stands in one placeholder item. // Graph-shape mistakes above still fail, because a dry run should catch // those. - let resolved = match context::lookup_flat(context, source) { - Some(raw_items) => { - match artifact::resolve_json_value(&raw_items, &services.run.run_store).await { - Ok(value) => Some(value), - Err(_) if simulated => None, - Err(err) => { - return Err(Outcome::fail_deterministic(format!( - "for_each source '{source}' could not be resolved: {err}" - ))); - } - } - } - None if simulated => None, - None => { + let resolved = match artifact::resolve_flat_context_value( + context, + source, + &services.run.run_store, + ) + .await + { + Ok(Some(value)) => Some(value), + Ok(None) | Err(_) if simulated => None, + Ok(None) => { return Err(Outcome::fail_deterministic(format!( "for_each source '{source}' was not found in workflow context" ))); } + Err(err) => { + return Err(Outcome::fail_deterministic(format!( + "for_each source '{source}' could not be resolved: {err}" + ))); + } }; let items = match resolved { Some(serde_json::Value::Array(items)) => items, diff --git a/lib/foundation/fabro-types/src/graph.rs b/lib/foundation/fabro-types/src/graph.rs index 9a7bcf93c..5d7e6b231 100644 --- a/lib/foundation/fabro-types/src/graph.rs +++ b/lib/foundation/fabro-types/src/graph.rs @@ -116,6 +116,19 @@ pub fn shape_to_handler_type(shape: &str) -> Option<&'static str> { } } +/// Presence and validity of a node attribute whose value names a workflow +/// context key (`for_each`, `stdin_source`). +/// +/// Consumers need three states: the attribute is not set, it is set but not a +/// usable key (non-string or blank), or it carries a key. Modeling this once +/// keeps lint rules and handlers agreeing on what "valid" means. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub enum ContextKeyAttr<'a> { + Absent, + Invalid, + Present(&'a str), +} + /// A node in the workflow graph. #[derive(Debug, Clone, PartialEq, Serialize, Deserialize)] pub struct Node { @@ -183,8 +196,14 @@ impl Node { } #[must_use] - pub fn stdin_source(&self) -> Option<&str> { - self.str_attr("stdin_source") + pub fn context_key_attr(&self, name: &str) -> ContextKeyAttr<'_> { + let Some(value) = self.attrs.get(name) else { + return ContextKeyAttr::Absent; + }; + match value.as_str() { + Some(source) if !source.trim().is_empty() => ContextKeyAttr::Present(source), + _ => ContextKeyAttr::Invalid, + } } #[must_use] @@ -611,7 +630,10 @@ 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.context_key_attr("stdin_source"), + ContextKeyAttr::Absent + ); assert_eq!(node.output_schema(), None); assert_eq!(node.output_retries(), 2); assert_eq!(node.max_retries(), None); @@ -694,14 +716,25 @@ mod tests { } #[test] - fn node_stdin_source_returns_context_source() { + fn node_context_key_attr_classifies_presence_and_validity() { 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")); + assert_eq!( + node.context_key_attr("stdin_source"), + ContextKeyAttr::Present("context.parallel.results") + ); + + for invalid in [AttrValue::String(" ".to_string()), AttrValue::Integer(3)] { + node.attrs.insert("stdin_source".to_string(), invalid); + assert_eq!( + node.context_key_attr("stdin_source"), + ContextKeyAttr::Invalid + ); + } } #[test] diff --git a/lib/foundation/fabro-types/src/lib.rs b/lib/foundation/fabro-types/src/lib.rs index 8b5288c7b..ad9142e51 100644 --- a/lib/foundation/fabro-types/src/lib.rs +++ b/lib/foundation/fabro-types/src/lib.rs @@ -72,8 +72,8 @@ pub use event_envelope::EventEnvelope; pub use fabro_model::ReasoningEffort; pub use failure_signature::FailureSignature; pub use graph::{ - AttrValue, Edge, Graph, KNOWN_HANDLER_TYPES, Node, is_known_handler_type, is_llm_handler_type, - shape_to_handler_type, + AttrValue, ContextKeyAttr, Edge, Graph, KNOWN_HANDLER_TYPES, Node, is_known_handler_type, + is_llm_handler_type, shape_to_handler_type, }; pub use interview::{ InterviewQuestionRecord, QuestionType, ReviewTarget, ReviewTargetError, ReviewTargetKind,