Heirarchical lineage

docs/ui/lineage.md

@@ -69,6 +69,25 @@ The Lineage page offers three different visualization modes:
 
 ![Artifact Execution Tree Lineage](../assets/artifact_exec_tree_lineage.png)
 
+### 4. Heirarchical Lineage
+
+**Purpose**: High-level, multi-column tree abstraction tracking task orchestration flows branching from root configurations down through execution chains.
+
+**Use Cases**:
+
+- Tracking simultaneous execution tracks branching from a single environment or pipeline root.
+- Tracing linear execution histories sequentially across multiple stages like preparation, features, training, and evaluation.
+- Auditing full procedural histories by analyzing vertically ordered step progressions.
+
+**Features**:
+
+- Environment-centric root anchoring branching into independent functional swimlanes.
+- Color-coded stage markers distinguishing entry hooks from down-stream operational tasks.
+- Distinct step blocks capturing individual execution hashes sequentially.
+- Vertical progression layouts tracing step dependencies from top to bottom.
+
+![Heirarchical Lineage](../assets/Hierarchical_lineage.png)
+
 
 ## Using the Lineage Page
 

server/app/main.py

@@ -13,7 +13,7 @@
 import asyncio
 from sqlalchemy.ext.asyncio import AsyncSession
 from collections import defaultdict
-from server.app.utils import extract_hostname, get_fqdn
+from server.app.utils import extract_hostname, get_fqdn, convert_to_stage_json
 from server.app.get_data import (
     get_mlmd_from_server,
     get_artifact_types,
@@ -611,6 +611,30 @@ async def artifact_execution_lineage(request: Request, pipeline_name: str):
     return response
 
 
+@app.get("/hierarchical-lineage/tangled-tree/{pipeline_name}")
+async def hierarchical_lineage(request: Request, pipeline_name: str):
+    """
+    Return MLMD data for the given pipeline converted into the UI `stage.json` schema.
+    """
+    # checks if mlmd file exists on server
+    await check_mlmd_file_exists()
+    # checks if pipeline exists
+    await check_pipeline_exists(pipeline_name)
+
+    # Pull MLMD JSON for the pipeline
+    json_payload = await async_api(get_mlmd_from_server, query, pipeline_name, None, None, dict_of_exe_ids)
+
+    if json_payload is None:
+        raise HTTPException(status_code=404, detail=f"Pipeline {pipeline_name} not found or has no MLMD data.")
+
+    try:
+        converted = convert_to_stage_json(json_payload, pipeline_name)
+    except Exception as e:
+        raise HTTPException(status_code=500, detail=f"Failed to convert MLMD to stage JSON: {e}")
+
+    return converted
+
+
 @app.get("/list-of-executions/{pipeline_name}")
 async def list_of_executions(request: Request, pipeline_name: str):
     '''

server/app/utils.py

@@ -99,3 +99,130 @@ def get_fqdn(name: str) -> str:
         return fqdn
     except Exception:
         return "127.0.0.1"
+
+import json
+
+def convert_to_stage_json(input_json: dict, pipeline_name: str) -> dict:
+    """
+    Converts MLMD JSON for a clean hierarchy map.
+    - Removes trailing child nodes underneath executions.
+    - Combines Execution Type name and a 6-digit truncated UUID.
+    - Includes rich metadata fields for frontend hover/tooltip parsing.
+    """
+
+    def extract_stages_recursively(data):
+        """Recursively parses and drills down to locate the raw stages list."""
+        if isinstance(data, str):
+            cleaned = data.strip()
+            try:
+                return extract_stages_recursively(json.loads(cleaned))
+            except Exception:
+                if "stages" in cleaned or "Pipeline" in cleaned:
+                    try:
+                        if not (cleaned.startswith('{') and cleaned.endswith('}')):
+                            return extract_stages_recursively(json.loads("{" + cleaned + "}"))
+                    except Exception:
+                        pass
+                return None
+
+        if isinstance(data, dict):
+            for key in ("stages", "stage_list", "stageList"):
+                if key in data and isinstance(data[key], list):
+                    return data[key]
+            for key in ("Pipeline", "pipelines", "name"):
+                if key in data:
+                    res = extract_stages_recursively(data[key])
+                    if res: return res
+            for v in data.values():
+                if isinstance(v, (dict, list, str)):
+                    res = extract_stages_recursively(v)
+                    if res: return res
+
+        if isinstance(data, list):
+            for item in data:
+                res = extract_stages_recursively(item)
+                if res: return res
+
+        return None
+
+    stages_candidate = extract_stages_recursively(input_json)
+    if not stages_candidate:
+        stages_candidate = []
+
+    out = {
+        "environment": pipeline_name,
+        "metadata": { "version": "4.0.0", "description": "Executions Lineage Map" },
+        "stages": []
+    }
+
+    stage_idx = 1
+    exec_idx = 1
+
+    for s_item in stages_candidate:
+        if not isinstance(s_item, dict):
+            continue
+
+        raw_executions = s_item.get("executions") or []
+        if isinstance(raw_executions, str):
+            try: raw_executions = json.loads(raw_executions)
+            except Exception: raw_executions = []
+
+        if not raw_executions:
+            continue
+
+        # Extract context attributes from the execution definition
+        first_exec = raw_executions[0] if isinstance(raw_executions, list) and len(raw_executions) > 0 else {}
+        props = first_exec.get("properties") or {} if isinstance(first_exec, dict) else {}
+
+        context_type = props.get("Context_Type") or props.get("Execution_type_name") or first_exec.get("type") or ""
+        if "/" in context_type:
+            stage_name = context_type.split("/", 1)[1]
+        elif context_type:
+            stage_name = context_type
+        else:
+            stage_name = f"stage_{stage_idx}"
+
+        stage_obj = {
+            "stage_id": f"stage_{stage_idx:02d}",
+            "stage_name": stage_name,
+            "status": "completed",
+            "executions": []
+        }
+        stage_idx += 1
+
+        for e in raw_executions:
+            if not isinstance(e, dict):
+                continue
+
+            e_props = e.get("properties") or {}
+
+            # 1. Fetch execution base name (e.g., "Prepare")
+            exec_base_name = e.get("type") or "Execution"
+
+            # 2. Use Execution_uuid first — same field the tangled-tree lineage keys off of,
+            #    so the same execution shows the same short UUID in both views.
+            full_uuid = str(e_props.get("Execution_uuid") or e_props.get("Git_End_Commit") or e_props.get("Git_Start_Commit") or e.get("id") or "")
+            truncated_uuid = full_uuid[:4] if full_uuid else f"ex_{exec_idx}"
+
+            # 3. Create the multi-line UI text block
+            # Note: '\n' handles line breaks in graph frameworks configured with CSS white-space: pre-wrap
+            display_text = f"{exec_base_name}\n{truncated_uuid}"
+
+            exec_obj = {
+                "execution_id": f"exec_{exec_idx:03d}",
+                "execution_type": display_text, 
+                "children": [],  # Empty array ensures no leaf artifact nodes render underneath
+
+                # Broad compatibility fields for the UI graphing framework's hover tooltip engine
+                "tooltip": full_uuid,
+                "title": full_uuid,
+                "description": f"Full UUID: {full_uuid}",
+                "full_uuid": full_uuid
+            }
+            exec_idx += 1
+
+            stage_obj["executions"].append(exec_obj)
+
+        out["stages"].append(stage_obj)
+
+    return out