diff --git a/apps/web/src/browser/ElectronBrowserHost.tsx b/apps/web/src/browser/ElectronBrowserHost.tsx index 02e45bb950bef..78b4d579c71f0 100644 --- a/apps/web/src/browser/ElectronBrowserHost.tsx +++ b/apps/web/src/browser/ElectronBrowserHost.tsx @@ -13,6 +13,7 @@ import { primaryEnvironmentIdAtom } from "~/state/primaryEnvironment"; import { isElectron } from "~/env"; import { useClientSettingsHydrated } from "~/hooks/useSettings"; +import { useDesktopHostedPreviewSessions } from "~/components/preview/usePreviewSession"; import { useActivePreviewSessions } from "~/previewStateStore"; import { previewEnvironment } from "~/state/preview"; import { useEnvironmentScope } from "~/state/session"; @@ -29,6 +30,7 @@ import { previewRuntimeTabId } from "./previewRuntimeTabId"; export function ElectronBrowserHost() { const previewByThreadKey = useActivePreviewSessions(); const primaryEnvironmentId = useAtomValue(primaryEnvironmentIdAtom); + useDesktopHostedPreviewSessions(primaryEnvironmentId); const activeSessions = Object.entries(previewByThreadKey).flatMap(([threadKey, previewState]) => { const threadRef = parseScopedThreadKey(threadKey); return threadRef 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.ts b/apps/web/src/components/preview/usePreviewSession.ts index 6166da8e4289f..a7e21faca8acc 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()( @@ -94,3 +99,37 @@ const previewSessionSyncAtom = Atom.family((threadKey: string) => { export function usePreviewSession(threadRef: ScopedThreadRef): void { useAtomValue(previewSessionSyncAtom(scopedThreadKey(threadRef))); } + +const inactiveAtom = Atom.make(() => undefined); + +// 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 syncedThreadIds = new Set(); + get.subscribe( + eventsAtom, + (result) => { + if (!AsyncResult.isSuccess(result) || result.value.type === "closed") return; + const threadId = result.value.threadId; + if (syncedThreadIds.has(threadId)) return; + syncedThreadIds.add(threadId); + get.mount( + previewSessionSyncAtom( + scopedThreadKey(scopeThreadRef(environmentId, ThreadId.make(threadId))), + ), + ); + }, + { immediate: true }, + ); + }).pipe(Atom.withLabel(`preview:desktop-hosted-sync:${environmentId}`)); +}); + +export function useDesktopHostedPreviewSessions(environmentId: EnvironmentId | null): void { + useAtomValue( + isElectron && environmentId !== null + ? desktopHostedPreviewSyncAtom(environmentId) + : inactiveAtom, + ); +} 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", diff --git a/packages/client-runtime/src/state/runtime.test.ts b/packages/client-runtime/src/state/runtime.test.ts index c389f454186e8..bc0cdcc40df8e 100644 --- a/packages/client-runtime/src/state/runtime.test.ts +++ b/packages/client-runtime/src/state/runtime.test.ts @@ -1,6 +1,7 @@ import { describe, expect, it, vi } from "@effect/vitest"; import { EnvironmentId, + type PreviewEvent, type TerminalAttachInput, type TerminalAttachStreamEvent, type TerminalMetadataStreamEvent, @@ -32,6 +33,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 { createTerminalEnvironmentAtoms } from "./terminal.ts"; import { @@ -96,6 +98,7 @@ const makeEnvironmentQueryHarness = Effect.fn("TestEnvironmentQuery.makeHarness" idleTtlMs, persistence, initialState, + rpcSession, isFinal, }: { readonly refreshTrigger?: Atom.Atom; @@ -103,10 +106,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 streamGeneration = yield* SubscriptionRef.make(0); const supervisor = EnvironmentSupervisor.EnvironmentSupervisor.of({ target: QUERY_ENVIRONMENT, @@ -320,6 +326,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",