core: fix Progress.speed_bps reading 0 on the polled path

DAEMON reported Progress.speed_bps reading 0 for the whole life of a live
throttled download while downloaded bytes visibly advanced. DAEMON reads
progress by polling DownloadHandle::progress() (engine_port_core.hpp), not
the on_progress push callback.

DownloadTaskState::snapshot_progress() -- the body behind progress() -- never
set speed_bps, per-segment speed_bps, or eta_seconds at all; only the
event-driven emit_progress_if_due() (which drives the on_progress callback)
computed them, from the same live SegWorker::speed_bps EMA seg_data()
maintains. snapshot_progress() now reads that same per-worker speed while
building its segment list, so a segment with no live worker (idle, paused,
complete, failed) correctly reports 0 and a segment with an active transfer
reports its real EMA, matching emit_progress_if_due()'s math including the
eta_seconds derivation.

engine_polled_progress_reports_nonzero_speed reproduces the bug (fails
without the fix, confirmed) by polling .progress() -- the same path DAEMON
uses -- during a throttled download and asserting speed_bps > 0 once real
progress has accumulated.

Co-Authored-By: Claude Sonnet 5 <[email protected]>
Claude-Session: https://claude.ai/code/session_01Q3QrF7rCt21bkAjt9BCDFQ

core/src/task/download_task.cpp

@@ -1217,6 +1217,25 @@ EngineState DownloadTaskState::snapshot_state() {
 Progress DownloadTaskState::snapshot_progress() {
     Progress p;
     std::lock_guard lk(mu);
+    // Per-segment instantaneous speed lives on the live SegWorker (seg_data's 0.5s-sampled
+    // EMA, see the `speed_bps` update below) -- a segment with no live worker (idle,
+    // paused, complete, failed) has no speed to report and stays at SegmentProgress's
+    // default 0. Read every live worker's speed up front so the seg->snapshot() loop below
+    // (which covers *every* segment, not just live ones -- unlike emit_progress_if_due's
+    // push-callback version, which only ever reports the segments it currently has
+    // workers for) can look each one up by index.
+    double agg_speed = 0;
+    std::unordered_map<std::uint32_t, double> worker_speed;
+    {
+        std::shared_lock wl(workers_mu);
+        worker_speed.reserve(workers.size());
+        for (auto &[idx, w] : workers) {
+            worker_speed.emplace(idx, w->speed_bps);
+            agg_speed += w->speed_bps;
+        }
+        p.effective_segments = static_cast<std::uint32_t>(workers.size());
+    }
+    p.speed_bps = static_cast<std::uint64_t>(agg_speed);
     if (seg) {
         p.downloaded = seg->downloaded();
         for (auto &v : seg->snapshot()) {
@@ -1226,15 +1245,15 @@ Progress DownloadTaskState::snapshot_progress() {
             sp.end = v.end;
             sp.completed = v.completed;
             sp.state = v.state;
+            if (auto it = worker_speed.find(v.index); it != worker_speed.end())
+                sp.speed_bps = static_cast<std::uint64_t>(it->second);
             p.segments.push_back(sp);
         }
     }
     p.total = total_size;
-    {
-        std::shared_lock wl(workers_mu);
-        p.effective_segments = static_cast<std::uint32_t>(workers.size());
-    }
     p.effective_buffer_bytes = effective_buffer;
+    if (p.speed_bps > 0 && total_size && *total_size > p.downloaded)
+        p.eta_seconds = static_cast<std::uint32_t>((*total_size - p.downloaded) / p.speed_bps);
     return p;
 }
 

core/tests/task/engine_test.cpp

@@ -457,3 +457,33 @@ VT_TEST(engine_content_length_mismatch_fails_honestly) {
     VT_CHECK(rec.saw(EngineState::failed));
     VT_CHECK_EQ(::access(td.file("clm.bin").c_str(), F_OK), -1);  // never renamed into place
 }
+
+// --- DAEMON-reported bug: Progress.speed_bps reads 0 for the whole life of a live
+// download while downloaded bytes visibly advance. DAEMON reads progress by polling
+// DownloadHandle::progress() (engine_port_core.hpp), not the on_progress push callback --
+// this exercises exactly that path. ---
+
+VT_TEST(engine_polled_progress_reports_nonzero_speed) {
+    TestServer srv;
+    VT_REQUIRE(srv.available());
+    TmpDir td;
+    Recorder rec;
+    Engine eng;
+    auto h = eng.start(spec_for(srv, "/throttled/file/4M", td.file("sp.bin")), rec.cbs());
+
+    // Give it real, sustained progress: the speed estimate only updates on a >=0.5s
+    // sample window (seg_data), so a snapshot taken too early would legitimately read 0
+    // even with the bug fixed. Poll until downloaded has clearly advanced twice over.
+    std::uint64_t speed = 0;
+    for (int i = 0; i < 400 && speed == 0; ++i) {
+        std::this_thread::sleep_for(20ms);
+        auto p = h.progress();
+        if (p.downloaded >= 256u * 1024)
+            speed = p.speed_bps;
+    }
+    VT_CHECK(speed > 0);
+
+    h.cancel(/*discard_partial=*/true);
+    auto r = rec.wait();
+    VT_REQUIRE(!r.has_value());
+}