Revertir cambios de archivos con checkpointing
Rastrear cambios de archivos durante sesiones de agente y restaurar archivos a cualquier estado anterior
El checkpointing de archivos rastrea las modificaciones de archivos realizadas a través de las herramientas Write, Edit y NotebookEdit durante una sesión de agente, lo que le permite revertir archivos a cualquier estado anterior. ¿Desea probarlo? Salte al ejemplo interactivo.
Con checkpointing, puede:
- Deshacer cambios no deseados restaurando archivos a un estado conocido y bueno
- Explorar alternativas restaurando a un checkpoint e intentando un enfoque diferente
- Recuperarse de errores cuando el agente realiza modificaciones incorrectas
Solo se rastrean los cambios realizados a través de las herramientas Write, Edit y NotebookEdit. Los cambios realizados a través de comandos Bash (como echo > file.txt o sed -i) no se capturan en el sistema de checkpoint.
Cómo funciona el checkpointing
Cuando habilita el checkpointing de archivos, el SDK crea copias de seguridad de archivos antes de modificarlos a través de las herramientas Write, Edit o NotebookEdit. Los mensajes de usuario en el flujo de respuesta incluyen un UUID de checkpoint que puede usar como punto de restauración.
Checkpoint funciona con estas herramientas integradas que el agente usa para modificar archivos:
| Herramienta | Descripción |
|---|---|
| Write | Crea un archivo nuevo o sobrescribe un archivo existente con contenido nuevo |
| Edit | Realiza ediciones dirigidas a partes específicas de un archivo existente |
| NotebookEdit | Modifica celdas en cuadernos Jupyter (archivos .ipynb) |
La reversión de archivos restaura archivos en disco a un estado anterior. No revierte la conversación en sí. El historial de conversación y el contexto permanecen intactos después de llamar a rewindFiles() (TypeScript) o rewind_files() (Python).
El sistema de checkpoint rastrea:
- Archivos creados durante la sesión
- Archivos modificados durante la sesión
- El contenido original de archivos modificados
Cuando revierte a un checkpoint, los archivos creados se eliminan y los archivos modificados se restauran a su contenido en ese punto.
Implementar checkpointing
Para usar el checkpointing de archivos, habilítelo en sus opciones, capture UUIDs de checkpoint del flujo de respuesta, luego llame a rewindFiles() (TypeScript) o rewind_files() (Python) cuando necesite restaurar.
El siguiente ejemplo muestra el flujo completo: habilitar checkpointing, capturar el UUID de checkpoint y el ID de sesión del flujo de respuesta, luego reanudar la sesión más tarde para revertir archivos. Cada paso se explica en detalle a continuación.
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())
import { query } from "@anthropic-ai/claude-agent-sdk";
async function main() {
// Step 1: Enable checkpointing
const opts = {
enableFileCheckpointing: true,
permissionMode: "acceptEdits" as const, // Auto-accept file edits without prompting
extraArgs: { "replay-user-messages": null } // Required to receive checkpoint UUIDs in the response stream
};
const response = query({
prompt: "Refactor the authentication module",
options: opts
});
let checkpointId: string | undefined;
let sessionId: string | undefined;
// Step 2: Capture checkpoint UUID from the first user message
for await (const message of response) {
if (message.type === "user" && message.uuid && !checkpointId) {
checkpointId = message.uuid;
}
if ("session_id" in message && !sessionId) {
sessionId = message.session_id;
}
}
// Step 3: Later, rewind by resuming the session with an empty prompt
if (checkpointId && sessionId) {
const rewindQuery = query({
prompt: "", // Empty prompt to open the connection
options: { ...opts, resume: sessionId }
});
for await (const msg of rewindQuery) {
await rewindQuery.rewindFiles(checkpointId);
break;
}
console.log(`Rewound to checkpoint: ${checkpointId}`);
}
}
main();
Habilitar checkpointing
Configure sus opciones de SDK para habilitar checkpointing y recibir UUIDs de checkpoint:
| Opción | Python | TypeScript | Descripción |
|---|---|---|---|
| Habilitar checkpointing | enable_file_checkpointing=True |
enableFileCheckpointing: true |
Rastrea cambios de archivos para reversión |
| Recibir UUIDs de checkpoint | extra_args={"replay-user-messages": None} |
extraArgs: { 'replay-user-messages': null } |
Requerido para obtener UUIDs de mensajes de usuario en el flujo |
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")
const response = query({
prompt: "Refactor the authentication module",
options: {
enableFileCheckpointing: true,
permissionMode: "acceptEdits" as const,
extraArgs: { "replay-user-messages": null }
}
});
Capturar UUID de checkpoint e ID de sesión
Con la opción replay-user-messages establecida (mostrada arriba), cada mensaje de usuario en el flujo de respuesta tiene un UUID que sirve como checkpoint.
Para la mayoría de los casos de uso, capture el UUID del primer mensaje de usuario (message.uuid); revertir a él restaura todos los archivos a su estado original. Para almacenar múltiples checkpoints y revertir a estados intermedios, consulte Múltiples puntos de restauración.
Capturar el ID de sesión (message.session_id) es opcional; solo lo necesita si desea revertir más tarde, después de que se complete el flujo. Si está llamando a rewindFiles() inmediatamente mientras aún procesa mensajes (como lo hace el ejemplo en Checkpoint antes de operaciones arriesgadas), puede omitir la captura del ID de sesión.
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
let checkpointId: string | undefined;
let sessionId: string | undefined;
for await (const message of response) {
// Update checkpoint on each user message (keeps the latest)
if (message.type === "user" && message.uuid) {
checkpointId = message.uuid;
}
// Capture session ID from any message that has it
if ("session_id" in message) {
sessionId = message.session_id;
}
}
Revertir archivos
Para revertir después de que se complete el flujo, reanude la sesión con un mensaje vacío y llame a rewind_files() (Python) o rewindFiles() (TypeScript) con su UUID de checkpoint. También puede revertir durante el flujo; consulte Checkpoint antes de operaciones arriesgadas para ese patrón.
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
const rewindQuery = query({
prompt: "", // Empty prompt to open the connection
options: { ...opts, resume: sessionId }
});
for await (const msg of rewindQuery) {
await rewindQuery.rewindFiles(checkpointId);
break;
}
Si captura el ID de sesión y el ID de checkpoint, también puede revertir desde la CLI:
claude -p --resume <session-id> --rewind-files <checkpoint-uuid>
Patrones comunes
Estos patrones muestran diferentes formas de capturar y usar UUIDs de checkpoint según su caso de uso.
Checkpoint antes de operaciones arriesgadas
Este patrón mantiene solo el UUID de checkpoint más reciente, actualizándolo antes de cada turno del agente. Si algo sale mal durante el procesamiento, puede revertir inmediatamente al último estado seguro y salir del bucle.
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())
import { query } from "@anthropic-ai/claude-agent-sdk";
async function main() {
const response = query({
prompt: "Refactor the authentication module",
options: {
enableFileCheckpointing: true,
permissionMode: "acceptEdits" as const,
extraArgs: { "replay-user-messages": null }
}
});
let safeCheckpoint: string | undefined;
for await (const message of response) {
// Update checkpoint before each agent turn starts
// This overwrites the previous checkpoint. Only keep the latest
if (message.type === "user" && message.uuid) {
safeCheckpoint = message.uuid;
}
// Decide when to revert based on your own logic
// For example: error detection, validation failure, or user input
if (yourRevertCondition && safeCheckpoint) {
await response.rewindFiles(safeCheckpoint);
// Exit the loop after rewinding, files are restored
break;
}
}
}
main();
Múltiples puntos de restauración
Si Claude realiza cambios en múltiples turnos, es posible que desee revertir a un punto específico en lugar de volver completamente. Por ejemplo, si Claude refactoriza un archivo en el turno uno y agrega pruebas en el turno dos, es posible que desee mantener la refactorización pero deshacer las pruebas.
Este patrón almacena todos los UUIDs de checkpoint en una matriz con metadatos. Después de que se complete la sesión, puede revertir a cualquier checkpoint anterior:
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())
import { query } from "@anthropic-ai/claude-agent-sdk";
// Store checkpoint metadata for better tracking
interface Checkpoint {
id: string;
description: string;
timestamp: Date;
}
async function main() {
const opts = {
enableFileCheckpointing: true,
permissionMode: "acceptEdits" as const,
extraArgs: { "replay-user-messages": null }
};
const response = query({
prompt: "Refactor the authentication module",
options: opts
});
const checkpoints: Checkpoint[] = [];
let sessionId: string | undefined;
for await (const message of response) {
if (message.type === "user" && message.uuid) {
checkpoints.push({
id: message.uuid,
description: `After turn ${checkpoints.length + 1}`,
timestamp: new Date()
});
}
if ("session_id" in message && !sessionId) {
sessionId = message.session_id;
}
}
// Later: rewind to any checkpoint by resuming the session
if (checkpoints.length > 0 && sessionId) {
const target = checkpoints[0]; // Pick any checkpoint
const rewindQuery = query({
prompt: "", // Empty prompt to open the connection
options: { ...opts, resume: sessionId }
});
for await (const msg of rewindQuery) {
await rewindQuery.rewindFiles(target.id);
break;
}
console.log(`Rewound to: ${target.description}`);
}
}
main();
Pruébelo
Este ejemplo completo crea un pequeño archivo de utilidad, hace que el agente agregue comentarios de documentación, le muestra los cambios, luego pregunta si desea revertir.
Antes de comenzar, asegúrese de tener el Claude Agent SDK instalado.
Crear un archivo de prueba
Cree un nuevo archivo llamado utils.py (Python) o utils.ts (TypeScript) y pegue el siguiente código:
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
export function add(a: number, b: number): number {
return a + b;
}
export function subtract(a: number, b: number): number {
return a - b;
}
export function multiply(a: number, b: number): number {
return a * b;
}
export function divide(a: number, b: number): number {
if (b === 0) {
throw new Error("Cannot divide by zero");
}
return a / b;
}
Ejecutar el ejemplo interactivo
Cree un nuevo archivo llamado try_checkpointing.py (Python) o try_checkpointing.ts (TypeScript) en el mismo directorio que su archivo de utilidad, y pegue el siguiente código.
Este script le pide a Claude que agregue comentarios de documentación a su archivo de utilidad, luego le da la opción de revertir y restaurar el original.
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())
import { query } from "@anthropic-ai/claude-agent-sdk";
import * as readline from "readline";
async function main() {
// Configure the SDK with checkpointing enabled
// - enableFileCheckpointing: Track file changes for rewinding
// - permissionMode: Auto-accept file edits without prompting
// - extraArgs: Required to receive user message UUIDs in the stream
const opts = {
enableFileCheckpointing: true,
permissionMode: "acceptEdits" as const,
extraArgs: { "replay-user-messages": null }
};
let sessionId: string | undefined; // Store the session ID for resuming
let checkpointId: string | undefined; // Store the user message UUID for rewinding
console.log("Running agent to add doc comments to utils.ts...\n");
// Run the agent and capture checkpoint data from the response stream
const response = query({
prompt: "Add doc comments to utils.ts",
options: opts
});
for await (const message of response) {
// Capture the first user message UUID - this is our restore point
if (message.type === "user" && message.uuid && !checkpointId) {
checkpointId = message.uuid;
}
// Capture the session ID so we can resume later
if ("session_id" in message) {
sessionId = message.session_id;
}
}
console.log("Done! Open utils.ts to see the added doc comments.\n");
// Ask the user if they want to rewind the changes
if (checkpointId && sessionId) {
const rl = readline.createInterface({
input: process.stdin,
output: process.stdout
});
const answer = await new Promise<string>((resolve) => {
rl.question("Rewind to remove the doc comments? (y/n): ", resolve);
});
rl.close();
if (answer.toLowerCase() === "y") {
// Resume the session with an empty prompt, then rewind
const rewindQuery = query({
prompt: "", // Empty prompt opens the connection
options: { ...opts, resume: sessionId }
});
for await (const msg of rewindQuery) {
await rewindQuery.rewindFiles(checkpointId); // Restore files
break;
}
console.log("\n✓ File restored! Open utils.ts to verify the doc comments are gone.");
} else {
console.log("\nKept the modified file.");
}
}
}
main();
Este ejemplo demuestra el flujo de trabajo completo de checkpointing:
- Habilitar checkpointing: configure el SDK con
enable_file_checkpointing=Trueypermission_mode="acceptEdits"para aprobar automáticamente ediciones de archivos - Capturar datos de checkpoint: mientras el agente se ejecuta, almacene el UUID del primer mensaje de usuario (su punto de restauración) y el ID de sesión
- Solicitar reversión: después de que el agente termine, verifique su archivo de utilidad para ver los comentarios de documentación, luego decida si desea deshacer los cambios
- Reanudar y revertir: si es así, reanude la sesión con un mensaje vacío y llame a
rewind_files()para restaurar el archivo original
Ejecutar el ejemplo
Ejecute el script desde el mismo directorio que su archivo de utilidad.
Abra su archivo de utilidad (utils.py o utils.ts) en su IDE o editor antes de ejecutar el script. Verá que el archivo se actualiza en tiempo real mientras el agente agrega comentarios de documentación, luego revierte al original cuando elige revertir.
python try_checkpointing.py
npx tsx try_checkpointing.ts
Verá que el agente agrega comentarios de documentación, luego un mensaje preguntando si desea revertir. Si elige sí, el archivo se restaura a su estado original.
Limitaciones
El checkpointing de archivos tiene las siguientes limitaciones:
| Limitación | Descripción |
|---|---|
| Solo herramientas Write/Edit/NotebookEdit | Los cambios realizados a través de comandos Bash no se rastrean |
| Misma sesión | Los checkpoints están vinculados a la sesión que los creó |
| Solo contenido de archivo | Crear, mover o eliminar directorios no se deshace al revertir |
| Archivos locales | Los archivos remotos o de red no se rastrean |
Solución de problemas
Las opciones de checkpointing no se reconocen
Si enableFileCheckpointing o rewindFiles() no está disponible, es posible que esté en una versión anterior del SDK.
Solución: Actualice a la última versión del SDK:
- Python:
pip install --upgrade claude-agent-sdk - TypeScript:
npm install @anthropic-ai/claude-agent-sdk@latest
Los mensajes de usuario no tienen UUIDs
Si message.uuid es undefined o está faltando, no está recibiendo UUIDs de checkpoint.
Causa: La opción replay-user-messages no está establecida.
Solución: Agregue extra_args={"replay-user-messages": None} (Python) o extraArgs: { 'replay-user-messages': null } (TypeScript) a sus opciones.
Error "No file checkpoint found for message"
Este error ocurre cuando los datos de checkpoint no existen para el UUID de mensaje de usuario especificado.
Causas comunes:
- El checkpointing de archivos no estaba habilitado en la sesión original (
enable_file_checkpointingoenableFileCheckpointingno estaba establecido entrue) - La sesión no se completó correctamente antes de intentar reanudar y revertir
Solución: Asegúrese de que enable_file_checkpointing=True (Python) o enableFileCheckpointing: true (TypeScript) estuviera establecido en la sesión original, luego use el patrón mostrado en los ejemplos: capture el UUID del primer mensaje de usuario, complete la sesión completamente, luego reanude con un mensaje vacío y llame a rewindFiles() una sola vez.
Error "ProcessTransport is not ready for writing"
Este error ocurre cuando llama a rewindFiles() o rewind_files() después de haber terminado de iterar a través de la respuesta. La conexión al proceso de CLI se cierra cuando se completa el bucle.
Solución: Reanude la sesión con un mensaje vacío, luego llame a rewind en la nueva consulta:
# 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
// Resume session with empty prompt, then rewind
const rewindQuery = query({
prompt: "",
options: { ...opts, resume: sessionId }
});
for await (const msg of rewindQuery) {
await rewindQuery.rewindFiles(checkpointId);
break;
}
Próximos pasos
- Sessions: aprenda cómo reanudar sesiones, que es necesario para revertir después de que se complete el flujo. Cubre IDs de sesión, reanudación de conversaciones y bifurcación de sesiones.
- Permissions: configure qué herramientas puede usar Claude y cómo se aprueban las modificaciones de archivos. Útil si desea más control sobre cuándo ocurren las ediciones.
- TypeScript SDK reference: referencia completa de API incluyendo todas las opciones para
query()y el métodorewindFiles(). - Python SDK reference: referencia completa de API incluyendo todas las opciones para
ClaudeAgentOptionsy el métodorewind_files().