◈ Agents & Manager/2026-03-30上級

Antigravity × Durable Execution: 障害に強い長時間AIタスクの設計パターン

Durable Execution パターンで Antigravity のエージェントを長時間タスクに投入する実装ガイド。チェックポイント・べき等性・自動リトライに加え、CAS 付きチェックポイントストアの実装、障害注入テスト、本番で踏みやすい 7 つの落とし穴まで具体的にまとめます。

antigravity⁴⁴⁹ durable-execution² ai-agent¹⁷ typescript²⁷ fault-tolerance workflow³⁹

✦ プレミアム記事

ある朝、AdMob の前日収益を集計する 1 時間半のパイプラインがあと 10 分というところで落ちていて、最初からやり直すのを諦めて出社の電車の中で同じ処理を再度キックした、という経験があります。個人開発で壁紙アプリや癒し系アプリを公開し続けていて、その収益を AdMob と App Store / Google Play のデータで毎日突合させているのですが、Durable Execution を導入する前は、こうしたパイプラインの「あと一歩で落ちた」状況がとにかく痛かった。AdMob API のレートリミット、Cloudflare Workers のタイムアウト、Supabase 側の一時切断、どれもこちらの設計次第で「ゼロからやり直し」になっていました。

「落ちても最初からやり直さない」という当たり前を実装で担保するのが Durable Execution です。個人開発で Antigravity のエージェントに長時間タスクを任せるようになってから、このパターンなしでは夜間ジョブを回せなくなりました。

このページでは、最小実装から始めて、本番で必要になるチェックポイントストア（同時実行の検知つき）、再開の正しさを壊して確かめるテスト、そして公式ドキュメントが触れない 7 つの落とし穴までを、実際に動かしているコードで追っていきます。

Durable Execution の 3 つの基本原則

このパターンを正しく理解するために、3 つの基本原則を押さえておきましょう。Antigravity でのエージェント設計にも、そのまま転用できる骨格です。

チェックポイントによる状態保存

ワークフローの各ステップが完了するたびに、その結果を永続ストレージ（データベースやキュー、Cloudflare KV など）に保存します。障害が発生した場合、最後に保存されたチェックポイントから処理を再開できます。私が運用している AdMob 集計パイプラインでは、再実行時間が 47 分 → 4 分（約 11 倍の短縮）にまで縮みました。

べき等性（Idempotency）の確保

同じ処理が複数回実行されても、結果が変わらないことを保証する設計です。リトライが発生したときに、二重にデータが書き込まれたり、メールが複数回送信されたりすることを防ぎます。べき等性の欠落は、特に決済・通知系の処理で取り返しのつかない事故に直結します。

自動リトライとバックオフ

一時的な障害（ネットワークタイムアウト、レート制限など）に対して、指数バックオフ付きで自動的にリトライします。永続的な障害と一時的な障害を区別し、適切に対処する点が肝心です。AdMob API の場合、UNAVAILABLE と RESOURCE_EXHAUSTED は再試行価値あり、PERMISSION_DENIED は即時打ち切りといった具体的な分岐を入れています。

Antigravity で Durable Execution を実装する最小コード

Antigravity のエージェント機能を使って、実際に Durable Execution パターンを実装してみましょう。ここでは TypeScript で、外部 API からデータを取得し、加工して保存する長時間ワークフローを構築します。

// durable-workflow.ts
// Durable Execution パターンの基本実装
 
interface WorkflowState {
  currentStep: string;
  completedSteps: string[];
  data: Record<string, unknown>;
  retryCount: number;
  lastCheckpoint: string;
}
 
class DurableWorkflow {
  private state: WorkflowState;
  private storePath: string;
 
  constructor(workflowId: string) {
    this.storePath = `./workflow-state/${workflowId}.json`;
    this.state = this.loadState();
  }
 
  // チェックポイントから状態を復元
  private loadState(): WorkflowState {
    try {
      const fs = require("fs");
      if (fs.existsSync(this.storePath)) {
        const saved = JSON.parse(fs.readFileSync(this.storePath, "utf-8"));
        console.log(`Resuming from checkpoint: ${saved.lastCheckpoint}`);
        return saved;
      }
    } catch (e) {
      console.warn("No checkpoint found, starting fresh");
    }
    return {
      currentStep: "init",
      completedSteps: [],
      data: {},
      retryCount: 0,
      lastCheckpoint: "",
    };
  }
 
  // チェックポイントを保存
  private saveCheckpoint(stepName: string): void {
    const fs = require("fs");
    const path = require("path");
    fs.mkdirSync(path.dirname(this.storePath), { recursive: true });
    this.state.lastCheckpoint = stepName;
    this.state.completedSteps.push(stepName);
    fs.writeFileSync(this.storePath, JSON.stringify(this.state, null, 2));
    console.log(`Checkpoint saved: ${stepName}`);
  }
 
  // べき等なステップ実行
  async executeStep<T>(
    stepName: string,
    fn: () => Promise<T>,
    maxRetries = 3
  ): Promise<T> {
    // 既に完了済みのステップはスキップ
    if (this.state.completedSteps.includes(stepName)) {
      console.log(`Skipping completed step: ${stepName}`);
      return this.state.data[stepName] as T;
    }
 
    this.state.currentStep = stepName;
 
    for (let attempt = 1; attempt <= maxRetries; attempt++) {
      try {
        const result = await fn();
        this.state.data[stepName] = result;
        this.saveCheckpoint(stepName);
        return result;
      } catch (error) {
        const waitMs = Math.min(1000 * Math.pow(2, attempt), 30000);
        console.error(
          `Step "${stepName}" failed (attempt ${attempt}/${maxRetries}):`,
          error
        );
        if (attempt === maxRetries) throw error;
        console.log(`Retrying in ${waitMs}ms...`);
        await new Promise((r) => setTimeout(r, waitMs));
      }
    }
    throw new Error(`Step "${stepName}" exhausted all retries`);
  }
}
 
// 使用例: 大量データの変換パイプライン
async function runDataPipeline() {
  const workflow = new DurableWorkflow("data-pipeline-001");
 
  // Step 1: データ取得（障害時はチェックポイントから再開）
  const rawData = await workflow.executeStep("fetch-data", async () => {
    const res = await fetch("https://api.example.com/large-dataset");
    return res.json();
    // 期待する出力: { records: [...], total: 10000 }
  });
 
  // Step 2: データ変換
  const transformed = await workflow.executeStep("transform", async () => {
    return rawData.records.map((r: any) => ({
      id: r.id,
      normalized: r.value.toLowerCase().trim(),
      processedAt: new Date().toISOString(),
    }));
    // 期待する出力: [{ id: "1", normalized: "...", processedAt: "..." }, ...]
  });
 
  // Step 3: バッチ保存
  const result = await workflow.executeStep("save-batch", async () => {
    const batchSize = 100;
    let saved = 0;
    for (let i = 0; i < transformed.length; i += batchSize) {
      const batch = transformed.slice(i, i + batchSize);
      await saveBatch(batch); // DB書き込み（べき等な upsert）
      saved += batch.length;
    }
    return { totalSaved: saved };
    // 期待する出力: { totalSaved: 10000 }
  });
 
  console.log("Pipeline complete:", result);
}
 
async function saveBatch(batch: any[]) {
  // べき等な upsert 処理（同じIDのレコードは上書き）
  console.log(`Saving batch of ${batch.length} records`);
}
 
runDataPipeline().catch(console.error);

このコードでは、各ステップの完了時にファイルシステムにチェックポイントを保存しています。プロセスが中断されても、再実行時には完了済みのステップを自動的にスキップし、未完了のステップから処理を再開します。ただしサーバーレス環境ではローカル FS が揮発しますし、それ以上に厄介な問題があります。次のセクションでそこを詰めます。

✦

ここまでお読みいただきありがとうございます。

この記事の続きを読む

この先には、実装コードやベンチマーク結果など、実務でお役に立てる内容をご用意しています。このサイトは広告を掲載しておらず、サーバーや開発にかかる費用はメンバーの皆様のご支援で成り立っています。もしお役に立てていましたら、ご支援いただけますと大変ありがたいです。

この記事で得られること

✦毎朝 04:30 に走る AdMob 集計パイプラインで Durable Execution が救った 3 つの障害ログ（再実行時間 47 分 → 4 分）

✦二重起動を検知する Compare-And-Swap 付きチェックポイントストア（Cloudflare KV 実装 / LeaseLostError の扱い）と、再開の正しさを機械的に保証する障害注入テスト

✦公式ドキュメントには書かれていない、本番運用 7 つの落とし穴（チェックポイント肥大 / 二重通知 / レートリミット / 部分復旧 / 時刻ドリフト / Cold Start / 監視盲点）

✦Antigravity に Durable Workflow を生成させる時のプロンプト分割テンプレート（4 工程 / 失敗列挙 / ジョブ単位の独立化）

Stripe による安全な決済 · いつでもキャンセル可能

✦

この記事を購入する

この先の内容をすべてお読みいただけます。一度のご購入で、いつでも何度でもアクセスできます。このサイトは広告を掲載しておらず、皆さまのご支援がサーバー費用などの運営を支えています。

または

メンバーシップなら全記事が読み放題 →

ローカル FS のチェックポイントを、そのまま本番に持ち込むと壊れます

先ほどの最小実装は、チェックポイントをローカルのファイルへ書いています。手元で挙動を確かめる分にはこれで十分です。ただ、同じコードを Cloudflare Workers や AWS Lambda にそのまま載せると、二段構えで壊れます。

ひとつは揮発です。インスタンスは処理が終われば消えますから、次の起動時に読みにいくファイルは存在しません。

もうひとつが厄介でした。同じワークフロー ID を、2 つのインスタンスが同時に掴むケースです。cron の自動リトライと、ダッシュボードから叩いた手動再実行が重なった朝に踏みました。両方が「Step 4 は未完了」と読み、両方が Step 4 を走らせ、Slack に同じ日次レポートが 2 通並びました。落ちてもいないのに壊れる、という一番気味の悪い壊れ方です。

必要なのは 2 つでした。ストレージを差し替えられる形にしておくこと。そして保存のたびに「自分が読んだ時点からバージョンが変わっていないこと」を確かめること。後者はいわゆる Compare-And-Swap で、これがないと上のような同時実行はそもそも検知できません。

// checkpoint-store.ts
// チェックポイントの永続化を差し替え可能にし、同時実行を検知できるようにする
 
export interface CheckpointRecord {
  state: WorkflowState;
  version: number;      // 楽観ロック用。保存のたびに +1 される
  owner: string;        // このランを掴んでいるインスタンス ID
  leaseUntil: number;   // リースの失効時刻（epoch ms）
}
 
export interface CheckpointStore {
  load(workflowId: string): Promise<CheckpointRecord | null>;
  /** expectedVersion と一致したときだけ書き込む。負けたら false を返す */
  compareAndSave(
    workflowId: string,
    next: Pick<CheckpointRecord, "state">,
    expectedVersion: number
  ): Promise<boolean>;
}
 
/**
 * 他インスタンスに追い越されたときに投げます。
 * リトライ対象にせず、その場で静かに降りるための型です。
 */
export class LeaseLostError extends Error {
  constructor(workflowId: string) {
    super(`Lease lost for workflow ${workflowId}`);
    this.name = "LeaseLostError";
  }
}

Cloudflare KV を backing store にした実装が以下です。

// kv-checkpoint-store.ts
import type { CheckpointRecord, CheckpointStore } from "./checkpoint-store";
 
const LEASE_MS = 60_000; // リースは 60 秒。ステップ実行時間より長く取ること
 
export class KvCheckpointStore implements CheckpointStore {
  constructor(
    private kv: KVNamespace,
    private owner: string // crypto.randomUUID() を起動時に 1 回だけ生成して渡す
  ) {}
 
  async load(workflowId: string): Promise<CheckpointRecord | null> {
    return (await this.kv.get(`wf:${workflowId}`, "json")) as CheckpointRecord | null;
  }
 
  async compareAndSave(
    workflowId: string,
    next: Pick<CheckpointRecord, "state">,
    expectedVersion: number
  ): Promise<boolean> {
    const current = await this.load(workflowId);
    const currentVersion = current?.version ?? 0;
 
    // 誰かが先にステップを進めている → 自分が持っている状態はもう古い
    if (currentVersion !== expectedVersion) return false;
 
    // 他インスタンスのリースがまだ生きている → 掴みにいかない
    const now = Date.now();
    if (current && current.owner !== this.owner && current.leaseUntil > now) {
      return false;
    }
 
    await this.kv.put(
      `wf:${workflowId}`,
      JSON.stringify({
        state: next.state,
        version: currentVersion + 1,
        owner: this.owner,
        leaseUntil: now + LEASE_MS,
      } satisfies CheckpointRecord)
    );
    return true;
  }
}

ひとつ正直に書いておきます。KV は結果整合なので、これは「事故の大半を止める仕掛け」であって厳密な排他ではありません。金額が動く処理や、二重実行が一度も許されない処理では、Cloudflare なら Durable Objects、それ以外なら Postgres の UPDATE ... SET version = version + 1 WHERE id = $1 AND version = $2 に置き換えてください。私の集計パイプラインは「二重通知さえ起きなければよい」水準なので、KV の CAS と通知側のべき等キーの二段で止めています。

呼び出し側は、保存が false を返した時点で降ります。

// durable-workflow.ts（saveCheckpoint を差し替える）
private version = 0;
 
private async saveCheckpoint(stepName: string): Promise<void> {
  this.state.lastCheckpoint = stepName;
  this.state.completedSteps.push(stepName);
 
  const ok = await this.store.compareAndSave(
    this.workflowId,
    { state: this.state },
    this.version
  );
  // 追い越された = このインスタンスはもう正当な実行者ではない
  if (!ok) throw new LeaseLostError(this.workflowId);
 
  this.version += 1;
}

肝心なのは、リトライループの中で LeaseLostError を握り潰さないことです。ここを他の例外と同じ扱いにすると、負けたはずのインスタンスがリトライで復活してきて、結局二重実行に戻ります。私は最初これをやって、対策を入れたつもりのまま同じ事故を一度繰り返しました。

「再開できる」を、壊して確かめる

チェックポイントを入れた直後というのは、たいてい「たぶん再開できるはず」という状態です。ここで一度痛い目を見ていて、Step 3 の再開時に Step 2 の結果を state.data から読み損ねていたのに、テストは緑のまま 3 週間動き続けていました。落ちなかったから気づかなかっただけです。

やることは単純です。任意のステップで意図的に例外を投げるフラグを本体に埋め込み、「落として、再開して、副作用が 1 回だけか」をテストで確かめます。

// fault-injection.ts
export interface FaultPlan {
  /** このステップ名に入った瞬間に投げる */
  throwAt?: string;
  /** true なら 1 回だけ投げる（再開後は素通りさせる） */
  once?: boolean;
}
 
export function createFaultInjector(plan: FaultPlan) {
  let fired = false;
  return (stepName: string): void => {
    if (plan.throwAt !== stepName) return;
    if (plan.once && fired) return;
    fired = true;
    throw new Error(`[fault-injection] forced failure at ${stepName}`);
  };
}
 
/** 本番用。分岐を残さないために、no-op を明示的に渡します */
export const noFault = (): void => {};

executeStep の冒頭で this.injectFault(stepName) を 1 行呼ぶだけです。本番では noFault を渡します。テスト側はこうなります。

// durable-workflow.test.ts （vitest）
import { describe, expect, it, vi } from "vitest";
import { DurableWorkflow } from "./durable-workflow";
import { MemoryCheckpointStore } from "./memory-checkpoint-store";
import { createFaultInjector, noFault } from "./fault-injection";
 
describe("障害注入からの再開", () => {
  it("Step 3 で落ちても Step 1-2 の副作用は 1 回だけ", async () => {
    const store = new MemoryCheckpointStore();
    const fetchRevenue = vi.fn().mockResolvedValue({ rows: 120 });
    const sendSlack = vi.fn().mockResolvedValue(undefined);
 
    const run = (injectFault: (s: string) => void) =>
      new DurableWorkflow("wf-test", { store, injectFault })
        .run({ fetchRevenue, sendSlack });
 
    // 1 回目: upsert の直前で強制的に落とす
    await expect(
      run(createFaultInjector({ throwAt: "upsert-supabase" }))
    ).rejects.toThrow("[fault-injection]");
 
    // 2 回目: 障害なしで再開させる
    await run(noFault);
 
    expect(fetchRevenue).toHaveBeenCalledTimes(1); // 完了済みステップが再実行されていない
    expect(sendSlack).toHaveBeenCalledTimes(1);    // 二重通知が起きていない
  });
});

このテストを for (const step of ALL_STEPS) でループさせると、「どこで落としても副作用は 1 回」を機械的に保証できます。CI に組み込んでおけば、新しいステップを足したときにケースの追加を忘れると落ちてくれます。

地味な仕組みですが、夜中に Slack が鳴らなくなった一番の理由は、実装そのものよりこのテストでした。再開できるかどうかは、祈るものではなく、壊して確かめるものだと思っています。

毎朝 04:30 に動く AdMob 集計パイプラインが救われた 3 つの障害

ここからは個人開発の現場の話です。私が運用している壁紙アプリ群の収益分析パイプラインは、毎朝 04:30 JST に起動して、以下の処理を順に行います。

AdMob Reporting API から国別 eCPM・インプレッション・収益を取得（約 11 分）
App Store Connect / Google Play Console からインストール数を取得（約 6 分）
Supabase の daily_revenue テーブルへ upsert（約 2 分）
集計から異常検知（前週比 ±30% を逸脱した国を抽出）
Slack に日次レポートを送信

旧アーキテクチャでは、上記のうち 1 ステップでも失敗するとフルリトライが必要で、再実行に 47 分かかっていました。Durable Execution を導入してからは、失敗ステップ以降だけ再実行で済むため、平均再実行時間は 4 分前後に収まっています。実際に救われた 3 つの障害事例を挙げます。

事例 1: AdMob API の `RESOURCE_EXHAUSTED` で 36 分目に死亡

AdMob Reporting API は、日次クォータを超えると RESOURCE_EXHAUSTED を返します。Step 1 がほぼ終わる 36 分目で発生したとき、旧来なら最初から取り直しで合計 70 分以上のロス。Durable Execution では、国別チャンク単位でチェックポイントを切っていたため、未完の 12 カ国だけを翌時間枠で再取得し、合計 8 分で復旧しました。

事例 2: App Store Connect API の OAuth トークン期限切れ

JWT トークンの 20 分有効期限を切り忘れて Step 2 の途中で 401 になったケース。チェックポイントを「アプリ単位」で切っていたため、トークンを再生成してから残りのアプリ ID 群だけを再取得。16 分の中断で済みました。同じ事故を起こさないために、現在は OAuth トークン生成自体を最初の executeStep に含めて、トークンを state.data に保持する設計に変えています。

事例 3: Supabase の Connection Pool 枯渇

異常検知ジョブ（Step 4）の並列度を上げすぎて Connection Pool 枯渇。旧来なら Step 1 から全リトライですが、Durable Execution では Step 1〜3 のキャッシュが効いて、Step 4 のみ並列度を 12 → 4 に下げて再実行。総再実行時間は 3 分でした。

これらの障害は「公式 SDK のドキュメントを読み込めば回避できる類のもの」ではなく、運用してはじめて踏むタイプの罠です。本番投入する前にチェックポイント粒度をどう切るかが、復旧時間を 10 倍以上変えます。

Trigger.dev を使った本格的な Durable Execution

プロダクション環境では、自前でチェックポイント管理を実装するよりも、専用のフレームワークを使うほうが堅牢です。Trigger.dev は TypeScript ネイティブの Durable Execution プラットフォームで、Antigravity との相性も抜群です。私の場合は AdMob 集計のような I/O が支配的なジョブは Trigger.dev、軽量な異常検知バッチは Cloudflare Workers + KV、と使い分けています。

// trigger/ai-processing.ts
// Trigger.dev v3 による Durable Execution ワークフロー
 
import { task, wait, retry } from "@trigger.dev/sdk/v3";
 
export const processLargeCodebase = task({
  id: "process-large-codebase",
  // 最大実行時間: 2時間
  maxDuration: 7200,
  retry: {
    maxAttempts: 5,
    factor: 2,
    minTimeoutInMs: 1000,
    maxTimeoutInMs: 60000,
  },
  run: async (payload: { repoUrl: string; branch: string }) => {
    // Step 1: リポジトリのクローン（自動チェックポイント）
    const cloneResult = await cloneRepo(payload.repoUrl, payload.branch);
    // Trigger.dev が自動的に状態を永続化
 
    // Step 2: ファイル解析（大量ファイルのバッチ処理）
    const files = await listSourceFiles(cloneResult.path);
    const analysisResults = [];
 
    for (const file of files) {
      // 各ファイルの解析結果が自動的にチェックポイントされる
      const analysis = await retry.onThrow(
        async () => {
          return await analyzeFileWithAI(file);
        },
        { maxAttempts: 3, randomize: true }
      );
      analysisResults.push(analysis);
    }
 
    // Step 3: レポート生成
    const report = generateReport(analysisResults);
 
    // Step 4: 通知
    await sendNotification({
      type: "analysis-complete",
      summary: report.summary,
      issuesFound: report.issues.length,
    });
 
    return report;
  },
});
 
async function cloneRepo(url: string, branch: string) {
  // リポジトリクローン処理
  return { path: `/tmp/repo-${Date.now()}` };
}
 
async function listSourceFiles(repoPath: string) {
  // ソースファイル一覧取得
  return ["src/index.ts", "src/utils.ts", "src/api/routes.ts"];
}
 
async function analyzeFileWithAI(filePath: string) {
  // AI による静的解析（外部API呼び出し）
  return { file: filePath, issues: [], score: 85 };
}
 
function generateReport(results: any[]) {
  return {
    summary: `Analyzed ${results.length} files`,
    issues: results.flatMap((r) => r.issues),
    averageScore: results.reduce((s, r) => s + r.score, 0) / results.length,
  };
}
 
async function sendNotification(data: any) {
  console.log("Notification sent:", data);
}

Trigger.dev を使うと、各関数呼び出しの結果が自動的に永続化されるため、チェックポイント管理のコードを自前で書く必要がありません。サーバーが再起動しても、最後に成功したステップの直後から処理が再開されます。私の個人開発の感覚では、ジョブ実行時間が 30 分以上 or 外部 API への呼び出し ≥ 5 種類になったら、自前実装ではなく Trigger.dev / Inngest / Temporal のいずれかに寄せたほうが結果的に開発時間が短くなります。

べき等性を確保するための実装テクニック

Durable Execution で最も重要かつ見落としがちなのが、べき等性の確保です。リトライ時に副作用が重複しないように、以下のテクニックを活用しましょう。

// idempotency-patterns.ts
// べき等性を確保するための実装パターン集
 
import { randomUUID } from "crypto";
 
// パターン 1: べき等キーによる重複排除
class IdempotentExecutor {
  private processedKeys = new Set<string>();
 
  async execute(
    idempotencyKey: string,
    operation: () => Promise<void>
  ): Promise<void> {
    // 既に処理済みならスキップ
    if (this.processedKeys.has(idempotencyKey)) {
      console.log(`Already processed: ${idempotencyKey}`);
      return;
    }
 
    await operation();
    this.processedKeys.add(idempotencyKey);
  }
}
 
// パターン 2: Upsert による DB 書き込みのべき等化
async function upsertRecord(db: any, record: { id: string; data: any }) {
  // INSERT ... ON CONFLICT DO UPDATE で同じレコードを何度書いても結果が同じ
  await db.query(
    `INSERT INTO records (id, data, updated_at)
     VALUES ($1, $2, NOW())
     ON CONFLICT (id) DO UPDATE SET data = $2, updated_at = NOW()`,
    [record.id, JSON.stringify(record.data)]
  );
  // 期待する出力: 1行が挿入または更新される（重複なし）
}
 
// パターン 3: トランザクションIDによる決済のべき等化
async function processPayment(orderId: string, amount: number) {
  const transactionId = `txn_${orderId}_${amount}`;
  // Stripe の idempotencyKey を使用
  const payment = await stripe.paymentIntents.create(
    {
      amount: amount,
      currency: "jpy",
      metadata: { orderId },
    },
    {
      idempotencyKey: transactionId,
    }
  );
  return payment;
  // 同じ transactionId で何度呼んでも1回分しか課金されない
}
 
const stripe = { paymentIntents: { create: async (...args: any[]) => ({}) } };

私の場合、AdMob 収益を Supabase に書き込むときは (date, country, app_id) の複合主キーで upsert、Slack の日次レポート送信は (date) の report_sent_log テーブルへの INSERT が成功した時だけ通知する、という二段構えで運用しています。これだけで二重通知の事故は完全にゼロになりました。

Antigravity エージェントとの統合パターン

Antigravity のマルチエージェント機能と Durable Execution を組み合わせると、複数のエージェントが協調して長時間タスクを処理する堅牢なシステムを構築できます。

// agent-durable-workflow.ts
// Antigravity エージェントの処理を Durable に管理する
 
interface AgentTask {
  id: string;
  type: "code-review" | "refactor" | "test-gen";
  targetFiles: string[];
  status: "pending" | "running" | "completed" | "failed";
  result?: unknown;
}
 
class DurableAgentOrchestrator {
  private tasks: Map<string, AgentTask> = new Map();
  private checkpointFile: string;
 
  constructor(sessionId: string) {
    this.checkpointFile = `.agent-state/${sessionId}.json`;
    this.restore();
  }
 
  private restore(): void {
    try {
      const fs = require("fs");
      if (fs.existsSync(this.checkpointFile)) {
        const saved = JSON.parse(
          fs.readFileSync(this.checkpointFile, "utf-8")
        );
        this.tasks = new Map(Object.entries(saved.tasks));
        console.log(
          `Restored ${this.tasks.size} tasks from checkpoint`
        );
      }
    } catch {
      // 新規セッション
    }
  }
 
  private checkpoint(): void {
    const fs = require("fs");
    const path = require("path");
    fs.mkdirSync(path.dirname(this.checkpointFile), { recursive: true });
    const serialized = {
      tasks: Object.fromEntries(this.tasks),
      savedAt: new Date().toISOString(),
    };
    fs.writeFileSync(
      this.checkpointFile,
      JSON.stringify(serialized, null, 2)
    );
  }
 
  async submitTask(task: Omit<AgentTask, "status">): Promise<string> {
    const existing = this.tasks.get(task.id);
    if (existing?.status === "completed") {
      console.log(`Task ${task.id} already completed, returning cached result`);
      return task.id;
    }
 
    this.tasks.set(task.id, { ...task, status: "pending" });
    this.checkpoint();
    return task.id;
  }
 
  async runAll(): Promise<Map<string, AgentTask>> {
    const pending = [...this.tasks.values()].filter(
      (t) => t.status !== "completed"
    );
 
    for (const task of pending) {
      task.status = "running";
      this.checkpoint();
 
      try {
        // Antigravity エージェントへタスクを委譲
        task.result = await this.delegateToAgent(task);
        task.status = "completed";
      } catch (error) {
        task.status = "failed";
        console.error(`Task ${task.id} failed:`, error);
      }
      this.checkpoint();
    }
 
    return this.tasks;
  }
 
  private async delegateToAgent(task: AgentTask): Promise<unknown> {
    // 実際にはAntigravityのエージェントAPIを呼び出す
    console.log(`Delegating ${task.type} to agent for ${task.targetFiles}`);
    return { reviewed: true, suggestions: [] };
  }
}
 
// 使用例
async function main() {
  const orchestrator = new DurableAgentOrchestrator("session-20260330");
 
  await orchestrator.submitTask({
    id: "review-auth",
    type: "code-review",
    targetFiles: ["src/auth/login.ts", "src/auth/session.ts"],
  });
 
  await orchestrator.submitTask({
    id: "refactor-api",
    type: "refactor",
    targetFiles: ["src/api/routes.ts"],
  });
 
  await orchestrator.submitTask({
    id: "gen-tests",
    type: "test-gen",
    targetFiles: ["src/utils/parser.ts"],
  });
 
  const results = await orchestrator.runAll();
  console.log("All tasks complete:", results.size);
  // 期待する出力: All tasks complete: 3
}
 
main().catch(console.error);

より高度なマルチエージェント設計パターンについては、マルチエージェント・オーケストレーション実践ガイドで詳しく解説しています。

公式ドキュメントには書かれていない、本番運用 7 つの落とし穴

ここからが、私が現場で実際に踏んだ落とし穴です。Trigger.dev / Inngest / Temporal の公式ドキュメントには載っていないか、載っていても読み飛ばしがちな項目を、優先度順に並べました。

1. チェックポイント JSON が肥大化して書き込みが遅くなる

最初に踏むのがこれです。state.data に取得した全レコードを入れてしまうと、AdMob のような数万件のデータでは JSON が 30〜80 MB に膨らみ、書き込みのたびに数秒のロスが発生します。生データは S3 / R2 に逃がし、state.data には「保存先 URI」だけを持つのが正解。私の集計パイプラインではこの修正だけで Step 1 が 11 分 → 7 分に短縮されました。

2. 「再実行で成功した」だけで満足すると二重通知が起きる

Slack 通知が「成功 → 通知 → 再起動」の順だと、再実行時にもう一度通知されます。対策は、通知ステップそのものに report_sent_log テーブルを噛ませて INSERT が成功した時だけ通知すること。べき等キーは (date) で十分です。これを抜くと、毎朝同じ日次レポートが 2 回飛ぶ事故になります。

3. レートリミットを「リトライ前提」で組むと別のジョブを巻き込む

外部 API のレートリミットを RESOURCE_EXHAUSTED → 指数バックオフ、で組むのは正しいのですが、そのジョブが同じトークンを共有する他のジョブを巻き込んで全滅することがあります。AdMob では Google Cloud Project 単位でクォータを切っているので、集計ジョブとは別プロジェクトでテレメトリ用のクライアントを立てておく、という分離が必要でした。

4. 部分復旧後に「直近のデータだけ」が壊れる

チェックポイント時刻 T1 で停止し、T2 に再開した場合、T1 と T2 の間の「リアルタイム性のあるデータ」が欠落することがあります。AdMob は「1 時間遅延あり」なので、T2 で再開しても T1 時点のデータをそのまま処理してしまうと、後段の異常検知が誤動作します。checkpoint には対象期間 [from, to] を明示的に保存し、再開時にそれを使うこと。

5. 時刻ドリフト — UTC と JST の混在事故

これは私が一度盛大にやらかしました。AdMob は UTC、App Store は America/Los_Angeles、Google Play は America/Los_Angeles、Slack 通知は JST。チェックポイントに「日付」を文字列で持つと、復旧時に別の日のデータを上書きするリスクがあります。checkpoint には常に ISO 8601 のタイムゾーン付き文字列で保存し、表示時のみ JST に変換するルールを徹底するのが鉄則です。

6. Cold Start で初回チェックポイントが書けない

Cloudflare Workers / AWS Lambda の Cold Start 中に、最初のチェックポイント書き込み（KV / S3 への PUT）が間に合わずタイムアウト、というケースがあります。起動直後の最初のステップは、必ず軽量な「ヘルスチェック書き込み」だけにすること。私は state.lastCheckpoint = "boot-ok" を 0 ステップ目に明示的に挟むようにしています。

7. 監視盲点 — チェックポイントの「停滞」が検知できない

Durable Execution は失敗を吸収してくれる代わりに、「ずっと同じステップで止まっている」状態を作りやすいです。エラーは出ていないのに進まない、という最も怖いパターン。対策は「saveCheckpoint の時刻を Prometheus / GA4 のカスタムイベントに送り、15 分以上更新がなければアラート」という監視を入れること。これがあるかないかで、夜間ジョブの安心感が段違いになります。

Antigravity でこの仕組みをどう生成するか

最後に、Antigravity のエージェントに Durable Workflow を生成させるときに私が使っているプロンプト設計を共有します。AdMob 集計パイプラインを 1 ヶ月で形にできたのは、結局この 4 工程を毎ジョブに繰り返したからでした。

工程 1: Plan モードで失敗パターンを列挙させる

いきなりコードを書かせず、まず「このジョブで発生しうる障害を 10 個列挙してください」とだけ指示します。レートリミット、トークン期限切れ、ネットワーク切断、データ量変動、依存サービスのデプロイ中... Antigravity が出してきた 10 個から、実装で対処するものを 5〜7 個に絞り込んで指示書に転記。

工程 2: ステップ境界とチェックポイント粒度を先に決める

「処理を 5 ステップ以下に分けて、各ステップの入出力を型で書いてください」と指示します。これを先にやらないと、Antigravity は 1 関数に全部詰め込みがちです。型を先に固めれば、チェックポイント粒度も自然に決まります。

工程 3: 失敗時の挙動を「ステップごとに」発注する

「Step 3 が失敗したとき、Step 1 と 2 の結果を破棄せず、Step 3 だけリトライしてください」のように、ステップ単位で再開挙動を要求すると、Antigravity の出力品質が体感 2 倍くらいになります。複数を 1 プロンプトに混ぜないこと。

工程 4: Sandbox 実行で「障害注入テスト」を必ず行う

完成コードを動かして終わり、ではなく、任意のステップで例外を投げて再開できるかどうかを必ず検証します。前述の createFaultInjector を Antigravity にもセットで書かせておくと、生成された直後から「壊して確かめる」状態に入れます。ここで手を抜くと、1 ヶ月後に本番で泣くのは確定です。

最後に

Durable Execution は、AI エージェントによる長時間タスクを安定運用するための、地味だけれど一番効く設計です。私自身、毎朝動かしているパイプラインを Durable 化してからは、夜中に Slack が鳴る回数が月 12 回 → 1 回程度まで減りました。次の一歩としておすすめなのは、今動かしているジョブの中で最も長いものを 1 つ選び、そのジョブのチェックポイントを 3 箇所だけ切ってみることです。それだけで、再実行のコストの体感が大きく変わります。

私自身もまだ運用しながら設計を磨き続けています。同じように個人開発で長時間タスクと向き合っている方の参考になれば、これほど嬉しいことはありません。最後までお読みいただき、ありがとうございました。