| 1 | """Git-backed version control for memory files, using dulwich.""" |
| 2 | |
| 3 | from __future__ import annotations |
| 4 | |
| 5 | import io |
| 6 | import time |
| 7 | from dataclasses import dataclass |
| 8 | from datetime import datetime, timezone |
| 9 | from pathlib import Path |
| 10 | |
| 11 | from loguru import logger |
| 12 | |
| 13 | |
| 14 | @dataclass |
| 15 | class CommitInfo: |
| 16 | sha: str # Short SHA (8 chars) |
| 17 | message: str |
| 18 | timestamp: str # Formatted datetime |
| 19 | |
| 20 | def format(self, diff: str = "") -> str: |
| 21 | """Format this commit for display, optionally with a diff.""" |
| 22 | header = f"## {self.message.splitlines()[0]}\n`{self.sha}` — {self.timestamp}\n" |
| 23 | if diff: |
| 24 | return f"{header}\n```diff\n{diff}\n```" |
| 25 | return f"{header}\n(no file changes)" |
| 26 | |
| 27 | |
| 28 | @dataclass |
| 29 | class LineAge: |
| 30 | """Age of a single line based on git blame.""" |
| 31 | |
| 32 | age_days: int # days since last modification |
| 33 | |
| 34 | |
| 35 | def _compute_line_ages(annotated) -> list[LineAge]: |
| 36 | """Convert annotate results to per-line ages.""" |
| 37 | now = datetime.now(tz=timezone.utc).date() |
| 38 | ages: list[LineAge] = [] |
| 39 | for (commit, _tree_entry), _line_bytes in annotated: |
| 40 | dt = datetime.fromtimestamp(commit.commit_time, tz=timezone.utc).date() |
| 41 | ages.append(LineAge(age_days=(now - dt).days)) |
| 42 | return ages |
| 43 | |
| 44 | |
| 45 | class GitStore: |
| 46 | """Git-backed version control for memory files.""" |
| 47 | |
| 48 | def __init__(self, workspace: Path, tracked_files: list[str]): |
| 49 | self._workspace = workspace |
| 50 | self._tracked_files = tracked_files |
| 51 | |
| 52 | def is_initialized(self) -> bool: |
| 53 | """Check if the git repo has been initialized.""" |
| 54 | return (self._workspace / ".git").is_dir() |
| 55 | |
| 56 | # -- init ------------------------------------------------------------------ |
| 57 | |
| 58 | def init(self) -> bool: |
| 59 | """Initialize a git repo if not already initialized. |
| 60 | |
| 61 | Creates .gitignore and makes an initial commit. |
| 62 | Returns True if a new repo was created, False if already exists. |
| 63 | """ |
| 64 | if self.is_initialized(): |
| 65 | return False |
| 66 | |
| 67 | if self._is_inside_git_repo(): |
| 68 | logger.warning( |
| 69 | "Workspace {} is already inside a git repo; " |
| 70 | "skipping nested repo initialization", |
| 71 | self._workspace, |
| 72 | ) |
| 73 | return False |
| 74 | |
| 75 | try: |
| 76 | from dulwich import porcelain |
| 77 | |
| 78 | porcelain.init(str(self._workspace)) |
| 79 | |
| 80 | # Write .gitignore (merge with existing if present) |
| 81 | gitignore = self._workspace / ".gitignore" |
| 82 | dream_entries = self._build_gitignore() |
| 83 | if gitignore.exists(): |
| 84 | existing = gitignore.read_text(encoding="utf-8") |
| 85 | existing_lines = set(existing.splitlines()) |
| 86 | new_lines = [ |
| 87 | line |
| 88 | for line in dream_entries.splitlines() |
| 89 | if line not in existing_lines |
| 90 | ] |
| 91 | if new_lines: |
| 92 | merged = existing.rstrip("\n") + "\n" + "\n".join(new_lines) + "\n" |
| 93 | gitignore.write_text(merged, encoding="utf-8") |
| 94 | else: |
| 95 | gitignore.write_text(dream_entries, encoding="utf-8") |
| 96 | |
| 97 | # Ensure tracked files exist (touch them if missing) so the initial |
| 98 | # commit has something to track. |
| 99 | for rel in self._tracked_files: |
| 100 | p = self._workspace / rel |
| 101 | p.parent.mkdir(parents=True, exist_ok=True) |
| 102 | if not p.exists(): |
| 103 | p.write_text("", encoding="utf-8") |
| 104 | |
| 105 | # Initial commit |
| 106 | porcelain.add(str(self._workspace), paths=[".gitignore"] + self._tracked_files) |
| 107 | porcelain.commit( |
| 108 | str(self._workspace), |
| 109 | message=b"init: nanobot memory store", |
| 110 | author=b"nanobot <nanobot@dream>", |
| 111 | committer=b"nanobot <nanobot@dream>", |
| 112 | ) |
| 113 | logger.info("Git store initialized at {}", self._workspace) |
| 114 | return True |
| 115 | except Exception: |
| 116 | logger.warning("Git store init failed for {}", self._workspace) |
| 117 | return False |
| 118 | |
| 119 | # -- daily operations ------------------------------------------------------ |
| 120 | |
| 121 | def auto_commit(self, message: str) -> str | None: |
| 122 | """Stage tracked memory files and commit if there are changes. |
| 123 | |
| 124 | Returns the short commit SHA, or None if nothing to commit. |
| 125 | """ |
| 126 | if not self.is_initialized(): |
| 127 | return None |
| 128 | |
| 129 | try: |
| 130 | from dulwich import porcelain |
| 131 | |
| 132 | # .gitignore excludes everything except tracked files, |
| 133 | # so any staged/unstaged change must be in our files. |
| 134 | st = porcelain.status(str(self._workspace)) |
| 135 | if not st.unstaged and not any(st.staged.values()): |
| 136 | return None |
| 137 | |
| 138 | msg_bytes = message.encode("utf-8") if isinstance(message, str) else message |
| 139 | porcelain.add(str(self._workspace), paths=self._tracked_files) |
| 140 | sha_bytes = porcelain.commit( |
| 141 | str(self._workspace), |
| 142 | message=msg_bytes, |
| 143 | author=b"nanobot <nanobot@dream>", |
| 144 | committer=b"nanobot <nanobot@dream>", |
| 145 | ) |
| 146 | if sha_bytes is None: |
| 147 | return None |
| 148 | sha = sha_bytes.hex()[:8] |
| 149 | logger.debug("Git auto-commit: {} ({})", sha, message) |
| 150 | return sha |
| 151 | except Exception: |
| 152 | logger.warning("Git auto-commit failed: {}", message) |
| 153 | return None |
| 154 | |
| 155 | # -- internal helpers ------------------------------------------------------ |
| 156 | |
| 157 | def _resolve_sha(self, short_sha: str) -> bytes | None: |
| 158 | """Resolve a short SHA prefix to the full SHA bytes.""" |
| 159 | try: |
| 160 | from dulwich.repo import Repo |
| 161 | |
| 162 | with Repo(str(self._workspace)) as repo: |
| 163 | try: |
| 164 | sha = repo.refs[b"HEAD"] |
| 165 | except KeyError: |
| 166 | return None |
| 167 | |
| 168 | while sha: |
| 169 | if sha.hex().startswith(short_sha): |
| 170 | return sha |
| 171 | commit = repo[sha] |
| 172 | if commit.type_name != b"commit": |
| 173 | break |
| 174 | sha = commit.parents[0] if commit.parents else None |
| 175 | return None |
| 176 | except Exception: |
| 177 | return None |
| 178 | |
| 179 | def _is_inside_git_repo(self) -> bool: |
| 180 | """Check if self._workspace is already inside a git repository. |
| 181 | |
| 182 | Walks up from self._workspace to the filesystem root, returning True |
| 183 | if any parent directory contains a .git entry. |
| 184 | |
| 185 | Git worktrees and submodules can use a ``.git`` file instead of a |
| 186 | directory, so we must treat either form as "already inside a repo". |
| 187 | """ |
| 188 | current = self._workspace.resolve() |
| 189 | while current != current.parent: |
| 190 | if (current / ".git").exists(): |
| 191 | return True |
| 192 | current = current.parent |
| 193 | return False |
| 194 | |
| 195 | def _build_gitignore(self) -> str: |
| 196 | """Generate .gitignore content from tracked files.""" |
| 197 | dirs: set[str] = set() |
| 198 | for f in self._tracked_files: |
| 199 | parent = str(Path(f).parent) |
| 200 | if parent != ".": |
| 201 | dirs.add(parent) |
| 202 | lines = ["/*"] |
| 203 | for d in sorted(dirs): |
| 204 | lines.append(f"!{d}/") |
| 205 | for f in self._tracked_files: |
| 206 | lines.append(f"!{f}") |
| 207 | lines.append("!.gitignore") |
| 208 | return "\n".join(lines) + "\n" |
| 209 | |
| 210 | # -- query ----------------------------------------------------------------- |
| 211 | |
| 212 | def log(self, max_entries: int = 20) -> list[CommitInfo]: |
| 213 | """Return simplified commit log.""" |
| 214 | if not self.is_initialized(): |
| 215 | return [] |
| 216 | |
| 217 | try: |
| 218 | from dulwich.repo import Repo |
| 219 | |
| 220 | entries: list[CommitInfo] = [] |
| 221 | with Repo(str(self._workspace)) as repo: |
| 222 | try: |
| 223 | head = repo.refs[b"HEAD"] |
| 224 | except KeyError: |
| 225 | return [] |
| 226 | |
| 227 | sha = head |
| 228 | while sha and len(entries) < max_entries: |
| 229 | commit = repo[sha] |
| 230 | if commit.type_name != b"commit": |
| 231 | break |
| 232 | ts = time.strftime( |
| 233 | "%Y-%m-%d %H:%M", |
| 234 | time.localtime(commit.commit_time), |
| 235 | ) |
| 236 | msg = commit.message.decode("utf-8", errors="replace").strip() |
| 237 | entries.append(CommitInfo( |
| 238 | sha=sha.hex()[:8], |
| 239 | message=msg, |
| 240 | timestamp=ts, |
| 241 | )) |
| 242 | sha = commit.parents[0] if commit.parents else None |
| 243 | |
| 244 | return entries |
| 245 | except Exception: |
| 246 | logger.warning("Git log failed") |
| 247 | return [] |
| 248 | |
| 249 | def line_ages(self, file_path: str) -> list[LineAge]: |
| 250 | """Compute the age of each line in a tracked file via git blame. |
| 251 | |
| 252 | Returns one LineAge per line, in order. |
| 253 | Returns an empty list if the repo is not initialized, the file is |
| 254 | empty, or annotation fails. |
| 255 | """ |
| 256 | |
| 257 | if not self.is_initialized(): |
| 258 | return [] |
| 259 | |
| 260 | target = self._workspace / file_path |
| 261 | if not target.exists() or target.stat().st_size == 0: |
| 262 | return [] |
| 263 | |
| 264 | try: |
| 265 | from dulwich import porcelain |
| 266 | |
| 267 | annotated = porcelain.annotate(str(self._workspace), file_path) |
| 268 | except Exception: |
| 269 | logger.warning("Git line_ages annotate failed for {}", file_path) |
| 270 | return [] |
| 271 | |
| 272 | if not annotated: |
| 273 | return [] |
| 274 | |
| 275 | return _compute_line_ages(annotated) |
| 276 | |
| 277 | def diff_commits(self, sha1: str, sha2: str) -> str: |
| 278 | """Show diff between two commits.""" |
| 279 | if not self.is_initialized(): |
| 280 | return "" |
| 281 | |
| 282 | try: |
| 283 | from dulwich import porcelain |
| 284 | |
| 285 | full1 = self._resolve_sha(sha1) |
| 286 | full2 = self._resolve_sha(sha2) |
| 287 | if not full1 or not full2: |
| 288 | return "" |
| 289 | |
| 290 | out = io.BytesIO() |
| 291 | porcelain.diff( |
| 292 | str(self._workspace), |
| 293 | commit=full1, |
| 294 | commit2=full2, |
| 295 | outstream=out, |
| 296 | ) |
| 297 | return out.getvalue().decode("utf-8", errors="replace") |
| 298 | except Exception: |
| 299 | logger.warning("Git diff_commits failed") |
| 300 | return "" |
| 301 | |
| 302 | def find_commit(self, short_sha: str, max_entries: int = 20) -> CommitInfo | None: |
| 303 | """Find a commit by short SHA prefix match.""" |
| 304 | for c in self.log(max_entries=max_entries): |
| 305 | if c.sha.startswith(short_sha): |
| 306 | return c |
| 307 | return None |
| 308 | |
| 309 | def show_commit_diff(self, short_sha: str, max_entries: int = 20) -> tuple[CommitInfo, str] | None: |
| 310 | """Find a commit and return it with its diff vs the parent.""" |
| 311 | commits = self.log(max_entries=max_entries) |
| 312 | for i, c in enumerate(commits): |
| 313 | if c.sha.startswith(short_sha): |
| 314 | if i + 1 < len(commits): |
| 315 | diff = self.diff_commits(commits[i + 1].sha, c.sha) |
| 316 | else: |
| 317 | diff = "" |
| 318 | return c, diff |
| 319 | return None |
| 320 | |
| 321 | # -- restore --------------------------------------------------------------- |
| 322 | |
| 323 | def revert(self, commit: str) -> str | None: |
| 324 | """Revert (undo) the changes introduced by the given commit. |
| 325 | |
| 326 | Restores all tracked memory files to the state at the commit's parent, |
| 327 | then creates a new commit recording the revert. |
| 328 | |
| 329 | Returns the new commit SHA, or None on failure. |
| 330 | """ |
| 331 | if not self.is_initialized(): |
| 332 | return None |
| 333 | |
| 334 | try: |
| 335 | from dulwich.repo import Repo |
| 336 | |
| 337 | full_sha = self._resolve_sha(commit) |
| 338 | if not full_sha: |
| 339 | logger.warning("Git revert: SHA not found: {}", commit) |
| 340 | return None |
| 341 | |
| 342 | with Repo(str(self._workspace)) as repo: |
| 343 | commit_obj = repo[full_sha] |
| 344 | if commit_obj.type_name != b"commit": |
| 345 | return None |
| 346 | |
| 347 | if not commit_obj.parents: |
| 348 | logger.warning("Git revert: cannot revert root commit {}", commit) |
| 349 | return None |
| 350 | |
| 351 | # Use the parent's tree — this undoes the commit's changes |
| 352 | parent_obj = repo[commit_obj.parents[0]] |
| 353 | tree = repo[parent_obj.tree] |
| 354 | |
| 355 | restored: list[str] = [] |
| 356 | for filepath in self._tracked_files: |
| 357 | content = self._read_blob_from_tree(repo, tree, filepath) |
| 358 | if content is not None: |
| 359 | dest = self._workspace / filepath |
| 360 | dest.write_text(content, encoding="utf-8") |
| 361 | restored.append(filepath) |
| 362 | |
| 363 | if not restored: |
| 364 | return None |
| 365 | |
| 366 | # Commit the restored state |
| 367 | msg = f"revert: undo {commit}" |
| 368 | return self.auto_commit(msg) |
| 369 | except Exception: |
| 370 | logger.warning("Git revert failed for {}", commit) |
| 371 | return None |
| 372 | |
| 373 | @staticmethod |
| 374 | def _read_blob_from_tree(repo, tree, filepath: str) -> str | None: |
| 375 | """Read a blob's content from a tree object by walking path parts.""" |
| 376 | parts = Path(filepath).parts |
| 377 | current = tree |
| 378 | for part in parts: |
| 379 | try: |
| 380 | entry = current[part.encode()] |
| 381 | except KeyError: |
| 382 | return None |
| 383 | obj = repo[entry[1]] |
| 384 | if obj.type_name == b"blob": |
| 385 | return obj.data.decode("utf-8", errors="replace") |
| 386 | if obj.type_name == b"tree": |
| 387 | current = obj |
| 388 | else: |
| 389 | return None |
| 390 | return None |
| 391 |