| 1 | """Retention for Time Travel shadow repositories. |
| 2 | |
| 3 | Time Travel keeps one hidden git repository per workspace under |
| 4 | ``/a0/usr/.time_travel/workspaces/<workspace_id>/repo.git`` and snapshots it on every file |
| 5 | change. Without retention those repositories accumulate unboundedly: a removed chat or project |
| 6 | leaves its shadow repository orphaned forever (nothing cleans it up), and a workspace whose |
| 7 | ``git add`` ever exceeded ``GIT_TIMEOUT_SECONDS`` strands a ``repo.git/index.lock`` that makes |
| 8 | every later snapshot fail with "index.lock: File exists". |
| 9 | |
| 10 | The sweep (driven from ``job_loop``, throttled by config) removes: |
| 11 | |
| 12 | - ORPHANS — shadow directories whose id matches no live workspace path. Live paths are |
| 13 | forward-enumerated (project folders, the configured workdir, per-chat workdirs) and hashed |
| 14 | with the same ``workspace_id_for`` derivation; anything outside that set has no owner and can |
| 15 | never be shown in the UI again. Deleted once last activity is past a grace window. |
| 16 | - AGED repositories — no snapshot in ``retention_max_age_days`` (0 = keep forever, the |
| 17 | default). |
| 18 | - STALE LOCKS — ``repo.git/index.lock`` older than ``retention_stale_lock_minutes``; Time |
| 19 | Travel kills its git subprocesses at ``GIT_TIMEOUT_SECONDS``, so no legitimate lock lives |
| 20 | that long. Removing it un-wedges future snapshots. |
| 21 | - INVALID BACKUPS — ``repo.git.invalid*`` set-asides made for corrupt repositories, past the |
| 22 | same grace window. |
| 23 | |
| 24 | Deleting a live workspace's shadow repository is always safe for the feature itself: the next |
| 25 | snapshot lazily re-initializes an empty history. Deletion is refused for any path outside the |
| 26 | shadow root. |
| 27 | |
| 28 | Durable state next to the workspaces dir: ``retention.json`` (running totals + last sweep |
| 29 | stamp) and ``retention.log`` (one JSON line per sweep with the names of everything removed, |
| 30 | tail-capped). |
| 31 | """ |
| 32 | |
| 33 | from __future__ import annotations |
| 34 | |
| 35 | import datetime |
| 36 | import json |
| 37 | import os |
| 38 | import shutil |
| 39 | import time |
| 40 | from typing import Any, Optional |
| 41 | |
| 42 | PLUGIN_NAME = "_time_travel" |
| 43 | |
| 44 | MARKER_FILE = "retention.json" |
| 45 | HISTORY_FILE = "retention.log" |
| 46 | HISTORY_MAX_LINES = 1000 |
| 47 | |
| 48 | DEFAULT_CONFIG: dict[str, Any] = { |
| 49 | "retention_enabled": True, |
| 50 | "retention_sweep_interval_hours": 6, |
| 51 | "retention_max_age_days": 0, |
| 52 | "retention_orphan_grace_hours": 24, |
| 53 | "retention_stale_lock_minutes": 30, |
| 54 | } |
| 55 | |
| 56 | |
| 57 | def _int_at_least(value: Any, minimum: int, fallback: int) -> int: |
| 58 | try: |
| 59 | return max(int(value), minimum) |
| 60 | except Exception: |
| 61 | return fallback |
| 62 | |
| 63 | |
| 64 | def effective_config(cfg: Optional[dict[str, Any]] = None) -> dict[str, Any]: |
| 65 | """Plugin config with defaults filled in and values clamped to sane minimums.""" |
| 66 | if cfg is None: |
| 67 | try: |
| 68 | from helpers import plugins |
| 69 | |
| 70 | cfg = plugins.get_plugin_config(PLUGIN_NAME) or {} |
| 71 | except Exception: |
| 72 | cfg = {} |
| 73 | merged = dict(DEFAULT_CONFIG) |
| 74 | merged.update({k: v for k, v in cfg.items() if k in DEFAULT_CONFIG and v is not None}) |
| 75 | merged["retention_enabled"] = bool(merged["retention_enabled"]) |
| 76 | merged["retention_sweep_interval_hours"] = _int_at_least( |
| 77 | merged["retention_sweep_interval_hours"], 1, 6 |
| 78 | ) |
| 79 | merged["retention_max_age_days"] = _int_at_least(merged["retention_max_age_days"], 0, 0) |
| 80 | merged["retention_orphan_grace_hours"] = _int_at_least( |
| 81 | merged["retention_orphan_grace_hours"], 1, 24 |
| 82 | ) |
| 83 | merged["retention_stale_lock_minutes"] = _int_at_least( |
| 84 | merged["retention_stale_lock_minutes"], 5, 30 |
| 85 | ) |
| 86 | return merged |
| 87 | |
| 88 | |
| 89 | def _state_dir() -> str: |
| 90 | from plugins._time_travel.helpers import time_travel |
| 91 | |
| 92 | return str(time_travel.real_path_for_display("/a0/usr/.time_travel")) |
| 93 | |
| 94 | |
| 95 | def _shadow_root() -> str: |
| 96 | from plugins._time_travel.helpers import time_travel |
| 97 | |
| 98 | return str(time_travel.real_path_for_display(time_travel.SHADOW_DISPLAY_ROOT)) |
| 99 | |
| 100 | |
| 101 | def live_workspace_ids() -> set[str]: |
| 102 | """Every workspace id resolvable from a path that exists right now: project folders, the |
| 103 | configured workdir, and per-chat workdirs (custom projects resolvers may mint workspaces |
| 104 | there; including them only makes the sweep more conservative).""" |
| 105 | from plugins._time_travel.helpers import time_travel |
| 106 | |
| 107 | ids: set[str] = set() |
| 108 | |
| 109 | projects_root = time_travel.real_path_for_display("/a0/usr/projects") |
| 110 | try: |
| 111 | for name in os.listdir(projects_root): |
| 112 | if os.path.isdir(os.path.join(projects_root, name)): |
| 113 | ids.add(time_travel.workspace_id_for(f"/a0/usr/projects/{name}")) |
| 114 | except Exception: |
| 115 | pass |
| 116 | |
| 117 | try: |
| 118 | ids.add(time_travel.workspace_id_for(time_travel.configured_workdir_display_path())) |
| 119 | except Exception: |
| 120 | ids.add(time_travel.workspace_id_for("/a0/usr/workdir")) |
| 121 | |
| 122 | chats_root = time_travel.real_path_for_display("/a0/usr/chats") |
| 123 | try: |
| 124 | for name in os.listdir(chats_root): |
| 125 | if os.path.isdir(os.path.join(chats_root, name, "workdir")): |
| 126 | ids.add(time_travel.workspace_id_for(f"/a0/usr/chats/{name}/workdir")) |
| 127 | except Exception: |
| 128 | pass |
| 129 | |
| 130 | return ids |
| 131 | |
| 132 | |
| 133 | def _read_json(path: str) -> dict[str, Any]: |
| 134 | try: |
| 135 | with open(path, "r", encoding="utf-8") as f: |
| 136 | return json.load(f) |
| 137 | except Exception: |
| 138 | return {} |
| 139 | |
| 140 | |
| 141 | def _write_marker(state_dir: str, sweep_stats: dict[str, int], stamp: str) -> None: |
| 142 | try: |
| 143 | os.makedirs(state_dir, exist_ok=True) |
| 144 | path = os.path.join(state_dir, MARKER_FILE) |
| 145 | payload = _read_json(path) |
| 146 | payload["sweeps"] = int(payload.get("sweeps", 0)) + 1 |
| 147 | for key, value in sweep_stats.items(): |
| 148 | payload[key] = int(payload.get(key, 0)) + int(value) |
| 149 | payload["last_sweep_at"] = stamp |
| 150 | tmp = path + ".tmp" |
| 151 | with open(tmp, "w", encoding="utf-8") as f: |
| 152 | json.dump(payload, f, indent=2, sort_keys=True) |
| 153 | os.replace(tmp, path) |
| 154 | except Exception: |
| 155 | pass |
| 156 | |
| 157 | |
| 158 | def _append_history(state_dir: str, entry: dict[str, Any]) -> None: |
| 159 | try: |
| 160 | os.makedirs(state_dir, exist_ok=True) |
| 161 | path = os.path.join(state_dir, HISTORY_FILE) |
| 162 | lines: list[str] = [] |
| 163 | try: |
| 164 | with open(path, "r", encoding="utf-8") as f: |
| 165 | lines = [ln for ln in f.read().splitlines() if ln.strip()] |
| 166 | except Exception: |
| 167 | lines = [] |
| 168 | lines.append(json.dumps(entry, sort_keys=True)) |
| 169 | if len(lines) > HISTORY_MAX_LINES: |
| 170 | lines = lines[-HISTORY_MAX_LINES:] |
| 171 | tmp = path + ".tmp" |
| 172 | with open(tmp, "w", encoding="utf-8") as f: |
| 173 | f.write("\n".join(lines) + "\n") |
| 174 | os.replace(tmp, path) |
| 175 | except Exception: |
| 176 | pass |
| 177 | |
| 178 | |
| 179 | def read_history(limit: int = 50, state_dir: Optional[str] = None) -> list[dict[str, Any]]: |
| 180 | try: |
| 181 | base = state_dir if state_dir is not None else _state_dir() |
| 182 | with open(os.path.join(base, HISTORY_FILE), "r", encoding="utf-8") as f: |
| 183 | lines = [ln for ln in f.read().splitlines() if ln.strip()] |
| 184 | return [json.loads(ln) for ln in lines[-limit:]] |
| 185 | except Exception: |
| 186 | return [] |
| 187 | |
| 188 | |
| 189 | def _last_activity(entry_path: str) -> float: |
| 190 | candidates = [ |
| 191 | os.path.join(entry_path, "repo.git", "refs", "heads", "current"), |
| 192 | os.path.join(entry_path, "repo.git", "packed-refs"), |
| 193 | os.path.join(entry_path, "repo.git", "HEAD"), |
| 194 | os.path.join(entry_path, "repo.git"), |
| 195 | entry_path, |
| 196 | ] |
| 197 | newest = 0.0 |
| 198 | for candidate in candidates: |
| 199 | try: |
| 200 | newest = max(newest, os.stat(candidate).st_mtime) |
| 201 | except Exception: |
| 202 | continue |
| 203 | return newest |
| 204 | |
| 205 | |
| 206 | def _tree_bytes(path: str) -> int: |
| 207 | total = 0 |
| 208 | try: |
| 209 | for root, _dirs, names in os.walk(path): |
| 210 | for name in names: |
| 211 | try: |
| 212 | total += os.stat(os.path.join(root, name)).st_size |
| 213 | except Exception: |
| 214 | pass |
| 215 | except Exception: |
| 216 | pass |
| 217 | return total |
| 218 | |
| 219 | |
| 220 | def _remove_tree(path: str, shadow_root: str) -> int: |
| 221 | """rmtree guarded to the shadow root; returns bytes reclaimed (0 on refusal/failure).""" |
| 222 | real = os.path.realpath(path) |
| 223 | root = os.path.realpath(shadow_root) |
| 224 | if not real.startswith(root + os.sep): |
| 225 | return 0 |
| 226 | size = _tree_bytes(real) |
| 227 | try: |
| 228 | shutil.rmtree(real) |
| 229 | return size |
| 230 | except Exception: |
| 231 | return 0 |
| 232 | |
| 233 | |
| 234 | def sweep( |
| 235 | cfg: Optional[dict[str, Any]] = None, |
| 236 | shadow_root: Optional[str] = None, |
| 237 | live_ids: Optional[set[str]] = None, |
| 238 | now_ts: Optional[float] = None, |
| 239 | state_dir: Optional[str] = None, |
| 240 | ) -> dict[str, int]: |
| 241 | """One retention pass. All inputs are injectable for tests; production callers pass |
| 242 | nothing and everything resolves from the plugin runtime.""" |
| 243 | stats = { |
| 244 | "orphans_removed": 0, |
| 245 | "aged_removed": 0, |
| 246 | "stale_locks_removed": 0, |
| 247 | "invalid_backups_removed": 0, |
| 248 | "bytes_reclaimed": 0, |
| 249 | } |
| 250 | config = effective_config(cfg) |
| 251 | if not config["retention_enabled"]: |
| 252 | return stats |
| 253 | root = shadow_root if shadow_root is not None else _shadow_root() |
| 254 | if not os.path.isdir(root): |
| 255 | return stats |
| 256 | ids = live_ids if live_ids is not None else live_workspace_ids() |
| 257 | base = state_dir if state_dir is not None else _state_dir() |
| 258 | now = time.time() if now_ts is None else now_ts |
| 259 | |
| 260 | max_age_s = config["retention_max_age_days"] * 86400 |
| 261 | grace_s = config["retention_orphan_grace_hours"] * 3600 |
| 262 | lock_s = config["retention_stale_lock_minutes"] * 60 |
| 263 | |
| 264 | detail: dict[str, list[str]] = {"orphans": [], "aged": [], "locks": [], "invalid": []} |
| 265 | |
| 266 | try: |
| 267 | entries = os.listdir(root) |
| 268 | except Exception: |
| 269 | return stats |
| 270 | |
| 271 | for name in entries: |
| 272 | entry = os.path.join(root, name) |
| 273 | if not os.path.isdir(entry): |
| 274 | continue |
| 275 | last = _last_activity(entry) |
| 276 | |
| 277 | if name not in ids: |
| 278 | if now - last > grace_s: |
| 279 | stats["bytes_reclaimed"] += _remove_tree(entry, root) |
| 280 | stats["orphans_removed"] += 1 |
| 281 | detail["orphans"].append(name) |
| 282 | continue |
| 283 | |
| 284 | if max_age_s and now - last > max_age_s: |
| 285 | stats["bytes_reclaimed"] += _remove_tree(entry, root) |
| 286 | stats["aged_removed"] += 1 |
| 287 | detail["aged"].append(name) |
| 288 | continue |
| 289 | |
| 290 | lock = os.path.join(entry, "repo.git", "index.lock") |
| 291 | try: |
| 292 | if os.path.isfile(lock) and now - os.stat(lock).st_mtime > lock_s: |
| 293 | os.remove(lock) |
| 294 | stats["stale_locks_removed"] += 1 |
| 295 | detail["locks"].append(name) |
| 296 | except Exception: |
| 297 | pass |
| 298 | |
| 299 | try: |
| 300 | for sub in os.listdir(entry): |
| 301 | if sub.startswith("repo.git.invalid"): |
| 302 | backup = os.path.join(entry, sub) |
| 303 | if now - os.stat(backup).st_mtime > grace_s: |
| 304 | stats["bytes_reclaimed"] += _remove_tree(backup, root) |
| 305 | stats["invalid_backups_removed"] += 1 |
| 306 | detail["invalid"].append(f"{name}/{sub}") |
| 307 | except Exception: |
| 308 | pass |
| 309 | |
| 310 | stamp = datetime.datetime.now(datetime.timezone.utc).isoformat() |
| 311 | _write_marker(base, stats, stamp) |
| 312 | _append_history(base, {"at": stamp, **stats, "removed": detail}) |
| 313 | return stats |
| 314 | |
| 315 | |
| 316 | def due( |
| 317 | cfg: Optional[dict[str, Any]] = None, |
| 318 | now_ts: Optional[float] = None, |
| 319 | state_dir: Optional[str] = None, |
| 320 | ) -> bool: |
| 321 | """True when retention is enabled and the configured interval has elapsed since the last |
| 322 | sweep (or no sweep ever ran).""" |
| 323 | config = effective_config(cfg) |
| 324 | if not config["retention_enabled"]: |
| 325 | return False |
| 326 | base = state_dir if state_dir is not None else _state_dir() |
| 327 | marker = _read_json(os.path.join(base, MARKER_FILE)) |
| 328 | last = str(marker.get("last_sweep_at") or "") |
| 329 | if not last: |
| 330 | return True |
| 331 | try: |
| 332 | last_dt = datetime.datetime.fromisoformat(last) |
| 333 | now = ( |
| 334 | datetime.datetime.now(datetime.timezone.utc) |
| 335 | if now_ts is None |
| 336 | else datetime.datetime.fromtimestamp(now_ts, datetime.timezone.utc) |
| 337 | ) |
| 338 | interval_s = config["retention_sweep_interval_hours"] * 3600 |
| 339 | return (now - last_dt).total_seconds() >= interval_s |
| 340 | except Exception: |
| 341 | return True |