test(graph): cover agent node inputs end to end and at the runtime guards

Two offline chain tests materialize an echo-inputs graph agent whose
script reports the width it sees: a parent with inputs.width gets the
parent's value back, a parent without inputs gets the child's YAML
default. Two agents.rs tests pin the exact runtime guard texts for a
config-only target and a reserved initial_prompt key.

Task T5 of plans/map-subgraph-branches-design.md §3.12.

src/function/agents.rs

@@ -1943,6 +1943,88 @@ 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:`)"
+            )
+        );
+    }
+
     fn auto_continue_ctx() -> RequestContext {
         let app = AppState {
             config: Arc::new(AppConfig {

src/graph/map.rs

@@ -1484,6 +1484,46 @@ nodes:
         .unwrap();
     }
 
+    const ECHO_INPUTS_AGENT: &str = "echo-inputs";
+
+    /// Materializes a graph agent whose only script echoes the `width` it
+    /// finds in `GRAPH_STATE` back as `output`. Its YAML default is
+    /// `width: 1`, so a parent that seeds `width` through `inputs:` is
+    /// observable against a parent that does not.
+    fn materialize_echo_inputs_agent() {
+        let graph_path = paths::agent_graph_file(ECHO_INPUTS_AGENT);
+        let agent_dir = graph_path.parent().unwrap();
+        fs::create_dir_all(agent_dir).unwrap();
+        fs::write(
+            agent_dir.join("echo.py"),
+            "#!/usr/bin/env python3\nimport os, json\n\
+             state = json.loads(os.environ.get(\"GRAPH_STATE\", \"{}\"))\n\
+             print(json.dumps({\"output\": str(state[\"width\"])}))\n",
+        )
+        .unwrap();
+        fs::write(
+            &graph_path,
+            format!(
+                r#"
+name: {ECHO_INPUTS_AGENT}
+start: echo
+initial_state:
+  width: 1
+nodes:
+  echo:
+    type: script
+    script: echo.py
+    timeout: 30
+    next: done
+  done:
+    type: end
+    output: "{{{{output}}}}"
+"#
+            ),
+        )
+        .unwrap();
+    }
+
     const DOUBLER_GRAPH: &str = r#"
 name: t
 start: doubler
@@ -1784,6 +1824,71 @@ nodes:
         assert!(peers.0.is_finished(&peers.1.0));
     }
 
+    fn echo_inputs_parent_graph(inputs_block: &str) -> String {
+        format!(
+            r#"
+name: t
+start: worker
+nodes:
+  worker:
+    type: agent
+    agent: {ECHO_INPUTS_AGENT}
+    prompt: "p"
+{inputs_block}    state_updates:
+      output: "{{{{output}}}}"
+"#
+        )
+    }
+
+    async fn run_echo_inputs_chain(parent_yaml: &str) -> StateManager {
+        let h = Harness::new(parent_yaml);
+        let mut state = StateManager::new(HashMap::from([("n".to_string(), json!(3))]));
+        let mut ctx = silent_ctx();
+
+        h.run("worker", None, &mut state, &mut ctx)
+            .await
+            .unwrap_or_else(|e| panic!("chain failed: {e:#}"));
+
+        state
+    }
+
+    /// The child's `width` default is overlaid by the parent's `inputs`
+    /// before the child graph starts, so its script sees the parent's value.
+    #[tokio::test]
+    #[serial]
+    async fn agent_node_inputs_seed_the_child_graph_state() {
+        if !cmd_available("python3") {
+            eprintln!("skipping: python3 not available");
+            return;
+        }
+        let _guard = TestConfigDirGuard::new();
+        materialize_echo_inputs_agent();
+        let parent = echo_inputs_parent_graph("    inputs:\n      width: \"{{n}}\"\n");
+
+        let state = run_echo_inputs_chain(&parent).await;
+
+        assert_eq!(state.state().get("output"), Some(&json!("3")));
+    }
+
+    /// Control for the test above: the same child with no `inputs` on the
+    /// parent node keeps its YAML default, so an override in the sibling
+    /// test cannot be the child ignoring its own `initial_state`.
+    #[tokio::test]
+    #[serial]
+    async fn agent_node_without_inputs_leaves_the_child_graph_defaults() {
+        if !cmd_available("python3") {
+            eprintln!("skipping: python3 not available");
+            return;
+        }
+        let _guard = TestConfigDirGuard::new();
+        materialize_echo_inputs_agent();
+        let parent = echo_inputs_parent_graph("");
+
+        let state = run_echo_inputs_chain(&parent).await;
+
+        assert_eq!(state.state().get("output"), Some(&json!("1")));
+    }
+
     #[tokio::test]
     async fn run_item_chain_reaches_flagged_agent_after_script_step() {
         if !cmd_available("python3") {