@@ -681,6 +681,107 @@ def parse_cli_args():
         default=None,
     )
 
+    # Nvcre
+    nvcre_args = parser.add_argument_group("Nvcre arguments")
+    nvcre_args.add_argument(
+        "--nvcre_namespace",
+        type=str,
+        help="Kubernetes namespace for Nvcre WorkloadRun. When set, uses the Nvcre executor instead of Slurm.",
+        required=False,
+    )
+    nvcre_args.add_argument(
+        "--nvcre_image_pull_secret",
+        type=str,
+        help="Kubernetes image pull secret name for pulling the container image.",
+        required=False,
+        default=None,
+    )
+    nvcre_args.add_argument(
+        "--nvcre_workdir_pvc",
+        type=str,
+        help="PVC name for syncing job workdir to the cluster before launch.",
+        required=False,
+        default=None,
+    )
+    nvcre_args.add_argument(
+        "--nvcre_workdir_pvc_path",
+        type=str,
+        help="Mount path for the workdir PVC inside the training pod.",
+        default="/nemo_run",
+        required=False,
+    )
+    nvcre_args.add_argument(
+        "--nvcre_workdir_local_path",
+        type=str,
+        help="Local directory rsynced into the workdir PVC before launch.",
+        required=False,
+        default=None,
+    )
+    nvcre_args.add_argument(
+        "--nvcre_node_selector_json",
+        type=str,
+        help="JSON-encoded dict of node selector labels for targeting specific nodes.",
+        required=False,
+        default=None,
+    )
+    nvcre_args.add_argument(
+        "--nvcre_volumes_json",
+        type=str,
+        help="JSON-encoded list of Kubernetes Volume dicts.",
+        required=False,
+        default=None,
+    )
+    nvcre_args.add_argument(
+        "--nvcre_volume_mounts_json",
+        type=str,
+        help="JSON-encoded list of Kubernetes VolumeMount dicts.",
+        required=False,
+        default=None,
+    )
+    nvcre_args.add_argument(
+        "--nvcre_timeout_per_job",
+        type=str,
+        help="Per-job timeout passed to Nvcre orchestration (e.g. '24h').",
+        required=False,
+        default="24h",
+    )
+
+    nvcre_args.add_argument(
+        "--nvcre_kubeconfig",
+        type=str,
+        help="Path to kubeconfig file for nvcrectl/kubectl.",
+        required=False,
+        default=None,
+    )
+    nvcre_args.add_argument(
+        "--nvcre_kube_context",
+        type=str,
+        help="Kubernetes context to use with nvcrectl/kubectl.",
+        required=False,
+        default=None,
+    )
+    nvcre_args.add_argument(
+        "--nvcre_gang_scheduler_name",
+        type=str,
+        help="Gang scheduler name for the WorkloadRun (e.g. 'kai-scheduler').",
+        required=False,
+        default=None,
+    )
+    nvcre_args.add_argument(
+        "--nvcre_hf_token_secret_name",
+        type=str,
+        help="Name of the Kubernetes Secret holding the HuggingFace token.",
+        required=False,
+        default=None,
+    )
+    nvcre_args.add_argument(
+        "--nvcre_hf_token_secret_key",
+        type=str,
+        help="Key inside the Kubernetes Secret that holds the HuggingFace token value.",
+        required=False,
+        default="token",
+    )
+
     # For performance
     performance_args = parser.add_argument_group("Performance arguments")
     performance_args.add_argument(
 
@@ -35,6 +35,12 @@
 from nemo_run import Plugin, Script, SlurmExecutor
 
 
+try:
+    from nemo_run.core.execution.nvcre import NvcreExecutor
+except ImportError:
+    NvcreExecutor = None  # type: ignore[assignment,misc]
+
+
 logger: logging.Logger = logging.getLogger(__name__)
 NSYS_SQLITE_EXPORT_ARG = "--export=sqlite"
 
@@ -185,6 +191,9 @@ def setup(self, task: Union["run.Partial", "run.Script"], executor: "run.Executo
         if isinstance(executor, SlurmExecutor):
             # NOTE: DO NOT change to f-string, `%q{}` is Slurm placeholder
             launcher.nsys_filename = "profile_%p_%q{SLURM_JOB_ID}_node%q{SLURM_NODEID}_rank%q{SLURM_PROCID}"
+        elif NvcreExecutor is not None and isinstance(executor, NvcreExecutor):
+            # Nvcre pods use PET_* env vars for distributed rank info.
+            launcher.nsys_filename = "profile_node%q{PET_NODE_RANK}_rank%p"
 
         if self.nsys_gpu_metrics:
             if hasattr(launcher, "nsys_gpu_metrics"):
 
@@ -88,22 +88,31 @@ def _filter_run_script_args(argv: List[str]) -> List[str]:
       carry JSON values whose ``{}`` / ``[]`` are brace/glob-expanded by the
       shell in the generated launch command, corrupting argv and leaking tokens
       into the training entrypoint's Hydra override parser.
+    * ``--offline`` — controls HF Hub access on the launcher node only; offline
+      behaviour in the rank-local script is governed by the ``HF_HUB_OFFLINE``
+      environment variable, not by this flag.
 
-    All of these take a value, passed either as ``--flag value`` (two tokens) or
-    ``--flag=value`` (one token).
+    Value-taking flags are passed as ``--flag value`` or ``--flag=value``.
+    Boolean flags (no following value) are listed in ``_LAUNCHER_ONLY_BOOL``.
     """
+    _LAUNCHER_ONLY_BOOL = {"--offline", "--dryrun", "-d"}
 
     def _is_launcher_only(flag: str) -> bool:
-        return flag in (
-            "-lgc",
-            "-lmc",
-            "-vb",
-            "--additional_slurm_params",
-            "--csp",
-            "--enable_vboost",
-            "--lock_gpu_freq",
-            "--peak_mem_clk",
-        ) or flag.startswith("--kubeflow_")
+        return (
+            flag
+            in (
+                "-lgc",
+                "-lmc",
+                "-vb",
+                "--additional_slurm_params",
+                "--csp",
+                "--enable_vboost",
+                "--lock_gpu_freq",
+                "--peak_mem_clk",
+            )
+            or flag.startswith("--kubeflow_")
+            or flag.startswith("--nvcre_")
+        )
 
     filtered_args = []
     skip_next = False
@@ -112,6 +121,8 @@ def _is_launcher_only(flag: str) -> bool:
         if skip_next:
             skip_next = False
             continue
+        if arg in _LAUNCHER_ONLY_BOOL:
+            continue
         if _is_launcher_only(arg.split("=", 1)[0]):
             skip_next = "=" not in arg
             continue
@@ -472,6 +483,23 @@ def maybe_increase_n_attempts_on_flaky_failure(
     return n_attempts
 
 
+def _nvcre_env_vars(custom_env_vars: Dict[str, str], hf_token_secret_name: Optional[str]) -> Dict[str, str]:
+    """Build the plain env-var dict for an NVCRE executor.
+
+    When an HF token secret is configured, HF_TOKEN is stripped from the plain
+    env (it may have arrived via cluster_config environment: or --env flags) and
+    injected via secret_env_vars instead, so the token never appears as a plain
+    value in the WorkloadRun manifest. HF connectivity is always enabled so the
+    pod can reach HuggingFace regardless of any --offline flag passed for
+    launcher-side setup.
+    """
+    env = custom_env_vars.copy()
+    if hf_token_secret_name:
+        env.pop("HF_TOKEN", None)
+        env.update({"HF_HUB_OFFLINE": "0", "TRANSFORMERS_OFFLINE": "0"})
+    return env
+
+
 def main(
     use_recipes: bool,
     model_family_name: str,
@@ -542,6 +570,21 @@ def main(
     kubeflow_container_kwargs_json: Optional[str],
     kubeflow_labels_json: Optional[str],
     kubeflow_pod_annotations_json: Optional[str],
+    nvcre_namespace: Optional[str] = None,
+    nvcre_image_pull_secret: Optional[str] = None,
+    nvcre_workdir_pvc: Optional[str] = None,
+    nvcre_workdir_pvc_path: str = "/nemo_run",
+    nvcre_workdir_local_path: Optional[str] = None,
+    nvcre_node_selector_json: Optional[str] = None,
+    nvcre_volumes_json: Optional[str] = None,
+    nvcre_volume_mounts_json: Optional[str] = None,
+    nvcre_timeout_per_job: str = "24h",
+    nvcre_kubeconfig: Optional[str] = None,
+    nvcre_kube_context: Optional[str] = None,
+    nvcre_gang_scheduler_name: Optional[str] = None,
+    nvcre_hf_token_secret_name: Optional[str] = None,
+    nvcre_hf_token_secret_key: str = "token",
+    deterministic: bool = False,
     config_variant: str | None = None,
     gres: Optional[str] = None,
     packager: str = "git",
@@ -594,7 +637,7 @@ def main(
         # runs. Creating the dir from the launcher would either fail (PVC
         # not present) or create a useless dir on the launcher's local FS.
         # Let the trainer container create its own dirs on first write.
-        if kubeflow_namespace is None:
+        if kubeflow_namespace is None and nvcre_namespace is None:
             save_dir_path.mkdir(parents=True, exist_ok=True)
             save_dir_mount = f"{save_dir_path}:{save_dir_path}"
             if save_dir_mount not in custom_mounts:
@@ -613,7 +656,7 @@ def main(
     # Kubeflow the trainer pod runs the image — which ships Megatron-Bridge at
     # /opt/Megatron-Bridge — and custom_mounts do not apply, so the launcher's
     # /tmp path does not exist in the pod; use the image's script path instead.
-    if kubeflow_namespace:
+    if kubeflow_namespace or nvcre_namespace:
         in_container_script_dir = "/opt/Megatron-Bridge/scripts/performance"
         in_container_script_path = f"{in_container_script_dir}/{script_name}"
     else:
@@ -634,11 +677,13 @@ def main(
     if peak_mem_clk == -1:
         peak_mem_clk = None
 
-    if kubeflow_namespace:
+    if kubeflow_namespace or nvcre_namespace:
         if enable_vboost or lock_gpu_freq is not None or peak_mem_clk is not None:
             logger.warning(
-                "--enable_vboost, --lock_gpu_freq, and --peak_mem_clk are Slurm-only and will be ignored on Kubeflow."
+                "--enable_vboost, --lock_gpu_freq, and --peak_mem_clk are Slurm-only and will be ignored on Kubeflow/NVCRE."
             )
+
+    if kubeflow_namespace:
         executor = kubeflow_executor(
             namespace=kubeflow_namespace,
             nodes=-(num_gpus // -gpus_per_node),
@@ -670,6 +715,31 @@ def main(
             labels=json.loads(kubeflow_labels_json) if kubeflow_labels_json else None,
             pod_annotations=(json.loads(kubeflow_pod_annotations_json) if kubeflow_pod_annotations_json else None),
         )
+    elif nvcre_namespace is not None:
+        try:
+            from utils.executors import nvcre_executor
+        except ImportError:
+            from .utils.executors import nvcre_executor
+        executor = nvcre_executor(
+            namespace=nvcre_namespace,
+            container_image=container_image,
+            nodes=-(num_gpus // -gpus_per_node),
+            num_gpus_per_node=gpus_per_node,
+            image_pull_secret=nvcre_image_pull_secret,
+            workdir_pvc=nvcre_workdir_pvc,
+            workdir_pvc_path=nvcre_workdir_pvc_path,
+            workdir_local_path=nvcre_workdir_local_path,
+            node_selector=json.loads(nvcre_node_selector_json) if nvcre_node_selector_json else None,
+            volumes=json.loads(nvcre_volumes_json) if nvcre_volumes_json else None,
+            volume_mounts=json.loads(nvcre_volume_mounts_json) if nvcre_volume_mounts_json else None,
+            timeout_per_job=nvcre_timeout_per_job,
+            kubeconfig=nvcre_kubeconfig,
+            kube_context=nvcre_kube_context,
+            gang_scheduler_name=nvcre_gang_scheduler_name,
+        )
+        executor.env_vars = _nvcre_env_vars(custom_env_vars, nvcre_hf_token_secret_name)
+        if nvcre_hf_token_secret_name:
+            executor.secret_env_vars["HF_TOKEN"] = (nvcre_hf_token_secret_name, nvcre_hf_token_secret_key)
     else:
         executor = slurm_executor(
             gpu=gpu,
@@ -788,15 +858,17 @@ def main(
                 logger.info("dryrun requested: exiting")
                 return
 
-            job_dir, job_status = get_job_dir_and_status_from_run(exp_name)
-
-            terminal_failure = job_status not in ["SUCCEEDED", "SUBMITTED", "PENDING", "RUNNING"]
-
             if detach:
+                # For detached runs (e.g. Nvcre), skip status polling — the
+                # caller (llmb-run) polls for completion via nvcrectl.
                 is_finished_experiment = True
                 is_testing_passed = True
                 break
 
+            job_dir, job_status = get_job_dir_and_status_from_run(exp_name)
+
+            terminal_failure = job_status not in ["SUCCEEDED", "SUBMITTED", "PENDING", "RUNNING"]
+
             log_file_paths = list(Path(f"{job_dir}").glob("log*.out"))
             ensure_logs_where_written(log_file_paths)
 
@@ -990,9 +1062,10 @@ def main(
         task=args.task,
         compute_dtype=args.compute_dtype,
         gpu=args.gpu,
-        hf_token=args.hf_token,
+        hf_token=args.hf_token or os.environ.get("HF_TOKEN"),
         offline=args.offline,
-        detach=args.detach,
+        # Force detach for Nvcre — llmb-run polls for completion via nvcrectl
+        detach=True if args.nvcre_namespace else args.detach,
         dryrun=args.dryrun,
         enable_vboost=args.enable_vboost,
         lock_gpu_freq=args.lock_gpu_freq,
@@ -1069,6 +1142,21 @@ def main(
         kubeflow_container_kwargs_json=args.kubeflow_container_kwargs_json,
         kubeflow_labels_json=args.kubeflow_labels_json,
         kubeflow_pod_annotations_json=args.kubeflow_pod_annotations_json,
+        nvcre_namespace=args.nvcre_namespace,
+        nvcre_image_pull_secret=args.nvcre_image_pull_secret,
+        nvcre_workdir_pvc=args.nvcre_workdir_pvc,
+        nvcre_workdir_pvc_path=args.nvcre_workdir_pvc_path,
+        nvcre_workdir_local_path=args.nvcre_workdir_local_path,
+        nvcre_node_selector_json=args.nvcre_node_selector_json,
+        nvcre_volumes_json=args.nvcre_volumes_json,
+        nvcre_volume_mounts_json=args.nvcre_volume_mounts_json,
+        nvcre_timeout_per_job=args.nvcre_timeout_per_job,
+        nvcre_kubeconfig=args.nvcre_kubeconfig,
+        nvcre_kube_context=args.nvcre_kube_context,
+        nvcre_gang_scheduler_name=args.nvcre_gang_scheduler_name,
+        nvcre_hf_token_secret_name=args.nvcre_hf_token_secret_name,
+        nvcre_hf_token_secret_key=args.nvcre_hf_token_secret_key,
+        deterministic=args.deterministic,
         config_variant=config_variant,
         gres=args.gres,
         packager=args.packager,