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/ - ハイドレーションと名前の解析ユーティリティ（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/ - コアランタイム（基礎、仕組み）
• @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" から import { 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 エージェント

AI SDK の WorkflowAgent を使用すると、Workflow 5 で永続性のあるエージェントを構築できます。これは、非推奨となった @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 はモデル呼び出しの回数を制限します。デフォルトでは、モデルがツールの呼び出しを停止すると実行を終了します
• 複数ターンの場合は、後続の agent.stream() 呼び出しに result.messages と新しいユーザーメッセージを渡します

詳しくは、インストール済みの AI SDK パッケージまたは https://ai-sdk.dev/v7/docs/agents/workflow-agent の WorkflowAgent ドキュメントをご確認ください。

ワークフローと子ワークフローの開始

API ルートからワークフローを起動するには start() を使用します。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 のプロパティアクセスやメソッド呼び出しは、それぞれがステップになります。

ランのサイズと同時実行数：分割するタイミングを把握する

上限が設定されている対象は 3 つあります。覚えている数字を確定値として扱わないでください。現在の値は Workflow run limits に公開されており、引用する場合はこの情報だけを参照してください。

ランあたりのイベント数。 ランのイベントログには上限があり、上限を超えるとランは MAX_EVENTS_EXCEEDED で失敗します。イベントはステップとは異なります。最初の試行で成功するステップでは 3 つ（step_created、step_started、step_completed）が記録され、再試行ではさらに 1 つまたは 2 つ記録されます。フック、スリープ、Webhook もそれぞれイベントを記録します。上限に達するかなり前に、子ワークフローへ分割してください。料金ページでは数千件を超えるあたりから分割することを推奨しています。ランが失敗するよりずっと前に、リプレイが遅くなるためです。

ランあたりのステップ数。 イベント数とは別に上限があるため、幅の狭い長い逐次チェーンにも上限があります。チェーンが 5 桁の数値に達する場合は、複数の項目を 1 つのステップにまとめてください。

同時実行数。 幅広いファンアウトは拒否されるのではなく、スロットリングされます。ランごとに 1 秒あたりのイベント作成数が制限されるため、数千件を対象としたフラットな Promise.all では、バックオフに多くの時間が費やされます。代わりにバッチ処理やバンドル処理を行い、リストをチャンクに分けるか、1 ステップで複数の項目を処理して、同時に実行する単位の数を減らしてください。項目ごとに 1 つの子ランを生成しても、ファンアウトの幅が狭まるわけではありません。各子ランのログを制限し、失敗を分離できるため、その目的には有効ですが、チャンク化の代わりにはなりません。

これらの上限を自分で引き上げることはできません。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 は両方を制限します。1 つのステップで BATCH 件を処理するため、BATCH が大きくなるほどイベント数が減るのは、この 3 つのうちこれだけです。

フック：外部イベントによる一時停止と再開

フックを使うと、ワークフローで外部データを待機できます。ワークフロー内で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;
}

複数のイベント（反復可能なフック）

フックは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を追加する必要があります。

パターン： 計算プロパティ構文を使い、クラス本体に2つの静的メソッドを追加します。

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. 計算プロパティ構文（static [WORKFLOW_SERIALIZE](...)）を使った静的メソッドとして、serde メソッドをクラス本体の内側に定義してください。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 を必要とする場合、正しい解決方法は/* @vite-ignore */ import(...)ではなく"use step"です。

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 のウェブプレイグラウンドでは、serde パターンが検出されると Serde Analysis パネルが表示されます。
• ビルド時の警告 -- ビルダーは、ワークフローバンドルに 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' }) を使って、データの種類ごとに複数の独立したストリームを作成します。ログと主要な出力、異なるログレベル、エージェントの出力、メトリクス、その他の異なるデータチャネルを分ける際に便利です。長時間実行するワークフローでは、名前空間付きストリームを使うと、詳細なログを別のストリームに保持しながら、重要なイベント（最終結果など）のみを再生できます。

例：ログレベルとエージェント出力の分離：

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 を使うと、ダッシュボードへのディープリンクを出力して終了します。ブラウザーは開かず、ローカルサーバーも起動しません。UI を開くのではなく、ユーザーにクリック可能なリンクを渡したい場合（PR コメント、Slack メッセージ、デバッグ概要など）に適した方法です。（--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 です。チームのスラッグを解決するには、プロジェクトをリンクした状態で 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）を使います
• Vercel Observability ダッシュボードをブラウザーで開くには --web を、ディープリンクを出力するには --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 と一緒にインストールし、2つのメジャーバージョンを揃えてください：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 内で実行されます）には届きません。また、生成されたバンドルが Vitest のモジュールランナーを通じて読み込まれる場合に限り、ステップコードに届きます。プロジェクト内のモジュールはステップバンドルにまとめられるため、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/ を grep してください）：

• 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 でシリアライズされたままです。使用可能な JS 値を取得するには、必ず hydrateResourceIO() を呼び出してください。

• ステータスのポーリング、進行状況ダッシュボード、実行一覧には '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 は、4 バイトの形式プレフィックス（devl）とともに devalue でシリアライズされます。ハイドレーションしない場合、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

エラー処理のパターン

さまざまな障害モードに対応する3つのエラー戦略です。

エラーの種類	使用する場面	動作
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);