conductor

README

Conductor

A self-hosted, Postgres-only durable task and workflow engine.

Run shell commands, HTTP calls and containers across a cluster of workers, with:

• retries with exponential back-off, timeouts and cancellation
• DAG workflows that undo completed steps when a later one fails (sagas)
• cron schedules, queues with concurrency and rate limits, and label-based routing
• namespaces with quotas
• AI steps: model calls with token budgets, human approvals, and agents that survive a crash without repeating finished tool calls
• a web dashboard, Prometheus metrics, OpenTelemetry tracing and live task output
• Kubernetes: a Helm chart for a highly available cluster, workers that scale on queue depth down to zero, tasks that run as their own pods, and an operator for managing workflows with GitOps

Postgres is the only dependency.

Status: pre-1.0 and under active development. See plan.md for the roadmap and DEVLOG.md for how it's built, with diagrams.

Documentation: 13 pages from overview to operations, in docs/slashdocs, published to slashdocs with make docs-publish.

Architecture

flowchart LR
    CLI["conductorctl / Go SDK / curl"] -- "HTTP + API key" --> API
    subgraph cluster [Conductor cluster]
        API["api<br/>HTTP :8081"] -- "gRPC + cluster token (+mTLS)" --> COORD["coordinator (leader)<br/>gRPC :8080"]
        STANDBY["coordinator (standby)"] -. "redirects to leader" .-> COORD
        COORD -- "dispatch / cancel" --> W1["worker<br/>shell · http"]
        COORD -- "dispatch / cancel" --> W2["worker<br/>shell · http"]
        COORD -- "dispatch / cancel" --> W3["container-worker<br/>+ container"]
        W1 & W2 & W3 -- "heartbeats + results" --> COORD
    end
    API -- "reads" --> PG[("PostgreSQL")]
    COORD -- "queue, workflows, schedules" --> PG
    STANDBY -. "campaigns for leadership" .-> PG

Component	Role
api	Stateless HTTP API. Authenticates API keys, reads state from Postgres, and sends commands to the coordinator.
coordinator	Tracks workers, claims tasks from Postgres with FOR UPDATE SKIP LOCKED (respecting queue limits and worker labels), dispatches them, retries failures, recovers tasks from dead workers, drives workflow DAGs, and fires cron schedules. Run two or more: one is elected leader through a Postgres advisory lock and the rest are hot standbys that take over within about a second. Every leader write is fenced by an epoch, and every result by an attempt number, so a deposed leader or a partitioned worker can't change anything.
worker	Runs tasks (shell, HTTP, containers, Kubernetes pods, model calls) in a clean environment. Cancellation kills everything a task started. Drains gracefully on shutdown.

All three are subcommands of one binary, conductor, shipped as one Docker image.

Quickstart

With Docker Compose (Docker and Go 1.24+):

git clone https://github.com/ChinmayNoob/conductor.git && cd conductor
docker compose up -d --build --scale worker=3

go build -o bin/conductorctl ./cmd/conductorctl
export CONDUCTOR_API_KEY=insecure-dev-api-key   # the development default

bin/conductorctl task submit -cmd 'echo hello from $(hostname)' -wait
bin/conductorctl workflow start order_pipeline -input '{"order_id": "A1"}' -wait

Then open the dashboard at http://localhost:8081 and sign in with the same key.

On Kubernetes: make kind-up builds the images and runs everything on a local kind cluster (API at http://localhost:30081; the key from ./scripts/kind-up.sh --api-key). On any other cluster, push the images to your registry first (releases will publish them), then:

docker build -t REGISTRY/conductor:TAG . && docker push REGISTRY/conductor:TAG
helm install conductor deploy/helm/conductor --namespace conductor --create-namespace \
  --set image.repository=REGISTRY/conductor --set image.tag=TAG
kubectl get secret -n conductor conductor-auth -o jsonpath='{.data.api-key}' | base64 -d   # the admin key
kubectl port-forward -n conductor svc/conductor-api 8081:8081

See Kubernetes below.

Or as a single process (only Postgres needed):

POSTGRES_PORT=5433 go run ./cmd/conductor dev    # coordinator + API + worker in one process

The compose file uses insecure development secrets by default. Copy .env.example to .env and set CONDUCTOR_CLUSTER_TOKEN and CONDUCTOR_API_KEY before running anywhere but your own machine.

Tasks

conductorctl task submit -cmd 'make backup' -priority 1 -retries 5 -timeout 600
conductorctl task submit -url https://example.com/hook -method POST -body '{"ok":true}'
conductorctl task submit -image alpine:3.20 -- cat /etc/alpine-release
conductorctl task submit -type k8s -image python:3.12 -memory 512Mi -- python -c 'print(42)'
conductorctl task list -status FAILED
conductorctl task cancel <id>         # kills the process if it's running
conductorctl dlq                      # permanently failed tasks
conductorctl task requeue <id>        # give one a fresh set of retries

Field	Default	Meaning
type	shell	shell runs command with sh -c. http sends http.{method,url,headers,body}. container runs container.{image,command,memory,cpus,network}. k8s runs k8s.{image,command,args,cpu,memory,service_account,node_selector} as its own pod. llm calls a model (see AI steps).
env		Extra environment variables for the task
labels		Only workers with all these labels run the task, e.g. {"gpu": "true"}
queue	default	Queues can have concurrency and rate limits (see below)
priority	5	1 (highest) to 10. Waiting tasks gain priority over time, so nothing starves.
max_retries / retry_delay_seconds	3 / 60	Attempt n waits delay × 2ⁿ (max 1 hour)
timeout_seconds	300	The task is killed after this long
delay_seconds / scheduled_at	now	Run later
idempotency_key		Submitting the same key twice returns the first task (HTTP 200, not 201)

Tasks run in a minimal environment. Worker secrets such as the cluster token are never visible to them. Each task gets CONDUCTOR_TASK_ID, CONDUCTOR_ATTEMPT and CONDUCTOR_OUTPUT. Lines written to $CONDUCTOR_OUTPUT as key=value become the task's outputs.

Delivery is at-least-once: if a worker dies mid-task, the task is retried on another worker, so make side effects idempotent.

Workflows

A workflow is a DAG of steps defined in YAML. Steps run as soon as their dependencies finish, so independent steps run in parallel. If a step fails, running steps are cancelled and completed steps are compensated (undone) in reverse dependency order.

name: order_pipeline
inputs:
  order_id: {required: true}
  amount: {default: "100"}
steps:
  - name: validate
    run: echo "customer=cus_$INPUT_ORDER_ID" >> "$CONDUCTOR_OUTPUT"

  - name: reserve_stock
    depends_on: [validate]
    run: echo "reservation=res_$INPUT_ORDER_ID" >> "$CONDUCTOR_OUTPUT"
    compensate: echo "releasing $OUTPUT_RESERVATION"

  - name: charge_card
    depends_on: [validate]
    env:
      CUSTOMER: "${{ steps.validate.outputs.customer }}"
    run: echo "charging $INPUT_AMOUNT to $CUSTOMER"
    compensate: echo "refunding"

  - name: ship
    depends_on: [reserve_stock, charge_card]
    run: ./ship.sh

flowchart LR
    V[validate] --> R[reserve_stock] & C[charge_card]
    R & C --> S[ship]
    S -. "fails" .-> UC["↩ refund"] & UR["↩ release"]

conductorctl workflow apply -f my_workflow.yaml     # new version only if it changed
conductorctl workflow defs                          # list definitions
conductorctl workflow start order_pipeline -input '{"order_id":"A1"}' -wait
conductorctl workflow cancel <id>                   # cancel running steps, compensate the rest

• Inputs are exposed as INPUT_<NAME> environment variables. Declared inputs can be required or have a default.
• Expressions ${{ inputs.x }}, ${{ steps.s.outputs.k }} and ${{ workflow.id }} may be used in env, HTTP fields and container commands, but never in run. Values reach shell commands only through environment variables, so inputs can't be interpreted as shell code.
• A compensation sees its own step's outputs as OUTPUT_<KEY>. It can be a string (a shell command) or a full action (type: http, …).
• Steps take retries, retry_delay, timeout, priority, queue and labels, and defaults: sets them for every step.
• Each run keeps a snapshot of the definition version it started with, so editing a workflow never affects runs in flight.

Bundled examples: trip_booking, trip_booking_fail, trip_booking_slow and order_pipeline (in examples/workflows).

AI steps

Models are slow, flaky and billed by the token, which is what durable workflows are good at. Set OPENAI_API_KEY (or CONDUCTOR_LLM_BASE_URL for Ollama, vLLM, LiteLLM or Azure) and workers can run:

name: triage
budget: {max_tokens: 20000, max_cost_usd: 0.50}   # the run fails, and compensates, if it overspends
steps:
  - name: classify
    llm:
      prompt: "Severity of this report: ${{ inputs.report }}"
      schema: {type: object, properties: {severity: {type: string}}, required: [severity], additionalProperties: false}
  - name: sign_off
    depends_on: [classify]
    type: approval
    approval: {message: "Page the on-call? severity is ${{ steps.classify.outputs.severity }}", timeout: 1h}
  - name: investigate
    depends_on: [sign_off]
    type: agent
    agent:
      prompt: "Find out why checkout is slow."
      max_turns: 8
      tools:
        - name: recent_errors
          description: Last errors from the checkout service
          run: tail -n 50 /var/log/checkout.log

• llm steps and tasks (conductorctl task submit -prompt "...") return the answer as text, or the fields of a JSON schema as outputs. A provider's Retry-After delays the retry; permanent errors (bad key, unknown model) are not retried. Queues can cap tokens_per_minute; namespaces can cap tokens and dollars per day.
• approval and signal steps wait for a person (conductorctl approvals, workflow approve|reject, or the dashboard) or an outside system (workflow signal), holding no worker.
• agent steps run a model with tools in a loop. Every model call and tool call is its own task, so killing the worker mid-run resumes from the last finished call; only the call in flight runs again. Tool arguments arrive as $ARG_<NAME> and $TOOL_ARGS, never in the command line. Limits: turns, tool calls, duration, budget.
• Costs are counted in tokens always, and in dollars only for models you price in CONDUCTOR_LLM_PRICES (none are built in).

Operations assistant (opt in per namespace: conductorctl namespace set NAME -ai-assist). Redacted failures go to the model, which adds a likely cause and a fix next to a rule-based class (transient, permanent, needs attention). It never changes retries and never saves anything on its own:

conductorctl task explain <id>                     # also appears on failed tasks in the dashboard
conductorctl ai ask "which tasks failed in the last hour, and why?"   # read-only lookups
conductorctl ai workflow "fetch and render in parallel, then publish" -save   # validated, dry run, asks before saving

Design notes: ADR 0003. For tests and demos, docker compose --profile mock-llm runs a free scripted model server.

Schedules

conductorctl schedule create nightly-report -cron '0 2 * * *' -tz Asia/Kolkata -cmd './report.sh'
conductorctl schedule create sync -cron '@every 5m' -workflow order_pipeline -input '{"order_id":"X"}'
conductorctl schedule list
conductorctl schedule trigger nightly-report   # fire now
conductorctl schedule pause nightly-report

Misfire policy (what happens to runs missed while the coordinator was down): skip (default), run_once, or catch_up. Each run time fires at most once, even across restarts.

Queues, priorities and labels

conductorctl queue set emails -concurrency 5 -rate 100 -per 60   # ≤5 at once, ≤100/minute
conductorctl queue set emails -pause
conductorctl queue list

Workers advertise labels (CONDUCTOR_WORKER_LABELS=gpu=true,region=eu) plus the task types they support. Tasks requiring labels nobody has simply wait. A dedicated worker also names labels a task must require to run there (CONDUCTOR_WORKER_EXCLUSIVE=gpu), so an expensive pool, or one that scales from zero, never picks up ordinary work.

Namespaces

Namespaces isolate teams. Each has its own tasks, workflows, schedules, queues and API keys, with optional quotas:

conductorctl namespace create team-a -max-pending 1000 -max-concurrency 20
conductorctl apikey create team-a-ci -namespace team-a
conductorctl -n team-a task list          # admin keys can act in any namespace

HTTP API

Every /v1 route needs Authorization: Bearer <api key>. Admin keys may add X-Conductor-Namespace. Errors look like {"error": "..."}.

Area	Endpoints
Tasks	POST /v1/tasks · GET /v1/tasks?status=&queue=&q=&before= · GET /v1/tasks/{id} · GET /v1/tasks/{id}/logs?follow=true · GET /v1/tasks/{id}/attempts · POST /v1/tasks/{id}/cancel · POST /v1/tasks/{id}/requeue · GET /v1/dead-letter · GET /v1/stats · GET /v1/stats/timeline
Definitions	PUT /v1/workflow-definitions (YAML or JSON body) · GET /v1/workflow-definitions · GET /v1/workflow-definitions/{name}?version=&format=yaml
Runs	POST /v1/workflows · GET /v1/workflows · GET /v1/workflows/{id} · POST /v1/workflows/{id}/cancel
Schedules	POST /v1/schedules · GET /v1/schedules · GET/DELETE /v1/schedules/{name} · POST /v1/schedules/{name}/{pause,resume,trigger}
Queues	GET /v1/queues · PUT/DELETE /v1/queues/{name}
Admin	GET /v1/workers · GET /v1/cluster · POST/GET /v1/namespaces · PUT /v1/namespaces/{name} · POST/GET /v1/api-keys · DELETE /v1/api-keys/{id}
Health	GET /health (no auth)

The Go SDK in pkg/client covers all of it:

c := client.New("http://localhost:8081", os.Getenv("CONDUCTOR_API_KEY"))
wf, _ := c.StartWorkflow(ctx, client.WorkflowRequest{Workflow: "order_pipeline", Input: map[string]any{"order_id": "A1"}})
wf, _ = c.WaitForWorkflow(ctx, wf.ID)

Dashboard and observability

Dashboard at /ui (/ redirects there), built into the binary. Sign in with an API key; it acts with that key's permissions. It shows what is running and waiting, every task with its attempts and live output, workflow runs drawn as a live diagram of which step failed and what was undone, approvals waiting for a decision, agent conversations, and workers, schedules and the dead-letter queue. You can cancel, requeue, pause or resume queues and schedules, and run schedules now. Design notes: ADR 0002.

Live output from the CLI or HTTP:

conductorctl task logs <id> -f      # follows a running task, across retries
conductorctl task attempts <id>     # every attempt, and why each one failed

Metrics: every process serves Prometheus metrics on :9090 (CONDUCTOR_METRICS_LISTEN). Tracing: set OTEL_EXPORTER_OTLP_ENDPOINT to send OpenTelemetry traces; a task's trace runs from the HTTP request through the queue into the task, which gets TRACEPARENT. To run all of it locally:

OTEL_EXPORTER_OTLP_ENDPOINT=http://jaeger:4317 docker compose --profile observability up -d --build --scale worker=3
# Grafana http://localhost:3000 · Prometheus http://localhost:9091 · Jaeger http://localhost:16686

If a port is taken, set GRAFANA_PORT, PROMETHEUS_PORT or JAEGER_PORT.

The profile provisions a Grafana dashboard and example alerts: no leader, no healthy workers, a queue backing up, a failure-rate spike, lost tasks.

Kubernetes

The chart in deploy/helm/conductor runs the whole cluster:

flowchart LR
    U["conductorctl / UI / operator"] --> SVC["api Service"] --> API["api ×2+<br/>(HPA)"]
    API --> CO["coordinator ×2<br/>(leader + standby)"]
    CO -- "dispatch" --> W["worker pools<br/>(Deployments)"]
    W -- "k8s tasks" --> P["task pods"]
    KEDA["KEDA"] -- "conductor_slot_demand()" --> PG
    KEDA -- "scales 0..N" --> W
    API & CO --> PG[("Postgres<br/>dev · CloudNativePG · external")]

• Highly available: two coordinators with leader election, an API with a HorizontalPodAutoscaler, PodDisruptionBudgets, readiness and liveness probes. A rolling upgrade loses no task and fails no request (tested in CI).
• Secure by default: mutual TLS between components with a generated CA, generated secrets kept across upgrades, non-root pods with read-only root filesystems, and NetworkPolicies that let each component accept only the connections it needs.
• Postgres: postgres.mode is dev (one pod, to try it out), cloudnativepg (an HA cluster via the CloudNativePG operator) or external (a managed database).
• Autoscaling: with autoscaling.enabled, KEDA scales a worker pool on the slots it could use right now (due tasks it can run, plus tasks running on it), down to zero when idle. Pools are listed in workerPools; make scale-to-zero pools dedicated.
• Tasks as pods: with kubernetesTasks.enabled, k8s tasks run as their own pods with their own image, resources and service account. They get no privileges by default (non-root, no capabilities, no token); tasks may only use service accounts from kubernetesTasks.serviceAccounts. Outputs work as for shell tasks. A pool of its own creates the pods; it runs no task code and is the only one holding Kubernetes credentials, so shell tasks elsewhere have none to read.
• Operator: with operator.enabled, Workflow, WorkflowRun and Schedule resources are applied to Conductor, so workflows live in Git and ship with kubectl apply or Argo CD. See operator/README.md.
• Monitoring: pods carry Prometheus scrape annotations; the chart can add a PodMonitor, the alert rules as a PrometheusRule, the Grafana dashboard, or a PodMonitoring for Google Managed Prometheus.

Reference setups: Terraform for GKE with Cloud SQL and GitOps with Argo CD. Design: ADR 0004.

Security

• API keys: CONDUCTOR_API_KEY creates the first admin key on startup. Other keys are bound to one namespace. Only SHA-256 hashes are stored.
• Cluster token: every gRPC call between components carries CONDUCTOR_CLUSTER_TOKEN.
• Mutual TLS: set CONDUCTOR_TLS_CERT, CONDUCTOR_TLS_KEY and CONDUCTOR_TLS_CA (try ./scripts/gen-dev-certs.sh with docker-compose.tls.yml).
• Task isolation: tasks run as an unprivileged user with a minimal environment. Container tasks need a worker with the Docker socket, which is opt-in (docker compose --profile containers up -d) because socket access is equivalent to root on the host. On Kubernetes, k8s tasks run as their own locked-down pods instead; only the pool that creates them holds a Kubernetes token (a Role for pods in the task namespace), and it runs no task code.

Configuration

All settings are environment variables; .env.example lists them. The most important:

Variable	Default	Purpose
CONDUCTOR_CLUSTER_TOKEN	(required)	Shared secret for gRPC between components
CONDUCTOR_API_KEY		Bootstrap admin API key
POSTGRES_HOST / PORT / USER / PASSWORD / DB	localhost / 5432 / …	Database
CONDUCTOR_COORDINATOR_ADDR	localhost:8080	Where the API and workers reach the coordinator
CONDUCTOR_WORKER_SLOTS	2	Tasks a worker runs at once
CONDUCTOR_WORKER_LABELS		e.g. gpu=true,region=eu
CONDUCTOR_WORKER_EXCLUSIVE		Labels a task must require to run here (a dedicated worker), e.g. gpu
CONDUCTOR_K8S_TASKS	false	Run k8s tasks as pods (workers in a cluster; the chart sets this up)
CONDUCTOR_WORKER_TASK_TYPES		Run only these task types, e.g. k8s (default: every type the worker can)
CONDUCTOR_WORKER_PASS_ENV		Worker variables tasks may see, e.g. AWS_REGION
CONDUCTOR_PRIORITY_AGING	60s	Waiting tasks gain one priority level per interval (0 disables it)
CONDUCTOR_SHUTDOWN_TIMEOUT	25s	How long a stopping worker lets running tasks finish
CONDUCTOR_METRICS_LISTEN	:9090	Prometheus metrics (off disables them)
OTEL_EXPORTER_OTLP_ENDPOINT		Send traces over OTLP/gRPC; standard OTEL_* variables apply
OPENAI_API_KEY / CONDUCTOR_LLM_BASE_URL		Enable llm and agent steps (workers) and the assistant (API)
CONDUCTOR_LLM_MODEL	gpt-4o-mini	Model for steps that name none
CONDUCTOR_LLM_PRICES		JSON price list, USD per million tokens; without it costs stay zero

Development

make help        # list targets
make test        # unit tests
make lint        # golangci-lint (runs in Docker)
make up          # start the stack with 3 workers
make e2e         # end-to-end suite against the running stack
make chaos       # chaos suite: faults injected under load, then invariant checks
make bench       # benchmarks at 1, 10 and 50 workers
make proto       # regenerate gRPC code (protoc runs in Docker)
make kind-up     # run Conductor on a local kind cluster (images, KEDA, the chart)
make e2e-k8s     # the end-to-end suite on kind
make operator    # build and test the operator (its own Go module)
make helm-lint   # lint the chart and render it in every mode

Database tests need a Postgres server; they create and drop a throwaway database per test:

CONDUCTOR_TEST_DATABASE_URL='postgres://postgres:postgres@localhost:5433/postgres?sslmode=disable' go test ./pkg/db/

CI runs lint, unit and database tests with the race detector, the end-to-end suite over plaintext (with tracing and the observability stack on) and mutual TLS, the operator's tests, the chart's lint and render checks, and the end-to-end suite again on a kind cluster, with dev Postgres and with CloudNativePG. The chaos suite runs nightly.

Reliability and performance

• High availability: killing the leader coordinator mid-workflow loses nothing; a standby takes over in 0.5–1 s.
• Chaos-tested: a steady workload survives killed workers and coordinators, a network partition, Postgres latency, a cut database connection and a Postgres restart, with every task completed and every saga compensated exactly where it should be.
• Benchmarks: about 800 no-op tasks/s drained and ~5 ms idle dispatch on a laptop; see BENCHMARKS.md.
• Design records: docs/adr.

License

Apache 2.0