pub mod audio;
pub mod boolean;
pub mod file;
pub mod files;
pub mod image;
pub mod json;
pub mod number;
pub mod text;
pub mod video;
pub fn definitions() -> Vec<NodeDefinition> {
vec![
text::definition(),
number::definition(),
boolean::definition(),
json::definition(),
file::definition(),
files::definition(),
image::definition(),
audio::definition(),
video::definition(),
]
}
use async_trait::async_trait;
use serde::{Deserialize, Deserializer, Serialize};
use serde_json::{Value, json};
use crate::pipeline::nodes::shared::file_ref::is_file_ref;
use crate::pipeline::nodes::shared::util::{resolve_path, with_answer};
use crate::pipeline::model::{
DslFlag, DslFlagKind, NodeExample, NodeFieldDef, NodeFieldType, PipelineGraph,
};
use crate::pipeline::{
NodeDefinition, PipelineError,
nodes::{NodeExecutionInput, NodeExecutionOutput, NodeHandler},
};
pub const KIND_PREFIX: &str = "input.";
const INPUT_PIN_IN: &str = "in";
const OUTPUT_PIN_OUT: &str = "out";
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum InputKind {
Text,
Number,
Boolean,
Json,
File,
Files,
Image,
Audio,
Video,
}
impl InputKind {
pub const ALL: [InputKind; 9] = [
InputKind::Text,
InputKind::Number,
InputKind::Boolean,
InputKind::Json,
InputKind::File,
InputKind::Files,
InputKind::Image,
InputKind::Audio,
InputKind::Video,
];
pub fn from_kind(kind: &str) -> Option<Self> {
Self::ALL.into_iter().find(|k| k.node_kind() == kind)
}
pub fn node_kind(self) -> &'static str {
match self {
InputKind::Text => text::NODE_KIND,
InputKind::Number => number::NODE_KIND,
InputKind::Boolean => boolean::NODE_KIND,
InputKind::Json => json::NODE_KIND,
InputKind::File => file::NODE_KIND,
InputKind::Files => files::NODE_KIND,
InputKind::Image => image::NODE_KIND,
InputKind::Audio => audio::NODE_KIND,
InputKind::Video => video::NODE_KIND,
}
}
pub fn short(self) -> &'static str {
self.node_kind().trim_start_matches(KIND_PREFIX)
}
pub fn is_file(self) -> bool {
matches!(
self,
InputKind::File | InputKind::Files | InputKind::Image | InputKind::Audio | InputKind::Video
)
}
pub fn preset_accept(self) -> Option<&'static str> {
match self {
InputKind::Image => Some("image"),
InputKind::Audio => Some("audio"),
InputKind::Video => Some("video"),
_ => None,
}
}
pub fn envelope_slot(self) -> &'static str {
if self.is_file() { "files" } else { "body" }
}
pub fn expects(self) -> &'static str {
match self {
InputKind::Text => "a non-empty string",
InputKind::Number => "a number",
InputKind::Boolean => "a boolean",
InputKind::Json => "a JSON value",
InputKind::File => "one file",
InputKind::Files => "one or more files",
InputKind::Image => "one image file",
InputKind::Audio => "one audio file",
InputKind::Video => "one video file",
}
}
fn title(self) -> &'static str {
match self {
InputKind::Text => "Input: Text",
InputKind::Number => "Input: Number",
InputKind::Boolean => "Input: Boolean",
InputKind::Json => "Input: JSON",
InputKind::File => "Input: File",
InputKind::Files => "Input: Files",
InputKind::Image => "Input: Image",
InputKind::Audio => "Input: Audio",
InputKind::Video => "Input: Video",
}
}
}
pub fn is_input_kind(kind: &str) -> bool {
InputKind::from_kind(kind).is_some()
}
fn string_or_list<'de, D: Deserializer<'de>>(de: D) -> Result<Vec<String>, D::Error> {
let raw = Value::deserialize(de)?;
Ok(match raw {
Value::Array(items) => {
let words: Vec<String> = items
.into_iter()
.filter_map(|v| v.as_str().map(|s| s.trim().to_ascii_lowercase()))
.filter(|s| !s.is_empty())
.collect();
if let Some(joined) = words.iter().find(|w| w.contains(',')) {
return Err(serde::de::Error::custom(format!(
"--accept '{joined}' holds a comma; give one word per flag: --accept {}",
joined.split(',').map(str::trim).collect::<Vec<_>>().join(" --accept ")
)));
}
words
}
Value::String(s) => s
.split(',')
.map(|p| p.trim().to_ascii_lowercase())
.filter(|p| !p.is_empty())
.collect(),
_ => Vec::new(),
})
}
fn number_or_string<'de, D: Deserializer<'de>>(de: D) -> Result<Option<f64>, D::Error> {
let raw = Value::deserialize(de)?;
Ok(match raw {
Value::Number(n) => n.as_f64(),
Value::String(s) => s.trim().parse::<f64>().ok(),
_ => None,
})
}
#[derive(Debug, Clone, Serialize, Deserialize, Default)]
pub struct Config {
#[serde(default)]
pub name: String,
#[serde(default)]
pub label: Option<String>,
#[serde(default)]
pub optional: bool,
#[serde(default)]
pub default: Option<Value>,
#[serde(default, deserialize_with = "string_or_list")]
pub accept: Vec<String>,
#[serde(default, deserialize_with = "number_or_string")]
pub min: Option<f64>,
#[serde(default, deserialize_with = "number_or_string")]
pub max: Option<f64>,
}
impl Config {
pub fn effective_accept(&self, kind: InputKind) -> Vec<String> {
if !self.accept.is_empty() {
return self.accept.clone();
}
kind.preset_accept().map(|k| vec![k.to_string()]).unwrap_or_default()
}
}
fn missing(kind: InputKind, name: &str) -> PipelineError {
PipelineError::new(
"FW_NODE_INPUT_MISSING",
format!(
"input '{name}' is required: expected {} at $trigger.{}.{name}, nothing was sent",
kind.expects(),
kind.envelope_slot()
),
)
}
fn invalid(kind: InputKind, name: &str, detail: impl AsRef<str>) -> PipelineError {
PipelineError::new(
"FW_NODE_INPUT_INVALID",
format!(
"input '{name}' expected {} at $trigger.{}.{name}: {}",
kind.expects(),
kind.envelope_slot(),
detail.as_ref()
),
)
}
fn describe(value: &Value) -> String {
match value {
Value::Null => "null".to_string(),
Value::Bool(_) => "a boolean".to_string(),
Value::Number(_) => "a number".to_string(),
Value::String(s) => format!("a string ({} chars)", s.chars().count()),
Value::Array(items) => format!("an array of {}", items.len()),
Value::Object(_) if is_file_ref(value) => "a file".to_string(),
Value::Object(_) => "an object".to_string(),
}
}
fn file_matches(file: &Value, accept: &str) -> bool {
let kind = file["kind"].as_str().unwrap_or("").to_ascii_lowercase();
let mime = file["mime"].as_str().unwrap_or("").to_ascii_lowercase();
let mime = mime.split(';').next().unwrap_or("").trim();
let ext = file["filename"]
.as_str()
.and_then(|f| f.rsplit_once('.'))
.map(|(_, e)| e.to_ascii_lowercase())
.unwrap_or_default();
let accept = accept.trim().trim_start_matches('.').to_ascii_lowercase();
if accept.is_empty() {
return true;
}
kind == accept
|| mime == accept
|| (accept.ends_with('/') && mime.starts_with(&accept))
|| (accept.ends_with("/*") && mime.starts_with(&accept[..accept.len() - 1]))
|| ext == accept
}
fn check_file(kind: InputKind, name: &str, accept: &[String], file: &Value) -> Result<(), PipelineError> {
if !is_file_ref(file) {
return Err(invalid(kind, name, format!("got {}", describe(file))));
}
if !accept.is_empty() && !accept.iter().any(|a| file_matches(file, a)) {
let got = format!(
"{} ({})",
file["filename"].as_str().unwrap_or("file"),
file["mime"].as_str().unwrap_or("unknown type")
);
return Err(invalid(
kind,
name,
format!("accepts {} — got {got}", accept.join(", ")),
));
}
Ok(())
}
pub fn check(kind: InputKind, config: &Config, envelope: &Value) -> Result<Value, PipelineError> {
let name = config.name.trim();
if name.is_empty() {
return Err(PipelineError::new(
"FW_NODE_INPUT_CONFIG",
format!("{}: the field name is required — `{} <name>`", kind.node_kind(), kind.short()),
));
}
let raw = sent_value(kind, name, envelope);
if is_absent(&raw) {
if !kind.is_file() {
if let Some(default) = config.default.as_ref().filter(|d| !d.is_null()) {
return check_value(kind, config, name, default.clone());
}
}
if config.optional {
return Ok(Value::Null);
}
return Err(missing(kind, name));
}
check_value(kind, config, name, raw)
}
fn sent_value(kind: InputKind, name: &str, envelope: &Value) -> Value {
let slot = envelope.get(kind.envelope_slot()).unwrap_or(&Value::Null);
resolve_path(slot, name).cloned().unwrap_or(Value::Null)
}
fn is_absent(raw: &Value) -> bool {
match raw {
Value::Null => true,
Value::String(s) => s.trim().is_empty(),
Value::Array(items) => items.is_empty(),
_ => false,
}
}
fn answer_at(name: &str, value: Value) -> Value {
let segments: Vec<&str> = name.split('.').map(str::trim).filter(|s| !s.is_empty()).collect();
segments.iter().rev().fold(value, |inner, segment| {
let mut map = serde_json::Map::new();
map.insert((*segment).to_string(), inner);
Value::Object(map)
})
}
fn check_value(kind: InputKind, config: &Config, name: &str, raw: Value) -> Result<Value, PipelineError> {
match kind {
InputKind::Text => {
let text = match raw {
Value::String(s) => s,
other => return Err(invalid(kind, name, format!("got {}", describe(&other)))),
};
if let Some(max) = config.max {
let len = text.chars().count();
if (len as f64) > max {
return Err(invalid(kind, name, format!("at most {max} characters, got {len}")));
}
}
Ok(Value::String(text))
}
InputKind::Number => {
let n = match &raw {
Value::Number(n) => n.as_f64(),
Value::String(s) => s.trim().parse::<f64>().ok(),
_ => None,
}
.ok_or_else(|| invalid(kind, name, format!("got {}", describe(&raw))))?;
if let Some(min) = config.min {
if n < min {
return Err(invalid(kind, name, format!("at least {min}, got {n}")));
}
}
if let Some(max) = config.max {
if n > max {
return Err(invalid(kind, name, format!("at most {max}, got {n}")));
}
}
Ok(serde_json::Number::from_f64(n)
.map(Value::Number)
.unwrap_or(raw))
}
InputKind::Boolean => {
let b = match &raw {
Value::Bool(b) => Some(*b),
Value::Number(n) => n.as_f64().map(|v| v != 0.0),
Value::String(s) => match s.trim().to_ascii_lowercase().as_str() {
"true" | "1" | "on" | "yes" => Some(true),
"false" | "0" | "off" | "no" => Some(false),
_ => None,
},
_ => None,
}
.ok_or_else(|| invalid(kind, name, format!("got {}", describe(&raw))))?;
Ok(Value::Bool(b))
}
InputKind::Json => match raw {
Value::String(s) => serde_json::from_str::<Value>(&s)
.map_err(|err| invalid(kind, name, format!("the string does not parse as JSON ({err})"))),
other => Ok(other),
},
InputKind::File | InputKind::Image | InputKind::Audio | InputKind::Video => {
let accept = config.effective_accept(kind);
let file = match raw {
Value::Array(mut items) if items.len() == 1 => items.remove(0),
Value::Array(items) => {
return Err(invalid(kind, name, format!("got {} files", items.len())));
}
other => other,
};
check_file(kind, name, &accept, &file)?;
Ok(file)
}
InputKind::Files => {
let accept = config.effective_accept(kind);
let items = match raw {
Value::Array(items) => items,
single if is_file_ref(&single) => vec![single],
other => return Err(invalid(kind, name, format!("got {}", describe(&other)))),
};
if let Some(max) = config.max {
if (items.len() as f64) > max {
return Err(invalid(kind, name, format!("at most {max} files, got {}", items.len())));
}
}
for item in &items {
check_file(kind, name, &accept, item)?;
}
Ok(Value::Array(items))
}
}
}
fn flag(flag: &str, key: &str, description: &str, kind: DslFlagKind) -> DslFlag {
DslFlag {
flag: flag.to_string(),
config_key: key.to_string(),
description: description.to_string(),
kind,
required: false,
..Default::default()
}
}
fn field(name: &str, label: &str, field_type: NodeFieldType, help: &str) -> NodeFieldDef {
NodeFieldDef {
name: name.to_string(),
label: label.to_string(),
field_type,
help: Some(help.to_string()),
..Default::default()
}
}
fn example(kind: InputKind) -> NodeExample {
match kind {
InputKind::Text => NodeExample::dsl("A caption the Run form asks for", r#"input.text prompt --label "Caption" --max 200"#)
.input(json!({ "manual": { "body": { "prompt": "A cat on a mat" } } }))
.output(json!({ "manual": { "body": { "prompt": "A cat on a mat" } }, "prompt": "A cat on a mat" }))
.note("Read from `$trigger.body.prompt`; the checked string is `input.prompt` next, `$nodes.<id>.prompt` anywhere later."),
InputKind::Number => NodeExample::dsl("A bounded count", "input.number count --min 1 --max 50 --default 10")
.input(json!({ "manual": { "body": {} } }))
.output(json!({ "manual": { "body": {} }, "count": 10.0 }))
.note("Nothing was sent, so the default stands: `input.count` and `$nodes.<id>.count` are 10."),
InputKind::Boolean => NodeExample::dsl("A switch", "input.boolean dry_run --optional")
.input(json!({ "manual": { "body": { "dry_run": "on" } } }))
.output(json!({ "manual": { "body": { "dry_run": "on" } }, "dry_run": true }))
.note("A form posts `on`; the node answers `dry_run: true`."),
InputKind::Json => NodeExample::dsl("A JSON document typed into the form", r#"input.json config --label "Settings""#)
.input(json!({ "manual": { "body": { "config": "{\"theme\":\"dark\"}" } } }))
.output(json!({ "manual": { "body": { "config": "{\"theme\":\"dark\"}" } }, "config": { "theme": "dark" } }))
.note("A JSON string in `body` is parsed; the node answers the object."),
InputKind::File => NodeExample::dsl("A spreadsheet or CSV", "input.file sheet --accept csv --accept xlsx")
.input(json!({ "webhook": { "body": {}, "files": { "sheet": { "__zf_type": "file_ref", "filename": "q3.xlsx", "mime": "application/vnd.openxmlformats-officedocument.spreadsheetml.sheet", "kind": "spreadsheet", "size": 18211 } } } }))
.output(json!({ "webhook": { "body": {}, "files": { "sheet": { "__zf_type": "file_ref", "filename": "q3.xlsx", "mime": "application/vnd.openxmlformats-officedocument.spreadsheetml.sheet", "kind": "spreadsheet", "size": 18211 } } }, "sheet": { "__zf_type": "file_ref", "filename": "q3.xlsx", "mime": "application/vnd.openxmlformats-officedocument.spreadsheetml.sheet", "kind": "spreadsheet", "size": 18211 } }))
.note("`--accept` takes FileRef kinds, mimes (`image/png`, `image/`) or extensions."),
InputKind::Files => NodeExample::dsl("Up to five attachments", "input.files attachments --accept pdf --accept image --max 5 --optional")
.note("Answers `attachments`: the array of FileRefs, `[]` when none were sent."),
InputKind::Image => NodeExample::dsl("A photo to process", "input.image photo")
.input(json!({ "manual": { "body": {}, "files": { "photo": { "__zf_type": "file_ref", "filename": "cat.png", "mime": "image/png", "kind": "image", "size": 4321 } } } }))
.output(json!({ "manual": { "body": {}, "files": { "photo": { "__zf_type": "file_ref", "filename": "cat.png", "mime": "image/png", "kind": "image", "size": 4321 } } }, "photo": { "__zf_type": "file_ref", "filename": "cat.png", "mime": "image/png", "kind": "image", "size": 4321 } }))
.note("`--accept` is preset to `image`; `--accept png --accept webp` narrows it. The next node takes `--from \"{{ input.photo }}\"`."),
InputKind::Audio => NodeExample::dsl("A recording", "input.audio clip --optional")
.note("`--accept` is preset to `audio`."),
InputKind::Video => NodeExample::dsl("A clip", "input.video clip --accept mp4,webm")
.note("`--accept` is preset to `video`; here narrowed to two containers."),
}
}
pub fn definition_for(kind: InputKind) -> NodeDefinition {
let short = kind.short();
let mut dsl_flags = vec![
DslFlag {
flag: "--name".to_string(),
config_key: "name".to_string(),
description: format!("The envelope field this node declares, and the key it answers — also the first bare token: `{short} <name>`."),
kind: DslFlagKind::Scalar,
required: true,
..Default::default()
},
flag("--label", "label", "Form label shown in the Run form. Default: the field name.", DslFlagKind::Scalar),
flag("--optional", "optional", "A missing value is allowed; the node's value is then null.", DslFlagKind::Bool),
];
let mut fields = vec![
field("name", "Field name", NodeFieldType::Text, "The envelope field this node declares: read at $trigger.body.<name> (or $trigger.files.<name> for file kinds), answered at <name>."),
field("label", "Label", NodeFieldType::Text, "Form label shown in the Run form. Defaults to the field name."),
field("optional", "Optional", NodeFieldType::Checkbox, "A missing value is allowed; the node's value is then null. Required by default."),
];
let mut properties = json!({
"name": { "type": "string", "description": "The envelope field this node declares." },
"label": { "type": "string", "description": "Form label; the field name when empty." },
"optional": { "type": "boolean", "description": "A missing value is allowed." },
});
if !kind.is_file() {
dsl_flags.push(flag("--default", "default", "Value used when the field is missing (text, number, boolean, json).", DslFlagKind::Scalar));
fields.push(field("default", "Default", NodeFieldType::Text, "Value used when the field is missing. Checked like a sent value."));
properties["default"] = json!({ "description": "Value used when the field is missing." });
}
if kind.is_file() {
let preset = kind.preset_accept().map(|k| format!(" Preset: `{k}`.")).unwrap_or_default();
dsl_flags.push(flag("--accept", "accept", &format!("FileRef kinds, mimes or extensions the file must match, one per flag, repeated (`--accept csv --accept xlsx`, `image/png`, `image/`).{preset}"), DslFlagKind::RepeatedList));
fields.push(field("accept", "Accept", NodeFieldType::Text, &format!("Comma-separated FileRef kinds, mimes or extensions the file must match.{preset}")));
properties["accept"] = json!({ "type": "array", "items": { "type": "string" }, "description": "What the file must match." });
}
if kind == InputKind::Number {
dsl_flags.push(flag("--min", "min", "Smallest value accepted.", DslFlagKind::Scalar));
fields.push(field("min", "Min", NodeFieldType::Number, "Smallest value accepted."));
properties["min"] = json!({ "type": "number", "description": "Smallest value accepted." });
}
let max_help = match kind {
InputKind::Text => Some("Longest string accepted, in characters."),
InputKind::Number => Some("Largest value accepted."),
InputKind::Files => Some("Most files accepted."),
_ => None,
};
if let Some(help) = max_help {
dsl_flags.push(flag("--max", "max", help, DslFlagKind::Scalar));
fields.push(field("max", "Max", NodeFieldType::Number, help));
properties["max"] = json!({ "type": "number", "description": help });
}
let value_word = match kind {
InputKind::Text => "a string",
InputKind::Number => "a number",
InputKind::Boolean => "a boolean",
InputKind::Json => "any JSON value",
InputKind::Files => "an array of FileRefs",
_ => "one FileRef",
};
let where_ = kind.envelope_slot();
let description = format!(
"Declares and checks one field of the trigger envelope: {value_word} at `$trigger.{where_}.<name>`. \
Required by default — a missing or invalid value refuses the run with `FW_NODE_INPUT_MISSING` / `FW_NODE_INPUT_INVALID`, \
naming the field and what was expected; `--optional` allows absence (the answer is then null). Answers one key, its `--name`: \
the next node reads `input.<name>`, any later one `$nodes.<id>.<name>`, and the rest of the payload is kept.{} \
The input nodes after a trigger are its declaration: the Studio builds the Run form from them and an agent reads them to know \
what to send. Works after `trigger.manual` and after `trigger.webhook`. Fetches nothing, stores nothing.{}",
if kind.is_file() {
""
} else {
" A field that was not sent and has `--default` answers the checked default, so every path (DSL/MCP execute, webhook, Run form) gives the same answer."
},
match kind {
InputKind::Json => " A JSON string in `body` is parsed.",
InputKind::Image | InputKind::Audio | InputKind::Video => " `--accept` is preset to the media kind.",
_ => "",
}
);
NodeDefinition {
kind: kind.node_kind().to_string(),
capabilities: vec![],
title: kind.title().to_string(),
description,
input_schema: json!({
"type": "object",
"description": format!("Any payload; this node reads `$trigger.{where_}.<name>`.")
}),
output_schema: json!({
"type": "object",
"description": "The payload plus the checked value at `--name` (null when optional and not sent)."
}),
input_pins: vec![INPUT_PIN_IN.to_string()],
output_pins: vec![OUTPUT_PIN_OUT.to_string()],
script_available: false,
script_bridge: None,
config_schema: json!({ "type": "object", "required": ["name"], "properties": properties }),
dsl_flags,
fields,
layout: vec![],
ai_tool: Default::default(),
examples: vec![example(kind)],
positional: Some("name".to_string()),
ui_category: "input".to_string(),
..Default::default()
}
}
const EMPTY_ENVELOPE_TRIGGERS: [&str; 1] = ["trigger.schedule"];
fn satisfied_by_an_empty_envelope(kind: InputKind, config: &Config) -> bool {
if config.optional {
return true;
}
!kind.is_file() && config.default.as_ref().is_some_and(|d| !d.is_null())
}
pub fn ensure_inputs_reachable_from_empty_triggers(
graph: &PipelineGraph,
) -> Result<(), PipelineError> {
for trigger in graph
.nodes
.iter()
.filter(|node| EMPTY_ENVELOPE_TRIGGERS.contains(&node.kind.as_str()))
{
let trigger_kind = trigger.kind.as_str();
let mut seen: std::collections::HashSet<&str> = std::collections::HashSet::new();
let mut queue = std::collections::VecDeque::from([trigger.id.as_str()]);
while let Some(current) = queue.pop_front() {
if !seen.insert(current) {
continue;
}
for edge in graph.edges.iter().filter(|edge| edge.from_node == current) {
queue.push_back(edge.to_node.as_str());
}
let Some(node) = graph.nodes.iter().find(|node| node.id == current) else {
continue;
};
let Some(kind) = InputKind::from_kind(&node.kind) else {
continue;
};
let config = serde_json::from_value::<Config>(node.config.clone())
.map_err(|err| PipelineError::new("FW_NODE_INPUT_CONFIG", err.to_string()))?;
if satisfied_by_an_empty_envelope(kind, &config) {
continue;
}
let fix = if kind.is_file() { "--optional" } else { "--default or --optional" };
return Err(PipelineError::new(
"FW_NODE_INPUT_UNREACHABLE",
format!(
"input '{}' (node {}) is required but {trigger_kind} delivers no {} — add {fix}",
config.name.trim(),
node.id,
kind.envelope_slot()
),
));
}
}
Ok(())
}
pub struct Node {
kind: InputKind,
config: Config,
}
impl Node {
pub fn new(kind: InputKind, config: Config) -> Self {
Self { kind, config }
}
pub fn for_kind(kind: &str, config: &Value) -> Option<Result<Self, PipelineError>> {
let input_kind = InputKind::from_kind(kind)?;
Some(
serde_json::from_value::<Config>(config.clone())
.map(|config| Self::new(input_kind, config))
.map_err(|err| PipelineError::new("FW_NODE_INPUT_CONFIG", err.to_string())),
)
}
}
#[async_trait]
impl NodeHandler for Node {
fn kind(&self) -> &'static str {
self.kind.node_kind()
}
fn input_pins(&self) -> &'static [&'static str] {
&[INPUT_PIN_IN]
}
fn output_pins(&self) -> &'static [&'static str] {
&[OUTPUT_PIN_OUT]
}
async fn execute_async(
&self,
input: NodeExecutionInput,
) -> Result<NodeExecutionOutput, PipelineError> {
let envelope = crate::pipeline::nodes::basic::trigger::envelope(&input.metadata);
let filled = !self.kind.is_file()
&& self.config.default.as_ref().is_some_and(|d| !d.is_null())
&& is_absent(&sent_value(self.kind, self.config.name.trim(), envelope));
let value = check(self.kind, &self.config, envelope)?;
let name = self.config.name.trim();
let mut trace = format!(
"node_kind={} field={} value={}",
self.kind.node_kind(),
name,
describe(&value)
);
if filled {
trace.push_str(" default=filled");
}
Ok(NodeExecutionOutput {
output_pins: vec![OUTPUT_PIN_OUT.to_string()],
payload: with_answer(&input.payload, answer_at(name, value)),
trace: vec![trace],
})
}
}
#[cfg(test)]
mod tests {
use super::*;
fn cfg(name: &str) -> Config {
Config { name: name.to_string(), ..Default::default() }
}
fn file_ref(filename: &str, mime: &str, kind: &str) -> Value {
json!({
"__zf_type": "file_ref", "backend": "zebfs", "ref": format!("tmp/runs/r/files/{filename}"),
"filename": filename, "mime": mime, "kind": kind, "size": 12,
"sha256": "sha256:e3b0c44298fc1c14a1b2c3d4e5f60718293a4b5c6d7e8f90a1b2c3d4e5f60718",
"lifecycle": "temporary", "origin": "manual", "trust": "user"
})
}
#[test]
fn a_required_field_that_was_not_sent_is_missing_and_names_the_field() {
let err = check(InputKind::Text, &cfg("prompt"), &json!({ "body": {} })).unwrap_err();
assert_eq!(err.code, "FW_NODE_INPUT_MISSING");
assert!(err.message.contains("'prompt'") && err.message.contains("body.prompt"), "{}", err.message);
let err = check(InputKind::Text, &cfg("prompt"), &json!({ "body": { "prompt": " " } })).unwrap_err();
assert_eq!(err.code, "FW_NODE_INPUT_MISSING");
}
#[test]
fn optional_and_default_stand_in_for_an_absent_value() {
let optional = Config { optional: true, ..cfg("note") };
assert_eq!(check(InputKind::Text, &optional, &json!({})).unwrap(), Value::Null);
let with_default = Config { default: Some(json!(10)), ..cfg("count") };
assert_eq!(check(InputKind::Number, &with_default, &json!({ "body": {} })).unwrap(), json!(10.0));
}
#[test]
fn text_number_boolean_and_json_coerce_what_a_form_posts() {
let long = Config { max: Some(3.0), ..cfg("t") };
assert_eq!(check(InputKind::Text, &long, &json!({ "body": { "t": "abc" } })).unwrap(), json!("abc"));
let err = check(InputKind::Text, &long, &json!({ "body": { "t": "abcd" } })).unwrap_err();
assert_eq!(err.code, "FW_NODE_INPUT_INVALID");
let bounded = Config { min: Some(1.0), max: Some(5.0), ..cfg("n") };
assert_eq!(check(InputKind::Number, &bounded, &json!({ "body": { "n": "4" } })).unwrap(), json!(4.0));
assert_eq!(check(InputKind::Number, &bounded, &json!({ "body": { "n": 9 } })).unwrap_err().code, "FW_NODE_INPUT_INVALID");
assert_eq!(check(InputKind::Number, &bounded, &json!({ "body": { "n": "four" } })).unwrap_err().code, "FW_NODE_INPUT_INVALID");
assert_eq!(check(InputKind::Boolean, &cfg("b"), &json!({ "body": { "b": "on" } })).unwrap(), json!(true));
assert_eq!(check(InputKind::Boolean, &cfg("b"), &json!({ "body": { "b": false } })).unwrap(), json!(false));
assert_eq!(check(InputKind::Boolean, &cfg("b"), &json!({ "body": { "b": "maybe" } })).unwrap_err().code, "FW_NODE_INPUT_INVALID");
assert_eq!(check(InputKind::Json, &cfg("j"), &json!({ "body": { "j": "{\"a\":1}" } })).unwrap(), json!({ "a": 1 }));
assert_eq!(check(InputKind::Json, &cfg("j"), &json!({ "body": { "j": { "a": 1 } } })).unwrap(), json!({ "a": 1 }));
assert_eq!(check(InputKind::Json, &cfg("j"), &json!({ "body": { "j": "{nope" } })).unwrap_err().code, "FW_NODE_INPUT_INVALID");
}
#[test]
fn a_file_is_read_from_files_and_checked_against_accept() {
let photo = file_ref("cat.png", "image/png", "image");
let payload = json!({ "body": {}, "files": { "photo": photo.clone() } });
assert_eq!(check(InputKind::Image, &cfg("photo"), &payload).unwrap(), photo);
let pdf = json!({ "files": { "photo": file_ref("cv.pdf", "application/pdf", "pdf") } });
let err = check(InputKind::Image, &cfg("photo"), &pdf).unwrap_err();
assert_eq!(err.code, "FW_NODE_INPUT_INVALID");
assert!(err.message.contains("accepts image"), "{}", err.message);
let sheet = Config { accept: vec!["csv".into(), "xlsx".into()], ..cfg("sheet") };
let xlsx = file_ref("q3.xlsx", "application/vnd.openxmlformats-officedocument.spreadsheetml.sheet", "spreadsheet");
assert!(check(InputKind::File, &sheet, &json!({ "files": { "sheet": xlsx } })).is_ok());
let family = Config { accept: vec!["image/".into()], ..cfg("f") };
assert!(check(InputKind::File, &family, &json!({ "files": { "f": file_ref("a.webp", "image/webp", "image") } })).is_ok());
let err = check(InputKind::File, &cfg("f"), &json!({ "files": { "f": "uploads/x.png" } })).unwrap_err();
assert_eq!(err.code, "FW_NODE_INPUT_INVALID");
}
#[test]
fn files_takes_a_list_bounded_by_max() {
let two = json!({ "files": { "docs": [file_ref("a.pdf", "application/pdf", "pdf"), file_ref("b.pdf", "application/pdf", "pdf")] } });
let capped = Config { max: Some(1.0), ..cfg("docs") };
assert_eq!(check(InputKind::Files, &capped, &two).unwrap_err().code, "FW_NODE_INPUT_INVALID");
let value = check(InputKind::Files, &cfg("docs"), &two).unwrap();
assert_eq!(value.as_array().map(Vec::len), Some(2));
let one = json!({ "files": { "docs": file_ref("a.pdf", "application/pdf", "pdf") } });
assert_eq!(check(InputKind::Files, &cfg("docs"), &one).unwrap().as_array().map(Vec::len), Some(1));
assert_eq!(check(InputKind::Files, &cfg("docs"), &json!({ "files": {} })).unwrap_err().code, "FW_NODE_INPUT_MISSING");
}
async fn run(node: &Node, envelope: Value) -> NodeExecutionOutput {
node.execute_async(NodeExecutionInput {
node_id: "n1".into(),
input_pin: "in".into(),
payload: json!({ "manual": envelope.clone() }),
metadata: json!({ "trigger": envelope }),
bus: None,
})
.await
.unwrap()
}
#[tokio::test]
async fn the_node_reads_the_envelope_and_answers_at_its_name() {
let envelope = json!({ "body": { "prompt": "hi", "other": 1 }, "files": {} });
let out = run(&Node::new(InputKind::Text, cfg("prompt")), envelope.clone()).await;
assert_eq!(out.payload, json!({ "manual": envelope, "prompt": "hi" }), "one key added, the rest kept");
}
#[tokio::test]
async fn the_envelope_is_read_from_trigger_not_from_the_payload() {
let node = Node::new(InputKind::Image, cfg("photo"));
let photo = file_ref("cat.png", "image/png", "image");
let out = node
.execute_async(NodeExecutionInput {
node_id: "n2".into(),
input_pin: "in".into(),
payload: json!({ "result": 1 }),
metadata: json!({ "trigger": { "body": {}, "files": { "photo": photo.clone() } } }),
bus: None,
})
.await
.unwrap();
assert_eq!(out.payload, json!({ "result": 1, "photo": photo }));
}
#[tokio::test]
async fn a_default_answers_for_the_missing_field_and_a_sent_value_wins() {
let node = Node::new(InputKind::Text, Config { default: Some(json!("world")), optional: true, ..cfg("who") });
let out = run(&node, json!({ "body": {} })).await;
assert_eq!(out.payload["who"], json!("world"));
assert!(out.trace[0].ends_with("default=filled"), "{}", out.trace[0]);
assert_eq!(run(&node, json!({})).await.payload["who"], json!("world"));
let out = run(&node, json!({ "body": { "who": "x", "other": 1 }, "files": {} })).await;
assert_eq!(out.payload["who"], json!("x"));
assert!(!out.trace[0].contains("default=filled"));
let node = Node::new(InputKind::Number, Config { default: Some(json!("10")), ..cfg("page.size") });
let out = run(&node, json!({ "body": { "page": {} } })).await;
assert_eq!(out.payload["page"], json!({ "size": 10.0 }));
}
#[tokio::test]
async fn an_optional_field_without_a_default_answers_null() {
let node = Node::new(InputKind::Text, Config { optional: true, ..cfg("who") });
let out = run(&node, json!({ "body": {} })).await;
assert_eq!(out.payload, json!({ "manual": { "body": {} }, "who": null }));
}
#[test]
fn every_kind_has_a_definition_with_the_positional_name() {
for kind in InputKind::ALL {
let def = definition_for(kind);
assert_eq!(def.positional.as_deref(), Some("name"), "{}", def.kind);
assert_eq!(def.ui_category, "input");
assert!(def.dsl_flags.iter().any(|f| f.flag == "--name"));
assert_eq!(def.dsl_flags.iter().any(|f| f.flag == "--accept"), kind.is_file(), "{}", def.kind);
assert_eq!(def.dsl_flags.iter().any(|f| f.flag == "--default"), !kind.is_file(), "{}", def.kind);
assert_eq!(InputKind::from_kind(&def.kind), Some(kind));
}
}
#[test]
fn accept_and_bounds_read_the_dialog_spelling_too() {
let config: Config = serde_json::from_value(json!({ "name": "f", "accept": "CSV, xlsx", "max": "3" })).unwrap();
assert_eq!(config.accept, vec!["csv", "xlsx"]);
assert_eq!(config.max, Some(3.0));
let config: Config = serde_json::from_value(json!({ "name": "f", "accept": ["pdf"], "max": 2 })).unwrap();
assert_eq!(config.accept, vec!["pdf"]);
assert_eq!(config.max, Some(2.0));
}
fn graph(body: &str) -> PipelineGraph {
crate::platform::shell::parser::build_pipeline_graph("schedule-inputs", body).expect("graph")
}
#[test]
fn a_required_input_after_a_schedule_is_refused_naming_the_fix() {
let err = ensure_inputs_reachable_from_empty_triggers(&graph(
"[a] trigger.schedule --cron \"0 * * * *\"\n[b] input.text prompt\n[a] -> [b]\n",
))
.unwrap_err();
assert_eq!(err.code, "FW_NODE_INPUT_UNREACHABLE");
assert_eq!(
err.message,
"input 'prompt' (node b) is required but trigger.schedule delivers no body — add --default or --optional"
);
let err = ensure_inputs_reachable_from_empty_triggers(&graph(
"[a] trigger.schedule --cron \"0 * * * *\"\n[b] input.text note --optional\n[c] input.image photo\n[a] -> [b]\n[b] -> [c]\n",
))
.unwrap_err();
assert_eq!(err.code, "FW_NODE_INPUT_UNREACHABLE");
assert_eq!(
err.message,
"input 'photo' (node c) is required but trigger.schedule delivers no files — add --optional"
);
}
#[test]
fn a_default_or_optional_input_after_a_schedule_passes() {
ensure_inputs_reachable_from_empty_triggers(&graph(
"[a] trigger.schedule --cron \"0 * * * *\"\n[b] input.text prompt --default hi\n[a] -> [b]\n",
))
.expect("--default satisfies the tick");
ensure_inputs_reachable_from_empty_triggers(&graph(
"[a] trigger.schedule --cron \"0 * * * *\"\n[b] input.text prompt --optional\n[a] -> [b]\n",
))
.expect("--optional satisfies the tick");
ensure_inputs_reachable_from_empty_triggers(&graph(
"[a] trigger.schedule --cron \"0 * * * *\"\n[b] input.file doc --optional\n[a] -> [b]\n",
))
.expect("an optional file satisfies the tick");
}
#[test]
fn each_trigger_is_checked_on_its_own() {
let err = ensure_inputs_reachable_from_empty_triggers(&graph(
"[a] trigger.schedule --cron \"0 * * * *\"\n[m] trigger.manual\n[b] input.text prompt\n[a] -> [b]\n[m] -> [b]\n",
))
.unwrap_err();
assert_eq!(err.code, "FW_NODE_INPUT_UNREACHABLE");
ensure_inputs_reachable_from_empty_triggers(&graph(
"[a] trigger.schedule --cron \"0 * * * *\"\n[m] trigger.manual\n[b] input.text prompt\n[s] javascript.script.run -- \"return input;\"\n[a] -> [s]\n[m] -> [b]\n",
))
.expect("the schedule reaches no input");
ensure_inputs_reachable_from_empty_triggers(&graph(
"[w] trigger.webhook --route /x --method POST\n[b] input.text prompt\n[w] -> [b]\n",
))
.expect("a webhook's missing field is the run's refusal, not activation's");
}
}