Events & streaming

The executor emits events as a run advances: a node starts, a node finishes, the run pauses at a gate, the run completes or fails, the checks are scored. An in-process EventBus fans those events out to every WebSocket subscribed to that run. There is exactly one streaming endpoint, and it is server-push only.

The stream endpoint

ws://localhost:8000/api/runs/af772b3602ef/stream

Authentication

The HTTP auth middleware does not cover WebSockets — it only intercepts HTTP requests. The stream endpoint therefore checks auth itself, and only when the server runs in AUTH_MODE=required:

• In AUTH_MODE=open (the local default) the socket accepts the connection with no credential.
• In AUTH_MODE=required you must pass a session token as a ?token= query parameter (a header is not an option for browser WebSockets). The server verifies it before accepting; a missing or invalid token closes the socket with code 4401 before it ever opens.

// swap ws:// for wss:// behind TLS
const token = localStorage.getItem("ballast_token");
const url = "ws://localhost:8000/api/runs/af772b3602ef/stream?token=" + encodeURIComponent(token);
const ws = new WebSocket(url);
// invalid/missing token → server closes with code 4401 (no 'open' fires)

The event envelope

Every frame is a JSON object. The bus stamps two fields on all of them —run_id and type — and each type adds a few fields of its own. Payloads are intentionally minimal: they tell you what happened and where, not the full step output. To read a step's output, cost, or the run's final state, refetch GET /api/runs/{id}.

Event types

step_started

A node's step has begun. Emitted for every node the moment its wave launches — including nodes running concurrently in the same wave.

{ "run_id": "af772b3602ef", "type": "step_started", "node_id": "classify", "label": "Classify Ticket" }

step_completed

A node finished successfully and its checkpoint has been written. The frame carries only the node_id— refetch the run to read the step's output and cost.

{ "run_id": "af772b3602ef", "type": "step_completed", "node_id": "classify" }

step_failed

A node exhausted its retries and failed. This is followed by a run_failed for the run as a whole.

{ "run_id": "af772b3602ef", "type": "step_failed", "node_id": "draft_reply",
  "error": "Tool 'custom:score-lead' timed out after 60s" }

run_paused

The run reached a human gate and parked. In-flight branch work finishes first; then the run waits for POST /api/runs/{id}/approve. No step_completed is emitted for the gate node — this event is its signal.

{ "run_id": "af772b3602ef", "type": "run_paused", "node_id": "review",
  "prompt": "Review the drafted reply before it is sent." }

run_completed

Every branch finished and the run settled as success. Refetch the run for its final output, cost_usd, and duration_ms.

{ "run_id": "af772b3602ef", "type": "run_completed" }

run_failed

The run settled as failed — from a node failure (the error reads Node '…' failed: …), an unexpected crash, or cancellation. The last checkpoint is intact, so POST /api/runs/{id}/resume can pick up where it left off.

{ "run_id": "af772b3602ef", "type": "run_failed", "error": "Node 'Draft Reply' failed: Tool timed out after 60s" }

run_evaluated

Emitted immediately after run_completedwhen the workflow has evaluation checks. The full per-check results are on the run's evals field.

{ "run_id": "af772b3602ef", "type": "run_evaluated", "passed": 2, "total": 2 }

Ordering & guarantees

• A successful node emits step_started then step_completed; a failing node emits step_started then step_failed, and the run then emits run_failed.
• With parallel branches, events from different nodes interleave — several step_started frames can arrive before the first step_completed. Key on node_id, not arrival order.
• A run ends with exactly one terminal event: run_completed (optionally followed by run_evaluated), run_failed, or a park at run_paused. After an approval the run resumes and streams more events until it reaches another terminal state.

Delivery model

The EventBus is an in-process publish/subscribe map from run id to the set of connected sockets. It has two properties worth designing around:

• No replay. Events are delivered only to sockets connected at the moment they fire — there is no buffer or backlog. If you subscribe after an event fired, you never see it, and a very fast run can finish before your socket opens.
• Per process. The bus lives inside the backend process that is executing the run. Behind multiple workers, connect to the same instance that owns the run (see Self-hosting). A dead socket is dropped from the subscriber set on the next publish.

Recommended client pattern

Open the WebSocket, and on any event just refetch the run — the response is always complete and correct, which keeps client logic trivial. Fall back to short-interval polling if the socket errors or closes while the run is still active, and tear everything down once the run reaches a terminal state.

const TERMINAL = new Set(["success", "failed"]);

function followRun(runId: string, onUpdate: (run: Run) => void) {
  let closed = false;
  let poll: ReturnType<typeof setInterval> | null = null;

  async function refetch() {
    const run = await fetch(`http://localhost:8000/api/runs/${runId}`).then((r) => r.json());
    onUpdate(run);
    if (TERMINAL.has(run.status)) stop();          // gate is 'paused' — keep watching
  }

  function startPolling() {
    if (closed || poll) return;
    poll = setInterval(refetch, 3000);             // 3s fallback, like the console
  }

  const token = localStorage.getItem("ballast_token");        // required mode only
  const suffix = token ? "?token=" + encodeURIComponent(token) : "";
  const ws = new WebSocket(`ws://localhost:8000/api/runs/${runId}/stream${suffix}`);

  ws.onopen = () => void refetch();                // catch anything before we connected
  ws.onmessage = () => void refetch();             // any event → refetch (payload-agnostic)
  ws.onerror = startPolling;
  ws.onclose = startPolling;

  function stop() {
    closed = true;
    ws.onopen = ws.onmessage = ws.onerror = ws.onclose = null;
    ws.close();
    if (poll) clearInterval(poll);
  }
  return stop;                                      // call to unsubscribe
}

How the console uses it

The run detail page in the Ballast console follows this exact shape. While a run is active (pending, running, or paused) it opens a WebSocket to /api/runs/{id}/stream (carrying the session token in required mode). On any message it refetches the run and re-renders the step timeline. If the socket errors or closes, it falls back to polling the run every three seconds. The effect tears down on unmount and the moment the run reaches success or failed, so a finished run holds no open socket. A live indicator reflects whether the socket is currently connected.

Governance & policy events

The run event stream above tells you where a run is. A separate, first-class stream tells you what the governance layer decided while the run executed: an inspectable, auditable record of every execution-time policy decision. It is a polling REST endpoint — not this WebSocket — and it is distinct both from the run lifecycle events on this page and from the tamper-evident audit log (which records API mutations). Think of it as the policy-decision plane.

[
  { "id": "a1b2c3d4e5f6", "type": "data_redacted",
    "workspace_id": "0a11b2c3d4e5", "workflow_id": "9bc62f209c52", "version": 2,
    "run_id": "af772b3602ef", "node_id": "classify", "policy": "dlp",
    "actor_id": null, "decision": "redacted",
    "reason": "Redacted 2 value(s) before the prompt left the box",
    "evidence": { "fields": ["email", "credit_card"], "count": 2 },
    "created_at": "2026-07-05T14:10:05Z" },
  { "id": "b2c3d4e5f6a1", "type": "model_fallback_blocked",
    "workspace_id": "0a11b2c3d4e5", "workflow_id": "9bc62f209c52", "version": 2,
    "run_id": "af772b3602ef", "node_id": "draft_reply", "policy": "model_allowlist",
    "actor_id": null, "decision": "blocked",
    "reason": "Fallback model is not on the workspace allowlist",
    "evidence": { "model": "gpt-4o-mini", "provider": "openai" },
    "created_at": "2026-07-05T14:11:02Z" }
]

Event types

• policy_denied— a tool call was refused because the tool's workflow allowlist did not permit this workflow.
• tool_argument_blocked— a tool call passed an argument value on the tool's blocked-arguments list and was refused at run time.
• data_redacted — DLP scrubbed one or more values (named PII field types and/or redact patterns) from a prompt before it left the box.
• budget_blocked — a trigger or step was stopped because a block budget was exhausted (the caller sees 402).
• model_fallback_blocked — a fallback model was refused because it is not on the workspace model/provider allowlist.
• approval_override_granted — a budget was overridden with approval, lifting the per-run cap and resuming the run.

What each event carries

Every event names the decision and enough context to locate and explain it — but the evidence is redacted: it is limited to counts, ids, and model/provider names, and never includes the scrubbed plaintext or a secret.