authorgravatar for adam.bogucki2018@gmail.comAdamulek123 <adam.bogucki2018@gmail.com> 2026-09-01 13:23:50+02:00
committergravatar for noreply@github.comGitHub <noreply@github.com> 2026-09-01 04:23:50-07:00
logb883fc066ea5c9bebbe1c3e9b4bc2471aab3685f
tree169b7fd375306cf1d71438fbd15eb3b227e95a34
parente86604d3372acccd9f6a33a2c4ae46f4e2685541
signature Signed by PGP key B5690EEEBB952194

perf(client-runtime): halve server config bootstrap traffic (#8367)

Reuse one server config subscription for session bootstrap and live updates. Preserve environment theme opt-in, replay, deletion, slow subscriber recovery, and config stream failure handling. Co-authored-by: Adamulek123 <adam.bogucki2018@gmail.com>

20 files changed, 1053 insertions(+), 151 deletions(-)

apps/web/src/cloud/linkEnvironment.test.ts+1
......@@ -91,6 +91,7 @@ function registryLayer(options?: {
9191 const session: RpcSession = {
9292 client,
9393 initialConfig: Effect.never,
94 subscribeServerConfig: (input) => client.subscribeServerConfig(input),
9495 ready: Effect.void,
9596 probe: Effect.void,
9697 closed: Effect.never,
apps/web/src/connection/runtime.ts+4-1
......@@ -30,7 +30,10 @@ type ConnectionLayerSource =
3030 | typeof backgroundActivityObserverLayer
3131 | typeof backgroundActivityReporterLayer;
3232
33const providedClientConnectionLayer = Layer.merge(Connection.layer, snapshotLoaderLayer).pipe(
33const providedClientConnectionLayer = Layer.merge(
34 Connection.layerWithOptions({ environmentThemes: true }),
35 snapshotLoaderLayer,
36).pipe(
3437 Layer.provideMerge(
3538 Layer.mergeAll(
3639 runtimeContextLayer,
packages/client-runtime/src/connection/layer.ts+26-27
......@@ -15,30 +15,29 @@ const resolverLayer = ConnectionResolver.layer.pipe(
1515 Layer.provide(RemoteEnvironmentAuthorization.layer),
1616);
1717
18const driverLayer = ConnectionDriver.layer.pipe(
19 Layer.provide(Layer.mergeAll(resolverLayer, RpcSession.layer)),
20);
21
22const registryLayer = EnvironmentRegistry.layer.pipe(Layer.provide(driverLayer));
23
24const onboardingLayer = ConnectionOnboarding.layer.pipe(Layer.provide(registryLayer));
25
26const connectionServicesLayer = Layer.mergeAll(
27 registryLayer,
28 RelayEnvironmentDiscovery.layer,
29 onboardingLayer,
30);
31
32const connectionStartupLayer = Layer.effectDiscard(
33 Effect.gen(function* () {
34 const registry = yield* EnvironmentRegistry.EnvironmentRegistry;
35 const platformSource = yield* PlatformConnectionSource.PlatformConnectionSource;
36 yield* registry.start;
37 yield* platformSource.registrations.pipe(
38 Stream.runForEach(registry.reconcilePlatform),
39 Effect.forkScoped,
40 );
41 }).pipe(Effect.withSpan("clientRuntime.connection.application.start")),
42);
43
44export const layer = connectionStartupLayer.pipe(Layer.provideMerge(connectionServicesLayer));
18export function layerWithOptions(options: RpcSession.RpcSessionOptions) {
19 const driverLayer = ConnectionDriver.layer.pipe(
20 Layer.provide(Layer.mergeAll(resolverLayer, RpcSession.layerWithOptions(options))),
21 );
22 const registryLayer = EnvironmentRegistry.layer.pipe(Layer.provide(driverLayer));
23 const onboardingLayer = ConnectionOnboarding.layer.pipe(Layer.provide(registryLayer));
24 const connectionServicesLayer = Layer.mergeAll(
25 registryLayer,
26 RelayEnvironmentDiscovery.layer,
27 onboardingLayer,
28 );
29 const connectionStartupLayer = Layer.effectDiscard(
30 Effect.gen(function* () {
31 const registry = yield* EnvironmentRegistry.EnvironmentRegistry;
32 const platformSource = yield* PlatformConnectionSource.PlatformConnectionSource;
33 yield* registry.start;
34 yield* platformSource.registrations.pipe(
35 Stream.runForEach(registry.reconcilePlatform),
36 Effect.forkScoped,
37 );
38 }).pipe(Effect.withSpan("clientRuntime.connection.application.start")),
39 );
40 return connectionStartupLayer.pipe(Layer.provideMerge(connectionServicesLayer));
41}
42
43export const layer = layerWithOptions({});
packages/client-runtime/src/connection/registry.test.ts+2
......@@ -356,6 +356,8 @@ const makeHarness = Effect.fn("TestEnvironmentRegistry.makeHarness")(function* (
356356 Effect.succeed({
357357 client: {} as RpcSession.RpcSession["client"],
358358 initialConfig: Effect.die(new Error("Config is not used by registry tests.")),
359 subscribeServerConfig: () =>
360 Stream.die(new Error("Config is not used by registry tests.")),
359361 ready: Effect.void,
360362 probe: Effect.void,
361363 closed: Deferred.await(closed),
packages/client-runtime/src/connection/supervisor.test.ts+1
......@@ -163,6 +163,7 @@ const makeHarness = Effect.fn("TestConnectionHarness.make")(function* (options?:
163163 Effect.succeed({
164164 client: TEST_RPC_CLIENT,
165165 initialConfig: Effect.die(new Error("Initial config is not used by supervisor tests.")),
166 subscribeServerConfig: (input) => TEST_RPC_CLIENT.subscribeServerConfig(input),
166167 ready: options?.ready?.(attempt) ?? Effect.void,
167168 probe: options?.probe?.(attempt) ?? Effect.void,
168169 closed: Deferred.await(closed),
packages/client-runtime/src/operations/commands.test.ts+1
......@@ -57,6 +57,7 @@ const makeSupervisor = Effect.fn("TestEnvironmentCommands.makeSupervisor")(funct
5757 const session: RpcSession.RpcSession = {
5858 client,
5959 initialConfig: Effect.never,
60 subscribeServerConfig: (input) => client.subscribeServerConfig(input),
6061 ready: Effect.void,
6162 probe: Effect.void,
6263 closed: Effect.never,
packages/client-runtime/src/rpc/client.test.ts+36
......@@ -1,6 +1,8 @@
11import {
2 DEFAULT_SERVER_SETTINGS,
23 EnvironmentId,
34 type RelayClientInstallProgressEvent,
5 type ServerConfigStreamEvent,
46 WS_METHODS,
57} from "@t3tools/contracts";
68import { describe, expect, it } from "@effect/vitest";
......@@ -47,6 +49,7 @@ function session(client: WsRpcProtocolClient): RpcSession.RpcSession {
4749 return {
4850 client,
4951 initialConfig: Effect.never,
52 subscribeServerConfig: (input) => client.subscribeServerConfig(input),
5053 ready: Effect.void,
5154 probe: Effect.void,
5255 closed: Effect.never,
......@@ -77,6 +80,39 @@ const makeHarness = Effect.fn("TestEnvironmentRpc.makeHarness")(function* () {
7780});
7881
7982describe("environment RPC", () => {
83 it.effect("reuses the session config stream instead of opening a duplicate subscription", () =>
84 Effect.gen(function* () {
85 const event: ServerConfigStreamEvent = {
86 version: 1,
87 type: "settingsUpdated",
88 payload: { settings: DEFAULT_SERVER_SETTINGS },
89 };
90 let duplicateSubscriptions = 0;
91 const client = {
92 [WS_METHODS.subscribeServerConfig]: () => {
93 duplicateSubscriptions += 1;
94 return Stream.never;
95 },
96 } as unknown as WsRpcProtocolClient;
97 const { activeSession, supervisor } = yield* makeHarness();
98 yield* SubscriptionRef.set(
99 activeSession,
100 Option.some({
101 ...session(client),
102 subscribeServerConfig: () => Stream.succeed(event),
103 }),
104 );
105
106 const received = yield* subscribe(WS_METHODS.subscribeServerConfig, {}).pipe(
107 Stream.runHead,
108 Effect.provideService(EnvironmentSupervisor.EnvironmentSupervisor, supervisor),
109 );
110
111 expect(received).toEqual(Option.some(event));
112 expect(duplicateSubscriptions).toBe(0);
113 }),
114 );
115
80116 it.effect("observes unary requests until they complete", () =>
81117 Effect.gen(function* () {
82118 const observations: string[] = [];
packages/client-runtime/src/rpc/client.ts+5-1
......@@ -203,7 +203,11 @@ export function subscribeDynamic<TTag extends EnvironmentSubscriptionRpcTag>(
203203 Option.match({
204204 onNone: () => Stream.empty,
205205 onSome: (session) => {
206 const method = session.client[tag] as (
206 const method = (
207 tag === WS_METHODS.subscribeServerConfig
208 ? session.subscribeServerConfig
209 : session.client[tag]
210 ) as (
207211 input: EnvironmentRpcInput<TTag>,
208212 ) => Stream.Stream<
209213 EnvironmentRpcStreamValue<TTag>,
packages/client-runtime/src/rpc/session.test.ts+675-23
......@@ -1,24 +1,41 @@
11import {
22 DEFAULT_SERVER_SETTINGS,
33 EnvironmentId,
4 ProviderDriverKind,
5 ProviderInstanceId,
46 ServerConfig,
57 type ServerConfig as ServerConfigType,
8 ServerConfigStreamEvent,
9 type ServerConfigStreamEvent as ServerConfigStreamEventType,
610 WS_METHODS,
711} from "@t3tools/contracts";
812import { describe, expect, it } from "@effect/vitest";
13import * as Cause from "effect/Cause";
14import * as Deferred from "effect/Deferred";
915import * as Effect from "effect/Effect";
16import * as Exit from "effect/Exit";
1017import * as Fiber from "effect/Fiber";
1118import * as Layer from "effect/Layer";
19import * as Option from "effect/Option";
20import * as Queue from "effect/Queue";
1221import * as Schema from "effect/Schema";
22import * as Stream from "effect/Stream";
23import * as SubscriptionRef from "effect/SubscriptionRef";
1324import * as TestClock from "effect/testing/TestClock";
1425import * as Socket from "effect/unstable/socket/Socket";
1526
1627import {
28 AVAILABLE_CONNECTION_STATE,
29 ConnectionBlockedError,
1730 ConnectionTransientError,
1831 PrimaryConnectionTarget,
1932 type PreparedConnection,
2033} from "../connection/model.ts";
34import * as EnvironmentSupervisor from "../connection/supervisor.ts";
35import * as Persistence from "../platform/persistence.ts";
2136import * as RpcSession from "./session.ts";
37import { makeEnvironmentServerConfigState } from "../state/server.ts";
38import { applyServerConfigProjection } from "../state/serverConfigProjection.ts";
2239
2340type SocketEventType = "open" | "message" | "close" | "error";
2441type SocketEvent = {
......@@ -139,10 +156,24 @@ const RpcRequest = Schema.TaggedStruct("Request", {
139156 tag: Schema.String,
140157});
141158const decodeJson = Schema.decodeUnknownSync(Schema.fromJsonString(Schema.Unknown));
142const decodeRpcRequest = Schema.decodeUnknownSync(RpcRequest);
159const isRpcRequest = Schema.is(RpcRequest);
160const isPing = Schema.is(Schema.Struct({ _tag: Schema.Literal("Ping") }));
143161const encodeJson = Schema.encodeUnknownSync(Schema.fromJsonString(Schema.Unknown));
144162const encodeServerConfig = Schema.encodeSync(ServerConfig);
163const encodeServerConfigStreamEvent = Schema.encodeSync(ServerConfigStreamEvent);
164const encodeDefect = Schema.encodeSync(Schema.Defect());
145165const ENCODED_SERVER_CONFIG = encodeServerConfig(SERVER_CONFIG);
166const THEME_SERVER_CONFIG: ServerConfigType = {
167 ...SERVER_CONFIG,
168 environment: {
169 ...SERVER_CONFIG.environment,
170 capabilities: {
171 ...SERVER_CONFIG.environment.capabilities,
172 environmentThemes: true,
173 },
174 },
175};
176const ENCODED_THEME_SERVER_CONFIG = encodeServerConfig(THEME_SERVER_CONFIG);
146177const LEGACY_SERVER_CONFIG = {
147178 ...ENCODED_SERVER_CONFIG,
148179 environment: {
......@@ -153,23 +184,26 @@ const LEGACY_SERVER_CONFIG = {
153184 },
154185};
155186
156const makeFactory = Effect.fn("TestRpcSessionFactory.make")(function* () {
187const makeFactory = Effect.fn("TestRpcSessionFactory.make")(function* (
188 options: RpcSession.RpcSessionOptions = {},
189) {
157190 const sockets: TestWebSocket[] = [];
158191 const constructorLayer = Layer.succeed(Socket.WebSocketConstructor, (url) => {
159192 const socket = new TestWebSocket(url);
160193 sockets.push(socket);
161194 return socket as unknown as globalThis.WebSocket;
162195 });
163 const layer = RpcSession.layer.pipe(Layer.provide(constructorLayer));
196 const layer = RpcSession.layerWithOptions(options).pipe(Layer.provide(constructorLayer));
164197 const factory = yield* RpcSession.RpcSessionFactory.pipe(Effect.provide(layer));
165198 return { factory, sockets };
166199});
167200
168201const awaitSocket = Effect.fn("TestRpcSessionFactory.awaitSocket")(function* (
169202 sockets: ReadonlyArray<TestWebSocket>,
203 index = 0,
170204) {
171205 for (let attempt = 0; attempt < 100; attempt += 1) {
172 const socket = sockets[0];
206 const socket = sockets[index];
173207 if (socket) {
174208 return socket;
175209 }
......@@ -183,9 +217,9 @@ const awaitRequest = Effect.fn("TestRpcSessionFactory.awaitRequest")(function* (
183217 index = 0,
184218) {
185219 for (let attempt = 0; attempt < 100; attempt += 1) {
186 const request = socket.sent[index];
220 const request = socket.sent.map((message) => decodeJson(message)).filter(isRpcRequest)[index];
187221 if (request) {
188 return decodeRpcRequest(decodeJson(request));
222 return request;
189223 }
190224 yield* Effect.yieldNow;
191225 }
......@@ -195,21 +229,33 @@ const awaitRequest = Effect.fn("TestRpcSessionFactory.awaitRequest")(function* (
195229const completeInitialConfig = Effect.fn("TestRpcSessionFactory.completeInitialConfig")(function* (
196230 socket: TestWebSocket,
197231 config: unknown = ENCODED_SERVER_CONFIG,
232 payload: unknown = {},
198233) {
199234 const request = yield* awaitRequest(socket);
200235 expect(request).toMatchObject({
201236 _tag: "Request",
202 tag: WS_METHODS.serverGetConfig,
203 payload: {},
237 tag: WS_METHODS.subscribeServerConfig,
238 payload,
204239 });
205240 socket.serverMessage(
206241 encodeJson({
207 _tag: "Exit",
242 _tag: "Chunk",
208243 requestId: request.id,
209 exit: {
210 _tag: "Success",
211 value: config,
212 },
244 values: [{ version: 1, type: "snapshot", config }],
245 }),
246 );
247});
248
249const publishConfigEvents = Effect.fn("TestRpcSessionFactory.publishConfigEvents")(function* (
250 socket: TestWebSocket,
251 events: ReadonlyArray<ServerConfigStreamEventType>,
252) {
253 const request = yield* awaitRequest(socket);
254 socket.serverMessage(
255 encodeJson({
256 _tag: "Chunk",
257 requestId: request.id,
258 values: events.map((event) => encodeServerConfigStreamEvent(event)),
213259 }),
214260 );
215261});
......@@ -229,7 +275,9 @@ describe("RpcSessionFactory", () => {
229275
230276 const config = yield* session.initialConfig;
231277 expect(config).toEqual(SERVER_CONFIG);
232 expect(socket.sent).toHaveLength(1);
278 expect(socket.sent.map((message) => decodeJson(message)).filter(isRpcRequest)).toHaveLength(
279 1,
280 );
233281
234282 const probeFiber = yield* Effect.forkChild(session.probe);
235283 const probeRequest = yield* awaitRequest(socket, 1);
......@@ -250,19 +298,25 @@ describe("RpcSessionFactory", () => {
250298 );
251299 yield* Fiber.join(probeFiber);
252300
253 expect(socket.sent.map((request) => decodeRpcRequest(decodeJson(request)).tag)).toEqual([
254 WS_METHODS.serverGetConfig,
255 WS_METHODS.serverProbe,
256 ]);
301 expect(
302 socket.sent
303 .map((message) => decodeJson(message))
304 .filter(isRpcRequest)
305 .map((request) => request.tag),
306 ).toEqual([WS_METHODS.subscribeServerConfig, WS_METHODS.serverProbe]);
257307
258308 socket.close(1012, "service restart");
259309 const error = yield* Effect.flip(session.closed);
310 const configStreamError = yield* session
311 .subscribeServerConfig({})
312 .pipe(Stream.runDrain, Effect.flip);
260313
261314 expect(error).toBeInstanceOf(ConnectionTransientError);
262315 expect(error).toMatchObject({
263316 reason: "transport",
264317 message: "Test environment disconnected.",
265318 });
319 expect(configStreamError).toMatchObject({ _tag: "RpcClientError" });
266320 yield* Effect.yieldNow;
267321 expect(sockets).toHaveLength(1);
268322 }),
......@@ -287,6 +341,602 @@ describe("RpcSessionFactory", () => {
287341 }),
288342 );
289343
344 it.effect("replays current config and broadcasts updates to every subscriber", () =>
345 Effect.scoped(
346 Effect.gen(function* () {
347 const { factory, sockets } = yield* makeFactory();
348 const session = yield* factory.connect(PREPARED);
349 const readyFiber = yield* Effect.forkChild(session.ready);
350 const socket = yield* awaitSocket(sockets);
351 socket.open();
352 yield* completeInitialConfig(socket);
353 yield* Fiber.join(readyFiber);
354
355 const collectTwo = session
356 .subscribeServerConfig({})
357 .pipe(Stream.take(2), Stream.runCollect);
358 const firstSubscriber = yield* Effect.forkChild(collectTwo);
359 const secondSubscriber = yield* Effect.forkChild(collectTwo);
360 yield* Effect.yieldNow;
361
362 const shortcut = {
363 key: "k",
364 metaKey: false,
365 ctrlKey: false,
366 shiftKey: false,
367 altKey: false,
368 modKey: true,
369 };
370 const request = yield* awaitRequest(socket);
371 socket.serverMessage(
372 encodeJson({
373 _tag: "Chunk",
374 requestId: request.id,
375 values: [
376 {
377 version: 1,
378 type: "keybindingsUpdated",
379 payload: {
380 keybindings: [{ command: "terminal.toggle", shortcut }],
381 issues: [],
382 },
383 },
384 ],
385 }),
386 );
387
388 const firstEvents = Array.from(yield* Fiber.join(firstSubscriber));
389 const secondEvents = Array.from(yield* Fiber.join(secondSubscriber));
390 expect(firstEvents.map((event) => event.type)).toEqual(["snapshot", "keybindingsUpdated"]);
391 expect(secondEvents).toEqual(firstEvents);
392
393 const replay = yield* session.subscribeServerConfig({}).pipe(Stream.runHead);
394 expect(replay).toMatchObject({
395 _tag: "Some",
396 value: {
397 type: "snapshot",
398 config: { keybindings: [{ command: "terminal.toggle", shortcut }] },
399 },
400 });
401 }),
402 ),
403 );
404
405 it.effect("shares only a config subscription with the same theme opt-in", () =>
406 Effect.scoped(
407 Effect.gen(function* () {
408 const { factory, sockets } = yield* makeFactory({ environmentThemes: true });
409 const session = yield* factory.connect(PREPARED);
410 const readyFiber = yield* Effect.forkChild(session.ready);
411 const socket = yield* awaitSocket(sockets);
412 socket.open();
413 yield* completeInitialConfig(socket, ENCODED_THEME_SERVER_CONFIG, {
414 environmentThemes: true,
415 });
416 yield* Fiber.join(readyFiber);
417
418 const shared = yield* session
419 .subscribeServerConfig({ environmentThemes: true })
420 .pipe(Stream.runHead);
421 expect(shared).toMatchObject({ _tag: "Some", value: { type: "snapshot" } });
422 expect(socket.sent.map((message) => decodeJson(message)).filter(isRpcRequest)).toHaveLength(
423 1,
424 );
425
426 const fallbackFiber = yield* session
427 .subscribeServerConfig({})
428 .pipe(Stream.runHead, Effect.forkChild);
429 const fallbackRequest = yield* awaitRequest(socket, 1);
430 expect(fallbackRequest).toMatchObject({
431 tag: WS_METHODS.subscribeServerConfig,
432 payload: {},
433 });
434 socket.serverMessage(
435 encodeJson({
436 _tag: "Chunk",
437 requestId: fallbackRequest.id,
438 values: [
439 {
440 version: 1,
441 type: "snapshot",
442 config: ENCODED_THEME_SERVER_CONFIG,
443 },
444 ],
445 }),
446 );
447 expect(yield* Fiber.join(fallbackFiber)).toMatchObject({
448 _tag: "Some",
449 value: { type: "snapshot" },
450 });
451 }),
452 ),
453 );
454
455 it.effect("replays theme updates and deletion as authoritative events", () =>
456 Effect.scoped(
457 Effect.gen(function* () {
458 const { factory, sockets } = yield* makeFactory({ environmentThemes: true });
459 const session = yield* factory.connect(PREPARED);
460 const readyFiber = yield* Effect.forkChild(session.ready);
461 const socket = yield* awaitSocket(sockets);
462 socket.open();
463 yield* completeInitialConfig(socket, ENCODED_THEME_SERVER_CONFIG, {
464 environmentThemes: true,
465 });
466 yield* Fiber.join(readyFiber);
467
468 const firstThemes = [
469 {
470 id: "nightfall",
471 name: "Nightfall",
472 appearance: "dark" as const,
473 canvas: "#1a1b26",
474 accent: "#7aa2f7",
475 },
476 ];
477 const replacementThemes = [
478 {
479 id: "midnight",
480 name: "Midnight",
481 appearance: "dark" as const,
482 canvas: "#000000",
483 accent: "#ffffff",
484 },
485 ];
486 const subscriberStarted = yield* Deferred.make<void>();
487 const subscriber = yield* session.subscribeServerConfig({ environmentThemes: true }).pipe(
488 Stream.mapEffect((event) =>
489 Deferred.succeed(subscriberStarted, undefined).pipe(Effect.as(event)),
490 ),
491 Stream.take(4),
492 Stream.runCollect,
493 Effect.forkChild,
494 );
495 yield* Deferred.await(subscriberStarted);
496 yield* publishConfigEvents(socket, [
497 {
498 version: 1,
499 type: "environmentThemesUpdated",
500 payload: { themes: firstThemes },
501 },
502 {
503 version: 1,
504 type: "environmentThemesUpdated",
505 payload: { themes: replacementThemes },
506 },
507 {
508 version: 1,
509 type: "environmentThemesUpdated",
510 payload: { themes: [] },
511 },
512 ]);
513
514 const liveEvents = Array.from(yield* Fiber.join(subscriber));
515 expect(liveEvents.map((event) => event.type)).toEqual([
516 "snapshot",
517 "environmentThemesUpdated",
518 "environmentThemesUpdated",
519 "environmentThemesUpdated",
520 ]);
521 expect(liveEvents[2]).toMatchObject({ payload: { themes: replacementThemes } });
522
523 const replay = Array.from(
524 yield* session
525 .subscribeServerConfig({ environmentThemes: true })
526 .pipe(Stream.take(2), Stream.runCollect),
527 );
528 expect(replay.map((event) => event.type)).toEqual(["snapshot", "environmentThemesUpdated"]);
529 expect(replay[1]).toMatchObject({ payload: { themes: [] } });
530
531 let projection = applyServerConfigProjection(Option.none(), {
532 version: 1,
533 type: "snapshot",
534 config: THEME_SERVER_CONFIG,
535 });
536 projection = applyServerConfigProjection(projection, {
537 version: 1,
538 type: "environmentThemesUpdated",
539 payload: { themes: firstThemes },
540 });
541 for (const event of replay) {
542 projection = applyServerConfigProjection(projection, event);
543 }
544 expect(Option.getOrThrow(projection).config.environmentThemes).toBeUndefined();
545 }),
546 ),
547 );
548
549 it.effect("recovers a slow subscriber after it misses theme deletion", () =>
550 Effect.scoped(
551 Effect.gen(function* () {
552 const { factory, sockets } = yield* makeFactory({ environmentThemes: true });
553 const session = yield* factory.connect(PREPARED);
554 const readyFiber = yield* Effect.forkChild(session.ready);
555 const socket = yield* awaitSocket(sockets);
556 socket.open();
557 yield* completeInitialConfig(socket, ENCODED_THEME_SERVER_CONFIG, {
558 environmentThemes: true,
559 });
560 yield* Fiber.join(readyFiber);
561
562 const slowSubscriberStarted = yield* Deferred.make<void>();
563 const releaseSlowSubscriber = yield* Deferred.make<void>();
564 let firstEvent = true;
565 const slowSubscriber = yield* session
566 .subscribeServerConfig({ environmentThemes: true })
567 .pipe(
568 Stream.mapEffect((event) => {
569 if (!firstEvent) return Effect.succeed(event);
570 firstEvent = false;
571 return Deferred.succeed(slowSubscriberStarted, undefined).pipe(
572 Effect.andThen(Deferred.await(releaseSlowSubscriber)),
573 Effect.as(event),
574 );
575 }),
576 Stream.take(3),
577 Stream.runCollect,
578 Effect.forkChild,
579 );
580 yield* Deferred.await(slowSubscriberStarted);
581
582 const firstThemes = [
583 {
584 id: "nightfall",
585 name: "Nightfall",
586 appearance: "dark" as const,
587 canvas: "#1a1b26",
588 accent: "#7aa2f7",
589 },
590 ];
591 const themeEvents: ServerConfigStreamEventType[] = [
592 {
593 version: 1,
594 type: "environmentThemesUpdated",
595 payload: { themes: firstThemes },
596 },
597 {
598 version: 1,
599 type: "environmentThemesUpdated",
600 payload: {
601 themes: [{ ...firstThemes[0]!, name: "Nightfall 2" }],
602 },
603 },
604 {
605 version: 1,
606 type: "environmentThemesUpdated",
607 payload: { themes: [] },
608 },
609 ];
610 const settingsEvents = Array.from(
611 { length: 65 },
612 (): ServerConfigStreamEventType => ({
613 version: 1,
614 type: "settingsUpdated",
615 payload: { settings: DEFAULT_SERVER_SETTINGS },
616 }),
617 );
618 const allEvents = [...themeEvents, ...settingsEvents];
619 const observedByFastSubscriber = yield* Queue.unbounded<ServerConfigStreamEventType>();
620 yield* session.subscribeServerConfig({ environmentThemes: true }).pipe(
621 Stream.runForEach((event) => Queue.offer(observedByFastSubscriber, event)),
622 Effect.forkChild,
623 );
624 expect((yield* Queue.take(observedByFastSubscriber)).type).toBe("snapshot");
625 for (const event of allEvents) {
626 yield* publishConfigEvents(socket, [event]);
627 expect(yield* Queue.take(observedByFastSubscriber)).toEqual(event);
628 }
629 yield* Deferred.succeed(releaseSlowSubscriber, undefined);
630
631 const recovered = Array.from(yield* Fiber.join(slowSubscriber));
632 expect(recovered.map((event) => event.type)).toEqual([
633 "snapshot",
634 "snapshot",
635 "environmentThemesUpdated",
636 ]);
637 expect(recovered[2]).toMatchObject({ payload: { themes: [] } });
638
639 let projection = applyServerConfigProjection(Option.none(), {
640 version: 1,
641 type: "snapshot",
642 config: THEME_SERVER_CONFIG,
643 });
644 projection = applyServerConfigProjection(projection, themeEvents[0]!);
645 for (const event of recovered.slice(1)) {
646 projection = applyServerConfigProjection(projection, event);
647 }
648 expect(Option.getOrThrow(projection).config.environmentThemes).toBeUndefined();
649 }),
650 ),
651 );
652
653 it.effect("closes the session when the config source dies", () =>
654 Effect.scoped(
655 Effect.gen(function* () {
656 const { factory, sockets } = yield* makeFactory();
657 const session = yield* factory.connect(PREPARED);
658 const readyFiber = yield* Effect.forkChild(session.ready);
659 const socket = yield* awaitSocket(sockets);
660 socket.open();
661 yield* completeInitialConfig(socket);
662 yield* Fiber.join(readyFiber);
663
664 const closedFiber = yield* session.closed.pipe(Effect.exit, Effect.forkChild);
665 socket.serverMessage(
666 encodeJson({
667 _tag: "Defect",
668 defect: encodeDefect(new Error("config stream died")),
669 }),
670 );
671
672 const closed = yield* Fiber.join(closedFiber);
673 expect(Exit.isFailure(closed)).toBe(true);
674 if (Exit.isFailure(closed)) {
675 expect(Cause.hasDies(closed.cause)).toBe(true);
676 }
677 }),
678 ),
679 );
680
681 it.effect.each([{ failure: "defect" as const }, { failure: "typed" as const }])(
682 "keeps durable config state alive after an owned $failure failure",
683 ({ failure }) =>
684 Effect.scoped(
685 Effect.gen(function* () {
686 const { factory, sockets } = yield* makeFactory({ environmentThemes: true });
687 const firstSession = yield* factory.connect(PREPARED);
688 const firstReady = yield* Effect.forkChild(firstSession.ready);
689 const firstSocket = yield* awaitSocket(sockets);
690 firstSocket.open();
691 yield* completeInitialConfig(firstSocket, ENCODED_THEME_SERVER_CONFIG, {
692 environmentThemes: true,
693 });
694 yield* Fiber.join(firstReady);
695
696 const activeSession = yield* SubscriptionRef.make(Option.some(firstSession));
697 const supervisor = EnvironmentSupervisor.EnvironmentSupervisor.of({
698 target: TARGET,
699 state: yield* SubscriptionRef.make(AVAILABLE_CONNECTION_STATE),
700 session: activeSession,
701 prepared: yield* SubscriptionRef.make(Option.some(PREPARED)),
702 connect: Effect.void,
703 disconnect: Effect.void,
704 retryNow: Effect.void,
705 } satisfies EnvironmentSupervisor.EnvironmentSupervisor["Service"]);
706 const cache = Persistence.EnvironmentCacheStore.of({
707 loadShell: () => Effect.succeed(Option.none()),
708 saveShell: () => Effect.void,
709 loadThread: () => Effect.succeed(Option.none()),
710 saveThread: () => Effect.void,
711 removeThread: () => Effect.void,
712 loadServerConfig: () => Effect.succeed(Option.none()),
713 saveServerConfig: () => Effect.void,
714 loadVcsRefs: () => Effect.succeed(Option.none()),
715 saveVcsRefs: () => Effect.void,
716 removeVcsRefs: () => Effect.void,
717 clearVcsRefs: () => Effect.void,
718 clear: () => Effect.void,
719 });
720 const configState = yield* makeEnvironmentServerConfigState(true).pipe(
721 Effect.provideService(EnvironmentSupervisor.EnvironmentSupervisor, supervisor),
722 Effect.provideService(Persistence.EnvironmentCacheStore, cache),
723 );
724 const awaitConfig = (predicate: (config: ServerConfigType) => boolean) =>
725 SubscriptionRef.changes(configState).pipe(
726 Stream.filter(Option.isSome),
727 Stream.map((projection) => projection.value.config),
728 Stream.filter(predicate),
729 Stream.runHead,
730 Effect.map(Option.getOrThrow),
731 );
732
733 const firstThemes = [
734 {
735 id: "first-theme",
736 name: "First theme",
737 appearance: "dark" as const,
738 canvas: "#111111",
739 accent: "#ffffff",
740 },
741 ];
742 const firstThemeState = yield* awaitConfig(
743 (config) => config.environmentThemes?.[0]?.id === "first-theme",
744 ).pipe(Effect.forkChild);
745 yield* publishConfigEvents(firstSocket, [
746 {
747 version: 1,
748 type: "environmentThemesUpdated",
749 payload: { themes: firstThemes },
750 },
751 ]);
752 expect((yield* Fiber.join(firstThemeState)).environmentThemes).toEqual(firstThemes);
753
754 const firstClosed = yield* firstSession.closed.pipe(Effect.exit, Effect.forkChild);
755 const firstRequest = yield* awaitRequest(firstSocket);
756 firstSocket.serverMessage(
757 failure === "defect"
758 ? encodeJson({
759 _tag: "Defect",
760 defect: encodeDefect(new Error("config stream died")),
761 })
762 : encodeJson({
763 _tag: "Exit",
764 requestId: firstRequest.id,
765 exit: {
766 _tag: "Failure",
767 cause: [
768 {
769 _tag: "Fail",
770 error: {
771 _tag: "EnvironmentAuthorizationError",
772 message: "config subscription rejected",
773 requiredScope: "orchestration:read",
774 },
775 },
776 ],
777 },
778 }),
779 );
780 const firstClosedExit = yield* Fiber.join(firstClosed);
781 expect(Exit.isFailure(firstClosedExit)).toBe(true);
782 if (failure === "typed" && Exit.isFailure(firstClosedExit)) {
783 expect(Cause.squash(firstClosedExit.cause)).toBeInstanceOf(ConnectionBlockedError);
784 expect(Cause.squash(firstClosedExit.cause)).toMatchObject({ reason: "permission" });
785 }
786 yield* SubscriptionRef.set(activeSession, Option.none());
787
788 const recoveredConfig = {
789 ...THEME_SERVER_CONFIG,
790 environment: {
791 ...THEME_SERVER_CONFIG.environment,
792 label: "Recovered environment",
793 },
794 } satisfies ServerConfigType;
795 const secondSession = yield* factory.connect(PREPARED);
796 const secondReady = yield* Effect.forkChild(secondSession.ready);
797 const secondSocket = yield* awaitSocket(sockets, 1);
798 secondSocket.open();
799 yield* completeInitialConfig(secondSocket, encodeServerConfig(recoveredConfig), {
800 environmentThemes: true,
801 });
802 yield* Fiber.join(secondReady);
803
804 const recoveredState = yield* awaitConfig(
805 (config) => config.environment.label === "Recovered environment",
806 ).pipe(Effect.forkChild);
807 yield* SubscriptionRef.set(activeSession, Option.some(secondSession));
808 expect((yield* Fiber.join(recoveredState)).environmentThemes).toEqual(firstThemes);
809
810 const recoveredThemes = [
811 {
812 id: "recovered-theme",
813 name: "Recovered theme",
814 appearance: "dark" as const,
815 canvas: "#000000",
816 accent: "#eeeeee",
817 },
818 ];
819 const liveRecoveredState = yield* awaitConfig(
820 (config) => config.environmentThemes?.[0]?.id === "recovered-theme",
821 ).pipe(Effect.forkChild);
822 yield* publishConfigEvents(secondSocket, [
823 {
824 version: 1,
825 type: "environmentThemesUpdated",
826 payload: { themes: recoveredThemes },
827 },
828 ]);
829 expect((yield* Fiber.join(liveRecoveredState)).environmentThemes).toEqual(
830 recoveredThemes,
831 );
832 }),
833 ),
834 );
835
836 it.effect.each<{
837 readonly event: ServerConfigStreamEventType;
838 readonly expectedConfig: Partial<ServerConfigType>;
839 }>([
840 {
841 event: {
842 version: 1,
843 type: "providerStatuses",
844 payload: {
845 providers: [
846 {
847 instanceId: ProviderInstanceId.make("codex"),
848 driver: ProviderDriverKind.make("codex"),
849 enabled: true,
850 installed: true,
851 version: "1.0.0",
852 status: "ready",
853 auth: { status: "authenticated" },
854 checkedAt: "2026-08-27T00:00:00.000Z",
855 models: [],
856 slashCommands: [],
857 skills: [],
858 },
859 ],
860 },
861 },
862 expectedConfig: {
863 providers: [
864 {
865 instanceId: ProviderInstanceId.make("codex"),
866 driver: ProviderDriverKind.make("codex"),
867 enabled: true,
868 installed: true,
869 version: "1.0.0",
870 status: "ready",
871 auth: { status: "authenticated" },
872 checkedAt: "2026-08-27T00:00:00.000Z",
873 models: [],
874 slashCommands: [],
875 skills: [],
876 },
877 ],
878 },
879 },
880 {
881 event: {
882 version: 1,
883 type: "settingsUpdated",
884 payload: {
885 settings: {
886 ...DEFAULT_SERVER_SETTINGS,
887 newWorktreesStartFromOrigin: !DEFAULT_SERVER_SETTINGS.newWorktreesStartFromOrigin,
888 },
889 },
890 },
891 expectedConfig: {
892 settings: {
893 ...DEFAULT_SERVER_SETTINGS,
894 newWorktreesStartFromOrigin: !DEFAULT_SERVER_SETTINGS.newWorktreesStartFromOrigin,
895 },
896 },
897 },
898 ])(
899 "preserves $event.type events and includes them in replay snapshots",
900 ({ event, expectedConfig }) =>
901 Effect.scoped(
902 Effect.gen(function* () {
903 const { factory, sockets } = yield* makeFactory();
904 const session = yield* factory.connect(PREPARED);
905 const readyFiber = yield* Effect.forkChild(session.ready);
906 const socket = yield* awaitSocket(sockets);
907 socket.open();
908 yield* completeInitialConfig(socket);
909 yield* Fiber.join(readyFiber);
910
911 const subscriber = yield* session
912 .subscribeServerConfig({})
913 .pipe(Stream.take(2), Stream.runCollect, Effect.forkChild);
914 yield* Effect.yieldNow;
915
916 const request = yield* awaitRequest(socket);
917 socket.serverMessage(
918 encodeJson({
919 _tag: "Chunk",
920 requestId: request.id,
921 values: [encodeServerConfigStreamEvent(event)],
922 }),
923 );
924
925 const events = Array.from(yield* Fiber.join(subscriber));
926 expect(events[1]).toEqual(event);
927
928 const replay = yield* session.subscribeServerConfig({}).pipe(Stream.runHead);
929 expect(replay).toMatchObject({
930 _tag: "Some",
931 value: {
932 type: "snapshot",
933 config: expectedConfig,
934 },
935 });
936 }),
937 ),
938 );
939
290940 it.effect("tolerates two missed pong windows before closing the session", () =>
291941 Effect.gen(function* () {
292942 const { factory, sockets } = yield* makeFactory();
......@@ -301,7 +951,7 @@ describe("RpcSessionFactory", () => {
301951
302952 yield* TestClock.adjust("15 seconds");
303953 expect(closedFiber.pollUnsafe()).toBeUndefined();
304 expect(socket.sent.slice(1).map((request) => decodeJson(request))).toEqual([
954 expect(socket.sent.map((message) => decodeJson(message)).filter(isPing)).toEqual([
305955 { _tag: "Ping" },
306956 { _tag: "Ping" },
307957 { _tag: "Ping" },
......@@ -379,10 +1029,12 @@ describe("RpcSessionFactory", () => {
3791029 );
3801030 yield* Fiber.join(probeFiber);
3811031
382 expect(socket.sent.map((request) => decodeRpcRequest(decodeJson(request)).tag)).toEqual([
383 WS_METHODS.serverGetConfig,
384 WS_METHODS.serverGetConfig,
385 ]);
1032 expect(
1033 socket.sent
1034 .map((message) => decodeJson(message))
1035 .filter(isRpcRequest)
1036 .map((request) => request.tag),
1037 ).toEqual([WS_METHODS.subscribeServerConfig, WS_METHODS.serverGetConfig]);
3861038 }),
3871039 ),
3881040 );
packages/client-runtime/src/rpc/session.ts+206-13
......@@ -1,11 +1,25 @@
1import { type ServerConfig, WS_METHODS } from "@t3tools/contracts";
1import {
2 type ServerConfig,
3 type ServerConfigStreamEvent,
4 WsSubscribeServerConfigRpc,
5 WS_METHODS,
6} from "@t3tools/contracts";
7import * as Cause from "effect/Cause";
28import * as Context from "effect/Context";
39import * as Deferred from "effect/Deferred";
410import * as Effect from "effect/Effect";
11import * as Equal from "effect/Equal";
12import * as Exit from "effect/Exit";
513import * as Layer from "effect/Layer";
14import * as Option from "effect/Option";
15import * as PubSub from "effect/PubSub";
16import * as Ref from "effect/Ref";
617import * as Schedule from "effect/Schedule";
718import type * as Scope from "effect/Scope";
19import * as Stream from "effect/Stream";
20import type * as Rpc from "effect/unstable/rpc/Rpc";
821import * as RpcClient from "effect/unstable/rpc/RpcClient";
22import * as RpcClientError from "effect/unstable/rpc/RpcClientError";
923import * as RpcSerialization from "effect/unstable/rpc/RpcSerialization";
1024import * as Socket from "effect/unstable/socket/Socket";
1125
......@@ -19,15 +33,27 @@ import {
1933 ConnectionBlockedError,
2034 ConnectionTransientError as ConnectionTransientErrorClass,
2135} from "../connection/model.ts";
36import {
37 applyServerConfigProjection,
38 type ServerConfigProjection,
39 withoutEnvironmentThemes,
40} from "../state/serverConfigProjection.ts";
2241
2342const SOCKET_OPEN_TIMEOUT = "15 seconds";
2443
2544export interface RpcSession {
2645 readonly client: WsRpcProtocolClient;
2746 readonly initialConfig: Effect.Effect<ServerConfig, ConnectionAttemptError>;
47 readonly subscribeServerConfig: (
48 input: ServerConfigSubscriptionInput,
49 ) => ServerConfigSubscription;
2850 readonly ready: Effect.Effect<void, ConnectionAttemptError>;
2951 readonly probe: Effect.Effect<void, ConnectionAttemptError>;
30 readonly closed: Effect.Effect<never, ConnectionTransientError>;
52 readonly closed: Effect.Effect<never, ConnectionAttemptError>;
53}
54
55export interface RpcSessionOptions {
56 readonly environmentThemes?: boolean;
3157}
3258
3359export class RpcSessionFactory extends Context.Service<
......@@ -43,8 +69,47 @@ type InitialConfigError = Effect.Error<
4369 ReturnType<WsRpcProtocolClient[typeof WS_METHODS.serverGetConfig]>
4470>;
4571type ProbeError = Effect.Error<ReturnType<WsRpcProtocolClient[typeof WS_METHODS.serverProbe]>>;
72type ServerConfigSubscriptionError =
73 | Rpc.ErrorExit<typeof WsSubscribeServerConfigRpc>
74 | RpcClientError.RpcClientError;
75type ServerConfigSubscription = Stream.Stream<
76 ServerConfigStreamEvent,
77 ServerConfigSubscriptionError
78>;
79type ServerConfigSubscriptionInput = Parameters<
80 WsRpcProtocolClient[typeof WS_METHODS.subscribeServerConfig]
81>[0];
82type EnvironmentThemesUpdatedEvent = Extract<
83 ServerConfigStreamEvent,
84 { readonly type: "environmentThemesUpdated" }
85>;
86
87interface ServerConfigReplayState {
88 readonly projection: ServerConfigProjection;
89 readonly revision: number;
90 readonly themesEvent: EnvironmentThemesUpdatedEvent | undefined;
91}
92
93interface BufferedServerConfigEvent {
94 readonly event: ServerConfigStreamEvent;
95 readonly replay: ServerConfigReplayState;
96 readonly revision: number;
97}
98
99function serverConfigReplayEvents(
100 state: ServerConfigReplayState,
101): ReadonlyArray<ServerConfigStreamEvent> {
102 const snapshot = {
103 version: 1 as const,
104 type: "snapshot" as const,
105 config: withoutEnvironmentThemes(state.projection.config),
106 };
107 return state.themesEvent === undefined ? [snapshot] : [snapshot, state.themesEvent];
108}
46109
47function mapSessionRpcError(error: InitialConfigError | ProbeError): ConnectionAttemptError {
110function mapSessionRpcError(
111 error: InitialConfigError | ProbeError | ServerConfigSubscriptionError,
112): ConnectionAttemptError {
48113 switch (error._tag) {
49114 case "EnvironmentAuthorizationError":
50115 return new ConnectionBlockedError({
......@@ -65,8 +130,12 @@ function mapSessionRpcError(error: InitialConfigError | ProbeError): ConnectionA
65130 }
66131}
67132
68export const make = Effect.gen(function* () {
133export const make = Effect.fn("RpcSessionFactory.make")(function* (
134 options: RpcSessionOptions = {},
135) {
69136 const webSocketConstructor = yield* Socket.WebSocketConstructor;
137 const serverConfigInput: ServerConfigSubscriptionInput =
138 options.environmentThemes === true ? { environmentThemes: true } : {};
70139
71140 const connect = Effect.fnUntraced(function* (connection: PreparedConnection) {
72141 yield* Effect.annotateCurrentSpan({
......@@ -113,18 +182,135 @@ export const make = Effect.gen(function* () {
113182 const protocolContext = yield* Layer.build(protocolLayer).pipe(
114183 Effect.withSpan("environment.websocket.connect"),
115184 );
116 const client = yield* makeWsRpcProtocolClient.pipe(Effect.provide(protocolContext));
117 const initialConfig = yield* Effect.cached(
118 client[WS_METHODS.serverGetConfig]({}).pipe(
185 const protocolClient = yield* makeWsRpcProtocolClient.pipe(Effect.provide(protocolContext));
186 const initialConfigDeferred = yield* Deferred.make<ServerConfig>();
187 const serverConfigExit = yield* Deferred.make<void, ServerConfigSubscriptionError>();
188 const configSubscriptionClosed = yield* Deferred.make<never, ConnectionAttemptError>();
189 const serverConfigState = yield* Ref.make(Option.none<ServerConfigReplayState>());
190 const serverConfigUpdates = yield* PubSub.sliding<BufferedServerConfigEvent>(64);
191 const configSubscriptionEndedError = new ConnectionTransientErrorClass({
192 reason: "remote-unavailable",
193 detail: `${connection.label} config subscription ended.`,
194 });
195 const serverConfigSource = protocolClient[WS_METHODS.subscribeServerConfig](
196 serverConfigInput,
197 ).pipe(
198 Stream.runForEach((event) =>
199 Effect.gen(function* () {
200 const buffered = yield* Ref.modify(serverConfigState, (current) => {
201 const projection = applyServerConfigProjection(
202 Option.map(current, (state) => state.projection),
203 event,
204 );
205 if (Option.isNone(projection)) {
206 return [Option.none<BufferedServerConfigEvent>(), current] as const;
207 }
208 const next = {
209 projection: projection.value,
210 revision: Option.match(current, {
211 onNone: () => 1,
212 onSome: (state) => state.revision + 1,
213 }),
214 themesEvent:
215 event.type === "environmentThemesUpdated"
216 ? event
217 : event.type === "snapshot" &&
218 event.config.environment.capabilities.environmentThemes !== true
219 ? undefined
220 : Option.getOrUndefined(current)?.themesEvent,
221 } satisfies ServerConfigReplayState;
222 return [
223 Option.some({ event, replay: next, revision: next.revision }),
224 Option.some(next),
225 ] as const;
226 });
227 if (Option.isSome(buffered)) {
228 yield* PubSub.publish(serverConfigUpdates, buffered.value);
229 }
230 if (event.type === "snapshot") {
231 yield* Deferred.succeed(initialConfigDeferred, event.config);
232 }
233 }),
234 ),
235 Effect.onExit((exit) => {
236 if (Exit.isSuccess(exit)) {
237 return Effect.all([
238 Deferred.succeed(serverConfigExit, undefined),
239 Deferred.fail(configSubscriptionClosed, configSubscriptionEndedError),
240 ]).pipe(Effect.asVoid);
241 }
242 if (Cause.hasInterruptsOnly(exit.cause)) {
243 return Effect.void;
244 }
245 return Effect.all([
246 Deferred.failCause(serverConfigExit, exit.cause),
247 Deferred.failCause(configSubscriptionClosed, Cause.map(exit.cause, mapSessionRpcError)),
248 ]).pipe(Effect.asVoid);
249 }),
250 );
251 yield* serverConfigSource.pipe(Effect.forkScoped);
252 const initialConfig = Effect.raceFirst(
253 Deferred.await(initialConfigDeferred),
254 Deferred.await(serverConfigExit).pipe(
119255 Effect.mapError(mapSessionRpcError),
120 Effect.withSpan("environment.initialSync"),
256 Effect.flatMap(() => Effect.fail(configSubscriptionEndedError)),
121257 ),
258 ).pipe(Effect.withSpan("environment.initialSync"));
259 const serverConfigEvents = Stream.unwrap(
260 Effect.gen(function* () {
261 const subscription = yield* PubSub.subscribe(serverConfigUpdates);
262 yield* Effect.raceFirst(
263 Deferred.await(initialConfigDeferred).pipe(Effect.asVoid),
264 Deferred.await(serverConfigExit),
265 );
266 const snapshot = yield* Ref.get(serverConfigState);
267 if (Option.isNone(snapshot)) {
268 return Stream.empty;
269 }
270 const updates = Stream.fromSubscription(subscription).pipe(
271 Stream.filter((buffered) => buffered.revision > snapshot.value.revision),
272 Stream.mapAccum(
273 () => snapshot.value.revision,
274 (revision, buffered) => [
275 buffered.revision,
276 buffered.revision === revision + 1
277 ? [buffered.event]
278 : serverConfigReplayEvents(buffered.replay),
279 ],
280 ),
281 );
282 const terminal = Stream.fromEffect(Deferred.await(serverConfigExit)).pipe(Stream.drain);
283 return Stream.concat(
284 Stream.fromIterable(serverConfigReplayEvents(snapshot.value)),
285 Stream.merge(updates, terminal, { haltStrategy: "either" }),
286 );
287 }),
288 ).pipe(
289 Stream.catchCause((cause) => {
290 if (Cause.hasInterruptsOnly(cause)) {
291 return Stream.failCause(cause);
292 }
293 // The supervisor keeps the original cause. Shared durable consumers
294 // need a transport-shaped failure so they wait for its replacement.
295 return Stream.fail(
296 new RpcClientError.RpcClientError({
297 reason: new RpcClientError.RpcClientDefect({
298 message: `${connection.label} config subscription failed.`,
299 cause,
300 }),
301 }),
302 );
303 }),
122304 );
305 const subscribeServerConfig = (input: ServerConfigSubscriptionInput) =>
306 Equal.equals(input, serverConfigInput)
307 ? serverConfigEvents
308 : protocolClient[WS_METHODS.subscribeServerConfig](input);
123309 const probe = initialConfig.pipe(
124310 Effect.flatMap((config) =>
125311 (config.environment.capabilities.connectionProbe === true
126 ? client[WS_METHODS.serverProbe]({})
127 : client[WS_METHODS.serverGetConfig]({})
312 ? protocolClient[WS_METHODS.serverProbe]({})
313 : protocolClient[WS_METHODS.serverGetConfig]({})
128314 ).pipe(Effect.mapError(mapSessionRpcError)),
129315 ),
130316 Effect.asVoid,
......@@ -132,19 +318,26 @@ export const make = Effect.gen(function* () {
132318 );
133319
134320 return {
135 client,
321 client: protocolClient,
136322 initialConfig,
323 subscribeServerConfig,
137324 ready: Deferred.await(connected).pipe(
138325 Effect.andThen(initialConfig),
139326 Effect.asVoid,
140327 Effect.raceFirst(Deferred.await(disconnected)),
141328 ),
142329 probe,
143 closed: Deferred.await(disconnected),
330 closed: Effect.raceFirst(
331 Deferred.await(disconnected),
332 Deferred.await(configSubscriptionClosed),
333 ),
144334 } satisfies RpcSession;
145335 });
146336
147337 return RpcSessionFactory.of({ connect });
148338});
149339
150export const layer = Layer.effect(RpcSessionFactory, make);
340export const layerWithOptions = (options: RpcSessionOptions) =>
341 Layer.effect(RpcSessionFactory, make(options));
342
343export const layer = layerWithOptions({});
packages/client-runtime/src/state/pullRequests.test.ts+1
......@@ -32,6 +32,7 @@ function session(client: WsRpcProtocolClient): RpcSession {
3232 return {
3333 client,
3434 initialConfig: Effect.never,
35 subscribeServerConfig: (input) => client.subscribeServerConfig(input),
3536 ready: Effect.void,
3637 probe: Effect.void,
3738 closed: Effect.never,
packages/client-runtime/src/state/server.test.ts+2-1
......@@ -30,7 +30,6 @@ import * as Persistence from "../platform/persistence.ts";
3030import type { WsRpcProtocolClient } from "../rpc/protocol.ts";
3131import type { RpcSession } from "../rpc/session.ts";
3232import {
33 applyServerConfigProjection,
3433 makeEnvironmentServerConfigState,
3534 isLegacyUpdateHandoffLoss,
3635 matchesServerUpdateReadyEvent,
......@@ -42,6 +41,7 @@ import {
4241 serverUpdateStateForServerVersion,
4342 validateServerUpdateReadyEvent,
4443} from "./server.ts";
44import { applyServerConfigProjection } from "./serverConfigProjection.ts";
4545
4646const CONFIG = {
4747 availableEditors: [],
......@@ -73,6 +73,7 @@ function session(client: WsRpcProtocolClient): RpcSession {
7373 return {
7474 client,
7575 initialConfig: Effect.succeed(CONFIG),
76 subscribeServerConfig: (input) => client.subscribeServerConfig(input),
7677 ready: Effect.void,
7778 probe: Effect.void,
7879 closed: Effect.never,
packages/client-runtime/src/state/server.ts+8-85
......@@ -41,6 +41,14 @@ import {
4141 type EnvironmentRpcInput,
4242} from "../rpc/client.ts";
4343import { followStreamInEnvironment } from "./runtime.ts";
44import {
45 applyServerConfigProjection,
46 type ServerConfigProjection,
47 withoutEnvironmentThemes,
48} from "./serverConfigProjection.ts";
49
50// Exported server state includes this type in its inferred public return type.
51export type { ServerConfigProjection } from "./serverConfigProjection.ts";
4452
4553export type ServerUpdateStage = "downloading" | "installing" | "resuming";
4654
......@@ -262,75 +270,6 @@ export function resolveServerUpdateProgressResult<E>(
262270 return Effect.fail(new ServerUpdateProgressIncompleteError({ targetVersion }));
263271}
264272
265export interface ServerConfigProjection {
266 readonly config: ServerConfig;
267 readonly latestEvent: ServerConfigStreamEvent;
268 readonly source: "cache" | "live";
269}
270
271export function applyServerConfigProjection(
272 current: Option.Option<ServerConfigProjection>,
273 event: ServerConfigStreamEvent,
274): Option.Option<ServerConfigProjection> {
275 switch (event.type) {
276 case "snapshot": {
277 // A snapshot never carries published themes -- the theme stream owns
278 // them -- so taking it wholesale would clear the set on every reconnect
279 // and repaint anyone wearing one until the follow-up event landed.
280 // Only from a server that still streams them. Reconnecting to one that
281 // predates the feature must drop the set rather than leave a palette on
282 // screen that nothing will ever update again.
283 const carried =
284 event.config.environment.capabilities.environmentThemes === true && Option.isSome(current)
285 ? current.value.config.environmentThemes
286 : undefined;
287 return Option.some({
288 config:
289 carried === undefined ? event.config : { ...event.config, environmentThemes: carried },
290 latestEvent: event,
291 source: "live" as const,
292 });
293 }
294 case "keybindingsUpdated":
295 return Option.map(current, (projection) => ({
296 config: {
297 ...projection.config,
298 keybindings: event.payload.keybindings,
299 issues: event.payload.issues,
300 },
301 latestEvent: event,
302 source: "live",
303 }));
304 case "providerStatuses":
305 return Option.map(current, (projection) => ({
306 config: {
307 ...projection.config,
308 providers: event.payload.providers,
309 },
310 latestEvent: event,
311 source: "live",
312 }));
313 case "settingsUpdated":
314 return Option.map(current, (projection) => ({
315 config: {
316 ...projection.config,
317 settings: event.payload.settings,
318 },
319 latestEvent: event,
320 source: "live",
321 }));
322 case "environmentThemesUpdated":
323 return Option.map(current, (projection) => ({
324 config: {
325 ...projection.config,
326 environmentThemes: event.payload.themes.length > 0 ? event.payload.themes : undefined,
327 },
328 latestEvent: event,
329 source: "live",
330 }));
331 }
332}
333
334273export function projectServerConfig(
335274 current: Option.Option<ServerConfigProjection>,
336275 event: ServerConfigStreamEvent,
......@@ -345,22 +284,6 @@ const cachedConfigSnapshotEvent = (config: ServerConfig): ServerConfigStreamEven
345284 config,
346285});
347286
348/**
349 * Keeps a complete server configuration available during reconnects. Server
350 * config carries the provider/model catalogue used by task creation, so it is
351 * useful—and safe—to retain after a transport session ends.
352 */
353/**
354 * Published themes live only as long as the machine publishes them, so they
355 * must not survive in the config cache: a restart or an offline load would
356 * otherwise hand clients palettes the environment has already dropped.
357 */
358function withoutEnvironmentThemes(config: ServerConfig): ServerConfig {
359 if (config.environmentThemes === undefined) return config;
360 const { environmentThemes: _ephemeral, ...rest } = config;
361 return rest;
362}
363
364287export const makeEnvironmentServerConfigState = Effect.fn("EnvironmentServerConfigState.make")(
365288 function* (environmentThemes?: boolean) {
366289 const supervisor = yield* EnvironmentSupervisor;
packages/client-runtime/src/state/serverConfigProjection.ts created+79
......@@ -0,0 +1,79 @@
1import type { ServerConfig, ServerConfigStreamEvent } from "@t3tools/contracts";
2import * as Option from "effect/Option";
3
4export interface ServerConfigProjection {
5 readonly config: ServerConfig;
6 readonly latestEvent: ServerConfigStreamEvent;
7 readonly source: "cache" | "live";
8}
9
10/**
11 * Cached config keeps the provider and model catalog available across reconnects.
12 * Published themes are current machine state, so a cache could restore themes
13 * that the machine no longer publishes. Replay sends themes as a separate event.
14 */
15export function withoutEnvironmentThemes(config: ServerConfig): ServerConfig {
16 if (config.environmentThemes === undefined) return config;
17 const { environmentThemes: _ephemeral, ...rest } = config;
18 return rest;
19}
20
21export function applyServerConfigProjection(
22 current: Option.Option<ServerConfigProjection>,
23 event: ServerConfigStreamEvent,
24): Option.Option<ServerConfigProjection> {
25 switch (event.type) {
26 case "snapshot": {
27 // Wire snapshots never contain published themes. Keep the previous set
28 // until a capable server sends its authoritative theme event. A legacy
29 // server cannot send a later removal, so a downgrade must clear the set.
30 const carried =
31 event.config.environment.capabilities.environmentThemes === true && Option.isSome(current)
32 ? current.value.config.environmentThemes
33 : undefined;
34 return Option.some({
35 config:
36 carried === undefined ? event.config : { ...event.config, environmentThemes: carried },
37 latestEvent: event,
38 source: "live" as const,
39 });
40 }
41 case "keybindingsUpdated":
42 return Option.map(current, (projection) => ({
43 config: {
44 ...projection.config,
45 keybindings: event.payload.keybindings,
46 issues: event.payload.issues,
47 },
48 latestEvent: event,
49 source: "live",
50 }));
51 case "providerStatuses":
52 return Option.map(current, (projection) => ({
53 config: {
54 ...projection.config,
55 providers: event.payload.providers,
56 },
57 latestEvent: event,
58 source: "live",
59 }));
60 case "settingsUpdated":
61 return Option.map(current, (projection) => ({
62 config: {
63 ...projection.config,
64 settings: event.payload.settings,
65 },
66 latestEvent: event,
67 source: "live",
68 }));
69 case "environmentThemesUpdated":
70 return Option.map(current, (projection) => ({
71 config: {
72 ...projection.config,
73 environmentThemes: event.payload.themes.length > 0 ? event.payload.themes : undefined,
74 },
75 latestEvent: event,
76 source: "live",
77 }));
78 }
79}
packages/client-runtime/src/state/shell-sync.test.ts+1
......@@ -51,6 +51,7 @@ function session(client: WsRpcProtocolClient): RpcSession.RpcSession {
5151 return {
5252 client,
5353 initialConfig: Effect.succeed({ shellResumeCompletionMarker: true } as never),
54 subscribeServerConfig: (input) => client.subscribeServerConfig(input),
5455 ready: Effect.void,
5556 probe: Effect.void,
5657 closed: Effect.never,
packages/client-runtime/src/state/sourceControl.test.ts+1
......@@ -50,6 +50,7 @@ function session(client: WsRpcProtocolClient): RpcSession {
5050 return {
5151 client,
5252 initialConfig: Effect.never,
53 subscribeServerConfig: (input) => client.subscribeServerConfig(input),
5354 ready: Effect.void,
5455 probe: Effect.void,
5556 closed: Effect.never,
packages/client-runtime/src/state/threads-pagination.test.ts+1
......@@ -156,6 +156,7 @@ const makeHarness = Effect.fn("TestThreadPagination.makeHarness")(function* (opt
156156 initialConfig: Effect.succeed({
157157 threadSnapshotPagination: options?.paginationCapability !== false,
158158 } as never),
159 subscribeServerConfig: (input) => client.subscribeServerConfig(input),
159160 ready: Effect.void,
160161 probe: Effect.void,
161162 closed: Effect.never,
packages/client-runtime/src/state/threads-sync.test.ts+1
......@@ -112,6 +112,7 @@ function testSession(
112112 ? ({ threadResumeCompletionMarker: true } as never)
113113 : ({} as never),
114114 ),
115 subscribeServerConfig: (input) => client.subscribeServerConfig(input),
115116 ready: Effect.void,
116117 probe: Effect.void,
117118 closed: Effect.never,
packages/client-runtime/src/state/vcs.test.ts+1
......@@ -86,6 +86,7 @@ function session(client: WsRpcProtocolClient): RpcSession {
8686 return {
8787 client,
8888 initialConfig: Effect.never,
89 subscribeServerConfig: (input) => client.subscribeServerConfig(input),
8990 ready: Effect.void,
9091 probe: Effect.void,
9192 closed: Effect.never,
packages/client-runtime/src/state/vcsAction.test.ts+1
......@@ -84,6 +84,7 @@ function session(client: WsRpcProtocolClient): RpcSession {
8484 return {
8585 client,
8686 initialConfig: Effect.never,
87 subscribeServerConfig: (input) => client.subscribeServerConfig(input),
8788 ready: Effect.void,
8889 probe: Effect.void,
8990 closed: Effect.never,