fix(progress): stop status updates throttling downloads (#11661)

* feat(progress): aggregate and coalesce gallery downloads Assisted-by: Codex:gpt-5 Signed-off-by: Ettore Di Giacinto <mudler@localai.io> * feat(ui): show rolling transfer speed Assisted-by: Codex:gpt-5 Signed-off-by: Ettore Di Giacinto <mudler@localai.io> * fix(ui): preserve legacy import byte labels Assisted-by: Codex:gpt-5 Signed-off-by: Ettore Di Giacinto <mudler@localai.io> --------- Signed-off-by: Ettore Di Giacinto <mudler@localai.io> Co-authored-by: Ettore Di Giacinto <mudler@localai.io>
2026-08-21 18:54:28 +02:00
parent 36ad21d1f5
commit 5429f569e0
12 changed files with 464 additions and 58 deletions
@@ -75,6 +75,9 @@ export default function OperationCard({ operation, onCancel, onPause, onDismiss,
    : ''
  const phaseKey = phaseKeys[operation.phase]
  const etaLabel = formatEta(operation.etaSeconds)
+  const rateLabel = Number.isFinite(operation.bytesPerSecond) && operation.bytesPerSecond > 0
+    ? `${formatBytes(operation.bytesPerSecond)}/s`
+    : ''
  // Same call the strip makes, for the same reason: a failed operation
  // stopped where it broke and a queued one has not moved, so neither has a
  // bar worth drawing.
@@ -124,7 +127,11 @@ export default function OperationCard({ operation, onCancel, onPause, onDismiss,
              <span className="operation-card__message" title={operation.message}>{operation.message}</span>
            )}
            {!failed && operation.isQueued && <span>{t('activity.waitingForInstaller')}</span>}
-            {!failed && byteLabel && <span className="operation-card__bytes">{byteLabel}</span>}
+            {!failed && byteLabel && (
+              <span className="operation-card__bytes">
+                {byteLabel}{rateLabel && ` · ${rateLabel}`}
+              </span>
+            )}
            {!failed && etaLabel && <span className="operation-card__bytes">{t('activity.timeLeft', { value: etaLabel })}</span>}
          </div>

@@ -78,6 +78,9 @@ export default function OperationsBar() {
  const byteLabel = Number.isFinite(shown.currentBytes) && Number.isFinite(shown.totalBytes) && shown.totalBytes > 0
    ? `${formatBytes(shown.currentBytes)} / ${formatBytes(shown.totalBytes)}`
    : ''
+  const rateLabel = Number.isFinite(shown.bytesPerSecond) && shown.bytesPerSecond > 0
+    ? `${formatBytes(shown.bytesPerSecond)}/s`
+    : ''
  const kind = shown.isBackend ? t('activity.kind.backend') : t('activity.kind.model')

  let modifier = ''
@@ -159,7 +162,12 @@ export default function OperationsBar() {
      <span className="operations-strip__name">{shown.name || shown.id}</span>
      {detail && <span className="operations-strip__sep" aria-hidden="true">·</span>}
      {detail && <span className="operations-strip__detail">{detail}</span>}
-      {byteLabel && !shown.error && <span className="operations-strip__bytes">{byteLabel}</span>}
+      {byteLabel && !shown.error && (
+        <span className="operations-strip__bytes">
+          {byteLabel}
+          {rateLabel && <span aria-live="off"> · {rateLabel}</span>}
+        </span>
+      )}
      <span className="operations-strip__spacer" />
      {showProgress && (
        <>
@@ -1,5 +1,6 @@
 import { createContext, useContext, useState, useEffect, useCallback, useMemo, useRef } from 'react'
 import { operationsApi } from '../utils/api'
+import { createTransferRateSampler } from '../utils/transferRate'
 import { useAuth } from '../context/AuthContext'

 // Serialize ops into a stable comparison key. Each op is a flat map of
@@ -167,63 +168,19 @@ export function OperationsProvider({ children, pollInterval = 1000 }) {
    }
  }, [fetchOperations])

-  // Time remaining is derived, not reported. We keep the previous
-  // (bytes, timestamp) sample per job and estimate from the delta.
-  //
-  // All or nothing on purpose: an estimate needs two samples, and one card
-  // showing "11 min left" while its neighbours show nothing reads as a
-  // rendering bug rather than as missing data.
-  const samplesRef = useRef(new Map())
+  const transferRateRef = useRef(createTransferRateSampler())
  const operationsWithEta = useMemo(() => {
    const now = Date.now()
-    const samples = samplesRef.current
    const seen = new Set()

    const withEta = operations.map((op) => {
      const key = op.jobID || op.id
      seen.add(key)
-      const current = op.currentBytes
-      const total = op.totalBytes
-      if (!Number.isFinite(current) || !Number.isFinite(total) || total <= 0) return op
-
-      const previous = samples.get(key)
-      samples.set(key, { bytes: current, at: now })
-      if (!previous || current <= previous.bytes) return op
-
-      const bytesPerMs = (current - previous.bytes) / Math.max(1, now - previous.at)
-      if (bytesPerMs <= 0) return op
-      return { ...op, etaSeconds: Math.round((total - current) / bytesPerMs / 1000) }
+      const metrics = transferRateRef.current.sample(key, op.currentBytes, op.totalBytes, now)
+      return Object.keys(metrics).length > 0 ? { ...op, ...metrics } : op
    })

-    // Drop samples for jobs that finished, so the map cannot grow forever.
-    for (const key of samples.keys()) {
-      if (!seen.has(key)) samples.delete(key)
-    }
-
-    // All or nothing: if any operation still transferring has no estimate yet,
-    // nobody shows one this tick.
-    //
-    // Only operations actually downloading get a vote. Every other phase
-    // reports bytes but stops advancing them: verifying hashes a finished file
-    // while the counter sits below the multi-file total, and committing sits
-    // pinned at the total. Both can last minutes, and counting them would
-    // blank every other operation's estimate for that whole window.
-    //
-    // The byte clauses are not redundant with the phase clause: a producer can
-    // report downloading with bytes already at the total. The undefined-phase
-    // arm keeps today's behaviour for producers that do not report a phase,
-    // which in practice do not report totalBytes either.
-    const tracked = withEta.filter(
-      (op) =>
-        Number.isFinite(op.totalBytes) &&
-        op.totalBytes > 0 &&
-        Number.isFinite(op.currentBytes) &&
-        op.currentBytes < op.totalBytes &&
-        (op.phase === undefined || op.phase === 'downloading')
-    )
-    if (tracked.length > 0 && tracked.some((op) => op.etaSeconds === undefined)) {
-      return withEta.map(({ etaSeconds: _etaSeconds, ...op }) => op)
-    }
+    transferRateRef.current.retain(seen)
    return withEta
  }, [operations])

@@ -2,6 +2,8 @@ import { useState, useRef, useCallback, useEffect, useMemo } from 'react'
 import { useNavigate, useOutletContext } from 'react-router-dom'
 import { useTranslation } from 'react-i18next'
 import { modelsApi, backendsApi } from '../utils/api'
+import { formatBytes } from '../utils/format'
+import { createTransferRateSampler } from '../utils/transferRate'
 import LoadingSpinner from '../components/LoadingSpinner'
 import PageHeader from '../components/PageHeader'
 import CodeEditor from '../components/CodeEditor'
@@ -136,6 +138,7 @@ export default function ImportModel() {
  // already reports progress, phase and byte counts, and the page used to
  // render only `message`.
  const [job, setJob] = useState(null)
+  const transferRateRef = useRef(createTransferRateSampler())

  const [prefs, setPrefs] = useState(DEFAULT_PREFS)
  const [customPrefs, setCustomPrefs] = useState([])
@@ -234,10 +237,12 @@ export default function ImportModel() {

  const startJobPolling = useCallback((jobId) => {
    if (pollRef.current) clearInterval(pollRef.current)
+    transferRateRef.current.retain([jobId])
    pollRef.current = setInterval(async () => {
      try {
        const data = await modelsApi.getJobStatus(jobId)
        if (data.completed) {
+          transferRateRef.current.reset(jobId)
          clearInterval(pollRef.current)
          pollRef.current = null
          setIsSubmitting(false)
@@ -247,6 +252,7 @@ export default function ImportModel() {
          return
        }
        if (data.error || (data.message && data.message.startsWith('error:'))) {
+          transferRateRef.current.reset(jobId)
          clearInterval(pollRef.current)
          pollRef.current = null
          setIsSubmitting(false)
@@ -263,7 +269,10 @@ export default function ImportModel() {
        // import endpoint registers it in the opcache) but drops it the moment
        // it finishes, which is indistinguishable from a cancel — so terminal
        // detection stays on this endpoint and only the rendering gets richer.
-        setJob(data)
+        const currentBytes = data.current_bytes
+        const totalBytes = data.total_bytes
+        const metrics = transferRateRef.current.sample(jobId, currentBytes, totalBytes)
+        setJob({ ...data, currentBytes, totalBytes, ...metrics })
      } catch (err) {
        console.error('Error polling job status:', err)
      }
@@ -566,8 +575,13 @@ export default function ImportModel() {
  // Everything the poller already returns and the old status card threw away.
  const progressPct = Number.isFinite(job?.progress) ? Math.round(job.progress) : null
  const jobName = job?.file_name || job?.gallery_element_name || ''
-  const jobBytes = job?.downloaded_size && job?.file_size
-    ? `${job.downloaded_size} / ${job.file_size}`
+  const jobBytes = Number.isFinite(job?.currentBytes) && Number.isFinite(job?.totalBytes) && job.totalBytes > 0
+    ? `${formatBytes(job.currentBytes)} / ${formatBytes(job.totalBytes)}`
+    : (job?.downloaded_size && job?.file_size
+        ? `${job.downloaded_size} / ${job.file_size}`
+        : '')
+  const jobRate = Number.isFinite(job?.bytesPerSecond) && job.bytesPerSecond > 0
+    ? `${formatBytes(job.bytesPerSecond)}/s`
    : ''

  return (
@@ -716,6 +730,7 @@ export default function ImportModel() {
                  <span className="import-progress__meta">
                    {job.phase || job.message || t('progress.working')}
                    {jobBytes && ` · ${jobBytes}`}
+                    {jobRate && ` · ${jobRate}`}
                  </span>
                </div>
              </div>
@@ -0,0 +1,52 @@
+const WINDOW_MS = 5_000
+
+export function createTransferRateSampler() {
+  const histories = new Map()
+
+  const reset = (jobID) => {
+    if (jobID === undefined) histories.clear()
+    else histories.delete(jobID)
+  }
+
+  const retain = (jobIDs) => {
+    const active = new Set(jobIDs)
+    for (const jobID of histories.keys()) {
+      if (!active.has(jobID)) histories.delete(jobID)
+    }
+  }
+
+  const sample = (jobID, currentBytes, totalBytes, now = Date.now()) => {
+    const valid = jobID !== undefined
+      && jobID !== null
+      && Number.isFinite(currentBytes)
+      && Number.isFinite(totalBytes)
+      && Number.isFinite(now)
+      && currentBytes >= 0
+      && totalBytes > 0
+      && currentBytes < totalBytes
+    if (!valid) {
+      if (jobID !== undefined && jobID !== null) reset(jobID)
+      return {}
+    }
+
+    let history = histories.get(jobID) || []
+    const newest = history.at(-1)
+    if (newest && (currentBytes < newest.bytes || now < newest.at)) history = []
+
+    history.push({ bytes: currentBytes, at: now })
+    history = history.filter((entry) => entry.at >= now - WINDOW_MS)
+    histories.set(jobID, history)
+
+    const oldest = history[0]
+    const elapsedSeconds = (now - oldest.at) / 1_000
+    const bytesPerSecond = (currentBytes - oldest.bytes) / elapsedSeconds
+    if (!Number.isFinite(bytesPerSecond) || bytesPerSecond <= 0) return {}
+
+    return {
+      bytesPerSecond,
+      etaSeconds: Math.round((totalBytes - currentBytes) / bytesPerSecond),
+    }
+  }
+
+  return { sample, retain, reset }
+}
@@ -0,0 +1,70 @@
+import assert from 'node:assert/strict'
+import test from 'node:test'
+
+import { createTransferRateSampler } from './transferRate.js'
+
+test('calculates speed and ETA from the oldest and newest samples in five seconds', () => {
+  const sampler = createTransferRateSampler()
+
+  assert.deepEqual(sampler.sample('job-1', 0, 10_000, 0), {})
+  assert.deepEqual(sampler.sample('job-1', 2_000, 10_000, 2_000), {
+    bytesPerSecond: 1_000,
+    etaSeconds: 8,
+  })
+  assert.deepEqual(sampler.sample('job-1', 8_000, 10_000, 7_000), {
+    bytesPerSecond: 1_200,
+    etaSeconds: 2,
+  })
+})
+
+test('resets samples after byte regression', () => {
+  const sampler = createTransferRateSampler()
+
+  sampler.sample('job-1', 4_000, 10_000, 0)
+  sampler.sample('job-1', 6_000, 10_000, 1_000)
+  assert.deepEqual(sampler.sample('job-1', 1_000, 10_000, 2_000), {})
+  assert.deepEqual(sampler.sample('job-1', 2_000, 10_000, 3_000), {
+    bytesPerSecond: 1_000,
+    etaSeconds: 8,
+  })
+})
+
+test('resets completed and invalid jobs', () => {
+  const sampler = createTransferRateSampler()
+
+  sampler.sample('job-1', 1_000, 10_000, 0)
+  assert.deepEqual(sampler.sample('job-1', 10_000, 10_000, 1_000), {})
+  assert.deepEqual(sampler.sample('job-1', 10_000, Number.NaN, 2_000), {})
+  assert.deepEqual(sampler.sample('job-1', 11_000, 20_000, 3_000), {})
+})
+
+test('keeps job histories independent and removes replaced jobs', () => {
+  const sampler = createTransferRateSampler()
+
+  sampler.sample('old-job', 1_000, 10_000, 0)
+  sampler.retain(['new-job'])
+  assert.deepEqual(sampler.sample('old-job', 2_000, 10_000, 1_000), {})
+  assert.deepEqual(sampler.sample('new-job', 1_000, 10_000, 1_000), {})
+})
+
+test('does not produce non-positive or non-finite rates', () => {
+  const sampler = createTransferRateSampler()
+
+  sampler.sample('job-1', 1_000, 10_000, 1_000)
+  assert.deepEqual(sampler.sample('job-1', 1_000, 10_000, 2_000), {})
+
+  sampler.reset('job-1')
+  sampler.sample('job-1', 1_000, 10_000, 2_000)
+  assert.deepEqual(sampler.sample('job-1', 2_000, 10_000, 2_000), {})
+})
+
+test('an invalid unnamed sample does not reset other jobs', () => {
+  const sampler = createTransferRateSampler()
+
+  sampler.sample('job-1', 1_000, 10_000, 0)
+  sampler.sample(undefined, 1_000, 10_000, 500)
+  assert.deepEqual(sampler.sample('job-1', 2_000, 10_000, 1_000), {
+    bytesPerSecond: 1_000,
+    etaSeconds: 8,
+  })
+})
@@ -1,12 +1,118 @@
 package galleryop

 import (
+	"strconv"
+	"strings"
 	"sync"
 	"time"

 	"github.com/mudler/LocalAI/pkg/modelartifacts"
 )

+type legacyProgressUpdate struct {
+	fileName   string
+	current    string
+	total      string
+	percentage float64
+}
+
+type legacyProgressCoalescer struct {
+	mu        sync.Mutex
+	forwardMu sync.Mutex
+	closed    bool
+	pending   *legacyProgressUpdate
+	ticker    artifactProgressTicker
+	done      chan struct{}
+	forward   func(legacyProgressUpdate)
+}
+
+func newLegacyProgressCoalescer(interval time.Duration, forward func(legacyProgressUpdate)) *legacyProgressCoalescer {
+	c := &legacyProgressCoalescer{
+		ticker:  newArtifactProgressTicker(interval),
+		done:    make(chan struct{}),
+		forward: forward,
+	}
+	go c.run()
+	return c
+}
+
+func (c *legacyProgressCoalescer) Sink(fileName, current, total string, percentage float64) {
+	c.mu.Lock()
+	defer c.mu.Unlock()
+	if c.closed {
+		return
+	}
+	c.pending = &legacyProgressUpdate{fileName: fileName, current: current, total: total, percentage: percentage}
+}
+
+func (c *legacyProgressCoalescer) Close() {
+	c.forwardMu.Lock()
+	defer c.forwardMu.Unlock()
+	c.mu.Lock()
+	if c.closed {
+		c.mu.Unlock()
+		return
+	}
+	c.closed = true
+	pending := c.pending
+	c.pending = nil
+	close(c.done)
+	c.ticker.Stop()
+	c.mu.Unlock()
+	c.forwardUpdate(pending)
+}
+
+func (c *legacyProgressCoalescer) run() {
+	for {
+		select {
+		case <-c.ticker.Chan():
+			c.flush()
+		case <-c.done:
+			return
+		}
+	}
+}
+
+func (c *legacyProgressCoalescer) flush() {
+	c.forwardMu.Lock()
+	defer c.forwardMu.Unlock()
+	c.mu.Lock()
+	if c.closed {
+		c.mu.Unlock()
+		return
+	}
+	pending := c.pending
+	c.pending = nil
+	c.mu.Unlock()
+	c.forwardUpdate(pending)
+}
+
+func (c *legacyProgressCoalescer) forwardUpdate(update *legacyProgressUpdate) {
+	if update != nil && c.forward != nil {
+		c.forward(*update)
+	}
+}
+
+func parseDisplayedBytes(value string) (int64, bool) {
+	parts := strings.Fields(value)
+	if len(parts) != 2 {
+		return 0, false
+	}
+	number, err := strconv.ParseFloat(parts[0], 64)
+	if err != nil || number < 0 {
+		return 0, false
+	}
+	multipliers := map[string]float64{
+		"B": 1, "KiB": 1 << 10, "MiB": 1 << 20, "GiB": 1 << 30,
+		"TiB": 1 << 40, "PiB": 1 << 50, "EiB": 1 << 60,
+	}
+	multiplier, ok := multipliers[parts[1]]
+	if !ok {
+		return 0, false
+	}
+	return int64(number * multiplier), true
+}
+
 type artifactProgressTicker interface {
 	Chan() <-chan time.Time
 	Stop()
@@ -61,6 +61,18 @@ func (m *modelOperationProgressManager) InstallModel(ctx context.Context, _ *Man

 func (m *modelOperationProgressManager) DeleteModel(string) error { return nil }

+type legacyModelOperationProgressManager struct {
+	err error
+}
+
+func (m *legacyModelOperationProgressManager) InstallModel(_ context.Context, _ *ManagementOp[gallery.GalleryModel, gallery.ModelConfig], progress ProgressCallback) error {
+	progress("model.bin", "25 B", "100 B", 25)
+	progress("model.bin", "50 B", "100 B", 50)
+	return m.err
+}
+
+func (m *legacyModelOperationProgressManager) DeleteModel(string) error { return nil }
+
 type recordingProgressClient struct {
 	mu      sync.Mutex
 	updates []*OpStatus
@@ -187,4 +199,42 @@ var _ = Describe("artifact progress coalescer", func() {
 			Cancellable:        true,
 		}))
 	})
+
+	It("coalesces legacy callback progress and flushes numeric bytes on close", func() {
+		installErr := errors.New("stop after legacy progress")
+		progressClient := &recordingProgressClient{}
+		service := NewGalleryService(&config.ApplicationConfig{}, nil)
+		service.modelManager = &legacyModelOperationProgressManager{err: installErr}
+		service.natsClient = progressClient
+		op := &ManagementOp[gallery.GalleryModel, gallery.ModelConfig]{
+			ID:                 "legacy-model-operation",
+			GalleryElementName: "model",
+			Context:            context.Background(),
+		}
+
+		Expect(service.modelHandler(op, nil, nil)).To(MatchError(installErr))
+		status := service.GetStatus(op.ID)
+		Expect(status).NotTo(BeNil())
+		Expect(status.Progress).To(Equal(float64(50)))
+		Expect(status.CurrentBytes).To(Equal(int64(50)))
+		Expect(status.TotalBytes).To(Equal(int64(100)))
+		Expect(status.DownloadedFileSize).To(Equal("50 B"))
+	})
+
+	It("forwards only the latest legacy callback update on each tick", func() {
+		updates := make(chan legacyProgressUpdate, 2)
+		coalescer := newLegacyProgressCoalescer(250*time.Millisecond, func(update legacyProgressUpdate) {
+			updates <- update
+		})
+		DeferCleanup(coalescer.Close)
+
+		coalescer.Sink("model.bin", "25 B", "100 B", 25)
+		coalescer.Sink("model.bin", "50 B", "100 B", 50)
+		Consistently(updates).ShouldNot(Receive())
+
+		ticker.channel <- time.Now()
+		Eventually(updates).Should(Receive(Equal(legacyProgressUpdate{
+			fileName: "model.bin", current: "50 B", total: "100 B", percentage: 50,
+		})))
+	})
 })
@@ -6,6 +6,7 @@ import (
 	"fmt"
 	"path/filepath"
 	"strings"
+	"time"

 	"github.com/mudler/LocalAI/core/config"
 	"github.com/mudler/LocalAI/core/gallery"
@@ -55,6 +56,18 @@ func (g *GalleryService) backendHandler(op *ManagementOp[gallery.GalleryBackend,
 	// ctx. DeleteBackend takes no context and cannot be interrupted, so a Cancel
 	// button on a running removal is one the server cannot honour.
 	g.UpdateStatus(op.ID, &OpStatus{Message: fmt.Sprintf("processing backend: %s", op.GalleryElementName), Progress: 0, Cancellable: !op.Delete})
+	legacyCoalescer := newLegacyProgressCoalescer(250*time.Millisecond, func(update legacyProgressUpdate) {
+		status := &OpStatus{Message: fmt.Sprintf(processingMessage, update.fileName, update.total, update.current), FileName: update.fileName, Progress: update.percentage, TotalFileSize: update.total, DownloadedFileSize: update.current, Cancellable: true}
+		if currentBytes, ok := parseDisplayedBytes(update.current); ok {
+			if totalBytes, totalOK := parseDisplayedBytes(update.total); totalOK {
+				status.CurrentBytes = currentBytes
+				status.TotalBytes = totalBytes
+			}
+		}
+		status.GalleryElementName = op.GalleryElementName
+		g.UpdateStatus(op.ID, status)
+	})
+	defer legacyCoalescer.Close()

 	// displayDownload displays the download progress
 	progressCallback := func(fileName string, current string, total string, percentage float64) {
@@ -66,7 +79,7 @@ func (g *GalleryService) backendHandler(op *ManagementOp[gallery.GalleryBackend,
 			default:
 			}
 		}
-		g.UpdateStatus(op.ID, &OpStatus{Message: fmt.Sprintf(processingMessage, fileName, total, current), FileName: fileName, Progress: percentage, TotalFileSize: total, DownloadedFileSize: current, Cancellable: true})
+		legacyCoalescer.Sink(fileName, current, total, percentage)
 		utils.DisplayDownloadFunction(fileName, current, total, percentage)
 	}

@@ -88,6 +101,7 @@ func (g *GalleryService) backendHandler(op *ManagementOp[gallery.GalleryBackend,
 		}
 	}
 	if err != nil {
+		legacyCoalescer.Close()
 		// Check if error is due to cancellation
 		if op.Context != nil && errors.Is(err, op.Context.Err()) {
 			g.UpdateStatus(op.ID, &OpStatus{
@@ -136,6 +150,7 @@ func (g *GalleryService) backendHandler(op *ManagementOp[gallery.GalleryBackend,
 		Op:      opName,
 	})

+	legacyCoalescer.Close()
 	g.UpdateStatus(op.ID,
 		&OpStatus{
 			Deletion:           op.Delete,
@@ -72,6 +72,19 @@ func (g *GalleryService) modelHandler(op *ManagementOp[gallery.GalleryModel, gal
 		operationCtx = context.Background()
 	}
 	operationCtx = modelartifacts.WithProgressSink(operationCtx, coalescer.Sink)
+	legacyCoalescer := newLegacyProgressCoalescer(250*time.Millisecond, func(update legacyProgressUpdate) {
+		percentage := bridge.ClampLegacy(update.percentage)
+		status := &OpStatus{Message: fmt.Sprintf(processingMessage, update.fileName, update.total, update.current), FileName: update.fileName, Progress: percentage, TotalFileSize: update.total, DownloadedFileSize: update.current, Cancellable: true}
+		if currentBytes, ok := parseDisplayedBytes(update.current); ok {
+			if totalBytes, totalOK := parseDisplayedBytes(update.total); totalOK {
+				status.CurrentBytes = currentBytes
+				status.TotalBytes = totalBytes
+			}
+		}
+		status.GalleryElementName = op.GalleryElementName
+		g.UpdateStatus(op.ID, status)
+	})
+	defer legacyCoalescer.Close()

 	// displayDownload displays the download progress
 	progressCallback := func(fileName string, current string, total string, percentage float64) {
@@ -83,8 +96,7 @@ func (g *GalleryService) modelHandler(op *ManagementOp[gallery.GalleryModel, gal
 			default:
 			}
 		}
-		percentage = bridge.ClampLegacy(percentage)
-		g.UpdateStatus(op.ID, &OpStatus{Message: fmt.Sprintf(processingMessage, fileName, total, current), FileName: fileName, Progress: percentage, TotalFileSize: total, DownloadedFileSize: current, Cancellable: true})
+		legacyCoalescer.Sink(fileName, current, total, percentage)
 		utils.DisplayDownloadFunction(fileName, current, total, percentage)
 	}

@@ -95,6 +107,7 @@ func (g *GalleryService) modelHandler(op *ManagementOp[gallery.GalleryModel, gal
 		err = g.modelManager.InstallModel(operationCtx, op, progressCallback)
 	}
 	if err != nil {
+		legacyCoalescer.Close()
 		// Check if error is due to cancellation
 		if op.Context != nil && errors.Is(err, op.Context.Err()) {
 			g.UpdateStatus(op.ID, &OpStatus{
@@ -112,6 +125,7 @@ func (g *GalleryService) modelHandler(op *ManagementOp[gallery.GalleryModel, gal
 	if op.Context != nil {
 		select {
 		case <-op.Context.Done():
+			legacyCoalescer.Close()
 			g.UpdateStatus(op.ID, &OpStatus{
 				Cancelled:          true,
 				Processed:          true,
@@ -147,6 +161,7 @@ func (g *GalleryService) modelHandler(op *ManagementOp[gallery.GalleryModel, gal
 		Op:      op2,
 	})

+	legacyCoalescer.Close()
 	g.UpdateStatus(op.ID,
 		&OpStatus{
 			Deletion:           op.Delete,
@@ -47,7 +47,23 @@ func DownloadFilesWithConcurrency(ctx context.Context, tasks []FileTask, status
 		concurrency = 1
 	}

-	if status != nil && concurrency > 1 {
+	var aggregateTotal int64
+	aggregateAvailable := status != nil && len(tasks) > 0
+	if aggregateAvailable {
+		for _, task := range tasks {
+			size, err := task.URI.ContentLength(ctx)
+			if err != nil || size < 0 {
+				aggregateAvailable = false
+				break
+			}
+			aggregateTotal += size
+		}
+	}
+	if aggregateTotal <= 0 {
+		aggregateAvailable = false
+	}
+
+	if status != nil && concurrency > 1 && !aggregateAvailable {
 		var statusMutex sync.Mutex
 		unsynchronized := status
 		status = func(fileName, current, total string, percent float64) {
@@ -63,9 +79,12 @@ func DownloadFilesWithConcurrency(ctx context.Context, tasks []FileTask, status
 	// context.Canceled the siblings observe.
 	group, groupCtx := errgroup.WithContext(ctx)
 	group.SetLimit(concurrency)
+	aggregateWritten := make([]int64, len(tasks))
+	var aggregateMutex sync.Mutex

 	for i := range tasks {
 		task := tasks[i]
+		taskIndex := i
 		if err := groupCtx.Err(); err != nil {
 			break
 		}
@@ -75,7 +94,27 @@ func DownloadFilesWithConcurrency(ctx context.Context, tasks []FileTask, status
 			}
 			taskOpts := append([]DownloadOption{}, opts...)
 			taskOpts = append(taskOpts, task.Options...)
-			if err := downloadTaskWithRetry(groupCtx, task, status, taskOpts); err != nil {
+			taskStatus := status
+			if aggregateAvailable {
+				existingSink := applyDownloadOptions(taskOpts).transferProgress
+				taskOpts = append(taskOpts, WithTransferProgress(func(event TransferProgress) {
+					if existingSink != nil {
+						existingSink(event)
+					}
+					aggregateMutex.Lock()
+					aggregateWritten[taskIndex] = event.Written
+					var written int64
+					for _, taskWritten := range aggregateWritten {
+						written += taskWritten
+					}
+					if status != nil {
+						status(event.FileName, formatBytes(written), formatBytes(aggregateTotal), float64(written)*100/float64(aggregateTotal))
+					}
+					aggregateMutex.Unlock()
+				}))
+				taskStatus = nil
+			}
+			if err := downloadTaskWithRetry(groupCtx, task, taskStatus, taskOpts); err != nil {
 				return err
 			}
 			if task.AfterDownload != nil {
@@ -1,11 +1,14 @@
 package downloader_test

 import (
+	"bytes"
 	"context"
+	"fmt"
 	"net/http"
 	"net/http/httptest"
 	"os"
 	"path/filepath"
+	"sync"

 	. "github.com/onsi/ginkgo/v2"
 	. "github.com/onsi/gomega"
@@ -41,4 +44,73 @@ var _ = Describe("DownloadFilesWithContext", func() {
 		Expect(err).NotTo(HaveOccurred())
 		Expect(hookCalled).To(BeTrue())
 	})
+
+	It("weights progress by bytes and preserves transfer sinks", func() {
+		payloads := map[string][]byte{
+			"/small": bytes.Repeat([]byte("s"), 10),
+			"/large": bytes.Repeat([]byte("l"), 90),
+		}
+		server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+			payload := payloads[r.URL.Path]
+			w.Header().Set("Content-Length", fmt.Sprintf("%d", len(payload)))
+			if r.Method != http.MethodHead {
+				_, _ = w.Write(payload)
+			}
+		}))
+		DeferCleanup(server.Close)
+
+		var mu sync.Mutex
+		percentages := []float64{}
+		events := []downloader.TransferProgress{}
+		tasks := []downloader.FileTask{}
+		for index, path := range []string{"/small", "/large"} {
+			tasks = append(tasks, downloader.FileTask{
+				URI:         downloader.URI(server.URL + path),
+				Destination: filepath.Join(GinkgoT().TempDir(), filepath.Base(path)),
+				FileIndex:   index,
+				TotalFiles:  2,
+			})
+		}
+
+		err := downloader.DownloadFilesWithContext(context.Background(), tasks, func(_ string, _, _ string, percentage float64) {
+			mu.Lock()
+			defer mu.Unlock()
+			percentages = append(percentages, percentage)
+		}, downloader.WithTransferProgress(func(event downloader.TransferProgress) {
+			mu.Lock()
+			defer mu.Unlock()
+			events = append(events, event)
+		}))
+
+		Expect(err).NotTo(HaveOccurred())
+		Expect(percentages).NotTo(BeEmpty())
+		Expect(percentages[0]).To(BeNumerically("~", 10.0, 0.01))
+		Expect(percentages).To(HaveEach(BeNumerically("<=", 100)))
+		Expect(events).To(HaveLen(2))
+	})
+
+	It("falls back to per-file progress when a size cannot be resolved", func() {
+		payload := []byte("downloaded")
+		server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+			if r.Method == http.MethodHead {
+				http.Error(w, "unavailable", http.StatusServiceUnavailable)
+				return
+			}
+			_, _ = w.Write(payload)
+		}))
+		DeferCleanup(server.Close)
+
+		percentages := []float64{}
+		err := downloader.DownloadFilesWithContext(context.Background(), []downloader.FileTask{{
+			URI:         downloader.URI(server.URL),
+			Destination: filepath.Join(GinkgoT().TempDir(), "fallback.bin"),
+			TotalFiles:  1,
+		}}, func(_ string, _, _ string, percentage float64) {
+			percentages = append(percentages, percentage)
+		})
+
+		Expect(err).NotTo(HaveOccurred())
+		Expect(percentages).NotTo(BeEmpty())
+		Expect(percentages[len(percentages)-1]).To(Equal(float64(100)))
+	})
 })