diff --git a/src/config/request_context.rs b/src/config/request_context.rs index e596ad7..d5b36f9 100644 --- a/src/config/request_context.rs +++ b/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 { + let mut lines: Vec = 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 { + 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] diff --git a/src/graph/map.rs b/src/graph/map.rs index e61eb1d..5ca4bdd 100644 --- a/src/graph/map.rs +++ b/src/graph/map.rs @@ -740,6 +740,119 @@ nodes: assert_eq!(collected(&result), vec![json!(3), json!(5), json!(7)]); } + const RETRY_CHAIN_GRAPH: &str = r#" +name: retry-chain +start: fan_out +settings: + validate_before_run: true +initial_state: + items: ["a", "b", "c"] +nodes: + fan_out: + type: map + over: "{{items}}" + as: item + branch: draft + output_key: result + collect_into: results + next: done + draft: + type: script + script: draft.py + next: check + check: + type: script + script: check.py + done: + type: end + output: "{{results}}" +"#; + + fn write_retry_chain_scripts(ws: &TestWorkspace) { + ws.write_py( + "draft.py", + r##"attempts = state.get("attempts", 0) + 1 +print(json.dumps({"attempts": attempts, "draft": f"{state['item']}#{attempts}"}))"##, + ); + ws.write_py( + "check.py", + r#"if state["attempts"] < 2: + print(json.dumps({"_next": "draft"})) +else: + print(json.dumps({"result": {"item": state["item"], "draft": state["draft"], "attempts": state["attempts"]}}))"#, + ); + } + + fn retry_chain_results() -> Vec { + ["a", "b", "c"] + .iter() + .map(|item| json!({"item": item, "draft": format!("{item}#2"), "attempts": 2})) + .collect() + } + + #[tokio::test] + async fn map_chain_retries_via_script_next_until_check_accepts() { + if !cmd_available("python3") { + eprintln!("skipping: python3 not available"); + return; + } + let ws = TestWorkspace::new(); + write_retry_chain_scripts(&ws); + + let result = run_graph(RETRY_CHAIN_GRAPH, &ws) + .await + .unwrap_or_else(|e| panic!("validated retry chain failed: {e:#}")); + + assert_eq!(collected(&result), retry_chain_results()); + } + + #[tokio::test] + async fn map_chain_retries_leave_parent_state_and_loop_counts_untouched() { + if !cmd_available("python3") { + eprintln!("skipping: python3 not available"); + return; + } + let ws = TestWorkspace::new(); + write_retry_chain_scripts(&ws); + let graph: Arc = Arc::new(serde_yaml::from_str(RETRY_CHAIN_GRAPH).unwrap()); + let NodeType::Map(map_node) = &graph.get_node("fan_out").unwrap().node_type else { + panic!("fan_out should be a map node"); + }; + let script = ScriptExecutor::new(&ws.dir); + let abort = create_abort_signal(); + let step_ctx = StepContext { + graph: Arc::clone(&graph), + script_executor: &script, + max_concurrency: 4, + abort_signal: &abort, + branch_mode: false, + }; + let items = json!(["a", "b", "c"]); + let mut parent = StateManager::new(HashMap::from([("items".to_string(), items.clone())])); + let mut ctx = silent_ctx(); + + MapNodeExecutor::execute(map_node, &mut parent, &mut ctx, &step_ctx, "fan_out") + .await + .unwrap_or_else(|e| panic!("map failed: {e:#}")); + + // Only `collect_into` lands on the parent: the per-item `attempts`, + // `draft` and `result` writes were fork-local scratch. + assert_eq!( + *parent.state().data(), + HashMap::from([ + ("items".to_string(), items), + ("results".to_string(), Value::Array(retry_chain_results())), + ]) + ); + for node in ["fan_out", "draft", "check"] { + assert_eq!( + parent.state().loop_count(node), + 0, + "parent loop count for '{node}'" + ); + } + } + #[tokio::test] async fn map_chain_script_next_loop_is_bounded_by_max_loop_iterations() { if !cmd_available("python3") {