use std::collections::HashMap;
use serde_json::{Map, Value, json};
use crate::contracts::kinds::INSTALLED_NODE_KIND_PREFIX;
use crate::pipeline::auto_tidy_pipeline_graph;
use crate::pipeline::model::{
DslFlag, DslFlagKind, NodeDefinition, PipelineEdge, PipelineGraph, PipelineNode, PipelineNote,};
use crate::pipeline::nodes::builtin_node_definitions;
#[derive(Debug, Clone)]
pub enum DslVerb {
Get {
resource: String,
path: Option<String>,
filter: Option<String>,
status: Option<String>,
},
Describe {
kind: String,
name: String,
compact: bool,
},
Read { kind: String, name: String },
Write {
kind: String,
name: String,
body: Option<String>,
},
Delete { kind: String, name: String },
Activate { file_rel_path: String },
Deactivate { file_rel_path: String },
Execute { file_rel_path: String, input: Value },
Register {
file_rel_path: String,
title: String,
description: String,
as_json: bool,
body: String,
},
Patch {
file_rel_path: String,
target: String,
node_id: String,
flags: HashMap<String, Value>,
body: Option<String>,
},
Run { body: String, dry_run: bool },
Check { file_rel_path: String, body: String },
Git {
subcommand: String,
args: Vec<String>,
body: Option<String>,
},
NodeHelp { kind: String },
Invalid { message: String },
CredentialBlocked { reason: String },
Unknown { raw: String },
}
pub fn tokenize(s: &str) -> Vec<String> {
let mut tokens = Vec::new();
let mut current = String::new();
let mut in_single = false;
let mut in_double = false;
let mut in_backtick = false;
let mut chars = s.chars().peekable();
while let Some(ch) = chars.next() {
if in_double && ch == '\\' && matches!(chars.peek(), Some('"') | Some('\\')) {
current.push(chars.next().unwrap_or(ch));
continue;
}
match ch {
'\'' if !in_double && !in_backtick => {
in_single = !in_single;
}
'"' if !in_single && !in_backtick => {
in_double = !in_double;
}
'`' if !in_single && !in_double => {
in_backtick = !in_backtick;
current.push(ch);
}
' ' | '\t' | '\n' | '\r' if !in_single && !in_double && !in_backtick => {
if !current.is_empty() {
tokens.push(current.clone());
current.clear();
}
}
_ => {
current.push(ch);
}
}
}
if !current.is_empty() {
tokens.push(current);
}
tokens
}
fn is_double_quote_escape(bytes: &[u8], i: usize) -> bool {
bytes[i] == b'\\' && matches!(bytes.get(i + 1), Some(b'"') | Some(b'\\'))
}
pub fn split_commands(dsl: &str) -> Vec<String> {
let joined = dsl.replace("\\\r\n", "\n").replace("\\\n", "\n");
let bytes = joined.as_bytes();
let mut segments = Vec::new();
let mut start = 0usize;
let mut in_single = false;
let mut in_double = false;
let mut in_backtick = false;
let mut in_opaque_body = false;
let mut i = 0;
while i < bytes.len() {
if !in_single
&& !in_double
&& !in_backtick
&& !in_opaque_body
&& bytes[i] == b'-'
&& i + 1 < bytes.len()
&& bytes[i + 1] == b'-'
&& is_standalone_token(bytes, i, i + 2)
&& command_accepts_opaque_body(&joined[start..i])
{
in_opaque_body = true;
i += 2;
continue;
}
if in_double && !in_opaque_body && is_double_quote_escape(bytes, i) {
i += 2;
continue;
}
match bytes[i] {
b'\'' if !in_opaque_body && !in_double && !in_backtick => in_single = !in_single,
b'"' if !in_opaque_body && !in_single && !in_backtick => in_double = !in_double,
b'`' if !in_opaque_body && !in_single && !in_double => in_backtick = !in_backtick,
b'&' if !in_opaque_body
&& !in_single
&& !in_double
&& !in_backtick
&& i + 1 < bytes.len()
&& bytes[i + 1] == b'&' =>
{
let seg = joined[start..i].trim();
if !seg.is_empty() {
segments.push(seg.to_string());
}
i += 2;
start = i;
in_opaque_body = false;
continue;
}
_ => {}
}
i += 1;
}
let last = joined[start..].trim();
if !last.is_empty() {
segments.push(last.to_string());
}
segments
}
fn is_standalone_token(bytes: &[u8], start: usize, end: usize) -> bool {
let before_ok = start == 0 || bytes[start - 1].is_ascii_whitespace();
let after_ok = end >= bytes.len() || bytes[end].is_ascii_whitespace();
before_ok && after_ok
}
fn command_accepts_opaque_body(prefix: &str) -> bool {
let tokens = tokenize(prefix);
let Some(verb) = tokens.first().map(|s| s.to_lowercase()) else {
return false;
};
matches!(
verb.as_str(),
"patch" | "register" | "reg" | "run" | "write" | "create" | "git"
)
}
fn resolve_catalog_kind(raw_kind: &str, definitions: &[NodeDefinition]) -> Option<String> {
if let Some(kind) = expand_kind(raw_kind) {
return Some(kind.to_string());
}
if definitions.iter().any(|def| def.kind == raw_kind) {
return Some(raw_kind.to_string());
}
None
}
fn unknown_kind_message(raw_kind: &str, definitions: &[NodeDefinition]) -> String {
use crate::pipeline::nodes::check::suggest::{did_you_mean_phrase, suggest_kinds};
let hint = did_you_mean_phrase(&suggest_kinds(raw_kind, definitions, 3));
format!("Unknown node kind: '{raw_kind}'{hint} — help(\"pipeline/nodes\") lists every kind")
}
fn looks_like_node_kind(raw_kind: &str) -> bool {
expand_kind(raw_kind).is_some() || raw_kind.starts_with(INSTALLED_NODE_KIND_PREFIX)
}
pub const NOTE_KEYWORD: &str = "note";
fn is_note_keyword(token: &str) -> bool {
token.eq_ignore_ascii_case(NOTE_KEYWORD)
}
fn parse_pair(value: &str, sep: char, what: &str) -> Result<(f64, f64), String> {
let (a, b) = value
.split_once(sep)
.ok_or_else(|| format!("note: {what}, got `{value}`"))?;
let a: f64 = a.trim().parse().map_err(|_| format!("note: {what}, got `{value}`"))?;
let b: f64 = b.trim().parse().map_err(|_| format!("note: {what}, got `{value}`"))?;
Ok((a, b))
}
fn parse_note_statement(default_id: &str, tokens: &[String], raw: &str) -> Result<PipelineNote, String> {
let mut note = PipelineNote {
id: default_id.to_string(),
text: String::new(),
x: 0.0,
y: 0.0,
width: 0.0,
height: 0.0,
color: String::new(),
};
let mut i = 0;
while i < tokens.len() {
let flag = tokens[i].as_str();
if flag == "--" {
break;
}
let value = tokens
.get(i + 1)
.cloned()
.ok_or_else(|| format!("note: `{flag}` needs a value"))?;
match flag {
"--id" => note.id = value,
"--text" => note.text = value,
"--color" => note.color = value,
"--at" => {
let (x, y) = parse_pair(&value, ',', "--at expects x,y")?;
note.x = x;
note.y = y;
}
"--size" => {
let (w, h) = parse_pair(&value, 'x', "--size expects WIDTHxHEIGHT")?;
note.width = w;
note.height = h;
}
other => {
return Err(format!(
"note: unknown flag `{other}` (flags: --text, --at x,y, --size WxH, --color, --id; or `-- text`)"
))
}
}
i += 2;
}
if let Some(body) = extract_raw_body_from(raw) {
note.text = body;
}
if note.id.trim().is_empty() {
return Err("note id must not be empty".to_string());
}
Ok(note)
}
fn note_to_statement(note: &PipelineNote, with_id: bool) -> String {
let mut parts = vec![NOTE_KEYWORD.to_string()];
if with_id {
parts.push("--id".into());
parts.push(quote_dsl_arg(¬e.id));
}
parts.push("--text".into());
parts.push(quote_dsl_arg(¬e.text));
if note.x != 0.0 || note.y != 0.0 {
parts.push(format!("--at {},{}", note.x, note.y));
}
if note.width != 0.0 || note.height != 0.0 {
parts.push(format!("--size {}x{}", note.width, note.height));
}
if !note.color.is_empty() {
parts.push(format!("--color {}", quote_dsl_arg(¬e.color)));
}
parts.join(" ")
}
pub fn expand_kind(word: &str) -> Option<&'static str> {
static NATIVE: std::sync::OnceLock<Vec<&'static str>> = std::sync::OnceLock::new();
let native = NATIVE.get_or_init(|| {
crate::pipeline::nodes::builtin_node_definitions()
.into_iter()
.map(|def| &*Box::leak(def.kind.into_boxed_str()))
.collect()
});
let word = word.trim();
native
.iter()
.copied()
.find(|kind| *kind == word)
}
pub fn default_pins(kind: &str) -> (Vec<String>, Vec<String>) {
if kind != "logic.match" {
if let Some(def) = crate::pipeline::nodes::builtin_node_definitions()
.into_iter()
.find(|def| def.kind == kind)
{
if !def.output_pins.is_empty() {
return (def.input_pins, def.output_pins);
}
}
}
match kind {
"trigger.webhook" | "trigger.schedule" | "trigger.manual" | "trigger.function" => {
(vec![], vec!["out".to_string()])
}
"postgres.query.run" | "sekejap.query.run" | "sekejap.record.create" | "sqlite.query.run"
| "table.data.convert" | "table.query.run" | "javascript.script.run" | "typescript.script.run" | "http.response.fetch"
| "logic.collect" | "ai.audio.generate" => (vec!["in".to_string()], vec!["out".to_string()]),
"logic.foreach" => (vec!["in".to_string()], vec!["item".to_string()]),
"logic.reduce" => (vec!["in".to_string()], vec!["out".to_string()]),
"logic.retry" => (
vec!["in".to_string()],
vec!["retry".to_string(), "failed".to_string()],
),
"logic.if" => (
vec!["in".to_string()],
vec!["true".to_string(), "false".to_string()],
),
"logic.match" => (vec!["in".to_string()], vec!["default".to_string()]),
"web.response.send" => (vec!["in".to_string()], vec!["out".to_string()]),
"trigger.room" | "trigger.socket" => (vec![], vec!["out".to_string()]),
"function.result.call" => (
vec!["in".to_string()],
vec!["out".to_string(), "error".to_string()],
),
_ => (vec!["in".to_string()], vec!["out".to_string()]),
}
}
fn find_first_pipe_in_raw(raw: &str) -> Option<usize> {
let mut in_single = false;
let mut in_double = false;
let mut in_backtick = false;
let mut escaped = false;
for (i, ch) in raw.char_indices() {
if escaped {
escaped = false;
continue;
}
if in_double && is_double_quote_escape(raw.as_bytes(), i) {
escaped = true;
continue;
}
match ch {
'\'' if !in_double && !in_backtick => in_single = !in_single,
'"' if !in_single && !in_backtick => in_double = !in_double,
'`' if !in_single && !in_double => in_backtick = !in_backtick,
'|' if !in_single && !in_double && !in_backtick => return Some(i),
_ => {}
}
}
None
}
fn find_first_graph_marker_in_raw(raw: &str) -> Option<usize> {
let mut in_single = false;
let mut in_double = false;
let mut in_backtick = false;
let mut escaped = false;
for (i, ch) in raw.char_indices() {
if escaped {
escaped = false;
continue;
}
if in_double && is_double_quote_escape(raw.as_bytes(), i) {
escaped = true;
continue;
}
match ch {
'\'' if !in_double && !in_backtick => in_single = !in_single,
'"' if !in_single && !in_backtick => in_double = !in_double,
'`' if !in_single && !in_double => in_backtick = !in_backtick,
'[' if !in_single && !in_double && !in_backtick => return Some(i),
_ => {}
}
}
None
}
fn extract_pipeline_body(raw: &str) -> String {
match (
find_first_pipe_in_raw(raw),
find_first_graph_marker_in_raw(raw),
) {
(Some(pipe), Some(graph)) => raw[pipe.min(graph)..].to_string(),
(Some(pipe), None) => raw[pipe..].to_string(),
(None, Some(graph)) => raw[graph..].to_string(),
(None, None) => String::new(),
}
}
fn strip_outer_quotes(s: &str) -> &str {
if s.len() >= 2
&& ((s.starts_with('"') && s.ends_with('"'))
|| (s.starts_with('\'') && s.ends_with('\''))
|| (s.starts_with('`') && s.ends_with('`')))
{
&s[1..s.len() - 1]
} else {
s
}
}
fn extract_raw_body_from(raw: &str) -> Option<String> {
find_body_delimiter(raw)
.map(|body_start| {
let after = raw[body_start..].trim();
strip_outer_quotes(after).to_string()
})
.filter(|s| !s.is_empty())
}
fn find_body_delimiter(raw: &str) -> Option<usize> {
let bytes = raw.as_bytes();
let mut in_single = false;
let mut in_double = false;
let mut in_backtick = false;
let mut i = 0usize;
while i < bytes.len() {
if in_double && is_double_quote_escape(bytes, i) {
i += 2;
continue;
}
match bytes[i] {
b'\'' if !in_double && !in_backtick => in_single = !in_single,
b'"' if !in_single && !in_backtick => in_double = !in_double,
b'`' if !in_single && !in_double => in_backtick = !in_backtick,
b'-' if !in_single
&& !in_double
&& !in_backtick
&& i + 1 < bytes.len()
&& bytes[i + 1] == b'-'
&& is_standalone_token(bytes, i, i + 2) =>
{
let mut body_start = i + 2;
while body_start < bytes.len() && bytes[body_start].is_ascii_whitespace() {
body_start += 1;
}
return Some(body_start);
}
_ => {}
}
i += 1;
}
None
}
fn coerce_scalar_value(s: &str) -> Value {
match s {
"true" => json!(true),
"false" => json!(false),
_ if is_plain_number(s) => {
if let Ok(n) = s.parse::<i64>() {
json!(n)
} else if let Ok(f) = s.parse::<f64>() {
json!(f)
} else {
json!(s)
}
}
_ => json!(s),
}
}
fn is_plain_number(s: &str) -> bool {
let digits = s.strip_prefix('-').unwrap_or(s);
let (whole, fraction) = match digits.split_once('.') {
Some((whole, fraction)) => (whole, Some(fraction)),
None => (digits, None),
};
let whole_ok = whole == "0" || (!whole.is_empty() && !whole.starts_with('0') && whole.bytes().all(|b| b.is_ascii_digit()));
let fraction_ok = fraction.is_none_or(|f| !f.is_empty() && f.bytes().all(|b| b.is_ascii_digit()));
whole_ok && fraction_ok
}
pub const PREVIEW_KINDS: [&str; 8] = [
"image", "video", "audio", "pdf", "json", "text", "table", "html",
];
pub fn preview_slot_for_config_key(config_key: &str) -> Option<&'static str> {
match config_key {
"preview.out" => Some("out"),
"preview.in" => Some("in"),
_ => None,
}
}
pub const PREVIEW_MIN_SIZE: (u64, u64) = (160, 60);
pub const PREVIEW_MAX_SIZE: (u64, u64) = (1200, 900);
fn parse_preview_size(flag: &str, value: &str, raw: &str) -> Result<(u64, u64), String> {
let bad = || {
format!(
"`{flag} {value}` — bad size `{raw}`; the form is <as>[:<path>]@WxH with \
W {}–{} and H {}–{}, e.g. `image@360x240`",
PREVIEW_MIN_SIZE.0, PREVIEW_MAX_SIZE.0, PREVIEW_MIN_SIZE.1, PREVIEW_MAX_SIZE.1
)
};
let (w, h) = raw.trim().split_once(['x', 'X']).ok_or_else(bad)?;
let width: u64 = w.trim().parse().map_err(|_| bad())?;
let height: u64 = h.trim().parse().map_err(|_| bad())?;
let in_range = (PREVIEW_MIN_SIZE.0..=PREVIEW_MAX_SIZE.0).contains(&width)
&& (PREVIEW_MIN_SIZE.1..=PREVIEW_MAX_SIZE.1).contains(&height);
if !in_range {
return Err(bad());
}
Ok((width, height))
}
pub fn parse_preview_flag_value(flag: &str, value: &str) -> Result<Option<Value>, String> {
let trimmed = value.trim();
if trimmed.is_empty() {
return Err(format!(
"`{flag}` requires a value: <as>[:<path>][@WxH] where as is one of {}, or `off`",
PREVIEW_KINDS.join(", ")
));
}
if trimmed.eq_ignore_ascii_case("off") {
return Ok(None);
}
let (cell_raw, size) = match trimmed.rsplit_once('@') {
Some((cell, size_raw)) => (cell, Some(parse_preview_size(flag, trimmed, size_raw)?)),
None => (trimmed, None),
};
let (as_raw, path_raw) = match cell_raw.split_once(':') {
Some((a, p)) => (a, Some(p)),
None => (cell_raw, None),
};
let as_kind = as_raw.trim().to_ascii_lowercase();
if !PREVIEW_KINDS.contains(&as_kind.as_str()) {
return Err(format!(
"`{flag} {trimmed}` — unknown preview kind `{as_kind}`; use one of {}, or `off`",
PREVIEW_KINDS.join(", ")
));
}
let mut cell = Map::new();
cell.insert("as".to_string(), json!(as_kind));
if let Some(path) = path_raw.map(str::trim).filter(|p| !p.is_empty()) {
cell.insert("path".to_string(), json!(path));
}
if let Some((width, height)) = size {
cell.insert("width".to_string(), json!(width));
cell.insert("height".to_string(), json!(height));
}
Ok(Some(Value::Object(cell)))
}
pub fn set_preview_config(config: &mut Map<String, Value>, slot: &str, cell: Option<Value>) {
match cell {
Some(value) => {
let entry = config
.entry("preview".to_string())
.or_insert_with(|| Value::Object(Map::new()));
if !entry.is_object() {
*entry = Value::Object(Map::new());
}
if let Value::Object(map) = entry {
map.insert(slot.to_string(), value);
}
}
None => {
let emptied = match config.get_mut("preview") {
Some(Value::Object(map)) => {
map.remove(slot);
map.is_empty()
}
Some(_) => true,
None => false,
};
if emptied {
config.remove("preview");
}
}
}
}
pub fn preview_flag_token(config: &Value, slot: &str) -> Option<String> {
let cell = config.get("preview")?.get(slot)?;
let as_kind = cell.get("as")?.as_str()?.trim();
if as_kind.is_empty() {
return None;
}
let mut token = match cell.get("path").and_then(Value::as_str).map(str::trim) {
Some(path) if !path.is_empty() => format!("{as_kind}:{path}"),
_ => as_kind.to_string(),
};
let dimension = |key: &str| cell.get(key).and_then(Value::as_u64).filter(|n| *n > 0);
if let (Some(width), Some(height)) = (dimension("width"), dimension("height")) {
token.push_str(&format!("@{width}x{height}"));
}
Some(token)
}
pub fn parse_node_config(
tokens: &[String],
raw: &str,
dsl_flags: &[DslFlag],
) -> Result<(Value, Option<String>), String> {
parse_node_config_with_positional(tokens, raw, dsl_flags, None)
}
pub fn parse_node_config_with_positional(
tokens: &[String],
raw: &str,
dsl_flags: &[DslFlag],
positional: Option<&str>,
) -> Result<(Value, Option<String>), String> {
let mut config = serde_json::Map::new();
let mut body: Option<String> = None;
let mut i = 0;
if let Some(key) = positional {
if let Some(first) = tokens.first() {
if !first.starts_with("--") && first != "--" {
reject_unquoted_expression(key, first)?;
config.insert(key.to_string(), Value::String(first.clone()));
i = 1;
}
}
}
while i < tokens.len() {
let t = &tokens[i];
if t == "--" {
body = extract_raw_body_from(raw);
break;
}
if let Some(key) = t.strip_prefix("--") {
let flag_str = format!("--{key}");
let dsl_flag = dsl_flags
.iter()
.find(|f| f.flag == flag_str)
.ok_or_else(|| {
let names: Vec<&str> = dsl_flags.iter().map(|f| f.flag.as_str()).collect();
let hint = crate::pipeline::nodes::check::suggest::did_you_mean(&flag_str, &names)
.map(|near| format!("; did you mean {near}?"))
.unwrap_or_default();
format!("unknown flag `--{key}` — not declared in this node's dsl_flags{hint} — it takes {}", names.join(" "))
})?;
match dsl_flag.kind {
DslFlagKind::Scalar => {
let val = tokens.get(i + 1).cloned().unwrap_or_default();
reject_unquoted_expression(&flag_str, &val)?;
if let Some(slot) = preview_slot_for_config_key(&dsl_flag.config_key) {
let cell = parse_preview_flag_value(&flag_str, &val)?;
set_preview_config(&mut config, slot, cell);
} else if dsl_flag.value == "text" {
config.insert(dsl_flag.config_key.clone(), json!(val));
} else {
config.insert(dsl_flag.config_key.clone(), coerce_scalar_value(&val));
}
i += 2;
}
DslFlagKind::RepeatedList => {
let val = tokens.get(i + 1).cloned().unwrap_or_default();
reject_unquoted_expression(&flag_str, &val)?;
push_list_flag_value(&mut config, dsl_flag, &val);
i += 2;
}
DslFlagKind::Bool => {
config.insert(dsl_flag.config_key.clone(), json!(true));
i += 1;
}
DslFlagKind::KeyValuePairs => {
let raw = tokens.get(i + 1).cloned().unwrap_or_default();
if is_whole_expression(&raw) {
if config.get(&dsl_flag.config_key).is_some_and(|v| v.as_object().is_some_and(|m| !m.is_empty())) {
return Err(format!(
"`{}` takes repeated key=value pairs or one {{{{ object }}}}, not both",
dsl_flag.flag
));
}
config.insert(dsl_flag.config_key.clone(), Value::String(raw.trim().to_string()));
i += 2;
continue;
}
if config.get(&dsl_flag.config_key).is_some_and(Value::is_string) {
return Err(format!(
"`{}` takes repeated key=value pairs or one {{{{ object }}}}, not both",
dsl_flag.flag
));
}
let (k, v) = if let Some(eq) = raw.find('=') {
(raw[..eq].trim().to_string(), raw[eq + 1..].to_string())
} else {
(raw.trim().to_string(), String::new())
};
reject_unquoted_expression(&flag_str, &raw)?;
reject_unquoted_expression(&flag_str, &v)?;
let entry = config
.entry(dsl_flag.config_key.clone())
.or_insert_with(|| Value::Object(serde_json::Map::new()));
if let Value::Object(m) = entry {
insert_pair(m, k, v);
}
i += 2;
}
DslFlagKind::SchemaField => {
let spec = tokens.get(i + 1).cloned().unwrap_or_default();
if spec.trim().is_empty() {
return Err(format!(
"schema field flag `{}` requires name:type",
dsl_flag.flag
));
}
let desc = tokens.get(i + 2).filter(|v| !v.starts_with("--")).cloned();
push_schema_field(&mut config, &dsl_flag.config_key, &spec, desc.as_deref())?;
i += if desc.is_some() { 3 } else { 2 };
}
}
} else {
i += 1;
}
}
Ok((Value::Object(config), body))
}
fn push_list_flag_value(config: &mut serde_json::Map<String, Value>, flag: &DslFlag, value: &str) {
if value.trim().is_empty() {
return;
}
let entry = config
.entry(flag.config_key.clone())
.or_insert_with(|| Value::Array(Vec::new()));
if let Value::Array(existing) = entry {
existing.push(json!(value));
}
}
fn push_schema_field(
config: &mut Map<String, Value>,
config_key: &str,
spec: &str,
description: Option<&str>,
) -> Result<(), String> {
let (name, property, required) = schema_field_from_spec(spec, description)?;
let entry = config
.entry(config_key.to_string())
.or_insert_with(|| json!({ "type": "object", "properties": {}, "required": [] }));
if !entry.is_object() {
*entry = json!({ "type": "object", "properties": {}, "required": [] });
}
let schema = entry.as_object_mut().expect("schema object");
schema.insert("type".to_string(), json!("object"));
let properties = schema
.entry("properties".to_string())
.or_insert_with(|| Value::Object(Map::new()));
if !properties.is_object() {
*properties = Value::Object(Map::new());
}
properties
.as_object_mut()
.expect("properties object")
.insert(name.clone(), property);
if required {
let required_values = schema
.entry("required".to_string())
.or_insert_with(|| Value::Array(Vec::new()));
if !required_values.is_array() {
*required_values = Value::Array(Vec::new());
}
let arr = required_values.as_array_mut().expect("required array");
if !arr.iter().any(|v| v.as_str() == Some(name.as_str())) {
arr.push(json!(name));
}
}
Ok(())
}
fn schema_field_from_spec(
spec: &str,
description: Option<&str>,
) -> Result<(String, Value, bool), String> {
let (raw_name, raw_type) = spec
.split_once(':')
.ok_or_else(|| format!("schema field `{spec}` must use name:type syntax"))?;
let name = raw_name.trim();
if name.is_empty() {
return Err(format!("schema field `{spec}` has empty name"));
}
let mut type_spec = raw_type.trim();
let required = type_spec.ends_with('!');
if required {
type_spec = type_spec.trim_end_matches('!').trim();
}
let mut property = schema_property_for_type(type_spec)?;
if let Some(desc) = description.map(str::trim).filter(|s| !s.is_empty()) {
if let Some(obj) = property.as_object_mut() {
obj.insert("description".to_string(), json!(desc));
}
}
Ok((name.to_string(), property, required))
}
fn schema_property_for_type(type_spec: &str) -> Result<Value, String> {
let trimmed = type_spec.trim();
if trimmed.is_empty() {
return Err("schema field type must not be empty".to_string());
}
if let Some(item) = trimmed.strip_suffix("[]") {
return Ok(json!({
"type": "array",
"items": schema_property_for_type(item)?
}));
}
match trimmed {
"string" | "number" | "integer" | "boolean" | "object" | "array" => {
Ok(json!({ "type": trimmed }))
}
"json" | "any" => Ok(json!({})),
"file" | "bytes" | "blob" => Ok(json!({
"type": "object",
"x-zebflow-type": trimmed
})),
other => Err(format!(
"unsupported schema field type `{other}`; use string, number, integer, boolean, object, array, any, json, file, bytes, blob, or []"
)),
}
}
fn reject_unquoted_expression(flag: &str, value: &str) -> Result<(), String> {
let trimmed = value.trim();
if trimmed.starts_with("{{") && !trimmed.contains("}}") {
return Err(format!(
"`{flag} {trimmed}` looks like an unquoted expression — the value was cut at the \
first space. Quote the whole thing: {flag} \"{{{{ … }}}}\""
));
}
Ok(())
}
fn parse_flags_for_patch(tokens: &[String], cmd: &str) -> (HashMap<String, Value>, Option<String>) {
let mut flags: HashMap<String, Value> = HashMap::new();
let mut body: Option<String> = None;
let mut i = 0;
while i < tokens.len() {
let t = &tokens[i];
if t == "--" {
body = extract_raw_body_from(cmd);
break;
}
if let Some(key) = t.strip_prefix("--") {
let val = tokens.get(i + 1).cloned().unwrap_or_default();
let config_key = key.replace('-', "_");
let new_val = coerce_scalar_value(&val);
match flags.entry(config_key) {
std::collections::hash_map::Entry::Occupied(mut e) => match e.get_mut() {
Value::Array(arr) => arr.push(new_val),
existing => {
let prev = existing.clone();
*existing = Value::Array(vec![prev, new_val]);
}
},
std::collections::hash_map::Entry::Vacant(e) => {
e.insert(new_val);
}
}
i += 2;
} else {
i += 1;
}
}
(flags, body)
}
fn parse_register(tokens: &[String], cmd: &str) -> DslVerb {
let file_rel_path = tokens.get(1).cloned().unwrap_or_default();
let mut title = String::new();
let mut description = String::new();
let mut as_json = false;
let mut i = 2;
while i < tokens.len() {
match tokens[i].as_str() {
"--title" => {
title = tokens.get(i + 1).cloned().unwrap_or_default();
i += 2;
}
"--description" => {
description = tokens.get(i + 1).cloned().unwrap_or_default();
i += 2;
}
"--as-json" => {
as_json = true;
i += 1;
}
_ => break,
}
}
let body = extract_pipeline_body(cmd);
let stray: Vec<&str> = tokens[i..]
.iter()
.map(String::as_str)
.take_while(|t| !t.starts_with('|') && !t.starts_with('['))
.filter(|t| *t != "--") .collect();
if !stray.is_empty() {
return DslVerb::Invalid {
message: format!(
"register: the body must start with `|` (pipe mode) or `[label]` (graph mode); found `{}` before it. \
Write: register {file_rel_path} | {} …",
stray.join(" "),
stray.join(" ")
),
};
}
DslVerb::Register {
file_rel_path,
title,
description,
as_json,
body,
}
}
fn parse_patch(tokens: &[String], cmd: &str) -> DslVerb {
let file_rel_path = tokens.get(2).cloned().unwrap_or_default();
let target = tokens
.get(3)
.map(|t| t.to_lowercase())
.unwrap_or_else(|| "node".to_string());
let node_id = tokens.get(4).cloned().unwrap_or_default();
let flag_tokens = if tokens.len() > 5 {
tokens[5..].to_vec()
} else {
vec![]
};
let (flags, body) = parse_flags_for_patch(&flag_tokens, cmd);
DslVerb::Patch {
file_rel_path,
target,
node_id,
flags,
body,
}
}
fn extract_flag(tokens: &[String], flag: &str) -> Option<String> {
let pos = tokens.iter().position(|t| t == flag)?;
tokens.get(pos + 1).cloned()
}
fn extract_body(tokens: &[String]) -> Option<String> {
let pos = tokens.iter().position(|t| t == "--")?;
let rest = tokens[pos + 1..].join(" ");
if rest.is_empty() { None } else { Some(rest) }
}
pub fn parse_one_command(cmd: &str) -> DslVerb {
let tokens = tokenize(cmd);
if tokens.is_empty() {
return DslVerb::Unknown {
raw: cmd.to_string(),
};
}
match tokens[0].to_lowercase().as_str() {
"get" => {
let resource = tokens.get(1).cloned().unwrap_or_default().to_lowercase();
let path = extract_flag(&tokens, "--path");
let filter = extract_flag(&tokens, "--filter");
let status = extract_flag(&tokens, "--status");
DslVerb::Get { resource, path, filter, status }
}
"describe" => {
let kind = tokens.get(1).cloned().unwrap_or_default().to_lowercase();
let name = tokens.get(2).cloned().unwrap_or_default();
let compact = tokens.iter().any(|t| t == "--compact");
DslVerb::Describe { kind, name, compact }
}
"read" => {
let kind = tokens.get(1).cloned().unwrap_or_default().to_lowercase();
let name = tokens.get(2).cloned().unwrap_or_default();
DslVerb::Read { kind, name }
}
"write" | "create" => {
let kind = tokens.get(1).cloned().unwrap_or_default().to_lowercase();
let name = tokens.get(2).cloned().unwrap_or_default();
let body = extract_body(&tokens);
DslVerb::Write { kind, name, body }
}
"delete" | "rm" | "remove" => {
let kind = tokens.get(1).cloned().unwrap_or_default().to_lowercase();
let name = tokens.get(2).cloned().unwrap_or_default();
DslVerb::Delete { kind, name }
}
"activate" => {
let mut idx = 1;
if tokens.get(1).map(|s| s.as_str()) == Some("pipeline") { idx = 2; }
let file_rel_path = tokens.get(idx).cloned().unwrap_or_default();
DslVerb::Activate { file_rel_path }
}
"deactivate" => {
let mut idx = 1;
if tokens.get(1).map(|s| s.as_str()) == Some("pipeline") { idx = 2; }
let file_rel_path = tokens.get(idx).cloned().unwrap_or_default();
DslVerb::Deactivate { file_rel_path }
}
"execute" | "exec" => {
let mut idx = 1;
if tokens.get(1).map(|s| s.as_str()) == Some("pipeline") { idx = 2; }
let file_rel_path = tokens.get(idx).cloned().unwrap_or_default();
let input_str = extract_flag(&tokens, "--input").unwrap_or_default();
let input = serde_json::from_str(&input_str).unwrap_or(json!({}));
DslVerb::Execute { file_rel_path, input }
}
"register" | "reg" => parse_register(&tokens, cmd),
"check" => {
let body = extract_pipeline_body(cmd);
let at = if tokens.get(1).map(String::as_str) == Some("pipeline") { 2 } else { 1 };
let file_rel_path = tokens
.get(at)
.filter(|t| body.is_empty() && !t.starts_with('|') && !t.starts_with('['))
.cloned()
.unwrap_or_default();
DslVerb::Check { file_rel_path, body }
}
"patch" => parse_patch(&tokens, cmd),
"run" => {
let dry_run = tokens.iter().any(|t| t == "--dry-run");
let body = extract_pipeline_body(cmd);
DslVerb::Run { body, dry_run }
}
"git" => {
let subcommand = tokens.get(1).cloned().unwrap_or_default().to_lowercase();
let dash_pos = tokens.iter().position(|t| t == "--");
let args = if let Some(pos) = dash_pos {
tokens[2..pos].to_vec()
} else {
tokens[2..].to_vec()
};
let body = dash_pos.and_then(|pos| {
let b = tokens[pos + 1..].join(" ");
if b.is_empty() { None } else { Some(b) }
});
DslVerb::Git { subcommand, args, body }
}
"node" => {
let sub = tokens.get(1).cloned().unwrap_or_default().to_lowercase();
if sub == "help" {
let kind = tokens.get(2).cloned().unwrap_or_default();
DslVerb::NodeHelp { kind }
} else {
DslVerb::Unknown { raw: cmd.to_string() }
}
}
"credential" | "credentials" | "secret" | "secrets"
| "set-secret" | "set-credential" | "set-env" => {
DslVerb::CredentialBlocked {
reason: "Credential writes are blocked via DSL for security. Use the Credentials UI at /projects/{owner}/{project}/credentials".to_string(),
}
}
_ => DslVerb::Unknown { raw: cmd.to_string() },
}
}
pub fn build_pipeline_graph(id: &str, body: &str) -> Result<PipelineGraph, String> {
let all_defs = builtin_node_definitions();
build_pipeline_graph_with_definitions(id, body, &all_defs)
}
pub fn build_pipeline_graph_with_definitions(
id: &str,
body: &str,
definitions: &[NodeDefinition],
) -> Result<PipelineGraph, String> {
let body = body.trim();
if body.is_empty() {
return Err("Pipeline body is empty".to_string());
}
if body.contains('[') && body.contains("] ->") {
build_graph_mode(id, body, definitions)
} else {
build_pipe_mode(id, body, definitions)
}
}
fn split_pipe_segments(body: &str) -> Vec<&str> {
let bytes = body.as_bytes();
let mut segments = Vec::new();
let mut start = 0usize;
let mut in_single = false;
let mut in_double = false;
let mut in_backtick = false;
let mut in_opaque_body = false;
let mut escaped = false;
for (byte_pos, ch) in body.char_indices() {
if escaped {
escaped = false;
continue;
}
if in_double && !in_opaque_body && is_double_quote_escape(bytes, byte_pos) {
escaped = true;
continue;
}
if !in_opaque_body
&& !in_single
&& !in_double
&& !in_backtick
&& ch == '-'
&& byte_pos + 1 < bytes.len()
&& bytes[byte_pos + 1] == b'-'
&& is_standalone_token(bytes, byte_pos, byte_pos + 2)
{
in_opaque_body = true;
continue;
}
match ch {
'\'' if !in_opaque_body && !in_double && !in_backtick => in_single = !in_single,
'"' if !in_opaque_body && !in_single && !in_backtick => in_double = !in_double,
'`' if !in_opaque_body && !in_single && !in_double => in_backtick = !in_backtick,
'|' if !in_single
&& !in_double
&& !in_backtick
&& (!in_opaque_body || pipe_starts_node_segment(body, byte_pos)) =>
{
let seg = body[start..byte_pos].trim();
if !seg.is_empty() {
segments.push(seg);
}
start = byte_pos + ch.len_utf8();
in_opaque_body = false;
}
_ => {}
}
}
let last = body[start..].trim();
if !last.is_empty() {
segments.push(last);
}
segments
}
fn pipe_starts_node_segment(body: &str, pipe_pos: usize) -> bool {
let bytes = body.as_bytes();
if pipe_pos > 0 && bytes[pipe_pos - 1] == b'|' {
return false;
}
if pipe_pos + 1 < bytes.len() && bytes[pipe_pos + 1] == b'|' {
return false;
}
let after = body[pipe_pos + 1..].trim_start();
let Some(raw_kind) = after.split_whitespace().next() else {
return false;
};
let kind = raw_kind.trim_matches(|ch: char| ch == ';' || ch == ',' || ch == ')');
looks_like_node_kind(kind)
}
fn build_graph_mode(
id: &str,
body: &str,
definitions: &[NodeDefinition],
) -> Result<PipelineGraph, String> {
let mut nodes: Vec<PipelineNode> = Vec::new();
let mut edges: Vec<PipelineEdge> = Vec::new();
let mut notes: Vec<PipelineNote> = Vec::new();
for statement in graph_mode_statements(body) {
if is_graph_edge_statement(&statement) {
parse_graph_edge(&statement, &mut edges)?;
} else if let Some((label, tokens)) = graph_note_statement(&statement) {
if notes.iter().any(|n| n.id == label) {
return Err(format!("duplicate note id '[{label}]'"));
}
notes.push(parse_note_statement(&label, &tokens, &statement)?);
} else {
parse_graph_node(&statement, &mut nodes, definitions)?;
}
}
let to_nodes: std::collections::HashSet<&str> =
edges.iter().map(|e| e.to_node.as_str()).collect();
let entry_nodes: Vec<String> = nodes
.iter()
.filter(|n| !to_nodes.contains(n.id.as_str()))
.map(|n| n.id.clone())
.collect();
let mut graph = PipelineGraph {
id: id.to_string(),
description: None,
metadata: None,
entry_nodes,
nodes,
edges,
notes,
};
auto_tidy_pipeline_graph(&mut graph);
Ok(graph)
}
fn graph_note_statement(line: &str) -> Option<(String, Vec<String>)> {
let inner = line.strip_prefix('[')?;
let (label, rest) = inner.split_once(']')?;
let tokens = tokenize(rest.trim());
if tokens.first().map(|t| is_note_keyword(t)).unwrap_or(false) {
Some((label.trim().to_string(), tokens[1..].to_vec()))
} else {
None
}
}
pub fn body_config_key(kind: &str) -> &'static str {
match kind {
"postgres.query.run" | "sekejap.query.run" | "sqlite.query.run" | "table.query.run" => "query",
"javascript.script.run" | "typescript.script.run" => "source",
"browser.page.run" => "code",
"ai.text.generate" => "prompt",
_ => "body",
}
}
fn graph_mode_statements(body: &str) -> Vec<String> {
let joined = body.replace("\\\r\n", " ").replace("\\\n", " ");
let mut statements = Vec::new();
let mut current: Option<String> = None;
for line in joined.lines().map(str::trim).filter(|s| !s.is_empty()) {
if is_graph_structural_statement(line) {
if let Some(statement) = current.take() {
statements.push(statement);
}
current = Some(line.to_string());
} else if let Some(statement) = current.as_mut() {
statement.push('\n');
statement.push_str(line);
}
}
if let Some(statement) = current {
statements.push(statement);
}
statements
}
fn is_graph_structural_statement(line: &str) -> bool {
is_graph_edge_statement(line) || is_graph_node_statement(line)
}
fn is_graph_edge_statement(line: &str) -> bool {
let Some(arrow_pos) = line.find("->") else {
return false;
};
graph_endpoint_is_complete(&line[..arrow_pos])
&& graph_endpoint_is_complete(&line[arrow_pos + 2..])
}
fn graph_endpoint_is_complete(value: &str) -> bool {
let value = value.trim();
let Some(inner) = value.strip_prefix('[') else {
return false;
};
let Some((label, rest)) = inner.split_once(']') else {
return false;
};
if label.trim().is_empty() {
return false;
}
let rest = rest.trim();
if rest.is_empty() {
return true;
}
let Some(pin) = rest.strip_prefix(':') else {
return false;
};
let pin = pin.trim();
!pin.is_empty() && !pin.chars().any(char::is_whitespace)
}
fn is_graph_node_statement(line: &str) -> bool {
let Some(inner) = line.strip_prefix('[') else {
return false;
};
let Some((label, rest)) = inner.split_once(']') else {
return false;
};
if label.trim().is_empty() {
return false;
}
let tokens = tokenize(rest.trim());
let Some(raw_kind) = tokens.first().map(String::as_str) else {
return false;
};
looks_like_node_kind(raw_kind) || is_dotted_kind(raw_kind) || is_note_keyword(raw_kind)
}
fn is_dotted_kind(raw_kind: &str) -> bool {
raw_kind.contains('.')
&& raw_kind.split('.').all(|part| {
part.starts_with(|c: char| c.is_ascii_lowercase())
&& part.chars().all(|c| c.is_ascii_lowercase() || c.is_ascii_digit() || c == '_' || c == '-')
})
}
pub fn insert_pair(map: &mut serde_json::Map<String, Value>, key: String, value: String) {
match map.get_mut(&key) {
Some(Value::Array(items)) => items.push(Value::String(value)),
Some(existing) => {
let first = existing.take();
*existing = Value::Array(vec![first, Value::String(value)]);
}
None => {
map.insert(key, Value::String(value));
}
}
}
pub fn is_whole_expression(raw: &str) -> bool {
let raw = raw.trim();
raw.starts_with("{{") && raw.ends_with("}}")
}
fn pair_values(value: &Value) -> Vec<String> {
let one = |v: &Value| match v {
Value::String(s) => s.clone(),
other => other.to_string(),
};
match value {
Value::Array(items) => items.iter().map(one).collect(),
other => vec![one(other)],
}
}
#[cfg(test)]
mod tests {
use crate::pipeline::model::{DslFlag, DslFlagKind, NodeDefinition, PipelineGraph, PipelineNode};
use serde_json::json;
use super::{
DslVerb, build_pipeline_graph, build_pipeline_graph_with_definitions, graph_to_dsl,
node_to_segment_no_body, parse_one_command, parse_preview_flag_value,
preview_flag_token, set_preview_config, split_commands,
};
fn preview_graph_node(dsl_flags: &str) -> PipelineNode {
let dsl = format!(
"[a] trigger.manual\n[b] javascript.script.run {dsl_flags} -- return input\n\n[a] -> [b]\n"
);
let graph = build_pipeline_graph("parser-preview-test", &dsl).expect("graph");
graph
.nodes
.into_iter()
.find(|node| node.id == "b")
.expect("script node")
}
#[test]
fn preview_flags_parse_into_the_nested_preview_object() {
let node = preview_graph_node("--preview image:response.file --preview-in json:body");
assert_eq!(
node.config.get("preview"),
Some(&json!({
"out": { "as": "image", "path": "response.file" },
"in": { "as": "json", "path": "body" }
}))
);
}
#[test]
fn preview_flag_without_a_path_stores_only_the_kind() {
let node = preview_graph_node("--preview table");
assert_eq!(
node.config.get("preview"),
Some(&json!({ "out": { "as": "table" } }))
);
}
#[test]
fn preview_off_removes_that_half_and_the_object_with_the_last_one() {
let mut config = serde_json::Map::new();
set_preview_config(&mut config, "out", Some(json!({ "as": "image" })));
set_preview_config(&mut config, "in", Some(json!({ "as": "json" })));
set_preview_config(&mut config, "out", None);
assert_eq!(
config.get("preview"),
Some(&json!({ "in": { "as": "json" } }))
);
set_preview_config(&mut config, "in", None);
assert!(config.get("preview").is_none());
}
#[test]
fn preview_off_in_dsl_leaves_no_preview_key() {
let node = preview_graph_node("--preview off --preview-in off");
assert!(node.config.get("preview").is_none());
}
#[test]
fn preview_refuses_a_kind_that_is_not_on_the_list() {
let err = build_pipeline_graph(
"parser-preview-bad-kind",
"[a] trigger.manual\n[b] javascript.script.run --preview gif -- return input\n\n[a] -> [b]\n",
)
.expect_err("unknown preview kind must fail");
assert!(err.contains("unknown preview kind `gif`"), "{err}");
assert!(err.contains("image, video, audio, pdf, json, text, table, html"), "{err}");
}
#[test]
fn preview_flags_round_trip_through_graph_to_dsl() {
let dsl = "[a] trigger.manual\n[b] javascript.script.run --preview image:response.file --preview-in json:body -- return input\n\n[a] -> [b]\n";
let graph = build_pipeline_graph("parser-preview-round-trip", dsl).expect("graph");
let rendered = graph_to_dsl(&graph);
assert!(
rendered.contains("--preview image:response.file"),
"{rendered}"
);
assert!(rendered.contains("--preview-in json:body"), "{rendered}");
let reparsed = build_pipeline_graph("parser-preview-round-trip-2", &rendered)
.expect("re-parsed graph");
let script_preview = |g: &PipelineGraph| {
g.nodes
.iter()
.find(|n| n.kind == "javascript.script.run")
.expect("script node")
.config
.get("preview")
.cloned()
};
assert_eq!(script_preview(&graph), script_preview(&reparsed));
}
#[test]
fn node_to_segment_no_body_renders_the_preview_flags() {
let node = preview_graph_node("--preview table:rows");
let segment = node_to_segment_no_body(&node);
assert!(segment.contains("--preview table:rows"), "{segment}");
assert!(!segment.contains("return input"), "{segment}");
}
#[test]
fn preview_size_suffix_parses_in_every_combination() {
let cell = |value: &str| {
parse_preview_flag_value("--preview", value)
.expect("parses")
.expect("a cell, not off")
};
assert_eq!(cell("image"), json!({ "as": "image" }));
assert_eq!(cell("table:rows"), json!({ "as": "table", "path": "rows" }));
assert_eq!(
cell("image@360x240"),
json!({ "as": "image", "width": 360, "height": 240 })
);
assert_eq!(
cell("table:rows@420x180"),
json!({ "as": "table", "path": "rows", "width": 420, "height": 180 })
);
let node = preview_graph_node("--preview image@360x240 --preview-in json:body@300x90");
assert_eq!(
node.config.get("preview"),
Some(&json!({
"out": { "as": "image", "width": 360, "height": 240 },
"in": { "as": "json", "path": "body", "width": 300, "height": 90 }
}))
);
}
#[test]
fn preview_size_suffix_round_trips_and_is_omitted_at_default_size() {
let dsl = "[a] trigger.manual\n[b] javascript.script.run --preview table:rows@420x180 --preview-in json@300x90 -- return input\n\n[a] -> [b]\n";
let graph = build_pipeline_graph("parser-preview-size-round-trip", dsl).expect("graph");
let rendered = graph_to_dsl(&graph);
assert!(rendered.contains("--preview table:rows@420x180"), "{rendered}");
assert!(rendered.contains("--preview-in json@300x90"), "{rendered}");
let reparsed = build_pipeline_graph("parser-preview-size-round-trip-2", &rendered)
.expect("re-parsed graph");
let script_preview = |g: &PipelineGraph| {
g.nodes
.iter()
.find(|n| n.kind == "javascript.script.run")
.expect("script node")
.config
.get("preview")
.cloned()
};
assert_eq!(script_preview(&graph), script_preview(&reparsed));
let plain = preview_graph_node("--preview image:response.file");
assert_eq!(node_to_segment_no_body(&plain).contains('@'), false);
let half = json!({ "preview": { "out": { "as": "image", "width": 300 } } });
assert_eq!(preview_flag_token(&half, "out").as_deref(), Some("image"));
let sized = json!({ "preview": { "out": { "as": "image", "width": 300, "height": 200 } } });
assert_eq!(preview_flag_token(&sized, "out").as_deref(), Some("image@300x200"));
}
#[test]
fn preview_size_suffix_refuses_a_bad_size_and_names_the_form() {
for bad in ["image@big", "image@360", "image@0x240", "image@360x", "image@-1x5", "image@10x10", "image@9999x240"] {
let err = parse_preview_flag_value("--preview", bad).expect_err(bad);
assert!(err.contains("bad size"), "{bad}: {err}");
assert!(err.contains("<as>[:<path>]@WxH"), "{bad}: {err}");
}
let err = parse_preview_flag_value("--preview", "gif@360x240").expect_err("bad kind");
assert!(err.contains("unknown preview kind `gif`"), "{err}");
let err = build_pipeline_graph(
"parser-preview-bad-size",
"[a] trigger.manual\n[b] javascript.script.run --preview image@wide -- return input\n\n[a] -> [b]\n",
)
.expect_err("bad size must fail");
assert!(err.contains("bad size `wide`"), "{err}");
}
#[test]
fn a_positional_field_name_round_trips_through_the_dsl() {
let dsl = r#"| trigger.manual | input.text prompt --label "What should happen?" --max 200 | input.image photo --optional | input.file sheet --accept csv --accept xlsx"#;
let graph = build_pipeline_graph("parser-positional-test", dsl).expect("graph");
let text = &graph.nodes[1];
assert_eq!(text.kind, "input.text");
assert_eq!(text.config["name"], json!("prompt"));
assert_eq!(text.config["label"], json!("What should happen?"));
assert_eq!(text.config["max"], json!(200));
assert_eq!(graph.nodes[2].config["name"], json!("photo"));
assert_eq!(graph.nodes[2].config["optional"], json!(true));
assert_eq!(graph.nodes[3].config["accept"], json!(["csv", "xlsx"]));
let rendered = graph_to_dsl(&graph);
assert!(rendered.contains(r#"input.text prompt --label "What should happen?" --max 200"#), "{rendered}");
assert!(rendered.contains("input.image photo --optional"), "{rendered}");
assert!(rendered.contains("input.file sheet --accept csv --accept xlsx"), "{rendered}");
assert!(!rendered.contains("--name"), "{rendered}");
let compact = node_to_segment_no_body(text);
assert!(compact.starts_with("input.text prompt "), "{compact}");
let again = build_pipeline_graph("parser-positional-test", &rendered).expect("re-parse");
for (a, b) in graph.nodes.iter().zip(again.nodes.iter()) {
assert_eq!(a.kind, b.kind);
assert_eq!(a.config, b.config, "{}", a.kind);
}
}
#[test]
fn the_flag_form_of_a_positional_still_works_and_wins() {
let graph = build_pipeline_graph(
"parser-positional-flag",
"| trigger.manual | input.number count --name total --min 1",
)
.expect("graph");
assert_eq!(graph.nodes[1].config["name"], json!("total"));
assert_eq!(graph.nodes[1].config["min"], json!(1));
let graph = build_pipeline_graph("parser-positional-none", "| trigger.manual | javascript.script.run stray -- return input")
.expect("graph");
assert!(graph.nodes[1].config.get("name").is_none());
let err = build_pipeline_graph("parser-positional-expr", "| trigger.manual | input.text {{ input.x }}")
.expect_err("an unquoted expression is refused, not stored as `{{`");
assert!(err.contains("unquoted expression"), "{err}");
}
#[test]
fn only_a_plainly_written_number_becomes_a_number() {
for (raw, want) in [
("12", json!(12)),
("-3", json!(-3)),
("0", json!(0)),
("0.5", json!(0.5)),
("2.75", json!(2.75)),
("true", json!(true)),
("0012", json!("0012")),
("+5", json!("+5")),
("1e3", json!("1e3")),
("inf", json!("inf")),
("nan", json!("nan")),
("1.", json!("1.")),
(".5", json!(".5")),
("-", json!("-")),
] {
assert_eq!(super::coerce_scalar_value(raw), want, "{raw}");
}
}
#[test]
fn logic_match_cases_are_repeated_and_each_is_a_pin() {
let dsl = r#"
[a] trigger.manual
[b] logic.match --from "$input.type" --case create --case "New York" --case 404 --default unknown
[a] -> [b]
"#;
let graph = build_pipeline_graph("parser-match-cases-test", dsl).expect("graph");
let node = graph
.nodes
.iter()
.find(|node| node.id == "b")
.expect("match node");
assert_eq!(node.config.get("cases"), Some(&json!(["create", "New York", "404"])));
assert_eq!(node.output_pins, vec!["create", "new-york", "404", "unknown"]);
}
#[test]
fn logic_match_renders_one_case_flag_per_case() {
let dsl = "trigger.manual | logic.match --from \"input.manual.kind\" --case a --case b --default other";
let graph = build_pipeline_graph("parser-match-render-test", dsl).expect("graph");
let rendered = graph_to_dsl(&graph);
let line = rendered.lines().find(|l| l.contains("logic.match")).expect("match line");
assert!(line.contains("--case a --case b"), "{line}");
assert!(line.contains("--from") && line.contains("--default other"), "{line}");
let again = build_pipeline_graph("parser-match-render-test", &rendered).expect("re-parse");
assert_eq!(again.nodes[1].config["cases"], graph.nodes[1].config["cases"]);
assert_eq!(again.nodes[1].output_pins, vec!["a", "b", "other"]);
}
#[test]
fn a_comma_inside_a_repeated_value_stays_one_value() {
let dsl = r#"
[a] trigger.manual
[b] logic.match --from "{{ input.kind }}" --case "a,b" --case c --default other
[c] ai.text.generate --provider openai --credential c --tool lookup --tool notify -- Hello
[a] -> [b]
[b] -> [c]
"#;
let graph = build_pipeline_graph("parser-list-comma-test", dsl).expect("graph");
assert_eq!(graph.nodes[1].config["cases"], json!(["a,b", "c"]));
assert_eq!(graph.nodes[2].config["tool"], json!(["lookup", "notify"]));
let again = build_pipeline_graph("parser-list-comma-test", &graph_to_dsl(&graph)).expect("re-parse");
assert_eq!(again.nodes[1].config["cases"], json!(["a,b", "c"]));
}
#[test]
fn parameters_render_back_as_parameters() {
let dsl = r#"| trigger.mcp --route /shop --name search --parameter q:string! "The words." --parameter tags:string[] --auth none"#;
let graph = build_pipeline_graph("parser-mcp-parameters", dsl).expect("graph");
let rendered = graph_to_dsl(&graph);
assert!(rendered.contains(r#"--parameter q:string! "The words.""#), "{rendered}");
let again = build_pipeline_graph("parser-mcp-parameters", &rendered).expect("re-parse");
assert_eq!(again.nodes[0].config["schema"], graph.nodes[0].config["schema"]);
}
#[test]
fn repeated_list_reconstructs_source_binding_objects() {
let node = PipelineNode {
id: "b".to_string(),
kind: "table.query.run".to_string(),
config: json!({
"from": [
{ "source": "datasets/posts.csv", "alias": "posts" },
{ "source": "$input.rows", "alias": "rows" }
],
"query": "select * from posts",
"rows": true
}),
input_pins: vec!["in".to_string()],
output_pins: vec!["out".to_string()],
};
let segment = node_to_segment_no_body(&node);
assert!(segment.contains("--from \"datasets/posts.csv as posts\""));
assert!(segment.contains("--from \"$input.rows as rows\""));
assert!(segment.contains("--rows"));
}
#[test]
fn split_commands_preserves_graph_mode_lines_for_run() {
let dsl = r#"
run \
[a] trigger.manual \
[b] javascript.script.run -- "return { ok: true };" \
[a] -> [b]
"#;
let commands = split_commands(dsl);
assert_eq!(commands.len(), 1);
let verb = parse_one_command(&commands[0]);
match verb {
DslVerb::Run { body, dry_run } => {
assert!(!dry_run);
let graph = build_pipeline_graph("graph-run-test", &body).expect("graph");
assert_eq!(graph.nodes.len(), 2);
assert_eq!(graph.edges.len(), 1);
}
other => panic!("expected run verb, got {other:?}"),
}
}
#[test]
fn split_commands_respects_quotes_around_ampersand() {
let dsl = r#"register pipelines/test [trigger] trigger.manual
[echo] javascript.script.run -- "if (a && b) { return 1; }"
[trigger] -> [echo]"#;
let commands = split_commands(dsl);
assert_eq!(commands.len(), 1, "&& inside double quotes must not split");
let dsl2 = r#"register pipelines/test [t] trigger.manual
[s] javascript.script.run -- `${a && b}`
[t] -> [s]"#;
let commands2 = split_commands(dsl2);
assert_eq!(commands2.len(), 1, "&& inside backticks must not split");
let dsl3 = "get pipelines && activate pipelines/foo";
let commands3 = split_commands(dsl3);
assert_eq!(commands3.len(), 2, "&& outside quotes must still split");
}
#[test]
fn register_graph_body_prefers_graph_marker_before_js_pipe() {
let dsl = r#"register pipelines/e1/rename [a] trigger.manual
[b] javascript.script.run -- const left = input.old_name;
const ok = /old|new/.test(left) || left === "legacy|name";
return { ok };
[a] -> [b]"#;
let commands = split_commands(dsl);
assert_eq!(commands.len(), 1);
let verb = parse_one_command(&commands[0]);
match verb {
DslVerb::Register { body, .. } => {
assert!(
body.trim_start().starts_with("[a]"),
"graph body must start at the first graph marker, not at a later JS pipe"
);
let graph = build_pipeline_graph("register-graph-js-pipe", &body).expect("graph");
assert_eq!(graph.nodes.len(), 2);
assert_eq!(graph.edges.len(), 1);
let script = graph
.nodes
.iter()
.find(|node| node.id == "b")
.expect("script node");
let source = script
.config
.get("source")
.and_then(|value| value.as_str())
.expect("source");
assert!(source.contains("/old|new/"));
assert!(source.contains("legacy|name"));
assert!(source.contains("||"));
}
other => panic!("expected register verb, got {other:?}"),
}
}
#[test]
fn pipe_mode_long_script_body_keeps_downstream_nodes() {
let dsl = r#"
| trigger.manual
| javascript.script.run -- const ok = !!(input.left || input.right);
const label = "left|right";
return { ok, label };
| javascript.script.run -- return { downstream: input.ok, label: input.label };
"#;
let graph = build_pipeline_graph("pipe-long-script-body", dsl).expect("graph");
assert_eq!(graph.nodes.len(), 3);
assert_eq!(graph.edges.len(), 2);
assert_eq!(graph.nodes[1].kind, "javascript.script.run");
assert_eq!(graph.nodes[2].kind, "javascript.script.run");
let first_source = graph.nodes[1]
.config
.get("source")
.and_then(|value| value.as_str())
.expect("first source");
assert!(first_source.contains("input.left || input.right"));
assert!(first_source.contains("left|right"));
let second_source = graph.nodes[2]
.config
.get("source")
.and_then(|value| value.as_str())
.expect("second source");
assert!(second_source.contains("downstream"));
}
#[test]
fn graph_mode_multiline_script_body_keeps_edges() {
let dsl = r#"
[a] trigger.manual
[b] javascript.script.run -- const values = [1, 2, 3];
[1, 2, 3].map((value) => value + 1);
return { values };
[a] -> [b]
"#;
let graph = build_pipeline_graph("graph-multiline-script-body", dsl).expect("graph");
assert_eq!(graph.nodes.len(), 2);
assert_eq!(graph.edges.len(), 1);
let source = graph
.nodes
.iter()
.find(|node| node.id == "b")
.expect("script node")
.config
.get("source")
.and_then(|value| value.as_str())
.expect("source");
assert!(source.contains("[1, 2, 3].map"));
assert!(source.contains("return { values };"));
}
#[test]
fn a_graph_line_naming_a_catalogue_node_by_its_short_name_is_its_own_node() {
let embed = NodeDefinition {
kind: "ai.embedding.generate".to_string(),
title: "AI Embedding".to_string(),
description: "Embed text.".to_string(),
input_pins: vec!["in".to_string()],
output_pins: vec!["out".to_string()],
dsl_flags: vec![DslFlag {
flag: "--model".to_string(),
config_key: "model".to_string(),
description: "The embedding model.".to_string(),
kind: DslFlagKind::Scalar,
required: false,
..Default::default()
}],
..Default::default()
};
let dsl = "[a] trigger.manual\n[b] javascript.script.run -- \"return { text: ['x'] }\"\n[c] ai.embedding.generate --model text-embedding-3-small\n[a] -> [b]\n[b] -> [c]";
let graph = build_pipeline_graph_with_definitions("short-kind", dsl, &[embed]).expect("graph");
let c = graph.nodes.iter().find(|n| n.id == "c").expect("[c] is a node, not glued onto [b]'s body");
assert_eq!(c.kind, "ai.embedding.generate");
let unknown = build_pipeline_graph_with_definitions("short-kind", "[a] trigger.manual\n[b] ai.nothing\n[a] -> [b]", &[]);
assert!(unknown.unwrap_err().contains("ai.nothing"), "an unknown short kind is refused by name");
}
#[test]
fn curated_and_third_party_kinds_both_resolve_from_the_catalog() {
let curated = NodeDefinition {
kind: "telegram.message.send".to_string(),
title: "Telegram Send".to_string(),
description: "Send a message.".to_string(),
input_pins: vec!["in".to_string()],
output_pins: vec!["out".to_string()],
..Default::default()
};
let third_party = NodeDefinition {
kind: "x.acme.thing.run".to_string(),
title: "Acme Thing".to_string(),
description: "Do the thing.".to_string(),
input_pins: vec!["in".to_string()],
output_pins: vec!["out".to_string()],
..Default::default()
};
let definitions = vec![curated, third_party];
assert_eq!(
super::resolve_catalog_kind("telegram.message.send", &definitions),
Some("telegram.message.send".to_string())
);
assert_eq!(
super::resolve_catalog_kind("n.telegram.message.send", &definitions),
None,
"a kind has one name: the retired n. prefix is not an alias"
);
assert_eq!(
super::resolve_catalog_kind("x.acme.thing.run", &definitions),
Some("x.acme.thing.run".to_string())
);
assert_eq!(
super::resolve_catalog_kind("telegramm.message.send", &definitions),
None,
"a kind the catalog does not have is still rejected"
);
}
#[test]
fn registry_definitions_parse_composite_dsl_flags() {
let mut definitions = crate::pipeline::nodes::builtin_node_definitions();
definitions.push(NodeDefinition {
kind: "x.openai_embedding.embedding.generate".to_string(),
title: "AI Embedding".to_string(),
description: "Composite embedding node.".to_string(),
input_pins: vec!["in".to_string()],
output_pins: vec!["out".to_string(), "error".to_string()],
dsl_flags: vec![
DslFlag {
flag: "--credential".to_string(),
config_key: "credential_id".to_string(),
description: "OpenAI-compatible credential.".to_string(),
kind: DslFlagKind::Scalar,
required: true,
..Default::default()
},
DslFlag {
flag: "--model".to_string(),
config_key: "model".to_string(),
description: "Embedding model.".to_string(),
kind: DslFlagKind::Scalar,
required: false,
..Default::default()
},
DslFlag {
flag: "--text".to_string(),
config_key: "text".to_string(),
description: "The text to embed.".to_string(),
kind: DslFlagKind::Scalar,
required: false,
..Default::default()
},
],
..Default::default()
});
let graph = build_pipeline_graph_with_definitions(
"composite-embedding-dsl",
r#"
| trigger.manual
| x.openai_embedding.embedding.generate --credential qwen-embed --model text-embedding-v4 --text hello
"#,
&definitions,
)
.expect("graph");
let node = graph
.nodes
.iter()
.find(|node| node.kind == "x.openai_embedding.embedding.generate")
.expect("composite node");
assert_eq!(node.config["credential_id"], json!("qwen-embed"));
assert_eq!(node.config["model"], json!("text-embedding-v4"));
assert_eq!(node.config["text"], json!("hello"));
}
#[test]
fn registry_definitions_parse_wasm_node_dsl_flags() {
let mut definitions = crate::pipeline::nodes::builtin_node_definitions();
definitions.push(NodeDefinition {
kind: "x.wasmpkg.number.add".to_string(),
title: "WASM Test Add".to_string(),
description: "WASM test node.".to_string(),
input_pins: vec!["in".to_string()],
output_pins: vec!["out".to_string(), "error".to_string()],
dsl_flags: vec![
DslFlag {
flag: "--a".to_string(),
config_key: "a".to_string(),
description: "Left integer operand.".to_string(),
kind: DslFlagKind::Scalar,
required: false,
..Default::default()
},
DslFlag {
flag: "--b".to_string(),
config_key: "b".to_string(),
description: "Right integer operand.".to_string(),
kind: DslFlagKind::Scalar,
required: false,
..Default::default()
},
],
..Default::default()
});
let graph = build_pipeline_graph_with_definitions(
"wasm-test-dsl",
r#"
| trigger.manual
| x.wasmpkg.number.add --a 2 --b 3
"#,
&definitions,
)
.expect("graph");
let node = graph
.nodes
.iter()
.find(|node| node.kind == "x.wasmpkg.number.add")
.expect("wasm node");
assert_eq!(node.config["a"], json!(2));
assert_eq!(node.config["b"], json!(3));
}
#[test]
fn patch_script_body_is_opaque_to_shell_separators() {
let source = r#"const lat = row.Stop_lat ?? row.stop_lat;
const lon = row.Stop_long ?? row.stop_long;
const ok = !!((lat || lon) && /Stop_lat|Stop_long/.test("Stop_lat|Stop_long"));
return { ok, label: "lat|lon", pair: `${lat || ""}|${lon || ""}` };"#;
let dsl = format!(
"patch pipeline pipelines/demo/functions/e1/ingestion-tools/e1-tool-ing-transform-tabular-spatial.zf.json node n1 -- {source}"
);
let commands = split_commands(&dsl);
assert_eq!(
commands.len(),
1,
"operators inside a patch body must not become command separators"
);
let verb = parse_one_command(&commands[0]);
match verb {
DslVerb::Patch {
file_rel_path,
node_id,
body,
..
} => {
assert_eq!(
file_rel_path,
"pipelines/demo/functions/e1/ingestion-tools/e1-tool-ing-transform-tabular-spatial.zf.json"
);
assert_eq!(node_id, "n1");
let body = body.expect("patch body");
assert!(body.contains("lat || lon"));
assert!(body.contains("/Stop_lat|Stop_long/"));
assert!(body.contains("\"lat|lon\""));
assert!(body.contains("`${lat || \"\"}|${lon || \"\"}`"));
}
other => panic!("expected patch verb, got {other:?}"),
}
}
#[test]
fn function_trigger_schema_field_flags_build_json_schema() {
let dsl = r#"
[fn] trigger.function --title "Inspect CSV" --description "Reads a CSV." --parameter source:file! "CSV file reference." --parameter options:any "Provider options." --result ok:boolean! "Whether it worked." --result columns:string[] "Detected columns."
[fn] -> [done]
[done] javascript.script.run -- return input;
"#;
let graph = build_pipeline_graph("function-schema-field-test", dsl).expect("graph");
let node = graph.nodes.iter().find(|node| node.id == "fn").expect("fn");
assert_eq!(node.config["description"], json!("Reads a CSV."));
assert_eq!(node.config["schema"]["type"], json!("object"));
assert_eq!(node.config["schema"]["required"], json!(["source"]));
assert_eq!(
node.config["schema"]["properties"]["source"]["x-zebflow-type"],
json!("file")
);
assert_eq!(
node.config["schema"]["properties"]["options"],
json!({"description": "Provider options."})
);
assert_eq!(node.config["result_schema"]["required"], json!(["ok"]));
assert_eq!(
node.config["result_schema"]["properties"]["columns"]["items"]["type"],
json!("string")
);
let segment = super::node_to_segment_no_body(node);
assert!(segment.contains("--schema ") && segment.contains("--result-schema "), "{segment}");
assert!(!segment.contains("--parameter") && !segment.contains("--result "), "{segment}");
}
}
fn parse_graph_node(
line: &str,
nodes: &mut Vec<PipelineNode>,
definitions: &[NodeDefinition],
) -> Result<(), String> {
let rest = line
.strip_prefix('[')
.ok_or("expected '[' at start of node declaration")?;
let (label, rest) = rest
.split_once(']')
.ok_or("expected ']' in node declaration")?;
let label = label.trim();
if label.is_empty() {
return Err("node label must not be empty".to_string());
}
let rest = rest.trim();
let tokens = tokenize(rest);
if tokens.is_empty() {
return Err(format!("node '[{label}]' has no kind"));
}
let raw_kind = &tokens[0];
let custom_kind: String;
let full_kind: &str = match resolve_catalog_kind(raw_kind, definitions) {
Some(kind) => {
custom_kind = kind;
custom_kind.as_str()
}
None => return Err(unknown_kind_message(raw_kind, definitions)),
};
let (input_pins, mut output_pins) = default_pins(full_kind);
let definition = definitions.iter().find(|d| d.kind == full_kind);
let dsl_flags = definition.map(|d| d.dsl_flags.as_slice()).unwrap_or(&[]);
let positional = definition.and_then(|d| d.positional.as_deref());
let (mut config, body_val) =
parse_node_config_with_positional(&tokens[1..], rest, dsl_flags, positional)?;
if let Some(bval) = body_val {
if let Value::Object(ref mut map) = config {
map.insert(body_config_key(full_kind).to_string(), json!(bval));
}
}
if full_kind == "logic.match" {
output_pins = crate::pipeline::nodes::basic::logic::match_::output_pins(&config);
}
nodes.push(PipelineNode {
id: label.to_string(),
kind: full_kind.to_string(),
input_pins,
output_pins,
config,
});
Ok(())
}
fn parse_graph_edge(line: &str, edges: &mut Vec<PipelineEdge>) -> Result<(), String> {
let arrow_pos = line.find("->").ok_or("expected '->' in edge declaration")?;
let from_part = line[..arrow_pos].trim();
let to_part = line[arrow_pos + 2..].trim();
let (from_node, from_pin) = parse_node_pin_part(from_part, "out")?;
let (to_node, to_pin) = parse_node_pin_part(to_part, "in")?;
edges.push(PipelineEdge {
from_node,
from_pin,
to_node,
to_pin,
});
Ok(())
}
fn parse_node_pin_part(s: &str, default_pin: &str) -> Result<(String, String), String> {
let s = s.trim();
let inner = s
.strip_prefix('[')
.ok_or_else(|| format!("expected '[' in edge endpoint: '{s}'"))?;
let (label, rest) = inner
.split_once(']')
.ok_or_else(|| format!("expected ']' in edge endpoint: '{s}'"))?;
let label = label.trim().to_string();
let pin = rest
.trim()
.strip_prefix(':')
.map(|p| p.trim().to_string())
.unwrap_or_else(|| default_pin.to_string());
Ok((label, pin))
}
pub fn graph_to_dsl(graph: &PipelineGraph) -> String {
if is_linear_graph(graph) {
graph_to_pipe_mode(graph)
} else {
graph_to_graph_mode(graph)
}
}
fn is_linear_graph(graph: &PipelineGraph) -> bool {
let mut in_count: HashMap<&str, usize> = HashMap::new();
let mut out_count: HashMap<&str, usize> = HashMap::new();
for edge in &graph.edges {
if edge.from_pin != "out" || edge.to_pin != "in" {
return false;
}
*in_count.entry(edge.to_node.as_str()).or_insert(0) += 1;
*out_count.entry(edge.from_node.as_str()).or_insert(0) += 1;
}
graph.nodes.iter().all(|n| {
*in_count.get(n.id.as_str()).unwrap_or(&0) <= 1
&& *out_count.get(n.id.as_str()).unwrap_or(&0) <= 1
})
}
fn node_to_segment(node: &PipelineNode) -> String {
let all_defs = builtin_node_definitions();
let definition = all_defs.iter().find(|d| d.kind == node.kind);
let dsl_flags = definition.map(|d| d.dsl_flags.as_slice()).unwrap_or(&[]);
let mut parts = vec![node.kind.clone()];
let positional = positional_segment_token(definition, &node.config);
if let Some((_, token)) = &positional {
parts.push(token.clone());
}
let body_key = body_config_key(&node.kind);
for flag in dsl_flags {
if flag.config_key == body_key {
continue;
}
if positional.as_ref().is_some_and(|(key, _)| *key == flag.config_key) {
continue;
}
if let Some(slot) = preview_slot_for_config_key(&flag.config_key) {
if let Some(token) = preview_flag_token(&node.config, slot) {
parts.push(flag.flag.clone());
parts.push(token);
}
continue;
}
let Some(val) = node.config.get(&flag.config_key) else {
continue;
};
match &flag.kind {
DslFlagKind::Bool => {
if val.as_bool().unwrap_or(false) {
parts.push(flag.flag.clone());
}
}
DslFlagKind::RepeatedList => {
if let Some(arr) = val.as_array() {
for item in arr.iter().filter_map(repeated_list_item_to_string) {
if item.is_empty() {
continue;
}
parts.push(flag.flag.clone());
parts.push(quote_dsl_arg(&item));
}
}
}
DslFlagKind::Scalar => {
let s = match val {
Value::String(s) if !s.is_empty() => s.clone(),
Value::Number(n) => n.to_string(),
Value::Bool(b) => b.to_string(),
_ => continue,
};
parts.push(flag.flag.clone());
parts.push(quote_dsl_arg(&s));
}
DslFlagKind::KeyValuePairs => {
if let Some(whole) = val.as_str().filter(|raw| is_whole_expression(raw)) {
parts.push(flag.flag.clone());
parts.push(quote_dsl_arg(whole));
} else if let Some(map) = val.as_object() {
for (k, v) in map {
for v_str in pair_values(v) {
parts.push(flag.flag.clone());
parts.push(quote_dsl_arg(&format!("{}={}", k, v_str)));
}
}
}
}
DslFlagKind::SchemaField => parts.extend(schema_field_tokens(dsl_flags, flag, val)),
}
}
if let Some(body) = node.config.get(body_key).and_then(|v| v.as_str()) {
let body = body.trim();
if !body.is_empty() {
let inline: String = body
.lines()
.map(str::trim)
.filter(|l| !l.is_empty())
.collect::<Vec<_>>()
.join(" ");
parts.push("--".to_string());
parts.push(inline);
}
}
parts.join(" ")
}
pub fn node_to_segment_no_body(node: &PipelineNode) -> String {
let all_defs = builtin_node_definitions();
let definition = all_defs.iter().find(|d| d.kind == node.kind);
let dsl_flags = definition.map(|d| d.dsl_flags.as_slice()).unwrap_or(&[]);
let mut parts = vec![node.kind.clone()];
let positional = positional_segment_token(definition, &node.config);
if let Some((_, token)) = &positional {
parts.push(token.clone());
}
for flag in dsl_flags {
if positional.as_ref().is_some_and(|(key, _)| *key == flag.config_key) {
continue;
}
let body_key = match node.kind.as_str() {
"postgres.query.run" => "query",
"sekejap.query.run" => "query",
"sqlite.query.run" => "query",
"table.query.run" => "query",
"javascript.script.run" | "typescript.script.run" => "source",
_ => "body",
};
if flag.config_key == body_key {
continue;
}
if let Some(slot) = preview_slot_for_config_key(&flag.config_key) {
if let Some(token) = preview_flag_token(&node.config, slot) {
parts.push(flag.flag.clone());
parts.push(token);
}
continue;
}
let Some(val) = node.config.get(&flag.config_key) else {
continue;
};
match &flag.kind {
DslFlagKind::Bool => {
if val.as_bool().unwrap_or(false) {
parts.push(flag.flag.clone());
}
}
DslFlagKind::RepeatedList => {
if let Some(arr) = val.as_array() {
for item in arr.iter().filter_map(repeated_list_item_to_string) {
if item.is_empty() {
continue;
}
parts.push(flag.flag.clone());
parts.push(quote_dsl_arg(&item));
}
}
}
DslFlagKind::Scalar => {
let s = match val {
Value::String(s) if !s.is_empty() => s.clone(),
Value::Number(n) => n.to_string(),
Value::Bool(b) => b.to_string(),
_ => continue,
};
parts.push(flag.flag.clone());
parts.push(quote_dsl_arg(&s));
}
DslFlagKind::KeyValuePairs => {
if let Some(whole) = val.as_str().filter(|raw| is_whole_expression(raw)) {
parts.push(flag.flag.clone());
parts.push(quote_dsl_arg(whole));
} else if let Some(map) = val.as_object() {
for (k, v) in map {
for v_str in pair_values(v) {
parts.push(flag.flag.clone());
parts.push(quote_dsl_arg(&format!("{}={}", k, v_str)));
}
}
}
}
DslFlagKind::SchemaField => parts.extend(schema_field_tokens(dsl_flags, flag, val)),
}
}
parts.join(" ")
}
fn schema_field_tokens(dsl_flags: &[DslFlag], flag: &DslFlag, schema: &Value) -> Vec<String> {
let has_scalar = dsl_flags
.iter()
.any(|other| other.config_key == flag.config_key && other.kind == DslFlagKind::Scalar);
if !has_scalar && let Some(fields) = schema_field_specs(schema) {
let mut parts = Vec::new();
for (spec, description) in fields {
parts.push(flag.flag.clone());
parts.push(quote_dsl_arg(&spec));
if let Some(description) = description {
parts.push(quote_dsl_arg(&description));
}
}
return parts;
}
match serde_json::to_string(schema) {
Ok(json) => vec![schema_flag_for(dsl_flags, flag), quote_dsl_arg(&json)],
Err(_) => Vec::new(),
}
}
fn schema_field_specs(schema: &Value) -> Option<Vec<(String, Option<String>)>> {
fn type_word(property: &Value) -> Option<String> {
let obj = property.as_object()?;
if obj.keys().any(|k| !matches!(k.as_str(), "type" | "description" | "items" | "x-zebflow-type")) {
return None;
}
if let Some(word) = obj.get("x-zebflow-type").and_then(Value::as_str) {
return Some(word.to_string());
}
match obj.get("type").and_then(Value::as_str) {
None => Some("any".to_string()),
Some("array") => match obj.get("items") {
Some(items) => Some(format!("{}[]", type_word(items)?)),
None => Some("array".to_string()),
},
Some(word) => Some(word.to_string()),
}
}
let properties = schema.get("properties")?.as_object()?;
let required: Vec<&str> = schema
.get("required")
.and_then(Value::as_array)
.map(|names| names.iter().filter_map(Value::as_str).collect())
.unwrap_or_default();
properties
.iter()
.map(|(name, property)| {
let word = type_word(property)?;
let bang = if required.contains(&name.as_str()) { "!" } else { "" };
let description = property.get("description").and_then(Value::as_str).map(str::to_string);
Some((format!("{name}:{word}{bang}"), description))
})
.collect()
}
fn schema_flag_for(dsl_flags: &[DslFlag], field_flag: &DslFlag) -> String {
dsl_flags
.iter()
.find(|other| other.config_key == field_flag.config_key && other.kind == DslFlagKind::Scalar)
.map(|other| other.flag.clone())
.unwrap_or_else(|| field_flag.flag.clone())
}
fn positional_segment_token(
definition: Option<&NodeDefinition>,
config: &Value,
) -> Option<(String, String)> {
let key = definition?.positional.as_deref()?;
let value = config.get(key)?.as_str()?.trim();
if value.is_empty() {
return None;
}
Some((key.to_string(), quote_dsl_arg(value)))
}
fn quote_dsl_arg(value: &str) -> String {
if value.chars().any(char::is_whitespace) || value.contains('"') || value.contains('\\') {
format!("\"{}\"", value.replace('\\', "\\\\").replace('"', "\\\""))
} else {
value.to_string()
}
}
fn repeated_list_item_to_string(value: &Value) -> Option<String> {
if let Some(s) = value.as_str() {
return Some(s.to_string());
}
let obj = value.as_object()?;
let source = obj.get("source").and_then(Value::as_str)?.trim();
let alias = obj.get("alias").and_then(Value::as_str)?.trim();
if source.is_empty() || alias.is_empty() {
return None;
}
Some(format!("{source} as {alias}"))
}
fn graph_to_pipe_mode(graph: &PipelineGraph) -> String {
let node_map: HashMap<&str, &PipelineNode> =
graph.nodes.iter().map(|n| (n.id.as_str(), n)).collect();
let next_map: HashMap<&str, &str> = graph
.edges
.iter()
.map(|e| (e.from_node.as_str(), e.to_node.as_str()))
.collect();
let mut ordered: Vec<&PipelineNode> = Vec::new();
if let Some(first_id) = graph.entry_nodes.first() {
let mut cur = first_id.as_str();
let mut visited = std::collections::HashSet::new();
loop {
if !visited.insert(cur) {
break; }
if let Some(&node) = node_map.get(cur) {
ordered.push(node);
}
match next_map.get(cur) {
Some(&next) => cur = next,
None => break,
}
}
}
let mut lines = ordered
.iter()
.map(|n| format!("| {}", node_to_segment(n)))
.collect::<Vec<_>>();
for note in &graph.notes {
lines.push(format!("| {}", note_to_statement(note, true)));
}
lines.join("\n")
}
fn graph_to_graph_mode(graph: &PipelineGraph) -> String {
let mut lines: Vec<String> = Vec::new();
for node in &graph.nodes {
lines.push(format!("[{}] {}", node.id, node_to_segment(node)));
}
if !graph.edges.is_empty() {
lines.push(String::new());
for edge in &graph.edges {
if edge.from_pin == "out" && edge.to_pin == "in" {
lines.push(format!("[{}] -> [{}]", edge.from_node, edge.to_node));
} else {
lines.push(format!(
"[{}]:{} -> [{}]:{}",
edge.from_node, edge.from_pin, edge.to_node, edge.to_pin
));
}
}
}
if !graph.notes.is_empty() {
lines.push(String::new());
for note in &graph.notes {
lines.push(format!("[{}] {}", note.id, note_to_statement(note, false)));
}
}
lines.join("\n")
}
fn build_pipe_mode(
id: &str,
body: &str,
definitions: &[NodeDefinition],
) -> Result<PipelineGraph, String> {
let body = body.trim_start_matches('|').trim();
let all_segments: Vec<&str> = split_pipe_segments(body);
let mut notes: Vec<PipelineNote> = Vec::new();
let mut segments: Vec<&str> = Vec::new();
for segment in all_segments {
let tokens = tokenize(segment);
if tokens.first().map(|t| is_note_keyword(t)).unwrap_or(false) {
let default_id = format!("note{}", notes.len() + 1);
let note = parse_note_statement(&default_id, &tokens[1..], segment)?;
if notes.iter().any(|n| n.id == note.id) {
return Err(format!("duplicate note id '{}'", note.id));
}
notes.push(note);
} else {
segments.push(segment);
}
}
if segments.is_empty() {
return Err("No nodes in pipeline body".to_string());
}
let mut nodes: Vec<PipelineNode> = Vec::new();
let mut edges: Vec<PipelineEdge> = Vec::new();
let first_tokens = tokenize(segments[0]);
let first_raw_kind = first_tokens.first().map(|s| s.as_str()).unwrap_or("");
let first_full_kind = expand_kind(first_raw_kind).unwrap_or(first_raw_kind);
let has_trigger_first = first_full_kind.starts_with("trigger.")
|| definitions.iter().any(|d| d.kind == first_full_kind && d.input_pins.is_empty());
if !has_trigger_first {
nodes.push(PipelineNode {
id: "trigger".to_string(),
kind: "trigger.manual".to_string(),
input_pins: vec![],
output_pins: vec!["out".to_string()],
config: json!({}),
});
}
for (idx, segment) in segments.iter().enumerate() {
let seg_tokens = tokenize(segment);
if seg_tokens.is_empty() {
continue;
}
let raw_kind = &seg_tokens[0];
let custom_kind: String;
let full_kind: &str = match resolve_catalog_kind(raw_kind, definitions) {
Some(kind) => {
custom_kind = kind;
custom_kind.as_str()
}
None => return Err(unknown_kind_message(raw_kind, definitions)),
};
let node_id = format!("n{idx}");
let (input_pins, mut output_pins) = default_pins(full_kind);
let definition = definitions.iter().find(|d| d.kind == full_kind);
let dsl_flags = definition.map(|d| d.dsl_flags.as_slice()).unwrap_or(&[]);
let positional = definition.and_then(|d| d.positional.as_deref());
let (mut config, body_val) =
parse_node_config_with_positional(&seg_tokens[1..], segment, dsl_flags, positional)?;
if let Some(bval) = body_val {
if let Value::Object(ref mut map) = config {
map.insert(body_config_key(full_kind).to_string(), json!(bval));
}
}
if full_kind == "logic.match" {
output_pins = crate::pipeline::nodes::basic::logic::match_::output_pins(&config);
}
let from_node = if idx == 0 {
if !has_trigger_first {
Some("trigger".to_string())
} else {
None }
} else {
Some(format!("n{}", idx - 1))
};
if let Some(from_id) = from_node {
edges.push(PipelineEdge {
from_node: from_id,
from_pin: "out".to_string(),
to_node: node_id.clone(),
to_pin: "in".to_string(),
});
}
nodes.push(PipelineNode {
id: node_id,
kind: full_kind.to_string(),
input_pins,
output_pins,
config,
});
}
let entry_nodes = nodes
.first()
.map(|n| vec![n.id.clone()])
.unwrap_or_default();
let mut graph = PipelineGraph {
id: id.to_string(),
description: None,
metadata: None,
entry_nodes,
nodes,
edges,
notes,
};
auto_tidy_pipeline_graph(&mut graph);
Ok(graph)
}
#[cfg(test)]
mod escape_tests {
use super::*;
#[test]
fn an_escaped_quote_stays_inside_its_value() {
assert_eq!(
tokenize(r#"web.response.send --header "Content-Disposition=attachment; filename=\"a b.pdf\"" --body ok"#),
vec!["web.response.send", "--header", r#"Content-Disposition=attachment; filename="a b.pdf""#, "--body", "ok"]
);
assert_eq!(tokenize(r#"--pattern "\d+\.\w" --dir "C:\data" --two "a\\b""#), vec!["--pattern", r"\d+\.\w", "--dir", r"C:\data", "--two", r"a\b"]);
assert_eq!(tokenize(r#"--dir "C:\\data\\" --next x"#), vec!["--dir", r"C:\data\", "--next", "x"]);
assert_eq!(tokenize(r#"--text 'say \"hi\"'"#), vec!["--text", r#"say \"hi\""#], "single quotes are literal");
assert_eq!(tokenize(r#"--when "input.a == \"b\" && input.c""#), vec!["--when", r#"input.a == "b" && input.c"#]);
}
#[test]
fn an_escaped_quote_does_not_end_the_string_for_the_scanners() {
let body = r#"| trigger.webhook --route /dl | web.response.send --header "X-Note=a \"|\" b" --body ok"#;
assert_eq!(split_pipe_segments(body).len(), 2, "{:?}", split_pipe_segments(body));
assert_eq!(split_commands(r#"activate pipeline "a \"&&\" b" && activate pipeline c"#).len(), 2);
assert_eq!(find_first_pipe_in_raw(r#"register a --title "a \"|\" b" | x"#), Some(31));
}
#[test]
fn a_quoted_value_round_trips() {
for value in [r#"attachment; filename="a b.pdf""#, r"C:\data\", r"\d+", r#"say \"x\""#, "plain", r#"a"b"#] {
let token = quote_dsl_arg(value);
assert_eq!(tokenize(&format!("--v {token}")), vec!["--v".to_string(), value.to_string()], "{value} as {token}");
}
}
}
#[cfg(test)]
mod register_shape_tests {
use super::*;
#[test]
fn a_body_without_a_leading_pipe_is_refused_not_truncated() {
match parse_one_command("register api/x trigger.webhook --route /x --method POST | ai.text.generate --credential c") {
DslVerb::Invalid { message } => {
assert!(message.contains("must start with `|`"), "{message}");
assert!(message.contains("trigger.webhook --route /x --method POST"), "{message}");
}
other => panic!("expected a refusal, got {other:?}"),
}
match parse_one_command("register api/x --title \"T\" -- | trigger.webhook --route /x | ai.text.generate --credential c") {
DslVerb::Register { body, title, .. } => {
assert_eq!(title, "T");
assert!(body.starts_with("| trigger.webhook"), "{body}");
}
other => panic!("{other:?}"),
}
match parse_one_command("register api/x\n[t] trigger.webhook --route /x\n[a] ai.text.generate --credential c\n[t] -> [a]") {
DslVerb::Register { body, .. } => assert!(body.starts_with("[t]"), "{body}"),
other => panic!("{other:?}"),
}
}
#[test]
fn a_body_key_that_is_also_a_flag_is_rendered_once() {
let g = build_pipeline_graph("p", "trigger.webhook --route /x | logic.if --when \"$trigger.query.who == 'm'\" | ai.text.generate --provider openai --credential c --answer-only -- Classify: {{ $trigger.body.review }}").expect("parse");
let dsl = graph_to_dsl(&g);
let if_line = dsl.lines().find(|l| l.contains("logic.if")).unwrap();
assert_eq!(if_line.matches("$trigger.query.who").count(), 1, "{if_line}");
assert!(if_line.contains("--when") && !if_line.contains(" -- "), "{if_line}");
let agent_line = dsl.lines().find(|l| l.contains("ai.text.generate")).unwrap();
assert_eq!(agent_line.matches("Classify:").count(), 1, "{agent_line}");
assert!(!agent_line.contains("--prompt"), "{agent_line}");
let again = build_pipeline_graph("p", &dsl).expect("re-parse");
assert_eq!(again.nodes[1].config.get("when"), g.nodes[1].config.get("when"));
assert_eq!(again.nodes[2].config.get("prompt"), g.nodes[2].config.get("prompt"));
}
#[test]
fn a_repeated_map_key_keeps_every_value_and_round_trips() {
let g = build_pipeline_graph(
"p",
"trigger.webhook --route /x | web.response.send --status 303 --header \"Location=/home\" --header \"Set-Cookie=a=1; Max-Age=60\" --header \"Set-Cookie=b=2\"",
)
.expect("parse");
let headers = &g.nodes[1].config["headers"];
assert_eq!(headers["Location"], "/home");
assert_eq!(headers["Set-Cookie"], serde_json::json!(["a=1; Max-Age=60", "b=2"]));
let dsl = graph_to_dsl(&g);
assert_eq!(dsl.matches("--header").count(), 3, "{dsl}");
let again = build_pipeline_graph("p", &dsl).expect("re-parse");
assert_eq!(&again.nodes[1].config["headers"], headers);
}
}
#[cfg(test)]
mod note_tests {
use super::*;
#[test]
fn graph_mode_note_is_a_note_not_a_node() {
let body = "[t] trigger.webhook --route /x\n[r] web.response.send --template pages/x.tsx\n[t] -> [r]\n[why] note --text \"Create the `smtp` credential first.\" --at 120,40 --size 320x140 --color amber";
let graph = build_pipeline_graph("p", body).expect("parse");
assert_eq!(graph.nodes.len(), 2);
assert_eq!(graph.edges.len(), 1);
assert_eq!(graph.notes.len(), 1);
let note = &graph.notes[0];
assert_eq!(note.id, "why");
assert_eq!(note.text, "Create the `smtp` credential first.");
assert_eq!((note.x, note.y, note.width, note.height), (120.0, 40.0, 320.0, 140.0));
assert_eq!(note.color, "amber");
}
#[test]
fn pipe_mode_note_does_not_break_the_chain() {
let body = "trigger.webhook --route /x | note --text \"first\" | web.response.send --template pages/x.tsx | note --id tail -- a longer text, after the flags";
let graph = build_pipeline_graph("p", body).expect("parse");
assert_eq!(graph.nodes.iter().map(|n| n.id.as_str()).collect::<Vec<_>>(), vec!["n0", "n1"]);
assert_eq!(graph.edges.len(), 1);
assert_eq!(graph.edges[0].from_node, "n0");
assert_eq!(graph.edges[0].to_node, "n1");
assert_eq!(graph.notes.len(), 2);
assert_eq!(graph.notes[0].id, "note1");
assert_eq!(graph.notes[1].id, "tail");
assert_eq!(graph.notes[1].text, "a longer text, after the flags");
}
#[test]
fn notes_survive_the_dsl_round_trip_in_both_modes() {
for body in [
"trigger.webhook --route /x | web.response.send --template pages/x.tsx | note --id n1 --text \"keep me\" --at 10,20 --size 200x80 --color blue",
"[t] trigger.webhook --route /x\n[a] logic.if --when \"true\"\n[b] web.response.send --template pages/a.tsx\n[c] web.response.send --template pages/b.tsx\n[t] -> [a]\n[a]:then -> [b]\n[a]:else -> [c]\n[n1] note --text \"keep me\" --color blue",
] {
let first = build_pipeline_graph("p", body).expect("parse");
let rendered = graph_to_dsl(&first);
assert!(rendered.contains("note"), "{rendered}");
let second = build_pipeline_graph("p", &rendered).expect("re-parse:\n{rendered}");
assert_eq!(second.notes, first.notes, "{rendered}");
}
}
#[test]
fn bad_note_flags_are_refused_with_the_flag_list() {
let err = build_pipeline_graph("p", "trigger.webhook --route /x | note --colour red").unwrap_err();
assert!(err.contains("unknown flag `--colour`"), "{err}");
let err = build_pipeline_graph("p", "trigger.webhook --route /x | note --at 12").unwrap_err();
assert!(err.contains("--at expects x,y"), "{err}");
let err = build_pipeline_graph("p", "[t] trigger.webhook --route /x\n[r] web.response.send --template pages/x.tsx\n[t] -> [r]\n[a] note --text one\n[a] note --text two").unwrap_err();
assert!(err.contains("duplicate note id"), "{err}");
}
#[test]
fn patch_verb_carries_its_target() {
match parse_one_command("patch pipeline api/x note why --text \"new\" --at 1,2") {
DslVerb::Patch { target, node_id, flags, .. } => {
assert_eq!(target, "note");
assert_eq!(node_id, "why");
assert_eq!(flags.get("text").and_then(|v| v.as_str()), Some("new"));
}
other => panic!("{other:?}"),
}
match parse_one_command("patch pipeline api/x node n1 --path /y") {
DslVerb::Patch { target, .. } => assert_eq!(target, "node"),
other => panic!("{other:?}"),
}
}
}
#[cfg(test)]
mod quoting_tests {
use super::*;
#[test]
fn every_catalogue_kind_is_reachable_as_written() {
let mut unreachable = Vec::new();
for definition in crate::pipeline::nodes::builtin_node_definitions() {
let kind = definition.kind.clone();
match expand_kind(&kind) {
Some(resolved) if resolved == kind => {}
other => unreachable.push(format!("{kind} (resolved to {other:?})")),
}
}
assert!(
unreachable.is_empty(),
"these kinds do not resolve to themselves in expand_kind:\n {}",
unreachable.join("\n ")
);
}
const COMPLEX: &str = r#"
[t] trigger.webhook --route /orders --method POST
[k] kv.entry.get --key "order:{{ $trigger.query.id }}"
[c] logic.if --when "input.entry.found"
[m] mail.message.send --credential relay --recipient "{{ input.entry.value.email }}" --subject "Order {{ $trigger.query.id }}" --text "Thank you."
[f] kv.entry.put --key "seen:{{ $trigger.query.id }}" --value "{{ input.entry.value }}" --ttl 10m
[w] web.response.send --status 200 --body "ok"
[e] web.response.send --status 404 --body "no such order"
[t] -> [k]
[k] -> [c]
[c]:true -> [m]
[m] -> [f]
[f] -> [w]
[c]:false -> [e]
"#;
#[test]
fn a_complex_graph_of_quoted_expressions_parses_whole() {
let graph = build_pipeline_graph("complex-quoting", COMPLEX).expect("graph parses");
assert_eq!(graph.nodes.len(), 7, "every node survived");
assert_eq!(graph.edges.len(), 6, "every edge survived");
let by_id = |id: &str| {
graph
.nodes
.iter()
.find(|n| n.id == id)
.unwrap_or_else(|| panic!("node {id} missing"))
};
assert_eq!(
by_id("k").config["key"], "order:{{ $trigger.query.id }}",
"an interpolated key kept its expression"
);
assert_eq!(
by_id("m").config["recipient"], json!(["{{ input.entry.value.email }}"]),
"a whole-field expression kept its braces"
);
assert_eq!(
by_id("m").config["subject"], "Order {{ $trigger.query.id }}",
"text around an expression survived tokenizing"
);
assert_eq!(
by_id("f").config["value"], "{{ input.entry.value }}",
"the --value flag carries an expression"
);
assert_eq!(by_id("f").config["ttl"], "10m", "a duration stays as written");
assert_eq!(by_id("t").config["route"], "/orders");
assert_eq!(by_id("w").config["status"], 200);
}
#[test]
fn an_unquoted_expression_is_refused_with_advice() {
let dsl = r#"
[t] trigger.manual
[m] mail.message.send --credential relay --recipient {{ input.email }} --subject hi --text hi
[t] -> [m]
"#;
let err = build_pipeline_graph("unquoted", dsl).expect_err("must refuse");
assert!(err.contains("unquoted expression"), "{err}");
assert!(err.contains("Quote the whole thing"), "{err}");
}
#[test]
fn a_spaceless_expression_needs_no_quotes() {
let dsl = r#"
[t] trigger.manual
[k] kv.entry.get --key {{input.id}}
[t] -> [k]
"#;
let graph = build_pipeline_graph("spaceless", dsl).expect("graph parses");
let k = graph.nodes.iter().find(|n| n.id == "k").expect("node k");
assert_eq!(k.config["key"], "{{input.id}}");
}
}
#[cfg(test)]
mod key_value_quoting_tests {
use super::*;
#[test]
fn an_unquoted_expression_in_a_pair_is_refused() {
let err = build_pipeline_graph(
"claims",
"[a] trigger.manual\n\
[b] auth.token.create --credential k --claim sub={{ input.sub }}\n\
[a] -> [b]\n",
)
.expect_err("an unquoted expression in a key=value pair must be refused");
let message = format!("{err:?}");
assert!(
message.contains("unquoted expression") || message.contains("cut at the"),
"the error should name the cause: {message}"
);
}
#[test]
fn a_public_claim_keeps_a_whole_expression_as_its_value() {
let graph = build_pipeline_graph(
"claims-public",
"[a] trigger.manual\n\
[b] auth.token.create --credential k --claim \"name:public={{ input.name }}\"\n\
[a] -> [b]\n",
)
.expect("a public claim must parse");
let node = graph.nodes.iter().find(|n| n.id == "b").expect("node b");
let claims = node.config.get("claims").expect("claims");
assert_eq!(claims.get("name:public").and_then(|v| v.as_str()), Some("{{ input.name }}"));
}
#[test]
fn a_quoted_expression_in_a_pair_survives_whole() {
let graph = build_pipeline_graph(
"claims-ok",
"[a] trigger.manual\n\
[b] auth.token.create --credential k --claim \"sub={{ input.sub }}\"\n\
[a] -> [b]\n",
)
.expect("a quoted expression must parse");
let node = graph.nodes.iter().find(|n| n.id == "b").expect("node b");
let claims = node.config.get("claims").expect("claims");
assert_eq!(
claims.get("sub").and_then(|v| v.as_str()),
Some("{{ input.sub }}"),
"the whole expression must survive: {claims:?}"
);
}
}