Fix ORCHESTRATE refresh workers path growth (#649)

Co-authored-by: jpmorard <focus@thalesgroup.com>

Author: jpmorard

src/agilab/orchestrate/orchestrate_cluster.py

@@ -419,6 +419,16 @@ def _workflow_sessions_root(cluster_share: Path, user: Any, workflow_name: Any)
     return _workflow_user_root(cluster_share, user) / safe_workflow
 
 
+def _workflow_cluster_share_root(cluster_share: Path, *, user: Any, workflow_name: Any) -> Path:
+    safe_user = _safe_workflow_component(user, "local-user")
+    safe_workflow = _safe_workflow_component(workflow_name, "default")
+    parts = cluster_share.parts
+    for index in range(len(parts) - 3, -1, -1):
+        if parts[index] == safe_user and parts[index + 1] == safe_workflow:
+            return Path(*parts[: index + 1])
+    return cluster_share
+
+
 def _list_workflow_sessions(sessions_root: Path) -> list[Path]:
     try:
         entries = [path for path in sessions_root.iterdir() if path.is_dir()]
@@ -471,6 +481,11 @@ def _resolve_workflow_session_path(
 def _workflow_data_path_text(cluster_share: Path, session_path: Path, env: Any) -> str:
     cluster_share_setting = _env_cluster_share_setting(env)
     if cluster_share_setting:
+        setting_path = _resolve_env_relative_path(cluster_share_setting, env)
+        if setting_path is not None and (
+            setting_path == session_path or _path_is_within(setting_path, session_path)
+        ):
+            return _home_relative_share_text(session_path, env) or str(session_path)
         try:
             suffix = session_path.relative_to(cluster_share)
         except ValueError:
@@ -1465,14 +1480,23 @@ def render_cluster_settings_ui(env: Any, deps: OrchestrateClusterDeps, *, show_r
             and cluster_share_candidate.is_dir()
             and os.access(cluster_share_candidate, os.W_OK)
         )
+        workflow_cluster_share_candidate = (
+            _workflow_cluster_share_root(
+                cluster_share_candidate,
+                user=sanitized_user,
+                workflow_name=workflow_id,
+            )
+            if cluster_share_candidate is not None
+            else None
+        )
         workflow_session_options: list[str] = []
-        if cluster_share_ready and cluster_share_candidate is not None:
+        if cluster_share_ready and workflow_cluster_share_candidate is not None:
             workflow_session_options = _workflow_session_names(
-                _workflow_sessions_root(cluster_share_candidate, sanitized_user, workflow_id)
+                _workflow_sessions_root(workflow_cluster_share_candidate, sanitized_user, workflow_id)
             )
             try:
                 workflow_session_path, workflow_session, workflow_policy = _resolve_workflow_session_path(
-                    cluster_share_candidate,
+                    workflow_cluster_share_candidate,
                     user=sanitized_user,
                     workflow_name=workflow_id,
                     policy=workflow_policy,
@@ -1484,7 +1508,7 @@ def render_cluster_settings_ui(env: Any, deps: OrchestrateClusterDeps, *, show_r
                 st.error(f"Could not prepare workflow session under `{cluster_share_candidate}`: {exc}")
             else:
                 workflow_workers_data_path = _workflow_data_path_text(
-                    cluster_share_candidate,
+                    workflow_cluster_share_candidate,
                     workflow_session_path,
                     env,
                 )
@@ -1530,7 +1554,7 @@ def render_cluster_settings_ui(env: Any, deps: OrchestrateClusterDeps, *, show_r
                 st.session_state.pop(workflow_existing_session_key, None)
                 st.info(
                     "No existing workflow session directories found under "
-                    f"`{_workflow_sessions_root(cluster_share_candidate, sanitized_user, workflow_id)}`."
+                    f"`{_workflow_sessions_root(workflow_cluster_share_candidate, sanitized_user, workflow_id)}`."
                 )
         else:
             st.session_state.pop(workflow_existing_session_key, None)
@@ -1547,10 +1571,10 @@ def render_cluster_settings_ui(env: Any, deps: OrchestrateClusterDeps, *, show_r
                 ),
             )
         workflow_session = _safe_workflow_component(workflow_session_input, "")
-        if cluster_share_ready and cluster_share_candidate is not None and not workflow_session_error_shown:
+        if cluster_share_ready and workflow_cluster_share_candidate is not None and not workflow_session_error_shown:
             try:
                 workflow_session_path, workflow_session, workflow_policy = _resolve_workflow_session_path(
-                    cluster_share_candidate,
+                    workflow_cluster_share_candidate,
                     user=sanitized_user,
                     workflow_name=workflow_id,
                     policy=workflow_policy,
@@ -1560,7 +1584,7 @@ def render_cluster_settings_ui(env: Any, deps: OrchestrateClusterDeps, *, show_r
                 st.error(f"Could not prepare workflow session under `{cluster_share_candidate}`: {exc}")
             else:
                 workflow_workers_data_path = _workflow_data_path_text(
-                    cluster_share_candidate,
+                    workflow_cluster_share_candidate,
                     workflow_session_path,
                     env,
                 )

test/test_orchestrate_cluster.py

@@ -632,6 +632,22 @@ def test_workflow_session_path_regression_uses_workflow_session_root_semantic(tm
     assert "workers" not in path.parts
     assert "flight_telemetry_project" not in path.parts
     assert "flight_telemetry" not in path.parts
+    assert (
+        orchestrate_cluster._workflow_cluster_share_root(
+            path,
+            user="agi",
+            workflow_name="workflows",
+        )
+        == share
+    )
+    assert (
+        orchestrate_cluster._workflow_cluster_share_root(
+            share / "workflows",
+            user="agi",
+            workflow_name="workflows",
+        )
+        == share / "workflows"
+    )
 
 
 def test_workflow_workers_data_path_regression_does_not_duplicate_module_dataset(tmp_path):
@@ -2025,6 +2041,90 @@ def test_render_cluster_settings_ui_refresh_replaces_stale_lan_discovery_state(m
     assert any("LAN discovery refreshed" in info for info in fake_st.infos)
 
 
+def test_render_cluster_settings_ui_refresh_keeps_session_scoped_share_root(monkeypatch, tmp_path):
+    app_name = "flight_trajectory_project"
+    session_id = "20260618T185326Z-4ffeb367"
+    widget_keys = orchestrate_cluster.cluster_widget_keys(app_name)
+    session_share = tmp_path / "clustershare" / "agi" / "workflows" / session_id
+    session_share.mkdir(parents=True)
+    refresh_calls = []
+    fake_st = _FakeStreamlit(
+        widget_values={
+            widget_keys["cluster_enabled"]: True,
+            widget_keys["cython"]: False,
+            widget_keys["pool"]: False,
+            widget_keys["rapids"]: False,
+            widget_keys["use_key"]: True,
+            widget_keys["workflow_session"]: session_id,
+        },
+        button_values={
+            orchestrate_cluster._lan_discovery_refresh_key(app_name): True,
+        },
+        session_state={
+            "app_settings": {
+                "cluster": {
+                    "cluster_enabled": True,
+                    "scheduler": "10.0.0.10:8786",
+                    "workers": {"10.0.0.11": 2},
+                    "workers_data_path": f"clustershare/agi/workflows/{session_id}",
+                    "workflow_session": session_id,
+                }
+            },
+            widget_keys["scheduler"]: "10.0.0.10:8786",
+            widget_keys["user"]: "agi",
+            widget_keys["workers"]: '{"10.0.0.11": 2}',
+            widget_keys["workers_data_path"]: f"clustershare/agi/workflows/{session_id}",
+            widget_keys["workflow_session"]: session_id,
+            "benchmark": False,
+        },
+    )
+    monkeypatch.setattr(orchestrate_cluster, "st", fake_st)
+    monkeypatch.setattr(
+        orchestrate_cluster,
+        "_lan_discovery_cluster_defaults",
+        lambda *_, **__: {
+            "scheduler": "192.168.3.103:8786",
+            "workers": {"192.168.3.35": 1},
+        },
+    )
+    monkeypatch.setattr(
+        orchestrate_cluster,
+        "_refresh_lan_discovery_cache",
+        lambda cache_path, **kwargs: refresh_calls.append((cache_path, kwargs))
+        or (True, "LAN discovery refreshed: 1 node(s), 1 ready."),
+    )
+    deps = orchestrate_cluster.OrchestrateClusterDeps(
+        parse_and_validate_scheduler=lambda raw: raw,
+        parse_and_validate_workers=lambda raw: {"192.168.3.35": 1} if "192.168.3.35" in raw else None,
+        write_app_settings_toml=lambda _path, settings: settings,
+        clear_load_toml_cache=lambda: None,
+        set_env_var=lambda _key, _value: None,
+        agi_env_envars={},
+    )
+    env = SimpleNamespace(
+        app=app_name,
+        home_abs=tmp_path,
+        is_managed_pc=False,
+        AGI_CLUSTER_SHARE=str(session_share),
+        agi_share_path=Path(f"clustershare/agi/workflows/{session_id}"),
+        share_root_path=lambda: session_share,
+        user="agi",
+        password=None,
+        ssh_key_path=None,
+        app_settings_file=tmp_path / "app_settings.toml",
+    )
+
+    orchestrate_cluster.render_cluster_settings_ui(env, deps)
+
+    cluster = fake_st.session_state.app_settings["cluster"]
+    expected = f"clustershare/agi/workflows/{session_id}"
+    duplicated = f"{expected}/agi/workflows/{session_id}"
+    assert cluster["workers_data_path"] == expected
+    assert fake_st.session_state[widget_keys["workers_data_path"]] == expected
+    assert duplicated not in cluster["workers_data_path"]
+    assert refresh_calls
+
+
 def test_render_cluster_settings_ui_builds_advisory_cluster_plan_without_applying(monkeypatch, tmp_path):
     app_name = "demo_project"
     widget_keys = orchestrate_cluster.cluster_widget_keys(app_name)