Воркфлоу

Долгие процессы с шагами, паузами, таймерами и возобновлением.

Воркфлоу хранит состояние долгого процесса на платформе. Он переживает перезапуск runtime и продолжает работу после внешнего события или таймера.

Используйте Conversation для одного хода диалога. Используйте Workflow, если процесс должен ждать, повторять сетевой вызов или выполняться по расписанию.

Объявите воркфлоу

src/workflows/delivery.ts

import { Workflow, z } from '@holocronlab/botruntime-runtime'

export const deliveryWorkflow = new Workflow({
  name: 'deliveryStatus',
  timeout: '24h',
  input: z.object({
    conversationId: z.string(),
    orderNumber: z.string(),
  }),
  requests: {
    deliveryAddress: z.object({ address: z.string().min(1) }),
  },
  output: z.object({
    status: z.string(),
  }),
  handler: async ({ input, step }) => {
    const address = await step.request(
      'deliveryAddress',
      'Уточните адрес доставки.',
    )

    const status = await step('check-delivery', async () => {
      return delivery.getStatus(input.orderNumber, address.address)
    })

    return { status }
  },
})

input, output, state, requests и notifications описываются zod-схемами. timeout принимает значения вроде 30s, 15m или 24h.

Поместите побочные эффекты в шаги

При возобновлении handler запускается с начала. Успешный step() возвращает сохранённый результат и не повторяет функцию.

const receipt = await step(
  'create-receipt',
  async () => billing.createReceipt(input.orderNumber),
  { maxAttempts: 3 },
)

Имя шага должно быть стабильным и уникальным внутри воркфлоу. Не помещайте сетевой вызов между шагами: при возобновлении он выполнится ещё раз.

По умолчанию шаг повторяет ошибку с ограниченным backoff. После исчерпания попыток воркфлоу завершается ошибкой.

Запросите данные у диалога

step.request(name, message) ставит процесс на паузу. Диалог получает событие workflow_request и отвечает через Workflow.provide().

src/conversations/index.ts

if (props.type === 'workflow_request') {
  state.pendingWorkflowEvent = props.event
  await props.conversation.send({
    type: 'text',
    payload: { text: props.event.payload.message },
  })
  return
}

После получения ответа:

await deliveryWorkflow.provide(state.pendingWorkflowEvent, {
  address: userAddress,
})
state.pendingWorkflowEvent = undefined

Данные проходят схему из requests. Невалидный ответ не возобновляет процесс.

step.notify(name, payload) отправляет типизированное уведомление без паузы.

Используйте таймеры

await step.sleep('retry-later', '15m')
await step.sleepUntil('send-tomorrow', tomorrowAtNine)

Таймер хранится на платформе. Перезапуск процесса не сбрасывает ожидание.

Запустите процесс

const workflow = await deliveryWorkflow.start({
  id: `delivery:${orderNumber}`,
  input: { conversationId, orderNumber },
})

Стабильный id помогает не создать второй экземпляр для той же операции. Используйте getOrCreate, если повторный запуск должен вернуть существующий процесс.

Воркфлоу можно передать агенту как инструмент:

await execute({
  instructions: 'Запускай проверку доставки после получения номера заказа.',
  tools: [deliveryWorkflow.asTool()],
})

Обработайте завершение

После завершения диалог получает workflow_callback.

if (props.type === 'workflow_callback') {
  const { status, output, error } = props.completion
  if (status !== 'completed') {
    console.error(`delivery workflow failed: ${error ?? status}`)
    return
  }

  await props.conversation.send({
    type: 'text',
    payload: { text: `Статус доставки: ${output.status}` },
  })
}

Обрабатывайте failed, canceled и timed_out. Не оставляйте сбой без технического сигнала и понятного ответа пользователю.

Запуск по расписанию

Поле schedule принимает пятичастное cron-выражение в UTC.

export default new Workflow({
  name: 'dailyCatalogSync',
  schedule: '0 6 * * *',
  input: z.object({}),
  output: z.object({ updated: z.number() }),
  handler: async ({ step }) => {
    const updated = await step('sync', syncCatalog)
    return { updated }
  },
})

Scheduled workflow всегда запускается с пустым input. Поэтому его схема не может содержать обязательные поля.

Расписание — durable часть deploy-контракта:

deploy создаёт или обновляет producer для каждой декларации schedule;
следующий запуск считается от запланированного cron-slot, поэтому задержка worker-а не сдвигает последующие запуски;
перезапуск cloudapi/runtime не удаляет расписание;
доставка каждого slot — at-least-once, поэтому используйте стабильные имена шагов и idempotency key для внешних побочных эффектов;
удаление schedule из декларации при следующем deploy удаляет producer.

Полная длительность scheduled workflow не ограничена одним runtime invocation. Для больших синхронизаций делите работу на несколько step() — подробнее в гарантиях runtime.