Add worktree restore script for symmetric one-click recovery

scripts/keep_codex_fast.py

@@ -17,7 +17,7 @@
 import sys
 import time
 from dataclasses import dataclass
-from datetime import datetime, timedelta
+from datetime import datetime, timedelta, timezone
 from pathlib import Path
 
 
@@ -249,7 +249,7 @@ def append_session_index_name(codex_home: Path, thread_id: str, name: str) -> No
     entry = {
         "id": thread_id,
         "thread_name": name,
-        "updated_at": datetime.utcnow().isoformat(timespec="milliseconds") + "Z",
+        "updated_at": datetime.now(timezone.utc).replace(tzinfo=None).isoformat(timespec="milliseconds") + "Z",
     }
     with path.open("a", encoding="utf-8") as handle:
         handle.write(json.dumps(entry, ensure_ascii=False) + "\n")
@@ -656,10 +656,32 @@ def move_stale_worktrees(codex_home: Path, backup_root: Path, days: int, stamp:
             item_size = size_bytes(source)
             shutil.move(str(source), str(dest))
             handle.write(json.dumps({"from": str(source), "to": str(dest), "bytes": item_size}) + "\n")
+    write_worktree_restore_script(manifest, backup_root)
     report(f"worktree_archive_root {archive_root}")
     report(f"worktree_manifest {manifest}")
 
 
+def write_worktree_restore_script(manifest: Path, backup_root: Path) -> None:
+    restore = backup_root / "restore-worktrees.py"
+    restore.write_text(
+        f'''import json
+import shutil
+from pathlib import Path
+
+manifest = Path(r"{manifest}")
+for line in manifest.read_text(encoding="utf-8").splitlines():
+    rec = json.loads(line)
+    src = Path(rec["to"])
+    dest = Path(rec["from"])
+    if src.exists():
+        dest.parent.mkdir(parents=True, exist_ok=True)
+        shutil.move(str(src), str(dest))
+''',
+        encoding="utf-8",
+    )
+    report(f"worktree_restore_script {restore}")
+
+
 def rotate_logs(codex_home: Path, threshold_mb: int, stamp: str, apply: bool) -> None:
     files = [path for path in codex_home.glob("logs_2.sqlite*") if path.is_file()]
     total = sum(path.stat().st_size for path in files)

tests/smoke_test.py

@@ -220,6 +220,7 @@ def assert_apply_mode(module) -> None:
         )
         assert (backup / "restore-sessions.py").exists()
         assert (backup / "restore-thread-metadata.py").exists()
+        assert (backup / "restore-worktrees.py").exists()
         assert (backup / "moved-sessions.jsonl").exists()
         assert (backup / "thread-metadata-repairs.jsonl").exists()
         assert (backup / "moved-worktrees.jsonl").exists()
@@ -261,13 +262,52 @@ def assert_normal_apply_does_not_repair_thread_metadata(module) -> None:
         assert not (backup / "restore-thread-metadata.py").exists()
 
 
+def assert_worktree_restore_round_trip(module) -> None:
+    with tempfile.TemporaryDirectory() as td:
+        root = Path(td)
+        codex_home = root / ".codex"
+        backup = root / "backup-worktree"
+        backup.mkdir(parents=True)
+        worktree = codex_home / "worktrees" / "oldtree" / "src"
+        worktree.mkdir(parents=True)
+        payload = worktree / "file.txt"
+        payload.write_text("worktree-payload", encoding="utf-8")
+        old_time = time.time() - 30 * 86400
+        os.utime(codex_home / "worktrees" / "oldtree", (old_time, old_time))
+
+        stamp = module.now_stamp()
+        module.move_stale_worktrees(codex_home, backup, days=7, stamp=stamp, apply=True)
+
+        # archived, not deleted
+        assert not (codex_home / "worktrees" / "oldtree").exists(), "stale worktree must be moved out of hot path"
+        archived = list((codex_home / "archived_worktrees").rglob("file.txt"))
+        assert len(archived) == 1, "worktree payload must exist in archive after apply"
+        assert archived[0].read_text(encoding="utf-8") == "worktree-payload"
+
+        manifest = backup / "moved-worktrees.jsonl"
+        restore = backup / "restore-worktrees.py"
+        assert manifest.exists(), "apply must write moved-worktrees manifest"
+        assert restore.exists(), "apply must write restore-worktrees.py"
+
+        # run the generated restore script and confirm the round-trip
+        import subprocess
+
+        result = subprocess.run([sys.executable, str(restore)], capture_output=True, text=True)
+        assert result.returncode == 0, f"restore script failed: {result.stderr}"
+        restored = codex_home / "worktrees" / "oldtree" / "src" / "file.txt"
+        assert restored.exists(), "restore must move worktree back to original path"
+        assert restored.read_text(encoding="utf-8") == "worktree-payload"
+        assert not list((codex_home / "archived_worktrees").rglob("file.txt")), "archive must be emptied after restore"
+
+
 def main() -> int:
     module = load_module()
     assert_report_mode(module)
     assert_backup_only_mode(module)
     assert_session_alias_detection(module)
     assert_normal_apply_does_not_repair_thread_metadata(module)
     assert_apply_mode(module)
+    assert_worktree_restore_round_trip(module)
     print("smoke tests passed")
     return 0