[feat] add MCP file browsing and cancellable transfers

This commit is contained in:
dijunkun
2026-11-06 20:36:48 +08:00
parent 0aed57ed4d
commit 6bfcd5cf3d
20 changed files with 537 additions and 23 deletions
+12
View File
@@ -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())
+6
View File
@@ -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.
+2 -1
View File
@@ -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")
@@ -13,7 +13,7 @@
#include <vector>
#include "file_transfer.h"
#include "features/file_transfer/file_browser.h"
#include <file_browser.h>
#include "runtime/remote_session.h"
namespace crossdesk {
@@ -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") ||
+1 -1
View File
@@ -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",
+1
View File
@@ -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"
+8
View File
@@ -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`。下载请求绑定目标目录及传输标识,接收器接受已为该请求预留的文件。人工暂停后仍可查询或取消已有任务。
+2 -1
View File
@@ -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"));
+26 -2
View File
@@ -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)
+28
View File
@@ -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<Data> {
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 });
@@ -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 <functional>
#include <memory>
#include <nlohmann/json.hpp>
#include <string>
namespace crossdesk {
class ControllerFiles {
public:
using Send = std::function<bool(const char *, const std::string &)>;
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> state_;
};
} // namespace crossdesk
#endif
@@ -8,6 +8,7 @@
#define _CONTROLLER_SESSION_H_
#include <chrono>
#include <controller_files.h>
#include <deque>
#include <minirtc.h>
#include <mutex>
@@ -80,6 +81,7 @@ public:
std::optional<nlohmann::json> 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<ControllerFiles> files_;
uint64_t generation_ = 0;
uint32_t preview_request_ = 0, preview_ack_ = 0;
bool control_ = false, preview_supported_ = false, resolution_ = false;
@@ -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<void()> 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<std::string>& 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<UploadJob>& 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_;
@@ -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<std::mutex> lock(mutex_);
+343
View File
@@ -0,0 +1,343 @@
#include "controller_files.h"
#include "file_browser.h"
#include "file_transfer_core.h"
#include <deque>
#include <filesystem>
#include <mutex>
#include <set>
#include <stdexcept>
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<std::pair<std::string, std::string>> output;
size_t output_bytes = 0;
struct Upload { uint32_t id; std::string path; };
std::deque<Upload> uploads;
std::map<uint32_t, json> files, listings;
struct Route { std::string peer; std::chrono::steady_clock::time_point expires; };
std::map<uint32_t, Route> routes;
std::map<std::string, json> history;
std::map<std::string, std::string> arguments;
std::string operation, state = "idle", destination;
bool download = false, prepared = false;
uint32_t sending = 0, completed_receive = 0;
std::unique_ptr<file_transfer::Session> transfer;
std::unique_ptr<FileBrowser> browser;
explicit State(Send callback) : send(std::move(callback)) {
transfer = std::make_unique<file_transfer::Session>(
[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<FileBrowser>(
[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<uint32_t>()] = {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<State>(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<uint64_t>() > 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<std::string>();
if (text.size() > 4096 || text.find('\0') != std::string::npos) throw std::invalid_argument("INVALID_ARGUMENT");
}
const auto command = request.at("operation").get<std::string>();
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<uint32_t>();
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<uint32_t>();
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<std::string>();
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<std::pair<uint32_t, uint64_t>> 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<std::string>();
const auto paths = request.at("paths").get<std::vector<std::string>>();
auto destination = request.at("destination").get<std::string>();
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<uint32_t>());
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<State::Upload> 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<std::pair<std::string, std::string>> 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
@@ -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
@@ -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<ControllerSession *>(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<ControllerFiles>([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;
@@ -1,4 +1,4 @@
#include "features/file_transfer/file_browser.h"
#include "file_browser.h"
#include <file_transfer_core.h>
#include <algorithm>
@@ -9,7 +9,6 @@
#include <fstream>
#include <set>
#include <stdexcept>
#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 <typename FileTime>
@@ -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<std::string> 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<std::string>& paths,
const std::string& destination, bool notify_peer) {
const std::string& destination, bool notify_peer, const std::string& transfer_id) {
std::vector<UploadItem> 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<std::string>
};
// 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<std::string>
}
auto job = std::make_shared<UploadJob>();
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<std::string>
void FileBrowser::UploadNext(const std::string& peer, const std::shared_ptr<UploadJob>& 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_ptr<Uplo
const auto end = job->next;
// 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);
+8
View File
@@ -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")