use std::collections::HashMap; use ruff_python_ast::{ CmpOp, ElifElseClause, ExceptHandler, Expr, ExprAttribute, ExprAwait, ExprCall, ExprName, Number, Parameters, Stmt, StmtExpr, StmtFor, StmtFunctionDef, StmtIf, StmtReturn, StmtTry, StmtWhile, }; use ruff_python_parser::parse_module; use ruff_text_size::{Ranged, TextSize}; use crate::dag::{DagEdge, DagNode, DagNodeType, Param, WorkflowDag}; use crate::validation::{self, CompileError}; struct LineIndex { newline_offsets: Vec, } impl LineIndex { fn new(source: &str) -> Self { let mut offsets = vec![0]; for (i, c) in source.char_indices() { if c == '\n' { offsets.push(i + 1); } } Self { newline_offsets: offsets } } fn line_of(&self, byte_offset: usize) -> usize { match self.newline_offsets.binary_search(&byte_offset) { Ok(line) => line + 1, Err(line) => line, } } } /// Maps task function name → optional external path (from `@task(path="...")`) type TaskFunctions = HashMap>; /// First pass: scan top-level `@task async def foo(...)` declarations /// and `foo = task_script("path")` / `foo = task_flow("path")` assignments. fn collect_task_functions(stmts: &[Stmt]) -> TaskFunctions { let mut tasks = HashMap::new(); for stmt in stmts { if let Stmt::FunctionDef(func @ StmtFunctionDef { is_async: true, .. }) = stmt { for dec in &func.decorator_list { match &dec.expression { // @task (bare decorator) Expr::Name(ExprName { id, .. }) if id.as_str() == "task" => { tasks.insert(func.name.to_string(), None); } // @task(path="...") Expr::Call(call) => { if let Expr::Name(ExprName { id, .. }) = call.func.as_ref() { if id.as_str() == "task" { let path = extract_task_path_kwarg(call); tasks.insert(func.name.to_string(), path); } } } _ => {} } } } // foo = task_script("path") or foo = task_flow("path") if let Stmt::Assign(assign) = stmt { if let Expr::Call(call) = assign.value.as_ref() { if let Expr::Name(ExprName { id, .. }) = call.func.as_ref() { if id.as_str() == "task_script" || id.as_str() == "task_flow" { // Extract the path from the first positional argument let path = call.arguments.args.first().and_then(|arg| match arg { Expr::StringLiteral(s) => Some(s.value.to_str().to_string()), _ => None, }); // Extract variable name from target if let Some(Expr::Name(ExprName { id: var_name, .. })) = assign.targets.first() { tasks.insert(var_name.to_string(), path); } } } } } } tasks } /// Extract the `path=` keyword argument from a `@task(path="...")` call. fn extract_task_path_kwarg(call: &ExprCall) -> Option { for kw in &call.arguments.keywords { if let Some(ref arg) = kw.arg { if arg.as_str() == "path" { if let Expr::StringLiteral(s) = &kw.value { return Some(s.value.to_str().to_string()); } } } } None } struct WacWalker { nodes: Vec, edges: Vec, errors: Vec, node_counter: usize, line_index: LineIndex, task_functions: TaskFunctions, in_nested_func: bool, in_comprehension: bool, } impl WacWalker { fn new(source: &str, task_functions: TaskFunctions) -> Self { Self { nodes: Vec::new(), edges: Vec::new(), errors: Vec::new(), node_counter: 0, line_index: LineIndex::new(source), task_functions, in_nested_func: false, in_comprehension: false, } } fn next_id(&mut self) -> String { let id = format!("step_{}", self.node_counter); self.node_counter += 1; id } fn add_node(&mut self, node: DagNode) -> String { let id = node.id.clone(); self.nodes.push(node); id } fn add_edge(&mut self, from: &str, to: &str, label: Option) { self.edges .push(DagEdge { from: from.to_string(), to: to.to_string(), label }); } fn line_of_expr(&self, expr: &Expr) -> usize { let offset = match expr { Expr::Call(_) | Expr::Await(_) | Expr::Attribute(_) | Expr::Name(_) => { expr.start().to_usize() } _ => 0, }; self.line_index.line_of(offset) } fn line_of_stmt(&self, stmt: &Stmt) -> usize { let offset = match stmt { Stmt::If(_) | Stmt::For(_) | Stmt::While(_) | Stmt::Return(_) | Stmt::Expr(_) | Stmt::Try(_) | Stmt::Assign(_) | Stmt::AnnAssign(_) | Stmt::FunctionDef(_) => stmt.start().to_usize(), _ => 0, }; self.line_index.line_of(offset) } /// Check if an expression is a call to a known @task function fn is_task_fn_call(&self, expr: &Expr) -> bool { if let Expr::Call(call) = expr { if let Expr::Name(ExprName { id, .. }) = call.func.as_ref() { return self.task_functions.contains_key(id.as_str()); } } false } /// Check if an expression is `asyncio.gather(...)` call fn is_asyncio_gather_call(expr: &Expr) -> bool { if let Expr::Call(call) = expr { if let Expr::Attribute(ExprAttribute { value, attr, .. }) = call.func.as_ref() { if attr.as_str() == "gather" { if let Expr::Name(ExprName { id, .. }) = value.as_ref() { return id.as_str() == "asyncio"; } } } } false } /// Extract step name and script from a task function call. /// Name = function name, script = task_path or function name. fn extract_step_info_from_task_call(&self, call: &ExprCall) -> Option<(String, String)> { if let Expr::Name(ExprName { id, .. }) = call.func.as_ref() { let name = id.to_string(); let script = self .task_functions .get(id.as_str()) .and_then(|p| p.clone()) .unwrap_or_else(|| name.clone()); Some((name, script)) } else { None } } fn expr_to_source(expr: &Expr) -> String { match expr { Expr::Compare(c) => { let left = Self::expr_to_source(&c.operands[0]); if let Some(comparator) = c.comparators().first() { let right = Self::expr_to_source(comparator); let op = match c.ops.first() { Some(CmpOp::Gt) => ">", Some(CmpOp::Lt) => "<", Some(CmpOp::GtE) => ">=", Some(CmpOp::LtE) => "<=", Some(CmpOp::Eq) => "==", Some(CmpOp::NotEq) => "!=", Some(CmpOp::In) => "in", Some(CmpOp::NotIn) => "not in", Some(CmpOp::Is) => "is", Some(CmpOp::IsNot) => "is not", None => "?", }; format!("{left} {op} {right}") } else { left } } Expr::Subscript(s) => { let value = Self::expr_to_source(&s.value); let slice = Self::expr_to_source(&s.slice); format!("{value}[{slice}]") } Expr::Attribute(a) => { let value = Self::expr_to_source(&a.value); format!("{value}.{}", a.attr) } Expr::Name(n) => n.id.to_string(), Expr::StringLiteral(s) => format!("\"{}\"", s.value.to_str()), Expr::NumberLiteral(n) => match &n.value { Number::Int(i) => i.to_string(), Number::Float(f) => f.to_string(), Number::Complex { .. } => "...".to_string(), }, Expr::BooleanLiteral(b) => b.value.to_string(), Expr::NoneLiteral(_) => "None".to_string(), _ => "...".to_string(), } } /// Check if a statement body contains any task function calls (recursively) fn body_contains_step(&self, body: &[Stmt]) -> bool { for stmt in body { if self.stmt_contains_step(stmt) { return true; } } false } fn stmt_contains_step(&self, stmt: &Stmt) -> bool { match stmt { Stmt::Expr(StmtExpr { value, .. }) => self.expr_contains_step(value), Stmt::Assign(a) => self.expr_contains_step(&a.value), Stmt::If(s) => { self.body_contains_step(&s.body) || s.elif_else_clauses .iter() .any(|clause| self.body_contains_step(&clause.body)) } // `async for` is not modelled: its body is skipped like any unsupported statement Stmt::For(s) if !s.is_async => { self.body_contains_step(&s.body) || self.body_contains_step(&s.orelse) } Stmt::While(s) => { self.body_contains_step(&s.body) || self.body_contains_step(&s.orelse) } Stmt::Try(s) => { self.body_contains_step(&s.body) || self.body_contains_step(&s.orelse) || self.body_contains_step(&s.finalbody) || s.handlers.iter().any(|h| match h { ExceptHandler::ExceptHandler(eh) => self.body_contains_step(&eh.body), }) } Stmt::Return(_) => false, _ => false, } } fn expr_contains_step(&self, expr: &Expr) -> bool { if self.is_task_fn_call(expr) { return true; } if Self::is_sdk_call(expr) { return true; } match expr { Expr::Await(ExprAwait { value, .. }) => self.expr_contains_step(value), Expr::Call(call) => { if Self::is_asyncio_gather_call(expr) { return call .arguments .args .iter() .any(|a| self.expr_contains_step(a)); } false } _ => false, } } /// Check if expr is a call to a known SDK function (step, sleep, wait_for_approval) fn is_sdk_call(expr: &Expr) -> bool { if let Expr::Call(call) = expr { if let Expr::Name(ExprName { id, .. }) = call.func.as_ref() { let name = id.as_str(); return name == "step" || name == "sleep" || name == "wait_for_approval"; } } false } /// Walk a list of statements, returning (first_node_id, last_node_id) fn walk_body(&mut self, body: &[Stmt]) -> Option<(String, String)> { let mut first_id: Option = None; let mut prev_id: Option = None; for stmt in body { if let Some((stmt_first, stmt_last)) = self.walk_stmt(stmt) { if let Some(ref prev) = prev_id { self.add_edge(prev, &stmt_first, None); } if first_id.is_none() { first_id = Some(stmt_first); } prev_id = Some(stmt_last); } } match (first_id, prev_id) { (Some(f), Some(l)) => Some((f, l)), _ => None, } } fn walk_stmt(&mut self, stmt: &Stmt) -> Option<(String, String)> { match stmt { Stmt::Expr(StmtExpr { value, .. }) => self.walk_expr_stmt(value), Stmt::Assign(a) => self.walk_expr_stmt(&a.value), Stmt::If(if_stmt) => self.walk_if(if_stmt), Stmt::For(for_stmt) if !for_stmt.is_async => self.walk_for(for_stmt), Stmt::While(while_stmt) => self.walk_while(while_stmt), Stmt::Try(try_stmt) => self.walk_try(try_stmt), Stmt::Return(ret) => self.walk_return(ret), Stmt::FunctionDef(_) => { if self.stmt_contains_step(stmt) { self.errors.push(validation::error_step_in_nested_function( self.line_of_stmt(stmt), )); } None } _ => None, } } fn walk_expr_stmt(&mut self, expr: &Expr) -> Option<(String, String)> { // await task_fn(...) / await step(...) / await sleep(...) / await wait_for_approval(...) if let Expr::Await(ExprAwait { value, .. }) = expr { // await task_fn(...) if let Expr::Call(call) = value.as_ref() { if self.is_task_fn_call(value) { return self.emit_step(call, expr); } // Check for SDK-level calls: step(), sleep(), wait_for_approval() if let Some(result) = self.try_emit_sdk_call(call, expr) { return Some(result); } } // await asyncio.gather(task_fn(...), task_fn(...), ...) if Self::is_asyncio_gather_call(value) { if let Expr::Call(gather_call) = value.as_ref() { return self.emit_parallel(gather_call, expr); } } } // Bare task_fn() without await — validation error if self.is_task_fn_call(expr) { self.errors .push(validation::error_missing_await(self.line_of_expr(expr))); } None } /// Try to emit a node for SDK-level calls: step(), sleep(), wait_for_approval() fn try_emit_sdk_call(&mut self, call: &ExprCall, expr: &Expr) -> Option<(String, String)> { let callee_name = match call.func.as_ref() { Expr::Name(ExprName { id, .. }) => Some(id.as_str()), _ => None, }?; let line = self.line_of_expr(expr); match callee_name { "step" => { // step("name", fn) — extract the name from the first string argument let name = call .arguments .args .first() .and_then(|arg| match arg { Expr::StringLiteral(s) => Some(s.value.to_str().to_string()), _ => None, }) .unwrap_or_else(|| "step".to_string()); let id = self.next_id(); let node_id = self.add_node(DagNode { id: id.clone(), node_type: DagNodeType::InlineStep { name: name.clone() }, label: name, line, }); Some((node_id.clone(), node_id)) } "sleep" => { let seconds = call .arguments .args .first() .map(|arg| Self::expr_to_source(arg)) .unwrap_or_else(|| "?".to_string()); let id = self.next_id(); let node_id = self.add_node(DagNode { id: id.clone(), node_type: DagNodeType::Sleep { seconds: seconds.clone() }, label: format!("sleep({seconds})"), line, }); Some((node_id.clone(), node_id)) } "wait_for_approval" => { let id = self.next_id(); let node_id = self.add_node(DagNode { id: id.clone(), node_type: DagNodeType::WaitForApproval, label: "wait_for_approval".to_string(), line, }); Some((node_id.clone(), node_id)) } _ => None, } } fn emit_step(&mut self, call: &ExprCall, expr: &Expr) -> Option<(String, String)> { if self.in_nested_func { self.errors.push(validation::error_step_in_nested_function( self.line_of_expr(expr), )); return None; } if self.in_comprehension { self.errors.push(validation::error_step_in_comprehension( self.line_of_expr(expr), )); return None; } let (name, script) = self .extract_step_info_from_task_call(call) .unwrap_or(("unknown".into(), "unknown".into())); let id = self.next_id(); let node_id = self.add_node(DagNode { id: id.clone(), node_type: DagNodeType::Step { name: name.clone(), script }, label: name, line: self.line_of_expr(expr), }); Some((node_id.clone(), node_id)) } fn emit_parallel(&mut self, gather_call: &ExprCall, expr: &Expr) -> Option<(String, String)> { let line = self.line_of_expr(expr); let start_id = self.next_id(); let start_node_id = self.add_node(DagNode { id: start_id.clone(), node_type: DagNodeType::ParallelStart, label: "parallel".to_string(), line, }); let mut step_ids = Vec::new(); for arg in &gather_call.arguments.args { // Each arg should be task_fn(...) if let Expr::Call(call) = arg { if self.is_task_fn_call(arg) { let (name, script) = self .extract_step_info_from_task_call(call) .unwrap_or(("unknown".into(), "unknown".into())); let step_id = self.next_id(); let node_id = self.add_node(DagNode { id: step_id.clone(), node_type: DagNodeType::Step { name: name.clone(), script }, label: name, line: self.line_of_expr(arg), }); self.add_edge(&start_node_id, &node_id, None); step_ids.push(node_id); } } } let end_id = self.next_id(); let end_node_id = self.add_node(DagNode { id: end_id.clone(), node_type: DagNodeType::ParallelEnd, label: "join".to_string(), line, }); for step_id in &step_ids { self.add_edge(step_id, &end_node_id, None); } Some((start_node_id, end_node_id)) } fn walk_if(&mut self, if_stmt: &StmtIf) -> Option<(String, String)> { self.walk_if_chain( if_stmt.start(), &if_stmt.test, &if_stmt.body, &if_stmt.elif_else_clauses, ) } /// An `elif` is laid out as an `if` nested alone in the `else` of the clause before it. fn walk_if_chain( &mut self, start: TextSize, test: &Expr, body: &[Stmt], clauses: &[ElifElseClause], ) -> Option<(String, String)> { let has_steps_in_body = self.body_contains_step(body); let has_steps_in_else = clauses .iter() .any(|clause| self.body_contains_step(&clause.body)); if !has_steps_in_body && !has_steps_in_else { return None; } let line = self.line_index.line_of(start.to_usize()); let condition_source = Self::expr_to_source(test); let branch_id = self.next_id(); let branch_node_id = self.add_node(DagNode { id: branch_id.clone(), node_type: DagNodeType::Branch { condition_source }, label: "if".to_string(), line, }); let mut last_ids = Vec::new(); if let Some((true_first, true_last)) = self.walk_body(body) { self.add_edge(&branch_node_id, &true_first, Some("true".to_string())); last_ids.push(true_last); } else { last_ids.push(branch_node_id.clone()); } if let Some((clause, clauses)) = clauses.split_first() { let orelse = match &clause.test { Some(test) => self.walk_if_chain(clause.start(), test, &clause.body, clauses), None => self.walk_body(&clause.body), }; if let Some((else_first, else_last)) = orelse { self.add_edge(&branch_node_id, &else_first, Some("false".to_string())); last_ids.push(else_last); } else { last_ids.push(branch_node_id.clone()); } } if last_ids.len() == 1 { Some((branch_node_id, last_ids.into_iter().next().unwrap())) } else { let merge_id = format!("{branch_id}_merge"); let merge_node_id = self.add_node(DagNode { id: merge_id, node_type: DagNodeType::Merge, label: "merge".to_string(), line, }); for last in last_ids { self.add_edge(&last, &merge_node_id, None); } Some((branch_node_id, merge_node_id)) } } fn walk_for(&mut self, for_stmt: &StmtFor) -> Option<(String, String)> { if !self.body_contains_step(&for_stmt.body) { return None; } let line = self.line_index.line_of(for_stmt.start().to_usize()); let iter_source = Self::expr_to_source(&for_stmt.iter); let start_id = self.next_id(); let start_node_id = self.add_node(DagNode { id: start_id.clone(), node_type: DagNodeType::LoopStart { iter_source }, label: "for".to_string(), line, }); if let Some((body_first, body_last)) = self.walk_body(&for_stmt.body) { self.add_edge(&start_node_id, &body_first, None); self.add_edge(&body_last, &start_node_id, Some("next".to_string())); } let end_id = self.next_id(); let end_node_id = self.add_node(DagNode { id: end_id.clone(), node_type: DagNodeType::LoopEnd, label: "end for".to_string(), line, }); self.add_edge(&start_node_id, &end_node_id, Some("done".to_string())); Some((start_node_id, end_node_id)) } fn walk_while(&mut self, while_stmt: &StmtWhile) -> Option<(String, String)> { if !self.body_contains_step(&while_stmt.body) { return None; } let line = self.line_index.line_of(while_stmt.start().to_usize()); let condition = Self::expr_to_source(&while_stmt.test); let start_id = self.next_id(); let start_node_id = self.add_node(DagNode { id: start_id.clone(), node_type: DagNodeType::LoopStart { iter_source: condition }, label: "while".to_string(), line, }); if let Some((body_first, body_last)) = self.walk_body(&while_stmt.body) { self.add_edge(&start_node_id, &body_first, None); self.add_edge(&body_last, &start_node_id, Some("next".to_string())); } let end_id = self.next_id(); let end_node_id = self.add_node(DagNode { id: end_id.clone(), node_type: DagNodeType::LoopEnd, label: "end while".to_string(), line, }); self.add_edge(&start_node_id, &end_node_id, Some("done".to_string())); Some((start_node_id, end_node_id)) } fn walk_try(&mut self, try_stmt: &StmtTry) -> Option<(String, String)> { let has_steps = self.body_contains_step(&try_stmt.body) || self.body_contains_step(&try_stmt.orelse) || self.body_contains_step(&try_stmt.finalbody) || try_stmt.handlers.iter().any(|h| match h { ExceptHandler::ExceptHandler(eh) => self.body_contains_step(&eh.body), }); if !has_steps { return None; } let line = self.line_index.line_of(try_stmt.start().to_usize()); self.emit_try_catch_branch( &try_stmt.body, &try_stmt.handlers, &try_stmt.finalbody, line, ) } fn emit_try_catch_branch( &mut self, try_body: &[Stmt], handlers: &[ExceptHandler], finally_body: &[Stmt], line: usize, ) -> Option<(String, String)> { let branch_id = self.next_id(); let branch_node_id = self.add_node(DagNode { id: branch_id.clone(), node_type: DagNodeType::Branch { condition_source: "try/except".to_string() }, label: "try".to_string(), line, }); let mut last_ids = Vec::new(); // Try body if let Some((try_first, try_last)) = self.walk_body(try_body) { self.add_edge(&branch_node_id, &try_first, Some("try".to_string())); last_ids.push(try_last); } else { last_ids.push(branch_node_id.clone()); } // Except handlers for handler in handlers { match handler { ExceptHandler::ExceptHandler(eh) => { if let Some((catch_first, catch_last)) = self.walk_body(&eh.body) { self.add_edge(&branch_node_id, &catch_first, Some("except".to_string())); last_ids.push(catch_last); } } } } // Finally body — sequential after merge let merge_last = if last_ids.len() == 1 { last_ids.into_iter().next().unwrap() } else { let merge_id = format!("{branch_id}_merge"); let merge_node_id = self.add_node(DagNode { id: merge_id, node_type: DagNodeType::Merge, label: "merge".to_string(), line, }); for last in last_ids { self.add_edge(&last, &merge_node_id, None); } merge_node_id }; if !finally_body.is_empty() { if let Some((finally_first, finally_last)) = self.walk_body(finally_body) { self.add_edge(&merge_last, &finally_first, None); return Some((branch_node_id, finally_last)); } } Some((branch_node_id, merge_last)) } fn walk_return(&mut self, ret: &StmtReturn) -> Option<(String, String)> { let line = self.line_index.line_of(ret.start().to_usize()); let id = self.next_id(); let node_id = self.add_node(DagNode { id: id.clone(), node_type: DagNodeType::Return, label: "return".to_string(), line, }); Some((node_id.clone(), node_id)) } } /// Extract workflow function parameters (no longer skips ctx) fn extract_params(args: &Parameters) -> Vec { let mut params = Vec::new(); for arg_with_default in args.args.iter().chain(args.posonlyargs.iter()) { let name = arg_with_default.parameter.name.to_string(); let typ = arg_with_default .parameter .annotation .as_ref() .map(|ann| WacWalker::expr_to_source(ann)); params.push(Param { name, typ }); } params } pub fn parse_python_workflow(code: &str) -> Result> { let ast = parse_module(code) .map_err(|e| vec![CompileError { message: format!("Parse error: {e}"), line: 0 }])? .into_suite(); // First pass: collect @task functions let task_functions = collect_task_functions(&ast); let workflow_fns = || { ast.iter().filter_map(|stmt| match stmt { Stmt::FunctionDef(func) if func.decorator_list.iter().any(|dec| { matches!(&dec.expression, Expr::Name(ExprName { id, .. }) if id.as_str() == "workflow") }) => { Some(func) } _ => None, }) }; // Find the @workflow async def let workflow_fn = workflow_fns().find(|func| func.is_async); // Check for non-async workflow function. The statement starts at its first decorator, // the name sits on the `def` line the error points at. let non_async_workflow = workflow_fns() .find(|func| !func.is_async) .map(|func| LineIndex::new(code).line_of(func.name.start().to_usize())); if let Some(line) = non_async_workflow { if workflow_fn.is_none() { return Err(vec![validation::error_not_async(line)]); } } let workflow_fn = workflow_fn.ok_or_else(|| { vec![CompileError { message: "No @workflow async function found.".to_string(), line: 0 }] })?; let params = extract_params(&workflow_fn.parameters); let source_hash = compute_source_hash(code); let mut walker = WacWalker::new(code, task_functions); walker.walk_body(&workflow_fn.body); if !walker.errors.is_empty() { return Err(walker.errors); } Ok(WorkflowDag { nodes: walker.nodes, edges: walker.edges, params, source_hash }) } fn compute_source_hash(code: &str) -> String { use sha2::{Digest, Sha256}; let mut hasher = Sha256::new(); hasher.update(code.as_bytes()); format!("{:x}", hasher.finalize()) }