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 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 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())

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