YangWX/crossdesk
1
0
Fork 0
mirror of https://github.com/kunkundi/crossdesk.git synced 2026-11-07 06:39:10 +08:00
Code Issues Packages Projects Releases Wiki Activity

[fix] stabilize MCP sessions and extend input controls

Browse Source
This commit is contained in:
dijunkun
2026-11-03 14:19:39 +08:00
parent e1ae449d36
commit fc1529ee2b
15 changed files with 596 additions and 318 deletions
Download Patch File Download Diff File Expand all files Collapse all files
apps/mcp/src/browser.ts
+1
View File
@@ -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",
apps/mcp/src/common.ts
+19 -10
View File
@@ -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"),
apps/mcp/src/native.ts
+38 -24
View File
@@ -95,8 +95,10 @@ export class NativeClient {
private child: ChildProcess;
private pending = new Map<number, Pending>();
private sequence = 0;
private buffer = "";
private chunks: Buffer[] = [];
private bufferedBytes = 0;
private closed = false;
private exited = false;
private closing?: Promise<void>;
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<Data> {
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<void>((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();
});
apps/mcp/src/server.ts
+18 -7
View File
@@ -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,
});
},
apps/mcp/src/sessions.ts
+80 -29
View File
@@ -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<void>;
connecting: boolean;
statusRequest?: Promise<Data>;
closing?: Promise<void>;
}
export class Sessions {
readonly views = new Map<string, View>();
private current?: Session;
private connecting = false;
private connecting?: AbortController;
private connectionDone?: Promise<void>;
private closing = false;
private binary?: Promise<string>;
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<Data>, signal?: AbortSignal) {
if (!signal) return request;
let abort: () => void = () => {};
try {
return await Promise.race([request, new Promise<never>((_, 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<void> {
return s.closing ??= s.native.close().finally(() => s.release());
}
async waitForConnection(viewId: string, waitMs: number, signal?: AbortSignal): Promise<Data> {
const deadline = Date.now() + waitMs;
let result = this.describe(this.view(viewId));
@@ -253,18 +282,23 @@ export class Sessions {
): Promise<Data> {
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<void>((resolve) => { complete = resolve; });
let release: (() => Promise<void>) | 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<Data> {
@@ -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();
}
}
apps/mcp/src/ui-actions.ts
+1 -1
View File
@@ -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":