fix: preserve pending branches in update_state(as_node); reject unenforced steering blocks at compile time

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:
 

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).

docs/modules/expressive-language/reference.md

@@ -77,26 +77,44 @@ Supported fields:
 - `routes`
 - `retry`
 - `timeout`
-- `steering`
 
 Example:
 
 ```tinyagents
 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`
 

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

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<NodeId> = 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<NodeId>, Option<Vec<PendingActivation>>) =
+            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<Activation> = 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<NodeId> = 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<NodeId> = 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();