From 028c5b01f9515dece7f4164c9564803b577de0b2 Mon Sep 17 00:00:00 2001
From: Dennis-yxchen <ddddennis.chan@gmail.com>
Date: Tue, 31 Mar 2026 17:38:39 +0800
Subject: [PATCH] fix: use hash-based workflow ID to prevent separator
 collisions

The previous __-joined scheme was not injective: a directory named
a__b and a nested path a/b both mapped to the same ID. Use a sha256
hash suffix of the resolved path instead, which is collision-free
and keeps the dir name as a human-readable prefix.

Added regression test for separator collision case.
---
 openspace/dashboard_server.py | 18 +++++++++---------
 tests/test_workflow_id.py     | 29 ++++++++++++++++++++++-------
 2 files changed, 31 insertions(+), 16 deletions(-)

diff --git a/openspace/dashboard_server.py b/openspace/dashboard_server.py
index 4333aef..f6916b6 100644
--- a/openspace/dashboard_server.py
+++ b/openspace/dashboard_server.py
@@ -420,15 +420,15 @@ def _build_lineage_payload(skill_id: str, store: SkillStore) -> Dict[str, Any]:
 
 
 def _workflow_id(workflow_dir: Path) -> str:
-    """Stable short ID for a workflow directory, unique across roots."""
-    resolved = workflow_dir.resolve()
-    for root in WORKFLOW_ROOTS:
-        try:
-            rel = resolved.relative_to(root.resolve())
-            return f"{root.name}__{'__'.join(rel.parts)}"
-        except ValueError:
-            continue
-    return workflow_dir.name
+    """Stable short ID for a workflow directory, unique across roots.
+
+    Uses a hash suffix derived from the resolved path to avoid collisions
+    when directory names contain the separator character.
+    """
+    import hashlib
+    resolved = str(workflow_dir.resolve())
+    path_hash = hashlib.sha256(resolved.encode()).hexdigest()[:8]
+    return f"{workflow_dir.name}_{path_hash}"
 
 
 def _discover_workflow_dirs() -> List[Path]:
diff --git a/tests/test_workflow_id.py b/tests/test_workflow_id.py
index d4ca82f..84bf3e1 100644
--- a/tests/test_workflow_id.py
+++ b/tests/test_workflow_id.py
@@ -28,27 +28,42 @@ class TestWorkflowId:
             id_b = dashboard_server._workflow_id(wf_b)
 
         assert id_a != id_b
-        assert "task1" in id_a
-        assert "task1" in id_b
+        assert id_a.startswith("task1_")
+        assert id_b.startswith("task1_")
 
-    def test_nested_workflow_id_includes_path(self, tmp_path):
+    def test_id_contains_dir_name_and_hash(self, tmp_path):
         root = tmp_path / "root"
-        wf = root / "sub" / "deep" / "task1"
+        wf = root / "my-task"
         _make_workflow(wf)
 
         with patch.object(dashboard_server, "WORKFLOW_ROOTS", [root]):
             wf_id = dashboard_server._workflow_id(wf)
 
-        assert wf_id == "root__sub__deep__task1"
+        assert wf_id.startswith("my-task_")
+        assert len(wf_id) == len("my-task_") + 8  # 8-char hex hash
 
-    def test_workflow_outside_roots_falls_back_to_name(self, tmp_path):
+    def test_separator_collision_produces_different_ids(self, tmp_path):
+        """Regression: a dir named 'a__b' must not collide with path a/b."""
+        root = tmp_path / "root"
+        flat = root / "a__b"
+        nested = root / "a" / "b"
+        _make_workflow(flat)
+        _make_workflow(nested)
+
+        with patch.object(dashboard_server, "WORKFLOW_ROOTS", [root]):
+            id_flat = dashboard_server._workflow_id(flat)
+            id_nested = dashboard_server._workflow_id(nested)
+
+        assert id_flat != id_nested
+
+    def test_workflow_outside_roots_still_works(self, tmp_path):
         wf = tmp_path / "orphan_workflow"
         _make_workflow(wf)
 
         with patch.object(dashboard_server, "WORKFLOW_ROOTS", []):
             wf_id = dashboard_server._workflow_id(wf)
 
-        assert wf_id == "orphan_workflow"
+        assert wf_id.startswith("orphan_workflow_")
 
 
 class TestDiscoverWorkflowDirs: