Workflow Executor Skill

Workflow Executor Skill

How to Use This Skill

1. Review the patterns and examples below
2. Apply the relevant patterns to your implementation
3. Follow the best practices outlined in this skill

Execute declarative workflow definitions as state-machine orchestrated multi-agent tasks.

Research Basis (2024-2025)

This skill implements patterns from leading AI agent orchestration research:

Pattern	Source	Implementation
State Machine Workflows	LangGraph 2025	FSM-based execution with node/edge model
Declarative + Imperative	DevOps Best Practices	YAML definitions + Python execution
Graph-Based Orchestration	Azure AI Patterns	Nodes (agents) + Edges (transitions)
Persistence/Resume	AWS Multi-Agent	Checkpoint/resume across sessions

When to Use This Skill

✅ Use this skill when:

• Executing workflows from docs/workflows/ as automated tasks
• Converting documentation workflows to executable state machines
• Coordinating multi-agent tasks with defined phases
• Need auditable, reproducible workflow execution
• Require checkpoint/resume for long-running workflows

❌ Don't use this skill when:

• Single-step tasks (no workflow needed)
• Ad-hoc exploration (use agents directly)
• Tasks without defined phases

Architecture

┌─────────────────────────────────────────────────────────────────┐
│                     WORKFLOW EXECUTOR                            │
├─────────────────────────────────────────────────────────────────┤
│                                                                  │
│  ┌──────────────────┐    ┌──────────────────┐                   │
│  │  Declarative     │    │  Imperative      │                   │
│  │  Definition      │───►│  Execution       │                   │
│  │  (YAML/JSON)     │    │  (FSM Engine)    │                   │
│  └──────────────────┘    └────────┬─────────┘                   │
│                                   │                              │
│         ┌─────────────────────────┼─────────────────────────┐   │
│         │                         │                         │   │
│         ▼                         ▼                         ▼   │
│  ┌────────────┐           ┌────────────┐           ┌────────────┐
│  │   Node 1   │──────────►│   Node 2   │──────────►│   Node N   │
│  │  (Agent)   │   edge    │  (Agent)   │   edge    │  (Agent)   │
│  └────────────┘           └────────────┘           └────────────┘
│                                                                  │
│  State: { current_node, completed_nodes, context, checkpoints } │
│                                                                  │
└─────────────────────────────────────────────────────────────────┘

Workflow Definition Schema

Workflows are defined as YAML files with nodes (agents) and edges (transitions):

# workflows/example-workflow.yaml
name: parallel-task-isolation
version: 1.0.0
description: Execute multiple tasks in parallel using git worktrees

# Metadata
metadata:
  category: devops
  estimated_duration: "10-30 minutes"
  token_budget: 80000
  requires_approval: false

# State machine definition
states:
  - INITIATE
  - CREATE_WORKTREES
  - EXECUTE_PARALLEL
  - MONITOR
  - MERGE
  - CLEANUP
  - COMPLETE

# Initial and terminal states
initial_state: INITIATE
terminal_states: [COMPLETE, FAILED]

# Nodes (agents/actions)
nodes:
  - id: task_parser
    type: function
    description: Parse task definitions from input
    handler: parse_tasks

  - id: worktree_creator
    type: agent
    agent: git-workflow-orchestrator
    description: Create isolated git worktrees for each task

  - id: parallel_executor
    type: multi-agent
    agents: ["$task.agent for task in tasks"]  # Dynamic agent assignment
    description: Execute tasks in parallel worktrees
    parallel: true

  - id: progress_monitor
    type: function
    handler: monitor_progress
    description: Monitor task completion status

  - id: merge_handler
    type: agent
    agent: git-workflow-orchestrator
    description: Merge completed tasks back to main branch

  - id: cleanup_handler
    type: function
    handler: cleanup_worktrees
    description: Remove worktrees and branches

# Edges (transitions)
edges:
  - from: INITIATE
    to: CREATE_WORKTREES
    node: task_parser
    condition: "tasks.length > 0"

  - from: CREATE_WORKTREES
    to: EXECUTE_PARALLEL
    node: worktree_creator
    condition: "all_worktrees_created"

  - from: EXECUTE_PARALLEL
    to: MONITOR
    node: parallel_executor
    condition: "agents_launched"

  - from: MONITOR
    to: MERGE
    node: progress_monitor
    condition: "all_tasks_complete"
    on_failure: MONITOR  # Loop until complete

  - from: MERGE
    to: CLEANUP
    node: merge_handler
    condition: "all_merged"
    on_failure: FAILED

  - from: CLEANUP
    to: COMPLETE
    node: cleanup_handler

# Error handling
error_handling:
  retry_limit: 3
  on_error: FAILED
  checkpoint_on_error: true

# Checkpoints (for resume)
checkpoints:
  - after: CREATE_WORKTREES
  - after: EXECUTE_PARALLEL
  - after: MERGE

Core Capabilities

1. Load Workflow Definition

from skills.workflow_executor.core import WorkflowExecutor

# Load from YAML file
executor = WorkflowExecutor.from_file("workflows/parallel-task-isolation.yaml")

# Or load from docs/workflows/ by name
executor = WorkflowExecutor.from_library("parallel-task-isolation")

2. Execute Workflow

# Execute with input
result = executor.execute({
    "tasks": [
        {"description": "Security audit", "agent": "security-specialist"},
        {"description": "Performance test", "agent": "performance-profiler"},
    ]
})

# Check result
print(f"Status: {result.status}")  # COMPLETE or FAILED
print(f"Output: {result.output}")
print(f"Duration: {result.duration}")

3. Resume from Checkpoint

# Resume failed or paused workflow
executor = WorkflowExecutor.from_checkpoint("workflow-123-checkpoint.json")
result = executor.resume()

4. Monitor Progress

# Get current state
state = executor.get_state()
print(f"Current: {state.current_node}")
print(f"Completed: {state.completed_nodes}")
print(f"Progress: {state.progress_percent}%")

CLI Usage

# Execute workflow by name
/execute-workflow parallel-task-isolation --tasks "Task 1" "Task 2"

# Execute from file
/execute-workflow --file workflows/custom.yaml

# Resume from checkpoint
/execute-workflow --resume workflow-123-checkpoint.json

# Dry run (preview without execution)
/execute-workflow parallel-task-isolation --dry-run

Integration with Existing Skills

multi-agent-workflow FSM

The workflow executor extends the existing FSM infrastructure:

# Existing FSM states
INITIATE → IDENTIFY → DOCUMENT → SOLVE → CODE → DEPLOY → TEST → VALIDATE → COMPLETE

# Workflow executor adds:
- Dynamic state loading from YAML definitions
- Node-to-agent mapping
- Edge conditions as Python expressions
- Parallel execution support
- Checkpoint/resume persistence

git-workflow-automation

For workflows involving git operations:

nodes:
  - id: git_ops
    type: skill
    skill: git-workflow-automation
    method: create_task_environment

Workflow Activation

Workflows can be activated like other components:

# Activate workflow
python3 scripts/update-component-activation.py activate workflow parallel-task-isolation

# Check status
python3 scripts/update-component-activation.py status workflow parallel-task-isolation

# List activated workflows
python3 scripts/update-component-activation.py list --type workflow --activated-only

Example Workflows

1. Parallel Task Isolation

/execute-workflow parallel-task-isolation \
  --tasks "Security audit:security-specialist" \
          "Performance test:performance-profiler"

2. Full-Stack Feature Development

/execute-workflow full-stack-feature \
  --feature "Add user authentication" \
  --modules backend frontend database

3. Security Audit

/execute-workflow security-audit \
  --scope full \
  --report-format markdown

Metrics & Monitoring

Metric	Target	Description
Workflow load time	<100ms	Parse YAML definition
State transition	<500ms	Move between nodes
Checkpoint creation	<1s	Persist state to disk
Resume time	<2s	Restore from checkpoint
Success rate	>95%	Workflows completing successfully

Error Handling

Error	Cause	Recovery
InvalidWorkflowDefinition	YAML syntax error	Fix definition, reload
AgentNotFound	Referenced agent doesn't exist	Check agent name
TransitionFailed	Edge condition not met	Check condition, retry
CheckpointCorrupted	Invalid checkpoint file	Start fresh

Best Practices

DO:

• Define clear state transitions with conditions
• Use checkpoints for long-running workflows
• Assign specialized agents to nodes
• Test workflows with --dry-run first
• Use parallel execution for independent tasks

DON'T:

• Create overly complex state machines (>10 states)
• Skip error handling definitions
• Ignore token budget limits
• Mix unrelated tasks in one workflow

Success Output

When successful, this skill MUST output:

✅ SKILL COMPLETE: workflow_executor

Workflow Execution Summary:
- Workflow name: {workflow_name}
- Workflow file: {path/to/workflow.yaml}
- Status: {COMPLETE|FAILED}
- Duration: {duration}
- Nodes executed: {N}
- State transitions: {N}

Completed:
- [x] Workflow definition loaded from YAML
- [x] FSM initialized with states and transitions
- [x] All nodes executed successfully
- [x] Edge conditions evaluated correctly
- [x] Checkpoints created at designated points
- [x] Final state reached: {terminal_state}
- [x] Output produced and verified

Outputs:
- Result status: {result.status}
- Result output: {result.output}
- Checkpoint files: {N} created
- Audit trail: {path/to/audit.log}
- Token budget: {used}/{budget} ({percentage}%)

Completion Checklist

Before marking this skill as complete, verify:

• Workflow definition YAML loaded successfully
• All required nodes defined (function, agent, multi-agent, skill)
• All edges have valid conditions
• Initial and terminal states defined
• FSM execution completed without errors
• All checkpoints created at designated points
• Error handling executed as defined
• Retry limits respected (if errors occurred)
• Final state is a terminal state (COMPLETE or FAILED)
• Audit trail generated with all transitions

Failure Indicators

This skill has FAILED if:

• ❌ Invalid workflow definition (YAML syntax error)
• ❌ Referenced agent does not exist
• ❌ Edge condition failed and no retry defined
• ❌ Checkpoint file corrupted (cannot resume)
• ❌ State transition failed (no valid edge)
• ❌ Infinite loop detected (exceeded max transitions)
• ❌ Token budget exceeded without checkpoint
• ❌ Required node handler not found
• ❌ Terminal state never reached
• ❌ Error handling failed to recover

When NOT to Use

Do NOT use this skill when:

• Single-step tasks requiring no state machine
• Ad-hoc exploration without defined phases
• Tasks without clear state transitions
• Workflows with undefined or changing requirements
• Simple scripts that don't need orchestration
• Real-time interactive workflows (FSM overhead too high)
• Workflows requiring human decision points (not automated)

Use alternatives instead:

• Direct agent invocation for single-step tasks
• Manual coordination for exploratory work
• Simple scripts for linear, deterministic workflows
• Interactive workflows for human-in-the-loop processes

Anti-Patterns (Avoid)

Anti-Pattern	Problem	Solution
Overly complex state machines (>10 states)	Hard to understand, maintain	Break into sub-workflows, simplify
Missing error handling	Unrecoverable failures	Define error_handling with retry limits
Ignoring token budget	Context overflow	Set token_budget in metadata, monitor
No checkpoints	Lost progress on failure	Define checkpoint points in YAML
Mixing unrelated tasks	Workflow confusion	Keep workflows focused, single purpose
Hard-coded values	Brittle, not reusable	Use workflow inputs, parameterize
Skipping dry-run	Unexpected execution	Always test with --dry-run first
No audit trail	Cannot debug failures	Ensure audit logging enabled
Circular dependencies	Infinite loops	Validate edge conditions prevent cycles
Missing terminal states	Workflow never completes	Always define COMPLETE and FAILED states

Principles

This skill embodies:

• #1 Recycle → Extend → Re-Use → Create - Reusable workflow definitions across projects
• #4 Separation of Concerns - Declarative (YAML) separate from imperative (FSM engine)
• #5 Eliminate Ambiguity - Explicit states, edges, conditions
• #6 Clear, Understandable, Explainable - YAML definitions are self-documenting
• #8 No Assumptions - Validate workflow definition before execution
• #11 Inform → Do → Verify - Dry-run → execute → verify terminal state reached

Workflow Executor Principles:

1. Define workflows declaratively in YAML (nodes, edges, states)
2. Execute imperatively via FSM engine
3. Support checkpoint/resume for long-running workflows
4. Validate workflow definitions before execution
5. Generate comprehensive audit trails
6. Respect token budgets and create checkpoints
7. Handle errors with retry logic and fallbacks
8. Test workflows with --dry-run before production

Workflow Activation:

python3 scripts/update-component-activation.py activate workflow {workflow-name}

Full Standard: CODITECT-STANDARD-AUTOMATION.md

---

Related Components

• multi-agent-workflow - FSM execution engine
• git-workflow-automation - Git operations
• orchestrator - Task coordination
• /execute-workflow - CLI command

---

Workflow Executor v1.0 - Declarative Workflow Definitions + Imperative FSM Execution

Last Updated: 2025-12-18 Owner: CODITECT Core Team Status: Production Ready Research Basis: LangGraph, Azure AI Patterns, AWS Multi-Agent Systems (2024-2025)

Sources:

• LangGraph 2025 Review
• Azure AI Agent Design Patterns
• LangGraph Multi-Agent Workflows
• AWS Multi-Agent Systems