feat: enhance progress tracking for execution jobs

- Added logic to calculate and manage column-level progress for job executions.
- Introduced `progress_columns_total` and `_column_done` fields for more granular progress updates.
- Improved overall progress computation by considering total columns and individual progress per column.

studio/backend/core/data_recipe/jobs/manager.py

@@ -86,12 +86,25 @@ class JobManager:
 
     def start(self, *, recipe: dict, run: dict) -> str:
         """Spawn the job subprocess (one at a time, no cap)."""
+        llm_columns = recipe.get("columns") or []
+        llm_column_count = 0
+        if isinstance(llm_columns, list):
+            for column in llm_columns:
+                if not isinstance(column, dict):
+                    continue
+                column_type = str(column.get("column_type") or "").strip().lower()
+                if column_type.startswith("llm"):
+                    llm_column_count += 1
+        if llm_column_count <= 0:
+            llm_column_count = 1
+
         with self._lock:
             if self._proc is not None and self._proc.is_alive():
                 raise RuntimeError("job already running")
 
             job_id = uuid.uuid4().hex
             self._job = Job(job_id=job_id, status="pending", started_at=time.time())
+            self._job.progress_columns_total = llm_column_count
             self._events.clear()
             self._seq = 0
 

studio/backend/core/data_recipe/jobs/parse.py

@@ -198,39 +198,36 @@ def apply_update(job: Job, update: ParsedUpdate) -> None:
 
 
 def _compute_overall_progress(job: Job, column_progress: Progress) -> Progress:
-    if not job.rows or not job.current_column:
+    if not job.rows:
         return column_progress
 
     total_rows = max(1, int(job.rows))
-    total_cols = max(
-        1,
-        len(job._seen_generation_columns),
-        int(job.cols or 0),
-    )
     current_done = 0 if column_progress.done is None else int(column_progress.done)
     current_done = max(0, min(current_done, total_rows))
+    total_columns = max(1, int(job.progress_columns_total or 1))
 
-    try:
-        col_index = job._seen_generation_columns.index(job.current_column)
-    except ValueError:
-        col_index = max(0, len(job._seen_generation_columns) - 1)
+    if job.current_column:
+        job._column_done[job.current_column] = current_done
+
+    if len(job._column_done) == 0:
+        done = current_done
+    else:
+        sum_done = sum(max(0, min(value, total_rows)) for value in job._column_done.values())
+        done = int(sum_done / total_columns)
 
-    col_index = max(0, min(col_index, total_cols - 1))
-    total = total_rows * total_cols
-    done = min(total, (col_index * total_rows) + current_done)
     prev_done = int(job.progress.done or 0)
     if done < prev_done:
         done = prev_done
-    if done > total:
-        done = total
-    percent = (done / total) * 100 if total > 0 else 100.0
+    if done > total_rows:
+        done = total_rows
+    percent = (done / total_rows) * 100 if total_rows > 0 else 100.0
     prev_percent = float(job.progress.percent or 0.0)
     if percent < prev_percent:
         percent = prev_percent
 
     return Progress(
         done=done,
-        total=total,
+        total=total_rows,
         percent=percent,
         eta_sec=column_progress.eta_sec,
         rate=column_progress.rate,

studio/backend/core/data_recipe/jobs/types.py

@@ -65,6 +65,8 @@ class Job:
     dataset: list[dict[str, Any]] | None = None
     processor_artifacts: dict[str, Any] | None = None
     model_usage: dict[str, ModelUsage] = field(default_factory=dict)
+    progress_columns_total: int | None = None
     _current_usage_model: str | None = None
     _in_usage_summary: bool = False
     _seen_generation_columns: list[str] = field(default_factory=list)
+    _column_done: dict[str, int] = field(default_factory=dict)