daemon: D4b — download.pause/resume/start/cancel and queue.start/stop drive the scheduler

download.pause/resume/start/cancel and queue.start/stop were stubs; now they call into
the scheduler and take effect immediately, not on the next 1s tick — pausing, resuming
or cancelling a live transfer can't wait, and per ADR 0013 §3 the governor never
touches a user-owned pause on its own.

New rpc::TaskActionPort interface (owned by rpc/, implemented by sched::Scheduler) is
what dispatcher.hpp depends on instead of sched/scheduler.hpp directly. Needed because
veloxd_sched already links veloxd_rpc (for EventHub); dispatcher.hpp pulling in
sched/scheduler.hpp directly would make it a real circular library dependency, breaking
anything that links veloxd_rpc alone (cli's tests, as it turned out — hit and fixed
during this change).

Scheduler::user_pause/user_resume/user_start/user_cancel + pause_queue follow tick()'s
existing to_pause pattern: call the engine (async, no synchronous effect) and transition
the store eagerly so download.get/list are correct the instant the RPC call returns.

Fixed a real bug surfaced while building this: transition() always overwrote
pause_reason to NULL when the engine's own delayed pause-ack callback (on_state to
paused, no error) arrived after whoever actually initiated the pause had already
written the real reason — now it preserves the stored reason when the callback supplies
none, instead of clobbering it. Covered by a regression check in sched_scheduler_test.

store/queues gets get() and set_state() (was list()-only) for queue.start/stop.

Verified against real veloxd + tools/testserver, not just unit tests: pausing a live
single-segment throttled transfer freezes downloadedBytes, resume continues it from
that point, cancel stops it; a bad taskId comes back in BulkTaskResult.failed with
-32010, not a crash; queue.stop(pauseRunning:true) pauses the queue's running task
immediately and queue.start resumes admission.

Known gap: download.start's contract "a task in 'queued' jumps its queue" (priority
bump) is not implemented — admission is still plain FIFO by created_at. Noted in
deferrals.md.

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

daemon/src/main.cpp

@@ -103,7 +103,7 @@ int main() {
     (void)scheduler.reload_config();
     (void)scheduler.tick();  // admit anything already queued in the DB
 
-    velox::daemon::rpc::VeloxDispatcher dispatcher(*db, hub);
+    velox::daemon::rpc::VeloxDispatcher dispatcher(*db, hub, &scheduler);
     dispatcher.set_on_mutation([&loop, &scheduler] {
         loop.post([&scheduler] { (void)scheduler.tick(); });
     });

daemon/src/rpc/dispatcher.cpp

@@ -28,6 +28,29 @@ proto::HandlerResult<T> not_implemented(const char* method) {
         std::string("not implemented in this build: ") + method});
 }
 
+// Shared by download.pause/resume/start/cancel: apply `action` to every id in `task_ids`
+// and fold the per-task velox::daemon::rpc::TaskActionPort::Result into a
+// BulkTaskResult (ADR: "a bulk call never fails as a whole because one id was bad").
+proto::BulkTaskResult bulk_apply(
+    const std::vector<std::string>& task_ids,
+    const std::function<velox::daemon::rpc::TaskActionPort::Result(const std::string&)>&
+        action) {
+    proto::BulkTaskResult out;
+    for (const auto& id : task_ids) {
+        const auto r = action(id);
+        if (!r.found) {
+            out.failed.push_back({id, proto::ErrorCode::TaskNotFound, "no such task"});
+            continue;
+        }
+        proto::BulkTaskResultUpdatedItem item;
+        item.taskId = id;
+        item.changed = r.changed;
+        if (auto st = proto::parse_TaskState(r.state)) item.state = *st;
+        out.updated.push_back(std::move(item));
+    }
+    return out;
+}
+
 // A v4 UUID for a new task id.
 std::string new_task_id() {
     std::random_device rd;
@@ -237,8 +260,13 @@ VeloxDispatcher::on_download_addBatch(const proto::DownloadAddBatchParams&) {
     return not_implemented<proto::DownloadAddBatchResult>("download.addBatch");
 }
 proto::HandlerResult<proto::BulkTaskResult>
-VeloxDispatcher::on_download_cancel(const proto::DownloadCancelParams&) {
-    return not_implemented<proto::BulkTaskResult>("download.cancel");
+VeloxDispatcher::on_download_cancel(const proto::DownloadCancelParams& params) {
+    if (!actions_) return not_implemented<proto::BulkTaskResult>("download.cancel");
+    auto result = bulk_apply(params.taskIds, [this](const std::string& id) {
+        return actions_->user_cancel(id, /*discard_partial=*/false);
+    });
+    if (on_mutation_) on_mutation_();
+    return result;
 }
 proto::HandlerResult<proto::TaskDetail>
 VeloxDispatcher::on_download_get(const proto::DownloadGetParams& params) {
@@ -273,8 +301,12 @@ VeloxDispatcher::on_download_get(const proto::DownloadGetParams& params) {
     return d;
 }
 proto::HandlerResult<proto::BulkTaskResult>
-VeloxDispatcher::on_download_pause(const proto::DownloadPauseParams&) {
-    return not_implemented<proto::BulkTaskResult>("download.pause");
+VeloxDispatcher::on_download_pause(const proto::DownloadPauseParams& params) {
+    if (!actions_) return not_implemented<proto::BulkTaskResult>("download.pause");
+    auto result = bulk_apply(params.taskIds,
+                             [this](const std::string& id) { return actions_->user_pause(id); });
+    if (on_mutation_) on_mutation_();
+    return result;
 }
 proto::HandlerResult<proto::DownloadProbeResult>
 VeloxDispatcher::on_download_probe(const proto::DownloadProbeParams&) {
@@ -293,12 +325,20 @@ VeloxDispatcher::on_download_remove(const proto::DownloadRemoveParams&) {
     return not_implemented<proto::DownloadRemoveResult>("download.remove");
 }
 proto::HandlerResult<proto::BulkTaskResult>
-VeloxDispatcher::on_download_resume(const proto::DownloadResumeParams&) {
-    return not_implemented<proto::BulkTaskResult>("download.resume");
+VeloxDispatcher::on_download_resume(const proto::DownloadResumeParams& params) {
+    if (!actions_) return not_implemented<proto::BulkTaskResult>("download.resume");
+    auto result = bulk_apply(
+        params.taskIds, [this](const std::string& id) { return actions_->user_resume(id); });
+    if (on_mutation_) on_mutation_();
+    return result;
 }
 proto::HandlerResult<proto::BulkTaskResult>
-VeloxDispatcher::on_download_start(const proto::DownloadStartParams&) {
-    return not_implemented<proto::BulkTaskResult>("download.start");
+VeloxDispatcher::on_download_start(const proto::DownloadStartParams& params) {
+    if (!actions_) return not_implemented<proto::BulkTaskResult>("download.start");
+    auto result = bulk_apply(
+        params.taskIds, [this](const std::string& id) { return actions_->user_start(id); });
+    if (on_mutation_) on_mutation_();
+    return result;
 }
 proto::HandlerResult<proto::TaskSummary>
 VeloxDispatcher::on_download_update(const proto::DownloadUpdateParams&) {
@@ -346,12 +386,54 @@ VeloxDispatcher::on_queue_reorder(const proto::QueueReorderParams&) {
     return not_implemented<proto::QueueReorderResult>("queue.reorder");
 }
 proto::HandlerResult<proto::QueueStartResult>
-VeloxDispatcher::on_queue_start(const proto::QueueStartParams&) {
-    return not_implemented<proto::QueueStartResult>("queue.start");
+VeloxDispatcher::on_queue_start(const proto::QueueStartParams& params) {
+    store::Queues queues(db_);
+    auto exists = queues.get(params.queueId);
+    if (!exists)
+        return std::unexpected(proto::HandlerError{proto::ErrorCode::InternalError,
+                                                   "queue.start: " + exists.error().message});
+    if (!exists->has_value())
+        return std::unexpected(proto::HandlerError{
+            proto::ErrorCode::InvalidParams, "no such queue",
+            nlohmann::json{{"queueId", params.queueId}}});
+
+    (void)queues.set_state(params.queueId, "running");
+    // Nudged through the same deferred tick() as download.add (on_mutation_ posts onto the
+    // loop) — which tasks actually get admitted happens asynchronously (tick()'s to_start
+    // even probes before it starts one), so startedTaskIds is not knowable synchronously
+    // here; the GUI learns the real outcome from event.task.state as each one lands.
+    if (on_mutation_) on_mutation_();
+
+    auto after = queues.get(params.queueId);
+    proto::QueueStartResult r;
+    if (after && after->has_value()) r.queue = **after;
+    return r;
 }
 proto::HandlerResult<proto::QueueStopResult>
-VeloxDispatcher::on_queue_stop(const proto::QueueStopParams&) {
-    return not_implemented<proto::QueueStopResult>("queue.stop");
+VeloxDispatcher::on_queue_stop(const proto::QueueStopParams& params) {
+    store::Queues queues(db_);
+    auto exists = queues.get(params.queueId);
+    if (!exists)
+        return std::unexpected(proto::HandlerError{proto::ErrorCode::InternalError,
+                                                   "queue.stop: " + exists.error().message});
+    if (!exists->has_value())
+        return std::unexpected(proto::HandlerError{
+            proto::ErrorCode::InvalidParams, "no such queue",
+            nlohmann::json{{"queueId", params.queueId}}});
+
+    (void)queues.set_state(params.queueId, "stopped");
+
+    proto::QueueStopResult r;
+    // "the difference between 'stop the queue' and 'stop everything', which IDM
+    // conflates" (the schema's own words) — absent means the soft stop: only halt new
+    // admissions, let what's running finish.
+    if (params.pauseRunning.value_or(false) && actions_)
+        r.pausedTaskIds = actions_->pause_queue(params.queueId);
+    if (on_mutation_) on_mutation_();
+
+    auto after = queues.get(params.queueId);
+    if (after && after->has_value()) r.queue = **after;
+    return r;
 }
 proto::HandlerResult<proto::QueueUpsertResult>
 VeloxDispatcher::on_queue_upsert(const proto::QueueUpsertParams&) {

daemon/src/rpc/dispatcher.hpp

@@ -15,6 +15,7 @@
 #include <functional>
 
 #include "rpc/event_hub.hpp"
+#include "rpc/task_action_port.hpp"
 #include "store/sqlite.hpp"
 #include "velox_proto.hpp"
 
@@ -22,7 +23,15 @@ namespace velox::daemon::rpc {
 
 class VeloxDispatcher final : public velox::proto::Dispatcher {
 public:
-    VeloxDispatcher(velox::daemon::store::Db& db, EventHub& hub) : db_(db), hub_(hub) {}
+    // `actions` drives download.pause/resume/start/cancel and queue.start/stop
+    // immediately (D4b) — those cannot wait for the next tick(), unlike download.add's
+    // on_mutation nudge. Optional so existing tests that only exercise download.add/list/
+    // get keep building with no scheduler at hand; a null actions_ makes those methods
+    // answer "not implemented" instead of crashing. See rpc/task_action_port.hpp for why
+    // this is an interface owned by rpc/ rather than a direct sched::Scheduler* (avoids a
+    // veloxd_rpc <-> veloxd_sched circular library dependency).
+    VeloxDispatcher(velox::daemon::store::Db& db, EventHub& hub, TaskActionPort* actions = nullptr)
+        : db_(db), hub_(hub), actions_(actions) {}
 
     // Called after a handler mutates task state (download.add for now). main.cpp wires it
     // to nudge the scheduler; unset in tests.
@@ -109,6 +118,7 @@ public:
 private:
     velox::daemon::store::Db& db_;
     EventHub& hub_;
+    TaskActionPort* actions_;
     std::function<void()> on_mutation_;
 };
 

daemon/src/rpc/task_action_port.hpp

@@ -0,0 +1,44 @@
+#pragma once
+
+// The seam between the dispatcher and the scheduler for user-initiated task/queue actions
+// (download.pause/resume/start/cancel, queue.stop's pauseRunning) — owned by rpc/ so
+// dispatcher.hpp (part of veloxd_rpc) never has to include sched/scheduler.hpp, which
+// would make veloxd_rpc depend on veloxd_sched at compile time. veloxd_sched already
+// depends on veloxd_rpc (for EventHub); the other direction too would be a real circular
+// library dependency, not just an inconvenience — anything linking veloxd_rpc alone (e.g.
+// the CLI's tests) would fail to link over symbols it never calls.
+//
+// sched::Scheduler implements this directly (it already lives in a library that depends on
+// rpc/, so adding an rpc-defined base costs nothing new); main.cpp hands the dispatcher a
+// `TaskActionPort*` pointing at the same Scheduler it constructs.
+
+#include <string>
+#include <vector>
+
+namespace velox::daemon::rpc {
+
+class TaskActionPort {
+public:
+    virtual ~TaskActionPort() = default;
+
+    // Mirrors sched::Scheduler::UserActionResult: whether the task was found at all,
+    // whether it actually changed state (a task already in the target/a terminal state is
+    // reported found=true, changed=false — BulkTaskResult's own "not an error" contract),
+    // and its resulting/current state spelling either way.
+    struct Result {
+        bool found = false;
+        bool changed = false;
+        std::string state;
+    };
+
+    virtual Result user_pause(const std::string& wire_id) = 0;
+    virtual Result user_resume(const std::string& wire_id) = 0;
+    virtual Result user_start(const std::string& wire_id) = 0;
+    virtual Result user_cancel(const std::string& wire_id, bool discard_partial) = 0;
+
+    // queue.stop(pauseRunning=true): pause every currently-running task in `queue_id` now.
+    // Returns the wire ids actually paused.
+    virtual std::vector<std::string> pause_queue(const std::string& queue_id) = 0;
+};
+
+}  // namespace velox::daemon::rpc

daemon/src/sched/scheduler.cpp

@@ -291,7 +291,16 @@ void Scheduler::transition(const std::string& wire_id, std::string_view to_state
     // 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";
+    if (to_state == "paused" && err) {
+        reason = "auto";
+    } else if (to_state == "paused" && !reason && before && before->has_value()) {
+        // No reason supplied — the common case is the engine's own pause-ack callback
+        // (on_state(_, paused, nullopt)) arriving after whoever actually initiated the
+        // pause (user_pause() or tick()'s to_pause loop) already wrote the real reason
+        // eagerly. Keep what's already stored instead of clobbering it back to NULL:
+        // set_state() always overwrites the column, reason or not.
+        reason = (**before).pause_reason;
+    }
     (void)tasks.set_state(wire_id, to_state, reason);
 
     if (err) {
@@ -472,4 +481,98 @@ std::vector<Scheduler::ProgressRow> Scheduler::progress_snapshot() {
     return out;
 }
 
+namespace {
+bool is_terminal_state(const std::string& s) {
+    return s == "complete" || s == "failed" || s == "cancelled";
+}
+}  // namespace
+
+rpc::TaskActionPort::Result Scheduler::user_pause(const std::string& wire_id) {
+    store::Tasks tasks(db_);
+    auto got = tasks.get(wire_id);
+    if (!got || !got->has_value()) return {false, false, {}};
+    const store::TaskRow& row = **got;
+    if (is_terminal_state(row.state) || row.state == "paused")
+        return {true, false, row.state};
+
+    if (auto eid = engine_id_of(wire_id)) engine_.pause(*eid);
+    transition(wire_id, "paused", std::string("user"), std::nullopt);
+    return {true, true, "paused"};
+}
+
+rpc::TaskActionPort::Result Scheduler::user_resume(const std::string& wire_id) {
+    store::Tasks tasks(db_);
+    auto got = tasks.get(wire_id);
+    if (!got || !got->has_value()) return {false, false, {}};
+    const store::TaskRow& row = **got;
+    if (row.state != "paused") return {true, false, row.state};
+
+    if (auto eid = engine_id_of(wire_id)) {
+        engine_.resume(*eid);
+        transition(wire_id, "connecting", std::nullopt, std::nullopt);
+        return {true, true, "connecting"};
+    }
+    transition(wire_id, "queued", std::nullopt, std::nullopt);
+    return {true, true, "queued"};
+}
+
+rpc::TaskActionPort::Result Scheduler::user_start(const std::string& wire_id) {
+    store::Tasks tasks(db_);
+    auto got = tasks.get(wire_id);
+    if (!got || !got->has_value()) return {false, false, {}};
+    const store::TaskRow& row = **got;
+    if (row.state != "paused" && row.state != "new") return {true, false, row.state};
+
+    if (auto eid = engine_id_of(wire_id)) {
+        engine_.resume(*eid);
+        transition(wire_id, "connecting", std::nullopt, std::nullopt);
+        return {true, true, "connecting"};
+    }
+    transition(wire_id, "queued", std::nullopt, std::nullopt);
+    return {true, true, "queued"};
+}
+
+rpc::TaskActionPort::Result Scheduler::user_cancel(const std::string& wire_id,
+                                                   bool discard_partial) {
+    store::Tasks tasks(db_);
+    auto got = tasks.get(wire_id);
+    if (!got || !got->has_value()) return {false, false, {}};
+    const store::TaskRow& row = **got;
+    if (is_terminal_state(row.state)) return {true, false, row.state};
+
+    // Same pattern as tick()'s to_pause loop: call the engine (async, no synchronous
+    // effect) and transition the store eagerly so download.get/list are correct the
+    // instant this call returns. The engine's own on_state(_, cancelled, nullopt) +
+    // on_finished arrive later via on_engine_state, which is what actually
+    // release()s/unmaps the handle — never done here.
+    if (auto eid = engine_id_of(wire_id)) engine_.cancel(*eid, discard_partial);
+    transition(wire_id, "cancelled", std::nullopt, std::nullopt);
+    return {true, true, "cancelled"};
+}
+
+std::vector<std::string> Scheduler::pause_queue(const std::string& queue_id) {
+    store::Tasks tasks(db_);
+    proto::TaskFilter filter;
+    filter.queueId = queue_id;
+    filter.states = non_terminal_states();
+    // No paging needed: a queue's max_concurrent is <= 32, so "everything non-terminal in
+    // this queue" is never a large page.
+    auto page = tasks.list(filter, std::nullopt, 0, 10000);
+    std::vector<std::string> paused;
+    if (!page) return paused;
+
+    for (const auto& row : page->rows) {
+        if (run_state_of(row.state) != RunState::Running) continue;
+        if (auto eid = engine_id_of(row.task_id)) engine_.pause(*eid);
+        transition(row.task_id, "paused", std::string("queue_stopped"), std::nullopt);
+        paused.push_back(row.task_id);
+    }
+    return paused;
+}
+
+void Scheduler::probe_now(const vdm::net::ProbeRequest& req,
+                          std::function<void(vdm::Result<vdm::net::ProbeResult>)> done) {
+    engine_.probe(req, std::move(done));
+}
+
 }  // namespace velox::daemon::sched

daemon/src/sched/scheduler.hpp

@@ -25,6 +25,7 @@
 #include <vector>
 
 #include "rpc/event_hub.hpp"
+#include "rpc/task_action_port.hpp"
 #include "sched/engine_port.hpp"
 #include "sched/governor.hpp"
 #include "store/sqlite.hpp"
@@ -45,7 +46,12 @@ struct TaskErrorFields {
     std::optional<std::int64_t> attempt;
 };
 
-class Scheduler {
+// Implements rpc::TaskActionPort directly — sched/ already depends on rpc/ (EventHub), so
+// this costs nothing new, and it's what lets dispatcher.hpp depend on the port interface
+// instead of on sched/scheduler.hpp (see rpc/task_action_port.hpp's top comment for why
+// that matters: it would otherwise make veloxd_rpc <-> veloxd_sched a circular library
+// dependency).
+class Scheduler final : public rpc::TaskActionPort {
 public:
     // `local_now` returns a fully-populated std::tm in local time; injected so tests can
     // pin the clock. `post_to_loop` marshals an engine-thread callback onto the loop
@@ -98,6 +104,42 @@ public:
     };
     std::vector<ProgressRow> progress_snapshot();
 
+    // rpc::TaskActionPort. These apply immediately — never wait for the next tick() —
+    // because pausing, resuming or cancelling a live transfer cannot wait up to 1s for the
+    // timerfd, and the governor will never do any of them on its own for a user-owned
+    // reason (ADR 0013 §3: "never touch a task paused for a reason it does not own").
+    // Idempotent: calling one on a task already in the target (or a terminal) state
+    // reports found=true, changed=false.
+    rpc::TaskActionPort::Result user_pause(const std::string& wire_id) override;
+    // A task still holding a live engine handle (paused mid-flight) is engine_.resume()'d
+    // straight back to `connecting`; one with no handle yet (parked since download.add
+    // with startMode 'later', or never admitted) goes to `queued` for the next tick's
+    // normal admission.
+    rpc::TaskActionPort::Result user_resume(const std::string& wire_id) override;
+    // "Begin or restart the given tasks" (download.start): same effect as user_resume for
+    // a paused/new task. NOTE: the contract's "a task in 'queued' jumps its queue" priority
+    // bump is not implemented — admission is still plain FIFO via the governor's
+    // created_at rank. Flagged in deferrals.md.
+    rpc::TaskActionPort::Result user_start(const std::string& wire_id) override;
+    // download.cancel == cancel(discard_partial=false); download.remove == cancel(true)
+    // plus the store row / file cleanup (that part is still D3).
+    rpc::TaskActionPort::Result user_cancel(const std::string& wire_id,
+                                            bool discard_partial) override;
+
+    // queue.stop(pauseRunning=true): pause every task in `queue_id` the governor would
+    // currently call Running, right now rather than waiting for the next tick — the same
+    // immediacy reasoning as the user_* actions above, with PauseReason::QueueStopped
+    // instead of User. Returns the wire ids actually paused.
+    std::vector<std::string> pause_queue(const std::string& queue_id) override;
+
+    // download.probe's standalone use (File Info dialog, no task row involved): a thin
+    // passthrough to the engine's own probe pool, outside the segment budget (ADR 0011
+    // §5). Never blocks — `done` arrives on an engine thread like every other EnginePort
+    // callback; the caller (the RPC server layer, not this synchronous dispatcher — see
+    // rpc/dispatcher.hpp's top comment) is responsible for marshalling the reply back.
+    void probe_now(const vdm::net::ProbeRequest& req,
+                   std::function<void(vdm::Result<vdm::net::ProbeResult>)> done);
+
     // Diagnostics / tests.
     std::optional<std::string> wire_id_of(vdm::TaskId id) const;
     std::optional<vdm::TaskId> engine_id_of(const std::string& wire_id) const;

daemon/src/store/queues.cpp

@@ -1,11 +1,44 @@
 #include "store/queues.hpp"
 
+#include <sqlite3.h>
+
 #include <nlohmann/json.hpp>
 
 namespace velox::daemon::store {
 
 namespace proto = velox::proto;
 
+namespace {
+
+// One queue row (columns queue_id, name, state, max_concurrent, schedule, in that order)
+// plus its member taskIds, read off the row a caller has already step()'d to.
+DbResult<proto::Queue> project_row(Db& db, Stmt& st) {
+    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);
+    return q;
+}
+
+}  // namespace
+
 DbResult<std::vector<proto::Queue>> Queues::list() {
     auto st = db_.prepare(
         "SELECT queue_id, name, state, max_concurrent, schedule FROM queues ORDER BY name");
@@ -16,33 +49,33 @@ DbResult<std::vector<proto::Queue>> Queues::list() {
         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));
+        auto q = project_row(db_, *st);
+        if (!q) return std::unexpected(q.error());
+        out.push_back(std::move(*q));
     }
     return out;
 }
 
+DbResult<std::optional<proto::Queue>> Queues::get(std::string_view queue_id) {
+    auto st = db_.prepare(
+        "SELECT queue_id, name, state, max_concurrent, schedule FROM queues WHERE queue_id = ?1");
+    if (!st) return std::unexpected(st.error());
+    if (auto b = st->bind(1, queue_id); !b) return std::unexpected(b.error());
+    auto row = st->step();
+    if (!row) return std::unexpected(row.error());
+    if (!*row) return std::optional<proto::Queue>{};
+    auto q = project_row(db_, *st);
+    if (!q) return std::unexpected(q.error());
+    return std::optional<proto::Queue>{std::move(*q)};
+}
+
+DbResult<bool> Queues::set_state(std::string_view queue_id, std::string_view state) {
+    auto st = db_.prepare("UPDATE queues SET state = ?2 WHERE queue_id = ?1");
+    if (!st) return std::unexpected(st.error());
+    if (auto b = st->bind(1, queue_id); !b) return std::unexpected(b.error());
+    if (auto b = st->bind(2, state); !b) return std::unexpected(b.error());
+    if (auto r = st->step(); !r) return std::unexpected(r.error());
+    return sqlite3_changes(db_.raw()) > 0;
+}
+
 }  // namespace velox::daemon::store

daemon/src/store/queues.hpp

@@ -5,6 +5,8 @@
 // — membership changes through download.update / queue.reorder, per Queue's own schema
 // note that a queue.upsert payload's taskIds is ignored.
 
+#include <optional>
+#include <string_view>
 #include <vector>
 
 #include "store/sqlite.hpp"
@@ -18,6 +20,13 @@ public:
 
     DbResult<std::vector<velox::proto::Queue>> list();
 
+    // nullopt (not an error) if no queue has this id.
+    DbResult<std::optional<velox::proto::Queue>> get(std::string_view queue_id);
+
+    // 'running' or 'stopped' (Queue.schema.json / the state column's CHECK). false if the
+    // id doesn't exist.
+    DbResult<bool> set_state(std::string_view queue_id, std::string_view state);
+
 private:
     Db& db_;
 };