チェックポイントによるファイル変更の巻き戻し

# 空のプロンプトでセッションを再開し、巻き戻す
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

import asyncio
import os
from claude_agent_sdk import ClaudeSDKClient, ClaudeAgentOptions, UserMessage, ResultMessage

async def main():
    # ステップ1: チェックポイントを有効にする
    options = ClaudeAgentOptions(
        enable_file_checkpointing=True,
        permission_mode="acceptEdits",  # プロンプトなしでファイル編集を自動承認
        extra_args={"replay-user-messages": None},  # レスポンスストリームでチェックポイントUUIDを受信するために必要
        env={**os.environ, "CLAUDE_CODE_ENABLE_SDK_FILE_CHECKPOINTING": "1"}
    )

    checkpoint_id = None
    session_id = None

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

        # ステップ2: 最初のユーザーメッセージからチェックポイント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

    # ステップ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 asyncio
import os
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},
        env={**os.environ, "CLAUDE_CODE_ENABLE_SDK_FILE_CHECKPOINTING": "1"}
    )

    safe_checkpoint = 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:
                safe_checkpoint = message.uuid

            # 独自のロジックに基づいて巻き戻すタイミングを決定
            # 例：エラー検出、バリデーション失敗、またはユーザー入力
            if your_revert_condition and safe_checkpoint:
                await client.rewind_files(safe_checkpoint)
                # 巻き戻し後にループを終了、ファイルは復元済み
                break

asyncio.run(main())

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

# より良い追跡のためにチェックポイントメタデータを保存
@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},
        env={**os.environ, "CLAUDE_CODE_ENABLE_SDK_FILE_CHECKPOINTING": "1"}
    )

    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

    # 後で：セッションを再開して任意のチェックポイントに巻き戻す
    if checkpoints and session_id:
        target = checkpoints[0]  # 任意のチェックポイントを選択
        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())