diff --git a/apps/mobile/src/Stack.tsx b/apps/mobile/src/Stack.tsx index db244d7c726..4eb4a5061e8 100644 --- a/apps/mobile/src/Stack.tsx +++ b/apps/mobile/src/Stack.tsx @@ -15,7 +15,6 @@ import { DynamicColorIOS, Platform, Pressable, ScrollView, StyleSheet } from "re import { useResolveClassNames } from "uniwind"; import { AppText as Text } from "./components/AppText"; -import { renderCompactBrandTitle } from "./components/CompactBrandTitle"; import { ArchivedThreadsRouteScreen } from "./features/archive/ArchivedThreadsRouteScreen"; import { useAgentNotificationNavigation } from "./features/agent-awareness/notificationNavigation"; import { ClerkSettingsSheetDetentProvider } from "./features/cloud/ClerkSettingsSheetDetent"; @@ -378,8 +377,7 @@ export const RootStack = createNativeStackNavigator({ ...GLASS_HEADER_OPTIONS, contentStyle: { backgroundColor: "transparent" }, headerBackVisible: false, - headerTitle: renderCompactBrandTitle, - title: "T3 Code", + title: "Threads", }, }), Thread: createNativeStackScreen({ diff --git a/apps/mobile/src/components/CompactBrandTitle.tsx b/apps/mobile/src/components/CompactBrandTitle.tsx deleted file mode 100644 index 2ecf34aa40c..00000000000 --- a/apps/mobile/src/components/CompactBrandTitle.tsx +++ /dev/null @@ -1,60 +0,0 @@ -import { View } from "react-native"; - -import { AppText as Text } from "./AppText"; -import { T3Wordmark } from "./T3Wordmark"; -import { useThemeColor } from "../lib/useThemeColor"; - -/** - * Compact brand lockup sized for native navigation bars. - */ -export function CompactBrandTitle() { - const iconColor = useThemeColor("--color-icon"); - const mutedColor = useThemeColor("--color-foreground-muted"); - const subtleColor = useThemeColor("--color-subtle"); - - return ( - - - - Code - - - - Alpha - - - - ); -} - -export function renderCompactBrandTitle() { - return ; -} diff --git a/apps/mobile/src/features/home/HomeRouteScreen.tsx b/apps/mobile/src/features/home/HomeRouteScreen.tsx index cd9467c774e..49cf06d85ec 100644 --- a/apps/mobile/src/features/home/HomeRouteScreen.tsx +++ b/apps/mobile/src/features/home/HomeRouteScreen.tsx @@ -4,7 +4,6 @@ import { useNavigation } from "@react-navigation/native"; import { useMemo, useState } from "react"; import { NativeHeaderToolbar, NativeStackScreenOptions } from "../../native/StackHeader"; -import { renderCompactBrandTitle } from "../../components/CompactBrandTitle"; import { useProjects, useThreadShells } from "../../state/entities"; import { usePendingNewTasks } from "../../state/use-pending-new-tasks"; import { useWorkspaceState } from "../../state/workspace"; @@ -87,9 +86,7 @@ export function HomeRouteScreen() { > <> {/* Restore the compact title in case the split branch blanked it. */} - + - {emptyState.loading ? ( + {emptyState.loading && !shouldShowConnectionStatus ? ( ) : null} + {shouldShowConnectionStatus && Platform.OS === "ios" ? ( + + + + ) : null} {connectionStatus} diff --git a/apps/mobile/src/features/home/WorkspaceConnectionStatus.test.ts b/apps/mobile/src/features/home/WorkspaceConnectionStatus.test.ts index 767f0e3dd3f..8c3c873cc9e 100644 --- a/apps/mobile/src/features/home/WorkspaceConnectionStatus.test.ts +++ b/apps/mobile/src/features/home/WorkspaceConnectionStatus.test.ts @@ -67,4 +67,21 @@ describe("workspace connection status", () => { expect(shouldShowWorkspaceConnectionStatus(state)).toBe(true); expect(workspaceConnectionStatusLabel(state)).toBe("Could not reach Julius’s Mac mini"); }); + + it("shows shell catch-up while cached threads remain visible", () => { + const state = workspaceState({ hasPendingShellSnapshot: true }); + + expect(shouldShowWorkspaceConnectionStatus(state)).toBe(true); + expect(workspaceConnectionStatusLabel(state)).toBe("Syncing threads..."); + }); + + it("distinguishes initial shell loading from cached catch-up", () => { + const state = workspaceState({ + hasLoadedShellSnapshot: false, + hasPendingShellSnapshot: true, + }); + + expect(shouldShowWorkspaceConnectionStatus(state)).toBe(true); + expect(workspaceConnectionStatusLabel(state)).toBe("Loading threads..."); + }); }); diff --git a/apps/mobile/src/features/home/WorkspaceConnectionStatus.tsx b/apps/mobile/src/features/home/WorkspaceConnectionStatus.tsx index 8acdacfbbb3..1e986ad1a50 100644 --- a/apps/mobile/src/features/home/WorkspaceConnectionStatus.tsx +++ b/apps/mobile/src/features/home/WorkspaceConnectionStatus.tsx @@ -12,7 +12,10 @@ export function WorkspaceConnectionStatus(props: { readonly variant?: "floating" | "sidebar"; }) { const iconColor = useThemeColor("--color-icon-muted"); - const isReconnecting = props.state.connectingEnvironments.length > 0; + const isSynchronizing = + props.state.networkStatus !== "offline" && + props.state.connectionError === null && + (props.state.connectingEnvironments.length > 0 || props.state.hasPendingShellSnapshot); const variant = props.variant ?? "floating"; return ( @@ -37,7 +40,7 @@ export function WorkspaceConnectionStatus(props: { : undefined } > - {isReconnecting ? ( + {isSynchronizing ? ( ) : ( diff --git a/apps/mobile/src/features/home/workspace-connection-status.ts b/apps/mobile/src/features/home/workspace-connection-status.ts index 7d2c6d840d4..d8eed4383b1 100644 --- a/apps/mobile/src/features/home/workspace-connection-status.ts +++ b/apps/mobile/src/features/home/workspace-connection-status.ts @@ -5,6 +5,7 @@ export function shouldShowWorkspaceConnectionStatus(state: WorkspaceState): bool state.networkStatus === "offline" || state.connectionError !== null || state.hasConnectingEnvironment || + state.hasPendingShellSnapshot || (state.hasLoadedShellSnapshot && !state.hasReadyEnvironment) ); } @@ -18,5 +19,8 @@ export function workspaceConnectionStatusLabel(state: WorkspaceState): string { return `Reconnecting ${state.connectingEnvironments.length} environments`; } if (state.connectionError !== null) return state.connectionError; + if (state.hasPendingShellSnapshot) { + return state.hasLoadedShellSnapshot ? "Syncing threads..." : "Loading threads..."; + } return "Not connected"; } diff --git a/apps/mobile/src/features/threads/sidebar-navigation-shell.tsx b/apps/mobile/src/features/threads/sidebar-navigation-shell.tsx index a94c033fa4d..f1e06926d0b 100644 --- a/apps/mobile/src/features/threads/sidebar-navigation-shell.tsx +++ b/apps/mobile/src/features/threads/sidebar-navigation-shell.tsx @@ -11,7 +11,6 @@ import { import type { ReactNode } from "react"; import { Platform, useColorScheme } from "react-native"; -import { renderCompactBrandTitle } from "../../components/CompactBrandTitle"; import { nativeHeaderScrollEdgeEffects } from "../../native/StackHeader"; const SCROLL_EDGE_EFFECTS = nativeHeaderScrollEdgeEffects(Platform.OS, Platform.Version); @@ -35,11 +34,10 @@ const SIDEBAR_SCREEN_OPTIONS: SidebarScreenOptions = { headerShadowVisible: false, headerShown: true, headerStyle: { backgroundColor: "transparent" }, - headerTitle: renderCompactBrandTitle, headerTitleStyle: { fontSize: 18, fontWeight: "800" }, headerTransparent: true, scrollEdgeEffects: SCROLL_EDGE_EFFECTS, - title: "T3 Code", + title: "Threads", unstable_navigationItemStyle: "editor", }; diff --git a/apps/server/src/server.test.ts b/apps/server/src/server.test.ts index b1f6c1f7251..a0f41cbf0fc 100644 --- a/apps/server/src/server.test.ts +++ b/apps/server/src/server.test.ts @@ -3717,6 +3717,8 @@ it.layer(NodeServices.layer)("server router seam", (it) => { assert.equal(response.environment.environmentId, testEnvironmentDescriptor.environmentId); assert.equal(response.auth.policy, "desktop-managed-local"); + assert.equal(response.shellResumeCompletionMarker, true); + assert.equal(response.threadResumeCompletionMarker, true); }).pipe(Effect.provide(NodeHttpServer.layerTest)), ); @@ -5607,6 +5609,113 @@ it.layer(NodeServices.layer)("server router seam", (it) => { }).pipe(Effect.provide(NodeHttpServer.layerTest)), ); + it.effect("marks an empty shell catch-up replay as synchronized when requested", () => + Effect.gen(function* () { + yield* buildAppUnderTest({ + layers: { + orchestrationEngine: { + readEvents: () => Stream.empty, + }, + }, + }); + + const wsUrl = yield* getWsServerUrl("/ws"); + const firstItem = yield* Effect.scoped( + withWsRpcClient(wsUrl, (client) => + client[ORCHESTRATION_WS_METHODS.subscribeShell]({ + afterSequence: 0, + requestCompletionMarker: true, + }).pipe(Stream.runHead), + ), + ); + + assert.deepEqual(Option.getOrThrow(firstItem), { kind: "synchronized" }); + }).pipe(Effect.provide(NodeHttpServer.layerTest)), + ); + + it.effect("marks a socket thread snapshot as synchronized when requested", () => + Effect.gen(function* () { + const thread = makeDefaultOrchestrationReadModel().threads[0]!; + yield* buildAppUnderTest({ + layers: { + projectionSnapshotQuery: { + getThreadDetailSnapshot: () => + Effect.succeed(Option.some({ snapshotSequence: 1, thread })), + }, + }, + }); + + const wsUrl = yield* getWsServerUrl("/ws"); + const items = yield* Effect.scoped( + withWsRpcClient(wsUrl, (client) => + client[ORCHESTRATION_WS_METHODS.subscribeThread]({ + threadId: defaultThreadId, + requestCompletionMarker: true, + }).pipe(Stream.take(2), Stream.runCollect), + ), + ); + + assert.equal(items[0]?.kind, "snapshot"); + assert.deepEqual(items[1], { kind: "synchronized" }); + }).pipe(Effect.provide(NodeHttpServer.layerTest)), + ); + + it.effect("buffers shell events published while the fallback snapshot loads", () => + Effect.gen(function* () { + const liveEvents = yield* PubSub.unbounded(); + const deletedEvent = { + sequence: 2, + eventId: EventId.make("event-shell-thread-deleted"), + aggregateKind: "thread", + aggregateId: defaultThreadId, + occurredAt: "2026-01-01T00:00:01.000Z", + commandId: null, + causationEventId: null, + correlationId: null, + metadata: {}, + type: "thread.deleted", + payload: { + threadId: defaultThreadId, + deletedAt: "2026-01-01T00:00:01.000Z", + }, + } satisfies Extract; + + yield* buildAppUnderTest({ + layers: { + orchestrationEngine: { + streamDomainEvents: Stream.fromPubSub(liveEvents), + }, + projectionSnapshotQuery: { + getShellSnapshot: () => + Effect.gen(function* () { + yield* Effect.sleep("25 millis"); + yield* PubSub.publish(liveEvents, deletedEvent); + return { + snapshotSequence: 1, + projects: [], + threads: [makeDefaultOrchestrationThreadShell()], + updatedAt: "2026-01-01T00:00:00.000Z", + }; + }), + }, + }, + }); + + const wsUrl = yield* getWsServerUrl("/ws"); + const items = yield* Effect.scoped( + withWsRpcClient(wsUrl, (client) => + client[ORCHESTRATION_WS_METHODS.subscribeShell]({ + requestCompletionMarker: true, + }).pipe(Stream.take(3), Stream.runCollect), + ), + ).pipe(Effect.timeout("2 seconds")); + + assert.equal(items[0]?.kind, "snapshot"); + assert.equal(items[1]?.kind, "thread-removed"); + assert.deepEqual(items[2], { kind: "synchronized" }); + }).pipe(Effect.provide(NodeHttpServer.layerTest), TestClock.withLive), + ); + it.effect("buffers thread events published while the initial snapshot loads", () => Effect.gen(function* () { const thread = makeDefaultOrchestrationReadModel().threads[0]!; diff --git a/apps/server/src/ws.ts b/apps/server/src/ws.ts index 32dcb8a13d6..4b04cf07b5b 100644 --- a/apps/server/src/ws.ts +++ b/apps/server/src/ws.ts @@ -935,6 +935,8 @@ const makeWsRpcLayer = ( otlpMetricsEnabled: config.otlpMetricsUrl !== undefined, }, settings, + shellResumeCompletionMarker: true, + threadResumeCompletionMarker: true, }; }); @@ -1104,11 +1106,28 @@ const makeWsRpcLayer = ( }), ), ); - return Stream.concat(catchUpStream, Stream.fromQueue(liveBuffer)); + const afterCatchUp = + input.requestCompletionMarker === true + ? Stream.concat( + Stream.fromEffect( + Queue.offer(liveBuffer, { kind: "synchronized" as const }), + ).pipe(Stream.drain), + Stream.fromQueue(liveBuffer), + ) + : Stream.fromQueue(liveBuffer); + return Stream.concat(catchUpStream, afterCatchUp); }), ); } + // The full-snapshot fallback needs the same replay-window safety + // as the resume path: subscribe before loading the projection so + // events published while the snapshot is read are buffered. + const liveBuffer = yield* Queue.unbounded(); + yield* Effect.forkScoped( + liveStream.pipe(Stream.runForEach((item) => Queue.offer(liveBuffer, item))), + ); + const bufferedLiveStream = Stream.fromQueue(liveBuffer); const snapshot = yield* projectionSnapshotQuery.getShellSnapshot().pipe( Effect.tapError((cause) => Effect.logError("orchestration shell snapshot load failed", { cause }), @@ -1122,12 +1141,21 @@ const makeWsRpcLayer = ( ), ); + const afterSnapshot = + input.requestCompletionMarker === true + ? Stream.concat( + Stream.fromEffect( + Queue.offer(liveBuffer, { kind: "synchronized" as const }), + ).pipe(Stream.drain), + bufferedLiveStream, + ) + : bufferedLiveStream; return Stream.concat( Stream.make({ kind: "snapshot" as const, snapshot, }), - liveStream, + afterSnapshot, ); }), { "rpc.aggregate": "orchestration" }, @@ -1206,7 +1234,16 @@ const makeWsRpcLayer = ( }), ), ); - return Stream.concat(catchUpStream, bufferedLiveStream); + const afterCatchUp = + input.requestCompletionMarker === true + ? Stream.concat( + Stream.fromEffect( + Queue.offer(liveBuffer, { kind: "synchronized" as const }), + ).pipe(Stream.drain), + bufferedLiveStream, + ) + : bufferedLiveStream; + return Stream.concat(catchUpStream, afterCatchUp); } const snapshot = yield* projectionSnapshotQuery @@ -1228,12 +1265,21 @@ const makeWsRpcLayer = ( }); } + const afterSnapshot = + input.requestCompletionMarker === true + ? Stream.concat( + Stream.fromEffect( + Queue.offer(liveBuffer, { kind: "synchronized" as const }), + ).pipe(Stream.drain), + bufferedLiveStream, + ) + : bufferedLiveStream; return Stream.concat( Stream.make({ kind: "snapshot" as const, snapshot: snapshot.value, }), - bufferedLiveStream, + afterSnapshot, ); }), { "rpc.aggregate": "orchestration" }, diff --git a/packages/client-runtime/src/rpc/client.ts b/packages/client-runtime/src/rpc/client.ts index 92892431e45..c7c928b3c95 100644 --- a/packages/client-runtime/src/rpc/client.ts +++ b/packages/client-runtime/src/rpc/client.ts @@ -11,6 +11,7 @@ import { RpcClientError } from "effect/unstable/rpc"; import { EnvironmentSupervisor } from "../connection/supervisor.ts"; import type { WsRpcProtocolClient } from "../rpc/protocol.ts"; +import type { RpcSession } from "../rpc/session.ts"; export class EnvironmentRpcUnavailableError extends Schema.TaggedErrorClass()( "EnvironmentRpcUnavailableError", @@ -147,15 +148,18 @@ export function runStream( ); } -export function subscribe( +interface SubscriptionOptions { + readonly onExpectedFailure?: ( + cause: Cause.Cause>, + ) => Effect.Effect; + readonly retryExpectedFailureAfter?: Duration.Input; + readonly resubscribe?: Stream.Stream; +} + +export function subscribeDynamic( tag: TTag, - input: EnvironmentRpcInput, - options?: { - readonly onExpectedFailure?: ( - cause: Cause.Cause>, - ) => Effect.Effect; - readonly retryExpectedFailureAfter?: Duration.Input; - }, + makeInput: (session: RpcSession) => Effect.Effect>, + options?: SubscriptionOptions, ): Stream.Stream< EnvironmentRpcStreamValue, EnvironmentRpcStreamFailure, @@ -163,8 +167,18 @@ export function subscribe( > { return Stream.unwrap( EnvironmentSupervisor.pipe( - Effect.map((supervisor) => - SubscriptionRef.changes(supervisor.session).pipe( + Effect.map((supervisor) => { + const sessionChanges = SubscriptionRef.changes(supervisor.session); + const sessions = + options?.resubscribe === undefined + ? sessionChanges + : Stream.merge( + sessionChanges, + options.resubscribe.pipe( + Stream.mapEffect(() => SubscriptionRef.get(supervisor.session)), + ), + ); + return sessions.pipe( Stream.switchMap( Option.match({ onNone: () => Stream.empty, @@ -180,54 +194,64 @@ export function subscribe( EnvironmentRpcStreamFailure > => Stream.suspend(() => - method(input).pipe( - Stream.catchCause((cause) => { - const hasOnlyExpectedFailures = - cause.reasons.length > 0 && - cause.reasons.every((reason) => reason._tag === "Fail"); - const isTransportFailure = - hasOnlyExpectedFailures && - cause.reasons.every( - (reason) => reason._tag === "Fail" && isRpcClientError(reason.error), - ); - if (isTransportFailure) { - return Stream.fromEffect( - Effect.logWarning( - "Durable RPC subscription lost its transport; waiting for the next session.", - { - cause: Cause.pretty(cause), - method: tag, - environmentId: supervisor.target.environmentId, - }, - ), - ).pipe(Stream.drain); - } - if (hasOnlyExpectedFailures && options?.onExpectedFailure !== undefined) { - const handled = Stream.fromEffect(options.onExpectedFailure(cause)).pipe( - Stream.drain, - ); - if (options.retryExpectedFailureAfter === undefined) { - return handled; - } - return handled.pipe( - Stream.concat( - Stream.fromEffect( - Effect.sleep(options.retryExpectedFailureAfter), - ).pipe(Stream.drain), - ), - Stream.concat(subscribeToSession()), - ); - } - return Stream.failCause(cause); - }), + Stream.unwrap( + makeInput(session).pipe( + Effect.map((input) => + method(input).pipe( + Stream.catchCause((cause) => { + const hasOnlyExpectedFailures = + cause.reasons.length > 0 && + cause.reasons.every((reason) => reason._tag === "Fail"); + const isTransportFailure = + hasOnlyExpectedFailures && + cause.reasons.every( + (reason) => + reason._tag === "Fail" && isRpcClientError(reason.error), + ); + if (isTransportFailure) { + return Stream.fromEffect( + Effect.logWarning( + "Durable RPC subscription lost its transport; waiting for the next session.", + { + cause: Cause.pretty(cause), + method: tag, + environmentId: supervisor.target.environmentId, + }, + ), + ).pipe(Stream.drain); + } + if ( + hasOnlyExpectedFailures && + options?.onExpectedFailure !== undefined + ) { + const handled = Stream.fromEffect( + options.onExpectedFailure(cause), + ).pipe(Stream.drain); + if (options.retryExpectedFailureAfter === undefined) { + return handled; + } + return handled.pipe( + Stream.concat( + Stream.fromEffect( + Effect.sleep(options.retryExpectedFailureAfter), + ).pipe(Stream.drain), + ), + Stream.concat(subscribeToSession()), + ); + } + return Stream.failCause(cause); + }), + ), + ), + ), ), ); return subscribeToSession(); }, }), ), - ), - ), + ); + }), ), ).pipe( Stream.withSpan("EnvironmentRpc.subscribe", { @@ -236,6 +260,18 @@ export function subscribe( ); } +export function subscribe( + tag: TTag, + input: EnvironmentRpcInput, + options?: SubscriptionOptions, +): Stream.Stream< + EnvironmentRpcStreamValue, + EnvironmentRpcStreamFailure, + EnvironmentSupervisor +> { + return subscribeDynamic(tag, () => Effect.succeed(input), options); +} + export const config = Effect.gen(function* () { const session = yield* currentSession(); return yield* session.initialConfig; diff --git a/packages/client-runtime/src/state/shell-sync.test.ts b/packages/client-runtime/src/state/shell-sync.test.ts index b6eb4e9f710..a4514d5f0e6 100644 --- a/packages/client-runtime/src/state/shell-sync.test.ts +++ b/packages/client-runtime/src/state/shell-sync.test.ts @@ -8,6 +8,7 @@ import { describe, expect, it } from "@effect/vitest"; import * as Effect from "effect/Effect"; import * as Option from "effect/Option"; import * as Queue from "effect/Queue"; +import * as Ref from "effect/Ref"; import * as Stream from "effect/Stream"; import * as SubscriptionRef from "effect/SubscriptionRef"; @@ -17,6 +18,7 @@ import { type PreparedConnection, } from "../connection/model.ts"; import * as EnvironmentSupervisor from "../connection/supervisor.ts"; +import * as ConnectionWakeups from "../connection/wakeups.ts"; import * as Persistence from "../platform/persistence.ts"; import * as RpcSession from "../rpc/session.ts"; import type { WsRpcProtocolClient } from "../rpc/protocol.ts"; @@ -48,7 +50,7 @@ const LIVE_SHELL_SNAPSHOT: OrchestrationShellSnapshot = { function session(client: WsRpcProtocolClient): RpcSession.RpcSession { return { client, - initialConfig: Effect.never, + initialConfig: Effect.succeed({ shellResumeCompletionMarker: true } as never), ready: Effect.void, probe: Effect.void, closed: Effect.never, @@ -112,6 +114,15 @@ describe("environment shell synchronization", () => { kind: "snapshot", snapshot: LIVE_SHELL_SNAPSHOT, }); + const synchronizing = yield* SubscriptionRef.changes(shellState).pipe( + Stream.filter((state) => state.status === "synchronizing" && Option.isSome(state.snapshot)), + Stream.runHead, + ); + expect(Option.getOrThrow(Option.getOrThrow(synchronizing).snapshot)).toEqual( + LIVE_SHELL_SNAPSHOT, + ); + + yield* Queue.offer(events, { kind: "synchronized" }); yield* SubscriptionRef.changes(shellState).pipe( Stream.filter((state) => state.status === "live"), Stream.runHead, @@ -137,21 +148,32 @@ describe("environment shell synchronization", () => { }), ); - it.effect("resumes a warm shell cache via afterSequence without an HTTP fetch", () => + it.effect("replaces a warm shell cache with an authoritative HTTP snapshot", () => Effect.gen(function* () { const cachedSnapshot: OrchestrationShellSnapshot = { snapshotSequence: 5, projects: [], - threads: [], + threads: [{ id: "stale-thread" } as never], updatedAt: "2026-06-06T00:00:00.000Z", }; + const httpSnapshot: OrchestrationShellSnapshot = { + ...cachedSnapshot, + snapshotSequence: 9, + threads: [], + updatedAt: "2026-06-07T00:00:00.000Z", + }; const events = yield* Queue.unbounded(); const capturedAfterSequence = yield* SubscriptionRef.make(undefined); + const capturedCompletionMarker = yield* Ref.make(undefined); const loaderCalls = yield* SubscriptionRef.make(0); const client = { - [ORCHESTRATION_WS_METHODS.subscribeShell]: (input: { readonly afterSequence?: number }) => + [ORCHESTRATION_WS_METHODS.subscribeShell]: (input: { + readonly afterSequence?: number; + readonly requestCompletionMarker?: boolean; + }) => Stream.unwrap( - SubscriptionRef.set(capturedAfterSequence, input.afterSequence).pipe( + Ref.set(capturedCompletionMarker, input.requestCompletionMarker).pipe( + Effect.andThen(SubscriptionRef.set(capturedAfterSequence, input.afterSequence)), Effect.as(Stream.fromQueue(events)), ), ), @@ -183,9 +205,11 @@ describe("environment shell synchronization", () => { }); const snapshotLoader = ShellSnapshotLoader.of({ load: () => - SubscriptionRef.update(loaderCalls, (count) => count + 1).pipe(Effect.as(Option.none())), + SubscriptionRef.update(loaderCalls, (count) => count + 1).pipe( + Effect.as(Option.some(httpSnapshot)), + ), }); - yield* makeEnvironmentShellState().pipe( + const shellState = yield* makeEnvironmentShellState().pipe( Effect.provideService(EnvironmentSupervisor.EnvironmentSupervisor, supervisor), Effect.provideService(Persistence.EnvironmentCacheStore, cache), Effect.provideService(ShellSnapshotLoader, snapshotLoader), @@ -197,8 +221,108 @@ describe("environment shell synchronization", () => { Stream.runHead, ); - expect(yield* SubscriptionRef.get(capturedAfterSequence)).toBe(5); - expect(yield* SubscriptionRef.get(loaderCalls)).toBe(0); + expect(yield* SubscriptionRef.get(capturedAfterSequence)).toBe(9); + expect(yield* Ref.get(capturedCompletionMarker)).toBe(true); + expect(yield* SubscriptionRef.get(loaderCalls)).toBe(1); + const synchronizing = yield* SubscriptionRef.get(shellState); + expect(synchronizing.status).toBe("synchronizing"); + expect(Option.getOrThrow(synchronizing.snapshot)).toEqual(httpSnapshot); + + yield* Queue.offer(events, { kind: "synchronized" }); + yield* SubscriptionRef.changes(shellState).pipe( + Stream.filter((value) => value.status === "live"), + Stream.runHead, + ); + }), + ); + + it.effect("refreshes the authoritative shell snapshot when the app becomes active", () => + Effect.gen(function* () { + const events = yield* Queue.unbounded(); + const wakeups = yield* Queue.unbounded(); + const loaderCalls = yield* Ref.make(0); + const subscriptionCount = yield* Ref.make(0); + const client = { + [ORCHESTRATION_WS_METHODS.subscribeShell]: () => + Stream.unwrap( + Ref.update(subscriptionCount, (count) => count + 1).pipe( + Effect.as(Stream.fromQueue(events)), + ), + ), + } as unknown as WsRpcProtocolClient; + const supervisorState = yield* SubscriptionRef.make(AVAILABLE_CONNECTION_STATE); + const supervisor = EnvironmentSupervisor.EnvironmentSupervisor.of({ + target: TARGET, + state: supervisorState, + session: yield* SubscriptionRef.make(Option.some(session(client))), + prepared: yield* SubscriptionRef.make(Option.some(PREPARED)), + connect: Effect.void, + disconnect: Effect.void, + retryNow: Effect.void, + } satisfies EnvironmentSupervisor.EnvironmentSupervisor["Service"]); + const cache = Persistence.EnvironmentCacheStore.of({ + loadShell: () => Effect.succeed(Option.some(LIVE_SHELL_SNAPSHOT)), + saveShell: () => Effect.void, + loadThread: () => Effect.succeed(Option.none()), + saveThread: () => Effect.void, + removeThread: () => Effect.void, + loadServerConfig: () => Effect.succeed(Option.none()), + saveServerConfig: () => Effect.void, + loadVcsRefs: () => Effect.succeed(Option.none()), + saveVcsRefs: () => Effect.void, + clear: () => Effect.void, + }); + const snapshotLoader = ShellSnapshotLoader.of({ + load: () => + Ref.updateAndGet(loaderCalls, (count) => count + 1).pipe( + Effect.map((count) => + Option.some({ ...LIVE_SHELL_SNAPSHOT, snapshotSequence: count * 10 }), + ), + ), + }); + const shellState = yield* makeEnvironmentShellState().pipe( + Effect.provideService(EnvironmentSupervisor.EnvironmentSupervisor, supervisor), + Effect.provideService(Persistence.EnvironmentCacheStore, cache), + Effect.provideService(ShellSnapshotLoader, snapshotLoader), + Effect.provideService( + ConnectionWakeups.ConnectionWakeups, + ConnectionWakeups.ConnectionWakeups.of({ changes: Stream.fromQueue(wakeups) }), + ), + ); + + yield* SubscriptionRef.changes(shellState).pipe( + Stream.filter( + (value) => + value.status === "synchronizing" && + Option.isSome(value.snapshot) && + value.snapshot.value.snapshotSequence === 10, + ), + Stream.runHead, + ); + yield* Queue.offer(events, { kind: "synchronized" }); + yield* SubscriptionRef.changes(shellState).pipe( + Stream.filter((value) => value.status === "live"), + Stream.runHead, + ); + + yield* Queue.offer(wakeups, "application-active"); + yield* SubscriptionRef.changes(shellState).pipe( + Stream.filter( + (value) => + value.status === "synchronizing" && + Option.isSome(value.snapshot) && + value.snapshot.value.snapshotSequence === 20, + ), + Stream.runHead, + ); + + for (let attempt = 0; attempt < 100; attempt += 1) { + if ((yield* Ref.get(subscriptionCount)) >= 2) break; + yield* Effect.yieldNow; + } + + expect(yield* Ref.get(loaderCalls)).toBe(2); + expect(yield* Ref.get(subscriptionCount)).toBe(2); }), ); }); diff --git a/packages/client-runtime/src/state/shell.ts b/packages/client-runtime/src/state/shell.ts index faa70bc4f3a..6ccb11797f5 100644 --- a/packages/client-runtime/src/state/shell.ts +++ b/packages/client-runtime/src/state/shell.ts @@ -9,6 +9,7 @@ import * as Cause from "effect/Cause"; import * as Effect from "effect/Effect"; import * as Option from "effect/Option"; import * as Queue from "effect/Queue"; +import * as Ref from "effect/Ref"; import * as Stream from "effect/Stream"; import * as SubscriptionRef from "effect/SubscriptionRef"; import { AsyncResult, Atom } from "effect/unstable/reactivity"; @@ -16,9 +17,10 @@ import { AsyncResult, Atom } from "effect/unstable/reactivity"; import { EnvironmentRegistry } from "../connection/registry.ts"; import { connectionProjectionPhase } from "../connection/model.ts"; import { EnvironmentSupervisor } from "../connection/supervisor.ts"; +import * as ConnectionWakeups from "../connection/wakeups.ts"; import { safeErrorLogAttributes } from "../errors/safeLog.ts"; import { EnvironmentCacheStore } from "../platform/persistence.ts"; -import { subscribe } from "../rpc/client.ts"; +import { subscribeDynamic } from "../rpc/client.ts"; import { ShellSnapshotLoader } from "./shellSnapshotHttp.ts"; import { applyShellStreamEvent } from "./shellReducer.ts"; import type { EnvironmentCatalogState } from "./connections.ts"; @@ -50,6 +52,7 @@ export const makeEnvironmentShellState = Effect.fn("EnvironmentShellState.make") const supervisor = yield* EnvironmentSupervisor; const cache = yield* EnvironmentCacheStore; const snapshotLoader = yield* ShellSnapshotLoader; + const wakeups = yield* Effect.serviceOption(ConnectionWakeups.ConnectionWakeups); const environmentId = supervisor.target.environmentId; const cachedSnapshot = yield* cache.loadShell(environmentId).pipe( Effect.catch((error) => @@ -67,6 +70,7 @@ export const makeEnvironmentShellState = Effect.fn("EnvironmentShellState.make") status: shellStatusForSnapshot(cachedSnapshot), error: Option.none(), }); + const awaitingCompletion = yield* Ref.make(false); const persistence = yield* Queue.sliding(1); const persist = Effect.fn("EnvironmentShellState.persist")(function* ( @@ -90,10 +94,14 @@ export const makeEnvironmentShellState = Effect.fn("EnvironmentShellState.make") Effect.forkScoped, ); - const setDisconnected = SubscriptionRef.update(state, (current) => ({ - ...current, - status: shellStatusForSnapshot(current.snapshot), - })); + const setDisconnected = Ref.set(awaitingCompletion, false).pipe( + Effect.andThen( + SubscriptionRef.update(state, (current) => ({ + ...current, + status: shellStatusForSnapshot(current.snapshot), + })), + ), + ); const setSynchronizing = SubscriptionRef.update(state, (current) => ({ ...current, status: "synchronizing" as const, @@ -109,7 +117,8 @@ export const makeEnvironmentShellState = Effect.fn("EnvironmentShellState.make") }, ); const setStreamError = (error: unknown) => - Effect.logWarning("Could not synchronize the environment shell.").pipe( + Ref.set(awaitingCompletion, false).pipe( + Effect.andThen(Effect.logWarning("Could not synchronize the environment shell.")), Effect.annotateLogs({ environmentId, ...safeErrorLogAttributes(error), @@ -126,6 +135,16 @@ export const makeEnvironmentShellState = Effect.fn("EnvironmentShellState.make") const applyItem = Effect.fn("EnvironmentShellState.applyItem")(function* ( item: OrchestrationShellStreamItem, ) { + if (item.kind === "synchronized") { + yield* Ref.set(awaitingCompletion, false); + yield* SubscriptionRef.update(state, (current) => + Option.isSome(current.snapshot) + ? { ...current, status: "live" as const, error: Option.none() } + : current, + ); + return; + } + const current = yield* SubscriptionRef.get(state); const nextSnapshot = item.kind === "snapshot" @@ -141,52 +160,64 @@ export const makeEnvironmentShellState = Effect.fn("EnvironmentShellState.make") return; } + const waiting = yield* Ref.get(awaitingCompletion); yield* SubscriptionRef.set(state, { snapshot: Option.some(nextSnapshot), - status: "live", + status: waiting ? "synchronizing" : "live", error: Option.none(), }); yield* Queue.offer(persistence, nextSnapshot); }); - yield* Effect.forkScoped( - Effect.gen(function* () { - // Establish the base shell snapshot to resume from, minimizing bytes over - // the wire: - // - Warm cache: reuse the cached snapshot (zero network) and resume via - // `afterSequence` so we only receive shell events since the cached - // sequence. - // - Cold cache: load the full shell snapshot over HTTP (gzip-compressible, - // and off the socket), then resume via `afterSequence`. - // If no base can be established we fall back to the socket-embedded - // snapshot so the shell still synchronizes. Overlapping/replayed events are - // deduped by sequence in applyItem. - const base = Option.isSome(cachedSnapshot) - ? cachedSnapshot - : yield* Effect.gen(function* () { - const prepared = yield* SubscriptionRef.changes(supervisor.prepared).pipe( - Stream.filter(Option.isSome), - Stream.map((current) => current.value), - Stream.runHead, - ); - return Option.isSome(prepared) - ? yield* snapshotLoader.load(prepared.value) - : Option.none(); - }); + const foregroundResubscriptions = Option.match(wakeups, { + onNone: () => Stream.never, + onSome: (service) => + service.changes.pipe(Stream.filter((reason) => reason === "application-active")), + }); - if (Option.isSome(base)) { - yield* applyItem({ kind: "snapshot", snapshot: base.value }); - } + yield* setSynchronizing; + yield* Effect.forkScoped( + subscribeDynamic( + ORCHESTRATION_WS_METHODS.subscribeShell, + Effect.fn("EnvironmentShellState.makeSubscribeInput")(function* (session) { + const supportsCompletionMarker = yield* session.initialConfig.pipe( + Effect.map((config) => config.shellResumeCompletionMarker === true), + Effect.orElseSucceed(() => false), + ); + yield* Ref.set(awaitingCompletion, supportsCompletionMarker); + yield* setSynchronizing; - const subscribeInput = Option.match(base, { - onNone: () => ({}), - onSome: (snapshot) => ({ afterSequence: snapshot.snapshotSequence }), - }); + const prepared = yield* SubscriptionRef.get(supervisor.prepared).pipe( + Effect.flatMap( + Option.match({ + onSome: Effect.succeed, + onNone: () => + SubscriptionRef.changes(supervisor.prepared).pipe( + Stream.filter(Option.isSome), + Stream.map((value) => value.value), + Stream.runHead, + Effect.map(Option.getOrThrow), + ), + }), + ), + ); + const httpSnapshot = yield* snapshotLoader.load(prepared); + if (Option.isSome(httpSnapshot)) { + yield* applyItem({ kind: "snapshot", snapshot: httpSnapshot.value }); + return { + afterSequence: httpSnapshot.value.snapshotSequence, + ...(supportsCompletionMarker ? { requestCompletionMarker: true as const } : {}), + }; + } - yield* subscribe(ORCHESTRATION_WS_METHODS.subscribeShell, subscribeInput, { + return supportsCompletionMarker ? { requestCompletionMarker: true as const } : {}; + }), + { onExpectedFailure: (cause) => setStreamError(Cause.squash(cause)), - }).pipe(Stream.runForEach(applyItem)); - }), + retryExpectedFailureAfter: "250 millis", + resubscribe: foregroundResubscriptions, + }, + ).pipe(Stream.runForEach(applyItem)), ); yield* SubscriptionRef.changes(supervisor.state).pipe( Stream.runForEach((connectionState) => { diff --git a/packages/client-runtime/src/state/threads-sync.test.ts b/packages/client-runtime/src/state/threads-sync.test.ts index 630a5f27d16..f3c4c4e6338 100644 --- a/packages/client-runtime/src/state/threads-sync.test.ts +++ b/packages/client-runtime/src/state/threads-sync.test.ts @@ -26,6 +26,7 @@ import { type PreparedConnection, type SupervisorConnectionState, } from "../connection/model.ts"; +import * as ConnectionWakeups from "../connection/wakeups.ts"; import * as EnvironmentSupervisor from "../connection/supervisor.ts"; import * as Persistence from "../platform/persistence.ts"; import * as RpcSession from "../rpc/session.ts"; @@ -98,10 +99,17 @@ const ACTIVE_THREAD: OrchestrationThread = { type TestThreadInput = OrchestrationThreadStreamItem | Error; -function testSession(client: WsRpcProtocolClient): RpcSession.RpcSession { +function testSession( + client: WsRpcProtocolClient, + options?: { readonly completionMarker?: boolean }, +): RpcSession.RpcSession { return { client, - initialConfig: Effect.never, + initialConfig: Effect.succeed( + options?.completionMarker === true + ? ({ threadResumeCompletionMarker: true } as never) + : ({} as never), + ), ready: Effect.void, probe: Effect.void, closed: Effect.never, @@ -122,6 +130,7 @@ function awaitThreadState( const makeHarness = Effect.fn("TestEnvironmentThreads.makeHarness")(function* (options?: { readonly cached?: OrchestrationThread; readonly httpSnapshot?: Option.Option; + readonly completionMarker?: boolean; }) { const inputs = yield* Queue.unbounded(); const observed = yield* Queue.unbounded(); @@ -130,8 +139,10 @@ const makeHarness = Effect.fn("TestEnvironmentThreads.makeHarness")(function* (o const subscriptionCount = yield* Ref.make(0); const loaderCalls = yield* Ref.make(0); const lastSubscribeAfterSequence = yield* Ref.make(undefined); + const lastRequestCompletionMarker = yield* Ref.make(undefined); const savedThreads = yield* Ref.make>([]); const removedThreads = yield* Ref.make>([]); + const wakeups = yield* Queue.unbounded(); const supervisorState = yield* SubscriptionRef.make( AVAILABLE_CONNECTION_STATE, ); @@ -142,16 +153,25 @@ const makeHarness = Effect.fn("TestEnvironmentThreads.makeHarness")(function* (o ), ); const client = { - [ORCHESTRATION_WS_METHODS.subscribeThread]: (input: { readonly afterSequence?: number }) => + [ORCHESTRATION_WS_METHODS.subscribeThread]: (input: { + readonly afterSequence?: number; + readonly requestCompletionMarker?: boolean; + }) => Stream.unwrap( Ref.updateAndGet(subscriptionCount, (count) => count + 1).pipe( Effect.andThen(Ref.set(lastSubscribeAfterSequence, input.afterSequence)), + Effect.andThen(Ref.set(lastRequestCompletionMarker, input.requestCompletionMarker)), Effect.as(streamFrom(inputs)), ), ), } as unknown as WsRpcProtocolClient; const supervisorSession = yield* SubscriptionRef.make>( - Option.some(testSession(client)), + Option.some( + testSession( + client, + options?.completionMarker === true ? { completionMarker: true } : undefined, + ), + ), ); const prepared = yield* SubscriptionRef.make>( Option.some(PREPARED), @@ -201,6 +221,10 @@ const makeHarness = Effect.fn("TestEnvironmentThreads.makeHarness")(function* (o Effect.provideService(EnvironmentSupervisor.EnvironmentSupervisor, supervisor), Effect.provideService(Persistence.EnvironmentCacheStore, cache), Effect.provideService(ThreadSnapshotLoader, snapshotLoader), + Effect.provideService( + ConnectionWakeups.ConnectionWakeups, + ConnectionWakeups.ConnectionWakeups.of({ changes: Stream.fromQueue(wakeups) }), + ), ); yield* SubscriptionRef.changes(threadState).pipe( Stream.runForEach((state) => @@ -217,11 +241,21 @@ const makeHarness = Effect.fn("TestEnvironmentThreads.makeHarness")(function* (o subscriptionCount, loaderCalls, lastSubscribeAfterSequence, + lastRequestCompletionMarker, supervisorState, supervisorSession, savedThreads, removedThreads, - replaceSession: SubscriptionRef.set(supervisorSession, Option.some(testSession(client))), + wakeups, + replaceSession: SubscriptionRef.set( + supervisorSession, + Option.some( + testSession( + client, + options?.completionMarker === true ? { completionMarker: true } : undefined, + ), + ), + ), }; }); @@ -233,6 +267,8 @@ const snapshot = (thread: OrchestrationThread): OrchestrationThreadStreamItem => }, }); +const synchronized = (): OrchestrationThreadStreamItem => ({ kind: "synchronized" }); + const titleUpdated = (title: string, sequence = 2): OrchestrationThreadStreamItem => ({ kind: "event", event: { @@ -415,6 +451,35 @@ describe("EnvironmentThreads", () => { }), ); + it.effect("does not resurrect a deleted thread when the app returns to the foreground", () => + Effect.gen(function* () { + const harness = yield* makeHarness({ + cached: BASE_THREAD, + completionMarker: true, + httpSnapshot: Option.some({ + snapshotSequence: 4, + thread: { ...BASE_THREAD, title: "Stale HTTP thread" }, + }), + }); + yield* Queue.offer(harness.inputs, snapshot(BASE_THREAD)); + yield* Queue.offer(harness.inputs, deleted()); + yield* awaitThreadState(harness.observed, (value) => value.status === "deleted"); + + expect(yield* Ref.get(harness.loaderCalls)).toBe(0); + yield* Queue.offer(harness.wakeups, "application-active"); + for (let attempt = 0; attempt < 100; attempt += 1) { + if ((yield* Ref.get(harness.subscriptionCount)) >= 2) break; + yield* Effect.yieldNow; + } + + const latest = yield* Ref.get(harness.latest); + expect(yield* Ref.get(harness.subscriptionCount)).toBe(2); + expect(yield* Ref.get(harness.loaderCalls)).toBe(0); + expect(latest.status).toBe("deleted"); + expect(Option.isNone(latest.data)).toBe(true); + }), + ); + it.effect("preserves data after a domain failure and resumes on a replacement session", () => Effect.gen(function* () { const harness = yield* makeHarness({ cached: BASE_THREAD }); @@ -529,4 +594,104 @@ describe("EnvironmentThreads", () => { expect((yield* Ref.get(harness.latest)).status).toBe("live"); }), ); + + it.effect("keeps replayed updates synchronizing until the completion marker arrives", () => + Effect.gen(function* () { + const harness = yield* makeHarness({ cached: BASE_THREAD, completionMarker: true }); + yield* awaitThreadState( + harness.observed, + (value) => value.status === "synchronizing" && Option.isSome(value.data), + ); + expect(yield* Ref.get(harness.lastRequestCompletionMarker)).toBe(true); + + yield* Queue.offer( + harness.inputs, + titleUpdated("Caught-up title", CACHED_SNAPSHOT_SEQUENCE + 1), + ); + const catchingUp = yield* awaitThreadState( + harness.observed, + (value) => + value.status === "synchronizing" && + Option.isSome(value.data) && + value.data.value.title === "Caught-up title", + ); + expect(catchingUp.status).toBe("synchronizing"); + + yield* Queue.offer(harness.inputs, synchronized()); + const live = yield* awaitThreadState( + harness.observed, + (value) => value.status === "live" && Option.isSome(value.data), + ); + expect(Option.getOrThrow(live.data).title).toBe("Caught-up title"); + }), + ); + + it.effect("resumes replacement sessions from the latest applied sequence", () => + Effect.gen(function* () { + const harness = yield* makeHarness({ cached: BASE_THREAD, completionMarker: true }); + yield* Queue.offer( + harness.inputs, + titleUpdated("Latest title", CACHED_SNAPSHOT_SEQUENCE + 1), + ); + yield* Queue.offer(harness.inputs, synchronized()); + yield* awaitThreadState( + harness.observed, + (value) => + value.status === "live" && + Option.isSome(value.data) && + value.data.value.title === "Latest title", + ); + + yield* harness.replaceSession; + for (let attempt = 0; attempt < 100; attempt += 1) { + if ((yield* Ref.get(harness.subscriptionCount)) >= 2) break; + yield* Effect.yieldNow; + } + + expect(yield* Ref.get(harness.subscriptionCount)).toBe(2); + expect(yield* Ref.get(harness.lastSubscribeAfterSequence)).toBe(CACHED_SNAPSHOT_SEQUENCE + 1); + expect((yield* Ref.get(harness.latest)).status).toBe("synchronizing"); + }), + ); + + it.effect("resubscribes on app foreground from the latest applied sequence", () => + Effect.gen(function* () { + const harness = yield* makeHarness({ cached: BASE_THREAD, completionMarker: true }); + yield* Queue.offer( + harness.inputs, + titleUpdated("Latest title", CACHED_SNAPSHOT_SEQUENCE + 1), + ); + yield* Queue.offer(harness.inputs, synchronized()); + yield* awaitThreadState( + harness.observed, + (value) => + value.status === "live" && + Option.isSome(value.data) && + value.data.value.title === "Latest title", + ); + + yield* Queue.offer(harness.wakeups, "application-active"); + const synchronizing = yield* awaitThreadState( + harness.observed, + (value) => value.status === "synchronizing" && Option.isSome(value.data), + ); + for (let attempt = 0; attempt < 100; attempt += 1) { + if ((yield* Ref.get(harness.subscriptionCount)) >= 2) break; + yield* Effect.yieldNow; + } + + expect(synchronizing.status).toBe("synchronizing"); + expect(yield* Ref.get(harness.subscriptionCount)).toBe(2); + expect(yield* Ref.get(harness.lastSubscribeAfterSequence)).toBe(CACHED_SNAPSHOT_SEQUENCE + 1); + expect(yield* Ref.get(harness.lastRequestCompletionMarker)).toBe(true); + expect(yield* Ref.get(harness.loaderCalls)).toBe(0); + + yield* Queue.offer(harness.inputs, synchronized()); + const live = yield* awaitThreadState( + harness.observed, + (value) => value.status === "live" && Option.isSome(value.data), + ); + expect(Option.getOrThrow(live.data).title).toBe("Latest title"); + }), + ); }); diff --git a/packages/client-runtime/src/state/threads.ts b/packages/client-runtime/src/state/threads.ts index 0c9ec340610..196229cc8b1 100644 --- a/packages/client-runtime/src/state/threads.ts +++ b/packages/client-runtime/src/state/threads.ts @@ -10,6 +10,7 @@ import * as Cause from "effect/Cause"; import * as Effect from "effect/Effect"; import * as Option from "effect/Option"; import * as Queue from "effect/Queue"; +import * as Ref from "effect/Ref"; import * as Stream from "effect/Stream"; import * as SubscriptionRef from "effect/SubscriptionRef"; import { Atom } from "effect/unstable/reactivity"; @@ -17,8 +18,9 @@ import { Atom } from "effect/unstable/reactivity"; import { EnvironmentRegistry } from "../connection/registry.ts"; import { connectionProjectionPhase } from "../connection/model.ts"; import { EnvironmentSupervisor } from "../connection/supervisor.ts"; +import * as ConnectionWakeups from "../connection/wakeups.ts"; import { EnvironmentCacheStore } from "../platform/persistence.ts"; -import { subscribe } from "../rpc/client.ts"; +import { subscribeDynamic } from "../rpc/client.ts"; import { ThreadSnapshotLoader } from "./threadSnapshotHttp.ts"; import { parseThreadKey, threadKey } from "./entities.ts"; import { applyThreadDetailEvent } from "./threadReducer.ts"; @@ -52,6 +54,7 @@ export const makeEnvironmentThreadState = Effect.fn("EnvironmentThreadState.make const supervisor = yield* EnvironmentSupervisor; const cache = yield* EnvironmentCacheStore; const snapshotLoader = yield* ThreadSnapshotLoader; + const wakeups = yield* Effect.serviceOption(ConnectionWakeups.ConnectionWakeups); const environmentId = supervisor.target.environmentId; const cached = yield* cache.loadThread(environmentId, threadId).pipe( Effect.catch((error) => @@ -76,6 +79,7 @@ export const makeEnvironmentThreadState = Effect.fn("EnvironmentThreadState.make const lastSequence = yield* SubscriptionRef.make( Option.match(cached, { onNone: () => 0, onSome: (snapshot) => snapshot.snapshotSequence }), ); + const awaitingCompletion = yield* Ref.make(false); const persistence = yield* Queue.sliding(1); const persist = Effect.fn("EnvironmentThreadState.persist")(function* ( @@ -100,11 +104,15 @@ export const makeEnvironmentThreadState = Effect.fn("EnvironmentThreadState.make Effect.forkScoped, ); - const setSynchronizing = SubscriptionRef.update(state, (current) => ({ - ...current, - status: "synchronizing" as const, - error: Option.none(), - })); + const setSynchronizing = SubscriptionRef.update(state, (current) => + current.status === "deleted" + ? current + : { + ...current, + status: "synchronizing" as const, + error: Option.none(), + }, + ); const setReady = SubscriptionRef.update(state, (current) => current.status === "live" || current.status === "deleted" ? current @@ -114,23 +122,32 @@ export const makeEnvironmentThreadState = Effect.fn("EnvironmentThreadState.make error: Option.none(), }, ); - const setDisconnected = SubscriptionRef.update(state, (current) => ({ - ...current, - status: current.status === "deleted" ? current.status : statusWithoutLiveData(current.data), - })); - const setStreamError = (cause: Cause.Cause) => - SubscriptionRef.update(state, (current) => ({ + const setDisconnected = Effect.gen(function* () { + yield* Ref.set(awaitingCompletion, false); + yield* SubscriptionRef.update(state, (current) => ({ ...current, status: current.status === "deleted" ? current.status : statusWithoutLiveData(current.data), - error: Option.some(formatThreadError(cause)), })); + }); + const setStreamError = (cause: Cause.Cause) => + Ref.set(awaitingCompletion, false).pipe( + Effect.andThen( + SubscriptionRef.update(state, (current) => ({ + ...current, + status: + current.status === "deleted" ? current.status : statusWithoutLiveData(current.data), + error: Option.some(formatThreadError(cause)), + })), + ), + ); const setThread = Effect.fn("EnvironmentThreadState.setThread")(function* ( thread: OrchestrationThread, ) { + const waiting = yield* Ref.get(awaitingCompletion); yield* SubscriptionRef.set(state, { data: Option.some(thread), - status: "live", + status: waiting ? "synchronizing" : "live", error: Option.none(), }); // Active threads can update many times per second and retain large tool @@ -143,6 +160,7 @@ export const makeEnvironmentThreadState = Effect.fn("EnvironmentThreadState.make }); const setDeleted = Effect.fn("EnvironmentThreadState.setDeleted")(function* () { + yield* Ref.set(awaitingCompletion, false); yield* SubscriptionRef.set(state, { data: Option.none(), status: "deleted", @@ -164,6 +182,16 @@ export const makeEnvironmentThreadState = Effect.fn("EnvironmentThreadState.make const applyItem = Effect.fn("EnvironmentThreadState.applyItem")(function* ( item: OrchestrationThreadStreamItem, ) { + if (item.kind === "synchronized") { + yield* Ref.set(awaitingCompletion, false); + yield* SubscriptionRef.update(state, (current) => + Option.isSome(current.data) && current.status !== "deleted" + ? { ...current, status: "live" as const, error: Option.none() } + : current, + ); + return; + } + if (item.kind === "snapshot") { yield* SubscriptionRef.set(lastSequence, item.snapshot.snapshotSequence); yield* setThread(item.snapshot.thread); @@ -205,48 +233,69 @@ export const makeEnvironmentThreadState = Effect.fn("EnvironmentThreadState.make Effect.forkScoped, ); + const foregroundResubscriptions = Option.match(wakeups, { + onNone: () => Stream.never, + onSome: (service) => + service.changes.pipe(Stream.filter((reason) => reason === "application-active")), + }); + yield* setSynchronizing; yield* Effect.forkScoped( - Effect.gen(function* () { - // Establish the base snapshot to resume from, minimizing bytes over the - // wire: - // - Warm cache: reuse the cached snapshot (zero network) and resume via - // `afterSequence` so we only receive events since the cached sequence. - // - Cold cache: load the full snapshot over HTTP (gzip-compressible, and - // off the socket), then resume via `afterSequence`. - // If no base can be established we fall back to the socket-embedded - // snapshot so the thread still synchronizes. Overlapping/replayed events - // are deduped by sequence in applyItem. - const base = Option.isSome(cached) - ? cached - : yield* Effect.gen(function* () { - // Cold cache only: wait for a prepared connection so we can - // authenticate the HTTP request; this mirrors the socket path, which - // likewise waits for a live session. - const prepared = yield* SubscriptionRef.changes(supervisor.prepared).pipe( - Stream.filter(Option.isSome), - Stream.map((current) => current.value), - Stream.runHead, - ); - return Option.isSome(prepared) - ? yield* snapshotLoader.load(prepared.value, threadId) - : Option.none(); - }); + subscribeDynamic( + ORCHESTRATION_WS_METHODS.subscribeThread, + Effect.fn("EnvironmentThreadState.makeSubscribeInput")(function* (session) { + const supportsCompletionMarker = yield* session.initialConfig.pipe( + Effect.map((config) => config.threadResumeCompletionMarker === true), + Effect.orElseSucceed(() => false), + ); + yield* Ref.set(awaitingCompletion, supportsCompletionMarker); + yield* setSynchronizing; - if (Option.isSome(base)) { - yield* applyItem({ kind: "snapshot", snapshot: base.value }); - } + let current = yield* SubscriptionRef.get(state); + if (Option.isNone(current.data) && current.status !== "deleted") { + const prepared = yield* SubscriptionRef.get(supervisor.prepared).pipe( + Effect.flatMap( + Option.match({ + onSome: Effect.succeed, + onNone: () => + SubscriptionRef.changes(supervisor.prepared).pipe( + Stream.filter(Option.isSome), + Stream.map((value) => value.value), + Stream.runHead, + Effect.map(Option.getOrThrow), + ), + }), + ), + ); + const httpSnapshot = yield* snapshotLoader.load(prepared, threadId); + if (Option.isSome(httpSnapshot)) { + yield* applyItem({ kind: "snapshot", snapshot: httpSnapshot.value }); + current = yield* SubscriptionRef.get(state); + } + } - const subscribeInput = Option.match(base, { - onNone: () => ({ threadId }), - onSome: (snapshot) => ({ threadId, afterSequence: snapshot.snapshotSequence }), - }); + const sequence = yield* SubscriptionRef.get(lastSequence); + const canResume = Option.isSome(current.data); + if (!supportsCompletionMarker && canResume) { + yield* SubscriptionRef.update(state, (value) => ({ + ...value, + status: value.status === "deleted" ? value.status : ("live" as const), + error: Option.none(), + })); + } - yield* subscribe(ORCHESTRATION_WS_METHODS.subscribeThread, subscribeInput, { + return { + threadId, + ...(canResume ? { afterSequence: sequence } : {}), + ...(supportsCompletionMarker ? { requestCompletionMarker: true as const } : {}), + }; + }), + { onExpectedFailure: setStreamError, retryExpectedFailureAfter: "250 millis", - }).pipe(Stream.runForEach(applyItem)); - }), + resubscribe: foregroundResubscriptions, + }, + ).pipe(Stream.runForEach(applyItem)), ); yield* Effect.addFinalizer(() => diff --git a/packages/contracts/src/orchestration.ts b/packages/contracts/src/orchestration.ts index 7e9c421b5df..c0c9c62d080 100644 --- a/packages/contracts/src/orchestration.ts +++ b/packages/contracts/src/orchestration.ts @@ -443,6 +443,9 @@ export const OrchestrationShellStreamEvent = Schema.Union([ export type OrchestrationShellStreamEvent = typeof OrchestrationShellStreamEvent.Type; export const OrchestrationShellStreamItem = Schema.Union([ + Schema.Struct({ + kind: Schema.Literal("synchronized"), + }), Schema.Struct({ kind: Schema.Literal("snapshot"), snapshot: OrchestrationShellSnapshot, @@ -461,6 +464,11 @@ export const OrchestrationSubscribeShellInput = Schema.Struct({ * client). */ afterSequence: Schema.optionalKey(NonNegativeInt), + /** + * Requests an explicit marker after the subscription has emitted its initial + * snapshot or catch-up replay and before it begins emitting live events. + */ + requestCompletionMarker: Schema.optionalKey(Schema.Boolean), }); export type OrchestrationSubscribeShellInput = typeof OrchestrationSubscribeShellInput.Type; @@ -474,6 +482,11 @@ export const OrchestrationSubscribeThreadInput = Schema.Struct({ * sequence on the client). */ afterSequence: Schema.optionalKey(NonNegativeInt), + /** + * Requests an explicit marker after the subscription has emitted its initial + * snapshot or catch-up replay and before it begins emitting live events. + */ + requestCompletionMarker: Schema.optionalKey(Schema.Boolean), }); export type OrchestrationSubscribeThreadInput = typeof OrchestrationSubscribeThreadInput.Type; @@ -1135,6 +1148,9 @@ export const OrchestrationEvent = Schema.Union([ export type OrchestrationEvent = typeof OrchestrationEvent.Type; export const OrchestrationThreadStreamItem = Schema.Union([ + Schema.Struct({ + kind: Schema.Literal("synchronized"), + }), Schema.Struct({ kind: Schema.Literal("snapshot"), snapshot: OrchestrationThreadDetailSnapshot, diff --git a/packages/contracts/src/server.ts b/packages/contracts/src/server.ts index b76ea965afe..316f09693ec 100644 --- a/packages/contracts/src/server.ts +++ b/packages/contracts/src/server.ts @@ -417,6 +417,10 @@ export const ServerConfig = Schema.Struct({ availableEditors: Schema.Array(EditorId), observability: ServerObservability, settings: ServerSettings, + /** Whether shell subscriptions can emit an opt-in catch-up completion marker. */ + shellResumeCompletionMarker: Schema.optionalKey(Schema.Boolean), + /** Whether thread subscriptions can emit an opt-in catch-up completion marker. */ + threadResumeCompletionMarker: Schema.optionalKey(Schema.Boolean), }); export type ServerConfig = typeof ServerConfig.Type;