diff --git a/cli/tests/client_test.cpp b/cli/tests/client_test.cpp
index 4819abb..92e8beb 100644
--- a/cli/tests/client_test.cpp
+++ b/cli/tests/client_test.cpp
@@ -11,6 +11,7 @@
 #include "client.hpp"
 #include "rpc/dispatcher.hpp"
 #include "rpc/event_loop.hpp"
+#include "rpc/event_hub.hpp"
 #include "rpc/uds_server.hpp"
 #include "store/migrations.hpp"
 #include "store/settings.hpp"
@@ -59,8 +60,9 @@ void run() {
         CHECK(settings.set_raw("saveTo.defaultDir",
                                "\"" + g_allowed_root + "\"").has_value());
     }
-    rpc::VeloxDispatcher dispatcher(*db);
-    rpc::UdsServer server(loop, dispatcher, server_sock);
+    rpc::EventHub hub;
+    rpc::VeloxDispatcher dispatcher(*db, hub);
+    rpc::UdsServer server(loop, dispatcher, hub, server_sock);
     const auto ec = server.start();
     CHECK(!ec);
     if (ec) return;
diff --git a/daemon/CMakeLists.txt b/daemon/CMakeLists.txt
index 1a3273d..4559005 100644
--- a/daemon/CMakeLists.txt
+++ b/daemon/CMakeLists.txt
@@ -35,6 +35,8 @@ add_library(veloxd_store STATIC
     src/store/pairings.cpp
     src/store/settings.cpp
     src/store/tasks.cpp
+    src/store/categories.cpp
+    src/store/queues.cpp
     ${_mig_hdr}
 )
 add_library(velox::daemon_store ALIAS veloxd_store)
@@ -65,12 +67,13 @@ add_library(velox::daemon_sched ALIAS veloxd_sched)
 target_include_directories(veloxd_sched PUBLIC ${CMAKE_CURRENT_SOURCE_DIR}/src)
 target_compile_features(veloxd_sched PUBLIC cxx_std_23)
 target_compile_options(veloxd_sched PRIVATE -Wall -Wextra -Wpedantic -Werror)
-target_link_libraries(veloxd_sched PUBLIC velox::proto velox::core veloxd_store nlohmann_json::nlohmann_json)
+target_link_libraries(veloxd_sched PUBLIC velox::proto velox::core veloxd_store veloxd_rpc nlohmann_json::nlohmann_json)
 
 # --- veloxd_rpc — the RPC transports + dispatcher ------------------------------------
 add_library(veloxd_rpc STATIC
     src/rpc/runtime_dir.cpp
     src/rpc/event_loop.cpp
+    src/rpc/event_hub.cpp
     src/rpc/uds_server.cpp
     src/rpc/ws_frame.cpp
     src/rpc/ws_handshake.cpp
diff --git a/daemon/docs/deferrals.md b/daemon/docs/deferrals.md
index b8e99b0..73778e6 100644
--- a/daemon/docs/deferrals.md
+++ b/daemon/docs/deferrals.md
@@ -7,7 +7,7 @@ close. Kept here (not buried in commit messages) so the next pass can see them a
 |---|---|---|---|---|
 | D1 | Pairing prompt is `EnvAutoApprover` (needs `VELOX_PAIR_AUTO=1`) | `rpc/pairing.hpp`, `main.cpp` | A GUI dialog / `org.freedesktop.Notifications` approver is integration work | Build step 7 (systemd + notifications) |
 | D2 | `download.probe` → `-32603` | `rpc/dispatcher.cpp` | `download.add` is wired (`fs/safepath` + store, real `-32011`); `download.probe` needs the engine's probe path for `-32013` | probe with the engine link (CORE stage 3 is landed; wire `Engine::probe`) |
-| D3 | Stub handlers for everything except `session.*`, `download.add/list/get` | `rpc/dispatcher.cpp` | No store behind them yet (categories/queues/rules/settings/limiter/schedule) | Per method, as the store query modules land behind them |
+| D3 | Stub handlers for the rest: `download.pause/resume/start/cancel/remove/addBatch/refreshUrl/provideAuth`, `rules.*`, `settings.*`, `limiter.*`, `schedule.*`, `queue.upsert/reorder/start/stop`, `category.upsert/remove`, `grabber.*`, `media.*` | `rpc/dispatcher.cpp` | No store/scheduler wiring behind them yet. `category.list` and `queue.list` are done (`store/categories`, `store/queues`) | Per method, as each wires to the store/scheduler |
 | ~~D4a~~ | **Closed** — `sched/engine_port_core.hpp` wraps `vdm::Engine` + `segment_budget()`; `main.cpp` constructs `Engine` + `Scheduler`, calls `reconcile_after_restart` / `reload_config` / `tick` at startup | — | — | done (`lane/core` stage 8 merged) |
 | D4b | timer + nudges: a 1 s `timerfd` re-runs `Scheduler::tick()` and `download.add` nudges via `on_mutation`. `download.pause`/`resume`/`start`/`cancel` and the queue.* handlers still don't touch the scheduler | `rpc/dispatcher.cpp` | those handlers are still stubs (D3) | as each handler is implemented behind the store, it calls `on_mutation` / drives the scheduler |
-| D5 | `event.*` fan-out not implemented; `session.subscribe` accepts and echoes but nothing is emitted | `rpc/uds_server.cpp`, `rpc/ws_server.cpp` | No task state to broadcast until the engine is wired. `Scheduler::on_engine_state` is the hook it will fire from | With D4a — the same engine-state callback feeds both the store and `event.task.state` |
+| ~~D5~~ | **Mostly closed** — `rpc/event_hub` fans out per-subscription; `session.subscribe` on both transports registers/updates/tears down a real subscription; `Scheduler::transition()` publishes `event.task.state` (with `previousState`) on every state change, scheduler-driven or engine-reported; `dispatcher::on_download_add` publishes `event.task.added`; a 250 ms timer batches `Scheduler::progress_snapshot()` into one `event.task.progress` array per AGENT-DAEMON.md item 5 / the schema's `x-maxRateHz: 4`. Verified live end to end. | — | `event.task.removed` has no source yet (`download.remove` is D3); `event.speed.global`, `event.notify`, `event.auth.required`, `event.settings.changed`, `event.grabber.progress` are unpublished — each lands with its owning handler | as each owning D3 handler lands |
diff --git a/daemon/src/main.cpp b/daemon/src/main.cpp
index 26a8c88..14a1567 100644
--- a/daemon/src/main.cpp
+++ b/daemon/src/main.cpp
@@ -17,7 +17,10 @@
 
 #include <sys/timerfd.h>
 
+#include <nlohmann/json.hpp>
+
 #include "rpc/dispatcher.hpp"
+#include "rpc/event_hub.hpp"
 #include "rpc/event_loop.hpp"
 #include "rpc/pairing.hpp"
 #include "rpc/runtime_dir.hpp"
@@ -28,6 +31,7 @@
 #include "sched/scheduler.hpp"
 #include "store/migrations.hpp"
 #include "store/sqlite.hpp"
+#include "util/time.hpp"
 #include "vdm/engine.hpp"
 #include "version.hpp"
 
@@ -108,10 +112,11 @@ int main() {
     }
 
     // --- engine + scheduler ---------------------------------------------------------
+    velox::daemon::rpc::EventHub hub;
     vdm::Engine engine;
     velox::daemon::sched::EnginePortCore engine_port(engine);
     velox::daemon::sched::Scheduler scheduler(
-        *db, engine_port, velox::daemon::sched::Governor{},
+        *db, engine_port, velox::daemon::sched::Governor{}, &hub,
         {/*local_now*/ {},
          /*post_to_loop*/ [&loop](std::function<void()> fn) { loop.post(std::move(fn)); }});
 
@@ -120,7 +125,7 @@ int main() {
     (void)scheduler.reload_config();
     (void)scheduler.tick();  // admit anything already queued in the DB
 
-    velox::daemon::rpc::VeloxDispatcher dispatcher(*db);
+    velox::daemon::rpc::VeloxDispatcher dispatcher(*db, hub);
     dispatcher.set_on_mutation([&loop, &scheduler] {
         loop.post([&scheduler] { (void)scheduler.tick(); });
     });
@@ -140,7 +145,45 @@ int main() {
         });
     }
 
-    velox::daemon::rpc::UdsServer uds(loop, dispatcher, rt.socket_path());
+    // event.task.progress: one array message at <=4 Hz (schema x-maxRateHz), never one
+    // notification per task (AGENT-DAEMON.md item 5). 250 ms keeps every active task's
+    // segment bar under 4 Hz without depending on how many tasks are running.
+    const int progress_fd = ::timerfd_create(CLOCK_MONOTONIC, TFD_NONBLOCK | TFD_CLOEXEC);
+    if (progress_fd >= 0) {
+        itimerspec spec{};
+        spec.it_value.tv_nsec = 250'000'000;
+        spec.it_interval.tv_nsec = 250'000'000;
+        ::timerfd_settime(progress_fd, 0, &spec, nullptr);
+        loop.add_fd(progress_fd, velox::daemon::rpc::kRead, [&](int fd, unsigned) {
+            std::uint64_t ticks = 0;
+            [[maybe_unused]] ssize_t n = ::read(fd, &ticks, sizeof(ticks));
+            const auto rows = scheduler.progress_snapshot();
+            if (rows.empty()) return;
+
+            nlohmann::json tasks_json = nlohmann::json::array();
+            for (const auto& r : rows) {
+                nlohmann::json t{{"taskId", r.task_id},
+                                 {"downloadedBytes", r.downloaded_bytes},
+                                 {"speedBps", r.speed_bps}};
+                t["etaSeconds"] = r.eta_seconds ? nlohmann::json(*r.eta_seconds) : nlohmann::json(nullptr);
+                if (!r.segments.empty()) {
+                    nlohmann::json segs = nlohmann::json::array();
+                    for (const auto& s : r.segments)
+                        segs.push_back({{"index", s.index},
+                                       {"downloadedBytes", s.downloaded_bytes},
+                                       {"speedBps", s.speed_bps}});
+                    t["segments"] = std::move(segs);
+                }
+                tasks_json.push_back(std::move(t));
+            }
+            const nlohmann::json params{{"tasks", std::move(tasks_json)},
+                                        {"at", velox::daemon::now_iso()}};
+            hub.publish(velox::proto::Event::TaskProgress,
+                       velox::proto::make_notification(velox::proto::Event::TaskProgress, params));
+        });
+    }
+
+    velox::daemon::rpc::UdsServer uds(loop, dispatcher, hub, rt.socket_path());
     if (const auto ec = uds.start()) {
         std::cerr << "veloxd: cannot listen on " << rt.socket_path() << ": " << ec.message()
                   << "\n";
@@ -154,7 +197,7 @@ int main() {
     // TODO(build step 7): replace EnvAutoApprover with a GUI-dialog / desktop-notification
     // approver. Until then pairing needs VELOX_PAIR_AUTO=1.
     velox::daemon::rpc::EnvAutoApprover approver;
-    velox::daemon::rpc::WsServer ws(loop, dispatcher, *db, approver, rt);
+    velox::daemon::rpc::WsServer ws(loop, dispatcher, *db, approver, hub, rt);
     if (const auto ec = ws.start()) {
         std::cerr << "veloxd: WebSocket transport unavailable (" << ec.message()
                   << "); the extension fallback will not work this run\n";
@@ -169,6 +212,10 @@ int main() {
         loop.del_fd(tick_fd);
         ::close(tick_fd);
     }
+    if (progress_fd >= 0) {
+        loop.del_fd(progress_fd);
+        ::close(progress_fd);
+    }
     g_loop = nullptr;
     ::close(lock_fd);
     return 0;
diff --git a/daemon/src/rpc/dispatcher.cpp b/daemon/src/rpc/dispatcher.cpp
index 150d916..014a5ad 100644
--- a/daemon/src/rpc/dispatcher.cpp
+++ b/daemon/src/rpc/dispatcher.cpp
@@ -6,6 +6,8 @@
 #include <nlohmann/json.hpp>
 
 #include "fs/safepath.hpp"
+#include "store/categories.hpp"
+#include "store/queues.hpp"
 #include "store/settings.hpp"
 #include "store/tasks.hpp"
 #include "util/time.hpp"
@@ -168,6 +170,14 @@ VeloxDispatcher::on_download_add(const proto::DownloadSpec& spec) {
         return std::unexpected(proto::HandlerError{proto::ErrorCode::InternalError,
                                                    "download.add: " + ins.error().message});
 
+    // event.task.added carries the summary so a client inserts the row without a
+    // follow-up download.get (schema note on the event).
+    hub_.publish(proto::Event::TaskAdded,
+                proto::make_notification(proto::Event::TaskAdded,
+                                         nlohmann::json{{"taskId", row.task_id},
+                                                        {"summary", store::to_summary(row)}}),
+                row.task_id);
+
     if (on_mutation_) on_mutation_();
 
     proto::DownloadAddResult r;
@@ -188,7 +198,14 @@ VeloxDispatcher::on_capture_offer(const proto::CaptureOfferParams&) {
 }
 proto::HandlerResult<proto::CategoryListResult>
 VeloxDispatcher::on_category_list(const proto::CategoryListParams&) {
-    return not_implemented<proto::CategoryListResult>("category.list");
+    store::Categories categories(db_);
+    auto items = categories.list();
+    if (!items)
+        return std::unexpected(proto::HandlerError{proto::ErrorCode::InternalError,
+                                                   "category.list: " + items.error().message});
+    proto::CategoryListResult r;
+    r.items = std::move(*items);
+    return r;
 }
 proto::HandlerResult<proto::CategoryRemoveResult>
 VeloxDispatcher::on_category_remove(const proto::CategoryRemoveParams&) {
@@ -289,7 +306,14 @@ VeloxDispatcher::on_media_listVariants(const proto::MediaListVariantsParams&) {
 }
 proto::HandlerResult<proto::QueueListResult>
 VeloxDispatcher::on_queue_list(const proto::QueueListParams&) {
-    return not_implemented<proto::QueueListResult>("queue.list");
+    store::Queues queues(db_);
+    auto items = queues.list();
+    if (!items)
+        return std::unexpected(proto::HandlerError{proto::ErrorCode::InternalError,
+                                                   "queue.list: " + items.error().message});
+    proto::QueueListResult r;
+    r.items = std::move(*items);
+    return r;
 }
 proto::HandlerResult<proto::QueueReorderResult>
 VeloxDispatcher::on_queue_reorder(const proto::QueueReorderParams&) {
diff --git a/daemon/src/rpc/dispatcher.hpp b/daemon/src/rpc/dispatcher.hpp
index 1868203..126b59c 100644
--- a/daemon/src/rpc/dispatcher.hpp
+++ b/daemon/src/rpc/dispatcher.hpp
@@ -14,6 +14,7 @@
 
 #include <functional>
 
+#include "rpc/event_hub.hpp"
 #include "store/sqlite.hpp"
 #include "velox_proto.hpp"
 
@@ -21,7 +22,7 @@ namespace velox::daemon::rpc {
 
 class VeloxDispatcher final : public velox::proto::Dispatcher {
 public:
-    explicit VeloxDispatcher(velox::daemon::store::Db& db) : db_(db) {}
+    VeloxDispatcher(velox::daemon::store::Db& db, EventHub& hub) : db_(db), hub_(hub) {}
 
     // Called after a handler mutates task state (download.add for now). main.cpp wires it
     // to nudge the scheduler; unset in tests.
@@ -107,6 +108,7 @@ public:
 
 private:
     velox::daemon::store::Db& db_;
+    EventHub& hub_;
     std::function<void()> on_mutation_;
 };
 
diff --git a/daemon/src/rpc/event_hub.cpp b/daemon/src/rpc/event_hub.cpp
new file mode 100644
index 0000000..807ee7f
--- /dev/null
+++ b/daemon/src/rpc/event_hub.cpp
@@ -0,0 +1,55 @@
+#include "rpc/event_hub.hpp"
+
+#include <algorithm>
+
+#include <nlohmann/json.hpp>
+
+namespace velox::daemon::rpc {
+
+namespace proto = velox::proto;
+
+EventHub::SubId EventHub::subscribe(Sink sink) {
+    std::lock_guard<std::mutex> lk(mu_);
+    const SubId id = next_++;
+    subs_.emplace(id, Sub{std::move(sink), {}, std::nullopt});
+    return id;
+}
+
+void EventHub::set_filter(SubId id, std::vector<proto::Event> events,
+                          std::optional<std::vector<std::string>> task_ids) {
+    std::lock_guard<std::mutex> lk(mu_);
+    if (auto it = subs_.find(id); it != subs_.end()) {
+        it->second.events = std::move(events);
+        it->second.task_ids = std::move(task_ids);
+    }
+}
+
+void EventHub::unsubscribe(SubId id) {
+    std::lock_guard<std::mutex> lk(mu_);
+    subs_.erase(id);
+}
+
+void EventHub::publish(proto::Event kind, const nlohmann::json& notification,
+                       std::string_view task_id) {
+    // Copy the sinks to call out to while holding the lock only long enough to build the
+    // list — a sink runs arbitrary connection code (framing + a write syscall) and must
+    // not run with mu_ held.
+    std::vector<Sink> targets;
+    {
+        std::lock_guard<std::mutex> lk(mu_);
+        targets.reserve(subs_.size());
+        for (const auto& [id, sub] : subs_) {
+            (void)id;
+            if (std::find(sub.events.begin(), sub.events.end(), kind) == sub.events.end())
+                continue;
+            if (!task_id.empty() && sub.task_ids &&
+                std::find(sub.task_ids->begin(), sub.task_ids->end(), task_id) ==
+                    sub.task_ids->end())
+                continue;
+            targets.push_back(sub.sink);
+        }
+    }
+    for (const auto& sink : targets) sink(notification);
+}
+
+}  // namespace velox::daemon::rpc
diff --git a/daemon/src/rpc/event_hub.hpp b/daemon/src/rpc/event_hub.hpp
new file mode 100644
index 0000000..40b7be0
--- /dev/null
+++ b/daemon/src/rpc/event_hub.hpp
@@ -0,0 +1,63 @@
+#pragma once
+
+// Per-subscription event fan-out, shared by both transports. A connection subscribes once
+// (session.subscribe) with the event kinds and optional task-id filter it wants; publish()
+// delivers a pre-built notification to every subscription that asked for that kind and
+// passes the per-task filter.
+//
+// event.task.progress is the one call site that matters for load: it is batched by the
+// caller (Scheduler::progress_snapshot + one publish) into a single array message at
+// <=4 Hz, never one publish per task — that batching happens before this class ever sees
+// it (AGENT-DAEMON.md item 5, event.task.progress.schema.json x-maxRateHz).
+
+#include <cstdint>
+#include <functional>
+#include <mutex>
+#include <optional>
+#include <string>
+#include <string_view>
+#include <unordered_map>
+#include <vector>
+
+#include <nlohmann/json_fwd.hpp>
+
+#include "velox_proto.hpp"
+
+namespace velox::daemon::rpc {
+
+class EventHub {
+public:
+    using SubId = std::uint64_t;
+    using Sink = std::function<void(const nlohmann::json&)>;
+
+    // Register a connection with no interest yet; session.subscribe calls set_filter to
+    // actually turn events on. Returns the id to unsubscribe with on disconnect.
+    SubId subscribe(Sink sink);
+
+    // Replaces the subscription's event set and task filter (session.subscribe replaces,
+    // never adds — matches the method's own description).
+    void set_filter(SubId id, std::vector<velox::proto::Event> events,
+                    std::optional<std::vector<std::string>> task_ids);
+
+    void unsubscribe(SubId id);
+
+    // `notification` is a complete {jsonrpc, method, params} object
+    // (velox::proto::make_notification). `task_id` is matched against each subscription's
+    // filter when set; empty means "not task-scoped" and reaches every subscriber of
+    // `kind` regardless of their filter.
+    void publish(velox::proto::Event kind, const nlohmann::json& notification,
+                std::string_view task_id = {});
+
+private:
+    struct Sub {
+        Sink sink;
+        std::vector<velox::proto::Event> events;
+        std::optional<std::vector<std::string>> task_ids;
+    };
+
+    std::mutex mu_;
+    std::unordered_map<SubId, Sub> subs_;
+    SubId next_ = 1;
+};
+
+}  // namespace velox::daemon::rpc
diff --git a/daemon/src/rpc/uds_server.cpp b/daemon/src/rpc/uds_server.cpp
index 9437090..5b3c36c 100644
--- a/daemon/src/rpc/uds_server.cpp
+++ b/daemon/src/rpc/uds_server.cpp
@@ -56,8 +56,9 @@ json rpc_error(const json& id, proto::ErrorCode code, std::string_view msg, json
 
 }  // namespace
 
-UdsServer::UdsServer(EventLoop& loop, proto::Dispatcher& dispatcher, std::string socket_path)
-    : loop_(loop), dispatcher_(dispatcher), path_(std::move(socket_path)) {}
+UdsServer::UdsServer(EventLoop& loop, proto::Dispatcher& dispatcher, EventHub& hub,
+                    std::string socket_path)
+    : loop_(loop), dispatcher_(dispatcher), hub_(hub), path_(std::move(socket_path)) {}
 
 UdsServer::~UdsServer() {
     for (auto& [fd, c] : conns_) {
@@ -251,11 +252,21 @@ bool UdsServer::handle_session_method(Conn& c, const std::string& method, const
                               json{{"path", p.error().path}});
             return true;
         }
-        // Event fan-out is not wired yet; accept the subscription and echo it back so a
-        // client can already register its interest without erroring.
+        if (!c.sub_id) {
+            const int fd = c.fd;
+            c.sub_id = hub_.subscribe([this, fd](const json& n) {
+                if (const auto it = conns_.find(fd); it != conns_.end()) queue_reply(*it->second, n);
+            });
+        }
+        std::vector<proto::Event> events;
         proto::SessionSubscribeResult r;
         r.ok = true;
-        for (const auto& ev : p->events) r.events.emplace_back(proto::to_string(ev));
+        for (const auto& ev : p->events) {
+            const auto name = proto::to_string(ev);
+            r.events.emplace_back(name);
+            if (auto e = proto::event_from_string(name)) events.push_back(*e);
+        }
+        hub_.set_filter(*c.sub_id, std::move(events), p->taskIds);
         reply = proto::make_result(id, r);
         return true;
     }
@@ -300,6 +311,7 @@ void UdsServer::flush(Conn& c) {
 
 void UdsServer::close_conn(int fd) {
     if (const auto it = conns_.find(fd); it != conns_.end()) {
+        if (it->second->sub_id) hub_.unsubscribe(*it->second->sub_id);
         loop_.del_fd(fd);
         ::close(fd);
         conns_.erase(it);
diff --git a/daemon/src/rpc/uds_server.hpp b/daemon/src/rpc/uds_server.hpp
index 48cf389..dd88faf 100644
--- a/daemon/src/rpc/uds_server.hpp
+++ b/daemon/src/rpc/uds_server.hpp
@@ -13,12 +13,14 @@
 #include <cstdint>
 #include <memory>
 #include <string>
+#include <optional>
 #include <system_error>
 #include <unordered_map>
 #include <vector>
 
 #include <nlohmann/json_fwd.hpp>
 
+#include "rpc/event_hub.hpp"
 #include "rpc/ndjson.hpp"
 #include "velox_proto.hpp"
 
@@ -28,7 +30,8 @@ class EventLoop;
 
 class UdsServer {
 public:
-    UdsServer(EventLoop& loop, velox::proto::Dispatcher& dispatcher, std::string socket_path);
+    UdsServer(EventLoop& loop, velox::proto::Dispatcher& dispatcher, EventHub& hub,
+             std::string socket_path);
     ~UdsServer();
 
     UdsServer(const UdsServer&) = delete;
@@ -51,6 +54,7 @@ private:
         bool close_after_flush = false;
         bool hello_ok = false;
         std::string session_id;
+        std::optional<EventHub::SubId> sub_id;
     };
 
     void on_listener_readable();
@@ -68,6 +72,7 @@ private:
 
     EventLoop& loop_;
     velox::proto::Dispatcher& dispatcher_;
+    EventHub& hub_;
     std::string path_;
     int listen_fd_ = -1;
     bool bound_ = false;   // path_ is ours to unlink on destruction
diff --git a/daemon/src/rpc/ws_server.cpp b/daemon/src/rpc/ws_server.cpp
index 36ca5b1..a3f2aa3 100644
--- a/daemon/src/rpc/ws_server.cpp
+++ b/daemon/src/rpc/ws_server.cpp
@@ -60,11 +60,12 @@ json rpc_error(const json& id, proto::ErrorCode code, std::string_view msg, json
 }  // namespace
 
 WsServer::WsServer(EventLoop& loop, proto::Dispatcher& dispatcher, store::Db& db,
-                   PairingApprover& approver, RuntimeDir runtime)
+                   PairingApprover& approver, EventHub& hub, RuntimeDir runtime)
     : loop_(loop),
       dispatcher_(dispatcher),
       db_(db),
       approver_(approver),
+      hub_(hub),
       runtime_(std::move(runtime)) {}
 
 WsServer::~WsServer() {
@@ -364,9 +365,21 @@ bool WsServer::handle_session_ws(Conn& c, const std::string& method, const json&
                               json{{"path", p.error().path}});
             return true;
         }
+        if (!c.sub_id) {
+            const int fd = c.fd;
+            c.sub_id = hub_.subscribe([this, fd](const json& n) {
+                if (const auto it = conns_.find(fd); it != conns_.end()) send_text(*it->second, n);
+            });
+        }
+        std::vector<proto::Event> events;
         proto::SessionSubscribeResult r;
         r.ok = true;
-        for (const auto& ev : p->events) r.events.emplace_back(proto::to_string(ev));
+        for (const auto& ev : p->events) {
+            const auto name = proto::to_string(ev);
+            r.events.emplace_back(name);
+            if (auto e = proto::event_from_string(name)) events.push_back(*e);
+        }
+        hub_.set_filter(*c.sub_id, std::move(events), p->taskIds);
         reply = proto::make_result(id, r);
         return true;
     }
@@ -423,6 +436,7 @@ void WsServer::flush(Conn& c) {
 
 void WsServer::close_conn(int fd) {
     if (const auto it = conns_.find(fd); it != conns_.end()) {
+        if (it->second->sub_id) hub_.unsubscribe(*it->second->sub_id);
         loop_.del_fd(fd);
         ::close(fd);
         conns_.erase(it);
diff --git a/daemon/src/rpc/ws_server.hpp b/daemon/src/rpc/ws_server.hpp
index 72e467a..2f8e8a9 100644
--- a/daemon/src/rpc/ws_server.hpp
+++ b/daemon/src/rpc/ws_server.hpp
@@ -8,12 +8,14 @@
 
 #include <cstdint>
 #include <memory>
+#include <optional>
 #include <string>
 #include <system_error>
 #include <unordered_map>
 
 #include <nlohmann/json_fwd.hpp>
 
+#include "rpc/event_hub.hpp"
 #include "rpc/pairing.hpp"
 #include "rpc/runtime_dir.hpp"
 #include "rpc/ws_frame.hpp"
@@ -30,7 +32,7 @@ class EventLoop;
 class WsServer {
 public:
     WsServer(EventLoop& loop, velox::proto::Dispatcher& dispatcher, velox::daemon::store::Db& db,
-             PairingApprover& approver, RuntimeDir runtime);
+             PairingApprover& approver, EventHub& hub, RuntimeDir runtime);
     ~WsServer();
 
     WsServer(const WsServer&) = delete;
@@ -60,6 +62,7 @@ private:
         bool authed = false;
         std::string pairing_id;
         std::string session_id;
+        std::optional<EventHub::SubId> sub_id;
     };
 
     void on_listener_readable();
@@ -80,6 +83,7 @@ private:
     velox::proto::Dispatcher& dispatcher_;
     velox::daemon::store::Db& db_;
     PairingApprover& approver_;
+    EventHub& hub_;
     RuntimeDir runtime_;
     PairingRateLimiter rate_limiter_;
 
diff --git a/daemon/src/sched/engine_port.hpp b/daemon/src/sched/engine_port.hpp
index 759bf37..ba9ef8d 100644
--- a/daemon/src/sched/engine_port.hpp
+++ b/daemon/src/sched/engine_port.hpp
@@ -14,6 +14,7 @@
 
 #include <cstdint>
 #include <functional>
+#include <optional>
 #include <string>
 #include <vector>
 
@@ -40,6 +41,10 @@ public:
     // The daemon is done with this task (it went terminal). Drop the handle. Idempotent.
     virtual void release(vdm::TaskId) = 0;
 
+    // A synchronous, lock-guarded snapshot (DownloadHandle::progress()). nullopt if the
+    // id is unknown (already released, or never started).
+    virtual std::optional<vdm::task::Progress> progress(vdm::TaskId) const = 0;
+
     // ADR 0011 admission surface. Values are DAEMON's; enforcement is the engine's.
     virtual void set_task_order(const std::vector<vdm::TaskId>& order) = 0;
     virtual void set_max_active_segments(std::uint32_t n) = 0;
diff --git a/daemon/src/sched/engine_port_core.hpp b/daemon/src/sched/engine_port_core.hpp
index 9f5bb51..4fff2be 100644
--- a/daemon/src/sched/engine_port_core.hpp
+++ b/daemon/src/sched/engine_port_core.hpp
@@ -44,6 +44,11 @@ public:
         if (auto* h = find(id)) h->refresh_url(url);
     }
     void release(vdm::TaskId id) override { handles_.erase(id); }
+    std::optional<vdm::task::Progress> progress(vdm::TaskId id) const override {
+        auto it = handles_.find(id);
+        if (it == handles_.end()) return std::nullopt;
+        return it->second.progress();
+    }
 
     void set_task_order(const std::vector<vdm::TaskId>& order) override {
         engine_.segment_budget().set_task_order(order);
diff --git a/daemon/src/sched/fake_engine_port.hpp b/daemon/src/sched/fake_engine_port.hpp
index 799a7b5..127b90d 100644
--- a/daemon/src/sched/fake_engine_port.hpp
+++ b/daemon/src/sched/fake_engine_port.hpp
@@ -5,6 +5,7 @@
 
 #include <cstdint>
 #include <string>
+#include <unordered_map>
 #include <vector>
 
 #include "sched/engine_port.hpp"
@@ -42,6 +43,13 @@ public:
     void decide(vdm::TaskId, vdm::task::Decision) override {}
     void refresh_url(vdm::TaskId, const std::string&) override {}
     void release(vdm::TaskId id) override { released.push_back(id); }
+    std::optional<vdm::task::Progress> progress(vdm::TaskId id) const override {
+        auto it = fake_progress.find(id.value);
+        return it == fake_progress.end() ? std::nullopt : std::optional(it->second);
+    }
+
+    // Tests set this to control what progress(id) returns.
+    std::unordered_map<std::uint64_t, vdm::task::Progress> fake_progress;
     void set_task_order(const std::vector<vdm::TaskId>& order) override { orders.push_back(order); }
     void set_max_active_segments(std::uint32_t n) override { max_active_segments.push_back(n); }
     void set_host_segment_cap(const std::string& h, std::uint32_t c) override {
diff --git a/daemon/src/sched/scheduler.cpp b/daemon/src/sched/scheduler.cpp
index 2e923cc..8f26954 100644
--- a/daemon/src/sched/scheduler.cpp
+++ b/daemon/src/sched/scheduler.cpp
@@ -102,8 +102,9 @@ std::vector<proto::TaskState> non_terminal_states() {
 
 }  // namespace
 
-Scheduler::Scheduler(store::Db& db, EnginePort& engine, Governor governor, Deps deps)
-    : db_(db), engine_(engine), governor_(std::move(governor)), deps_(std::move(deps)) {
+Scheduler::Scheduler(store::Db& db, EnginePort& engine, Governor governor, rpc::EventHub* hub,
+                     Deps deps)
+    : db_(db), engine_(engine), governor_(std::move(governor)), hub_(hub), deps_(std::move(deps)) {
     if (!deps_.local_now) deps_.local_now = local_now_default;
     if (!deps_.post_to_loop) deps_.post_to_loop = [](std::function<void()> f) { f(); };
 }
@@ -237,7 +238,7 @@ store::DbResult<void> Scheduler::tick() {
 
         vdm::task::DownloadCallbacks cbs;
         const std::string id_copy = wire_id;
-        cbs.on_state = [this, id_copy](vdm::task::EngineState, vdm::task::EngineState to,
+        cbs.on_state = [this, id_copy](vdm::task::EngineState from, vdm::task::EngineState to,
                                        const std::optional<vdm::ErrorInfo>& err) {
             std::optional<TaskErrorFields> ef;
             if (err) {
@@ -247,20 +248,22 @@ store::DbResult<void> Scheduler::tick() {
                 if (err->http_status != 0) ef->http_status = err->http_status;
                 ef->retryable = err->retryable;
             }
+            const std::string from_name = engine_state_name(from);
             const std::string to_name = engine_state_name(to);
-            deps_.post_to_loop(
-                [this, id_copy, to_name, ef]() { on_engine_state(id_copy, to_name, ef); });
+            deps_.post_to_loop([this, id_copy, from_name, to_name, ef]() {
+                on_engine_state(id_copy, from_name, to_name, ef);
+            });
         };
 
         const vdm::TaskId engine_id = engine_.start(spec, std::move(cbs));
         map(wire_id, engine_id);
-        (void)tasks.set_state(wire_id, "probing", std::nullopt);
+        transition(wire_id, "probing", std::nullopt, std::nullopt);
     }
 
     for (const auto& wire_id : d.to_resume) {
         if (auto eid = engine_id_of(wire_id)) {
             engine_.resume(*eid);
-            (void)tasks.set_state(wire_id, "connecting", std::nullopt);
+            transition(wire_id, "connecting", std::nullopt, std::nullopt);
         }
     }
 
@@ -269,7 +272,7 @@ store::DbResult<void> Scheduler::tick() {
                                 ? pause_reason_str(d.pause_reasons.at(wire_id))
                                 : "user";
         if (auto eid = engine_id_of(wire_id)) engine_.pause(*eid);
-        (void)tasks.set_state(wire_id, "paused", std::string(reason));
+        transition(wire_id, "paused", std::string(reason), std::nullopt);
     }
 
     std::vector<vdm::TaskId> order;
@@ -281,14 +284,18 @@ store::DbResult<void> Scheduler::tick() {
     return {};
 }
 
-void Scheduler::on_engine_state(const std::string& wire_id, std::string_view engine_state,
-                                const std::optional<TaskErrorFields>& err) {
+void Scheduler::transition(const std::string& wire_id, std::string_view to_state,
+                           std::optional<std::string> pause_reason,
+                           const std::optional<TaskErrorFields>& err) {
     store::Tasks tasks(db_);
-    // pause_reason: an engine-initiated pause carries an error => 'auto' (ADR 0013 §2);
-    // otherwise set_state clears the column.
-    std::optional<std::string> reason;
-    if (engine_state == "paused" && err) reason = "auto";
-    (void)tasks.set_state(wire_id, engine_state, reason);
+    const auto before = tasks.get(wire_id);
+    const std::string previous = (before && before->has_value()) ? (**before).state : std::string();
+
+    // An engine-initiated pause carries an error => 'auto' (ADR 0013 §2), overriding
+    // whatever the caller passed (a scheduler-driven pause never carries an error here).
+    std::optional<std::string> reason = pause_reason;
+    if (to_state == "paused" && err) reason = "auto";
+    (void)tasks.set_state(wire_id, to_state, reason);
 
     if (err) {
         auto st = db_.prepare(
@@ -306,7 +313,30 @@ void Scheduler::on_engine_state(const std::string& wire_id, std::string_view eng
         }
     }
 
-    if (engine_state == "complete" || engine_state == "failed" || engine_state == "cancelled") {
+    if (!hub_) return;
+    auto after = tasks.get(wire_id);
+    if (!after || !after->has_value()) return;
+    const proto::TaskSummary summary = store::to_summary(**after);
+
+    nlohmann::json params{
+        {"taskId", wire_id},
+        {"state", std::string(to_state)},
+        {"previousState", previous.empty() ? nlohmann::json(nullptr) : nlohmann::json(previous)},
+        {"summary", summary},
+        {"error", summary.error.has_value() ? nlohmann::json(*summary.error) : nlohmann::json(nullptr)},
+    };
+    hub_->publish(proto::Event::TaskState, proto::make_notification(proto::Event::TaskState, params),
+                 wire_id);
+}
+
+void Scheduler::on_engine_state(const std::string& wire_id, std::string_view from_state,
+                                std::string_view to_state,
+                                const std::optional<TaskErrorFields>& err) {
+    (void)from_state;  // transition() reads the store's own current state as previousState,
+                       // which is authoritative regardless of engine/store timing
+    transition(wire_id, to_state, std::nullopt, err);
+
+    if (to_state == "complete" || to_state == "failed" || to_state == "cancelled") {
         if (auto eid = engine_id_of(wire_id)) {
             engine_.release(*eid);
             unmap_engine(*eid);
@@ -314,4 +344,30 @@ void Scheduler::on_engine_state(const std::string& wire_id, std::string_view eng
     }
 }
 
+std::vector<Scheduler::ProgressRow> Scheduler::progress_snapshot() {
+    std::vector<ProgressRow> out;
+    if (to_engine_.empty()) return out;
+
+    store::Tasks tasks(db_);
+    out.reserve(to_engine_.size());
+    for (const auto& [wire_id, engine_id] : to_engine_) {
+        const auto p = engine_.progress(engine_id);
+        if (!p) continue;
+
+        ProgressRow row;
+        row.task_id = wire_id;
+        row.downloaded_bytes = p->downloaded;
+        row.speed_bps = p->speed_bps;
+        row.eta_seconds = p->eta_seconds;
+        for (const auto& s : p->segments)
+            row.segments.push_back({s.index, s.completed, s.speed_bps});
+        out.push_back(std::move(row));
+
+        (void)tasks.update_progress(wire_id, static_cast<std::int64_t>(p->downloaded),
+                                    static_cast<std::int64_t>(p->effective_segments),
+                                    static_cast<std::int64_t>(p->effective_buffer_bytes));
+    }
+    return out;
+}
+
 }  // namespace velox::daemon::sched
diff --git a/daemon/src/sched/scheduler.hpp b/daemon/src/sched/scheduler.hpp
index 0e82f9b..a7430a3 100644
--- a/daemon/src/sched/scheduler.hpp
+++ b/daemon/src/sched/scheduler.hpp
@@ -4,13 +4,17 @@
 // wire-UUID <-> vdm::TaskId map and is the only thing that calls EnginePort::start /
 // pause / resume / set_task_order.
 //
-// Threading: tick(), reload_config(), reconcile_after_restart() and the on_engine_*
-// callbacks all run on ONE thread (the RPC loop). Engine callbacks arrive on engine
-// threads, so the real wiring passes a `post_to_loop` that marshals them here; the
-// default runs them inline (tests, single-threaded).
+// Threading: tick(), reload_config(), reconcile_after_restart(), progress_snapshot() and
+// the on_engine_* callbacks all run on ONE thread (the RPC loop). Engine callbacks arrive
+// on engine threads, so the real wiring passes a `post_to_loop` that marshals them here;
+// the default runs them inline (tests, single-threaded).
 //
-// Not yet wired into veloxd — that plus the real EnginePort land when velox::core's
-// stage-8 bodies reach main (daemon/docs/deferrals.md D4).
+// event.task.state (D5): on_engine_state publishes it when a hub is supplied — the same
+// callback that keeps the store row current also keeps subscribed clients current.
+// event.task.progress is NOT published here: it must be batched into one array message at
+// <=4 Hz (event.task.progress.schema.json x-maxRateHz), so the caller collects
+// progress_snapshot() on its own 250 ms timer and does one hub_.publish() with the whole
+// array, never one per task.
 
 #include <cstdint>
 #include <ctime>
@@ -18,7 +22,9 @@
 #include <optional>
 #include <string>
 #include <unordered_map>
+#include <vector>
 
+#include "rpc/event_hub.hpp"
 #include "sched/engine_port.hpp"
 #include "sched/governor.hpp"
 #include "store/sqlite.hpp"
@@ -46,7 +52,10 @@ public:
         std::function<void(std::function<void()>)> post_to_loop;
     };
 
-    Scheduler(store::Db& db, EnginePort& engine, Governor governor, Deps deps = {});
+    // `hub` is optional so unit tests can build a Scheduler with no event fan-out at all;
+    // production wiring always supplies one.
+    Scheduler(store::Db& db, EnginePort& engine, Governor governor, rpc::EventHub* hub = nullptr,
+             Deps deps = {});
 
     // ADR 0013 §5: on daemon start, every task whose persisted state is a CORE-owned one
     // (probing..verifying) is rewritten to `queued`; `paused` keeps its pauseReason. The
@@ -62,11 +71,29 @@ public:
     // (start / resume / pause via the engine, update task state rows, push set_task_order).
     store::DbResult<void> tick();
 
-    // Engine lifecycle callback -> store projection, so the next tick sees ground truth.
-    // Keyed by wire UUID (known when the callback is built, before start() returns the
-    // TaskId). Also the hook the event.task.state fan-out will use — D5.
-    void on_engine_state(const std::string& wire_id, std::string_view engine_state,
-                         const std::optional<TaskErrorFields>& err);
+    // Engine lifecycle callback -> store projection, so the next tick sees ground truth,
+    // and (when a hub was supplied) the event.task.state publish. Keyed by wire UUID
+    // (known when the callback is built, before start() returns the TaskId).
+    void on_engine_state(const std::string& wire_id, std::string_view from_state,
+                         std::string_view to_state, const std::optional<TaskErrorFields>& err);
+
+    // One row per task the engine is currently tracking, for the caller's
+    // event.task.progress batch. Also writes downloaded_bytes / eff_segments /
+    // eff_buffer_bytes back to the store so download.list / download.get stay current
+    // between state transitions.
+    struct ProgressRow {
+        std::string task_id;
+        std::uint64_t downloaded_bytes;
+        std::uint64_t speed_bps;
+        std::optional<std::uint32_t> eta_seconds;
+        struct Segment {
+            std::uint32_t index;
+            std::uint64_t downloaded_bytes;
+            std::uint64_t speed_bps;
+        };
+        std::vector<Segment> segments;
+    };
+    std::vector<ProgressRow> progress_snapshot();
 
     // Diagnostics / tests.
     std::optional<std::string> wire_id_of(vdm::TaskId id) const;
@@ -76,9 +103,17 @@ private:
     void map(const std::string& wire_id, vdm::TaskId engine_id);
     void unmap_engine(vdm::TaskId engine_id);
 
+    // The one place a task's state row changes and (if a hub is set) event.task.state
+    // publishes. previousState is read from the store's own current row, not passed in —
+    // authoritative regardless of engine/scheduler timing.
+    void transition(const std::string& wire_id, std::string_view to_state,
+                    std::optional<std::string> pause_reason,
+                    const std::optional<TaskErrorFields>& err);
+
     store::Db& db_;
     EnginePort& engine_;
     Governor governor_;
+    rpc::EventHub* hub_;
     Deps deps_;
 
     std::unordered_map<std::string, vdm::TaskId> to_engine_;
diff --git a/daemon/src/store/categories.cpp b/daemon/src/store/categories.cpp
new file mode 100644
index 0000000..8532e8d
--- /dev/null
+++ b/daemon/src/store/categories.cpp
@@ -0,0 +1,36 @@
+#include "store/categories.hpp"
+
+#include <nlohmann/json.hpp>
+
+namespace velox::daemon::store {
+
+namespace proto = velox::proto;
+
+DbResult<std::vector<proto::Category>> Categories::list() {
+    auto st = db_.prepare(
+        "SELECT category_id, name, save_dir, extensions, builtin FROM categories "
+        "ORDER BY builtin DESC, name");
+    if (!st) return std::unexpected(st.error());
+
+    std::vector<proto::Category> out;
+    for (;;) {
+        auto row = st->step();
+        if (!row) return std::unexpected(row.error());
+        if (!*row) break;
+
+        proto::Category c;
+        c.categoryId = st->column_text(0);
+        c.name = st->column_text(1);
+        c.saveDir = st->column_text(2);
+        auto j = nlohmann::json::parse(st->column_text(3), nullptr, false);
+        if (j.is_array()) {
+            for (const auto& e : j)
+                if (e.is_string()) c.extensions.push_back(e.get<std::string>());
+        }
+        c.builtin = st->column_int(4) != 0;
+        out.push_back(std::move(c));
+    }
+    return out;
+}
+
+}  // namespace velox::daemon::store
diff --git a/daemon/src/store/categories.hpp b/daemon/src/store/categories.hpp
new file mode 100644
index 0000000..b4439e8
--- /dev/null
+++ b/daemon/src/store/categories.hpp
@@ -0,0 +1,24 @@
+#pragma once
+
+// Read access to the `categories` table, projected onto proto::Category. Owned here
+// rather than duplicated per handler since category.list and download.add (rule
+// matching, later) both need it.
+
+#include <vector>
+
+#include "store/sqlite.hpp"
+#include "velox_proto.hpp"
+
+namespace velox::daemon::store {
+
+class Categories {
+public:
+    explicit Categories(Db& db) : db_(db) {}
+
+    DbResult<std::vector<velox::proto::Category>> list();
+
+private:
+    Db& db_;
+};
+
+}  // namespace velox::daemon::store
diff --git a/daemon/src/store/queues.cpp b/daemon/src/store/queues.cpp
new file mode 100644
index 0000000..97c63b8
--- /dev/null
+++ b/daemon/src/store/queues.cpp
@@ -0,0 +1,48 @@
+#include "store/queues.hpp"
+
+#include <nlohmann/json.hpp>
+
+namespace velox::daemon::store {
+
+namespace proto = velox::proto;
+
+DbResult<std::vector<proto::Queue>> Queues::list() {
+    auto st = db_.prepare(
+        "SELECT queue_id, name, state, max_concurrent, schedule FROM queues ORDER BY name");
+    if (!st) return std::unexpected(st.error());
+
+    std::vector<proto::Queue> out;
+    for (;;) {
+        auto row = st->step();
+        if (!row) return std::unexpected(row.error());
+        if (!*row) break;
+
+        proto::Queue q;
+        q.queueId = st->column_text(0);
+        q.name = st->column_text(1);
+        if (auto s = proto::parse_QueueState(st->column_text(2))) q.state = *s;
+        q.maxConcurrent = st->column_int(3);
+        if (!st->column_is_null(4)) {
+            auto j = nlohmann::json::parse(st->column_text(4), nullptr, false);
+            if (auto sched = proto::parse<proto::Schedule>(j, "schedule")) q.schedule = *sched;
+        }
+
+        auto ts = db_.prepare(
+            "SELECT task_id FROM tasks WHERE queue_id = ?1 ORDER BY queue_position");
+        if (!ts) return std::unexpected(ts.error());
+        if (auto b = ts->bind(1, std::string_view(q.queueId)); !b) return std::unexpected(b.error());
+        std::vector<std::string> ids;
+        for (;;) {
+            auto r = ts->step();
+            if (!r) return std::unexpected(r.error());
+            if (!*r) break;
+            ids.push_back(ts->column_text(0));
+        }
+        q.taskIds = std::move(ids);
+
+        out.push_back(std::move(q));
+    }
+    return out;
+}
+
+}  // namespace velox::daemon::store
diff --git a/daemon/src/store/queues.hpp b/daemon/src/store/queues.hpp
new file mode 100644
index 0000000..90460bb
--- /dev/null
+++ b/daemon/src/store/queues.hpp
@@ -0,0 +1,25 @@
+#pragma once
+
+// Read access to the `queues` table, projected onto proto::Queue. taskIds is derived from
+// `tasks` (queue_id = this queue, ordered by queue_position), not stored on the queue row
+// — membership changes through download.update / queue.reorder, per Queue's own schema
+// note that a queue.upsert payload's taskIds is ignored.
+
+#include <vector>
+
+#include "store/sqlite.hpp"
+#include "velox_proto.hpp"
+
+namespace velox::daemon::store {
+
+class Queues {
+public:
+    explicit Queues(Db& db) : db_(db) {}
+
+    DbResult<std::vector<velox::proto::Queue>> list();
+
+private:
+    Db& db_;
+};
+
+}  // namespace velox::daemon::store
diff --git a/daemon/src/store/tasks.cpp b/daemon/src/store/tasks.cpp
index cd809b3..7ea946b 100644
--- a/daemon/src/store/tasks.cpp
+++ b/daemon/src/store/tasks.cpp
@@ -264,6 +264,20 @@ DbResult<bool> Tasks::remove(std::string_view task_id) {
     return sqlite3_changes(db_.raw()) > 0;
 }
 
+DbResult<bool> Tasks::update_progress(std::string_view task_id, std::int64_t downloaded_bytes,
+                                      std::int64_t eff_segments, std::int64_t eff_buffer_bytes) {
+    auto st = db_.prepare(
+        "UPDATE tasks SET downloaded_bytes = ?2, eff_segments = ?3, eff_buffer_bytes = ?4 "
+        "WHERE task_id = ?1");
+    if (!st) return std::unexpected(st.error());
+    if (auto r = st->bind(1, task_id); !r) return std::unexpected(r.error());
+    if (auto r = st->bind(2, downloaded_bytes); !r) return std::unexpected(r.error());
+    if (auto r = st->bind(3, eff_segments); !r) return std::unexpected(r.error());
+    if (auto r = st->bind(4, eff_buffer_bytes); !r) return std::unexpected(r.error());
+    if (auto r = st->step(); !r) return std::unexpected(r.error());
+    return sqlite3_changes(db_.raw()) > 0;
+}
+
 DbResult<std::int64_t> Tasks::count() {
     auto st = db_.prepare("SELECT count(*) FROM tasks");
     if (!st) return std::unexpected(st.error());
diff --git a/daemon/src/store/tasks.hpp b/daemon/src/store/tasks.hpp
index 4cc384c..cecf8b5 100644
--- a/daemon/src/store/tasks.hpp
+++ b/daemon/src/store/tasks.hpp
@@ -78,6 +78,11 @@ public:
     DbResult<bool> remove(std::string_view task_id);
     DbResult<std::int64_t> count();
 
+    // Byte-counter update from an engine progress tick — cheaper than a full row rewrite,
+    // and keeps download.list / download.get current between state transitions.
+    DbResult<bool> update_progress(std::string_view task_id, std::int64_t downloaded_bytes,
+                                   std::int64_t eff_segments, std::int64_t eff_buffer_bytes);
+
 private:
     Db& db_;
 };
diff --git a/daemon/tests/CMakeLists.txt b/daemon/tests/CMakeLists.txt
index fac594e..9c84ca4 100644
--- a/daemon/tests/CMakeLists.txt
+++ b/daemon/tests/CMakeLists.txt
@@ -20,4 +20,6 @@ veloxd_test(sched_window     LIBS veloxd_sched)
 veloxd_test(sched_governor   LIBS veloxd_sched)
 veloxd_test(safepath         LIBS veloxd_fs)
 veloxd_test(store_tasks      LIBS veloxd_store)
-veloxd_test(sched_scheduler  LIBS veloxd_sched)
+veloxd_test(sched_scheduler  LIBS veloxd_sched veloxd_rpc)
+veloxd_test(event_hub               LIBS veloxd_rpc)
+veloxd_test(store_categories_queues LIBS veloxd_store)
diff --git a/daemon/tests/event_hub_test.cpp b/daemon/tests/event_hub_test.cpp
new file mode 100644
index 0000000..adae1a0
--- /dev/null
+++ b/daemon/tests/event_hub_test.cpp
@@ -0,0 +1,68 @@
+#include "rpc/event_hub.hpp"
+
+#include <nlohmann/json.hpp>
+
+#include "check.hpp"
+
+using velox::daemon::rpc::EventHub;
+namespace proto = velox::proto;
+
+void run() {
+    EventHub hub;
+
+    // --- no subscribers: publish is a no-op, not an error --------------------------
+    hub.publish(proto::Event::TaskAdded, nlohmann::json{{"x", 1}});
+
+    // --- a subscription with no filter set receives nothing -------------------
+    std::vector<nlohmann::json> a;
+    const auto sub_a = hub.subscribe([&](const nlohmann::json& n) { a.push_back(n); });
+    hub.publish(proto::Event::TaskAdded, nlohmann::json{{"n", 1}});
+    CHECK_EQ(a.size(), 0u);
+
+    // --- filtering by event kind ------------------------------------------------
+    hub.set_filter(sub_a, {proto::Event::TaskAdded}, std::nullopt);
+    hub.publish(proto::Event::TaskAdded, nlohmann::json{{"n", 2}});
+    hub.publish(proto::Event::TaskRemoved, nlohmann::json{{"n", 3}});  // not subscribed
+    CHECK_EQ(a.size(), 1u);
+    CHECK_EQ(a[0]["n"].get<int>(), 2);
+
+    // --- two subscribers, independent filters -----------------------------------
+    std::vector<nlohmann::json> b;
+    const auto sub_b = hub.subscribe([&](const nlohmann::json& n) { b.push_back(n); });
+    hub.set_filter(sub_b, {proto::Event::TaskRemoved}, std::nullopt);
+    hub.publish(proto::Event::TaskAdded, nlohmann::json{{"n", 4}});
+    hub.publish(proto::Event::TaskRemoved, nlohmann::json{{"n", 5}});
+    CHECK_EQ(a.size(), 2u);   // got the TaskAdded
+    CHECK_EQ(b.size(), 1u);   // got the TaskRemoved
+    CHECK_EQ(b[0]["n"].get<int>(), 5);
+
+    // --- per-task filter: only the named ids reach the subscriber ---------------
+    std::vector<nlohmann::json> c;
+    const auto sub_c = hub.subscribe([&](const nlohmann::json& n) { c.push_back(n); });
+    hub.set_filter(sub_c, {proto::Event::TaskState},
+                   std::vector<std::string>{"t1", "t2"});
+    hub.publish(proto::Event::TaskState, nlohmann::json{{"n", 6}}, "t1");
+    hub.publish(proto::Event::TaskState, nlohmann::json{{"n", 7}}, "t9");  // filtered out
+    hub.publish(proto::Event::TaskState, nlohmann::json{{"n", 8}}, "t2");
+    CHECK_EQ(c.size(), 2u);
+    CHECK_EQ(c[0]["n"].get<int>(), 6);
+    CHECK_EQ(c[1]["n"].get<int>(), 8);
+
+    // --- a non-task-scoped publish (no task_id) reaches a task-filtered sub too --
+    // (matches "session.subscribe narrows task events" — a global event isn't one)
+    hub.publish(proto::Event::TaskState, nlohmann::json{{"n", 9}});
+    CHECK_EQ(c.size(), 3u);
+
+    // --- set_filter replaces, it does not add -----------------------------------
+    hub.set_filter(sub_a, {proto::Event::TaskRemoved}, std::nullopt);
+    hub.publish(proto::Event::TaskAdded, nlohmann::json{{"n", 10}});
+    CHECK_EQ(a.size(), 2u);  // still 2 -- TaskAdded no longer reaches sub_a
+
+    // --- unsubscribe stops delivery ------------------------------------------
+    hub.unsubscribe(sub_b);
+    hub.publish(proto::Event::TaskRemoved, nlohmann::json{{"n", 11}});
+    CHECK_EQ(b.size(), 1u);  // unchanged
+    CHECK_EQ(a.size(), 3u);  // sub_a still gets TaskRemoved
+}
+
+TEST_MAIN()
diff --git a/daemon/tests/sched_scheduler_test.cpp b/daemon/tests/sched_scheduler_test.cpp
index 3d614b7..3adf08b 100644
--- a/daemon/tests/sched_scheduler_test.cpp
+++ b/daemon/tests/sched_scheduler_test.cpp
@@ -3,7 +3,10 @@
 
 #include <string>
 
+#include <nlohmann/json.hpp>
+
 #include "check.hpp"
+#include "rpc/event_hub.hpp"
 #include "sched/fake_engine_port.hpp"
 #include "sched/governor.hpp"
 #include "sched/scheduler.hpp"
@@ -91,10 +94,10 @@ void run() {
         CHECK_EQ(engine.starts.size(), 1u);  // b0 only
         CHECK_EQ(task_state(*db, "b0"), std::string("probing"));
 
-        sched.on_engine_state("b0", "downloading", std::nullopt);
+        sched.on_engine_state("b0", "probing", "downloading", std::nullopt);
         CHECK_EQ(task_state(*db, "b0"), std::string("downloading"));
 
-        sched.on_engine_state("b0", "complete", std::nullopt);
+        sched.on_engine_state("b0", "downloading", "complete", std::nullopt);
         CHECK_EQ(task_state(*db, "b0"), std::string("complete"));
 
         CHECK(sched.tick().has_value());  // b0 terminal -> b1 admitted
@@ -113,7 +116,7 @@ void run() {
         CHECK(tasks.insert(task("q0", "queued", "2026-09-10T10:00:00Z", "main", 0)).has_value());
         CHECK(sched.tick().has_value());
         CHECK_EQ(engine.starts.size(), 1u);
-        sched.on_engine_state("q0", "downloading", std::nullopt);
+        sched.on_engine_state("q0", "connecting", "downloading", std::nullopt);
 
         CHECK(db->exec("UPDATE queues SET state='stopped' WHERE queue_id='main'").has_value());
         CHECK(sched.tick().has_value());
@@ -143,7 +146,7 @@ void run() {
         ef.code = "auth_required";
         ef.message = "401";
         ef.http_status = 401;
-        sched.on_engine_state("auth", "paused", ef);
+        sched.on_engine_state("auth", "connecting", "paused", ef);
 
         store::Tasks t(*db);
         auto row = t.get("auth").value().value();
@@ -186,6 +189,76 @@ void run() {
         CHECK_EQ(engine.max_active_segments.back(), 12u);
         CHECK_EQ(sched.reload_config().has_value() ? 0 : 1, 0);
     }
+
+    // --- event.task.state: published on admission, on a paused transition, and on an
+    // engine-reported transition; previousState reflects the store's prior row ----------
+    {
+        for (const char* id : {"r0", "r1", "r2"}) tasks.remove(id);  // leftover from above
+        FakeEnginePort engine;
+        rpc::EventHub hub;
+        Scheduler sched(*db, engine, Governor(GovernorConfig{.max_concurrent_downloads = 10,
+                                                              .max_active_segments = 32}),
+                        &hub);
+        std::vector<nlohmann::json> received;
+        const auto sub = hub.subscribe([&](const nlohmann::json& n) { received.push_back(n); });
+        hub.set_filter(sub, {velox::proto::Event::TaskState}, std::nullopt);
+
+        CHECK(tasks.insert(task("ev0", "queued", "2026-09-10T10:00:00Z")).has_value());
+        CHECK(sched.tick().has_value());  // admits ev0: queued -> probing
+        CHECK_EQ(received.size(), 1u);
+        CHECK_EQ(received[0]["params"]["taskId"].get<std::string>(), std::string("ev0"));
+        CHECK_EQ(received[0]["params"]["state"].get<std::string>(), std::string("probing"));
+        CHECK_EQ(received[0]["params"]["previousState"].get<std::string>(), std::string("queued"));
+        CHECK(received[0]["params"]["error"].is_null());
+        CHECK_EQ(received[0]["params"]["summary"]["taskId"].get<std::string>(), std::string("ev0"));
+
+        sched.on_engine_state("ev0", "probing", "downloading", std::nullopt);
+        CHECK_EQ(received.size(), 2u);
+        CHECK_EQ(received[1]["params"]["previousState"].get<std::string>(), std::string("probing"));
+        CHECK_EQ(received[1]["params"]["state"].get<std::string>(), std::string("downloading"));
+
+        sched::TaskErrorFields ef;
+        ef.code = "connection_reset";
+        ef.message = "reset";
+        sched.on_engine_state("ev0", "downloading", "paused", ef);
+        CHECK_EQ(received.size(), 3u);
+        CHECK(!received[2]["params"]["error"].is_null());
+        CHECK_EQ(received[2]["params"]["error"]["code"].get<std::string>(),
+                 std::string("connection_reset"));
+
+        tasks.remove("ev0");
+    }
+
+    // --- progress_snapshot: only started tasks, plus a store side-effect --------------
+    {
+        FakeEnginePort engine;
+        Scheduler sched(*db, engine,
+                        Governor(GovernorConfig{.max_concurrent_downloads = 10,
+                                                .max_active_segments = 32}));
+        CHECK(tasks.insert(task("pr0", "queued", "2026-09-10T10:00:00Z")).has_value());
+        CHECK(sched.tick().has_value());
+        CHECK_EQ(engine.starts.size(), 1u);
+
+        vdm::task::Progress p;
+        p.downloaded = 12345;
+        p.speed_bps = 999;
+        p.effective_segments = 4;
+        p.effective_buffer_bytes = 65536;
+        engine.fake_progress[engine.starts[0].id.value] = p;
+
+        const auto snap = sched.progress_snapshot();
+        CHECK_EQ(snap.size(), 1u);
+        CHECK_EQ(snap[0].task_id, std::string("pr0"));
+        CHECK_EQ(snap[0].downloaded_bytes, 12345u);
+        CHECK_EQ(snap[0].speed_bps, 999u);
+
+        auto row = tasks.get("pr0").value().value();
+        CHECK_EQ(row.downloaded_bytes, 12345);
+        CHECK_EQ(row.eff_segments, 4);
+        CHECK((row.eff_buffer_bytes.has_value() && *row.eff_buffer_bytes == 65536));
+
+        tasks.remove("pr0");
+    }
 }
 
 TEST_MAIN()
diff --git a/daemon/tests/store_categories_queues_test.cpp b/daemon/tests/store_categories_queues_test.cpp
new file mode 100644
index 0000000..bd4f696
--- /dev/null
+++ b/daemon/tests/store_categories_queues_test.cpp
@@ -0,0 +1,78 @@
+// store/categories + store/queues: the two D3 handlers GUI's category panel and queue
+// view need against a real daemon.
+
+#include <string>
+
+#include "check.hpp"
+#include "store/categories.hpp"
+#include "store/migrations.hpp"
+#include "store/queues.hpp"
+#include "store/sqlite.hpp"
+#include "store/tasks.hpp"
+
+using namespace velox::daemon::store;
+
+void run() {
+    auto db = Db::open(":memory:");
+    CHECK(db.has_value());
+    if (!db) return;
+    CHECK(migrate_to_head(*db).has_value());
+
+    // --- categories: the six seeded built-ins, builtin first --------------------
+    {
+        Categories categories(*db);
+        auto items = categories.list();
+        CHECK(items.has_value());
+        if (items) {
+            CHECK_EQ(items->size(), 6u);
+            for (const auto& c : *items) CHECK(c.builtin);
+            const auto& programs =
+                *std::find_if(items->begin(), items->end(),
+                              [](const auto& c) { return c.categoryId == "programs"; });
+            CHECK(!programs.extensions.empty());
+            CHECK(std::find(programs.extensions.begin(), programs.extensions.end(), "iso") !=
+                  programs.extensions.end());
+        }
+    }
+
+    // --- queues: the seeded "main" queue, empty task list ------------------------
+    {
+        Queues queues(*db);
+        auto items = queues.list();
+        CHECK(items.has_value());
+        if (items) {
+            CHECK_EQ(items->size(), 1u);
+            CHECK_EQ(items->front().queueId, std::string("main"));
+            CHECK(items->front().taskIds.has_value());
+            CHECK(items->front().taskIds->empty());
+        }
+    }
+
+    // --- queue.taskIds reflects membership, in queue_position order ------------
+    {
+        Tasks tasks(*db);
+        for (int i = 0; i < 3; ++i) {
+            TaskRow r;
+            r.task_id = "t" + std::to_string(i);
+            r.url = "https://example.com/" + r.task_id;
+            r.save_dir = "/tmp";
+            r.filename = r.task_id;
+            r.created_at = "2026-09-11T00:00:00Z";
+            r.queue_id = "main";
+            r.queue_position = 2 - i;  // reverse insertion order
+            CHECK(tasks.insert(r).has_value());
+        }
+        Queues queues(*db);
+        auto items = queues.list();
+        CHECK(items.has_value());
+        if (items && !items->empty()) {
+            const auto& ids = *items->front().taskIds;
+            CHECK_EQ(ids.size(), 3u);
+            CHECK_EQ(ids[0], std::string("t2"));  // queue_position 0
+            CHECK_EQ(ids[1], std::string("t1"));
+            CHECK_EQ(ids[2], std::string("t0"));
+        }
+    }
+}
+
+TEST_MAIN()
diff --git a/daemon/tests/uds_roundtrip_test.cpp b/daemon/tests/uds_roundtrip_test.cpp
index b707591..9cc79d2 100644
--- a/daemon/tests/uds_roundtrip_test.cpp
+++ b/daemon/tests/uds_roundtrip_test.cpp
@@ -14,6 +14,7 @@
 
 #include "check.hpp"
 #include "rpc/dispatcher.hpp"
+#include "rpc/event_hub.hpp"
 #include "rpc/event_loop.hpp"
 #include "rpc/ndjson.hpp"
 #include "rpc/uds_server.hpp"
@@ -71,8 +72,9 @@ void run() {
     CHECK(velox::daemon::store::migrate_to_head(*db).has_value());
 
     rpc::EventLoop loop;
-    rpc::VeloxDispatcher dispatcher(*db);
-    rpc::UdsServer server(loop, dispatcher, sock);
+    rpc::EventHub hub;
+    rpc::VeloxDispatcher dispatcher(*db, hub);
+    rpc::UdsServer server(loop, dispatcher, hub, sock);
     const auto ec = server.start();
     CHECK(!ec);
     if (ec) return;
diff --git a/daemon/tests/ws_server_test.cpp b/daemon/tests/ws_server_test.cpp
index 22ed527..f3b404b 100644
--- a/daemon/tests/ws_server_test.cpp
+++ b/daemon/tests/ws_server_test.cpp
@@ -16,6 +16,7 @@
 
 #include "check.hpp"
 #include "rpc/dispatcher.hpp"
+#include "rpc/event_hub.hpp"
 #include "rpc/event_loop.hpp"
 #include "rpc/pairing.hpp"
 #include "rpc/runtime_dir.hpp"
@@ -123,9 +124,10 @@ void run() {
     rpc::RuntimeDir rt{dir ? dir : "/tmp"};
 
     rpc::EventLoop loop;
-    rpc::VeloxDispatcher dispatcher(*db);
+    rpc::EventHub hub;
+    rpc::VeloxDispatcher dispatcher(*db, hub);
     rpc::EnvAutoApprover approver;
-    rpc::WsServer server(loop, dispatcher, *db, approver, rt);
+    rpc::WsServer server(loop, dispatcher, *db, approver, hub, rt);
     const auto ec = server.start();
     CHECK(!ec);
     if (ec) return;