scriptoriumadmin/fabro
1
0
Fork 0
mirror of https://github.com/fabro-sh/fabro.git synced 2026-08-28 05:27:41 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 4

Simplify stdin_source plumbing after review

Browse source 
- 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 <noreply@anthropic.com>
This commit is contained in:
Bryan Helmkamp 2026-07-29 11:14:48 -04:00
parent 5d72f9a538
commit 0eda219376
No known key found for this signature in database
15 changed files with 372 additions and 375 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

4
docs/public/workflows/stages-and-nodes.mdx
View file

@ -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

14
lib/components/fabro-agent/src/tools.rs
View file

@ -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()))
}

125
lib/components/fabro-sandbox/src/daytona/mod.rs
View file

@ -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::<u64>()
);
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;

63
lib/components/fabro-sandbox/src/docker.rs
View file

@ -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<u64>,
working_dir: Option<&str>,
env_vars: Option<&HashMap<String, String>>,
cancel_token: Option<CancellationToken>,
stdin: Option<Vec<u8>>,
output_callback: Option<CommandOutputCallback>,
request: ExecStreamingRequest<'_>,
) -> crate::Result<ExecStreamingResult> {
let ExecStreamingRequest {
command,
timeout_ms,
working_dir,
env_vars,
cancel_token,
stdin,
output_callback,
} = request;
let start = Instant::now();
let 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<ExecStreamingResult> {
let ExecStreamingRequest {
command,
timeout_ms,
working_dir,
env_vars,
cancel_token,
stdin,
output_callback,
} = request;
let dir = working_dir.map(|path| self.resolve_container_path(path));
self.docker_exec_shell_streaming(
command,
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
}

50
lib/components/fabro-sandbox/src/local.rs
View file

@ -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");

59
lib/components/fabro-sandbox/src/sandbox.rs
View file

@ -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<u64>,
@ -788,50 +792,27 @@ impl<'a> ExecStreamingRequest<'a> {
output_callback: None,
}
}
#[must_use]
pub fn timeout_ms(mut self, timeout_ms: Option<u64>) -> Self {
self.timeout_ms = timeout_ms;
self
}
#[must_use]
pub fn working_dir(mut self, working_dir: Option<&'a str>) -> Self {
self.working_dir = working_dir;
self
}
#[must_use]
pub fn env_vars(mut self, env_vars: Option<&'a HashMap<String, String>>) -> Self {
self.env_vars = env_vars;
self
}
#[must_use]
pub fn cancel_token(mut self, cancel_token: Option<CancellationToken>) -> Self {
self.cancel_token = cancel_token;
self
}
#[must_use]
pub fn stdin(mut self, stdin: Option<Vec<u8>>) -> Self {
self.stdin = stdin;
self
}
#[must_use]
pub fn output_callback(mut self, output_callback: Option<CommandOutputCallback>) -> Self {
self.output_callback = output_callback;
self
}
}
pub(crate) async fn write_process_stdin<W>(mut writer: W, stdin: &[u8]) -> crate::Result<()>
where
W: AsyncWrite + Unpin,
{
// 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,

26
lib/components/fabro-sandbox/tests/daytona_streaming_live.rs
View file

@ -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();

30
lib/components/fabro-sandbox/tests/docker_streaming.rs
View file

@ -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");

58
lib/components/fabro-validate/src/rules/inert_attribute.rs
View file

@ -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<dyn LintRule> {
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::<Vec<_>>()
.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()
});

16
lib/components/fabro-validate/src/rules/stdin_source_valid.rs
View file

@ -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()),

36
lib/components/fabro-workflow/src/artifact.rs
View file

@ -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<Value> {
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<Value> {
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<Option<Value>> {
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
);
}

186
lib/components/fabro-workflow/src/handler/command.rs
View file

@ -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<Outcome, Error> {
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<Option<&str>, 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<Option<&str>, 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<Option<Vec<u8>>, Outcome> {
let Some(source) = stdin_source(node).map_err(Outcome::fail_deterministic)? else {
) -> Result<Option<Vec<u8>>, 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<Option<String>>,
captured_env_vars: std::sync::Mutex<Option<std::collections::HashMap<String, String>>>,
captured_cancel_token: std::sync::Mutex<Option<bool>>,
captured_stdin: std::sync::Mutex<Option<Vec<u8>>>,
}
impl SpySandbox {
@ -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<String> {
self.captured_command.lock().unwrap().clone()
}
fn captured_stdin(&self) -> Option<Vec<u8>> {
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<fabro_agent::sandbox::ExecStreamingResult> {
*self.captured_stdin.lock().unwrap() = request.stdin;
let result = self
.exec_command(
request.command,
request.timeout_ms.unwrap_or(u64::MAX),
request.working_dir,
request.env_vars,
request.cancel_token,
)
.await?;
if let Some(callback) = request.output_callback.as_ref() {
if !result.stdout.is_empty() {
callback(
fabro_types::CommandOutputStream::Stdout,
result.stdout.as_bytes().to_vec(),
)
.await?;
}
if !result.stderr.is_empty() {
callback(
fabro_types::CommandOutputStream::Stderr,
result.stderr.as_bytes().to_vec(),
)
.await?;
}
}
Ok(fabro_agent::sandbox::ExecStreamingResult {
result,
streams_separated: true,
live_streaming: false,
})
}
async fn grep(
&self,
_: &str,
@ -1557,7 +1527,7 @@ mod tests {
}
}
fn make_spy_services(sandbox: std::sync::Arc<SpySandbox>) -> EngineServices {
fn make_sandbox_services(sandbox: std::sync::Arc<dyn fabro_agent::Sandbox>) -> 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(&parallel_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)

33
lib/components/fabro-workflow/src/handler/parallel.rs
View file

@ -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,

43
lib/foundation/fabro-types/src/graph.rs
View file

@ -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]

4
lib/foundation/fabro-types/src/lib.rs
View file

@ -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,