跳到主要内容
非官方简体中文翻译，内容版权归 Mastra 所有，遵循 Apache-2.0 许可。

Run.timeTravelStream()

The .timeTravelStream() method re-executes a workflow starting from any specific step with streaming events. You can receive real-time updates during time travel execution while maintaining full visibility into each step's progress.

Usage example
Usage example的直接链接

const run = await workflow.createRun()

const output = run.timeTravelStream({
  step: 'step2',
  inputData: { value: 10 },
})

// Process events as they arrive
for await (const event of output.fullStream) {
  console.log(event.type, event.payload)
}

// Get the final result
const result = await output.result

Parameters
Parameters的直接链接

All parameters are the same as Run.timeTravel(). See the timeTravel reference for detailed parameter documentation.

Returns
Returns的直接链接

output:

WorkflowRunOutput<WorkflowResult<TState, TInput, TOutput, TSteps>>
An object containing both the stream and result promise

output.fullStream:

ReadableStream<WorkflowStreamEvent>
A readable stream that emits workflow events as execution progresses

output.result:

Promise<WorkflowResult<TState, TInput, TOutput, TSteps>>
A promise that resolves to the final workflow execution result

output.runId:

string
The ID of the workflow run

output.workflowId:

string
The ID of the workflow

Stream events
Stream events的直接链接

The stream emits workflow events during execution:

  • workflow-step-start: Emitted when a step begins execution
  • workflow-step-result: Emitted when a step produces a result
  • workflow-step-error: Emitted when a step encounters an error
  • workflow-step-suspended: Emitted when a step suspends
  • Additional events depending on step types (agents, tools, etc.)

Extended usage examples
Extended usage examples的直接链接

Processing events during time travel
Processing events during time travel的直接链接

const run = await workflow.createRun()

const output = run.timeTravelStream({
  step: 'step2',
  inputData: { value: 10 },
})

for await (const event of output.fullStream) {
  switch (event.type) {
    case 'workflow-step-start':
      console.log(`Starting step: ${event.payload.stepName}`)
      break
    case 'workflow-step-result':
      console.log(`Completed step: ${event.payload.id}`)
      break
    case 'workflow-step-error':
      console.error(`Error in step: ${event.payload.stepName}`, event.payload.error)
      break
  }
}

const result = await output.result
console.log('Time travel completed:', result)

Time travel stream with context
Time travel stream with context的直接链接

const output = run.timeTravelStream({
  step: 'step2',
  context: {
    step1: {
      status: 'success',
      payload: { value: 0 },
      output: { step1Result: 2 },
      startedAt: Date.now(),
      endedAt: Date.now(),
    },
  },
})

for await (const event of output.fullStream) {
  // Handle events
  console.log(event)
}

const result = await output.result

Time travel stream with nested workflows
Time travel stream with nested workflows的直接链接

const output = run.timeTravelStream({
  step: ['nestedWorkflow', 'step3'],
  inputData: { value: 10 },
  nestedStepsContext: {
    nestedWorkflow: {
      step2: {
        status: 'success',
        payload: { step1Result: 2 },
        output: { step2Result: 3 },
        startedAt: Date.now(),
        endedAt: Date.now(),
      },
    },
  },
})

for await (const event of output.fullStream) {
  console.log(event.type, event.payload)
}

const result = await output.result

Notes
Notes的直接链接

  • The stream closes automatically when time travel execution completes or encounters an error
  • You can process events from the stream while the workflow is still executing
  • The result promise resolves only after all steps have completed
  • Stream events follow the same format as regular workflow streaming
  • Time travel streaming requires storage to be configured since it relies on persisted workflow snapshots
On this page