Orchestrator

Responsibility

Loads runtime definitions and executes workflow steps via workflow_runner.py.

For asynchronous execution and durability, orchestrator uses:

• RabbitMQ command queues (small/large tracks + retry + DLQ)
• PostgreSQL run/step state persistence
• MinIO payload references for large messages

Inbound Authentication

• HTTP invocation auth (OAuth2 scopes and/or bearer token fallback)
• scheduled invocation auth
• admin reload token (X-Reload-Token)

Outbound Authentication

• Calls egress services over internal network
• Uses storage bearer token for runtime reads/writes

Key Config

• ORCH_RUNTIME_STORE
• WORKFLOWS_DIR, CONNECTIONS_DIR, POLICIES_DIR
• ORCH_RELOAD_TOKEN
• EGRESS_*_URL
• ORCH_MESSAGE_TRACK_THRESHOLD_BYTES (default target: 524288)
• RABBITMQ_URL / queue and exchange names
• POSTGRES_DSN
• MINIO_ENDPOINT, MINIO_BUCKET, credentials

Health

• GET /healthz
• GET /readyz

Async Queue Entry

• POST /run/{workflow_name}/enqueue (returns 202)
• Requires async queue mode enabled (ORCH_ASYNC_QUEUE_ENABLED=true)
• Small track (<= 512KB): queues payload inline
• Large track (> 512KB): uploads payload as artifact and queues payload_ref

Queue Workers

• python -m app.queue_worker (no HTTP server)
• ORCH_WORKER_TRACK=small|large selects queue when ORCH_MESSAGE_CONSUMER_QUEUE is unset
• Kubernetes deployments:
• orchestrator-worker-small
• orchestrator-worker-large
• Uses the same orchestrator image/runtime config as the API pod (workflows, connections, traces, registry discovery)

Run state persistence

• Driver: psycopg (async pool)
• Env: ORCH_MESSAGE_DATABASE_DSN
• Schema: engine/services/orchestrator/db/migrations/0001_workflow_messaging.sql
• Tables: workflow_runs, workflow_steps, message_receipts, workflow_events
• When DSN is unset, run-state writes are skipped (queue flow still works)

Workflow Validation Contract

Orchestrator validates workflow YAML through POST /admin/validate using a layered model:

• contract validation (always)
• provider semantic validation (optional when provider capability supports it)

Recommended response shape (aligned with API reference):

{
    "valid": true,
    "contract_status": "passed",
    "provider_status": "passed",
    "steps": [
        {
            "step_id": "publish_order",
            "contract": {
                "status": "passed",
                "diagnostics": []
            },
            "provider": {
                "status": "passed",
                "diagnostics": []
            }
        }
    ]
}

Diagnostic entry fields:

• severity: error | warning | info
• code: stable machine-readable code
• message: human-readable explanation
• path: optional field path in step/request payload

Blocking policy recommendation:

• contract errors always block publish/reload
• provider errors block only in strict mode
• provider unavailable yields provider_status=skipped in non-strict mode

Two-Track Message Policy

• Small track (<= 512KB)
• Inline payload in queue message.
• In-flow transformations allowed (bounded by execution limits).

• Content-based routing using CEL expressions.

• Large track (> 512KB)

• Payload stored in object storage; queue carries payload_ref.
• No inline transforms in runner.
• Transformations only through provider service_call steps.

See ADR 0006 for the normative contract and queue topology.