authorgravatar for julius0216@outlook.comJulius Marminge <julius0216@outlook.com> 2026-07-20 11:54:59+02:00
committergravatar for noreply@github.comGitHub <noreply@github.com> 2026-07-20 11:54:59+02:00
log8e3467fe60c49ee739b278229156108032b72e25
tree5b44e2488e27969f366526faab7213f06f2a8922
parent2fae0d0ac300353be774c9f9fe79d3af025d4f91
signature Signed by PGP key B5690EEEBB952194

Synchronize mobile threads with authoritative shell snapshots (#4163)

Co-authored-by: codex <codex@users.noreply.github.com>

17 files changed, 779 insertions(+), 233 deletions(-)

apps/mobile/src/Stack.tsx+1-3
...@@ -15,7 +15,6 @@ import { DynamicColorIOS, Platform, Pressable, ScrollView, StyleSheet } from "re...@@ -15,7 +15,6 @@ import { DynamicColorIOS, Platform, Pressable, ScrollView, StyleSheet } from "re
15import { useResolveClassNames } from "uniwind";15import { useResolveClassNames } from "uniwind";
1616
17import { AppText as Text } from "./components/AppText";17import { AppText as Text } from "./components/AppText";
18import { renderCompactBrandTitle } from "./components/CompactBrandTitle";
19import { ArchivedThreadsRouteScreen } from "./features/archive/ArchivedThreadsRouteScreen";18import { ArchivedThreadsRouteScreen } from "./features/archive/ArchivedThreadsRouteScreen";
20import { useAgentNotificationNavigation } from "./features/agent-awareness/notificationNavigation";19import { useAgentNotificationNavigation } from "./features/agent-awareness/notificationNavigation";
21import { ClerkSettingsSheetDetentProvider } from "./features/cloud/ClerkSettingsSheetDetent";20import { ClerkSettingsSheetDetentProvider } from "./features/cloud/ClerkSettingsSheetDetent";
...@@ -378,8 +377,7 @@ export const RootStack = createNativeStackNavigator({...@@ -378,8 +377,7 @@ export const RootStack = createNativeStackNavigator({
378 ...GLASS_HEADER_OPTIONS,377 ...GLASS_HEADER_OPTIONS,
379 contentStyle: { backgroundColor: "transparent" },378 contentStyle: { backgroundColor: "transparent" },
380 headerBackVisible: false,379 headerBackVisible: false,
381 headerTitle: renderCompactBrandTitle,380 title: "Threads",
382 title: "T3 Code",
383 },381 },
384 }),382 }),
385 Thread: createNativeStackScreen({383 Thread: createNativeStackScreen({
apps/mobile/src/components/CompactBrandTitle.tsx deleted-60
...@@ -1,60 +0,0 @@
1import { View } from "react-native";
2
3import { AppText as Text } from "./AppText";
4import { T3Wordmark } from "./T3Wordmark";
5import { useThemeColor } from "../lib/useThemeColor";
6
7/**
8 * Compact brand lockup sized for native navigation bars.
9 */
10export function CompactBrandTitle() {
11 const iconColor = useThemeColor("--color-icon");
12 const mutedColor = useThemeColor("--color-foreground-muted");
13 const subtleColor = useThemeColor("--color-subtle");
14
15 return (
16 <View
17 aria-level={1}
18 accessibilityLabel="T3 Code"
19 accessible
20 role="heading"
21 style={{ alignItems: "center", flexDirection: "row", gap: 6 }}
22 >
23 <T3Wordmark color={iconColor} height={11} />
24 <Text
25 style={{
26 color: mutedColor,
27 fontFamily: "DMSans-Medium",
28 fontSize: 14,
29 letterSpacing: -0.35,
30 }}
31 >
32 Code
33 </Text>
34 <View
35 style={{
36 backgroundColor: subtleColor,
37 borderRadius: 999,
38 paddingHorizontal: 6,
39 paddingVertical: 2,
40 }}
41 >
42 <Text
43 style={{
44 color: mutedColor,
45 fontFamily: "DMSans-Bold",
46 fontSize: 9,
47 letterSpacing: 0.9,
48 textTransform: "uppercase",
49 }}
50 >
51 Alpha
52 </Text>
53 </View>
54 </View>
55 );
56}
57
58export function renderCompactBrandTitle() {
59 return <CompactBrandTitle />;
60}
apps/mobile/src/features/home/HomeRouteScreen.tsx+1-4
...@@ -4,7 +4,6 @@ import { useNavigation } from "@react-navigation/native";...@@ -4,7 +4,6 @@ import { useNavigation } from "@react-navigation/native";
4import { useMemo, useState } from "react";4import { useMemo, useState } from "react";
55
6import { NativeHeaderToolbar, NativeStackScreenOptions } from "../../native/StackHeader";6import { NativeHeaderToolbar, NativeStackScreenOptions } from "../../native/StackHeader";
7import { renderCompactBrandTitle } from "../../components/CompactBrandTitle";
8import { useProjects, useThreadShells } from "../../state/entities";7import { useProjects, useThreadShells } from "../../state/entities";
9import { usePendingNewTasks } from "../../state/use-pending-new-tasks";8import { usePendingNewTasks } from "../../state/use-pending-new-tasks";
10import { useWorkspaceState } from "../../state/workspace";9import { useWorkspaceState } from "../../state/workspace";
...@@ -87,9 +86,7 @@ export function HomeRouteScreen() {...@@ -87,9 +86,7 @@ export function HomeRouteScreen() {
87 >86 >
88 <>87 <>
89 {/* Restore the compact title in case the split branch blanked it. */}88 {/* Restore the compact title in case the split branch blanked it. */}
90 <NativeStackScreenOptions89 <NativeStackScreenOptions options={{ title: "Threads", headerTitle: "Threads" }} />
91 options={{ title: "T3 Code", headerTitle: renderCompactBrandTitle }}
92 />
93 <HomeHeader90 <HomeHeader
94 environments={environments}91 environments={environments}
95 searchQuery={searchQuery}92 searchQuery={searchQuery}
apps/mobile/src/features/home/HomeScreen.tsx+10-1
...@@ -399,11 +399,20 @@ export function HomeScreen(props: HomeScreenProps) {...@@ -399,11 +399,20 @@ export function HomeScreen(props: HomeScreenProps) {
399 onAction={!props.catalogState.hasReadyEnvironment ? props.onAddConnection : undefined}399 onAction={!props.catalogState.hasReadyEnvironment ? props.onAddConnection : undefined}
400 variant="plain"400 variant="plain"
401 />401 />
402 {emptyState.loading ? (402 {emptyState.loading && !shouldShowConnectionStatus ? (
403 <View className="mt-4 items-center">403 <View className="mt-4 items-center">
404 <ActivityIndicator color={accentColor} />404 <ActivityIndicator color={accentColor} />
405 </View>405 </View>
406 ) : null}406 ) : null}
407 {shouldShowConnectionStatus && Platform.OS === "ios" ? (
408 <View className="mt-4">
409 <WorkspaceConnectionStatus
410 state={props.catalogState}
411 onPress={props.onOpenEnvironments}
412 variant="sidebar"
413 />
414 </View>
415 ) : null}
407 </View>416 </View>
408 {connectionStatus}417 {connectionStatus}
409 </View>418 </View>
apps/mobile/src/features/home/WorkspaceConnectionStatus.test.ts+17
...@@ -67,4 +67,21 @@ describe("workspace connection status", () => {...@@ -67,4 +67,21 @@ describe("workspace connection status", () => {
67 expect(shouldShowWorkspaceConnectionStatus(state)).toBe(true);67 expect(shouldShowWorkspaceConnectionStatus(state)).toBe(true);
68 expect(workspaceConnectionStatusLabel(state)).toBe("Could not reach Julius’s Mac mini");68 expect(workspaceConnectionStatusLabel(state)).toBe("Could not reach Julius’s Mac mini");
69 });69 });
70
71 it("shows shell catch-up while cached threads remain visible", () => {
72 const state = workspaceState({ hasPendingShellSnapshot: true });
73
74 expect(shouldShowWorkspaceConnectionStatus(state)).toBe(true);
75 expect(workspaceConnectionStatusLabel(state)).toBe("Syncing threads...");
76 });
77
78 it("distinguishes initial shell loading from cached catch-up", () => {
79 const state = workspaceState({
80 hasLoadedShellSnapshot: false,
81 hasPendingShellSnapshot: true,
82 });
83
84 expect(shouldShowWorkspaceConnectionStatus(state)).toBe(true);
85 expect(workspaceConnectionStatusLabel(state)).toBe("Loading threads...");
86 });
70});87});
apps/mobile/src/features/home/WorkspaceConnectionStatus.tsx+5-2
...@@ -12,7 +12,10 @@ export function WorkspaceConnectionStatus(props: {...@@ -12,7 +12,10 @@ export function WorkspaceConnectionStatus(props: {
12 readonly variant?: "floating" | "sidebar";12 readonly variant?: "floating" | "sidebar";
13}) {13}) {
14 const iconColor = useThemeColor("--color-icon-muted");14 const iconColor = useThemeColor("--color-icon-muted");
15 const isReconnecting = props.state.connectingEnvironments.length > 0;15 const isSynchronizing =
16 props.state.networkStatus !== "offline" &&
17 props.state.connectionError === null &&
18 (props.state.connectingEnvironments.length > 0 || props.state.hasPendingShellSnapshot);
16 const variant = props.variant ?? "floating";19 const variant = props.variant ?? "floating";
1720
18 return (21 return (
...@@ -37,7 +40,7 @@ export function WorkspaceConnectionStatus(props: {...@@ -37,7 +40,7 @@ export function WorkspaceConnectionStatus(props: {
37 : undefined40 : undefined
38 }41 }
39 >42 >
40 {isReconnecting ? (43 {isSynchronizing ? (
41 <ActivityIndicator color={iconColor} size="small" />44 <ActivityIndicator color={iconColor} size="small" />
42 ) : (45 ) : (
43 <SymbolView name="wifi.slash" size={15} tintColor={iconColor} type="monochrome" />46 <SymbolView name="wifi.slash" size={15} tintColor={iconColor} type="monochrome" />
apps/mobile/src/features/home/workspace-connection-status.ts+4
...@@ -5,6 +5,7 @@ export function shouldShowWorkspaceConnectionStatus(state: WorkspaceState): bool...@@ -5,6 +5,7 @@ export function shouldShowWorkspaceConnectionStatus(state: WorkspaceState): bool
5 state.networkStatus === "offline" ||5 state.networkStatus === "offline" ||
6 state.connectionError !== null ||6 state.connectionError !== null ||
7 state.hasConnectingEnvironment ||7 state.hasConnectingEnvironment ||
8 state.hasPendingShellSnapshot ||
8 (state.hasLoadedShellSnapshot && !state.hasReadyEnvironment)9 (state.hasLoadedShellSnapshot && !state.hasReadyEnvironment)
9 );10 );
10}11}
...@@ -18,5 +19,8 @@ export function workspaceConnectionStatusLabel(state: WorkspaceState): string {...@@ -18,5 +19,8 @@ export function workspaceConnectionStatusLabel(state: WorkspaceState): string {
18 return `Reconnecting ${state.connectingEnvironments.length} environments`;19 return `Reconnecting ${state.connectingEnvironments.length} environments`;
19 }20 }
20 if (state.connectionError !== null) return state.connectionError;21 if (state.connectionError !== null) return state.connectionError;
22 if (state.hasPendingShellSnapshot) {
23 return state.hasLoadedShellSnapshot ? "Syncing threads..." : "Loading threads...";
24 }
21 return "Not connected";25 return "Not connected";
22}26}
apps/mobile/src/features/threads/sidebar-navigation-shell.tsx+1-3
...@@ -11,7 +11,6 @@ import {...@@ -11,7 +11,6 @@ import {
11import type { ReactNode } from "react";11import type { ReactNode } from "react";
12import { Platform, useColorScheme } from "react-native";12import { Platform, useColorScheme } from "react-native";
1313
14import { renderCompactBrandTitle } from "../../components/CompactBrandTitle";
15import { nativeHeaderScrollEdgeEffects } from "../../native/StackHeader";14import { nativeHeaderScrollEdgeEffects } from "../../native/StackHeader";
1615
17const SCROLL_EDGE_EFFECTS = nativeHeaderScrollEdgeEffects(Platform.OS, Platform.Version);16const SCROLL_EDGE_EFFECTS = nativeHeaderScrollEdgeEffects(Platform.OS, Platform.Version);
...@@ -35,11 +34,10 @@ const SIDEBAR_SCREEN_OPTIONS: SidebarScreenOptions = {...@@ -35,11 +34,10 @@ const SIDEBAR_SCREEN_OPTIONS: SidebarScreenOptions = {
35 headerShadowVisible: false,34 headerShadowVisible: false,
36 headerShown: true,35 headerShown: true,
37 headerStyle: { backgroundColor: "transparent" },36 headerStyle: { backgroundColor: "transparent" },
38 headerTitle: renderCompactBrandTitle,
39 headerTitleStyle: { fontSize: 18, fontWeight: "800" },37 headerTitleStyle: { fontSize: 18, fontWeight: "800" },
40 headerTransparent: true,38 headerTransparent: true,
41 scrollEdgeEffects: SCROLL_EDGE_EFFECTS,39 scrollEdgeEffects: SCROLL_EDGE_EFFECTS,
42 title: "T3 Code",40 title: "Threads",
43 unstable_navigationItemStyle: "editor",41 unstable_navigationItemStyle: "editor",
44};42};
4543
apps/server/src/server.test.ts+109
...@@ -3717,6 +3717,8 @@ it.layer(NodeServices.layer)("server router seam", (it) => {...@@ -3717,6 +3717,8 @@ it.layer(NodeServices.layer)("server router seam", (it) => {
37173717
3718 assert.equal(response.environment.environmentId, testEnvironmentDescriptor.environmentId);3718 assert.equal(response.environment.environmentId, testEnvironmentDescriptor.environmentId);
3719 assert.equal(response.auth.policy, "desktop-managed-local");3719 assert.equal(response.auth.policy, "desktop-managed-local");
3720 assert.equal(response.shellResumeCompletionMarker, true);
3721 assert.equal(response.threadResumeCompletionMarker, true);
3720 }).pipe(Effect.provide(NodeHttpServer.layerTest)),3722 }).pipe(Effect.provide(NodeHttpServer.layerTest)),
3721 );3723 );
37223724
...@@ -5607,6 +5609,113 @@ it.layer(NodeServices.layer)("server router seam", (it) => {...@@ -5607,6 +5609,113 @@ it.layer(NodeServices.layer)("server router seam", (it) => {
5607 }).pipe(Effect.provide(NodeHttpServer.layerTest)),5609 }).pipe(Effect.provide(NodeHttpServer.layerTest)),
5608 );5610 );
56095611
5612 it.effect("marks an empty shell catch-up replay as synchronized when requested", () =>
5613 Effect.gen(function* () {
5614 yield* buildAppUnderTest({
5615 layers: {
5616 orchestrationEngine: {
5617 readEvents: () => Stream.empty,
5618 },
5619 },
5620 });
5621
5622 const wsUrl = yield* getWsServerUrl("/ws");
5623 const firstItem = yield* Effect.scoped(
5624 withWsRpcClient(wsUrl, (client) =>
5625 client[ORCHESTRATION_WS_METHODS.subscribeShell]({
5626 afterSequence: 0,
5627 requestCompletionMarker: true,
5628 }).pipe(Stream.runHead),
5629 ),
5630 );
5631
5632 assert.deepEqual(Option.getOrThrow(firstItem), { kind: "synchronized" });
5633 }).pipe(Effect.provide(NodeHttpServer.layerTest)),
5634 );
5635
5636 it.effect("marks a socket thread snapshot as synchronized when requested", () =>
5637 Effect.gen(function* () {
5638 const thread = makeDefaultOrchestrationReadModel().threads[0]!;
5639 yield* buildAppUnderTest({
5640 layers: {
5641 projectionSnapshotQuery: {
5642 getThreadDetailSnapshot: () =>
5643 Effect.succeed(Option.some({ snapshotSequence: 1, thread })),
5644 },
5645 },
5646 });
5647
5648 const wsUrl = yield* getWsServerUrl("/ws");
5649 const items = yield* Effect.scoped(
5650 withWsRpcClient(wsUrl, (client) =>
5651 client[ORCHESTRATION_WS_METHODS.subscribeThread]({
5652 threadId: defaultThreadId,
5653 requestCompletionMarker: true,
5654 }).pipe(Stream.take(2), Stream.runCollect),
5655 ),
5656 );
5657
5658 assert.equal(items[0]?.kind, "snapshot");
5659 assert.deepEqual(items[1], { kind: "synchronized" });
5660 }).pipe(Effect.provide(NodeHttpServer.layerTest)),
5661 );
5662
5663 it.effect("buffers shell events published while the fallback snapshot loads", () =>
5664 Effect.gen(function* () {
5665 const liveEvents = yield* PubSub.unbounded<OrchestrationEvent>();
5666 const deletedEvent = {
5667 sequence: 2,
5668 eventId: EventId.make("event-shell-thread-deleted"),
5669 aggregateKind: "thread",
5670 aggregateId: defaultThreadId,
5671 occurredAt: "2026-01-01T00:00:01.000Z",
5672 commandId: null,
5673 causationEventId: null,
5674 correlationId: null,
5675 metadata: {},
5676 type: "thread.deleted",
5677 payload: {
5678 threadId: defaultThreadId,
5679 deletedAt: "2026-01-01T00:00:01.000Z",
5680 },
5681 } satisfies Extract<OrchestrationEvent, { type: "thread.deleted" }>;
5682
5683 yield* buildAppUnderTest({
5684 layers: {
5685 orchestrationEngine: {
5686 streamDomainEvents: Stream.fromPubSub(liveEvents),
5687 },
5688 projectionSnapshotQuery: {
5689 getShellSnapshot: () =>
5690 Effect.gen(function* () {
5691 yield* Effect.sleep("25 millis");
5692 yield* PubSub.publish(liveEvents, deletedEvent);
5693 return {
5694 snapshotSequence: 1,
5695 projects: [],
5696 threads: [makeDefaultOrchestrationThreadShell()],
5697 updatedAt: "2026-01-01T00:00:00.000Z",
5698 };
5699 }),
5700 },
5701 },
5702 });
5703
5704 const wsUrl = yield* getWsServerUrl("/ws");
5705 const items = yield* Effect.scoped(
5706 withWsRpcClient(wsUrl, (client) =>
5707 client[ORCHESTRATION_WS_METHODS.subscribeShell]({
5708 requestCompletionMarker: true,
5709 }).pipe(Stream.take(3), Stream.runCollect),
5710 ),
5711 ).pipe(Effect.timeout("2 seconds"));
5712
5713 assert.equal(items[0]?.kind, "snapshot");
5714 assert.equal(items[1]?.kind, "thread-removed");
5715 assert.deepEqual(items[2], { kind: "synchronized" });
5716 }).pipe(Effect.provide(NodeHttpServer.layerTest), TestClock.withLive),
5717 );
5718
5610 it.effect("buffers thread events published while the initial snapshot loads", () =>5719 it.effect("buffers thread events published while the initial snapshot loads", () =>
5611 Effect.gen(function* () {5720 Effect.gen(function* () {
5612 const thread = makeDefaultOrchestrationReadModel().threads[0]!;5721 const thread = makeDefaultOrchestrationReadModel().threads[0]!;
apps/server/src/ws.ts+50-4
...@@ -936,6 +936,8 @@ const makeWsRpcLayer = (...@@ -936,6 +936,8 @@ const makeWsRpcLayer = (
936 otlpMetricsEnabled: config.otlpMetricsUrl !== undefined,936 otlpMetricsEnabled: config.otlpMetricsUrl !== undefined,
937 },937 },
938 settings,938 settings,
939 shellResumeCompletionMarker: true,
940 threadResumeCompletionMarker: true,
939 };941 };
940 });942 });
941943
...@@ -1105,11 +1107,28 @@ const makeWsRpcLayer = (...@@ -1105,11 +1107,28 @@ const makeWsRpcLayer = (
1105 }),1107 }),
1106 ),1108 ),
1107 );1109 );
1108 return Stream.concat(catchUpStream, Stream.fromQueue(liveBuffer));1110 const afterCatchUp =
1111 input.requestCompletionMarker === true
1112 ? Stream.concat(
1113 Stream.fromEffect(
1114 Queue.offer(liveBuffer, { kind: "synchronized" as const }),
1115 ).pipe(Stream.drain),
1116 Stream.fromQueue(liveBuffer),
1117 )
1118 : Stream.fromQueue(liveBuffer);
1119 return Stream.concat(catchUpStream, afterCatchUp);
1109 }),1120 }),
1110 );1121 );
1111 }1122 }
11121123
1124 // The full-snapshot fallback needs the same replay-window safety
1125 // as the resume path: subscribe before loading the projection so
1126 // events published while the snapshot is read are buffered.
1127 const liveBuffer = yield* Queue.unbounded<OrchestrationShellStreamItem>();
1128 yield* Effect.forkScoped(
1129 liveStream.pipe(Stream.runForEach((item) => Queue.offer(liveBuffer, item))),
1130 );
1131 const bufferedLiveStream = Stream.fromQueue(liveBuffer);
1113 const snapshot = yield* projectionSnapshotQuery.getShellSnapshot().pipe(1132 const snapshot = yield* projectionSnapshotQuery.getShellSnapshot().pipe(
1114 Effect.tapError((cause) =>1133 Effect.tapError((cause) =>
1115 Effect.logError("orchestration shell snapshot load failed", { cause }),1134 Effect.logError("orchestration shell snapshot load failed", { cause }),
...@@ -1123,12 +1142,21 @@ const makeWsRpcLayer = (...@@ -1123,12 +1142,21 @@ const makeWsRpcLayer = (
1123 ),1142 ),
1124 );1143 );
11251144
1145 const afterSnapshot =
1146 input.requestCompletionMarker === true
1147 ? Stream.concat(
1148 Stream.fromEffect(
1149 Queue.offer(liveBuffer, { kind: "synchronized" as const }),
1150 ).pipe(Stream.drain),
1151 bufferedLiveStream,
1152 )
1153 : bufferedLiveStream;
1126 return Stream.concat(1154 return Stream.concat(
1127 Stream.make({1155 Stream.make({
1128 kind: "snapshot" as const,1156 kind: "snapshot" as const,
1129 snapshot,1157 snapshot,
1130 }),1158 }),
1131 liveStream,1159 afterSnapshot,
1132 );1160 );
1133 }),1161 }),
1134 { "rpc.aggregate": "orchestration" },1162 { "rpc.aggregate": "orchestration" },
...@@ -1207,7 +1235,16 @@ const makeWsRpcLayer = (...@@ -1207,7 +1235,16 @@ const makeWsRpcLayer = (
1207 }),1235 }),
1208 ),1236 ),
1209 );1237 );
1210 return Stream.concat(catchUpStream, bufferedLiveStream);1238 const afterCatchUp =
1239 input.requestCompletionMarker === true
1240 ? Stream.concat(
1241 Stream.fromEffect(
1242 Queue.offer(liveBuffer, { kind: "synchronized" as const }),
1243 ).pipe(Stream.drain),
1244 bufferedLiveStream,
1245 )
1246 : bufferedLiveStream;
1247 return Stream.concat(catchUpStream, afterCatchUp);
1211 }1248 }
12121249
1213 const snapshot = yield* projectionSnapshotQuery1250 const snapshot = yield* projectionSnapshotQuery
...@@ -1229,12 +1266,21 @@ const makeWsRpcLayer = (...@@ -1229,12 +1266,21 @@ const makeWsRpcLayer = (
1229 });1266 });
1230 }1267 }
12311268
1269 const afterSnapshot =
1270 input.requestCompletionMarker === true
1271 ? Stream.concat(
1272 Stream.fromEffect(
1273 Queue.offer(liveBuffer, { kind: "synchronized" as const }),
1274 ).pipe(Stream.drain),
1275 bufferedLiveStream,
1276 )
1277 : bufferedLiveStream;
1232 return Stream.concat(1278 return Stream.concat(
1233 Stream.make({1279 Stream.make({
1234 kind: "snapshot" as const,1280 kind: "snapshot" as const,
1235 snapshot: snapshot.value,1281 snapshot: snapshot.value,
1236 }),1282 }),
1237 bufferedLiveStream,1283 afterSnapshot,
1238 );1284 );
1239 }),1285 }),
1240 { "rpc.aggregate": "orchestration" },1286 { "rpc.aggregate": "orchestration" },
packages/client-runtime/src/rpc/client.ts+88-52
...@@ -11,6 +11,7 @@ import { RpcClientError } from "effect/unstable/rpc";...@@ -11,6 +11,7 @@ import { RpcClientError } from "effect/unstable/rpc";
1111
12import { EnvironmentSupervisor } from "../connection/supervisor.ts";12import { EnvironmentSupervisor } from "../connection/supervisor.ts";
13import type { WsRpcProtocolClient } from "../rpc/protocol.ts";13import type { WsRpcProtocolClient } from "../rpc/protocol.ts";
14import type { RpcSession } from "../rpc/session.ts";
1415
15export class EnvironmentRpcUnavailableError extends Schema.TaggedErrorClass<EnvironmentRpcUnavailableError>()(16export class EnvironmentRpcUnavailableError extends Schema.TaggedErrorClass<EnvironmentRpcUnavailableError>()(
16 "EnvironmentRpcUnavailableError",17 "EnvironmentRpcUnavailableError",
...@@ -147,15 +148,18 @@ export function runStream<TTag extends EnvironmentStreamCommandRpcTag>(...@@ -147,15 +148,18 @@ export function runStream<TTag extends EnvironmentStreamCommandRpcTag>(
147 );148 );
148}149}
149150
150export function subscribe<TTag extends EnvironmentSubscriptionRpcTag>(151interface SubscriptionOptions<TTag extends EnvironmentSubscriptionRpcTag> {
152 readonly onExpectedFailure?: (
153 cause: Cause.Cause<EnvironmentRpcStreamFailure<TTag>>,
154 ) => Effect.Effect<void, never, never>;
155 readonly retryExpectedFailureAfter?: Duration.Input;
156 readonly resubscribe?: Stream.Stream<unknown, never, never>;
157}
158
159export function subscribeDynamic<TTag extends EnvironmentSubscriptionRpcTag>(
151 tag: TTag,160 tag: TTag,
152 input: EnvironmentRpcInput<TTag>,161 makeInput: (session: RpcSession) => Effect.Effect<EnvironmentRpcInput<TTag>>,
153 options?: {162 options?: SubscriptionOptions<TTag>,
154 readonly onExpectedFailure?: (
155 cause: Cause.Cause<EnvironmentRpcStreamFailure<TTag>>,
156 ) => Effect.Effect<void, never, never>;
157 readonly retryExpectedFailureAfter?: Duration.Input;
158 },
159): Stream.Stream<163): Stream.Stream<
160 EnvironmentRpcStreamValue<TTag>,164 EnvironmentRpcStreamValue<TTag>,
161 EnvironmentRpcStreamFailure<TTag>,165 EnvironmentRpcStreamFailure<TTag>,
...@@ -163,8 +167,18 @@ export function subscribe<TTag extends EnvironmentSubscriptionRpcTag>(...@@ -163,8 +167,18 @@ export function subscribe<TTag extends EnvironmentSubscriptionRpcTag>(
163> {167> {
164 return Stream.unwrap(168 return Stream.unwrap(
165 EnvironmentSupervisor.pipe(169 EnvironmentSupervisor.pipe(
166 Effect.map((supervisor) =>170 Effect.map((supervisor) => {
167 SubscriptionRef.changes(supervisor.session).pipe(171 const sessionChanges = SubscriptionRef.changes(supervisor.session);
172 const sessions =
173 options?.resubscribe === undefined
174 ? sessionChanges
175 : Stream.merge(
176 sessionChanges,
177 options.resubscribe.pipe(
178 Stream.mapEffect(() => SubscriptionRef.get(supervisor.session)),
179 ),
180 );
181 return sessions.pipe(
168 Stream.switchMap(182 Stream.switchMap(
169 Option.match({183 Option.match({
170 onNone: () => Stream.empty,184 onNone: () => Stream.empty,
...@@ -180,54 +194,64 @@ export function subscribe<TTag extends EnvironmentSubscriptionRpcTag>(...@@ -180,54 +194,64 @@ export function subscribe<TTag extends EnvironmentSubscriptionRpcTag>(
180 EnvironmentRpcStreamFailure<TTag>194 EnvironmentRpcStreamFailure<TTag>
181 > =>195 > =>
182 Stream.suspend(() =>196 Stream.suspend(() =>
183 method(input).pipe(197 Stream.unwrap(
184 Stream.catchCause((cause) => {198 makeInput(session).pipe(
185 const hasOnlyExpectedFailures =199 Effect.map((input) =>
186 cause.reasons.length > 0 &&200 method(input).pipe(
187 cause.reasons.every((reason) => reason._tag === "Fail");201 Stream.catchCause((cause) => {
188 const isTransportFailure =202 const hasOnlyExpectedFailures =
189 hasOnlyExpectedFailures &&203 cause.reasons.length > 0 &&
190 cause.reasons.every(204 cause.reasons.every((reason) => reason._tag === "Fail");
191 (reason) => reason._tag === "Fail" && isRpcClientError(reason.error),205 const isTransportFailure =
192 );206 hasOnlyExpectedFailures &&
193 if (isTransportFailure) {207 cause.reasons.every(
194 return Stream.fromEffect(208 (reason) =>
195 Effect.logWarning(209 reason._tag === "Fail" && isRpcClientError(reason.error),
196 "Durable RPC subscription lost its transport; waiting for the next session.",210 );
197 {211 if (isTransportFailure) {
198 cause: Cause.pretty(cause),212 return Stream.fromEffect(
199 method: tag,213 Effect.logWarning(
200 environmentId: supervisor.target.environmentId,214 "Durable RPC subscription lost its transport; waiting for the next session.",
201 },215 {
202 ),216 cause: Cause.pretty(cause),
203 ).pipe(Stream.drain);217 method: tag,
204 }218 environmentId: supervisor.target.environmentId,
205 if (hasOnlyExpectedFailures && options?.onExpectedFailure !== undefined) {219 },
206 const handled = Stream.fromEffect(options.onExpectedFailure(cause)).pipe(220 ),
207 Stream.drain,221 ).pipe(Stream.drain);
208 );222 }
209 if (options.retryExpectedFailureAfter === undefined) {223 if (
210 return handled;224 hasOnlyExpectedFailures &&
211 }225 options?.onExpectedFailure !== undefined
212 return handled.pipe(226 ) {
213 Stream.concat(227 const handled = Stream.fromEffect(
214 Stream.fromEffect(228 options.onExpectedFailure(cause),
215 Effect.sleep(options.retryExpectedFailureAfter),229 ).pipe(Stream.drain);
216 ).pipe(Stream.drain),230 if (options.retryExpectedFailureAfter === undefined) {
217 ),231 return handled;
218 Stream.concat(subscribeToSession()),232 }
219 );233 return handled.pipe(
220 }234 Stream.concat(
221 return Stream.failCause(cause);235 Stream.fromEffect(
222 }),236 Effect.sleep(options.retryExpectedFailureAfter),
237 ).pipe(Stream.drain),
238 ),
239 Stream.concat(subscribeToSession()),
240 );
241 }
242 return Stream.failCause(cause);
243 }),
244 ),
245 ),
246 ),
223 ),247 ),
224 );248 );
225 return subscribeToSession();249 return subscribeToSession();
226 },250 },
227 }),251 }),
228 ),252 ),
229 ),253 );
230 ),254 }),
231 ),255 ),
232 ).pipe(256 ).pipe(
233 Stream.withSpan("EnvironmentRpc.subscribe", {257 Stream.withSpan("EnvironmentRpc.subscribe", {
...@@ -236,6 +260,18 @@ export function subscribe<TTag extends EnvironmentSubscriptionRpcTag>(...@@ -236,6 +260,18 @@ export function subscribe<TTag extends EnvironmentSubscriptionRpcTag>(
236 );260 );
237}261}
238262
263export function subscribe<TTag extends EnvironmentSubscriptionRpcTag>(
264 tag: TTag,
265 input: EnvironmentRpcInput<TTag>,
266 options?: SubscriptionOptions<TTag>,
267): Stream.Stream<
268 EnvironmentRpcStreamValue<TTag>,
269 EnvironmentRpcStreamFailure<TTag>,
270 EnvironmentSupervisor
271> {
272 return subscribeDynamic(tag, () => Effect.succeed(input), options);
273}
274
239export const config = Effect.gen(function* () {275export const config = Effect.gen(function* () {
240 const session = yield* currentSession();276 const session = yield* currentSession();
241 return yield* session.initialConfig;277 return yield* session.initialConfig;
packages/client-runtime/src/state/shell-sync.test.ts+133-9
...@@ -8,6 +8,7 @@ import { describe, expect, it } from "@effect/vitest";...@@ -8,6 +8,7 @@ import { describe, expect, it } from "@effect/vitest";
8import * as Effect from "effect/Effect";8import * as Effect from "effect/Effect";
9import * as Option from "effect/Option";9import * as Option from "effect/Option";
10import * as Queue from "effect/Queue";10import * as Queue from "effect/Queue";
11import * as Ref from "effect/Ref";
11import * as Stream from "effect/Stream";12import * as Stream from "effect/Stream";
12import * as SubscriptionRef from "effect/SubscriptionRef";13import * as SubscriptionRef from "effect/SubscriptionRef";
1314
...@@ -17,6 +18,7 @@ import {...@@ -17,6 +18,7 @@ import {
17 type PreparedConnection,18 type PreparedConnection,
18} from "../connection/model.ts";19} from "../connection/model.ts";
19import * as EnvironmentSupervisor from "../connection/supervisor.ts";20import * as EnvironmentSupervisor from "../connection/supervisor.ts";
21import * as ConnectionWakeups from "../connection/wakeups.ts";
20import * as Persistence from "../platform/persistence.ts";22import * as Persistence from "../platform/persistence.ts";
21import * as RpcSession from "../rpc/session.ts";23import * as RpcSession from "../rpc/session.ts";
22import type { WsRpcProtocolClient } from "../rpc/protocol.ts";24import type { WsRpcProtocolClient } from "../rpc/protocol.ts";
...@@ -48,7 +50,7 @@ const LIVE_SHELL_SNAPSHOT: OrchestrationShellSnapshot = {...@@ -48,7 +50,7 @@ const LIVE_SHELL_SNAPSHOT: OrchestrationShellSnapshot = {
48function session(client: WsRpcProtocolClient): RpcSession.RpcSession {50function session(client: WsRpcProtocolClient): RpcSession.RpcSession {
49 return {51 return {
50 client,52 client,
51 initialConfig: Effect.never,53 initialConfig: Effect.succeed({ shellResumeCompletionMarker: true } as never),
52 ready: Effect.void,54 ready: Effect.void,
53 probe: Effect.void,55 probe: Effect.void,
54 closed: Effect.never,56 closed: Effect.never,
...@@ -112,6 +114,15 @@ describe("environment shell synchronization", () => {...@@ -112,6 +114,15 @@ describe("environment shell synchronization", () => {
112 kind: "snapshot",114 kind: "snapshot",
113 snapshot: LIVE_SHELL_SNAPSHOT,115 snapshot: LIVE_SHELL_SNAPSHOT,
114 });116 });
117 const synchronizing = yield* SubscriptionRef.changes(shellState).pipe(
118 Stream.filter((state) => state.status === "synchronizing" && Option.isSome(state.snapshot)),
119 Stream.runHead,
120 );
121 expect(Option.getOrThrow(Option.getOrThrow(synchronizing).snapshot)).toEqual(
122 LIVE_SHELL_SNAPSHOT,
123 );
124
125 yield* Queue.offer(events, { kind: "synchronized" });
115 yield* SubscriptionRef.changes(shellState).pipe(126 yield* SubscriptionRef.changes(shellState).pipe(
116 Stream.filter((state) => state.status === "live"),127 Stream.filter((state) => state.status === "live"),
117 Stream.runHead,128 Stream.runHead,
...@@ -137,21 +148,32 @@ describe("environment shell synchronization", () => {...@@ -137,21 +148,32 @@ describe("environment shell synchronization", () => {
137 }),148 }),
138 );149 );
139150
140 it.effect("resumes a warm shell cache via afterSequence without an HTTP fetch", () =>151 it.effect("replaces a warm shell cache with an authoritative HTTP snapshot", () =>
141 Effect.gen(function* () {152 Effect.gen(function* () {
142 const cachedSnapshot: OrchestrationShellSnapshot = {153 const cachedSnapshot: OrchestrationShellSnapshot = {
143 snapshotSequence: 5,154 snapshotSequence: 5,
144 projects: [],155 projects: [],
145 threads: [],156 threads: [{ id: "stale-thread" } as never],
146 updatedAt: "2026-06-06T00:00:00.000Z",157 updatedAt: "2026-06-06T00:00:00.000Z",
147 };158 };
159 const httpSnapshot: OrchestrationShellSnapshot = {
160 ...cachedSnapshot,
161 snapshotSequence: 9,
162 threads: [],
163 updatedAt: "2026-06-07T00:00:00.000Z",
164 };
148 const events = yield* Queue.unbounded<OrchestrationShellStreamItem>();165 const events = yield* Queue.unbounded<OrchestrationShellStreamItem>();
149 const capturedAfterSequence = yield* SubscriptionRef.make<number | undefined>(undefined);166 const capturedAfterSequence = yield* SubscriptionRef.make<number | undefined>(undefined);
167 const capturedCompletionMarker = yield* Ref.make<boolean | undefined>(undefined);
150 const loaderCalls = yield* SubscriptionRef.make(0);168 const loaderCalls = yield* SubscriptionRef.make(0);
151 const client = {169 const client = {
152 [ORCHESTRATION_WS_METHODS.subscribeShell]: (input: { readonly afterSequence?: number }) =>170 [ORCHESTRATION_WS_METHODS.subscribeShell]: (input: {
171 readonly afterSequence?: number;
172 readonly requestCompletionMarker?: boolean;
173 }) =>
153 Stream.unwrap(174 Stream.unwrap(
154 SubscriptionRef.set(capturedAfterSequence, input.afterSequence).pipe(175 Ref.set(capturedCompletionMarker, input.requestCompletionMarker).pipe(
176 Effect.andThen(SubscriptionRef.set(capturedAfterSequence, input.afterSequence)),
155 Effect.as(Stream.fromQueue(events)),177 Effect.as(Stream.fromQueue(events)),
156 ),178 ),
157 ),179 ),
...@@ -183,9 +205,11 @@ describe("environment shell synchronization", () => {...@@ -183,9 +205,11 @@ describe("environment shell synchronization", () => {
183 });205 });
184 const snapshotLoader = ShellSnapshotLoader.of({206 const snapshotLoader = ShellSnapshotLoader.of({
185 load: () =>207 load: () =>
186 SubscriptionRef.update(loaderCalls, (count) => count + 1).pipe(Effect.as(Option.none())),208 SubscriptionRef.update(loaderCalls, (count) => count + 1).pipe(
209 Effect.as(Option.some(httpSnapshot)),
210 ),
187 });211 });
188 yield* makeEnvironmentShellState().pipe(212 const shellState = yield* makeEnvironmentShellState().pipe(
189 Effect.provideService(EnvironmentSupervisor.EnvironmentSupervisor, supervisor),213 Effect.provideService(EnvironmentSupervisor.EnvironmentSupervisor, supervisor),
190 Effect.provideService(Persistence.EnvironmentCacheStore, cache),214 Effect.provideService(Persistence.EnvironmentCacheStore, cache),
191 Effect.provideService(ShellSnapshotLoader, snapshotLoader),215 Effect.provideService(ShellSnapshotLoader, snapshotLoader),
...@@ -197,8 +221,108 @@ describe("environment shell synchronization", () => {...@@ -197,8 +221,108 @@ describe("environment shell synchronization", () => {
197 Stream.runHead,221 Stream.runHead,
198 );222 );
199223
200 expect(yield* SubscriptionRef.get(capturedAfterSequence)).toBe(5);224 expect(yield* SubscriptionRef.get(capturedAfterSequence)).toBe(9);
201 expect(yield* SubscriptionRef.get(loaderCalls)).toBe(0);225 expect(yield* Ref.get(capturedCompletionMarker)).toBe(true);
226 expect(yield* SubscriptionRef.get(loaderCalls)).toBe(1);
227 const synchronizing = yield* SubscriptionRef.get(shellState);
228 expect(synchronizing.status).toBe("synchronizing");
229 expect(Option.getOrThrow(synchronizing.snapshot)).toEqual(httpSnapshot);
230
231 yield* Queue.offer(events, { kind: "synchronized" });
232 yield* SubscriptionRef.changes(shellState).pipe(
233 Stream.filter((value) => value.status === "live"),
234 Stream.runHead,
235 );
236 }),
237 );
238
239 it.effect("refreshes the authoritative shell snapshot when the app becomes active", () =>
240 Effect.gen(function* () {
241 const events = yield* Queue.unbounded<OrchestrationShellStreamItem>();
242 const wakeups = yield* Queue.unbounded<ConnectionWakeups.ConnectionWakeup>();
243 const loaderCalls = yield* Ref.make(0);
244 const subscriptionCount = yield* Ref.make(0);
245 const client = {
246 [ORCHESTRATION_WS_METHODS.subscribeShell]: () =>
247 Stream.unwrap(
248 Ref.update(subscriptionCount, (count) => count + 1).pipe(
249 Effect.as(Stream.fromQueue(events)),
250 ),
251 ),
252 } as unknown as WsRpcProtocolClient;
253 const supervisorState = yield* SubscriptionRef.make(AVAILABLE_CONNECTION_STATE);
254 const supervisor = EnvironmentSupervisor.EnvironmentSupervisor.of({
255 target: TARGET,
256 state: supervisorState,
257 session: yield* SubscriptionRef.make(Option.some(session(client))),
258 prepared: yield* SubscriptionRef.make(Option.some(PREPARED)),
259 connect: Effect.void,
260 disconnect: Effect.void,
261 retryNow: Effect.void,
262 } satisfies EnvironmentSupervisor.EnvironmentSupervisor["Service"]);
263 const cache = Persistence.EnvironmentCacheStore.of({
264 loadShell: () => Effect.succeed(Option.some(LIVE_SHELL_SNAPSHOT)),
265 saveShell: () => Effect.void,
266 loadThread: () => Effect.succeed(Option.none()),
267 saveThread: () => Effect.void,
268 removeThread: () => Effect.void,
269 loadServerConfig: () => Effect.succeed(Option.none()),
270 saveServerConfig: () => Effect.void,
271 loadVcsRefs: () => Effect.succeed(Option.none()),
272 saveVcsRefs: () => Effect.void,
273 clear: () => Effect.void,
274 });
275 const snapshotLoader = ShellSnapshotLoader.of({
276 load: () =>
277 Ref.updateAndGet(loaderCalls, (count) => count + 1).pipe(
278 Effect.map((count) =>
279 Option.some({ ...LIVE_SHELL_SNAPSHOT, snapshotSequence: count * 10 }),
280 ),
281 ),
282 });
283 const shellState = yield* makeEnvironmentShellState().pipe(
284 Effect.provideService(EnvironmentSupervisor.EnvironmentSupervisor, supervisor),
285 Effect.provideService(Persistence.EnvironmentCacheStore, cache),
286 Effect.provideService(ShellSnapshotLoader, snapshotLoader),
287 Effect.provideService(
288 ConnectionWakeups.ConnectionWakeups,
289 ConnectionWakeups.ConnectionWakeups.of({ changes: Stream.fromQueue(wakeups) }),
290 ),
291 );
292
293 yield* SubscriptionRef.changes(shellState).pipe(
294 Stream.filter(
295 (value) =>
296 value.status === "synchronizing" &&
297 Option.isSome(value.snapshot) &&
298 value.snapshot.value.snapshotSequence === 10,
299 ),
300 Stream.runHead,
301 );
302 yield* Queue.offer(events, { kind: "synchronized" });
303 yield* SubscriptionRef.changes(shellState).pipe(
304 Stream.filter((value) => value.status === "live"),
305 Stream.runHead,
306 );
307
308 yield* Queue.offer(wakeups, "application-active");
309 yield* SubscriptionRef.changes(shellState).pipe(
310 Stream.filter(
311 (value) =>
312 value.status === "synchronizing" &&
313 Option.isSome(value.snapshot) &&
314 value.snapshot.value.snapshotSequence === 20,
315 ),
316 Stream.runHead,
317 );
318
319 for (let attempt = 0; attempt < 100; attempt += 1) {
320 if ((yield* Ref.get(subscriptionCount)) >= 2) break;
321 yield* Effect.yieldNow;
322 }
323
324 expect(yield* Ref.get(loaderCalls)).toBe(2);
325 expect(yield* Ref.get(subscriptionCount)).toBe(2);
202 }),326 }),
203 );327 );
204});328});
packages/client-runtime/src/state/shell.ts+72-41
...@@ -9,6 +9,7 @@ import * as Cause from "effect/Cause";...@@ -9,6 +9,7 @@ import * as Cause from "effect/Cause";
9import * as Effect from "effect/Effect";9import * as Effect from "effect/Effect";
10import * as Option from "effect/Option";10import * as Option from "effect/Option";
11import * as Queue from "effect/Queue";11import * as Queue from "effect/Queue";
12import * as Ref from "effect/Ref";
12import * as Stream from "effect/Stream";13import * as Stream from "effect/Stream";
13import * as SubscriptionRef from "effect/SubscriptionRef";14import * as SubscriptionRef from "effect/SubscriptionRef";
14import { AsyncResult, Atom } from "effect/unstable/reactivity";15import { AsyncResult, Atom } from "effect/unstable/reactivity";
...@@ -16,9 +17,10 @@ import { AsyncResult, Atom } from "effect/unstable/reactivity";...@@ -16,9 +17,10 @@ import { AsyncResult, Atom } from "effect/unstable/reactivity";
16import { EnvironmentRegistry } from "../connection/registry.ts";17import { EnvironmentRegistry } from "../connection/registry.ts";
17import { connectionProjectionPhase } from "../connection/model.ts";18import { connectionProjectionPhase } from "../connection/model.ts";
18import { EnvironmentSupervisor } from "../connection/supervisor.ts";19import { EnvironmentSupervisor } from "../connection/supervisor.ts";
20import * as ConnectionWakeups from "../connection/wakeups.ts";
19import { safeErrorLogAttributes } from "../errors/safeLog.ts";21import { safeErrorLogAttributes } from "../errors/safeLog.ts";
20import { EnvironmentCacheStore } from "../platform/persistence.ts";22import { EnvironmentCacheStore } from "../platform/persistence.ts";
21import { subscribe } from "../rpc/client.ts";23import { subscribeDynamic } from "../rpc/client.ts";
22import { ShellSnapshotLoader } from "./shellSnapshotHttp.ts";24import { ShellSnapshotLoader } from "./shellSnapshotHttp.ts";
23import { applyShellStreamEvent } from "./shellReducer.ts";25import { applyShellStreamEvent } from "./shellReducer.ts";
24import type { EnvironmentCatalogState } from "./connections.ts";26import type { EnvironmentCatalogState } from "./connections.ts";
...@@ -50,6 +52,7 @@ export const makeEnvironmentShellState = Effect.fn("EnvironmentShellState.make")...@@ -50,6 +52,7 @@ export const makeEnvironmentShellState = Effect.fn("EnvironmentShellState.make")
50 const supervisor = yield* EnvironmentSupervisor;52 const supervisor = yield* EnvironmentSupervisor;
51 const cache = yield* EnvironmentCacheStore;53 const cache = yield* EnvironmentCacheStore;
52 const snapshotLoader = yield* ShellSnapshotLoader;54 const snapshotLoader = yield* ShellSnapshotLoader;
55 const wakeups = yield* Effect.serviceOption(ConnectionWakeups.ConnectionWakeups);
53 const environmentId = supervisor.target.environmentId;56 const environmentId = supervisor.target.environmentId;
54 const cachedSnapshot = yield* cache.loadShell(environmentId).pipe(57 const cachedSnapshot = yield* cache.loadShell(environmentId).pipe(
55 Effect.catch((error) =>58 Effect.catch((error) =>
...@@ -67,6 +70,7 @@ export const makeEnvironmentShellState = Effect.fn("EnvironmentShellState.make")...@@ -67,6 +70,7 @@ export const makeEnvironmentShellState = Effect.fn("EnvironmentShellState.make")
67 status: shellStatusForSnapshot(cachedSnapshot),70 status: shellStatusForSnapshot(cachedSnapshot),
68 error: Option.none(),71 error: Option.none(),
69 });72 });
73 const awaitingCompletion = yield* Ref.make(false);
70 const persistence = yield* Queue.sliding<OrchestrationShellSnapshot>(1);74 const persistence = yield* Queue.sliding<OrchestrationShellSnapshot>(1);
7175
72 const persist = Effect.fn("EnvironmentShellState.persist")(function* (76 const persist = Effect.fn("EnvironmentShellState.persist")(function* (
...@@ -90,10 +94,14 @@ export const makeEnvironmentShellState = Effect.fn("EnvironmentShellState.make")...@@ -90,10 +94,14 @@ export const makeEnvironmentShellState = Effect.fn("EnvironmentShellState.make")
90 Effect.forkScoped,94 Effect.forkScoped,
91 );95 );
9296
93 const setDisconnected = SubscriptionRef.update(state, (current) => ({97 const setDisconnected = Ref.set(awaitingCompletion, false).pipe(
94 ...current,98 Effect.andThen(
95 status: shellStatusForSnapshot(current.snapshot),99 SubscriptionRef.update(state, (current) => ({
96 }));100 ...current,
101 status: shellStatusForSnapshot(current.snapshot),
102 })),
103 ),
104 );
97 const setSynchronizing = SubscriptionRef.update(state, (current) => ({105 const setSynchronizing = SubscriptionRef.update(state, (current) => ({
98 ...current,106 ...current,
99 status: "synchronizing" as const,107 status: "synchronizing" as const,
...@@ -109,7 +117,8 @@ export const makeEnvironmentShellState = Effect.fn("EnvironmentShellState.make")...@@ -109,7 +117,8 @@ export const makeEnvironmentShellState = Effect.fn("EnvironmentShellState.make")
109 },117 },
110 );118 );
111 const setStreamError = (error: unknown) =>119 const setStreamError = (error: unknown) =>
112 Effect.logWarning("Could not synchronize the environment shell.").pipe(120 Ref.set(awaitingCompletion, false).pipe(
121 Effect.andThen(Effect.logWarning("Could not synchronize the environment shell.")),
113 Effect.annotateLogs({122 Effect.annotateLogs({
114 environmentId,123 environmentId,
115 ...safeErrorLogAttributes(error),124 ...safeErrorLogAttributes(error),
...@@ -126,6 +135,16 @@ export const makeEnvironmentShellState = Effect.fn("EnvironmentShellState.make")...@@ -126,6 +135,16 @@ export const makeEnvironmentShellState = Effect.fn("EnvironmentShellState.make")
126 const applyItem = Effect.fn("EnvironmentShellState.applyItem")(function* (135 const applyItem = Effect.fn("EnvironmentShellState.applyItem")(function* (
127 item: OrchestrationShellStreamItem,136 item: OrchestrationShellStreamItem,
128 ) {137 ) {
138 if (item.kind === "synchronized") {
139 yield* Ref.set(awaitingCompletion, false);
140 yield* SubscriptionRef.update(state, (current) =>
141 Option.isSome(current.snapshot)
142 ? { ...current, status: "live" as const, error: Option.none() }
143 : current,
144 );
145 return;
146 }
147
129 const current = yield* SubscriptionRef.get(state);148 const current = yield* SubscriptionRef.get(state);
130 const nextSnapshot =149 const nextSnapshot =
131 item.kind === "snapshot"150 item.kind === "snapshot"
...@@ -141,52 +160,64 @@ export const makeEnvironmentShellState = Effect.fn("EnvironmentShellState.make")...@@ -141,52 +160,64 @@ export const makeEnvironmentShellState = Effect.fn("EnvironmentShellState.make")
141 return;160 return;
142 }161 }
143162
163 const waiting = yield* Ref.get(awaitingCompletion);
144 yield* SubscriptionRef.set(state, {164 yield* SubscriptionRef.set(state, {
145 snapshot: Option.some(nextSnapshot),165 snapshot: Option.some(nextSnapshot),
146 status: "live",166 status: waiting ? "synchronizing" : "live",
147 error: Option.none(),167 error: Option.none(),
148 });168 });
149 yield* Queue.offer(persistence, nextSnapshot);169 yield* Queue.offer(persistence, nextSnapshot);
150 });170 });
151171
152 yield* Effect.forkScoped(172 const foregroundResubscriptions = Option.match(wakeups, {
153 Effect.gen(function* () {173 onNone: () => Stream.never,
154 // Establish the base shell snapshot to resume from, minimizing bytes over174 onSome: (service) =>
155 // the wire:175 service.changes.pipe(Stream.filter((reason) => reason === "application-active")),
156 // - Warm cache: reuse the cached snapshot (zero network) and resume via176 });
157 // `afterSequence` so we only receive shell events since the cached
158 // sequence.
159 // - Cold cache: load the full shell snapshot over HTTP (gzip-compressible,
160 // and off the socket), then resume via `afterSequence`.
161 // If no base can be established we fall back to the socket-embedded
162 // snapshot so the shell still synchronizes. Overlapping/replayed events are
163 // deduped by sequence in applyItem.
164 const base = Option.isSome(cachedSnapshot)
165 ? cachedSnapshot
166 : yield* Effect.gen(function* () {
167 const prepared = yield* SubscriptionRef.changes(supervisor.prepared).pipe(
168 Stream.filter(Option.isSome),
169 Stream.map((current) => current.value),
170 Stream.runHead,
171 );
172 return Option.isSome(prepared)
173 ? yield* snapshotLoader.load(prepared.value)
174 : Option.none<OrchestrationShellSnapshot>();
175 });
176177
177 if (Option.isSome(base)) {178 yield* setSynchronizing;
178 yield* applyItem({ kind: "snapshot", snapshot: base.value });179 yield* Effect.forkScoped(
179 }180 subscribeDynamic(
181 ORCHESTRATION_WS_METHODS.subscribeShell,
182 Effect.fn("EnvironmentShellState.makeSubscribeInput")(function* (session) {
183 const supportsCompletionMarker = yield* session.initialConfig.pipe(
184 Effect.map((config) => config.shellResumeCompletionMarker === true),
185 Effect.orElseSucceed(() => false),
186 );
187 yield* Ref.set(awaitingCompletion, supportsCompletionMarker);
188 yield* setSynchronizing;
180189
181 const subscribeInput = Option.match(base, {190 const prepared = yield* SubscriptionRef.get(supervisor.prepared).pipe(
182 onNone: () => ({}),191 Effect.flatMap(
183 onSome: (snapshot) => ({ afterSequence: snapshot.snapshotSequence }),192 Option.match({
184 });193 onSome: Effect.succeed,
194 onNone: () =>
195 SubscriptionRef.changes(supervisor.prepared).pipe(
196 Stream.filter(Option.isSome),
197 Stream.map((value) => value.value),
198 Stream.runHead,
199 Effect.map(Option.getOrThrow),
200 ),
201 }),
202 ),
203 );
204 const httpSnapshot = yield* snapshotLoader.load(prepared);
205 if (Option.isSome(httpSnapshot)) {
206 yield* applyItem({ kind: "snapshot", snapshot: httpSnapshot.value });
207 return {
208 afterSequence: httpSnapshot.value.snapshotSequence,
209 ...(supportsCompletionMarker ? { requestCompletionMarker: true as const } : {}),
210 };
211 }
185212
186 yield* subscribe(ORCHESTRATION_WS_METHODS.subscribeShell, subscribeInput, {213 return supportsCompletionMarker ? { requestCompletionMarker: true as const } : {};
214 }),
215 {
187 onExpectedFailure: (cause) => setStreamError(Cause.squash(cause)),216 onExpectedFailure: (cause) => setStreamError(Cause.squash(cause)),
188 }).pipe(Stream.runForEach(applyItem));217 retryExpectedFailureAfter: "250 millis",
189 }),218 resubscribe: foregroundResubscriptions,
219 },
220 ).pipe(Stream.runForEach(applyItem)),
190 );221 );
191 yield* SubscriptionRef.changes(supervisor.state).pipe(222 yield* SubscriptionRef.changes(supervisor.state).pipe(
192 Stream.runForEach((connectionState) => {223 Stream.runForEach((connectionState) => {
packages/client-runtime/src/state/threads-sync.test.ts+170-5
...@@ -26,6 +26,7 @@ import {...@@ -26,6 +26,7 @@ import {
26 type PreparedConnection,26 type PreparedConnection,
27 type SupervisorConnectionState,27 type SupervisorConnectionState,
28} from "../connection/model.ts";28} from "../connection/model.ts";
29import * as ConnectionWakeups from "../connection/wakeups.ts";
29import * as EnvironmentSupervisor from "../connection/supervisor.ts";30import * as EnvironmentSupervisor from "../connection/supervisor.ts";
30import * as Persistence from "../platform/persistence.ts";31import * as Persistence from "../platform/persistence.ts";
31import * as RpcSession from "../rpc/session.ts";32import * as RpcSession from "../rpc/session.ts";
...@@ -98,10 +99,17 @@ const ACTIVE_THREAD: OrchestrationThread = {...@@ -98,10 +99,17 @@ const ACTIVE_THREAD: OrchestrationThread = {
9899
99type TestThreadInput = OrchestrationThreadStreamItem | Error;100type TestThreadInput = OrchestrationThreadStreamItem | Error;
100101
101function testSession(client: WsRpcProtocolClient): RpcSession.RpcSession {102function testSession(
103 client: WsRpcProtocolClient,
104 options?: { readonly completionMarker?: boolean },
105): RpcSession.RpcSession {
102 return {106 return {
103 client,107 client,
104 initialConfig: Effect.never,108 initialConfig: Effect.succeed(
109 options?.completionMarker === true
110 ? ({ threadResumeCompletionMarker: true } as never)
111 : ({} as never),
112 ),
105 ready: Effect.void,113 ready: Effect.void,
106 probe: Effect.void,114 probe: Effect.void,
107 closed: Effect.never,115 closed: Effect.never,
...@@ -122,6 +130,7 @@ function awaitThreadState(...@@ -122,6 +130,7 @@ function awaitThreadState(
122const makeHarness = Effect.fn("TestEnvironmentThreads.makeHarness")(function* (options?: {130const makeHarness = Effect.fn("TestEnvironmentThreads.makeHarness")(function* (options?: {
123 readonly cached?: OrchestrationThread;131 readonly cached?: OrchestrationThread;
124 readonly httpSnapshot?: Option.Option<OrchestrationThreadDetailSnapshot>;132 readonly httpSnapshot?: Option.Option<OrchestrationThreadDetailSnapshot>;
133 readonly completionMarker?: boolean;
125}) {134}) {
126 const inputs = yield* Queue.unbounded<TestThreadInput>();135 const inputs = yield* Queue.unbounded<TestThreadInput>();
127 const observed = yield* Queue.unbounded<EnvironmentThreadState>();136 const observed = yield* Queue.unbounded<EnvironmentThreadState>();
...@@ -130,8 +139,10 @@ const makeHarness = Effect.fn("TestEnvironmentThreads.makeHarness")(function* (o...@@ -130,8 +139,10 @@ const makeHarness = Effect.fn("TestEnvironmentThreads.makeHarness")(function* (o
130 const subscriptionCount = yield* Ref.make(0);139 const subscriptionCount = yield* Ref.make(0);
131 const loaderCalls = yield* Ref.make(0);140 const loaderCalls = yield* Ref.make(0);
132 const lastSubscribeAfterSequence = yield* Ref.make<number | undefined>(undefined);141 const lastSubscribeAfterSequence = yield* Ref.make<number | undefined>(undefined);
142 const lastRequestCompletionMarker = yield* Ref.make<boolean | undefined>(undefined);
133 const savedThreads = yield* Ref.make<ReadonlyArray<OrchestrationThreadDetailSnapshot>>([]);143 const savedThreads = yield* Ref.make<ReadonlyArray<OrchestrationThreadDetailSnapshot>>([]);
134 const removedThreads = yield* Ref.make<ReadonlyArray<ThreadId>>([]);144 const removedThreads = yield* Ref.make<ReadonlyArray<ThreadId>>([]);
145 const wakeups = yield* Queue.unbounded<ConnectionWakeups.ConnectionWakeup>();
135 const supervisorState = yield* SubscriptionRef.make<SupervisorConnectionState>(146 const supervisorState = yield* SubscriptionRef.make<SupervisorConnectionState>(
136 AVAILABLE_CONNECTION_STATE,147 AVAILABLE_CONNECTION_STATE,
137 );148 );
...@@ -142,16 +153,25 @@ const makeHarness = Effect.fn("TestEnvironmentThreads.makeHarness")(function* (o...@@ -142,16 +153,25 @@ const makeHarness = Effect.fn("TestEnvironmentThreads.makeHarness")(function* (o
142 ),153 ),
143 );154 );
144 const client = {155 const client = {
145 [ORCHESTRATION_WS_METHODS.subscribeThread]: (input: { readonly afterSequence?: number }) =>156 [ORCHESTRATION_WS_METHODS.subscribeThread]: (input: {
157 readonly afterSequence?: number;
158 readonly requestCompletionMarker?: boolean;
159 }) =>
146 Stream.unwrap(160 Stream.unwrap(
147 Ref.updateAndGet(subscriptionCount, (count) => count + 1).pipe(161 Ref.updateAndGet(subscriptionCount, (count) => count + 1).pipe(
148 Effect.andThen(Ref.set(lastSubscribeAfterSequence, input.afterSequence)),162 Effect.andThen(Ref.set(lastSubscribeAfterSequence, input.afterSequence)),
163 Effect.andThen(Ref.set(lastRequestCompletionMarker, input.requestCompletionMarker)),
149 Effect.as(streamFrom(inputs)),164 Effect.as(streamFrom(inputs)),
150 ),165 ),
151 ),166 ),
152 } as unknown as WsRpcProtocolClient;167 } as unknown as WsRpcProtocolClient;
153 const supervisorSession = yield* SubscriptionRef.make<Option.Option<RpcSession.RpcSession>>(168 const supervisorSession = yield* SubscriptionRef.make<Option.Option<RpcSession.RpcSession>>(
154 Option.some(testSession(client)),169 Option.some(
170 testSession(
171 client,
172 options?.completionMarker === true ? { completionMarker: true } : undefined,
173 ),
174 ),
155 );175 );
156 const prepared = yield* SubscriptionRef.make<Option.Option<PreparedConnection>>(176 const prepared = yield* SubscriptionRef.make<Option.Option<PreparedConnection>>(
157 Option.some(PREPARED),177 Option.some(PREPARED),
...@@ -201,6 +221,10 @@ const makeHarness = Effect.fn("TestEnvironmentThreads.makeHarness")(function* (o...@@ -201,6 +221,10 @@ const makeHarness = Effect.fn("TestEnvironmentThreads.makeHarness")(function* (o
201 Effect.provideService(EnvironmentSupervisor.EnvironmentSupervisor, supervisor),221 Effect.provideService(EnvironmentSupervisor.EnvironmentSupervisor, supervisor),
202 Effect.provideService(Persistence.EnvironmentCacheStore, cache),222 Effect.provideService(Persistence.EnvironmentCacheStore, cache),
203 Effect.provideService(ThreadSnapshotLoader, snapshotLoader),223 Effect.provideService(ThreadSnapshotLoader, snapshotLoader),
224 Effect.provideService(
225 ConnectionWakeups.ConnectionWakeups,
226 ConnectionWakeups.ConnectionWakeups.of({ changes: Stream.fromQueue(wakeups) }),
227 ),
204 );228 );
205 yield* SubscriptionRef.changes(threadState).pipe(229 yield* SubscriptionRef.changes(threadState).pipe(
206 Stream.runForEach((state) =>230 Stream.runForEach((state) =>
...@@ -217,11 +241,21 @@ const makeHarness = Effect.fn("TestEnvironmentThreads.makeHarness")(function* (o...@@ -217,11 +241,21 @@ const makeHarness = Effect.fn("TestEnvironmentThreads.makeHarness")(function* (o
217 subscriptionCount,241 subscriptionCount,
218 loaderCalls,242 loaderCalls,
219 lastSubscribeAfterSequence,243 lastSubscribeAfterSequence,
244 lastRequestCompletionMarker,
220 supervisorState,245 supervisorState,
221 supervisorSession,246 supervisorSession,
222 savedThreads,247 savedThreads,
223 removedThreads,248 removedThreads,
224 replaceSession: SubscriptionRef.set(supervisorSession, Option.some(testSession(client))),249 wakeups,
250 replaceSession: SubscriptionRef.set(
251 supervisorSession,
252 Option.some(
253 testSession(
254 client,
255 options?.completionMarker === true ? { completionMarker: true } : undefined,
256 ),
257 ),
258 ),
225 };259 };
226});260});
227261
...@@ -233,6 +267,8 @@ const snapshot = (thread: OrchestrationThread): OrchestrationThreadStreamItem =>...@@ -233,6 +267,8 @@ const snapshot = (thread: OrchestrationThread): OrchestrationThreadStreamItem =>
233 },267 },
234});268});
235269
270const synchronized = (): OrchestrationThreadStreamItem => ({ kind: "synchronized" });
271
236const titleUpdated = (title: string, sequence = 2): OrchestrationThreadStreamItem => ({272const titleUpdated = (title: string, sequence = 2): OrchestrationThreadStreamItem => ({
237 kind: "event",273 kind: "event",
238 event: {274 event: {
...@@ -415,6 +451,35 @@ describe("EnvironmentThreads", () => {...@@ -415,6 +451,35 @@ describe("EnvironmentThreads", () => {
415 }),451 }),
416 );452 );
417453
454 it.effect("does not resurrect a deleted thread when the app returns to the foreground", () =>
455 Effect.gen(function* () {
456 const harness = yield* makeHarness({
457 cached: BASE_THREAD,
458 completionMarker: true,
459 httpSnapshot: Option.some({
460 snapshotSequence: 4,
461 thread: { ...BASE_THREAD, title: "Stale HTTP thread" },
462 }),
463 });
464 yield* Queue.offer(harness.inputs, snapshot(BASE_THREAD));
465 yield* Queue.offer(harness.inputs, deleted());
466 yield* awaitThreadState(harness.observed, (value) => value.status === "deleted");
467
468 expect(yield* Ref.get(harness.loaderCalls)).toBe(0);
469 yield* Queue.offer(harness.wakeups, "application-active");
470 for (let attempt = 0; attempt < 100; attempt += 1) {
471 if ((yield* Ref.get(harness.subscriptionCount)) >= 2) break;
472 yield* Effect.yieldNow;
473 }
474
475 const latest = yield* Ref.get(harness.latest);
476 expect(yield* Ref.get(harness.subscriptionCount)).toBe(2);
477 expect(yield* Ref.get(harness.loaderCalls)).toBe(0);
478 expect(latest.status).toBe("deleted");
479 expect(Option.isNone(latest.data)).toBe(true);
480 }),
481 );
482
418 it.effect("preserves data after a domain failure and resumes on a replacement session", () =>483 it.effect("preserves data after a domain failure and resumes on a replacement session", () =>
419 Effect.gen(function* () {484 Effect.gen(function* () {
420 const harness = yield* makeHarness({ cached: BASE_THREAD });485 const harness = yield* makeHarness({ cached: BASE_THREAD });
...@@ -529,4 +594,104 @@ describe("EnvironmentThreads", () => {...@@ -529,4 +594,104 @@ describe("EnvironmentThreads", () => {
529 expect((yield* Ref.get(harness.latest)).status).toBe("live");594 expect((yield* Ref.get(harness.latest)).status).toBe("live");
530 }),595 }),
531 );596 );
597
598 it.effect("keeps replayed updates synchronizing until the completion marker arrives", () =>
599 Effect.gen(function* () {
600 const harness = yield* makeHarness({ cached: BASE_THREAD, completionMarker: true });
601 yield* awaitThreadState(
602 harness.observed,
603 (value) => value.status === "synchronizing" && Option.isSome(value.data),
604 );
605 expect(yield* Ref.get(harness.lastRequestCompletionMarker)).toBe(true);
606
607 yield* Queue.offer(
608 harness.inputs,
609 titleUpdated("Caught-up title", CACHED_SNAPSHOT_SEQUENCE + 1),
610 );
611 const catchingUp = yield* awaitThreadState(
612 harness.observed,
613 (value) =>
614 value.status === "synchronizing" &&
615 Option.isSome(value.data) &&
616 value.data.value.title === "Caught-up title",
617 );
618 expect(catchingUp.status).toBe("synchronizing");
619
620 yield* Queue.offer(harness.inputs, synchronized());
621 const live = yield* awaitThreadState(
622 harness.observed,
623 (value) => value.status === "live" && Option.isSome(value.data),
624 );
625 expect(Option.getOrThrow(live.data).title).toBe("Caught-up title");
626 }),
627 );
628
629 it.effect("resumes replacement sessions from the latest applied sequence", () =>
630 Effect.gen(function* () {
631 const harness = yield* makeHarness({ cached: BASE_THREAD, completionMarker: true });
632 yield* Queue.offer(
633 harness.inputs,
634 titleUpdated("Latest title", CACHED_SNAPSHOT_SEQUENCE + 1),
635 );
636 yield* Queue.offer(harness.inputs, synchronized());
637 yield* awaitThreadState(
638 harness.observed,
639 (value) =>
640 value.status === "live" &&
641 Option.isSome(value.data) &&
642 value.data.value.title === "Latest title",
643 );
644
645 yield* harness.replaceSession;
646 for (let attempt = 0; attempt < 100; attempt += 1) {
647 if ((yield* Ref.get(harness.subscriptionCount)) >= 2) break;
648 yield* Effect.yieldNow;
649 }
650
651 expect(yield* Ref.get(harness.subscriptionCount)).toBe(2);
652 expect(yield* Ref.get(harness.lastSubscribeAfterSequence)).toBe(CACHED_SNAPSHOT_SEQUENCE + 1);
653 expect((yield* Ref.get(harness.latest)).status).toBe("synchronizing");
654 }),
655 );
656
657 it.effect("resubscribes on app foreground from the latest applied sequence", () =>
658 Effect.gen(function* () {
659 const harness = yield* makeHarness({ cached: BASE_THREAD, completionMarker: true });
660 yield* Queue.offer(
661 harness.inputs,
662 titleUpdated("Latest title", CACHED_SNAPSHOT_SEQUENCE + 1),
663 );
664 yield* Queue.offer(harness.inputs, synchronized());
665 yield* awaitThreadState(
666 harness.observed,
667 (value) =>
668 value.status === "live" &&
669 Option.isSome(value.data) &&
670 value.data.value.title === "Latest title",
671 );
672
673 yield* Queue.offer(harness.wakeups, "application-active");
674 const synchronizing = yield* awaitThreadState(
675 harness.observed,
676 (value) => value.status === "synchronizing" && Option.isSome(value.data),
677 );
678 for (let attempt = 0; attempt < 100; attempt += 1) {
679 if ((yield* Ref.get(harness.subscriptionCount)) >= 2) break;
680 yield* Effect.yieldNow;
681 }
682
683 expect(synchronizing.status).toBe("synchronizing");
684 expect(yield* Ref.get(harness.subscriptionCount)).toBe(2);
685 expect(yield* Ref.get(harness.lastSubscribeAfterSequence)).toBe(CACHED_SNAPSHOT_SEQUENCE + 1);
686 expect(yield* Ref.get(harness.lastRequestCompletionMarker)).toBe(true);
687 expect(yield* Ref.get(harness.loaderCalls)).toBe(0);
688
689 yield* Queue.offer(harness.inputs, synchronized());
690 const live = yield* awaitThreadState(
691 harness.observed,
692 (value) => value.status === "live" && Option.isSome(value.data),
693 );
694 expect(Option.getOrThrow(live.data).title).toBe("Latest title");
695 }),
696 );
532});697});
packages/client-runtime/src/state/threads.ts+98-49
...@@ -10,6 +10,7 @@ import * as Cause from "effect/Cause";...@@ -10,6 +10,7 @@ import * as Cause from "effect/Cause";
10import * as Effect from "effect/Effect";10import * as Effect from "effect/Effect";
11import * as Option from "effect/Option";11import * as Option from "effect/Option";
12import * as Queue from "effect/Queue";12import * as Queue from "effect/Queue";
13import * as Ref from "effect/Ref";
13import * as Stream from "effect/Stream";14import * as Stream from "effect/Stream";
14import * as SubscriptionRef from "effect/SubscriptionRef";15import * as SubscriptionRef from "effect/SubscriptionRef";
15import { Atom } from "effect/unstable/reactivity";16import { Atom } from "effect/unstable/reactivity";
...@@ -17,8 +18,9 @@ import { Atom } from "effect/unstable/reactivity";...@@ -17,8 +18,9 @@ import { Atom } from "effect/unstable/reactivity";
17import { EnvironmentRegistry } from "../connection/registry.ts";18import { EnvironmentRegistry } from "../connection/registry.ts";
18import { connectionProjectionPhase } from "../connection/model.ts";19import { connectionProjectionPhase } from "../connection/model.ts";
19import { EnvironmentSupervisor } from "../connection/supervisor.ts";20import { EnvironmentSupervisor } from "../connection/supervisor.ts";
21import * as ConnectionWakeups from "../connection/wakeups.ts";
20import { EnvironmentCacheStore } from "../platform/persistence.ts";22import { EnvironmentCacheStore } from "../platform/persistence.ts";
21import { subscribe } from "../rpc/client.ts";23import { subscribeDynamic } from "../rpc/client.ts";
22import { ThreadSnapshotLoader } from "./threadSnapshotHttp.ts";24import { ThreadSnapshotLoader } from "./threadSnapshotHttp.ts";
23import { parseThreadKey, threadKey } from "./entities.ts";25import { parseThreadKey, threadKey } from "./entities.ts";
24import { applyThreadDetailEvent } from "./threadReducer.ts";26import { applyThreadDetailEvent } from "./threadReducer.ts";
...@@ -52,6 +54,7 @@ export const makeEnvironmentThreadState = Effect.fn("EnvironmentThreadState.make...@@ -52,6 +54,7 @@ export const makeEnvironmentThreadState = Effect.fn("EnvironmentThreadState.make
52 const supervisor = yield* EnvironmentSupervisor;54 const supervisor = yield* EnvironmentSupervisor;
53 const cache = yield* EnvironmentCacheStore;55 const cache = yield* EnvironmentCacheStore;
54 const snapshotLoader = yield* ThreadSnapshotLoader;56 const snapshotLoader = yield* ThreadSnapshotLoader;
57 const wakeups = yield* Effect.serviceOption(ConnectionWakeups.ConnectionWakeups);
55 const environmentId = supervisor.target.environmentId;58 const environmentId = supervisor.target.environmentId;
56 const cached = yield* cache.loadThread(environmentId, threadId).pipe(59 const cached = yield* cache.loadThread(environmentId, threadId).pipe(
57 Effect.catch((error) =>60 Effect.catch((error) =>
...@@ -76,6 +79,7 @@ export const makeEnvironmentThreadState = Effect.fn("EnvironmentThreadState.make...@@ -76,6 +79,7 @@ export const makeEnvironmentThreadState = Effect.fn("EnvironmentThreadState.make
76 const lastSequence = yield* SubscriptionRef.make(79 const lastSequence = yield* SubscriptionRef.make(
77 Option.match(cached, { onNone: () => 0, onSome: (snapshot) => snapshot.snapshotSequence }),80 Option.match(cached, { onNone: () => 0, onSome: (snapshot) => snapshot.snapshotSequence }),
78 );81 );
82 const awaitingCompletion = yield* Ref.make(false);
79 const persistence = yield* Queue.sliding<OrchestrationThreadDetailSnapshot>(1);83 const persistence = yield* Queue.sliding<OrchestrationThreadDetailSnapshot>(1);
8084
81 const persist = Effect.fn("EnvironmentThreadState.persist")(function* (85 const persist = Effect.fn("EnvironmentThreadState.persist")(function* (
...@@ -100,11 +104,15 @@ export const makeEnvironmentThreadState = Effect.fn("EnvironmentThreadState.make...@@ -100,11 +104,15 @@ export const makeEnvironmentThreadState = Effect.fn("EnvironmentThreadState.make
100 Effect.forkScoped,104 Effect.forkScoped,
101 );105 );
102106
103 const setSynchronizing = SubscriptionRef.update(state, (current) => ({107 const setSynchronizing = SubscriptionRef.update(state, (current) =>
104 ...current,108 current.status === "deleted"
105 status: "synchronizing" as const,109 ? current
106 error: Option.none(),110 : {
107 }));111 ...current,
112 status: "synchronizing" as const,
113 error: Option.none(),
114 },
115 );
108 const setReady = SubscriptionRef.update(state, (current) =>116 const setReady = SubscriptionRef.update(state, (current) =>
109 current.status === "live" || current.status === "deleted"117 current.status === "live" || current.status === "deleted"
110 ? current118 ? current
...@@ -114,23 +122,32 @@ export const makeEnvironmentThreadState = Effect.fn("EnvironmentThreadState.make...@@ -114,23 +122,32 @@ export const makeEnvironmentThreadState = Effect.fn("EnvironmentThreadState.make
114 error: Option.none(),122 error: Option.none(),
115 },123 },
116 );124 );
117 const setDisconnected = SubscriptionRef.update(state, (current) => ({125 const setDisconnected = Effect.gen(function* () {
118 ...current,126 yield* Ref.set(awaitingCompletion, false);
119 status: current.status === "deleted" ? current.status : statusWithoutLiveData(current.data),127 yield* SubscriptionRef.update(state, (current) => ({
120 }));
121 const setStreamError = (cause: Cause.Cause<unknown>) =>
122 SubscriptionRef.update(state, (current) => ({
123 ...current,128 ...current,
124 status: current.status === "deleted" ? current.status : statusWithoutLiveData(current.data),129 status: current.status === "deleted" ? current.status : statusWithoutLiveData(current.data),
125 error: Option.some(formatThreadError(cause)),
126 }));130 }));
131 });
132 const setStreamError = (cause: Cause.Cause<unknown>) =>
133 Ref.set(awaitingCompletion, false).pipe(
134 Effect.andThen(
135 SubscriptionRef.update(state, (current) => ({
136 ...current,
137 status:
138 current.status === "deleted" ? current.status : statusWithoutLiveData(current.data),
139 error: Option.some(formatThreadError(cause)),
140 })),
141 ),
142 );
127143
128 const setThread = Effect.fn("EnvironmentThreadState.setThread")(function* (144 const setThread = Effect.fn("EnvironmentThreadState.setThread")(function* (
129 thread: OrchestrationThread,145 thread: OrchestrationThread,
130 ) {146 ) {
147 const waiting = yield* Ref.get(awaitingCompletion);
131 yield* SubscriptionRef.set(state, {148 yield* SubscriptionRef.set(state, {
132 data: Option.some(thread),149 data: Option.some(thread),
133 status: "live",150 status: waiting ? "synchronizing" : "live",
134 error: Option.none(),151 error: Option.none(),
135 });152 });
136 // Active threads can update many times per second and retain large tool153 // Active threads can update many times per second and retain large tool
...@@ -143,6 +160,7 @@ export const makeEnvironmentThreadState = Effect.fn("EnvironmentThreadState.make...@@ -143,6 +160,7 @@ export const makeEnvironmentThreadState = Effect.fn("EnvironmentThreadState.make
143 });160 });
144161
145 const setDeleted = Effect.fn("EnvironmentThreadState.setDeleted")(function* () {162 const setDeleted = Effect.fn("EnvironmentThreadState.setDeleted")(function* () {
163 yield* Ref.set(awaitingCompletion, false);
146 yield* SubscriptionRef.set(state, {164 yield* SubscriptionRef.set(state, {
147 data: Option.none(),165 data: Option.none(),
148 status: "deleted",166 status: "deleted",
...@@ -164,6 +182,16 @@ export const makeEnvironmentThreadState = Effect.fn("EnvironmentThreadState.make...@@ -164,6 +182,16 @@ export const makeEnvironmentThreadState = Effect.fn("EnvironmentThreadState.make
164 const applyItem = Effect.fn("EnvironmentThreadState.applyItem")(function* (182 const applyItem = Effect.fn("EnvironmentThreadState.applyItem")(function* (
165 item: OrchestrationThreadStreamItem,183 item: OrchestrationThreadStreamItem,
166 ) {184 ) {
185 if (item.kind === "synchronized") {
186 yield* Ref.set(awaitingCompletion, false);
187 yield* SubscriptionRef.update(state, (current) =>
188 Option.isSome(current.data) && current.status !== "deleted"
189 ? { ...current, status: "live" as const, error: Option.none() }
190 : current,
191 );
192 return;
193 }
194
167 if (item.kind === "snapshot") {195 if (item.kind === "snapshot") {
168 yield* SubscriptionRef.set(lastSequence, item.snapshot.snapshotSequence);196 yield* SubscriptionRef.set(lastSequence, item.snapshot.snapshotSequence);
169 yield* setThread(item.snapshot.thread);197 yield* setThread(item.snapshot.thread);
...@@ -205,48 +233,69 @@ export const makeEnvironmentThreadState = Effect.fn("EnvironmentThreadState.make...@@ -205,48 +233,69 @@ export const makeEnvironmentThreadState = Effect.fn("EnvironmentThreadState.make
205 Effect.forkScoped,233 Effect.forkScoped,
206 );234 );
207235
236 const foregroundResubscriptions = Option.match(wakeups, {
237 onNone: () => Stream.never,
238 onSome: (service) =>
239 service.changes.pipe(Stream.filter((reason) => reason === "application-active")),
240 });
241
208 yield* setSynchronizing;242 yield* setSynchronizing;
209 yield* Effect.forkScoped(243 yield* Effect.forkScoped(
210 Effect.gen(function* () {244 subscribeDynamic(
211 // Establish the base snapshot to resume from, minimizing bytes over the245 ORCHESTRATION_WS_METHODS.subscribeThread,
212 // wire:246 Effect.fn("EnvironmentThreadState.makeSubscribeInput")(function* (session) {
213 // - Warm cache: reuse the cached snapshot (zero network) and resume via247 const supportsCompletionMarker = yield* session.initialConfig.pipe(
214 // `afterSequence` so we only receive events since the cached sequence.248 Effect.map((config) => config.threadResumeCompletionMarker === true),
215 // - Cold cache: load the full snapshot over HTTP (gzip-compressible, and249 Effect.orElseSucceed(() => false),
216 // off the socket), then resume via `afterSequence`.250 );
217 // If no base can be established we fall back to the socket-embedded251 yield* Ref.set(awaitingCompletion, supportsCompletionMarker);
218 // snapshot so the thread still synchronizes. Overlapping/replayed events252 yield* setSynchronizing;
219 // are deduped by sequence in applyItem.
220 const base = Option.isSome(cached)
221 ? cached
222 : yield* Effect.gen(function* () {
223 // Cold cache only: wait for a prepared connection so we can
224 // authenticate the HTTP request; this mirrors the socket path, which
225 // likewise waits for a live session.
226 const prepared = yield* SubscriptionRef.changes(supervisor.prepared).pipe(
227 Stream.filter(Option.isSome),
228 Stream.map((current) => current.value),
229 Stream.runHead,
230 );
231 return Option.isSome(prepared)
232 ? yield* snapshotLoader.load(prepared.value, threadId)
233 : Option.none<OrchestrationThreadDetailSnapshot>();
234 });
235253
236 if (Option.isSome(base)) {254 let current = yield* SubscriptionRef.get(state);
237 yield* applyItem({ kind: "snapshot", snapshot: base.value });255 if (Option.isNone(current.data) && current.status !== "deleted") {
238 }256 const prepared = yield* SubscriptionRef.get(supervisor.prepared).pipe(
257 Effect.flatMap(
258 Option.match({
259 onSome: Effect.succeed,
260 onNone: () =>
261 SubscriptionRef.changes(supervisor.prepared).pipe(
262 Stream.filter(Option.isSome),
263 Stream.map((value) => value.value),
264 Stream.runHead,
265 Effect.map(Option.getOrThrow),
266 ),
267 }),
268 ),
269 );
270 const httpSnapshot = yield* snapshotLoader.load(prepared, threadId);
271 if (Option.isSome(httpSnapshot)) {
272 yield* applyItem({ kind: "snapshot", snapshot: httpSnapshot.value });
273 current = yield* SubscriptionRef.get(state);
274 }
275 }
239276
240 const subscribeInput = Option.match(base, {277 const sequence = yield* SubscriptionRef.get(lastSequence);
241 onNone: () => ({ threadId }),278 const canResume = Option.isSome(current.data);
242 onSome: (snapshot) => ({ threadId, afterSequence: snapshot.snapshotSequence }),279 if (!supportsCompletionMarker && canResume) {
243 });280 yield* SubscriptionRef.update(state, (value) => ({
281 ...value,
282 status: value.status === "deleted" ? value.status : ("live" as const),
283 error: Option.none(),
284 }));
285 }
244286
245 yield* subscribe(ORCHESTRATION_WS_METHODS.subscribeThread, subscribeInput, {287 return {
288 threadId,
289 ...(canResume ? { afterSequence: sequence } : {}),
290 ...(supportsCompletionMarker ? { requestCompletionMarker: true as const } : {}),
291 };
292 }),
293 {
246 onExpectedFailure: setStreamError,294 onExpectedFailure: setStreamError,
247 retryExpectedFailureAfter: "250 millis",295 retryExpectedFailureAfter: "250 millis",
248 }).pipe(Stream.runForEach(applyItem));296 resubscribe: foregroundResubscriptions,
249 }),297 },
298 ).pipe(Stream.runForEach(applyItem)),
250 );299 );
251300
252 yield* Effect.addFinalizer(() =>301 yield* Effect.addFinalizer(() =>
packages/contracts/src/orchestration.ts+16
...@@ -443,6 +443,9 @@ export const OrchestrationShellStreamEvent = Schema.Union([...@@ -443,6 +443,9 @@ export const OrchestrationShellStreamEvent = Schema.Union([
443export type OrchestrationShellStreamEvent = typeof OrchestrationShellStreamEvent.Type;443export type OrchestrationShellStreamEvent = typeof OrchestrationShellStreamEvent.Type;
444444
445export const OrchestrationShellStreamItem = Schema.Union([445export const OrchestrationShellStreamItem = Schema.Union([
446 Schema.Struct({
447 kind: Schema.Literal("synchronized"),
448 }),
446 Schema.Struct({449 Schema.Struct({
447 kind: Schema.Literal("snapshot"),450 kind: Schema.Literal("snapshot"),
448 snapshot: OrchestrationShellSnapshot,451 snapshot: OrchestrationShellSnapshot,
...@@ -461,6 +464,11 @@ export const OrchestrationSubscribeShellInput = Schema.Struct({...@@ -461,6 +464,11 @@ export const OrchestrationSubscribeShellInput = Schema.Struct({
461 * client).464 * client).
462 */465 */
463 afterSequence: Schema.optionalKey(NonNegativeInt),466 afterSequence: Schema.optionalKey(NonNegativeInt),
467 /**
468 * Requests an explicit marker after the subscription has emitted its initial
469 * snapshot or catch-up replay and before it begins emitting live events.
470 */
471 requestCompletionMarker: Schema.optionalKey(Schema.Boolean),
464});472});
465export type OrchestrationSubscribeShellInput = typeof OrchestrationSubscribeShellInput.Type;473export type OrchestrationSubscribeShellInput = typeof OrchestrationSubscribeShellInput.Type;
466474
...@@ -474,6 +482,11 @@ export const OrchestrationSubscribeThreadInput = Schema.Struct({...@@ -474,6 +482,11 @@ export const OrchestrationSubscribeThreadInput = Schema.Struct({
474 * sequence on the client).482 * sequence on the client).
475 */483 */
476 afterSequence: Schema.optionalKey(NonNegativeInt),484 afterSequence: Schema.optionalKey(NonNegativeInt),
485 /**
486 * Requests an explicit marker after the subscription has emitted its initial
487 * snapshot or catch-up replay and before it begins emitting live events.
488 */
489 requestCompletionMarker: Schema.optionalKey(Schema.Boolean),
477});490});
478export type OrchestrationSubscribeThreadInput = typeof OrchestrationSubscribeThreadInput.Type;491export type OrchestrationSubscribeThreadInput = typeof OrchestrationSubscribeThreadInput.Type;
479492
...@@ -1135,6 +1148,9 @@ export const OrchestrationEvent = Schema.Union([...@@ -1135,6 +1148,9 @@ export const OrchestrationEvent = Schema.Union([
1135export type OrchestrationEvent = typeof OrchestrationEvent.Type;1148export type OrchestrationEvent = typeof OrchestrationEvent.Type;
11361149
1137export const OrchestrationThreadStreamItem = Schema.Union([1150export const OrchestrationThreadStreamItem = Schema.Union([
1151 Schema.Struct({
1152 kind: Schema.Literal("synchronized"),
1153 }),
1138 Schema.Struct({1154 Schema.Struct({
1139 kind: Schema.Literal("snapshot"),1155 kind: Schema.Literal("snapshot"),
1140 snapshot: OrchestrationThreadDetailSnapshot,1156 snapshot: OrchestrationThreadDetailSnapshot,
packages/contracts/src/server.ts+4
...@@ -417,6 +417,10 @@ export const ServerConfig = Schema.Struct({...@@ -417,6 +417,10 @@ export const ServerConfig = Schema.Struct({
417 availableEditors: Schema.Array(EditorId),417 availableEditors: Schema.Array(EditorId),
418 observability: ServerObservability,418 observability: ServerObservability,
419 settings: ServerSettings,419 settings: ServerSettings,
420 /** Whether shell subscriptions can emit an opt-in catch-up completion marker. */
421 shellResumeCompletionMarker: Schema.optionalKey(Schema.Boolean),
422 /** Whether thread subscriptions can emit an opt-in catch-up completion marker. */
423 threadResumeCompletionMarker: Schema.optionalKey(Schema.Boolean),
420});424});
421export type ServerConfig = typeof ServerConfig.Type;425export type ServerConfig = typeof ServerConfig.Type;
422426