Skip to main content

Workflow Progress

During workflow execution, you can provide real-time progress information to the client application. The framework supports three distinct types of progress updates to keep clients informed about the task execution status:

Progress Metrics​

Currently available via Async API​

Progress metrics allow you to report completion percentage and estimated remaining time for a task. This is particularly useful for long-running operations where clients need to track overall progress.


@Executor("example-task")
export class ExampleExecutor implements TaskExecutor {
    async processTask(command: any, metadata: TaskMetadata, flow: Flow): Promise<ExampleResponse> {
        await flow.emitProgressMetrics({
            percent: 27, // progress percentage (0 to 100)
            estimatedSecondsLeft: 14, // optional remaining time in seconds
        });
        sleep(5000);
        return {
            name: 'Joe',
            surname: `Doe`
        }
    }
}

For tasks with predictable durations, you can automatically schedule progress updates based on the average execution time:


@Executor("example-task")
export class ExampleExecutor implements TaskExecutor {
    async processTask(command: any, metadata: TaskMetadata, flow: Flow): Promise<ExampleResponse> {
        await flow.scheduleEstimatedProgressMetrics(60_000);
        sleep(5000);
        return {
            name: 'Joe',
            surname: `Doe`
        }
    }
}

Partial Result​

Currently available via Async API​

Partial results enable you to send intermediate results to the client during task execution. These results should be a subset of your final result schema, allowing clients to start processing data before the task completes.

Important: Only the most recently emitted partial result is accessible to clients during polling.


@Executor("example-task")
export class ExampleExecutor implements TaskExecutor {
    async processTask(command: any, metadata: TaskMetadata, flow: Flow): Promise<ExampleResponse> {
        await flow.emitPartialResult({
            name: 'Joe'
        });
        sleep(5000);
        return {
            name: 'Joe',
            surname: `Doe`
        }
    }
}

Events​

Available via Stream API and Async API​

Events provide a way to send structured updates about specific occurrences during workflow execution. Unlike partial results, events are accumulated and can be used to communicate various execution milestones or state changes.

  • Stream API: Events are streamed to the client in real-time as they are emitted.
  • Async API: Events are stored and returned as part of the events array when polling the task result. The framework assigns a unique id to each event for deduplication across polls.
note

Async API support for events requires @picsart/pa-pluggable-workers-core version 6.15.0 or higher.

When using the workflows client, events can be consumed via the onEvent callback passed to workflows.run() in both Stream and Async modes. In Async mode, the client uses the ids to deduplicate events across polls; without the client, you must track seen ids yourself.

Always await emitEvent

flow.emitEvent() returns a Promise. You must await each call to guarantee that events are delivered in order and that all events are included in the final result.


@Executor("example-task")
export class ExampleExecutor implements TaskExecutor {
    async processTask(command: any, metadata: TaskMetadata, flow: Flow): Promise<ExampleResponse> {
        await flow.emitEvent({
            type: 'name.generated',
            data: {
                name: 'Joe'
            }
        });
        sleep(5000);
        await flow.emitEvent({
            type: 'surname.generated',
            data: {
                surname: 'Doe'
            }
        });
        return {
            name: 'Joe',
            surname: `Doe`
        }
    }
}

Declaring event schemas with @EventModel​

Use @EventModel("<task-name>") to declare the schema of an event your task emits. Apply it once per event variant — the registered models are published in the OpenAPI schema for the task's stream endpoint and the registered type is used to stamp emitted events automatically.

@EventModel("example-task", { type: "name.generated" })
export class NameGeneratedEvent extends TaskEvent {
    @ApiProperty()
    public progress: number;
}

When you call flow.emitEvent() without a type, the framework stamps the registered type onto the event, so you don't have to repeat it:

// "type" is set to "name.generated" automatically
await flow.emitEvent({
    progress: 20
});
note

A type registered via @EventModel is only stamped when the emitted event has no type of its own — a caller-provided type is always kept. If a task registers more than one event type, none is stamped automatically, so you must set the type explicitly on each emitEvent call.

Real-time delivery over the Socket Gateway​

All three updates above — progress metrics, partial results, and events — plus the task's terminal result (completed/failed) can be delivered to clients pushed in real time over the socket gateway, in addition to the usual Async polling and Stream delivery.

The same calls documented above — flow.emitProgressMetrics(), flow.emitPartialResult(), flow.emitEvent() — and the task's completion fan out to the gateway automatically once it's enabled. To turn it on, set socketGateway.enabled on the task definition (see Register Tasks → Socket Gateway Config):

tasks:
  - name: example-task
    socketGateway:
      enabled: true

Events are published only when all of these hold:

  • socketGateway.enabled: true for the task,
  • the task carries a userId — the events are user-scoped, so an anonymous task publishes nothing, and
  • it is the root task — the top-level task the caller submitted. Sub-tasks spawned during execution publish nothing to the socket gateway.

Delivery is best-effort and independent of the worker's own result: a failed publish is logged and swallowed, never failing the task. On the client, consume these with ExecutionMode.SOCKET / subscribe().