From a357bf832b381aad01b245ab3b9ade0444cdda45 Mon Sep 17 00:00:00 2001 From: Mahir Abbas Date: Thu, 15 Oct 2026 15:51:58 +0100 Subject: [PATCH 1/5] fix(web): agent tabs in background threads open in the desktop browser The desktop only loaded preview tabs for threads whose chat was on screen. A tab an agent opened in a background thread never got a webview, so after 10 s the server fell back to its headless browser, whose software-rendered GPU process can use several cores. The always-mounted browser host now follows the environment's preview events and starts the existing per-thread sync for each thread that opens a tab, so the desktop renders it. Refs #18165, #16901 Co-Authored-By: Claude Opus 5.5 --- apps/web/src/browser/ElectronBrowserHost.tsx | 2 + .../preview/usePreviewSession.test.tsx | 99 +++++++++++++++++++ .../components/preview/usePreviewSession.ts | 45 ++++++++- 3 files changed, 144 insertions(+), 2 deletions(-) create mode 100644 apps/web/src/components/preview/usePreviewSession.test.tsx diff --git a/apps/web/src/browser/ElectronBrowserHost.tsx b/apps/web/src/browser/ElectronBrowserHost.tsx index b076a30a5221f..f5ddc17a70cb4 100644 --- a/apps/web/src/browser/ElectronBrowserHost.tsx +++ b/apps/web/src/browser/ElectronBrowserHost.tsx @@ -10,6 +10,7 @@ import { primaryEnvironmentIdAtom } from "~/state/primaryEnvironment"; import { isElectron } from "~/env"; import { useClientSettingsHydrated } from "~/hooks/useSettings"; import { useTheme } from "~/hooks/useTheme"; +import { useDesktopHostedPreviewSessions } from "~/components/preview/usePreviewSession"; import { useActivePreviewSessions } from "~/previewStateStore"; import { previewEnvironment } from "~/state/preview"; import { useEnvironmentScope } from "~/state/session"; @@ -27,6 +28,7 @@ export function ElectronBrowserHost() { const { resolvedTheme } = useTheme(); const previewByThreadKey = useActivePreviewSessions(); const primaryEnvironmentId = useAtomValue(primaryEnvironmentIdAtom); + useDesktopHostedPreviewSessions(primaryEnvironmentId); const sessions = useMemo( () => Object.entries(previewByThreadKey).flatMap(([threadKey, previewState]) => { diff --git a/apps/web/src/components/preview/usePreviewSession.test.tsx b/apps/web/src/components/preview/usePreviewSession.test.tsx new file mode 100644 index 0000000000000..6ea7dc808a726 --- /dev/null +++ b/apps/web/src/components/preview/usePreviewSession.test.tsx @@ -0,0 +1,99 @@ +import { scopeThreadRef } from "@t3tools/client-runtime/environment"; +import { + EnvironmentId, + type PreviewEvent, + type PreviewSessionSnapshot, + ThreadId, +} from "@t3tools/contracts"; +import { AsyncResult, Atom } from "effect/reactivity"; +import { act, createElement } from "react"; +import { create, type ReactTestRenderer } from "react-test-renderer"; +import { afterEach, beforeEach, describe, expect, it, vi } from "vite-plus/test"; + +// Stand-ins for the environment's preview queries, keyed like the real atom families. +const queryAtoms = vi.hoisted( + () => new Map>>(), +); + +vi.mock("~/env", () => ({ isElectron: true })); + +vi.mock("~/state/preview", async () => { + const { AsyncResult, Atom } = await import("effect/reactivity"); + const query = (key: string) => { + let atom = queryAtoms.get(key); + if (!atom) { + atom = Atom.make>(AsyncResult.initial()).pipe( + Atom.keepAlive, + ); + queryAtoms.set(key, atom); + } + return atom; + }; + return { + previewEnvironment: { + events: ({ environmentId }: { environmentId: string }) => query(`events:${environmentId}`), + list: ({ environmentId, input }: { environmentId: string; input: { threadId: string } }) => + query(`list:${environmentId}:${input.threadId}`), + }, + }; +}); + +import { readThreadPreviewState, resetPreviewStateForTests } from "~/previewStateStore"; +import { AppAtomRegistryProvider, appAtomRegistry } from "~/rpc/atomRegistry"; +import { useDesktopHostedPreviewSessions } from "./usePreviewSession"; + +const environmentId = EnvironmentId.make("env-1"); +const threadRef = scopeThreadRef(environmentId, ThreadId.make("thread-background")); + +const snapshot: PreviewSessionSnapshot = { + threadId: "thread-background", + tabId: "tab_agent", + navStatus: { _tag: "Loading", url: "http://localhost:5173/", title: "" }, + canGoBack: false, + canGoForward: false, + updatedAt: "2026-01-01T00:00:00.000Z", +}; + +function Host() { + useDesktopHostedPreviewSessions(environmentId); + return null; +} + +let renderer: ReactTestRenderer | undefined; + +beforeEach(() => { + vi.stubGlobal("IS_REACT_ACT_ENVIRONMENT", true); + resetPreviewStateForTests(); +}); + +afterEach(async () => { + await act(() => renderer?.unmount()); + renderer = undefined; + vi.unstubAllGlobals(); +}); + +describe("useDesktopHostedPreviewSessions", () => { + it("tracks tabs an agent opens in a thread whose chat is not mounted", async () => { + await act(async () => { + renderer = create(createElement(AppAtomRegistryProvider, null, createElement(Host))); + }); + expect(readThreadPreviewState(threadRef).sessions).toEqual({}); + + const opened: PreviewEvent = { + type: "opened", + threadId: "thread-background", + tabId: "tab_agent", + createdAt: "2026-01-01T00:00:00.000Z", + serverEpoch: "server-a", + revision: 1, + snapshot, + }; + await act(async () => { + appAtomRegistry.set(queryAtoms.get(`events:${environmentId}`)!, AsyncResult.success(opened)); + }); + + // The thread's own sync is now running, so its list and events reach the store. + expect(queryAtoms.has(`list:${environmentId}:thread-background`)).toBe(true); + expect(Object.keys(readThreadPreviewState(threadRef).sessions)).toEqual(["tab_agent"]); + }); +}); diff --git a/apps/web/src/components/preview/usePreviewSession.ts b/apps/web/src/components/preview/usePreviewSession.ts index b529d044f3931..8024bd3cd5640 100644 --- a/apps/web/src/components/preview/usePreviewSession.ts +++ b/apps/web/src/components/preview/usePreviewSession.ts @@ -1,8 +1,12 @@ "use client"; import { useAtomValue } from "@effect/atom-react"; -import { parseScopedThreadKey, scopedThreadKey } from "@t3tools/client-runtime/environment"; -import type { ScopedThreadRef } from "@t3tools/contracts"; +import { + parseScopedThreadKey, + scopeThreadRef, + scopedThreadKey, +} from "@t3tools/client-runtime/environment"; +import { type EnvironmentId, type ScopedThreadRef, ThreadId } from "@t3tools/contracts"; import * as Schema from "effect/Schema"; import { AsyncResult, Atom } from "effect/reactivity"; @@ -11,6 +15,7 @@ import { readThreadPreviewState, reconcilePreviewServerSessions, } from "~/previewStateStore"; +import { isElectron } from "~/env"; import { previewEnvironment } from "~/state/preview"; class PreviewSessionThreadKeyParseError extends Schema.TaggedError()( @@ -90,3 +95,39 @@ const previewSessionSyncAtom = Atom.family((threadKey: string) => { export function usePreviewSession(threadRef: ScopedThreadRef): void { useAtomValue(previewSessionSyncAtom(scopedThreadKey(threadRef))); } + +const inactiveAtom = Atom.make(() => undefined); + +/** + * The desktop hosts its own server's tabs even while their chat is not shown. + * Otherwise an agent tab in a background thread never gets a webview and the + * server falls back to its headless browser. Keep each thread's sync running + * once the environment reports a tab for it. + */ +const desktopHostedPreviewSyncAtom = Atom.family((environmentId: EnvironmentId) => { + const eventsAtom = previewEnvironment.events({ environmentId, input: {} }); + return Atom.make((get) => { + const syncedThreadKeys = new Set(); + get.subscribe( + eventsAtom, + (result) => { + if (!AsyncResult.isSuccess(result) || result.value.type === "closed") return; + const threadKey = scopedThreadKey( + scopeThreadRef(environmentId, ThreadId.make(result.value.threadId)), + ); + if (syncedThreadKeys.has(threadKey)) return; + syncedThreadKeys.add(threadKey); + get.mount(previewSessionSyncAtom(threadKey)); + }, + { immediate: true }, + ); + }).pipe(Atom.withLabel(`preview:desktop-hosted-sync:${environmentId}`)); +}); + +export function useDesktopHostedPreviewSessions(environmentId: EnvironmentId | null): void { + useAtomValue( + isElectron && environmentId !== null + ? desktopHostedPreviewSyncAtom(environmentId) + : inactiveAtom, + ); +} From 199bf70beae935e07c571c6fe7a10eed785d495a Mon Sep 17 00:00:00 2001 From: Julius Marminge Date: Mon, 19 Oct 2026 03:44:45 +0000 Subject: [PATCH 2/5] test(web): cover background preview session discovery --- .../previewSessionSync.browser.test.ts | 54 +++++++++- .../preview/usePreviewSession.test.tsx | 99 ------------------- 2 files changed, 51 insertions(+), 102 deletions(-) delete mode 100644 apps/web/src/components/preview/usePreviewSession.test.tsx diff --git a/apps/web/src/components/preview/previewSessionSync.browser.test.ts b/apps/web/src/components/preview/previewSessionSync.browser.test.ts index 65c222e4126ad..ba6a60cdbb44d 100644 --- a/apps/web/src/components/preview/previewSessionSync.browser.test.ts +++ b/apps/web/src/components/preview/previewSessionSync.browser.test.ts @@ -26,6 +26,8 @@ const harness = vi.hoisted(() => ({ events: new Map>(), })); +vi.mock("~/env", () => ({ isElectron: true })); + vi.mock("~/state/preview", async () => { const { AsyncResult, Atom } = await import("effect/reactivity"); return { @@ -74,7 +76,7 @@ import { } from "~/previewStateStore"; import { AppAtomRegistryProvider, appAtomRegistry } from "~/rpc/atomRegistry"; import { previewEnvironment } from "~/state/preview"; -import { usePreviewSession } from "./usePreviewSession"; +import { useDesktopHostedPreviewSessions, usePreviewSession } from "./usePreviewSession"; let root: Root | null = null; let container: HTMLDivElement | null = null; @@ -107,11 +109,16 @@ const result = ( })), }); -async function mount() { +function DesktopSessionSync() { + useDesktopHostedPreviewSessions(ref.environmentId); + return null; +} + +async function mount(Component = SessionSync) { container = document.body.appendChild(document.createElement("div")); root = createRoot(container); await act(async () => { - root!.render(createElement(AppAtomRegistryProvider, null, createElement(SessionSync))); + root!.render(createElement(AppAtomRegistryProvider, null, createElement(Component))); }); } @@ -134,6 +141,47 @@ afterEach(async () => { }); describe("native preview session synchronization", () => { + it("discovers and closes a server tab in a background thread without mounting its chat", async () => { + await mount(DesktopSessionSync); + expect(readThreadPreviewState(ref).sessions).toEqual({}); + previewEnvironment.events({ environmentId: ref.environmentId, input: {} }); + const events = harness.events.get(ref.environmentId)!; + const baseline = result(1, ["agent-tab"]); + const snapshot = baseline.sessions[0]!; + await act(async () => { + appAtomRegistry.set( + events, + AsyncResult.success({ + type: "opened", + threadId: ref.threadId, + tabId: snapshot.tabId, + snapshot, + serverEpoch: baseline.serverEpoch, + revision: baseline.revision, + createdAt: snapshot.updatedAt, + }), + ); + }); + expect(readThreadPreviewState(ref).sessions).toEqual({ "agent-tab": snapshot }); + appAtomRegistry.set(listAtom(), AsyncResult.success(baseline)); + expect(readThreadPreviewState(ref).listLoaded).toBe(true); + expect(readThreadPreviewState(ref).snapshot).toEqual(snapshot); + + appAtomRegistry.set( + events, + AsyncResult.success({ + type: "closed", + threadId: ref.threadId, + tabId: snapshot.tabId, + serverEpoch: baseline.serverEpoch, + revision: 2, + createdAt: snapshot.updatedAt, + }), + ); + expect(readThreadPreviewState(ref).sessions).toEqual({}); + expect(readThreadPreviewState(ref).activeTabId).toBeNull(); + }); + it("evaluates the lazy list once and restores sessions that predate mounting", async () => { const atom = listAtom(); await mount(); diff --git a/apps/web/src/components/preview/usePreviewSession.test.tsx b/apps/web/src/components/preview/usePreviewSession.test.tsx deleted file mode 100644 index 6ea7dc808a726..0000000000000 --- a/apps/web/src/components/preview/usePreviewSession.test.tsx +++ /dev/null @@ -1,99 +0,0 @@ -import { scopeThreadRef } from "@t3tools/client-runtime/environment"; -import { - EnvironmentId, - type PreviewEvent, - type PreviewSessionSnapshot, - ThreadId, -} from "@t3tools/contracts"; -import { AsyncResult, Atom } from "effect/reactivity"; -import { act, createElement } from "react"; -import { create, type ReactTestRenderer } from "react-test-renderer"; -import { afterEach, beforeEach, describe, expect, it, vi } from "vite-plus/test"; - -// Stand-ins for the environment's preview queries, keyed like the real atom families. -const queryAtoms = vi.hoisted( - () => new Map>>(), -); - -vi.mock("~/env", () => ({ isElectron: true })); - -vi.mock("~/state/preview", async () => { - const { AsyncResult, Atom } = await import("effect/reactivity"); - const query = (key: string) => { - let atom = queryAtoms.get(key); - if (!atom) { - atom = Atom.make>(AsyncResult.initial()).pipe( - Atom.keepAlive, - ); - queryAtoms.set(key, atom); - } - return atom; - }; - return { - previewEnvironment: { - events: ({ environmentId }: { environmentId: string }) => query(`events:${environmentId}`), - list: ({ environmentId, input }: { environmentId: string; input: { threadId: string } }) => - query(`list:${environmentId}:${input.threadId}`), - }, - }; -}); - -import { readThreadPreviewState, resetPreviewStateForTests } from "~/previewStateStore"; -import { AppAtomRegistryProvider, appAtomRegistry } from "~/rpc/atomRegistry"; -import { useDesktopHostedPreviewSessions } from "./usePreviewSession"; - -const environmentId = EnvironmentId.make("env-1"); -const threadRef = scopeThreadRef(environmentId, ThreadId.make("thread-background")); - -const snapshot: PreviewSessionSnapshot = { - threadId: "thread-background", - tabId: "tab_agent", - navStatus: { _tag: "Loading", url: "http://localhost:5173/", title: "" }, - canGoBack: false, - canGoForward: false, - updatedAt: "2026-01-01T00:00:00.000Z", -}; - -function Host() { - useDesktopHostedPreviewSessions(environmentId); - return null; -} - -let renderer: ReactTestRenderer | undefined; - -beforeEach(() => { - vi.stubGlobal("IS_REACT_ACT_ENVIRONMENT", true); - resetPreviewStateForTests(); -}); - -afterEach(async () => { - await act(() => renderer?.unmount()); - renderer = undefined; - vi.unstubAllGlobals(); -}); - -describe("useDesktopHostedPreviewSessions", () => { - it("tracks tabs an agent opens in a thread whose chat is not mounted", async () => { - await act(async () => { - renderer = create(createElement(AppAtomRegistryProvider, null, createElement(Host))); - }); - expect(readThreadPreviewState(threadRef).sessions).toEqual({}); - - const opened: PreviewEvent = { - type: "opened", - threadId: "thread-background", - tabId: "tab_agent", - createdAt: "2026-01-01T00:00:00.000Z", - serverEpoch: "server-a", - revision: 1, - snapshot, - }; - await act(async () => { - appAtomRegistry.set(queryAtoms.get(`events:${environmentId}`)!, AsyncResult.success(opened)); - }); - - // The thread's own sync is now running, so its list and events reach the store. - expect(queryAtoms.has(`list:${environmentId}:thread-background`)).toBe(true); - expect(Object.keys(readThreadPreviewState(threadRef).sessions)).toEqual(["tab_agent"]); - }); -}); From d109f54c228293adaec5a5a4a75f536341cd9a9a Mon Sep 17 00:00:00 2001 From: Julius Marminge Date: Mon, 19 Oct 2026 03:47:33 +0000 Subject: [PATCH 3/5] refactor(web): simplify background preview sync retention --- .../components/preview/usePreviewSession.ts | 22 +++++++++---------- 1 file changed, 10 insertions(+), 12 deletions(-) diff --git a/apps/web/src/components/preview/usePreviewSession.ts b/apps/web/src/components/preview/usePreviewSession.ts index ef595a823041d..a7e21faca8acc 100644 --- a/apps/web/src/components/preview/usePreviewSession.ts +++ b/apps/web/src/components/preview/usePreviewSession.ts @@ -102,26 +102,24 @@ export function usePreviewSession(threadRef: ScopedThreadRef): void { const inactiveAtom = Atom.make(() => undefined); -/** - * The desktop hosts its own server's tabs even while their chat is not shown. - * Otherwise an agent tab in a background thread never gets a webview and the - * server falls back to its headless browser. Keep each thread's sync running - * once the environment reports a tab for it. - */ +// Discover background tabs without listing every thread at startup. Retain each +// sync for the host's lifetime so later events still arrive while chat is absent. const desktopHostedPreviewSyncAtom = Atom.family((environmentId: EnvironmentId) => { const eventsAtom = previewEnvironment.events({ environmentId, input: {} }); return Atom.make((get) => { - const syncedThreadKeys = new Set(); + const syncedThreadIds = new Set(); get.subscribe( eventsAtom, (result) => { if (!AsyncResult.isSuccess(result) || result.value.type === "closed") return; - const threadKey = scopedThreadKey( - scopeThreadRef(environmentId, ThreadId.make(result.value.threadId)), + const threadId = result.value.threadId; + if (syncedThreadIds.has(threadId)) return; + syncedThreadIds.add(threadId); + get.mount( + previewSessionSyncAtom( + scopedThreadKey(scopeThreadRef(environmentId, ThreadId.make(threadId))), + ), ); - if (syncedThreadKeys.has(threadKey)) return; - syncedThreadKeys.add(threadKey); - get.mount(previewSessionSyncAtom(threadKey)); }, { immediate: true }, ); From c71f7548ba32daec939e7d6fa0a8b077ccf84bf0 Mon Sep 17 00:00:00 2001 From: Julius Marminge Date: Mon, 19 Oct 2026 03:57:31 +0000 Subject: [PATCH 4/5] test(client-runtime): cover batched preview event delivery --- .../client-runtime/src/state/runtime.test.ts | 86 ++++++++++++++++++- 1 file changed, 84 insertions(+), 2 deletions(-) diff --git a/packages/client-runtime/src/state/runtime.test.ts b/packages/client-runtime/src/state/runtime.test.ts index c36bcaf3bd373..f294da5cb02df 100644 --- a/packages/client-runtime/src/state/runtime.test.ts +++ b/packages/client-runtime/src/state/runtime.test.ts @@ -1,5 +1,5 @@ import { describe, expect, it, vi } from "@effect/vitest"; -import { EnvironmentId } from "@t3tools/contracts"; +import { EnvironmentId, type PreviewEvent, WS_METHODS } from "@t3tools/contracts"; import * as Cause from "effect/Cause"; import * as Deferred from "effect/Deferred"; import * as Effect from "effect/Effect"; @@ -26,6 +26,7 @@ import * as EnvironmentRegistry from "../connection/registry.ts"; import * as EnvironmentSupervisor from "../connection/supervisor.ts"; import { EnvironmentRpcUnavailableError } from "../rpc/client.ts"; import type * as RpcSession from "../rpc/session.ts"; +import { createPreviewEnvironmentAtoms } from "./preview.ts"; import type { QueryPersistence } from "./queryPersistence.ts"; import { environmentRpcKey, @@ -89,6 +90,7 @@ const makeEnvironmentQueryHarness = Effect.fn("TestEnvironmentQuery.makeHarness" idleTtlMs, persistence, initialState, + rpcSession, isFinal, }: { readonly refreshTrigger?: Atom.Atom; @@ -96,10 +98,13 @@ const makeEnvironmentQueryHarness = Effect.fn("TestEnvironmentQuery.makeHarness" readonly idleTtlMs?: number; readonly persistence?: QueryPersistence; readonly initialState?: SupervisorConnectionState; + readonly rpcSession?: RpcSession.RpcSession; } = {}, ) { const supervisorState = yield* SubscriptionRef.make(initialState ?? queryConnectionState()); - const supervisorSession = yield* SubscriptionRef.make(Option.some(QUERY_RPC_SESSION)); + const supervisorSession = yield* SubscriptionRef.make( + Option.some(rpcSession ?? QUERY_RPC_SESSION), + ); const supervisor = EnvironmentSupervisor.EnvironmentSupervisor.of({ target: QUERY_ENVIRONMENT, state: supervisorState, @@ -134,6 +139,7 @@ const makeEnvironmentQueryHarness = Effect.fn("TestEnvironmentQuery.makeHarness" }); return { + runtime, atom: family({ environmentId: QUERY_ENVIRONMENT.environmentId, input: undefined }), supervisorSession, supervisorState, @@ -305,6 +311,82 @@ describe("environmentRpcKey", () => { }); }); +describe("preview event subscriptions", () => { + it.effect("delivers every open and close when RPC events arrive in batches", () => + Effect.scoped( + Effect.gen(function* () { + const opened: PreviewEvent[] = ["thread-a", "thread-b"].map((threadId, index) => ({ + type: "opened", + threadId, + tabId: `tab-${index}`, + serverEpoch: "epoch-1", + revision: index + 1, + createdAt: "2026-10-02T00:00:00.000Z", + snapshot: { + threadId, + tabId: `tab-${index}`, + navStatus: { _tag: "Idle" }, + canGoBack: false, + canGoForward: false, + updatedAt: "2026-10-02T00:00:00.000Z", + }, + })); + const closed: PreviewEvent[] = opened.map((event, index) => ({ + type: "closed", + threadId: event.threadId, + tabId: event.tabId, + serverEpoch: event.serverEpoch, + revision: opened.length + index + 1, + createdAt: event.createdAt, + })); + const releaseEvents = yield* Deferred.make(); + // Each iterable is one RPC chunk; separate emissions would hide lost events. + const stream = Stream.fromEffect(Deferred.await(releaseEvents)).pipe( + Stream.flatMap(() => + Stream.fromIterable(opened).pipe( + Stream.concat(Stream.fromIterable(closed)), + Stream.concat(Stream.never), + ), + ), + ); + const harness = yield* makeEnvironmentQueryHarness(Effect.void, { + rpcSession: { + client: { [WS_METHODS.subscribePreviewEvents]: () => stream }, + } as unknown as RpcSession.RpcSession, + }); + const atom = createPreviewEnvironmentAtoms(harness.runtime).events({ + environmentId: QUERY_ENVIRONMENT.environmentId, + input: {}, + }); + const registry = AtomRegistry.make(); + const observed: PreviewEvent[] = []; + const sawLast = yield* Deferred.make(); + const unsubscribe = registry.subscribe( + atom, + (result) => { + if (!AsyncResult.isSuccess(result)) return; + observed.push(result.value); + if (result.value === closed[closed.length - 1]) { + Deferred.doneUnsafe(sawLast, Exit.void); + } + }, + { immediate: true }, + ); + yield* Effect.addFinalizer(() => + Effect.sync(() => { + unsubscribe(); + registry.dispose(); + }), + ); + + yield* Deferred.succeed(releaseEvents, undefined); + yield* Deferred.await(sawLast); + expect(observed).toEqual([...opened, ...closed]); + }), + ), + ); +}); + describe("finite environment subscription lifecycle", () => { it.effect( "exposes early values and settles on the terminal value without waiting for a reconnect", From 68c1401c7658857110ef8660cc904778c37d2895 Mon Sep 17 00:00:00 2001 From: Julius Marminge Date: Mon, 19 Oct 2026 04:00:18 +0000 Subject: [PATCH 5/5] fix(client-runtime): preserve batched preview events --- packages/client-runtime/src/state/preview.ts | 3 +++ 1 file changed, 3 insertions(+) diff --git a/packages/client-runtime/src/state/preview.ts b/packages/client-runtime/src/state/preview.ts index 72d47b7426994..99f3b2f10e1d4 100644 --- a/packages/client-runtime/src/state/preview.ts +++ b/packages/client-runtime/src/state/preview.ts @@ -1,4 +1,5 @@ import { WS_METHODS } from "@t3tools/contracts"; +import * as Stream from "effect/Stream"; import { Atom } from "effect/reactivity"; import type { EnvironmentRegistry } from "../connection/registry.ts"; @@ -28,6 +29,8 @@ export function createPreviewEnvironmentAtoms( events: createEnvironmentRpcSubscriptionAtomFamily(runtime, { label: "environment-data:preview:events", tag: WS_METHODS.subscribePreviewEvents, + // Stream atoms publish only the last value per chunk; every preview event must arrive. + transform: (stream) => stream.pipe(Stream.rechunk(1)), }), discoveredServers: createEnvironmentRpcSubscriptionAtomFamily(runtime, { label: "environment-data:preview:discovered-servers",