merge: lane/core

2026-09-10 18:55:19 +04:00
parent 7289943375 3da4cd6e91
commit b97ed5a5c6
9 changed files with 926 additions and 3 deletions
@@ -0,0 +1,186 @@
 # CORE → DAEMON — the engine API (M1)
 **Status: reviewed and signed off** by DAEMON on 2026-09-10
 (`daemon/docs/engine-api-review.md`). Nothing forced a `sched/` or dispatch rewrite. The
 five open questions are resolved at the bottom; the review's four confirms are folded into
 the headers (`sha512` added to `Checksum::Algo`; the parent-directory contract made
 explicit; `cancel()` ordering documented; `rate_limiter()` accessor added).
 Integration order: DAEMON wires this in after `daemon/src/sched/` lands (the scheduler is
 what calls `start`/`pause`/`resume`/`cancel` and drives `set_task_order`); `sched/` builds
 against these headers in parallel with CORE stage 8 and does not need the bodies.
 ---
 This is the `libveloxcore` public surface `veloxd` links and calls
 to actually run a download. It is the thing the AGENT-CORE brief asked for on day one and
 that slipped: DAEMON has an RPC surface and a store and, until this is agreed, nothing to
 call. Sketch headers: `core/include/vdm/engine.hpp`, `core/include/vdm/task/download.hpp`
 (both compile now; `Engine` / `DownloadHandle` bodies land in CORE stage 8). Nothing here
 touches `contracts/` — DAEMON projects these callbacks onto `TaskSummary` / `TaskDetail` /
 `event.*`.
 Please review the field semantics, the threading/lifetime rules, and the
 pause/resume/cancel contract, and raise anything that would force a `daemon/src/sched/` or
 RPC-dispatch rewrite later. Open questions are at the bottom.
 ---
 ## 1. `DownloadSpec` — what DAEMON hands the engine
 DAEMON has already run the rules engine, picked the category folder, canonicalised the
 path and checked it against the allowed roots (`-32011` is DAEMON's error, raised before
 `start()`), and resolved the filename. The engine's spec is the concrete result.
 | field | who fills it | notes |
 |---|---|---|
 | `url`, `mirrors` | DAEMON | `mirrors` are alternative URLs for the *same bytes*; the segmenter's requeue prefers a different host after 3 connection failures (docs/04 §3). |
 | `headers`, `cookies`, `referrer`, `user_agent` | DAEMON, verbatim from the capture | replayed on every segment request and on the probe. |
 | `save_path` | DAEMON | **absolute and final.** The engine never canonicalises or root-checks. `<save_path>.veloxpart` and `.veloxpart.meta` sit beside it; on success the part file is renamed in place. |
 | `segments` | user / `connection.maxSegmentsPerDownload` | requested upper bound 1–32; the engine lowers it to the per-host cap and to 1 for a non-resumable source. Effective value comes back in `Progress.effective_segments`. |
 | `buffer_bytes` | user / `connection.bufferBytes` | requested per-segment; silently reduced to fit `maxTotalBufferBytes` across all live segments. Effective value in `Progress.effective_buffer_bytes` → your `TaskDetail.effectiveBufferBytes`. |
 | `proxy`, `auth` | DAEMON | `auth` carries credentials only if known up front (Secret Service). Leave `scheme == none` to get an `on_auth_required` on a 401/407 instead. |
 | `checksum` | user (`download.add.checksum`) | `{algo, hex}`. Verified during `verifying`; a mismatch is a terminal `failed` with `Error::checksum_mismatch`. |
 | `probe_hint` | DAEMON | the `ProbeResult` you already got for the File Info dialog. Supplying it skips the engine's own probe — the task starts in `connecting`, not `probing`. The engine still revalidates with `If-Range` on resume. |
 | `allow_resume` | DAEMON | `true`: if a CRC-valid `.veloxpart.meta` sits beside `save_path`, resume from it. `false`: start fresh, overwrite. Your restart flow (ADR 0013 §5) sets this per task. |
 | `max_retries` | user / default 10 | per segment, before `failed` with `Error::max_retries_exhausted`. |
 `start()` returns immediately. It never throws and never blocks on the network; a bad URL,
 DNS failure, or unwritable `save_path` is delivered through `on_finished`.
 ## 2. The state machine the engine drives
 `EngineState` is the CORE-owned subset of the wire `TaskState` (ADR 0013 §1):
 ```
 probing ─▶ connecting ⇄ downloading ─▶ assembling ─▶ verifying ─▶ complete
   │            │  ▲          │  │            (M4 mux; a no-op rename in M1)
   │            │  └─ retry_wait ┘
   ▼            ▼
 (any) ──────▶ paused ──(resume)──▶ connecting
 (any CORE state) ──────────────────▶ failed        (terminal, engine-initiated)
 (any state, on cancel()) ──────────▶ cancelled     (terminal, DAEMON-initiated)
 ```
 `new` and `queued` are yours; the engine never emits them. `start()` corresponds to your
 `queued → probing`. Every transition is reported through `on_state(from, to, error?)`,
 including the auto-pauses (§4) and the terminals. `previousState` on your
 `event.task.state` maps straight from the `from` argument.
 ## 3. Threading and lifetime
 - **Callbacks run on an engine thread** — a transfer worker, the progress timer, or a
  dispatch thread — **never** the thread that called `start()` / `pause()` / etc.
 - **Per task, callbacks are serialised.** You will never get two callbacks for the same
  handle at once. Across tasks they run concurrently.
 - **A callback must not block.** It runs on a thread doing real transfer work; a slow
  callback stalls that work. Hand off to your own queue/loop.
 - **A callback must not re-enter the same handle synchronously** — no `pause()` from
  inside `on_state`, etc. Post it. (Calling into a *different* handle, or into
  `segment_budget()`, is fine.)
 - **`on_finished` is always the last callback** for a task. After it returns the engine
  makes no further callbacks for that handle and the handle's control methods are no-ops.
 - **The handle is copyable and thread-safe.** Dropping the last copy does **not** cancel —
  the task runs on. Call `cancel()` to stop it. (DAEMON holds the handle for the task's
  life anyway.)
 - **`Engine` must outlive every handle.** `~Engine()` cancels all running tasks and joins
  their workers before returning — expect it to block briefly.
 - **Logging**: the engine writes through `vdm::set_log_sink()` (a `core/util` global).
  Install your sink once at startup; the engine never opens a file itself.
 ## 4. `paused` is shared, and idempotency is the contract (ADR 0013 §2)
 Both sides put a task in `paused`, for disjoint reasons:
 - **DAEMON-initiated**: `handle.pause()` — user pause, a schedule window closing, a queue
  stop, `Queue.onComplete`, the admission governor reconciling a lowered
  `maxActiveSegments`.
 - **Engine auto-pause**: `on_auth_required` (401/407), `on_decision_needed`
  (`server_file_changed` / stale range), disk full. The engine transitions to `paused`
  on its own and fires `on_state(_, paused, ErrorInfo{...})` — the same path as any other
  transition. `ErrorInfo.code` present ⇒ engine-initiated; absent ⇒ you did it. That is
  the only discriminator, and it is what your `error`-on-`paused` widening (open item 3 of
  ADR 0013) carries on the wire.
 **Idempotency, now a signature:**
 | call | already in that state / terminal | otherwise |
 |---|---|---|
 | `pause()` | no-op, no error | stop new segment requests, flush + `fdatasync` in-flight buffers, write `.veloxpart.meta`, release the budget slots, `on_state(_, paused, nullopt)`. Bounded by the slowest in-flight flush. |
 | `resume()` | no-op if not `paused`; no-op if terminal | revalidate with `If-Range`, re-acquire budget slots, `paused → connecting`, resume from the sidecar offsets. |
 | `cancel(discard_partial)` | no-op if already terminal | stop everything, `on_state(_, cancelled, nullopt)`, `on_finished(Err{cancelled})`. `discard_partial` also unlinks `.veloxpart[.meta]` — wire this to `download.remove {deleteFile}`. |
 `resume()` after an auto-pause for `auth_required` **without** a preceding
 `provide_auth()` is a no-op — the task stays paused. This is ADR 0013 §3's "resume must
 not cross reasons", enforced on CORE's side: the scheduler cannot accidentally un-pause a
 task waiting on credentials.
 - **`provide_auth(user, pass, remember)`** — acts only while the task is auto-paused for
  auth; supplies the credential for the retry and resumes. `remember` asks *you* to
  persist to the Secret Service; the engine never stores it. No-op otherwise.
 - **`decide(Decision)`** — acts only while auto-paused for a decision. `restart` discards
  the partial and re-downloads; `keep_partial` continues against what is on disk (the
  user's stated risk); `abort` → `failed`. No-op otherwise.
 - **`refresh_url(url, headers)`** — IDM's "Refresh Download Address": the engine re-probes
  the new URL to validate it, then **restarts every segment on it** (the signed-URL-expiry
  case the wire method exists for), keeping the bytes already on disk. Mirror rotation is
  *not* this — that is `spec.mirrors` + the segmenter's requeue-to-a-different-host.
 - **`on_decision_needed`** is only for the cases the engine cannot resolve itself. A
  routine 416 / stale range is the engine's own re-probe + re-split loop; it escalates
  `DecisionRequest{range_metadata_stale}` **only when that loop fails**, at which point
  "retry the same range" is already exhausted — so `{restart, keep_partial, abort}` is the
  whole choice set.
 ## 5. Progress
 `on_progress` is coalesced to **≤ 4 Hz per task** inside the engine — the same cadence as
 `event.task.progress`, so your batcher can forward without re-throttling. It carries
 aggregate `downloaded` / `speed_bps` / `eta_seconds`, the effective segment count and
 buffer size, and a `SegmentProgress[]` (index, inclusive `[start,end]`, `completed`,
 per-segment speed, state) for the GUI's segment bars. `total` is absent for a chunked
 source until the stream ends.
 `handle.state()` and `handle.progress()` are synchronous lock-guarded snapshots for
 `download.get` / `download.list` — call them any time, including from your RPC thread.
 ## 6. What the engine does NOT do
 - No filename resolution, no category matching, no path canonicalisation, no allowed-root
  check — all DAEMON, before `start()`.
 - No queueing, scheduling, priority, or "when queue completes" — DAEMON, via
  `segment_budget().set_task_order()` and by choosing when to call `start()` / `pause()`.
 - No persistence beyond `.veloxpart.meta`. On a daemon restart the engine knows nothing;
  you reload from SQLite, rewrite CORE-owned states to `queued`, and re-`start()` with
  `allow_resume = true` (ADR 0013 §5).
 - No credential storage. Ever (`CLAUDE.md` §4).
 ---
 ## Resolved (DAEMON review, 2026-09-10)
 1. **`probe_hint` stays optional.** DAEMON has a `ProbeResult` only on the File-Info path;
   capture-take, `velox add`, `addBatch` and restart have none. The engine probes when
   it's absent.
 2. **One `cancel(discard_partial)`.** `download.cancel` = `cancel(false)`;
   `download.remove` = `cancel(true)` for a live task (plus DAEMON's row/file cleanup), or
   pure DAEMON-side for an already-terminal one. No `handle.remove()`.
 3. **`{restart, keep_partial, abort}` is the whole set.** The engine owns the routine
   416 re-probe/re-split and only escalates `on_decision_needed` when that loop fails —
   "retry the same range" is exhausted by then.
 4. **Per-task 4 Hz is fine.** DAEMON coalesces across tasks for `event.task.progress`
   regardless; `on_progress_batch` is a nice-to-have and must not block stage 8.
 5. **`refresh_url` restarts all segments on the new URL** after a validating re-probe (the
   signed-URL case). Mirror rotation is `spec.mirrors` + the segmenter, not this.
 ## Review confirms, folded in
 - **(a)** `vdm::TaskId` is a cheap-copy hashable value; DAEMON never constructs one — it
  only receives it from `start()` / callbacks and passes it back to `set_task_order()`
  etc. ✔ (`vdm/ids.hpp`)
 - **(b)** DAEMON `mkdir -p`s `save_path`'s parent before `start()`. The engine opens the
  file and fails with `Error::path_rejected` if the directory is missing. ✔ (documented on
  `DownloadSpec`)
 - **(c)** `Checksum::Algo` now has `sha512`, matching the wire `Checksum` set. ✔
 - **(d)** `cancel()` always fires `on_state(_, cancelled, nullopt)` then
  `on_finished(Err{Error::canceled})` (note: the taxonomy value is `canceled`), in that
  order. ✔ (documented on `DownloadHandle::cancel`)
@@ -1,8 +1,12 @@
 # `libveloxcore` — public API
-**Status: M1 in progress.** Only `util/` is landed. The download-facing API
+**Status: M1 in progress.** `util/`, `net/` (http_client, probe, url, content_disposition),
-(`DownloadSpec`, `DownloadTask`, probe, typed callbacks) arrives with later stages and
+`io/` (sparse_file, write_buffer), `meta/veloxpart`, and `segment/` (segmenter, budget)
-is reviewed by DAEMON before M2 (AGENT-CORE DoD).
+are landed. The **download entry point** — `vdm::Engine`, `vdm::task::DownloadSpec` /
 `DownloadHandle` / `DownloadCallbacks` — is sketched in `vdm/engine.hpp` and
 `vdm/task/download.hpp` and **out for DAEMON review**: see
 [`core/docs/engine-api-m1.md`](../../docs/engine-api-m1.md). Bodies land in CORE stage 8;
 build against the value types now.
 Layering (CLAUDE.md §3): this library knows nothing about JSON, SQL, Qt, or RPC. Input is
 a spec value; output is bytes on disk plus typed callbacks. DAEMON projects engine state
@@ -0,0 +1,82 @@
 // vdm/engine.hpp — the download engine's single entry point. REVIEW SKETCH (stage 7
 // pre-work); bodies land in stage 8. See core/docs/engine-api-m1.md.
 //
 // The engine owns the HTTP client, the probe pool, the segment budget, and the disk I/O.
 // Its input is a DownloadSpec; its output is bytes at save_path plus typed callbacks. No
 // JSON, no SQL, no Qt, no RPC — DAEMON projects the callbacks onto the wire contract.
 //
 // This header compiles standalone.
 #ifndef VDM_ENGINE_HPP
 #define VDM_ENGINE_HPP
 #include <cstdint>
 #include <memory>
 #include "vdm/rate/token_bucket.hpp"
 #include "vdm/segment/budget.hpp"
 #include "vdm/task/download.hpp"
 namespace vdm {
 class Engine {
  public:
    struct Config {
        // Defaults used when a DownloadSpec leaves the field unset. Live-adjustable via
        // the setters below (they take effect on the next segment (re)assignment, not by
        // resizing an in-flight buffer).
        std::uint32_t default_segments = 8;                   // connection.maxSegmentsPerDownload
        std::uint64_t default_buffer_bytes = 1u << 20;        // connection.bufferBytes (1 MiB)
        std::uint64_t min_segment_bytes = 1u << 20;           // never split below this
        std::uint64_t max_total_buffer_bytes = 128ull << 20;  // connection.maxTotalBufferBytes
        std::uint32_t max_active_segments = 32;               // connection.maxActiveSegments
        std::uint32_t probe_pool_size = 4;                    // ADR 0011 §5, outside the budget
        long default_max_retries = 10;                        // per segment
        std::uint32_t http_workers = 0;                       // 0 => hardware-derived (<=4)
    };
    Engine();  // default Config
    explicit Engine(Config cfg);
    ~Engine();  // cancels every running task and joins before returning
    Engine(const Engine &) = delete;
    Engine &operator=(const Engine &) = delete;
    // Start a download. Returns immediately with a handle; the task begins in `probing`
    // (or `connecting` when spec.probe_hint is supplied). Every failure — bad URL, DNS,
    // an unwritable save_path — is delivered through callbacks.on_finished, never thrown.
    [[nodiscard]] task::DownloadHandle start(task::DownloadSpec spec,
                                             task::DownloadCallbacks callbacks);
    // The global segment allocator. DAEMON's scheduler drives admission through this
    // (set_max_active_segments / set_host_segment_cap / set_task_order) and reads
    // occupancy from it (budget() / segments_active() / starved_tasks() /
    // on_budget_changed). See ADR 0011.
    [[nodiscard]] segment::SegmentBudget &segment_budget() noexcept;
    // The hierarchical speed limiter (docs/04 §6): global -> per-queue -> per-task token
    // buckets. `limiter.set {globalBps, enabled}` -> rate_limiter().set_global_limit();
    // per-queue / per-task limits and the task<->queue attachment come from DAEMON too.
    // The engine paces every segment read through it.
    [[nodiscard]] rate::RateLimiter &rate_limiter() noexcept;
    // Live settings (connection.* changes from settings.set). Each affects future work.
    void set_default_segments(std::uint32_t n);
    void set_default_buffer_bytes(std::uint64_t bytes);
    void set_max_total_buffer_bytes(std::uint64_t bytes);
    void set_probe_pool_size(std::uint32_t n);
    // A standalone probe for the File Info dialog, on the same pool as spec-less probes
    // (never charged against the segment budget). capture.offer's 750 ms deadline is
    // DAEMON's to enforce — it should answer `ignore` and probe after, never block on
    // this.
    void probe(net::ProbeRequest req, std::function<void(Result<net::ProbeResult>)> done);
  private:
    struct Impl;
    std::unique_ptr<Impl> impl_;
 };
 }  // namespace vdm
 #endif  // VDM_ENGINE_HPP
@@ -24,6 +24,15 @@ struct TaskId {
    friend constexpr auto operator<=>(const TaskId &, const TaskId &) = default;
 };
 // A scheduler queue. A task may belong to one — for the per-queue rate limit and
 // per-queue concurrency. A task with no queue rate-limits against the global bucket only.
 struct QueueId {
    std::uint64_t value = 0;
    [[nodiscard]] constexpr bool valid() const noexcept { return value != 0; }
    friend constexpr auto operator<=>(const QueueId &, const QueueId &) = default;
 };
 using SteadyTime = std::chrono::steady_clock::time_point;
 }  // namespace vdm
@@ -35,4 +44,11 @@ struct std::hash<vdm::TaskId> {
    }
 };
 template <>
 struct std::hash<vdm::QueueId> {
    std::size_t operator()(vdm::QueueId id) const noexcept {
        return std::hash<std::uint64_t>{}(id.value);
    }
 };
 #endif  // VDM_IDS_HPP
@@ -0,0 +1,186 @@
 // vdm/rate/token_bucket.hpp — a lazily-refilled token bucket, and the global -> queue ->
 // task limiter hierarchy built on it (docs/04 §6).
 //
 // A segment worker calls RateLimiter::acquire(task, n) after receiving n body bytes. If
 // every applicable level (task, its queue, global) has n tokens, it consumes n from each
 // and returns 0. Otherwise it consumes nothing and returns how long to wait before
 // retrying — the worker returns CURL_WRITEFUNC_PAUSE and schedules a curl_easy_pause
 // resume after that delay (the "precision" layer on top of CURLOPT_MAX_RECV_SPEED_LARGE).
 //
 // This header compiles standalone.
 #ifndef VDM_RATE_TOKEN_BUCKET_HPP
 #define VDM_RATE_TOKEN_BUCKET_HPP
 #include <algorithm>
 #include <chrono>
 #include <cstdint>
 #include <mutex>
 #include <optional>
 #include <unordered_map>
 #include "vdm/ids.hpp"
 namespace vdm::rate {
 // rate_bps == 0 means unlimited: consume() always succeeds and never waits.
 class TokenBucket {
  public:
    TokenBucket() = default;
    // `burst` caps how many tokens accumulate while idle; 0 => 1 second's worth.
    explicit TokenBucket(std::uint64_t rate_bps, std::uint64_t burst = 0) {
        set_rate(rate_bps, burst);
    }
    void set_rate(std::uint64_t rate_bps, std::uint64_t burst = 0) {
        std::lock_guard lk(mu_);
        const bool was_unlimited = rate_ == 0;
        rate_ = rate_bps;
        cap_ = burst ? burst : rate_bps;  // 1 s of burst by default
        // A freshly-limited bucket starts full: you may transfer at burst speed
        // immediately, then it throttles (classic token bucket / IDM behaviour). Lowering
        // an existing limit only clamps down — it never hands out a fresh burst.
        if (rate_bps > 0 && was_unlimited)
            tokens_ = cap_;
        else if (tokens_ > cap_)
            tokens_ = cap_;
        last_ = clock::now();
    }
    [[nodiscard]] std::uint64_t rate() const {
        std::lock_guard lk(mu_);
        return rate_;
    }
    // Consume `n` if available; otherwise consume nothing. Returns the wait until `n`
    // tokens *would* be available (0 when it consumed).
    [[nodiscard]] std::chrono::nanoseconds consume(std::uint64_t n) {
        std::lock_guard lk(mu_);
        if (rate_ == 0)
            return {};
        refill_locked();
        if (tokens_ >= n) {
            tokens_ -= n;
            return {};
        }
        const std::uint64_t deficit = n - tokens_;
        // ns to earn `deficit` tokens at rate_ bytes/s
        return std::chrono::nanoseconds{
            static_cast<std::int64_t>((deficit * 1'000'000'000ull + rate_ - 1) / rate_)};
    }
    // Two-phase for the hierarchy: check every level, then commit on all or none.
    [[nodiscard]] std::chrono::nanoseconds peek(std::uint64_t n) {
        std::lock_guard lk(mu_);
        if (rate_ == 0)
            return {};
        refill_locked();
        if (tokens_ >= n)
            return {};
        const std::uint64_t deficit = n - tokens_;
        return std::chrono::nanoseconds{
            static_cast<std::int64_t>((deficit * 1'000'000'000ull + rate_ - 1) / rate_)};
    }
    void commit(std::uint64_t n) {
        std::lock_guard lk(mu_);
        if (rate_ == 0)
            return;
        tokens_ = tokens_ >= n ? tokens_ - n : 0;
    }
  private:
    using clock = std::chrono::steady_clock;
    void refill_locked() {
        auto now = clock::now();
        auto dt = std::chrono::duration_cast<std::chrono::nanoseconds>(now - last_).count();
        if (dt <= 0)
            return;
        last_ = now;
        // added = rate_ * dt / 1e9, guarding overflow for very long idle gaps
        long double added =
            static_cast<long double>(rate_) * static_cast<long double>(dt) / 1'000'000'000.0L;
        std::uint64_t add =
            added >= static_cast<long double>(cap_) ? cap_ : static_cast<std::uint64_t>(added);
        tokens_ = tokens_ + add > cap_ ? cap_ : tokens_ + add;
    }
    mutable std::mutex mu_;
    std::uint64_t rate_ = 0;
    std::uint64_t cap_ = 0;
    std::uint64_t tokens_ = 0;
    clock::time_point last_ = clock::now();
 };
 // The hierarchy. All limits default to 0 (unlimited). A task with no queue is limited by
 // task + global only.
 class RateLimiter {
  public:
    void set_global_limit(std::uint64_t bps) { global_.set_rate(bps); }
    [[nodiscard]] std::uint64_t global_limit() const { return global_.rate(); }
    void set_queue_limit(QueueId q, std::uint64_t bps) {
        std::lock_guard lk(mu_);
        queues_[q].set_rate(bps);
    }
    void set_task_limit(TaskId t, std::uint64_t bps) {
        std::lock_guard lk(mu_);
        tasks_[t].set_rate(bps);
    }
    void attach_task(TaskId t, std::optional<QueueId> q) {
        std::lock_guard lk(mu_);
        tasks_.try_emplace(t);
        if (q) {
            task_queue_[t] = *q;
            queues_.try_emplace(*q);
        } else {
            task_queue_.erase(t);
        }
    }
    void detach_task(TaskId t) {
        std::lock_guard lk(mu_);
        tasks_.erase(t);
        task_queue_.erase(t);
    }
    // Consume `n` bytes against task, queue and global. 0 => consumed everywhere. > 0 =>
    // consumed nowhere; wait that long and retry. Held under mu_ for its whole duration
    // so a concurrent detach_task() can't invalidate the bucket it is using.
    [[nodiscard]] std::chrono::nanoseconds acquire(TaskId t, std::uint64_t n) {
        std::lock_guard lk(mu_);
        TokenBucket *tb = nullptr;
        TokenBucket *qb = nullptr;
        if (auto it = tasks_.find(t); it != tasks_.end())
            tb = &it->second;
        if (auto qit = task_queue_.find(t); qit != task_queue_.end())
            if (auto q = queues_.find(qit->second); q != queues_.end())
                qb = &q->second;
        // peek all, then commit all or none — no level "leaks" tokens on a partial miss.
        std::chrono::nanoseconds wait{};
        if (tb)
            wait = std::max(wait, tb->peek(n));
        if (qb)
            wait = std::max(wait, qb->peek(n));
        wait = std::max(wait, global_.peek(n));
        if (wait.count() > 0)
            return wait;
        if (tb)
            tb->commit(n);
        if (qb)
            qb->commit(n);
        global_.commit(n);
        return {};
    }
  private:
    mutable std::mutex mu_;  // guards the maps AND serialises acquire()
    TokenBucket global_;
    std::unordered_map<QueueId, TokenBucket> queues_;
    std::unordered_map<TaskId, TokenBucket> tasks_;
    std::unordered_map<TaskId, QueueId> task_queue_;
 };
 }  // namespace vdm::rate
 #endif  // VDM_RATE_TOKEN_BUCKET_HPP
@@ -0,0 +1,218 @@
 // vdm/task/download.hpp — the public download API: what DAEMON hands the engine and how
 // the engine reports back. REVIEW SKETCH (stage 7 pre-work) — value types are final
 // enough to build against; Engine/DownloadHandle bodies land in stage 8.
 //
 // See core/docs/engine-api-m1.md for the threading, lifetime, and pause/resume/cancel
 // contract that goes with these signatures.
 //
 // This header compiles standalone.
 #ifndef VDM_TASK_DOWNLOAD_HPP
 #define VDM_TASK_DOWNLOAD_HPP
 #include <chrono>
 #include <cstdint>
 #include <functional>
 #include <memory>
 #include <optional>
 #include <string>
 #include <vector>
 #include "vdm/ids.hpp"
 #include "vdm/net/http_types.hpp"
 #include "vdm/net/probe.hpp"
 #include "vdm/segment/segmenter.hpp"
 #include "vdm/util/error.hpp"
 #include "vdm/util/result.hpp"
 namespace vdm {
 class Engine;  // owns and fills DownloadHandle (see vdm/engine.hpp)
 }  // namespace vdm
 namespace vdm::task {
 // --- input ---------------------------------------------------------------------------
 struct Checksum {
    enum class Algo { md5, sha1, sha256, sha512 };  // matches the wire Checksum set
    Algo algo = Algo::sha256;
    std::string hex;  // lower-case, no separators
 };
 // Everything the engine needs to run ONE download. DAEMON has already run the rules
 // engine, canonicalised the path, checked it against the allowed roots, resolved the
 // filename, and created the parent directory — `save_path` is absolute and final and its
 // directory exists. `<save_path>.veloxpart` and `<save_path>.veloxpart.meta` live beside
 // it during the transfer; on success the part file is renamed in place. If the directory
 // is missing at open time the task fails with Error::path_rejected.
 struct DownloadSpec {
    std::string url;
    std::vector<std::string> mirrors;  // alternative URLs for the same bytes
    std::vector<net::HeaderField> headers;  // the browser's, verbatim
    std::vector<net::Cookie> cookies;
    std::string referrer;
    std::string user_agent;
    std::string save_path;  // absolute; the engine never canonicalises or root-checks
    std::optional<std::uint32_t> segments;      // requested 1..32; nullopt => engine default
    std::optional<std::uint64_t> buffer_bytes;  // requested per segment; nullopt => default
    net::ProxyConfig proxy;
    net::AuthConfig auth;  // credentials known up front (e.g. from the Secret Service);
                           // leave scheme == none to be prompted on a 401/407
    std::optional<Checksum> checksum;  // verified during `verifying`; mismatch => failed
    // DAEMON usually probed already for the File Info dialog. Pass it to skip a second
    // probe; the engine still revalidates on resume. nullopt => the engine probes.
    std::optional<net::ProbeResult> probe_hint;
    bool allow_resume = true;         // if a valid .veloxpart.meta sits beside save_path, resume
                                      // from it; false starts fresh and overwrites
    std::optional<long> max_retries;  // per-segment; nullopt => engine default (10)
 };
 // --- lifecycle (the CORE-owned subset of the wire TaskState; ADR 0013 §1) -------------
 enum class EngineState {
    probing,
    connecting,
    downloading,
    paused,      // shared with DAEMON; entered by either side, idempotently
    retry_wait,  // the engine's own backoff timer
    assembling,  // no-op rename in M1; a real mux step for HLS/DASH (M4)
    verifying,   // checksum
    complete,    // terminal
    failed,      // terminal
    cancelled,   // terminal; always DAEMON- or user-initiated
 };
 [[nodiscard]] constexpr bool is_terminal(EngineState s) noexcept {
    return s == EngineState::complete || s == EngineState::failed || s == EngineState::cancelled;
 }
 // --- progress ---------------------------------------------------------------------------
 struct SegmentProgress {
    std::uint32_t index = 0;
    std::uint64_t start = 0;
    std::uint64_t end = 0;  // inclusive
    std::uint64_t completed = 0;
    std::uint64_t speed_bps = 0;
    segment::SegState state = segment::SegState::idle;
 };
 struct Progress {
    std::uint64_t downloaded = 0;
    std::optional<std::uint64_t> total;  // absent for a chunked source until it ends
    std::uint64_t speed_bps = 0;         // aggregate over the last window
    std::optional<std::uint32_t> eta_seconds;
    std::uint32_t effective_segments = 0;      // slots the budget granted (held)
    std::uint64_t effective_buffer_bytes = 0;  // per segment, after the maxTotal clamp
    std::vector<SegmentProgress> segments;
 };
 // --- interaction callbacks ----------------------------------------------------------
 // A 401/407. The task has already auto-paused (state -> paused, error == auth_required).
 // DAEMON collects credentials and calls handle.provide_auth().
 struct AuthChallenge {
    std::string host;
    std::string realm;
    enum class Scheme { basic, digest, ntlm, negotiate, unknown };
    Scheme scheme = Scheme::unknown;
 };
 // The server's copy changed under us (a 200 where a 206 was expected, or an If-Range /
 // ETag mismatch on resume — docs/04 §5), or the range metadata went stale (416). The
 // task has auto-paused. DAEMON asks the user and calls handle.decide().
 struct DecisionRequest {
    enum class Kind { server_file_changed, range_metadata_stale };
    Kind kind = Kind::server_file_changed;
    std::string detail;  // human-readable, for the dialog body
 };
 enum class Decision {
    restart,       // discard the partial file, download again from scratch
    keep_partial,  // trust what is on disk and continue (the user's risk)
    abort,         // give up: the task goes to `failed`
 };
 struct DownloadOutcome {
    std::string final_path;
    std::uint64_t bytes = 0;
    std::optional<std::string> sha256_hex;  // present when a checksum was requested/derived
    std::chrono::milliseconds elapsed{0};
 };
 // All callbacks are optional. See core/docs/engine-api-m1.md for the rules; in short:
 // they arrive on an engine thread, are serialised per task, must not block, and must not
 // re-enter THIS task's handle synchronously.
 struct DownloadCallbacks {
    // Coalesced to <= 4 Hz per task (matches the wire event.task.progress cadence).
    std::function<void(const Progress &)> on_progress;
    // Every lifecycle transition, including the auto-pauses above (to == paused with a
    // populated ErrorInfo) and terminals.
    std::function<void(EngineState from, EngineState to, const std::optional<ErrorInfo> &)>
        on_state;
    std::function<void(const AuthChallenge &)> on_auth_required;
    std::function<void(const DecisionRequest &)> on_decision_needed;
    // Fired exactly once, last. Success carries the outcome; failure carries the mapped
    // ErrorInfo. After it returns the engine makes no further callbacks for this task and
    // the handle's control methods become no-ops.
    std::function<void(Result<DownloadOutcome>)> on_finished;
 };
 // --- the handle -------------------------------------------------------------------------
 // Copyable (shared state). Every method is safe to call from any thread; each posts to
 // the engine and returns immediately. Dropping the last handle does NOT cancel the task —
 // call cancel() for that. Bodies land in stage 8.
 class DownloadHandle {
  public:
    DownloadHandle() = default;
    [[nodiscard]] TaskId id() const noexcept;
    [[nodiscard]] bool valid() const noexcept { return static_cast<bool>(state_); }
    // Idempotent. pause() on an already-paused or terminal task is a no-op (no error);
    // likewise resume() on a task that is not paused. The resulting state is observed via
    // on_state / this->state(), never a return value (ADR 0013 §2).
    void pause();
    void resume();
    // Idempotent, terminal. discard_partial also removes the .veloxpart[.meta] files.
    // Always fires on_state(_, cancelled, nullopt) then on_finished(Err{Error::canceled}),
    // in that order. `download.cancel` == cancel(false); `download.remove` == cancel(true)
    // (plus DAEMON's own row/file cleanup).
    void cancel(bool discard_partial = false);
    // Only act while the task is awaiting the matching input (auto-paused for auth /
    // decision); otherwise a no-op. `remember` asks DAEMON to persist to the Secret
    // Service — the engine never stores a credential.
    void provide_auth(std::string username, std::string password, bool remember);
    void decide(Decision d);
    // IDM's "Refresh Download Address": swap the URL (e.g. a fresh signed URL) on a live
    // or paused task without losing progress. Empty `headers` keeps the current ones.
    void refresh_url(std::string url, std::vector<net::HeaderField> headers = {});
    // Synchronous snapshots — cheap, lock-guarded, safe any time.
    [[nodiscard]] EngineState state() const;
    [[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_;
 };
 }  // namespace vdm::task
 #endif  // VDM_TASK_DOWNLOAD_HPP
@@ -30,6 +30,8 @@ vdm_add_test(veloxcore_write_buffer_test io/write_buffer_test.cpp)
 vdm_add_test(veloxcore_veloxpart_test meta/veloxpart_test.cpp)
 vdm_add_test(veloxcore_segmenter_test segment/segmenter_test.cpp)
 vdm_add_test(veloxcore_budget_test segment/budget_test.cpp)
 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)
 foreach(net_it http_client probe)
@@ -0,0 +1,145 @@
 #include "vdm/rate/token_bucket.hpp"
 #include <atomic>
 #include <chrono>
 #include <thread>
 #include <vector>
 #include "vtest.hpp"
 using namespace vdm;
 using namespace vdm::rate;
 using namespace std::chrono_literals;
 namespace {
 TaskId tid(std::uint64_t v) {
    return TaskId{v};
 }
 QueueId qid(std::uint64_t v) {
    return QueueId{v};
 }
 }  // namespace
 VT_TEST(tb_unlimited_never_waits) {
    TokenBucket b(0);
    for (int i = 0; i < 1000; ++i)
        VT_CHECK_EQ(b.consume(1'000'000).count(), 0);
 }
 VT_TEST(tb_burst_then_throttle) {
    // 1000 B/s, default burst = 1 s = 1000 tokens.
    TokenBucket b(1000);
    VT_CHECK_EQ(b.consume(1000).count(), 0);  // drains the burst
    auto w = b.consume(1000);                 // empty now: must wait ~1 s
    VT_CHECK(w >= 900ms && w <= 1100ms);
 }
 VT_TEST(tb_refills_over_time) {
    TokenBucket b(10'000, /*burst=*/10'000);
    VT_CHECK_EQ(b.consume(10'000).count(), 0);
    std::this_thread::sleep_for(120ms);  // ~1200 tokens back
    auto w = b.consume(1000);
    VT_CHECK_EQ(w.count(), 0);  // affordable from the refill
    auto w2 = b.consume(5000);
    VT_CHECK(w2.count() > 0);  // not that much yet
 }
 VT_TEST(tb_burst_caps_accumulation) {
    TokenBucket b(1000, /*burst=*/2000);
    std::this_thread::sleep_for(100ms);  // idle far longer than burst/rate would fill
    std::this_thread::sleep_for(100ms);
    VT_CHECK_EQ(b.consume(2000).count(), 0);  // at most the 2000 cap accumulated
    VT_CHECK(b.consume(1).count() > 0);       // and no more
 }
 VT_TEST(tb_set_rate_zero_makes_unlimited) {
    TokenBucket b(1000);
    VT_CHECK_EQ(b.consume(1000).count(), 0);
    VT_CHECK(b.consume(1000).count() > 0);
    b.set_rate(0);
    VT_CHECK_EQ(b.consume(1'000'000).count(), 0);
 }
 // --- the hierarchy --------------------------------------------------------------------
 VT_TEST(rl_all_unlimited_by_default) {
    RateLimiter rl;
    rl.attach_task(tid(1), std::nullopt);
    for (int i = 0; i < 100; ++i)
        VT_CHECK_EQ(rl.acquire(tid(1), 1'000'000).count(), 0);
 }
 VT_TEST(rl_tightest_level_binds) {
    RateLimiter rl;
    rl.set_global_limit(100'000);
    rl.set_queue_limit(qid(9), 20'000);
    rl.set_task_limit(tid(1), 50'000);
    rl.attach_task(tid(1), qid(9));
    // burst: task 50k, queue 20k, global 100k -> the queue's 20k is the ceiling
    VT_CHECK_EQ(rl.acquire(tid(1), 20'000).count(), 0);
    auto w = rl.acquire(tid(1), 5'000);
    VT_CHECK(w.count() > 0);  // queue bucket is dry even though task & global aren't
 }
 VT_TEST(rl_no_partial_consumption_on_miss) {
    RateLimiter rl;
    rl.set_global_limit(1'000'000);   // plenty
    rl.set_task_limit(tid(1), 1000);  // 1 s burst
    rl.attach_task(tid(1), std::nullopt);
    VT_CHECK_EQ(rl.acquire(tid(1), 1000).count(), 0);  // drain the task bucket
    for (int i = 0; i < 5; ++i)
        VT_CHECK(rl.acquire(tid(1), 1000).count() > 0);  // task bucket blocks, repeatedly
    // global must NOT have been charged for any of those blocked attempts: a fresh task
    // limited only by the global bucket can still spend nearly its whole burst (only the
    // one *successful* 1000-byte acquire above was charged).
    rl.attach_task(tid(2), std::nullopt);
    VT_CHECK_EQ(rl.acquire(tid(2), 990'000).count(), 0);
 }
 VT_TEST(rl_detach_then_acquire_is_safe_and_unlimited) {
    RateLimiter rl;
    rl.set_task_limit(tid(1), 1000);
    rl.attach_task(tid(1), std::nullopt);
    VT_CHECK_EQ(rl.acquire(tid(1), 1000).count(), 0);
    rl.detach_task(tid(1));
    // unknown task -> no task/queue bucket, only global (unlimited here)
    VT_CHECK_EQ(rl.acquire(tid(1), 1'000'000).count(), 0);
 }
 VT_TEST(rl_enforces_aggregate_rate_under_load) {
    RateLimiter rl;
    const std::uint64_t rate = 4'000'000;  // 4 MB/s global
    rl.set_global_limit(rate);
    for (std::uint64_t i = 1; i <= 8; ++i)
        rl.attach_task(tid(i), std::nullopt);
    std::atomic<std::uint64_t> moved{0};
    auto t0 = std::chrono::steady_clock::now();
    std::vector<std::jthread> ws;
    for (std::uint64_t i = 1; i <= 8; ++i) {
        ws.emplace_back([&, id = tid(i)] {
            for (int k = 0; k < 400; ++k) {
                std::uint64_t chunk = 16 * 1024;
                for (;;) {
                    auto w = rl.acquire(id, chunk);
                    if (w.count() == 0)
                        break;
                    std::this_thread::sleep_for(
                        std::min<std::chrono::nanoseconds>(w, std::chrono::milliseconds(20)));
                }
                moved.fetch_add(chunk);
            }
        });
    }
    ws.clear();  // join
    auto secs = std::chrono::duration<double>(std::chrono::steady_clock::now() - t0).count();
    double effective = moved.load() / secs;
    // Allow one burst's worth of slop plus scheduling noise: effective rate should sit
    // within ~2x of the configured limit, never wildly above.
    VT_CHECK(effective <= rate * 2.5);
    VT_CHECK(moved.load() == 8u * 400u * 16u * 1024u);
 }
@@ -0,0 +1,84 @@
 // The engine API sketch must compile and its value types must behave. Engine /
 // DownloadHandle bodies land in stage 8; this only exercises the data shapes DAEMON
 // builds against.
 #include "vdm/engine.hpp"
 #include "vdm/task/download.hpp"
 #include <type_traits>
 #include "vtest.hpp"
 using namespace vdm;
 using namespace vdm::task;
 VT_TEST(api_download_spec_defaults) {
    DownloadSpec s;
    s.url = "https://example.com/big.iso";
    s.save_path = "/home/u/Downloads/big.iso";
    VT_CHECK(s.mirrors.empty());
    VT_CHECK(!s.segments.has_value());
    VT_CHECK(!s.buffer_bytes.has_value());
    VT_CHECK(!s.checksum.has_value());
    VT_CHECK(!s.probe_hint.has_value());
    VT_CHECK(s.allow_resume);
    VT_CHECK(s.proxy.kind == net::ProxyKind::none);
    VT_CHECK(s.auth.scheme == net::AuthScheme::none);
 }
 VT_TEST(api_state_helpers) {
    VT_CHECK(is_terminal(EngineState::complete));
    VT_CHECK(is_terminal(EngineState::failed));
    VT_CHECK(is_terminal(EngineState::cancelled));
    VT_CHECK(!is_terminal(EngineState::paused));
    VT_CHECK(!is_terminal(EngineState::downloading));
 }
 VT_TEST(api_callbacks_are_all_optional) {
    DownloadCallbacks cb;  // every std::function default-constructs empty
    VT_CHECK(!cb.on_progress);
    VT_CHECK(!cb.on_state);
    VT_CHECK(!cb.on_auth_required);
    VT_CHECK(!cb.on_decision_needed);
    VT_CHECK(!cb.on_finished);
    cb.on_state = [](EngineState, EngineState, const std::optional<vdm::ErrorInfo> &) {};
    cb.on_finished = [](Result<DownloadOutcome>) {};
    VT_CHECK(cb.on_state && cb.on_finished);
 }
 VT_TEST(api_value_types_roundtrip) {
    Progress p;
    p.downloaded = 1234;
    p.total = 5000;
    p.effective_segments = 4;
    SegmentProgress sp;
    sp.index = 0;
    sp.end = 1249;
    sp.completed = 1234;
    p.segments.push_back(sp);
    VT_CHECK_EQ(p.segments.size(), 1u);
    VT_CHECK_EQ(p.segments[0].end, 1249u);
    DownloadOutcome o;
    o.final_path = "/x";
    o.bytes = 5000;
    VT_CHECK_EQ(o.bytes, 5000u);
    AuthChallenge a;
    a.host = "h";
    a.scheme = AuthChallenge::Scheme::digest;
    VT_CHECK(a.scheme == AuthChallenge::Scheme::digest);
    DecisionRequest d;
    d.kind = DecisionRequest::Kind::server_file_changed;
    d.detail = "changed";
    VT_CHECK(d.kind == DecisionRequest::Kind::server_file_changed);
 }
 VT_TEST(api_handle_and_engine_are_move_only_shaped) {
    static_assert(!std::is_copy_constructible_v<Engine>, "Engine is non-copyable");
    static_assert(std::is_copy_constructible_v<DownloadHandle>, "handle is a shared handle");
    DownloadHandle h;  // default handle is invalid until Engine::start() fills it
    VT_CHECK(!h.valid());
 }