Skip to content
Merged
Show file tree
Hide file tree
Changes from 1 commit
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
8 changes: 7 additions & 1 deletion crates/adaptive/src/closing/keep.rs
Original file line number Diff line number Diff line change
Expand Up @@ -57,7 +57,13 @@ a planner reads when deciding whether this procedure does what a NEW goal asks.
- reusable: false when this graph only makes sense for the one goal it was
written for, whatever its inputs say. A one-off kept in the catalogue is a row
every future planner reads and none can use, so say so rather than reaching
for a description that sounds general.";
for a description that sounds general.

Also false when the goal's specifics — a topic, a name, a repository, a
value — sit inside a node's prompt or config instead of arriving through a
declared input. Run unchanged for the NEXT goal of its class, that graph
does the old goal's specific thing, which is worse than not being found:
it is found, run, and wrong. Reusable means reusable as-is.";

/// What was kept, when anything was.
#[derive(Debug, Clone)]
Expand Down
228 changes: 227 additions & 1 deletion crates/adaptive/src/intake/author.rs
Original file line number Diff line number Diff line change
Expand Up @@ -10,6 +10,7 @@
//! is an error from intake rather than a run-time failure that reads like the
//! work failing.

use serde_json::Value;
use tinyflows::caps::Capabilities;
use tinyflows::catalog::{NodeKindContract, all_contracts};
use tinyflows::model::WorkflowGraph;
Expand Down Expand Up @@ -49,12 +50,21 @@ everything else is a literal.
=item.name a field of the direct predecessor's output
=nodes.fetch.item.json.body a field of any completed node, by node id
=run.trigger.payload what the trigger carried
=run.inputs.topic a declared workflow input, by its name
=.items | length a leading dot makes the rest a jq program

There are no braces. `={{ ... }}` is not a binding — it is a jq program that
fails to compile, and a failed program is null, so the step runs with an empty
value and reports success.

An `=` anywhere but the FIRST character is literal text, not a binding:
`\"about: =run.inputs.topic\"` sends those exact characters to the model. To
put a value inside prose — an agent prompt, a message — the whole string is
one expression, jq with explicit dots:

=\"Write a poem about \\(.run.inputs.topic)\"
=\"Summarise \" + .run.inputs.repo

`agent`, `tool_call` and `http_request` wrap their output in
`{json, text, raw}`. Their fields are under `.json`: write
`=nodes.fetch.item.json.body`, never `=nodes.fetch.item.body`. The second form
Expand Down Expand Up @@ -107,7 +117,44 @@ pub async fn author(
}
);

let answer = ask(caps, conn, Tier::Author, SYSTEM, &user).await?;
// The gates below produce readable refusals on purpose, and this loop is
// where they earn it: a refused graph goes back to the model with the
// refusal, once per round, rather than costing the whole episode. Bounded,
// because a model that cannot fix its graph in two more tries is telling
// us the answer.
//
// A reply that never became an answer — no JSON object, a transport
// failure — is retried too, but with the prompt unchanged: there is no
// graph to give feedback on, and a resample is the whole remedy.
let mut prompt = user;
let mut last: Option<IntakeError> = None;
for _ in 0..ROUNDS {
let answer = match ask(caps, conn, Tier::Author, SYSTEM, &prompt).await {
Ok(answer) => answer,
Err(err) => {
last = Some(err);
continue;
}
};
match gated(&answer, facts, policy) {
Ok(attempt) => return Ok(attempt),
Err(err) => {
prompt = format!(
"{prompt}\n\n# Your previous graph was refused — fix exactly this\n\
{err}\n\nReturn the corrected, complete JSON reply."
);
last = Some(err);
}
}
}
Err(last.unwrap_or_else(|| IntakeError::Inference("the author was never asked".to_string())))
}

/// How many replies the author gets before the failure is the answer.
const ROUNDS: usize = 3;

/// One reply through every gate, or why it was refused.
fn gated(answer: &Value, facts: &HostFacts, policy: &dyn HostPolicy) -> Result<Attempt> {
let raw = answer
.get("graph")
.cloned()
Expand All @@ -129,6 +176,17 @@ pub async fn author(
));
}

// The single most common authoring mistake, across every model tried: a
// binding path inside prose — `"about: .run.inputs.topic"` — which the
// engine reads as those literal characters, so the step runs on garbage
// and reports success. Mechanically detectable, so it is refused here and
// fixed through the feedback loop rather than found by a judge two
// minutes and one model call later.
let prose = prose_bindings(&graph);
if !prose.is_empty() {
return Err(IntakeError::Invalid(prose.join("; ")));
}

// Three gates, and the order is cost. `validate_all` is structural and
// free. `HostFacts::check` is our own reading of the machine's config.
// `check_graph` is the host's, which may know things we were not told —
Expand All @@ -153,6 +211,50 @@ pub async fn author(
})
}

/// Config strings that embed a binding path in prose instead of being one.
///
/// A string that does not start with `=` is a literal, whole. One that
/// mentions `run.inputs.`, `run.trigger.` or `nodes.<id>.` inside prose was
/// almost certainly meant to interpolate — and will instead hand the model,
/// the tool or the request those exact characters. Expression strings
/// (leading `=`) are exempt: `="about \(.run.inputs.topic)"` legitimately
/// contains the path.
fn prose_bindings(graph: &WorkflowGraph) -> Vec<String> {
const PATHS: [&str; 4] = ["run.inputs.", "run.trigger.", "=run.", "=nodes."];

Comment thread
coderabbitai[bot] marked this conversation as resolved.
Outdated
fn scan(node: &str, field: &str, value: &Value, out: &mut Vec<String>) {
match value {
Value::String(s) if !s.starts_with('=') => {
if PATHS.iter().any(|p| s.contains(p)) {
out.push(format!(
"node `{node}` config `{field}` embeds a binding path in literal \
text, which the engine passes through as those exact characters. \
Make the whole string one expression instead: \
=\"… \\(.run.inputs.name) …\""
));
}
}
Value::Object(map) => {
for (key, nested) in map {
scan(node, key, nested, out);
}
}
Value::Array(items) => {
for nested in items {
scan(node, field, nested, out);
}
}
_ => {}
}
}

let mut out = Vec::new();
for node in &graph.nodes {
scan(&node.id, "config", &node.config, &mut out);
}
out
}

/// A digest of the graph's runnable shape.
///
/// Nodes, edges and declared inputs — not the name, not the description. Two
Expand Down Expand Up @@ -246,6 +348,130 @@ mod tests {
);
}

#[tokio::test]
async fn a_refused_graph_goes_back_to_the_model_with_the_refusal() {
use std::sync::Mutex;

use tinyflows::caps::LlmProvider;
use tinyflows::caps::mock::mock_capabilities;

/// First reply: a graph with no trigger. Second: a valid one — but
/// only if the follow-up prompt actually carries the refusal.
struct Corrigible {
prompts: Mutex<Vec<String>>,
}

#[async_trait::async_trait]
impl LlmProvider for Corrigible {
async fn complete(
&self,
request: Value,
_conn: Option<&str>,
) -> tinyflows::error::Result<Value> {
let shown = request["messages"][1]["content"]
.as_str()
.unwrap_or_default()
.to_string();
let mut prompts = self.prompts.lock().expect("prompt log");
prompts.push(shown.clone());
let graph = if prompts.len() == 1 {
// No trigger: fails `validate_all`, must come back.
serde_json::json!({
"schema_version": 1, "name": "broken",
"inputs": [], "nodes": [], "edges": []
})
} else {
assert!(
shown.contains("refused"),
"the retry prompt must carry the refusal, got: {shown}"
);
serde_json::json!({
"schema_version": 1, "name": "fixed", "inputs": [],
"nodes": [{
"id": "start", "kind": "trigger", "name": "manual",
"config": { "trigger_kind": "manual" }
}],
"edges": []
})
};
Ok(serde_json::json!({ "graph": graph, "why": "test", "inputs": {} }))
}
}

#[derive(Debug, Default)]
struct Permissive;
impl HostPolicy for Permissive {}

let provider = std::sync::Arc::new(Corrigible {
prompts: Mutex::new(Vec::new()),
});
let caps = Capabilities {
llm: provider.clone(),
..mock_capabilities()
};
let attempt = author(
&Goal::new("do the thing"),
&HostFacts::unknown(),
&Permissive,
"",
&caps,
None,
)
.await
.expect("the corrected graph must land");
assert_eq!(attempt.graph.name, "fixed");
assert_eq!(provider.prompts.lock().expect("prompt log").len(), 2);
}

#[test]
fn a_binding_path_inside_prose_is_refused_with_the_remedy() {
use tinyflows::model::{Edge, Node, NodeKind};

let graph = WorkflowGraph {
schema_version: 1,
name: "poem".into(),
nodes: vec![
Node {
id: "start".into(),
kind: NodeKind::Trigger,
type_version: 1,
name: "manual".into(),
config: serde_json::json!({ "trigger_kind": "manual" }),
ports: Vec::new(),
position: None,
},
Node {
id: "poet".into(),
kind: NodeKind::Agent,
type_version: 1,
name: "poet".into(),
config: serde_json::json!({
"prompt": "Write a poem about: .run.inputs.topic"
}),
ports: Vec::new(),
position: None,
},
],
edges: vec![Edge {
from_node: "start".into(),
from_port: "main".into(),
to_node: "poet".into(),
to_port: "main".into(),
}],
..WorkflowGraph::default()
};

let found = prose_bindings(&graph);
assert_eq!(found.len(), 1, "{found:?}");
assert!(found[0].contains("poet"), "names the node: {}", found[0]);

// The legitimate form is exempt: the whole string is an expression.
let mut fixed = graph;
fixed.nodes[1].config =
serde_json::json!({ "prompt": "=\"Write a poem about \\(.run.inputs.topic)\"" });
assert!(prose_bindings(&fixed).is_empty());
}

#[test]
fn a_graph_with_no_trigger_is_refused_rather_than_returned() {
// Not reachable through `author` without a provider, so the invariant
Expand Down
38 changes: 34 additions & 4 deletions crates/adaptive/src/intake/mod.rs
Original file line number Diff line number Diff line change
Expand Up @@ -135,10 +135,40 @@ pub async fn decide(
// `select` answers with an id; the graph and the input check come from
// the store. Returning the choice unbound would hand the engine an
// empty graph, which compiles to nothing and reads as the work failing.
return bind(chosen, store).map(|attempt| Attempt {
lessons_shown: shown,
..attempt
});
match bind(chosen, store) {
Ok(attempt) => {
return Ok(Attempt {
lessons_shown: shown,
..attempt
});
}
Err(refusal) => {
// A sound choice that failed to bind — the model asserted
// inputs it did not supply. That is a correctable slip, not a
// reason to end the episode: one more selection round with
// the refusal on the table, and authoring after that, because
// authoring can always produce something runnable.
let noted = format!(
"{past}\n\n# A selection just failed to bind\n{refusal}\n\
Supply a value for every required input this time, or \
decline so a graph is written instead."
);
if let Some(retry) = select(goal, &candidates, &noted, caps, conn).await?
&& let Ok(attempt) = bind(retry, store)
Comment thread
coderabbitai[bot] marked this conversation as resolved.
Outdated
{
return Ok(Attempt {
lessons_shown: shown,
..attempt
});
}
return author(goal, facts, store.policy(), &noted, caps, conn)
Comment thread
coderabbitai[bot] marked this conversation as resolved.
Outdated
.await
.map(|attempt| Attempt {
lessons_shown: shown,
..attempt
});
}
}
}
author(goal, facts, store.policy(), &past, caps, conn)
.await
Expand Down
11 changes: 11 additions & 0 deletions crates/adaptive/src/intake/select.rs
Original file line number Diff line number Diff line change
Expand Up @@ -147,6 +147,12 @@ pub async fn select(
/// The model is confident about inputs it did not actually find in the goal, so
/// the cheap deterministic check catches what the expensive one asserted.
///
/// The check runs in both directions. A required input the model did not
/// supply is an error. An input the model supplied that the graph never
/// declared is *dropped*: the engine rejects undeclared keys before any node
/// executes, so one invented key — and models invent them freely — would
/// otherwise turn a sound selection into an attempt that ran nothing.
///
/// # Errors
/// When the workflow is gone, or an input has no value.
pub fn bind(attempt: Attempt, store: &dyn WorkflowStore) -> Result<Attempt> {
Expand Down Expand Up @@ -177,6 +183,11 @@ pub fn bind(attempt: Attempt, store: &dyn WorkflowStore) -> Result<Attempt> {
}
}

let mut attempt = attempt;
attempt
.inputs
.retain(|name, _| record.graph.inputs.iter().any(|d| d.name == *name));

Ok(Attempt {
graph: record.graph,
..attempt
Expand Down
Loading