daemon: D3 (partial) — category.upsert/remove, queue.upsert, download.remove/addBatch/provideAuth

Closes a bounded, high-value slice of the remaining D3 stubs. settings.*/rules.*/ limiter.*/schedule.*/grabber.*/media.*/capture.*/queue.reorder/download.refreshUrl/ download.update stay deferred — reasons noted individually in deferrals.md (settings.* specifically: a real, large field<->key<->JSON-type mapping table across ~43 fields / ~50 SettingKeys, not started rather than rushed). category.upsert/remove: store/categories.hpp gains get/upsert/remove. upsert generates an id when absent and always ignores the payload's `builtin` (preserved from the existing row on replace, false on create); saveDir goes through the same fs::resolve_target canonicalize-and-root-check as download.add. remove refuses a builtin at both layers (dispatcher's -32602 pre-check; the store's own "DELETE ... AND builtin = 0" as defense in depth) and reassigns member tasks to reassignTo (default "general") inside one transaction before deleting the row. queue.upsert: store/queues.hpp gains get/upsert (set_state already existed from D4b). Same create-generates-id pattern; a create always starts 'stopped', a replace keeps the queue's current run state (upsert edits config, not run state — that's queue.start/stop). Also fixed in passing: on_complete has been a real column since migration 0001 but Queues::list/get never projected it onto Queue.onComplete. download.remove: cancels with discard_partial=true (always drops the .veloxpart pair — unlike download.cancel, which keeps them, the row is gone either way), deletes the finished file only when deleteFile is true and the task was complete (best-effort), deletes the row (segments cascade via the FK), and publishes event.task.removed (closing the last open note under D5). download.addBatch: on_download_add's body is now a shared add_one(), called once per item after merging each item's unset fields against params.defaults. download.provideAuth: forwards to EnginePort::provide_auth via a new TaskActionPort::provide_auth. `remember`/persisting to the Secret Service is accepted but not acted on — nothing in this build talks to libsecret yet. Verified against real veloxd + tools/testserver: category create/replace/ remove-with-reassignment, queue create/replace-keeps-run-state, a batch add sharing defaults.saveDir, and download.remove with deleteFile actually deleting the file and the task then 404ing download.get with -32010. Full ctest: 39/39. Co-Authored-By: Claude Sonnet 5 <[email protected]> Claude-Session: https://claude.ai/code/session_01GRDjHGgpYmMoPE2UFbe7pP
2026-09-11 17:43:43 +04:00
parent 4f6fb1029d
commit c89158ea09
11 changed files with 493 additions and 31 deletions
@@ -7,7 +7,11 @@ 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~~ | **Closed** — `download.probe` is real on both transports. It's genuinely async (the engine's probe pool, up to the schema's 30s `x-deadlineMs`) and so cannot fit `VeloxDispatcher::on_download_probe`'s synchronous `HandlerResult<T>` return — `uds_server.cpp`/`ws_server.cpp` special-case `"download.probe"` before the generic `dispatch()`, exactly the way they already special-case `session.hello`/`session.subscribe`, and queue the reply whenever the callback fires. `rpc::TaskActionPort::probe_now` (kept in proto/std terms, no `vdm::net::*`, so `veloxd_rpc` never needs `core/include`'s vdm headers) is what both transports call; `sched::Scheduler::probe_now` is the implementation — builds a `vdm::net::ProbeRequest`, runs it on the engine's probe pool, maps a failure to `-32013 ProbeFailed` (with `data.httpStatus` when there was one), and fills `suggestedCategoryId`/`suggestedSaveDir` with a plain extension match against the categories table (not the real rules engine — that's still D3). Verified live: a real probe answers in ~5ms; a bad host maps to `-32013`; a connection issuing a 10s `slow-loris` probe does not block a second connection's `download.list` (answered in ~1ms) — confirms the async design actually keeps the loop free, not just compiles. | `rpc/task_action_port.hpp`, `rpc/{uds_server,ws_server}.{hpp,cpp}`, `sched/scheduler.{cpp,hpp}` | — | done |
-| D3 | Stub handlers for the rest: `download.remove/addBatch/refreshUrl/provideAuth/update`, `rules.*`, `settings.*`, `limiter.*`, `schedule.*`, `queue.upsert/reorder`, `category.upsert/remove`, `grabber.*`, `media.*`, `capture.*` | `rpc/dispatcher.cpp` | No store/scheduler wiring behind them yet. `category.list`, `queue.list`, `queue.start`/`stop` are done | Per method, as each wires to the store/scheduler |
+| D3 | Stub handlers for the rest: `download.refreshUrl/update`, `rules.*`, `settings.*`, `limiter.*`, `schedule.*`, `queue.reorder`, `grabber.*`, `media.*`, `capture.*` | `rpc/dispatcher.cpp` | No store/scheduler wiring behind them yet, or (`settings.*`) sound but large — see the note below. `category.list/upsert/remove`, `queue.list/upsert/start/stop`, `download.remove/addBatch/provideAuth` are done | Per method, as each wires to the store/scheduler |
+| — | **`settings.get`/`settings.set` specifically, not started:** `proto::Settings` is a flat struct of ~43 `std::optional` fields, one per `SettingKey` (~50 keys) in `Settings.schema.json`; `store::Settings` already has `get_raw`/`set_raw`/`overrides` keyed by the same dotted strings the JSON uses (`"connection.maxSegmentsPerDownload"`, …). The handlers are a mechanical field <-> key <-> JSON-type mapping table in both directions (get: row-or-default -> struct field; set: struct field -> validate against the key's schema type -> `set_raw`, collecting `changed`) — real work, just long and repetitive rather than hard. Left alone this pass rather than rushed; every other read of settings in this codebase already goes through `store::Settings`'s typed helpers directly (`reload_config`, `on_download_add`'s segment default, `capture.offer`'s allowed roots when that lands), so nothing downstream is blocked on the RPC surface existing. | `rpc/dispatcher.cpp`, `store/settings.{hpp,cpp}` | the mapping table is genuinely large, not genuinely hard | its own pass |
+| ~~D3a~~ | **Closed** — `category.upsert`/`category.remove`: `store/categories.hpp` gains `get`/`upsert`/`remove`. `upsert` generates an id when absent (create) and always ignores the payload's `builtin` (preserved from the existing row on replace, false on create — a client can never mint or revoke it); the `saveDir` goes through the same `fs::resolve_target` canonicalize-and-root-check as `download.add` (`-32011` on failure). `remove` refuses a builtin at both layers (dispatcher pre-checks for the `-32602` error text; the store's own `DELETE ... AND builtin = 0` is defense in depth) and reassigns member tasks to `reassignTo` (default `"general"`) inside one transaction before deleting the row. Note: the `categories` table (0001) has no columns for `Category.mimeTypes`/`.sortOrder` — accepted on `upsert` but not persisted. | `store/categories.{hpp,cpp}`, `rpc/dispatcher.cpp` | — | done, `mimeTypes`/`sortOrder` gap noted |
+| ~~D3b~~ | **Closed** — `queue.upsert`: `store/queues.hpp` gains `get`/`upsert` (`set_state` already existed from D4b). Same create-generates-id pattern as categories; `taskIds` in the payload is ignored (schema's own note) and a create always starts `'stopped'` while a replace keeps the queue's current run state — `queue.upsert` edits config, not run state (that's `queue.start`/`stop`). Also fixed: `on_complete` was a real column since 0001 but `Queues::list`/`get` never projected it onto `Queue.onComplete` — now they do. | `store/queues.{hpp,cpp}`, `rpc/dispatcher.cpp` | — | done |
+| ~~D3c~~ | **Closed** — `download.remove`: cancels with `discard_partial=true` through `TaskActionPort` (always drops any `.veloxpart`/`.veloxpart.meta` — the row is gone either way, unlike `download.cancel`, which keeps them), deletes the finished file only when `deleteFile` is true and the task was `complete` (best-effort — a missing file doesn't fail the call), deletes the row (segments cascade via the FK), and publishes `event.task.removed` (closing the last open note under D5). `download.addBatch`: `on_download_add`'s body is now a shared `add_one()`, called once per item after merging each item's unset fields against `params.defaults`. `download.provideAuth`: forwards to `EnginePort::provide_auth` through a new `TaskActionPort::provide_auth`; `remember`/persisting to the Secret Service is accepted but not acted on — nothing in this build talks to libsecret yet (verified: no such integration exists anywhere in the tree). Verified against real `veloxd` + `tools/testserver`: category create/replace/remove-with-reassignment, queue create/replace-keeps-state, a batch add with shared `defaults.saveDir`, and remove-with-deleteFile actually deleting the file and the task then 404ing `download.get` with `-32010`. | `rpc/dispatcher.{hpp,cpp}`, `rpc/task_action_port.hpp`, `sched/scheduler.{cpp,hpp}` | `download.provideAuth`'s `remember` (needs the Secret Service, unbuilt) | done, `remember` persistence gap noted |
 | ~~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~~ | **Closed** — `download.pause`/`resume`/`start`/`cancel` and `queue.start`/`stop` all drive the scheduler now, and apply *immediately* (not deferred to the next tick — pausing/resuming/cancelling a live transfer can't wait up to 1s, 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 the seam dispatcher.hpp depends on instead of `sched/scheduler.hpp` directly — avoids a real `veloxd_rpc` <-> `veloxd_sched` circular library dependency (`veloxd_sched` already links `veloxd_rpc` for `EventHub`). `Scheduler::user_pause/resume/start/cancel` + `pause_queue` engine-call-then-eager-transition, matching `tick()`'s existing `to_pause` pattern. Fixed a real bug hit while building this: `transition()` always overwrote `pause_reason` to NULL when the engine's own delayed pause-ack callback arrived with no explicit reason, clobbering whatever the actual initiator (user or governor) had just written — now it preserves the stored reason when none is supplied. Verified against real `veloxd` + `tools/testserver`: pausing a live single-segment throttled transfer freezes `downloadedBytes`, resume continues it from that point, cancel stops it; `queue.stop(pauseRunning:true)` pauses the queue's running task immediately. NOTE: `download.start`'s contract "a task in 'queued' jumps its queue" (priority bump) is not implemented — admission is still plain FIFO by `created_at`. | `sched/scheduler.{cpp,hpp}`, `rpc/task_action_port.hpp`, `rpc/dispatcher.{hpp,cpp}`, `store/queues.{cpp,hpp}` | — | done, except the queue-jump priority bump noted above |
 | ~~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 |
@@ -1,5 +1,6 @@
 #include "rpc/dispatcher.hpp"

+#include <filesystem>
 #include <random>
 #include <string>

@@ -141,9 +142,13 @@ VeloxDispatcher::on_download_list(const proto::DownloadListParams& params) {
 }

 // --- download.add : canonicalise + root-check the destination, then persist -----------
+//
+// add_one() is the whole of download.add's body; on_download_add and on_download_addBatch
+// (each item merged against `defaults` first) both call it, so there is exactly one place
+// that turns a DownloadSpec into a stored, admitted task.

-proto::HandlerResult<proto::DownloadAddResult>
-VeloxDispatcher::on_download_add(const proto::DownloadSpec& spec) {
+proto::HandlerResult<proto::DownloadAddResult> VeloxDispatcher::add_one(
+    const proto::DownloadSpec& spec) {
    store::Settings settings(db_);

    std::string save_dir = spec.saveDir && !spec.saveDir->empty()
@@ -226,6 +231,11 @@ VeloxDispatcher::on_download_add(const proto::DownloadSpec& spec) {
    return r;
 }

+proto::HandlerResult<proto::DownloadAddResult>
+VeloxDispatcher::on_download_add(const proto::DownloadSpec& spec) {
+    return add_one(spec);
+}
+
 // --- everything else : not implemented until the store and scheduler land -------------

 proto::HandlerResult<proto::CaptureRules>
@@ -248,16 +258,94 @@ VeloxDispatcher::on_category_list(const proto::CategoryListParams&) {
    return r;
 }
 proto::HandlerResult<proto::CategoryRemoveResult>
-VeloxDispatcher::on_category_remove(const proto::CategoryRemoveParams&) {
-    return not_implemented<proto::CategoryRemoveResult>("category.remove");
+VeloxDispatcher::on_category_remove(const proto::CategoryRemoveParams& params) {
+    store::Categories categories(db_);
+    // A builtin category refuses with -32602 (category.remove's own words) — checked here
+    // rather than inferred from RemoveResult.removed=false, which also covers "no such
+    // category" and would otherwise collapse two different error stories into one.
+    auto existing = categories.get(params.categoryId);
+    if (!existing)
+        return std::unexpected(proto::HandlerError{proto::ErrorCode::InternalError,
+                                                   "category.remove: " + existing.error().message});
+    if (!existing->has_value())
+        return std::unexpected(proto::HandlerError{
+            proto::ErrorCode::InvalidParams, "no such category",
+            nlohmann::json{{"categoryId", params.categoryId}}});
+    if ((*existing)->builtin)
+        return std::unexpected(proto::HandlerError{
+            proto::ErrorCode::InvalidParams, "builtin categories cannot be removed",
+            nlohmann::json{{"categoryId", params.categoryId}}});
+
+    auto removed = categories.remove(params.categoryId, params.reassignTo);
+    if (!removed)
+        return std::unexpected(proto::HandlerError{proto::ErrorCode::InternalError,
+                                                   "category.remove: " + removed.error().message});
+    proto::CategoryRemoveResult r;
+    r.removed = removed->removed;
+    r.reassignedTaskIds = removed->reassigned_task_ids;
+    if (on_mutation_) on_mutation_();  // download.list's categoryId column just moved
+    return r;
 }
 proto::HandlerResult<proto::CategoryUpsertResult>
-VeloxDispatcher::on_category_upsert(const proto::CategoryUpsertParams&) {
-    return not_implemented<proto::CategoryUpsertResult>("category.upsert");
+VeloxDispatcher::on_category_upsert(const proto::CategoryUpsertParams& params) {
+    store::Settings settings(db_);
+    std::string save_dir = expand_tilde(params.category.saveDir);
+
+    std::vector<std::string> roots;
+    for (const auto& r : settings.get_string_array("saveTo.allowedRoots")) {
+        if (auto c = fs::canonicalize_root(r)) roots.push_back(*c);
+    }
+    // No file is ever written for this marker leaf — resolve_target only validates/creates
+    // the directory chain (fs/safepath.hpp); it never creates the leaf itself.
+    auto target = fs::resolve_target(save_dir, ".category-marker", roots);
+    if (!target) {
+        return std::unexpected(proto::HandlerError{
+            proto::ErrorCode::InvalidPath, target.error().message,
+            nlohmann::json{{"path", params.category.saveDir}}});
+    }
+
+    store::Categories categories(db_);
+    proto::Category to_store = params.category;
+    to_store.saveDir = target->dir;
+    auto stored = categories.upsert(std::move(to_store));
+    if (!stored)
+        return std::unexpected(proto::HandlerError{proto::ErrorCode::InternalError,
+                                                   "category.upsert: " + stored.error().message});
+    proto::CategoryUpsertResult r;
+    r.category = std::move(*stored);
+    return r;
 }
 proto::HandlerResult<proto::DownloadAddBatchResult>
-VeloxDispatcher::on_download_addBatch(const proto::DownloadAddBatchParams&) {
-    return not_implemented<proto::DownloadAddBatchResult>("download.addBatch");
+VeloxDispatcher::on_download_addBatch(const proto::DownloadAddBatchParams& params) {
+    proto::DownloadAddBatchResult r;
+    for (std::size_t i = 0; i < params.items.size(); ++i) {
+        proto::DownloadSpec spec = params.items[i];  // "its url is ignored" only for defaults
+        if (params.defaults) {
+            const auto& d = *params.defaults;
+            if (!spec.headers) spec.headers = d.headers;
+            if (!spec.cookies) spec.cookies = d.cookies;
+            if (!spec.referrer) spec.referrer = d.referrer;
+            if (!spec.userAgent) spec.userAgent = d.userAgent;
+            if (!spec.filename) spec.filename = d.filename;
+            if (!spec.saveDir) spec.saveDir = d.saveDir;
+            if (!spec.categoryId) spec.categoryId = d.categoryId;
+            if (!spec.queueId) spec.queueId = d.queueId;
+            if (!spec.segments) spec.segments = d.segments;
+            if (!spec.bufferBytes) spec.bufferBytes = d.bufferBytes;
+            if (!spec.startMode) spec.startMode = d.startMode;
+            if (!spec.description) spec.description = d.description;
+            if (!spec.checksum) spec.checksum = d.checksum;
+        }
+
+        auto added = add_one(spec);
+        if (added) {
+            r.taskIds.push_back(added->taskId);
+        } else {
+            r.failed.push_back({static_cast<std::int64_t>(i), added.error().code,
+                               added.error().message});
+        }
+    }
+    return r;
 }
 proto::HandlerResult<proto::BulkTaskResult>
 VeloxDispatcher::on_download_cancel(const proto::DownloadCancelParams& params) {
@@ -313,16 +401,56 @@ VeloxDispatcher::on_download_probe(const proto::DownloadProbeParams&) {
    return not_implemented<proto::DownloadProbeResult>("download.probe");
 }
 proto::HandlerResult<proto::DownloadProvideAuthResult>
-VeloxDispatcher::on_download_provideAuth(const proto::DownloadProvideAuthParams&) {
-    return not_implemented<proto::DownloadProvideAuthResult>("download.provideAuth");
+VeloxDispatcher::on_download_provideAuth(const proto::DownloadProvideAuthParams& params) {
+    if (!actions_) return not_implemented<proto::DownloadProvideAuthResult>("download.provideAuth");
+    proto::DownloadProvideAuthResult r;
+    r.ok = actions_->provide_auth(params.taskId, params.username, params.password,
+                                  params.save.value_or(false));
+    return r;
 }
 proto::HandlerResult<proto::DownloadRefreshUrlResult>
 VeloxDispatcher::on_download_refreshUrl(const proto::DownloadRefreshUrlParams&) {
    return not_implemented<proto::DownloadRefreshUrlResult>("download.refreshUrl");
 }
 proto::HandlerResult<proto::DownloadRemoveResult>
-VeloxDispatcher::on_download_remove(const proto::DownloadRemoveParams&) {
-    return not_implemented<proto::DownloadRemoveResult>("download.remove");
+VeloxDispatcher::on_download_remove(const proto::DownloadRemoveParams& params) {
+    store::Tasks tasks(db_);
+    proto::DownloadRemoveResult r;
+    for (const auto& id : params.taskIds) {
+        auto got = tasks.get(id);
+        if (!got || !got->has_value()) {
+            r.failed.push_back({id, proto::ErrorCode::TaskNotFound, "no such task"});
+            continue;
+        }
+        const store::TaskRow row = **got;  // copy: row is gone once tasks.remove() runs
+
+        // Always discard the .veloxpart/.veloxpart.meta pair (the schema's own words) —
+        // download.remove deletes the row outright, so there is no reason to leave a
+        // partial behind the way download.cancel does. A no-op if the task never had a
+        // live engine handle (already terminal, or never started).
+        if (actions_) (void)actions_->user_cancel(id, /*discard_partial=*/true);
+
+        // The finished file only when deleteFile is true — never touched for a partial or
+        // failed transfer (there is nothing there but what discard_partial above already
+        // took care of). Best-effort: a missing file is not a reason to fail the remove.
+        bool deleted_file = false;
+        if (params.deleteFile && row.state == "complete") {
+            std::error_code ec;
+            deleted_file =
+                std::filesystem::remove(std::filesystem::path(row.save_dir) / row.filename, ec);
+        }
+
+        (void)tasks.remove(id);  // segments cascade via the FK (ON DELETE CASCADE)
+        r.removed.push_back(id);
+
+        hub_.publish(proto::Event::TaskRemoved,
+                    proto::make_notification(
+                        proto::Event::TaskRemoved,
+                        nlohmann::json{{"taskId", id}, {"deletedFile", deleted_file}}),
+                    id);
+    }
+    if (on_mutation_) on_mutation_();
+    return r;
 }
 proto::HandlerResult<proto::BulkTaskResult>
 VeloxDispatcher::on_download_resume(const proto::DownloadResumeParams& params) {
@@ -436,8 +564,15 @@ VeloxDispatcher::on_queue_stop(const proto::QueueStopParams& params) {
    return r;
 }
 proto::HandlerResult<proto::QueueUpsertResult>
-VeloxDispatcher::on_queue_upsert(const proto::QueueUpsertParams&) {
-    return not_implemented<proto::QueueUpsertResult>("queue.upsert");
+VeloxDispatcher::on_queue_upsert(const proto::QueueUpsertParams& params) {
+    store::Queues queues(db_);
+    auto stored = queues.upsert(params.queue);
+    if (!stored)
+        return std::unexpected(proto::HandlerError{proto::ErrorCode::InternalError,
+                                                   "queue.upsert: " + stored.error().message});
+    proto::QueueUpsertResult r;
+    r.queue = std::move(*stored);
+    return r;
 }
 proto::HandlerResult<proto::RulesListResult>
 VeloxDispatcher::on_rules_list(const proto::RulesListParams&) {
@@ -116,6 +116,12 @@ public:
    on_settings_set(const velox::proto::SettingsSetParams&) override;

 private:
+    // The whole of download.add's body; on_download_add and on_download_addBatch (each
+    // item merged against DownloadAddBatchParams.defaults first) both call this — exactly
+    // one place turns a DownloadSpec into a stored, admitted task.
+    velox::proto::HandlerResult<velox::proto::DownloadAddResult> add_one(
+        const velox::proto::DownloadSpec& spec);
+
    velox::daemon::store::Db& db_;
    EventHub& hub_;
    TaskActionPort* actions_;
@@ -43,6 +43,14 @@ public:
    // Returns the wire ids actually paused.
    virtual std::vector<std::string> pause_queue(const std::string& queue_id) = 0;

+    // download.provideAuth: answers a task auto-paused on a 401/407. false if the task
+    // isn't currently holding a live engine handle (nothing waiting on credentials).
+    // `remember` is accepted but not yet acted on — persisting to the Secret Service isn't
+    // wired anywhere in this build yet (CLAUDE.md §4: never SQLite, never logs); this
+    // always does the "this retry only" half. Noted in deferrals.md.
+    virtual bool provide_auth(const std::string& wire_id, const std::string& username,
+                              const std::string& password, bool remember) = 0;
+
    // download.probe (D2), the File Info dialog's own network round trip — no task row
    // involved. Genuinely async (the engine's probe pool; up to the schema's 30s
    // x-deadlineMs) and so cannot fit VeloxDispatcher's synchronous on_download_probe:
@@ -571,6 +571,15 @@ std::vector<std::string> Scheduler::pause_queue(const std::string& queue_id) {
    return paused;
 }

+bool Scheduler::provide_auth(const std::string& wire_id, const std::string& username,
+                             const std::string& password, bool remember) {
+    (void)remember;  // not yet wired to the Secret Service anywhere in this build
+    auto eid = engine_id_of(wire_id);
+    if (!eid) return false;
+    engine_.provide_auth(*eid, username, password, remember);
+    return true;
+}
+
 namespace {

 // Extension match against the categories table (categories.extensions, per category.list),
@@ -132,6 +132,9 @@ public:
    // instead of User. Returns the wire ids actually paused.
    std::vector<std::string> pause_queue(const std::string& queue_id) override;

+    bool provide_auth(const std::string& wire_id, const std::string& username,
+                      const std::string& password, bool remember) override;
+
    // rpc::TaskActionPort. Builds a vdm::net::ProbeRequest from `params`, runs it on the
    // engine's probe pool (outside the segment budget, ADR 0011 §5), and converts the
    // result back to proto terms — including the suggestedCategoryId/-SaveDir guess (a
@@ -1,11 +1,35 @@
 #include "store/categories.hpp"

+#include <sqlite3.h>
+
+#include <cstdio>
+#include <random>
+#include <string>
+
 #include <nlohmann/json.hpp>

 namespace velox::daemon::store {

 namespace proto = velox::proto;

+namespace {
+
+proto::Category project_row(Stmt& st) {
+    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;
+    return c;
+}
+
+}  // namespace
+
 DbResult<std::vector<proto::Category>> Categories::list() {
    auto st = db_.prepare(
        "SELECT category_id, name, save_dir, extensions, builtin FROM categories "
@@ -17,20 +41,111 @@ DbResult<std::vector<proto::Category>> Categories::list() {
        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));
+        out.push_back(project_row(*st));
    }
    return out;
 }

+DbResult<std::optional<proto::Category>> Categories::get(std::string_view category_id) {
+    auto st = db_.prepare(
+        "SELECT category_id, name, save_dir, extensions, builtin FROM categories "
+        "WHERE category_id = ?1");
+    if (!st) return std::unexpected(st.error());
+    if (auto b = st->bind(1, category_id); !b) return std::unexpected(b.error());
+    auto row = st->step();
+    if (!row) return std::unexpected(row.error());
+    if (!*row) return std::optional<proto::Category>{};
+    return std::optional<proto::Category>{project_row(*st)};
+}
+
+DbResult<proto::Category> Categories::upsert(proto::Category category) {
+    // A replace keeps the existing row's builtin flag; a create is never builtin. Either
+    // way the payload's own `builtin` is ignored — a client cannot mint or revoke it.
+    bool builtin = false;
+    if (!category.categoryId.empty()) {
+        auto existing = get(category.categoryId);
+        if (!existing) return std::unexpected(existing.error());
+        if (existing->has_value()) builtin = (*existing)->builtin;
+    } else {
+        // Reuse Tasks' id scheme (v4 UUID) would need a cross-module include for one
+        // function; a category id has no wire format requirement beyond "a string", so a
+        // timestamp-free random hex id keeps this module self-contained.
+        std::random_device rd;
+        std::uniform_int_distribution<std::uint64_t> d;
+        char buf[17];
+        std::snprintf(buf, sizeof(buf), "%016llx", static_cast<unsigned long long>(d(rd)));
+        category.categoryId = std::string(buf);
+    }
+
+    nlohmann::json ext = nlohmann::json::array();
+    for (const auto& e : category.extensions) ext.push_back(e);
+
+    auto st = db_.prepare(
+        "INSERT INTO categories(category_id, name, save_dir, extensions, builtin) "
+        "VALUES(?1,?2,?3,?4,?5) "
+        "ON CONFLICT(category_id) DO UPDATE SET "
+        "name=excluded.name, save_dir=excluded.save_dir, extensions=excluded.extensions");
+    if (!st) return std::unexpected(st.error());
+    if (auto b = st->bind(1, std::string_view(category.categoryId)); !b)
+        return std::unexpected(b.error());
+    if (auto b = st->bind(2, std::string_view(category.name)); !b) return std::unexpected(b.error());
+    if (auto b = st->bind(3, std::string_view(category.saveDir)); !b)
+        return std::unexpected(b.error());
+    if (auto b = st->bind(4, std::string_view(ext.dump())); !b) return std::unexpected(b.error());
+    if (auto b = st->bind(5, static_cast<std::int64_t>(builtin)); !b)
+        return std::unexpected(b.error());
+    if (auto r = st->step(); !r) return std::unexpected(r.error());
+
+    category.builtin = builtin;
+    return category;
+}
+
+DbResult<Categories::RemoveResult> Categories::remove(
+    std::string_view category_id, const std::optional<std::string>& reassign_to) {
+    RemoveResult out;
+    const std::string target = reassign_to.value_or("general");
+
+    // Db::transaction only threads a DbResult<void> lambda; `out` is filled in-place and
+    // returned once the transaction (which may still fail and roll back) succeeds.
+    auto txn = db_.transaction([&]() -> DbResult<void> {
+        {
+            // A builtin category is never removed, and — since it was never going to be
+            // removed — its tasks must not be reassigned away from it either.
+            auto chk = db_.prepare("SELECT builtin FROM categories WHERE category_id = ?1");
+            if (!chk) return std::unexpected(chk.error());
+            if (auto b = chk->bind(1, category_id); !b) return std::unexpected(b.error());
+            auto row = chk->step();
+            if (!row) return std::unexpected(row.error());
+            if (!*row) return {};  // no such category: removed stays false
+            if (chk->column_int(0) != 0) return {};  // builtin: removed stays false
+        }
+        {
+            auto sel = db_.prepare("SELECT task_id FROM tasks WHERE category_id = ?1");
+            if (!sel) return std::unexpected(sel.error());
+            if (auto b = sel->bind(1, category_id); !b) return std::unexpected(b.error());
+            for (;;) {
+                auto row = sel->step();
+                if (!row) return std::unexpected(row.error());
+                if (!*row) break;
+                out.reassigned_task_ids.push_back(sel->column_text(0));
+            }
+        }
+        if (!out.reassigned_task_ids.empty()) {
+            auto upd = db_.prepare("UPDATE tasks SET category_id = ?2 WHERE category_id = ?1");
+            if (!upd) return std::unexpected(upd.error());
+            if (auto b = upd->bind(1, category_id); !b) return std::unexpected(b.error());
+            if (auto b = upd->bind(2, std::string_view(target)); !b) return std::unexpected(b.error());
+            if (auto r = upd->step(); !r) return std::unexpected(r.error());
+        }
+        auto del = db_.prepare("DELETE FROM categories WHERE category_id = ?1 AND builtin = 0");
+        if (!del) return std::unexpected(del.error());
+        if (auto b = del->bind(1, category_id); !b) return std::unexpected(b.error());
+        if (auto r = del->step(); !r) return std::unexpected(r.error());
+        out.removed = sqlite3_changes(db_.raw()) > 0;
+        return {};
+    });
+    if (!txn) return std::unexpected(txn.error());
+    return out;
+}
+
 }  // namespace velox::daemon::store
@@ -3,7 +3,13 @@
 // 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.
+//
+// The table has no columns for Category.mimeTypes / .sortOrder (0001_initial.sql predates
+// those fields); upsert() accepts them but they are not persisted — round-tripped as unset
+// on the next list()/get(). Noted in daemon/docs/deferrals.md.

+#include <optional>
+#include <string>
 #include <vector>

 #include "store/sqlite.hpp"
@@ -16,6 +22,24 @@ public:
    explicit Categories(Db& db) : db_(db) {}

    DbResult<std::vector<velox::proto::Category>> list();
+    DbResult<std::optional<velox::proto::Category>> get(std::string_view category_id);
+
+    // "Omit categoryId to create; supply it to replace" (category.upsert's own words) —
+    // the caller (dispatcher) decides create vs replace by whether `category.categoryId`
+    // is empty and generates the id; this just writes the row. `builtin` is never taken
+    // from the payload: preserved from the existing row on a replace, always false on a
+    // create (a client can never mint a builtin category).
+    DbResult<velox::proto::Category> upsert(velox::proto::Category category);
+
+    // False for "no such category". A builtin category is never removed — the caller
+    // checks that (category.remove -> -32602) before calling this, since that check needs
+    // ErrorCode, which this module (like the rest of store/) does not depend on.
+    struct RemoveResult {
+        bool removed = false;
+        std::vector<std::string> reassigned_task_ids;
+    };
+    DbResult<RemoveResult> remove(std::string_view category_id,
+                                  const std::optional<std::string>& reassign_to);

 private:
    Db& db_;
@@ -2,6 +2,10 @@

 #include <sqlite3.h>

+#include <cstdio>
+#include <random>
+#include <string>
+
 #include <nlohmann/json.hpp>

 namespace velox::daemon::store {
@@ -10,8 +14,8 @@ 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.
+// One queue row (columns queue_id, name, state, max_concurrent, schedule, on_complete, 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);
@@ -22,6 +26,7 @@ DbResult<proto::Queue> project_row(Db& db, Stmt& st) {
        auto j = nlohmann::json::parse(st.column_text(4), nullptr, false);
        if (auto sched = proto::parse<proto::Schedule>(j, "schedule")) q.schedule = *sched;
    }
+    if (auto oc = proto::parse_QueueOnComplete(st.column_text(5))) q.onComplete = *oc;

    auto ts = db.prepare("SELECT task_id FROM tasks WHERE queue_id = ?1 ORDER BY queue_position");
    if (!ts) return std::unexpected(ts.error());
@@ -41,7 +46,8 @@ DbResult<proto::Queue> project_row(Db& db, Stmt& st) {

 DbResult<std::vector<proto::Queue>> Queues::list() {
    auto st = db_.prepare(
-        "SELECT queue_id, name, state, max_concurrent, schedule FROM queues ORDER BY name");
+        "SELECT queue_id, name, state, max_concurrent, schedule, on_complete FROM queues "
+        "ORDER BY name");
    if (!st) return std::unexpected(st.error());

    std::vector<proto::Queue> out;
@@ -58,7 +64,8 @@ DbResult<std::vector<proto::Queue>> Queues::list() {

 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");
+        "SELECT queue_id, name, state, max_concurrent, schedule, on_complete 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();
@@ -78,4 +85,48 @@ DbResult<bool> Queues::set_state(std::string_view queue_id, std::string_view sta
    return sqlite3_changes(db_.raw()) > 0;
 }

+DbResult<proto::Queue> Queues::upsert(proto::Queue queue) {
+    if (queue.queueId.empty()) {
+        std::random_device rd;
+        std::uniform_int_distribution<std::uint64_t> d;
+        char buf[17];
+        std::snprintf(buf, sizeof(buf), "%016llx", static_cast<unsigned long long>(d(rd)));
+        queue.queueId = std::string(buf);
+    }
+
+    // A create defaults to 'stopped' (never auto-runs a brand-new queue); a replace keeps
+    // whatever run state the queue is already in — queue.upsert edits the config, not the
+    // run state (that's queue.start/stop).
+    std::string state = "stopped";
+    if (auto existing = get(queue.queueId); existing && existing->has_value())
+        state = std::string(proto::to_string((*existing)->state));
+
+    const std::string schedule_json =
+        queue.schedule ? nlohmann::json(*queue.schedule).dump() : std::string();
+    const std::string on_complete =
+        std::string(proto::to_string(queue.onComplete.value_or(proto::QueueOnComplete::Nothing)));
+
+    auto st = db_.prepare(
+        "INSERT INTO queues(queue_id, name, state, max_concurrent, schedule, on_complete) "
+        "VALUES(?1,?2,?3,?4,?5,?6) "
+        "ON CONFLICT(queue_id) DO UPDATE SET "
+        "name=excluded.name, max_concurrent=excluded.max_concurrent, "
+        "schedule=excluded.schedule, on_complete=excluded.on_complete");
+    if (!st) return std::unexpected(st.error());
+    if (auto b = st->bind(1, std::string_view(queue.queueId)); !b) return std::unexpected(b.error());
+    if (auto b = st->bind(2, std::string_view(queue.name)); !b) return std::unexpected(b.error());
+    if (auto b = st->bind(3, std::string_view(state)); !b) return std::unexpected(b.error());
+    if (auto b = st->bind(4, queue.maxConcurrent); !b) return std::unexpected(b.error());
+    if (auto r = queue.schedule ? st->bind(5, std::string_view(schedule_json)) : st->bind_null(5); !r)
+        return std::unexpected(r.error());
+    if (auto b = st->bind(6, std::string_view(on_complete)); !b) return std::unexpected(b.error());
+    if (auto r = st->step(); !r) return std::unexpected(r.error());
+
+    auto stored = get(queue.queueId);
+    if (!stored) return std::unexpected(stored.error());
+    if (!stored->has_value())
+        return std::unexpected(DbError{0, "queue.upsert: row vanished after insert"});
+    return **stored;
+}
+
 }  // namespace velox::daemon::store
@@ -27,6 +27,12 @@ public:
    // id doesn't exist.
    DbResult<bool> set_state(std::string_view queue_id, std::string_view state);

+    // "Omit queueId to create" (queue.upsert's own words) — an empty id generates one.
+    // taskIds is ignored (membership changes only through download.update / queue.reorder,
+    // per the schema's own note); a create defaults to 'stopped', a replace keeps the
+    // queue's current run state (queue.upsert edits config, not run state).
+    DbResult<velox::proto::Queue> upsert(velox::proto::Queue queue);
+
 private:
    Db& db_;
 };
@@ -1,6 +1,8 @@
 // store/categories + store/queues: the two D3 handlers GUI's category panel and queue
 // view need against a real daemon.

+#include <algorithm>
+#include <optional>
 #include <string>

 #include "check.hpp"
@@ -73,6 +75,105 @@ void run() {
            CHECK_EQ(ids[2], std::string("t0"));
        }
    }
+
+    // --- Categories::upsert: create generates an id, builtin is never settable ------
+    {
+        Categories categories(*db);
+
+        velox::proto::Category in;
+        in.name = "ISOs";
+        in.saveDir = "/tmp/isos";
+        in.extensions = {"iso"};
+        in.builtin = true;  // ignored on create: a client cannot mint a builtin category
+        auto created = categories.upsert(in);
+        CHECK(created.has_value());
+        if (created) {
+            CHECK(!created->categoryId.empty());
+            CHECK(!created->builtin);
+
+            // A replace keeps builtin=false too, and can rename/re-point.
+            velox::proto::Category patch = *created;
+            patch.name = "ISO Images";
+            patch.builtin = true;  // still ignored
+            auto replaced = categories.upsert(patch);
+            CHECK(replaced.has_value());
+            if (replaced) {
+                CHECK_EQ(replaced->categoryId, created->categoryId);
+                CHECK_EQ(replaced->name, std::string("ISO Images"));
+                CHECK(!replaced->builtin);
+            }
+
+            // A builtin category is untouched by remove(), and its tasks are not
+            // reassigned away from it — the store enforces this even without the
+            // dispatcher's own -32602 pre-check.
+            auto builtin_attempt = categories.remove("general", std::nullopt);
+            CHECK(builtin_attempt.has_value());
+            if (builtin_attempt) CHECK(!builtin_attempt->removed);
+
+            // remove() reassigns member tasks (default target: "general") and deletes
+            // the row.
+            Tasks tasks(*db);
+            TaskRow r;
+            r.task_id = "cat-owner";
+            r.url = "https://example.com/x";
+            r.save_dir = "/tmp";
+            r.filename = "x";
+            r.created_at = "2026-09-11T00:00:00Z";
+            r.category_id = created->categoryId;
+            CHECK(tasks.insert(r).has_value());
+
+            auto removed = categories.remove(created->categoryId, std::nullopt);
+            CHECK(removed.has_value());
+            if (removed) {
+                CHECK(removed->removed);
+                CHECK_EQ(removed->reassigned_task_ids.size(), std::size_t{1});
+                CHECK_EQ(removed->reassigned_task_ids[0], std::string("cat-owner"));
+            }
+            auto owner = tasks.get("cat-owner");
+            CHECK(owner.has_value() && owner->has_value());
+            if (owner && *owner)
+                CHECK_EQ((*owner)->category_id.value_or(""), std::string("general"));
+
+            // Gone: a second remove() finds nothing.
+            auto gone = categories.remove(created->categoryId, std::nullopt);
+            CHECK(gone.has_value());
+            if (gone) CHECK(!gone->removed);
+
+            tasks.remove("cat-owner");
+        }
+    }
+
+    // --- Queues::upsert: create generates an id; replace keeps the run state --------
+    {
+        Queues queues(*db);
+
+        velox::proto::Queue in;
+        in.name = "Nightly";
+        in.state = velox::proto::QueueState::Running;  // ignored on create: always 'stopped'
+        in.maxConcurrent = 3;
+        auto created = queues.upsert(in);
+        CHECK(created.has_value());
+        if (created) {
+            CHECK(!created->queueId.empty());
+            CHECK(created->state == velox::proto::QueueState::Stopped);
+
+            CHECK(queues.set_state(created->queueId, "running").has_value());
+
+            velox::proto::Queue patch = *created;
+            patch.name = "Nightly Batch";
+            patch.maxConcurrent = 5;
+            patch.state = velox::proto::QueueState::Stopped;  // ignored on replace too
+            auto replaced = queues.upsert(patch);
+            CHECK(replaced.has_value());
+            if (replaced) {
+                CHECK_EQ(replaced->name, std::string("Nightly Batch"));
+                CHECK_EQ(replaced->maxConcurrent, std::int64_t{5});
+                // Run state survived the config edit — still 'running' from set_state above,
+                // not reset by the payload's (ignored) 'stopped'.
+                CHECK(replaced->state == velox::proto::QueueState::Running);
+            }
+        }
+    }
 }

 TEST_MAIN()