From ff5f586d2b1020ab270aa68f93ffd45b6ce6af9f Mon Sep 17 00:00:00 2001 From: Julius Marminge Date: Mon, 19 Oct 2026 10:52:16 +0000 Subject: [PATCH 1/6] feat(server): manage opt-in Cua Driver lifecycle --- apps/server/src/cua/CuaDriver.test.ts | 387 ++++++++++++++++++ apps/server/src/cua/CuaDriver.ts | 376 +++++++++++++++++ .../DesktopTelemetryReceiver.ts | 24 +- apps/server/src/server.ts | 7 + packages/contracts/src/cua.ts | 38 ++ packages/contracts/src/index.ts | 1 + packages/contracts/src/resourceTelemetry.ts | 3 + packages/contracts/src/settings.ts | 2 + 8 files changed, 837 insertions(+), 1 deletion(-) create mode 100644 apps/server/src/cua/CuaDriver.test.ts create mode 100644 apps/server/src/cua/CuaDriver.ts create mode 100644 packages/contracts/src/cua.ts diff --git a/apps/server/src/cua/CuaDriver.test.ts b/apps/server/src/cua/CuaDriver.test.ts new file mode 100644 index 0000000000000..78de4c7191807 --- /dev/null +++ b/apps/server/src/cua/CuaDriver.test.ts @@ -0,0 +1,387 @@ +import * as NodeCrypto from "@effect/platform-node/NodeCrypto"; +import { expect, it } from "@effect/vitest"; +import { + DEFAULT_SERVER_SETTINGS, + type CuaDriverMcpConfiguration, + type DesktopCuaDriverReport, +} from "@t3tools/contracts"; +import type { EmbeddedDriverConnection } from "@trycua/cua-driver/embedded"; +import * as HostProcess from "@t3tools/shared/HostProcess"; +import * as Deferred from "effect/Deferred"; +import * as Effect from "effect/Effect"; +import * as Exit from "effect/Exit"; +import * as Fiber from "effect/Fiber"; +import * as Option from "effect/Option"; +import * as PubSub from "effect/PubSub"; +import * as Ref from "effect/Ref"; +import * as Scope from "effect/Scope"; +import * as Stream from "effect/Stream"; +import * as TestClock from "effect/testing/TestClock"; + +import * as DesktopTelemetryReceiver from "../resourceTelemetry/DesktopTelemetryReceiver.ts"; +import * as ServerSettings from "../serverSettings.ts"; +import * as CuaDriver from "./CuaDriver.ts"; + +const mcp: CuaDriverMcpConfiguration = { command: "/driver", args: ["mcp"], environment: [] }; + +const makeHost = Effect.fnUntraced(function* () { + const started = yield* Deferred.make(); + const stopped = yield* Deferred.make(); + const watching = yield* Deferred.make(); + const ready = yield* Deferred.make(); + const exited = yield* Deferred.make(); + let stops = 0; + const host: CuaDriver.CuaDriverHost = { + start: Deferred.succeed(started, undefined).pipe(Effect.andThen(Deferred.await(ready))), + stop: Effect.sync(() => { + stops++; + }).pipe(Effect.andThen(Deferred.succeed(stopped, undefined)), Effect.asVoid), + waitForExit: Deferred.succeed(watching, undefined).pipe(Effect.andThen(Deferred.await(exited))), + }; + return { host, started, stopped, watching, ready, exited, stops: () => stops }; +}); + +const makeService = Effect.fnUntraced(function* ( + createHost: Parameters[0], + initialEnabled = true, +) { + const settings = yield* ServerSettings.ServerSettingsService; + const values = yield* settings.getSettings; + const enabled = yield* Ref.make(initialEnabled); + const changes = yield* PubSub.unbounded(); + const service = yield* CuaDriver.make(createHost).pipe( + Effect.provideService(ServerSettings.ServerSettingsService, { + ...settings, + getSettings: Ref.get(enabled).pipe(Effect.map((enableCua) => ({ ...values, enableCua }))), + subscribeChanges: PubSub.subscribe(changes).pipe( + Effect.map((subscription) => + Stream.fromSubscription(subscription).pipe( + Stream.map((enableCua) => ({ ...values, enableCua })), + ), + ), + ), + }), + ); + return { + service, + setEnabled: (value: boolean) => + Ref.set(enabled, value).pipe(Effect.andThen(PubSub.publish(changes, value))), + }; +}); + +it.layer(ServerSettings.layerTest())("CuaDriver lifecycle", (it) => { + it.effect("defaults to disabled and never acquires a host before opt-in", () => + Effect.gen(function* () { + expect(DEFAULT_SERVER_SETTINGS.enableCua).toBe(false); + const { service } = yield* makeService(Effect.die("unexpected host acquisition"), false); + expect(yield* service.enabled).toBe(false); + expect(yield* service.acquire).toEqual(Option.none()); + }).pipe(Effect.scoped), + ); + + it.effect("shares a lazy start, survives one caller cancellation, and owns cleanup", () => + Effect.gen(function* () { + const h = yield* makeHost(); + let creates = 0; + yield* Effect.gen(function* () { + const { service } = yield* makeService( + Effect.sync(() => { + creates++; + return h.host; + }), + ); + expect(creates).toBe(0); + const first = yield* service.acquire.pipe(Effect.forkChild); + yield* Deferred.await(h.started); + const second = yield* service.acquire.pipe(Effect.forkChild({ startImmediately: true })); + yield* Fiber.interrupt(first); + yield* Deferred.succeed(h.ready, mcp); + expect(yield* Fiber.join(second)).toEqual(Option.some(mcp)); + expect(yield* service.acquire).toEqual(Option.some(mcp)); + expect(creates).toBe(1); + expect(h.stops()).toBe(0); + }).pipe(Effect.scoped); + expect(h.stops()).toBe(1); + }), + ); + + it.effect("retries failed starts, clears the failure, and replaces a crashed connection", () => + Effect.gen(function* () { + const failed = yield* makeHost(); + const crashed = yield* makeHost(); + const replacement = yield* makeHost(); + const hosts = [failed, crashed, replacement]; + let creates = 0; + const { service } = yield* makeService(Effect.sync(() => hosts[creates++]!.host)); + yield* Deferred.fail( + failed.ready, + new CuaDriver.CuaDriverStartError({ + binaryPath: "/driver", + cause: new Error("native start failed"), + }), + ); + expect(yield* service.acquire).toEqual(Option.none()); + expect(failed.stops()).toBe(1); + expect(Option.isSome(yield* service.lastFailure)).toBe(true); + yield* Deferred.succeed(crashed.ready, mcp); + expect(yield* service.acquire).toEqual(Option.some(mcp)); + expect(yield* service.lastFailure).toEqual(Option.none()); + yield* Deferred.await(crashed.watching); + yield* Deferred.succeed(crashed.exited, undefined); + yield* Deferred.await(crashed.stopped); + const next = { ...mcp, args: ["mcp", "--generation=next"] }; + yield* Deferred.succeed(replacement.ready, next); + expect(yield* service.acquire).toEqual(Option.some(next)); + expect(creates).toBe(3); + }).pipe(Effect.scoped), + ); + + it.effect("disabling stops the active host and re-enabling starts a fresh one", () => + Effect.gen(function* () { + const first = yield* makeHost(); + const second = yield* makeHost(); + let creates = 0; + const h = yield* makeService(Effect.sync(() => (creates++ === 0 ? first.host : second.host))); + yield* Deferred.succeed(first.ready, mcp); + expect(yield* h.service.acquire).toEqual(Option.some(mcp)); + yield* h.setEnabled(false); + yield* Deferred.await(first.stopped); + expect(yield* h.service.acquire).toEqual(Option.none()); + expect(creates).toBe(1); + yield* h.setEnabled(true); + yield* Deferred.succeed(second.ready, mcp); + expect(yield* h.service.acquire).toEqual(Option.some(mcp)); + expect(creates).toBe(2); + expect(first.stops()).toBe(1); + }).pipe(Effect.scoped), + ); + + it.effect("a revoked start cannot publish its late descriptor or stop its replacement", () => + Effect.gen(function* () { + const first = yield* makeHost(); + const second = yield* makeHost(); + const lateFinished = yield* Deferred.make(); + let creates = 0; + const h = yield* makeService( + Effect.sync(() => + creates++ === 0 + ? { + ...first.host, + start: first.host.start.pipe( + Effect.uninterruptible, + Effect.ensuring(Deferred.succeed(lateFinished, undefined)), + ), + } + : second.host, + ), + ); + const pending = yield* h.service.acquire.pipe(Effect.forkChild); + yield* Deferred.await(first.started); + yield* h.setEnabled(false); + yield* Deferred.await(first.stopped); + expect(yield* Fiber.join(pending)).toEqual(Option.none()); + yield* h.setEnabled(true); + const next = { ...mcp, command: "/replacement" }; + yield* Deferred.succeed(second.ready, next); + expect(yield* h.service.acquire).toEqual(Option.some(next)); + yield* Deferred.succeed(first.ready, mcp); + yield* Deferred.await(lateFinished); + expect(yield* h.service.acquire).toEqual(Option.some(next)); + expect(first.stops()).toBe(1); + expect(second.stops()).toBe(0); + }).pipe(Effect.scoped), + ); +}); + +const nativeFixture = Effect.fnUntraced(function* () { + const start = Promise.withResolvers(); + const stop = Promise.withResolvers(); + const started = yield* Deferred.make(); + const stopping = yield* Deferred.make(); + const destroyed = yield* Deferred.make(); + const connection: EmbeddedDriverConnection = { + socketPath: "/socket", + pid: 1, + generation: "first", + driverVersion: "test", + contractVersion: "test", + mcpProtocolVersion: "test", + mcp: { command: mcp.command, args: [...mcp.args], environment: [...mcp.environment] }, + }; + let destroys = 0; + let creates = 0; + const factory = yield* CuaDriver.makeStandaloneHostFactory( + Effect.succeed("/driver"), + Effect.succeed({ + EmbeddedCuaDriverHost: class { + constructor() { + creates++; + } + start() { + Deferred.doneUnsafe(started, Effect.void); + return start.promise; + } + stop() { + Deferred.doneUnsafe(stopping, Effect.void); + return stop.promise; + } + waitForExit(_generation: string, options?: { signal: AbortSignal }) { + return new Promise((_resolve, reject) => { + if (options?.signal.aborted) reject(new Error("aborted")); + else + options?.signal.addEventListener("abort", () => reject(new Error("aborted")), { + once: true, + }); + }); + } + uniffiDestroy() { + destroys++; + Deferred.doneUnsafe(destroyed, Effect.void); + } + }, + }), + ); + return { + factory, + connection, + start, + stop, + started, + stopping, + destroyed, + creates: () => creates, + destroys: () => destroys, + }; +}); + +it.layer(ServerSettings.layerTest())("CuaDriver native ownership", (it) => { + it.effect.each(["start", "stop"] as const)( + "bounds shutdown with native %s pending without destroying or reusing its owner", + (pending) => + Effect.gen(function* () { + const f = yield* nativeFixture(); + const scope = yield* Scope.make(); + const { service } = yield* makeService(f.factory).pipe( + Effect.provideService(Scope.Scope, scope), + ); + const acquiring = yield* service.acquire.pipe(Effect.forkChild); + yield* Deferred.await(f.started); + if (pending === "stop") { + f.start.resolve(f.connection); + expect(yield* Fiber.join(acquiring)).toEqual( + Option.some({ ...mcp, socketPath: "/socket" }), + ); + } + const closing = yield* Scope.close(scope, Exit.void).pipe(Effect.forkChild); + if (pending === "stop") yield* Deferred.await(f.stopping); + yield* TestClock.adjust("5 seconds"); + yield* Fiber.join(closing); + expect(f.destroys()).toBe(0); + expect(Exit.isFailure(yield* Effect.exit(f.factory))).toBe(true); + expect(f.creates()).toBe(1); + f.start.resolve(f.connection); + yield* Deferred.await(f.stopping); + expect(f.destroys()).toBe(0); + f.stop.resolve(); + yield* Deferred.await(f.destroyed); + const replacement = yield* f.factory; + expect(f.creates()).toBe(2); + yield* replacement.stop; + expect(f.destroys()).toBe(2); + expect(yield* service.acquire).toEqual(Option.none()); + }).pipe(Effect.scoped), + ); +}); + +it.effect.each([false, true])( + "uses the supplied desktop receiver for correlated replies (synchronous: %s) and acknowledged stop", + (synchronous) => + Effect.gen(function* () { + const reports = yield* PubSub.unbounded(); + const requested = yield* Deferred.make(); + const requests: boolean[] = []; + const factory = yield* CuaDriver.makeDesktopHostFactory().pipe( + Effect.provide( + DesktopTelemetryReceiver.layerTest({ + cuaReports: Stream.fromPubSub(reports), + requestCuaDriver: (requestId, enabled) => + Effect.gen(function* () { + requests.push(enabled); + if (enabled) { + yield* Deferred.succeed(requested, requestId); + if (synchronous) + yield* PubSub.publish(reports, { + version: 1, + type: "cuaDriverReport", + requestId, + status: "ready", + mcp, + }); + } else + yield* PubSub.publish(reports, { + version: 1, + type: "cuaDriverReport", + requestId, + status: "stopped", + }); + }), + }), + ), + ); + yield* Effect.gen(function* () { + const { service } = yield* makeService(factory); + const acquiring = yield* service.acquire.pipe(Effect.forkChild); + const requestId = yield* Deferred.await(requested); + yield* PubSub.publish(reports, { + version: 1, + type: "cuaDriverReport", + requestId: "unrelated", + status: "ready", + mcp: { ...mcp, command: "/wrong" }, + }); + yield* PubSub.publish(reports, { + version: 1, + type: "cuaDriverReport", + requestId, + status: "ready", + mcp, + }); + expect(yield* Fiber.join(acquiring)).toEqual(Option.some(mcp)); + expect(yield* service.acquire).toEqual(Option.some(mcp)); + }).pipe(Effect.scoped); + expect(requests).toEqual([true, false]); + }).pipe(Effect.provide([ServerSettings.layerTest(), NodeCrypto.layer]), Effect.scoped), +); + +it.effect("uses an explicit driver path without consulting the installer", () => + Effect.gen(function* () { + let binaryPath = ""; + const factory = yield* CuaDriver.makeStandaloneHostFactory( + Effect.die("installer must not run"), + Effect.succeed({ + EmbeddedCuaDriverHost: class { + constructor(path: string) { + binaryPath = path; + } + start() { + return Promise.reject(new Error("unused")); + } + stop() { + return Promise.resolve(); + } + waitForExit() { + return Promise.reject(new Error("unused")); + } + uniffiDestroy() {} + }, + }), + ).pipe( + Effect.provideService(HostProcess.Environment, { + T3CODE_CUA_DRIVER_PATH: "/explicit/driver", + }), + ); + const host = yield* factory; + expect(binaryPath).toBe("/explicit/driver"); + yield* host.stop; + }), +); diff --git a/apps/server/src/cua/CuaDriver.ts b/apps/server/src/cua/CuaDriver.ts new file mode 100644 index 0000000000000..27ff61b543005 --- /dev/null +++ b/apps/server/src/cua/CuaDriver.ts @@ -0,0 +1,376 @@ +import { CuaDriverMcpConfiguration } from "@t3tools/contracts"; +import type { EmbeddedCuaDriverHost, EmbeddedDriverConnection } from "@trycua/cua-driver/embedded"; +import * as HostProcess from "@t3tools/shared/HostProcess"; +import * as Cause from "effect/Cause"; +import * as Context from "effect/Context"; +import * as Crypto from "effect/Crypto"; +import * as Deferred from "effect/Deferred"; +import * as Effect from "effect/Effect"; +import * as Exit from "effect/Exit"; +import * as Fiber from "effect/Fiber"; +import * as Layer from "effect/Layer"; +import * as Option from "effect/Option"; +import * as Schema from "effect/Schema"; +import * as Scope from "effect/Scope"; +import * as Stream from "effect/Stream"; + +import * as ServerConfig from "../config.ts"; +import * as DesktopTelemetryReceiver from "../resourceTelemetry/DesktopTelemetryReceiver.ts"; +import * as ServerSettings from "../serverSettings.ts"; +import * as CuaDriverInstallation from "./CuaDriverInstallation.ts"; + +export class CuaDriverStartError extends Schema.TaggedError()( + "CuaDriverStartError", + { binaryPath: Schema.String, cause: Schema.Defect() }, +) { + override get message() { + return "Failed to start Cua Driver."; + } +} + +export class CuaDriverHostError extends Schema.TaggedError()( + "CuaDriverHostError", + { + operation: Schema.Literals(["busy", "start", "stop", "exit", "desktop", "timeout"]), + cause: Schema.optional(Schema.Defect()), + }, +) { + override get message() { + return `Cua Driver host failed during ${this.operation}.`; + } +} + +export type CuaDriverError = CuaDriverStartError | CuaDriverHostError; + +const decodeMcp = Schema.decodeEffect(CuaDriverMcpConfiguration); +const isStartError = Schema.is(CuaDriverStartError); + +export interface CuaDriverHost { + readonly start: Effect.Effect; + readonly stop: Effect.Effect; + readonly waitForExit: Effect.Effect; +} + +export class CuaDriver extends Context.Service< + CuaDriver, + { + readonly acquire: Effect.Effect>; + readonly enabled: Effect.Effect; + readonly lastFailure: Effect.Effect>; + } +>()("t3/cua/CuaDriver") {} + +interface Acquisition { + readonly result: Deferred.Deferred>; + host?: CuaDriverHost; + cleanup?: Fiber.Fiber; + watcher?: Fiber.Fiber | undefined; + revoked: boolean; +} + +export const make = Effect.fn("CuaDriver.make")(function* ( + factory: Effect.Effect, +) { + const settings = yield* ServerSettings.ServerSettingsService; + const changes = yield* settings.subscribeChanges; + let active: Acquisition | undefined; + let closed = false; + let failure = Option.none(); + + const stop = Effect.fnUntraced(function* (entry: Acquisition) { + if (entry.watcher) yield* Fiber.interrupt(entry.watcher).pipe(Effect.forkDetach); + if (!entry.cleanup && entry.host) { + entry.cleanup = yield* entry.host.stop.pipe( + Effect.catch((error) => + Effect.sync(() => { + if (active === entry || active === undefined) failure = Option.some(error); + }), + ), + Effect.ignoreCause({ log: true }), + Effect.forkDetach({ startImmediately: true }), + ); + } + if (entry.cleanup) yield* Fiber.join(entry.cleanup).pipe(Effect.timeoutOption("4 seconds")); + }); + + const revoke = Effect.fnUntraced(function* (entry: Acquisition) { + entry.revoked = true; + if (active === entry) active = undefined; + yield* Deferred.succeed(entry.result, Option.none()); + yield* stop(entry); + }); + + yield* Effect.addFinalizer(() => + Effect.gen(function* () { + closed = true; + if (active) yield* revoke(active); + }), + ); + yield* changes.pipe( + Stream.runForEach((value) => + Effect.gen(function* () { + if (!value.enableCua && active) yield* revoke(active); + }), + ), + Effect.forkScoped, + ); + + const start = Effect.fnUntraced( + function* (entry: Acquisition) { + if (entry.revoked) return; + entry.host = yield* factory; + if (entry.revoked) { + yield* stop(entry); + return; + } + const mcp = yield* entry.host.start; + if (entry.revoked) return; + failure = Option.none(); + yield* Deferred.succeed(entry.result, Option.some(mcp)); + entry.watcher = yield* entry.host.waitForExit.pipe( + Effect.catch((error) => + Effect.sync(() => { + if (active === entry) failure = Option.some(error); + }), + ), + Effect.andThen( + Effect.gen(function* () { + if (active !== entry) return; + failure = Option.some(new CuaDriverHostError({ operation: "exit" })); + // The watcher must not interrupt itself while retiring its host. + entry.watcher = undefined; + yield* revoke(entry); + }), + ), + Effect.forkDetach, + ); + }, + (effect, entry) => + effect.pipe( + Effect.catchCause((cause) => + Effect.gen(function* () { + if (active !== entry) return; + failure = Option.some( + Option.getOrElse( + Cause.findErrorOption(cause), + () => new CuaDriverHostError({ operation: "start", cause: Cause.squash(cause) }), + ), + ); + yield* stop(entry); + yield* revoke(entry); + }), + ), + ), + ); + + const enabled = settings.getSettings.pipe( + Effect.map((value) => !closed && value.enableCua), + Effect.catch((cause) => + Effect.sync(() => { + failure = Option.some(new CuaDriverHostError({ operation: "start", cause })); + return false; + }), + ), + ); + const acquire = Effect.gen(function* () { + if (!(yield* enabled)) { + if (active) yield* revoke(active); + return Option.none(); + } + let entry = active; + if (!entry) { + entry = { + result: yield* Deferred.make>(), + revoked: false, + }; + active = entry; + yield* start(entry).pipe(Effect.forkDetach); + } + const result = yield* Deferred.await(entry.result).pipe(Effect.timeoutOption("30 seconds")); + if (Option.isSome(result)) return result.value; + if (active === entry) { + failure = Option.some(new CuaDriverHostError({ operation: "timeout" })); + yield* revoke(entry); + } + return Option.none(); + }); + return CuaDriver.of({ acquire, enabled, lastFailure: Effect.sync(() => failure) }); +}); + +type NativeModule = { + readonly EmbeddedCuaDriverHost: new ( + binaryPath: string, + bundleId: string, + ) => Pick; +}; + +export const makeStandaloneHostFactory = Effect.fn("CuaDriver.makeStandaloneHostFactory")( + function* (executable: Effect.Effect, module?: Effect.Effect) { + const environment = yield* HostProcess.Environment; + let owned = false; + return Effect.gen(function* () { + if (owned) return yield* new CuaDriverHostError({ operation: "busy" }); + const binaryPath = + environment.T3CODE_CUA_DRIVER_PATH?.trim() || + (yield* executable.pipe( + Effect.mapError((cause) => new CuaDriverHostError({ operation: "start", cause })), + )); + const sdk = yield* module === undefined + ? Effect.tryPromise({ + try: () => import("@trycua/cua-driver/embedded"), + catch: (cause) => new CuaDriverStartError({ binaryPath, cause }), + }) + : module.pipe(Effect.mapError((cause) => new CuaDriverStartError({ binaryPath, cause }))); + // Acquisition can suspend during installation/import, so check ownership again. + if (owned) return yield* new CuaDriverHostError({ operation: "busy" }); + const native = yield* Effect.try({ + try: () => new sdk.EmbeddedCuaDriverHost(binaryPath, "com.t3tools.t3code.server"), + catch: (cause) => new CuaDriverStartError({ binaryPath, cause }), + }); + owned = true; + let starting: Promise | undefined; + let stopping: Promise | undefined; + const watchers = new Map>(); + + const start = Effect.tryPromise({ + try: () => { + if (stopping) return Promise.reject(new CuaDriverHostError({ operation: "start" })); + starting ??= native.start(); + return starting; + }, + catch: (cause) => new CuaDriverStartError({ binaryPath, cause }), + }).pipe( + Effect.flatMap((connection) => + decodeMcp({ + ...connection.mcp, + socketPath: connection.socketPath, + }), + ), + Effect.mapError((cause) => + isStartError(cause) ? cause : new CuaDriverStartError({ binaryPath, cause }), + ), + ); + + const stop = Effect.tryPromise({ + try: () => + (stopping ??= (async () => { + // UniFFI cancellation ends the JS wait before native work necessarily settles. + // Retain the owner until every owned operation has settled, even after public shutdown. + await starting?.catch(() => undefined); + for (const controller of watchers.keys()) controller.abort(); + await Promise.allSettled(watchers.values()); + try { + await native.stop(); + } finally { + owned = false; + native.uniffiDestroy(); + } + })()), + catch: (cause) => new CuaDriverHostError({ operation: "stop", cause }), + }); + const waitForExit = Effect.tryPromise({ + try: async (signal) => { + const connection = await starting; + if (!connection || stopping) return; + const controller = new AbortController(); + const abort = () => controller.abort(); + signal.addEventListener("abort", abort, { once: true }); + if (signal.aborted) controller.abort(); + const waiting = native.waitForExit(connection.generation, { signal: controller.signal }); + watchers.set(controller, waiting); + try { + await waiting; + } finally { + watchers.delete(controller); + signal.removeEventListener("abort", abort); + } + }, + catch: (cause) => new CuaDriverHostError({ operation: "exit", cause }), + }); + return { start, stop, waitForExit } satisfies CuaDriverHost; + }); + }, +); + +export const makeDesktopHostFactory = Effect.fn("CuaDriver.makeDesktopHostFactory")(function* () { + const receiver = yield* DesktopTelemetryReceiver.DesktopTelemetryReceiver; + const crypto = yield* Crypto.Crypto; + return Effect.gen(function* () { + const requestId = yield* crypto + .randomUUIDv4() + .pipe(Effect.mapError((cause) => new CuaDriverHostError({ operation: "desktop", cause }))); + const scope = yield* Scope.make(); + const ready = yield* Deferred.make(); + const stopped = yield* Deferred.make(); + let requested = false; + const unavailable = new CuaDriverHostError({ operation: "desktop" }); + const start = Effect.gen(function* () { + // Start the subscription before writing: the desktop may reply synchronously. + yield* receiver.cuaReports.pipe( + Stream.runForEach((report) => + Effect.gen(function* () { + if (report.requestId !== requestId) return; + if (report.status === "ready") yield* Deferred.succeed(ready, report.mcp); + else { + yield* Deferred.fail(ready, unavailable); + yield* Deferred.succeed(stopped, undefined); + } + }), + ), + Effect.ensuring( + Deferred.fail(ready, unavailable).pipe( + Effect.andThen(Deferred.succeed(stopped, undefined)), + ), + ), + Effect.forkIn(scope, { startImmediately: true }), + ); + const health = yield* receiver.subscribeHealth.pipe( + Effect.provideService(Scope.Scope, scope), + ); + if (health.latest.status === "unavailable" || health.latest.status === "stopped") + return yield* unavailable; + yield* health.changes.pipe( + Stream.runForEach((value) => + value.status === "stopped" || + value.status === "unavailable" || + value.status === "degraded" + ? Deferred.fail(ready, unavailable).pipe( + Effect.andThen(Deferred.succeed(stopped, undefined)), + ) + : Effect.void, + ), + Effect.forkIn(scope, { startImmediately: true }), + ); + requested = true; + yield* receiver + .requestCuaDriver(requestId, true) + .pipe(Effect.mapError((cause) => new CuaDriverHostError({ operation: "desktop", cause }))); + return yield* Deferred.await(ready); + }); + const stop = Effect.gen(function* () { + if (!requested) return; + yield* receiver + .requestCuaDriver(requestId, false) + .pipe(Effect.mapError((cause) => new CuaDriverHostError({ operation: "desktop", cause }))); + yield* Deferred.await(stopped).pipe(Effect.timeoutOption("3 seconds")); + }).pipe(Effect.ensuring(Scope.close(scope, Exit.void))); + return { start, stop, waitForExit: Deferred.await(stopped) } satisfies CuaDriverHost; + }); +}); + +export const layer = Layer.effect( + CuaDriver, + Effect.gen(function* () { + const config = yield* ServerConfig.ServerConfig; + const factory = + config.mode === "desktop" + ? yield* makeDesktopHostFactory() + : yield* Effect.gen(function* () { + const installation = yield* CuaDriverInstallation.makeCuaDriverInstallation({ + baseDir: config.baseDir, + }); + return yield* makeStandaloneHostFactory(installation.executable); + }); + return yield* make(factory); + }), +); diff --git a/apps/server/src/resourceTelemetry/DesktopTelemetryReceiver.ts b/apps/server/src/resourceTelemetry/DesktopTelemetryReceiver.ts index a911c2b05d3d6..d94c7523e5b50 100644 --- a/apps/server/src/resourceTelemetry/DesktopTelemetryReceiver.ts +++ b/apps/server/src/resourceTelemetry/DesktopTelemetryReceiver.ts @@ -5,6 +5,7 @@ import * as NodeNet from "node:net"; import * as NodeStream from "@effect/platform-node/NodeStream"; import { DesktopHostTelemetryMessage, + type DesktopCuaDriverReport, type DesktopHostTelemetryMessage as DesktopHostTelemetryMessageValue, type DesktopHostTelemetrySnapshot, DesktopTelemetryControlMessage, @@ -190,6 +191,11 @@ export class DesktopTelemetryReceiver extends Context.Service< never, Scope.Scope >; + readonly cuaReports: Stream.Stream; + readonly requestCuaDriver: ( + requestId: string, + enabled: boolean, + ) => Effect.Effect; } >()("t3/resourceTelemetry/DesktopTelemetryReceiver") {} @@ -344,6 +350,7 @@ export const make = Effect.fn("resourceTelemetry.desktopTelemetryReceiver.make") const healthChanges = yield* PubSub.sliding(4); const latestUpdateReport = yield* Ref.make(Option.none()); const updateReportChanges = yield* PubSub.sliding(16); + const cuaReports = yield* PubSub.unbounded(); const controlMutex = yield* Semaphore.make(1); const snapshotMutex = yield* Semaphore.make(1); const health = yield* Ref.make({ @@ -529,6 +536,13 @@ export const make = Effect.fn("resourceTelemetry.desktopTelemetryReceiver.make") ); } + if (message.type === "cuaDriverReport") { + return recordContact.pipe( + Effect.andThen(PubSub.publish(cuaReports, message)), + Effect.asVoid, + ); + } + const sampledAt = DateTime.makeUnsafe(message.sampledAtUnixMs); return snapshotMutex.withPermits(1)( recordContact.pipe( @@ -623,6 +637,11 @@ export const make = Effect.fn("resourceTelemetry.desktopTelemetryReceiver.make") } return DesktopTelemetryReceiver.of({ + cuaReports: Stream.fromPubSub(cuaReports), + requestCuaDriver: (requestId, enabled) => + config.desktopTelemetryControlFd === undefined || config.desktopTelemetryFd === undefined + ? Effect.fail(new DesktopTelemetryDescriptorUnavailable({ mode: config.mode })) + : sendControlMessage({ version: 1, type: "cuaDriverRequest", requestId, enabled }), latest: Ref.get(latest), changes: Stream.fromPubSub(changes), subscribe: snapshotMutex.withPermits(1)( @@ -669,7 +688,7 @@ export const layerTest = ( const health = overrides.health ?? Effect.succeed({ - status: "unavailable" as const, + status: overrides.cuaReports === undefined ? ("unavailable" as const) : ("healthy" as const), lastSampleAt: Option.none(), lastError: Option.some("Desktop telemetry test implementation is unavailable."), }); @@ -699,6 +718,9 @@ export const layerTest = ( requestDesktopUpdate: () => Effect.void, commitDesktopUpdate: () => Effect.void, cancelDesktopUpdate: () => Effect.void, + cuaReports: Stream.empty, + requestCuaDriver: () => + Effect.fail(new DesktopTelemetryDescriptorUnavailable({ mode: "test" })), desktopUpdates: overrides.desktopUpdates ?? Effect.succeed({ diff --git a/apps/server/src/server.ts b/apps/server/src/server.ts index 59daedc151071..6e6938cc8e2ef 100644 --- a/apps/server/src/server.ts +++ b/apps/server/src/server.ts @@ -152,6 +152,7 @@ import * as CloudCliTokenManager from "./cloud/CliTokenManager.ts"; import * as CloudCliState from "./cloud/CliState.ts"; import * as ServerSelfUpdate from "./cloud/selfUpdate.ts"; import * as DesktopAppUpdate from "./desktopUpdate/DesktopAppUpdate.ts"; +import * as CuaDriver from "./cua/CuaDriver.ts"; import * as ServiceLauncherClient from "./cloud/serviceLauncherClient.ts"; import * as ProcessDiagnostics from "./diagnostics/ProcessDiagnostics.ts"; import * as HostResources from "./resourceTelemetry/HostResources.ts"; @@ -253,6 +254,11 @@ const layerDesktopAppUpdate = DesktopAppUpdate.layer.pipe( Layer.provide(layerDesktopTelemetryReceiver), ); +const layerCuaDriver = CuaDriver.layer.pipe( + Layer.provide(layerDesktopTelemetryReceiver), + Layer.provide(layerServerSettings), +); + const layerBackground = BackgroundPolicy.layer.pipe( Layer.provide(layerHostPowerMonitor), Layer.provideMerge(layerServerSettings), @@ -590,6 +596,7 @@ const layerRuntimeCoreDependenciesBase = Layer.mergeAll( Layer.provideMerge(layerOrchestrationApplication), Layer.provideMerge(RuntimeLayer.layerEventInfrastructure), Layer.provideMerge(Layer.merge(ProjectStore.layer, ThreadSearch.layer)), + Layer.provideMerge(layerCuaDriver), Layer.provideMerge(layerServerSettings), Layer.provideMerge(layerSourceControlProviderRegistry), Layer.provideMerge(layerGit), diff --git a/packages/contracts/src/cua.ts b/packages/contracts/src/cua.ts new file mode 100644 index 0000000000000..7fe8d9fc41928 --- /dev/null +++ b/packages/contracts/src/cua.ts @@ -0,0 +1,38 @@ +import * as Schema from "effect/Schema"; + +import { TrimmedNonEmptyString } from "./baseSchemas.ts"; + +/** Native host descriptor, confined to the desktop/server control pipe. */ +export const CuaDriverMcpConfiguration = Schema.Struct({ + command: TrimmedNonEmptyString, + args: Schema.Array(Schema.String), + environment: Schema.Array(Schema.Struct({ name: TrimmedNonEmptyString, value: Schema.String })), + socketPath: Schema.optionalKey(TrimmedNonEmptyString), +}); +export type CuaDriverMcpConfiguration = typeof CuaDriverMcpConfiguration.Type; + +export const DesktopCuaDriverRequest = Schema.Struct({ + version: Schema.Literal(1), + type: Schema.Literal("cuaDriverRequest"), + requestId: TrimmedNonEmptyString, + enabled: Schema.Boolean, +}); +export type DesktopCuaDriverRequest = typeof DesktopCuaDriverRequest.Type; + +export const DesktopCuaDriverReport = Schema.Union([ + Schema.Struct({ + version: Schema.Literal(1), + type: Schema.Literal("cuaDriverReport"), + requestId: TrimmedNonEmptyString, + status: Schema.Literal("ready"), + mcp: CuaDriverMcpConfiguration, + }), + Schema.Struct({ + version: Schema.Literal(1), + type: Schema.Literal("cuaDriverReport"), + requestId: TrimmedNonEmptyString, + status: Schema.Literals(["stopped", "unavailable"]), + message: Schema.optionalKey(Schema.String), + }), +]); +export type DesktopCuaDriverReport = typeof DesktopCuaDriverReport.Type; diff --git a/packages/contracts/src/index.ts b/packages/contracts/src/index.ts index 600d503ade39b..9fc8b3873b085 100644 --- a/packages/contracts/src/index.ts +++ b/packages/contracts/src/index.ts @@ -9,6 +9,7 @@ export * from "./environment.ts"; export * from "./environmentHttp.ts"; export * from "./relayClient.ts"; export * from "./desktopBootstrap.ts"; +export * from "./cua.ts"; export * from "./desktopBrowser.ts"; export * from "./desktopAppActivation.ts"; export * from "./remoteAccess.ts"; diff --git a/packages/contracts/src/resourceTelemetry.ts b/packages/contracts/src/resourceTelemetry.ts index 40220c603e957..adec89aa817ae 100644 --- a/packages/contracts/src/resourceTelemetry.ts +++ b/packages/contracts/src/resourceTelemetry.ts @@ -2,6 +2,7 @@ import * as Schema from "effect/Schema"; import { NonNegativeInt, PositiveInt, TrimmedNonEmptyString } from "./baseSchemas.ts"; import { HostPowerSnapshot } from "./background.ts"; +import { DesktopCuaDriverReport, DesktopCuaDriverRequest } from "./cua.ts"; import { DesktopUpdateStateSchema } from "./ipc.ts"; export const RESOURCE_MONITOR_PROTOCOL_VERSION = 3 as const; @@ -299,6 +300,7 @@ export const DesktopHostTelemetryMessage = Schema.Union([ DesktopHostTelemetryHello, DesktopHostTelemetrySnapshot, DesktopUpdateStatusReport, + DesktopCuaDriverReport, ]); export type DesktopHostTelemetryMessage = typeof DesktopHostTelemetryMessage.Type; @@ -350,6 +352,7 @@ export const DesktopTelemetryControlMessage = Schema.Union([ DesktopTelemetryRequestDesktopUpdate, DesktopTelemetryCommitDesktopUpdate, DesktopTelemetryCancelDesktopUpdate, + DesktopCuaDriverRequest, ]); export type DesktopTelemetryControlMessage = typeof DesktopTelemetryControlMessage.Type; diff --git a/packages/contracts/src/settings.ts b/packages/contracts/src/settings.ts index dc93afc809023..1fc502e18b59c 100644 --- a/packages/contracts/src/settings.ts +++ b/packages/contracts/src/settings.ts @@ -1124,6 +1124,7 @@ export const ServerSettings = Schema.Struct({ projectSettingsFolded: Schema.Boolean.pipe(Schema.withDecodingDefault(Effect.succeed(false))), enableAgentDeviceAccess: Schema.Boolean.pipe(Schema.withDecodingDefault(Effect.succeed(false))), + enableCua: Schema.Boolean.pipe(Schema.withDecodingDefault(Effect.succeed(false))), /** * Kept separate from agent access so enabling the user's Device panel does * not also grant providers control of simulators and emulators. @@ -1436,6 +1437,7 @@ export const ServerSettingsPatch = Schema.Struct({ ), enableAgentDeviceAccess: Schema.optionalKey(Schema.Boolean), enableDeviceSupport: Schema.optionalKey(Schema.Boolean), + enableCua: Schema.optionalKey(Schema.Boolean), deviceOnboardingCompleted: Schema.optionalKey(Schema.Boolean), deviceHosts: Schema.optionalKey(SshDeviceHostConfigs), sidebarAutoSettleAfterDays: Schema.optionalKey(Schema.NullOr(SidebarAutoSettleAfterDays)), From 98a57893c67b68c9b20bb8befc7f0199e0ef0856 Mon Sep 17 00:00:00 2001 From: Julius Marminge Date: Mon, 19 Oct 2026 11:42:52 +0000 Subject: [PATCH 2/6] fix(server): preserve Cua hosting in executable and stale telemetry --- apps/server/src/cua/CuaDriver.test.ts | 133 +++++++++++++----- apps/server/src/cua/CuaDriver.ts | 42 ++++-- .../DesktopTelemetryReceiver.ts | 2 +- 3 files changed, 136 insertions(+), 41 deletions(-) diff --git a/apps/server/src/cua/CuaDriver.test.ts b/apps/server/src/cua/CuaDriver.test.ts index 78de4c7191807..4c5fd6c5708e7 100644 --- a/apps/server/src/cua/CuaDriver.test.ts +++ b/apps/server/src/cua/CuaDriver.test.ts @@ -1,4 +1,5 @@ import * as NodeCrypto from "@effect/platform-node/NodeCrypto"; +import * as NodeServices from "@effect/platform-node/NodeServices"; import { expect, it } from "@effect/vitest"; import { DEFAULT_SERVER_SETTINGS, @@ -13,6 +14,7 @@ import * as Exit from "effect/Exit"; import * as Fiber from "effect/Fiber"; import * as Option from "effect/Option"; import * as PubSub from "effect/PubSub"; +import * as Queue from "effect/Queue"; import * as Ref from "effect/Ref"; import * as Scope from "effect/Scope"; import * as Stream from "effect/Stream"; @@ -240,7 +242,7 @@ const nativeFixture = Effect.fnUntraced(function* () { } }, }), - ); + ).pipe(Effect.provide(NodeServices.layer)); return { factory, connection, @@ -293,41 +295,73 @@ it.layer(ServerSettings.layerTest())("CuaDriver native ownership", (it) => { ); }); +const healthy: DesktopTelemetryReceiver.DesktopTelemetryReceiverHealth = { + status: "healthy", + lastSampleAt: Option.none(), + lastError: Option.none(), +}; + +const desktopFixture = Effect.fnUntraced(function* ( + synchronous = false, + healthChanges: Stream.Stream = Stream.empty, +) { + const reports = yield* PubSub.unbounded(); + const requested = yield* Deferred.make(); + const disabled = yield* Deferred.make(); + const watching = yield* Deferred.make(); + const requests: boolean[] = []; + const factory = yield* CuaDriver.makeDesktopHostFactory().pipe( + Effect.provide( + DesktopTelemetryReceiver.layerTest({ + health: Effect.succeed(healthy), + subscribeHealth: Effect.succeed({ latest: healthy, changes: healthChanges }), + cuaReports: Stream.fromPubSub(reports), + requestCuaDriver: (requestId, enabled) => + Effect.gen(function* () { + requests.push(enabled); + if (enabled) { + yield* Deferred.succeed(requested, requestId); + if (synchronous) + yield* PubSub.publish(reports, { + version: 1, + type: "cuaDriverReport", + requestId, + status: "ready", + mcp, + }); + } else { + yield* PubSub.publish(reports, { + version: 1, + type: "cuaDriverReport", + requestId, + status: "stopped", + }); + yield* Deferred.succeed(disabled, undefined); + } + }), + }), + ), + ); + return { + factory: factory.pipe( + Effect.map((host) => ({ + ...host, + waitForExit: Deferred.succeed(watching, undefined).pipe(Effect.andThen(host.waitForExit)), + })), + ), + reports, + requested, + disabled, + watching, + requests, + }; +}); + it.effect.each([false, true])( "uses the supplied desktop receiver for correlated replies (synchronous: %s) and acknowledged stop", (synchronous) => Effect.gen(function* () { - const reports = yield* PubSub.unbounded(); - const requested = yield* Deferred.make(); - const requests: boolean[] = []; - const factory = yield* CuaDriver.makeDesktopHostFactory().pipe( - Effect.provide( - DesktopTelemetryReceiver.layerTest({ - cuaReports: Stream.fromPubSub(reports), - requestCuaDriver: (requestId, enabled) => - Effect.gen(function* () { - requests.push(enabled); - if (enabled) { - yield* Deferred.succeed(requested, requestId); - if (synchronous) - yield* PubSub.publish(reports, { - version: 1, - type: "cuaDriverReport", - requestId, - status: "ready", - mcp, - }); - } else - yield* PubSub.publish(reports, { - version: 1, - type: "cuaDriverReport", - requestId, - status: "stopped", - }); - }), - }), - ), - ); + const { factory, reports, requested, requests } = yield* desktopFixture(synchronous); yield* Effect.gen(function* () { const { service } = yield* makeService(factory); const acquiring = yield* service.acquire.pipe(Effect.forkChild); @@ -353,6 +387,40 @@ it.effect.each([false, true])( }).pipe(Effect.provide([ServerSettings.layerTest(), NodeCrypto.layer]), Effect.scoped), ); +it.effect("preserves the desktop driver through degraded telemetry until the receiver stops", () => + Effect.gen(function* () { + const streamQueue = + yield* Queue.unbounded< + Stream.Stream + >(); + const processed = yield* Deferred.make(); + const f = yield* desktopFixture(true, Stream.fromQueue(streamQueue).pipe(Stream.flatten)); + const { service } = yield* makeService(f.factory); + const descriptor = yield* service.acquire; + expect(descriptor).toEqual(Option.some(mcp)); + // Acquisition resolves before the exit watcher starts; wait so a false exit cannot go unnoticed. + yield* Deferred.await(f.watching); + yield* Queue.offer( + streamQueue, + Stream.make({ ...healthy, status: "degraded" as const }).pipe( + // The next stream runs only after the health listener consumes the degraded event. + Stream.concat(Stream.fromEffect(Deferred.succeed(processed, undefined)).pipe(Stream.drain)), + ), + ); + yield* Effect.raceFirst(Deferred.await(processed), Deferred.await(f.disabled)); + expect(f.requests).toEqual([true]); + expect(yield* service.acquire).toEqual(descriptor); + expect(f.requests).toEqual([true]); + expect(yield* service.lastFailure).toEqual(Option.none()); + yield* Queue.offer(streamQueue, Stream.make({ ...healthy, status: "stopped" as const })); + yield* Deferred.await(f.disabled); + expect(f.requests).toEqual([true, false]); + expect(yield* service.lastFailure).toEqual( + Option.some(new CuaDriver.CuaDriverHostError({ operation: "exit" })), + ); + }).pipe(Effect.provide([ServerSettings.layerTest(), NodeCrypto.layer]), Effect.scoped), +); + it.effect("uses an explicit driver path without consulting the installer", () => Effect.gen(function* () { let binaryPath = ""; @@ -376,6 +444,7 @@ it.effect("uses an explicit driver path without consulting the installer", () => }, }), ).pipe( + Effect.provide(NodeServices.layer), Effect.provideService(HostProcess.Environment, { T3CODE_CUA_DRIVER_PATH: "/explicit/driver", }), diff --git a/apps/server/src/cua/CuaDriver.ts b/apps/server/src/cua/CuaDriver.ts index 27ff61b543005..5a2895531fe5c 100644 --- a/apps/server/src/cua/CuaDriver.ts +++ b/apps/server/src/cua/CuaDriver.ts @@ -1,3 +1,5 @@ +import * as NodeModule from "node:module"; + import { CuaDriverMcpConfiguration } from "@t3tools/contracts"; import type { EmbeddedCuaDriverHost, EmbeddedDriverConnection } from "@trycua/cua-driver/embedded"; import * as HostProcess from "@t3tools/shared/HostProcess"; @@ -8,8 +10,10 @@ import * as Deferred from "effect/Deferred"; import * as Effect from "effect/Effect"; import * as Exit from "effect/Exit"; import * as Fiber from "effect/Fiber"; +import * as FileSystem from "effect/FileSystem"; import * as Layer from "effect/Layer"; import * as Option from "effect/Option"; +import * as Path from "effect/Path"; import * as Schema from "effect/Schema"; import * as Scope from "effect/Scope"; import * as Stream from "effect/Stream"; @@ -204,9 +208,15 @@ type NativeModule = { ) => Pick; }; +// SEA can load file-backed SDKs through require. This SDK only exports import +// conditions, so resolve its package directory and require the entry file directly. +const requireForCuaDriver = NodeModule.createRequire(import.meta.url); + export const makeStandaloneHostFactory = Effect.fn("CuaDriver.makeStandaloneHostFactory")( function* (executable: Effect.Effect, module?: Effect.Effect) { const environment = yield* HostProcess.Environment; + const fs = yield* FileSystem.FileSystem; + const path = yield* Path.Path; let owned = false; return Effect.gen(function* () { if (owned) return yield* new CuaDriverHostError({ operation: "busy" }); @@ -216,12 +226,30 @@ export const makeStandaloneHostFactory = Effect.fn("CuaDriver.makeStandaloneHost Effect.mapError((cause) => new CuaDriverHostError({ operation: "start", cause })), )); const sdk = yield* module === undefined - ? Effect.tryPromise({ - try: () => import("@trycua/cua-driver/embedded"), - catch: (cause) => new CuaDriverStartError({ binaryPath, cause }), - }) + ? Effect.gen(function* () { + for (const lookup of requireForCuaDriver.resolve.paths("@trycua/cua-driver") ?? []) { + const directory = path.join(lookup, "@trycua", "cua-driver"); + if (yield* fs.exists(path.join(directory, "package.json"))) { + return yield* Effect.try({ + try: () => + requireForCuaDriver( + path.join(directory, "dist", "embedded.js"), + ) as NativeModule, + catch: (cause) => new CuaDriverStartError({ binaryPath, cause }), + }); + } + } + return yield* new CuaDriverStartError({ + binaryPath, + cause: "@trycua/cua-driver is not installed beside this server.", + }); + }).pipe( + Effect.mapError((cause) => + isStartError(cause) ? cause : new CuaDriverStartError({ binaryPath, cause }), + ), + ) : module.pipe(Effect.mapError((cause) => new CuaDriverStartError({ binaryPath, cause }))); - // Acquisition can suspend during installation/import, so check ownership again. + // Acquisition can suspend during installation/loading, so check ownership again. if (owned) return yield* new CuaDriverHostError({ operation: "busy" }); const native = yield* Effect.try({ try: () => new sdk.EmbeddedCuaDriverHost(binaryPath, "com.t3tools.t3code.server"), @@ -331,9 +359,7 @@ export const makeDesktopHostFactory = Effect.fn("CuaDriver.makeDesktopHostFactor return yield* unavailable; yield* health.changes.pipe( Stream.runForEach((value) => - value.status === "stopped" || - value.status === "unavailable" || - value.status === "degraded" + value.status === "stopped" || value.status === "unavailable" ? Deferred.fail(ready, unavailable).pipe( Effect.andThen(Deferred.succeed(stopped, undefined)), ) diff --git a/apps/server/src/resourceTelemetry/DesktopTelemetryReceiver.ts b/apps/server/src/resourceTelemetry/DesktopTelemetryReceiver.ts index d94c7523e5b50..51637d83eb5d5 100644 --- a/apps/server/src/resourceTelemetry/DesktopTelemetryReceiver.ts +++ b/apps/server/src/resourceTelemetry/DesktopTelemetryReceiver.ts @@ -688,7 +688,7 @@ export const layerTest = ( const health = overrides.health ?? Effect.succeed({ - status: overrides.cuaReports === undefined ? ("unavailable" as const) : ("healthy" as const), + status: "unavailable" as const, lastSampleAt: Option.none(), lastError: Option.some("Desktop telemetry test implementation is unavailable."), }); From fe9e8919741102da8be63b5f5d83753e240c7523 Mon Sep 17 00:00:00 2001 From: Julius Marminge Date: Mon, 19 Oct 2026 12:11:44 +0000 Subject: [PATCH 3/6] fix(server): declare Cua SDK with lifecycle consumer --- apps/server/package.json | 1 + pnpm-lock.yaml | 158 +++++++++++++++++++++++++++++++++++++++ 2 files changed, 159 insertions(+) diff --git a/apps/server/package.json b/apps/server/package.json index b474b7012d6a8..abb152c5c4008 100644 --- a/apps/server/package.json +++ b/apps/server/package.json @@ -52,6 +52,7 @@ "@t3tools/source-control-gitcafe": "workspace:*", "@t3tools/source-control-github": "workspace:*", "@t3tools/source-control-gitlab": "workspace:*", + "@trycua/cua-driver": "catalog:", "effect": "catalog:", "jose": "catalog:", "node-pty": "^1.2.0-beta.15", diff --git a/pnpm-lock.yaml b/pnpm-lock.yaml index 01e7f2be6dc46..f2b0151efa84e 100644 --- a/pnpm-lock.yaml +++ b/pnpm-lock.yaml @@ -30,6 +30,9 @@ catalogs: '@pierre/diffs': specifier: 1.5.2 version: 1.5.2 + '@trycua/cua-driver': + specifier: 0.34.1 + version: 0.34.1 '@types/plist': specifier: 3.0.5 version: 3.0.5 @@ -718,6 +721,9 @@ importers: '@t3tools/source-control-gitlab': specifier: workspace:* version: link:../../packages/source-control-gitlab + '@trycua/cua-driver': + specifier: 'catalog:' + version: 0.34.1 effect: specifier: 4.0.5 version: 4.0.5 @@ -6487,6 +6493,41 @@ packages: '@tiptap/starter-kit@3.31.3': resolution: {integrity: sha512-WKof9RewdmGHvWJ1wn0/HVNG2mV+HOgVRyJkKekuM9fgr6BZAAH/xZsWE1eon+94JnQ+KtK2ThydXQM/qc6b2A==} + '@trycua/cua-driver-darwin-arm64@0.34.1': + resolution: {integrity: sha512-GSTwGvNVkg7uMrdvaxwqD9m6cbknDpceaJAdL4bnVTPHjSIk1Jw5Sf/3DpYT1NSqzLgz1ipia306l0LxlsDpmQ==} + cpu: [arm64] + os: [darwin] + + '@trycua/cua-driver-darwin-x64@0.34.1': + resolution: {integrity: sha512-9LNIqARZrdptxiAu3xRO45MPxUuvGtA7Na2yAaibI92p0kps9bRYGV+Rcm+yIxuupdg07Jwyo8Up+FO4rxrQNw==} + cpu: [x64] + os: [darwin] + + '@trycua/cua-driver-linux-arm64-gnu@0.34.1': + resolution: {integrity: sha512-0UDbpOSyGCfq4w5kgKLzKOD7Ms2K5likULF4FY+T2nvmeL3Sf+LyhXEhs8TNuZ8fwgW5TauMq1kYb6LK5ofFGQ==} + cpu: [arm64] + os: [linux] + libc: [glibc] + + '@trycua/cua-driver-linux-x64-gnu@0.34.1': + resolution: {integrity: sha512-RyrYm28rmi8cXzlrdaR5Qi8WQ+jRaZ3Qo8govoOyq5vcjqUefN3G5fPRjxl4MpqqanXs7pwUYVF+d4Gwjq7HoQ==} + cpu: [x64] + os: [linux] + libc: [glibc] + + '@trycua/cua-driver-win32-arm64-msvc@0.34.1': + resolution: {integrity: sha512-jmqDxZTL1+mAWMAnVKfZF+J5bEZok3SjwUOdA4o8hzD74mPLPJO8RUXsvPp4SYtMnnUO6JsDZvz4xrW5m99d6w==} + cpu: [arm64] + os: [win32] + + '@trycua/cua-driver-win32-x64-msvc@0.34.1': + resolution: {integrity: sha512-DBjd+cb8x+HM8xvTvhq4//5HxZm2+ZNS5ZENy969KZppwAN9jzdzXEy0qboWbutGGzFfFv1TglPBLGxRYA3UGA==} + cpu: [x64] + os: [win32] + + '@trycua/cua-driver@0.34.1': + resolution: {integrity: sha512-6r/s28tw2mf9PBawwzfHSSwdbx9kZ+rqy6Hpv4oTRsz2CwG/vl5oWLlNQu/YQy9zuZzCdwXGidmZR0tla2k+4w==} + '@tsc-rs/darwin-arm64@0.7.0': resolution: {integrity: sha512-3execGkNh7DXNZr03dhQP2wqmsjB1Q4OJnTMgKIibjeQMsjNK8HTalJZPIG6PkiXeusXUweHP3+4yJjbi5auNw==} engines: {node: '>=16.20.0'} @@ -6915,6 +6956,56 @@ packages: cpu: [x64] os: [win32] + '@ubjs/core@0.31.0-3': + resolution: {integrity: sha512-39XrJgUZ2VVb561sSnkXPhczNoeBsNiSRArecsV0JE7CJq69ajFkcn9/tBAUS2NpgHkLIDU+z6Ks2+1wXnboxg==} + + '@ubjs/node-darwin-arm64@0.31.0-3': + resolution: {integrity: sha512-GGQVPLkVo4Gc8qVLW4IGvS8bjl8eHXyeP4a97ntGmsAdXwE5gsS29o8xUEFROjhwHKD+9sgVeblCSWVG3CpHsw==} + cpu: [arm64] + os: [darwin] + + '@ubjs/node-darwin-x64@0.31.0-3': + resolution: {integrity: sha512-2sc47u4XFYOsbmP5EW+Gx8m/yGrYnfFDFQm6+kz7goSWTNg84eEiz3COs9HKJDVuNJ5Khv5XipTO8CFadMLXCw==} + cpu: [x64] + os: [darwin] + + '@ubjs/node-linux-arm64-gnu@0.31.0-3': + resolution: {integrity: sha512-YStVXhYz/5jvlWf/p4fhiVT72unYAbGugifFC9QmO/+hnroQDAQ5t8SARbsc15G4olMcamdIB+GETiUB7gmaYg==} + cpu: [arm64] + os: [linux] + libc: [glibc] + + '@ubjs/node-linux-arm64-musl@0.31.0-3': + resolution: {integrity: sha512-Izp4nvfy/LmibzFowAztkoDOksCR2fb2zl6fh1ojR1HEsg0rAGruxtI8d3fn8DI0lBXwqmn6SF///oD4mFNJPQ==} + cpu: [arm64] + os: [linux] + libc: [musl] + + '@ubjs/node-linux-x64-gnu@0.31.0-3': + resolution: {integrity: sha512-Xdm21blyg5U/kW6s7OMvgrr8coGTkUlt26DVR9x8gKISif+E3YwdEskbFScWqystAiJnfjw7xHEc8UMu0Qlz7Q==} + cpu: [x64] + os: [linux] + libc: [glibc] + + '@ubjs/node-linux-x64-musl@0.31.0-3': + resolution: {integrity: sha512-fFQ9BWS6i2LUH9SJgD9oEiKXXo/say59vHy7usFe1t7C2xvwvP34f5SuxXmWP9R8fHyA0aC4kIZ+TTwyHSv1Kw==} + cpu: [x64] + os: [linux] + libc: [musl] + + '@ubjs/node-win32-arm64-msvc@0.31.0-3': + resolution: {integrity: sha512-ID6rSz1NmPsWTNBBNAw4OnJ5Dj8pcbtNJtdPB3OxcGigLBd/e0x7buhSI7os6Mo5iYtEdCynBRQHFJwub5XSPg==} + cpu: [arm64] + os: [win32] + + '@ubjs/node-win32-x64-msvc@0.31.0-3': + resolution: {integrity: sha512-wevs+Y+szwcCUT8IJFbB4/1nfxyRv/51l8oG7FGUPWD1xLPyuHfJBG27C+PDeC7KRzes/2n6aLo4DGZbr/LYTw==} + cpu: [x64] + os: [win32] + + '@ubjs/node@0.31.0-3': + resolution: {integrity: sha512-qNMpi2LICNwxGXZyRF8fSDBSpbezyZbEsydrbiMPJOmtOWr4tmZIEl7jkWGHVGShoBvHfFo4eHp5B4UVP928Cg==} + '@ungap/structured-clone@1.3.1': resolution: {integrity: sha512-mUFwbeTqrVgDQxFveS+df2yfap6iuP20NAKAsBt5jDEoOTDew+zwLAOilHCeQJOVSvmgCX4ogqIrA0mnyr08yQ==} @@ -17155,6 +17246,36 @@ snapshots: '@tiptap/extensions': 3.31.3(@tiptap/core@3.31.3(@tiptap/pm@3.31.3))(@tiptap/pm@3.31.3) '@tiptap/pm': 3.31.3 + '@trycua/cua-driver-darwin-arm64@0.34.1': + optional: true + + '@trycua/cua-driver-darwin-x64@0.34.1': + optional: true + + '@trycua/cua-driver-linux-arm64-gnu@0.34.1': + optional: true + + '@trycua/cua-driver-linux-x64-gnu@0.34.1': + optional: true + + '@trycua/cua-driver-win32-arm64-msvc@0.34.1': + optional: true + + '@trycua/cua-driver-win32-x64-msvc@0.34.1': + optional: true + + '@trycua/cua-driver@0.34.1': + dependencies: + '@ubjs/core': 0.31.0-3 + '@ubjs/node': 0.31.0-3 + optionalDependencies: + '@trycua/cua-driver-darwin-arm64': 0.34.1 + '@trycua/cua-driver-darwin-x64': 0.34.1 + '@trycua/cua-driver-linux-arm64-gnu': 0.34.1 + '@trycua/cua-driver-linux-x64-gnu': 0.34.1 + '@trycua/cua-driver-win32-arm64-msvc': 0.34.1 + '@trycua/cua-driver-win32-x64-msvc': 0.34.1 + '@tsc-rs/darwin-arm64@0.7.0': optional: true @@ -17561,6 +17682,43 @@ snapshots: '@typescript/typescript-win32-x64@7.0.2': optional: true + '@ubjs/core@0.31.0-3': {} + + '@ubjs/node-darwin-arm64@0.31.0-3': + optional: true + + '@ubjs/node-darwin-x64@0.31.0-3': + optional: true + + '@ubjs/node-linux-arm64-gnu@0.31.0-3': + optional: true + + '@ubjs/node-linux-arm64-musl@0.31.0-3': + optional: true + + '@ubjs/node-linux-x64-gnu@0.31.0-3': + optional: true + + '@ubjs/node-linux-x64-musl@0.31.0-3': + optional: true + + '@ubjs/node-win32-arm64-msvc@0.31.0-3': + optional: true + + '@ubjs/node-win32-x64-msvc@0.31.0-3': + optional: true + + '@ubjs/node@0.31.0-3': + optionalDependencies: + '@ubjs/node-darwin-arm64': 0.31.0-3 + '@ubjs/node-darwin-x64': 0.31.0-3 + '@ubjs/node-linux-arm64-gnu': 0.31.0-3 + '@ubjs/node-linux-arm64-musl': 0.31.0-3 + '@ubjs/node-linux-x64-gnu': 0.31.0-3 + '@ubjs/node-linux-x64-musl': 0.31.0-3 + '@ubjs/node-win32-arm64-msvc': 0.31.0-3 + '@ubjs/node-win32-x64-msvc': 0.31.0-3 + '@ungap/structured-clone@1.3.1': {} '@upsetjs/venn.js@2.0.0': From fe06f033e8ec8942c268601cf360bd03c35ed941 Mon Sep 17 00:00:00 2001 From: Julius Marminge Date: Mon, 19 Oct 2026 12:20:21 +0000 Subject: [PATCH 4/6] fix(server): preserve Cua Driver exit failures --- .../src/telemetry/DesktopTelemetryPublisher.ts | 2 ++ apps/server/src/cua/CuaDriver.test.ts | 9 +++++++-- apps/server/src/cua/CuaDriver.ts | 13 ++++++------- 3 files changed, 15 insertions(+), 9 deletions(-) diff --git a/apps/desktop/src/telemetry/DesktopTelemetryPublisher.ts b/apps/desktop/src/telemetry/DesktopTelemetryPublisher.ts index 1155a2b15b0ce..964294f28efdc 100644 --- a/apps/desktop/src/telemetry/DesktopTelemetryPublisher.ts +++ b/apps/desktop/src/telemetry/DesktopTelemetryPublisher.ts @@ -320,6 +320,8 @@ export const make = Effect.fn("desktop.telemetryPublisher.make")(function* () { message, ) => { switch (message.type) { + case "cuaDriverRequest": + return Effect.void; case "setDiagnosticsDemand": return Ref.modify(diagnosticsDemandSources, (sources) => { const previous = sources.size > 0; diff --git a/apps/server/src/cua/CuaDriver.test.ts b/apps/server/src/cua/CuaDriver.test.ts index 4c5fd6c5708e7..d35b7fbd011c8 100644 --- a/apps/server/src/cua/CuaDriver.test.ts +++ b/apps/server/src/cua/CuaDriver.test.ts @@ -31,7 +31,7 @@ const makeHost = Effect.fnUntraced(function* () { const stopped = yield* Deferred.make(); const watching = yield* Deferred.make(); const ready = yield* Deferred.make(); - const exited = yield* Deferred.make(); + const exited = yield* Deferred.make(); let stops = 0; const host: CuaDriver.CuaDriverHost = { start: Deferred.succeed(started, undefined).pipe(Effect.andThen(Deferred.await(ready))), @@ -129,8 +129,13 @@ it.layer(ServerSettings.layerTest())("CuaDriver lifecycle", (it) => { expect(yield* service.acquire).toEqual(Option.some(mcp)); expect(yield* service.lastFailure).toEqual(Option.none()); yield* Deferred.await(crashed.watching); - yield* Deferred.succeed(crashed.exited, undefined); + const waitError = new CuaDriver.CuaDriverHostError({ + operation: "exit", + cause: new Error("native waitForExit failed"), + }); + yield* Deferred.fail(crashed.exited, waitError); yield* Deferred.await(crashed.stopped); + expect(Option.getOrThrow(yield* service.lastFailure)).toBe(waitError); const next = { ...mcp, args: ["mcp", "--generation=next"] }; yield* Deferred.succeed(replacement.ready, next); expect(yield* service.acquire).toEqual(Option.some(next)); diff --git a/apps/server/src/cua/CuaDriver.ts b/apps/server/src/cua/CuaDriver.ts index 5a2895531fe5c..f9a65b861d68f 100644 --- a/apps/server/src/cua/CuaDriver.ts +++ b/apps/server/src/cua/CuaDriver.ts @@ -132,15 +132,14 @@ export const make = Effect.fn("CuaDriver.make")(function* ( failure = Option.none(); yield* Deferred.succeed(entry.result, Option.some(mcp)); entry.watcher = yield* entry.host.waitForExit.pipe( - Effect.catch((error) => - Effect.sync(() => { - if (active === entry) failure = Option.some(error); - }), - ), - Effect.andThen( + Effect.match({ + onFailure: (error) => error, + onSuccess: () => new CuaDriverHostError({ operation: "exit" }), + }), + Effect.flatMap((error) => Effect.gen(function* () { if (active !== entry) return; - failure = Option.some(new CuaDriverHostError({ operation: "exit" })); + failure = Option.some(error); // The watcher must not interrupt itself while retiring its host. entry.watcher = undefined; yield* revoke(entry); From 817274756ee3ae5c34e755240c785fe0f2de13f8 Mon Sep 17 00:00:00 2001 From: Julius Marminge Date: Mon, 19 Oct 2026 12:38:59 +0000 Subject: [PATCH 5/6] fix(server): prevent enabling retired desktop Cua hosts --- apps/server/src/cua/CuaDriver.test.ts | 34 ++++++++++++++++++++++++++- apps/server/src/cua/CuaDriver.ts | 3 +++ 2 files changed, 36 insertions(+), 1 deletion(-) diff --git a/apps/server/src/cua/CuaDriver.test.ts b/apps/server/src/cua/CuaDriver.test.ts index d35b7fbd011c8..7b2dad2235481 100644 --- a/apps/server/src/cua/CuaDriver.test.ts +++ b/apps/server/src/cua/CuaDriver.test.ts @@ -309,6 +309,7 @@ const healthy: DesktopTelemetryReceiver.DesktopTelemetryReceiverHealth = { const desktopFixture = Effect.fnUntraced(function* ( synchronous = false, healthChanges: Stream.Stream = Stream.empty, + beforeHealth = Effect.void, ) { const reports = yield* PubSub.unbounded(); const requested = yield* Deferred.make(); @@ -319,7 +320,7 @@ const desktopFixture = Effect.fnUntraced(function* ( Effect.provide( DesktopTelemetryReceiver.layerTest({ health: Effect.succeed(healthy), - subscribeHealth: Effect.succeed({ latest: healthy, changes: healthChanges }), + subscribeHealth: beforeHealth.pipe(Effect.as({ latest: healthy, changes: healthChanges })), cuaReports: Stream.fromPubSub(reports), requestCuaDriver: (requestId, enabled) => Effect.gen(function* () { @@ -392,6 +393,37 @@ it.effect.each([false, true])( }).pipe(Effect.provide([ServerSettings.layerTest(), NodeCrypto.layer]), Effect.scoped), ); +it.effect("does not enable a desktop host retired while subscribing to health", () => + Effect.gen(function* () { + const subscribing = yield* Deferred.make(); + const releaseHealth = yield* Deferred.make(); + const startFinished = + yield* Deferred.make>(); + const f = yield* desktopFixture( + false, + Stream.empty, + Deferred.succeed(subscribing, undefined).pipe(Effect.andThen(Deferred.await(releaseHealth))), + ); + const scope = yield* Scope.make(); + const { service } = yield* makeService( + f.factory.pipe( + Effect.map((host) => ({ + ...host, + start: host.start.pipe(Effect.onExit((exit) => Deferred.succeed(startFinished, exit))), + })), + ), + ).pipe(Effect.provideService(Scope.Scope, scope)); + const acquiring = yield* service.acquire.pipe(Effect.forkChild); + yield* Deferred.await(subscribing); + yield* Scope.close(scope, Exit.void); + expect(yield* Fiber.join(acquiring)).toEqual(Option.none()); + yield* Deferred.succeed(releaseHealth, undefined); + const result = yield* Deferred.await(startFinished); + expect(result).toEqual(Exit.fail(new CuaDriver.CuaDriverHostError({ operation: "desktop" }))); + expect(f.requests).toEqual([]); + }).pipe(Effect.provide([ServerSettings.layerTest(), NodeCrypto.layer]), Effect.scoped), +); + it.effect("preserves the desktop driver through degraded telemetry until the receiver stops", () => Effect.gen(function* () { const streamQueue = diff --git a/apps/server/src/cua/CuaDriver.ts b/apps/server/src/cua/CuaDriver.ts index f9a65b861d68f..3a52b50149d2b 100644 --- a/apps/server/src/cua/CuaDriver.ts +++ b/apps/server/src/cua/CuaDriver.ts @@ -330,6 +330,7 @@ export const makeDesktopHostFactory = Effect.fn("CuaDriver.makeDesktopHostFactor const ready = yield* Deferred.make(); const stopped = yield* Deferred.make(); let requested = false; + let retired = false; const unavailable = new CuaDriverHostError({ operation: "desktop" }); const start = Effect.gen(function* () { // Start the subscription before writing: the desktop may reply synchronously. @@ -366,6 +367,7 @@ export const makeDesktopHostFactory = Effect.fn("CuaDriver.makeDesktopHostFactor ), Effect.forkIn(scope, { startImmediately: true }), ); + if (retired) return yield* unavailable; requested = true; yield* receiver .requestCuaDriver(requestId, true) @@ -373,6 +375,7 @@ export const makeDesktopHostFactory = Effect.fn("CuaDriver.makeDesktopHostFactor return yield* Deferred.await(ready); }); const stop = Effect.gen(function* () { + retired = true; if (!requested) return; yield* receiver .requestCuaDriver(requestId, false) From dc307c8a75312243d2519c498c4346b1af505099 Mon Sep 17 00:00:00 2001 From: Julius Marminge Date: Mon, 19 Oct 2026 13:10:38 +0000 Subject: [PATCH 6/6] fix(server): limit Cua startup after preparation --- apps/server/src/cua/CuaDriver.test.ts | 34 ++++++++++++++++++++++++--- apps/server/src/cua/CuaDriver.ts | 15 ++++++------ 2 files changed, 38 insertions(+), 11 deletions(-) diff --git a/apps/server/src/cua/CuaDriver.test.ts b/apps/server/src/cua/CuaDriver.test.ts index 7b2dad2235481..445fd7ceb81ce 100644 --- a/apps/server/src/cua/CuaDriver.test.ts +++ b/apps/server/src/cua/CuaDriver.test.ts @@ -81,22 +81,28 @@ it.layer(ServerSettings.layerTest())("CuaDriver lifecycle", (it) => { }).pipe(Effect.scoped), ); - it.effect("shares a lazy start, survives one caller cancellation, and owns cleanup", () => + it.effect("shares slow preparation, survives one caller cancellation, and owns cleanup", () => Effect.gen(function* () { const h = yield* makeHost(); + const preparing = yield* Deferred.make(); + const prepared = yield* Deferred.make(); let creates = 0; yield* Effect.gen(function* () { const { service } = yield* makeService( - Effect.sync(() => { + Effect.gen(function* () { creates++; + yield* Deferred.succeed(preparing, undefined); + yield* Deferred.await(prepared); return h.host; }), ); expect(creates).toBe(0); const first = yield* service.acquire.pipe(Effect.forkChild); - yield* Deferred.await(h.started); + yield* Deferred.await(preparing); const second = yield* service.acquire.pipe(Effect.forkChild({ startImmediately: true })); yield* Fiber.interrupt(first); + yield* TestClock.adjust("31 seconds"); + yield* Deferred.succeed(prepared, undefined); yield* Deferred.succeed(h.ready, mcp); expect(yield* Fiber.join(second)).toEqual(Option.some(mcp)); expect(yield* service.acquire).toEqual(Option.some(mcp)); @@ -262,6 +268,28 @@ const nativeFixture = Effect.fnUntraced(function* () { }); it.layer(ServerSettings.layerTest())("CuaDriver native ownership", (it) => { + it.effect("times out native startup after 30 seconds and cleans up when it settles", () => + Effect.gen(function* () { + const f = yield* nativeFixture(); + const { service } = yield* makeService(f.factory); + const acquiring = yield* service.acquire.pipe(Effect.forkChild); + yield* Deferred.await(f.started); + yield* TestClock.adjust("29 seconds"); + expect(yield* service.lastFailure).toEqual(Option.none()); + yield* TestClock.adjust("1 second"); + expect(yield* service.lastFailure).toEqual( + Option.some(new CuaDriver.CuaDriverHostError({ operation: "timeout" })), + ); + expect(f.destroys()).toBe(0); + f.start.resolve(f.connection); + yield* Deferred.await(f.stopping); + f.stop.resolve(); + yield* Deferred.await(f.destroyed); + expect(yield* Fiber.join(acquiring)).toEqual(Option.none()); + expect(f.destroys()).toBe(1); + }).pipe(Effect.scoped), + ); + it.effect.each(["start", "stop"] as const)( "bounds shutdown with native %s pending without destroying or reusing its owner", (pending) => diff --git a/apps/server/src/cua/CuaDriver.ts b/apps/server/src/cua/CuaDriver.ts index 3a52b50149d2b..af13fc54f17e5 100644 --- a/apps/server/src/cua/CuaDriver.ts +++ b/apps/server/src/cua/CuaDriver.ts @@ -127,7 +127,12 @@ export const make = Effect.fn("CuaDriver.make")(function* ( yield* stop(entry); return; } - const mcp = yield* entry.host.start; + const mcp = yield* entry.host.start.pipe( + Effect.timeoutOrElse({ + duration: "30 seconds", + orElse: () => Effect.fail(new CuaDriverHostError({ operation: "timeout" })), + }), + ); if (entry.revoked) return; failure = Option.none(); yield* Deferred.succeed(entry.result, Option.some(mcp)); @@ -189,13 +194,7 @@ export const make = Effect.fn("CuaDriver.make")(function* ( active = entry; yield* start(entry).pipe(Effect.forkDetach); } - const result = yield* Deferred.await(entry.result).pipe(Effect.timeoutOption("30 seconds")); - if (Option.isSome(result)) return result.value; - if (active === entry) { - failure = Option.some(new CuaDriverHostError({ operation: "timeout" })); - yield* revoke(entry); - } - return Option.none(); + return yield* Deferred.await(entry.result); }); return CuaDriver.of({ acquire, enabled, lastFailure: Effect.sync(() => failure) }); });