From 134f2f0765bf03e98da10fc3f7c1614dbe99a9fe Mon Sep 17 00:00:00 2001 From: clover caruso Date: Tue, 13 Oct 2026 11:10:31 -0700 Subject: [PATCH] feat: isolate terminal transport and use Electron IPC locally Clover uses interactive terminals for everyday inspection, including vi. Keep terminal input, output, and control independent of chat traffic. Use Electron IPC for the desktop local environment and a lazy authenticated WebSocket for remote environments, preserving destination permissions and supervised reconnects. Keep metadata on the primary connection and bound ordinary output batches without splitting sequenced events or snapshots. Assisted-by: gpt-6.1-sol --- .../backend/DesktopBackendConfiguration.ts | 2 + .../src/backend/DesktopBackendManager.test.ts | 7 + .../src/backend/DesktopBackendManager.ts | 37 ++ .../src/backend/DesktopBackendPool.test.ts | 7 + .../desktop/src/backend/DesktopBackendPool.ts | 2 + .../src/backend/DesktopTerminalHost.test.ts | 168 ++++++++ .../src/backend/DesktopTerminalHost.ts | 176 +++++++++ apps/desktop/src/ipc/channels.ts | 1 + apps/desktop/src/main.ts | 2 + apps/desktop/src/preload.ts | 25 ++ apps/server/src/cli/config.ts | 2 + apps/server/src/config.ts | 2 + apps/server/src/server.ts | 2 + .../src/terminal/DesktopTerminalRpc.test.ts | 129 ++++++ .../server/src/terminal/DesktopTerminalRpc.ts | 203 ++++++++++ .../src/terminal/OutputProtocol.test.ts | 130 +++++- apps/server/src/terminal/OutputProtocol.ts | 45 ++- apps/server/src/terminal/Rpc.ts | 45 +++ apps/server/src/ws.ts | 32 +- apps/web/src/connection/runtime.ts | 2 + docs/internals/terminal-runtime.md | 7 + .../client-runtime/src/connection/driver.ts | 5 +- .../client-runtime/src/rpc/desktopTerminal.ts | 67 ++++ packages/client-runtime/src/rpc/index.ts | 1 + .../client-runtime/src/rpc/session.test.ts | 370 ++++++++++++++++-- packages/client-runtime/src/rpc/session.ts | 196 +++++++--- packages/contracts/src/desktopBootstrap.ts | 2 + packages/contracts/src/desktopTerminal.ts | 25 ++ packages/contracts/src/index.ts | 1 + packages/contracts/src/ipc.ts | 9 + packages/contracts/src/rpc.ts | 25 +- 31 files changed, 1604 insertions(+), 123 deletions(-) create mode 100644 apps/desktop/src/backend/DesktopTerminalHost.test.ts create mode 100644 apps/desktop/src/backend/DesktopTerminalHost.ts create mode 100644 apps/server/src/terminal/DesktopTerminalRpc.test.ts create mode 100644 apps/server/src/terminal/DesktopTerminalRpc.ts create mode 100644 apps/server/src/terminal/Rpc.ts create mode 100644 packages/client-runtime/src/rpc/desktopTerminal.ts create mode 100644 packages/contracts/src/desktopTerminal.ts 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