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-api/world/ - World SDK (runs.mdx, steps.mdx, hooks.mdx, events.mdx, streams.mdx, queue.mdx, observability.mdx)
• ai/ - AI SDK 統合ドキュメント
• errors/ - エラーコードドキュメント

関連パッケージにもバンドルドキュメントが含まれています:

• @workflow/ai: node_modules/@workflow/ai/docs/ - DurableAgent と AI 統合
• @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 - async 関数を耐久性ものにする
"use step";      // First line - 関数をキャッシュ可能で再試行可能なユニットにする

必須のインポート:

// Workflow プリミティブ
import { sleep, fetch, createHook, createWebhook, getWritable } from "workflow";
import { FatalError, RetryableError } from "workflow";
import { getWorkflowMetadata, getStepMetadata } from "workflow";

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

// 可観測性とデータハイドレーション
import { hydrateResourceIO, observabilityRevivers, parseStepName, parseWorkflowName } from "workflow/observability";

// フレームワーク統合
import { withWorkflow } from "workflow/next";
import { workflow } from "workflow/vite";
import { workflow } from "workflow/astro";
// または modules: ["workflow/nitro"] を使用 (Nitro/Nuxt 用)

// AI エージェント
import { DurableAgent } from "@workflow/ai/agent";

サンドボックスエラーを避けるためにステップ関数を優先する

"use workflow" 関数はサンドボックス VM で実行されます。"use step" 関数は Node.js へのフルアクセス を持ちます。ロジックをステップに入れ、ワークフロー関数は純粋にオーケストレーション用途で使用してください。

// ステップは Node.js と npm へのフルアクセスを持つ
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 はステップで回避策なしで動作
  return await generateText({
    model: openai("gpt-4"),
    prompt: `Process: ${JSON.stringify(data)}`,
  });
}

// ワークフローがステップをオーケストレート - サンドボックス問題なし
export async function dataProcessingWorkflow(userId: string) {
  "use workflow";
  const data = await fetchUserData(userId);
  const processed = await processWithAI(data);
  return { success: true, processed };
}

メリット: ステップは自動再試行、結果はリプレイ用に永続化、サンドボックス制限なし。

ワークフロー サンドボックスの制限

ロジックがワークフロー関数で直接必要な場合 (ステップではなく)、これらの制限が適用されます:

制限	解決策
fetch() なし	import { fetch } from "workflow" してから 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;  // AI SDK と HTTP ライブラリに必須
  // これで generateText() その他のライブラリが動作
}

注: @workflow/ai の DurableAgent は fetch 割り当てを自動的に処理します。

DurableAgent — ワークフロー内の AI エージェント

DurableAgent を使用して、状態を保持し中断から復旧する AI エージェントを構築します。ワークフロー サンドボックスを自動的に処理します (手動の globalThis.fetch 割り当てが不要)。

import { DurableAgent } from "@workflow/ai/agent";
import { getWritable } from "workflow";
import { z } from "zod";
import type { UIMessageChunk } from "ai";

async function lookupData({ query }: { query: string }) {
  "use step";
  // ステップ関数は Node.js へのフルアクセスを持つ
  return `Results for "${query}"`;
}

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

  const agent = new DurableAgent({
    model: "anthropic/claude-sonnet-4-5",
    system: "You are a helpful assistant.",
    tools: {
      lookupData: {
        description: "Search for information",
        inputSchema: z.object({ query: z.string() }),
        execute: lookupData,
      },
    },
  });

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

  return result.messages;
}

重要なポイント:

• getWritable<UIMessageChunk>() はワークフロー実行のデフォルトストリームに出力をストリーム配信
• Node.js/npm アクセスが必要なツール execute 関数は "use step" を使用すべき
• ワークフロー プリミティブ (sleep()、createHook()) を使用するツール execute 関数は "use step" を使用 してはいけません — ワークフローレベルで実行
• maxSteps は LLM 呼び出し数を制限 (デフォルトは無制限)
• マルチターン: 後続の agent.stream() 呼び出しに result.messages と新しいユーザーメッセージを渡す

DurableAgent の詳細は node_modules/@workflow/ai/docs/ の AI ドキュメントを確認してください。

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

API ルートからワークフローを起動するには start() を使用します。start() はワークフロー コンテキスト内で直接呼び出せません — ステップ関数でラップしてください。

import { start } from "workflow/api";

// API ルートから — 直接動作
export async function POST() {
  const run = await start(myWorkflow, [arg1, arg2]);
  return Response.json({ runId: run.runId });
}

// 引数なしワークフロー
const run = await start(noArgWorkflow);

ワークフロー内から子ワークフローを開始 — ステップを使用する必要があります:

import { start } from "workflow/api";

// start() をステップ関数でラップ
async function triggerChild(data: string) {
  "use step";
  const run = await start(childWorkflow, [data]);
  return run.runId;
}

export async function parentWorkflow() {
  "use workflow";
  const childRunId = await triggerChild("some data");  // ステップ経由でファイア・アンド・フォーゲット
  await sleep("1h");
}

start() は即座に返ります — ワークフロー完了を待ちません。run.returnValue を使用して完了を待ちます。

フック — 外部イベントで一時停止と再開

フックでワークフローが外部データを待機できます。ワークフロー内で 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",  // 外部システム用の確定的トークン
  });

  const result = await hook;  // ワークフローここで一時停止
  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 >= 400 && res.status < 500) {
  throw new FatalError(`Client error: ${res.status}`);
}
if (res.status === 429) {
  throw new RetryableError("Rate limited", { retryAfter: "5m" });
}

シリアライゼーション

ワークフローとステップに渡される/戻されるすべてのデータはシリアライズ可能である必要があります。

サポートされた組み込み型: string, number, boolean, null, undefined, bigint, プレーンオブジェクト, 配列, Date, RegExp, URL, URLSearchParams, Map, Set, Headers, ArrayBuffer, 型付き配列, Request, Response, ReadableStream, WritableStream。

サポートされていない: 関数, シンボル, 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;
  }

  // シリアライズ: プレーンデータを返す (devalue 互換型のみである必要があります)
  static [WORKFLOW_SERIALIZE](instance: Point) {
    return { x: instance.x, y: instance.y };
  }

  // デシリアライズ: プレーンデータから再構築
  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. "use step" を Node.js 依存インスタンスメソッドに追加します。 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' }) を使用して複数の独立したストリームを作成し、異なるデータ型を処理します。これはプライマリ出力からログを分離、異なるログレベル, エージェント出力, メトリクス, またはその他の個別データチャネルに有用です。長時間実行ワークフローは重要なイベント (例: 最終結果) のみをリプレイし、詳細ログを別ストリームに保持できるため、名前空間付きストリームの恩恵を受けます。

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

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";
  // 重要な結果はデフォルトストリームに、簡単なリプレイ用
  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...");
  
  // ... エージェント処理 ...
  
  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"]);

  // 名前空間別に特定ストリームアクセス
  const results = run.getReadable({ namespace: undefined }); // デフォルトストリーム (重要な結果)
  const infoLogs = run.getReadable({ namespace: "logs:info" });
  const debugLogs = run.getReadable({ namespace: "logs:debug" });
  const thoughts = run.getReadable({ namespace: "agent:thoughts" });

  // ほとんどのクライアントに重要な結果のみ返す
  return new Response(results, { headers: { "Content-Type": "application/json" } });
}

// 特定のポイントから再開 (長時間セッション用途に有用)
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);
  // 重要なストリームのみリプレイ, 詳細デバッグログをスキップ
  const stream = run.getReadable({ startIndex });
  
  return new Response(stream);
}

**プロ チップ: 非常に長時間実行セッション (50 分以上)、名前空間付きストリームはリプレイパフォーマンス管理に役立ちます。詳細/デバッグ出力を別の名前空間に入れ、重要なイベントのみをすばやくリプレイできます。

デバッグ

# ワークフロー エンドポイントが到達可能かチェック
npx workflow health
npx workflow health --port 3001  # デフォルト以外のポート

# 実行の可視的ダッシュボード
npx workflow web
npx workflow web <run_id>

# CLI 検査 (機械可読出力用に --json、完全使用法用に --help)
npx workflow inspect runs
npx workflow inspect run <run_id>

# Vercel デプロイプロジェクト用、バックエンドとプロジェクト指定
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>

# ブラウザで特定実行用 Vercel ダッシュボードを開く
npx workflow inspect run <run_id> --web
npx workflow web <run_id> --backend vercel --project <project-name> --team <team-slug>

# 実行中ワークフロー キャンセル
npx workflow cancel <run_id>
npx workflow cancel <run_id> --backend vercel --project <project-name> --team <team-slug>
# --env のデフォルト "production"; プレビューデプロイ用に --env preview

デバッグのコツ:

• コマンドで機械可読出力用に --json (-j) を使用
• ブラウザで Vercel 可観測性ダッシュボードを開くに --web を使用
• 完全使用法詳細用にいかなるコマンドでも --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("test@example.com");
    expect(user.email).toBe("test@example.com");
  });
});

統合テスト: sleep()、フック、ウェブフック、再試行を使用するワークフロー用に @workflow/vitest を使用:

// 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"]);

    // フック待機, その後再開
    await waitForHook(run, { token: "approval:doc-123" });
    await resumeHook("approval:doc-123", { approved: true, reviewer: "alice" });

    // sleep 待機, その後起動
    const sleepId = await waitForSleep(run);
    await getRun(run.runId).wakeUp({ correlationIds: [sleepId] });

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

ウェブフック テスト: resumeWebhook() を Request オブジェクト用に使用 — HTTP サーバー不要:

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

const run = await start(ingestWorkflow, ["ep-1"]);
const hook = await waitForHook(run);  // ランダムウェブフック トークン検出
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() 呼び出しをスキップ

ベストプラクティス:

• ユニットテスト (プラグインなし) と統合テスト (workflow() プラグイン) を別々の設定に保持
• テストデータに基づく確定的フック トークンを使用して再開しやすく
• 寛容な testTimeout を設定 — ワークフロー は標準的なユニットテストより長く実行される可能性
• vi.mock() は統合テストで 動作しません — ステップ依存関係は esbuild でバンドル

可観測性と World SDK

await getWorld() を使用して可観測性ダッシュボード, 管理パネル, ワークフロー状態検査を構築。getWorld() は非同期で Promise<World> を返します (動的インポート / env ベース設定)。

キー インポート:

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

キー ドキュメント (node_modules/workflow/docs/ を grep して完全詳細):

• api-reference/workflow-api/world/storage.mdx — events, runs, steps, hooks (events がトゥルースソース; 他は具現化ビュー)
• api-reference/workflow-api/world/observability.mdx — ハイドレーション, 解析, 暗号化

World SDK メソッドシグネチャ

⚠️ ページネーションはネストされています: { pagination: { cursor } } — NOT { cursor } 直接。

const world = await getWorld();

// 実行
const { data, cursor } = await world.runs.list({ pagination: { cursor }, resolveData: 'all' | 'none' });
const run = await world.runs.get(runId, { resolveData: 'all' | 'none' });
// キャンセル はイベント作成経由 (実行上に cancel() メソッドなし)
await world.events.create(runId, { eventType: 'run_cancelled' });

// ステップ — runId はトップレベル, ページネーション内 NOT
const { data, cursor } = await world.steps.list({ runId, pagination: { cursor }, resolveData: 'all' | 'none' });
const step = await world.steps.get(runId, stepId, { resolveData: 'all' | 'none' });

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

// フック
const hook = await world.hooks.get(hookId);
const hook = await world.hooks.getByToken(token);

// ストリーム (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);

// キュー (メソッドは world 上に直接存在 — 内部 SDK インフラ)
await world.queue(queueName, payload, opts);
const deploymentId = await world.getDeploymentId();

resolveData パラメータ

入力/出力データが応答に 含まれる かどうかを制御。'all' (デフォルト) または 'none' を受け入れます。

重要: 'all' でも、データは devalue シリアライズされています。使用可能な JS 値を取得するには hydrateResourceIO() を呼び出す 必須 です。

• 'none' を使用 ステータスポーリング, 進捗ダッシュボード, 実行リスト用
• 'all' を使用 (または省略) 実際のステップ I/O データを検査する必要時 — その後 常にハイドレーション

// 軽量ステータス チェック — I/O ロードなし
const run = await world.runs.get(runId, { resolveData: 'none' });
console.log(run.status); // 'running' | 'completed' | 'failed' | 'cancelled'

// 完全な検査 — resolveData はデータ含む, hydrateResourceIO これをデシリアライズ
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] (実際の関数引数)
// hydrated[0].output → 125 (実際の戻り値)

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"
// ⚠️ フォーマット一致しない場合 null を返す

イベント型

イベントは追記専用トゥルースソース。実行/ステップ/フックは具現化ビュー。

カテゴリ	型
実行	run_created, run_started, run_completed, run_failed, run_cancelled
ステップ	step_created, step_started, step_completed, step_failed, step_retrying
フック	hook_created, hook_received, hook_disposed, hook_conflict
待機	wait_created, wait_completed

エラーハンドリングパターン

異なる失敗モード用の 3 つのエラー戦略:

エラー型	使用時	動作
FatalError	永続的失敗 (不正な入力, 認証拒否)	ワークフロー即座終了, 再試行なし
RetryableError	一時的失敗 (レート制限, タイムアウト)	オプション retryAfter 遅延で再試行
Promise.allSettled	混合重要度の並列ステップ	一部ステップ失敗でも続行

import { FatalError, RetryableError } from "workflow";

// 永続的失敗 — ワークフロー終了
throw new FatalError("Invalid input: missing required field");

// 一時的失敗 — 再試行
throw new RetryableError("API rate limited", { retryAfter: "5m" });

// 混合重要度並列実行
const results = await Promise.allSettled([
  criticalStep(data),    // 成功必須
  optionalStep(data),    // 失敗 OK
  enrichmentStep(data),  // 失敗 OK
]);
const [critical, optional, enrichment] = results;
if (critical.status === "rejected") throw new FatalError(critical.reason);