Fix workflow worker data path normalization (#644)

Author: jpmorard

src/agilab/orchestrate/orchestrate_cluster.py

@@ -490,14 +490,6 @@ def _path_matches_workflow_session_default(data_path: Path, sessions_root: Path,
     return len(relative_path.parts) == 2 and relative_path.parts[1] == module_name
 
 
-def _path_is_current_workflow_module_or_child(data_path: Path, sessions_root: Path, module_name: str) -> bool:
-    try:
-        relative_path = data_path.relative_to(sessions_root)
-    except ValueError:
-        return False
-    return len(relative_path.parts) >= 3 and relative_path.parts[1] == module_name
-
-
 def _workers_data_path_should_follow_workflow_session(
     value: Any,
     env: Any,
@@ -523,7 +515,7 @@ def _workers_data_path_should_follow_workflow_session(
         return True
     user_root = _workflow_user_root(cluster_share, user)
     if _path_is_within(data_path, user_root):
-        return not _path_is_current_workflow_module_or_child(data_path, sessions_root, module_name)
+        return True
     return False
 
 
@@ -1604,7 +1596,21 @@ def render_cluster_settings_ui(env: Any, deps: OrchestrateClusterDeps, *, show_r
             placeholder="/path/to/data",
             help="Path to data directory on workers.",
         )
-        cluster_params["workers_data_path"] = str(workers_data_path_input or "").strip()
+        workers_data_path_value = str(workers_data_path_input or "").strip()
+        if (
+            workflow_workers_data_path
+            and _clean_path_text(workers_data_path_value)
+            and _workers_data_path_should_follow_workflow_session(
+                workers_data_path_value,
+                env,
+                user=sanitized_user,
+                workflow_name=workflow_id,
+                project_name=app_state_name,
+            )
+        ):
+            workers_data_path_value = workflow_workers_data_path
+            st.session_state[workers_data_path_widget_key] = workflow_workers_data_path
+        cluster_params["workers_data_path"] = workers_data_path_value
 
         workers_widget_key = widget_keys["workers"]
         workers_dict = cluster_params.get("workers", {})

test/test_orchestrate_cluster.py

@@ -562,7 +562,7 @@ def test_workflow_session_path_policies(tmp_path):
         workflow_name="workflows",
         project_name="demo_project",
     )
-    assert not orchestrate_cluster._workers_data_path_should_follow_workflow_session(
+    assert orchestrate_cluster._workers_data_path_should_follow_workflow_session(
         "clustershare/agi/workflows/latest-run/demo/custom-data",
         env,
         user="agi",
@@ -657,13 +657,27 @@ def test_workers_data_path_regression_rewrites_misplaced_managed_share_values(tm
         workflow_name="workflows",
         project_name="flight_telemetry_project",
     )
-    assert not orchestrate_cluster._workers_data_path_should_follow_workflow_session(
+    assert orchestrate_cluster._workers_data_path_should_follow_workflow_session(
         "clustershare/agi/workflows/20260618T093102Z-492de776/flight_telemetry/custom-data",
         env,
         user="agi",
         workflow_name="workflows",
         project_name="flight_telemetry_project",
     )
+    assert orchestrate_cluster._workers_data_path_should_follow_workflow_session(
+        "clustershare/agi/workflows/20260618T093102Z-492de776/flight_trajectory/dataset",
+        env,
+        user="agi",
+        workflow_name="workflows",
+        project_name="flight_trajectory_project",
+    )
+    assert orchestrate_cluster._workers_data_path_should_follow_workflow_session(
+        "clustershare/agi/workflows/20260618T093102Z-492de776/flight_trajectory/flight_trajectory/dataset",
+        env,
+        user="agi",
+        workflow_name="workflows",
+        project_name="flight_trajectory_project",
+    )
 
 
 def test_cluster_advisor_recommends_workers_and_writes_evidence(tmp_path):
@@ -1561,12 +1575,13 @@ def test_render_cluster_settings_ui_empty_workflow_session_auto_selects_latest(m
     assert cluster["workers_data_path"] == "cluster-share/agi/workflows/session-b"
 
 
-def test_render_cluster_settings_ui_preserves_custom_workers_data_path_under_workflow_root(monkeypatch, tmp_path):
-    app_name = "demo_project"
+def test_render_cluster_settings_ui_rewrites_stale_app_child_workers_data_path(monkeypatch, tmp_path):
+    app_name = "flight_trajectory_project"
     widget_keys = orchestrate_cluster.cluster_widget_keys(app_name)
     share = tmp_path / "cluster-share"
     (share / "agi" / "workflows" / "session-a").mkdir(parents=True)
-    custom_data_path = "cluster-share/agi/workflows/session-a/demo/custom-data"
+    stale_data_path = "cluster-share/agi/workflows/session-a/flight_trajectory/dataset"
+    session_data_path = "cluster-share/agi/workflows/session-a"
     fake_st = _FakeStreamlit(
         widget_values={
             widget_keys["cluster_enabled"]: True,
@@ -1575,18 +1590,18 @@ def test_render_cluster_settings_ui_preserves_custom_workers_data_path_under_wor
             widget_keys["rapids"]: False,
             widget_keys["use_key"]: True,
             widget_keys["workflow_session"]: "session-a",
-            widget_keys["workers_data_path"]: custom_data_path,
+            widget_keys["workers_data_path"]: stale_data_path,
         },
         session_state={
             "app_settings": {
                 "cluster": {
                     "cluster_enabled": True,
                     "workflow_session": "session-a",
-                    "workers_data_path": custom_data_path,
+                    "workers_data_path": stale_data_path,
                 }
             },
             widget_keys["workflow_session"]: "session-a",
-            widget_keys["workers_data_path"]: custom_data_path,
+            widget_keys["workers_data_path"]: stale_data_path,
             "benchmark": False,
         },
     )
@@ -1617,8 +1632,8 @@ def test_render_cluster_settings_ui_preserves_custom_workers_data_path_under_wor
 
     cluster = fake_st.session_state.app_settings["cluster"]
     assert cluster["workflow_session"] == "session-a"
-    assert cluster["workers_data_path"] == custom_data_path
-    assert fake_st.session_state[widget_keys["workers_data_path"]] == custom_data_path
+    assert cluster["workers_data_path"] == session_data_path
+    assert fake_st.session_state[widget_keys["workers_data_path"]] == session_data_path
 
 
 def test_render_cluster_settings_ui_selects_existing_workflow_session_directory(monkeypatch, tmp_path):