From 58428bf2c4764d4475224886838346de903d964a Mon Sep 17 00:00:00 2001 From: sanil-23 Date: Mon, 17 Aug 2026 23:50:48 +0530 Subject: [PATCH] =?UTF-8?q?feat(adaptive):=20recipe=20authoring=20?= =?UTF-8?q?=E2=80=94=20steps=20in,=20lowered=20graphs=20out?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Nine field runs against a live host established why full-graph authoring fails at small model tiers: the author must emit exact tokens across three foreign syntaxes at once — the graph dialect, the host's tools, and whatever CLI its scripts drive — blind, with feedback one round away. One wrong token anywhere is a dead graph, and the defects surface serially (observed: an invented tool arg, an invented gh field, a split_out on unparsed stdout, a binding embedded in prose — one per round, never converging). So the model no longer writes graphs. It writes a recipe — steps that either `run` a script or `ask` an agent, with `reads` naming which earlier steps' output an agent needs — and `intake::recipe::lower` compiles that into a valid WorkflowGraph deterministically. Every expression, envelope path and edge is machine-generated by code that knows the engine's shapes exactly; the prose-binding failure class is impossible by construction. The lowered graph still walks every downstream gate, and the author feedback rounds now carry the lowering's refusals, each of which states the fix. Because every generated prompt is now an expression, reuse::baked_in learned to scan the QUOTED LITERALS of expressions (paths stay exempt): ="review acme/thing" is a paste however it is spelled, and the old skip-expressions rule would have blinded the keep gate entirely. The node catalogue leaves the authoring prompt — the model never writes node syntax, so teaching it was pure token cost. Test fixtures across author/driver/intake speak the recipe dialect. Verified end to end on a live host: the cheapest model tier authored a working multi-tool research workflow in two attempts, 9m37s wall clock, where six full-graph runs had produced nothing. Co-Authored-By: Claude Fable 5 --- crates/adaptive/src/intake/author.rs | 230 ++----------- crates/adaptive/src/intake/mod.rs | 1 + crates/adaptive/src/intake/recipe.rs | 371 +++++++++++++++++++++ crates/adaptive/src/intake/recipe_tests.rs | 133 ++++++++ crates/adaptive/src/reuse.rs | 72 +++- crates/adaptive/tests/driver.rs | 76 ++--- crates/adaptive/tests/intake.rs | 104 +++--- 7 files changed, 659 insertions(+), 328 deletions(-) create mode 100644 crates/adaptive/src/intake/recipe.rs create mode 100644 crates/adaptive/src/intake/recipe_tests.rs diff --git a/crates/adaptive/src/intake/author.rs b/crates/adaptive/src/intake/author.rs index 0e78667..a43197e 100644 --- a/crates/adaptive/src/intake/author.rs +++ b/crates/adaptive/src/intake/author.rs @@ -1,95 +1,22 @@ -//! Writing a graph when nothing stored fits. +//! Authoring when nothing stored fits: a recipe in, a lowered graph out. //! -//! Two things make the difference between a graph that runs and one that -//! validates and then does nothing, and both are here rather than in the -//! prompt's good intentions: -//! -//! * the node catalogue is **generated from the engine**, not described from -//! memory, so a config field cannot be invented; and -//! * the result is **validated before it is returned**, so an authoring mistake -//! is an error from intake rather than a run-time failure that reads like the -//! work failing. +//! The model never writes graph syntax — see [`super::recipe`] for why and +//! for the surface it writes instead. What stays here is the conversation: +//! bounded feedback rounds, and the gates every candidate walks before it +//! may become an attempt. The gates run on the LOWERED graph; by +//! construction they should all pass, and a construction bug surfacing as a +//! refusal rather than a run-time null is exactly why they still run. use serde_json::Value; use tinyflows::caps::Capabilities; -use tinyflows::catalog::{NodeKindContract, all_contracts}; use tinyflows::model::WorkflowGraph; use tinyflows::store::HostPolicy; use tinyflows::validate::validate_all; -use super::{Attempt, IntakeError, Result, ask}; +use super::{Attempt, IntakeError, Result, ask, recipe}; use crate::contracts::{Approach, Goal, Tier}; use crate::host::HostFacts; -const SYSTEM: &str = "\ -You write a workflow graph that achieves a goal. - -Return JSON: {\"graph\": , \"why\": str, \"inputs\": {name: value}} - -The graph is the engine's own format: -{\"schema_version\": 1, \"name\": str, \"inputs\": [{\"name\", \"required\", \"description\"}], - \"nodes\": [{\"id\", \"kind\", \"name\", \"config\": {...}}], - \"edges\": [{\"from_node\", \"from_port\", \"to_node\", \"to_port\"}]} - -Rules that are checked, not requested: - -- Exactly one `trigger` node, and it is where the graph starts. -- Every node kind and every config field must come from the catalogue below. - It is generated from the engine, so it is the truth; a field you remember - from elsewhere is a field that resolves to null at run time. -- Ports default to `main` on both ends. Name one only where the catalogue says - a node has others (a `condition` emits `true`/`false`; a `loop` emits `body` - and `done`). -- Declare in `inputs` anything the goal supplies as data — a repository, a path, - an id — and read it in config rather than pasting the literal. A graph with - the value baked in is a graph that works once. - -How one node reads another. A config string starting with `=` is an expression; -everything else is a literal. - - =item.name a field of the direct predecessor's output - =nodes.fetch.item.json.body a field of any completed node, by node id - =run.trigger.payload what the trigger carried - =run.inputs.topic a declared workflow input, by its name - =.items | length a leading dot makes the rest a jq program - -There are no braces. `={{ ... }}` is not a binding — it is a jq program that -fails to compile, and a failed program is null, so the step runs with an empty -value and reports success. - -An `=` anywhere but the FIRST character is literal text, not a binding: -`\"about: =run.inputs.topic\"` sends those exact characters to the model. To -put a value inside prose — an agent prompt, a message — the whole string is -one expression, jq with explicit dots: - - =\"Write a poem about \\(.run.inputs.topic)\" - =\"Summarise \" + .run.inputs.repo - -`agent`, `tool_call` and `http_request` wrap their output in -`{json, text, raw}`. Their fields are under `.json`: write -`=nodes.fetch.item.json.body`, never `=nodes.fetch.item.body`. The second form -validates, dry-runs green, and resolves to null every time. - -Design guidance, which is judgement rather than a check: - -- Fewer nodes is better. An `agent` node is a whole coding-agent session on some - hosts — minutes, not seconds — so a graph of eight is usually a worse answer - than a graph of three. -- Use `agent` for work that cannot be specified, and the determined kinds for - everything else. Fetching, reshaping and branching are not agent work. -- Say what a step is for, concretely. The agent running it sees the goal and - that instruction and nothing else — not the other nodes, not what they found. - -Where a section below states what this host permits, it is the machine's own -configuration and is enforced when the graph runs. A graph that ignores it saves -cleanly, validates cleanly, and fails the first time it matters. - -Where a section lists what this episode already tried, write something -DIFFERENT. Not the same graph with a reworded prompt — a different shape: other -nodes, another order, a step that checks what the last attempt assumed. If every -approach you can think of is already on that list, say so in `why` and write the -smallest graph that would establish which assumption is wrong."; - /// Write a graph for `goal`, grounded on the engine's own node catalogue. /// /// # Errors @@ -107,9 +34,8 @@ pub async fn author( ) -> Result { let permitted = facts.render(); let user = format!( - "# Goal\n{}\n\n# Node catalogue — the only kinds and fields that exist\n{}{}{past}", + "# Goal\n{}{}{past}", goal.text.trim(), - catalogue(), if permitted.is_empty() { String::new() } else { @@ -129,7 +55,7 @@ pub async fn author( let mut prompt = user; let mut last: Option = None; for _ in 0..ROUNDS { - let answer = match ask(caps, conn, Tier::Author, SYSTEM, &prompt).await { + let answer = match ask(caps, conn, Tier::Author, recipe::SYSTEM, &prompt).await { Ok(answer) => answer, Err(err) => { last = Some(err); @@ -155,13 +81,7 @@ const ROUNDS: usize = 3; /// One reply through every gate, or why it was refused. fn gated(answer: &Value, facts: &HostFacts, policy: &dyn HostPolicy) -> Result { - let raw = answer - .get("graph") - .cloned() - .ok_or_else(|| IntakeError::Inference("the reply has no `graph` key".to_string()))?; - - let graph: WorkflowGraph = serde_json::from_value(raw) - .map_err(|e| IntakeError::Invalid(format!("not a workflow graph: {e}")))?; + let (graph, mut inputs, why) = recipe::lower(answer)?; // Every failure at once, not the first. A model handed one error fixes it // and returns with the next; handed all four it fixes all four. @@ -204,7 +124,6 @@ fn gated(answer: &Value, facts: &HostFacts, policy: &dyn HostPolicy) -> Result Result String { crate::reuse::digest_hex(&crate::reuse::shape_bytes(graph)) } -/// The node catalogue, rendered for a prompt. -/// -/// Generated from [`all_contracts`] rather than written out here, so a node -/// kind the engine gains appears without this file being touched — and a field -/// this file could describe wrongly cannot exist. -fn catalogue() -> String { - all_contracts() - .iter() - .map(render) - .collect::>() - .join("\n") -} - -fn render(contract: &NodeKindContract) -> String { - let fields = contract - .config_fields - .iter() - .map(|field| { - let mark = if field.required { "*" } else { " " }; - let allowed = match field.enum_values.as_ref() { - Some(values) if !values.is_empty() => format!(" [{}]", values.join("|")), - _ => String::new(), - }; - format!(" {mark}{}: {}{allowed}", field.name, field.value_type) - }) - .collect::>() - .join("\n"); - - // Only the outputs, and only when they are not the default. `from_port` is - // the field an author gets wrong; inputs are almost always `main` and - // listing them on every kind is noise that hides the one that matters. - let outputs = &contract.ports.outputs; - let ports = if outputs.as_slice() == ["main".to_string()] || outputs.is_empty() { - String::new() - } else { - format!(" out ports: {}\n", outputs.join(", ")) - }; - - format!("{}: {}\n{ports}{fields}", contract.kind, contract.summary) -} - #[cfg(test)] mod tests { use super::*; - #[test] - fn the_catalogue_is_generated_from_the_engine() { - // If this list were written by hand it would already be wrong: it is - // the thing the prompt calls the truth. - let rendered = catalogue(); - for kind in [ - "trigger", - "agent", - "tool_call", - "http_request", - "condition", - "loop", - ] { - assert!(rendered.contains(kind), "catalogue is missing {kind}"); - } - } - - #[test] - fn required_fields_are_marked() { - let rendered = catalogue(); - // `trigger_kind` is required on a trigger; a model that misses it - // authors a graph that cannot start. - assert!(rendered.contains("*trigger_kind"), "{rendered}"); - } - - #[test] - fn enum_fields_show_their_allowed_values() { - assert!( - catalogue().contains("manual"), - "trigger_kind's values must be listed" - ); - } - #[tokio::test] async fn a_refused_graph_goes_back_to_the_model_with_the_refusal() { use std::sync::Mutex; @@ -398,27 +243,19 @@ mod tests { .to_string(); let mut prompts = self.prompts.lock().expect("prompt log"); prompts.push(shown.clone()); - let graph = if prompts.len() == 1 { - // No trigger: fails `validate_all`, must come back. - serde_json::json!({ - "schema_version": 1, "name": "broken", - "inputs": [], "nodes": [], "edges": [] - }) - } else { - assert!( - shown.contains("refused"), - "the retry prompt must carry the refusal, got: {shown}" - ); - serde_json::json!({ - "schema_version": 1, "name": "fixed", "inputs": [], - "nodes": [{ - "id": "start", "kind": "trigger", "name": "manual", - "config": { "trigger_kind": "manual" } - }], - "edges": [] - }) - }; - Ok(serde_json::json!({ "graph": graph, "why": "test", "inputs": {} })) + if prompts.len() == 1 { + // No steps: refused by the lowering, must come back. + return Ok(serde_json::json!({ "why": "broken", "inputs": {} })); + } + assert!( + shown.contains("refused"), + "the retry prompt must carry the refusal, got: {shown}" + ); + Ok(serde_json::json!({ + "why": "fixed", + "inputs": {}, + "steps": [{ "id": "do_it", "ask": "Do the thing directly." }] + })) } } @@ -444,6 +281,7 @@ mod tests { .await .expect("the corrected graph must land"); assert_eq!(attempt.graph.name, "fixed"); + assert_eq!(attempt.graph.nodes[1].id, "do_it"); assert_eq!(provider.prompts.lock().expect("prompt log").len(), 2); } @@ -473,15 +311,6 @@ mod tests { .to_string(); let mut prompts = self.prompts.lock().expect("prompt log"); prompts.push(shown.clone()); - let graph = serde_json::json!({ - "schema_version": 1, "name": "needs-topic", - "inputs": [{ "name": "topic", "type": "string", "required": true }], - "nodes": [{ - "id": "start", "kind": "trigger", "name": "manual", - "config": { "trigger_kind": "manual" } - }], - "edges": [] - }); let inputs = if prompts.len() == 1 { serde_json::json!({ "extraneous": "trimmed anyway" }) } else { @@ -491,7 +320,12 @@ mod tests { ); serde_json::json!({ "topic": "flash models", "extraneous": "still here" }) }; - Ok(serde_json::json!({ "graph": graph, "why": "test", "inputs": inputs })) + Ok(serde_json::json!({ + "why": "test", + "declared": [{ "name": "topic", "description": "what about", "required": true }], + "inputs": inputs, + "steps": [{ "id": "write", "ask": "Write it." }] + })) } } diff --git a/crates/adaptive/src/intake/mod.rs b/crates/adaptive/src/intake/mod.rs index 687bfda..f6b237d 100644 --- a/crates/adaptive/src/intake/mod.rs +++ b/crates/adaptive/src/intake/mod.rs @@ -15,6 +15,7 @@ //! A graph leaves here compilable or not at all. mod author; +pub mod recipe; mod select; pub use author::author; diff --git a/crates/adaptive/src/intake/recipe.rs b/crates/adaptive/src/intake/recipe.rs new file mode 100644 index 0000000..451e1d8 --- /dev/null +++ b/crates/adaptive/src/intake/recipe.rs @@ -0,0 +1,371 @@ +//! The simple authoring surface: a recipe of steps, lowered to a real graph. +//! +//! Six field-test runs established why full-graph authoring fails at small +//! model tiers: the author must emit exact tokens across three foreign +//! syntaxes at once — the graph dialect (`=`-expressions, envelope paths), +//! the host's tools, and whatever CLI its scripts drive — blind, with +//! feedback one round away. One wrong token anywhere is a dead graph, and +//! the defects surface serially. +//! +//! So the model does not write graphs. It writes a **recipe** — steps that +//! either `run` a script or `ask` an agent, with `reads` naming which +//! earlier steps' output an agent needs — and [`lower`] compiles that into a +//! valid [`WorkflowGraph`] deterministically. Every expression, envelope +//! path and edge is generated here, by code that knows the engine's shapes +//! exactly. The model's remaining obligations are things models are good +//! at: choosing steps, writing commands, writing prose. +//! +//! The lowered graph still walks every downstream gate. By construction it +//! should pass them all; a construction bug surfacing as a refusal instead +//! of a run-time null is the point of keeping them. + +use serde_json::{Map, Value, json}; +use tinyflows::model::{Edge, InputType, Node, NodeKind, WorkflowGraph, WorkflowInput}; + +use super::IntakeError; + +/// The authoring prompt for the recipe surface. +/// +/// Deliberately free of graph syntax: nothing here teaches nodes, edges, +/// bindings or envelopes, because the model never writes them. +pub const SYSTEM: &str = "\ +You plan how to achieve a goal as a short sequence of steps. + +Return JSON: +{\"why\": str, + \"declared\": [{\"name\": str, \"description\": str, \"required\": bool}], + \"inputs\": {name: value}, + \"steps\": [ + {\"id\": str, \"run\": str}, + {\"id\": str, \"ask\": str, \"reads\": [str], \"worker\": str?} + ]} + +- Steps execute in the order listed. Each step has an `id` (a short + snake_case name) and exactly ONE of: + run a shell script. It must PRINT its result to stdout — a result in a + file or nowhere is a result the next step cannot see. Print JSON + when the output is structured. + ask an instruction for an AI agent. Say exactly what to produce and + that it should be produced directly. The agent's reply text is the + step's output. +- `reads` (ask steps only): the ids of EARLIER steps whose output this agent + needs. Their output is attached to the instruction automatically — do not + describe how to fetch it, and do not use placeholders for it. +- `worker` (ask steps only, optional): a worker this host lists, when the + step must run somewhere specific. Omit it otherwise. +- `declared`: the workflow's inputs — anything the goal supplies as data (a + repository, a topic, an id), so the plan works for the NEXT goal of its + kind with different values. `inputs` supplies this run's value for every + required one. Declared values are attached to ask steps automatically. +- The LAST step's output is the run's answer: make it the step that produces + the deliverable. + +Keep it short. One step is often right: an agent asked for the whole +deliverable, with the goal's data declared as inputs. Use `run` steps for +deterministic fetching or checking, not for judgement. + +Where a section below states what this host permits, it is enforced when the +plan runs. Where a section lists what this episode already tried, produce a +DIFFERENT plan, not the same one reworded."; + +/// One parsed step of a recipe. +struct Step { + id: String, + action: Action, + reads: Vec, +} + +enum Action { + Run(String), + Ask { + prompt: String, + worker: Option, + }, +} + +/// Lower a recipe reply into a runnable graph plus its run values. +/// +/// # Errors +/// [`IntakeError::Invalid`] naming every structural problem at once — absent +/// steps, duplicate or malformed ids, `reads` pointing forward or nowhere, a +/// step that is both `run` and `ask` or neither. The messages are written +/// for the feedback round: each states the fix, not just the fault. +pub fn lower(answer: &Value) -> Result<(WorkflowGraph, Map, String), IntakeError> { + let why = answer["why"].as_str().unwrap_or_default().to_string(); + let steps = parse_steps(answer)?; + let declared = parse_declared(answer); + let inputs = answer["inputs"].as_object().cloned().unwrap_or_default(); + + let mut nodes = vec![Node { + id: "start".into(), + kind: NodeKind::Trigger, + type_version: 1, + name: "start".into(), + config: json!({ "trigger_kind": "manual" }), + ports: Vec::new(), + position: None, + }]; + let mut edges = Vec::new(); + let mut previous = "start".to_string(); + + for step in &steps { + let node = match &step.action { + Action::Run(script) => Node { + id: step.id.clone(), + kind: NodeKind::Shell, + type_version: 1, + name: step.id.clone(), + config: json!({ "script": script }), + ports: Vec::new(), + position: None, + }, + Action::Ask { prompt, worker } => { + let mut config = json!({ + "prompt": ask_expression(prompt, &step.reads, &steps, &declared) + }); + if let Some(worker) = worker { + config["agent_ref"] = json!(worker); + } + Node { + id: step.id.clone(), + kind: NodeKind::Agent, + type_version: 1, + name: step.id.clone(), + config, + ports: Vec::new(), + position: None, + } + } + }; + nodes.push(node); + edges.push(Edge { + from_node: previous.clone(), + from_port: "main".into(), + to_node: step.id.clone(), + to_port: "main".into(), + }); + previous = step.id.clone(); + } + + let graph = WorkflowGraph { + schema_version: 1, + id: None, + name: graph_name(&why, &steps), + inputs: declared + .iter() + .map(|(name, description, required)| { + let input = WorkflowInput::new(name.clone(), InputType::String) + .with_description(description.clone()); + if *required { input.required() } else { input } + }) + .collect(), + agents: Vec::new(), + nodes, + edges, + }; + Ok((graph, inputs, why)) +} + +/// The generated prompt expression for an ask step. +/// +/// A jq program the model never sees: the instruction as a quoted literal, +/// then every declared input, then each read step's output through the path +/// its kind actually produces — `stdout` for a script (with the engine's +/// pre-parsed `stdout_json` unnecessary here: the agent reads text), `text` +/// for an upstream agent. Missing values render as an explicit marker +/// rather than vanishing, because an agent told "output: (missing)" says so +/// instead of improvising. +fn ask_expression( + prompt: &str, + reads: &[String], + steps: &[Step], + declared: &[(String, String, bool)], +) -> String { + let mut program = format!("={}", jq_quote(prompt)); + for (name, _, _) in declared { + program.push_str(&format!( + " + {} + ((.run.inputs.{name} // \"(not provided)\") | tostring)", + jq_quote(&format!("\n\n# Input `{name}`\n")) + )); + } + for read in reads { + let path = match steps + .iter() + .find(|step| &step.id == read) + .map(|step| &step.action) + { + Some(Action::Run(_)) => format!("(.nodes.{read}.item.json.stdout // \"(no output)\")"), + _ => format!("(.nodes.{read}.item.json.text // \"(no output)\")"), + }; + program.push_str(&format!( + " + {} + ({path} | tostring)", + jq_quote(&format!("\n\n# Output of step `{read}`\n")) + )); + } + program +} + +/// A string as a jq literal: quoted, with the characters jq treats specially +/// escaped. Newlines become `\n` so the program stays one line. +fn jq_quote(text: &str) -> String { + let mut quoted = String::with_capacity(text.len() + 2); + quoted.push('"'); + for ch in text.chars() { + match ch { + '"' => quoted.push_str("\\\""), + '\\' => quoted.push_str("\\\\"), + '\n' => quoted.push_str("\\n"), + '\r' => quoted.push_str("\\r"), + '\t' => quoted.push_str("\\t"), + other => quoted.push(other), + } + } + quoted.push('"'); + quoted +} + +fn parse_steps(answer: &Value) -> Result, IntakeError> { + let raw = answer["steps"] + .as_array() + .filter(|steps| !steps.is_empty()) + .ok_or_else(|| { + IntakeError::Invalid( + "the reply has no `steps` — return at least one step with an `id` and a \ + `run` script or an `ask` instruction" + .to_string(), + ) + })?; + + let mut problems = Vec::new(); + let mut steps: Vec = Vec::new(); + for (index, step) in raw.iter().enumerate() { + let id = sanitize_id(step["id"].as_str().unwrap_or_default()); + if id.is_empty() { + problems.push(format!("step {index} has no usable `id`")); + continue; + } + if id == "start" || steps.iter().any(|existing| existing.id == id) { + problems.push(format!("step id `{id}` is taken — ids must be unique")); + continue; + } + let run = step["run"] + .as_str() + .map(str::trim) + .filter(|s| !s.is_empty()); + let ask = step["ask"] + .as_str() + .map(str::trim) + .filter(|s| !s.is_empty()); + let action = match (run, ask) { + (Some(script), None) => Action::Run(script.to_string()), + (None, Some(prompt)) => Action::Ask { + prompt: prompt.to_string(), + worker: step["worker"] + .as_str() + .map(str::trim) + .filter(|s| !s.is_empty()) + .map(ToString::to_string), + }, + (Some(_), Some(_)) => { + problems.push(format!( + "step `{id}` has both `run` and `ask` — split it into two steps" + )); + continue; + } + (None, None) => { + problems.push(format!( + "step `{id}` has neither a `run` script nor an `ask` instruction" + )); + continue; + } + }; + let mut reads = Vec::new(); + if let Some(raw_reads) = step["reads"].as_array() { + for read in raw_reads { + let read = sanitize_id(read.as_str().unwrap_or_default()); + if steps.iter().any(|existing| existing.id == read) { + reads.push(read); + } else { + problems.push(format!( + "step `{id}` reads `{read}`, which is not an EARLIER step id" + )); + } + } + } + if matches!(action, Action::Run(_)) && !reads.is_empty() { + problems.push(format!( + "step `{id}`: `reads` only works on ask steps — a run script sees nothing" + )); + } + steps.push(Step { id, action, reads }); + } + if !problems.is_empty() { + return Err(IntakeError::Invalid(problems.join("; "))); + } + Ok(steps) +} + +fn parse_declared(answer: &Value) -> Vec<(String, String, bool)> { + answer["declared"] + .as_array() + .map(|declared| { + declared + .iter() + .filter_map(|input| { + let name = sanitize_id(input["name"].as_str().unwrap_or_default()); + if name.is_empty() { + return None; + } + Some(( + name, + input["description"] + .as_str() + .unwrap_or_default() + .to_string(), + input["required"].as_bool().unwrap_or(false), + )) + }) + .collect() + }) + .unwrap_or_default() +} + +/// A graph name from the recipe: the first ask step's opening words, or the +/// step ids — something a shelf listing can show, not an id. +fn graph_name(why: &str, steps: &[Step]) -> String { + let head: String = why.split_whitespace().take(6).collect::>().join(" "); + if !head.is_empty() { + return head; + } + steps + .iter() + .map(|step| step.id.as_str()) + .collect::>() + .join(" → ") +} + +/// Identifiers the engine and jq both accept: lowercase, alnum and `_`. +fn sanitize_id(raw: &str) -> String { + let mut id: String = raw + .trim() + .chars() + .map(|ch| { + if ch.is_ascii_alphanumeric() { + ch.to_ascii_lowercase() + } else { + '_' + } + }) + .collect(); + while id.starts_with('_') { + id.remove(0); + } + while id.ends_with('_') { + id.pop(); + } + id +} + +#[cfg(test)] +#[path = "recipe_tests.rs"] +mod tests; diff --git a/crates/adaptive/src/intake/recipe_tests.rs b/crates/adaptive/src/intake/recipe_tests.rs new file mode 100644 index 0000000..84aec23 --- /dev/null +++ b/crates/adaptive/src/intake/recipe_tests.rs @@ -0,0 +1,133 @@ +//! The lowering is where authoring mistakes used to live — every test here +//! is a defect class a real episode paid for. + +use serde_json::json; +use tinyflows::model::NodeKind; +use tinyflows::validate::validate_all; + +use super::lower; + +fn review_recipe() -> serde_json::Value { + json!({ + "why": "fetch then review", + "declared": [ + { "name": "repo", "description": "owner/name to review", "required": true } + ], + "inputs": { "repo": "acme/thing" }, + "steps": [ + { "id": "fetch", "run": "gh issue list --json number,title" }, + { "id": "review", "ask": "Write the verdict report.", "reads": ["fetch"] } + ] + }) +} + +#[test] +fn a_recipe_lowers_to_a_graph_that_validates() { + let (graph, inputs, why) = lower(&review_recipe()).expect("lowers"); + assert!( + validate_all(&graph).is_empty(), + "a lowered graph must always validate: {:?}", + validate_all(&graph) + ); + assert_eq!(graph.nodes.len(), 3, "trigger + two steps"); + assert_eq!(graph.nodes[1].kind, NodeKind::Shell); + assert_eq!(graph.nodes[2].kind, NodeKind::Agent); + assert_eq!(inputs["repo"], "acme/thing"); + assert_eq!(why, "fetch then review"); +} + +#[test] +fn the_generated_prompt_is_one_expression_with_the_right_envelope_paths() { + let (graph, _, _) = lower(&review_recipe()).expect("lowers"); + let prompt = graph.nodes[2].config["prompt"].as_str().expect("prompt"); + // The whole string is an expression — the prose-binding failure class + // cannot occur by construction. + assert!(prompt.starts_with('='), "{prompt}"); + // A shell upstream is read through `stdout`, the field its kind emits — + // the exact path a blind author guessed wrong three runs straight. + assert!(prompt.contains(".nodes.fetch.item.json.stdout"), "{prompt}"); + // Declared inputs are attached without the model writing any binding. + assert!(prompt.contains(".run.inputs.repo"), "{prompt}"); + // Absent values surface as markers, not silent nothing. + assert!(prompt.contains("(no output)"), "{prompt}"); +} + +#[test] +fn an_agent_upstream_is_read_through_text_not_stdout() { + let recipe = json!({ + "why": "chain of agents", + "steps": [ + { "id": "draft", "ask": "Draft it." }, + { "id": "polish", "ask": "Polish the draft.", "reads": ["draft"] } + ] + }); + let (graph, _, _) = lower(&recipe).expect("lowers"); + let prompt = graph.nodes[2].config["prompt"].as_str().expect("prompt"); + assert!(prompt.contains(".nodes.draft.item.json.text"), "{prompt}"); +} + +#[test] +fn quotes_and_newlines_in_an_ask_survive_as_a_valid_jq_literal() { + let recipe = json!({ + "why": "quoting", + "steps": [ + { "id": "speak", "ask": "Say \"hello\",\nthen stop." } + ] + }); + let (graph, _, _) = lower(&recipe).expect("lowers"); + let prompt = graph.nodes[1].config["prompt"].as_str().expect("prompt"); + assert!(prompt.contains("\\\"hello\\\""), "{prompt}"); + assert!(prompt.contains("\\n"), "{prompt}"); +} + +#[test] +fn a_worker_on_an_ask_step_becomes_agent_ref() { + let recipe = json!({ + "why": "placed work", + "steps": [ + { "id": "build", "ask": "Build it.", "worker": "ci-box" } + ] + }); + let (graph, _, _) = lower(&recipe).expect("lowers"); + assert_eq!(graph.nodes[1].config["agent_ref"], "ci-box"); +} + +#[test] +fn every_structural_problem_is_reported_at_once_with_the_fix() { + let recipe = json!({ + "why": "broken", + "steps": [ + { "id": "fetch", "run": "true", "reads": ["later"] }, + { "id": "fetch", "ask": "duplicate id" }, + { "id": "confused", "run": "true", "ask": "both" }, + { "id": "empty" } + ] + }); + let err = lower(&recipe).expect_err("refused").to_string(); + for fragment in ["EARLIER", "unique", "split it into two", "neither"] { + assert!(err.contains(fragment), "missing `{fragment}` in: {err}"); + } +} + +#[test] +fn a_reply_with_no_steps_says_what_to_return() { + let err = lower(&json!({ "why": "empty" })) + .expect_err("refused") + .to_string(); + assert!(err.contains("at least one step"), "{err}"); +} + +#[test] +fn ids_are_sanitized_into_engine_and_jq_safe_names() { + let recipe = json!({ + "why": "messy ids", + "steps": [ + { "id": " Fetch-Issues! ", "run": "true" }, + { "id": "review", "ask": "Review.", "reads": ["Fetch-Issues!"] } + ] + }); + let (graph, _, _) = lower(&recipe).expect("lowers"); + assert_eq!(graph.nodes[1].id, "fetch_issues"); + let prompt = graph.nodes[2].config["prompt"].as_str().expect("prompt"); + assert!(prompt.contains(".nodes.fetch_issues.item"), "{prompt}"); +} diff --git a/crates/adaptive/src/reuse.rs b/crates/adaptive/src/reuse.rs index 3336162..c7920ff 100644 --- a/crates/adaptive/src/reuse.rs +++ b/crates/adaptive/src/reuse.rs @@ -84,12 +84,18 @@ pub fn baked_in(graph: &WorkflowGraph, inputs: &serde_json::Map) collect_strings(&node.config, &mut leaves); for leaf in leaves { for value in &distinctive { - // An expression that happens to mention the value is still - // reading it from somewhere; a literal is not. - if leaf.starts_with('=') { - continue; - } - if leaf.contains(value) && !found.iter().any(|f| f == value) { + // Expressions are scanned too, but only their QUOTED + // literals: `=.run.inputs.repo` reads the value from the + // run and is fine; `="review acme/thing"` welded it in — + // and since generated prompts are all expressions now, an + // expression-shaped paste is the common shape, not the + // exception. + let pasted = if leaf.starts_with('=') { + quoted_literals(&leaf).iter().any(|lit| lit.contains(value)) + } else { + leaf.contains(value) + }; + if pasted && !found.iter().any(|f| f == value) { found.push((*value).to_string()); } } @@ -131,6 +137,38 @@ pub(crate) fn shape_bytes(graph: &WorkflowGraph) -> Vec { /// old 28-bit truncation also put birthday collisions within reach of a few /// tens of thousands of records; 64 bits does not. FNV-1a is fixed forever, /// fits in six lines, and needs no dependency. +/// The quoted string literals of a jq expression, unescaped enough to +/// substring-search: `="a \"b\"" + .x` yields `a "b"`. +fn quoted_literals(expression: &str) -> Vec { + let mut literals = Vec::new(); + let mut current: Option = None; + let mut chars = expression.chars(); + while let Some(ch) = chars.next() { + match ch { + '"' => match current.take() { + Some(literal) => literals.push(literal), + None => current = Some(String::new()), + }, + '\\' if current.is_some() => { + if let (Some(literal), Some(escaped)) = (current.as_mut(), chars.next()) { + literal.push(match escaped { + 'n' => '\n', + 't' => '\t', + 'r' => '\r', + other => other, + }); + } + } + other => { + if let Some(literal) = current.as_mut() { + literal.push(other); + } + } + } + } + literals +} + pub(crate) fn digest_hex(bytes: &[u8]) -> String { let mut hash: u64 = 0xcbf2_9ce4_8422_2325; for byte in bytes { @@ -210,14 +248,26 @@ mod tests { } #[test] - fn an_expression_mentioning_the_value_is_still_reading_it() { - // `=run.inputs.repo | ascii_downcase` names nothing literally, but a jq - // program can contain the text and still be a binding rather than a - // paste. Anything starting with `=` is resolved at run time. - let graph = graph_with(json!({ "prompt": "=\"acme/thing\" " })); + fn an_expression_reading_the_value_by_path_is_not_a_paste() { + // `=run.inputs.repo | ascii_downcase` resolves the value at run time — + // the graph works for the next repo too. + let graph = graph_with(json!({ "prompt": "=run.inputs.repo | ascii_downcase" })); assert!(baked_in(&graph, &inputs(&[("repo", "acme/thing")])).is_empty()); } + #[test] + fn a_value_welded_into_an_expressions_quoted_literal_is_a_paste() { + // `="acme/thing"` evaluates to exactly the pasted text: expression + // syntax around a literal changes nothing about its reusability. Since + // recipe lowering made every generated prompt an expression, this is + // the common shape of a paste, not an edge case. + let graph = graph_with(json!({ "prompt": "=\"review acme/thing directly\"" })); + assert_eq!( + baked_in(&graph, &inputs(&[("repo", "acme/thing")])), + vec!["acme/thing".to_string()] + ); + } + #[test] fn plain_short_words_prove_nothing_and_are_not_evidence() { // `main` is the default port name on every edge in the graph, so a node diff --git a/crates/adaptive/tests/driver.rs b/crates/adaptive/tests/driver.rs index 1539afe..219de35 100644 --- a/crates/adaptive/tests/driver.rs +++ b/crates/adaptive/tests/driver.rs @@ -12,7 +12,6 @@ use serde_json::{Value, json}; use tinyflows::caps::mock::mock_capabilities; use tinyflows::caps::{Capabilities, LlmProvider}; use tinyflows::error::Result as EngineResult; -use tinyflows::model::{Edge, Node, NodeKind, WorkflowGraph}; use tinyflows::store::{FileWorkflowStore, WorkflowStore}; use tinyflows_adaptive::contracts::Goal; use tinyflows_adaptive::driver::{Clock, Loop}; @@ -75,42 +74,6 @@ fn caps_with(llm: Arc) -> Capabilities { } } -fn tiny(name: &str) -> WorkflowGraph { - WorkflowGraph { - schema_version: 1, - id: Some(name.into()), - name: name.into(), - inputs: Vec::new(), - agents: Vec::new(), - nodes: vec![ - Node { - id: "start".into(), - kind: NodeKind::Trigger, - type_version: 1, - name: "manual".into(), - config: json!({ "trigger_kind": "manual" }), - ports: Vec::new(), - position: None, - }, - Node { - id: "done".into(), - kind: NodeKind::Transform, - type_version: 1, - name: "done".into(), - config: json!({ "set": { "ok": true } }), - ports: Vec::new(), - position: None, - }, - ], - edges: vec![Edge { - from_node: "start".into(), - from_port: "main".into(), - to_node: "done".into(), - to_port: "main".into(), - }], - } -} - fn store(tag: &str) -> Arc { let root = std::env::temp_dir().join(format!("adaptive-driver-{}-{tag}", std::process::id())); let _ = std::fs::remove_dir_all(&root); @@ -123,9 +86,9 @@ fn store(tag: &str) -> Arc { fn authoring() -> Arc { Always::new(json!({ - "graph": tiny("attempt"), "why": "nothing stored fits", "inputs": {}, + "steps": [{ "id": "attempt", "run": "echo attempt-done" }], })) } @@ -335,19 +298,21 @@ async fn a_run_drives_to_a_stand_down_and_consolidates_once() { /// A graph parameterised by a declared input, which is what the authoring /// prompt asks for and what makes a procedure worth keeping. -fn parameterised() -> WorkflowGraph { - let mut g = tiny("review"); - g.inputs = vec![ - tinyflows::model::WorkflowInput::new("repo", tinyflows::model::InputType::String) - .required(), - ]; - g.nodes[1].config = json!({ "set": { "target": "=run.inputs.repo" } }); - g +fn parameterised() -> Value { + json!({ + "why": "review", + "declared": [{ "name": "repo", "description": "the repository", "required": true }], + "inputs": { "repo": "acme/thing" }, + "steps": [{ + "id": "review", + "ask": "Review the open pull requests and summarise them directly." + }], + }) } /// Authors `graph`, judges every run satisfied, and answers the naming call. struct Succeeds { - graph: WorkflowGraph, + authored: Value, reusable: bool, seen: Mutex>, } @@ -365,18 +330,14 @@ impl LlmProvider for Succeeds { "description": "Reviews the open pull requests on a repository. Takes the repository as an input.", "reusable": self.reusable, }), - _ => json!({ - "graph": self.graph, - "why": "nothing stored fits", - "inputs": { "repo": "acme/thing" }, - }), + _ => self.authored.clone(), }) } } -fn succeeding(graph: WorkflowGraph, reusable: bool) -> Arc { +fn succeeding(authored: Value, reusable: bool) -> Arc { Arc::new(Succeeds { - graph, + authored, reusable, seen: Mutex::new(Vec::new()), }) @@ -443,10 +404,11 @@ async fn a_graph_that_was_authored_and_worked_becomes_a_stored_procedure() { #[tokio::test] async fn a_graph_that_pasted_its_inputs_is_not_kept() { - // Same run, same success — but the goal's specifics are welded into a node, - // so it matches one task and never another. No model is asked. + // Same run, same success — but the goal's specifics are welded into a + // step, so it matches one task and never another. No model is asked. let mut baked = parameterised(); - baked.nodes[1].config = json!({ "set": { "target": "acme/thing" } }); + baked["steps"][0]["ask"] = + json!("Review the open pull requests on acme/thing and summarise them."); let llm = succeeding(baked, true); let caps = Capabilities { diff --git a/crates/adaptive/tests/intake.rs b/crates/adaptive/tests/intake.rs index 7bfb95f..9424e00 100644 --- a/crates/adaptive/tests/intake.rs +++ b/crates/adaptive/tests/intake.rs @@ -81,6 +81,21 @@ fn empty_store(tag: &str) -> (FileWorkflowStore, std::path::PathBuf) { } /// A minimal graph that validates: one trigger, one transform. +/// An author reply in the recipe surface, with `label` distinguishing its +/// lowered shape (and so its fingerprint) from any other reply's. +fn authored_reply(label: &str, required_input: Option<&str>) -> Value { + let mut reply = json!({ + "why": label, + "inputs": {}, + "steps": [{ "id": "work", "ask": format!("Do the {label} work directly.") }], + }); + if let Some(name) = required_input { + reply["declared"] = json!([{ "name": name, "description": "", "required": true }]); + reply["inputs"] = json!({ name: "acme/thing" }); + } + reply +} + fn tiny_graph(name: &str, required_input: Option<&str>) -> WorkflowGraph { WorkflowGraph { schema_version: 1, @@ -135,11 +150,7 @@ fn stored(id: &str, description: &str, required_input: Option<&str>) -> Workflow async fn an_empty_store_authors_without_asking_whether_to_select() { // With nothing to choose from the answer can only be "none". Spending a // call to be told so is the cost of every cold start. - let llm = std::sync::Arc::new(Scripted::new(vec![json!({ - "graph": tiny_graph("fresh", None), - "why": "nothing stored", - "inputs": {}, - })])); + let llm = std::sync::Arc::new(Scripted::new(vec![authored_reply("fresh", None)])); let caps = caps_with(llm.clone()); let (store, _root) = empty_store("1"); let ledger = MemoryLedger::new(); @@ -163,8 +174,8 @@ async fn an_empty_store_authors_without_asking_whether_to_select() { "exactly one call: the authoring one" ); assert!( - llm.prompts()[0].contains("Node catalogue"), - "authoring must be grounded on the catalogue" + llm.prompts()[0].contains("You plan how to achieve a goal"), + "authoring must speak the recipe surface, not graph syntax" ); } @@ -212,7 +223,7 @@ async fn a_matching_workflow_is_selected_and_its_graph_is_loaded() { async fn declining_falls_through_to_authoring() { let llm = std::sync::Arc::new(Scripted::new(vec![ json!({ "workflow_id": null, "why": "none of these fetch anything" }), - json!({ "graph": tiny_graph("written", None), "why": "had to write one", "inputs": {} }), + authored_reply("written", None), ])); let caps = caps_with(llm.clone()); let (store, _root) = empty_store("3"); @@ -246,11 +257,7 @@ async fn a_workflow_already_tried_this_episode_is_not_offered_again() { // The property the whole retry edge rests on. Without it attempt two // re-selects what attempt one already failed on, and the episode pays // twice for one dead end. - let llm = std::sync::Arc::new(Scripted::new(vec![json!({ - "graph": tiny_graph("written", None), - "why": "the only candidate was already spent", - "inputs": {}, - })])); + let llm = std::sync::Arc::new(Scripted::new(vec![authored_reply("written", None)])); let caps = caps_with(llm.clone()); let (store, _root) = empty_store("4"); store @@ -335,7 +342,7 @@ async fn a_selection_that_still_cannot_bind_falls_back_to_authoring() { let llm = std::sync::Arc::new(Scripted::new(vec![ json!({ "workflow_id": "needs-repo", "why": "matches", "inputs": {} }), json!({ "workflow_id": "needs-repo", "why": "still sure", "inputs": {} }), - json!({ "graph": tiny_graph("fresh", None), "why": "wrote one instead", "inputs": {} }), + authored_reply("fresh", None), ])); let caps = caps_with(llm.clone()); let (store, _root) = empty_store("5b"); @@ -404,7 +411,7 @@ async fn inputs_the_graph_never_declared_are_trimmed_before_the_engine_sees_them async fn a_hallucinated_workflow_id_reads_as_a_decline() { let llm = std::sync::Arc::new(Scripted::new(vec![ json!({ "workflow_id": "pr-reviewer", "why": "close, but no such id" }), - json!({ "graph": tiny_graph("written", None), "why": "wrote one", "inputs": {} }), + authored_reply("written", None), ])); let caps = caps_with(llm.clone()); let (store, _root) = empty_store("6"); @@ -437,8 +444,7 @@ async fn an_authored_graph_that_does_not_validate_is_an_error_not_a_return_value // that reads like the work failing. The author retries with the refusal // fed back, so the script holds a model that stays wrong for every round. let broken = json!({ - "graph": { "schema_version": 1, "name": "empty", "nodes": [], "edges": [] }, - "why": "forgot the trigger", + "why": "forgot the steps", "inputs": {}, }); let llm = std::sync::Arc::new(Scripted::new(vec![broken.clone(), broken.clone(), broken])); @@ -462,11 +468,7 @@ async fn an_authored_graph_that_does_not_validate_is_an_error_not_a_return_value #[tokio::test] async fn a_disabled_workflow_is_never_offered() { - let llm = std::sync::Arc::new(Scripted::new(vec![json!({ - "graph": tiny_graph("written", None), - "why": "the only one was disabled", - "inputs": {}, - })])); + let llm = std::sync::Arc::new(Scripted::new(vec![authored_reply("written", None)])); let caps = caps_with(llm.clone()); let (store, _root) = empty_store("8"); let mut off = stored("switched-off", "would have matched", None); @@ -498,22 +500,13 @@ async fn a_graph_naming_a_worker_this_host_lacks_is_refused_before_it_runs() { // The whole point of collecting host facts. Without this the graph saves // cleanly, validates cleanly, and fails at run time — usually overnight, // to nobody watching. - let mut agent_graph = tiny_graph("uses-an-agent", None); - agent_graph.nodes[1] = Node { - id: "work".into(), - kind: NodeKind::Agent, - type_version: 1, - name: "do it".into(), - config: json!({ "prompt": "do the thing", "agent_ref": "desktop" }), - ports: Vec::new(), - position: None, - }; - agent_graph.edges[0].to_node = "work".into(); - + // // Three copies: the author feeds refusals back, and this model never // learns that the worker does not exist. let insistent = json!({ - "graph": agent_graph, "why": "needs an agent", "inputs": {}, + "why": "needs an agent", + "inputs": {}, + "steps": [{ "id": "work", "ask": "do the thing", "worker": "desktop" }], }); let llm = std::sync::Arc::new(Scripted::new(vec![ insistent.clone(), @@ -554,8 +547,12 @@ async fn a_graph_naming_a_worker_this_host_lacks_is_refused_before_it_runs() { #[tokio::test] async fn the_authoring_prompt_carries_what_the_host_permits() { + // The facts below say agent work must name a worker, so the reply's ask + // step names one — the same gate this test exists to see rendered. let llm = std::sync::Arc::new(Scripted::new(vec![json!({ - "graph": tiny_graph("fine", None), "why": "ok", "inputs": {}, + "why": "fine", + "inputs": {}, + "steps": [{ "id": "work", "ask": "Do it directly.", "worker": "laptop" }], })])); let caps = caps_with(llm.clone()); let (store, _root) = empty_store("facts-rendered"); @@ -626,11 +623,7 @@ async fn repaired_family( async fn offered(store: &FileWorkflowStore, ledger: &MemoryLedger) -> String { let llm = std::sync::Arc::new(Scripted::new(vec![ json!({"workflow_id": "none"}), - json!({ - "graph": tiny_graph("fallback", None), - "why": "declined", - "inputs": {}, - }), + authored_reply("fallback", None), ])); let caps = caps_with(llm.clone()); let _ = decide( @@ -759,11 +752,7 @@ async fn the_author_is_shown_what_this_episode_already_tried() { // because nothing told it otherwise. The exclusion list only guards // *selection*; authoring has no structural guard at all. let (store, ledger, _root) = with_history("retry-1").await; - let llm = std::sync::Arc::new(Scripted::new(vec![json!({ - "graph": tiny_graph("third-idea", None), - "why": "the first two both trusted the model for figures", - "inputs": {}, - })])); + let llm = std::sync::Arc::new(Scripted::new(vec![authored_reply("third-idea", None)])); let caps = caps_with(llm.clone()); decide( @@ -785,7 +774,7 @@ async fn the_author_is_shown_what_this_episode_already_tried() { "{prompt}" ); assert!(prompt.contains("it invented the figures"), "{prompt}"); - assert!(prompt.contains("write something\nDIFFERENT"), "{prompt}"); + assert!(prompt.contains("DIFFERENT plan"), "{prompt}"); } #[tokio::test] @@ -843,11 +832,7 @@ async fn lessons_from_other_episodes_reach_the_planner() { .await .expect("promote"); - let llm = std::sync::Arc::new(Scripted::new(vec![json!({ - "graph": tiny_graph("informed", None), - "why": "nothing stored", - "inputs": {}, - })])); + let llm = std::sync::Arc::new(Scripted::new(vec![authored_reply("informed", None)])); let caps = caps_with(llm.clone()); decide( @@ -873,11 +858,7 @@ async fn a_first_attempt_is_told_nothing_it_would_have_to_ignore() { // empty "already tried" heading reads as a claim that something was. let (store, _root) = empty_store("retry-4"); let ledger = MemoryLedger::new(); - let llm = std::sync::Arc::new(Scripted::new(vec![json!({ - "graph": tiny_graph("first", None), - "why": "nothing stored", - "inputs": {}, - })])); + let llm = std::sync::Arc::new(Scripted::new(vec![authored_reply("first", None)])); let caps = caps_with(llm.clone()); decide( @@ -906,11 +887,10 @@ async fn two_authored_attempts_leave_two_distinct_signatures() { let mut signatures = Vec::new(); for (n, name) in [(0, "shape-one"), (1, "shape-two")] { - let llm = std::sync::Arc::new(Scripted::new(vec![json!({ - "graph": tiny_graph(name, if n == 1 { Some("repo") } else { None }), - "why": "nothing stored", - "inputs": { "repo": "acme/thing" }, - })])); + let llm = std::sync::Arc::new(Scripted::new(vec![authored_reply( + name, + if n == 1 { Some("repo") } else { None }, + )])); let attempt = decide( &Goal::new("write the weekly report"), "ep-sigs",