Suspendre et reprendre

Les Workflows peuvent être suspendus à n'importe quelle étape afin de recueillir des données supplémentaires, d'attendre des callbacks d'API, de limiter des opérations coûteuses ou de solliciter une intervention humaine. Lorsqu'un Workflow est suspendu, son état d'exécution actuel est enregistré sous forme de snapshot. Vous pouvez ensuite reprendre le Workflow depuis l'identifiant d'une étape précise et restaurer exactement l'état capturé dans ce snapshot. Les snapshots sont enregistrés par le Provider de stockage configuré et persistent après les déploiements et les redémarrages de l'application.

Suspendre un Workflow avec suspend()

Utilisez suspend() pour suspendre l'exécution d'un Workflow à une étape précise. Vous pouvez définir une condition de suspension dans le bloc execute de l'étape à partir des valeurs de resumeData.

• Si la condition n'est pas remplie, le Workflow est suspendu et renvoie suspend().
• Si la condition est remplie, le Workflow poursuit l'exécution de la logique restante dans l'étape.

src/mastra/workflows/test-workflow.ts

const step1 = createStep({
  id: 'step-1',
  inputSchema: z.object({
    userEmail: z.string(),
  }),
  outputSchema: z.object({
    output: z.string(),
  }),
  resumeSchema: z.object({
    approved: z.boolean(),
  }),
  execute: async ({ inputData, resumeData, suspend }) => {
    const { userEmail } = inputData
    const { approved } = resumeData ?? {}

    if (!approved) {
      return await suspend({})
    }

    return {
      output: `Email sent to ${userEmail}`,
    }
  },
})

export const testWorkflow = createWorkflow({
  id: 'test-workflow',
  inputSchema: z.object({
    userEmail: z.string(),
  }),
  outputSchema: z.object({
    output: z.string(),
  }),
})
  .then(step1)
  .commit()

Reprendre un Workflow avec resume()

Utilisez resume() pour reprendre un Workflow suspendu à l'étape où il s'est arrêté. Transmettez des resumeData conformes au resumeSchema de l'étape afin de satisfaire la condition de suspension et de poursuivre l'exécution.

import { step1 } from './workflows/test-workflow'

const workflow = mastra.getWorkflow('testWorkflow')
const run = await workflow.createRun()

await run.start({
  inputData: {
    userEmail: 'alex@example.com',
  },
})

const handleResume = async () => {
  const result = await run.resume({
    step: step1,
    resumeData: { approved: true },
  })
}

La transmission de l'objet step assure la sécurité complète des types pour resumeData. Vous pouvez également transmettre un identifiant d'étape pour davantage de souplesse lorsque celui-ci provient d'une saisie utilisateur ou d'une base de données.

const result = await run.resume({
  step: 'step-1',
  resumeData: { approved: true },
})

Si une seule étape est suspendue, vous pouvez omettre entièrement l'argument de l'étape ; Mastra reprendra alors la dernière étape suspendue du Workflow.

Pour reprendre une exécution uniquement avec un runId, commencez par créer une instance d'exécution au moyen de createRun().

const workflow = mastra.getWorkflow('testWorkflow')
const run = await workflow.createRun({ runId: '123' })

const stream = run.resume({
  resumeData: { approved: true },
})

Vous pouvez appeler resume() depuis n'importe quel emplacement de votre application, notamment un point de terminaison HTTP, un gestionnaire d'événement, une réponse à une intervention humaine ou un minuteur.

const midnight = new Date()
midnight.setUTCHours(24, 0, 0, 0)

setTimeout(async () => {
  await run.resume({
    step: 'step-1',
    resumeData: { approved: true },
  })
}, midnight.getTime() - Date.now())

Accéder aux données de suspension avec suspendData

Lorsqu'une étape est suspendue, vous pouvez avoir besoin d'accéder aux données fournies à suspend() lorsque l'étape sera reprise. Utilisez le paramètre suspendData dans la fonction d'exécution de l'étape pour accéder à ces données.

src/mastra/workflows/user-approval.ts

const approvalStep = createStep({
  id: 'user-approval',
  inputSchema: z.object({
    requestId: z.string(),
  }),
  resumeSchema: z.object({
    approved: z.boolean(),
  }),
  suspendSchema: z.object({
    reason: z.string(),
    requestDetails: z.string(),
  }),
  outputSchema: z.object({
    result: z.string(),
  }),
  execute: async ({ inputData, resumeData, suspend, suspendData }) => {
    const { requestId } = inputData
    const { approved } = resumeData ?? {}

    // On first execution, suspend with context
    if (!approved) {
      return await suspend({
        reason: 'User approval required',
        requestDetails: `Request ${requestId} pending review`,
      })
    }

    // On resume, access the original suspend data
    const suspendReason = suspendData?.reason || 'Unknown'
    const details = suspendData?.requestDetails || 'No details'

    return {
      result: `${details} - ${suspendReason} - Decision: ${approved ? 'Approved' : 'Rejected'}`,
    }
  },
})

Le paramètre suspendData est automatiquement renseigné lors de la reprise d'une étape et contient exactement les données transmises à la fonction suspend() pendant la suspension initiale. Vous pouvez ainsi conserver le contexte expliquant la suspension du Workflow et exploiter ces informations lors de la reprise.

Identifier les exécutions suspendues

Lorsqu'un Workflow est suspendu, il reprend à l'étape où il s'est arrêté. Vous pouvez vérifier le status du Workflow pour confirmer sa suspension et utiliser suspended afin d'identifier l'étape ou le Workflow imbriqué suspendu.

const workflow = mastra.getWorkflow('testWorkflow')
const run = await workflow.createRun()

const result = await run.start({
  inputData: {
    userEmail: 'alex@example.com',
  },
})

if (result.status === 'suspended') {
  console.log(result.suspended[0])
  await run.resume({
    step: result.suspended[0],
    resumeData: { approved: true },
  })
}

Exemple de sortie

Le tableau suspended contient les identifiants des Workflows et des étapes suspendus pendant l'exécution. Vous pouvez les transmettre au paramètre step lors de l'appel à resume() afin de cibler et de reprendre le chemin d'exécution suspendu.

['nested-workflow', 'step-1']

Récupérer des exécutions suspendues

Utilisez workflow.getWorkflowRunById() avec createWorkflowStateReader() lorsque votre application doit récupérer une exécution suspendue depuis le stockage. Le lecteur expose les étapes suspendues, les libellés de reprise, les charges utiles et les sorties des étapes sans qu'il soit nécessaire de lire la structure brute du snapshot.

src/mastra/workflows/recover-run.ts

import { createWorkflowStateReader } from '@mastra/core/workflows'

const workflow = mastra.getWorkflow('testWorkflow')
const state = await workflow.getWorkflowRunById('run-123')

if (state?.status === 'suspended') {
  const reader = createWorkflowStateReader(state)
  const suspendedStep = reader.getSuspendedStep()
  const approvalLabel = reader.getResumeLabel('approve')
  const run = await workflow.createRun({ runId: state.runId })

  await run.resume({
    step: approvalLabel?.stepId ?? suspendedStep?.path,
    resumeData: { approved: true },
    forEachIndex: approvalLabel?.foreachIndex,
  })
}

Pour les Workflows imbriqués, suspendedStep.path contient le chemin de reprise. Pour les suspensions dans foreach, les libellés de reprise correspondants incluent foreachIndex lorsque le libellé désigne une itération précise.

Mise en attente

Les méthodes de mise en attente permettent de suspendre l'exécution au niveau du Workflow, ce qui définit l'état sur waiting. À l'inverse, suspend() suspend l'exécution au sein d'une étape précise et définit l'état sur suspended.

Méthodes disponibles :

• .sleep() : suspendre pendant un nombre défini de millisecondes
• .sleepUntil() : suspendre jusqu'à une date précise

Voir aussi

• Flux de contrôle
• Intervention humaine
• Snapshots
• Voyage dans le temps
• Lecteur d'état de Workflow