From 3db0d01f9d76658ed0bb2fe331eb9871401974a9 Mon Sep 17 00:00:00 2001 From: sami Date: Thu, 10 Sep 2026 19:58:45 +0400 Subject: [PATCH] daemon: download.add / download.list / download.get behind the store MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit VeloxDispatcher now takes a store::Db& and three handlers are real: - download.list -> store::Tasks::list (filter / sort / paging in SQL) -> to_summary per row. No more empty-table stub. - download.add -> resolve saveDir (spec, else saveTo.defaultDir; ~ expanded) and the leaf (spec.filename, else the URL's last segment percent-decoded, else download.bin) -> fs::resolve_target against canonicalize_root'd saveTo.allowedRoots. Any path-destination failure is -32011 with the *original* saveDir in data.path. On success a TaskRow is inserted in state `queued` (or `new` for startMode "manual") and {taskId, state} returned. The scheduler that would then admit it is D4. - download.get -> store::Tasks::get; a real -32010 + data.taskId for an unknown id, else a TaskDetail (segmentDetail empty until the engine segments the task, which the schema permits). util/time.hpp: now_iso() factored out of ws_server.cpp. main.cpp constructs the dispatcher with the opened db. The three integration tests build an in-memory migrated db for it; velox.client now drives the full slice through the CLI — add outside roots -> -32011 with data.path, add into an allowed root -> a task that download.list shows and download.get details, unknown id -> -32010. Verified with the real binaries: velox add persists, velox ls shows it, it survives a daemon restart, /etc is refused. ASan+UBSan and TSan clean; 34 daemon/cli tests green. deferrals.md: D2 down to just download.probe; D3 down to categories/queues/rules/ settings/limiter/schedule. Co-Authored-By: Claude Sonnet 5 Claude-Session: https://claude.ai/code/session_01Upd9WhG9oppieig5nRDLig --- cli/tests/client_test.cpp | 62 ++++++++++- daemon/docs/deferrals.md | 4 +- daemon/src/main.cpp | 2 +- daemon/src/rpc/dispatcher.cpp | 164 ++++++++++++++++++++++++++-- daemon/src/rpc/dispatcher.hpp | 21 ++-- daemon/src/rpc/ws_server.cpp | 10 +- daemon/src/util/time.hpp | 21 ++++ daemon/tests/CMakeLists.txt | 4 +- daemon/tests/uds_roundtrip_test.cpp | 9 +- daemon/tests/ws_server_test.cpp | 2 +- 10 files changed, 257 insertions(+), 42 deletions(-) create mode 100644 daemon/src/util/time.hpp diff --git a/cli/tests/client_test.cpp b/cli/tests/client_test.cpp index 0f1f1a7..4819abb 100644 --- a/cli/tests/client_test.cpp +++ b/cli/tests/client_test.cpp @@ -12,8 +12,13 @@ #include "rpc/dispatcher.hpp" #include "rpc/event_loop.hpp" #include "rpc/uds_server.hpp" +#include "store/migrations.hpp" +#include "store/settings.hpp" +#include "store/sqlite.hpp" namespace rpc = velox::daemon::rpc; + +static std::string g_allowed_root; using velox::cli::CallError; using velox::cli::Client; @@ -39,7 +44,22 @@ void run() { const std::string server_sock = velox_dir + "/velox.sock"; rpc::EventLoop loop; - rpc::VeloxDispatcher dispatcher; + auto db = velox::daemon::store::Db::open(":memory:"); + CHECK(db.has_value()); + if (!db) return; + CHECK(velox::daemon::store::migrate_to_head(*db).has_value()); + + char root_tmpl[] = "/tmp/velox-cli-root-XXXXXX"; + g_allowed_root = ::mkdtemp(root_tmpl); + CHECK(!g_allowed_root.empty()); + { + velox::daemon::store::Settings settings(*db); + CHECK(settings.set_raw("saveTo.allowedRoots", + "[\"" + g_allowed_root + "\"]").has_value()); + CHECK(settings.set_raw("saveTo.defaultDir", + "\"" + g_allowed_root + "\"").has_value()); + } + rpc::VeloxDispatcher dispatcher(*db); rpc::UdsServer server(loop, dispatcher, server_sock); const auto ec = server.start(); CHECK(!ec); @@ -64,10 +84,42 @@ void run() { CHECK(ls->at("items").is_array()); } - // A not-yet-implemented method surfaces the daemon's error, not a transport error. - auto add = c.call("download.add", {{"url", "https://example.com/x"}}); - CHECK(!add.has_value()); - if (!add) CHECK_EQ(add.error().code, -32603); + // download.add outside every allowed root -> -32011, original saveDir echoed. + auto bad = c.call("download.add", + {{"url", "https://example.com/x"}, {"saveDir", "/etc"}}); + CHECK(!bad.has_value()); + if (!bad) { + CHECK_EQ(bad.error().code, -32011); + CHECK_EQ(bad.error().data.value("path", ""), std::string("/etc")); + } + + // download.add into an allowed root -> a task id that then shows up in the list. + auto ok = c.call( + "download.add", + {{"url", "https://example.com/movie.mp4"}, {"saveDir", g_allowed_root}}); + CHECK(ok.has_value()); + std::string task_id; + if (ok) { + task_id = ok->value("taskId", ""); + CHECK(!task_id.empty()); + CHECK_EQ(ok->value("state", ""), std::string("queued")); + } + + auto ls2 = c.call("download.list", nlohmann::json::object()); + CHECK(ls2.has_value()); + if (ls2) { + CHECK_EQ(ls2->value("total", -1), 1); + CHECK_EQ(ls2->at("items").at(0).value("filename", ""), std::string("movie.mp4")); + } + + auto detail = c.call("download.get", {{"taskId", task_id}}); + CHECK(detail.has_value()); + if (detail) CHECK_EQ(detail->at("summary").value("taskId", ""), task_id); + + auto missing = + c.call("download.get", {{"taskId", "00000000-0000-4000-8000-000000000000"}}); + CHECK(!missing.has_value()); + if (!missing) CHECK_EQ(missing.error().code, -32010); } loop.stop(); diff --git a/daemon/docs/deferrals.md b/daemon/docs/deferrals.md index ebeb73b..8bf668b 100644 --- a/daemon/docs/deferrals.md +++ b/daemon/docs/deferrals.md @@ -6,7 +6,7 @@ close. Kept here (not buried in commit messages) so the next pass can see them a | # | What | Where | Why deferred | Closes when | |---|---|---|---|---| | 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.add` → `-32603`, `download.probe` → `-32603` | `rpc/dispatcher.cpp` | The path boundary (`-32011`) is built and tested (`fs/safepath`, `daemon/docs/safepath-adversarial.md`); still need it wired into the `download.add` handler with the store behind it, plus the probe path (`-32013`) which needs the engine | `download.add` glue (dispatcher ↔ store ↔ `fs/safepath`); probe with the engine link | -| D3 | Stub handlers for everything except `session.*`, `download.list`, `download.get` | `rpc/dispatcher.cpp` | No store behind them yet | Per method, as the store/scheduler wire in | +| 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 | | D4 | `sched/` is the pure `Governor` + schedule window only; no `Scheduler` wiring to store/engine/timer | `sched/` | `Engine` bodies land in CORE stage 8; `Scheduler` needs the UUID↔`vdm::TaskId` map, a store query layer, and a timer | After CORE stage 8 lands `Engine::start()` | | 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 | With the callback → `event.*` projection | diff --git a/daemon/src/main.cpp b/daemon/src/main.cpp index f8e5d11..55a6c8e 100644 --- a/daemon/src/main.cpp +++ b/daemon/src/main.cpp @@ -101,7 +101,7 @@ int main() { return 1; } - velox::daemon::rpc::VeloxDispatcher dispatcher; + velox::daemon::rpc::VeloxDispatcher dispatcher(*db); velox::daemon::rpc::UdsServer uds(loop, dispatcher, rt.socket_path()); if (const auto ec = uds.start()) { diff --git a/daemon/src/rpc/dispatcher.cpp b/daemon/src/rpc/dispatcher.cpp index 39b76f4..bc9657b 100644 --- a/daemon/src/rpc/dispatcher.cpp +++ b/daemon/src/rpc/dispatcher.cpp @@ -1,5 +1,15 @@ #include "rpc/dispatcher.hpp" +#include +#include + +#include + +#include "fs/safepath.hpp" +#include "store/settings.hpp" +#include "store/tasks.hpp" +#include "util/time.hpp" + namespace velox::daemon::rpc { namespace proto = velox::proto; @@ -15,6 +25,56 @@ proto::HandlerResult not_implemented(const char* method) { std::string("not implemented in this build: ") + method}); } +// A v4 UUID for a new task id. +std::string new_task_id() { + std::random_device rd; + std::uniform_int_distribution d; + std::uint32_t a = d(rd), b = d(rd), c = d(rd), e = d(rd); + b = (b & 0xFFFF0FFFu) | 0x00004000u; + c = (c & 0x3FFFFFFFu) | 0x80000000u; + char buf[37]; + std::snprintf(buf, sizeof(buf), "%08x-%04x-%04x-%04x-%04x%08x", a, (b >> 16), (b & 0xFFFF), + (c >> 16), (c & 0xFFFF), e); + return std::string(buf); +} + +// Expand a leading "~" against $HOME. Configured dirs may be stored that way. +std::string expand_tilde(std::string p) { + if (p == "~" || p.rfind("~/", 0) == 0) { + if (const char* home = std::getenv("HOME"); home != nullptr && home[0] != '\0') + p = std::string(home) + (p.size() > 1 ? p.substr(1) : std::string{}); + } + return p; +} + +// Last path segment of a URL, percent-decoded, as a filename fallback when the caller gave +// none and there is no probe yet. Empty => the handler uses "download.bin". +std::string filename_from_url(std::string_view url) { + auto q = url.find_first_of("?#"); + if (q != std::string_view::npos) url = url.substr(0, q); + auto slash = url.find_last_of('/'); + std::string_view leaf = slash == std::string_view::npos ? url : url.substr(slash + 1); + std::string out; + for (std::size_t i = 0; i < leaf.size(); ++i) { + if (leaf[i] == '%' && i + 2 < leaf.size()) { + auto hex = [](char ch) -> int { + if (ch >= '0' && ch <= '9') return ch - '0'; + if (ch >= 'a' && ch <= 'f') return ch - 'a' + 10; + if (ch >= 'A' && ch <= 'F') return ch - 'A' + 10; + return -1; + }; + int hi = hex(leaf[i + 1]), lo = hex(leaf[i + 2]); + if (hi >= 0 && lo >= 0) { + out.push_back(static_cast((hi << 4) | lo)); + i += 2; + continue; + } + } + out.push_back(leaf[i]); + } + return out; +} + } // namespace // --- session.* : handled in the server layer, unreachable here in the running daemon --- @@ -36,12 +96,81 @@ VeloxDispatcher::on_session_subscribe(const proto::SessionSubscribeParams&) { return not_implemented("session.subscribe"); } -// --- download.list : an empty table, so a client can connect and render --------------- +// --- download.list : the main table; the store does the filter / sort / page ---------- proto::HandlerResult -VeloxDispatcher::on_download_list(const proto::DownloadListParams&) { +VeloxDispatcher::on_download_list(const proto::DownloadListParams& params) { + store::Tasks tasks(db_); + auto page = tasks.list(params.filter, params.sort, params.offset.value_or(0), + params.limit.value_or(0)); + if (!page) + return std::unexpected(proto::HandlerError{proto::ErrorCode::InternalError, + "download.list: " + page.error().message}); + proto::DownloadListResult r; - r.total = 0; + r.total = page->total; + r.items.reserve(page->rows.size()); + for (const auto& row : page->rows) r.items.push_back(store::to_summary(row)); + return r; +} + +// --- download.add : canonicalise + root-check the destination, then persist ----------- + +proto::HandlerResult +VeloxDispatcher::on_download_add(const proto::DownloadSpec& spec) { + store::Settings settings(db_); + + std::string save_dir = spec.saveDir && !spec.saveDir->empty() + ? *spec.saveDir + : settings.get_string("saveTo.defaultDir"); + save_dir = expand_tilde(std::move(save_dir)); + + std::string leaf = spec.filename && !spec.filename->empty() ? *spec.filename + : filename_from_url(spec.url); + if (leaf.empty()) leaf = "download.bin"; + + std::vector roots; + for (const auto& r : settings.get_string_array("saveTo.allowedRoots")) { + if (auto c = fs::canonicalize_root(r)) roots.push_back(*c); + } + + auto target = fs::resolve_target(save_dir, leaf, roots); + if (!target) { + // Everything path-destination-related is -32011 with the *original* saveDir in + // data.path (never the resolved path — daemon/docs/safepath-adversarial.md rule 6). + return std::unexpected(proto::HandlerError{ + proto::ErrorCode::InvalidPath, target.error().message, + nlohmann::json{{"path", spec.saveDir.value_or(save_dir)}}}); + } + + store::TaskRow row; + row.task_id = new_task_id(); + row.url = spec.url; + row.save_dir = target->dir; + row.filename = target->leaf; + row.created_at = velox::daemon::now_iso(); + row.start_mode = spec.startMode ? std::string(proto::to_string(*spec.startMode)) : "auto"; + // startMode 'manual' parks the task in `new`; anything else makes it eligible for the + // scheduler (`queued`). The scheduler itself is not wired yet (deferrals.md D4). + row.state = row.start_mode == "manual" ? "new" : "queued"; + row.category_id = spec.categoryId; + row.queue_id = spec.queueId; + row.description = spec.description; + row.req_segments = spec.segments; + row.req_buffer_bytes = spec.bufferBytes; + if (spec.checksum) { + row.checksum_algo = std::string(proto::to_string(spec.checksum->algorithm)); + row.checksum_value = spec.checksum->value; + } + + store::Tasks tasks(db_); + if (auto ins = tasks.insert(row); !ins) + return std::unexpected(proto::HandlerError{proto::ErrorCode::InternalError, + "download.add: " + ins.error().message}); + + proto::DownloadAddResult r; + r.taskId = row.task_id; + if (auto st = proto::parse_TaskState(row.state)) r.state = *st; return r; } @@ -67,10 +196,6 @@ proto::HandlerResult VeloxDispatcher::on_category_upsert(const proto::CategoryUpsertParams&) { return not_implemented("category.upsert"); } -proto::HandlerResult -VeloxDispatcher::on_download_add(const proto::DownloadSpec&) { - return not_implemented("download.add"); -} proto::HandlerResult VeloxDispatcher::on_download_addBatch(const proto::DownloadAddBatchParams&) { return not_implemented("download.addBatch"); @@ -81,11 +206,26 @@ VeloxDispatcher::on_download_cancel(const proto::DownloadCancelParams&) { } proto::HandlerResult VeloxDispatcher::on_download_get(const proto::DownloadGetParams& params) { - // No store is wired yet, so no task exists and every id is genuinely not-found. This - // is the real -32010 answer (contracts/ error fixture download.get.not-found), not a - // placeholder; it becomes a store lookup when store/ is wired in. - return std::unexpected(proto::HandlerError{proto::ErrorCode::TaskNotFound, "no such task", - nlohmann::json{{"taskId", params.taskId}}}); + store::Tasks tasks(db_); + auto got = tasks.get(params.taskId); + if (!got) + return std::unexpected(proto::HandlerError{proto::ErrorCode::InternalError, + "download.get: " + got.error().message}); + if (!got->has_value()) + return std::unexpected(proto::HandlerError{proto::ErrorCode::TaskNotFound, "no such task", + nlohmann::json{{"taskId", params.taskId}}}); + + const store::TaskRow& row = **got; + proto::TaskDetail d; + d.summary = store::to_summary(row); + // segmentDetail stays empty until the engine has segmented the task — the schema + // permits that ("empty before the task has been segmented"). + d.mime = row.content_type; + d.bufferBytes = row.req_buffer_bytes; + d.effectiveBufferBytes = row.eff_buffer_bytes; + if (row.state != "complete" && row.state != "cancelled") + d.partPath = row.save_dir + "/" + row.filename + ".veloxpart"; + return d; } proto::HandlerResult VeloxDispatcher::on_download_pause(const proto::DownloadPauseParams&) { diff --git a/daemon/src/rpc/dispatcher.hpp b/daemon/src/rpc/dispatcher.hpp index 1e21e70..04a0f52 100644 --- a/daemon/src/rpc/dispatcher.hpp +++ b/daemon/src/rpc/dispatcher.hpp @@ -5,22 +5,22 @@ // parse; a method here only ever sees a validated, typed params struct and returns a // typed result. // -// Scope of this drop (AGENT-DAEMON.md build order): the transport is real, the store is -// not. session.hello / session.pair / session.subscribe are handled in the server layer -// (they are connection- and transport-stateful) and never reach this class in the running -// daemon. download.list answers with an empty table so a client can connect and render. -// Every other method returns "not implemented in this build" — which the generated -// dispatch() surfaces as -32603 — until the store and scheduler land. -// -// The -32603 collapse for genuine in-handler errors (-32010 / -32011 / -32013) is a known -// codegen gap, filed as P1 in daemon/docs/proto-requests-m1.md. Not worked around here. +// Scope of this drop (AGENT-DAEMON.md build order): the transports are real and the store +// is behind download.add / download.list / download.get. session.hello / session.pair / +// session.subscribe are handled in the server layer (connection- and transport-stateful) +// and never reach this class in the running daemon. Everything else still returns +// "not implemented in this build" (-> -32603) until its handler and the scheduler land; +// see daemon/docs/deferrals.md. +#include "store/sqlite.hpp" #include "velox_proto.hpp" namespace velox::daemon::rpc { class VeloxDispatcher final : public velox::proto::Dispatcher { public: + explicit VeloxDispatcher(velox::daemon::store::Db& db) : db_(db) {} + velox::proto::HandlerResult on_capture_getRules(const velox::proto::CaptureGetRulesParams&) override; velox::proto::HandlerResult @@ -98,6 +98,9 @@ public: on_settings_get(const velox::proto::SettingsGetParams&) override; velox::proto::HandlerResult on_settings_set(const velox::proto::SettingsSetParams&) override; + +private: + velox::daemon::store::Db& db_; }; } // namespace velox::daemon::rpc diff --git a/daemon/src/rpc/ws_server.cpp b/daemon/src/rpc/ws_server.cpp index d6c701c..36ca5b1 100644 --- a/daemon/src/rpc/ws_server.cpp +++ b/daemon/src/rpc/ws_server.cpp @@ -16,6 +16,7 @@ #include "rpc/event_loop.hpp" #include "rpc/ws_handshake.hpp" +#include "util/time.hpp" #include "store/pairings.hpp" #include "store/sqlite.hpp" #include "version.hpp" @@ -32,15 +33,6 @@ std::error_code errc(int e) { return std::error_code(e, std::generic_category()) constexpr std::size_t kMaxOutBytes = 16 * 1024 * 1024; constexpr std::size_t kMaxHandshakeBytes = 16 * 1024; -std::string now_iso() { - std::time_t t = std::time(nullptr); - std::tm tm{}; - ::gmtime_r(&t, &tm); - char buf[32]; - std::strftime(buf, sizeof(buf), "%Y-%m-%dT%H:%M:%SZ", &tm); - return std::string(buf); -} - std::string uuid4() { std::random_device rd; std::uniform_int_distribution d; diff --git a/daemon/src/util/time.hpp b/daemon/src/util/time.hpp new file mode 100644 index 0000000..d0182e4 --- /dev/null +++ b/daemon/src/util/time.hpp @@ -0,0 +1,21 @@ +#pragma once + +// RFC 3339 UTC timestamp, second precision ("2026-09-10T14:55:02Z"). The wire uses this +// spelling verbatim for createdAt / lastTryAt / completedAt, so a stored value projects +// with no reformatting. + +#include +#include + +namespace velox::daemon { + +inline std::string now_iso() { + const std::time_t t = std::time(nullptr); + std::tm tm{}; + ::gmtime_r(&t, &tm); + char buf[32]; + std::strftime(buf, sizeof(buf), "%Y-%m-%dT%H:%M:%SZ", &tm); + return std::string(buf); +} + +} // namespace velox::daemon diff --git a/daemon/tests/CMakeLists.txt b/daemon/tests/CMakeLists.txt index 5c14b75..736e92e 100644 --- a/daemon/tests/CMakeLists.txt +++ b/daemon/tests/CMakeLists.txt @@ -11,11 +11,11 @@ function(veloxd_test name) endfunction() veloxd_test(ndjson LIBS veloxd_rpc) -veloxd_test(uds_roundtrip LIBS veloxd_rpc) +veloxd_test(uds_roundtrip LIBS veloxd_rpc veloxd_store) veloxd_test(store_migrations LIBS veloxd_store) veloxd_test(pairings LIBS veloxd_store veloxd_rpc) veloxd_test(ws_frame LIBS veloxd_rpc) -veloxd_test(ws_server LIBS veloxd_rpc) +veloxd_test(ws_server LIBS veloxd_rpc veloxd_store) veloxd_test(sched_window LIBS veloxd_sched) veloxd_test(sched_governor LIBS veloxd_sched) veloxd_test(safepath LIBS veloxd_fs) diff --git a/daemon/tests/uds_roundtrip_test.cpp b/daemon/tests/uds_roundtrip_test.cpp index a221f9a..b707591 100644 --- a/daemon/tests/uds_roundtrip_test.cpp +++ b/daemon/tests/uds_roundtrip_test.cpp @@ -17,6 +17,8 @@ #include "rpc/event_loop.hpp" #include "rpc/ndjson.hpp" #include "rpc/uds_server.hpp" +#include "store/migrations.hpp" +#include "store/sqlite.hpp" #include "velox_proto.hpp" using nlohmann::json; @@ -63,8 +65,13 @@ json call(int fd, const json& request) { void run() { const std::string sock = make_temp_socket_path(); + auto db = velox::daemon::store::Db::open(":memory:"); + CHECK(db.has_value()); + if (!db) return; + CHECK(velox::daemon::store::migrate_to_head(*db).has_value()); + rpc::EventLoop loop; - rpc::VeloxDispatcher dispatcher; + rpc::VeloxDispatcher dispatcher(*db); rpc::UdsServer server(loop, dispatcher, sock); const auto ec = server.start(); CHECK(!ec); diff --git a/daemon/tests/ws_server_test.cpp b/daemon/tests/ws_server_test.cpp index e241012..22ed527 100644 --- a/daemon/tests/ws_server_test.cpp +++ b/daemon/tests/ws_server_test.cpp @@ -123,7 +123,7 @@ void run() { rpc::RuntimeDir rt{dir ? dir : "/tmp"}; rpc::EventLoop loop; - rpc::VeloxDispatcher dispatcher; + rpc::VeloxDispatcher dispatcher(*db); rpc::EnvAutoApprover approver; rpc::WsServer server(loop, dispatcher, *db, approver, rt); const auto ec = server.start();