From 7e28bdd53b1b2a14357fb24f43a02dea5b0e5fcc Mon Sep 17 00:00:00 2001 From: Steven Enamakel Date: Fri, 31 Jul 2026 13:29:11 +0300 Subject: [PATCH 1/5] feat(model): declare typed workflow inputs A WorkflowGraph now carries `inputs`: named, typed, optionally-defaulted parameters that form the workflow's public signature, independent of the trigger kind. `resolve_inputs` validates supplied values against the declarations, and `validate_all` rejects declarations an expression could not address or that contradict themselves. --- src/error.rs | 120 ++++++++++- src/model/inputs.rs | 492 ++++++++++++++++++++++++++++++++++++++++++++ src/model/mod.rs | 46 +++++ src/validate.rs | 148 ++++++++++++- 4 files changed, 792 insertions(+), 14 deletions(-) create mode 100644 src/model/inputs.rs diff --git a/src/error.rs b/src/error.rs index 2ae752b..c704adf 100644 --- a/src/error.rs +++ b/src/error.rs @@ -99,6 +99,35 @@ pub enum ValidationError { /// The edge's actual (invalid) `from_port` value. from_port: String, }, + + /// Two declared inputs share the same name, so one would shadow the other + /// in the `inputs` expression scope. + #[error("duplicate workflow input name: {0}")] + DuplicateInputName(String), + + /// A declared input's name is not a plain identifier, so `=inputs.` + /// could not address it without jq quoting. + #[error( + "invalid workflow input name {0:?} — names must match [A-Za-z_][A-Za-z0-9_]* so \ + `=inputs.` can address them" + )] + InvalidInputName(String), + + /// A declared input's `default` does not satisfy its own declared type, so + /// an omitted value would inject a wrongly-typed one. + #[error("workflow input {name:?} has a default that is not of its declared type {expected}")] + InputDefaultTypeMismatch { + /// The offending input's name. + name: String, + /// The declared type's wire name. + expected: &'static str, + }, + + /// A declared input is both `required` and has a `default`. The default + /// always supplies a value, so the requirement could never fire — one of + /// the two is a mistake. + #[error("workflow input {0:?} is both required and has a default; a default makes it optional")] + RequiredInputWithDefault(String), } impl ValidationError { @@ -121,15 +150,21 @@ impl ValidationError { Self::InvalidOnError { .. } => "invalid_on_error", Self::SchemaVersionTooNew { .. } => "schema_version_too_new", Self::InvalidConditionRouting { .. } => "invalid_condition_routing", + Self::DuplicateInputName(_) => "duplicate_input_name", + Self::InvalidInputName(_) => "invalid_input_name", + Self::InputDefaultTypeMismatch { .. } => "input_default_type_mismatch", + Self::RequiredInputWithDefault(_) => "required_input_with_default", } } /// The node id this error is anchored to, when it is node-specific. /// /// Returns `None` for graph-wide errors (`MissingTrigger`, - /// `SchemaVersionTooNew`) and for `MultipleTriggers` (which carries many - /// ids in its payload rather than a single anchor). Lets a host attach the - /// error to the right node in a structured validation report. + /// `SchemaVersionTooNew`), for `MultipleTriggers` (which carries many ids in + /// its payload rather than a single anchor), and for the declared-input + /// errors (anchored to an input, not a node — see [`Self::input_name`]). + /// Lets a host attach the error to the right node in a structured + /// validation report. pub fn node_id(&self) -> Option<&str> { match self { Self::UnknownNode(id) @@ -140,9 +175,28 @@ impl ValidationError { | Self::InvalidOnError { node, .. } | Self::InvalidConditionRouting { node, .. } => Some(node), Self::DuplicateEdge { from_node, .. } => Some(from_node), - Self::MissingTrigger | Self::MultipleTriggers(_) | Self::SchemaVersionTooNew { .. } => { - None - } + Self::MissingTrigger + | Self::MultipleTriggers(_) + | Self::SchemaVersionTooNew { .. } + | Self::DuplicateInputName(_) + | Self::InvalidInputName(_) + | Self::InputDefaultTypeMismatch { .. } + | Self::RequiredInputWithDefault(_) => None, + } + } + + /// The declared input this error is anchored to, when it is input-specific. + /// + /// The counterpart to [`Self::node_id`]: lets a host attach the error to the + /// right field of an inputs editor. Returns `None` for every node-anchored + /// and graph-wide error. + pub fn input_name(&self) -> Option<&str> { + match self { + Self::DuplicateInputName(name) + | Self::InvalidInputName(name) + | Self::RequiredInputWithDefault(name) => Some(name), + Self::InputDefaultTypeMismatch { name, .. } => Some(name), + _ => None, } } } @@ -154,6 +208,13 @@ pub enum EngineError { #[error("validation failed: {0}")] Validation(#[from] ValidationError), + /// The values supplied for the workflow's declared inputs were rejected. + /// + /// Raised before any node executes and before the run is recorded, so a + /// caller that gets this can be certain nothing ran. + #[error("input error: {0}")] + Input(#[from] crate::model::InputError), + /// A feature required by the graph is not yet implemented in this stage. #[error("not yet implemented: {0}")] Unimplemented(&'static str), @@ -243,6 +304,40 @@ mod tests { ); } + #[test] + fn declared_input_validation_error_display_and_anchors() { + let dup = ValidationError::DuplicateInputName("repo".to_string()); + assert_eq!(dup.to_string(), "duplicate workflow input name: repo"); + assert_eq!(dup.code(), "duplicate_input_name"); + assert_eq!(dup.input_name(), Some("repo")); + assert_eq!(dup.node_id(), None); + + assert_eq!( + ValidationError::InvalidInputName("repo-url".to_string()).to_string(), + "invalid workflow input name \"repo-url\" — names must match \ + [A-Za-z_][A-Za-z0-9_]* so `=inputs.` can address them" + ); + assert_eq!( + ValidationError::InputDefaultTypeMismatch { + name: "depth".to_string(), + expected: "number", + } + .to_string(), + "workflow input \"depth\" has a default that is not of its declared type number" + ); + assert_eq!( + ValidationError::RequiredInputWithDefault("repo".to_string()).to_string(), + "workflow input \"repo\" is both required and has a default; \ + a default makes it optional" + ); + + // Node-anchored errors are not input-anchored, and vice versa. + assert_eq!( + ValidationError::UnknownNode("ghost".to_string()).input_name(), + None + ); + } + #[test] fn engine_error_display() { assert_eq!( @@ -257,6 +352,19 @@ mod tests { EngineError::Validation(ValidationError::MissingTrigger).to_string(), "validation failed: workflow has no trigger node" ); + assert_eq!( + EngineError::Input(crate::model::InputError::Missing("repo".to_string())).to_string(), + "input error: workflow input \"repo\" is required but was not supplied" + ); + } + + #[test] + fn input_error_lifts_into_engine_error() { + let engine: EngineError = crate::model::InputError::Unknown("reop".to_string()).into(); + match engine { + EngineError::Input(inner) => assert_eq!(inner.input_name(), "reop"), + other => panic!("expected lifted input error, got {other:?}"), + } } #[test] diff --git a/src/model/inputs.rs b/src/model/inputs.rs new file mode 100644 index 0000000..ab56e29 --- /dev/null +++ b/src/model/inputs.rs @@ -0,0 +1,492 @@ +//! Declared workflow inputs: the typed, caller-supplied parameters of a run. +//! +//! A [`WorkflowGraph`](crate::model::WorkflowGraph) declares zero or more +//! [`WorkflowInput`]s. They are the workflow's *public signature* — what a +//! caller must (or may) provide to run it — and are deliberately independent of +//! the trigger kind, so a manually run graph, a scheduled one, and one invoked +//! as a `sub_workflow` all expose the same parameters. +//! +//! Declared inputs are distinct from the free-form trigger payload: +//! +//! | | trigger payload (`run.trigger`) | declared inputs (`inputs`) | +//! |---|---|---| +//! | shape | whatever fired the run (webhook body, chat message, …) | named, typed, validated | +//! | discoverable | no | yes — from the graph | +//! | addressed as | `=run.trigger.` | `=inputs.` | +//! +//! Supplied values are checked by [`resolve_inputs`] **before** a run starts, so +//! a missing required parameter fails loudly instead of surfacing as a `null` +//! deep inside some node's configuration. +//! +//! Inputs are not a secret channel. Credentials reach a workflow through the +//! opaque connection reference the host resolves (see [`crate::caps`]); an input +//! is ordinary run data and is journalled and observable like any other. + +use serde::{Deserialize, Serialize}; +use serde_json::{Map, Value}; +use thiserror::Error; + +/// The declared type of a [`WorkflowInput`]. +/// +/// Type checking is deliberately shallow: it catches a caller passing a string +/// where a number was declared, and nothing more. Anything with real structure +/// is declared [`Json`](InputType::Json) and validated by the workflow itself. +#[derive(Debug, Clone, Copy, PartialEq, Eq, Default, Serialize, Deserialize)] +#[serde(rename_all = "snake_case")] +pub enum InputType { + /// A JSON string. The default when `type` is omitted. + #[default] + String, + /// A JSON number (integer or float). + Number, + /// A JSON boolean. + Boolean, + /// Any JSON value — object, array, or scalar. Accepts everything. + Json, +} + +impl InputType { + /// The `snake_case` wire name of this type, as it appears in JSON. + /// + /// ``` + /// use tinyflows::model::InputType; + /// + /// assert_eq!(InputType::Boolean.as_str(), "boolean"); + /// ``` + #[must_use] + pub fn as_str(self) -> &'static str { + match self { + Self::String => "string", + Self::Number => "number", + Self::Boolean => "boolean", + Self::Json => "json", + } + } + + /// Whether `value` satisfies this declared type. + /// + /// [`Json`](InputType::Json) accepts any value, including `null`. The scalar + /// types accept only their own JSON type — no coercion, so `"3"` is not a + /// [`Number`](InputType::Number). Hosts that read values from a + /// string-shaped surface (a CLI flag, a text prompt) are expected to coerce + /// before calling; see the `medulla` CLI's `--set` handling for a reference. + /// + /// ``` + /// use tinyflows::model::InputType; + /// use serde_json::json; + /// + /// assert!(InputType::Number.accepts(&json!(3))); + /// assert!(!InputType::Number.accepts(&json!("3"))); + /// assert!(InputType::Json.accepts(&json!({"any": "shape"}))); + /// ``` + #[must_use] + pub fn accepts(self, value: &Value) -> bool { + match self { + Self::String => value.is_string(), + Self::Number => value.is_number(), + Self::Boolean => value.is_boolean(), + Self::Json => true, + } + } +} + +/// One declared parameter of a workflow. +/// +/// ``` +/// use tinyflows::model::{InputType, WorkflowInput}; +/// +/// let declared: WorkflowInput = serde_json::from_str( +/// r#"{"name":"repo","type":"string","required":true,"description":"Repo to review"}"#, +/// ) +/// .unwrap(); +/// assert_eq!(declared.name, "repo"); +/// assert_eq!(declared.ty, InputType::String); +/// assert!(declared.required); +/// +/// // `type` defaults to `string`, and an input is optional unless declared otherwise. +/// let minimal: WorkflowInput = serde_json::from_str(r#"{"name":"note"}"#).unwrap(); +/// assert_eq!(minimal.ty, InputType::String); +/// assert!(!minimal.required); +/// ``` +#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)] +pub struct WorkflowInput { + /// The parameter's name, and the key it is addressed by in expressions + /// (`=inputs.`). Must be a plain identifier — see + /// [`is_valid_input_name`]. + pub name: String, + /// The declared JSON type. Defaults to [`InputType::String`]. + #[serde(rename = "type", default)] + pub ty: InputType, + /// Human-readable explanation, shown by authoring and run surfaces (the + /// hint line of a prompt, the description of a generated tool parameter). + #[serde(default)] + pub description: Option, + /// Whether a caller must supply this input. A required input with no + /// supplied value fails the run before it starts. + #[serde(default)] + pub required: bool, + /// Value used when the caller supplies none. Mutually exclusive with + /// `required` — a default means there is always a value. + #[serde(default)] + pub default: Option, +} + +impl WorkflowInput { + /// A minimal optional input of the given name and type. + #[must_use] + pub fn new(name: impl Into, ty: InputType) -> Self { + Self { + name: name.into(), + ty, + description: None, + required: false, + default: None, + } + } + + /// Marks this input as required, so a run without it fails to start. + #[must_use] + pub fn required(mut self) -> Self { + self.required = true; + self + } + + /// Sets the value used when the caller supplies none. + #[must_use] + pub fn with_default(mut self, default: Value) -> Self { + self.default = Some(default); + self + } + + /// Sets the human-readable description. + #[must_use] + pub fn with_description(mut self, description: impl Into) -> Self { + self.description = Some(description.into()); + self + } +} + +/// Whether `name` is usable as an input name. +/// +/// Names must match `[A-Za-z_][A-Za-z0-9_]*` so that `=inputs.` resolves +/// through the fast dotted-path walk in [`crate::expr`] without jq quoting. A +/// name that needed escaping would work in one expression form and silently +/// misbehave in the other, so it is rejected at validation time instead. +/// +/// ``` +/// use tinyflows::model::is_valid_input_name; +/// +/// assert!(is_valid_input_name("repo_url")); +/// assert!(!is_valid_input_name("repo-url")); // would need jq quoting +/// assert!(!is_valid_input_name("2fa")); // leading digit +/// assert!(!is_valid_input_name("")); +/// ``` +#[must_use] +pub fn is_valid_input_name(name: &str) -> bool { + let mut chars = name.chars(); + match chars.next() { + Some(c) if c.is_ascii_alphabetic() || c == '_' => {} + _ => return false, + } + chars.all(|c| c.is_ascii_alphanumeric() || c == '_') +} + +/// Why a set of supplied input values was rejected. +/// +/// Every variant names the offending input, so a host can attach the failure to +/// the right form field rather than showing a whole-run error. +#[derive(Debug, Clone, PartialEq, Eq, Error)] +pub enum InputError { + /// A required input was not supplied and has no default. + #[error("workflow input {0:?} is required but was not supplied")] + Missing(String), + + /// A supplied value's JSON type does not match the declared type. + #[error("workflow input {name:?} expects type {expected} but received {found}")] + TypeMismatch { + /// The declared input's name. + name: String, + /// The declared type's wire name. + expected: &'static str, + /// The supplied value's JSON type name. + found: &'static str, + }, + + /// A value was supplied under a name the workflow does not declare. + /// + /// Rejected rather than ignored: a silently dropped value looks identical to + /// a workflow that read it and did nothing. Callers with genuinely + /// free-form data should use the trigger payload instead. + #[error("workflow does not declare an input named {0:?}")] + Unknown(String), +} + +impl InputError { + /// A stable, machine-readable code for this variant, for hosts that surface + /// structured errors rather than prose. + #[must_use] + pub fn code(&self) -> &'static str { + match self { + Self::Missing(_) => "input_missing", + Self::TypeMismatch { .. } => "input_type_mismatch", + Self::Unknown(_) => "input_unknown", + } + } + + /// The name of the input this error is about. + #[must_use] + pub fn input_name(&self) -> &str { + match self { + Self::Missing(name) | Self::Unknown(name) => name, + Self::TypeMismatch { name, .. } => name, + } + } +} + +/// The JSON type name of `value`, for [`InputError::TypeMismatch`] reporting. +fn json_type_name(value: &Value) -> &'static str { + match value { + Value::Null => "null", + Value::Bool(_) => "boolean", + Value::Number(_) => "number", + Value::String(_) => "string", + Value::Array(_) => "array", + Value::Object(_) => "object", + } +} + +/// Validates `supplied` against `declared` and returns the resolved values. +/// +/// The returned map has **exactly one entry per declared input** — callers can +/// index it without checking for absence. Resolution per input: +/// +/// - supplied → type-checked, then used as-is; +/// - absent with a `default` → the default; +/// - absent and `required` → [`InputError::Missing`]; +/// - absent, optional, no default → [`Value::Null`]. +/// +/// A supplied key with no matching declaration is [`InputError::Unknown`]. +/// +/// # Errors +/// Returns the first problem found, scanning declarations in order and then +/// checking for undeclared keys. +/// +/// ``` +/// use tinyflows::model::{resolve_inputs, InputError, InputType, WorkflowInput}; +/// use serde_json::{json, Map}; +/// +/// let declared = vec![ +/// WorkflowInput::new("repo", InputType::String).required(), +/// WorkflowInput::new("depth", InputType::Number).with_default(json!(3)), +/// WorkflowInput::new("note", InputType::String), +/// ]; +/// +/// let mut supplied = Map::new(); +/// supplied.insert("repo".into(), json!("acme/api")); +/// +/// let resolved = resolve_inputs(&declared, &supplied).unwrap(); +/// assert_eq!(resolved["repo"], json!("acme/api")); +/// assert_eq!(resolved["depth"], json!(3)); // default applied +/// assert_eq!(resolved["note"], json!(null)); // optional, no default +/// +/// // A required input with no value fails before the run starts. +/// let err = resolve_inputs(&declared, &Map::new()).unwrap_err(); +/// assert_eq!(err, InputError::Missing("repo".into())); +/// ``` +pub fn resolve_inputs( + declared: &[WorkflowInput], + supplied: &Map, +) -> std::result::Result, InputError> { + let mut resolved = Map::new(); + + for input in declared { + let value = match supplied.get(&input.name) { + Some(value) => { + if !input.ty.accepts(value) { + return Err(InputError::TypeMismatch { + name: input.name.clone(), + expected: input.ty.as_str(), + found: json_type_name(value), + }); + } + value.clone() + } + None => match &input.default { + Some(default) => default.clone(), + None if input.required => return Err(InputError::Missing(input.name.clone())), + None => Value::Null, + }, + }; + resolved.insert(input.name.clone(), value); + } + + // Undeclared keys are checked last so a caller fixing a real declaration + // problem is not first told about a typo in an unrelated key. + for name in supplied.keys() { + if !resolved.contains_key(name) { + return Err(InputError::Unknown(name.clone())); + } + } + + Ok(resolved) +} + +#[cfg(test)] +mod tests { + use super::*; + use serde_json::json; + + fn supplied(pairs: &[(&str, Value)]) -> Map { + pairs + .iter() + .map(|(k, v)| ((*k).to_string(), v.clone())) + .collect() + } + + #[test] + fn input_type_defaults_to_string_and_round_trips() { + let input: WorkflowInput = serde_json::from_str(r#"{"name":"note"}"#).unwrap(); + assert_eq!(input.ty, InputType::String); + assert!(!input.required); + assert_eq!(input.default, None); + assert_eq!(input.description, None); + + let json = serde_json::to_value(&input).unwrap(); + assert_eq!(json["type"], "string"); + let back: WorkflowInput = serde_json::from_value(json).unwrap(); + assert_eq!(back, input); + } + + #[test] + fn input_type_accepts_matching_json_only() { + assert!(InputType::String.accepts(&json!("x"))); + assert!(!InputType::String.accepts(&json!(1))); + assert!(InputType::Number.accepts(&json!(1.5))); + assert!(!InputType::Number.accepts(&json!("1.5"))); + assert!(InputType::Boolean.accepts(&json!(true))); + assert!(!InputType::Boolean.accepts(&json!("true"))); + for value in [json!(null), json!([1]), json!({"a": 1}), json!("s")] { + assert!(InputType::Json.accepts(&value), "json rejected {value}"); + } + } + + #[test] + fn valid_input_names() { + for name in ["a", "_", "repo", "repo_url", "_x1", "A9"] { + assert!(is_valid_input_name(name), "{name} should be valid"); + } + for name in ["", "1a", "repo-url", "repo.url", "repo url", "café"] { + assert!(!is_valid_input_name(name), "{name} should be invalid"); + } + } + + #[test] + fn resolves_supplied_default_and_null() { + let declared = vec![ + WorkflowInput::new("repo", InputType::String).required(), + WorkflowInput::new("depth", InputType::Number).with_default(json!(3)), + WorkflowInput::new("note", InputType::String), + ]; + let resolved = + resolve_inputs(&declared, &supplied(&[("repo", json!("acme/api"))])).unwrap(); + + assert_eq!(resolved.len(), 3, "one entry per declared input"); + assert_eq!(resolved["repo"], json!("acme/api")); + assert_eq!(resolved["depth"], json!(3)); + assert_eq!(resolved["note"], json!(null)); + } + + #[test] + fn supplied_value_overrides_default() { + let declared = vec![WorkflowInput::new("depth", InputType::Number).with_default(json!(3))]; + let resolved = resolve_inputs(&declared, &supplied(&[("depth", json!(9))])).unwrap(); + assert_eq!(resolved["depth"], json!(9)); + } + + #[test] + fn missing_required_input_is_rejected() { + let declared = vec![WorkflowInput::new("repo", InputType::String).required()]; + let err = resolve_inputs(&declared, &Map::new()).unwrap_err(); + assert_eq!(err, InputError::Missing("repo".into())); + assert_eq!(err.code(), "input_missing"); + assert_eq!(err.input_name(), "repo"); + assert_eq!( + err.to_string(), + "workflow input \"repo\" is required but was not supplied" + ); + } + + #[test] + fn type_mismatch_is_rejected_without_coercion() { + let declared = vec![WorkflowInput::new("depth", InputType::Number)]; + let err = resolve_inputs(&declared, &supplied(&[("depth", json!("3"))])).unwrap_err(); + assert_eq!( + err, + InputError::TypeMismatch { + name: "depth".into(), + expected: "number", + found: "string", + } + ); + assert_eq!(err.code(), "input_type_mismatch"); + assert_eq!( + err.to_string(), + "workflow input \"depth\" expects type number but received string" + ); + } + + #[test] + fn explicit_null_fails_a_scalar_type_but_passes_json() { + let scalar = vec![WorkflowInput::new("depth", InputType::Number)]; + assert!(resolve_inputs(&scalar, &supplied(&[("depth", json!(null))])).is_err()); + + let any = vec![WorkflowInput::new("payload", InputType::Json)]; + let resolved = resolve_inputs(&any, &supplied(&[("payload", json!(null))])).unwrap(); + assert_eq!(resolved["payload"], json!(null)); + } + + #[test] + fn undeclared_key_is_rejected() { + let declared = vec![WorkflowInput::new("repo", InputType::String)]; + let err = resolve_inputs( + &declared, + &supplied(&[("repo", json!("a")), ("reop", json!("typo"))]), + ) + .unwrap_err(); + assert_eq!(err, InputError::Unknown("reop".into())); + assert_eq!(err.code(), "input_unknown"); + } + + #[test] + fn declaration_errors_are_reported_before_undeclared_keys() { + // A caller who both forgot a required input and mistyped another key + // hears about the required one first — it is the actionable failure. + let declared = vec![WorkflowInput::new("repo", InputType::String).required()]; + let err = resolve_inputs(&declared, &supplied(&[("reop", json!("typo"))])).unwrap_err(); + assert_eq!(err, InputError::Missing("repo".into())); + } + + #[test] + fn no_declarations_accepts_nothing_and_yields_empty() { + assert!(resolve_inputs(&[], &Map::new()).unwrap().is_empty()); + assert_eq!( + resolve_inputs(&[], &supplied(&[("x", json!(1))])).unwrap_err(), + InputError::Unknown("x".into()) + ); + } + + #[test] + fn builders_compose() { + let input = WorkflowInput::new("depth", InputType::Number) + .with_default(json!(3)) + .with_description("How deep to recurse"); + assert_eq!(input.default, Some(json!(3))); + assert_eq!(input.description.as_deref(), Some("How deep to recurse")); + assert!(!input.required); + assert!( + WorkflowInput::new("repo", InputType::String) + .required() + .required + ); + } +} diff --git a/src/model/mod.rs b/src/model/mod.rs index 1d1a32a..cc78c10 100644 --- a/src/model/mod.rs +++ b/src/model/mod.rs @@ -15,9 +15,17 @@ //! //! Both fields are `#[serde(default)]`, so definitions persisted before they //! existed still load. Load-time upgrades are performed by [`crate::migrate`]. +//! +//! ## Inputs +//! +//! A graph also declares its parameters — see [`WorkflowInput`] and +//! [`resolve_inputs`]. They are the workflow's public signature, validated +//! before a run starts and addressed from node config as `=inputs.`. +mod inputs; mod node_kind; +pub use inputs::{InputError, InputType, WorkflowInput, is_valid_input_name, resolve_inputs}; pub use node_kind::{NodeKind, TriggerKind}; use serde::{Deserialize, Serialize}; @@ -140,6 +148,14 @@ pub struct WorkflowGraph { /// Human-readable workflow name. #[serde(default)] pub name: String, + /// The workflow's declared parameters — its public signature. Empty for + /// graphs authored before inputs existed, and for graphs that take none. + /// + /// Values supplied by a caller are validated against these declarations + /// before the run starts (see [`resolve_inputs`]) and exposed to node + /// configuration as `=inputs.`. + #[serde(default)] + pub inputs: Vec, /// The nodes in the graph. #[serde(default)] pub nodes: Vec, @@ -156,6 +172,7 @@ impl Default for WorkflowGraph { schema_version: CURRENT_SCHEMA_VERSION, id: None, name: String::new(), + inputs: Vec::new(), nodes: Vec::new(), edges: Vec::new(), } @@ -249,6 +266,12 @@ mod tests { schema_version: CURRENT_SCHEMA_VERSION, id: Some("wf_1".to_string()), name: "demo".to_string(), + inputs: vec![ + WorkflowInput::new("repo", InputType::String).required(), + WorkflowInput::new("depth", InputType::Number) + .with_default(serde_json::json!(3)) + .with_description("How deep to recurse"), + ], nodes: vec![node("t", NodeKind::Trigger), node("a", NodeKind::Agent)], edges: vec![Edge { from_node: "t".to_string(), @@ -371,12 +394,35 @@ mod tests { assert_eq!(graph.successors("t"), vec!["a", "a"]); } + #[test] + fn graphs_authored_before_inputs_existed_still_load() { + let graph: WorkflowGraph = serde_json::from_str( + r#"{"name":"legacy","nodes":[{"id":"t","kind":"trigger","name":"start"}],"edges":[]}"#, + ) + .expect("deserialize"); + assert!(graph.inputs.is_empty()); + } + + #[test] + fn declared_inputs_survive_a_json_round_trip() { + let graph = WorkflowGraph { + inputs: vec![WorkflowInput::new("repo", InputType::String).required()], + nodes: vec![node("t", NodeKind::Trigger)], + ..Default::default() + }; + let json = serde_json::to_string(&graph).expect("serialize"); + assert!(json.contains(r#""inputs":[{"name":"repo","type":"string""#)); + let back: WorkflowGraph = serde_json::from_str(&json).expect("deserialize"); + assert_eq!(back, graph); + } + #[test] fn round_trip_preserves_version_fields() { let graph = WorkflowGraph { schema_version: CURRENT_SCHEMA_VERSION, id: Some("wf_1".to_string()), name: "demo".to_string(), + inputs: Vec::new(), nodes: vec![Node { id: "t".to_string(), kind: NodeKind::Trigger, diff --git a/src/validate.rs b/src/validate.rs index fc6012b..7035c4b 100644 --- a/src/validate.rs +++ b/src/validate.rs @@ -5,7 +5,7 @@ use std::collections::HashSet; use serde_json::Value; use crate::error::ValidationError; -use crate::model::{NodeKind, WorkflowGraph}; +use crate::model::{NodeKind, WorkflowGraph, is_valid_input_name}; /// The node kind's wire discriminator (`tool_call`, `sub_workflow`, …) for use /// in error messages, so a validation error names the kind the way the graph @@ -20,10 +20,11 @@ fn kind_name(kind: &NodeKind) -> String { /// Validates a workflow graph's structure. /// /// Currently checks: unique node ids, exactly one trigger node, that every edge -/// references existing nodes, no duplicate edges, and per-node `on_error` policy -/// sanity (a known value, and an `error` edge when the policy is `route`). -/// Cycle-legality and per-kind configuration checks are completed in stages -/// A1–A2. +/// references existing nodes, no duplicate edges, per-node `on_error` policy +/// sanity (a known value, and an `error` edge when the policy is `route`), and +/// declared-input sanity (addressable, unique names; defaults that match their +/// declared type). Cycle-legality and per-kind configuration checks are +/// completed in stages A1–A2. /// /// # Errors /// Returns the first [`ValidationError`] encountered. For a full list of every @@ -42,9 +43,9 @@ pub fn validate(graph: &WorkflowGraph) -> Result<(), ValidationError> { /// /// Returns an empty `Vec` for a valid graph. The checks are ordered /// deterministically (duplicate ids → trigger count → edge integrity → -/// `on_error` policy → per-kind config → condition routing), and every error is -/// self-contained (no check can panic on a graph that failed an earlier one), -/// so accumulating is safe. The first element is identical to what +/// `on_error` policy → per-kind config → condition routing → declared inputs), +/// and every error is self-contained (no check can panic on a graph that failed +/// an earlier one), so accumulating is safe. The first element is identical to what /// [`validate`] returns, preserving the historical single-error contract. /// /// This is what a host should surface to an author or agent: fixing five @@ -422,6 +423,35 @@ pub fn validate_all(graph: &WorkflowGraph) -> Vec { } } + // Declared-input checks. These are author-time mistakes that would otherwise + // surface as a confusing runtime `null`: a name that `=inputs.` cannot + // address, two declarations racing for the same key, a default the input's + // own type would reject, or `required` alongside a default (which makes the + // requirement unreachable — a default always supplies a value). + let mut seen_inputs = HashSet::new(); + for input in &graph.inputs { + if !is_valid_input_name(&input.name) { + errors.push(ValidationError::InvalidInputName(input.name.clone())); + } + if !seen_inputs.insert(input.name.as_str()) { + errors.push(ValidationError::DuplicateInputName(input.name.clone())); + } + match &input.default { + Some(default) if !input.ty.accepts(default) => { + errors.push(ValidationError::InputDefaultTypeMismatch { + name: input.name.clone(), + expected: input.ty.as_str(), + }); + } + Some(_) if input.required => { + errors.push(ValidationError::RequiredInputWithDefault( + input.name.clone(), + )); + } + _ => {} + } + } + errors } @@ -442,6 +472,108 @@ mod tests { } } + /// A graph with one trigger and no edges — the minimum that passes every + /// structural check, so an inputs test sees only inputs errors. + fn graph_with_inputs(inputs: Vec) -> WorkflowGraph { + WorkflowGraph { + inputs, + nodes: vec![node("t", NodeKind::Trigger)], + ..Default::default() + } + } + + #[test] + fn accepts_declared_inputs() { + use crate::model::{InputType, WorkflowInput}; + + let graph = graph_with_inputs(vec![ + WorkflowInput::new("repo", InputType::String).required(), + WorkflowInput::new("depth", InputType::Number).with_default(serde_json::json!(3)), + WorkflowInput::new("payload", InputType::Json), + ]); + assert_eq!(validate(&graph), Ok(())); + } + + #[test] + fn rejects_duplicate_input_names() { + use crate::model::{InputType, WorkflowInput}; + + let graph = graph_with_inputs(vec![ + WorkflowInput::new("repo", InputType::String), + WorkflowInput::new("repo", InputType::Number), + ]); + assert_eq!( + validate(&graph), + Err(ValidationError::DuplicateInputName("repo".to_string())) + ); + } + + #[test] + fn rejects_input_names_expressions_could_not_address() { + use crate::model::{InputType, WorkflowInput}; + + for bad in ["repo-url", "2fa", "", "repo.url"] { + let graph = graph_with_inputs(vec![WorkflowInput::new(bad, InputType::String)]); + assert_eq!( + validate(&graph), + Err(ValidationError::InvalidInputName(bad.to_string())), + "{bad} should be rejected" + ); + } + } + + #[test] + fn rejects_default_that_violates_its_own_type() { + use crate::model::{InputType, WorkflowInput}; + + let graph = graph_with_inputs(vec![ + WorkflowInput::new("depth", InputType::Number).with_default(serde_json::json!("3")), + ]); + assert_eq!( + validate(&graph), + Err(ValidationError::InputDefaultTypeMismatch { + name: "depth".to_string(), + expected: "number", + }) + ); + } + + #[test] + fn rejects_required_input_with_a_default() { + use crate::model::{InputType, WorkflowInput}; + + let graph = graph_with_inputs(vec![ + WorkflowInput::new("repo", InputType::String) + .required() + .with_default(serde_json::json!("acme/api")), + ]); + assert_eq!( + validate(&graph), + Err(ValidationError::RequiredInputWithDefault( + "repo".to_string() + )) + ); + } + + #[test] + fn collects_every_input_error_in_one_pass() { + use crate::model::{InputType, WorkflowInput}; + + let graph = graph_with_inputs(vec![ + WorkflowInput::new("repo-url", InputType::String), + WorkflowInput::new("depth", InputType::Number).with_default(serde_json::json!("3")), + WorkflowInput::new("depth", InputType::Number), + ]); + let errors = validate_all(&graph); + assert_eq!(errors.len(), 3, "got {errors:?}"); + assert!(errors.contains(&ValidationError::InvalidInputName("repo-url".to_string()))); + assert!(errors.contains(&ValidationError::InputDefaultTypeMismatch { + name: "depth".to_string(), + expected: "number", + })); + assert!(errors.contains(&ValidationError::DuplicateInputName("depth".to_string()))); + } + #[test] fn accepts_a_minimal_valid_graph() { let graph = WorkflowGraph { From d26e4c84693df694a335495c16192af0e4f3d786 Mon Sep 17 00:00:00 2001 From: Steven Enamakel Date: Fri, 31 Jul 2026 13:36:16 +0300 Subject: [PATCH 2/5] feat(engine): resolve declared inputs and expose them to expressions MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Entry points now take `impl Into` — a struct carrying the trigger payload alongside values for the workflow's declared inputs. `From` keeps every existing caller compiling unchanged. `build_and_run` resolves the values against the graph's declarations before minting a run id, notifying the observer, or building the graph, so an input error means provably nothing ran. Resolved values are seeded at `run.inputs` and lifted to the `inputs` expression scope, so config reads `=inputs.`. A `sub_workflow` node forwards values to its child via a new `inputs` config object, each field resolved against the parent's scope. --- src/engine.rs | 177 ++++++++++++++++++-------- src/nodes/integration/sub_workflow.rs | 56 +++++++- src/nodes/mod.rs | 13 +- 3 files changed, 191 insertions(+), 55 deletions(-) diff --git a/src/engine.rs b/src/engine.rs index fce276d..d2b8a01 100644 --- a/src/engine.rs +++ b/src/engine.rs @@ -4,8 +4,11 @@ //! [`WorkflowGraph`](crate::model::WorkflowGraph) — capturing the run's host //! [`Capabilities`] in each node handler — then drives it and returns the final //! run state. State is a [`serde_json::Value`] laid out as -//! `{ "run": { "trigger": … }, "nodes": { "": { "items": [ … ] } } }`; -//! a merge reducer folds each node's item output into that map. +//! `{ "run": { "trigger": …, "inputs": { … } }, "nodes": { "": { "items": [ … ] } } }`; +//! a merge reducer folds each node's item output into that map. `run.trigger` is +//! the free-form payload that fired the run; `run.inputs` is the workflow's +//! resolved declared inputs (see [`RunInput`]), which node config addresses as +//! `=inputs.`. //! //! Lowering covers the **linear** path (one successor per node), **conditional //! branching** (successors on distinct ports), **parallel fan-out** (several @@ -98,6 +101,67 @@ impl CancellationToken { } } +/// What a caller hands a run: the trigger payload plus values for the +/// workflow's declared inputs. +/// +/// The two are deliberately separate channels. The **trigger payload** is +/// whatever fired the run — a webhook body, a chat message, an empty object for +/// a manual start — and is free-form by nature. **Inputs** are the workflow's +/// declared, typed parameters (see [`crate::model::WorkflowInput`]); they are +/// validated against the graph's declarations before anything executes. +/// +/// Every entry point takes `impl Into`, and [`Value`] converts, so a +/// caller with no declared inputs passes a bare payload exactly as before: +/// +/// ``` +/// use tinyflows::engine::RunInput; +/// use serde_json::json; +/// +/// // Trigger payload only — the historical form. +/// let plain: RunInput = json!({"from": "webhook"}).into(); +/// assert!(plain.inputs.is_empty()); +/// +/// // With declared input values. +/// let mut values = serde_json::Map::new(); +/// values.insert("repo".into(), json!("acme/api")); +/// let parameterized = RunInput::new(json!({})).with_inputs(values); +/// assert_eq!(parameterized.inputs["repo"], json!("acme/api")); +/// ``` +#[derive(Debug, Clone, Default)] +pub struct RunInput { + /// The trigger payload, seeded as the trigger node's item and `run.trigger`. + pub trigger: Value, + /// Caller-supplied values for the workflow's declared inputs, by name. + /// Validated by [`crate::model::resolve_inputs`] before the run starts. + pub inputs: Map, +} + +impl RunInput { + /// A run carrying only a trigger payload and no declared-input values. + #[must_use] + pub fn new(trigger: Value) -> Self { + Self { + trigger, + inputs: Map::new(), + } + } + + /// Attaches values for the workflow's declared inputs. + #[must_use] + pub fn with_inputs(mut self, inputs: Map) -> Self { + self.inputs = inputs; + self + } +} + +impl From for RunInput { + /// Treats a bare JSON value as a trigger payload with no declared inputs — + /// what every caller meant before inputs existed. + fn from(trigger: Value) -> Self { + Self::new(trigger) + } +} + /// The result of a completed workflow run. #[derive(Debug, Clone)] pub struct RunOutcome { @@ -447,7 +511,7 @@ fn error_item(node_id: &str, e: &EngineError) -> Item { /// not yet implemented surfaces its `Unimplemented` error here. pub async fn run( workflow: &CompiledWorkflow, - input: Value, + input: impl Into, capabilities: &Capabilities, ) -> Result { run_with_observer( @@ -470,7 +534,7 @@ pub async fn run( /// Same as [`run`]. pub async fn run_with_observer( workflow: &CompiledWorkflow, - input: Value, + input: impl Into, capabilities: &Capabilities, observer: &Arc, ) -> Result { @@ -509,7 +573,7 @@ pub async fn run_with_observer( /// Same as [`run`]. pub async fn run_cancellable( workflow: &CompiledWorkflow, - input: Value, + input: impl Into, capabilities: &Capabilities, token: CancellationToken, ) -> Result { @@ -524,7 +588,7 @@ pub async fn run_cancellable( /// Same as [`run_cancellable`]. pub async fn run_cancellable_with_observer( workflow: &CompiledWorkflow, - input: Value, + input: impl Into, capabilities: &Capabilities, token: CancellationToken, observer: &Arc, @@ -570,7 +634,7 @@ pub const MAX_SUB_WORKFLOW_DEPTH: u64 = 8; /// Same as [`run`]. pub(crate) async fn run_sub_workflow( workflow: &CompiledWorkflow, - input: Value, + input: impl Into, capabilities: &Capabilities, depth: u64, ) -> Result { @@ -1316,7 +1380,7 @@ fn default_thread_id(workflow: &CompiledWorkflow) -> Result { #[allow(clippy::too_many_arguments)] async fn build_and_run( workflow: &CompiledWorkflow, - input: Value, + input: impl Into, capabilities: &Capabilities, observer: &Arc, checkpointer: Arc>, @@ -1325,6 +1389,14 @@ async fn build_and_run( run_meta_overlay: Option, token: CancellationToken, ) -> Result<(CompiledGraph, String, RunOutcome, GraphRunIds)> { + // Declared inputs are resolved FIRST — before the run id is minted, before + // the observer is told a run started, and before any graph is built. A + // caller that gets an `Input` error can therefore be certain nothing ran and + // nothing was observed, which is what lets a host reject a bad call without + // recording a phantom run. + let RunInput { trigger, inputs } = input.into(); + let resolved_inputs = crate::model::resolve_inputs(&workflow.graph.inputs, &inputs)?; + // Process-local, monotonic run id — no time/random source. let run_id = format!("run-{}", NEXT_RUN_ID.fetch_add(1, Ordering::Relaxed)); observer.on_run_start(&run_id); @@ -1347,9 +1419,13 @@ async fn build_and_run( token.clone(), )?; - let seed_items = items_update(&trigger_id, &[Item::new(input.clone())], None) + let seed_items = items_update(&trigger_id, &[Item::new(trigger.clone())], None) .map_err(|e| EngineError::Capability(e.to_string()))?; - let mut initial = json!({ "run": { "trigger": input } }); + // `run.inputs` holds the resolved declared inputs — one entry per + // declaration, defaults already applied. `expr_scope_for` lifts it to the + // top-level `inputs` scope key, so node config addresses it as + // `=inputs.` (and jq programs walking `run` still see it too). + let mut initial = json!({ "run": { "trigger": trigger, "inputs": resolved_inputs } }); merge(&mut initial, seed_items); // Optional run-level metadata overlaid onto `run` before the run starts — // e.g. the `sub_workflow_depth` counter a nested `sub_workflow` run threads @@ -1441,15 +1517,33 @@ async fn build_and_run( /// execution of the resumed run fails. pub async fn resume( workflow: &CompiledWorkflow, - input: Value, + input: impl Into, newly_approved: Vec, capabilities: &Capabilities, ) -> Result { - // Union `newly_approved` into `input["approvals"]`: start from any existing - // approvals array (ignoring non-string entries), then append each newly - // approved id that is not already present. Reading defensively — a missing or - // non-array `approvals` simply yields an empty starting set, never a panic. - let mut approvals: Vec = input + run( + workflow, + merge_approvals(input, newly_approved), + capabilities, + ) + .await +} + +/// Unions `newly_approved` into the run input's `trigger["approvals"]`, leaving +/// the declared-input values untouched. +/// +/// Reads defensively: a missing or non-array `approvals` yields an empty +/// starting set, and a non-object trigger (which carries no fields to preserve) +/// is replaced by a fresh object holding just the merged approvals. Declared +/// inputs ride along unchanged — a resume re-runs the *same* parameterized +/// workflow, so dropping them would silently change what it does. +fn merge_approvals(input: impl Into, newly_approved: Vec) -> RunInput { + let RunInput { + mut trigger, + inputs, + } = input.into(); + + let mut approvals: Vec = trigger .get("approvals") .and_then(Value::as_array) .map(|existing| { @@ -1466,16 +1560,13 @@ pub async fn resume( } } - let mut merged_input = input; - if let Value::Object(map) = &mut merged_input { + if let Value::Object(map) = &mut trigger { map.insert("approvals".to_string(), json!(approvals)); } else { - // A non-object input carries no fields to preserve, so replace it with a - // fresh object holding just the merged approvals. - merged_input = json!({ "approvals": approvals }); + trigger = json!({ "approvals": approvals }); } - run(workflow, merged_input, capabilities).await + RunInput { trigger, inputs } } /// Like [`resume`], but observes `token`: cancelling it winds the resumed run @@ -1487,36 +1578,18 @@ pub async fn resume( /// Same as [`resume`]. pub async fn resume_cancellable( workflow: &CompiledWorkflow, - input: Value, + input: impl Into, newly_approved: Vec, capabilities: &Capabilities, token: CancellationToken, ) -> Result { - let mut approvals: Vec = input - .get("approvals") - .and_then(Value::as_array) - .map(|existing| { - existing - .iter() - .filter_map(Value::as_str) - .map(str::to_string) - .collect() - }) - .unwrap_or_default(); - for id in newly_approved { - if !approvals.contains(&id) { - approvals.push(id); - } - } - - let mut merged_input = input; - if let Value::Object(map) = &mut merged_input { - map.insert("approvals".to_string(), json!(approvals)); - } else { - merged_input = json!({ "approvals": approvals }); - } - - run_cancellable(workflow, merged_input, capabilities, token).await + run_cancellable( + workflow, + merge_approvals(input, newly_approved), + capabilities, + token, + ) + .await } /// A live, resumable workflow run. @@ -1601,7 +1674,7 @@ impl ResumableRun { /// Same as [`run`]. pub async fn run_resumable( workflow: &CompiledWorkflow, - input: Value, + input: impl Into, capabilities: &Capabilities, ) -> Result { let observer = Arc::new(crate::observability::NoopObserver) as Arc; @@ -1654,7 +1727,7 @@ pub async fn run_resumable( /// execution (including any node executor error) fails. pub async fn run_with_checkpointer( workflow: &CompiledWorkflow, - input: Value, + input: impl Into, capabilities: &Capabilities, checkpointer: Arc>, thread_id: &str, @@ -1740,7 +1813,7 @@ pub async fn resume_with_checkpointer( /// Same as [`run_with_checkpointer`]. pub async fn run_with_checkpointer_journaled( workflow: &CompiledWorkflow, - input: Value, + input: impl Into, capabilities: &Capabilities, checkpointer: Arc>, thread_id: &str, @@ -1775,7 +1848,7 @@ pub async fn run_with_checkpointer_journaled( /// Same as [`run_with_checkpointer_journaled`]. pub async fn run_with_checkpointer_journaled_observed( workflow: &CompiledWorkflow, - input: Value, + input: impl Into, capabilities: &Capabilities, checkpointer: Arc>, thread_id: &str, diff --git a/src/nodes/integration/sub_workflow.rs b/src/nodes/integration/sub_workflow.rs index 43ada75..54433fd 100644 --- a/src/nodes/integration/sub_workflow.rs +++ b/src/nodes/integration/sub_workflow.rs @@ -44,6 +44,24 @@ use crate::nodes::{NodeContext, NodeExecutor, NodeOutput}; /// The depth guard below is per child run, so a fan-out widens a run without /// deepening it — N siblings at depth d+1, never d+N. /// +/// ## Passing the child's declared inputs +/// +/// An optional `inputs` config object supplies values for the child's declared +/// [`WorkflowInput`](crate::model::WorkflowInput)s. Each field is resolved +/// against the parent's expression scope, so a parent can forward its own +/// inputs or an upstream node's output: +/// +/// ```json +/// { +/// "workflow_id": "review-and-fix", +/// "inputs": { "repo": "=inputs.repo", "depth": 2 } +/// } +/// ``` +/// +/// The child validates what arrives against its own declarations, so a parent +/// that omits a required child input fails the same way a top-level caller +/// would — before the child executes anything. +/// /// ## Cycle / depth handling /// /// Every nested `sub_workflow` run (inline or by id) increments a @@ -87,6 +105,35 @@ fn reject_self_reference(child: &WorkflowGraph, workflow_id: &str) -> Result<()> Ok(()) } +/// Builds the values passed to the child's declared inputs from this node's +/// `inputs` config object. +/// +/// Each field's value is resolved against the **parent's** scope, so a parent +/// can forward its own inputs (`"repo": "=inputs.repo"`), an upstream node's +/// output (`"=nodes.fetch.item.url"`), or a literal. The child then validates +/// what arrives against its own declarations, exactly as a top-level caller +/// would — a parent that forgets a required child input fails loudly. +/// +/// An absent or non-object `inputs` config yields an empty map, so a +/// `sub_workflow` node authored before inputs existed keeps working against a +/// child that declares none. +fn child_inputs(config: &Value, scope: &Value) -> Result> { + let Some(declared) = config.get("inputs") else { + return Ok(serde_json::Map::new()); + }; + let Some(fields) = declared.as_object() else { + return Err(EngineError::Capability( + "sub_workflow node: `inputs` must be an object mapping the child's declared input \ + names to values" + .to_string(), + )); + }; + Ok(fields + .iter() + .map(|(name, value)| (name.clone(), crate::expr::resolve(value, scope))) + .collect()) +} + #[async_trait] impl NodeExecutor for SubWorkflowNode { async fn execute(&self, ctx: NodeContext<'_>) -> Result { @@ -190,12 +237,17 @@ async fn run_child( } let compiled = crate::compiler::compile(&child)?; - let input = + let trigger = serde_json::to_value(child_input).map_err(|e| EngineError::Capability(e.to_string()))?; + // Resolved against the same `scope` as `workflow_id`, so a `per_item` run + // forwards values derived from *its* element (`"=item.repo"`) rather than + // from the batch — the whole point of resolving inputs in here rather than + // once at the call site. + let child_inputs = child_inputs(&ctx.node.config, scope)?; // Box the recursive engine call so the async future type stays sized. let outcome = Box::pin(crate::engine::run_sub_workflow( &compiled, - input, + crate::engine::RunInput::new(trigger).with_inputs(child_inputs), ctx.caps, child_depth, )) diff --git a/src/nodes/mod.rs b/src/nodes/mod.rs index 02c75a0..0c78164 100644 --- a/src/nodes/mod.rs +++ b/src/nodes/mod.rs @@ -57,7 +57,12 @@ pub struct NodeContext<'a> { /// `=.nodes["fetch_recipient"].items[0].email` — including non-adjacent /// (grandparent) nodes and specific predecessors of a fan-in node. Node /// **id** is the addressing key (stable across renames); names are not -/// indexed. +/// indexed; +/// - `inputs` — the workflow's resolved declared inputs, keyed by name (see +/// [`crate::model::WorkflowInput`]), so a config field reads +/// `=inputs.repo`. One entry per declaration with defaults already applied, +/// so a binding to a declared name is never *absent* — at worst it is the +/// explicit `null` of an optional input nobody supplied. #[must_use] pub(crate) fn expr_scope(ctx: &NodeContext) -> Value { let item = ctx @@ -80,6 +85,12 @@ pub(crate) fn expr_scope_for(ctx: &NodeContext, item: Value) -> Value { "items": items, "run": ctx.run, "nodes": nodes_scope(ctx.nodes), + // Lifted out of `run` rather than carried separately on `NodeContext`: + // the engine seeds the resolved declared inputs at `run.inputs`, and + // this promotes them to a top-level key so authors write the short + // `=inputs.` while jq programs walking `run` still find them. + // `Null` when the run predates inputs or the graph declares none. + "inputs": ctx.run.get("inputs").cloned().unwrap_or(Value::Null), }) } From f6e2f34dcb1b4ee9f11aab83f5c4ea5e5fb49916 Mon Sep 17 00:00:00 2001 From: Steven Enamakel Date: Fri, 31 Jul 2026 13:43:05 +0300 Subject: [PATCH 3/5] feat(authoring): expose declared inputs to graph ops, catalog and docs MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Adds GraphOp::SetWorkflowInputs (whole-list replace), documents the child `inputs` map on the sub_workflow contract, and points the trigger contract at the graph-level declarations so an authoring agent does not invent a trigger-config form schema. Collapses the transform node's hand-rolled expression scope onto a single build_expr_scope constructor. That drift is what made `=inputs.` resolve to null in transform while working everywhere else — a node that builds its own scope silently loses keys, and the binding fails quietly rather than erroring. tests/inputs_e2e.rs covers the chain end to end: supplied values reaching node config, defaults, a rejected call running nothing, the bare-Value call shape still working, and a parent forwarding inputs to a sub_workflow child. --- src/catalog.rs | 18 +- src/graph_ops.rs | 69 ++++++ src/nodes/control_flow/transform.rs | 14 +- src/nodes/mod.rs | 22 +- tests/inputs_e2e.rs | 316 ++++++++++++++++++++++++++++ wiki/Architecture.md | 44 +++- wiki/Node-Catalog.md | 8 +- 7 files changed, 479 insertions(+), 12 deletions(-) create mode 100644 tests/inputs_e2e.rs diff --git a/src/catalog.rs b/src/catalog.rs index 78c0b46..d0ce7c6 100644 --- a/src/catalog.rs +++ b/src/catalog.rs @@ -206,6 +206,12 @@ pub fn contract_for(kind: &str) -> Option { notes: vec![ "Exactly ONE trigger node per graph — zero or multiple is a hard reject." .to_string(), + "A workflow's typed PARAMETERS are NOT declared here. They live in the graph's \ + top-level `inputs` array (name/type/required/default/description), are validated \ + before the run starts, and are read from any node as \"=inputs.\". The \ + trigger payload — free-form, whatever fired the run — stays at \ + \"=run.trigger.\"." + .to_string(), ], }, "agent" => NodeKindContract { @@ -458,15 +464,25 @@ pub fn contract_for(kind: &str) -> Option { "string", "The id of a saved workflow to run as the child. Provide this OR workflow.", ), + ConfigField::optional( + "inputs", + "object", + "Values for the child's declared workflow inputs, by name. Each value is \ + resolved against THIS node's scope, so a parent can forward its own inputs \ + (\"=inputs.repo\") or an upstream node's output.", + ), ], ports: PortSpec::linear(), example: json!({ "id": "sub", "kind": "sub_workflow", "name": "Enrich", - "config": { "workflow_id": "flow-123" } + "config": { "workflow_id": "flow-123", "inputs": { "repo": "=inputs.repo" } } }), notes: vec![ "Exactly one of workflow / workflow_id — having both or neither is a hard reject." .to_string(), + "The child validates config.inputs against its OWN declarations, so omitting a \ + required child input fails before the child executes anything." + .to_string(), ], }, "memory" => NodeKindContract { diff --git a/src/graph_ops.rs b/src/graph_ops.rs index 8f1fe32..d2b96de 100644 --- a/src/graph_ops.rs +++ b/src/graph_ops.rs @@ -102,6 +102,19 @@ pub enum GraphOp { /// The new canvas position. position: Position, }, + /// Replace the workflow's declared inputs wholesale. + /// + /// A whole-list replace rather than per-input add/remove/update ops: the + /// list is short, order is meaningful (it drives the order a host prompts + /// for values), and a rename is otherwise two ops that must not be + /// interleaved with anything else. Never fails structurally — the resulting + /// declarations are checked by [`crate::validate::validate_all`], which + /// reports duplicate names, unaddressable names, and self-contradictory + /// defaults. + SetWorkflowInputs { + /// The complete new set of declared inputs (empty clears them). + inputs: Vec, + }, } impl GraphOp { @@ -116,6 +129,7 @@ impl GraphOp { Self::AddEdge { .. } => "add_edge", Self::RemoveEdge { .. } => "remove_edge", Self::SetNodePosition { .. } => "set_node_position", + Self::SetWorkflowInputs { .. } => "set_workflow_inputs", } } } @@ -307,6 +321,9 @@ fn apply_one(graph: &mut WorkflowGraph, op: &GraphOp) -> Result<(), GraphOpError node_index(graph, id).ok_or_else(|| GraphOpErrorKind::NodeNotFound(id.clone()))?; graph.nodes[idx].position = Some(*position); } + GraphOp::SetWorkflowInputs { inputs } => { + graph.inputs = inputs.clone(); + } } Ok(()) } @@ -368,6 +385,58 @@ mod tests { } } + #[test] + fn set_workflow_inputs_replaces_the_whole_list() { + use crate::model::{InputType, WorkflowInput}; + + let mut base = base(); + base.inputs = vec![WorkflowInput::new("stale", InputType::String)]; + + let g = apply_ops( + &base, + &[GraphOp::SetWorkflowInputs { + inputs: vec![ + WorkflowInput::new("repo", InputType::String).required(), + WorkflowInput::new("depth", InputType::Number).with_default(json!(3)), + ], + }], + ) + .unwrap(); + + let names: Vec<&str> = g.inputs.iter().map(|i| i.name.as_str()).collect(); + assert_eq!(names, vec!["repo", "depth"], "replaced, not merged"); + assert_eq!(g.nodes.len(), base.nodes.len(), "nodes are untouched"); + } + + #[test] + fn set_workflow_inputs_can_clear_declarations() { + use crate::model::{InputType, WorkflowInput}; + + let mut base = base(); + base.inputs = vec![WorkflowInput::new("repo", InputType::String)]; + + let g = apply_ops(&base, &[GraphOp::SetWorkflowInputs { inputs: vec![] }]).unwrap(); + assert!(g.inputs.is_empty()); + } + + #[test] + fn set_workflow_inputs_round_trips_through_its_serde_tag() { + let op: GraphOp = serde_json::from_str( + r#"{"op":"set_workflow_inputs","inputs":[{"name":"repo","required":true}]}"#, + ) + .expect("deserialize"); + assert_eq!(op.name(), "set_workflow_inputs"); + match &op { + GraphOp::SetWorkflowInputs { inputs } => { + assert_eq!(inputs.len(), 1); + assert!(inputs[0].required); + } + other => panic!("expected set_workflow_inputs, got {other:?}"), + } + let back: GraphOp = serde_json::from_value(serde_json::to_value(&op).unwrap()).unwrap(); + assert_eq!(back, op); + } + #[test] fn add_node_appends_and_rejects_duplicates() { let g = apply_ops( diff --git a/src/nodes/control_flow/transform.rs b/src/nodes/control_flow/transform.rs index 8060e17..a843c20 100644 --- a/src/nodes/control_flow/transform.rs +++ b/src/nodes/control_flow/transform.rs @@ -28,12 +28,14 @@ impl NodeExecutor for TransformNode { for (index, item) in ctx.input.iter().enumerate() { // `item` is this loop's current item; `items` exposes the full input // batch; `nodes` addresses any completed upstream node by id. - let scope = serde_json::json!({ - "item": item.json.clone(), - "items": items.clone(), - "run": ctx.run, - "nodes": nodes.clone(), - }); + // Built through the shared constructor so this node can never drift + // out of sync with the scope every other node sees. + let scope = crate::nodes::build_expr_scope( + item.json.clone(), + items.clone(), + ctx.run, + nodes.clone(), + ); let mut json = item.json.clone(); if let Some(set) = &set { if !json.is_object() { diff --git a/src/nodes/mod.rs b/src/nodes/mod.rs index 0c78164..659b37c 100644 --- a/src/nodes/mod.rs +++ b/src/nodes/mod.rs @@ -80,17 +80,33 @@ pub(crate) fn expr_scope(ctx: &NodeContext) -> Value { #[must_use] pub(crate) fn expr_scope_for(ctx: &NodeContext, item: Value) -> Value { let items: Vec = ctx.input.iter().map(|i| i.json.clone()).collect(); + build_expr_scope(item, items, ctx.run, nodes_scope(ctx.nodes)) +} + +/// THE single constructor for an expression scope — every `=`-expression in the +/// crate is evaluated against an object built here. +/// +/// Takes an already-projected `nodes` scope so a per-item loop can project it +/// once and reuse it across the batch (see the `transform` node), while callers +/// with a [`NodeContext`] go through [`expr_scope`] / [`expr_scope_for`]. +/// +/// Keeping this in one place is load-bearing, not tidiness: a node that +/// hand-rolls the object silently loses whatever key it was written before, and +/// the binding fails as a quiet `null` rather than an error. Add a key here and +/// every node sees it. +#[must_use] +pub(crate) fn build_expr_scope(item: Value, items: Vec, run: &Value, nodes: Value) -> Value { serde_json::json!({ "item": item, "items": items, - "run": ctx.run, - "nodes": nodes_scope(ctx.nodes), + "run": run, + "nodes": nodes, // Lifted out of `run` rather than carried separately on `NodeContext`: // the engine seeds the resolved declared inputs at `run.inputs`, and // this promotes them to a top-level key so authors write the short // `=inputs.` while jq programs walking `run` still find them. // `Null` when the run predates inputs or the graph declares none. - "inputs": ctx.run.get("inputs").cloned().unwrap_or(Value::Null), + "inputs": run.get("inputs").cloned().unwrap_or(Value::Null), }) } diff --git a/tests/inputs_e2e.rs b/tests/inputs_e2e.rs new file mode 100644 index 0000000..dccc70d --- /dev/null +++ b/tests/inputs_e2e.rs @@ -0,0 +1,316 @@ +#![cfg(feature = "mock")] +//! End-to-end tests for declared workflow inputs. +//! +//! A graph declares typed parameters in its top-level `inputs` array; a caller +//! supplies values through [`RunInput`], they are validated before anything +//! executes, and node config reads them as `=inputs.` (see +//! `src/model/inputs.rs` and the `inputs` scope key in `src/nodes/mod.rs`). +//! +//! These tests assert the *whole chain* rather than any one layer: that a +//! supplied value reaches a node's resolved config, that defaults are applied, +//! that a rejected call runs nothing at all, and that a parent forwards values +//! to a `sub_workflow` child. +//! +//! Gated behind the `mock` feature, so plain `cargo test` skips it while +//! `cargo test --features mock` runs it. + +use serde_json::{Map, Value, json}; +use tinyflows::caps::mock::{ + MockWorkflowResolver, mock_capabilities, mock_capabilities_with_resolver, +}; +use tinyflows::compiler::compile; +use tinyflows::engine::{RunInput, run}; +use tinyflows::error::EngineError; +use tinyflows::model::{ + Edge, InputType, Node, NodeKind, TriggerKind, WorkflowGraph, WorkflowInput, +}; + +/// Builds a node with the given id, kind, and config. +fn node(id: &str, kind: NodeKind, config: Value) -> Node { + Node { + id: id.to_string(), + kind, + type_version: 1, + name: id.to_string(), + config, + ports: vec![], + position: None, + } +} + +/// Builds a trigger node with the given firing mode. +fn trigger(id: &str, kind: TriggerKind) -> Node { + node(id, NodeKind::Trigger, json!({ "kind": kind })) +} + +/// Builds an edge from `from_node`'s `main` port into `to_node`'s `main` port. +fn edge(from_node: &str, to_node: &str) -> Edge { + Edge { + from_node: from_node.to_string(), + from_port: "main".to_string(), + to_node: to_node.to_string(), + to_port: "main".to_string(), + } +} + +/// Collects `pairs` into the supplied-values map a caller hands [`RunInput`]. +fn values(pairs: &[(&str, Value)]) -> Map { + pairs + .iter() + .map(|(k, v)| ((*k).to_string(), v.clone())) + .collect() +} + +/// A graph declaring `repo` (required) and `depth` (defaulted), whose single +/// `transform` node copies both into its output via `=inputs.`. +/// +/// `transform` is used because the mock capabilities echo it deterministically, +/// so the assertion is on the *resolved* config values, not on any provider's +/// behaviour. +fn parameterized_graph() -> WorkflowGraph { + WorkflowGraph { + name: "parameterized".to_string(), + inputs: vec![ + WorkflowInput::new("repo", InputType::String) + .required() + .with_description("Repo to review"), + WorkflowInput::new("depth", InputType::Number).with_default(json!(3)), + WorkflowInput::new("note", InputType::String), + ], + nodes: vec![ + trigger("t", TriggerKind::Manual), + node( + "shape", + NodeKind::Transform, + json!({ "set": { "repo": "=inputs.repo", "depth": "=inputs.depth", "note": "=inputs.note" } }), + ), + ], + edges: vec![edge("t", "shape")], + ..Default::default() + } +} + +#[tokio::test] +async fn supplied_input_reaches_node_config_and_defaults_are_applied() { + let compiled = compile(¶meterized_graph()).expect("compile"); + let caps = mock_capabilities(); + + let outcome = run( + &compiled, + RunInput::new(json!({})).with_inputs(values(&[("repo", json!("acme/api"))])), + &caps, + ) + .await + .expect("run"); + + let shaped = &outcome.output["nodes"]["shape"]["items"][0]["json"]; + assert_eq!(shaped["repo"], json!("acme/api"), "supplied value"); + assert_eq!(shaped["depth"], json!(3), "declared default applied"); + assert_eq!( + shaped["note"], + json!(null), + "optional input with no default resolves to null, not absent" + ); +} + +#[tokio::test] +async fn resolved_inputs_are_readable_through_the_run_slice_too() { + // `run.inputs` is the seeded location; the `inputs` scope key is lifted from + // it. Both are part of the contract, so a jq program walking `run` works. + let compiled = compile(¶meterized_graph()).expect("compile"); + let caps = mock_capabilities(); + + let outcome = run( + &compiled, + RunInput::new(json!({ "from": "webhook" })) + .with_inputs(values(&[("repo", json!("acme/api"))])), + &caps, + ) + .await + .expect("run"); + + assert_eq!(outcome.output["run"]["inputs"]["repo"], json!("acme/api")); + assert_eq!(outcome.output["run"]["inputs"]["depth"], json!(3)); + // The trigger payload is a separate channel and is untouched by inputs. + assert_eq!( + outcome.output["run"]["trigger"], + json!({ "from": "webhook" }) + ); +} + +#[tokio::test] +async fn a_missing_required_input_runs_nothing() { + let compiled = compile(¶meterized_graph()).expect("compile"); + let caps = mock_capabilities(); + + let err = run(&compiled, json!({}), &caps) + .await + .expect_err("a missing required input must fail the run"); + + match err { + EngineError::Input(inner) => { + assert_eq!(inner.code(), "input_missing"); + assert_eq!(inner.input_name(), "repo"); + } + other => panic!("expected an input error, got: {other:?}"), + } +} + +#[tokio::test] +async fn a_wrongly_typed_input_is_rejected_before_the_run() { + let compiled = compile(¶meterized_graph()).expect("compile"); + let caps = mock_capabilities(); + + let err = run( + &compiled, + RunInput::new(json!({})).with_inputs(values(&[ + ("repo", json!("acme/api")), + ("depth", json!("3")), + ])), + &caps, + ) + .await + .expect_err("a string for a number input must be rejected"); + + assert!( + matches!(&err, EngineError::Input(inner) if inner.code() == "input_type_mismatch"), + "expected a type mismatch, got: {err:?}" + ); +} + +#[tokio::test] +async fn an_undeclared_input_is_rejected_rather_than_silently_dropped() { + let compiled = compile(¶meterized_graph()).expect("compile"); + let caps = mock_capabilities(); + + let err = run( + &compiled, + RunInput::new(json!({})).with_inputs(values(&[ + ("repo", json!("acme/api")), + ("reop", json!("typo")), + ])), + &caps, + ) + .await + .expect_err("an undeclared key must be rejected"); + + assert!( + matches!(&err, EngineError::Input(inner) if inner.input_name() == "reop"), + "expected the typo to be named, got: {err:?}" + ); +} + +#[tokio::test] +async fn a_graph_declaring_no_inputs_still_runs_from_a_bare_payload() { + // The historical call shape — a bare `Value` — must keep working untouched. + let graph = WorkflowGraph { + nodes: vec![ + trigger("t", TriggerKind::Manual), + node( + "shape", + NodeKind::Transform, + json!({ "set": { "seen": "=run.trigger.hi" } }), + ), + ], + edges: vec![edge("t", "shape")], + ..Default::default() + }; + let compiled = compile(&graph).expect("compile"); + let caps = mock_capabilities(); + + let outcome = run(&compiled, json!({ "hi": 1 }), &caps) + .await + .expect("run"); + assert_eq!( + outcome.output["nodes"]["shape"]["items"][0]["json"]["seen"], + json!(1) + ); +} + +#[tokio::test] +async fn a_parent_forwards_its_own_inputs_to_a_sub_workflow_child() { + // The child declares `repo`; the parent declares `repo` too and forwards it + // through the sub_workflow node's `inputs` config. + let child = WorkflowGraph { + name: "child".to_string(), + inputs: vec![WorkflowInput::new("repo", InputType::String).required()], + nodes: vec![ + trigger("ct", TriggerKind::Manual), + node( + "echo", + NodeKind::Transform, + json!({ "set": { "child_repo": "=inputs.repo" } }), + ), + ], + edges: vec![edge("ct", "echo")], + ..Default::default() + }; + let caps = + mock_capabilities_with_resolver(MockWorkflowResolver::default().with("child-1", child)); + + let parent = WorkflowGraph { + name: "parent".to_string(), + inputs: vec![WorkflowInput::new("repo", InputType::String).required()], + nodes: vec![ + trigger("t", TriggerKind::Manual), + node( + "sub", + NodeKind::SubWorkflow, + json!({ "workflow_id": "child-1", "inputs": { "repo": "=inputs.repo" } }), + ), + ], + edges: vec![edge("t", "sub")], + ..Default::default() + }; + let compiled = compile(&parent).expect("compile parent"); + + let outcome = run( + &compiled, + RunInput::new(json!({})).with_inputs(values(&[("repo", json!("acme/api"))])), + &caps, + ) + .await + .expect("run parent"); + + let child_state = &outcome.output["nodes"]["sub"]["items"][0]["json"]; + assert_eq!( + child_state["nodes"]["echo"]["items"][0]["json"]["child_repo"], + json!("acme/api"), + "the parent's input should reach the child's `=inputs.repo`" + ); +} + +#[tokio::test] +async fn a_parent_that_omits_a_required_child_input_fails() { + let child = WorkflowGraph { + name: "child".to_string(), + inputs: vec![WorkflowInput::new("repo", InputType::String).required()], + nodes: vec![trigger("ct", TriggerKind::Manual)], + ..Default::default() + }; + let caps = + mock_capabilities_with_resolver(MockWorkflowResolver::default().with("child-1", child)); + + let parent = WorkflowGraph { + name: "parent".to_string(), + nodes: vec![ + trigger("t", TriggerKind::Manual), + node( + "sub", + NodeKind::SubWorkflow, + json!({ "workflow_id": "child-1" }), + ), + ], + edges: vec![edge("t", "sub")], + ..Default::default() + }; + let compiled = compile(&parent).expect("compile parent"); + + let err = run(&compiled, json!({}), &caps) + .await + .expect_err("the child's requirement must be enforced across the boundary"); + assert!( + err.to_string().contains("repo"), + "the error should name the missing child input, got: {err}" + ); +} diff --git a/wiki/Architecture.md b/wiki/Architecture.md index 2f1c2f2..4ac9f9a 100644 --- a/wiki/Architecture.md +++ b/wiki/Architecture.md @@ -41,7 +41,10 @@ Run state is a single `serde_json::Value`: ```json { - "run": { "trigger": { /* trigger payload */ } }, + "run": { + "trigger": { /* free-form payload that fired the run */ }, + "inputs": { /* resolved declared inputs, one entry per declaration */ } + }, "nodes": { "": { "items": [ /* Item… */ ], "port": "true" } } } ``` @@ -53,6 +56,45 @@ items and emits output items. A merge reducer folds each node's partial writes under its own id, independent updates never collide, which keeps parallel fan-out correct. Field references use `=`-prefixed expressions. +## Workflow inputs + +A graph declares its parameters in a top-level `inputs` array — its public +signature, independent of how the workflow is triggered: + +```json +{ + "name": "review-and-fix", + "inputs": [ + { "name": "repo", "type": "string", "required": true, "description": "Repo to review" }, + { "name": "depth", "type": "number", "default": 3 } + ], + "nodes": [ /* … */ ] +} +``` + +A caller supplies values through `engine::RunInput`, which carries them +alongside the trigger payload. They are validated against the declarations +**before** the run id is minted, the observer is notified, or the graph is +built — so an input error means provably nothing ran. Missing required values, +type mismatches, and undeclared keys are all rejected; declared inputs the +caller omits fall back to their `default`, or to `null` when optional. + +Resolved values land at `run.inputs` and are lifted to the top-level `inputs` +expression scope, so node config reads them as `=inputs.repo`. Keep the two +channels distinct: + +| | `run.trigger` | `inputs` | +|---|---|---| +| what it is | whatever fired the run | the workflow's declared parameters | +| shape | free-form | named, typed, validated | +| discoverable from the graph | no | yes | +| read as | `=run.trigger.` | `=inputs.` | + +Inputs are **not** a secret channel — credentials reach a workflow through the +opaque connection reference the host resolves. A `sub_workflow` node forwards +values to its child with an `inputs` config object, each field resolved against +the parent's scope (`{"repo": "=inputs.repo"}`). + ## Host-agnostic seam The crate never hard-codes an LLM, tool, HTTP, code, or persistence vendor. diff --git a/wiki/Node-Catalog.md b/wiki/Node-Catalog.md index 684a3b6..daf2379 100644 --- a/wiki/Node-Catalog.md +++ b/wiki/Node-Catalog.md @@ -15,6 +15,12 @@ fires it; tinyflows injects the trigger payload as the initial run state. |------|---------|---------------------| | `trigger` | Entry node that starts the run | Out `main`; config `trigger_kind` | +A workflow's typed **parameters** are not declared on the trigger. They live in +the graph's top-level `inputs` array, are validated before the run starts, and +are read from any node as `=inputs.` — see +[Architecture → Workflow inputs](Architecture). The trigger payload stays at +`=run.trigger.`. + ## Control-flow nodes (native) Native routing logic — no host capabilities required. @@ -39,7 +45,7 @@ traits](Capability-Traits). | `http_request` | Outbound HTTP request | Config `method`, `url`, `headers`, `query`, `body` — via `HttpClient` | | `code` | Runs sandboxed user code | Config `language` (`javascript`/`python`), `source` — via `CodeRunner` | | `output_parser` | Parses/validates an agent's output into a structured shape | May use `LlmProvider` for auto-fixing; can nest as a sub-agent | -| `sub_workflow` | Runs another workflow as a nested sub-graph | Config `workflow_id`, `input` mapping | +| `sub_workflow` | Runs another workflow as a nested sub-graph | Config: exactly one of `workflow` (inline) / `workflow_id`; optional `inputs` map for the child's declared inputs | The capability-backed integration nodes (`agent`, `tool_call`, `http_request`) resolve `=` expressions anywhere in their config against the `{ item, items, run }` From df9b9367d6d496d141fbec3b24435a1d56e5a122 Mon Sep 17 00:00:00 2001 From: Steven Enamakel Date: Fri, 31 Jul 2026 15:45:37 +0300 Subject: [PATCH 4/5] docs(expr): warn that a jq program must write .inputs. MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit `inputs` is the one scope key that collides with a jq builtin — jq's own `inputs` reads further program inputs. `=inputs.repo` works because a simple dotted path is walked directly, but anything jq compiles (a concatenation, a conditional, a pipe) binds the builtin instead and silently yields nothing. Found by dry-running the new shipped example, which reported an empty agent prompt rather than an error. Documented at the scope, in the wiki, and pinned by a test that asserts both forms — including the failing one, so the warning cannot go stale without the test noticing. --- src/nodes/mod.rs | 10 ++++++++ tests/inputs_e2e.rs | 57 ++++++++++++++++++++++++++++++++++++++++++++ wiki/Architecture.md | 12 ++++++++-- 3 files changed, 77 insertions(+), 2 deletions(-) diff --git a/src/nodes/mod.rs b/src/nodes/mod.rs index 659b37c..d779fd8 100644 --- a/src/nodes/mod.rs +++ b/src/nodes/mod.rs @@ -63,6 +63,16 @@ pub struct NodeContext<'a> { /// `=inputs.repo`. One entry per declaration with defaults already applied, /// so a binding to a declared name is never *absent* — at worst it is the /// explicit `null` of an optional input nobody supplied. +/// +/// **Write `.inputs.` inside a real jq program.** `=inputs.repo` works +/// because a simple dotted path is walked directly, never compiled. Anything +/// jq actually compiles — a concatenation, a conditional, a pipe — resolves +/// bare `inputs` as jq's own `inputs` *builtin* (which reads further program +/// inputs) rather than this scope key, and the expression quietly yields +/// nothing instead of erroring. The leading dot forces the object lookup: +/// `="Review " + .inputs.repo` is right, `="Review " + inputs.repo` is not. +/// No other scope key has this problem; `inputs` is the one name jq already +/// uses. #[must_use] pub(crate) fn expr_scope(ctx: &NodeContext) -> Value { let item = ctx diff --git a/tests/inputs_e2e.rs b/tests/inputs_e2e.rs index dccc70d..b3fa1ab 100644 --- a/tests/inputs_e2e.rs +++ b/tests/inputs_e2e.rs @@ -314,3 +314,60 @@ async fn a_parent_that_omits_a_required_child_input_fails() { "the error should name the missing child input, got: {err}" ); } + +#[tokio::test] +async fn a_jq_program_must_address_inputs_with_a_leading_dot() { + // `inputs` is the one scope key that collides with a jq builtin (jq's own + // `inputs` reads further program inputs). A simple dotted path is walked + // directly and is fine; anything jq compiles needs the leading dot, and + // getting it wrong yields nothing rather than erroring — which is exactly + // why this is pinned rather than left to a doc comment. + let graph = WorkflowGraph { + name: "jq-inputs".to_string(), + inputs: vec![WorkflowInput::new("repo", InputType::String).required()], + nodes: vec![ + trigger("t", TriggerKind::Manual), + node( + "shape", + NodeKind::Transform, + json!({ "set": { + // The fast dotted path: no jq compilation, so bare `inputs` + // resolves against the scope. + "direct": "=inputs.repo", + // A real jq program, addressed correctly. + "dotted": "=\"repo: \" + .inputs.repo", + // The same program with the collision. Kept in the test on + // purpose: if a future change makes this resolve, the doc + // warning is stale and should be removed with it. + "bare": "=\"repo: \" + inputs.repo", + } }), + ), + ], + edges: vec![edge("t", "shape")], + ..Default::default() + }; + let compiled = compile(&graph).expect("compile"); + let caps = mock_capabilities(); + + let outcome = run( + &compiled, + RunInput::new(json!({})).with_inputs(values(&[("repo", json!("acme/api"))])), + &caps, + ) + .await + .expect("run"); + + let shaped = &outcome.output["nodes"]["shape"]["items"][0]["json"]; + assert_eq!(shaped["direct"], json!("acme/api"), "dotted-path fast form"); + assert_eq!( + shaped["dotted"], + json!("repo: acme/api"), + "a jq program addressing `.inputs`" + ); + assert_eq!( + shaped["bare"], + json!(null), + "bare `inputs` in a jq program hits jq's builtin and yields nothing — \ + the reason authors are told to write `.inputs.` there" + ); +} diff --git a/wiki/Architecture.md b/wiki/Architecture.md index 4ac9f9a..3d164b8 100644 --- a/wiki/Architecture.md +++ b/wiki/Architecture.md @@ -80,8 +80,16 @@ type mismatches, and undeclared keys are all rejected; declared inputs the caller omits fall back to their `default`, or to `null` when optional. Resolved values land at `run.inputs` and are lifted to the top-level `inputs` -expression scope, so node config reads them as `=inputs.repo`. Keep the two -channels distinct: +expression scope, so node config reads them as `=inputs.repo`. + +> **Inside a real jq program, write `.inputs.`.** `=inputs.repo` works +> because a simple dotted path is walked directly rather than compiled. Anything +> jq compiles — a concatenation, a conditional, a pipe — resolves bare `inputs` +> as jq's own `inputs` builtin and silently yields nothing. So +> `="Review " + .inputs.repo` is right and `="Review " + inputs.repo` is not. +> `inputs` is the only scope key that collides with a jq builtin. + +Keep the two channels distinct: | | `run.trigger` | `inputs` | |---|---|---| From 548dbe389eee7f1bd7b8b7075e60b9ec0346f2c2 Mon Sep 17 00:00:00 2001 From: Steven Enamakel Date: Fri, 31 Jul 2026 16:04:11 +0300 Subject: [PATCH 5/5] feat(sub_workflow): forward declared inputs per item as well Resolving the child `inputs` map inside `run_child` rather than at the call site means a `per_item` fan-out resolves each child's values against its own element, the same scope `workflow_id` already used. Resolving once outside would hand every child the first element's values. --- src/catalog.rs | 4 +- src/nodes/integration/sub_workflow.rs | 12 +++++ tests/inputs_e2e.rs | 74 +++++++++++++++++++++++++++ 3 files changed, 89 insertions(+), 1 deletion(-) diff --git a/src/catalog.rs b/src/catalog.rs index d0ce7c6..7381435 100644 --- a/src/catalog.rs +++ b/src/catalog.rs @@ -469,7 +469,9 @@ pub fn contract_for(kind: &str) -> Option { "object", "Values for the child's declared workflow inputs, by name. Each value is \ resolved against THIS node's scope, so a parent can forward its own inputs \ - (\"=inputs.repo\") or an upstream node's output.", + (\"=inputs.repo\") or an upstream node's output. Under \ + execution=\"per_item\" the scope is the current element, so each child in a \ + fan-out gets values from its own item (\"=item.name\").", ), ], ports: PortSpec::linear(), diff --git a/src/nodes/integration/sub_workflow.rs b/src/nodes/integration/sub_workflow.rs index 54433fd..f72cf85 100644 --- a/src/nodes/integration/sub_workflow.rs +++ b/src/nodes/integration/sub_workflow.rs @@ -62,6 +62,18 @@ use crate::nodes::{NodeContext, NodeExecutor, NodeOutput}; /// that omits a required child input fails the same way a top-level caller /// would — before the child executes anything. /// +/// Under `execution: "per_item"` the fields are resolved against **the current +/// element**, exactly like `workflow_id` is, so each child in a fan-out gets +/// values derived from its own item: +/// +/// ```json +/// { +/// "workflow_id": "review-and-fix", +/// "execution": "per_item", +/// "inputs": { "repo": "=item.name" } +/// } +/// ``` +/// /// ## Cycle / depth handling /// /// Every nested `sub_workflow` run (inline or by id) increments a diff --git a/tests/inputs_e2e.rs b/tests/inputs_e2e.rs index b3fa1ab..2fcea97 100644 --- a/tests/inputs_e2e.rs +++ b/tests/inputs_e2e.rs @@ -371,3 +371,77 @@ async fn a_jq_program_must_address_inputs_with_a_leading_dot() { the reason authors are told to write `.inputs.` there" ); } + +#[tokio::test] +async fn a_per_item_sub_workflow_forwards_inputs_derived_from_its_own_element() { + // The `inputs` map is resolved inside `run_child`, against the same scope + // as `workflow_id`. For a `per_item` fan-out that scope is the *current + // element*, so each child receives values derived from its own item rather + // than from the batch — resolving once at the call site would give every + // child the first element's values. + let child = WorkflowGraph { + name: "child".to_string(), + inputs: vec![WorkflowInput::new("repo", InputType::String).required()], + nodes: vec![ + trigger("ct", TriggerKind::Manual), + node( + "echo", + NodeKind::Transform, + json!({ "set": { "child_repo": "=inputs.repo" } }), + ), + ], + edges: vec![edge("ct", "echo")], + ..Default::default() + }; + let caps = + mock_capabilities_with_resolver(MockWorkflowResolver::default().with("child-1", child)); + + let parent = WorkflowGraph { + name: "parent".to_string(), + nodes: vec![ + trigger("t", TriggerKind::Manual), + // Fan the trigger payload's array out into one item per element… + node("fan", NodeKind::SplitOut, json!({ "path": "repos" })), + // …and run the child once per element, each with its own `repo`. + node( + "sub", + NodeKind::SubWorkflow, + json!({ + "workflow_id": "child-1", + "execution": "per_item", + "inputs": { "repo": "=item.name" } + }), + ), + ], + edges: vec![edge("t", "fan"), edge("fan", "sub")], + ..Default::default() + }; + let compiled = compile(&parent).expect("compile parent"); + + let outcome = run( + &compiled, + json!({ "repos": [{ "name": "acme/api" }, { "name": "acme/web" }] }), + &caps, + ) + .await + .expect("run parent"); + + let children = outcome.output["nodes"]["sub"]["items"] + .as_array() + .expect("one output item per element"); + assert_eq!(children.len(), 2, "both elements should have run"); + + let seen: Vec<&str> = children + .iter() + .map(|item| { + item["json"]["nodes"]["echo"]["items"][0]["json"]["child_repo"] + .as_str() + .unwrap_or("") + }) + .collect(); + assert_eq!( + seen, + vec!["acme/api", "acme/web"], + "each child should have received its own element's value" + ); +}