Running Composer — The Full Stack, a First Workflow and the Test Suites

Everything is built. Now run it and push a real workflow through every layer you wrote.

#Step 1: Start the stack

You need Docker with Compose:

make docker-up    # PostgreSQL, Redis, coordinator, 2 workers, language servers, web UI, observability

Service	Address
Web UI	http://localhost:3000 (sign in as admin / admin123, change it after first login)
Coordinator HTTP API	http://localhost:8080
Coordinator gRPC (workers)	localhost:50051
Jaeger (traces)	http://localhost:16686
Grafana (dashboards)	http://localhost:3001
Prometheus	http://localhost:9090
PostgreSQL / Redis	localhost:5432 / localhost:6379

make docker-logs follows the logs; make docker-down stops everything.

#Or run the pieces yourself

psql "$DATABASE_URL" -f migrations/V001__init.sql      # schema, partitions, seed data

./bin/coordinator --store.url "$DATABASE_URL" --redis.addrs localhost:6379

(cd cmd/language-server-py && PORT=50053 uv run python main.py)
(cd cmd/language-server-js && PORT=50052 npm start)

LANG_WORKER_ADDRS=python:localhost:50053,js:localhost:50052 ./bin/worker

(cd web && npm run dev)                                 # http://localhost:3000

Coordinator flags (or COMPOSER_-prefixed environment variables, e.g. COMPOSER_STORE_URL):

Flag	Default	Meaning
--store.url	postgres://postgres:password@localhost:5432/composer	PostgreSQL URL
--redis.addrs	localhost:6379	Redis address
--api.http_port / --api.grpc_port	8080 / 50051	API ports
--stream_shards	4	streams per task queue (published to Redis for workers)
--retention_days	0	drop run history for months that ended more than N days ago (0 keeps everything)
--outbox.poll_interval	1s	outbox fallback poll
--outbox.batch_size / --outbox.concurrency	50 / 2	outbox publishing

#Step 2: Log in

TOKEN=$(curl -s localhost:8080/api/v1/auth/login \
  -d '{"username":"admin","password":"admin123"}' | jq -r .token)

Every other endpoint needs Authorization: Bearer $TOKEN.

#Step 3: Register, deploy and run a two-step workflow

Register version 1 of a workflow that fetches JSON over HTTP and then upper-cases a field in Python. The spec is YAML inside the JSON body:

curl -s localhost:8080/api/v1/workflows -H "Authorization: Bearer $TOKEN" -d @- <<'JSON'
{
  "workflow_id": "hello",
  "name": "Hello",
  "version": 1,
  "spec": "nodes:\n  - {node_id: fetch, name: Fetch, type: HTTP, spec: {url: 'https://httpbin.org/json', method: GET}}\n  - node_id: shout\n    name: Shout\n    type: CODE\n    runtime: python\n    spec:\n      code: |\n        def handler(input):\n            return {'title': input['title'].upper()}\n      title: '{{tasks.fetch.output.slideshow.title}}'\nedges:\n  - {from: fetch, to: shout}\n"
}
JSON

Deploy it and start a run:

curl -s -X POST "localhost:8080/api/v1/workflows/hello/deploy?version=1" -H "Authorization: Bearer $TOKEN"
RUN=$(curl -s localhost:8080/api/v1/workflows/hello/execute -H "Authorization: Bearer $TOKEN" \
  -d '{"input":{}}' | jq -r .run_id)

Inspect it:

curl -s localhost:8080/api/v1/executions/$RUN -H "Authorization: Bearer $TOKEN" | jq

#What just happened, layer by layer

Look at the run's tasks in the database to see your schema at work:

psql "$DATABASE_URL" -c "SELECT node_id, state, attempt, lease_owner FROM task_instances WHERE run_id = '$RUN'"

#Step 4: Try signals, retries and cancels

With the order workflow from the model page (WAIT_FOR_EVENT on approve-{{input.order_id}}), start a large order, then approve it:

curl -s localhost:8080/api/v1/executions/$RUN/events -H "Authorization: Bearer $TOKEN" \
  -d '{"event": "approve-42", "payload": {"approved_by": "manager"}}'
# → {"delivered": true}   a waiting node consumed it
# → {"delivered": false}  buffered: the run has not reached the wait yet

Restart a finished run from a node, or cancel a running one:

curl -s localhost:8080/api/v1/executions/$RUN/retry  -H "Authorization: Bearer $TOKEN" -d '{"node_id": "ship"}'
curl -s -X POST localhost:8080/api/v1/executions/$RUN/cancel -H "Authorization: Bearer $TOKEN"

The full REST surface:

Method	Path	Purpose
POST	/auth/login	Exchange username/password for a JWT
POST	/workflows	Register a definition version (spec is YAML)
GET	/workflows, /workflows/{id}	List / get definitions
POST	/workflows/{id}/deploy?version=N, /workflows/{id}/rollback	Choose the deployed version
POST	/workflows/{id}/execute	Start a run: {"input": {…}, "execution_options": {…}} → {"run_id"}
GET	/executions, /executions/{id}	List runs / run details with every task
POST	/executions/{id}/events	Signal a run → {"delivered": bool}
POST	/executions/{id}/retry	Restart a finished run from a node
POST	/executions/{id}/cancel	Cancel a run

Errors map straight from the state machine's error values: 400 invalid request, 404 not found, 409 conflict (retrying a run with tasks in flight, signalling a finished run).

#Step 5: Break things on purpose

The best way to trust the engine is to watch it heal:

cd infra/docker

# Kill one of the two worker replicas in the middle of a slow step.
docker kill $(docker compose ps -q worker | head -1)
#   → within ~90 s the lease expires; the other worker reclaims and finishes the task.

# Restart Redis while runs are in flight.
docker compose restart redis
#   → publishes fail and back off; dispatch resumes when Redis is back.

# Restart the coordinator.
docker compose restart coordinator
#   → in-flight reports fail and are retried by workers; nothing is half-applied.

Watch the recovery in Grafana (outbox backlog, stream depth) and follow one run end to end in Jaeger.

#Step 6: Run the test suites

go test -short ./...                         # unit tests only (no Docker needed)
make test                                    # everything, incl. container-backed tests (needs Docker)
go test ./tests/ -run TestEngine -v          # engine scenarios end to end

(cd cmd/language-server-py && uv run python -m unittest test_sandbox -v)   # Python sandbox
(cd cmd/language-server-js && npm test)                                    # JS sandbox (Node 22)

Suite	What it covers
pkg/statemachine	every engine rule, event by event, against an in-memory snapshot
pkg/model, pkg/template, pkg/stream	DAG index, template references, CEL resolution, queue sharding
pkg/coord, pkg/api/..., pkg/outbox, pkg/auth	query building, run ID validation, error mapping, request validation, outbox backoff, JWT
worker	the worker pipeline against an in-memory stream and a fake coordinator
tests/	full system: fork/join, switches, retries, waits, signals, queues, lease recovery, NOTIFY latency, partition retention
sandbox suites	known escape techniques are blocked; ordinary code keeps working

#What you built

Layer	What it does
pkg/model + pkg/template	a workflow is an indexed DAG + CEL
pkg/statemachine	pure rules: events in, actions out
migrations + repository	partitioned, fenced, minimal indexes
pkg/coord	lock → decide → persist → commit
pkg/outbox + pkg/stream	commit-time NOTIFY, SKIP LOCKED, shards
pkg/lease + worker/	one executor per task, renew or abandon
recovery loops	expired leases, lost messages, timers

A handful of ideas carry the whole design: keep the decision logic pure, commit decisions and their effects together, serialize per run with a row lock, fence every result with an attempt number, and make every background job safe to run everywhere at once. With those in place, adding a new node type is a few lines in the state machine and a new executor on the worker.