From 6bfcd5cf3df9a38cf4912f65a9476df5fc3ad8b3 Mon Sep 17 00:00:00 2001 From: dijunkun Date: Fri, 6 Nov 2026 20:36:48 +0800 Subject: [PATCH] [feat] add MCP file browsing and cancellable transfers --- apps/cli/agent.cpp | 12 + apps/cli/docs/automation-protocol.md | 6 + apps/cli/xmake.lua | 3 +- .../file_transfer/file_transfer_manager.h | 2 +- .../src/gui/runtime/automation_runtime.cpp | 2 +- apps/desktop/xmake/targets.lua | 2 +- apps/harmony/desktop-port/CMakeLists.txt | 1 + apps/mcp/README.md | 8 + apps/mcp/src/native.ts | 3 +- apps/mcp/src/server.ts | 28 +- apps/mcp/src/sessions.ts | 28 ++ libs/controller/include/controller_files.h | 31 ++ libs/controller/include/controller_session.h | 3 + .../controller/include}/file_browser.h | 7 +- libs/controller/include/file_transfer_core.h | 10 + libs/controller/src/controller_files.cpp | 343 ++++++++++++++++++ libs/controller/src/controller_requests.cpp | 8 + libs/controller/src/controller_session.cpp | 22 ++ .../controller/src}/file_browser.cpp | 33 +- xmake.lua | 8 + 20 files changed, 537 insertions(+), 23 deletions(-) create mode 100644 libs/controller/include/controller_files.h rename {apps/desktop/src/gui/features/file_transfer => libs/controller/include}/file_browser.h (96%) create mode 100644 libs/controller/src/controller_files.cpp rename {apps/desktop/src/gui/features/file_transfer => libs/controller/src}/file_browser.cpp (96%) diff --git a/apps/cli/agent.cpp b/apps/cli/agent.cpp index 3ffce27d..9283661a 100644 --- a/apps/cli/agent.cpp +++ b/apps/cli/agent.cpp @@ -788,6 +788,18 @@ private: } else if (method == "action") { StartAction(j, id); return false; + } else if (method == "files") { + Fields(j, {"id", "method", "operation", "operation_id", "path", "paths", + "destination", "request_id", "offset"}); + const auto operation = String(j, "operation", 16); + const bool existing = operation == "status" || operation == "cancel" || + operation == "list_result" || operation == "list_cancel"; + if (!allowed_ || (!existing && (paused_ || !session_.ControlEnabled()))) + throw std::invalid_argument("CONTROL_DENIED"); + auto request = j; + request.erase("id"); request.erase("method"); + Reply(id, session_.Files(request)); + return false; } else if (method == "clipboard") { Fields(j, {"id", "method", "operation", "operation_id", "text"}); if (!allowed_ || paused_ || !session_.ControlEnabled()) diff --git a/apps/cli/docs/automation-protocol.md b/apps/cli/docs/automation-protocol.md index 74ef1560..10da036c 100644 --- a/apps/cli/docs/automation-protocol.md +++ b/apps/cli/docs/automation-protocol.md @@ -145,3 +145,9 @@ Updated desktop hosts answer the versioned `automation` capability handshake. Wh ## Clipboard `clipboard` accepts `operation: get | set`. Writes additionally require `operation_id` and `text` (up to 16384 UTF-8 bytes, including an empty string to clear). Reads return up to 128 KiB of text. The existing chunked `web-clipboard` stream carries values and applied/rejected replies. Additive `read`, `empty`, and `clear` frames are enabled by the remote `clipboard` capability. Clipboard access requires control permission and respects human pause. Cancellation of an issued write reports an uncertain effect. Repeated write operation IDs retrieve the cached result. + +## File operations + +`files` supports `list`, `list_result`, `list_cancel`, `upload`, `download`, `status`, and `cancel`. Listing requests return a numeric `request_id`. Transfer requests require an `operation_id`, absolute `paths`, and an existing absolute `destination` directory. Progress and listing pages use `offset` and return `next_offset` with at most 256 entries. + +The desktop and CLI share `FileBrowser` and the existing file transfer engine. Preparation and completion messages carry a transfer identifier to bind incoming files to the requested destination. A cancelled transfer cleans unfinished staging files. The capability handshake advertises `files` on compatible hosts. Queries and cancellation of existing tasks remain available during human pause. diff --git a/apps/cli/xmake.lua b/apps/cli/xmake.lua index 99aa5c96..b4f02b2a 100644 --- a/apps/cli/xmake.lua +++ b/apps/cli/xmake.lua @@ -12,12 +12,13 @@ end target("crossdesk_controller") set_kind("static") - add_deps("minirtc", "crossdesk_wire") + add_deps("minirtc", "crossdesk_wire", "crossdesk_file_browser") add_packages("spdlog", "libpng", "libyuv") add_includedirs(path.join(controller_dir, "include"), path.join(minirtc_dir, "src/api"), {public = true}) add_includedirs(path.join(minirtc_dir, "src/log")) add_files(path.join(controller_dir, "src/*.cpp")) + remove_files(path.join(controller_dir, "src/file_browser.cpp")) target_end() target("crossdesk_terminal") diff --git a/apps/desktop/src/gui/features/file_transfer/file_transfer_manager.h b/apps/desktop/src/gui/features/file_transfer/file_transfer_manager.h index 5d3fdd1c..f46e2ff4 100644 --- a/apps/desktop/src/gui/features/file_transfer/file_transfer_manager.h +++ b/apps/desktop/src/gui/features/file_transfer/file_transfer_manager.h @@ -13,7 +13,7 @@ #include #include "file_transfer.h" -#include "features/file_transfer/file_browser.h" +#include #include "runtime/remote_session.h" namespace crossdesk { diff --git a/apps/desktop/src/gui/runtime/automation_runtime.cpp b/apps/desktop/src/gui/runtime/automation_runtime.cpp index 2d2bad0a..6c82a833 100644 --- a/apps/desktop/src/gui/runtime/automation_runtime.cpp +++ b/apps/desktop/src/gui/runtime/automation_runtime.cpp @@ -21,7 +21,7 @@ void GuiRuntime::HandleAutomation(const std::string& remote_id, try { const auto method = request.value("method", std::string{}); if (method == "capabilities") { - reply(true, {{"input_ack", true}, {"clipboard", true}}); + reply(true, {{"input_ack", true}, {"clipboard", true}, {"files", true}}); } else if (method == "input") { RemoteAction action{}; if (!request.contains("arguments") || diff --git a/apps/desktop/xmake/targets.lua b/apps/desktop/xmake/targets.lua index 9aeeab4d..e288be2a 100644 --- a/apps/desktop/xmake/targets.lua +++ b/apps/desktop/xmake/targets.lua @@ -608,7 +608,7 @@ function setup_targets() "single_instance", "path_manager", "screen_capturer", "speaker_capturer", "device_controller", "thumbnail", "version_checker", "tools", - "crossdesk_wire", "privacy", "crossdesk_ui") + "crossdesk_wire", "crossdesk_file_browser", "privacy", "crossdesk_ui") add_files("apps/desktop/src/gui/render.cpp", "apps/desktop/src/gui/application/gui_application.cpp", "apps/desktop/src/gui/rendering/*.cpp", "apps/desktop/src/gui/runtime/*.cpp", diff --git a/apps/harmony/desktop-port/CMakeLists.txt b/apps/harmony/desktop-port/CMakeLists.txt index 2356b426..644ccca6 100644 --- a/apps/harmony/desktop-port/CMakeLists.txt +++ b/apps/harmony/desktop-port/CMakeLists.txt @@ -7,6 +7,7 @@ file(GLOB_RECURSE FEATURES CONFIGURE_DEPENDS "${DESKTOP}/gui/features/*.cpp") file(GLOB COMMON CONFIGURE_DEPENDS "${DESKTOP}/common/*.cpp" "${DESKTOP}/thumbnail/*.cpp" "${DESKTOP}/version_checker/*.cpp" "${CROSSDESK_ROOT}/libs/wire/src/*.cpp") target_sources(slint_desktop PRIVATE ${RUNTIME} ${FEATURES} ${COMMON} + "${CROSSDESK_ROOT}/libs/controller/src/file_browser.cpp" "${DESKTOP}/gui/application/gui_application.cpp" "${DESKTOP}/gui/rendering/slint_video_presenter.cpp" "${DESKTOP}/platform/common/gui/service_status_runtime.cpp" diff --git a/apps/mcp/README.md b/apps/mcp/README.md index fc6da058..7c37b19a 100644 --- a/apps/mcp/README.md +++ b/apps/mcp/README.md @@ -254,3 +254,11 @@ npm run pack:npm -- --out-dir /absolute/path/artifacts ### 远端剪贴板 `crossdesk_clipboard_get` 读取远端文本剪贴板，`crossdesk_clipboard_set` 设置文本，空字符串清空。读取上限为 128 KiB，单次工具写入上限为 16 KiB UTF-8。写入需要 `operation_id`，相同请求查询原结果。工具使用已授权控制会话，并遵守人工暂停及解锁等待状态。更新后的被控端通过 `capabilities.clipboard` 声明支持，写入成功由系统剪贴板更新回执确认。 + +### 文件浏览与传输 + +`crossdesk_files_list` 浏览远端目录，空路径选择远端主目录，`@roots` 返回文件系统根目录。每页最多 256 项，使用 `request_id` 和 `next_offset` 继续读取。超时返回可继续查询的请求标识。 + +`crossdesk_files_transfer` 支持上传和下载文件、文件夹，源路径为绝对路径，目标目录需要已存在。传输使用 `operation_id` 去重，通过 `crossdesk_files_status` 查询每个文件的字节进度和终态，通过 `crossdesk_files_cancel` 取消并清理未完成文件。每个连接同时处理一个传输任务，多文件按现有传输引擎排队。已完成任务保留四份历史记录，状态中的文件清单分页返回。目标同名文件按现有冲突命名规则另存。 + +文件操作需要更新后的被控端声明 `capabilities.files`。下载请求绑定目标目录及传输标识，接收器接受已为该请求预留的文件。人工暂停后仍可查询或取消已有任务。 diff --git a/apps/mcp/src/native.ts b/apps/mcp/src/native.ts index 4386db12..c01b7a7c 100644 --- a/apps/mcp/src/native.ts +++ b/apps/mcp/src/native.ts @@ -174,7 +174,8 @@ export class NativeClient { return Promise.reject(new Failure("BUSY", false, true)); const id = ++this.sequence; const sideEffect = method === "action" || method === "unlock" || - (method === "clipboard" && args.operation === "set"); + (method === "clipboard" && args.operation === "set") || + (method === "files" && ["upload", "download", "cancel"].includes(args.operation)); const line = JSON.stringify({ ...args, id, method }) + "\n"; if (Buffer.byteLength(line) > 65536) return Promise.reject(new Failure("MESSAGE_TOO_LARGE")); diff --git a/apps/mcp/src/server.ts b/apps/mcp/src/server.ts index ed7d86cf..31f20003 100644 --- a/apps/mcp/src/server.ts +++ b/apps/mcp/src/server.ts @@ -124,7 +124,8 @@ export function createServer( inputSchema: schema, annotations: { readOnlyHint: !!d.readOnly, - destructiveHint: ["crossdesk_action", "crossdesk_unlock"].includes(d.name), + destructiveHint: ["crossdesk_action", "crossdesk_unlock", "crossdesk_clipboard_set", + "crossdesk_files_transfer", "crossdesk_files_cancel"].includes(d.name), openWorldHint: true, }, ...(d.ui || d.app @@ -316,6 +317,28 @@ export function createServer( "Read the remote text clipboard within the authorized control session.", async (p, ctx) => output(await sessions.clipboard({ ...p, operation: "get" }, ctx.mcpReq.signal)), { readOnly: true }); + const filePath = z.string().max(4096).refine((path) => !path.includes("\0")); + add("crossdesk_files_list", z.strictObject({ session_id: id, path: filePath.default(""), + request_id: z.int().min(1).max(4294967295).optional(), offset: z.int().min(0).default(0) }), + "List a remote directory. Empty path selects the remote home and @roots lists filesystem roots. Results contain up to 256 entries. Continue with request_id and next_offset while present. A timed-out listing remains queryable with its request_id.", + async (p, ctx) => output(await sessions.files({ ...p, operation: "list" }, ctx.mcpReq.signal)), + { readOnly: true }); + add("crossdesk_files_transfer", z.strictObject({ session_id: id, operation_id: id, + direction: z.enum(["upload", "download"]), paths: z.array(filePath.min(1)).min(1).max(256), + destination: filePath.min(1) }), + "Upload local files/folders or download remote files/folders into an existing absolute destination directory. paths are absolute source paths. Returns a transfer operation to monitor with crossdesk_files_status. Reuse the same operation_id and arguments to query the original transfer. Existing destination files receive a distinct name on conflict.", + async (p, ctx) => { + const { direction, ...args } = p; + return output(await sessions.files({ ...args, operation: direction }, ctx.mcpReq.signal)); + }); + add("crossdesk_files_status", z.strictObject({ session_id: id, operation_id: id, + offset: z.int().min(0).default(0) }), + "Get transfer progress, per-file completion and errors. Continue file entries with next_offset. Wait for completed, failed or cancelled before reporting a finished transfer.", + async (p, ctx) => output(await sessions.files({ ...p, operation: "status" }, ctx.mcpReq.signal)), + { readOnly: true }); + add("crossdesk_files_cancel", z.strictObject({ session_id: id, operation_id: id }), + "Cancel a transfer and remove its unfinished staging files. Completed files remain available.", + async (p, ctx) => output(await sessions.files({ ...p, operation: "cancel" }, ctx.mcpReq.signal))); add("crossdesk_clipboard_set", z.strictObject({ session_id: id, operation_id: id, text: z.string().max(16384).refine((text) => Buffer.byteLength(text) <= 16384 && !text.includes("\0")) }), "Set or clear the remote text clipboard. Reuse the same operation_id to query the original result. Returns applied after the remote clipboard update.", @@ -414,7 +437,8 @@ export function createServer( inputSchema, annotations: { readOnlyHint: !!d.readOnly, - destructiveHint: ["crossdesk_action", "crossdesk_unlock"].includes(d.name), + destructiveHint: ["crossdesk_action", "crossdesk_unlock", "crossdesk_clipboard_set", + "crossdesk_files_transfer", "crossdesk_files_cancel"].includes(d.name), openWorldHint: true, }, ...((d.ui || d.app) && apps(ctx) diff --git a/apps/mcp/src/sessions.ts b/apps/mcp/src/sessions.ts index b29ead83..0e308bf7 100644 --- a/apps/mcp/src/sessions.ts +++ b/apps/mcp/src/sessions.ts @@ -473,6 +473,34 @@ export class Sessions { if (s.unlock && this.unlockPending(s.unlock)) throw new Failure("UNLOCK_PENDING"); return s.native.request("clipboard", args, 7000, signal); } + async files(p: Data, signal?: AbortSignal): Promise { + const { session_id, ...args } = p; + const s = this.session(session_id); + this.checkUnlock(s); + if (s.unlock && this.unlockPending(s.unlock) && !["status", "cancel"].includes(args.operation)) + throw new Failure("UNLOCK_PENDING"); + if (args.operation !== "list") return s.native.request("files", args, 7000, signal); + const requestId = args.request_id ?? (await s.native.request("files", + { operation: "list", path: args.path ?? "" }, 7000, signal)).request_id; + try { + const deadline = Date.now() + 25000; + while (true) { + const result = await s.native.request("files", { + operation: "list_result", request_id: requestId, offset: args.offset ?? 0, + }, 7000, signal); + if (result.state !== "pending") { + if (result.next_offset == null) + await s.native.request("files", { operation: "list_cancel", request_id: requestId }); + return result; + } + if (Date.now() >= deadline) return { request_id: requestId, state: "pending", wait_timed_out: true }; + await delay(100, undefined, { signal }); + } + } catch (error) { + await s.native.request("files", { operation: "list_cancel", request_id: requestId }).catch(() => {}); + throw error; + } + } async control(id: string, enabled: boolean) { if (!enabled) this.cancelUnlock(this.session(id)); return this.session(id).native.request("control", { enabled }); diff --git a/libs/controller/include/controller_files.h b/libs/controller/include/controller_files.h new file mode 100644 index 00000000..ec95fbc3 --- /dev/null +++ b/libs/controller/include/controller_files.h @@ -0,0 +1,31 @@ +/* + * @Author: DI JUNKUN + * @Date: 2026-11-06 + * Copyright (c) 2026 by DI JUNKUN, All Rights Reserved. + */ + +#ifndef _CONTROLLER_FILES_H_ +#define _CONTROLLER_FILES_H_ + +#include +#include +#include +#include + +namespace crossdesk { +class ControllerFiles { +public: + using Send = std::function; + explicit ControllerFiles(Send send); + ~ControllerFiles(); + nlohmann::json Command(const nlohmann::json &request); + void Message(nlohmann::json message); + void Receive(const char *data, size_t size, bool acknowledgement); + void Tick(); +private: + struct State; + std::unique_ptr state_; +}; +} // namespace crossdesk + +#endif diff --git a/libs/controller/include/controller_session.h b/libs/controller/include/controller_session.h index a0bb8286..22325533 100644 --- a/libs/controller/include/controller_session.h +++ b/libs/controller/include/controller_session.h @@ -8,6 +8,7 @@ #define _CONTROLLER_SESSION_H_ #include +#include #include #include #include @@ -80,6 +81,7 @@ public: std::optional TakeRemoteResult(uint64_t request); void ForgetRemoteRequest(uint64_t request) { remote_results_.erase(request); } uint64_t Clipboard(bool write, const std::string &text = {}); + nlohmann::json Files(const nlohmann::json &request); void BeginInput(); std::string InputResult(); bool InputAcknowledged() const { return remote_capabilities_.value("input_ack", nlohmann::json{}) == true; } @@ -126,6 +128,7 @@ private: bool input_ack_active_ = false; bool clipboard_ready_ = false; web_clipboard::Assembler clipboard_assembler_; + std::unique_ptr files_; uint64_t generation_ = 0; uint32_t preview_request_ = 0, preview_ack_ = 0; bool control_ = false, preview_supported_ = false, resolution_ = false; diff --git a/apps/desktop/src/gui/features/file_transfer/file_browser.h b/libs/controller/include/file_browser.h similarity index 96% rename from apps/desktop/src/gui/features/file_transfer/file_browser.h rename to libs/controller/include/file_browser.h index bc7a59cf..d8518df0 100644 --- a/apps/desktop/src/gui/features/file_transfer/file_browser.h +++ b/libs/controller/include/file_browser.h @@ -73,6 +73,7 @@ class FileBrowser { size_t next = 0; std::string destination; bool notify_peer = false; + std::string transfer_id; }; void Post(const std::string& peer, std::function work); void Run(); @@ -85,10 +86,12 @@ class FileBrowser { void ReleaseList(const std::string& peer, const std::string& cursor); Json DirectoryPage(const std::string& peer, const Json& message); void Upload(const std::string& peer, const std::vector& paths, - const std::string& destination, bool notify_peer = false); + const std::string& destination, bool notify_peer = false, + const std::string& transfer_id = {}); void UploadNext(const std::string& peer, const std::shared_ptr& job); bool Current(const std::string& peer, uint64_t generation); - void Report(const std::string& peer, Json result, bool notify_peer); + void Report(const std::string& peer, Json result, bool notify_peer, + const std::string& transfer_id = {}); bool SendMessage(const std::string& peer, Json message); Send send_; Enqueue enqueue_; diff --git a/libs/controller/include/file_transfer_core.h b/libs/controller/include/file_transfer_core.h index 1de83ffb..a1396cb3 100644 --- a/libs/controller/include/file_transfer_core.h +++ b/libs/controller/include/file_transfer_core.h @@ -204,6 +204,16 @@ class Session { if (send_) send_(kFileStream, packet.data(), packet.size()); return true; } + void CancelReceive(uint32_t id) { + uint64_t total = 0; + { + std::lock_guard lock(mutex_); + const auto found = incoming_.find(id); + if (found != incoming_.end()) total = found->second.total; + } + const auto packet = EncodeFileCancel(id, total); + Receive(packet.data(), packet.size()); + } // A cancelled connection cannot be restarted by a late data callback. void Cancel() { std::lock_guard lock(mutex_); diff --git a/libs/controller/src/controller_files.cpp b/libs/controller/src/controller_files.cpp new file mode 100644 index 00000000..8aa223a7 --- /dev/null +++ b/libs/controller/src/controller_files.cpp @@ -0,0 +1,343 @@ +#include "controller_files.h" +#include "file_browser.h" +#include "file_transfer_core.h" +#include +#include +#include +#include +#include + +namespace crossdesk { +namespace fs = std::filesystem; +using nlohmann::json; +namespace { +std::string Utf8(const fs::path &path) { + const auto text = path.u8string(); + return {text.begin(), text.end()}; +} +bool Terminal(const std::string &state) { + return state == "completed" || state == "failed" || state == "cancelled"; +} +} +struct ControllerFiles::State { + Send send; + std::mutex mutex; + std::deque> output; + size_t output_bytes = 0; + struct Upload { uint32_t id; std::string path; }; + std::deque uploads; + std::map files, listings; + struct Route { std::string peer; std::chrono::steady_clock::time_point expires; }; + std::map routes; + std::map history; + std::map arguments; + std::string operation, state = "idle", destination; + bool download = false, prepared = false; + uint32_t sending = 0, completed_receive = 0; + std::unique_ptr transfer; + std::unique_ptr browser; + + explicit State(Send callback) : send(std::move(callback)) { + transfer = std::make_unique( + [this](const char *stream, const char *data, size_t size) { + return Queue(stream, std::string(data, size)) ? 0 : -1; + }, + [this](const std::string &, double progress, bool outbound, const std::string &path) { + std::lock_guard lock(mutex); + if (outbound && sending && files.count(sending)) { + auto &file = files.at(sending); + const auto total = file.value("total", uint64_t{}); + file["bytes"] = progress < 0 ? file.value("bytes", uint64_t{}) : uint64_t(progress * total); + file["state"] = progress < 0 ? "failed" : progress >= 1 ? "completed" : "transferring"; + } + if (!outbound && !path.empty() && completed_receive) { + if (files.count(completed_receive)) files[completed_receive]["local_path"] = path; + completed_receive = 0; + } + }, file_transfer::Options{false, false, 32}, + [this](const file_transfer::ReceiveState &progress) { + std::lock_guard lock(mutex); + auto found = files.find(progress.id); + if (found == files.end()) return; + auto &file = found->second; + if (progress.status == file_transfer::ReceiveStatus::Completed) completed_receive = progress.id; + file["name"] = progress.name; + file["total"] = progress.total; + file["bytes"] = progress.received; + file["state"] = progress.status == file_transfer::ReceiveStatus::Completed ? "completed" + : progress.status == file_transfer::ReceiveStatus::Cancelled ? "cancelled" + : progress.status == file_transfer::ReceiveStatus::Failed ? "failed" : "transferring"; + }); + browser = std::make_unique( + [this](const std::string &peer, const json &message) { + auto packet = message; + if (packet.value("op", "") != "reply" && packet.value("op", "") != "notice") { + std::lock_guard lock(mutex); + if (routes.size() >= 512) return false; + routes[packet.at("id").get()] = {peer, + std::chrono::steady_clock::now() + std::chrono::seconds(30)}; + if (peer != "remote") packet["transfer_id"] = peer; + } + return Queue(kControlStream, packet.dump()); + }, + [this](const std::string &peer, const std::string &path, uint32_t id) { + std::error_code error; + const auto total = fs::file_size(fs::u8path(path), error); + std::lock_guard lock(mutex); + if (error || peer != operation || Terminal(state) || files.size() >= 4096) return false; + files[id] = {{"file_id", id}, {"name", Utf8(fs::u8path(path).filename())}, + {"direction", "upload"}, {"total", total}, {"bytes", 0}, {"state", "queued"}}; + uploads.push_back({id, path}); + return true; + }, + [this](const std::string &peer, uint32_t id, const fs::path &directory) { + { + std::lock_guard lock(mutex); + if (!download || peer != operation || Terminal(state) || files.size() >= 4096) return false; + } + if (!transfer->ReserveDirectory(id, directory)) return false; + { + std::lock_guard lock(mutex); + if (peer == operation && !Terminal(state)) { + files[id] = {{"file_id", id}, {"direction", "download"}, {"total", 0}, + {"bytes", 0}, {"state", "queued"}, {"directory", Utf8(directory)}}; + return true; + } + } + transfer->CancelReceive(id); + return false; + }); + } + ~State() { browser->Stop(); transfer->Cancel(); } + bool Queue(const char *stream, std::string message) { + std::lock_guard lock(mutex); + if (output.size() >= 256 || output_bytes + message.size() > 2 * 1024 * 1024) return false; + output_bytes += message.size(); + output.emplace_back(stream, std::move(message)); + return true; + } + json Snapshot(size_t offset = 0, size_t limit = 256) { + json result{{"operation_id", operation}, {"state", state}, {"files", json::array()}, + {"total_files", files.size()}, {"offset", offset}, {"prepared", prepared}}; + size_t index = 0; + for (const auto &[id, file] : files) { + if (index++ < offset) continue; + if (result["files"].size() >= limit) break; + result["files"].push_back(file); + } + result["next_offset"] = offset + result["files"].size() < files.size() + ? json(offset + result["files"].size()) : json(nullptr); + return result; + } +}; + +ControllerFiles::ControllerFiles(Send send) : state_(std::make_unique(std::move(send))) {} +ControllerFiles::~ControllerFiles() = default; + +json ControllerFiles::Command(const json &request) { + auto &s = *state_; + for (const auto *key : {"offset", "request_id"}) + if (request.contains(key) && (!request[key].is_number_unsigned() || request[key].get() > UINT32_MAX)) + throw std::invalid_argument("INVALID_ARGUMENT"); + for (const auto *key : {"path", "destination", "operation_id"}) { + if (!request.contains(key)) continue; + if (!request[key].is_string()) throw std::invalid_argument("INVALID_ARGUMENT"); + const auto text = request[key].get(); + if (text.size() > 4096 || text.find('\0') != std::string::npos) throw std::invalid_argument("INVALID_ARGUMENT"); + } + const auto command = request.at("operation").get(); + if (command == "list") { + { + std::lock_guard lock(s.mutex); + for (auto it = s.listings.begin(); it != s.listings.end() && s.listings.size() >= 32;) { + if (it->second.value("state", "") != "pending") it = s.listings.erase(it); + else ++it; + } + if (s.listings.size() >= 32) throw std::invalid_argument("FILE_REQUEST_LIMIT"); + } + const auto id = s.browser->List("remote", request.value("path", ""), false); + std::lock_guard lock(s.mutex); + s.listings[id] = {{"state", "pending"}}; + return {{"request_id", id}, {"state", "pending"}}; + } + if (command == "list_result") { + std::lock_guard lock(s.mutex); + const auto id = request.at("request_id").get(); + const auto found = s.listings.find(id); + if (found == s.listings.end()) throw std::invalid_argument("FILE_REQUEST_NOT_FOUND"); + auto result = found->second; + if (!result.contains("entries")) return result; + const auto &entries = found->second.at("entries"); + const size_t offset = request.value("offset", size_t{}); + if (offset > entries.size()) throw std::invalid_argument("INVALID_ARGUMENT"); + result["entries"] = json::array(); + for (size_t i = offset; i < std::min(offset + 256, entries.size()); ++i) + result["entries"].push_back(entries[i]); + result["next_offset"] = offset + result["entries"].size() < entries.size() + ? json(offset + result["entries"].size()) : json(nullptr); + result["request_id"] = id; + return result; + } + if (command == "list_cancel") { + const auto id = request.at("request_id").get(); + s.browser->CancelList("remote", id); + std::lock_guard lock(s.mutex); + s.listings.erase(id); + return {{"cancelled", true}}; + } + if (command == "status") { + std::lock_guard lock(s.mutex); + const auto operation = request.at("operation_id").get(); + if (operation == s.operation) return s.Snapshot(request.value("offset", size_t{})); + const auto found = s.history.find(operation); + if (found == s.history.end()) throw std::invalid_argument("FILE_REQUEST_NOT_FOUND"); + auto result = found->second; + const auto &all = found->second.at("files"); + const size_t offset = request.value("offset", size_t{}); + if (offset > all.size()) throw std::invalid_argument("INVALID_ARGUMENT"); + result["files"] = json::array(); + for (size_t i = offset; i < std::min(offset + 256, all.size()); ++i) result["files"].push_back(all[i]); + result["offset"] = offset; + result["next_offset"] = offset + result["files"].size() < all.size() ? json(offset + result["files"].size()) : json(nullptr); + return result; + } + if (command == "cancel") { + std::vector> incoming; + { + std::lock_guard lock(s.mutex); + if (request.at("operation_id") != s.operation) throw std::invalid_argument("FILE_REQUEST_NOT_FOUND"); + if (Terminal(s.state)) return s.Snapshot(); + for (auto &[id, file] : s.files) { + if (Terminal(file.value("state", ""))) continue; + file["state"] = "cancelled"; + if (file.value("direction", "") == "download") incoming.emplace_back(id, file.value("total", uint64_t{})); + } + s.uploads.clear(); s.state = "cancelled"; + s.output.clear(); s.output_bytes = 0; + for (auto &[id, listing] : s.listings) + if (listing.value("state", "") == "pending") listing = {{"state", "failed"}, {"error", "CANCELLED"}}; + } + s.browser->Disconnect(s.operation); + if (s.transfer->CancelSend()) { + std::lock_guard lock(s.mutex); + if (s.files.count(s.sending)) s.files[s.sending]["state"] = "cancelled"; + } + for (const auto &[id, total] : incoming) { + s.transfer->CancelReceive(id); + s.browser->CancelTransfer(s.operation, id); + } + std::lock_guard lock(s.mutex); + return s.Snapshot(); + } + if (command != "upload" && command != "download") throw std::invalid_argument("INVALID_ARGUMENT"); + const auto operation = request.at("operation_id").get(); + const auto paths = request.at("paths").get>(); + auto destination = request.at("destination").get(); + if (paths.empty() || paths.size() > 256 || operation.empty() || destination.empty()) + throw std::invalid_argument("INVALID_ARGUMENT"); + if (command == "download") destination = Utf8(fs::canonical(fs::u8path(destination))); + { + std::lock_guard lock(s.mutex); + if (s.arguments.count(operation)) { + if (s.arguments.at(operation) != request.dump()) throw std::invalid_argument("OPERATION_ID_CONFLICT"); + return operation == s.operation ? s.Snapshot() : s.history.at(operation); + } + if (!s.operation.empty() && !Terminal(s.state)) throw std::invalid_argument("FILE_TRANSFER_BUSY"); + if (!s.operation.empty()) s.history[s.operation] = s.Snapshot(0, 4096); + if (s.history.size() >= 4) { + s.arguments.erase(s.history.begin()->first); + s.history.erase(s.history.begin()); + } + s.arguments[operation] = request.dump(); + s.operation = operation; s.state = "preparing"; s.prepared = false; + s.destination = destination; s.download = command == "download"; + s.files.clear(); s.uploads.clear(); + } + s.browser->Copy(operation, paths, destination, command == "download"); + return {{"operation_id", operation}, {"state", "preparing"}}; +} + +void ControllerFiles::Message(json message) { + auto &s = *state_; + std::string peer; + { + std::lock_guard lock(s.mutex); + const auto op = message.value("op", json{}); + if (op == "reply") { + if (!message.contains("id") || !message["id"].is_number_unsigned()) return; + const auto found = s.routes.find(message["id"].get()); + if (found == s.routes.end()) return; + peer = found->second.peer; s.routes.erase(found); + } else if (op == "prepare" || op == "notice") { + if (Terminal(s.state) || message.value("transfer_id", json{}) != s.operation) return; + if (op == "prepare" && (!s.download || message.value("destination", json{}) != s.destination)) return; + peer = s.operation; + } else return; + } + s.browser->Handle(peer, std::move(message)); +} + +void ControllerFiles::Receive(const char *data, size_t size, bool acknowledgement) { + auto &s = *state_; + if (acknowledgement) { s.transfer->Ack(data, size); return; } + FileChunkView chunk; + if (!DecodeFileChunk(data, size, &chunk)) return; + { + std::lock_guard lock(s.mutex); + const auto found = s.files.find(chunk.header.file_id); + if (found == s.files.end() || found->second.value("direction", "") != "download") return; + if (Terminal(found->second.value("state", ""))) return; + } + s.transfer->Receive(data, size); +} + +void ControllerFiles::Tick() { + auto &s = *state_; + auto events = s.browser->TakeEvents("remote"); + if (!s.operation.empty()) { + auto transfers = s.browser->TakeEvents(s.operation); + for (auto &event : transfers) events.push_back(std::move(event)); + } + for (auto &event : events) { + std::lock_guard lock(s.mutex); + if (event.id) { + if (s.listings.count(event.id)) { event.result["state"] = event.result.value("ok", json{}) == true ? "completed" : "failed"; s.listings[event.id] = std::move(event.result); } + } else if (!s.operation.empty() && !Terminal(s.state)) { + if (event.result.value("ok", json{}) == false) s.state = "failed"; + if (event.result.value("message", json{}) == "queued" && + (!s.download || event.result.value("op", json{}) == "notice")) s.prepared = true; + } + } + std::optional next; + if (!s.transfer->Sending()) { + std::lock_guard lock(s.mutex); + if (!s.uploads.empty() && !Terminal(s.state)) { + next = s.uploads.front(); s.uploads.pop_front(); s.sending = next->id; + } + } + if (next && !s.transfer->Start(fs::u8path(next->path), Utf8(fs::u8path(next->path).filename()), + file_transfer::Session::Clock::now(), next->id)) { + std::lock_guard lock(s.mutex); s.files[next->id]["state"] = "failed"; + } + s.transfer->Pump(); + std::deque> output; + { + std::lock_guard lock(s.mutex); output.swap(s.output); s.output_bytes = 0; + for (auto it = s.routes.begin(); it != s.routes.end();) { + if (it->second.expires <= std::chrono::steady_clock::now()) it = s.routes.erase(it); + else ++it; + } + if (s.prepared && !Terminal(s.state)) { + bool complete = true, failed = false, cancelled = false; + for (const auto &[id, file] : s.files) { + const auto state = file.value("state", "queued"); + complete &= Terminal(state); failed |= state == "failed"; cancelled |= state == "cancelled"; + } + s.state = complete ? (failed ? "failed" : cancelled ? "cancelled" : "completed") : "transferring"; + } + } + for (const auto &[stream, message] : output) { + if (!s.send(stream.c_str(), message)) { std::lock_guard lock(s.mutex); s.state = "failed"; } + } +} +} // namespace crossdesk diff --git a/libs/controller/src/controller_requests.cpp b/libs/controller/src/controller_requests.cpp index 9204789f..b123c664 100644 --- a/libs/controller/src/controller_requests.cpp +++ b/libs/controller/src/controller_requests.cpp @@ -70,3 +70,11 @@ uint64_t ControllerSession::Clipboard(bool write, const std::string &text) { return id; } } // namespace crossdesk + +namespace crossdesk { +nlohmann::json ControllerSession::Files(const nlohmann::json &request) { + if (!Connected()) throw std::invalid_argument("NOT_READY"); + if (!files_) throw std::invalid_argument("FILES_UNSUPPORTED"); + return files_->Command(request); +} +} // namespace crossdesk diff --git a/libs/controller/src/controller_session.cpp b/libs/controller/src/controller_session.cpp index c2df2ef9..f9c32ceb 100644 --- a/libs/controller/src/controller_session.cpp +++ b/libs/controller/src/controller_session.cpp @@ -43,6 +43,13 @@ void ControllerSession::Data(const char *data, size_t n, const char *, size_t, const char *stream, size_t sn, void *p) { auto &session = *static_cast(p); try { + if (data && n && n <= 256 * 1024 && stream && sn < 64) { + const std::string_view source(stream, sn); + if (source == kFileStream || source == kFileFeedbackStream) { + session.Enqueue({source == kFileStream ? 4 : 5, 0, std::string(data, n)}); + return; + } + } if (data && n && n <= web_clipboard::kHeaderBytes + web_clipboard::kChunkBytes && stream && sn == std::strlen(web_clipboard::kStream) && std::string_view(stream, sn) == web_clipboard::kStream) { @@ -165,6 +172,8 @@ void ControllerSession::Connect(SessionOptions options, std::string device, for (auto stream : {kMouseStream, kKeyboardStream, kControlStream}) AddDataStream(peer_, stream, true); AddDataStream(peer_, web_clipboard::kStream, true); + AddDataStream(peer_, kFileStream, true); + AddDataStream(peer_, kFileFeedbackStream, true); if (Init(peer_) != 0) Fail("media_initialization_failed"); } @@ -250,6 +259,8 @@ void ControllerSession::Tick() { Fail(Connected() ? "connection_lost" : "connection_failed"); else if (e.code == Closed || e.code == Disconnected) Fail("remote_disconnected"); + } else if ((e.kind == 4 || e.kind == 5) && Connected()) { + if (files_) files_->Receive(e.data.data(), e.data.size(), e.kind == 5); } else if (e.kind == 3 && Connected()) { namespace clip = web_clipboard; if (e.data.size() < clip::kHeaderBytes) continue; @@ -272,6 +283,10 @@ void ControllerSession::Tick() { auto j = ParseBoundedJson(e.data); if (!j.is_object()) continue; + if (j.value("type", json{}) == "file_browser") { + if (files_) files_->Message(std::move(j)); + continue; + } if (j.value("type", json{}) == "automation_result" && j.value("version", json{}) == 1 && j.contains("id") && j["id"].is_number_unsigned()) { @@ -285,6 +300,11 @@ void ControllerSession::Tick() { if (remote_capabilities_.value("clipboard", json{}) == true) Send(web_clipboard::kStream, web_clipboard::Frame( web_clipboard::kHello, 0, web_clipboard::kVersion)); + if (remote_capabilities_.value("files", json{}) == true) + files_ = std::make_unique([this](const char *stream, const std::string &data) { + if (!peer_ || !Connected()) return false; + return SendReliableDataFrame(peer_, data.data(), data.size(), stream) == 0; + }); remote_results_.erase(found); } } @@ -320,6 +340,7 @@ void ControllerSession::Tick() { FreeRemoteAction(a); } } + if (files_) files_->Tick(); } bool ControllerSession::Send(const char *stream, const std::string &payload) { return peer_ && Connected() && !payload.empty() && @@ -327,6 +348,7 @@ bool ControllerSession::Send(const char *stream, const std::string &payload) { 0; } void ControllerSession::ClosePeers() { + files_.reset(); { std::lock_guard lock(mutex_); stopping_ = true; diff --git a/apps/desktop/src/gui/features/file_transfer/file_browser.cpp b/libs/controller/src/file_browser.cpp similarity index 96% rename from apps/desktop/src/gui/features/file_transfer/file_browser.cpp rename to libs/controller/src/file_browser.cpp index cf062200..52ff4e47 100644 --- a/apps/desktop/src/gui/features/file_transfer/file_browser.cpp +++ b/libs/controller/src/file_browser.cpp @@ -1,4 +1,4 @@ -#include "features/file_transfer/file_browser.h" +#include "file_browser.h" #include #include @@ -9,7 +9,6 @@ #include #include #include -#include "filesystem_utf8.h" namespace crossdesk { namespace { @@ -27,7 +26,7 @@ fs::path Home() { #else const auto* home = std::getenv("HOME"); #endif - return home ? PathFromUtf8(home) : fs::current_path(); + return home ? std::filesystem::u8path(home) : fs::current_path(); } Json Failure(const std::string& reason) { return {{"ok", false}, {"error", reason}}; } template @@ -77,7 +76,7 @@ std::string Text(const Json& value, const char* key) { } fs::path ExistingDirectory(const std::string& path) { if (path.empty()) throw std::runtime_error("invalid_path"); - auto input = PathFromUtf8(path); + auto input = std::filesystem::u8path(path); if (!input.is_absolute()) throw std::runtime_error("invalid_path"); std::error_code ec; auto result = fs::canonical(input, ec); @@ -85,7 +84,7 @@ fs::path ExistingDirectory(const std::string& path) { return result; } fs::path Destination(const fs::path& root, const std::string& relative) { - auto path = PathFromUtf8(relative); + auto path = std::filesystem::u8path(relative); if (path.is_absolute() || path.has_root_name()) throw std::runtime_error("invalid_path"); auto result = root; for (const auto& component : path) { @@ -155,7 +154,9 @@ bool FileBrowser::Current(const std::string& peer, uint64_t generation) { std::lock_guard lock(mutex_); return !stopping_ && generations_[peer] == generation; } -void FileBrowser::Report(const std::string& peer, Json result, bool notify_peer) { +void FileBrowser::Report(const std::string& peer, Json result, bool notify_peer, + const std::string& transfer_id) { + if (!transfer_id.empty()) result["transfer_id"] = transfer_id; if (!Current(peer, active_generation_)) return; Publish(peer, 0, result); if (notify_peer) { @@ -357,7 +358,7 @@ FileBrowser::Json FileBrowser::Directory(const std::string& value, const std::fu for (char drive = 'A'; drive <= 'Z'; ++drive) { const std::string root = std::string(1, drive) + ":\\"; std::error_code ec; - if (fs::is_directory(PathFromUtf8(root), ec)) + if (fs::is_directory(std::filesystem::u8path(root), ec)) entries.push_back({{"name", root}, {"path", root}, {"directory", true}, {"size", 0}, {"modified", ""}, {"link", false}}); } #else @@ -483,16 +484,18 @@ void FileBrowser::Serve(const std::string& peer, const Json& message) { if (!paths.is_array() || paths.empty() || paths.size() > 256) throw std::runtime_error("invalid_request"); std::vector sources; for (const auto& path : paths) sources.push_back(Text(Json{{"path", path}}, "path")); - Upload(peer, sources, Text(message, "destination"), true); + Upload(peer, sources, Text(message, "destination"), true, + message.contains("transfer_id") ? Text(message, "transfer_id") : ""); result = {{"ok", true}, {"message", "queued"}}; } else throw std::runtime_error("unsupported"); } catch (const std::exception& error) { result = Failure(error.what()); } + if (message.contains("transfer_id")) result["transfer_id"] = Text(message, "transfer_id"); result["op"] = "reply"; result["id"] = message.at("id"); SendMessage(peer, std::move(result)); } void FileBrowser::Upload(const std::string& peer, const std::vector& paths, - const std::string& destination, bool notify_peer) { + const std::string& destination, bool notify_peer, const std::string& transfer_id) { std::vector files; try { if (paths.empty() || paths.size() > 256 || destination.empty()) throw std::runtime_error("invalid_request"); @@ -511,7 +514,7 @@ void FileBrowser::Upload(const std::string& peer, const std::vector }; // Validate the entire selection before creating remote folders or sending. for (const auto& source : paths) { - auto path = PathFromUtf8(source); + auto path = std::filesystem::u8path(source); if (!path.is_absolute() || path.filename().empty()) throw std::runtime_error("invalid_path"); add(path, path.filename()); if (fs::is_directory(fs::symlink_status(path))) { @@ -523,6 +526,7 @@ void FileBrowser::Upload(const std::string& peer, const std::vector } auto job = std::make_shared(); job->files = std::move(files); job->destination = destination; job->notify_peer = notify_peer; + job->transfer_id = transfer_id; UploadNext(peer, job); } catch (const std::exception& error) { Publish(peer, 0, Failure(error.what())); @@ -533,7 +537,7 @@ void FileBrowser::Upload(const std::string& peer, const std::vector void FileBrowser::UploadNext(const std::string& peer, const std::shared_ptr& job) { if (!Current(peer, active_generation_)) return; if (job->next == job->files.size()) { - Report(peer, {{"ok", true}, {"message", "queued"}}, job->notify_peer); + Report(peer, {{"ok", true}, {"message", "queued"}}, job->notify_peer, job->transfer_id); return; } const auto start = job->next; @@ -549,14 +553,15 @@ void FileBrowser::UploadNext(const std::string& peer, const std::shared_ptrnext; // Only one preparation batch per operation is in flight. Large folder // copies do not exhaust pending requests or reserve thousands of paths. - Request(peer, {{"op", "prepare"}, {"destination", job->destination}, {"items", items}}, + Request(peer, {{"op", "prepare"}, {"destination", job->destination}, {"items", items}, + {"transfer_id", job->transfer_id}}, [this, peer, job, start, end](const Json& reply) { - if (!reply.value("ok", false)) { Report(peer, reply, job->notify_peer); return; } + if (!reply.value("ok", false)) { Report(peer, reply, job->notify_peer, job->transfer_id); return; } for (auto index = start; index < end; ++index) { if (!Current(peer, active_generation_)) return; const auto& file = job->files[index]; if (file.id && !enqueue_(peer, file.path, file.id)) { - Report(peer, Failure("file_unreadable"), job->notify_peer); return; + Report(peer, Failure("file_unreadable"), job->notify_peer, job->transfer_id); return; } } UploadNext(peer, job); diff --git a/xmake.lua b/xmake.lua index 56df8ee1..88df2a68 100644 --- a/xmake.lua +++ b/xmake.lua @@ -59,6 +59,14 @@ target("crossdesk_wire") add_includedirs(path.join(wire_dir, "include"), {public = true}) target_end() +target("crossdesk_file_browser") + set_kind("static") + set_languages("c++20") + add_deps("crossdesk_wire") + add_includedirs(path.join(source_dir, "libs/controller/include"), {public = true}) + add_files(path.join(source_dir, "libs/controller/src/file_browser.cpp")) +target_end() + -- CLI source packages omit tests; full checkouts retain these existing targets. if os.isdir(path.join(wire_dir, "tests")) then target("crossdesk_wire_test")