daemon: D2 — download.probe, genuinely async on both transports

download.probe was a stub (-32603). It needs an HTTP round trip on the engine's probe
pool (up to the schema's 30s x-deadlineMs), which cannot fit VeloxDispatcher's
synchronous on_download_probe -> HandlerResult<T> return without blocking the RPC
loop for the duration — a hard no per CLAUDE.md ("never block the RPC loop") and
AGENT-DAEMON.md build step 1.

uds_server.cpp and ws_server.cpp special-case "download.probe" before the generic
dispatch(), exactly the way they already special-case session.hello/session.subscribe:
parse the params, call the port, and queue the reply whenever the callback fires
(dropped silently if the connection is gone by then).

rpc::TaskActionPort gains probe_now(DownloadProbeParams, callback) — kept in proto/std
terms, no vdm::net::* in the signature, so veloxd_rpc never needs core/include's vdm
headers just to declare this. sched::Scheduler::probe_now is the implementation:
builds a vdm::net::ProbeRequest, runs it on the engine's probe pool, marshals the
engine-thread callback back onto the loop (deps_.post_to_loop, same as every other
engine callback here), maps a probe failure to -32013 ProbeFailed (data.httpStatus set
when there was an HTTP response), and fills suggestedCategoryId/suggestedSaveDir with a
plain extension match against the categories table — not the real rules engine, which
is still D3; noted in a comment.

Verified against real veloxd + tools/testserver, not just unit tests: a real probe
answers in ~5ms with size/resumable/etag/redirect chain; a bad host maps to -32013;
and — the actual point of the async design — a connection running a 10s slow-loris
probe does not block a second connection's download.list, which answers in ~1ms while
the probe is still outstanding.

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

daemon/src/rpc/task_action_port.hpp

@@ -12,9 +12,12 @@
 // rpc/, so adding an rpc-defined base costs nothing new); main.cpp hands the dispatcher a
 // `TaskActionPort*` pointing at the same Scheduler it constructs.
 
+#include <functional>
 #include <string>
 #include <vector>
 
+#include "velox_proto.hpp"
+
 namespace velox::daemon::rpc {
 
 class TaskActionPort {
@@ -39,6 +42,22 @@ public:
     // queue.stop(pauseRunning=true): pause every currently-running task in `queue_id` now.
     // Returns the wire ids actually paused.
     virtual std::vector<std::string> pause_queue(const std::string& queue_id) = 0;
+
+    // download.probe (D2), the File Info dialog's own network round trip — no task row
+    // involved. Genuinely async (the engine's probe pool; up to the schema's 30s
+    // x-deadlineMs) and so cannot fit VeloxDispatcher's synchronous on_download_probe:
+    // the RPC server layer (uds_server.cpp / ws_server.cpp) special-cases "download.probe"
+    // before the generic dispatch(), the same way it already special-cases session.hello,
+    // calls this, and queues the reply whenever `done` fires — on an engine thread, so the
+    // implementation must marshal back to the loop before calling it, the same as every
+    // other EnginePort callback. Kept in std::string/proto terms (not vdm::net::*) so this
+    // header — included by dispatcher.hpp, part of veloxd_rpc — never needs core/include's
+    // vdm headers; the vdm::net::ProbeRequest/ProbeResult conversion lives in sched/, which
+    // already depends on vdm.
+    virtual void probe_now(
+        const velox::proto::DownloadProbeParams& params,
+        std::function<void(velox::proto::HandlerResult<velox::proto::DownloadProbeResult>)>
+            done) = 0;
 };
 
 }  // namespace velox::daemon::rpc

daemon/src/rpc/uds_server.cpp

@@ -57,8 +57,9 @@ json rpc_error(const json& id, proto::ErrorCode code, std::string_view msg, json
 }  // namespace
 
 UdsServer::UdsServer(EventLoop& loop, proto::Dispatcher& dispatcher, EventHub& hub,
-                    std::string socket_path)
-    : loop_(loop), dispatcher_(dispatcher), hub_(hub), path_(std::move(socket_path)) {}
+                    std::string socket_path, TaskActionPort* actions)
+    : loop_(loop), dispatcher_(dispatcher), hub_(hub), actions_(actions),
+      path_(std::move(socket_path)) {}
 
 UdsServer::~UdsServer() {
     for (auto& [fd, c] : conns_) {
@@ -202,12 +203,47 @@ void UdsServer::handle_line(Conn& c, const std::string& line) {
         }
     }
 
+    if (method == "download.probe") {
+        handle_download_probe(c, req);
+        return;
+    }
+
     // Everything else: the generated router. It returns a null json for a notification
     // that needs no reply.
     json reply = proto::dispatch(dispatcher_, proto::Transport::Uds, req);
     if (!reply.is_null()) queue_reply(c, reply);
 }
 
+void UdsServer::handle_download_probe(Conn& c, const json& request) {
+    const json id = request.contains("id") ? request.at("id") : json(nullptr);
+    const json params_json = request.contains("params") ? request.at("params") : json::object();
+
+    auto parsed = proto::parse<proto::DownloadProbeParams>(params_json, "params");
+    if (!parsed) {
+        queue_reply(c, rpc_error(id, proto::ErrorCode::InvalidParams, parsed.error().message,
+                                 json{{"path", parsed.error().path}}));
+        return;
+    }
+    if (!actions_) {
+        queue_reply(c, rpc_error(id, proto::ErrorCode::InternalError,
+                                 "not implemented in this build: download.probe"));
+        return;
+    }
+
+    const int fd = c.fd;
+    actions_->probe_now(
+        *parsed, [this, fd, id](proto::HandlerResult<proto::DownloadProbeResult> r) {
+            auto it = conns_.find(fd);
+            if (it == conns_.end()) return;  // client gone while the probe was outstanding
+            if (r) {
+                queue_reply(*it->second, proto::make_result(id, *r));
+            } else {
+                queue_reply(*it->second,
+                           rpc_error(id, r.error().code, r.error().message, r.error().data));
+            }
+        });
+}
+
 bool UdsServer::handle_session_method(Conn& c, const std::string& method, const json& request,
                                       json& reply) {
     const json id = request.contains("id") ? request.at("id") : json(nullptr);

daemon/src/rpc/uds_server.hpp

@@ -22,6 +22,7 @@
 
 #include "rpc/event_hub.hpp"
 #include "rpc/ndjson.hpp"
+#include "rpc/task_action_port.hpp"
 #include "velox_proto.hpp"
 
 namespace velox::daemon::rpc {
@@ -30,8 +31,11 @@ class EventLoop;
 
 class UdsServer {
 public:
+    // `actions` is optional (nullptr in tests that don't need download.probe) — see
+    // handle_download_probe's own comment for why this method can't go through the
+    // generic dispatch() path like everything else.
     UdsServer(EventLoop& loop, velox::proto::Dispatcher& dispatcher, EventHub& hub,
-             std::string socket_path);
+             std::string socket_path, TaskActionPort* actions = nullptr);
     ~UdsServer();
 
     UdsServer(const UdsServer&) = delete;
@@ -66,6 +70,13 @@ private:
     bool handle_session_method(Conn& c, const std::string& method, const nlohmann::json& request,
                                nlohmann::json& reply);
 
+    // download.probe is genuinely async (up to the schema's 30s x-deadlineMs, on the
+    // engine's probe pool) and so cannot fit the synchronous generic dispatch() path —
+    // special-cased here exactly the way handle_session_method special-cases session.*.
+    // Queues the reply itself, later, when actions_->probe_now()'s callback fires; does
+    // nothing if the connection is gone by then (client disconnected mid-probe).
+    void handle_download_probe(Conn& c, const nlohmann::json& request);
+
     void queue_reply(Conn& c, const nlohmann::json& reply);
     void flush(Conn& c);
     void close_conn(int fd);
@@ -73,6 +84,7 @@ private:
     EventLoop& loop_;
     velox::proto::Dispatcher& dispatcher_;
     EventHub& hub_;
+    TaskActionPort* actions_;
     std::string path_;
     int listen_fd_ = -1;
     bool bound_ = false;   // path_ is ours to unlink on destruction

daemon/src/rpc/ws_server.cpp

@@ -60,12 +60,14 @@ json rpc_error(const json& id, proto::ErrorCode code, std::string_view msg, json
 }  // namespace
 
 WsServer::WsServer(EventLoop& loop, proto::Dispatcher& dispatcher, store::Db& db,
-                   PairingApprover& approver, EventHub& hub, RuntimeDir runtime)
+                   PairingApprover& approver, EventHub& hub, RuntimeDir runtime,
+                   TaskActionPort* actions)
     : loop_(loop),
       dispatcher_(dispatcher),
       db_(db),
       approver_(approver),
       hub_(hub),
+      actions_(actions),
       runtime_(std::move(runtime)) {}
 
 WsServer::~WsServer() {
@@ -261,10 +263,45 @@ void WsServer::handle_rpc(Conn& c, const std::string& text) {
         return;
     }
 
+    if (method == "download.probe") {
+        handle_download_probe(c, req);
+        return;
+    }
+
     json reply = proto::dispatch(dispatcher_, proto::Transport::Ws, req);
     if (!reply.is_null()) send_text(c, reply);
 }
 
+void WsServer::handle_download_probe(Conn& c, const json& request) {
+    const json id = request.contains("id") ? request.at("id") : json(nullptr);
+    const json params_json = request.contains("params") ? request.at("params") : json::object();
+
+    auto parsed = proto::parse<proto::DownloadProbeParams>(params_json, "params");
+    if (!parsed) {
+        send_text(c, rpc_error(id, proto::ErrorCode::InvalidParams, parsed.error().message,
+                               json{{"path", parsed.error().path}}));
+        return;
+    }
+    if (!actions_) {
+        send_text(c, rpc_error(id, proto::ErrorCode::InternalError,
+                               "not implemented in this build: download.probe"));
+        return;
+    }
+
+    const int fd = c.fd;
+    actions_->probe_now(
+        *parsed, [this, fd, id](proto::HandlerResult<proto::DownloadProbeResult> r) {
+            auto it = conns_.find(fd);
+            if (it == conns_.end()) return;  // client gone while the probe was outstanding
+            if (r) {
+                send_text(*it->second, proto::make_result(id, *r));
+            } else {
+                send_text(*it->second,
+                         rpc_error(id, r.error().code, r.error().message, r.error().data));
+            }
+        });
+}
+
 bool WsServer::handle_session_ws(Conn& c, const std::string& method, const json& request,
                                  json& reply) {
     const json id = request.contains("id") ? request.at("id") : json(nullptr);

daemon/src/rpc/ws_server.hpp

@@ -18,6 +18,7 @@
 #include "rpc/event_hub.hpp"
 #include "rpc/pairing.hpp"
 #include "rpc/runtime_dir.hpp"
+#include "rpc/task_action_port.hpp"
 #include "rpc/ws_frame.hpp"
 #include "velox_proto.hpp"
 
@@ -31,8 +32,12 @@ class EventLoop;
 
 class WsServer {
 public:
+    // `actions` is optional (nullptr in tests that don't need download.probe) — see
+    // handle_download_probe's own comment for why this method can't go through the
+    // generic dispatch() path like everything else.
     WsServer(EventLoop& loop, velox::proto::Dispatcher& dispatcher, velox::daemon::store::Db& db,
-             PairingApprover& approver, EventHub& hub, RuntimeDir runtime);
+             PairingApprover& approver, EventHub& hub, RuntimeDir runtime,
+             TaskActionPort* actions = nullptr);
     ~WsServer();
 
     WsServer(const WsServer&) = delete;
@@ -73,6 +78,10 @@ private:
     bool handle_session_ws(Conn& c, const std::string& method, const nlohmann::json& request,
                            nlohmann::json& reply);
 
+    // See UdsServer::handle_download_probe — same reasoning, same pattern, duplicated per
+    // transport because each owns its own Conn/send mechanics.
+    void handle_download_probe(Conn& c, const nlohmann::json& request);
+
     void send_text(Conn& c, const nlohmann::json& value);
     void send_frame(Conn& c, WsOpcode op, std::string_view payload);
     void begin_close(Conn& c, std::uint16_t code, std::string_view reason);
@@ -84,6 +93,7 @@ private:
     velox::daemon::store::Db& db_;
     PairingApprover& approver_;
     EventHub& hub_;
+    TaskActionPort* actions_;
     RuntimeDir runtime_;
     PairingRateLimiter rate_limiter_;