From 18a8cd9819131d606f87bbf38c8630df5c302fa9 Mon Sep 17 00:00:00 2001 From: sanil-23 Date: Tue, 18 Aug 2026 20:15:53 +0530 Subject: [PATCH 1/6] feat(adaptive): a `use` step, so a plan can call a saved workflow MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit A goal is often two jobs that already exist, or one that exists plus work on top. The loop could not express that: `select` returns at most one workflow id, and a recipe could only `run` a script or `ask` an agent. So a goal that needed two stored procedures got a graph that reinvented both — badly, and discarding everything the originals had proved about themselves. The engine has supported composition all along (`NodeKind::SubWorkflow`, by `workflow_id`, resolved through the host's `WorkflowResolver`, with input forwarding and a depth guard). Nothing could reach it, because the authoring surface had no way to say so. This adds the third step kind: { "id": "audit", "use": "pr-audit-review", "with": { "repo": "@input.repo" } } lowered to a `sub_workflow` node exactly as `run` and `ask` lower to `shell` and `agent`. The model still writes no graph syntax. `@input.x` and `@step.y` are the whole reference vocabulary — they become the engine expressions that read them, so a child's input can be wired to live data without the author knowing jq exists. By reference, never inlined: the callee keeps its own identity, its own scores, and whatever it becomes next. Four refusals at intake rather than mid-run, each standing for a run it saves: an id nobody offered (the resolver would raise a capability error after the earlier steps had been paid for), a required input left unfilled, a `with` key the callee never declared — silently dropping that one is the worst case, because the run then completes while the model believes it passed something — and a `@`-reference to an input that was never declared. The author is shown what it may call. That listing is deliberately NOT the chooser's: it is unfiltered by what the episode already tried, because a workflow that fell short as the whole answer is exactly the one worth calling as a part. `WorkflowSummary` already carries declared inputs, so it costs no extra store traffic. Two fixes fall out of building it. **An agent's prose is at `item.text`, not `item.json.text`.** The two are siblings on the envelope — `json` is the structured value, `text` is the prose derived from it — so `item.json.text` reads a `text` field *inside* the structured value, which a prose reply has not got. It resolved to null, and every `reads` of an agent step has been rendering "(no output)": the silent-null class this surface exists to make impossible, sitting inside the surface. A script's `stdout` genuinely is nested, which is what made the two paths look symmetric enough to write side by side. Found by evaluating a generated prompt instead of string-matching its spelling — which is what let the wrong spelling ship, so the new tests evaluate. **Reading a `use` step needs a projection.** A `sub_workflow` node emits the child's entire final run state, so a naive read hands the next agent the child's bookkeeping with the deliverable buried in it. The projection keeps each child step's readable leaf, labelled with the step it came from — not just the last one, because the child's node slots are a JSON object whose key order is alphabetical rather than the order they ran. Written defensively at every hop, since it walks a state this graph did not choose the shape of, and evaluated against real and empty child states so a jq error cannot fail the parent instead of the step. Also lists each candidate's declared inputs to the chooser, which was being asked to supply values for inputs it had never been seen. Stacks on #63 (open): its shell-source-key commit is the parent here. Merge that first and this rebases to its own commit. --- crates/adaptive/src/intake/author.rs | 26 +- crates/adaptive/src/intake/mod.rs | 41 ++- crates/adaptive/src/intake/recipe.rs | 361 +++++++++++++++++++-- crates/adaptive/src/intake/recipe_tests.rs | 359 +++++++++++++++++++- crates/adaptive/src/intake/select.rs | 26 +- 5 files changed, 772 insertions(+), 41 deletions(-) diff --git a/crates/adaptive/src/intake/author.rs b/crates/adaptive/src/intake/author.rs index a43197e..5cb632b 100644 --- a/crates/adaptive/src/intake/author.rs +++ b/crates/adaptive/src/intake/author.rs @@ -16,6 +16,7 @@ use tinyflows::validate::validate_all; use super::{Attempt, IntakeError, Result, ask, recipe}; use crate::contracts::{Approach, Goal, Tier}; use crate::host::HostFacts; +use recipe::Callable; /// Write a graph for `goal`, grounded on the engine's own node catalogue. /// @@ -27,19 +28,29 @@ use crate::host::HostFacts; pub async fn author( goal: &Goal, facts: &HostFacts, + callables: &[Callable], policy: &dyn HostPolicy, past: &str, caps: &Capabilities, conn: Option<&str>, ) -> Result { let permitted = facts.render(); + // The callable listing goes beside what the host permits, because it is + // the same kind of fact: these exist, the rest do not. A plan naming one + // that is not here is refused at intake, not discovered mid-run. + let offered = recipe::render_callables(callables); let user = format!( - "# Goal\n{}{}{past}", + "# Goal\n{}{}{}{past}", goal.text.trim(), if permitted.is_empty() { String::new() } else { format!("\n\n{permitted}") + }, + if offered.is_empty() { + String::new() + } else { + format!("\n\n{offered}") } ); @@ -62,7 +73,7 @@ pub async fn author( continue; } }; - match gated(&answer, facts, policy) { + match gated(&answer, facts, callables, policy) { Ok(attempt) => return Ok(attempt), Err(err) => { prompt = format!( @@ -80,8 +91,13 @@ pub async fn author( 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 (graph, mut inputs, why) = recipe::lower(answer)?; +fn gated( + answer: &Value, + facts: &HostFacts, + callables: &[Callable], + policy: &dyn HostPolicy, +) -> Result { + let (graph, mut inputs, why) = recipe::lower(answer, callables)?; // 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. @@ -273,6 +289,7 @@ mod tests { let attempt = author( &Goal::new("do the thing"), &HostFacts::unknown(), + &[], &Permissive, "", &caps, @@ -343,6 +360,7 @@ mod tests { let attempt = author( &Goal::new("do the thing"), &HostFacts::unknown(), + &[], &Permissive, "", &caps, diff --git a/crates/adaptive/src/intake/mod.rs b/crates/adaptive/src/intake/mod.rs index f6b237d..8842e77 100644 --- a/crates/adaptive/src/intake/mod.rs +++ b/crates/adaptive/src/intake/mod.rs @@ -19,6 +19,7 @@ pub mod recipe; mod select; pub use author::author; +pub use recipe::{Callable, render_callables}; pub use select::{Candidate, bind, select}; use serde_json::{Map, Value}; @@ -113,6 +114,12 @@ pub async fn decide( let rows = ledger.rows(episode).await?; let tried = crate::ledger::signatures(&rows); let candidates = catalogue(store, ledger, &tried).await?; + // Two views of the same shelf, and the difference is the point. The + // chooser sees what it has not already tried, because repeating a + // selection cannot teach the episode anything. The author sees the WHOLE + // shelf, because a workflow that fell short as the entire answer is + // exactly the one worth calling as one step of a bigger plan. + let callables = callables(store)?; // Both planners see the same past, in the same words. The exclusion list // stops a *selection* being repeated, but nothing structural stops the @@ -169,7 +176,7 @@ pub async fn decide( Err(err) => return Err(err), } } - return author(goal, facts, store.policy(), ¬ed, caps, conn) + return author(goal, facts, &callables, store.policy(), ¬ed, caps, conn) .await .map(|attempt| Attempt { lessons_shown: shown, @@ -183,7 +190,7 @@ pub async fn decide( Err(err) => return Err(err), } } - author(goal, facts, store.policy(), &past, caps, conn) + author(goal, facts, &callables, store.policy(), &past, caps, conn) .await .map(|attempt| Attempt { lessons_shown: shown, @@ -210,6 +217,31 @@ pub async fn decide( /// clause is not a choice, it is noise, and a planner asked to make it is being /// asked to guess. Which member survives is decided on score, never on being /// the newest — see [`crate::promotion`]. +/// Every enabled workflow, as something a plan may call. +/// +/// Unfiltered on purpose — see the note at the call site. Built from the same +/// listing the chooser's catalogue reads, so composition costs no extra store +/// traffic: `WorkflowSummary` already carries the declared inputs, precisely +/// so a caller need not fetch a whole graph to learn what it takes. +fn callables(store: &dyn WorkflowStore) -> Result> { + Ok(store + .list() + .map_err(|e| IntakeError::Store(e.to_string()))? + .into_iter() + .filter(|summary| summary.enabled) + .map(|summary| Callable { + id: summary.id, + name: summary.name, + description: summary.description, + inputs: summary + .inputs + .into_iter() + .map(|input| (input.name, input.required)) + .collect(), + }) + .collect()) +} + async fn catalogue( store: &dyn WorkflowStore, ledger: &dyn Ledger, @@ -236,6 +268,11 @@ async fn catalogue( node_count: summary.node_count, applied: score.applied, helped: score.helped, + inputs: summary + .inputs + .into_iter() + .map(|input| (input.name, input.required)) + .collect(), }); } collapse_families(out, ledger).await diff --git a/crates/adaptive/src/intake/recipe.rs b/crates/adaptive/src/intake/recipe.rs index 2a2f8af..590f2d1 100644 --- a/crates/adaptive/src/intake/recipe.rs +++ b/crates/adaptive/src/intake/recipe.rs @@ -37,7 +37,8 @@ Return JSON: \"inputs\": {name: value}, \"steps\": [ {\"id\": str, \"run\": str}, - {\"id\": str, \"ask\": str, \"reads\": [str], \"worker\": str?} + {\"id\": str, \"ask\": str, \"reads\": [str], \"worker\": str?}, + {\"id\": str, \"use\": str, \"with\": {name: value}} ]} - Steps execute in the order listed. Each step has an `id` (a short @@ -48,11 +49,17 @@ Return JSON: 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. + use the id of a saved workflow, from the list of ones you can call. It + runs as one step and its output is what it produced. - `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. +- `with` (use steps only): a value for each input the saved workflow declares. + Write `\"@input.\"` to pass one of YOUR declared inputs through, or + `\"@step.\"` to pass an earlier step's output. Anything else is used as + a literal value. - `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 @@ -64,10 +71,86 @@ 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. +A `use` step is for a WHOLE job somebody already solved — the goal asks for +two of them, or for one plus your own work on top. It is not for scavenging: +if you only want part of what a saved workflow does, write the step yourself. + 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 saved workflow a plan may call as a step. +/// +/// The same rows the chooser weighs, minus its scores: composition asks "does +/// this job exist" rather than "is this the whole answer", so the record that +/// decides a *selection* is noise here — and unlike the chooser's list this +/// one is not filtered by what the episode already tried, because a workflow +/// that fell short alone is exactly the one worth calling as a part. +#[derive(Debug, Clone)] +pub struct Callable { + /// The id a `use` step names. + pub id: String, + /// Display name; falls back to the id when blank. + pub name: String, + /// What it does — the only thing that can justify calling it. + pub description: String, + /// Its declared inputs: name and whether it is required. + pub inputs: Vec<(String, bool)>, +} + +impl Callable { + /// The prompt listing for one callable workflow. + fn render(&self) -> String { + let name = if self.name.is_empty() { + &self.id + } else { + &self.name + }; + let description = if self.description.is_empty() { + "(no description — do not call this; nobody can say what it does)" + } else { + &self.description + }; + let inputs = if self.inputs.is_empty() { + "takes no inputs".to_string() + } else { + let listed: Vec = self + .inputs + .iter() + .map(|(name, required)| { + if *required { + name.clone() + } else { + format!("{name} (optional)") + } + }) + .collect(); + format!("with: {}", listed.join(", ")) + }; + format!( + "- id: {}\n name: {name}\n {inputs}\n {description}", + self.id + ) + } +} + +/// The prompt section listing what a `use` step may name. +/// +/// Empty when nothing is callable, so a cold store's author is not told about +/// a step kind it cannot use — an offer with an empty list reads as a missing +/// list, and the model invents an id to fill it. +#[must_use] +pub fn render_callables(callables: &[Callable]) -> String { + if callables.is_empty() { + return String::new(); + } + let listed: Vec = callables.iter().map(Callable::render).collect(); + format!( + "# Saved workflows you can call with a `use` step\n{}", + listed.join("\n") + ) +} + /// One parsed step of a recipe. struct Step { id: String, @@ -81,6 +164,11 @@ enum Action { prompt: String, worker: Option, }, + /// Call a saved workflow, forwarding values for its declared inputs. + Use { + workflow_id: String, + with: Map, + }, } /// Lower a recipe reply into a runnable graph plus its run values. @@ -90,10 +178,13 @@ enum Action { /// 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> { +pub fn lower( + answer: &Value, + callables: &[Callable], +) -> 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 steps = parse_steps(answer, callables, &declared)?; let inputs = answer["inputs"].as_object().cloned().unwrap_or_default(); let mut nodes = vec![Node { @@ -139,6 +230,18 @@ pub fn lower(answer: &Value) -> Result<(WorkflowGraph, Map, Strin position: None, } } + Action::Use { workflow_id, with } => Node { + id: step.id.clone(), + kind: NodeKind::SubWorkflow, + type_version: 1, + name: step.id.clone(), + // `workflow_id`, not an inline graph: the child is resolved + // from the store at run time, so the callee keeps its own + // identity, its own scores, and whatever it becomes next. + config: json!({ "workflow_id": workflow_id, "inputs": with }), + ports: Vec::new(), + position: None, + }, }; nodes.push(node); edges.push(Edge { @@ -192,14 +295,10 @@ fn ask_expression( )); } 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)\")"), - }; + let path = format!( + "({} // \"(no output)\")", + output_of(read, kind_of(read, steps)) + ); program.push_str(&format!( " + {} + ({path} | tostring)", jq_quote(&format!("\n\n# Output of step `{read}`\n")) @@ -208,6 +307,73 @@ fn ask_expression( program } +/// Which kind of step `id` is, for choosing how to read its output. +fn kind_of<'a>(id: &str, steps: &'a [Step]) -> Option<&'a Action> { + steps + .iter() + .find(|step| step.id == id) + .map(|step| &step.action) +} + +/// The jq path that yields a step's output as something readable. +/// +/// Each node kind puts its result somewhere different, and this is the one +/// place that knows where. +/// +/// **An agent's prose is at `item.text`, not `item.json.text`.** The two are +/// siblings on the envelope — `json` is the structured value, `text` is the +/// prose `text_of` derived from it — so `item.json.text` reads a `text` field +/// *inside* the structured value, which a prose reply does not have. It +/// resolved to null, and every `reads` of an agent step rendered +/// "(no output)": the exact silent-null class this whole surface exists to +/// make impossible, sitting inside the surface. A script's `stdout` genuinely +/// is nested (`json` holds `{exit_code, stdout}`), which is what made the two +/// paths look symmetric enough to write side by side. +/// +/// For a called workflow it is a projection, because a `sub_workflow` node +/// emits the child's entire final run state, every node of it, wrapped around +/// the answer. Handing an agent that whole object would bury the deliverable +/// in the child's own bookkeeping. +/// +/// The projection keeps each child step's readable leaf, labelled with the +/// step id it came from. Not just the last one: the child's node slots are a +/// JSON object, whose key order is alphabetical rather than the order the +/// steps ran, so "the last one" is not a thing this expression can ask for. +/// Everything the child produced, named, is the honest answer — and for the +/// ordinary child whose one agent step writes the deliverable, it is exactly +/// that deliverable. +fn output_of(id: &str, action: Option<&Action>) -> String { + match action { + Some(Action::Run(_)) => format!(".nodes.{id}.item.json.stdout"), + Some(Action::Use { .. }) => child_answer(id), + _ => format!(".nodes.{id}.item.text"), + } +} + +/// The projection that turns a called workflow's run state into prose. +/// +/// Written defensively at every hop — a slot with no `items`, an empty array, +/// an item whose payload is not an object — because this walks a *child's* +/// state, whose shape this graph did not choose. A jq error here would fail +/// the parent node rather than the step that actually produced nothing. +fn child_answer(id: &str) -> String { + // Two different shapes in one expression, which is the whole hazard here. + // `.nodes.{id}.item` is the *scope* projection — the child's final run + // state. Inside it, `.nodes.` is a raw run-state slot, which stores + // serialized items (`{"json": …}`) rather than the bare payloads the scope + // exposes. So the outer hop drops `items`/`json` and the inner one needs + // both. + format!( + "([((.nodes.{id}.item.nodes // {{}}) | to_entries[]) \ + | . as $step \ + | ((.value.items // []) | .[-1] | .json) as $out \ + | ($out.text? // $out.stdout? // $out.json.stdout? // empty) as $said \ + | \"## \" + $step.key + \"\\n\" + ($said | tostring)] \ + | join(\"\\n\\n\") \ + | if . == \"\" then null else . end)" + ) +} + /// 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 { @@ -227,14 +393,18 @@ fn jq_quote(text: &str) -> String { quoted } -fn parse_steps(answer: &Value) -> Result, IntakeError> { +fn parse_steps( + answer: &Value, + callables: &[Callable], + declared: &[(String, String, bool)], +) -> 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" + `run` script, an `ask` instruction or a `use` workflow id" .to_string(), ) })?; @@ -259,9 +429,24 @@ fn parse_steps(answer: &Value) -> Result, IntakeError> { .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 { + let called = step["use"] + .as_str() + .map(str::trim) + .filter(|s| !s.is_empty()); + let given = [run.is_some(), ask.is_some(), called.is_some()] + .iter() + .filter(|given| **given) + .count(); + if given > 1 { + problems.push(format!( + "step `{id}` has more than one of `run`, `ask` and `use` — a step does \ + exactly one thing, so split it" + )); + continue; + } + let action = match (run, ask, called) { + (Some(script), _, _) => Action::Run(script.to_string()), + (_, Some(prompt), _) => Action::Ask { prompt: prompt.to_string(), worker: step["worker"] .as_str() @@ -269,15 +454,19 @@ fn parse_steps(answer: &Value) -> Result, IntakeError> { .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; + (_, _, Some(workflow_id)) => { + match call(&id, workflow_id, &step["with"], callables, declared, &steps) { + Ok(action) => action, + Err(problem) => { + problems.push(problem); + continue; + } + } } - (None, None) => { + (None, None, None) => { problems.push(format!( - "step `{id}` has neither a `run` script nor an `ask` instruction" + "step `{id}` has none of a `run` script, an `ask` instruction or a \ + `use` workflow id" )); continue; } @@ -300,6 +489,12 @@ fn parse_steps(answer: &Value) -> Result, IntakeError> { "step `{id}`: `reads` only works on ask steps — a run script sees nothing" )); } + if matches!(action, Action::Use { .. }) && !reads.is_empty() { + problems.push(format!( + "step `{id}`: `reads` only works on ask steps — a called workflow takes \ + what you pass it, so put `\"@step.\"` in `with` instead" + )); + } steps.push(Step { id, action, reads }); } if !problems.is_empty() { @@ -308,6 +503,126 @@ fn parse_steps(answer: &Value) -> Result, IntakeError> { Ok(steps) } +/// Build one `use` step, refusing everything that could only fail later. +/// +/// Three checks, and each stands for a run this saves: an id nobody offered +/// is a hallucination that the resolver would turn into a mid-run capability +/// error; a required input left unfilled fails the child's own declaration +/// check after the parent has already spent its earlier steps; and a `with` +/// key the child never declared is a value the model believes it is passing +/// and the child will never see. +fn call( + id: &str, + workflow_id: &str, + with: &Value, + callables: &[Callable], + declared: &[(String, String, bool)], + earlier: &[Step], +) -> Result { + let Some(callable) = callables.iter().find(|c| c.id == workflow_id) else { + let offered: Vec<&str> = callables.iter().map(|c| c.id.as_str()).collect(); + return Err(if offered.is_empty() { + format!( + "step `{id}` uses `{workflow_id}`, but this host has no saved workflows to \ + call — write the step yourself" + ) + } else { + format!( + "step `{id}` uses `{workflow_id}`, which is not one of the workflows you \ + can call ({})", + offered.join(", ") + ) + }); + }; + + let given = match with { + Value::Null => Map::new(), + Value::Object(fields) => fields.clone(), + _ => { + return Err(format!( + "step `{id}`: `with` must be an object mapping `{workflow_id}`'s input \ + names to values" + )); + } + }; + + for (name, _) in callable.inputs.iter().filter(|(_, required)| *required) { + if !given.contains_key(name) { + return Err(format!( + "step `{id}`: `{workflow_id}` requires the input `{name}` and `with` does \ + not supply it" + )); + } + } + let mut forwarded = Map::new(); + for (name, value) in given { + if !callable.inputs.iter().any(|(input, _)| input == &name) { + return Err(format!( + "step `{id}`: `{workflow_id}` declares no input `{name}` — it takes {}", + if callable.inputs.is_empty() { + "none".to_string() + } else { + callable + .inputs + .iter() + .map(|(input, _)| input.as_str()) + .collect::>() + .join(", ") + } + )); + } + forwarded.insert( + name, + forward(&value, declared, earlier).map_err(|why| format!("step `{id}`: {why}"))?, + ); + } + Ok(Action::Use { + workflow_id: workflow_id.to_string(), + with: forwarded, + }) +} + +/// Turn one `with` value into what the child should receive. +/// +/// `@input.x` and `@step.y` become the engine expressions that read them; a +/// plain value is passed as itself. The sigil exists so a model can wire a +/// child's input to live data without writing jq — the same bargain the rest +/// of this surface makes. +fn forward( + value: &Value, + declared: &[(String, String, bool)], + earlier: &[Step], +) -> Result { + let Some(reference) = value.as_str().and_then(|text| text.strip_prefix('@')) else { + return Ok(value.clone()); + }; + if let Some(name) = reference.strip_prefix("input.") { + let name = sanitize_id(name); + if !declared.iter().any(|(declared, _, _)| declared == &name) { + return Err(format!( + "`@input.{name}` names an input you did not declare — add it to `declared`" + )); + } + return Ok(Value::String(format!("=.run.inputs.{name}"))); + } + if let Some(step) = reference.strip_prefix("step.") { + let step = sanitize_id(step); + let Some(action) = kind_of(&step, earlier) else { + return Err(format!( + "`@step.{step}` names a step that is not an EARLIER step of this plan" + )); + }; + return Ok(Value::String(format!( + "={}", + output_of(&step, Some(action)) + ))); + } + Err(format!( + "`{reference}` is not a reference this understands — write `@input.`, \ + `@step.`, or a plain value" + )) +} + fn parse_declared(answer: &Value) -> Vec<(String, String, bool)> { answer["declared"] .as_array() diff --git a/crates/adaptive/src/intake/recipe_tests.rs b/crates/adaptive/src/intake/recipe_tests.rs index ab0cc0a..4833c82 100644 --- a/crates/adaptive/src/intake/recipe_tests.rs +++ b/crates/adaptive/src/intake/recipe_tests.rs @@ -23,7 +23,7 @@ fn review_recipe() -> serde_json::Value { #[test] fn a_recipe_lowers_to_a_graph_that_validates() { - let (graph, inputs, why) = lower(&review_recipe()).expect("lowers"); + let (graph, inputs, why) = lower(&review_recipe(), &[]).expect("lowers"); assert!( validate_all(&graph).is_empty(), "a lowered graph must always validate: {:?}", @@ -43,7 +43,7 @@ fn a_recipe_lowers_to_a_graph_that_validates() { #[test] fn the_generated_prompt_is_one_expression_with_the_right_envelope_paths() { - let (graph, _, _) = lower(&review_recipe()).expect("lowers"); + 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. @@ -66,9 +66,9 @@ fn an_agent_upstream_is_read_through_text_not_stdout() { { "id": "polish", "ask": "Polish the draft.", "reads": ["draft"] } ] }); - let (graph, _, _) = lower(&recipe).expect("lowers"); + 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}"); + assert!(prompt.contains(".nodes.draft.item.text"), "{prompt}"); } #[test] @@ -79,7 +79,7 @@ fn quotes_and_newlines_in_an_ask_survive_as_a_valid_jq_literal() { { "id": "speak", "ask": "Say \"hello\",\nthen stop." } ] }); - let (graph, _, _) = lower(&recipe).expect("lowers"); + 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}"); @@ -93,7 +93,7 @@ fn a_worker_on_an_ask_step_becomes_agent_ref() { { "id": "build", "ask": "Build it.", "worker": "ci-box" } ] }); - let (graph, _, _) = lower(&recipe).expect("lowers"); + let (graph, _, _) = lower(&recipe, &[]).expect("lowers"); assert_eq!(graph.nodes[1].config["agent_ref"], "ci-box"); } @@ -108,15 +108,15 @@ fn every_structural_problem_is_reported_at_once_with_the_fix() { { "id": "empty" } ] }); - let err = lower(&recipe).expect_err("refused").to_string(); - for fragment in ["EARLIER", "unique", "split it into two", "neither"] { + let err = lower(&recipe, &[]).expect_err("refused").to_string(); + for fragment in ["EARLIER", "unique", "so split it", "has none of"] { 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" })) + let err = lower(&json!({ "why": "empty" }), &[]) .expect_err("refused") .to_string(); assert!(err.contains("at least one step"), "{err}"); @@ -135,7 +135,7 @@ fn the_lowered_shell_config_satisfies_the_engines_own_contract() { "why": "fetch", "steps": [{ "id": "fetch", "run": "echo hi" }] }); - let (graph, _, _) = lower(&recipe).expect("lowers"); + let (graph, _, _) = lower(&recipe, &[]).expect("lowers"); let shell = tinyflows::catalog::all_contracts() .iter() .find(|contract| contract.kind == "shell") @@ -173,8 +173,345 @@ fn ids_are_sanitized_into_engine_and_jq_safe_names() { { "id": "review", "ask": "Review.", "reads": ["Fetch-Issues!"] } ] }); - let (graph, _, _) = lower(&recipe).expect("lowers"); + 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}"); } + +// --------------------------------------------------------------------------- +// `use` steps: calling a saved workflow as one step of a plan. +// --------------------------------------------------------------------------- + +use super::Callable; + +fn audit() -> Callable { + Callable { + id: "pr-audit-review".to_string(), + name: "PR audit review".to_string(), + description: "reviews a pull request and posts the verdict".to_string(), + inputs: vec![("repo".to_string(), true), ("depth".to_string(), false)], + } +} + +fn compose_recipe() -> serde_json::Value { + json!({ + "why": "audit the PR, then summarise what it found", + "declared": [ + { "name": "repo", "description": "owner/name", "required": true } + ], + "inputs": { "repo": "acme/thing" }, + "steps": [ + { "id": "audit", "use": "pr-audit-review", "with": { "repo": "@input.repo" } }, + { "id": "summary", "ask": "Summarise the audit in three bullets.", + "reads": ["audit"] } + ] + }) +} + +#[test] +fn a_use_step_lowers_to_a_sub_workflow_node_that_references_the_callee() { + let (graph, _, _) = lower(&compose_recipe(), &[audit()]).expect("lowers"); + assert!( + validate_all(&graph).is_empty(), + "{:?}", + validate_all(&graph) + ); + let node = graph + .nodes + .iter() + .find(|node| node.id == "audit") + .expect("the use step became a node"); + assert_eq!(node.kind, NodeKind::SubWorkflow); + // By reference, never inlined: the callee keeps its own identity, its own + // scores, and whatever it becomes next. + assert_eq!(node.config["workflow_id"], json!("pr-audit-review")); + assert!( + node.config.get("workflow").is_none(), + "an inlined child would fork the callee at authoring time" + ); + // `@input.repo` became the expression that reads the parent's run input. + assert_eq!(node.config["inputs"]["repo"], json!("=.run.inputs.repo")); +} + +#[test] +fn a_step_reference_in_with_reads_the_earlier_step_the_way_its_kind_produces() { + let recipe = json!({ + "why": "fetch the diff, then hand it to a saved reviewer", + "declared": [], + "steps": [ + { "id": "diff", "run": "git diff" }, + { "id": "review", "use": "reviewer", "with": { "patch": "@step.diff" } } + ] + }); + let callable = Callable { + id: "reviewer".to_string(), + name: String::new(), + description: "reviews a patch".to_string(), + inputs: vec![("patch".to_string(), true)], + }; + let (graph, _, _) = lower(&recipe, &[callable]).expect("lowers"); + let node = graph + .nodes + .iter() + .find(|node| node.id == "review") + .expect("the use step became a node"); + // A run step's output is its stdout, and the reference knows that without + // the model having to. + assert_eq!( + node.config["inputs"]["patch"], + json!("=.nodes.diff.item.json.stdout") + ); +} + +#[test] +fn a_use_step_naming_a_workflow_nobody_offered_is_refused_at_intake() { + // Not deferred to the resolver: a hallucinated id would surface as a + // capability error mid-run, after the earlier steps had already been paid + // for, and be attributed to the work rather than to the plan. + let err = lower(&compose_recipe(), &[]) + .expect_err("refused") + .to_string(); + assert!(err.contains("no saved workflows to call"), "{err}"); + + let other = Callable { + id: "something-else".to_string(), + ..audit() + }; + let err = lower(&compose_recipe(), &[other]) + .expect_err("refused") + .to_string(); + assert!( + err.contains("something-else"), + "names what IS callable: {err}" + ); +} + +#[test] +fn a_use_step_that_omits_a_required_input_is_refused_with_the_name() { + let recipe = json!({ + "why": "call it with nothing", + "steps": [{ "id": "audit", "use": "pr-audit-review", "with": {} }] + }); + let err = lower(&recipe, &[audit()]).expect_err("refused").to_string(); + assert!(err.contains("requires the input `repo`"), "{err}"); +} + +#[test] +fn a_with_key_the_callee_never_declared_is_refused_rather_than_dropped() { + // Silently dropping it would leave the model believing it passed a value + // the child will never see — the worst kind of pass, because the run + // completes. + let recipe = json!({ + "why": "wrong input name", + "declared": [{ "name": "repo", "description": "owner/name", "required": true }], + "steps": [{ + "id": "audit", "use": "pr-audit-review", + "with": { "repo": "@input.repo", "reponame": "acme/thing" } + }] + }); + let err = lower(&recipe, &[audit()]).expect_err("refused").to_string(); + assert!(err.contains("declares no input `reponame`"), "{err}"); + assert!(err.contains("repo, depth"), "says what it does take: {err}"); +} + +#[test] +fn an_input_reference_to_something_undeclared_is_refused() { + let recipe = json!({ + "why": "reference an input that does not exist", + "declared": [], + "steps": [{ + "id": "audit", "use": "pr-audit-review", "with": { "repo": "@input.repo" } + }] + }); + let err = lower(&recipe, &[audit()]).expect_err("refused").to_string(); + assert!(err.contains("did not declare"), "{err}"); +} + +#[test] +fn reads_on_a_use_step_points_at_with_instead() { + let recipe = json!({ + "why": "reads does not apply", + "declared": [{ "name": "repo", "description": "owner/name", "required": true }], + "steps": [ + { "id": "diff", "run": "git diff" }, + { "id": "audit", "use": "pr-audit-review", "reads": ["diff"], + "with": { "repo": "@input.repo" } } + ] + }); + let err = lower(&recipe, &[audit()]).expect_err("refused").to_string(); + assert!(err.contains("`with`"), "{err}"); +} + +#[test] +fn a_step_reading_a_use_step_gets_the_childs_answer_not_its_run_state() { + // The defect this projection exists for: a `sub_workflow` node emits the + // child's ENTIRE final run state, so a naive read hands the next agent the + // child's bookkeeping with the deliverable buried in it. + let (graph, _, _) = lower(&compose_recipe(), &[audit()]).expect("lowers"); + let summary = graph + .nodes + .iter() + .find(|node| node.id == "summary") + .expect("the ask step"); + let prompt = summary.config["prompt"] + .as_str() + .expect("a generated prompt expression"); + + // Run the generated expression against a real child run state, through the + // engine's own evaluator — the only thing that proves the projection is + // valid jq and picks the right leaves. + let state = json!({ + "run": { "inputs": { "repo": "acme/thing" } }, + "inputs": { "repo": "acme/thing" }, + "nodes": { "audit": { "item": child_run_state(), "items": [child_run_state()] } } + }); + let rendered = tinyflows::expr::resolve(&json!(prompt), &state); + let rendered = rendered.as_str().expect("resolves to a string"); + + assert!( + rendered.contains("Requesting changes."), + "the child's deliverable must reach the reader: {rendered}" + ); + assert!( + rendered.contains("3 files changed"), + "and so must every other leaf it produced: {rendered}" + ); + assert!( + rendered.contains("## verdict"), + "each labelled with the child step it came from: {rendered}" + ); + assert!( + !rendered.contains("trigger"), + "but not the child's own bookkeeping: {rendered}" + ); +} + +#[test] +fn a_child_that_produced_nothing_readable_says_so_rather_than_erroring() { + // The projection walks a state this graph did not choose the shape of, so + // every hop is written defensively; a jq error here would fail the parent + // node instead of reporting the step that produced nothing. + let (graph, _, _) = lower(&compose_recipe(), &[audit()]).expect("lowers"); + let prompt = graph + .nodes + .iter() + .find(|node| node.id == "summary") + .expect("the ask step") + .config["prompt"] + .as_str() + .expect("a generated prompt expression") + .to_string(); + + for state in [ + json!({ "run": {}, "nodes": { "audit": { "item": { "nodes": {} }, "items": [] } } }), + json!({ "run": {}, "nodes": { "audit": { "item": null, "items": [] } } }), + json!({ "run": {}, "nodes": {} }), + ] { + let rendered = tinyflows::expr::resolve(&json!(prompt), &state); + let rendered = rendered.as_str().unwrap_or_default(); + assert!( + rendered.contains("(no output)"), + "empty child state {state} rendered: {rendered}" + ); + } +} + +#[test] +fn the_callable_listing_names_the_inputs_a_call_must_fill() { + // A model asked to supply `with` for inputs it was never shown is a model + // guessing — the same defect the chooser had. + let rendered = super::render_callables(&[audit()]); + assert!(rendered.contains("pr-audit-review"), "{rendered}"); + assert!( + rendered.contains("with: repo, depth (optional)"), + "{rendered}" + ); + assert!( + super::render_callables(&[]).is_empty(), + "a cold store offers no `use` list at all, rather than an empty one" + ); +} + +/// A child workflow's final run state, in the shape the engine records it. +/// +/// Raw run-state slots, so items are serialized (`{"json": …}`) — one shape in +/// from the parent's scope projection, which exposes bare payloads. Getting +/// that boundary wrong is precisely what `child_answer` has to survive. +fn child_run_state() -> serde_json::Value { + json!({ + "run": { "trigger": [], "inputs": { "repo": "acme/thing" } }, + "nodes": { + "fetch_pr": { "items": [{ "json": { + "json": { "exit_code": 0, "stdout": "3 files changed" }, + "text": null, "raw": {} + } }] }, + "verdict": { "items": [{ "json": { + "json": "Requesting changes.", + "text": "Requesting changes.", + "raw": "Requesting changes." + } }] } + } + }) +} + +#[test] +fn an_agents_prose_is_read_from_the_envelopes_text_not_from_inside_its_json() { + // The regression this file exists to prevent, found in this file's own + // output: `item.json.text` reads a `text` field inside the STRUCTURED + // value, which a prose reply has not got, so every `reads` of an agent + // step rendered "(no output)". Evaluated rather than string-matched — + // asserting the path spelling is what let the wrong spelling ship. + 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"); + + let scope = json!({ + "run": {}, "inputs": null, "item": null, "items": [], + "nodes": { "draft": { "item": { + "json": "The draft, in prose.", + "text": "The draft, in prose.", + "raw": "The draft, in prose." + } } } + }); + let rendered = tinyflows::expr::resolve(&json!(prompt), &scope); + let rendered = rendered.as_str().expect("resolves to a string"); + assert!( + rendered.contains("The draft, in prose."), + "an upstream agent's reply must reach the next step: {rendered}" + ); + assert!( + !rendered.contains("(no output)"), + "it rendered the missing-value marker instead: {rendered}" + ); +} + +#[test] +fn a_scripts_stdout_is_read_from_inside_its_json_because_that_is_where_it_is() { + // The asymmetry that made the agent path look right: a shell node's + // structured value genuinely holds `{exit_code, stdout}`, so this one IS + // nested. Pinned by evaluation so the two never get "harmonised". + let (graph, _, _) = lower(&review_recipe(), &[]).expect("lowers"); + let prompt = graph.nodes[2].config["prompt"].as_str().expect("prompt"); + let scope = json!({ + "run": { "inputs": { "repo": "acme/thing" } }, + "inputs": { "repo": "acme/thing" }, "item": null, "items": [], + "nodes": { "fetch": { "item": { + "json": { "exit_code": 0, "stdout": "#41 flaky test" }, + "text": null, "raw": {} + } } } + }); + let rendered = tinyflows::expr::resolve(&json!(prompt), &scope); + let rendered = rendered.as_str().expect("resolves to a string"); + assert!(rendered.contains("#41 flaky test"), "{rendered}"); + assert!( + rendered.contains("acme/thing"), + "declared inputs too: {rendered}" + ); +} diff --git a/crates/adaptive/src/intake/select.rs b/crates/adaptive/src/intake/select.rs index 244d29a..39c07d4 100644 --- a/crates/adaptive/src/intake/select.rs +++ b/crates/adaptive/src/intake/select.rs @@ -33,6 +33,13 @@ pub struct Candidate { pub applied: u32, /// Times that ended satisfied. pub helped: u32, + /// Its declared inputs: name and whether it is required. + /// + /// Listed because the chooser is asked to supply values for them. It was + /// being asked to fill inputs it had never been shown, which is a guess + /// dressed as a binding — and a required input guessed wrong is a run + /// that fails after the choice has already been made. + pub inputs: Vec<(String, bool)>, } impl Candidate { @@ -54,8 +61,24 @@ impl Candidate { 0 => "never run".to_string(), applied => format!("run {applied}×, satisfied {}×", self.helped), }; + let inputs = if self.inputs.is_empty() { + String::new() + } else { + let listed: Vec = self + .inputs + .iter() + .map(|(name, required)| { + if *required { + name.clone() + } else { + format!("{name} (optional)") + } + }) + .collect(); + format!("\n inputs: {}", listed.join(", ")) + }; format!( - "- id: {}\n name: {name}\n steps: {}, {record}\n {description}", + "- id: {}\n name: {name}\n steps: {}, {record}{inputs}\n {description}", self.id, self.node_count ) } @@ -210,6 +233,7 @@ mod tests { node_count: 4, applied, helped, + inputs: vec![("repo".to_string(), true)], } } From 281ac82312b3b67bf7ccc6d01a60003aef63de58 Mon Sep 17 00:00:00 2001 From: sanil-23 Date: Tue, 18 Aug 2026 20:40:16 +0530 Subject: [PATCH 2/6] fix(adaptive): guard the child projection by type, not by `?` MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Found by running it: a composed plan reached its combining agent with nothing, eleven attempts in a row, while the two called workflows had both plainly succeeded and written their poems into the run. `?` does not do here what it looks like it does. In `jaq`, `.a?` over a non-object yields no output as expected — but a two-hop `.a.b?` fails the whole enclosing expression instead of yielding nothing, so one bad slot resolved the entire prompt to null. And a child always has a bad slot: its trigger holds the seeded item ARRAY, not an object. Guarded with an explicit `type == "object"` instead, which needs no assumption about how a suppression operator propagates. The unit fixture now carries a real child run state, trigger slot included — the synthetic one whose slots were all objects passed against the broken spelling, which is the whole reason this shipped. --- crates/adaptive/src/intake/recipe.rs | 18 +++++++++++++++--- crates/adaptive/src/intake/recipe_tests.rs | 10 ++++++++-- 2 files changed, 23 insertions(+), 5 deletions(-) diff --git a/crates/adaptive/src/intake/recipe.rs b/crates/adaptive/src/intake/recipe.rs index 590f2d1..5b9e29a 100644 --- a/crates/adaptive/src/intake/recipe.rs +++ b/crates/adaptive/src/intake/recipe.rs @@ -354,8 +354,18 @@ fn output_of(id: &str, action: Option<&Action>) -> String { /// /// Written defensively at every hop — a slot with no `items`, an empty array, /// an item whose payload is not an object — because this walks a *child's* -/// state, whose shape this graph did not choose. A jq error here would fail -/// the parent node rather than the step that actually produced nothing. +/// state, whose shape this graph did not choose. A failure here would take +/// down the parent node rather than report the step that produced nothing, +/// and a child always has at least one payload that is not an object: its +/// trigger slot holds the seeded item **array**. +/// +/// Guarded with an explicit `type == "object"` rather than the `?` operator, +/// which does not do what it looks like it does here. In `jaq`, `.a?` over a +/// non-object yields no output as expected, but a two-hop `.a.b?` fails the +/// whole enclosing expression instead — so the "defensive" spelling of this +/// projection resolved the entire prompt to null, and every composed plan +/// reached its combining agent with nothing. Found by evaluating against a +/// real child run state; a synthetic one whose slots were all objects passed. fn child_answer(id: &str) -> String { // Two different shapes in one expression, which is the whole hazard here. // `.nodes.{id}.item` is the *scope* projection — the child's final run @@ -367,7 +377,9 @@ fn child_answer(id: &str) -> String { "([((.nodes.{id}.item.nodes // {{}}) | to_entries[]) \ | . as $step \ | ((.value.items // []) | .[-1] | .json) as $out \ - | ($out.text? // $out.stdout? // $out.json.stdout? // empty) as $said \ + | (($out | if type == \"object\" then \ + (.text // .stdout // (.json | if type == \"object\" then .stdout else null end)) \ + else null end) // empty) as $said \ | \"## \" + $step.key + \"\\n\" + ($said | tostring)] \ | join(\"\\n\\n\") \ | if . == \"\" then null else . end)" diff --git a/crates/adaptive/src/intake/recipe_tests.rs b/crates/adaptive/src/intake/recipe_tests.rs index 4833c82..710ff7d 100644 --- a/crates/adaptive/src/intake/recipe_tests.rs +++ b/crates/adaptive/src/intake/recipe_tests.rs @@ -442,14 +442,20 @@ fn child_run_state() -> serde_json::Value { json!({ "run": { "trigger": [], "inputs": { "repo": "acme/thing" } }, "nodes": { + // The trigger slot, verbatim from a real run: its payload is the + // seeded item ARRAY, not an object. Every child has one, and it is + // what made the first spelling of the projection fail — a + // fixture whose slots were all objects passed while the real + // thing resolved the whole prompt to null. + "start": { "items": [{ "json": [{ "json": {} }] }] }, "fetch_pr": { "items": [{ "json": { "json": { "exit_code": 0, "stdout": "3 files changed" }, "text": null, "raw": {} } }] }, "verdict": { "items": [{ "json": { - "json": "Requesting changes.", + "json": { "text": "Requesting changes.", "worker": "local" }, "text": "Requesting changes.", - "raw": "Requesting changes." + "raw": { "text": "Requesting changes." } } }] } } }) From 14df5c634b66ab8e64bae9710aea1fa53a89c913 Mon Sep 17 00:00:00 2001 From: sanil-23 Date: Wed, 19 Aug 2026 01:55:25 +0530 Subject: [PATCH 3/6] feat(adaptive): score the workflows a plan calls, not just the one it picks MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit A workflow only ever used as a component stayed Unproven forever. The chooser weighs `run N×, satisfied M×`, the promotion gate reads the same counters, and neither ever moved for a callee — so composition was a place a procedure went to stop earning a reputation, and the first `use` step ever written made its callees permanently less trusted than they were before anyone called them. The rule is the one a selection is already held to — it ran, and the attempt was judged satisfied — plus one condition a selection does not need. A selected workflow IS the attempt, so the attempt's verdict is its verdict. A called one is a part, so its own step must also have succeeded: a child that errored inside a plan that recovered around it has been exercised, not vindicated, and reads `applied` without `helped`. A call the run never reached earns nothing at all, not even `applied` — it is not evidence of anything. Read off the graph rather than reported by the runner, so a host implementing `Runner` does not have to know this scoring exists to participate in it. An id written as an `=`-expression is skipped: it names a workflow only once the run resolves it, and scoring the literal text would move counters on a workflow that does not exist. Weaker evidence than a selection's, and worth saying so: nothing here judges the child's OUTPUT, so a child that ran cleanly and contributed nothing to an episode its siblings satisfied is credited anyway. Establishing more would cost a judge call per child, which is the expense the loop's economics exist to avoid. One level only — a grandchild's calls live inside the child's run state, not in this attempt's steps. --- crates/adaptive/src/closing/mod.rs | 61 ++++++++++ crates/adaptive/src/driver.rs | 1 + crates/adaptive/tests/closing.rs | 189 +++++++++++++++++++++++++++++ 3 files changed, 251 insertions(+) diff --git a/crates/adaptive/src/closing/mod.rs b/crates/adaptive/src/closing/mod.rs index 6e22654..c1e8ddc 100644 --- a/crates/adaptive/src/closing/mod.rs +++ b/crates/adaptive/src/closing/mod.rs @@ -21,9 +21,45 @@ pub use keep::{Kept, keep}; pub use repair::{Variant, graph_is_suspect, repair}; use crate::contracts::{Approach, Budget, Goal, Verdict}; +use crate::execute::StepRecord; use crate::intake::Result; use crate::ledger::{Episode, EpisodeStatus, Ledger, LedgerRow}; use tinyflows::caps::Capabilities; +use tinyflows::model::{NodeKind, WorkflowGraph}; + +/// The workflows this graph called, and whether each one's step succeeded. +/// +/// Read off the graph rather than reported by the runner: which nodes are +/// calls is a property of the plan, so a host implementing [`Runner`] does not +/// have to know this scoring exists to participate in it. +/// +/// A node with no step record never ran — the graph stopped short of it — and +/// is credited with nothing at all, not even `applied`. An id written as an +/// `=`-expression is skipped: it names a workflow only once the run resolves +/// it, and scoring the literal text would move counters on a workflow that +/// does not exist. +/// +/// [`Runner`]: crate::execute::Runner +fn called_workflows(graph: &WorkflowGraph, steps: &[StepRecord]) -> Vec<(String, bool)> { + graph + .nodes + .iter() + .filter(|node| node.kind == NodeKind::SubWorkflow) + .filter_map(|node| { + let called = node + .config + .get("workflow_id") + .and_then(serde_json::Value::as_str) + .map(str::trim) + .filter(|id| !id.is_empty() && !id.starts_with('='))?; + let step = steps.iter().find(|step| step.node_id == node.id)?; + Some(( + called.to_string(), + step.status == crate::execute::StepOutcome::Success, + )) + }) + .collect() +} /// What the loop should do next. #[derive(Debug, Clone, PartialEq, Eq)] @@ -78,6 +114,7 @@ pub async fn close( episode: &str, attempt: u32, approach: &Approach, + graph: &WorkflowGraph, ran: &crate::execute::Ran, budget: &Budget, ledger: &dyn Ledger, @@ -139,6 +176,30 @@ pub async fn close( ledger.score_workflow(&id, verdict.satisfied).await?; } + // And the workflows this attempt *called*. Without this a workflow only + // ever used as a component stays Unproven forever: the chooser distrusts + // it, the promotion gate cannot see it, and composition becomes a place + // procedures go to stop earning a reputation. + // + // Same standard a selection is held to — it ran, and the attempt was + // judged satisfied — with one addition a selection does not need. A + // selected workflow IS the attempt, so the attempt's verdict is its + // verdict. A called one is a part, so its own step must also have + // succeeded: a child that errored inside a plan that recovered around it + // has been exercised, not vindicated, and reads `applied` without + // `helped`. + // + // Weaker evidence than a selection's, and worth knowing it: nothing here + // judges the child's *output*, so a child that ran cleanly and + // contributed nothing to an episode satisfied by its siblings is credited + // anyway. Establishing more would cost a judge call per child, which is + // the thing the loop's economics are built to avoid. + for (called, worked) in called_workflows(graph, &ran.steps) { + ledger + .score_workflow(&called, worked && verdict.satisfied) + .await?; + } + let stalled = if verdict.satisfied || verdict.advanced { 0 } else { diff --git a/crates/adaptive/src/driver.rs b/crates/adaptive/src/driver.rs index 45cd3f6..a1df3b7 100644 --- a/crates/adaptive/src/driver.rs +++ b/crates/adaptive/src/driver.rs @@ -152,6 +152,7 @@ impl Loop<'_> { episode, attempt, &planned.approach, + &planned.graph, &ran, &self.budget, self.ledger, diff --git a/crates/adaptive/tests/closing.rs b/crates/adaptive/tests/closing.rs index 33d8185..8c5152d 100644 --- a/crates/adaptive/tests/closing.rs +++ b/crates/adaptive/tests/closing.rs @@ -15,6 +15,7 @@ use tinyflows::caps::{Capabilities, LlmProvider}; use tinyflows::diagnostics::{Diagnosis, NeverRan}; use tinyflows::engine::RunOutcome; use tinyflows::error::Result as EngineResult; +use tinyflows::model::WorkflowGraph; use tinyflows_adaptive::closing::{Next, close, consolidate}; use tinyflows_adaptive::contracts::{Approach, Blocker, Budget, Goal}; use tinyflows_adaptive::execute::Ran; @@ -124,6 +125,7 @@ async fn a_failed_attempt_is_still_recorded_and_still_scored() { "ep-1", 1, &selected("weekly"), + &WorkflowGraph::default(), &ran(&outcome, &diagnosis, "wrote report.md"), &Budget::default(), &ledger, @@ -165,6 +167,7 @@ async fn a_satisfied_attempt_moves_both_halves_of_the_score() { "ep-2", 1, &selected("weekly"), + &WorkflowGraph::default(), &ran(&outcome, &diagnosis, "wrote report.md"), &Budget::default(), &ledger, @@ -202,6 +205,7 @@ async fn a_run_where_nothing_happened_never_reaches_the_model() { "ep-3", 1, &selected("weekly"), + &WorkflowGraph::default(), &ran(&outcome, &diagnosis, ""), &Budget::default(), &ledger, @@ -241,6 +245,7 @@ async fn a_parked_approval_is_not_a_failure() { "ep-4", 1, &selected("blog"), + &WorkflowGraph::default(), &ran(&outcome, &diagnosis, ""), &Budget::default(), &ledger, @@ -282,6 +287,7 @@ async fn two_flat_attempts_in_a_row_stand_down_on_the_stall_rule() { why: format!("attempt {attempt}"), fingerprint: "0000000".into(), }, + &WorkflowGraph::default(), &ran(&outcome, &diagnosis, ""), &budget, &ledger, @@ -466,3 +472,186 @@ async fn an_episode_with_no_attempts_asks_nothing() { assert!(kept.is_empty()); assert_eq!(llm.call_count(), 0); } + +/// A plan that calls two saved workflows, with the second one's step failing. +fn composed() -> WorkflowGraph { + use tinyflows::model::{Edge, Node, NodeKind}; + let call = |id: &str, workflow: &str| Node { + id: id.to_string(), + kind: NodeKind::SubWorkflow, + type_version: 1, + name: id.to_string(), + config: json!({ "workflow_id": workflow }), + ports: Vec::new(), + position: None, + }; + WorkflowGraph { + schema_version: 1, + id: None, + name: "composed".into(), + inputs: Vec::new(), + agents: Vec::new(), + 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, + }, + call("write_haiku", "haiku-writer"), + call("write_limerick", "limerick-writer"), + call("never_reached", "epic-writer"), + ], + edges: vec![ + Edge { + from_node: "start".into(), + from_port: "main".into(), + to_node: "write_haiku".into(), + to_port: "main".into(), + }, + Edge { + from_node: "write_haiku".into(), + from_port: "main".into(), + to_node: "write_limerick".into(), + to_port: "main".into(), + }, + Edge { + from_node: "write_limerick".into(), + from_port: "main".into(), + to_node: "never_reached".into(), + to_port: "main".into(), + }, + ], + } +} + +fn step(node: &str, ok: bool) -> tinyflows_adaptive::execute::StepRecord { + use tinyflows_adaptive::execute::{StepOutcome, StepRecord}; + StepRecord { + node_id: node.to_string(), + status: if ok { + StepOutcome::Success + } else { + StepOutcome::Error + }, + output: Value::Null, + duration_ms: 1, + null_bindings: Vec::new(), + } +} + +#[tokio::test] +async fn a_workflow_called_by_a_plan_earns_the_same_record_a_chosen_one_does() { + // Without this a workflow only ever used as a component stays Unproven + // forever: the chooser distrusts it and the promotion gate cannot see it, + // so composition becomes a place procedures go to stop earning a + // reputation. + let llm = Scripted::new(vec![json!({ + "satisfied": true, "blocker": "none", "gap": "", "advanced": true + })]); + let ledger = MemoryLedger::new(); + let diagnosis = Diagnosis::default(); + let outcome = RunOutcome { + output: json!({}), + pending_approvals: Vec::new(), + cancelled: false, + }; + let mut finished = ran(&outcome, &diagnosis, "wrote the document"); + finished.steps = vec![ + step("write_haiku", true), + // Errored inside a plan that recovered around it. + step("write_limerick", false), + ]; + + close( + &Goal::new("a haiku and a limerick"), + "ep-compose", + 1, + &Approach::Authored { + why: "compose the two writers".into(), + fingerprint: "abc1234".into(), + }, + &composed(), + &finished, + &Budget::default(), + &ledger, + &caps_with(llm), + None, + "2026-01-01T00:00:00Z", + ) + .await + .expect("closes"); + + let haiku = ledger.workflow_score("haiku-writer").await.expect("scored"); + assert_eq!( + (haiku.applied, haiku.helped), + (1, 1), + "it ran and the attempt was satisfied — the standard a selection is held to" + ); + + let limerick = ledger + .workflow_score("limerick-writer") + .await + .expect("scored"); + assert_eq!( + (limerick.applied, limerick.helped), + (1, 0), + "a child that errored inside a satisfied plan was exercised, not vindicated" + ); + + let never = ledger.workflow_score("epic-writer").await.expect("scored"); + assert_eq!( + (never.applied, never.helped), + (0, 0), + "a call the run never reached is not evidence of anything" + ); +} + +#[tokio::test] +async fn a_called_workflow_earns_nothing_from_an_attempt_that_fell_short() { + // The counters must stay readable as evidence: an unsatisfied episode + // gives a component `applied` and no more, exactly as it would a chosen + // workflow that failed. + let llm = Scripted::new(vec![json!({ + "satisfied": false, "blocker": "goal_not_met", + "gap": "the document is missing the limerick", "advanced": false + })]); + let ledger = MemoryLedger::new(); + let diagnosis = Diagnosis::default(); + let outcome = RunOutcome { + output: json!({}), + pending_approvals: Vec::new(), + cancelled: false, + }; + let mut finished = ran(&outcome, &diagnosis, "wrote half a document"); + finished.steps = vec![step("write_haiku", true)]; + + close( + &Goal::new("a haiku and a limerick"), + "ep-short", + 1, + &Approach::Authored { + why: "compose the two writers".into(), + fingerprint: "abc1234".into(), + }, + &composed(), + &finished, + &Budget::default(), + &ledger, + &caps_with(llm), + None, + "2026-01-01T00:00:00Z", + ) + .await + .expect("closes"); + + let haiku = ledger.workflow_score("haiku-writer").await.expect("scored"); + assert_eq!( + (haiku.applied, haiku.helped), + (1, 0), + "the child ran cleanly, but nothing it was part of was satisfied" + ); +} From 42343a94c5c517c4bcd42619f0bb1a653061b035 Mon Sep 17 00:00:00 2001 From: sanil-23 Date: Wed, 19 Aug 2026 11:12:46 +0530 Subject: [PATCH 4/6] fix(adaptive): count every call, and quote the paths a plan generates MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Four fixes from review, each one a case where the surface reported something other than what happened. **Every activation of a call is scored, not just the first.** A node inside a loop produces one `StepRecord` per iteration, and reading only the first credited a workflow once for work it did three times — and let an early success hide a later error, so a child that failed a pass read as clean. The counters are the only evidence the chooser and the promotion gate have, so the walk is now over the records with the graph as a lookup, not over the nodes taking one record each. **Interpolated ids are quoted in generated jq paths.** `sanitize_id` keeps `[a-z0-9_]`, which is wider than the identifiers jq's dot syntax accepts: it permits a leading digit, and nothing upstream rejects a step id like `2024_report`. `.nodes.2024_report` does not compile, so the whole prompt expression resolved to nothing — a plan refused by the evaluator over how its author happened to name a step. Bracket access takes any key. The test evaluates rather than string-matches, and fails against the old spelling. **A required input present but empty is refused.** `contains_key` accepted `"repo": null` and `"repo": ""`, which `forward` then passed through unchanged, so the child failed its own declaration check mid-run — the exact failure the intake refusal exists to prevent. `gated` in `author.rs` already read unfilled that way; the two checks disagreeing is what let the value through. **One store listing, not two.** `decide` called `catalogue` and `callables` and each listed the store, so composition cost a second directory scan and a second parse of every record on every attempt — while the doc comment claimed it cost nothing. Listed once, both views derived from it, and the `(name, required)` mapping they shared lives in one place. Also: the declared-input listing both prompts render is one helper rather than two copies of a convention the model is being asked to obey, and the sub-workflow lowering gets the contract-drift test the shell lowering already had, asserting the engine still reads `workflow_id` and `inputs`. Co-Authored-By: Claude Opus 5 (1M context) Co-authored-by: Medulla --- crates/adaptive/src/closing/mod.rs | 20 +++- crates/adaptive/src/intake/mod.rs | 79 +++++++------- crates/adaptive/src/intake/recipe.rs | 77 +++++++++---- crates/adaptive/src/intake/recipe_tests.rs | 121 ++++++++++++++++++++- crates/adaptive/src/intake/select.rs | 13 +-- crates/adaptive/tests/closing.rs | 54 +++++++++ 6 files changed, 287 insertions(+), 77 deletions(-) diff --git a/crates/adaptive/src/closing/mod.rs b/crates/adaptive/src/closing/mod.rs index c1e8ddc..f2a6aed 100644 --- a/crates/adaptive/src/closing/mod.rs +++ b/crates/adaptive/src/closing/mod.rs @@ -24,6 +24,7 @@ use crate::contracts::{Approach, Budget, Goal, Verdict}; use crate::execute::StepRecord; use crate::intake::Result; use crate::ledger::{Episode, EpisodeStatus, Ledger, LedgerRow}; +use std::collections::HashMap; use tinyflows::caps::Capabilities; use tinyflows::model::{NodeKind, WorkflowGraph}; @@ -39,9 +40,15 @@ use tinyflows::model::{NodeKind, WorkflowGraph}; /// it, and scoring the literal text would move counters on a workflow that /// does not exist. /// +/// **One entry per activation, not per node.** A node inside a loop produces +/// one [`StepRecord`] per iteration, so the walk is over the *records* with the +/// graph as a lookup, not over the nodes taking the first record each. Reading +/// only the first would drop every later call and — worse — let an early +/// success hide a later error, crediting a workflow for a run that failed. +/// /// [`Runner`]: crate::execute::Runner fn called_workflows(graph: &WorkflowGraph, steps: &[StepRecord]) -> Vec<(String, bool)> { - graph + let calls: HashMap<&str, &str> = graph .nodes .iter() .filter(|node| node.kind == NodeKind::SubWorkflow) @@ -52,9 +59,16 @@ fn called_workflows(graph: &WorkflowGraph, steps: &[StepRecord]) -> Vec<(String, .and_then(serde_json::Value::as_str) .map(str::trim) .filter(|id| !id.is_empty() && !id.starts_with('='))?; - let step = steps.iter().find(|step| step.node_id == node.id)?; + Some((node.id.as_str(), called)) + }) + .collect(); + + steps + .iter() + .filter_map(|step| { + let called = calls.get(step.node_id.as_str())?; Some(( - called.to_string(), + (*called).to_string(), step.status == crate::execute::StepOutcome::Success, )) }) diff --git a/crates/adaptive/src/intake/mod.rs b/crates/adaptive/src/intake/mod.rs index 8842e77..1cf2d37 100644 --- a/crates/adaptive/src/intake/mod.rs +++ b/crates/adaptive/src/intake/mod.rs @@ -25,7 +25,7 @@ pub use select::{Candidate, bind, select}; use serde_json::{Map, Value}; use tinyflows::caps::Capabilities; use tinyflows::model::WorkflowGraph; -use tinyflows::store::WorkflowStore; +use tinyflows::store::{WorkflowStore, WorkflowSummary}; use crate::contracts::{Approach, Goal}; use crate::host::HostFacts; @@ -113,13 +113,21 @@ pub async fn decide( // attempt, against whatever database the host brought. let rows = ledger.rows(episode).await?; let tried = crate::ledger::signatures(&rows); - let candidates = catalogue(store, ledger, &tried).await?; - // Two views of the same shelf, and the difference is the point. The - // chooser sees what it has not already tried, because repeating a - // selection cannot teach the episode anything. The author sees the WHOLE - // shelf, because a workflow that fell short as the entire answer is - // exactly the one worth calling as one step of a bigger plan. - let callables = callables(store)?; + // One listing, two views of the same shelf, and the difference is the + // point. The chooser sees what it has not already tried, because repeating + // a selection cannot teach the episode anything. The author sees the WHOLE + // shelf, because a workflow that fell short as the entire answer is exactly + // the one worth calling as one step of a bigger plan. + // + // Listed once for the same reason `ledger.rows` is read once above: against + // a file-backed store a second `list` is a second directory scan and a + // second parse of every record, paid on every attempt for a result already + // in hand. + let listed = store + .list() + .map_err(|e| IntakeError::Store(e.to_string()))?; + let candidates = catalogue(&listed, ledger, &tried).await?; + let callables = callables(&listed); // Both planners see the same past, in the same words. The exclusion list // stops a *selection* being repeated, but nothing structural stops the @@ -219,38 +227,39 @@ pub async fn decide( /// the newest — see [`crate::promotion`]. /// Every enabled workflow, as something a plan may call. /// -/// Unfiltered on purpose — see the note at the call site. Built from the same -/// listing the chooser's catalogue reads, so composition costs no extra store +/// Unfiltered on purpose — see the note at the call site. Takes the listing the +/// chooser's catalogue already read, so composition costs no extra store /// traffic: `WorkflowSummary` already carries the declared inputs, precisely /// so a caller need not fetch a whole graph to learn what it takes. -fn callables(store: &dyn WorkflowStore) -> Result> { - Ok(store - .list() - .map_err(|e| IntakeError::Store(e.to_string()))? - .into_iter() +fn callables(listed: &[WorkflowSummary]) -> Vec { + listed + .iter() .filter(|summary| summary.enabled) .map(|summary| Callable { - id: summary.id, - name: summary.name, - description: summary.description, - inputs: summary - .inputs - .into_iter() - .map(|input| (input.name, input.required)) - .collect(), + id: summary.id.clone(), + name: summary.name.clone(), + description: summary.description.clone(), + inputs: declared_inputs(summary), }) - .collect()) + .collect() +} + +/// A summary's declared inputs as the `(name, required)` pairs both prompts +/// render. One mapping, because two copies of it is how the chooser and the +/// author come to disagree about which inputs a workflow demands. +fn declared_inputs(summary: &WorkflowSummary) -> Vec<(String, bool)> { + summary + .inputs + .iter() + .map(|input| (input.name.clone(), input.required)) + .collect() } async fn catalogue( - store: &dyn WorkflowStore, + listed: &[WorkflowSummary], ledger: &dyn Ledger, tried: &[String], ) -> Result> { - let listed = store - .list() - .map_err(|e| IntakeError::Store(e.to_string()))?; - let mut out = Vec::new(); for summary in listed { if !summary.enabled { @@ -262,17 +271,13 @@ async fn catalogue( } let score = ledger.workflow_score(&summary.id).await?; out.push(Candidate { - id: summary.id, - name: summary.name, - description: summary.description, + id: summary.id.clone(), + name: summary.name.clone(), + description: summary.description.clone(), node_count: summary.node_count, applied: score.applied, helped: score.helped, - inputs: summary - .inputs - .into_iter() - .map(|input| (input.name, input.required)) - .collect(), + inputs: declared_inputs(summary), }); } collapse_families(out, ledger).await diff --git a/crates/adaptive/src/intake/recipe.rs b/crates/adaptive/src/intake/recipe.rs index 5b9e29a..6d0c6ad 100644 --- a/crates/adaptive/src/intake/recipe.rs +++ b/crates/adaptive/src/intake/recipe.rs @@ -114,18 +114,7 @@ impl Callable { let inputs = if self.inputs.is_empty() { "takes no inputs".to_string() } else { - let listed: Vec = self - .inputs - .iter() - .map(|(name, required)| { - if *required { - name.clone() - } else { - format!("{name} (optional)") - } - }) - .collect(); - format!("with: {}", listed.join(", ")) + format!("with: {}", render_inputs(&self.inputs)) }; format!( "- id: {}\n name: {name}\n {inputs}\n {description}", @@ -134,6 +123,32 @@ impl Callable { } } +/// Declared inputs as one comma-separated listing: `repo, depth (optional)`. +/// +/// A required input is named bare and an optional one is marked, because the +/// only thing a planner does with this line is decide what it must supply. Both +/// prompts that carry it — the chooser's candidate listing and the author's +/// callable listing — render it here rather than each spelling the rule out, +/// since two copies of a convention a model is being asked to obey drift the +/// first time the wording changes and then teach two different things. +/// +/// The prefix and the empty-list wording stay with each caller: the chooser +/// says nothing at all when there are no inputs, the author says "takes no +/// inputs", and that difference is deliberate. +pub(super) fn render_inputs(inputs: &[(String, bool)]) -> String { + inputs + .iter() + .map(|(name, required)| { + if *required { + name.clone() + } else { + format!("{name} (optional)") + } + }) + .collect::>() + .join(", ") +} + /// The prompt section listing what a `use` step may name. /// /// Empty when nothing is callable, so a cold store's author is not told about @@ -290,8 +305,9 @@ fn ask_expression( 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")) + " + {} + ((.run.inputs{} // \"(not provided)\") | tostring)", + jq_quote(&format!("\n\n# Input `{name}`\n")), + jq_field(name) )); } for read in reads { @@ -343,10 +359,11 @@ fn kind_of<'a>(id: &str, steps: &'a [Step]) -> Option<&'a Action> { /// ordinary child whose one agent step writes the deliverable, it is exactly /// that deliverable. fn output_of(id: &str, action: Option<&Action>) -> String { + let slot = jq_field(id); match action { - Some(Action::Run(_)) => format!(".nodes.{id}.item.json.stdout"), + Some(Action::Run(_)) => format!(".nodes{slot}.item.json.stdout"), Some(Action::Use { .. }) => child_answer(id), - _ => format!(".nodes.{id}.item.text"), + _ => format!(".nodes{slot}.item.text"), } } @@ -373,8 +390,9 @@ fn child_answer(id: &str) -> String { // serialized items (`{"json": …}`) rather than the bare payloads the scope // exposes. So the outer hop drops `items`/`json` and the inner one needs // both. + let slot = jq_field(id); format!( - "([((.nodes.{id}.item.nodes // {{}}) | to_entries[]) \ + "([((.nodes{slot}.item.nodes // {{}}) | to_entries[]) \ | . as $step \ | ((.value.items // []) | .[-1] | .json) as $out \ | (($out | if type == \"object\" then \ @@ -386,6 +404,19 @@ fn child_answer(id: &str) -> String { ) } +/// One object key as a jq path step: `["fetch_pr"]`, never `.fetch_pr`. +/// +/// `sanitize_id` keeps `[a-z0-9_]`, which is *not* the same set as the +/// identifiers jq's dot syntax accepts: it permits a leading digit, and nothing +/// upstream rejects a step id such as `2024_report`. `.nodes.2024_report` does +/// not compile, so the whole prompt expression fails at run time — a plan +/// refused by the evaluator for the way its author spelled an id. Bracket +/// access takes any key, so this is the only spelling used for an interpolated +/// name. +fn jq_field(name: &str) -> String { + format!("[{}]", jq_quote(name)) +} + /// 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 { @@ -559,7 +590,15 @@ fn call( }; for (name, _) in callable.inputs.iter().filter(|(_, required)| *required) { - if !given.contains_key(name) { + // Present-but-empty is not supplied. `"repo": null` and `"repo": ""` + // pass a `contains_key` check and are then forwarded unchanged, so the + // child fails its own declaration check mid-run — the exact failure + // this refusal exists to move to intake. `gated` in `author.rs` already + // reads unfilled the same way; the two must not disagree. + let filled = given + .get(name) + .is_some_and(|value| !value.is_null() && value.as_str() != Some("")); + if !filled { return Err(format!( "step `{id}`: `{workflow_id}` requires the input `{name}` and `with` does \ not supply it" @@ -615,7 +654,7 @@ fn forward( "`@input.{name}` names an input you did not declare — add it to `declared`" )); } - return Ok(Value::String(format!("=.run.inputs.{name}"))); + return Ok(Value::String(format!("=.run.inputs{}", jq_field(&name)))); } if let Some(step) = reference.strip_prefix("step.") { let step = sanitize_id(step); diff --git a/crates/adaptive/src/intake/recipe_tests.rs b/crates/adaptive/src/intake/recipe_tests.rs index 710ff7d..4271596 100644 --- a/crates/adaptive/src/intake/recipe_tests.rs +++ b/crates/adaptive/src/intake/recipe_tests.rs @@ -50,9 +50,12 @@ fn the_generated_prompt_is_one_expression_with_the_right_envelope_paths() { 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}"); + 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}"); + assert!(prompt.contains(".run.inputs[\"repo\"]"), "{prompt}"); // Absent values surface as markers, not silent nothing. assert!(prompt.contains("(no output)"), "{prompt}"); } @@ -68,7 +71,7 @@ fn an_agent_upstream_is_read_through_text_not_stdout() { }); let (graph, _, _) = lower(&recipe, &[]).expect("lowers"); let prompt = graph.nodes[2].config["prompt"].as_str().expect("prompt"); - assert!(prompt.contains(".nodes.draft.item.text"), "{prompt}"); + assert!(prompt.contains(".nodes[\"draft\"].item.text"), "{prompt}"); } #[test] @@ -176,7 +179,41 @@ fn ids_are_sanitized_into_engine_and_jq_safe_names() { 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}"); + assert!(prompt.contains(".nodes[\"fetch_issues\"].item"), "{prompt}"); +} + +#[test] +fn a_step_id_starting_with_a_digit_still_compiles_as_jq() { + // `sanitize_id` keeps `[a-z0-9_]`, which is a wider set than the + // identifiers jq's dot syntax accepts, and nothing upstream rejects an id + // beginning with a digit. Spelled `.nodes.2024_report` the whole prompt + // fails to compile — a plan refused by the evaluator over how its author + // happened to name a step. Evaluated, not string-matched: asserting the + // spelling is what let the previous path bug ship. + let recipe = json!({ + "why": "read a numerically named step", + "declared": [ + { "name": "repo", "description": "owner/name", "required": true } + ], + "inputs": { "repo": "acme/thing" }, + "steps": [ + { "id": "2024 report", "run": "cat report" }, + { "id": "summary", "ask": "Summarise it.", "reads": ["2024 report"] } + ] + }); + let (graph, _, _) = lower(&recipe, &[]).expect("lowers"); + assert_eq!(graph.nodes[1].id, "2024_report"); + let prompt = graph.nodes[2].config["prompt"].as_str().expect("prompt"); + let scope = json!({ + "run": { "inputs": { "repo": "acme/thing" } }, + "nodes": { "2024_report": { "item": { "json": { "stdout": "12 findings" } } } } + }); + let rendered = tinyflows::expr::resolve(&json!(prompt), &scope); + let rendered = rendered.as_str().unwrap_or_default(); + assert!( + rendered.contains("12 findings"), + "a digit-leading step id must produce a compilable path: {prompt} -> {rendered}" + ); } // --------------------------------------------------------------------------- @@ -231,7 +268,56 @@ fn a_use_step_lowers_to_a_sub_workflow_node_that_references_the_callee() { "an inlined child would fork the callee at authoring time" ); // `@input.repo` became the expression that reads the parent's run input. - assert_eq!(node.config["inputs"]["repo"], json!("=.run.inputs.repo")); + assert_eq!( + node.config["inputs"]["repo"], + json!("=.run.inputs[\"repo\"]") + ); +} + +#[test] +fn the_lowered_sub_workflow_config_satisfies_the_engines_own_contract() { + // The same drift guard the shell lowering has, for the same reason: the + // `use` step's whole value is that the engine already knows how to run a + // child, and it knows it by reading `workflow_id` and `inputs`. A rename on + // either side would surface as a capability error mid-run, attributed to + // the work rather than to the plan. + let (graph, _, _) = lower(&compose_recipe(), &[audit()]).expect("lowers"); + let config = &graph + .nodes + .iter() + .find(|node| node.id == "audit") + .expect("the use step became a node") + .config; + let contract = tinyflows::catalog::all_contracts() + .iter() + .find(|contract| contract.kind == "sub_workflow") + .expect("the engine has a sub_workflow contract") + .clone(); + let fields: Vec<&str> = contract + .config_fields + .iter() + .map(|field| field.name.as_str()) + .collect(); + for key in ["workflow_id", "inputs"] { + assert!( + fields.contains(&key), + "the engine's sub_workflow contract no longer declares `{key}`: {fields:?}" + ); + assert!( + config.get(key).is_some(), + "a lowered use step must fill config.{key}: {config}" + ); + } + // Required fields are the engine's own list; filling one is not enough if + // it grows another. + for field in contract.config_fields.iter().filter(|field| field.required) { + assert!( + config.get(&field.name).is_some(), + "the lowering fills none of the engine's required sub_workflow field \ + `{}`: {config}", + field.name + ); + } } #[test] @@ -260,10 +346,33 @@ fn a_step_reference_in_with_reads_the_earlier_step_the_way_its_kind_produces() { // the model having to. assert_eq!( node.config["inputs"]["patch"], - json!("=.nodes.diff.item.json.stdout") + json!("=.nodes[\"diff\"].item.json.stdout") ); } +#[test] +fn a_required_input_present_but_empty_is_refused_the_way_an_absent_one_is() { + // `contains_key` accepts `null` and `""`, which `forward` then passes + // through unchanged — so the child fails its OWN declaration check + // mid-run, which is the failure this refusal exists to move to intake. + // `gated` in `author.rs` already reads unfilled this way; the two checks + // disagreeing is what let the value through. + for empty in [json!(null), json!("")] { + let recipe = json!({ + "why": "audit the PR", + "declared": [], + "steps": [ + { "id": "audit", "use": "pr-audit-review", "with": { "repo": empty } } + ] + }); + let err = lower(&recipe, &[audit()]).expect_err("refused").to_string(); + assert!( + err.contains("requires the input `repo`"), + "an empty value is not a supplied value ({empty}): {err}" + ); + } +} + #[test] fn a_use_step_naming_a_workflow_nobody_offered_is_refused_at_intake() { // Not deferred to the resolver: a hallucinated id would surface as a diff --git a/crates/adaptive/src/intake/select.rs b/crates/adaptive/src/intake/select.rs index 39c07d4..011f46a 100644 --- a/crates/adaptive/src/intake/select.rs +++ b/crates/adaptive/src/intake/select.rs @@ -64,18 +64,7 @@ impl Candidate { let inputs = if self.inputs.is_empty() { String::new() } else { - let listed: Vec = self - .inputs - .iter() - .map(|(name, required)| { - if *required { - name.clone() - } else { - format!("{name} (optional)") - } - }) - .collect(); - format!("\n inputs: {}", listed.join(", ")) + format!("\n inputs: {}", super::recipe::render_inputs(&self.inputs)) }; format!( "- id: {}\n name: {name}\n steps: {}, {record}{inputs}\n {description}", diff --git a/crates/adaptive/tests/closing.rs b/crates/adaptive/tests/closing.rs index 8c5152d..4d7ce4e 100644 --- a/crates/adaptive/tests/closing.rs +++ b/crates/adaptive/tests/closing.rs @@ -655,3 +655,57 @@ async fn a_called_workflow_earns_nothing_from_an_attempt_that_fell_short() { "the child ran cleanly, but nothing it was part of was satisfied" ); } + +#[tokio::test] +async fn every_activation_of_a_looped_call_is_scored_not_just_the_first() { + // A node inside a loop produces one `StepRecord` per iteration. Reading + // only the first record credits the workflow once for work it did three + // times — and, worse, lets an early success hide a later error, so a child + // that failed a pass reads as clean. The counters are the only evidence the + // chooser and the promotion gate have; they have to count what happened. + let llm = Scripted::new(vec![json!({ + "satisfied": true, "blocker": "none", "gap": "", "advanced": true + })]); + let ledger = MemoryLedger::new(); + let diagnosis = Diagnosis::default(); + let outcome = RunOutcome { + output: json!({}), + pending_approvals: Vec::new(), + cancelled: false, + }; + let mut finished = ran(&outcome, &diagnosis, "wrote three haiku"); + // One node, three passes, mixed outcomes — the first one succeeding is + // exactly the arrangement that made the old reading look correct. + finished.steps = vec![ + step("write_haiku", true), + step("write_haiku", false), + step("write_haiku", true), + ]; + + close( + &Goal::new("three haiku"), + "ep-loop", + 1, + &Approach::Authored { + why: "call the writer once per subject".into(), + fingerprint: "abc1234".into(), + }, + &composed(), + &finished, + &Budget::default(), + &ledger, + &caps_with(llm), + None, + "2026-01-01T00:00:00Z", + ) + .await + .expect("closes"); + + let haiku = ledger.workflow_score("haiku-writer").await.expect("scored"); + assert_eq!( + (haiku.applied, haiku.helped), + (3, 2), + "three activations, and the one that errored is not vindicated by the \ + two that did not" + ); +} From 2ee4fc7606dc4168e5f1a18aae51b47bc82612c8 Mon Sep 17 00:00:00 2001 From: sanil-23 Date: Wed, 19 Aug 2026 11:25:19 +0530 Subject: [PATCH 5/6] feat(adaptive): continue a repaired run from where it broke, when that is sound MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit A failed run leaves its prefix committed. The loop threw it away: every retry started at the trigger, so a graph whose third step broke paid for the first two again — and if either of them posted a comment or opened a pull request, paid for that too, twice. The engine can now continue such a run. What decides whether the loop *should* is the new gate, and it is mechanical on purpose — no model is asked whether its own edit was safe to skip work over. **The rule: continue only when no edit touches a node that is an ancestor of the failed one — in EITHER graph.** Both graphs, because the two directions fail differently and each is invisible from the other side. An edit that removes an upstream node is an ancestor only in the graph that ran. An edit that ADds one — a fetch step wired in ahead of the node that starved without it — is an ancestor only in the repaired graph, and it is the worse of the two: the new node has no committed output, so continuing would re-enter the failed node with the upstream it was just given still missing. The fix would look like it had not worked, and the next repair would chase the wrong thing. Editing the failed node itself is fine and is the case worth having: a continue re-runs it, so its config is read fresh. So is editing anything downstream, none of which has run. The plumbing is deliberately narrow. `Ran::resume` is what a runner reports — `None` from any host without a checkpointer, which is every host until it opts in. `Attempt::resume` is what the loop hands back, and only ever to an attempt that selected the very workflow the repair produced: the chooser is free to pick something else, and a prefix committed by one graph is not a prefix for another. `ResumePoint` carries the workflow id for exactly that check, so both sides can confirm they mean the same run. It is carried through one attempt rather than stored, because it is worth exactly one: it names a boundary in a checkpointer whose thread the next run writes over, and a stale one would re-enter a graph on some other run's prefix. `Loop::run` threads it; `attempt_continuing` is the seam for a host driving attempts itself, and plain `attempt` passes `None` and behaves exactly as before. A `false` from the gate is not a refusal to retry. It means the retry starts from the trigger — which is what every attempt did before this existed, so the floor is unchanged and continuing is strictly an improvement on top of it. One bug the tests caught while being written, worth recording because the first shape of it looked right: an op that names no node but changes what every node reads (`SetWorkflowInputs`) was represented as an unmatchable sentinel node id. A set containing only that id is disjoint from every real ancestor set, so the op read as touching *nothing* — the exact inversion of what it means. The absence is in the type now. --- crates/adaptive/src/closing/mod.rs | 2 + crates/adaptive/src/closing/resume.rs | 145 ++++++++++++ crates/adaptive/src/closing/resume_tests.rs | 246 ++++++++++++++++++++ crates/adaptive/src/contracts.rs | 33 +++ crates/adaptive/src/driver.rs | 92 ++++++-- crates/adaptive/src/execute/mod.rs | 9 + crates/adaptive/src/execute/wire.rs | 4 + crates/adaptive/src/intake/author.rs | 2 + crates/adaptive/src/intake/mod.rs | 9 + crates/adaptive/src/intake/select.rs | 3 + crates/adaptive/tests/closing.rs | 1 + crates/adaptive/tests/execute.rs | 1 + 12 files changed, 533 insertions(+), 14 deletions(-) create mode 100644 crates/adaptive/src/closing/resume.rs create mode 100644 crates/adaptive/src/closing/resume_tests.rs diff --git a/crates/adaptive/src/closing/mod.rs b/crates/adaptive/src/closing/mod.rs index f2a6aed..8535a58 100644 --- a/crates/adaptive/src/closing/mod.rs +++ b/crates/adaptive/src/closing/mod.rs @@ -14,11 +14,13 @@ mod consolidate; mod judge; mod keep; mod repair; +mod resume; pub use consolidate::consolidate; pub use judge::{Evidence, judge}; pub use keep::{Kept, keep}; pub use repair::{Variant, graph_is_suspect, repair}; +pub use resume::may_continue; use crate::contracts::{Approach, Budget, Goal, Verdict}; use crate::execute::StepRecord; diff --git a/crates/adaptive/src/closing/resume.rs b/crates/adaptive/src/closing/resume.rs new file mode 100644 index 0000000..69deeab --- /dev/null +++ b/crates/adaptive/src/closing/resume.rs @@ -0,0 +1,145 @@ +//! Whether a repaired graph may continue the run its parent broke, or has to +//! start over. +//! +//! A failed run leaves its prefix committed: the engine's failure boundary +//! holds everything that finished before the node that broke, so the next +//! attempt can re-enter at that node instead of redoing the lot. For a graph +//! with effects in it that is not an optimisation — re-running a step that +//! posted a comment posts a second one. +//! +//! But a repair produces a *different graph*, and the committed prefix was +//! produced by the old one. Continuing is only sound when the two agree about +//! everything that already ran. This module is that check, and it is +//! deliberately a mechanical one: no model is asked whether its own edit was +//! safe to skip work over. +//! +//! **The rule.** Continue only when no edit touches a node that is an ancestor +//! of the failed one — in *either* graph. +//! +//! Both graphs, because the two directions fail differently and each is +//! invisible from the other side. An edit that *removes* an upstream node is +//! only an ancestor in the parent. An edit that *adds* one — a fetch step +//! wired in ahead of the node that starved without it — is only an ancestor in +//! the child, and it is the more dangerous of the two: the new node has no +//! committed output, so continuing would re-enter the failed node with the +//! upstream it was just given still missing, and the fix would look like it +//! had not worked. +//! +//! Editing the failed node itself is fine, and is the case worth having: a +//! continue re-runs it, so its config is read fresh. So is editing anything +//! downstream — none of it has run. + +use std::collections::{HashMap, HashSet}; + +use tinyflows::graph_ops::GraphOp; +use tinyflows::model::WorkflowGraph; + +/// Whether the run that failed at `failed_node` may be continued under `child`. +/// +/// `parent` is the graph that ran, `child` the repaired one, and `ops` the +/// edits between them. `false` is the safe answer and the default for anything +/// this cannot reason about — an unknown node id, an empty failed node. +/// +/// A `false` here is not a refusal to retry. It means the retry starts from +/// the trigger, which is what every attempt did before continuing existed. +#[must_use] +pub fn may_continue( + parent: &WorkflowGraph, + child: &WorkflowGraph, + failed_node: &str, + ops: &[GraphOp], +) -> bool { + if failed_node.is_empty() { + return false; + } + // A node the run never reached cannot be where it stopped; something is + // out of step, and guessing is the one thing this must not do. + if !parent.nodes.iter().any(|node| node.id == failed_node) { + return false; + } + let mut upstream = ancestors(parent, failed_node); + upstream.extend(ancestors(child, failed_node)); + let Some(edited) = touched(ops) else { + // A whole-graph edit invalidates every committed output at once. Still + // safe when there is no prefix to invalidate — a node with no + // ancestors has nothing committed ahead of it. + return upstream.is_empty(); + }; + upstream.is_disjoint(&edited) +} + +/// Every node that can reach `target` by following edges forward. +/// +/// The committed prefix is exactly this set's output, so it is exactly the set +/// a continue takes on trust. Walks the reverse edges breadth-first; a cycle +/// terminates because a node already seen is not queued again. +fn ancestors(graph: &WorkflowGraph, target: &str) -> HashSet { + let mut incoming: HashMap<&str, Vec<&str>> = HashMap::new(); + for edge in &graph.edges { + incoming + .entry(edge.to_node.as_str()) + .or_default() + .push(edge.from_node.as_str()); + } + let mut seen: HashSet = HashSet::new(); + let mut queue = vec![target]; + while let Some(node) = queue.pop() { + for from in incoming.get(node).into_iter().flatten() { + if seen.insert((*from).to_string()) { + queue.push(from); + } + } + } + seen +} + +/// Every node id an edit names, or `None` when an edit reaches all of them. +/// +/// Both endpoints of an edge op, because an edge is a statement about two +/// nodes: rewiring what an ancestor feeds changes that ancestor's meaning as +/// much as editing its config does. +/// +/// `None` is for an op that names no node and yet changes what every node +/// reads — today only `SetWorkflowInputs`. Returning it as an unmatchable +/// sentinel id was the first attempt and was silently wrong: a set containing +/// only that id is disjoint from every real ancestor set, so the op read as +/// touching *nothing*. The absence has to be in the type. +fn touched(ops: &[GraphOp]) -> Option> { + let mut names: HashSet = HashSet::new(); + for op in ops { + match op { + GraphOp::AddNode { node } => { + names.insert(node.id.clone()); + } + GraphOp::UpdateNodeConfig { id, .. } + | GraphOp::SetNodeName { id, .. } + | GraphOp::RemoveNode { id } + | GraphOp::SetNodePosition { id, .. } => { + names.insert(id.clone()); + } + GraphOp::RenameNode { id, new_id } => { + names.insert(id.clone()); + names.insert(new_id.clone()); + } + GraphOp::AddEdge { edge } => { + names.insert(edge.from_node.clone()); + names.insert(edge.to_node.clone()); + } + GraphOp::RemoveEdge { + from_node, to_node, .. + } => { + names.insert(from_node.clone()); + names.insert(to_node.clone()); + } + // Declared inputs are read by `=`-expressions anywhere in the + // graph, including in the prefix that already ran, so a change to + // them invalidates every committed output at once. + GraphOp::SetWorkflowInputs { .. } => return None, + } + } + Some(names) +} + +#[cfg(test)] +#[path = "resume_tests.rs"] +mod tests; diff --git a/crates/adaptive/src/closing/resume_tests.rs b/crates/adaptive/src/closing/resume_tests.rs new file mode 100644 index 0000000..cd0e113 --- /dev/null +++ b/crates/adaptive/src/closing/resume_tests.rs @@ -0,0 +1,246 @@ +//! The ancestor gate: which repairs may skip a committed prefix. +//! +//! Every case here is a way of getting this wrong that would not fail loudly. +//! A run that continues on a prefix the new graph would not have produced +//! completes, reports success, and is wrong about it — so the tests are +//! written from the unsafe side, and the safe cases exist to prove the gate is +//! not simply refusing everything. + +use serde_json::json; +use tinyflows::graph_ops::GraphOp; +use tinyflows::model::{Edge, Node, NodeKind, WorkflowGraph, WorkflowInput}; + +use super::may_continue; + +/// `start → fetch → review → post`. `review` is where these runs break: it has +/// a real upstream to protect and a real downstream that has not run. +fn graph() -> WorkflowGraph { + chain(&["start", "fetch", "review", "post"]) +} + +fn chain(ids: &[&str]) -> WorkflowGraph { + let nodes = ids + .iter() + .map(|id| Node { + id: (*id).to_string(), + kind: if *id == "start" { + NodeKind::Trigger + } else { + NodeKind::Agent + }, + type_version: 1, + name: (*id).to_string(), + config: json!({}), + ports: Vec::new(), + position: None, + }) + .collect(); + let edges = ids + .windows(2) + .map(|pair| Edge { + from_node: pair[0].to_string(), + from_port: "main".to_string(), + to_node: pair[1].to_string(), + to_port: "main".to_string(), + }) + .collect(); + WorkflowGraph { + schema_version: 1, + id: None, + name: "review".to_string(), + inputs: Vec::new(), + agents: Vec::new(), + nodes, + edges, + } +} + +fn edit(id: &str) -> GraphOp { + GraphOp::UpdateNodeConfig { + id: id.to_string(), + config: json!({ "prompt": "=.item.text" }), + } +} + +#[test] +fn editing_the_node_that_failed_may_continue() { + // The case worth having. A continue re-runs the failed node, so its new + // config is read fresh — and the prefix that produced its input is + // untouched. + assert!(may_continue( + &graph(), + &graph(), + "review", + &[edit("review")] + )); +} + +#[test] +fn editing_something_downstream_may_continue() { + // Nothing downstream of the failure has run, so nothing about it can + // conflict with what is committed. + assert!(may_continue(&graph(), &graph(), "review", &[edit("post")])); +} + +#[test] +fn editing_an_upstream_node_must_start_over() { + // `fetch` already ran and its output is committed. Continuing would run + // the repaired `review` against the *old* fetch's output while the store + // says the graph is the new one — green, and wrong about it. + assert!(!may_continue( + &graph(), + &graph(), + "review", + &[edit("fetch")] + )); +} + +#[test] +fn editing_the_trigger_must_start_over() { + // The trigger is an ancestor like any other, and its output seeds + // everything. + assert!(!may_continue( + &graph(), + &graph(), + "review", + &[edit("start")] + )); +} + +#[test] +fn a_repair_that_adds_an_upstream_node_must_start_over() { + // The dangerous direction, and the one a parent-only ancestor set misses: + // the new node is not an ancestor in the graph that RAN, only in the + // repaired one. Continuing would re-enter `review` with the upstream it + // was just given still missing — so the fix would look like it had not + // worked, and the next repair would chase the wrong thing. + let child = chain(&["start", "fetch", "enrich", "review", "post"]); + let ops = vec![ + GraphOp::AddNode { + node: Node { + id: "enrich".to_string(), + kind: NodeKind::Agent, + type_version: 1, + name: "enrich".to_string(), + config: json!({}), + ports: Vec::new(), + position: None, + }, + }, + GraphOp::AddEdge { + edge: Edge { + from_node: "enrich".to_string(), + from_port: "main".to_string(), + to_node: "review".to_string(), + to_port: "main".to_string(), + }, + }, + ]; + assert!(!may_continue(&graph(), &child, "review", &ops)); +} + +#[test] +fn a_repair_that_removes_an_upstream_node_must_start_over() { + // The other direction: `fetch` is an ancestor only in the graph that ran. + let child = chain(&["start", "review", "post"]); + let ops = vec![GraphOp::RemoveNode { + id: "fetch".to_string(), + }]; + assert!(!may_continue(&graph(), &child, "review", &ops)); +} + +#[test] +fn rewiring_an_edge_out_of_an_ancestor_must_start_over() { + // An edge is a statement about two nodes. Changing what an ancestor feeds + // changes that ancestor's meaning as surely as editing its config, and + // only one endpoint has to be upstream for that to bite. + let ops = vec![GraphOp::RemoveEdge { + from_node: "fetch".to_string(), + from_port: "main".to_string(), + to_node: "review".to_string(), + to_port: "main".to_string(), + }]; + assert!(!may_continue(&graph(), &graph(), "review", &ops)); +} + +#[test] +fn changing_the_declared_inputs_must_start_over() { + // Declared values are read by expressions anywhere, including in the + // prefix that already ran, so this invalidates every committed output at + // once — even though it names no node. + let ops = vec![GraphOp::SetWorkflowInputs { + inputs: vec![WorkflowInput::new( + "repo".to_string(), + tinyflows::model::InputType::String, + )], + }]; + assert!(!may_continue(&graph(), &graph(), "review", &ops)); +} + +#[test] +fn a_failed_node_the_graph_does_not_have_must_start_over() { + // Something is out of step — a stale point, the wrong thread — and + // guessing is the one thing this must not do. + assert!(!may_continue(&graph(), &graph(), "nonexistent", &[])); + assert!(!may_continue(&graph(), &graph(), "", &[])); +} + +#[test] +fn a_failure_in_the_first_step_may_continue_whatever_was_edited() { + // Not a special case, a consequence: a node with no ancestors has no + // committed prefix, so there is nothing an edit could invalidate. The + // continue saves nothing here, and is still correct. + assert!(may_continue(&graph(), &graph(), "start", &[edit("post")])); + assert!(may_continue(&graph(), &graph(), "start", &[edit("start")])); +} + +#[test] +fn a_diamond_protects_both_branches() { + // Ancestry is transitive and not a straight line. `left` and `right` both + // feed `join`, and an edit to either invalidates what `join` was given. + let mut graph = chain(&["start", "join", "post"]); + for id in ["left", "right"] { + graph.nodes.push(Node { + id: id.to_string(), + kind: NodeKind::Agent, + type_version: 1, + name: id.to_string(), + config: json!({}), + ports: Vec::new(), + position: None, + }); + graph.edges.push(Edge { + from_node: "start".to_string(), + from_port: "main".to_string(), + to_node: id.to_string(), + to_port: "main".to_string(), + }); + graph.edges.push(Edge { + from_node: id.to_string(), + from_port: "main".to_string(), + to_node: "join".to_string(), + to_port: "main".to_string(), + }); + } + assert!(!may_continue(&graph, &graph, "join", &[edit("left")])); + assert!(!may_continue(&graph, &graph, "join", &[edit("right")])); + assert!( + may_continue(&graph, &graph, "join", &[edit("post")]), + "the node after the join still has not run" + ); +} + +#[test] +fn a_cycle_upstream_does_not_hang_the_walk() { + // Loop nodes are closed by a back-edge, so the ancestor walk meets cycles + // in ordinary graphs. Seen-set termination, asserted rather than assumed. + let mut graph = chain(&["start", "head", "body", "review"]); + graph.edges.push(Edge { + from_node: "body".to_string(), + from_port: "main".to_string(), + to_node: "head".to_string(), + to_port: "main".to_string(), + }); + assert!(!may_continue(&graph, &graph, "review", &[edit("body")])); + assert!(may_continue(&graph, &graph, "review", &[edit("review")])); +} diff --git a/crates/adaptive/src/contracts.rs b/crates/adaptive/src/contracts.rs index a5defc9..934102a 100644 --- a/crates/adaptive/src/contracts.rs +++ b/crates/adaptive/src/contracts.rs @@ -421,3 +421,36 @@ mod tests { assert!(v.advanced); } } + +/// Where a failed run stopped, so a later attempt can carry on from it. +/// +/// A run that failed at a node leaves its prefix committed in the engine's +/// failure boundary. This is the handle to that: which thread holds it, and +/// which node it stopped at. A [`Runner`](crate::execute::Runner) reports one +/// on [`Ran`](crate::execute::Ran) when the host it runs on keeps +/// checkpoints; the loop hands one back on the next +/// [`Attempt`](crate::intake::Attempt) when — and only when — the repair it +/// made is safe to skip the prefix over +/// ([`may_continue`](crate::closing::may_continue)). +/// +/// A host with no checkpointer reports `None` and receives `None`, and every +/// attempt starts at the trigger exactly as it always did. +#[derive(Debug, Clone, PartialEq, Eq)] +pub struct ResumePoint { + /// The engine thread the failed run is checkpointed under — for most hosts + /// the run id they gave it. + pub thread: String, + /// The node whose failure ended that run. What a continue re-runs first, + /// and what the ancestor gate is computed against. + pub failed_node: String, + /// The workflow whose graph committed that prefix. + /// + /// Carried so the two sides can both check they mean the same run. The + /// loop only hands a point to an attempt that selected *this* workflow — + /// the chooser is free to pick something else, and a prefix committed by + /// one graph is not a prefix for another. A runner may check it again + /// against the graph it is about to run; a mismatch means continue was + /// asked for on the wrong thing, and starting from the trigger is the + /// correct answer. + pub workflow: String, +} diff --git a/crates/adaptive/src/driver.rs b/crates/adaptive/src/driver.rs index a1df3b7..320ac0a 100644 --- a/crates/adaptive/src/driver.rs +++ b/crates/adaptive/src/driver.rs @@ -30,6 +30,8 @@ use std::sync::Arc; +use crate::contracts::ResumePoint; +use crate::intake::Attempt; use tinyflows::caps::Capabilities; use tinyflows::store::WorkflowStore; @@ -132,6 +134,32 @@ impl Loop<'_> { /// When intake cannot decide, or the ledger cannot be read or written. /// Running never errors — see [`crate::execute`]. pub async fn attempt(&self, episode: &str, goal: &Goal) -> Result { + Ok(self.attempt_continuing(episode, goal, None).await?.0) + } + + /// [`attempt`](Self::attempt), given the chance to continue a run the + /// previous one left unfinished. + /// + /// Returns the close *and* the continuation the next attempt may use — + /// `Some` only when this attempt's repair left the failed node's whole + /// upstream untouched, and only for the variant that repair produced. + /// [`run`](Self::run) threads that value; a host driving attempts itself + /// threads it the same way, or passes `None` and pays for the prefix + /// again, which is what every attempt did before continuing existed. + /// + /// The continuation is carried rather than stored because it is worth + /// exactly one attempt: it names a boundary in a checkpointer whose thread + /// the *next* run will write over, and a stale one would re-enter a graph + /// on some other run's prefix. + /// + /// # Errors + /// As [`attempt`](Self::attempt). + pub async fn attempt_continuing( + &self, + episode: &str, + goal: &Goal, + from: Option, + ) -> Result<(Closed, Option)> { let record = self.start(episode, goal).await?; let attempt = record.attempt + 1; @@ -146,6 +174,22 @@ impl Loop<'_> { ) .await?; + // The continuation only applies to the workflow the repair produced. + // The chooser is free to pick something else entirely — a different + // variant, the parent, nothing at all — and a prefix committed by one + // graph is not a prefix for another. + let planned = match (from, &planned.approach) { + (Some(point), Approach::Selected { workflow_id, .. }) + if workflow_id == &point.workflow => + { + Attempt { + resume: Some(point), + ..planned + } + } + _ => planned, + }; + let ran = self.runner.run(&planned).await; let closed = closing::close( goal, @@ -173,13 +217,14 @@ impl Loop<'_> { .await; } - if closed.verdict.satisfied { + let next = if closed.verdict.satisfied { self.keep_if_it_generalises(goal, &planned).await; + None } else { - self.repair_if_the_graph_is_at_fault(goal, &closed, &planned.approach, &ran) - .await; - } - Ok(closed) + self.repair_if_the_graph_is_at_fault(goal, &closed, &planned, &ran) + .await + }; + Ok((closed, next)) } /// Drive an episode until it is satisfied or stands down. @@ -192,8 +237,10 @@ impl Loop<'_> { /// # Errors /// As [`attempt`](Self::attempt). pub async fn run(&self, episode: &str, goal: &Goal) -> Result { + let mut carry: Option = None; loop { - let closed = self.attempt(episode, goal).await?; + let (closed, next) = self.attempt_continuing(episode, goal, carry.take()).await?; + carry = next; let status = match &closed.next { Next::Retry => continue, Next::Done => EpisodeStatus::Satisfied, @@ -279,28 +326,28 @@ impl Loop<'_> { &self, goal: &Goal, closed: &Closed, - approach: &Approach, + planned: &Attempt, ran: &crate::execute::Ran, - ) { + ) -> Option { if closed.verdict.satisfied { - return; + return None; } // Whatever ran is the parent of the next repair — including a variant, // which makes a second generation. `Ledger::lineage` walks to the root, // so a grandchild is still compared inside one family. - let parent = match approach { + let parent = match &planned.approach { Approach::Selected { workflow_id, .. } => workflow_id, // Nothing to repair: an authored graph was written for this goal // and the next attempt writes another, seeing why this one fell // short. A variant of a one-off is a stored procedure nobody asked // for. - Approach::Authored { .. } => return, + Approach::Authored { .. } => return None, }; let evidence = ran.evidence(); if !graph_is_suspect(&closed.verdict, &evidence) { - return; + return None; } - let _ = closing::repair( + let variant = closing::repair( goal, &closed.verdict, &evidence, @@ -310,7 +357,24 @@ impl Loop<'_> { self.caps, self.conn, ) - .await; + .await + .ok() + .flatten()?; + + // The variant exists either way. What the gate decides is only whether + // the next attempt may skip the prefix — a `false` costs the work + // again, which is what every attempt cost before this existed. + let stopped = ran.resume.as_ref()?; + closing::may_continue( + &planned.graph, + &variant.record.graph, + &stopped.failed_node, + &variant.ops, + ) + .then(|| ResumePoint { + workflow: variant.record.id.clone(), + ..stopped.clone() + }) } } diff --git a/crates/adaptive/src/execute/mod.rs b/crates/adaptive/src/execute/mod.rs index 48faa35..328101c 100644 --- a/crates/adaptive/src/execute/mod.rs +++ b/crates/adaptive/src/execute/mod.rs @@ -129,6 +129,15 @@ pub struct Ran { pub steps: Vec, /// What the run cost, in the runner's unit. Zero means not measured. pub cost_usd: f64, + /// Where this run stopped, when it failed at a node and the host kept a + /// resumable boundary for it. + /// + /// `None` is always allowed and is the right answer for a host with no + /// checkpointer, a run that completed, or one that broke somewhere a node + /// boundary cannot describe. The loop treats it as "start the next attempt + /// from the trigger", which is what every attempt did before continuing + /// existed. + pub resume: Option, } impl Ran { diff --git a/crates/adaptive/src/execute/wire.rs b/crates/adaptive/src/execute/wire.rs index e1821a6..4db60f7 100644 --- a/crates/adaptive/src/execute/wire.rs +++ b/crates/adaptive/src/execute/wire.rs @@ -205,6 +205,10 @@ impl RunReport { } Ran { + // A wire report says what happened, not where to pick it back up: + // the boundary lives in the host's checkpointer, and only a host + // that keeps one can name it. + resume: None, outcome: RunOutcome { output: Value::Object(output), pending_approvals: self.pending_approvals, diff --git a/crates/adaptive/src/intake/author.rs b/crates/adaptive/src/intake/author.rs index 5cb632b..3693e5b 100644 --- a/crates/adaptive/src/intake/author.rs +++ b/crates/adaptive/src/intake/author.rs @@ -159,6 +159,8 @@ fn gated( inputs.retain(|name, _| graph.inputs.iter().any(|d| d.name == *name)); Ok(Attempt { + // See `select`: continuing is the loop's call, not intake's. + resume: None, approach: Approach::Authored { why, fingerprint: fingerprint(&graph), diff --git a/crates/adaptive/src/intake/mod.rs b/crates/adaptive/src/intake/mod.rs index 1cf2d37..b4e2b20 100644 --- a/crates/adaptive/src/intake/mod.rs +++ b/crates/adaptive/src/intake/mod.rs @@ -43,6 +43,15 @@ pub struct Attempt { pub graph: WorkflowGraph, /// Values for the graph's declared inputs, by name. pub inputs: Map, + /// Continue a previous run rather than starting this graph from its + /// trigger. + /// + /// Set only by the loop, only after a repair whose edits left the failed + /// node's whole upstream alone. A [`Runner`](crate::execute::Runner) that + /// sees `Some` re-enters the named node on the committed prefix; one that + /// cannot must run the graph normally rather than fail — continuing is an + /// optimisation over a correctness floor, never a requirement. + pub resume: Option, /// The lessons this attempt's planner was shown. /// /// Carried so the closing pass can score them against what happened. A diff --git a/crates/adaptive/src/intake/select.rs b/crates/adaptive/src/intake/select.rs index 011f46a..2a5304b 100644 --- a/crates/adaptive/src/intake/select.rs +++ b/crates/adaptive/src/intake/select.rs @@ -148,6 +148,9 @@ pub async fn select( }, graph: WorkflowGraph::default(), inputs: inputs_of(&answer), + // Intake never continues a run: only the loop knows whether the + // repair it just made is safe to skip a prefix over. + resume: None, // Filled by `decide`, which is what knows what the planner was shown. lessons_shown: Vec::new(), })) diff --git a/crates/adaptive/tests/closing.rs b/crates/adaptive/tests/closing.rs index 4d7ce4e..4d5c4e8 100644 --- a/crates/adaptive/tests/closing.rs +++ b/crates/adaptive/tests/closing.rs @@ -96,6 +96,7 @@ fn ran(outcome: &RunOutcome, diagnosis: &Diagnosis, changed: &str) -> Ran { failed: None, steps: Vec::new(), cost_usd: 0.0, + resume: None, } } diff --git a/crates/adaptive/tests/execute.rs b/crates/adaptive/tests/execute.rs index e45cf56..7f03a29 100644 --- a/crates/adaptive/tests/execute.rs +++ b/crates/adaptive/tests/execute.rs @@ -68,6 +68,7 @@ fn graph(nodes: Vec, edges: Vec) -> WorkflowGraph { fn attempt(graph: WorkflowGraph) -> Attempt { Attempt { + resume: None, approach: Approach::Authored { why: "for the test".into(), fingerprint: "0000000".into(), From 9a9843865c2abcc077cbbf23963dd8614db15d1f Mon Sep 17 00:00:00 2001 From: sanil-23 Date: Wed, 19 Aug 2026 12:45:55 +0530 Subject: [PATCH 6/6] fix(adaptive): give the merged-in recipe tests the catalogue argument MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit CI failed on the PR's merge commit while the branch alone compiled, which is the tell: `lower` gained a `&[Callable]` parameter here, and #61 landed on main afterwards with four new call sites that pass one argument. Git merged both cleanly — the conflict is semantic, and only the compiler sees it. They name no saved workflow, so the catalogue they get is empty. --- crates/adaptive/src/intake/recipe_tests.rs | 8 ++++---- 1 file changed, 4 insertions(+), 4 deletions(-) diff --git a/crates/adaptive/src/intake/recipe_tests.rs b/crates/adaptive/src/intake/recipe_tests.rs index aa6991e..e31ac96 100644 --- a/crates/adaptive/src/intake/recipe_tests.rs +++ b/crates/adaptive/src/intake/recipe_tests.rs @@ -133,7 +133,7 @@ fn a_declared_value_pasted_into_an_ask_is_refused_with_the_remedy() { { "id": "write", "ask": "Write a two-line poem about the topic 'warm caches'." } ] }); - let err = lower(&recipe).expect_err("refused").to_string(); + let err = lower(&recipe, &[]).expect_err("refused").to_string(); assert!(err.contains("pastes the value"), "{err}"); assert!(err.contains("attached automatically"), "{err}"); @@ -146,7 +146,7 @@ fn a_declared_value_pasted_into_an_ask_is_refused_with_the_remedy() { { "id": "write", "ask": "Write a two-line poem about the given topic." } ] }); - lower(&clean).expect("keepable"); + lower(&clean, &[]).expect("keepable"); } #[test] @@ -160,7 +160,7 @@ fn an_undeclared_input_value_in_an_ask_is_not_a_paste() { { "id": "write", "ask": "Write a two-line poem about warm caches." } ] }); - lower(&recipe).expect("not a paste — nothing declared"); + lower(&recipe, &[]).expect("not a paste — nothing declared"); } #[test] @@ -175,7 +175,7 @@ fn an_indistinct_input_value_in_an_ask_is_not_a_paste() { { "id": "flip", "ask": "Turn the feature on if the mode input says so." } ] }); - lower(&recipe).expect("not a paste"); + lower(&recipe, &[]).expect("not a paste"); } #[test]