Skip to content
2 changes: 2 additions & 0 deletions apps/web/src/browser/ElectronBrowserHost.tsx
Original file line number Diff line number Diff line change
Expand Up @@ -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";
Expand All @@ -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
Expand Down
54 changes: 51 additions & 3 deletions apps/web/src/components/preview/previewSessionSync.browser.test.ts
Original file line number Diff line number Diff line change
Expand Up @@ -26,6 +26,8 @@ const harness = vi.hoisted(() => ({
events: new Map<string, Atom.Writable<EventResult>>(),
}));

vi.mock("~/env", () => ({ isElectron: true }));

vi.mock("~/state/preview", async () => {
const { AsyncResult, Atom } = await import("effect/reactivity");
return {
Expand Down Expand Up @@ -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;
Expand Down Expand Up @@ -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)));
});
}

Expand All @@ -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();
Expand Down
43 changes: 41 additions & 2 deletions apps/web/src/components/preview/usePreviewSession.ts
Original file line number Diff line number Diff line change
@@ -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";

Expand All @@ -11,6 +15,7 @@ import {
readThreadPreviewState,
reconcilePreviewServerSessions,
} from "~/previewStateStore";
import { isElectron } from "~/env";
import { previewEnvironment } from "~/state/preview";

class PreviewSessionThreadKeyParseError extends Schema.TaggedError<PreviewSessionThreadKeyParseError>()(
Expand Down Expand Up @@ -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<string>();
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,
);
}
3 changes: 3 additions & 0 deletions packages/client-runtime/src/state/preview.ts
Original file line number Diff line number Diff line change
@@ -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";
Expand Down Expand Up @@ -28,6 +29,8 @@ export function createPreviewEnvironmentAtoms<R, E>(
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",
Expand Down
84 changes: 83 additions & 1 deletion packages/client-runtime/src/state/runtime.test.ts
Original file line number Diff line number Diff line change
@@ -1,6 +1,7 @@
import { describe, expect, it, vi } from "@effect/vitest";
import {
EnvironmentId,
type PreviewEvent,
type TerminalAttachInput,
type TerminalAttachStreamEvent,
type TerminalMetadataStreamEvent,
Expand Down Expand Up @@ -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 {
Expand Down Expand Up @@ -96,17 +98,21 @@ const makeEnvironmentQueryHarness = Effect.fn("TestEnvironmentQuery.makeHarness"
idleTtlMs,
persistence,
initialState,
rpcSession,
isFinal,
}: {
readonly refreshTrigger?: Atom.Atom<unknown>;
readonly isFinal?: (value: A) => boolean;
readonly idleTtlMs?: number;
readonly persistence?: QueryPersistence<A>;
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,
Expand Down Expand Up @@ -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<void>();
// 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<void>();
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",
Expand Down
Loading