authorclover caruso <git@paperclover.net> 2026-10-13 11:10:31-07:00
committerclover caruso <git@paperclover.net> 2026-10-16 16:52:04-07:00
log134f2f0765bf03e98da10fc3f7c1614dbe99a9fe
treea62bb6ffe28d7a83dd2c4d3881665d59133b21e4
parente05b6a353e4701d919165a7a32ee30b2ec61dce0
signature Signed by SSH key SHA256:cOKiuRFOeSRxne6EWgHtdQQSlBxjOXm2hOCFnCdLQbQ (~clover)

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

31 files changed, 1604 insertions(+), 123 deletions(-)

apps/desktop/src/backend/DesktopBackendConfiguration.ts+2
......@@ -577,6 +577,8 @@ const resolvePrimaryStartConfig = Effect.fn("desktop.backendConfiguration.resolv
577577 desktopTelemetryControlFd: 5,
578578 desktopBrowserFd: 6,
579579 desktopBrowserControlFd: 7,
580 desktopTerminalFd: 8,
581 desktopTerminalControlFd: 9,
580582 ...Option.match(input.resourceMonitorPath, {
581583 onNone: () => ({}),
582584 onSome: (resourceMonitorPath) => ({ resourceMonitorPath }),
apps/desktop/src/backend/DesktopBackendManager.test.ts+7
......@@ -1,3 +1,4 @@
1import * as DesktopTerminalHost from "./DesktopTerminalHost.ts";
12import {
23 SERVER_EXIT_CODE_STATE_DIR_OWNED,
34 DesktopBackendBootstrap,
......@@ -176,6 +177,12 @@ function makeTestInstance(input: MakeInstanceInput) {
176177 updateCancellations: Stream.empty,
177178 ...input.desktopTelemetryPublisher,
178179 }),
180 Layer.succeed(DesktopTerminalHost.DesktopTerminalHost, {
181 requests: Stream.empty,
182 handleLine: () => Effect.void,
183 reset: Effect.void,
184 configure: () => Effect.void,
185 }),
179186 DesktopBrowserHost.layer,
180187 DesktopWslEnvironment.layerTest(
181188 input.pruneRuntimes === undefined ? {} : { pruneRuntimes: input.pruneRuntimes },
apps/desktop/src/backend/DesktopBackendManager.ts+37
......@@ -53,6 +53,7 @@ import { waitForHttpReady as waitForHttpReadyShared } from "@t3tools/shared/http
5353
5454import * as DesktopObservability from "../app/DesktopObservability.ts";
5555import * as DesktopTelemetryPublisher from "../telemetry/DesktopTelemetryPublisher.ts";
56import * as DesktopTerminalHost from "./DesktopTerminalHost.ts";
5657import * as DesktopBrowserHost from "../preview/DesktopBrowserHost.ts";
5758import * as DesktopWslEnvironment from "../wsl/DesktopWslEnvironment.ts";
5859
......@@ -226,6 +227,9 @@ export type BackendProcessError = typeof BackendProcessError.Type;
226227
227228interface RunBackendProcessOptions extends DesktopBackendStartConfig {
228229 readonly desktopTelemetryStream: Stream.Stream<Uint8Array>;
230 readonly desktopTerminalStream?: Stream.Stream<Uint8Array>;
231 readonly onDesktopTerminalClosed?: Effect.Effect<void>;
232 readonly onDesktopTerminalLine?: (line: string) => Effect.Effect<void>;
229233 /** Events for the desktop's browser tabs, written to `desktopBrowserFd`. */
230234 readonly desktopBrowserStream?: Stream.Stream<Uint8Array>;
231235 /** Each line the backend writes to `desktopBrowserControlFd`. */
......@@ -494,6 +498,17 @@ export const runBackendProcess = Effect.fn("runBackendProcess")(function* (
494498 additionalFds[`fd${options.bootstrap.desktopBrowserControlFd}`] = { type: "output" };
495499 }
496500 }
501 if (
502 options.bootstrap.desktopTerminalFd !== undefined &&
503 options.bootstrap.desktopTerminalControlFd !== undefined &&
504 options.desktopTerminalStream
505 ) {
506 additionalFds[`fd${options.bootstrap.desktopTerminalFd}`] = {
507 type: "input",
508 stream: options.desktopTerminalStream,
509 };
510 additionalFds[`fd${options.bootstrap.desktopTerminalControlFd}`] = { type: "output" };
511 }
497512 const command = ChildProcess.make(options.executablePath, options.args, {
498513 cwd: options.cwd,
499514 env: options.env,
......@@ -563,6 +578,16 @@ export const runBackendProcess = Effect.fn("runBackendProcess")(function* (
563578 Effect.forkScoped,
564579 );
565580 }
581 if (options.bootstrap.desktopTerminalControlFd !== undefined && options.onDesktopTerminalLine) {
582 yield* handle.getOutputFd(options.bootstrap.desktopTerminalControlFd).pipe(
583 Stream.decodeText(),
584 Stream.splitLines,
585 Stream.runForEach(options.onDesktopTerminalLine),
586 Effect.catchCause(() => Effect.void),
587 Effect.ensuring(options.onDesktopTerminalClosed ?? Effect.void),
588 Effect.forkScoped,
589 );
590 }
566591 if (
567592 options.bootstrap.desktopBrowserControlFd !== undefined &&
568593 options.onDesktopBrowserCommand !== undefined
......@@ -683,6 +708,7 @@ export const makeBackendInstance = Effect.fn("makeBackendInstance")(function* (
683708 | HttpClient.HttpClient
684709 | DesktopObservability.DesktopBackendOutputLogFactory
685710 | DesktopTelemetryPublisher.DesktopTelemetryPublisher
711 | DesktopTerminalHost.DesktopTerminalHost
686712 | DesktopBrowserHost.DesktopBrowserHost
687713 | DesktopWslEnvironment.DesktopWslEnvironment
688714 | Scope.Scope
......@@ -692,6 +718,7 @@ export const makeBackendInstance = Effect.fn("makeBackendInstance")(function* (
692718 const backendOutputLogFactory = yield* DesktopObservability.DesktopBackendOutputLogFactory;
693719 const backendOutputLog = yield* backendOutputLogFactory.forInstance(spec.id);
694720 const desktopTelemetryPublisher = yield* DesktopTelemetryPublisher.DesktopTelemetryPublisher;
721 const desktopTerminalHost = yield* DesktopTerminalHost.DesktopTerminalHost;
695722 const desktopBrowserHost = yield* DesktopBrowserHost.DesktopBrowserHost;
696723 const wslEnvironment = yield* DesktopWslEnvironment.DesktopWslEnvironment;
697724 const spawner = yield* ChildProcessSpawner.ChildProcessSpawner;
......@@ -954,10 +981,15 @@ export const makeBackendInstance = Effect.fn("makeBackendInstance")(function* (
954981 );
955982 });
956983
984 if (config.value.bootstrap.desktopTerminalFd !== undefined)
985 yield* desktopTerminalHost.configure(config.value.httpBaseUrl.href);
957986 const program = runBackendProcess({
958987 ...config.value,
959988 desktopTelemetryStream: desktopTelemetryPublisher.encoded,
960989 // Only a bootstrap that names the browser fds (the local primary) gets them.
990 desktopTerminalStream: desktopTerminalHost.requests,
991 onDesktopTerminalLine: desktopTerminalHost.handleLine,
992 onDesktopTerminalClosed: desktopTerminalHost.reset,
961993 desktopBrowserStream: desktopBrowserHost.events,
962994 onDesktopBrowserCommand: desktopBrowserHost.handleCommandLine,
963995 onDesktopTelemetryControl: (message) =>
......@@ -1027,6 +1059,11 @@ export const makeBackendInstance = Effect.fn("makeBackendInstance")(function* (
10271059 onSuccess: (exit) => finalizeRun(exit.reason, Option.getOrUndefined(exit.code)),
10281060 }),
10291061 Effect.ensuring(Scope.close(runScope, Exit.void).pipe(Effect.ignore)),
1062 Effect.ensuring(
1063 config.value.bootstrap.desktopTerminalFd === undefined
1064 ? Effect.void
1065 : desktopTerminalHost.reset,
1066 ),
10301067 );
10311068
10321069 const fiber = yield* Effect.forkIn(program, parentScope);
apps/desktop/src/backend/DesktopBackendPool.test.ts+7
......@@ -1,3 +1,4 @@
1import * as DesktopTerminalHost from "./DesktopTerminalHost.ts";
12import { assert, describe, it } from "@effect/vitest";
23import * as Duration from "effect/Duration";
34import * as Effect from "effect/Effect";
......@@ -77,6 +78,12 @@ function layerPool(labelRef: Ref.Ref<string>): Layer.Layer<DesktopBackendPool.De
7778 updateCommits: Stream.empty,
7879 updateCancellations: Stream.empty,
7980 }),
81 Layer.succeed(DesktopTerminalHost.DesktopTerminalHost, {
82 requests: Stream.empty,
83 handleLine: () => Effect.void,
84 reset: Effect.void,
85 configure: () => Effect.void,
86 }),
8087 DesktopBrowserHost.layer,
8188 Layer.succeed(DesktopBackendConfiguration.DesktopBackendConfiguration, {
8289 resolvePrimary: Effect.die("unexpected primary config resolve"),
apps/desktop/src/backend/DesktopBackendPool.ts+2
......@@ -98,6 +98,7 @@ import * as DesktopBackendManager from "./DesktopBackendManager.ts";
9898import * as DesktopObservability from "../app/DesktopObservability.ts";
9999import * as DesktopAppSettings from "../settings/DesktopAppSettings.ts";
100100import * as DesktopTelemetryPublisher from "../telemetry/DesktopTelemetryPublisher.ts";
101import * as DesktopTerminalHost from "./DesktopTerminalHost.ts";
101102import * as DesktopBrowserHost from "../preview/DesktopBrowserHost.ts";
102103import * as DesktopWindow from "../window/DesktopWindow.ts";
103104import * as DesktopWslEnvironment from "../wsl/DesktopWslEnvironment.ts";
......@@ -182,6 +183,7 @@ export type BackendInstanceFactoryRequirements =
182183 | HttpClient.HttpClient
183184 | DesktopObservability.DesktopBackendOutputLogFactory
184185 | DesktopTelemetryPublisher.DesktopTelemetryPublisher
186 | DesktopTerminalHost.DesktopTerminalHost
185187 | DesktopBrowserHost.DesktopBrowserHost
186188 | DesktopWslEnvironment.DesktopWslEnvironment;
187189
apps/desktop/src/backend/DesktopTerminalHost.test.ts created+168
......@@ -0,0 +1,168 @@
1import { describe, expect, it } from "@effect/vitest";
2import { vi } from "vite-plus/test";
3import { DESKTOP_TERMINAL_CHANNEL } from "../ipc/channels.ts";
4import * as NodeServices from "@effect/platform-node/NodeServices";
5import * as Effect from "effect/Effect";
6import * as Fiber from "effect/Fiber";
7import * as Layer from "effect/Layer";
8import * as Option from "effect/Option";
9import * as Stream from "effect/Stream";
10import * as DesktopEnvironment from "../app/DesktopEnvironment.ts";
11import * as DesktopConfig from "../app/DesktopConfig.ts";
12import * as ElectronWindow from "../electron/ElectronWindow.ts";
13import * as DesktopTerminalHost from "./DesktopTerminalHost.ts";
14
15const electron = await vi.hoisted(async () => {
16 const { EventEmitter } = await import("node:events");
17 const sender = Object.assign(new EventEmitter(), { mainFrame: { url: "t3code-dev://app/" } });
18 class Port extends EventEmitter {
19 messages: unknown[] = [];
20 closed = false;
21 postMessage(data: unknown) {
22 this.messages.push(data);
23 }
24 start() {}
25 close() {
26 if (!this.closed) {
27 this.closed = true;
28 this.emit("close");
29 }
30 }
31 }
32 return {
33 ipcMain: new EventEmitter(),
34 sender,
35 Port,
36 window: { webContents: sender, isDestroyed: () => false },
37 };
38});
39vi.mock("electron", () => ({
40 ipcMain: electron.ipcMain,
41 BrowserWindow: { getAllWindows: () => [electron.window] },
42}));
43
44const layerEnvironment = DesktopEnvironment.layer({
45 dirname: "/repo/apps/desktop/dist-electron",
46 homeDirectory: "/tmp/t3-terminal-test",
47 platform: "darwin",
48 processArch: "arm64",
49 appVersion: "0.0.45",
50 appPath: "/tmp/T3 Code.app",
51 isPackaged: false,
52 resourcesPath: "/tmp/resources",
53 runningUnderArm64Translation: false,
54}).pipe(
55 Layer.provide(
56 Layer.mergeAll(
57 NodeServices.layer,
58 DesktopConfig.layerTest({ VITE_DEV_SERVER_URL: "http://localhost:5733" }),
59 ),
60 ),
61);
62const layer = DesktopTerminalHost.layer.pipe(
63 Layer.provide(ElectronWindow.layer),
64 Layer.provide(layerEnvironment),
65);
66
67const event = (port: InstanceType<typeof electron.Port>, frame = electron.sender.mainFrame) => {
68 const replies: boolean[] = [];
69 let firstReply = false;
70 return {
71 sender: electron.sender,
72 senderFrame: frame,
73 ports: [port],
74 replies,
75 get returnValue() {
76 return firstReply;
77 },
78 set returnValue(value: boolean) {
79 replies.push(value);
80 if (replies.length === 1) firstReply = value;
81 },
82 };
83};
84const input = {
85 environmentId: "local-environment",
86 url: "ws://127.0.0.1:3773/ws?wsTicket=private",
87};
88const readRequest = (host: DesktopTerminalHost.DesktopTerminalHost["Service"]) =>
89 host.requests.pipe(
90 Stream.runHead,
91 Effect.map((data) => JSON.parse(new TextDecoder().decode(Option.getOrThrow(data)))),
92 Effect.forkScoped({ startImmediately: true }),
93 );
94
95describe("DesktopTerminalHost", () => {
96 it.effect("selects the native route before ready and binds the announced identity", () =>
97 Effect.gen(function* () {
98 const host = yield* DesktopTerminalHost.DesktopTerminalHost;
99 yield* host.configure("http://127.0.0.1:3773");
100 const probe = (environmentId: string, httpBaseUrl = "http://127.0.0.1:3773") => {
101 const request = event(new electron.Port());
102 electron.ipcMain.emit(`${DESKTOP_TERMINAL_CHANNEL}:available`, request, {
103 environmentId,
104 httpBaseUrl,
105 });
106 expect(request.replies).toHaveLength(1);
107 return request.returnValue;
108 };
109 expect(probe(input.environmentId)).toBe(true);
110 expect(probe(input.environmentId, "http://127.0.0.1:3774")).toBe(false);
111 yield* host.handleLine(JSON.stringify({ type: "ready", environmentId: input.environmentId }));
112 expect(probe("other-environment")).toBe(false);
113 yield* host.reset;
114 expect(probe(input.environmentId)).toBe(false);
115 }).pipe(Effect.provide(layer), Effect.scoped),
116 );
117
118 it.effect("forwards RPC frames through the private pipe and closes on navigation", () =>
119 Effect.gen(function* () {
120 const host = yield* DesktopTerminalHost.DesktopTerminalHost;
121 yield* host.configure("http://127.0.0.1:3773");
122 const port = new electron.Port();
123 const opening = yield* readRequest(host);
124 electron.ipcMain.emit(DESKTOP_TERMINAL_CHANNEL, event(port), input);
125 const request = yield* Fiber.join(opening);
126 expect(request).toEqual({ type: "open", id: request.id, ...input });
127 const data = '{"_tag":"Request","tag":"terminal.write"}';
128 const message = yield* readRequest(host);
129 port.emit("message", { data });
130 expect(yield* Fiber.join(message)).toEqual({ type: "message", id: request.id, data });
131 yield* host.handleLine(JSON.stringify({ type: "message", id: request.id, data: "response" }));
132 expect(port.messages).toEqual([{ type: "message", id: request.id, data: "response" }]);
133 const closing = yield* readRequest(host);
134 electron.sender.emit("did-start-navigation", { isMainFrame: false, isSameDocument: false });
135 expect(port.closed).toBe(false);
136 electron.sender.emit("did-start-navigation", { isMainFrame: true, isSameDocument: true });
137 expect(port.closed).toBe(false);
138 electron.sender.emit("did-start-navigation", { isMainFrame: true, isSameDocument: false });
139 expect(yield* Fiber.join(closing)).toEqual({ type: "close", id: request.id });
140 expect(port.closed).toBe(true);
141 }).pipe(Effect.provide(layer), Effect.scoped),
142 );
143
144 it.effect("rejects embedded frames, other windows and custom-scheme origins", () =>
145 Effect.gen(function* () {
146 const host = yield* DesktopTerminalHost.DesktopTerminalHost;
147 yield* host.configure("http://127.0.0.1:3773");
148 for (const frame of [{ url: "t3code-dev://app/" }, { url: "t3code-dev://evil/" }]) {
149 const port = new electron.Port();
150 electron.ipcMain.emit(DESKTOP_TERMINAL_CHANNEL, event(port, frame), input);
151 expect(port.closed).toBe(true);
152 }
153 const oldUrl = electron.sender.mainFrame.url;
154 electron.sender.mainFrame.url = "t3code-dev://evil/";
155 const navigated = new electron.Port();
156 electron.ipcMain.emit(DESKTOP_TERMINAL_CHANNEL, event(navigated), input);
157 expect(navigated.closed).toBe(true);
158 electron.sender.mainFrame.url = oldUrl;
159 const foreign = new electron.Port();
160 electron.ipcMain.emit(
161 DESKTOP_TERMINAL_CHANNEL,
162 { ...event(foreign), sender: { mainFrame: electron.sender.mainFrame } },
163 input,
164 );
165 expect(foreign.closed).toBe(true);
166 }).pipe(Effect.provide(layer), Effect.scoped),
167 );
168});
apps/desktop/src/backend/DesktopTerminalHost.ts created+176
......@@ -0,0 +1,176 @@
1import { DESKTOP_TERMINAL_CHANNEL } from "../ipc/channels.ts";
2import { DesktopTerminalResponse, DesktopTerminalRequest } from "@t3tools/contracts";
3import * as Context from "effect/Context";
4import * as Effect from "effect/Effect";
5import * as Layer from "effect/Layer";
6import * as PubSub from "effect/PubSub";
7import * as Schema from "effect/Schema";
8import * as Stream from "effect/Stream";
9import * as Electron from "electron";
10import { getDesktopUrl } from "../electron/ElectronProtocol.ts";
11import * as DesktopEnvironment from "../app/DesktopEnvironment.ts";
12import * as ElectronWindow from "../electron/ElectronWindow.ts";
13
14const encode = Schema.encodeSync(Schema.fromJsonString(DesktopTerminalRequest));
15const decode = Schema.decodeUnknownOption(Schema.fromJsonString(DesktopTerminalResponse));
16const decodeAvailability = Schema.decodeUnknownOption(
17 Schema.Struct({ httpBaseUrl: Schema.String, environmentId: Schema.String }),
18);
19const decodeOpen = Schema.decodeUnknownOption(
20 Schema.Struct({ environmentId: Schema.String, url: Schema.String }),
21);
22
23export class DesktopTerminalHost extends Context.Service<
24 DesktopTerminalHost,
25 {
26 readonly requests: Stream.Stream<Uint8Array>;
27 readonly handleLine: (line: string) => Effect.Effect<void>;
28 readonly reset: Effect.Effect<void>;
29 readonly configure: (httpBaseUrl: string) => Effect.Effect<void>;
30 }
31>()("@t3tools/desktop/backend/DesktopTerminalHost") {}
32
33export const layer = Layer.effect(
34 DesktopTerminalHost,
35 Effect.gen(function* () {
36 const environment = yield* DesktopEnvironment.DesktopEnvironment;
37 const appUrl = new URL(getDesktopUrl(environment.isDevelopment));
38 const windows = yield* ElectronWindow.ElectronWindow;
39 const requests = yield* PubSub.unbounded<Uint8Array>();
40 const run = Effect.runForkWith(yield* Effect.context<never>());
41 let nextId = 0;
42 const ports = new Map<string, Electron.MessagePortMain>();
43 let environmentId: string | null = null;
44 let backendOrigin: string | null = null;
45 const emit = (request: DesktopTerminalRequest) =>
46 run(PubSub.publish(requests, new TextEncoder().encode(`${encode(request)}\n`)));
47 const reset = Effect.sync(() => {
48 environmentId = null;
49 backendOrigin = null;
50 for (const port of ports.values()) {
51 port.postMessage({ type: "close" });
52 port.close();
53 }
54 ports.clear();
55 });
56 const trusted = (event: Electron.IpcMainEvent) => {
57 const window = Effect.runSync(windows.currentMainOrFirst);
58 return (
59 window._tag === "Some" &&
60 window.value.webContents === event.sender &&
61 event.senderFrame === event.sender.mainFrame &&
62 new URL(event.senderFrame.url).protocol === appUrl.protocol &&
63 new URL(event.senderFrame.url).host === appUrl.host
64 );
65 };
66 const onAvailable = (event: Electron.IpcMainEvent, input: unknown) => {
67 const value = decodeAvailability(input);
68 let available = false;
69 if (trusted(event) && value._tag === "Some" && backendOrigin !== null) {
70 try {
71 available =
72 new URL(value.value.httpBaseUrl).origin === backendOrigin &&
73 (environmentId === null || value.value.environmentId === environmentId);
74 } catch {}
75 }
76 event.returnValue = available;
77 };
78 const onOpen = (event: Electron.IpcMainEvent, input: unknown) => {
79 const port = event.ports[0];
80 if (!port) return;
81 const decoded = decodeOpen(input);
82 if (
83 !trusted(event) ||
84 decoded._tag === "None" ||
85 (environmentId !== null && decoded.value.environmentId !== environmentId)
86 ) {
87 port.postMessage({ type: "close" });
88 port.close();
89 return;
90 }
91 try {
92 const url = new URL(decoded.value.url);
93 url.protocol = url.protocol === "wss:" ? "https:" : "http:";
94 if (url.origin !== backendOrigin) {
95 port.postMessage({ type: "close" });
96 port.close();
97 return;
98 }
99 } catch {
100 port.postMessage({ type: "close" });
101 port.close();
102 return;
103 }
104 const id = String(++nextId);
105 ports.set(id, port);
106 const close = () => {
107 if (ports.get(id) !== port) return;
108 ports.delete(id);
109 emit({ type: "close", id });
110 port.close();
111 };
112 port.on("message", ({ data }: { data: unknown }) => {
113 if (typeof data === "string") emit({ type: "message", id, data });
114 else close();
115 });
116 port.on("close", close);
117 const onNavigation = (
118 details: Electron.Event<Electron.WebContentsDidStartNavigationEventParams>,
119 ) => {
120 if (details.isMainFrame && !details.isSameDocument) close();
121 };
122 event.sender.once("destroyed", close);
123 event.sender.on("did-start-navigation", onNavigation);
124 port.once("close", () => {
125 event.sender.removeListener("destroyed", close);
126 event.sender.removeListener("did-start-navigation", onNavigation);
127 });
128 emit({ type: "open", id, ...decoded.value });
129 port.start();
130 };
131 yield* Effect.acquireRelease(
132 Effect.sync(() => {
133 Electron.ipcMain.on(`${DESKTOP_TERMINAL_CHANNEL}:available`, onAvailable);
134 Electron.ipcMain.on(DESKTOP_TERMINAL_CHANNEL, onOpen);
135 }),
136 () =>
137 reset.pipe(
138 Effect.andThen(
139 Effect.sync(() => {
140 Electron.ipcMain.removeListener(`${DESKTOP_TERMINAL_CHANNEL}:available`, onAvailable);
141 Electron.ipcMain.removeListener(DESKTOP_TERMINAL_CHANNEL, onOpen);
142 }),
143 ),
144 ),
145 );
146 return DesktopTerminalHost.of({
147 requests: Stream.fromPubSub(requests),
148 configure: (url) =>
149 reset.pipe(
150 Effect.andThen(
151 Effect.sync(() => {
152 backendOrigin = new URL(url).origin;
153 }),
154 ),
155 ),
156 reset,
157 handleLine: (line) =>
158 Effect.sync(() => {
159 const response = decode(line);
160 if (response._tag === "None") return;
161 const value = response.value;
162 if (value.type === "ready") {
163 environmentId = value.environmentId;
164 return;
165 }
166 const port = ports.get(value.id);
167 if (!port) return;
168 port.postMessage(value);
169 if (value.type === "close") {
170 ports.delete(value.id);
171 port.close();
172 }
173 }),
174 });
175 }),
176);
apps/desktop/src/ipc/channels.ts+1
......@@ -123,3 +123,4 @@ export const RECEIVE_PROVIDER_AUTH_CALLBACK_CHANNEL = "desktop:receive-provider-
123123export const CANCEL_PROVIDER_AUTH_CALLBACK_CHANNEL = "desktop:cancel-provider-auth-callback";
124124export const TAKE_LEGACY_LOCAL_STORAGE_CHANNEL = "desktop:take-legacy-local-storage";
125125export const COMPLETE_LEGACY_LOCAL_STORAGE_CHANNEL = "desktop:complete-legacy-local-storage";
126export const DESKTOP_TERMINAL_CHANNEL = "desktop:terminal";
apps/desktop/src/main.ts+2
......@@ -43,6 +43,7 @@ import * as DesktopCliCommand from "./app/DesktopCliCommand.ts";
4343import * as DesktopApplicationMenu from "./window/DesktopApplicationMenu.ts";
4444import * as DesktopAssets from "./app/DesktopAssets.ts";
4545import * as DesktopBackendConfiguration from "./backend/DesktopBackendConfiguration.ts";
46import * as DesktopTerminalHost from "./backend/DesktopTerminalHost.ts";
4647import * as DesktopBackendPool from "./backend/DesktopBackendPool.ts";
4748import * as DesktopLocalEnvironmentAuth from "./backend/DesktopLocalEnvironmentAuth.ts";
4849import * as DesktopNetworkInterfaces from "./backend/DesktopNetworkInterfaces.ts";
......@@ -193,6 +194,7 @@ const layerDesktopAppActivation = DesktopAppActivation.layer.pipe(
193194// WSL second instance gets registered later in the migration. See
194195// DesktopBackendPool.ts header for the full rollout plan.
195196const layerDesktopBackend = DesktopBackendPool.layer.pipe(
197 Layer.provideMerge(DesktopTerminalHost.layer),
196198 Layer.provideMerge(DesktopAppIdentity.layer),
197199 Layer.provideMerge(DesktopBackendConfiguration.layer),
198200 Layer.provideMerge(DesktopWslEnvironment.layer),
apps/desktop/src/preload.ts+25
......@@ -77,6 +77,31 @@ function unwrapEnsureSshEnvironmentResult(result: unknown) {
7777}
7878
7979contextBridge.exposeInMainWorld("desktopBridge", {
80 supportsLocalTerminal: (input) =>
81 ipcRenderer.sendSync(`${IpcChannels.DESKTOP_TERMINAL_CHANNEL}:available`, input) === true,
82 openLocalTerminal: (input, onMessage, onClose) => {
83 const { port1, port2 } = new MessageChannel();
84 let closed = false;
85 const close = () => {
86 if (closed) return;
87 closed = true;
88 port1.postMessage(null);
89 port1.close();
90 onClose();
91 };
92 port1.addEventListener("message", ({ data }) => {
93 if (data?.type === "message" && typeof data.data === "string") onMessage(data.data);
94 else if (data?.type === "close") close();
95 });
96 ipcRenderer.postMessage(IpcChannels.DESKTOP_TERMINAL_CHANNEL, input, [port2]);
97 port1.start();
98 return {
99 send: (data) => {
100 if (!closed) port1.postMessage(data);
101 },
102 close,
103 };
104 },
80105 getAppBranding: () => {
81106 const result = ipcRenderer.sendSync(IpcChannels.GET_APP_BRANDING_CHANNEL);
82107 if (typeof result !== "object" || result === null) {
apps/server/src/cli/config.ts+2
......@@ -472,6 +472,8 @@ export const resolveServerConfig = (
472472 ...(shellEnvironmentPrepared === undefined ? {} : { shellEnvironmentPrepared }),
473473 desktopTelemetryFd,
474474 desktopTelemetryControlFd,
475 desktopTerminalFd: bootstrap?.desktopTerminalFd,
476 desktopTerminalControlFd: bootstrap?.desktopTerminalControlFd,
475477 desktopBrowserFd,
476478 desktopBrowserControlFd,
477479 resourceMonitorPath,
apps/server/src/config.ts+2
......@@ -98,6 +98,8 @@ export class ServerConfig extends Context.Service<
9898 readonly shellEnvironmentPrepared?: boolean | undefined;
9999 readonly desktopTelemetryFd?: number | undefined;
100100 readonly desktopTelemetryControlFd?: number | undefined;
101 readonly desktopTerminalFd?: number | undefined;
102 readonly desktopTerminalControlFd?: number | undefined;
101103 readonly desktopBrowserFd?: number | undefined;
102104 readonly desktopBrowserControlFd?: number | undefined;
103105 readonly resourceMonitorPath?: string | undefined;
apps/server/src/server.ts+2
......@@ -69,6 +69,7 @@ import * as DeviceHubProxy from "./device/DeviceHubProxy.ts";
6969import * as PreviewManager from "./preview/Manager.ts";
7070import * as PortScanner from "./preview/PortScanner.ts";
7171import * as ServerBrowser from "./preview/ServerBrowser.ts";
72import * as DesktopTerminalRpc from "./terminal/DesktopTerminalRpc.ts";
7273import * as DesktopBrowserChannel from "./preview/DesktopBrowserChannel.ts";
7374import * as ServerBrowserStream from "./preview/ServerBrowserStream.ts";
7475import * as PreviewBrowser from "./preview/PreviewBrowser.ts";
......@@ -717,6 +718,7 @@ const layerMakeRoutes = Layer.mergeAll(
717718 ServerBrowserStream.routeLayer,
718719 ServerHttp.layerStaticAndDevRoute,
719720 Ws.layer,
721 DesktopTerminalRpc.layer,
720722 ),
721723 // The MCP session registry is provided globally (shared with V2 provider
722724 // sessions) rather than inline here. The orchestrator toolkit resolves
apps/server/src/terminal/DesktopTerminalRpc.test.ts created+129
......@@ -0,0 +1,129 @@
1import { expect, it } from "@effect/vitest";
2import * as NodeServices from "@effect/platform-node/NodeServices";
3import {
4 ORCHESTRATION_PROTOCOL_QUERY_PARAM,
5 ORCHESTRATION_PROTOCOL_VERSION,
6 WS_METHODS,
7 type DesktopTerminalRequest,
8 type DesktopTerminalResponse,
9} from "@t3tools/contracts";
10import * as Effect from "effect/Effect";
11import * as Deferred from "effect/Deferred";
12import * as Layer from "effect/Layer";
13import * as Option from "effect/Option";
14import * as Queue from "effect/Queue";
15import * as Stream from "effect/Stream";
16import * as ServerConfig from "../config.ts";
17import * as ServerEnvironment from "../environment/ServerEnvironment.ts";
18import * as EnvironmentAuth from "../auth/EnvironmentAuth.ts";
19import * as SessionStore from "../auth/SessionStore.ts";
20import * as ServerSecretStore from "../auth/ServerSecretStore.ts";
21import * as SqlitePersistence from "../persistence/Sqlite.ts";
22import * as TerminalManager from "./Manager.ts";
23import * as DesktopTerminalRpc from "./DesktopTerminalRpc.ts";
24
25const config = ServerConfig.layerTest(process.cwd(), { prefix: "t3-terminal-pipe-test-" });
26const auth = EnvironmentAuth.layer.pipe(
27 Layer.provideMerge(SqlitePersistence.layerMemory),
28 Layer.provide(ServerSecretStore.layer),
29 Layer.provideMerge(ServerEnvironment.layerIdentity),
30 Layer.provide(config),
31);
32
33it.layer(NodeServices.layer)("DesktopTerminalRpc", (it) => {
34 it.effect("port close releases the running RPC, stream subscription and session connection", () =>
35 Effect.gen(function* () {
36 const requests = yield* Queue.unbounded<DesktopTerminalRequest>();
37 const responses = yield* Queue.unbounded<DesktopTerminalResponse>();
38 const unsubscribed = yield* Deferred.make<void>();
39 const sessions = yield* SessionStore.SessionStore;
40 const identity = yield* ServerEnvironment.ServerEnvironmentIdentity;
41 const environmentId = yield* identity.getEnvironmentId;
42 const session = yield* sessions.issue({ scopes: ["terminal:read"] });
43 const ticket = yield* sessions.issueWebSocketToken(session.sessionId);
44 const manager = Layer.mock(TerminalManager.TerminalManager, {
45 observeStream: (input, listener) =>
46 listener({
47 type: "output",
48 threadId: input.threadId,
49 terminalId: input.terminalId,
50 data: "ready",
51 }).pipe(
52 Effect.as(() => {
53 Deferred.doneUnsafe(unsubscribed, Effect.void);
54 }),
55 ),
56 });
57 yield* DesktopTerminalRpc.run(Stream.fromQueue(requests), (response) =>
58 Queue.offer(responses, response).pipe(Effect.asVoid),
59 ).pipe(Effect.provide(manager), Effect.forkScoped);
60 expect((yield* Queue.take(responses)).type).toBe("ready");
61 const url = `ws://127.0.0.1/ws?${ORCHESTRATION_PROTOCOL_QUERY_PARAM}=${ORCHESTRATION_PROTOCOL_VERSION}&wsTicket=${ticket.token}`;
62 yield* Queue.offer(requests, { type: "open", id: "port-1", environmentId, url });
63 yield* Queue.offer(requests, {
64 type: "message",
65 id: "port-1",
66 data: JSON.stringify({
67 _tag: "Request",
68 id: "1",
69 tag: WS_METHODS.terminalObserve,
70 payload: { threadId: "thread", terminalId: "term" },
71 headers: [],
72 }),
73 });
74 const response = yield* Queue.take(responses);
75 expect(response.type).toBe("message");
76 expect(
77 (yield* sessions.listActive()).find((value) => value.sessionId === session.sessionId)
78 ?.connected,
79 ).toBe(true);
80 yield* Queue.offer(requests, { type: "close", id: "port-1" });
81 yield* Deferred.await(unsubscribed);
82 const closed = yield* Stream.fromQueue(responses).pipe(
83 Stream.filter((response) => response.type === "close"),
84 Stream.runHead,
85 );
86 expect(Option.getOrThrow(closed)).toEqual({ type: "close", id: "port-1" });
87 expect(
88 (yield* sessions.listActive()).find((value) => value.sessionId === session.sessionId)
89 ?.connected,
90 ).toBe(false);
91
92 yield* Queue.offer(requests, {
93 type: "open",
94 id: "wrong-environment",
95 environmentId: "other-environment",
96 url,
97 });
98 expect(yield* Queue.take(responses)).toEqual({ type: "close", id: "wrong-environment" });
99 yield* Queue.offer(requests, {
100 type: "open",
101 id: "invalid-ticket",
102 environmentId,
103 url: url.replace(ticket.token, "invalid"),
104 });
105 expect(yield* Queue.take(responses)).toEqual({ type: "close", id: "invalid-ticket" });
106 yield* Queue.offer(requests, { type: "open", id: "read-only", environmentId, url });
107 yield* Queue.offer(requests, {
108 type: "message",
109 id: "read-only",
110 data: JSON.stringify({
111 _tag: "Request",
112 id: "2",
113 tag: WS_METHODS.terminalWrite,
114 payload: { threadId: "thread", terminalId: "term", data: "j" },
115 headers: [],
116 }),
117 });
118 const denied = yield* Queue.take(responses);
119 expect(denied.type).toBe("message");
120 if (denied.type === "message") expect(denied.data).toContain("EnvironmentAuthorizationError");
121 yield* sessions.revoke(session.sessionId);
122 const revoked = yield* Stream.fromQueue(responses).pipe(
123 Stream.filter((response) => response.type === "close"),
124 Stream.runHead,
125 );
126 expect(Option.getOrThrow(revoked)).toEqual({ type: "close", id: "read-only" });
127 }).pipe(Effect.provide(auth), Effect.scoped),
128 );
129});
apps/server/src/terminal/DesktopTerminalRpc.ts created+203
......@@ -0,0 +1,203 @@
1// @effect-diagnostics nodeBuiltinImport:off -- Reads the private pipes inherited from Electron.
2import {
3 DesktopTerminalRequest,
4 DesktopTerminalResponse,
5 TerminalRpcGroup,
6 ORCHESTRATION_PROTOCOL_VERSION,
7 ORCHESTRATION_PROTOCOL_QUERY_PARAM,
8} from "@t3tools/contracts";
9import * as NodeStream from "@effect/platform-node/NodeStream";
10import * as Exit from "effect/Exit";
11import * as Effect from "effect/Effect";
12import * as Layer from "effect/Layer";
13import * as Option from "effect/Option";
14import * as Queue from "effect/Queue";
15import * as Schema from "effect/Schema";
16import * as Scope from "effect/Scope";
17import * as Semaphore from "effect/Semaphore";
18import * as Stream from "effect/Stream";
19import * as Ndjson from "effect/encoding/Ndjson";
20import { HttpServerRequest } from "effect/http";
21import { RpcServer, RpcSerialization } from "effect/rpc";
22import type { FromClientEncoded } from "effect/rpc/RpcMessage";
23import * as NodeNet from "node:net";
24import * as ServerConfig from "../config.ts";
25import * as EnvironmentAuth from "../auth/EnvironmentAuth.ts";
26import * as SessionStore from "../auth/SessionStore.ts";
27import * as RpcAuthorization from "../auth/RpcAuthorization.ts";
28import {
29 RpcInstrumentation,
30 rpcInstrumentationLayer,
31} from "../observability/RpcInstrumentation.ts";
32import * as ServerEnvironment from "../environment/ServerEnvironment.ts";
33import * as TerminalManager from "./Manager.ts";
34import { makeTerminalRpcHandlers } from "./Rpc.ts";
35import { withTerminalOutputWindow } from "./OutputProtocol.ts";
36
37const group = TerminalRpcGroup.middleware(RpcInstrumentation);
38const decode = Schema.decodeUnknownOption(DesktopTerminalRequest);
39const encode = Schema.encodeSync(Schema.fromJsonString(DesktopTerminalResponse));
40
41export const run = (
42 requests: Stream.Stream<DesktopTerminalRequest>,
43 send: (response: DesktopTerminalResponse) => Effect.Effect<void>,
44) =>
45 Effect.gen(function* () {
46 const auth = yield* EnvironmentAuth.EnvironmentAuth;
47 const sessions = yield* SessionStore.SessionStore;
48 const manager = yield* TerminalManager.TerminalManager;
49 const identity = yield* ServerEnvironment.ServerEnvironmentIdentity;
50 const environmentId = yield* identity.getEnvironmentId;
51 const parentScope = yield* Scope.Scope;
52 const serialization = yield* RpcSerialization.RpcSerialization;
53 const connections = new Map<
54 string,
55 { scope: Scope.Closeable; messages: Queue.Queue<string> }
56 >();
57 const close = (id: string) =>
58 Effect.gen(function* () {
59 const connection = connections.get(id);
60 if (!connection) return;
61 connections.delete(id);
62 yield* Scope.close(connection.scope, Exit.void);
63 yield* send({ type: "close", id });
64 });
65 const open = (request: Extract<DesktopTerminalRequest, { type: "open" }>) =>
66 Effect.gen(function* () {
67 if (connections.has(request.id)) return;
68 const scope = yield* Scope.fork(parentScope);
69 const messages = yield* Queue.unbounded<string>();
70 connections.set(request.id, { scope, messages });
71 yield* Effect.gen(function* () {
72 yield* Effect.addFinalizer(() => Queue.shutdown(messages));
73 const url = new URL(request.url);
74 if (
75 request.environmentId !== environmentId ||
76 url.pathname !== "/ws" ||
77 url.searchParams.get(ORCHESTRATION_PROTOCOL_QUERY_PARAM) !==
78 String(ORCHESTRATION_PROTOCOL_VERSION)
79 )
80 return;
81 url.protocol = url.protocol === "wss:" ? "https:" : "http:";
82 const httpRequest = HttpServerRequest.fromWeb(new Request(url));
83 const session = yield* auth.authenticateWebSocketUpgrade(httpRequest);
84 const protocol = yield* RpcServer.Protocol.make((receive) =>
85 Effect.gen(function* () {
86 const parser = serialization.makeUnsafe();
87 yield* Stream.fromQueue(messages).pipe(
88 Stream.runForEach((data) =>
89 // The RPC parser owns encoded envelopes; RpcServer validates payloads against the group.
90 Effect.forEach(
91 parser.decode(data) as ReadonlyArray<FromClientEncoded>,
92 (message) => receive(0, message),
93 { discard: true },
94 ),
95 ),
96 Effect.forkScoped,
97 );
98 return {
99 disconnects: yield* Queue.unbounded<number>(),
100 send: (_clientId, message) =>
101 Effect.suspend(() => {
102 const data = parser.encode(message);
103 return data === undefined
104 ? Effect.void
105 : send({
106 type: "message",
107 id: request.id,
108 data: typeof data === "string" ? data : new TextDecoder().decode(data),
109 });
110 }),
111 end: () => Queue.shutdown(messages),
112 clientIds: Effect.succeed(new Set([0])),
113 initialMessage: Effect.succeedNone,
114 supportsAck: true,
115 supportsTransferables: false,
116 supportsSpanPropagation: true,
117 supportsNotifications: true,
118 codecFor: serialization.codecFor,
119 };
120 }),
121 );
122 yield* Effect.acquireUseRelease(
123 sessions.markConnected(session.sessionId),
124 () =>
125 Effect.raceFirst(
126 RpcServer.make(group, { disableTracing: true, disableFatalDefects: true }).pipe(
127 Effect.provideService(RpcServer.Protocol, withTerminalOutputWindow(protocol)),
128 Effect.provide(
129 Layer.mergeAll(
130 group.toLayer(makeTerminalRpcHandlers(manager)),
131 RpcAuthorization.layer(session.scopes),
132 rpcInstrumentationLayer,
133 ),
134 ),
135 ),
136 sessions.awaitInvalidation(session.sessionId),
137 ),
138 () => sessions.markDisconnected(session.sessionId),
139 );
140 }).pipe(
141 Scope.provide(scope),
142 Effect.catchCause(() => Effect.void),
143 Effect.ensuring(Effect.forkIn(close(request.id), parentScope).pipe(Effect.asVoid)),
144 Effect.forkIn(scope),
145 );
146 });
147 yield* send({ type: "ready", environmentId });
148 yield* requests.pipe(
149 Stream.runForEach((value) => {
150 const request = value;
151 if (request.type === "open") return open(request);
152 if (request.type === "close") return close(request.id);
153 const connection = connections.get(request.id);
154 return connection ? Queue.offer(connection.messages, request.data) : Effect.void;
155 }),
156 Effect.ensuring(
157 Effect.suspend(() => Effect.forEach([...connections.keys()], close, { discard: true })),
158 ),
159 );
160 }).pipe(Effect.provide(RpcSerialization.layerJson));
161
162export const layer = Layer.effectDiscard(
163 Effect.gen(function* () {
164 const config = yield* ServerConfig.ServerConfig;
165 if (config.desktopTerminalFd === undefined || config.desktopTerminalControlFd === undefined)
166 return;
167 const outputFd = config.desktopTerminalControlFd;
168 const lock = yield* Semaphore.make(1);
169 const writable = yield* Effect.acquireRelease(
170 Effect.sync(() => {
171 const socket = new NodeNet.Socket({ fd: outputFd, readable: false, writable: true });
172 socket.on("error", () => socket.destroy());
173 return socket;
174 }),
175 (socket) => Effect.sync(() => socket.destroy()),
176 );
177 const readable = yield* Effect.acquireRelease(
178 Effect.sync(
179 () => new NodeNet.Socket({ fd: config.desktopTerminalFd, readable: true, writable: false }),
180 ),
181 (stream) => Effect.sync(() => stream.destroy()),
182 );
183 const requests = NodeStream.fromReadable({
184 evaluate: () => readable,
185 onError: (cause) => cause,
186 }).pipe(
187 Stream.pipeThroughChannel(Ndjson.decode({ ignoreEmptyLines: true })),
188 Stream.map((value) => decode(value)),
189 Stream.filter(Option.isSome),
190 Stream.map((value) => value.value),
191 Stream.orDie,
192 );
193 yield* run(requests, (response) =>
194 lock.withPermits(1)(
195 Effect.callback<void>((resume) => {
196 writable.write(Buffer.from(`${encode(response)}\n`), (error) =>
197 resume(error ? Effect.die(error) : Effect.void),
198 );
199 }),
200 ),
201 ).pipe(Effect.forkScoped);
202 }),
203);
apps/server/src/terminal/OutputProtocol.test.ts+128-2
......@@ -1,5 +1,5 @@
11import { assert, describe, it } from "@effect/vitest";
2import { WS_METHODS } from "@t3tools/contracts";
2import { TerminalAttachStreamEvent, WS_METHODS } from "@t3tools/contracts";
33import * as Deferred from "effect/Deferred";
44import * as Effect from "effect/Effect";
55import * as Queue from "effect/Queue";
......@@ -8,7 +8,7 @@ import * as Stream from "effect/Stream";
88import * as TestClock from "effect/testing/TestClock";
99import { Rpc, RpcGroup, RpcMessage, RpcSerialization, RpcServer } from "effect/rpc";
1010
11import { withTerminalOutputWindow } from "./OutputProtocol.ts";
11import { batchTerminalOutput, withTerminalOutputWindow } from "./OutputProtocol.ts";
1212
1313describe("terminal output window", () => {
1414 it.effect.each([
......@@ -77,3 +77,129 @@ describe("terminal output window", () => {
7777 }).pipe(Effect.provide(RpcSerialization.layerJson), Effect.scoped),
7878 );
7979});
80
81describe("terminal output batches", () => {
82 it.effect("bounds an acknowledged backlog without changing bytes or event order", () =>
83 Effect.gen(function* () {
84 const group = RpcGroup.make(
85 Rpc.make(WS_METHODS.terminalAttach, { success: TerminalAttachStreamEvent, stream: true }),
86 );
87 const output = yield* Queue.unbounded<TerminalAttachStreamEvent>();
88 const responses = yield* Queue.unbounded<RpcMessage.FromServerEncoded>();
89 const receive = yield* Deferred.make<Parameters<RpcServer.Protocol["Service"]["run"]>[0]>();
90 const protocol = yield* RpcServer.Protocol.make((write) =>
91 Effect.gen(function* () {
92 yield* Deferred.succeed(receive, write);
93 const serialization = yield* RpcSerialization.RpcSerialization;
94 return {
95 disconnects: yield* Queue.unbounded<number>(),
96 send: (_clientId, response) => Queue.offer(responses, response),
97 end: () => Effect.void,
98 clientIds: Effect.succeed(new Set([0])),
99 initialMessage: Effect.succeedNone,
100 supportsAck: true,
101 supportsTransferables: false,
102 supportsSpanPropagation: false,
103 supportsNotifications: true,
104 codecFor: serialization.codecFor,
105 };
106 }),
107 );
108 yield* RpcServer.make(group).pipe(
109 Effect.provide(
110 group.toLayerHandler(WS_METHODS.terminalAttach, () =>
111 Stream.fromQueue(output).pipe(batchTerminalOutput),
112 ),
113 ),
114 Effect.provideService(RpcServer.Protocol, withTerminalOutputWindow(protocol)),
115 Effect.forkScoped,
116 );
117 const write = yield* Deferred.await(receive);
118 yield* write(0, {
119 _tag: "Request",
120 id: "1",
121 tag: WS_METHODS.terminalAttach,
122 payload: null,
123 headers: [],
124 });
125 const event = (sequence: number): TerminalAttachStreamEvent => ({
126 type: "output",
127 threadId: "thread-1",
128 terminalId: "term-1",
129 sequence,
130 data: `\u001b[31m${"🙂\\\n".repeat(400)}\u001b[0m`,
131 });
132 for (let index = 0; index < 8; index++) {
133 yield* Queue.offer(output, event(index));
134 const response = yield* Queue.take(responses);
135 assert.equal(response._tag, "Chunk");
136 }
137
138 const expected = Array.from({ length: 1_024 }, (_, index) => event(index + 8));
139 for (const value of expected) yield* Queue.offer(output, value);
140 yield* TestClock.adjust(0);
141 assert.equal(yield* Queue.size(output), expected.length);
142 assert.equal(yield* Queue.size(responses), 0);
143
144 for (let index = 0; index < 8; index++) {
145 yield* write(0, { _tag: "Ack", requestId: "1" });
146 }
147 const received: unknown[] = [];
148 let batchCount = 0;
149 while (received.length < expected.length) {
150 const response = yield* Queue.take(responses);
151 assert.equal(response._tag, "Chunk");
152 if (response._tag !== "Chunk") continue;
153
154 assert.isAtMost(Buffer.byteLength(JSON.stringify(response)), 64 * 1024);
155 received.push(...response.values);
156 batchCount += 1;
157 yield* write(0, { _tag: "Ack", requestId: "1" });
158 }
159 assert.isAbove(batchCount, 1);
160 assert.deepEqual(received, expected);
161 }).pipe(Effect.provide(RpcSerialization.layerJson), Effect.scoped),
162 );
163
164 it.effect("preserves large snapshots and oversized sequenced events as individual batches", () =>
165 Effect.gen(function* () {
166 const snapshot: TerminalAttachStreamEvent = {
167 type: "snapshot",
168 snapshot: {
169 threadId: "thread-1",
170 terminalId: "term-1",
171 cwd: "/tmp",
172 worktreePath: null,
173 status: "running",
174 pid: 123,
175 history: "x".repeat(8 * 1024 * 1024),
176 label: "sh",
177 exitCode: null,
178 exitSignal: null,
179 updatedAt: "2026-10-15T00:00:00.000Z",
180 },
181 };
182 const output: TerminalAttachStreamEvent = {
183 type: "output",
184 threadId: "thread-1",
185 terminalId: "term-1",
186 sequence: 12,
187 data: "🙂".repeat(32_768),
188 };
189 const exit: TerminalAttachStreamEvent = {
190 type: "exited",
191 threadId: "thread-1",
192 terminalId: "term-1",
193 sequence: 13,
194 exitCode: 0,
195 exitSignal: null,
196 };
197 const batches = yield* Stream.fromArray([snapshot, output, exit]).pipe(
198 batchTerminalOutput,
199 Stream.chunks,
200 Stream.runCollect,
201 );
202 assert.deepEqual(batches, [[snapshot], [output], [exit]]);
203 }),
204 );
205});
apps/server/src/terminal/OutputProtocol.ts+44-1
......@@ -1,5 +1,6 @@
1import { WS_METHODS } from "@t3tools/contracts";
1import { WS_METHODS, type TerminalAttachStreamEvent, type TerminalEvent } from "@t3tools/contracts";
22import * as Effect from "effect/Effect";
3import * as Stream from "effect/Stream";
34import type { RpcServer } from "effect/rpc";
45
56const MAX_PENDING_CHUNKS = 8;
......@@ -8,6 +9,48 @@ const isFull = (sizes: number[]) =>
89 sizes.length >= MAX_PENDING_CHUNKS ||
910 sizes.reduce((total, size) => total + size, 0) >= MAX_PENDING_BYTES;
1011
12/** Keeps sequenced events intact, including snapshots larger than the batch budget. */
13export function batchTerminalOutput<A extends TerminalAttachStreamEvent | TerminalEvent, E, R>(
14 stream: Stream.Stream<A, E, R>,
15): Stream.Stream<A, E, R> {
16 return Stream.transformPull(stream, (pull) =>
17 Effect.sync(() => {
18 let pending: ReadonlyArray<A> = [];
19 let index = 0;
20
21 return Effect.gen(function* () {
22 if (index === pending.length) {
23 pending = yield* pull;
24 index = 0;
25 }
26
27 const first = pending[index++]!;
28 const batch: [A, ...A[]] = [first];
29 // @effect-diagnostics-next-line preferSchemaOverJson:off -- Match the RPC JSON wire encoding.
30 let bytes = Buffer.byteLength(JSON.stringify(first)) + 2;
31
32 // Reserve 1 KiB for RPC framing; oversized events remain indivisible.
33 while (index < pending.length && bytes < MAX_PENDING_BYTES - 1024) {
34 const next = pending[index]!;
35 // @effect-diagnostics-next-line preferSchemaOverJson:off -- Match the RPC JSON wire encoding.
36 const size = Buffer.byteLength(JSON.stringify(next)) + 1;
37 if (bytes + size > MAX_PENDING_BYTES - 1024) break;
38
39 batch.push(next);
40 bytes += size;
41 index += 1;
42 }
43
44 if (index === pending.length) {
45 pending = [];
46 index = 0;
47 }
48 return batch;
49 });
50 }),
51 );
52}
53
1154export function withTerminalOutputWindow(
1255 protocol: RpcServer.Protocol["Service"],
1356): RpcServer.Protocol["Service"] {
apps/server/src/terminal/Rpc.ts created+45
......@@ -0,0 +1,45 @@
1import {
2 TerminalRpcGroup,
3 WS_METHODS,
4 type TerminalAttachStreamEvent,
5 type TerminalEvent,
6 type TerminalError,
7} from "@t3tools/contracts";
8import * as Effect from "effect/Effect";
9import * as Queue from "effect/Queue";
10import * as Stream from "effect/Stream";
11import * as TerminalManager from "./Manager.ts";
12import { batchTerminalOutput } from "./OutputProtocol.ts";
13
14export const makeTerminalRpcHandlers = (
15 terminalManager: TerminalManager.TerminalManager["Service"],
16) =>
17 TerminalRpcGroup.of({
18 [WS_METHODS.terminalOpen]: (input) => terminalManager.open(input),
19 [WS_METHODS.terminalAttach]: (input) =>
20 Stream.callback<TerminalAttachStreamEvent, TerminalError>((queue) =>
21 Effect.acquireRelease(
22 terminalManager.attachStream(input, (event) => Queue.offer(queue, event)),
23 (unsubscribe) => Effect.sync(unsubscribe),
24 ).pipe(Effect.catchCause((cause) => Queue.failCause(queue, cause))),
25 ).pipe(batchTerminalOutput),
26 [WS_METHODS.terminalWrite]: (input) => terminalManager.write(input),
27 [WS_METHODS.terminalResize]: (input) => terminalManager.resize(input),
28 [WS_METHODS.terminalClear]: (input) => terminalManager.clear(input),
29 [WS_METHODS.terminalRestart]: (input) => terminalManager.restart(input),
30 [WS_METHODS.terminalClose]: (input) => terminalManager.close(input),
31 [WS_METHODS.terminalObserve]: (input) =>
32 Stream.callback<TerminalAttachStreamEvent, TerminalError>((queue) =>
33 Effect.acquireRelease(
34 terminalManager.observeStream(input, (event) => Queue.offer(queue, event)),
35 (unsubscribe) => Effect.sync(unsubscribe),
36 ).pipe(Effect.catchCause((cause) => Queue.failCause(queue, cause))),
37 ).pipe(batchTerminalOutput),
38 [WS_METHODS.subscribeTerminalEvents]: (_input) =>
39 Stream.callback<TerminalEvent>((queue) =>
40 Effect.acquireRelease(
41 terminalManager.subscribe((event) => Queue.offer(queue, event)),
42 (unsubscribe) => Effect.sync(unsubscribe),
43 ),
44 ).pipe(batchTerminalOutput),
45 });
apps/server/src/ws.ts+2-30
......@@ -94,9 +94,6 @@ import {
9494 type ProviderDriverKind,
9595 type ProviderInstanceId,
9696 ThreadId,
97 type TerminalAttachStreamEvent,
98 type TerminalError,
99 type TerminalEvent,
10097 type TerminalMetadataStreamEvent,
10198 type PullRequestRef,
10299 WS_METHODS,
......@@ -178,6 +175,7 @@ import * as ServerRuntimeStartup from "./serverRuntimeStartup.ts";
178175import * as ServerSettings from "./serverSettings.ts";
179176import * as StorageCleanup from "./storageCleanup.ts";
180177import * as TerminalManager from "./terminal/Manager.ts";
178import { makeTerminalRpcHandlers } from "./terminal/Rpc.ts";
181179import { withTerminalOutputWindow } from "./terminal/OutputProtocol.ts";
182180import * as PreviewAutomationBroker from "./mcp/PreviewAutomationBroker.ts";
183181import * as ServerBrowser from "./preview/ServerBrowser.ts";
......@@ -2892,33 +2890,7 @@ const layerWsRpc = (
28922890 vcsProvisioning.initRepository(input).pipe(Effect.tap(() => refreshGitStatus(input.cwd))),
28932891 [WS_METHODS.reviewGetDiffPreview]: (input) => review.getDiffPreview(input),
28942892 [WS_METHODS.reviewGetDiffFileContents]: (input) => review.getDiffFileContents(input),
2895 [WS_METHODS.terminalOpen]: (input) => terminalManager.open(input),
2896 [WS_METHODS.terminalAttach]: (input) =>
2897 Stream.callback<TerminalAttachStreamEvent, TerminalError>((queue) =>
2898 Effect.acquireRelease(
2899 terminalManager.attachStream(input, (event) => Queue.offer(queue, event)),
2900 (unsubscribe) => Effect.sync(unsubscribe),
2901 ).pipe(Effect.catchCause((cause) => Queue.failCause(queue, cause))),
2902 ),
2903 [WS_METHODS.terminalWrite]: (input) => terminalManager.write(input),
2904 [WS_METHODS.terminalResize]: (input) => terminalManager.resize(input),
2905 [WS_METHODS.terminalClear]: (input) => terminalManager.clear(input),
2906 [WS_METHODS.terminalRestart]: (input) => terminalManager.restart(input),
2907 [WS_METHODS.terminalClose]: (input) => terminalManager.close(input),
2908 [WS_METHODS.terminalObserve]: (input) =>
2909 Stream.callback<TerminalAttachStreamEvent, TerminalError>((queue) =>
2910 Effect.acquireRelease(
2911 terminalManager.observeStream(input, (event) => Queue.offer(queue, event)),
2912 (unsubscribe) => Effect.sync(unsubscribe),
2913 ).pipe(Effect.catchCause((cause) => Queue.failCause(queue, cause))),
2914 ),
2915 [WS_METHODS.subscribeTerminalEvents]: (_input) =>
2916 Stream.callback<TerminalEvent>((queue) =>
2917 Effect.acquireRelease(
2918 terminalManager.subscribe((event) => Queue.offer(queue, event)),
2919 (unsubscribe) => Effect.sync(unsubscribe),
2920 ),
2921 ),
2893 ...makeTerminalRpcHandlers(terminalManager),
29222894 [WS_METHODS.subscribeTerminalMetadata]: (_input) =>
29232895 Stream.callback<TerminalMetadataStreamEvent>((queue) =>
29242896 Effect.acquireRelease(
apps/web/src/connection/runtime.ts+2
......@@ -1,3 +1,4 @@
1import { desktopTerminalSocket } from "@t3tools/client-runtime/rpc";
12import { Connection } from "@t3tools/client-runtime/connection";
23import { ShellSnapshotLoader } from "@t3tools/client-runtime/state/shell";
34import {
......@@ -32,6 +33,7 @@ type ConnectionLayerSource =
3233const layerProvidedClientConnection = layerSnapshotLoader.pipe(
3334 Layer.provideMerge(
3435 Connection.layerWithOptions({
36 terminalSocket: (prepared) => desktopTerminalSocket(window.desktopBridge, prepared),
3537 environmentThemes: true,
3638 usageLimitSources: true,
3739 usageLimitsCommand: true,
docs/internals/terminal-runtime.md+7
......@@ -5,6 +5,13 @@ client, including the desktop renderer, attaches through the environment connect
55This lets clients reconnect or share a running session. Renderer choices stay local
66to each client and do not change terminal contracts.
77
8Terminal I/O stays independent of chat traffic so stalled streams and queued
9compression cannot block keystrokes. The native desktop's own backend uses a
10MessagePort and inherited pipes; other environments use a lazy second WebSocket.
11Both paths share terminal handlers, ticket authorization, and session supervision.
12Metadata stays on the primary connection so an idle sidebar opens neither path.
13WSL uses WebSockets because its launcher does not preserve inherited descriptors.
14
815## Output and retention
916
1017[Terminal history](../../apps/server/src/terminal/Manager.ts) is incremental.
packages/client-runtime/src/connection/driver.ts+4-1
......@@ -178,7 +178,10 @@ export const make = Effect.gen(function* () {
178178 Effect.fnUntraced(function* (route) {
179179 const prepared = yield* resolver.prepare(routeEntry(entry, route));
180180 yield* reportProgress({ stage: "opening", prepared });
181 const session = yield* sessions.connect(prepared);
181 const session = yield* sessions.connect(
182 prepared,
183 resolver.prepare(routeEntry(entry, route)),
184 );
182185 yield* reportProgress({ stage: "synchronizing", prepared });
183186 yield* session.ready;
184187 return { prepared, session } satisfies EnvironmentConnectionLease;
packages/client-runtime/src/rpc/desktopTerminal.ts created+67
......@@ -0,0 +1,67 @@
1import type { DesktopBridge, DesktopTerminalConnection } from "@t3tools/contracts";
2import * as Deferred from "effect/Deferred";
3import * as Effect from "effect/Effect";
4import * as Queue from "effect/Queue";
5import * as Socket from "effect/socket/Socket";
6import type { PreparedConnection } from "../connection/model.ts";
7
8export function desktopTerminalSocket(
9 bridge: Pick<DesktopBridge, "supportsLocalTerminal" | "openLocalTerminal"> | undefined,
10 prepared: PreparedConnection,
11): Socket.Socket | undefined {
12 if (!bridge?.openLocalTerminal || bridge.supportsLocalTerminal?.(prepared) !== true) return;
13 const open = bridge.openLocalTerminal;
14 const connected = Deferred.makeUnsafe<DesktopTerminalConnection>();
15 const closed = () =>
16 new Socket.SocketError({ reason: new Socket.SocketCloseError({ code: 1001 }) });
17 return {
18 [Socket.TypeId]: Socket.TypeId,
19 reader: Effect.gen(function* () {
20 const queue = yield* Queue.unbounded<string, Socket.SocketError>();
21 const run = Effect.runForkWith(yield* Effect.context<never>());
22 yield* Effect.acquireRelease(
23 Effect.sync(() => {
24 const connection = open(
25 { environmentId: prepared.environmentId, url: prepared.socketUrl },
26 (data) => {
27 run(Queue.offer(queue, data));
28 },
29 () => {
30 run(Queue.fail(queue, closed()));
31 },
32 );
33 run(Deferred.succeed(connected, connection));
34 return connection;
35 }),
36 (connection) =>
37 Effect.sync(connection.close).pipe(Effect.andThen(Queue.fail(queue, closed()))),
38 );
39 return {
40 pull: Queue.take(queue).pipe(Effect.map((data) => [data] as const)),
41 upgrade: () =>
42 Effect.fail(new Socket.SocketError({ reason: new Socket.SocketUpgradeError({}) })),
43 };
44 }),
45 writer: Effect.succeed({
46 write: (data) =>
47 Deferred.await(connected).pipe(
48 Effect.flatMap((connection) =>
49 Effect.sync(() => {
50 if (Socket.isCloseEvent(data)) connection.close();
51 else
52 connection.send(typeof data === "string" ? data : new TextDecoder().decode(data));
53 }),
54 ),
55 ),
56 writeAll: (chunks) =>
57 Deferred.await(connected).pipe(
58 Effect.flatMap((connection) =>
59 Effect.sync(() => {
60 for (const data of chunks)
61 connection.send(typeof data === "string" ? data : new TextDecoder().decode(data));
62 }),
63 ),
64 ),
65 }),
66 };
67}
packages/client-runtime/src/rpc/index.ts+1
......@@ -2,3 +2,4 @@ export * from "./client.ts";
22export * from "./http.ts";
33export type { WsRpcProtocolClient } from "./protocol.ts";
44export { type RpcSession } from "./session.ts";
5export { desktopTerminalSocket } from "./desktopTerminal.ts";
packages/client-runtime/src/rpc/session.test.ts+345-25
......@@ -1,3 +1,4 @@
1import { desktopTerminalSocket } from "./desktopTerminal.ts";
12import {
23 DEFAULT_SERVER_SETTINGS,
34 EnvironmentId,
......@@ -288,10 +289,326 @@ const publishConfigEvents = Effect.fn("TestRpcSessionFactory.publishConfigEvents
288289});
289290
290291describe("RpcSessionFactory", () => {
292 it.effect("uses the local pipe for terminal RPC without opening a terminal WebSocket", () =>
293 Effect.gen(function* () {
294 const incoming = yield* Queue.unbounded<string>();
295 let respond: (data: string) => void = () => {};
296 let onClose: () => void = () => {};
297 let opened = 0;
298 let closed = 0;
299 const { factory, sockets } = yield* makeFactory({
300 terminalSocket: (prepared) =>
301 desktopTerminalSocket(
302 {
303 supportsLocalTerminal: ({ environmentId }) =>
304 environmentId === PREPARED.environmentId,
305 openLocalTerminal: (input, receive, disconnected) => {
306 expect(input.url).toContain("wsTicket=pipe-ticket");
307 opened += 1;
308 respond = receive;
309 onClose = disconnected;
310 return {
311 send: (data) => {
312 Queue.offerUnsafe(incoming, data);
313 },
314 close: () => {
315 closed += 1;
316 },
317 };
318 },
319 },
320 prepared,
321 ),
322 });
323 const session = yield* factory.connect(
324 PREPARED,
325 Effect.succeed({ ...PREPARED, socketUrl: `${PREPARED.socketUrl}&wsTicket=pipe-ticket` }),
326 );
327 const main = yield* awaitSocket(sockets);
328 main.open();
329 yield* completeInitialConfig(main);
330 yield* session.ready;
331 expect(opened).toBe(0);
332 const write = yield* session.client[WS_METHODS.terminalWrite]({
333 threadId: "thread",
334 terminalId: "term-1",
335 data: "j",
336 }).pipe(Effect.forkChild);
337 const message = JSON.parse(yield* Queue.take(incoming));
338 expect(message.tag).toBe(WS_METHODS.terminalWrite);
339 expect(message.payload.data).toBe("j");
340 respond(
341 encodeJson({ _tag: "Exit", requestId: message.id, exit: { _tag: "Success", value: null } }),
342 );
343 yield* Fiber.join(write);
344 expect(sockets).toHaveLength(1);
345 expect(opened).toBe(1);
346 onClose();
347 expect((yield* Effect.exit(session.closed))._tag).toBe("Failure");
348 expect(closed).toBeLessThanOrEqual(1);
349 }).pipe(Effect.scoped),
350 );
351
352 it.effect(
353 "opens one terminal socket lazily with fresh authorization and isolates blocked metadata",
354 () =>
355 Effect.gen(function* () {
356 const { factory, sockets } = yield* makeFactory();
357 let authorizations = 0;
358 const session = yield* factory.connect(
359 PREPARED,
360 Effect.sync(() => {
361 authorizations += 1;
362 return {
363 ...PREPARED,
364 socketUrl: `${PREPARED.socketUrl}&wsTicket=fresh-terminal-ticket`,
365 };
366 }),
367 );
368 const primary = yield* awaitSocket(sockets);
369 primary.open();
370 yield* completeInitialConfig(primary);
371 yield* session.ready;
372 const metadata = yield* session.client[WS_METHODS.subscribeTerminalMetadata](
373 {},
374 { asQueue: true },
375 );
376 const metadataRequest = yield* awaitRequest(primary, 1);
377 expect(authorizations).toBe(0);
378 expect(sockets).toHaveLength(1);
379 primary.serverMessage(
380 encodeJson({
381 _tag: "Chunk",
382 requestId: metadataRequest.id,
383 values: Array.from({ length: 17 }, () => ({
384 type: "remove",
385 threadId: "thread",
386 terminalId: "term-1",
387 })),
388 }),
389 );
390 yield* Effect.yieldNow;
391 expect(yield* Queue.size(metadata)).toBe(16);
392 const probed = yield* Deferred.make<void>();
393 const probe = yield* Effect.forkChild(
394 session.probe.pipe(Effect.tap(() => Deferred.succeed(probed, undefined))),
395 );
396 const probeRequest = yield* awaitRequest(primary, 2);
397 primary.serverMessage(
398 encodeJson({
399 _tag: "Exit",
400 requestId: probeRequest.id,
401 exit: { _tag: "Success", value: {} },
402 }),
403 );
404 const observed = yield* session.client[WS_METHODS.terminalObserve](
405 { threadId: "thread", terminalId: "term-1" },
406 { asQueue: true },
407 );
408 const terminal = yield* awaitSocket(sockets, 1);
409 terminal.open();
410 const observeRequest = yield* awaitRequest(terminal);
411 expect(observeRequest.tag).toBe(WS_METHODS.terminalObserve);
412 expect(terminal.url).toContain("wsTicket=fresh-terminal-ticket");
413 const write = yield* session.client[WS_METHODS.terminalWrite]({
414 threadId: "thread",
415 terminalId: "term-1",
416 data: "j",
417 }).pipe(Effect.forkChild);
418 const writeRequest = yield* awaitRequest(terminal, 1);
419 terminal.serverMessage(
420 encodeJson({
421 _tag: "Exit",
422 requestId: writeRequest.id,
423 exit: { _tag: "Success", value: null },
424 }),
425 );
426 terminal.serverMessage(
427 encodeJson({
428 _tag: "Chunk",
429 requestId: observeRequest.id,
430 values: [{ type: "output", threadId: "thread", terminalId: "term-1", data: "j" }],
431 }),
432 );
433 yield* Fiber.join(write);
434 expect((yield* Queue.take(observed)).type).toBe("output");
435 expect(yield* Deferred.isDone(probed)).toBe(false);
436 expect(authorizations).toBe(1);
437 expect(sockets).toHaveLength(2);
438 yield* Queue.take(metadata);
439 yield* Fiber.join(probe);
440 }),
441 );
442
443 it.effect(
444 "closes both transports with the session and obtains fresh authorization on its replacement",
445 () =>
446 Effect.gen(function* () {
447 const { factory, sockets } = yield* makeFactory();
448 let authorizations = 0;
449 const prepareTerminal = Effect.sync(() => ({
450 ...PREPARED,
451 socketUrl: `${PREPARED.socketUrl}&wsTicket=terminal-${++authorizations}`,
452 }));
453 for (let attempt = 0; attempt < 2; attempt += 1) {
454 yield* Effect.scoped(
455 Effect.gen(function* () {
456 const session = yield* factory.connect(PREPARED, prepareTerminal);
457 const primary = yield* awaitSocket(sockets, attempt * 2);
458 primary.open();
459 yield* completeInitialConfig(primary);
460 yield* session.ready;
461 const resize = yield* session.client[WS_METHODS.terminalResize]({
462 threadId: "thread",
463 terminalId: "term-1",
464 cols: 80,
465 rows: 24,
466 }).pipe(Effect.forkChild);
467 const terminal = yield* awaitSocket(sockets, attempt * 2 + 1);
468 terminal.open();
469 const request = yield* awaitRequest(terminal);
470 terminal.serverMessage(
471 encodeJson({
472 _tag: "Exit",
473 requestId: request.id,
474 exit: { _tag: "Success", value: null },
475 }),
476 );
477 yield* Fiber.join(resize);
478 expect(terminal.url).toContain(`wsTicket=terminal-${attempt + 1}`);
479 (attempt === 0 ? primary : terminal).close(1012, "restart");
480 expect(yield* Effect.flip(session.closed)).toMatchObject({ reason: "transport" });
481 }),
482 );
483 expect(sockets[attempt * 2]?.readyState).toBe(TestWebSocket.CLOSED);
484 expect(sockets[attempt * 2 + 1]?.readyState).toBe(TestWebSocket.CLOSED);
485 }
486 expect(authorizations).toBe(2);
487 }),
488 );
489
490 it.effect("keeps terminal initialization alive when its first command is interrupted", () =>
491 Effect.gen(function* () {
492 const { factory, sockets } = yield* makeFactory();
493 const authorizing = yield* Deferred.make<void>();
494 const authorized = yield* Deferred.make<PreparedConnection>();
495 let authorizations = 0;
496 const session = yield* factory.connect(
497 PREPARED,
498 Effect.gen(function* () {
499 authorizations += 1;
500 yield* Deferred.succeed(authorizing, undefined);
501 return yield* Deferred.await(authorized);
502 }),
503 );
504 const primary = yield* awaitSocket(sockets);
505 primary.open();
506 yield* completeInitialConfig(primary);
507 yield* session.ready;
508 const first = yield* session.client[WS_METHODS.terminalWrite]({
509 threadId: "thread",
510 terminalId: "term-1",
511 data: "first",
512 }).pipe(Effect.forkChild);
513 yield* Deferred.await(authorizing);
514 yield* Fiber.interrupt(first);
515 yield* Deferred.succeed(authorized, PREPARED);
516 const terminal = yield* awaitSocket(sockets, 1);
517 terminal.open();
518 const second = yield* session.client[WS_METHODS.terminalWrite]({
519 threadId: "thread",
520 terminalId: "term-1",
521 data: "second",
522 }).pipe(Effect.forkChild);
523 const request = yield* awaitRequest(terminal);
524 expect(request.payload).toMatchObject({ data: "second" });
525 terminal.serverMessage(
526 encodeJson({ _tag: "Exit", requestId: request.id, exit: { _tag: "Success", value: null } }),
527 );
528 yield* Fiber.join(second);
529 expect(authorizations).toBe(1);
530 expect(primary.readyState).toBe(TestWebSocket.OPEN);
531 expect(terminal.readyState).toBe(TestWebSocket.OPEN);
532 }),
533 );
534
535 it.effect("keeps the terminal transport after its observing caller finishes", () =>
536 Effect.gen(function* () {
537 const { factory, sockets } = yield* makeFactory();
538 const session = yield* factory.connect(PREPARED, Effect.succeed(PREPARED));
539 const primary = yield* awaitSocket(sockets);
540 primary.open();
541 yield* completeInitialConfig(primary);
542 yield* session.ready;
543 const observing = yield* session.client[WS_METHODS.terminalObserve]({
544 threadId: "thread",
545 terminalId: "term-1",
546 }).pipe(Stream.take(1), Stream.runCollect, Effect.forkChild);
547 const terminal = yield* awaitSocket(sockets, 1);
548 terminal.open();
549 const observeRequest = yield* awaitRequest(terminal);
550 terminal.serverMessage(
551 encodeJson({
552 _tag: "Chunk",
553 requestId: observeRequest.id,
554 values: [{ type: "output", threadId: "thread", terminalId: "term-1", data: "a" }],
555 }),
556 );
557 yield* Fiber.join(observing);
558 const writing = yield* session.client[WS_METHODS.terminalWrite]({
559 threadId: "thread",
560 terminalId: "term-1",
561 data: "j",
562 }).pipe(Effect.forkChild);
563 const writeRequest = yield* awaitRequest(terminal, 1);
564 terminal.serverMessage(
565 encodeJson({
566 _tag: "Exit",
567 requestId: writeRequest.id,
568 exit: { _tag: "Success", value: null },
569 }),
570 );
571 yield* Fiber.join(writing);
572 expect(sockets).toHaveLength(2);
573 expect(terminal.readyState).toBe(TestWebSocket.OPEN);
574 }),
575 );
576
577 it.effect.each(["write", "observe", "queue"] as const)(
578 "fails %s and the session if terminal authorization is revoked before first use",
579 (consumer) =>
580 Effect.gen(function* () {
581 const { factory, sockets } = yield* makeFactory();
582 const revoked = new ConnectionBlockedError({
583 reason: "permission",
584 detail: "Terminal access was revoked.",
585 });
586 const session = yield* factory.connect(PREPARED, Effect.fail(revoked));
587 const primary = yield* awaitSocket(sockets);
588 primary.open();
589 yield* completeInitialConfig(primary);
590 yield* session.ready;
591 const input = { threadId: "thread", terminalId: "term-1" };
592 const call =
593 consumer === "write"
594 ? session.client[WS_METHODS.terminalWrite]({ ...input, data: "j" })
595 : consumer === "observe"
596 ? session.client[WS_METHODS.terminalObserve](input).pipe(Stream.runDrain)
597 : Effect.flatMap(
598 session.client[WS_METHODS.terminalObserve](input, { asQueue: true }),
599 (queue) => Stream.fromQueue(queue).pipe(Stream.runDrain),
600 );
601 const result = yield* Effect.flip(call);
602 expect(result._tag).toBe("RpcClientError");
603 expect(yield* Effect.flip(session.closed)).toBe(revoked);
604 expect(sockets).toHaveLength(1);
605 }),
606 );
607
291608 it.effect("owns one scoped websocket attempt and exposes readiness and closure", () =>
292609 Effect.gen(function* () {
293610 const { factory, sockets } = yield* makeFactory();
294 const session = yield* factory.connect(PREPARED);
611 const session = yield* factory.connect(PREPARED, Effect.succeed(PREPARED));
295612 const readyFiber = yield* Effect.forkChild(session.ready);
296613 const socket = yield* awaitSocket(sockets);
297614
......@@ -355,7 +672,7 @@ describe("RpcSessionFactory", () => {
355672
356673 yield* Effect.scoped(
357674 Effect.gen(function* () {
358 const session = yield* factory.connect(PREPARED);
675 const session = yield* factory.connect(PREPARED, Effect.succeed(PREPARED));
359676 const readyFiber = yield* Effect.forkChild(session.ready);
360677 const socket = yield* awaitSocket(sockets);
361678 socket.open();
......@@ -372,7 +689,7 @@ describe("RpcSessionFactory", () => {
372689 Effect.scoped(
373690 Effect.gen(function* () {
374691 const { factory, sockets } = yield* makeFactory();
375 const session = yield* factory.connect(PREPARED);
692 const session = yield* factory.connect(PREPARED, Effect.succeed(PREPARED));
376693 const readyFiber = yield* Effect.forkChild(session.ready);
377694 const socket = yield* awaitSocket(sockets);
378695 socket.open();
......@@ -437,7 +754,7 @@ describe("RpcSessionFactory", () => {
437754 Effect.scoped(
438755 Effect.gen(function* () {
439756 const { factory, sockets } = yield* makeFactory(options);
440 const session = yield* factory.connect(PREPARED);
757 const session = yield* factory.connect(PREPARED, Effect.succeed(PREPARED));
441758 const readyFiber = yield* Effect.forkChild(session.ready);
442759 const socket = yield* awaitSocket(sockets);
443760 socket.open();
......@@ -486,7 +803,7 @@ describe("RpcSessionFactory", () => {
486803 Effect.scoped(
487804 Effect.gen(function* () {
488805 const { factory, sockets } = yield* makeFactory(options);
489 const session = yield* factory.connect(PREPARED);
806 const session = yield* factory.connect(PREPARED, Effect.succeed(PREPARED));
490807 const ready = yield* Effect.forkChild(session.ready);
491808 const socket = yield* awaitSocket(sockets);
492809 socket.open();
......@@ -560,7 +877,7 @@ describe("RpcSessionFactory", () => {
560877 Effect.scoped(
561878 Effect.gen(function* () {
562879 const { factory, sockets } = yield* makeFactory({ environmentThemes: true });
563 const session = yield* factory.connect(PREPARED);
880 const session = yield* factory.connect(PREPARED, Effect.succeed(PREPARED));
564881 const readyFiber = yield* Effect.forkChild(session.ready);
565882 const socket = yield* awaitSocket(sockets);
566883 socket.open();
......@@ -657,7 +974,7 @@ describe("RpcSessionFactory", () => {
657974 environmentThemes: true,
658975 usageLimitSources: true,
659976 });
660 const session = yield* factory.connect(PREPARED);
977 const session = yield* factory.connect(PREPARED, Effect.succeed(PREPARED));
661978 const readyFiber = yield* Effect.forkChild(session.ready);
662979 const socket = yield* awaitSocket(sockets);
663980 socket.open();
......@@ -769,7 +1086,7 @@ describe("RpcSessionFactory", () => {
7691086 Effect.scoped(
7701087 Effect.gen(function* () {
7711088 const { factory, sockets } = yield* makeFactory();
772 const session = yield* factory.connect(PREPARED);
1089 const session = yield* factory.connect(PREPARED, Effect.succeed(PREPARED));
7731090 const readyFiber = yield* Effect.forkChild(session.ready);
7741091 const socket = yield* awaitSocket(sockets);
7751092 socket.open();
......@@ -799,7 +1116,7 @@ describe("RpcSessionFactory", () => {
7991116 Effect.scoped(
8001117 Effect.gen(function* () {
8011118 const { factory, sockets } = yield* makeFactory({ environmentThemes: true });
802 const firstSession = yield* factory.connect(PREPARED);
1119 const firstSession = yield* factory.connect(PREPARED, Effect.succeed(PREPARED));
8031120 const firstReady = yield* Effect.forkChild(firstSession.ready);
8041121 const firstSocket = yield* awaitSocket(sockets);
8051122 firstSocket.open();
......@@ -909,7 +1226,7 @@ describe("RpcSessionFactory", () => {
9091226 label: "Recovered environment",
9101227 },
9111228 } satisfies ServerConfigType;
912 const secondSession = yield* factory.connect(PREPARED);
1229 const secondSession = yield* factory.connect(PREPARED, Effect.succeed(PREPARED));
9131230 const secondReady = yield* Effect.forkChild(secondSession.ready);
9141231 const secondSocket = yield* awaitSocket(sockets, 1);
9151232 secondSocket.open();
......@@ -1018,7 +1335,7 @@ describe("RpcSessionFactory", () => {
10181335 Effect.scoped(
10191336 Effect.gen(function* () {
10201337 const { factory, sockets } = yield* makeFactory();
1021 const session = yield* factory.connect(PREPARED);
1338 const session = yield* factory.connect(PREPARED, Effect.succeed(PREPARED));
10221339 const readyFiber = yield* Effect.forkChild(session.ready);
10231340 const socket = yield* awaitSocket(sockets);
10241341 socket.open();
......@@ -1057,7 +1374,7 @@ describe("RpcSessionFactory", () => {
10571374 it.effect("rejects a server config for a different environment", () =>
10581375 Effect.gen(function* () {
10591376 const { factory, sockets } = yield* makeFactory();
1060 const session = yield* factory.connect(PREPARED);
1377 const session = yield* factory.connect(PREPARED, Effect.succeed(PREPARED));
10611378 const readyFiber = yield* Effect.forkChild(Effect.flip(session.ready));
10621379 const configFiber = yield* session
10631380 .subscribeServerConfig({})
......@@ -1088,7 +1405,7 @@ describe("RpcSessionFactory", () => {
10881405 it.effect("closes the session after 15 seconds without a frame from the server", () =>
10891406 Effect.gen(function* () {
10901407 const { factory, sockets } = yield* makeFactory();
1091 const session = yield* factory.connect(PREPARED);
1408 const session = yield* factory.connect(PREPARED, Effect.succeed(PREPARED));
10921409 const readyFiber = yield* Effect.forkChild(session.ready);
10931410 const closedFiber = yield* Effect.forkChild(Effect.flip(session.closed));
10941411 const socket = yield* awaitSocket(sockets);
......@@ -1118,7 +1435,7 @@ describe("RpcSessionFactory", () => {
11181435 it.effect("keeps reading replies after closing a stream with a full buffer", () =>
11191436 Effect.gen(function* () {
11201437 const { factory, sockets } = yield* makeFactory();
1121 const session = yield* factory.connect(PREPARED);
1438 const session = yield* factory.connect(PREPARED, Effect.succeed(PREPARED));
11221439 const readyFiber = yield* Effect.forkChild(session.ready);
11231440 const socket = yield* awaitSocket(sockets);
11241441 socket.open();
......@@ -1162,7 +1479,7 @@ describe("RpcSessionFactory", () => {
11621479 it.effect("reaches ready when a newer server sends unknown config members", () =>
11631480 Effect.gen(function* () {
11641481 const { factory, sockets } = yield* makeFactory();
1165 const session = yield* factory.connect(PREPARED);
1482 const session = yield* factory.connect(PREPARED, Effect.succeed(PREPARED));
11661483 const readyFiber = yield* Effect.forkChild(session.ready);
11671484 const socket = yield* awaitSocket(sockets);
11681485 socket.open();
......@@ -1197,7 +1514,7 @@ describe("RpcSessionFactory", () => {
11971514 Effect.scoped(
11981515 Effect.gen(function* () {
11991516 const { factory, sockets } = yield* makeFactory();
1200 const session = yield* factory.connect(PREPARED);
1517 const session = yield* factory.connect(PREPARED, Effect.succeed(PREPARED));
12011518 const readyFiber = yield* Effect.forkChild(session.ready);
12021519 const socket = yield* awaitSocket(sockets);
12031520
......@@ -1243,15 +1560,18 @@ describe("RpcSessionFactory", () => {
12431560
12441561 const error = yield* Effect.scoped(
12451562 Effect.gen(function* () {
1246 const session = yield* factory.connect({
1247 ...PREPARED,
1248 target: relay
1249 ? new RelayConnectionTarget({
1250 environmentId: TARGET.environmentId,
1251 label: TARGET.label,
1252 })
1253 : TARGET,
1254 });
1563 const session = yield* factory.connect(
1564 {
1565 ...PREPARED,
1566 target: relay
1567 ? new RelayConnectionTarget({
1568 environmentId: TARGET.environmentId,
1569 label: TARGET.label,
1570 })
1571 : TARGET,
1572 },
1573 Effect.succeed(PREPARED),
1574 );
12551575 const readyFiber = yield* Effect.forkChild(Effect.flip(session.ready));
12561576 yield* awaitSocket(sockets);
12571577
packages/client-runtime/src/rpc/session.ts+142-54
......@@ -10,13 +10,14 @@ import * as Deferred from "effect/Deferred";
1010import * as Effect from "effect/Effect";
1111import * as Equal from "effect/Equal";
1212import * as Exit from "effect/Exit";
13import * as Fiber from "effect/Fiber";
1314import * as Layer from "effect/Layer";
1415import * as Option from "effect/Option";
1516import * as PubSub from "effect/PubSub";
1617import * as Ref from "effect/Ref";
1718import * as Schedule from "effect/Schedule";
1819import * as Schema from "effect/Schema";
19import type * as Scope from "effect/Scope";
20import * as Scope from "effect/Scope";
2021import * as Stream from "effect/Stream";
2122import type * as Rpc from "effect/rpc/Rpc";
2223import * as RpcClient from "effect/rpc/RpcClient";
......@@ -26,11 +27,7 @@ import * as Socket from "effect/socket/Socket";
2627
2728import { makeWsRpcProtocolClient, PING_TIMEOUT, type WsRpcProtocolClient } from "./protocol.ts";
2829import { NETWORK_BLOCKING_HINT } from "../errors/network.ts";
29import type {
30 ConnectionAttemptError,
31 ConnectionTransientError,
32 PreparedConnection,
33} from "../connection/model.ts";
30import type { ConnectionAttemptError, PreparedConnection } from "../connection/model.ts";
3431import {
3532 ConnectionBlockedError,
3633 ConnectionTransientError as ConnectionTransientErrorClass,
......@@ -56,6 +53,7 @@ export interface RpcSession {
5653}
5754
5855export interface RpcSessionOptions {
56 readonly terminalSocket?: (prepared: PreparedConnection) => Socket.Socket | undefined;
5957 readonly environmentThemes?: boolean;
6058 readonly usageLimitSources?: boolean;
6159 /** This client answers /usage-limits itself, so the server may advertise it. */
......@@ -67,6 +65,7 @@ export class RpcSessionFactory extends Context.Service<
6765 {
6866 readonly connect: (
6967 connection: PreparedConnection,
68 prepareTerminalConnection: Effect.Effect<PreparedConnection, ConnectionAttemptError>,
7069 ) => Effect.Effect<RpcSession, ConnectionAttemptError, Scope.Scope>;
7170 }
7271>()("@t3tools/client-runtime/rpc/session/RpcSessionFactory") {}
......@@ -159,7 +158,10 @@ export const make = Effect.fn("RpcSessionFactory.make")(function* (
159158 ...(options.usageLimitsCommand === true ? { usageLimitsCommand: true } : {}),
160159 };
161160
162 const connect = Effect.fnUntraced(function* (connection: PreparedConnection) {
161 const connect = Effect.fnUntraced(function* (
162 connection: PreparedConnection,
163 prepareTerminalConnection: Effect.Effect<PreparedConnection, ConnectionAttemptError>,
164 ) {
163165 const networkHint =
164166 connection.target._tag === "RelayConnectionTarget" ? ` ${NETWORK_BLOCKING_HINT}` : "";
165167 const mapRpcError = (error: Parameters<typeof mapSessionRpcError>[0]) =>
......@@ -168,56 +170,142 @@ export const make = Effect.fn("RpcSessionFactory.make")(function* (
168170 "connection.environment.id": connection.environmentId,
169171 });
170172
171 const connected = yield* Deferred.make<void>();
172 const disconnected = yield* Deferred.make<never, ConnectionTransientError>();
173 // Set when the socket closes because the server stopped answering, so the failure says so
174 // instead of looking like the server closed the connection.
175 const pingTimedOut = yield* Ref.make(false);
176 const hooks = RpcClient.ConnectionHooks.of({
177 onConnect: Deferred.succeed(connected, undefined).pipe(Effect.asVoid),
178 onDisconnect: Effect.all([Deferred.isDone(connected), Ref.get(pingTimedOut)]).pipe(
179 Effect.flatMap(([wasConnected, timedOut]) =>
180 Deferred.fail(
181 disconnected,
182 new ConnectionTransientErrorClass({
183 reason: "transport",
184 detail: `${
185 !wasConnected
186 ? `${connection.label} could not establish a WebSocket connection.`
187 : timedOut
188 ? `${connection.label} stopped responding.`
189 : `${connection.label} disconnected.`
190 }${networkHint}`,
191 }),
173 const scope = yield* Effect.scope;
174 const disconnected = yield* Deferred.make<never, ConnectionAttemptError>();
175 const openTransport = Effect.fnUntraced(function* (
176 prepared: PreparedConnection,
177 terminal = false,
178 ) {
179 const connected = yield* Deferred.make<void>();
180 const pingTimedOut = yield* Ref.make(false);
181 const hooks = RpcClient.ConnectionHooks.of({
182 onConnect: Deferred.succeed(connected, undefined).pipe(Effect.asVoid),
183 onDisconnect: Effect.all([Deferred.isDone(connected), Ref.get(pingTimedOut)]).pipe(
184 Effect.flatMap(([wasConnected, timedOut]) =>
185 Deferred.fail(
186 disconnected,
187 new ConnectionTransientErrorClass({
188 reason: "transport",
189 detail: `${
190 !wasConnected
191 ? `${connection.label} could not establish a WebSocket connection.`
192 : timedOut
193 ? `${connection.label} stopped responding.`
194 : `${connection.label} disconnected.`
195 }${networkHint}`,
196 }),
197 ),
192198 ),
199 Effect.asVoid,
193200 ),
194 Effect.asVoid,
195 ),
196 });
197 const layerSocket = Socket.layerWebSocket(connection.socketUrl, {
198 openTimeout: SOCKET_OPEN_TIMEOUT,
199 }).pipe(Layer.provide(Layer.succeed(Socket.WebSocketConstructor, webSocketConstructor)));
200 const layerProtocol = Layer.effect(
201 RpcClient.Protocol,
202 RpcClient.makeProtocolSocket({
203 pingTimeout: PING_TIMEOUT,
204 retryTransientErrors: false,
205 retryPolicy: Schedule.recurs(0),
206 onPingTimeout: Ref.set(pingTimedOut, true),
207 }),
208 ).pipe(
209 Layer.provide(
210 Layer.mergeAll(
211 layerSocket,
212 RpcSerialization.layerJson,
213 Layer.succeed(RpcClient.ConnectionHooks, hooks),
201 });
202 const localSocket = terminal ? options.terminalSocket?.(prepared) : undefined;
203 const layerSocket = localSocket
204 ? Layer.succeed(Socket.Socket, localSocket)
205 : Socket.layerWebSocket(prepared.socketUrl, {
206 openTimeout: SOCKET_OPEN_TIMEOUT,
207 }).pipe(Layer.provide(Layer.succeed(Socket.WebSocketConstructor, webSocketConstructor)));
208 const layerProtocol = Layer.effect(
209 RpcClient.Protocol,
210 RpcClient.makeProtocolSocket({
211 pingTimeout: PING_TIMEOUT,
212 retryTransientErrors: false,
213 retryPolicy: Schedule.recurs(0),
214 onPingTimeout: Ref.set(pingTimedOut, true),
215 }),
216 ).pipe(
217 Layer.provide(
218 Layer.mergeAll(
219 layerSocket,
220 RpcSerialization.layerJson,
221 Layer.succeed(RpcClient.ConnectionHooks, hooks),
222 ),
214223 ),
224 );
225 const protocolContext = yield* Layer.build(layerProtocol).pipe(
226 Effect.withSpan("environment.websocket.connect"),
227 );
228 const client = yield* makeWsRpcProtocolClient.pipe(Effect.provide(protocolContext));
229 return { client, connected: Deferred.await(connected) };
230 });
231 const transport = yield* openTransport(connection);
232 const protocolClient = transport.client;
233 // Canceling one command must not abandon the connection shared by other terminals.
234 const terminalTransport = yield* prepareTerminalConnection.pipe(
235 Effect.flatMap((prepared) => openTransport(prepared, true)),
236 Effect.tapError((error) => Deferred.fail(disconnected, error)),
237 Effect.map(({ client }) => client),
238 Effect.mapError(
239 (cause) =>
240 new RpcClientError.RpcClientError({
241 reason: new RpcClientError.RpcClientDefect({
242 message: `${connection.label} could not establish a WebSocket connection.`,
243 cause,
244 }),
245 }),
215246 ),
247 Scope.provide(scope),
248 Effect.forkIn(scope),
249 Effect.cached,
216250 );
217 const protocolContext = yield* Layer.build(layerProtocol).pipe(
218 Effect.withSpan("environment.websocket.connect"),
219 );
220 const protocolClient = yield* makeWsRpcProtocolClient.pipe(Effect.provide(protocolContext));
251 const terminalClient = Effect.flatMap(terminalTransport, Fiber.join);
252 const terminalStream = <
253 Tag extends
254 | typeof WS_METHODS.terminalAttach
255 | typeof WS_METHODS.terminalObserve
256 | typeof WS_METHODS.subscribeTerminalEvents,
257 >(
258 tag: Tag,
259 ): WsRpcProtocolClient[Tag] => {
260 type Method = WsRpcProtocolClient[Tag];
261 type Input = Parameters<Method>[0];
262 type Options = Parameters<Method>[1];
263
264 // The tag correlates its payload and result; `asQueue` selects the RPC overload at runtime.
265 return ((input: Input, options?: Options) => {
266 const stream = Stream.unwrap(
267 Effect.map(terminalClient, (client) => {
268 const method = client[tag] as (
269 input: Input,
270 options: Options & { readonly asQueue: false },
271 ) => Stream.Stream<unknown, unknown>;
272 return method(input, { ...options, asQueue: false });
273 }),
274 );
275 return options?.asQueue
276 ? Stream.toQueue(stream, { capacity: options.streamBufferSize ?? 16 })
277 : stream;
278 }) as Method;
279 };
280 // Metadata stays on the main socket: sidebars observe it without opening a terminal.
281 const client: WsRpcProtocolClient = {
282 ...protocolClient,
283 [WS_METHODS.terminalOpen]: (input, options) =>
284 Effect.flatMap(terminalClient, (client) => client[WS_METHODS.terminalOpen](input, options)),
285 [WS_METHODS.terminalWrite]: (input, options) =>
286 Effect.flatMap(terminalClient, (client) =>
287 client[WS_METHODS.terminalWrite](input, options),
288 ),
289 [WS_METHODS.terminalResize]: (input, options) =>
290 Effect.flatMap(terminalClient, (client) =>
291 client[WS_METHODS.terminalResize](input, options),
292 ),
293 [WS_METHODS.terminalClear]: (input, options) =>
294 Effect.flatMap(terminalClient, (client) =>
295 client[WS_METHODS.terminalClear](input, options),
296 ),
297 [WS_METHODS.terminalRestart]: (input, options) =>
298 Effect.flatMap(terminalClient, (client) =>
299 client[WS_METHODS.terminalRestart](input, options),
300 ),
301 [WS_METHODS.terminalClose]: (input, options) =>
302 Effect.flatMap(terminalClient, (client) =>
303 client[WS_METHODS.terminalClose](input, options),
304 ),
305 [WS_METHODS.terminalAttach]: terminalStream(WS_METHODS.terminalAttach),
306 [WS_METHODS.terminalObserve]: terminalStream(WS_METHODS.terminalObserve),
307 [WS_METHODS.subscribeTerminalEvents]: terminalStream(WS_METHODS.subscribeTerminalEvents),
308 };
221309 const initialConfigDeferred = yield* Deferred.make<ServerConfig>();
222310 const serverConfigExit = yield* Deferred.make<void, ServerConfigSubscriptionError>();
223311 const configSubscriptionClosed = yield* Deferred.make<never, ConnectionAttemptError>();
......@@ -383,10 +471,10 @@ export const make = Effect.fn("RpcSessionFactory.make")(function* (
383471 );
384472
385473 return {
386 client: protocolClient,
474 client,
387475 initialConfig,
388476 subscribeServerConfig,
389 ready: Deferred.await(connected).pipe(
477 ready: transport.connected.pipe(
390478 Effect.andThen(initialConfig),
391479 Effect.asVoid,
392480 Effect.raceFirst(Deferred.await(disconnected)),
packages/contracts/src/desktopBootstrap.ts+2
......@@ -35,6 +35,8 @@ export const DesktopBackendBootstrap = Schema.Struct({
3535 desktopBrowserFd: Schema.optionalKey(PositiveInt),
3636 /** Server -> desktop: commands for those tabs. */
3737 desktopBrowserControlFd: Schema.optionalKey(PositiveInt),
38 desktopTerminalFd: Schema.optionalKey(PositiveInt),
39 desktopTerminalControlFd: Schema.optionalKey(PositiveInt),
3840 resourceMonitorPath: Schema.optionalKey(TrimmedNonEmptyString),
3941});
4042
packages/contracts/src/desktopTerminal.ts created+25
......@@ -0,0 +1,25 @@
1import * as Schema from "effect/Schema";
2
3export const DesktopTerminalRequest = Schema.Union([
4 Schema.Struct({
5 type: Schema.Literal("open"),
6 id: Schema.String,
7 environmentId: Schema.String,
8 url: Schema.String,
9 }),
10 Schema.Struct({ type: Schema.Literal("message"), id: Schema.String, data: Schema.String }),
11 Schema.Struct({ type: Schema.Literal("close"), id: Schema.String }),
12]);
13export type DesktopTerminalRequest = typeof DesktopTerminalRequest.Type;
14
15export const DesktopTerminalResponse = Schema.Union([
16 Schema.Struct({ type: Schema.Literal("ready"), environmentId: Schema.String }),
17 Schema.Struct({ type: Schema.Literal("message"), id: Schema.String, data: Schema.String }),
18 Schema.Struct({ type: Schema.Literal("close"), id: Schema.String }),
19]);
20export type DesktopTerminalResponse = typeof DesktopTerminalResponse.Type;
21
22export interface DesktopTerminalConnection {
23 readonly send: (data: string) => void;
24 readonly close: () => void;
25}
packages/contracts/src/index.ts+1
......@@ -65,3 +65,4 @@ export * from "./rpc.ts";
6565export * from "./worktreeSetup.ts";
6666export * from "./secretRequest.ts";
6767export * from "./clientRpcPermissions.ts";
68export * from "./desktopTerminal.ts";
packages/contracts/src/ipc.ts+9
......@@ -1123,6 +1123,15 @@ export const SystemSettingsPaneSchema = Schema.Literals(["full-disk-access"]);
11231123export type SystemSettingsPane = typeof SystemSettingsPaneSchema.Type;
11241124
11251125export interface DesktopBridge {
1126 supportsLocalTerminal?: (input: {
1127 readonly httpBaseUrl: string;
1128 readonly environmentId: string;
1129 }) => boolean;
1130 openLocalTerminal?: (
1131 input: { readonly environmentId: string; readonly url: string },
1132 onMessage: (data: string) => void,
1133 onClose: () => void,
1134 ) => import("./desktopTerminal.ts").DesktopTerminalConnection;
11261135 getAppBranding: () => DesktopAppBranding | null;
11271136 /** Absolute path of a dropped or picked file; absent on desktop builds predating it. */
11281137 getPathForFile?: (file: File) => string;
packages/contracts/src/rpc.ts+15-10
......@@ -1875,6 +1875,18 @@ export class RpcScopeAuthorization extends RpcMiddleware.Service<RpcScopeAuthori
18751875 { error: EnvironmentAuthorizationError },
18761876) {}
18771877
1878export const TerminalRpcGroup = RpcGroup.make(
1879 WsTerminalOpenRpc,
1880 WsTerminalAttachRpc,
1881 WsTerminalObserveRpc,
1882 WsTerminalWriteRpc,
1883 WsTerminalResizeRpc,
1884 WsTerminalClearRpc,
1885 WsTerminalRestartRpc,
1886 WsTerminalCloseRpc,
1887 WsSubscribeTerminalEventsRpc,
1888).middleware(RpcScopeAuthorization);
1889
18781890export const WsRpcGroup = RpcGroup.make(
18791891 WsServerProbeRpc,
18801892 WsServerGetConfigRpc,
......@@ -2018,15 +2030,6 @@ export const WsRpcGroup = RpcGroup.make(
20182030 WsVcsInitRpc,
20192031 WsReviewGetDiffPreviewRpc,
20202032 WsReviewGetDiffFileContentsRpc,
2021 WsTerminalOpenRpc,
2022 WsTerminalAttachRpc,
2023 WsTerminalObserveRpc,
2024 WsTerminalWriteRpc,
2025 WsTerminalResizeRpc,
2026 WsTerminalClearRpc,
2027 WsTerminalRestartRpc,
2028 WsTerminalCloseRpc,
2029 WsSubscribeTerminalEventsRpc,
20302033 WsSubscribeTerminalMetadataRpc,
20312034 WsPreviewOpenRpc,
20322035 WsPreviewNavigateRpc,
......@@ -2069,4 +2072,6 @@ export const WsRpcGroup = RpcGroup.make(
20692072 WsOrchestrationV2SubscribeArchivedShellRpc,
20702073 WsOrchestrationV2SubscribeShellRpc,
20712074 WsOrchestrationV2SubscribeThreadRpc,
2072).middleware(RpcScopeAuthorization);
2075)
2076 .merge(TerminalRpcGroup)
2077 .middleware(RpcScopeAuthorization);