Attempt to render DAG from DB

fallback to tasks file

Author: MariusWirtz

src/rushti/commands.py

@@ -1664,37 +1664,63 @@ def _stats_visualize(args) -> None:
 
         # --- Attempt DAG generation ---
         dag_generated = False
-        taskfile_path = None
 
-        # Find the first accessible taskfile_path from runs (most recent first)
+        # Prefer DB-based DAG (no taskfile on disk needed); fall back to taskfile if unavailable.
         runs = data["runs"]
-        for run in runs:
-            candidate = run.get("taskfile_path")
-            if candidate and not candidate.startswith("TM1:") and os.path.isfile(candidate):
-                taskfile_path = candidate
-                break
+        workflow_runs = [r for r in runs if (r.get("workflow") or "") == args.workflow]
+        latest_run = workflow_runs[0] if workflow_runs else None
 
-        if taskfile_path:
+        dag_generated_from_db = False
+        if latest_run:
             try:
-                from rushti.taskfile_ops import visualize_dag
-
-                # Ensure output directory exists
-                Path(dag_path).parent.mkdir(parents=True, exist_ok=True)
-
-                visualize_dag(
-                    source=taskfile_path,
-                    output_path=dag_path,
-                    dashboard_url=dashboard_filename,
-                )
-                dag_generated = True
-                print(f"DAG visualization generated: {dag_path}")
+                from rushti.taskfile_ops import visualize_dag_from_db_results
+
+                latest_run_id = latest_run["run_id"]
+                latest_task_results = [
+                    tr for tr in data["task_results"] if tr["run_id"] == latest_run_id
+                ]
+
+                if latest_task_results:
+                    Path(dag_path).parent.mkdir(parents=True, exist_ok=True)
+                    visualize_dag_from_db_results(
+                        task_results=latest_task_results,
+                        output_path=dag_path,
+                        dashboard_url=dashboard_filename,
+                    )
+                    dag_generated = True
+                    dag_generated_from_db = True
+                    print(f"DAG visualization generated: {dag_path}")
             except Exception as e:
-                logger.warning(f"Could not generate DAG visualization: {e}")
-                print(f"Warning: DAG visualization skipped ({e})")
-        else:
-            print(
-                "Warning: No accessible taskfile found in run history, skipping DAG visualization"
-            )
+                logger.warning(f"Could not generate DAG from DB: {e}")
+
+        if not dag_generated_from_db:
+            # Fall back to taskfile on disk (most recent accessible one for this workflow)
+            taskfile_path = None
+            for run in workflow_runs:
+                candidate = run.get("taskfile_path")
+                if candidate and not candidate.startswith("TM1:") and os.path.isfile(candidate):
+                    taskfile_path = candidate
+                    break
+
+            if taskfile_path:
+                try:
+                    from rushti.taskfile_ops import visualize_dag
+
+                    Path(dag_path).parent.mkdir(parents=True, exist_ok=True)
+                    visualize_dag(
+                        source=taskfile_path,
+                        output_path=dag_path,
+                        dashboard_url=dashboard_filename,
+                    )
+                    dag_generated = True
+                    print(f"DAG visualization generated from taskfile: {dag_path}")
+                except Exception as e:
+                    logger.warning(f"Could not generate DAG visualization: {e}")
+                    print(f"Warning: DAG visualization skipped ({e})")
+            else:
+                print(
+                    "Warning: No DB task results or accessible taskfile found, skipping DAG visualization"
+                )
 
         # --- Generate dashboard ---
         output_file = generate_dashboard(

src/rushti/taskfile_ops.py

@@ -797,6 +797,72 @@ def validate_taskfile_full(
     return result
 
 
+def visualize_dag_from_db_results(
+    task_results: List[Dict[str, Any]],
+    output_path: str,
+    dashboard_url: Optional[str] = None,
+) -> str:
+    """Generate interactive HTML DAG visualization from DB task results.
+
+    Builds the DAG from task result records stored in the stats database,
+    so no taskfile on disk is required.
+
+    :param task_results: List of task result dicts for a single run (from get_visualization_data)
+    :param output_path: Path to output HTML file
+    :param dashboard_url: Optional relative URL back to the dashboard HTML
+    :return: Path to the generated HTML file
+    """
+    tasks_by_id: Dict[str, "TaskDefinition"] = {}
+    adjacency: Dict[str, List[str]] = {}
+
+    for tr in task_results:
+        task_id = tr.get("task_id")
+        if not task_id or task_id in tasks_by_id:
+            continue
+
+        params = tr.get("parameters")
+        if isinstance(params, str):
+            try:
+                params = json.loads(params)
+            except Exception:
+                params = {}
+        params = params or {}
+
+        predecessors = tr.get("predecessors")
+        if isinstance(predecessors, str):
+            try:
+                predecessors = json.loads(predecessors)
+            except Exception:
+                predecessors = []
+        predecessors = predecessors or []
+
+        tasks_by_id[task_id] = TaskDefinition(
+            id=task_id,
+            instance=tr.get("instance") or "",
+            process=tr.get("process") or "",
+            parameters=params,
+            predecessors=predecessors,
+            stage=tr.get("stage"),
+        )
+
+        for pred in predecessors:
+            if pred not in adjacency:
+                adjacency[pred] = []
+            adjacency[pred].append(task_id)
+
+    output_path_obj = Path(output_path)
+    output_base = (
+        str(output_path_obj.with_suffix("")) if output_path_obj.suffix else str(output_path_obj)
+    )
+
+    return _visualize_dag_html(
+        adjacency=adjacency,
+        tasks_by_id=tasks_by_id,
+        filename=output_base,
+        dashboard_url=dashboard_url,
+    )
+
+
 def _check_dag_cycles(tasks: List[TaskDefinition]) -> List[str]:
     """Check for cycles in task dependencies.