Claude Media
Claude CodeでTemporalのワークフローを書く — 決定性違反をテストで見つける
Claude Codeチュートリアル

Claude CodeでTemporalのワークフローを書く — 決定性違反をテストで見つける

Claude CodeにTemporal(Python)のワークフローを書かせるとき、非決定的な処理を持ち込ませない規約と、テストスイートを完了条件にする設定をまとめます。

読了目安 約27分

Temporalのワークフローでは、Claude Codeに何を書かせないか

Temporalのワークフローコードは、再生(replay)されても同じ順序で同じ処理を呼ぶ決定的なコードでなければなりません。Claude Codeに書かせるときの要点は、非決定的な処理をワークフローに書かせないことと、書いてしまっても自動で気づける状態にしておくことの2つです。

Temporal Python SDKの公式ドキュメントは、ワークフローコードで避けるものを次のように挙げています。

  • スレッドの利用
  • 乱数
  • 外部プロセスの呼び出し
  • ネットワークI/O
  • グローバル状態の書き換え
  • システムの日付・時刻

普通のPythonコードとしては自然な書き方ばかりです。datetime.now()もuuid.uuid4()も、何も指示しなければ生成AIは平然と書きます。そこで本記事では、規約をCLAUDE.mdに置き、テストスイートとStopフックで完了条件にする流れを順に組みます。同じ「ワークフローをClaude Codeで書く」型でも、Pythonの関数にデコレータを足すPrefectとは前提が違います。Prefectの手順はClaude CodeでPrefectのワークフローを書くにあります。

CLAUDE.mdに置く決定性の規約

CLAUDE.mdは強制ではなく文脈として読まれます。Claude Codeのドキュメントも、検証できる具体さで書くことを勧めています。そこで「決定的に書くこと」のような抽象語ではなく、禁止する呼び出しと置き換え先を対で書きます。

置き換え先は、TemporalのPython SDKガイドにあるものです。

ワークフロー内で書かせない代わりに使うもの
datetime.now() / time.time()代わりに使うものworkflow.now()
random.random() などのrandomモジュール代わりに使うものworkflow.random()
uuid.uuid4()代わりに使うものworkflow.uuid4()
print() / 標準のlogging代わりに使うものworkflow.logger
API呼び出し・DB・LLM呼び出しなどの外部通信代わりに使うものActivityに切り出す

workflow.now()は直近のWorkflow Taskの時刻を返すため、再生しても値が変わりません。外部との通信をActivityに出す理由もガイドに書かれています。Activityは再生の経路の外で実行され、自動で再試行されるので、非決定性エラーの原因になりません。

これをtests/とworkflows/の置き場所とあわせてCLAUDE.mdに書くと、次のような形になります。

## Temporal(Python)の規約
 
- `workflows/` 配下のコードは決定的に書く。次は禁止:
  `datetime.now()`, `time.time()`, `random.*`, `uuid.uuid4()`,
  `print()`, ネットワークI/O, ファイル・プロセス操作, スレッド
- 時刻・乱数・UUID・ログは `workflow.now()` / `workflow.random()` /
  `workflow.uuid4()` / `workflow.logger` を使う
- 外部サービスの呼び出しは `activities/` に書き、ワークフローから呼ぶ
- `sandbox_unrestricted()` と `@workflow.defn(sandboxed=False)` は使わない
- 変更後は `python -m pytest tests -q` を通す。落ちたら直してから完了とする

最後の2行が効きどころです。サンドボックスを外す手段はサンドボックスのページに載っていますが、外すと決定性の検査も外れます。Claudeがテストを通すために安易に使わないよう、禁止として明示します。

テストスイートを完了条件にする3層

Temporalのテストは、エンドツーエンド、統合、ユニットの3種類に分けられます。テストの大半は統合テストで書くことが勧められています。Claudeに書かせるテストも、次の3つを揃えておくと決定性の違反と時間まわりの不具合を拾えます。

  1. Activityをモックしたワークフローの統合テスト
  2. 時間をスキップするテスト
  3. 保存した履歴によるリプレイテスト

Activityをモックして統合テストを書く

WorkerにモックのActivityを渡すと、ワークフローのロジックだけを切り離して試せます。注意点は、モックの名前とシグネチャを本物と同じにすることです。名前がずれるとワークフローは本物のActivityを探しに行きます。この点はClaudeへの指示文に入れておく価値があります。

import uuid
 
import pytest
from temporalio import activity
from temporalio.testing import WorkflowEnvironment
from temporalio.worker import Worker
 
from workflows import GreetingWorkflow
 
 
@activity.defn(name="compose_greeting")
async def compose_greeting_mocked(name: str) -> str:
    return f"Hello, {name} from mocked activity!"
 
 
@pytest.mark.asyncio
async def test_greeting_with_mocked_activity():
    task_queue = str(uuid.uuid4())
    async with await WorkflowEnvironment.start_local() as env:
        async with Worker(
            env.client,
            task_queue=task_queue,
            workflows=[GreetingWorkflow],
            activities=[compose_greeting_mocked],
        ):
            result = await env.client.execute_workflow(
                GreetingWorkflow.run,
                "World",
                id=str(uuid.uuid4()),
                task_queue=task_queue,
            )
            assert result == "Hello, World from mocked activity!"

これはテストスイートのページの例に沿って組んだ形で、GreetingWorkflowは説明用の名前です。start_local()はローカルのTemporalサーバーを立てるので時間のスキップはありません。タイマーは実時間で待ちます。

時間をスキップして長いsleepを検証する

数日待つワークフローを実時間で試すことはできません。インメモリのテストサーバーを使うstart_time_skipping()なら、タイマー(sleepと条件付きタイムアウトを含む)が早送りされます。ただしActivityの実行中は早送りされない点に注意が必要です。

from datetime import timedelta
 
from temporalio import workflow
 
 
@workflow.defn
class ReminderWorkflow:
    @workflow.run
    async def run(self, user: str) -> str:
        await workflow.sleep(timedelta(days=1))
        return f"reminded {user}"
@pytest.mark.asyncio
async def test_reminder_skips_one_day():
    task_queue = str(uuid.uuid4())
    async with await WorkflowEnvironment.start_time_skipping() as env:
        async with Worker(
            env.client, task_queue=task_queue, workflows=[ReminderWorkflow]
        ):
            result = await env.client.execute_workflow(
                ReminderWorkflow.run,
                "alice",
                id=str(uuid.uuid4()),
                task_queue=task_queue,
            )
            assert result == "reminded alice"

テストスイートのページによると、時間はWorkflowEnvironmentのインスタンス全体で共有されます。自動スキップと手動スキップのテストを同じインスタンスで並走させると互いに影響するため、モードごとにテストをまとめるか、インスタンスを分けます。Claudeに複数のテストを書かせるときは、この分離も指示に入れます。手動で進めたいときはenv.sleep(秒数)で時間を進められます。

履歴を再生して決定性の違反を検出する

決定性の違反を直接見つけるのは、リプレイです。CIには次の手順を入れることが勧められています。

  1. テスト対象のWorkerが扱うワークフロー型かタスクキューを決める
  2. 各タスクキューから、直近の実行中・完了済みワークフローのイベント履歴を代表的な数だけ取得する
  3. 履歴をリプレイにかける
  4. エラーが1件でもあればCIを落とす

履歴を取得してJSONとしてtests/histories/に置いておけば、Claudeがワークフローを書き換えるたびに既存の履歴との互換性を検査できます。単一の履歴をJSON文字列から再生する例は次のとおりです。

replayer = Replayer(workflows=[YourWorkflow])
await replayer.replay_workflow(WorkflowHistory.from_json(history_json_str))

イベント履歴が非決定的であれば、エラーが投げられます。複数の履歴を再生するreplay_workflowsでは、fail_fastをfalseにすると全履歴の再生を待ってから結果を見られます。

履歴を保存するときの落とし穴が1つあります。サーバーから直接取得した履歴やエクスポートした履歴はprotobufエンコードのbytesのことがあり、Replayerが扱うのはデコード済みのdictであることが多いとされています。dictとbytesの食い違いでTypeErrorが出たら、デコードしてから渡します。Claudeにリプレイテストを書かせた直後にこのエラーが出たときは、まずここを疑えます。

Stopフックでテストが通るまで終わらせない

CLAUDE.mdは文脈にすぎないので、テストを走らせ忘れる余地が残ります。確実に走らせたいならStopフックです。StopはClaudeが応答を終えるときに発火し、フックがブロックすればClaudeは止まらずに作業を続けます。コマンド型のフックでは、終了コード2でブロックし、標準エラーに理由を書きます。

#!/bin/bash
# .claude/hooks/run-temporal-tests.sh
input=$(cat)
 
# すでにStopフックの結果を受けて続行中なら、無限ループを避けて通す
if [ "$(echo "$input" | jq -r '.stop_hook_active')" = "true" ]; then
  exit 0
fi
 
if ! out=$(python -m pytest tests -x -q 2>&1); then
  echo "Temporalのテストが失敗しています。直してから完了してください。" >&2
  echo "$out" | tail -30 >&2
  exit 2
fi
{
  "hooks": {
    "Stop": [
      {
        "hooks": [
          {
            "type": "command",
            "command": ".claude/hooks/run-temporal-tests.sh"
          }
        ]
      }
    ]
  }
}

stop_hook_activeは、Claude Codeがすでにストップフックの結果で続行しているときにtrueになるフィールドで、解消しない条件でブロックし続けないための確認用です。連続して続行できる回数にも上限があります。上限を調整する環境変数はCLAUDE_CODE_STOP_HOOK_BLOCK_CAPの記事にまとめています。フックの種類やイベントの全体像はClaude Code Hooks完全ガイドで確認できます。

上のスクリプトは、stop_hook_activeがtrueのときに通す設計です。そのため、1回差し戻されたあとの再テストはこのフックでは行われません。差し戻し後もテストを必須にしたい場合は、このガードを外して上限に任せる設計もありますが、通らないテストでセッションが長引くおそれがあります。トレードオフを理解したうえで選んでください。

決定性違反を意図的に仕込んで確認する

規約とフックを入れたら、検出が働くことを一度確かめておきます。ワークフローにdatetime.now()を書いたブランチを作り、テストを走らせます。

Python SDKのサンドボックスはワークフローの実行時に既知の非決定的な呼び出しを制限します。非決定的な処理が実行されると例外が投げられ、Workerのタスクが失敗します。そしてコードが直るまでワークフローは進みません。

つまり、違反は「テストが例外で落ちる」とは限らず、「ワークフローが完了しない」形で現れます。この場合のexecute_workflowは待ち続けることがあるため、確認用のテストには実行タイムアウトを付けておくと、ハングせずに失敗として拾えます。ここはサンドボックスの説明からの推論なので、手元のSDKバージョンで一度動かして挙動を確かめてください。

サンドボックスには限界もあります。サンドボックスは完全には隔離されておらず、ライブラリが内部で状態を書き換えると決定性が壊れうると明記されています。サンドボックスが通ったから安全とは言えません。そのためリプレイテストを併用します。

もう1つ、passthrough_modulesの扱いにも注意が要ります。サードパーティのモジュールは、通常ワークフローの実行ごとに再読み込みされます。パススルーに指定すると再読み込みされなくなり、性能は上がりますが、副作用のないモジュールだけに使うよう注意書きがあります。Claudeがimportの遅さを理由に安易にパススルーを足さないよう、CLAUDE.mdの規約に「パススルーの追加は人が判断する」と一行足しておくと安全です。

テストの設計で押さえる点

  • 統合テストを中心に、ユニットは補助にします。テストスイートのページの推奨に沿った配分です。
  • Activity単体のテストにはActivityEnvironmentを使います。on_heartbeatにコールバックを設定すれば、Heartbeatの内容を検証できます。
  • pytestを使う場合は、フィクスチャでテスト環境の起動と終了をまとめられます。
  • 履歴のリプレイは、ワークフロー定義を変えるたびに回します。定義の変更が走行中のワークフローと互換かどうかは、リプレイでしか検出できません。

ワークフローの定義を変えるときは、コマンドの順序を変えること自体が非決定性エラーの原因になります。ワークフロー定義のページでは、タイマーの開始、Activityの実行予約、子ワークフローの開始、ワークフロー終了などがコマンドを生成する呼び出しであり、バージョニングの手当てなしに順序の入れ替え・追加・削除をしてはならないとしています。Claudeに既存ワークフローを修正させるときは、「走行中のワークフローがあるなら、まずリプレイテストを通す」ことを指示に加えます。テストの書き方そのものはClaude Codeで単体テストを書かせる手順も参考になります。

まとめ

Temporalのワークフローで怖いのは、動くのに再生で壊れるコードです。規約をCLAUDE.mdに置くだけでは守られる保証がなく、モック付きの統合テスト、時間のスキップ、リプレイをStopフックにつないで初めて「通るまで終わらない」状態になります。サンドボックスを外す手段をClaudeに使わせないことと、履歴のリプレイをCIに常設することが、この構成の肝です。

この記事を共有:XはてブLinkedIn

関連する記事

Claude Code をもっと見る →