From 8cfb7329e8df391e1439b1f47989393d80861c23 Mon Sep 17 00:00:00 2001 From: Steven Enamakel Date: Sat, 8 Aug 2026 13:21:50 +0300 Subject: [PATCH 1/2] fix(graph): update_state(as_node) merges routed successors into remaining pending work instead of replacing it Co-authored-by: Medulla --- docs/modules/graph/checkpointing.md | 8 +- src/graph/compiled/state_api.rs | 169 +++++++++++-------- src/graph/compiled/test.rs | 252 ++++++++++++++++++++++++++++ 3 files changed, 357 insertions(+), 72 deletions(-) diff --git a/docs/modules/graph/checkpointing.md b/docs/modules/graph/checkpointing.md index 07e8e26..b2a3740 100644 --- a/docs/modules/graph/checkpointing.md +++ b/docs/modules/graph/checkpointing.md @@ -292,8 +292,12 @@ snapshot, its `parent_config`, the listing `metadata`, and any `update` is folded through the same `StateReducer` the executor uses, on top of the thread's latest committed state, and persisted as a new checkpoint with source `update`. `as_node` must name a real node (`MissingNode` otherwise); the - write is attributed to it and the new checkpoint's pending nodes become that - node's routing successors. With `as_node == None` the latest pending set is + write is attributed to it: the node is treated as just-completed (it leaves + the pending set) and its routing successors are merged into the base + checkpoint's remaining pending work, so branches the write never touched keep + running — with their `Send` args intact. A successor behind a waiting edge is + barrier-gated exactly as during a run, and the retained predecessors are what + later clear the join. With `as_node == None` the latest pending set is preserved. - `bulk_update_state(thread_id, updates)` — applies a sequence of `(update, as_node)` pairs as successive `update` checkpoints, each layered on diff --git a/src/graph/compiled/state_api.rs b/src/graph/compiled/state_api.rs index 52d4ab5..188d392 100644 --- a/src/graph/compiled/state_api.rs +++ b/src/graph/compiled/state_api.rs @@ -66,18 +66,25 @@ where /// [`StateReducer`](crate::graph::StateReducer) the executor uses, on top of /// the thread's latest committed state. When `as_node` is supplied it must /// name a real node (else [`TinyAgentsError::MissingNode`]); the write is - /// attributed to that node and the new checkpoint's pending nodes become that - /// node's routing successors (so a subsequent resume continues from after the - /// attributed node). A command node cannot be used as `as_node` (it routes - /// dynamically and has no static successors); doing so returns - /// [`TinyAgentsError::Graph`] rather than silently producing a non-resumable - /// checkpoint. A successor reached by a waiting edge is barrier-gated - /// exactly as it would be during a run: it is scheduled only once every - /// required predecessor has arrived (the write records the attributed - /// node's arrival), so a manual write can never fire a join ahead of a - /// still-pending branch. With `as_node == None` the latest pending node set is - /// preserved. Requires a configured checkpointer and an existing checkpoint - /// for the thread. + /// attributed to that node, which is treated as having just completed: it + /// leaves the pending set and its routing successors are *merged into* the + /// base checkpoint's remaining pending work (so a subsequent resume + /// continues from after the attributed node without dropping the branches + /// it never touched). Sibling branches keep their `Send` args, and a + /// successor that is already pending is not scheduled twice. When the + /// attributed node has several pending `Send` activations, the write + /// completes all of them at once — a manual write cannot name which packet + /// it stands for — and the successor is scheduled once. A command node + /// cannot be used as `as_node` (it routes dynamically and has no static + /// successors); doing so returns [`TinyAgentsError::Graph`] rather than + /// silently producing a non-resumable checkpoint. A successor reached by a + /// waiting edge is barrier-gated exactly as it would be during a run: it is + /// scheduled only once every required predecessor has arrived (the write + /// records the attributed node's arrival), so a manual write can never fire + /// a join ahead of a still-pending branch — and because the other pending + /// predecessors are retained, they still run and clear the join. With + /// `as_node == None` the latest pending node set is preserved. Requires a + /// configured checkpointer and an existing checkpoint for the thread. pub async fn update_state( &self, thread_id: &str, @@ -118,69 +125,91 @@ where // Manual writes preserve any accumulated barrier arrivals, and an // attributed write records its own arrival into them. let mut arrivals = barriers_from_persisted(&base.barrier_arrivals); - // Pending nodes: the attributed node's successors, or the inherited set. - let mut withheld_by_barrier = false; - // Set only when the base checkpoint's pending set was reinstated below, - // which is the one case where the pending *activations* must be - // reinstated with it. Keying that off `withheld_by_barrier` would let a - // routing that withholds one target while scheduling another persist - // `next_nodes` and `pending_activations` that disagree — and resume - // prefers the activations, silently dropping the scheduled successor. - let mut used_base_fallback = false; - let next_nodes: Vec = match &as_node { - Some(node) => { - let mut next = Vec::new(); - for target in self.route(node, None, &new_state)? { - let tnode = target.node().clone(); - if tnode.as_str() == END { - continue; - } - // Apply the same barrier gate the executor applies in - // `route_completed`: a waiting node stays unscheduled until - // every required predecessor has arrived. Without this an - // attributed write would fire a join ahead of a predecessor - // that is still pending — the data loss the waiting edge - // exists to prevent. - if let Some(required) = self.waiting.get(&tnode) { - let arrived = arrivals.entry(tnode.clone()).or_default(); - arrived.insert(node.clone()); - if !required.is_subset(arrived) { - withheld_by_barrier = true; + // Pending schedule: the attributed node's successors *merged into* the + // base checkpoint's still-pending work, or the inherited set verbatim. + // + // `next_nodes` and `pending_activations` are derived from one merged + // activation list so they can never disagree — resume prefers the + // activations, so a node named by only one of them would be silently + // dropped (or re-scheduled without its `Send` arg). + // + // The merge is unconditional rather than a fallback for the + // nothing-was-scheduled case. `route(node, None, ..)` resolves a static + // or conditional edge, so today it yields at most one target and a + // withheld barrier is the only way to end up with none — but keying the + // merge on that would silently drop the untouched branches the moment a + // single call ever resolves a withheld target *and* a schedulable one. + let (next_nodes, pending_activations): (Vec, Option>) = + match &as_node { + Some(node) => { + // The attributed node counts as completed, so it leaves the + // schedule; every other branch the base checkpoint had in + // flight (with its `Send` arg, when it carried one) stays. + let mut merged: Vec = match &base.pending_activations { + Some(pending) if !pending.is_empty() => pending + .iter() + .map(Activation::from) + .filter(|activation| activation.node != *node) + .collect(), + // Checkpoints written before `pending_activations` + // existed only carry the node-id projection. + _ => base + .next_nodes + .iter() + .filter(|pending| *pending != node) + .cloned() + .map(Activation::node) + .collect(), + }; + let mut seen: HashSet = merged + .iter() + .filter(|activation| activation.send_arg.is_none()) + .map(|activation| activation.node.clone()) + .collect(); + for target in self.route(node, None, &new_state)? { + let tnode = target.node().clone(); + if tnode.as_str() == END { continue; } - arrivals.remove(&tnode); + // Apply the same barrier gate the executor applies in + // `route_completed`: a waiting node stays unscheduled + // until every required predecessor has arrived. Without + // this an attributed write would fire a join ahead of a + // predecessor that is still pending — the data loss the + // waiting edge exists to prevent. The barrier's other + // predecessors are still scheduled (they are part of + // `merged` above), so they run and clear the join. + if let Some(required) = self.waiting.get(&tnode) { + let arrived = arrivals.entry(tnode.clone()).or_default(); + arrived.insert(node.clone()); + if !required.is_subset(arrived) { + continue; + } + arrivals.remove(&tnode); + } + // `Send` activations may legitimately repeat a node + // (each carries its own arg); plain ones are + // deduplicated so a successor already pending is not + // scheduled twice. + let send_arg = target.send_arg().cloned(); + if send_arg.is_some() || seen.insert(tnode.clone()) { + merged.push(Activation { + node: tnode, + send_arg, + }); + } } - next.push(tnode); - } - // An unsatisfied barrier leaves nothing to schedule, which would - // make the checkpoint non-resumable. Keep the base checkpoint's - // still-pending nodes (minus the attributed one) so the barrier's - // remaining predecessors still run and clear the join. - if next.is_empty() && withheld_by_barrier { - used_base_fallback = true; - next.extend(base.next_nodes.iter().filter(|n| *n != node).cloned()); + let nodes = activation_nodes(&merged); + let activations = if merged.is_empty() { + None + } else { + Some(merged.iter().map(PendingActivation::from).collect()) + }; + (nodes, activations) } - next - } - None => base.next_nodes.clone(), - }; + None => (base.next_nodes.clone(), base.pending_activations.clone()), + }; let completed_tasks: Vec = as_node.iter().cloned().collect(); - // With `as_node`, pending becomes that node's (plain) successors, so no - // send args carry over; without it, inherit the base checkpoint's - // pending activations verbatim so any pending `Send` args survive. The - // barrier-withheld fallback above re-schedules base pending nodes, so it - // keeps their activations (and `Send` args) too. - let pending_activations = match (&as_node, used_base_fallback) { - (Some(node), true) => base.pending_activations.as_ref().map(|pending| { - pending - .iter() - .filter(|activation| activation.node != *node) - .cloned() - .collect() - }), - (Some(_), false) => None, - (None, _) => base.pending_activations.clone(), - }; let barrier_arrivals = barriers_to_persisted(&arrivals); let checkpoint_id = next_checkpoint_id(); diff --git a/src/graph/compiled/test.rs b/src/graph/compiled/test.rs index a8e308e..b6537a6 100644 --- a/src/graph/compiled/test.rs +++ b/src/graph/compiled/test.rs @@ -2604,6 +2604,258 @@ async fn attributed_update_does_not_fire_an_unsatisfied_barrier() { assert_eq!(done.state.value, 113); } +/// Fan-out graph used by the attributed-write scheduling tests: `super` forks +/// into `b -> x -> y` and `c`, where `c` interrupts on its first activation. +/// The pause therefore leaves a checkpoint with two independent pending +/// branches (`x` from the completed `b`, and the still-interrupted `c`). +fn forked_interrupt_graph( + cp: Arc>, + interrupted: Arc, +) -> CompiledGraph { + GraphBuilder::::new() + .with_parallel(true) + .set_reducer(ClosureStateReducer::new(|mut s: Counter, u: i32| { + s.value += u; + s.log.push(format!("+{u}")); + Ok(s) + })) + .add_node("super", |_s: Counter, _c: NodeContext| async move { + Ok(NodeResult::Command( + Command::default().with_goto(["b", "c"]), + )) + }) + .add_node("b", |_s: Counter, _c: NodeContext| async move { + Ok(NodeResult::Update(1)) + }) + .add_node("x", |_s: Counter, _c: NodeContext| async move { + Ok(NodeResult::Update(20)) + }) + .add_node("y", |_s: Counter, _c: NodeContext| async move { + Ok(NodeResult::Update(40)) + }) + .add_node("c", move |_s: Counter, _c: NodeContext| { + let once = interrupted.clone(); + async move { + if once.swap(true, AtomicOrdering::SeqCst) { + Ok(NodeResult::Update(2)) + } else { + Ok(NodeResult::Interrupt(Interrupt::new("c", json!({})))) + } + } + }) + .set_entry("super") + .mark_command_routing("super") + .add_sequence(["b", "x", "y"]) + .set_finish("y") + .set_finish("c") + .compile() + .unwrap() + .with_checkpointer(cp) +} + +#[tokio::test] +async fn attributed_update_keeps_other_pending_branches_scheduled() { + // Two independent branches are pending (`x` and the interrupted `c`). A + // manual write attributed to `x` schedules x's successor `y`, but it must + // not discard `c`: the attributed node's successors *add to* the schedule + // rather than replacing it, or the untouched branch is silently dropped and + // never runs again. + let cp = Arc::new(InMemoryCheckpointer::::new()); + let graph = forked_interrupt_graph(cp.clone(), Arc::new(AtomicBool::new(false))); + + let paused = graph + .run_with_thread( + "t-fork-update", + Counter { + value: 0, + log: vec![], + }, + ) + .await + .unwrap(); + assert!(paused.is_interrupted()); + + let before = cp.get("t-fork-update", None).await.unwrap().unwrap(); + assert!( + before.next_nodes.iter().any(|n| n.as_str() == "x") + && before.next_nodes.iter().any(|n| n.as_str() == "c"), + "precondition: both branches pending, got {:?}", + before.next_nodes + ); + + graph + .update_state("t-fork-update", 10, Some(NodeId::from("x"))) + .await + .unwrap(); + let written = cp.get("t-fork-update", None).await.unwrap().unwrap(); + assert!( + written.next_nodes.iter().any(|n| n.as_str() == "y"), + "the attributed node's successor must be scheduled, got {:?}", + written.next_nodes + ); + assert!( + written.next_nodes.iter().any(|n| n.as_str() == "c"), + "the untouched pending branch must stay scheduled, got {:?}", + written.next_nodes + ); + assert!( + !written.next_nodes.iter().any(|n| n.as_str() == "x"), + "the attributed node itself is completed, not pending: {:?}", + written.next_nodes + ); + // Resume prefers `pending_activations` over `next_nodes`, so the two must + // never disagree. + if let Some(pending) = &written.pending_activations { + assert_eq!( + pending.iter().map(|a| a.node.clone()).collect::>(), + written.next_nodes, + "pending activations and next nodes must describe the same schedule" + ); + } + + let done = graph.retry("t-fork-update").await.unwrap(); + assert!( + done.visited.iter().any(|n| n.as_str() == "c"), + "the dropped branch must still run, visited {:?}", + done.visited + ); + // 1 (b) + 10 (manual write) + 2 (c) + 40 (y). + assert_eq!(done.state.value, 53); +} + +#[tokio::test] +async fn attributed_update_to_sink_node_keeps_other_pending_branches() { + // `c` is terminal, so an attributed write to it schedules nothing of its + // own. Replacing the pending set with that empty routing would drop the + // sibling `x` branch *and* leave a checkpoint with nothing to resume. + let cp = Arc::new(InMemoryCheckpointer::::new()); + let graph = forked_interrupt_graph(cp.clone(), Arc::new(AtomicBool::new(false))); + + let paused = graph + .run_with_thread( + "t-fork-sink", + Counter { + value: 0, + log: vec![], + }, + ) + .await + .unwrap(); + assert!(paused.is_interrupted()); + + graph + .update_state("t-fork-sink", 10, Some(NodeId::from("c"))) + .await + .unwrap(); + let written = cp.get("t-fork-sink", None).await.unwrap().unwrap(); + assert_eq!( + written + .next_nodes + .iter() + .map(|n| n.to_string()) + .collect::>(), + vec!["x".to_string()], + "the sibling branch must survive an attributed write to a sink node" + ); + + let done = graph.retry("t-fork-sink").await.unwrap(); + // 1 (b) + 10 (manual write) + 20 (x) + 40 (y); `c` is attributed as done. + assert_eq!(done.state.value, 71); +} + +#[tokio::test] +async fn attributed_update_preserves_pending_send_args_of_other_branches() { + // Three `Send` activations of `worker` are pending behind an interrupt. A + // write attributed to an unrelated node must carry them over *with* their + // args — dropping them loses the fanout, and re-scheduling them by node id + // alone loses each packet's payload. + let cp = Arc::new(InMemoryCheckpointer::::new()); + let graph = GraphBuilder::::new() + .with_parallel(true) + .set_reducer(ClosureStateReducer::new(|mut s: Counter, u: i32| { + s.value += u; + s.log.push(format!("w:{u}")); + Ok(s) + })) + .add_node("dispatch", |_s: Counter, _c: NodeContext| async move { + Ok(NodeResult::Command(Command::send([ + Send::new("worker", json!(1)), + Send::new("worker", json!(2)), + Send::new("worker", json!(3)), + ]))) + }) + .add_node("worker", |_s: Counter, c: NodeContext| async move { + let arg = c + .send_arg + .clone() + .expect("worker scheduled via Send must carry its arg") + .as_i64() + .unwrap() as i32; + if arg == 1 && c.resume.is_none() { + return Ok(NodeResult::Interrupt(Interrupt::new("worker", json!({})))); + } + Ok(NodeResult::Update(arg)) + }) + .add_node("side", |_s: Counter, _c: NodeContext| async move { + Ok(NodeResult::Update(0)) + }) + .add_node("tail", |_s: Counter, _c: NodeContext| async move { + Ok(NodeResult::Update(0)) + }) + .set_entry("dispatch") + .mark_command_routing("dispatch") + .add_edge("side", "tail") + .set_finish("worker") + .set_finish("tail") + .compile() + .unwrap() + .with_checkpointer(cp.clone()); + + let paused = graph + .run_with_thread( + "t-send-update", + Counter { + value: 0, + log: vec![], + }, + ) + .await + .unwrap(); + assert!(paused.is_interrupted()); + + graph + .update_state("t-send-update", 0, Some(NodeId::from("side"))) + .await + .unwrap(); + let written = cp.get("t-send-update", None).await.unwrap().unwrap(); + let pending = written + .pending_activations + .clone() + .expect("an attributed write must persist the merged activations"); + let mut args: Vec = pending + .iter() + .filter(|a| a.node.as_str() == "worker") + .map(|a| { + a.send_arg + .as_ref() + .expect("pending Send activations keep their arg") + .as_i64() + .unwrap() + }) + .collect(); + args.sort_unstable(); + assert_eq!(args, vec![1, 2, 3], "every pending Send packet survives"); + assert!( + pending.iter().any(|a| a.node.as_str() == "tail"), + "the attributed node's successor is scheduled alongside them" + ); + assert_eq!( + pending.iter().map(|a| a.node.clone()).collect::>(), + written.next_nodes, + "pending activations and next nodes must describe the same schedule" + ); +} + #[tokio::test] async fn async_durability_drains_background_writes_on_abort() { // The recursion-limit abort returns `Err` mid-run. Any in-flight background From aea4de73bdb4335e960d61f60faa282cb6165651 Mon Sep 17 00:00:00 2001 From: Steven Enamakel Date: Sat, 8 Aug 2026 13:21:50 +0300 Subject: [PATCH 2/2] fix(language): reject steering blocks at compile time until the runtime can enforce them Co-authored-by: Medulla --- docs/modules/expressive-language/README.md | 8 ++- .../implementation-status.md | 9 ++- docs/modules/expressive-language/reference.md | 36 +++++++++--- src/language/ast.rs | 16 ++++-- src/language/compiler.rs | 22 +++++++ src/language/parser.rs | 6 ++ src/language/test/extended_grammar.rs | 57 ++++++++++++++++++- 7 files changed, 135 insertions(+), 19 deletions(-) diff --git a/docs/modules/expressive-language/README.md b/docs/modules/expressive-language/README.md index e86e327..3a7ac20 100644 --- a/docs/modules/expressive-language/README.md +++ b/docs/modules/expressive-language/README.md @@ -251,6 +251,10 @@ checkpoint_decl = "checkpoint" ident node_ref = ident | "END" ``` +`steering_decl` is reserved grammar only: it parses, and the compiler then +rejects it, because no faithful lowering onto the runtime steering policy exists +yet. See the `subagent` section of [`reference.md`](reference.md). + ## AST ```rust @@ -371,8 +375,8 @@ Required errors: - checkpoint policy incompatible with interrupts - state channel missing reducer - send target missing input mapping -- steering target not allowed -- steering policy references unknown actor or capability +- steering target not allowed (future — today the compiler rejects `steering` blocks wholesale) +- steering policy references unknown actor or capability (future — same) Example diagnostic: diff --git a/docs/modules/expressive-language/implementation-status.md b/docs/modules/expressive-language/implementation-status.md index 8edf335..de3afe9 100644 --- a/docs/modules/expressive-language/implementation-status.md +++ b/docs/modules/expressive-language/implementation-status.md @@ -84,7 +84,14 @@ against the registered scripts. ## Not yet implemented - State-schema declarations (`state Name { … }`). -- Steering policy lowering for `subagent` nodes (parsed shape only is partial). +- Steering policy lowering for `subagent` nodes. The `steering { … }` block + parses (the grammar reserves the shape), but the compiler **rejects** any node + that declares one rather than discarding it silently: the runtime + `harness::steering::SteeringPolicy` is a single flat command allowlist with no + `parent`/`human` actor separation, no delivery policy, and no + `add_instruction`/`request_status` commands, so no faithful lowering exists. + Build the `SteeringPolicy` in the Rust `NodeFactory` instead. See + `reference.md`, `subagent` section. - Duration literals like `60s` (write timeouts as a number or quoted string). - Formatter and round-trip golden tests (milestone L8). - Agent-authored review gates and blueprint provenance (milestone L7). diff --git a/docs/modules/expressive-language/reference.md b/docs/modules/expressive-language/reference.md index a3cdc71..6aa5aa6 100644 --- a/docs/modules/expressive-language/reference.md +++ b/docs/modules/expressive-language/reference.md @@ -77,7 +77,6 @@ Supported fields: - `routes` - `retry` - `timeout` -- `steering` Example: @@ -85,18 +84,37 @@ Example: node research { kind subagent agent "researcher" - steering { - parent allow ["add_instruction", "request_status", "cancel"] - human allow ["add_instruction", "pause", "resume", "cancel"] - delivery "safe_boundary" - } next synthesize } ``` -Steering policies lower into harness steering policy and graph task policy. They -can narrow a child agent's model/tool/runtime limits but cannot grant -capabilities absent from the registry or parent run policy. +#### `steering` — reserved, rejected by the compiler + +```tinyagents +steering { + parent allow ["add_instruction", "request_status", "cancel"] + human allow ["add_instruction", "pause", "resume", "cancel"] + delivery "safe_boundary" +} +``` + +This block **parses** — the grammar reserves the shape above — but `compile` +**rejects** any node that carries it, with a `TinyAgentsError::Compile` +diagnostic. It is not enforced, and it is deliberately not accepted-and-ignored: +a silently discarded policy would let an operator deploy a blueprint believing a +child agent's steering is restricted when the runtime receives no restriction at +all. + +There is no faithful lowering yet. `harness::steering::SteeringPolicy` is a +single flat allowlist of `SteeringCommandKind`s (`pause`, `resume`, `cancel`, +`inject_message`, `redirect`, `set_metadata`); it has no `parent`/`human` actor +separation, no delivery policy, and no `add_instruction` or `request_status` +command — so three of the four elements in the block above have no runtime +counterpart. + +Until declarative steering is implemented end to end, restrict a child agent by +building the `SteeringPolicy` in the Rust `NodeFactory` that materialises the +node, where the policy is actually attached to the run's `SteeringHandle`. ### `repl_agent` diff --git a/src/language/ast.rs b/src/language/ast.rs index eb31257..72c1d2a 100644 --- a/src/language/ast.rs +++ b/src/language/ast.rs @@ -167,10 +167,18 @@ pub struct NodeDecl { /// or a human may send to this `subagent` node, plus a delivery policy. /// /// Parsing accepts the shape documented in -/// `docs/modules/expressive-language/reference.md`; lowering it into a -/// runtime `harness::steering` policy is not yet implemented (tracked -/// separately), so [`crate::language::compiler::compile`] currently parses -/// and retains this declaration without acting on it. +/// `docs/modules/expressive-language/reference.md`, but lowering it into a +/// runtime `harness::steering` policy is not yet implemented: that policy is a +/// single flat allowlist of +/// [`SteeringCommandKind`](crate::harness::steering::SteeringCommandKind)s with +/// no `parent`/`human` actor separation, no delivery policy, and no +/// `add_instruction` / `request_status` commands. +/// +/// Because a partial lowering would silently weaken the declared restrictions, +/// [`crate::language::compiler::compile`] **rejects** any node carrying this +/// declaration with a [`TinyAgentsError::Compile`](crate::error::TinyAgentsError::Compile) +/// diagnostic. The AST node exists so the documented grammar still parses (and +/// so tooling can read the declaration), not because compilation accepts it. #[derive(Clone, Debug, Default, PartialEq)] pub struct SteeringDecl { /// Steering commands a parent orchestrator run may send (`parent allow [...]`). diff --git a/src/language/compiler.rs b/src/language/compiler.rs index ea22609..2b445a9 100644 --- a/src/language/compiler.rs +++ b/src/language/compiler.rs @@ -267,6 +267,28 @@ fn compile_graph(graph: &crate::language::types::GraphDecl) -> Result } } + // Reject a `steering { … }` declaration rather than dropping it on the + // floor. The grammar accepts the documented shape (parser.rs), but + // there is no faithful lowering onto the runtime today: + // `harness::steering::SteeringPolicy` is a single flat allowlist of + // `SteeringCommandKind`s with no `parent`/`human` actor separation, no + // delivery policy, and no `add_instruction` / `request_status` kinds. + // Any partial lowering would have to widen or invent semantics, and a + // silent no-op is worse still: an operator would deploy a blueprint + // believing the declared restrictions are enforced while the runtime + // receives none. Fail loudly until declarative steering is implemented + // end to end. + if node.steering.is_some() { + return Err(compile_err(format!( + "node `{}` declares a `steering {{ … }}` block, but declarative steering is parsed and not yet enforced: \ +the runtime policy (`harness::steering::SteeringPolicy`) is one flat command allowlist with no `parent`/`human` actor \ +separation, no delivery policy, and no `add_instruction`/`request_status` commands, so the declaration cannot be lowered \ +faithfully. Remove the block and apply a `SteeringPolicy` from the Rust `NodeFactory` that builds this node until \ +declarative steering lowering lands.", + node.name + ))); + } + // Determine routing. Precedence: explicit `routes` > `next` > command // `goto` > top-level edge > terminal. let routing = if has_routes { diff --git a/src/language/parser.rs b/src/language/parser.rs index de84f85..f3a420d 100644 --- a/src/language/parser.rs +++ b/src/language/parser.rs @@ -553,6 +553,12 @@ impl Parser<'_> { /// Parses a `steering { parent allow [...] human allow [...] delivery "…" }` /// block. The `steering` keyword has already been consumed. + /// + /// The block is reserved grammar: it parses into a + /// [`SteeringDecl`](crate::language::ast::SteeringDecl), and + /// [`compile`](crate::language::compiler::compile) then rejects any node + /// carrying one, because no faithful lowering onto + /// [`SteeringPolicy`](crate::harness::steering::SteeringPolicy) exists yet. fn parse_steering_block(&mut self) -> Result { self.expect(&Token::LBrace)?; let mut steering = SteeringDecl::default(); diff --git a/src/language/test/extended_grammar.rs b/src/language/test/extended_grammar.rs index 773f29c..d357d2a 100644 --- a/src/language/test/extended_grammar.rs +++ b/src/language/test/extended_grammar.rs @@ -381,10 +381,61 @@ fn steering_block_on_subagent_node_parses() { vec!["add_instruction", "pause", "resume", "cancel"] ); assert_eq!(steering.delivery.as_deref(), Some("safe_boundary")); +} + +#[test] +fn steering_block_is_rejected_at_compile_time_until_it_is_enforced() { + // Parsing the documented shape must not imply enforcement. Nothing lowers + // `SteeringDecl` into `harness::steering::SteeringPolicy` (which has no + // parent/human actor split, no delivery policy, and no + // `add_instruction`/`request_status` kinds), so compilation fails loudly + // rather than deploying a blueprint whose steering restrictions exist only + // in the source text. + let src = r#" + graph g { + start research + node research { + kind subagent + agent "researcher" + steering { + parent allow ["cancel"] + human allow ["pause", "resume", "cancel"] + delivery "safe_boundary" + } + next END + } + } + "#; + let program = parse_str(src).unwrap(); + let err = compile(&program).unwrap_err(); + match err { + crate::error::TinyAgentsError::Compile(message) => { + assert!(message.contains("node `research`"), "{message}"); + assert!(message.contains("steering"), "{message}"); + assert!(message.contains("not yet enforced"), "{message}"); + assert!(message.contains("NodeFactory"), "{message}"); + } + other => panic!("expected a compile error, got {other:?}"), + } +} - // The declaration survives compilation (lowering into the blueprint stays - // a no-op; the point is that documented source is no longer rejected). - compile(&program).unwrap(); +#[test] +fn subagent_node_without_steering_still_compiles() { + // The rejection above is scoped to the `steering` block itself: an + // otherwise identical `subagent` node must keep compiling, and the + // compiled spec carries no steering field to mistake for a policy. + let src = r#" + graph g { + start research + node research { + kind subagent + agent "researcher" + next END + } + } + "#; + let blueprint = compile(&parse_str(src).unwrap()).unwrap().remove(0); + assert_eq!(blueprint.nodes[0].agent.as_deref(), Some("researcher")); } #[test]