feat(checkpoints): shadow-git file snapshots before mutating tool calls#335
feat(checkpoints): shadow-git file snapshots before mutating tool calls#335DouweM wants to merge 5 commits into
Conversation
Coding agents can damage files with no built-in way to undo. Five surveyed harnesses converged on a shadow git repository that snapshots the working tree before each file-mutating tool call, independent of the user's own version control (broad survey, recommendation 5). `Checkpoints` snapshots the project before a tool whose name is in `mutating_tools` runs, into a repo at `<state_dir>/checkpoints/<project-slug>/` that has its own committer identity, gpg signing off, and no dependency on the user's `.git`. It works in non-git projects, respects `.gitignore`, and debounces so unchanged trees do not produce empty commits. Restore and list the snapshots from Python, or expose them to the model behind `expose_tool`. Files only for v1; conversation rewind/fork pairs with the branchable session-history track (#321). Co-Authored-By: Claude Opus <noreply@anthropic.com>
Some git versions render `%cI` with a `Z` suffix, which `datetime.fromisoformat` rejects before Python 3.11. Read `%ct` (Unix timestamp) instead, which parses identically across git and Python versions. Co-Authored-By: Claude Opus <noreply@anthropic.com>
|
@coderabbitai review |
…ed state_dir The shadow-git README claimed `GIT_CONFIG_GLOBAL`/`GIT_CONFIG_SYSTEM` are "unset", but the code points them at `os.devnull`; reword to match. Also warn that a `state_dir` nested under `project_root` would be swept into every snapshot by `git add -A`. Co-Authored-By: Claude Opus <noreply@anthropic.com>
Merge `main` and relocate from `experimental/checkpoints` to top-level `pydantic_ai_harness/checkpoints/`, matching #347/#354. Drops the experimental import warning, fixes import paths, moves tests. Adds the `docs/checkpoints.md` page (nav.json, index.md, parity gate) and a README matrix row. No behavior change. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
📝 WalkthroughWalkthroughAdds a Possibly related PRs
🚥 Pre-merge checks | ✅ 4✅ Passed checks (4 passed)
✨ Finishing Touches📝 Generate docstrings
🧪 Generate unit tests (beta)
Comment |
There was a problem hiding this comment.
Actionable comments posted: 4
🧹 Nitpick comments (3)
tests/checkpoints/test_checkpoints.py (3)
231-234: 📐 Maintainability & Code Quality | 🔵 Trivial | ⚡ Quick winSimplify the defensive
getattron message parts.
getattr(m, 'parts', [])treats.partsas possibly absent and effectively types asAny, but bothModelMessagevariants expose.partsdirectly. Iteratingm.partskeeps type narrowing intact per the**/*.pytyping guideline.🧹 Proposed fix
- if not any(isinstance(p, ToolCallPart) for m in messages for p in getattr(m, 'parts', [])): + if not any(isinstance(p, ToolCallPart) for m in messages for p in m.parts):🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@tests/checkpoints/test_checkpoints.py` around lines 231 - 234, Update the message iteration in model_fn to access the parts attribute directly via m.parts instead of using getattr with a fallback, while preserving the existing ToolCallPart detection and response behavior.Source: Coding guidelines
339-339: 📐 Maintainability & Code Quality | 🔵 Trivial | ⚡ Quick winUse a normal
datetimeimport instead of__import__.
__import__('datetime').datetime.now()returns an effectively untyped value, which conflicts with the "avoidAny" guideline for**/*.py, and is just harder to read than a top-level import.🧹 Proposed fix
+from datetime import datetime + def test_checkpoint_is_frozen_dataclass() -> None: - cp = Checkpoint(id='abc', time=__import__('datetime').datetime.now(), tool_name='write_file', files_changed=[]) + cp = Checkpoint(id='abc', time=datetime.now(), tool_name='write_file', files_changed=[])🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@tests/checkpoints/test_checkpoints.py` at line 339, Replace the dynamic __import__('datetime') usage in the Checkpoint construction with a normal top-level datetime import, then call datetime.datetime.now() through that import while leaving the checkpoint fields unchanged.Source: Coding guidelines
30-30: 📐 Maintainability & Code Quality | 🔵 Trivial | ⚡ Quick winExpose
shadow_dir_forthrough the public checkpoints package, or switch the test to a public API.pydantic_ai_harness.checkpoints.__init__does not re-export it, so this test is reaching into_shadowdirectly.🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@tests/checkpoints/test_checkpoints.py` at line 30, Update the test’s use of shadow_dir_for to avoid importing it from the private _shadow module: either re-export shadow_dir_for from the public checkpoints package __init__, or change the test import and usage to an existing public checkpoints API.Source: Coding guidelines
🤖 Prompt for all review comments with AI agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
Inline comments:
In `@pydantic_ai_harness/checkpoints/_shadow.py`:
- Around line 92-101: Protect the multi-step mutation flows in
CheckpointStore.snapshot() and CheckpointStore.restore() with a lock shared by
all callers using the same shadow_dir, covering the entire sequence of git
operations. Preserve the documented concurrent shared-repository behavior and
last-writer-wins semantics; do not use a lock scoped only to an individual call
or instance if instances can share the repository.
- Around line 152-158: Update the shadow-directory setup before git init to
create it with mode 0700 and verify any existing directory is owned by the
current user and has no group or other permissions; reject unsafe reuse with
CheckpointError. Anchor the change in the directory creation logic before
self._git('init', ...) while preserving the existing error handling.
- Around line 149-176: Update the shadow-directory initialization and validation
around ensure_initialized to reject any shadow_dir that overlaps project_root,
including when either path is an ancestor of the other. Resolve both paths
before comparing them, require the shadow repository to be a disjoint
harness-owned location, and do not treat an existing HEAD as sufficient
validation; preserve normal initialization for valid paths.
In `@pydantic_ai_harness/checkpoints/_toolset.py`:
- Around line 36-46: Normalize an empty paths list to None before calling
_store.restore in restore_checkpoint, so paths=[] restores all files
consistently with the existing “all files” confirmation message. Preserve
explicit non-empty path lists and the current response formatting.
---
Nitpick comments:
In `@tests/checkpoints/test_checkpoints.py`:
- Around line 231-234: Update the message iteration in model_fn to access the
parts attribute directly via m.parts instead of using getattr with a fallback,
while preserving the existing ToolCallPart detection and response behavior.
- Line 339: Replace the dynamic __import__('datetime') usage in the Checkpoint
construction with a normal top-level datetime import, then call
datetime.datetime.now() through that import while leaving the checkpoint fields
unchanged.
- Line 30: Update the test’s use of shadow_dir_for to avoid importing it from
the private _shadow module: either re-export shadow_dir_for from the public
checkpoints package __init__, or change the test import and usage to an existing
public checkpoints API.
🪄 Autofix (Beta)
Fix all unresolved CodeRabbit comments on this PR:
- Push a commit to this branch (recommended)
- Create a new PR with the fixes
ℹ️ Review info
⚙️ Run configuration
Configuration used: Organization UI
Review profile: CHILL
Plan: Pro Plus
Run ID: 45bc5313-6f58-46ff-8a7f-7650e001cbe7
📒 Files selected for processing (12)
README.mddocs/checkpoints.mddocs/index.mddocs/nav.jsonpydantic_ai_harness/checkpoints/README.mdpydantic_ai_harness/checkpoints/__init__.pypydantic_ai_harness/checkpoints/_capability.pypydantic_ai_harness/checkpoints/_shadow.pypydantic_ai_harness/checkpoints/_toolset.pytests/checkpoints/__init__.pytests/checkpoints/test_checkpoints.pytests/test_docs_parity.py
🔗 Linked repositories identified
CodeRabbit considers these linked repositories for cross-repo context during reviews:
pydantic/logfire(manual)pydantic/pydantic-ai(manual)pydantic/monty(auto-detected)pydantic/pydantic(auto-detected)
| @dataclass | ||
| class CheckpointStore: | ||
| """Snapshot and restore a project's files through a shadow git repository. | ||
|
|
||
| The store is stateless beyond its configuration -- every method reads the | ||
| current state from the shadow repo on disk -- so it is cheap to build one per | ||
| call and safe to use from several `Agent.run` calls against the same project. | ||
| Concurrent runs share the shadow repo with last-writer-wins semantics: each | ||
| snapshot commits the whole work tree as it looks at that instant. | ||
| """ |
There was a problem hiding this comment.
🗄️ Data Integrity & Integration | 🟠 Major | ⚡ Quick win
🧩 Analysis chain
🏁 Script executed:
#!/bin/bash
set -euo pipefail
echo '--- file list ---'
git ls-files pydantic_ai_harness/checkpoints/_shadow.py
echo '--- ast outline ---'
ast-grep outline pydantic_ai_harness/checkpoints/_shadow.py --view expanded || true
echo '--- relevant ranges ---'
nl -ba pydantic_ai_harness/checkpoints/_shadow.py | sed -n '1,260p'
echo '--- search for locks / concurrency / snapshot usage ---'
rg -n "FileLock|lock|snapshot\(|restore\(|_index_matches_head|_has_head|shadow_dir" pydantic_ai_harness -SRepository: pydantic/pydantic-ai-harness
Length of output: 1867
🏁 Script executed:
#!/bin/bash
set -euo pipefail
python3 - <<'PY'
from pathlib import Path
p = Path('pydantic_ai_harness/checkpoints/_shadow.py')
for start, end in [(120, 240)]:
print(f'--- {start}-{end} ---')
with p.open() as f:
for i, line in enumerate(f, start=1):
if start <= i <= end:
print(f'{i:4}: {line.rstrip()}')
PY
echo '--- filelock usage / dependency ---'
rg -n "filelock|FileLock" -S .
echo '--- pyproject dependency snippets ---'
sed -n '1,220p' pyproject.tomlRepository: pydantic/pydantic-ai-harness
Length of output: 6787
🏁 Script executed:
#!/bin/bash
set -euo pipefail
python3 - <<'PY'
from pathlib import Path
p = Path('pydantic_ai_harness/checkpoints/_shadow.py')
for start, end in [(145, 235)]:
print(f'--- {start}-{end} ---')
with p.open() as f:
for i, line in enumerate(f, start=1):
if start <= i <= end:
print(f'{i:4}: {line.rstrip()}')
PY
echo '--- filelock usage / dependency ---'
rg -n "filelock|FileLock" -S .
echo '--- pyproject dependency snippets ---'
python3 - <<'PY'
from pathlib import Path
p = Path('pyproject.toml')
text = p.read_text()
for key in ['filelock', 'dependencies', 'project']:
if key in text:
print(f'contains {key}')
print('--- first 220 lines ---')
for i, line in enumerate(text.splitlines(), start=1):
if i <= 220:
print(f'{i:4}: {line}')
PYRepository: pydantic/pydantic-ai-harness
Length of output: 5388
🏁 Script executed:
#!/bin/bash
set -euo pipefail
python3 - <<'PY'
from pathlib import Path
p = Path('pydantic_ai_harness/checkpoints/_shadow.py')
for start, end in [(145, 235)]:
print(f'--- {start}-{end} ---')
with p.open() as f:
for i, line in enumerate(f, start=1):
if start <= i <= end:
print(f'{i:4}: {line.rstrip()}')
PY
echo '--- filelock usage / dependency ---'
rg -n "filelock|FileLock" -S .
echo '--- pyproject dependency snippets ---'
python3 - <<'PY'
from pathlib import Path
text = Path('pyproject.toml').read_text()
for i, line in enumerate(text.splitlines(), start=1):
if i <= 220:
print(f'{i:4}: {line}')
PYRepository: pydantic/pydantic-ai-harness
Length of output: 5388
Guard shadow-repo mutations with a lock. snapshot() and restore() span multiple git calls, so concurrent callers sharing one shadow_dir can interleave and violate the stated last-writer-wins behavior. Add a lock around the whole mutation path, or narrow the docstring if shared concurrent use is not supported.
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@pydantic_ai_harness/checkpoints/_shadow.py` around lines 92 - 101, Protect
the multi-step mutation flows in CheckpointStore.snapshot() and
CheckpointStore.restore() with a lock shared by all callers using the same
shadow_dir, covering the entire sequence of git operations. Preserve the
documented concurrent shared-repository behavior and last-writer-wins semantics;
do not use a lock scoped only to an individual call or instance if instances can
share the repository.
| if (self.shadow_dir / 'HEAD').exists(): | ||
| return | ||
| try: | ||
| self.shadow_dir.mkdir(parents=True, exist_ok=True) | ||
| except OSError as exc: | ||
| raise CheckpointError(f'could not create shadow repo at {self.shadow_dir}: {exc}') from exc | ||
| self._git('init', '-q', str(self.shadow_dir)) | ||
| info_dir = self.shadow_dir / 'info' | ||
| info_dir.mkdir(exist_ok=True) | ||
| (info_dir / 'exclude').write_text('.git/\n', encoding='utf-8') | ||
|
|
||
| def _has_head(self) -> bool: | ||
| return self._git('rev-parse', '--verify', '--quiet', 'HEAD', check=False).returncode == 0 | ||
|
|
||
| def _index_matches_head(self) -> bool: | ||
| # rc 0 == no staged difference from HEAD, i.e. the work tree is unchanged | ||
| # since the last checkpoint, so a new snapshot would be an empty duplicate. | ||
| return self._git('diff', '--cached', '--quiet', check=False).returncode == 0 | ||
|
|
||
| def snapshot(self, *, tool_name: str | None = None, run_id: str | None = None) -> Checkpoint: | ||
| """Commit the current work tree as a checkpoint, or reuse the last one if nothing changed. | ||
|
|
||
| Stages every non-ignored path (`git add -A`) and commits it. If the work | ||
| tree is byte-identical to the most recent checkpoint, no commit is made and | ||
| that checkpoint is returned instead (debounce -- no empty duplicates). | ||
| """ | ||
| self.ensure_initialized() | ||
| self._git('add', '-A') |
There was a problem hiding this comment.
🔒 Security & Privacy | 🟠 Major | ⚡ Quick win
🧩 Analysis chain
🏁 Script executed:
#!/bin/bash
set -euo pipefail
echo '--- file list ---'
git ls-files pydantic_ai_harness/checkpoints/_shadow.py pydantic_ai_harness/checkpoints
echo '--- outline ---'
ast-grep outline pydantic_ai_harness/checkpoints/_shadow.py --view expanded || true
echo '--- relevant ranges ---'
sed -n '1,260p' pydantic_ai_harness/checkpoints/_shadow.py
echo '--- search for shadow_dir/project_root usage ---'
rg -n "shadow_dir|project_root|ensure_initialized|_has_head|_index_matches_head|git add -A|gitattributes|clean-filter|filter" pydantic_ai_harness -SRepository: pydantic/pydantic-ai-harness
Length of output: 22132
🏁 Script executed (no clone):
#!/bin/bash
set -euo pipefail
python3 - <<'PY'
import subprocess, tempfile, os, textwrap, pathlib, sys
with tempfile.TemporaryDirectory() as td:
repo = pathlib.Path(td) / 'repo'
repo.mkdir()
subprocess.run(['git', 'init', '-q', str(repo)], check=True)
# Create a filter that logs when invoked
log = repo / 'filter.log'
(repo / 'filt.sh').write_text(textwrap.dedent(f"""\
#!/bin/sh
echo "FILTER:$@" >> "{log}"
cat
"""))
os.chmod(repo / 'filt.sh', 0o755)
# Configure a local clean filter and attributes file
subprocess.run(['git', '-C', str(repo), 'config', 'filter.evil.clean', str(repo / 'filt.sh')], check=True)
subprocess.run(['git', '-C', str(repo), 'config', 'filter.evil.smudge', 'cat'], check=True)
(repo / '.gitattributes').write_text('x filter=evil\n')
(repo / 'x').write_text('payload')
# Run add -A and see whether filter is invoked
r = subprocess.run(['git', '-C', str(repo), 'add', '-A'], capture_output=True, text=True)
print('add rc', r.returncode)
print('stdout', r.stdout)
print('stderr', r.stderr)
print('log exists', log.exists())
if log.exists():
print(log.read_text())
PY
Length of output: 237
🏁 Script executed:
#!/bin/bash
set -euo pipefail
# Read the surrounding code with line numbers.
nl -ba pydantic_ai_harness/checkpoints/_shadow.py | sed -n '1,260p'
# Find the constructor and any path validation around shadow_dir.
rg -n "class .*Shadow|def __init__|shadow_dir|project_root|private|sentinel|ownership|HEAD" pydantic_ai_harness/checkpoints/_shadow.py -n -S
# Inspect any tests covering overlapping paths or existing repos.
rg -n "shadow_dir|project_root|HEAD|gitattributes|filter=|clean filter|checkpoint" tests pydantic_ai_harness -SRepository: pydantic/pydantic-ai-harness
Length of output: 206
🏁 Script executed:
#!/bin/bash
set -euo pipefail
# Read the capability layer and README around the state_dir guidance.
sed -n '1,240p' pydantic_ai_harness/checkpoints/_capability.py
echo '--- README excerpt ---'
sed -n '55,80p' pydantic_ai_harness/checkpoints/README.mdRepository: pydantic/pydantic-ai-harness
Length of output: 9772
Injection (CWE-78): Improper Neutralization of Special Elements used in an OS Command ('OS Command Injection')
Reachability: External
● Entry
pydantic_ai_harness/checkpoints/_capability.py:148
_store
│
▼
● Hop
tests/checkpoints/test_checkpoints.py
│
▼
● Hop
pydantic_ai_harness/checkpoints/__init__.py:17
CheckpointError
│
▼
● Sink
pydantic_ai_harness/checkpoints/_shadow.py
Reject overlapping shadow_dir and project_root If the shadow repo lives under the work tree, file-mutating tools can alter its local config, and the next git add -A will honor a .gitattributes clean filter and run it. Enforce a disjoint, harness-owned path instead of trusting any directory with HEAD.
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@pydantic_ai_harness/checkpoints/_shadow.py` around lines 149 - 176, Update
the shadow-directory initialization and validation around ensure_initialized to
reject any shadow_dir that overlaps project_root, including when either path is
an ancestor of the other. Resolve both paths before comparing them, require the
shadow repository to be a disjoint harness-owned location, and do not treat an
existing HEAD as sufficient validation; preserve normal initialization for valid
paths.
| self.shadow_dir.mkdir(parents=True, exist_ok=True) | ||
| except OSError as exc: | ||
| raise CheckpointError(f'could not create shadow repo at {self.shadow_dir}: {exc}') from exc | ||
| self._git('init', '-q', str(self.shadow_dir)) | ||
| info_dir = self.shadow_dir / 'info' | ||
| info_dir.mkdir(exist_ok=True) | ||
| (info_dir / 'exclude').write_text('.git/\n', encoding='utf-8') |
There was a problem hiding this comment.
🔒 Security & Privacy | 🟡 Minor | ⚡ Quick win
🧩 Analysis chain
🏁 Script executed:
#!/bin/bash
set -euo pipefail
git ls-files pydantic_ai_harness/checkpoints/_shadow.py
wc -l pydantic_ai_harness/checkpoints/_shadow.py
sed -n '1,260p' pydantic_ai_harness/checkpoints/_shadow.pyRepository: pydantic/pydantic-ai-harness
Length of output: 10816
🏁 Script executed:
#!/bin/bash
set -euo pipefail
rg -n "shadow_dir_for|ensure_initialized|state_dir|checkpoints/" pydantic_ai_harness -g '!**/__pycache__/**'Repository: pydantic/pydantic-ai-harness
Length of output: 1796
🏁 Script executed:
#!/bin/bash
set -euo pipefail
sed -n '1,240p' pydantic_ai_harness/checkpoints/_capability.py
printf '\n--- README ---\n'
sed -n '1,140p' pydantic_ai_harness/checkpoints/README.mdRepository: pydantic/pydantic-ai-harness
Length of output: 14043
🏁 Script executed:
#!/bin/bash
set -euo pipefail
rg -n "mkdir\\(|resolve\\(|symlink|is_symlink|chmod|0700|0o700|pydantic-ai-harness" pydantic_ai_harness -g '!**/__pycache__/**'Repository: pydantic/pydantic-ai-harness
Length of output: 13952
Sensitive Data Exposure (CWE-732): Incorrect Permission Assignment for Critical Resource
Reachability: External
● Entry
pydantic_ai_harness/checkpoints/_capability.py:148
_store
│
▼
● Hop
tests/checkpoints/test_checkpoints.py
│
▼
● Hop
pydantic_ai_harness/checkpoints/__init__.py:17
CheckpointError
│
▼
● Sink
pydantic_ai_harness/checkpoints/_shadow.py
Make the shadow repo private before git init
mkdir() leaves <state_dir>/checkpoints/... at umask-derived perms, so with a common 022 umask the checkpoint store and its Git objects can be read by other local users. That can expose snapshots of 0600 files. Create the directory with 0700 and reject reuse of an existing state dir unless it is owned by the current user and stays private.
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@pydantic_ai_harness/checkpoints/_shadow.py` around lines 152 - 158, Update
the shadow-directory setup before git init to create it with mode 0700 and
verify any existing directory is owned by the current user and has no group or
other permissions; reject unsafe reuse with CheckpointError. Anchor the change
in the directory creation logic before self._git('init', ...) while preserving
the existing error handling.
| async def restore_checkpoint(self, checkpoint_id: str, paths: list[str] | None = None) -> str: | ||
| """Restore project files from a checkpoint. | ||
|
|
||
| Args: | ||
| checkpoint_id: The id of a checkpoint from `list_checkpoints`. | ||
| paths: Project-relative paths to restore. Omit to restore every file the | ||
| checkpoint captured. | ||
| """ | ||
| await anyio.to_thread.run_sync(lambda: self._store.restore(checkpoint_id, paths=paths)) | ||
| scope = ', '.join(paths) if paths else 'all files' | ||
| return f'Restored {scope} from checkpoint {checkpoint_id}.' |
There was a problem hiding this comment.
🎯 Functional Correctness | 🟡 Minor | ⚡ Quick win
Empty paths list contradicts the "all files" confirmation message.
scope treats paths=[] as "all files" for messaging, but Line 44 forwards the empty list unchanged to store.restore, whose contract (_shadow.py) turns paths=None into ['.'] but leaves an explicit empty list as an empty pathspec, restoring nothing. A model passing paths=[] would get told everything was restored when nothing was.
🐛 Proposed fix to normalize empty paths
async def restore_checkpoint(self, checkpoint_id: str, paths: list[str] | None = None) -> str:
"""Restore project files from a checkpoint.
Args:
checkpoint_id: The id of a checkpoint from `list_checkpoints`.
paths: Project-relative paths to restore. Omit to restore every file the
checkpoint captured.
"""
- await anyio.to_thread.run_sync(lambda: self._store.restore(checkpoint_id, paths=paths))
+ normalized = paths or None
+ await anyio.to_thread.run_sync(lambda: self._store.restore(checkpoint_id, paths=normalized))
scope = ', '.join(paths) if paths else 'all files'
return f'Restored {scope} from checkpoint {checkpoint_id}.'📝 Committable suggestion
‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.
| async def restore_checkpoint(self, checkpoint_id: str, paths: list[str] | None = None) -> str: | |
| """Restore project files from a checkpoint. | |
| Args: | |
| checkpoint_id: The id of a checkpoint from `list_checkpoints`. | |
| paths: Project-relative paths to restore. Omit to restore every file the | |
| checkpoint captured. | |
| """ | |
| await anyio.to_thread.run_sync(lambda: self._store.restore(checkpoint_id, paths=paths)) | |
| scope = ', '.join(paths) if paths else 'all files' | |
| return f'Restored {scope} from checkpoint {checkpoint_id}.' | |
| async def restore_checkpoint(self, checkpoint_id: str, paths: list[str] | None = None) -> str: | |
| """Restore project files from a checkpoint. | |
| Args: | |
| checkpoint_id: The id of a checkpoint from `list_checkpoints`. | |
| paths: Project-relative paths to restore. Omit to restore every file the | |
| checkpoint captured. | |
| """ | |
| normalized = paths or None | |
| await anyio.to_thread.run_sync(lambda: self._store.restore(checkpoint_id, paths=normalized)) | |
| scope = ', '.join(paths) if paths else 'all files' | |
| return f'Restored {scope} from checkpoint {checkpoint_id}.' |
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@pydantic_ai_harness/checkpoints/_toolset.py` around lines 36 - 46, Normalize
an empty paths list to None before calling _store.restore in restore_checkpoint,
so paths=[] restores all files consistently with the existing “all files”
confirmation message. Preserve explicit non-empty path lists and the current
response formatting.
What
Adds
Checkpoints, a private/experimental capability that snapshots the working tree into a shadow git repo before any file-mutating tool runs, so an agent's file damage is restorable. Recommendation 5 of the broad CLI-harness survey; five harnesses (Gemini CLI, opencode, Cline, Roo, goose) converged on this.How
before_tool_executeon a tool whose name is inmutating_toolssnapshots the working tree into a shadow repo at<state_dir>/checkpoints/<project-slug>/—GIT_DIRthere,GIT_WORK_TREEat the project root, its own committer identity,commit.gpgsign=false, no hooks. It never touches the user's own.git, works in non-git projects (git inits the shadow repo itself), and respects.gitignore.checkpoints() -> list[Checkpoint],restore(checkpoint_id, *, paths=None). Optional model-facinglist_checkpoints/restore_checkpointtools behindexpose_tool=False(default off — restoring is usually a human/app action).Deliberate scope
restoreis additive (git checkoutsemantics — re-creates files present at the checkpoint, does not delete files created since; documented, not silent).CheckpointWarningand lets the run continue rather than aborting the agent.state_dirdefaults to~/.pydantic-ai-harness(noplatformdirsdep added).Tests
26 tests over tmp_path projects: snapshot-before-write, restore round-trip, path-scoped restore, non-git project, user's own
.gituntouched, debounce (no duplicate empty checkpoints),mutating_toolsconfig respected. 100% branch coverage on the module; full suite green. (Fixed one cross-version bug found in CI: git's%cItimestamp format rejected by Python <3.11 → read from%ct.)🤖 Generated with Claude Code