merge: stage 8 engine, O_NOFOLLOW fix, auth handshake

core/CMakeLists.txt

@@ -6,6 +6,7 @@
 
 find_package(Threads REQUIRED)
 find_package(CURL 8.0 REQUIRED)
+find_package(OpenSSL REQUIRED)
 
 add_library(veloxcore STATIC
     src/util/error.cpp
@@ -22,6 +23,9 @@ add_library(veloxcore STATIC
     src/meta/veloxpart.cpp
     src/segment/segmenter.cpp
     src/segment/budget.cpp
+    src/task/digest.cpp
+    src/task/download_task.cpp
+    src/engine.cpp
 )
 add_library(velox::core ALIAS veloxcore)
 
@@ -38,7 +42,7 @@ target_compile_options(veloxcore PRIVATE
     -Wall -Wextra -Wpedantic -Werror
 )
 
-target_link_libraries(veloxcore PUBLIC Threads::Threads CURL::libcurl)
+target_link_libraries(veloxcore PUBLIC Threads::Threads CURL::libcurl PRIVATE OpenSSL::Crypto)
 
 # Later stages add: find_package(OpenSSL) for meta/ (streaming SHA-256 + resume CRC).
 

core/include/vdm/net/probe.hpp

@@ -35,6 +35,7 @@ struct ProbeRequest {
     std::string user_agent;
     std::string referrer;
     ProxyConfig proxy;
+    AuthConfig auth;  // credentials for a re-probe after a 401 (leave scheme == none otherwise)
 
     long connect_timeout_ms = 15000;
     long overall_timeout_ms = 25000;  // download.probe deadline is 30 s

core/include/vdm/segment/segmenter.hpp

@@ -131,6 +131,10 @@ class Segmenter {
     [[nodiscard]] SegState segment_state(std::uint32_t idx) const noexcept;
     void set_segment_state(std::uint32_t idx, SegState s) noexcept;
 
+    // Hand a segment back to the pool without touching its `completed`: it becomes an
+    // unassigned idle range that the next assign_slot() picks up (used on pause).
+    void release_segment(std::uint32_t idx) noexcept;
+
     // Sum of bytes done across every segment (active + already complete). Locks.
     [[nodiscard]] std::uint64_t downloaded() const;
     [[nodiscard]] bool all_complete() const;

core/include/vdm/task/download.hpp

@@ -31,6 +31,11 @@ class Engine;  // owns and fills DownloadHandle (see vdm/engine.hpp)
 
 namespace vdm::task {
 
+// The task control block. Opaque: defined only in the engine's translation unit. A handle
+// holds a shared_ptr to one; the engine keeps its own copy so the task outlives a caller
+// that drops its handle.
+struct DownloadTaskState;
+
 // --- input ---------------------------------------------------------------------------
 
 struct Checksum {
@@ -177,6 +182,9 @@ struct DownloadCallbacks {
 class DownloadHandle {
   public:
     DownloadHandle() = default;
+    // The engine builds handles; `DownloadTaskState` is incomplete everywhere else, so
+    // this is effectively engine-only without a friend declaration.
+    explicit DownloadHandle(std::shared_ptr<DownloadTaskState> s) : state_(std::move(s)) {}
 
     [[nodiscard]] TaskId id() const noexcept;
     [[nodiscard]] bool valid() const noexcept { return static_cast<bool>(state_); }
@@ -207,10 +215,7 @@ class DownloadHandle {
     [[nodiscard]] Progress progress() const;
 
   private:
-    friend class vdm::Engine;
-    struct State;
-    explicit DownloadHandle(std::shared_ptr<State> s) : state_(std::move(s)) {}
-    std::shared_ptr<State> state_;
+    std::shared_ptr<DownloadTaskState> state_;
 };
 
 }  // namespace vdm::task

core/src/engine.cpp

@@ -0,0 +1,169 @@
+// vdm/engine.cpp
+
+#include "vdm/engine.hpp"
+
+#include <atomic>
+#include <condition_variable>
+#include <cstdint>
+#include <map>
+#include <mutex>
+#include <thread>
+#include <unordered_map>
+
+#include "task/download_task.hpp"
+
+namespace vdm {
+
+struct Engine::Impl : task::TaskHost {
+    explicit Impl(Config c)
+        : cfg_(c),
+          http_(net::HttpClient::Options{.workers = c.http_workers}),
+          prober_(c.probe_pool_size ? c.probe_pool_size : 4),
+          budget_(segment::SegmentBudget::Options{.max_active_segments = c.max_active_segments}) {
+        timer_ = std::jthread([this](std::stop_token st) { timer_loop(st); });
+    }
+
+    ~Impl() override {
+        // Quiesce every task first so no callback fires during or after teardown: mark it
+        // retired and cancel its transfers. Then stop the timer thread (a fn already
+        // running holds a shared_ptr and finishes, but seg_finished early-returns on
+        // `retired`). Then drop the registry; http_/prober_/budget_ destruct after.
+        {
+            std::lock_guard lk(reg_mu_);
+            for (auto &[id, t] : tasks_)
+                task::quiesce_task(t);
+        }
+        timer_.request_stop();
+        timer_cv_.notify_all();
+        if (timer_.joinable())
+            timer_.join();
+        {
+            std::lock_guard lk(reg_mu_);
+            tasks_.clear();
+        }
+    }
+
+    // --- TaskHost -------------------------------------------------------------------
+    net::HttpClient &http() override { return http_; }
+    segment::SegmentBudget &budget() override { return budget_; }
+    rate::RateLimiter &limiter() override { return limiter_; }
+    const Config &config() override { return cfg_; }
+
+    task::TimerId schedule(SteadyTime at, std::function<void()> fn) override {
+        task::TimerId id;
+        {
+            std::lock_guard lk(timer_mu_);
+            id = ++next_timer_;
+            timers_.emplace(at, Entry{id, std::move(fn)});
+        }
+        timer_cv_.notify_all();
+        return id;
+    }
+    void cancel_timer(task::TimerId id) override {
+        std::lock_guard lk(timer_mu_);
+        for (auto it = timers_.begin(); it != timers_.end(); ++it)
+            if (it->second.id == id) {
+                timers_.erase(it);
+                return;
+            }
+    }
+    void probe(net::ProbeRequest req, std::function<void(Result<net::ProbeResult>)> done) override {
+        prober_.probe(std::move(req), std::move(done));
+    }
+    void task_retired(TaskId id) override {
+        std::lock_guard lk(reg_mu_);
+        tasks_.erase(id);
+    }
+
+    // --- engine surface ----------------------------------------------------------------
+    task::DownloadHandle start(task::DownloadSpec spec, task::DownloadCallbacks cbs) {
+        TaskId id{++next_id_};
+        auto st = task::create_task(*this, id, std::move(spec), std::move(cbs));
+        {
+            std::lock_guard lk(reg_mu_);
+            tasks_[id] = st;
+        }
+        return task::DownloadHandle(std::move(st));
+    }
+
+    Config cfg_;
+    net::HttpClient http_;
+    net::Prober prober_;
+    segment::SegmentBudget budget_;
+    rate::RateLimiter limiter_;
+
+    std::atomic<std::uint64_t> next_id_{0};
+
+    std::mutex reg_mu_;
+    std::unordered_map<TaskId, std::shared_ptr<task::DownloadTaskState>> tasks_;
+
+    struct Entry {
+        task::TimerId id;
+        std::function<void()> fn;
+    };
+    std::mutex timer_mu_;
+    std::condition_variable timer_cv_;
+    std::multimap<SteadyTime, Entry> timers_;
+    std::atomic<std::uint64_t> next_timer_{0};
+    std::jthread timer_;
+
+    void timer_loop(std::stop_token st) {
+        std::unique_lock lk(timer_mu_);
+        while (!st.stop_requested()) {
+            if (timers_.empty()) {
+                timer_cv_.wait_for(lk, std::chrono::seconds(1));
+                continue;
+            }
+            auto next_at = timers_.begin()->first;
+            if (timer_cv_.wait_until(lk, next_at, [&] {
+                    return st.stop_requested() ||
+                           (!timers_.empty() && timers_.begin()->first < next_at);
+                })) {
+                continue;  // stop, or an earlier timer landed — re-evaluate
+            }
+            // fire everything due
+            std::vector<std::function<void()>> due;
+            auto now = std::chrono::steady_clock::now();
+            for (auto it = timers_.begin(); it != timers_.end() && it->first <= now;)
+                due.push_back(std::move(it->second.fn)), it = timers_.erase(it);
+            lk.unlock();
+            for (auto &fn : due)
+                fn();
+            lk.lock();
+        }
+    }
+};
+
+// --- Engine ---------------------------------------------------------------------------
+
+Engine::Engine() : Engine(Config{}) {}
+Engine::Engine(Config cfg) : impl_(std::make_unique<Impl>(cfg)) {}
+Engine::~Engine() = default;
+
+task::DownloadHandle Engine::start(task::DownloadSpec spec, task::DownloadCallbacks cbs) {
+    return impl_->start(std::move(spec), std::move(cbs));
+}
+
+segment::SegmentBudget &Engine::segment_budget() noexcept {
+    return impl_->budget_;
+}
+rate::RateLimiter &Engine::rate_limiter() noexcept {
+    return impl_->limiter_;
+}
+
+void Engine::set_default_segments(std::uint32_t n) {
+    impl_->cfg_.default_segments = n ? n : 1;
+}
+void Engine::set_default_buffer_bytes(std::uint64_t b) {
+    impl_->cfg_.default_buffer_bytes = b;
+}
+void Engine::set_max_total_buffer_bytes(std::uint64_t b) {
+    impl_->cfg_.max_total_buffer_bytes = b;
+}
+void Engine::set_probe_pool_size(std::uint32_t) { /* prober pool is fixed at construction in M1 */ }
+
+void Engine::probe(net::ProbeRequest req, std::function<void(Result<net::ProbeResult>)> done) {
+    impl_->prober_.probe(std::move(req), std::move(done));
+}
+
+}  // namespace vdm

core/src/io/sparse_file.cpp

@@ -74,7 +74,12 @@ Result<void> SparseFile::open(std::string_view path, const OpenOptions &opts) {
         return ErrorInfo(Error::internal, "SparseFile already open");
 
     std::string p(path);
-    int flags = O_WRONLY | O_CREAT | O_CLOEXEC;
+    // O_NOFOLLOW: the final component of a download target must never be a symlink, on
+    // create or on resume. DAEMON canonicalises the path and checks it against the allowed
+    // roots before start(), but a symlink swapped in afterwards would redirect our writes
+    // outside those roots (daemon/docs/safepath-adversarial.md leans on this open closing
+    // that TOCTOU window). A symlinked leaf fails here with ELOOP -> Error::path_rejected.
+    int flags = O_WRONLY | O_CREAT | O_CLOEXEC | O_NOFOLLOW;
     if (opts.truncate_existing)
         flags |= O_TRUNC;
 

core/src/net/http_client.cpp

@@ -195,6 +195,13 @@ struct HttpClient::Impl {
         std::string_view line(buf, total);
 
         if (line.starts_with("HTTP/")) {
+            // A new status line after we already delivered a 401/407 means libcurl's
+            // CURLAUTH_ANY handshake just resent with credentials: let the head of this
+            // second response be delivered too, so callers see the real (2xx/4xx) status
+            // rather than the challenge. Redirects never reach here delivered — their head
+            // is suppressed below — so this only fires for the auth resend.
+            if (st->head_delivered && (st->line_status == 401 || st->line_status == 407))
+                st->head_delivered = false;
             st->line_status = status_from_line(line);
             st->head.headers.clear();  // keep only the final response's headers
             return total;

core/src/net/probe.cpp

@@ -134,6 +134,7 @@ struct Prober::Impl {
         r.user_agent = pr.user_agent;
         r.referrer = pr.referrer;
         r.proxy = pr.proxy;
+        r.auth = pr.auth;
         r.follow_redirects = true;
         r.accept_encoding = false;
         r.connect_timeout_ms = pr.connect_timeout_ms;
@@ -155,7 +156,7 @@ struct Prober::Impl {
         TransferCallbacks cbs;
         cbs.on_head = [p](const ResponseHead &h) {
             absorb_head(p->result, h);
-            return DataAction::abort;
+            return head_action(p, h);
         };
         cbs.on_data = [](ConstByteSpan) { return DataAction::abort; };
         cbs.on_finished = [p](Result<TransferStats> r) { on_head_done(p, std::move(r)); };
@@ -165,6 +166,17 @@ struct Prober::Impl {
         client_.start(std::move(req), std::move(cbs));
     }
 
+    // We want no body from a probe, so the head callback normally aborts after headers.
+    // The exception: a 401/407 when we were handed credentials — libcurl's CURLAUTH_ANY
+    // has to see that response before it resends with Authorization, so let this one
+    // through (a HEAD has no body; the ranged GET's is a single byte). The final status
+    // then lands on the next header block.
+    static DataAction head_action(const std::shared_ptr<P> &p, const ResponseHead &h) {
+        if ((h.status == 401 || h.status == 407) && p->job.req.auth.scheme != AuthScheme::none)
+            return DataAction::proceed;
+        return DataAction::abort;
+    }
+
     static void absorb_head(ProbeResult &res, const ResponseHead &h) {
         if (h.status)
             res.http_status = h.status;
@@ -242,7 +254,7 @@ struct Prober::Impl {
         TransferCallbacks cbs;
         cbs.on_head = [p](const ResponseHead &h) {
             absorb_range_head(p->result, h);
-            return DataAction::abort;  // we don't need the one body byte
+            return head_action(p, h);  // abort after headers, except a 401/407 with creds
         };
         cbs.on_data = [](ConstByteSpan) { return DataAction::abort; };
         cbs.on_finished = [p](Result<TransferStats> r) { on_range_done(p, std::move(r)); };

core/src/segment/segmenter.cpp

@@ -263,6 +263,15 @@ void Segmenter::set_segment_state(std::uint32_t idx, SegState st) noexcept {
         segs_[idx].state.store(st);
 }
 
+void Segmenter::release_segment(std::uint32_t idx) noexcept {
+    std::lock_guard lk(mu_);
+    if (idx >= segs_.size())
+        return;
+    segs_[idx].assigned = false;
+    if (segs_[idx].state.load() != SegState::complete)
+        segs_[idx].state.store(SegState::idle);
+}
+
 // --- public: queries (take the lock) ----------------------------------------------
 
 std::uint64_t Segmenter::downloaded() const {

core/src/task/digest.cpp

@@ -0,0 +1,89 @@
+// vdm/task/digest.cpp
+
+#include "task/digest.hpp"
+
+#include <fcntl.h>
+#include <unistd.h>
+
+#include <array>
+#include <cerrno>
+#include <cstring>
+
+#include <openssl/evp.h>
+
+namespace vdm::task {
+namespace {
+
+const EVP_MD *md_for(Checksum::Algo a) {
+    switch (a) {
+        case Checksum::Algo::md5:
+            return EVP_md5();
+        case Checksum::Algo::sha1:
+            return EVP_sha1();
+        case Checksum::Algo::sha256:
+            return EVP_sha256();
+        case Checksum::Algo::sha512:
+            return EVP_sha512();
+    }
+    return EVP_sha256();
+}
+
+std::string to_hex(const unsigned char *p, unsigned n) {
+    static const char *h = "0123456789abcdef";
+    std::string s;
+    s.reserve(n * 2);
+    for (unsigned i = 0; i < n; ++i) {
+        s.push_back(h[p[i] >> 4]);
+        s.push_back(h[p[i] & 0xF]);
+    }
+    return s;
+}
+
+}  // namespace
+
+Result<std::string> hash_file(std::string_view path, Checksum::Algo algo) {
+    std::string p(path);
+    int fd = ::open(p.c_str(), O_RDONLY | O_CLOEXEC);
+    if (fd < 0)
+        return ErrorInfo(Error::path_rejected,
+                         std::string("open ") + p + ": " + std::strerror(errno));
+
+    EVP_MD_CTX *ctx = EVP_MD_CTX_new();
+    if (!ctx) {
+        ::close(fd);
+        return ErrorInfo(Error::internal, "EVP_MD_CTX_new");
+    }
+    auto fail = [&](Error e, std::string msg) {
+        EVP_MD_CTX_free(ctx);
+        ::close(fd);
+        return Result<std::string>(ErrorInfo(e, std::move(msg)));
+    };
+
+    if (EVP_DigestInit_ex(ctx, md_for(algo), nullptr) != 1)
+        return fail(Error::internal, "EVP_DigestInit_ex");
+
+    std::array<unsigned char, 256 * 1024> buf{};
+    for (;;) {
+        ssize_t n = ::read(fd, buf.data(), buf.size());
+        if (n < 0) {
+            if (errno == EINTR)
+                continue;
+            return fail(Error::io_error, std::string("read: ") + std::strerror(errno));
+        }
+        if (n == 0)
+            break;
+        if (EVP_DigestUpdate(ctx, buf.data(), static_cast<std::size_t>(n)) != 1)
+            return fail(Error::internal, "EVP_DigestUpdate");
+    }
+
+    unsigned char out[EVP_MAX_MD_SIZE];
+    unsigned out_len = 0;
+    if (EVP_DigestFinal_ex(ctx, out, &out_len) != 1)
+        return fail(Error::internal, "EVP_DigestFinal_ex");
+
+    EVP_MD_CTX_free(ctx);
+    ::close(fd);
+    return to_hex(out, out_len);
+}
+
+}  // namespace vdm::task

core/src/task/digest.hpp

@@ -0,0 +1,20 @@
+// vdm/task/digest.hpp — internal: hash a finished file for checksum verification.
+
+#ifndef VDM_TASK_DIGEST_HPP
+#define VDM_TASK_DIGEST_HPP
+
+#include <string>
+#include <string_view>
+
+#include "vdm/task/download.hpp"
+#include "vdm/util/result.hpp"
+
+namespace vdm::task {
+
+// Stream `path` through the digest and return it lower-case hex. io_error on a read
+// failure, path_rejected if the file can't be opened.
+[[nodiscard]] Result<std::string> hash_file(std::string_view path, Checksum::Algo algo);
+
+}  // namespace vdm::task
+
+#endif  // VDM_TASK_DIGEST_HPP

core/src/task/download_task.hpp

@@ -0,0 +1,58 @@
+// vdm/task/download_task.hpp — internal: the task machine behind DownloadHandle, and the
+// narrow interface it uses to reach engine-owned resources (so this TU doesn't depend on
+// Engine::Impl).
+
+#ifndef VDM_TASK_DOWNLOAD_TASK_HPP
+#define VDM_TASK_DOWNLOAD_TASK_HPP
+
+#include <cstdint>
+#include <functional>
+#include <memory>
+
+#include "vdm/engine.hpp"
+#include "vdm/ids.hpp"
+#include "vdm/net/http_client.hpp"
+#include "vdm/net/probe.hpp"
+#include "vdm/rate/token_bucket.hpp"
+#include "vdm/segment/budget.hpp"
+#include "vdm/task/download.hpp"
+
+namespace vdm::task {
+
+using TimerId = std::uint64_t;
+
+// Implemented by Engine::Impl. Every method is safe to call from any thread.
+struct TaskHost {
+    virtual ~TaskHost() = default;
+
+    virtual net::HttpClient &http() = 0;
+    virtual segment::SegmentBudget &budget() = 0;
+    virtual rate::RateLimiter &limiter() = 0;
+    virtual const Engine::Config &config() = 0;
+
+    // One-shot timer. `fn` runs on the engine's timer thread. cancel_timer is a no-op if
+    // it already fired or never existed.
+    virtual TimerId schedule(SteadyTime at, std::function<void()> fn) = 0;
+    virtual void cancel_timer(TimerId id) = 0;
+
+    // Probe pool, outside the segment budget (ADR 0011 §5).
+    virtual void probe(net::ProbeRequest req,
+                       std::function<void(Result<net::ProbeResult>)> done) = 0;
+
+    // The task reached a terminal state — drop it from the engine's registry.
+    virtual void task_retired(TaskId id) = 0;
+};
+
+// Create a task and begin it (probe or connect). The returned control block is what
+// DownloadHandle wraps (DownloadHandle{state}); the engine keeps its own copy so the task
+// outlives a caller that drops its handle.
+[[nodiscard]] std::shared_ptr<DownloadTaskState> create_task(TaskHost &host, TaskId id,
+                                                             DownloadSpec spec,
+                                                             DownloadCallbacks cbs);
+
+// Engine shutdown: stop every transfer and fire no further callbacks. Safe on nullptr.
+void quiesce_task(const std::shared_ptr<DownloadTaskState> &s);
+
+}  // namespace vdm::task
+
+#endif  // VDM_TASK_DOWNLOAD_TASK_HPP

core/tests/CMakeLists.txt

@@ -34,6 +34,13 @@ vdm_add_test(veloxcore_engine_api_test task/api_compiles_test.cpp)
 vdm_add_test(veloxcore_token_bucket_test rate/token_bucket_test.cpp)
 
 set(_testserver ${CMAKE_SOURCE_DIR}/tools/testserver/testserver.py)
+vdm_add_test(veloxcore_engine_test task/engine_test.cpp)
+target_include_directories(veloxcore_engine_test PRIVATE ${CMAKE_CURRENT_SOURCE_DIR}/net ${CMAKE_SOURCE_DIR}/core/src)
+if(EXISTS ${_testserver})
+    target_compile_definitions(veloxcore_engine_test PRIVATE VDM_TESTSERVER_PY="${_testserver}")
+    set_tests_properties(veloxcore_engine_test PROPERTIES TIMEOUT 300)
+endif()
+
 foreach(net_it http_client probe)
     vdm_add_test(veloxcore_${net_it}_test net/${net_it}_test.cpp)
     target_include_directories(veloxcore_${net_it}_test

core/tests/io/sparse_file_test.cpp

@@ -117,6 +117,23 @@ VT_TEST(sparse_open_bad_path_is_path_rejected) {
     VT_CHECK(!f.is_open());
 }
 
+VT_TEST(sparse_symlinked_target_is_rejected) {
+    // A symlink swapped in as the final path component after DAEMON's canonicalise-and-check
+    // must not be followed: the open is O_NOFOLLOW, so it fails with ELOOP -> path_rejected
+    // rather than redirecting our writes through the link.
+    TempPath link;    // the download target the caller hands us
+    TempPath target;  // where the symlink points (would-be victim, outside allowed roots)
+    VT_REQUIRE(::symlink(target.path.c_str(), link.path.c_str()) == 0);
+
+    SparseFile f;
+    auto r = f.open(link.path, {.total_size = 4096});
+    VT_REQUIRE(!r.has_value());
+    VT_CHECK_EQ(r.error().code, Error::path_rejected);
+    VT_CHECK(!f.is_open());
+    // the link target was never created/written through
+    VT_CHECK_EQ(::access(target.path.c_str(), F_OK), -1);
+}
+
 VT_TEST(sparse_ops_on_closed_file_error) {
     SparseFile f;
     VT_CHECK_EQ(f.write_at(0, bytes("x")).error().code, Error::internal);

core/tests/support/vtest.hpp

@@ -81,9 +81,17 @@ std::string show(const T &v) {
     }
 }
 
-inline int run_all() {
+inline int run_all(const std::vector<std::string> &filters = {}) {
     int failed_cases = 0;
     for (const auto &c : registry()) {
+        if (!filters.empty()) {
+            bool match = false;
+            for (const auto &f : filters)
+                if (std::string_view(c.name).find(f) != std::string_view::npos)
+                    match = true;
+            if (!match)
+                continue;
+        }
         int before = stats().failures;
         stats().current_fatal = false;
         std::fprintf(stderr, "[ RUN  ] %s\n", c.name);

core/tests/support/vtest_main.cpp

@@ -1,6 +1,29 @@
 // vtest_main.cpp — shared entry point for every CORE test binary.
+#include <cstdlib>
+#include <string>
+#include <vector>
+
 #include "vtest.hpp"
 
-int main() {
-    return ::vt::run_all();
+// Optional filters: each argv argument (or a comma-separated entry in $VT_ONLY) is a
+// substring; a test runs only if its name contains one of them. No filters => run all.
+int main(int argc, char **argv) {
+    std::vector<std::string> filters;
+    for (int i = 1; i < argc; ++i)
+        filters.emplace_back(argv[i]);
+    if (const char *env = std::getenv("VT_ONLY")) {
+        std::string cur;
+        for (const char *p = env;; ++p) {
+            if (*p == ',' || *p == '\0') {
+                if (!cur.empty())
+                    filters.push_back(cur);
+                cur.clear();
+                if (*p == '\0')
+                    break;
+            } else {
+                cur.push_back(*p);
+            }
+        }
+    }
+    return ::vt::run_all(filters);
 }

core/tests/task/engine_test.cpp

@@ -0,0 +1,328 @@
+// End-to-end: a real Engine against tools/testserver, covering the CORE M1 DoD paths.
+
+#include "vdm/engine.hpp"
+
+#include <fcntl.h>
+#include <unistd.h>
+
+#include <atomic>
+#include <chrono>
+#include <cstdlib>
+#include <future>
+#include <string>
+#include <vector>
+
+#include "task/digest.hpp"
+#include "testserver_fixture.hpp"
+#include "vtest.hpp"
+
+using namespace vdm;
+using namespace vdm::task;
+using vdm::testing::TestServer;
+using namespace std::chrono_literals;
+
+namespace {
+
+struct TmpDir {
+    std::string path;
+    TmpDir() {
+        const char *d = std::getenv("TMPDIR");
+        path = (d ? d : "/tmp");
+        path += "/vdm_engine_XXXXXX";
+        path = ::mkdtemp(path.data()) ? path : "";
+    }
+    ~TmpDir() {
+        // best-effort recursive cleanup of our flat dir
+        if (path.empty())
+            return;
+        std::string cmd = "rm -rf '" + path + "'";
+        (void)std::system(cmd.c_str());
+    }
+    std::string file(const std::string &name) const { return path + "/" + name; }
+};
+
+struct Recorder {
+    std::promise<Result<DownloadOutcome>> done;
+    std::future<Result<DownloadOutcome>> fut = done.get_future();
+    std::atomic<bool> fired{false};
+    std::vector<EngineState> states;
+    std::mutex mu;
+    std::atomic<int> auth_calls{0};
+    std::atomic<int> decision_calls{0};
+    // A probe callback can fire before the caller has stored the handle returned by
+    // eng.start(). Callbacks that reach back into the handle wait on this.
+    std::atomic<bool> handle_ready{false};
+
+    void arm(DownloadHandle &) { handle_ready.store(true, std::memory_order_release); }
+
+    DownloadCallbacks cbs(DownloadHandle *h = nullptr, std::string user = "",
+                          std::string pass = "") {
+        DownloadCallbacks c;
+        c.on_state = [this](EngineState, EngineState to, const std::optional<ErrorInfo> &) {
+            std::lock_guard lk(mu);
+            states.push_back(to);
+        };
+        c.on_finished = [this](Result<DownloadOutcome> r) {
+            if (!fired.exchange(true))
+                done.set_value(std::move(r));
+        };
+        if (h) {
+            c.on_auth_required = [this, h, user, pass](const AuthChallenge &) {
+                auth_calls.fetch_add(1);
+                while (!handle_ready.load(std::memory_order_acquire))
+                    std::this_thread::sleep_for(1ms);
+                h->provide_auth(user, pass, false);
+            };
+        }
+        return c;
+    }
+    Result<DownloadOutcome> wait(std::chrono::seconds to = 40s) {
+        if (fut.wait_for(to) != std::future_status::ready)
+            return Err{Error::timeout, "engine test wait"};
+        return fut.get();
+    }
+    bool saw(EngineState s) {
+        std::lock_guard lk(mu);
+        for (auto x : states)
+            if (x == s)
+                return true;
+        return false;
+    }
+};
+
+std::uint64_t file_size(const std::string &p) {
+    int fd = ::open(p.c_str(), O_RDONLY);
+    if (fd < 0)
+        return ~0ull;
+    off_t e = ::lseek(fd, 0, SEEK_END);
+    ::close(fd);
+    return e < 0 ? ~0ull : static_cast<std::uint64_t>(e);
+}
+
+// The reference SHA-256 the testserver will report for a given path.
+std::string server_sha(TestServer &srv, const std::string &mode, const std::string &size) {
+    // one-shot GET /<mode>/sha256/<size> via a throwaway Engine::probe? no — use raw curl
+    // through a small helper. Simplest: shell out.
+    std::string url = srv.url("/" + mode + "/sha256/" + size);
+    std::string cmd = "curl -s '" + url + "'";
+    std::string out;
+    if (FILE *f = ::popen(cmd.c_str(), "r")) {
+        char buf[512];
+        while (std::fgets(buf, sizeof buf, f))
+            out += buf;
+        ::pclose(f);
+    }
+    auto q = out.find("\"sha256\"");
+    if (q == std::string::npos)
+        return {};
+    auto colon = out.find(':', q);
+    auto open = out.find('"', colon);
+    auto close = out.find('"', open + 1);
+    if (open == std::string::npos || close == std::string::npos)
+        return {};
+    return out.substr(open + 1, close - open - 1);
+}
+
+DownloadSpec spec_for(TestServer &srv, const std::string &urlpath, const std::string &save) {
+    DownloadSpec s;
+    s.url = srv.url(urlpath);
+    s.save_path = save;
+    return s;
+}
+
+}  // namespace
+
+VT_TEST(engine_plain_multisegment_download) {
+    TestServer srv;
+    VT_REQUIRE(srv.available());
+    TmpDir td;
+    VT_REQUIRE(!td.path.empty());
+
+    Recorder rec;
+    Engine eng;
+    auto h = eng.start(spec_for(srv, "/plain/file/4M", td.file("a.bin")), rec.cbs());
+
+    auto r = rec.wait();
+    VT_REQUIRE(r.has_value());
+    VT_CHECK_EQ(r.value().final_path, td.file("a.bin"));
+    VT_CHECK_EQ(r.value().bytes, 4u * 1024 * 1024);
+    VT_CHECK_EQ(file_size(td.file("a.bin")), 4u * 1024 * 1024);
+    VT_CHECK(rec.saw(EngineState::downloading));
+    VT_CHECK(rec.saw(EngineState::complete));
+
+    auto got = hash_file(td.file("a.bin"), Checksum::Algo::sha256);
+    VT_REQUIRE(got.has_value());
+    VT_CHECK_EQ(got.value(), server_sha(srv, "plain", "4M"));
+    // the sidecar is gone on success
+    VT_CHECK_EQ(::access((td.file("a.bin") + ".veloxpart.meta").c_str(), F_OK), -1);
+}
+
+VT_TEST(engine_checksum_pass_and_mismatch) {
+    TestServer srv;
+    VT_REQUIRE(srv.available());
+    TmpDir td;
+    Engine eng;
+
+    std::string want = server_sha(srv, "plain", "1M");
+    VT_REQUIRE(!want.empty());
+
+    {
+        Recorder rec;
+        auto s = spec_for(srv, "/plain/file/1M", td.file("ok.bin"));
+        s.checksum = Checksum{Checksum::Algo::sha256, want};
+        auto h = eng.start(std::move(s), rec.cbs());
+        auto r = rec.wait();
+        VT_REQUIRE(r.has_value());
+        VT_CHECK(rec.saw(EngineState::verifying));
+    }
+    {
+        Recorder rec;
+        auto s = spec_for(srv, "/plain/file/1M", td.file("bad.bin"));
+        s.checksum = Checksum{Checksum::Algo::sha256, std::string(64, 'a')};
+        auto h = eng.start(std::move(s), rec.cbs());
+        auto r = rec.wait();
+        VT_REQUIRE(!r.has_value());
+        VT_CHECK_EQ(r.error().code, Error::checksum_mismatch);
+    }
+}
+
+VT_TEST(engine_non_resumable_single_segment) {
+    TestServer srv;
+    VT_REQUIRE(srv.available());
+    TmpDir td;
+    Recorder rec;
+    Engine eng;
+    auto h = eng.start(spec_for(srv, "/no-range/file/2M", td.file("nr.bin")), rec.cbs());
+    auto r = rec.wait();
+    VT_REQUIRE(r.has_value());
+    VT_CHECK_EQ(file_size(td.file("nr.bin")), 2u * 1024 * 1024);
+    auto got = hash_file(td.file("nr.bin"), Checksum::Algo::sha256);
+    VT_CHECK_EQ(got.value(), server_sha(srv, "no-range", "2M"));
+}
+
+VT_TEST(engine_404_is_an_error) {
+    TestServer srv;
+    VT_REQUIRE(srv.available());
+    TmpDir td;
+    Recorder rec;
+    Engine eng;
+    auto h = eng.start(spec_for(srv, "/plain/nope", td.file("x.bin")), rec.cbs());
+    auto r = rec.wait();
+    VT_REQUIRE(!r.has_value());
+    VT_CHECK_EQ(r.error().code, Error::not_found);
+    VT_CHECK(rec.saw(EngineState::failed));
+}
+
+VT_TEST(engine_cancel_mid_download) {
+    TestServer srv;
+    VT_REQUIRE(srv.available());
+    TmpDir td;
+    Recorder rec;
+    Engine eng;
+    auto h = eng.start(spec_for(srv, "/throttled/file/8M", td.file("c.bin")), rec.cbs());
+
+    for (int i = 0; i < 200 && !rec.saw(EngineState::downloading); ++i)
+        std::this_thread::sleep_for(10ms);
+    VT_REQUIRE(rec.saw(EngineState::downloading));
+    h.cancel(/*discard_partial=*/true);
+
+    auto r = rec.wait();
+    VT_REQUIRE(!r.has_value());
+    VT_CHECK_EQ(r.error().code, Error::canceled);
+    VT_CHECK(rec.saw(EngineState::cancelled));
+    VT_CHECK_EQ(::access((td.file("c.bin") + ".veloxpart").c_str(), F_OK), -1);  // discarded
+}
+
+VT_TEST(engine_pause_resume_completes) {
+    TestServer srv;
+    VT_REQUIRE(srv.available());
+    TmpDir td;
+    Recorder rec;
+    Engine eng;
+    auto h = eng.start(spec_for(srv, "/throttled/file/2M", td.file("pr.bin")), rec.cbs());
+
+    for (int i = 0; i < 200 && !rec.saw(EngineState::downloading); ++i)
+        std::this_thread::sleep_for(10ms);
+    h.pause();
+    for (int i = 0; i < 100 && h.state() != EngineState::paused; ++i)
+        std::this_thread::sleep_for(20ms);
+    VT_CHECK_EQ(h.state(), EngineState::paused);
+    h.resume();
+
+    auto r = rec.wait(90s);
+    VT_REQUIRE(r.has_value());
+    VT_CHECK_EQ(file_size(td.file("pr.bin")), 2u * 1024 * 1024);
+    auto got = hash_file(td.file("pr.bin"), Checksum::Algo::sha256);
+    VT_CHECK_EQ(got.value(), server_sha(srv, "throttled", "2M"));
+}
+
+VT_TEST(engine_resume_after_a_fresh_task) {
+    // Simulates kill -9: cancel WITHOUT discard, then a new task with allow_resume picks
+    // up the .veloxpart[.meta] and finishes with a byte-identical file.
+    TestServer srv;
+    VT_REQUIRE(srv.available());
+    TmpDir td;
+    std::string save = td.file("resume.bin");
+    std::string want = server_sha(srv, "throttled", "3M");
+
+    {
+        Recorder rec;
+        Engine eng;
+        auto h = eng.start(spec_for(srv, "/throttled/file/3M", save), rec.cbs());
+        for (int i = 0; i < 800 && (h.progress().downloaded < 512u * 1024); ++i)
+            std::this_thread::sleep_for(10ms);
+        VT_REQUIRE(h.progress().downloaded >= 512u * 1024);
+        h.cancel(/*discard_partial=*/false);
+        (void)rec.wait();
+        VT_CHECK_EQ(::access((save + ".veloxpart.meta").c_str(), F_OK), 0);  // sidecar kept
+    }
+    {
+        Recorder rec;
+        Engine eng;
+        auto s = spec_for(srv, "/throttled/file/3M", save);  // same source -> sidecar validates
+        s.allow_resume = true;
+        auto h = eng.start(std::move(s), rec.cbs());
+        auto r = rec.wait(120s);
+        VT_REQUIRE(r.has_value());
+        VT_CHECK_EQ(file_size(save), 3u * 1024 * 1024);
+        auto got = hash_file(save, Checksum::Algo::sha256);
+        VT_CHECK_EQ(got.value(), want);  // byte-identical after resume
+    }
+}
+
+VT_TEST(engine_flaky_reset_retries_to_completion) {
+    TestServer srv;
+    VT_REQUIRE(srv.available());
+    TmpDir td;
+    Recorder rec;
+    Engine eng;
+    // flaky-reset RSTs the first two attempts per (path,range); a single-segment request
+    // therefore needs the retry loop.
+    auto s = spec_for(srv, "/flaky-reset/file/256K", td.file("fl.bin"));
+    s.segments = 1;
+    s.max_retries = 40;  // the server RSTs at the halfway point every attempt -> ~18 halvings
+    auto h = eng.start(std::move(s), rec.cbs());
+    auto r = rec.wait(120s);
+    VT_REQUIRE(r.has_value());
+    VT_CHECK_EQ(file_size(td.file("fl.bin")), 256u * 1024);
+    auto got = hash_file(td.file("fl.bin"), Checksum::Algo::sha256);
+    VT_CHECK_EQ(got.value(), server_sha(srv, "flaky-reset", "256K"));
+    VT_CHECK(rec.saw(EngineState::retry_wait) || rec.saw(EngineState::connecting));
+}
+
+VT_TEST(engine_401_then_provide_auth_completes) {
+    TestServer srv;
+    VT_REQUIRE(srv.available());
+    TmpDir td;
+    Recorder rec;
+    Engine eng;
+    DownloadHandle h;
+    auto cbs = rec.cbs(&h, "test", "test");
+    h = eng.start(spec_for(srv, "/401-basic/file/1M", td.file("au.bin")), std::move(cbs));
+    rec.arm(h);  // publish h to the auth callback (which may already be waiting)
+
+    auto r = rec.wait();
+    VT_REQUIRE(r.has_value());
+    VT_CHECK(rec.auth_calls.load() >= 1);
+    VT_CHECK_EQ(file_size(td.file("au.bin")), 1u * 1024 * 1024);
+}

docs/04-engine-design.md

@@ -61,8 +61,14 @@ Rules:
 
 ## 4. Disk I/O — the buffer setting you asked for
 
-One file, opened once, `O_WRONLY`. Each segment `pwrite()`s at its own absolute offset, so
-**there is no reassembly pass and no second write of the whole file.**
+One file, opened once, `O_WRONLY | O_NOFOLLOW`. Each segment `pwrite()`s at its own absolute
+offset, so **there is no reassembly pass and no second write of the whole file.**
+
+`O_NOFOLLOW` on the part-file open: DAEMON canonicalises the save path and checks it against
+the allowed roots before `start()`, but the final component could be swapped for a symlink
+in the window between that check and our open. A symlinked leaf is rejected here (`ELOOP` →
+`Error::path_rejected`), not followed — it closes the TOCTOU residual that
+`daemon/docs/safepath-adversarial.md` accepts on those grounds.
 
 - `posix_fallocate()` the full size up front → contiguous extents, no ENOSPC surprise at
   99 %, no fragmentation.