hh

SYS/models.py

@@ -7,6 +7,7 @@ import os
 import shutil
 import sys
 import time
+from threading import RLock
 from dataclasses import dataclass, field
 from pathlib import Path
 from typing import Any, Callable, Dict, List, Optional, Protocol, TextIO
@@ -755,6 +756,7 @@ class PipelineLiveProgress:
     def __init__(self, pipe_labels: List[str], *, enabled: bool = True) -> None:
         self._enabled = bool(enabled)
         self._pipe_labels = [str(x) for x in (pipe_labels or [])]
+        self._lock = RLock()
 
         self._console: Optional[Console] = None
         self._live: Optional[Live] = None
@@ -826,26 +828,27 @@ class PipelineLiveProgress:
         the spinner without needing manual Live.update() calls.
         """
 
-        pipe_progress = self._pipe_progress
-        status = self._status
-        transfers = self._transfers
-        overall = self._overall
-        if pipe_progress is None or transfers is None or overall is None:
-            # Not started (or stopped).
-            yield Panel("", title="Pipeline", expand=False)
-            return
+        with self._lock:
+            pipe_progress = self._pipe_progress
+            status = self._status
+            transfers = self._transfers
+            overall = self._overall
+            if pipe_progress is None or transfers is None or overall is None:
+                # Not started (or stopped).
+                yield Panel("", title="Pipeline", expand=False)
+                return
 
-        body_parts: List[Any] = [pipe_progress]
-        if status is not None and self._status_tasks:
-            body_parts.append(status)
-        body_parts.append(transfers)
+            body_parts: List[Any] = [pipe_progress]
+            if status is not None and self._status_tasks:
+                body_parts.append(status)
+            body_parts.append(transfers)
 
-        yield Group(
-            Panel(Group(*body_parts),
-                  title=self._title_text(),
-                  expand=False),
-            overall
-        )
+            yield Group(
+                Panel(Group(*body_parts),
+                      title=self._title_text(),
+                      expand=False),
+                overall
+            )
 
     def _render_group(self) -> Group:
         # Backward-compatible helper (some callers may still expect a Group).
@@ -1029,52 +1032,58 @@ class PipelineLiveProgress:
             return
         if not self._ensure_pipe(int(pipe_index)):
             return
-        prog = self._status
-        if prog is None:
-            return
+        
+        with self._lock:
+            prog = self._status
+            if prog is None:
+                return
 
-        try:
-            pidx = int(pipe_index)
-            msg = str(text or "").strip()
-        except Exception:
-            return
-
-        # For long single-item work, hide the per-item spinner line and use this
-        # dedicated status line instead.
-        if self._pipe_percent_mode.get(pidx, False):
             try:
-                self._hide_pipe_subtasks(pidx)
+                pidx = int(pipe_index)
+                msg = str(text or "").strip()
+            except Exception:
+                return
+
+            # For long single-item work, hide the per-item spinner line and use this
+            # dedicated status line instead.
+            if self._pipe_percent_mode.get(pidx, False):
+                try:
+                    self._hide_pipe_subtasks(pidx)
+                except Exception:
+                    pass
+
+            task_id = self._status_tasks.get(pidx)
+            if task_id is None:
+                try:
+                    task_id = prog.add_task(msg)
+                except Exception:
+                    return
+                self._status_tasks[pidx] = task_id
+
+            try:
+                prog.update(task_id, description=msg, refresh=True)
             except Exception:
                 pass
 
-        task_id = self._status_tasks.get(pidx)
-        if task_id is None:
+    def clear_pipe_status_text(self, pipe_index: int) -> None:
+        if not self._enabled:
+            return
+        
+        with self._lock:
+            prog = self._status
+            if prog is None:
+                return
             try:
-                task_id = prog.add_task(msg)
+                pidx = int(pipe_index)
             except Exception:
                 return
-            self._status_tasks[pidx] = task_id
-
-        try:
-            prog.update(task_id, description=msg, refresh=True)
-        except Exception:
-            pass
-
-    def clear_pipe_status_text(self, pipe_index: int) -> None:
-        prog = self._status
-        if prog is None:
-            return
-        try:
-            pidx = int(pipe_index)
-        except Exception:
-            return
-        task_id = self._status_tasks.pop(pidx, None)
-        if task_id is None:
-            return
-        try:
-            prog.remove_task(task_id)
-        except Exception:
-            pass
+            task_id = self._status_tasks.pop(pidx, None)
+            if task_id is None:
+                return
+            try:
+                prog.remove_task(task_id)
+            except Exception:
+                pass
 
     def set_pipe_percent(self, pipe_index: int, percent: int) -> None:
         """Update the pipe bar as a percent (only when single-item mode is enabled)."""
@@ -1095,6 +1104,31 @@ class PipelineLiveProgress:
             pct = max(0, min(100, int(percent)))
             pipe_task = self._pipe_tasks[pidx]
             pipe_progress.update(pipe_task, completed=pct, total=100, refresh=True)
+            self._update_overall()
+        except Exception:
+            pass
+
+    def _update_overall(self) -> None:
+        """Update the overall pipeline progress task."""
+        if self._overall is None or self._overall_task is None:
+            return
+
+        completed = 0
+        try:
+            # Count a pipe as completed if its 'done' count matches or exceeds the advertised total.
+            completed = sum(
+                1 for i in range(len(self._pipe_labels))
+                if self._pipe_done[i] >= max(1, self._pipe_totals[i])
+            )
+        except Exception:
+            completed = 0
+
+        try:
+            self._overall.update(
+                self._overall_task,
+                completed=min(completed, max(1, len(self._pipe_labels))),
+                description=f"Pipeline: {completed}/{len(self._pipe_labels)} pipes completed",
+            )
         except Exception:
             pass
 
@@ -1108,24 +1142,25 @@ class PipelineLiveProgress:
         if not self._ensure_pipe(int(pipe_index)):
             return
 
-        try:
-            pidx = int(pipe_index)
-            tot = max(1, int(total_steps))
-        except Exception:
-            return
+        with self._lock:
+            try:
+                pidx = int(pipe_index)
+                tot = max(1, int(total_steps))
+            except Exception:
+                return
 
-        self._pipe_step_total[pidx] = tot
-        self._pipe_step_done[pidx] = 0
+            self._pipe_step_total[pidx] = tot
+            self._pipe_step_done[pidx] = 0
 
-        # Reset status line and percent.
-        try:
-            self.clear_pipe_status_text(pidx)
-        except Exception:
-            pass
-        try:
-            self.set_pipe_percent(pidx, 0)
-        except Exception:
-            pass
+            # Reset status line and percent.
+            try:
+                self.clear_pipe_status_text(pidx)
+            except Exception:
+                pass
+            try:
+                self.set_pipe_percent(pidx, 0)
+            except Exception:
+                pass
 
     def advance_pipe_step(self, pipe_index: int, text: str) -> None:
         """Advance the pipe's step counter by one.
@@ -1287,6 +1322,8 @@ class PipelineLiveProgress:
         except Exception:
             pass
 
+        self._update_overall()
+
         labels: List[str] = []
         if isinstance(items_preview, list) and items_preview:
             labels = [_pipeline_progress_item_label(x) for x in items_preview]
@@ -1372,6 +1409,8 @@ class PipelineLiveProgress:
         else:
             pipe_progress.update(pipe_task, completed=done)
 
+        self._update_overall()
+
         # Clear any status line now that it emitted.
         try:
             self.clear_pipe_status_text(pipe_index)
@@ -1452,23 +1491,7 @@ class PipelineLiveProgress:
         except Exception:
             pass
 
-        if self._overall_task is not None:
-            completed = 0
-            try:
-                completed = sum(
-                    1 for i in range(len(self._pipe_labels))
-                    if self._pipe_done[i] >= max(1, self._pipe_totals[i])
-                )
-            except Exception:
-                completed = 0
-            overall.update(
-                self._overall_task,
-                completed=min(completed,
-                              max(1,
-                                  len(self._pipe_labels))),
-                description=
-                f"Pipeline: {completed}/{len(self._pipe_labels)} pipes completed",
-            )
+        self._update_overall()
 
 
 class PipelineStageContext: