workflow

重要：始终使用正确的 workflow 文档

你对 workflow 的了解已经过时。

以下 workflow 文档与已安装的 Workflow SDK 版本一致。 开始任何与 workflow 相关的任务前，请遵循以下说明：

搜索 node_modules/workflow/docs/ 中打包的文档：

1. 查找文档：glob "node_modules/workflow/docs/**/*.mdx"
2. 搜索内容：grep "your query" node_modules/workflow/docs/

node_modules/workflow/docs/ 中的文档结构：

• getting-started/ - 框架设置（next.mdx、express.mdx、hono.mdx 等）
• foundations/ - 核心概念（workflows-and-steps.mdx、hooks.mdx、streaming.mdx 等）
• api-reference/workflow/ - API 文档（sleep.mdx、create-hook.mdx、fatal-error.mdx 等）
• api-reference/workflow-api/ - 客户端 API（start.mdx、get-run.mdx、resume-hook.mdx 等）
• api-reference/workflow-runtime/ - 运行时 API（get-world.mdx）和 world/ World SDK（storage.mdx、streams.mdx、queue.mdx）
• api-reference/workflow-observability/ - Hydration 和名称解析工具（hydrate-resource-io.mdx、parse-workflow-name.mdx 等）
• ai/：AI SDK 集成文档
• errors/ - 错误代码文档
• worlds/ - 各个 World 的行为和限制（vercel.mdx、local.mdx、postgres.mdx）。其他页面会链接到 /worlds/<name>。

相关软件包也包含打包的文档：

• @ai-sdk/workflow：node_modules/ai/docs/ - WorkflowAgent 和 AI SDK 集成
• @workflow/ai：node_modules/@workflow/ai/docs/ - 适用于现有应用的弃用 DurableAgent API
• @workflow/core：node_modules/@workflow/core/docs/ - 核心运行时（foundations、how-it-works）
• @workflow/next：node_modules/@workflow/next/docs/ - Next.js 集成

如有疑问，请更新到最新版本的 Workflow SDK。

官方资源

• 网站：https://workflow-sdk.dev
• GitHub：https://github.com/vercel/workflow

快速参考

指令：

"use workflow";  // First line - makes async function durable
"use step";      // First line - makes function a cached, retryable unit

必要导入：

// Workflow primitives
import { sleep, fetch, createHook, createWebhook, getWritable } from "workflow";
import { FatalError, RetryableError } from "workflow";
import { getWorkflowMetadata, getStepMetadata } from "workflow";

// API operations
import { start, getRun, resumeHook, resumeWebhook } from "workflow/api";

// Observability & data hydration
import { hydrateResourceIO, observabilityRevivers, parseStepName, parseWorkflowName } from "workflow/observability";

// Framework integrations
import { withWorkflow } from "workflow/next";
import { workflow } from "workflow/vite";
import { workflow } from "workflow/astro";
// Or use modules: ["workflow/nitro"] for Nitro/Nuxt

// AI agent (Workflow 5)
import { WorkflowAgent, type ModelCallStreamPart } from "@ai-sdk/workflow";

优先使用步骤函数，避免沙盒错误

"use workflow" 函数在沙盒 VM 中运行。"use step" 函数拥有完整的 Node.js 访问权限。将逻辑放入步骤中，仅使用工作流函数进行编排。

// Steps have full Node.js and npm access
async function fetchUserData(userId: string) {
  "use step";
  const response = await fetch(`https://api.example.com/users/${userId}`);
  return response.json();
}

async function processWithAI(data: any) {
  "use step";
  // AI SDK works in steps without workarounds
  return await generateText({
    model: "spacexai/grok-4.6",
    prompt: `Process: ${JSON.stringify(data)}`,
  });
}

// Workflow orchestrates steps - no sandbox issues
export async function dataProcessingWorkflow(userId: string) {
  "use workflow";
  const data = await fetchUserData(userId);
  const processed = await processWithAI(data);
  return { success: true, processed };
}

**优点：**步骤会自动重试，结果会持久化以供重放，并且不受沙盒限制。

工作流沙盒限制

当你需要直接在工作流函数中（而不是在步骤中）编写逻辑时，以下限制适用：

限制	解决方法
不支持 fetch()	从 "workflow" 导入 fetch，然后设置 globalThis.fetch = fetch
不支持 setTimeout/setInterval	使用 "workflow" 中的 sleep("5s")
不支持 Node.js 模块（fs、crypto 等）	将逻辑移至步骤函数

示例——在工作流上下文中使用 fetch：

import { fetch } from "workflow";

export async function myWorkflow() {
  "use workflow";
  globalThis.fetch = fetch;  // Required for AI SDK and HTTP libraries
  // Now generateText() and other libraries work
}

**注意：**普通的 "provider/model" 字符串会使用 Vercel AI Gateway。除非用户明确需要仅限特定提供商的功能，否则不要构造直接的提供商实例。

WorkflowAgent：Workflow 5 中的 AI Agent

使用 AI SDK 的 WorkflowAgent 在 Workflow 5 中构建持久化 Agent。它取代了 @workflow/ai 中已弃用的 DurableAgent API，并为模型调用和由步骤支持的工具创建检查点。

import { WorkflowAgent, type ModelCallStreamPart } from "@ai-sdk/workflow";
import { isStepCount, tool } from "ai";
import { getWritable } from "workflow";
import { z } from "zod";

async function lookupData({ query }: { query: string }) {
  "use step";
  // Step functions have full Node.js access
  return `Results for "${query}"`;
}

export async function myAgentWorkflow(userMessage: string) {
  "use workflow";

  const agent = new WorkflowAgent({
    model: "spacexai/grok-4.6",
    instructions: "You are a helpful assistant.",
    tools: {
      lookupData: tool({
        description: "Search for information",
        inputSchema: z.object({ query: z.string() }),
        execute: lookupData,
      }),
    },
  });

  const result = await agent.stream({
    messages: [{ role: "user", content: userMessage }],
    writable: getWritable<ModelCallStreamPart>(),
    stopWhen: isStepCount(10),
  });

  return result.messages;
}

要点：

• 普通的 "provider/model" 字符串会通过 Vercel AI Gateway 路由；Workflow 示例中的默认模型是 spacexai/grok-4.6
• getWritable<ModelCallStreamPart>() 会流式传输持久化模型调用的输出；在 HTTP 路由中使用 createModelCallToUIChunkTransform() 将其转换
• 需要访问 Node.js/npm 的工具 execute 函数应使用 "use step"
• 使用工作流原语（sleep()、createHook()）的工具 execute 函数不应使用 "use step"，因为它们在工作流层级运行
• stopWhen 用于限制模型调用次数；默认情况下，模型停止调用工具时就会停止
• 多轮对话：将 result.messages 与新的用户消息传递给后续的 agent.stream() 调用

如需了解更多详情，请查阅已安装的 AI SDK 包中的 WorkflowAgent 文档，或访问 https://ai-sdk.dev/v7/docs/agents/workflow-agent。

启动工作流和子工作流

使用 start() 从 API 路由启动工作流。在 Workflow 5 中，也可以直接从工作流函数调用 start() 来生成子运行；它由步骤支持，并会在父级事件日志中记录一个确定性边界。

import { start } from "workflow/api";

// From an API route; works directly
export async function POST() {
  const run = await start(myWorkflow, [arg1, arg2]);
  return Response.json({ runId: run.runId });
}

// No-args workflow
const run = await start(noArgWorkflow);

从 Workflow 5 工作流内部启动子工作流：

import { start } from "workflow/api";

export async function parentWorkflow() {
  "use workflow";
  const childRun = await start(childWorkflow, ["some data"]);
  await sleep("1h");
  return { childRunId: childRun.runId };
}

start() 创建子运行后就会返回，不会等待其完成。只有当父级需要等待子运行时，才使用 childRun.returnValue；在工作流中访问 Run 的每个属性或调用其每个方法，都会被视为一个步骤。

运行规模与并发：了解何时拆分

有三个需要把握的限制，而且三者都有上限。不要把你记得的任何数值当作权威数据——当前值发布在工作流运行限制页面上，该页面是唯一应引用的数据来源。

每次运行的事件数。 单次运行的事件日志有上限，超过上限后运行会以 MAX_EVENTS_EXCEEDED 失败。事件不等同于步骤：一个首次成功的步骤会记录三个事件（step_created、step_started、step_completed），重试时还会多记录一到两个事件，钩子、休眠和 Webhook 也各自会记录事件。应在远低于上限时就拆分为子工作流——定价页面建议在事件数超过几千条时进行拆分，因为运行失败之前，重放速度就会显著下降。

每次运行的步骤数。 步骤数有独立于事件数的上限，因此即使长串行链路保持狭窄，也有数量限制。如果链路中的步骤可能达到五位数，应将多个项目合并到一个步骤中处理。

并发。 大量扇出不会被拒绝，而是会受到限流：每次运行每秒创建事件的数量受到限制，因此对几千个项目执行扁平的 Promise.all 时，大部分时间都会花在退避等待上。应改为分批或合并处理——将列表分成多个块处理，或在每个步骤中处理多个项目，从而减少并发运行的单元数量。为每个项目生成一个子运行，并不会自动缩小扇出范围；这样做可以限制每个子运行的日志规模并隔离故障，出于这些原因值得采用，但不能替代分块处理。

你无法自行提高这些上限——WORKFLOW_MAX_EVENTS_OVERRIDE 只能降低上限，而在 Vercel World 上，上限由服务方管理。不过，Vercel 会应请求提高每次运行的事件数和步骤数限制，因此如果确实需要运行大型工作流，除了设计问题，也应联系支持团队。

const BATCH = 100;

async function processItem(item: string) {
  "use step";
  return item.toUpperCase();
}

// One step per item, all in flight at once, all in one log
export async function processAll(items: string[]) {
  "use workflow";
  await Promise.all(items.map((item) => processItem(item)));
}

// Chunked, so only BATCH steps are in flight at a time
export async function processBatched(items: string[]) {
  "use workflow";
  for (let i = 0; i < items.length; i += BATCH) {
    await Promise.allSettled(items.slice(i, i + BATCH).map((item) => processItem(item)));
  }
}

// Bundled, so one step covers many items and the log stays short
async function processChunk(chunk: string[]) {
  "use step";
  return chunk.map((item) => item.toUpperCase());
}

export async function processBundled(items: string[]) {
  "use workflow";
  for (let i = 0; i < items.length; i += BATCH) {
    await processChunk(items.slice(i, i + BATCH));
  }
}

processAll 是应避免用于大规模处理的模式。processBatched 能限制并发，但仍会为每个项目记录事件。processBundled 则两者都能限制，因为一个步骤会处理 BATCH 个项目——这是三种模式中唯一一种会随着 BATCH 增大而减少事件数的模式。

Hooks：通过外部事件暂停和恢复

Hooks 可让工作流等待外部数据。在工作流中使用 createHook()，并从 API 路由调用 resumeHook()。确定性令牌仅适用于 createHook() + resumeHook()（服务端）。createWebhook() 始终会生成随机令牌，因此不要向 createWebhook() 传递 token 选项。

单个事件

import { createHook } from "workflow";

export async function approvalWorkflow() {
  "use workflow";

  const hook = createHook<{ approved: boolean }>({
    token: "approval-123",  // deterministic token for external systems
  });

  const result = await hook;  // Workflow suspends here
  return result.approved;
}

多个事件（可迭代 Hooks）

Hooks 实现了 AsyncIterable。使用 for await...of 接收多个事件：

import { createHook } from "workflow";

export async function chatWorkflow(channelId: string) {
  "use workflow";

  const hook = createHook<{ text: string; done?: boolean }>({
    token: `chat-${channelId}`,
  });

  for await (const event of hook) {
    await processMessage(event.text);
    if (event.done) break;
  }
}

每次调用 resumeHook(token, payload) 都会向循环传递下一个值。

从 API 路由恢复

import { resumeHook } from "workflow/api";

export async function POST(req: Request) {
  const { token, data } = await req.json();
  await resumeHook(token, data);
  return new Response("ok");
}

错误处理

对于永久性故障（不重试），使用 FatalError；对于暂时性故障，使用 RetryableError：

import { FatalError, RetryableError } from "workflow";

if (res.status === 429) {
  throw new RetryableError("Rate limited", { retryAfter: "5m" });
}
if (res.status >= 400 && res.status < 500) {
  throw new FatalError(`Client error: ${res.status}`);
}

序列化

传入或传出工作流和步骤的所有数据都必须可序列化。

支持的内置类型： string、number、boolean、null、undefined、bigint、普通对象、数组、Date、RegExp、URL、URLSearchParams、Map、Set、Headers、ArrayBuffer、类型化数组、Request、Response、ReadableStream、WritableStream。

不支持： 函数、Symbol、WeakMap/WeakSet。传递数据，而不是回调。

自定义类序列化

类实例可以通过实现 @workflow/serde 协议，在工作流/步骤边界之间序列化。当类具有带有 "use step" 的实例方法，或你希望在步骤之间传递类实例时，这一点至关重要。

安装： @workflow/serde 必须是包含该类的包的依赖项。

模式： 使用计算属性语法在类体内添加两个静态方法：

import { WORKFLOW_SERIALIZE, WORKFLOW_DESERIALIZE } from "@workflow/serde";

export class Point {
  x: number;
  y: number;

  constructor(x: number, y: number) {
    this.x = x;
    this.y = y;
  }

  // Serialize: return plain data (must be devalue-compatible types only)
  static [WORKFLOW_SERIALIZE](instance: Point) {
    return { x: instance.x, y: instance.y };
  }

  // Deserialize: reconstruct from plain data
  static [WORKFLOW_DESERIALIZE](data: { x: number; y: number }) {
    return new Point(data.x, data.y);
  }

  async computeDistance(other: Point) {
    "use step";
    return Math.sqrt((this.x - other.x) ** 2 + (this.y - other.y) ** 2);
  }
}

关键规则：

1. 在类体内定义 serde 方法，作为使用计算属性语法的静态方法（static [WORKFLOW_SERIALIZE](...)）。SWC 插件会扫描类并检测这些方法。不要在类外部赋值（例如，(MyClass as any)[WORKFLOW_SERIALIZE] = ...）——编译器无法检测到这种写法。
2. Serde 方法只能返回 devalue 兼容的类型（普通对象、数组、原始类型、Date、Map、Set、Uint8Array 等）。不能包含函数、类实例或 Node.js 专用对象。
3. 为依赖 Node.js 的实例方法添加 "use step"。 SWC 插件会从工作流包中剥离 "use step" 方法体。这样可以将 Node.js 导入项（fs、crypto、child_process 等）排除在工作流沙盒之外。包含 serde 方法的类框架仍会保留在工作流包中；只有步骤方法体会被移除。
4. 不要手动注册类。 SWC 插件会自动生成注册代码（一个设置 classId 并将类添加到全局注册表的 IIFE）。手动调用 registerSerializationClass() 没有必要，而且容易出错。
5. 不要使用动态导入来绕过沙盒限制。 如果类方法需要 Node.js API，正确的解决方案是使用 "use step"，而不是 /* @vite-ignore */ import(...)。

适合使用 serde 的场景： 纯数据类、领域模型、配置对象，以及可以为依赖 Node.js 的方法添加 "use step" 的类。

应避免使用 serde 的场景： 如果一个类在本质上无法与 Node.js API 分离（每个方法都需要 fs、net 等），且无法以类框架的形式存在于工作流沙盒中，则应将其完全保留在步骤函数中，并在边界之间传递普通数据对象。

验证 serde 合规性

使用以下工具验证类是否设置正确：

• workflow transform <file> --check-serde -- 显示文件的 SWC 转换输出，并检查 serde 类是否合规（工作流包中不再包含 Node.js 导入）。
• workflow validate -- 扫描所有工作流文件并报告 serde 合规性问题。使用 --json 获取机器可读的输出。
• SWC Playground -- workbench/swc-playground 中的 Web 操场在检测到 serde 模式时会显示 Serde Analysis 面板。
• 构建时警告 -- 构建器会在工作流包中的 serde 类仍有 Node.js 内置模块导入时自动发出警告。

流式传输

使用 getWritable() 从工作流流式传输数据。getWritable() 可在工作流和步骤上下文中调用，但不能直接在工作流函数中与流交互（调用 getWriter()、write()、close()）。必须将流传递给步骤函数以执行实际 I/O，也可以由步骤自行调用 getWritable()。

在工作流中获取流，并将其传递给步骤：

import { getWritable } from "workflow";

export async function myWorkflow() {
  "use workflow";
  const writable = getWritable();
  await writeData(writable, "hello world");
}

async function writeData(writable: WritableStream, chunk: string) {
  "use step";
  const writer = writable.getWriter();
  try {
    await writer.write(chunk);
  } finally {
    writer.releaseLock();
  }
}

直接在步骤中调用 getWritable()（无需传递流）：

import { getWritable } from "workflow";

async function streamData(chunk: string) {
  "use step";
  const writer = getWritable().getWriter();
  try {
    await writer.write(chunk);
  } finally {
    writer.releaseLock();
  }
}

命名空间流

使用 getWritable({ namespace: 'name' }) 为不同类型的数据创建多个独立流。这适用于分离日志与主要输出、不同日志级别、Agent 输出、指标或任何不同的数据通道。长时间运行的工作流可从命名空间流中受益，因为你可以只重放重要事件（例如最终结果），同时将详细日志保存在单独的流中。

示例：分离日志级别和 Agent 输出：

import { getWritable } from "workflow";

type LogEntry = { level: "debug" | "info" | "warn" | "error"; message: string; timestamp: number };
type AgentOutput = { type: "thought" | "action" | "result"; content: string };

async function logDebug(message: string) {
  "use step";
  const writer = getWritable<LogEntry>({ namespace: "logs:debug" }).getWriter();
  try {
    await writer.write({ level: "debug", message, timestamp: Date.now() });
  } finally {
    writer.releaseLock();
  }
}

async function logInfo(message: string) {
  "use step";
  const writer = getWritable<LogEntry>({ namespace: "logs:info" }).getWriter();
  try {
    await writer.write({ level: "info", message, timestamp: Date.now() });
  } finally {
    writer.releaseLock();
  }
}

async function emitAgentThought(thought: string) {
  "use step";
  const writer = getWritable<AgentOutput>({ namespace: "agent:thoughts" }).getWriter();
  try {
    await writer.write({ type: "thought", content: thought });
  } finally {
    writer.releaseLock();
  }
}

async function emitAgentResult(result: string) {
  "use step";
  // Important results go to the default stream for replay
  const writer = getWritable<AgentOutput>().getWriter();
  try {
    await writer.write({ type: "result", content: result });
  } finally {
    writer.releaseLock();
  }
}

export async function agentWorkflow(task: string) {
  "use workflow";
  
  await logInfo(`Starting task: ${task}`);
  await logDebug("Initializing agent context");
  await emitAgentThought("Analyzing the task requirements...");
  
  // ... agent processing ...
  
  await emitAgentResult("Task completed successfully");
  await logInfo("Workflow finished");
}

使用命名空间流：

import { start, getRun } from "workflow/api";
import { agentWorkflow } from "./workflows/agent";

export async function POST(request: Request) {
  const run = await start(agentWorkflow, ["process data"]);

  // Access specific streams by namespace
  const results = run.getReadable({ namespace: undefined }); // Default stream (important results)
  const infoLogs = run.getReadable({ namespace: "logs:info" });
  const debugLogs = run.getReadable({ namespace: "logs:debug" });
  const thoughts = run.getReadable({ namespace: "agent:thoughts" });

  // Return only important results for most clients
  return new Response(results, { headers: { "Content-Type": "application/json" } });
}

// Resume from a specific point (useful for long sessions)
export async function GET(request: Request) {
  const { searchParams } = new URL(request.url);
  const runId = searchParams.get("runId")!;
  const startIndex = parseInt(searchParams.get("startIndex") || "0", 10);
  
  const run = getRun(runId);
  // Resume only the important stream, skip verbose debug logs
  const stream = run.getReadable({ startIndex });
  
  return new Response(stream);
}

对于长时间运行的会话（50 分钟以上），命名空间流有助于提升重放性能。将详细/调试输出放在单独的命名空间中，这样你就可以只重放重要事件。

调试

# Check workflow endpoints are reachable
npx workflow health
npx workflow health --port 3001  # Non-default port

# Visual dashboard for runs
npx workflow web
npx workflow web <run_id>

# CLI inspection (use --json for machine-readable output, --help for full usage)
npx workflow inspect runs
npx workflow inspect run <run_id>

# For Vercel-deployed projects, specify backend and project
npx workflow inspect runs --backend vercel --project <project-name> --team <team-slug>
npx workflow inspect run <run_id> --backend vercel --project <project-name> --team <team-slug>

# Open Vercel dashboard in browser for a specific run
npx workflow inspect run <run_id> --web
npx workflow web <run_id> --backend vercel --project <project-name> --team <team-slug>

# Cancel a running workflow
npx workflow cancel <run_id>
npx workflow cancel <run_id> --backend vercel --project <project-name> --team <team-slug>
# --env defaults to "production"; use --env preview for preview deployments

深度链接到某次运行（分享 URL，无需浏览器）

使用 --url 打印仪表板深度链接并退出。不会打开浏览器，也 不会启动本地服务。当你需要向用户提供可点击的链接（PR 评论、Slack 消息、调试摘要），而不是打开 UI 时，这是合适的工具。（--web 会打开仪表板；--url 只会打印链接。）

# Vercel run: prints the Vercel dashboard URL for the run
npx workflow inspect run <run_id> --backend vercel --project <project> --team <team> --url
npx workflow web <run_id> --backend vercel --project <project> --team <team> --env preview --url

# Local run: prints the local web UI deep link
npx workflow inspect run <run_id> --url

# Machine-readable: --url --json prints { "url": "..." } to stdout
npx workflow inspect run <run_id> --backend vercel --url --json

生成的 URL 格式：

• Vercel： https://vercel.com/<team-slug>/<project-slug>/workflows/runs/<run_id>?environment=<production|preview> （--env 用于选择环境；默认为 production。解析团队 slug 需要先通过 vercel login 登录，并关联项目。）
• 本地： http://localhost:<port>?resource=run&id=<run_id>（端口默认为 3456；运行 npx workflow web 服务期间，此链接有效）。

stdout 仅包含 URL（或 JSON 对象）。其他所有输出都会发送到 stderr，因此你可以直接捕获它，例如：URL=$(npx workflow web <run_id> --backend vercel --url)。

调试提示：

• 对任何命令使用 --json（-j）以获取机器可读的输出
• 使用 --web 在浏览器中打开 Vercel Observability 仪表板，或使用 --url 打印深度链接
• 对任何命令使用 --help 查看完整用法详情
• 只导入实际使用的工作流 API。未使用的导入可能导致 500 错误。

测试工作流

Workflow SDK 提供了 Vitest 插件，可在进程内测试工作流，无需运行服务。

单元测试步骤： 步骤是函数；没有编译器时，"use step" 不执行任何操作。直接测试这些函数：

import { describe, it, expect } from "vitest";
import { createUser } from "./user-signup";

describe("createUser step", () => {
  it("should create a user", async () => {
    const user = await createUser("[email protected]");
    expect(user.email).toBe("[email protected]");
  });
});

集成测试： 对使用 sleep()、钩子、Webhook 或重试的工作流，使用 @workflow/vitest。将它与 workflow 一起安装，并确保两者的主版本号相同：npm i -D @workflow/vitest。如果插件的 @workflow/core 主版本号与应用不一致，它会使运行失败。

// vitest.integration.config.ts
import { defineConfig } from "vitest/config";
import { workflow } from "@workflow/vitest";

export default defineConfig({
  plugins: [workflow()],
  test: {
    include: ["**/*.integration.test.ts"],
    testTimeout: 60_000,
  },
});

// approval.integration.test.ts
import { describe, it, expect } from "vitest";
import { start, getRun, resumeHook } from "workflow/api";
import { waitForHook, waitForSleep } from "@workflow/vitest";
import { approvalWorkflow } from "./approval";

describe("approvalWorkflow", () => {
  it("should publish when approved", async () => {
    const run = await start(approvalWorkflow, ["doc-123"]);

    // Wait for the hook, then resume it
    await waitForHook(run, { token: "approval:doc-123" });
    await resumeHook("approval:doc-123", { approved: true, reviewer: "alice" });

    // Wait for sleep, then wake it up
    const sleepId = await waitForSleep(run);
    await getRun(run.runId).wakeUp({ correlationIds: [sleepId] });

    const result = await run.returnValue;
    expect(result).toEqual({ status: "published", reviewer: "alice" });
  });
});

测试 Webhook： 使用带有 Request 对象的 resumeWebhook()。无需 HTTP 服务：

import { start, resumeWebhook } from "workflow/api";
import { waitForHook } from "@workflow/vitest";

const run = await start(ingestWorkflow, ["ep-1"]);
const hook = await waitForHook(run);  // Discovers the random webhook token
await resumeWebhook(hook.token, new Request("https://example.com/webhook", {
  method: "POST",
  body: JSON.stringify({ event: "order.created" }),
}));

主要 API：

• start()：触发工作流
• run.returnValue：等待工作流完成
• waitForHook(run, { token? }) / waitForSleep(run)：等待工作流到达暂停点
• resumeHook(token, data) / resumeWebhook(token, request)：恢复已暂停的工作流
• getRun(runId).wakeUp({ correlationIds })：跳过 sleep() 调用
• getWorkflowRef(name) / listWorkflowRefs()：当测试无法导入函数时，从测试构建的清单中查找工作流（切勿手动编写 workflow//... ID）

最佳实践：

• 将单元测试（不使用插件）和集成测试（使用 workflow() 插件）放在不同的配置中
• 安装 @workflow/vitest 时，确保其主版本号与 workflow 相同，并同时升级两者
• 使用基于测试数据的确定性钩子令牌，以便更轻松地恢复执行
• 设置较宽裕的 testTimeout 值，因为工作流的运行时间可能长于典型的单元测试
• vi.mock() 无法作用于工作流主体（它们在 VM 中运行）；只有当生成的 bundle 通过 Vitest 模块运行器加载时，vi.mock() 才能作用于步骤代码。项目本地模块会打包进步骤 bundle，因此应模拟 npm 叶子依赖、注入依赖，或对步骤进行单元测试

可观测性与 World SDK

使用 await getWorld() 构建可观测性仪表板、管理面板，并检查工作流状态。getWorld() 是异步的，返回 Promise<World>（基于动态导入／环境的设置）。

主要导入：

import { getWorld } from "workflow/runtime";
import { hydrateResourceIO, observabilityRevivers, parseStepName, parseWorkflowName } from "workflow/observability";

主要文档（搜索 node_modules/workflow/docs/ 获取完整详情）：

• api-reference/workflow-runtime/world/storage.mdx：事件、运行、步骤和钩子（事件是事实来源；其他内容是物化视图）
• api-reference/workflow-observability/：水合与名称解析

World SDK 方法签名

⚠️ 分页参数嵌套在 { pagination: { cursor } } 中，而不是直接使用 { cursor }。

const world = await getWorld();

// Runs
const { data, cursor } = await world.runs.list({ pagination: { cursor }, resolveData: 'all' | 'none' });
const run = await world.runs.get(runId, { resolveData: 'all' | 'none' });
// Cancel via event creation (no cancel() method on runs)
await world.events.create(runId, { eventType: 'run_cancelled' });

// Steps: runId is top-level, NOT inside pagination
const { data, cursor } = await world.steps.list({ runId, pagination: { cursor }, resolveData: 'all' | 'none' });
const step = await world.steps.get(runId, stepId, { resolveData: 'all' | 'none' });

// Events
const { data, cursor } = await world.events.list({ runId, pagination: { cursor } });
await world.events.create(runId, { eventType: 'run_cancelled' });

// Hooks
const hook = await world.hooks.get(hookId);
const hook = await world.hooks.getByToken(token);

// Streams (methods on world.streams)
await world.streams.write(runId, name, chunk);
await world.streams.writeMulti?.(runId, name, chunks);
const readable = await world.streams.get(runId, name, startIndex);
await world.streams.close(runId, name);
const streamNames = await world.streams.list(runId);
const chunks = await world.streams.getChunks(runId, name, { limit, cursor });
const info = await world.streams.getInfo(runId, name);

// Queue (methods live directly on world as internal SDK infrastructure)
await world.queue(queueName, payload, opts);
const deploymentId = await world.getDeploymentId();

resolveData 参数

控制是否在响应中包含输入／输出数据。可接受 'all'（默认值）或 'none'。

重要： 即使设置为 'all'，数据仍会经过 devalue 序列化。必须调用 hydrateResourceIO() 才能获取可用的 JS 值。

• 对于状态轮询、进度仪表板和运行列表，使用 'none'
• 需要检查实际步骤 I/O 数据时，使用 'all'（或省略此参数），然后务必进行水合

// Lightweight status check with no I/O loaded
const run = await world.runs.get(runId, { resolveData: 'none' });
console.log(run.status); // 'running' | 'completed' | 'failed' | 'cancelled'

// Full inspection: resolveData includes data, hydrateResourceIO deserializes it
const step = await world.steps.get(runId, stepId); // defaults to 'all'
const hydrated = hydrateResourceIO(step, observabilityRevivers);

常见错误：设置 resolveData: 'all' 后检查 step.input !== undefined，并认为数据已可直接使用。数据确实存在，但仍处于序列化状态，因此务必先进行水合。

数据水合（devalue 格式）

步骤 I/O 使用 devalue 序列化，并带有 4 字节格式前缀（devl）。未经水合时，input／output 是类似 Uint8Array、带有数字键的对象： {"0":100,"1":101,"2":118,"3":108,...} 中包含的值未经水合后无法使用。

使用 I/O 数据前务必先进行水合：

import { hydrateResourceIO, observabilityRevivers } from "workflow/observability";

const { data: steps } = await world.steps.list({ runId, resolveData: 'all' });
const hydrated = steps.map(s => hydrateResourceIO(s, observabilityRevivers));
// hydrated[0].input → [123, 2] (actual function arguments)
// hydrated[0].output → 125 (actual return value)

hydrateResourceIO 同时适用于 Step 和 WorkflowRun 对象。对于加密工作流，请使用 getEncryptionKeyForRun() + hydrateResourceIOWithKey()。

名称解析

parseWorkflowName()、parseStepName() 和 parseClassName() 返回 { shortName: string, moduleSpecifier: string } | null。务必使用可选链：

const parsed = parseWorkflowName("workflow//./src/workflows/order//processOrder");
// parsed?.shortName → "processOrder"
// parsed?.moduleSpecifier → "./src/workflows/order"
// ⚠️ Returns null if format doesn't match

事件类型

事件是仅追加的数据事实来源。Run/Step/Hook 是物化视图。

类别	类型
Run	run_created、run_started、run_completed、run_failed、run_cancelled
Step	step_created、step_started、step_completed、step_failed、step_retrying
Hook	hook_created、hook_received、hook_disposed、hook_conflict
Wait	wait_created、wait_completed

错误处理模式

针对不同故障模式的三种错误策略：

错误类型	适用场景	行为
FatalError	永久性故障（输入无效、身份验证被拒绝）	立即终止工作流，不重试
RetryableError	暂时性故障（速率限制、超时）	重试，可选择设置 retryAfter 延迟
Promise.allSettled	关键性各不相同的并行步骤	即使部分步骤失败，也会继续执行

import { FatalError, RetryableError } from "workflow";

// Permanent failure, so the workflow terminates
throw new FatalError("Invalid input: missing required field");

// Transient failure, so it will retry
throw new RetryableError("API rate limited", { retryAfter: "5m" });

// Mixed criticality parallel execution
const results = await Promise.allSettled([
  criticalStep(data),    // Must succeed
  optionalStep(data),    // OK to fail
  enrichmentStep(data),  // OK to fail
]);
const [critical, optional, enrichment] = results;
if (critical.status === "rejected") throw new FatalError(critical.reason);