Give CheckpointJob the training state it already owned

_has_training_job_completed and _mark_training_job_completed read nothing
but the job, and the summary re-derived the legacy rule from
physics_backend rather than asking. They become is_trained and
mark_trained, so the completion marker sits next to the only state that
explains it and the legacy branch is stated once.

is_trained is now consistent between training and reporting: a legacy run
predates the marker, so requiring one there would retrain every published
legacy checkpoint. Tests pin both halves of that rule. The collect locals
follow the declared-checkpoint vocabulary.

Author: hujc7

scripts/tools/test/test_train_and_publish_checkpoints.py

@@ -120,6 +120,37 @@ def test_legacy_collection_preserves_task_directory(tmp_path: Path) -> None:
     assert path == str(tmp_path / "rsl_rl" / "Isaac-Test" / "checkpoint.pt")
 
 
+def test_core_job_needs_the_completion_marker_to_count_as_trained(
+    tmp_path: Path, monkeypatch: pytest.MonkeyPatch
+) -> None:
+    """A run killed after writing a checkpoint must be retried, not skipped."""
+    monkeypatch.chdir(tmp_path)
+    job = CheckpointJob(
+        workflow="rsl_rl", task_name="Isaac-Test", physics_backend="newtonmjwarp", render_backend="none"
+    )
+    run_dir = Path(job.log_root) / "2026-01-01_00-00-00"
+    run_dir.mkdir(parents=True)
+    (run_dir / "model_0.pt").write_bytes(b"")
+
+    assert job.has_finished
+    assert not job.is_trained
+
+    job.mark_trained()
+
+    assert job.is_trained
+
+
+def test_legacy_job_counts_as_trained_without_the_marker(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
+    """Legacy runs predate the marker, so requiring one would retrain every published checkpoint."""
+    monkeypatch.chdir(tmp_path)
+    job = CheckpointJob(workflow="rsl_rl", task_name="Isaac-Test")
+    run_dir = Path(job.log_root) / "2026-01-01_00-00-00"
+    run_dir.mkdir(parents=True)
+    (run_dir / "model_0.pt").write_bytes(b"")
+
+    assert job.is_trained
+
+
 def test_publish_uses_collected_checkpoint_without_training_logs(
     tmp_path: Path,
     capsys: pytest.CaptureFixture[str],

scripts/tools/train_and_publish_checkpoints.py

@@ -111,6 +111,29 @@ def job_id(self) -> str:
             return f"{self.workflow}:{self.task_name}"
         return f"{self.workflow}:{self.task_name}:{self.physics_backend}:{self.render_backend}"
 
+    @property
+    def is_trained(self) -> bool:
+        """Whether this job's training finished and left a checkpoint.
+
+        A core job must also carry the marker its training subprocess writes, so a run killed
+        after writing a checkpoint is retried. Legacy jobs predate the marker.
+        """
+        if self.is_legacy:
+            return self.has_finished
+        run_path = self.latest_run
+        if run_path is None or not os.path.isfile(os.path.join(run_path, _TRAINING_COMPLETE_FILENAME)):
+            return False
+        return self.has_finished
+
+    def mark_trained(self) -> None:
+        """Record that the latest training subprocess exited successfully."""
+        run_path = self.latest_run
+        if run_path is None:
+            raise RuntimeError(f"Unable to determine the latest run for {self.job_id}")
+        marker_path = os.path.join(run_path, _TRAINING_COMPLETE_FILENAME)
+        with open(marker_path, "w", encoding="utf-8") as marker_file:
+            marker_file.write(f"{self.job_id}\n")
+
     @property
     def preset_args(self) -> list[str]:
         """Return typed preset selectors for this job."""
@@ -437,27 +460,9 @@ def _run_command(command: list[str], dry_run: bool) -> int:
     return subprocess.run(command, check=False, cwd=_REPO_ROOT, env=env).returncode
 
 
-def _has_training_job_completed(job: CheckpointJob) -> bool:
-    """Return whether the latest run exited successfully with a checkpoint."""
-    run_path = job.latest_run
-    if run_path is None or not os.path.isfile(os.path.join(run_path, _TRAINING_COMPLETE_FILENAME)):
-        return False
-    return job.has_finished
-
-
-def _mark_training_job_completed(job: CheckpointJob) -> None:
-    """Record that the latest training subprocess exited successfully."""
-    run_path = job.latest_run
-    if run_path is None:
-        raise RuntimeError(f"Unable to determine the latest run for {job.job_id}")
-    marker_path = os.path.join(run_path, _TRAINING_COMPLETE_FILENAME)
-    with open(marker_path, "w", encoding="utf-8") as marker_file:
-        marker_file.write(f"{job.job_id}\n")
-
-
 def train_job(job: CheckpointJob, args: argparse.Namespace, smoke: bool = False) -> bool:
     """Train or smoke-test one checkpoint job."""
-    if not smoke and not args.force and _has_training_job_completed(job):
+    if not smoke and not args.force and job.is_trained:
         print(f"Skipping completed training job {job.job_id}")
         return True
 
@@ -470,7 +475,7 @@ def train_job(job: CheckpointJob, args: argparse.Namespace, smoke: bool = False)
     if not job.has_finished:
         print(f"Training did not produce a checkpoint for {job.job_id}", file=sys.stderr)
         return False
-    _mark_training_job_completed(job)
+    job.mark_trained()
     return True
 
 
@@ -490,13 +495,13 @@ def collect_pretrained_checkpoint(job: CheckpointJob, output_dir: str, dry_run:
     os.makedirs(os.path.dirname(destination), exist_ok=True)
     shutil.copy2(source_path, destination)
     for checkpoint in job.checkpoints:
-        aux_source = job.trained_path(checkpoint)
-        if aux_source is None:
+        declared_source = job.trained_path(checkpoint)
+        if declared_source is None:
             print(f"No {checkpoint.name} checkpoint matched {checkpoint.run_glob!r} for {job.job_id}")
             continue
-        aux_destination = job.collected_path(output_dir, checkpoint)
-        print(f"Collecting {aux_source} -> {aux_destination}")
-        shutil.copy2(aux_source, aux_destination)
+        declared_destination = job.collected_path(output_dir, checkpoint)
+        print(f"Collecting {declared_source} -> {declared_destination}")
+        shutil.copy2(declared_source, declared_destination)
     return destination
 
 
@@ -582,7 +587,7 @@ def publish_pretrained_checkpoint(job: CheckpointJob, args: argparse.Namespace)
 def _summary_row(job: CheckpointJob, output_dir: str) -> list[str | bool]:
     """Return one CSV summary row."""
     has_run = job.has_run
-    has_finished = _has_training_job_completed(job) if job.physics_backend is not None else job.has_finished
+    has_finished = job.is_trained
     collected_path = job.collected_path(output_dir)
     review = job.review
     return [