Workflow Graph Engine

August 29, 2026 · View on GitHub

The workflow graph engine defines all factory modes as directed graphs of typed nodes. One source of truth, two execution formats: headless automation via WorkflowExecutor and interactive CEO sessions via auto-generated SKILL.md files.

How it works

definitions.py          (8 Python functions, each returning a Workflow)
       │
       ├──► WorkflowExecutor    (headless: walks the DAG deterministically)
       │       factory workflow run improve --project /path
       │
       └──► skill_export.py     (interactive: converts graph → SKILL.md)
               skills/workflow-improve/SKILL.md
               └── CEO reads this at runtime as its mode-specific playbook

In interactive mode, factory ceo launches a Claude Code session with ceo.md as the system prompt. The CEO detects project state, then reads the appropriate SKILL.md into context and follows it step by step. The SKILL.md files are prose translations of the same graph the executor walks — so both paths execute the same pipeline.

Node types

Every workflow is a graph of 6 node types connected by edges:

Node	Class	Purpose	Example
Agent	`AgentNode`	Spawn a Claude Code specialist agent	Researcher, Builder, QA
Function	`FnNode`	Run a shell command	`factory eval {project_path}`
Gate	`GateNode`	Decision point producing PROCEED / RELOOP / HALT	CEO reviewing research quality
Fork	`ForkNode`	Launch multiple targets in parallel	3 researchers simultaneously
Join	`JoinNode`	Barrier — wait for all parallel branches	Wait for all researchers
Study	`Study`	Distinguished `FnNode` wrapping `factory study`	Local codebase analysis

Each node declares reads and writes — the set of files it consumes and produces. The graph validator (validation.py) uses these to verify data flow: every file a node reads must be written by a predecessor. Pre-existing project files (e.g. CLAUDE.md, factory.md) should not be declared as reads since no workflow node produces them.

Edges and verdicts

Edges connect nodes. Unconditional edges always fire. Conditional edges fire only on a specific verdict from a GateNode:

Edge(source="gate_qa", target="gate_precheck", condition=VerdictType.PROCEED)
Edge(source="gate_qa", target="builder",        condition=VerdictType.RELOOP)

Three verdict types:

PROCEED — output is satisfactory, continue to the next step
RELOOP — output needs improvement, go back to a target node (max 3 iterations)
HALT — something is fundamentally wrong, stop the workflow

Workflows

8 workflows are registered in definitions.py:

Name	Function	Trigger	Purpose
`build`	`build_workflow()`	`no_repo` or `incomplete`	Build a new project from idea/spec
`design`	`design_workflow()`	`no_repo` + interactive	Same as build but with user approval gate at strategy
`improve`	`improve_workflow()`	`has_factory`	Improve an existing project through experiments
`research`	`research_workflow()`	`has_factory` + `research_target`	Research-driven optimization with failure analysis
`meta`	`meta_workflow()`	`has_factory` + `mode=meta`	Improve the factory itself + ACE playbook evolution
`discover`	`discover_workflow()`	`no_factory`	Auto-discover eval dimensions
`review`	`review_workflow()`	`evals_pending_review`	Verify eval dimensions and initialize factory config
`refine`	`refine_workflow()`	`has_factory` + `--refine`	Lightweight pipeline for user-directed refinements

Relationships: W2 (design) = W1 (build) with gate_strategy.evaluator_type = "user". W4 (research) extends W3 (improve) with baseline measurement, failure analyst, surface constraints, and plateau detection.

Creating a new workflow

Here is the discover workflow (simplest — 3 nodes) as an example:

from factory.workflow.primitives import (
    AgentRole, Edge, FnNode, GateNode, VerdictType, Workflow,
)

def discover_workflow() -> Workflow:
    nodes = {}
    edges = []

    # Step 1: Run discovery command
    nodes["discover"] = FnNode(
        id="discover",
        command="factory discover {project_path}",
        writes={".factory/eval_profile.json", "eval/score.py"},
    )

    # Step 2: CEO reviews the result
    nodes["gate_discover"] = GateNode(
        id="gate_discover",
        evaluator_type="agent",
        evaluator_role=AgentRole.CEO,
        gate_prompt="Verify the discovered eval profile makes sense...",
        reads={".factory/eval_profile.json", "eval/score.py"},
    )

    # Step 3: Re-detect project state
    nodes["redetect"] = FnNode(
        id="redetect",
        command="factory detect {project_path}",
        reads={".factory/eval_profile.json"},
    )

    # Wire them: discover → gate → redetect (on PROCEED)
    #                         └→ discover (on RELOOP — retry)
    edges = [
        Edge(source="discover", target="gate_discover"),
        Edge(source="gate_discover", target="redetect", condition=VerdictType.PROCEED),
        Edge(source="gate_discover", target="discover", condition=VerdictType.RELOOP),
    ]

    # Auto-select when project has no factory setup
    def trigger(state, ctx):
        return state == ProjectState.NO_FACTORY

    return Workflow(
        name="discover",
        nodes=nodes,
        edges=edges,
        start_node="discover",
        trigger=trigger,
    )

To register it, add the function to register_all() in definitions.py:

def register_all() -> dict[str, Workflow]:
    return {
        ...
        "discover": discover_workflow(),
    }

Common patterns

QA iteration loop (builder → QA → gate with RELOOP back to builder, max 3 iterations):

nodes["builder"] = AgentNode(id="builder", role=AgentRole.BUILDER, ...)
nodes["gate_build"] = GateNode(id="gate_build", ...)
nodes["qa"] = AgentNode(id="qa", role=AgentRole.QA, ...)
nodes["gate_qa"] = GateNode(id="gate_qa", ...)
nodes["gate_precheck"] = GateNode(id="gate_precheck", ...)

edges = [
    Edge(source="builder", target="gate_build"),
    Edge(source="gate_build", target="qa", condition=VerdictType.PROCEED),
    Edge(source="qa", target="gate_qa"),
    Edge(source="gate_qa", target="gate_precheck", condition=VerdictType.PROCEED),
    Edge(source="gate_qa", target="builder", condition=VerdictType.RELOOP),  # retry
]

Parallel research (fork 3 researchers, join, then gate):

nodes["fork_research"] = ForkNode(
    id="fork_research",
    targets=["researcher_a", "researcher_b", "researcher_c"],
)
nodes["join_research"] = JoinNode(
    id="join_research",
    sources=["researcher_a", "researcher_b", "researcher_c"],
)

Non-blocking archivist (fire-and-forget):

nodes["archivist"] = AgentNode(
    id="archivist", role=AgentRole.ARCHIVIST,
    model="haiku", blocking=False,
)

Validation

The graph validator (validation.py) checks:

Start node exists in the node set
All edge sources and targets reference existing nodes
All nodes are reachable from the start node
Cycles only pass through GateNodes with RELOOP edges
Fork targets match their ForkNode's target list
Join sources match their JoinNode's source list
Every file a node reads is written by a predecessor (data flow integrity)

Run validation:

factory workflow validate              # All 8 workflows
python -c "from factory.workflow.definitions import register_all
for name, wf in register_all().items():
    issues = wf.validate_graph()
    print(f'{name}: {\"CLEAN\" if not issues else issues}')"

CLI commands

# Run a workflow (headless, deterministic graph execution)
factory workflow run improve --project /path/to/project
factory workflow run build --project /path/to/project --dry-run

# List all registered workflows
factory workflow list

# Show a workflow's structure (nodes, edges, triggers)
factory workflow show improve

# Validate all workflow graphs
factory workflow validate

# Regenerate SKILL.md files from graph definitions
factory workflow export-skills

Launching the factory

Interactive mode (CEO + skills)

# Improve an existing project
factory ceo /path/to/project

# Build from an idea — brainstorm first
factory ceo "a weather CLI in Rust" --mode design

# Build directly (clear spec)
factory ceo "a weather CLI in Rust"

# Focus on one thing
factory ceo /path/to/project --focus "add auth"
factory ceo /path/to/project --focus 42  # GitHub issue number

# Research-driven optimization
factory ceo "SWE-bench solver" --mode research

# Self-improve the factory
factory ceo /path/to/factory --mode meta

# Quick refinement
factory ceo /path/to/project --refine "fix the login bug"

What happens under the hood:

cmd_ceo() resolves path, mode, focus directives
Creates a git worktree for isolation
Builds a task string describing what the CEO should do
Resolves the CEO system prompt from factory/agents/prompts/ceo.md
Launches claude (or another runner) with the CEO prompt + task
The CEO detects project state → reads the matching skills/workflow-*/SKILL.md → follows it step by step, spawning specialist agents via factory agent <role>
On exit, the worktree is cleaned up

Headless mode (graph executor)

# Direct graph execution — no CEO agent, the executor walks the DAG
factory workflow run improve --project /path/to/project

# With dry-run (no actual agent spawns or commands)
factory workflow run build --project /path/to/project --dry-run

# Headless CEO (pipe mode — for scripting, cron, tmux)
factory ceo /path/to/project --headless
factory run /path/to/project --loop --interval 1800

Continuous loop

# Heartbeat loop — run improve every 30 minutes
factory run /path/to/project --loop --interval 1800

# In a detached tmux session
factory tmux /path/to/project --loop

Package ecosystem (`package.py`)

A Package wraps a workflow subgraph behind a typed interface and composes with other Packages:

Port — named artifact slot (data plane)
StateContract — preconditions/postconditions (control plane)
OptKnob — parameter the outer loop can mutate (with optional expandable=True for runtime expansion)
MemoryDeclaration — what a package persists and how

Composition operators: Sequential, Parallel, Conditional, Loop. All lower to flat Workflow DAGs via Package.compile().

The compiled Workflow carries three knob-related fields populated by compile():

knob_values: dict[str, str | float] — current knob defaults
knob_bounds: dict[str, list[str | float]] — allowed values per knob
knob_expandable: dict[str, str] — expansion hints for knobs the optimizer can grow

These fields are preserved through to_dict()/from_dict() serialization and _deep_copy_workflow() in the mutation operators.

See docs/design/package-ecosystem.md for the full design doc.

File layout

factory/workflow/
├── __init__.py          # Public API: re-exports all primitives + executor
├── primitives.py        # Pydantic models: Node types, Edge, Verdict, Workflow
├── package.py           # Package ecosystem: Package, Port, StateContract, OptKnob,
│                        #   MemoryDeclaration, Sequential, Parallel, Conditional, Loop
├── definitions.py       # 8 workflow functions returning Workflow objects
├── executor.py          # WorkflowExecutor — async graph walker
├── validation.py        # NetworkX-based graph validator
├── events.py            # Structured event types for .factory/events.jsonl
├── skill_export.py      # Graph → SKILL.md converter
└── cli.py               # CLI subcommands: run, list, show, validate, export-skills

skills/
├── workflow-build/SKILL.md       # Auto-generated from build_workflow()
├── workflow-design/SKILL.md      # Auto-generated from design_workflow()
├── workflow-discover/SKILL.md    # Auto-generated from discover_workflow()
├── workflow-improve/SKILL.md     # Auto-generated from improve_workflow()
├── workflow-meta/SKILL.md        # Auto-generated from meta_workflow()
├── workflow-refine/SKILL.md      # Auto-generated from refine_workflow()
├── workflow-research/SKILL.md    # Auto-generated from research_workflow()
└── workflow-review/SKILL.md      # Auto-generated from review_workflow()

Agent pool

The default agent pool maps roles to models:

Role	Model	Purpose
researcher	sonnet	Web research + local analysis
strategist	opus	Hypothesis generation
builder	opus	Code implementation
qa	opus	Health check + code review + adversarial QA
failure_analyst	opus	Research mode failure classification
ceo	opus	Orchestration + gate evaluation
archivist	haiku	Fast, cheap summarization
refiner	opus	Refinement scoping

Configured in DEFAULT_AGENT_POOL in primitives.py. Override per-node with AgentNode(model="sonnet").