Отмена изменений файлов с помощью контрольных точек

# Возобновите сеанс с пустым запросом, затем отмените
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"Отменено до контрольной точки: {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"Отменено до: {target.description}")

asyncio.run(main())