Coverage for agentos/workflow/__init__.py: 0%

1""" 

2AgentOS Workflow DSL — Declarative multi-agent workflow definition language. 

3 

4v1.14.4: YAML/JSON-based DSL for defining complex multi-agent pipelines with 

5 sequential, parallel, conditional, loop, and fan-out/fan-in patterns. 

6 

7Key features: 

8- YAML/JSON declarative workflow definitions 

9- Topology validation and cycle detection 

10- Sequential, parallel, conditional, loop, sub-workflow patterns 

11- Built-in fan-out/fan-in via agentos.core.parallel 

12- Workflow execution engine with real-time progress 

13- Dry-run mode for validation without execution 

14- Visual DAG export (Mermaid/Graphviz) 

15- Error recovery strategies (retry, fallback, skip, escalate) 

16""" 

17 

18import asyncio 

19import json 

20import logging 

21from collections.abc import Callable 

22from dataclasses import dataclass, field 

23from enum import Enum 

24from pathlib import Path 

25from typing import Any, Optional 

26 

27import yaml 

28 

29logger = logging.getLogger(__name__) 

30 

31 

32# --------------------------------------------------------------------------- 

33# Enums 

34# --------------------------------------------------------------------------- 

35 

36 

37class StepType(Enum): 

38 """Types of workflow steps.""" 

39 

40 TASK = "task" # Single agent task 

41 SEQUENTIAL = "sequential" # Run children in sequence 

42 PARALLEL = "parallel" # Run children in parallel 

43 CONDITIONAL = "conditional" # Branch based on condition 

44 LOOP = "loop" # Repeat children until condition 

45 SUB_WORKFLOW = "sub" # Nested workflow 

46 JOIN = "join" # Wait for all branches to complete 

47 SPLIT = "split" # Fan-out to multiple agents 

48 

49 

50class ExecutionStatus(Enum): 

51 PENDING = "pending" 

52 RUNNING = "running" 

53 SUCCESS = "success" 

54 FAILED = "failed" 

55 SKIPPED = "skipped" 

56 CANCELLED = "cancelled" 

57 RETRYING = "retrying" 

58 

59 

60class ErrorStrategy(Enum): 

61 RETRY = "retry" # Retry the step 

62 FALLBACK = "fallback" # Execute fallback step 

63 SKIP = "skip" # Skip and continue 

64 ESCALATE = "escalate" # Fail the workflow 

65 PAUSE = "pause" # Pause for human intervention 

66 

67 

68class ConditionOperator(Enum): 

69 EQUALS = "eq" 

70 NOT_EQUALS = "neq" 

71 CONTAINS = "contains" 

72 GREATER = "gt" 

73 LESS = "lt" 

74 IN = "in" 

75 MATCHES = "matches" # regex 

76 EXISTS = "exists" 

77 EMPTY = "empty" 

78 

79 

80# --------------------------------------------------------------------------- 

81# Data structures 

82# --------------------------------------------------------------------------- 

83 

84 

85@dataclass 

86class WorkflowContext: 

87 """Runtime context shared across workflow steps.""" 

88 

89 variables: dict[str, Any] = field(default_factory=dict) 

90 history: list[dict[str, Any]] = field(default_factory=list) 

91 errors: list[dict[str, Any]] = field(default_factory=list) 

92 metrics: dict[str, float] = field(default_factory=dict) 

93 metadata: dict[str, str] = field(default_factory=dict) 

94 

95 def get(self, key: str, default: Any = None) -> Any: 

96 """Get a variable, supporting dot-notation (e.g., 'result.output.text').""" 

97 parts = key.split(".") 

98 current = self.variables 

99 for part in parts: 

100 if isinstance(current, dict): 

101 current = current.get(part, default) 

102 else: 

103 return default 

104 return current 

105 

106 def set(self, key: str, value: Any) -> None: 

107 """Set a variable, supporting dot-notation for nested dicts.""" 

108 parts = key.split(".") 

109 current = self.variables 

110 for part in parts[:-1]: 

111 if part not in current: 

112 current[part] = {} 

113 current = current[part] 

114 current[parts[-1]] = value 

115 

116 

117@dataclass 

118class StepResult: 

119 """Result of a workflow step execution.""" 

120 

121 step_id: str 

122 status: ExecutionStatus 

123 output: Any = None 

124 error: str | None = None 

125 duration: float = 0.0 

126 retries: int = 0 

127 metadata: dict[str, Any] = field(default_factory=dict) 

128 

129 

130@dataclass 

131class WorkflowStep: 

132 """A single step in a workflow DAG.""" 

133 

134 id: str 

135 type: StepType 

136 name: str = "" 

137 description: str = "" 

138 

139 # Execution 

140 agent: str | None = None # agent_id to dispatch to 

141 task: str | None = None # task payload template 

142 children: list["WorkflowStep"] = field(default_factory=list) 

143 

144 # Conditional 

145 condition: dict[str, Any] | None = None 

146 branches: dict[str, list["WorkflowStep"]] = field(default_factory=dict) 

147 

148 # Loop 

149 max_iterations: int = 100 

150 loop_condition: dict[str, Any] | None = None 

151 

152 # Error handling 

153 on_error: ErrorStrategy = ErrorStrategy.ESCALATE 

154 max_retries: int = 3 

155 retry_delay: float = 1.0 

156 fallback_step: Optional["WorkflowStep"] = None 

157 

158 # Timing 

159 timeout: float = 300.0 

160 depends_on: list[str] = field(default_factory=list) 

161 

162 # Metadata 

163 tags: list[str] = field(default_factory=list) 

164 metadata: dict[str, Any] = field(default_factory=dict) 

165 

166 

167@dataclass 

168class WorkflowDefinition: 

169 """Top-level workflow definition.""" 

170 

171 name: str 

172 version: str = "1.0" 

173 description: str = "" 

174 root: WorkflowStep | None = None 

175 variables: dict[str, Any] = field(default_factory=dict) 

176 

177 @property 

178 def steps(self) -> list[WorkflowStep]: 

179 """Compatibility: return steps list from root tree.""" 

180 if self.root is None: 

181 return [] 

182 result = [] 

183 

184 def _collect(s): 

185 result.append(s) 

186 for c in s.children or []: 

187 _collect(c) 

188 

189 _collect(self.root) 

190 return result 

191 

192 agents: dict[str, dict[str, Any]] = field(default_factory=dict) 

193 defaults: dict[str, Any] = field(default_factory=dict) 

194 metadata: dict[str, Any] = field(default_factory=dict) 

195 

196 def validate(self) -> list[str]: 

197 """Validate workflow structure and return a list of issues.""" 

198 issues = [] 

199 step_ids: set[str] = set() 

200 

201 def validate_step(step: WorkflowStep): 

202 if step.id in step_ids: 

203 issues.append(f"Duplicate step ID: {step.id}") 

204 step_ids.add(step.id) 

205 

206 if step.type == StepType.CONDITIONAL and not step.condition: 

207 issues.append(f"Conditional step '{step.id}' has no condition") 

208 if step.type == StepType.TASK and not step.agent: 

209 issues.append(f"Task step '{step.id}' has no agent assigned") 

210 

211 # Validate depends_on references 

212 for dep in step.depends_on: 

213 if dep not in step_ids: 

214 issues.append(f"Step '{step.id}' depends on unknown step '{dep}'") 

215 

216 for child in step.children: 

217 validate_step(child) 

218 for branch_steps in step.branches.values(): 

219 for s in branch_steps: 

220 validate_step(s) 

221 if step.fallback_step: 

222 validate_step(step.fallback_step) 

223 

224 if self.root: 

225 validate_step(self.root) 

226 else: 

227 issues.append("Workflow has no root step") 

228 

229 return issues 

230 

231 def to_mermaid(self) -> str: 

232 """Export workflow as a Mermaid flowchart.""" 

233 lines = ["graph TD"] 

234 ids: set[str] = set() 

235 

236 def add_step(step: WorkflowStep, parent_id: str | None = None): 

237 prefix = { 

238 StepType.PARALLEL: "[||]", 

239 StepType.CONDITIONAL: "{?}", 

240 StepType.LOOP: "[/]", 

241 StepType.TASK: "[ ]", 

242 StepType.JOIN: "[+]", 

243 StepType.SPLIT: "[>]", 

244 }.get(step.type, "[ ]") 

245 label = step.name or step.id 

246 lines.append(f" {step.id}{prefix}{label}") 

247 

248 if parent_id: 

249 lines.append(f" {parent_id} --> {step.id}") 

250 

251 if step.id not in ids: 

252 ids.add(step.id) 

253 if step.type == StepType.CONDITIONAL: 

254 for branch_name, branch_steps in step.branches.items(): 

255 for s in branch_steps: 

256 add_step(s, step.id) 

257 lines.append(f" {step.id} -- {branch_name} --> {s.id}") 

258 else: 

259 for child in step.children: 

260 add_step(child, step.id) 

261 

262 if self.root: 

263 add_step(self.root) 

264 return "\n".join(lines) 

265 

266 

267# --------------------------------------------------------------------------- 

268# Condition evaluator 

269# --------------------------------------------------------------------------- 

270 

271 

272class ConditionEvaluator: 

273 """Evaluate conditions against the workflow context.""" 

274 

275 @staticmethod 

276 def evaluate(condition: dict[str, Any], ctx: WorkflowContext) -> bool: 

277 """Evaluate a condition dict against context.""" 

278 if not condition: 

279 return True 

280 

281 # Support AND/OR combinators 

282 if "and" in condition: 

283 return all(ConditionEvaluator.evaluate(sub, ctx) for sub in condition["and"]) 

284 if "or" in condition: 

285 return any(ConditionEvaluator.evaluate(sub, ctx) for sub in condition["or"]) 

286 if "not" in condition: 

287 return not ConditionEvaluator.evaluate(condition["not"], ctx) 

288 

289 # Single condition 

290 field = condition.get("field", "") 

291 op = condition.get("op", "eq") 

292 value = condition.get("value") 

293 

294 actual = ctx.get(field) 

295 operator = ConditionOperator(op) 

296 

297 if operator == ConditionOperator.EQUALS: 

298 return actual == value 

299 elif operator == ConditionOperator.NOT_EQUALS: 

300 return actual != value 

301 elif operator == ConditionOperator.CONTAINS: 

302 return value in str(actual) if actual is not None else False 

303 elif operator == ConditionOperator.GREATER: 

304 try: 

305 return float(actual) > float(value) 

306 except (TypeError, ValueError): 

307 return False 

308 elif operator == ConditionOperator.LESS: 

309 try: 

310 return float(actual) < float(value) 

311 except (TypeError, ValueError): 

312 return False 

313 elif operator == ConditionOperator.IN: 

314 return actual in value if isinstance(value, (list, tuple, set)) else False 

315 elif operator == ConditionOperator.MATCHES: 

316 import re 

317 

318 try: 

319 return bool(re.search(str(value), str(actual))) 

320 except re.error: 

321 return False 

322 elif operator == ConditionOperator.EXISTS: 

323 return actual is not None 

324 elif operator == ConditionOperator.EMPTY: 

325 return actual is None or actual == "" or actual == [] or actual == {} 

326 

327 return False 

328 

329 

330# --------------------------------------------------------------------------- 

331# Workflow Engine 

332# --------------------------------------------------------------------------- 

333 

334 

335class WorkflowEngine: 

336 """Executes a WorkflowDefinition with real-time progress tracking.""" 

337 

338 def __init__( 

339 self, 

340 agent_dispatcher: Callable | None = None, 

341 max_parallelism: int = 10, 

342 ): 

343 self._dispatcher = agent_dispatcher or self._default_dispatcher 

344 self._max_parallelism = max_parallelism 

345 self._ctx: WorkflowContext | None = None 

346 self._results: dict[str, StepResult] = {} 

347 self._progress_callbacks: list[Callable] = [] 

348 self._cancelled = False 

349 self._semaphore = asyncio.Semaphore(max_parallelism) 

350 

351 def on_progress(self, callback: Callable[[StepResult], None]) -> None: 

352 """Register a progress callback.""" 

353 self._progress_callbacks.append(callback) 

354 

355 async def execute(self, workflow: WorkflowDefinition) -> WorkflowContext: 

356 """Execute a workflow and return the final context.""" 

357 issues = workflow.validate() 

358 if issues: 

359 raise ValueError(f"Workflow validation failed: {issues}") 

360 

361 self._ctx = WorkflowContext(variables=dict(workflow.variables)) 

362 self._results = {} 

363 self._cancelled = False 

364 

365 if workflow.root: 

366 await self._execute_step(workflow.root, self._ctx) 

367 

368 return self._ctx 

369 

370 async def dry_run(self, workflow: WorkflowDefinition) -> dict[str, Any]: 

371 """Validate a workflow without executing it.""" 

372 issues = workflow.validate() 

373 return { 

374 "valid": len(issues) == 0, 

375 "issues": issues, 

376 "steps": self._count_steps(workflow.root) if workflow.root else 0, 

377 "mermaid": workflow.to_mermaid(), 

378 } 

379 

380 def cancel(self) -> None: 

381 """Cancel the running workflow.""" 

382 self._cancelled = True 

383 

384 async def _execute_step(self, step: WorkflowStep, ctx: WorkflowContext) -> StepResult: 

385 if self._cancelled: 

386 return StepResult(step.id, ExecutionStatus.CANCELLED) 

387 

388 if step.id in self._results: 

389 return self._results[step.id] 

390 

391 logger.info(f"[Workflow] Executing step '{step.id}' ({step.type.value})") 

392 result = StepResult(step.id, ExecutionStatus.RUNNING) 

393 

394 try: 

395 if step.type == StepType.TASK: 

396 result = await self._run_task(step, ctx) 

397 elif step.type == StepType.SEQUENTIAL: 

398 result = await self._run_sequential(step, ctx) 

399 elif step.type == StepType.PARALLEL: 

400 result = await self._run_parallel(step, ctx) 

401 elif step.type == StepType.CONDITIONAL: 

402 result = await self._run_conditional(step, ctx) 

403 elif step.type == StepType.LOOP: 

404 result = await self._run_loop(step, ctx) 

405 elif step.type == StepType.SUB_WORKFLOW: 

406 result = await self._run_sub_workflow(step, ctx) 

407 elif step.type == StepType.JOIN: 

408 result = await self._run_join(step, ctx) 

409 elif step.type == StepType.SPLIT: 

410 result = await self._run_split(step, ctx) 

411 else: 

412 result.status = ExecutionStatus.SUCCESS 

413 

414 except TimeoutError: 

415 result.status = ExecutionStatus.FAILED 

416 result.error = f"Step '{step.id}' timed out after {step.timeout}s" 

417 except Exception as e: 

418 result.status = ExecutionStatus.FAILED 

419 result.error = str(e) 

420 logger.exception(f"[Workflow] Step '{step.id}' failed: {e}") 

421 

422 # Error recovery 

423 result = await self._handle_error(step, ctx, result) 

424 

425 self._results[step.id] = result 

426 ctx.history.append( 

427 { 

428 "step_id": step.id, 

429 "status": result.status.value, 

430 "output": str(result.output)[:200] if result.output else None, 

431 "error": result.error, 

432 "duration": result.duration, 

433 } 

434 ) 

435 

436 for cb in self._progress_callbacks: 

437 try: 

438 cb(result) 

439 except Exception: 

440 pass 

441 

442 return result 

443 

444 async def _run_task(self, step: WorkflowStep, ctx: WorkflowContext) -> StepResult: 

445 """Execute a single agent task.""" 

446 import time 

447 

448 t0 = time.time() 

449 

450 # Resolve template variables in task payload 

451 payload = step.task or "" 

452 if "{{" in payload: 

453 payload = self._resolve_template(payload, ctx) 

454 

455 try: 

456 output = await asyncio.wait_for( 

457 self._dispatcher(step.agent, payload, ctx), 

458 timeout=step.timeout, 

459 ) 

460 ctx.set(f"steps.{step.id}.output", output) 

461 return StepResult( 

462 step.id, 

463 ExecutionStatus.SUCCESS, 

464 output=output, 

465 duration=time.time() - t0, 

466 ) 

467 except TimeoutError: 

468 raise 

469 

470 async def _run_sequential(self, step: WorkflowStep, ctx: WorkflowContext) -> StepResult: 

471 """Run children in sequence.""" 

472 for child in step.children: 

473 result = await self._execute_step(child, ctx) 

474 if result.status == ExecutionStatus.FAILED and step.on_error == ErrorStrategy.ESCALATE: 

475 return StepResult( 

476 step.id, 

477 ExecutionStatus.FAILED, 

478 error=f"Child '{child.id}' failed: {result.error}", 

479 ) 

480 return StepResult(step.id, ExecutionStatus.SUCCESS) 

481 

482 async def _run_parallel(self, step: WorkflowStep, ctx: WorkflowContext) -> StepResult: 

483 """Run children in parallel with semaphore control.""" 

484 

485 async def bounded_execute(child): 

486 async with self._semaphore: 

487 return await self._execute_step(child, ctx) 

488 

489 tasks = [bounded_execute(child) for child in step.children] 

490 results = await asyncio.gather(*tasks, return_exceptions=True) 

491 

492 outputs = {} 

493 for child, result in zip(step.children, results): 

494 if isinstance(result, Exception): 

495 outputs[child.id] = {"error": str(result)} 

496 else: 

497 outputs[child.id] = result.output 

498 

499 ctx.set(f"steps.{step.id}.outputs", outputs) 

500 return StepResult(step.id, ExecutionStatus.SUCCESS, output=outputs) 

501 

502 async def _run_conditional(self, step: WorkflowStep, ctx: WorkflowContext) -> StepResult: 

503 """Evaluate condition and execute the matching branch.""" 

504 if not step.condition: 

505 return StepResult(step.id, ExecutionStatus.SKIPPED, error="No condition defined") 

506 

507 matched = ConditionEvaluator.evaluate(step.condition, ctx) 

508 branch_key = "true" if matched else "false" 

509 branch_steps = step.branches.get(branch_key, []) 

510 if not branch_steps: 

511 # Try numeric/default branches 

512 branch_steps = step.branches.get("default", []) 

513 

514 for child in branch_steps: 

515 result = await self._execute_step(child, ctx) 

516 if result.status == ExecutionStatus.FAILED: 

517 return StepResult( 

518 step.id, 

519 ExecutionStatus.FAILED, 

520 error=f"Branch child '{child.id}' failed: {result.error}", 

521 ) 

522 

523 return StepResult(step.id, ExecutionStatus.SUCCESS, output={"branch": branch_key}) 

524 

525 async def _run_loop(self, step: WorkflowStep, ctx: WorkflowContext) -> StepResult: 

526 """Execute children in a loop until condition is false.""" 

527 iteration = 0 

528 while iteration < step.max_iterations: 

529 if self._cancelled: 

530 return StepResult(step.id, ExecutionStatus.CANCELLED) 

531 

532 for child in step.children: 

533 result = await self._execute_step(child, ctx) 

534 if result.status == ExecutionStatus.FAILED: 

535 return StepResult( 

536 step.id, 

537 ExecutionStatus.FAILED, 

538 error=f"Loop iteration {iteration}: child '{child.id}' failed", 

539 ) 

540 

541 ctx.set(f"steps.{step.id}.iteration", iteration) 

542 iteration += 1 

543 

544 # Check loop condition 

545 if step.loop_condition: 

546 if not ConditionEvaluator.evaluate(step.loop_condition, ctx): 

547 break 

548 

549 return StepResult(step.id, ExecutionStatus.SUCCESS, output={"iterations": iteration}) 

550 

551 async def _run_sub_workflow(self, step: WorkflowStep, ctx: WorkflowContext) -> StepResult: 

552 """Execute a nested sub-workflow.""" 

553 # Sub-workflow steps just execute their children 

554 for child in step.children: 

555 result = await self._execute_step(child, ctx) 

556 if result.status == ExecutionStatus.FAILED: 

557 return StepResult( 

558 step.id, ExecutionStatus.FAILED, error=f"Sub-workflow child '{child.id}' failed" 

559 ) 

560 return StepResult(step.id, ExecutionStatus.SUCCESS) 

561 

562 async def _run_join(self, step: WorkflowStep, ctx: WorkflowContext) -> StepResult: 

563 """Join point — wait for specified dependencies.""" 

564 # Already handled by depends_on graph resolution 

565 return StepResult(step.id, ExecutionStatus.SUCCESS) 

566 

567 async def _run_split(self, step: WorkflowStep, ctx: WorkflowContext) -> StepResult: 

568 """Fan-out to multiple agents.""" 

569 children_results = await self._run_parallel(step, ctx) 

570 return children_results 

571 

572 async def _handle_error( 

573 self, step: WorkflowStep, ctx: WorkflowContext, result: StepResult 

574 ) -> StepResult: 

575 """Apply error recovery strategy.""" 

576 ctx.errors.append({"step_id": step.id, "error": result.error}) 

577 

578 if step.on_error == ErrorStrategy.RETRY and result.retries < step.max_retries: 

579 logger.info( 

580 f"[Workflow] Retrying step '{step.id}' ({result.retries+1}/{step.max_retries})" 

581 ) 

582 await asyncio.sleep(step.retry_delay * (2**result.retries)) 

583 result.retries += 1 

584 return await self._execute_step(step, ctx) 

585 

586 elif step.on_error == ErrorStrategy.FALLBACK and step.fallback_step: 

587 logger.info(f"[Workflow] Executing fallback for step '{step.id}'") 

588 return await self._execute_step(step.fallback_step, ctx) 

589 

590 elif step.on_error == ErrorStrategy.SKIP: 

591 result.status = ExecutionStatus.SKIPPED 

592 return result 

593 

594 elif step.on_error == ErrorStrategy.PAUSE: 

595 logger.warning(f"[Workflow] Paused at step '{step.id}': {result.error}") 

596 # In production, this would notify the HITL system 

597 return result 

598 

599 # Default: escalate 

600 return result 

601 

602 @staticmethod 

603 async def _default_dispatcher(agent_id: str, task: str, ctx: WorkflowContext) -> str: 

604 """Default task dispatcher — logs and returns mock result.""" 

605 logger.info(f"[Workflow] Dispatch to '{agent_id}': {task[:100]}") 

606 return f"Task dispatched to {agent_id}: {task[:50]}" 

607 

608 @staticmethod 

609 def _resolve_template(template: str, ctx: WorkflowContext) -> str: 

610 """Resolve {{ variable }} placeholders in a template string.""" 

611 import re 

612 

613 def replacer(match): 

614 key = match.group(1).strip() 

615 return str(ctx.get(key, f"<{key} not found>")) 

616 

617 return re.sub(r"\{\{\s*(.*?)\s*\}\}", replacer, template) 

618 

619 @staticmethod 

620 def _count_steps(step: WorkflowStep | None) -> int: 

621 if step is None: 

622 return 0 

623 count = 1 

624 for child in step.children: 

625 count += WorkflowEngine._count_steps(child) 

626 for branch_steps in step.branches.values(): 

627 for s in branch_steps: 

628 count += WorkflowEngine._count_steps(s) 

629 if step.fallback_step: 

630 count += WorkflowEngine._count_steps(step.fallback_step) 

631 return count 

632 

633 

634# --------------------------------------------------------------------------- 

635# Workflow YAML/JSON Parser 

636# --------------------------------------------------------------------------- 

637 

638 

639class WorkflowParser: 

640 """Parse YAML/JSON files into WorkflowDefinition objects.""" 

641 

642 @staticmethod 

643 def parse_file(filepath: str | Path) -> WorkflowDefinition: 

644 """Parse a .yaml/.yml/.json file into a WorkflowDefinition.""" 

645 path = Path(filepath) 

646 with open(path, encoding="utf-8") as f: 

647 if path.suffix in (".json",): 

648 data = json.load(f) 

649 else: 

650 data = yaml.safe_load(f) 

651 return WorkflowParser.parse_dict(data) 

652 

653 @staticmethod 

654 def parse_str(text: str) -> WorkflowDefinition: 

655 """Parse a YAML/JSON string into a WorkflowDefinition.""" 

656 try: 

657 data = json.loads(text) 

658 except json.JSONDecodeError: 

659 data = yaml.safe_load(text) 

660 return WorkflowParser.parse_dict(data) 

661 

662 @staticmethod 

663 def parse_dict(data: dict[str, Any]) -> WorkflowDefinition: 

664 """Parse a dict into a WorkflowDefinition.""" 

665 wf = WorkflowDefinition( 

666 name=data.get("name", "unnamed"), 

667 version=data.get("version", "1.0"), 

668 description=data.get("description", ""), 

669 variables=data.get("variables", {}), 

670 agents=data.get("agents", {}), 

671 defaults=data.get("defaults", {}), 

672 metadata=data.get("metadata", {}), 

673 ) 

674 

675 if "steps" in data: 

676 wf.root = WorkflowParser._parse_steps(data["steps"]) 

677 

678 return wf 

679 

680 @staticmethod 

681 def _parse_steps(steps: list[dict[str, Any]]) -> WorkflowStep: 

682 """Parse a list of step dicts into a tree. First step is root.""" 

683 if not steps: 

684 raise ValueError("No steps defined") 

685 

686 parsed = [WorkflowParser._parse_step(s) for s in steps] 

687 

688 # Build parent-child relationships 

689 for i in range(len(parsed) - 1): 

690 if not parsed[i].children: 

691 parsed[i].children = [parsed[i + 1]] 

692 

693 return parsed[0] 

694 

695 @staticmethod 

696 def _parse_step(data: dict[str, Any]) -> WorkflowStep: 

697 """Parse a single step dict.""" 

698 step = WorkflowStep( 

699 id=data.get("id", ""), 

700 type=StepType(data.get("type", "task")), 

701 name=data.get("name", ""), 

702 description=data.get("description", ""), 

703 agent=data.get("agent"), 

704 task=data.get("task"), 

705 timeout=data.get("timeout", 300.0), 

706 depends_on=data.get("depends_on", []), 

707 tags=data.get("tags", []), 

708 metadata=data.get("metadata", {}), 

709 ) 

710 

711 if "condition" in data: 

712 step.condition = data["condition"] 

713 

714 if "branches" in data: 

715 for branch_name, branch_steps in data["branches"].items(): 

716 step.branches[branch_name] = [WorkflowParser._parse_step(s) for s in branch_steps] 

717 

718 if "children" in data: 

719 step.children = [WorkflowParser._parse_step(c) for c in data["children"]] 

720 

721 if "on_error" in data: 

722 step.on_error = ErrorStrategy(data["on_error"]) 

723 if "max_retries" in data: 

724 step.max_retries = data["max_retries"] 

725 if "retry_delay" in data: 

726 step.retry_delay = data["retry_delay"] 

727 if "fallback" in data: 

728 step.fallback_step = WorkflowParser._parse_step(data["fallback"]) 

729 

730 if step.type == StepType.LOOP: 

731 step.max_iterations = data.get("max_iterations", 100) 

732 if "loop_condition" in data: 

733 step.loop_condition = data["loop_condition"] 

734 

735 return step 

736 

737 @staticmethod 

738 def to_yaml(workflow: WorkflowDefinition) -> str: 

739 """Serialize a WorkflowDefinition to YAML string.""" 

740 return yaml.dump(WorkflowParser._to_dict(workflow), default_flow_style=False) 

741 

742 @staticmethod 

743 def to_json(workflow: WorkflowDefinition) -> str: 

744 """Serialize a WorkflowDefinition to JSON string.""" 

745 return json.dumps(WorkflowParser._to_dict(workflow), indent=2) 

746 

747 @staticmethod 

748 def _to_dict(wf: WorkflowDefinition) -> dict[str, Any]: 

749 data = { 

750 "name": wf.name, 

751 "version": wf.version, 

752 "description": wf.description, 

753 "variables": wf.variables, 

754 "agents": wf.agents, 

755 "defaults": wf.defaults, 

756 "metadata": wf.metadata, 

757 } 

758 if wf.root: 

759 data["steps"] = [WorkflowParser._step_to_dict(wf.root)] 

760 return data 

761 

762 @staticmethod 

763 def _step_to_dict(step: WorkflowStep) -> dict[str, Any]: 

764 d = { 

765 "id": step.id, 

766 "type": step.type.value, 

767 "name": step.name, 

768 "description": step.description, 

769 "agent": step.agent, 

770 "task": step.task, 

771 "timeout": step.timeout, 

772 "depends_on": step.depends_on, 

773 "tags": step.tags, 

774 "metadata": step.metadata, 

775 } 

776 if step.condition: