diff --git a/packages/client-runtime/src/rpc/client.test.ts b/packages/client-runtime/src/rpc/client.test.ts index 507d137caccb..a33dcf9b22a4 100644 --- a/packages/client-runtime/src/rpc/client.test.ts +++ b/packages/client-runtime/src/rpc/client.test.ts @@ -1,6 +1,8 @@ import { + DEFAULT_SERVER_SETTINGS, EnvironmentId, type RelayClientInstallProgressEvent, + type ServerConfigStreamEvent, WS_METHODS, } from "@t3tools/contracts"; import { describe, expect, it } from "@effect/vitest"; @@ -77,6 +79,39 @@ const makeHarness = Effect.fn("TestEnvironmentRpc.makeHarness")(function* () { }); describe("environment RPC", () => { + it.effect("reuses the session config stream instead of opening a duplicate subscription", () => + Effect.gen(function* () { + const event: ServerConfigStreamEvent = { + version: 1, + type: "settingsUpdated", + payload: { settings: DEFAULT_SERVER_SETTINGS }, + }; + let duplicateSubscriptions = 0; + const client = { + [WS_METHODS.subscribeServerConfig]: () => { + duplicateSubscriptions += 1; + return Stream.never; + }, + } as unknown as WsRpcProtocolClient; + const { activeSession, supervisor } = yield* makeHarness(); + yield* SubscriptionRef.set( + activeSession, + Option.some({ + ...session(client), + serverConfigEvents: Stream.succeed(event), + }), + ); + + const received = yield* subscribe(WS_METHODS.subscribeServerConfig, {}).pipe( + Stream.runHead, + Effect.provideService(EnvironmentSupervisor.EnvironmentSupervisor, supervisor), + ); + + expect(received).toEqual(Option.some(event)); + expect(duplicateSubscriptions).toBe(0); + }), + ); + it.effect("observes unary requests until they complete", () => Effect.gen(function* () { const observations: string[] = []; diff --git a/packages/client-runtime/src/rpc/client.ts b/packages/client-runtime/src/rpc/client.ts index bfe57a6c0dd5..e95e2298622e 100644 --- a/packages/client-runtime/src/rpc/client.ts +++ b/packages/client-runtime/src/rpc/client.ts @@ -203,7 +203,11 @@ export function subscribeDynamic( Option.match({ onNone: () => Stream.empty, onSome: (session) => { - const method = session.client[tag] as ( + const method = ( + tag === WS_METHODS.subscribeServerConfig && session.serverConfigEvents !== undefined + ? () => session.serverConfigEvents + : session.client[tag] + ) as ( input: EnvironmentRpcInput, ) => Stream.Stream< EnvironmentRpcStreamValue, diff --git a/packages/client-runtime/src/rpc/session.test.ts b/packages/client-runtime/src/rpc/session.test.ts index 0af5850bf6c7..a6e50fe56d9e 100644 --- a/packages/client-runtime/src/rpc/session.test.ts +++ b/packages/client-runtime/src/rpc/session.test.ts @@ -1,8 +1,12 @@ import { DEFAULT_SERVER_SETTINGS, EnvironmentId, + ProviderDriverKind, + ProviderInstanceId, ServerConfig, type ServerConfig as ServerConfigType, + ServerConfigStreamEvent, + type ServerConfigStreamEvent as ServerConfigStreamEventType, WS_METHODS, } from "@t3tools/contracts"; import { describe, expect, it } from "@effect/vitest"; @@ -10,6 +14,7 @@ import * as Effect from "effect/Effect"; import * as Fiber from "effect/Fiber"; import * as Layer from "effect/Layer"; import * as Schema from "effect/Schema"; +import * as Stream from "effect/Stream"; import * as TestClock from "effect/testing/TestClock"; import * as Socket from "effect/unstable/socket/Socket"; @@ -139,9 +144,11 @@ const RpcRequest = Schema.TaggedStruct("Request", { tag: Schema.String, }); const decodeJson = Schema.decodeUnknownSync(Schema.fromJsonString(Schema.Unknown)); -const decodeRpcRequest = Schema.decodeUnknownSync(RpcRequest); +const isRpcRequest = Schema.is(RpcRequest); +const isPing = Schema.is(Schema.Struct({ _tag: Schema.Literal("Ping") })); const encodeJson = Schema.encodeUnknownSync(Schema.fromJsonString(Schema.Unknown)); const encodeServerConfig = Schema.encodeSync(ServerConfig); +const encodeServerConfigStreamEvent = Schema.encodeSync(ServerConfigStreamEvent); const ENCODED_SERVER_CONFIG = encodeServerConfig(SERVER_CONFIG); const LEGACY_SERVER_CONFIG = { ...ENCODED_SERVER_CONFIG, @@ -183,9 +190,9 @@ const awaitRequest = Effect.fn("TestRpcSessionFactory.awaitRequest")(function* ( index = 0, ) { for (let attempt = 0; attempt < 100; attempt += 1) { - const request = socket.sent[index]; + const request = socket.sent.map((message) => decodeJson(message)).filter(isRpcRequest)[index]; if (request) { - return decodeRpcRequest(decodeJson(request)); + return request; } yield* Effect.yieldNow; } @@ -199,17 +206,14 @@ const completeInitialConfig = Effect.fn("TestRpcSessionFactory.completeInitialCo const request = yield* awaitRequest(socket); expect(request).toMatchObject({ _tag: "Request", - tag: WS_METHODS.serverGetConfig, + tag: WS_METHODS.subscribeServerConfig, payload: {}, }); socket.serverMessage( encodeJson({ - _tag: "Exit", + _tag: "Chunk", requestId: request.id, - exit: { - _tag: "Success", - value: config, - }, + values: [{ version: 1, type: "snapshot", config }], }), ); }); @@ -229,7 +233,9 @@ describe("RpcSessionFactory", () => { const config = yield* session.initialConfig; expect(config).toEqual(SERVER_CONFIG); - expect(socket.sent).toHaveLength(1); + expect(socket.sent.map((message) => decodeJson(message)).filter(isRpcRequest)).toHaveLength( + 1, + ); const probeFiber = yield* Effect.forkChild(session.probe); const probeRequest = yield* awaitRequest(socket, 1); @@ -250,19 +256,26 @@ describe("RpcSessionFactory", () => { ); yield* Fiber.join(probeFiber); - expect(socket.sent.map((request) => decodeRpcRequest(decodeJson(request)).tag)).toEqual([ - WS_METHODS.serverGetConfig, - WS_METHODS.serverProbe, - ]); + expect( + socket.sent + .map((message) => decodeJson(message)) + .filter(isRpcRequest) + .map((request) => request.tag), + ).toEqual([WS_METHODS.subscribeServerConfig, WS_METHODS.serverProbe]); socket.close(1012, "service restart"); const error = yield* Effect.flip(session.closed); + const configStreamError = yield* session.serverConfigEvents!.pipe( + Stream.runDrain, + Effect.flip, + ); expect(error).toBeInstanceOf(ConnectionTransientError); expect(error).toMatchObject({ reason: "transport", message: "Test environment disconnected.", }); + expect(configStreamError).toMatchObject({ _tag: "RpcClientError" }); yield* Effect.yieldNow; expect(sockets).toHaveLength(1); }), @@ -287,6 +300,171 @@ describe("RpcSessionFactory", () => { }), ); + it.effect("replays current config and broadcasts updates to every subscriber", () => + Effect.scoped( + Effect.gen(function* () { + const { factory, sockets } = yield* makeFactory(); + const session = yield* factory.connect(PREPARED); + const readyFiber = yield* Effect.forkChild(session.ready); + const socket = yield* awaitSocket(sockets); + socket.open(); + yield* completeInitialConfig(socket); + yield* Fiber.join(readyFiber); + + const collectTwo = session.serverConfigEvents!.pipe(Stream.take(2), Stream.runCollect); + const firstSubscriber = yield* Effect.forkChild(collectTwo); + const secondSubscriber = yield* Effect.forkChild(collectTwo); + yield* Effect.yieldNow; + + const shortcut = { + key: "k", + metaKey: false, + ctrlKey: false, + shiftKey: false, + altKey: false, + modKey: true, + }; + const request = yield* awaitRequest(socket); + socket.serverMessage( + encodeJson({ + _tag: "Chunk", + requestId: request.id, + values: [ + { + version: 1, + type: "keybindingsUpdated", + payload: { + keybindings: [{ command: "terminal.toggle", shortcut }], + issues: [], + }, + }, + ], + }), + ); + + const firstEvents = Array.from(yield* Fiber.join(firstSubscriber)); + const secondEvents = Array.from(yield* Fiber.join(secondSubscriber)); + expect(firstEvents.map((event) => event.type)).toEqual(["snapshot", "keybindingsUpdated"]); + expect(secondEvents).toEqual(firstEvents); + + const replay = yield* session.serverConfigEvents!.pipe(Stream.runHead); + expect(replay).toMatchObject({ + _tag: "Some", + value: { + type: "snapshot", + config: { keybindings: [{ command: "terminal.toggle", shortcut }] }, + }, + }); + }), + ), + ); + + it.effect.each<{ + readonly event: ServerConfigStreamEventType; + readonly expectedConfig: Partial; + }>([ + { + event: { + version: 1, + type: "providerStatuses", + payload: { + providers: [ + { + instanceId: ProviderInstanceId.make("codex"), + driver: ProviderDriverKind.make("codex"), + enabled: true, + installed: true, + version: "1.0.0", + status: "ready", + auth: { status: "authenticated" }, + checkedAt: "2026-08-27T00:00:00.000Z", + models: [], + slashCommands: [], + skills: [], + }, + ], + }, + }, + expectedConfig: { + providers: [ + { + instanceId: ProviderInstanceId.make("codex"), + driver: ProviderDriverKind.make("codex"), + enabled: true, + installed: true, + version: "1.0.0", + status: "ready", + auth: { status: "authenticated" }, + checkedAt: "2026-08-27T00:00:00.000Z", + models: [], + slashCommands: [], + skills: [], + }, + ], + }, + }, + { + event: { + version: 1, + type: "settingsUpdated", + payload: { + settings: { + ...DEFAULT_SERVER_SETTINGS, + newWorktreesStartFromOrigin: !DEFAULT_SERVER_SETTINGS.newWorktreesStartFromOrigin, + }, + }, + }, + expectedConfig: { + settings: { + ...DEFAULT_SERVER_SETTINGS, + newWorktreesStartFromOrigin: !DEFAULT_SERVER_SETTINGS.newWorktreesStartFromOrigin, + }, + }, + }, + ])( + "preserves $event.type events and includes them in replay snapshots", + ({ event, expectedConfig }) => + Effect.scoped( + Effect.gen(function* () { + const { factory, sockets } = yield* makeFactory(); + const session = yield* factory.connect(PREPARED); + const readyFiber = yield* Effect.forkChild(session.ready); + const socket = yield* awaitSocket(sockets); + socket.open(); + yield* completeInitialConfig(socket); + yield* Fiber.join(readyFiber); + + const subscriber = yield* session.serverConfigEvents!.pipe( + Stream.take(2), + Stream.runCollect, + Effect.forkChild, + ); + yield* Effect.yieldNow; + + const request = yield* awaitRequest(socket); + socket.serverMessage( + encodeJson({ + _tag: "Chunk", + requestId: request.id, + values: [encodeServerConfigStreamEvent(event)], + }), + ); + + const events = Array.from(yield* Fiber.join(subscriber)); + expect(events[1]).toEqual(event); + + const replay = yield* session.serverConfigEvents!.pipe(Stream.runHead); + expect(replay).toMatchObject({ + _tag: "Some", + value: { + type: "snapshot", + config: expectedConfig, + }, + }); + }), + ), + ); + it.effect("tolerates two missed pong windows before closing the session", () => Effect.gen(function* () { const { factory, sockets } = yield* makeFactory(); @@ -301,7 +479,7 @@ describe("RpcSessionFactory", () => { yield* TestClock.adjust("15 seconds"); expect(closedFiber.pollUnsafe()).toBeUndefined(); - expect(socket.sent.slice(1).map((request) => decodeJson(request))).toEqual([ + expect(socket.sent.map((message) => decodeJson(message)).filter(isPing)).toEqual([ { _tag: "Ping" }, { _tag: "Ping" }, { _tag: "Ping" }, @@ -379,10 +557,12 @@ describe("RpcSessionFactory", () => { ); yield* Fiber.join(probeFiber); - expect(socket.sent.map((request) => decodeRpcRequest(decodeJson(request)).tag)).toEqual([ - WS_METHODS.serverGetConfig, - WS_METHODS.serverGetConfig, - ]); + expect( + socket.sent + .map((message) => decodeJson(message)) + .filter(isRpcRequest) + .map((request) => request.tag), + ).toEqual([WS_METHODS.subscribeServerConfig, WS_METHODS.serverGetConfig]); }), ), ); diff --git a/packages/client-runtime/src/rpc/session.ts b/packages/client-runtime/src/rpc/session.ts index 9625effa406f..70f299eae364 100644 --- a/packages/client-runtime/src/rpc/session.ts +++ b/packages/client-runtime/src/rpc/session.ts @@ -1,11 +1,22 @@ -import { type ServerConfig, WS_METHODS } from "@t3tools/contracts"; +import { + type EnvironmentAuthorizationError, + type KeybindingsConfigError, + type ServerConfig, + type ServerConfigStreamEvent, + type ServerSettingsError, + WS_METHODS, +} from "@t3tools/contracts"; import * as Context from "effect/Context"; import * as Deferred from "effect/Deferred"; import * as Effect from "effect/Effect"; import * as Layer from "effect/Layer"; +import * as PubSub from "effect/PubSub"; +import * as Ref from "effect/Ref"; import * as Schedule from "effect/Schedule"; import type * as Scope from "effect/Scope"; +import * as Stream from "effect/Stream"; import * as RpcClient from "effect/unstable/rpc/RpcClient"; +import type * as RpcClientError from "effect/unstable/rpc/RpcClientError"; import * as RpcSerialization from "effect/unstable/rpc/RpcSerialization"; import * as Socket from "effect/unstable/socket/Socket"; @@ -25,6 +36,10 @@ const SOCKET_OPEN_TIMEOUT = "15 seconds"; export interface RpcSession { readonly client: WsRpcProtocolClient; readonly initialConfig: Effect.Effect; + readonly serverConfigEvents?: Stream.Stream< + ServerConfigStreamEvent, + ServerConfigSubscriptionError + >; readonly ready: Effect.Effect; readonly probe: Effect.Effect; readonly closed: Effect.Effect; @@ -43,8 +58,46 @@ type InitialConfigError = Effect.Error< ReturnType >; type ProbeError = Effect.Error>; +type ServerConfigSubscriptionError = + | EnvironmentAuthorizationError + | KeybindingsConfigError + | ServerSettingsError + | RpcClientError.RpcClientError; + +interface ServerConfigReplayState { + readonly config: ServerConfig; + readonly revision: number; +} + +interface BufferedServerConfigEvent { + readonly config: ServerConfig; + readonly event: ServerConfigStreamEvent; + readonly revision: number; +} -function mapSessionRpcError(error: InitialConfigError | ProbeError): ConnectionAttemptError { +function applyServerConfigEvent( + config: ServerConfig, + event: ServerConfigStreamEvent, +): ServerConfig { + switch (event.type) { + case "snapshot": + return event.config; + case "keybindingsUpdated": + return { + ...config, + keybindings: event.payload.keybindings, + issues: event.payload.issues, + }; + case "providerStatuses": + return { ...config, providers: event.payload.providers }; + case "settingsUpdated": + return { ...config, settings: event.payload.settings }; + } +} + +function mapSessionRpcError( + error: InitialConfigError | ProbeError | ServerConfigSubscriptionError, +): ConnectionAttemptError { switch (error._tag) { case "EnvironmentAuthorizationError": return new ConnectionBlockedError({ @@ -114,11 +167,96 @@ export const make = Effect.gen(function* () { Effect.withSpan("environment.websocket.connect"), ); const client = yield* makeWsRpcProtocolClient.pipe(Effect.provide(protocolContext)); - const initialConfig = yield* Effect.cached( - client[WS_METHODS.serverGetConfig]({}).pipe( - Effect.mapError(mapSessionRpcError), - Effect.withSpan("environment.initialSync"), + const initialConfigDeferred = yield* Deferred.make(); + const serverConfigExit = yield* Deferred.make(); + const serverConfigState = yield* Ref.make(undefined); + const serverConfigUpdates = yield* PubSub.sliding(64); + const serverConfigSource = client[WS_METHODS.subscribeServerConfig]({}).pipe( + Stream.runForEach((event) => + Effect.gen(function* () { + if (event.type === "snapshot") { + yield* Deferred.succeed(initialConfigDeferred, event.config); + } + const buffered = yield* Ref.modify(serverConfigState, (current) => { + let config: ServerConfig; + if (current === undefined) { + if (event.type !== "snapshot") { + return [undefined, current] as const; + } + config = event.config; + } else { + config = applyServerConfigEvent(current.config, event); + } + const next = { + config, + revision: (current?.revision ?? 0) + 1, + }; + return [{ config: next.config, event, revision: next.revision }, next] as const; + }); + if (buffered !== undefined) { + yield* PubSub.publish(serverConfigUpdates, buffered); + } + }), + ), + Effect.tapError((error) => + Effect.all([ + Deferred.fail(initialConfigDeferred, mapSessionRpcError(error)), + Deferred.fail(serverConfigExit, error), + ]).pipe(Effect.asVoid), ), + Effect.ensuring( + Effect.all([ + Deferred.fail( + initialConfigDeferred, + new ConnectionTransientErrorClass({ + reason: "remote-unavailable", + detail: `${connection.label} config subscription ended before its initial snapshot.`, + }), + ), + Deferred.succeed(serverConfigExit, undefined), + ]).pipe(Effect.asVoid), + ), + ); + yield* serverConfigSource.pipe(Effect.forkScoped); + const initialConfig = Deferred.await(initialConfigDeferred).pipe( + Effect.withSpan("environment.initialSync"), + ); + const serverConfigEvents = Stream.unwrap( + Effect.gen(function* () { + const subscription = yield* PubSub.subscribe(serverConfigUpdates); + yield* initialConfig.pipe(Effect.option); + const snapshot = yield* Ref.get(serverConfigState); + if (snapshot === undefined) { + return Stream.empty; + } + const updates = Stream.fromSubscription(subscription).pipe( + Stream.filter((buffered) => buffered.revision > snapshot.revision), + Stream.mapAccum( + () => snapshot.revision, + (revision, buffered) => [ + buffered.revision, + [ + buffered.revision === revision + 1 + ? buffered.event + : ({ + version: 1, + type: "snapshot", + config: buffered.config, + } satisfies ServerConfigStreamEvent), + ], + ], + ), + ); + const terminal = Stream.fromEffect(Deferred.await(serverConfigExit)).pipe(Stream.drain); + return Stream.concat( + Stream.succeed({ + version: 1 as const, + type: "snapshot" as const, + config: snapshot.config, + }), + Stream.merge(updates, terminal, { haltStrategy: "either" }), + ); + }), ); const probe = initialConfig.pipe( Effect.flatMap((config) => @@ -134,6 +272,7 @@ export const make = Effect.gen(function* () { return { client, initialConfig, + serverConfigEvents, ready: Deferred.await(connected).pipe( Effect.andThen(initialConfig), Effect.asVoid,