Add safe backfill for the public experiment archive

docs/torch.md

@@ -180,6 +180,42 @@ The archive is public. Review new artifact types before adding them to the run
 directory, and never write secrets there. Local W&B state, common credential and
 key filenames, and incomplete temporary files are excluded from publication.
 
+## Backfill Historical Experiments to the Public OSN Archive
+
+After configuring the authenticated rclone remote above, an operator can copy
+completed and failed experiment directories from Torch scratch. Run the tool
+once per authorized user root; it never scans all of `/scratch`, and it copies
+only direct child run directories. It never deletes local or remote files and is
+a dry run unless `--apply` is present.
+
+Set the owner to the user whose run directory is being migrated, then inspect
+the complete plan or select named runs:
+
+```bash
+export ARCHIVE_OWNER=aa1234
+python3 scripts/experiment_archive.py --owner "$ARCHIVE_OWNER" \
+  backfill "/scratch/$ARCHIVE_OWNER/runs"
+python3 scripts/experiment_archive.py --owner "$ARCHIVE_OWNER" \
+  backfill "/scratch/$ARCHIVE_OWNER/runs" \
+  --run 2026-08-01-baseline \
+  --run 2026-08-02-ablation
+```
+
+Then perform and verify the copy, retaining the migration report:
+
+```bash
+python3 scripts/experiment_archive.py --owner "$ARCHIVE_OWNER" \
+  backfill "/scratch/$ARCHIVE_OWNER/runs" \
+  --apply \
+  --report experiment-backfill-report.json
+```
+
+The default destination is
+`nyu-osn:m2lines-pubs/Samudra/experiments/<owner>/<run-name>`. Inspect every
+selected run before publishing because its contents become publicly readable.
+The operator must have read permission for the source path and write credentials
+for OSN.
+
 ## Fast Iteration With Ref-Built Code Overlays
 
 The code-layer builder fetches a pushed Git ref, resolves it to a full commit,

scripts/experiment_archive.py

@@ -25,7 +25,7 @@
 import threading
 from datetime import datetime, timezone
 from pathlib import Path
-from typing import Dict, List, Optional, Sequence
+from typing import Dict, Iterable, List, Optional, Sequence
 
 DEFAULT_ARCHIVE_BASE = "nyu-osn:m2lines-pubs/Samudra/experiments"
 STATUS_FILENAME = "archive-status.json"
@@ -123,6 +123,40 @@ def validate_run_dir(run_dir: Path) -> Path:
     return resolved
 
 
+def _validate_source_root(source_root: Path) -> Path:
+    if source_root.is_symlink():
+        raise ArchiveError(f"refusing symlinked source root: {source_root}")
+    if not source_root.is_dir():
+        raise ArchiveError(f"source root does not exist: {source_root}")
+    resolved = source_root.resolve()
+    if resolved in (Path("/"), Path("/scratch")):
+        raise ArchiveError(
+            "source root is too broad; pass the directory containing run directories"
+        )
+    return resolved
+
+
+def discover_runs(source_root: Path, requested: Iterable[str]) -> List[Path]:
+    source_root = _validate_source_root(source_root)
+    requested = list(requested)
+    if requested:
+        runs = []
+        for run_name in requested:
+            archive_destination("validation", "validation", run_name)
+            runs.append(validate_run_dir(source_root / run_name))
+        return runs
+
+    runs = []
+    for candidate in sorted(source_root.iterdir(), key=lambda path: path.name):
+        if candidate.name.startswith(".") or candidate.is_symlink():
+            continue
+        if candidate.is_dir():
+            runs.append(candidate.resolve())
+    if not runs:
+        raise ArchiveError(f"no run directories found under {source_root}")
+    return runs
+
+
 def publish_run(
     run_dir: Path,
     archive_base: str,
@@ -251,6 +285,61 @@ def watch_run(
     return 1 if failures else 0
 
 
+def backfill(
+    source_root: Path,
+    archive_base: str,
+    owner: str,
+    requested_runs: Iterable[str],
+    apply: bool,
+    report_path: Optional[Path],
+    rclone_bin: str = "rclone",
+) -> int:
+    source_root = _validate_source_root(source_root)
+    owner = _validate_archive_segment(owner, "archive owner")
+    runs = discover_runs(source_root, requested_runs)
+    run_reports = []  # type: List[Dict[str, object]]
+    report = {
+        "schema_version": 1,
+        "created_at": utc_now(),
+        "source_root": str(source_root),
+        "archive_base": archive_base,
+        "owner": owner,
+        "apply": apply,
+        "runs": run_reports,
+    }  # type: Dict[str, object]
+    failures = 0
+
+    for run_dir in runs:
+        try:
+            result = publish_run(
+                run_dir,
+                archive_base,
+                owner,
+                apply=apply,
+                verify=True,
+                rclone_bin=rclone_bin,
+            )
+        except subprocess.CalledProcessError as error:
+            failures += 1
+            result = {
+                "source": str(run_dir),
+                "destination": archive_destination(archive_base, owner, run_dir.name),
+                "owner": owner,
+                "status": "failed",
+                "verified": False,
+                "failed_command": _display_command(error.cmd),
+                "returncode": error.returncode,
+            }
+        run_reports.append(result)
+
+    report["completed_at"] = utc_now()
+    report["failure_count"] = failures
+    if report_path is not None:
+        _write_json_atomic(report_path, report)
+        print(f"Wrote migration report: {report_path.resolve()}")
+    return 1 if failures else 0
+
+
 def _build_parser() -> argparse.ArgumentParser:
     parser = argparse.ArgumentParser(description=__doc__)
     parser.add_argument(
@@ -295,6 +384,26 @@ def _build_parser() -> argparse.ArgumentParser:
     status_parser.add_argument("state", choices=VALID_STATES)
     status_parser.add_argument("--exit-code", type=int)
 
+    backfill_parser = subparsers.add_parser(
+        "backfill", help="publish historical child run directories"
+    )
+    backfill_parser.add_argument("source_root", type=Path)
+    backfill_parser.add_argument(
+        "--run",
+        action="append",
+        default=[],
+        help="publish only this direct child; repeat to select several runs",
+    )
+    backfill_parser.add_argument(
+        "--apply",
+        action="store_true",
+        help="perform copies; without this flag only print the migration plan",
+    )
+    backfill_parser.add_argument(
+        "--report",
+        type=Path,
+        help="write a JSON report containing every planned or verified run",
+    )
     return parser
 
 
@@ -314,6 +423,16 @@ def main(argv: Optional[Sequence[str]] = None) -> int:
                 rclone_bin=args.rclone_bin,
             )
             return 0
+        if args.command == "backfill":
+            return backfill(
+                args.source_root,
+                args.archive_base,
+                args.owner,
+                args.run,
+                apply=args.apply,
+                report_path=args.report,
+                rclone_bin=args.rclone_bin,
+            )
         if args.command == "status":
             write_status(args.run_dir, args.owner, args.state, args.exit_code)
             return 0

tests/test_experiment_archive.py

@@ -101,6 +101,57 @@ def record(command, check):
     assert result["verified"] is True
 
 
+def test_discover_runs_uses_direct_non_hidden_directories(tmp_path):
+    (tmp_path / "b-run").mkdir()
+    (tmp_path / "a-run").mkdir()
+    (tmp_path / ".cache").mkdir()
+    (tmp_path / "file.txt").write_text("not a run", encoding="utf-8")
+    (tmp_path / "linked-run").symlink_to(tmp_path / "a-run", target_is_directory=True)
+
+    runs = experiment_archive.discover_runs(tmp_path, [])
+
+    assert [run.name for run in runs] == ["a-run", "b-run"]
+
+
+def test_backfill_writes_report_and_returns_nonzero_for_failures(tmp_path, monkeypatch):
+    source_root = tmp_path / "runs"
+    source_root.mkdir()
+    (source_root / "good-run").mkdir()
+    (source_root / "bad-run").mkdir()
+    report_path = tmp_path / "report.json"
+
+    def fake_run(command, check):
+        assert check is True
+        if "bad-run" in command[2]:
+            raise subprocess.CalledProcessError(7, command)
+
+    monkeypatch.setattr(subprocess, "run", fake_run)
+
+    result = experiment_archive.backfill(
+        source_root,
+        "remote:bucket/archive",
+        "owner-a",
+        [],
+        apply=True,
+        report_path=report_path,
+    )
+
+    assert result == 1
+    report = json.loads(report_path.read_text(encoding="utf-8"))
+    assert report["failure_count"] == 1
+    assert report["owner"] == "owner-a"
+    assert {run["status"] for run in report["runs"]} == {"failed", "verified"}
+    assert all(
+        run["destination"].startswith("remote:bucket/archive/owner-a/")
+        for run in report["runs"]
+    )
+
+
+def test_source_root_rejects_filesystem_root():
+    with pytest.raises(experiment_archive.ArchiveError, match="too broad"):
+        experiment_archive.discover_runs(Path("/"), [])
+
+
 def test_status_tracks_run_lifecycle(tmp_path, monkeypatch):
     run_dir = tmp_path / "run-a"
     run_dir.mkdir()