diff --git a/apps/desktop/src/backend/DesktopBackendConfiguration.ts b/apps/desktop/src/backend/DesktopBackendConfiguration.ts index b306975beb24b49dab4faa6bddd16ce7c3e43e83..c5f60e963dc0adadfff9751951b4f091dace2a6d 100644 --- a/apps/desktop/src/backend/DesktopBackendConfiguration.ts +++ b/apps/desktop/src/backend/DesktopBackendConfiguration.ts @@ -577,6 +577,8 @@ const resolvePrimaryStartConfig = Effect.fn("desktop.backendConfiguration.resolv desktopTelemetryControlFd: 5, desktopBrowserFd: 6, desktopBrowserControlFd: 7, + desktopTerminalFd: 8, + desktopTerminalControlFd: 9, ...Option.match(input.resourceMonitorPath, { onNone: () => ({}), onSome: (resourceMonitorPath) => ({ resourceMonitorPath }), diff --git a/apps/desktop/src/backend/DesktopBackendManager.test.ts b/apps/desktop/src/backend/DesktopBackendManager.test.ts index 760e5a24f897163ee5d685d66a9251da7cb3c4e4..b6a2df1ec5fcdecc2399e31cbae33a9a6f7b35d3 100644 --- a/apps/desktop/src/backend/DesktopBackendManager.test.ts +++ b/apps/desktop/src/backend/DesktopBackendManager.test.ts @@ -1,3 +1,4 @@ +import * as DesktopTerminalHost from "./DesktopTerminalHost.ts"; import { SERVER_EXIT_CODE_STATE_DIR_OWNED, DesktopBackendBootstrap, @@ -176,6 +177,12 @@ function makeTestInstance(input: MakeInstanceInput) { updateCancellations: Stream.empty, ...input.desktopTelemetryPublisher, }), + Layer.succeed(DesktopTerminalHost.DesktopTerminalHost, { + requests: Stream.empty, + handleLine: () => Effect.void, + reset: Effect.void, + configure: () => Effect.void, + }), DesktopBrowserHost.layer, DesktopWslEnvironment.layerTest( input.pruneRuntimes === undefined ? {} : { pruneRuntimes: input.pruneRuntimes }, diff --git a/apps/desktop/src/backend/DesktopBackendManager.ts b/apps/desktop/src/backend/DesktopBackendManager.ts index c758b27dd282ce5f2f4f40b8f97056e58d746ba7..5b26e0e64dfb0d51f58b0686bc5a27d12450cd00 100644 --- a/apps/desktop/src/backend/DesktopBackendManager.ts +++ b/apps/desktop/src/backend/DesktopBackendManager.ts @@ -53,6 +53,7 @@ import { waitForHttpReady as waitForHttpReadyShared } from "@t3tools/shared/http import * as DesktopObservability from "../app/DesktopObservability.ts"; import * as DesktopTelemetryPublisher from "../telemetry/DesktopTelemetryPublisher.ts"; +import * as DesktopTerminalHost from "./DesktopTerminalHost.ts"; import * as DesktopBrowserHost from "../preview/DesktopBrowserHost.ts"; import * as DesktopWslEnvironment from "../wsl/DesktopWslEnvironment.ts"; @@ -226,6 +227,9 @@ export type BackendProcessError = typeof BackendProcessError.Type; interface RunBackendProcessOptions extends DesktopBackendStartConfig { readonly desktopTelemetryStream: Stream.Stream; + readonly desktopTerminalStream?: Stream.Stream; + readonly onDesktopTerminalClosed?: Effect.Effect; + readonly onDesktopTerminalLine?: (line: string) => Effect.Effect; /** Events for the desktop's browser tabs, written to `desktopBrowserFd`. */ readonly desktopBrowserStream?: Stream.Stream; /** Each line the backend writes to `desktopBrowserControlFd`. */ @@ -494,6 +498,17 @@ export const runBackendProcess = Effect.fn("runBackendProcess")(function* ( additionalFds[`fd${options.bootstrap.desktopBrowserControlFd}`] = { type: "output" }; } } + if ( + options.bootstrap.desktopTerminalFd !== undefined && + options.bootstrap.desktopTerminalControlFd !== undefined && + options.desktopTerminalStream + ) { + additionalFds[`fd${options.bootstrap.desktopTerminalFd}`] = { + type: "input", + stream: options.desktopTerminalStream, + }; + additionalFds[`fd${options.bootstrap.desktopTerminalControlFd}`] = { type: "output" }; + } const command = ChildProcess.make(options.executablePath, options.args, { cwd: options.cwd, env: options.env, @@ -563,6 +578,16 @@ export const runBackendProcess = Effect.fn("runBackendProcess")(function* ( Effect.forkScoped, ); } + if (options.bootstrap.desktopTerminalControlFd !== undefined && options.onDesktopTerminalLine) { + yield* handle.getOutputFd(options.bootstrap.desktopTerminalControlFd).pipe( + Stream.decodeText(), + Stream.splitLines, + Stream.runForEach(options.onDesktopTerminalLine), + Effect.catchCause(() => Effect.void), + Effect.ensuring(options.onDesktopTerminalClosed ?? Effect.void), + Effect.forkScoped, + ); + } if ( options.bootstrap.desktopBrowserControlFd !== undefined && options.onDesktopBrowserCommand !== undefined @@ -683,6 +708,7 @@ export const makeBackendInstance = Effect.fn("makeBackendInstance")(function* ( | HttpClient.HttpClient | DesktopObservability.DesktopBackendOutputLogFactory | DesktopTelemetryPublisher.DesktopTelemetryPublisher + | DesktopTerminalHost.DesktopTerminalHost | DesktopBrowserHost.DesktopBrowserHost | DesktopWslEnvironment.DesktopWslEnvironment | Scope.Scope @@ -692,6 +718,7 @@ export const makeBackendInstance = Effect.fn("makeBackendInstance")(function* ( const backendOutputLogFactory = yield* DesktopObservability.DesktopBackendOutputLogFactory; const backendOutputLog = yield* backendOutputLogFactory.forInstance(spec.id); const desktopTelemetryPublisher = yield* DesktopTelemetryPublisher.DesktopTelemetryPublisher; + const desktopTerminalHost = yield* DesktopTerminalHost.DesktopTerminalHost; const desktopBrowserHost = yield* DesktopBrowserHost.DesktopBrowserHost; const wslEnvironment = yield* DesktopWslEnvironment.DesktopWslEnvironment; const spawner = yield* ChildProcessSpawner.ChildProcessSpawner; @@ -954,10 +981,15 @@ export const makeBackendInstance = Effect.fn("makeBackendInstance")(function* ( ); }); + if (config.value.bootstrap.desktopTerminalFd !== undefined) + yield* desktopTerminalHost.configure(config.value.httpBaseUrl.href); const program = runBackendProcess({ ...config.value, desktopTelemetryStream: desktopTelemetryPublisher.encoded, // Only a bootstrap that names the browser fds (the local primary) gets them. + desktopTerminalStream: desktopTerminalHost.requests, + onDesktopTerminalLine: desktopTerminalHost.handleLine, + onDesktopTerminalClosed: desktopTerminalHost.reset, desktopBrowserStream: desktopBrowserHost.events, onDesktopBrowserCommand: desktopBrowserHost.handleCommandLine, onDesktopTelemetryControl: (message) => @@ -1027,6 +1059,11 @@ export const makeBackendInstance = Effect.fn("makeBackendInstance")(function* ( onSuccess: (exit) => finalizeRun(exit.reason, Option.getOrUndefined(exit.code)), }), Effect.ensuring(Scope.close(runScope, Exit.void).pipe(Effect.ignore)), + Effect.ensuring( + config.value.bootstrap.desktopTerminalFd === undefined + ? Effect.void + : desktopTerminalHost.reset, + ), ); const fiber = yield* Effect.forkIn(program, parentScope); diff --git a/apps/desktop/src/backend/DesktopBackendPool.test.ts b/apps/desktop/src/backend/DesktopBackendPool.test.ts index c5f31fc47a09bd31184b0876fe553015721fa953..2ca5ca313da55e6bf961edf30b0288407f90fd6c 100644 --- a/apps/desktop/src/backend/DesktopBackendPool.test.ts +++ b/apps/desktop/src/backend/DesktopBackendPool.test.ts @@ -1,3 +1,4 @@ +import * as DesktopTerminalHost from "./DesktopTerminalHost.ts"; import { assert, describe, it } from "@effect/vitest"; import * as Duration from "effect/Duration"; import * as Effect from "effect/Effect"; @@ -77,6 +78,12 @@ function layerPool(labelRef: Ref.Ref): Layer.Layer Effect.void, + reset: Effect.void, + configure: () => Effect.void, + }), DesktopBrowserHost.layer, Layer.succeed(DesktopBackendConfiguration.DesktopBackendConfiguration, { resolvePrimary: Effect.die("unexpected primary config resolve"), diff --git a/apps/desktop/src/backend/DesktopBackendPool.ts b/apps/desktop/src/backend/DesktopBackendPool.ts index 3a3ec70c57b804160a9d33eb9698afe53145d56f..6abfa02edd221a976b370aa70239acbce5f759d6 100644 --- a/apps/desktop/src/backend/DesktopBackendPool.ts +++ b/apps/desktop/src/backend/DesktopBackendPool.ts @@ -98,6 +98,7 @@ import * as DesktopBackendManager from "./DesktopBackendManager.ts"; import * as DesktopObservability from "../app/DesktopObservability.ts"; import * as DesktopAppSettings from "../settings/DesktopAppSettings.ts"; import * as DesktopTelemetryPublisher from "../telemetry/DesktopTelemetryPublisher.ts"; +import * as DesktopTerminalHost from "./DesktopTerminalHost.ts"; import * as DesktopBrowserHost from "../preview/DesktopBrowserHost.ts"; import * as DesktopWindow from "../window/DesktopWindow.ts"; import * as DesktopWslEnvironment from "../wsl/DesktopWslEnvironment.ts"; @@ -182,6 +183,7 @@ export type BackendInstanceFactoryRequirements = | HttpClient.HttpClient | DesktopObservability.DesktopBackendOutputLogFactory | DesktopTelemetryPublisher.DesktopTelemetryPublisher + | DesktopTerminalHost.DesktopTerminalHost | DesktopBrowserHost.DesktopBrowserHost | DesktopWslEnvironment.DesktopWslEnvironment; diff --git a/apps/desktop/src/backend/DesktopTerminalHost.test.ts b/apps/desktop/src/backend/DesktopTerminalHost.test.ts new file mode 100644 index 0000000000000000000000000000000000000000..7079948210645e65d6efc7c68fb5cd2a0f58c7e5 --- /dev/null +++ b/apps/desktop/src/backend/DesktopTerminalHost.test.ts @@ -0,0 +1,168 @@ +import { describe, expect, it } from "@effect/vitest"; +import { vi } from "vite-plus/test"; +import { DESKTOP_TERMINAL_CHANNEL } from "../ipc/channels.ts"; +import * as NodeServices from "@effect/platform-node/NodeServices"; +import * as Effect from "effect/Effect"; +import * as Fiber from "effect/Fiber"; +import * as Layer from "effect/Layer"; +import * as Option from "effect/Option"; +import * as Stream from "effect/Stream"; +import * as DesktopEnvironment from "../app/DesktopEnvironment.ts"; +import * as DesktopConfig from "../app/DesktopConfig.ts"; +import * as ElectronWindow from "../electron/ElectronWindow.ts"; +import * as DesktopTerminalHost from "./DesktopTerminalHost.ts"; + +const electron = await vi.hoisted(async () => { + const { EventEmitter } = await import("node:events"); + const sender = Object.assign(new EventEmitter(), { mainFrame: { url: "t3code-dev://app/" } }); + class Port extends EventEmitter { + messages: unknown[] = []; + closed = false; + postMessage(data: unknown) { + this.messages.push(data); + } + start() {} + close() { + if (!this.closed) { + this.closed = true; + this.emit("close"); + } + } + } + return { + ipcMain: new EventEmitter(), + sender, + Port, + window: { webContents: sender, isDestroyed: () => false }, + }; +}); +vi.mock("electron", () => ({ + ipcMain: electron.ipcMain, + BrowserWindow: { getAllWindows: () => [electron.window] }, +})); + +const layerEnvironment = DesktopEnvironment.layer({ + dirname: "/repo/apps/desktop/dist-electron", + homeDirectory: "/tmp/t3-terminal-test", + platform: "darwin", + processArch: "arm64", + appVersion: "0.0.45", + appPath: "/tmp/T3 Code.app", + isPackaged: false, + resourcesPath: "/tmp/resources", + runningUnderArm64Translation: false, +}).pipe( + Layer.provide( + Layer.mergeAll( + NodeServices.layer, + DesktopConfig.layerTest({ VITE_DEV_SERVER_URL: "http://localhost:5733" }), + ), + ), +); +const layer = DesktopTerminalHost.layer.pipe( + Layer.provide(ElectronWindow.layer), + Layer.provide(layerEnvironment), +); + +const event = (port: InstanceType, frame = electron.sender.mainFrame) => { + const replies: boolean[] = []; + let firstReply = false; + return { + sender: electron.sender, + senderFrame: frame, + ports: [port], + replies, + get returnValue() { + return firstReply; + }, + set returnValue(value: boolean) { + replies.push(value); + if (replies.length === 1) firstReply = value; + }, + }; +}; +const input = { + environmentId: "local-environment", + url: "ws://127.0.0.1:3773/ws?wsTicket=private", +}; +const readRequest = (host: DesktopTerminalHost.DesktopTerminalHost["Service"]) => + host.requests.pipe( + Stream.runHead, + Effect.map((data) => JSON.parse(new TextDecoder().decode(Option.getOrThrow(data)))), + Effect.forkScoped({ startImmediately: true }), + ); + +describe("DesktopTerminalHost", () => { + it.effect("selects the native route before ready and binds the announced identity", () => + Effect.gen(function* () { + const host = yield* DesktopTerminalHost.DesktopTerminalHost; + yield* host.configure("http://127.0.0.1:3773"); + const probe = (environmentId: string, httpBaseUrl = "http://127.0.0.1:3773") => { + const request = event(new electron.Port()); + electron.ipcMain.emit(`${DESKTOP_TERMINAL_CHANNEL}:available`, request, { + environmentId, + httpBaseUrl, + }); + expect(request.replies).toHaveLength(1); + return request.returnValue; + }; + expect(probe(input.environmentId)).toBe(true); + expect(probe(input.environmentId, "http://127.0.0.1:3774")).toBe(false); + yield* host.handleLine(JSON.stringify({ type: "ready", environmentId: input.environmentId })); + expect(probe("other-environment")).toBe(false); + yield* host.reset; + expect(probe(input.environmentId)).toBe(false); + }).pipe(Effect.provide(layer), Effect.scoped), + ); + + it.effect("forwards RPC frames through the private pipe and closes on navigation", () => + Effect.gen(function* () { + const host = yield* DesktopTerminalHost.DesktopTerminalHost; + yield* host.configure("http://127.0.0.1:3773"); + const port = new electron.Port(); + const opening = yield* readRequest(host); + electron.ipcMain.emit(DESKTOP_TERMINAL_CHANNEL, event(port), input); + const request = yield* Fiber.join(opening); + expect(request).toEqual({ type: "open", id: request.id, ...input }); + const data = '{"_tag":"Request","tag":"terminal.write"}'; + const message = yield* readRequest(host); + port.emit("message", { data }); + expect(yield* Fiber.join(message)).toEqual({ type: "message", id: request.id, data }); + yield* host.handleLine(JSON.stringify({ type: "message", id: request.id, data: "response" })); + expect(port.messages).toEqual([{ type: "message", id: request.id, data: "response" }]); + const closing = yield* readRequest(host); + electron.sender.emit("did-start-navigation", { isMainFrame: false, isSameDocument: false }); + expect(port.closed).toBe(false); + electron.sender.emit("did-start-navigation", { isMainFrame: true, isSameDocument: true }); + expect(port.closed).toBe(false); + electron.sender.emit("did-start-navigation", { isMainFrame: true, isSameDocument: false }); + expect(yield* Fiber.join(closing)).toEqual({ type: "close", id: request.id }); + expect(port.closed).toBe(true); + }).pipe(Effect.provide(layer), Effect.scoped), + ); + + it.effect("rejects embedded frames, other windows and custom-scheme origins", () => + Effect.gen(function* () { + const host = yield* DesktopTerminalHost.DesktopTerminalHost; + yield* host.configure("http://127.0.0.1:3773"); + for (const frame of [{ url: "t3code-dev://app/" }, { url: "t3code-dev://evil/" }]) { + const port = new electron.Port(); + electron.ipcMain.emit(DESKTOP_TERMINAL_CHANNEL, event(port, frame), input); + expect(port.closed).toBe(true); + } + const oldUrl = electron.sender.mainFrame.url; + electron.sender.mainFrame.url = "t3code-dev://evil/"; + const navigated = new electron.Port(); + electron.ipcMain.emit(DESKTOP_TERMINAL_CHANNEL, event(navigated), input); + expect(navigated.closed).toBe(true); + electron.sender.mainFrame.url = oldUrl; + const foreign = new electron.Port(); + electron.ipcMain.emit( + DESKTOP_TERMINAL_CHANNEL, + { ...event(foreign), sender: { mainFrame: electron.sender.mainFrame } }, + input, + ); + expect(foreign.closed).toBe(true); + }).pipe(Effect.provide(layer), Effect.scoped), + ); +}); diff --git a/apps/desktop/src/backend/DesktopTerminalHost.ts b/apps/desktop/src/backend/DesktopTerminalHost.ts new file mode 100644 index 0000000000000000000000000000000000000000..9e4a53997eb99ecc36d6fb1f0a374d2e3243091b --- /dev/null +++ b/apps/desktop/src/backend/DesktopTerminalHost.ts @@ -0,0 +1,176 @@ +import { DESKTOP_TERMINAL_CHANNEL } from "../ipc/channels.ts"; +import { DesktopTerminalResponse, DesktopTerminalRequest } from "@t3tools/contracts"; +import * as Context from "effect/Context"; +import * as Effect from "effect/Effect"; +import * as Layer from "effect/Layer"; +import * as PubSub from "effect/PubSub"; +import * as Schema from "effect/Schema"; +import * as Stream from "effect/Stream"; +import * as Electron from "electron"; +import { getDesktopUrl } from "../electron/ElectronProtocol.ts"; +import * as DesktopEnvironment from "../app/DesktopEnvironment.ts"; +import * as ElectronWindow from "../electron/ElectronWindow.ts"; + +const encode = Schema.encodeSync(Schema.fromJsonString(DesktopTerminalRequest)); +const decode = Schema.decodeUnknownOption(Schema.fromJsonString(DesktopTerminalResponse)); +const decodeAvailability = Schema.decodeUnknownOption( + Schema.Struct({ httpBaseUrl: Schema.String, environmentId: Schema.String }), +); +const decodeOpen = Schema.decodeUnknownOption( + Schema.Struct({ environmentId: Schema.String, url: Schema.String }), +); + +export class DesktopTerminalHost extends Context.Service< + DesktopTerminalHost, + { + readonly requests: Stream.Stream; + readonly handleLine: (line: string) => Effect.Effect; + readonly reset: Effect.Effect; + readonly configure: (httpBaseUrl: string) => Effect.Effect; + } +>()("@t3tools/desktop/backend/DesktopTerminalHost") {} + +export const layer = Layer.effect( + DesktopTerminalHost, + Effect.gen(function* () { + const environment = yield* DesktopEnvironment.DesktopEnvironment; + const appUrl = new URL(getDesktopUrl(environment.isDevelopment)); + const windows = yield* ElectronWindow.ElectronWindow; + const requests = yield* PubSub.unbounded(); + const run = Effect.runForkWith(yield* Effect.context()); + let nextId = 0; + const ports = new Map(); + let environmentId: string | null = null; + let backendOrigin: string | null = null; + const emit = (request: DesktopTerminalRequest) => + run(PubSub.publish(requests, new TextEncoder().encode(`${encode(request)}\n`))); + const reset = Effect.sync(() => { + environmentId = null; + backendOrigin = null; + for (const port of ports.values()) { + port.postMessage({ type: "close" }); + port.close(); + } + ports.clear(); + }); + const trusted = (event: Electron.IpcMainEvent) => { + const window = Effect.runSync(windows.currentMainOrFirst); + return ( + window._tag === "Some" && + window.value.webContents === event.sender && + event.senderFrame === event.sender.mainFrame && + new URL(event.senderFrame.url).protocol === appUrl.protocol && + new URL(event.senderFrame.url).host === appUrl.host + ); + }; + const onAvailable = (event: Electron.IpcMainEvent, input: unknown) => { + const value = decodeAvailability(input); + let available = false; + if (trusted(event) && value._tag === "Some" && backendOrigin !== null) { + try { + available = + new URL(value.value.httpBaseUrl).origin === backendOrigin && + (environmentId === null || value.value.environmentId === environmentId); + } catch {} + } + event.returnValue = available; + }; + const onOpen = (event: Electron.IpcMainEvent, input: unknown) => { + const port = event.ports[0]; + if (!port) return; + const decoded = decodeOpen(input); + if ( + !trusted(event) || + decoded._tag === "None" || + (environmentId !== null && decoded.value.environmentId !== environmentId) + ) { + port.postMessage({ type: "close" }); + port.close(); + return; + } + try { + const url = new URL(decoded.value.url); + url.protocol = url.protocol === "wss:" ? "https:" : "http:"; + if (url.origin !== backendOrigin) { + port.postMessage({ type: "close" }); + port.close(); + return; + } + } catch { + port.postMessage({ type: "close" }); + port.close(); + return; + } + const id = String(++nextId); + ports.set(id, port); + const close = () => { + if (ports.get(id) !== port) return; + ports.delete(id); + emit({ type: "close", id }); + port.close(); + }; + port.on("message", ({ data }: { data: unknown }) => { + if (typeof data === "string") emit({ type: "message", id, data }); + else close(); + }); + port.on("close", close); + const onNavigation = ( + details: Electron.Event, + ) => { + if (details.isMainFrame && !details.isSameDocument) close(); + }; + event.sender.once("destroyed", close); + event.sender.on("did-start-navigation", onNavigation); + port.once("close", () => { + event.sender.removeListener("destroyed", close); + event.sender.removeListener("did-start-navigation", onNavigation); + }); + emit({ type: "open", id, ...decoded.value }); + port.start(); + }; + yield* Effect.acquireRelease( + Effect.sync(() => { + Electron.ipcMain.on(`${DESKTOP_TERMINAL_CHANNEL}:available`, onAvailable); + Electron.ipcMain.on(DESKTOP_TERMINAL_CHANNEL, onOpen); + }), + () => + reset.pipe( + Effect.andThen( + Effect.sync(() => { + Electron.ipcMain.removeListener(`${DESKTOP_TERMINAL_CHANNEL}:available`, onAvailable); + Electron.ipcMain.removeListener(DESKTOP_TERMINAL_CHANNEL, onOpen); + }), + ), + ), + ); + return DesktopTerminalHost.of({ + requests: Stream.fromPubSub(requests), + configure: (url) => + reset.pipe( + Effect.andThen( + Effect.sync(() => { + backendOrigin = new URL(url).origin; + }), + ), + ), + reset, + handleLine: (line) => + Effect.sync(() => { + const response = decode(line); + if (response._tag === "None") return; + const value = response.value; + if (value.type === "ready") { + environmentId = value.environmentId; + return; + } + const port = ports.get(value.id); + if (!port) return; + port.postMessage(value); + if (value.type === "close") { + ports.delete(value.id); + port.close(); + } + }), + }); + }), +); diff --git a/apps/desktop/src/ipc/channels.ts b/apps/desktop/src/ipc/channels.ts index a6b617385b57842da503e88eef356916cfa3f8ed..8e2ebec04f95883050fa53341abdad0281d1582a 100644 --- a/apps/desktop/src/ipc/channels.ts +++ b/apps/desktop/src/ipc/channels.ts @@ -123,3 +123,4 @@ export const RECEIVE_PROVIDER_AUTH_CALLBACK_CHANNEL = "desktop:receive-provider- export const CANCEL_PROVIDER_AUTH_CALLBACK_CHANNEL = "desktop:cancel-provider-auth-callback"; export const TAKE_LEGACY_LOCAL_STORAGE_CHANNEL = "desktop:take-legacy-local-storage"; export const COMPLETE_LEGACY_LOCAL_STORAGE_CHANNEL = "desktop:complete-legacy-local-storage"; +export const DESKTOP_TERMINAL_CHANNEL = "desktop:terminal"; diff --git a/apps/desktop/src/main.ts b/apps/desktop/src/main.ts index ef00c4d66ed89b41f11f52aa6b754dccc7cc6446..5720858f6993a31beae743e5c4dd9acfa1dfd99f 100644 --- a/apps/desktop/src/main.ts +++ b/apps/desktop/src/main.ts @@ -43,6 +43,7 @@ import * as DesktopCliCommand from "./app/DesktopCliCommand.ts"; import * as DesktopApplicationMenu from "./window/DesktopApplicationMenu.ts"; import * as DesktopAssets from "./app/DesktopAssets.ts"; import * as DesktopBackendConfiguration from "./backend/DesktopBackendConfiguration.ts"; +import * as DesktopTerminalHost from "./backend/DesktopTerminalHost.ts"; import * as DesktopBackendPool from "./backend/DesktopBackendPool.ts"; import * as DesktopLocalEnvironmentAuth from "./backend/DesktopLocalEnvironmentAuth.ts"; import * as DesktopNetworkInterfaces from "./backend/DesktopNetworkInterfaces.ts"; @@ -193,6 +194,7 @@ const layerDesktopAppActivation = DesktopAppActivation.layer.pipe( // WSL second instance gets registered later in the migration. See // DesktopBackendPool.ts header for the full rollout plan. const layerDesktopBackend = DesktopBackendPool.layer.pipe( + Layer.provideMerge(DesktopTerminalHost.layer), Layer.provideMerge(DesktopAppIdentity.layer), Layer.provideMerge(DesktopBackendConfiguration.layer), Layer.provideMerge(DesktopWslEnvironment.layer), diff --git a/apps/desktop/src/preload.ts b/apps/desktop/src/preload.ts index efd49ffa3864895c7a22abfb82a81915ffd29131..c4e3d10e54d73724241177f688b7d088f10e81b0 100644 --- a/apps/desktop/src/preload.ts +++ b/apps/desktop/src/preload.ts @@ -77,6 +77,31 @@ function unwrapEnsureSshEnvironmentResult(result: unknown) { } contextBridge.exposeInMainWorld("desktopBridge", { + supportsLocalTerminal: (input) => + ipcRenderer.sendSync(`${IpcChannels.DESKTOP_TERMINAL_CHANNEL}:available`, input) === true, + openLocalTerminal: (input, onMessage, onClose) => { + const { port1, port2 } = new MessageChannel(); + let closed = false; + const close = () => { + if (closed) return; + closed = true; + port1.postMessage(null); + port1.close(); + onClose(); + }; + port1.addEventListener("message", ({ data }) => { + if (data?.type === "message" && typeof data.data === "string") onMessage(data.data); + else if (data?.type === "close") close(); + }); + ipcRenderer.postMessage(IpcChannels.DESKTOP_TERMINAL_CHANNEL, input, [port2]); + port1.start(); + return { + send: (data) => { + if (!closed) port1.postMessage(data); + }, + close, + }; + }, getAppBranding: () => { const result = ipcRenderer.sendSync(IpcChannels.GET_APP_BRANDING_CHANNEL); if (typeof result !== "object" || result === null) { diff --git a/apps/server/src/cli/config.ts b/apps/server/src/cli/config.ts index be2dd5789627d1b6aacc1abd1f745102b609d719..58fde17f8299848549d4836059212e3dea670dda 100644 --- a/apps/server/src/cli/config.ts +++ b/apps/server/src/cli/config.ts @@ -472,6 +472,8 @@ export const resolveServerConfig = ( ...(shellEnvironmentPrepared === undefined ? {} : { shellEnvironmentPrepared }), desktopTelemetryFd, desktopTelemetryControlFd, + desktopTerminalFd: bootstrap?.desktopTerminalFd, + desktopTerminalControlFd: bootstrap?.desktopTerminalControlFd, desktopBrowserFd, desktopBrowserControlFd, resourceMonitorPath, diff --git a/apps/server/src/config.ts b/apps/server/src/config.ts index ad2d182a3fadeeb706e35393fc4b967e196f2239..f4290a697ea82163c04f76e013e4741542415f2f 100644 --- a/apps/server/src/config.ts +++ b/apps/server/src/config.ts @@ -98,6 +98,8 @@ export class ServerConfig extends Context.Service< readonly shellEnvironmentPrepared?: boolean | undefined; readonly desktopTelemetryFd?: number | undefined; readonly desktopTelemetryControlFd?: number | undefined; + readonly desktopTerminalFd?: number | undefined; + readonly desktopTerminalControlFd?: number | undefined; readonly desktopBrowserFd?: number | undefined; readonly desktopBrowserControlFd?: number | undefined; readonly resourceMonitorPath?: string | undefined; diff --git a/apps/server/src/server.ts b/apps/server/src/server.ts index 23981a932e0524e226470ef2afdc991b4b574f3d..20a3f603ff9a190f6111f58aa26e2e2a9cbbc418 100644 --- a/apps/server/src/server.ts +++ b/apps/server/src/server.ts @@ -69,6 +69,7 @@ import * as DeviceHubProxy from "./device/DeviceHubProxy.ts"; import * as PreviewManager from "./preview/Manager.ts"; import * as PortScanner from "./preview/PortScanner.ts"; import * as ServerBrowser from "./preview/ServerBrowser.ts"; +import * as DesktopTerminalRpc from "./terminal/DesktopTerminalRpc.ts"; import * as DesktopBrowserChannel from "./preview/DesktopBrowserChannel.ts"; import * as ServerBrowserStream from "./preview/ServerBrowserStream.ts"; import * as PreviewBrowser from "./preview/PreviewBrowser.ts"; @@ -717,6 +718,7 @@ const layerMakeRoutes = Layer.mergeAll( ServerBrowserStream.routeLayer, ServerHttp.layerStaticAndDevRoute, Ws.layer, + DesktopTerminalRpc.layer, ), // The MCP session registry is provided globally (shared with V2 provider // sessions) rather than inline here. The orchestrator toolkit resolves diff --git a/apps/server/src/terminal/DesktopTerminalRpc.test.ts b/apps/server/src/terminal/DesktopTerminalRpc.test.ts new file mode 100644 index 0000000000000000000000000000000000000000..8401078dc94fedd8568e2a33e18810a0b2bf1053 --- /dev/null +++ b/apps/server/src/terminal/DesktopTerminalRpc.test.ts @@ -0,0 +1,129 @@ +import { expect, it } from "@effect/vitest"; +import * as NodeServices from "@effect/platform-node/NodeServices"; +import { + ORCHESTRATION_PROTOCOL_QUERY_PARAM, + ORCHESTRATION_PROTOCOL_VERSION, + WS_METHODS, + type DesktopTerminalRequest, + type DesktopTerminalResponse, +} from "@t3tools/contracts"; +import * as Effect from "effect/Effect"; +import * as Deferred from "effect/Deferred"; +import * as Layer from "effect/Layer"; +import * as Option from "effect/Option"; +import * as Queue from "effect/Queue"; +import * as Stream from "effect/Stream"; +import * as ServerConfig from "../config.ts"; +import * as ServerEnvironment from "../environment/ServerEnvironment.ts"; +import * as EnvironmentAuth from "../auth/EnvironmentAuth.ts"; +import * as SessionStore from "../auth/SessionStore.ts"; +import * as ServerSecretStore from "../auth/ServerSecretStore.ts"; +import * as SqlitePersistence from "../persistence/Sqlite.ts"; +import * as TerminalManager from "./Manager.ts"; +import * as DesktopTerminalRpc from "./DesktopTerminalRpc.ts"; + +const config = ServerConfig.layerTest(process.cwd(), { prefix: "t3-terminal-pipe-test-" }); +const auth = EnvironmentAuth.layer.pipe( + Layer.provideMerge(SqlitePersistence.layerMemory), + Layer.provide(ServerSecretStore.layer), + Layer.provideMerge(ServerEnvironment.layerIdentity), + Layer.provide(config), +); + +it.layer(NodeServices.layer)("DesktopTerminalRpc", (it) => { + it.effect("port close releases the running RPC, stream subscription and session connection", () => + Effect.gen(function* () { + const requests = yield* Queue.unbounded(); + const responses = yield* Queue.unbounded(); + const unsubscribed = yield* Deferred.make(); + const sessions = yield* SessionStore.SessionStore; + const identity = yield* ServerEnvironment.ServerEnvironmentIdentity; + const environmentId = yield* identity.getEnvironmentId; + const session = yield* sessions.issue({ scopes: ["terminal:read"] }); + const ticket = yield* sessions.issueWebSocketToken(session.sessionId); + const manager = Layer.mock(TerminalManager.TerminalManager, { + observeStream: (input, listener) => + listener({ + type: "output", + threadId: input.threadId, + terminalId: input.terminalId, + data: "ready", + }).pipe( + Effect.as(() => { + Deferred.doneUnsafe(unsubscribed, Effect.void); + }), + ), + }); + yield* DesktopTerminalRpc.run(Stream.fromQueue(requests), (response) => + Queue.offer(responses, response).pipe(Effect.asVoid), + ).pipe(Effect.provide(manager), Effect.forkScoped); + expect((yield* Queue.take(responses)).type).toBe("ready"); + const url = `ws://127.0.0.1/ws?${ORCHESTRATION_PROTOCOL_QUERY_PARAM}=${ORCHESTRATION_PROTOCOL_VERSION}&wsTicket=${ticket.token}`; + yield* Queue.offer(requests, { type: "open", id: "port-1", environmentId, url }); + yield* Queue.offer(requests, { + type: "message", + id: "port-1", + data: JSON.stringify({ + _tag: "Request", + id: "1", + tag: WS_METHODS.terminalObserve, + payload: { threadId: "thread", terminalId: "term" }, + headers: [], + }), + }); + const response = yield* Queue.take(responses); + expect(response.type).toBe("message"); + expect( + (yield* sessions.listActive()).find((value) => value.sessionId === session.sessionId) + ?.connected, + ).toBe(true); + yield* Queue.offer(requests, { type: "close", id: "port-1" }); + yield* Deferred.await(unsubscribed); + const closed = yield* Stream.fromQueue(responses).pipe( + Stream.filter((response) => response.type === "close"), + Stream.runHead, + ); + expect(Option.getOrThrow(closed)).toEqual({ type: "close", id: "port-1" }); + expect( + (yield* sessions.listActive()).find((value) => value.sessionId === session.sessionId) + ?.connected, + ).toBe(false); + + yield* Queue.offer(requests, { + type: "open", + id: "wrong-environment", + environmentId: "other-environment", + url, + }); + expect(yield* Queue.take(responses)).toEqual({ type: "close", id: "wrong-environment" }); + yield* Queue.offer(requests, { + type: "open", + id: "invalid-ticket", + environmentId, + url: url.replace(ticket.token, "invalid"), + }); + expect(yield* Queue.take(responses)).toEqual({ type: "close", id: "invalid-ticket" }); + yield* Queue.offer(requests, { type: "open", id: "read-only", environmentId, url }); + yield* Queue.offer(requests, { + type: "message", + id: "read-only", + data: JSON.stringify({ + _tag: "Request", + id: "2", + tag: WS_METHODS.terminalWrite, + payload: { threadId: "thread", terminalId: "term", data: "j" }, + headers: [], + }), + }); + const denied = yield* Queue.take(responses); + expect(denied.type).toBe("message"); + if (denied.type === "message") expect(denied.data).toContain("EnvironmentAuthorizationError"); + yield* sessions.revoke(session.sessionId); + const revoked = yield* Stream.fromQueue(responses).pipe( + Stream.filter((response) => response.type === "close"), + Stream.runHead, + ); + expect(Option.getOrThrow(revoked)).toEqual({ type: "close", id: "read-only" }); + }).pipe(Effect.provide(auth), Effect.scoped), + ); +}); diff --git a/apps/server/src/terminal/DesktopTerminalRpc.ts b/apps/server/src/terminal/DesktopTerminalRpc.ts new file mode 100644 index 0000000000000000000000000000000000000000..233e8ed81b78f9236278bb3ea68c4bf1ae190d87 --- /dev/null +++ b/apps/server/src/terminal/DesktopTerminalRpc.ts @@ -0,0 +1,203 @@ +// @effect-diagnostics nodeBuiltinImport:off -- Reads the private pipes inherited from Electron. +import { + DesktopTerminalRequest, + DesktopTerminalResponse, + TerminalRpcGroup, + ORCHESTRATION_PROTOCOL_VERSION, + ORCHESTRATION_PROTOCOL_QUERY_PARAM, +} from "@t3tools/contracts"; +import * as NodeStream from "@effect/platform-node/NodeStream"; +import * as Exit from "effect/Exit"; +import * as Effect from "effect/Effect"; +import * as Layer from "effect/Layer"; +import * as Option from "effect/Option"; +import * as Queue from "effect/Queue"; +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 Ndjson from "effect/encoding/Ndjson"; +import { HttpServerRequest } from "effect/http"; +import { RpcServer, RpcSerialization } from "effect/rpc"; +import type { FromClientEncoded } from "effect/rpc/RpcMessage"; +import * as NodeNet from "node:net"; +import * as ServerConfig from "../config.ts"; +import * as EnvironmentAuth from "../auth/EnvironmentAuth.ts"; +import * as SessionStore from "../auth/SessionStore.ts"; +import * as RpcAuthorization from "../auth/RpcAuthorization.ts"; +import { + RpcInstrumentation, + rpcInstrumentationLayer, +} from "../observability/RpcInstrumentation.ts"; +import * as ServerEnvironment from "../environment/ServerEnvironment.ts"; +import * as TerminalManager from "./Manager.ts"; +import { makeTerminalRpcHandlers } from "./Rpc.ts"; +import { withTerminalOutputWindow } from "./OutputProtocol.ts"; + +const group = TerminalRpcGroup.middleware(RpcInstrumentation); +const decode = Schema.decodeUnknownOption(DesktopTerminalRequest); +const encode = Schema.encodeSync(Schema.fromJsonString(DesktopTerminalResponse)); + +export const run = ( + requests: Stream.Stream, + send: (response: DesktopTerminalResponse) => Effect.Effect, +) => + Effect.gen(function* () { + const auth = yield* EnvironmentAuth.EnvironmentAuth; + const sessions = yield* SessionStore.SessionStore; + const manager = yield* TerminalManager.TerminalManager; + const identity = yield* ServerEnvironment.ServerEnvironmentIdentity; + const environmentId = yield* identity.getEnvironmentId; + const parentScope = yield* Scope.Scope; + const serialization = yield* RpcSerialization.RpcSerialization; + const connections = new Map< + string, + { scope: Scope.Closeable; messages: Queue.Queue } + >(); + const close = (id: string) => + Effect.gen(function* () { + const connection = connections.get(id); + if (!connection) return; + connections.delete(id); + yield* Scope.close(connection.scope, Exit.void); + yield* send({ type: "close", id }); + }); + const open = (request: Extract) => + Effect.gen(function* () { + if (connections.has(request.id)) return; + const scope = yield* Scope.fork(parentScope); + const messages = yield* Queue.unbounded(); + connections.set(request.id, { scope, messages }); + yield* Effect.gen(function* () { + yield* Effect.addFinalizer(() => Queue.shutdown(messages)); + const url = new URL(request.url); + if ( + request.environmentId !== environmentId || + url.pathname !== "/ws" || + url.searchParams.get(ORCHESTRATION_PROTOCOL_QUERY_PARAM) !== + String(ORCHESTRATION_PROTOCOL_VERSION) + ) + return; + url.protocol = url.protocol === "wss:" ? "https:" : "http:"; + const httpRequest = HttpServerRequest.fromWeb(new Request(url)); + const session = yield* auth.authenticateWebSocketUpgrade(httpRequest); + const protocol = yield* RpcServer.Protocol.make((receive) => + Effect.gen(function* () { + const parser = serialization.makeUnsafe(); + yield* Stream.fromQueue(messages).pipe( + Stream.runForEach((data) => + // The RPC parser owns encoded envelopes; RpcServer validates payloads against the group. + Effect.forEach( + parser.decode(data) as ReadonlyArray, + (message) => receive(0, message), + { discard: true }, + ), + ), + Effect.forkScoped, + ); + return { + disconnects: yield* Queue.unbounded(), + send: (_clientId, message) => + Effect.suspend(() => { + const data = parser.encode(message); + return data === undefined + ? Effect.void + : send({ + type: "message", + id: request.id, + data: typeof data === "string" ? data : new TextDecoder().decode(data), + }); + }), + end: () => Queue.shutdown(messages), + clientIds: Effect.succeed(new Set([0])), + initialMessage: Effect.succeedNone, + supportsAck: true, + supportsTransferables: false, + supportsSpanPropagation: true, + supportsNotifications: true, + codecFor: serialization.codecFor, + }; + }), + ); + yield* Effect.acquireUseRelease( + sessions.markConnected(session.sessionId), + () => + Effect.raceFirst( + RpcServer.make(group, { disableTracing: true, disableFatalDefects: true }).pipe( + Effect.provideService(RpcServer.Protocol, withTerminalOutputWindow(protocol)), + Effect.provide( + Layer.mergeAll( + group.toLayer(makeTerminalRpcHandlers(manager)), + RpcAuthorization.layer(session.scopes), + rpcInstrumentationLayer, + ), + ), + ), + sessions.awaitInvalidation(session.sessionId), + ), + () => sessions.markDisconnected(session.sessionId), + ); + }).pipe( + Scope.provide(scope), + Effect.catchCause(() => Effect.void), + Effect.ensuring(Effect.forkIn(close(request.id), parentScope).pipe(Effect.asVoid)), + Effect.forkIn(scope), + ); + }); + yield* send({ type: "ready", environmentId }); + yield* requests.pipe( + Stream.runForEach((value) => { + const request = value; + if (request.type === "open") return open(request); + if (request.type === "close") return close(request.id); + const connection = connections.get(request.id); + return connection ? Queue.offer(connection.messages, request.data) : Effect.void; + }), + Effect.ensuring( + Effect.suspend(() => Effect.forEach([...connections.keys()], close, { discard: true })), + ), + ); + }).pipe(Effect.provide(RpcSerialization.layerJson)); + +export const layer = Layer.effectDiscard( + Effect.gen(function* () { + const config = yield* ServerConfig.ServerConfig; + if (config.desktopTerminalFd === undefined || config.desktopTerminalControlFd === undefined) + return; + const outputFd = config.desktopTerminalControlFd; + const lock = yield* Semaphore.make(1); + const writable = yield* Effect.acquireRelease( + Effect.sync(() => { + const socket = new NodeNet.Socket({ fd: outputFd, readable: false, writable: true }); + socket.on("error", () => socket.destroy()); + return socket; + }), + (socket) => Effect.sync(() => socket.destroy()), + ); + const readable = yield* Effect.acquireRelease( + Effect.sync( + () => new NodeNet.Socket({ fd: config.desktopTerminalFd, readable: true, writable: false }), + ), + (stream) => Effect.sync(() => stream.destroy()), + ); + const requests = NodeStream.fromReadable({ + evaluate: () => readable, + onError: (cause) => cause, + }).pipe( + Stream.pipeThroughChannel(Ndjson.decode({ ignoreEmptyLines: true })), + Stream.map((value) => decode(value)), + Stream.filter(Option.isSome), + Stream.map((value) => value.value), + Stream.orDie, + ); + yield* run(requests, (response) => + lock.withPermits(1)( + Effect.callback((resume) => { + writable.write(Buffer.from(`${encode(response)}\n`), (error) => + resume(error ? Effect.die(error) : Effect.void), + ); + }), + ), + ).pipe(Effect.forkScoped); + }), +); diff --git a/apps/server/src/terminal/OutputProtocol.test.ts b/apps/server/src/terminal/OutputProtocol.test.ts index a12f9d6ecbf856831f42f4e32afa0a1be3112cac..75bc2d64c1f500cd658cfbbff6ae796f63e9488b 100644 --- a/apps/server/src/terminal/OutputProtocol.test.ts +++ b/apps/server/src/terminal/OutputProtocol.test.ts @@ -1,5 +1,5 @@ import { assert, describe, it } from "@effect/vitest"; -import { WS_METHODS } from "@t3tools/contracts"; +import { TerminalAttachStreamEvent, WS_METHODS } from "@t3tools/contracts"; import * as Deferred from "effect/Deferred"; import * as Effect from "effect/Effect"; import * as Queue from "effect/Queue"; @@ -8,7 +8,7 @@ import * as Stream from "effect/Stream"; import * as TestClock from "effect/testing/TestClock"; import { Rpc, RpcGroup, RpcMessage, RpcSerialization, RpcServer } from "effect/rpc"; -import { withTerminalOutputWindow } from "./OutputProtocol.ts"; +import { batchTerminalOutput, withTerminalOutputWindow } from "./OutputProtocol.ts"; describe("terminal output window", () => { it.effect.each([ @@ -77,3 +77,129 @@ describe("terminal output window", () => { }).pipe(Effect.provide(RpcSerialization.layerJson), Effect.scoped), ); }); + +describe("terminal output batches", () => { + it.effect("bounds an acknowledged backlog without changing bytes or event order", () => + Effect.gen(function* () { + const group = RpcGroup.make( + Rpc.make(WS_METHODS.terminalAttach, { success: TerminalAttachStreamEvent, stream: true }), + ); + const output = yield* Queue.unbounded(); + const responses = yield* Queue.unbounded(); + const receive = yield* Deferred.make[0]>(); + const protocol = yield* RpcServer.Protocol.make((write) => + Effect.gen(function* () { + yield* Deferred.succeed(receive, write); + const serialization = yield* RpcSerialization.RpcSerialization; + return { + disconnects: yield* Queue.unbounded(), + send: (_clientId, response) => Queue.offer(responses, response), + end: () => Effect.void, + clientIds: Effect.succeed(new Set([0])), + initialMessage: Effect.succeedNone, + supportsAck: true, + supportsTransferables: false, + supportsSpanPropagation: false, + supportsNotifications: true, + codecFor: serialization.codecFor, + }; + }), + ); + yield* RpcServer.make(group).pipe( + Effect.provide( + group.toLayerHandler(WS_METHODS.terminalAttach, () => + Stream.fromQueue(output).pipe(batchTerminalOutput), + ), + ), + Effect.provideService(RpcServer.Protocol, withTerminalOutputWindow(protocol)), + Effect.forkScoped, + ); + const write = yield* Deferred.await(receive); + yield* write(0, { + _tag: "Request", + id: "1", + tag: WS_METHODS.terminalAttach, + payload: null, + headers: [], + }); + const event = (sequence: number): TerminalAttachStreamEvent => ({ + type: "output", + threadId: "thread-1", + terminalId: "term-1", + sequence, + data: `\u001b[31m${"🙂\\\n".repeat(400)}\u001b[0m`, + }); + for (let index = 0; index < 8; index++) { + yield* Queue.offer(output, event(index)); + const response = yield* Queue.take(responses); + assert.equal(response._tag, "Chunk"); + } + + const expected = Array.from({ length: 1_024 }, (_, index) => event(index + 8)); + for (const value of expected) yield* Queue.offer(output, value); + yield* TestClock.adjust(0); + assert.equal(yield* Queue.size(output), expected.length); + assert.equal(yield* Queue.size(responses), 0); + + for (let index = 0; index < 8; index++) { + yield* write(0, { _tag: "Ack", requestId: "1" }); + } + const received: unknown[] = []; + let batchCount = 0; + while (received.length < expected.length) { + const response = yield* Queue.take(responses); + assert.equal(response._tag, "Chunk"); + if (response._tag !== "Chunk") continue; + + assert.isAtMost(Buffer.byteLength(JSON.stringify(response)), 64 * 1024); + received.push(...response.values); + batchCount += 1; + yield* write(0, { _tag: "Ack", requestId: "1" }); + } + assert.isAbove(batchCount, 1); + assert.deepEqual(received, expected); + }).pipe(Effect.provide(RpcSerialization.layerJson), Effect.scoped), + ); + + it.effect("preserves large snapshots and oversized sequenced events as individual batches", () => + Effect.gen(function* () { + const snapshot: TerminalAttachStreamEvent = { + type: "snapshot", + snapshot: { + threadId: "thread-1", + terminalId: "term-1", + cwd: "/tmp", + worktreePath: null, + status: "running", + pid: 123, + history: "x".repeat(8 * 1024 * 1024), + label: "sh", + exitCode: null, + exitSignal: null, + updatedAt: "2026-10-15T00:00:00.000Z", + }, + }; + const output: TerminalAttachStreamEvent = { + type: "output", + threadId: "thread-1", + terminalId: "term-1", + sequence: 12, + data: "🙂".repeat(32_768), + }; + const exit: TerminalAttachStreamEvent = { + type: "exited", + threadId: "thread-1", + terminalId: "term-1", + sequence: 13, + exitCode: 0, + exitSignal: null, + }; + const batches = yield* Stream.fromArray([snapshot, output, exit]).pipe( + batchTerminalOutput, + Stream.chunks, + Stream.runCollect, + ); + assert.deepEqual(batches, [[snapshot], [output], [exit]]); + }), + ); +}); diff --git a/apps/server/src/terminal/OutputProtocol.ts b/apps/server/src/terminal/OutputProtocol.ts index 402ed32c022ae0232c56698d7bf4599ee27882f1..171aef2586e62a0adf7223ebe36bb6c8fcdd25cf 100644 --- a/apps/server/src/terminal/OutputProtocol.ts +++ b/apps/server/src/terminal/OutputProtocol.ts @@ -1,5 +1,6 @@ -import { WS_METHODS } from "@t3tools/contracts"; +import { WS_METHODS, type TerminalAttachStreamEvent, type TerminalEvent } from "@t3tools/contracts"; import * as Effect from "effect/Effect"; +import * as Stream from "effect/Stream"; import type { RpcServer } from "effect/rpc"; const MAX_PENDING_CHUNKS = 8; @@ -8,6 +9,48 @@ const isFull = (sizes: number[]) => sizes.length >= MAX_PENDING_CHUNKS || sizes.reduce((total, size) => total + size, 0) >= MAX_PENDING_BYTES; +/** Keeps sequenced events intact, including snapshots larger than the batch budget. */ +export function batchTerminalOutput( + stream: Stream.Stream, +): Stream.Stream { + return Stream.transformPull(stream, (pull) => + Effect.sync(() => { + let pending: ReadonlyArray = []; + let index = 0; + + return Effect.gen(function* () { + if (index === pending.length) { + pending = yield* pull; + index = 0; + } + + const first = pending[index++]!; + const batch: [A, ...A[]] = [first]; + // @effect-diagnostics-next-line preferSchemaOverJson:off -- Match the RPC JSON wire encoding. + let bytes = Buffer.byteLength(JSON.stringify(first)) + 2; + + // Reserve 1 KiB for RPC framing; oversized events remain indivisible. + while (index < pending.length && bytes < MAX_PENDING_BYTES - 1024) { + const next = pending[index]!; + // @effect-diagnostics-next-line preferSchemaOverJson:off -- Match the RPC JSON wire encoding. + const size = Buffer.byteLength(JSON.stringify(next)) + 1; + if (bytes + size > MAX_PENDING_BYTES - 1024) break; + + batch.push(next); + bytes += size; + index += 1; + } + + if (index === pending.length) { + pending = []; + index = 0; + } + return batch; + }); + }), + ); +} + export function withTerminalOutputWindow( protocol: RpcServer.Protocol["Service"], ): RpcServer.Protocol["Service"] { diff --git a/apps/server/src/terminal/Rpc.ts b/apps/server/src/terminal/Rpc.ts new file mode 100644 index 0000000000000000000000000000000000000000..55c2ffc5e963521ab9ae038c8d80ced8054e738e --- /dev/null +++ b/apps/server/src/terminal/Rpc.ts @@ -0,0 +1,45 @@ +import { + TerminalRpcGroup, + WS_METHODS, + type TerminalAttachStreamEvent, + type TerminalEvent, + type TerminalError, +} from "@t3tools/contracts"; +import * as Effect from "effect/Effect"; +import * as Queue from "effect/Queue"; +import * as Stream from "effect/Stream"; +import * as TerminalManager from "./Manager.ts"; +import { batchTerminalOutput } from "./OutputProtocol.ts"; + +export const makeTerminalRpcHandlers = ( + terminalManager: TerminalManager.TerminalManager["Service"], +) => + TerminalRpcGroup.of({ + [WS_METHODS.terminalOpen]: (input) => terminalManager.open(input), + [WS_METHODS.terminalAttach]: (input) => + Stream.callback((queue) => + Effect.acquireRelease( + terminalManager.attachStream(input, (event) => Queue.offer(queue, event)), + (unsubscribe) => Effect.sync(unsubscribe), + ).pipe(Effect.catchCause((cause) => Queue.failCause(queue, cause))), + ).pipe(batchTerminalOutput), + [WS_METHODS.terminalWrite]: (input) => terminalManager.write(input), + [WS_METHODS.terminalResize]: (input) => terminalManager.resize(input), + [WS_METHODS.terminalClear]: (input) => terminalManager.clear(input), + [WS_METHODS.terminalRestart]: (input) => terminalManager.restart(input), + [WS_METHODS.terminalClose]: (input) => terminalManager.close(input), + [WS_METHODS.terminalObserve]: (input) => + Stream.callback((queue) => + Effect.acquireRelease( + terminalManager.observeStream(input, (event) => Queue.offer(queue, event)), + (unsubscribe) => Effect.sync(unsubscribe), + ).pipe(Effect.catchCause((cause) => Queue.failCause(queue, cause))), + ).pipe(batchTerminalOutput), + [WS_METHODS.subscribeTerminalEvents]: (_input) => + Stream.callback((queue) => + Effect.acquireRelease( + terminalManager.subscribe((event) => Queue.offer(queue, event)), + (unsubscribe) => Effect.sync(unsubscribe), + ), + ).pipe(batchTerminalOutput), + }); diff --git a/apps/server/src/ws.ts b/apps/server/src/ws.ts index 96994ee37c1301a99e59e3ba026871340b1a2a1a..139f3646a5416e1420ea324541a125b56e4d5528 100644 --- a/apps/server/src/ws.ts +++ b/apps/server/src/ws.ts @@ -94,9 +94,6 @@ import { type ProviderDriverKind, type ProviderInstanceId, ThreadId, - type TerminalAttachStreamEvent, - type TerminalError, - type TerminalEvent, type TerminalMetadataStreamEvent, type PullRequestRef, WS_METHODS, @@ -178,6 +175,7 @@ import * as ServerRuntimeStartup from "./serverRuntimeStartup.ts"; import * as ServerSettings from "./serverSettings.ts"; import * as StorageCleanup from "./storageCleanup.ts"; import * as TerminalManager from "./terminal/Manager.ts"; +import { makeTerminalRpcHandlers } from "./terminal/Rpc.ts"; import { withTerminalOutputWindow } from "./terminal/OutputProtocol.ts"; import * as PreviewAutomationBroker from "./mcp/PreviewAutomationBroker.ts"; import * as ServerBrowser from "./preview/ServerBrowser.ts"; @@ -2892,33 +2890,7 @@ const layerWsRpc = ( vcsProvisioning.initRepository(input).pipe(Effect.tap(() => refreshGitStatus(input.cwd))), [WS_METHODS.reviewGetDiffPreview]: (input) => review.getDiffPreview(input), [WS_METHODS.reviewGetDiffFileContents]: (input) => review.getDiffFileContents(input), - [WS_METHODS.terminalOpen]: (input) => terminalManager.open(input), - [WS_METHODS.terminalAttach]: (input) => - Stream.callback((queue) => - Effect.acquireRelease( - terminalManager.attachStream(input, (event) => Queue.offer(queue, event)), - (unsubscribe) => Effect.sync(unsubscribe), - ).pipe(Effect.catchCause((cause) => Queue.failCause(queue, cause))), - ), - [WS_METHODS.terminalWrite]: (input) => terminalManager.write(input), - [WS_METHODS.terminalResize]: (input) => terminalManager.resize(input), - [WS_METHODS.terminalClear]: (input) => terminalManager.clear(input), - [WS_METHODS.terminalRestart]: (input) => terminalManager.restart(input), - [WS_METHODS.terminalClose]: (input) => terminalManager.close(input), - [WS_METHODS.terminalObserve]: (input) => - Stream.callback((queue) => - Effect.acquireRelease( - terminalManager.observeStream(input, (event) => Queue.offer(queue, event)), - (unsubscribe) => Effect.sync(unsubscribe), - ).pipe(Effect.catchCause((cause) => Queue.failCause(queue, cause))), - ), - [WS_METHODS.subscribeTerminalEvents]: (_input) => - Stream.callback((queue) => - Effect.acquireRelease( - terminalManager.subscribe((event) => Queue.offer(queue, event)), - (unsubscribe) => Effect.sync(unsubscribe), - ), - ), + ...makeTerminalRpcHandlers(terminalManager), [WS_METHODS.subscribeTerminalMetadata]: (_input) => Stream.callback((queue) => Effect.acquireRelease( diff --git a/apps/web/src/connection/runtime.ts b/apps/web/src/connection/runtime.ts index 5e3aa3655102e189622463e5f6d6d2c01b82821c..c573a637b5ff0e2f196bff55b90cc4a81cedea73 100644 --- a/apps/web/src/connection/runtime.ts +++ b/apps/web/src/connection/runtime.ts @@ -1,3 +1,4 @@ +import { desktopTerminalSocket } from "@t3tools/client-runtime/rpc"; import { Connection } from "@t3tools/client-runtime/connection"; import { ShellSnapshotLoader } from "@t3tools/client-runtime/state/shell"; import { @@ -32,6 +33,7 @@ type ConnectionLayerSource = const layerProvidedClientConnection = layerSnapshotLoader.pipe( Layer.provideMerge( Connection.layerWithOptions({ + terminalSocket: (prepared) => desktopTerminalSocket(window.desktopBridge, prepared), environmentThemes: true, usageLimitSources: true, usageLimitsCommand: true, diff --git a/docs/internals/terminal-runtime.md b/docs/internals/terminal-runtime.md index ca5f0d46ead6e5eed89c61ce5a23ff2d2ca587fe..3b76a23ee7b273317ed15c78c70ce13fa537a956 100644 --- a/docs/internals/terminal-runtime.md +++ b/docs/internals/terminal-runtime.md @@ -5,6 +5,13 @@ client, including the desktop renderer, attaches through the environment connect This lets clients reconnect or share a running session. Renderer choices stay local to each client and do not change terminal contracts. +Terminal I/O stays independent of chat traffic so stalled streams and queued +compression cannot block keystrokes. The native desktop's own backend uses a +MessagePort and inherited pipes; other environments use a lazy second WebSocket. +Both paths share terminal handlers, ticket authorization, and session supervision. +Metadata stays on the primary connection so an idle sidebar opens neither path. +WSL uses WebSockets because its launcher does not preserve inherited descriptors. + ## Output and retention [Terminal history](../../apps/server/src/terminal/Manager.ts) is incremental. diff --git a/packages/client-runtime/src/connection/driver.ts b/packages/client-runtime/src/connection/driver.ts index d34bc1a9ba49a4efbaae9d96c3dbc2f7b23a401f..0400062256211ca22d7a589338181d361f1aa6eb 100644 --- a/packages/client-runtime/src/connection/driver.ts +++ b/packages/client-runtime/src/connection/driver.ts @@ -178,7 +178,10 @@ export const make = Effect.gen(function* () { Effect.fnUntraced(function* (route) { const prepared = yield* resolver.prepare(routeEntry(entry, route)); yield* reportProgress({ stage: "opening", prepared }); - const session = yield* sessions.connect(prepared); + const session = yield* sessions.connect( + prepared, + resolver.prepare(routeEntry(entry, route)), + ); yield* reportProgress({ stage: "synchronizing", prepared }); yield* session.ready; return { prepared, session } satisfies EnvironmentConnectionLease; diff --git a/packages/client-runtime/src/rpc/desktopTerminal.ts b/packages/client-runtime/src/rpc/desktopTerminal.ts new file mode 100644 index 0000000000000000000000000000000000000000..f9b6b2271a24ec8919929eeb77e7dc2571619a5e --- /dev/null +++ b/packages/client-runtime/src/rpc/desktopTerminal.ts @@ -0,0 +1,67 @@ +import type { DesktopBridge, DesktopTerminalConnection } from "@t3tools/contracts"; +import * as Deferred from "effect/Deferred"; +import * as Effect from "effect/Effect"; +import * as Queue from "effect/Queue"; +import * as Socket from "effect/socket/Socket"; +import type { PreparedConnection } from "../connection/model.ts"; + +export function desktopTerminalSocket( + bridge: Pick | undefined, + prepared: PreparedConnection, +): Socket.Socket | undefined { + if (!bridge?.openLocalTerminal || bridge.supportsLocalTerminal?.(prepared) !== true) return; + const open = bridge.openLocalTerminal; + const connected = Deferred.makeUnsafe(); + const closed = () => + new Socket.SocketError({ reason: new Socket.SocketCloseError({ code: 1001 }) }); + return { + [Socket.TypeId]: Socket.TypeId, + reader: Effect.gen(function* () { + const queue = yield* Queue.unbounded(); + const run = Effect.runForkWith(yield* Effect.context()); + yield* Effect.acquireRelease( + Effect.sync(() => { + const connection = open( + { environmentId: prepared.environmentId, url: prepared.socketUrl }, + (data) => { + run(Queue.offer(queue, data)); + }, + () => { + run(Queue.fail(queue, closed())); + }, + ); + run(Deferred.succeed(connected, connection)); + return connection; + }), + (connection) => + Effect.sync(connection.close).pipe(Effect.andThen(Queue.fail(queue, closed()))), + ); + return { + pull: Queue.take(queue).pipe(Effect.map((data) => [data] as const)), + upgrade: () => + Effect.fail(new Socket.SocketError({ reason: new Socket.SocketUpgradeError({}) })), + }; + }), + writer: Effect.succeed({ + write: (data) => + Deferred.await(connected).pipe( + Effect.flatMap((connection) => + Effect.sync(() => { + if (Socket.isCloseEvent(data)) connection.close(); + else + connection.send(typeof data === "string" ? data : new TextDecoder().decode(data)); + }), + ), + ), + writeAll: (chunks) => + Deferred.await(connected).pipe( + Effect.flatMap((connection) => + Effect.sync(() => { + for (const data of chunks) + connection.send(typeof data === "string" ? data : new TextDecoder().decode(data)); + }), + ), + ), + }), + }; +} diff --git a/packages/client-runtime/src/rpc/index.ts b/packages/client-runtime/src/rpc/index.ts index e0e269bdfa8ad31d4d273da5809a9f3e91caa63a..50ef9664ea121369a490b52d1d994b3e2dfea731 100644 --- a/packages/client-runtime/src/rpc/index.ts +++ b/packages/client-runtime/src/rpc/index.ts @@ -2,3 +2,4 @@ export * from "./client.ts"; export * from "./http.ts"; export type { WsRpcProtocolClient } from "./protocol.ts"; export { type RpcSession } from "./session.ts"; +export { desktopTerminalSocket } from "./desktopTerminal.ts"; diff --git a/packages/client-runtime/src/rpc/session.test.ts b/packages/client-runtime/src/rpc/session.test.ts index 38e1b7f89d31e0ecbe0ec05d49f27bc82aed7881..c533e06a7f24221c1a12f518447d56afef69a56d 100644 --- a/packages/client-runtime/src/rpc/session.test.ts +++ b/packages/client-runtime/src/rpc/session.test.ts @@ -1,3 +1,4 @@ +import { desktopTerminalSocket } from "./desktopTerminal.ts"; import { DEFAULT_SERVER_SETTINGS, EnvironmentId, @@ -288,10 +289,326 @@ const publishConfigEvents = Effect.fn("TestRpcSessionFactory.publishConfigEvents }); describe("RpcSessionFactory", () => { + it.effect("uses the local pipe for terminal RPC without opening a terminal WebSocket", () => + Effect.gen(function* () { + const incoming = yield* Queue.unbounded(); + let respond: (data: string) => void = () => {}; + let onClose: () => void = () => {}; + let opened = 0; + let closed = 0; + const { factory, sockets } = yield* makeFactory({ + terminalSocket: (prepared) => + desktopTerminalSocket( + { + supportsLocalTerminal: ({ environmentId }) => + environmentId === PREPARED.environmentId, + openLocalTerminal: (input, receive, disconnected) => { + expect(input.url).toContain("wsTicket=pipe-ticket"); + opened += 1; + respond = receive; + onClose = disconnected; + return { + send: (data) => { + Queue.offerUnsafe(incoming, data); + }, + close: () => { + closed += 1; + }, + }; + }, + }, + prepared, + ), + }); + const session = yield* factory.connect( + PREPARED, + Effect.succeed({ ...PREPARED, socketUrl: `${PREPARED.socketUrl}&wsTicket=pipe-ticket` }), + ); + const main = yield* awaitSocket(sockets); + main.open(); + yield* completeInitialConfig(main); + yield* session.ready; + expect(opened).toBe(0); + const write = yield* session.client[WS_METHODS.terminalWrite]({ + threadId: "thread", + terminalId: "term-1", + data: "j", + }).pipe(Effect.forkChild); + const message = JSON.parse(yield* Queue.take(incoming)); + expect(message.tag).toBe(WS_METHODS.terminalWrite); + expect(message.payload.data).toBe("j"); + respond( + encodeJson({ _tag: "Exit", requestId: message.id, exit: { _tag: "Success", value: null } }), + ); + yield* Fiber.join(write); + expect(sockets).toHaveLength(1); + expect(opened).toBe(1); + onClose(); + expect((yield* Effect.exit(session.closed))._tag).toBe("Failure"); + expect(closed).toBeLessThanOrEqual(1); + }).pipe(Effect.scoped), + ); + + it.effect( + "opens one terminal socket lazily with fresh authorization and isolates blocked metadata", + () => + Effect.gen(function* () { + const { factory, sockets } = yield* makeFactory(); + let authorizations = 0; + const session = yield* factory.connect( + PREPARED, + Effect.sync(() => { + authorizations += 1; + return { + ...PREPARED, + socketUrl: `${PREPARED.socketUrl}&wsTicket=fresh-terminal-ticket`, + }; + }), + ); + const primary = yield* awaitSocket(sockets); + primary.open(); + yield* completeInitialConfig(primary); + yield* session.ready; + const metadata = yield* session.client[WS_METHODS.subscribeTerminalMetadata]( + {}, + { asQueue: true }, + ); + const metadataRequest = yield* awaitRequest(primary, 1); + expect(authorizations).toBe(0); + expect(sockets).toHaveLength(1); + primary.serverMessage( + encodeJson({ + _tag: "Chunk", + requestId: metadataRequest.id, + values: Array.from({ length: 17 }, () => ({ + type: "remove", + threadId: "thread", + terminalId: "term-1", + })), + }), + ); + yield* Effect.yieldNow; + expect(yield* Queue.size(metadata)).toBe(16); + const probed = yield* Deferred.make(); + const probe = yield* Effect.forkChild( + session.probe.pipe(Effect.tap(() => Deferred.succeed(probed, undefined))), + ); + const probeRequest = yield* awaitRequest(primary, 2); + primary.serverMessage( + encodeJson({ + _tag: "Exit", + requestId: probeRequest.id, + exit: { _tag: "Success", value: {} }, + }), + ); + const observed = yield* session.client[WS_METHODS.terminalObserve]( + { threadId: "thread", terminalId: "term-1" }, + { asQueue: true }, + ); + const terminal = yield* awaitSocket(sockets, 1); + terminal.open(); + const observeRequest = yield* awaitRequest(terminal); + expect(observeRequest.tag).toBe(WS_METHODS.terminalObserve); + expect(terminal.url).toContain("wsTicket=fresh-terminal-ticket"); + const write = yield* session.client[WS_METHODS.terminalWrite]({ + threadId: "thread", + terminalId: "term-1", + data: "j", + }).pipe(Effect.forkChild); + const writeRequest = yield* awaitRequest(terminal, 1); + terminal.serverMessage( + encodeJson({ + _tag: "Exit", + requestId: writeRequest.id, + exit: { _tag: "Success", value: null }, + }), + ); + terminal.serverMessage( + encodeJson({ + _tag: "Chunk", + requestId: observeRequest.id, + values: [{ type: "output", threadId: "thread", terminalId: "term-1", data: "j" }], + }), + ); + yield* Fiber.join(write); + expect((yield* Queue.take(observed)).type).toBe("output"); + expect(yield* Deferred.isDone(probed)).toBe(false); + expect(authorizations).toBe(1); + expect(sockets).toHaveLength(2); + yield* Queue.take(metadata); + yield* Fiber.join(probe); + }), + ); + + it.effect( + "closes both transports with the session and obtains fresh authorization on its replacement", + () => + Effect.gen(function* () { + const { factory, sockets } = yield* makeFactory(); + let authorizations = 0; + const prepareTerminal = Effect.sync(() => ({ + ...PREPARED, + socketUrl: `${PREPARED.socketUrl}&wsTicket=terminal-${++authorizations}`, + })); + for (let attempt = 0; attempt < 2; attempt += 1) { + yield* Effect.scoped( + Effect.gen(function* () { + const session = yield* factory.connect(PREPARED, prepareTerminal); + const primary = yield* awaitSocket(sockets, attempt * 2); + primary.open(); + yield* completeInitialConfig(primary); + yield* session.ready; + const resize = yield* session.client[WS_METHODS.terminalResize]({ + threadId: "thread", + terminalId: "term-1", + cols: 80, + rows: 24, + }).pipe(Effect.forkChild); + const terminal = yield* awaitSocket(sockets, attempt * 2 + 1); + terminal.open(); + const request = yield* awaitRequest(terminal); + terminal.serverMessage( + encodeJson({ + _tag: "Exit", + requestId: request.id, + exit: { _tag: "Success", value: null }, + }), + ); + yield* Fiber.join(resize); + expect(terminal.url).toContain(`wsTicket=terminal-${attempt + 1}`); + (attempt === 0 ? primary : terminal).close(1012, "restart"); + expect(yield* Effect.flip(session.closed)).toMatchObject({ reason: "transport" }); + }), + ); + expect(sockets[attempt * 2]?.readyState).toBe(TestWebSocket.CLOSED); + expect(sockets[attempt * 2 + 1]?.readyState).toBe(TestWebSocket.CLOSED); + } + expect(authorizations).toBe(2); + }), + ); + + it.effect("keeps terminal initialization alive when its first command is interrupted", () => + Effect.gen(function* () { + const { factory, sockets } = yield* makeFactory(); + const authorizing = yield* Deferred.make(); + const authorized = yield* Deferred.make(); + let authorizations = 0; + const session = yield* factory.connect( + PREPARED, + Effect.gen(function* () { + authorizations += 1; + yield* Deferred.succeed(authorizing, undefined); + return yield* Deferred.await(authorized); + }), + ); + const primary = yield* awaitSocket(sockets); + primary.open(); + yield* completeInitialConfig(primary); + yield* session.ready; + const first = yield* session.client[WS_METHODS.terminalWrite]({ + threadId: "thread", + terminalId: "term-1", + data: "first", + }).pipe(Effect.forkChild); + yield* Deferred.await(authorizing); + yield* Fiber.interrupt(first); + yield* Deferred.succeed(authorized, PREPARED); + const terminal = yield* awaitSocket(sockets, 1); + terminal.open(); + const second = yield* session.client[WS_METHODS.terminalWrite]({ + threadId: "thread", + terminalId: "term-1", + data: "second", + }).pipe(Effect.forkChild); + const request = yield* awaitRequest(terminal); + expect(request.payload).toMatchObject({ data: "second" }); + terminal.serverMessage( + encodeJson({ _tag: "Exit", requestId: request.id, exit: { _tag: "Success", value: null } }), + ); + yield* Fiber.join(second); + expect(authorizations).toBe(1); + expect(primary.readyState).toBe(TestWebSocket.OPEN); + expect(terminal.readyState).toBe(TestWebSocket.OPEN); + }), + ); + + it.effect("keeps the terminal transport after its observing caller finishes", () => + Effect.gen(function* () { + const { factory, sockets } = yield* makeFactory(); + const session = yield* factory.connect(PREPARED, Effect.succeed(PREPARED)); + const primary = yield* awaitSocket(sockets); + primary.open(); + yield* completeInitialConfig(primary); + yield* session.ready; + const observing = yield* session.client[WS_METHODS.terminalObserve]({ + threadId: "thread", + terminalId: "term-1", + }).pipe(Stream.take(1), Stream.runCollect, Effect.forkChild); + const terminal = yield* awaitSocket(sockets, 1); + terminal.open(); + const observeRequest = yield* awaitRequest(terminal); + terminal.serverMessage( + encodeJson({ + _tag: "Chunk", + requestId: observeRequest.id, + values: [{ type: "output", threadId: "thread", terminalId: "term-1", data: "a" }], + }), + ); + yield* Fiber.join(observing); + const writing = yield* session.client[WS_METHODS.terminalWrite]({ + threadId: "thread", + terminalId: "term-1", + data: "j", + }).pipe(Effect.forkChild); + const writeRequest = yield* awaitRequest(terminal, 1); + terminal.serverMessage( + encodeJson({ + _tag: "Exit", + requestId: writeRequest.id, + exit: { _tag: "Success", value: null }, + }), + ); + yield* Fiber.join(writing); + expect(sockets).toHaveLength(2); + expect(terminal.readyState).toBe(TestWebSocket.OPEN); + }), + ); + + it.effect.each(["write", "observe", "queue"] as const)( + "fails %s and the session if terminal authorization is revoked before first use", + (consumer) => + Effect.gen(function* () { + const { factory, sockets } = yield* makeFactory(); + const revoked = new ConnectionBlockedError({ + reason: "permission", + detail: "Terminal access was revoked.", + }); + const session = yield* factory.connect(PREPARED, Effect.fail(revoked)); + const primary = yield* awaitSocket(sockets); + primary.open(); + yield* completeInitialConfig(primary); + yield* session.ready; + const input = { threadId: "thread", terminalId: "term-1" }; + const call = + consumer === "write" + ? session.client[WS_METHODS.terminalWrite]({ ...input, data: "j" }) + : consumer === "observe" + ? session.client[WS_METHODS.terminalObserve](input).pipe(Stream.runDrain) + : Effect.flatMap( + session.client[WS_METHODS.terminalObserve](input, { asQueue: true }), + (queue) => Stream.fromQueue(queue).pipe(Stream.runDrain), + ); + const result = yield* Effect.flip(call); + expect(result._tag).toBe("RpcClientError"); + expect(yield* Effect.flip(session.closed)).toBe(revoked); + expect(sockets).toHaveLength(1); + }), + ); + it.effect("owns one scoped websocket attempt and exposes readiness and closure", () => Effect.gen(function* () { const { factory, sockets } = yield* makeFactory(); - const session = yield* factory.connect(PREPARED); + const session = yield* factory.connect(PREPARED, Effect.succeed(PREPARED)); const readyFiber = yield* Effect.forkChild(session.ready); const socket = yield* awaitSocket(sockets); @@ -355,7 +672,7 @@ describe("RpcSessionFactory", () => { yield* Effect.scoped( Effect.gen(function* () { - const session = yield* factory.connect(PREPARED); + const session = yield* factory.connect(PREPARED, Effect.succeed(PREPARED)); const readyFiber = yield* Effect.forkChild(session.ready); const socket = yield* awaitSocket(sockets); socket.open(); @@ -372,7 +689,7 @@ describe("RpcSessionFactory", () => { Effect.scoped( Effect.gen(function* () { const { factory, sockets } = yield* makeFactory(); - const session = yield* factory.connect(PREPARED); + const session = yield* factory.connect(PREPARED, Effect.succeed(PREPARED)); const readyFiber = yield* Effect.forkChild(session.ready); const socket = yield* awaitSocket(sockets); socket.open(); @@ -437,7 +754,7 @@ describe("RpcSessionFactory", () => { Effect.scoped( Effect.gen(function* () { const { factory, sockets } = yield* makeFactory(options); - const session = yield* factory.connect(PREPARED); + const session = yield* factory.connect(PREPARED, Effect.succeed(PREPARED)); const readyFiber = yield* Effect.forkChild(session.ready); const socket = yield* awaitSocket(sockets); socket.open(); @@ -486,7 +803,7 @@ describe("RpcSessionFactory", () => { Effect.scoped( Effect.gen(function* () { const { factory, sockets } = yield* makeFactory(options); - const session = yield* factory.connect(PREPARED); + const session = yield* factory.connect(PREPARED, Effect.succeed(PREPARED)); const ready = yield* Effect.forkChild(session.ready); const socket = yield* awaitSocket(sockets); socket.open(); @@ -560,7 +877,7 @@ describe("RpcSessionFactory", () => { Effect.scoped( Effect.gen(function* () { const { factory, sockets } = yield* makeFactory({ environmentThemes: true }); - const session = yield* factory.connect(PREPARED); + const session = yield* factory.connect(PREPARED, Effect.succeed(PREPARED)); const readyFiber = yield* Effect.forkChild(session.ready); const socket = yield* awaitSocket(sockets); socket.open(); @@ -657,7 +974,7 @@ describe("RpcSessionFactory", () => { environmentThemes: true, usageLimitSources: true, }); - const session = yield* factory.connect(PREPARED); + const session = yield* factory.connect(PREPARED, Effect.succeed(PREPARED)); const readyFiber = yield* Effect.forkChild(session.ready); const socket = yield* awaitSocket(sockets); socket.open(); @@ -769,7 +1086,7 @@ describe("RpcSessionFactory", () => { Effect.scoped( Effect.gen(function* () { const { factory, sockets } = yield* makeFactory(); - const session = yield* factory.connect(PREPARED); + const session = yield* factory.connect(PREPARED, Effect.succeed(PREPARED)); const readyFiber = yield* Effect.forkChild(session.ready); const socket = yield* awaitSocket(sockets); socket.open(); @@ -799,7 +1116,7 @@ describe("RpcSessionFactory", () => { Effect.scoped( Effect.gen(function* () { const { factory, sockets } = yield* makeFactory({ environmentThemes: true }); - const firstSession = yield* factory.connect(PREPARED); + const firstSession = yield* factory.connect(PREPARED, Effect.succeed(PREPARED)); const firstReady = yield* Effect.forkChild(firstSession.ready); const firstSocket = yield* awaitSocket(sockets); firstSocket.open(); @@ -909,7 +1226,7 @@ describe("RpcSessionFactory", () => { label: "Recovered environment", }, } satisfies ServerConfigType; - const secondSession = yield* factory.connect(PREPARED); + const secondSession = yield* factory.connect(PREPARED, Effect.succeed(PREPARED)); const secondReady = yield* Effect.forkChild(secondSession.ready); const secondSocket = yield* awaitSocket(sockets, 1); secondSocket.open(); @@ -1018,7 +1335,7 @@ describe("RpcSessionFactory", () => { Effect.scoped( Effect.gen(function* () { const { factory, sockets } = yield* makeFactory(); - const session = yield* factory.connect(PREPARED); + const session = yield* factory.connect(PREPARED, Effect.succeed(PREPARED)); const readyFiber = yield* Effect.forkChild(session.ready); const socket = yield* awaitSocket(sockets); socket.open(); @@ -1057,7 +1374,7 @@ describe("RpcSessionFactory", () => { it.effect("rejects a server config for a different environment", () => Effect.gen(function* () { const { factory, sockets } = yield* makeFactory(); - const session = yield* factory.connect(PREPARED); + const session = yield* factory.connect(PREPARED, Effect.succeed(PREPARED)); const readyFiber = yield* Effect.forkChild(Effect.flip(session.ready)); const configFiber = yield* session .subscribeServerConfig({}) @@ -1088,7 +1405,7 @@ describe("RpcSessionFactory", () => { it.effect("closes the session after 15 seconds without a frame from the server", () => Effect.gen(function* () { const { factory, sockets } = yield* makeFactory(); - const session = yield* factory.connect(PREPARED); + const session = yield* factory.connect(PREPARED, Effect.succeed(PREPARED)); const readyFiber = yield* Effect.forkChild(session.ready); const closedFiber = yield* Effect.forkChild(Effect.flip(session.closed)); const socket = yield* awaitSocket(sockets); @@ -1118,7 +1435,7 @@ describe("RpcSessionFactory", () => { it.effect("keeps reading replies after closing a stream with a full buffer", () => Effect.gen(function* () { const { factory, sockets } = yield* makeFactory(); - const session = yield* factory.connect(PREPARED); + const session = yield* factory.connect(PREPARED, Effect.succeed(PREPARED)); const readyFiber = yield* Effect.forkChild(session.ready); const socket = yield* awaitSocket(sockets); socket.open(); @@ -1162,7 +1479,7 @@ describe("RpcSessionFactory", () => { it.effect("reaches ready when a newer server sends unknown config members", () => Effect.gen(function* () { const { factory, sockets } = yield* makeFactory(); - const session = yield* factory.connect(PREPARED); + const session = yield* factory.connect(PREPARED, Effect.succeed(PREPARED)); const readyFiber = yield* Effect.forkChild(session.ready); const socket = yield* awaitSocket(sockets); socket.open(); @@ -1197,7 +1514,7 @@ describe("RpcSessionFactory", () => { Effect.scoped( Effect.gen(function* () { const { factory, sockets } = yield* makeFactory(); - const session = yield* factory.connect(PREPARED); + const session = yield* factory.connect(PREPARED, Effect.succeed(PREPARED)); const readyFiber = yield* Effect.forkChild(session.ready); const socket = yield* awaitSocket(sockets); @@ -1243,15 +1560,18 @@ describe("RpcSessionFactory", () => { const error = yield* Effect.scoped( Effect.gen(function* () { - const session = yield* factory.connect({ - ...PREPARED, - target: relay - ? new RelayConnectionTarget({ - environmentId: TARGET.environmentId, - label: TARGET.label, - }) - : TARGET, - }); + const session = yield* factory.connect( + { + ...PREPARED, + target: relay + ? new RelayConnectionTarget({ + environmentId: TARGET.environmentId, + label: TARGET.label, + }) + : TARGET, + }, + Effect.succeed(PREPARED), + ); const readyFiber = yield* Effect.forkChild(Effect.flip(session.ready)); yield* awaitSocket(sockets); diff --git a/packages/client-runtime/src/rpc/session.ts b/packages/client-runtime/src/rpc/session.ts index 62410d7c6c5b80f23011aff49ba4adbe26cff512..f1d1c918f5810689052165c8b2454417c463846e 100644 --- a/packages/client-runtime/src/rpc/session.ts +++ b/packages/client-runtime/src/rpc/session.ts @@ -10,13 +10,14 @@ import * as Deferred from "effect/Deferred"; import * as Effect from "effect/Effect"; import * as Equal from "effect/Equal"; import * as Exit from "effect/Exit"; +import * as Fiber from "effect/Fiber"; import * as Layer from "effect/Layer"; import * as Option from "effect/Option"; import * as PubSub from "effect/PubSub"; import * as Ref from "effect/Ref"; import * as Schedule from "effect/Schedule"; import * as Schema from "effect/Schema"; -import type * as Scope from "effect/Scope"; +import * as Scope from "effect/Scope"; import * as Stream from "effect/Stream"; import type * as Rpc from "effect/rpc/Rpc"; import * as RpcClient from "effect/rpc/RpcClient"; @@ -26,11 +27,7 @@ import * as Socket from "effect/socket/Socket"; import { makeWsRpcProtocolClient, PING_TIMEOUT, type WsRpcProtocolClient } from "./protocol.ts"; import { NETWORK_BLOCKING_HINT } from "../errors/network.ts"; -import type { - ConnectionAttemptError, - ConnectionTransientError, - PreparedConnection, -} from "../connection/model.ts"; +import type { ConnectionAttemptError, PreparedConnection } from "../connection/model.ts"; import { ConnectionBlockedError, ConnectionTransientError as ConnectionTransientErrorClass, @@ -56,6 +53,7 @@ export interface RpcSession { } export interface RpcSessionOptions { + readonly terminalSocket?: (prepared: PreparedConnection) => Socket.Socket | undefined; readonly environmentThemes?: boolean; readonly usageLimitSources?: boolean; /** This client answers /usage-limits itself, so the server may advertise it. */ @@ -67,6 +65,7 @@ export class RpcSessionFactory extends Context.Service< { readonly connect: ( connection: PreparedConnection, + prepareTerminalConnection: Effect.Effect, ) => Effect.Effect; } >()("@t3tools/client-runtime/rpc/session/RpcSessionFactory") {} @@ -159,7 +158,10 @@ export const make = Effect.fn("RpcSessionFactory.make")(function* ( ...(options.usageLimitsCommand === true ? { usageLimitsCommand: true } : {}), }; - const connect = Effect.fnUntraced(function* (connection: PreparedConnection) { + const connect = Effect.fnUntraced(function* ( + connection: PreparedConnection, + prepareTerminalConnection: Effect.Effect, + ) { const networkHint = connection.target._tag === "RelayConnectionTarget" ? ` ${NETWORK_BLOCKING_HINT}` : ""; const mapRpcError = (error: Parameters[0]) => @@ -168,56 +170,142 @@ export const make = Effect.fn("RpcSessionFactory.make")(function* ( "connection.environment.id": connection.environmentId, }); - const connected = yield* Deferred.make(); - const disconnected = yield* Deferred.make(); - // Set when the socket closes because the server stopped answering, so the failure says so - // instead of looking like the server closed the connection. - const pingTimedOut = yield* Ref.make(false); - const hooks = RpcClient.ConnectionHooks.of({ - onConnect: Deferred.succeed(connected, undefined).pipe(Effect.asVoid), - onDisconnect: Effect.all([Deferred.isDone(connected), Ref.get(pingTimedOut)]).pipe( - Effect.flatMap(([wasConnected, timedOut]) => - Deferred.fail( - disconnected, - new ConnectionTransientErrorClass({ - reason: "transport", - detail: `${ - !wasConnected - ? `${connection.label} could not establish a WebSocket connection.` - : timedOut - ? `${connection.label} stopped responding.` - : `${connection.label} disconnected.` - }${networkHint}`, - }), + const scope = yield* Effect.scope; + const disconnected = yield* Deferred.make(); + const openTransport = Effect.fnUntraced(function* ( + prepared: PreparedConnection, + terminal = false, + ) { + const connected = yield* Deferred.make(); + const pingTimedOut = yield* Ref.make(false); + const hooks = RpcClient.ConnectionHooks.of({ + onConnect: Deferred.succeed(connected, undefined).pipe(Effect.asVoid), + onDisconnect: Effect.all([Deferred.isDone(connected), Ref.get(pingTimedOut)]).pipe( + Effect.flatMap(([wasConnected, timedOut]) => + Deferred.fail( + disconnected, + new ConnectionTransientErrorClass({ + reason: "transport", + detail: `${ + !wasConnected + ? `${connection.label} could not establish a WebSocket connection.` + : timedOut + ? `${connection.label} stopped responding.` + : `${connection.label} disconnected.` + }${networkHint}`, + }), + ), ), + Effect.asVoid, ), - Effect.asVoid, - ), + }); + const localSocket = terminal ? options.terminalSocket?.(prepared) : undefined; + const layerSocket = localSocket + ? Layer.succeed(Socket.Socket, localSocket) + : Socket.layerWebSocket(prepared.socketUrl, { + openTimeout: SOCKET_OPEN_TIMEOUT, + }).pipe(Layer.provide(Layer.succeed(Socket.WebSocketConstructor, webSocketConstructor))); + const layerProtocol = Layer.effect( + RpcClient.Protocol, + RpcClient.makeProtocolSocket({ + pingTimeout: PING_TIMEOUT, + retryTransientErrors: false, + retryPolicy: Schedule.recurs(0), + onPingTimeout: Ref.set(pingTimedOut, true), + }), + ).pipe( + Layer.provide( + Layer.mergeAll( + layerSocket, + RpcSerialization.layerJson, + Layer.succeed(RpcClient.ConnectionHooks, hooks), + ), + ), + ); + const protocolContext = yield* Layer.build(layerProtocol).pipe( + Effect.withSpan("environment.websocket.connect"), + ); + const client = yield* makeWsRpcProtocolClient.pipe(Effect.provide(protocolContext)); + return { client, connected: Deferred.await(connected) }; }); - const layerSocket = Socket.layerWebSocket(connection.socketUrl, { - openTimeout: SOCKET_OPEN_TIMEOUT, - }).pipe(Layer.provide(Layer.succeed(Socket.WebSocketConstructor, webSocketConstructor))); - const layerProtocol = Layer.effect( - RpcClient.Protocol, - RpcClient.makeProtocolSocket({ - pingTimeout: PING_TIMEOUT, - retryTransientErrors: false, - retryPolicy: Schedule.recurs(0), - onPingTimeout: Ref.set(pingTimedOut, true), - }), - ).pipe( - Layer.provide( - Layer.mergeAll( - layerSocket, - RpcSerialization.layerJson, - Layer.succeed(RpcClient.ConnectionHooks, hooks), - ), + const transport = yield* openTransport(connection); + const protocolClient = transport.client; + // Canceling one command must not abandon the connection shared by other terminals. + const terminalTransport = yield* prepareTerminalConnection.pipe( + Effect.flatMap((prepared) => openTransport(prepared, true)), + Effect.tapError((error) => Deferred.fail(disconnected, error)), + Effect.map(({ client }) => client), + Effect.mapError( + (cause) => + new RpcClientError.RpcClientError({ + reason: new RpcClientError.RpcClientDefect({ + message: `${connection.label} could not establish a WebSocket connection.`, + cause, + }), + }), ), + Scope.provide(scope), + Effect.forkIn(scope), + Effect.cached, ); - const protocolContext = yield* Layer.build(layerProtocol).pipe( - Effect.withSpan("environment.websocket.connect"), - ); - const protocolClient = yield* makeWsRpcProtocolClient.pipe(Effect.provide(protocolContext)); + const terminalClient = Effect.flatMap(terminalTransport, Fiber.join); + const terminalStream = < + Tag extends + | typeof WS_METHODS.terminalAttach + | typeof WS_METHODS.terminalObserve + | typeof WS_METHODS.subscribeTerminalEvents, + >( + tag: Tag, + ): WsRpcProtocolClient[Tag] => { + type Method = WsRpcProtocolClient[Tag]; + type Input = Parameters[0]; + type Options = Parameters[1]; + + // The tag correlates its payload and result; `asQueue` selects the RPC overload at runtime. + return ((input: Input, options?: Options) => { + const stream = Stream.unwrap( + Effect.map(terminalClient, (client) => { + const method = client[tag] as ( + input: Input, + options: Options & { readonly asQueue: false }, + ) => Stream.Stream; + return method(input, { ...options, asQueue: false }); + }), + ); + return options?.asQueue + ? Stream.toQueue(stream, { capacity: options.streamBufferSize ?? 16 }) + : stream; + }) as Method; + }; + // Metadata stays on the main socket: sidebars observe it without opening a terminal. + const client: WsRpcProtocolClient = { + ...protocolClient, + [WS_METHODS.terminalOpen]: (input, options) => + Effect.flatMap(terminalClient, (client) => client[WS_METHODS.terminalOpen](input, options)), + [WS_METHODS.terminalWrite]: (input, options) => + Effect.flatMap(terminalClient, (client) => + client[WS_METHODS.terminalWrite](input, options), + ), + [WS_METHODS.terminalResize]: (input, options) => + Effect.flatMap(terminalClient, (client) => + client[WS_METHODS.terminalResize](input, options), + ), + [WS_METHODS.terminalClear]: (input, options) => + Effect.flatMap(terminalClient, (client) => + client[WS_METHODS.terminalClear](input, options), + ), + [WS_METHODS.terminalRestart]: (input, options) => + Effect.flatMap(terminalClient, (client) => + client[WS_METHODS.terminalRestart](input, options), + ), + [WS_METHODS.terminalClose]: (input, options) => + Effect.flatMap(terminalClient, (client) => + client[WS_METHODS.terminalClose](input, options), + ), + [WS_METHODS.terminalAttach]: terminalStream(WS_METHODS.terminalAttach), + [WS_METHODS.terminalObserve]: terminalStream(WS_METHODS.terminalObserve), + [WS_METHODS.subscribeTerminalEvents]: terminalStream(WS_METHODS.subscribeTerminalEvents), + }; const initialConfigDeferred = yield* Deferred.make(); const serverConfigExit = yield* Deferred.make(); const configSubscriptionClosed = yield* Deferred.make(); @@ -383,10 +471,10 @@ export const make = Effect.fn("RpcSessionFactory.make")(function* ( ); return { - client: protocolClient, + client, initialConfig, subscribeServerConfig, - ready: Deferred.await(connected).pipe( + ready: transport.connected.pipe( Effect.andThen(initialConfig), Effect.asVoid, Effect.raceFirst(Deferred.await(disconnected)), diff --git a/packages/contracts/src/desktopBootstrap.ts b/packages/contracts/src/desktopBootstrap.ts index 8a3f379df329013248fd08a397f0f6974d0b6141..77771e5ce36b0aa6bc6ca46f4c16f2a28fb65e11 100644 --- a/packages/contracts/src/desktopBootstrap.ts +++ b/packages/contracts/src/desktopBootstrap.ts @@ -35,6 +35,8 @@ export const DesktopBackendBootstrap = Schema.Struct({ desktopBrowserFd: Schema.optionalKey(PositiveInt), /** Server -> desktop: commands for those tabs. */ desktopBrowserControlFd: Schema.optionalKey(PositiveInt), + desktopTerminalFd: Schema.optionalKey(PositiveInt), + desktopTerminalControlFd: Schema.optionalKey(PositiveInt), resourceMonitorPath: Schema.optionalKey(TrimmedNonEmptyString), }); diff --git a/packages/contracts/src/desktopTerminal.ts b/packages/contracts/src/desktopTerminal.ts new file mode 100644 index 0000000000000000000000000000000000000000..70f82c0a479cc9382496da4db9b5d1a754ebb6b8 --- /dev/null +++ b/packages/contracts/src/desktopTerminal.ts @@ -0,0 +1,25 @@ +import * as Schema from "effect/Schema"; + +export const DesktopTerminalRequest = Schema.Union([ + Schema.Struct({ + type: Schema.Literal("open"), + id: Schema.String, + environmentId: Schema.String, + url: Schema.String, + }), + Schema.Struct({ type: Schema.Literal("message"), id: Schema.String, data: Schema.String }), + Schema.Struct({ type: Schema.Literal("close"), id: Schema.String }), +]); +export type DesktopTerminalRequest = typeof DesktopTerminalRequest.Type; + +export const DesktopTerminalResponse = Schema.Union([ + Schema.Struct({ type: Schema.Literal("ready"), environmentId: Schema.String }), + Schema.Struct({ type: Schema.Literal("message"), id: Schema.String, data: Schema.String }), + Schema.Struct({ type: Schema.Literal("close"), id: Schema.String }), +]); +export type DesktopTerminalResponse = typeof DesktopTerminalResponse.Type; + +export interface DesktopTerminalConnection { + readonly send: (data: string) => void; + readonly close: () => void; +} diff --git a/packages/contracts/src/index.ts b/packages/contracts/src/index.ts index 600d503ade39b817747413cb841ae09a1d0dfb13..f2e3689fe30dcb62288fdd8ad94ce58b4006fb4b 100644 --- a/packages/contracts/src/index.ts +++ b/packages/contracts/src/index.ts @@ -65,3 +65,4 @@ export * from "./rpc.ts"; export * from "./worktreeSetup.ts"; export * from "./secretRequest.ts"; export * from "./clientRpcPermissions.ts"; +export * from "./desktopTerminal.ts"; diff --git a/packages/contracts/src/ipc.ts b/packages/contracts/src/ipc.ts index bea353c2bef4308c7f8e1cdb6aa2ea8701975c6c..47b16efc36359c88f0eb1885097a357daef2be8b 100644 --- a/packages/contracts/src/ipc.ts +++ b/packages/contracts/src/ipc.ts @@ -1123,6 +1123,15 @@ export const SystemSettingsPaneSchema = Schema.Literals(["full-disk-access"]); export type SystemSettingsPane = typeof SystemSettingsPaneSchema.Type; export interface DesktopBridge { + supportsLocalTerminal?: (input: { + readonly httpBaseUrl: string; + readonly environmentId: string; + }) => boolean; + openLocalTerminal?: ( + input: { readonly environmentId: string; readonly url: string }, + onMessage: (data: string) => void, + onClose: () => void, + ) => import("./desktopTerminal.ts").DesktopTerminalConnection; getAppBranding: () => DesktopAppBranding | null; /** Absolute path of a dropped or picked file; absent on desktop builds predating it. */ getPathForFile?: (file: File) => string; diff --git a/packages/contracts/src/rpc.ts b/packages/contracts/src/rpc.ts index 61304c52f860fa7f608ce07032d5d6b6771ca145..508c56813c3eea9771befccfcbfa93fd5d6977e5 100644 --- a/packages/contracts/src/rpc.ts +++ b/packages/contracts/src/rpc.ts @@ -1875,6 +1875,18 @@ export class RpcScopeAuthorization extends RpcMiddleware.Service