체크포인팅으로 파일 변경사항 되돌리기

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