diff --git a/apps/cli/agent.cpp b/apps/cli/agent.cpp index e4ace219..235f5bdc 100644 --- a/apps/cli/agent.cpp +++ b/apps/cli/agent.cpp @@ -1,4 +1,5 @@ #include "agent.h" +#include "automation_protocol.h" #include "terminal_io.h" #include "video_server.h" #include @@ -7,7 +8,6 @@ #include #include #include -#include #include #include #include @@ -15,49 +15,10 @@ namespace crossdesk::cli { namespace { using nlohmann::json; +using namespace automation; using Clock = std::chrono::steady_clock; using Ms = std::chrono::milliseconds; constexpr size_t kMaxResponse = 4 * 1024 * 1024; -void Fields(const json &j, std::initializer_list names) { - if (!j.is_object()) - throw std::invalid_argument("INVALID_REQUEST"); - for (auto it = j.begin(); it != j.end(); ++it) - if (std::none_of(names.begin(), names.end(), - [&](const char *s) { return it.key() == s; })) - throw std::invalid_argument("UNKNOWN_FIELD"); -} -int Number(const json &j, const char *key, int low, int high, int fallback) { - if (!j.contains(key)) - return fallback; - if (!j[key].is_number_integer()) - throw std::invalid_argument("INVALID_REQUEST"); - const auto n = j[key].get(); - if (n < low || n > high) - throw std::invalid_argument("INVALID_REQUEST"); - return int(n); -} -uint64_t Sequence(const json &j, const char *key, uint64_t fallback = 0) { - if (!j.contains(key)) - return fallback; - if (!j[key].is_number_unsigned()) - throw std::invalid_argument("INVALID_REQUEST"); - return j[key].get(); -} -bool Flag(const json &j, const char *key, bool fallback = false) { - if (!j.contains(key)) - return fallback; - if (!j[key].is_boolean()) - throw std::invalid_argument("INVALID_REQUEST"); - return j[key].get(); -} -std::string String(const json &j, const char *key, size_t limit = 128) { - if (!j.contains(key) || !j[key].is_string()) - throw std::invalid_argument("INVALID_REQUEST"); - auto s = j[key].get(); - if (s.empty() || s.size() > limit || s.find('\0') != std::string::npos) - throw std::invalid_argument("INVALID_REQUEST"); - return s; -} int64_t Age(Clock::time_point queued) { return std::chrono::duration_cast(Clock::now() - queued).count(); } @@ -125,14 +86,9 @@ struct TerminalEncoding { json id; uint64_t generation, epoch; Clock::time_point queued; + bool cancelled = false; std::future result; }; -struct Step { - int delay = 0, kind = 0, key = 0, flag = 0, wheel = 0; - float x = 0, y = 0; - bool down = false; - std::string text; -}; struct RunningAction { json id; std::string operation; @@ -147,25 +103,6 @@ struct Record { json result; Clock::time_point created; }; -int KeyCode(const std::string &key) { - if (key.size() == 1 && - ((key[0] >= 'A' && key[0] <= 'Z') || (key[0] >= '0' && key[0] <= '9'))) - return key[0]; - static const std::map keys = { - {"BACKSPACE", 8}, {"TAB", 9}, {"ENTER", 13}, {"SHIFT", 16}, - {"CTRL", 17}, {"ALT", 18}, {"PAUSE", 19}, {"CAPS_LOCK", 20}, - {"ESC", 27}, {"SPACE", 32}, {"PAGE_UP", 33}, {"PAGE_DOWN", 34}, - {"END", 35}, {"HOME", 36}, {"LEFT", 37}, {"UP", 38}, - {"RIGHT", 39}, {"DOWN", 40}, {"INSERT", 45}, {"DELETE", 46}, - {"META", 91}, {"F1", 112}, {"F2", 113}, {"F3", 114}, - {"F4", 115}, {"F5", 116}, {"F6", 117}, {"F7", 118}, - {"F8", 119}, {"F9", 120}, {"F10", 121}, {"F11", 122}, - {"F12", 123}}; - auto it = keys.find(key); - if (it == keys.end()) - throw std::invalid_argument("INVALID_KEY"); - return it->second; -} class Automation { public: Automation(int output, SessionOptions options, int password, bool video) @@ -190,18 +127,24 @@ public: eof = true; else if (errno != EAGAIN && errno != EINTR) eof = true; - if (buffer.size() > 65536) { - Error(nullptr, "MESSAGE_TOO_LARGE"); - quit = true; - } for (int count = 0; count < 16 && !quit; ++count) { const auto end = buffer.find('\n'); + if ((end == std::string::npos && buffer.size() > 65536) || + (end != std::string::npos && end > 65536)) { + Error(nullptr, "MESSAGE_TOO_LARGE"); + quit = true; + break; + } if (end == std::string::npos) break; auto request = ParseBoundedJson(buffer.substr(0, end)); buffer.erase(0, end + 1); quit = Handle(request); } + if (!quit && buffer.size() > 128 * 1024) { + Error(nullptr, "INPUT_BACKPRESSURE"); + quit = true; + } if (eof && buffer.find('\n') == std::string::npos) { if (!buffer.empty()) Error(nullptr, "INCOMPLETE_MESSAGE"); @@ -313,16 +256,10 @@ private: code == "BUSY")}, {"side_effect_possible", side_effect}}}}); } - void Release() { - const bool enabled = session_.Status()["control"]; - session_.SetControl(false); - if (enabled && allowed_ && !paused_ && session_.Connected()) - session_.SetControl(true); - } void FinishAction(const char *error = nullptr) { if (!action_) return; - Release(); + session_.ReleaseInput(); json result = { {"status", error ? (action_->submitted ? "outcome_unknown" : "failed_before_submit") @@ -351,12 +288,18 @@ private: Error(encoding_->capture.id, code); encoding_->cancelled = true; } + if (terminal_encoding_ && !terminal_encoding_->cancelled) { + Error(terminal_encoding_->id, code); + terminal_encoding_->cancelled = true; + } observations_.clear(); preview_ = json(); } void Tick() { session_.Tick(); - if (auto f = session_.TakeFrame()) { + // Recycle only uniquely owned frames; an encoder/video worker may still be + // reading the old pixels. Save geometry before Take can swap its storage. + if (auto f = session_.TakeFrame(frame_ && frame_.use_count() == 1 ? frame_.get() : nullptr)) { if (!frame_ || f->generation != frame_->generation || f->width != frame_->width || f->height != frame_->height) { ++epoch_; @@ -384,15 +327,15 @@ private: action_->submitted = true; // A failed send may still have reached the peer. const bool sent = - step.kind == 0 + step.kind == StepKind::Pointer ? session_.Pointer(step.x, step.y, step.flag, step.wheel) - : step.kind == 1 ? session_.Key(step.key, step.down) + : step.kind == StepKind::Key ? session_.Key(step.key, step.down) : session_.Text(step.text); if (!sent) { FinishAction("ACTION_OUTCOME_UNKNOWN"); break; } - if (step.kind == 0) { + if (step.kind == StepKind::Pointer) { cursor_x_ = step.x; cursor_y_ = step.y; cursor_visible_ = true; @@ -407,7 +350,9 @@ private: auto job = std::move(terminal_encoding_); try { auto grid = job->result.get(); - if (!session_.Connected() || job->generation != session_.Generation() || + if (job->cancelled) { + // Consume the worker result without replying twice. + } else if (!session_.Connected() || job->generation != session_.Generation() || job->epoch != epoch_) Error(job->id, "STALE_OBSERVATION"); else if (Age(job->queued) > 2000) @@ -425,7 +370,7 @@ private: {"lines", terminal::Lines(grid)}, {"cursor", cursor}, {"generation", job->generation}, {"geometry_epoch", job->epoch}}); } - } catch (...) { Error(job->id, "FRAME_RENDER_FAILED"); } + } catch (...) { if (!job->cancelled) Error(job->id, "FRAME_RENDER_FAILED"); } } if (encoding_ && encoding_->result.wait_for(Ms(0)) == std::future_status::ready) { @@ -436,8 +381,7 @@ private: if (!frame_ || job->epoch != epoch_ || job->frame->generation != session_.Generation()) Error(job->capture.id, "STALE_OBSERVATION"); - else if (Age(job->frame->queued) > 2000 || - Clock::now() > job->capture.deadline) + else if (Age(job->frame->queued) > 2000) Error(job->capture.id, "FRAME_TIMEOUT"); else { const auto &f = *job->frame; @@ -492,7 +436,7 @@ private: } else ++it; } - if (!encoding_ && !captures_.empty() && frame_ && + if (!encoding_ && !captures_.empty() && frame_ && session_.Connected() && Age(frame_->queued) <= 2000) { auto it = std::find_if( captures_.begin(), captures_.end(), [&](const Capture &c) { @@ -574,12 +518,24 @@ private: } if (captures_.size() >= 16) throw std::invalid_argument("BUSY"); - Capture c{id, j, preview, - Clock::now() + Ms(Number(j, "timeout_ms", 0, 5000, 2000))}; + const int timeout = Number(j, "timeout_ms", 0, 5000, 2000); + Capture c{id, j, preview, Clock::now() + Ms(timeout)}; + // timeout_ms bounds waiting for a suitable frame, not PNG encoding. + // A zero timeout can capture an already available frame immediately. + const bool available = frame_ && frame_->generation == session_.Generation() && + Age(frame_->queued) <= 2000 && + (!j.contains("after_frame_id") || frame_->frame_id > Sequence(j, "after_frame_id")); + if (!encoding_ && captures_.empty() && available) { + Encode(std::move(c)); + return; + } + if (timeout == 0) + throw std::invalid_argument(encoding_ || !captures_.empty() ? "BUSY" : "FRAME_TIMEOUT"); if (preview) captures_.push_back(c); else - captures_.push_front(c); + captures_.insert(std::find_if(captures_.begin(), captures_.end(), + [](const Capture &queued) { return queued.preview; }), c); } const Observation &Observe(const json &j) { auto name = String(j, "observation_id"); @@ -604,9 +560,9 @@ private: Reply(id, r.result); return; } - if (!allowed_ || paused_ || !session_.Status()["control"].get()) + if (!allowed_ || paused_ || !session_.ControlEnabled()) throw std::invalid_argument("CONTROL_DENIED"); - if (!session_.Connected() || !frame_) + if (!session_.Connected() || !frame_ || Age(frame_->queued) > 2000) throw std::invalid_argument("NOT_READY"); if (action_) throw std::invalid_argument("BUSY"); @@ -618,120 +574,10 @@ private: running->operation = operation; running->generation = o.generation; running->epoch = o.epoch; - auto point = [&](const json &p, int delay, int flag = 0, int wheel = 0) { - auto coordinate = [&](const char *key, int size, int offset, int span, - int source) { - if (!p.contains(key) || !p[key].is_number()) - throw std::invalid_argument("INVALID_COORDINATE"); - const auto value = p[key].get(); - if (!std::isfinite(value) || value < 0 || value >= size) - throw std::invalid_argument("INVALID_COORDINATE"); - return float(std::clamp((offset + (value + .5) * span / size) / source, - 0.0, 1.0)); - }; - Step step; - step.delay = delay; - step.flag = flag; - step.wheel = wheel; - step.x = - coordinate("x", o.width, o.region.x, o.region.width, o.source_width); - step.y = coordinate("y", o.height, o.region.y, o.region.height, - o.source_height); - running->steps.push_back(std::move(step)); - }; - auto key = [&](int code, bool down, int delay) { - Step step; - step.kind = 1; - step.key = code; - step.down = down; - step.delay = delay; - running->steps.push_back(std::move(step)); - }; - if (type == "move" || type == "click" || type == "double_click") { - Fields(a, type == "move" - ? std::initializer_list{"type", "x", "y"} - : std::initializer_list{"type", "x", "y", - "button"}); - point(a, 0); - if (type != "move") { - const auto button = - a.contains("button") ? String(a, "button", 8) : "left"; - const int flag = button == "left" ? 1 - : button == "right" ? 3 - : button == "middle" ? 5 - : -1; - if (flag < 0) - throw std::invalid_argument("INVALID_BUTTON"); - point(a, 10, flag); - point(a, 40, flag + 1); - if (type == "double_click") { - point(a, 110, flag); - point(a, 140, flag + 1); - } - } - } else if (type == "scroll") { - Fields(a, {"type", "x", "y", "direction", "ticks"}); - const auto direction = String(a, "direction", 8); - if (direction != "up" && direction != "down" && direction != "left" && - direction != "right") - throw std::invalid_argument("INVALID_DIRECTION"); - const int ticks = Number(a, "ticks", 1, 10, 1); - point(a, 0); - point(a, 10, direction == "left" || direction == "right" ? 8 : 7, - (direction == "up" || direction == "right" ? 120 : -120) * ticks); - } else if (type == "drag") { - Fields(a, {"type", "path", "duration_ms"}); - const auto &path = a.at("path"); - if (!path.is_array() || path.size() < 2 || path.size() > 128) - throw std::invalid_argument("INVALID_PATH"); - const int duration = Number(a, "duration_ms", 100, 2000, 500); - for (const auto &p : path) - Fields(p, {"x", "y"}); - point(path.front(), 0); - point(path.front(), 10, 1); - for (size_t i = 1; i < path.size(); ++i) - point(path[i], 10 + int(i * duration / (path.size() - 1))); - point(path.back(), duration + 30, 2); - } else if (type == "key_press") { - Fields(a, {"type", "key"}); - const auto code = KeyCode(String(a, "key", 16)); - key(code, true, 0); - key(code, false, 40); - } else if (type == "hotkey") { - Fields(a, {"type", "keys"}); - const auto &keys = a.at("keys"); - if (!keys.is_array() || keys.size() < 2 || keys.size() > 6) - throw std::invalid_argument("INVALID_KEYS"); - std::set seen; - std::vector codes; - for (const auto &item : keys) { - if (!item.is_string()) - throw std::invalid_argument("INVALID_KEY"); - int code = KeyCode(item.get()); - if (!seen.insert(code).second) - throw std::invalid_argument("DUPLICATE_KEY"); - codes.push_back(code); - } - int delay = 0; - for (int code : codes) { - key(code, true, delay); - delay += 15; - } - delay += 30; - for (auto it = codes.rbegin(); it != codes.rend(); ++it) { - key(*it, false, delay); - delay += 15; - } - } else if (type == "type_text") { - Fields(a, {"type", "text"}); - if (!session_.Status()["text_input"].get()) - throw std::invalid_argument("TEXT_INPUT_UNSUPPORTED"); - Step step; - step.kind = 2; - step.text = String(a, "text", 16384); - running->steps.push_back(std::move(step)); - } else - throw std::invalid_argument("UNSUPPORTED_ACTION"); + if (type == "type_text" && !session_.Status()["text_input"].get()) + throw std::invalid_argument("TEXT_INPUT_UNSUPPORTED"); + running->steps = automation::BuildInputSteps(a, + {o.width, o.height, o.source_width, o.source_height, o.region}); while (records_.size() >= 128 || (!records_.empty() && Age(records_.front().created) > 300000)) records_.pop_front(); @@ -884,8 +730,12 @@ private: throw std::invalid_argument("INVALID_REQUEST"); } else if (method == "display") { Fields(j, {"id", "method", "index"}); + const int index = Number(j, "index", 0, 63, -1); + if (!session_.Connected() || index < 0 || + size_t(index) >= session_.Status()["displays"].size()) + throw std::invalid_argument("DISPLAY_UNAVAILABLE"); Stop("SESSION_CHANGED"); - if (!session_.SelectDisplay(Number(j, "index", 0, 63, -1))) + if (!session_.SelectDisplay(index)) throw std::invalid_argument("DISPLAY_UNAVAILABLE"); frame_.reset(); ++epoch_; @@ -902,6 +752,10 @@ private: encoding_->cancelled = true; Error(target, "CANCELLED"); } + if (terminal_encoding_ && !terminal_encoding_->cancelled && terminal_encoding_->id == target) { + terminal_encoding_->cancelled = true; + Error(target, "CANCELLED"); + } for (auto it = captures_.begin(); it != captures_.end();) { if (it->id == target) { Error(target, "CANCELLED"); @@ -954,7 +808,7 @@ private: bool hello_ = false, connected_once_ = false, allowed_ = false, paused_ = false; uint64_t epoch_ = 0, observation_sequence_ = 0; - std::shared_ptr frame_; + std::shared_ptr frame_; std::deque observations_; std::deque captures_; std::unique_ptr encoding_; diff --git a/apps/cli/automation_protocol.cpp b/apps/cli/automation_protocol.cpp new file mode 100644 index 00000000..bd54d647 --- /dev/null +++ b/apps/cli/automation_protocol.cpp @@ -0,0 +1,262 @@ +#include "automation_protocol.h" + +#include +#include +#include +#include +#include + +namespace crossdesk::cli::automation { +void Fields(const json &j, std::initializer_list names) { + if (!j.is_object()) + throw std::invalid_argument("INVALID_REQUEST"); + for (auto it = j.begin(); it != j.end(); ++it) + if (std::none_of(names.begin(), names.end(), + [&](const char *s) { return it.key() == s; })) + throw std::invalid_argument("UNKNOWN_FIELD"); +} +int Number(const json &j, const char *key, int low, int high, int fallback) { + if (!j.contains(key)) + return fallback; + if (!j[key].is_number_integer()) + throw std::invalid_argument("INVALID_REQUEST"); + const auto n = j[key].get(); + if (n < low || n > high) + throw std::invalid_argument("INVALID_REQUEST"); + return int(n); +} +uint64_t Sequence(const json &j, const char *key, uint64_t fallback) { + if (!j.contains(key)) + return fallback; + if (!j[key].is_number_unsigned()) + throw std::invalid_argument("INVALID_REQUEST"); + return j[key].get(); +} +bool Flag(const json &j, const char *key, bool fallback) { + if (!j.contains(key)) + return fallback; + if (!j[key].is_boolean()) + throw std::invalid_argument("INVALID_REQUEST"); + return j[key].get(); +} +std::string String(const json &j, const char *key, size_t limit) { + if (!j.contains(key) || !j[key].is_string()) + throw std::invalid_argument("INVALID_REQUEST"); + auto s = j[key].get(); + if (s.empty() || s.size() > limit || s.find('\0') != std::string::npos) + throw std::invalid_argument("INVALID_REQUEST"); + return s; +} +namespace { +int KeyCode(const std::string &key) { + if (key.size() == 1 && + ((key[0] >= 'A' && key[0] <= 'Z') || (key[0] >= '0' && key[0] <= '9'))) + return key[0]; + static const std::map keys = {{"BACKSPACE", 8}, + {"TAB", 9}, + {"ENTER", 13}, + {"SHIFT", 16}, + {"CTRL", 17}, + {"ALT", 18}, + {"PAUSE", 19}, + {"CAPS_LOCK", 20}, + {"ESC", 27}, + {"SPACE", 32}, + {"PAGE_UP", 33}, + {"PAGE_DOWN", 34}, + {"END", 35}, + {"HOME", 36}, + {"LEFT", 37}, + {"UP", 38}, + {"RIGHT", 39}, + {"DOWN", 40}, + {"INSERT", 45}, + {"DELETE", 46}, + {"META", 91}, + {"F1", 112}, + {"F2", 113}, + {"F3", 114}, + {"F4", 115}, + {"F5", 116}, + {"F6", 117}, + {"F7", 118}, + {"F8", 119}, + {"F9", 120}, + {"F10", 121}, + {"F11", 122}, + {"F12", 123}, + {"SEMICOLON", 186}, + {"EQUAL", 187}, + {"COMMA", 188}, + {"MINUS", 189}, + {"PERIOD", 190}, + {"SLASH", 191}, + {"BACKQUOTE", 192}, + {"BRACKET_LEFT", 219}, + {"BACKSLASH", 220}, + {"BRACKET_RIGHT", 221}, + {"QUOTE", 222}}; + auto it = keys.find(key); + if (it == keys.end()) + throw std::invalid_argument("INVALID_KEY"); + return it->second; +} +} // namespace + +std::vector BuildInputSteps(const json &a, const InputGeometry &o) { + const auto type = String(a, "type", 32); + std::vector steps; + auto point = [&](const json &p, int delay, int flag = 0, int wheel = 0) { + auto coordinate = [&](const char *key, int size, int offset, int span, + int source) { + if (!p.contains(key) || !p[key].is_number()) + throw std::invalid_argument("INVALID_COORDINATE"); + const auto value = p[key].get(); + if (!std::isfinite(value) || value < 0 || value >= size) + throw std::invalid_argument("INVALID_COORDINATE"); + const auto center = std::min(value + .5, double(size) - .5); + return float(std::clamp((offset + center * span / size) / source, 0.0, 1.0)); + }; + Step step; + step.delay = delay; + step.flag = flag; + step.wheel = wheel; + step.x = + coordinate("x", o.width, o.region.x, o.region.width, o.source_width); + step.y = + coordinate("y", o.height, o.region.y, o.region.height, o.source_height); + steps.push_back(std::move(step)); + }; + auto key = [&](int code, bool down, int delay) { + Step step; + step.kind = StepKind::Key; + step.key = code; + step.down = down; + step.delay = delay; + steps.push_back(std::move(step)); + }; + if (type == "move" || type == "click" || type == "double_click") { + Fields(a, type == "move" + ? std::initializer_list{"type", "x", "y", + "modifiers"} + : std::initializer_list{"type", "x", "y", + "button", "modifiers"}); + point(a, 0); + if (type != "move") { + const auto button = + a.contains("button") ? String(a, "button", 8) : "left"; + const int flag = button == "left" ? 1 + : button == "right" ? 3 + : button == "middle" ? 5 + : -1; + if (flag < 0) + throw std::invalid_argument("INVALID_BUTTON"); + point(a, 10, flag); + point(a, 40, flag + 1); + if (type == "double_click") { + point(a, 110, flag); + point(a, 140, flag + 1); + } + } + } else if (type == "scroll") { + Fields(a, {"type", "x", "y", "direction", "ticks", "modifiers"}); + const auto direction = String(a, "direction", 8); + if (direction != "up" && direction != "down" && direction != "left" && + direction != "right") + throw std::invalid_argument("INVALID_DIRECTION"); + const int ticks = Number(a, "ticks", 1, 10, 1); + point(a, 0); + point(a, 10, direction == "left" || direction == "right" ? 8 : 7, + (direction == "up" || direction == "right" ? 120 : -120) * ticks); + } else if (type == "drag") { + Fields(a, {"type", "path", "duration_ms", "button", "modifiers"}); + const auto button = a.contains("button") ? String(a, "button", 8) : "left"; + const int flag = button == "left" ? 1 + : button == "right" ? 3 + : button == "middle" ? 5 + : -1; + if (flag < 0) + throw std::invalid_argument("INVALID_BUTTON"); + const auto &path = a.at("path"); + if (!path.is_array() || path.size() < 2 || path.size() > 128) + throw std::invalid_argument("INVALID_PATH"); + const int duration = Number(a, "duration_ms", 100, 2000, 500); + for (const auto &p : path) + Fields(p, {"x", "y"}); + point(path.front(), 0); + point(path.front(), 10, flag); + for (size_t i = 1; i < path.size(); ++i) + point(path[i], 10 + int(i * duration / (path.size() - 1))); + point(path.back(), duration + 30, flag + 1); + } else if (type == "key_press") { + Fields(a, {"type", "key", "modifiers"}); + const auto code = KeyCode(String(a, "key", 16)); + key(code, true, 0); + key(code, false, 40); + } else if (type == "hotkey") { + Fields(a, {"type", "keys"}); + const auto &keys = a.at("keys"); + if (!keys.is_array() || keys.size() < 2 || keys.size() > 6) + throw std::invalid_argument("INVALID_KEYS"); + std::set seen; + std::vector codes; + for (const auto &item : keys) { + if (!item.is_string()) + throw std::invalid_argument("INVALID_KEY"); + int code = KeyCode(item.get()); + if (!seen.insert(code).second) + throw std::invalid_argument("DUPLICATE_KEY"); + codes.push_back(code); + } + int delay = 0; + for (int code : codes) { + key(code, true, delay); + delay += 15; + } + delay += 30; + for (auto it = codes.rbegin(); it != codes.rend(); ++it) { + key(*it, false, delay); + delay += 15; + } + } else if (type == "type_text") { + Fields(a, {"type", "text"}); + Step step; + step.kind = StepKind::Text; + step.text = String(a, "text", 16384); + steps.push_back(std::move(step)); + } else + throw std::invalid_argument("UNSUPPORTED_ACTION"); + if (a.contains("modifiers")) { + const auto &modifiers = a["modifiers"]; + if (!modifiers.is_array() || modifiers.size() > 4) + throw std::invalid_argument("INVALID_KEYS"); + std::set seen; + std::vector prefix; + for (const auto &item : modifiers) { + if (!item.is_string()) + throw std::invalid_argument("INVALID_KEY"); + const auto name = item.get(); + if (name != "CTRL" && name != "ALT" && name != "SHIFT" && name != "META") + throw std::invalid_argument("INVALID_KEY"); + const int code = KeyCode(name); + if (!seen.insert(code).second || + (type == "key_press" && String(a, "key", 16) == name)) + throw std::invalid_argument("DUPLICATE_KEY"); + Step step; + step.kind = StepKind::Key; + step.key = code; + step.down = true; + step.delay = int(prefix.size()) * 15; + prefix.push_back(step); + } + const int offset = int(prefix.size()) * 15; + for (auto &step : steps) + step.delay += offset; + steps.insert(steps.begin(), prefix.begin(), prefix.end()); + int delay = steps.back().delay + 15; + for (auto it = prefix.rbegin(); it != prefix.rend(); ++it, delay += 15) + key(it->key, false, delay); + } + return steps; +} +} // namespace crossdesk::cli::automation diff --git a/apps/cli/automation_protocol.h b/apps/cli/automation_protocol.h new file mode 100644 index 00000000..fa3ac80a --- /dev/null +++ b/apps/cli/automation_protocol.h @@ -0,0 +1,41 @@ +/* + * @Author: DI JUNKUN + * @Date: 2026-11-03 + * Copyright (c) 2026 by DI JUNKUN, All Rights Reserved. + */ + +#ifndef _AUTOMATION_PROTOCOL_H_ +#define _AUTOMATION_PROTOCOL_H_ + +#include +#include +#include +#include + +namespace crossdesk::cli::automation { +using nlohmann::json; +void Fields(const json &, std::initializer_list names); +int Number(const json &, const char *key, int low, int high, int fallback); +uint64_t Sequence(const json &, const char *key, uint64_t fallback = 0); +bool Flag(const json &, const char *key, bool fallback = false); +std::string String(const json &, const char *key, size_t limit = 128); + +struct InputGeometry { + int width, height, source_width, source_height; + ImageRegion region; +}; +enum class StepKind { Pointer, Key, Text }; +struct Step { + StepKind kind = StepKind::Pointer; + int delay = 0, key = 0, flag = 0, wheel = 0; + float x = 0, y = 0; + bool down = false; + std::string text; +}; + +// Pure validation and translation: no session access or input side effects. +std::vector BuildInputSteps(const json &action, + const InputGeometry &geometry); +} // namespace crossdesk::cli::automation + +#endif diff --git a/apps/cli/docs/automation-protocol.md b/apps/cli/docs/automation-protocol.md index 8a37c35c..0aa5eed8 100644 --- a/apps/cli/docs/automation-protocol.md +++ b/apps/cli/docs/automation-protocol.md @@ -93,6 +93,8 @@ NV12 frames must be tightly packed, even dimensions, at most 8192 per axis and 1 `after_frame_id` requires `generation`. A screenshot waits up to `timeout_ms` for a newer frame; a preview can return `{unchanged:true, frame_age_ms, stale, generation, frame_id}`. Neither path treats an image as an input acknowledgement. +`timeout_ms` bounds frame/worker availability, not the subsequent PNG encoding. With `timeout_ms: 0`, a fresh eligible frame is encoded immediately if the worker and queue are idle; otherwise return `FRAME_TIMEOUT` or `BUSY`. Captures of the same priority retain arrival order. Encoding still rejects obsolete geometry and frames that become older than two seconds. Terminal rendering also honors cancellation without sending a second result when the worker finishes. + ## Actions All coordinates refer to the immutable returned observation image. The native layer maps them through its source rectangle to normalized remote coordinates, without terminal layout padding. Coordinates must satisfy `0 <= x < image_width` and `0 <= y < image_height`. Observations expire after 30 seconds, retain at most 32 entries, and are invalidated by geometry/display generation changes. @@ -106,12 +108,16 @@ Supported payloads: - `move`: `x`, `y`. - `click`, `double_click`: `x`, `y`, optional `button` (`left`, `right`, `middle`). - `scroll`: `x`, `y`, `direction` (`up`, `down`, `left`, `right`), optional `ticks` (1–10). -- `drag`: `path` of 2–128 `{x,y}` points, optional `duration_ms` (100–2000). +- `drag`: `path` of 2–128 `{x,y}` points, optional `duration_ms` (100–2000), optional `button` (`left`, `right`, `middle`; default `left`). - `key_press`: `key`; `hotkey`: 2–6 unique `keys`. - `type_text`: `text`, at most 16 KiB UTF-8; requires negotiated remote text-input capability. +All pointer actions and `key_press` accept optional `modifiers`, an array of at most four unique names from `CTRL`, `ALT`, `SHIFT`, `META`. They are pressed before the action and released afterwards in reverse order; a `key_press` cannot also list its own key as a modifier. `hotkey` already describes the complete chord and does not accept `modifiers`. Punctuation VK names are `SEMICOLON`, `EQUAL`, `COMMA`, `MINUS`, `PERIOD`, `SLASH`, `BACKQUOTE`, `BRACKET_LEFT`, `BACKSLASH`, `BRACKET_RIGHT`, `QUOTE`; these name keys, not layout-independent text. + Only one input sequence runs at once. Timed steps are advanced by the owner loop; cancellation, disconnect, display changes and pause release held buttons/keys. A failed send is conservatively considered potentially submitted. +New actions require a current frame no older than two seconds as well as a valid observation. Cached operation lookups remain available with a stalled feed or paused control. Failed key/button releases remain tracked for subsequent cleanup, including after control is disabled; a successful local send still does not prove remote execution. Invalid display indices leave existing observations and input sequences intact. + Results use `status: submitted | in_progress | failed_before_submit | outcome_unknown`, with `operation_id`, `retryable`, `side_effect_possible` and, for completed actions, `generation`/`after_frame_id`. The same operation ID and arguments retrieve the cached result; different arguments conflict. Retention is bounded to 128 entries, with entries older than five minutes eligible for removal on a new action. Never retry an uncertain effect with a new ID. `submitted` does not claim remote execution. The native process trusts its parent. The parent must authenticate panel-only `viewer_control` calls and protect descriptor 3; this private pipe does not isolate arbitrary programs running as the same OS user. diff --git a/apps/cli/xmake.lua b/apps/cli/xmake.lua index b2113665..99aa5c96 100644 --- a/apps/cli/xmake.lua +++ b/apps/cli/xmake.lua @@ -37,7 +37,7 @@ target("crossdesk-cli") set_version("0.2.0") add_deps("crossdesk_terminal") add_files("main.cpp") - add_files("agent.cpp", "video_server.cpp") + add_files("agent.cpp", "automation_protocol.cpp", "video_server.cpp") add_packages("asio", "websocketpp") add_files(is_plat("windows") and "terminal_io_windows.cpp" or "terminal_io_posix.cpp") if is_plat("windows") then diff --git a/apps/mcp/README.md b/apps/mcp/README.md index 122d503a..48cf9059 100644 --- a/apps/mcp/README.md +++ b/apps/mcp/README.md @@ -61,6 +61,8 @@ Windows 和 Linux 使用各自构建产物路径;`--native-binary` 必须指 `crossdesk_open` 立即返回,让 Apps 宿主有机会渲染面板;等待连接在后续 `crossdesk_get_session` 调用中完成。按 `view_id` 查询默认最多等待 25 秒,`wait_ms: 0` 则立即查询;未就绪时返回 `wait_timed_out: true`,Agent 在原任务仍待执行时继续等待。按 `session_id` 查询仍立即返回。取消等待只取消本次查询,不断开用户正在建立或使用的连接。用户主动断开后返回 `closed`,Agent 停止等待,不自动重连。 +面板发起的连接失败后,按 `view_id` 等待也会返回 `failed` 和错误码,避免 Agent 一直停留在等待用户状态。断开正在建立的连接会取消整个连接尝试。多个面板和 Agent 同时查询状态时共用一次原生查询,单次查询取消或暂时繁忙不会断开会话。`crossdesk_connect` 在连接成功但面板初始化失败时仍返回有效的 `session_id`,并附加 `panel_error`。 + 自动继续依赖宿主保持当前 Agent 任务及工具调用流程。它不通过网页伪造用户消息,也不承诺从已经结束的对话中主动唤醒 Agent;此类唤醒由宿主负责。 人工暂停由原生层锁定,模型不能通过 `set_control(true)` 恢复;只有当前授权面板的恢复按钮可以恢复。仅查看连接不能升级为控制连接,需重新连接并授权。关闭/隐藏面板仅停止预览,远程会话保留;断开按钮或 `crossdesk_disconnect` 才结束会话。 @@ -162,8 +164,16 @@ Apps 依赖宿主遵守 app-only 工具可见性和 `_meta` 的模型不可见 动作包括 `move`、`click`、`double_click`、`scroll`、`drag`、`key_press`、`hotkey`、`type_text`。不提供 shell、任意键按下保持、剪贴板或文件操作。按键使用 Windows VK 对应的标准名称:`A`–`Z`、`0`–`9`、`F1`–`F12`、`CTRL`、`ALT`、`SHIFT`、`META`、`ENTER`、`ESC`、方向键等;应用快捷键仍需根据远端系统选择。 +鼠标动作和 `key_press` 支持可选的 `modifiers: ["CTRL", "ALT", "SHIFT", "META"]`(选择需要的键,不能重复),用于组合点击、选择、拖拽和快捷键。修饰键仅在本次动作期间保持,结束、取消或人工暂停时释放。`drag` 可指定 `button: "left" | "right" | "middle"`,默认左键。标点键支持 `SEMICOLON`、`EQUAL`、`COMMA`、`MINUS`、`PERIOD`、`SLASH`、`BACKQUOTE`、`BRACKET_LEFT`、`BACKSLASH`、`BRACKET_RIGHT`、`QUOTE`;这些是 VK 键位名称,实际字符取决于远端键盘布局,输入文字应使用 `type_text`。例如: + +```json +{"session_id":"…","operation_id":"select-item-1","observation_id":"…","action":{"type":"click","x":120,"y":80,"modifiers":["SHIFT"]}} +``` + 截图默认最长边 1280,最大 2048;允许指定源帧像素区域。返回 `source_size`、`source_rect`、`image_size`、`generation`、`geometry_epoch`、`frame_id`、`frame_age_ms` 和 `observation_id`。动作坐标使用返回图像坐标,不使用浏览器 CSS 像素。观测保留最多 32 个,30 秒过期;切换显示器、连接代际或图像几何变化立即失效。截图使用不超过 2 秒的帧,PNG 不超过 2 MiB,必要时继续缩小输出并返回真实尺寸。 +`timeout_ms` 限制等待可用帧的时间,PNG 编码独立执行;`0` 表示只使用当前已就绪的新鲜帧,无帧返回 `FRAME_TIMEOUT`,编码器繁忙返回 `BUSY`。截图按到达顺序处理,并优先于预览。视频超过 2 秒未更新时拒绝新的键鼠输入,恢复新鲜画面后可继续;已提交操作的去重查询仍可读取原结果。 + `operation_id` 在当前原生会话中缓存最近 128 个操作,至少在未被容量淘汰的 5 分钟窗口内支持同参数查询;不同参数复用 ID 返回冲突。`submitted` 仅说明本地发送完成,远端没有执行 ACK。超时、断网、取消或发送失败可能返回 `outcome_unknown`/`side_effect_possible`,禁止自动重放。`observe_after` 默认开启,后续图像只证明取得更新帧,不证明操作执行成功。 ## 实时视频与按需截图 @@ -211,3 +221,11 @@ npm run pack:npm -- --out-dir /absolute/path/artifacts 脚本生成可发布 tarball,包含面板、运行代码、源代码、许可和固定版本的五个平台可选依赖。CI 的 `MCP package` 工作流仅构建并上传 `mcp-package`,不发布 npm。原生包继续由 CLI 工作流构建,`doctor` 必须同时声明自动化协议和 `automation_video_protocols: ["h264-annexb/1", "nv12/1"]` 和 `automation_video_decoders: ["auto", "software"]`,以及 `automation_video_transport: "loopback-websocket/1"`、`automation_terminal_protocol: "ansi/1"`;先发布对应 0.2.0 平台包,再发布 MCP tarball。尚未进行真实跨平台发布/远端验收时,不应把构建成功等同于各宿主和设备上的联调完成。 从发布 tarball 重建 TypeScript/面板时,将包中的 `build-package.json` 复制为 `package.json`,运行 `npm ci && npm run build`。原生源码和依赖来源随平台包提供,详见 [源码来源](../cli/docs/source-provenance.md)。 + +## 本地审阅验证(2026-11-03) + +动作参数校验和输入步骤生成集中在 `apps/cli/automation_protocol.*`,主循环负责调度、取消和权限检查。截图按色度网格对齐后仅转换所需区域,PNG 在容量受限的缓冲区中单次压缩;原生帧仅在没有其他读取者时复用内存。Node 按行累计响应字节,在完整响应到达时解码,避免反复扫描和复制大截图。 + +本机 macOS arm64 完成 TypeScript/面板构建、原生 CLI 构建、现有 wire/keyboard-state 测试,以及工程外的 MCP、会话和原生动作回归检查。ASan/UBSan 检查覆盖动作调度和输入清理;140 组随机图像覆盖完整帧、奇偶裁剪边界、缩放及容量回退,优化前后 PNG 字节一致。 + +使用 1920 × 1080 合成 NV12、最长边 1280、每项 6 次的 `-O2` 本地基准:完整画面编码约 22.5 → 11.5 ms,321 × 241 裁剪约 2.1 → 0.9 ms。数字仅衡量图像转换与编码,不包含网络、解码、远端执行或宿主渲染;未连接真实远端或完成 Windows/Linux 验收。额外测试代码、脚本、数据和产物均存放于工程外系统临时目录。 diff --git a/apps/mcp/src/browser.ts b/apps/mcp/src/browser.ts index 6afeacbb..f058989b 100644 --- a/apps/mcp/src/browser.ts +++ b/apps/mcp/src/browser.ts @@ -367,6 +367,7 @@ export async function pairBrowser( resolve(undefined); }); socket.once("error", () => resolve(undefined)); + socket.once("close", () => resolve(undefined)); socket.once("connect", () => socket.write( JSON.stringify({ secret: descriptor.secret, view_id: view }) + "\n", diff --git a/apps/mcp/src/common.ts b/apps/mcp/src/common.ts index 298f5596..b56e6f0d 100644 --- a/apps/mcp/src/common.ts +++ b/apps/mcp/src/common.ts @@ -61,29 +61,38 @@ export const point = z.strictObject({ y: z.number().finite().min(0).max(8192), }); const button = z.enum(["left", "right", "middle"]).optional(); -const key = z - .string() - .min(1) - .max(16) - .regex(/^[A-Z0-9_]+$/); +const unique = (keys: string[]) => new Set(keys).size === keys.length; +const modifiers = z.array(z.enum(["CTRL", "ALT", "SHIFT", "META"])).max(4).refine(unique) + .describe("Modifiers held during this action and released afterwards, including on cancellation.").optional(); +const key = z.enum([ + ..."ABCDEFGHIJKLMNOPQRSTUVWXYZ0123456789".split(""), + "BACKSPACE", "TAB", "ENTER", "SHIFT", "CTRL", "ALT", "PAUSE", "CAPS_LOCK", + "ESC", "SPACE", "PAGE_UP", "PAGE_DOWN", "END", "HOME", "LEFT", "UP", "RIGHT", "DOWN", + "INSERT", "DELETE", "META", "F1", "F2", "F3", "F4", "F5", "F6", "F7", "F8", "F9", "F10", "F11", "F12", + "SEMICOLON", "EQUAL", "COMMA", "MINUS", "PERIOD", "SLASH", "BACKQUOTE", "BRACKET_LEFT", "BACKSLASH", "BRACKET_RIGHT", "QUOTE", +]); export const action = z.discriminatedUnion("type", [ - point.extend({ type: z.literal("move") }), - point.extend({ type: z.literal("click"), button }), - point.extend({ type: z.literal("double_click"), button }), + point.extend({ type: z.literal("move"), modifiers }), + point.extend({ type: z.literal("click"), button, modifiers }), + point.extend({ type: z.literal("double_click"), button, modifiers }), point.extend({ type: z.literal("scroll"), direction: z.enum(["up", "down", "left", "right"]), ticks: z.int().min(1).max(10).optional(), + modifiers, }), z.strictObject({ type: z.literal("drag"), path: z.array(point).min(2).max(128), duration_ms: z.int().min(100).max(2000).optional(), + button, + modifiers, }), - z.strictObject({ type: z.literal("key_press"), key }), + z.strictObject({ type: z.literal("key_press"), key, modifiers }) + .refine((p) => !p.modifiers?.some((modifier) => modifier === p.key)), z.strictObject({ type: z.literal("hotkey"), - keys: z.array(key).min(2).max(6), + keys: z.array(key).min(2).max(6).refine(unique), }), z.strictObject({ type: z.literal("type_text"), diff --git a/apps/mcp/src/native.ts b/apps/mcp/src/native.ts index efc2fa9f..8e5ec16c 100644 --- a/apps/mcp/src/native.ts +++ b/apps/mcp/src/native.ts @@ -95,8 +95,10 @@ export class NativeClient { private child: ChildProcess; private pending = new Map(); private sequence = 0; - private buffer = ""; + private chunks: Buffer[] = []; + private bufferedBytes = 0; private closed = false; + private exited = false; private closing?: Promise; constructor(binary: string, server: string, port: number, logs: string, readonly videoOptions: VideoOptions = { decode_path: "browser", decoder: "auto" }) { @@ -122,12 +124,14 @@ export class NativeClient { env: nativeEnvironment(), }, ); - this.child.stdout!.setEncoding("utf8"); - this.child.stdout!.on("data", (part: string) => this.receive(part)); + this.child.stdout!.on("data", (part: Buffer) => this.receive(part)); this.child.stdin!.on("error", () => this.fail()); this.child.stdio[3]!.on("error", () => this.fail()); this.child.once("error", () => this.fail()); - this.child.once("exit", () => this.fail()); + // Process exit can precede the final stdout data. Drain replies until EOF + // or close before rejecting requests that have no response. + this.child.once("close", () => { this.exited = true; this.fail(); }); + this.child.stdout!.once("end", () => this.fail()); } async connect( device: string, @@ -161,8 +165,8 @@ export class NativeClient { timeout = 7000, signal?: AbortSignal, ): Promise { - if (this.closed) - return Promise.reject(new Failure("SESSION_CLOSED", method === "action")); + if (this.closed || (this.closing && method !== "disconnect")) + return Promise.reject(new Failure("SESSION_CLOSED")); if (signal?.aborted) return Promise.reject(new Failure("CANCELLED")); if (this.pending.size >= 32 || this.child.stdin!.writableLength > 262144) return Promise.reject(new Failure("BUSY", false, true)); @@ -201,22 +205,30 @@ export class NativeClient { this.child.stdin!.write(line); }); } - private receive(part: string) { - this.buffer += part; - if (Buffer.byteLength(this.buffer) > 8 * 1024 * 1024) { - this.fail(); - this.child.kill(); - return; - } - for (;;) { - const end = this.buffer.indexOf("\n"); - if (end < 0) break; - const line = this.buffer.slice(0, end); - this.buffer = this.buffer.slice(end + 1); + private receive(part: Buffer) { + if (this.closed) return; + for (let start = 0; start < part.length;) { + const end = part.indexOf(10, start); + const chunk = part.subarray(start, end < 0 ? part.length : end); + this.bufferedBytes += chunk.length; + if (this.bufferedBytes > 4 * 1024 * 1024) { + this.fail(); + this.child.kill(); + return; + } + this.chunks.push(chunk); + if (end < 0) return; + // One allocation and UTF-8 decode per response, including split UTF-8 + // codepoints. Do not repeatedly scan/copy a multi-megabyte PNG response. + const line = this.chunks.length === 1 ? chunk : Buffer.concat(this.chunks, this.bufferedBytes); + this.chunks = []; + this.bufferedBytes = 0; + start = end + 1; try { - if (Buffer.byteLength(line) > 4 * 1024 * 1024) throw new Error(); - const reply = JSON.parse(line); - if (!Number.isSafeInteger(reply.id) || typeof reply.ok !== "boolean") + const reply = JSON.parse(line.toString("utf8")); + if (!reply || !Number.isSafeInteger(reply.id) || reply.id < 0 || typeof reply.ok !== "boolean" || + (reply.ok && (!reply.result || typeof reply.result !== "object" || Array.isArray(reply.result))) || + (!reply.ok && (!reply.error || typeof reply.error.code !== "string"))) throw new Error(); const p = this.pending.get(reply.id); if (!p) continue; @@ -242,6 +254,8 @@ export class NativeClient { } private fail() { this.closed = true; + this.chunks = []; + this.bufferedBytes = 0; for (const p of this.pending.values()) { p.clean(); p.reject(new Failure("SESSION_CLOSED", p.action)); @@ -254,13 +268,13 @@ export class NativeClient { if (!this.closed) await this.request("disconnect", {}, 1000).catch(() => {}); this.child.stdin?.end(); - if (this.child.exitCode === null && this.child.signalCode === null) + if (!this.exited && this.child.exitCode === null && this.child.signalCode === null) await new Promise((resolve) => { const kill = setTimeout(() => { this.child.kill("SIGKILL"); - resolve(); }, 2000); - this.child.once("exit", () => { + // Keep the device lock until the child really exits. + this.child.once("close", () => { clearTimeout(kill); resolve(); }); diff --git a/apps/mcp/src/server.ts b/apps/mcp/src/server.ts index 3552470e..6eb48a20 100644 --- a/apps/mcp/src/server.ts +++ b/apps/mcp/src/server.ts @@ -203,17 +203,28 @@ export function createServer( "Connect only a saved, previously authorized device with available credentials. AUTH_REQUIRED means the user must authorize through the panel or setup.", async (p, ctx) => { const selected = sessions.choose(supported(ctx)); - const result = await sessions.connectSaved( - p.device_alias, - p.mode, - p.display_index, - ctx.mcpReq.signal, - ); + // Reserve panel capacity before starting a connection. Once connected, + // panel setup errors must not hide the live session from the caller. const v = sessions.create(selected, p.device_alias); + let result: Data; + try { + result = await sessions.connectSaved( + p.device_alias, + p.mode, + p.display_index, + ctx.mcpReq.signal, + ); + } catch (error) { + sessions.views.delete(v.id); + throw error; + } v.session = result.session_id; + let panelResult: Data; + try { panelResult = await openPanel(v); } + catch (error) { panelResult = { ...sessions.describe(v), panel_error: failure(error) }; } return viewResult(v, { ...result, - ...await openPanel(v), + ...panelResult, state: result.state, }); }, diff --git a/apps/mcp/src/sessions.ts b/apps/mcp/src/sessions.ts index 105fe57c..9eb46dff 100644 --- a/apps/mcp/src/sessions.ts +++ b/apps/mcp/src/sessions.ts @@ -2,6 +2,7 @@ import { join } from "node:path"; import { setTimeout as delay } from "node:timers/promises"; import { Failure, + failure, token, equal, parse, @@ -25,6 +26,7 @@ export interface View { alias?: string; session?: string; disconnected?: boolean; + connectionError?: Data; } interface Session { id: string; @@ -32,11 +34,15 @@ interface Session { native: NativeClient; status: Data; release: () => Promise; + connecting: boolean; + statusRequest?: Promise; + closing?: Promise; } export class Sessions { readonly views = new Map(); private current?: Session; - private connecting = false; + private connecting?: AbortController; + private connectionDone?: Promise; private closing = false; private binary?: Promise; constructor( @@ -94,11 +100,12 @@ export class Sessions { ? "not_confirmed" : "waiting", session_id: v.session ?? null, - state: v.session ? "session_available" : v.disconnected ? "closed" : "awaiting_user", + state: v.session ? "session_available" : v.disconnected ? "closed" : v.connectionError ? "failed" : "awaiting_user", + ...(v.connectionError ? { error: v.connectionError } : {}), device_alias: v.alias ?? null, next_step: v.session ? "Call crossdesk_get_session with view_id." - : v.disconnected + : v.disconnected || v.connectionError ? "The session was disconnected. Do not automatically reconnect." : v.panel === "off" ? "Use setup to configure credentials and authorization, then crossdesk_connect with the saved device alias. Never send passwords in chat." @@ -117,25 +124,31 @@ export class Sessions { const s = this.session(id!); // A panel connection may still be performing the native hello handshake. // Reading native status here would race initialization and could close it. - if (this.connecting) return { + if (s.connecting) return { ...(view ? this.describe(view) : {}), session_id: s.id, device_alias: s.device.alias, state: "connecting", control: false, }; - try { - s.status = await s.native.request("status", {}, 5000, signal); - } catch { - if (signal?.aborted) throw new Failure("CANCELLED"); - s.status = { state: "closed", control: false }; + // Concurrent panel/model polls share one native read. A cancelled waiter + // must not cancel another caller's status request or disconnect the session. + if (signal?.aborted) throw new Failure("CANCELLED"); + if (!["closed", "failed"].includes(s.status.state)) { + s.statusRequest ??= s.native.request("status", {}, 5000) + .then((status) => (s.status = status)) + .catch((error) => { + if (!(error instanceof Failure) || error.code !== "SESSION_CLOSED") throw error; + return (s.status = { state: "closed", control: false }); + }) + .finally(() => { s.statusRequest = undefined; }); + await this.waitStatus(s.statusRequest, signal); } // The user may have disconnected or replaced the session during the read. if (this.current !== s || (view && view.session !== id)) return view ? this.describe(view) : { session_id: id, state: "closed", control: false }; if (["closed", "failed"].includes(s.status.state)) { - await s.native.close(); - await s.release(); + await this.dispose(s); } return { ...s.status, @@ -150,6 +163,22 @@ export class Sessions { } : {}), }; } + private async waitStatus(request: Promise, signal?: AbortSignal) { + if (!signal) return request; + let abort: () => void = () => {}; + try { + return await Promise.race([request, new Promise((_, reject) => { + abort = () => reject(new Failure("CANCELLED")); + signal.addEventListener("abort", abort, { once: true }); + if (signal.aborted) abort(); + })]); + } finally { + signal.removeEventListener("abort", abort); + } + } + private dispose(s: Session): Promise { + return s.closing ??= s.native.close().finally(() => s.release()); + } async waitForConnection(viewId: string, waitMs: number, signal?: AbortSignal): Promise { const deadline = Date.now() + waitMs; let result = this.describe(this.view(viewId)); @@ -253,18 +282,23 @@ export class Sessions { ): Promise { if (this.closing) throw new Failure("SESSION_CLOSED"); if (this.connecting) throw new Failure("BUSY"); - if ( - this.current && - !["closed", "failed", "disconnected"].includes( - (await this.status(this.current.id)).state, - ) - ) - throw new Failure("SESSION_LIMIT"); - this.connecting = true; + // Reserve the single connection slot before the first await. + const controller = this.connecting = new AbortController(); + signal = signal ? AbortSignal.any([signal, controller.signal]) : controller.signal; + let complete!: () => void; + this.connectionDone = new Promise((resolve) => { complete = resolve; }); let release: (() => Promise) | undefined; let native: NativeClient | undefined; + let session: Session | undefined; try { + if (this.current && !["closed", "failed", "disconnected"].includes( + (await this.status(this.current.id, undefined, signal)).state, + )) throw new Failure("SESSION_LIMIT"); if (this.current) await this.disconnect(this.current.id); + if (view) { + view.disconnected = false; + view.connectionError = undefined; + } release = await this.store.lock( `device:${d.server}:${d.port}:${d.device_id}`, ); @@ -281,12 +315,13 @@ export class Sessions { join(this.store.directory, "logs"), this.videoOptions ?? (await this.store.config()).video, ); - const s: Session = { + const s: Session = session = { id: token(), device: d, native, status: {}, release, + connecting: true, }; this.current = s; if (view) { @@ -306,8 +341,11 @@ export class Sessions { while (Date.now() < deadline) { if (this.closing || signal?.aborted) throw new Failure("CANCELLED"); s.status = await native.request("status", {}, 5000, signal); - if (s.status.state === "ready" && s.status.frame_age_ms <= 2000) + if (this.current !== s || signal.aborted) throw new Failure("CANCELLED"); + if (s.status.state === "ready" && typeof s.status.frame_age_ms === "number" && s.status.frame_age_ms <= 2000) { + s.connecting = false; return { ...s.status, session_id: s.id, device_alias: s.device.alias }; + } if (["failed", "closed", "disconnected"].includes(s.status.state)) throw new Failure( s.status.error === "incorrect_password" @@ -318,14 +356,23 @@ export class Sessions { } throw new Failure("CONNECTION_TIMEOUT"); } catch (error) { - await native?.close(); - await release?.(); + if (session) await this.dispose(session); + else { + await native?.close(); + await release?.(); + } if (this.current?.native === native) this.current = undefined; - if (view) view.session = undefined; - throw error; + const reason = signal.aborted ? new Failure("CANCELLED") : error; + if (view && (!view.session || view.session === session?.id)) { + view.session = undefined; + if (!view.disconnected) view.connectionError = failure(reason); + } + throw reason; } finally { password = ""; - this.connecting = false; + this.connecting = undefined; + this.connectionDone = undefined; + complete(); } } async screenshot(p: Data, signal?: AbortSignal): Promise { @@ -369,14 +416,15 @@ export class Sessions { } async disconnect(id: string) { const s = this.session(id); - await s.native.close(); - await s.release(); - if (this.current === s) this.current = undefined; + if (s.connecting) this.connecting?.abort(); for (const v of this.views.values()) if (v.session === id) { v.session = undefined; v.disconnected = true; + v.connectionError = undefined; } + await this.dispose(s); + if (this.current === s) this.current = undefined; return { session_id: id, state: "closed" }; } async viewerControl(v: View, operation: string) { @@ -428,7 +476,10 @@ export class Sessions { } async close() { this.closing = true; + this.connecting?.abort(); + const pending = this.connectionDone; if (this.current) await this.disconnect(this.current.id); + await pending; this.views.clear(); } } diff --git a/apps/mcp/src/ui-actions.ts b/apps/mcp/src/ui-actions.ts index ff513101..04cb038b 100644 --- a/apps/mcp/src/ui-actions.ts +++ b/apps/mcp/src/ui-actions.ts @@ -35,7 +35,7 @@ export async function uiAction( }; } case "status": - return sessions.status(undefined, view.id); + return sessions.status(undefined, view.id, signal); case "connect": return sessions.connectUI(view, p, signal); case "control": diff --git a/libs/controller/include/controller_session.h b/libs/controller/include/controller_session.h index 1b04728f..4e56b3c7 100644 --- a/libs/controller/include/controller_session.h +++ b/libs/controller/include/controller_session.h @@ -70,6 +70,7 @@ public: void SetPreview(int fps, int width, int height, int quality = -1, int max_bitrate_bps = -1); void SetControl(bool enabled); + void ReleaseInput(); bool Text(const std::string &utf8); bool Key(int key, bool down); bool Pointer(float x, float y, int flag, int wheel = 0); @@ -93,7 +94,6 @@ private: Params Parameters(); bool Send(const char *stream, const std::string &payload); void Preview(); - void ReleaseInput(); static void Signal(SignalStatus, const char *, size_t, void *); static void Connection(ConnectionStatus, const char *, size_t, void *); static void Data(const char *, size_t, const char *, size_t, const char *, diff --git a/libs/controller/src/controller_session.cpp b/libs/controller/src/controller_session.cpp index 5dd41350..cf97ff41 100644 --- a/libs/controller/src/controller_session.cpp +++ b/libs/controller/src/controller_session.cpp @@ -288,6 +288,8 @@ void ControllerSession::Fail(const char *reason) { state_ = "failed"; control_ = false; ClosePeers(); + keys_.clear(); + buttons_.clear(); } void ControllerSession::Disconnect() { if (state_ == "closed" && !control_) @@ -295,6 +297,8 @@ void ControllerSession::Disconnect() { ReleaseInput(); control_ = false; ClosePeers(); + keys_.clear(); + buttons_.clear(); state_ = "closed"; } bool ControllerSession::SelectDisplay(int index) { @@ -367,30 +371,34 @@ void ControllerSession::SetControl(bool enabled) { control_ = enabled; } void ControllerSession::ReleaseInput() { - auto keys = keys_; - for (int k : keys) - Key(k, false); + // Release buttons before modifiers, so interrupted modified drags do not + // turn into an unmodified drop. Failed releases remain tracked for cleanup. auto buttons = buttons_; for (int b : buttons) Pointer(pointer_x_, pointer_y_, b + 1); - keys_.clear(); - buttons_.clear(); + auto keys = keys_; + for (auto it = keys.rbegin(); it != keys.rend(); ++it) + Key(*it, false); } bool ControllerSession::Key(int key, bool down) { - if (!control_ || !Connected() || key < 0 || key > 255) + if ((!control_ && (down || !keys_.count(key))) || !Connected() || key < 0 || + key > 255) return false; RemoteAction a{}; a.type = keyboard; a.k = {size_t(key), 0, false, down ? key_down : key_up}; if (down) keys_.insert(key); - else + const bool sent = Send(kKeyboardStream, a.to_json()); + if (sent && !down) keys_.erase(key); - return Send(kKeyboardStream, a.to_json()); + return sent; } bool ControllerSession::Pointer(float x, float y, int flag, int wheel) { - if (!control_ || !Connected() || !std::isfinite(x) || !std::isfinite(y) || - x < 0 || y < 0 || x > 1 || y > 1 || flag < 0 || flag > 8 || + const bool releasing = + flag >= 2 && flag <= 6 && !(flag & 1) && buttons_.count(flag - 1); + if ((!control_ && !releasing) || !Connected() || !std::isfinite(x) || + !std::isfinite(y) || x < 0 || y < 0 || x > 1 || y > 1 || flag < 0 || flag > 8 || std::abs(int64_t(wheel)) > 1200) return false; RemoteAction a{}; @@ -398,13 +406,12 @@ bool ControllerSession::Pointer(float x, float y, int flag, int wheel) { a.m = {x, y, wheel, static_cast(flag)}; pointer_x_ = x; pointer_y_ = y; - if (flag >= 1 && flag <= 6) { - if (flag & 1) - buttons_.insert(flag); - else - buttons_.erase(flag - 1); - } - return Send(kMouseStream, a.to_json()); + if (flag >= 1 && flag <= 6 && (flag & 1)) + buttons_.insert(flag); + const bool sent = Send(kMouseStream, a.to_json()); + if (sent && flag >= 2 && flag <= 6 && !(flag & 1)) + buttons_.erase(flag - 1); + return sent; } bool ControllerSession::Text(const std::string &text) { if (!control_ || !Connected() || !text_supported_ || text.empty() || diff --git a/libs/controller/src/frame_image.cpp b/libs/controller/src/frame_image.cpp index 15ff5ac5..da06676c 100644 --- a/libs/controller/src/frame_image.cpp +++ b/libs/controller/src/frame_image.cpp @@ -19,10 +19,15 @@ FrameImage EncodeFrameImage(const DecodedFrame &f, ImageRegion r, int max_edge, r.width > int(f.width) || r.height > int(f.height) || r.x > int(f.width) - r.width || r.y > int(f.height) - r.height) throw std::invalid_argument("INVALID_REGION"); - std::vector argb(size_t(f.width) * f.height * 4); - if (libyuv::NV12ToARGB(f.nv12.data(), f.width, - f.nv12.data() + size_t(f.width) * f.height, f.width, - argb.data(), f.width * 4, f.width, f.height)) + // Align the NV12 crop to its chroma grid, then offset the RGB crop by at + // most one pixel. Small observations do not need a full-frame RGB buffer. + const int left = r.x & ~1, top = r.y & ~1; + const int width = r.width + r.x - left, height = r.height + r.y - top; + std::vector argb(size_t(width) * height * 4); + if (libyuv::NV12ToARGB(f.nv12.data() + size_t(top) * f.width + left, f.width, + f.nv12.data() + size_t(f.width) * f.height + + size_t(top / 2) * f.width + left, + f.width, argb.data(), width * 4, width, height)) throw std::runtime_error("IMAGE_CONVERSION_FAILED"); FrameImage result; result.region = r; @@ -33,8 +38,8 @@ FrameImage EncodeFrameImage(const DecodedFrame &f, ImageRegion r, int max_edge, result.height = std::max(1, int(r.height * scale)); std::vector scaled(size_t(result.width) * result.height * 4); std::vector rgb(size_t(result.width) * result.height * 3); - if (libyuv::ARGBScale(argb.data() + (size_t(r.y) * f.width + r.x) * 4, - f.width * 4, r.width, r.height, scaled.data(), + if (libyuv::ARGBScale(argb.data() + (size_t(r.y - top) * width + r.x - left) * 4, + width * 4, r.width, r.height, scaled.data(), result.width * 4, result.width, result.height, libyuv::kFilterBox) || libyuv::ARGBToRAW(scaled.data(), result.width * 4, rgb.data(), @@ -45,23 +50,22 @@ FrameImage EncodeFrameImage(const DecodedFrame &f, ImageRegion r, int max_edge, image.width = result.width; image.height = result.height; image.format = PNG_FORMAT_RGB; - png_alloc_size_t size = 0; - if (!png_image_write_to_memory(&image, nullptr, &size, 0, rgb.data(), 0, - nullptr)) { - png_image_free(&image); - throw std::runtime_error("IMAGE_ENCODING_FAILED"); - } - if (size <= max_bytes) { - result.png.resize(size); - const auto ok = png_image_write_to_memory( - &image, result.png.data(), &size, 0, rgb.data(), 0, nullptr); - png_image_free(&image); - if (!ok) - throw std::runtime_error("IMAGE_ENCODING_FAILED"); + // libpng reports the required size on buffer exhaustion. Compress once + // into the bounded output, instead of compressing once to measure and + // again to write every successful screenshot. + const auto capacity = + std::min(png_alloc_size_t(max_bytes), PNG_IMAGE_PNG_SIZE_MAX(image)); + result.png.resize(capacity); + png_alloc_size_t size = capacity; + const auto ok = png_image_write_to_memory( + &image, result.png.data(), &size, 0, rgb.data(), 0, nullptr); + png_image_free(&image); + if (ok) { result.png.resize(size); return result; } - png_image_free(&image); + if (size <= max_bytes) + throw std::runtime_error("IMAGE_ENCODING_FAILED"); if (edge == 32) throw std::runtime_error("IMAGE_TOO_LARGE"); }