diff --git a/packages/nmp_customization_common/src/nmp/customization_common/service/context.py b/packages/nmp_customization_common/src/nmp/customization_common/service/context.py
index adad780d50..29068f05d0 100644
--- a/packages/nmp_customization_common/src/nmp/customization_common/service/context.py
+++ b/packages/nmp_customization_common/src/nmp/customization_common/service/context.py
@@ -14,20 +14,20 @@
 
 from nmp.common.entities.constants import DEFAULT_WORKSPACE
 from nmp.common.jobs.constants import (
-    DEFAULT_NEMO_JOB_STEP_CONFIG_FILE_PATH,
     NEMO_JOB_ATTEMPT_ID_ENVVAR,
     NEMO_JOB_ID_ENVVAR,
-    NEMO_JOB_STEP_CONFIG_FILE_PATH_ENVVAR,
     NEMO_JOB_STEP_ENVVAR,
     NEMO_JOB_TASK_ENVVAR,
     NEMO_JOB_WORKSPACE_ENVVAR,
-    PERSISTENT_JOB_STORAGE_PATH_ENVVAR,
 )
 from nmp.customization_common.service.constants import (
-    DEFAULT_JOB_STORAGE_PATH,
     NMP_FILES_URL_ENVVAR,
     NMP_JOBS_URL_ENVVAR,
 )
+from nmp.customization_common.service.path_utils import (
+    get_job_step_config_path_from_env,
+    get_job_storage_path_from_env,
+)
 
 DEFAULT_JOB_ID = "unknown-job-id"
 DEFAULT_ATTEMPT_ID = "attempt-0"
@@ -97,8 +97,6 @@ def from_env(cls) -> Self:
             task=os.environ.get(NEMO_JOB_TASK_ENVVAR, DEFAULT_TASK),
             jobs_url=os.environ.get(NMP_JOBS_URL_ENVVAR),
             files_url=os.environ.get(NMP_FILES_URL_ENVVAR),
-            storage_path=Path(os.environ.get(PERSISTENT_JOB_STORAGE_PATH_ENVVAR, DEFAULT_JOB_STORAGE_PATH)),
-            config_path=Path(
-                os.environ.get(NEMO_JOB_STEP_CONFIG_FILE_PATH_ENVVAR, DEFAULT_NEMO_JOB_STEP_CONFIG_FILE_PATH)
-            ),
+            storage_path=get_job_storage_path_from_env(),
+            config_path=get_job_step_config_path_from_env(),
         )
diff --git a/packages/nmp_customization_common/src/nmp/customization_common/service/path_utils.py b/packages/nmp_customization_common/src/nmp/customization_common/service/path_utils.py
new file mode 100644
index 0000000000..9906f276dd
--- /dev/null
+++ b/packages/nmp_customization_common/src/nmp/customization_common/service/path_utils.py
@@ -0,0 +1,93 @@
+# SPDX-FileCopyrightText: Copyright (c) 2025-2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved.
+# SPDX-License-Identifier: Apache-2.0
+
+"""Path helpers for customization job storage mounts."""
+
+import os
+from pathlib import Path
+
+from nmp.common.jobs.constants import (
+    DEFAULT_JOB_STORAGE_PATH,
+    DEFAULT_NEMO_JOB_STEP_CONFIG_FILE_PATH,
+    NEMO_JOB_STEP_CONFIG_FILE_PATH_ENVVAR,
+    PERSISTENT_JOB_STORAGE_PATH_ENVVAR,
+)
+
+CURRENT_PERSISTENT_JOB_STORAGE_PATH_ENVVAR = "NEMO_JOB_PERSISTENT_JOB_STORAGE_PATH"
+LEGACY_PERSISTENT_JOB_STORAGE_PATH_ENVVARS = ("NMP_JOB_PERSISTENT_JOB_STORAGE_PATH",)
+
+CURRENT_NEMO_JOB_STEP_CONFIG_FILE_PATH_ENVVAR = "NEMO_JOB_STEP_CONFIG_FILE_PATH"
+LEGACY_NEMO_JOB_STEP_CONFIG_FILE_PATH_ENVVARS = ("NMP_JOB_STEP_CONFIG_FILE_PATH",)
+
+CURRENT_JOB_STORAGE_PATH = Path("/var/run/scratch/job")
+LEGACY_JOB_STORAGE_PATHS = (Path("/run/scratch/job"),)
+
+
+def _first_env_path(names: tuple[str, ...], default: str) -> Path:
+    for name in names:
+        value = os.environ.get(name)
+        if value:
+            return Path(value)
+    return Path(default)
+
+
+def get_job_storage_path_from_env() -> Path:
+    """Return the mounted persistent job storage path.
+
+    The explicit current env name is checked before the imported constant so a
+    task image still works if another installed package has an older constant.
+    """
+
+    return _first_env_path(
+        (
+            CURRENT_PERSISTENT_JOB_STORAGE_PATH_ENVVAR,
+            PERSISTENT_JOB_STORAGE_PATH_ENVVAR,
+            *LEGACY_PERSISTENT_JOB_STORAGE_PATH_ENVVARS,
+        ),
+        DEFAULT_JOB_STORAGE_PATH,
+    )
+
+
+def get_job_step_config_path_from_env() -> Path:
+    """Return the mounted step config file path."""
+
+    return _first_env_path(
+        (
+            CURRENT_NEMO_JOB_STEP_CONFIG_FILE_PATH_ENVVAR,
+            NEMO_JOB_STEP_CONFIG_FILE_PATH_ENVVAR,
+            *LEGACY_NEMO_JOB_STEP_CONFIG_FILE_PATH_ENVVARS,
+        ),
+        DEFAULT_NEMO_JOB_STEP_CONFIG_FILE_PATH,
+    )
+
+
+def _known_job_storage_roots(storage_path: Path) -> tuple[Path, ...]:
+    roots = (
+        storage_path,
+        Path(DEFAULT_JOB_STORAGE_PATH),
+        CURRENT_JOB_STORAGE_PATH,
+        *LEGACY_JOB_STORAGE_PATHS,
+    )
+    return tuple(dict.fromkeys(roots))
+
+
+def remap_job_storage_path(storage_path: Path, user_path: str | Path) -> Path:
+    """Remap known absolute job-storage roots to the mounted storage path.
+
+    Older customizer task configs used ``/run/scratch/job`` while the current
+    Jobs runner mounts persistent storage at ``/var/run/scratch/job``. This
+    keeps absolute paths produced by either side usable when API and task images
+    are briefly out of sync. Unknown absolute paths are returned unchanged so
+    callers can reject them with their normal traversal checks.
+    """
+
+    raw_path = Path(user_path)
+    if not raw_path.is_absolute():
+        return raw_path
+
+    for root in _known_job_storage_roots(storage_path):
+        try:
+            return storage_path / raw_path.relative_to(root)
+        except ValueError:
+            continue
+    return raw_path
diff --git a/packages/nmp_customization_common/src/nmp/customization_common/tasks/file_io_utils.py b/packages/nmp_customization_common/src/nmp/customization_common/tasks/file_io_utils.py
index 32331342a2..d0f83e80a1 100644
--- a/packages/nmp_customization_common/src/nmp/customization_common/tasks/file_io_utils.py
+++ b/packages/nmp_customization_common/src/nmp/customization_common/tasks/file_io_utils.py
@@ -28,6 +28,7 @@
     PathTraversalError,
     ProgressReportError,
 )
+from nmp.customization_common.service.path_utils import remap_job_storage_path
 
 logger = logging.getLogger(__name__)
 
@@ -166,7 +167,10 @@ def validate_safe_path(base_path: Path, user_path: str) -> Path:
         PathTraversalError: If the resolved path would escape base_path.
     """
     resolved_base = base_path.resolve()
-    resolved_path = (base_path / user_path).resolve()
+    candidate_path = remap_job_storage_path(base_path, user_path)
+    if not candidate_path.is_absolute():
+        candidate_path = base_path / candidate_path
+    resolved_path = candidate_path.resolve()
 
     if not resolved_path.is_relative_to(resolved_base):
         raise PathTraversalError(
diff --git a/packages/nmp_customization_common/tests/tasks/test_file_io_utils.py b/packages/nmp_customization_common/tests/tasks/test_file_io_utils.py
index 7cdb6acf48..5e3b037598 100644
--- a/packages/nmp_customization_common/tests/tasks/test_file_io_utils.py
+++ b/packages/nmp_customization_common/tests/tasks/test_file_io_utils.py
@@ -3,7 +3,8 @@
 
 from pathlib import Path
 
-from nmp.customization_common.tasks.file_io_utils import list_local_files
+import pytest
+from nmp.customization_common.tasks.file_io_utils import list_local_files, validate_safe_path
 
 
 def test_list_local_files_skips_unreadable_entries(tmp_path: Path) -> None:
@@ -34,3 +35,23 @@ def test_list_local_files_single_file(tmp_path: Path) -> None:
     assert len(files) == 1
     assert files[0].path == "weights.bin"
     assert files[0].size == 3
+
+
+def test_validate_safe_path_remaps_legacy_job_storage_root(tmp_path: Path) -> None:
+    storage = tmp_path / "job"
+    target = storage / "output_model"
+    target.mkdir(parents=True)
+
+    resolved = validate_safe_path(storage, "/run/scratch/job/output_model")
+
+    assert resolved == target.resolve()
+
+
+def test_validate_safe_path_rejects_unknown_absolute_path(tmp_path: Path) -> None:
+    from nmp.customization_common.schemas.file_io import PathTraversalError
+
+    storage = tmp_path / "job"
+    storage.mkdir()
+
+    with pytest.raises(PathTraversalError):
+        validate_safe_path(storage, "/etc/passwd")
diff --git a/services/automodel/src/nmp/automodel/app/jobs/compiler.py b/services/automodel/src/nmp/automodel/app/jobs/compiler.py
index a872846a1c..b419cba445 100644
--- a/services/automodel/src/nmp/automodel/app/jobs/compiler.py
+++ b/services/automodel/src/nmp/automodel/app/jobs/compiler.py
@@ -381,9 +381,17 @@ async def platform_job_config_compiler(
     workspace: str,
     job_spec: CustomizationJobOutput,
     sdk: AsyncNeMoPlatform,
+    *,
+    job_name: str | None = None,
+    profile: str | None = None,
 ) -> PlatformJobSpec:
     """Compile canonical job spec into a four-step PlatformJobSpec."""
+    del job_name  # reserved for future scheduling decisions
     transformed_spec = job_spec
+    if profile is not None and transformed_spec.training.execution_profile is None:
+        transformed_spec = transformed_spec.model_copy(
+            update={"training": transformed_spec.training.model_copy(update={"execution_profile": profile})},
+        )
     logger.info("Compiling Automodel job to PlatformJobSpec: %s", transformed_spec.model_dump_json(indent=2))
 
     try:
@@ -394,6 +402,7 @@ async def platform_job_config_compiler(
     # output is a required field in CustomizationJobOutput
     cpu_resources = _get_cpu_resources()
     base_env = _get_base_environment()
+    task_profile = transformed_spec.training.execution_profile or config.default_training_execution_profile
 
     # Fetch the primary model entity
     me = await fetch_model_entity(transformed_spec.model, workspace, sdk)
@@ -447,17 +456,11 @@ async def platform_job_config_compiler(
     trust_remote_code = me.trust_remote_code or False
     model_entity_config = _build_model_entity_config(workspace, transformed_spec, trust_remote_code)
 
-    cpu_profile = (
-        transformed_spec.training.execution_profile
-        if transformed_spec.training.execution_profile is not None
-        else config.default_training_execution_profile
-    )
-
     steps = [
         # Step 1: Download model and dataset files from Files service
         PlatformJobStep(
             name="model-and-dataset-download",
-            executor=_cpu_tasks_executor(FILE_IO_TASK_COMMAND, cpu_resources, cpu_profile),
+            executor=_cpu_tasks_executor(FILE_IO_TASK_COMMAND, cpu_resources, task_profile),
             environment=base_env,
             config=file_io_download_config.model_dump(mode="json"),
         ),
@@ -471,14 +474,14 @@ async def platform_job_config_compiler(
         # Step 3: Upload customized model
         PlatformJobStep(
             name="model-upload",
-            executor=_cpu_tasks_executor(FILE_IO_TASK_COMMAND, cpu_resources, cpu_profile),
+            executor=_cpu_tasks_executor(FILE_IO_TASK_COMMAND, cpu_resources, task_profile),
             environment=base_env,
             config=file_io_upload_config.model_dump(mode="json"),
         ),
         # Step 4: Create model entity
         PlatformJobStep(
             name="model-entity-creation",
-            executor=_cpu_tasks_executor(MODEL_ENTITY_TASK_COMMAND, cpu_resources, cpu_profile),
+            executor=_cpu_tasks_executor(MODEL_ENTITY_TASK_COMMAND, cpu_resources, task_profile),
             environment=base_env,
             config=model_entity_config.model_dump(mode="json"),
         ),
diff --git a/services/automodel/src/nmp/automodel/compile.py b/services/automodel/src/nmp/automodel/compile.py
index 1e4e194136..04f0c42d7d 100644
--- a/services/automodel/src/nmp/automodel/compile.py
+++ b/services/automodel/src/nmp/automodel/compile.py
@@ -25,14 +25,12 @@ async def platform_job_config_compiler(
     """Compile Automodel job spec (plugin or legacy shape) to PlatformJobSpec."""
     if not isinstance(job_spec, CustomizationJobOutput):
         job_spec = automodel_spec_to_compiler_output(job_spec)
-    if profile and job_spec.training.execution_profile is None:
-        job_spec = job_spec.model_copy(
-            update={"training": job_spec.training.model_copy(update={"execution_profile": profile})},
-        )
     return await _compile_canonical(
         workspace,
         job_spec,
         sdk,
+        job_name=job_name,
+        profile=profile,
     )
 
 
diff --git a/services/automodel/src/nmp/automodel/tasks/training/runner.py b/services/automodel/src/nmp/automodel/tasks/training/runner.py
index c81988ea6f..49bcfbb07f 100644
--- a/services/automodel/src/nmp/automodel/tasks/training/runner.py
+++ b/services/automodel/src/nmp/automodel/tasks/training/runner.py
@@ -19,6 +19,7 @@
 import yaml
 from nmp.automodel.app.constants import DEFAULT_TRAINING_RESULT_FILE_NAME
 from nmp.customization_common.service.context import NMPJobContext
+from nmp.customization_common.service.path_utils import remap_job_storage_path
 
 from .backends.backend import AUTOMODEL_CONFIG_FILENAME, AutomodelBackend
 from .distributed import DistributedContext
@@ -116,7 +117,36 @@ def _get_barrier_dir(self) -> Path:
 
     def _load_config(self, config_path: Path) -> TrainingStepConfig:
         with open(config_path) as f:
-            return TrainingStepConfig.model_validate(json.load(f))
+            config = TrainingStepConfig.model_validate(json.load(f))
+        return self._normalize_storage_paths(config)
+
+    def _normalize_storage_paths(self, config: TrainingStepConfig) -> TrainingStepConfig:
+        storage_path = self._job_ctx.storage_path
+
+        model = config.model.model_copy(
+            update={"path": str(remap_job_storage_path(storage_path, config.model.path))},
+        )
+        dataset = config.dataset.model_copy(
+            update={"path": str(remap_job_storage_path(storage_path, config.dataset.path))},
+        )
+        training = config.training
+        if training.kd is not None:
+            teacher_model = training.kd.teacher_model.model_copy(
+                update={"path": str(remap_job_storage_path(storage_path, training.kd.teacher_model.path))},
+            )
+            training = training.model_copy(
+                update={"kd": training.kd.model_copy(update={"teacher_model": teacher_model})},
+            )
+
+        return config.model_copy(
+            update={
+                "model": model,
+                "dataset": dataset,
+                "training": training,
+                "workspace_path": str(remap_job_storage_path(storage_path, config.workspace_path)),
+                "output_path": str(remap_job_storage_path(storage_path, config.output_path)),
+            },
+        )
 
     def _get_library_config_path(self) -> Path:
         return self._workspace_path / AUTOMODEL_CONFIG_FILENAME
diff --git a/services/automodel/tests/tasks/training/test_runner.py b/services/automodel/tests/tasks/training/test_runner.py
new file mode 100644
index 0000000000..938ea86cc5
--- /dev/null
+++ b/services/automodel/tests/tasks/training/test_runner.py
@@ -0,0 +1,64 @@
+# SPDX-FileCopyrightText: Copyright (c) 2025-2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved.
+# SPDX-License-Identifier: Apache-2.0
+
+"""Unit tests for the Automodel training runner."""
+
+import sys
+from pathlib import Path
+from unittest.mock import MagicMock
+
+sys.modules["nemo_automodel"] = MagicMock()
+sys.modules["nemo_automodel._transformers"] = MagicMock()
+sys.modules["nemo_automodel._transformers.registry"] = MagicMock()
+
+from nmp.automodel.entities.values import TrainingType  # noqa: E402
+from nmp.automodel.tasks.training.runner import TrainingRunner  # noqa: E402
+from nmp.automodel.tasks.training.schemas import DistillationConfig, ModelConfig, TrainingStepConfig  # noqa: E402
+from nmp.customization_common.service.context import NMPJobContext  # noqa: E402
+
+
+def _job_context(storage_path: Path) -> NMPJobContext:
+    return NMPJobContext(
+        workspace="default",
+        job_id="job-1",
+        attempt_id="attempt-0",
+        step="training",
+        task="task-1",
+        jobs_url=None,
+        files_url=None,
+        storage_path=storage_path,
+        config_path=storage_path / "config.json",
+    )
+
+
+def _config() -> TrainingStepConfig:
+    return TrainingStepConfig(
+        model=ModelConfig(path="/run/scratch/job/model"),
+        dataset=TrainingStepConfig.DatasetConfig(path="/run/scratch/job/dataset"),
+        training=TrainingStepConfig.TrainingConfig(
+            training_type=TrainingType.DISTILLATION,
+            kd=DistillationConfig(teacher_model=ModelConfig(path="/run/scratch/job/teacher_model")),
+        ),
+        schedule=TrainingStepConfig.ScheduleConfig(),
+        batch=TrainingStepConfig.BatchConfig(),
+        optimizer=TrainingStepConfig.OptimizerConfig(),
+        parallelism=TrainingStepConfig.ParallelismConfig(),
+        output_model="trained-model",
+        workspace_path="/run/scratch/job/training",
+        output_path="/run/scratch/job/output_model",
+    )
+
+
+def test_normalizes_legacy_job_storage_paths_to_runtime_mount(tmp_path: Path) -> None:
+    storage = tmp_path / "job"
+    runner = TrainingRunner.__new__(TrainingRunner)
+    runner._job_ctx = _job_context(storage)
+
+    normalized = runner._normalize_storage_paths(_config())
+
+    assert normalized.model.path == str(storage / "model")
+    assert normalized.dataset.path == str(storage / "dataset")
+    assert normalized.training.kd is not None
+    assert normalized.training.kd.teacher_model.path == str(storage / "teacher_model")
+    assert normalized.workspace_path == str(storage / "training")
+    assert normalized.output_path == str(storage / "output_model")
diff --git a/services/automodel/tests/test_compiler.py b/services/automodel/tests/test_compiler.py
index b043624d2f..b617509e12 100644
--- a/services/automodel/tests/test_compiler.py
+++ b/services/automodel/tests/test_compiler.py
@@ -445,3 +445,15 @@ def _step_image(step: Any) -> str:
     assert _step_image(steps[1]) == get_training_image()
     assert _step_image(steps[2]) == get_tasks_image()
     assert _step_image(steps[3]) == get_tasks_image()
+
+
+@pytest.mark.asyncio
+async def test_platform_job_config_compiler_applies_profile_to_task_steps(mock_sdk, monkeypatch):
+    monkeypatch.setattr(
+        "nmp.automodel.app.jobs.compiler.fetch_model_entity",
+        AsyncMock(return_value=_make_mock_model_entity()),
+    )
+
+    spec = await platform_job_config_compiler(_make_job_output(), "default", mock_sdk, profile="custom-gpu")
+
+    assert [step.executor.profile for step in spec.steps] == ["custom-gpu"] * 4
diff --git a/services/automodel/tests/test_job_context.py b/services/automodel/tests/test_job_context.py
index 82c9c929d4..7a6a763925 100644
--- a/services/automodel/tests/test_job_context.py
+++ b/services/automodel/tests/test_job_context.py
@@ -25,10 +25,35 @@
     DEFAULT_TASK,
     NMPJobContext,
 )
+from nmp.customization_common.service.path_utils import (
+    CURRENT_NEMO_JOB_STEP_CONFIG_FILE_PATH_ENVVAR,
+    CURRENT_PERSISTENT_JOB_STORAGE_PATH_ENVVAR,
+    LEGACY_NEMO_JOB_STEP_CONFIG_FILE_PATH_ENVVARS,
+    LEGACY_PERSISTENT_JOB_STORAGE_PATH_ENVVARS,
+)
+
+PATH_ENVVARS = tuple(
+    dict.fromkeys(
+        (
+            CURRENT_PERSISTENT_JOB_STORAGE_PATH_ENVVAR,
+            PERSISTENT_JOB_STORAGE_PATH_ENVVAR,
+            *LEGACY_PERSISTENT_JOB_STORAGE_PATH_ENVVARS,
+            CURRENT_NEMO_JOB_STEP_CONFIG_FILE_PATH_ENVVAR,
+            NEMO_JOB_STEP_CONFIG_FILE_PATH_ENVVAR,
+            *LEGACY_NEMO_JOB_STEP_CONFIG_FILE_PATH_ENVVARS,
+        )
+    )
+)
+
+
+def _clear_path_env(monkeypatch: pytest.MonkeyPatch) -> None:
+    for var in PATH_ENVVARS:
+        monkeypatch.delenv(var, raising=False)
 
 
 class TestNMPJobContextFromEnv:
     def test_uses_defaults_when_env_vars_not_set(self, monkeypatch: pytest.MonkeyPatch) -> None:
+        _clear_path_env(monkeypatch)
         for var in (
             NEMO_JOB_WORKSPACE_ENVVAR,
             NEMO_JOB_ID_ENVVAR,
@@ -37,8 +62,6 @@ def test_uses_defaults_when_env_vars_not_set(self, monkeypatch: pytest.MonkeyPat
             NEMO_JOB_TASK_ENVVAR,
             NMP_JOBS_URL_ENVVAR,
             NMP_FILES_URL_ENVVAR,
-            PERSISTENT_JOB_STORAGE_PATH_ENVVAR,
-            NEMO_JOB_STEP_CONFIG_FILE_PATH_ENVVAR,
         ):
             monkeypatch.delenv(var, raising=False)
 
@@ -55,6 +78,7 @@ def test_uses_defaults_when_env_vars_not_set(self, monkeypatch: pytest.MonkeyPat
         assert ctx.config_path == Path(DEFAULT_NEMO_JOB_STEP_CONFIG_FILE_PATH)
 
     def test_uses_env_vars_when_set(self, monkeypatch: pytest.MonkeyPatch) -> None:
+        _clear_path_env(monkeypatch)
         monkeypatch.setenv(NEMO_JOB_WORKSPACE_ENVVAR, "test-workspace")
         monkeypatch.setenv(NEMO_JOB_ID_ENVVAR, "job-123")
         monkeypatch.setenv(NEMO_JOB_ATTEMPT_ID_ENVVAR, "attempt-5")
@@ -71,3 +95,31 @@ def test_uses_env_vars_when_set(self, monkeypatch: pytest.MonkeyPatch) -> None:
         assert ctx.job_id == "job-123"
         assert ctx.normalized_task == "task-train-model"
         assert ctx.jobs_url == "http://jobs.example.com"
+        assert ctx.files_url == "http://files.example.com"
+        assert ctx.storage_path == Path("/custom/storage")
+        assert ctx.config_path == Path("/custom/config.json")
+
+    def test_current_storage_env_wins_over_legacy(self, monkeypatch: pytest.MonkeyPatch) -> None:
+        _clear_path_env(monkeypatch)
+        monkeypatch.setenv(CURRENT_PERSISTENT_JOB_STORAGE_PATH_ENVVAR, "/var/run/scratch/job")
+        monkeypatch.setenv(LEGACY_PERSISTENT_JOB_STORAGE_PATH_ENVVARS[0], "/run/scratch/job")
+
+        ctx = NMPJobContext.from_env()
+
+        assert ctx.storage_path == Path("/var/run/scratch/job")
+
+    def test_uses_legacy_storage_env_when_current_env_missing(self, monkeypatch: pytest.MonkeyPatch) -> None:
+        _clear_path_env(monkeypatch)
+        monkeypatch.setenv(LEGACY_PERSISTENT_JOB_STORAGE_PATH_ENVVARS[0], "/run/scratch/job")
+
+        ctx = NMPJobContext.from_env()
+
+        assert ctx.storage_path == Path("/run/scratch/job")
+
+    def test_uses_legacy_config_env_when_current_env_missing(self, monkeypatch: pytest.MonkeyPatch) -> None:
+        _clear_path_env(monkeypatch)
+        monkeypatch.setenv(LEGACY_NEMO_JOB_STEP_CONFIG_FILE_PATH_ENVVARS[0], "/run/scratch/config/job.json")
+
+        ctx = NMPJobContext.from_env()
+
+        assert ctx.config_path == Path("/run/scratch/config/job.json")
diff --git a/services/core/jobs/src/nmp/core/jobs/app/dispatcher.py b/services/core/jobs/src/nmp/core/jobs/app/dispatcher.py
index bd96de0026..6f25f4562c 100644
--- a/services/core/jobs/src/nmp/core/jobs/app/dispatcher.py
+++ b/services/core/jobs/src/nmp/core/jobs/app/dispatcher.py
@@ -42,11 +42,13 @@
     PlatformJobSpec,
 )
 from nmp.core.jobs.entities import (
+    STEP_SPEC_NAME_CONFIG_KEY,
     PlatformJob,
     PlatformJobAttempt,
     PlatformJobResult,
     PlatformJobStep,
     PlatformJobTask,
+    get_step_spec_name,
 )
 from opentelemetry import metrics, trace
 
@@ -615,7 +617,7 @@ async def _start_attempt(self, job: PlatformJob, attempt: PlatformJobAttempt) ->
         # With parent-scoped uniqueness, step names are unique per attempt (parent)
         # Store original spec name in config for reference
         step_config = dict(first_step.config) if first_step.config else {}
-        step_config["_step_spec_name"] = first_step.name
+        step_config[STEP_SPEC_NAME_CONFIG_KEY] = first_step.name
         await self.store.create(
             PlatformJobStep(
                 name=first_step.name,  # Simple name, unique per attempt via parent-scoped uniqueness
@@ -1009,7 +1011,7 @@ async def _update_job_status_from_step_locked(
         new_attempt_status = attempt.status
 
         # Get the original step spec name from config (step entity names have suffixes for uniqueness)
-        step_spec_name = saved_step.config.get("_step_spec_name", saved_step.name)
+        step_spec_name = get_step_spec_name(saved_step.config, saved_step.name) or saved_step.name
 
         if (
             saved_step.status == PlatformJobStatus.PENDING
@@ -1043,7 +1045,7 @@ async def _update_job_status_from_step_locked(
             if next_step:
                 # With parent-scoped uniqueness, use simple step name (unique per attempt)
                 next_step_config = dict(next_step.config) if next_step.config else {}
-                next_step_config["_step_spec_name"] = next_step.name
+                next_step_config[STEP_SPEC_NAME_CONFIG_KEY] = next_step.name
                 try:
                     await self.store.create(
                         PlatformJobStep(
diff --git a/services/core/jobs/src/nmp/core/jobs/controllers/backends/base.py b/services/core/jobs/src/nmp/core/jobs/controllers/backends/base.py
index a4ab60262c..992ae7edb4 100644
--- a/services/core/jobs/src/nmp/core/jobs/controllers/backends/base.py
+++ b/services/core/jobs/src/nmp/core/jobs/controllers/backends/base.py
@@ -46,6 +46,7 @@
 from nmp.common.platform_endpoint import parse_platform_endpoint
 from nmp.common.sdk_factory import get_entity_parts
 from nmp.core.jobs.app.providers import ComputeResources
+from nmp.core.jobs.entities import get_step_spec_name, is_final_platform_step
 from pydantic import BaseModel, model_validator
 
 logger = logging.getLogger(__name__)
@@ -417,6 +418,55 @@ def check_job_is_terminal(self, job: str, workspace: str) -> bool:
         except Exception as e:
             raise RuntimeError(f"Failed to fetch job '{workspace}/{job}' to check if terminal") from e
 
+    def check_job_persistent_storage_cleanup_allowed(self, job: str, step_name: str, workspace: str) -> bool:
+        """Return whether a completed step may delete persistent job storage.
+
+        The jobs controller owns aggregate terminal-state transitions. Backend cleanup only checks that the
+        successful resource belongs to the final configured step before deleting storage shared by the attempt.
+        """
+        try:
+            job_response = self._jobs.get_job(name=job, workspace=workspace).data()
+        except ClientNotFoundError:
+            # If the job entity is gone (e.g. workspace deletion), allow backend cleanup to reclaim storage.
+            return True
+        except Exception as e:
+            raise RuntimeError(f"Failed to fetch job '{workspace}/{job}' to check storage cleanup eligibility") from e
+
+        job_status = getattr(job_response.status, "value", job_response.status)
+        if job_status not in ("cancelled", "error", "completed"):
+            return False
+
+        try:
+            step = self.get_step(job=job, step_name=step_name, workspace=workspace)
+        except ClientNotFoundError:
+            return True
+        except Exception as e:
+            raise RuntimeError(
+                f"Could not fetch job step '{job}/{step_name}' to check storage cleanup eligibility"
+            ) from e
+
+        step_spec_name = get_step_spec_name(step.config, getattr(step, "name", None) or step_name)
+
+        platform_spec = getattr(job_response, "platform_spec", None)
+        steps = getattr(platform_spec, "steps", None) or []
+        if not steps:
+            return True
+
+        if is_final_platform_step(platform_spec, step_spec_name):
+            return True
+
+        logger.debug(
+            "Skipping persistent storage cleanup because completed step is not the final job step",
+            extra={
+                "workspace": workspace,
+                "job": job,
+                "step": step_name,
+                "step_spec_name": step_spec_name,
+                "final_step_name": getattr(steps[-1], "name", None),
+            },
+        )
+        return False
+
     def check_step_ttl(self, step: PlatformJobStepWithContext, ttl_seconds: int) -> bool:
         # Ensure created_at is timezone-aware (assume UTC if naive)
         if step.created_at is None:
diff --git a/services/core/jobs/src/nmp/core/jobs/controllers/backends/docker.py b/services/core/jobs/src/nmp/core/jobs/controllers/backends/docker.py
index 53edc833c6..0e5e2aff73 100644
--- a/services/core/jobs/src/nmp/core/jobs/controllers/backends/docker.py
+++ b/services/core/jobs/src/nmp/core/jobs/controllers/backends/docker.py
@@ -2185,16 +2185,16 @@ def cleanup_single_container(self, container: Container) -> None:
                 self.get_label_from_container(container, JOB_USES_PERSISTENT_STORAGE_LABEL) == "true"
             )
             if uses_persistent_storage and exit_code == 0:
-                # Verify the job is in a terminal state before cleaning up persistent storage
-                if self.check_job_is_terminal(job=job, workspace=workspace):
+                step_name = self.get_label_from_container(container, JOB_STEP_NAME_LABEL)
+                if self.check_job_persistent_storage_cleanup_allowed(job=job, step_name=step_name, workspace=workspace):
                     logger.debug(
                         "Cleaning up persistent storage for successful job", extra={"workspace": workspace, "job": job}
                     )
                     self.cleanup_job_persistent_storage(workspace, job)
                 else:
                     logger.debug(
-                        "Skipping persistent storage cleanup for job because it is not in terminal state yet",
-                        extra={"workspace": workspace, "job": job},
+                        "Skipping persistent storage cleanup for job",
+                        extra={"workspace": workspace, "job": job, "step": step_name},
                     )
 
     @abstractmethod
diff --git a/services/core/jobs/src/nmp/core/jobs/controllers/backends/kubernetes/kubernetes_job.py b/services/core/jobs/src/nmp/core/jobs/controllers/backends/kubernetes/kubernetes_job.py
index f74d11d322..fc12bdd65e 100644
--- a/services/core/jobs/src/nmp/core/jobs/controllers/backends/kubernetes/kubernetes_job.py
+++ b/services/core/jobs/src/nmp/core/jobs/controllers/backends/kubernetes/kubernetes_job.py
@@ -591,8 +591,9 @@ def cleanup_steps(self):
 
                 uses_persistent_storage = job.metadata.labels.get(JOB_USES_PERSISTENT_STORAGE_LABEL) == "true"
                 if uses_persistent_storage and self._execution_profile_config.storage:
-                    # Verify the job is in a terminal state before cleaning up persistent storage
-                    if self.check_job_is_terminal(job=job_id, workspace=workspace_id):
+                    if self.check_job_persistent_storage_cleanup_allowed(
+                        job=job_id, step_name=step_name, workspace=workspace_id
+                    ):
                         logger.info(
                             "Cleaning up persistent storage for successful job",
                             extra={
@@ -616,10 +617,11 @@ def cleanup_steps(self):
                         )
                     else:
                         logger.debug(
-                            "Skipping persistent storage cleanup for job as job is not in terminal state yet",
+                            "Skipping persistent storage cleanup for job",
                             extra={
                                 "workspace_id": workspace_id,
                                 "job_id": job_id,
+                                "step_name": step_name,
                             },
                         )
 
diff --git a/services/core/jobs/src/nmp/core/jobs/controllers/backends/kubernetes/volcano_job.py b/services/core/jobs/src/nmp/core/jobs/controllers/backends/kubernetes/volcano_job.py
index 545dd70850..2820a15e31 100644
--- a/services/core/jobs/src/nmp/core/jobs/controllers/backends/kubernetes/volcano_job.py
+++ b/services/core/jobs/src/nmp/core/jobs/controllers/backends/kubernetes/volcano_job.py
@@ -456,8 +456,9 @@ def cleanup_steps(self):
                     job.get("metadata", {}).get("labels", {}).get(JOB_USES_PERSISTENT_STORAGE_LABEL) == "true"
                 )
                 if uses_persistent_storage and self._execution_profile_config.storage:
-                    # Verify the job is in a terminal state before cleaning up persistent storage
-                    if self.check_job_is_terminal(job=job_id, workspace=workspace_id):
+                    if self.check_job_persistent_storage_cleanup_allowed(
+                        job=job_id, step_name=step_name, workspace=workspace_id
+                    ):
                         logger.info(
                             "Cleaning up persistent storage for successful job",
                             extra={"workspace_id": workspace_id, "job_id": job_id},
@@ -478,8 +479,8 @@ def cleanup_steps(self):
                         )
                     else:
                         logger.debug(
-                            "Skipping persistent storage cleanup for job as job is not in terminal state yet",
-                            extra={"workspace_id": workspace_id, "job_id": job_id},
+                            "Skipping persistent storage cleanup for job",
+                            extra={"workspace_id": workspace_id, "job_id": job_id, "step_name": step_name},
                         )
 
                 logger.debug(
@@ -504,8 +505,9 @@ def cleanup_steps(self):
                             job.get("metadata", {}).get("labels", {}).get(JOB_USES_PERSISTENT_STORAGE_LABEL) == "true"
                         )
                         if uses_persistent_storage and self._execution_profile_config.storage:
-                            # Verify the job is in a terminal state before cleaning up persistent storage
-                            if self.check_job_is_terminal(job=job_id, workspace=workspace_id):
+                            if self.check_job_persistent_storage_cleanup_allowed(
+                                job=job_id, step_name=step_name, workspace=workspace_id
+                            ):
                                 logger.info(
                                     "Cleaning up persistent storage for successful volcano job",
                                     extra={"workspace_id": workspace_id, "job_id": job_id},
@@ -526,8 +528,8 @@ def cleanup_steps(self):
                                 )
                             else:
                                 logger.debug(
-                                    "Skipping persistent storage cleanup for volcano job as job is not in terminal state yet",
-                                    extra={"workspace_id": workspace_id, "job_id": job_id},
+                                    "Skipping persistent storage cleanup for volcano job",
+                                    extra={"workspace_id": workspace_id, "job_id": job_id, "step_name": step_name},
                                 )
 
                     logger.debug(
diff --git a/services/core/jobs/src/nmp/core/jobs/entities.py b/services/core/jobs/src/nmp/core/jobs/entities.py
index 5eae2f57a7..833b58bed7 100644
--- a/services/core/jobs/src/nmp/core/jobs/entities.py
+++ b/services/core/jobs/src/nmp/core/jobs/entities.py
@@ -7,6 +7,7 @@
 Jobs are identified by ID rather than by a unique name within a namespace.
 """
 
+from collections.abc import Mapping
 from typing import Any, ClassVar, Dict, Optional, Self
 
 from nmp.common.auth import AuthContext
@@ -15,6 +16,24 @@
 from nmp.core.jobs.app.schemas import PlatformJobSpec, PlatformJobStepSpec
 from pydantic import Field, PrivateAttr, computed_field, model_validator
 
+STEP_SPEC_NAME_CONFIG_KEY = "_step_spec_name"
+
+
+def get_step_spec_name(config: Mapping[str, Any] | None, fallback_name: str | None = None) -> str | None:
+    """Return the original platform step spec name stored on a step entity."""
+    if isinstance(config, Mapping):
+        step_spec_name = config.get(STEP_SPEC_NAME_CONFIG_KEY)
+        if isinstance(step_spec_name, str):
+            return step_spec_name
+    return fallback_name
+
+
+def is_final_platform_step(platform_spec: PlatformJobSpec, step_spec_name: str | None) -> bool:
+    """Return whether ``step_spec_name`` names the final configured step."""
+    if step_spec_name is None or not platform_spec.steps:
+        return False
+    return platform_spec.steps[-1].name == step_spec_name
+
 
 class PlatformJob(EntityBase):
     """Platform job storage entity.
@@ -102,6 +121,10 @@ def get_next_step_spec(self, current_step_name: str) -> Optional[PlatformJobStep
                 found_current = True
         return None
 
+    def is_final_step_spec(self, step_name: str | None) -> bool:
+        """Return whether ``step_name`` is the final configured step in this attempt."""
+        return is_final_platform_step(self.platform_spec, step_name)
+
 
 class PlatformJobStep(EntityBase):
     """A single step within an attempt.
diff --git a/services/core/jobs/tests/controllers/test_docker_backend.py b/services/core/jobs/tests/controllers/test_docker_backend.py
index 556186029a..83be8b2240 100644
--- a/services/core/jobs/tests/controllers/test_docker_backend.py
+++ b/services/core/jobs/tests/controllers/test_docker_backend.py
@@ -85,6 +85,7 @@
     SchedulingDeferred,
 )
 from nmp.core.jobs.controllers.backends.workload_tokens import WORKLOAD_DELEGATION_TTL_BUFFER_SECONDS
+from nmp.core.jobs.entities import STEP_SPEC_NAME_CONFIG_KEY
 from pydantic import ValidationError
 
 from services.core.jobs.tests.controllers.client_mocks import data_response
@@ -2307,10 +2308,8 @@ def test_cleanup_steps_by_ttl(docker_job, docker_client_mock, test_job_step, cle
     # Make an older timestamp (120 minutes ago) so TTL-based cleanup does trigger
     older_time = (datetime.datetime.now(datetime.UTC) - datetime.timedelta(minutes=120)).isoformat()
 
-    # Mock the check_step_is_terminal and check_job_is_terminal methods to return True for all containers
-    # This allows the cleanup to proceed
+    # Mock the step terminal check to allow container cleanup to proceed.
     docker_job.check_step_is_terminal = MagicMock(return_value=True)
-    docker_job.check_job_is_terminal = MagicMock(return_value=True)
 
     # Create mock container that exited normally (exit code 0)
     mock_container_success = MagicMock()
@@ -3156,8 +3155,8 @@ def test_docker_job_schedule_with_auth_context_sdk_model_none_groups(
     assert principal_data["groups"] == []  # Default factory kicks in, not None
 
 
-def test_cleanup_single_container_checks_job_terminal_before_persistent_storage_cleanup(docker_job, docker_client_mock):
-    """Test that cleanup_single_container only cleans up persistent storage when the job is terminal."""
+def test_cleanup_single_container_checks_storage_cleanup_allowed(docker_job, docker_client_mock):
+    """Test that cleanup_single_container only cleans up persistent storage when cleanup is allowed."""
     # Create a mock container with persistent storage label that exited successfully
     mock_container = MagicMock()
     mock_container.name = "test-container-success"
@@ -3168,6 +3167,7 @@ def test_cleanup_single_container_checks_job_terminal_before_persistent_storage_
     mock_container.labels = {
         JOB_WORKSPACE_ID_LABEL: "default",
         JOB_ID_LABEL: "test-job-id",
+        JOB_STEP_NAME_LABEL: "test-step",
         JOB_TASK_ID_LABEL: "task-success",
         JOB_USES_PERSISTENT_STORAGE_LABEL: "true",
         JOB_MANAGED_BY_LABEL: JOB_MANAGED_BY_JOBS_CONTROLLER,
@@ -3178,8 +3178,8 @@ def test_cleanup_single_container_checks_job_terminal_before_persistent_storage_
     mock_volume = MagicMock()
     docker_client_mock.volumes.get.return_value = mock_volume
 
-    # Test Case 1: Job is NOT in terminal state - should skip persistent storage cleanup
-    docker_job.check_job_is_terminal = MagicMock(return_value=False)
+    # Test Case 1: cleanup is NOT allowed - should skip persistent storage cleanup
+    docker_job.check_job_persistent_storage_cleanup_allowed = MagicMock(return_value=False)
     docker_job.cleanup_job_persistent_storage = MagicMock()
 
     docker_job.cleanup_single_container(mock_container)
@@ -3191,17 +3191,19 @@ def test_cleanup_single_container_checks_job_terminal_before_persistent_storage_
     # Verify persistent storage cleanup was NOT called
     docker_job.cleanup_job_persistent_storage.assert_not_called()
 
-    # Verify check_job_is_terminal was called
-    docker_job.check_job_is_terminal.assert_called_once_with(job="test-job-id", workspace="default")
+    # Verify persistent storage cleanup eligibility was checked
+    docker_job.check_job_persistent_storage_cleanup_allowed.assert_called_once_with(
+        job="test-job-id", step_name="test-step", workspace="default"
+    )
 
     # Reset mocks
     mock_container.remove.reset_mock()
     docker_client_mock.volumes.get.reset_mock()
     docker_job.cleanup_job_persistent_storage.reset_mock()
-    docker_job.check_job_is_terminal.reset_mock()
+    docker_job.check_job_persistent_storage_cleanup_allowed.reset_mock()
 
-    # Test Case 2: Job IS in terminal state - should proceed with persistent storage cleanup
-    docker_job.check_job_is_terminal = MagicMock(return_value=True)
+    # Test Case 2: cleanup is allowed - should proceed with persistent storage cleanup
+    docker_job.check_job_persistent_storage_cleanup_allowed = MagicMock(return_value=True)
 
     docker_job.cleanup_single_container(mock_container)
 
@@ -3212,8 +3214,56 @@ def test_cleanup_single_container_checks_job_terminal_before_persistent_storage_
     # Verify persistent storage cleanup WAS called
     docker_job.cleanup_job_persistent_storage.assert_called_once_with("default", "test-job-id")
 
-    # Verify check_job_is_terminal was called
-    docker_job.check_job_is_terminal.assert_called_once_with(job="test-job-id", workspace="default")
+    # Verify persistent storage cleanup eligibility was checked
+    docker_job.check_job_persistent_storage_cleanup_allowed.assert_called_once_with(
+        job="test-job-id", step_name="test-step", workspace="default"
+    )
+
+
+def test_persistent_storage_cleanup_rejects_non_final_step_when_job_is_terminal(docker_job):
+    """Keep shared job storage when a completed container belongs to an intermediate step."""
+    docker_job._jobs.get_job.return_value = data_response(
+        SimpleNamespace(
+            status=PlatformJobStatus.COMPLETED,
+            platform_spec=SimpleNamespace(
+                steps=[
+                    SimpleNamespace(name="download"),
+                    SimpleNamespace(name="training"),
+                ]
+            ),
+        )
+    )
+    docker_job._jobs.get_job_step.return_value = data_response(
+        SimpleNamespace(name="download-1", config={STEP_SPEC_NAME_CONFIG_KEY: "download"})
+    )
+
+    assert (
+        docker_job.check_job_persistent_storage_cleanup_allowed(
+            job="customizer-job", step_name="download-1", workspace="default"
+        )
+        is False
+    )
+
+
+def test_persistent_storage_cleanup_allows_final_step_when_job_is_terminal(docker_job):
+    docker_job._jobs.get_job.return_value = data_response(
+        SimpleNamespace(
+            status=PlatformJobStatus.COMPLETED,
+            platform_spec=SimpleNamespace(
+                steps=[
+                    SimpleNamespace(name="download"),
+                    SimpleNamespace(name="training"),
+                ]
+            ),
+        )
+    )
+    docker_job._jobs.get_job_step.return_value = data_response(
+        SimpleNamespace(name="training-1", config={STEP_SPEC_NAME_CONFIG_KEY: "training"})
+    )
+
+    assert docker_job.check_job_persistent_storage_cleanup_allowed(
+        job="customizer-job", step_name="training-1", workspace="default"
+    )
 
 
 def test_cleanup_single_container_without_persistent_storage_label(docker_job, docker_client_mock):
@@ -3238,7 +3288,7 @@ def test_cleanup_single_container_without_persistent_storage_label(docker_job, d
     mock_volume = MagicMock()
     docker_client_mock.volumes.get.return_value = mock_volume
 
-    docker_job.check_job_is_terminal = MagicMock()
+    docker_job.check_job_persistent_storage_cleanup_allowed = MagicMock()
     docker_job.cleanup_job_persistent_storage = MagicMock()
 
     docker_job.cleanup_single_container(mock_container)
@@ -3247,8 +3297,8 @@ def test_cleanup_single_container_without_persistent_storage_label(docker_job, d
     assert mock_container.remove.call_count == 1
     assert docker_client_mock.volumes.get.call_count == 3  # task storage + config + workload identity volumes
 
-    # Verify job terminal check was NOT called (no persistent storage to cleanup)
-    docker_job.check_job_is_terminal.assert_not_called()
+    # Verify persistent storage cleanup eligibility was NOT checked (no persistent storage to cleanup)
+    docker_job.check_job_persistent_storage_cleanup_allowed.assert_not_called()
 
     # Verify persistent storage cleanup was NOT called
     docker_job.cleanup_job_persistent_storage.assert_not_called()
@@ -3320,8 +3370,8 @@ def test_cleanup_single_container_step_terminal_but_job_has_more_steps(docker_jo
     # Mock check_step_is_terminal to return True (step 1 is complete)
     docker_job.check_step_is_terminal = MagicMock(return_value=True)
 
-    # Mock check_job_is_terminal to return False (job is not terminal - step 2 still needs to run)
-    docker_job.check_job_is_terminal = MagicMock(return_value=False)
+    # Mock persistent storage cleanup eligibility to return False (step 2 still needs to run)
+    docker_job.check_job_persistent_storage_cleanup_allowed = MagicMock(return_value=False)
 
     docker_job.cleanup_job_persistent_storage = MagicMock()
 
@@ -3332,8 +3382,10 @@ def test_cleanup_single_container_step_terminal_but_job_has_more_steps(docker_jo
     assert mock_container.remove.call_count == 1
     assert docker_client_mock.volumes.get.call_count == 3  # task storage + config + workload identity volumes
 
-    # Verify job terminal check WAS called (since container uses persistent storage)
-    docker_job.check_job_is_terminal.assert_called_once_with(job="multi-step-job", workspace="default")
+    # Verify persistent storage cleanup eligibility was checked
+    docker_job.check_job_persistent_storage_cleanup_allowed.assert_called_once_with(
+        job="multi-step-job", step_name="step1", workspace="default"
+    )
 
     # Verify persistent storage cleanup was NOT called (job is not terminal yet)
     docker_job.cleanup_job_persistent_storage.assert_not_called()
@@ -3363,8 +3415,8 @@ def check_step_side_effect(job, step_name, workspace):
 
     docker_job.check_step_is_terminal = MagicMock(side_effect=check_step_side_effect)
 
-    # Mock check_job_is_terminal to return False (job is not terminal - has more steps)
-    docker_job.check_job_is_terminal = MagicMock(return_value=False)
+    # Mock persistent storage cleanup eligibility to return False (job has more steps)
+    docker_job.check_job_persistent_storage_cleanup_allowed = MagicMock(return_value=False)
 
     # Create mock container for step 1 that exited successfully
     mock_container_step1 = MagicMock()
@@ -3436,8 +3488,10 @@ def check_step_side_effect(job, step_name, workspace):
     docker_client_mock.volumes.get.assert_any_call("task-config-default-multi-step-job-task-step1")
     docker_client_mock.volumes.get.assert_any_call("task-workload-identity-default-multi-step-job-task-step1")
 
-    # Verify job terminal check was called for step 1
-    docker_job.check_job_is_terminal.assert_called_once_with(job="multi-step-job", workspace="default")
+    # Verify persistent storage cleanup eligibility was checked for step 1
+    docker_job.check_job_persistent_storage_cleanup_allowed.assert_called_once_with(
+        job="multi-step-job", step_name="step1", workspace="default"
+    )
 
     # Verify persistent storage cleanup was NOT called
     # because the job is not terminal yet (step 2 still running)
diff --git a/services/core/jobs/tests/controllers/test_kubernetes_backend.py b/services/core/jobs/tests/controllers/test_kubernetes_backend.py
index 0dc627a1aa..e3f960ebed 100644
--- a/services/core/jobs/tests/controllers/test_kubernetes_backend.py
+++ b/services/core/jobs/tests/controllers/test_kubernetes_backend.py
@@ -2068,9 +2068,8 @@ def test_schedule_with_additional_volumes(kubernetes_job, cpu_execution_provider
 def test_cleanup_steps_by_ttl(kubernetes_job, cleanup_completed_jobs_immediately):
     kubernetes_job._execution_profile_config.cleanup_completed_jobs_immediately = cleanup_completed_jobs_immediately
 
-    # Both return True when terminal or when entity not found (404). Persistent storage cleanup uses check_job_is_terminal.
+    # Step cleanup proceeds for terminal steps or when the step entity is gone.
     kubernetes_job.check_step_is_terminal = MagicMock(return_value=True)
-    kubernetes_job.check_job_is_terminal = MagicMock(return_value=True)
 
     # Mock active job status
     mock_job_spec = MagicMock()
@@ -2623,8 +2622,8 @@ def check_step_side_effect(job, step_name, workspace):
 
     kubernetes_job.check_step_is_terminal = MagicMock(side_effect=check_step_side_effect)
 
-    # Mock check_job_is_terminal to return False (job is not terminal - has more steps)
-    kubernetes_job.check_job_is_terminal = MagicMock(return_value=False)
+    # Mock persistent storage cleanup eligibility to return False (job has more steps)
+    kubernetes_job.check_job_persistent_storage_cleanup_allowed = MagicMock(return_value=False)
 
     # Create mock Kubernetes job for step 1 that completed successfully
     mock_job_step1_spec = MagicMock()
@@ -2701,8 +2700,10 @@ def check_step_side_effect(job, step_name, workspace):
             job="multi-step-job", step_name="step1", workspace="default"
         )
 
-        # Verify job terminal check was called for step 1
-        kubernetes_job.check_job_is_terminal.assert_called_once_with(job="multi-step-job", workspace="default")
+        # Verify persistent storage cleanup eligibility was checked for step 1
+        kubernetes_job.check_job_persistent_storage_cleanup_allowed.assert_called_once_with(
+            job="multi-step-job", step_name="step1", workspace="default"
+        )
 
         # Verify persistent storage cleanup was NOT called
         # because the job is not terminal yet (step 2 still active)
@@ -2712,13 +2713,12 @@ def check_step_side_effect(job, step_name, workspace):
 def test_cleanup_steps_proceeds_when_entity_not_found(kubernetes_job):
     """When step/job entities are gone (e.g. workspace deleted) but the backend job is terminal and ours, still clean up.
 
-    check_step_is_terminal and check_job_is_terminal return True when terminal or when the entity is not found (404).
+    check_step_is_terminal returns True when terminal or when the entity is not found (404).
     """
     kubernetes_job._execution_profile_config.cleanup_completed_jobs_immediately = True
 
     # Simulate entity-not-found: both return True so cleanup proceeds
     kubernetes_job.check_step_is_terminal = MagicMock(return_value=True)
-    kubernetes_job.check_job_is_terminal = MagicMock(return_value=True)
 
     mock_job_spec = MagicMock()
     mock_job_spec.suspend = False
@@ -2751,7 +2751,7 @@ def test_cleanup_steps_proceeds_when_entity_not_found(kubernetes_job):
 def test_cleanup_steps_proceeds_when_job_entity_not_found_with_persistent_storage(kubernetes_job):
     """When job entity is not found (404) but backend job is completed and uses persistent storage, still run full cleanup.
 
-    check_job_is_terminal returns True when job is terminal or when job entity is not found (404).
+    check_job_persistent_storage_cleanup_allowed returns True when cleanup should proceed.
     """
     kubernetes_job._execution_profile_config.cleanup_completed_jobs_immediately = True
     kubernetes_job._execution_profile_config.storage = MagicMock()
@@ -2759,7 +2759,7 @@ def test_cleanup_steps_proceeds_when_job_entity_not_found_with_persistent_storag
     kubernetes_job._execution_profile_config.storage.volume_permissions_image = "busybox"
 
     kubernetes_job.check_step_is_terminal = MagicMock(return_value=True)
-    kubernetes_job.check_job_is_terminal = MagicMock(return_value=True)  # job entity 404 → True
+    kubernetes_job.check_job_persistent_storage_cleanup_allowed = MagicMock(return_value=True)
 
     mock_job_spec = MagicMock()
     mock_job_spec.suspend = False
diff --git a/services/core/jobs/tests/controllers/test_volcano_backend.py b/services/core/jobs/tests/controllers/test_volcano_backend.py
index c22e26daf6..a9897b5cff 100644
--- a/services/core/jobs/tests/controllers/test_volcano_backend.py
+++ b/services/core/jobs/tests/controllers/test_volcano_backend.py
@@ -1164,9 +1164,8 @@ def test_schedule_with_storage_integration(
 def test_cleanup_steps_by_ttl(volcano_job: VolcanoJobBackend, status):
     """Test job cleanup with one active, one recently completed, one harvestable completed."""
 
-    # Both return True when terminal or when entity not found (404). Persistent storage cleanup uses check_job_is_terminal.
+    # Step cleanup proceeds for terminal steps or when the step entity is gone.
     volcano_job.check_step_is_terminal = MagicMock(return_value=True)  # type: ignore[assignment]
-    volcano_job.check_job_is_terminal = MagicMock(return_value=True)  # type: ignore[assignment]
 
     status_key = status.lower()
     two_hours_ago = datetime.datetime.now(datetime.UTC) - datetime.timedelta(seconds=7200)
@@ -1434,8 +1433,8 @@ def check_step_side_effect(job, step_name, workspace):
 
     volcano_job.check_step_is_terminal = MagicMock(side_effect=check_step_side_effect)  # type: ignore[assignment]
 
-    # Mock check_job_is_terminal to return False (job is not terminal - has more steps)
-    volcano_job.check_job_is_terminal = MagicMock(return_value=False)  # type: ignore[assignment]
+    # Mock persistent storage cleanup eligibility to return False (job has more steps)
+    volcano_job.check_job_persistent_storage_cleanup_allowed = MagicMock(return_value=False)  # type: ignore[assignment]
 
     # Create mock Volcano job for step 1 that completed successfully
     two_minutes_ago = datetime.datetime.now(datetime.UTC) - datetime.timedelta(seconds=120)
@@ -1499,8 +1498,10 @@ def check_step_side_effect(job, step_name, workspace):
         volcano_job.check_step_is_terminal.assert_any_call(job="multi-step-job", step_name="step1", workspace="default")
         volcano_job.check_step_is_terminal.assert_any_call(job="multi-step-job", step_name="step2", workspace="default")
 
-        # Verify job terminal check was called once for step 1 (only for terminal steps with persistent storage)
-        volcano_job.check_job_is_terminal.assert_called_once_with(job="multi-step-job", workspace="default")
+        # Verify persistent storage cleanup eligibility was checked once for step 1
+        volcano_job.check_job_persistent_storage_cleanup_allowed.assert_called_once_with(
+            job="multi-step-job", step_name="step1", workspace="default"
+        )
 
         # Verify persistent storage cleanup was NOT called
         # because the job is not terminal yet (step 2 still active)
@@ -1510,13 +1511,12 @@ def check_step_side_effect(job, step_name, workspace):
 def test_cleanup_steps_proceeds_when_entity_not_found(volcano_job: VolcanoJobBackend):
     """When step/job entities are gone (e.g. workspace deleted) but the backend job is terminal and ours, still clean up.
 
-    check_step_is_terminal and check_job_is_terminal return True when terminal or when the entity is not found (404).
+    check_step_is_terminal returns True when terminal or when the entity is not found (404).
     """
     volcano_job._execution_profile_config.cleanup_completed_jobs_immediately = True
 
     # Simulate entity-not-found: both return True so cleanup proceeds
     volcano_job.check_step_is_terminal = MagicMock(return_value=True)  # type: ignore[assignment]
-    volcano_job.check_job_is_terminal = MagicMock(return_value=True)  # type: ignore[assignment]
 
     two_minutes_ago = datetime.datetime.now(datetime.UTC) - datetime.timedelta(seconds=120)
     mock_job = {
@@ -1548,7 +1548,7 @@ def test_cleanup_steps_proceeds_when_entity_not_found(volcano_job: VolcanoJobBac
 def test_cleanup_steps_proceeds_when_job_entity_not_found_with_persistent_storage(volcano_job: VolcanoJobBackend):
     """When job entity is not found (404) but backend job is completed and uses persistent storage, still run full cleanup.
 
-    check_job_is_terminal returns True when job is terminal or when job entity is not found (404).
+    check_job_persistent_storage_cleanup_allowed returns True when cleanup should proceed.
     """
     volcano_job._execution_profile_config.cleanup_completed_jobs_immediately = True
     volcano_job._execution_profile_config.storage = MagicMock()
@@ -1556,7 +1556,7 @@ def test_cleanup_steps_proceeds_when_job_entity_not_found_with_persistent_storag
     volcano_job._execution_profile_config.storage.volume_permissions_image = "busybox"
 
     volcano_job.check_step_is_terminal = MagicMock(return_value=True)  # type: ignore[assignment]
-    volcano_job.check_job_is_terminal = MagicMock(return_value=True)  # type: ignore[assignment]  # job entity 404 → True
+    volcano_job.check_job_persistent_storage_cleanup_allowed = MagicMock(return_value=True)  # type: ignore[assignment]
 
     two_minutes_ago = datetime.datetime.now(datetime.UTC) - datetime.timedelta(seconds=120)
     mock_job = {
diff --git a/services/core/jobs/tests/test_entities.py b/services/core/jobs/tests/test_entities.py
new file mode 100644
index 0000000000..5c7c75588a
--- /dev/null
+++ b/services/core/jobs/tests/test_entities.py
@@ -0,0 +1,46 @@
+# SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved.
+# SPDX-License-Identifier: Apache-2.0
+
+from typing import Any
+
+from nmp.common.jobs.schemas import PlatformJobStatus
+from nmp.core.jobs.app.schemas import PlatformJobStepSpec
+from nmp.core.jobs.app.test_helpers import TestConstants
+from nmp.core.jobs.entities import (
+    STEP_SPEC_NAME_CONFIG_KEY,
+    PlatformJobAttempt,
+    get_step_spec_name,
+)
+
+
+def test_get_step_spec_name_prefers_stored_spec_name():
+    assert get_step_spec_name({STEP_SPEC_NAME_CONFIG_KEY: "download"}, fallback_name="download-1") == "download"
+
+
+def test_get_step_spec_name_uses_fallback_without_stored_spec_name():
+    assert get_step_spec_name({}, fallback_name="download") == "download"
+
+
+def test_get_step_spec_name_uses_fallback_for_non_mapping_config():
+    config: Any = ["not", "a", "mapping"]
+
+    assert get_step_spec_name(config, fallback_name="download") == "download"
+
+
+def test_platform_job_attempt_identifies_final_step_spec():
+    platform_spec = TestConstants.PLATFORM_SPEC.model_copy(deep=True)
+    platform_spec.steps.append(
+        PlatformJobStepSpec(name="finalize", executor=TestConstants.TEST_EXECUTOR, config={}),
+    )
+    attempt = PlatformJobAttempt(
+        name="attempt-1",
+        workspace=TestConstants.WORKSPACE,
+        job="job-1",
+        seq=0,
+        status=PlatformJobStatus.ACTIVE,
+        spec=TestConstants.SPEC_BASIC,
+        platform_spec=platform_spec,
+    )
+
+    assert attempt.is_final_step_spec("basic") is False
+    assert attempt.is_final_step_spec("finalize") is True
diff --git a/services/unsloth/src/nmp/unsloth/app/jobs/compiler.py b/services/unsloth/src/nmp/unsloth/app/jobs/compiler.py
index c125f82160..3a294204f7 100644
--- a/services/unsloth/src/nmp/unsloth/app/jobs/compiler.py
+++ b/services/unsloth/src/nmp/unsloth/app/jobs/compiler.py
@@ -234,6 +234,7 @@ async def platform_job_config_compiler(
 
     cpu_resources = _get_cpu_resources()
     base_env = _get_base_environment()
+    task_profile = profile or config.default_training_execution_profile
 
     validation_dataset_path = _resolve_validation_dataset_path(job_spec, workspace=workspace)
     download_config = _build_file_download_config(job_spec, me, workspace=workspace)
@@ -249,6 +250,7 @@ async def platform_job_config_compiler(
             name="model-and-dataset-download",
             executor=CPUExecutionProviderSpec(
                 provider="cpu",
+                profile=task_profile,
                 container=ContainerSpec(
                     image=get_tasks_image(),
                     entrypoint=UNSLOTH_PYTHON_ENTRYPOINT,
@@ -263,12 +265,13 @@ async def platform_job_config_compiler(
             job_spec,
             base_env,
             validation_dataset_path=validation_dataset_path,
-            profile=profile,
+            profile=task_profile,
         ),
         PlatformJobStep(
             name="model-upload",
             executor=CPUExecutionProviderSpec(
                 provider="cpu",
+                profile=task_profile,
                 container=ContainerSpec(
                     image=get_tasks_image(),
                     entrypoint=UNSLOTH_PYTHON_ENTRYPOINT,
@@ -283,6 +286,7 @@ async def platform_job_config_compiler(
             name="model-entity-creation",
             executor=CPUExecutionProviderSpec(
                 provider="cpu",
+                profile=task_profile,
                 container=ContainerSpec(
                     image=get_tasks_image(),
                     entrypoint=UNSLOTH_PYTHON_ENTRYPOINT,
diff --git a/services/unsloth/src/nmp/unsloth/tasks/training/__main__.py b/services/unsloth/src/nmp/unsloth/tasks/training/__main__.py
index 924657ad92..07dd6ca9eb 100644
--- a/services/unsloth/src/nmp/unsloth/tasks/training/__main__.py
+++ b/services/unsloth/src/nmp/unsloth/tasks/training/__main__.py
@@ -65,17 +65,14 @@ def main() -> int:
     # Local imports so the parent process (e.g. CLI discovery, pytest
     # collection) does not pay the ML import cost.
     from nemo_platform_plugin.job_context import JobContext, StoragePaths
-    from nmp.common.jobs.constants import (
-        DEFAULT_JOB_STORAGE_PATH,
-        PERSISTENT_JOB_STORAGE_PATH_ENVVAR,
-    )
+    from nmp.customization_common.service.path_utils import get_job_storage_path_from_env, remap_job_storage_path
     from nmp.unsloth.app.jobs.training.schemas import TrainingStepConfig
     from nmp.unsloth.tasks.training.backends.unsloth_sft import train_sft
 
     config = TrainingStepConfig.model_validate(raw)
     spec = config.spec
 
-    persistent_root = Path(os.environ.get(PERSISTENT_JOB_STORAGE_PATH_ENVVAR, DEFAULT_JOB_STORAGE_PATH))
+    persistent_root = get_job_storage_path_from_env()
     storage = StoragePaths(
         ephemeral=persistent_root / "ephemeral",
         persistent=persistent_root,
@@ -113,14 +110,19 @@ def main() -> int:
         f"Container: UNSLOTH_COMPILE_LOCATION={os.environ['UNSLOTH_COMPILE_LOCATION']} HF_HOME={os.environ['HF_HOME']}"
     )
 
+    def _storage_path(value: str | None) -> str | None:
+        if value is None:
+            return None
+        return str(remap_job_storage_path(persistent_root, value))
+
     try:
         result = train_sft(
             spec,
             ctx,
-            model_path=config.model_path,
-            dataset_path=config.dataset_path,
-            validation_path=config.validation_path,
-            output_path=config.output_path,
+            model_path=_storage_path(config.model_path),
+            dataset_path=_storage_path(config.dataset_path),
+            validation_path=_storage_path(config.validation_path),
+            output_path=_storage_path(config.output_path),
         )
     except Exception:
         logger.exception("Unsloth training step failed")
diff --git a/services/unsloth/tests/test_compiler_validation_path.py b/services/unsloth/tests/test_compiler_validation_path.py
index 02241f5e1b..b91fee21d8 100644
--- a/services/unsloth/tests/test_compiler_validation_path.py
+++ b/services/unsloth/tests/test_compiler_validation_path.py
@@ -115,3 +115,22 @@ async def test_upload_step_stamps_output_metadata() -> None:
 
     upload = next(s for s in job["steps"] if s["name"] == "model-upload")
     assert upload["config"]["upload"][0]["metadata"] is None
+
+
+@pytest.mark.asyncio
+async def test_compiler_applies_profile_to_task_steps() -> None:
+    from nmp.unsloth.app.jobs import compiler as compiler_mod
+
+    original_fetch = compiler_mod.fetch_model_entity
+    compiler_mod.fetch_model_entity = AsyncMock(return_value=_model_entity())
+    try:
+        job = await platform_job_config_compiler(
+            workspace="default",
+            job_spec=_spec(validation_path=None),
+            sdk=MagicMock(),
+            profile="custom-gpu",
+        )
+    finally:
+        compiler_mod.fetch_model_entity = original_fetch
+
+    assert [step["executor"]["profile"] for step in job["steps"]] == ["custom-gpu"] * 4
diff --git a/services/unsloth/tests/test_main.py b/services/unsloth/tests/test_main.py
index 442fb3bae0..47dc88a6d3 100644
--- a/services/unsloth/tests/test_main.py
+++ b/services/unsloth/tests/test_main.py
@@ -92,6 +92,95 @@ def _stub_train_sft(*_args: object, **_kwargs: object) -> dict[str, object]:
         assert (ephemeral / "unsloth_compiled_cache").is_dir()
         assert (ephemeral / "hf").is_dir()
 
+    def test_remaps_legacy_storage_paths_to_runtime_mount(
+        self,
+        tmp_path: Path,
+        monkeypatch: pytest.MonkeyPatch,
+    ) -> None:
+        from nmp.common.jobs.constants import PERSISTENT_JOB_STORAGE_PATH_ENVVAR
+        from nmp.unsloth.tasks.training.backends import unsloth_sft
+
+        config = _step_config().model_copy(
+            update={
+                "model_path": "/run/scratch/job/model",
+                "dataset_path": "/run/scratch/job/dataset",
+                "validation_path": "/run/scratch/job/validation_dataset",
+                "output_path": "/run/scratch/job/output_model",
+            },
+        )
+        config_file = tmp_path / "step.json"
+        config_file.write_text(json.dumps(config.model_dump(mode="json")))
+
+        storage = tmp_path / "job"
+        monkeypatch.setenv("NEMO_JOB_STEP_CONFIG_FILE_PATH", str(config_file))
+        monkeypatch.setenv(PERSISTENT_JOB_STORAGE_PATH_ENVVAR, str(storage))
+        monkeypatch.delenv("UNSLOTH_COMPILE_LOCATION", raising=False)
+        monkeypatch.delenv("HF_HOME", raising=False)
+
+        captured: dict[str, str | None] = {}
+
+        def _stub_train_sft(*_args: object, **kwargs: object) -> dict[str, object]:
+            captured.update(
+                {
+                    "model_path": kwargs["model_path"],
+                    "dataset_path": kwargs["dataset_path"],
+                    "validation_path": kwargs["validation_path"],
+                    "output_path": kwargs["output_path"],
+                },
+            )
+            return {}
+
+        monkeypatch.setattr(unsloth_sft, "train_sft", _stub_train_sft)
+
+        rc = main()
+
+        assert rc == 0
+        assert captured == {
+            "model_path": str(storage / "model"),
+            "dataset_path": str(storage / "dataset"),
+            "validation_path": str(storage / "validation_dataset"),
+            "output_path": str(storage / "output_model"),
+        }
+
+    def test_uses_legacy_storage_env_for_runtime_mount(
+        self,
+        tmp_path: Path,
+        monkeypatch: pytest.MonkeyPatch,
+    ) -> None:
+        from nmp.common.jobs.constants import PERSISTENT_JOB_STORAGE_PATH_ENVVAR
+        from nmp.customization_common.service.path_utils import (
+            CURRENT_PERSISTENT_JOB_STORAGE_PATH_ENVVAR,
+            LEGACY_PERSISTENT_JOB_STORAGE_PATH_ENVVARS,
+        )
+        from nmp.unsloth.tasks.training.backends import unsloth_sft
+
+        config_file = tmp_path / "step.json"
+        config_file.write_text(json.dumps(_step_config().model_dump(mode="json")))
+
+        storage = tmp_path / "legacy-job"
+        monkeypatch.setenv("NEMO_JOB_STEP_CONFIG_FILE_PATH", str(config_file))
+        monkeypatch.delenv(CURRENT_PERSISTENT_JOB_STORAGE_PATH_ENVVAR, raising=False)
+        monkeypatch.delenv(PERSISTENT_JOB_STORAGE_PATH_ENVVAR, raising=False)
+        monkeypatch.setenv(LEGACY_PERSISTENT_JOB_STORAGE_PATH_ENVVARS[0], str(storage))
+        monkeypatch.delenv("UNSLOTH_COMPILE_LOCATION", raising=False)
+        monkeypatch.delenv("HF_HOME", raising=False)
+
+        captured: dict[str, str | None] = {}
+
+        def _stub_train_sft(*_args: object, **_kwargs: object) -> dict[str, object]:
+            captured["UNSLOTH_COMPILE_LOCATION"] = os.environ.get("UNSLOTH_COMPILE_LOCATION")
+            captured["HF_HOME"] = os.environ.get("HF_HOME")
+            return {}
+
+        monkeypatch.setattr(unsloth_sft, "train_sft", _stub_train_sft)
+
+        rc = main()
+
+        assert rc == 0
+        ephemeral = storage / "ephemeral"
+        assert captured["UNSLOTH_COMPILE_LOCATION"] == str(ephemeral / "unsloth_compiled_cache")
+        assert captured["HF_HOME"] == str(ephemeral / "hf")
+
 
 class TestEntrypointWithoutStepConfig:
     def test_returns_2_when_step_config_env_missing(