diff --git a/docs/torch.md b/docs/torch.md index 4913bf85..4f1aabad 100644 --- a/docs/torch.md +++ b/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//`. 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, diff --git a/scripts/experiment_archive.py b/scripts/experiment_archive.py index e9db29fa..a8b5c8fb 100644 --- a/scripts/experiment_archive.py +++ b/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 diff --git a/tests/test_experiment_archive.py b/tests/test_experiment_archive.py index 156b17a8..72bdb7c8 100644 --- a/tests/test_experiment_archive.py +++ b/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()