diff --git a/src/catalog.rs b/src/catalog.rs new file mode 100644 index 0000000..39eb459 --- /dev/null +++ b/src/catalog.rs @@ -0,0 +1,542 @@ +//! Machine-readable authoring contracts for the node kinds — the queryable DSL +//! schema. +//! +//! A [`Node`](crate::model::Node)'s `config` is free-form +//! [`serde_json::Value`](serde_json::Value): each executor reads the keys it +//! needs at run time, so the per-kind config *shape* was, until now, documented +//! only in prose in downstream hosts. This module makes that shape a typed, +//! host-agnostic **source of truth** — one [`NodeKindContract`] per +//! [`NodeKind`](crate::model::NodeKind) — so a host (or an agent authoring a +//! graph) can enumerate the kinds and fetch one kind's config fields, ports, an +//! example node, and the authoring gotchas without reading a prompt. +//! +//! **Host-agnostic by construction** (the crate's core rule): these contracts +//! describe only what the tinyflows model and executors define. Anything a +//! specific host layers on top of the opaque fields — what a `tool_call` slug +//! resolves to, how its output is wrapped, which trigger kinds actually +//! dispatch — is deliberately *not* here; a host augments these contracts with +//! its own notes. + +use serde::{Deserialize, Serialize}; +use serde_json::{Value, json}; + +/// The node kinds, in the canonical order used wherever the DSL is enumerated +/// (matches [`NodeKind`](crate::model::NodeKind)'s serde discriminators). +pub const NODE_KINDS: [&str; 12] = [ + "trigger", + "agent", + "tool_call", + "http_request", + "code", + "condition", + "switch", + "merge", + "split_out", + "transform", + "output_parser", + "sub_workflow", +]; + +/// One config field a node of a given kind reads at run time. +#[derive(Debug, Clone, Serialize, Deserialize, PartialEq)] +pub struct ConfigField { + /// The `config.` key. + pub name: String, + /// Whether the node is malformed / a no-op without it. + pub required: bool, + /// A human-readable value-shape hint (`string`, `object`, `"=expr"`, + /// `enum`, `WorkflowGraph`, …) — descriptive, not a JSON Schema `type`. + pub value_type: String, + /// What the field means and how to fill it. + pub description: String, + /// The allowed values, when the field is a closed enum (e.g. + /// `trigger_kind`, `code.language`); `None` otherwise. + #[serde(default, skip_serializing_if = "Option::is_none")] + pub enum_values: Option>, +} + +impl ConfigField { + /// A required config field. + pub fn required(name: &str, value_type: &str, description: &str) -> Self { + Self { + name: name.to_string(), + required: true, + value_type: value_type.to_string(), + description: description.to_string(), + enum_values: None, + } + } + + /// An optional config field. + pub fn optional(name: &str, value_type: &str, description: &str) -> Self { + Self { + name: name.to_string(), + required: false, + value_type: value_type.to_string(), + description: description.to_string(), + enum_values: None, + } + } + + /// Marks this field a closed enum with the given allowed values. + #[must_use] + pub fn with_enum(mut self, values: &[&str]) -> Self { + self.enum_values = Some(values.iter().map(|s| s.to_string()).collect()); + self + } +} + +/// The input/output ports a node exposes. Routing is keyed exclusively on the +/// source node's `from_port` (see [`crate::validate`]'s condition-routing +/// check), so the output-port list is what an author wires branch edges onto. +#[derive(Debug, Clone, Serialize, Deserialize, PartialEq)] +pub struct PortSpec { + /// Named input ports. Almost always just `["main"]`. + pub inputs: Vec, + /// Named output ports. `["main"]` for a linear node; `["true","false"]` + /// for `condition`; case ports + `"default"` for `switch`. Every node can + /// additionally emit on `"error"` when its `on_error` policy is `"route"`. + pub outputs: Vec, +} + +impl PortSpec { + /// One `main` input and one `main` output — the shape of every linear node. + #[must_use] + pub fn linear() -> Self { + Self { + inputs: vec!["main".to_string()], + outputs: vec!["main".to_string()], + } + } + + /// Custom input/output port lists. + #[must_use] + pub fn new(inputs: &[&str], outputs: &[&str]) -> Self { + Self { + inputs: inputs.iter().map(|s| s.to_string()).collect(), + outputs: outputs.iter().map(|s| s.to_string()).collect(), + } + } +} + +/// The full machine-readable contract for one node kind. +#[derive(Debug, Clone, Serialize, Deserialize, PartialEq)] +pub struct NodeKindContract { + /// The kind discriminator (`trigger`, `agent`, …) — the `kind` field value. + pub kind: String, + /// One-line summary, safe to render in a compact list. + pub summary: String, + /// Fuller description of the node's role and how to author it. + pub description: String, + /// The `config.*` fields this kind reads. + pub config_fields: Vec, + /// Input/output ports. + pub ports: PortSpec, + /// A complete, valid example node (`{id, kind, name, config}`). + pub example: Value, + /// Authoring gotchas that bite in practice (envelope semantics, the + /// `from_port` branch rule, the `sub_workflow` XOR, …). + #[serde(default, skip_serializing_if = "Vec::is_empty")] + pub notes: Vec, +} + +impl NodeKindContract { + /// Appends a host-specific caveat to this contract's [`notes`](Self::notes), + /// returning the modified contract. + /// + /// The mechanism a host uses to augment the portable contract with facts it + /// owns — how a `tool_call` slug resolves, how output is wrapped, which + /// triggers dispatch — without editing the crate. + #[must_use] + pub fn with_note(mut self, note: impl Into) -> Self { + self.notes.push(note.into()); + self + } +} + +/// All node-kind contracts, in [`NODE_KINDS`] order. +pub fn all_contracts() -> Vec { + NODE_KINDS + .iter() + .map(|k| contract_for(k).expect("every NODE_KINDS entry has a contract")) + .collect() +} + +/// The contract for one node kind, or `None` if `kind` is not one of the 12. +pub fn contract_for(kind: &str) -> Option { + let c = match kind { + "trigger" => NodeKindContract { + kind: "trigger".to_string(), + summary: "The single entry point of the flow (exactly one required).".to_string(), + description: "Every graph has exactly one trigger. config.trigger_kind selects how it \ + fires; whether a given kind actually dispatches unattended is a host concern." + .to_string(), + config_fields: vec![ + ConfigField::required( + "trigger_kind", + "enum", + "How the flow fires. manual = run on demand; schedule = a timer; the rest are \ + event/host-driven.", + ) + .with_enum(&[ + "manual", + "schedule", + "webhook", + "app_event", + "form", + "chat_message", + "evaluation", + "system", + "execute_by_workflow", + ]), + ConfigField::optional( + "schedule", + "object", + "Required when trigger_kind=schedule: {kind:\"cron\",expr,tz?} | \ + {kind:\"at\",at} | {kind:\"every\",every_ms}.", + ), + ], + ports: PortSpec::new(&[], &["main"]), + example: json!({ + "id": "t", "kind": "trigger", "name": "Every morning", + "config": { "trigger_kind": "schedule", "schedule": { "kind": "cron", "expr": "0 9 * * *" } } + }), + notes: vec![ + "Exactly ONE trigger node per graph — zero or multiple is a hard reject." + .to_string(), + ], + }, + "agent" => NodeKindContract { + kind: "agent".to_string(), + summary: "An LLM step, run via the host's LlmProvider capability.".to_string(), + description: "Runs config.prompt through the injected LlmProvider. config.agent_ref \ + may select a host-registered agent persona; config.output_parser.schema requests a \ + structured, field-addressable output item. The exact data-input convention (how \ + the upstream item reaches the prompt) is defined by the host's LlmProvider." + .to_string(), + config_fields: vec![ + ConfigField::required("prompt", "string", "The instruction sent to the model."), + ConfigField::optional( + "agent_ref", + "string", + "A host-registered agent-kind id to run this step as (persona/model).", + ), + ConfigField::optional( + "output_parser", + "object", + "Set output_parser.schema (a JSON Schema object) to coerce the output into a \ + structured item whose fields downstream nodes can address; without it the \ + agent emits {text:\"...\"} only.", + ), + ConfigField::optional( + "connection_ref", + "string", + "An opaque connection reference passed to the LlmProvider, when the host needs \ + one.", + ), + ], + ports: PortSpec::linear(), + example: json!({ + "id": "classify", "kind": "agent", "name": "Classify", + "config": { + "prompt": "Classify the message as urgent, normal, or low priority.", + "output_parser": { "schema": { "type": "object", "properties": { "priority": { "type": "string" } } } } + } + }), + notes: vec![ + "If the output feeds a condition, declare that field \"type\":\"boolean\" in \ + output_parser.schema — an untyped field can carry the truthy string \"false\" and \ + route to the wrong port." + .to_string(), + ], + }, + "tool_call" => NodeKindContract { + kind: "tool_call".to_string(), + summary: "Invoke a tool via the host's ToolInvoker capability.".to_string(), + description: "config.slug names the tool (opaque to the engine — the host resolves \ + it); config.args are the arguments; config.connection_ref is an opaque account \ + reference. What slugs exist, their arg schemas, and how their output is shaped are \ + host concerns." + .to_string(), + config_fields: vec![ + ConfigField::required( + "slug", + "string", + "The tool identifier, resolved by the host's ToolInvoker.", + ), + ConfigField::optional( + "args", + "object", + "Arguments passed to the tool. Values may be literals or =bindings.", + ), + ConfigField::optional( + "connection_ref", + "string", + "An opaque connected-account reference the host resolves.", + ), + ], + ports: PortSpec::linear(), + example: json!({ + "id": "act", "kind": "tool_call", "name": "Do the thing", + "config": { "slug": "SOME_TOOL_ACTION", "args": { "to": "=nodes.pick.item.json.email" } } + }), + notes: vec![], + }, + "http_request" => NodeKindContract { + kind: "http_request".to_string(), + summary: "A raw HTTP call via the host's HttpClient capability.".to_string(), + description: "config.method + config.url, with optional headers/body. \ + config.connection_ref may reference a host credential for auth." + .to_string(), + config_fields: vec![ + ConfigField::required("method", "string", "HTTP method, e.g. GET / POST."), + ConfigField::required( + "url", + "string", + "The request URL (may be a =binding or contain =interpolated parts).", + ), + ConfigField::optional("headers", "object", "Request headers."), + ConfigField::optional("body", "any", "Request body (object or string)."), + ConfigField::optional( + "connection_ref", + "string", + "An opaque credential reference for authentication.", + ), + ], + ports: PortSpec::linear(), + example: json!({ + "id": "fetch", "kind": "http_request", "name": "Fetch", + "config": { "method": "GET", "url": "https://api.example.com/items" } + }), + notes: vec![], + }, + "code" => NodeKindContract { + kind: "code".to_string(), + summary: "Run a sandboxed JavaScript or Python snippet.".to_string(), + description: "config.language + config.source, run via the host's CodeRunner \ + capability." + .to_string(), + config_fields: vec![ + ConfigField::required("language", "enum", "The runtime language.") + .with_enum(&["javascript", "python"]), + ConfigField::required("source", "string", "The code to run."), + ], + ports: PortSpec::linear(), + example: json!({ + "id": "shape", "kind": "code", "name": "Shape", + "config": { "language": "javascript", "source": "return { total: item.a + item.b };" } + }), + notes: vec![], + }, + "condition" => NodeKindContract { + kind: "condition".to_string(), + summary: "A boolean gate that routes to the `true` or `false` port.".to_string(), + description: "Evaluates config.field and routes on from_port \"true\" or \"false\". \ + Wire both branches (or the unwired one dead-ends)." + .to_string(), + config_fields: vec![ConfigField::required( + "field", + "\"=expr\"", + "The boolean expression/field to gate on.", + )], + ports: PortSpec::new(&["main"], &["true", "false"]), + example: json!({ + "id": "gate", "kind": "condition", "name": "Urgent?", + "config": { "field": "=nodes.classify.item.json.priority == \"urgent\"" } + }), + notes: vec![ + "HARD RULE: the branch label goes on the edge's from_port, e.g. \ + {from_node:\"gate\",from_port:\"true\",to_node:\"x\",to_port:\"main\"}. Putting \ + the label on to_port instead silently turns the branch into an unconditional \ + fan-out (BOTH branches run) and is a hard reject." + .to_string(), + ], + }, + "switch" => NodeKindContract { + kind: "switch".to_string(), + summary: "Multi-way routing to the matching case port, else `default`.".to_string(), + description: "Evaluates config.expression (or config.field) and routes on from_port \ + equal to the matching case value, falling back to the \"default\" port." + .to_string(), + config_fields: vec![ + ConfigField::optional( + "expression", + "\"=expr\"", + "The expression whose value selects the case port. Provide this OR field.", + ), + ConfigField::optional( + "field", + "\"=expr\"", + "A field whose value selects the case port. Provide this OR expression.", + ), + ], + ports: PortSpec::new(&["main"], &["…", "default"]), + example: json!({ + "id": "route", "kind": "switch", "name": "By type", + "config": { "field": "=item.type" } + }), + notes: vec![ + "Like condition, case labels go on the edge's from_port; to_port stays \"main\"." + .to_string(), + ], + }, + "merge" => NodeKindContract { + kind: "merge".to_string(), + summary: "A fan-in barrier that passes its inputs through.".to_string(), + description: "Waits for its inbound branches and passes the collected items through. \ + No config." + .to_string(), + config_fields: vec![], + ports: PortSpec::linear(), + example: json!({ "id": "join", "kind": "merge", "name": "Join" }), + notes: vec![], + }, + "split_out" => NodeKindContract { + kind: "split_out".to_string(), + summary: "Fan out one item per element of an array field.".to_string(), + description: "config.path names an array within the current item; the node emits one \ + item per element." + .to_string(), + config_fields: vec![ConfigField::required( + "path", + "string", + "Dotted path to the array field to fan out over, e.g. \"json.data.messages\".", + )], + ports: PortSpec::linear(), + example: json!({ + "id": "each", "kind": "split_out", "name": "Each item", + "config": { "path": "json.items" } + }), + notes: vec![], + }, + "transform" => NodeKindContract { + kind: "transform".to_string(), + summary: "Merge computed keys onto each item.".to_string(), + description: "config.set = { key: \"=expr\" } — each expression is evaluated and \ + merged onto every item flowing through." + .to_string(), + config_fields: vec![ConfigField::required( + "set", + "object", + "A map of output key -> \"=expr\" merged onto each item.", + )], + ports: PortSpec::linear(), + example: json!({ + "id": "enrich", "kind": "transform", "name": "Add name", + "config": { "set": { "full_name": "=item.first + \" \" + item.last" } } + }), + notes: vec![], + }, + "output_parser" => NodeKindContract { + kind: "output_parser".to_string(), + summary: "Passthrough today; no config required.".to_string(), + description: "A passthrough node reserved for structured-output parsing. Requires no \ + config." + .to_string(), + config_fields: vec![], + ports: PortSpec::linear(), + example: json!({ "id": "parse", "kind": "output_parser", "name": "Parse" }), + notes: vec![], + }, + "sub_workflow" => NodeKindContract { + kind: "sub_workflow".to_string(), + summary: "Run a child workflow — inline or by reference.".to_string(), + description: "References its child EXACTLY one way: config.workflow (an inline child \ + WorkflowGraph) OR config.workflow_id (resolved by the host's WorkflowResolver) — \ + never both, never neither." + .to_string(), + config_fields: vec![ + ConfigField::optional( + "workflow", + "WorkflowGraph", + "An inline child graph. Provide this OR workflow_id, not both.", + ), + ConfigField::optional( + "workflow_id", + "string", + "The id of a saved workflow to run as the child. Provide this OR workflow.", + ), + ], + ports: PortSpec::linear(), + example: json!({ + "id": "sub", "kind": "sub_workflow", "name": "Enrich", + "config": { "workflow_id": "flow-123" } + }), + notes: vec![ + "Exactly one of workflow / workflow_id — having both or neither is a hard reject." + .to_string(), + ], + }, + _ => return None, + }; + Some(c) +} + +#[cfg(test)] +mod tests { + use super::*; + use crate::model::NodeKind; + + #[test] + fn every_node_kind_has_a_contract() { + for kind in NODE_KINDS { + let c = contract_for(kind).unwrap_or_else(|| panic!("no contract for {kind}")); + assert_eq!(c.kind, kind); + assert!(!c.summary.is_empty(), "{kind} has empty summary"); + assert!(!c.description.is_empty(), "{kind} has empty description"); + assert_eq!( + c.example.get("kind").and_then(Value::as_str), + Some(kind), + "{kind} example has the wrong kind" + ); + for f in &c.config_fields { + if f.value_type == "enum" { + assert!( + f.enum_values.is_some(), + "{kind}.{} is an enum but lists no values", + f.name + ); + } + } + } + assert_eq!(all_contracts().len(), 12); + } + + #[test] + fn node_kinds_match_the_model_enum() { + // Every catalog entry must deserialize back to a real NodeKind, and the + // count must match — a new NodeKind without a contract fails here. + for kind in NODE_KINDS { + let parsed: NodeKind = serde_json::from_value(Value::String(kind.to_string())) + .unwrap_or_else(|_| { + panic!("catalog kind {kind} is not a real NodeKind discriminator") + }); + // round-trips back to the same string + assert_eq!( + serde_json::to_value(parsed).unwrap(), + Value::String(kind.to_string()) + ); + } + } + + #[test] + fn unknown_kind_has_no_contract() { + assert!(contract_for("not_a_kind").is_none()); + assert!(contract_for("").is_none()); + } + + #[test] + fn with_note_appends_a_host_caveat() { + let c = contract_for("tool_call").unwrap().with_note("host says hi"); + assert_eq!(c.notes.last().map(String::as_str), Some("host says hi")); + } + + #[test] + fn contracts_are_serde_round_trippable() { + for c in all_contracts() { + let json = serde_json::to_value(&c).unwrap(); + let back: NodeKindContract = serde_json::from_value(json).unwrap(); + assert_eq!(c, back); + } + } +} diff --git a/src/error.rs b/src/error.rs index 70ce1db..2ae752b 100644 --- a/src/error.rs +++ b/src/error.rs @@ -101,6 +101,52 @@ pub enum ValidationError { }, } +impl ValidationError { + /// A stable, machine-readable code for this error variant. + /// + /// Unlike the human-readable [`Display`](std::fmt::Display) form, this is a + /// fixed `snake_case` identifier safe for a host to switch on or surface to + /// an agent as a structured `code` field. It never changes for a given + /// variant. + pub fn code(&self) -> &'static str { + match self { + Self::MissingTrigger => "missing_trigger", + Self::MultipleTriggers(_) => "multiple_triggers", + Self::UnknownNode(_) => "unknown_node", + Self::DuplicateNodeId(_) => "duplicate_node_id", + Self::IllegalCycle(_) => "illegal_cycle", + Self::InvalidNodeConfig { .. } => "invalid_node_config", + Self::MissingErrorRoute(_) => "missing_error_route", + Self::DuplicateEdge { .. } => "duplicate_edge", + Self::InvalidOnError { .. } => "invalid_on_error", + Self::SchemaVersionTooNew { .. } => "schema_version_too_new", + Self::InvalidConditionRouting { .. } => "invalid_condition_routing", + } + } + + /// 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. + pub fn node_id(&self) -> Option<&str> { + match self { + Self::UnknownNode(id) + | Self::DuplicateNodeId(id) + | Self::IllegalCycle(id) + | Self::MissingErrorRoute(id) => Some(id), + Self::InvalidNodeConfig { node, .. } + | Self::InvalidOnError { node, .. } + | Self::InvalidConditionRouting { node, .. } => Some(node), + Self::DuplicateEdge { from_node, .. } => Some(from_node), + Self::MissingTrigger | Self::MultipleTriggers(_) | Self::SchemaVersionTooNew { .. } => { + None + } + } + } +} + /// Errors produced while compiling or running a workflow. #[derive(Debug, Error)] pub enum EngineError { diff --git a/src/graph_ops.rs b/src/graph_ops.rs new file mode 100644 index 0000000..8c31af8 --- /dev/null +++ b/src/graph_ops.rs @@ -0,0 +1,652 @@ +//! Structured, incremental edits to a [`WorkflowGraph`] — the patch-op layer. +//! +//! Authoring a flow by re-emitting the *entire* graph on every change is +//! token-heavy and the #1 source of accidental regressions (a dropped node, a +//! mangled edge). This module lets a caller express a change as a small list of +//! [`GraphOp`]s — add a node, merge-patch one node's config, rewire an edge — +//! applied to a base graph with precise, per-op errors. +//! +//! [`apply_ops`] performs only the **structural mutation**; it deliberately +//! does not run [`crate::validate`]. The intended pipeline is +//! `apply_ops` → `validate_all` → (host gates), so a caller gets a clear +//! "op 3 (add_edge) failed" for a malformed *operation* separately from the +//! structural validation of the resulting graph. +//! +//! Host-agnostic: these are edits to the portable model, with no knowledge of +//! what any `config` field means. + +use serde::{Deserialize, Serialize}; +use serde_json::{Map, Value}; + +use crate::model::{Edge, Node, NodeId, Position, WorkflowGraph}; + +/// Serde default for an op's port fields — matches [`Edge`]'s `"main"` default. +fn default_port() -> String { + "main".to_string() +} + +/// One structured edit to a [`WorkflowGraph`]. +/// +/// Serialized as an internally-tagged object: `{ "op": "add_node", ... }`. +#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)] +#[serde(tag = "op", rename_all = "snake_case")] +pub enum GraphOp { + /// Append a new node. Fails if its `id` is empty or already present. + AddNode { + /// The node to add. + node: Node, + }, + /// Merge-patch a node's `config` (RFC 7386 JSON Merge Patch): keys in + /// `config` are recursively merged onto the node's existing config, and a + /// `null` value deletes that key. Fails if no node has `id`. + UpdateNodeConfig { + /// The target node id. + id: NodeId, + /// The partial config to merge (a `null` leaf deletes the key). + config: Value, + }, + /// Replace a node's human-readable `name`. Fails if no node has `id`. + SetNodeName { + /// The target node id. + id: NodeId, + /// The new display name. + name: String, + }, + /// Change a node's `id`, rewiring every edge that referenced the old id. + /// + /// Note: `=nodes.…` references inside *other* nodes' config expressions + /// are NOT rewritten (that would require parsing jq) — a caller renaming a + /// node that others bind to should re-point those bindings itself. Fails if + /// `new_id` is empty or already in use, or if no node has `id`. + RenameNode { + /// The current node id. + id: NodeId, + /// The new node id. + new_id: NodeId, + }, + /// Remove a node and every edge incident on it. Fails if no node has `id`. + RemoveNode { + /// The node id to remove. + id: NodeId, + }, + /// Add a directed edge. Fails if either endpoint node is missing or the + /// exact edge (same `from`/`to` node and port) already exists. + AddEdge { + /// The edge to add. + edge: Edge, + }, + /// Remove every edge matching the given `from`/`to` node and port (ports + /// default to `"main"`). Fails if no edge matches. + RemoveEdge { + /// Source node id. + from_node: NodeId, + /// Source port (defaults to `"main"`). + #[serde(default = "default_port")] + from_port: String, + /// Target node id. + to_node: NodeId, + /// Target port (defaults to `"main"`). + #[serde(default = "default_port")] + to_port: String, + }, + /// Set (or move) a node's canvas position. Fails if no node has `id`. + SetNodePosition { + /// The target node id. + id: NodeId, + /// The new canvas position. + position: Position, + }, +} + +impl GraphOp { + /// The op's stable, machine-readable name (its serde tag), for diagnostics. + pub fn name(&self) -> &'static str { + match self { + Self::AddNode { .. } => "add_node", + Self::UpdateNodeConfig { .. } => "update_node_config", + Self::SetNodeName { .. } => "set_node_name", + Self::RenameNode { .. } => "rename_node", + Self::RemoveNode { .. } => "remove_node", + Self::AddEdge { .. } => "add_edge", + Self::RemoveEdge { .. } => "remove_edge", + Self::SetNodePosition { .. } => "set_node_position", + } + } +} + +/// A 4-tuple identifying an edge, for edge-related errors. Boxed inside +/// [`GraphOpErrorKind`] so those variants stay small. +#[derive(Debug, Clone, PartialEq, Eq)] +pub struct EdgeRef { + /// Source node id. + pub from_node: NodeId, + /// Source port. + pub from_port: String, + /// Target node id. + pub to_node: NodeId, + /// Target port. + pub to_port: String, +} + +impl std::fmt::Display for EdgeRef { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + write!( + f, + "{}.{} -> {}.{}", + self.from_node, self.from_port, self.to_node, self.to_port + ) + } +} + +/// Why a single [`GraphOp`] could not be applied. +#[derive(Debug, Clone, PartialEq, Eq, thiserror::Error)] +pub enum GraphOpErrorKind { + /// A referenced node id does not exist in the graph. + #[error("no node with id {0}")] + NodeNotFound(NodeId), + /// A node id that must be new is already taken. + #[error("a node with id {0} already exists")] + NodeIdExists(NodeId), + /// A node id (new or renamed-to) is empty. + #[error("node id must not be empty")] + EmptyNodeId, + /// An edge endpoint references a node that does not exist. + #[error("edge references unknown node id {0}")] + EdgeEndpointMissing(NodeId), + /// The exact edge already exists. + #[error("edge {0} already exists")] + EdgeExists(Box), + /// No edge matched a [`GraphOp::RemoveEdge`]. + #[error("no edge {0} to remove")] + EdgeNotFound(Box), +} + +/// A failure to apply an op, carrying which op (0-based index) failed. +#[derive(Debug, Clone, PartialEq, Eq, thiserror::Error)] +#[error("op {index} ({op}): {kind}")] +pub struct GraphOpError { + /// 0-based index of the failing op in the input list. + pub index: usize, + /// The failing op's name (serde tag). + pub op: &'static str, + /// What went wrong. + pub kind: GraphOpErrorKind, +} + +/// Applies `ops` to a clone of `base`, in order, returning the mutated graph. +/// +/// Purely structural: on the first op that cannot be applied, returns a +/// [`GraphOpError`] naming the op index and reason, leaving `base` untouched +/// (the working copy is discarded). Run [`crate::validate::validate_all`] on +/// the result to check the resulting graph's structure. +pub fn apply_ops(base: &WorkflowGraph, ops: &[GraphOp]) -> Result { + let mut graph = base.clone(); + for (index, op) in ops.iter().enumerate() { + apply_one(&mut graph, op).map_err(|kind| GraphOpError { + index, + op: op.name(), + kind, + })?; + } + Ok(graph) +} + +fn node_index(graph: &WorkflowGraph, id: &str) -> Option { + graph.nodes.iter().position(|n| n.id == id) +} + +fn apply_one(graph: &mut WorkflowGraph, op: &GraphOp) -> Result<(), GraphOpErrorKind> { + match op { + GraphOp::AddNode { node } => { + if node.id.is_empty() { + return Err(GraphOpErrorKind::EmptyNodeId); + } + if node_index(graph, &node.id).is_some() { + return Err(GraphOpErrorKind::NodeIdExists(node.id.clone())); + } + graph.nodes.push(node.clone()); + } + GraphOp::UpdateNodeConfig { id, config } => { + let idx = + node_index(graph, id).ok_or_else(|| GraphOpErrorKind::NodeNotFound(id.clone()))?; + json_merge_patch(&mut graph.nodes[idx].config, config); + } + GraphOp::SetNodeName { id, name } => { + let idx = + node_index(graph, id).ok_or_else(|| GraphOpErrorKind::NodeNotFound(id.clone()))?; + graph.nodes[idx].name = name.clone(); + } + GraphOp::RenameNode { id, new_id } => { + if new_id.is_empty() { + return Err(GraphOpErrorKind::EmptyNodeId); + } + if node_index(graph, id).is_none() { + return Err(GraphOpErrorKind::NodeNotFound(id.clone())); + } + if new_id != id && node_index(graph, new_id).is_some() { + return Err(GraphOpErrorKind::NodeIdExists(new_id.clone())); + } + for node in &mut graph.nodes { + if node.id == *id { + node.id = new_id.clone(); + } + } + for edge in &mut graph.edges { + if edge.from_node == *id { + edge.from_node = new_id.clone(); + } + if edge.to_node == *id { + edge.to_node = new_id.clone(); + } + } + } + GraphOp::RemoveNode { id } => { + if node_index(graph, id).is_none() { + return Err(GraphOpErrorKind::NodeNotFound(id.clone())); + } + graph.nodes.retain(|n| n.id != *id); + graph + .edges + .retain(|e| e.from_node != *id && e.to_node != *id); + } + GraphOp::AddEdge { edge } => { + if node_index(graph, &edge.from_node).is_none() { + return Err(GraphOpErrorKind::EdgeEndpointMissing( + edge.from_node.clone(), + )); + } + if node_index(graph, &edge.to_node).is_none() { + return Err(GraphOpErrorKind::EdgeEndpointMissing(edge.to_node.clone())); + } + let exists = graph.edges.iter().any(|e| { + e.from_node == edge.from_node + && e.from_port == edge.from_port + && e.to_node == edge.to_node + && e.to_port == edge.to_port + }); + if exists { + return Err(GraphOpErrorKind::EdgeExists(Box::new(EdgeRef { + from_node: edge.from_node.clone(), + from_port: edge.from_port.clone(), + to_node: edge.to_node.clone(), + to_port: edge.to_port.clone(), + }))); + } + graph.edges.push(edge.clone()); + } + GraphOp::RemoveEdge { + from_node, + from_port, + to_node, + to_port, + } => { + let before = graph.edges.len(); + graph.edges.retain(|e| { + !(e.from_node == *from_node + && e.from_port == *from_port + && e.to_node == *to_node + && e.to_port == *to_port) + }); + if graph.edges.len() == before { + return Err(GraphOpErrorKind::EdgeNotFound(Box::new(EdgeRef { + from_node: from_node.clone(), + from_port: from_port.clone(), + to_node: to_node.clone(), + to_port: to_port.clone(), + }))); + } + } + GraphOp::SetNodePosition { id, position } => { + let idx = + node_index(graph, id).ok_or_else(|| GraphOpErrorKind::NodeNotFound(id.clone()))?; + graph.nodes[idx].position = Some(*position); + } + } + Ok(()) +} + +/// Applies an RFC 7386 JSON Merge Patch of `patch` onto `target` in place. +/// +/// Object values are merged recursively; a `null` leaf deletes the +/// corresponding key; any non-object patch replaces the target wholesale. +fn json_merge_patch(target: &mut Value, patch: &Value) { + match patch { + Value::Object(patch_map) => { + if !target.is_object() { + *target = Value::Object(Map::new()); + } + let target_map = target.as_object_mut().expect("just ensured object"); + for (key, patch_val) in patch_map { + if patch_val.is_null() { + target_map.remove(key); + } else { + json_merge_patch( + target_map.entry(key.clone()).or_insert(Value::Null), + patch_val, + ); + } + } + } + _ => *target = patch.clone(), + } +} + +#[cfg(test)] +mod tests { + use super::*; + use crate::model::NodeKind; + use serde_json::json; + + fn node(id: &str, kind: NodeKind) -> Node { + Node { + id: id.to_string(), + kind, + type_version: 1, + name: id.to_string(), + config: Value::Null, + ports: Vec::new(), + position: None, + } + } + + fn base() -> WorkflowGraph { + WorkflowGraph { + nodes: vec![node("t", NodeKind::Trigger), node("a", NodeKind::Agent)], + edges: vec![Edge { + from_node: "t".to_string(), + from_port: "main".to_string(), + to_node: "a".to_string(), + to_port: "main".to_string(), + }], + ..Default::default() + } + } + + #[test] + fn add_node_appends_and_rejects_duplicates() { + let g = apply_ops( + &base(), + &[GraphOp::AddNode { + node: node("b", NodeKind::Merge), + }], + ) + .unwrap(); + assert_eq!(g.nodes.len(), 3); + + let err = apply_ops( + &base(), + &[GraphOp::AddNode { + node: node("a", NodeKind::Agent), + }], + ) + .unwrap_err(); + assert_eq!(err.index, 0); + assert_eq!(err.op, "add_node"); + assert!(matches!(err.kind, GraphOpErrorKind::NodeIdExists(_))); + } + + #[test] + fn add_node_rejects_empty_id() { + let err = apply_ops( + &base(), + &[GraphOp::AddNode { + node: node("", NodeKind::Merge), + }], + ) + .unwrap_err(); + assert!(matches!(err.kind, GraphOpErrorKind::EmptyNodeId)); + } + + #[test] + fn update_node_config_merge_patches() { + let mut n = node("a", NodeKind::Agent); + n.config = json!({ "prompt": "hi", "keep": 1 }); + let g0 = WorkflowGraph { + nodes: vec![node("t", NodeKind::Trigger), n], + ..Default::default() + }; + let g = apply_ops( + &g0, + &[GraphOp::UpdateNodeConfig { + id: "a".to_string(), + config: json!({ "prompt": "bye", "added": true, "keep": null }), + }], + ) + .unwrap(); + let cfg = &g.nodes[1].config; + assert_eq!(cfg["prompt"], "bye"); + assert_eq!(cfg["added"], true); + assert!( + cfg.get("keep").is_none(), + "null leaf deletes the key: {cfg}" + ); + } + + #[test] + fn update_node_config_on_null_config_creates_object() { + let g = apply_ops( + &base(), + &[GraphOp::UpdateNodeConfig { + id: "a".to_string(), + config: json!({ "x": 1 }), + }], + ) + .unwrap(); + assert_eq!(g.nodes[1].config["x"], 1); + } + + #[test] + fn update_node_config_missing_node_errors() { + let err = apply_ops( + &base(), + &[GraphOp::UpdateNodeConfig { + id: "ghost".to_string(), + config: json!({}), + }], + ) + .unwrap_err(); + assert!(matches!(err.kind, GraphOpErrorKind::NodeNotFound(_))); + } + + #[test] + fn rename_node_rewires_edges() { + let g = apply_ops( + &base(), + &[GraphOp::RenameNode { + id: "a".to_string(), + new_id: "agent1".to_string(), + }], + ) + .unwrap(); + assert!(g.nodes.iter().any(|n| n.id == "agent1")); + assert!(g.nodes.iter().all(|n| n.id != "a")); + assert_eq!(g.edges[0].to_node, "agent1"); + } + + #[test] + fn rename_node_rejects_collision_and_missing() { + let err = apply_ops( + &base(), + &[GraphOp::RenameNode { + id: "a".to_string(), + new_id: "t".to_string(), + }], + ) + .unwrap_err(); + assert!(matches!(err.kind, GraphOpErrorKind::NodeIdExists(_))); + + let err = apply_ops( + &base(), + &[GraphOp::RenameNode { + id: "ghost".to_string(), + new_id: "z".to_string(), + }], + ) + .unwrap_err(); + assert!(matches!(err.kind, GraphOpErrorKind::NodeNotFound(_))); + } + + #[test] + fn remove_node_drops_incident_edges() { + let g = apply_ops( + &base(), + &[GraphOp::RemoveNode { + id: "a".to_string(), + }], + ) + .unwrap(); + assert_eq!(g.nodes.len(), 1); + assert!(g.edges.is_empty(), "incident edge removed"); + } + + #[test] + fn add_edge_validates_endpoints_and_dupes() { + // both endpoints must exist + let err = apply_ops( + &base(), + &[GraphOp::AddEdge { + edge: Edge { + from_node: "a".to_string(), + from_port: "main".to_string(), + to_node: "ghost".to_string(), + to_port: "main".to_string(), + }, + }], + ) + .unwrap_err(); + assert!(matches!(err.kind, GraphOpErrorKind::EdgeEndpointMissing(_))); + + // duplicate rejected + let err = apply_ops( + &base(), + &[GraphOp::AddEdge { + edge: Edge { + from_node: "t".to_string(), + from_port: "main".to_string(), + to_node: "a".to_string(), + to_port: "main".to_string(), + }, + }], + ) + .unwrap_err(); + assert!(matches!(err.kind, GraphOpErrorKind::EdgeExists(_))); + } + + #[test] + fn remove_edge_matches_or_errors() { + let g = apply_ops( + &base(), + &[GraphOp::RemoveEdge { + from_node: "t".to_string(), + from_port: "main".to_string(), + to_node: "a".to_string(), + to_port: "main".to_string(), + }], + ) + .unwrap(); + assert!(g.edges.is_empty()); + + let err = apply_ops( + &base(), + &[GraphOp::RemoveEdge { + from_node: "t".to_string(), + from_port: "main".to_string(), + to_node: "ghost".to_string(), + to_port: "main".to_string(), + }], + ) + .unwrap_err(); + assert!(matches!(err.kind, GraphOpErrorKind::EdgeNotFound(_))); + } + + #[test] + fn set_node_position_sets_coords() { + let g = apply_ops( + &base(), + &[GraphOp::SetNodePosition { + id: "a".to_string(), + position: Position { x: 10.0, y: 20.0 }, + }], + ) + .unwrap(); + assert_eq!(g.nodes[1].position, Some(Position { x: 10.0, y: 20.0 })); + } + + #[test] + fn ops_apply_in_sequence_and_base_is_untouched() { + let b = base(); + let g = apply_ops( + &b, + &[ + GraphOp::AddNode { + node: node("b", NodeKind::Merge), + }, + GraphOp::AddEdge { + edge: Edge { + from_node: "a".to_string(), + from_port: "main".to_string(), + to_node: "b".to_string(), + to_port: "main".to_string(), + }, + }, + GraphOp::RenameNode { + id: "b".to_string(), + new_id: "merge1".to_string(), + }, + ], + ) + .unwrap(); + assert_eq!(g.nodes.len(), 3); + assert_eq!(g.edges.len(), 2); + assert!(g.edges.iter().any(|e| e.to_node == "merge1")); + // base untouched + assert_eq!(b.nodes.len(), 2); + assert_eq!(b.edges.len(), 1); + } + + #[test] + fn error_index_points_at_the_failing_op() { + // op 0 ok, op 1 fails. + let err = apply_ops( + &base(), + &[ + GraphOp::SetNodeName { + id: "a".to_string(), + name: "Renamed".to_string(), + }, + GraphOp::RemoveNode { + id: "ghost".to_string(), + }, + ], + ) + .unwrap_err(); + assert_eq!(err.index, 1); + assert_eq!(err.op, "remove_node"); + } + + #[test] + fn graph_op_deserializes_from_tagged_json() { + let op: GraphOp = serde_json::from_value(json!({ + "op": "update_node_config", + "id": "a", + "config": { "prompt": "hi" } + })) + .unwrap(); + assert!(matches!(op, GraphOp::UpdateNodeConfig { .. })); + + // remove_edge ports default to "main" + let op: GraphOp = serde_json::from_value(json!({ + "op": "remove_edge", "from_node": "t", "to_node": "a" + })) + .unwrap(); + match op { + GraphOp::RemoveEdge { + from_port, to_port, .. + } => { + assert_eq!(from_port, "main"); + assert_eq!(to_port, "main"); + } + _ => panic!("wrong variant"), + } + } +} diff --git a/src/lib.rs b/src/lib.rs index 24a8e8a..ac313bf 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -20,11 +20,13 @@ #![warn(missing_docs)] pub mod caps; +pub mod catalog; pub mod compiler; pub mod data; pub mod engine; pub mod error; pub mod expr; +pub mod graph_ops; pub mod migrate; pub mod model; pub mod nodes; diff --git a/src/validate.rs b/src/validate.rs index daea24e..463ff76 100644 --- a/src/validate.rs +++ b/src/validate.rs @@ -16,12 +16,36 @@ use crate::model::{NodeKind, WorkflowGraph}; /// A1–A2. /// /// # Errors -/// Returns the first [`ValidationError`] encountered. +/// Returns the first [`ValidationError`] encountered. For a full list of every +/// structural problem in one pass (so an author can fix them all at once +/// instead of one round-trip per error), use [`validate_all`]; this function is +/// exactly its first element and is kept for the fail-fast compile path. pub fn validate(graph: &WorkflowGraph) -> Result<(), ValidationError> { + match validate_all(graph).into_iter().next() { + Some(err) => Err(err), + None => Ok(()), + } +} + +/// Validates a workflow graph's structure, collecting **every** independent +/// error in one pass. +/// +/// 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 +/// [`validate`] returns, preserving the historical single-error contract. +/// +/// This is what a host should surface to an author or agent: fixing five +/// problems then costs one validate call, not five. +pub fn validate_all(graph: &WorkflowGraph) -> Vec { + let mut errors = Vec::new(); + let mut seen = HashSet::new(); for node in &graph.nodes { if !seen.insert(node.id.as_str()) { - return Err(ValidationError::DuplicateNodeId(node.id.clone())); + errors.push(ValidationError::DuplicateNodeId(node.id.clone())); } } @@ -32,18 +56,18 @@ pub fn validate(graph: &WorkflowGraph) -> Result<(), ValidationError> { .map(|n| n.id.clone()) .collect(); match triggers.len() { - 0 => return Err(ValidationError::MissingTrigger), + 0 => errors.push(ValidationError::MissingTrigger), 1 => {} - _ => return Err(ValidationError::MultipleTriggers(triggers)), + _ => errors.push(ValidationError::MultipleTriggers(triggers)), } let mut seen_edges = HashSet::new(); for edge in &graph.edges { if graph.node(&edge.from_node).is_none() { - return Err(ValidationError::UnknownNode(edge.from_node.clone())); + errors.push(ValidationError::UnknownNode(edge.from_node.clone())); } if graph.node(&edge.to_node).is_none() { - return Err(ValidationError::UnknownNode(edge.to_node.clone())); + errors.push(ValidationError::UnknownNode(edge.to_node.clone())); } // Reject two identical edges (same source node/port and destination // node/port); a redundant duplicate is almost always an authoring slip. @@ -53,7 +77,7 @@ pub fn validate(graph: &WorkflowGraph) -> Result<(), ValidationError> { edge.to_node.as_str(), edge.to_port.as_str(), )) { - return Err(ValidationError::DuplicateEdge { + errors.push(ValidationError::DuplicateEdge { from_node: edge.from_node.clone(), from_port: edge.from_port.clone(), to_node: edge.to_node.clone(), @@ -78,11 +102,11 @@ pub fn validate(graph: &WorkflowGraph) -> Result<(), ValidationError> { .iter() .any(|e| e.from_node == node.id && e.from_port == "error"); if !has_error_edge { - return Err(ValidationError::MissingErrorRoute(node.id.clone())); + errors.push(ValidationError::MissingErrorRoute(node.id.clone())); } } other => { - return Err(ValidationError::InvalidOnError { + errors.push(ValidationError::InvalidOnError { node: node.id.clone(), value: other.to_string(), }); @@ -103,7 +127,7 @@ pub fn validate(graph: &WorkflowGraph) -> Result<(), ValidationError> { .and_then(serde_json::Value::as_str) .is_some_and(|s| !s.is_empty()); if has_inline == has_ref { - return Err(ValidationError::InvalidNodeConfig { + errors.push(ValidationError::InvalidNodeConfig { node: node.id.clone(), reason: "sub_workflow requires exactly one of `workflow` (inline) or \ `workflow_id` (reference)" @@ -132,14 +156,14 @@ pub fn validate(graph: &WorkflowGraph) -> Result<(), ValidationError> { && edge.from_port != "true" && edge.from_port != "false" { - return Err(ValidationError::InvalidConditionRouting { + errors.push(ValidationError::InvalidConditionRouting { node: edge.from_node.clone(), from_port: edge.from_port.clone(), }); } } - Ok(()) + errors } #[cfg(test)] @@ -643,4 +667,140 @@ mod tests { Err(ValidationError::DuplicateNodeId("dup".to_string())) ); } + + #[test] + fn validate_all_is_empty_for_a_valid_graph() { + let graph = WorkflowGraph { + nodes: vec![node("t", NodeKind::Trigger), node("a", NodeKind::Agent)], + edges: vec![Edge { + from_node: "t".to_string(), + from_port: "main".to_string(), + to_node: "a".to_string(), + to_port: "main".to_string(), + }], + ..Default::default() + }; + assert!(validate_all(&graph).is_empty()); + } + + #[test] + fn validate_all_first_element_matches_validate() { + // The single-error contract of `validate` must stay exactly the first + // element of `validate_all` — same graph, same lead error. + let graph = WorkflowGraph { + nodes: vec![node("dup", NodeKind::Agent), node("dup", NodeKind::Agent)], + ..Default::default() + }; + assert_eq!( + validate_all(&graph).into_iter().next(), + validate(&graph).err() + ); + } + + #[test] + fn validate_all_accumulates_independent_errors() { + // A graph riddled with problems: no trigger, a duplicate node id, a + // dangling edge, an unknown on_error value, and a mis-wired condition. + // One pass should surface all of them, not just the first. + let graph = WorkflowGraph { + nodes: vec![ + node("dup", NodeKind::Agent), + node("dup", NodeKind::Agent), + condition_node("gate"), + tool_node("x", serde_json::json!({ "on_error": "explode" })), + ], + edges: vec![Edge { + from_node: "gate".to_string(), + from_port: "maybe".to_string(), + to_node: "ghost".to_string(), + to_port: "main".to_string(), + }], + ..Default::default() + }; + let errors = validate_all(&graph); + assert!( + errors.contains(&ValidationError::DuplicateNodeId("dup".to_string())), + "missing duplicate-id error in {errors:?}" + ); + assert!( + errors.contains(&ValidationError::MissingTrigger), + "missing trigger error in {errors:?}" + ); + assert!( + errors.contains(&ValidationError::UnknownNode("ghost".to_string())), + "missing unknown-node error in {errors:?}" + ); + assert!( + errors.contains(&ValidationError::InvalidOnError { + node: "x".to_string(), + value: "explode".to_string(), + }), + "missing invalid-on_error error in {errors:?}" + ); + assert!( + errors.contains(&ValidationError::InvalidConditionRouting { + node: "gate".to_string(), + from_port: "maybe".to_string(), + }), + "missing condition-routing error in {errors:?}" + ); + // Five distinct problems, five errors — no fail-fast truncation. + assert!( + errors.len() >= 5, + "expected >=5 accumulated errors, got {errors:?}" + ); + } + + #[test] + fn validate_all_reports_every_duplicate_and_every_dangling_edge() { + // Two separate dangling edges must both be reported (fail-fast would + // stop at the first). + let graph = WorkflowGraph { + nodes: vec![node("t", NodeKind::Trigger)], + edges: vec![ + Edge { + from_node: "t".to_string(), + from_port: "main".to_string(), + to_node: "ghost1".to_string(), + to_port: "main".to_string(), + }, + Edge { + from_node: "t".to_string(), + from_port: "main".to_string(), + to_node: "ghost2".to_string(), + to_port: "main".to_string(), + }, + ], + ..Default::default() + }; + let errors = validate_all(&graph); + assert!(errors.contains(&ValidationError::UnknownNode("ghost1".to_string()))); + assert!(errors.contains(&ValidationError::UnknownNode("ghost2".to_string()))); + } + + #[test] + fn validation_error_code_and_node_id_accessors() { + assert_eq!(ValidationError::MissingTrigger.code(), "missing_trigger"); + assert_eq!(ValidationError::MissingTrigger.node_id(), None); + assert_eq!( + ValidationError::UnknownNode("ghost".to_string()).code(), + "unknown_node" + ); + assert_eq!( + ValidationError::UnknownNode("ghost".to_string()).node_id(), + Some("ghost") + ); + assert_eq!( + ValidationError::InvalidConditionRouting { + node: "gate".to_string(), + from_port: "main".to_string(), + } + .node_id(), + Some("gate") + ); + assert_eq!( + ValidationError::MultipleTriggers(vec!["a".to_string()]).node_id(), + None + ); + } }