1
Curious builder0 XP earned · 300 to level 2
0 daysFinish a lesson to begin
Badge collection0 of 6 unlocked
46 small wins to finish your pathNext lesson →
File checkpointing and rewinding files
File checkpointing copies a file before the agent changes it, so you can put it back. Each user message is a restore point, named by that message's uuid.
Last updated: 20 Oct, 2026 · Claude Agent SDK
Resuming brings a conversation back and leaves the files alone. This is the other half: the file checkpointing page says a rewind restores files on disk and does not rewind the conversation, which stays exactly as it was.
Two options, not one
enable_file_checkpointing=True turns the copying on. On its own it is not enough, because the ids you rewind to ride on user messages that the stream does not replay by default.
options = ClaudeAgentOptions(
enable_file_checkpointing=True,
# without this the user messages carry no uuid, so there is no id to use
extra_args={"replay-user-messages": None},
)enable_file_checkpointing builds no command line flag at all. The installed SDK sets the environment variable CLAUDE_CODE_ENABLE_SDK_FILE_CHECKPOINTING on the Claude Code process instead, which is also why the file checkpointing page tells you to set that variable by hand when rewinding from the bare command line.
A tool that writes a file
The shop's own tools only read, so desk_notes.py adds one that writes. Its file_path argument is what the desk checkpoints, the same argument Write and Edit carry.
from pathlib import Path
from claude_agent_sdk import create_sdk_mcp_server, tool
# file_path is the argument the desk checkpoints, as Write and Edit do
@tool("note_order", "Write a note about an order to a file.",
{"order_id": str, "file_path": str})
async def note_order(args):
path = Path(args["file_path"])
path.write_text(f"Order {args['order_id']} needs a call back.\n")
return {"content": [{"type": "text", "text": f"Wrote {path}."}]}
notes = create_sdk_mcp_server("notes", tools=[note_order])Teaching the desk to copy and restore
shop_desk.py adds a backups dictionary to __init__, a copy step at the top of run, and two new methods, write and rewind. The file also imports json now.
def __init__(self, options=None, model=None):
super().__init__(options)
self.model = model or ShopModel()
self.said, self.marks, self.tools = [], [], {}
# one snapshot per user message, which is what a checkpoint id names
self.backups = {} # Claude Code copies a file before a Write or an Edit changes it
if self.options.enable_file_checkpointing and "file_path" in args:
saved = self.backups.setdefault(self.marks[-1], {})
path = Path(args["file_path"])
saved.setdefault(str(path), path.read_text() if path.exists() else None) async def write(self, data):
# rewind_files is the one control request the desk answers with real work
request = json.loads(data).get("request") or {}
if request.get("subtype") == "rewind_files":
await self.rewind(request["user_message_id"])
await super().write(data) async def rewind(self, mark):
if not self.options.enable_file_checkpointing:
raise ProcessError("File rewinding is not enabled", exit_code=1)
# a path with no saved text did not exist at the checkpoint, so it goes
for path, text in self.backups.get(mark, {}).items():
Path(path).unlink() if text is None else Path(path).write_text(text)client.rewind_files(uuid) sends a control request, the same channel the permission questions travel on. Watching for it in write is how the desk turns that request into real work on real files.
Project files used on this pageThis lesson builds on a project from earlier lessons. The code below imports this file. Click a file to see its code, or follow the link to the lesson that wrote it. To run the code yourself, keep it in the same folder.
View the code here
shop_desk.py
import asyncio
import itertools
import json
import os
import re
from pathlib import Path
from claude_agent_sdk import (ClaudeAgentOptions, ProcessError, Transport,
project_key_for_directory)
ORDER = re.compile(r"\b[A-Z]\d+\b")
SESSIONS = itertools.count(1)
# Claude Code keeps one .jsonl file per session; the desk keeps one dictionary
SAVED = {}
class ShopModel:
"""Stands in for Claude: reads the conversation, picks the next step."""
def __init__(self, script=None):
self.script = list(script or [])
def reply(self, said, results, tools):
# a scripted reply lets a lesson force a step the stand-in would not pick
if self.script:
return self.script.pop(0)
if results:
return {"text": " ".join(results)}
# the newest message first, then the last id named earlier in the chat
orders = ORDER.findall(said[-1]) or ORDER.findall(" ".join(said))[-1:]
wanted = "refund_order" if "refund" in said[-1].lower() else "lookup_order"
tool = next((name for name in tools if name.endswith(wanted)), None)
if orders and tool:
return {"calls": [(tool, {"order_id": order}) for order in orders]}
return {"text": f"I cannot help with order {orders[0]} yet." if orders
else "Which order is this about?"}
class ShopDesk(Transport):
"""Stands in for the Claude Code binary: it runs the loop over JSON lines."""
def __init__(self, options=None, model=None):
self.options = options or ClaudeAgentOptions()
self.model = model or ShopModel()
# a counted id, not a UUID, so the printed output is the same every run
self.session_id = f"session-{next(SESSIONS)}"
self.inbox, self.outbox = asyncio.Queue(), asyncio.Queue()
self.prompts, self.said, self.marks = asyncio.Queue(), [], []
# one snapshot per user message, which is what a checkpoint id names
self.backups = {}
self.ids, self.setup, self.waiting = itertools.count(1), {}, {}
self.tools, self.failed, self.ready = {}, None, False
async def connect(self):
# a refused resume has to fail where the SDK starts the session
self.restore()
self.ready = True
def restore(self):
o = self.options
source = o.resume or (self.newest() if o.continue_conversation else None)
if source is None or source not in SAVED:
return
marks, said = SAVED[source]
if o.resume_session_at:
marks, said = self.truncate(marks, said)
# the loaded turns are the context the next question is answered against
self.marks, self.said = list(marks), list(said)
# a fork keeps the copied history under its own new id, so the original
# session is never appended to again
self.session_id = self.session_id if o.fork_session else source
def newest(self):
# continue_conversation takes the most recent session and needs no id
return list(SAVED)[-1] if SAVED else None
def truncate(self, marks, said):
# resume_session_at loads up to and including that message, and no further
end = marks.index(self.options.resume_session_at) + 1
drops = self.options.resume_drops_turn
# resume_drops_turn names the one turn the cut is allowed to throw away
if drops is not None and marks[end:end + 1] != [drops]:
raise ProcessError("Resume rejected by --resume-drops-turn: "
"the dropped range is not that turn", exit_code=1)
return marks[:end], said[:end]
def is_ready(self):
return self.ready
async def write(self, data):
# one JSON line at a time, which is what the SDK writes to a real process
message = json.loads(data)
# rewind_files is the one request the desk answers with real work
request = message.get("request") or {}
if request.get("subtype") == "rewind_files":
await self.rewind(request["user_message_id"])
await self.inbox.put(message)
async def rewind(self, mark):
if not self.options.enable_file_checkpointing:
raise ProcessError("File rewinding is not enabled", exit_code=1)
# a path with no saved text did not exist at the checkpoint, so it goes
for path, text in self.backups.get(mark, {}).items():
Path(path).unlink() if text is None else Path(path).write_text(text)
async def end_input(self):
# for a real process this would close its standard input
await self.inbox.put(None)
async def close(self):
self.ready = False
async def send(self, message):
await self.outbox.put(message)
async def read_messages(self):
# the listener and the worker run alongside the SDK's own read loop
asyncio.create_task(self.listen())
self.worker = asyncio.create_task(self.work())
while (message := await self.outbox.get()) is not None:
if isinstance(message, Exception):
raise message
yield message
async def listen(self):
while (message := await self.inbox.get()) is not None:
# a control request is the SDK asking a question; initialize is first
if message["type"] == "control_request":
if message["request"]["subtype"] == "initialize":
self.setup = message["request"]
await self.respond(message["request_id"])
# a control response answers a question the desk itself asked
elif message["type"] == "control_response":
answer = message["response"]
self.waiting.pop(answer["request_id"]).set_result(answer)
elif message["type"] == "user":
await self.prompts.put(message["message"]["content"])
# the input has ended: finish the work in hand, then pass on any failure
await self.prompts.put(None)
await self.worker
await self.send(self.failed)
async def respond(self, request_id):
# the stand-in says yes to every request the SDK makes of it
await self.send({"type": "control_response", "response": {
"subtype": "success", "request_id": request_id, "response": {}}})
async def ask(self, request):
# a future per request, so the answer can be awaited where it was asked
request_id = f"desk_{next(self.ids)}"
self.waiting[request_id] = asyncio.get_running_loop().create_future()
await self.send({"type": "control_request", "request_id": request_id,
"request": request})
return await self.waiting[request_id]
async def work(self):
while (prompt := await self.prompts.get()) is not None:
await self.turn(prompt)
async def start(self, prompt):
self.said.append(prompt)
self.marks.append(f"uuid_{len(self.marks) + 1}")
# replay-user-messages asks for the prompt back with the uuid it was
# stored under, which is the only way to learn that id
if "replay-user-messages" in self.options.extra_args:
await self.send({"type": "user", "uuid": self.marks[-1],
"parent_tool_use_id": None, "session_id": self.session_id,
"message": {"role": "user", "content": prompt}})
if not self.tools:
self.tools = await self.find_tools()
await self.send({"type": "system", "subtype": "init", "model": "shop-desk",
"session_id": self.session_id, "tools": list(self.tools)})
async def turn(self, prompt):
# UserPromptSubmit fires before the model is given the prompt
await self.hooks("UserPromptSubmit", {"prompt": prompt})
await self.start(prompt)
results, turns, limit = [], 0, self.options.max_turns
while "calls" in (step := self.model.reply(self.said, results, self.tools)):
turns += 1
# max_turns is the desk's own stop, not something the model respects
if limit and turns > limit:
message = f"Reached maximum number of turns ({limit})"
return await self.finish("error_max_turns", turns, message)
results = [await self.use(name, args) for name, args in step["calls"]]
await self.say(step["text"])
# Stop fires once the model has answered and wants no more tools
await self.hooks("Stop", {"stop_hook_active": False})
await self.finish("success", turns + 1, step["text"])
async def say(self, text):
# with include_partial_messages on, the text goes out a word at a time
if self.options.include_partial_messages:
for word in text.split(" "):
delta = {"type": "text_delta", "text": word + " "}
await self.send({"type": "stream_event", "uuid": f"ev_{next(self.ids)}",
"session_id": self.session_id, "event": {
"type": "content_block_delta", "index": 0,
"delta": delta}})
# the whole message still follows, so code written earlier keeps working
await self.send(self.message("assistant", [{"type": "text", "text": text}]))
async def finish(self, subtype, turns, text):
ok = subtype == "success"
result = {"type": "result", "subtype": subtype, "is_error": not ok,
"duration_ms": 0, "duration_api_ms": 0, "num_turns": turns,
"session_id": self.session_id, "total_cost_usd": 0.0,
"usage": {"input_tokens": len(" ".join(self.said).split()),
"output_tokens": len(text.split())}}
# cost is zero and usage counts words, because no model was called
result.update({"result": text} if ok else {"errors": [text]})
# the turn is on the shelf now, which is what a resume will read back
SAVED[self.session_id] = (list(self.marks), list(self.said))
# the frame goes before the result, because the result is what makes the
# SDK flush the batch to your store
await self.mirror()
await self.send(result)
self.failed = ProcessError("Command failed", exit_code=1) if not ok else None
def message(self, kind, content, parent=None):
body = {"role": kind, "content": content}
if kind == "assistant":
body.update(model="shop-desk", id=f"msg_{next(self.ids)}")
return {"type": kind, "message": body, "parent_tool_use_id": parent,
"session_id": self.session_id}
async def mirror(self):
# one frame per turn, holding the lines Claude Code wrote to the file
if self.options.session_store is None:
return
entries = [{"type": "user", "uuid": mark, "sessionId": self.session_id,
"message": {"role": "user", "content": text}}
for mark, text in zip(self.marks, self.said)]
await self.send({"type": "transcript_mirror", "entries": entries,
"filePath": self.transcript()})
def transcript(self):
# the SDK reads the store key off this path, so it has to look like the
# file Claude Code would have written
config = (self.options.env.get("CLAUDE_CONFIG_DIR")
or os.environ.get("CLAUDE_CONFIG_DIR") or f"{Path.home()}/.claude")
project = project_key_for_directory(self.options.cwd)
return f"{config}/projects/{project}/{self.session_id}.jsonl"
async def mcp(self, server, method, params):
message = {"jsonrpc": "2.0", "id": next(self.ids), "method": method,
"params": params}
answer = await self.ask({"subtype": "mcp_message", "server_name": server,
"message": message})
return answer["response"]["mcp_response"]["result"]
async def find_tools(self):
tools = {}
for server in self.options.mcp_servers:
await self.mcp(server, "initialize", {"protocolVersion": "2025-06-18",
"capabilities": {}, "clientInfo": {"name": "desk", "version": "1"}})
for tool in (await self.mcp(server, "tools/list", {}))["tools"]:
name = f"mcp__{server}__{tool['name']}"
# a tool left out here never reaches the model at all
if name not in self.options.disallowed_tools:
tools[name] = (server, tool["name"])
return tools
async def use(self, name, args):
call = f"toolu_{next(self.ids)}"
block = {"type": "tool_use", "id": call, "name": name, "input": args}
await self.send(self.message("assistant", [block]))
allowed, args = await self.permit(name, args, call)
text, failed = (await self.run(name, args, call)) if allowed else (args, True)
# a denied or raising call gets its own event, so a failure is never silent
if failed:
await self.hooks("PostToolUseFailure", {"tool_name": name, "error": text,
"tool_input": args, "tool_use_id": call})
result = {"type": "tool_result", "tool_use_id": call, "content": text,
"is_error": failed}
await self.send(self.message("user", [result]))
return text
async def run(self, name, args, call):
# the full name carries the server it came from, so the call goes back there
server, tool = self.tools[name]
# Claude Code copies a file before a Write or an Edit changes it
if self.options.enable_file_checkpointing and "file_path" in args:
saved = self.backups.setdefault(self.marks[-1], {})
path = Path(args["file_path"])
saved.setdefault(str(path), path.read_text() if path.exists() else None)
result = await self.mcp(server, "tools/call", {"name": tool, "arguments": args})
# the hooks see the answer before the model does, with the same call id
text = result["content"][0]["text"]
for output in await self.hooks("PostToolUse", {"tool_name": name, "tool_input": args,
"tool_response": text, "tool_use_id": call}):
# updatedToolOutput replaces what the model is about to read
decision = output.get("hookSpecificOutput", {})
text = decision.get("updatedToolOutput", text)
return text, result.get("isError", False)
def hook_input(self, event, data):
# the fields a hook input carries whatever the event is, plus the
# event's own fields on top
return {"hook_event_name": event, "session_id": self.session_id,
"cwd": ".", "transcript_path": "", **data}
async def hooks(self, event, data):
outputs = []
for group in (self.setup.get("hooks") or {}).get(event, []):
# a matcher of None means the group runs for every tool
# an event with no tool name tests the matcher against an empty string
if group["matcher"] and not re.fullmatch(group["matcher"],
data.get("tool_name", "")):
continue
# the SDK sent ids, not functions, so the call goes back by id
for callback in group["hookCallbackIds"]:
await self.lifecycle(event, "hook_started")
answer = await self.ask({"subtype": "hook_callback",
"callback_id": callback,
"tool_use_id": data.get("tool_use_id"),
"input": self.hook_input(event, data)})
outputs.append(answer["response"])
await self.lifecycle(event, "hook_response")
return outputs
async def lifecycle(self, event, phase):
# include_hook_events asks for these two frames around every callback
if self.options.include_hook_events:
await self.send({"type": "system", "subtype": phase,
"hook_event": event, "session_id": self.session_id,
"uuid": f"ev_{next(self.ids)}"})
async def pre_hooks(self, name, args, call):
verdict = None
for output in await self.hooks("PreToolUse", {"tool_name": name,
"tool_input": args, "tool_use_id": call}):
decision = output.get("hookSpecificOutput", {})
# one deny ends it, and the other hooks' opinions no longer matter
if decision.get("permissionDecision") == "deny":
return "deny", decision.get("permissionDecisionReason", "Blocked by a hook")
verdict = decision.get("permissionDecision", verdict)
args = decision.get("updatedInput", args)
return verdict, args
async def permit(self, name, args, call):
# hooks are the first step, so nothing below here can skip them
verdict, args = await self.pre_hooks(name, args, call)
if verdict in ("allow", "deny"):
return verdict == "allow", args
o = self.options
if o.permission_mode == "bypassPermissions" or name in o.allowed_tools:
return True, args
# nothing approved the call and there is nobody left to ask
if o.can_use_tool is None or o.permission_mode == "dontAsk":
return False, f"Permission to use {name} was not granted."
return await self.ask_permission(name, args, call)
async def ask_permission(self, name, args, call):
# the SDK runs your callback and writes its answer back on this channel
decision = (await self.ask({"subtype": "can_use_tool", "tool_name": name,
"input": args, "tool_use_id": call,
"permission_suggestions": None,
"blocked_path": None}))["response"]
# an allow may hand back different arguments to run with
if decision["behavior"] == "allow":
return True, decision["updatedInput"]
return False, decision["message"]
Restoring a file the agent changed
One helper runs a turn that writes the file and then rewinds to the turn's own uuid, so the three runs below differ only in their options.
from pathlib import Path
from claude_agent_sdk import (ClaudeAgentOptions, ClaudeSDKClient, ResultMessage,
UserMessage)
from desk_notes import notes
from shop_desk import ShopDesk, ShopModel
# the one tool call the scripted model will ask for
NOTE = Path("orders.txt")
CALL = {"calls": [("mcp__notes__note_order",
{"order_id": "A17", "file_path": "orders.txt"})]}
OPTIONS = dict(mcp_servers={"notes": notes}, allowed_tools=["mcp__notes__note_order"],
extra_args={"replay-user-messages": None})async def write_then_rewind(**extra):
# one turn that writes the file, then a rewind to that turn's own uuid
options = ClaudeAgentOptions(**OPTIONS, **extra)
desk, mark = ShopDesk(options, ShopModel(script=[CALL])), None
async with ClaudeSDKClient(options, transport=desk) as client:
await client.query("Note order A17 in orders.txt")
async for message in client.receive_response():
if isinstance(message, UserMessage) and message.uuid:
mark = message.uuid
# the rewind has to go out while the connection is still open
if isinstance(message, ResultMessage):
print("after :", NOTE.read_text().strip(), "| checkpoint", mark)
await client.rewind_files(mark)NOTE.write_text("nothing yet\n")
print("before :", NOTE.read_text().strip())
await write_then_rewind(enable_file_checkpointing=True)
print("rewound :", NOTE.read_text().strip())Output
before : nothing yet after : Order A17 needs a call back. | checkpoint uuid_1 rewound : nothing yet
- The file changed. The tool wrote its note, and the line printed inside the run shows it on disk.
- The checkpoint id is the user message's uuid, which only arrived because
replay-user-messageswas set. - The file came back. Not a fresh file with the same words: the text the desk copied before the tool ran.
Restoring a file that did not exist
A rewind undoes a creation too, because the backup for a path that was not there is nothing, and putting nothing back means deleting it.
NOTE.unlink(missing_ok=True)
print("before :", NOTE.exists())
await write_then_rewind(enable_file_checkpointing=True)
print("rewound :", NOTE.exists())Output
before : False after : Order A17 needs a call back. | checkpoint uuid_1 rewound : False
That is the behaviour the file checkpointing page describes: a rewind deletes the files the agent created and restores the ones it modified.
The same run with checkpointing off
Take the option away and the rewind has nothing to work with.
NOTE.write_text("nothing yet\n")
await write_then_rewind()Output
after : Order A17 needs a call back. | checkpoint uuid_1
Traceback (most recent call last):
File "main.py", line 2, in <module>
await write_then_rewind()
claude_agent_sdk._errors.ProcessError: File rewinding is not enabled (exit code: 1)The real Claude Code answers the same request with File rewinding is not enabled, and the file checkpointing page lists two ways to earn it: the option was off on the session that made the changes, or it is off on the session doing the rewinding. Both have to have it.
What a checkpoint does not cover
- Only Write, Edit and NotebookEdit. A file a Bash command changed was never copied, so a rewind leaves it as it is.
- Only this session. A checkpoint id belongs to the session that created it, and both that session and the one rewinding need the option.
- Only file contents. A directory the agent created, moved or deleted stays as it is.
- Not a nested agent's edits. Those are untracked, except a skill running in the foreground, so git is what you reach for there.
Checkpointing compared with forking
| fork_session | enable_file_checkpointing | |
|---|---|---|
| Branches | the conversation | nothing |
| Restores | nothing | files on disk |
| Granularity | a whole session | one user message |
| Works with a session store | yes | no, the SDK refuses the pair |
| Undoes a Bash change | no | no |
When to turn it on
- Before a risky instruction. Keep the latest uuid as you read the stream, and rewind the moment a check fails.
- Several restore points. Collect every user message uuid in a list and rewind to the one you want, keeping turn one's work and dropping turn two's.
- A directory that is not a git repository. Where there is no commit to go back to, this is the undo.
Watch out. Calling
rewind_files after the message loop has finished fails with ProcessTransport is not ready for writing against the real Claude Code: the file checkpointing page explains that the connection closes when the loop ends. Rewind inside the loop, as the runs above do, or resume the session with an empty prompt and rewind from the new connection.Related
- Previous: Session storage with a SessionStore
- Next: System prompts
- Reference: Rewind file changes with checkpointing
Try it yourself
- Collect every user message uuid in a list across two turns, then rewind to the first.
- Point the tool at a path inside a folder that does not exist and read what happens.
- Drop
extra_argsfrom the options and print the checkpoint id. What do you get?
This is what real progress feels like.