diff --git a/studio/backend/core/data_recipe/jobs/manager.py b/studio/backend/core/data_recipe/jobs/manager.py index e5e05b0868..a71e8eac8e 100644 --- a/studio/backend/core/data_recipe/jobs/manager.py +++ b/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 diff --git a/studio/backend/core/data_recipe/jobs/parse.py b/studio/backend/core/data_recipe/jobs/parse.py index 4a3ca99000..9096216676 100644 --- a/studio/backend/core/data_recipe/jobs/parse.py +++ b/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, diff --git a/studio/backend/core/data_recipe/jobs/types.py b/studio/backend/core/data_recipe/jobs/types.py index f532741370..24a63d062c 100644 --- a/studio/backend/core/data_recipe/jobs/types.py +++ b/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)