This guide provides comprehensive instructions for creating Visor workflows. It covers the structure, available check types, configuration patterns, testing DSL, and best practices.
- Workflow Structure
- Check Types Reference
- Configuration Patterns
- Testing DSL
- Style Guide
- Example Patterns
- Common Pitfalls
Every Visor workflow follows this structure:
version: "1.0"
# Optional: Workflow metadata
id: my-workflow
name: My Workflow Name
description: What this workflow does
# Optional: Global routing configuration
routing:
max_loops: 5 # Prevent infinite routing loops
# Optional: Workflow-level outputs
outputs:
- name: result
description: The aggregated result
value_js: |
const all = Object.values(outputs || {});
return all.map(v => v?.issues || []).flat();
# Required: Steps definition
steps:
step-one:
type: ai
prompt: "Analyze the code"
# ... step configuration
step-two:
type: command
depends_on: [step-one]
exec: "echo done"
# Optional: Inline tests
tests:
defaults:
strict: true
ai_provider: mock
cases:
- name: basic-flow
event: manual
# ... test case configuration| Section | Required | Description |
|---|---|---|
version |
Yes | Always "1.0" |
steps |
Yes | Map of step names to configurations |
outputs |
No | Workflow-level output definitions |
routing |
No | Global routing configuration |
tests |
No | Inline test cases (or use separate .tests.yaml) |
imports |
No | External workflow files to import |
Steps can declare which events trigger them:
steps:
on-pr-open:
type: ai
on: [pr_opened] # Only on PR open
on-pr-changes:
type: ai
on: [pr_opened, pr_updated] # PR open or update
on-any-event:
type: command
# No 'on:' means the check runs on ANY event (event-agnostic)Available events:
pr_opened- Pull request openedpr_updated- Pull request synchronized/updatedpr_closed- Pull request closedissue_opened- Issue createdissue_comment- Comment on issue or PRmanual- CLI execution (no event)schedule- Scheduled/cron-triggered executionwebhook_received- HTTP webhook was received (via http_input provider)
AI-powered analysis using LLMs (Gemini, Claude, OpenAI).
steps:
analyze:
type: ai
on: [pr_opened, pr_updated]
# The prompt sent to the AI
prompt: |
Analyze the code changes for security issues.
Files changed: {{ files | json }}
PR Title: {{ pr.title }}
# Output schema (JSON Schema or named schema)
schema: code-review # Named schema
# OR inline schema:
schema:
type: object
properties:
issues:
type: array
items:
type: object
properties:
severity: { type: string, enum: [critical, error, warning, info] }
message: { type: string }
required: [severity, message]
required: [issues]
# AI configuration
ai:
provider: anthropic # google, anthropic, openai
model: claude-sonnet-4-20250514
skip_code_context: false # Include code context in prompt
disableTools: false # Allow tool use
system_prompt: |
You are a security expert.Advanced AI with MCP tools, file editing, and subagents.
steps:
comprehensive-analysis:
type: claude-code
prompt: |
Perform a comprehensive code review:
{{ outputs['get-requirements'].text }}
claude_code:
allowedTools: ['Read', 'Grep', 'Edit', 'Write', 'Bash']
maxTurns: 10
systemPrompt: |
You are an expert code reviewer.
# Bash permissions
allowBash: true
bashConfig:
allow:
- 'npm test'
- 'npm run lint'
- 'git status'
- 'git diff'
deny:
- 'rm -rf'
- 'git push'
# MCP server configuration
mcpServers:
analyzer:
command: "node"
args: ["./tools/analyzer.js"]
env:
MODE: "deep"Execute shell commands.
steps:
build:
type: command
exec: "npm run build"
# Working directory
cwd: "{{ outputs.checkout.path }}"
# Environment variables
env:
NODE_ENV: production
API_KEY: "{{ env.API_KEY }}"
# Output format
output_format: json # text (default), json
# Output schema for JSON
schema:
type: object
properties:
success: { type: boolean }
errors: { type: array }
multi-command:
type: command
exec: |
npm ci
npm run lint
npm testPause for user input.
steps:
get-input:
type: human-input
prompt: |
What would you like to accomplish?
Be specific about constraints and requirements.
placeholder: "Enter your task description..."
multiline: true # Allow multi-line input
allow_empty: false # Require input
default: "yes" # Default value
timeout: 300 # Timeout in secondsOutput structure: { text: string, ts: number }
Tip: Use
--tuimode for interactive prompts and real-time visualization:visor --tui --config workflow.yamlTUI provides a chat-style interface for human-input prompts. See Interactive TUI Mode.
Output messages to the console/log. Useful for debugging workflows and displaying execution information.
steps:
finish:
type: log
depends_on: [process]
message: |
Processing complete!
Results:
{% for item in outputs['process'].results %}
- {{ item.name }}: {{ item.status }}
{% endfor %}
level: info # debug, info, warn, error
include_pr_context: false
include_dependencies: false
include_metadata: falseNote: The type must be log (not logger).
Execute JavaScript code.
steps:
transform:
type: script
content: |
const input = outputs['previous-step'];
const filtered = input.items.filter(i => i.valid);
return {
total: input.items.length,
valid: filtered.length,
items: filtered
};
# Schema for output validation
schema:
type: object
required: [total, valid, items]Perform GitHub API operations.
steps:
add-labels:
type: github
criticality: external
depends_on: [analyze]
assume:
- "(outputs['analyze']?.labels?.length ?? 0) > 0"
op: labels.add
values:
- "{{ outputs['analyze'].labels | json }}"
create-comment:
type: github
op: comment.create
values:
body: |
## Analysis Complete
{{ outputs['analyze'].summary }}Available operations:
labels.add,labels.remove,labels.setcomment.create,comment.updatereview.create,review.approve,review.request_changesstatus.create
Store and retrieve state across steps.
steps:
store:
type: memory
operation: set
key: "analysis_result"
value: "{{ outputs['analyze'] | json }}"
namespace: "my-workflow"
retrieve:
type: memory
operation: get
key: "analysis_result"
namespace: "my-workflow"
increment:
type: memory
operation: increment
key: "attempt_count"
value: 1
namespace: "retry-loop"Call another workflow as a step.
steps:
security-scan:
type: workflow
workflow: security-scan # Workflow ID
args:
severity_threshold: high
scan_dependencies: true
output_mapping:
vulnerabilities: scan_resultsCheckout code from a repository.
steps:
checkout:
type: git-checkout
repository: owner/repo # GitHub repository
ref: "{{ pr.head }}" # Branch, tag, or commit
# Optional configuration
fetch_depth: 1 # Shallow clone
fetch_tags: false
submodules: false # true, false, or 'recursive'
working_directory: /tmp/checkoutOutput: { success, path, ref, commit, repository }
Make HTTP requests.
steps:
fetch-data:
type: http_client
url: "https://api.example.com/data"
method: POST
headers:
Authorization: "Bearer {{ env.API_TOKEN }}"
Content-Type: application/json
body: |
{ "query": "{{ outputs['input'].query }}" }
schema:
type: object
properties:
data: { type: array }Receive data via webhook.
steps:
webhook-receiver:
type: http_input
path: /webhook/data
method: POSTA pass-through step for orchestration.
steps:
checkpoint:
type: noop
depends_on: [step-a, step-b, step-c]
# All dependencies must complete before dependents runControl execution order with depends_on:
steps:
first:
type: command
exec: "echo first"
second:
type: command
depends_on: [first] # Runs after 'first'
exec: "echo second"
parallel-a:
type: command
depends_on: [second]
exec: "echo parallel-a"
parallel-b:
type: command
depends_on: [second] # Runs in parallel with parallel-a
exec: "echo parallel-b"
final:
type: command
depends_on: [parallel-a, parallel-b] # Waits for both
exec: "echo final"Skip steps conditionally:
steps:
conditional:
type: command
depends_on: [check]
if: "outputs['check']?.should_run === true"
exec: "echo running"Assert preconditions before execution:
steps:
process:
type: command
depends_on: [fetch]
assume:
- "outputs['fetch']?.data != null"
- "(outputs['fetch']?.data?.length ?? 0) > 0"
exec: "process-data"Validate output:
steps:
analyze:
type: ai
prompt: "Analyze code"
# Schema validation
schema:
type: object
required: [issues]
properties:
issues: { type: array }
# Post-execution guarantee
guarantee: "output.issues != null && Array.isArray(output.issues)"Mark step as failed based on output:
steps:
validate:
type: command
exec: "./validate.sh"
fail_if: "output.code !== 0"
ai-check:
type: ai
prompt: "Check for issues"
fail_if: "output.issues?.some(i => i.severity === 'critical')"Control flow after step completion:
steps:
validate:
type: command
exec: "npm test"
fail_if: "output.code !== 0"
on_fail:
run: [fix-issues] # Run remediation step
goto: validate # Then retry (ancestor only)
retry:
max: 2
backoff:
mode: exponential
delay_ms: 1000
on_success:
goto: finalize
fix-issues:
type: claude-code
prompt: "Fix the test failures"
claude_code:
allowedTools: ['Read', 'Edit']
finalize:
type: log
message: "All tests pass!"Process arrays in parallel:
steps:
extract-items:
type: ai
forEach: true # Output is treated as array
prompt: "Extract items from: {{ pr.body }}"
schema:
type: array
items:
type: object
properties:
id: { type: string }
task: { type: string }
process-item:
type: command
depends_on: [extract-items]
fanout: map # Run once per item
exec: "process {{ outputs['extract-items'].task }}"
aggregate:
type: script
depends_on: [process-item]
fanout: reduce # Run once with all results
content: |
const results = outputs_history['process-item'] || [];
return { total: results.length };Continue AI conversations across steps:
steps:
initial-analysis:
type: ai
prompt: "Analyze the code structure"
follow-up:
type: ai
depends_on: [initial-analysis]
reuse_ai_session: initial-analysis
session_mode: clone # or 'append'
prompt: "Now look for security issues in what we discussed"Session modes:
clone(default): Copy the conversation history to a new sessionappend: Share the conversation history (subsequent messages append to the same session)
Transform step outputs before they're consumed by dependent steps:
steps:
fetch-data:
type: http_client
url: "https://api.example.com/data"
# Liquid transform
transform: |
{% assign items = response.data.items %}
{{ items | json }}
# OR JavaScript transform (alternative)
transform_js: |
const data = JSON.parse(output);
return data.items.filter(i => i.active);Run preprocessing steps before a step executes:
steps:
ai-review:
type: ai
on_init:
run:
# Invoke a tool to enrich context
- tool: fetch-jira
with:
issue_key: "{{ pr.body | regex_search: '[A-Z]+-[0-9]+' }}"
as: jira_context
# Or invoke another step
- step: enrich-context
as: extra_context
prompt: |
Review the code with this context:
JIRA: {{ outputs['jira_context'] | json }}The on_init hook allows:
run: Array of tool invocations, step invocations, or workflow invocationsrun_js: Dynamic computation of what to runtransitions: Declarative routing rules
See on_init Hook RFC for detailed documentation.
Tests can be inline in the workflow or in a separate file:
# workflow-name.tests.yaml
version: "1.0"
extends: "./workflow-name.yaml"
tests:
defaults:
strict: true # Every executed step must be asserted
ai_provider: mock # Use mock AI provider
prompt_max_chars: 16000 # Truncate captured prompts
tags: "fast" # Only run steps with these tags
exclude_tags: "slow" # Skip steps with these tags
cases:
- name: basic-flow
description: Tests the happy path
event: manual # or pr_opened, pr_updated, etc.
fixture: local.minimal # Built-in or custom fixture
mocks:
step-one: "mock response"
step-two:
field: "value"
items: [1, 2, 3]
# Array mocks for loops
step-three[]:
- { attempt: 1, status: "fail" }
- { attempt: 2, status: "pass" }
expect:
calls:
- step: step-one
exactly: 1
- step: step-two
at_least: 1
at_most: 3
no_calls:
- step: should-not-run
prompts:
- step: step-one
contains:
- "expected text"
not_contains:
- "unwanted text"
outputs:
- step: step-two
path: items.length
equals: 3
- step: step-two
path: status
matches: "^(pass|success)$"Built-in fixtures:
gh.pr_open.minimal- Minimal PR opened eventgh.pr_sync.minimal- Minimal PR sync eventgh.issue_open.minimal- Minimal issue opened eventgh.issue_comment.standard- Standard issue commentlocal.minimal- Minimal local/manual fixture
Custom fixtures:
tests:
fixtures:
- name: my-fixture
extends: gh.pr_open.minimal
overrides:
pr:
title: "Custom PR Title"
labels: ["bug", "urgent"]mocks:
# Simple string mock (for human-input or command stdout)
get-input: "user input text"
# JSON object mock (for AI with schema)
analyze:
issues: []
summary: "All good"
# Array mock for forEach or loops
extract[]:
- { id: 1, name: "item1" }
- { id: 2, name: "item2" }
# Command mock
build:
stdout: '{"success": true}'
stderr: ""
exit_code: 0expect:
# Call count assertions
calls:
- step: my-step
exactly: 1 # Exactly N times
- step: retry-step
at_least: 1 # At least N times
at_most: 5 # At most N times
# Negative assertions
no_calls:
- step: should-skip
# Prompt assertions
prompts:
- step: ai-step
index: last # first, last, or number
contains: ["keyword"]
not_contains: ["bad"]
matches: "pattern.*"
# Output assertions
outputs:
- step: process
path: result.status # Dot notation path
equals: "success"
- step: process
path: items
contains_unordered: ["a", "b"]
- step: process
where: { path: type, equals: "important" }
path: value
matches: "\\d+"
# Failure assertions
fail:
message_contains: "expected error"Test sequences of events:
tests:
cases:
- name: multi-event-flow
flow:
- name: pr-opened
event: pr_opened
fixture: gh.pr_open.minimal
mocks:
overview: { text: "Initial review" }
expect:
calls:
- step: overview
exactly: 1
- name: pr-updated
event: pr_updated
fixture: gh.pr_sync.minimal
mocks:
overview: { text: "Updated review" }
expect:
calls:
- step: overview
exactly: 1- One step, one responsibility - Keep steps focused and composable
- Declare intent before mechanics - Readers should understand what/when before how
- Guard and contract every important step - Use
assumeandschema/guarantee - Avoid hidden control flow - Prefer declarative routing over imperative logic
For each step, use this order:
my-step:
# 1. Identity & Intent
type: ai
criticality: external # external, internal, policy, info
group: analysis
tags: [security, slow]
description: Analyzes code for security issues
# 2. Triggers & Dependencies
on: [pr_opened, pr_updated]
depends_on: [overview]
# 3. Preconditions (Guards)
assume:
- "outputs['overview']?.text != null"
if: "outputs['overview']?.shouldAnalyze === true"
# 4. Provider Configuration
prompt: |
Analyze for security issues...
ai:
provider: anthropic
model: claude-sonnet-4-20250514
# 5. Contracts (Post-Exec)
schema: code-review
guarantee: "output.issues != null"
# 6. Failure Policies
fail_if: "output.issues?.some(i => i.severity === 'critical')"
continue_on_failure: false
# 7. Routing & Transitions
on_success:
goto: next-step
on_fail:
run: [fix-step]
goto: my-step
# 8. Runtime Controls
timeout: 120
retries: 2-
external: Side effects outside repo/CI (GitHub ops, webhooks)- Requires:
assumeorifprecondition - Requires:
schemaorguaranteefor outputs
- Requires:
-
internal: Orchestration/state within CI- Same requirements as
external
- Same requirements as
-
policy: Evaluative checks (security, quality)- Guards/contracts optional
-
info: Purely informational, never gates dependents
Do:
- Declare
criticalityand follow guard/contract rules - Keep expressions short and defensive:
outputs?.x?.length ?? 0 - Add
schemawhenever output shape matters - Use meaningful step names
- Include tests for all workflows
Don't:
- Hide control flow in templates or long JS snippets
- Mix unrelated responsibilities in a single step
- Depend on outputs you didn't guard
- Use magic numbers without explanation
- Create workflows without tests
steps:
get-task:
type: human-input
prompt: "Describe what you want to accomplish"
multiline: true
allow_empty: false
refine:
type: ai
depends_on: [get-task]
ai:
disableTools: true
schema:
type: object
properties:
refined: { type: boolean }
text: { type: string }
required: [refined, text]
prompt: |
Refine this task into clear, actionable requirements:
{{ outputs['get-task'].text }}
If clarification is needed, set refined=false and ask in text.
If complete, set refined=true with the final specification.
fail_if: "output.refined !== true"
on_fail:
goto: get-taskrouting:
max_loops: 5
steps:
generate:
type: ai
prompt: "Generate code for: {{ inputs.task }}"
validate:
type: command
depends_on: [generate]
exec: "npm run lint && npm test"
fail_if: "output.code !== 0"
on_fail:
run: [fix]
on_success:
goto: complete
fix:
type: claude-code
depends_on: [validate]
if: "outputs['validate']?.code !== 0"
prompt: |
Fix these errors:
{{ outputs['validate'].stderr }}
claude_code:
allowedTools: ['Read', 'Edit']
maxTurns: 5
on_success:
goto: validate
complete:
type: log
depends_on: [validate]
message: "Validation passed!"steps:
overview:
type: ai
on: [pr_opened, pr_updated]
prompt: "Provide PR overview"
schema: overview
security:
type: ai
depends_on: [overview]
prompt: "Analyze for security issues"
schema: code-review
performance:
type: ai
depends_on: [overview]
prompt: "Analyze for performance issues"
schema: code-review
aggregate:
type: script
depends_on: [security, performance]
content: |
const all = [
...(outputs['security']?.issues || []),
...(outputs['performance']?.issues || [])
];
return {
issues: all,
hasErrors: all.some(i => i.severity === 'critical' || i.severity === 'error')
};steps:
analyze:
type: ai
on: [pr_opened]
prompt: "Analyze and suggest labels"
schema:
type: object
properties:
labels: { type: array, items: { type: string } }
effort: { type: integer, minimum: 1, maximum: 5 }
apply-labels:
type: github
criticality: external
depends_on: [analyze]
assume:
- "(outputs['analyze']?.labels?.length ?? 0) > 0"
op: labels.add
values:
- "{{ outputs['analyze'].labels | json }}"
- "effort:{{ outputs['analyze'].effort }}"Wrong:
steps:
process:
type: command
exec: "process {{ outputs['fetch'].data }}" # fetch not declared as dependencyRight:
steps:
process:
type: command
depends_on: [fetch]
exec: "process {{ outputs['fetch'].data }}"Wrong:
steps:
step-a:
on_fail:
goto: step-b
step-b:
depends_on: [step-a]
on_fail:
goto: step-a # Infinite loop!Right:
routing:
max_loops: 3 # Limit iterations
steps:
step-a:
on_fail:
goto: step-b
retry:
max: 2 # Limit retriesWrong:
steps:
add-labels:
type: github
op: labels.add
values:
- "{{ outputs['analyze'].labels }}" # May be null!Right:
steps:
add-labels:
type: github
criticality: external
depends_on: [analyze]
assume:
- "(outputs['analyze']?.labels?.length ?? 0) > 0"
op: labels.add
values:
- "{{ outputs['analyze'].labels | json }}"Wrong:
tests:
cases:
- name: test
mocks:
step-one: "value"
expect:
outputs:
- step: step-one
path: text
equals: "value"
# Missing call assertions - unexecuted steps go unnoticedRight:
tests:
defaults:
strict: true # Require call assertions for all executed steps
cases:
- name: test
mocks:
step-one: "value"
expect:
calls:
- step: step-one
exactly: 1
outputs:
- step: step-one
path: text
equals: "value"Wrong:
steps:
extract:
type: ai
forEach: true
prompt: "Extract items"
process:
depends_on: [extract]
# Runs once with first item only!Right:
steps:
extract:
type: ai
forEach: true
prompt: "Extract items"
process:
depends_on: [extract]
fanout: map # Runs for each itemWrong:
expect:
outputs:
- step: calculate
path: result
equals: 42 # Magic number - why 42?Right:
# Use meaningful values that relate to the mock inputs
mocks:
input: { value: 6 }
multiplier: { value: 7 }
expect:
outputs:
- step: calculate
path: result
equals: 42 # 6 * 7 = 42, derivable from inputsAvailable in prompts and Liquid templates:
| Variable | Description |
|---|---|
pr |
PR metadata (title, body, author, labels, etc.) |
files |
Changed files list |
outputs |
Map of dependency outputs |
outputs['step-name'] |
Specific step output |
outputs_history['step-name'] |
All historical outputs for step |
outputs_raw['step-name'] |
Raw/aggregate output |
env |
Environment variables |
event |
Current event details |
inputs |
Workflow input parameters |
| Filter | Example | Description |
|---|---|---|
json |
{{ data | json }} |
JSON encode |
default |
{{ x | default: 'fallback' }} |
Default value |
size |
{{ arr | size }} |
Array/string length |
first |
{{ arr | first }} |
First element |
last |
{{ arr | last }} |
Last element |
join |
{{ arr | join: ', ' }} |
Join array |
split |
{{ str | split: ',' }} |
Split string |
upcase |
{{ str | upcase }} |
Uppercase |
downcase |
{{ str | downcase }} |
Lowercase |
Available in if, fail_if, assume, guarantee, run_js, goto_js:
| Variable | Description |
|---|---|
output |
Current step's output |
outputs |
Map of dependency outputs |
outputs.history |
Historical outputs map |
attempt |
Current attempt number |
loop |
Current routing loop number |
step |
Current step metadata |
pr |
PR metadata |
files |
Changed files |
env |
Environment variables |
event |
Event metadata |
memory |
Memory access (get/set/increment) |
# Run workflow
visor --config workflow.yaml
# Run with message (for human-input)
visor --config workflow.yaml --message "input text"
# Validate configuration
visor validate --config workflow.yaml
# Run tests
visor test --config workflow.tests.yaml
# Run specific test
visor test --only test-name
# List tests
visor test --list
# Validate tests only
visor test --validate
# Simulate event
visor --config workflow.yaml --event pr_openedTUI mode provides a persistent terminal interface for running any workflow:
# Start interactive TUI
visor --tui --config workflow.yaml
# TUI with debug logging
visor --tui --config workflow.yaml --debugTUI Tabs:
- Chat (1): Interactive prompts and workflow results
- Logs (2): Execution logs and debug output
- Traces (3): Real-time OpenTelemetry execution tree
Key Bindings:
| Key | Action |
|---|---|
Shift+Tab |
Cycle between tabs |
1 / 2 / 3 |
Jump to Chat / Logs / Traces |
e |
Toggle engine states in Traces tab |
Enter |
Submit input |
Ctrl+C |
Abort workflow |
q |
Exit (when complete) |
The Traces tab shows:
- Execution tree with check hierarchy
- forEach iterations grouped under parent
- IN/OUT/ERR lines for each span
- Press
eto show/hide engine internals (LevelDispatch, WavePlanning)
For more detailed information on specific topics:
- Configuration Reference - Full configuration options
- Event Triggers - Event types and filtering
- Dependencies - Dependency patterns and execution order
- Failure Routing - Retry, remediation, and routing
- AI Configuration - AI provider settings
- Claude Code - Claude Code provider details
- MCP Provider - MCP tool integration
- Command Provider - Shell command execution
- HTTP Provider - HTTP client and webhooks
- Git Checkout - Repository checkout
- Memory - State persistence
- Script - JavaScript execution
- Testing Getting Started - Quick start for testing
- Testing DSL Reference - Complete test syntax
- Fixtures and Mocks - Test data setup
- Assertions - Assertion syntax
- Workflow Style Guide - Best practices
- Criticality Modes - Safety levels
- Fault Management - Contracts and guards
- on_init Hook - Context preprocessing
- Output History - Accessing historical outputs
- Tag Filtering - Selective execution
- Liquid Templates - Template syntax
- Debugging - Troubleshooting workflows