merge: engine wired into veloxd — vertical slice closes

daemon/CMakeLists.txt

@@ -59,12 +59,13 @@ target_compile_options(veloxd_fs PRIVATE -Wall -Wextra -Wpedantic -Werror)
 add_library(veloxd_sched STATIC
     src/sched/schedule_window.cpp
     src/sched/governor.cpp
+    src/sched/scheduler.cpp
 )
 add_library(velox::daemon_sched ALIAS veloxd_sched)
 target_include_directories(veloxd_sched PUBLIC ${CMAKE_CURRENT_SOURCE_DIR}/src)
 target_compile_features(veloxd_sched PUBLIC cxx_std_23)
 target_compile_options(veloxd_sched PRIVATE -Wall -Wextra -Wpedantic -Werror)
-target_link_libraries(veloxd_sched PUBLIC velox::proto nlohmann_json::nlohmann_json)
+target_link_libraries(veloxd_sched PUBLIC velox::proto velox::core veloxd_store nlohmann_json::nlohmann_json)
 
 # --- veloxd_rpc — the RPC transports + dispatcher ------------------------------------
 add_library(veloxd_rpc STATIC

daemon/docs/deferrals.md

@@ -8,5 +8,6 @@ close. Kept here (not buried in commit messages) so the next pass can see them a
 | D1 | Pairing prompt is `EnvAutoApprover` (needs `VELOX_PAIR_AUTO=1`) | `rpc/pairing.hpp`, `main.cpp` | A GUI dialog / `org.freedesktop.Notifications` approver is integration work | Build step 7 (systemd + notifications) |
 | D2 | `download.probe` → `-32603` | `rpc/dispatcher.cpp` | `download.add` is wired (`fs/safepath` + store, real `-32011`); `download.probe` needs the engine's probe path for `-32013` | probe with the engine link (CORE stage 3 is landed; wire `Engine::probe`) |
 | D3 | Stub handlers for everything except `session.*`, `download.add/list/get` | `rpc/dispatcher.cpp` | No store behind them yet (categories/queues/rules/settings/limiter/schedule) | Per method, as the store query modules land behind them |
-| 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()` |
+| ~~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) |
-| 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 |
+| D4b | timer + nudges: a 1 s `timerfd` re-runs `Scheduler::tick()` and `download.add` nudges via `on_mutation`. `download.pause`/`resume`/`start`/`cancel` and the queue.* handlers still don't touch the scheduler | `rpc/dispatcher.cpp` | those handlers are still stubs (D3) | as each handler is implemented behind the store, it calls `on_mutation` / drives the scheduler |
+| D5 | `event.*` fan-out not implemented; `session.subscribe` accepts and echoes but nothing is emitted | `rpc/uds_server.cpp`, `rpc/ws_server.cpp` | No task state to broadcast until the engine is wired. `Scheduler::on_engine_state` is the hook it will fire from | With D4a — the same engine-state callback feeds both the store and `event.task.state` |

daemon/docs/safepath-adversarial.md

@@ -58,32 +58,27 @@ Roots for the examples: `allowedRoots = ["/home/u/Downloads", "/data/dl"]`, alre
    `-32011`. Then re-derive the final dir's path from its fd (`/proc/self/fd/N`) and
    re-assert containment. (A8 for the created tail, A14)
 5. **Best-effort leaf check:** `fstatat(dir_fd, leaf, AT_SYMLINK_NOFOLLOW)` — refuse if it
-   is already a symlink. This narrows, but does not close, the create-after-check race on
+   is already a symlink. This narrows the create-after-check race on the leaf; it is fully
-   the leaf: a symlink planted *after* this `fstatat` and *before* CORE opens the file is
+   closed by CORE opening the download target with `O_NOFOLLOW`. **Verified 2026-09-11:
-   still followed. Closing it needs CORE to open with `O_NOFOLLOW` (plus `O_EXCL` on a
+   `core/src/io/sparse_file.cpp` opens `O_WRONLY | O_CREAT | O_CLOEXEC | O_NOFOLLOW`** — a
-   fresh download). **Verified 2026-09-10: it does not yet** —
+   symlink swapped in as the leaf after our check fails there with `ELOOP` ->
-   `core/src/io/sparse_file.cpp:77` is `O_WRONLY | O_CREAT | O_CLOEXEC`. The flag change
+   `Error::path_rejected`. (No `O_EXCL`: resume must be able to open an existing
-   has been raised with CORE; until it lands this race is open, see the residual below.
+   `.veloxpart`.)
 6. **Every failure is `-32011`, `data.path` = the *original* `saveDir`** — never the
    resolved path, which would leak where the roots actually live. The one exception is a
    `filename` that violates the schema's own `maxLength`, which is `-32602` at the param
    layer before this code runs.
 
-### Residual — currently OPEN, tracked
+### Residual — one gap, narrowed
 
-Two TOCTOU gaps this code does not close on its own:
+**The leaf-symlink TOCTOU is closed** (step 5, verified 2026-09-11: CORE opens the target
+`O_NOFOLLOW`). What remains:
 
 1. **An existing intermediate directory** swapped for an out-of-root symlink between our
    `realpath` (step 3) and the write. Step 3 trusts `realpath` for the pre-existing
-   prefix; a full `O_NOFOLLOW` chase would reject the legitimate symlinked directories
+   prefix; `O_NOFOLLOW` on the *file* open does not re-check the *directories* above it,
-   A16 requires us to allow.
+   and a full `O_NOFOLLOW` directory chase would reject the legitimate symlinked
-2. **The leaf** swapped for a symlink between our `fstatat` (step 5) and CORE's `open`.
+   directories A16 requires us to allow.
-
-Both are closed by CORE opening the file `O_NOFOLLOW` (and, for a fresh download,
-`O_EXCL`). **As verified on 2026-09-10 that is not yet the case** —
-`core/src/io/sparse_file.cpp:77` opens `O_WRONLY | O_CREAT | O_CLOEXEC`. The flag change has
-been raised with CORE; when it lands, update step 5 and this paragraph and re-verify the
-flags at that line.
 
 What limits the exposure *today*: the download directory lives under `~/.local/share` /
 `~/Downloads`, both `0700` — an attacker planting a symlink there already has write access

daemon/src/main.cpp

@@ -15,14 +15,20 @@
 #include <sys/un.h>
 #include <unistd.h>
 
+#include <sys/timerfd.h>
+
 #include "rpc/dispatcher.hpp"
 #include "rpc/event_loop.hpp"
 #include "rpc/pairing.hpp"
 #include "rpc/runtime_dir.hpp"
 #include "rpc/uds_server.hpp"
 #include "rpc/ws_server.hpp"
+#include "sched/engine_port_core.hpp"
+#include "sched/governor.hpp"
+#include "sched/scheduler.hpp"
 #include "store/migrations.hpp"
 #include "store/sqlite.hpp"
+#include "vdm/engine.hpp"
 #include "version.hpp"
 
 namespace {
@@ -101,7 +107,38 @@ int main() {
         return 1;
     }
 
+    // --- engine + scheduler ---------------------------------------------------------
+    vdm::Engine engine;
+    velox::daemon::sched::EnginePortCore engine_port(engine);
+    velox::daemon::sched::Scheduler scheduler(
+        *db, engine_port, velox::daemon::sched::Governor{},
+        {/*local_now*/ {},
+         /*post_to_loop*/ [&loop](std::function<void()> fn) { loop.post(std::move(fn)); }});
+
+    if (const auto ec = scheduler.reconcile_after_restart(); !ec)
+        std::cerr << "veloxd: restart reconcile: " << ec.error().to_string() << "\n";
+    (void)scheduler.reload_config();
+    (void)scheduler.tick();  // admit anything already queued in the DB
+
     velox::daemon::rpc::VeloxDispatcher dispatcher(*db);
+    dispatcher.set_on_mutation([&loop, &scheduler] {
+        loop.post([&scheduler] { (void)scheduler.tick(); });
+    });
+
+    // A 1 s timer re-runs the scheduler so schedule windows opening/closing and any
+    // missed nudge are picked up. Registered on the loop, no extra thread.
+    const int tick_fd = ::timerfd_create(CLOCK_MONOTONIC, TFD_NONBLOCK | TFD_CLOEXEC);
+    if (tick_fd >= 0) {
+        itimerspec spec{};
+        spec.it_value.tv_sec = 1;
+        spec.it_interval.tv_sec = 1;
+        ::timerfd_settime(tick_fd, 0, &spec, nullptr);
+        loop.add_fd(tick_fd, velox::daemon::rpc::kRead, [&](int fd, unsigned) {
+            std::uint64_t ticks = 0;
+            [[maybe_unused]] ssize_t n = ::read(fd, &ticks, sizeof(ticks));
+            (void)scheduler.tick();
+        });
+    }
 
     velox::daemon::rpc::UdsServer uds(loop, dispatcher, rt.socket_path());
     if (const auto ec = uds.start()) {
@@ -128,6 +165,10 @@ int main() {
     loop.run();
     std::cout << "veloxd: shutting down\n";
 
+    if (tick_fd >= 0) {
+        loop.del_fd(tick_fd);
+        ::close(tick_fd);
+    }
     g_loop = nullptr;
     ::close(lock_fd);
     return 0;

daemon/src/rpc/dispatcher.cpp

@@ -151,7 +151,7 @@ VeloxDispatcher::on_download_add(const proto::DownloadSpec& spec) {
     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).
+    // scheduler (`queued`); on_mutation_ nudges it.
     row.state = row.start_mode == "manual" ? "new" : "queued";
     row.category_id = spec.categoryId;
     row.queue_id = spec.queueId;
@@ -168,6 +168,8 @@ VeloxDispatcher::on_download_add(const proto::DownloadSpec& spec) {
         return std::unexpected(proto::HandlerError{proto::ErrorCode::InternalError,
                                                    "download.add: " + ins.error().message});
 
+    if (on_mutation_) on_mutation_();
+
     proto::DownloadAddResult r;
     r.taskId = row.task_id;
     if (auto st = proto::parse_TaskState(row.state)) r.state = *st;

daemon/src/rpc/dispatcher.hpp

@@ -12,6 +12,8 @@
 // "not implemented in this build" (-> -32603) until its handler and the scheduler land;
 // see daemon/docs/deferrals.md.
 
+#include <functional>
+
 #include "store/sqlite.hpp"
 #include "velox_proto.hpp"
 
@@ -21,6 +23,10 @@ class VeloxDispatcher final : public velox::proto::Dispatcher {
 public:
     explicit VeloxDispatcher(velox::daemon::store::Db& db) : db_(db) {}
 
+    // Called after a handler mutates task state (download.add for now). main.cpp wires it
+    // to nudge the scheduler; unset in tests.
+    void set_on_mutation(std::function<void()> fn) { on_mutation_ = std::move(fn); }
+
     velox::proto::HandlerResult<velox::proto::CaptureRules>
     on_capture_getRules(const velox::proto::CaptureGetRulesParams&) override;
     velox::proto::HandlerResult<velox::proto::CaptureOfferResult>
@@ -101,6 +107,7 @@ public:
 
 private:
     velox::daemon::store::Db& db_;
+    std::function<void()> on_mutation_;
 };
 
 }  // namespace velox::daemon::rpc

daemon/src/rpc/event_loop.cpp

@@ -45,6 +45,23 @@ void EventLoop::stop() noexcept {
     wake();
 }
 
+void EventLoop::post(std::function<void()> fn) {
+    {
+        std::lock_guard<std::mutex> lk(post_mu_);
+        posts_.push_back(std::move(fn));
+    }
+    wake();
+}
+
+void EventLoop::drain_posts() {
+    std::vector<std::function<void()>> batch;
+    {
+        std::lock_guard<std::mutex> lk(post_mu_);
+        batch.swap(posts_);
+    }
+    for (auto& fn : batch) fn();
+}
+
 void EventLoop::drain_wakeup() noexcept {
     std::uint64_t sink = 0;
     while (::read(wake_fd_, &sink, sizeof(sink)) > 0) {
@@ -87,6 +104,8 @@ void EventLoop::run() {
             if (p.revents != 0) fired.push_back(p.fd);
         }
 
+        drain_posts();
+
         for (const int fd : fired) {
             const auto it = fds_.find(fd);
             if (it == fds_.end()) continue;  // removed by an earlier callback this pass

daemon/src/rpc/event_loop.hpp

@@ -12,7 +12,9 @@
 #include <atomic>
 #include <cstdint>
 #include <functional>
+#include <mutex>
 #include <unordered_map>
+#include <vector>
 
 namespace velox::daemon::rpc {
 
@@ -52,6 +54,10 @@ public:
     // callback. Async-signal-safe.
     void wake() noexcept;
 
+    // Run `fn` on the loop thread at the next iteration. Thread-safe; the intended way to
+    // marshal an engine-thread callback back onto the RPC loop.
+    void post(std::function<void()> fn);
+
 private:
     struct Entry {
         unsigned interest;
@@ -59,11 +65,15 @@ private:
     };
 
     void drain_wakeup() noexcept;
+    void drain_posts();
 
     int wake_fd_;                      // eventfd, always registered
     bool running_ = false;
     std::atomic<bool> stop_requested_ = false;   // set from stop(), read by run()
     std::unordered_map<int, Entry> fds_;
+
+    std::mutex post_mu_;
+    std::vector<std::function<void()>> posts_;
 };
 
 }  // namespace velox::daemon::rpc

daemon/src/sched/engine_port.hpp

@@ -0,0 +1,49 @@
+#pragma once
+
+// The seam between the Scheduler and CORE's engine. Everything the Scheduler drives on the
+// engine goes through this interface, so the Scheduler is unit-testable without a live
+// engine and the daemon is not bound to the concrete `vdm::Engine`. The real
+// implementation (engine_port_core, added when velox::core's stage-8 bodies reach main)
+// wraps `vdm::Engine` + `vdm::segment::SegmentBudget`; FakeEnginePort records calls.
+//
+// Task ids here are `vdm::TaskId` — the engine assigns one from start() and the Scheduler
+// keeps the wire-UUID <-> TaskId map (ADR 0013). Admission is the Scheduler's: it calls
+// start() only for a task the governor admitted, and the engine begins probing at once
+// (it does not queue). The min-1 fairness rule in SegmentBudget then guarantees each
+// started task a slot; set_task_order pushes the priority.
+
+#include <cstdint>
+#include <functional>
+#include <string>
+#include <vector>
+
+#include "vdm/ids.hpp"
+#include "vdm/task/download.hpp"
+
+namespace velox::daemon::sched {
+
+class EnginePort {
+public:
+    virtual ~EnginePort() = default;
+
+    virtual vdm::TaskId start(const vdm::task::DownloadSpec& spec,
+                              vdm::task::DownloadCallbacks callbacks) = 0;
+
+    virtual void pause(vdm::TaskId) = 0;
+    virtual void resume(vdm::TaskId) = 0;
+    virtual void cancel(vdm::TaskId, bool discard_partial) = 0;
+    virtual void provide_auth(vdm::TaskId, const std::string& username,
+                              const std::string& password, bool remember) = 0;
+    virtual void decide(vdm::TaskId, vdm::task::Decision) = 0;
+    virtual void refresh_url(vdm::TaskId, const std::string& url) = 0;
+
+    // The daemon is done with this task (it went terminal). Drop the handle. Idempotent.
+    virtual void release(vdm::TaskId) = 0;
+
+    // ADR 0011 admission surface. Values are DAEMON's; enforcement is the engine's.
+    virtual void set_task_order(const std::vector<vdm::TaskId>& order) = 0;
+    virtual void set_max_active_segments(std::uint32_t n) = 0;
+    virtual void set_host_segment_cap(const std::string& host, std::uint32_t cap) = 0;
+};
+
+}  // namespace velox::daemon::sched

daemon/src/sched/engine_port_core.hpp

@@ -0,0 +1,68 @@
+#pragma once
+
+// The real EnginePort: forwards to a live vdm::Engine and keeps the DownloadHandle per
+// task so pause/resume/cancel/… have something to call. All methods run on the RPC loop
+// thread (the Scheduler's thread); the handles map is only touched there.
+
+#include <unordered_map>
+
+#include "sched/engine_port.hpp"
+#include "vdm/engine.hpp"
+#include "vdm/task/download.hpp"
+
+namespace velox::daemon::sched {
+
+class EnginePortCore final : public EnginePort {
+public:
+    explicit EnginePortCore(vdm::Engine& engine) : engine_(engine) {}
+
+    vdm::TaskId start(const vdm::task::DownloadSpec& spec,
+                      vdm::task::DownloadCallbacks callbacks) override {
+        vdm::task::DownloadHandle h = engine_.start(spec, std::move(callbacks));
+        const vdm::TaskId id = h.id();
+        handles_.insert_or_assign(id, std::move(h));
+        return id;
+    }
+
+    void pause(vdm::TaskId id) override {
+        if (auto* h = find(id)) h->pause();
+    }
+    void resume(vdm::TaskId id) override {
+        if (auto* h = find(id)) h->resume();
+    }
+    void cancel(vdm::TaskId id, bool discard_partial) override {
+        if (auto* h = find(id)) h->cancel(discard_partial);
+    }
+    void provide_auth(vdm::TaskId id, const std::string& u, const std::string& p,
+                      bool remember) override {
+        if (auto* h = find(id)) h->provide_auth(u, p, remember);
+    }
+    void decide(vdm::TaskId id, vdm::task::Decision d) override {
+        if (auto* h = find(id)) h->decide(d);
+    }
+    void refresh_url(vdm::TaskId id, const std::string& url) override {
+        if (auto* h = find(id)) h->refresh_url(url);
+    }
+    void release(vdm::TaskId id) override { handles_.erase(id); }
+
+    void set_task_order(const std::vector<vdm::TaskId>& order) override {
+        engine_.segment_budget().set_task_order(order);
+    }
+    void set_max_active_segments(std::uint32_t n) override {
+        engine_.segment_budget().set_max_active_segments(n);
+    }
+    void set_host_segment_cap(const std::string& host, std::uint32_t cap) override {
+        engine_.segment_budget().set_host_segment_cap(host, cap);
+    }
+
+private:
+    vdm::task::DownloadHandle* find(vdm::TaskId id) {
+        auto it = handles_.find(id);
+        return it == handles_.end() ? nullptr : &it->second;
+    }
+
+    vdm::Engine& engine_;
+    std::unordered_map<vdm::TaskId, vdm::task::DownloadHandle> handles_;
+};
+
+}  // namespace velox::daemon::sched

daemon/src/sched/fake_engine_port.hpp

@@ -0,0 +1,57 @@
+#pragma once
+
+// A recording EnginePort for Scheduler tests. Every call is logged; start() hands back a
+// sequential TaskId. No threads, no real work.
+
+#include <cstdint>
+#include <string>
+#include <vector>
+
+#include "sched/engine_port.hpp"
+
+namespace velox::daemon::sched {
+
+class FakeEnginePort final : public EnginePort {
+public:
+    struct StartCall {
+        vdm::TaskId id;
+        std::string url;
+        std::string save_path;
+        vdm::task::DownloadCallbacks callbacks;
+    };
+
+    std::vector<StartCall> starts;
+    std::vector<vdm::TaskId> paused;
+    std::vector<vdm::TaskId> resumed;
+    std::vector<std::pair<vdm::TaskId, bool>> cancelled;
+    std::vector<vdm::TaskId> released;
+    std::vector<std::vector<vdm::TaskId>> orders;
+    std::vector<std::uint32_t> max_active_segments;
+    std::vector<std::pair<std::string, std::uint32_t>> host_caps;
+
+    vdm::TaskId start(const vdm::task::DownloadSpec& spec,
+                      vdm::task::DownloadCallbacks callbacks) override {
+        const vdm::TaskId id{next_++};
+        starts.push_back({id, spec.url, spec.save_path, std::move(callbacks)});
+        return id;
+    }
+    void pause(vdm::TaskId id) override { paused.push_back(id); }
+    void resume(vdm::TaskId id) override { resumed.push_back(id); }
+    void cancel(vdm::TaskId id, bool discard) override { cancelled.emplace_back(id, discard); }
+    void provide_auth(vdm::TaskId, const std::string&, const std::string&, bool) override {}
+    void decide(vdm::TaskId, vdm::task::Decision) override {}
+    void refresh_url(vdm::TaskId, const std::string&) override {}
+    void release(vdm::TaskId id) override { released.push_back(id); }
+    void set_task_order(const std::vector<vdm::TaskId>& order) override { orders.push_back(order); }
+    void set_max_active_segments(std::uint32_t n) override { max_active_segments.push_back(n); }
+    void set_host_segment_cap(const std::string& h, std::uint32_t c) override {
+        host_caps.emplace_back(h, c);
+    }
+
+    const std::vector<vdm::TaskId>& last_order() const { return orders.back(); }
+
+private:
+    std::uint64_t next_ = 1;
+};
+
+}  // namespace velox::daemon::sched

daemon/src/sched/scheduler.cpp

@@ -0,0 +1,317 @@
+#include "sched/scheduler.hpp"
+
+#include <algorithm>
+#include <cctype>
+
+#include <nlohmann/json.hpp>
+
+#include "sched/schedule_window.hpp"
+#include "store/settings.hpp"
+#include "store/tasks.hpp"
+
+namespace velox::daemon::sched {
+
+namespace proto = velox::proto;
+
+namespace {
+
+std::tm local_now_default() {
+    const std::time_t t = std::time(nullptr);
+    std::tm tm{};
+    ::localtime_r(&t, &tm);
+    return tm;
+}
+
+// host[:port] out of a URL, lowercased. "" when it cannot be parsed (opts the task out of
+// the per-host cap rather than lumping unrelated tasks under "").
+std::string host_of(std::string_view url) {
+    auto scheme = url.find("://");
+    std::string_view rest = scheme == std::string_view::npos ? url : url.substr(scheme + 3);
+    const auto at = rest.find('@');
+    if (at != std::string_view::npos) rest = rest.substr(at + 1);
+    const auto end = rest.find_first_of("/:?#");
+    std::string h(end == std::string_view::npos ? rest : rest.substr(0, end));
+    std::transform(h.begin(), h.end(), h.begin(),
+                   [](unsigned char c) { return static_cast<char>(std::tolower(c)); });
+    return h;
+}
+
+// ISO-8601 timestamp -> a monotonic integer for FIFO tiebreaking ("2026-09-10T15:57:50Z"
+// -> 20260910155750). Lexical order of the digits is chronological.
+std::int64_t rank_of(std::string_view iso) {
+    std::string digits;
+    for (char c : iso)
+        if (std::isdigit(static_cast<unsigned char>(c))) digits.push_back(c);
+    digits.resize(std::min<std::size_t>(digits.size(), 17));  // fits in int64
+    return digits.empty() ? 0 : std::stoll(digits);
+}
+
+RunState run_state_of(const std::string& s) {
+    if (s == "queued") return RunState::Queued;
+    if (s == "paused") return RunState::Paused;
+    if (s == "complete" || s == "failed" || s == "cancelled") return RunState::Terminal;
+    return RunState::Running;  // probing / connecting / downloading / retry_wait / assembling / verifying
+}
+
+std::optional<PauseReason> pause_reason_of(const std::optional<std::string>& s) {
+    if (!s) return std::nullopt;
+    if (*s == "user") return PauseReason::User;
+    if (*s == "schedule") return PauseReason::Schedule;
+    if (*s == "queue_stopped") return PauseReason::QueueStopped;
+    if (*s == "admission_reconcile") return PauseReason::AdmissionReconcile;
+    if (*s == "auto") return PauseReason::Auto;
+    return std::nullopt;
+}
+
+const char* pause_reason_str(PauseReason r) {
+    switch (r) {
+    case PauseReason::User: return "user";
+    case PauseReason::Schedule: return "schedule";
+    case PauseReason::QueueStopped: return "queue_stopped";
+    case PauseReason::AdmissionReconcile: return "admission_reconcile";
+    case PauseReason::Auto: return "auto";
+    }
+    return "user";
+}
+
+const char* engine_state_name(vdm::task::EngineState s) {
+    using E = vdm::task::EngineState;
+    switch (s) {
+    case E::probing: return "probing";
+    case E::connecting: return "connecting";
+    case E::downloading: return "downloading";
+    case E::paused: return "paused";
+    case E::retry_wait: return "retry_wait";
+    case E::assembling: return "assembling";
+    case E::verifying: return "verifying";
+    case E::complete: return "complete";
+    case E::failed: return "failed";
+    case E::cancelled: return "cancelled";
+    }
+    return "connecting";
+}
+
+// The non-terminal states the scheduler cares about — feeds the tasks.list filter.
+std::vector<proto::TaskState> non_terminal_states() {
+    return {proto::TaskState::New,        proto::TaskState::Probing,
+            proto::TaskState::Queued,     proto::TaskState::Connecting,
+            proto::TaskState::Downloading, proto::TaskState::Paused,
+            proto::TaskState::RetryWait,  proto::TaskState::Assembling,
+            proto::TaskState::Verifying};
+}
+
+}  // namespace
+
+Scheduler::Scheduler(store::Db& db, EnginePort& engine, Governor governor, Deps deps)
+    : db_(db), engine_(engine), governor_(std::move(governor)), deps_(std::move(deps)) {
+    if (!deps_.local_now) deps_.local_now = local_now_default;
+    if (!deps_.post_to_loop) deps_.post_to_loop = [](std::function<void()> f) { f(); };
+}
+
+void Scheduler::map(const std::string& wire_id, vdm::TaskId engine_id) {
+    to_engine_[wire_id] = engine_id;
+    to_wire_[engine_id] = wire_id;
+}
+void Scheduler::unmap_engine(vdm::TaskId engine_id) {
+    if (auto it = to_wire_.find(engine_id); it != to_wire_.end()) {
+        to_engine_.erase(it->second);
+        to_wire_.erase(it);
+    }
+}
+std::optional<std::string> Scheduler::wire_id_of(vdm::TaskId id) const {
+    auto it = to_wire_.find(id);
+    return it == to_wire_.end() ? std::nullopt : std::optional<std::string>(it->second);
+}
+std::optional<vdm::TaskId> Scheduler::engine_id_of(const std::string& wire_id) const {
+    auto it = to_engine_.find(wire_id);
+    return it == to_engine_.end() ? std::nullopt : std::optional<vdm::TaskId>(it->second);
+}
+
+store::DbResult<void> Scheduler::reconcile_after_restart() {
+    // Any CORE-owned state (probing..verifying) becomes `queued`; the engine knows nothing
+    // across a restart and will re-probe / re-resume from the .veloxpart.meta sidecar when
+    // the scheduler starts it again (ADR 0013 §5). `paused` and `new` are left alone.
+    return db_.exec(
+        "UPDATE tasks SET state = 'queued', pause_reason = NULL "
+        "WHERE state IN ('probing','connecting','downloading','retry_wait','assembling','verifying')");
+}
+
+store::DbResult<void> Scheduler::reload_config() {
+    store::Settings settings(db_);
+    GovernorConfig cfg;
+    cfg.max_concurrent_downloads = settings.get_int("connection.maxConcurrentDownloads");
+    cfg.max_active_segments = settings.get_int("connection.maxActiveSegments");
+
+    // Per-host caps: a JSON object {host: n} under a daemon-local key. Absent => none.
+    if (auto raw = settings.get_raw("saveTo.hostSegmentCaps"); raw && *raw) {
+        auto j = nlohmann::json::parse(**raw, nullptr, false);
+        if (j.is_object()) {
+            for (const auto& [host, n] : j.items()) {
+                if (n.is_number_integer()) {
+                    const auto cap = n.get<std::int64_t>();
+                    cfg.host_caps[host] = cap;
+                    engine_.set_host_segment_cap(host, static_cast<std::uint32_t>(std::max<std::int64_t>(cap, 0)));
+                }
+            }
+        }
+    }
+    governor_.set_config(cfg);
+    engine_.set_max_active_segments(
+        static_cast<std::uint32_t>(std::max<std::int64_t>(cfg.max_active_segments, 1)));
+    return {};
+}
+
+store::DbResult<void> Scheduler::tick() {
+    // --- snapshot: queues -----------------------------------------------------------
+    std::vector<QueueView> queues;
+    {
+        auto st = db_.prepare("SELECT queue_id, state, max_concurrent, schedule FROM queues");
+        if (!st) return std::unexpected(st.error());
+        const std::tm now = deps_.local_now();
+        for (;;) {
+            auto row = st->step();
+            if (!row) return std::unexpected(row.error());
+            if (!*row) break;
+            QueueView q;
+            q.queue_id = st->column_text(0);
+            q.running = st->column_text(1) == "running";
+            q.max_concurrent = st->column_int(2);
+            if (!st->column_is_null(3)) {
+                auto j = nlohmann::json::parse(st->column_text(3), nullptr, false);
+                proto::Schedule sched;
+                if (auto p = proto::parse<proto::Schedule>(j, "schedule")) sched = *p;
+                q.window_open = window_open(sched, now);
+            }
+            queues.push_back(std::move(q));
+        }
+    }
+
+    // --- snapshot: tasks ----------------------------------------------------------
+    store::Tasks tasks(db_);
+    proto::TaskFilter filter;
+    filter.states = non_terminal_states();
+    auto page = tasks.list(filter, std::nullopt, 0, 100000);
+    if (!page) return std::unexpected(page.error());
+
+    std::vector<TaskView> views;
+    views.reserve(page->rows.size());
+    for (const auto& r : page->rows) {
+        if (r.state == "new") continue;  // parked until the user starts it
+        TaskView v;
+        v.task_id = r.task_id;
+        v.run_state = run_state_of(r.state);
+        v.pause_reason = pause_reason_of(r.pause_reason);
+        v.queue_id = r.queue_id;
+        v.queue_position = r.queue_position.value_or(0);
+        v.host = host_of(r.url);
+        v.admit_rank = rank_of(r.created_at);
+        views.push_back(std::move(v));
+    }
+
+    const Decision d = governor_.evaluate(views, queues);
+
+    // --- apply ------------------------------------------------------------------
+    for (const auto& wire_id : d.to_start) {
+        auto got = tasks.get(wire_id);
+        if (!got || !got->has_value()) continue;
+        const store::TaskRow& row = **got;
+
+        vdm::task::DownloadSpec spec;
+        spec.url = row.url;
+        spec.save_path = row.save_dir + "/" + row.filename;
+        if (row.req_segments) spec.segments = static_cast<std::uint32_t>(*row.req_segments);
+        if (row.req_buffer_bytes)
+            spec.buffer_bytes = static_cast<std::uint64_t>(*row.req_buffer_bytes);
+        if (row.checksum_algo && row.checksum_value) {
+            vdm::task::Checksum ck;
+            ck.hex = *row.checksum_value;
+            if (*row.checksum_algo == "md5") ck.algo = vdm::task::Checksum::Algo::md5;
+            else if (*row.checksum_algo == "sha1") ck.algo = vdm::task::Checksum::Algo::sha1;
+            else if (*row.checksum_algo == "sha512") ck.algo = vdm::task::Checksum::Algo::sha512;
+            else ck.algo = vdm::task::Checksum::Algo::sha256;
+            spec.checksum = ck;
+        }
+        spec.allow_resume = true;  // resume from a sidecar if one is beside save_path
+        // headers / cookies / referrer / user_agent are not persisted yet (a URL-only
+        // `velox add` has none); the capture path will fill them when it lands.
+
+        vdm::task::DownloadCallbacks cbs;
+        const std::string id_copy = wire_id;
+        cbs.on_state = [this, id_copy](vdm::task::EngineState, vdm::task::EngineState to,
+                                       const std::optional<vdm::ErrorInfo>& err) {
+            std::optional<TaskErrorFields> ef;
+            if (err) {
+                ef = TaskErrorFields{};
+                ef->code = std::string(vdm::error_name(err->code));  // matches TaskErrorCode
+                ef->message = err->context;
+                if (err->http_status != 0) ef->http_status = err->http_status;
+                ef->retryable = err->retryable;
+            }
+            const std::string to_name = engine_state_name(to);
+            deps_.post_to_loop(
+                [this, id_copy, to_name, ef]() { on_engine_state(id_copy, to_name, ef); });
+        };
+
+        const vdm::TaskId engine_id = engine_.start(spec, std::move(cbs));
+        map(wire_id, engine_id);
+        (void)tasks.set_state(wire_id, "probing", std::nullopt);
+    }
+
+    for (const auto& wire_id : d.to_resume) {
+        if (auto eid = engine_id_of(wire_id)) {
+            engine_.resume(*eid);
+            (void)tasks.set_state(wire_id, "connecting", std::nullopt);
+        }
+    }
+
+    for (const auto& wire_id : d.to_pause) {
+        const auto reason = d.pause_reasons.count(wire_id)
+                                ? pause_reason_str(d.pause_reasons.at(wire_id))
+                                : "user";
+        if (auto eid = engine_id_of(wire_id)) engine_.pause(*eid);
+        (void)tasks.set_state(wire_id, "paused", std::string(reason));
+    }
+
+    std::vector<vdm::TaskId> order;
+    order.reserve(d.priority_order.size());
+    for (const auto& wire_id : d.priority_order)
+        if (auto eid = engine_id_of(wire_id)) order.push_back(*eid);
+    engine_.set_task_order(order);
+
+    return {};
+}
+
+void Scheduler::on_engine_state(const std::string& wire_id, std::string_view engine_state,
+                                const std::optional<TaskErrorFields>& err) {
+    store::Tasks tasks(db_);
+    // pause_reason: an engine-initiated pause carries an error => 'auto' (ADR 0013 §2);
+    // otherwise set_state clears the column.
+    std::optional<std::string> reason;
+    if (engine_state == "paused" && err) reason = "auto";
+    (void)tasks.set_state(wire_id, engine_state, reason);
+
+    if (err) {
+        auto st = db_.prepare(
+            "UPDATE tasks SET error_code=?2, error_message=?3, error_http_status=?4, "
+            "error_retryable=?5 WHERE task_id=?1");
+        if (st) {
+            (void)st->bind(1, std::string_view(wire_id));
+            (void)st->bind(2, std::string_view(err->code));
+            (void)st->bind(3, std::string_view(err->message));
+            if (err->http_status) (void)st->bind(4, *err->http_status);
+            else (void)st->bind_null(4);
+            if (err->retryable) (void)st->bind(5, static_cast<std::int64_t>(*err->retryable));
+            else (void)st->bind_null(5);
+            (void)st->step();
+        }
+    }
+
+    if (engine_state == "complete" || engine_state == "failed" || engine_state == "cancelled") {
+        if (auto eid = engine_id_of(wire_id)) {
+            engine_.release(*eid);
+            unmap_engine(*eid);
+        }
+    }
+}
+
+}  // namespace velox::daemon::sched

daemon/src/sched/scheduler.hpp

@@ -0,0 +1,88 @@
+#pragma once
+
+// The Scheduler ties the pure Governor to the store and the engine. It owns the
+// wire-UUID <-> vdm::TaskId map and is the only thing that calls EnginePort::start /
+// pause / resume / set_task_order.
+//
+// Threading: tick(), reload_config(), reconcile_after_restart() and the on_engine_*
+// callbacks all run on ONE thread (the RPC loop). Engine callbacks arrive on engine
+// threads, so the real wiring passes a `post_to_loop` that marshals them here; the
+// default runs them inline (tests, single-threaded).
+//
+// Not yet wired into veloxd — that plus the real EnginePort land when velox::core's
+// stage-8 bodies reach main (daemon/docs/deferrals.md D4).
+
+#include <cstdint>
+#include <ctime>
+#include <functional>
+#include <optional>
+#include <string>
+#include <unordered_map>
+
+#include "sched/engine_port.hpp"
+#include "sched/governor.hpp"
+#include "store/sqlite.hpp"
+#include "vdm/ids.hpp"
+
+namespace velox::daemon::sched {
+
+// The subset of proto::TaskError the store row carries; passed by on_engine_state so a
+// failed / retry_wait / auto-paused transition lands in the DB.
+struct TaskErrorFields {
+    std::string code;             // TaskErrorCode spelling
+    std::string message;
+    std::optional<std::int64_t> http_status;
+    std::optional<bool> retryable;
+    std::optional<std::int64_t> attempt;
+};
+
+class Scheduler {
+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
+    // thread; the default calls it inline.
+    struct Deps {
+        std::function<std::tm()> local_now;
+        std::function<void(std::function<void()>)> post_to_loop;
+    };
+
+    Scheduler(store::Db& db, EnginePort& engine, Governor governor, Deps deps = {});
+
+    // ADR 0013 §5: on daemon start, every task whose persisted state is a CORE-owned one
+    // (probing..verifying) is rewritten to `queued`; `paused` keeps its pauseReason. The
+    // engine holds no state across a restart.
+    store::DbResult<void> reconcile_after_restart();
+
+    // Re-read connection.maxConcurrentDownloads / maxActiveSegments and the per-host cap
+    // table; push the caps to the engine and update the governor config. Call at startup
+    // and on settings.set of a connection.* key.
+    store::DbResult<void> reload_config();
+
+    // One scheduling pass: snapshot the store, run the governor, apply its decision
+    // (start / resume / pause via the engine, update task state rows, push set_task_order).
+    store::DbResult<void> tick();
+
+    // Engine lifecycle callback -> store projection, so the next tick sees ground truth.
+    // Keyed by wire UUID (known when the callback is built, before start() returns the
+    // TaskId). Also the hook the event.task.state fan-out will use — D5.
+    void on_engine_state(const std::string& wire_id, std::string_view engine_state,
+                         const std::optional<TaskErrorFields>& err);
+
+    // 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;
+
+private:
+    void map(const std::string& wire_id, vdm::TaskId engine_id);
+    void unmap_engine(vdm::TaskId engine_id);
+
+    store::Db& db_;
+    EnginePort& engine_;
+    Governor governor_;
+    Deps deps_;
+
+    std::unordered_map<std::string, vdm::TaskId> to_engine_;
+    std::unordered_map<vdm::TaskId, std::string> to_wire_;
+};
+
+}  // namespace velox::daemon::sched

daemon/tests/CMakeLists.txt

@@ -20,3 +20,4 @@ veloxd_test(sched_window     LIBS veloxd_sched)
 veloxd_test(sched_governor   LIBS veloxd_sched)
 veloxd_test(safepath         LIBS veloxd_fs)
 veloxd_test(store_tasks      LIBS veloxd_store)
+veloxd_test(sched_scheduler  LIBS veloxd_sched)

daemon/tests/sched_scheduler_test.cpp

@@ -0,0 +1,191 @@
+// Scheduler against a FakeEnginePort and an in-memory store: admission, priority order,
+// queue stop, restart reconciliation, and the engine-state -> store projection.
+
+#include <string>
+
+#include "check.hpp"
+#include "sched/fake_engine_port.hpp"
+#include "sched/governor.hpp"
+#include "sched/scheduler.hpp"
+#include "store/migrations.hpp"
+#include "store/settings.hpp"
+#include "store/sqlite.hpp"
+#include "store/tasks.hpp"
+
+using namespace velox::daemon;
+using sched::FakeEnginePort;
+using sched::Governor;
+using sched::GovernorConfig;
+using sched::Scheduler;
+
+namespace {
+
+store::TaskRow task(std::string id, std::string state, std::string created,
+                    std::optional<std::string> queue = std::nullopt, std::int64_t pos = 0) {
+    store::TaskRow r;
+    r.task_id = std::move(id);
+    r.url = "https://cdn.example/" + r.task_id;
+    r.save_dir = "/tmp";
+    r.filename = r.task_id + ".bin";
+    r.state = std::move(state);
+    r.created_at = std::move(created);
+    r.queue_id = std::move(queue);
+    if (r.queue_id) r.queue_position = pos;
+    return r;
+}
+
+std::string task_state(store::Db& db, const std::string& id) {
+    store::Tasks t(db);
+    auto g = t.get(id);
+    return (g && *g) ? (*g)->state : std::string("<none>");
+}
+
+}  // namespace
+
+void run() {
+    auto db = store::Db::open(":memory:");
+    CHECK(db.has_value());
+    if (!db) return;
+    CHECK(store::migrate_to_head(*db).has_value());
+    store::Tasks tasks(*db);
+
+    // --- admission: 4 queued, cap 2 -> start the 2 oldest, in order -----------------
+    {
+        FakeEnginePort engine;
+        Scheduler sched(*db, engine,
+                        Governor(GovernorConfig{.max_concurrent_downloads = 2,
+                                                .max_active_segments = 32}));
+        for (int i = 0; i < 4; ++i)
+            CHECK(tasks.insert(task("a" + std::to_string(i), "queued",
+                                    "2026-09-10T10:0" + std::to_string(i) + ":00Z"))
+                      .has_value());
+
+        CHECK(sched.tick().has_value());
+        CHECK_EQ(engine.starts.size(), 2u);
+        CHECK_EQ(engine.starts[0].url, std::string("https://cdn.example/a0"));
+        CHECK_EQ(engine.starts[1].url, std::string("https://cdn.example/a1"));
+        CHECK_EQ(engine.starts[0].save_path, std::string("/tmp/a0.bin"));
+        CHECK_EQ(task_state(*db, "a0"), std::string("probing"));
+        CHECK_EQ(task_state(*db, "a2"), std::string("queued"));  // not admitted
+
+        // set_task_order carries exactly the started tasks, oldest first.
+        CHECK_EQ(engine.last_order().size(), 2u);
+        CHECK(engine.last_order()[0] == engine.starts[0].id);
+
+        // A second tick with no free slots starts nothing new.
+        CHECK(sched.tick().has_value());
+        CHECK_EQ(engine.starts.size(), 2u);
+    }
+
+    // --- engine reports downloading, then one completes -> a slot frees ------------
+    {
+        for (const char* id : {"a0", "a1", "a2", "a3"}) tasks.remove(id);
+        FakeEnginePort engine;
+        Scheduler sched(*db, engine,
+                        Governor(GovernorConfig{.max_concurrent_downloads = 1,
+                                                .max_active_segments = 32}));
+        CHECK(tasks.insert(task("b0", "queued", "2026-09-10T10:00:00Z")).has_value());
+        CHECK(tasks.insert(task("b1", "queued", "2026-09-10T10:01:00Z")).has_value());
+
+        CHECK(sched.tick().has_value());
+        CHECK_EQ(engine.starts.size(), 1u);  // b0 only
+        CHECK_EQ(task_state(*db, "b0"), std::string("probing"));
+
+        sched.on_engine_state("b0", "downloading", std::nullopt);
+        CHECK_EQ(task_state(*db, "b0"), std::string("downloading"));
+
+        sched.on_engine_state("b0", "complete", std::nullopt);
+        CHECK_EQ(task_state(*db, "b0"), std::string("complete"));
+
+        CHECK(sched.tick().has_value());  // b0 terminal -> b1 admitted
+        CHECK_EQ(engine.starts.size(), 2u);
+        CHECK_EQ(engine.starts[1].url, std::string("https://cdn.example/b1"));
+    }
+
+    // --- a stopped queue: running tasks get paused with reason queue_stopped -------
+    {
+        for (const char* id : {"b0", "b1"}) tasks.remove(id);
+        CHECK(db->exec("UPDATE queues SET state='running' WHERE queue_id='main'").has_value());
+        FakeEnginePort engine;
+        Scheduler sched(*db, engine,
+                        Governor(GovernorConfig{.max_concurrent_downloads = 10,
+                                                .max_active_segments = 32}));
+        CHECK(tasks.insert(task("q0", "queued", "2026-09-10T10:00:00Z", "main", 0)).has_value());
+        CHECK(sched.tick().has_value());
+        CHECK_EQ(engine.starts.size(), 1u);
+        sched.on_engine_state("q0", "downloading", std::nullopt);
+
+        CHECK(db->exec("UPDATE queues SET state='stopped' WHERE queue_id='main'").has_value());
+        CHECK(sched.tick().has_value());
+        CHECK_EQ(engine.paused.size(), 1u);
+        CHECK_EQ(task_state(*db, "q0"), std::string("paused"));
+        store::Tasks t(*db);
+        CHECK_EQ(t.get("q0").value().value().pause_reason.value_or(""), std::string("queue_stopped"));
+
+        // Restart the queue -> the task resumes (not a fresh start).
+        CHECK(db->exec("UPDATE queues SET state='running' WHERE queue_id='main'").has_value());
+        CHECK(sched.tick().has_value());
+        CHECK_EQ(engine.resumed.size(), 1u);
+        CHECK_EQ(engine.starts.size(), 1u);  // no new start
+    }
+
+    // --- an engine auto-pause (error present) -> pause_reason 'auto', not touched --
+    {
+        for (const char* id : {"q0"}) tasks.remove(id);
+        FakeEnginePort engine;
+        Scheduler sched(*db, engine,
+                        Governor(GovernorConfig{.max_concurrent_downloads = 10,
+                                                .max_active_segments = 32}));
+        CHECK(tasks.insert(task("auth", "queued", "2026-09-10T10:00:00Z")).has_value());
+        CHECK(sched.tick().has_value());
+
+        sched::TaskErrorFields ef;
+        ef.code = "auth_required";
+        ef.message = "401";
+        ef.http_status = 401;
+        sched.on_engine_state("auth", "paused", ef);
+
+        store::Tasks t(*db);
+        auto row = t.get("auth").value().value();
+        CHECK_EQ(row.state, std::string("paused"));
+        CHECK_EQ(row.pause_reason.value_or(""), std::string("auto"));
+        CHECK_EQ(row.error_code.value_or(""), std::string("auth_required"));
+
+        // A tick must NOT resume an auto-paused task.
+        engine.resumed.clear();
+        CHECK(sched.tick().has_value());
+        CHECK_EQ(engine.resumed.size(), 0u);
+    }
+
+    // --- reconcile_after_restart: CORE-owned states -> queued --------------------
+    {
+        for (const char* id : {"auth"}) tasks.remove(id);
+        CHECK(tasks.insert(task("r0", "downloading", "2026-09-10T10:00:00Z")).has_value());
+        CHECK(tasks.insert(task("r1", "verifying", "2026-09-10T10:01:00Z")).has_value());
+        CHECK(tasks.insert(task("r2", "paused", "2026-09-10T10:02:00Z")).has_value());
+        CHECK(db->exec("UPDATE tasks SET pause_reason='user' WHERE task_id='r2'").has_value());
+
+        FakeEnginePort engine;
+        Scheduler sched(*db, engine, Governor(GovernorConfig{}));
+        CHECK(sched.reconcile_after_restart().has_value());
+        CHECK_EQ(task_state(*db, "r0"), std::string("queued"));
+        CHECK_EQ(task_state(*db, "r1"), std::string("queued"));
+        CHECK_EQ(task_state(*db, "r2"), std::string("paused"));  // paused survives
+        store::Tasks t(*db);
+        CHECK_EQ(t.get("r2").value().value().pause_reason.value_or(""), std::string("user"));
+    }
+
+    // --- reload_config pushes the caps to the engine ---------------------------
+    {
+        store::Settings settings(*db);
+        CHECK(settings.set_raw("connection.maxActiveSegments", "12").has_value());
+        FakeEnginePort engine;
+        Scheduler sched(*db, engine, Governor(GovernorConfig{}));
+        CHECK(sched.reload_config().has_value());
+        CHECK_EQ(engine.max_active_segments.size(), 1u);
+        CHECK_EQ(engine.max_active_segments.back(), 12u);
+        CHECK_EQ(sched.reload_config().has_value() ? 0 : 1, 0);
+    }
+}
+
+TEST_MAIN()