From 0f2542b8b320cc5a834801ddc3477578634f1563 Mon Sep 17 00:00:00 2001 From: Will Eaton Date: Wed, 7 Oct 2026 14:21:17 -0400 Subject: [PATCH] Let a route decide each item of a list A route could not declare over, so a playbook that fans work out over a list could only gate the whole fan-out on one decision. A route may now map over a list (C-ROUTE-EACH, ADR-0062): each item is one decision, recorded as its instance's row, so a resume asks only about items it has no answer for. Two tasks mapped over the same producer field are aligned. An edge between them is read per item: instance K receives the dependency's instance K in place of its fold, under any join, and a when on an aligned route is read against that route's instance K. An item whose answer is not listed settles not taken without dispatch; a fan-out counts such items under not_taken instead of failing on them, and one whose every item was untaken settles not taken. keyed lets any mapped task narrow an object field of an unmapped dependency to its item's entry. A route's files put dependencies' schema-declared JSON files in the model's state under files (C-ROUTE-FILES), so a route can judge an agent's RESULT.json. An output-decided mapped route reads each item's answer from its aligned source, or from an unmapped source's object keyed by item. The decision binding and its key are withheld from every task process (C-INFERENCE, ADR-0064): only a route asks the decision model. plan_admitted carries keyed; the fan-out summary carries not_taken. CONTRACT_VERSION 1.16.0. --- crucible-contract/src/emits.rs | 9 + crucible-contract/src/lib.rs | 2 +- crucible-contract/src/session.rs | 6 + crucible/src/agent/mod.rs | 2 + crucible/src/control/recovery.rs | 2 + crucible/src/inference.rs | 17 +- crucible/src/plan/cli.rs | 4 +- crucible/src/plan/events.rs | 1 + crucible/src/plan/exec.rs | 785 +++++++++++++++--- crucible/src/plan/harness.rs | 86 ++ crucible/src/plan/ir.rs | 590 ++++++++++++- crucible/src/plan/machine.rs | 6 + crucible/src/plan/runner.rs | 14 + crucible/src/plan/starlark.rs | 86 +- crucible/src/plan/starlark/error.rs | 12 + crucible/src/plan/starlark/idents.rs | 4 + crucible/src/plan/starlark/reference.rs | 32 + crucible/src/plan/template.rs | 1 + crucible/src/runloop/graph.rs | 3 + crucible/tests/contract_version.rs | 2 +- crucible/tests/route_e2e.rs | 88 ++ docs/adr/ADR-0062.md | 94 +++ docs/adr/ADR-0064.md | 56 ++ docs/crucible-contract.md | 6 + docs/dsl-reference.md | 8 + docs/playbook-patterns.md | 57 ++ docs/rfc/RFC-0001.md | 41 +- ...ver-the-list-the-gated-work-maps-over.toml | 125 +++ ...-only-a-route-asks-the-decision-model.toml | 82 ++ gov/rfc/RFC-0001/clauses/C-INFERENCE.toml | 2 +- gov/rfc/RFC-0001/clauses/C-ROUTE-EACH.toml | 26 + gov/rfc/RFC-0001/clauses/C-ROUTE-FILES.toml | 16 + gov/rfc/RFC-0001/clauses/C-ROUTE.toml | 2 +- gov/rfc/RFC-0001/rfc.toml | 7 + 34 files changed, 2120 insertions(+), 154 deletions(-) create mode 100644 docs/adr/ADR-0062.md create mode 100644 docs/adr/ADR-0064.md create mode 100644 gov/adr/ADR-0062-per-element-decisions-are-routes-mapped-over-the-list-the-gated-work-maps-over.toml create mode 100644 gov/adr/ADR-0064-only-a-route-asks-the-decision-model.toml create mode 100644 gov/rfc/RFC-0001/clauses/C-ROUTE-EACH.toml create mode 100644 gov/rfc/RFC-0001/clauses/C-ROUTE-FILES.toml diff --git a/crucible-contract/src/emits.rs b/crucible-contract/src/emits.rs index dc501d57..6bba33d3 100644 --- a/crucible-contract/src/emits.rs +++ b/crucible-contract/src/emits.rs @@ -134,6 +134,15 @@ impl FieldType { } } + /// True for a type every value of which is a JSON object. + pub fn is_object(&self) -> bool { + match self { + FieldType::Object => true, + FieldType::Schema(schema) => schema.coarse() == Some(FieldType::Object), + _ => false, + } + } + /// True for a type every value of which is a JSON array. pub fn is_list(&self) -> bool { match self { diff --git a/crucible-contract/src/lib.rs b/crucible-contract/src/lib.rs index 8e1847e9..b3646a14 100644 --- a/crucible-contract/src/lib.rs +++ b/crucible-contract/src/lib.rs @@ -34,7 +34,7 @@ pub mod verdict; /// `crucible --contract-version` and the runtime image carries it as the /// `io.crucible.contract-version` OCI label, so a deployed image can be matched against the /// controller it talks to without a probe. -pub const CONTRACT_VERSION: &str = "1.15.0"; +pub const CONTRACT_VERSION: &str = "1.16.0"; pub use admission::{ ADMISSION_WIRE_VERSION, AdmissionEvent, AdmissionKey, AdmissionOutcome, AdmittedInput, diff --git a/crucible-contract/src/session.rs b/crucible-contract/src/session.rs index 9a0b660c..e3127d71 100644 --- a/crucible-contract/src/session.rs +++ b/crucible-contract/src/session.rs @@ -142,6 +142,10 @@ pub struct PlanTaskWire { /// worst-case width before any spend. #[serde(default)] pub max_fanout: u32, + /// Each `producer.field` a mapped task narrows to its own element's entry (contract 1.16.0), + /// empty when it declares none. + #[serde(default, skip_serializing_if = "Vec::is_empty")] + pub keyed: Vec, /// `route.question in a|b` when the task runs only on those answers, empty otherwise. #[serde(default)] pub when: String, @@ -927,6 +931,7 @@ mod tests { stage: "iteration".into(), over: "discover.targets".into(), max_fanout: 8, + keyed: vec!["discover.facts".into()], when: String::new(), revise: vec!["draft".into(), "check".into()], max_rounds: 3, @@ -1024,6 +1029,7 @@ mod tests { stage: "iteration".into(), over: String::new(), max_fanout: 0, + keyed: Vec::new(), when: String::new(), revise, max_rounds: 0, diff --git a/crucible/src/agent/mod.rs b/crucible/src/agent/mod.rs index 79056974..75fdafe9 100644 --- a/crucible/src/agent/mod.rs +++ b/crucible/src/agent/mod.rs @@ -137,6 +137,7 @@ fn command_backend( session: Option<&crate::agent::agent_session::SessionTurn>, ) -> Command { let mut c = Command::new("sh"); + crucible::inference::withhold_from_task(&mut c); c.arg("-c") .arg(cmd) .current_dir(&p.workspace) @@ -191,6 +192,7 @@ fn local_command( return Err(std::io::Error::other("harness produced an empty argv")); }; let mut cmd = Command::new(program); + crucible::inference::withhold_from_task(&mut cmd); cmd.args(rest) .current_dir(&p.workspace) .stdin(Stdio::null()) diff --git a/crucible/src/control/recovery.rs b/crucible/src/control/recovery.rs index d0b1fd44..90c99fa0 100644 --- a/crucible/src/control/recovery.rs +++ b/crucible/src/control/recovery.rs @@ -932,6 +932,7 @@ mod tests { stage: String::new(), over: String::new(), max_fanout: 0, + keyed: Vec::new(), when: String::new(), revise: Vec::new(), timeout: String::new(), @@ -951,6 +952,7 @@ mod tests { stage: String::new(), over: String::new(), max_fanout: 0, + keyed: Vec::new(), when: String::new(), revise: Vec::new(), timeout: String::new(), diff --git a/crucible/src/inference.rs b/crucible/src/inference.rs index 1a108564..75f7e149 100644 --- a/crucible/src/inference.rs +++ b/crucible/src/inference.rs @@ -1,7 +1,22 @@ //! The run's injected inference bindings, read from the process environment once per use. -use crucible_contract::inference::{ENV_INFERENCE, InferenceError, ResolvedInference}; +use crucible_contract::inference::{ + ENV_INFERENCE, InferenceError, InferenceRole, ResolvedInference, +}; pub fn from_process_env() -> Result { ResolvedInference::parse(&std::env::var(ENV_INFERENCE).unwrap_or_default()) } + +/// Keep the inference document and the decision binding's credential out of a task process. A +/// decision has no consumer outside the engine (RFC-0001 C-INFERENCE): only a route task asks it. +pub fn withhold_from_task(cmd: &mut std::process::Command) { + if let Ok(inference) = from_process_env() + && let Some(key) = inference + .binding(InferenceRole::Decision) + .and_then(|binding| binding.key_env.as_ref()) + { + cmd.env_remove(key.as_str()); + } + cmd.env_remove(ENV_INFERENCE); +} diff --git a/crucible/src/plan/cli.rs b/crucible/src/plan/cli.rs index 6f5aebf7..26354bf6 100644 --- a/crucible/src/plan/cli.rs +++ b/crucible/src/plan/cli.rs @@ -114,7 +114,9 @@ pub fn render(plan: &ValidPlan, caps: &BTreeSet) -> String { .unwrap_or_default() ), TaskKind::TopK { k, .. } => format!("top_k[k={k}]"), - TaskKind::Route { questions, decider } => format!( + TaskKind::Route { + questions, decider, .. + } => format!( "route[{}; {}]", questions .keys() diff --git a/crucible/src/plan/events.rs b/crucible/src/plan/events.rs index f0109cf4..90fc6b22 100644 --- a/crucible/src/plan/events.rs +++ b/crucible/src/plan/events.rs @@ -30,6 +30,7 @@ pub(crate) fn plan_admitted_event( .map(crate::plan::ir::OutputRef::to_string) .unwrap_or_default(), max_fanout: t.max_fanout.unwrap_or_default(), + keyed: t.keyed.iter().map(ToString::to_string).collect(), when: t.when.as_ref().map(ToString::to_string).unwrap_or_default(), revise: t .revise diff --git a/crucible/src/plan/exec.rs b/crucible/src/plan/exec.rs index 778c12c8..6b1f0936 100644 --- a/crucible/src/plan/exec.rs +++ b/crucible/src/plan/exec.rs @@ -21,7 +21,8 @@ use crate::diagram::IllegalTransition; use crate::plan::history::SeriesHistory; use crate::plan::ir::{ COMPLETE_FIELD, Decider, HISTORY_INPUT, ITEM_INPUT, Join, OUTCOME_INPUT, PARAMS_INPUT, - REASON_FIELD, REVISION_INPUT, ReviseLoop, Stage, Task, TaskKind, TaskName, ValidPlan, + REASON_FIELD, REVISION_INPUT, ReviseLoop, Stage, Task, TaskKind, TaskName, ValidPlan, When, + aligned, }; use crate::plan::machine::{ BlockedReason, PlanEvent, PlanMachine, TaskEvent, TaskMachine, TaskState, @@ -419,6 +420,8 @@ pub struct FanoutSummary { pub instances: usize, pub passed: usize, pub failed: usize, + #[serde(default)] + pub not_taken: usize, } #[derive(Clone, Debug, PartialEq, Serialize)] @@ -786,7 +789,7 @@ pub fn execute( if t.depends_on.iter().any(|d| !results.contains_key(d)) { continue; } - if let Some((event, note)) = not_taken(t, &results) { + if let Some((event, note)) = not_taken(plan, t, &results) { let r = TaskResult::undispatched(TaskStatus::NotTaken, note); record( &mut *runner, @@ -801,7 +804,7 @@ pub fn execute( )?; continue; } - if !dependencies_allow(t, &results) { + if !dependencies_allow(plan, t, &results) { // Nothing runs on top of a failure (or a skip): advisory failures gate // their dependents even though they never gate validity. let reason = BlockedReason::DependencyDidNotPass; @@ -995,16 +998,51 @@ pub fn execute( Some((&instance.name, results.get(&instance.name)?.clone())) }) .collect(); - // An instance is staged under its own name: it runs in its own root, and - // the node itself never runs. - let producers: Vec<&Task> = producers_for + let node_producers: &[Task] = producers_for .get(&node.name) - .map(|producers| producers.iter().collect()) + .map(Vec::as_slice) .unwrap_or_default(); - let refused = instances - .iter() - .filter(|instance| !folded.contains_key(&instance.name)) - .find_map(|instance| runner.stage(instance, &producers).err()); + // An instance reads an aligned dependency's files from that dependency's + // instance for the same key, and no other instance of it. + let producers_of = |key: &str| -> Vec { + node_producers + .iter() + .filter(|p| { + !node.depends_on.iter().any(|d| { + aligned_dependency(plan, node, d) + && is_instance_of(d, &p.name) + && p.name != instance_name(d, key) + }) + }) + .cloned() + .collect() + }; + let mut prepared: BTreeMap> = + BTreeMap::new(); + let mut undispatched: BTreeMap = + BTreeMap::new(); + let mut refused = None; + for (instance, key) in instances.iter().zip(&keys) { + if folded.contains_key(&instance.name) { + continue; + } + let producers = producers_of(key); + match element(plan, node, key, &base, &results, &producers) { + Element::Settle(result, event) => { + undispatched.insert(instance.name.clone(), (result, event)); + } + Element::Dispatch(inputs) => { + // An instance is staged under its own name: it runs in its own + // root, and the node itself never runs. + let staged: Vec<&Task> = producers.iter().collect(); + if let Err(why) = runner.stage(instance, &staged) { + refused = Some(why); + break; + } + prepared.insert(instance.name.clone(), inputs); + } + } + } if let Some(why) = refused { let reason = BlockedReason::StagingRefused(why); let r = TaskResult::blocked(&reason); @@ -1025,23 +1063,35 @@ pub fn execute( .entry(node.name.clone()) .or_default() .advance(TaskEvent::FannedOut)?; - let item_inputs = |key: &String| { - let mut inputs = base.clone(); - inputs.insert(TaskName(ITEM_INPUT.to_string()), Value::String(key.clone())); - inputs - }; // Each instance settles in its own right, so a reader sees one row per item // rather than one row standing for all of them. let mut settled: Vec<(String, TaskResult)> = Vec::new(); + for (instance, key) in instances.iter().zip(&keys) { + let Some((result, event)) = undispatched.remove(&instance.name) else { + continue; + }; + settled.push((key.clone(), result.clone())); + record( + &mut *runner, + instance, + result, + event, + &mut results, + &mut machines, + &mut halted, + &mut plan_machine, + false, + )?; + } + let to_run = |task: &Task| prepared.contains_key(&task.name); if node.isolation.is_some() { let batch: Vec> = instances .iter() - .zip(&keys) - .filter(|(task, _)| !folded.contains_key(&task.name)) - .map(|(task, key)| BatchItem { + .filter(|task| to_run(task)) + .map(|task| BatchItem { task, attempt: 1, - inputs: item_inputs(key), + inputs: prepared.get(&task.name).cloned().unwrap_or_default(), deadline: None, }) .collect(); @@ -1066,6 +1116,9 @@ pub fn execute( settled.push((key.clone(), result.clone())); continue; } + if !to_run(instance) { + continue; + } let Some((task, result, event)) = live.next() else { break; }; @@ -1092,6 +1145,9 @@ pub fn execute( settled.push((key.clone(), result.clone())); continue; } + let Some(inputs) = prepared.get(&instance.name) else { + continue; + }; if spent >= budget { halt(&mut halted, &mut plan_machine, Halt::Budget)?; } else if clock.past(cfg.wall_clock) { @@ -1118,15 +1174,30 @@ pub fn execute( .entry(instance.name.clone()) .or_default() .advance(TaskEvent::Dispatched)?; - let (result, event, overrun) = run_with_retries( - instance, - &item_inputs(key), - cfg, - run_ceiling, - runner, - &mut spent, - budget, - ); + let (result, event, overrun) = match &instance.task { + TaskKind::Route { + questions, + decider: Decider::Output { task: source }, + .. + } => { + let decided = decide_from_output(inputs, questions, source); + let event = if decided.status == TaskStatus::Pass { + TaskEvent::Passed + } else { + TaskEvent::Failed + }; + (decided, event, Overrun::default()) + } + _ => run_with_retries( + instance, + inputs, + cfg, + run_ceiling, + runner, + &mut spent, + budget, + ), + }; overrun.halt(node.stage, &mut halted, &mut plan_machine)?; runner.settled(instance, result.status == TaskStatus::Pass); settled.push((key.clone(), result.clone())); @@ -1143,11 +1214,14 @@ pub fn execute( )?; } } + // Settled order follows the keys, so a fold reads the same whichever path + // settled an instance. + settled.sort_by_key(|(key, _)| keys.iter().position(|k| k == key)); let folded = fold_instances(settled); - let event = if folded.status == TaskStatus::Pass { - TaskEvent::InstancesPassed - } else { - TaskEvent::InstancesFailed + let event = match folded.status { + TaskStatus::Pass => TaskEvent::InstancesPassed, + TaskStatus::NotTaken => TaskEvent::InstancesNotTaken, + _ => TaskEvent::InstancesFailed, }; record( &mut *runner, @@ -1182,6 +1256,7 @@ pub fn execute( TaskKind::Route { questions, decider: Decider::Output { task: source }, + .. } => { let decided = decide_from_output(&inputs, questions, source); let event = if decided.status == TaskStatus::Pass { @@ -1792,9 +1867,9 @@ where let settled = if let Some(halt) = halted { let reason = halt.blocked(); Err((TaskResult::blocked(&reason), reason.event())) - } else if let Some((event, note)) = not_taken(t, &view) { + } else if let Some((event, note)) = not_taken(plan, t, &view) { Err((TaskResult::undispatched(TaskStatus::NotTaken, note), event)) - } else if !dependencies_allow(t, &view) { + } else if !dependencies_allow(plan, t, &view) { let reason = BlockedReason::DependencyDidNotPass; Err((TaskResult::blocked(&reason), reason.event())) } else if let Some(halt) = ceiling_reached(cfg, ledger) { @@ -2113,18 +2188,31 @@ fn rounds_event(status: TaskStatus) -> TaskEvent { } /// Whether `t`'s join lets it dispatch on its dependencies' settled results. -fn dependencies_allow(t: &Task, results: &BTreeMap) -> bool { +/// Whether `d` is a dependency `t` reads per element: both map over the same list. +fn aligned_dependency(plan: &ValidPlan, t: &Task, d: &TaskName) -> bool { + plan.get(d).is_some_and(|dependency| aligned(t, dependency)) +} + +/// Whether the node's dependencies let it dispatch. An aligned dependency is judged per element +/// once the node expands ([`element`]), so only the others count here. +fn dependencies_allow( + plan: &ValidPlan, + t: &Task, + results: &BTreeMap, +) -> bool { + let mut unaligned = t + .depends_on + .iter() + .filter(|d| !aligned_dependency(plan, t, d)); match t.join { - Join::All => t - .depends_on - .iter() - .all(|d| results.get(d).map(|r| r.status) == Some(TaskStatus::Pass)), + Join::All => unaligned.all(|d| results.get(d).map(|r| r.status) == Some(TaskStatus::Pass)), // Only passing outputs feed a lossy join. A mapped node contributes when any of its - // instances passed, so a fan-out reads the same as a set of siblings. - Join::Passed => t - .depends_on - .iter() - .any(|d| results.get(d).is_some_and(TaskResult::contributed)), + // instances passed, so a fan-out reads the same as a set of siblings. An aligned + // dependency may contribute for some elements, which only the elements can tell. + Join::Passed => { + t.depends_on.iter().any(|d| aligned_dependency(plan, t, d)) + || unaligned.any(|d| results.get(d).is_some_and(TaskResult::contributed)) + } // Every dependency already holds a result by the caller's guard, so terminality is // established and no status blocks the dispatch. Join::Settled => true, @@ -2137,11 +2225,10 @@ fn dependencies_allow(t: &Task, results: &BTreeMap) -> boo /// dependent that wants whatever survived says so with `join = "passed"`, which is the vocabulary /// that already exists; a fan-out needs no join of its own. fn fold_instances(settled: Vec<(String, TaskResult)>) -> TaskResult { - let passed = settled - .iter() - .filter(|(_, r)| r.status == TaskStatus::Pass) - .count(); - let failed = settled.len() - passed; + let count = |status| settled.iter().filter(|(_, r)| r.status == status).count(); + let passed = count(TaskStatus::Pass); + let not_taken = count(TaskStatus::NotTaken); + let failed = settled.len() - passed - not_taken; let cost: f64 = settled.iter().map(|(_, r)| r.cost_usd).sum(); // Only passing instances feed a downstream join. A failed instance contributed nothing, and // a null under its key reads as "it ran and found nothing", which is a different claim. @@ -2153,7 +2240,7 @@ fn fold_instances(settled: Vec<(String, TaskResult)>) -> TaskResult { let note = (failed > 0).then(|| { let names: Vec<&str> = settled .iter() - .filter(|(_, r)| r.status != TaskStatus::Pass) + .filter(|(_, r)| !matches!(r.status, TaskStatus::Pass | TaskStatus::NotTaken)) .map(|(key, _)| key.as_str()) .collect(); format!( @@ -2162,20 +2249,24 @@ fn fold_instances(settled: Vec<(String, TaskResult)>) -> TaskResult { names.join(", ") ) }); + let status = if failed > 0 { + TaskStatus::Fail + } else if !settled.is_empty() && not_taken == settled.len() { + TaskStatus::NotTaken + } else { + TaskStatus::Pass + }; TaskResult { blocked: None, transport: None, - status: if failed == 0 { - TaskStatus::Pass - } else { - TaskStatus::Fail - }, + status, attempts: 1, cost_usd: cost, output: Some(serde_json::json!({ "instances": settled.len(), "passed": passed, "failed": failed, + "not_taken": not_taken, "outputs": Value::Object(outputs), })), note, @@ -2183,6 +2274,7 @@ fn fold_instances(settled: Vec<(String, TaskResult)>) -> TaskResult { instances: settled.len(), passed, failed, + not_taken, }), repairs: Vec::new(), } @@ -2529,49 +2621,22 @@ fn run_batch_with_retries<'a>( /// Engine-built-in fold: keep the k best inputs by their `score` field. /// Output: `{"kept": [{"task": ..., "score": ...}, ...]}`, best first. /// Why `t` settles not taken without dispatch, if it does. Every dependency holds a result. -fn not_taken(t: &Task, results: &BTreeMap) -> Option<(TaskEvent, String)> { - let status = |d: &TaskName| results.get(d).map(|r| r.status); - if let Some(when) = &t.when { - match status(&when.task) { - Some(TaskStatus::NotTaken) => { - return Some(( - TaskEvent::BranchNotTaken, - format!("route {} was not taken", when.task), - )); - } - Some(TaskStatus::Pass) => { - let label = results - .get(&when.task) - .and_then(|r| r.output.as_ref()) - .and_then(|o| o.get(when.question.as_str())) - .and_then(|a| a.get("label")) - .and_then(Value::as_str); - if !when.is.iter().any(|l| Some(l.as_str()) == label) { - return Some(( - TaskEvent::ConditionUnmet, - format!( - "{}.{} resolved to {}", - when.task, - when.question, - label.unwrap_or("nothing") - ), - )); - } - } - _ if t.join != Join::All => { - return Some(( - TaskEvent::ConditionUnmet, - format!("route {} did not decide", when.task), - )); - } - _ => {} - } +fn not_taken( + plan: &ValidPlan, + t: &Task, + results: &BTreeMap, +) -> Option<(TaskEvent, String)> { + if let Some(when) = &t.when + && !aligned_dependency(plan, t, &when.task) + && let Some(unmet) = when_unmet(when, &when.task, results.get(&when.task), t.join) + { + return Some(unmet); } if t.join == Join::All - && let Some(d) = t - .depends_on - .iter() - .find(|d| status(d) == Some(TaskStatus::NotTaken)) + && let Some(d) = t.depends_on.iter().find(|d| { + !aligned_dependency(plan, t, d) + && results.get(*d).map(|r| r.status) == Some(TaskStatus::NotTaken) + }) { return Some(( TaskEvent::BranchNotTaken, @@ -2581,6 +2646,217 @@ fn not_taken(t: &Task, results: &BTreeMap) -> Option<(Task None } +/// Why `when` keeps a task from dispatching, read against `decided`, the result of the route +/// (or, for an element, of the route's instance for the same key) recorded under `route`. +/// `None` lets the task through, or leaves an undecided route under `all` to block it. +fn when_unmet( + when: &When, + route: &TaskName, + decided: Option<&TaskResult>, + join: Join, +) -> Option<(TaskEvent, String)> { + match decided.map(|r| r.status) { + Some(TaskStatus::NotTaken) => Some(( + TaskEvent::BranchNotTaken, + format!("route {route} was not taken"), + )), + Some(TaskStatus::Pass) => { + let label = decided + .and_then(|r| r.output.as_ref()) + .and_then(|o| o.get(when.question.as_str())) + .and_then(|a| a.get("label")) + .and_then(Value::as_str); + (!when.is.iter().any(|l| Some(l.as_str()) == label)).then(|| { + ( + TaskEvent::ConditionUnmet, + format!( + "{route}.{} resolved to {}", + when.question, + label.unwrap_or("nothing") + ), + ) + }) + } + _ if join != Join::All => Some(( + TaskEvent::ConditionUnmet, + format!("route {route} did not decide"), + )), + _ => None, + } +} + +/// What one element of a mapped node dispatches with, or how it settles without dispatch. +enum Element { + Dispatch(BTreeMap), + Settle(TaskResult, TaskEvent), +} + +/// One element of a mapped node ([[RFC-0001:C-ROUTE-EACH]]): a `when` on an aligned route and +/// every aligned dependency are read through their instance for `key`, each keyed field is +/// narrowed to its entry under `key`, and the item itself rides under [`ITEM_INPUT`]. +/// +/// ```text +/// triage[K] ──when──► judge[K] ◄──all── analyze[K] instance K reads instance K +/// │ │ +/// └── keyed: prepare.lanes ──► prepare.lanes[K] unmapped object narrowed to K +/// ``` +fn element( + plan: &ValidPlan, + node: &Task, + key: &str, + base: &BTreeMap, + results: &BTreeMap, + staged: &[Task], +) -> Element { + let instance = |d: &TaskName| results.get(&instance_name(d, key)); + if let Some(when) = &node.when + && aligned_dependency(plan, node, &when.task) + { + let route = instance_name(&when.task, key); + if let Some((event, note)) = when_unmet(when, &route, instance(&when.task), node.join) { + return Element::Settle(TaskResult::undispatched(TaskStatus::NotTaken, note), event); + } + } + let mut inputs = base.clone(); + let mut contributed = node.join != Join::Passed + || node.depends_on.iter().any(|d| { + !aligned_dependency(plan, node, d) + && results.get(d).is_some_and(TaskResult::contributed) + }); + for d in node + .depends_on + .iter() + .filter(|d| aligned_dependency(plan, node, d)) + { + let settled = instance(d); + let status = settled.map_or(TaskStatus::Blocked, |r| r.status); + match node.join { + Join::All if status == TaskStatus::NotTaken => { + return Element::Settle( + TaskResult::undispatched( + TaskStatus::NotTaken, + format!("dependency {} was not taken", instance_name(d, key)), + ), + TaskEvent::BranchNotTaken, + ); + } + Join::All if status != TaskStatus::Pass => { + let reason = BlockedReason::DependencyDidNotPass; + return Element::Settle(TaskResult::blocked(&reason), reason.event()); + } + Join::All => { + let output = settled + .and_then(|r| r.output.clone()) + .unwrap_or(Value::Null); + inputs.insert(d.clone(), output); + } + Join::Passed if status == TaskStatus::Pass => { + let output = settled + .and_then(|r| r.output.clone()) + .unwrap_or(Value::Null); + inputs.insert(d.clone(), output); + contributed = true; + } + Join::Passed => { + inputs.remove(d); + } + Join::Settled => { + let name = instance_name(d, key); + let files = staged.iter().any(|p| p.name == name); + let entry = match settled { + Some(r) => settled_entry(r, files), + None => settled_entry( + &TaskResult::blocked(&BlockedReason::DependencyDidNotPass), + false, + ), + }; + inputs.insert(d.clone(), Value::Object(entry)); + } + } + } + if !contributed { + let reason = BlockedReason::DependencyDidNotPass; + return Element::Settle(TaskResult::blocked(&reason), reason.event()); + } + let mut narrowed: Vec = node.keyed.clone(); + if let TaskKind::Route { + questions, + decider: Decider::Output { task: source }, + .. + } = &node.task + && !aligned_dependency(plan, node, source) + { + narrowed.extend(questions.keys().map(|question| crate::plan::ir::OutputRef { + task: source.clone(), + field: crate::plan::ir::OutputField(question.to_string()), + })); + } + for reference in &narrowed { + if let Err(note) = narrow(&mut inputs, reference, node.join, key) { + return Element::Settle( + TaskResult::undispatched(TaskStatus::Fail, note), + TaskEvent::ElementInputInvalid, + ); + } + } + inputs.insert( + TaskName(ITEM_INPUT.to_string()), + Value::String(key.to_owned()), + ); + Element::Dispatch(inputs) +} + +/// Replace an object field keyed by element with its entry under `key`. +fn narrow( + inputs: &mut BTreeMap, + reference: &crate::plan::ir::OutputRef, + join: Join, + key: &str, +) -> Result<(), String> { + let slot = field_slot(inputs, reference, join) + .ok_or_else(|| format!("keyed {reference} is absent"))?; + let entry = match &*slot { + Value::Object(entries) => entries + .get(key) + .cloned() + .ok_or_else(|| format!("keyed {reference} has no entry for {key}"))?, + other => { + return Err(format!( + "keyed {reference} is a {}, not an object keyed by element", + json_kind(other) + )); + } + }; + *slot = entry; + Ok(()) +} + +/// Where a keyed field's value sits in a dispatch's inputs: in the producer's output, which a +/// `settled` join wraps in its entry. +fn field_slot<'a>( + inputs: &'a mut BTreeMap, + reference: &crate::plan::ir::OutputRef, + join: Join, +) -> Option<&'a mut Value> { + let entry = inputs.get_mut(&reference.task)?; + let output = match join { + Join::Settled => entry.get_mut("output")?, + Join::All | Join::Passed => entry, + }; + output.get_mut(reference.field.0.as_str()) +} + +fn json_kind(value: &Value) -> &'static str { + match value { + Value::Null => "null", + Value::Bool(_) => "boolean", + Value::Number(_) => "number", + Value::String(_) => "string", + Value::Array(_) => "list", + Value::Object(_) => "object", + } +} + /// The deterministic decider: read one declared label per question from `source`'s output. fn decide_from_output( inputs: &BTreeMap, @@ -2873,6 +3149,7 @@ mod tests { emits_files: Vec::new(), over: None, max_fanout: None, + keyed: Vec::new(), when: None, revise: None, timeout: None, @@ -3647,6 +3924,7 @@ mod tests { emits_files: Vec::new(), over: None, max_fanout: None, + keyed: Vec::new(), when: None, revise: None, timeout: None, @@ -3710,6 +3988,7 @@ mod tests { emits_files: Vec::new(), over: None, max_fanout: None, + keyed: Vec::new(), when: None, revise: None, timeout: None, @@ -3914,6 +4193,7 @@ mod tests { emits_files: Vec::new(), over: None, max_fanout: None, + keyed: Vec::new(), when: None, revise: None, timeout: None, @@ -4538,6 +4818,7 @@ mod tests { decider: Decider::Output { task: "classify".into(), }, + files: Vec::new(), }, ..task("gate", &["classify"], "any", required) } @@ -9030,4 +9311,314 @@ mod tests { assert_eq!(named(name), None, "{name}"); } } + + fn over_targets(mut t: Task) -> Task { + t.over = Some(crate::plan::ir::OutputRef { + task: "discover".into(), + field: crate::plan::ir::OutputField("targets".to_string()), + }); + t.max_fanout = Some(8); + t + } + + fn keyed_on(mut t: Task, producer: &str, field: &str) -> Task { + t.keyed.push(crate::plan::ir::OutputRef { + task: producer.into(), + field: crate::plan::ir::OutputField(field.to_string()), + }); + t + } + + fn fix_question() -> Question { + question( + crucible_contract::decision::QuestionKind::Noul, + &["no", "uncertain"], + ) + } + + /// A route a model decides, mapped over discover's targets and reading analyze per element. + fn judge(required: bool) -> Task { + let mut judge = over_targets(task( + "judge", + &["discover", "analyze"], + "decision", + required, + )); + judge.task = TaskKind::Route { + questions: BTreeMap::from([(QuestionId::new("fix").unwrap(), fix_question())]), + decider: Decider::Model { + min_confidence: 0.5, + }, + files: Vec::new(), + }; + keyed_on(judge, "discover", "facts") + } + + fn fix_when(t: Task) -> Task { + let mut t = t; + t.when = Some(crate::plan::ir::When { + task: "judge".into(), + question: QuestionId::new("fix").unwrap(), + is: vec![label("yes")], + }); + t + } + + /// discover -> analyze[k] -> judge[k] (route) -> fix[k] when judge[k].fix = yes -> wrap + fn judged(judge_required: bool) -> ValidPlan { + valid( + vec![ + task("discover", &[], "any", true), + over_targets(task("analyze", &["discover"], "any", true)), + judge(judge_required), + fix_when(over_targets(task( + "fix", + &["discover", "judge", "analyze"], + "any", + true, + ))), + task("wrap", &["fix"], "any", true), + ], + 10.0, + ) + } + + fn decided(fix: &str) -> Value { + let yes = if fix == "yes" { 0.9 } else { 0.1 }; + serde_json::json!({"fix": {"label": fix, "confidence": 0.9, "probabilities": {"yes": yes, "no": 1.0 - yes}}}) + } + + fn judged_runner(answers: &[(&str, &str)], facts: Value) -> ScriptRunner { + let mut r = ScriptRunner::new(); + let targets: Vec<&str> = answers.iter().map(|(k, _)| *k).collect(); + r.outputs.insert( + "discover".into(), + serde_json::json!({"targets": targets, "facts": facts}), + ); + for (key, fix) in answers { + r.outputs.insert( + format!("analyze[{key}]"), + serde_json::json!({"claim": format!("claim-{key}")}), + ); + r.outputs.insert(format!("judge[{key}]"), decided(fix)); + } + r + } + + fn decision_substrate() -> Substrate { + Substrate { + caps: BTreeSet::from([crate::plan::ir::NEEDS_DECISION.to_owned()]), + } + } + + /// A mapped route decides each element on its own; the aligned fan-out runs only the elements + /// its answer allows, and an untaken element neither runs nor fails its node. + #[test] + fn a_mapped_route_gates_each_element_of_an_aligned_fanout() { + let mut r = judged_runner( + &[("a", "yes"), ("b", "no"), ("c", "yes")], + serde_json::json!({"a": {"lane": "x"}, "b": {"lane": "y"}, "c": {"lane": "z"}}), + ); + let out = execute( + &judged(true), + &decision_substrate(), + ExecCfg::default(), + &mut r, + |_, _| {}, + ); + assert!(out.valid, "{:?}", out.results); + let status = |n: &str| out.results[&TaskName(n.to_string())].status; + assert_eq!(status("fix[a]"), TaskStatus::Pass); + assert_eq!(status("fix[b]"), TaskStatus::NotTaken); + assert_eq!(status("fix[c]"), TaskStatus::Pass); + assert_eq!(status("fix"), TaskStatus::Pass); + assert_eq!( + out.results[&TaskName("fix".into())].fanout, + Some(FanoutSummary { + instances: 3, + passed: 2, + failed: 0, + not_taken: 1, + }) + ); + assert!( + !r.dispatched.iter().any(|(n, _)| n == "fix[b]"), + "an untaken element is never dispatched" + ); + assert_eq!( + out.results[&TaskName("fix[b]".into())].note.as_deref(), + Some("judge[b].fix resolved to no") + ); + + let seen = &r.seen_values["judge[b]"]; + assert_eq!( + seen[&TaskName("analyze".into())], + serde_json::json!({"claim": "claim-b"}) + ); + assert_eq!( + seen[&TaskName("discover".into())]["facts"], + serde_json::json!({"lane": "y"}), + "keyed narrows the object to the element's entry" + ); + assert_eq!(seen[&TaskName(ITEM_INPUT.into())], serde_json::json!("b")); + assert_eq!( + r.seen_values["fix[a]"][&TaskName("judge".into())], + decided("yes"), + "an aligned route reads as its own element's decision" + ); + } + + /// When every element is untaken, the node is untaken, and so is everything joining it with + /// `all`. + #[test] + fn a_fanout_with_every_element_untaken_is_untaken_to_its_leaves() { + let mut r = judged_runner( + &[("a", "no"), ("b", "no")], + serde_json::json!({"a": {}, "b": {}}), + ); + let out = execute( + &judged(true), + &decision_substrate(), + ExecCfg::default(), + &mut r, + |_, _| {}, + ); + assert!(out.valid, "{:?}", out.results); + assert_eq!( + out.results[&TaskName("fix".into())].status, + TaskStatus::NotTaken + ); + assert_eq!( + out.results[&TaskName("wrap".into())].status, + TaskStatus::NotTaken + ); + assert_eq!( + out.results[&TaskName("fix".into())] + .output + .as_ref() + .unwrap()["not_taken"], + 2 + ); + } + + /// A keyed object with no entry for an element fails that element alone, before any request; + /// the aligned work for it is blocked and the other elements go on. + #[test] + fn a_missing_keyed_entry_fails_only_its_element() { + let mut r = judged_runner(&[("a", "yes"), ("b", "yes")], serde_json::json!({"a": {}})); + let plan = valid( + vec![ + task("discover", &[], "any", true), + over_targets(task("analyze", &["discover"], "any", true)), + judge(false), + fix_when(over_targets(task( + "fix", + &["discover", "judge", "analyze"], + "any", + false, + ))), + joining(task("wrap", &["fix"], "any", true), Join::Settled), + ], + 10.0, + ); + let out = execute( + &plan, + &decision_substrate(), + ExecCfg::default(), + &mut r, + |_, _| {}, + ); + let result = |n: &str| &out.results[&TaskName(n.to_string())]; + assert_eq!(result("judge[a]").status, TaskStatus::Pass); + assert_eq!(result("judge[b]").status, TaskStatus::Fail); + assert_eq!( + result("judge[b]").note.as_deref(), + Some("keyed discover.facts has no entry for b") + ); + assert!(!r.dispatched.iter().any(|(n, _)| n == "judge[b]")); + assert_eq!(result("fix[a]").status, TaskStatus::Pass); + assert_eq!(result("fix[b]").status, TaskStatus::Blocked); + assert_eq!(result("fix").status, TaskStatus::Fail); + } + + /// An output-decided mapped route reads each element's answer from an unmapped source's + /// object keyed by element, and spends nothing. + #[test] + fn an_output_decided_mapped_route_reads_each_elements_answer() { + let mut gate = over_targets(task("gate", &["discover"], "any", true)); + gate.task = TaskKind::Route { + questions: BTreeMap::from([(QuestionId::new("fix").unwrap(), fix_question())]), + decider: Decider::Output { + task: "discover".into(), + }, + files: Vec::new(), + }; + let mut work = over_targets(task("work", &["discover", "gate"], "any", true)); + work.when = Some(crate::plan::ir::When { + task: "gate".into(), + question: QuestionId::new("fix").unwrap(), + is: vec![label("yes")], + }); + let plan = valid(vec![task("discover", &[], "any", true), gate, work], 10.0); + let mut r = ScriptRunner::new(); + r.outputs.insert( + "discover".into(), + serde_json::json!({"targets": ["a", "b"], "fix": {"a": "yes", "b": "no"}}), + ); + let out = execute( + &plan, + &any_substrate(), + ExecCfg::default(), + &mut r, + |_, _| {}, + ); + assert!(out.valid, "{:?}", out.results); + assert_eq!( + out.results[&TaskName("work[a]".into())].status, + TaskStatus::Pass + ); + assert_eq!( + out.results[&TaskName("work[b]".into())].status, + TaskStatus::NotTaken + ); + assert!(!r.dispatched.iter().any(|(n, _)| n.starts_with("gate"))); + assert_eq!(out.results[&TaskName("gate".into())].cost_usd, 0.0); + } + + fn output_routed() -> ValidPlan { + let mut gate = over_targets(task("gate", &["discover"], "any", true)); + gate.task = TaskKind::Route { + questions: BTreeMap::from([(QuestionId::new("fix").unwrap(), fix_question())]), + decider: Decider::Output { + task: "discover".into(), + }, + files: Vec::new(), + }; + let mut work = over_targets(task("work", &["discover", "gate"], "any", true)); + work.when = Some(crate::plan::ir::When { + task: "gate".into(), + question: QuestionId::new("fix").unwrap(), + is: vec![label("yes")], + }); + valid( + vec![ + task("discover", &[], "any", true), + gate, + work, + task("wrap", &["work"], "any", true), + ], + 10.0, + ) + } + + /// Each element's decision is its instance's row, so a resume asks nothing it already has. + #[test] + fn a_mapped_route_resumes_from_every_cut() { + assert_resumes_from_every_cut(&output_routed(), ExecCfg::default(), |r| { + r.outputs.insert( + "discover".into(), + serde_json::json!({"targets": ["a", "b", "c"], "fix": {"a": "yes", "b": "no", "c": "yes"}}), + ); + }); + } } diff --git a/crucible/src/plan/harness.rs b/crucible/src/plan/harness.rs index e99d1042..fd9554f6 100644 --- a/crucible/src/plan/harness.rs +++ b/crucible/src/plan/harness.rs @@ -83,6 +83,35 @@ fn captured_path(state: &Path, task: &str, declared: &str) -> PathBuf { captured_dir(state, task).join(declared) } +/// What a route's state carries under [`crate::plan::ir::FILES_INPUT`]: each listed file a +/// dependency declared and this route had staged, parsed, under the dependency's name. A mapped +/// dependency's instance is named by its node, since the route reads only its own element's. A +/// file that does not parse is `null`. +fn route_files(state: &Path, staged: &[StagedInput], files: &[String]) -> Value { + let mut by_producer: serde_json::Map = serde_json::Map::new(); + for input in staged.iter().filter(|s| files.contains(&s.declared)) { + let node = input + .producer + .split_once('[') + .map_or(input.producer.as_str(), |(node, _)| node); + let path = captured_path(state, &input.producer, &input.declared); + let Ok(text) = std::fs::read_to_string(&path) else { + continue; + }; + let parsed = serde_json::from_str(&text).unwrap_or_else(|error| { + tracing::warn!(file = %input.declared, producer = %input.producer, %error, "route file does not parse; the model reads null"); + Value::Null + }); + if let Value::Object(entries) = by_producer + .entry(node.to_owned()) + .or_insert_with(|| Value::Object(serde_json::Map::new())) + { + entries.insert(input.declared.clone(), parsed); + } + } + Value::Object(by_producer) +} + /// The mode the engine gives a captured file, regardless of what the source carried. A /// declared output derived from a staged input arrives 0444, and a capture that kept that mode /// would make the next run's copy fail with EACCES. @@ -227,6 +256,20 @@ impl TaskRunner for HarnessRunner { inputs: &BTreeMap, deadline: Option, ) -> Attempt { + let with_files; + let inputs = match &task.task { + TaskKind::Route { files, .. } if !files.is_empty() => { + let staged = self.staged.get(&task.name).map(Vec::as_slice); + let mut extended = inputs.clone(); + extended.insert( + TaskName(crate::plan::ir::FILES_INPUT.to_string()), + route_files(&self.paths.state, staged.unwrap_or_default(), files), + ); + with_files = extended; + &with_files + } + _ => inputs, + }; run_task( &Dispatch { args: &self.args, @@ -1362,6 +1405,7 @@ mod tests { emits_files: Vec::new(), over: None, max_fanout: None, + keyed: Vec::new(), when: None, revise: None, timeout: None, @@ -1522,6 +1566,7 @@ mod tests { emits_files: files.iter().map(|f| DeclaredFile::from(*f)).collect(), over: None, max_fanout: None, + keyed: Vec::new(), when: None, revise: None, timeout: None, @@ -3507,6 +3552,7 @@ workflow(type = "playbook", tasks = [analyze, implement, report]) emits_files: Vec::new(), over: None, max_fanout: None, + keyed: Vec::new(), when: None, revise: None, timeout: None, @@ -5268,4 +5314,44 @@ workflow(type = "playbook", tasks = [author, repro]) "a prompt without history is laid out as before: {prompt}" ); } + + /// A route reads each listed file its staged producers captured, parsed, under the + /// producer's node name; a file that does not parse is null, and an unlisted one is absent. + #[test] + fn a_route_reads_its_listed_files_by_dependency() { + let state = tempfile::tempdir().unwrap(); + let write = |task: &str, path: &str, body: &str| { + let dir = state.path().join("files").join(task); + std::fs::create_dir_all(&dir).unwrap(); + std::fs::write(dir.join(path), body).unwrap(); + }; + write( + "analyze[T-1]", + "RESULT.json", + r#"{"reachability": "unreachable"}"#, + ); + write("analyze[T-1]", "notes.txt", "not listed"); + write("triage", "RESULT.json", "{ torn"); + let staged = [ + StagedInput { + producer: "analyze[T-1]".into(), + declared: "RESULT.json".into(), + }, + StagedInput { + producer: "analyze[T-1]".into(), + declared: "notes.txt".into(), + }, + StagedInput { + producer: "triage".into(), + declared: "RESULT.json".into(), + }, + ]; + assert_eq!( + route_files(state.path(), &staged, &["RESULT.json".to_owned()]), + serde_json::json!({ + "analyze": {"RESULT.json": {"reachability": "unreachable"}}, + "triage": {"RESULT.json": null}, + }) + ); + } } diff --git a/crucible/src/plan/ir.rs b/crucible/src/plan/ir.rs index 418f5768..f84c708d 100644 --- a/crucible/src/plan/ir.rs +++ b/crucible/src/plan/ir.rs @@ -20,6 +20,8 @@ pub const ITEM_INPUT: &str = "item"; /// Reserved epilogue input key: `loop_graph` injects the kept candidate's /// context into every epilogue task's inputs under this name. pub const KEPT_INPUT: &str = "kept"; +/// The reserved key a route's state carries its dependencies' declared files under. +pub const FILES_INPUT: &str = "files"; /// The reserved input every epilogue task receives the main graph's outcome under. pub const OUTCOME_INPUT: &str = "outcome"; /// The reserved input every command and evaluate task receives the plan's bound parameter values @@ -179,6 +181,12 @@ impl<'de> Deserialize<'de> for Emits { } } +/// Two tasks mapped over the same producer field: their instances pair by element key, and an +/// edge between them is read per element. +pub fn aligned(a: &Task, b: &Task) -> bool { + a.over.is_some() && a.over == b.over +} + /// One declared output field of one task: what a mapped task fans out over. #[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)] pub struct OutputRef { @@ -345,6 +353,10 @@ pub enum TaskKind { Route { questions: BTreeMap, decider: Decider, + /// Declared JSON files of its dependencies the decision model reads under + /// [`FILES_INPUT`]. + #[serde(default, skip_serializing_if = "Vec::is_empty")] + files: Vec, }, /// A capability-owned engine operation. Engine { @@ -488,6 +500,9 @@ pub struct Task { /// whatever a global default happened to be. #[serde(default, skip_serializing_if = "Option::is_none")] pub max_fanout: Option, + /// Object fields of unmapped dependencies a mapped task narrows to its own element's entry. + #[serde(default, skip_serializing_if = "Vec::is_empty")] + pub keyed: Vec, #[serde(default, skip_serializing_if = "Option::is_none")] pub when: Option, /// Sends a failing verdict back to a dependency (see [`Revise`]). @@ -882,8 +897,75 @@ pub enum PlanError { }, #[error("{0}")] WhenNeverAnswered(Box), - #[error("route task {task:?} declares `over`; routing each element of a list is not supported")] - RouteWithOver { task: String }, + #[error( + "task {task:?}: when names the mapped route {route:?}, so it must map over {reference} too" + )] + WhenOnUnalignedRoute { + task: String, + route: String, + reference: String, + }, + #[error( + "task {task:?} declares keyed without over; only a mapped task has an element to key by" + )] + KeyedWithoutOver { task: String }, + #[error( + "task {task:?}: keyed {reference} names {producer:?}, which is not one of its dependencies" + )] + KeyedNotADependency { + task: String, + reference: String, + producer: String, + }, + #[error( + "task {task:?}: keyed {reference} names a mapped task; read a mapped dependency by mapping over the same list instead" + )] + KeyedFromMapped { task: String, reference: String }, + #[error("task {task:?}: keyed {reference} names a field {producer:?} does not declare")] + KeyedFieldOmitted { + task: String, + reference: String, + producer: String, + }, + #[error( + "task {task:?}: keyed {reference} is declared {declared}, not an object keyed by element" + )] + KeyedNotAnObject { + task: String, + reference: String, + declared: FieldType, + }, + #[error("route {task:?} depends on a task named {name:?}, which its state reserves")] + RouteReservedDependency { task: String, name: String }, + #[error( + "route {task:?} reads its answers from {source_task:?} and asks no model, so it takes no files" + )] + RouteFilesWithoutModel { task: String, source_task: String }, + #[error( + "route {task:?} lists file {path:?}, which none of its dependencies declares with a schema" + )] + RouteFileUndeclared { task: String, path: String }, + #[error( + "route {task:?} lists file {path:?}, which only {producer:?} declares, and it is mapped over another list" + )] + RouteFileFromUnaligned { + task: String, + path: String, + producer: String, + }, + #[error( + "mapped route {task:?} reads its answers from {source_task:?}, which is mapped over another list" + )] + RouteSourceUnaligned { task: String, source_task: String }, + #[error( + "mapped route {task:?} reads question {question:?} from {source_task:?}, which declares it {declared}; an unmapped source answers each element under an object keyed by element" + )] + RouteSourceNotKeyed { + task: String, + source_task: String, + question: String, + declared: FieldType, + }, #[error("task {task:?}: when names {route:?}, which is not one of its dependencies")] WhenNotADependency { task: String, route: String }, #[error("task {task:?}: when names {route:?}, which is a {kind} task, not a route")] @@ -1065,6 +1147,7 @@ pub(crate) fn typed_answers<'a>( let TaskKind::Route { questions, decider: Decider::Output { task: source }, + .. } = &route.task else { return None; @@ -1324,7 +1407,22 @@ impl Plan { } } } - if let TaskKind::Route { questions, decider } = &t.task { + if let TaskKind::Route { + questions, + decider, + files, + } = &t.task + { + if let Some(name) = t + .depends_on + .iter() + .find(|d| d.0 == ITEM_INPUT || d.0 == FILES_INPUT) + { + return Err(PlanError::RouteReservedDependency { + task: task(), + name: name.0.clone(), + }); + } if questions.is_empty() { return Err(PlanError::RouteWithoutQuestions { task: task() }); } @@ -1359,33 +1457,131 @@ impl Plan { source_task: source.0.clone(), }); } - let emits = &self.tasks[index[source]].emits; - for (id, question) in questions { - match emits.field(id.as_str()) { - Declared::Unchecked | Declared::Untyped => {} - Declared::Omitted => { - return Err(PlanError::RouteSourceOmitsQuestion { - task: task(), - source_task: source.0.clone(), - question: id.to_string(), - }); + if !files.is_empty() { + return Err(PlanError::RouteFilesWithoutModel { + task: task(), + source_task: source.0.clone(), + }); + } + let source_task = &self.tasks[index[source]]; + let emits = &source_task.emits; + if t.over.is_some() && !aligned(t, source_task) { + if source_task.over.is_some() { + return Err(PlanError::RouteSourceUnaligned { + task: task(), + source_task: source.0.clone(), + }); + } + for id in questions.keys() { + match emits.field(id.as_str()) { + Declared::Unchecked | Declared::Untyped => {} + Declared::Typed(declared) if declared.is_object() => {} + Declared::Omitted => { + return Err(PlanError::RouteSourceOmitsQuestion { + task: task(), + source_task: source.0.clone(), + question: id.to_string(), + }); + } + Declared::Typed(declared) => { + return Err(PlanError::RouteSourceNotKeyed { + task: task(), + source_task: source.0.clone(), + question: id.to_string(), + declared: declared.clone(), + }); + } } - Declared::Typed(declared) if answers(question, declared) => {} - Declared::Typed(declared) => { - return Err(PlanError::RouteSourceUnanswerable { - task: task(), - source_task: source.0.clone(), - question: id.to_string(), - declared: Box::new(declared.clone()), - expected: answerable_types(question), - }); + } + } else { + for (id, question) in questions { + match emits.field(id.as_str()) { + Declared::Unchecked | Declared::Untyped => {} + Declared::Omitted => { + return Err(PlanError::RouteSourceOmitsQuestion { + task: task(), + source_task: source.0.clone(), + question: id.to_string(), + }); + } + Declared::Typed(declared) if answers(question, declared) => {} + Declared::Typed(declared) => { + return Err(PlanError::RouteSourceUnanswerable { + task: task(), + source_task: source.0.clone(), + question: id.to_string(), + declared: Box::new(declared.clone()), + expected: answerable_types(question), + }); + } } } } } } - if t.over.is_some() { - return Err(PlanError::RouteWithOver { task: task() }); + for path in files { + let declaring: Vec<&Task> = t + .depends_on + .iter() + .map(|d| &self.tasks[index[d]]) + .filter(|d| { + d.emits_files + .iter() + .any(|f| f.path == *path && f.schema.is_some()) + }) + .collect(); + if declaring.is_empty() { + return Err(PlanError::RouteFileUndeclared { + task: task(), + path: path.clone(), + }); + } + if let Some(only) = declaring.first() + && declaring.iter().all(|d| d.over.is_some() && !aligned(t, d)) + { + return Err(PlanError::RouteFileFromUnaligned { + task: task(), + path: path.clone(), + producer: only.name.0.clone(), + }); + } + } + } + if !t.keyed.is_empty() && t.over.is_none() { + return Err(PlanError::KeyedWithoutOver { task: task() }); + } + for reference in &t.keyed { + if !t.depends_on.contains(&reference.task) { + return Err(PlanError::KeyedNotADependency { + task: task(), + reference: reference.to_string(), + producer: reference.task.0.clone(), + }); + } + let producer = &self.tasks[index[&reference.task]]; + if producer.over.is_some() { + return Err(PlanError::KeyedFromMapped { + task: task(), + reference: reference.to_string(), + }); + } + match producer.emits.field(&reference.field.0) { + Declared::Unchecked | Declared::Untyped => {} + Declared::Typed(declared) if declared.is_object() => {} + Declared::Omitted => { + return Err(PlanError::KeyedFieldOmitted { + task: task(), + reference: reference.to_string(), + producer: reference.task.0.clone(), + }); + } + Declared::Typed(declared) => { + return Err(PlanError::KeyedNotAnObject { + task: task(), + reference: reference.to_string(), + declared: declared.clone(), + }); + } } } if let Some(when) = &t.when { @@ -1405,6 +1601,15 @@ impl Plan { kind: target.task.label(), }); }; + if let Some(reference) = &target.over + && !aligned(t, target) + { + return Err(PlanError::WhenOnUnalignedRoute { + task: task(), + route: route(), + reference: reference.to_string(), + }); + } let Some(asked) = questions.get(&when.question) else { return Err(PlanError::WhenUnknownQuestion { task: task(), @@ -1900,6 +2105,7 @@ mod tests { emits_files: Vec::new(), over: None, max_fanout: None, + keyed: Vec::new(), when: None, revise: None, timeout: None, @@ -1950,6 +2156,7 @@ mod tests { decider: Decider::Model { min_confidence: 0.8, }, + files: Vec::new(), }, needs: NEEDS_DECISION.into(), ..agent(name, deps) @@ -1963,6 +2170,7 @@ mod tests { decider: Decider::Output { task: source.into(), }, + files: Vec::new(), }, ..agent(name, &[source]) } @@ -2014,7 +2222,10 @@ mod tests { let from_json = Plan::from_json_str(&serde_json::to_string(&original).unwrap()).unwrap(); for back in [from_toml, from_json] { assert_eq!(back.tasks[2].when, original.tasks[2].when); - let TaskKind::Route { questions, decider } = &back.tasks[1].task else { + let TaskKind::Route { + questions, decider, .. + } = &back.tasks[1].task + else { panic!("gate is not a route"); }; assert_eq!(questions[&qid("area")], area_question(&[])); @@ -2079,6 +2290,7 @@ mod tests { decider: Decider::Model { min_confidence: 0.8, }, + files: Vec::new(), }; assert_eq!( plan(vec![gate]).validate().unwrap_err(), @@ -2162,19 +2374,7 @@ mod tests { } #[test] - fn a_route_rejects_over_and_emits() { - let mut gate = model_route("gate", &["scan"], &[]); - gate.over = Some(OutputRef { - task: "scan".into(), - field: OutputField("issues".into()), - }); - gate.max_fanout = Some(4); - assert_eq!( - plan(vec![agent("scan", &[]), gate]).validate().unwrap_err(), - PlanError::RouteWithOver { - task: "gate".into() - } - ); + fn a_route_rejects_emits() { let mut gate = model_route("gate", &[], &[]); gate.emits = fields(&["area"]); assert_eq!( @@ -2540,6 +2740,7 @@ mod tests { emits_files: Vec::new(), over: None, max_fanout: None, + keyed: Vec::new(), when: None, revise: None, timeout: None, @@ -2691,6 +2892,7 @@ mod tests { emits_files: Vec::new(), over: None, max_fanout: None, + keyed: Vec::new(), when: None, revise: None, timeout: None, @@ -3080,6 +3282,7 @@ emits = ["lines"] decider: Decider::Output { task: source.into(), }, + files: Vec::new(), }, ..agent(name, &[source]) } @@ -4288,4 +4491,315 @@ emits = ["lines"] ); } } + + fn mapped(mut t: Task, producer: &str, field: &str) -> Task { + t.over = Some(OutputRef { + task: producer.into(), + field: OutputField(field.into()), + }); + t.max_fanout = Some(8); + t + } + + fn keyed(mut t: Task, producer: &str, field: &str) -> Task { + t.keyed.push(OutputRef { + task: producer.into(), + field: OutputField(field.into()), + }); + t + } + + fn scan() -> Task { + Task { + emits: typed(&[ + ("items", FieldType::List), + ("facts", FieldType::Object), + ("count", FieldType::Integer), + ]), + ..agent("scan", &[]) + } + } + + fn on_area(mut t: Task, route: &str) -> Task { + t.when = Some(When { + task: route.into(), + question: qid("area"), + is: vec![label("scheduler"), label("frontend"), label("uncertain")], + }); + t + } + + /// A route may map over a list, and a task mapped over the same list may branch on it; one + /// that is not mapped, or maps over another list, may not. + #[test] + fn a_when_on_a_mapped_route_must_come_from_an_aligned_task() { + let gate = mapped(model_route("gate", &["scan"], &[]), "scan", "items"); + let work = on_area( + mapped(agent("work", &["scan", "gate"]), "scan", "items"), + "gate", + ); + plan(vec![scan(), gate.clone(), work]).validate().unwrap(); + + let unmapped = on_area(agent("work", &["scan", "gate"]), "gate"); + assert_eq!( + plan(vec![scan(), gate.clone(), unmapped]) + .validate() + .unwrap_err(), + PlanError::WhenOnUnalignedRoute { + task: "work".into(), + route: "gate".into(), + reference: "scan.items".into(), + } + ); + let other = Task { + emits: typed(&[("items", FieldType::List)]), + ..agent("other", &[]) + }; + let elsewhere = on_area( + mapped(agent("work", &["scan", "other", "gate"]), "other", "items"), + "gate", + ); + assert!(matches!( + plan(vec![scan(), other, gate, elsewhere]) + .validate() + .unwrap_err(), + PlanError::WhenOnUnalignedRoute { .. } + )); + } + + /// keyed narrows an object field of an unmapped dependency, and only on a mapped task. + #[test] + fn keyed_names_an_object_field_of_an_unmapped_dependency() { + let work = |t: Task| keyed(mapped(t, "scan", "items"), "scan", "facts"); + plan(vec![scan(), work(agent("work", &["scan"]))]) + .validate() + .unwrap(); + + let cases = [ + ( + keyed(agent("work", &["scan"]), "scan", "facts"), + PlanError::KeyedWithoutOver { + task: "work".into(), + }, + ), + ( + keyed( + mapped(agent("work", &["scan"]), "scan", "items"), + "lost", + "facts", + ), + PlanError::KeyedNotADependency { + task: "work".into(), + reference: "lost.facts".into(), + producer: "lost".into(), + }, + ), + ( + keyed( + mapped(agent("work", &["scan"]), "scan", "items"), + "scan", + "nope", + ), + PlanError::KeyedFieldOmitted { + task: "work".into(), + reference: "scan.nope".into(), + producer: "scan".into(), + }, + ), + ( + keyed( + mapped(agent("work", &["scan"]), "scan", "items"), + "scan", + "count", + ), + PlanError::KeyedNotAnObject { + task: "work".into(), + reference: "scan.count".into(), + declared: FieldType::Integer, + }, + ), + ]; + for (task, error) in cases { + assert_eq!(plan(vec![scan(), task]).validate().unwrap_err(), error); + } + let per = mapped(agent("per", &["scan"]), "scan", "items"); + let from_mapped = keyed( + mapped(agent("work", &["scan", "per"]), "scan", "items"), + "per", + "x", + ); + assert_eq!( + plan(vec![scan(), per, from_mapped]).validate().unwrap_err(), + PlanError::KeyedFromMapped { + task: "work".into(), + reference: "per.x".into(), + } + ); + } + + fn with_files(mut t: Task, paths: &[&str]) -> Task { + if let TaskKind::Route { files, .. } = &mut t.task { + *files = paths.iter().map(|p| (*p).to_owned()).collect(); + } + t + } + + fn producing(name: &str, deps: &[&str], path: &str, schema: bool) -> Task { + Task { + emits_files: vec![DeclaredFile { + path: path.into(), + schema: schema.then(|| { + crucible_contract::emits::JsonSchema::parse(r#"{"type":"object"}"#).unwrap() + }), + }], + ..agent(name, deps) + } + } + + /// A route reads only files a dependency declares with a schema, from an aligned task when the + /// producer is mapped; an output-decided route reads none. + #[test] + fn a_routes_files_are_schema_declared_by_a_dependency_it_can_read() { + let analyze = mapped( + producing("analyze", &["scan"], "RESULT.json", true), + "scan", + "items", + ); + let judge = with_files( + mapped( + model_route("judge", &["scan", "analyze"], &[]), + "scan", + "items", + ), + &["RESULT.json"], + ); + plan(vec![scan(), analyze.clone(), judge]) + .validate() + .unwrap(); + + let undeclared = with_files( + mapped( + model_route("judge", &["scan", "analyze"], &[]), + "scan", + "items", + ), + &["OTHER.json"], + ); + assert_eq!( + plan(vec![scan(), analyze.clone(), undeclared]) + .validate() + .unwrap_err(), + PlanError::RouteFileUndeclared { + task: "judge".into(), + path: "OTHER.json".into(), + } + ); + let unschema = producing("note", &["scan"], "RESULT.json", false); + let reads_unschema = with_files( + model_route("judge", &["scan", "note"], &[]), + &["RESULT.json"], + ); + assert!(matches!( + plan(vec![scan(), unschema, reads_unschema]) + .validate() + .unwrap_err(), + PlanError::RouteFileUndeclared { .. } + )); + let unaligned = with_files( + model_route("judge", &["scan", "analyze"], &[]), + &["RESULT.json"], + ); + assert_eq!( + plan(vec![scan(), analyze, unaligned]) + .validate() + .unwrap_err(), + PlanError::RouteFileFromUnaligned { + task: "judge".into(), + path: "RESULT.json".into(), + producer: "analyze".into(), + } + ); + let source = Task { + emits: fields(&["area"]), + ..producing("classify", &[], "RESULT.json", true) + }; + let decided = with_files(output_route("gate", "classify", &[]), &["RESULT.json"]); + assert_eq!( + plan(vec![source, decided]).validate().unwrap_err(), + PlanError::RouteFilesWithoutModel { + task: "gate".into(), + source_task: "classify".into(), + } + ); + } + + /// A mapped route decided by an unmapped dependency reads each element's answer from an + /// object keyed by element; a source typed otherwise, or mapped over another list, is refused. + #[test] + fn a_mapped_output_route_reads_a_keyed_source() { + let source = |area: FieldType| Task { + emits: typed(&[("items", FieldType::List), ("area", area)]), + ..agent("scan", &[]) + }; + let gate = || mapped(output_route("gate", "scan", &[]), "scan", "items"); + plan(vec![source(FieldType::Object), gate()]) + .validate() + .unwrap(); + assert!(matches!( + plan(vec![source(FieldType::String), gate()]) + .validate() + .unwrap_err(), + PlanError::RouteSourceNotKeyed { .. } + )); + let other = Task { + emits: fields(&["area"]), + ..mapped(agent("per", &["scan"]), "scan", "items") + }; + let mut gate_over_per = mapped(output_route("gate", "per", &[]), "scan", "items"); + gate_over_per.depends_on.push("scan".into()); + gate_over_per.over = Some(OutputRef { + task: "scan".into(), + field: OutputField("items".into()), + }); + plan(vec![ + source(FieldType::Object), + other.clone(), + gate_over_per, + ]) + .validate() + .unwrap(); + let mut misaligned = output_route("gate", "per", &[]); + misaligned.depends_on.push("scan".into()); + misaligned = mapped(misaligned, "scan", "count"); + let scan_with_count = Task { + emits: typed(&[ + ("items", FieldType::List), + ("count", FieldType::List), + ("area", FieldType::Object), + ]), + ..agent("scan", &[]) + }; + assert!(matches!( + plan(vec![scan_with_count, other, misaligned]) + .validate() + .unwrap_err(), + PlanError::RouteSourceUnaligned { .. } + )); + } + + /// A route reads its dependencies' files under `files` and its own item under `item`, so it + /// may not depend on tasks by those names. + #[test] + fn a_route_may_not_depend_on_a_task_named_like_its_state_keys() { + let files = agent("files", &[]); + assert_eq!( + plan(vec![files, model_route("gate", &["files"], &[])]) + .validate() + .unwrap_err(), + PlanError::RouteReservedDependency { + task: "gate".into(), + name: "files".into(), + } + ); + } } diff --git a/crucible/src/plan/machine.rs b/crucible/src/plan/machine.rs index 3cef7ccb..e020aced 100644 --- a/crucible/src/plan/machine.rs +++ b/crucible/src/plan/machine.rs @@ -72,6 +72,10 @@ pub enum TaskEvent { FanoutItemsInvalid, InstancesPassed, InstancesFailed, + /// Every instance of a fan-out node settled not taken. + InstancesNotTaken, + /// An instance's keyed input was absent or not an object keyed by element. + ElementInputInvalid, /// A reviewer and the task it revises entered their rounds. RoundsStarted, RoundsPassed, @@ -106,6 +110,8 @@ pub const TASK_TRANSITIONS: &[(TaskState, TaskEvent, TaskState)] = { (S::Running, E::TransportCutByBudget, S::Transport), (S::Fanout, E::InstancesPassed, S::Pass), (S::Fanout, E::InstancesFailed, S::Fail), + (S::Fanout, E::InstancesNotTaken, S::NotTaken), + (S::Pending, E::ElementInputInvalid, S::Fail), (S::Pending, E::RoundsStarted, S::Revising), (S::Revising, E::RoundsPassed, S::Pass), (S::Revising, E::RoundsFailed, S::Fail), diff --git a/crucible/src/plan/runner.rs b/crucible/src/plan/runner.rs index 528dea83..ca696095 100644 --- a/crucible/src/plan/runner.rs +++ b/crucible/src/plan/runner.rs @@ -85,6 +85,7 @@ impl ShellRunner { ) -> Attempt { let mut cmd = Command::new("sh"); cmd.arg("-c").current_dir(&self.workdir); + crate::inference::withhold_from_task(&mut cmd); cmd.env(crate::plan::TASK_NAME_ENV, &task.name.0); let mut env_inputs = inputs.clone(); let history = match task.task { @@ -193,6 +194,7 @@ impl ShellRunner { TaskKind::Route { questions, decider: Decider::Model { min_confidence }, + .. } => { let inference = match crate::inference::from_process_env() { Ok(inference) => inference, @@ -505,6 +507,7 @@ mod tests { emits_files: Vec::new(), over: None, max_fanout: None, + keyed: Vec::new(), when: None, revise: None, timeout: None, @@ -530,6 +533,7 @@ mod tests { emits_files: Vec::new(), over: None, max_fanout: None, + keyed: Vec::new(), when: None, revise: None, timeout: None, @@ -751,6 +755,7 @@ mod tests { emits_files: Vec::new(), over: None, max_fanout: None, + keyed: Vec::new(), when: None, revise: None, timeout: None, @@ -787,6 +792,7 @@ mod tests { emits_files: Vec::new(), over: None, max_fanout: None, + keyed: Vec::new(), when: None, revise: None, timeout: None, @@ -830,6 +836,7 @@ mod tests { emits_files: Vec::new(), over: None, max_fanout: None, + keyed: Vec::new(), when: None, revise: None, timeout: None, @@ -862,6 +869,7 @@ mod tests { emits_files: Vec::new(), over: None, max_fanout: None, + keyed: Vec::new(), when: None, revise: None, timeout: None, @@ -903,6 +911,7 @@ mod tests { emits_files: Vec::new(), over: None, max_fanout: None, + keyed: Vec::new(), when: None, revise: None, timeout: None, @@ -944,6 +953,7 @@ mod tests { emits_files: Vec::new(), over: None, max_fanout: None, + keyed: Vec::new(), when: None, revise: None, timeout: None, @@ -977,6 +987,7 @@ mod tests { emits_files: Vec::new(), over: None, max_fanout: None, + keyed: Vec::new(), when: None, revise: None, timeout: None, @@ -1020,6 +1031,7 @@ mod tests { emits_files: Vec::new(), over: None, max_fanout: None, + keyed: Vec::new(), when: None, revise: None, history: Some(4), @@ -1098,6 +1110,7 @@ mod tests { emits_files: Vec::new(), over: None, max_fanout: None, + keyed: Vec::new(), when: None, revise: None, timeout: None, @@ -1131,6 +1144,7 @@ mod tests { emits_files: Vec::new(), over: None, max_fanout: None, + keyed: Vec::new(), when: None, revise: None, timeout: None, diff --git a/crucible/src/plan/starlark.rs b/crucible/src/plan/starlark.rs index c293af88..45486ab2 100644 --- a/crucible/src/plan/starlark.rs +++ b/crucible/src/plan/starlark.rs @@ -576,6 +576,7 @@ fn known_kwargs(function: &str) -> &'static [&'static str] { "stage", "over", "max_fanout", + "keyed", "revise", "max_rounds", "timeout", @@ -604,6 +605,7 @@ fn known_kwargs(function: &str) -> &'static [&'static str] { "stage", "over", "max_fanout", + "keyed", "revise", "max_rounds", "timeout", @@ -626,6 +628,7 @@ fn known_kwargs(function: &str) -> &'static [&'static str] { "stage", "over", "max_fanout", + "keyed", "revise", "max_rounds", "timeout", @@ -649,6 +652,7 @@ fn known_kwargs(function: &str) -> &'static [&'static str] { "stage", "over", "max_fanout", + "keyed", "revise", "max_rounds", "timeout", @@ -664,10 +668,14 @@ fn known_kwargs(function: &str) -> &'static [&'static str] { "questions", "min_confidence", "source", + "files", "depends_on", "required", "join", "stage", + "over", + "max_fanout", + "keyed", "when", "answers", "otherwise", @@ -874,6 +882,7 @@ fn constructor( emits_files: Vec::new(), over: None, max_fanout: None, + keyed: Vec::new(), when: None, revise: None, timeout: None, @@ -915,6 +924,7 @@ fn constructor( emits_files: Vec::new(), over: None, max_fanout: None, + keyed: Vec::new(), revise: None, timeout: None, when: None, @@ -935,8 +945,13 @@ fn constructor( (None, Some(task)) => (Decider::Output { task }, "any"), _ => return Err(CompileError::RouteDecider { task: name.0 }), }; - Task { - task: TaskKind::Route { questions, decider }, + let files = take_route_files(&mut named)?; + let task = Task { + task: TaskKind::Route { + questions, + decider, + files, + }, depends_on: take_task_names(&mut named)?, session: None, needs: needs.to_owned(), @@ -946,15 +961,18 @@ fn constructor( stage: parse_stage(&take_string_default(&mut named, "stage", "iteration")?)?, emits: crate::plan::ir::Emits::default(), emits_files: Vec::new(), - over: None, - max_fanout: None, + over: take_over(&mut named)?, + max_fanout: take_optional_fanout(&mut named)?, + keyed: take_keyed(&mut named)?, revise: None, timeout: None, when: take_when(&mut named, state, &name)?, name, history: None, repair: 0, - } + }; + check_fanout(&task)?; + task } "propose" => { let session = take_session(&mut named, state, at)?; @@ -1201,6 +1219,7 @@ fn dsl_task( emits_files: take_emitted_files(named)?, over: take_over(named)?, max_fanout: take_optional_fanout(named)?, + keyed: take_keyed(named)?, when, revise: take_revise(named)?, timeout: take_timeout(named)?, @@ -1525,6 +1544,44 @@ fn take_over(named: &mut BTreeMap) -> Result> { } } +/// The object fields a mapped task narrows to its own element's entry, each written as +/// `producer.field`. +fn take_keyed(named: &mut BTreeMap) -> Result> { + let items = match named.remove("keyed") { + None | Some(Value::None) => return Ok(Vec::new()), + Some(Value::List(items)) => items, + Some(_) => return Err(CompileError::KeyedNotOutputFields), + }; + items + .into_iter() + .map(|item| match item { + Value::Output(output) => match output.ty { + Some(declared) if !declared.is_object() => Err(CompileError::KeyedNotAnObject { + reference: output.reference.to_string(), + declared, + }), + _ => Ok(output.reference), + }, + _ => Err(CompileError::KeyedNotOutputFields), + }) + .collect() +} + +/// The dependency files a route's decision model reads, as workspace-relative paths. +fn take_route_files(named: &mut BTreeMap) -> Result> { + match named.remove("files") { + None | Some(Value::None) => Ok(Vec::new()), + Some(Value::List(items)) => items + .into_iter() + .map(|item| match item { + Value::String(path) => Ok(path), + _ => Err(CompileError::RouteFilesNotPaths), + }) + .collect(), + Some(_) => Err(CompileError::RouteFilesNotPaths), + } +} + fn take_optional_fanout(named: &mut BTreeMap) -> Result> { match named.remove("max_fanout") { None | Some(Value::None) => Ok(None), @@ -1639,6 +1696,7 @@ fn engine(name: &str, op: EngineOp, source: Option, depends_on: Vec CompileError { CompileError::OverNotAList { .. } | CompileError::OverItemsNotStrings { .. } => { (None, "over") } + CompileError::KeyedNotOutputFields | CompileError::KeyedNotAnObject { .. } => { + (None, "keyed") + } + CompileError::RouteFilesNotPaths => (Some("route"), "files"), CompileError::EmitsNotList | CompileError::EmitsEntryNotString | CompileError::UnknownFieldType { .. } diff --git a/crucible/src/plan/starlark/reference.rs b/crucible/src/plan/starlark/reference.rs index db9acb9e..fdac18b7 100644 --- a/crucible/src/plan/starlark/reference.rs +++ b/crucible/src/plan/starlark/reference.rs @@ -132,6 +132,7 @@ fn task_knobs() -> Vec { "Instance cap for `over`, within the engine's ceiling of {MAX_FANOUT_CEILING}." ), ), + keyed_kwarg(), Kwarg::new( "revise", "task | list[task]", @@ -176,6 +177,15 @@ fn task_knobs() -> Vec { ] } +fn keyed_kwarg() -> Kwarg { + Kwarg::new( + "keyed", + "list[producer.field]", + "Object fields of unmapped dependencies, keyed by item, that each instance of a mapped \ + task reads narrowed to its own item's entry.", + ) +} + fn when_kwarg() -> Kwarg { Kwarg::new( "when", @@ -576,6 +586,28 @@ pub fn functions() -> Vec { "Which dependency outputs form the state, as on any task.", ), Kwarg::new("stage", "\"iteration\" | \"epilogue\"", "As on any task."), + Kwarg::new( + "files", + "list[str]", + "Declared JSON files of its dependencies the decision model reads, under \ + `files` in its state, by dependency. Each path must be declared with a \ + schema. A model decider only.", + ), + Kwarg::new( + "over", + "producer.field", + "Decide once per item of a dependency's emitted list. A dependency mapped \ + over the same list is read per item, and so is a `when` on this route in a \ + task mapped over it.", + ), + Kwarg::new( + "max_fanout", + "int", + format!( + "Item cap for `over`, within the engine's ceiling of {MAX_FANOUT_CEILING}." + ), + ), + keyed_kwarg(), when_kwarg(), answers_kwarg(), otherwise_kwarg(), diff --git a/crucible/src/plan/template.rs b/crucible/src/plan/template.rs index 6a70ce21..ab97cb64 100644 --- a/crucible/src/plan/template.rs +++ b/crucible/src/plan/template.rs @@ -48,6 +48,7 @@ pub(crate) fn iteration_template( emits_files: Vec::new(), over: None, max_fanout: None, + keyed: Vec::new(), when: None, revise: None, timeout: None, diff --git a/crucible/src/runloop/graph.rs b/crucible/src/runloop/graph.rs index e85905a8..7cfed26a 100644 --- a/crucible/src/runloop/graph.rs +++ b/crucible/src/runloop/graph.rs @@ -1026,6 +1026,7 @@ fn wide_template(cfg: &WideConfig, prep: &Prepared, direction: Direction) -> Res emits_files: Vec::new(), over: None, max_fanout: None, + keyed: Vec::new(), when: None, revise: None, timeout: None, @@ -1052,6 +1053,7 @@ fn wide_template(cfg: &WideConfig, prep: &Prepared, direction: Direction) -> Res emits_files: Vec::new(), over: None, max_fanout: None, + keyed: Vec::new(), when: None, revise: None, timeout: None, @@ -1078,6 +1080,7 @@ fn wide_template(cfg: &WideConfig, prep: &Prepared, direction: Direction) -> Res emits_files: Vec::new(), over: None, max_fanout: None, + keyed: Vec::new(), when: None, revise: None, timeout: None, diff --git a/crucible/tests/contract_version.rs b/crucible/tests/contract_version.rs index 376d367f..e9c6675c 100644 --- a/crucible/tests/contract_version.rs +++ b/crucible/tests/contract_version.rs @@ -43,7 +43,7 @@ fn wire_versions_are_pinned_to_contract_version() { ); assert_eq!( (CONTRACT_VERSION, wire), - ("1.15.0", (1, 1, 1, "v2")), + ("1.16.0", (1, 1, 1, "v2")), "a wire version changed without bumping CONTRACT_VERSION" ); } diff --git a/crucible/tests/route_e2e.rs b/crucible/tests/route_e2e.rs index 90729dfb..11269a60 100644 --- a/crucible/tests/route_e2e.rs +++ b/crucible/tests/route_e2e.rs @@ -296,6 +296,94 @@ fn an_openai_decisions_binding_routes_the_same_plan_the_same_way() { let _ = std::fs::remove_dir_all(&dir); } +const MAPPED_PLAN: &str = r#" +version = 1 +[budget] +usd = 5.0 + +[[task]] +name = "scan" +kind = "command" +command = "echo '{\"items\": [\"a\", \"b\"], \"facts\": {\"a\": {\"lane\": \"needs_reach\"}, \"b\": {\"lane\": \"review\"}}}'" +emits = ["items", "facts"] + +[[task]] +name = "analyze" +kind = "command" +depends_on = ["scan"] +over = { task = "scan", field = "items" } +max_fanout = 4 +command = "echo '{}'" + +[[task]] +name = "judge" +kind = "route" +needs = "decision" +depends_on = ["scan", "analyze"] +over = { task = "scan", field = "items" } +max_fanout = 4 +keyed = [{ task = "scan", field = "facts" }] +decider = { kind = "model", min_confidence = 0.5 } +[task.questions.urgent] +instructions = "Does the evidence support the claim?" +type = "noul" +drop = ["no", "uncertain"] + +[[task]] +name = "fix" +kind = "command" +depends_on = ["scan", "judge"] +over = { task = "scan", field = "items" } +max_fanout = 4 +when = { task = "judge", question = "urgent", is = ["yes"] } +command = "if [ -n \"$ROUTE_E2E_KEY$CRUCIBLE_INFERENCE\" ]; then echo '{\"saw\": true}'; else echo '{\"saw\": false}'; fi" +"#; + +/// A route mapped over a list asks the model once per item, with that item and its keyed entry +/// in its state, and the work mapped over the same list runs per item on its answer. No +/// task process sees the decision binding or its key. +#[test] +fn a_mapped_route_asks_once_per_item_with_that_items_state() { + let (url, seen) = serve_at( + "/v1/decisions", + "200 OK", + r#"{"answers":[{"type":"predicate","name":"urgent","probability":0.97}]}"#.to_string(), + ); + let dir = workdir("mapped", MAPPED_PLAN); + let document = serde_json::json!({"version": 1, "bindings": [{ + "role": "decision", "protocol": "decisions", "url": url, "model": "gpt-6-luna", + "key_env": "ROUTE_E2E_KEY", + }]}) + .to_string(); + let run = run_with(&dir, Some(&document), Some("sk-test")); + assert!(run.ok, "{}\n{}", run.stdout, run.stderr); + assert_eq!(status_of(&run.stdout, "judge"), "pass"); + assert_eq!(status_of(&run.stdout, "fix"), "pass"); + assert_eq!( + run.stdout.matches("{\"saw\":false}").count(), + 2, + "every item ran, and no task process saw the decision binding: {}", + run.stdout + ); + + let requests = seen.lock().unwrap(); + assert_eq!(requests.len(), 2, "one decision per item"); + let mut states: Vec = requests + .iter() + .map(|request| { + let body: serde_json::Value = + serde_json::from_str(request.rsplit('\n').next().unwrap()).unwrap(); + serde_json::from_str(body["input"].as_str().unwrap()).unwrap() + }) + .collect(); + states.sort_by_key(|state| state["item"].as_str().map(str::to_owned)); + for (state, (item, lane)) in states.iter().zip([("a", "needs_reach"), ("b", "review")]) { + assert_eq!(state["item"], item); + assert_eq!(state["scan"]["facts"], serde_json::json!({"lane": lane})); + } + let _ = std::fs::remove_dir_all(&dir); +} + #[test] fn an_answer_below_min_confidence_takes_the_uncertain_branch() { let (url, _seen) = serve("200 OK", answers(0.55, 0.97)); diff --git a/docs/adr/ADR-0062.md b/docs/adr/ADR-0062.md new file mode 100644 index 00000000..9d339de0 --- /dev/null +++ b/docs/adr/ADR-0062.md @@ -0,0 +1,94 @@ + + + +# ADR-0062: Per-element decisions are routes mapped over the list the gated work maps over + +> **Status:** proposed | **Date:** 2026-10-07 + +**References:** [RFC-0001:C-ROUTE](../rfc/RFC-0001.md#rfc-0001c-route), [RFC-0001:C-ROUTE-EACH](../rfc/RFC-0001.md#rfc-0001c-route-each), [RFC-0001:C-ROUTE-FILES](../rfc/RFC-0001.md#rfc-0001c-route-files), [RFC-0001:C-WORKFLOW](../rfc/RFC-0001.md#rfc-0001c-workflow), [RFC-0002:C-SETTLED-JOIN](../rfc/RFC-0002.md#rfc-0002c-settled-join), [RFC-0002:C-PLAYBOOK-RESUME](../rfc/RFC-0002.md#rfc-0002c-playbook-resume), [RFC-0002:C-REVISE-LOOP](../rfc/RFC-0002.md#rfc-0002c-revise-loop), [ADR-0064](../adr/ADR-0064.md) + +## Context + +A route ([RFC-0001:C-ROUTE](../rfc/RFC-0001.md#rfc-0001c-route)) asks a decision model typed questions once and gates downstream tasks with when. A route cannot declare over, so a playbook that fans agent turns out over a list cannot put a decision in front of each turn: the only decision available is one for the whole run. + +The motivating case is a CVE triage playbook: a command stages per-ticket facts and emits the ticket keys, an agent task maps over the keys with one turn per ticket, a command applies the verdicts and emits the keys worth fixing, and a second agent task maps over those. Most of the cost is the per-ticket turns, and the decisions that would cut it are per ticket: skip a turn when the staged facts already settle reachability, give hard tickets more effort than easy ones, gate a fix on whether the version bump keeps the API, judge whether an agent's result file is supported by its own evidence, and flag likely duplicates. + +What exists today: + +- A mapped task names each instance by its item, `node[key]`, and a resumed run keeps an instance's result by that key ([RFC-0001:C-WORKFLOW](../rfc/RFC-0001.md#rfc-0001c-workflow)). +- A mapped dependency reaches its dependents as one folded output; a fold passes only when every instance passed. +- A when is evaluated once per node, before the node expands. +- Not taken is a terminal status, and an untaken branch is untaken to its leaves through "all" joins. +- A route's state is its dependencies' JSON outputs. A task whose only output is a declared file gives a route nothing to read. +- A revise loop refuses a body that maps over a list, and its reviewer must be an agent, command, or evaluate task ([RFC-0002:C-REVISE-LOOP](../rfc/RFC-0002.md#rfc-0002c-revise-loop)). + +## Decision + +Let a route map over a list, and evaluate a when on a mapped route per element in a task that maps over the same list. Two tasks mapped over the same producer field are aligned; their instances pair by key, which is already the identity a resume relies on. The normative rules are [RFC-0001:C-ROUTE-EACH](../rfc/RFC-0001.md#rfc-0001c-route-each) and [RFC-0001:C-ROUTE-FILES](../rfc/RFC-0001.md#rfc-0001c-route-files). + +- **One decision per element.** A mapped route's instance for a key is that element's decision, recorded and folded on resume like any instance, so an element is asked once per run. +- **Per-element reads.** On every aligned edge, the dependency reaches an instance as its own instance for the same key, under the dependency's own name, instead of the fold. One rule covers routes, the tasks they gate, and plain per-element pipelines alike, and an untaken element stays untaken down the whole aligned chain. +- **Gating.** when keeps its meaning and gains a per-element reading. Partitioning by answer is several aligned tasks with whens on the same question, and the existing coverage rule makes every label lead somewhere. +- **State.** An element's state is the route's inputs plus its key. keyed narrows a per-element object field to the element's entry, for any mapped task, so a hundred-element fan-out does not hand every instance, model request or agent turn, the whole field, and one element's facts cannot sway another's answer. A route may also read declared JSON files, which is how it judges an agent whose output is a file. +- **Folds.** An instance not taken is not a failure. A fan-out whose instances were all untaken is itself untaken, so its "all" dependents are untaken as any branch's are and a "settled" reducer still runs. + +For the triage playbook, sketched (illustrative; the clauses above are the contract): + +```python +gate = route(name = "gate", depends_on = [prepare], over = prepare.members, max_fanout = 120, + keyed = [prepare.facts], min_confidence = 0.8, + questions = {"depth": choice(ask = "...", options = {"settled": "...", "quick": "...", "deep": "..."}, + drop = ["settled"])}) +quick = agent(name = "quick", effort = "low", depends_on = [prepare, gate], over = prepare.members, + max_fanout = 120, when = gate.depth.quick, ...) +deep = agent(name = "deep", effort = "high", depends_on = [prepare, gate], over = prepare.members, + max_fanout = 120, when = gate.depth.otherwise, ...) # deep and uncertain +judge = route(name = "judge", depends_on = [quick, deep], join = "settled", over = prepare.members, + max_fanout = 120, files = ["RESULT.json"], min_confidence = 0.8, + questions = {"supported": noul(ask = "Does the evidence support the verdict?", drop = ["yes"])}) +review = agent(name = "review", depends_on = [judge], over = prepare.members, max_fanout = 120, + when = judge.supported.otherwise, ...) # no and uncertain +``` + +A ticket answered "settled" costs one decision and no turn. A deterministic output-decided route can sit in front of the model-decided one when the staged facts already carry an answer; the model route then declares a when on it and is asked only about the rest. A fix gate is the same shape over the fix list. + +## Consequences + +- The expensive per-element turns become conditional per element, and an untaken element shows as a not-taken row under its own key rather than disappearing from the run. +- Every element costs one decision request. max_fanout bounds the requests per run, and the run's ceilings are checked before each one. Decision requests are not charged to the run's budget: they cost little next to the turns they gate, so the ceiling check before each request is in practice a wall-clock check, and max_fanout bounds the count. +- An aligned dependent waits for the whole mapped dependency to settle before it expands. Starting element k's turn as soon as element k's decision lands is left out; the decision is the cheap stage, so the added latency is small next to the turns it gates. +- An aligned mapped-to-mapped edge changes what the dependent instance reads: an instance, not the fold. This is breaking for a pack whose mapped task consumes the fold of another task mapped over the same list. No published example or known pack has such an edge; a task mapped over a different list, or not mapped, still reads the fold. +- The fan-out summary gains a not_taken count. Readers that predate it read zero, and a fold that would have failed only because instances were untaken now passes. +- Judging an agent's result feeds a review lane, not a send-back ([ADR-0064](../adr/ADR-0064.md)). +- A playbook that relied on every element getting a full turn now relies on the decision model's calibration and min_confidence. uncertain still has to be routed, so the author decides what happens when the model is unsure. + +## Alternatives Considered + +### Map a route over the list, read every aligned edge per element, and read a when on a mapped route per element (accepted) + +- **Pros:** Reuses the instance key, the not-taken status, the coverage rule, and resume folding as they are., Every element keeps a row in every gated task, so the record shows why a turn did not run., Partitioning by answer and chaining a deterministic route in front of a model route need no new construct. +- **Cons:** An aligned mapped-to-mapped edge reads an instance instead of the fold, a breaking change for any pack that relied on the fold there., Each element is a separate decision request. + +### Map gated tasks over a filtered list taken from the route's fold (over = gate.depth.quick) (rejected) + +- **Pros:** No per-element when; a gated task simply has fewer instances., Instance names stay keyed by item. +- **Cons:** An element the route sent elsewhere has no row in the gated task, so a settled reducer cannot tell untaken from never listed., The coverage check would have to be rebuilt for list references instead of when labels., over would read a list the engine synthesises from a route's record, a second meaning for a field reference. +- **Rejected because:** Loses the not-taken record and duplicates the coverage rule in a new place. + +### Quantified when on an unmapped route over the fold (any or all elements answered a label) (rejected) + +- **Pros:** Small change to when; useful for run-level gates such as skipping a report. +- **Cons:** Still one decision for the whole list, so it cannot gate element k's turn on element k's answer., Does not address the cost driver, which is per-element turns. +- **Rejected because:** Answers a different question; it can be added later on top of mapped routes if a run-level gate is needed. + +### One route request per run carrying every element, answered as a map keyed by element (rejected) + +- **Pros:** One request instead of N. +- **Cons:** Neither decision API answers a question per element; the questions would have to be multiplied per key at run time, which C-ROUTE forbids because questions are compile-time., One malformed answer fails every element, and a resume re-asks all of them. +- **Rejected because:** Breaks compile-time questions and per-element failure isolation. + +### Read aligned edges per element only along chains that involve a route (rejected) + +- **Pros:** Existing mapped-to-mapped edges keep reading the fold. +- **Cons:** Two read rules for the same shape of edge, chosen by what is upstream., Needs a transitive per-element marking so an untaken element stays untaken past the first hop. +- **Rejected because:** No known pack consumes the fold on an aligned edge, so the compatibility it buys is not worth a second rule. diff --git a/docs/adr/ADR-0064.md b/docs/adr/ADR-0064.md new file mode 100644 index 00000000..d766544a --- /dev/null +++ b/docs/adr/ADR-0064.md @@ -0,0 +1,56 @@ + + + +# ADR-0064: Only a route asks the decision model + +> **Status:** proposed | **Date:** 2026-10-07 + +**References:** [RFC-0001:C-ROUTE](../rfc/RFC-0001.md#rfc-0001c-route), [RFC-0001:C-INFERENCE](../rfc/RFC-0001.md#rfc-0001c-inference), [RFC-0001:C-ROUTE-EACH](../rfc/RFC-0001.md#rfc-0001c-route-each), [RFC-0001:C-ROUTE-FILES](../rfc/RFC-0001.md#rfc-0001c-route-files), [RFC-0002:C-PLAYBOOK-CAPS](../rfc/RFC-0002.md#rfc-0002c-playbook-caps), [RFC-0002:C-REVISE-LOOP](../rfc/RFC-0002.md#rfc-0002c-revise-loop), [ADR-0062](../adr/ADR-0062.md) + +## Context + +The decision binding is used by the engine's own process, and its credential must not leave the engine ([RFC-0001:C-INFERENCE](../rfc/RFC-0001.md#rfc-0001c-inference)). A route is its only consumer today. Two pressures push against that: + +- A command task in a triage playbook already holds the per-element facts and could ask the decision model itself, through a broker tool that keeps the credential engine-side, instead of the playbook declaring a route for each question. +- A route that judges an agent's output ("does this result file's evidence support its claim?") wants to feed its verdict back to the agent, as a reviewer in a revise loop does ([RFC-0002:C-REVISE-LOOP](../rfc/RFC-0002.md#rfc-0002c-revise-loop)), not only gate a later lane. + +A route's guarantees come from being a graph node: its questions and labels are compile-time and in the generated TOML, every label must lead somewhere, its answer and full distribution are the task's recorded output, and a resumed run never asks again. A tool call from inside a task has none of those. + +## Decision + +Keep every model decision a route. The engine does not relay the decision binding into any sandbox or task process and offers no tool that asks it ([RFC-0001:C-INFERENCE](../rfc/RFC-0001.md#rfc-0001c-inference)). Per-element decisions are mapped routes ([RFC-0001:C-ROUTE-EACH](../rfc/RFC-0001.md#rfc-0001c-route-each)), and a route judges an agent's output by reading its declared JSON files ([RFC-0001:C-ROUTE-FILES](../rfc/RFC-0001.md#rfc-0001c-route-files)) and gating a review lane with when. + +A judgment is not yet a send-back, and C-REVISE-LOOP does not cover it: a reviewer must be an agent, command, or evaluate task, and a body must not map over a list. Admitting a route reviewer needs a stated mapping from labels to a failing verdict, and admitting mapped bodies needs per-element rounds. Both are separate decisions, deferred until a playbook needs the judged conversation resumed rather than a second agent reviewing. + +## Consequences + +- Every model decision in a run is visible in the plan, covered by the label rule, recorded with its distribution, and folded on resume. +- A command that would like a quick model opinion must be split around a route, which costs a node and an edge per question set. +- Questions cannot be composed at run time. A playbook that needs a question per discovered category has to declare the categories up front, which is the constraint C-ROUTE already imposes. +- The credential boundary stays simple: no sandbox egress rule, no broker tool, and no per-task decision budget to police. +- A judged agent result that fails review goes to a separate review turn, which starts a new conversation and pays to re-read the work. Resuming the original agent's session with the judgment is the follow-up if that cost matters. + +## Alternatives Considered + +### Only route tasks ask the decision model (accepted) + +- **Pros:** Compile-time questions, label coverage, recorded distributions, and resume folding hold for every decision., No credential, endpoint, or tool reaches a task process. +- **Cons:** A task cannot ask an ad hoc question mid-work. + +### A broker tool that lets a command task ask the decision model, credential kept in the engine (rejected) + +- **Pros:** The task asks exactly when it has the facts, with no extra node., The credential still never leaves the engine. +- **Cons:** Questions and labels are authored at run time by the task, outside the plan and the coverage rule., The answer is whatever the task writes into its own output, not a recorded decision; a resume re-runs the task and asks again., Calls are unbounded per task; max_fanout and the run's ceilings do not see them., A sandboxed task would need egress or a broker channel for decision traffic. +- **Rejected because:** Gives up every guarantee that makes a route auditable, for flexibility the per-element route already provides. + +### Declared question sets a command task may ask by name through the broker (rejected) + +- **Pros:** Questions stay compile-time., A task chooses when to ask. +- **Cons:** The answer still lands in the task's own output rather than a decision record., Gating on it needs the task to emit it and an output-decided route to read it, which is a route again. +- **Rejected because:** Ends up as a route with extra plumbing. + +### Admit a route as a revise reviewer, failing on listed labels (rejected) + +- **Pros:** A judgment can resume the judged agent's conversation. +- **Cons:** Needs a label-to-verdict mapping and its own coverage rule., Useless for the per-element case until revise bodies may map over a list. +- **Rejected because:** Deferred: needs a label-to-verdict mapping and per-element revise rounds, each a separate decision. diff --git a/docs/crucible-contract.md b/docs/crucible-contract.md index 77036b8d..924cb09e 100644 --- a/docs/crucible-contract.md +++ b/docs/crucible-contract.md @@ -781,6 +781,12 @@ the external results the task reported, one object per url with `url`, `provider task declared `link` or `links`, so every url in it is an http(s) url it validated. A reader renders these rather than reading urls out of `output`. +A `plan_admitted` task carries `keyed` (contract 1.16.0), each `producer.field` a mapped task +narrows to its own item's entry, omitted when empty. A mapped node's folded output and its +`task_result` fan-out summary count the instances that settled not taken under `not_taken`; a +reader that predates it reads the count as zero. A route may declare `over`, and its instances +settle as `route[K]`, each with the decision for item `K` as its output. + A `plan_admitted` task carries `emits_files` (contract 1.14.0): each declared path, as a string, or `{path, schema}` when the declaration gave a schema. A field typed by a schema carries `{"schema": ""}` as its `type`. A `task_result` event carries `repairs` (contract 1.14.0), one diff --git a/docs/dsl-reference.md b/docs/dsl-reference.md index ac03366f..a91bfaf4 100644 --- a/docs/dsl-reference.md +++ b/docs/dsl-reference.md @@ -35,6 +35,7 @@ An agent turn driven by a prompt. | `emits_files` | `list[str] \| dict[str, schema \| None]` | Workspace files the task produces. A dependent is staged with the declared files of every dependency that passed. The dict form maps a path to `schema_file(...)` or `None`; a file with a schema must hold JSON the schema admits when the task passes, or the task fails. | | `over` | `producer.field` | Map the task over a dependency's emitted list, one instance per item. | | `max_fanout` | `int` | Instance cap for `over`, within the engine's ceiling of 256. | +| `keyed` | `list[producer.field]` | Object fields of unmapped dependencies, keyed by item, that each instance of a mapped task reads narrowed to its own item's entry. | | `revise` | `task \| list[task]` | The tasks this task sends back when it settles failing. They run again, in dependency order, with the verdict under `revision`, then this task does, until this task stops failing or `max_rounds` is spent. Every task on a path between them and this task must be listed. Playbooks only; not with `over`. | | `max_rounds` | `int` | Round cap for `revise`, counting the first, from 2 to the engine's ceiling of 5. | | `timeout` | `str` | How long one attempt may run, as `90s`, `10m`, or `2h`. At the limit the attempt's whole process group is killed and the task settles failed with a note naming the limit; it is not retried. The run's `--max-time` bounds every attempt as well, and a playbook refuses a timeout longer than it before dispatching anything. | @@ -68,6 +69,7 @@ An agent turn whose prompt is a skill's instructions plus its arguments. | `emits_files` | `list[str] \| dict[str, schema \| None]` | Workspace files the task produces. A dependent is staged with the declared files of every dependency that passed. The dict form maps a path to `schema_file(...)` or `None`; a file with a schema must hold JSON the schema admits when the task passes, or the task fails. | | `over` | `producer.field` | Map the task over a dependency's emitted list, one instance per item. | | `max_fanout` | `int` | Instance cap for `over`, within the engine's ceiling of 256. | +| `keyed` | `list[producer.field]` | Object fields of unmapped dependencies, keyed by item, that each instance of a mapped task reads narrowed to its own item's entry. | | `revise` | `task \| list[task]` | The tasks this task sends back when it settles failing. They run again, in dependency order, with the verdict under `revision`, then this task does, until this task stops failing or `max_rounds` is spent. Every task on a path between them and this task must be listed. Playbooks only; not with `over`. | | `max_rounds` | `int` | Round cap for `revise`, counting the first, from 2 to the engine's ceiling of 5. | | `timeout` | `str` | How long one attempt may run, as `90s`, `10m`, or `2h`. At the limit the attempt's whole process group is killed and the task settles failed with a note naming the limit; it is not retried. The run's `--max-time` bounds every attempt as well, and a playbook refuses a timeout longer than it before dispatching anything. | @@ -94,6 +96,7 @@ A deterministic shell task in the candidate workspace. | `emits_files` | `list[str] \| dict[str, schema \| None]` | Workspace files the task produces. A dependent is staged with the declared files of every dependency that passed. The dict form maps a path to `schema_file(...)` or `None`; a file with a schema must hold JSON the schema admits when the task passes, or the task fails. | | `over` | `producer.field` | Map the task over a dependency's emitted list, one instance per item. | | `max_fanout` | `int` | Instance cap for `over`, within the engine's ceiling of 256. | +| `keyed` | `list[producer.field]` | Object fields of unmapped dependencies, keyed by item, that each instance of a mapped task reads narrowed to its own item's entry. | | `revise` | `task \| list[task]` | The tasks this task sends back when it settles failing. They run again, in dependency order, with the verdict under `revision`, then this task does, until this task stops failing or `max_rounds` is spent. Every task on a path between them and this task must be listed. Playbooks only; not with `over`. | | `max_rounds` | `int` | Round cap for `revise`, counting the first, from 2 to the engine's ceiling of 5. | | `timeout` | `str` | How long one attempt may run, as `90s`, `10m`, or `2h`. At the limit the attempt's whole process group is killed and the task settles failed with a note naming the limit; it is not retried. The run's `--max-time` bounds every attempt as well, and a playbook refuses a timeout longer than it before dispatching anything. | @@ -122,6 +125,7 @@ A measurement command. Its last non-empty stdout line is a JSON object; `pass = | `emits_files` | `list[str] \| dict[str, schema \| None]` | Workspace files the task produces. A dependent is staged with the declared files of every dependency that passed. The dict form maps a path to `schema_file(...)` or `None`; a file with a schema must hold JSON the schema admits when the task passes, or the task fails. | | `over` | `producer.field` | Map the task over a dependency's emitted list, one instance per item. | | `max_fanout` | `int` | Instance cap for `over`, within the engine's ceiling of 256. | +| `keyed` | `list[producer.field]` | Object fields of unmapped dependencies, keyed by item, that each instance of a mapped task reads narrowed to its own item's entry. | | `revise` | `task \| list[task]` | The tasks this task sends back when it settles failing. They run again, in dependency order, with the verdict under `revision`, then this task does, until this task stops failing or `max_rounds` is spent. Every task on a path between them and this task must be listed. Playbooks only; not with `over`. | | `max_rounds` | `int` | Round cap for `revise`, counting the first, from 2 to the engine's ceiling of 5. | | `timeout` | `str` | How long one attempt may run, as `90s`, `10m`, or `2h`. At the limit the attempt's whole process group is killed and the task settles failed with a note naming the limit; it is not retried. The run's `--max-time` bounds every attempt as well, and a playbook refuses a timeout longer than it before dispatching anything. | @@ -276,6 +280,10 @@ Engine-owned decision: answers typed questions about its dependencies' outputs a | `required` | `bool` | False makes the route advisory. | | `join` | `"all" \| "passed" \| "settled"` | Which dependency outputs form the state, as on any task. | | `stage` | `"iteration" \| "epilogue"` | As on any task. | +| `files` | `list[str]` | Declared JSON files of its dependencies the decision model reads, under `files` in its state, by dependency. Each path must be declared with a schema. A model decider only. | +| `over` | `producer.field` | Decide once per item of a dependency's emitted list. A dependency mapped over the same list is read per item, and so is a `when` on this route in a task mapped over it. | +| `max_fanout` | `int` | Item cap for `over`, within the engine's ceiling of 256. | +| `keyed` | `list[producer.field]` | Object fields of unmapped dependencies, keyed by item, that each instance of a mapped task reads narrowed to its own item's entry. | | `when` | `route.question` | Run only on a listed answer to one question of a `route()` this task depends on. Otherwise the task settles `not_taken`: no dispatch, no spend, no effect on validity, and every `all`-join dependent is not taken with it. Rejoin branches with `join = "passed"` or `join = "settled"`. Playbook and custom workflows only. | | `answers` | `str \| list[str]` | The answers `when` accepts: labels the question declares, or `"uncertain"`. Defaults to `"yes"` for a noul and is required for a choice. | | `otherwise` | `bool` | In place of `answers`: every answer no other task lists and the question does not `drop`, `"uncertain"` included. Expanded at compile time. An unreachable `otherwise` is an error. | diff --git a/docs/playbook-patterns.md b/docs/playbook-patterns.md index 4babee2b..c81bd0b4 100644 --- a/docs/playbook-patterns.md +++ b/docs/playbook-patterns.md @@ -300,6 +300,63 @@ def auditor(topic, blocking): auditors = [auditor(topic, blocking) for topic, blocking in AUDITS] ``` +## Decide per item + +A route maps over a list like any task, and decides each item on its own. A task mapped over the +same list reads it per item: instance `fix[K]` sees `judge[K]`'s answer and `analyze[K]`'s output, +and its `when` is read against `judge[K]` alone. An item whose answer is not listed settles +not taken, without dispatch or spend, and neither fails its fan-out nor holds up the other items: + +```python +analyze = agent( + name = "analyze", + prompt = prompt_file("analyze.md"), + depends_on = [prepare], + over = prepare.members, + max_fanout = 120, + emits_files = {"RESULT.json": schema_file("result.schema.json")}, +) + +judge = route( + name = "judge", + depends_on = [prepare, analyze], + over = prepare.members, + max_fanout = 120, + keyed = [prepare.lanes], + files = ["RESULT.json"], + min_confidence = 0.8, + required = False, + questions = { + "supported": noul( + ask = "Does the evidence in RESULT.json support its reachability answer?", + drop = ["yes"], + ), + }, +) + +review = command( + name = "review", + run = "./flag.sh", + depends_on = [prepare, judge], + over = prepare.members, + max_fanout = 120, + when = judge.supported, + answers = ["no", "uncertain"], +) +``` + +- `files` puts dependencies' declared JSON files in the decision model's state under `files`, by + dependency. A file has to be declared with a schema, and a mapped dependency's file is read only + by a route aligned with it, so `judge[K]` reads `analyze[K]`'s `RESULT.json`. +- `keyed` narrows an object field of an unmapped dependency to the item's entry: `prepare.lanes` + is keyed by item, and `judge[K]` sees `lanes[K]` in place of the whole object. Any mapped task + may declare it. +- A route's state also carries the item under `item`. +- A fan-out whose every item was not taken settles not taken, and so does everything joining it + with `all`. Its folded output counts the items under `not_taken`. +- Each item's decision is its own row, so a resumed run asks only about the items it has no + answer for. + ## Keep a conversation Agent tasks that share a `session` continue one conversation, in dependency order: diff --git a/docs/rfc/RFC-0001.md b/docs/rfc/RFC-0001.md index feebce49..881f2cf3 100644 --- a/docs/rfc/RFC-0001.md +++ b/docs/rfc/RFC-0001.md @@ -1,5 +1,5 @@ - + # RFC-0001: Crucible implementation contract @@ -314,7 +314,7 @@ A reviewer under [RFC-0002:C-REVISE-LOOP](../rfc/RFC-0002.md#rfc-0002c-revise-lo A task whose when is not satisfied must settle as not taken, without dispatch and without spend. A when over a route that itself settled not taken is not satisfied. A when over a route that settled any other way without passing is not satisfied either, except that a task joining "all" on that route is blocked by the failure as any dependent is. Not taken is a terminal status distinct from skipped. It must not count against validity, must not short-circuit the run, and must not be reported as a failure. A task joining "all" on a dependency that settled not taken must itself settle not taken, so an untaken branch is untaken to its leaves. A task joining "passed" or "settled" is dispatched as those joins already define, and is how branches rejoin. A required task is required when taken: if its when is satisfied it gates validity as any required task does. The workflow's result task must not be conditional and must not be able to settle not taken through "all" joins alone. -A route task must not declare over; routing each element of a list is not defined by this clause. +A route task may declare over. [RFC-0001:C-ROUTE-EACH](../rfc/RFC-0001.md#rfc-0001c-route-each) states how a mapped route decides each element of a list and how a when reads that decision; where it and this clause differ for a mapped route, it governs. The route and when constructs are admitted in the playbook and custom lanes. An autoresearch workflow must reject both: the keep rule is sourced from a frozen measure, and a model decision must not sit on the path to it. @@ -326,7 +326,7 @@ An orchestrator tells the engine where models are reached through one document, A binding states what the inference is for and how it is reached: a role, a protocol, a model, optionally a URL, and optionally the name of the environment variable that holds its credential. An absent URL means the protocol's own service address: for decisions, OpenAI's https://api.openai.com/v1/decisions. A system_one binding must name one, because system_one has none. The document must never carry a credential's value. The roles are agent, the model an agent turn thinks with, and decision, the model a route task asks under [RFC-0001:C-ROUTE](../rfc/RFC-0001.md#rfc-0001c-route). The protocols are messages, chat_completions, responses, system_one, and decisions. A document must hold at most one binding per role. An agent binding must speak messages, chat_completions, or responses; a decision binding must speak system_one or decisions. A URL must be absolute http or https, name a host, and carry no credentials. A credential variable name must be a valid environment variable name and must not be CRUCIBLE_INFERENCE. The engine must reject a document that breaks any of these, naming the binding. -A binding says nothing about delivery. The engine derives it from the role's consumer. Where the engine's own process is the client, as for a decision, the engine must use the binding directly and the credential must not leave the engine. Where the client runs inside a sandbox, as an agent harness does, the engine relays: it must project the binding into the sandbox under the names that client reads, and must add the binding's endpoint to the sandbox's egress policy. Those names are the engine's concern and are not part of this contract; an orchestrator must not be required to know them. +A binding says nothing about delivery. The engine derives it from the role's consumer. Where the engine's own process is the client, as for a decision, the engine must use the binding directly and the credential must not leave the engine. Where the client runs inside a sandbox, as an agent harness does, the engine relays: it must project the binding into the sandbox under the names that client reads, and must add the binding's endpoint to the sandbox's egress policy. Those names are the engine's concern and are not part of this contract; an orchestrator must not be required to know them. A decision binding has no relayed consumer: the engine must not project it, its endpoint, its credential, or any interface that reaches it into any sandbox or task process, so a model decision is made only by a route task under [RFC-0001:C-ROUTE](../rfc/RFC-0001.md#rfc-0001c-route). The decision capability of [RFC-0002:C-PLAYBOOK-CAPS](../rfc/RFC-0002.md#rfc-0002c-playbook-caps) is available exactly when the document holds a decision binding. A named credential variable that is unset or empty when its binding is first used must fail that use with an error naming the variable, and must not be treated as an unauthenticated endpoint. @@ -334,6 +334,36 @@ An agent binding is the whole answer for agent turns. Its model is the run's mod *Since: v0.2.0* +### [RFC-0001:C-ROUTE-EACH] Per-element routes (Normative) + +A route task may declare over = producer.field with max_fanout, under the rules for over in [RFC-0001:C-WORKFLOW](../rfc/RFC-0001.md#rfc-0001c-workflow): the field must be a list of strings, and the route's instance for an element is named by that element's item, its key. A mapped route decides once per element. Each element's decision is one request to the decision model, transport retries aside, or one read for an output-decided route, and its instance settles on its own: its output is that element's decision in the form [RFC-0001:C-ROUTE](../rfc/RFC-0001.md#rfc-0001c-route) defines for a route, and an element that fails or transport-fails must not change the decision recorded for any other element. The engine must check the run's cost and wall-clock ceilings before each element's request as it does before any dispatch, and an element not yet asked when a ceiling is exhausted must settle blocked. + +Two tasks are aligned when both declare over naming the same producer and the same field. Every edge from a task to a dependency aligned with it is read per element, whatever kind of task either is. Across such an edge the dependent must not expand until the dependency has settled as a node, and must then expand whatever the node's status, unless dispatch has stopped for another reason. Its instance for an element must receive the dependency's instance for the same key in place of the node: under "all" or "passed" that instance's output under the dependency's name, and under "settled" that instance's entry as [RFC-0002:C-SETTLED-JOIN](../rfc/RFC-0002.md#rfc-0002c-settled-join) defines it, with no per_instance mapping. The instance must apply its join to that instance's status in place of the node's, and must read an absent instance as blocked. An instance joining "all" on an aligned dependency whose instance for the same key settled not taken therefore settles not taken, so an untaken element is untaken to the leaves of its aligned chain as [RFC-0001:C-ROUTE](../rfc/RFC-0001.md#rfc-0001c-route) requires of a branch. + +The state sent for an element must be the route's dependency inputs, read per element across aligned edges, with the element's key added under the reserved key "item". A mapped route must not depend on a task named "item", and validation must reject one that does. + +Any mapped task, a route or not, may declare keyed, a list of producer.field references. Each must name a field declared by one of the task's dependencies that is not mapped, and where the declaration types it, as "object" or by a schema whose top-level type is "object"; validation must reject any other, naming the reference, and must reject keyed on a task that does not declare over. For each keyed field, the inputs an instance receives, and for a route the state it sends, must carry in place of the whole field only the field's entry under the instance's key, read from the producer's output whatever the task's join; the producer's other fields are unchanged. A keyed value that is not an object at run time must fail every instance, naming the field, and an absent entry must fail that instance, naming the key and the field. keyed must reach the generated TOML, and the admitted-plan event must carry each task's keyed references as a list of "producer.field" strings, empty for a task that declares none, so a reader can draw which edges are narrowed. + +An output-decided route that declares over reads each element's answers from its source per element. Where the source is aligned with the route, the answers are read from the source's instance for the element, under the rules [RFC-0001:C-ROUTE](../rfc/RFC-0001.md#rfc-0001c-route) states for a source's output. Otherwise the source must not be mapped, and its output must carry, under each question id, an object keyed by element key whose entry for the element is the answer under those rules; validation must reject a source whose emits types a question's field as anything but "object" or an object schema, and an absent entry must fail that element. + +A when that names a mapped route must be declared by a task aligned with that route. Validation must reject a when on a mapped route declared by a task that is not mapped, or that maps over a different list, naming both references. Such a when is evaluated per element: the dependent's instance for an element applies the rules of [RFC-0001:C-ROUTE](../rfc/RFC-0001.md#rfc-0001c-route) to the route's instance for the same key, so it settles not taken, without dispatch and without spend, when that element's resolved label is not among the labels the when lists, and the instances for other elements are unaffected. The rules of [RFC-0001:C-ROUTE](../rfc/RFC-0001.md#rfc-0001c-route) for drop, otherwise, and label coverage apply to a mapped route's questions unchanged. Several tasks aligned with one mapped route may each declare a when on the same question, and so partition the elements among them by answer. A mapped route may itself declare a when on an aligned route, so it is asked only about the elements that route's answer leaves to it. A when on a route that does not declare over, declared by a mapped task, is evaluated once for the node under [RFC-0001:C-ROUTE](../rfc/RFC-0001.md#rfc-0001c-route). + +A mapped task's instance that settled not taken must not count as failed. A mapped node with no instances must settle passing. Otherwise it must settle not taken when every instance was not taken, passing when every instance passed or was not taken, and failing otherwise. Its folded output must carry the number of instances not taken beside the numbers that passed and failed, and must carry no output for an instance not taken. The task-result event's fan-out summary must carry the same count under not_taken; a reader that predates it reads a summary without it as zero. + +Each element's decision is its instance's recorded output, and an instance settled not taken is terminal. On resume, under [RFC-0002:C-PLAYBOOK-RESUME](../rfc/RFC-0002.md#rfc-0002c-playbook-resume), the engine must fold both, must not ask again for an element that has a recorded decision, and must ask only for the elements that have none. + +*Since: v0.2.0* + +### [RFC-0001:C-ROUTE-FILES] Declared files in a route's state (Normative) + +A route may declare files, a list of workspace-relative paths. Each path must be declared under emits_files by at least one of the route's dependencies with a JSON Schema ([RFC-0002:C-TASK-FILES](../rfc/RFC-0002.md#rfc-0002c-task-files)); validation must reject a path no dependency declares, a path every declaring dependency declares without a schema, and a path declared only by a mapped dependency that is not aligned with the route under [RFC-0001:C-ROUTE-EACH](../rfc/RFC-0001.md#rfc-0001c-route-each), naming the path. + +The state sent to the decision model must carry, under the reserved key "files", one entry per dependency that declares a listed path, under that dependency's name, mapping each such path to the parsed content of the file staged for the route. For a route mapped over a list, an aligned dependency's file is its instance's for the same element. A dependency whose file was not staged for the route contributes no entry for that path. A staged file that does not parse as JSON, as a failed dependency's file under a "settled" join may not, must contribute null in its place, and the engine must record the path on the route's result. A route must not depend on a task named "files", whether or not it declares files, and validation must reject one that does. A route that declares no files sends no "files" key. + +An output-decided route reads its answers from its source's JSON output and must reject files at compile time, because it asks no model. + +*Since: v0.2.0* + --- ## Changelog @@ -370,6 +400,11 @@ Admit the playbook lane - skill joins the enumerated constructors: an agent task whose prompt is assembled from a shipped SKILL.md and the invocation's arguments - a decision binding may speak decisions, the OpenAI Decisions API, reached at OpenAI's address when it names no URL (C-INFERENCE, C-ROUTE) - the capability a model-decided route needs is decision, not systemone, since it names the role rather than one API (C-ROUTE) +- a route may map over a list and decide each element on its own; a task reads a dependency mapped over the same list per element, a when on a mapped route is read per element, and an instance not taken does not fail its fan-out (C-ROUTE-EACH) +- keyed: a mapped task may narrow an object field of an unmapped dependency to its own element's entry, and the admitted-plan event carries it (C-ROUTE-EACH) +- a route may put declared JSON files of its dependencies in the decision model's state (C-ROUTE-FILES) +- a route task may declare over (C-ROUTE) +- a decision binding is never relayed into a sandbox or task process, so only a route asks the decision model (C-INFERENCE) ### v0.1.0 (2026-08-19) diff --git a/gov/adr/ADR-0062-per-element-decisions-are-routes-mapped-over-the-list-the-gated-work-maps-over.toml b/gov/adr/ADR-0062-per-element-decisions-are-routes-mapped-over-the-list-the-gated-work-maps-over.toml new file mode 100644 index 00000000..ee4010fe --- /dev/null +++ b/gov/adr/ADR-0062-per-element-decisions-are-routes-mapped-over-the-list-the-gated-work-maps-over.toml @@ -0,0 +1,125 @@ +#:schema ../schema/adr.schema.json + +[govctl] +id = "ADR-0062" +title = "Per-element decisions are routes mapped over the list the gated work maps over" +status = "proposed" +date = "2026-10-07" +refs = [ + "RFC-0001:C-ROUTE", + "RFC-0001:C-ROUTE-EACH", + "RFC-0001:C-ROUTE-FILES", + "RFC-0001:C-WORKFLOW", + "RFC-0002:C-SETTLED-JOIN", + "RFC-0002:C-PLAYBOOK-RESUME", + "RFC-0002:C-REVISE-LOOP", + "ADR-0064", +] + +[content] +context = """ +A route ([[RFC-0001:C-ROUTE]]) asks a decision model typed questions once and gates downstream tasks with when. A route cannot declare over, so a playbook that fans agent turns out over a list cannot put a decision in front of each turn: the only decision available is one for the whole run. + +The motivating case is a CVE triage playbook: a command stages per-ticket facts and emits the ticket keys, an agent task maps over the keys with one turn per ticket, a command applies the verdicts and emits the keys worth fixing, and a second agent task maps over those. Most of the cost is the per-ticket turns, and the decisions that would cut it are per ticket: skip a turn when the staged facts already settle reachability, give hard tickets more effort than easy ones, gate a fix on whether the version bump keeps the API, judge whether an agent's result file is supported by its own evidence, and flag likely duplicates. + +What exists today: + +- A mapped task names each instance by its item, `node[key]`, and a resumed run keeps an instance's result by that key ([[RFC-0001:C-WORKFLOW]]). +- A mapped dependency reaches its dependents as one folded output; a fold passes only when every instance passed. +- A when is evaluated once per node, before the node expands. +- Not taken is a terminal status, and an untaken branch is untaken to its leaves through "all" joins. +- A route's state is its dependencies' JSON outputs. A task whose only output is a declared file gives a route nothing to read. +- A revise loop refuses a body that maps over a list, and its reviewer must be an agent, command, or evaluate task ([[RFC-0002:C-REVISE-LOOP]]).""" +decision = """ +Let a route map over a list, and evaluate a when on a mapped route per element in a task that maps over the same list. Two tasks mapped over the same producer field are aligned; their instances pair by key, which is already the identity a resume relies on. The normative rules are [[RFC-0001:C-ROUTE-EACH]] and [[RFC-0001:C-ROUTE-FILES]]. + +- **One decision per element.** A mapped route's instance for a key is that element's decision, recorded and folded on resume like any instance, so an element is asked once per run. +- **Per-element reads.** On every aligned edge, the dependency reaches an instance as its own instance for the same key, under the dependency's own name, instead of the fold. One rule covers routes, the tasks they gate, and plain per-element pipelines alike, and an untaken element stays untaken down the whole aligned chain. +- **Gating.** when keeps its meaning and gains a per-element reading. Partitioning by answer is several aligned tasks with whens on the same question, and the existing coverage rule makes every label lead somewhere. +- **State.** An element's state is the route's inputs plus its key. keyed narrows a per-element object field to the element's entry, for any mapped task, so a hundred-element fan-out does not hand every instance, model request or agent turn, the whole field, and one element's facts cannot sway another's answer. A route may also read declared JSON files, which is how it judges an agent whose output is a file. +- **Folds.** An instance not taken is not a failure. A fan-out whose instances were all untaken is itself untaken, so its "all" dependents are untaken as any branch's are and a "settled" reducer still runs. + +For the triage playbook, sketched (illustrative; the clauses above are the contract): + +```python +gate = route(name = "gate", depends_on = [prepare], over = prepare.members, max_fanout = 120, + keyed = [prepare.facts], min_confidence = 0.8, + questions = {"depth": choice(ask = "...", options = {"settled": "...", "quick": "...", "deep": "..."}, + drop = ["settled"])}) +quick = agent(name = "quick", effort = "low", depends_on = [prepare, gate], over = prepare.members, + max_fanout = 120, when = gate.depth.quick, ...) +deep = agent(name = "deep", effort = "high", depends_on = [prepare, gate], over = prepare.members, + max_fanout = 120, when = gate.depth.otherwise, ...) # deep and uncertain +judge = route(name = "judge", depends_on = [quick, deep], join = "settled", over = prepare.members, + max_fanout = 120, files = ["RESULT.json"], min_confidence = 0.8, + questions = {"supported": noul(ask = "Does the evidence support the verdict?", drop = ["yes"])}) +review = agent(name = "review", depends_on = [judge], over = prepare.members, max_fanout = 120, + when = judge.supported.otherwise, ...) # no and uncertain +``` + +A ticket answered "settled" costs one decision and no turn. A deterministic output-decided route can sit in front of the model-decided one when the staged facts already carry an answer; the model route then declares a when on it and is asked only about the rest. A fix gate is the same shape over the fix list.""" +consequences = """ +- The expensive per-element turns become conditional per element, and an untaken element shows as a not-taken row under its own key rather than disappearing from the run. +- Every element costs one decision request. max_fanout bounds the requests per run, and the run's ceilings are checked before each one. Decision requests are not charged to the run's budget: they cost little next to the turns they gate, so the ceiling check before each request is in practice a wall-clock check, and max_fanout bounds the count. +- An aligned dependent waits for the whole mapped dependency to settle before it expands. Starting element k's turn as soon as element k's decision lands is left out; the decision is the cheap stage, so the added latency is small next to the turns it gates. +- An aligned mapped-to-mapped edge changes what the dependent instance reads: an instance, not the fold. This is breaking for a pack whose mapped task consumes the fold of another task mapped over the same list. No published example or known pack has such an edge; a task mapped over a different list, or not mapped, still reads the fold. +- The fan-out summary gains a not_taken count. Readers that predate it read zero, and a fold that would have failed only because instances were untaken now passes. +- Judging an agent's result feeds a review lane, not a send-back ([[ADR-0064]]). +- A playbook that relied on every element getting a full turn now relies on the decision model's calibration and min_confidence. uncertain still has to be routed, so the author decides what happens when the model is unsure.""" + +[[content.alternatives]] +text = "Map a route over the list, read every aligned edge per element, and read a when on a mapped route per element" +status = "accepted" +pros = [ + "Reuses the instance key, the not-taken status, the coverage rule, and resume folding as they are.", + "Every element keeps a row in every gated task, so the record shows why a turn did not run.", + "Partitioning by answer and chaining a deterministic route in front of a model route need no new construct.", +] +cons = [ + "An aligned mapped-to-mapped edge reads an instance instead of the fold, a breaking change for any pack that relied on the fold there.", + "Each element is a separate decision request.", +] + +[[content.alternatives]] +text = "Map gated tasks over a filtered list taken from the route's fold (over = gate.depth.quick)" +status = "rejected" +pros = [ + "No per-element when; a gated task simply has fewer instances.", + "Instance names stay keyed by item.", +] +cons = [ + "An element the route sent elsewhere has no row in the gated task, so a settled reducer cannot tell untaken from never listed.", + "The coverage check would have to be rebuilt for list references instead of when labels.", + "over would read a list the engine synthesises from a route's record, a second meaning for a field reference.", +] +rejection_reason = "Loses the not-taken record and duplicates the coverage rule in a new place." + +[[content.alternatives]] +text = "Quantified when on an unmapped route over the fold (any or all elements answered a label)" +status = "rejected" +pros = ["Small change to when; useful for run-level gates such as skipping a report."] +cons = [ + "Still one decision for the whole list, so it cannot gate element k's turn on element k's answer.", + "Does not address the cost driver, which is per-element turns.", +] +rejection_reason = "Answers a different question; it can be added later on top of mapped routes if a run-level gate is needed." + +[[content.alternatives]] +text = "One route request per run carrying every element, answered as a map keyed by element" +status = "rejected" +pros = ["One request instead of N."] +cons = [ + "Neither decision API answers a question per element; the questions would have to be multiplied per key at run time, which C-ROUTE forbids because questions are compile-time.", + "One malformed answer fails every element, and a resume re-asks all of them.", +] +rejection_reason = "Breaks compile-time questions and per-element failure isolation." + +[[content.alternatives]] +text = "Read aligned edges per element only along chains that involve a route" +status = "rejected" +pros = ["Existing mapped-to-mapped edges keep reading the fold."] +cons = [ + "Two read rules for the same shape of edge, chosen by what is upstream.", + "Needs a transitive per-element marking so an untaken element stays untaken past the first hop.", +] +rejection_reason = "No known pack consumes the fold on an aligned edge, so the compatibility it buys is not worth a second rule." diff --git a/gov/adr/ADR-0064-only-a-route-asks-the-decision-model.toml b/gov/adr/ADR-0064-only-a-route-asks-the-decision-model.toml new file mode 100644 index 00000000..34f78295 --- /dev/null +++ b/gov/adr/ADR-0064-only-a-route-asks-the-decision-model.toml @@ -0,0 +1,82 @@ +#:schema ../schema/adr.schema.json + +[govctl] +id = "ADR-0064" +title = "Only a route asks the decision model" +status = "proposed" +date = "2026-10-07" +refs = [ + "RFC-0001:C-ROUTE", + "RFC-0001:C-INFERENCE", + "RFC-0001:C-ROUTE-EACH", + "RFC-0001:C-ROUTE-FILES", + "RFC-0002:C-PLAYBOOK-CAPS", + "RFC-0002:C-REVISE-LOOP", + "ADR-0062", +] + +[content] +context = """ +The decision binding is used by the engine's own process, and its credential must not leave the engine ([[RFC-0001:C-INFERENCE]]). A route is its only consumer today. Two pressures push against that: + +- A command task in a triage playbook already holds the per-element facts and could ask the decision model itself, through a broker tool that keeps the credential engine-side, instead of the playbook declaring a route for each question. +- A route that judges an agent's output ("does this result file's evidence support its claim?") wants to feed its verdict back to the agent, as a reviewer in a revise loop does ([[RFC-0002:C-REVISE-LOOP]]), not only gate a later lane. + +A route's guarantees come from being a graph node: its questions and labels are compile-time and in the generated TOML, every label must lead somewhere, its answer and full distribution are the task's recorded output, and a resumed run never asks again. A tool call from inside a task has none of those.""" +decision = """ +Keep every model decision a route. The engine does not relay the decision binding into any sandbox or task process and offers no tool that asks it ([[RFC-0001:C-INFERENCE]]). Per-element decisions are mapped routes ([[RFC-0001:C-ROUTE-EACH]]), and a route judges an agent's output by reading its declared JSON files ([[RFC-0001:C-ROUTE-FILES]]) and gating a review lane with when. + +A judgment is not yet a send-back, and C-REVISE-LOOP does not cover it: a reviewer must be an agent, command, or evaluate task, and a body must not map over a list. Admitting a route reviewer needs a stated mapping from labels to a failing verdict, and admitting mapped bodies needs per-element rounds. Both are separate decisions, deferred until a playbook needs the judged conversation resumed rather than a second agent reviewing.""" +consequences = """ +- Every model decision in a run is visible in the plan, covered by the label rule, recorded with its distribution, and folded on resume. +- A command that would like a quick model opinion must be split around a route, which costs a node and an edge per question set. +- Questions cannot be composed at run time. A playbook that needs a question per discovered category has to declare the categories up front, which is the constraint C-ROUTE already imposes. +- The credential boundary stays simple: no sandbox egress rule, no broker tool, and no per-task decision budget to police. +- A judged agent result that fails review goes to a separate review turn, which starts a new conversation and pays to re-read the work. Resuming the original agent's session with the judgment is the follow-up if that cost matters.""" + +[[content.alternatives]] +text = "Only route tasks ask the decision model" +status = "accepted" +pros = [ + "Compile-time questions, label coverage, recorded distributions, and resume folding hold for every decision.", + "No credential, endpoint, or tool reaches a task process.", +] +cons = ["A task cannot ask an ad hoc question mid-work."] + +[[content.alternatives]] +text = "A broker tool that lets a command task ask the decision model, credential kept in the engine" +status = "rejected" +pros = [ + "The task asks exactly when it has the facts, with no extra node.", + "The credential still never leaves the engine.", +] +cons = [ + "Questions and labels are authored at run time by the task, outside the plan and the coverage rule.", + "The answer is whatever the task writes into its own output, not a recorded decision; a resume re-runs the task and asks again.", + "Calls are unbounded per task; max_fanout and the run's ceilings do not see them.", + "A sandboxed task would need egress or a broker channel for decision traffic.", +] +rejection_reason = "Gives up every guarantee that makes a route auditable, for flexibility the per-element route already provides." + +[[content.alternatives]] +text = "Declared question sets a command task may ask by name through the broker" +status = "rejected" +pros = [ + "Questions stay compile-time.", + "A task chooses when to ask.", +] +cons = [ + "The answer still lands in the task's own output rather than a decision record.", + "Gating on it needs the task to emit it and an output-decided route to read it, which is a route again.", +] +rejection_reason = "Ends up as a route with extra plumbing." + +[[content.alternatives]] +text = "Admit a route as a revise reviewer, failing on listed labels" +status = "rejected" +pros = ["A judgment can resume the judged agent's conversation."] +cons = [ + "Needs a label-to-verdict mapping and its own coverage rule.", + "Useless for the per-element case until revise bodies may map over a list.", +] +rejection_reason = "Deferred: needs a label-to-verdict mapping and per-element revise rounds, each a separate decision." diff --git a/gov/rfc/RFC-0001/clauses/C-INFERENCE.toml b/gov/rfc/RFC-0001/clauses/C-INFERENCE.toml index c23f7cb7..57932bc1 100644 --- a/gov/rfc/RFC-0001/clauses/C-INFERENCE.toml +++ b/gov/rfc/RFC-0001/clauses/C-INFERENCE.toml @@ -13,7 +13,7 @@ An orchestrator tells the engine where models are reached through one document, A binding states what the inference is for and how it is reached: a role, a protocol, a model, optionally a URL, and optionally the name of the environment variable that holds its credential. An absent URL means the protocol's own service address: for decisions, OpenAI's https://api.openai.com/v1/decisions. A system_one binding must name one, because system_one has none. The document must never carry a credential's value. The roles are agent, the model an agent turn thinks with, and decision, the model a route task asks under [[RFC-0001:C-ROUTE]]. The protocols are messages, chat_completions, responses, system_one, and decisions. A document must hold at most one binding per role. An agent binding must speak messages, chat_completions, or responses; a decision binding must speak system_one or decisions. A URL must be absolute http or https, name a host, and carry no credentials. A credential variable name must be a valid environment variable name and must not be CRUCIBLE_INFERENCE. The engine must reject a document that breaks any of these, naming the binding. -A binding says nothing about delivery. The engine derives it from the role's consumer. Where the engine's own process is the client, as for a decision, the engine must use the binding directly and the credential must not leave the engine. Where the client runs inside a sandbox, as an agent harness does, the engine relays: it must project the binding into the sandbox under the names that client reads, and must add the binding's endpoint to the sandbox's egress policy. Those names are the engine's concern and are not part of this contract; an orchestrator must not be required to know them. +A binding says nothing about delivery. The engine derives it from the role's consumer. Where the engine's own process is the client, as for a decision, the engine must use the binding directly and the credential must not leave the engine. Where the client runs inside a sandbox, as an agent harness does, the engine relays: it must project the binding into the sandbox under the names that client reads, and must add the binding's endpoint to the sandbox's egress policy. Those names are the engine's concern and are not part of this contract; an orchestrator must not be required to know them. A decision binding has no relayed consumer: the engine must not project it, its endpoint, its credential, or any interface that reaches it into any sandbox or task process, so a model decision is made only by a route task under [[RFC-0001:C-ROUTE]]. The decision capability of [[RFC-0002:C-PLAYBOOK-CAPS]] is available exactly when the document holds a decision binding. A named credential variable that is unset or empty when its binding is first used must fail that use with an error naming the variable, and must not be treated as an unauthenticated endpoint. diff --git a/gov/rfc/RFC-0001/clauses/C-ROUTE-EACH.toml b/gov/rfc/RFC-0001/clauses/C-ROUTE-EACH.toml new file mode 100644 index 00000000..a3d0c28a --- /dev/null +++ b/gov/rfc/RFC-0001/clauses/C-ROUTE-EACH.toml @@ -0,0 +1,26 @@ +#:schema ../../../schema/clause.schema.json + +[govctl] +id = "C-ROUTE-EACH" +title = "Per-element routes" +kind = "normative" +status = "active" +since = "0.2.0" + +[content] +text = """ +A route task may declare over = producer.field with max_fanout, under the rules for over in [[RFC-0001:C-WORKFLOW]]: the field must be a list of strings, and the route's instance for an element is named by that element's item, its key. A mapped route decides once per element. Each element's decision is one request to the decision model, transport retries aside, or one read for an output-decided route, and its instance settles on its own: its output is that element's decision in the form [[RFC-0001:C-ROUTE]] defines for a route, and an element that fails or transport-fails must not change the decision recorded for any other element. The engine must check the run's cost and wall-clock ceilings before each element's request as it does before any dispatch, and an element not yet asked when a ceiling is exhausted must settle blocked. + +Two tasks are aligned when both declare over naming the same producer and the same field. Every edge from a task to a dependency aligned with it is read per element, whatever kind of task either is. Across such an edge the dependent must not expand until the dependency has settled as a node, and must then expand whatever the node's status, unless dispatch has stopped for another reason. Its instance for an element must receive the dependency's instance for the same key in place of the node: under "all" or "passed" that instance's output under the dependency's name, and under "settled" that instance's entry as [[RFC-0002:C-SETTLED-JOIN]] defines it, with no per_instance mapping. The instance must apply its join to that instance's status in place of the node's, and must read an absent instance as blocked. An instance joining "all" on an aligned dependency whose instance for the same key settled not taken therefore settles not taken, so an untaken element is untaken to the leaves of its aligned chain as [[RFC-0001:C-ROUTE]] requires of a branch. + +The state sent for an element must be the route's dependency inputs, read per element across aligned edges, with the element's key added under the reserved key "item". A mapped route must not depend on a task named "item", and validation must reject one that does. + +Any mapped task, a route or not, may declare keyed, a list of producer.field references. Each must name a field declared by one of the task's dependencies that is not mapped, and where the declaration types it, as "object" or by a schema whose top-level type is "object"; validation must reject any other, naming the reference, and must reject keyed on a task that does not declare over. For each keyed field, the inputs an instance receives, and for a route the state it sends, must carry in place of the whole field only the field's entry under the instance's key, read from the producer's output whatever the task's join; the producer's other fields are unchanged. A keyed value that is not an object at run time must fail every instance, naming the field, and an absent entry must fail that instance, naming the key and the field. keyed must reach the generated TOML, and the admitted-plan event must carry each task's keyed references as a list of "producer.field" strings, empty for a task that declares none, so a reader can draw which edges are narrowed. + +An output-decided route that declares over reads each element's answers from its source per element. Where the source is aligned with the route, the answers are read from the source's instance for the element, under the rules [[RFC-0001:C-ROUTE]] states for a source's output. Otherwise the source must not be mapped, and its output must carry, under each question id, an object keyed by element key whose entry for the element is the answer under those rules; validation must reject a source whose emits types a question's field as anything but "object" or an object schema, and an absent entry must fail that element. + +A when that names a mapped route must be declared by a task aligned with that route. Validation must reject a when on a mapped route declared by a task that is not mapped, or that maps over a different list, naming both references. Such a when is evaluated per element: the dependent's instance for an element applies the rules of [[RFC-0001:C-ROUTE]] to the route's instance for the same key, so it settles not taken, without dispatch and without spend, when that element's resolved label is not among the labels the when lists, and the instances for other elements are unaffected. The rules of [[RFC-0001:C-ROUTE]] for drop, otherwise, and label coverage apply to a mapped route's questions unchanged. Several tasks aligned with one mapped route may each declare a when on the same question, and so partition the elements among them by answer. A mapped route may itself declare a when on an aligned route, so it is asked only about the elements that route's answer leaves to it. A when on a route that does not declare over, declared by a mapped task, is evaluated once for the node under [[RFC-0001:C-ROUTE]]. + +A mapped task's instance that settled not taken must not count as failed. A mapped node with no instances must settle passing. Otherwise it must settle not taken when every instance was not taken, passing when every instance passed or was not taken, and failing otherwise. Its folded output must carry the number of instances not taken beside the numbers that passed and failed, and must carry no output for an instance not taken. The task-result event's fan-out summary must carry the same count under not_taken; a reader that predates it reads a summary without it as zero. + +Each element's decision is its instance's recorded output, and an instance settled not taken is terminal. On resume, under [[RFC-0002:C-PLAYBOOK-RESUME]], the engine must fold both, must not ask again for an element that has a recorded decision, and must ask only for the elements that have none.""" diff --git a/gov/rfc/RFC-0001/clauses/C-ROUTE-FILES.toml b/gov/rfc/RFC-0001/clauses/C-ROUTE-FILES.toml new file mode 100644 index 00000000..33e8bd9e --- /dev/null +++ b/gov/rfc/RFC-0001/clauses/C-ROUTE-FILES.toml @@ -0,0 +1,16 @@ +#:schema ../../../schema/clause.schema.json + +[govctl] +id = "C-ROUTE-FILES" +title = "Declared files in a route's state" +kind = "normative" +status = "active" +since = "0.2.0" + +[content] +text = """ +A route may declare files, a list of workspace-relative paths. Each path must be declared under emits_files by at least one of the route's dependencies with a JSON Schema ([[RFC-0002:C-TASK-FILES]]); validation must reject a path no dependency declares, a path every declaring dependency declares without a schema, and a path declared only by a mapped dependency that is not aligned with the route under [[RFC-0001:C-ROUTE-EACH]], naming the path. + +The state sent to the decision model must carry, under the reserved key "files", one entry per dependency that declares a listed path, under that dependency's name, mapping each such path to the parsed content of the file staged for the route. For a route mapped over a list, an aligned dependency's file is its instance's for the same element. A dependency whose file was not staged for the route contributes no entry for that path. A staged file that does not parse as JSON, as a failed dependency's file under a "settled" join may not, must contribute null in its place, and the engine must record the path on the route's result. A route must not depend on a task named "files", whether or not it declares files, and validation must reject one that does. A route that declares no files sends no "files" key. + +An output-decided route reads its answers from its source's JSON output and must reject files at compile time, because it asks no model.""" diff --git a/gov/rfc/RFC-0001/clauses/C-ROUTE.toml b/gov/rfc/RFC-0001/clauses/C-ROUTE.toml index 84086844..cb139332 100644 --- a/gov/rfc/RFC-0001/clauses/C-ROUTE.toml +++ b/gov/rfc/RFC-0001/clauses/C-ROUTE.toml @@ -29,6 +29,6 @@ A reviewer under [[RFC-0002:C-REVISE-LOOP]] must not declare when: a reviewer ru A task whose when is not satisfied must settle as not taken, without dispatch and without spend. A when over a route that itself settled not taken is not satisfied. A when over a route that settled any other way without passing is not satisfied either, except that a task joining "all" on that route is blocked by the failure as any dependent is. Not taken is a terminal status distinct from skipped. It must not count against validity, must not short-circuit the run, and must not be reported as a failure. A task joining "all" on a dependency that settled not taken must itself settle not taken, so an untaken branch is untaken to its leaves. A task joining "passed" or "settled" is dispatched as those joins already define, and is how branches rejoin. A required task is required when taken: if its when is satisfied it gates validity as any required task does. The workflow's result task must not be conditional and must not be able to settle not taken through "all" joins alone. -A route task must not declare over; routing each element of a list is not defined by this clause. +A route task may declare over. [[RFC-0001:C-ROUTE-EACH]] states how a mapped route decides each element of a list and how a when reads that decision; where it and this clause differ for a mapped route, it governs. The route and when constructs are admitted in the playbook and custom lanes. An autoresearch workflow must reject both: the keep rule is sourced from a frozen measure, and a model decision must not sit on the path to it.""" diff --git a/gov/rfc/RFC-0001/rfc.toml b/gov/rfc/RFC-0001/rfc.toml index c0db04ba..967ceaea 100644 --- a/gov/rfc/RFC-0001/rfc.toml +++ b/gov/rfc/RFC-0001/rfc.toml @@ -35,6 +35,8 @@ clauses = [ "clauses/C-CAPABILITY-DISCLOSURE.toml", "clauses/C-ROUTE.toml", "clauses/C-INFERENCE.toml", + "clauses/C-ROUTE-EACH.toml", + "clauses/C-ROUTE-FILES.toml", ] [[changelog]] @@ -67,6 +69,11 @@ changed = [ "skill joins the enumerated constructors: an agent task whose prompt is assembled from a shipped SKILL.md and the invocation's arguments", "a decision binding may speak decisions, the OpenAI Decisions API, reached at OpenAI's address when it names no URL (C-INFERENCE, C-ROUTE)", "the capability a model-decided route needs is decision, not systemone, since it names the role rather than one API (C-ROUTE)", + "a route may map over a list and decide each element on its own; a task reads a dependency mapped over the same list per element, a when on a mapped route is read per element, and an instance not taken does not fail its fan-out (C-ROUTE-EACH)", + "keyed: a mapped task may narrow an object field of an unmapped dependency to its own element's entry, and the admitted-plan event carries it (C-ROUTE-EACH)", + "a route may put declared JSON files of its dependencies in the decision model's state (C-ROUTE-FILES)", + "a route task may declare over (C-ROUTE)", + "a decision binding is never relayed into a sandbox or task process, so only a route asks the decision model (C-INFERENCE)", ] [[changelog]]