6716b72ef2
Move retention, orphan pruning, status and clear operations into a topical sibling and route callers and tests directly to their defining module. Keep the shared store paths and git execution in checkpoint_manager; shorten the local child-env WHAT docstring.
1539 lines
61 KiB
Python
1539 lines
61 KiB
Python
"""
|
||
Checkpoint Manager — Transparent filesystem snapshots via a single shared
|
||
shadow git store.
|
||
|
||
Creates automatic snapshots of working directories before file-mutating
|
||
operations (``write_file``, ``patch``, ``terminal`` with destructive flags),
|
||
triggered once per conversation turn. Provides rollback to any previous
|
||
checkpoint.
|
||
|
||
This is NOT a tool — the LLM never sees it. It's transparent infrastructure
|
||
controlled by the ``checkpoints`` config flag or ``--checkpoints`` CLI flag.
|
||
|
||
Storage layout (single shared store, git objects deduplicated across projects)
|
||
-----------------------------------------------------------------------------
|
||
|
||
~/.hermes/checkpoints/
|
||
store/ — single bare-ish git repo
|
||
HEAD, config, objects/ — standard git internals (shared)
|
||
refs/hermes/<hash16> — per-project branch tip
|
||
indexes/<hash16> — per-project git index
|
||
projects/<hash16>.json — {workdir, created_at, last_touch}
|
||
info/exclude — default excludes (shared)
|
||
.last_prune — auto-prune idempotency marker
|
||
legacy-<timestamp>/ — archived pre-v2 per-project shadow
|
||
repos (auto-migrated on first init)
|
||
|
||
Why a single store?
|
||
-------------------
|
||
|
||
The pre-v2 design kept a full shadow repo per working directory. Each one
|
||
re-stored most of the project's files under its own ``objects/`` tree, with
|
||
zero sharing across worktrees of the same project. A single user with a
|
||
dozen worktrees of the same repo burned ~40 MB each (~500 MB total) storing
|
||
the same blobs over and over. A single shared store lets git's content-
|
||
addressable object DB deduplicate across projects and across turns, so adding
|
||
a new worktree costs near-zero.
|
||
|
||
The shadow store uses ``GIT_DIR`` + ``GIT_WORK_TREE`` + ``GIT_INDEX_FILE``
|
||
so no git state leaks into the user's project directory.
|
||
|
||
Auto-maintenance (``tools.checkpoint_maintenance``)
|
||
---------------------------------------------------
|
||
|
||
Shadow state accumulates over time. ``prune_checkpoints`` deletes refs whose
|
||
recorded working directory no longer exists (orphan) or whose last touch is
|
||
older than ``retention_days`` (stale), then runs ``git gc --prune=now`` to
|
||
reclaim object storage. A size-cap pass drops the oldest checkpoints per
|
||
project until total store size is under ``max_total_size_mb``.
|
||
"""
|
||
|
||
import hashlib
|
||
import json
|
||
import logging
|
||
import os
|
||
import re
|
||
import shutil
|
||
import subprocess
|
||
import time
|
||
from pathlib import Path
|
||
from hermes_constants import get_hermes_home
|
||
from hermes_cli._subprocess_compat import selected_git_env, windows_hide_flags
|
||
from hermes_cli.gitlock import clear_stale_tmp_packs
|
||
from typing import Dict, List, Optional, Set, Tuple
|
||
|
||
from utils import env_int, rmtree_readonly
|
||
|
||
logger = logging.getLogger(__name__)
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# Constants
|
||
# ---------------------------------------------------------------------------
|
||
|
||
CHECKPOINT_BASE = get_hermes_home() / "checkpoints"
|
||
_CHECKPOINT_BASE_AT_IMPORT = CHECKPOINT_BASE
|
||
|
||
|
||
def _resolve_checkpoint_base() -> Path:
|
||
"""Active profile's checkpoint root at call time: the patched ``CHECKPOINT_BASE`` when a test
|
||
changed it, else live profile-scoped HERMES_HOME — under the multiplexed gateway one process
|
||
serves every profile, so the import-time constant would write every profile's code-edit
|
||
checkpoints into the launch profile's store."""
|
||
return CHECKPOINT_BASE if CHECKPOINT_BASE != _CHECKPOINT_BASE_AT_IMPORT else get_hermes_home() / "checkpoints"
|
||
|
||
# Single shared store directory under CHECKPOINT_BASE.
|
||
_STORE_DIRNAME = "store"
|
||
_REFS_PREFIX = "refs/hermes"
|
||
_INDEXES_DIRNAME = "indexes"
|
||
_PROJECTS_DIRNAME = "projects"
|
||
_LEDGERS_DIRNAME = "ledgers"
|
||
_LEGACY_PREFIX = "legacy-"
|
||
_PRUNE_MARKER_NAME = ".last_prune"
|
||
|
||
# Agent-write ledger cap: newest entries retained per project.
|
||
_LEDGER_MAX_ENTRIES = 2000
|
||
|
||
DEFAULT_EXCLUDES = [
|
||
# Dependency / build output
|
||
"node_modules/",
|
||
"dist/",
|
||
"build/",
|
||
"target/",
|
||
"out/",
|
||
".next/",
|
||
".nuxt/",
|
||
# Caches
|
||
"__pycache__/",
|
||
"*.pyc",
|
||
"*.pyo",
|
||
".cache/",
|
||
".pytest_cache/",
|
||
".mypy_cache/",
|
||
".ruff_cache/",
|
||
"coverage/",
|
||
".coverage",
|
||
# Virtualenvs
|
||
".venv/",
|
||
"venv/",
|
||
"env/",
|
||
# VCS
|
||
".git/",
|
||
".hg/",
|
||
".svn/",
|
||
# Worktrees (Hermes convention — don't recursively snapshot siblings)
|
||
".worktrees/",
|
||
# Native / compiled binaries
|
||
"*.so",
|
||
"*.dylib",
|
||
"*.dll",
|
||
"*.o",
|
||
"*.a",
|
||
"*.jar",
|
||
"*.class",
|
||
"*.exe",
|
||
"*.obj",
|
||
# Media / large binaries
|
||
"*.mp4",
|
||
"*.mov",
|
||
"*.mkv",
|
||
"*.webm",
|
||
"*.zip",
|
||
"*.tar",
|
||
"*.tar.gz",
|
||
"*.tgz",
|
||
"*.7z",
|
||
"*.rar",
|
||
"*.iso",
|
||
# Secrets
|
||
".env",
|
||
".env.*",
|
||
".env.local",
|
||
".env.*.local",
|
||
# OS junk
|
||
".DS_Store",
|
||
"Thumbs.db",
|
||
# Logs
|
||
"*.log",
|
||
]
|
||
|
||
# Git subprocess timeout (seconds).
|
||
_GIT_TIMEOUT: int = max(10, min(60, env_int("HERMES_CHECKPOINT_TIMEOUT", 30)))
|
||
|
||
# Max files to snapshot — skip huge directories to avoid slowdowns.
|
||
_MAX_FILES = 50_000
|
||
|
||
# Valid git commit hash pattern: 4–40 hex chars (short or full SHA-1/SHA-256).
|
||
_COMMIT_HASH_RE = re.compile(r'^[0-9a-fA-F]{4,64}$')
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# Input validation helpers
|
||
# ---------------------------------------------------------------------------
|
||
|
||
def _validate_commit_hash(commit_hash: str) -> Optional[str]:
|
||
"""Validate a commit hash to prevent git argument injection.
|
||
|
||
Returns an error string if invalid, None if valid.
|
||
Values starting with '-' would be interpreted as git flags
|
||
(e.g., '--patch', '-p') instead of revision specifiers.
|
||
"""
|
||
if not commit_hash or not commit_hash.strip():
|
||
return "Empty commit hash"
|
||
if commit_hash.startswith("-"):
|
||
return f"Invalid commit hash (must not start with '-'): {commit_hash!r}"
|
||
if not _COMMIT_HASH_RE.match(commit_hash):
|
||
return f"Invalid commit hash (expected 4-64 hex characters): {commit_hash!r}"
|
||
return None
|
||
|
||
|
||
def _validate_file_path(file_path: str, working_dir: str) -> Optional[str]:
|
||
"""Validate a file path to prevent path traversal outside the working directory.
|
||
|
||
Returns an error string if invalid, None if valid.
|
||
"""
|
||
if not file_path or not file_path.strip():
|
||
return "Empty file path"
|
||
if os.path.isabs(file_path):
|
||
return f"File path must be relative, got absolute path: {file_path!r}"
|
||
abs_workdir = _normalize_path(working_dir)
|
||
resolved = (abs_workdir / file_path).resolve()
|
||
try:
|
||
resolved.relative_to(abs_workdir)
|
||
except ValueError:
|
||
return f"File path escapes the working directory via traversal: {file_path!r}"
|
||
return None
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# Path / hash helpers
|
||
# ---------------------------------------------------------------------------
|
||
|
||
def _normalize_path(path_value: str) -> Path:
|
||
"""Return a canonical absolute path for checkpoint operations."""
|
||
return Path(path_value).expanduser().resolve()
|
||
|
||
|
||
def _project_hash(working_dir: str) -> str:
|
||
"""Deterministic per-project hash: sha256(abs_path)[:16]."""
|
||
abs_path = str(_normalize_path(working_dir))
|
||
return hashlib.sha256(abs_path.encode()).hexdigest()[:16]
|
||
|
||
|
||
def _store_path(base: Optional[Path] = None) -> Path:
|
||
"""Return the single shared shadow store path."""
|
||
return (base or _resolve_checkpoint_base()) / _STORE_DIRNAME
|
||
|
||
|
||
def _store_has_head(store: Path) -> bool:
|
||
return (store / "HEAD").exists()
|
||
|
||
|
||
def _index_path(store: Path, dir_hash: str) -> Path:
|
||
return store / _INDEXES_DIRNAME / dir_hash
|
||
|
||
|
||
def _ledger_path(store: Path, dir_hash: str) -> Path:
|
||
return store / _LEDGERS_DIRNAME / f"{dir_hash}.json"
|
||
|
||
|
||
def _hash_file(path: Path) -> Optional[str]:
|
||
"""Streaming sha256 of a file's bytes. None if unreadable/missing."""
|
||
try:
|
||
h = hashlib.sha256()
|
||
with open(path, "rb") as fh:
|
||
for chunk in iter(lambda: fh.read(1 << 20), b""):
|
||
h.update(chunk)
|
||
return h.hexdigest()
|
||
except OSError:
|
||
return None
|
||
|
||
|
||
def _load_ledger(store: Path, dir_hash: str) -> Dict[str, Dict]:
|
||
"""Load the agent-write ledger: {relpath: {"sha256": ..., "ts": ...}}.
|
||
|
||
The ledger records the content hash of every file the last successful
|
||
``write_file`` / ``patch`` produced, so restores can tell "Hermes wrote
|
||
this" apart from "the user hand-edited this afterwards".
|
||
"""
|
||
try:
|
||
raw = _ledger_path(store, dir_hash).read_text(encoding="utf-8-sig")
|
||
data = json.loads(raw)
|
||
return data if isinstance(data, dict) else {}
|
||
except (OSError, ValueError):
|
||
return {}
|
||
|
||
|
||
def _save_ledger(store: Path, dir_hash: str, ledger: Dict[str, Dict]) -> None:
|
||
"""Persist the agent-write ledger, capped to the newest entries."""
|
||
try:
|
||
if len(ledger) > _LEDGER_MAX_ENTRIES:
|
||
newest = sorted(
|
||
ledger.items(),
|
||
key=lambda kv: kv[1].get("ts", 0) if isinstance(kv[1], dict) else 0,
|
||
reverse=True,
|
||
)[:_LEDGER_MAX_ENTRIES]
|
||
ledger = dict(newest)
|
||
path = _ledger_path(store, dir_hash)
|
||
path.parent.mkdir(parents=True, exist_ok=True)
|
||
tmp = path.with_suffix(".json.tmp")
|
||
tmp.write_text(json.dumps(ledger), encoding="utf-8")
|
||
tmp.replace(path)
|
||
except OSError:
|
||
logger.debug("Failed to save agent-write ledger for %s", dir_hash, exc_info=True)
|
||
|
||
|
||
def _ref_name(dir_hash: str) -> str:
|
||
return f"{_REFS_PREFIX}/{dir_hash}"
|
||
|
||
|
||
def _project_meta_path(store: Path, dir_hash: str) -> Path:
|
||
return store / _PROJECTS_DIRNAME / f"{dir_hash}.json"
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# Git env
|
||
# ---------------------------------------------------------------------------
|
||
|
||
def _git_env(
|
||
store: Path,
|
||
working_dir: str,
|
||
index_file: Optional[Path] = None,
|
||
) -> dict:
|
||
"""Build env dict that redirects git to the shared store.
|
||
|
||
The shared store is internal Hermes infrastructure — it must NOT inherit
|
||
the user's global or system git config. User-level settings like
|
||
``commit.gpgsign = true``, signing hooks, or credential helpers would
|
||
either break background snapshots or, worse, spawn interactive prompts
|
||
(pinentry GUI windows) mid-session every time a file is written.
|
||
|
||
Isolation strategy:
|
||
* ``GIT_CONFIG_GLOBAL=<os.devnull>`` — ignore ``~/.gitconfig`` (git 2.32+).
|
||
* ``GIT_CONFIG_SYSTEM=<os.devnull>`` — ignore ``/etc/gitconfig`` (git 2.32+).
|
||
* ``GIT_CONFIG_NOSYSTEM=1`` — legacy belt-and-suspenders for older git.
|
||
|
||
``index_file``, if given, forces git to use a per-project index under
|
||
``store/indexes/<hash>`` so projects don't race on a shared index.
|
||
"""
|
||
normalized_working_dir = _normalize_path(working_dir)
|
||
# git child with hand-isolated config env; exact preservation — a HOME
|
||
# rewrite would change which ~/.gitconfig the isolation vars are hiding.
|
||
from tools.environments.local import build_subprocess_env
|
||
|
||
env = selected_git_env(build_subprocess_env(scrub_secrets=False, inherit_profile_home=False))
|
||
env["GIT_DIR"] = str(store)
|
||
env["GIT_WORK_TREE"] = str(normalized_working_dir)
|
||
env.pop("GIT_NAMESPACE", None)
|
||
env.pop("GIT_ALTERNATE_OBJECT_DIRECTORIES", None)
|
||
if index_file is not None:
|
||
env["GIT_INDEX_FILE"] = str(index_file)
|
||
else:
|
||
env.pop("GIT_INDEX_FILE", None)
|
||
env["GIT_CONFIG_GLOBAL"] = os.devnull
|
||
env["GIT_CONFIG_SYSTEM"] = os.devnull
|
||
env["GIT_CONFIG_NOSYSTEM"] = "1"
|
||
return env
|
||
|
||
|
||
def _run_git(
|
||
args: List[str],
|
||
store: Path,
|
||
working_dir: str,
|
||
timeout: int = _GIT_TIMEOUT,
|
||
allowed_returncodes: Optional[Set[int]] = None,
|
||
index_file: Optional[Path] = None,
|
||
extra_env: Optional[Dict[str, str]] = None,
|
||
) -> Tuple[bool, str, str]:
|
||
"""Run a git command against the shared store. Returns (ok, stdout, stderr).
|
||
|
||
``allowed_returncodes`` suppresses error logging for known/expected non-zero
|
||
exits while preserving the normal ``ok = (returncode == 0)`` contract.
|
||
Example: ``git diff --cached --quiet`` returns 1 when changes exist.
|
||
"""
|
||
normalized_working_dir = _normalize_path(working_dir)
|
||
if not normalized_working_dir.exists():
|
||
msg = f"working directory not found: {normalized_working_dir}"
|
||
logger.error("Git command skipped: %s (%s)", " ".join(["git"] + list(args)), msg)
|
||
return False, "", msg
|
||
if not normalized_working_dir.is_dir():
|
||
msg = f"working directory is not a directory: {normalized_working_dir}"
|
||
logger.error("Git command skipped: %s (%s)", " ".join(["git"] + list(args)), msg)
|
||
return False, "", msg
|
||
|
||
env = _git_env(store, str(normalized_working_dir), index_file=index_file)
|
||
if extra_env:
|
||
env.update(extra_env)
|
||
git = shutil.which("git", path=env.get("PATH", ""))
|
||
if git is None:
|
||
return False, "", "git is not installed or not on PATH"
|
||
cmd = [git, *args]
|
||
allowed_returncodes = allowed_returncodes or set()
|
||
|
||
try:
|
||
# NUL-delimited git output contains literal filenames, not text lines.
|
||
text_options = {} if "-z" in args else {"text": True, "encoding": "utf-8", "errors": "replace"}
|
||
result = subprocess.run(
|
||
cmd,
|
||
capture_output=True,
|
||
**text_options,
|
||
timeout=timeout,
|
||
env=env,
|
||
cwd=str(normalized_working_dir),
|
||
stdin=subprocess.DEVNULL,
|
||
# Checkpoints fire several bare git calls per turn from the
|
||
# console-less desktop/gateway backend; suppress the per-call
|
||
# conhost flash on Windows (no-op on POSIX).
|
||
creationflags=windows_hide_flags(),
|
||
)
|
||
ok = result.returncode == 0
|
||
stdout = os.fsdecode(result.stdout) if "-z" in args else result.stdout.strip()
|
||
stderr = result.stderr.decode("utf-8", errors="replace").strip() if "-z" in args else result.stderr.strip()
|
||
if not ok and result.returncode not in allowed_returncodes:
|
||
logger.error(
|
||
"Git command failed: %s (rc=%d) stderr=%s",
|
||
" ".join(cmd), result.returncode, stderr,
|
||
)
|
||
return ok, stdout, stderr
|
||
except subprocess.TimeoutExpired:
|
||
msg = f"git timed out after {timeout}s: {' '.join(cmd)}"
|
||
logger.error(msg, exc_info=True)
|
||
return False, "", msg
|
||
except FileNotFoundError as exc:
|
||
missing_target = getattr(exc, "filename", None)
|
||
if missing_target == "git":
|
||
logger.error("Git executable not found: %s", " ".join(cmd), exc_info=True)
|
||
return False, "", "git not found"
|
||
msg = f"working directory not found: {normalized_working_dir}"
|
||
logger.error("Git command failed before execution: %s (%s)", " ".join(cmd), msg, exc_info=True)
|
||
return False, "", msg
|
||
except Exception as exc:
|
||
logger.error("Unexpected git error running %s: %s", " ".join(cmd), exc, exc_info=True)
|
||
return False, "", str(exc)
|
||
|
||
|
||
def _git_out(args: List[str], store: Path, working_dir: str, rc: Optional[Set[int]] = None) -> str:
|
||
"""stdout of a successful git call, else ``""``."""
|
||
ok, out, _ = _run_git(args, store, working_dir, allowed_returncodes=rc)
|
||
return out if ok else ""
|
||
|
||
|
||
def _ref_tip(store: Path, working_dir: str, ref: str) -> Optional[str]:
|
||
"""Commit sha at ``ref``, or None when the ref does not exist yet."""
|
||
return _git_out(["rev-parse", "--verify", ref + "^{commit}"], store, working_dir, {128}) or None
|
||
|
||
|
||
def _list_project_refs(store: Path, working_dir: str) -> List[str]:
|
||
out = _git_out(["for-each-ref", "--format=%(refname)", _REFS_PREFIX], store, working_dir, {128})
|
||
return [r for r in out.splitlines() if r.strip()]
|
||
|
||
|
||
def _unlink_quiet(path: Path) -> None:
|
||
try:
|
||
path.unlink(missing_ok=True)
|
||
except OSError:
|
||
pass
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# Store initialisation + legacy migration
|
||
# ---------------------------------------------------------------------------
|
||
|
||
def _migrate_legacy_store(base: Path) -> Optional[Path]:
|
||
"""Move pre-v2 per-project shadow repos into a ``legacy-<ts>/`` dir.
|
||
|
||
The pre-v2 layout had one shadow git repo per working directory directly
|
||
under ``CHECKPOINT_BASE``. The v2 layout wants a single ``store/`` dir.
|
||
Rather than delete the old data (users might want to recover), rename
|
||
everything except our own v2 entries into ``legacy-<timestamp>/``. The
|
||
legacy dir is subject to the same retention sweep and can be manually
|
||
cleared with ``hermes checkpoints clear-legacy``.
|
||
|
||
Returns the legacy-archive path, or None if nothing to migrate.
|
||
"""
|
||
if not base.exists():
|
||
return None
|
||
store = _store_path(base)
|
||
legacy_root: Optional[Path] = None
|
||
# Reserved top-level entries managed by v2.
|
||
reserved = {_STORE_DIRNAME, _PRUNE_MARKER_NAME}
|
||
for child in list(base.iterdir()):
|
||
name = child.name
|
||
if name in reserved or name.startswith(_LEGACY_PREFIX):
|
||
continue
|
||
# Candidate: pre-v2 shadow repo (has HEAD) OR stray dir. Either way
|
||
# we archive it so v2 starts clean.
|
||
if legacy_root is None:
|
||
stamp = time.strftime("%Y%m%d-%H%M%S")
|
||
legacy_root = base / f"{_LEGACY_PREFIX}{stamp}"
|
||
try:
|
||
legacy_root.mkdir(parents=True, exist_ok=True)
|
||
except OSError as exc:
|
||
logger.warning("Could not create legacy archive dir: %s", exc)
|
||
return None
|
||
dest = legacy_root / name
|
||
try:
|
||
shutil.move(str(child), str(dest))
|
||
except OSError as exc:
|
||
logger.warning("Could not archive legacy checkpoint %s: %s", child, exc)
|
||
# If the store still hasn't been created, create it here.
|
||
_ = store
|
||
if legacy_root is not None:
|
||
logger.info(
|
||
"Migrated pre-v2 checkpoint repos to %s. "
|
||
"Clear with `hermes checkpoints clear-legacy` when safe.",
|
||
legacy_root,
|
||
)
|
||
return legacy_root
|
||
|
||
|
||
def _init_store(store: Path, working_dir: str) -> Optional[str]:
|
||
"""Initialise the shared shadow store if needed. Returns error or None.
|
||
|
||
Also performs one-time migration of pre-v2 per-directory shadow repos
|
||
into ``legacy-<timestamp>/``.
|
||
"""
|
||
base = store.parent
|
||
# One-time legacy migration before we create the store.
|
||
if not store.exists():
|
||
try:
|
||
base.mkdir(parents=True, exist_ok=True)
|
||
except OSError as exc:
|
||
return f"Could not create checkpoint base: {exc}"
|
||
# Only migrate if the base dir has pre-existing content that isn't
|
||
# our own v2 layout.
|
||
_migrate_legacy_store(base)
|
||
|
||
if (store / "HEAD").exists():
|
||
return None
|
||
|
||
store.mkdir(parents=True, exist_ok=True)
|
||
(store / _INDEXES_DIRNAME).mkdir(exist_ok=True)
|
||
(store / _PROJECTS_DIRNAME).mkdir(exist_ok=True)
|
||
|
||
# ``git init --bare`` rejects GIT_WORK_TREE, so we can't use _run_git
|
||
# here (which always sets GIT_DIR + GIT_WORK_TREE). Use a raw
|
||
# subprocess with just the config-isolation env vars.
|
||
from tools.environments.local import build_subprocess_env
|
||
|
||
init_env = selected_git_env(build_subprocess_env(scrub_secrets=False, inherit_profile_home=False))
|
||
init_env["GIT_CONFIG_GLOBAL"] = os.devnull
|
||
init_env["GIT_CONFIG_SYSTEM"] = os.devnull
|
||
init_env["GIT_CONFIG_NOSYSTEM"] = "1"
|
||
# Drop any inherited GIT_* that would interfere.
|
||
for k in ("GIT_DIR", "GIT_WORK_TREE", "GIT_INDEX_FILE", "GIT_NAMESPACE",
|
||
"GIT_ALTERNATE_OBJECT_DIRECTORIES"):
|
||
init_env.pop(k, None)
|
||
try:
|
||
git = shutil.which("git", path=init_env.get("PATH", ""))
|
||
if git is None:
|
||
return "Shadow store init failed: git is not installed or not on PATH"
|
||
result = subprocess.run(
|
||
[git, "init", "--bare", str(store)],
|
||
capture_output=True, text=True, encoding='utf-8', errors='replace',
|
||
env=init_env, timeout=_GIT_TIMEOUT,
|
||
stdin=subprocess.DEVNULL,
|
||
creationflags=windows_hide_flags(),
|
||
)
|
||
if result.returncode != 0:
|
||
return f"Shadow store init failed: {result.stderr.strip()}"
|
||
except (subprocess.TimeoutExpired, FileNotFoundError) as exc:
|
||
return f"Shadow store init failed: {exc}"
|
||
|
||
# Per-store config (isolated by env vars above, but belt-and-suspenders).
|
||
# Use the base dir as the working_dir for config commands — it always
|
||
# exists since we just created the store inside it.
|
||
cfg_wd = str(base)
|
||
_run_git(["config", "user.email", "hermes@local"], store, cfg_wd)
|
||
_run_git(["config", "user.name", "Hermes Checkpoint"], store, cfg_wd)
|
||
_run_git(["config", "commit.gpgsign", "false"], store, cfg_wd)
|
||
_run_git(["config", "tag.gpgSign", "false"], store, cfg_wd)
|
||
_run_git(["config", "gc.auto", "0"], store, cfg_wd)
|
||
|
||
info_dir = store / "info"
|
||
info_dir.mkdir(exist_ok=True)
|
||
(info_dir / "exclude").write_text(
|
||
"\n".join(DEFAULT_EXCLUDES) + "\n", encoding="utf-8"
|
||
)
|
||
|
||
logger.debug("Initialised checkpoint store at %s", store)
|
||
return None
|
||
|
||
|
||
def _volume_evidence(workdir: Path) -> Dict:
|
||
"""Record the identity of ``workdir``'s parent while the project is live.
|
||
|
||
``(st_dev, st_ino)`` of the parent directory, captured at a moment when
|
||
the workdir itself is reachable, identifies the *directory* — not just
|
||
the path. A mount point resolves to the mounted filesystem's root while
|
||
the volume is attached and to the underlying (underlay) directory after
|
||
unmount: same path, different directory, different ``(st_dev, st_ino)``.
|
||
Orphan pruning uses this to distinguish "the project was deleted out of
|
||
the directory we knew" from "a different directory is now visible at
|
||
that path because the volume is detached".
|
||
|
||
Returns ``{}`` when the workdir is not currently reachable, when the
|
||
filesystem does not provide a usable directory identity (a zero
|
||
``st_dev`` or ``st_ino`` — e.g. Windows filesystems without file IDs and
|
||
some network shares), or when the probe fails — callers treat all of
|
||
these as "no evidence recorded" and orphan pruning stays conservative
|
||
for the project (never classified as orphan; retention still applies).
|
||
"""
|
||
try:
|
||
if not workdir.exists():
|
||
return {}
|
||
st = workdir.parent.stat()
|
||
if not st.st_dev or not st.st_ino:
|
||
return {}
|
||
return {
|
||
"workdir_parent_dev": st.st_dev,
|
||
"workdir_parent_ino": st.st_ino,
|
||
}
|
||
except OSError:
|
||
return {}
|
||
|
||
|
||
def _register_project(store: Path, working_dir: str) -> None:
|
||
"""Create or update ``projects/<hash>.json`` with workdir + timestamps."""
|
||
dir_hash = _project_hash(working_dir)
|
||
meta_path = _project_meta_path(store, dir_hash)
|
||
now = time.time()
|
||
meta: Dict = {"workdir": str(_normalize_path(working_dir)),
|
||
"created_at": now, "last_touch": now}
|
||
evidence = _volume_evidence(_normalize_path(working_dir))
|
||
if evidence:
|
||
meta.update(evidence)
|
||
if meta_path.exists():
|
||
try:
|
||
existing = json.loads(meta_path.read_text(encoding="utf-8-sig"))
|
||
if isinstance(existing, dict):
|
||
meta["created_at"] = existing.get("created_at", now)
|
||
if not evidence:
|
||
# Fresh probe failed — keep the previously recorded
|
||
# parent identity rather than dropping it. Stale evidence
|
||
# only makes pruning MORE conservative (mismatch => not
|
||
# an orphan).
|
||
for key in ("workdir_parent_dev", "workdir_parent_ino"):
|
||
if key in existing:
|
||
meta[key] = existing[key]
|
||
except (OSError, ValueError):
|
||
pass
|
||
try:
|
||
meta_path.parent.mkdir(parents=True, exist_ok=True)
|
||
meta_path.write_text(json.dumps(meta), encoding="utf-8")
|
||
except OSError as exc:
|
||
logger.debug("Could not write project metadata %s: %s", meta_path, exc)
|
||
|
||
|
||
def _touch_project(store: Path, working_dir: str) -> None:
|
||
"""Update last_touch for a project, preserving created_at."""
|
||
dir_hash = _project_hash(working_dir)
|
||
meta_path = _project_meta_path(store, dir_hash)
|
||
if not meta_path.exists():
|
||
_register_project(store, working_dir)
|
||
return
|
||
try:
|
||
meta = json.loads(meta_path.read_text(encoding="utf-8-sig"))
|
||
except (OSError, ValueError):
|
||
meta = {}
|
||
if not isinstance(meta, dict):
|
||
meta = {}
|
||
meta["workdir"] = str(_normalize_path(working_dir))
|
||
meta["last_touch"] = time.time()
|
||
meta.setdefault("created_at", meta["last_touch"])
|
||
# Refresh the parent-directory identity while the project is observably
|
||
# live — a remount can legitimately change it (new device, new inode).
|
||
# On probe failure the previous evidence is kept: stale evidence can only
|
||
# make pruning MORE conservative (mismatch => not an orphan).
|
||
evidence = _volume_evidence(_normalize_path(working_dir))
|
||
if evidence:
|
||
meta.update(evidence)
|
||
try:
|
||
meta_path.write_text(json.dumps(meta), encoding="utf-8")
|
||
except OSError as exc:
|
||
logger.debug("Could not update project metadata %s: %s", meta_path, exc)
|
||
|
||
|
||
def _list_projects(store: Path) -> List[Dict]:
|
||
"""Return all registered projects under the store."""
|
||
projects_dir = store / _PROJECTS_DIRNAME
|
||
if not projects_dir.exists():
|
||
return []
|
||
out: List[Dict] = []
|
||
for meta_path in projects_dir.glob("*.json"):
|
||
dir_hash = meta_path.stem
|
||
try:
|
||
meta = json.loads(meta_path.read_text(encoding="utf-8-sig"))
|
||
except (OSError, ValueError):
|
||
continue
|
||
if not isinstance(meta, dict):
|
||
continue
|
||
meta["_hash"] = dir_hash
|
||
out.append(meta)
|
||
return out
|
||
|
||
|
||
def _pre_v2_shadow_repos(base: Path) -> List[Dict]:
|
||
"""Return pre-v2 per-project shadow repos still directly under ``base``.
|
||
|
||
Pre-v2 layout kept one shadow git repo per working directory directly
|
||
under ``CHECKPOINT_BASE`` (identified by a ``HEAD`` file). This is the
|
||
single source of truth for that scan so a preview built from it (e.g.
|
||
``store_status``) always matches what ``prune_checkpoints`` deletes.
|
||
"""
|
||
out: List[Dict] = []
|
||
if not base.exists():
|
||
return out
|
||
for child in base.iterdir():
|
||
if not child.is_dir():
|
||
continue
|
||
if child.name == _STORE_DIRNAME or child.name.startswith(_LEGACY_PREFIX):
|
||
continue
|
||
if not (child / "HEAD").exists():
|
||
continue
|
||
workdir: Optional[str] = None
|
||
marker_unreadable = False
|
||
wd_marker = child / "HERMES_WORKDIR"
|
||
if wd_marker.exists():
|
||
try:
|
||
workdir = wd_marker.read_text(encoding="utf-8-sig").strip()
|
||
except (OSError, UnicodeDecodeError):
|
||
# The marker is there, we just could not read it. That is
|
||
# not evidence the project is gone — never delete on it.
|
||
workdir = None
|
||
marker_unreadable = True
|
||
out.append({
|
||
"path": child,
|
||
"workdir": workdir,
|
||
"exists": bool(workdir) and Path(workdir).exists(),
|
||
"marker_unreadable": marker_unreadable,
|
||
})
|
||
return out
|
||
|
||
|
||
def _dir_file_count(path: str) -> int:
|
||
"""Quick file count estimate (stops early if over _MAX_FILES)."""
|
||
count = 0
|
||
try:
|
||
for _ in Path(path).rglob("*"):
|
||
count += 1
|
||
if count > _MAX_FILES:
|
||
return count
|
||
except (PermissionError, OSError):
|
||
pass
|
||
return count
|
||
|
||
|
||
def _dir_size_bytes(path: Path) -> int:
|
||
"""Best-effort recursive size in bytes. Returns 0 on error."""
|
||
total = 0
|
||
try:
|
||
for p in path.rglob("*"):
|
||
try:
|
||
if p.is_file():
|
||
total += p.stat().st_size
|
||
except OSError:
|
||
continue
|
||
except OSError:
|
||
pass
|
||
return total
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# CheckpointManager
|
||
# ---------------------------------------------------------------------------
|
||
|
||
class CheckpointManager:
|
||
"""Manages automatic filesystem checkpoints.
|
||
|
||
Designed to be owned by AIAgent. Call ``new_turn()`` at the start of
|
||
each conversation turn and ``ensure_checkpoint(dir, reason)`` before
|
||
any file-mutating tool call. The manager deduplicates so at most one
|
||
snapshot is taken per directory per turn.
|
||
|
||
Parameters
|
||
----------
|
||
enabled : bool
|
||
Master switch (from config / CLI flag).
|
||
max_snapshots : int
|
||
Keep at most this many checkpoints per directory.
|
||
max_total_size_mb : int
|
||
Hard ceiling on total store size. Oldest checkpoints per project
|
||
are dropped when the store exceeds this after a commit.
|
||
max_file_size_mb : int
|
||
Skip adding any single file larger than this to a checkpoint.
|
||
(Implemented via ``.gitignore`` excludes + a post-stage size check.)
|
||
"""
|
||
|
||
def __init__(
|
||
self,
|
||
enabled: bool = False,
|
||
max_snapshots: int = 20,
|
||
max_total_size_mb: int = 500,
|
||
max_file_size_mb: int = 10,
|
||
):
|
||
self.enabled = enabled
|
||
self.max_snapshots = max(1, int(max_snapshots))
|
||
self.max_total_size_mb = max(0, int(max_total_size_mb))
|
||
self.max_file_size_mb = max(0, int(max_file_size_mb))
|
||
self._checkpointed_dirs: Set[str] = set()
|
||
|
||
# ------------------------------------------------------------------
|
||
# Turn lifecycle
|
||
# ------------------------------------------------------------------
|
||
|
||
def new_turn(self) -> None:
|
||
"""Reset per-turn dedup. Call at the start of each agent iteration."""
|
||
self._checkpointed_dirs.clear()
|
||
|
||
def unsupported_backend_reason(self, task_id: str = "default") -> Optional[str]:
|
||
"""Explain why host checkpoints are off limits for a container-backed session.
|
||
|
||
Classifies the task's backend at call time (nothing is remembered), so /rollback is
|
||
refused before the first mutation and follows a backend change within the session."""
|
||
from tools.file_tools_paths import container_backend_for_task
|
||
backend = container_backend_for_task(task_id)
|
||
if backend is None:
|
||
return None
|
||
return (
|
||
f"Checkpoints are not taken for terminal.backend={backend}: "
|
||
"file paths belong to the container, not this host."
|
||
)
|
||
|
||
# ------------------------------------------------------------------
|
||
# Public API
|
||
# ------------------------------------------------------------------
|
||
|
||
def record_agent_write(self, file_path: str) -> None:
|
||
"""Record the content hash of a file Hermes just successfully wrote.
|
||
|
||
Feeds the agent-write ledger used by :meth:`restore` in safe mode:
|
||
at restore time, a file whose current content no longer matches the
|
||
recorded hash was hand-edited by the user after Hermes last touched
|
||
it, and is skipped instead of clobbered.
|
||
|
||
Never raises — the ledger is best-effort bookkeeping.
|
||
"""
|
||
if not self.enabled:
|
||
return
|
||
try:
|
||
from tools.checkpoint_pruning import store_lock
|
||
|
||
path = _normalize_path(file_path)
|
||
digest = _hash_file(path)
|
||
if digest is None:
|
||
return
|
||
with store_lock(_resolve_checkpoint_base()):
|
||
store = _store_path()
|
||
dir_hash = self._ledger_key(str(path))
|
||
ledger = _load_ledger(store, dir_hash)
|
||
ledger[str(path)] = {"sha256": digest, "ts": time.time()}
|
||
_save_ledger(store, dir_hash, ledger)
|
||
except Exception as exc:
|
||
logger.debug("record_agent_write failed for %s: %s", file_path, exc)
|
||
|
||
def _safe_restore_plan(self, working_dir: str, commit_hash: str) -> Dict:
|
||
"""Classify files changed since ``commit_hash`` for a safe restore.
|
||
|
||
Returns ``{"success", "restore": [rel...], "skipped": [rel...],
|
||
"error"?}`` where ``restore`` lists files whose current content
|
||
still matches what Hermes last wrote (per the agent-write ledger)
|
||
and ``skipped`` lists files the user hand-edited after Hermes'
|
||
last write or that Hermes never wrote at all.
|
||
"""
|
||
hash_err = _validate_commit_hash(commit_hash)
|
||
if hash_err:
|
||
return {"success": False, "error": hash_err}
|
||
|
||
abs_dir = str(_normalize_path(working_dir))
|
||
store = _store_path()
|
||
if not (store / "HEAD").exists():
|
||
return {"success": False, "error": "No checkpoints exist for this directory"}
|
||
|
||
dir_hash = _project_hash(abs_dir)
|
||
index_file = _index_path(store, dir_hash)
|
||
|
||
# Stage the current tree so the name-only diff sees new files too.
|
||
_run_git(["add", "-A"], store, abs_dir,
|
||
timeout=_GIT_TIMEOUT * 2, index_file=index_file)
|
||
ok, names_out, err = _run_git(
|
||
["diff", "--name-only", "-z", commit_hash, "--cached"],
|
||
store, abs_dir, index_file=index_file,
|
||
)
|
||
# Reset the index back to the project ref so it doesn't drift.
|
||
_run_git(["read-tree", _ref_name(dir_hash)], store, abs_dir,
|
||
index_file=index_file, allowed_returncodes={128})
|
||
if not ok:
|
||
return {"success": False, "error": f"Could not compute changed files: {err}"}
|
||
|
||
# Read the same marker-walked project key as record_agent_write.
|
||
ledger = _load_ledger(store, self._ledger_key(abs_dir))
|
||
if not ledger:
|
||
# No agent-write ledger yet (pre-existing store, or Hermes has
|
||
# not written any files here since the ledger was introduced).
|
||
# Signal callers to fall back to a full restore rather than
|
||
# skipping every file.
|
||
return {"success": True, "restore": [], "skipped": [],
|
||
"ledger_empty": True}
|
||
restore: List[str] = []
|
||
skipped: List[str] = []
|
||
for rel in filter(None, names_out.split("\x00")):
|
||
abs_path = Path(abs_dir) / rel
|
||
entry = ledger.get(str(abs_path))
|
||
recorded = entry.get("sha256") if isinstance(entry, dict) else None
|
||
if recorded is None:
|
||
# Hermes never wrote this file (or the ledger predates it) —
|
||
# do not touch it in safe mode.
|
||
skipped.append(rel)
|
||
continue
|
||
current = _hash_file(abs_path)
|
||
if current is None:
|
||
# File deleted since Hermes wrote it: restoring it back is
|
||
# safe — its last content was Hermes-authored.
|
||
restore.append(rel)
|
||
elif current == recorded:
|
||
restore.append(rel)
|
||
else:
|
||
skipped.append(rel)
|
||
return {"success": True, "restore": restore, "skipped": skipped}
|
||
|
||
def ensure_checkpoint(self, working_dir: str, reason: str = "auto") -> bool:
|
||
"""Take a checkpoint if enabled and not already done this turn.
|
||
|
||
Returns True if a checkpoint was taken, False otherwise.
|
||
Never raises — all errors are silently logged.
|
||
"""
|
||
if not self.enabled:
|
||
return False
|
||
|
||
abs_dir = str(_normalize_path(working_dir))
|
||
|
||
# Skip root, home, and other overly broad directories
|
||
if abs_dir in {"/", str(Path.home())}:
|
||
logger.debug("Checkpoint skipped: directory too broad (%s)", abs_dir)
|
||
return False
|
||
|
||
if abs_dir in self._checkpointed_dirs:
|
||
return False
|
||
|
||
self._checkpointed_dirs.add(abs_dir)
|
||
|
||
try:
|
||
from tools.checkpoint_pruning import store_lock
|
||
|
||
with store_lock(_resolve_checkpoint_base()):
|
||
return self._take(abs_dir, reason)
|
||
except Exception as e:
|
||
logger.debug("Checkpoint failed (non-fatal): %s", e)
|
||
return False
|
||
|
||
def list_checkpoints(self, working_dir: str) -> List[Dict]:
|
||
"""List available checkpoints for a directory (most recent first)."""
|
||
abs_dir = str(_normalize_path(working_dir))
|
||
store = _store_path()
|
||
|
||
if not (store / "HEAD").exists():
|
||
return []
|
||
|
||
ref = _ref_name(_project_hash(abs_dir))
|
||
ok, stdout, _ = _run_git(
|
||
["log", ref, "--format=%H|%h|%aI|%s", "-n", str(self.max_snapshots)],
|
||
store, abs_dir,
|
||
allowed_returncodes={128, 129},
|
||
)
|
||
|
||
if not ok or not stdout:
|
||
return []
|
||
|
||
results: List[Dict] = []
|
||
for line in stdout.splitlines():
|
||
parts = line.split("|", 3)
|
||
if len(parts) == 4:
|
||
entry = {
|
||
"hash": parts[0],
|
||
"short_hash": parts[1],
|
||
"timestamp": parts[2],
|
||
"reason": parts[3],
|
||
"files_changed": 0,
|
||
"insertions": 0,
|
||
"deletions": 0,
|
||
}
|
||
stat_ok, stat_out, _ = _run_git(
|
||
["diff", "--shortstat", f"{parts[0]}~1", parts[0]],
|
||
store, abs_dir,
|
||
allowed_returncodes={128, 129},
|
||
)
|
||
if stat_ok and stat_out:
|
||
self._parse_shortstat(stat_out, entry)
|
||
results.append(entry)
|
||
return results
|
||
|
||
def list_all_checkpoints(self) -> List[Dict]:
|
||
"""List checkpoints across every registered project (most recent first).
|
||
|
||
Surgical reapply of PR #10633 by @nightq (#10505) onto the v2
|
||
single-store layout: iterate ``projects/<hash>.json`` metadata via
|
||
``_list_projects`` instead of the pre-v2 per-shadow-dir scan. Each
|
||
entry carries the extra ``workdir`` key so callers can label which
|
||
project a checkpoint belongs to.
|
||
"""
|
||
store = _store_path()
|
||
if not (store / "HEAD").exists():
|
||
return []
|
||
results: List[Dict] = []
|
||
for meta in _list_projects(store):
|
||
workdir = meta.get("workdir") or ""
|
||
if not workdir:
|
||
continue
|
||
for entry in self.list_checkpoints(workdir):
|
||
entry["workdir"] = workdir
|
||
results.append(entry)
|
||
results.sort(key=lambda x: x.get("timestamp", ""), reverse=True)
|
||
return results
|
||
|
||
@staticmethod
|
||
def _parse_shortstat(stat_line: str, entry: Dict) -> None:
|
||
"""Parse git --shortstat output into entry dict."""
|
||
m = re.search(r'(\d+) file', stat_line)
|
||
if m:
|
||
entry["files_changed"] = int(m.group(1))
|
||
m = re.search(r'(\d+) insertion', stat_line)
|
||
if m:
|
||
entry["insertions"] = int(m.group(1))
|
||
m = re.search(r'(\d+) deletion', stat_line)
|
||
if m:
|
||
entry["deletions"] = int(m.group(1))
|
||
|
||
def diff(self, working_dir: str, commit_hash: str) -> Dict:
|
||
"""Show diff between a checkpoint and the current working tree."""
|
||
from tools.checkpoint_pruning import PruneError, store_lock
|
||
|
||
try:
|
||
with store_lock(_resolve_checkpoint_base()):
|
||
return self._diff(working_dir, commit_hash)
|
||
except (PruneError, OSError) as exc:
|
||
return {"success": False, "error": str(exc)}
|
||
|
||
def _diff(self, working_dir: str, commit_hash: str) -> Dict:
|
||
hash_err = _validate_commit_hash(commit_hash)
|
||
if hash_err:
|
||
return {"success": False, "error": hash_err}
|
||
|
||
abs_dir = str(_normalize_path(working_dir))
|
||
store = _store_path()
|
||
|
||
if not (store / "HEAD").exists():
|
||
return {"success": False, "error": "No checkpoints exist for this directory"}
|
||
|
||
ok, _, err = _run_git(
|
||
["cat-file", "-t", commit_hash], store, abs_dir,
|
||
)
|
||
if not ok:
|
||
return {"success": False, "error": f"Checkpoint '{commit_hash}' not found"}
|
||
|
||
dir_hash = _project_hash(abs_dir)
|
||
index_file = _index_path(store, dir_hash)
|
||
|
||
# Stage current state into the per-project index to compare.
|
||
_run_git(["add", "-A"], store, abs_dir,
|
||
timeout=_GIT_TIMEOUT * 2, index_file=index_file)
|
||
|
||
ok_stat, stat_out, _ = _run_git(
|
||
["diff", "--stat", commit_hash, "--cached"],
|
||
store, abs_dir, index_file=index_file,
|
||
)
|
||
ok_diff, diff_out, _ = _run_git(
|
||
["diff", commit_hash, "--cached", "--no-color"],
|
||
store, abs_dir, index_file=index_file,
|
||
)
|
||
|
||
# Reset staged tree back to the project's last checkpoint so the
|
||
# index doesn't drift out of sync with the ref.
|
||
ref = _ref_name(dir_hash)
|
||
_run_git(["read-tree", ref], store, abs_dir,
|
||
index_file=index_file,
|
||
allowed_returncodes={128})
|
||
|
||
if not ok_stat and not ok_diff:
|
||
return {"success": False, "error": "Could not generate diff"}
|
||
|
||
return {
|
||
"success": True,
|
||
"stat": stat_out if ok_stat else "",
|
||
"diff": diff_out if ok_diff else "",
|
||
}
|
||
|
||
def session_diff(self, working_dir: str) -> Dict:
|
||
"""Show the cumulative diff of everything changed in this directory.
|
||
|
||
This powers ``/diff session``. It answers "what has Hermes changed
|
||
here?" by diffing the *earliest retained checkpoint* — the snapshot
|
||
taken before the first recorded edit — against the current working
|
||
tree. Because checkpoints are captured just before each file-mutating
|
||
tool call, that baseline is the pre-edit state, so the diff covers the
|
||
first edit and everything after it.
|
||
|
||
Note: checkpoints are a persistent per-project ref, so the earliest
|
||
*retained* checkpoint may predate the current session (or, after
|
||
pruning, postdate its true start). It is an approximation of "what
|
||
Hermes changed", not an exact per-session ledger.
|
||
|
||
Returns the same shape as :meth:`diff` (``{"success", "stat",
|
||
"diff"}``). When no checkpoints exist yet — nothing has been edited —
|
||
the call still *succeeds* with empty output and ``"empty": True`` so
|
||
callers can show a friendly "no changes" message rather than an error.
|
||
"""
|
||
checkpoints = self.list_checkpoints(working_dir)
|
||
if not checkpoints:
|
||
return {"success": True, "stat": "", "diff": "", "empty": True}
|
||
|
||
baseline = checkpoints[-1].get("hash") or ""
|
||
result = self.diff(working_dir, baseline)
|
||
if result.get("success"):
|
||
result.setdefault("baseline", baseline)
|
||
if not result.get("stat") and not result.get("diff"):
|
||
result["empty"] = True
|
||
return result
|
||
|
||
def restore(
|
||
self,
|
||
working_dir: str,
|
||
commit_hash: str,
|
||
file_path: str = None,
|
||
safe: bool = False,
|
||
) -> Dict:
|
||
"""Restore files to a checkpoint state.
|
||
|
||
With ``safe=True`` (full-directory restores only), files the user
|
||
hand-edited after Hermes' last write — per the agent-write ledger —
|
||
are left untouched, and only Hermes-authored changes are reverted.
|
||
The result gains ``skipped_user_edits`` listing the preserved paths,
|
||
``skipped_oversize`` listing paths kept because the size cap excluded
|
||
them from every checkpoint, and — only when a delete failed —
|
||
``failed_deletes`` listing paths that could not be removed.
|
||
"""
|
||
from tools.checkpoint_pruning import PruneError, store_lock
|
||
|
||
try:
|
||
with store_lock(_resolve_checkpoint_base()):
|
||
return self._restore(working_dir, commit_hash, file_path, safe)
|
||
except (PruneError, OSError) as exc:
|
||
return {"success": False, "error": str(exc)}
|
||
|
||
def _restore(self, working_dir: str, commit_hash: str, file_path: str | None, safe: bool) -> Dict:
|
||
hash_err = _validate_commit_hash(commit_hash)
|
||
if hash_err:
|
||
return {"success": False, "error": hash_err}
|
||
|
||
abs_dir = str(_normalize_path(working_dir))
|
||
|
||
if file_path:
|
||
path_err = _validate_file_path(file_path, abs_dir)
|
||
if path_err:
|
||
return {"success": False, "error": path_err}
|
||
|
||
store = _store_path()
|
||
|
||
if not (store / "HEAD").exists():
|
||
return {"success": False, "error": "No checkpoints exist for this directory"}
|
||
|
||
ok, _, err = _run_git(
|
||
["cat-file", "-t", commit_hash], store, abs_dir,
|
||
)
|
||
if not ok:
|
||
return {"success": False, "error": f"Checkpoint '{commit_hash}' not found",
|
||
"debug": err or None}
|
||
|
||
skipped_user_edits: List[str] = []
|
||
kept_oversize: List[str] = []
|
||
failed_deletes: List[str] = []
|
||
restore_paths: Optional[List[str]] = None
|
||
if safe and not file_path:
|
||
plan = self._safe_restore_plan(abs_dir, commit_hash)
|
||
if not plan.get("success"):
|
||
return {"success": False, "error": plan.get("error", "Safe-restore plan failed")}
|
||
if plan.get("ledger_empty"):
|
||
# No agent-write history to compare against — fall back to
|
||
# the classic full restore rather than restoring nothing.
|
||
restore_paths = None
|
||
else:
|
||
restore_paths = plan["restore"]
|
||
skipped_user_edits = plan["skipped"]
|
||
if not restore_paths:
|
||
return {
|
||
"success": True,
|
||
"restored_to": commit_hash[:8],
|
||
"reason": "nothing to restore (all changed files were user-edited)",
|
||
"directory": abs_dir,
|
||
"restored_files": [],
|
||
"skipped_user_edits": skipped_user_edits,
|
||
"skipped_oversize": [],
|
||
}
|
||
|
||
# Take a pre-rollback snapshot so you can undo the undo.
|
||
self._take(abs_dir, f"pre-rollback snapshot (restoring to {commit_hash[:8]})", prune=False)
|
||
|
||
dir_hash = _project_hash(abs_dir)
|
||
index_file = _index_path(store, dir_hash)
|
||
|
||
if restore_paths is not None:
|
||
# Split into files present in the checkpoint (checkout) and
|
||
# Hermes-created files absent from it (delete to restore state).
|
||
checkout_targets: List[str] = []
|
||
delete_targets: List[str] = []
|
||
for rel in restore_paths:
|
||
ok_in_commit, _, _ = _run_git(
|
||
["cat-file", "-e", f"{commit_hash}:{rel}"],
|
||
store, abs_dir, allowed_returncodes={1, 128},
|
||
)
|
||
if ok_in_commit:
|
||
checkout_targets.append(rel)
|
||
elif self._exceeds_size_cap(Path(abs_dir) / rel):
|
||
# Absent from the checkpoint because ``max_file_size_mb``
|
||
# kept it out (_drop_oversize_from_index), not because
|
||
# Hermes created it. Deleting it would not restore a prior
|
||
# state — no checkpoint holds one — it would destroy the
|
||
# only copy. The ledger records a content hash, not whether
|
||
# a write created or modified the file, so an oversize path
|
||
# cannot be proven agent-created; leaving it costs a stale
|
||
# file, deleting it costs the file.
|
||
kept_oversize.append(rel)
|
||
else:
|
||
delete_targets.append(rel)
|
||
for rel in delete_targets:
|
||
try:
|
||
target = Path(abs_dir) / rel
|
||
if target.is_file() or target.is_symlink():
|
||
target.unlink()
|
||
except OSError as exc:
|
||
logger.warning(
|
||
"Safe restore: could not remove %s: %s", rel, exc,
|
||
)
|
||
failed_deletes.append(rel)
|
||
if not checkout_targets:
|
||
ok, stdout, err = True, "", ""
|
||
else:
|
||
ok, stdout, err = _run_git(
|
||
["checkout", commit_hash, "--", *checkout_targets],
|
||
store, abs_dir, timeout=_GIT_TIMEOUT * 2,
|
||
index_file=index_file,
|
||
)
|
||
else:
|
||
ok, stdout, err = _run_git(
|
||
["checkout", commit_hash, "--", file_path if file_path else "."],
|
||
store, abs_dir, timeout=_GIT_TIMEOUT * 2,
|
||
index_file=index_file,
|
||
)
|
||
|
||
if not ok:
|
||
return {"success": False, "error": f"Restore failed: {err}",
|
||
"debug": err or None}
|
||
|
||
ok2, reason_out, _ = _run_git(
|
||
["log", "--format=%s", "-1", commit_hash], store, abs_dir,
|
||
)
|
||
reason = reason_out if ok2 else "unknown"
|
||
|
||
result = {
|
||
"success": True,
|
||
"restored_to": commit_hash[:8],
|
||
"reason": reason,
|
||
"directory": abs_dir,
|
||
}
|
||
if file_path:
|
||
result["file"] = file_path
|
||
if restore_paths is not None:
|
||
# Only what was actually acted on. A kept oversize path was not
|
||
# restored (and a failed unlink left the file in place), and
|
||
# reporting either as restored is how the data loss above stayed
|
||
# silent: the user was told "Restored" for a file that had just
|
||
# been unlinked.
|
||
not_restored = set(kept_oversize) | set(failed_deletes)
|
||
result["restored_files"] = [
|
||
rel for rel in restore_paths if rel not in not_restored
|
||
]
|
||
result["skipped_user_edits"] = skipped_user_edits
|
||
result["skipped_oversize"] = kept_oversize
|
||
if failed_deletes:
|
||
result["failed_deletes"] = failed_deletes
|
||
# The selected tree was needed until checkout completed. Only now may
|
||
# the safety snapshot's count/size budget make that tree unreachable.
|
||
self._prune(store, abs_dir, _ref_name(dir_hash))
|
||
return result
|
||
|
||
def _ledger_key(self, path: str) -> str:
|
||
"""Agent-write ledger key: hash of the marker-walked project dir, for writer and reader alike."""
|
||
return _project_hash(self.get_working_dir_for_path(path))
|
||
|
||
def get_working_dir_for_path(self, file_path: str) -> str:
|
||
"""Resolve a file path to its working directory for checkpointing."""
|
||
path = _normalize_path(file_path)
|
||
if path.is_dir():
|
||
candidate = path
|
||
else:
|
||
candidate = path.parent
|
||
|
||
# An explicitly checkpointed root owns its ledger even when an
|
||
# ancestor carries a project marker. Prefer the nearest owner.
|
||
roots = [Path(meta["workdir"]) for meta in _list_projects(_store_path()) if meta.get("workdir")]
|
||
owners = [root for root in roots if candidate.is_relative_to(root)]
|
||
if owners:
|
||
return str(max(owners, key=lambda root: len(root.parts)))
|
||
markers = {".git", "pyproject.toml", "package.json", "Cargo.toml",
|
||
"go.mod", "Makefile", "pom.xml", ".hg", "Gemfile"}
|
||
home = Path.home().resolve()
|
||
broad = {home, *home.parents}
|
||
check = candidate
|
||
while check != check.parent and check not in broad:
|
||
if any((check / m).exists() for m in markers):
|
||
return str(check)
|
||
check = check.parent
|
||
|
||
return str(candidate)
|
||
|
||
# ------------------------------------------------------------------
|
||
# Internal
|
||
# ------------------------------------------------------------------
|
||
|
||
def _take(self, working_dir: str, reason: str, *, prune: bool = True) -> bool:
|
||
"""Take a snapshot. Returns True on success."""
|
||
store = _store_path()
|
||
|
||
err = _init_store(store, working_dir)
|
||
if err:
|
||
logger.debug("Checkpoint store init failed: %s", err)
|
||
return False
|
||
|
||
_touch_project(store, working_dir)
|
||
|
||
# Quick size guard — don't try to snapshot enormous directories
|
||
if _dir_file_count(working_dir) > _MAX_FILES:
|
||
logger.debug("Checkpoint skipped: >%d files in %s", _MAX_FILES, working_dir)
|
||
return False
|
||
|
||
dir_hash = _project_hash(working_dir)
|
||
index_file = _index_path(store, dir_hash)
|
||
ref = _ref_name(dir_hash)
|
||
|
||
# Seed the per-project index from the last checkpoint, if any, so the
|
||
# diff/commit machinery sees only changes since then. On first call,
|
||
# clear the index so ``git add -A`` produces a clean tree.
|
||
if index_file.exists():
|
||
# Reset index to current ref tip to avoid accumulating stale paths.
|
||
ok_ref, ref_commit, _ = _run_git(
|
||
["rev-parse", "--verify", ref + "^{commit}"],
|
||
store, working_dir,
|
||
allowed_returncodes={128},
|
||
)
|
||
if ok_ref and ref_commit:
|
||
_run_git(
|
||
["read-tree", ref_commit],
|
||
store, working_dir,
|
||
index_file=index_file,
|
||
allowed_returncodes={128},
|
||
)
|
||
else:
|
||
try:
|
||
index_file.unlink()
|
||
except OSError:
|
||
pass
|
||
else:
|
||
# First snapshot for this project.
|
||
index_file.parent.mkdir(parents=True, exist_ok=True)
|
||
|
||
# Stage with per-project index. Include a per-stage file-size filter
|
||
# via ``core.bigFileThreshold`` is not what we want — instead, we
|
||
# rely on the exclude file for broad patterns and post-stage prune
|
||
# any path whose size exceeds max_file_size_mb.
|
||
ok, _, err = _run_git(
|
||
["add", "-A"], store, working_dir,
|
||
timeout=_GIT_TIMEOUT * 2, index_file=index_file,
|
||
)
|
||
if not ok:
|
||
logger.debug("Checkpoint git-add failed: %s", err)
|
||
return False
|
||
|
||
if self.max_file_size_mb > 0:
|
||
self._drop_oversize_from_index(store, working_dir, index_file)
|
||
|
||
# Compare against the current ref tip (not HEAD — HEAD points to a
|
||
# branch that doesn't exist on a bare store, so ``diff --cached``
|
||
# against HEAD would always show "new file" for every staged path).
|
||
ok_ref, ref_commit, _ = _run_git(
|
||
["rev-parse", "--verify", ref + "^{commit}"],
|
||
store, working_dir,
|
||
allowed_returncodes={128},
|
||
)
|
||
has_ref = ok_ref and bool(ref_commit)
|
||
|
||
if has_ref:
|
||
ok_diff, _, _ = _run_git(
|
||
["diff-index", "--cached", "--quiet", ref_commit],
|
||
store, working_dir,
|
||
allowed_returncodes={1},
|
||
index_file=index_file,
|
||
)
|
||
if ok_diff:
|
||
logger.debug("Checkpoint skipped: no changes in %s", working_dir)
|
||
return False
|
||
else:
|
||
# No ref yet — skip only if the index is empty.
|
||
ok_ls, ls_out, _ = _run_git(
|
||
["ls-files", "--cached"],
|
||
store, working_dir,
|
||
index_file=index_file,
|
||
)
|
||
if ok_ls and not ls_out.strip():
|
||
logger.debug("Checkpoint skipped: empty tree in %s", working_dir)
|
||
return False
|
||
|
||
# Write tree from per-project index.
|
||
ok_tree, tree_sha, err = _run_git(
|
||
["write-tree"], store, working_dir,
|
||
index_file=index_file,
|
||
)
|
||
if not ok_tree or not tree_sha:
|
||
logger.debug("Checkpoint write-tree failed: %s", err)
|
||
return False
|
||
|
||
# Build commit (parent = current ref tip, if any).
|
||
commit_args = ["commit-tree", tree_sha, "-m", reason, "--no-gpg-sign"]
|
||
if has_ref:
|
||
commit_args = ["commit-tree", tree_sha, "-p", ref_commit, "-m", reason, "--no-gpg-sign"]
|
||
ok_commit, new_sha, err = _run_git(
|
||
commit_args, store, working_dir,
|
||
index_file=index_file,
|
||
)
|
||
if not ok_commit or not new_sha:
|
||
logger.debug("Checkpoint commit-tree failed: %s", err)
|
||
return False
|
||
|
||
# Update the per-project ref.
|
||
update_args = ["update-ref", ref, new_sha]
|
||
if has_ref:
|
||
update_args = ["update-ref", ref, new_sha, ref_commit]
|
||
ok_update, _, err = _run_git(
|
||
update_args, store, working_dir,
|
||
)
|
||
if not ok_update:
|
||
logger.debug("Checkpoint update-ref failed: %s", err)
|
||
return False
|
||
|
||
logger.debug("Checkpoint taken in %s: %s (%s)", working_dir, reason, new_sha[:8])
|
||
|
||
# Count and size budgets share one failure boundary.
|
||
if prune:
|
||
self._prune(store, working_dir, ref)
|
||
|
||
return True
|
||
|
||
def _exceeds_size_cap(self, path: Path) -> bool:
|
||
"""Whether *path* is larger than ``max_file_size_mb``.
|
||
|
||
The same test :meth:`_drop_oversize_from_index` applies when building a
|
||
checkpoint, so "excluded from the checkpoint" and "refused deletion at
|
||
restore" agree on one definition. A cap of 0 disables it, and an
|
||
unstattable path is not claimed to be oversize.
|
||
"""
|
||
cap = self.max_file_size_mb * 1024 * 1024
|
||
if cap <= 0:
|
||
return False
|
||
try:
|
||
return path.stat().st_size > cap
|
||
except OSError:
|
||
return False
|
||
|
||
def _drop_oversize_from_index(
|
||
self, store: Path, working_dir: str, index_file: Path,
|
||
) -> None:
|
||
"""Remove any staged file larger than ``max_file_size_mb`` from the index.
|
||
|
||
Lets the agent keep snapshotting source code while refusing to
|
||
swallow generated assets (datasets, model weights, logs, videos).
|
||
"""
|
||
if self.max_file_size_mb <= 0:
|
||
return
|
||
ok, stdout, _ = _run_git(
|
||
["ls-files", "--cached", "-z"],
|
||
store, working_dir, index_file=index_file,
|
||
)
|
||
if not ok or not stdout:
|
||
return
|
||
# NUL separators preserve whitespace within each literal filename.
|
||
paths = [p for p in stdout.split("\x00") if p]
|
||
abs_workdir = _normalize_path(working_dir)
|
||
# Same predicate safe restore consults, called rather than restated:
|
||
# a threshold that drifted between the two would make a file both
|
||
# absent from the checkpoint and not recognised as capped at restore,
|
||
# which is precisely the deletion this change exists to prevent.
|
||
oversize = [
|
||
rel for rel in paths if self._exceeds_size_cap(abs_workdir / rel)
|
||
]
|
||
if not oversize:
|
||
return
|
||
logger.debug(
|
||
"Checkpoint: dropping %d oversize file(s) (>%d MB) from index",
|
||
len(oversize), self.max_file_size_mb,
|
||
)
|
||
# Use --pathspec-from-file for safety with many paths.
|
||
# Chunk into manageable batches.
|
||
BATCH = 200
|
||
for i in range(0, len(oversize), BATCH):
|
||
chunk = oversize[i:i + BATCH]
|
||
_run_git(
|
||
["rm", "--cached", "--quiet", "--"] + chunk,
|
||
store, working_dir, index_file=index_file,
|
||
allowed_returncodes={128},
|
||
)
|
||
|
||
def _prune(self, store: Path, working_dir: str, ref: str) -> None:
|
||
"""Checkpoint-take path: snapshot-count budget plus one size round, gc deferred to the
|
||
periodic prune — a repack here held the tool call for the whole gc on a large store."""
|
||
from tools.checkpoint_pruning import Pruner, PruneError
|
||
|
||
pruner = Pruner(_run_git, store, working_dir, _GIT_TIMEOUT, _dir_size_bytes, _REFS_PREFIX)
|
||
try:
|
||
pruner.trim(ref, self.max_snapshots)
|
||
if pruner.drop_one_round(self.max_total_size_mb * 1024 * 1024):
|
||
logger.info("Checkpoint store exceeded %d MB — dropped the oldest snapshot per project; "
|
||
"space is reclaimed by the next prune", self.max_total_size_mb)
|
||
except (PruneError, OSError) as exc:
|
||
logger.warning("Checkpoint pruning stopped: %s", exc)
|
||
|
||
|
||
def format_checkpoint_list(checkpoints: List[Dict], directory: str) -> str:
|
||
"""Format checkpoint list for display to user."""
|
||
if not checkpoints:
|
||
return f"No checkpoints found for {directory}"
|
||
|
||
lines = [f"📸 Checkpoints for {directory}:\n"]
|
||
for i, cp in enumerate(checkpoints, 1):
|
||
ts = cp["timestamp"]
|
||
if "T" in ts:
|
||
ts = ts.split("T")[1].split("+")[0].split("-")[0][:5]
|
||
date = cp["timestamp"].split("T")[0]
|
||
ts = f"{date} {ts}"
|
||
|
||
files = cp.get("files_changed", 0)
|
||
ins = cp.get("insertions", 0)
|
||
dele = cp.get("deletions", 0)
|
||
if files:
|
||
stat = f" ({files} file{'s' if files != 1 else ''}, +{ins}/-{dele})"
|
||
else:
|
||
stat = ""
|
||
|
||
# Label per-project entries when showing the cross-project view
|
||
# (workdir key only present on list_all_checkpoints results).
|
||
workdir = cp.get("workdir", "")
|
||
if workdir and directory == "all directories":
|
||
workdir_short = Path(workdir).name or workdir
|
||
lines.append(
|
||
f" {i}. {cp['short_hash']} {ts} [{workdir_short}] {cp['reason']}{stat}"
|
||
)
|
||
else:
|
||
lines.append(f" {i}. {cp['short_hash']} {ts} {cp['reason']}{stat}")
|
||
|
||
lines.append("\n /rollback <N> restore to checkpoint N")
|
||
lines.append(" /rollback diff <N> preview changes since checkpoint N")
|
||
lines.append(" /rollback <N> <file> restore a single file from checkpoint N")
|
||
return "\n".join(lines)
|