fix: store job_id explicitly to avoid separator collision in app_id parsing

When a role name ends with '_', the app_id string looks like:
  experiment___role_name____job_id
Splitting on '___' produces job_id = '_job_id' (spurious leading '_'),
causing the status/cancel/log_iter calls to use a wrong ID and get 404.

Fix: _save_job_dir now stores the actual job_id in the JSON record.
describe(), log_iter(), and _cancel_existing() all read job_id from
the stored record, falling back to app_id.split('___')[-1] for
backwards compatibility with existing saved jobs.

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>

Author: ko3n1g

nemo_run/run/torchx_backend/schedulers/dgxcloud.py

@@ -161,7 +161,7 @@ def schedule(self, dryrun_info: AppDryRunInfo[DGXRequest]) -> str:
 
         # Store a status entry or logs path if available
         # Currently, the DGXExecutor status is placeholder, but we keep the pattern
-        _save_job_dir(app_id, job_status=status, executor=executor)
+        _save_job_dir(app_id, job_status=status, executor=executor, job_id=job_id)
 
         return app_id
 
@@ -173,7 +173,8 @@ def describe(self, app_id: str) -> Optional[DescribeAppResponse]:
         # We split out the stored values from the JSON file
         stored_data = _get_job_dirs()
         job_info = stored_data.get(app_id)
-        _, role_name, job_id = app_id.split("___")
+        parts = app_id.split("___")
+        role_name = parts[1] if len(parts) > 1 else app_id
         roles = [Role(name=role_name, image="", num_replicas=1)]
         roles_statuses = [
             RoleStatus(
@@ -191,6 +192,7 @@ def describe(self, app_id: str) -> Optional[DescribeAppResponse]:
         if not executor:
             return None
 
+        job_id = job_info.get("job_id") or parts[-1]
         dgx_state = executor.status(job_id) or DGXCloudState.UNKNOWN
         app_state = DGX_STATES.get(dgx_state, AppState.PENDING)
         roles_statuses[0].replicas[0].state = app_state
@@ -217,7 +219,7 @@ def log_iter(
     ) -> Iterable[str]:
         stored_data = _get_job_dirs()
         job_info = stored_data.get(app_id)
-        _, _, job_id = app_id.split("___")
+        job_id = job_info.get("job_id") or app_id.split("___")[-1]
         executor: Optional[DGXCloudExecutor] = job_info.get("executor", None)  # type: ignore
         if not executor:
             return [""]
@@ -240,7 +242,7 @@ def _cancel_existing(self, app_id: str) -> None:
         """
         stored_data = _get_job_dirs()
         job_info = stored_data.get(app_id)
-        _, _, job_id = app_id.split("___")
+        job_id = job_info.get("job_id") or app_id.split("___")[-1]
         executor: DGXCloudExecutor = job_info.get("executor", None)  # type: ignore
         if not executor:
             return None
@@ -257,7 +259,9 @@ def create_scheduler(session_name: str, **kwargs: Any) -> DGXCloudScheduler:
     return DGXCloudScheduler(session_name=session_name)
 
 
-def _save_job_dir(app_id: str, job_status: str, executor: DGXCloudExecutor) -> None:
+def _save_job_dir(
+    app_id: str, job_status: str, executor: DGXCloudExecutor, job_id: str = ""
+) -> None:
     """
     Saves or updates local record of job status in JSON for demonstration.
     """
@@ -276,6 +280,7 @@ def _save_job_dir(app_id: str, job_status: str, executor: DGXCloudExecutor) -> N
 
         app = {
             "job_status": job_status,
+            "job_id": job_id,
             "executor": serializer.serialize(
                 fdl_dc.convert_dataclasses_to_configs(executor, allow_post_init=True)
             ),

test/run/torchx_backend/schedulers/test_dgxcloud.py

@@ -110,6 +110,7 @@ def test_describe(dgx_cloud_scheduler, dgx_cloud_executor):
         mock_get_job_dirs.return_value = {
             "test_experiment___test_role___test_job_id": {
                 "job_status": "RUNNING",
+                "job_id": "test_job_id",
                 "executor": dgx_cloud_executor,
             }
         }
@@ -132,6 +133,7 @@ def test_cancel_existing(dgx_cloud_scheduler, dgx_cloud_executor):
         mock_get_job_dirs.return_value = {
             "test_experiment___test_role___test_job_id": {
                 "job_status": "RUNNING",
+                "job_id": "test_job_id",
                 "executor": dgx_cloud_executor,
             }
         }
@@ -159,10 +161,11 @@ def test_save_and_get_job_dirs():
             pvc_nemo_run_dir="/workspace/nemo_run",
         )
 
-        _save_job_dir("test_app_id", "RUNNING", executor)
+        _save_job_dir("test_app_id", "RUNNING", executor, job_id="actual_job_id")
         job_dirs = _get_job_dirs()
 
         assert "test_app_id" in job_dirs
+        assert job_dirs["test_app_id"]["job_id"] == "actual_job_id"
         assert isinstance(job_dirs["test_app_id"]["executor"], DGXCloudExecutor)
 
 
@@ -173,6 +176,7 @@ def test_log_iter(dgx_cloud_scheduler, dgx_cloud_executor):
         mock_get_job_dirs.return_value = {
             "test_session___test_role___test_container_id": {
                 "job_status": "RUNNING",
+                "job_id": "test_job_id",
                 "executor": dgx_cloud_executor,
             }
         }
@@ -188,6 +192,34 @@ def test_log_iter(dgx_cloud_scheduler, dgx_cloud_executor):
         assert logs == ["log2", "log3"]
 
 
+def test_describe_uses_stored_job_id_not_split(dgx_cloud_scheduler, dgx_cloud_executor):
+    # Regression test: when a role name ends with '_', splitting app_id on '___'
+    # produces a job_id with a spurious leading '_' (e.g. role 'W-foo_' + sep '___'
+    # gives '____' which splits into 'role_' and '_job_id'). describe() must use
+    # the job_id stored at schedule time, not re-derive it from the app_id string.
+    real_job_id = "48db46d2-ae56-4c9d-9abd-ba0d873e50eb"
+    # role name ending with '_' triggers the collision
+    app_id = f"experiment___role_name___{real_job_id}"
+
+    with (
+        mock.patch(
+            "nemo_run.run.torchx_backend.schedulers.dgxcloud._get_job_dirs"
+        ) as mock_get_job_dirs,
+        mock.patch.object(DGXCloudExecutor, "status", return_value=DGXCloudState.RUNNING) as mock_status,
+    ):
+        mock_get_job_dirs.return_value = {
+            app_id: {
+                "job_status": "RUNNING",
+                "job_id": real_job_id,
+                "executor": dgx_cloud_executor,
+            }
+        }
+
+        response = dgx_cloud_scheduler.describe(app_id)
+        assert response is not None
+        mock_status.assert_called_once_with(real_job_id)
+
+
 def test_unknown_state_maps_to_pending_not_failed():
     # DGXCloudState.UNKNOWN must map to PENDING so transient API errors during
     # job startup do not cause wait_and_exit() to treat the job as terminal.
@@ -207,6 +239,7 @@ def test_describe_returns_pending_when_status_is_none(dgx_cloud_scheduler, dgx_c
         mock_get_job_dirs.return_value = {
             "test_experiment___test_role___test_job_id": {
                 "job_status": "RUNNING",
+                "job_id": "test_job_id",
                 "executor": dgx_cloud_executor,
             }
         }
@@ -229,6 +262,7 @@ def test_describe_returns_pending_when_status_is_unknown(dgx_cloud_scheduler, dg
         mock_get_job_dirs.return_value = {
             "test_experiment___test_role___test_job_id": {
                 "job_status": "RUNNING",
+                "job_id": "test_job_id",
                 "executor": dgx_cloud_executor,
             }
         }
@@ -245,6 +279,7 @@ def test_log_iter_str(dgx_cloud_scheduler, dgx_cloud_executor):
         mock_get_job_dirs.return_value = {
             "test_session___test_role___test_container_id": {
                 "job_status": "RUNNING",
+                "job_id": "test_job_id",
                 "executor": dgx_cloud_executor,
             }
         }