Make mini-train split symlink setup retry-safe

src/agents/steps/data_split.py

@@ -15,6 +15,7 @@
 from runtime.filesystem import (
     chown_tree_to_root,
     create_absolute_symlink,
+    remove_path,
     rewrite_symlinks_to_absolute,
     validate_symlinks_targets_in,
 )
@@ -197,7 +198,9 @@ async def validate_split_dataset(ctx: RunContext[dict], result: AgenticStepOutpu
             if is_mini_train_only:
                 step_dir = self.config.current_step_dir
                 for split_name in [TRAIN_SPLIT, VALIDATION_SPLIT]:
-                    create_absolute_symlink(self.config.dataset_dir / split_name, step_dir / split_name)
+                    split_link = step_dir / split_name
+                    remove_path(split_link)
+                    create_absolute_symlink(self.config.dataset_dir / split_name, split_link)
                 result.train_path = str(step_dir / TRAIN_SPLIT)
                 result.val_path = str(step_dir / VALIDATION_SPLIT)
                 train_path = Path(result.train_path)
@@ -206,7 +209,8 @@ async def validate_split_dataset(ctx: RunContext[dict], result: AgenticStepOutpu
             if train_path.parent != val_path.parent or train_path.parent != mini_train_path.parent:
                 raise ModelRetry("Train, validation, and mini_train split folders must be in the same directory.")
             if not train_path.is_relative_to(self.config.splits_dir) or not val_path.is_relative_to(self.config.splits_dir) or not mini_train_path.is_relative_to(self.config.splits_dir):
-                for split_path in [train_path, val_path, mini_train_path]:
+                split_paths_to_validate = [mini_train_path] if is_mini_train_only else [train_path, val_path, mini_train_path]
+                for split_path in split_paths_to_validate:
                     try:
                         validate_symlinks_targets_in(split_path, self.config.dataset_dir)
                     except ValueError as e:
@@ -358,4 +362,3 @@ def on_iteration_end(self, iteration: int) -> None:
             is_allowed=bool(iteration_state["full_split_allowed_at_start"]),
         )
         #TODO log if split has changed?
-

src/runtime/filesystem.py

@@ -15,7 +15,25 @@ def remove_path(path: Path) -> None:
 
 def create_absolute_symlink(target: Path, link: Path) -> None:
     """Create a symlink at link pointing at the resolved absolute path of target."""
-    os.symlink(Path(target).resolve(), link)
+    absolute_target = Path(target).resolve()
+    link.parent.mkdir(parents=True, exist_ok=True)
+
+    try:
+        os.symlink(absolute_target, link)
+        return
+    except FileExistsError:
+        pass
+
+    if not link.is_symlink():
+        raise FileExistsError(
+            f"Cannot create symlink {link} -> {absolute_target}: path exists and is not a symlink"
+        )
+
+    if link.resolve(strict=False) == absolute_target:
+        return
+
+    link.unlink()
+    os.symlink(absolute_target, link)
 
 def find_symlinks_in_dir(root: Path, include_root: bool = False) -> list[Path]:
     root = Path(root)