perf: pipeline GSAM2 tracking + rendering with startup buffer

Pipeline tracking and rendering so segments stream as they're tracked
instead of waiting for all tracking to complete before rendering begins.

- Add on_segment callback to process_video() for incremental feeding
- Hoist render/writer infrastructure above tracking branch
- Single-GPU: callback-based incremental feeding via on_segment
- Multi-GPU: streaming reconciliation with segment_buffer pattern
- Writer startup buffer (60 frames) before streaming begins
- 3x frame duplication for 18 FPS stream (6 FPS processing throughput)
- Safety threshold: pause streaming when buffer < 20 frames
- try/finally sentinel safety for render worker shutdown
- Increase render_out queue from 64 to 128 for buffer headroom

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>

inference.py

@@ -1195,9 +1195,6 @@ def run_inference(
                                     display_labels.append("")
                                     continue
                                 lbl = d.get('label', 'obj')
-                                # Append Track ID
-                                if 'track_id' in d:
-                                    lbl = f"{d['track_id']} {lbl}"
                                 display_labels.append(lbl)
 
                             p_frame = draw_boxes(p_frame, display_boxes, label_names=display_labels)
@@ -1675,303 +1672,15 @@ def run_grounded_sam2_tracking(
 
         num_gpus = torch.cuda.device_count()
 
-        # ==================================================================
-        # Phase 1-4: Tracking  (single-GPU fallback vs multi-GPU pipeline)
-        # ==================================================================
-        if num_gpus <= 1:
-            # ---------- Single-GPU fallback ----------
-            device_str = "cuda:0" if torch.cuda.is_available() else "cpu"
-            _seg_kw = {"num_maskmem": num_maskmem} if num_maskmem is not None else {}
-            segmenter = load_segmenter_on_device(active_segmenter, device_str, **_seg_kw)
-            _check_cancellation(job_id)
-
-            if _perf_metrics is not None:
-                segmenter._perf_metrics = _perf_metrics
-                segmenter._perf_lock = None
-
-            if _perf_metrics is not None:
-                _t_track = time.perf_counter()
-
-            tracking_results = segmenter.process_video(
-                frame_dir, frame_names, queries,
-            )
-
-            if _perf_metrics is not None:
-                _perf_metrics["tracking_total_ms"] = (time.perf_counter() - _t_track) * 1000.0
-
-            logging.info(
-                "Single-GPU tracking complete: %d frames",
-                len(tracking_results),
-            )
-        else:
-            # ---------- Multi-GPU pipeline ----------
-            logging.info(
-                "Multi-GPU GSAM2 tracking: %d GPUs, %d frames, step=%d",
-                num_gpus, total_frames, step,
-            )
-
-            # Phase 1: Load one segmenter per GPU (parallel)
-            segmenters = []
-            with ThreadPoolExecutor(max_workers=num_gpus) as pool:
-                _seg_kw_multi = {"num_maskmem": num_maskmem} if num_maskmem is not None else {}
-                futs = [
-                    pool.submit(
-                        load_segmenter_on_device,
-                        active_segmenter,
-                        f"cuda:{i}",
-                        **_seg_kw_multi,
-                    )
-                    for i in range(num_gpus)
-                ]
-                segmenters = [f.result() for f in futs]
-            logging.info("Loaded %d segmenters", len(segmenters))
-
-            if _perf_metrics is not None:
-                import threading as _th
-                _actual_lock = _perf_lock or _th.Lock()
-                for seg in segmenters:
-                    seg._perf_metrics = _perf_metrics
-                    seg._perf_lock = _actual_lock
-
-            # Phase 2: Init SAM2 models/state per GPU (parallel)
-            def _init_seg_state(seg):
-                seg._ensure_models_loaded()
-                return seg._video_predictor.init_state(
-                    video_path=frame_dir,
-                    offload_video_to_cpu=True,
-                    async_loading_frames=True,
-                )
-
-            with ThreadPoolExecutor(max_workers=len(segmenters)) as pool:
-                futs = [pool.submit(_init_seg_state, seg) for seg in segmenters]
-                inference_states = [f.result() for f in futs]
-
-            if _perf_metrics is not None:
-                _t_track = time.perf_counter()
-
-            # Phase 3: Parallel segment processing (queue-based workers)
-            segments = list(range(0, total_frames, step))
-            seg_queue_in: Queue = Queue()
-            seg_queue_out: Queue = Queue()
-
-            for i, start_idx in enumerate(segments):
-                seg_queue_in.put((i, start_idx))
-            for _ in segmenters:
-                seg_queue_in.put(None)  # sentinel
-
-            iou_thresh = segmenters[0].iou_threshold
-
-            def _segment_worker(gpu_idx: int):
-                seg = segmenters[gpu_idx]
-                state = inference_states[gpu_idx]
-                device_type = seg.device.split(":")[0]
-                ac = (
-                    torch.autocast(device_type=device_type, dtype=torch.bfloat16)
-                    if device_type == "cuda"
-                    else nullcontext()
-                )
-                with ac:
-                    while True:
-                        if job_id:
-                            try:
-                                _check_cancellation(job_id)
-                            except RuntimeError as e:
-                                if "cancelled" in str(e).lower():
-                                    logging.info(
-                                        "Segment worker %d cancelled.",
-                                        gpu_idx,
-                                    )
-                                    break
-                                raise
-                        item = seg_queue_in.get()
-                        if item is None:
-                            break
-                        seg_idx, start_idx = item
-                        try:
-                            logging.info(
-                                "GPU %d processing segment %d (frame %d)",
-                                gpu_idx, seg_idx, start_idx,
-                            )
-                            img_path = os.path.join(
-                                frame_dir, frame_names[start_idx]
-                            )
-                            with PILImage.open(img_path) as pil_img:
-                                image = pil_img.convert("RGB")
-
-                            if job_id:
-                                _check_cancellation(job_id)
-                            masks, boxes, labels = seg.detect_keyframe(
-                                image, queries,
-                            )
-
-                            if masks is None:
-                                seg_queue_out.put(
-                                    (seg_idx, start_idx, None, {})
-                                )
-                                continue
-
-                            mask_dict = MaskDictionary()
-                            mask_dict.add_new_frame_annotation(
-                                mask_list=torch.tensor(masks).to(seg.device),
-                                box_list=(
-                                    boxes.clone()
-                                    if torch.is_tensor(boxes)
-                                    else torch.tensor(boxes)
-                                ),
-                                label_list=labels,
-                            )
-
-                            segment_output = seg.propagate_segment(
-                                state, start_idx, mask_dict, step,
-                            )
-                            seg_queue_out.put(
-                                (seg_idx, start_idx, mask_dict, segment_output)
-                            )
-                        except RuntimeError as e:
-                            if "cancelled" in str(e).lower():
-                                logging.info(
-                                    "Segment worker %d cancelled.",
-                                    gpu_idx,
-                                )
-                                break
-                            raise
-                        except Exception:
-                            logging.exception(
-                                "Segment %d failed on GPU %d",
-                                seg_idx, gpu_idx,
-                            )
-                            seg_queue_out.put(
-                                (seg_idx, start_idx, None, {})
-                            )
-
-            seg_workers = []
-            for i in range(num_gpus):
-                t = Thread(
-                    target=_segment_worker, args=(i,), daemon=True,
-                )
-                t.start()
-                seg_workers.append(t)
-
-            for t in seg_workers:
-                t.join()
-
-            # Collect all segment outputs
-            segment_data: Dict[int, Tuple] = {}
-            while not seg_queue_out.empty():
-                seg_idx, start_idx, mask_dict, segment_output = seg_queue_out.get()
-                segment_data[seg_idx] = (start_idx, mask_dict, segment_output)
-
-            # Phase 4: Sequential ID reconciliation
-            if _perf_metrics is not None:
-                _t_recon = time.perf_counter()
-
-            global_id_counter = 0
-            sam2_masks = MaskDictionary()
-            tracking_results: Dict[int, Dict[int, ObjectInfo]] = {}
-
-            def _mask_to_cpu(mask):
-                """Normalize mask to CPU tensor (still used for keyframe mask_dict)."""
-                if torch.is_tensor(mask):
-                    return mask.detach().cpu()
-                return mask
-
-            for seg_idx in sorted(segment_data.keys()):
-                start_idx, mask_dict, segment_output = segment_data[seg_idx]
-
-                if mask_dict is None or not mask_dict.labels:
-                    # No detections — carry forward previous masks
-                    for fi in range(
-                        start_idx, min(start_idx + step, total_frames)
-                    ):
-                        if fi not in tracking_results:
-                            tracking_results[fi] = (
-                                {
-                                    k: ObjectInfo(
-                                        instance_id=v.instance_id,
-                                        mask=v.mask,
-                                        class_name=v.class_name,
-                                        x1=v.x1, y1=v.y1,
-                                        x2=v.x2, y2=v.y2,
-                                    )
-                                    for k, v in sam2_masks.labels.items()
-                                }
-                                if sam2_masks.labels
-                                else {}
-                            )
-                    continue
-
-                # Normalize keyframe masks to CPU before cross-GPU IoU matching.
-                for info in mask_dict.labels.values():
-                    info.mask = _mask_to_cpu(info.mask)
-
-                # IoU match + get local→global remapping
-                global_id_counter, remapping = (
-                    mask_dict.update_masks_with_remapping(
-                        tracking_dict=sam2_masks,
-                        iou_threshold=iou_thresh,
-                        objects_count=global_id_counter,
-                    )
-                )
-
-                if not mask_dict.labels:
-                    for fi in range(
-                        start_idx, min(start_idx + step, total_frames)
-                    ):
-                        tracking_results[fi] = {}
-                    continue
-
-                # Bulk CPU transfer: 3 CUDA syncs total (was 100+ per-mask syncs)
-                segment_results = segment_output.to_object_dicts()
-
-                # Apply remapping to every frame in this segment
-                for frame_idx, frame_objects in segment_results.items():
-                    remapped: Dict[int, ObjectInfo] = {}
-                    for local_id, obj_info in frame_objects.items():
-                        global_id = remapping.get(local_id)
-                        if global_id is None:
-                            continue
-                        remapped[global_id] = ObjectInfo(
-                            instance_id=global_id,
-                            mask=obj_info.mask,
-                            class_name=obj_info.class_name,
-                            x1=obj_info.x1, y1=obj_info.y1,
-                            x2=obj_info.x2, y2=obj_info.y2,
-                        )
-                    tracking_results[frame_idx] = remapped
-
-                # Update running tracker with last frame of this segment
-                if segment_results:
-                    last_fi = max(segment_results.keys())
-                    last_objs = tracking_results.get(last_fi, {})
-                    sam2_masks = MaskDictionary()
-                    sam2_masks.labels = copy.deepcopy(last_objs)
-                    if last_objs:
-                        first_info = next(iter(last_objs.values()))
-                        if first_info.mask is not None:
-                            m = first_info.mask
-                            sam2_masks.mask_height = (
-                                m.shape[-2] if m.ndim >= 2 else 0
-                            )
-                            sam2_masks.mask_width = (
-                                m.shape[-1] if m.ndim >= 2 else 0
-                            )
-
-            if _perf_metrics is not None:
-                _perf_metrics["id_reconciliation_ms"] = (time.perf_counter() - _t_recon) * 1000.0
-                _perf_metrics["tracking_total_ms"] = (time.perf_counter() - _t_track) * 1000.0
-
-            logging.info(
-                "Multi-GPU reconciliation complete: %d frames, %d objects",
-                len(tracking_results), global_id_counter,
-            )
-
         # ==================================================================
         # Phase 5: Parallel rendering + sequential video writing
+        # (Hoisted above tracking so render pipeline starts before tracking
+        #  completes — segments are fed incrementally via callback / queue.)
         # ==================================================================
         _check_cancellation(job_id)
 
         render_in: Queue = Queue(maxsize=32)
-        render_out: Queue = Queue(maxsize=64)
+        render_out: Queue = Queue(maxsize=128)
         render_done = False
         num_render_workers = min(4, os.cpu_count() or 1)
 
@@ -2112,6 +1821,11 @@ def run_grounded_sam2_tracking(
             next_idx = 0
             buf: Dict[int, Tuple] = {}
 
+            # Streaming constants
+            STARTUP_BUFFER = 60
+            SAFETY_THRESHOLD = 20
+            FRAME_DUP = 3
+
             # Per-track bbox history (replaces ByteTracker for GSAM2)
             track_history: Dict[int, List] = {}
             speed_est = SpeedEstimator(fps=fps) if enable_gpt else None
@@ -2127,6 +1841,28 @@ def run_grounded_sam2_tracking(
                 with StreamingVideoWriter(
                     output_video_path, fps, width, height
                 ) as writer:
+                    # --- Phase 1: Startup buffering ---
+                    playback_started = False
+                    while not playback_started:
+                        try:
+                            idx, frm, fobjs = render_out.get(timeout=1.0)
+                            buf[idx] = (frm, fobjs)
+                        except Empty:
+                            if not any(t.is_alive() for t in r_workers) and render_out.empty():
+                                playback_started = True
+                                break
+                            continue
+
+                        ahead = sum(1 for k in buf if k >= next_idx)
+                        if ahead >= STARTUP_BUFFER or ahead >= total_frames:
+                            playback_started = True
+
+                    logging.info(
+                        "Startup buffer filled (%d frames), beginning playback",
+                        len(buf),
+                    )
+
+                    # --- Phase 2: Write + stream with safety gating ---
                     while next_idx < total_frames:
                         try:
                             while next_idx not in buf:
@@ -2239,22 +1975,40 @@ def run_grounded_sam2_tracking(
                             if _perf_metrics is not None:
                                 _t_w = time.perf_counter()
 
+                            # Write to video file (always, single copy)
                             writer.write(frm)
 
                             if _perf_metrics is not None:
                                 _perf_metrics["writer_total_ms"] += (time.perf_counter() - _t_w) * 1000.0
 
-                            if stream_queue:
-                                try:
-                                    from jobs.streaming import (
-                                        publish_frame as _pub,
-                                    )
+                            # --- Streaming with buffer gating + frame duplication ---
+                            if stream_queue or job_id:
+                                # Drain any immediately available frames for accurate buffer level
+                                while True:
+                                    try:
+                                        idx2, frm2, fobjs2 = render_out.get_nowait()
+                                        buf[idx2] = (frm2, fobjs2)
+                                    except Empty:
+                                        break
+
+                                buffer_ahead = sum(1 for k in buf if k > next_idx)
+
+                                if buffer_ahead >= SAFETY_THRESHOLD or next_idx >= total_frames - 1:
+                                    from jobs.streaming import publish_frame as _pub
                                     if job_id:
-                                        _pub(job_id, frm)
+                                        for _ in range(FRAME_DUP):
+                                            _pub(job_id, frm)
                                     else:
-                                        stream_queue.put(frm, timeout=0.01)
-                                except Exception:
-                                    pass
+                                        for _ in range(FRAME_DUP):
+                                            try:
+                                                stream_queue.put(frm, timeout=0.01)
+                                            except Exception:
+                                                pass
+                                else:
+                                    logging.debug(
+                                        "Stream paused: buffer=%d < threshold=%d at frame %d",
+                                        buffer_ahead, SAFETY_THRESHOLD, next_idx,
+                                    )
 
                             next_idx += 1
                             if next_idx % 30 == 0:
@@ -2285,15 +2039,335 @@ def run_grounded_sam2_tracking(
         writer_thread = Thread(target=_writer_loop, daemon=True)
         writer_thread.start()
 
-        # Feed render queue
-        for fidx in range(total_frames):
-            _check_cancellation(job_id)
-            fobjs = tracking_results.get(fidx, {})
-            render_in.put((fidx, fobjs))
+        # ==================================================================
+        # Phase 1-4: Tracking  (single-GPU fallback vs multi-GPU pipeline)
+        # Segments are fed incrementally to render_in as they complete.
+        # ==================================================================
+        try:
+            if num_gpus <= 1:
+                # ---------- Single-GPU fallback ----------
+                device_str = "cuda:0" if torch.cuda.is_available() else "cpu"
+                _seg_kw = {"num_maskmem": num_maskmem} if num_maskmem is not None else {}
+                segmenter = load_segmenter_on_device(active_segmenter, device_str, **_seg_kw)
+                _check_cancellation(job_id)
+
+                if _perf_metrics is not None:
+                    segmenter._perf_metrics = _perf_metrics
+                    segmenter._perf_lock = None
+
+                if _perf_metrics is not None:
+                    _t_track = time.perf_counter()
+
+                def _feed_segment(seg_frames):
+                    for fidx in sorted(seg_frames.keys()):
+                        render_in.put((fidx, seg_frames[fidx]))
+
+                tracking_results = segmenter.process_video(
+                    frame_dir, frame_names, queries,
+                    on_segment=_feed_segment,
+                )
+
+                if _perf_metrics is not None:
+                    _perf_metrics["tracking_total_ms"] = (time.perf_counter() - _t_track) * 1000.0
+
+                logging.info(
+                    "Single-GPU tracking complete: %d frames",
+                    len(tracking_results),
+                )
+            else:
+                # ---------- Multi-GPU pipeline ----------
+                logging.info(
+                    "Multi-GPU GSAM2 tracking: %d GPUs, %d frames, step=%d",
+                    num_gpus, total_frames, step,
+                )
+
+                # Phase 1: Load one segmenter per GPU (parallel)
+                segmenters = []
+                with ThreadPoolExecutor(max_workers=num_gpus) as pool:
+                    _seg_kw_multi = {"num_maskmem": num_maskmem} if num_maskmem is not None else {}
+                    futs = [
+                        pool.submit(
+                            load_segmenter_on_device,
+                            active_segmenter,
+                            f"cuda:{i}",
+                            **_seg_kw_multi,
+                        )
+                        for i in range(num_gpus)
+                    ]
+                    segmenters = [f.result() for f in futs]
+                logging.info("Loaded %d segmenters", len(segmenters))
+
+                if _perf_metrics is not None:
+                    import threading as _th
+                    _actual_lock = _perf_lock or _th.Lock()
+                    for seg in segmenters:
+                        seg._perf_metrics = _perf_metrics
+                        seg._perf_lock = _actual_lock
+
+                # Phase 2: Init SAM2 models/state per GPU (parallel)
+                def _init_seg_state(seg):
+                    seg._ensure_models_loaded()
+                    return seg._video_predictor.init_state(
+                        video_path=frame_dir,
+                        offload_video_to_cpu=True,
+                        async_loading_frames=True,
+                    )
 
-        # Sentinels for render workers
-        for _ in r_workers:
-            render_in.put(None)
+                with ThreadPoolExecutor(max_workers=len(segmenters)) as pool:
+                    futs = [pool.submit(_init_seg_state, seg) for seg in segmenters]
+                    inference_states = [f.result() for f in futs]
+
+                if _perf_metrics is not None:
+                    _t_track = time.perf_counter()
+
+                # Phase 3: Parallel segment processing (queue-based workers)
+                segments = list(range(0, total_frames, step))
+                num_total_segments = len(segments)
+                seg_queue_in: Queue = Queue()
+                seg_queue_out: Queue = Queue()
+
+                for i, start_idx in enumerate(segments):
+                    seg_queue_in.put((i, start_idx))
+                for _ in segmenters:
+                    seg_queue_in.put(None)  # sentinel
+
+                iou_thresh = segmenters[0].iou_threshold
+
+                def _segment_worker(gpu_idx: int):
+                    seg = segmenters[gpu_idx]
+                    state = inference_states[gpu_idx]
+                    device_type = seg.device.split(":")[0]
+                    ac = (
+                        torch.autocast(device_type=device_type, dtype=torch.bfloat16)
+                        if device_type == "cuda"
+                        else nullcontext()
+                    )
+                    with ac:
+                        while True:
+                            if job_id:
+                                try:
+                                    _check_cancellation(job_id)
+                                except RuntimeError as e:
+                                    if "cancelled" in str(e).lower():
+                                        logging.info(
+                                            "Segment worker %d cancelled.",
+                                            gpu_idx,
+                                        )
+                                        break
+                                    raise
+                            item = seg_queue_in.get()
+                            if item is None:
+                                break
+                            seg_idx, start_idx = item
+                            try:
+                                logging.info(
+                                    "GPU %d processing segment %d (frame %d)",
+                                    gpu_idx, seg_idx, start_idx,
+                                )
+                                img_path = os.path.join(
+                                    frame_dir, frame_names[start_idx]
+                                )
+                                with PILImage.open(img_path) as pil_img:
+                                    image = pil_img.convert("RGB")
+
+                                if job_id:
+                                    _check_cancellation(job_id)
+                                masks, boxes, labels = seg.detect_keyframe(
+                                    image, queries,
+                                )
+
+                                if masks is None:
+                                    seg_queue_out.put(
+                                        (seg_idx, start_idx, None, {})
+                                    )
+                                    continue
+
+                                mask_dict = MaskDictionary()
+                                mask_dict.add_new_frame_annotation(
+                                    mask_list=torch.tensor(masks).to(seg.device),
+                                    box_list=(
+                                        boxes.clone()
+                                        if torch.is_tensor(boxes)
+                                        else torch.tensor(boxes)
+                                    ),
+                                    label_list=labels,
+                                )
+
+                                segment_output = seg.propagate_segment(
+                                    state, start_idx, mask_dict, step,
+                                )
+                                seg_queue_out.put(
+                                    (seg_idx, start_idx, mask_dict, segment_output)
+                                )
+                            except RuntimeError as e:
+                                if "cancelled" in str(e).lower():
+                                    logging.info(
+                                        "Segment worker %d cancelled.",
+                                        gpu_idx,
+                                    )
+                                    break
+                                raise
+                            except Exception:
+                                logging.exception(
+                                    "Segment %d failed on GPU %d",
+                                    seg_idx, gpu_idx,
+                                )
+                                seg_queue_out.put(
+                                    (seg_idx, start_idx, None, {})
+                                )
+
+                seg_workers = []
+                for i in range(num_gpus):
+                    t = Thread(
+                        target=_segment_worker, args=(i,), daemon=True,
+                    )
+                    t.start()
+                    seg_workers.append(t)
+
+                # Phase 4: Streaming reconciliation — process segments in order
+                # as they arrive, feeding render_in incrementally.
+                if _perf_metrics is not None:
+                    _t_recon = time.perf_counter()
+
+                global_id_counter = 0
+                sam2_masks = MaskDictionary()
+                tracking_results: Dict[int, Dict[int, ObjectInfo]] = {}
+
+                def _mask_to_cpu(mask):
+                    """Normalize mask to CPU tensor (still used for keyframe mask_dict)."""
+                    if torch.is_tensor(mask):
+                        return mask.detach().cpu()
+                    return mask
+
+                next_seg_idx = 0
+                segment_buffer: Dict[int, Tuple] = {}
+
+                while next_seg_idx < num_total_segments:
+                    try:
+                        seg_idx, start_idx, mask_dict, segment_output = seg_queue_out.get(timeout=1.0)
+                    except Empty:
+                        if job_id:
+                            _check_cancellation(job_id)
+                        # Check if all segment workers are still alive
+                        if not any(t.is_alive() for t in seg_workers) and seg_queue_out.empty():
+                            logging.error(
+                                "All segment workers stopped while waiting for segment %d",
+                                next_seg_idx,
+                            )
+                            break
+                        continue
+                    segment_buffer[seg_idx] = (start_idx, mask_dict, segment_output)
+
+                    # Process contiguous ready segments in order
+                    while next_seg_idx in segment_buffer:
+                        start_idx, mask_dict, segment_output = segment_buffer.pop(next_seg_idx)
+
+                        if mask_dict is None or not mask_dict.labels:
+                            # No detections — carry forward previous masks
+                            for fi in range(
+                                start_idx, min(start_idx + step, total_frames)
+                            ):
+                                if fi not in tracking_results:
+                                    tracking_results[fi] = (
+                                        {
+                                            k: ObjectInfo(
+                                                instance_id=v.instance_id,
+                                                mask=v.mask,
+                                                class_name=v.class_name,
+                                                x1=v.x1, y1=v.y1,
+                                                x2=v.x2, y2=v.y2,
+                                            )
+                                            for k, v in sam2_masks.labels.items()
+                                        }
+                                        if sam2_masks.labels
+                                        else {}
+                                    )
+                                render_in.put((fi, tracking_results.get(fi, {})))
+                            next_seg_idx += 1
+                            continue
+
+                        # Normalize keyframe masks to CPU before cross-GPU IoU matching.
+                        for info in mask_dict.labels.values():
+                            info.mask = _mask_to_cpu(info.mask)
+
+                        # IoU match + get local→global remapping
+                        global_id_counter, remapping = (
+                            mask_dict.update_masks_with_remapping(
+                                tracking_dict=sam2_masks,
+                                iou_threshold=iou_thresh,
+                                objects_count=global_id_counter,
+                            )
+                        )
+
+                        if not mask_dict.labels:
+                            for fi in range(
+                                start_idx, min(start_idx + step, total_frames)
+                            ):
+                                tracking_results[fi] = {}
+                                render_in.put((fi, {}))
+                            next_seg_idx += 1
+                            continue
+
+                        # Bulk CPU transfer: 3 CUDA syncs total (was 100+ per-mask syncs)
+                        segment_results = segment_output.to_object_dicts()
+
+                        # Apply remapping to every frame in this segment
+                        for frame_idx, frame_objects in segment_results.items():
+                            remapped: Dict[int, ObjectInfo] = {}
+                            for local_id, obj_info in frame_objects.items():
+                                global_id = remapping.get(local_id)
+                                if global_id is None:
+                                    continue
+                                remapped[global_id] = ObjectInfo(
+                                    instance_id=global_id,
+                                    mask=obj_info.mask,
+                                    class_name=obj_info.class_name,
+                                    x1=obj_info.x1, y1=obj_info.y1,
+                                    x2=obj_info.x2, y2=obj_info.y2,
+                                )
+                            tracking_results[frame_idx] = remapped
+
+                        # Update running tracker with last frame of this segment
+                        if segment_results:
+                            last_fi = max(segment_results.keys())
+                            last_objs = tracking_results.get(last_fi, {})
+                            sam2_masks = MaskDictionary()
+                            sam2_masks.labels = copy.deepcopy(last_objs)
+                            if last_objs:
+                                first_info = next(iter(last_objs.values()))
+                                if first_info.mask is not None:
+                                    m = first_info.mask
+                                    sam2_masks.mask_height = (
+                                        m.shape[-2] if m.ndim >= 2 else 0
+                                    )
+                                    sam2_masks.mask_width = (
+                                        m.shape[-1] if m.ndim >= 2 else 0
+                                    )
+
+                        # Feed reconciled frames to render immediately
+                        for fi in range(start_idx, min(start_idx + step, total_frames)):
+                            render_in.put((fi, tracking_results.get(fi, {})))
+
+                        next_seg_idx += 1
+
+                for t in seg_workers:
+                    t.join()
+
+                if _perf_metrics is not None:
+                    _perf_metrics["id_reconciliation_ms"] = (time.perf_counter() - _t_recon) * 1000.0
+                    _perf_metrics["tracking_total_ms"] = (time.perf_counter() - _t_track) * 1000.0
+
+                logging.info(
+                    "Multi-GPU reconciliation complete: %d frames, %d objects",
+                    len(tracking_results), global_id_counter,
+                )
+        finally:
+            # Sentinels for render workers — always sent even on error/cancellation
+            for _ in r_workers:
+                try:
+                    render_in.put(None, timeout=5.0)
+                except Full:
+                    pass
 
         for t in r_workers:
             t.join()

models/segmenters/grounded_sam2.py

@@ -13,7 +13,7 @@ import logging
 import time
 from contextlib import nullcontext
 from dataclasses import dataclass, field
-from typing import Any, Dict, List, Optional, Sequence, Tuple, TYPE_CHECKING
+from typing import Any, Callable, Dict, List, Optional, Sequence, Tuple, TYPE_CHECKING
 
 import numpy as np
 import torch
@@ -673,6 +673,7 @@ class GroundedSAM2Segmenter(Segmenter):
         frame_dir: str,
         frame_names: List[str],
         text_prompts: List[str],
+        on_segment: Optional[Callable[[Dict[int, Dict[int, "ObjectInfo"]]], None]] = None,
     ) -> Dict[int, Dict[int, ObjectInfo]]:
         """Run full Grounded-SAM-2 tracking pipeline on extracted JPEG frames.
 
@@ -680,6 +681,8 @@ class GroundedSAM2Segmenter(Segmenter):
             frame_dir: Directory containing JPEG frames.
             frame_names: Sorted list of frame filenames.
             text_prompts: Text queries for Grounding DINO.
+            on_segment: Optional callback invoked after each segment completes.
+                Receives ``{frame_idx: {obj_id: ObjectInfo}}`` for the segment.
 
         Returns:
             Dict mapping frame_idx -> {obj_id: ObjectInfo} with masks,
@@ -764,6 +767,7 @@ class GroundedSAM2Segmenter(Segmenter):
                 if input_boxes.shape[0] == 0:
                     logging.info("No detections on keyframe %d, propagating previous masks", start_idx)
                     # Fill empty results for this segment
+                    seg_results: Dict[int, Dict[int, ObjectInfo]] = {}
                     for fi in range(start_idx, min(start_idx + step, total_frames)):
                         if fi not in all_results:
                             # Carry forward last known masks
@@ -776,6 +780,9 @@ class GroundedSAM2Segmenter(Segmenter):
                                 )
                                 for k, v in sam2_masks.labels.items()
                             } if sam2_masks.labels else {}
+                            seg_results[fi] = all_results[fi]
+                    if on_segment and seg_results:
+                        on_segment(seg_results)
                     continue
 
                 # -- SAM2 image predictor on keyframe --
@@ -831,8 +838,12 @@ class GroundedSAM2Segmenter(Segmenter):
                         _pm["id_reconciliation_ms"] += _d
 
                 if len(mask_dict.labels) == 0:
+                    seg_results_empty: Dict[int, Dict[int, ObjectInfo]] = {}
                     for fi in range(start_idx, min(start_idx + step, total_frames)):
                         all_results[fi] = {}
+                        seg_results_empty[fi] = {}
+                    if on_segment:
+                        on_segment(seg_results_empty)
                     continue
 
                 # -- SAM2 video predictor: propagate masks --
@@ -846,6 +857,8 @@ class GroundedSAM2Segmenter(Segmenter):
 
                 if segment_results:
                     all_results.update(segment_results)
+                    if on_segment:
+                        on_segment(segment_results)
                     last_fi = segment_output.last_frame_idx()
                     if last_fi is not None:
                         last_frame_objects = all_results.get(last_fi, {})