feat: Time Travel shadow-repo retention (orphan cleanup, optional age-out, stale-lock repair)

Time Travel keeps a hidden git repository per workspace under
/a0/usr/.time_travel/workspaces/<id>/repo.git and snapshots on every file
change, but ships no retention: nothing deletes a shadow repository when its
chat or project is removed (chat_remove never touches .time_travel), there is
no delete/prune endpoint in the API or web UI, and a workspace whose 'git add'
ever exceeds GIT_TIMEOUT_SECONDS strands repo.git/index.lock, failing every
later snapshot with 'index.lock: File exists'. Observed on a live instance:
518 shadow repositories / 12 GB, most belonging to long-deleted chats, plus a
permanently wedged workspace.

This adds a throttled retention sweep driven from job_loop:

- Orphans: live workspace paths (project folders, the configured workdir,
  per-chat workdirs) are forward-enumerated and hashed with the existing
  workspace_id_for derivation; a shadow directory matching none of them is
  unreachable from the UI forever and is removed after a grace window.
- Optional age-out: retention_max_age_days (default 0 = keep forever) removes
  a live workspace's history when it has had no snapshot in N days. Time
  Travel lazily re-initializes an empty history on the next snapshot, so this
  is always safe for the feature.
- Stale locks: repo.git/index.lock older than retention_stale_lock_minutes is
  removed (git subprocesses are killed at GIT_TIMEOUT_SECONDS, so no
  legitimate lock lives that long), un-wedging future snapshots.
- Corrupt-repo set-asides (repo.git.invalid*) past the grace window.

Deletion is refused for any path outside the shadow root. Settings render in
the existing plugin-config pattern (default_config.yaml + webui/config.html,
settings_sections: agent): enable toggle, sweep interval, age-out days,
orphan grace, stale-lock age. Durable evidence lives next to the workspaces
dir: retention.json (running totals + last sweep stamp) and retention.log
(one JSON line per sweep naming everything removed, tail-capped at 1000).

Sweeps run in a worker thread off the event loop, at most one in flight,
default every 6 hours.

Tests: tests/test_time_travel_retention.py (config defaults/clamps, the full
sweep matrix, keep-forever default, disabled no-op, deletion guard, marker
accrual, per-sweep history + cap, throttle, and workspace_id_for parity).

plugins/_time_travel/default_config.yaml

@@ -0,0 +1,5 @@
+retention_enabled: true
+retention_sweep_interval_hours: 6
+retention_max_age_days: 0 # 0 = keep workspace history forever
+retention_orphan_grace_hours: 24
+retention_stale_lock_minutes: 30

plugins/_time_travel/extensions/python/job_loop/_50_retention.py

@@ -0,0 +1,39 @@
+"""Throttled Time Travel retention sweep (see helpers/retention.py)."""
+
+import asyncio
+from typing import Any
+
+from helpers.extension import Extension
+from helpers.print_style import PrintStyle
+
+_in_flight = False
+
+
+class TimeTravelRetention(Extension):
+
+    async def execute(self, **kwargs: Any) -> None:
+        global _in_flight
+        if _in_flight:
+            return
+        try:
+            from plugins._time_travel.helpers import retention
+        except Exception:
+            return
+        try:
+            if not retention.due():
+                return
+            _in_flight = True
+            try:
+                stats = await asyncio.to_thread(retention.sweep)
+            finally:
+                _in_flight = False
+            if any(stats.values()):
+                PrintStyle().print(
+                    "Time Travel retention: "
+                    f"orphans={stats['orphans_removed']} aged={stats['aged_removed']} "
+                    f"locks={stats['stale_locks_removed']} "
+                    f"invalid={stats['invalid_backups_removed']} "
+                    f"reclaimed={stats['bytes_reclaimed']}b"
+                )
+        except Exception:
+            return

plugins/_time_travel/helpers/retention.py

@@ -0,0 +1,341 @@
+"""Retention for Time Travel shadow repositories.
+
+Time Travel keeps one hidden git repository per workspace under
+``/a0/usr/.time_travel/workspaces/<workspace_id>/repo.git`` and snapshots it on every file
+change. Without retention those repositories accumulate unboundedly: a removed chat or project
+leaves its shadow repository orphaned forever (nothing cleans it up), and a workspace whose
+``git add`` ever exceeded ``GIT_TIMEOUT_SECONDS`` strands a ``repo.git/index.lock`` that makes
+every later snapshot fail with "index.lock: File exists".
+
+The sweep (driven from ``job_loop``, throttled by config) removes:
+
+- ORPHANS — shadow directories whose id matches no live workspace path. Live paths are
+  forward-enumerated (project folders, the configured workdir, per-chat workdirs) and hashed
+  with the same ``workspace_id_for`` derivation; anything outside that set has no owner and can
+  never be shown in the UI again. Deleted once last activity is past a grace window.
+- AGED repositories — no snapshot in ``retention_max_age_days`` (0 = keep forever, the
+  default).
+- STALE LOCKS — ``repo.git/index.lock`` older than ``retention_stale_lock_minutes``; Time
+  Travel kills its git subprocesses at ``GIT_TIMEOUT_SECONDS``, so no legitimate lock lives
+  that long. Removing it un-wedges future snapshots.
+- INVALID BACKUPS — ``repo.git.invalid*`` set-asides made for corrupt repositories, past the
+  same grace window.
+
+Deleting a live workspace's shadow repository is always safe for the feature itself: the next
+snapshot lazily re-initializes an empty history. Deletion is refused for any path outside the
+shadow root.
+
+Durable state next to the workspaces dir: ``retention.json`` (running totals + last sweep
+stamp) and ``retention.log`` (one JSON line per sweep with the names of everything removed,
+tail-capped).
+"""
+
+from __future__ import annotations
+
+import datetime
+import json
+import os
+import shutil
+import time
+from typing import Any, Optional
+
+PLUGIN_NAME = "_time_travel"
+
+MARKER_FILE = "retention.json"
+HISTORY_FILE = "retention.log"
+HISTORY_MAX_LINES = 1000
+
+DEFAULT_CONFIG: dict[str, Any] = {
+    "retention_enabled": True,
+    "retention_sweep_interval_hours": 6,
+    "retention_max_age_days": 0,
+    "retention_orphan_grace_hours": 24,
+    "retention_stale_lock_minutes": 30,
+}
+
+
+def _int_at_least(value: Any, minimum: int, fallback: int) -> int:
+    try:
+        return max(int(value), minimum)
+    except Exception:
+        return fallback
+
+
+def effective_config(cfg: Optional[dict[str, Any]] = None) -> dict[str, Any]:
+    """Plugin config with defaults filled in and values clamped to sane minimums."""
+    if cfg is None:
+        try:
+            from helpers import plugins
+
+            cfg = plugins.get_plugin_config(PLUGIN_NAME) or {}
+        except Exception:
+            cfg = {}
+    merged = dict(DEFAULT_CONFIG)
+    merged.update({k: v for k, v in cfg.items() if k in DEFAULT_CONFIG and v is not None})
+    merged["retention_enabled"] = bool(merged["retention_enabled"])
+    merged["retention_sweep_interval_hours"] = _int_at_least(
+        merged["retention_sweep_interval_hours"], 1, 6
+    )
+    merged["retention_max_age_days"] = _int_at_least(merged["retention_max_age_days"], 0, 0)
+    merged["retention_orphan_grace_hours"] = _int_at_least(
+        merged["retention_orphan_grace_hours"], 1, 24
+    )
+    merged["retention_stale_lock_minutes"] = _int_at_least(
+        merged["retention_stale_lock_minutes"], 5, 30
+    )
+    return merged
+
+
+def _state_dir() -> str:
+    from plugins._time_travel.helpers import time_travel
+
+    return str(time_travel.real_path_for_display("/a0/usr/.time_travel"))
+
+
+def _shadow_root() -> str:
+    from plugins._time_travel.helpers import time_travel
+
+    return str(time_travel.real_path_for_display(time_travel.SHADOW_DISPLAY_ROOT))
+
+
+def live_workspace_ids() -> set[str]:
+    """Every workspace id resolvable from a path that exists right now: project folders, the
+    configured workdir, and per-chat workdirs (custom projects resolvers may mint workspaces
+    there; including them only makes the sweep more conservative)."""
+    from plugins._time_travel.helpers import time_travel
+
+    ids: set[str] = set()
+
+    projects_root = time_travel.real_path_for_display("/a0/usr/projects")
+    try:
+        for name in os.listdir(projects_root):
+            if os.path.isdir(os.path.join(projects_root, name)):
+                ids.add(time_travel.workspace_id_for(f"/a0/usr/projects/{name}"))
+    except Exception:
+        pass
+
+    try:
+        ids.add(time_travel.workspace_id_for(time_travel.configured_workdir_display_path()))
+    except Exception:
+        ids.add(time_travel.workspace_id_for("/a0/usr/workdir"))
+
+    chats_root = time_travel.real_path_for_display("/a0/usr/chats")
+    try:
+        for name in os.listdir(chats_root):
+            if os.path.isdir(os.path.join(chats_root, name, "workdir")):
+                ids.add(time_travel.workspace_id_for(f"/a0/usr/chats/{name}/workdir"))
+    except Exception:
+        pass
+
+    return ids
+
+
+def _read_json(path: str) -> dict[str, Any]:
+    try:
+        with open(path, "r", encoding="utf-8") as f:
+            return json.load(f)
+    except Exception:
+        return {}
+
+
+def _write_marker(state_dir: str, sweep_stats: dict[str, int], stamp: str) -> None:
+    try:
+        os.makedirs(state_dir, exist_ok=True)
+        path = os.path.join(state_dir, MARKER_FILE)
+        payload = _read_json(path)
+        payload["sweeps"] = int(payload.get("sweeps", 0)) + 1
+        for key, value in sweep_stats.items():
+            payload[key] = int(payload.get(key, 0)) + int(value)
+        payload["last_sweep_at"] = stamp
+        tmp = path + ".tmp"
+        with open(tmp, "w", encoding="utf-8") as f:
+            json.dump(payload, f, indent=2, sort_keys=True)
+        os.replace(tmp, path)
+    except Exception:
+        pass
+
+
+def _append_history(state_dir: str, entry: dict[str, Any]) -> None:
+    try:
+        os.makedirs(state_dir, exist_ok=True)
+        path = os.path.join(state_dir, HISTORY_FILE)
+        lines: list[str] = []
+        try:
+            with open(path, "r", encoding="utf-8") as f:
+                lines = [ln for ln in f.read().splitlines() if ln.strip()]
+        except Exception:
+            lines = []
+        lines.append(json.dumps(entry, sort_keys=True))
+        if len(lines) > HISTORY_MAX_LINES:
+            lines = lines[-HISTORY_MAX_LINES:]
+        tmp = path + ".tmp"
+        with open(tmp, "w", encoding="utf-8") as f:
+            f.write("\n".join(lines) + "\n")
+        os.replace(tmp, path)
+    except Exception:
+        pass
+
+
+def read_history(limit: int = 50, state_dir: Optional[str] = None) -> list[dict[str, Any]]:
+    try:
+        base = state_dir if state_dir is not None else _state_dir()
+        with open(os.path.join(base, HISTORY_FILE), "r", encoding="utf-8") as f:
+            lines = [ln for ln in f.read().splitlines() if ln.strip()]
+        return [json.loads(ln) for ln in lines[-limit:]]
+    except Exception:
+        return []
+
+
+def _last_activity(entry_path: str) -> float:
+    candidates = [
+        os.path.join(entry_path, "repo.git", "refs", "heads", "current"),
+        os.path.join(entry_path, "repo.git", "packed-refs"),
+        os.path.join(entry_path, "repo.git", "HEAD"),
+        os.path.join(entry_path, "repo.git"),
+        entry_path,
+    ]
+    newest = 0.0
+    for candidate in candidates:
+        try:
+            newest = max(newest, os.stat(candidate).st_mtime)
+        except Exception:
+            continue
+    return newest
+
+
+def _tree_bytes(path: str) -> int:
+    total = 0
+    try:
+        for root, _dirs, names in os.walk(path):
+            for name in names:
+                try:
+                    total += os.stat(os.path.join(root, name)).st_size
+                except Exception:
+                    pass
+    except Exception:
+        pass
+    return total
+
+
+def _remove_tree(path: str, shadow_root: str) -> int:
+    """rmtree guarded to the shadow root; returns bytes reclaimed (0 on refusal/failure)."""
+    real = os.path.realpath(path)
+    root = os.path.realpath(shadow_root)
+    if not real.startswith(root + os.sep):
+        return 0
+    size = _tree_bytes(real)
+    try:
+        shutil.rmtree(real)
+        return size
+    except Exception:
+        return 0
+
+
+def sweep(
+    cfg: Optional[dict[str, Any]] = None,
+    shadow_root: Optional[str] = None,
+    live_ids: Optional[set[str]] = None,
+    now_ts: Optional[float] = None,
+    state_dir: Optional[str] = None,
+) -> dict[str, int]:
+    """One retention pass. All inputs are injectable for tests; production callers pass
+    nothing and everything resolves from the plugin runtime."""
+    stats = {
+        "orphans_removed": 0,
+        "aged_removed": 0,
+        "stale_locks_removed": 0,
+        "invalid_backups_removed": 0,
+        "bytes_reclaimed": 0,
+    }
+    config = effective_config(cfg)
+    if not config["retention_enabled"]:
+        return stats
+    root = shadow_root if shadow_root is not None else _shadow_root()
+    if not os.path.isdir(root):
+        return stats
+    ids = live_ids if live_ids is not None else live_workspace_ids()
+    base = state_dir if state_dir is not None else _state_dir()
+    now = time.time() if now_ts is None else now_ts
+
+    max_age_s = config["retention_max_age_days"] * 86400
+    grace_s = config["retention_orphan_grace_hours"] * 3600
+    lock_s = config["retention_stale_lock_minutes"] * 60
+
+    detail: dict[str, list[str]] = {"orphans": [], "aged": [], "locks": [], "invalid": []}
+
+    try:
+        entries = os.listdir(root)
+    except Exception:
+        return stats
+
+    for name in entries:
+        entry = os.path.join(root, name)
+        if not os.path.isdir(entry):
+            continue
+        last = _last_activity(entry)
+
+        if name not in ids:
+            if now - last > grace_s:
+                stats["bytes_reclaimed"] += _remove_tree(entry, root)
+                stats["orphans_removed"] += 1
+                detail["orphans"].append(name)
+            continue
+
+        if max_age_s and now - last > max_age_s:
+            stats["bytes_reclaimed"] += _remove_tree(entry, root)
+            stats["aged_removed"] += 1
+            detail["aged"].append(name)
+            continue
+
+        lock = os.path.join(entry, "repo.git", "index.lock")
+        try:
+            if os.path.isfile(lock) and now - os.stat(lock).st_mtime > lock_s:
+                os.remove(lock)
+                stats["stale_locks_removed"] += 1
+                detail["locks"].append(name)
+        except Exception:
+            pass
+
+        try:
+            for sub in os.listdir(entry):
+                if sub.startswith("repo.git.invalid"):
+                    backup = os.path.join(entry, sub)
+                    if now - os.stat(backup).st_mtime > grace_s:
+                        stats["bytes_reclaimed"] += _remove_tree(backup, root)
+                        stats["invalid_backups_removed"] += 1
+                        detail["invalid"].append(f"{name}/{sub}")
+        except Exception:
+            pass
+
+    stamp = datetime.datetime.now(datetime.timezone.utc).isoformat()
+    _write_marker(base, stats, stamp)
+    _append_history(base, {"at": stamp, **stats, "removed": detail})
+    return stats
+
+
+def due(
+    cfg: Optional[dict[str, Any]] = None,
+    now_ts: Optional[float] = None,
+    state_dir: Optional[str] = None,
+) -> bool:
+    """True when retention is enabled and the configured interval has elapsed since the last
+    sweep (or no sweep ever ran)."""
+    config = effective_config(cfg)
+    if not config["retention_enabled"]:
+        return False
+    base = state_dir if state_dir is not None else _state_dir()
+    marker = _read_json(os.path.join(base, MARKER_FILE))
+    last = str(marker.get("last_sweep_at") or "")
+    if not last:
+        return True
+    try:
+        last_dt = datetime.datetime.fromisoformat(last)
+        now = (
+            datetime.datetime.now(datetime.timezone.utc)
+            if now_ts is None
+            else datetime.datetime.fromtimestamp(now_ts, datetime.timezone.utc)
+        )
+        interval_s = config["retention_sweep_interval_hours"] * 3600
+        return (now - last_dt).total_seconds() >= interval_s
+    except Exception:
+        return True

plugins/_time_travel/plugin.yaml

@@ -3,6 +3,7 @@ title: Time Travel
 description: Agent Zero-owned workdir/project history, diff inspection, travel, and revert for /a0/usr workspaces.
 version: 0.1.0
 always_enabled: false
-settings_sections: []
+settings_sections:
+  - agent
 per_project_config: false
 per_agent_config: false

plugins/_time_travel/webui/config.html

@@ -0,0 +1,104 @@
+<html>
+
+<head>
+    <title>Time Travel</title>
+</head>
+
+<body>
+    <div x-data>
+        <template x-if="config">
+            <div>
+                <div class="section-title">Time Travel retention</div>
+                <div class="section-description">
+                    Time Travel keeps a hidden history of every workspace it snapshots. Retention
+                    keeps that storage bounded by cleaning up histories nothing can reach anymore
+                    (deleted chats and projects) and, optionally, aging out old history.
+                </div>
+
+                <div class="field">
+                    <div class="field-label">
+                        <div class="field-title">Enable retention</div>
+                        <div class="field-description">
+                            Periodically clean up orphaned workspace histories, stale snapshot
+                            locks, and corrupt-repository backups. Turning this off means Time
+                            Travel storage grows without bound.
+                        </div>
+                    </div>
+                    <div class="field-control">
+                        <label class="toggle">
+                            <input type="checkbox" x-model="config.retention_enabled"
+                                x-init="if (config.retention_enabled === undefined || config.retention_enabled === null) config.retention_enabled = true" />
+                            <span class="toggler"></span>
+                        </label>
+                    </div>
+                </div>
+
+                <div class="field">
+                    <div class="field-label">
+                        <div class="field-title">Sweep interval (hours)</div>
+                        <div class="field-description">
+                            How often the cleanup runs.
+                        </div>
+                    </div>
+                    <div class="field-control">
+                        <input type="number" min="1" step="1"
+                            x-init="if (config.retention_sweep_interval_hours === undefined || config.retention_sweep_interval_hours === null) config.retention_sweep_interval_hours = 6"
+                            x-model.number="config.retention_sweep_interval_hours" />
+                    </div>
+                </div>
+
+                <div class="field">
+                    <div class="field-label">
+                        <div class="field-title">Delete history older than (days)</div>
+                        <div class="field-description">
+                            Leave at 0 (the default) and workspace history is kept forever. Set a
+                            number — say 30 — and a workspace with no snapshot in that many days
+                            has its history deleted. Time Travel simply starts a fresh history on
+                            the workspace's next change.
+                        </div>
+                    </div>
+                    <div class="field-control">
+                        <input type="number" min="0" step="1"
+                            x-init="if (config.retention_max_age_days === undefined || config.retention_max_age_days === null) config.retention_max_age_days = 0"
+                            x-model.number="config.retention_max_age_days" />
+                    </div>
+                </div>
+
+                <div class="field">
+                    <div class="field-label">
+                        <div class="field-title">Orphan grace period (hours)</div>
+                        <div class="field-description">
+                            History belonging to a deleted chat or project is unreachable from the
+                            UI and gets cleaned up — but only after it has been inactive this long,
+                            so recent work is never raced.
+                        </div>
+                    </div>
+                    <div class="field-control">
+                        <input type="number" min="1" step="1"
+                            x-init="if (config.retention_orphan_grace_hours === undefined || config.retention_orphan_grace_hours === null) config.retention_orphan_grace_hours = 24"
+                            x-model.number="config.retention_orphan_grace_hours" />
+                    </div>
+                </div>
+
+                <div class="field">
+                    <div class="field-label">
+                        <div class="field-title">Stale snapshot lock age (minutes)</div>
+                        <div class="field-description">
+                            A leftover git index.lock older than this is removed so snapshots stop
+                            failing with "index.lock: File exists". Time Travel's own git
+                            operations are killed after 20 seconds, so no legitimate lock lives
+                            this long.
+                        </div>
+                    </div>
+                    <div class="field-control">
+                        <input type="number" min="5" step="1"
+                            x-init="if (config.retention_stale_lock_minutes === undefined || config.retention_stale_lock_minutes === null) config.retention_stale_lock_minutes = 30"
+                            x-model.number="config.retention_stale_lock_minutes" />
+                    </div>
+                </div>
+            </div>
+        </template>
+    </div>
+</body>
+
+</html>

tests/test_time_travel_retention.py

@@ -0,0 +1,226 @@
+import hashlib
+import importlib.util
+import json
+import os
+import sys
+import time
+from pathlib import Path
+
+import pytest
+
+PROJECT_ROOT = Path(__file__).resolve().parents[1]
+if str(PROJECT_ROOT) not in sys.path:
+    sys.path.insert(0, str(PROJECT_ROOT))
+
+MOD_PATH = PROJECT_ROOT / "plugins" / "_time_travel" / "helpers" / "retention.py"
+_spec = importlib.util.spec_from_file_location("time_travel_retention", MOD_PATH)
+retention = importlib.util.module_from_spec(_spec)
+assert _spec and _spec.loader
+_spec.loader.exec_module(retention)
+
+DAY = 86400
+HOUR = 3600
+NOW = time.time()
+
+CFG_AGING = {"retention_max_age_days": 30}
+
+
+def _hex_id(seed: str) -> str:
+    return hashlib.sha256(seed.encode()).hexdigest()[:32]
+
+
+def _mk_repo(shadow_root, name, age_s, lock_age_s=None, invalid_age_s=None):
+    entry = os.path.join(shadow_root, name)
+    ref = os.path.join(entry, "repo.git", "refs", "heads")
+    os.makedirs(ref, exist_ok=True)
+    cur = os.path.join(ref, "current")
+    with open(cur, "w") as f:
+        f.write("deadbeef")
+    stamp = NOW - age_s
+    for path in (cur, os.path.join(entry, "repo.git"), entry):
+        os.utime(path, (stamp, stamp))
+    if lock_age_s is not None:
+        lock = os.path.join(entry, "repo.git", "index.lock")
+        with open(lock, "w") as f:
+            f.write("")
+        os.utime(lock, (NOW - lock_age_s, NOW - lock_age_s))
+    if invalid_age_s is not None:
+        backup = os.path.join(entry, "repo.git.invalid-20260101")
+        os.makedirs(backup, exist_ok=True)
+        with open(os.path.join(backup, "junk"), "w") as f:
+            f.write("x" * 100)
+        os.utime(backup, (NOW - invalid_age_s, NOW - invalid_age_s))
+    return entry
+
+
+def test_effective_config_defaults_and_clamps():
+    cfg = retention.effective_config({})
+    assert cfg["retention_enabled"] is True
+    assert cfg["retention_max_age_days"] == 0
+    assert cfg["retention_sweep_interval_hours"] == 6
+    clamped = retention.effective_config(
+        {
+            "retention_sweep_interval_hours": 0,
+            "retention_max_age_days": -5,
+            "retention_orphan_grace_hours": 0,
+            "retention_stale_lock_minutes": 1,
+            "retention_enabled": 1,
+        }
+    )
+    assert clamped["retention_sweep_interval_hours"] == 1
+    assert clamped["retention_max_age_days"] == 0
+    assert clamped["retention_orphan_grace_hours"] == 1
+    assert clamped["retention_stale_lock_minutes"] == 5
+    assert clamped["retention_enabled"] is True
+    garbage = retention.effective_config({"retention_sweep_interval_hours": "nope"})
+    assert garbage["retention_sweep_interval_hours"] == 6
+
+
+def test_sweep_matrix(tmp_path):
+    shadow = str(tmp_path / "workspaces")
+    state = str(tmp_path / "state")
+    os.makedirs(shadow)
+
+    live_recent = _hex_id("alpha")
+    live_aged = _hex_id("beta")
+    live_locked = _hex_id("gamma")
+    live_fresh_lock = _hex_id("delta")
+    live_invalid = _hex_id("epsilon")
+    live_ids = {live_recent, live_aged, live_locked, live_fresh_lock, live_invalid}
+
+    _mk_repo(shadow, live_recent, age_s=1 * HOUR)
+    _mk_repo(shadow, live_aged, age_s=40 * DAY)
+    _mk_repo(shadow, live_locked, age_s=1 * HOUR, lock_age_s=1 * HOUR)
+    _mk_repo(shadow, live_fresh_lock, age_s=1 * HOUR, lock_age_s=60)
+    _mk_repo(shadow, live_invalid, age_s=1 * HOUR, invalid_age_s=48 * HOUR)
+    _mk_repo(shadow, "0" * 32, age_s=48 * HOUR)  # orphan past grace
+    _mk_repo(shadow, "1" * 32, age_s=1 * HOUR)  # orphan inside grace
+    with open(os.path.join(shadow, "stray-file"), "w") as f:
+        f.write("ignore me")
+
+    stats = retention.sweep(
+        cfg=CFG_AGING, shadow_root=shadow, live_ids=live_ids, now_ts=NOW, state_dir=state
+    )
+
+    assert os.path.isdir(os.path.join(shadow, live_recent))
+    assert not os.path.exists(os.path.join(shadow, live_aged))
+    assert stats["aged_removed"] == 1
+    assert not os.path.exists(os.path.join(shadow, "0" * 32))
+    assert os.path.isdir(os.path.join(shadow, "1" * 32))
+    assert stats["orphans_removed"] == 1
+    assert os.path.isdir(os.path.join(shadow, live_locked))
+    assert not os.path.exists(os.path.join(shadow, live_locked, "repo.git", "index.lock"))
+    assert os.path.exists(os.path.join(shadow, live_fresh_lock, "repo.git", "index.lock"))
+    assert stats["stale_locks_removed"] == 1
+    assert os.path.isdir(os.path.join(shadow, live_invalid))
+    assert not os.path.exists(
+        os.path.join(shadow, live_invalid, "repo.git.invalid-20260101")
+    )
+    assert stats["invalid_backups_removed"] == 1
+    assert stats["bytes_reclaimed"] > 0
+    assert os.path.isfile(os.path.join(shadow, "stray-file"))
+
+
+def test_max_age_zero_keeps_history_forever(tmp_path):
+    shadow = str(tmp_path / "workspaces")
+    state = str(tmp_path / "state")
+    os.makedirs(shadow)
+    ancient = _hex_id("ancient")
+    _mk_repo(shadow, ancient, age_s=400 * DAY)
+
+    stats = retention.sweep(
+        cfg={"retention_max_age_days": 0},
+        shadow_root=shadow,
+        live_ids={ancient},
+        now_ts=NOW,
+        state_dir=state,
+    )
+    assert stats["aged_removed"] == 0
+    assert os.path.isdir(os.path.join(shadow, ancient))
+
+
+def test_disabled_sweep_is_noop(tmp_path):
+    shadow = str(tmp_path / "workspaces")
+    os.makedirs(shadow)
+    _mk_repo(shadow, "0" * 32, age_s=48 * HOUR)
+    stats = retention.sweep(
+        cfg={"retention_enabled": False},
+        shadow_root=shadow,
+        live_ids=set(),
+        now_ts=NOW,
+        state_dir=str(shadow),
+    )
+    assert stats["orphans_removed"] == 0
+    assert os.path.isdir(os.path.join(shadow, "0" * 32))
+
+
+def test_remove_tree_refuses_outside_root(tmp_path):
+    shadow = str(tmp_path / "workspaces")
+    outside = str(tmp_path / "outside")
+    os.makedirs(shadow)
+    os.makedirs(outside)
+    assert retention._remove_tree(outside, shadow) == 0
+    assert os.path.isdir(outside)
+
+
+def test_marker_and_history(tmp_path):
+    shadow = str(tmp_path / "workspaces")
+    state = str(tmp_path / "state")
+    os.makedirs(shadow)
+    _mk_repo(shadow, "0" * 32, age_s=48 * HOUR)
+    retention.sweep(cfg={}, shadow_root=shadow, live_ids=set(), now_ts=NOW, state_dir=state)
+    _mk_repo(shadow, "2" * 32, age_s=48 * HOUR)
+    retention.sweep(cfg={}, shadow_root=shadow, live_ids=set(), now_ts=NOW, state_dir=state)
+
+    marker = json.load(open(os.path.join(state, retention.MARKER_FILE)))
+    assert marker["sweeps"] == 2
+    assert marker["orphans_removed"] == 2
+    assert marker["last_sweep_at"]
+
+    history = retention.read_history(state_dir=state)
+    assert len(history) == 2
+    assert history[0]["removed"]["orphans"] == ["0" * 32]
+    assert history[1]["removed"]["orphans"] == ["2" * 32]
+    assert history[0]["at"]
+
+
+def test_history_tail_cap(tmp_path):
+    state = str(tmp_path / "state")
+    os.makedirs(state)
+    with open(os.path.join(state, retention.HISTORY_FILE), "w") as f:
+        for i in range(retention.HISTORY_MAX_LINES + 20):
+            f.write('{"at": "old-%d"}\n' % i)
+    retention._append_history(state, {"at": "newest"})
+    history = retention.read_history(limit=retention.HISTORY_MAX_LINES + 100, state_dir=state)
+    assert len(history) == retention.HISTORY_MAX_LINES
+    assert history[-1]["at"] == "newest"
+    assert history[0]["at"] != "old-0"
+
+
+def test_due_throttle(tmp_path):
+    state = str(tmp_path / "state")
+    shadow = str(tmp_path / "workspaces")
+    os.makedirs(shadow)
+    assert retention.due(cfg={}, state_dir=state)
+    assert not retention.due(cfg={"retention_enabled": False}, state_dir=state)
+
+    retention.sweep(cfg={}, shadow_root=shadow, live_ids=set(), now_ts=NOW, state_dir=state)
+    assert not retention.due(cfg={}, now_ts=time.time(), state_dir=state)
+    assert retention.due(cfg={}, now_ts=time.time() + 7 * HOUR, state_dir=state)
+    assert not retention.due(
+        cfg={"retention_sweep_interval_hours": 12},
+        now_ts=time.time() + 7 * HOUR,
+        state_dir=state,
+    )
+
+
+def test_workspace_id_parity_with_time_travel():
+    time_travel = pytest.importorskip(
+        "plugins._time_travel.helpers.time_travel",
+        reason="requires the full runtime environment",
+    )
+    path = "/a0/usr/projects/example"
+    expected = hashlib.sha256(
+        time_travel.canonical_workspace_display_path(path).rstrip("/").encode("utf-8")
+    ).hexdigest()[:32]
+    assert time_travel.workspace_id_for(path) == expected