""" Workflows System for PraisonAI Agents. Provides reusable multi-step workflows similar to Windsurf's Workflows, allowing users to define and execute complex task sequences. Features: - YAML/Markdown workflow definitions - Auto-discovery from .praisonai/workflows/ - Step-by-step execution with context passing - Variable substitution - Conditional steps - Lazy loading for performance Storage Structure: .praisonai/workflows/ ├── deploy.md # Deployment workflow ├── test.md # Testing workflow └── review.md # Code review workflow """ import os import re import json import copy import time import logging import collections from praisonaiagents._logging import get_logger from praisonaiagents.utils.safe_path import safe_child from pathlib import Path from typing import Any, Dict, List, Optional, Callable, Tuple, Union from dataclasses import dataclass, field from .workflow_configs import ( WorkflowPlanningConfig, WorkflowMemoryConfig, TaskContextConfig, TaskOutputConfig, TaskExecutionConfig, TaskRoutingConfig, ) from ..task.task import Task from ..llm.model_providers import default_auxiliary_model logger = get_logger(__name__) # Default maximum parallel workers to prevent rate limiting issues DEFAULT_MAX_PARALLEL_WORKERS = 3 # Minimum number of parallel branches before an extra LLM summarisation call is # worth its cost. Below this, the cheaper truncation fallback is used since the # summarisation request consumes roughly as many tokens as it saves. MIN_BRANCHES_FOR_LLM_SUMMARY = 3 import contextvars # Generation of the run a branch belongs to. A fail_fast/timeout Parallel block # abandons still-running branches (threads cannot be force-killed in Python); if # that branch later fails it must NOT scribble "failed" onto a *subsequent* run's # shared status. Each run() stamps its generation here; branches capture it via # contextvars (propagated into the ThreadPoolExecutor by copy_context_to_callable) # and status writes are suppressed once the owning run has moved on. None means # "no active run" (e.g. a bare _execute_single_step_internal call in tests). _run_generation: "contextvars.ContextVar[Optional[int]]" = contextvars.ContextVar( "praisonai_workflow_run_generation", default=None ) from .._run_lock import ensure_run_lock class _WriteTrackingDict(dict): """A ``dict`` that records which keys were assigned after construction. A parallel branch runs against its own deep copy of the shared variables and the branch's writes must be merged back (see ``_merge_branch_variables``). The only reliable way to know *which* keys a branch wrote - not equality (drops an equal-valued write, mis-flags a value whose ``!=`` is not a bool) nor identity (blind to a rebind to an interned immutable like ``x = "SAME"``) - is to record the assignment as it happens. Keys set at construction time seed the branch's starting scope and are deliberately *not* counted as writes. """ __slots__ = ("written_keys",) def __init__(self, *args, **kwargs): super().__init__(*args, **kwargs) # A set, not a copy of the values: only the identity of *which* keys were # written matters, and the current value is read from the dict at merge. self.written_keys = set() def __setitem__(self, key, value): self.written_keys.add(key) super().__setitem__(key, value) def update(self, *args, **kwargs): # dict.update does NOT route through __setitem__, so a StepResult.variables # merge (``all_variables.update(result.variables)``) would otherwise be an # invisible write. Record its keys explicitly. merged = dict(*args, **kwargs) self.written_keys.update(merged) super().update(merged) def setdefault(self, key, default=None): # setdefault also bypasses __setitem__ and creates the key when absent. if key not in self: self.written_keys.add(key) return super().setdefault(key, default) def _isolate_value(value: Any, memo: dict) -> Any: """Deep-copy ``value`` but degrade to sharing by reference where it can't. This is ``copy.deepcopy`` with two differences that matter for a parallel scope: * **Shared memo.** All values in one scope are copied with the *same* ``memo`` so cross-variable aliasing survives: if ``variables['records']`` and ``variables['state']['records']`` are the same list, the branch still sees one shared list, exactly as the old whole-scope ``copy.deepcopy`` did. A write through one name is visible through the other within the branch. * **Granular fallback.** A value that cannot be deep-copied (thread lock, DB/HTTP/LLM client, open file, a live Agent, generator, ...) would abort the whole run with a pickling ``TypeError`` even when no branch touches it. Only the *exact* un-copyable node is shared by reference; the containers around it are still copied, so an ordinary mutable sibling (``{'client': c, 'rows': []}`` -> ``rows``) stays isolated and an in-place mutation cannot leak to a sibling branch or loop iteration. """ # Already isolated under this scope (cross-variable alias): reuse the copy. if id(value) in memo: return memo[id(value)] pre_existing = set(memo) try: return copy.deepcopy(value, memo) except Exception: pass # ``copy.deepcopy`` failed somewhere inside ``value``. A failed deepcopy can # leave *unfinished* copies in ``memo`` (e.g. an empty dict for a container it # had started but not populated before the un-copyable leaf raised). Reusing # those would silently drop nested seed data, so discard everything this # attempt added and rebuild the known container types node-by-node. The # un-copyable leaves are shared by reference; everything around them is still # isolated, and container subclasses (namedtuple, defaultdict, OrderedDict, # subclassed list/set/tuple) keep their type and behaviour. for key in set(memo) - pre_existing: del memo[key] return _rebuild_isolated(value, memo) def _rebuild_isolated(value: Any, memo: dict) -> Any: """Node-by-node fallback copy that shares only the un-copyable leaves. Used by :func:`_isolate_value` after a whole-value ``copy.deepcopy`` fails. Rebuilds ``dict``/``list``/``set``/``tuple`` (and their subclasses, so a ``defaultdict`` keeps its factory and a ``namedtuple`` keeps its fields), recursing through :func:`_isolate_value` so each nested child first gets a clean ``deepcopy`` attempt and only genuinely un-copyable nodes are shared. """ if id(value) in memo: return memo[id(value)] if isinstance(value, dict): try: copied = value.__class__() # preserves defaultdict/OrderedDict/... except Exception: copied = {} if isinstance(value, collections.defaultdict): copied.default_factory = value.default_factory memo[id(value)] = copied for k, v in value.items(): copied[_isolate_value(k, memo)] = _isolate_value(v, memo) return copied if isinstance(value, list): try: copied = value.__class__() # preserves a list subclass where possible except Exception: copied = [] # subclass constructor requires args: fall back to list memo[id(value)] = copied copied.extend(_isolate_value(v, memo) for v in value) return copied if isinstance(value, set): try: copied = value.__class__() # preserves a set subclass where possible except Exception: copied = set() # subclass constructor requires args: fall back to set memo[id(value)] = copied for v in value: copied.add(_isolate_value(v, memo)) return copied if isinstance(value, tuple): items = [_isolate_value(v, memo) for v in value] if hasattr(value, "_fields"): # namedtuple keeps its type and fields try: return value.__class__(*items) except Exception: return tuple(items) return tuple(items) # Unknown/atomic object that could not be deep-copied: share by reference as # a last resort (read-only intent). return value def _isolate_scope(variables: dict) -> dict: """Give a parallel branch/iteration its own copy of the variable scope. A branch needs its own container so concurrent writes are not a data race, and mutable data (lists/dicts) is deep-copied so a sibling cannot observe another branch's in-place mutation. Values are copied with one shared ``memo`` so cross-variable aliasing is preserved, and any value that cannot be deep-copied degrades to a by-reference share of only the un-copyable node rather than crashing the run. See ``_isolate_value``. """ memo: dict = {} return {k: _isolate_value(v, memo) for k, v in variables.items()} class WorkflowStepError(Exception): """Exception raised when workflow step execution fails.""" def __init__(self, message: str, cause: Exception = None, errors: List = None): super().__init__(message) self.cause = cause self.errors = errors or [] def _parse_json_output(output: Any, step_name: str = "step") -> Any: """ Parse JSON from LLM output if it's a string. Handles: - Direct JSON strings: '{"key": "value"}' - Markdown code blocks: ```json\n{"key": "value"}\n``` - LLM echoing schema: {'type': 'array', 'items': [...]} -> extract items Returns: Parsed dict/list if successful, original output otherwise """ if not isinstance(output, str) or not output: # Handle LLM echoing schema structure even for non-string outputs if isinstance(output, dict): return _extract_from_schema_echo(output) return output # Try direct JSON parse first try: parsed = json.loads(output) # Check if LLM echoed the schema structure if isinstance(parsed, dict): parsed = _extract_from_schema_echo(parsed) return parsed except json.JSONDecodeError: pass # Try extracting from markdown code block json_match = re.search(r'```(?:json)?\s*\n?([\s\S]*?)\n?```', output) if json_match: try: parsed = json.loads(json_match.group(1).strip()) if isinstance(parsed, dict): parsed = _extract_from_schema_echo(parsed) return parsed except json.JSONDecodeError: pass # Try finding JSON object/array in text # Look for {...} or [...] for pattern in [r'(\{[^}]+\})', r'(\[[^\]]+\])']: match = re.search(pattern, output) if match: try: parsed = json.loads(match.group(1)) if isinstance(parsed, dict): parsed = _extract_from_schema_echo(parsed) return parsed except json.JSONDecodeError: continue # Return original if can't parse logger.debug(f"Could not parse JSON from step '{step_name}' output") return output def _substitute_action_variables(action: str, all_variables: dict, previous_output: Any, task_input: str) -> str: """ Substitute templating placeholders in a prompt/action string. Performs, in order: - ``{{key}}`` substitution for each key/value in ``all_variables`` - ``{{previous_output}}`` in-place replacement when present, otherwise appending a "Context from previous step" section (only if there is a previous output) - ``{{input}}`` substitution Returns the rendered action string. """ for key, value in all_variables.items(): action = action.replace(f"{{{{{key}}}}}", str(value)) if previous_output: if "{{previous_output}}" in action: action = action.replace("{{previous_output}}", str(previous_output)) else: # Auto-append context if not explicitly referenced action = f"{action}\n\nContext from previous step:\n{previous_output}" return action.replace("{{input}}", task_input) def _extract_from_schema_echo(data: dict) -> Any: """ Extract actual data when LLM echoes the JSON schema structure. Common patterns: - {'type': 'array', 'items': [...]} -> return [...] - {'type': 'object', 'properties': {...}, 'data': {...}} -> return data - {'items': [...]} -> return [...] (native structured output wrapper) - {'keywords': [...]} -> return as-is (normal output) Args: data: Dict that might contain echoed schema Returns: Extracted data or original dict """ # Check if this looks like a schema echo with 'type' and actual data if 'type' in data: data_type = data.get('type') # Array schema echo: {'type': 'array', 'items': [...]} if data_type == 'array' and 'items' in data: items = data['items'] # If items is a list, that's our actual data if isinstance(items, list): logger.debug("Extracted items from schema-echoed array output") return items # Object schema echo: {'type': 'object', 'properties': {...}, 'data': {...}} if data_type == 'object': # Check for actual data field if 'data' in data and isinstance(data['data'], dict): logger.debug("Extracted data from schema-echoed object output") return data['data'] # Handle native structured output wrapper: {'items': [...]} # This is created by _build_response_format when wrapping arrays if 'items' in data and len(data) == 1 and isinstance(data['items'], list): logger.debug("Extracted items from native structured output array wrapper") return data['items'] # Not a schema echo, return as-is return data @dataclass class WorkflowContext: """Context passed to step handlers. Contains all information about the current workflow state.""" input: str = "" # Original workflow input previous_result: Optional[str] = None # Output from previous step current_step: str = "" # Current step name variables: Dict[str, Any] = field(default_factory=dict) # All workflow variables @dataclass class StepResult: """Result returned from step handlers.""" output: str = "" # Step output content stop_workflow: bool = False # If True, stop the entire workflow early variables: Dict[str, Any] = field(default_factory=dict) # Variables to add/update skipped: bool = False # Completion observer outcome when no handler ran # Aliases for backward compatibility StepInput = WorkflowContext StepOutput = StepResult # ============================================================================= # Workflow Pattern Helpers # ============================================================================= @dataclass class Route: """ Decision-based branching. Routes to different steps based on previous output. Usage: workflow = Workflow(steps=[ decision_maker, route({ "approve": [approve_step], "reject": [reject_step], "default": [fallback_step] }) ]) """ routes: Dict[str, List] = field(default_factory=dict) default: Optional[List] = None def __init__(self, routes: Dict[str, List], default: Optional[List] = None): self.routes = routes self.default = default or routes.get("default", []) @dataclass class Parallel: """ Execute multiple steps concurrently and combine results. Usage: workflow = Workflow(steps=[ parallel([agent1, agent2, agent3], max_workers=5, on_failure="partial_ok"), aggregator ]) Failure strategies: - "partial_ok": Continue with partial results if some branches fail (default) - "fail_fast": Cancel remaining branches and fail immediately on first error - "fail_all": Wait for all branches to complete, then fail if any failed timeout: optional per-Parallel wall-clock bound (seconds). Must be a positive number, or None for no bound (a non-positive value raises ValueError rather than silently disabling the limit). The clock starts when the block begins, including any context-preparation work. If branches are still running after it elapses the block raises WorkflowStepError instead of hanging forever. Threads running user code cannot be force-killed in Python, so a timed-out/cancelled branch may keep running in the background, but the caller is no longer blocked on it and its late failure cannot corrupt a later run's status. Context strategies (how the upstream step's output reaches each branch): - "full": Pass the exact upstream output to every branch (default, no data loss) - "truncate": Excerpt head+tail to save tokens (lossy; opt-in) - "summarize": LLM-summarise for large fan-outs, else truncate (lossy; opt-in) """ steps: List = field(default_factory=list) max_workers: Optional[int] = None # None = use system default on_failure: str = "partial_ok" # "partial_ok" | "fail_fast" | "fail_all" timeout: Optional[float] = None # None = no wall-clock bound context: str = "full" # "full" | "truncate" | "summarize" def __init__(self, steps: List, max_workers: Optional[int] = None, on_failure: str = "partial_ok", timeout: Optional[float] = None, context: str = "full"): valid_on_failure = {"partial_ok", "fail_fast", "fail_all"} if on_failure not in valid_on_failure: raise ValueError( f"Invalid on_failure='{on_failure}'. Must be one of {valid_on_failure}. " "See Parallel docstring for semantics." ) # Reject a non-positive timeout: `timeout=0`/negatives are truthiness-false # and would silently disable the bound, letting a hung branch wait forever # even though the caller explicitly asked for a limit. `None` is the only # way to mean "no bound". if timeout is not None and timeout <= 0: raise ValueError( f"Invalid timeout={timeout!r}. Must be a positive number of seconds, " "or None for no wall-clock bound." ) valid_context = {"full", "truncate", "summarize"} if context not in valid_context: raise ValueError( f"Invalid context='{context}'. Must be one of {valid_context}. " "See Parallel docstring for semantics." ) self.steps = steps self.max_workers = max_workers self.on_failure = on_failure self.timeout = timeout self.context = context @dataclass class Loop: """ Iterate over a list or CSV file, executing step(s) for each item. Usage: # Loop over list variable (sequential) - single step workflow = Workflow( steps=[loop(processor, over="items")], variables={"items": ["a", "b", "c"]} ) # Loop over CSV file workflow = Workflow(steps=[ loop(processor, from_csv="data.csv") ]) # Parallel loop over items (concurrent execution) workflow = Workflow( steps=[loop(processor, over="items", parallel=True)], variables={"items": ["a", "b", "c"]} ) # Parallel with limited workers workflow = Workflow( steps=[loop(processor, over="items", parallel=True, max_workers=4)], variables={"items": ["a", "b", "c"]} ) # Multi-step loop (NEW) - execute multiple steps per iteration workflow = Workflow( steps=[loop( steps=[researcher, writer, publisher], # Multiple steps over="topics", parallel=True, max_workers=4 )], variables={"topics": [...]} ) """ step: Any = None # Single step (backward compat) steps: Optional[List[Any]] = None # Multiple steps (NEW) over: Optional[str] = None # Variable name containing list from_csv: Optional[str] = None # CSV file path from_file: Optional[str] = None # Text file path (one item per line) var_name: str = "item" # Variable name for current item parallel: bool = False # Execute iterations in parallel max_workers: Optional[int] = None # Max parallel workers (None = unlimited) output_variable: Optional[str] = None # Store loop results in this variable name context: str = "full" # "full" | "truncate" — how previous_output reaches parallel iterations def __init__( self, step: Any = None, steps: Optional[List[Any]] = None, over: Optional[str] = None, from_csv: Optional[str] = None, from_file: Optional[str] = None, var_name: str = "item", parallel: bool = False, max_workers: Optional[int] = None, output_variable: Optional[str] = None, context: str = "full" ): # Validation: cannot have both step and steps if step is not None and steps is not None: raise ValueError("Cannot specify both 'step' and 'steps'") # Validation: must have at least one if step is None and steps is None: raise ValueError("Loop requires 'step' or 'steps'") # Validation: steps cannot be empty if steps is not None and len(steps) == 0: raise ValueError("Loop 'steps' cannot be empty") valid_context = {"full", "truncate"} if context not in valid_context: raise ValueError( f"Invalid context='{context}'. Must be one of {valid_context}." ) self.step = step self.steps = steps self.over = over self.from_csv = from_csv self.from_file = from_file self.var_name = var_name self.parallel = parallel self.max_workers = max_workers self.output_variable = output_variable self.context = context class Discussion: """N agents take turns on the same thread until a criterion is met. PraisonAI had no N-agent conversation loop at all -- no round-robin, no speaker selection -- so having three agents debate for a few turns meant writing the loop by hand. ``repeat([a, b])`` was the obvious workaround and silently misbehaved until it was fixed, which is what made this worth building rather than documenting around. Discussion([critic, author], rounds=3, until=lambda ctx: "AGREED" in (ctx.previous_result or "")) Each turn receives the previous turn's output via ``ctx.previous_result``, so this is a conversation rather than N independent answers. The full running transcript is exposed as the ``discussion_transcript`` variable -- a speaker action sees the whole thread only when it references ``{{discussion_transcript}}``. ``rounds`` counts full passes over the speakers; it defaults to a bounded 3 and must be >= 1 -- an unbounded debate is a bill, not a feature. """ def __init__( self, agents=None, rounds: int = 3, until=None, select=None, name: str = "discussion", ): speakers = list(agents or []) if not speakers: raise ValueError( "Discussion needs at least one speaker; an empty discussion would " "run zero turns and report success." ) if rounds < 1: raise ValueError("Discussion rounds must be >= 1.") self.agents = speakers self.rounds = rounds #: Stop early when this returns True, checked after every turn. self.until = until #: Choose the next speaker: (turn_index, context) -> agent. Defaults to #: round-robin, which is the behaviour people expect from "discussion". self.select = select self.name = name @dataclass class Repeat: """ Repeat a step until a condition is met (evaluator-optimizer pattern). Usage: workflow = Workflow(steps=[ repeat( generator, until=lambda ctx: "done" in ctx.previous_result.lower(), max_iterations=5 ) ]) """ step: Any = None until: Optional[Callable[[WorkflowContext], bool]] = None max_iterations: int = 10 def __init__( self, step: Any, until: Optional[Callable[[WorkflowContext], bool]] = None, max_iterations: int = 10 ): self.step = step self.until = until self.max_iterations = max_iterations # Convenience functions for cleaner API def route(routes: Dict[str, List], default: Optional[List] = None) -> Route: """Create a routing decision point.""" return Route(routes=routes, default=default) def parallel( steps: List, max_workers: Optional[int] = None, on_failure: str = "partial_ok", timeout: Optional[float] = None, context: str = "full", ) -> Parallel: """Execute steps in parallel. Args: steps: Steps to execute concurrently. max_workers: Optional cap on ThreadPoolExecutor workers. When unset, defaults to min(DEFAULT_MAX_PARALLEL_WORKERS, len(steps)). on_failure: Failure strategy — "partial_ok" (default), "fail_fast", or "fail_all". timeout: Optional per-Parallel wall-clock bound (seconds); must be positive (or None for no bound). Raises instead of hanging if branches are still running when it elapses. context: How the upstream output reaches each branch — "full" (default, exact pass-through, no data loss), "truncate", or "summarize". """ return Parallel(steps=steps, max_workers=max_workers, on_failure=on_failure, timeout=timeout, context=context) def loop(step: Any = None, steps: Optional[List[Any]] = None, over: Optional[str] = None, from_csv: Optional[str] = None, from_file: Optional[str] = None, var_name: str = "item", parallel: bool = False, max_workers: Optional[int] = None, output_variable: Optional[str] = None, context: str = "full") -> Loop: """Loop over items executing step(s) for each. Args: step: Single step (agent/function) to execute for each item (backward compat) steps: Multiple steps to execute sequentially for each item (NEW) over: Variable name containing the list to iterate over from_csv: Path to CSV file to read items from from_file: Path to text file with one item per line var_name: Variable name for current item (default: "item") parallel: If True, execute iterations in parallel (default: False) max_workers: Max parallel workers when parallel=True (default: None = unlimited) output_variable: Variable name to store all loop outputs (default: None = "loop_outputs") context: How previous_output reaches parallel iterations — "full" (default, exact pass-through, no data loss) or "truncate" (lossy; opt-in) Returns: Loop object configured for iteration """ return Loop(step=step, steps=steps, over=over, from_csv=from_csv, from_file=from_file, var_name=var_name, parallel=parallel, max_workers=max_workers, output_variable=output_variable, context=context) def repeat(step: Any, until: Optional[Callable[[WorkflowContext], bool]] = None, max_iterations: int = 10) -> Repeat: """Repeat step until condition is met.""" return Repeat(step=step, until=until, max_iterations=max_iterations) @dataclass class Include: """ Include another recipe or workflow as a workflow step. Enables modular composition - recipes can include other recipes or workflows. Usage: workflow = Workflow(steps=[ content_writer, include("wordpress-publisher"), # Include another recipe include(workflow=my_other_workflow), # Include another workflow ]) YAML syntax: steps: - agent: content_writer - include: wordpress-publisher # Or with configuration: - include: recipe: wordpress-publisher input: "{{previous_output}}" """ recipe: Optional[str] = None # Recipe name or path workflow: Optional[Any] = None # Workflow instance for composition input: Optional[str] = None # Input override (default: previous_output) def __init__(self, recipe: Optional[str] = None, workflow: Optional[Any] = None, input: Optional[str] = None): if recipe is None and workflow is None: raise ValueError("Either 'recipe' or 'workflow' must be provided") self.recipe = recipe self.workflow = workflow self.input = input def include(recipe: Optional[str] = None, workflow: Optional[Any] = None, input: Optional[str] = None) -> Include: """Include another recipe or workflow as a workflow step. Args: recipe: Recipe name or path to include workflow: Workflow instance to include input: Input override (default: previous_output) Returns: Include object for workflow composition """ return Include(recipe=recipe, workflow=workflow, input=input) # Maximum nesting depth for patterns (prevents stack overflow) MAX_NESTING_DEPTH = 5 # --------------------------------------------------------------------------- # Replay step events (step_start / step_end). Every helper is a no-op returning # None when the context emitter is disabled, so an untraced run pays one # attribute check per step. # --------------------------------------------------------------------------- # The innermost pattern span (Route/If) so it can record the branch it chose. _replay_step_span: "contextvars.ContextVar[Optional[dict]]" = contextvars.ContextVar( "praisonai_replay_step_span", default=None ) def _replay_step_begin(flow, step, name=None, pattern=None): """Emit ``step_start`` for a definition object; returns a span or None.""" from ..trace.context_events import get_context_emitter emitter = get_context_emitter() if not emitter._enabled: return None node_id = (getattr(flow, "_replay_node_ids", None) or {}).get(id(step)) if name is None: if node_id is not None: name = node_id[len("step:"):] if node_id.startswith("step:") else node_id else: from .diagram import _label name = pattern or _label(step) if node_id is None: node_id = f"step:{name}" agent = getattr(step, "agent", None) if not hasattr(step, "chat") else step agent = getattr(agent, "name", None) if agent is not None else None emitter.step_start(name, pattern=pattern, node_id=node_id, agent=agent) return {"emitter": emitter, "name": name, "node_id": node_id, "agent": agent, "started": time.perf_counter()} def _replay_step_end(span, status, output=None): """Emit ``step_end`` for a span from :func:`_replay_step_begin`.""" if span is None: return span["emitter"].step_end( span["name"], status, duration_ms=(time.perf_counter() - span["started"]) * 1000, output=output, route=span.get("route"), node_id=span["node_id"], agent=span["agent"], ) def _replay_note_route(label): """Record which branch the enclosing Route/If chose (tracing on only).""" span = _replay_step_span.get() if span is not None: span["route"] = label def _replay_pattern(pattern): """Wrap a pattern executor (``_execute_route`` ...) in step_start/step_end. Top-level and nested patterns both reach the executor, so wrapping it here covers every nesting level once. """ def decorate(execute): def wrapper(self, step, *args, **kwargs): name = getattr(step, "name", None) if pattern == "discussion" else None span = _replay_step_begin(self, step, name, pattern) if span is None: return execute(self, step, *args, **kwargs) token = _replay_step_span.set(span) try: result = execute(self, step, *args, **kwargs) except BaseException: _replay_step_end(span, "failed") raise finally: _replay_step_span.reset(token) is_dict = isinstance(result, dict) _replay_step_end( span, "failed" if is_dict and result.get("stop") else "completed", result.get("output") if is_dict else None, ) return result wrapper.__name__ = execute.__name__ wrapper.__qualname__ = execute.__qualname__ wrapper.__doc__ = execute.__doc__ wrapper.__wrapped__ = execute return wrapper return decorate @dataclass class If: """ Conditional branching pattern for workflows. Evaluates a condition and executes either then_steps or else_steps. Usage: workflow = Workflow(steps=[ if_( condition="{{score}} > 80", then_steps=[approve_step], else_steps=[reject_step] ) ]) YAML syntax: steps: - if: condition: "{{score}} > 80" then: - agent: approver else: - agent: rejector Supported condition formats: - Numeric comparison: "{{var}} > 80", "{{var}} >= 50", "{{var}} < 10" - String equality: "{{var}} == approved", "{{var}} != rejected" - Contains check: "error in {{message}}", "{{status}} contains success" - Boolean: "{{flag}}" (truthy check) """ condition: str = "" # Condition expression with {{var}} placeholders then_steps: List[Any] = field(default_factory=list) # Steps if condition is true else_steps: Optional[List[Any]] = None # Steps if condition is false (optional) def __init__( self, condition: str, then_steps: List[Any], else_steps: Optional[List[Any]] = None ): self.condition = condition self.then_steps = then_steps self.else_steps = else_steps or [] def when( condition: str, then_steps: List[Any], else_steps: Optional[List[Any]] = None ) -> If: """ Create a conditional branching step. This is the preferred function for conditional branching (cleaner name than if_). Args: condition: Condition expression with {{var}} placeholders then_steps: Steps to execute if condition is true else_steps: Steps to execute if condition is false (optional) Returns: If object for conditional execution Example: when( condition="{{score}} > 80", then_steps=[approve_agent], else_steps=[reject_agent] ) """ return If(condition=condition, then_steps=then_steps, else_steps=else_steps) def if_( condition: str, then_steps: List[Any], else_steps: Optional[List[Any]] = None ) -> If: """ Create a conditional branching step. Alias for :func:`when`. Both functions work identically. Args: condition: Condition expression with {{var}} placeholders then_steps: Steps to execute if condition is true else_steps: Steps to execute if condition is false (optional) Returns: If object for conditional execution """ return If(condition=condition, then_steps=then_steps, else_steps=else_steps) @dataclass class AgentFlow: """ A complete workflow with multiple steps (deterministic pipeline). AgentFlow-centric API with consolidated feature parameters. All params follow precedence: Instance > Config > String > Bool > Default Usage: from praisonaiagents import AgentFlow, Agent # Simple flow flow = AgentFlow( steps=[Agent(instructions="Write content"), Agent(instructions="Edit content")], output="verbose", ) result = flow.run("Write about AI") # With consolidated configs from praisonaiagents.workflows import WorkflowOutputConfig, WorkflowPlanningConfig flow = AgentFlow( steps=[...], output=WorkflowOutputConfig(verbose=True, stream=True), planning=WorkflowPlanningConfig(llm="gpt-4o", reasoning=True), hooks=WorkflowHooksConfig(on_step_complete=my_callback), ) Note: The class was renamed from `Workflow` to `AgentFlow` in v1.0. `Workflow` and `Pipeline` remain as silent aliases for backward compatibility. """ name: str = "AgentFlow" description: str = "" steps: List = field(default_factory=list) # Can be Task, Agent, or function variables: Dict[str, Any] = field(default_factory=dict) #: Cache step results by their inputs. True for a process-scoped LRU, or #: any object with get()/set(). OPT-IN: an agent step is not always #: deterministic, and silently serving a cached answer for a call the user #: expected to happen again is worse than paying for it. cache: Any = None file_path: Optional[str] = None # Default configuration for all steps default_agent_config: Optional[Dict[str, Any]] = None # Default agent for all steps # Default model for steps that did not name one. A ready-made Agent step # constructed with its own llm=/model=/auth= keeps it -- same rule as # AgentTeam. manager_llm below is separate: it is the hierarchical # manager's model and never touches the steps. # model= is the canonical name; llm= is the deprecated alias kept for # compatibility (renamed from default_llm). Passing both raises TypeError # -- see resolve_model_alias. model: Optional[str] = None llm: Optional[str] = None # Process type: "sequential" (default) or "hierarchical" (manager-based validation) process: str = "sequential" # "sequential", "hierarchical" manager_llm: Optional[str] = None # LLM for manager agent (hierarchical mode) # ============================================================ # CONSOLIDATED FEATURE PARAMS (workflow-centric API) # Precedence: Instance > Config > Array > Dict > String > Bool > Default # Workflow is "Agents in an organised way" - supports same params as Agents # ============================================================ output: Optional[Any] = None # Union[str, WorkflowOutputConfig] - verbose/stream planning: Optional[Any] = False # Union[bool, str, WorkflowPlanningConfig] - planning mode memory: Optional[Any] = None # Union[bool, WorkflowMemoryConfig, instance] - memory hooks: Optional[Any] = None # WorkflowHooksConfig - lifecycle callbacks context: Optional[Any] = True # Union[bool, ManagerConfig, ContextManager] - context management (enabled by default for workflows) # NEW: Agent-like consolidated params for feature parity autonomy: Optional[Any] = None # Union[bool, AutonomyConfig] - agent autonomy knowledge: Optional[Any] = None # Union[bool, List[str], KnowledgeConfig] - RAG guardrails: Optional[Any] = None # Union[bool, Callable, GuardrailConfig] - validation web: Optional[Any] = None # Union[bool, WebConfig] - web search/fetch reflection: Optional[Any] = None # Union[bool, ReflectionConfig] - self-reflection execution: Optional[Any] = None # Union[str, ExecutionConfig] - execution control caching: Optional[Any] = None # Union[bool, CachingConfig] - caching learn: Optional[Any] = None # Union[bool, LearnConfig] - continuous learning # ============================================================ # REMOTE EXECUTION # ============================================================ # Union[str, ComputeProviderProtocol] - where each step's shell/file tools # run: "docker", "e2b", "modal", "daytona", "flyio", "tenki", "local", # "subprocess", "sandlock", "ssh", "novita". # Every step shares ONE sandbox and one working directory, so a file written # by one step is visible to later steps. Orchestration -- routing, parallel, # repeat -- stays local, and so does the thinking. Pass a configured # provider instance instead of a name to customise resources. tools_run_on: Optional[Union['ToolPlace', str, Any]] = None # Accepted only to fail with a useful message: on an Agent, run_on= moves # the whole agent to a managed runtime, so honouring it here as a synonym # for tools_run_on= would make one word mean two things. run_on: Optional[Any] = None # ============================================================ # ROBUSTNESS PARAMS (debugging & audit trail) # ============================================================ history: bool = False # Enable execution history tracking for debugging # Status tracking status: str = "not_started" # not_started, running, completed, failed step_statuses: Dict[str, str] = field(default_factory=dict) # {step_name: status} #: Optional type for ``variables``: a Pydantic model or a dataclass. When #: set, the initial variables are validated and ``flow.state`` gives typed #: access, so a misspelled variable is an error instead of a value written #: once and never read. Without it nothing changes -- ``variables`` stays #: the untyped dict it has always been. Declared last so it never shifts #: the positional binding of existing fields like ``file_path``. state_model: Optional[type] = None # Private resolved fields (set in __post_init__) _verbose: bool = field(default=False, repr=False) _stream: bool = field(default=True, repr=False) _output_config: Optional[Any] = field(default=None, repr=False) # Full OutputConfig for propagation _planning_enabled: bool = field(default=False, repr=False) _planning_llm: Optional[str] = field(default=None, repr=False) _reasoning: bool = field(default=False, repr=False) _memory_config: Optional[Any] = field(default=None, repr=False) _hooks_config: Optional[Any] = field(default=None, repr=False) _context_manager: Optional[Any] = field(default=None, repr=False) _context_manager_initialized: bool = field(default=False, repr=False) _session_dedup_cache: Optional[Any] = field(default=None, repr=False) # Shared session deduplication cache # NEW: Resolved configs for agent-like params (for propagation to steps/agents) _autonomy_config: Optional[Any] = field(default=None, repr=False) _knowledge_config: Optional[Any] = field(default=None, repr=False) _guardrails_config: Optional[Any] = field(default=None, repr=False) _web_config: Optional[Any] = field(default=None, repr=False) _reflection_config: Optional[Any] = field(default=None, repr=False) # Execution history for debugging (only populated when history=True) _execution_history: List[Dict[str, Any]] = field(default_factory=list, repr=False) # Gap 3c: Cross-step handoff cycle detection _handoff_chain: List[str] = field(default_factory=list, repr=False) # Guards per-run mutable state (status/step_statuses/_handoff_chain) so a # shared Workflow instance cannot be corrupted by concurrent run() calls. _run_lock: Optional[Any] = field(default=None, repr=False, compare=False) # Monotonic id of the currently-executing run. Bumped at each run() so an # abandoned fail_fast/timeout Parallel branch can tell it outlived its run # and must not mutate a later run's shared status (see _begin_run_generation # and _is_current_run_generation). _current_run_generation: int = field(default=0, repr=False, compare=False) # Reset token for the _run_generation contextvar captured at run() start and # consumed in run()'s finally to restore the caller's generation (nested runs). _run_generation_token: Any = field(default=None, repr=False, compare=False) def _begin_run_generation(self) -> Any: """Stamp a fresh generation for the run starting now and publish it to the current context so branches submitted by this run capture it. Returns the contextvar reset token for the *previous* generation. The caller MUST pass it to :meth:`_end_run_generation` in a ``finally`` so a nested ``Workflow.run()`` (e.g. an ``Include`` step launching another flow) restores the outer run's generation on the way out instead of leaving its own value stamped on the outer thread's context -- otherwise a later task in the outer flow would be judged against the wrong generation and have its failure silently suppressed. """ self._current_run_generation += 1 return _run_generation.set(self._current_run_generation) def _end_run_generation(self, token: Any) -> None: """Restore the generation the context carried before :meth:`_begin_run_generation`. A no-op when ``token`` is None so callers can guard unconditionally. """ if token is not None: _run_generation.reset(token) def _is_current_run_generation(self) -> bool: """True if the caller still belongs to the instance's live run. A branch abandoned by a previous fail_fast/timeout Parallel block carries an older generation (captured via the _run_generation contextvar), so this returns False for it and its late status writes are suppressed. A bare internal call outside any run (generation None) is treated as current to preserve existing single-step behaviour. """ gen = _run_generation.get() return gen is None or gen == self._current_run_generation @property def _execution_lock(self): """Lazily-created re-entrancy lock for run()/astart() (zero overhead until used). Creation is serialized through a module-level guard so two threads that first reach run()/astart() concurrently observe the *same* lock object (double-checked locking) rather than each minting and acquiring its own. """ return ensure_run_lock(self) def __post_init__(self): """Resolve consolidated params to internal values.""" from ..utils.model_alias import resolve_model_name from .step_cache import resolve_step_cache # Resolved once, not per step: `cache=True` must mean ONE cache shared # across the run, not a fresh empty one for every step (which would # never hit and would look like the feature simply not working). self._step_cache = resolve_step_cache(self.cache) # Validate the declared state up front. A misspelled variable found when # the flow is BUILT costs nothing; found mid-run it has already burned # the steps before it, and may never be found at all. if self.state_model is not None: from .state import validate_variables as _validate_state _validate_state(self.state_model, self.variables) # One rule for the alias pair, shared with Agent and AgentTeam. Must # UNWRAP an LLMConfig to its model string: this value seeds the agents # AgentFlow builds, and an agent holding an LLMConfig fails inside chat(). self.llm = resolve_model_name(self.llm, self.model, type(self).__name__) self.model = self.llm from ..agent.placement import resolve_placement resolve_placement( "AgentFlow", run_on=self.run_on, tools_run_on=self.tools_run_on, supports_run_on=False, ) from .workflow_configs import ( resolve_output_config, resolve_planning_config, resolve_memory_config, resolve_hooks_config, ) # Resolve output param - now uses OUTPUT_PRESETS (same as Agent) output_cfg = resolve_output_config(self.output) self._output_config = output_cfg # Store full config for propagation self._verbose = getattr(output_cfg, 'verbose', False) if output_cfg else False self._stream = getattr(output_cfg, 'stream', True) if output_cfg else True # Enable status/trace output if configured (same as Agent) if output_cfg: status_trace = getattr(output_cfg, 'status_trace', False) actions_trace = getattr(output_cfg, 'actions_trace', False) json_output = getattr(output_cfg, 'json_output', False) simple_output = getattr(output_cfg, 'simple_output', False) metrics = getattr(output_cfg, 'metrics', False) if status_trace: try: from ..output.trace import enable_trace_output, is_trace_output_enabled if not is_trace_output_enabled(): enable_trace_output(use_color=True, show_timestamps=True) except ImportError: pass elif actions_trace: try: from ..output.status import enable_status_output, is_status_output_enabled if not is_status_output_enabled(): output_format = "jsonl" if json_output else "text" enable_status_output( redact=True, use_color=True, format=output_format, show_timestamps=not simple_output, show_metrics=metrics ) except ImportError: pass # Resolve planning param planning_cfg = resolve_planning_config(self.planning) if planning_cfg: self._planning_enabled = planning_cfg.enabled self._planning_llm = planning_cfg.llm self._reasoning = planning_cfg.reasoning else: self._planning_enabled = False self._planning_llm = None self._reasoning = False # Resolve memory param self._memory_config = resolve_memory_config(self.memory) # Resolve hooks param self._hooks_config = resolve_hooks_config(self.hooks) # Resolve NEW agent-like params using canonical resolver # These are stored for propagation to steps/agents from ..config.param_resolver import resolve, ArrayMode from ..config.presets import ( WEB_PRESETS, REFLECTION_PRESETS, GUARDRAIL_PRESETS, AUTONOMY_PRESETS, KNOWLEDGE_PRESETS, ) try: from ..config.feature_configs import ( AutonomyConfig, KnowledgeConfig, GuardrailConfig, WebConfig, ReflectionConfig, ) except ImportError: AutonomyConfig = KnowledgeConfig = GuardrailConfig = None WebConfig = ReflectionConfig = None # Resolve autonomy if AutonomyConfig and self.autonomy is not None: self._autonomy_config = resolve( value=self.autonomy, param_name="autonomy", config_class=AutonomyConfig, presets=AUTONOMY_PRESETS, default=None, ) # Resolve knowledge if KnowledgeConfig and self.knowledge is not None: self._knowledge_config = resolve( value=self.knowledge, param_name="knowledge", config_class=KnowledgeConfig, presets=KNOWLEDGE_PRESETS, array_mode=ArrayMode.SOURCES, default=None, ) # Resolve guardrails if GuardrailConfig and self.guardrails is not None: self._guardrails_config = resolve( value=self.guardrails, param_name="guardrails", config_class=GuardrailConfig, presets=GUARDRAIL_PRESETS, default=None, ) # Resolve web if WebConfig and self.web is not None: self._web_config = resolve( value=self.web, param_name="web", config_class=WebConfig, presets=WEB_PRESETS, default=None, ) # Resolve reflection if ReflectionConfig and self.reflection is not None: self._reflection_config = resolve( value=self.reflection, param_name="reflection", config_class=ReflectionConfig, presets=REFLECTION_PRESETS, default=None, ) # Initialize session deduplication cache for cross-agent deduplication if self.context: try: from ..context.manager import SessionDeduplicationCache self._session_dedup_cache = SessionDeduplicationCache() except ImportError: pass def get_history(self) -> List[Dict[str, Any]]: """Return execution trace for debugging. Returns a list of step execution records, each containing: - step: Step name - timestamp: ISO format timestamp - success: Whether the step succeeded - output: Truncated output (first 500 chars) - error: Error message if failed Only populated when history=True is set on the workflow. """ return self._execution_history def to_dict(self) -> Dict[str, Any]: return { "name": self.name, "description": self.description, "steps": [s.to_dict() if hasattr(s, 'to_dict') else str(s) for s in self.steps], "variables": self.variables, "file_path": self.file_path, "default_agent_config": self.default_agent_config, "default_llm": self.llm, "process": self.process, "manager_llm": self.manager_llm, "output": {"verbose": self._verbose, "stream": self._stream}, "planning": {"enabled": self._planning_enabled, "llm": self._planning_llm, "reasoning": self._reasoning}, "memory": self._memory_config.to_dict() if hasattr(self._memory_config, 'to_dict') else bool(self._memory_config), "hooks": self._hooks_config.to_dict() if hasattr(self._hooks_config, 'to_dict') else None, } # Property accessors for backward compatibility during transition @property def verbose(self) -> bool: """Get verbose setting from resolved output config.""" return self._verbose @verbose.setter def verbose(self, value: bool): """Set verbose setting.""" self._verbose = value @property def stream(self) -> bool: """Get stream setting from resolved output config.""" return self._stream @property def planning_llm(self) -> Optional[str]: """Get planning LLM from resolved planning config.""" return self._planning_llm @property def reasoning(self) -> bool: """Get reasoning setting from resolved planning config.""" return self._reasoning @property def memory_config(self) -> Optional[Dict[str, Any]]: """Get memory config dict for backward compatibility.""" if self._memory_config: if hasattr(self._memory_config, 'to_dict'): return self._memory_config.to_dict() elif isinstance(self._memory_config, dict): return self._memory_config return None @property def on_workflow_start(self) -> Optional[Callable]: """Get on_workflow_start callback from hooks config.""" return self._hooks_config.on_workflow_start if self._hooks_config else None @property def on_workflow_complete(self) -> Optional[Callable]: """Get on_workflow_complete callback from hooks config.""" return self._hooks_config.on_workflow_complete if self._hooks_config else None @property def on_step_start(self) -> Optional[Callable]: """Get on_step_start callback from hooks config.""" return self._hooks_config.on_step_start if self._hooks_config else None @property def on_step_complete(self) -> Optional[Callable]: """Get on_step_complete callback from hooks config.""" return self._hooks_config.on_step_complete if self._hooks_config else None @property def on_step_error(self) -> Optional[Callable]: """Get on_step_error callback from hooks config.""" return self._hooks_config.on_step_error if self._hooks_config else None def _resolve_pydantic_class(self, class_name: str) -> Optional[Any]: """ Resolve a Pydantic class from string reference (Option B for structured output). Looks for the class in: 1. tools.py in the workflow's directory 2. The global namespace Args: class_name: Name of the Pydantic class to resolve Returns: The Pydantic class if found, None otherwise """ import sys # Try to find the class in tools.py from the workflow's directory if self.file_path: from pathlib import Path workflow_dir = Path(self.file_path).parent tools_path = workflow_dir / "tools.py" if tools_path.exists(): _local_ok = os.environ.get("PRAISONAI_ALLOW_LOCAL_TOOLS", "").strip().lower() in ( "true", "1", "yes", ) if not _local_ok: logger.debug( "Skipping tools.py load for %s: set PRAISONAI_ALLOW_LOCAL_TOOLS=true", class_name, ) return None try: import importlib.util spec = importlib.util.spec_from_file_location("tools", tools_path) if spec and spec.loader: tools_module = importlib.util.module_from_spec(spec) spec.loader.exec_module(tools_module) if hasattr(tools_module, class_name): cls = getattr(tools_module, class_name) # Verify it's a Pydantic model if hasattr(cls, 'model_json_schema'): return cls except Exception as e: logger.debug(f"Failed to load Pydantic class from tools.py: {e}") # Try to find in already-imported modules for module_name, module in sys.modules.items(): if module and hasattr(module, class_name): cls = getattr(module, class_name) if hasattr(cls, 'model_json_schema'): return cls return None @classmethod def from_template( cls, uri: str, config: Optional[Dict[str, Any]] = None, offline: bool = False, **kwargs ) -> 'Workflow': """ Create a Workflow from a template. Args: uri: Template URI (local path, package ref, or github ref) Examples: - "./my-template" (local path) - "transcript-generator" (default recipes repo) - "github:owner/repo/template@v1.0.0" (GitHub with version) - "package:agent_recipes/transcript-generator" (installed package) config: Optional configuration overrides offline: If True, only use cached templates (no network) **kwargs: Additional Workflow constructor arguments Returns: Configured Workflow instance Example: ```python from praisonaiagents import Workflow # From default recipes repo workflow = Workflow.from_template("transcript-generator") result = workflow.run("./audio.mp3") # With config overrides workflow = Workflow.from_template( "data-transformer", config={"output_format": "json"} ) ``` """ try: # Lazy import to avoid circular dependencies and keep core SDK lean from praisonai.templates.loader import create_workflow_from_template return create_workflow_from_template(uri, config=config, offline=offline, **kwargs) except ImportError: raise ImportError( "Template support requires the 'praisonai' package. " "Install with: pip install praisonai" ) def _check_handoff_cycle(self, step: Any) -> None: """Check for cross-step handoff cycles (Gap 3c fix).""" step_id = getattr(step, 'name', str(step)) # If step involves handoff (basic check for agent steps) if hasattr(step, 'agent') and step.agent: # Track this step in handoff chain if step_id in self._handoff_chain: # Cycle detected! cycle_path = self._handoff_chain[self._handoff_chain.index(step_id):] + [step_id] from ..errors import HandoffCycleError raise HandoffCycleError( f"Cross-step handoff cycle detected: {' -> '.join(cycle_path)}", cycle_path=cycle_path ) # Add to chain (limit chain length to prevent memory issues) self._handoff_chain.append(step_id) if len(self._handoff_chain) > 100: # Reasonable limit self._handoff_chain = self._handoff_chain[-50:] # Keep last 50 else: # Non-agent step, reset chain self._handoff_chain.clear() def _validate_framework(self) -> None: """Validate the workflow's ``framework`` before execution. The YAML parser may set a ``framework`` attribute on the workflow (e.g. from ``framework: crewai`` in a workflow YAML). The native execution engine only supports ``praisonai``, so any other value must fail fast with a clear, actionable error rather than being silently ignored. Backward-compatible: when ``framework`` is unset, empty, or ``"praisonai"`` (case-insensitive), this is a no-op. """ framework = getattr(self, 'framework', None) if framework is None: return if str(framework).strip().lower() in ("", "praisonai"): return # Include workflow identity (name/file) so included-recipe failures # can be traced back to the exact YAML that triggered them. identity_parts = [] name = getattr(self, 'name', None) if name: identity_parts.append(f"workflow '{name}'") file_path = getattr(self, 'file_path', None) if file_path: identity_parts.append(f"file '{file_path}'") identity = f" ({', '.join(identity_parts)})" if identity_parts else "" raise ValueError( f"Unsupported workflow framework: '{framework}'{identity}. " "Workflow YAML execution only supports the native PraisonAI " "engine (framework: praisonai). Remove the 'framework' field or " "set it to 'praisonai'." ) def run( self, input: str = "", llm: Optional[str] = None, verbose: bool = False, stream: bool = None ) -> Dict[str, Any]: """ Run the workflow with the given input. This is the simplest way to execute a workflow: ```python from praisonaiagents import Workflow, Agent workflow = Workflow( steps=[ Agent(name="Writer", role="Write content"), my_validator_function, Agent(name="Editor", role="Edit content"), ] ) result = workflow.run("Write about AI") ``` Args: input: The input text/prompt for the workflow llm: LLM model to use (default: gpt-4o-mini) verbose: Print step outputs stream: Enable streaming responses (default: use workflow's stream setting) Returns: Dict with 'output' (final result) and 'steps' (all step results) """ # Gap 3: refuse concurrent re-entrancy on the same instance. Per-run # mutable state (status/step_statuses/_handoff_chain) is not safe to # share across concurrent run()/astart() calls, so acquire a lock for # the whole run and raise a clear error instead of silently corrupting. if not self._execution_lock.acquire(blocking=False): raise RuntimeError( "This Workflow instance is already running; a Workflow is not " "safe to run() concurrently on the same object. Create a " "separate Workflow instance per concurrent run." ) try: from ..trace.context_events import _workflow_scope _process = "hierarchical" if self.process == "hierarchical" else "flow" with _workflow_scope("flow", self.name, _process, graph_fn=self._replay_graph) as _run: if self.tools_run_on is None: result = self._run_impl(input, llm, verbose, stream) else: # One sandbox for the whole run; torn down even if a step raises. with self._shared_compute() as shared: shared.attach(self._collect_agents()) # Expose the live sandbox so agents created *during* the run # (e.g. an action-only step's temporary agent) can be attached # too, instead of silently executing tools locally. self._active_shared_compute = shared try: result = self._run_impl(input, llm, verbose, stream) finally: self._active_shared_compute = None if _run is not None and isinstance(result, dict) and result.get("status") == "failed": _run["status"] = "failed" return result finally: # Restore the generation the caller's context carried before this # run stamped its own, so a nested Workflow.run() (Include) does not # leave a stale generation on the outer run's thread. token, self._run_generation_token = self._run_generation_token, None self._end_run_generation(token) self._execution_lock.release() def to_mermaid(self) -> str: """Render this flow's DEFINITION as a mermaid diagram. Unlike the telemetry diagrams, which need a captured execution trace, this reads the step list -- so a flow can be checked before it is run and before it costs anything. Mermaid needs no extra dependency and renders inline on GitHub and in the docs. print(flow.to_mermaid()) """ from .diagram import flow_to_mermaid return flow_to_mermaid(self) def to_graph(self) -> Dict[str, Any]: """This flow's definition as graph data (``praisonai.graph/1``). Node ids are stable across processes (derived from step names and positions), and are the ``node_id`` values replay trace events carry. """ from .diagram import to_graph return to_graph(self) @property def state(self): """The flow's variables as the declared ``state_model``. Returns None when no model was declared, so callers can tell "untyped" from "typed and empty" rather than being handed a misleading blank. """ from .state import build_state if self.state_model is None: return None return build_state(self.state_model, self.variables) def validate_variables(self) -> None: """Raise unless the current variables fit ``state_model``. Called for you at construction; exposed so a step that writes variables can re-check before the next step reads them. """ from .state import validate_variables validate_variables(self.state_model, self.variables) def __repr__(self): """Show where this workflow's steps run. Defined in the class body on purpose: @dataclass leaves a hand-written __repr__ alone, and the generated one listed every field while saying nothing about execution location. """ from ..agent.execution_location import repr_fields return ( f"AgentFlow(name={self.name!r}, steps={len(self.steps or [])}, " f"{repr_fields(self, shared=True)})" ) def where_does_it_run(self) -> str: """Plain-English answer to "where does my code actually run?".""" from ..agent.execution_location import explain return explain(self, shared=True) def _shared_compute(self): """Build the SharedCompute for this run (see ``tools_run_on=``).""" from ..managed.shared_compute import SharedCompute return SharedCompute(self.tools_run_on) def _collect_agents(self) -> List[Any]: """Walk steps (including nested route/parallel/loop/repeat/if) for Agents. Returns each Agent once, in first-seen order. Anything that is not an Agent -- plain functions, Tasks without an agent -- is ignored. """ found: List[Any] = [] seen = set() def visit(step: Any) -> None: if step is None: return if isinstance(step, (list, tuple)): for item in step: visit(item) return # Nested step containers. if isinstance(step, Route): for branch in step.routes.values(): visit(branch) visit(step.default) return if isinstance(step, Parallel): visit(step.steps) return if isinstance(step, Loop): visit(step.step) visit(step.steps) return if isinstance(step, Discussion): visit(step.agents) return if isinstance(step, Repeat): visit(step.step) return if isinstance(step, If): visit(step.then_steps) visit(step.else_steps) return # A Task carries the agent that will execute it. agent = getattr(step, "agent", None) candidate = agent if agent is not None else step # Duck-type an Agent: has chat() and instructions/role. if hasattr(candidate, "chat") and ( hasattr(candidate, "instructions") or hasattr(candidate, "role") ): if id(candidate) not in seen: seen.add(id(candidate)) found.append(candidate) visit(self.steps) return found def _notify_step_start(self, raw_step: Any, step: Any, context: Any) -> Optional[dict]: """Emit replay ``step_start`` then call ``on_step_start``; returns the span.""" span = _replay_step_begin(self, raw_step, step.name) if self.on_step_start: try: self.on_step_start(step.name, context) except Exception as e: logger.error(f"on_step_start callback failed: {e}") return span def _notify_step_complete( self, span: Optional[dict], step_name: str, step_result: Any, status: str, output: Any = None, ) -> None: """Call ``on_step_complete`` then emit replay ``step_end``.""" if self.on_step_complete: try: self.on_step_complete(step_name, step_result) except Exception as e: logger.error(f"on_step_complete callback failed: {e}") _replay_step_end(span, status, output) def _replay_graph(self) -> Dict[str, Any]: """Graph for ``workflow_start``; also maps step objects to node ids.""" from .diagram import _flow_graph graph = _flow_graph(self) self._replay_node_ids = graph.ids return graph.to_dict() def _run_impl( self, input: str = "", llm: Optional[str] = None, verbose: bool = False, stream: bool = None ) -> Dict[str, Any]: """Internal run implementation (see run() for the concurrency guard).""" # Gap 3c: Clear handoff chain at start of new workflow run self._handoff_chain.clear() # Fall back to the YAML-defined default input when no explicit input # is provided. The parser stores the top-level `input:` field as # `default_input`; honoring it here ensures `{{input}}` substitution # works for both CLI and programmatic `start()`/`run()` calls. if not input: input = getattr(self, "default_input", "") or "" # Fail fast if a non-praisonai framework is requested via YAML. # The YAML parser stores `framework` on the workflow, but execution # always uses the native PraisonAI engine. Silently ignoring a # non-praisonai value (e.g. framework: crewai) is a correctness bug, # so raise a clear error instead of producing misleading results. self._validate_framework() # Use default LLM if not specified model = default_auxiliary_model(llm or self.llm) logger.debug(f"Workflow using model: {model} (llm={llm}, default_llm={self.llm})") # Agents the flow builds itself already receive the default via # llm=config.get("llm", model). Ready-made Agent steps were skipped, so # AgentFlow(llm=...) silently did nothing for them. Apply the same rule # here: fill in agents that never named a model, leave the rest alone. # The "gpt-4o-mini" fallback above is deliberately not propagated -- it # is not a caller's choice, and Agent has its own env-aware default. _default_model = llm or self.llm if _default_model: for _agent in self._collect_agents(): _apply = getattr(_agent, "_apply_default_llm", None) if callable(_apply): _apply(_default_model) # Use workflow verbose setting if not overridden verbose = verbose or self.verbose # Use workflow stream setting if not overridden if stream is None: stream = self.stream # Add input to variables all_variables = {**self.variables, "input": input} results = [] previous_output = None # Update workflow status self.status = "running" # Stamp a fresh generation for this run so a still-running branch from a # previous fail_fast/timeout Parallel block (which cannot be force-killed) # cannot write "failed" onto this run's status. Branches capture this via # the _run_generation contextvar at submit time. The reset token is kept # on the instance so run()'s finally can restore the context the caller # had -- critical for nested Workflow.run() (Include), which must not # leave its own generation stamped on the outer run's thread. self._run_generation_token = self._begin_run_generation() # Set YAML-approved tools only when caller allows dangerous tools _approval_token = None approve_tools = getattr(self, 'approve_tools', []) allow_dangerous = getattr(self, 'allow_dangerous_tools', False) if approve_tools and allow_dangerous: from ..approval import set_yaml_approved_tools _approval_token = set_yaml_approved_tools(approve_tools) # Call on_workflow_start callback if self.on_workflow_start: try: self.on_workflow_start(self, input) except Exception as e: logger.error(f"on_workflow_start callback failed: {e}") # Planning mode - create execution plan before running if self.planning: plan = self._create_plan(input, model, verbose) if plan and verbose: print(f"📋 Execution Plan: {plan}") all_variables["execution_plan"] = plan # Hierarchical mode - use manager-based validation if self.process == "hierarchical": if verbose: print("🔄 Running workflow in hierarchical mode with manager validation") return self._run_hierarchical( input=input, model=model, verbose=verbose, stream=stream, all_variables=all_variables, _approval_token=_approval_token ) # Process steps (may include Route, Parallel, Loop, Repeat) i = 0 while i < len(self.steps): step = self.steps[i] # Handle special pattern types if isinstance(step, Route): # Decision-based routing route_result = self._execute_route( step, previous_output, input, all_variables, model, verbose, stream ) results.extend(route_result["steps"]) previous_output = route_result["output"] all_variables.update(route_result.get("variables", {})) if route_result.get("stop"): self.status = "failed" if verbose: print("🛑 Workflow stopped by nested route step") break i += 1 continue elif isinstance(step, Parallel): # Parallel execution parallel_result = self._execute_parallel( step, previous_output, input, all_variables, model, verbose, stream ) results.extend(parallel_result["steps"]) previous_output = parallel_result["output"] all_variables.update(parallel_result.get("variables", {})) # A nested step with on_error="stop" halts the whole workflow, # not just the parallel block: honor the propagated stop signal. if parallel_result.get("stop"): self.status = "failed" if verbose: print("🛑 Workflow stopped by nested parallel step") break i += 1 continue elif isinstance(step, Loop): # Loop over items loop_result = self._execute_loop( step, previous_output, input, all_variables, model, verbose, stream ) results.extend(loop_result["steps"]) previous_output = loop_result["output"] all_variables.update(loop_result.get("variables", {})) # A nested step with on_error="stop" halts the whole workflow, # not just the loop: honor the propagated stop signal here. if loop_result.get("stop"): self.status = "failed" if verbose: print("🛑 Workflow stopped by nested loop step") break i += 1 continue elif isinstance(step, Discussion): discussion_result = self._execute_discussion( step, previous_output, input, all_variables, model, verbose, stream ) results.extend(discussion_result["steps"]) previous_output = discussion_result["output"] all_variables.update(discussion_result.get("variables", {})) # A speaker with on_error="stop" halts the whole workflow, not # just the discussion: honor the propagated stop signal here. if discussion_result.get("stop"): self.status = "failed" if verbose: print("🛑 Workflow stopped by nested discussion step") break i += 1 continue elif isinstance(step, Repeat): # Repeat until condition repeat_result = self._execute_repeat( step, previous_output, input, all_variables, model, verbose, stream ) results.extend(repeat_result["steps"]) previous_output = repeat_result["output"] all_variables.update(repeat_result.get("variables", {})) if repeat_result.get("stop"): self.status = "failed" if verbose: print("🛑 Workflow stopped by nested repeat step") break i += 1 continue elif isinstance(step, Include): # Include another recipe include_result = self._execute_include( step, previous_output, input, all_variables, model, verbose, stream ) results.extend(include_result["steps"]) previous_output = include_result["output"] all_variables.update(include_result.get("variables", {})) i += 1 continue elif isinstance(step, If): # Conditional branching if_result = self._execute_if( step, previous_output, input, all_variables, model, verbose, stream ) results.extend(if_result["steps"]) previous_output = if_result["output"] all_variables.update(if_result.get("variables", {})) if if_result.get("stop"): self.status = "failed" if verbose: print("🛑 Workflow stopped by nested conditional step") break i += 1 continue # Normalize single step raw_step = step step = self._normalize_single_step(step, i) # Create context for handlers context = WorkflowContext( input=input, previous_result=str(previous_output) if previous_output else None, current_step=step.name, variables=all_variables.copy() ) # Start a step attempt before notifying lifecycle observers. The # attempt may finish as skipped or from cache without a handler call. if hasattr(step, 'status'): step.status = "running" self.step_statuses[step.name] = "running" # Preserve the preparation hook before the gate: it may populate # context variables used by should_run, including on cached runs. _replay = self._notify_step_start(raw_step, step, context) # Check should_run condition if step.should_run: try: if not step.should_run(context): if verbose: print(f"⏭️ Skipped: {step.name}") if hasattr(step, 'status'): step.status = "skipped" self.step_statuses[step.name] = "skipped" self._notify_step_complete( _replay, step.name, StepResult(output="", skipped=True), "skipped", ) i += 1 continue except Exception as e: logger.error(f"should_run failed for {step.name}: {e}") # Cache only the result, never the decision to run. A condition can # observe external state that is absent from the cache key, so it # must be evaluated even when identical inputs have a cached result. _step_cache = getattr(self, "_step_cache", None) _cache_key = None if _step_cache is not None: from .step_cache import make_step_key if step.handler: _cache_key = make_step_key(step, context.previous_result, context.input, context.variables) else: # Agent prompts and attachments consume workflow inputs, # not the handler context prepared by lifecycle hooks. _cache_key = make_step_key(step, previous_output, input, all_variables) try: _cached = copy.deepcopy(_step_cache.get(_cache_key)) except Exception: _cached = None logger.warning("Could not restore cached workflow step '%s'", step.name) if _cached is not None: if verbose: print(f"↩︎ cache hit: {step.name}") previous_output = _cached.get("output") # Replay the full step delta recorded on the cold run so a # hit is indistinguishable from re-executing: the step's # status/retries, any handler-supplied variables (and the # generated output variable, present only when the cold run # did not stop first), and -- crucially -- the stop signal. # A hit that only appended {step, output} and continued lost # the stop flag and the handler's variables, so a workflow # that stopped on the cold run ran on through the cached one. cached_record = _cached.get("step_record") or { "step": step.name, "output": previous_output, "status": "completed", "retries": 0, } results.append(dict(cached_record)) self.step_statuses[step.name] = cached_record["status"] if hasattr(step, "status"): step.status = cached_record["status"] if _cached.get("variables"): all_variables.update(_cached["variables"]) self._notify_step_complete( _replay, step.name, StepResult( output=previous_output or "", stop_workflow=bool(_cached.get("stop")), ), "cached", previous_output, ) if _cached.get("stop"): if verbose: print(f"🛑 Workflow stopped at: {step.name}") break i += 1 continue # Gap 3c: Check for cross-step handoff cycles self._check_handoff_cycle(step) # Execute step with retry and guardrail support output = None stop = False step_error = None max_retries = getattr(step, 'max_retries', 3) retry_count = 0 validation_feedback = None guardrail_failed = False cached_variable_updates = {} if _cache_key is not None else None while retry_count <= max_retries: step_error = None # Add validation feedback to context if retrying if validation_feedback: context.variables["validation_feedback"] = validation_feedback try: if step.handler: # Custom handler function result = step.handler(context) if isinstance(result, StepResult): output = result.output stop = result.stop_workflow if result.variables: # Accumulate across retries so the cached # snapshot matches the cold run: a rejected # attempt that wrote {a, b} followed by an # accepted attempt that wrote {a} leaves both # keys in the live run, so both must be cached. all_variables.update(result.variables) if cached_variable_updates is not None: cached_variable_updates.update(result.variables) else: output = str(result) elif step.agent: # Direct agent with tools # Propagate context management to existing agent if workflow has it enabled if self.context and hasattr(step.agent, '_context_manager_initialized'): if not step.agent._context_manager_initialized: step.agent._context_param = self.context # Share session deduplication cache for cross-agent deduplication if self._session_dedup_cache: step.agent._session_dedup_cache = self._session_dedup_cache # Also set on existing context manager if already initialized if step.agent._context_manager and hasattr(step.agent._context_manager, '_session_cache'): step.agent._context_manager._session_cache = self._session_dedup_cache # Substitute variables in action action = step.action or input action = _substitute_action_variables(action, all_variables, previous_output, input) # Add reasoning prompt if enabled if self.reasoning: action = f"Think step by step and reason through this task:\n\n{action}" # Add validation feedback if retrying if validation_feedback: action = f"{action}\n\nPrevious attempt feedback: {validation_feedback}" # Check if this is a specialized agent (AudioAgent, VideoAgent, ImageAgent, OCRAgent) agent_class_name = step.agent.__class__.__name__ if agent_class_name in ('AudioAgent', 'VideoAgent', 'ImageAgent', 'OCRAgent', 'DeepResearchAgent'): # Use specialized agent dispatch output = self._execute_specialized_agent( step.agent, agent_class_name, action, step, all_variables, stream ) else: # Standard Agent with chat() method # Handle images/attachments if present - check step.images first, then variables step_attachments = getattr(step, 'images', None) if not step_attachments: # Check for image_path, image_url, or image in variables image_var = all_variables.get('image_path') or all_variables.get('image_url') or all_variables.get('image') if image_var: step_attachments = [image_var] if isinstance(image_var, str) else image_var logger.debug(f"Passing image attachment to agent: {image_var}") # Get structured output options from step step_output_json = getattr(step, '_output_json', None) step_output_pydantic = getattr(step, '_output_pydantic', None) # Resolve Pydantic class from string reference (Option B) # If output_pydantic is a string, try to resolve it from tools.py if step_output_pydantic and isinstance(step_output_pydantic, str): resolved_class = self._resolve_pydantic_class(step_output_pydantic) if resolved_class: step_output_pydantic = resolved_class else: logger.warning(f"Could not resolve Pydantic class '{step_output_pydantic}' from tools.py") step_output_pydantic = None # Build chat kwargs chat_kwargs = {"stream": stream} if step_attachments: chat_kwargs["attachments"] = step_attachments if step_output_json: chat_kwargs["output_json"] = step_output_json if step_output_pydantic: chat_kwargs["output_pydantic"] = step_output_pydantic # Pass tool_choice from YAML if specified (auto, required, none) # This forces the LLM to use tools when tool_choice='required' yaml_tool_choice = getattr(step.agent, '_yaml_tool_choice', None) if yaml_tool_choice: chat_kwargs["tool_choice"] = yaml_tool_choice output = step.agent.chat(action, **chat_kwargs) # Parse JSON output if output_json was requested and output is a string if step_output_json and output and isinstance(output, str): output = _parse_json_output(output, step.name) # Handle output_pydantic if present output_pydantic = getattr(step, 'output_pydantic', None) if output_pydantic and output: try: # Try to parse output as Pydantic model if hasattr(output_pydantic, 'model_validate_json'): parsed = output_pydantic.model_validate_json(output) output = parsed.model_dump_json() except Exception as e: logger.debug(f"Pydantic parsing failed: {e}") elif step.action: # Action with agent_config - create temporary agent from ..agent.agent import Agent config = step.agent_config or self.default_agent_config or {} # Get tools from step or config step_tools = step.tools or config.get("tools", []) temp_agent = Agent( name=config.get("name", step.name), role=config.get("role", "Assistant"), goal=config.get("goal", "Complete the task"), llm=config.get("llm", model), tools=step_tools if step_tools else None, output=self.output, # Propagate output config context=self.context, # Propagate context management memory=self.memory, # Propagate memory config knowledge=self.knowledge, # Propagate knowledge/RAG guardrails=self.guardrails, # Propagate guardrails web=self.web, # Propagate web search/fetch reflection=self.reflection, # Propagate self-reflection autonomy=self.autonomy, # Propagate autonomy settings caching=self.caching, # Propagate caching config hooks=self.hooks, # Propagate hooks/callbacks ) # An action-only step builds its agent here, after the # run's initial attach(). Bind it to the shared sandbox # too, so its shell/file tools land in the same instance # as every other step instead of running locally. shared = getattr(self, "_active_shared_compute", None) if shared is not None: shared.attach([temp_agent]) # Substitute variables in action action = step.action action = _substitute_action_variables(action, all_variables, previous_output, input) # Add reasoning prompt if enabled if self.reasoning: action = f"Think step by step and reason through this task:\n\n{action}" if validation_feedback: action = f"{action}\n\nPrevious attempt feedback: {validation_feedback}" output = temp_agent.chat(action, stream=stream) # Parse JSON output if output_json was requested step_output_json = getattr(step, '_output_json', None) if step_output_json and output and isinstance(output, str): output = _parse_json_output(output, step.name) except Exception as e: step_error = e output = f"Error: {e}" # Gap 3b fix: Check if error is retryable and implement exponential backoff is_retryable = getattr(e, 'is_retryable', True) # Default to retryable if not is_retryable: # Non-retryable error - break out of retry loop immediately if verbose: print(f"❌ {step.name} failed with non-retryable error: {e}") break retry_count += 1 if retry_count <= max_retries: # Exponential backoff (shared with _apply_step_policies) self._retry_backoff(step, retry_count, max_retries, e, verbose) continue # Retry if self.on_step_error: try: self.on_step_error(step.name, e) except Exception: pass # Check guardrail if present (guardrails is canonical, guardrail is deprecated) guardrail = getattr(step, 'guardrails', None) or getattr(step, 'guardrail', None) if guardrail and output and not step_error: try: is_valid, feedback = guardrail(StepResult(output=output)) if not is_valid: validation_feedback = str(feedback) guardrail_failed = True retry_count += 1 if verbose: print(f"⚠️ {step.name} failed validation (attempt {retry_count}/{max_retries}): {feedback}") continue # Retry guardrail_failed = False except Exception as e: logger.error(f"Guardrail failed for {step.name}: {e}") # Success - break out of retry loop break # A step whose guardrail never validated after exhausting retries is # a failure too, even though no exception was raised. Treat it like # step_error so on_error flow control and the final status are honored # instead of silently reporting the step "completed". # # So is an LLM-backed step that produced no output at all. `chat()` # returns None when the underlying call failed and the agent # swallowed the error (an auth 401 is the common case): the engine # already logs "Output is None" a few lines below, so it knows the # step produced nothing — it just used to report the run # "completed" anyway, and every consumer (CLI, API server, Python # callers) was told the run succeeded. Only agent/action steps are # judged this way: a custom `handler` may legitimately return None, # and a skipped step never reaches here. produced_nothing = ( output is None and not step_error and not step.handler and (step.agent is not None or bool(step.action)) ) step_failed = bool(step_error) or guardrail_failed or produced_nothing failure_reason = ( str(step_error) if step_error else (f"guardrail validation failed: {validation_feedback}" if guardrail_failed else ("agent produced no output (None) — the model call " "most likely failed" if produced_nothing else None)) ) # Update step status if step_failed: if hasattr(step, 'status'): step.status = "failed" self.step_statuses[step.name] = "failed" else: if hasattr(step, 'status'): step.status = "completed" self.step_statuses[step.name] = "completed" # Honor the step's on_error flow-control setting. When a step # exhausts its retries (error or unresolved guardrail) and # on_error == "stop" (the Task default), abort the workflow and mark # it failed instead of feeding the error string forward into the next # step and falsely reporting overall success. if step_failed and getattr(step, 'on_error', 'stop') == 'stop': self.status = "failed" results.append({ "step": step.name, "output": output, "status": "failed", "retries": retry_count, "error": failure_reason, }) # Still notify the step-complete callback so lifecycle observers # see the (failed) terminal step before the workflow aborts. self._notify_step_complete( _replay, step.name, StepResult(output=output or "", stop_workflow=True), "failed", output, ) if verbose: print(f"🛑 Workflow stopped: step '{step.name}' failed (on_error='stop')") break # Create step result for callback step_result = StepResult(output=output or "", stop_workflow=stop) # Call on_step_complete callback self._notify_step_complete( _replay, step.name, step_result, "failed" if step_failed else "completed", output, ) # Handle output_file - save output to file if hasattr(step, 'output_file') and step.output_file and output: try: import os output_path = step.output_file # Substitute variables in path for key, value in all_variables.items(): output_path = output_path.replace(f"{{{{{key}}}}}", str(value)) # Create directory if needed os.makedirs(os.path.dirname(output_path) if os.path.dirname(output_path) else ".", exist_ok=True) with open(output_path, "w") as f: f.write(str(output)) if verbose: print(f"📁 Saved output to: {output_path}") except Exception as e: logger.error(f"Failed to save output to file: {e}") # Store result. A step that failed under on_error="continue" reaches # here (the on_error="stop" branch above already returned); carry its # failure_reason through so the final result surfaces *why* it failed # rather than the generic "step(s) failed: " fallback. step_record = { "step": step.name, "output": output, "status": self.step_statuses.get(step.name, "completed"), "retries": retry_count } if step_failed and failure_reason: step_record["error"] = failure_reason results.append(step_record) previous_output = output # Only a SUCCESSFUL step is cached. Caching a failure would serve # the failure again on every re-run, turning a transient error into # a permanent one that no retry could clear. The step's output # variables and stop signal must be stored too: a cache hit must # preserve handler state and control flow, not just output text. # Store only this step's variable updates, including its output # variable unless it stopped before that variable was written. if _cache_key is not None and not step_failed: if not stop: _cached_var_name = step.output_variable or f"{step.name}_output" cached_variable_updates[_cached_var_name] = output try: _step_cache.set( _cache_key, copy.deepcopy({ "output": output, "variables": cached_variable_updates, "stop": stop, "step_record": step_record, }), ) except Exception: # Custom caches may only accept serializable state. Keep # the successful workflow result when storage rejects it. logger.warning("Could not cache workflow step '%s'", step.name) if verbose: print(f"✅ {step.name}: {str(output)}") # Handle early stop if stop: if verbose: print(f"🛑 Workflow stopped at: {step.name}") break # Store output in variables var_name = step.output_variable or f"{step.name}_output" all_variables[var_name] = output # Validate output and warn about issues if output is None: logger.warning(f"⚠️ Step '{step.name}': Output is None. Agent may not have returned expected format.") if verbose: print(f"⚠️ WARNING: Step '{step.name}' output is None!") else: # Check type against output_json schema if defined expected_schema = getattr(step, '_output_json', None) if expected_schema and isinstance(expected_schema, dict): expected_type = expected_schema.get('type') actual_type = type(output).__name__ if expected_type == 'object' and not isinstance(output, dict): logger.warning(f"⚠️ Step '{step.name}': Expected object/dict, got {actual_type}") if verbose: print(f"⚠️ Step '{step.name}': Expected 'object', received '{actual_type}'") elif expected_type == 'array' and not isinstance(output, list): logger.warning(f"⚠️ Step '{step.name}': Expected array/list, got {actual_type}") if verbose: print(f"⚠️ Step '{step.name}': Expected 'array', received '{actual_type}'") i += 1 # Update workflow status. Reflect any unresolved step failure (e.g. a # step with on_error="continue" that still failed) instead of always # reporting "completed". A prior on_error="stop" break already set # self.status = "failed". if self.status != "failed" and any( r.get("status") == "failed" for r in results ): self.status = "failed" elif self.status != "failed": self.status = "completed" # Reset YAML-approved tools context if it was set if _approval_token is not None: from ..approval import reset_yaml_approved_tools reset_yaml_approved_tools(_approval_token) final_result = { "output": previous_output, "steps": results, "variables": all_variables, "status": self.status } # Surface *why* a failed run failed. Callers (the CLI included) read # result["error"]; without it a genuine failure printed # "Workflow failed: Unknown error". if self.status == "failed": reasons = [ f"{r.get('step')}: {r.get('error')}" for r in results if r.get("status") == "failed" and r.get("error") ] if reasons: final_result["error"] = "; ".join(reasons) else: failed_names = [ str(r.get("step")) for r in results if r.get("status") == "failed" ] final_result["error"] = ( "step(s) failed: " + ", ".join(failed_names) if failed_names else "workflow failed" ) # Call on_workflow_complete callback if self.on_workflow_complete: try: self.on_workflow_complete(self, final_result) except Exception as e: logger.error(f"on_workflow_complete callback failed: {e}") return final_result async def astart( self, input: str = "", llm: Optional[str] = None, verbose: bool = False ) -> Dict[str, Any]: """Async version of start() for async workflow execution. Args: input: The input text/prompt for the workflow llm: LLM model to use (default: gpt-4o-mini) verbose: Print step outputs Returns: Dict with 'output' (final result) and 'steps' (all step results) """ import asyncio # Run the synchronous version in a thread pool # Use copy_context_to_callable to propagate contextvars (needed for trace emission) from ..trace.context_events import copy_context_to_callable loop = asyncio.get_event_loop() result = await loop.run_in_executor( None, copy_context_to_callable(lambda: self.run(input, llm, verbose)) ) return result async def arun( self, input: str = "", llm: Optional[str] = None, verbose: bool = False ) -> Dict[str, Any]: """Alias for astart() for backward compatibility.""" return await self.astart(input, llm, verbose) def _run_hierarchical( self, input: str, model: str, verbose: bool, stream: bool, all_variables: Dict[str, Any], _approval_token: Optional[Any] = None ) -> Dict[str, Any]: """Run workflow with hierarchical manager-based validation. In hierarchical mode, a manager agent validates each step's output before proceeding to the next step. If the manager rejects a step, the workflow stops with status='failed' and includes a failure_reason. This pattern is adapted from Process.hierarchical() for Workflow use. Args: input: The workflow input model: LLM model to use for steps verbose: Print step outputs stream: Enable streaming all_variables: Variables dict _approval_token: Token for YAML-approved tools Returns: Dict with output, steps, variables, status, and optionally failure_reason """ from ..llm import LLM results = [] previous_output = input failure_reason = None manager_model = self.manager_llm or model normalized_steps = self._normalize_steps() for i, step in enumerate(normalized_steps): step_name = getattr(step, 'name', f'step_{i}') if self.on_step_start: try: self.on_step_start(self, step, i) except Exception as e: logger.error(f"on_step_start callback failed: {e}") if hasattr(step, 'status'): step.status = "running" self.step_statuses[step_name] = "running" try: # Use _execute_single_step_internal which returns a dict step_result_internal = self._execute_single_step_internal( step, previous_output, input, all_variables, model, verbose, i, stream ) if step_result_internal.get("skipped"): results.append({"step": step_name, "output": None, "status": "skipped"}) continue output = step_result_internal.get("output", "") stop = step_result_internal.get("stop", False) step_vars = step_result_internal.get("variables", {}) all_variables.update(step_vars) # An LLM-backed step that produced nothing is a failure here # too, exactly as in the sequential path: `chat()` returns None # when the model call failed and the agent swallowed the error. # This loop used to mark every non-raising step "completed" # before the manager ever saw it. produced_nothing = ( output is None and not getattr(step, 'handler', None) and (getattr(step, 'agent', None) is not None or bool(getattr(step, 'action', None))) ) step_result = { "step": step_name, "output": output, "status": "failed" if produced_nothing else "completed" } results.append(step_result) if produced_nothing: failure_reason = ( f"Step '{step_name}' produced no output (None) -- the " f"model call most likely failed" ) logger.error(failure_reason) self.status = "failed" if hasattr(step, 'status'): step.status = "failed" self.step_statuses[step_name] = "failed" step_result["failure_reason"] = failure_reason if self.on_step_error: try: self.on_step_error(self, step, Exception(failure_reason)) except Exception as callback_e: logger.error(f"on_step_error callback failed: {callback_e}") break if hasattr(step, 'status'): step.status = "completed" self.step_statuses[step_name] = "completed" if self.on_step_complete: try: self.on_step_complete(self, step, step_result) except Exception as e: logger.error(f"on_step_complete callback failed: {e}") if stop: if verbose: print(f"🛑 Workflow stopped at: {step_name}") break # Check if step has tools assigned step_has_tools = bool(getattr(step, 'tools', None) or (hasattr(step, 'agent') and getattr(step.agent, 'tools', None))) # Build tool evidence guidance for manager tool_evidence_guidance = "" if step_has_tools: tool_evidence_guidance = """ 4. TOOL USAGE EVIDENCE: If the step required tool usage (e.g., search_web), check for EVIDENCE in the output: - URLs or links (e.g., https://..., [Source](url)) - Structured data with sources/citations - Specific facts with references - Multiple distinct items from external sources NOTE: The agent may have called tools successfully even if it doesn't say "I called the tool". Look for RESULTS from tools, not explicit statements about calling them.""" validation_prompt = f"""You are reviewing the output of a workflow step. Step Name: {step_name} Step Action: {getattr(step, 'action', 'Execute task')} Expected Output: {getattr(step, 'expected_output', 'Task completed successfully')} Agent Output: {output} Did this step complete successfully? Consider: 1. Does the output address the task? 2. Is the output meaningful (not an error message)? 3. Does it meet the expected output criteria?{tool_evidence_guidance} IMPORTANT: Approve if the output contains substantive content that addresses the task. Only reject if the output is clearly an error, empty, or completely off-topic. Respond with JSON: {{"approved": true/false, "reason": "Brief explanation"}} """ try: llm_instance = LLM(model=manager_model, verbose=False) validation_response = llm_instance.get_response( prompt=validation_prompt, system_prompt="You are a workflow quality manager. Respond only with valid JSON.", output_json=True ) # Parse the manager's verdict with the module's own JSON # helper, which already strips the ```json fences models # routinely emit. A bare json.loads() raised JSONDecodeError # on every fenced reply, and the handler then FAILED OPEN -- # turning "approved": false into "assuming success". A # manager rejection was therefore silently discarded and the # run reported completed, which is the same class of defect # as an all-None run reporting success. decision_data = _parse_json_output( validation_response, step_name ) if not isinstance(decision_data, dict): # Fail CLOSED: an unreadable verdict is not approval. decision_data = { "approved": False, "reason": ( "Manager validation response could not be " f"parsed: {str(validation_response)[:200]!r}" ), } approved = decision_data.get("approved", True) reason = decision_data.get("reason", "No reason provided") if verbose: status_icon = "✅" if approved else "❌" print(f"{status_icon} Manager validation for '{step_name}': {reason}") if not approved: failure_reason = f"Manager rejected step '{step_name}': {reason}" self.status = "failed" if hasattr(step, 'status'): step.status = "failed" self.step_statuses[step_name] = "failed" step_result["status"] = "failed" step_result["failure_reason"] = failure_reason if self.on_step_error: try: self.on_step_error(self, step, Exception(failure_reason)) except Exception as e: logger.error(f"on_step_error callback failed: {e}") break except Exception as e: # The manager is a GATE. If it could not run, it has not # passed -- reporting "completed successfully" because the # quality check itself failed is the same lie as approving a # rejection. This block already fails closed on an # unparseable verdict two lines up, so failing open here # would be incoherent. Hierarchical mode is opt-in: a caller # who asked for manager validation wants it enforced. failure_reason = ( f"Manager validation could not run for step " f"'{step_name}': {e}" ) logger.error(failure_reason) self.status = "failed" if hasattr(step, 'status'): step.status = "failed" self.step_statuses[step_name] = "failed" step_result["status"] = "failed" step_result["failure_reason"] = failure_reason if self.on_step_error: try: self.on_step_error(self, step, e) except Exception as callback_e: logger.error(f"on_step_error callback failed: {callback_e}") break previous_output = output # Store output in variables for next step's variable substitution # This is critical for {{agent_name}}_output references to work var_name = f"{step_name}_output" all_variables[var_name] = output except Exception as e: error_msg = str(e) if hasattr(step, 'status'): step.status = "failed" self.step_statuses[step_name] = "failed" step_result = { "step": step_name, "output": f"Error: {error_msg}", "status": "failed" } results.append(step_result) if self.on_step_error: try: self.on_step_error(self, step, e) except Exception as callback_e: logger.error(f"on_step_error callback failed: {callback_e}") failure_reason = f"Step '{step_name}' failed with error: {error_msg}" self.status = "failed" break if failure_reason is None: self.status = "completed" if _approval_token is not None: from ..approval import reset_yaml_approved_tools reset_yaml_approved_tools(_approval_token) final_result = { "output": previous_output, "steps": results, "variables": all_variables, "status": self.status } if failure_reason: final_result["failure_reason"] = failure_reason # Also expose it as `error`, the key every caller (the CLI # included) reads. Without it a genuine hierarchical failure # printed "Workflow failed: Unknown error". final_result.setdefault("error", failure_reason) if self.on_workflow_complete: try: self.on_workflow_complete(self, final_result) except Exception as e: logger.error(f"on_workflow_complete callback failed: {e}") return final_result def _normalize_steps(self) -> List['Task']: """Convert mixed steps (Agent, function, Task) to Task list. This method uses _normalize_single_step to ensure consistent normalization and avoid duplicated code paths (DRY principle). """ return [self._normalize_single_step(step, i) for i, step in enumerate(self.steps)] def _create_plan(self, input: str, model: str, verbose: bool) -> Optional[str]: """Create an execution plan for the workflow using LLM. Args: input: The workflow input model: LLM model to use verbose: Print verbose output Returns: Execution plan as string, or None if planning fails """ try: from ..agent.agent import Agent # Describe the steps step_descriptions = [] for i, step in enumerate(self.steps): if isinstance(step, Task): step_descriptions.append(f"{i+1}. {step.name}: {step.description or step.action or 'Execute step'}") elif hasattr(step, 'name'): role = getattr(step, 'role', 'Agent') step_descriptions.append(f"{i+1}. {step.name} ({role})") elif callable(step): step_descriptions.append(f"{i+1}. {getattr(step, '__name__', 'function')}") elif isinstance(step, (Route, Parallel, Loop, Repeat)): step_descriptions.append(f"{i+1}. {type(step).__name__} pattern") else: step_descriptions.append(f"{i+1}. Step {i+1}") steps_text = "\n".join(step_descriptions) planning_prompt = f"""You are a workflow planner. Given the following workflow steps and input, create a brief execution plan. Workflow: {self.name} Description: {self.description} Steps: {steps_text} Input: {input} Create a brief execution plan (2-3 sentences) describing how to best accomplish this task using the available steps. Focus on the key objectives and any important considerations.""" planner = Agent( name="Planner", role="Workflow Planner", goal="Create efficient execution plans", llm=self.planning_llm or model, ) plan = planner.chat(planning_prompt) return plan except Exception as e: logger.error(f"Planning failed: {e}") return None def _execute_specialized_agent( self, agent: Any, agent_class_name: str, action: str, step: 'Task', variables: Dict[str, Any], stream: bool ) -> Any: """ Execute a specialized agent (AudioAgent, VideoAgent, ImageAgent, OCRAgent). Dispatches to the appropriate method based on agent type and action. Falls back to chat() for standard Agent class. Args: agent: The agent instance agent_class_name: Name of the agent class action: The action/prompt to execute step: The workflow step containing additional config variables: Current workflow variables stream: Whether to stream output Returns: Output from the agent execution """ # Get yaml_action which specifies the method to call yaml_action = getattr(agent, '_yaml_action', None) or action # AudioAgent - TTS (speech) or STT (transcribe) if agent_class_name == 'AudioAgent': if yaml_action == 'speech' or 'speech' in yaml_action.lower(): # Text-to-Speech text = variables.get('text', action) output_file = variables.get('output') or variables.get('audio_output') or 'output.mp3' voice = variables.get('voice', 'alloy') result = agent.speech(text, output=output_file, voice=voice) # Store the output file path in variables for next step variables['_last_audio_file'] = output_file return result elif yaml_action == 'transcribe' or 'transcribe' in yaml_action.lower(): # Speech-to-Text - check multiple variable names for input file input_file = ( variables.get('audio_file') or variables.get('audio') or variables.get('_last_audio_file') or # From previous TTS step variables.get('input') ) # If input is empty string or template, use default if not input_file or input_file == '{{input}}' or 'Context from previous' in str(input_file): input_file = 'output.mp3' # Default TTS output language = variables.get('language', 'en') return agent.transcribe(input_file, language=language) else: # Default to speech if action contains text return agent.speech(action, output=variables.get('output', 'output.mp3')) # VideoAgent - generate elif agent_class_name == 'VideoAgent': prompt = variables.get('prompt', action) output_file = variables.get('output', 'output.mp4') return agent.generate(prompt, output=output_file) # ImageAgent - generate with output path support elif agent_class_name == 'ImageAgent': prompt = variables.get('prompt', action) output_path = variables.get('output') or variables.get('image_output') result = agent.generate(prompt) # Extract URL from result and store in variables for next step if hasattr(result, 'data') and result.data: image_url = result.data[0].url if hasattr(result.data[0], 'url') else None if image_url: variables['_last_image_url'] = image_url # Save to file if output path specified if output_path: try: import urllib.request urllib.request.urlretrieve(image_url, output_path) variables['_last_image_file'] = output_path except Exception as e: logger.warning(f"Failed to save image to {output_path}: {e}") return result # OCRAgent - extract with retry logic for external API failures elif agent_class_name == 'OCRAgent': source = variables.get('source') or variables.get('url') or variables.get('document') # If no explicit source variable, try to extract URL from action text if not source: import re url_pattern = r'https?://[^\s<>"{}|\\^`\[\]]+' urls = re.findall(url_pattern, action) source = urls[0] if urls else action # Retry logic for external API failures max_retries = 2 last_error = None for attempt in range(max_retries + 1): try: return agent.extract(source) except Exception as e: last_error = e error_str = str(e).lower() # Only retry on transient errors (network, timeout, fetch failures) if any(x in error_str for x in ['fetch', 'timeout', 'connection', 'network']): if attempt < max_retries: time.sleep(1 * (attempt + 1)) # Exponential backoff continue # Non-retryable error, raise immediately raise # All retries exhausted raise last_error # DeepResearchAgent - research elif agent_class_name == 'DeepResearchAgent': query = variables.get('query', action) return agent.research(query) # Default: standard Agent with chat() method else: return agent.chat(action, stream=stream) def _normalize_single_step(self, step: Any, index: int) -> 'Task': """Normalize a single step to Task. Supports: - Task objects (passed through) - Agent objects (wrapped with agent reference and tools) - Specialized agents (AudioAgent, VideoAgent, ImageAgent, OCRAgent) - Callable functions (wrapped as handler) - Strings (used as action) """ if isinstance(step, Task): return step # Check for standard Agent (has chat method) or specialized agents agent_class_name = step.__class__.__name__ is_specialized_agent = agent_class_name in ('AudioAgent', 'VideoAgent', 'ImageAgent', 'OCRAgent', 'DeepResearchAgent') if hasattr(step, 'chat') or is_specialized_agent: # It's an Agent - wrap with agent reference and preserve tools agent_tools = getattr(step, 'tools', None) # Check for _yaml_action from YAML parser (canonical: action, alias: description) yaml_action = getattr(step, '_yaml_action', None) # Check for _yaml_output_variable from YAML parser yaml_output_variable = getattr(step, '_yaml_output_variable', None) # Check for _yaml_output_json from YAML parser (structured output - Option A) yaml_output_json = getattr(step, '_yaml_output_json', None) # Check for _yaml_output_pydantic from YAML parser (structured output - Option B) yaml_output_pydantic = getattr(step, '_yaml_output_pydantic', None) # Check for _yaml_step_name from YAML parser yaml_step_name = getattr(step, '_yaml_step_name', None) # Use yaml_action if set, otherwise fall back to previous_output/input default_action = "{{previous_output}}" if index > 0 else "{{input}}" action = yaml_action if yaml_action else default_action # Build output config with all structured output options output_config = {} if yaml_output_variable: output_config["variable"] = yaml_output_variable if yaml_output_json: output_config["json_model"] = yaml_output_json if yaml_output_pydantic: output_config["pydantic_model"] = yaml_output_pydantic # Only pass output_config if it has values output_json_model = output_config.get('json_model') if output_config else None output_pydantic_model = output_config.get('pydantic_model') if output_config else None # Extract remaining YAML step-level attributes (previously dead stores) yaml_guardrail = getattr(step, '_yaml_guardrail', None) yaml_max_retries = getattr(step, '_yaml_max_retries', None) yaml_expected_output = getattr(step, '_yaml_expected_output', None) yaml_output_file = getattr(step, '_yaml_output_file', None) yaml_skip_on_failure = getattr(step, '_yaml_skip_on_failure', None) yaml_retry_delay = getattr(step, '_yaml_retry_delay', None) yaml_create_directory = getattr(step, '_yaml_create_directory', None) yaml_callback = getattr(step, '_yaml_callback', None) yaml_async_execution = getattr(step, '_yaml_async_execution', None) # Build Task kwargs — only include YAML overrides when present task_kwargs = dict( name=yaml_step_name or getattr(step, 'name', f'agent_{index+1}'), agent=step, tools=agent_tools, action=action, output_json=output_json_model, output_pydantic=output_pydantic_model, ) if yaml_guardrail is not None: task_kwargs['guardrails'] = yaml_guardrail if yaml_max_retries is not None: task_kwargs['max_retries'] = yaml_max_retries if yaml_expected_output is not None: task_kwargs['expected_output'] = yaml_expected_output if yaml_output_file is not None: task_kwargs['output_file'] = yaml_output_file if yaml_skip_on_failure is not None: task_kwargs['skip_on_failure'] = yaml_skip_on_failure if yaml_retry_delay is not None: task_kwargs['retry_delay'] = yaml_retry_delay if yaml_create_directory is not None: task_kwargs['create_directory'] = yaml_create_directory if yaml_callback is not None: task_kwargs['callback'] = yaml_callback if yaml_async_execution is not None: task_kwargs['async_execution'] = yaml_async_execution return Task(**task_kwargs) elif callable(step): return Task( name=getattr(step, '__name__', f'step_{index+1}'), handler=step ) else: return Task( name=f'step_{index+1}', action=str(step) ) def _execute_single_step_internal( self, step: Any, previous_output: Optional[str], input: str, all_variables: Dict[str, Any], model: str, verbose: bool, index: int = 0, stream: bool = True, depth: int = 0 ) -> Dict[str, Any]: """Cache wrapper. The uncached body is _execute_single_step_uncached.""" # Patterns are control flow, not deterministic leaf results. Re-enter # them on every run so their conditions and nested Task gates execute; # their individual leaves can still reuse cached results. if isinstance(step, (Loop, Parallel, Route, Repeat, Discussion, If, Include)): # Pattern executors emit their own replay step events. return self._execute_single_step_uncached( step, previous_output, input, all_variables, model, verbose, index, stream=stream, depth=depth, ) span = _replay_step_begin(self, step) if span is None: return self._execute_leaf_step( step, previous_output, input, all_variables, model, verbose, index, stream=stream, depth=depth, ) try: result = self._execute_leaf_step( step, previous_output, input, all_variables, model, verbose, index, stream=stream, depth=depth, _replay=span, ) except BaseException: _replay_step_end(span, "failed") raise is_dict = isinstance(result, dict) status = span.get("status") or ( "skipped" if is_dict and result.get("skipped") else "failed" if is_dict and result.get("error") else "completed" ) _replay_step_end(span, status, result.get("output") if is_dict else None) return result def _execute_leaf_step( self, step: Any, previous_output: Optional[str], input: str, all_variables: Dict[str, Any], model: str, verbose: bool, index: int = 0, stream: bool = True, depth: int = 0, _replay: Optional[dict] = None, ) -> Dict[str, Any]: """Leaf-step body of :meth:`_execute_single_step_internal` (gate + cache).""" if depth > MAX_NESTING_DEPTH: raise ValueError( f"Maximum nesting depth ({MAX_NESTING_DEPTH}) exceeded. " "Simplify your workflow or reduce pattern nesting." ) step = self._normalize_single_step(step, index) if step.should_run: context = WorkflowContext( input=input, previous_result=str(previous_output) if previous_output else None, current_step=step.name, variables=all_variables.copy(), ) try: should_run = step.should_run(context) except Exception as e: logger.error(f"should_run failed for {step.name}: {e}") should_run = True # Preserve the linear executor's error policy. if not should_run: step.status = "skipped" self.step_statuses[step.name] = "skipped" return { "step": step.name, "output": previous_output, "stop": False, "skipped": True, "variables": {}, } cache = getattr(self, "_step_cache", None) if cache is None: return self._execute_single_step_uncached( step, previous_output, input, all_variables, model, verbose, index, stream=stream, depth=depth, ) from .step_cache import make_step_key key = make_step_key(step, previous_output, input, all_variables) try: hit = copy.deepcopy(cache.get(key)) except Exception: hit = None logger.warning("Could not restore cached workflow step") if hit is not None: if verbose: print(f"↩︎ cache hit: {getattr(step, 'name', getattr(step, '__name__', step))}") # A cache hit means the gated-in step is being served a prior result, # so its status is "completed" -- not whatever a previous run left in # step_statuses. Without this reset, a step that was gated OUT on an # earlier run ("skipped") and is now gated IN would still report # "skipped" to callers even though its cached output is returned. if hasattr(step, "status"): step.status = "completed" self.step_statuses[step.name] = "completed" if _replay is not None: _replay["status"] = "cached" # Copy: a caller mutating a returned result must not edit the cache. return dict(hit) result = self._execute_single_step_uncached( step, previous_output, input, all_variables, model, verbose, index, stream=stream, depth=depth, ) # Only cache a SUCCESSFUL result. A nested step that exhausts its # retries or fails a guardrail returns a dict carrying an "error" key # (see the failure branches of _execute_single_step_uncached); caching # that would replay the failure on every identical re-run and never let # the step retry -- turning a transient error into a permanent one. if isinstance(result, dict) and not result.get("error"): try: cache.set(key, copy.deepcopy(result)) except Exception: logger.warning("Could not cache workflow step") return result def _execute_single_step_uncached( self, step: Any, previous_output: Optional[str], input: str, all_variables: Dict[str, Any], model: str, verbose: bool, index: int = 0, stream: bool = True, depth: int = 0 ) -> Dict[str, Any]: """Execute a single step and return result. Args: step: The step to execute (Agent, function, or pattern like Loop/Parallel/Route/If) previous_output: Output from previous step input: Original workflow input all_variables: Current workflow variables model: LLM model to use verbose: Whether to print verbose output index: Step index stream: Whether to stream responses depth: Current nesting depth (for pattern recursion safety) Returns: Dict with 'step', 'output', 'stop', 'variables' """ # Check nesting depth limit for nested patterns if depth > MAX_NESTING_DEPTH: raise ValueError( f"Maximum nesting depth ({MAX_NESTING_DEPTH}) exceeded. " "Simplify your workflow or reduce pattern nesting." ) # Handle nested patterns (Loop, Parallel, Route, Repeat, If) if isinstance(step, Loop): loop_result = self._execute_loop( step, previous_output, input, all_variables, model, verbose, stream, depth=depth+1 ) return { "step": f"loop_{index}", "output": loop_result.get("output", ""), # Propagate a nested stop request so an on_error="stop" step # inside the loop halts the enclosing workflow, not just the loop. "stop": loop_result.get("stop", False), "variables": loop_result.get("variables", all_variables) } if isinstance(step, Parallel): parallel_result = self._execute_parallel( step, previous_output, input, all_variables, model, verbose, stream, depth=depth+1 ) return { "step": f"parallel_{index}", "output": parallel_result.get("output", ""), # Propagate a nested stop request so an on_error="stop" step # inside the parallel block halts the enclosing workflow. "stop": parallel_result.get("stop", False), "variables": parallel_result.get("variables", all_variables) } if isinstance(step, Route): route_result = self._execute_route( step, previous_output, input, all_variables, model, verbose, stream, depth=depth+1 ) return { "step": f"route_{index}", "output": route_result.get("output", ""), "stop": route_result.get("stop", False), "variables": route_result.get("variables", all_variables) } if isinstance(step, Repeat): repeat_result = self._execute_repeat( step, previous_output, input, all_variables, model, verbose, stream, depth=depth+1 ) return { "step": f"repeat_{index}", "output": repeat_result.get("output", ""), "stop": repeat_result.get("stop", False), "variables": repeat_result.get("variables", all_variables) } if isinstance(step, Discussion): discussion_result = self._execute_discussion( step, previous_output, input, all_variables, model, verbose, stream, depth=depth+1 ) return { "step": f"discussion_{index}", "output": discussion_result.get("output", ""), # Propagate a nested stop request so a speaker with # on_error="stop" halts the enclosing workflow, not just the # discussion, and so a nested Discussion is not stringified into # a Task action by _normalize_single_step. "stop": discussion_result.get("stop", False), "variables": discussion_result.get("variables", all_variables) } if isinstance(step, If): if_result = self._execute_if( step, previous_output, input, all_variables, model, verbose, stream, depth=depth+1 ) return { "step": f"if_{index}", "output": if_result.get("output", ""), "stop": if_result.get("stop", False), "variables": if_result.get("variables", all_variables) } # Handle Include steps (for include inside loop) if isinstance(step, Include): include_result = self._execute_include( step, previous_output, input, all_variables, model, verbose, stream ) # Return in the expected format for single step combined_output = include_result.get("output", "") return { "step": f"include:{step.recipe}", "output": combined_output, "stop": False, "variables": include_result.get("variables", all_variables) } normalized = self._normalize_single_step(step, index) state = {"stop": False} def _run_body(validation_feedback: Optional[str]) -> Any: """Execute the step body once; used by the shared policy wrapper.""" context = WorkflowContext( input=input, previous_result=str(previous_output) if previous_output else None, current_step=normalized.name, variables=all_variables.copy() ) if validation_feedback: context.variables["validation_feedback"] = validation_feedback if normalized.handler: result = normalized.handler(context) if isinstance(result, StepResult): if result.variables: all_variables.update(result.variables) state["stop"] = result.stop_workflow return result.output return str(result) elif normalized.agent: # Propagate context management to existing agent if workflow has it enabled if self.context and hasattr(normalized.agent, '_context_manager_initialized'): if not normalized.agent._context_manager_initialized: normalized.agent._context_param = self.context # Share session deduplication cache for cross-agent deduplication if self._session_dedup_cache: normalized.agent._session_dedup_cache = self._session_dedup_cache # Also set on existing context manager if already initialized if normalized.agent._context_manager and hasattr(normalized.agent._context_manager, '_session_cache'): normalized.agent._context_manager._session_cache = self._session_dedup_cache action = normalized.action or input # Substitute variables action = _substitute_action_variables(action, all_variables, previous_output, input) if validation_feedback: action = f"{action}\n\nPrevious attempt feedback: {validation_feedback}" # Check if this is a specialized agent (AudioAgent, VideoAgent, ImageAgent, OCRAgent) agent_class_name = normalized.agent.__class__.__name__ out = self._execute_specialized_agent( normalized.agent, agent_class_name, action, normalized, all_variables, stream ) # Parse JSON output if output_json was requested step_output_json = getattr(normalized, '_output_json', None) if step_output_json and out and isinstance(out, str): out = _parse_json_output(out, normalized.name) return out elif normalized.action: from ..agent.agent import Agent config = normalized.agent_config or self.default_agent_config or {} temp_agent = Agent( name=config.get("name", normalized.name), role=config.get("role", "Assistant"), goal=config.get("goal", "Complete the task"), llm=config.get("llm", model), output=self.output, # Propagate output config context=self.context, # Propagate context management memory=self.memory, # Propagate memory config knowledge=self.knowledge, # Propagate knowledge/RAG guardrails=self.guardrails, # Propagate guardrails web=self.web, # Propagate web search/fetch reflection=self.reflection, # Propagate self-reflection autonomy=self.autonomy, # Propagate autonomy settings caching=self.caching, # Propagate caching config hooks=self.hooks, # Propagate hooks/callbacks ) action = normalized.action action = _substitute_action_variables(action, all_variables, previous_output, input) if validation_feedback: action = f"{action}\n\nPrevious attempt feedback: {validation_feedback}" out = temp_agent.chat(action, stream=stream) step_output_json = getattr(normalized, '_output_json', None) if step_output_json and out and isinstance(out, str): out = _parse_json_output(out, normalized.name) return out return None # Apply the same retry / guardrail / output_file policies used by the # top-level run() loop so nested patterns (Parallel/Loop/Route/If/Repeat) # and hierarchical mode don't silently drop these guarantees (Gap 2). output, step_error = self._apply_step_policies( normalized, _run_body, all_variables, verbose ) # Honor the step's on_error flow-control and status bookkeeping exactly # as the top-level _run_impl loop does. Previously a failing step inside a # nested pattern folded f"Error: {e}" into its output and reported the run # as "completed", so downstream steps consumed the exception text as data # and failure detection was lost. Instead: mark the run failed and, when # on_error == "stop" (the Task default), signal a stop and do NOT hand the # error string forward as this step's output. if step_error is not None: if hasattr(normalized, 'status'): normalized.status = "failed" # Suppress shared-status writes from a branch that outlived its run: a # fail_fast/timeout Parallel block abandons still-running branches, and # a late failure from one must not corrupt a *subsequent* run's status. if self._is_current_run_generation(): self.step_statuses[normalized.name] = "failed" self.status = "failed" if getattr(normalized, 'on_error', 'stop') == 'stop': if verbose: print(f"🛑 Step '{normalized.name}' failed (on_error='stop'): {step_error}") return { "step": normalized.name, "output": output if output is not None else previous_output, "stop": True, "error": str(step_error), "variables": all_variables } # on_error == "continue": the run is still marked failed above, but we # keep going. Do not pass the exception text on as real output. if verbose: print(f"⚠️ Step '{normalized.name}' failed (on_error='continue'): {step_error}") return { "step": normalized.name, "output": output, "stop": state["stop"], "error": str(step_error), "variables": all_variables } # Success: record status and write output_variable so nested patterns get # the same variable-scoping guarantee the top-level loop provides. if hasattr(normalized, 'status'): normalized.status = "completed" self.step_statuses[normalized.name] = "completed" var_name = getattr(normalized, 'output_variable', None) or f"{normalized.name}_output" all_variables[var_name] = output if verbose: print(f"✅ {normalized.name}: {str(output)}") return { "step": normalized.name, "output": output, "stop": state["stop"], "variables": all_variables } @staticmethod def _retry_backoff(step, retry_count, max_retries, error, verbose): """Sleep with exponential backoff and print retry progress. Shared by the top-level loop and ``_apply_step_policies`` so the retry-delay/print is defined exactly once. Behaviour is byte-for-byte identical to the previous inline copies: backoff is ``2**(retry_count-1)`` seconds with the same verbose progress line. """ backoff_seconds = 2 ** (retry_count - 1) if verbose: print(f"🔄 {step.name} failed (attempt {retry_count}/{max_retries}), retrying in {backoff_seconds}s: {error}") time.sleep(backoff_seconds) def _apply_step_policies(self, step, run_body, all_variables, verbose): """Run a step body with retry, guardrail-retry and output_file handling. Shared by the top-level loop and nested-pattern execution so guarded / retrying / file-writing steps behave identically wherever they appear. Args: step: The normalized step (provides max_retries/guardrails/output_file). run_body: Callable(validation_feedback) -> output. Raises on error. all_variables: Current workflow variables (for output_file substitution). verbose: Whether to print progress. Returns: Tuple ``(output, step_error)``. ``step_error`` is an Exception when the step ultimately failed (an exception that exhausted retries, or a guardrail that never validated) and ``None`` otherwise. The exception is NOT folded into ``output``: downstream steps cannot distinguish an error string from a real answer, so surfacing it lets the caller's ``on_error`` policy and the run status react exactly as ``_run_impl`` does at the top level. """ max_retries = getattr(step, 'max_retries', 3) retry_count = 0 validation_feedback = None output = None step_error = None guardrail_failed = False while retry_count <= max_retries: step_error = None try: output = run_body(validation_feedback) except Exception as e: step_error = e is_retryable = getattr(e, 'is_retryable', True) if not is_retryable: break retry_count += 1 if retry_count <= max_retries: self._retry_backoff(step, retry_count, max_retries, e, verbose) continue break # Guardrail check (guardrails canonical, guardrail deprecated) guardrail = getattr(step, 'guardrails', None) or getattr(step, 'guardrail', None) if guardrail and output and not step_error: try: is_valid, feedback = guardrail(StepResult(output=output)) if not is_valid: validation_feedback = str(feedback) retry_count += 1 if verbose: print(f"⚠️ {step.name} failed validation (attempt {retry_count}/{max_retries}): {feedback}") if retry_count <= max_retries: continue # A guardrail that never validated after exhausting retries # is a failure too, even without an exception. Surface it # so on_error flow control and status match _run_impl. guardrail_failed = True break except Exception as e: logger.error(f"Guardrail failed for {step.name}: {e}") break if step_error is None and guardrail_failed: step_error = ValueError( f"guardrail validation failed: {validation_feedback}" ) # Handle output_file - save output to file if hasattr(step, 'output_file') and step.output_file and output and not step_error: try: output_path = step.output_file for key, value in all_variables.items(): output_path = output_path.replace(f"{{{{{key}}}}}", str(value)) os.makedirs(os.path.dirname(output_path) if os.path.dirname(output_path) else ".", exist_ok=True) with open(output_path, "w") as f: f.write(str(output)) if verbose: print(f"📁 Saved output to: {output_path}") except Exception as e: logger.error(f"Failed to save output to file: {e}") return output, step_error @_replay_pattern("route") def _execute_route( self, route_step: Route, previous_output: Optional[str], input: str, all_variables: Dict[str, Any], model: str, verbose: bool, stream: bool = True, depth: int = 0 ) -> Dict[str, Any]: """Execute routing based on previous output.""" results = [] output = previous_output # Find matching route matched_route = None prev_lower = str(previous_output).lower() if previous_output else "" for key in route_step.routes: if key == "default": continue # Use word-boundary matching instead of substring containment pattern = r'\b' + re.escape(key.lower()) + r'\b' if re.search(pattern, prev_lower): matched_route = route_step.routes[key] break if matched_route is None: matched_route = route_step.default or [] _replay_note_route("default") else: _replay_note_route(key) if verbose: route_name = next((k for k in route_step.routes if route_step.routes[k] == matched_route), "default") print(f"🔀 Routing to: {route_name}") # Execute matched route steps route_stopped = False for idx, step in enumerate(matched_route): step_result = self._execute_single_step_internal( step, output, input, all_variables, model, verbose, idx, stream=stream, depth=depth+1 ) results.append({"step": step_result["step"], "output": step_result["output"]}) output = step_result["output"] all_variables.update(step_result.get("variables", {})) if step_result.get("stop"): route_stopped = True break return {"steps": results, "output": output, "variables": all_variables, "stop": route_stopped} def _llm_summarize_for_parallel( self, content: str, num_branches: int, model: str, verbose: bool = False ) -> str: """ Use LLM to create a concise summary of content for parallel distribution. This reduces token usage when the same context is passed to multiple parallel branches. """ from ..context.tokens import estimate_tokens_heuristic # Only summarize if content is large enough tokens = estimate_tokens_heuristic(content) if tokens < 1500: return content # Target summary size: ~500 tokens to leave room for each branch's work target_tokens = min(500, tokens // num_branches) try: import litellm prompt = f"""Summarize the following content concisely, preserving all key information, data, and actionable items. Keep the summary under {target_tokens * 4} characters. Focus on facts, numbers, and specific details. CONTENT: {content[:8000]} CONCISE SUMMARY:""" response = litellm.completion( model=model, messages=[{"role": "user", "content": prompt}], max_tokens=target_tokens, temperature=0.1, ) summary = response.choices[0].message.content.strip() if verbose: print(f" 🤖 LLM summarized context: {tokens:,} → {estimate_tokens_heuristic(summary):,} tokens") return summary except Exception as e: logger.debug(f"LLM summarization failed, using truncation: {e}") raise # Let caller handle fallback def _truncate_context_for_branches( self, previous_output: str, num_branches: int ) -> str: """Truncation-based context summarization for parallel branch distribution. Extracts key information: first part (context) + last part (recent output). """ max_chars = min(2500, len(previous_output) // max(num_branches, 2)) if len(previous_output) > max_chars: return previous_output[:max_chars * 2 // 3] + "\n\n[... context summarized for parallel efficiency ...]\n\n" + previous_output[-max_chars // 3:] return previous_output def _effective_workers( self, user_max: Optional[int], n: int, label: str = "Parallel" ) -> int: """Compute effective worker count, clamped by item count and default cap.""" if user_max is not None: if user_max > DEFAULT_MAX_PARALLEL_WORKERS: logger.info( f"{label} max_workers={user_max} exceeds default {DEFAULT_MAX_PARALLEL_WORKERS}. " f"Consider rate limiting if using LLM-backed agents." ) return min(user_max, n) return min(DEFAULT_MAX_PARALLEL_WORKERS, n) @staticmethod def _branch_variable_delta( branch_variables: Optional[Dict[str, Any]] ) -> Dict[str, Any]: """Return only the variables a parallel branch actually wrote. A branch runs against its *own* ``copy.deepcopy`` of the shared scope, so merging the whole returned dict back would copy every untouched key as well - replacing the parent's objects with per-branch clones and letting a stale copy overwrite a value some other part of the run had moved on from. Only the delta is merged: the keys the branch actually assigned. Writes are read from ``_WriteTrackingDict.written_keys`` - the exact set of keys the branch bound via ``variables[name] = ...`` (an ``output_variable`` write, a ``StepResult.variables`` merge, or a nested pattern's own merge). Recording the assignment itself is the only detection that is always right: * value **equality** dropped an equal-valued write (writing a variable back to the value it already held), losing declaration-order semantics on a reused ``output_variable``, and it misclassified an untouched value whose ``!=`` is not a bool (numpy arrays, some ORM/pydantic objects) as a write; * object **identity** still could not see a write whose value is an interned immutable (``x = "SAME"`` rebinds to the very object already there). Tracking the assignment sidesteps both: an untouched key is never in ``written_keys`` (so the parent keeps its own object and no ``!=`` is ever evaluated), and every real write is present regardless of its value. """ if not branch_variables: return {} written = getattr(branch_variables, "written_keys", None) if written is None: # Not a tracking dict (a branch replaced ``ctx.variables`` wholesale, or # a nested pattern returned a plain dict). Fall back to merging every # key so a real write is never dropped - the bug this method prevents. return dict(branch_variables) return {key: branch_variables[key] for key in written if key in branch_variables} @staticmethod def _loop_control_variables( loop_step: Any, item: Any, idx: int ) -> Dict[str, Any]: """The variables a ``Loop`` injects itself for one iteration. ``item`` (or the loop's ``var_name``), ``loop_index`` and the flattened ``item.`` accessors are loop *machinery*, not results the loop body produced, so they stay scoped to the loop in **both** execution modes: the sequential path restores whatever those names held before the loop, and the parallel path drops them from the merged delta. Keeping the two modes identical here is what lets the loop body's own writes - which do escape - be compared between them. """ control: Dict[str, Any] = {loop_step.var_name: item, "loop_index": idx} # Also expand nested item properties for template access (e.g., {{item.title}}) if isinstance(item, dict): for key, value in item.items(): control[f"{loop_step.var_name}.{key}"] = value return control @classmethod def _loop_variable_delta( cls, loop_variables: Optional[Dict[str, Any]], control: Dict[str, Any] ) -> Dict[str, Any]: """Writes one parallel-loop iteration made, minus the loop's control vars.""" delta = cls._branch_variable_delta(loop_variables) for key in control: delta.pop(key, None) return delta @staticmethod def _restore_loop_scope( all_variables: Dict[str, Any], saved: Dict[str, Any], introduced: set, preexisting_writes: set ) -> None: """Undo a sequential loop's own control variables once the loop is done. The sequential loop runs against the shared scope (see ``_execute_loop``), so ``item``/``loop_index`` would otherwise leak out of the loop - and, when the loop sits inside a ``Parallel`` branch, would be merged back into the enclosing workflow as if the branch had written them. Restore the previous value (or remove the name) and un-record the write. """ for key in introduced: if key in saved: all_variables[key] = saved[key] else: all_variables.pop(key, None) written = getattr(all_variables, "written_keys", None) if written is not None: for key in introduced: if key not in preexisting_writes: written.discard(key) @staticmethod def _merge_branch_variables( all_variables: Dict[str, Any], branch_deltas: List[Tuple[int, Dict[str, Any]]], warn_on_collision: bool = True ) -> None: """Merge concurrent branches' variable writes back into the shared scope. Used by ``Parallel`` (branch declaration order) and by the parallel ``Loop`` (iteration order). ``warn_on_collision`` is the only difference between them: two *branches* writing one variable is almost certainly an authoring accident worth a warning, whereas N *iterations* of one loop body necessarily write the loop body's ``output_variable`` N times - that is the normal shape of a loop, not a mistake, so the loop passes False rather than emitting a warning per item. THE MERGE RULE (please do not "simplify" this back into a race): 1. Branches never share a variables dict - each gets a deep copy - because concurrent writes to one dict is a data race. Isolation is the safe half of the design; this function is the missing other half, the merge back. 2. Deltas are applied in **branch declaration order**, never completion order, so the result does not depend on thread scheduling. Two runs of the same workflow produce the same variables. 3. On a **key collision** - two branches writing the same variable with different values - the later branch in declaration order wins, which is exactly what would happen if the same steps ran sequentially (Parallel is meant to be a faster way to run steps, not a different semantics). The collision is logged as a warning naming the key and both branches, because a silently discarded branch result is the bug this fixes; the workflow author has almost certainly reused an ``output_variable`` or a step name by accident. 4. Only real writes are merged (see ``_branch_variable_delta``), so an untouched key keeps the parent's own object. """ winners: Dict[str, int] = {} for idx, delta in branch_deltas: for key, value in delta.items(): previous_idx = winners.get(key) if previous_idx is not None: try: conflicting = bool(all_variables[key] != value) except Exception: conflicting = True if conflicting and warn_on_collision: logger.warning( f"Parallel branches {previous_idx} and {idx} both wrote " f"variable '{key}' with different values; branch {idx} wins " f"(declaration order). Give the branches distinct " f"output_variable names to avoid losing a result." ) winners[key] = idx all_variables[key] = value @_replay_pattern("parallel") def _execute_parallel( self, parallel_step: Parallel, previous_output: Optional[str], input: str, all_variables: Dict[str, Any], model: str, verbose: bool, stream: bool = True, depth: int = 0 ) -> Dict[str, Any]: """Execute steps in parallel using ThreadPoolExecutor.""" import concurrent.futures results = [] outputs = [] # Start the wall-clock bound *before* any preparation work. The context # summarisation below can make a blocking litellm.completion call, so a # deadline started only at submission time would let a Parallel(timeout=N) # block wait far longer than N before its clock even begins. timeout = getattr(parallel_step, "timeout", None) deadline = (time.monotonic() + timeout) if timeout else None if verbose: print(f"⚡ Running {len(parallel_step.steps)} steps in parallel...") # Context distribution: by default every branch sees the exact upstream # output (context="full", no data loss). Token-saving excerpting is opt-in # via context="truncate"/"summarize" because silently handing downstream # steps a smaller payload than the upstream produced is a correctness hazard. optimized_previous = previous_output num_branches = len(parallel_step.steps) context_mode = getattr(parallel_step, "context", "full") if context_mode != "full" and previous_output and len(previous_output) > 3000: from ..context.tokens import estimate_tokens_heuristic tokens = estimate_tokens_heuristic(previous_output) if tokens > 1000: if context_mode == "summarize" and num_branches >= MIN_BRANCHES_FOR_LLM_SUMMARY: # LLM summarisation only pays off for larger fan-outs; for small # branch counts the extra call costs roughly what it saves, so use # the cheaper truncation fallback instead. try: optimized_previous = self._llm_summarize_for_parallel(previous_output, num_branches, model, verbose) except Exception: # Fallback to truncation-based summarization optimized_previous = self._truncate_context_for_branches(previous_output, num_branches) else: optimized_previous = self._truncate_context_for_branches(previous_output, num_branches) if optimized_previous != previous_output: new_tokens = estimate_tokens_heuristic(optimized_previous) saved = tokens - new_tokens logger.warning( f"Parallel context='{context_mode}' reduced upstream output for " f"{num_branches} branches: {len(previous_output):,} → {len(optimized_previous):,} chars " f"({tokens:,} → {new_tokens:,} tokens). Middle content is dropped." ) if verbose: print(f" 📦 Optimized context for {num_branches} parallel branches: {tokens:,} → {new_tokens:,} tokens (saved {saved:,} per branch)") # Use ThreadPoolExecutor for parallel execution from ..trace.context_events import copy_context_to_callable, get_context_emitter branch_deltas = [] # [(branch_idx, {var: value}), ...] in declaration order # Determine effective workers based on user configuration user_max = getattr(parallel_step, 'max_workers', None) effective_workers = self._effective_workers(user_max, len(parallel_step.steps), label="Parallel") # Index-keyed collectors so branches can be consumed in *completion* # order (fail_fast must raise as soon as any branch fails, not after # every earlier-declared branch finishes) while the final merge below # still rebuilds results/outputs/branch_deltas in declaration order - # so parallel_outputs[idx] alignment and fail_all ordering are unchanged. succeeded = {} # idx -> {"result": {...}, "delta": {...}, "stop": bool} failed = {} # idx -> {"error": e, "delta": {...} | None, "stop": bool} parallel_stopped = False executor = concurrent.futures.ThreadPoolExecutor(max_workers=effective_workers) try: future_to_idx = {} for idx, step in enumerate(parallel_step.steps): # Wrap execution to propagate context and set branch_id for parallel tracking # Each branch writes into its OWN deep copy: concurrent writes to # one shared dict would be a data race. The copy is a # _WriteTrackingDict so the merge back (see _merge_branch_variables) # knows exactly which keys the branch assigned - the branch's # output_variable writes are not thrown away, and untouched keys are # not merged as per-branch clones. Keys present at construction seed # the scope and are not counted as writes. branch_vars = _WriteTrackingDict(_isolate_scope(all_variables)) def execute_with_branch(step=step, idx=idx, opt_prev=optimized_previous, branch_vars=branch_vars): emitter = get_context_emitter() emitter.set_branch(f"parallel_{idx}") try: return self._execute_single_step_internal( step, opt_prev, input, branch_vars, model, False, idx, stream, depth=depth+1 ) finally: emitter.clear_branch() future = executor.submit(copy_context_to_callable(execute_with_branch)) future_to_idx[future] = idx pending = set(future_to_idx) fail_fast_error = None # (idx, branch_error) of the first fail_fast failure while pending: remaining = None if deadline is None else max(0.0, deadline - time.monotonic()) done, pending = concurrent.futures.wait( pending, timeout=remaining, return_when=concurrent.futures.FIRST_COMPLETED, ) if not done: # Timed out with branches still running. Cancel what we can and # stop blocking the caller on threads that cannot be interrupted. for f in pending: f.cancel() self.status = "failed" raise WorkflowStepError( f"{len(pending)} parallel branch(es) timed out after {timeout}s", errors=[{"step": future_to_idx[f], "error": "timeout"} for f in pending], ) for future in done: idx = future_to_idx[future] try: step_result = future.result() except Exception as e: # A branch that raised before _apply_step_policies could catch # it (e.g. a pattern-level error). Treat like a branch failure. branch_error = e step_result = None else: # A nested step that failed now surfaces its error via the # result dict instead of raising (so on_error can be honored), # so the future resolves successfully. Detect that here so the # Parallel on_failure modes still fire instead of silently # treating the exception text as a real branch output. branch_error = step_result.get("error") branch_stop = bool(step_result and step_result.get("stop")) if branch_error is None: if step_result.get("skipped"): # A gated-out branch produced no output. Record it as # skipped and keep its index aligned, but merge no # variables and feed nothing into the combined output # (the executor returns the prior context as "output", # which must not be duplicated into sibling results). succeeded[idx] = { "result": {"step": step_result["step"], "output": None, "status": "skipped"}, "output": None, "delta": None, "stop": branch_stop, } continue succeeded[idx] = { "result": {"step": step_result["step"], "output": step_result["output"]}, "output": step_result["output"], "delta": self._branch_variable_delta(step_result.get("variables")), "stop": branch_stop, } continue logger.error(f"Parallel branch {idx} failed: {branch_error}") # The whole run has a failed branch regardless of on_failure mode. self.status = "failed" failed[idx] = { "error": branch_error, # partial_ok keeps the variables the branch's *completed* # steps wrote; fail_fast/fail_all raise, so nothing from a # failed branch is merged under those modes. "delta": self._branch_variable_delta(step_result.get("variables")) if step_result is not None else None, "stop": branch_stop, } if parallel_step.on_failure == "fail_fast": # Record the first failure, stop waiting, cancel the rest, # and raise immediately without blocking on running threads. fail_fast_error = (idx, branch_error) for f in pending: f.cancel() pending = set() break if fail_fast_error is not None: idx, branch_error = fail_fast_error # Only chain a real exception; nested steps surface the error # as a string, so guard against `raise ... from ` which # would itself raise TypeError and mask the failure. cause = branch_error if isinstance(branch_error, BaseException) else None raise WorkflowStepError( f"Parallel branch {idx} failed", cause=cause, errors=[{"step": idx, "error": branch_error}], ) from cause finally: # Do not block the caller on branches that cannot be interrupted. # shutdown(wait=True) (the `with` default) is what held the fail_fast # raise until the slowest sibling finished. executor.shutdown(wait=False) # Rebuild results/outputs/branch_deltas in declaration order so # parallel_outputs[idx] alignment and fail_all ordering are preserved # regardless of the order branches completed in. errors = [] for idx in range(len(parallel_step.steps)): if idx in succeeded: info = succeeded[idx] results.append(info["result"]) outputs.append(info["output"]) # A skipped branch carries delta=None: it wrote nothing, so it # must not contribute a merge entry (the failed path guards the # same way). if info["delta"] is not None: branch_deltas.append((idx, info["delta"])) if info["stop"]: parallel_stopped = True elif idx in failed: info = failed[idx] errors.append({"step": idx, "error": info["error"]}) # partial_ok: record the failure but keep going; fold neither the # exception text into outputs nor skip the index alignment. results.append({"step": f"parallel_{idx}", "output": None, "error": str(info["error"])}) if info["delta"] is not None: branch_deltas.append((idx, info["delta"])) outputs.append(None) if info["stop"]: parallel_stopped = True # Check if we should fail after all branches completed if errors and parallel_step.on_failure == "fail_all": first_error = errors[0]["error"] cause = first_error if isinstance(first_error, BaseException) else None raise WorkflowStepError( f"{len(errors)} parallel branches failed", errors=errors, cause=cause ) from cause # Merge each branch's variable writes back into the shared scope. Without # this every `output_variable` set inside a Parallel block was written to a # deep copy that was discarded when the branch returned (silent data loss). self._merge_branch_variables(all_variables, branch_deltas) # Combine outputs combined_output = "\n---\n".join(str(o) for o in outputs if o is not None) all_variables["parallel_outputs"] = outputs if verbose: print(f"✅ Parallel complete: {len(outputs)} results") return {"steps": results, "output": combined_output, "variables": all_variables, "stop": parallel_stopped} @_replay_pattern("loop") def _execute_loop( self, loop_step: Loop, previous_output: Optional[str], input: str, all_variables: Dict[str, Any], model: str, verbose: bool, stream: bool = True, depth: int = 0 ) -> Dict[str, Any]: """Execute step for each item in loop. When loop_step.parallel is True, executes iterations concurrently using ThreadPoolExecutor for faster processing. """ import csv import concurrent.futures results = [] outputs = [] items = [] # Tracks whether any iteration requested a workflow stop (on_error="stop") # so the signal can propagate to the enclosing workflow, not just this loop. loop_stopped = False # Get items from variable, CSV, or file if loop_step.over: items = all_variables.get(loop_step.over, []) if isinstance(items, str): # Try to parse as JSON list first (handles agent output like '["a", "b", "c"]') items = self._parse_list_from_string(items) elif loop_step.from_csv: try: with open(loop_step.from_csv, 'r', encoding='utf-8') as f: reader = csv.DictReader(f) items = list(reader) except Exception as e: logger.error(f"Failed to read CSV {loop_step.from_csv}: {e}") items = [] elif loop_step.from_file: try: with open(loop_step.from_file, 'r', encoding='utf-8') as f: items = [line.strip() for line in f if line.strip()] except Exception as e: logger.error(f"Failed to read file {loop_step.from_file}: {e}") items = [] num_items = len(items) # Determine steps to run (multi-step or single step) steps_to_run = loop_step.steps if loop_step.steps else [loop_step.step] is_multi_step = loop_step.steps is not None and len(loop_step.steps) > 1 if loop_step.parallel and num_items > 1: # Parallel execution with configurable worker limits user_max = loop_step.max_workers max_workers = self._effective_workers(user_max, num_items, label="Loop") if verbose: step_info = f" ({len(steps_to_run)} steps each)" if is_multi_step else "" print(f"⚡🔁 Parallel looping over {num_items} items{step_info} (max_workers={max_workers})...") # Context distribution: by default each iteration sees the exact upstream # output (context="full", no data loss). Token-saving excerpting is opt-in # via context="truncate". optimized_previous = previous_output context_mode = getattr(loop_step, "context", "full") if context_mode != "full" and previous_output and len(previous_output) > 3000: from ..context.tokens import estimate_tokens_heuristic tokens = estimate_tokens_heuristic(previous_output) # Target: max 800 tokens per branch to stay well under rate limits if tokens > 1000: optimized_previous = self._truncate_context_for_branches(previous_output, num_items) if optimized_previous != previous_output: new_tokens = estimate_tokens_heuristic(optimized_previous) saved = tokens - new_tokens logger.warning( f"Loop context='{context_mode}' reduced upstream output for " f"{num_items} branches: {len(previous_output):,} → {len(optimized_previous):,} chars " f"({tokens:,} → {new_tokens:,} tokens). Middle content is dropped." ) if verbose: print(f" 📦 Optimized context for {num_items} parallel branches: {tokens:,} → {new_tokens:,} tokens (saved {saved:,} per branch)") # Use copy_context_to_callable to propagate contextvars (needed for trace emission) from ..trace.context_events import copy_context_to_callable, get_context_emitter def execute_item(idx_item_tuple, opt_prev=optimized_previous): """Execute step(s) for a single item in thread.""" idx, item = idx_item_tuple # Set branch context for parallel tracking emitter = get_context_emitter() emitter.set_branch(f"loop_{idx}") try: # CRITICAL: Deep copy variables to ensure thread isolation # (concurrent iterations writing one dict is a data race). # Isolation is only half of the design: whatever the iteration # writes must still be merged back, or every output_variable # set inside the loop body is silently discarded. The copy is a # _WriteTrackingDict so the merge (see below) knows exactly # which keys this iteration assigned. The loop's own control # variables are seeded at construction time and are therefore # deliberately not counted as writes. control = self._loop_control_variables(loop_step, item, idx) seed = _isolate_scope(all_variables) seed.update(control) loop_vars = _WriteTrackingDict(seed) # Execute all steps sequentially within this iteration iteration_output = opt_prev iteration_results = [] iteration_stopped = False for step_idx, step in enumerate(steps_to_run): step_result = self._execute_single_step_internal( step, iteration_output, input, loop_vars, model, False, step_idx, stream=False, depth=depth+1 ) iteration_output = step_result.get("output") iteration_results.append(step_result) # Update variables if step set any. _execute_single_step_internal # writes into - and returns - the very dict it was handed, so # ``loop_vars.update(loop_vars)`` would record every key as # written and defeat the delta detection below. step_vars = step_result.get("variables") if step_vars and step_vars is not loop_vars: loop_vars.update(step_vars) # A failed step with on_error="stop" signals a stop; halt # this iteration instead of feeding later steps forward. if step_result.get("stop"): iteration_stopped = True break # Return final output (last step's output). Surface the stop # signal so the loop can propagate it to the outer workflow. final_result = { "step": f"loop_{idx}", "output": iteration_output, "steps": iteration_results, "stop": iteration_stopped, # Only what this iteration actually wrote, so untouched # variables keep the parent's own objects instead of being # replaced by this iteration's deep-copied clones. "variable_delta": self._loop_variable_delta(loop_vars, control), } return idx, final_result finally: emitter.clear_branch() with concurrent.futures.ThreadPoolExecutor(max_workers=max_workers) as executor: # Map each future back to its original index so error attribution # survives even if execute_item raises before returning its index. future_to_idx = {} for idx, item in enumerate(items): fut = executor.submit(copy_context_to_callable(lambda pair=(idx, item): execute_item(pair))) future_to_idx[fut] = idx # Collect results in order indexed_results = [] for future in concurrent.futures.as_completed(future_to_idx): idx = future_to_idx[future] try: _, step_result = future.result() indexed_results.append((idx, step_result)) if verbose: print(f" ✓ Item {idx + 1}/{num_items} complete") except Exception as e: logger.error(f"Parallel loop iteration {idx} failed: {e}") self.status = "failed" indexed_results.append((idx, {"step": f"loop_{idx}", "output": None, "error": str(e)})) # Sort by index to maintain order indexed_results.sort(key=lambda x: x[0]) iteration_deltas = [] # [(iteration_idx, {var: value}), ...] in item order for idx, step_result in indexed_results: results.append({"step": f"{step_result['step']}_{idx}", "output": step_result["output"]}) outputs.append(step_result["output"]) iteration_deltas.append((idx, step_result.get("variable_delta") or {})) if step_result.get("stop"): loop_stopped = True # Merge each iteration's writes back into the shared scope. Without # this every output_variable set inside the loop body was written to a # deep copy discarded when the iteration returned (silent data loss). # Deltas are applied in *item* order, never completion order, so the # surviving value does not depend on thread scheduling - it is the # last item's, exactly what the sequential loop above produces. A # cross-iteration collision is inherent to looping, so unlike Parallel # it is not warned about. self._merge_branch_variables( all_variables, iteration_deltas, warn_on_collision=False ) if verbose: print(f"✅ Parallel loop complete: {len(outputs)} results") else: # Sequential execution (original behavior) if verbose: step_info = f" ({len(steps_to_run)} steps each)" if is_multi_step else "" print(f"🔁 Looping over {num_items} items{step_info}...") # A sequential loop is simply its body unrolled, so - like Repeat, # If and Route - it runs against the shared scope. There is no # concurrency to isolate from here, and the per-iteration # ``copy.deepcopy(all_variables)`` that used to sit in this loop # bought no safety while silently discarding every output_variable # the body wrote and hiding iteration N-1's writes from iteration N. # Only the loop's own control variables stay loop-scoped; they are # saved here and restored in the finally below. control_saved: Dict[str, Any] = {} control_introduced = set() preexisting_writes = set(getattr(all_variables, "written_keys", None) or ()) prev_control_keys: set = set() try: for idx, item in enumerate(items): # Add current item (and loop_index / item.) to variables control = self._loop_control_variables(loop_step, item, idx) # Drop the previous item's flattened ``item.`` accessors # that this item does not have. Because the loop runs against # the shared scope, ``update`` alone would leave a stale # ``item.k`` visible while iterating an item that only has # ``item.m`` - the body would read the previous item's value. # ``_restore_loop_scope`` already cleans these up after the # loop; this keeps each iteration's view correct too. for stale_key in prev_control_keys - control.keys(): all_variables.pop(stale_key, None) prev_control_keys = set(control.keys()) for key in control: if key not in control_introduced: control_introduced.add(key) if key in all_variables: control_saved[key] = all_variables[key] all_variables.update(control) # Execute all steps sequentially within this iteration iteration_output = previous_output iteration_stopped = False for step_idx, step in enumerate(steps_to_run): step_result = self._execute_single_step_internal( step, iteration_output, input, all_variables, model, verbose, step_idx, stream=stream, depth=depth+1 ) iteration_output = step_result.get("output") # Update variables if step set any. The step writes into the # dict it was handed, so guard against self-update. step_vars = step_result.get("variables") if step_vars and step_vars is not all_variables: all_variables.update(step_vars) # A failed step with on_error="stop" signals a stop; halt this # iteration instead of feeding later steps the error forward. if step_result.get("stop"): iteration_stopped = True break results.append({"step": f"loop_{idx}", "output": iteration_output}) outputs.append(iteration_output) previous_output = iteration_output # Abort the remaining items too: a step that asked to stop the # workflow (on_error="stop") must not silently keep looping. if iteration_stopped: loop_stopped = True break finally: self._restore_loop_scope( all_variables, control_saved, control_introduced, preexisting_writes ) # Store outputs in user-specified variable or default to loop_outputs output_var_name = loop_step.output_variable or "loop_outputs" all_variables[output_var_name] = outputs all_variables["loop_outputs"] = outputs # Also keep for backward compatibility combined_output = "\n".join(str(o) for o in outputs) if outputs else "" # Validate outputs and warn about issues none_count = sum(1 for o in outputs if o is None) if none_count > 0: logger.warning(f"⚠️ Loop '{output_var_name}': {none_count}/{len(outputs)} outputs are None. " f"Check if agent returned expected format.") if verbose: print(f"⚠️ WARNING: {none_count}/{len(outputs)} loop outputs are None!") # Check for type consistency if outputs and len(outputs) > 0: expected_type = loop_step.step._output_json if hasattr(loop_step.step, '_output_json') else None if expected_type: expected_schema_type = expected_type.get('type') if isinstance(expected_type, dict) else None for i, o in enumerate(outputs): if o is None: continue actual_type = type(o).__name__ if expected_schema_type == 'object' and not isinstance(o, dict): logger.warning(f"⚠️ Loop output[{i}]: Expected object/dict, got {actual_type}") if verbose: print(f"⚠️ Loop output[{i}]: Expected 'object', received '{actual_type}'") elif expected_schema_type == 'array' and not isinstance(o, list): logger.warning(f"⚠️ Loop output[{i}]: Expected array/list, got {actual_type}") if verbose: print(f"⚠️ Loop output[{i}]: Expected 'array', received '{actual_type}'") # Debug logging for output_variable if verbose: print(f"📦 Loop stored {len(outputs)} results in variable: '{output_var_name}'") return {"steps": results, "output": combined_output, "variables": all_variables, "stop": loop_stopped} def _parse_list_from_string(self, text: str) -> List[Any]: """ Parse a list from a string, handling multiple formats. Supports: 1. Pure JSON array: '["a", "b", "c"]' 2. JSON array embedded in text: 'Here are topics: ["a", "b", "c"]' 3. Newline-separated items (fallback) Args: text: String that may contain a list Returns: List of items, or [text] if no list found """ import json if not text or not text.strip(): return [] text = text.strip() # 1. Try direct JSON parse (pure JSON array). # RecursionError guards against deeply-nested bracket input, which the # C/Python JSON scanner surfaces instead of JSONDecodeError (CWE-674). try: parsed = json.loads(text) if isinstance(parsed, list): return parsed except (json.JSONDecodeError, RecursionError): pass # 2. Search for an embedded list without a backtracking regex. # The previous nested-repeat pattern could take exponential time on # adversarial bracket/quote input (CWE-1333/CWE-730). JSONDecoder # validates each candidate using the JSON grammar. raw_decode is # anchored with the ``idx`` argument so we never allocate an # overlapping ``text[start:]`` copy per bracket. To bound the total # work on bracket-dense adversarial output (e.g. deeply nested "[[[[" # where each raw_decode still rescans toward the end, giving O(n^2)), # we cap both the input size and the number of decode attempts. decoder = json.JSONDecoder() max_scan_chars = 1_000_000 if len(text) > max_scan_chars: logger.warning("Loop list text exceeds the safe parsing limit") return [text] max_decode_attempts = 10_000 start = text.find('[') attempts = 0 while start != -1 and attempts < max_decode_attempts: attempts += 1 try: parsed, _ = decoder.raw_decode(text, start) except (json.JSONDecodeError, RecursionError): start = text.find('[', start + 1) continue if isinstance(parsed, list) and len(parsed) > 0: return parsed start = text.find('[', start + 1) # 3. Fallback for Python-style lists with single quotes, e.g. # "['topic1', 'topic2']", which strict JSON rejects. ast.literal_eval # only evaluates literals (no code execution) so it is safe here and # preserves the single-quoted parsing the previous regex supported. # A single attempt on the outermost bracket span keeps this linear; # LLM output places the list contiguously rather than fragmented. import ast first = text.find('[') last = text.rfind(']') if first != -1 and last > first: try: parsed = ast.literal_eval(text[first:last + 1]) if isinstance(parsed, (list, tuple)) and len(parsed) > 0: return list(parsed) except (ValueError, SyntaxError, RecursionError): pass # 4. Fallback: wrap as single item return [text] @_replay_pattern("discussion") def _execute_discussion( self, discussion, previous_output, input, all_variables, model, verbose, stream=True, depth=0 ): """Run a round-robin discussion and return the same shape as the other patterns.""" results = [] output = previous_output transcript = [] stopped = False discussion_stopped = False if verbose: names = [getattr(a, "name", getattr(a, "__name__", str(a))) for a in discussion.agents] print(f"💬 Discussion: {' → '.join(names)} for {discussion.rounds} round(s)") turn = 0 for round_index in range(discussion.rounds): for position in range(len(discussion.agents)): context = WorkflowContext( input=input, previous_result=str(output) if output else None, current_step=f"{discussion.name}_r{round_index}", variables=all_variables.copy(), ) # A custom selector may repeat or skip a speaker; round-robin is # only the default, not an assumption baked into the loop. speaker = ( discussion.select(turn, context) if discussion.select else discussion.agents[position] ) step_result = self._execute_single_step_internal( speaker, output, input, all_variables, model, verbose, turn, stream=stream, depth=depth + 1, ) output = step_result["output"] speaker_name = getattr(speaker, "name", getattr(speaker, "__name__", step_result["step"])) # The transcript is what makes this a conversation rather than N # independent answers: each speaker is handed what came before. transcript.append(f"{speaker_name}: {output}") all_variables.update(step_result.get("variables", {})) all_variables["discussion_transcript"] = "\n".join(transcript) results.append({ "step": f"{discussion.name}_r{round_index}_{speaker_name}", "output": output, }) turn += 1 # A speaker with on_error="stop" (the Task default) halts the # whole workflow, not just the discussion: honor its stop signal # rather than burning the rest of the round budget. if step_result.get("stop"): stopped = True discussion_stopped = True break if discussion.until: check = WorkflowContext( input=input, previous_result=str(output) if output else None, current_step=discussion.name, variables=all_variables.copy(), ) try: if discussion.until(check): stopped = True break except Exception as exc: # A broken until() must not silently mean "never stop": # that turns a bounded discussion into rounds of spend. raise ValueError( f"Discussion until() raised {type(exc).__name__}: {exc}. " f"It is checked after every turn, so a failing " f"condition would run the full round budget." ) from exc if stopped: break return { "steps": results, "output": output, "variables": {"discussion_transcript": "\n".join(transcript)}, "stop": discussion_stopped, } @_replay_pattern("repeat") def _execute_repeat( self, repeat_step: Repeat, previous_output: Optional[str], input: str, all_variables: Dict[str, Any], model: str, verbose: bool, stream: bool = True, depth: int = 0 ) -> Dict[str, Any]: """Repeat step until condition is met.""" results = [] output = previous_output repeat_stopped = False if verbose: print(f"🔄 Repeating up to {repeat_step.max_iterations} times...") # ``Repeat.step`` is typed ``Any`` and was handed straight to the # single-step executor, so ``repeat([a, b])`` was ACCEPTED and then # stringified into a prompt -- the list became text and the default # agent answered it. Since there is no N-agent discussion loop, that is # exactly the workaround people reach for, which made the silent # misbehaviour worse than a rejection. A list now runs its members in # order, once per iteration. steps = repeat_step.step if isinstance(repeat_step.step, (list, tuple)) else [repeat_step.step] for iteration in range(repeat_step.max_iterations): last_name = None for position, inner_step in enumerate(steps): step_result = self._execute_single_step_internal( inner_step, output, input, all_variables, model, verbose, iteration, stream=stream, depth=depth+1 ) last_name = step_result['step'] # Each member sees the previous member's output, so a two-step # repeat composes the way a two-step flow does. output = step_result["output"] all_variables.update(step_result.get("variables", {})) if len(steps) > 1: results.append({ "step": f"{last_name}_{iteration}_{position}", "output": step_result["output"], }) # A member that requests a stop must halt the remaining members # immediately; otherwise a later member's success would silently # discard the stop request and let the flow continue. if step_result.get("stop"): repeat_stopped = True break if len(steps) == 1 and last_name is not None: results.append({"step": f"{last_name}_{iteration}", "output": output}) if repeat_stopped: break # Check until condition if repeat_step.until: context = WorkflowContext( input=input, previous_result=str(output) if output else None, current_step=f"repeat_{iteration}", variables=all_variables.copy() ) try: if repeat_step.until(context): if verbose: print(f"✅ Repeat condition met at iteration {iteration + 1}") break except Exception as e: logger.error(f"Repeat until condition failed: {e}") all_variables["repeat_iterations"] = iteration + 1 if repeat_step.max_iterations > 0 else 0 return {"steps": results, "output": output, "variables": all_variables, "stop": repeat_stopped} @_replay_pattern("if") def _execute_if( self, if_step: If, previous_output: Optional[str], input: str, all_variables: Dict[str, Any], model: str, verbose: bool, stream: bool = True, depth: int = 0 ) -> Dict[str, Any]: """Execute conditional branching based on condition evaluation. Args: if_step: The If step with condition and branches previous_output: Output from previous step input: Original workflow input all_variables: Current workflow variables model: LLM model to use verbose: Verbose output stream: Enable streaming depth: Current nesting depth Returns: Dict with 'steps', 'output', and 'variables' """ results = [] output = previous_output # Evaluate the condition condition_result = self._evaluate_condition(if_step.condition, all_variables, previous_output) _replay_note_route("true" if condition_result else "false") if verbose: branch = "then" if condition_result else "else" print(f"🔀 Condition '{if_step.condition}' → {condition_result}, executing {branch} branch") # Select the appropriate branch steps_to_execute = if_step.then_steps if condition_result else if_step.else_steps # Execute the selected branch if_stopped = False for idx, step in enumerate(steps_to_execute): step_result = self._execute_single_step_internal( step, output, input, all_variables, model, verbose, idx, stream=stream, depth=depth ) results.append({"step": step_result["step"], "output": step_result["output"]}) output = step_result["output"] all_variables.update(step_result.get("variables", {})) if step_result.get("stop"): if_stopped = True break return {"steps": results, "output": output, "variables": all_variables, "stop": if_stopped} def _evaluate_condition( self, condition: str, variables: Dict[str, Any], previous_output: Optional[str] = None ) -> bool: """Evaluate a condition expression with variable substitution. This method delegates to the shared evaluate_condition() function from praisonaiagents.conditions for DRY condition handling. Supported formats: - Numeric comparison: "{{var}} > 80", "{{var}} >= 50", "{{var}} < 10" - String equality: "{{var}} == approved", "{{var}} != rejected" - Contains check: "error in {{message}}", "{{status}} contains success" - Boolean: "{{flag}}" (truthy check) - Nested property: "{{item.score}} >= 60" Args: condition: Condition expression with {{var}} placeholders variables: Current workflow variables previous_output: Output from previous step Returns: Boolean result of condition evaluation """ from praisonaiagents.conditions.evaluator import evaluate_condition return evaluate_condition(condition, variables, previous_output) @_replay_pattern("include") def _execute_include( self, include_step: Include, previous_output: Optional[str], input: str, all_variables: Dict[str, Any], model: str, verbose: bool, stream: bool = True, visited_recipes: Optional[set] = None ) -> Dict[str, Any]: """Execute an included recipe as a workflow step. Args: include_step: The Include step with recipe name previous_output: Output from previous step input: Original workflow input all_variables: Current workflow variables model: LLM model to use verbose: Verbose output stream: Enable streaming visited_recipes: Set of already-visited recipe names (for cycle detection) Returns: Dict with 'steps', 'output', and 'variables' """ recipe_name = include_step.recipe # Cycle detection if visited_recipes is None: visited_recipes = set() if recipe_name in visited_recipes: error_msg = f"Circular include detected: {recipe_name} was already included in this execution chain" logger.error(error_msg) return { "steps": [{"step": f"include:{recipe_name}", "output": f"Error: {error_msg}"}], "output": error_msg, "variables": all_variables } visited_recipes.add(recipe_name) if verbose: print(f"📦 Including recipe: {recipe_name}") try: # Try to resolve recipe path using agent_recipes if available recipe_path = None try: from agent_recipes import get_template_path recipe_path = get_template_path(recipe_name) except ImportError: # agent_recipes not installed, try local path from pathlib import Path if Path(recipe_name).exists(): recipe_path = Path(recipe_name) else: # Try as relative path to current working directory import os cwd = Path(os.getcwd()) potential_path = cwd / recipe_name if potential_path.exists(): recipe_path = potential_path if not recipe_path: error_msg = f"Recipe not found: {recipe_name}" logger.error(error_msg) return { "steps": [{"step": f"include:{recipe_name}", "output": f"Error: {error_msg}"}], "output": error_msg, "variables": all_variables } # Find the YAML file (agents.yaml or workflow.yaml) recipe_yaml = None from pathlib import Path recipe_path = Path(recipe_path) for yaml_name in ["agents.yaml", "workflow.yaml", "TEMPLATE.yaml"]: yaml_path = recipe_path / yaml_name if yaml_path.exists(): recipe_yaml = yaml_path break if not recipe_yaml: error_msg = f"No workflow YAML found in {recipe_path}" logger.error(error_msg) return { "steps": [{"step": f"include:{recipe_name}", "output": f"Error: {error_msg}"}], "output": error_msg, "variables": all_variables } # Load tools from the recipe's tools.py if present (opt-in only) tool_registry = {} tools_snapshot = None tools_py = recipe_path / "tools.py" if tools_py.exists(): try: import os _local_ok = os.environ.get("PRAISONAI_ALLOW_LOCAL_TOOLS", "").strip().lower() in ( "1", "true", "yes", "on" ) _template_ok = os.environ.get("PRAISONAI_ALLOW_TEMPLATE_TOOLS", "").strip().lower() in ( "1", "true", "yes", "on" ) if _local_ok or _template_ok: import importlib.util spec = importlib.util.spec_from_file_location("recipe_tools", tools_py) recipe_module = importlib.util.module_from_spec(spec) # Execute and cache the same source snapshot, bypassing stale pyc files. tools_content = tools_py.read_bytes() exec(compile(tools_content, str(tools_py), "exec"), recipe_module.__dict__) tools_snapshot = (str(tools_py.resolve()), tools_content) else: logger.warning( "Skipping included recipe tools.py: set PRAISONAI_ALLOW_LOCAL_TOOLS " "or PRAISONAI_ALLOW_TEMPLATE_TOOLS to enable." ) recipe_module = None if recipe_module is not None: # Extract callable tools for name in dir(recipe_module): if not name.startswith("_"): obj = getattr(recipe_module, name) if callable(obj): tool_registry[name] = obj except Exception as e: logger.warning(f"Failed to load tools from {tools_py}: {e}") # Determine input for the included recipe recipe_input = include_step.input or previous_output or input # Convert non-string inputs to JSON string for variable substitution if recipe_input and not isinstance(recipe_input, str): recipe_input = json.dumps(recipe_input) # Substitute variables in the input (only if it's a string) if recipe_input and isinstance(recipe_input, str): for key, value in all_variables.items(): recipe_input = recipe_input.replace(f"{{{{{key}}}}}", str(value)) if previous_output: prev_str = previous_output if isinstance(previous_output, str) else json.dumps(previous_output) recipe_input = recipe_input.replace("{{previous_output}}", prev_str) recipe_input = recipe_input.replace("{{input}}", input) # Parse and execute the included workflow from .yaml_parser import YAMLWorkflowParser parser = YAMLWorkflowParser(tool_registry=tool_registry) # Match parse_file's decoding policy, but read only once: the cache # namespace must describe the same snapshot that the parser saw. with open(recipe_yaml, 'r') as recipe_file: recipe_content = recipe_file.read() included_workflow = parser.parse_string(recipe_content) parent_cache = getattr(self, "_step_cache", None) if parent_cache is not None: from .step_cache import _ScopedStepCache included_workflow._step_cache = _ScopedStepCache( parent_cache, f"{recipe_yaml.resolve()}:{recipe_content}:{tools_snapshot!r}" ) # Merge parent variables into included workflow included_workflow.variables.update(all_variables) included_workflow.variables["previous_output"] = previous_output # Execute the included workflow result = included_workflow.run( input=recipe_input or "", llm=model, verbose=verbose, stream=stream ) if verbose: print(f"✅ Included recipe {recipe_name} completed") # Return result in expected format return { "steps": result.get("steps", []), "output": result.get("output", ""), "variables": {**all_variables, **result.get("variables", {})} } except Exception as e: logger.error(f"Failed to execute included recipe {recipe_name}: {e}") return { "steps": [{"step": f"include:{recipe_name}", "output": f"Error: {e}"}], "output": f"Error executing {recipe_name}: {e}", "variables": all_variables } finally: # Remove from visited after execution completes (allow re-use in different branches) visited_recipes.discard(recipe_name) def start(self, input: str = "", **kwargs) -> Dict[str, Any]: """Alias for run() for consistency with Agents.""" return self.run(input, **kwargs) # Backward compatibility aliases (silent - no deprecation warnings) # AgentFlow is the primary name (v1.0+) # Workflow and Pipeline are silent aliases Workflow = AgentFlow Pipeline = AgentFlow class WorkflowManager: """ Manages workflow discovery, loading, and execution. Workflows are defined in markdown files with YAML frontmatter and step definitions. Example workflow file (.praisonai/workflows/deploy.md): ```markdown --- name: Deploy to Production description: Deploy the application to production variables: environment: production branch: main --- ## Step 1: Run Tests Run all tests to ensure code quality. ```action Run the test suite with pytest ``` ## Step 2: Build Build the application for production. ```action Build the application with production settings ``` ## Step 3: Deploy Deploy to the production server. ```action Deploy to {{environment}} from {{branch}} ``` ``` Example usage: ```python manager = WorkflowManager(workspace_path="/path/to/project") # List available workflows workflows = manager.list_workflows() # Get a specific workflow deploy = manager.get_workflow("deploy") # Execute a workflow results = manager.execute( "deploy", executor=lambda prompt: agent.chat(prompt), variables={"branch": "release-1.0"} ) ``` """ from ..paths import DEFAULT_DIR_NAME as _DIR_NAME WORKFLOWS_DIR = f"{_DIR_NAME}/workflows" # Only markdown is parsed by this manager; YAML workflows are handled by # YAMLWorkflowParser. A discovered YAML file fails loudly in _load_workflow. SUPPORTED_EXTENSIONS = [".md"] def __init__( self, workspace_path: Optional[str] = None, verbose: int = 0 ): """ Initialize WorkflowManager. Args: workspace_path: Path to workspace/project root verbose: Verbosity level """ self.workspace_path = Path(workspace_path) if workspace_path else Path.cwd() self.verbose = verbose self._workflows: Dict[str, Workflow] = {} self._loaded = False def _log(self, msg: str, level: int = logging.INFO): """Log message if verbose.""" if self.verbose >= 1: logger.log(level, msg) def _ensure_loaded(self): """Lazy load workflows on first access.""" if not self._loaded: self._discover_workflows() self._loaded = True def _parse_frontmatter(self, content: str) -> tuple[Dict[str, Any], str]: """Parse YAML frontmatter from markdown content.""" frontmatter = {} body = content if content.startswith("---"): parts = content.split("---", 2) if len(parts) >= 3: yaml_content = parts[1].strip() body = parts[2].strip() # Simple YAML parsing for line in yaml_content.split("\n"): line = line.strip() if ":" in line and not line.startswith("#"): key, value = line.split(":", 1) key = key.strip() value = value.strip() # Handle basic types if value.lower() in ("true", "false"): value = value.lower() == "true" elif value.isdigit(): value = int(value) elif value.startswith('"') and value.endswith('"'): value = value[1:-1] frontmatter[key] = value return frontmatter, body def _parse_steps(self, body: str) -> List[Task]: """Parse workflow steps from markdown body.""" steps = [] # Split by ## headers step_pattern = re.compile(r'^##\s+(?:Step\s+\d+[:\s]*)?(.+)$', re.MULTILINE) action_pattern = re.compile(r'```action\s*\n(.*?)\n```', re.DOTALL) condition_pattern = re.compile(r'```condition\s*\n(.*?)\n```', re.DOTALL) agent_pattern = re.compile(r'```agent\s*\n(.*?)\n```', re.DOTALL) tools_pattern = re.compile(r'```tools\s*\n(.*?)\n```', re.DOTALL) context_from_pattern = re.compile(r'context_from:\s*\[([^\]]+)\]', re.IGNORECASE) retain_context_pattern = re.compile(r'retain_full_context:\s*(true|false)', re.IGNORECASE) output_var_pattern = re.compile(r'output_variable:\s*(\w+)', re.IGNORECASE) next_steps_pattern = re.compile(r'next_steps:\s*\[([^\]]+)\]', re.IGNORECASE) loop_over_pattern = re.compile(r'loop_over:\s*(\w+)', re.IGNORECASE) loop_var_pattern = re.compile(r'loop_var:\s*(\w+)', re.IGNORECASE) branch_pattern = re.compile(r'```branch\s*\n(.*?)\n```', re.DOTALL) # New patterns for route, parallel, images, output_file route_pattern = re.compile(r'```route\s*\n(.*?)\n```', re.DOTALL) parallel_pattern = re.compile(r'```parallel\s*\n(.*?)\n```', re.DOTALL) images_pattern = re.compile(r'```images\s*\n(.*?)\n```', re.DOTALL) output_file_pattern = re.compile(r'output_file:\s*(.+)', re.IGNORECASE) repeat_pattern = re.compile(r'```repeat\s*\n(.*?)\n```', re.DOTALL) # Find all step headers matches = list(step_pattern.finditer(body)) for i, match in enumerate(matches): step_name = match.group(1).strip() # Get content until next step or end start = match.end() end = matches[i + 1].start() if i + 1 < len(matches) else len(body) step_content = body[start:end].strip() # Extract action action_match = action_pattern.search(step_content) action = action_match.group(1).strip() if action_match else "" # Extract condition condition_match = condition_pattern.search(step_content) condition = condition_match.group(1).strip() if condition_match else None # Extract agent config agent_config = None agent_match = agent_pattern.search(step_content) if agent_match: agent_config = self._parse_agent_config(agent_match.group(1).strip()) # Extract tools list tools = None tools_match = tools_pattern.search(step_content) if tools_match: tools_str = tools_match.group(1).strip() # Parse tools as list of strings (tool names) tools = [t.strip().strip('"\'') for t in tools_str.split('\n') if t.strip()] # Extract context_from context_from = None context_from_match = context_from_pattern.search(step_content) if context_from_match: context_from = [s.strip().strip('"\'') for s in context_from_match.group(1).split(',')] # Extract retain_full_context retain_full_context = True retain_match = retain_context_pattern.search(step_content) if retain_match: retain_full_context = retain_match.group(1).lower() == 'true' # Extract output_variable output_variable = None output_var_match = output_var_pattern.search(step_content) if output_var_match: output_variable = output_var_match.group(1) # Extract next_steps for branching next_steps = None next_steps_match = next_steps_pattern.search(step_content) if next_steps_match: next_steps = [s.strip().strip('"\'') for s in next_steps_match.group(1).split(',')] # Extract branch_condition branch_condition = None branch_match = branch_pattern.search(step_content) if branch_match: branch_condition = self._parse_branch_condition(branch_match.group(1).strip()) # Extract loop_over loop_over = None loop_over_match = loop_over_pattern.search(step_content) if loop_over_match: loop_over = loop_over_match.group(1) # Extract loop_var loop_var = "item" loop_var_match = loop_var_pattern.search(step_content) if loop_var_match: loop_var = loop_var_match.group(1) # Extract route pattern (same as branch_condition) route_match = route_pattern.search(step_content) if route_match and not branch_condition: branch_condition = self._parse_branch_condition(route_match.group(1).strip()) # Extract parallel pattern parallel_steps = None parallel_match = parallel_pattern.search(step_content) if parallel_match: parallel_str = parallel_match.group(1).strip() # Parse as list of step names parallel_steps = [s.strip().lstrip('- ').strip('"\'') for s in parallel_str.split('\n') if s.strip()] # Set next_steps to parallel steps if not already set if not next_steps: next_steps = parallel_steps # Extract images images = None images_match = images_pattern.search(step_content) if images_match: images_str = images_match.group(1).strip() images = [s.strip() for s in images_str.split('\n') if s.strip()] # Extract output_file output_file = None output_file_match = output_file_pattern.search(step_content) if output_file_match: output_file = output_file_match.group(1).strip() # Extract repeat config repeat_config = None repeat_match = repeat_pattern.search(step_content) if repeat_match: repeat_config = self._parse_agent_config(repeat_match.group(1).strip()) # Get description (text before action block) description = step_content if action_match: description = step_content[:action_match.start()].strip() # Clean up description (remove code blocks and config lines) description = re.sub(r'```.*?```', '', description, flags=re.DOTALL) description = re.sub(r'context_from:.*', '', description) description = re.sub(r'retain_full_context:.*', '', description) description = re.sub(r'output_variable:.*', '', description) description = re.sub(r'next_steps:.*', '', description) description = re.sub(r'loop_over:.*', '', description) description = re.sub(r'loop_var:.*', '', description) description = re.sub(r'output_file:.*', '', description) description = description.strip() # Set max_retries from repeat config max_retries = 3 if repeat_config and 'max_iterations' in repeat_config: max_retries = int(repeat_config['max_iterations']) # Build consolidated context config if needed context_config = None if context_from or not retain_full_context: # Only create config if context_from is set OR retain_full_context is False (non-default) context_config = TaskContextConfig( from_steps=context_from, retain_full=retain_full_context ) # Build consolidated output config if needed output_config = None if output_variable or output_file: output_config = TaskOutputConfig( variable=output_variable, file=output_file ) # Build consolidated execution config if needed execution_config = None if max_retries != 3: # Only if non-default execution_config = TaskExecutionConfig( max_retries=max_retries ) # Build consolidated routing config if needed routing_config = None if next_steps or branch_condition: routing_config = TaskRoutingConfig( next_steps=next_steps, branches=branch_condition ) steps.append(Task( name=step_name, description=description, action=action, condition=condition, agent_config=agent_config, tools=tools, context=context_config, output=output_config, execution=execution_config, routing=routing_config, loop_over=loop_over, loop_var=loop_var, images=images )) return steps def _parse_branch_condition(self, branch_str: str) -> Dict[str, List[str]]: """Parse branch condition from a code block.""" condition = {} for line in branch_str.split('\n'): line = line.strip() if ':' in line and not line.startswith('#'): key, value = line.split(':', 1) key = key.strip() value = value.strip() # Parse as list if value.startswith('[') and value.endswith(']'): value = [s.strip().strip('"\'') for s in value[1:-1].split(',')] else: value = [value.strip('"\'')] condition[key] = value return condition def _parse_agent_config(self, agent_str: str) -> Dict[str, Any]: """Parse agent configuration from a code block.""" config = {} for line in agent_str.split('\n'): line = line.strip() if ':' in line and not line.startswith('#'): key, value = line.split(':', 1) key = key.strip() value = value.strip() # Handle basic types if value.lower() in ('true', 'false'): value = value.lower() == 'true' elif value.isdigit(): value = int(value) elif value.startswith('"') and value.endswith('"'): value = value[1:-1] elif value.startswith("'") and value.endswith("'"): value = value[1:-1] config[key] = value return config def _load_workflow(self, file_path: Path) -> Optional[Workflow]: """Load a workflow from a file.""" # The markdown parser below cannot read YAML workflows; loading a # ``.yaml``/``.yml`` here would silently produce zero steps. Fail loudly # so the mismatch is visible instead of running an empty workflow. if file_path.suffix.lower() in (".yaml", ".yml"): logger.error( "Workflow file '%s' is YAML, but WorkflowManager only parses " "markdown workflows. Use YAMLWorkflowParser / 'praisonai workflow " "run' for YAML workflows.", file_path, ) return None try: content = file_path.read_text(encoding="utf-8") frontmatter, body = self._parse_frontmatter(content) name = frontmatter.get("name", file_path.stem) description = frontmatter.get("description", "") variables = frontmatter.get("variables", {}) # Parse variables if it's a string (simple YAML) if isinstance(variables, str): variables = {} # Parse workflow-level configuration (llm preferred, default_llm for backward compat) llm = frontmatter.get("llm") or frontmatter.get("default_llm") planning_enabled = frontmatter.get("planning", False) planning_llm = frontmatter.get("planning_llm") memory_config = frontmatter.get("memory_config") default_agent_config = frontmatter.get("default_agent_config") steps = self._parse_steps(body) # Build consolidated planning config if needed planning_config = None if planning_enabled or planning_llm: planning_config = WorkflowPlanningConfig( enabled=planning_enabled, llm=planning_llm ) # Build consolidated memory config if needed memory_cfg = None if memory_config: if isinstance(memory_config, dict): memory_cfg = WorkflowMemoryConfig(**memory_config) else: memory_cfg = memory_config workflow = Workflow( name=name, description=description, steps=steps, variables=variables, file_path=str(file_path), llm=llm, planning=planning_config, memory=memory_cfg, default_agent_config=default_agent_config ) self._log(f"Loaded workflow '{name}' with {len(steps)} steps") return workflow except Exception as e: self._log(f"Error loading workflow {file_path}: {e}", logging.WARNING) return None def _discover_workflows(self): """Discover all workflows in the workspace.""" self._workflows = {} workflows_dir = self.workspace_path / self.WORKFLOWS_DIR.replace("/", os.sep) if not workflows_dir.exists(): return for ext in self.SUPPORTED_EXTENSIONS: for file_path in workflows_dir.glob(f"*{ext}"): workflow = self._load_workflow(file_path) if workflow: self._workflows[workflow.name.lower()] = workflow self._log(f"Discovered {len(self._workflows)} workflows") def reload(self): """Reload all workflows from disk.""" self._loaded = False self._ensure_loaded() def list_workflows(self) -> List[Workflow]: """List all available workflows.""" self._ensure_loaded() return list(self._workflows.values()) def get_workflow(self, name: str) -> Optional[Workflow]: """Get a workflow by name.""" self._ensure_loaded() return self._workflows.get(name.lower()) def _substitute_variables( self, text: str, variables: Dict[str, Any] ) -> str: """ Substitute {{variable}} placeholders in text. Delegates to shared substitute_variables() utility for DRY compliance. Resolution order: 1. Dynamic variable providers ({{now}}, {{today}}, {{uuid}}, etc.) 2. Static variables from workflow variables dict 3. Keep original placeholder if not found """ from praisonaiagents.utils.variables import substitute_variables return substitute_variables(text, variables) def _build_step_context( self, step: Task, step_index: int, results: List[Dict[str, Any]], variables: Dict[str, Any] ) -> str: """ Build context for a step from previous step outputs. Args: step: Current workflow step step_index: Index of current step results: Results from previous steps variables: Current variables dict Returns: Context string to prepend to action """ if step_index == 0 or not results: return "" context_parts = [] # Use getattr with defaults for workflow-specific attributes that may not exist on Task context_from = getattr(step, 'context_from', None) retain_full_context = getattr(step, 'retain_full_context', False) if context_from: # Include only specified steps for step_name in context_from: for prev_result in results: if prev_result["step"] == step_name and prev_result.get("output"): context_parts.append(f"## {step_name} Output:\n{prev_result['output']}") elif retain_full_context: # Include all previous outputs for prev_result in results: if prev_result.get("output") and prev_result.get("status") == "success": context_parts.append(f"## {prev_result['step']} Output:\n{prev_result['output']}") else: # Include only the last step's output last_result = results[-1] if last_result.get("output") and last_result.get("status") == "success": context_parts.append(f"## Previous Output:\n{last_result['output']}") if context_parts: return "# Context from previous steps:\n\n" + "\n\n".join(context_parts) + "\n\n" return "" def _update_variables_with_output( self, step: Task, output: str, variables: Dict[str, Any], results: List[Dict[str, Any]] ) -> None: """ Update variables dict with step output for variable substitution. Args: step: Current workflow step output: Step output variables: Variables dict to update results: Results list for previous_output """ # Store output in custom variable name if specified if step.output_variable: variables[step.output_variable] = output # Store output with step name (normalized) step_var_name = f"{step.name.lower().replace(' ', '_')}_output" variables[step_var_name] = output # Update previous_output for next step variables["previous_output"] = output def load_yaml( self, file_path: Union[str, Path], tool_registry: Optional[Dict[str, Callable]] = None ) -> "Workflow": """ Load a workflow from a YAML file. Args: file_path: Path to the YAML workflow file tool_registry: Optional dictionary mapping tool names to callable functions Returns: Workflow object ready for execution Example: ```python manager = WorkflowManager() workflow = manager.load_yaml("research_workflow.yaml") result = workflow.start("Research AI trends") ``` """ from .yaml_parser import YAMLWorkflowParser parser = YAMLWorkflowParser(tool_registry=tool_registry) workflow = parser.parse_file(file_path) # Store in manager's workflow cache workflow_name = workflow.name.lower().replace(' ', '_') self._workflows[workflow_name] = workflow return workflow def execute_yaml( self, file_path: Union[str, Path], input_data: Optional[str] = None, variables: Optional[Dict[str, Any]] = None, tool_registry: Optional[Dict[str, Callable]] = None, verbose: bool = False ) -> Dict[str, Any]: """ Load and execute a YAML workflow file. Args: file_path: Path to the YAML workflow file input_data: Initial input for the workflow variables: Additional variables to merge with workflow variables tool_registry: Optional dictionary mapping tool names to callable functions verbose: Enable verbose output Returns: Workflow execution results Example: ```python manager = WorkflowManager() result = manager.execute_yaml( "research_workflow.yaml", input_data="Research AI trends", variables={"topic": "Machine Learning"} ) print(result["output"]) ``` """ workflow = self.load_yaml(file_path, tool_registry=tool_registry) # Merge additional variables if variables: workflow.variables.update(variables) # Set verbose if specified if verbose: workflow.verbose = verbose # Execute the workflow return workflow.start(input_data or "") async def aexecute_yaml( self, file_path: Union[str, Path], input_data: Optional[str] = None, variables: Optional[Dict[str, Any]] = None, tool_registry: Optional[Dict[str, Callable]] = None, verbose: bool = False ) -> Dict[str, Any]: """ Async version of execute_yaml. Args: file_path: Path to the YAML workflow file input_data: Initial input for the workflow variables: Additional variables to merge with workflow variables tool_registry: Optional dictionary mapping tool names to callable functions verbose: Enable verbose output Returns: Workflow execution results """ workflow = self.load_yaml(file_path, tool_registry=tool_registry) # Merge additional variables if variables: workflow.variables.update(variables) # Set verbose if specified if verbose: workflow.verbose = verbose # Execute the workflow asynchronously return await workflow.astart(input_data or "") def execute( self, workflow_name: str, executor: Optional[Callable[[str], str]] = None, variables: Optional[Dict[str, Any]] = None, on_step: Optional[Callable[[Task, int], None]] = None, on_result: Optional[Callable[[Task, str], None]] = None, default_agent: Optional[Any] = None, default_llm: Optional[str] = None, memory: Optional[Any] = None, planning: bool = False, stream: bool = False, verbose: int = 0, checkpoint: Optional[str] = None, resume: Optional[str] = None, rebase_checkpoint: bool = False ) -> Dict[str, Any]: """ Execute a workflow with context passing between steps. Args: workflow_name: Name of the workflow to execute executor: Function to execute each step (e.g., agent.chat). Optional if default_agent provided. variables: Variables to substitute in steps on_step: Callback before each step (step, index) on_result: Callback after each step (step, result) default_agent: Default agent to use for steps without agent_config default_llm: Default LLM model for agent creation memory: Shared memory instance planning: Enable planning mode stream: Enable streaming output verbose: Verbosity level checkpoint: Save checkpoint after each step with this name resume: Resume from checkpoint with this name rebase_checkpoint: Force resume at the same step index even when the workflow definition changed since the checkpoint (skips the fingerprint check with a warning) Returns: Execution results with step outputs and status """ loop = self._prepare_workflow_loop( workflow_name=workflow_name, variables=variables, default_llm=default_llm, planning=planning, resume=resume, rebase_checkpoint=rebase_checkpoint, ) if "error" in loop: return loop["error"] state = loop["state"] # Drive the shared step-loop generator with a synchronous step-runner. step_iter = self._run_workflow_loop( workflow=loop["workflow"], workflow_name=workflow_name, state=state, checkpoint=checkpoint, on_step=on_step, ) request = next(step_iter, None) while request is not None: step, step_idx = request step_result = self._execute_single_step( step=step, step_idx=step_idx, results=state["results"], all_variables=state["all_variables"], executor=executor, default_agent=default_agent, default_llm=loop["default_llm"], memory=memory, planning=loop["planning"], on_step=None, # loop handles the on_step callback on_result=on_result, ) try: request = step_iter.send(step_result) except StopIteration: break return self._workflow_loop_result(loop["workflow"], state) def _prepare_workflow_loop( self, workflow_name: str, variables: Optional[Dict[str, Any]], default_llm: Optional[str], planning: bool, resume: Optional[str], rebase_checkpoint: bool = False, ) -> Dict[str, Any]: """Resolve the workflow and build the initial loop state. Shared by the sync ``execute`` and async ``aexecute`` drivers so the setup (framework validation, variable merge, checkpoint resume) lives in a single place. Returns ``{"error": }`` on failure. On resume the stored ``definition_fingerprint`` is compared against the current workflow. A mismatch means the file was edited between save and resume, so the numeric step index no longer points at the same step; this raises unless ``rebase_checkpoint`` is set to deliberately continue at the same index against the edited definition. """ workflow = self.get_workflow(workflow_name) if not workflow: return { "error": { "success": False, "error": f"Workflow '{workflow_name}' not found", "results": [] } } # Fail fast on unsupported frameworks. This path iterates steps # directly instead of calling Workflow.run(), so re-apply the same # guard here to avoid silently ignoring framework: . workflow._validate_framework() # Merge variables all_variables = {**workflow.variables} if variables: all_variables.update(variables) # Use workflow-level defaults if not provided if default_llm is None: default_llm = workflow.llm if planning is False and workflow.planning: planning = workflow.planning results: List[Dict[str, Any]] = [] start_step = 0 resumed_from_step = None fingerprint = self._definition_fingerprint(workflow) # Resume from checkpoint if specified if resume: checkpoint_data = self._load_checkpoint(resume) if not checkpoint_data: # Fail closed: an explicit resume onto an absent/misspelled/ # deleted checkpoint must not silently re-run from step 1 and # repeat already-completed side effects. return { "error": { "success": False, "error": ( f"checkpoint '{resume}' not found; refusing to resume. " "Run `workflow checkpoints` to list saved checkpoints, " "or re-run without --resume to start fresh." ), "results": [], } } if checkpoint_data: saved_fingerprint = checkpoint_data.get("definition_fingerprint") if ( saved_fingerprint and saved_fingerprint != fingerprint and not rebase_checkpoint ): return { "error": { "success": False, "error": ( "workflow definition changed since the checkpoint " f"(fingerprint mismatch: checkpoint={saved_fingerprint}, " f"current={fingerprint}). Re-run without --resume to start " "fresh, or pass --rebase-checkpoint to deliberately continue " "at the same step index against the edited definition." ), "results": [], } } if ( saved_fingerprint and saved_fingerprint != fingerprint and rebase_checkpoint ): self._log( "Warning: workflow definition changed since the checkpoint " f"(checkpoint={saved_fingerprint}, current={fingerprint}); " "continuing at the same step index because --rebase-checkpoint was set." ) results = checkpoint_data.get("results", []) # Layer, do not replace. The checkpoint's variables used to # overwrite the caller's outright, so a value supplied ON the # resume was accepted and silently thrown away: # # praisonai workflow run wf --resume --var reviewer_decision=approved # # which is precisely how a human hands their answer back to a # run that paused for review. Precedence is workflow defaults, # then the checkpoint, then whatever the caller supplied now -- # the freshest statement of intent wins. all_variables = { **all_variables, **(checkpoint_data.get("variables") or {}), **(variables or {}), } start_step = checkpoint_data.get("completed_steps", 0) resumed_from_step = start_step self._log(f"Resuming workflow from step {start_step + 1}") return { "workflow": workflow, "default_llm": default_llm, "planning": planning, "state": { "results": results, "all_variables": all_variables, "success": True, "start_step": start_step, "resumed_from_step": resumed_from_step, "definition_fingerprint": fingerprint, }, } def _workflow_loop_result(self, workflow: Workflow, state: Dict[str, Any]) -> Dict[str, Any]: """Assemble the return dict shared by ``execute`` and ``aexecute``.""" result = { "success": state["success"], "workflow": workflow.name, "results": state["results"], "variables": state["all_variables"], } # Surface resume provenance so a resumed run is visible after the fact. if state.get("resumed_from_step") is not None: result["resumed_from_step"] = state["resumed_from_step"] return result def _run_workflow_loop( self, workflow: Workflow, workflow_name: str, state: Dict[str, Any], checkpoint: Optional[str] = None, on_step: Optional[Callable[[Task, int], None]] = None, ): """Single source of truth for the workflow step-loop semantics. This is a *sans-io* generator: it owns all orchestration (condition skip, ``loop_over`` iteration, checkpoint save, ``branch_condition`` / ``next_steps`` routing, early-stop and ``on_error`` handling) but never runs a step itself. For each step invocation it yields ``(step, idx)`` and expects the driver to ``send`` back the step-result dict produced by the sync or async step-runner. This keeps the loop identical for both ``execute`` and ``aexecute`` while the only sync/async seam is how a single step is run. """ results = state["results"] all_variables = state["all_variables"] # Build step lookup for branching step_lookup = {step.name: (i, step) for i, step in enumerate(workflow.steps)} current_step_idx = state["start_step"] max_iterations = len(workflow.steps) * 10 # Prevent infinite loops iteration = 0 while current_step_idx < len(workflow.steps) and iteration < max_iterations: iteration += 1 step = workflow.steps[current_step_idx] i = current_step_idx # Check condition (only string conditions are evaluable here; a # dict condition is a routing table handled by branch logic below) if step.condition and isinstance(step.condition, str): condition = self._substitute_variables(step.condition, all_variables) # Simple condition evaluation (could be enhanced) if condition.lower() in ("false", "no", "skip", "0"): results.append({ "step": step.name, "status": "skipped", "output": None }) current_step_idx += 1 continue # Handle loop_over - iterate over a variable if step.loop_over and step.loop_over in all_variables: loop_items = all_variables[step.loop_over] loop_stopped = False if isinstance(loop_items, (list, tuple)): loop_results = [] for item_idx, item in enumerate(loop_items): # Set loop variable all_variables[step.loop_var] = item all_variables["_loop_index"] = item_idx # Callback before step if on_step: on_step(step, i) # Execute step for this item via the driver-provided runner step_result = yield (step, i) loop_results.append(step_result) if not step_result["success"] and step.on_error == "stop": state["success"] = False loop_stopped = True break # Store all loop results results.append({ "step": step.name, "status": "success" if all(r["success"] for r in loop_results) else "partial", "output": [r["output"] for r in loop_results], "loop_results": loop_results }) # Clean up loop variables all_variables.pop(step.loop_var, None) all_variables.pop("_loop_index", None) else: self._log(f"loop_over variable '{step.loop_over}' is not iterable") # A stop-on-error inside the loop aborts the whole workflow, # not just the inner item iteration. if loop_stopped: break # Persist checkpoint after a completed loop step so resume does # not re-run every loop item (parity with single-step saves). if checkpoint: self._save_checkpoint( name=checkpoint, workflow_name=workflow_name, completed_steps=i + 1, results=results, variables=all_variables, definition_fingerprint=state.get("definition_fingerprint") ) current_step_idx += 1 continue # Callback before step if on_step: on_step(step, i) # Execute single step via the driver-provided runner step_result = yield (step, i) # Handle skipped steps if step_result.get("skipped"): results.append({ "step": step.name, "status": "skipped", "output": None }) current_step_idx += 1 continue results.append({ "step": step.name, "status": "success" if step_result["success"] else "failed", "output": step_result["output"], "error": step_result.get("error") }) # Handle early stop if step_result.get("stop"): self._log(f"Workflow stopped early at step '{step.name}'") break # Save checkpoint after each step if enabled if checkpoint: self._save_checkpoint( name=checkpoint, workflow_name=workflow_name, completed_steps=i + 1, results=results, variables=all_variables, definition_fingerprint=state.get("definition_fingerprint") ) # Handle failure if not step_result["success"]: if step.on_error == "stop": state["success"] = False break elif step.on_error == "continue": current_step_idx += 1 continue # Handle branching next_step_idx = None # Use getattr with defaults for workflow-specific attributes that may not exist on Task branch_condition = getattr(step, 'branch_condition', None) next_steps = getattr(step, 'next_steps', None) # Unified condition syntax: Task(when=..., then_task=..., else_task=...) # redirects execution by step name based on the evaluated condition. if getattr(step, 'when', None) is not None: when_context = dict(all_variables) when_context['previous_output'] = step_result.get("output") target = step.get_next_task(when_context) if target and target in step_lookup: current_step_idx, _ = step_lookup[target] continue if branch_condition and step_result["output"]: # Evaluate branch condition based on output output_lower = str(step_result["output"]).lower() for branch_key, branch_targets in branch_condition.items(): if branch_key.lower() in output_lower or output_lower.startswith(branch_key.lower()): if branch_targets and branch_targets[0] in step_lookup: next_step_idx, _ = step_lookup[branch_targets[0]] break elif next_steps: # Use explicit next_steps if next_steps[0] in step_lookup: next_step_idx, _ = step_lookup[next_steps[0]] # Move to next step if next_step_idx is not None: current_step_idx = next_step_idx else: current_step_idx += 1 def _prepare_step_context( self, step: Task, results: List[Dict[str, Any]], all_variables: Dict[str, Any], original_input: str, ) -> Tuple[Optional["WorkflowContext"], Optional[Dict[str, Any]]]: """Shared prep for the sync/async single-step runners. Builds the handler ``WorkflowContext`` (only when a consumer exists) and evaluates ``should_run``. Returns ``(context, early_result)`` where ``early_result`` is the skip return-dict when ``should_run`` gates the step out, otherwise ``None`` and ``context`` is ready for handler use. """ # Get previous step output previous_output = results[-1].get("output") if results else None # Create context for step handlers (only when a consumer exists) context = None if step.should_run or step.handler: context = WorkflowContext( input=original_input, previous_result=str(previous_output) if previous_output else None, current_step=step.name, variables=all_variables.copy() ) # Check should_run condition if provided if step.should_run: try: if not step.should_run(context): return context, { "success": True, "output": None, "skipped": True, "stop": False } except Exception as e: self._log(f"should_run check for step '{step.name}' failed: {e}") return context, None def _build_step_action( self, step: Task, step_idx: int, results: List[Dict[str, Any]], all_variables: Dict[str, Any], ) -> str: """Shared prep: build step context, substitute variables, prepend context.""" # Build context from previous steps context = self._build_step_context(step, step_idx, results, all_variables) # Substitute variables in action action = self._substitute_variables(step.action, all_variables) # Prepend context to action if available if context: action = f"{context}# Current Task:\n{action}" return action def _finalize_step_result( self, step: Task, output: Optional[str], step_success: bool, error: Optional[str], all_variables: Dict[str, Any], results: List[Dict[str, Any]], on_result: Optional[Callable[[Task, str], None]], ) -> Dict[str, Any]: """Shared finalize for the sync/async single-step runners.""" # Update variables with step output if output and step_success: self._update_variables_with_output(step, output, all_variables, results) # Callback after step if on_result and output: on_result(step, output) return { "success": step_success, "output": output, "stop": False, # Normal execution doesn't request stop "error": error } def _execute_single_step( self, step: Task, step_idx: int, results: List[Dict[str, Any]], all_variables: Dict[str, Any], executor: Optional[Callable[[str], str]] = None, default_agent: Optional[Any] = None, default_llm: Optional[str] = None, memory: Optional[Any] = None, planning: bool = False, verbose: int = 0, on_step: Optional[Callable[[Task, int], None]] = None, on_result: Optional[Callable[[Task, str], None]] = None, original_input: str = "" ) -> Dict[str, Any]: """Cache wrapper for the top-level run() path. There are two step executors -- this one and _execute_single_step_internal, used inside patterns. Both are wrapped, because caching only one would make the feature work or not depending on whether the step happened to sit inside a Parallel or an If. """ cache = getattr(self, "_step_cache", None) if cache is None: return self._execute_single_step_nocache( step, step_idx, results, all_variables, executor, default_agent, default_llm, memory, planning, verbose, on_step, on_result, original_input, ) from .step_cache import make_step_key key = make_step_key(step, None, original_input, all_variables) try: hit = copy.deepcopy(cache.get(key)) except Exception: hit = None logger.warning("Could not restore cached workflow step") if hit is not None: if verbose: print(f"↩︎ cache hit: {getattr(step, 'name', getattr(step, '__name__', step))}") return dict(hit) result = self._execute_single_step_nocache( step, step_idx, results, all_variables, executor, default_agent, default_llm, memory, planning, verbose, on_step, on_result, original_input, ) # Only cache a successful result -- a failed step must be free to retry # on the next run rather than replay a cached failure forever. if isinstance(result, dict) and not result.get("error"): try: cache.set(key, copy.deepcopy(result)) except Exception: logger.warning("Could not cache workflow step") return result def _execute_single_step_nocache( self, step: Task, step_idx: int, results: List[Dict[str, Any]], all_variables: Dict[str, Any], executor: Optional[Callable[[str], str]] = None, default_agent: Optional[Any] = None, default_llm: Optional[str] = None, memory: Optional[Any] = None, planning: bool = False, verbose: int = 0, on_step: Optional[Callable[[Task, int], None]] = None, on_result: Optional[Callable[[Task, str], None]] = None, original_input: str = "" ) -> Dict[str, Any]: """Execute a single workflow step.""" # Shared prep: handler context + should_run gating context, early_result = self._prepare_step_context( step, results, all_variables, original_input ) if early_result is not None: return early_result # If step has a custom handler function if step.handler: try: result = step.handler(context) # Handle StepResult if isinstance(result, StepResult): # Update variables from result if result.variables: all_variables.update(result.variables) return { "success": True, "output": result.output, "stop": result.stop_workflow, # Early termination flag "error": None } else: return { "success": True, "output": str(result), "stop": False, "error": None } except Exception as e: return { "success": False, "output": None, "stop": False, "error": str(e) } # Shared prep: build step context + substitute variables action = self._build_step_action(step, step_idx, results, all_variables) # Get or create executor for this step step_executor = executor if step_executor is None: # Use step's direct agent if provided step_agent = step.agent # Otherwise create agent from config if step_agent is None: step_agent = self._create_step_agent( step=step, default_agent=default_agent, default_llm=default_llm, memory=memory, planning=planning, verbose=verbose ) if step_agent: if planning and hasattr(step_agent, 'start'): def agent_executor(prompt, agent=step_agent): return agent.start(prompt) step_executor = agent_executor else: def agent_executor(prompt, agent=step_agent): return agent.chat(prompt) step_executor = agent_executor if step_executor is None: return { "success": False, "output": None, "stop": False, "error": f"No executor available for step '{step.name}'" } # Execute step with retries retries = 0 step_success = False output = None error = None while retries <= step.max_retries and not step_success: try: output = step_executor(action) step_success = True except Exception as e: error = str(e) retries += 1 if retries <= step.max_retries: self._log(f"Step '{step.name}' failed, retrying ({retries}/{step.max_retries})") # Shared finalize: variable propagation + on_result callback + return dict return self._finalize_step_result( step, output, step_success, error, all_variables, results, on_result ) async def aexecute( self, workflow_name: str, executor: Optional[Callable[[str], str]] = None, variables: Optional[Dict[str, Any]] = None, on_step: Optional[Callable[[Task, int], None]] = None, on_result: Optional[Callable[[Task, str], None]] = None, default_agent: Optional[Any] = None, default_llm: Optional[str] = None, memory: Optional[Any] = None, planning: bool = False, stream: bool = False, verbose: int = 0, checkpoint: Optional[str] = None, resume: Optional[str] = None, rebase_checkpoint: bool = False ) -> Dict[str, Any]: """ Async version of execute() for workflow execution. Args: workflow_name: Name of the workflow to execute executor: Async function to execute each step. Optional if default_agent provided. variables: Variables to substitute in steps on_step: Callback before each step (step, index) on_result: Callback after each step (step, result) default_agent: Default agent to use for steps without agent_config default_llm: Default LLM model for agent creation memory: Shared memory instance planning: Enable planning mode stream: Enable streaming output verbose: Verbosity level checkpoint: Save checkpoint after each step with this name resume: Resume from checkpoint with this name rebase_checkpoint: Force resume at the same step index even when the workflow definition changed since the checkpoint (skips the fingerprint check with a warning) Returns: Execution results with step outputs and status """ loop = self._prepare_workflow_loop( workflow_name=workflow_name, variables=variables, default_llm=default_llm, planning=planning, resume=resume, rebase_checkpoint=rebase_checkpoint, ) if "error" in loop: return loop["error"] state = loop["state"] # Drive the same shared step-loop generator as execute(), but run each # step with a genuinely async step-runner so async executors / agent.achat # are awaited in-loop (true-async parity, not sync-in-a-thread). step_iter = self._run_workflow_loop( workflow=loop["workflow"], workflow_name=workflow_name, state=state, checkpoint=checkpoint, on_step=on_step, ) request = next(step_iter, None) while request is not None: step, step_idx = request step_result = await self._aexecute_single_step( step=step, step_idx=step_idx, results=state["results"], all_variables=state["all_variables"], executor=executor, default_agent=default_agent, default_llm=loop["default_llm"], memory=memory, planning=loop["planning"], on_result=on_result, ) try: request = step_iter.send(step_result) except StopIteration: break return self._workflow_loop_result(loop["workflow"], state) async def _aexecute_single_step( self, step: Task, step_idx: int, results: List[Dict[str, Any]], all_variables: Dict[str, Any], executor: Optional[Callable[[str], str]] = None, default_agent: Optional[Any] = None, default_llm: Optional[str] = None, memory: Optional[Any] = None, planning: bool = False, on_result: Optional[Callable[[Task, str], None]] = None, original_input: str = "" ) -> Dict[str, Any]: """Async mirror of ``_execute_single_step``. Structurally identical to the sync helper, but this is the only sync/async seam: it ``await``s async executors, ``agent.achat`` and async handlers so ``aexecute`` keeps genuine async execution while sharing the loop skeleton in ``_run_workflow_loop``. """ import asyncio # Shared prep: handler context + should_run gating context, early_result = self._prepare_step_context( step, results, all_variables, original_input ) if early_result is not None: return early_result # If step has a custom handler function if step.handler: try: result = step.handler(context) if asyncio.iscoroutine(result): result = await result # Handle StepResult if isinstance(result, StepResult): if result.variables: all_variables.update(result.variables) return { "success": True, "output": result.output, "stop": result.stop_workflow, "error": None } else: return { "success": True, "output": str(result), "stop": False, "error": None } except Exception as e: return { "success": False, "output": None, "stop": False, "error": str(e) } # Shared prep: build step context + substitute variables action = self._build_step_action(step, step_idx, results, all_variables) # Get or create executor for this step step_executor = executor if step_executor is None: # Use step's direct agent if provided step_agent = step.agent # Otherwise create agent from config if step_agent is None: step_agent = self._create_step_agent( step=step, default_agent=default_agent, default_llm=default_llm, memory=memory, planning=planning, ) if step_agent: # Prefer a genuinely async chat method when available if hasattr(step_agent, 'achat') and asyncio.iscoroutinefunction(getattr(step_agent, 'achat', None)): async def async_agent_executor(prompt, agent=step_agent): return await agent.achat(prompt) step_executor = async_agent_executor elif planning and hasattr(step_agent, 'start'): def agent_executor(prompt, agent=step_agent): return agent.start(prompt) step_executor = agent_executor else: def agent_executor(prompt, agent=step_agent): return agent.chat(prompt) step_executor = agent_executor if step_executor is None: return { "success": False, "output": None, "stop": False, "error": f"No executor available for step '{step.name}'" } # Execute step with retries retries = 0 step_success = False output = None error = None while retries <= step.max_retries and not step_success: try: if asyncio.iscoroutinefunction(step_executor): output = await step_executor(action) else: output = step_executor(action) if asyncio.iscoroutine(output): output = await output step_success = True except Exception as e: error = str(e) retries += 1 if retries <= step.max_retries: self._log(f"Step '{step.name}' failed, retrying ({retries}/{step.max_retries})") # Shared finalize: variable propagation + on_result callback + return dict return self._finalize_step_result( step, output, step_success, error, all_variables, results, on_result ) def _create_step_agent( self, step: Task, default_agent: Optional[Any] = None, default_llm: Optional[str] = None, memory: Optional[Any] = None, planning: bool = False, verbose: int = 0 ) -> Optional[Any]: """ Create an agent for a specific step. Args: step: Workflow step with optional agent_config default_agent: Default agent to use if no config default_llm: Default LLM model memory: Shared memory instance planning: Enable planning mode for the agent verbose: Verbosity level Returns: Agent instance or None """ if step.agent_config: try: from ..agent.agent import Agent config = step.agent_config.copy() # Remove verbose as Agent no longer accepts it config.pop("verbose", None) config.setdefault("name", f"{step.name}Agent") config.setdefault("llm", default_llm) # config.setdefault("verbose", verbose) # Agent no longer accepts verbose config.setdefault("planning", planning) if step.tools: config["tools"] = step.tools if memory: config["memory"] = memory return Agent(**config) except Exception as e: self._log(f"Failed to create agent for step '{step.name}': {e}", logging.ERROR) return default_agent return default_agent def create_workflow( self, name: str, description: str = "", steps: Optional[List[Dict[str, str]]] = None, variables: Optional[Dict[str, Any]] = None ) -> Workflow: """ Create a new workflow file. Args: name: Workflow name description: Workflow description steps: List of step dicts with name, description, action variables: Default variables Returns: Created Workflow object """ workflows_dir = self.workspace_path / self.WORKFLOWS_DIR.replace("/", os.sep) workflows_dir.mkdir(parents=True, exist_ok=True) file_path = workflows_dir / f"{name.lower().replace(' ', '_')}.md" # Build content lines = ["---"] lines.append(f"name: {name}") if description: lines.append(f"description: {description}") if variables: lines.append("variables:") for k, v in variables.items(): lines.append(f" {k}: {v}") lines.append("---") lines.append("") # Add steps workflow_steps = [] if steps: for i, step in enumerate(steps, 1): step_name = step.get("name", f"Step {i}") step_desc = step.get("description", "") step_action = step.get("action", "") lines.append(f"## Step {i}: {step_name}") if step_desc: lines.append(step_desc) lines.append("") lines.append("```action") lines.append(step_action) lines.append("```") lines.append("") workflow_steps.append(Task( name=step_name, description=step_desc, action=step_action )) # Write file file_path.write_text("\n".join(lines), encoding="utf-8") # Create and register workflow workflow = Workflow( name=name, description=description, steps=workflow_steps, variables=variables or {}, file_path=str(file_path) ) self._workflows[name.lower()] = workflow self._log(f"Created workflow '{name}' at {file_path}") return workflow def get_stats(self) -> Dict[str, Any]: """Get workflow statistics.""" self._ensure_loaded() total_steps = sum(len(w.steps) for w in self._workflows.values()) return { "total_workflows": len(self._workflows), "total_steps": total_steps, "workflows": [w.name for w in self._workflows.values()] } # ------------------------------------------------------------------------- # Checkpoint Methods # ------------------------------------------------------------------------- def _get_checkpoints_dir(self) -> Path: """Get the checkpoints directory path.""" from ..paths import DEFAULT_DIR_NAME checkpoints_dir = self.workspace_path / DEFAULT_DIR_NAME / "checkpoints" checkpoints_dir.mkdir(parents=True, exist_ok=True) return checkpoints_dir def _definition_fingerprint(self, workflow: Workflow) -> str: """Stable content hash of a workflow's step definition. Normalised like ``managed/protocols.py`` hashes environments: it hashes the ordered list of each step's identity (name / action / agent / full config), not the raw file bytes. Whitespace/comment-only edits therefore keep the same fingerprint, while adding, removing, reordering, or changing a step's action or agent configuration produces a new one. Used to refuse a silent resume onto a stale step index after the workflow file was edited. Config *values* (not just keys) are hashed so that semantic edits — e.g. changing an agent's instructions, model, tools, condition, or routing — also invalidate the checkpoint rather than silently combining a stale step position with an edited definition. """ import hashlib import json def _step_identity(step: Any) -> Dict[str, Any]: agent = getattr(step, "agent", None) agent_name = getattr(agent, "name", None) if agent is not None else None config = getattr(step, "agent_config", None) or {} return { "name": getattr(step, "name", None), "action": getattr(step, "action", None), "agent": agent_name, "config": config, } payload = [_step_identity(s) for s in (workflow.steps or [])] blob = json.dumps(payload, sort_keys=True, separators=(",", ":"), default=str) return hashlib.sha256(blob.encode("utf-8")).hexdigest()[:12] def _save_checkpoint( self, name: str, workflow_name: str, completed_steps: int, results: List[Dict[str, Any]], variables: Dict[str, Any], definition_fingerprint: Optional[str] = None ) -> str: """ Save workflow checkpoint for later resumption. Args: name: Checkpoint name workflow_name: Name of the workflow completed_steps: Number of completed steps results: Results from completed steps variables: Current variable state definition_fingerprint: Stable hash of the workflow definition, compared on resume to refuse a stale step index after edits Returns: Path to checkpoint file """ import json from datetime import datetime checkpoint_file = safe_child(self._get_checkpoints_dir(), name, ".json") checkpoint_data = { "name": name, "workflow_name": workflow_name, "completed_steps": completed_steps, "results": results, "variables": variables, "definition_fingerprint": definition_fingerprint, "saved_at": time.time(), "saved_at_iso": datetime.now().isoformat() } # Atomic write: serialise to a temp file and os.replace into place so a # crash/OOM/kill mid-write cannot destroy the previous good checkpoint # (which would force the workflow to restart from step 0 and re-run and # re-bill completed steps). from ..utils.atomic_io import atomic_write_json atomic_write_json(str(checkpoint_file), checkpoint_data, indent=2, default=str) self._log(f"Saved checkpoint '{name}' at step {completed_steps}") return str(checkpoint_file) def _load_checkpoint(self, name: str) -> Optional[Dict[str, Any]]: """ Load a workflow checkpoint. Args: name: Checkpoint name Returns: Checkpoint data or None if not found """ import json checkpoint_file = safe_child(self._get_checkpoints_dir(), name, ".json") if not checkpoint_file.exists(): self._log(f"Checkpoint '{name}' not found") return None with open(checkpoint_file, 'r') as f: return json.load(f) def list_checkpoints(self) -> List[Dict[str, Any]]: """ List all saved checkpoints. Returns: List of checkpoint info dicts """ import json checkpoints = [] checkpoints_dir = self._get_checkpoints_dir() for checkpoint_file in checkpoints_dir.glob("*.json"): try: with open(checkpoint_file, 'r') as f: data = json.load(f) checkpoints.append({ "name": data.get("name", checkpoint_file.stem), "workflow": data.get("workflow_name", ""), "completed_steps": data.get("completed_steps", 0), "definition_fingerprint": data.get("definition_fingerprint"), "saved_at": data.get("saved_at_iso", "") }) except Exception: continue checkpoints.sort(key=lambda x: x.get("saved_at", ""), reverse=True) return checkpoints def delete_checkpoint(self, name: str) -> bool: """ Delete a checkpoint. Args: name: Checkpoint name Returns: True if deleted successfully """ checkpoint_file = safe_child(self._get_checkpoints_dir(), name, ".json") if checkpoint_file.exists(): checkpoint_file.unlink() self._log(f"Deleted checkpoint '{name}'") return True return False def create_workflow_manager( workspace_path: Optional[str] = None, **kwargs ) -> WorkflowManager: """ Create a WorkflowManager instance. Args: workspace_path: Path to workspace **kwargs: Additional configuration Returns: WorkflowManager instance """ return WorkflowManager(workspace_path=workspace_path, **kwargs)