Merge pull request #24 from Dark-Alex-17/feat/map-subgraph-branches

feat(graph): map branches as per-item subgraph chains + templated max_concurrency

assets/agents/deep-research/graph.yaml

@@ -13,7 +13,7 @@ description: |
   node type (script, llm, rag, map, agent, input, approval, end) and
   both static fan-out and dynamic map fan-out.
 
-version: "1.0"
+version: '1.0'
 
 global_tools:
   - web_search_coyote.sh
@@ -24,8 +24,8 @@ mcp_servers:
   - ddg-search
 
 conversation_starters:
-  - "How does HTTP/3 differ from HTTP/2?"
-  - "Summarize recent advances in solid-state battery chemistry"
+  - 'How does HTTP/3 differ from HTTP/2?'
+  - 'Summarize recent advances in solid-state battery chemistry'
 
 settings:
   max_loop_iterations: 40
@@ -34,15 +34,14 @@ settings:
   max_concurrency: 4
 
 initial_state:
-  research_feedback: ""
+  research_feedback: ''
   research_attempts: 0
-  local_context: ""
-  local_sources: ""
+  local_context: ''
+  local_sources: ''
 
 start: parse_request
 
 nodes:
-
   parse_request:
     id: parse_request
     type: script
@@ -52,10 +51,10 @@ nodes:
   ask_topic:
     id: ask_topic
     type: input
-    question: "What would you like me to research?"
-    validation: "len(input) > 0"
+    question: 'What would you like me to research?'
+    validation: 'len(input) > 0'
     state_updates:
-      topic: "{{input}}"
+      topic: '{{input}}'
     next: bootstrap_research
 
   bootstrap_research:
@@ -77,7 +76,7 @@ nodes:
       are precise, self-contained queries (each one is sent on its own
       to a separate research worker, so they must be answerable
       without each other's context).
-    prompt: "Research topic: {{topic}}"
+    prompt: 'Research topic: {{topic}}'
     tools: []
     output_schema:
       type: object
@@ -99,22 +98,23 @@ nodes:
     type: rag
     documents:
       - ./knowledge/
-    query: "{{topic}}"
+    query: '{{topic}}'
     top_k: 6
     chunk_size: 1000
     chunk_overlap: 100
     state_updates:
-      local_context: "{{output.context}}"
-      local_sources: "{{output.sources}}"
+      local_context: '{{output.context}}'
+      local_sources: '{{output.sources}}'
     next: research_each_question
 
   research_each_question:
     id: research_each_question
     type: map
-    over: "{{questions}}"
+    over: '{{questions}}'
     as: question
     branch: research_one_question
     collect_into: question_findings
+    output_key: finding
     max_concurrency: 3
     next: combine_findings
 
@@ -151,7 +151,8 @@ nodes:
       - mcp:ddg-search
     max_iterations: 10
     max_attempts: 2
-    temperature: 0.1
+    state_updates:
+      finding: '{{output}}'
 
   combine_findings:
     id: combine_findings
@@ -177,7 +178,7 @@ nodes:
       - classify_source
     max_iterations: 15
     state_updates:
-      source_assessment: "{{output}}"
+      source_assessment: '{{output}}'
     next: critique
 
   critique:
@@ -217,7 +218,7 @@ nodes:
       {{source_assessment}}
     tools: []
     state_updates:
-      critique: "{{output}}"
+      critique: '{{output}}'
     next: reflexion_gate
 
   reflexion_gate:
@@ -242,7 +243,7 @@ nodes:
       Produce the final report following your instructions.
     timeout: 300
     state_updates:
-      report: "{{output}}"
+      report: '{{output}}'
     next: verify_sources
 
   verify_sources:
@@ -266,14 +267,14 @@ nodes:
       Accept this report? Pick "accept" or "reject", or type specific
       feedback to send the research back for another pass.
     options:
-      - "accept"
-      - "reject"
+      - 'accept'
+      - 'reject'
     routes:
-      "accept": end_accepted
-      "reject": end_rejected
+      'accept': end_accepted
+      'reject': end_rejected
     on_other: incorporate_feedback
     state_updates:
-      decision: "{{choice}}"
+      decision: '{{choice}}'
 
   incorporate_feedback:
     id: incorporate_feedback
@@ -283,7 +284,7 @@ nodes:
   end_accepted:
     id: end_accepted
     type: end
-    output: "{{report}}"
+    output: '{{report}}'
 
   end_rejected:
     id: end_rejected

assets/agents/finding-verifier/graph.yaml

@@ -27,9 +27,6 @@ settings:
   log_state_snapshots: true
   timeout: 900
 
-reducers:
-  output: overwrite
-
 initial_state:
   findings_list: []
   verdicts: []
@@ -95,6 +92,7 @@ nodes:
     as: finding
     branch: verify_one
     collect_into: verdicts
+    output_key: verdict
     max_concurrency: 4
     next: done
 
@@ -149,7 +147,8 @@ nodes:
       - fs_grep
     max_iterations: 8
     max_attempts: 2
-    temperature: 0.1
+    state_updates:
+      verdict: '{{output}}'
 
   done:
     id: done

assets/agents/librarian/graph.yaml

@@ -32,9 +32,6 @@ settings:
   log_state_snapshots: true
   timeout: 600
 
-reducers:
-  output: overwrite
-
 initial_state:
   language_ecosystem: 'general'
   doc_domain_hints: ''

graph.example.yaml

@@ -116,9 +116,12 @@ variables:
 # ---------------------------------------------------------------------------
 settings:
   max_loop_iterations: 100   # Per-node visit cap. If one node id is entered more
-                             # than this many times, execution aborts. Default 100.
+                             # than this many times, execution aborts. Default 100; 0 = no cap.
   timeout: 600               # Optional wall-clock cap (seconds) on the whole run,
-                             # checked between node transitions.
+                             # checked between node transitions. 0 = no bound.
+                             # Ctrl-C (or the enclosing turn aborting) ends the run
+                             # promptly from anywhere: in-flight branches, tools, and
+                             # child graph agents are cancelled with it.
   log_state_snapshots: true  # Log state before each node (debug/trace). Default true.
   validate_before_run: true  # Run the graph validator at startup. Default true.
   max_concurrency: 4         # Cap on simultaneously running branches in any
@@ -242,7 +245,7 @@ nodes:
       - https://example.com/reference
     query: "{{topic}}"         # Retrieval query (templated). Default: {{initial_prompt}}.
     top_k: 5                   # Chunks to retrieve. Default = the KB's own top_k.
-    timeout: 120               # Retrieval timeout in seconds. Default 120.
+    timeout: 120               # Retrieval timeout in seconds. Default 120; 0 = no bound.
     # Knowledge-base build config (optional; used only when the KB is first
     # built). When embedding_model + chunk_size + chunk_overlap are all set,
     # the KB builds with no interactive prompts (works in non-interactive runs).
@@ -311,7 +314,7 @@ nodes:
     id: synthesize
     type: script
     script: scripts/synthesize.py  # Path relative to the agent directory
-    timeout: 30                # Seconds. Default 30.
+    timeout: 30                # Seconds. Default 30; 0 = no bound.
     state_updates:             # Applied after the stdout JSON is merged
       decided_for: "{{topic}}"
     next: summarize            # Default route if the script emits no `_next`
@@ -333,7 +336,21 @@ nodes:
     prompt: |                  # User message sent to the child (templated)
       Research {{topic}} in depth. Existing context:
       {{context}}
-    timeout: 600               # Optional wall-clock cap, seconds. Default 300.
+    # inputs:                  # Optional. Only valid when `agent` is a GRAPH agent
+    #   max_parallel: "{{n}}"  # (has a graph.yaml; deep-research is one). Each value
+    #   context: "Repo: {{r}}" # is a template against THIS graph's state. The resolved
+    #                          # map is overlaid on the child's `initial_state` before
+    #                          # its graph starts, so the child can read {{max_parallel}}
+    #                          # and {{context}} from its first node on. A lone {{key}}
+    #                          # keeps the value's JSON type (numbers stay numbers, lists
+    #                          # stay lists, null overlays the child's default); mixed text
+    #                          # becomes a string. `initial_prompt` is reserved (always
+    #                          # seeded from `prompt:`) and is rejected as an input key.
+    #                          # `inputs:` on a config-only agent is a load-time error,
+    #                          # and `inputs` values count as reads for the cross-branch
+    #                          # read check. See the wiki: Graph Agents -> "Passing state
+    #                          # into a child graph agent (inputs)".
+    timeout: 600               # Optional wall-clock cap, seconds. Default 300; 0 = no bound.
     teammates: false           # Optional. When true and this node runs concurrently
                                # with other flagged agent nodes (map fan-out branches
                                # or the same super-step), the siblings can message
@@ -343,7 +360,13 @@ nodes:
                                # super-step frontier form two separate teams and cannot
                                # message each other. Peers only span a single map
                                # node's branches or a single super-step's flagged
-                               # agent nodes.
+                               # agent nodes. Inside a map branch chain the identity
+                               # belongs to the ITEM, not the node: it lives for the
+                               # item's whole chain, is handed to each `teammates: true`
+                               # agent step in that chain (unflagged steps never see it),
+                               # and is retired when the chain ends. A message delivered
+                               # after the chain's last flagged agent step is accepted
+                               # but never read.
     output_schema:             # Optional. Same extraction as llm nodes
       type: object
       properties:
@@ -357,9 +380,20 @@ nodes:
     next: review               # Required for agent nodes
 
   # --- map node (Dynamic fan-out. Think: LangGraph's `Send` API) ----------------
-  # Spawns one parallel sub-branch per item in `over`. Each sub-branch runs
-  # the node referenced by `branch:` with the item bound to `as:`. Outputs
-  # collect into the array named by `collect_into:`, preserving input order.
+  # Spawns one parallel sub-branch per item in `over`. Each sub-branch runs a
+  # per-item CHAIN of nodes on a private fork of the state: it starts at the
+  # node referenced by `branch:` (with the item bound to `as:`) and follows
+  # that node's `next:` / script `_next` / `fallback` edges until it reaches a
+  # node with no `next` and no `_next`. When every chain has finished, the
+  # value of `output_key` from each item's final fork state is collected into
+  # the array named by `collect_into:`, preserving input order. Everything
+  # else the chain wrote is fork-local scratch and is discarded at collect.
+  #
+  # Not supported inside a branch chain (validator errors): fan-out `next:`
+  # lists, another `map`, `end`, `approval`, `input`. A branch node must not
+  # also be reachable from `start` on the main flow. Per-item loops are script
+  # `_next` loops (bounded by settings.max_loop_iterations on the fork; `0` = unbounded); a
+  # declared `next`/`fallback` edge back to an earlier chain node is a cycle.
   #
   # Reach via `synthesize`'s `_next: subjects_map`. The producer is expected
   # to have written a list at `subjects` (e.g. an upstream LLM node with an
@@ -371,31 +405,53 @@ nodes:
                                 # Empty list is allowed. It means no branches spawn,
                                 # and thus `collect_into` is written as [].
     as: subject                 # Required. Per-branch state key holding the
-                                # current item. Read with {{subject}} inside
-                                # the branch node's prompt.
-    branch: research_subject    # Required. Node id to invoke per item.
-                                # Must point to an llm | agent | rag | script
-                                # node satisfying the map branch contract:
-                                #   - no `next:` (atomic, joined at map exit)
-                                #   - no `state_updates:` other than via the
-                                #     map's `collect_into` channel
-                                #   - no `output_schema:` (top-level merge
-                                #     would clash with collect_into)
-                                # Validator enforces all three.
+                                # current item. Read with {{subject}} from any
+                                # node in the branch chain. Must differ from
+                                # `output_key` (load-time error otherwise).
+    branch: research_subject    # Required. ENTRY node of the per-item chain.
+                                # Every node reachable from it via `next:` /
+                                # `fallback:` is part of the chain and must be
+                                # an llm | agent | rag | script node. Chain
+                                # nodes may use `next:`, `state_updates:`
+                                # (fork-local scratch such as a draft or a
+                                # retry counter), `output_schema:`, and
+                                # `fallback:` (honored per item; it must point
+                                # at a node inside the chain, never at an
+                                # `end_*` node). A script's `_next` must also
+                                # stay inside the chain; routing out of it
+                                # fails the item.
     collect_into: subject_findings  # Required. State key for the array of
                                     # per-branch outputs, in input order
                                     # (not spawn-finish order).
-    max_concurrency: 3          # Optional per-map cap. Defaults to
-                                # settings.max_concurrency above.
-    output_key: output          # Optional. State key the branch's output
-                                # appears under. Default "output". Useful
-                                # only if the branch reads its own bound
-                                # name back (rare).
+    max_concurrency: 3          # Optional cap on items in flight (one slot is
+                                # held for an item's WHOLE chain). Defaults to
+                                # settings.max_concurrency above. Accepts an
+                                # integer or a template that is exactly one
+                                # "{{key}}" (surrounding whitespace allowed,
+                                # nothing else) and resolves to a positive integer,
+                                # e.g. `max_concurrency: "{{research_budget}}"`.
+    output_key: subject_summary # Optional. State key whose value in the item's
+                                # FINAL fork state becomes the item's result.
+                                # Default "output"; must differ from `as:`. The
+                                # key is removed from the fork before the chain
+                                # starts, so the chain
+                                # must write it (via state_updates, an
+                                # output_schema property, or a script's JSON);
+                                # an item that never writes it is an error, and
+                                # a chain that reads it before writing it sees
+                                # it absent (prompts fail, state_updates render
+                                # "", scripts get no key) - seed per-item input
+                                # through `as:` instead. Prefer a named key: with
+                                # the default, the collector errors if the chain
+                                # never writes it.
     next: aggregate_subjects    # Where to go after all sub-branches finish.
 
-  # Branch node for subjects_map. Each invocation receives a different
-  # `subject` in state. The branch is "atomic", meaning it cannot route on
-  # its own; the surrounding `map` joins after all invocations finish.
+  # Entry node of subjects_map's per-item chain. Each item gets its own copy
+  # of this node with a different `subject` in state. With no `next:` it is
+  # also the LAST chain node, so the chain is a single step; give it `next:`
+  # (or emit `_next` from a script step) to extend the chain, e.g.
+  # research -> verify -> gate-script that re-emits `_next: research_subject`
+  # until the draft passes or a retry counter in fork state runs out.
   research_subject:
     id: research_subject
     type: llm
@@ -403,8 +459,11 @@ nodes:
     prompt: "Research {{subject}}: pull the key facts and one citation."
     tools:
       - web_search_coyote
-    # No `next:`, `state_updates:`, or `output_schema:` here. Map branches
-    # have a strict contract (see `subjects_map.branch` comment).
+    # Recommended pattern: write the completion to a named key via
+    # `state_updates` and point the map's `output_key` at that same key.
+    # `{{output}}` is bound only while these templates render.
+    state_updates:
+      subject_summary: "{{output}}"
 
   # Aggregator that runs after the map joins. Reads the collected list.
   aggregate_subjects:
@@ -434,9 +493,9 @@ nodes:
     temperature: 0.3           # Optional per-node sampling override
     reasoning_effort: null     # Optional per-node reasoning effort override (e.g. low, medium, high)
     max_attempts: 2            # Retry count on transient errors only. Default 1.
-    max_iterations: 10         # Tool-call-loop turn cap. Default 10.
+    max_iterations: 10         # Tool-call-loop turn cap. Default 10; 0 = no cap.
     fallback: review           # Route here if all attempts fail
-    timeout: 300               # Optional node wall-clock cap, seconds (unset = no timeout)
+    timeout: 300               # Optional node wall-clock cap, seconds (unset or 0 = no timeout)
     state_updates:
       research: "{{output}}"
     next: review               # Required for llm nodes: the success route

src/config/agent.rs

@@ -633,6 +633,7 @@ impl Agent {
             Some(self.name().to_string()),
             tool_timeout,
             false,
+            None,
         )?;
         match value {
             Some(v) => Ok(v),

src/config/request_context.rs

@@ -8076,7 +8076,28 @@ mod tests {
     #[serial]
     fn bundled_graph_agents_parse_and_validate() {
         use crate::graph::GraphParser;
-        use crate::graph::validator::GraphValidator;
+        use crate::graph::validator::{GraphValidator, ValidationResult};
+        use std::collections::BTreeMap;
+
+        const UNREACHABLE: &str = "Node is unreachable from the start node via declared edges \
+                                   (script `_next` routing is not analyzed)";
+
+        fn warning_lines(result: &ValidationResult) -> Vec<String> {
+            let mut lines: Vec<String> = result
+                .warnings
+                .iter()
+                .map(|w| format!("{}: {}", w.node_id.as_deref().unwrap_or("-"), w.message))
+                .collect();
+            lines.sort();
+            lines
+        }
+
+        fn unreachable(node_ids: &[&str]) -> Vec<String> {
+            node_ids
+                .iter()
+                .map(|id| format!("{id}: {UNREACHABLE}"))
+                .collect()
+        }
 
         let _guard = TestConfigDirGuard::new();
 
@@ -8084,6 +8105,7 @@ mod tests {
         Skill::install_builtin_skills(false).unwrap();
 
         let mut checked = Vec::new();
+        let mut warnings_by_agent = BTreeMap::new();
         for entry in std::fs::read_dir(paths::agents_data_dir()).unwrap() {
             let dir = entry.unwrap().path();
             let graph_path = dir.join("graph.yaml");
@@ -8100,6 +8122,7 @@ mod tests {
                 "graph.yaml for '{name}' failed validation: {:#?}",
                 result.errors
             );
+            warnings_by_agent.insert(name.clone(), warning_lines(&result));
             checked.push(name);
         }
         checked.sort();
@@ -8109,6 +8132,89 @@ mod tests {
                 "expected bundled graph agent '{expected}' to be checked; found {checked:?}"
             );
         }
+
+        // Warning parity with the pre-subgraph validator: the only warnings any
+        // bundled graph emits are the `_next`-routed unreachable nodes. A new
+        // rule that fires on a shipped asset (or a new bundled graph) must be
+        // recorded here deliberately.
+        let expected_warnings = BTreeMap::from([
+            (
+                "coder".to_string(),
+                unreachable(&[
+                    "analyze_request",
+                    "end_rejected",
+                    "end_success",
+                    "fix_loop_gate",
+                    "gate_approval",
+                    "implement",
+                    "route_complexity",
+                    "route_review_result",
+                    "self_review",
+                    "verify_build",
+                    "verify_tests",
+                ]),
+            ),
+            ("deep-research".to_string(), unreachable(&["ask_topic"])),
+            ("finding-verifier".to_string(), Vec::new()),
+            ("librarian".to_string(), Vec::new()),
+            (
+                "step-runner".to_string(),
+                unreachable(&[
+                    "check_handoff",
+                    "edge_case_sweep",
+                    "end_blocked",
+                    "end_rejected",
+                    "end_success",
+                    "fix_loop_gate",
+                    "gate_blocked",
+                    "gate_deviation",
+                    "gate_user_review",
+                    "get_revision",
+                    "independent_review",
+                    "revise_from_choice",
+                    "route_review",
+                    "route_sweep",
+                    "verify_build",
+                    "verify_format_lint",
+                    "verify_tests",
+                    "write_handoff",
+                ]),
+            ),
+        ]);
+        assert_eq!(
+            warnings_by_agent, expected_warnings,
+            "bundled graph validator warnings drifted from the recorded baseline"
+        );
+
+        // The shipped reference graph must load through the real parser and
+        // validate cleanly too; its `scripts/synthesize.py` only has to exist.
+        let base = _guard.path.join("example-base");
+        create_dir_all(base.join("scripts")).unwrap();
+        write(base.join("scripts").join("synthesize.py"), "").unwrap();
+        let example = GraphParser::new(&base)
+            .load_from_string(include_str!("../../graph.example.yaml"))
+            .unwrap_or_else(|e| panic!("graph.example.yaml failed to parse: {e}"));
+        let result = GraphValidator::new(&base).validate(&example);
+        assert!(
+            result.errors.is_empty(),
+            "graph.example.yaml failed validation: {:#?}",
+            result.errors
+        );
+        let example_warnings = warning_lines(&result);
+        assert!(
+            example_warnings.iter().all(|w| !w.contains("output_key")),
+            "graph.example.yaml's map must declare a writer for its output_key: {example_warnings:#?}"
+        );
+        assert_eq!(
+            example_warnings,
+            unreachable(&[
+                "aggregate_subjects",
+                "deep_dive",
+                "research_subject",
+                "subjects_map",
+            ]),
+            "graph.example.yaml validator warnings drifted from the recorded baseline"
+        );
     }
 
     #[test]

src/function/agents.rs

@@ -18,6 +18,7 @@ use indexmap::IndexMap;
 use log::{debug, warn};
 use parking_lot::RwLock;
 use serde_json::{Value, json};
+use std::collections::HashMap;
 use std::pin::Pin;
 use std::sync::Arc;
 use std::time::Duration;
@@ -535,18 +536,57 @@ pub async fn handle_agent_tool(
 }
 
 pub fn run_child_agent(
-    mut child_ctx: RequestContext,
+    child_ctx: RequestContext,
     initial_input: Input,
     abort_signal: AbortSignal,
+) -> Pin<Box<dyn Future<Output = Result<String>> + Send>> {
+    run_child_agent_with_graph_inputs(child_ctx, initial_input, abort_signal, None)
+}
+
+/// Stops a child agent's own spawned subagents if the future running the
+/// child is dropped mid-flight (parent timeout, abort) or the child fails
+/// before reaching its own cleanup. Disarmed on the success paths, which
+/// already cancel explicitly.
+struct CancelOnDrop(Option<Arc<RwLock<Supervisor>>>);
+
+impl CancelOnDrop {
+    fn disarm(&mut self) {
+        self.0.take();
+    }
+}
+
+impl Drop for CancelOnDrop {
+    fn drop(&mut self) {
+        if let Some(supervisor) = self.0.take() {
+            supervisor.read().cancel_recursive();
+        }
+    }
+}
+
+/// Runs a child agent to completion. When the child is a graph agent,
+/// `graph_inputs` is overlaid on its `initial_state` before the graph
+/// starts. Passing inputs to a config-only agent is rejected before the
+/// child runs, since nothing could consume them.
+fn run_child_agent_with_graph_inputs(
+    mut child_ctx: RequestContext,
+    initial_input: Input,
+    abort_signal: AbortSignal,
+    graph_inputs: Option<HashMap<String, Value>>,
 ) -> Pin<Box<dyn Future<Output = Result<String>> + Send>> {
     Box::pin(async move {
+        let mut cancel_on_drop = CancelOnDrop(child_ctx.supervisor.clone());
         if graph::active_agent_graph_name(&child_ctx).is_some() {
-            return graph::run_active_agent_graph(
+            let result = graph::run_active_agent_graph_with_inputs(
                 &mut child_ctx,
                 &initial_input.text(),
                 abort_signal,
+                graph_inputs,
             )
             .await;
+            if result.is_ok() {
+                cancel_on_drop.disarm();
+            }
+            return result;
         }
 
         let mut accumulated_output = String::new();
@@ -606,6 +646,7 @@ pub fn run_child_agent(
         if let Some(supervisor) = child_ctx.supervisor.clone() {
             supervisor.read().cancel_recursive();
         }
+        cancel_on_drop.disarm();
 
         Ok(accumulated_output)
     })
@@ -685,10 +726,13 @@ fn effective_max_agent_depth(parent_ctx: &RequestContext) -> usize {
 /// output. This is similar to `handle_spawn` but runs the child agent in the
 /// current task (no tokio::spawn, no supervisor handle registration) so the
 /// graph executor can sequence agent nodes directly.
+/// `graph_inputs` seeds the child's graph state and is only accepted when the
+/// target is a graph agent.
 pub async fn run_agent_for_graph(
     parent_ctx: &mut RequestContext,
     agent_name: &str,
     prompt: &str,
+    graph_inputs: Option<HashMap<String, Value>>,
 ) -> Result<String> {
     let peer_assignment = parent_ctx.peer_assignment.take();
     let assigned_peer = peer_assignment.is_some();
@@ -757,8 +801,22 @@ pub async fn run_agent_for_graph(
         Arc::clone(&child_inbox),
         agent_id.clone(),
     );
+    child_ctx.session_abort = parent_ctx.session_abort.clone();
     child_ctx.rag = agent.rag();
     child_ctx.agent = Some(agent);
+    if graph_inputs.is_some() && graph::active_agent_graph_name(&child_ctx).is_none() {
+        bail!(
+            "agent '{agent_name}' has no graph.yaml; `inputs:` is only valid on agent nodes that target a graph agent"
+        );
+    }
+    if graph_inputs
+        .as_ref()
+        .is_some_and(|inputs| inputs.contains_key("initial_prompt"))
+    {
+        bail!(
+            "agent '{agent_name}': `inputs.initial_prompt` is reserved (the dispatcher seeds it from `prompt:`)"
+        );
+    }
     if assigned_peer {
         child_ctx.peer_registry = peer_registry.clone();
     }
@@ -786,13 +844,7 @@ pub async fn run_agent_for_graph(
 
     debug!("Spawning agent '{agent_name}' for graph node as '{agent_id}'");
 
-    let result = run_child_agent(child_ctx, input, child_abort).await;
-
-    if let Some(registry) = &peer_registry {
-        registry.mark_finished(&agent_id);
-    }
-
-    result
+    run_child_agent_with_graph_inputs(child_ctx, input, child_abort, graph_inputs).await
 }
 
 async fn populate_agent_mcp_runtime(ctx: &mut RequestContext, server_ids: &[String]) -> Result<()> {
@@ -1919,6 +1971,141 @@ mod tests {
         assert_eq!(effective_max_agent_depth(&ctx), 5);
     }
 
+    fn unique_agent_name(prefix: &str) -> String {
+        format!(
+            "{prefix}_{}",
+            SystemTime::now()
+                .duration_since(UNIX_EPOCH)
+                .unwrap()
+                .as_nanos()
+        )
+    }
+
+    fn write_config_only_agent(agent_name: &str) {
+        let agent_dir = paths::agent_data_dir(agent_name);
+        create_dir_all(&agent_dir).unwrap();
+        write(
+            agent_dir.join("config.yaml"),
+            format!("name: {agent_name}\ninstructions: hi\n"),
+        )
+        .unwrap();
+    }
+
+    fn write_graph_agent(agent_name: &str) {
+        let agent_dir = paths::agent_data_dir(agent_name);
+        create_dir_all(&agent_dir).unwrap();
+        write(
+            agent_dir.join("graph.yaml"),
+            format!(
+                "name: {agent_name}\nstart: done\nnodes:\n  done:\n    type: end\n    output: done\n"
+            ),
+        )
+        .unwrap();
+    }
+
+    fn run_agent_for_graph_error(
+        agent_name: &str,
+        graph_inputs: Option<HashMap<String, Value>>,
+    ) -> String {
+        let mut ctx = RequestContext::new(default_app_state(), WorkingMode::Cmd);
+        let err = run_async(run_agent_for_graph(
+            &mut ctx,
+            agent_name,
+            "hi",
+            graph_inputs,
+        ))
+        .expect_err("run_agent_for_graph should reject the inputs");
+        format!("{err:#}")
+    }
+
+    #[test]
+    #[serial]
+    fn run_agent_for_graph_rejects_inputs_for_config_only_agent() {
+        let _guard = TestConfigDirGuard::new();
+        let agent_name = unique_agent_name("test_inputs_plain_agent");
+        write_config_only_agent(&agent_name);
+
+        let chain = run_agent_for_graph_error(&agent_name, Some(HashMap::new()));
+
+        assert_eq!(
+            chain,
+            format!(
+                "agent '{agent_name}' has no graph.yaml; `inputs:` is only valid on agent nodes that target a graph agent"
+            )
+        );
+    }
+
+    #[test]
+    #[serial]
+    fn run_agent_for_graph_rejects_initial_prompt_input_on_graph_agent() {
+        let _guard = TestConfigDirGuard::new();
+        let agent_name = unique_agent_name("test_inputs_graph_agent");
+        write_graph_agent(&agent_name);
+        let inputs = HashMap::from([("initial_prompt".to_string(), json!("x"))]);
+
+        let chain = run_agent_for_graph_error(&agent_name, Some(inputs));
+
+        assert_eq!(
+            chain,
+            format!(
+                "agent '{agent_name}': `inputs.initial_prompt` is reserved (the dispatcher seeds it from `prompt:`)"
+            )
+        );
+    }
+
+    /// The child ctx must carry the parent's session flag: a session already
+    /// aborted before the child starts trips the child graph's own pre-check,
+    /// so it never reaches its (otherwise trivially successful) end node.
+    /// Without the inheritance the child sees `None` and returns "done".
+    #[test]
+    #[serial]
+    fn run_agent_for_graph_child_inherits_session_abort() {
+        let _guard = TestConfigDirGuard::new();
+        let agent_name = unique_agent_name("test_session_abort_graph_agent");
+        write_graph_agent(&agent_name);
+        let session = create_abort_signal();
+        session.set_ctrlc();
+        let mut ctx = RequestContext::new(default_app_state(), WorkingMode::Cmd);
+        ctx.session_abort = Some(Arc::clone(&session));
+
+        let err = run_async(run_agent_for_graph(&mut ctx, &agent_name, "hi", None))
+            .expect_err("the inherited session abort must end the child graph");
+
+        let chain = format!("{err:#}");
+        assert!(
+            chain.contains(&format!("Graph execution failed for agent '{agent_name}'")),
+            "{chain}"
+        );
+        assert!(chain.contains("aborted before super-step"), "{chain}");
+        assert!(
+            Arc::ptr_eq(ctx.session_abort.as_ref().unwrap(), &session),
+            "the parent's own session signal is untouched"
+        );
+    }
+
+    #[test]
+    fn cancel_on_drop_guard_cancels_when_armed_and_not_when_disarmed() {
+        run_async(async {
+            let mut ctx = ctx_with_supervisor(4, 3);
+            let armed_abort = register_running_agent(&mut ctx, "a1", "explore");
+            drop(CancelOnDrop(ctx.supervisor.clone()));
+            assert!(
+                armed_abort.aborted(),
+                "an armed guard cancels the supervisor's agents when dropped"
+            );
+
+            let mut ctx = ctx_with_supervisor(4, 3);
+            let disarmed_abort = register_running_agent(&mut ctx, "a2", "explore");
+            let mut guard = CancelOnDrop(ctx.supervisor.clone());
+            guard.disarm();
+            drop(guard);
+            assert!(
+                !disarmed_abort.aborted(),
+                "a disarmed guard leaves the supervisor's agents running"
+            );
+        });
+    }
+
     fn auto_continue_ctx() -> RequestContext {
         let app = AppState {
             config: Arc::new(AppConfig {

src/function/mod.rs

@@ -1684,6 +1684,7 @@ impl ToolCall {
                 agent_name,
                 ctx.app.config.tool_timeout,
                 quiet,
+                ctx.session_abort.clone(),
             ) {
                 Ok(Some(contents)) => serde_json::from_str(&contents)
                     .ok()
@@ -2293,6 +2294,7 @@ pub fn run_llm_function(
     agent_name: Option<String>,
     tool_timeout: Option<u64>,
     quiet: bool,
+    abort: Option<AbortSignal>,
 ) -> Result<Option<String>> {
     let mut bin_dirs: Vec<PathBuf> = vec![];
     let mut command_name = cmd_name.clone();
@@ -2440,6 +2442,21 @@ pub fn run_llm_function(
 
             return Ok(Some(error_json.to_string()));
         }
+        if abort.as_ref().is_some_and(|a| a.aborted()) {
+            let _ = child.kill();
+            let _ = child.wait();
+            drop(stdout_thread);
+            drop(stderr_thread);
+            let tool_error_message = format!("Tool call '{command_name}' aborted");
+            emit_tool_warning(
+                quiet,
+                &format!("⚠️ {tool_error_message} ⚠️"),
+                &tool_error_message,
+            );
+            return Ok(Some(
+                json!({"tool_call_error": tool_error_message}).to_string(),
+            ));
+        }
         thread::sleep(Duration::from_millis(100));
     };
     let stdout_bytes = stdout_thread.join().unwrap_or_default();
@@ -3389,7 +3406,8 @@ mod tests {
                 continue;
             }
             let name = path.file_stem().unwrap().to_string_lossy().to_string();
-            let declarations = Functions::generate_declarations(&path)
+            let src = std::fs::read_to_string(&path).unwrap();
+            let (_, declarations) = bash::build_bash_tool(&src, &name)
                 .unwrap_or_else(|e| panic!("bundled tool '{name}' failed to parse: {e}"));
             assert!(
                 !declarations.is_empty(),
@@ -4561,6 +4579,7 @@ mod tests {
             None,
             None,
             false,
+            None,
         )
         .unwrap()
         .expect("nonzero exit must return an error payload");
@@ -4589,6 +4608,7 @@ mod tests {
             None,
             None,
             true,
+            None,
         )
         .unwrap()
         .expect("nonzero exit must return an error payload");
@@ -4604,6 +4624,38 @@ mod tests {
         assert_eq!(json["output"], "partial-output\n");
     }
 
+    /// An already-aborted signal is noticed on the first wait-loop poll, so a
+    /// tool that would run for seconds is killed and reported within one poll
+    /// interval instead of running to completion or `tool_timeout`.
+    #[cfg(unix)]
+    #[test]
+    fn run_llm_function_aborts_running_tool() {
+        if which::which("bash").is_err() {
+            eprintln!("skipping: bash not available");
+            return;
+        }
+        let abort = create_abort_signal();
+        abort.set_ctrlc();
+
+        let started = Instant::now();
+        let result = run_llm_function(
+            "bash".into(),
+            vec!["-c".into(), "sleep 5".into()],
+            HashMap::new(),
+            None,
+            None,
+            true,
+            Some(abort),
+        )
+        .unwrap()
+        .expect("an aborted tool must return an error payload");
+        let elapsed = started.elapsed();
+
+        assert!(elapsed < Duration::from_secs(1), "took {elapsed:?}");
+        let json: serde_json::Value = serde_json::from_str(&result).unwrap();
+        assert_eq!(json["tool_call_error"], "Tool call 'bash' aborted");
+    }
+
     #[test]
     #[serial]
     fn tool_timeout_secs_prefers_env_then_config_then_default() {

src/graph/agent.rs

@@ -1,14 +1,16 @@
 use super::state::StateManager;
+use super::state_updates;
 use super::structured;
 use super::types::AgentNode;
+use super::wall_clock;
 use crate::config::RequestContext;
 use crate::function::agents::run_agent_for_graph;
 use anyhow::{Context, Result};
+use log::debug;
 use serde_json::Value;
-use std::time::Duration;
+use std::collections::HashMap;
 use tokio::time::timeout;
 
-const OUTPUT_KEY: &str = "output";
 const DEFAULT_TIMEOUT_SECS: u64 = 300;
 
 pub struct AgentNodeExecutor;
@@ -18,26 +20,44 @@ impl AgentNodeExecutor {
         node: &AgentNode,
         state_manager: &mut StateManager,
         parent_ctx: &mut RequestContext,
+        retire_peer_on_return: bool,
     ) -> Result<String> {
         let prompt = state_manager
             .interpolate(&node.prompt)
             .with_context(|| format!("Failed to interpolate prompt for agent '{}'", node.agent))?;
 
-        let timeout_dur = Duration::from_secs(node.timeout.unwrap_or(DEFAULT_TIMEOUT_SECS));
+        let graph_inputs = resolve_inputs(node, state_manager)?;
+        if let Some(inputs) = &graph_inputs {
+            let mut keys: Vec<&String> = inputs.keys().collect();
+            keys.sort_unstable();
+            debug!("Agent '{}' graph inputs: {keys:?}", node.agent);
+        }
 
-        let raw = timeout(
-            timeout_dur,
-            run_agent_for_graph(parent_ctx, &node.agent, &prompt),
-        )
-        .await
-        .with_context(|| {
-            format!(
-                "Agent '{}' timed out after {}s",
-                node.agent,
-                timeout_dur.as_secs()
-            )
-        })?
-        .with_context(|| format!("Agent '{}' failed", node.agent))?;
+        let secs = node.timeout.unwrap_or(DEFAULT_TIMEOUT_SECS);
+        let bound = wall_clock(secs);
+
+        // run_agent_for_graph takes the identity off the ctx; keep a handle so
+        // a frontier peer is retired the moment the agent stops, not after
+        // extraction. Chains re-arm the same identity for later steps and
+        // leave retirement to the chain runner.
+        let peer = parent_ctx.peer_registry.clone().zip(
+            parent_ctx
+                .peer_assignment
+                .as_ref()
+                .map(|(id, _)| id.clone()),
+        );
+
+        let fut = run_agent_for_graph(parent_ctx, &node.agent, &prompt, graph_inputs);
+        let raw_result = match bound {
+            Some(d) => timeout(d, fut).await,
+            None => Ok(fut.await),
+        };
+        if retire_peer_on_return && let Some((registry, id)) = &peer {
+            registry.mark_finished(id);
+        }
+        let raw = raw_result
+            .with_context(|| format!("Agent '{}' timed out after {}s", node.agent, secs))?
+            .with_context(|| format!("Agent '{}' failed", node.agent))?;
 
         let output_value = match &node.output_schema {
             Some(schema) => structured::extract(&raw, schema, parent_ctx)
@@ -57,46 +77,48 @@ impl AgentNodeExecutor {
     }
 }
 
-fn apply_state_updates(node: &AgentNode, state_manager: &mut StateManager, output: &Value) {
-    if node.output_schema.is_some()
-        && let Some(obj) = output.as_object()
-    {
-        for (k, v) in obj {
-            state_manager.state_mut().set(k.clone(), v.clone());
-        }
-    }
-
-    let Some(updates) = &node.state_updates else {
-        return;
+/// Resolves each `inputs` template against the parent state. A lone
+/// `{{key}}` yields the state value as-is (numbers, arrays, objects and
+/// `null` all survive); anything else renders to a string. Every key is
+/// strict: a missing reference fails the node before the child starts.
+fn resolve_inputs(
+    node: &AgentNode,
+    state_manager: &StateManager,
+) -> Result<Option<HashMap<String, Value>>> {
+    let Some(inputs) = &node.inputs else {
+        return Ok(None);
     };
-    let prev_output = state_manager.state().get(OUTPUT_KEY).cloned();
-    state_manager
-        .state_mut()
-        .set(OUTPUT_KEY.into(), output.clone());
-
-    for (key, template) in updates {
-        let value = state_manager.interpolate_lenient(template);
-        state_manager
-            .state_mut()
-            .set(key.clone(), Value::String(value));
+    let mut resolved = HashMap::with_capacity(inputs.len());
+    for (key, template) in inputs {
+        let value = state_manager.interpolate_raw(template).with_context(|| {
+            format!(
+                "Failed to interpolate inputs.{key} for agent '{}'",
+                node.agent
+            )
+        })?;
+        resolved.insert(key.clone(), value);
     }
+    Ok(Some(resolved))
+}
 
-    match prev_output {
-        Some(v) => state_manager.state_mut().set(OUTPUT_KEY.into(), v),
-        None => {
-            state_manager
-                .state_mut()
-                .set(OUTPUT_KEY.into(), Value::Null);
-        }
-    }
+fn apply_state_updates(node: &AgentNode, state_manager: &mut StateManager, output: &Value) {
+    state_updates::apply(
+        state_manager,
+        output,
+        node.output_schema.is_some(),
+        node.state_updates.as_ref(),
+    );
 }
 
 #[cfg(test)]
 mod tests {
     use super::super::types::AgentNode;
     use super::*;
+    use crate::config::{AppState, WorkingMode, default_max_agent_depth};
+    use crate::supervisor::mailbox::{Inbox, PeerRegistry, graph_agent_id};
     use serde_json::json;
     use std::collections::HashMap;
+    use std::sync::Arc;
 
     fn manager_with(pairs: &[(&str, Value)]) -> StateManager {
         let mut map = HashMap::new();
@@ -113,10 +135,117 @@ mod tests {
             state_updates: updates,
             output_schema: None,
             timeout: None,
+            inputs: None,
             teammates: false,
         }
     }
 
+    fn ctx_at_max_depth_with_peer() -> (RequestContext, Arc<PeerRegistry>, String) {
+        let registry = Arc::new(PeerRegistry::new());
+        let id = graph_agent_id("test_agent");
+        let inbox = Arc::new(Inbox::new());
+        registry.insert(id.clone(), "worker[0]".into(), Arc::clone(&inbox));
+        let mut ctx = RequestContext::new(Arc::new(AppState::test_default()), WorkingMode::Cmd);
+        ctx.current_depth = default_max_agent_depth();
+        ctx.peer_registry = Some(Arc::clone(&registry));
+        ctx.peer_assignment = Some((id.clone(), inbox));
+        (ctx, registry, id)
+    }
+
+    async fn execute_past_max_depth(ctx: &mut RequestContext, retire_peer_on_return: bool) {
+        let mut node = node_with("hi", None);
+        node.teammates = true;
+        let mut state = manager_with(&[]);
+
+        let err = AgentNodeExecutor::execute(&node, &mut state, ctx, retire_peer_on_return)
+            .await
+            .expect_err("agent past max depth should fail before running");
+
+        let chain = format!("{err:#}");
+        assert!(chain.contains("Agent 'test_agent' failed"), "{chain}");
+        assert!(chain.contains("Max agent depth exceeded"), "{chain}");
+    }
+
+    #[tokio::test]
+    async fn execute_retires_peer_identity_on_frontier_when_agent_fails() {
+        let (mut ctx, registry, id) = ctx_at_max_depth_with_peer();
+
+        execute_past_max_depth(&mut ctx, true).await;
+
+        assert!(registry.is_finished(&id));
+    }
+
+    #[tokio::test]
+    async fn execute_leaves_peer_identity_live_in_branch_mode_when_agent_fails() {
+        let (mut ctx, registry, id) = ctx_at_max_depth_with_peer();
+
+        execute_past_max_depth(&mut ctx, false).await;
+
+        assert!(!registry.is_finished(&id));
+    }
+
+    fn node_with_inputs(pairs: &[(&str, &str)]) -> AgentNode {
+        let mut node = node_with("hi", None);
+        node.inputs = Some(
+            pairs
+                .iter()
+                .map(|(k, v)| ((*k).to_string(), (*v).to_string()))
+                .collect(),
+        );
+        node
+    }
+
+    #[test]
+    fn resolve_inputs_is_none_when_node_has_no_inputs() {
+        let node = node_with("hi", None);
+        let state = manager_with(&[("n", json!(3))]);
+
+        assert_eq!(resolve_inputs(&node, &state).unwrap(), None);
+    }
+
+    #[test]
+    fn resolve_inputs_lone_reference_keeps_the_array_value() {
+        let node = node_with_inputs(&[("items", "{{list}}")]);
+        let state = manager_with(&[("list", json!(["a", "b"]))]);
+
+        let inputs = resolve_inputs(&node, &state).unwrap().unwrap();
+
+        assert_eq!(inputs.get("items"), Some(&json!(["a", "b"])));
+    }
+
+    #[test]
+    fn resolve_inputs_mixed_text_renders_to_a_string() {
+        let node = node_with_inputs(&[("label", "n={{n}}")]);
+        let state = manager_with(&[("n", json!(3))]);
+
+        let inputs = resolve_inputs(&node, &state).unwrap().unwrap();
+
+        assert_eq!(inputs.get("label"), Some(&json!("n=3")));
+    }
+
+    #[test]
+    fn resolve_inputs_lone_reference_to_null_yields_null() {
+        let node = node_with_inputs(&[("x", "{{k}}")]);
+        let state = manager_with(&[("k", Value::Null)]);
+
+        let inputs = resolve_inputs(&node, &state).unwrap().unwrap();
+
+        assert_eq!(inputs.get("x"), Some(&Value::Null));
+    }
+
+    #[test]
+    fn resolve_inputs_missing_key_errors_naming_the_input() {
+        let node = node_with_inputs(&[("width", "{{nope}}")]);
+        let state = manager_with(&[]);
+
+        let err = resolve_inputs(&node, &state).expect_err("missing key must fail");
+
+        let chain = format!("{err:#}");
+        assert!(chain.contains("inputs.width"), "{chain}");
+        assert!(chain.contains("for agent 'test_agent'"), "{chain}");
+        assert!(chain.contains("'nope' not found in state"), "{chain}");
+    }
+
     #[test]
     fn state_updates_use_output_placeholder() {
         let node = {
@@ -162,7 +291,7 @@ mod tests {
 
         apply_state_updates(&node, &mut state, &json!("anything"));
 
-        assert_eq!(state.state().get("output"), Some(&Value::Null));
+        assert!(state.state().get("output").is_none());
     }
 
     #[test]

src/graph/dispatch.rs

@@ -5,6 +5,7 @@ use crate::utils::AbortSignal;
 use anyhow::{Context, Result, anyhow};
 use log::info;
 use serde_json::Value;
+use std::collections::HashMap;
 
 pub fn active_agent_graph_name(ctx: &RequestContext) -> Option<String> {
     let name = ctx.agent.as_ref()?.name().to_string();
@@ -15,6 +16,18 @@ pub async fn run_active_agent_graph(
     ctx: &mut RequestContext,
     prompt: &str,
     abort_signal: AbortSignal,
+) -> Result<String> {
+    run_active_agent_graph_with_inputs(ctx, prompt, abort_signal, None).await
+}
+
+/// Runs the active agent's graph with `inputs` overlaid on its
+/// `initial_state` before the executor starts. `None` seeds exactly what
+/// `run_active_agent_graph` does: the prompt under `initial_prompt`.
+pub async fn run_active_agent_graph_with_inputs(
+    ctx: &mut RequestContext,
+    prompt: &str,
+    abort_signal: AbortSignal,
+    inputs: Option<HashMap<String, Value>>,
 ) -> Result<String> {
     let agent_name =
         active_agent_graph_name(ctx).ok_or_else(|| anyhow!("Active agent has no graph.yaml"))?;
@@ -29,9 +42,7 @@ pub async fn run_active_agent_graph(
         .load_from_file(&graph_path)
         .with_context(|| format!("Failed to load graph.yaml for agent '{agent_name}'"))?;
 
-    graph
-        .initial_state
-        .insert("initial_prompt".into(), Value::String(prompt.to_string()));
+    seed_initial_state(&mut graph.initial_state, prompt, inputs);
 
     let executor = GraphExecutor::new(graph, agent_dir);
     let output = executor
@@ -45,3 +56,91 @@ pub async fn run_active_agent_graph(
 
     Ok(output)
 }
+
+/// Seeds a child graph's `initial_state`, later writers winning: the graph's
+/// own YAML defaults, then the caller's `inputs`, then `initial_prompt`,
+/// which the dispatcher always owns.
+pub fn seed_initial_state(
+    initial_state: &mut HashMap<String, Value>,
+    prompt: &str,
+    inputs: Option<HashMap<String, Value>>,
+) {
+    if let Some(inputs) = inputs {
+        initial_state.extend(inputs);
+    }
+    initial_state.insert("initial_prompt".into(), Value::String(prompt.to_string()));
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use serde_json::json;
+
+    fn yaml_defaults() -> HashMap<String, Value> {
+        HashMap::from([
+            ("width".to_string(), json!(4)),
+            ("mode".to_string(), json!("fast")),
+        ])
+    }
+
+    #[test]
+    fn seed_initial_state_without_inputs_only_adds_initial_prompt() {
+        let mut state = yaml_defaults();
+
+        seed_initial_state(&mut state, "hello", None);
+
+        let mut expected = yaml_defaults();
+        expected.insert("initial_prompt".into(), json!("hello"));
+        assert_eq!(state, expected);
+    }
+
+    #[test]
+    fn seed_initial_state_without_inputs_on_empty_state_inserts_exactly_one_key() {
+        let mut state = HashMap::new();
+
+        seed_initial_state(&mut state, "hello", None);
+
+        assert_eq!(
+            state,
+            HashMap::from([("initial_prompt".to_string(), json!("hello"))])
+        );
+    }
+
+    #[test]
+    fn seed_initial_state_inputs_override_yaml_defaults_and_keep_the_rest() {
+        let mut state = yaml_defaults();
+        let inputs = HashMap::from([
+            ("width".to_string(), json!(2)),
+            ("extra".to_string(), json!(["a", "b"])),
+        ]);
+
+        seed_initial_state(&mut state, "hello", Some(inputs));
+
+        assert_eq!(state.get("width"), Some(&json!(2)));
+        assert_eq!(state.get("mode"), Some(&json!("fast")));
+        assert_eq!(state.get("extra"), Some(&json!(["a", "b"])));
+        assert_eq!(state.get("initial_prompt"), Some(&json!("hello")));
+        assert_eq!(state.len(), 4);
+    }
+
+    #[test]
+    fn seed_initial_state_dispatcher_prompt_beats_an_initial_prompt_input() {
+        let mut state = HashMap::from([("initial_prompt".to_string(), json!("from yaml"))]);
+        let inputs = HashMap::from([("initial_prompt".to_string(), json!("from inputs"))]);
+
+        seed_initial_state(&mut state, "from dispatcher", Some(inputs));
+
+        assert_eq!(state.get("initial_prompt"), Some(&json!("from dispatcher")));
+        assert_eq!(state.len(), 1);
+    }
+
+    #[test]
+    fn seed_initial_state_null_input_overlays_a_default() {
+        let mut state = yaml_defaults();
+        let inputs = HashMap::from([("width".to_string(), Value::Null)]);
+
+        seed_initial_state(&mut state, "hello", Some(inputs));
+
+        assert_eq!(state.get("width"), Some(&Value::Null));
+    }
+}

src/graph/llm.rs

@@ -1,6 +1,8 @@
 use super::state::StateManager;
+use super::state_updates;
 use super::structured;
 use super::types::LlmNode;
+use super::wall_clock;
 use crate::client::{Model, ModelType, call_chat_completions};
 use crate::config::prompts::DEFAULT_SKILL_INSTRUCTIONS;
 use crate::config::{
@@ -9,17 +11,14 @@ use crate::config::{
 use crate::function::agents::{GuardrailAction, check_pending_tasks_guardrail};
 use crate::function::jobs::reap_jobs;
 use crate::function::skill::skill_function_declarations;
-use crate::utils::create_abort_signal;
+use crate::utils::AbortSignal;
 use anyhow::{Context, Error, Result, anyhow, bail};
 use log::warn;
 use serde_json::Value;
 use std::collections::HashSet;
 use std::sync::Arc;
-use std::time::Duration;
 use tokio::time::timeout;
 
-const OUTPUT_KEY: &str = "output";
-
 #[derive(Debug, PartialEq, Eq)]
 pub(super) enum LlmExecutionOutcome {
     Continue,
@@ -34,8 +33,9 @@ impl LlmNodeExecutor {
         node: &LlmNode,
         state_manager: &mut StateManager,
         parent_ctx: &mut RequestContext,
+        abort: &AbortSignal,
     ) -> Result<LlmExecutionOutcome> {
-        let result = run(node_id, node, state_manager, parent_ctx).await;
+        let result = run(node_id, node, state_manager, parent_ctx, abort).await;
         let (output, failure_reason) = match result {
             Ok(raw) => match &node.output_schema {
                 Some(schema) => match structured::extract(&raw, schema, parent_ctx).await {
@@ -84,6 +84,7 @@ async fn run(
     node: &LlmNode,
     state_manager: &mut StateManager,
     parent_ctx: &mut RequestContext,
+    abort: &AbortSignal,
 ) -> Result<String> {
     let mut instructions: Option<String> = match &node.instructions {
         Some(s) => Some(
@@ -186,17 +187,12 @@ async fn run(
         node.mcp_tools.clone().map(|map| (node_id.to_string(), map)),
     );
     parent_ctx.refresh_mcp_tool_filters();
-    let result = match node.timeout {
-        Some(secs) => match timeout(
-            Duration::from_secs(secs),
-            run_with_retries(node, &prompt, parent_ctx),
-        )
-        .await
-        {
+    let result = match node.timeout.and_then(wall_clock) {
+        Some(d) => match timeout(d, run_with_retries(node, &prompt, parent_ctx, abort)).await {
             Ok(r) => r,
-            Err(_) => Err(anyhow!("llm node timed out after {secs}s")),
+            Err(_) => Err(anyhow!("llm node timed out after {}s", d.as_secs())),
         },
-        None => run_with_retries(node, &prompt, parent_ctx).await,
+        None => run_with_retries(node, &prompt, parent_ctx, abort).await,
     };
     parent_ctx.role = saved_role;
     let node_jobs =
@@ -248,10 +244,11 @@ async fn run_with_retries(
     node: &LlmNode,
     prompt: &str,
     ctx: &mut RequestContext,
+    abort: &AbortSignal,
 ) -> Result<String> {
     let mut last_err: Option<Error> = None;
     for attempt in 1..=node.max_attempts {
-        match run_chat_loop(node, prompt, ctx).await {
+        match run_chat_loop(node, prompt, ctx, abort).await {
             Ok(out) => return Ok(out),
             Err(e) if is_transient(&e) && attempt < node.max_attempts => {
                 warn!("llm node attempt {attempt} failed (transient): {e}; retrying");
@@ -263,14 +260,28 @@ async fn run_with_retries(
     Err(last_err.unwrap_or_else(|| anyhow!("llm node exhausted retries")))
 }
 
-async fn run_chat_loop(node: &LlmNode, prompt: &str, ctx: &mut RequestContext) -> Result<String> {
-    let abort = create_abort_signal();
+/// Whether `turn` (0-based) is the final turn the node's cap allows. A cap of 0 means no cap.
+pub(crate) fn is_last_turn(turn: u32, max_iterations: u32) -> bool {
+    max_iterations > 0 && turn == max_iterations - 1
+}
+
+async fn run_chat_loop(
+    node: &LlmNode,
+    prompt: &str,
+    ctx: &mut RequestContext,
+    abort: &AbortSignal,
+) -> Result<String> {
+    let abort = abort.clone();
     let app_cfg = Arc::clone(&ctx.app.config);
     let role_for_input = ctx.role.clone();
     let mut input = Input::from_str(ctx, prompt, role_for_input)?;
     let mut accumulated = String::new();
 
-    for turn in 0..node.max_iterations {
+    let mut turn: u32 = 0;
+    loop {
+        if abort.aborted() {
+            bail!("llm node aborted");
+        }
         let client = input.create_client()?;
         ctx.before_chat_completion(&input)?;
         let (output, tool_results) =
@@ -297,7 +308,7 @@ async fn run_chat_loop(node: &LlmNode, prompt: &str, ctx: &mut RequestContext) -
                     return Ok(accumulated);
                 }
                 GuardrailAction::Inject(prompt) => {
-                    if turn + 1 == node.max_iterations {
+                    if is_last_turn(turn, node.max_iterations) {
                         bail!(
                             "llm node hit max_iterations ({}) before LLM concluded",
                             node.max_iterations
@@ -305,22 +316,19 @@ async fn run_chat_loop(node: &LlmNode, prompt: &str, ctx: &mut RequestContext) -
                     }
                     let role = ctx.role.clone();
                     input = Input::from_str(ctx, &prompt, role)?;
-                    continue;
                 }
             }
+        } else {
+            if is_last_turn(turn, node.max_iterations) {
+                bail!(
+                    "llm node hit max_iterations ({}) before LLM concluded",
+                    node.max_iterations
+                );
+            }
+            input = input.merge_tool_results(output, tool_results);
         }
-
-        if turn + 1 == node.max_iterations {
-            bail!(
-                "llm node hit max_iterations ({}) before LLM concluded",
-                node.max_iterations
-            );
-        }
-
-        input = input.merge_tool_results(output, tool_results);
+        turn = turn.saturating_add(1);
     }
-
-    bail!("llm node ended without producing output")
 }
 
 fn build_inline_role(
@@ -456,37 +464,12 @@ fn apply_state_updates_with_output(
     state_manager: &mut StateManager,
     output: &Value,
 ) {
-    if node.output_schema.is_some()
-        && let Some(obj) = output.as_object()
-    {
-        for (k, v) in obj {
-            state_manager.state_mut().set(k.clone(), v.clone());
-        }
-    }
-
-    let Some(updates) = &node.state_updates else {
-        return;
-    };
-    let prev_output = state_manager.state().get(OUTPUT_KEY).cloned();
-    state_manager
-        .state_mut()
-        .set(OUTPUT_KEY.into(), output.clone());
-
-    for (key, template) in updates {
-        let value = state_manager.interpolate_lenient(template);
-        state_manager
-            .state_mut()
-            .set(key.clone(), Value::String(value));
-    }
-
-    match prev_output {
-        Some(v) => state_manager.state_mut().set(OUTPUT_KEY.into(), v),
-        None => {
-            state_manager
-                .state_mut()
-                .set(OUTPUT_KEY.into(), Value::Null);
-        }
-    }
+    state_updates::apply(
+        state_manager,
+        output,
+        node.output_schema.is_some(),
+        node.state_updates.as_ref(),
+    );
 }
 
 fn format_schema_hint(schema: &Value) -> String {
@@ -499,10 +482,14 @@ fn format_schema_hint(schema: &Value) -> String {
 
 #[cfg(test)]
 mod tests {
+    use super::super::state_updates::OUTPUT_KEY;
     use super::super::types::*;
     use super::*;
+    use crate::config::{Agent, AgentConfig, AppState, WorkingMode};
+    use crate::utils::create_abort_signal;
     use serde_json::json;
     use std::collections::HashMap;
+    use std::time::Duration;
 
     fn manager_with(pairs: &[(&str, Value)]) -> StateManager {
         let mut map = HashMap::new();
@@ -568,7 +555,7 @@ mod tests {
 
         apply_state_updates_with_output(&node, &mut state, &json!("anything"));
 
-        assert_eq!(state.state().get(OUTPUT_KEY), Some(&json!(null)));
+        assert!(state.state().get(OUTPUT_KEY).is_none());
     }
 
     #[test]
@@ -756,4 +743,64 @@ mod tests {
         assert!(!is_transient(&anyhow!("hit max_iterations")));
         assert!(!is_transient(&anyhow!("authentication failed")));
     }
+
+    #[test]
+    fn zero_timeout_resolves_to_no_bound() {
+        assert!(Some(0u64).and_then(wall_clock).is_none());
+        assert_eq!(
+            Some(5u64).and_then(wall_clock),
+            Some(Duration::from_secs(5))
+        );
+    }
+
+    #[test]
+    fn is_last_turn_bounded() {
+        assert!(is_last_turn(0, 1));
+        assert!(!is_last_turn(0, 10));
+        assert!(is_last_turn(9, 10));
+        assert!(!is_last_turn(10, 10));
+    }
+
+    #[test]
+    fn is_last_turn_zero_cap_never_fires() {
+        assert!(!is_last_turn(0, 0));
+        assert!(!is_last_turn(9, 0));
+        assert!(!is_last_turn(u32::MAX, 0));
+    }
+
+    #[test]
+    fn is_last_turn_at_u32_max_does_not_overflow() {
+        assert!(is_last_turn(u32::MAX - 1, u32::MAX));
+        assert!(!is_last_turn(u32::MAX, u32::MAX));
+    }
+
+    /// The turn-top abort check runs before any client is created, so an
+    /// already-aborted graph never issues a model call and the node fails
+    /// with the abort reason rather than a connection error.
+    #[tokio::test]
+    async fn run_chat_loop_bails_before_first_call_when_aborted() {
+        let node = node_with(None);
+        let mut state = manager_with(&[]);
+        let mut ctx = RequestContext::new(Arc::new(AppState::test_default()), WorkingMode::Cmd);
+        ctx.agent = Some(Agent::test_new(AgentConfig::default()));
+        let abort = create_abort_signal();
+        abort.set_ctrlc();
+
+        let err = LlmNodeExecutor::execute("think", &node, &mut state, &mut ctx, &abort)
+            .await
+            .expect_err("a pre-set abort must fail the node");
+
+        let chain = format!("{err:#}");
+        assert!(
+            chain.contains(
+                "LLM node failed and no fallback declared: LLM call failed: llm node aborted"
+            ),
+            "{chain}"
+        );
+        assert_eq!(
+            state.state().get(OUTPUT_KEY),
+            None,
+            "no output is recorded for an aborted node without state_updates"
+        );
+    }
 }

src/graph/mod.rs

@@ -10,21 +10,30 @@ pub mod reducer;
 pub mod script;
 pub mod staging;
 pub mod state;
+pub mod state_updates;
 pub mod structured;
 pub mod types;
 pub mod user_interaction;
 pub mod validator;
 
-pub use dispatch::{active_agent_graph_name, run_active_agent_graph};
+pub use dispatch::{
+    active_agent_graph_name, run_active_agent_graph, run_active_agent_graph_with_inputs,
+};
 pub use executor::GraphExecutor;
 pub use parser::{GraphParser, agent_has_graph};
 use serde_json::Value;
+use std::time::Duration;
 pub use types::{Graph, NodeType};
 
 pub const GRAPH_SCHEMA_VERSION: &str = "1.0";
 
 pub const DEFAULT_MAX_LOOP_ITERATIONS: usize = 100;
 
+/// A timeout of `0` disables the wall-clock bound.
+pub(crate) fn wall_clock(secs: u64) -> Option<Duration> {
+    (secs != 0).then(|| Duration::from_secs(secs))
+}
+
 pub const MAX_STATE_SIZE_BYTES: usize = 32 * 1024;
 
 pub(in crate::graph) fn type_name(value: &Value) -> &'static str {
@@ -37,3 +46,18 @@ pub(in crate::graph) fn type_name(value: &Value) -> &'static str {
         Value::Object(_) => "object",
     }
 }
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn wall_clock_zero_is_no_bound() {
+        assert!(wall_clock(0).is_none());
+    }
+
+    #[test]
+    fn wall_clock_nonzero_is_that_many_seconds() {
+        assert_eq!(wall_clock(7), Some(Duration::from_secs(7)));
+    }
+}

src/graph/rag.rs

@@ -1,13 +1,13 @@
 use super::state::StateManager;
+use super::state_updates;
 use super::types::RagNode;
+use super::wall_clock;
 use crate::config::RequestContext;
 use crate::utils::create_abort_signal;
 use anyhow::{Context, Result, anyhow};
 use serde_json::{Map, Value};
-use std::time::Duration;
 use tokio::time::timeout;
 
-const OUTPUT_KEY: &str = "output";
 const DEFAULT_QUERY: &str = "{{initial_prompt}}";
 const DEFAULT_RAG_TIMEOUT_SECS: u64 = 120;
 
@@ -34,18 +34,16 @@ impl RagNodeExecutor {
         let top_k = node.top_k.unwrap_or_else(|| rag.configured_top_k());
         let rerank = rag.configured_reranker();
 
-        let timeout_dur = Duration::from_secs(node.timeout.unwrap_or(DEFAULT_RAG_TIMEOUT_SECS));
+        let secs = node.timeout.unwrap_or(DEFAULT_RAG_TIMEOUT_SECS);
+        let bound = wall_clock(secs);
         let abort = create_abort_signal();
-        let (context, sources_str, _ids) =
-            timeout(timeout_dur, rag.search(&query, top_k, rerank, abort))
-                .await
-                .with_context(|| {
-                    format!(
-                        "rag node '{node_id}' timed out after {}s",
-                        timeout_dur.as_secs()
-                    )
-                })?
-                .with_context(|| format!("rag node '{node_id}' retrieval failed"))?;
+        let fut = rag.search(&query, top_k, rerank, abort);
+        let (context, sources_str, _ids) = match bound {
+            Some(d) => timeout(d, fut).await,
+            None => Ok(fut.await),
+        }
+        .with_context(|| format!("rag node '{node_id}' timed out after {secs}s"))?
+        .with_context(|| format!("rag node '{node_id}' retrieval failed"))?;
 
         let output = build_rag_output(context, &sources_str);
         apply_state_updates(node, state_manager, &output);
@@ -70,33 +68,14 @@ fn build_rag_output(context: String, sources_str: &str) -> Value {
 }
 
 fn apply_state_updates(node: &RagNode, state_manager: &mut StateManager, output: &Value) {
-    let Some(updates) = &node.state_updates else {
-        return;
-    };
-    let prev_output = state_manager.state().get(OUTPUT_KEY).cloned();
-    state_manager
-        .state_mut()
-        .set(OUTPUT_KEY.into(), output.clone());
-
-    for (key, template) in updates {
-        let value = state_manager.interpolate_lenient(template);
-        state_manager
-            .state_mut()
-            .set(key.clone(), Value::String(value));
-    }
-
-    match prev_output {
-        Some(v) => state_manager.state_mut().set(OUTPUT_KEY.into(), v),
-        None => state_manager
-            .state_mut()
-            .set(OUTPUT_KEY.into(), Value::Null),
-    }
+    state_updates::apply(state_manager, output, false, node.state_updates.as_ref());
 }
 
 #[cfg(test)]
 mod tests {
     use super::*;
     use serde_json::json;
+    use std::time::Duration;
 
     #[test]
     fn build_rag_output_splits_bullet_sources_into_array() {
@@ -126,4 +105,15 @@ mod tests {
 
         assert_eq!(out["sources"], json!(["plain/path"]));
     }
+
+    /// Mirrors the resolution expression in `execute`.
+    fn resolve(timeout: Option<u64>) -> Option<Duration> {
+        wall_clock(timeout.unwrap_or(DEFAULT_RAG_TIMEOUT_SECS))
+    }
+
+    #[test]
+    fn zero_timeout_resolves_to_no_bound() {
+        assert!(resolve(Some(0)).is_none());
+        assert_eq!(resolve(None), Some(Duration::from_secs(120)));
+    }
 }

src/graph/script.rs

@@ -1,5 +1,6 @@
 use super::state::{StateManager, StateRepresentation};
 use super::types::ScriptNode;
+use super::wall_clock;
 use crate::config::paths;
 use crate::function::Language;
 use anyhow::{Context, Result, anyhow, bail};
@@ -8,7 +9,6 @@ use std::collections::HashMap;
 use std::env;
 use std::path::{Path, PathBuf};
 use std::process::Stdio;
-use std::time::Duration;
 use tokio::process::Command;
 use tokio::time::timeout;
 
@@ -66,22 +66,25 @@ impl ScriptExecutor {
             }
         }
 
-        let timeout_dur = Duration::from_secs(node.timeout);
-        let output = timeout(timeout_dur, cmd.output())
-            .await
-            .with_context(|| {
-                format!(
-                    "Script '{}' timed out after {}s",
-                    script_path.display(),
-                    node.timeout
-                )
-            })?
-            .with_context(|| {
-                format!(
-                    "Failed to spawn script process for '{}'",
-                    script_path.display()
-                )
-            })?;
+        let bound = wall_clock(node.timeout);
+        let fut = cmd.output();
+        let output = match bound {
+            Some(d) => timeout(d, fut).await,
+            None => Ok(fut.await),
+        }
+        .with_context(|| {
+            format!(
+                "Script '{}' timed out after {}s",
+                script_path.display(),
+                node.timeout
+            )
+        })?
+        .with_context(|| {
+            format!(
+                "Failed to spawn script process for '{}'",
+                script_path.display()
+            )
+        })?;
 
         if !output.status.success() {
             let stderr = String::from_utf8_lossy(&output.stderr);
@@ -260,6 +263,33 @@ echo '{"quality": 0.85, "issues": 3, "_next": "approve"}'
         cleanup(&dir);
     }
 
+    #[tokio::test]
+    async fn zero_timeout_is_unbounded() {
+        if !cmd_available("bash") {
+            return;
+        }
+        let (dir, path) = write_script(
+            r#"#!/bin/bash
+sleep 1.5
+echo '{"ok":true}'
+"#,
+            "sh",
+        );
+        let mut state = StateManager::new(HashMap::new());
+        let executor = ScriptExecutor::new(&dir);
+
+        executor
+            .execute(
+                &node_for(path.file_name().unwrap().to_str().unwrap(), 0),
+                &mut state,
+            )
+            .await
+            .unwrap_or_else(|e| panic!("zero timeout should not bound the script: {e:#}"));
+
+        assert_eq!(state.state().get("ok"), Some(&json!(true)));
+        cleanup(&dir);
+    }
+
     #[tokio::test]
     async fn bash_script_can_read_state_from_env() {
         if !cmd_available("bash") {

src/graph/state.rs

@@ -320,6 +320,26 @@ pub(super) fn template_root_keys(template: &str) -> Vec<String> {
         .collect()
 }
 
+// Whether `s` is exactly one `{{key}}` reference and nothing else (after the
+// same trim `interpolate_raw` applies) whose path parses with the grammar
+// `get_nested_value` uses, so this accepts exactly the strings the runtime
+// can resolve as a pure reference.
+pub(super) fn is_lone_template(s: &str) -> bool {
+    let s = s.trim();
+    let Ok(Some(m)) = TEMPLATE_VAR_RE.captures(s) else {
+        return false;
+    };
+    let whole = m.get(0).expect("group 0 always present");
+    if whole.start() != 0 || whole.end() != s.len() {
+        return false;
+    }
+    let path = m.get(1).expect("template regex has one group").as_str();
+    let mut parts = path.split('.');
+    let first = parts.next().and_then(split_indices);
+    first.is_some_and(|(name, _)| !name.is_empty())
+        && parts.all(|part| split_indices(part).is_some())
+}
+
 fn value_to_string(value: &Value) -> String {
     match value {
         Value::String(s) => s.clone(),
@@ -931,6 +951,60 @@ mod tests {
         assert_eq!(result, json!("v"));
     }
 
+    #[test]
+    fn is_lone_template_cases() {
+        for accepted in [
+            "{{budget}}",
+            "{{cfg.limits[0]}}",
+            "  {{budget}}  ",
+            "\n{{k}}\t",
+        ] {
+            assert!(
+                is_lone_template(accepted),
+                "{accepted:?} should be a lone template"
+            );
+        }
+        for rejected in [
+            "",
+            "4",
+            "{{}}",
+            "{{ key }}",
+            "n={{k}}",
+            "{{k}}x",
+            "{{a}} {{b",
+            "{{a}}{{b}}",
+            "{{a-b}}",
+            "{{k",
+        ] {
+            assert!(
+                !is_lone_template(rejected),
+                "{rejected:?} must not be a lone template"
+            );
+        }
+    }
+
+    #[test]
+    fn is_lone_template_matches_runtime_path_grammar() {
+        for accepted in [
+            "{{a.b[0][1].c}}",
+            "{{a[0][1]}}",
+            "{{a.}}",
+            "{{a..b}}",
+            "{{a.[0]}}",
+        ] {
+            assert!(
+                is_lone_template(accepted),
+                "{accepted:?} resolves at run time and must be accepted"
+            );
+        }
+        for rejected in ["{{limits[foo]}}", "{{a[]}}", "{{a[0}}", "{{.a}}"] {
+            assert!(
+                !is_lone_template(rejected),
+                "{rejected:?} never resolves at run time and must be rejected"
+            );
+        }
+    }
+
     #[test]
     fn interpolate_raw_inner_spaces_treated_as_mixed() {
         let manager = manager_with(&[("k", json!("v"))]);

src/graph/state_updates.rs

@@ -0,0 +1,168 @@
+use super::state::StateManager;
+use serde_json::Value;
+use std::collections::HashMap;
+
+pub(super) const OUTPUT_KEY: &str = "output";
+
+/// Merge `output_schema` top-level keys (when `has_schema`) and apply `state_updates` with
+/// `{{output}}` bound to `output` for the duration of interpolation only. An explicit `output`
+/// key in `updates` is honored; otherwise `output` is restored to its prior value, or removed
+/// again if it was absent.
+pub(super) fn apply(
+    state_manager: &mut StateManager,
+    output: &Value,
+    has_schema: bool,
+    updates: Option<&HashMap<String, String>>,
+) {
+    if has_schema && let Some(obj) = output.as_object() {
+        for (k, v) in obj {
+            state_manager.state_mut().set(k.clone(), v.clone());
+        }
+    }
+
+    let Some(updates) = updates else {
+        return;
+    };
+    let prev_output = state_manager.state().get(OUTPUT_KEY).cloned();
+    state_manager
+        .state_mut()
+        .set(OUTPUT_KEY.into(), output.clone());
+
+    let computed: Vec<(String, Value)> = updates
+        .iter()
+        .map(|(key, template)| {
+            (
+                key.clone(),
+                Value::String(state_manager.interpolate_lenient(template)),
+            )
+        })
+        .collect();
+
+    match prev_output {
+        Some(prev) => state_manager.state_mut().set(OUTPUT_KEY.into(), prev),
+        None => {
+            state_manager.state_mut().remove(OUTPUT_KEY);
+        }
+    }
+
+    for (key, value) in computed {
+        state_manager.state_mut().set(key, value);
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use serde_json::json;
+
+    fn manager_with(pairs: &[(&str, Value)]) -> StateManager {
+        let mut map = HashMap::new();
+        for (k, v) in pairs {
+            map.insert((*k).into(), v.clone());
+        }
+        StateManager::new(map)
+    }
+
+    fn updates(pairs: &[(&str, &str)]) -> HashMap<String, String> {
+        pairs
+            .iter()
+            .map(|(k, v)| ((*k).to_string(), (*v).to_string()))
+            .collect()
+    }
+
+    #[test]
+    fn explicit_output_state_update_is_honored() {
+        let u = updates(&[("output", "{{output}}")]);
+        let mut state = manager_with(&[("output", json!("stale"))]);
+
+        apply(&mut state, &json!("the completion"), false, Some(&u));
+
+        assert_eq!(
+            state.state().get(OUTPUT_KEY),
+            Some(&json!("the completion"))
+        );
+    }
+
+    #[test]
+    fn explicit_output_is_honored_alongside_other_updates() {
+        let u = updates(&[("output", "{{output}}"), ("summary", "got: {{output}}")]);
+        let mut state = manager_with(&[]);
+
+        apply(&mut state, &json!("done"), false, Some(&u));
+
+        assert_eq!(state.state().get(OUTPUT_KEY), Some(&json!("done")));
+        assert_eq!(state.state().get("summary"), Some(&json!("got: done")));
+    }
+
+    #[test]
+    fn no_updates_is_a_noop() {
+        let mut state = manager_with(&[("keep", json!(1))]);
+
+        apply(&mut state, &json!("ignored"), false, None);
+
+        assert_eq!(state.state().get("keep"), Some(&json!(1)));
+        assert!(state.state().get(OUTPUT_KEY).is_none());
+    }
+
+    #[test]
+    fn output_is_restored_to_previous_value() {
+        let u = updates(&[("response", "{{output}}")]);
+        let mut state = manager_with(&[("output", json!("preserved"))]);
+
+        apply(&mut state, &json!("new"), false, Some(&u));
+
+        assert_eq!(state.state().get("response"), Some(&json!("new")));
+        assert_eq!(state.state().get(OUTPUT_KEY), Some(&json!("preserved")));
+    }
+
+    #[test]
+    fn output_is_removed_when_absent() {
+        let u = updates(&[("response", "{{output}}")]);
+        let mut state = manager_with(&[]);
+
+        apply(&mut state, &json!("new"), false, Some(&u));
+
+        assert_eq!(state.state().get("response"), Some(&json!("new")));
+        assert!(state.state().get(OUTPUT_KEY).is_none());
+    }
+
+    #[test]
+    fn explicit_null_prior_output_is_restored_as_null() {
+        let u = updates(&[("response", "{{output}}")]);
+        let mut state = manager_with(&[("output", json!(null))]);
+
+        apply(&mut state, &json!("new"), false, Some(&u));
+
+        assert_eq!(state.state().get("response"), Some(&json!("new")));
+        assert_eq!(state.state().get(OUTPUT_KEY), Some(&json!(null)));
+    }
+
+    #[test]
+    fn schema_merges_top_level_keys_when_has_schema() {
+        let mut state = manager_with(&[]);
+
+        apply(&mut state, &json!({"summary": "s", "score": 7}), true, None);
+
+        assert_eq!(state.state().get("summary"), Some(&json!("s")));
+        assert_eq!(state.state().get("score"), Some(&json!(7)));
+        assert!(state.state().get(OUTPUT_KEY).is_none());
+    }
+
+    #[test]
+    fn object_output_is_not_merged_without_schema() {
+        let u = updates(&[("ctx", "{{output.context}}")]);
+        let mut state = manager_with(&[]);
+
+        apply(
+            &mut state,
+            &json!({"context": "c", "sources": ["a"]}),
+            false,
+            Some(&u),
+        );
+
+        assert!(state.state().get("context").is_none());
+        assert!(state.state().get("sources").is_none());
+        assert_eq!(state.state().get("ctx"), Some(&json!("c")));
+        assert!(state.state().get(OUTPUT_KEY).is_none());
+    }
+}

src/graph/types.rs

@@ -240,6 +240,9 @@ pub struct AgentNode {
     #[serde(default, skip_serializing_if = "Option::is_none")]
     pub timeout: Option<u64>,
 
+    #[serde(default, skip_serializing_if = "Option::is_none")]
+    pub inputs: Option<HashMap<String, String>>,
+
     #[serde(default)]
     pub teammates: bool,
 }
@@ -426,13 +429,60 @@ pub struct MapNode {
     pub collect_into: String,
 
     #[serde(default, skip_serializing_if = "Option::is_none")]
-    pub max_concurrency: Option<usize>,
+    pub max_concurrency: Option<ConcurrencyCap>,
 }
 
 fn default_map_output_key() -> String {
     "output".to_string()
 }
 
+/// A map node's per-item concurrency cap: either a literal integer or a
+/// `{{template}}` resolved against graph state when the map runs.
+#[derive(Debug, Clone, Serialize, PartialEq)]
+#[serde(untagged)]
+pub enum ConcurrencyCap {
+    Fixed(usize),
+    Template(String),
+}
+
+// Hand-written so a malformed value (`-1`, `4.5`, `true`) reports what was
+// received and what is accepted, instead of serde's untagged-enum "did not
+// match any variant".
+impl<'de> Deserialize<'de> for ConcurrencyCap {
+    fn deserialize<D>(deserializer: D) -> Result<Self, D::Error>
+    where
+        D: serde::Deserializer<'de>,
+    {
+        struct CapVisitor;
+
+        impl serde::de::Visitor<'_> for CapVisitor {
+            type Value = ConcurrencyCap;
+
+            fn expecting(&self, f: &mut std::fmt::Formatter) -> std::fmt::Result {
+                f.write_str("a positive integer or a `{{template}}` string for max_concurrency")
+            }
+
+            fn visit_u64<E: serde::de::Error>(self, v: u64) -> Result<Self::Value, E> {
+                usize::try_from(v)
+                    .map(ConcurrencyCap::Fixed)
+                    .map_err(|_| E::invalid_value(serde::de::Unexpected::Unsigned(v), &self))
+            }
+
+            fn visit_i64<E: serde::de::Error>(self, v: i64) -> Result<Self::Value, E> {
+                usize::try_from(v)
+                    .map(ConcurrencyCap::Fixed)
+                    .map_err(|_| E::invalid_value(serde::de::Unexpected::Signed(v), &self))
+            }
+
+            fn visit_str<E: serde::de::Error>(self, v: &str) -> Result<Self::Value, E> {
+                Ok(ConcurrencyCap::Template(v.to_owned()))
+            }
+        }
+
+        deserializer.deserialize_any(CapVisitor)
+    }
+}
+
 #[derive(Debug, Clone, Copy, Deserialize, Serialize, PartialEq, Eq)]
 #[serde(rename_all = "lowercase")]
 pub enum Reducer {
@@ -449,6 +499,7 @@ pub enum Reducer {
 #[derive(Debug, Clone, Default)]
 pub struct GraphState {
     data: HashMap<String, Value>,
+    #[cfg(test)]
     history: Vec<String>,
     loop_counts: HashMap<String, usize>,
 }
@@ -457,6 +508,7 @@ impl GraphState {
     pub fn new(initial: HashMap<String, Value>) -> Self {
         Self {
             data: initial,
+            #[cfg(test)]
             history: Vec::new(),
             loop_counts: HashMap::new(),
         }
@@ -470,6 +522,10 @@ impl GraphState {
         self.data.insert(key, value);
     }
 
+    pub fn remove(&mut self, key: &str) -> Option<Value> {
+        self.data.remove(key)
+    }
+
     pub fn merge(&mut self, json_obj: &serde_json::Map<String, Value>) {
         for (key, value) in json_obj {
             self.data.insert(key.clone(), value.clone());
@@ -481,6 +537,7 @@ impl GraphState {
     }
 
     pub fn visit_node(&mut self, node_id: &str) {
+        #[cfg(test)]
         self.history.push(node_id.to_string());
         *self.loop_counts.entry(node_id.to_string()).or_insert(0) += 1;
     }
@@ -583,6 +640,48 @@ nodes:
         }
     }
 
+    #[test]
+    fn agent_node_inputs_parse_optionally_and_empty_map_is_some() {
+        let yaml = r#"
+name: g
+start: with_inputs
+nodes:
+  with_inputs:
+    id: with_inputs
+    type: agent
+    agent: helper
+    prompt: hi
+    inputs:
+      width: "{{n}}"
+      note: "Repo: {{repo}}"
+    next: without_inputs
+  without_inputs:
+    id: without_inputs
+    type: agent
+    agent: helper
+    prompt: hi
+    next: empty_inputs
+  empty_inputs:
+    id: empty_inputs
+    type: agent
+    agent: helper
+    prompt: hi
+    inputs: {}
+"#;
+        let graph: Graph = serde_yaml::from_str(yaml).unwrap();
+        let inputs_of = |id: &str| match &graph.get_node(id).unwrap().node_type {
+            NodeType::Agent(n) => n.inputs.clone(),
+            _ => panic!("expected Agent variant"),
+        };
+
+        let with = inputs_of("with_inputs").expect("inputs should parse");
+        assert_eq!(with.len(), 2);
+        assert_eq!(with.get("width").map(String::as_str), Some("{{n}}"));
+        assert_eq!(with.get("note").map(String::as_str), Some("Repo: {{repo}}"));
+        assert_eq!(inputs_of("without_inputs"), None);
+        assert_eq!(inputs_of("empty_inputs"), Some(HashMap::new()));
+    }
+
     #[test]
     fn deserializes_mcp_tools_at_graph_and_node_level() {
         let yaml = r#"
@@ -1232,7 +1331,7 @@ next: rank
         assert_eq!(map.branch, "research_subject");
         assert_eq!(map.output_key, "research_result");
         assert_eq!(map.collect_into, "research_results");
-        assert_eq!(map.max_concurrency, Some(5));
+        assert_eq!(map.max_concurrency, Some(ConcurrencyCap::Fixed(5)));
     }
 
     #[test]
@@ -1256,6 +1355,78 @@ collect_into: results
         assert!(map.max_concurrency.is_none());
     }
 
+    fn map_node_yaml_with_cap(cap: &str) -> String {
+        format!(
+            "id: fan_out\ntype: map\nover: \"{{{{items}}}}\"\nas: item\nbranch: process\n\
+             collect_into: results\nmax_concurrency: {cap}\n"
+        )
+    }
+
+    fn parse_map_cap(cap: &str) -> Option<ConcurrencyCap> {
+        let node: Node = serde_yaml::from_str(&map_node_yaml_with_cap(cap)).unwrap();
+        match node.node_type {
+            NodeType::Map(m) => m.max_concurrency,
+            _ => panic!("expected Map variant"),
+        }
+    }
+
+    #[test]
+    fn map_max_concurrency_integer_parses_as_fixed() {
+        assert_eq!(parse_map_cap("4"), Some(ConcurrencyCap::Fixed(4)));
+    }
+
+    #[test]
+    fn map_max_concurrency_zero_parses_as_fixed_zero() {
+        assert_eq!(parse_map_cap("0"), Some(ConcurrencyCap::Fixed(0)));
+    }
+
+    #[test]
+    fn map_max_concurrency_string_parses_as_template() {
+        assert_eq!(
+            parse_map_cap("\"{{budget}}\""),
+            Some(ConcurrencyCap::Template("{{budget}}".into()))
+        );
+    }
+
+    #[test]
+    fn map_max_concurrency_quoted_integer_parses_as_template() {
+        assert_eq!(
+            parse_map_cap("\"4\""),
+            Some(ConcurrencyCap::Template("4".into()))
+        );
+    }
+
+    #[test]
+    fn map_max_concurrency_malformed_values_name_accepted_forms() {
+        for bad in ["-1", "4.5", "true", "[4]", "{n: 4}"] {
+            let err = serde_yaml::from_str::<Node>(&map_node_yaml_with_cap(bad))
+                .expect_err(&format!("max_concurrency: {bad} should fail to parse"));
+            let msg = err.to_string();
+            assert!(
+                msg.contains("a positive integer or a `{{template}}` string for max_concurrency"),
+                "max_concurrency: {bad} produced an unhelpful error: {msg}"
+            );
+        }
+    }
+
+    #[test]
+    fn map_max_concurrency_negative_error_names_offending_value() {
+        let err = serde_yaml::from_str::<Node>(&map_node_yaml_with_cap("-1")).unwrap_err();
+        assert!(err.to_string().contains("-1"), "{err}");
+    }
+
+    #[test]
+    fn concurrency_cap_serializes_as_bare_scalar() {
+        assert_eq!(
+            serde_json::to_value(ConcurrencyCap::Fixed(4)).unwrap(),
+            serde_json::json!(4)
+        );
+        assert_eq!(
+            serde_json::to_value(ConcurrencyCap::Template("{{k}}".into())).unwrap(),
+            serde_json::json!("{{k}}")
+        );
+    }
+
     #[test]
     fn full_graph_with_all_new_phase_a_fields_parses() {
         let yaml = r#"
@@ -1386,4 +1557,20 @@ nodes:
 
         assert!(!serde_yaml::to_string(&graph).unwrap().contains("driver"));
     }
+
+    #[test]
+    fn graph_state_remove_returns_previous_value_and_drops_key() {
+        let mut state = GraphState::new(HashMap::from([("k".to_string(), json!(1))]));
+
+        assert_eq!(state.remove("k"), Some(json!(1)));
+        assert_eq!(state.get("k"), None);
+        assert!(state.data().is_empty());
+    }
+
+    #[test]
+    fn graph_state_remove_missing_key_is_none() {
+        let mut state = GraphState::new(HashMap::new());
+
+        assert_eq!(state.remove("k"), None);
+    }
 }

src/parsers/bash.rs

@@ -18,23 +18,32 @@ pub fn generate_bash_declarations(
         .with_context(|| format!("Failed to load script at '{tool_file:?}'"))?;
 
     debug!("Building script at '{tool_file:?}'");
+    let (build_script, declarations) = build_bash_tool(&src, file_name)
+        .with_context(|| format!("Failed to build script '{file_name}'"))?;
+    function::write_file_atomic(tools_file_path, &build_script, Some(0o755))
+        .with_context(|| format!("Failed to write built script to '{tools_file_path:?}'"))?;
+
+    Ok(declarations)
+}
+
+/// Builds the argc script and derives its declarations without touching the
+/// filesystem, so callers can inspect a tool source without rewriting it.
+pub fn build_bash_tool(src: &str, file_name: &str) -> Result<(String, Vec<FunctionDeclaration>)> {
     let build_script = argc::build(
-        &src,
+        src,
         "",
         env::var("TERM_WIDTH").ok().and_then(|v| v.parse().ok()),
     )?;
     let build_script = allow_empty_required_values(&build_script);
-    function::write_file_atomic(tools_file_path, &build_script, Some(0o755))
-        .with_context(|| format!("Failed to write built script to '{tools_file_path:?}'"))?;
 
     let command_value = argc::export(&build_script, file_name)
-        .with_context(|| format!("Failed to parse script at '{tool_file:?}'"))?;
+        .with_context(|| format!("Failed to parse script '{file_name}'"))?;
     if command_value.subcommands.is_empty() {
         let function_declaration =
             command_to_function_declaration(&command_value).ok_or_else(|| {
                 anyhow::format_err!("Tool definition missing or empty description: {file_name}")
             })?;
-        Ok(vec![function_declaration])
+        Ok((build_script, vec![function_declaration]))
     } else {
         let mut declarations = vec![];
         for subcommand in &command_value.subcommands {
@@ -54,7 +63,7 @@ pub fn generate_bash_declarations(
             }
         }
 
-        Ok(declarations)
+        Ok((build_script, declarations))
     }
 }
 

src/utils/spinner.rs

@@ -2,6 +2,11 @@ use super::{AbortSignal, IS_STDOUT_TERMINAL, poll_abort_signal, wait_abort_signa
 
 use anyhow::{Result, bail};
 use crossterm::{cursor, queue, style, terminal};
+#[cfg(test)]
+use std::sync::{
+    Arc,
+    atomic::{AtomicUsize, Ordering},
+};
 use std::{
     future::Future,
     io::{Write, stdout},
@@ -19,11 +24,22 @@ use tokio::{
 pub struct SpinnerInner {
     index: usize,
     message: String,
+    #[cfg(test)]
+    cleared: Option<Arc<AtomicUsize>>,
 }
 
 impl SpinnerInner {
     const DATA: [&'static str; 10] = ["⠋", "⠙", "⠹", "⠸", "⠼", "⠴", "⠦", "⠧", "⠇", "⠏"];
 
+    #[cfg(test)]
+    fn observed(cleared: Arc<AtomicUsize>) -> Self {
+        Self {
+            index: 0,
+            message: String::new(),
+            cleared: Some(cleared),
+        }
+    }
+
     fn step(&mut self) -> Result<()> {
         if !*IS_STDOUT_TERMINAL || self.message.is_empty() {
             return Ok(());
@@ -50,10 +66,17 @@ impl SpinnerInner {
     }
 
     fn clear_message(&mut self) -> Result<()> {
-        if !*IS_STDOUT_TERMINAL || self.message.is_empty() {
+        if self.message.is_empty() {
             return Ok(());
         }
         self.message.clear();
+        #[cfg(test)]
+        if let Some(cleared) = &self.cleared {
+            cleared.fetch_add(1, Ordering::SeqCst);
+        }
+        if !*IS_STDOUT_TERMINAL {
+            return Ok(());
+        }
         let mut writer = stdout();
         queue!(
             writer,
@@ -66,6 +89,17 @@ impl SpinnerInner {
     }
 }
 
+/// Restores the terminal when the spinner goes away by any route, not just the
+/// loop's normal exit: a future cancelled mid-spin (the surrounding task was
+/// aborted) or an early `?` return would otherwise leave the last frame on
+/// screen and the cursor hidden. `clear_message` is a no-op once the message
+/// is gone, so a normal-path clear leaves nothing for the drop to redo.
+impl Drop for SpinnerInner {
+    fn drop(&mut self) {
+        let _ = self.clear_message();
+    }
+}
+
 #[derive(Clone)]
 pub struct Spinner(mpsc::UnboundedSender<SpinnerEvent>);
 
@@ -215,3 +249,65 @@ async fn run_abortable_spinner(
     spinner.clear_message()?;
     Ok(())
 }
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    fn observed(message: &str) -> (SpinnerInner, Arc<AtomicUsize>) {
+        let cleared = Arc::new(AtomicUsize::new(0));
+        let mut spinner = SpinnerInner::observed(cleared.clone());
+        spinner.set_message(message.to_string()).unwrap();
+        (spinner, cleared)
+    }
+
+    #[test]
+    fn drop_with_active_message_clears_once() {
+        let (spinner, cleared) = observed("Thinking");
+        assert_eq!(cleared.load(Ordering::SeqCst), 0);
+        drop(spinner);
+        assert_eq!(cleared.load(Ordering::SeqCst), 1);
+    }
+
+    #[test]
+    fn explicit_clear_then_drop_clears_exactly_once() {
+        let (mut spinner, cleared) = observed("Thinking");
+        spinner.clear_message().unwrap();
+        assert_eq!(cleared.load(Ordering::SeqCst), 1);
+        drop(spinner);
+        assert_eq!(cleared.load(Ordering::SeqCst), 1);
+    }
+
+    #[test]
+    fn drop_without_message_is_silent() {
+        let (spinner, cleared) = observed("");
+
+        drop(spinner);
+
+        assert_eq!(cleared.load(Ordering::SeqCst), 0);
+    }
+
+    #[test]
+    fn set_message_replaces_and_clears_previous_once() {
+        let (mut spinner, cleared) = observed("Thinking");
+        spinner.set_message("Fetching".to_string()).unwrap();
+        assert_eq!(cleared.load(Ordering::SeqCst), 1);
+        drop(spinner);
+        assert_eq!(cleared.load(Ordering::SeqCst), 2);
+    }
+
+    #[tokio::test]
+    async fn cancelled_task_restores_terminal_on_drop() {
+        let (spinner, cleared) = observed("Thinking");
+        let task = tokio::spawn(async move {
+            let _spinner = spinner;
+            std::future::pending::<()>().await;
+        });
+        tokio::task::yield_now().await;
+        assert_eq!(cleared.load(Ordering::SeqCst), 0);
+
+        task.abort();
+        assert!(task.await.unwrap_err().is_cancelled());
+        assert_eq!(cleared.load(Ordering::SeqCst), 1);
+    }
+}