From 16d1bc2e1214432333fb47042ba9898906ae0300 Mon Sep 17 00:00:00 2001 From: Adamulek123 Date: Thu, 27 Aug 2026 00:20:52 +0200 Subject: [PATCH 01/10] perf(client-runtime): reuse server config subscription --- .../client-runtime/src/rpc/client.test.ts | 35 +++++++++++++ packages/client-runtime/src/rpc/client.ts | 6 ++- .../client-runtime/src/rpc/session.test.ts | 42 ++++++++------- packages/client-runtime/src/rpc/session.ts | 51 ++++++++++++++++--- 4 files changed, 108 insertions(+), 26 deletions(-) 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..4e934ea8b028 100644 --- a/packages/client-runtime/src/rpc/session.test.ts +++ b/packages/client-runtime/src/rpc/session.test.ts @@ -139,7 +139,8 @@ 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 ENCODED_SERVER_CONFIG = encodeServerConfig(SERVER_CONFIG); @@ -183,9 +184,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 +200,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 +227,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,10 +250,12 @@ 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); @@ -301,7 +303,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 +381,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..f56c81526870 100644 --- a/packages/client-runtime/src/rpc/session.ts +++ b/packages/client-runtime/src/rpc/session.ts @@ -1,11 +1,20 @@ -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 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 +34,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 +56,15 @@ type InitialConfigError = Effect.Error< ReturnType >; type ProbeError = Effect.Error>; +type ServerConfigSubscriptionError = + | EnvironmentAuthorizationError + | KeybindingsConfigError + | ServerSettingsError + | RpcClientError.RpcClientError; -function mapSessionRpcError(error: InitialConfigError | ProbeError): ConnectionAttemptError { +function mapSessionRpcError( + error: InitialConfigError | ProbeError | ServerConfigSubscriptionError, +): ConnectionAttemptError { switch (error._tag) { case "EnvironmentAuthorizationError": return new ConnectionBlockedError({ @@ -114,11 +134,29 @@ 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 serverConfigEvents = client[WS_METHODS.subscribeServerConfig]({}).pipe( + Stream.tap((event) => + event.type === "snapshot" + ? Deferred.succeed(initialConfigDeferred, event.config).pipe(Effect.asVoid) + : Effect.void, + ), + Stream.tapError((error) => + Deferred.fail(initialConfigDeferred, mapSessionRpcError(error)).pipe(Effect.asVoid), ), + Stream.ensuring( + Deferred.fail( + initialConfigDeferred, + new ConnectionTransientErrorClass({ + reason: "remote-unavailable", + detail: `${connection.label} config subscription ended before its initial snapshot.`, + }), + ).pipe(Effect.asVoid), + ), + ); + const serverConfigQueue = yield* Stream.toQueue(serverConfigEvents, { capacity: 64 }); + const initialConfig = Deferred.await(initialConfigDeferred).pipe( + Effect.withSpan("environment.initialSync"), ); const probe = initialConfig.pipe( Effect.flatMap((config) => @@ -134,6 +172,7 @@ export const make = Effect.gen(function* () { return { client, initialConfig, + serverConfigEvents: Stream.fromQueue(serverConfigQueue), ready: Deferred.await(connected).pipe( Effect.andThen(initialConfig), Effect.asVoid, From 675cf82fbf6d386b00392cf4ab527051018cad9b Mon Sep 17 00:00:00 2001 From: Adamulek123 Date: Thu, 27 Aug 2026 10:43:27 +0200 Subject: [PATCH 02/10] fix(client-runtime): replay config to every subscriber --- .../client-runtime/src/rpc/session.test.ts | 60 ++++++++++++ packages/client-runtime/src/rpc/session.ts | 94 +++++++++++++++++-- 2 files changed, 145 insertions(+), 9 deletions(-) diff --git a/packages/client-runtime/src/rpc/session.test.ts b/packages/client-runtime/src/rpc/session.test.ts index 4e934ea8b028..5112ecdf0fe7 100644 --- a/packages/client-runtime/src/rpc/session.test.ts +++ b/packages/client-runtime/src/rpc/session.test.ts @@ -10,6 +10,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"; @@ -289,6 +290,65 @@ 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("tolerates two missed pong windows before closing the session", () => Effect.gen(function* () { const { factory, sockets } = yield* makeFactory(); diff --git a/packages/client-runtime/src/rpc/session.ts b/packages/client-runtime/src/rpc/session.ts index f56c81526870..23c070966be7 100644 --- a/packages/client-runtime/src/rpc/session.ts +++ b/packages/client-runtime/src/rpc/session.ts @@ -10,6 +10,8 @@ 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"; @@ -62,6 +64,36 @@ type ServerConfigSubscriptionError = | ServerSettingsError | RpcClientError.RpcClientError; +interface ServerConfigReplayState { + readonly config: ServerConfig; + readonly revision: number; +} + +interface BufferedServerConfigEvent { + readonly event: ServerConfigStreamEvent; + readonly revision: number; +} + +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 { @@ -135,16 +167,39 @@ export const make = Effect.gen(function* () { ); const client = yield* makeWsRpcProtocolClient.pipe(Effect.provide(protocolContext)); const initialConfigDeferred = yield* Deferred.make(); - const serverConfigEvents = client[WS_METHODS.subscribeServerConfig]({}).pipe( - Stream.tap((event) => - event.type === "snapshot" - ? Deferred.succeed(initialConfigDeferred, event.config).pipe(Effect.asVoid) - : Effect.void, + const serverConfigState = yield* Ref.make(undefined); + const serverConfigUpdates = yield* PubSub.bounded(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 [{ event, revision: next.revision }, next] as const; + }); + if (buffered !== undefined) { + yield* PubSub.publish(serverConfigUpdates, buffered); + } + }), ), - Stream.tapError((error) => + Effect.tapError((error) => Deferred.fail(initialConfigDeferred, mapSessionRpcError(error)).pipe(Effect.asVoid), ), - Stream.ensuring( + Effect.ensuring( Deferred.fail( initialConfigDeferred, new ConnectionTransientErrorClass({ @@ -154,10 +209,31 @@ export const make = Effect.gen(function* () { ).pipe(Effect.asVoid), ), ); - const serverConfigQueue = yield* Stream.toQueue(serverConfigEvents, { capacity: 64 }); + 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; + } + return Stream.concat( + Stream.succeed({ + version: 1 as const, + type: "snapshot" as const, + config: snapshot.config, + }), + Stream.fromSubscription(subscription).pipe( + Stream.filter((buffered) => buffered.revision > snapshot.revision), + Stream.map((buffered) => buffered.event), + ), + ); + }), + ); const probe = initialConfig.pipe( Effect.flatMap((config) => (config.environment.capabilities.connectionProbe === true @@ -172,7 +248,7 @@ export const make = Effect.gen(function* () { return { client, initialConfig, - serverConfigEvents: Stream.fromQueue(serverConfigQueue), + serverConfigEvents, ready: Deferred.await(connected).pipe( Effect.andThen(initialConfig), Effect.asVoid, From 53584a297ee6d113ae6b4afc5e08b32f1952fa10 Mon Sep 17 00:00:00 2001 From: Adamulek123 Date: Thu, 27 Aug 2026 10:51:54 +0200 Subject: [PATCH 03/10] fix(client-runtime): keep config fanout nonblocking --- .../client-runtime/src/rpc/session.test.ts | 7 ++- packages/client-runtime/src/rpc/session.ts | 45 ++++++++++++------- 2 files changed, 36 insertions(+), 16 deletions(-) diff --git a/packages/client-runtime/src/rpc/session.test.ts b/packages/client-runtime/src/rpc/session.test.ts index 5112ecdf0fe7..e73de59abff2 100644 --- a/packages/client-runtime/src/rpc/session.test.ts +++ b/packages/client-runtime/src/rpc/session.test.ts @@ -260,12 +260,17 @@ describe("RpcSessionFactory", () => { 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); }), @@ -334,7 +339,7 @@ describe("RpcSessionFactory", () => { 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(firstEvents.map((event) => event.type)).toEqual(["snapshot", "snapshot"]); expect(secondEvents).toEqual(firstEvents); const replay = yield* session.serverConfigEvents!.pipe(Stream.runHead); diff --git a/packages/client-runtime/src/rpc/session.ts b/packages/client-runtime/src/rpc/session.ts index 23c070966be7..d547f99a1b58 100644 --- a/packages/client-runtime/src/rpc/session.ts +++ b/packages/client-runtime/src/rpc/session.ts @@ -70,7 +70,7 @@ interface ServerConfigReplayState { } interface BufferedServerConfigEvent { - readonly event: ServerConfigStreamEvent; + readonly config: ServerConfig; readonly revision: number; } @@ -167,8 +167,9 @@ export const make = Effect.gen(function* () { ); const client = yield* makeWsRpcProtocolClient.pipe(Effect.provide(protocolContext)); const initialConfigDeferred = yield* Deferred.make(); + const serverConfigExit = yield* Deferred.make(); const serverConfigState = yield* Ref.make(undefined); - const serverConfigUpdates = yield* PubSub.bounded(64); + const serverConfigUpdates = yield* PubSub.sliding(64); const serverConfigSource = client[WS_METHODS.subscribeServerConfig]({}).pipe( Stream.runForEach((event) => Effect.gen(function* () { @@ -189,7 +190,7 @@ export const make = Effect.gen(function* () { config, revision: (current?.revision ?? 0) + 1, }; - return [{ event, revision: next.revision }, next] as const; + return [{ config: next.config, revision: next.revision }, next] as const; }); if (buffered !== undefined) { yield* PubSub.publish(serverConfigUpdates, buffered); @@ -197,16 +198,22 @@ export const make = Effect.gen(function* () { }), ), Effect.tapError((error) => - Deferred.fail(initialConfigDeferred, mapSessionRpcError(error)).pipe(Effect.asVoid), + Effect.all([ + Deferred.fail(initialConfigDeferred, mapSessionRpcError(error)), + Deferred.fail(serverConfigExit, error), + ]).pipe(Effect.asVoid), ), Effect.ensuring( - Deferred.fail( - initialConfigDeferred, - new ConnectionTransientErrorClass({ - reason: "remote-unavailable", - detail: `${connection.label} config subscription ended before its initial snapshot.`, - }), - ).pipe(Effect.asVoid), + 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); @@ -221,16 +228,24 @@ export const make = Effect.gen(function* () { if (snapshot === undefined) { return Stream.empty; } + const updates = Stream.fromSubscription(subscription).pipe( + Stream.filter((buffered) => buffered.revision > snapshot.revision), + Stream.map( + (buffered): ServerConfigStreamEvent => ({ + version: 1, + type: "snapshot", + config: buffered.config, + }), + ), + ); + 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.fromSubscription(subscription).pipe( - Stream.filter((buffered) => buffered.revision > snapshot.revision), - Stream.map((buffered) => buffered.event), - ), + Stream.merge(updates, terminal, { haltStrategy: "either" }), ); }), ); From 3785988410f4e2ec65b20dc850dca220adb863c8 Mon Sep 17 00:00:00 2001 From: Adamulek123 Date: Thu, 27 Aug 2026 11:01:30 +0200 Subject: [PATCH 04/10] fix(client-runtime): preserve config update events --- .../client-runtime/src/rpc/session.test.ts | 2 +- packages/client-runtime/src/rpc/session.ts | 23 +++++++++++++------ 2 files changed, 17 insertions(+), 8 deletions(-) diff --git a/packages/client-runtime/src/rpc/session.test.ts b/packages/client-runtime/src/rpc/session.test.ts index e73de59abff2..13364bed8426 100644 --- a/packages/client-runtime/src/rpc/session.test.ts +++ b/packages/client-runtime/src/rpc/session.test.ts @@ -339,7 +339,7 @@ describe("RpcSessionFactory", () => { const firstEvents = Array.from(yield* Fiber.join(firstSubscriber)); const secondEvents = Array.from(yield* Fiber.join(secondSubscriber)); - expect(firstEvents.map((event) => event.type)).toEqual(["snapshot", "snapshot"]); + expect(firstEvents.map((event) => event.type)).toEqual(["snapshot", "keybindingsUpdated"]); expect(secondEvents).toEqual(firstEvents); const replay = yield* session.serverConfigEvents!.pipe(Stream.runHead); diff --git a/packages/client-runtime/src/rpc/session.ts b/packages/client-runtime/src/rpc/session.ts index d547f99a1b58..70f299eae364 100644 --- a/packages/client-runtime/src/rpc/session.ts +++ b/packages/client-runtime/src/rpc/session.ts @@ -71,6 +71,7 @@ interface ServerConfigReplayState { interface BufferedServerConfigEvent { readonly config: ServerConfig; + readonly event: ServerConfigStreamEvent; readonly revision: number; } @@ -190,7 +191,7 @@ export const make = Effect.gen(function* () { config, revision: (current?.revision ?? 0) + 1, }; - return [{ config: next.config, revision: next.revision }, next] as const; + return [{ config: next.config, event, revision: next.revision }, next] as const; }); if (buffered !== undefined) { yield* PubSub.publish(serverConfigUpdates, buffered); @@ -230,12 +231,20 @@ export const make = Effect.gen(function* () { } const updates = Stream.fromSubscription(subscription).pipe( Stream.filter((buffered) => buffered.revision > snapshot.revision), - Stream.map( - (buffered): ServerConfigStreamEvent => ({ - version: 1, - type: "snapshot", - config: buffered.config, - }), + 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); From df049311040f9f8c0bd10c623cabc161321a7b39 Mon Sep 17 00:00:00 2001 From: Adamulek123 Date: Thu, 27 Aug 2026 14:19:51 +0200 Subject: [PATCH 05/10] test(client-runtime): cover config event replay --- .../client-runtime/src/rpc/session.test.ts | 111 ++++++++++++++++++ 1 file changed, 111 insertions(+) diff --git a/packages/client-runtime/src/rpc/session.test.ts b/packages/client-runtime/src/rpc/session.test.ts index 13364bed8426..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"; @@ -144,6 +148,7 @@ 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, @@ -354,6 +359,112 @@ describe("RpcSessionFactory", () => { ), ); + 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(); From 4ea0362edea498f26e059749c363db355ba84701 Mon Sep 17 00:00:00 2001 From: Theo Browne Date: Tue, 1 Sep 2026 03:33:25 -0700 Subject: [PATCH 06/10] fix(client-runtime): preserve environment theme config events --- apps/web/src/cloud/linkEnvironment.test.ts | 1 + apps/web/src/connection/runtime.ts | 5 +- .../client-runtime/src/connection/layer.ts | 53 ++- .../src/connection/registry.test.ts | 2 + .../src/connection/supervisor.test.ts | 1 + .../src/operations/commands.test.ts | 1 + .../client-runtime/src/rpc/client.test.ts | 3 +- packages/client-runtime/src/rpc/client.ts | 4 +- .../client-runtime/src/rpc/session.test.ts | 340 +++++++++++++++++- packages/client-runtime/src/rpc/session.ts | 234 +++++++----- .../src/state/pullRequests.test.ts | 1 + .../client-runtime/src/state/server.test.ts | 1 + packages/client-runtime/src/state/server.ts | 95 +---- .../src/state/serverConfigProjection.ts | 74 ++++ .../src/state/shell-sync.test.ts | 1 + .../src/state/sourceControl.test.ts | 1 + .../src/state/threads-pagination.test.ts | 1 + .../src/state/threads-sync.test.ts | 1 + packages/client-runtime/src/state/vcs.test.ts | 1 + .../src/state/vcsAction.test.ts | 1 + 20 files changed, 592 insertions(+), 229 deletions(-) create mode 100644 packages/client-runtime/src/state/serverConfigProjection.ts diff --git a/apps/web/src/cloud/linkEnvironment.test.ts b/apps/web/src/cloud/linkEnvironment.test.ts index 38e205beabbb..7ae5e7ed03a9 100644 --- a/apps/web/src/cloud/linkEnvironment.test.ts +++ b/apps/web/src/cloud/linkEnvironment.test.ts @@ -91,6 +91,7 @@ function registryLayer(options?: { const session: RpcSession = { client, initialConfig: Effect.never, + subscribeServerConfig: (input) => client.subscribeServerConfig(input), ready: Effect.void, probe: Effect.void, closed: Effect.never, diff --git a/apps/web/src/connection/runtime.ts b/apps/web/src/connection/runtime.ts index b63d01999036..06c8bf0ccfed 100644 --- a/apps/web/src/connection/runtime.ts +++ b/apps/web/src/connection/runtime.ts @@ -30,7 +30,10 @@ type ConnectionLayerSource = | typeof backgroundActivityObserverLayer | typeof backgroundActivityReporterLayer; -const providedClientConnectionLayer = Layer.merge(Connection.layer, snapshotLoaderLayer).pipe( +const providedClientConnectionLayer = Layer.merge( + Connection.layerWithOptions({ environmentThemes: true }), + snapshotLoaderLayer, +).pipe( Layer.provideMerge( Layer.mergeAll( runtimeContextLayer, diff --git a/packages/client-runtime/src/connection/layer.ts b/packages/client-runtime/src/connection/layer.ts index 798ec01e2f0d..7927151e5d42 100644 --- a/packages/client-runtime/src/connection/layer.ts +++ b/packages/client-runtime/src/connection/layer.ts @@ -15,30 +15,29 @@ const resolverLayer = ConnectionResolver.layer.pipe( Layer.provide(RemoteEnvironmentAuthorization.layer), ); -const driverLayer = ConnectionDriver.layer.pipe( - Layer.provide(Layer.mergeAll(resolverLayer, RpcSession.layer)), -); - -const registryLayer = EnvironmentRegistry.layer.pipe(Layer.provide(driverLayer)); - -const onboardingLayer = ConnectionOnboarding.layer.pipe(Layer.provide(registryLayer)); - -const connectionServicesLayer = Layer.mergeAll( - registryLayer, - RelayEnvironmentDiscovery.layer, - onboardingLayer, -); - -const connectionStartupLayer = Layer.effectDiscard( - Effect.gen(function* () { - const registry = yield* EnvironmentRegistry.EnvironmentRegistry; - const platformSource = yield* PlatformConnectionSource.PlatformConnectionSource; - yield* registry.start; - yield* platformSource.registrations.pipe( - Stream.runForEach(registry.reconcilePlatform), - Effect.forkScoped, - ); - }).pipe(Effect.withSpan("clientRuntime.connection.application.start")), -); - -export const layer = connectionStartupLayer.pipe(Layer.provideMerge(connectionServicesLayer)); +export function layerWithOptions(options: RpcSession.RpcSessionOptions) { + const driverLayer = ConnectionDriver.layer.pipe( + Layer.provide(Layer.mergeAll(resolverLayer, RpcSession.layerWithOptions(options))), + ); + const registryLayer = EnvironmentRegistry.layer.pipe(Layer.provide(driverLayer)); + const onboardingLayer = ConnectionOnboarding.layer.pipe(Layer.provide(registryLayer)); + const connectionServicesLayer = Layer.mergeAll( + registryLayer, + RelayEnvironmentDiscovery.layer, + onboardingLayer, + ); + const connectionStartupLayer = Layer.effectDiscard( + Effect.gen(function* () { + const registry = yield* EnvironmentRegistry.EnvironmentRegistry; + const platformSource = yield* PlatformConnectionSource.PlatformConnectionSource; + yield* registry.start; + yield* platformSource.registrations.pipe( + Stream.runForEach(registry.reconcilePlatform), + Effect.forkScoped, + ); + }).pipe(Effect.withSpan("clientRuntime.connection.application.start")), + ); + return connectionStartupLayer.pipe(Layer.provideMerge(connectionServicesLayer)); +} + +export const layer = layerWithOptions({}); diff --git a/packages/client-runtime/src/connection/registry.test.ts b/packages/client-runtime/src/connection/registry.test.ts index 2d0ce41daf7c..4dbbfe45fe00 100644 --- a/packages/client-runtime/src/connection/registry.test.ts +++ b/packages/client-runtime/src/connection/registry.test.ts @@ -356,6 +356,8 @@ const makeHarness = Effect.fn("TestEnvironmentRegistry.makeHarness")(function* ( Effect.succeed({ client: {} as RpcSession.RpcSession["client"], initialConfig: Effect.die(new Error("Config is not used by registry tests.")), + subscribeServerConfig: () => + Stream.die(new Error("Config is not used by registry tests.")), ready: Effect.void, probe: Effect.void, closed: Deferred.await(closed), diff --git a/packages/client-runtime/src/connection/supervisor.test.ts b/packages/client-runtime/src/connection/supervisor.test.ts index 5e50c44d9610..d9f54bb326ca 100644 --- a/packages/client-runtime/src/connection/supervisor.test.ts +++ b/packages/client-runtime/src/connection/supervisor.test.ts @@ -163,6 +163,7 @@ const makeHarness = Effect.fn("TestConnectionHarness.make")(function* (options?: Effect.succeed({ client: TEST_RPC_CLIENT, initialConfig: Effect.die(new Error("Initial config is not used by supervisor tests.")), + subscribeServerConfig: (input) => TEST_RPC_CLIENT.subscribeServerConfig(input), ready: options?.ready?.(attempt) ?? Effect.void, probe: options?.probe?.(attempt) ?? Effect.void, closed: Deferred.await(closed), diff --git a/packages/client-runtime/src/operations/commands.test.ts b/packages/client-runtime/src/operations/commands.test.ts index 0cb1650066c4..36bc6a7b296f 100644 --- a/packages/client-runtime/src/operations/commands.test.ts +++ b/packages/client-runtime/src/operations/commands.test.ts @@ -57,6 +57,7 @@ const makeSupervisor = Effect.fn("TestEnvironmentCommands.makeSupervisor")(funct const session: RpcSession.RpcSession = { client, initialConfig: Effect.never, + subscribeServerConfig: (input) => client.subscribeServerConfig(input), ready: Effect.void, probe: Effect.void, closed: Effect.never, diff --git a/packages/client-runtime/src/rpc/client.test.ts b/packages/client-runtime/src/rpc/client.test.ts index a33dcf9b22a4..4e6baba8bef4 100644 --- a/packages/client-runtime/src/rpc/client.test.ts +++ b/packages/client-runtime/src/rpc/client.test.ts @@ -49,6 +49,7 @@ function session(client: WsRpcProtocolClient): RpcSession.RpcSession { return { client, initialConfig: Effect.never, + subscribeServerConfig: (input) => client.subscribeServerConfig(input), ready: Effect.void, probe: Effect.void, closed: Effect.never, @@ -98,7 +99,7 @@ describe("environment RPC", () => { activeSession, Option.some({ ...session(client), - serverConfigEvents: Stream.succeed(event), + subscribeServerConfig: () => Stream.succeed(event), }), ); diff --git a/packages/client-runtime/src/rpc/client.ts b/packages/client-runtime/src/rpc/client.ts index e95e2298622e..50cc029eccff 100644 --- a/packages/client-runtime/src/rpc/client.ts +++ b/packages/client-runtime/src/rpc/client.ts @@ -204,8 +204,8 @@ export function subscribeDynamic( onNone: () => Stream.empty, onSome: (session) => { const method = ( - tag === WS_METHODS.subscribeServerConfig && session.serverConfigEvents !== undefined - ? () => session.serverConfigEvents + tag === WS_METHODS.subscribeServerConfig + ? session.subscribeServerConfig : session.client[tag] ) as ( input: EnvironmentRpcInput, diff --git a/packages/client-runtime/src/rpc/session.test.ts b/packages/client-runtime/src/rpc/session.test.ts index a6e50fe56d9e..9fdf5907078a 100644 --- a/packages/client-runtime/src/rpc/session.test.ts +++ b/packages/client-runtime/src/rpc/session.test.ts @@ -10,9 +10,14 @@ import { WS_METHODS, } from "@t3tools/contracts"; import { describe, expect, it } from "@effect/vitest"; +import * as Cause from "effect/Cause"; +import * as Deferred from "effect/Deferred"; import * as Effect from "effect/Effect"; +import * as Exit from "effect/Exit"; import * as Fiber from "effect/Fiber"; import * as Layer from "effect/Layer"; +import * as Option from "effect/Option"; +import * as Queue from "effect/Queue"; import * as Schema from "effect/Schema"; import * as Stream from "effect/Stream"; import * as TestClock from "effect/testing/TestClock"; @@ -24,6 +29,7 @@ import { type PreparedConnection, } from "../connection/model.ts"; import * as RpcSession from "./session.ts"; +import { applyServerConfigProjection } from "../state/serverConfigProjection.ts"; type SocketEventType = "open" | "message" | "close" | "error"; type SocketEvent = { @@ -149,7 +155,19 @@ 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 encodeDefect = Schema.encodeSync(Schema.Defect()); const ENCODED_SERVER_CONFIG = encodeServerConfig(SERVER_CONFIG); +const THEME_SERVER_CONFIG: ServerConfigType = { + ...SERVER_CONFIG, + environment: { + ...SERVER_CONFIG.environment, + capabilities: { + ...SERVER_CONFIG.environment.capabilities, + environmentThemes: true, + }, + }, +}; +const ENCODED_THEME_SERVER_CONFIG = encodeServerConfig(THEME_SERVER_CONFIG); const LEGACY_SERVER_CONFIG = { ...ENCODED_SERVER_CONFIG, environment: { @@ -160,14 +178,16 @@ const LEGACY_SERVER_CONFIG = { }, }; -const makeFactory = Effect.fn("TestRpcSessionFactory.make")(function* () { +const makeFactory = Effect.fn("TestRpcSessionFactory.make")(function* ( + options: RpcSession.RpcSessionOptions = {}, +) { const sockets: TestWebSocket[] = []; const constructorLayer = Layer.succeed(Socket.WebSocketConstructor, (url) => { const socket = new TestWebSocket(url); sockets.push(socket); return socket as unknown as globalThis.WebSocket; }); - const layer = RpcSession.layer.pipe(Layer.provide(constructorLayer)); + const layer = RpcSession.layerWithOptions(options).pipe(Layer.provide(constructorLayer)); const factory = yield* RpcSession.RpcSessionFactory.pipe(Effect.provide(layer)); return { factory, sockets }; }); @@ -202,12 +222,13 @@ const awaitRequest = Effect.fn("TestRpcSessionFactory.awaitRequest")(function* ( const completeInitialConfig = Effect.fn("TestRpcSessionFactory.completeInitialConfig")(function* ( socket: TestWebSocket, config: unknown = ENCODED_SERVER_CONFIG, + payload: unknown = {}, ) { const request = yield* awaitRequest(socket); expect(request).toMatchObject({ _tag: "Request", tag: WS_METHODS.subscribeServerConfig, - payload: {}, + payload, }); socket.serverMessage( encodeJson({ @@ -218,6 +239,20 @@ const completeInitialConfig = Effect.fn("TestRpcSessionFactory.completeInitialCo ); }); +const publishConfigEvents = Effect.fn("TestRpcSessionFactory.publishConfigEvents")(function* ( + socket: TestWebSocket, + events: ReadonlyArray, +) { + const request = yield* awaitRequest(socket); + socket.serverMessage( + encodeJson({ + _tag: "Chunk", + requestId: request.id, + values: events.map((event) => encodeServerConfigStreamEvent(event)), + }), + ); +}); + describe("RpcSessionFactory", () => { it.effect("owns one scoped websocket attempt and exposes readiness and closure", () => Effect.gen(function* () { @@ -265,10 +300,9 @@ describe("RpcSessionFactory", () => { socket.close(1012, "service restart"); const error = yield* Effect.flip(session.closed); - const configStreamError = yield* session.serverConfigEvents!.pipe( - Stream.runDrain, - Effect.flip, - ); + const configStreamError = yield* session + .subscribeServerConfig({}) + .pipe(Stream.runDrain, Effect.flip); expect(error).toBeInstanceOf(ConnectionTransientError); expect(error).toMatchObject({ @@ -311,7 +345,9 @@ describe("RpcSessionFactory", () => { yield* completeInitialConfig(socket); yield* Fiber.join(readyFiber); - const collectTwo = session.serverConfigEvents!.pipe(Stream.take(2), Stream.runCollect); + const collectTwo = session + .subscribeServerConfig({}) + .pipe(Stream.take(2), Stream.runCollect); const firstSubscriber = yield* Effect.forkChild(collectTwo); const secondSubscriber = yield* Effect.forkChild(collectTwo); yield* Effect.yieldNow; @@ -347,7 +383,7 @@ describe("RpcSessionFactory", () => { expect(firstEvents.map((event) => event.type)).toEqual(["snapshot", "keybindingsUpdated"]); expect(secondEvents).toEqual(firstEvents); - const replay = yield* session.serverConfigEvents!.pipe(Stream.runHead); + const replay = yield* session.subscribeServerConfig({}).pipe(Stream.runHead); expect(replay).toMatchObject({ _tag: "Some", value: { @@ -359,6 +395,282 @@ describe("RpcSessionFactory", () => { ), ); + it.effect("shares only a config subscription with the same theme opt-in", () => + Effect.scoped( + Effect.gen(function* () { + const { factory, sockets } = yield* makeFactory({ environmentThemes: true }); + const session = yield* factory.connect(PREPARED); + const readyFiber = yield* Effect.forkChild(session.ready); + const socket = yield* awaitSocket(sockets); + socket.open(); + yield* completeInitialConfig(socket, ENCODED_THEME_SERVER_CONFIG, { + environmentThemes: true, + }); + yield* Fiber.join(readyFiber); + + const shared = yield* session + .subscribeServerConfig({ environmentThemes: true }) + .pipe(Stream.runHead); + expect(shared).toMatchObject({ _tag: "Some", value: { type: "snapshot" } }); + expect(socket.sent.map((message) => decodeJson(message)).filter(isRpcRequest)).toHaveLength( + 1, + ); + + const fallbackFiber = yield* session + .subscribeServerConfig({}) + .pipe(Stream.runHead, Effect.forkChild); + const fallbackRequest = yield* awaitRequest(socket, 1); + expect(fallbackRequest).toMatchObject({ + tag: WS_METHODS.subscribeServerConfig, + payload: {}, + }); + socket.serverMessage( + encodeJson({ + _tag: "Chunk", + requestId: fallbackRequest.id, + values: [ + { + version: 1, + type: "snapshot", + config: ENCODED_THEME_SERVER_CONFIG, + }, + ], + }), + ); + expect(yield* Fiber.join(fallbackFiber)).toMatchObject({ + _tag: "Some", + value: { type: "snapshot" }, + }); + }), + ), + ); + + it.effect("replays theme updates and deletion as authoritative events", () => + Effect.scoped( + Effect.gen(function* () { + const { factory, sockets } = yield* makeFactory({ environmentThemes: true }); + const session = yield* factory.connect(PREPARED); + const readyFiber = yield* Effect.forkChild(session.ready); + const socket = yield* awaitSocket(sockets); + socket.open(); + yield* completeInitialConfig(socket, ENCODED_THEME_SERVER_CONFIG, { + environmentThemes: true, + }); + yield* Fiber.join(readyFiber); + + const firstThemes = [ + { + id: "nightfall", + name: "Nightfall", + appearance: "dark" as const, + canvas: "#1a1b26", + accent: "#7aa2f7", + }, + ]; + const replacementThemes = [ + { + id: "midnight", + name: "Midnight", + appearance: "dark" as const, + canvas: "#000000", + accent: "#ffffff", + }, + ]; + const subscriberStarted = yield* Deferred.make(); + const subscriber = yield* session.subscribeServerConfig({ environmentThemes: true }).pipe( + Stream.mapEffect((event) => + Deferred.succeed(subscriberStarted, undefined).pipe(Effect.as(event)), + ), + Stream.take(4), + Stream.runCollect, + Effect.forkChild, + ); + yield* Deferred.await(subscriberStarted); + yield* publishConfigEvents(socket, [ + { + version: 1, + type: "environmentThemesUpdated", + payload: { themes: firstThemes }, + }, + { + version: 1, + type: "environmentThemesUpdated", + payload: { themes: replacementThemes }, + }, + { + version: 1, + type: "environmentThemesUpdated", + payload: { themes: [] }, + }, + ]); + + const liveEvents = Array.from(yield* Fiber.join(subscriber)); + expect(liveEvents.map((event) => event.type)).toEqual([ + "snapshot", + "environmentThemesUpdated", + "environmentThemesUpdated", + "environmentThemesUpdated", + ]); + expect(liveEvents[2]).toMatchObject({ payload: { themes: replacementThemes } }); + + const replay = Array.from( + yield* session + .subscribeServerConfig({ environmentThemes: true }) + .pipe(Stream.take(2), Stream.runCollect), + ); + expect(replay.map((event) => event.type)).toEqual(["snapshot", "environmentThemesUpdated"]); + expect(replay[1]).toMatchObject({ payload: { themes: [] } }); + + let projection = applyServerConfigProjection(Option.none(), { + version: 1, + type: "snapshot", + config: THEME_SERVER_CONFIG, + }); + projection = applyServerConfigProjection(projection, { + version: 1, + type: "environmentThemesUpdated", + payload: { themes: firstThemes }, + }); + for (const event of replay) { + projection = applyServerConfigProjection(projection, event); + } + expect(Option.getOrThrow(projection).config.environmentThemes).toBeUndefined(); + }), + ), + ); + + it.effect("recovers a slow subscriber after it misses theme deletion", () => + Effect.scoped( + Effect.gen(function* () { + const { factory, sockets } = yield* makeFactory({ environmentThemes: true }); + const session = yield* factory.connect(PREPARED); + const readyFiber = yield* Effect.forkChild(session.ready); + const socket = yield* awaitSocket(sockets); + socket.open(); + yield* completeInitialConfig(socket, ENCODED_THEME_SERVER_CONFIG, { + environmentThemes: true, + }); + yield* Fiber.join(readyFiber); + + const slowSubscriberStarted = yield* Deferred.make(); + const releaseSlowSubscriber = yield* Deferred.make(); + let firstEvent = true; + const slowSubscriber = yield* session + .subscribeServerConfig({ environmentThemes: true }) + .pipe( + Stream.mapEffect((event) => { + if (!firstEvent) return Effect.succeed(event); + firstEvent = false; + return Deferred.succeed(slowSubscriberStarted, undefined).pipe( + Effect.andThen(Deferred.await(releaseSlowSubscriber)), + Effect.as(event), + ); + }), + Stream.take(3), + Stream.runCollect, + Effect.forkChild, + ); + yield* Deferred.await(slowSubscriberStarted); + + const firstThemes = [ + { + id: "nightfall", + name: "Nightfall", + appearance: "dark" as const, + canvas: "#1a1b26", + accent: "#7aa2f7", + }, + ]; + const themeEvents: ServerConfigStreamEventType[] = [ + { + version: 1, + type: "environmentThemesUpdated", + payload: { themes: firstThemes }, + }, + { + version: 1, + type: "environmentThemesUpdated", + payload: { + themes: [{ ...firstThemes[0]!, name: "Nightfall 2" }], + }, + }, + { + version: 1, + type: "environmentThemesUpdated", + payload: { themes: [] }, + }, + ]; + const settingsEvents = Array.from( + { length: 65 }, + (): ServerConfigStreamEventType => ({ + version: 1, + type: "settingsUpdated", + payload: { settings: DEFAULT_SERVER_SETTINGS }, + }), + ); + const allEvents = [...themeEvents, ...settingsEvents]; + const observedByFastSubscriber = yield* Queue.unbounded(); + yield* session.subscribeServerConfig({ environmentThemes: true }).pipe( + Stream.runForEach((event) => Queue.offer(observedByFastSubscriber, event)), + Effect.forkChild, + ); + expect((yield* Queue.take(observedByFastSubscriber)).type).toBe("snapshot"); + for (const event of allEvents) { + yield* publishConfigEvents(socket, [event]); + expect(yield* Queue.take(observedByFastSubscriber)).toEqual(event); + } + yield* Deferred.succeed(releaseSlowSubscriber, undefined); + + const recovered = Array.from(yield* Fiber.join(slowSubscriber)); + expect(recovered.map((event) => event.type)).toEqual([ + "snapshot", + "snapshot", + "environmentThemesUpdated", + ]); + expect(recovered[2]).toMatchObject({ payload: { themes: [] } }); + + let projection = applyServerConfigProjection(Option.none(), { + version: 1, + type: "snapshot", + config: THEME_SERVER_CONFIG, + }); + projection = applyServerConfigProjection(projection, themeEvents[0]!); + for (const event of recovered.slice(1)) { + projection = applyServerConfigProjection(projection, event); + } + expect(Option.getOrThrow(projection).config.environmentThemes).toBeUndefined(); + }), + ), + ); + + it.effect("closes the session when the config source dies", () => + 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 closedFiber = yield* session.closed.pipe(Effect.exit, Effect.forkChild); + socket.serverMessage( + encodeJson({ + _tag: "Defect", + defect: encodeDefect(new Error("config stream died")), + }), + ); + + const closed = yield* Fiber.join(closedFiber); + expect(Exit.isFailure(closed)).toBe(true); + if (Exit.isFailure(closed)) { + expect(Cause.hasDies(closed.cause)).toBe(true); + } + }), + ), + ); + it.effect.each<{ readonly event: ServerConfigStreamEventType; readonly expectedConfig: Partial; @@ -434,11 +746,9 @@ describe("RpcSessionFactory", () => { yield* completeInitialConfig(socket); yield* Fiber.join(readyFiber); - const subscriber = yield* session.serverConfigEvents!.pipe( - Stream.take(2), - Stream.runCollect, - Effect.forkChild, - ); + const subscriber = yield* session + .subscribeServerConfig({}) + .pipe(Stream.take(2), Stream.runCollect, Effect.forkChild); yield* Effect.yieldNow; const request = yield* awaitRequest(socket); @@ -453,7 +763,7 @@ describe("RpcSessionFactory", () => { const events = Array.from(yield* Fiber.join(subscriber)); expect(events[1]).toEqual(event); - const replay = yield* session.serverConfigEvents!.pipe(Stream.runHead); + const replay = yield* session.subscribeServerConfig({}).pipe(Stream.runHead); expect(replay).toMatchObject({ _tag: "Some", value: { diff --git a/packages/client-runtime/src/rpc/session.ts b/packages/client-runtime/src/rpc/session.ts index 70f299eae364..2043d0b25d90 100644 --- a/packages/client-runtime/src/rpc/session.ts +++ b/packages/client-runtime/src/rpc/session.ts @@ -1,20 +1,23 @@ import { - type EnvironmentAuthorizationError, - type KeybindingsConfigError, type ServerConfig, type ServerConfigStreamEvent, - type ServerSettingsError, + WsSubscribeServerConfigRpc, WS_METHODS, } from "@t3tools/contracts"; +import * as Cause from "effect/Cause"; import * as Context from "effect/Context"; import * as Deferred from "effect/Deferred"; import * as Effect from "effect/Effect"; +import * as Equal from "effect/Equal"; +import * as Exit from "effect/Exit"; import * as Layer from "effect/Layer"; +import * as Option from "effect/Option"; import * as PubSub from "effect/PubSub"; import * as Ref from "effect/Ref"; import * as Schedule from "effect/Schedule"; import type * as Scope from "effect/Scope"; import * as Stream from "effect/Stream"; +import type * as Rpc from "effect/unstable/rpc/Rpc"; 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"; @@ -30,19 +33,27 @@ import { ConnectionBlockedError, ConnectionTransientError as ConnectionTransientErrorClass, } from "../connection/model.ts"; +import { + applyServerConfigProjection, + type ServerConfigProjection, + withoutEnvironmentThemes, +} from "../state/serverConfigProjection.ts"; const SOCKET_OPEN_TIMEOUT = "15 seconds"; export interface RpcSession { readonly client: WsRpcProtocolClient; readonly initialConfig: Effect.Effect; - readonly serverConfigEvents?: Stream.Stream< - ServerConfigStreamEvent, - ServerConfigSubscriptionError - >; + readonly subscribeServerConfig: ( + input: ServerConfigSubscriptionInput, + ) => ServerConfigSubscription; readonly ready: Effect.Effect; readonly probe: Effect.Effect; - readonly closed: Effect.Effect; + readonly closed: Effect.Effect; +} + +export interface RpcSessionOptions { + readonly environmentThemes?: boolean; } export class RpcSessionFactory extends Context.Service< @@ -59,40 +70,41 @@ type InitialConfigError = Effect.Error< >; type ProbeError = Effect.Error>; type ServerConfigSubscriptionError = - | EnvironmentAuthorizationError - | KeybindingsConfigError - | ServerSettingsError + | Rpc.ErrorExit | RpcClientError.RpcClientError; +type ServerConfigSubscription = Stream.Stream< + ServerConfigStreamEvent, + ServerConfigSubscriptionError +>; +type ServerConfigSubscriptionInput = Parameters< + WsRpcProtocolClient[typeof WS_METHODS.subscribeServerConfig] +>[0]; +type EnvironmentThemesUpdatedEvent = Extract< + ServerConfigStreamEvent, + { readonly type: "environmentThemesUpdated" } +>; interface ServerConfigReplayState { - readonly config: ServerConfig; + readonly projection: ServerConfigProjection; readonly revision: number; + readonly themesEvent: EnvironmentThemesUpdatedEvent | undefined; } interface BufferedServerConfigEvent { - readonly config: ServerConfig; readonly event: ServerConfigStreamEvent; + readonly replay: ServerConfigReplayState; readonly revision: number; } -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 serverConfigReplayEvents( + state: ServerConfigReplayState, +): ReadonlyArray { + const snapshot = { + version: 1 as const, + type: "snapshot" as const, + config: withoutEnvironmentThemes(state.projection.config), + }; + return state.themesEvent === undefined ? [snapshot] : [snapshot, state.themesEvent]; } function mapSessionRpcError( @@ -118,8 +130,12 @@ function mapSessionRpcError( } } -export const make = Effect.gen(function* () { +export const make = Effect.fn("RpcSessionFactory.make")(function* ( + options: RpcSessionOptions = {}, +) { const webSocketConstructor = yield* Socket.WebSocketConstructor; + const serverConfigInput: ServerConfigSubscriptionInput = + options.environmentThemes === true ? { environmentThemes: true } : {}; const connect = Effect.fnUntraced(function* (connection: PreparedConnection) { yield* Effect.annotateCurrentSpan({ @@ -166,103 +182,119 @@ export const make = Effect.gen(function* () { const protocolContext = yield* Layer.build(protocolLayer).pipe( Effect.withSpan("environment.websocket.connect"), ); - const client = yield* makeWsRpcProtocolClient.pipe(Effect.provide(protocolContext)); - const initialConfigDeferred = yield* Deferred.make(); + const protocolClient = yield* makeWsRpcProtocolClient.pipe(Effect.provide(protocolContext)); + const initialConfigDeferred = yield* Deferred.make(); const serverConfigExit = yield* Deferred.make(); - const serverConfigState = yield* Ref.make(undefined); + const configSubscriptionClosed = yield* Deferred.make(); + const serverConfigState = yield* Ref.make(Option.none()); const serverConfigUpdates = yield* PubSub.sliding(64); - const serverConfigSource = client[WS_METHODS.subscribeServerConfig]({}).pipe( + const configSubscriptionEndedError = new ConnectionTransientErrorClass({ + reason: "remote-unavailable", + detail: `${connection.label} config subscription ended.`, + }); + const serverConfigSource = protocolClient[WS_METHODS.subscribeServerConfig]( + serverConfigInput, + ).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 projection = applyServerConfigProjection( + Option.map(current, (state) => state.projection), + event, + ); + if (Option.isNone(projection)) { + return [Option.none(), current] as const; } const next = { - config, - revision: (current?.revision ?? 0) + 1, - }; - return [{ config: next.config, event, revision: next.revision }, next] as const; + projection: projection.value, + revision: Option.match(current, { + onNone: () => 1, + onSome: (state) => state.revision + 1, + }), + themesEvent: + event.type === "environmentThemesUpdated" + ? event + : event.type === "snapshot" && + event.config.environment.capabilities.environmentThemes !== true + ? undefined + : Option.getOrUndefined(current)?.themesEvent, + } satisfies ServerConfigReplayState; + return [ + Option.some({ event, replay: next, revision: next.revision }), + Option.some(next), + ] as const; }); - if (buffered !== undefined) { - yield* PubSub.publish(serverConfigUpdates, buffered); + if (Option.isSome(buffered)) { + yield* PubSub.publish(serverConfigUpdates, buffered.value); + } + if (event.type === "snapshot") { + yield* Deferred.succeed(initialConfigDeferred, event.config); } }), ), - 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), - ), + Effect.onExit((exit) => { + if (Exit.isSuccess(exit)) { + return Effect.all([ + Deferred.succeed(serverConfigExit, undefined), + Deferred.fail(configSubscriptionClosed, configSubscriptionEndedError), + ]).pipe(Effect.asVoid); + } + if (Cause.hasInterruptsOnly(exit.cause)) { + return Effect.void; + } + return Effect.all([ + Deferred.failCause(serverConfigExit, exit.cause), + Deferred.failCause(configSubscriptionClosed, Cause.map(exit.cause, mapSessionRpcError)), + ]).pipe(Effect.asVoid); + }), ); yield* serverConfigSource.pipe(Effect.forkScoped); - const initialConfig = Deferred.await(initialConfigDeferred).pipe( - Effect.withSpan("environment.initialSync"), - ); + const initialConfig = Effect.raceFirst( + Deferred.await(initialConfigDeferred), + Deferred.await(serverConfigExit).pipe( + Effect.mapError(mapSessionRpcError), + Effect.flatMap(() => Effect.fail(configSubscriptionEndedError)), + ), + ).pipe(Effect.withSpan("environment.initialSync")); const serverConfigEvents = Stream.unwrap( Effect.gen(function* () { const subscription = yield* PubSub.subscribe(serverConfigUpdates); - yield* initialConfig.pipe(Effect.option); + yield* Effect.raceFirst( + Deferred.await(initialConfigDeferred).pipe(Effect.asVoid), + Deferred.await(serverConfigExit), + ); const snapshot = yield* Ref.get(serverConfigState); - if (snapshot === undefined) { + if (Option.isNone(snapshot)) { return Stream.empty; } const updates = Stream.fromSubscription(subscription).pipe( - Stream.filter((buffered) => buffered.revision > snapshot.revision), + Stream.filter((buffered) => buffered.revision > snapshot.value.revision), Stream.mapAccum( - () => snapshot.revision, + () => snapshot.value.revision, (revision, buffered) => [ buffered.revision, - [ - buffered.revision === revision + 1 - ? buffered.event - : ({ - version: 1, - type: "snapshot", - config: buffered.config, - } satisfies ServerConfigStreamEvent), - ], + buffered.revision === revision + 1 + ? [buffered.event] + : serverConfigReplayEvents(buffered.replay), ], ), ); 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.fromIterable(serverConfigReplayEvents(snapshot.value)), Stream.merge(updates, terminal, { haltStrategy: "either" }), ); }), ); + const subscribeServerConfig = (input: ServerConfigSubscriptionInput) => + Equal.equals(input, serverConfigInput) + ? serverConfigEvents + : protocolClient[WS_METHODS.subscribeServerConfig](input); const probe = initialConfig.pipe( Effect.flatMap((config) => (config.environment.capabilities.connectionProbe === true - ? client[WS_METHODS.serverProbe]({}) - : client[WS_METHODS.serverGetConfig]({}) + ? protocolClient[WS_METHODS.serverProbe]({}) + : protocolClient[WS_METHODS.serverGetConfig]({}) ).pipe(Effect.mapError(mapSessionRpcError)), ), Effect.asVoid, @@ -270,20 +302,26 @@ export const make = Effect.gen(function* () { ); return { - client, + client: protocolClient, initialConfig, - serverConfigEvents, + subscribeServerConfig, ready: Deferred.await(connected).pipe( Effect.andThen(initialConfig), Effect.asVoid, Effect.raceFirst(Deferred.await(disconnected)), ), probe, - closed: Deferred.await(disconnected), + closed: Effect.raceFirst( + Deferred.await(disconnected), + Deferred.await(configSubscriptionClosed), + ), } satisfies RpcSession; }); return RpcSessionFactory.of({ connect }); }); -export const layer = Layer.effect(RpcSessionFactory, make); +export const layerWithOptions = (options: RpcSessionOptions) => + Layer.effect(RpcSessionFactory, make(options)); + +export const layer = layerWithOptions({}); diff --git a/packages/client-runtime/src/state/pullRequests.test.ts b/packages/client-runtime/src/state/pullRequests.test.ts index e5dde9d8427c..618d5c39418b 100644 --- a/packages/client-runtime/src/state/pullRequests.test.ts +++ b/packages/client-runtime/src/state/pullRequests.test.ts @@ -32,6 +32,7 @@ function session(client: WsRpcProtocolClient): RpcSession { return { client, initialConfig: Effect.never, + subscribeServerConfig: (input) => client.subscribeServerConfig(input), ready: Effect.void, probe: Effect.void, closed: Effect.never, diff --git a/packages/client-runtime/src/state/server.test.ts b/packages/client-runtime/src/state/server.test.ts index 17dc09998fbf..c383bd4dc44f 100644 --- a/packages/client-runtime/src/state/server.test.ts +++ b/packages/client-runtime/src/state/server.test.ts @@ -73,6 +73,7 @@ function session(client: WsRpcProtocolClient): RpcSession { return { client, initialConfig: Effect.succeed(CONFIG), + subscribeServerConfig: (input) => client.subscribeServerConfig(input), ready: Effect.void, probe: Effect.void, closed: Effect.never, diff --git a/packages/client-runtime/src/state/server.ts b/packages/client-runtime/src/state/server.ts index 547fa4a3af28..e88216065d14 100644 --- a/packages/client-runtime/src/state/server.ts +++ b/packages/client-runtime/src/state/server.ts @@ -41,6 +41,16 @@ import { type EnvironmentRpcInput, } from "../rpc/client.ts"; import { followStreamInEnvironment } from "./runtime.ts"; +import { + applyServerConfigProjection, + type ServerConfigProjection, + withoutEnvironmentThemes, +} from "./serverConfigProjection.ts"; + +export { + applyServerConfigProjection, + type ServerConfigProjection, +} from "./serverConfigProjection.ts"; export type ServerUpdateStage = "downloading" | "installing" | "resuming"; @@ -262,75 +272,6 @@ export function resolveServerUpdateProgressResult( return Effect.fail(new ServerUpdateProgressIncompleteError({ targetVersion })); } -export interface ServerConfigProjection { - readonly config: ServerConfig; - readonly latestEvent: ServerConfigStreamEvent; - readonly source: "cache" | "live"; -} - -export function applyServerConfigProjection( - current: Option.Option, - event: ServerConfigStreamEvent, -): Option.Option { - switch (event.type) { - case "snapshot": { - // A snapshot never carries published themes -- the theme stream owns - // them -- so taking it wholesale would clear the set on every reconnect - // and repaint anyone wearing one until the follow-up event landed. - // Only from a server that still streams them. Reconnecting to one that - // predates the feature must drop the set rather than leave a palette on - // screen that nothing will ever update again. - const carried = - event.config.environment.capabilities.environmentThemes === true && Option.isSome(current) - ? current.value.config.environmentThemes - : undefined; - return Option.some({ - config: - carried === undefined ? event.config : { ...event.config, environmentThemes: carried }, - latestEvent: event, - source: "live" as const, - }); - } - case "keybindingsUpdated": - return Option.map(current, (projection) => ({ - config: { - ...projection.config, - keybindings: event.payload.keybindings, - issues: event.payload.issues, - }, - latestEvent: event, - source: "live", - })); - case "providerStatuses": - return Option.map(current, (projection) => ({ - config: { - ...projection.config, - providers: event.payload.providers, - }, - latestEvent: event, - source: "live", - })); - case "settingsUpdated": - return Option.map(current, (projection) => ({ - config: { - ...projection.config, - settings: event.payload.settings, - }, - latestEvent: event, - source: "live", - })); - case "environmentThemesUpdated": - return Option.map(current, (projection) => ({ - config: { - ...projection.config, - environmentThemes: event.payload.themes.length > 0 ? event.payload.themes : undefined, - }, - latestEvent: event, - source: "live", - })); - } -} - export function projectServerConfig( current: Option.Option, event: ServerConfigStreamEvent, @@ -345,22 +286,6 @@ const cachedConfigSnapshotEvent = (config: ServerConfig): ServerConfigStreamEven config, }); -/** - * Keeps a complete server configuration available during reconnects. Server - * config carries the provider/model catalogue used by task creation, so it is - * useful—and safe—to retain after a transport session ends. - */ -/** - * Published themes live only as long as the machine publishes them, so they - * must not survive in the config cache: a restart or an offline load would - * otherwise hand clients palettes the environment has already dropped. - */ -function withoutEnvironmentThemes(config: ServerConfig): ServerConfig { - if (config.environmentThemes === undefined) return config; - const { environmentThemes: _ephemeral, ...rest } = config; - return rest; -} - export const makeEnvironmentServerConfigState = Effect.fn("EnvironmentServerConfigState.make")( function* (environmentThemes?: boolean) { const supervisor = yield* EnvironmentSupervisor; diff --git a/packages/client-runtime/src/state/serverConfigProjection.ts b/packages/client-runtime/src/state/serverConfigProjection.ts new file mode 100644 index 000000000000..dd893ab6fbec --- /dev/null +++ b/packages/client-runtime/src/state/serverConfigProjection.ts @@ -0,0 +1,74 @@ +import type { ServerConfig, ServerConfigStreamEvent } from "@t3tools/contracts"; +import * as Option from "effect/Option"; + +export interface ServerConfigProjection { + readonly config: ServerConfig; + readonly latestEvent: ServerConfigStreamEvent; + readonly source: "cache" | "live"; +} + +/** Removes machine-owned themes before a config enters durable storage or a replay snapshot. */ +export function withoutEnvironmentThemes(config: ServerConfig): ServerConfig { + if (config.environmentThemes === undefined) return config; + const { environmentThemes: _ephemeral, ...rest } = config; + return rest; +} + +export function applyServerConfigProjection( + current: Option.Option, + event: ServerConfigStreamEvent, +): Option.Option { + switch (event.type) { + case "snapshot": { + // Wire snapshots never contain published themes. Keep the previous set + // until a capable server sends its authoritative theme event. + const carried = + event.config.environment.capabilities.environmentThemes === true && Option.isSome(current) + ? current.value.config.environmentThemes + : undefined; + return Option.some({ + config: + carried === undefined ? event.config : { ...event.config, environmentThemes: carried }, + latestEvent: event, + source: "live" as const, + }); + } + case "keybindingsUpdated": + return Option.map(current, (projection) => ({ + config: { + ...projection.config, + keybindings: event.payload.keybindings, + issues: event.payload.issues, + }, + latestEvent: event, + source: "live", + })); + case "providerStatuses": + return Option.map(current, (projection) => ({ + config: { + ...projection.config, + providers: event.payload.providers, + }, + latestEvent: event, + source: "live", + })); + case "settingsUpdated": + return Option.map(current, (projection) => ({ + config: { + ...projection.config, + settings: event.payload.settings, + }, + latestEvent: event, + source: "live", + })); + case "environmentThemesUpdated": + return Option.map(current, (projection) => ({ + config: { + ...projection.config, + environmentThemes: event.payload.themes.length > 0 ? event.payload.themes : undefined, + }, + latestEvent: event, + source: "live", + })); + } +} diff --git a/packages/client-runtime/src/state/shell-sync.test.ts b/packages/client-runtime/src/state/shell-sync.test.ts index 40e9bd80dc5b..1c0d838026fb 100644 --- a/packages/client-runtime/src/state/shell-sync.test.ts +++ b/packages/client-runtime/src/state/shell-sync.test.ts @@ -51,6 +51,7 @@ function session(client: WsRpcProtocolClient): RpcSession.RpcSession { return { client, initialConfig: Effect.succeed({ shellResumeCompletionMarker: true } as never), + subscribeServerConfig: (input) => client.subscribeServerConfig(input), ready: Effect.void, probe: Effect.void, closed: Effect.never, diff --git a/packages/client-runtime/src/state/sourceControl.test.ts b/packages/client-runtime/src/state/sourceControl.test.ts index 393be8e3227d..33c566bf82b6 100644 --- a/packages/client-runtime/src/state/sourceControl.test.ts +++ b/packages/client-runtime/src/state/sourceControl.test.ts @@ -50,6 +50,7 @@ function session(client: WsRpcProtocolClient): RpcSession { return { client, initialConfig: Effect.never, + subscribeServerConfig: (input) => client.subscribeServerConfig(input), ready: Effect.void, probe: Effect.void, closed: Effect.never, diff --git a/packages/client-runtime/src/state/threads-pagination.test.ts b/packages/client-runtime/src/state/threads-pagination.test.ts index 62cad18f89e0..2cede4f5b3e2 100644 --- a/packages/client-runtime/src/state/threads-pagination.test.ts +++ b/packages/client-runtime/src/state/threads-pagination.test.ts @@ -156,6 +156,7 @@ const makeHarness = Effect.fn("TestThreadPagination.makeHarness")(function* (opt initialConfig: Effect.succeed({ threadSnapshotPagination: options?.paginationCapability !== false, } as never), + subscribeServerConfig: (input) => client.subscribeServerConfig(input), ready: Effect.void, probe: Effect.void, closed: Effect.never, diff --git a/packages/client-runtime/src/state/threads-sync.test.ts b/packages/client-runtime/src/state/threads-sync.test.ts index c2df434e8e77..d94ed3a3fd74 100644 --- a/packages/client-runtime/src/state/threads-sync.test.ts +++ b/packages/client-runtime/src/state/threads-sync.test.ts @@ -112,6 +112,7 @@ function testSession( ? ({ threadResumeCompletionMarker: true } as never) : ({} as never), ), + subscribeServerConfig: (input) => client.subscribeServerConfig(input), ready: Effect.void, probe: Effect.void, closed: Effect.never, diff --git a/packages/client-runtime/src/state/vcs.test.ts b/packages/client-runtime/src/state/vcs.test.ts index 0a6264c62078..d7a4692fc317 100644 --- a/packages/client-runtime/src/state/vcs.test.ts +++ b/packages/client-runtime/src/state/vcs.test.ts @@ -86,6 +86,7 @@ function session(client: WsRpcProtocolClient): RpcSession { return { client, initialConfig: Effect.never, + subscribeServerConfig: (input) => client.subscribeServerConfig(input), ready: Effect.void, probe: Effect.void, closed: Effect.never, diff --git a/packages/client-runtime/src/state/vcsAction.test.ts b/packages/client-runtime/src/state/vcsAction.test.ts index b936246dc823..905972975606 100644 --- a/packages/client-runtime/src/state/vcsAction.test.ts +++ b/packages/client-runtime/src/state/vcsAction.test.ts @@ -84,6 +84,7 @@ function session(client: WsRpcProtocolClient): RpcSession { return { client, initialConfig: Effect.never, + subscribeServerConfig: (input) => client.subscribeServerConfig(input), ready: Effect.void, probe: Effect.void, closed: Effect.never, From 0e185b07f36fd51007fd7ac97257ca66767ebcde Mon Sep 17 00:00:00 2001 From: Theo Browne Date: Tue, 1 Sep 2026 03:38:40 -0700 Subject: [PATCH 07/10] refactor(client-runtime): import config projection directly --- packages/client-runtime/src/state/server.test.ts | 2 +- packages/client-runtime/src/state/server.ts | 5 ----- 2 files changed, 1 insertion(+), 6 deletions(-) diff --git a/packages/client-runtime/src/state/server.test.ts b/packages/client-runtime/src/state/server.test.ts index c383bd4dc44f..ea170c22e830 100644 --- a/packages/client-runtime/src/state/server.test.ts +++ b/packages/client-runtime/src/state/server.test.ts @@ -30,7 +30,6 @@ import * as Persistence from "../platform/persistence.ts"; import type { WsRpcProtocolClient } from "../rpc/protocol.ts"; import type { RpcSession } from "../rpc/session.ts"; import { - applyServerConfigProjection, makeEnvironmentServerConfigState, isLegacyUpdateHandoffLoss, matchesServerUpdateReadyEvent, @@ -42,6 +41,7 @@ import { serverUpdateStateForServerVersion, validateServerUpdateReadyEvent, } from "./server.ts"; +import { applyServerConfigProjection } from "./serverConfigProjection.ts"; const CONFIG = { availableEditors: [], diff --git a/packages/client-runtime/src/state/server.ts b/packages/client-runtime/src/state/server.ts index e88216065d14..dc763a6e9f31 100644 --- a/packages/client-runtime/src/state/server.ts +++ b/packages/client-runtime/src/state/server.ts @@ -47,11 +47,6 @@ import { withoutEnvironmentThemes, } from "./serverConfigProjection.ts"; -export { - applyServerConfigProjection, - type ServerConfigProjection, -} from "./serverConfigProjection.ts"; - export type ServerUpdateStage = "downloading" | "installing" | "resuming"; export type ServerUpdateState = From 491e5ac6904ef8429215a4004007c7779fff3c44 Mon Sep 17 00:00:00 2001 From: Theo Browne Date: Tue, 1 Sep 2026 03:42:02 -0700 Subject: [PATCH 08/10] fix(client-runtime): export config projection type --- packages/client-runtime/src/state/server.ts | 3 +++ 1 file changed, 3 insertions(+) diff --git a/packages/client-runtime/src/state/server.ts b/packages/client-runtime/src/state/server.ts index dc763a6e9f31..176b2631e0c5 100644 --- a/packages/client-runtime/src/state/server.ts +++ b/packages/client-runtime/src/state/server.ts @@ -47,6 +47,9 @@ import { withoutEnvironmentThemes, } from "./serverConfigProjection.ts"; +// Exported server state includes this type in its inferred public return type. +export type { ServerConfigProjection } from "./serverConfigProjection.ts"; + export type ServerUpdateStage = "downloading" | "installing" | "resuming"; export type ServerUpdateState = From a68af458ec521b9d447ee0598de061893886c2d1 Mon Sep 17 00:00:00 2001 From: Theo Browne Date: Tue, 1 Sep 2026 03:44:12 -0700 Subject: [PATCH 09/10] docs(client-runtime): explain environment theme lifecycle --- .../client-runtime/src/state/serverConfigProjection.ts | 9 +++++++-- 1 file changed, 7 insertions(+), 2 deletions(-) diff --git a/packages/client-runtime/src/state/serverConfigProjection.ts b/packages/client-runtime/src/state/serverConfigProjection.ts index dd893ab6fbec..6f4a812cf7e9 100644 --- a/packages/client-runtime/src/state/serverConfigProjection.ts +++ b/packages/client-runtime/src/state/serverConfigProjection.ts @@ -7,7 +7,11 @@ export interface ServerConfigProjection { readonly source: "cache" | "live"; } -/** Removes machine-owned themes before a config enters durable storage or a replay snapshot. */ +/** + * Cached config keeps the provider and model catalog available across reconnects. + * Published themes are current machine state, so a cache could restore themes + * that the machine no longer publishes. Replay sends themes as a separate event. + */ export function withoutEnvironmentThemes(config: ServerConfig): ServerConfig { if (config.environmentThemes === undefined) return config; const { environmentThemes: _ephemeral, ...rest } = config; @@ -21,7 +25,8 @@ export function applyServerConfigProjection( switch (event.type) { case "snapshot": { // Wire snapshots never contain published themes. Keep the previous set - // until a capable server sends its authoritative theme event. + // until a capable server sends its authoritative theme event. A legacy + // server cannot send a later removal, so a downgrade must clear the set. const carried = event.config.environment.capabilities.environmentThemes === true && Option.isSome(current) ? current.value.config.environmentThemes From 9114a74cf67cefcb2cf57ef2242ac58cdb91f12b Mon Sep 17 00:00:00 2001 From: Theo Browne Date: Tue, 1 Sep 2026 04:10:01 -0700 Subject: [PATCH 10/10] fix(client): recover config state after session failure --- .../client-runtime/src/rpc/session.test.ts | 164 +++++++++++++++++- packages/client-runtime/src/rpc/session.ts | 18 +- 2 files changed, 180 insertions(+), 2 deletions(-) diff --git a/packages/client-runtime/src/rpc/session.test.ts b/packages/client-runtime/src/rpc/session.test.ts index 9fdf5907078a..aedd85c5de47 100644 --- a/packages/client-runtime/src/rpc/session.test.ts +++ b/packages/client-runtime/src/rpc/session.test.ts @@ -20,15 +20,21 @@ import * as Option from "effect/Option"; import * as Queue from "effect/Queue"; import * as Schema from "effect/Schema"; import * as Stream from "effect/Stream"; +import * as SubscriptionRef from "effect/SubscriptionRef"; import * as TestClock from "effect/testing/TestClock"; import * as Socket from "effect/unstable/socket/Socket"; import { + AVAILABLE_CONNECTION_STATE, + ConnectionBlockedError, ConnectionTransientError, PrimaryConnectionTarget, type PreparedConnection, } from "../connection/model.ts"; +import * as EnvironmentSupervisor from "../connection/supervisor.ts"; +import * as Persistence from "../platform/persistence.ts"; import * as RpcSession from "./session.ts"; +import { makeEnvironmentServerConfigState } from "../state/server.ts"; import { applyServerConfigProjection } from "../state/serverConfigProjection.ts"; type SocketEventType = "open" | "message" | "close" | "error"; @@ -194,9 +200,10 @@ const makeFactory = Effect.fn("TestRpcSessionFactory.make")(function* ( const awaitSocket = Effect.fn("TestRpcSessionFactory.awaitSocket")(function* ( sockets: ReadonlyArray, + index = 0, ) { for (let attempt = 0; attempt < 100; attempt += 1) { - const socket = sockets[0]; + const socket = sockets[index]; if (socket) { return socket; } @@ -671,6 +678,161 @@ describe("RpcSessionFactory", () => { ), ); + it.effect.each([{ failure: "defect" as const }, { failure: "typed" as const }])( + "keeps durable config state alive after an owned $failure failure", + ({ failure }) => + Effect.scoped( + Effect.gen(function* () { + const { factory, sockets } = yield* makeFactory({ environmentThemes: true }); + const firstSession = yield* factory.connect(PREPARED); + const firstReady = yield* Effect.forkChild(firstSession.ready); + const firstSocket = yield* awaitSocket(sockets); + firstSocket.open(); + yield* completeInitialConfig(firstSocket, ENCODED_THEME_SERVER_CONFIG, { + environmentThemes: true, + }); + yield* Fiber.join(firstReady); + + const activeSession = yield* SubscriptionRef.make(Option.some(firstSession)); + const supervisor = EnvironmentSupervisor.EnvironmentSupervisor.of({ + target: TARGET, + state: yield* SubscriptionRef.make(AVAILABLE_CONNECTION_STATE), + session: activeSession, + prepared: yield* SubscriptionRef.make(Option.some(PREPARED)), + connect: Effect.void, + disconnect: Effect.void, + retryNow: Effect.void, + } satisfies EnvironmentSupervisor.EnvironmentSupervisor["Service"]); + const cache = Persistence.EnvironmentCacheStore.of({ + loadShell: () => Effect.succeed(Option.none()), + saveShell: () => Effect.void, + loadThread: () => Effect.succeed(Option.none()), + saveThread: () => Effect.void, + removeThread: () => Effect.void, + loadServerConfig: () => Effect.succeed(Option.none()), + saveServerConfig: () => Effect.void, + loadVcsRefs: () => Effect.succeed(Option.none()), + saveVcsRefs: () => Effect.void, + removeVcsRefs: () => Effect.void, + clearVcsRefs: () => Effect.void, + clear: () => Effect.void, + }); + const configState = yield* makeEnvironmentServerConfigState(true).pipe( + Effect.provideService(EnvironmentSupervisor.EnvironmentSupervisor, supervisor), + Effect.provideService(Persistence.EnvironmentCacheStore, cache), + ); + const awaitConfig = (predicate: (config: ServerConfigType) => boolean) => + SubscriptionRef.changes(configState).pipe( + Stream.filter(Option.isSome), + Stream.map((projection) => projection.value.config), + Stream.filter(predicate), + Stream.runHead, + Effect.map(Option.getOrThrow), + ); + + const firstThemes = [ + { + id: "first-theme", + name: "First theme", + appearance: "dark" as const, + canvas: "#111111", + accent: "#ffffff", + }, + ]; + const firstThemeState = yield* awaitConfig( + (config) => config.environmentThemes?.[0]?.id === "first-theme", + ).pipe(Effect.forkChild); + yield* publishConfigEvents(firstSocket, [ + { + version: 1, + type: "environmentThemesUpdated", + payload: { themes: firstThemes }, + }, + ]); + expect((yield* Fiber.join(firstThemeState)).environmentThemes).toEqual(firstThemes); + + const firstClosed = yield* firstSession.closed.pipe(Effect.exit, Effect.forkChild); + const firstRequest = yield* awaitRequest(firstSocket); + firstSocket.serverMessage( + failure === "defect" + ? encodeJson({ + _tag: "Defect", + defect: encodeDefect(new Error("config stream died")), + }) + : encodeJson({ + _tag: "Exit", + requestId: firstRequest.id, + exit: { + _tag: "Failure", + cause: [ + { + _tag: "Fail", + error: { + _tag: "EnvironmentAuthorizationError", + message: "config subscription rejected", + requiredScope: "orchestration:read", + }, + }, + ], + }, + }), + ); + const firstClosedExit = yield* Fiber.join(firstClosed); + expect(Exit.isFailure(firstClosedExit)).toBe(true); + if (failure === "typed" && Exit.isFailure(firstClosedExit)) { + expect(Cause.squash(firstClosedExit.cause)).toBeInstanceOf(ConnectionBlockedError); + expect(Cause.squash(firstClosedExit.cause)).toMatchObject({ reason: "permission" }); + } + yield* SubscriptionRef.set(activeSession, Option.none()); + + const recoveredConfig = { + ...THEME_SERVER_CONFIG, + environment: { + ...THEME_SERVER_CONFIG.environment, + label: "Recovered environment", + }, + } satisfies ServerConfigType; + const secondSession = yield* factory.connect(PREPARED); + const secondReady = yield* Effect.forkChild(secondSession.ready); + const secondSocket = yield* awaitSocket(sockets, 1); + secondSocket.open(); + yield* completeInitialConfig(secondSocket, encodeServerConfig(recoveredConfig), { + environmentThemes: true, + }); + yield* Fiber.join(secondReady); + + const recoveredState = yield* awaitConfig( + (config) => config.environment.label === "Recovered environment", + ).pipe(Effect.forkChild); + yield* SubscriptionRef.set(activeSession, Option.some(secondSession)); + expect((yield* Fiber.join(recoveredState)).environmentThemes).toEqual(firstThemes); + + const recoveredThemes = [ + { + id: "recovered-theme", + name: "Recovered theme", + appearance: "dark" as const, + canvas: "#000000", + accent: "#eeeeee", + }, + ]; + const liveRecoveredState = yield* awaitConfig( + (config) => config.environmentThemes?.[0]?.id === "recovered-theme", + ).pipe(Effect.forkChild); + yield* publishConfigEvents(secondSocket, [ + { + version: 1, + type: "environmentThemesUpdated", + payload: { themes: recoveredThemes }, + }, + ]); + expect((yield* Fiber.join(liveRecoveredState)).environmentThemes).toEqual( + recoveredThemes, + ); + }), + ), + ); + it.effect.each<{ readonly event: ServerConfigStreamEventType; readonly expectedConfig: Partial; diff --git a/packages/client-runtime/src/rpc/session.ts b/packages/client-runtime/src/rpc/session.ts index 2043d0b25d90..7d975be5c9d3 100644 --- a/packages/client-runtime/src/rpc/session.ts +++ b/packages/client-runtime/src/rpc/session.ts @@ -19,7 +19,7 @@ import type * as Scope from "effect/Scope"; import * as Stream from "effect/Stream"; import type * as Rpc from "effect/unstable/rpc/Rpc"; import * as RpcClient from "effect/unstable/rpc/RpcClient"; -import type * as RpcClientError from "effect/unstable/rpc/RpcClientError"; +import * as RpcClientError from "effect/unstable/rpc/RpcClientError"; import * as RpcSerialization from "effect/unstable/rpc/RpcSerialization"; import * as Socket from "effect/unstable/socket/Socket"; @@ -285,6 +285,22 @@ export const make = Effect.fn("RpcSessionFactory.make")(function* ( Stream.merge(updates, terminal, { haltStrategy: "either" }), ); }), + ).pipe( + Stream.catchCause((cause) => { + if (Cause.hasInterruptsOnly(cause)) { + return Stream.failCause(cause); + } + // The supervisor keeps the original cause. Shared durable consumers + // need a transport-shaped failure so they wait for its replacement. + return Stream.fail( + new RpcClientError.RpcClientError({ + reason: new RpcClientError.RpcClientDefect({ + message: `${connection.label} config subscription failed.`, + cause, + }), + }), + ); + }), ); const subscribeServerConfig = (input: ServerConfigSubscriptionInput) => Equal.equals(input, serverConfigInput)