Control and observability

Rewind file changes with checkpointing

Rewind file changes with checkpointing 这一页讲的，就是 Rewind file changes with checkpointing 这件事在 Claude Code 里到底怎么用。

页面信息

对应原页

页面性质

第三方中文解释页

使用建议

先看人话解释，再对照原页命令和代码

这页不是官方原文，而是顺着官方文档结构做的中文解释版。命令、参数、配置名这些硬东西尽量保留，解释部分则尽量讲成人能照着做的话。

如果你碰到特别敏感的配置、权限或企业环境差异，最好顺手点上面的“查看原始文档”再核一遍。

这一页先讲明白

这页主要讲 Rewind file changes with checkpointing：Track file changes during agent sessions and restore files to any previous state

你可以把它当成"Control and observability"这块里专门管这一摊事的说明书。

你可以把"Rewind file changes with checkpointing"理解成 Control and observability 这一栏里的一把专门工具。这页不是让你背书，而是教你什么时候该把这把工具拿出来。

原文这页大多会按 How checkpointing works、Implement checkpointing、Common patterns、Checkpoint before risky operations 这些环节往下讲。

翻成人话，大概就是：How checkpointing works

第一，先别一上来全开全配。先按最小一步试通，确认没跑偏，再继续往下加。

第二，命令、配置名、参数名这些硬东西尽量保留原样。人话解释是帮你听懂，不是帮你改关键字。

第三，照着原文这几个环节挨个过：How checkpointing works -> Implement checkpointing -> Common patterns -> Checkpoint before risky operations。像下地先看水路、再试机器、再正式开干，一步一步最稳。

关键片段

原页关键片段：Implement checkpointing 1

"Implement checkpointing"这一段里最要紧的原始写法在下面，先看它怎么落地。

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


async def main():
    # Step 1: Enable checkpointing
    options = ClaudeAgentOptions(
        enable_file_checkpointing=True,
        permission_mode="acceptEdits",  # Auto-accept file edits without prompting
        extra_args={
            "replay-user-messages": None
        },  # Required to receive checkpoint UUIDs in the response stream
    )

    checkpoint_id = None
    session_id = None

    # Run the query and capture checkpoint UUID and session ID
    async with ClaudeSDKClient(options) as client:
        await client.query("Refactor the authentication module")

        # Step 2: Capture checkpoint UUID from the first user message
        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: Later, rewind by resuming the session with an empty prompt
    if checkpoint_id and session_id:
        async with ClaudeSDKClient(
            ClaudeAgentOptions(enable_file_checkpointing=True, resume=session_id)
        ) as client:
            await client.query("")  # Empty prompt to open the connection
            async for message in client.receive_response():
                await client.rewind_files(checkpoint_id)
                break
        print(f"Rewound to checkpoint: {checkpoint_id}")


asyncio.run(main())

关键片段

原页关键片段：Implement checkpointing 2

"Implement checkpointing"这一段里最要紧的原始写法在下面，先看它怎么落地。

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

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

关键片段

原页关键片段：Implement checkpointing 3

"Implement checkpointing"这一段里最要紧的原始写法在下面，先看它怎么落地。

checkpoint_id = None
session_id = None

async for message in client.receive_response():
    # Update checkpoint on each user message (keeps the latest)
    if isinstance(message, UserMessage) and message.uuid:
        checkpoint_id = message.uuid
    # Capture session ID from the result message
    if isinstance(message, ResultMessage):
        session_id = message.session_id

关键片段

原页关键片段：Checkpoint before risky operations

"Checkpoint before risky operations"这一段里最要紧的原始写法在下面，先看它怎么落地。

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():
            # Update checkpoint before each agent turn starts
            # This overwrites the previous checkpoint. Only keep the latest
            if isinstance(message, UserMessage) and message.uuid:
                safe_checkpoint = message.uuid

            # Decide when to revert based on your own logic
            # For example: error detection, validation failure, or user input
            if your_revert_condition and safe_checkpoint:
                await client.rewind_files(safe_checkpoint)
                # Exit the loop after rewinding, files are restored
                break


asyncio.run(main())

关键片段

原页关键片段：Multiple restore points

"Multiple restore points"这一段里最要紧的原始写法在下面，先看它怎么落地。

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


# Store checkpoint metadata for better tracking
@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

    # Later: rewind to any checkpoint by resuming the session
    if checkpoints and session_id:
        target = checkpoints[0]  # Pick any checkpoint
        async with ClaudeSDKClient(
            ClaudeAgentOptions(enable_file_checkpointing=True, resume=session_id)
        ) as client:
            await client.query("")  # Empty prompt to open the connection
            async for message in client.receive_response():
                await client.rewind_files(target.id)
                break
        print(f"Rewound to: {target.description}")


asyncio.run(main())

关键片段

原页关键片段：Try it out 1

这一段要真抓重点，通常就抓下面这块原文。

def add(a, b):
    return a + b


def subtract(a, b):
    return a - b


def multiply(a, b):
    return a * b


def divide(a, b):
    if b == 0:
        raise ValueError("Cannot divide by zero")
    return a / b

关键片段

原页关键片段：Try it out 2

这一段要真抓重点，通常就抓下面这块原文。

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


async def main():
    # Configure the SDK with checkpointing enabled
    # - enable_file_checkpointing: Track file changes for rewinding
    # - permission_mode: Auto-accept file edits without prompting
    # - extra_args: Required to receive user message UUIDs in the stream
    options = ClaudeAgentOptions(
        enable_file_checkpointing=True,
        permission_mode="acceptEdits",
        extra_args={"replay-user-messages": None},
    )

    checkpoint_id = None  # Store the user message UUID for rewinding
    session_id = None  # Store the session ID for resuming

    print("Running agent to add doc comments to utils.py...\n")

    # Run the agent and capture checkpoint data from the response stream
    async with ClaudeSDKClient(options) as client:
        await client.query("Add doc comments to utils.py")

        async for message in client.receive_response():
            # Capture the first user message UUID - this is our restore point
            if isinstance(message, UserMessage) and message.uuid and not checkpoint_id:
                checkpoint_id = message.uuid
            # Capture the session ID so we can resume later
            if isinstance(message, ResultMessage):
                session_id = message.session_id

    print("Done! Open utils.py to see the added doc comments.\n")

    # Ask the user if they want to rewind the changes
    if checkpoint_id and session_id:
        response = input("Rewind to remove the doc comments? (y/n): ")

        if response.lower() == "y":
            # Resume the session with an empty prompt, then rewind
            async with ClaudeSDKClient(
                ClaudeAgentOptions(enable_file_checkpointing=True, resume=session_id)
            ) as client:
                await client.query("")  # Empty prompt opens the connection
                async for message in client.receive_response():
                    await client.rewind_files(checkpoint_id)  # Restore files
                    break

            print(
                "\n✓ File restored! Open utils.py to verify the doc comments are gone."
            )
        else:
            print("\nKept the modified file.")


asyncio.run(main())

关键片段

原页关键片段：Try it out 3

这一段要真抓重点，通常就抓下面这块原文。

python try_checkpointing.py

关键片段

原页关键片段：”ProcessTransport is not ready for writing” error

"”ProcessTransport is not ready for writing” error"这一段里最要紧的原始写法在下面，先看它怎么落地。

# Resume session with empty prompt, then rewind
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

预留广告位

正文中段响应式广告 等你后面真接 AdSense，这里再放正式广告。

Documentation Index

这里不是让你背"Documentation Index"这个词，而是让你看它真干活时怎么使。

这里还牵扯作用域，意思就是这条规则到底管当前项目、你个人，还是只管这一趟会话。

How checkpointing works

这里讲底层是怎么运转的。你把它看明白，后面遇到异常时就知道该往哪儿查。

这里还牵扯作用域，意思就是这条规则到底管当前项目、你个人，还是只管这一趟会话。

Implement checkpointing

这一块主要是在说"Implement checkpointing"真到手上该怎么用，哪里最容易踩坑。

这里还牵扯作用域，意思就是这条规则到底管当前项目、你个人，还是只管这一趟会话。

关键片段

Implement checkpointing 1

"Implement checkpointing"这一段里最要紧的原始写法在下面，先看它怎么落地。

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


async def main():
    # Step 1: Enable checkpointing
    options = ClaudeAgentOptions(
        enable_file_checkpointing=True,
        permission_mode="acceptEdits",  # Auto-accept file edits without prompting
        extra_args={
            "replay-user-messages": None
        },  # Required to receive checkpoint UUIDs in the response stream
    )

    checkpoint_id = None
    session_id = None

    # Run the query and capture checkpoint UUID and session ID
    async with ClaudeSDKClient(options) as client:
        await client.query("Refactor the authentication module")

        # Step 2: Capture checkpoint UUID from the first user message
        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: Later, rewind by resuming the session with an empty prompt
    if checkpoint_id and session_id:
        async with ClaudeSDKClient(
            ClaudeAgentOptions(enable_file_checkpointing=True, resume=session_id)
        ) as client:
            await client.query("")  # Empty prompt to open the connection
            async for message in client.receive_response():
                await client.rewind_files(checkpoint_id)
                break
        print(f"Rewound to checkpoint: {checkpoint_id}")


asyncio.run(main())

关键片段

Implement checkpointing 2

"Implement checkpointing"这一段里最要紧的原始写法在下面，先看它怎么落地。

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

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

关键片段

Implement checkpointing 3

"Implement checkpointing"这一段里最要紧的原始写法在下面，先看它怎么落地。

checkpoint_id = None
session_id = None

async for message in client.receive_response():
    # Update checkpoint on each user message (keeps the latest)
    if isinstance(message, UserMessage) and message.uuid:
        checkpoint_id = message.uuid
    # Capture session ID from the result message
    if isinstance(message, ResultMessage):
        session_id = message.session_id

关键片段

Implement checkpointing 4

"Implement checkpointing"这一段里最要紧的原始写法在下面，先看它怎么落地。

async with ClaudeSDKClient(
    ClaudeAgentOptions(enable_file_checkpointing=True, resume=session_id)
) as client:
    await client.query("")  # Empty prompt to open the connection
    async for message in client.receive_response():
        await client.rewind_files(checkpoint_id)
        break

Common patterns

看到这里，就当是前人踩完坑留下来的顺手做法。

Checkpoint before risky operations

这一段不只是挂个标题，它是在说明"Checkpoint before risky operations"这一块到底负责什么。

关键片段

Checkpoint before risky operations

"Checkpoint before risky operations"这一段里最要紧的原始写法在下面，先看它怎么落地。

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():
            # Update checkpoint before each agent turn starts
            # This overwrites the previous checkpoint. Only keep the latest
            if isinstance(message, UserMessage) and message.uuid:
                safe_checkpoint = message.uuid

            # Decide when to revert based on your own logic
            # For example: error detection, validation failure, or user input
            if your_revert_condition and safe_checkpoint:
                await client.rewind_files(safe_checkpoint)
                # Exit the loop after rewinding, files are restored
                break


asyncio.run(main())

Multiple restore points

这一段更像在讲判断条件，什么时候该上，什么时候先别急。把触发条件看清，比背标题更重要。

这里还牵扯作用域，意思就是这条规则到底管当前项目、你个人，还是只管这一趟会话。

关键片段

Multiple restore points

"Multiple restore points"这一段里最要紧的原始写法在下面，先看它怎么落地。

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


# Store checkpoint metadata for better tracking
@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

    # Later: rewind to any checkpoint by resuming the session
    if checkpoints and session_id:
        target = checkpoints[0]  # Pick any checkpoint
        async with ClaudeSDKClient(
            ClaudeAgentOptions(enable_file_checkpointing=True, resume=session_id)
        ) as client:
            await client.query("")  # Empty prompt to open the connection
            async for message in client.receive_response():
                await client.rewind_files(target.id)
                break
        print(f"Rewound to: {target.description}")


asyncio.run(main())

Try it out

这段看着像个标题，其实是在说"Try it out"管到哪儿。

这里还牵扯作用域，意思就是这条规则到底管当前项目、你个人，还是只管这一趟会话。

关键片段

Try it out 1

这一段要真抓重点，通常就抓下面这块原文。

def add(a, b):
    return a + b


def subtract(a, b):
    return a - b


def multiply(a, b):
    return a * b


def divide(a, b):
    if b == 0:
        raise ValueError("Cannot divide by zero")
    return a / b

关键片段

Try it out 2

这一段要真抓重点，通常就抓下面这块原文。

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


async def main():
    # Configure the SDK with checkpointing enabled
    # - enable_file_checkpointing: Track file changes for rewinding
    # - permission_mode: Auto-accept file edits without prompting
    # - extra_args: Required to receive user message UUIDs in the stream
    options = ClaudeAgentOptions(
        enable_file_checkpointing=True,
        permission_mode="acceptEdits",
        extra_args={"replay-user-messages": None},
    )

    checkpoint_id = None  # Store the user message UUID for rewinding
    session_id = None  # Store the session ID for resuming

    print("Running agent to add doc comments to utils.py...\n")

    # Run the agent and capture checkpoint data from the response stream
    async with ClaudeSDKClient(options) as client:
        await client.query("Add doc comments to utils.py")

        async for message in client.receive_response():
            # Capture the first user message UUID - this is our restore point
            if isinstance(message, UserMessage) and message.uuid and not checkpoint_id:
                checkpoint_id = message.uuid
            # Capture the session ID so we can resume later
            if isinstance(message, ResultMessage):
                session_id = message.session_id

    print("Done! Open utils.py to see the added doc comments.\n")

    # Ask the user if they want to rewind the changes
    if checkpoint_id and session_id:
        response = input("Rewind to remove the doc comments? (y/n): ")

        if response.lower() == "y":
            # Resume the session with an empty prompt, then rewind
            async with ClaudeSDKClient(
                ClaudeAgentOptions(enable_file_checkpointing=True, resume=session_id)
            ) as client:
                await client.query("")  # Empty prompt opens the connection
                async for message in client.receive_response():
                    await client.rewind_files(checkpoint_id)  # Restore files
                    break

            print(
                "\n✓ File restored! Open utils.py to verify the doc comments are gone."
            )
        else:
            print("\nKept the modified file.")


asyncio.run(main())

终端里敲

Try it out 3

先别急着往下翻，下面这条命令跑完，心里才有底。

npx tsx try_checkpointing.ts

Limitations

这一块主要是在说"Limitations"真到手上该怎么用，哪里最容易踩坑。

Troubleshooting

这里讲的是怎么找毛病，先查明白哪一步出错，再决定怎么修。

Checkpointing options not recognized

这一段更像在讲判断条件，什么时候该上，什么时候先别急。把触发条件看清，比背标题更重要。

User messages don’t have UUIDs

这一段更像在讲判断条件，什么时候该上，什么时候先别急。把触发条件看清，比背标题更重要。

这里还牵扯作用域，意思就是这条规则到底管当前项目、你个人，还是只管这一趟会话。

”No file checkpoint found for message” error

这一段不只是挂个标题，它是在说明"”No file checkpoint found for message” error"这一块到底负责什么。

这里还牵扯作用域，意思就是这条规则到底管当前项目、你个人，还是只管这一趟会话。

”ProcessTransport is not ready for writing” error

这一段不只是挂个标题，它是在说明"”ProcessTransport is not ready for writing” error"这一块到底负责什么。

这里还牵扯作用域，意思就是这条规则到底管当前项目、你个人，还是只管这一趟会话。

关键片段

”ProcessTransport is not ready for writing” error

"”ProcessTransport is not ready for writing” error"这一段里最要紧的原始写法在下面，先看它怎么落地。

# Resume session with empty prompt, then rewind
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

Next steps

这里不是让你背"Next steps"这个词，而是让你看它真干活时怎么使。

这里还牵扯作用域，意思就是这条规则到底管当前项目、你个人，还是只管这一趟会话。

照着做一遍

如果你不想来回翻，就先照这几步顺着做。

每做完一步就看一下结果，再决定要不要继续往下。

关键片段

第 1 步：Implement checkpointing 1

"Implement checkpointing"这一段里最要紧的原始写法在下面，先看它怎么落地。

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


async def main():
    # Step 1: Enable checkpointing
    options = ClaudeAgentOptions(
        enable_file_checkpointing=True,
        permission_mode="acceptEdits",  # Auto-accept file edits without prompting
        extra_args={
            "replay-user-messages": None
        },  # Required to receive checkpoint UUIDs in the response stream
    )

    checkpoint_id = None
    session_id = None

    # Run the query and capture checkpoint UUID and session ID
    async with ClaudeSDKClient(options) as client:
        await client.query("Refactor the authentication module")

        # Step 2: Capture checkpoint UUID from the first user message
        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: Later, rewind by resuming the session with an empty prompt
    if checkpoint_id and session_id:
        async with ClaudeSDKClient(
            ClaudeAgentOptions(enable_file_checkpointing=True, resume=session_id)
        ) as client:
            await client.query("")  # Empty prompt to open the connection
            async for message in client.receive_response():
                await client.rewind_files(checkpoint_id)
                break
        print(f"Rewound to checkpoint: {checkpoint_id}")


asyncio.run(main())

关键片段

第 2 步：Implement checkpointing 2

"Implement checkpointing"这一段里最要紧的原始写法在下面，先看它怎么落地。

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

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

关键片段

第 3 步：Implement checkpointing 3

"Implement checkpointing"这一段里最要紧的原始写法在下面，先看它怎么落地。

checkpoint_id = None
session_id = None

async for message in client.receive_response():
    # Update checkpoint on each user message (keeps the latest)
    if isinstance(message, UserMessage) and message.uuid:
        checkpoint_id = message.uuid
    # Capture session ID from the result message
    if isinstance(message, ResultMessage):
        session_id = message.session_id

关键片段

第 4 步：Checkpoint before risky operations

"Checkpoint before risky operations"这一段里最要紧的原始写法在下面，先看它怎么落地。

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():
            # Update checkpoint before each agent turn starts
            # This overwrites the previous checkpoint. Only keep the latest
            if isinstance(message, UserMessage) and message.uuid:
                safe_checkpoint = message.uuid

            # Decide when to revert based on your own logic
            # For example: error detection, validation failure, or user input
            if your_revert_condition and safe_checkpoint:
                await client.rewind_files(safe_checkpoint)
                # Exit the loop after rewinding, files are restored
                break


asyncio.run(main())

一眼看懂这一页

先把这页到底在讲什么看明白，再去碰具体命令和配置，最不容易绕晕。

Rewind file changes with checkpointing
   |
   v
这是 Control and observability 里的一摊要紧活
   |
   v
先弄懂，再下手

文末提醒

这站会按官方 docs 的导航和内容变化继续重生成，原站加页、删页、改页时，这里会跟着更新。

人话解释会尽量顺着原页往下讲，但命令、参数名、配置名这些硬东西还是保留原样，免得你抄过去跑不起来。