f

SYS/pipeline_progress.py

@@ -146,6 +146,26 @@ class PipelineProgress:
         except Exception:
             return
 
+    def begin_pipe(
+        self,
+        *,
+        total_items: int,
+        items_preview: Optional[Sequence[Any]] = None
+    ) -> None:
+        ui, pipe_idx = self.ui_and_pipe_index()
+        if ui is None:
+            return
+        try:
+            fn = getattr(ui, "begin_pipe", None)
+            if callable(fn):
+                fn(
+                    int(pipe_idx),
+                    total_items=max(1, int(total_items)),
+                    items_preview=list(items_preview or []),
+                )
+        except Exception:
+            return
+
     def on_emit(self, emitted: Any) -> None:
         """Advance local pipe progress after pipeline_context.emit().