checkpointing でファイル変更を巻き戻す

import asyncio
from claude_agent_sdk import (
    ClaudeSDKClient,
    ClaudeAgentOptions,
    UserMessage,
    ResultMessage,
)


async def main():
    # Step 1: checkpointing を有効にする
    options = ClaudeAgentOptions(
        enable_file_checkpointing=True,
        permission_mode="acceptEdits",  # プロンプトなしでファイル編集を自動承認
        extra_args={
            "replay-user-messages": None
        },  # レスポンスストリームで checkpoint UUID を受け取るために必須
    )

    checkpoint_id = None
    session_id = None

    # クエリを実行し、checkpoint UUID とセッション ID をキャプチャ
    async with ClaudeSDKClient(options) as client:
        await client.query("Refactor the authentication module")

        # Step 2: 最初のユーザーメッセージから checkpoint UUID をキャプチャ
        async for message in client.receive_response():
            if isinstance(message, UserMessage) and message.uuid and not checkpoint_id:
                checkpoint_id = message.uuid
            if isinstance(message, ResultMessage) and not session_id:
                session_id = message.session_id

    # Step 3: 後で、空のプロンプトでセッションを再開して巻き戻す
    if checkpoint_id and session_id:
        async with ClaudeSDKClient(
            ClaudeAgentOptions(enable_file_checkpointing=True, resume=session_id)
        ) as client:
            await client.query("")  # 接続を開くための空のプロンプト
            async for message in client.receive_response():
                await client.rewind_files(checkpoint_id)
                break
        print(f"Rewound to checkpoint: {checkpoint_id}")


asyncio.run(main())

import { query } from "@anthropic-ai/claude-agent-sdk";

async function main() {
  // Step 1: checkpointing を有効にする
  const opts = {
    enableFileCheckpointing: true,
    permissionMode: "acceptEdits" as const, // プロンプトなしでファイル編集を自動承認
    extraArgs: { "replay-user-messages": null } // レスポンスストリームで checkpoint UUID を受け取るために必須
  };

  const response = query({
    prompt: "Refactor the authentication module",
    options: opts
  });

  let checkpointId: string | undefined;
  let sessionId: string | undefined;

  // Step 2: 最初のユーザーメッセージから checkpoint UUID をキャプチャ
  try {
    for await (const message of response) {
      if (message.type === "user" && message.uuid && !checkpointId) {
        checkpointId = message.uuid;
      }
      if ("session_id" in message && !sessionId) {
        sessionId = message.session_id;
      }
    }
  } catch (error) {
    // A single-shot query() throws after yielding an error result. If the
    // failure was an error result, sessionId and checkpointId were already
    // captured by the loop above; connection or process failures yield no
    // result message.
    console.error(`Session ended with an error: ${error}`);
  }

  // Step 3: 後で、空のプロンプトでセッションを再開して巻き戻す
  if (checkpointId && sessionId) {
    const rewindQuery = query({
      prompt: "", // 接続を開くための空のプロンプト
      options: { ...opts, resume: sessionId }
    });

    for await (const msg of rewindQuery) {
      await rewindQuery.rewindFiles(checkpointId);
      break;
    }
    console.log(`Rewound to checkpoint: ${checkpointId}`);
  }
}

main();

import asyncio
from claude_agent_sdk import ClaudeSDKClient, ClaudeAgentOptions, UserMessage


async def main():
    options = ClaudeAgentOptions(
        enable_file_checkpointing=True,
        permission_mode="acceptEdits",
        extra_args={"replay-user-messages": None},
    )

    safe_checkpoint = None

    async with ClaudeSDKClient(options) as client:
        await client.query("Refactor the authentication module")

        async for message in client.receive_response():
            # 各エージェントターンが開始する前に checkpoint を更新
            # これは前の checkpoint を上書きします。最新のみを保持
            if isinstance(message, UserMessage) and message.uuid:
                safe_checkpoint = message.uuid

            # 独自のロジックに基づいて復帰するかどうかを決定
            # 例：エラー検出、検証失敗、またはユーザー入力
            if your_revert_condition and safe_checkpoint:
                await client.rewind_files(safe_checkpoint)
                # 巻き戻し後、ループを終了します。ファイルは復元されます
                break


asyncio.run(main())

import { query } from "@anthropic-ai/claude-agent-sdk";

async function main() {
  const response = query({
    prompt: "Refactor the authentication module",
    options: {
      enableFileCheckpointing: true,
      permissionMode: "acceptEdits" as const,
      extraArgs: { "replay-user-messages": null }
    }
  });

  let safeCheckpoint: string | undefined;

  for await (const message of response) {
    // 各エージェントターンが開始する前に checkpoint を更新
    // これは前の checkpoint を上書きします。最新のみを保持
    if (message.type === "user" && message.uuid) {
      safeCheckpoint = message.uuid;
    }

    // 独自のロジックに基づいて復帰するかどうかを決定
    // 例：エラー検出、検証失敗、またはユーザー入力
    if (yourRevertCondition && safeCheckpoint) {
      await response.rewindFiles(safeCheckpoint);
      // 巻き戻し後、ループを終了します。ファイルは復元されます
      break;
    }
  }
}

main();

import asyncio
from dataclasses import dataclass
from datetime import datetime
from claude_agent_sdk import (
    ClaudeSDKClient,
    ClaudeAgentOptions,
    UserMessage,
    ResultMessage,
)


# より良い追跡のために checkpoint メタデータを保存
@dataclass
class Checkpoint:
    id: str
    description: str
    timestamp: datetime


async def main():
    options = ClaudeAgentOptions(
        enable_file_checkpointing=True,
        permission_mode="acceptEdits",
        extra_args={"replay-user-messages": None},
    )

    checkpoints = []
    session_id = None

    async with ClaudeSDKClient(options) as client:
        await client.query("Refactor the authentication module")

        async for message in client.receive_response():
            if isinstance(message, UserMessage) and message.uuid:
                checkpoints.append(
                    Checkpoint(
                        id=message.uuid,
                        description=f"After turn {len(checkpoints) + 1}",
                        timestamp=datetime.now(),
                    )
                )
            if isinstance(message, ResultMessage) and not session_id:
                session_id = message.session_id

    # 後で：セッションを再開して任意の checkpoint に巻き戻す
    if checkpoints and session_id:
        target = checkpoints[0]  # 任意の checkpoint を選択
        async with ClaudeSDKClient(
            ClaudeAgentOptions(enable_file_checkpointing=True, resume=session_id)
        ) as client:
            await client.query("")  # 接続を開くための空のプロンプト
            async for message in client.receive_response():
                await client.rewind_files(target.id)
                break
        print(f"Rewound to: {target.description}")


asyncio.run(main())

import { query } from "@anthropic-ai/claude-agent-sdk";

// より良い追跡のために checkpoint メタデータを保存
interface Checkpoint {
  id: string;
  description: string;
  timestamp: Date;
}

async function main() {
  const opts = {
    enableFileCheckpointing: true,
    permissionMode: "acceptEdits" as const,
    extraArgs: { "replay-user-messages": null }
  };

  const response = query({
    prompt: "Refactor the authentication module",
    options: opts
  });

  const checkpoints: Checkpoint[] = [];
  let sessionId: string | undefined;

  try {
    for await (const message of response) {
      if (message.type === "user" && message.uuid) {
        checkpoints.push({
          id: message.uuid,
          description: `After turn ${checkpoints.length + 1}`,
          timestamp: new Date()
        });
      }
      if ("session_id" in message && !sessionId) {
        sessionId = message.session_id;
      }
    }
  } catch (error) {
    // 単一ショットの query() はエラー結果を生成した後にスローします。失敗がエラー結果だった場合、sessionId とチェックポイント配列は上記のループによってすでに入力されています。接続またはプロセスの失敗は結果メッセージを生成しません。
    console.error(`Session ended with an error: ${error}`);
  }

  // 後で：セッションを再開して任意の checkpoint に巻き戻す
  if (checkpoints.length > 0 && sessionId) {
    const target = checkpoints[0]; // 任意の checkpoint を選択
    const rewindQuery = query({
      prompt: "", // 接続を開くための空のプロンプト
      options: { ...opts, resume: sessionId }
    });

    for await (const msg of rewindQuery) {
      await rewindQuery.rewindFiles(target.id);
      break;
    }
    console.log(`Rewound to: ${target.description}`);
  }
}

main();