Real-time Events over the Socket Gateway

ExecutionMode.SOCKET runs a task asynchronously but delivers its events pushed over the socket gateway instead of polled. It's the low-latency variant of async: progress, partial results, and custom events arrive as the worker emits them, and run() resolves on the terminal event.

What it adds over async polling

Async (ExecutionMode.ASYNC) polls the result endpoint on an interval. SOCKET mode delivers the same events, but:

• Lower latency — events arrive as they're produced, not on the next poll tick.
• Less load — no repeated GET .../result requests while the task runs.

A brief disconnect is handled for you — the gateway restores your subscription and replays the events you missed (see Reconnection). A disconnect longer than the recovery window can't be recovered, and the call ends with an error so it never hangs on a stream that's gone. If you need a run to survive arbitrarily long outages, ASYNC polling is the more forgiving choice.

Prerequisites: give the client a socket

The task events are user-scoped, so the client needs a socket connection authenticated as the user. Two tokens are involved, and they are different:

Token	Where	Purpose
App token (paat-…)	identityToken (HTTP)	Authorizes the workflow submit/result calls.
User OAuth token (JWT)	socket handshake auth.token	Authenticates the socket to the gateway.

You can either bring your own socket or let the client create and manage one:

import { WorkflowsClient } from '@picsart/workflows-client';
import { io } from 'socket.io-client';

// Option A — bring your own socket.io socket (you own its lifecycle):
const socket = io('https://api.picsart.com', {
    path: '/socket-gateway',
    transports: ['websocket'],
    auth: { token: `Bearer ${userOAuthToken}` }, // user JWT, not the app token
});
const workflows = new WorkflowsClient({ identityToken, socket });

// Option B — let the client create + own the socket (needs socket.io-client installed):
const workflows = new WorkflowsClient({
    identityToken,
    baseUrl: 'https://api.picsart.com/',
    socketConnection: {
        token: userOAuthToken,   // 'Bearer ' is added if missing
        // url defaults to the client's baseUrl; path to '/socket-gateway', transports to ['websocket']
    },
});

socket.io-client is an optional peer dependency — it is imported on demand only when you use socketConnection, so consumers that don't use sockets never load it.

When a socket/socketConnection is configured, the client opens the connection eagerly at construction, so it's ready before your first run/subscribe (no connect latency on first use).

Running a task with pushed events

import { ExecutionMode, WorkflowEvent } from '@picsart/workflows-client';

const { result, usage } = await workflows.run<MyResponseType>('example-task', params, {
    mode: ExecutionMode.SOCKET,
    onAccepted:      (taskId) => console.log('submitted:', taskId),
    onProgress:      (p)  => console.log('progress:', p),      // { percent, estimatedSecondsLeft? }
    onPartialResult: (pr) => console.log('partial:', pr),      // latest partial snapshot
    onEvent:         (e: WorkflowEvent) => console.log('event:', e), // custom flow.emitEvent() events
});

mode: ExecutionMode.SOCKET selects socket delivery. Like the other modes, a remote-settings executionMode (if one is configured for this workflow) overrides the mode you pass — so a workflow pinned to ASYNC/SYNC/STREAM in remote settings won't switch to socket delivery (see Settings Integration). The mode requires a socket or socketConnection — without one, run() rejects with an invalid_state error.

The callbacks match the other modes: onProgress for metrics, onPartialResult for partial result snapshots, and onEvent for custom events (the event. prefix is stripped from event.type). run() resolves with { result, usage } on completion and rejects on failure.

Watching an existing task

run submits and runs a task. subscribe watches work that's already running — a task submitted elsewhere, or before a page reload. It's shaped for observation rather than running: an async iterable you consume with for await, where each item is the raw gateway message { taskId, workflow?, type, payload }.

import { EventTypes } from '@picsart/workflows-client';

for await (const msg of workflows.subscribe({ name: 'example-task', taskId })) {
  switch (msg.type) {
    case EventTypes.METRICS:
      console.log('progress', msg.payload);
      break;
    case EventTypes.PARTIAL_RESULT:
      console.log('partial', msg.payload);
      break;
    case EventTypes.COMPLETED:
      console.log('done', msg.payload.result, msg.payload.usage);
      break;
    case EventTypes.FAILED:
      console.log('failed', msg.payload);
      break;
    default:
      console.log('custom event', msg.type, msg.payload); // 'event.<name>'
  }
}

• Scope — with a taskId the loop ends after that task's COMPLETED/FAILED; without one it watches every task of the workflow and keeps running (one task finishing doesn't stop it).
• Stopping — break the loop, or pass signal: anAbortSignal and abort it.
• Errors — an unrecoverable socket session makes the for await throw (try/catch to re-subscribe); a task failing is just a FAILED message, not a throw.
• Custom-event names keep the event. prefix (it's what separates them from lifecycle events in the single stream) — use msg.type.replace(/^event\./, '') for the bare name. run's onEvent strips it for you.

To poll /result instead of using the socket, use runPolling.

Closing the socket

await workflows.disconnect();

disconnect() closes a socket the client created from socketConnection. If you injected your own socket, it is left untouched — you own its lifecycle. Safe to call more than once.

Reconnection

The socket gateway has Connection State Recovery enabled. If the socket drops and reconnects within the recovery window (~2 minutes), the gateway transparently restores your subscription and replays every event you missed during the gap — progress, partial results, and custom events. You don't do anything; run() and subscribe() just keep delivering.

If the disconnect is longer than the window, the session can't be recovered — the events are gone and the client fails fast so nothing hangs:

• run() rejects with a 503 error.
• subscribe()'s for await loop throws a 503 error.

In both cases, re-run or re-subscribe to start a fresh stream.

Channels

You don't set channels directly — run and subscribe derive them from name/taskId. For reference, events are delivered on these gateway channels:

• workflows:all — every task of every workflow (firehose).
• workflows:<workflow>:all — every task of one workflow (subscribe({ name }), and what run uses).
• workflows:<workflow>:<taskId> — a single task (subscribe({ name, taskId })).