daemon: wire the engine into veloxd — the vertical slice runs end to end

CORE stage 8 merged, so vdm::Engine is linkable. This closes D4a and
narrows D4b: `velox add <url>` now actually downloads.

- sched/engine_port_core.hpp — the real EnginePort: forwards to a live
  vdm::Engine, keeps the DownloadHandle per task for pause/resume/
  cancel/provide_auth/decide/refresh_url, drives set_task_order /
  set_max_active_segments / set_host_segment_cap via
  engine.segment_budget(). CORE confirmed the admission model: DAEMON
  decides when to start(); the engine's own download_task calls
  register_task/set_want internally — DAEMON never touches per-task
  budget calls. EnginePort gains release(TaskId) so the port drops a
  handle when the task goes terminal.
- rpc/event_loop — EventLoop::post(fn): thread-safe, runs fn on the
  loop thread next iteration. The marshaller for engine-thread
  callbacks.
- main.cpp — constructs vdm::Engine + EnginePortCore + Scheduler
  (post_to_loop = loop.post). At startup: reconcile_after_restart()
  (ADR 0013 §5), reload_config(), tick(). A 1 s timerfd on the loop
  re-runs tick() (schedule windows, missed nudges); download.add nudges
  via dispatcher.set_on_mutation.

End-to-end verified against tools/testserver: `velox add
http://127.0.0.1:.../file/512K` -> task queued -> scheduler admits ->
engine downloads 524288 bytes -> complete, file on disk. First
byte-path all the way through the project.

safepath-adversarial.md: re-verified per its own note — CORE landed
O_NOFOLLOW on the target open (core/src/io/sparse_file.cpp), so the
leaf-symlink TOCTOU is now closed; residual is down to one
intermediate-dir gap (documented post-M1 chase).

36 daemon/cli tests green; scheduler + uds_roundtrip TSan-clean.

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

daemon/docs/deferrals.md

@@ -8,6 +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 |
-| D4a | `sched/scheduler` is built and unit-tested against `FakeEnginePort`, but there is no real `EnginePortCore` (wraps `vdm::Engine` + `SegmentBudget`) and it is not wired into `veloxd` | `sched/` | CORE stage 8 is on `lane/core`, not yet in `main` (`core/src/task/` is still `.gitkeep` there) — linking `vdm::Engine` would be an unresolved symbol | `lane/core` merges to `main`: add `engine_port_core.{hpp,cpp}` (~100 lines) + construct `Engine`/`Scheduler` in `main.cpp`, run `tick()` on the timer thread and on RPC-driven changes |
-| D4b | no timer thread driving `Scheduler::tick()`; `download.add`/`pause`/`resume`/queue handlers don't nudge the scheduler | `daemon/src/main.cpp`, `rpc/dispatcher.cpp` | depends on D4a | with D4a |
+| ~~D4a~~ | **Closed** — `sched/engine_port_core.hpp` wraps `vdm::Engine` + `segment_budget()`; `main.cpp` constructs `Engine` + `Scheduler`, calls `reconcile_after_restart` / `reload_config` / `tick` at startup | — | — | done (`lane/core` stage 8 merged) |
+| D4b | timer + nudges: a 1 s `timerfd` re-runs `Scheduler::tick()` and `download.add` nudges via `on_mutation`. `download.pause`/`resume`/`start`/`cancel` and the queue.* handlers still don't touch the scheduler | `rpc/dispatcher.cpp` | those handlers are still stubs (D3) | as each handler is implemented behind the store, it calls `on_mutation` / drives the scheduler |
 | D5 | `event.*` fan-out not implemented; `session.subscribe` accepts and echoes but nothing is emitted | `rpc/uds_server.cpp`, `rpc/ws_server.cpp` | No task state to broadcast until the engine is wired. `Scheduler::on_engine_state` is the hook it will fire from | With D4a — the same engine-state callback feeds both the store and `event.task.state` |

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
-   the leaf: a symlink planted *after* this `fstatat` and *before* CORE opens the file is
-   still followed. Closing it needs CORE to open with `O_NOFOLLOW` (plus `O_EXCL` on a
-   fresh download). **Verified 2026-09-10: it does not yet** —
-   `core/src/io/sparse_file.cpp:77` is `O_WRONLY | O_CREAT | O_CLOEXEC`. The flag change
-   has been raised with CORE; until it lands this race is open, see the residual below.
+   is already a symlink. This narrows the create-after-check race on the leaf; it is fully
+   closed by CORE opening the download target with `O_NOFOLLOW`. **Verified 2026-09-11:
+   `core/src/io/sparse_file.cpp` opens `O_WRONLY | O_CREAT | O_CLOEXEC | O_NOFOLLOW`** — a
+   symlink swapped in as the leaf after our check fails there with `ELOOP` ->
+   `Error::path_rejected`. (No `O_EXCL`: resume must be able to open an existing
+   `.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
-   A16 requires us to allow.
-2. **The leaf** swapped for a symlink between our `fstatat` (step 5) and CORE's `open`.
-
-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.
+   prefix; `O_NOFOLLOW` on the *file* open does not re-check the *directories* above it,
+   and a full `O_NOFOLLOW` directory chase would reject the legitimate symlinked
+   directories A16 requires us to allow.
 
 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

@@ -37,6 +37,9 @@ public:
     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;

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

@@ -24,6 +24,7 @@ public:
     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;
@@ -40,6 +41,7 @@ public:
     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 {

daemon/src/sched/scheduler.cpp

@@ -307,7 +307,10 @@ void Scheduler::on_engine_state(const std::string& wire_id, std::string_view eng
     }
 
     if (engine_state == "complete" || engine_state == "failed" || engine_state == "cancelled") {
-        if (auto eid = engine_id_of(wire_id)) unmap_engine(*eid);
+        if (auto eid = engine_id_of(wire_id)) {
+            engine_.release(*eid);
+            unmap_engine(*eid);
+        }
     }
 }