ブログ一覧

Cloudflare Computer の replay と callId で投稿を 1回にする

Web開発者向け

Cloudflare Computer（Durable Object の中でエージェントが書いたコードを動かすライブラリ）の 0.7.0 で、再起動で途切れた実行を再開する replay と、人の承認を待つ仕組みが入りました。そこで、Issue にコメントを投稿するようなエージェントがデプロイやクラッシュで止まったとき、投稿を二重にせずに続きから動けるのかを、手元の wrangler dev で確かめました。

試したのは、Issue を読む、コメントを投稿する、通知を送るの 3つの外部 API を順に呼ぶコード（LLM は使わない）です。replay を付けると、エージェントのコードに進捗の記録を 1行も書かずに最後まで動き、wrangler dev を kill -9 したときもソースを書き換えて読み込み直したときも、レスポンスが返った外部 API の呼び出しはもう一度送られませんでした。止まった瞬間に送っている最中だった投稿は、storage に進捗を書いて続きから動かす自前実装と同じく 2回送られましたが、外部 API の呼び出しごとに付く ID である callId を冪等キーとして API に渡すと、投稿は 1回で済みました。

replay は進捗を書かずに、済んだ呼び出しを繰り返さず完了する

途切れた実行を扱う Computer の文書によると、replay（onInterrupt: "replay"）は途切れた実行を最初から動かし直し、済んだ呼び出しには最初に動いたときに記録しておいた結果で答えます。済んだ呼び出しはもう一度送られないので、外から見ると続きから動いたことになります。

同じ作業を Durable Object の storage に「どのステップまで終わったか」を書く自前実装と replay の 2通りで書き、外部 API の代わりに置いたモック API で投稿の回数を数えました（@cloudflare/computer 0.8.0、wrangler 4.154.0）。Computer では、エージェントのコードは Computer の実行環境（Dynamic Worker）の中で動き、外部 API には Durable Object の側に書いた関数（以下 DO 側の関数）を通して届きます。エージェントのコードと、DO 側の関数を渡す Durable Object の側のコードは次のとおりです。

import { readIssue, postComment, notify } from "ws:api";
export default async function main(input) {
  const issue = await readIssue(input.issue);
  const comment = await postComment(input.issue, "summary: " + issue.title);
  await notify(comment.id);
  return { commentId: comment.id };
}
new WorkerJavaScriptBackend({
  loader: env.LOADER,
  onInterrupt: "replay",
  modules: {
    "ws:api": {
      readIssue: ([n]) => readIssue(Number(n)),
      postComment: ([n, body], { callId }) => postComment(Number(n), String(body), callId),
      notify: ([id]) => notify(String(id)),
    },
  },
}),

上の投稿は次の節で使う callId を渡す形で載せていて、対策なしで試すときはこの引数を渡していません。投稿の関数に何の対策もしないまま、通知の最中に wrangler dev を kill -9 して立ち上げ直すと、replay は次のリクエストで続きから動いて完了しました。Issue を読む呼び出しも投稿ももう一度は送られず、送っている最中だった通知だけが 2回送られていて、自前実装でも回数は同じでした。ソースを 1行書き換えて Durable Object が作り直されたときも、同じく続きから動いて完了しています。

投稿の回数は同じで、違ったのはコードに書く再開のための行数です。空行とコメントを除いて数えると、自前実装は 28行のうち 13行が再開のための行で、replay では onInterrupt: "replay" の 1行、エージェントのコードには 0行でした。Computer では外部 API の呼び出しごとに DO 側の関数を 1行、計 3行書いていますが、これは外部 API に届くための行なので数に入れていません。外部 API の呼び出しを 1つ足すとき、自前実装は再開のための行が 3行増え、Computer は DO 側の関数が 1行増えます。readIssue などは自前実装と同じ関数で、自前実装の再開の部分は次の形です（投稿は冪等キーを渡す形）。

async resume() {
  const job = await this.ctx.storage.get<Job>("job");
  if (!job) return { error: "no job" };
  if (job.step < 1) {
    job.title = (await readIssue(job.issue)).title;
    job.step = 1; await this.ctx.storage.put("job", job);
  }
  if (job.step < 2) {
    const key = `${this.ctx.id}:${job.issue}:comment`;
    job.commentId = (await postComment(job.issue, `summary: ${job.title}`, key)).id;
    job.step = 2; await this.ctx.storage.put("job", job);
  }
  if (job.step < 3) {
    await notify(job.commentId!);
    job.step = 3; await this.ctx.storage.put("job", job);
  }
  return job;
}

再開が始まるのは、誰かがその Durable Object を呼んだときのようです。立ち上げ直して 20秒何も呼ばなかった間は続きが動かず、実行の結果を取りに行くと動きました。

送っている最中だった投稿は callId を冪等キーにして 1回にする

Computer の文書によると、送っている最中だった呼び出しはレスポンスが返っていないので replay でもう一度動きます。手元でも、モック API がコメントを受け付けてレスポンスを返す前に kill -9 すると、対策なしでは replay でも自前実装でも投稿が 2回になりました。

投稿の最中に wrangler dev を止めたときの投稿の回数。自前実装と Computer の replay はどちらも対策なしで 2回、callId を冪等キーとして渡すと 1回、unsafeToRepeat() で包むと 1回で実行はエラーで止まる

上の DO 側の postComment のように、関数の 2つ目の引数から取った callId を冪等キーとして API に渡すと、再起動の後の 2回目にも同じ callId が来て API が弾き、投稿は 1回で完了しました。自前実装でも、冪等キーを自分で組み立てて渡せば同じく 1回です。文書は、冪等キーを受けない呼び出し（課金など）を unsafeToRepeat() で包むよう勧めています。投稿の関数を包んで試すと、再起動の後は送り直さずに実行がエラーで止まりました。

投稿の前に人の承認を挟むなら、DO 側の関数の中で 2つ目の引数の interrupt() を呼びます。実行は答えが来るまで止まり、Computer を操作する API である Workspace の pending() に待っている呼び出しが出ます。

postCommentApproved: async ([n, body], c) => {
  const { approved } = await c.interrupt({ kind: "approval", issue: Number(n), body: String(body) });
  if (!approved) throw new Error("The comment was rejected.");
  return postComment(Number(n), String(body), c.callId);
},

待っている間に wrangler dev を kill -9 して立ち上げ直しても pending() には同じ呼び出しが残り、respond() で承認すると実行は完了して投稿は 1回、承認しない答えを返すと実行は失敗しました。0.8.0 では Worker 側のクライアントに pending()・respond() が無いため、私が Durable Object にこれを呼ぶメソッドを足して、Worker から呼びました。文書の承認待ちの節によると、interrupt() より前の処理は承認の後にもう一度動くため、投稿は interrupt() の後に書きます。

承認を待っている実行が、wrangler dev を kill -9 して立ち上げ直した後も pending() に同じ中身で残り、承認すると完了して投稿が 1回になった端末の出力

書き込み先が冪等キーを受けるかで、callId と unsafeToRepeat() を使い分ける

外部 API の呼び出しを 1つ足すとき、自前実装は再開のための行が 3行増え、replay で再開のために書くのは onInterrupt: "replay" の 1行のままです。

使う前に知っておく事実もあります。replay と承認待ちはパッケージの README の状態の表で Experimental です。Worker 本体の wrangler.jsonc の互換性日付とは別に、Computer の実行環境にも互換性日付（compatibility date、compatibilityDate）があり、これを変えて読み込み直すと途中の実行は次のエラーで失敗しました。

Execution was interrupted and can't resume, because the JavaScript runtime changed since it started.

wrangler.jsonc の互換性日付を変えたときは失敗していません。文書の制限の節によると、replay の実装が変わるパッケージの更新でも途中の実行は失敗します。待つ人がいない実行のために alarm から recover() を呼ぶ形も、文書の再開の条件に示されているだけで試していません。

手元で試していないのは、ほかに次の 3つです。

  1. 本番の Cloudflare の上でのデプロイとクラッシュ
  2. 外部 API の呼び出しを DO 側の関数にせず、エージェントのコードの fetch で直接書く場合（文書では、送っている最中だった fetch は送り直される）
  3. 本物の LLM がその場で書いたコードや、ツールを呼ぶエージェント

試した範囲では、書き込み先が冪等キーを受けるなら、DO 側の関数で callId を渡すだけで、進捗の記録を書かずに投稿を 1回にできました。受けない呼び出しは unsafeToRepeat() で包み、途中で止まった実行は失敗にして、その後どうするかを人が決めます。投稿の前に人の承認を挟みたいなら interrupt() を使い、待っている間に再起動しても待ちは残りました。

この記事をシェア

ブログ一覧