Streaming Updates

createTask / Task returns an object that supports streaming updates as the task executes.

Iterating Updates

TypeScript
Python

createTask is both a Promise and an AsyncIterable. Use for await to stream updates:

import { createTask } from '@requence/task'

const task = createTask({
  taskTemplate: 'my-template',
  input: { data: [1, 2, 3] },
})

for await (const update of task) {
  switch (update.type) {
    case 'taskStart':
      console.log(`Task ${update.taskId} started`)
      break
    case 'nodeStart':
      console.log(`Node ${update.node.alias ?? update.node.id} started`)
      break
    case 'nodeUpdate':
      console.log(`Node output:`, update.data)
      break
    case 'nodeError':
      console.log(`Node error:`, update.error)
      break
    case 'nodeEnd':
      console.log(`Node ${update.node.alias ?? update.node.id} finished`)
      break
    case 'taskEnd':
      console.log(`Task completed:`, update.context.result)
      break
    case 'taskError':
      console.log(`Task failed:`, update.reason)
      break
    case 'taskAborted':
      console.log(`Task aborted:`, update.reason)
      break
  }
}

Use the sync.updates generator for synchronous iteration:

from requence.task import Task

task = Task(
    task_template="my-template",
    input={"data": [1, 2, 3]},
)

for update in task.sync.updates:
    match update["type"]:
        case "taskStart":
            print(f"Task {update['taskId']} started")
        case "nodeStart":
            print(f"Node {update['node'].get('alias', update['node']['id'])} started")
        case "nodeUpdate":
            print(f"Node output: {update['data']}")
        case "nodeError":
            print(f"Node error: {update['error']}")
        case "taskEnd":
            print(f"Task completed: {update['context'].result}")
        case "taskError":
            print(f"Task failed: {update.get('reason')}")
        case "taskAborted":
            print(f"Task aborted: {update.get('reason')}")

For async usage, iterate over task.updates:

async for update in task.updates:
    print(update["type"])

Using a Callback (TypeScript)

You can also provide an onUpdate callback as the second argument:

const result = await createTask(
  {
    taskTemplate: 'my-template',
    input: {},
  },
  (update) => {
    console.log(`[${update.type}]`, update.timestamp)
  },
)

This approach lets you stream updates while still awaiting the final result.

Update Types

| Type | Description | | ------------- | ---------------------------------------------------------------------- | | taskStart | Task execution has begun. Contains input and taskId. | | nodeStart | A node has started processing. Contains node info (id, type, alias). | | nodeUpdate | A node has produced output. Contains data. | | nodeError | A node encountered an error. Contains error message. | | nodeDefer | A node has been deferred (e.g., waiting for retry). | | nodeEnd | A node has finished processing. | | taskEnd | The entire task completed successfully. Contains final result. | | taskError | The task failed. Contains reason. | | taskAborted | The task was aborted. Contains reason. |

Context on Updates

Every update includes a context object with the current state of the task:

TypeScript
Python

update.context.input // The task input
update.context.taskId // The task ID
update.context.result // Current accumulated result (partial until taskEnd)
update.context.getNodeData(alias) // Output from a specific node
update.context.getNodeError(alias) // Error from a specific node

update["context"].input # The task input
update["context"].task_id # The task ID
update["context"].result # Current accumulated result
update["context"].get_node_data(alias) # Output from a specific node
update["context"].get_node_error(alias) # Error from a specific node