From d12b80189668c60cae1f2ac9f1d138dc0b859e8c Mon Sep 17 00:00:00 2001 From: Francesco Bonacci Date: Wed, 9 Sep 2026 00:44:54 -0500 Subject: [PATCH 1/2] feat(desktop): add opt-in Cua Driver computer use --- apps/desktop/package.json | 1 + apps/desktop/src/app/DesktopApp.ts | 3 + .../src/backend/DesktopBackendManager.test.ts | 3 + .../src/backend/DesktopBackendManager.ts | 2 +- .../src/backend/DesktopBackendPool.test.ts | 3 + apps/desktop/src/cua/DesktopCuaDriver.test.ts | 288 +++++++++++++ apps/desktop/src/cua/DesktopCuaDriver.ts | 288 +++++++++++++ apps/desktop/src/main.ts | 2 + .../DesktopTelemetryPublisher.test.ts | 66 ++- .../telemetry/DesktopTelemetryPublisher.ts | 37 ++ .../src/updates/DesktopRemoteUpdates.test.ts | 3 + apps/server/package.json | 2 + apps/server/src/cua/CuaDriver.test.ts | 405 ++++++++++++++++++ apps/server/src/cua/CuaDriver.ts | 345 +++++++++++++++ .../src/cua/codexCuaConfiguration.test.ts | 104 +++++ apps/server/src/cua/codexCuaConfiguration.ts | 70 +++ apps/server/src/cua/resolveCodexCua.test.ts | 114 +++++ apps/server/src/cua/resolveCodexCua.ts | 54 +++ .../src/provider/Layers/CodexAdapter.test.ts | 79 ++++ .../src/provider/Layers/CodexAdapter.ts | 22 +- .../Layers/CodexSessionRuntime.test.ts | 15 + .../provider/Layers/CodexSessionRuntime.ts | 9 +- .../DesktopTelemetryReceiver.ts | 19 + apps/server/src/server.ts | 7 + .../settings/ProjectDefaultsSettings.tsx | 44 ++ .../components/settings/SettingsPanels.tsx | 3 + .../src/components/settings/settingsSearch.ts | 8 + docs/user/providers-codex.md | 21 + packages/contracts/src/cua.ts | 37 ++ packages/contracts/src/index.ts | 1 + packages/contracts/src/resourceTelemetry.ts | 3 + packages/contracts/src/settings.test.ts | 28 ++ packages/contracts/src/settings.ts | 3 + pnpm-lock.yaml | 163 ++++++- scripts/build-desktop-artifact.test.ts | 144 +++++++ scripts/build-desktop-artifact.ts | 238 ++++++++++ scripts/lib/cli-external-packages.test.ts | 5 +- scripts/lib/cli-external-packages.ts | 2 + 38 files changed, 2630 insertions(+), 11 deletions(-) create mode 100644 apps/desktop/src/cua/DesktopCuaDriver.test.ts create mode 100644 apps/desktop/src/cua/DesktopCuaDriver.ts create mode 100644 apps/server/src/cua/CuaDriver.test.ts create mode 100644 apps/server/src/cua/CuaDriver.ts create mode 100644 apps/server/src/cua/codexCuaConfiguration.test.ts create mode 100644 apps/server/src/cua/codexCuaConfiguration.ts create mode 100644 apps/server/src/cua/resolveCodexCua.test.ts create mode 100644 apps/server/src/cua/resolveCodexCua.ts create mode 100644 packages/contracts/src/cua.ts diff --git a/apps/desktop/package.json b/apps/desktop/package.json index ab5c82bec329..837831054725 100644 --- a/apps/desktop/package.json +++ b/apps/desktop/package.json @@ -22,6 +22,7 @@ "@t3tools/shared": "workspace:*", "@t3tools/ssh": "workspace:*", "@t3tools/tailscale": "workspace:*", + "@trycua/cua-driver": "0.24.0", "dbus-next": "0.10.2", "effect": "catalog:", "electron": "44.1.0", diff --git a/apps/desktop/src/app/DesktopApp.ts b/apps/desktop/src/app/DesktopApp.ts index e84e6d5c730d..c4d1d5419df0 100644 --- a/apps/desktop/src/app/DesktopApp.ts +++ b/apps/desktop/src/app/DesktopApp.ts @@ -27,6 +27,7 @@ import * as DesktopServerExposure from "../backend/DesktopServerExposure.ts"; import * as DesktopAppSettings from "../settings/DesktopAppSettings.ts"; import * as DesktopShellEnvironment from "../shell/DesktopShellEnvironment.ts"; import * as DesktopState from "./DesktopState.ts"; +import { DesktopCuaDriver } from "../cua/DesktopCuaDriver.ts"; import * as DesktopRemoteUpdates from "../updates/DesktopRemoteUpdates.ts"; import * as DesktopUpdates from "../updates/DesktopUpdates.ts"; import * as DesktopSnapShot from "../snapShot/DesktopSnapShot.ts"; @@ -243,6 +244,7 @@ const startup = Effect.gen(function* () { const preReadyElectronOptions = yield* DesktopPreReadyPlatform.DesktopPreReadyElectronOptions; const safeStorage = yield* ElectronSafeStorage.ElectronSafeStorage; const updates = yield* DesktopUpdates.DesktopUpdates; + const cuaDriver = yield* DesktopCuaDriver; const environment = yield* DesktopEnvironment.DesktopEnvironment; yield* shellEnvironment.installIntoProcess; @@ -300,6 +302,7 @@ const startup = Effect.gen(function* () { yield* applicationMenu.configure; yield* updates.configure; yield* DesktopRemoteUpdates.listen; + yield* cuaDriver.listen; yield* linuxUrlHandler.register; yield* bootstrap.pipe(Effect.catchCause((cause) => fatalStartupCause("bootstrap", cause))); }).pipe(Effect.withSpan("desktop.startup")); diff --git a/apps/desktop/src/backend/DesktopBackendManager.test.ts b/apps/desktop/src/backend/DesktopBackendManager.test.ts index 53ccf5a756eb..296bc12c5c2c 100644 --- a/apps/desktop/src/backend/DesktopBackendManager.test.ts +++ b/apps/desktop/src/backend/DesktopBackendManager.test.ts @@ -163,6 +163,9 @@ function makeTestInstance(input: MakeInstanceInput) { latest: Effect.succeed(Option.none()), changes: Stream.empty, encoded: input.desktopTelemetryStream ?? Stream.empty, + encodedForSource: () => input.desktopTelemetryStream ?? Stream.empty, + cuaRequests: Stream.empty, + publishCuaReport: () => Effect.void, handleControl: () => Effect.void, handleControlForSource: (_sourceId, message) => (input.desktopTelemetryPublisher?.handleControl ?? (() => Effect.void))(message), diff --git a/apps/desktop/src/backend/DesktopBackendManager.ts b/apps/desktop/src/backend/DesktopBackendManager.ts index 6f1ea139bdca..cc8358d7779d 100644 --- a/apps/desktop/src/backend/DesktopBackendManager.ts +++ b/apps/desktop/src/backend/DesktopBackendManager.ts @@ -906,7 +906,7 @@ export const makeBackendInstance = Effect.fn("makeBackendInstance")(function* ( const program = runBackendProcess({ ...config.value, - desktopTelemetryStream: desktopTelemetryPublisher.encoded, + desktopTelemetryStream: desktopTelemetryPublisher.encodedForSource(spec.id), onDesktopTelemetryControl: (message) => desktopTelemetryPublisher.handleControlForSource(spec.id, message), onStarted: Effect.fn("desktop.backendInstance.onStarted")(function* (pid) { diff --git a/apps/desktop/src/backend/DesktopBackendPool.test.ts b/apps/desktop/src/backend/DesktopBackendPool.test.ts index b9a96d72a77a..97e7b7642600 100644 --- a/apps/desktop/src/backend/DesktopBackendPool.test.ts +++ b/apps/desktop/src/backend/DesktopBackendPool.test.ts @@ -70,6 +70,9 @@ function makePoolLayer( latest: Effect.succeed(Option.none()), changes: Stream.empty, encoded: Stream.empty, + encodedForSource: () => Stream.empty, + cuaRequests: Stream.empty, + publishCuaReport: () => Effect.void, handleControl: () => Effect.void, handleControlForSource: () => Effect.void, removeControlSource: () => Effect.void, diff --git a/apps/desktop/src/cua/DesktopCuaDriver.test.ts b/apps/desktop/src/cua/DesktopCuaDriver.test.ts new file mode 100644 index 000000000000..50336f7ca5e7 --- /dev/null +++ b/apps/desktop/src/cua/DesktopCuaDriver.test.ts @@ -0,0 +1,288 @@ +import * as NodePath from "@effect/platform-node/NodePath"; +import { assert, describe, it } from "@effect/vitest"; +import { + DesktopHostTelemetryMessage, + type DesktopCuaDriverReport, + type DesktopCuaDriverRequest, +} from "@t3tools/contracts"; +import * as Fiber from "effect/Fiber"; +import * as Exit from "effect/Exit"; +import * as Scope from "effect/Scope"; +import * as TestClock from "effect/testing/TestClock"; +import * as Effect from "effect/Effect"; +import * as Layer from "effect/Layer"; +import * as Queue from "effect/Queue"; +import * as Schema from "effect/Schema"; +import * as Stream from "effect/Stream"; + +import * as DesktopConfig from "../app/DesktopConfig.ts"; +import * as DesktopEnvironment from "../app/DesktopEnvironment.ts"; +import { DesktopTelemetryPublisher } from "../telemetry/DesktopTelemetryPublisher.ts"; +import { make, type DesktopCuaDriverDependencies } from "./DesktopCuaDriver.ts"; + +const encodeMessage = Schema.encodeEffect(Schema.fromJsonString(DesktopHostTelemetryMessage)); +const decodeMessage = Schema.decodeEffect(Schema.fromJsonString(DesktopHostTelemetryMessage)); + +const descriptor = { + command: "/native/driver", + args: ["mcp", "--session", "opaque"], + environment: [{ name: "CUA_SESSION", value: "test-value" }], +}; +const environmentLayer = (platform: NodeJS.Platform = "linux") => + DesktopEnvironment.layer({ + dirname: "/app", + homeDirectory: "/home/test", + platform, + processArch: "x64", + appVersion: "1.0.0", + appPath: "/app", + isPackaged: true, + resourcesPath: "/resources", + runningUnderArm64Translation: false, + }).pipe(Layer.provide(Layer.mergeAll(NodePath.layerPosix, DesktopConfig.layerTest({})))); + +const fixture = Effect.fn(function* ( + failStart = false, + denyPermissions = false, + control: { holdStart?: boolean; holdStop?: boolean } = {}, +) { + const requests = yield* Queue.unbounded(); + const reports = yield* Queue.unbounded(); + const exits: Array<() => void> = []; + const starting = yield* Queue.unbounded(); + const stopping = yield* Queue.unbounded(); + const destroyed = yield* Queue.unbounded(); + const startGate = Promise.withResolvers(); + const stopGate = Promise.withResolvers(); + let starts = 0; + let stops = 0; + let destroys = 0; + const dependencies: DesktopCuaDriverDependencies = { + loadElectron: async () => ({ + requestMacOSPermissions: () => ({ accessibility: false, screenRecording: false }), + hasRequiredMacOSPermissions: () => false, + }), + loadEmbedded: async () => ({ + EmbeddedCuaDriverHost: class { + async start() { + starts++; + Queue.offerUnsafe(starting, undefined); + if (control.holdStart) await startGate.promise; + if (failStart) throw new Error("private host details"); + return { + socketPath: "/socket", + pid: 100, + generation: String(starts), + driverVersion: "0.24", + contractVersion: "1", + mcpProtocolVersion: "1", + mcp: descriptor, + }; + } + async stop() { + stops++; + Queue.offerUnsafe(stopping, undefined); + if (control.holdStop) await stopGate.promise; + } + uniffiDestroy() { + destroys++; + Queue.offerUnsafe(destroyed, undefined); + } + waitForExit(generation: string, options?: { signal: AbortSignal }) { + return new Promise<{ generation: string; success: boolean }>((resolve) => { + exits.push(() => resolve({ generation, success: false })); + options?.signal.addEventListener( + "abort", + () => resolve({ generation, success: false }), + { once: true }, + ); + }); + } + }, + }), + }; + const publisher = DesktopTelemetryPublisher.of({ + latest: Effect.succeedNone, + changes: Stream.empty, + encoded: Stream.empty, + encodedForSource: () => Stream.empty, + handleControl: () => Effect.void, + handleControlForSource: () => Effect.void, + removeControlSource: () => Effect.void, + updateRequests: Stream.empty, + updateCommits: Stream.empty, + updateCancellations: Stream.empty, + publishUpdateReport: () => Effect.void, + cuaRequests: Stream.fromQueue(requests), + publishCuaReport: (report) => Queue.offer(reports, report).pipe(Effect.asVoid), + }); + const driver = yield* make(dependencies).pipe( + Effect.provideService(DesktopTelemetryPublisher, publisher), + Effect.provide(environmentLayer(denyPermissions ? "darwin" : "linux")), + ); + yield* driver.listen; + return { + request: (requestId: string, enabled: boolean) => + Queue.offer(requests, { version: 1, type: "cuaDriverRequest", requestId, enabled }), + next: Queue.take(reports), + starting: Queue.take(starting), + stopping: Queue.take(stopping), + destroyed: Queue.take(destroyed), + releaseStart: Effect.sync(() => startGate.resolve()), + releaseStop: Effect.sync(() => stopGate.resolve()), + exits, + counts: () => ({ starts, stops, destroys }), + }; +}); + +describe("DesktopCuaDriver", () => { + it.effect("returns the exact descriptor, reuses the host, and stops it once", () => + Effect.scoped( + Effect.gen(function* () { + const test = yield* fixture(); + yield* test.request("start", true); + const ready = yield* test.next; + const json = yield* encodeMessage(ready); + assert.deepEqual(yield* decodeMessage(json), { + version: 1, + type: "cuaDriverReport", + requestId: "start", + status: "ready", + mcp: descriptor, + }); + yield* test.request("same", true); + assert.equal((yield* test.next).requestId, "same"); + yield* test.request("stop", false); + assert.equal((yield* test.next).status, "stopped"); + yield* test.request("stop-again", false); + yield* test.next; + assert.deepEqual(test.counts(), { starts: 1, stops: 1, destroys: 1 }); + }), + ), + ); + + it.effect("reports a crash against the active request and allows a new host", () => + Effect.scoped( + Effect.gen(function* () { + const test = yield* fixture(); + yield* test.request("first", true); + yield* test.next; + yield* test.request("current", true); + yield* test.next; + yield* Effect.sync(() => test.exits[0]!()); + const report = yield* test.next; + assert.equal(report.status, "unavailable"); + assert.equal(report.requestId, "current"); + yield* test.request("restart", true); + assert.equal((yield* test.next).status, "ready"); + assert.deepEqual(test.counts(), { starts: 2, stops: 1, destroys: 1 }); + }), + ), + ); + + it.effect("cleans up failed starts without exposing native error details", () => + Effect.scoped( + Effect.gen(function* () { + const test = yield* fixture(true); + yield* test.request("failed", true); + const report = yield* test.next; + assert.equal(report.status, "unavailable"); + if (report.status !== "ready") + assert.equal(report.message?.includes("private host details"), false); + assert.deepEqual(test.counts(), { starts: 1, stops: 1, destroys: 1 }); + }), + ), + ); + + it.effect("reports missing permissions without constructing a native host", () => + Effect.scoped( + Effect.gen(function* () { + const test = yield* fixture(false, true); + yield* test.request("permissions", true); + const report = yield* test.next; + assert.equal(report.status, "unavailable"); + if (report.status !== "ready") assert.include(report.message ?? "", "Accessibility"); + assert.deepEqual(test.counts(), { starts: 0, stops: 0, destroys: 0 }); + }), + ), + ); + + it.effect( + "times out a hanging start and cleans up its late result without publishing ready", + () => + Effect.scoped( + Effect.gen(function* () { + const test = yield* fixture(false, false, { holdStart: true }); + yield* test.request("hanging", true); + yield* test.starting; + yield* TestClock.adjust("23 seconds"); + assert.equal((yield* test.next).status, "unavailable"); + yield* test.request("disable", false); + yield* TestClock.adjust("3 seconds"); + assert.equal((yield* test.next).status, "unavailable"); + assert.deepEqual(test.counts(), { starts: 1, stops: 0, destroys: 0 }); + yield* test.releaseStart; + yield* test.destroyed; + yield* test.request("disabled", false); + const report = yield* test.next; + assert.equal(report.requestId, "disabled"); + assert.equal(report.status, "stopped"); + assert.deepEqual(test.counts(), { starts: 1, stops: 1, destroys: 1 }); + }), + ), + ); + + it.effect("bounds a hanging stop and prevents another native host until cleanup settles", () => + Effect.scoped( + Effect.gen(function* () { + const test = yield* fixture(false, false, { holdStop: true }); + yield* test.request("start", true); + yield* test.next; + yield* test.request("stop", false); + yield* test.stopping; + yield* TestClock.adjust("3 seconds"); + assert.equal((yield* test.next).status, "unavailable"); + yield* test.request("too-soon", true); + assert.equal((yield* test.next).status, "unavailable"); + assert.deepEqual(test.counts(), { starts: 1, stops: 1, destroys: 0 }); + yield* test.releaseStop; + yield* test.destroyed; + yield* test.request("restart", true); + assert.equal((yield* test.next).status, "ready"); + assert.deepEqual(test.counts(), { starts: 2, stops: 1, destroys: 1 }); + }), + ), + ); + + it.effect("closes its scope while start hangs and cleans up after late completion", () => + Effect.gen(function* () { + const scope = yield* Scope.make(); + const test = yield* fixture(false, false, { holdStart: true }).pipe( + Effect.provideService(Scope.Scope, scope), + ); + yield* test.request("start", true); + yield* test.starting; + const closing = yield* Scope.close(scope, Exit.void).pipe(Effect.forkChild); + yield* TestClock.adjust("3 seconds"); + yield* Fiber.join(closing); + assert.deepEqual(test.counts(), { starts: 1, stops: 0, destroys: 0 }); + yield* test.releaseStart; + yield* test.destroyed; + assert.deepEqual(test.counts(), { starts: 1, stops: 1, destroys: 1 }); + }), + ); + + it.effect("cleans up the active host when its scope closes", () => + Effect.gen(function* () { + const test = yield* Effect.scoped( + Effect.gen(function* () { + const test = yield* fixture(); + yield* test.request("start", true); + yield* test.next; + return test; + }), + ); + assert.deepEqual(test.counts(), { starts: 1, stops: 1, destroys: 1 }); + }), + ); +}); diff --git a/apps/desktop/src/cua/DesktopCuaDriver.ts b/apps/desktop/src/cua/DesktopCuaDriver.ts new file mode 100644 index 000000000000..6e3ce75e5e95 --- /dev/null +++ b/apps/desktop/src/cua/DesktopCuaDriver.ts @@ -0,0 +1,288 @@ +import { CuaDriverMcpConfiguration, type DesktopCuaDriverRequest } from "@t3tools/contracts"; +import type { EmbeddedCuaDriverHost, EmbeddedDriverConnection } from "@trycua/cua-driver/embedded"; +import * as Cause from "effect/Cause"; +import * as Exit from "effect/Exit"; +import * as Context from "effect/Context"; +import * as Effect from "effect/Effect"; +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 Semaphore from "effect/Semaphore"; +import * as Stream from "effect/Stream"; + +import * as DesktopEnvironment from "../app/DesktopEnvironment.ts"; +import { DesktopTelemetryPublisher } from "../telemetry/DesktopTelemetryPublisher.ts"; + +export interface DesktopCuaDriverDependencies { + readonly loadEmbedded: () => Promise<{ + readonly EmbeddedCuaDriverHost: new ( + path: string, + hostBundleId: string, + ) => Pick; + }>; + readonly loadElectron: () => Promise< + Pick< + typeof import("@trycua/cua-driver/electron"), + "requestMacOSPermissions" | "hasRequiredMacOSPermissions" + > + >; +} +const decodeMcpConfiguration = Schema.decodeEffect(CuaDriverMcpConfiguration); + +const defaultDependencies: DesktopCuaDriverDependencies = { + loadEmbedded: () => import("@trycua/cua-driver/embedded"), + loadElectron: () => import("@trycua/cua-driver/electron"), +}; + +export const resolveEmbeddedDriverPath = ( + environment: NodeJS.ProcessEnv, + desktop: Pick< + DesktopEnvironment.DesktopEnvironment["Service"], + "isPackaged" | "platform" | "resourcesPath" | "path" + >, +): Option.Option => + desktop.isPackaged + ? Option.some( + desktop.platform === "darwin" + ? desktop.path.join(desktop.resourcesPath, "cua-driver") + : desktop.path.join( + desktop.resourcesPath, + "cua-driver", + desktop.platform === "win32" ? "cua-driver.exe" : "cua-driver", + ), + ) + : Option.fromNullishOr(environment.T3CODE_CUA_DRIVER_PATH).pipe( + Option.map((value) => value.trim()), + Option.filter((value) => value.length > 0), + ); + +interface OwnedHost { + readonly host: Pick; + readonly abort: AbortController; + startPromise?: Promise; + monitorPromise?: Promise; + releasePromise?: Promise; +} + +interface HostedDriver { + readonly owned: OwnedHost; + readonly connection: EmbeddedDriverConnection; + requestId: string; +} + +export class DesktopCuaDriver extends Context.Service< + DesktopCuaDriver, + { + readonly listen: Effect.Effect; + } +>()("@t3tools/desktop/cua/DesktopCuaDriver") {} + +export const make = Effect.fn("desktop.cuaDriver.make")(function* ( + dependencies: DesktopCuaDriverDependencies = defaultDependencies, +) { + const environment = yield* DesktopEnvironment.DesktopEnvironment; + const publisher = yield* DesktopTelemetryPublisher; + const scope = yield* Scope.Scope; + const mutex = yield* Semaphore.make(1); + let active: HostedDriver | undefined; + const hosts = new Set(); + + const cleanup = (owned: OwnedHost) => + Effect.suspend(() => { + if (active?.owned === owned) active = undefined; + if (!owned.releasePromise) { + owned.abort.abort(); + // A native call may ignore cancellation. Keep its host alive until it + // settles, then stop it and destroy it exactly once, even after timeout. + owned.releasePromise = (async () => { + try { + await owned.startPromise; + } catch { + /* Failed starts still own a host. */ + } + try { + await owned.host.stop(); + } catch { + /* Destruction must still run. */ + } + try { + await owned.monitorPromise; + } catch { + /* Cancellation ends the monitor. */ + } + try { + owned.host.uniffiDestroy(); + } catch { + /* Cleanup cannot crash Electron. */ + } + hosts.delete(owned); + })(); + } + return Effect.tryPromise(() => owned.releasePromise!).pipe( + Effect.interruptible, + Effect.timeout("3 seconds"), + Effect.ignoreCause, + ); + }); + const stop = Effect.suspend(() => { + const current = [...hosts]; + active = undefined; + return Effect.forEach(current, cleanup, { concurrency: "unbounded", discard: true }); + }); + const retireNewHosts = Effect.suspend(() => { + active = undefined; + return Effect.forEach( + [...hosts].filter((owned) => !owned.releasePromise), + cleanup, + { concurrency: "unbounded", discard: true }, + ); + }); + yield* Effect.addFinalizer(() => retireNewHosts); + + const unavailable = (requestId: string, message: string) => + publisher.publishCuaReport({ + version: 1, + type: "cuaDriverReport", + requestId, + status: "unavailable", + message, + }); + const handle = Effect.fn("desktop.cuaDriver.handle")(function* ( + request: DesktopCuaDriverRequest, + ) { + if (!request.enabled) { + yield* stop; + yield* hosts.size === 0 + ? publisher.publishCuaReport({ + version: 1, + type: "cuaDriverReport", + requestId: request.requestId, + status: "stopped", + }) + : unavailable( + request.requestId, + "Cua Driver is still stopping. Wait for it to exit before starting another session.", + ); + return; + } + if (active) { + active.requestId = request.requestId; + yield* publisher.publishCuaReport({ + version: 1, + type: "cuaDriverReport", + requestId: request.requestId, + status: "ready", + mcp: active.connection.mcp, + }); + return; + } + if (hosts.size > 0) { + yield* unavailable( + request.requestId, + "The previous Cua Driver is still stopping. Try a new agent session after it exits.", + ); + return; + } + const binaryPath = resolveEmbeddedDriverPath(process.env, environment); + if (Option.isNone(binaryPath)) { + yield* unavailable( + request.requestId, + "Cua Driver is not configured for this desktop installation.", + ); + return; + } + if (environment.platform === "darwin") { + const helpers = yield* Effect.tryPromise(dependencies.loadElectron); + const allowed = yield* Effect.try(() => + helpers.hasRequiredMacOSPermissions(helpers.requestMacOSPermissions()), + ); + if (!allowed) { + yield* unavailable( + request.requestId, + "T3 Code needs Accessibility and Screen Recording access before Cua Driver can start.", + ); + return; + } + } + const module = yield* Effect.tryPromise(dependencies.loadEmbedded); + const owned = yield* Effect.try(() => { + const owned: OwnedHost = { + host: new module.EmbeddedCuaDriverHost(binaryPath.value, environment.appUserModelId), + abort: new AbortController(), + }; + hosts.add(owned); + return owned; + }); + yield* Effect.gen(function* () { + const connection = yield* Effect.tryPromise((signal) => { + signal.addEventListener("abort", () => owned.abort.abort(), { once: true }); + owned.startPromise = owned.host.start({ signal: owned.abort.signal }); + return owned.startPromise; + }); + const mcp = yield* decodeMcpConfiguration(connection.mcp); + yield* Effect.try(() => { + owned.monitorPromise = owned.host.waitForExit(connection.generation, { + signal: owned.abort.signal, + }); + void owned.monitorPromise.catch(() => undefined); + }); + const hosted: HostedDriver = { owned, connection, requestId: request.requestId }; + active = hosted; + yield* publisher.publishCuaReport({ + version: 1, + type: "cuaDriverReport", + requestId: request.requestId, + status: "ready", + mcp, + }); + yield* Effect.tryPromise(() => owned.monitorPromise!).pipe( + Effect.ignore, + Effect.andThen( + mutex.withPermits(1)( + Effect.suspend(() => { + if (active !== hosted) return Effect.void; + active = undefined; + return cleanup(owned).pipe( + Effect.andThen( + unavailable( + hosted.requestId, + "Cua Driver exited. A new agent session can try starting it again.", + ), + ), + ); + }), + ), + ), + Effect.forkIn(scope), + ); + }).pipe(Effect.onExit((exit) => (Exit.isFailure(exit) ? cleanup(owned) : Effect.void))); + }); + return DesktopCuaDriver.of({ + listen: publisher.cuaRequests.pipe( + Stream.runForEach((request) => + mutex.withPermits(1)( + handle(request).pipe( + Effect.timeout("20 seconds"), + Effect.catchCause((cause) => + Cause.hasInterrupts(cause) + ? Effect.failCause(cause) + : retireNewHosts.pipe( + Effect.andThen( + unavailable( + request.requestId, + "Cua Driver could not start. Check its installation and operating system permissions.", + ), + ), + ), + ), + ), + ), + ), + Effect.forkScoped, + Effect.asVoid, + ), + }); +}); + +export const layer = Layer.effect(DesktopCuaDriver, make()); diff --git a/apps/desktop/src/main.ts b/apps/desktop/src/main.ts index ed920abcdc8f..61221ec48c64 100644 --- a/apps/desktop/src/main.ts +++ b/apps/desktop/src/main.ts @@ -57,6 +57,7 @@ import * as DesktopShellEnvironment from "./shell/DesktopShellEnvironment.ts"; import * as DesktopSshEnvironment from "./ssh/DesktopSshEnvironment.ts"; import * as DesktopSshPasswordPrompts from "./ssh/DesktopSshPasswordPrompts.ts"; import * as DesktopState from "./app/DesktopState.ts"; +import * as DesktopCuaDriver from "./cua/DesktopCuaDriver.ts"; import * as DesktopTelemetryPublisher from "./telemetry/DesktopTelemetryPublisher.ts"; import * as DesktopUpdates from "./updates/DesktopUpdates.ts"; import * as BrowserImport from "./preview/BrowserImport/BrowserImport.ts"; @@ -207,6 +208,7 @@ const desktopApplicationLayer = Layer.mergeAll( ).pipe( Layer.provideMerge(desktopSnapShotLayer), Layer.provideMerge(DesktopUpdates.layer), + Layer.provideMerge(DesktopCuaDriver.layer), Layer.provideMerge(desktopWslBackendLayer), Layer.provideMerge(desktopLocalEnvironmentAuthLayer), ); diff --git a/apps/desktop/src/telemetry/DesktopTelemetryPublisher.test.ts b/apps/desktop/src/telemetry/DesktopTelemetryPublisher.test.ts index d705252ef588..aef562a4c1f0 100644 --- a/apps/desktop/src/telemetry/DesktopTelemetryPublisher.test.ts +++ b/apps/desktop/src/telemetry/DesktopTelemetryPublisher.test.ts @@ -19,6 +19,10 @@ import * as ElectronApp from "../electron/ElectronApp.ts"; import * as ElectronPowerMonitor from "../electron/ElectronPowerMonitor.ts"; import * as DesktopTelemetryPublisher from "./DesktopTelemetryPublisher.ts"; +const decodeMessage = Schema.decodeUnknownEffect( + Schema.fromJsonString(DesktopHostTelemetryMessage), +); + function makeElectronAppLayer( metrics: ReadonlyArray, onMetricsRead: () => void = () => undefined, @@ -150,9 +154,6 @@ describe("DesktopTelemetryPublisher", () => { const publisher = yield* DesktopTelemetryPublisher.DesktopTelemetryPublisher; const encoded = yield* publisher.encoded.pipe(Stream.take(2), Stream.runCollect); const decoder = new TextDecoder(); - const decodeMessage = Schema.decodeUnknownEffect( - Schema.fromJsonString(DesktopHostTelemetryMessage), - ); const messages = yield* Effect.forEach(encoded, (bytes) => decodeMessage(decoder.decode(bytes).trim()), ); @@ -411,6 +412,53 @@ describe("DesktopTelemetryPublisher", () => { yield* Effect.gen(function* () { const publisher = yield* DesktopTelemetryPublisher.DesktopTelemetryPublisher; + yield* publisher.handleControlForSource("wsl:Ubuntu", { + version: 1, + type: "cuaDriverRequest", + requestId: "remote", + enabled: true, + }); + yield* publisher.handleControlForSource("primary", { + version: 1, + type: "cuaDriverRequest", + requestId: "local", + enabled: true, + }); + const cuaRequest = yield* Stream.runHead(publisher.cuaRequests); + assert.equal(Option.getOrThrow(cuaRequest).requestId, "local"); + yield* publisher.removeControlSource("wsl:Ubuntu"); + yield* publisher.removeControlSource("primary"); + const detachRequest = Option.getOrThrow(yield* Stream.runHead(publisher.cuaRequests)); + assert.equal(detachRequest.requestId, "desktop-primary-detached"); + assert.equal(detachRequest.enabled, false); + + const primaryAttached = yield* Deferred.make(); + const cuaReportFiber = yield* publisher.encodedForSource("primary").pipe( + Stream.mapEffect((bytes) => decodeMessage(new TextDecoder().decode(bytes).trim())), + Stream.tap((message) => + message.type === "desktopTelemetryHello" + ? Deferred.succeed(primaryAttached, undefined) + : Effect.void, + ), + Stream.filter((message) => message.type === "cuaDriverReport"), + Stream.runHead, + Effect.forkChild, + ); + yield* Deferred.await(primaryAttached); + const cuaReport = { + version: 1, + type: "cuaDriverReport", + requestId: "local", + status: "ready", + mcp: { + command: "/driver", + args: ["mcp"], + environment: [{ name: "SESSION", value: "opaque" }], + }, + } as const; + yield* publisher.publishCuaReport(cuaReport); + assert.deepEqual(Option.getOrThrow(yield* Fiber.join(cuaReportFiber)), cuaReport); + const requestFiber = yield* Stream.runHead(publisher.updateRequests).pipe(Effect.forkChild); yield* Effect.yieldNow; yield* publisher.handleControlForSource("test", { @@ -449,9 +497,6 @@ describe("DesktopTelemetryPublisher", () => { // A subscriber that attaches after the publish (the backend spawned // by a relaunch) still sees the latest report replayed. const decoder = new TextDecoder(); - const decodeMessage = Schema.decodeUnknownEffect( - Schema.fromJsonString(DesktopHostTelemetryMessage), - ); const replayed = yield* publisher.encoded.pipe( Stream.mapEffect((bytes) => decodeMessage(decoder.decode(bytes).trim())), Stream.filter((message) => message.type === "desktopUpdateStatus"), @@ -463,6 +508,15 @@ describe("DesktopTelemetryPublisher", () => { } assert.equal(replayedReport.outcome, "up-to-date"); assert.equal(replayedReport.state.currentVersion, "1.2.3"); + const wslMessages = yield* publisher.encodedForSource("wsl:Ubuntu").pipe( + Stream.mapEffect((bytes) => decodeMessage(decoder.decode(bytes).trim())), + Stream.takeUntil((message) => message.type === "desktopUpdateStatus"), + Stream.runCollect, + ); + assert.equal( + wslMessages.some((message) => message.type === "cuaDriverReport"), + false, + ); }).pipe(Effect.provide(layer)); }), ); diff --git a/apps/desktop/src/telemetry/DesktopTelemetryPublisher.ts b/apps/desktop/src/telemetry/DesktopTelemetryPublisher.ts index 8fd478c821e7..e9d83a807983 100644 --- a/apps/desktop/src/telemetry/DesktopTelemetryPublisher.ts +++ b/apps/desktop/src/telemetry/DesktopTelemetryPublisher.ts @@ -1,5 +1,7 @@ import { DesktopHostTelemetryMessage, + type DesktopCuaDriverRequest, + type DesktopCuaDriverReport, type DesktopHostTelemetrySnapshot, type DesktopTelemetryControlMessage, type DesktopTelemetryCancelDesktopUpdate, @@ -62,6 +64,9 @@ export class DesktopTelemetryPublisher extends Context.Service< { readonly latest: Effect.Effect>; readonly changes: Stream.Stream; + readonly cuaRequests: Stream.Stream; + readonly publishCuaReport: (report: DesktopCuaDriverReport) => Effect.Effect; + readonly encodedForSource: (sourceId: string) => Stream.Stream; readonly encoded: Stream.Stream; readonly handleControl: (message: DesktopTelemetryControlMessage) => Effect.Effect; readonly handleControlForSource: ( @@ -175,6 +180,8 @@ export const make = Effect.fn("desktop.telemetryPublisher.make")(function* () { const sequence = yield* Ref.make(0); const latestUpdateReport = yield* Ref.make(Option.none()); const updateReportChanges = yield* PubSub.sliding(16); + const cuaRequests = yield* Queue.unbounded(); + const cuaReports = yield* PubSub.unbounded(); const updateRequestQueue = yield* Queue.unbounded(); const updateCommitQueue = yield* Queue.unbounded(); const updateCancellationQueue = yield* Queue.unbounded(); @@ -320,6 +327,10 @@ export const make = Effect.fn("desktop.telemetryPublisher.make")(function* () { message, ) => { switch (message.type) { + case "cuaDriverRequest": + return sourceId === "primary" + ? Queue.offer(cuaRequests, message).pipe(Effect.asVoid) + : Effect.void; case "setDiagnosticsDemand": return Ref.modify(diagnosticsDemandSources, (sources) => { const previous = sources.size > 0; @@ -364,6 +375,16 @@ export const make = Effect.fn("desktop.telemetryPublisher.make")(function* () { ? Effect.void : Queue.offer(sampleTriggers, undefined).pipe(Effect.asVoid), ), + Effect.andThen( + sourceId === "primary" + ? Queue.offer(cuaRequests, { + version: 1, + type: "cuaDriverRequest", + requestId: "desktop-primary-detached", + enabled: false, + }).pipe(Effect.asVoid) + : Effect.void, + ), ); const handleControl: DesktopTelemetryPublisher["Service"]["handleControl"] = (message) => handleControlForSource("legacy", message); @@ -415,6 +436,22 @@ export const make = Effect.fn("desktop.telemetryPublisher.make")(function* () { latest: Ref.get(latest), changes: Stream.fromPubSub(changes), encoded, + encodedForSource: (sourceId) => + sourceId === "primary" + ? Stream.unwrap( + Effect.gen(function* () { + const subscription = yield* PubSub.subscribe(cuaReports); + return Stream.merge( + encoded, + Stream.fromSubscription(subscription).pipe( + Stream.map((report) => textEncoder.encode(`${encodeMessage(report)}\n`)), + ), + ); + }), + ) + : encoded, + cuaRequests: Stream.fromQueue(cuaRequests), + publishCuaReport: (report) => PubSub.publish(cuaReports, report).pipe(Effect.asVoid), handleControl, handleControlForSource, removeControlSource, diff --git a/apps/desktop/src/updates/DesktopRemoteUpdates.test.ts b/apps/desktop/src/updates/DesktopRemoteUpdates.test.ts index 0f4cb970da2c..42ce3a976b33 100644 --- a/apps/desktop/src/updates/DesktopRemoteUpdates.test.ts +++ b/apps/desktop/src/updates/DesktopRemoteUpdates.test.ts @@ -57,6 +57,9 @@ function runRemoteUpdatesTest( latest: Effect.succeedNone, changes: Stream.empty, encoded: Stream.empty, + encodedForSource: () => Stream.empty, + cuaRequests: Stream.empty, + publishCuaReport: () => Effect.void, handleControl: () => Effect.void, handleControlForSource: () => Effect.void, removeControlSource: () => Effect.void, diff --git a/apps/server/package.json b/apps/server/package.json index 7d27d39d2eb3..795a4feb1bbe 100644 --- a/apps/server/package.json +++ b/apps/server/package.json @@ -28,9 +28,11 @@ "@effect/sql-sqlite-bun": "catalog:", "@ff-labs/fff-node": "0.9.4", "@opencode-ai/sdk": "^1.3.15", + "@trycua/cua-driver": "0.24.0", "effect": "catalog:", "msgpackr-extract": "3.0.4", "node-pty": "^1.1.0", + "smol-toml": "1.7.0", "stream-chain": "^4.2.5", "stream-json": "3.6.0", "yaml": "catalog:", diff --git a/apps/server/src/cua/CuaDriver.test.ts b/apps/server/src/cua/CuaDriver.test.ts new file mode 100644 index 000000000000..a0856fcf2507 --- /dev/null +++ b/apps/server/src/cua/CuaDriver.test.ts @@ -0,0 +1,405 @@ +import { it } from "@effect/vitest"; +import type { CuaDriverMcpConfiguration, DesktopCuaDriverReport } from "@t3tools/contracts"; +import type { EmbeddedDriverConnection } from "@trycua/cua-driver/embedded"; +import { Deferred, Effect, Exit, Fiber, Option, PubSub, Ref, Scope, Stream } from "effect"; +import { TestClock } from "effect/testing"; +import { describe, expect } from "vite-plus/test"; + +import { + CuaDriverError, + makeCuaDriver, + makeDesktopHostFactory, + makeStandaloneHostFactory, +} from "./CuaDriver.ts"; + +const mcp: CuaDriverMcpConfiguration = { command: "/driver", args: ["mcp"], environment: [] }; + +const standaloneFixture = Effect.fn(function* () { + const start = Promise.withResolvers(); + const stop = Promise.withResolvers(); + const monitor = Promise.withResolvers(); + const started = Promise.withResolvers(); + const stopped = Promise.withResolvers(); + const watching = Promise.withResolvers(); + const destroyed = Promise.withResolvers(); + let creates = 0; + let stops = 0; + let destroys = 0; + const factory = yield* makeStandaloneHostFactory("/driver", async () => ({ + EmbeddedCuaDriverHost: class { + constructor() { + creates++; + } + start() { + started.resolve(); + return start.promise; + } + stop() { + stops++; + stopped.resolve(); + return stop.promise; + } + waitForExit(_generation: string, options?: { signal: AbortSignal }) { + watching.resolve(options!.signal); + return monitor.promise; + } + uniffiDestroy() { + destroys++; + destroyed.resolve(); + } + }, + })); + 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] }, + }; + return { + factory, + start, + stop, + monitor, + started, + stopped, + watching, + destroyed, + connection, + counts: () => ({ creates, stops, destroys }), + }; +}); + +describe("standalone Cua Driver ownership", () => { + it.effect("retains a cancelled start until start and stop settle, with one cleanup", () => + Effect.gen(function* () { + const f = yield* standaloneFixture(); + const host = yield* f.factory; + const starting = yield* host.start.pipe(Effect.forkChild({ startImmediately: true })); + yield* Effect.promise(() => f.started.promise); + yield* Fiber.interrupt(starting); + const stopping = yield* host.stop.pipe(Effect.forkChild({ startImmediately: true })); + const stoppingAgain = yield* host.stop.pipe(Effect.forkChild({ startImmediately: true })); + expect(f.counts()).toEqual({ creates: 1, stops: 0, destroys: 0 }); + expect(Exit.isFailure(yield* Effect.exit(f.factory))).toBe(true); + f.start.resolve(f.connection); + yield* Effect.promise(() => f.stopped.promise); + expect(f.counts()).toEqual({ creates: 1, stops: 1, destroys: 0 }); + f.stop.resolve(); + yield* Effect.all([Fiber.join(stopping), Fiber.join(stoppingAgain)]); + expect(f.counts()).toEqual({ creates: 1, stops: 1, destroys: 1 }); + const replacement = yield* f.factory; + yield* replacement.stop; + expect(f.counts()).toEqual({ creates: 2, stops: 2, destroys: 2 }); + }), + ); + + it.effect("aborts the monitor and retains ownership after the stop waiter is cancelled", () => + Effect.gen(function* () { + const f = yield* standaloneFixture(); + const host = yield* f.factory; + f.start.resolve(f.connection); + expect(yield* host.start).toEqual(mcp); + const signal = yield* Effect.promise(() => f.watching.promise); + const stopping = yield* host.stop.pipe(Effect.forkChild({ startImmediately: true })); + yield* Effect.promise(() => f.stopped.promise); + expect(signal.aborted).toBe(true); + yield* Fiber.interrupt(stopping); + f.stop.resolve(); + expect(Exit.isFailure(yield* Effect.exit(f.factory))).toBe(true); + expect(f.counts()).toEqual({ creates: 1, stops: 1, destroys: 0 }); + f.monitor.reject(new Error("monitor cancelled")); + yield* Effect.promise(() => f.destroyed.promise); + yield* host.stop; + expect(f.counts()).toEqual({ creates: 1, stops: 1, destroys: 1 }); + }), + ); + + it.effect("destroys failed starts after a rejected native stop settles", () => + Effect.gen(function* () { + const f = yield* standaloneFixture(); + const host = yield* f.factory; + const starting = yield* host.start.pipe(Effect.forkChild({ startImmediately: true })); + yield* Effect.promise(() => f.started.promise); + f.start.reject(new Error("start failed")); + expect(Exit.isFailure(yield* Fiber.await(starting))).toBe(true); + const stopping = yield* host.stop.pipe(Effect.forkChild({ startImmediately: true })); + yield* Effect.promise(() => f.stopped.promise); + f.stop.reject(new Error("stop failed")); + yield* Fiber.join(stopping); + expect(f.counts()).toEqual({ creates: 1, stops: 1, destroys: 1 }); + }), + ); +}); + +const fixture = Effect.fn(function* (initialEnabled = true) { + const enabled = yield* Ref.make(initialEnabled); + const changes = yield* PubSub.unbounded(); + const subscription = yield* PubSub.subscribe(changes); + const started = yield* Deferred.make(); + const stopped = yield* Deferred.make(); + const result = yield* Deferred.make(); + const exited = yield* Deferred.make(); + const watching = yield* Deferred.make(); + let starts = 0; + let stops = 0; + const service = yield* makeCuaDriver({ + enabled: Ref.get(enabled), + changes: Stream.fromSubscription(subscription), + createHost: Effect.succeed({ + start: Effect.sync(() => { + starts++; + }).pipe( + Effect.andThen(Deferred.succeed(started, undefined)), + Effect.andThen(Deferred.await(result)), + ), + stop: Effect.sync(() => { + stops++; + }).pipe(Effect.andThen(Deferred.succeed(stopped, undefined)), Effect.asVoid), + waitForExit: Deferred.succeed(watching, undefined).pipe( + Effect.andThen(Deferred.await(exited)), + ), + }), + }); + const setEnabled = (value: boolean) => + Ref.set(enabled, value).pipe(Effect.andThen(PubSub.publish(changes, value))); + return { + service, + started, + stopped, + result, + exited, + watching, + setEnabled, + starts: () => starts, + stops: () => stops, + }; +}); + +describe("CuaDriver", () => { + it.effect("bounds shutdown when the host never acknowledges stop", () => + Effect.gen(function* () { + const scope = yield* Scope.make(); + const stopping = yield* Deferred.make(); + const service = yield* makeCuaDriver({ + enabled: Effect.succeed(true), + changes: Stream.empty, + createHost: Effect.succeed({ + start: Effect.succeed(mcp), + stop: Deferred.succeed(stopping, undefined).pipe(Effect.andThen(Effect.never)), + waitForExit: Effect.never, + }), + }).pipe(Scope.provide(scope)); + expect(yield* service.acquire).toEqual(Option.some(mcp)); + const closing = yield* Scope.close(scope, Exit.void).pipe(Effect.forkChild); + yield* Deferred.await(stopping); + yield* TestClock.adjust("5 seconds"); + yield* Fiber.join(closing); + expect(yield* service.acquire).toEqual(Option.none()); + }), + ); + + it.effect("late cleanup from a revoked start cannot stop its replacement", () => + Effect.scoped( + Effect.gen(function* () { + const enabled = yield* Ref.make(true); + const changes = yield* PubSub.unbounded(); + const subscription = yield* PubSub.subscribe(changes); + const firstStarted = yield* Deferred.make(); + const firstStopped = yield* Deferred.make(); + const lateReady = yield* Deferred.make(); + const lateConsumed = yield* Deferred.make(); + let count = 0; + let oldStops = 0; + let newStops = 0; + const service = yield* makeCuaDriver({ + enabled: Ref.get(enabled), + changes: Stream.fromSubscription(subscription), + createHost: Effect.sync(() => { + count++; + return count === 1 + ? { + start: Deferred.succeed(firstStarted, undefined).pipe( + Effect.andThen(Deferred.await(lateReady)), + Effect.tap(() => Deferred.succeed(lateConsumed, undefined)), + ), + stop: Effect.sync(() => { + oldStops++; + }).pipe(Effect.andThen(Deferred.succeed(firstStopped, undefined)), Effect.asVoid), + waitForExit: Effect.never, + } + : { + start: Effect.succeed(mcp), + stop: Effect.sync(() => { + newStops++; + }), + waitForExit: Effect.never, + }; + }), + }); + const old = yield* service.acquire.pipe(Effect.forkChild); + yield* Deferred.await(firstStarted); + yield* Ref.set(enabled, false); + yield* PubSub.publish(changes, false); + yield* Deferred.await(firstStopped); + expect(yield* Fiber.join(old)).toEqual(Option.none()); + yield* Ref.set(enabled, true); + expect(yield* service.acquire).toEqual(Option.some(mcp)); + yield* Deferred.succeed(lateReady, mcp); + yield* Deferred.await(lateConsumed); + expect(yield* service.acquire).toEqual(Option.some(mcp)); + expect(count).toBe(2); + expect(oldStops).toBe(1); + expect(newStops).toBe(0); + }), + ), + ); + + it.effect("does not start a disabled host", () => + Effect.scoped( + Effect.gen(function* () { + const f = yield* fixture(false); + expect(yield* f.service.enabled).toBe(false); + expect(yield* f.service.acquire).toEqual(Option.none()); + expect(f.starts()).toBe(0); + }), + ), + ); + + it.effect("shares a lazy start across concurrent callers and owns shutdown", () => + Effect.gen(function* () { + const scope = yield* Scope.make(); + const f = yield* fixture().pipe(Scope.provide(scope)); + const first = yield* f.service.acquire.pipe(Effect.forkChild); + const second = yield* f.service.acquire.pipe(Effect.forkChild); + yield* Deferred.await(f.started); + yield* Deferred.succeed(f.result, mcp); + expect(yield* Fiber.join(first)).toEqual(Option.some(mcp)); + expect(yield* Fiber.join(second)).toEqual(Option.some(mcp)); + expect(f.starts()).toBe(1); + yield* Scope.close(scope, Exit.void); + expect(f.stops()).toBe(1); + }), + ); + + it.effect("caller cancellation cannot cancel the environment start", () => + Effect.scoped( + Effect.gen(function* () { + const f = yield* fixture(); + const caller = yield* f.service.acquire.pipe(Effect.forkChild); + yield* Deferred.await(f.started); + yield* Fiber.interrupt(caller); + yield* Deferred.succeed(f.result, mcp); + expect(yield* f.service.acquire).toEqual(Option.some(mcp)); + expect(f.starts()).toBe(1); + }), + ), + ); + + it.effect("disabling during start revokes waiters and rejects late readiness", () => + Effect.scoped( + Effect.gen(function* () { + const f = yield* fixture(); + const caller = yield* f.service.acquire.pipe(Effect.forkChild); + yield* Deferred.await(f.started); + yield* f.setEnabled(false); + yield* Deferred.await(f.stopped); + expect(yield* Fiber.join(caller)).toEqual(Option.none()); + yield* Deferred.succeed(f.result, mcp); + expect(yield* f.service.acquire).toEqual(Option.none()); + expect(f.stops()).toBe(1); + }), + ), + ); + + it.effect("shutdown during start resolves callers and stops the host", () => + Effect.gen(function* () { + const scope = yield* Scope.make(); + const f = yield* fixture().pipe(Scope.provide(scope)); + const caller = yield* f.service.acquire.pipe(Effect.forkChild); + yield* Deferred.await(f.started); + yield* Scope.close(scope, Exit.void); + expect(yield* Fiber.join(caller)).toEqual(Option.none()); + expect(f.stops()).toBe(1); + expect(yield* f.service.acquire).toEqual(Option.none()); + }), + ); + + it.effect("cleans a failed start and allows the next session to retry", () => + Effect.scoped( + Effect.gen(function* () { + const f = yield* fixture(); + const caller = yield* f.service.acquire.pipe(Effect.forkChild); + yield* Deferred.await(f.started); + yield* Deferred.fail(f.result, new CuaDriverError({ message: "test failure" })); + expect(yield* Fiber.join(caller)).toEqual(Option.none()); + expect(f.stops()).toBe(1); + expect(yield* f.service.acquire).toEqual(Option.none()); + expect(f.starts()).toBe(2); + expect(f.stops()).toBe(2); + }), + ), + ); + + it.effect("bounds startup and cleans the failed attempt", () => + Effect.scoped( + Effect.gen(function* () { + const f = yield* fixture(); + const caller = yield* f.service.acquire.pipe(Effect.forkChild); + yield* Deferred.await(f.started); + yield* TestClock.adjust("30 seconds"); + expect(yield* Fiber.join(caller)).toEqual(Option.none()); + expect(f.stops()).toBe(1); + }), + ), + ); + + it.effect("clears a crashed generation without restarting it automatically", () => + Effect.scoped( + Effect.gen(function* () { + const f = yield* fixture(); + yield* Deferred.succeed(f.result, mcp); + expect(yield* f.service.acquire).toEqual(Option.some(mcp)); + yield* Deferred.await(f.watching); + yield* Deferred.succeed(f.exited, undefined); + yield* Deferred.await(f.stopped); + expect(f.starts()).toBe(1); + expect(yield* f.service.acquire).toEqual(Option.some(mcp)); + expect(f.starts()).toBe(2); + }), + ), + ); + + it.effect("subscribes before synchronous desktop replies and observes crashes", () => + Effect.scoped( + Effect.gen(function* () { + const reports = yield* PubSub.unbounded(); + let startId = ""; + const factory = yield* makeDesktopHostFactory({ + cuaReports: Stream.fromPubSub(reports), + requestCuaDriver: (requestId, enabled) => { + if (enabled) startId = requestId; + return PubSub.publish( + reports, + enabled + ? { version: 1, type: "cuaDriverReport", requestId, status: "ready", mcp } + : { version: 1, type: "cuaDriverReport", requestId, status: "stopped" }, + ).pipe(Effect.asVoid); + }, + }); + const host = yield* factory; + expect(yield* host.start).toEqual(mcp); + yield* PubSub.publish(reports, { + version: 1, + type: "cuaDriverReport", + requestId: startId, + status: "unavailable", + }); + yield* host.waitForExit; + yield* host.stop; + }), + ), + ); +}); diff --git a/apps/server/src/cua/CuaDriver.ts b/apps/server/src/cua/CuaDriver.ts new file mode 100644 index 000000000000..28b734866dc6 --- /dev/null +++ b/apps/server/src/cua/CuaDriver.ts @@ -0,0 +1,345 @@ +import * as NodeCrypto from "node:crypto"; + +import type { CuaDriverMcpConfiguration, DesktopCuaDriverReport } from "@t3tools/contracts"; +import type { EmbeddedCuaDriverHost, EmbeddedDriverConnection } from "@trycua/cua-driver/embedded"; +import { Context, Deferred, Effect, Layer, Option, Schema, Scope, Semaphore, Stream } from "effect"; + +import { ServerConfig } from "../config.ts"; +import { DesktopTelemetryReceiver } from "../resourceTelemetry/DesktopTelemetryReceiver.ts"; +import { ServerSettingsService } from "../serverSettings.ts"; + +export class CuaDriver extends Context.Service< + CuaDriver, + { + readonly enabled: Effect.Effect; + readonly acquire: Effect.Effect>; + } +>()("t3/cua/CuaDriver") {} + +export class CuaDriverError extends Schema.TaggedError()("CuaDriverError", { + message: Schema.String, +}) {} + +export interface CuaDriverHost { + readonly start: Effect.Effect; + readonly stop: Effect.Effect; + readonly waitForExit: Effect.Effect; +} + +export interface CuaDriverOptions { + readonly createHost: Effect.Effect; + readonly enabled: Effect.Effect; + readonly changes: Stream.Stream; +} + +const START_TIMEOUT = "30 seconds"; +const STOP_TIMEOUT = "5 seconds"; + +/** One environment owns the host; cancelling a session only cancels its wait. */ +export const makeCuaDriver = Effect.fn("CuaDriver.make")(function* (options: CuaDriverOptions) { + const scope = yield* Effect.scope; + const workers = yield* Scope.fork(scope); + const mutex = yield* Semaphore.make(1); + type Attempt = { + readonly ready: Deferred.Deferred>; + host?: CuaDriverHost; + stopped?: boolean; + }; + let current: Attempt | undefined; + let closed = false; + + const stop = (attempt: Attempt) => + Effect.suspend(() => { + if (!attempt.host || attempt.stopped) return Effect.void; + attempt.stopped = true; + return attempt.host.stop.pipe( + Effect.interruptible, + Effect.timeout(STOP_TIMEOUT), + Effect.catchCause(() => Effect.logWarning("Cua Driver host cleanup did not complete.")), + ); + }); + + const revoke = mutex.withPermits(1)( + Effect.gen(function* () { + const previous = current; + current = undefined; + if (previous) { + yield* Deferred.succeed(previous.ready, Option.none()); + yield* stop(previous); + } + }), + ); + + yield* Effect.addFinalizer(() => + Effect.sync(() => { + closed = true; + }).pipe(Effect.andThen(revoke)), + ); + yield* options.changes.pipe( + Stream.runForEach((enabled) => (enabled ? Effect.void : revoke)), + Effect.forkIn(workers, { startImmediately: true }), + ); + + const launch = Effect.fn("CuaDriver.launch")( + function* (attempt: Attempt) { + const host = yield* options.createHost.pipe(Effect.timeout(START_TIMEOUT)); + attempt.host = host; + if (closed || current !== attempt) { + yield* stop(attempt); + return; + } + const mcp = yield* host.start.pipe(Effect.timeout(START_TIMEOUT)); + yield* mutex.withPermits(1)( + Effect.gen(function* () { + if (closed || current !== attempt || !(yield* options.enabled)) { + yield* stop(attempt); + yield* Deferred.succeed(attempt.ready, Option.none()); + if (current === attempt) current = undefined; + return; + } + yield* Deferred.succeed(attempt.ready, Option.some(mcp)); + }), + ); + if (current !== attempt) return; + yield* host.waitForExit; + yield* mutex.withPermits(1)( + Effect.gen(function* () { + if (current !== attempt) return; + current = undefined; + yield* stop(attempt); + }), + ); + }, + (effect, attempt) => + effect.pipe( + Effect.catchCause(() => + mutex.withPermits(1)( + Effect.gen(function* () { + if (current === attempt) current = undefined; + yield* stop(attempt); + yield* Deferred.succeed(attempt.ready, Option.none()); + yield* Effect.logWarning( + "Cua Driver is unavailable; this coding session will continue without computer use.", + ); + }), + ), + ), + ), + ); + + const acquire = Effect.gen(function* () { + const attempt = yield* mutex + .withPermits(1)( + Effect.gen(function* () { + if (closed || !(yield* options.enabled)) return undefined; + if (current) return current; + const next: Attempt = { + ready: yield* Deferred.make>(), + }; + current = next; + yield* launch(next).pipe(Effect.interruptible, Effect.forkIn(workers)); + return next; + }), + ) + .pipe(Effect.uninterruptible); + return attempt ? yield* Deferred.await(attempt.ready) : Option.none(); + }); + + return CuaDriver.of({ enabled: options.enabled, acquire }); +}); + +type StandaloneHostModule = { + readonly EmbeddedCuaDriverHost: new ( + path: string, + hostBundleId: string, + ) => Pick; +}; + +/** A timed-out native cleanup still owns the factory until every call settles. */ +export const makeStandaloneHostFactory = Effect.fn("CuaDriver.standaloneHostFactory")(function* ( + binaryPath: string, + loadEmbedded: () => Promise = () => import("@trycua/cua-driver/embedded"), +) { + const mutex = yield* Semaphore.make(1); + let occupied = false; + // @effect-diagnostics-next-line returnEffectInGen:off + return mutex.withPermits(1)( + Effect.gen(function* () { + if (occupied) { + return yield* new CuaDriverError({ message: "The previous Cua Driver is still stopping." }); + } + const { EmbeddedCuaDriverHost } = yield* Effect.tryPromise({ + try: loadEmbedded, + catch: () => new CuaDriverError({ message: "Could not load the Cua Driver SDK." }), + }); + const host = yield* Effect.try({ + try: () => new EmbeddedCuaDriverHost(binaryPath, "com.t3tools.t3code.server"), + catch: () => new CuaDriverError({ message: "Could not create the Cua Driver host." }), + }); + occupied = true; + const abort = new AbortController(); + let starting: Promise | undefined; + let monitoring: Promise | undefined; + let releasing: Promise | undefined; + let retired = false; + const stop = Effect.tryPromise({ + try: () => { + if (!releasing) { + retired = true; + abort.abort(); + releasing = (async () => { + try { + await starting; + } catch { + /* Failed starts still own a host. */ + } + try { + await host.stop(); + } catch { + /* Destruction must still run. */ + } + try { + await monitoring; + } catch { + /* Cancellation ends the monitor. */ + } + host.uniffiDestroy(); + occupied = false; + })(); + } + return releasing; + }, + catch: () => new CuaDriverError({ message: "Could not stop Cua Driver." }), + }); + return { + start: Effect.tryPromise({ + try: async (signal) => { + if (retired) throw new Error("Cua Driver start was revoked."); + signal.addEventListener("abort", () => abort.abort(), { once: true }); + starting ??= (async () => { + const connection = await host.start({ signal: abort.signal }); + if (retired) throw new Error("Cua Driver start was revoked."); + monitoring = host.waitForExit(connection.generation, { signal: abort.signal }); + void monitoring.catch(() => undefined); + return connection; + })(); + return (await starting).mcp; + }, + catch: () => new CuaDriverError({ message: "Could not start Cua Driver." }), + }), + stop, + waitForExit: Effect.tryPromise({ + try: async () => { + await monitoring; + }, + catch: () => new CuaDriverError({ message: "Cua Driver exited unexpectedly." }), + }), + } satisfies CuaDriverHost; + }), + ); +}); + +export const makeDesktopHostFactory = Effect.fn("CuaDriver.desktopHostFactory")(function* ( + receiver: Pick, +) { + const pending = new Map>(); + const exits = new Map>(); + yield* receiver.cuaReports.pipe( + Stream.runForEach((report) => + Effect.gen(function* () { + const reply = pending.get(report.requestId); + if (reply) yield* Deferred.succeed(reply, report); + if (report.status !== "ready") { + const exit = exits.get(report.requestId); + if (exit) yield* Deferred.succeed(exit, undefined); + } + }), + ), + Effect.forkScoped({ startImmediately: true }), + ); + + // Subscribe once for the service, then allocate a fresh request for each lazy start. + // @effect-diagnostics-next-line returnEffectInGen:off + return Effect.gen(function* () { + const requestId = NodeCrypto.randomUUID(); + const ready = yield* Deferred.make(); + const exit = yield* Deferred.make(); + let requested = false; + pending.set(requestId, ready); + exits.set(requestId, exit); + return { + start: Effect.gen(function* () { + requested = true; + yield* receiver + .requestCuaDriver(requestId, true) + .pipe( + Effect.mapError( + () => new CuaDriverError({ message: "Could not request desktop Cua Driver." }), + ), + ); + const report = yield* Deferred.await(ready); + if (report.status === "ready") return report.mcp; + if (report.message) + yield* Effect.logWarning("Cua Driver host reported unavailable.", { + message: report.message, + }); + return yield* new CuaDriverError({ message: "Desktop Cua Driver is unavailable." }); + }), + stop: Effect.gen(function* () { + if (!requested) { + pending.delete(requestId); + exits.delete(requestId); + return; + } + const stopId = NodeCrypto.randomUUID(); + const stopped = yield* Deferred.make(); + pending.set(stopId, stopped); + yield* receiver.requestCuaDriver(stopId, false).pipe( + Effect.mapError( + () => new CuaDriverError({ message: "Could not stop desktop Cua Driver." }), + ), + Effect.andThen(Deferred.await(stopped)), + Effect.ensuring( + Effect.sync(() => { + pending.delete(stopId); + pending.delete(requestId); + exits.delete(requestId); + }), + ), + ); + }), + waitForExit: Deferred.await(exit), + } satisfies CuaDriverHost; + }); +}); + +export const layer = Layer.effect( + CuaDriver, + Effect.gen(function* () { + const config = yield* ServerConfig; + const settings = yield* ServerSettingsService; + const receiver = yield* DesktopTelemetryReceiver; + const changes = yield* settings.subscribeChanges; + const binaryPath = process.env.T3CODE_CUA_DRIVER_PATH?.trim(); + const createHost = + config.mode === "desktop" && config.desktopTelemetryControlFd !== undefined + ? yield* makeDesktopHostFactory(receiver) + : binaryPath + ? yield* makeStandaloneHostFactory(binaryPath) + : Effect.logWarning( + "Cua Driver requires the T3 desktop host or an explicit T3CODE_CUA_DRIVER_PATH on this server.", + ).pipe( + Effect.andThen( + Effect.fail(new CuaDriverError({ message: "No Cua Driver host configured." })), + ), + ); + return yield* makeCuaDriver({ + createHost, + enabled: settings.getSettings.pipe( + Effect.map((value) => value.enableCua), + Effect.orElseSucceed(() => false), + ), + changes: changes.pipe(Stream.map((value) => value.enableCua)), + }); + }), +); diff --git a/apps/server/src/cua/codexCuaConfiguration.test.ts b/apps/server/src/cua/codexCuaConfiguration.test.ts new file mode 100644 index 000000000000..3f4fd3229f7d --- /dev/null +++ b/apps/server/src/cua/codexCuaConfiguration.test.ts @@ -0,0 +1,104 @@ +import * as NodeAssert from "node:assert/strict"; + +import { parse as parseToml } from "smol-toml"; +import { describe, it } from "vite-plus/test"; + +import { buildCuaDriverAppServerArgs, hasConfiguredCuaDriver } from "./codexCuaConfiguration.ts"; + +describe("hasConfiguredCuaDriver", () => { + it.each([ + '[mcp_servers.cua-driver]\ncommand = "custom"', + '["mcp_servers"."cua-driver"]\nenabled = false', + "['mcp_servers'.'cua-driver']\nargs = []", + 'mcp_servers = { "cua-driver" = { command = "custom" } }', + 'mcp_servers . "cua-driver" . command = "custom"', + ])("preserves a configured server in TOML: %s", (config) => { + NodeAssert.equal(hasConfiguredCuaDriver([], config), true); + }); + + it.each(["-c", "--config", "-c=", "--config=", "-cattached"])( + "detects explicit overrides with %s", + (flag) => { + const override = '"mcp_servers" . "cua-driver" . command = custom-driver'; + const argv = + flag === "-cattached" + ? [`-c${override}`] + : flag.endsWith("=") + ? [`${flag}${override}`] + : [flag, override]; + NodeAssert.equal(hasConfiguredCuaDriver(argv, undefined), true); + }, + ); + + it("detects inline table overrides and explicit disablement", () => { + NodeAssert.equal( + hasConfiguredCuaDriver(["-c", 'mcp_servers = { "cua-driver" = { enabled = false } }'], ""), + true, + ); + }); + + it("ignores unrelated servers, comments, string values and similarly named keys", () => { + NodeAssert.equal( + hasConfiguredCuaDriver( + ["--config", "model = gpt-5", "-c", 'mcp_servers.other.command = "cua-driver"'], + '# [mcp_servers.cua-driver]\n[mcp_servers.cua-driver-backup]\ncommand = "cua-driver"\n[other.mcp_servers.cua-driver]\ncommand = "custom"', + ), + false, + ); + NodeAssert.equal(hasConfiguredCuaDriver([], '"mcp_servers.cua-driver" = {}'), false); + }); + + it("respects argument boundaries without consuming adjacent flags", () => { + NodeAssert.equal( + hasConfiguredCuaDriver(["--", "-c", "mcp_servers.cua-driver = {}"], ""), + false, + ); + NodeAssert.equal(hasConfiguredCuaDriver(["--config", "--strict-config", "-c"], ""), false); + NodeAssert.equal( + hasConfiguredCuaDriver(["--config", "--config=mcp_servers.cua-driver = {}"], ""), + true, + ); + NodeAssert.equal(hasConfiguredCuaDriver(["mcp_servers.cua-driver = {}"], ""), false); + }); + + it("conservatively suppresses injection when configuration is malformed", () => { + NodeAssert.equal(hasConfiguredCuaDriver([], "[broken"), true); + NodeAssert.equal(hasConfiguredCuaDriver(["-c", "mcp_servers = { broken"], ""), true); + NodeAssert.equal(hasConfiguredCuaDriver(["--config=broken"], ""), true); + NodeAssert.equal(hasConfiguredCuaDriver([], undefined), false); + }); +}); + +describe("buildCuaDriverAppServerArgs", () => { + it("round-trips command, argv and environment without shell interpretation", () => { + const descriptor = { + command: 'C:\\Program Files\\Cua\\driver "preview".exe', + args: ["mcp", "--path", "/a path/with spaces", "", "$(echo untouched)", "line\n\t\u007f"], + environment: [ + { name: "DOT.KEY", value: 'C:\\a path\\"file"' }, + { name: 'QUOTED"KEY', value: "line\nreturn\r\t\b\f\u0000" }, + { name: "SPACE KEY", value: "emoji: \u{1f600}" }, + ], + }; + const argv = buildCuaDriverAppServerArgs(descriptor); + NodeAssert.equal(argv.length, 2); + NodeAssert.equal(argv[0], "-c"); + NodeAssert.deepStrictEqual(parseToml(argv[1]!), { + mcp_servers: { + "cua-driver": { + command: descriptor.command, + args: descriptor.args, + env: Object.fromEntries(descriptor.environment.map(({ name, value }) => [name, value])), + }, + }, + }); + NodeAssert.equal(hasConfiguredCuaDriver(argv, undefined), true); + }); + + it("encodes empty args and environment as TOML collections", () => { + const argv = buildCuaDriverAppServerArgs({ command: "driver", args: [], environment: [] }); + NodeAssert.deepStrictEqual(parseToml(argv[1]!), { + mcp_servers: { "cua-driver": { command: "driver", args: [], env: {} } }, + }); + }); +}); diff --git a/apps/server/src/cua/codexCuaConfiguration.ts b/apps/server/src/cua/codexCuaConfiguration.ts new file mode 100644 index 000000000000..d487cb6df7da --- /dev/null +++ b/apps/server/src/cua/codexCuaConfiguration.ts @@ -0,0 +1,70 @@ +import { Predicate } from "effect"; +import { parse as parseToml } from "smol-toml"; + +const parsedConfigHasCuaDriver = (config: unknown): boolean => + Predicate.isObject(config) && + Predicate.isObject(config.mcp_servers) && + Object.hasOwn(config.mcp_servers, "cua-driver"); + +const overrideHasCuaDriver = (override: string): boolean => { + try { + return parsedConfigHasCuaDriver(parseToml(override)); + } catch { + // Codex also accepts unquoted string values. Parse only the key in that case. + const assignmentIndex = override.indexOf("="); + if (assignmentIndex === -1) return true; + try { + const keyConfig = parseToml(`${override.slice(0, assignmentIndex)} = true`); + // An unparseable replacement of the entire MCP table is ambiguous. + return parsedConfigHasCuaDriver(keyConfig) || keyConfig.mcp_servers === true; + } catch { + return true; + } + } +}; + +/** Invalid configuration suppresses automatic injection so user configuration always wins. */ +export const hasConfiguredCuaDriver = ( + argv: readonly string[], + configToml: string | undefined, +): boolean => { + for (let index = 0; index < argv.length; index++) { + const argument = argv[index]; + if (argument === "--") break; + if (argument === "-c" || argument === "--config") { + const override = argv[index + 1]; + if (override === undefined || override.startsWith("-")) continue; + if (overrideHasCuaDriver(override)) return true; + index++; + } else if (argument?.startsWith("--config=") || argument?.startsWith("-c=")) { + if (overrideHasCuaDriver(argument.slice(argument.indexOf("=") + 1))) return true; + } else if (argument?.startsWith("-c") && argument.length > 2) { + if (overrideHasCuaDriver(argument.slice(2))) return true; + } + } + + if (configToml === undefined) return false; + try { + return parsedConfigHasCuaDriver(parseToml(configToml)); + } catch { + return true; + } +}; + +const tomlString = (value: string): string => JSON.stringify(value).replace(/\u007f/g, "\\u007f"); + +/** Returns argv directly; the inline table keeps environment keys out of Codex's dotted-key path. */ +export const buildCuaDriverAppServerArgs = (descriptor: { + readonly command: string; + readonly args: readonly string[]; + readonly environment: readonly { readonly name: string; readonly value: string }[]; +}): readonly string[] => { + const args = descriptor.args.map(tomlString).join(", "); + const environment = descriptor.environment + .map(({ name, value }) => `${tomlString(name)} = ${tomlString(value)}`) + .join(", "); + return [ + "-c", + `mcp_servers.cua-driver = { command = ${tomlString(descriptor.command)}, args = [${args}], env = { ${environment} } }`, + ]; +}; diff --git a/apps/server/src/cua/resolveCodexCua.test.ts b/apps/server/src/cua/resolveCodexCua.test.ts new file mode 100644 index 000000000000..56f9735aed53 --- /dev/null +++ b/apps/server/src/cua/resolveCodexCua.test.ts @@ -0,0 +1,114 @@ +// @effect-diagnostics nodeBuiltinImport:off +import * as NodeAssert from "node:assert/strict"; +import * as NodePath from "node:path"; +import { it } from "@effect/vitest"; +import * as Effect from "effect/Effect"; +import * as FileSystem from "effect/FileSystem"; +import * as Option from "effect/Option"; +import * as NodePathLayer from "@effect/platform-node/NodePath"; +import { parse } from "smol-toml"; + +import { resolveCodexCua } from "./resolveCodexCua.ts"; + +const descriptor = { command: "cua-driver", args: ["mcp", "--proxy"], environment: [] }; + +it.effect("does no config I/O or host startup while disabled", () => + resolveCodexCua( + { enabled: Effect.succeed(false), acquire: Effect.die("unexpected host startup") }, + { cwd: "/project", homePath: "/codex-home", launchArgs: "" }, + ).pipe( + Effect.provide(NodePathLayer.layer), + Effect.map((args) => NodeAssert.deepEqual(args, [])), + Effect.provide(FileSystem.layerNoop({ readFileString: () => Effect.die("unexpected I/O") })), + ), +); + +it.effect("explicit user launch configuration avoids reading files or starting a host", () => + resolveCodexCua( + { enabled: Effect.succeed(true), acquire: Effect.die("unexpected host startup") }, + { + cwd: "/project", + homePath: "/codex-home", + launchArgs: "-c 'mcp_servers.cua-driver.command=\"custom\"'", + }, + ).pipe( + Effect.provide(NodePathLayer.layer), + Effect.map((args) => NodeAssert.deepEqual(args, [])), + Effect.provide(FileSystem.layerNoop({ readFileString: () => Effect.die("unexpected I/O") })), + ), +); + +for (const location of ["home", "project", "parent", "malformed"] as const) { + it.effect(`preserves ${location} Codex configuration without acquiring managed Cua`, () => { + const home = NodePath.resolve("/custom-codex-home"); + const project = NodePath.resolve("/workspace/project"); + const selectedPath = + location === "home" || location === "malformed" + ? NodePath.join(home, "config.toml") + : NodePath.join( + location === "project" ? project : NodePath.dirname(project), + ".codex", + "config.toml", + ); + return resolveCodexCua( + { enabled: Effect.succeed(true), acquire: Effect.die("unexpected host startup") }, + { cwd: project, homePath: home, launchArgs: "", environment: { CODEX_HOME: "/wrong-home" } }, + ).pipe( + Effect.provide(NodePathLayer.layer), + Effect.map((args) => NodeAssert.deepEqual(args, [])), + Effect.provide( + FileSystem.layerNoop({ + readFileString: (path) => + Effect.succeed( + path === selectedPath + ? location === "malformed" + ? "not valid TOML" + : "[mcp_servers.cua-driver]\nenabled = false" + : "", + ), + }), + ), + ); + }); +} + +it.effect("acquires once after reading the instance home and emits structured argv", () => { + const paths: string[] = []; + let acquired = 0; + return resolveCodexCua( + { + enabled: Effect.succeed(true), + acquire: Effect.sync(() => { + acquired++; + return Option.some(descriptor); + }), + }, + { + cwd: "/workspace/project", + homePath: "", + environment: { CODEX_HOME: "/instance-home" }, + launchArgs: "", + }, + ).pipe( + Effect.provide(NodePathLayer.layer), + Effect.map((args) => { + NodeAssert.equal(acquired, 1); + NodeAssert.equal(paths[0], NodePath.join("/instance-home", "config.toml")); + NodeAssert.equal(args[0], "-c"); + NodeAssert.deepEqual(parse(args[1]!), { + mcp_servers: { + "cua-driver": { command: descriptor.command, args: descriptor.args, env: {} }, + }, + }); + }), + Effect.provide( + FileSystem.layerNoop({ + readFileString: (path) => + Effect.sync(() => { + paths.push(path); + return ""; + }), + }), + ), + ); +}); diff --git a/apps/server/src/cua/resolveCodexCua.ts b/apps/server/src/cua/resolveCodexCua.ts new file mode 100644 index 000000000000..d57089a84b8d --- /dev/null +++ b/apps/server/src/cua/resolveCodexCua.ts @@ -0,0 +1,54 @@ +import { tokenizeCliArgs } from "@t3tools/shared/cliArgs"; +import * as Effect from "effect/Effect"; +import * as FileSystem from "effect/FileSystem"; +import * as Option from "effect/Option"; +import * as Path from "effect/Path"; +import * as Result from "effect/Result"; + +import { expandHomePath } from "../pathExpansion.ts"; +import type { CuaDriver } from "./CuaDriver.ts"; +import { buildCuaDriverAppServerArgs, hasConfiguredCuaDriver } from "./codexCuaConfiguration.ts"; + +/** Only interactive sessions acquire managed Cua; existing host/project configuration wins. */ +export const resolveCodexCua = Effect.fn("cua.resolveCodexCua")(function* ( + driver: CuaDriver["Service"], + input: { + readonly cwd: string; + readonly homePath: string; + readonly launchArgs: string; + readonly environment?: NodeJS.ProcessEnv; + }, +) { + if (!(yield* driver.enabled)) return []; + const argv = tokenizeCliArgs(input.launchArgs); + if (hasConfiguredCuaDriver(argv, undefined)) return []; + + const fs = yield* FileSystem.FileSystem; + const pathService = yield* Path.Path; + const home = expandHomePath( + input.homePath || (input.environment ?? process.env).CODEX_HOME?.trim() || "~/.codex", + ); + const paths = new Set([pathService.join(home, "config.toml")]); + // Codex can inherit project configuration from ancestors of the session cwd. + for ( + let directory = pathService.resolve(input.cwd); + ; + directory = pathService.dirname(directory) + ) { + paths.add(pathService.join(directory, ".codex", "config.toml")); + if (directory === pathService.dirname(directory)) break; + } + for (const path of paths) { + const config = yield* fs.readFileString(path).pipe(Effect.result); + if (Result.isFailure(config)) { + if (config.failure.reason._tag === "NotFound") continue; + yield* Effect.logWarning("Could not read Codex configuration; withholding managed Cua.", { + cause: config.failure, + }); + return []; + } + if (hasConfiguredCuaDriver([], config.success)) return []; + } + const descriptor = yield* driver.acquire; + return Option.isSome(descriptor) ? buildCuaDriverAppServerArgs(descriptor.value) : []; +}); diff --git a/apps/server/src/provider/Layers/CodexAdapter.test.ts b/apps/server/src/provider/Layers/CodexAdapter.test.ts index f7c6036885d9..e042eada4ddb 100644 --- a/apps/server/src/provider/Layers/CodexAdapter.test.ts +++ b/apps/server/src/provider/Layers/CodexAdapter.test.ts @@ -26,6 +26,7 @@ import * as Context from "effect/Context"; 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 Queue from "effect/Queue"; @@ -37,6 +38,10 @@ import * as CodexErrors from "effect-codex-app-server/errors"; import { ServerConfig } from "../../config.ts"; import { ServerSettingsService } from "../../serverSettings.ts"; +import { CuaDriver } from "../../cua/CuaDriver.ts"; +import * as McpProviderSession from "../../mcp/McpProviderSession.ts"; +import { EnvironmentId } from "@t3tools/contracts"; +import { parse as parseToml } from "smol-toml"; import { ProviderAdapterValidationError } from "../Errors.ts"; import type { CodexAdapterShape } from "../Services/CodexAdapter.ts"; import { ProviderSessionDirectory } from "../Services/ProviderSessionDirectory.ts"; @@ -511,6 +516,80 @@ sessionErrorLayer("CodexAdapterLive session errors", (it) => { }).pipe(Effect.provide(layer)); }); + for (const browserEnabled of [false, true]) { + it.effect(`attaches managed Cua without changing browser access (${browserEnabled})`, () => { + const runtimeFactory = makeRuntimeFactory(); + const threadId = asThreadId(`cua-session-${browserEnabled}`); + const descriptor = { + command: "managed-cua-driver", + args: ["mcp", "--proxy"], + environment: [{ name: "CUA_SOCKET_PATH", value: "/synthetic/socket" }], + }; + const layer = Layer.effect( + CodexAdapter, + makeCodexAdapter(decodeCodexSettings({ homePath: "/synthetic/codex-home" }), { + environment: {}, + makeRuntime: runtimeFactory.factory, + }).pipe(Effect.provide(FileSystem.layerNoop({ readFileString: () => Effect.succeed("") }))), + ).pipe( + Layer.provideMerge( + Layer.succeed(CuaDriver, { + enabled: Effect.succeed(true), + acquire: Effect.succeed(Option.some(descriptor)), + }), + ), + Layer.provideMerge(ServerConfig.layerTest(process.cwd(), process.cwd())), + Layer.provideMerge(ServerSettingsService.layerTest()), + Layer.provideMerge(providerSessionDirectoryTestLayer), + Layer.provideMerge(NodeServices.layer), + ); + return Effect.scoped( + Effect.gen(function* () { + yield* Effect.addFinalizer(() => + Effect.sync(() => McpProviderSession.clearMcpProviderSession(threadId)), + ); + if (browserEnabled) { + McpProviderSession.setMcpProviderSession({ + environmentId: EnvironmentId.make("cua-test-environment"), + threadId, + providerInstanceId: ProviderInstanceId.make("codex"), + providerSessionId: "cua-test-provider-session", + endpoint: "http://127.0.0.1:1234/mcp", + authorizationHeader: "Bearer synthetic-test-token", + }); + } + const adapter = yield* CodexAdapter; + yield* adapter.startSession({ + provider: ProviderDriverKind.make("codex"), + threadId, + runtimeMode: "full-access", + }); + const runtime = runtimeFactory.lastRuntime; + NodeAssert.ok(runtime); + const args = runtime.options.appServerArgs ?? []; + NodeAssert.deepEqual(parseToml(args[1]!), { + mcp_servers: { + "cua-driver": { + command: descriptor.command, + args: descriptor.args, + env: { CUA_SOCKET_PATH: "/synthetic/socket" }, + }, + }, + }); + NodeAssert.equal( + args.some((arg) => arg.startsWith("mcp_servers.t3-code.")), + browserEnabled, + ); + NodeAssert.equal(runtime.options.launchArgs, ""); + NodeAssert.equal( + runtime.options.environment?.T3_MCP_BEARER_TOKEN, + browserEnabled ? "synthetic-test-token" : undefined, + ); + }), + ).pipe(Effect.provide(layer)); + }); + } + it.effect("maps codex model options for the adapter's bound custom instance id", () => { const customInstanceId = ProviderInstanceId.make("codex_personal"); const customRuntimeFactory = makeRuntimeFactory(); diff --git a/apps/server/src/provider/Layers/CodexAdapter.ts b/apps/server/src/provider/Layers/CodexAdapter.ts index 2d88e58dc1fb..5d444c1e50bd 100644 --- a/apps/server/src/provider/Layers/CodexAdapter.ts +++ b/apps/server/src/provider/Layers/CodexAdapter.ts @@ -38,6 +38,8 @@ import * as Exit from "effect/Exit"; import * as Fiber from "effect/Fiber"; import * as FileSystem from "effect/FileSystem"; import * as Queue from "effect/Queue"; +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"; @@ -48,6 +50,8 @@ import * as EffectCodexSchema from "effect-codex-app-server/schema"; import { getModelSelectionStringOptionValue } from "@t3tools/shared/model"; import { getCodexServiceTierOptionValue } from "../../codexModelOptions.ts"; import * as McpProviderSession from "../../mcp/McpProviderSession.ts"; +import { CuaDriver } from "../../cua/CuaDriver.ts"; +import { resolveCodexCua } from "../../cua/resolveCodexCua.ts"; import { ProviderAdapterRequestError, @@ -2219,6 +2223,8 @@ export const makeCodexAdapter = Effect.fn("makeCodexAdapter")(function* ( ) { const boundInstanceId = options?.instanceId ?? ProviderInstanceId.make("codex"); const fileSystem = yield* FileSystem.FileSystem; + const path = yield* Path.Path; + const cuaDriver = yield* Effect.serviceOption(CuaDriver); const childProcessSpawner = yield* ChildProcessSpawner.ChildProcessSpawner; const crypto = yield* Crypto.Crypto; const serverConfig = yield* Effect.service(ServerConfig); @@ -2255,12 +2261,25 @@ export const makeCodexAdapter = Effect.fn("makeCodexAdapter")(function* ( ? getCodexServiceTierOptionValue(input.modelSelection) : undefined; const mcpSession = McpProviderSession.readMcpProviderSession(input.threadId); + const launchArgs = resolveCodexLaunchArgs(codexConfig.launchArgs, options?.environment); + const cuaArgs = Option.isSome(cuaDriver) + ? yield* resolveCodexCua(cuaDriver.value, { + cwd: input.cwd ?? process.cwd(), + homePath: codexConfig.homePath, + launchArgs, + ...(options?.environment ? { environment: options.environment } : {}), + }).pipe( + Effect.provideService(FileSystem.FileSystem, fileSystem), + Effect.provideService(Path.Path, path), + ) + : []; const runtimeInput: CodexSessionRuntimeOptions = { threadId: input.threadId, providerInstanceId: boundInstanceId, cwd: input.cwd ?? process.cwd(), binaryPath: codexConfig.binaryPath, - launchArgs: resolveCodexLaunchArgs(codexConfig.launchArgs, options?.environment), + launchArgs, + ...(cuaArgs.length > 0 ? { appServerArgs: cuaArgs } : {}), ...(options?.environment ? { environment: options.environment } : {}), ...(codexConfig.homePath ? { homePath: codexConfig.homePath } : {}), ...(isCodexResumeCursorSchema(input.resumeCursor) @@ -2278,6 +2297,7 @@ export const makeCodexAdapter = Effect.fn("makeCodexAdapter")(function* ( T3_MCP_BEARER_TOKEN: mcpSession.authorizationHeader.replace(/^Bearer\s+/, ""), }, appServerArgs: [ + ...cuaArgs, "-c", `mcp_servers.t3-code.url=${mcpSession.endpoint}`, "-c", diff --git a/apps/server/src/provider/Layers/CodexSessionRuntime.test.ts b/apps/server/src/provider/Layers/CodexSessionRuntime.test.ts index 3385137a2dae..eb530dd71d2d 100644 --- a/apps/server/src/provider/Layers/CodexSessionRuntime.test.ts +++ b/apps/server/src/provider/Layers/CodexSessionRuntime.test.ts @@ -15,6 +15,7 @@ import { buildTurnStartParams, describeMcpElicitation, hasConfiguredMcpServer, + hasConfiguredBrowserMcpServer, isRecoverableThreadResumeError, makeMemoryConsolidationNotificationFilter, openCodexThread, @@ -554,6 +555,20 @@ describe("hasConfiguredMcpServer", () => { true, ); }); + it("refreshes Cua tools without advertising T3 preview tools", () => { + const cua = ["-c", 'mcp_servers.cua-driver = { command = "driver" }']; + NodeAssert.equal(hasConfiguredMcpServer(cua), true); + NodeAssert.equal(hasConfiguredBrowserMcpServer(cua), false); + NodeAssert.equal(hasConfiguredBrowserMcpServer(undefined), false); + NodeAssert.equal( + hasConfiguredBrowserMcpServer([ + ...cua, + "-c", + 'mcp_servers.t3-code.url="http://127.0.0.1/mcp"', + ]), + true, + ); + }); }); function makeThreadStartedNotification( diff --git a/apps/server/src/provider/Layers/CodexSessionRuntime.ts b/apps/server/src/provider/Layers/CodexSessionRuntime.ts index 78f4b9aa8e50..733c23d32802 100644 --- a/apps/server/src/provider/Layers/CodexSessionRuntime.ts +++ b/apps/server/src/provider/Layers/CodexSessionRuntime.ts @@ -66,6 +66,13 @@ export function hasConfiguredMcpServer(appServerArgs: ReadonlyArray | un return appServerArgs?.some((argument) => argument.includes("mcp_servers.")) === true; } +export function hasConfiguredBrowserMcpServer( + appServerArgs: ReadonlyArray | undefined, +): boolean { + // The adapter emits this exact path for T3's preview tools. Other MCPs do not provide them. + return appServerArgs?.some((argument) => argument.startsWith("mcp_servers.t3-code.")) === true; +} + export const CodexResumeCursorSchema = Schema.Struct({ threadId: Schema.String, }); @@ -2352,7 +2359,7 @@ export const makeCodexSessionRuntime = ( // Derived from the session's own MCP configuration rather than the // setting, so the prompt describes the tools this turn actually // has even if the setting changed after the session started. - browserToolsAvailable: hasConfiguredMcpServer(options.appServerArgs), + browserToolsAvailable: hasConfiguredBrowserMcpServer(options.appServerArgs), }); const rawResponse = yield* client.raw.request("turn/start", params); const response = yield* decodeV2TurnStartResponse(rawResponse).pipe( diff --git a/apps/server/src/resourceTelemetry/DesktopTelemetryReceiver.ts b/apps/server/src/resourceTelemetry/DesktopTelemetryReceiver.ts index 67842addd7df..1de947d077bb 100644 --- a/apps/server/src/resourceTelemetry/DesktopTelemetryReceiver.ts +++ b/apps/server/src/resourceTelemetry/DesktopTelemetryReceiver.ts @@ -4,6 +4,7 @@ import * as NodeFS from "node:fs"; import * as NodeStream from "@effect/platform-node/NodeStream"; import { DesktopHostTelemetryMessage, + type DesktopCuaDriverReport, type DesktopHostTelemetryMessage as DesktopHostTelemetryMessageValue, type DesktopHostTelemetrySnapshot, DesktopTelemetryControlMessage, @@ -169,6 +170,11 @@ export class DesktopTelemetryReceiver extends Context.Service< never, Scope.Scope >; + readonly cuaReports: Stream.Stream; + readonly requestCuaDriver: ( + requestId: string, + enabled: boolean, + ) => Effect.Effect; readonly setDiagnosticsDemand: ( enabled: boolean, ) => Effect.Effect; @@ -348,6 +354,7 @@ export const make = Effect.fn("resourceTelemetry.desktopTelemetryReceiver.make") const changes = yield* PubSub.sliding(8); const healthChanges = yield* PubSub.sliding(4); const latestUpdateReport = yield* Ref.make(Option.none()); + const cuaReports = yield* PubSub.unbounded(); const updateReportChanges = yield* PubSub.sliding(16); const controlMutex = yield* Semaphore.make(1); const snapshotMutex = yield* Semaphore.make(1); @@ -528,6 +535,13 @@ export const make = Effect.fn("resourceTelemetry.desktopTelemetryReceiver.make") ); } + if (message.type === "cuaDriverReport") { + return recordContact.pipe( + Effect.andThen(PubSub.publish(cuaReports, message)), + Effect.asVoid, + ); + } + // Not a resource sample: do not touch `latest` or sample health. if (message.type === "desktopUpdateStatus") { return recordContact.pipe( @@ -645,6 +659,9 @@ export const make = Effect.fn("resourceTelemetry.desktopTelemetryReceiver.make") ), health: Ref.get(health), subscribeHealth: subscribeBeforeSnapshotWithoutMutex(healthChanges, Ref.get(health)), + cuaReports: Stream.fromPubSub(cuaReports), + requestCuaDriver: (requestId, enabled) => + sendControlMessage({ version: 1, type: "cuaDriverRequest", requestId, enabled }), setDiagnosticsDemand, requestDesktopUpdate: (requestId) => sendControlMessage({ @@ -703,6 +720,8 @@ export const layerTest = ( changes: Stream.empty, })), ), + cuaReports: Stream.empty, + requestCuaDriver: () => Effect.void, setDiagnosticsDemand: () => Effect.void, requestDesktopUpdate: () => Effect.void, commitDesktopUpdate: () => Effect.void, diff --git a/apps/server/src/server.ts b/apps/server/src/server.ts index fd8ee4a4f699..42da0567d62a 100644 --- a/apps/server/src/server.ts +++ b/apps/server/src/server.ts @@ -120,6 +120,7 @@ import * as HostResources from "./resourceTelemetry/HostResources.ts"; import * as ProcessResourceMonitor from "./diagnostics/ProcessResourceMonitor.ts"; import * as TraceDiagnostics from "./diagnostics/TraceDiagnostics.ts"; import * as DesktopTelemetryReceiver from "./resourceTelemetry/DesktopTelemetryReceiver.ts"; +import * as CuaDriver from "./cua/CuaDriver.ts"; import * as NativeTelemetryClient from "./resourceTelemetry/NativeTelemetryClient.ts"; import * as ResourceAttribution from "./resourceTelemetry/ResourceAttribution.ts"; import * as ResourceMonitorBinary from "./resourceTelemetry/ResourceMonitorBinary.ts"; @@ -178,6 +179,11 @@ const DesktopTelemetryReceiverLayerLive = DesktopTelemetryReceiver.layer.pipe( Layer.provideMerge(ServerSettingsLayerLive), ); +const CuaDriverLayerLive = CuaDriver.layer.pipe( + Layer.provide(ServerSettingsLayerLive), + Layer.provide(DesktopTelemetryReceiverLayerLive), +); + const ResourceTelemetryLayerLive = ResourceTelemetry.layer.pipe( Layer.provideMerge(NativeTelemetryLayerLive), Layer.provideMerge(DesktopTelemetryReceiverLayerLive), @@ -481,6 +487,7 @@ const RuntimeCoreDependenciesLive = ReactorLayerLive.pipe( Layer.provideMerge(ProviderInstanceRegistryHydrationLive), ).pipe( Layer.provideMerge(AntigravityInstallation.layer), + Layer.provideMerge(CuaDriverLayerLive), // Shared native/canonical NDJSON writers used by both the per-instance // drivers (native stream, written from inside each `Adapter`) and // `ProviderService` (canonical stream, written after event normalization). diff --git a/apps/web/src/components/settings/ProjectDefaultsSettings.tsx b/apps/web/src/components/settings/ProjectDefaultsSettings.tsx index 938000e01002..09bc67800734 100644 --- a/apps/web/src/components/settings/ProjectDefaultsSettings.tsx +++ b/apps/web/src/components/settings/ProjectDefaultsSettings.tsx @@ -96,6 +96,9 @@ export function ProjectDefaultsSettings({ target.serverConfig?.settings.enableAgentBrowserAccess !== serverSettings.enableAgentBrowserAccess, ); + const mixedCua = targets.some( + (target) => target.serverConfig?.settings.enableCua !== serverSettings.enableCua, + ); const disabled = (key: keyof ServerSettingsPatch) => targets.length === 0 || saving.has(key); const mixedAutoPull = targets.some( (target) => target.serverConfig?.settings.defaultAutoPull !== serverSettings.defaultAutoPull, @@ -388,6 +391,47 @@ export function ProjectDefaultsSettings({ } /> + void save({ enableCua: DEFAULT_SERVER_SETTINGS.enableCua })} + /> + ) : null + } + control={ + + } + /> void) { ...(settings.enableAgentBrowserAccess !== DEFAULT_UNIFIED_SETTINGS.enableAgentBrowserAccess ? ["Agent browser access"] : []), + ...(settings.enableCua !== DEFAULT_UNIFIED_SETTINGS.enableCua ? ["Cua computer use"] : []), ], [ isTextGenerationModelDirty, @@ -598,6 +599,7 @@ export function useSettingsRestore(onRestored?: () => void) { settings.browserAutoShowFloatingPreview, settings.appearanceContrast, settings.enableAgentBrowserAccess, + settings.enableCua, settings.confirmQuit, settings.confirmThreadArchive, settings.confirmThreadDelete, @@ -749,6 +751,7 @@ export function useSettingsRestore(onRestored?: () => void) { // name, so a user restoring defaults is told the agent regains access // rather than discovering it later. enableAgentBrowserAccess: DEFAULT_UNIFIED_SETTINGS.enableAgentBrowserAccess, + enableCua: DEFAULT_UNIFIED_SETTINGS.enableCua, }); onRestored?.(); }, [ diff --git a/apps/web/src/components/settings/settingsSearch.ts b/apps/web/src/components/settings/settingsSearch.ts index e32baf6f3987..97e18d469d90 100644 --- a/apps/web/src/components/settings/settingsSearch.ts +++ b/apps/web/src/components/settings/settingsSearch.ts @@ -396,6 +396,14 @@ export const SETTINGS_SEARCH_ITEMS = [ to: "/settings/projects", searchTerms: ["allow open drive preview tools sessions"], }, + { + id: "cua-computer-use", + title: "Cua computer use", + to: "/settings/projects", + searchTerms: [ + "Codex Cua Driver computer machine host permissions accessibility screen capture", + ], + }, { id: "browser-profiles", title: "Browser profiles", diff --git a/docs/user/providers-codex.md b/docs/user/providers-codex.md index 06c59c6f6aee..c97dec3eea7a 100644 --- a/docs/user/providers-codex.md +++ b/docs/user/providers-codex.md @@ -65,6 +65,27 @@ in the thread on web, desktop, or mobile. Some tools offer access for one reques the current session, or permanently. See [Permission modes](./permission-modes.md) for command and file approvals. +## Let Codex use the host computer + +**Cua computer use** lets Codex control the machine running your T3 Code server +through Cua Driver. It is off by default and separate from **Agent browser access**. + +Complete setup on the host machine using T3 Code desktop or web: + +1. Use the Cua Driver included with packaged T3 Code desktop. For a standalone + server, set `T3CODE_CUA_DRIVER_PATH` to the absolute path of your Cua Driver + executable before starting the server. +2. Grant the required operating-system permissions on that host. Connecting from + another computer or phone does not grant permission to control the host. +3. Open **Settings > Projects**, select the host machine, and enable + **Cua computer use** in its defaults. Start a new Codex session to use it. + +Once the host is configured, you can direct its Codex sessions from web, desktop, +or mobile. The setting applies to the selected environment, with no per-project +override. Disabling it revokes T3 Code's managed Cua access. It does not remove +MCP servers you configured yourself. Other providers do not receive this managed +integration. + ## Codex says I hit a usage limit When Codex stops on a usage limit, the thread names the window that ran out and diff --git a/packages/contracts/src/cua.ts b/packages/contracts/src/cua.ts new file mode 100644 index 000000000000..6dd1c7970917 --- /dev/null +++ b/packages/contracts/src/cua.ts @@ -0,0 +1,37 @@ +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 })), +}); +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 74a1b4939f1a..5b1c802f4167 100644 --- a/packages/contracts/src/index.ts +++ b/packages/contracts/src/index.ts @@ -39,3 +39,4 @@ export * from "./previewAutomation.ts"; export * from "./resourceTelemetry.ts"; export * from "./usage.ts"; export * from "./rpc.ts"; +export * from "./cua.ts"; diff --git a/packages/contracts/src/resourceTelemetry.ts b/packages/contracts/src/resourceTelemetry.ts index ee9d2b3ac258..b38b38c725aa 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; @@ -307,6 +308,7 @@ export const DesktopUpdateStatusReport = Schema.Struct({ export type DesktopUpdateStatusReport = typeof DesktopUpdateStatusReport.Type; export const DesktopHostTelemetryMessage = Schema.Union([ + DesktopCuaDriverReport, DesktopHostTelemetryHello, DesktopHostTelemetrySnapshot, DesktopUpdateStatusReport, @@ -356,6 +358,7 @@ export const DesktopTelemetryCancelDesktopUpdate = Schema.Struct({ export type DesktopTelemetryCancelDesktopUpdate = typeof DesktopTelemetryCancelDesktopUpdate.Type; export const DesktopTelemetryControlMessage = Schema.Union([ + DesktopCuaDriverRequest, DesktopTelemetrySetDiagnosticsDemand, DesktopTelemetrySetHostPowerIntervals, DesktopTelemetryRequestDesktopUpdate, diff --git a/packages/contracts/src/settings.test.ts b/packages/contracts/src/settings.test.ts index 7d3cd2ceafe7..d27707167bff 100644 --- a/packages/contracts/src/settings.test.ts +++ b/packages/contracts/src/settings.test.ts @@ -20,6 +20,34 @@ const decodeServerSettingsPatch = Schema.decodeUnknownSync(ServerSettingsPatch); const encodeServerSettings = Schema.encodeSync(ServerSettings); const decodeClaudeSettings = Schema.decodeUnknownSync(ClaudeSettings); +describe("ServerSettings Cua computer use", () => { + it("keeps legacy settings opted out independently of browser access", () => { + expect(DEFAULT_SERVER_SETTINGS.enableCua).toBe(false); + expect(decodeServerSettings({}).enableCua).toBe(false); + expect(decodeServerSettings({ enableAgentBrowserAccess: true }).enableCua).toBe(false); + }); + + it.each([true, false])("round-trips an explicit opt-in value of %s", (enableCua) => { + const settings = decodeServerSettings({ enableCua, enableAgentBrowserAccess: false }); + expect(settings.enableCua).toBe(enableCua); + expect(settings.enableAgentBrowserAccess).toBe(false); + expect(encodeServerSettings(settings).enableCua).toBe(enableCua); + expect(decodeServerSettingsPatch({ enableCua })).toEqual({ enableCua }); + }); + + it("does not introduce a Cua change into unrelated patches", () => { + expect(decodeServerSettingsPatch({})).not.toHaveProperty("enableCua"); + expect(decodeServerSettingsPatch({ enableAgentBrowserAccess: true })).toEqual({ + enableAgentBrowserAccess: true, + }); + }); + + it.each(["true", 1, null])("rejects a non-boolean opt-in of %s", (enableCua) => { + expect(() => decodeServerSettings({ enableCua })).toThrow(); + expect(() => decodeServerSettingsPatch({ enableCua })).toThrow(); + }); +}); + describe("ServerSettings usage price overrides", () => { const prices = { inputCostPerMillionTokens: 2, outputCostPerMillionTokens: 8 }; diff --git a/packages/contracts/src/settings.ts b/packages/contracts/src/settings.ts index 3491103da94f..4e2f333e0bed 100644 --- a/packages/contracts/src/settings.ts +++ b/packages/contracts/src/settings.ts @@ -946,6 +946,8 @@ export const ServerSettings = Schema.Struct({ * between a desktop window and a phone attached to the same server. */ enableAgentBrowserAccess: Schema.Boolean.pipe(Schema.withDecodingDefault(Effect.succeed(true))), + /** Environment-owned opt-in for managed Codex access to Cua Driver. */ + enableCua: Schema.Boolean.pipe(Schema.withDecodingDefault(Effect.succeed(false))), projectAgentBrowserAccessOverrides: Schema.Record(ProjectId, Schema.Boolean).pipe( Schema.withDecodingDefault(Effect.succeed({})), ), @@ -1222,6 +1224,7 @@ export const ServerSettingsPatch = Schema.Struct({ enableProviderUpdateChecks: Schema.optionalKey(Schema.Boolean), continueThreadsAfterServerUpdate: Schema.optionalKey(Schema.Boolean), enableAgentBrowserAccess: Schema.optionalKey(Schema.Boolean), + enableCua: Schema.optionalKey(Schema.Boolean), projectAgentBrowserAccessOverrides: Schema.optionalKey( Schema.Record(ProjectId, Schema.NullOr(Schema.Boolean)), ), diff --git a/pnpm-lock.yaml b/pnpm-lock.yaml index 6c58c455b465..55efc6f7f13e 100644 --- a/pnpm-lock.yaml +++ b/pnpm-lock.yaml @@ -158,6 +158,9 @@ importers: '@t3tools/tailscale': specifier: workspace:* version: link:../../packages/tailscale + '@trycua/cua-driver': + specifier: 0.24.0 + version: 0.24.0 dbus-next: specifier: 0.10.2 version: 0.10.2(patch_hash=cfff57561b0ee59b5addb3b2e6c6f20906e967507a530ab67e8db8108e520ba4) @@ -509,6 +512,9 @@ importers: '@opencode-ai/sdk': specifier: ^1.3.15 version: 1.15.13 + '@trycua/cua-driver': + specifier: 0.24.0 + version: 0.24.0 effect: specifier: 4.0.0-rc.112 version: 4.0.0-rc.112(patch_hash=8bef799f35729cf3465eb428196f617b434a7c9f658d56acb2874d74b21b98b2) @@ -518,6 +524,9 @@ importers: node-pty: specifier: ^1.1.0 version: 1.1.0 + smol-toml: + specifier: 1.7.0 + version: 1.7.0 stream-chain: specifier: ^4.2.5 version: 4.2.5 @@ -5306,6 +5315,41 @@ packages: peerDependencies: '@testing-library/dom': '>=7.21.4' + '@trycua/cua-driver-darwin-arm64@0.24.0': + resolution: {integrity: sha512-V6C580ZFIYmB6fxjqM3eEoXahvzGjWc34I1SoVSBjDRjQSn2fXLldlNsCBxEUHs4TH/JopTyP1hTChbNtCC06w==} + cpu: [arm64] + os: [darwin] + + '@trycua/cua-driver-darwin-x64@0.24.0': + resolution: {integrity: sha512-4KmFTUsGegTXmt+Crr/M+Zf/92zI+tGmJWFfM3CkCNGaCBw+wqJFmAYp9mVaR2F+zmdpaWyptB/movaHC88Lcg==} + cpu: [x64] + os: [darwin] + + '@trycua/cua-driver-linux-arm64-gnu@0.24.0': + resolution: {integrity: sha512-SWiyMz8z6Oax/Q+Fp2zN4WjxcXE1pSQH9SsET63W2+C6t2CZdhLzypB1h4xNStcX8eCQZ/dl0ayiV2h0U2c/Zw==} + cpu: [arm64] + os: [linux] + libc: [glibc] + + '@trycua/cua-driver-linux-x64-gnu@0.24.0': + resolution: {integrity: sha512-7FDtyemjyaZwNIOahfYPr6NcJOr86i3WXGcaJ9Q+a5wG13srkr3vhA0K7mPQcGd/tZTyZAulXA0fnjYv7oZiXg==} + cpu: [x64] + os: [linux] + libc: [glibc] + + '@trycua/cua-driver-win32-arm64-msvc@0.24.0': + resolution: {integrity: sha512-NktveoX98qafCjgBePzqTsHqjlFHdz6ositvEtwVTS3xA7hNEgCsr9w55sRQYuUSGfz5xjxpFioSn6Mh/f5brw==} + cpu: [arm64] + os: [win32] + + '@trycua/cua-driver-win32-x64-msvc@0.24.0': + resolution: {integrity: sha512-8tudsgdnfgCgeavIyhIOVCAAkfKezTkhTaZvPqZnikYN+5xKrO4o+vPhyyuEurYZLBLKluAxRwGr9DP/7syZjw==} + cpu: [x64] + os: [win32] + + '@trycua/cua-driver@0.24.0': + resolution: {integrity: sha512-OVYuBBvo7HNxGmqZDaWlfQ5tdr7FruA0YfytJ0p+aTZ9QF/1T8G5nQNsLVJht7JBAybktxO6JW+lxBNn9QHp8A==} + '@tybys/wasm-util@0.10.3': resolution: {integrity: sha512-F3fo1MYrRJYL3zER0OUOmkutjr1Vp23m7OsSgp7nq4SP6OqX6C/56XFIPAl5bt3zaBRjmW7SGz3u/6LwFpYcOg==} @@ -5587,6 +5631,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==} @@ -11467,7 +11561,7 @@ snapshots: picomatch: 4.0.7 retext-smartypants: 6.2.0 shiki: 4.2.0 - smol-toml: 1.8.0 + smol-toml: 1.7.0 unified: 11.0.5 '@astrojs/language-server@2.16.10(prettier@3.8.3)(typescript@6.0.3)': @@ -15785,6 +15879,36 @@ snapshots: dependencies: '@testing-library/dom': 10.4.1 + '@trycua/cua-driver-darwin-arm64@0.24.0': + optional: true + + '@trycua/cua-driver-darwin-x64@0.24.0': + optional: true + + '@trycua/cua-driver-linux-arm64-gnu@0.24.0': + optional: true + + '@trycua/cua-driver-linux-x64-gnu@0.24.0': + optional: true + + '@trycua/cua-driver-win32-arm64-msvc@0.24.0': + optional: true + + '@trycua/cua-driver-win32-x64-msvc@0.24.0': + optional: true + + '@trycua/cua-driver@0.24.0': + dependencies: + '@ubjs/core': 0.31.0-3 + '@ubjs/node': 0.31.0-3 + optionalDependencies: + '@trycua/cua-driver-darwin-arm64': 0.24.0 + '@trycua/cua-driver-darwin-x64': 0.24.0 + '@trycua/cua-driver-linux-arm64-gnu': 0.24.0 + '@trycua/cua-driver-linux-x64-gnu': 0.24.0 + '@trycua/cua-driver-win32-arm64-msvc': 0.24.0 + '@trycua/cua-driver-win32-x64-msvc': 0.24.0 + '@tybys/wasm-util@0.10.3': dependencies: tslib: 2.8.1 @@ -16040,6 +16164,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': {} '@vercel/config@0.3.0': diff --git a/scripts/build-desktop-artifact.test.ts b/scripts/build-desktop-artifact.test.ts index b601d3997fd7..773bc52a5e23 100644 --- a/scripts/build-desktop-artifact.test.ts +++ b/scripts/build-desktop-artifact.test.ts @@ -15,6 +15,12 @@ import * as Stream from "effect/Stream"; import { ChildProcess, ChildProcessSpawner } from "effect/unstable/process"; import { + CUA_DRIVER_EXTRA_RESOURCE, + CuaDriverBundleMissingFileError, + CuaDriverChecksumMismatchError, + stageCuaDriverExecutable, + resolveCuaDriverAsset, + stageCuaDriverBundle, BundleNotSelfContainedError, BuildCommandFailedError, buildWslRuntimeArchiveArgs, @@ -258,6 +264,139 @@ const makeWindowsPayloadFixture = Effect.fn("test.makeWindowsPayloadFixture")(fu }); it.layer(NodeServices.layer)("build-desktop-artifact", (it) => { + it.effect("rejects corrupt release downloads before extraction or staging", () => + Effect.scoped( + Effect.gen(function* () { + const fs = yield* FileSystem.FileSystem; + const path = yield* Path.Path; + const root = yield* fs.makeTempDirectoryScoped({ prefix: "t3-cua-checksum-" }); + const commands: string[] = []; + const spawnerLayer = Layer.succeed( + ChildProcessSpawner.ChildProcessSpawner, + ChildProcessSpawner.make((command) => { + const child = command as unknown as { command: string; args: readonly string[] }; + commands.push(child.command); + const output = child.args[child.args.indexOf("--output") + 1]; + return child.command === "curl" && output !== undefined + ? Effect.as(fs.writeFileString(output, "corrupted archive"), mockProcess(0)) + : Effect.succeed(mockProcess(0)); + }), + ); + const error = yield* Effect.flip( + stageCuaDriverExecutable({ + platform: "linux", + arch: "x64", + repoRoot: root, + stageRoot: root, + stageResourcesDir: path.join(root, "resources"), + verbose: false, + }).pipe(Effect.provide(spawnerLayer)), + ); + assert.instanceOf(error, CuaDriverChecksumMismatchError); + assert.deepStrictEqual(commands, ["curl"]); + assert.isFalse(yield* fs.exists(path.join(root, "resources", "cua-driver"))); + const cache = path.join(root, "node_modules/.cache/t3code/cua-driver"); + assert.deepStrictEqual(yield* fs.readDirectory(cache), []); + }), + ), + ); + + it("rejects universal Cua Driver targets outside macOS", () => { + for (const platform of ["linux", "win"] as const) { + assert.throws(() => resolveCuaDriverAsset(platform, "universal"), /no release asset/); + } + }); + + for (const platform of ["mac", "linux", "win"] as const) { + for (const arch of [ + "arm64", + "x64", + ...(platform === "mac" ? ["universal" as const] : []), + ] as const) { + it.effect(`stages the ${platform}/${arch} release layout with its companions`, () => + Effect.scoped( + Effect.gen(function* () { + const fs = yield* FileSystem.FileSystem; + const path = yield* Path.Path; + const root = yield* fs.makeTempDirectoryScoped({ prefix: "t3-cua-bundle-" }); + const extractDir = path.join(root, "extract"); + const stageResourcesDir = path.join(root, "resources"); + // These are the actual release archive layouts, independent of the resolver. + const releaseArch = arch === "x64" ? "x86_64" : arch; + const executable = + platform === "mac" && arch !== "universal" + ? `cua-driver-rs-0.24.0-darwin-${releaseArch}/cua-driver` + : platform === "win" + ? "cua-driver.exe" + : "cua-driver"; + const companions = + platform === "win" + ? [ + "cua-driver-uia.exe", + "cua-cursor-theme.exe", + "cua_driver_sdk.dll", + "cua_driver_node_runtime.node", + ] + : platform === "linux" + ? [ + "cua-cursor-theme", + "libcua_driver_sdk.so", + "cua_driver_node_runtime.node", + "wayland-helper/winrects@cua/extension.js", + ] + : []; + for (const member of [executable, ...companions, "cua_driver_abi.h"]) { + const target = path.join(extractDir, member); + yield* fs.makeDirectory(path.dirname(target), { recursive: true }); + yield* fs.writeFileString(target, member); + } + const asset = resolveCuaDriverAsset(platform, arch); + assert.equal(asset.executablePath, executable); + assert.match(asset.sha256, /^[0-9a-f]{64}$/); + assert.include(asset.url, "/cua-driver-rs-v0.24.0/"); + yield* stageCuaDriverBundle({ platform, arch, extractDir, stageResourcesDir }); + const destination = path.join(stageResourcesDir, "cua-driver"); + if (platform === "mac") { + assert.equal(yield* fs.readFileString(destination), executable); + } else { + for (const member of [executable, ...companions, "cua_driver_abi.h"]) { + assert.equal(yield* fs.readFileString(path.join(destination, member)), member); + } + } + if ((yield* HostProcessPlatform) !== "win32" && platform !== "win") { + const binary = platform === "mac" ? destination : path.join(destination, executable); + assert.equal((yield* fs.stat(binary)).mode & 0o777, 0o755); + } + }), + ), + ); + } + } + + it.effect("rejects an incomplete Windows bundle before publishing a staged CLI", () => + Effect.scoped( + Effect.gen(function* () { + const fs = yield* FileSystem.FileSystem; + const path = yield* Path.Path; + const root = yield* fs.makeTempDirectoryScoped({ prefix: "t3-cua-incomplete-" }); + const extractDir = path.join(root, "extract"); + const stageResourcesDir = path.join(root, "resources"); + yield* fs.makeDirectory(extractDir); + yield* fs.writeFileString(path.join(extractDir, "cua-driver.exe"), "cli"); + const error = yield* Effect.flip( + stageCuaDriverBundle({ + platform: "win", + arch: "x64", + extractDir, + stageResourcesDir, + }), + ); + assert.instanceOf(error, CuaDriverBundleMissingFileError); + assert.isFalse(yield* fs.exists(path.join(stageResourcesDir, "cua-driver"))); + }), + ), + ); + it("resolves the dedicated nightly updater channel from nightly versions", () => { assert.equal(resolveDesktopUpdateChannel("0.0.17-nightly.20260413.42"), "nightly"); assert.equal(resolveDesktopUpdateChannel("0.0.17"), "latest"); @@ -559,6 +698,8 @@ it.layer(NodeServices.layer)("build-desktop-artifact", (it) => { assert.deepStrictEqual(DESKTOP_FILE_EXCLUSIONS, [ "!**/node_modules/@anthropic-ai/claude-agent-sdk-*/**/*", + "!apps/desktop/prod-resources/cua-driver", + "!apps/desktop/prod-resources/cua-driver/**/*", "!apps/desktop/resources/browser-secret", "!apps/desktop/resources/browser-secret/**/*", "!apps/desktop/prod-resources/browser-secret", @@ -638,6 +779,7 @@ it.layer(NodeServices.layer)("build-desktop-artifact", (it) => { { from: "apps/desktop/prod-resources/browser-secret", to: "browser-secret" }, ]); assert.deepStrictEqual(win.extraResources, [ + CUA_DRIVER_EXTRA_RESOURCE, { from: "apps/desktop/prod-resources/resource-monitor", to: "resource-monitor", @@ -648,6 +790,7 @@ it.layer(NodeServices.layer)("build-desktop-artifact", (it) => { // No Linux prebuild means the sidecar staging never writes the archive, // so listing it here would fail the build on a missing source file. assert.deepStrictEqual(winWithoutWslPrebuild.extraResources, [ + CUA_DRIVER_EXTRA_RESOURCE, { from: "apps/desktop/prod-resources/resource-monitor", to: "resource-monitor", @@ -1856,6 +1999,7 @@ it.layer(NodeServices.layer)("build-desktop-artifact", (it) => { it("stages the resource monitor as an external executable resource", () => { assert.deepStrictEqual(DESKTOP_EXTRA_RESOURCES, [ + CUA_DRIVER_EXTRA_RESOURCE, { from: "apps/desktop/prod-resources/resource-monitor", to: "resource-monitor", diff --git a/scripts/build-desktop-artifact.ts b/scripts/build-desktop-artifact.ts index 44d5f0ab12f4..ba60c03d497d 100644 --- a/scripts/build-desktop-artifact.ts +++ b/scripts/build-desktop-artifact.ts @@ -59,6 +59,44 @@ const APPLE_TEAM_ID_PATTERN = /^[A-Z0-9]{10}$/u; const BuildPlatform = Schema.Literals(["mac", "linux", "win"]); const BuildArch = Schema.Literals(["arm64", "x64", "universal"]); +const CUA_DRIVER_RELEASE_VERSION = "0.24.0"; +const CUA_DRIVER_RELEASE_BASE_URL = `https://github.com/trycua/cua/releases/download/cua-driver-rs-v${CUA_DRIVER_RELEASE_VERSION}`; + +interface CuaDriverMacAsset { + readonly archiveName: string; + readonly executablePath: string; + readonly sha256: string; +} + +const CUA_DRIVER_MAC_ASSETS: Record = { + arm64: { + archiveName: `cua-driver-rs-${CUA_DRIVER_RELEASE_VERSION}-darwin-arm64.tar.gz`, + executablePath: `cua-driver-rs-${CUA_DRIVER_RELEASE_VERSION}-darwin-arm64/cua-driver`, + sha256: "fd0cf565db831ad34d44a3c2321439575e02a6ce3ca97d04f267db1da7883685", + }, + x64: { + archiveName: `cua-driver-rs-${CUA_DRIVER_RELEASE_VERSION}-darwin-x86_64.tar.gz`, + executablePath: `cua-driver-rs-${CUA_DRIVER_RELEASE_VERSION}-darwin-x86_64/cua-driver`, + sha256: "66db9b244c12e0f416212ca3421afb7cda6b1b927fb6536943919bb7eff5e10b", + }, + universal: { + archiveName: `cua-driver-rs-${CUA_DRIVER_RELEASE_VERSION}-darwin-universal-binary.tar.gz`, + executablePath: "cua-driver", + sha256: "31790cb49baa206f6455fbc259f8f83ae27e86be908f5c8cac5ec2f8521f8382", + }, +}; + +const CUA_DRIVER_PLATFORM_SHA256 = { + linux: { + arm64: "2c526bf14fb81a46db19d242ddd2e0be766bc1ce1082ed8a708ad235de74a28e", + x64: "b3b8ff52595feb111219aa0ac90e3b36618cc686aa8205aed4f3e7e1a67c7b64", + }, + win: { + arm64: "d973df1cfa421ca7801015a2987bbe1526eef1ac167cd3f281b417bb282037e9", + x64: "cc22d7a44ad526f779f2df7e6da053dd898ef8e5014b1ecfc01728645f691be0", + }, +} as const; + const WorkspaceConfig = Schema.Struct({ catalog: Schema.optional(Schema.Record(Schema.String, Schema.String)), overrides: Schema.optional(Schema.Record(Schema.String, Schema.String)), @@ -151,6 +189,41 @@ const PLATFORM_CONFIG: Record = { }, }; +export function resolveCuaDriverMacAsset(arch: typeof BuildArch.Type) { + if (serverPackageJson.dependencies["@trycua/cua-driver"] !== CUA_DRIVER_RELEASE_VERSION) { + throw new Error("Cua Driver release assets must be updated to match the SDK version."); + } + const asset = CUA_DRIVER_MAC_ASSETS[arch]; + return { + ...asset, + url: `${CUA_DRIVER_RELEASE_BASE_URL}/${asset.archiveName}`, + }; +} + +export function resolveCuaDriverAsset( + platform: typeof BuildPlatform.Type, + arch: typeof BuildArch.Type, +) { + if (serverPackageJson.dependencies["@trycua/cua-driver"] !== CUA_DRIVER_RELEASE_VERSION) { + throw new Error("Cua Driver release assets must be updated to match the SDK version."); + } + if (platform === "mac") return resolveCuaDriverMacAsset(arch); + if (arch === "universal") { + throw new Error(`Cua Driver has no release asset for ${platform}/${arch}.`); + } + + const os = platform === "win" ? "windows" : "linux"; + const releaseArch = arch === "x64" ? "x86_64" : "arm64"; + const extension = platform === "win" ? "zip" : "tar.gz"; + const archiveName = `cua-driver-rs-${CUA_DRIVER_RELEASE_VERSION}-${os}-${releaseArch}-binary.${extension}`; + return { + archiveName, + executablePath: platform === "win" ? "cua-driver.exe" : "cua-driver", + sha256: CUA_DRIVER_PLATFORM_SHA256[platform][arch], + url: `${CUA_DRIVER_RELEASE_BASE_URL}/${archiveName}`, + }; +} + interface BuildCliInput { readonly platform: Option.Option; readonly target: Option.Option; @@ -284,6 +357,19 @@ export class InvalidMockUpdateServerPortError extends Schema.TaggedError()( + "CuaDriverChecksumMismatchError", + { + archiveName: Schema.String, + expected: Schema.String, + actual: Schema.String, + }, +) { + override get message(): string { + return `Cua Driver checksum mismatch for ${this.archiveName}.`; + } +} + export class BuildCommandFailedError extends Schema.TaggedError()( "BuildCommandFailedError", { @@ -959,6 +1045,8 @@ export const DESKTOP_FILE_EXCLUSIONS = [ // so the SDK's optional platform packages (each a ~200MB bundled executable) // are dead weight. The trailing dash keeps the SDK's own JS package. "!**/node_modules/@anthropic-ai/claude-agent-sdk-*/**/*", + "!apps/desktop/prod-resources/cua-driver", + "!apps/desktop/prod-resources/cua-driver/**/*", "!apps/desktop/resources/browser-secret", "!apps/desktop/resources/browser-secret/**/*", "!apps/desktop/prod-resources/browser-secret", @@ -1089,7 +1177,13 @@ export const WSL_RUNTIME_EXTRA_RESOURCES = [ WSL_RUNTIME_ARCHIVE_EXTRA_RESOURCE, WSL_RUNTIME_ARCHIVE_HASH_EXTRA_RESOURCE, ] as const; +export const CUA_DRIVER_EXTRA_RESOURCE = { + from: "apps/desktop/prod-resources/cua-driver", + to: "cua-driver", +} as const; + export const DESKTOP_EXTRA_RESOURCES = [ + CUA_DRIVER_EXTRA_RESOURCE, { from: "apps/desktop/prod-resources/resource-monitor", to: "resource-monitor", @@ -2801,6 +2895,141 @@ export const createBuildConfig = Effect.fn("createBuildConfig")(function* ( return buildConfig; }); +export class CuaDriverBundleMissingFileError extends Schema.TaggedError()( + "CuaDriverBundleMissingFileError", + { filePath: Schema.String }, +) { + override get message(): string { + return `Cua Driver release bundle is missing required file: ${this.filePath}`; + } +} + +export const stageCuaDriverBundle = Effect.fn("stageCuaDriverBundle")(function* (input: { + readonly platform: typeof BuildPlatform.Type; + readonly arch: typeof BuildArch.Type; + readonly extractDir: string; + readonly stageResourcesDir: string; +}) { + const fs = yield* FileSystem.FileSystem; + const path = yield* Path.Path; + const asset = resolveCuaDriverAsset(input.platform, input.arch); + const requiredFiles = + input.platform === "mac" + ? [asset.executablePath] + : input.platform === "win" + ? [ + "cua-driver.exe", + "cua-driver-uia.exe", + "cua-cursor-theme.exe", + "cua_driver_sdk.dll", + "cua_driver_node_runtime.node", + ] + : [ + "cua-driver", + "cua-cursor-theme", + "libcua_driver_sdk.so", + "cua_driver_node_runtime.node", + "wayland-helper/winrects@cua/extension.js", + ]; + for (const member of requiredFiles) { + const filePath = path.join(input.extractDir, member); + if (!(yield* fs.exists(filePath)) || (yield* fs.stat(filePath)).type !== "File") { + return yield* new CuaDriverBundleMissingFileError({ filePath }); + } + } + + yield* fs.makeDirectory(input.stageResourcesDir, { recursive: true }); + const destination = path.join(input.stageResourcesDir, "cua-driver"); + if (input.platform === "mac") { + // osx-sign discovers this Mach-O alongside the other external resources. + yield* fs.copyFile(path.join(input.extractDir, asset.executablePath), destination); + yield* fs.chmod(destination, 0o755); + } else { + yield* fs.copy(input.extractDir, destination); + if (input.platform === "linux") { + yield* fs.chmod(path.join(destination, "cua-driver"), 0o755); + yield* fs.chmod(path.join(destination, "cua-cursor-theme"), 0o755); + } + } +}); + +export const stageCuaDriverExecutable = Effect.fn("stageCuaDriverExecutable")(function* (input: { + readonly platform: typeof BuildPlatform.Type; + readonly arch: typeof BuildArch.Type; + readonly repoRoot: string; + readonly stageRoot: string; + readonly stageResourcesDir: string; + readonly verbose: boolean; +}) { + const fs = yield* FileSystem.FileSystem; + const path = yield* Path.Path; + const asset = resolveCuaDriverAsset(input.platform, input.arch); + const cacheDir = path.join(input.repoRoot, "node_modules/.cache/t3code/cua-driver"); + const archivePath = path.join(cacheDir, asset.archiveName); + const extractDir = path.join(input.stageRoot, "cua-driver/extract"); + yield* fs.makeDirectory(cacheDir, { recursive: true }); + yield* fs.makeDirectory(extractDir, { recursive: true }); + + const checksum = (filePath: string) => + fs + .readFile(filePath) + .pipe( + Effect.map((contents) => NodeCrypto.createHash("sha256").update(contents).digest("hex")), + ); + if ((yield* fs.exists(archivePath)) && (yield* checksum(archivePath)) !== asset.sha256) { + yield* fs.remove(archivePath, { force: true }); + } + + if (!(yield* fs.exists(archivePath))) { + const temporaryArchivePath = `${archivePath}.${process.pid}.tmp`; + yield* Effect.gen(function* () { + yield* runCommand( + ChildProcess.make("curl", [ + "--fail", + "--location", + "--silent", + "--show-error", + "--output", + temporaryArchivePath, + asset.url, + ]), + { label: `download ${asset.archiveName}`, verbose: input.verbose }, + ); + const actualChecksum = yield* checksum(temporaryArchivePath); + if (actualChecksum !== asset.sha256) { + return yield* new CuaDriverChecksumMismatchError({ + archiveName: asset.archiveName, + expected: asset.sha256, + actual: actualChecksum, + }); + } + yield* fs.rename(temporaryArchivePath, archivePath); + }).pipe(Effect.ensuring(fs.remove(temporaryArchivePath, { force: true }).pipe(Effect.ignore))); + } + + const extractCommand = + input.platform === "win" + ? ChildProcess.make("powershell.exe", [ + "-NoProfile", + "-NonInteractive", + "-Command", + `Expand-Archive -LiteralPath '${archivePath.replaceAll("'", "''")}' -DestinationPath '${extractDir.replaceAll("'", "''")}' -Force`, + ]) + : ChildProcess.make("tar", ["-xzf", archivePath, "-C", extractDir]); + yield* runCommand(extractCommand, { + label: `extract ${asset.archiveName}`, + verbose: input.verbose, + }); + + yield* stageCuaDriverBundle({ + platform: input.platform, + arch: input.arch, + extractDir, + stageResourcesDir: input.stageResourcesDir, + }); + yield* Effect.log(`[desktop-artifact] Staged Cua Driver ${CUA_DRIVER_RELEASE_VERSION}.`); +}); + const assertPlatformBuildResources = Effect.fn("assertPlatformBuildResources")(function* ( platform: typeof BuildPlatform.Type, stageResourcesDir: string, @@ -3691,6 +3920,15 @@ const buildDesktopArtifact = Effect.fn("buildDesktopArtifact")(function* ( verbose: options.verbose, }); + yield* stageCuaDriverExecutable({ + platform: options.platform, + arch: options.arch, + repoRoot, + stageRoot, + stageResourcesDir, + verbose: options.verbose, + }); + yield* assertPlatformBuildResources( options.platform, stageResourcesDir, diff --git a/scripts/lib/cli-external-packages.test.ts b/scripts/lib/cli-external-packages.test.ts index e894359eccb5..6063cbc7e2f6 100644 --- a/scripts/lib/cli-external-packages.test.ts +++ b/scripts/lib/cli-external-packages.test.ts @@ -48,6 +48,9 @@ describe("shouldBundleCliDependency", () => { "@yuuang/ffi-rs-win32-x64-msvc", "@ff-labs/fff-node", "@clerk/electron-passkeys", + "@trycua/cua-driver/embedded", + "@trycua/cua-driver-darwin-arm64", + "@ubjs/node", "msgpackr-extract", "@msgpackr-extract/msgpackr-extract-win32-x64", ]) { @@ -87,7 +90,7 @@ describe("selectCliRuntimeExternalDependencies", () => { it("selects every external root declared by the server", () => { assert.deepStrictEqual( Object.keys(selectCliRuntimeExternalDependencies(serverPackageJson.dependencies)).sort(), - ["@ff-labs/fff-node", "msgpackr-extract", "node-pty"], + ["@ff-labs/fff-node", "@trycua/cua-driver", "msgpackr-extract", "node-pty"], ); }); }); diff --git a/scripts/lib/cli-external-packages.ts b/scripts/lib/cli-external-packages.ts index d7a89bc408a4..312442e12666 100644 --- a/scripts/lib/cli-external-packages.ts +++ b/scripts/lib/cli-external-packages.ts @@ -31,6 +31,8 @@ export const CLI_RUNTIME_EXTERNAL_PREFIXES = [ "@yuuang/", "@ff-labs/", "@clerk/electron-passkeys", + "@trycua/cua-driver", + "@ubjs/", "@msgpackr-extract/", "msgpackr-extract", "node-gyp-build", From 201eb1307b77956e49151b1ce57035000c6d58ef Mon Sep 17 00:00:00 2001 From: Francesco Bonacci Date: Wed, 9 Sep 2026 17:15:21 -0500 Subject: [PATCH 2/2] fix(server): release Cua host slot after destruction fails --- apps/server/src/cua/CuaDriver.test.ts | 23 ++++++++++++++++++++++- apps/server/src/cua/CuaDriver.ts | 7 +++++-- 2 files changed, 27 insertions(+), 3 deletions(-) diff --git a/apps/server/src/cua/CuaDriver.test.ts b/apps/server/src/cua/CuaDriver.test.ts index a0856fcf2507..61e587cfd660 100644 --- a/apps/server/src/cua/CuaDriver.test.ts +++ b/apps/server/src/cua/CuaDriver.test.ts @@ -14,7 +14,7 @@ import { const mcp: CuaDriverMcpConfiguration = { command: "/driver", args: ["mcp"], environment: [] }; -const standaloneFixture = Effect.fn(function* () { +const standaloneFixture = Effect.fn(function* (failFirstDestroy = false) { const start = Promise.withResolvers(); const stop = Promise.withResolvers(); const monitor = Promise.withResolvers(); @@ -46,6 +46,7 @@ const standaloneFixture = Effect.fn(function* () { uniffiDestroy() { destroys++; destroyed.resolve(); + if (failFirstDestroy && destroys === 1) throw new Error("destroy failed"); } }, })); @@ -73,6 +74,26 @@ const standaloneFixture = Effect.fn(function* () { }); describe("standalone Cua Driver ownership", () => { + it.effect("releases ownership after native destruction throws without retrying cleanup", () => + Effect.gen(function* () { + const f = yield* standaloneFixture(true); + const host = yield* f.factory; + f.start.resolve(f.connection); + yield* host.start; + f.stop.resolve(); + f.monitor.reject(new Error("monitor cancelled")); + expect(yield* Effect.flip(host.stop)).toEqual( + new CuaDriverError({ message: "Could not stop Cua Driver." }), + ); + expect(Exit.isFailure(yield* Effect.exit(host.stop))).toBe(true); + expect(f.counts()).toEqual({ creates: 1, stops: 1, destroys: 1 }); + const replacement = yield* f.factory; + expect(yield* replacement.start).toEqual(mcp); + yield* replacement.stop; + expect(f.counts()).toEqual({ creates: 2, stops: 2, destroys: 2 }); + }), + ); + it.effect("retains a cancelled start until start and stop settle, with one cleanup", () => Effect.gen(function* () { const f = yield* standaloneFixture(); diff --git a/apps/server/src/cua/CuaDriver.ts b/apps/server/src/cua/CuaDriver.ts index 28b734866dc6..ce54c9b461ef 100644 --- a/apps/server/src/cua/CuaDriver.ts +++ b/apps/server/src/cua/CuaDriver.ts @@ -203,8 +203,11 @@ export const makeStandaloneHostFactory = Effect.fn("CuaDriver.standaloneHostFact } catch { /* Cancellation ends the monitor. */ } - host.uniffiDestroy(); - occupied = false; + try { + host.uniffiDestroy(); + } finally { + occupied = false; + } })(); } return releasing;