diff --git a/apps/server/src/server.test.ts b/apps/server/src/server.test.ts index 9134db08ca08..54eb605e4053 100644 --- a/apps/server/src/server.test.ts +++ b/apps/server/src/server.test.ts @@ -46,12 +46,14 @@ import * as Deferred from "effect/Deferred"; import * as DateTime from "effect/DateTime"; import * as Duration from "effect/Duration"; import * as Effect from "effect/Effect"; +import * as Fiber from "effect/Fiber"; import * as FileSystem from "effect/FileSystem"; import * as Layer from "effect/Layer"; import * as ManagedRuntime from "effect/ManagedRuntime"; import * as Option from "effect/Option"; import * as Path from "effect/Path"; import * as PubSub from "effect/PubSub"; +import * as Queue from "effect/Queue"; import * as Stream from "effect/Stream"; import * as TestClock from "effect/testing/TestClock"; import { ChildProcessSpawner } from "effect/unstable/process"; @@ -5997,6 +5999,252 @@ it.layer(NodeServices.layer)("server router seam", (it) => { }).pipe(Effect.provide(NodeHttpServer.layerTest)), ); + it.effect("serves a fresh shell snapshot instead of replaying an ancient cursor", () => + Effect.gen(function* () { + const snapshot = { + snapshotSequence: 2_500, + projects: [], + threads: [ + makeDefaultOrchestrationThreadShell({ + id: ThreadId.make("thread-fresh-shell"), + title: "Fresh shell thread", + }), + ], + updatedAt: "2026-07-08T16:00:00.000Z", + }; + let readEventsCalled = false; + + yield* buildAppUnderTest({ + layers: { + projectionSnapshotQuery: { + getSnapshotSequence: () => + Effect.succeed({ snapshotSequence: snapshot.snapshotSequence }), + getShellSnapshot: () => Effect.succeed(snapshot), + }, + orchestrationEngine: { + readEvents: () => { + readEventsCalled = true; + return Stream.empty; + }, + }, + }, + }); + + const wsUrl = yield* getWsServerUrl("/ws"); + const items = yield* Effect.scoped( + withWsRpcClient(wsUrl, (client) => + client[ORCHESTRATION_WS_METHODS.subscribeShell]({ afterSequence: 1 }).pipe( + Stream.take(1), + Stream.runCollect, + ), + ), + ); + + assert.deepEqual(Array.from(items), [{ kind: "snapshot", snapshot }]); + assert.equal(readEventsCalled, false); + }).pipe(Effect.provide(NodeHttpServer.layerTest)), + ); + + it.effect("acknowledges an up-to-date shell replay with no emitted events", () => + Effect.gen(function* () { + let shellSnapshotCalled = false; + + yield* buildAppUnderTest({ + layers: { + projectionSnapshotQuery: { + getSnapshotSequence: () => Effect.succeed({ snapshotSequence: 2_500 }), + getShellSnapshot: () => { + shellSnapshotCalled = true; + return Effect.succeed({ + snapshotSequence: 2_500, + projects: [], + threads: [], + updatedAt: "2026-07-08T16:00:00.000Z", + }); + }, + }, + orchestrationEngine: { + readEvents: () => Stream.empty, + }, + }, + }); + + const wsUrl = yield* getWsServerUrl("/ws"); + const items = yield* Effect.scoped( + withWsRpcClient(wsUrl, (client) => + client[ORCHESTRATION_WS_METHODS.subscribeShell]({ afterSequence: 2_499 }).pipe( + Stream.take(1), + Stream.runCollect, + ), + ), + ); + + assert.deepEqual(Array.from(items), [{ kind: "caught-up", sequence: 2_500 }]); + assert.equal(shellSnapshotCalled, false); + }).pipe(Effect.provide(NodeHttpServer.layerTest)), + ); + + it.effect("serves an authoritative shell snapshot when the client cursor is ahead", () => + Effect.gen(function* () { + const snapshot = { + snapshotSequence: 25, + projects: [], + threads: [ + makeDefaultOrchestrationThreadShell({ + id: ThreadId.make("thread-reset-shell"), + title: "Reset shell thread", + }), + ], + updatedAt: "2026-07-08T16:00:00.000Z", + }; + let readEventsCalled = false; + + yield* buildAppUnderTest({ + layers: { + projectionSnapshotQuery: { + getSnapshotSequence: () => + Effect.succeed({ snapshotSequence: snapshot.snapshotSequence }), + getShellSnapshot: () => Effect.succeed(snapshot), + }, + orchestrationEngine: { + readEvents: () => { + readEventsCalled = true; + return Stream.empty; + }, + }, + }, + }); + + const wsUrl = yield* getWsServerUrl("/ws"); + const items = yield* Effect.scoped( + withWsRpcClient(wsUrl, (client) => + client[ORCHESTRATION_WS_METHODS.subscribeShell]({ afterSequence: 250 }).pipe( + Stream.take(1), + Stream.runCollect, + ), + ), + ); + + assert.deepEqual(Array.from(items), [{ kind: "snapshot", snapshot, force: true }]); + assert.equal(readEventsCalled, false); + }).pipe(Effect.provide(NodeHttpServer.layerTest)), + ); + + it.effect("fails shell replay instead of acknowledging dropped projection lookups", () => + Effect.gen(function* () { + const projectionError = new PersistenceSqlError({ + operation: "ProjectionSnapshotQuery.getProjectShellById:test", + detail: "failed to read replayed project shell", + }); + + yield* buildAppUnderTest({ + layers: { + projectionSnapshotQuery: { + getSnapshotSequence: () => Effect.succeed({ snapshotSequence: 1 }), + getProjectShellById: () => Effect.fail(projectionError), + }, + orchestrationEngine: { + readEvents: () => + Stream.make({ + sequence: 1, + eventId: EventId.make("event-shell-projection-failure"), + aggregateKind: "project", + aggregateId: defaultProjectId, + occurredAt: "2026-04-05T00:00:00.000Z", + commandId: null, + causationEventId: null, + correlationId: null, + metadata: {}, + type: "project.created", + payload: { + projectId: defaultProjectId, + title: "Default Project", + workspaceRoot: "/tmp/default-project", + defaultModelSelection, + scripts: [], + createdAt: "2026-04-05T00:00:00.000Z", + updatedAt: "2026-04-05T00:00:00.000Z", + }, + } satisfies Extract), + }, + }, + }); + + const wsUrl = yield* getWsServerUrl("/ws"); + const result = yield* Effect.scoped( + withWsRpcClient(wsUrl, (client) => + client[ORCHESTRATION_WS_METHODS.subscribeShell]({ afterSequence: 0 }).pipe( + Stream.runCollect, + ), + ).pipe(Effect.result), + ); + + assertTrue(result._tag === "Failure"); + assertTrue(result.failure._tag === "OrchestrationGetSnapshotError"); + assert.include(result.failure.message, "Failed to load orchestration project shell"); + }).pipe(Effect.provide(NodeHttpServer.layerTest)), + ); + + it.effect("propagates live shell projection failures from the replay buffer", () => + Effect.gen(function* () { + const liveEvents = yield* Queue.unbounded(); + const projectionError = new PersistenceSqlError({ + operation: "ProjectionSnapshotQuery.getProjectShellById:live", + detail: "failed to read live project shell", + }); + + yield* buildAppUnderTest({ + layers: { + projectionSnapshotQuery: { + getSnapshotSequence: () => Effect.succeed({ snapshotSequence: 1 }), + getProjectShellById: () => Effect.fail(projectionError), + }, + orchestrationEngine: { + readEvents: () => Stream.empty, + streamDomainEvents: Stream.fromQueue(liveEvents), + }, + }, + }); + + const wsUrl = yield* getWsServerUrl("/ws"); + const result = yield* Effect.scoped( + withWsRpcClient(wsUrl, (client) => + Effect.gen(function* () { + const fiber = yield* client[ORCHESTRATION_WS_METHODS.subscribeShell]({ + afterSequence: 1, + }).pipe(Stream.runCollect, Effect.result, Effect.forkScoped); + yield* Queue.offer(liveEvents, { + sequence: 2, + eventId: EventId.make("event-live-shell-projection-failure"), + aggregateKind: "project", + aggregateId: defaultProjectId, + occurredAt: "2026-04-05T00:00:00.000Z", + commandId: null, + causationEventId: null, + correlationId: null, + metadata: {}, + type: "project.created", + payload: { + projectId: defaultProjectId, + title: "Default Project", + workspaceRoot: "/tmp/default-project", + defaultModelSelection, + scripts: [], + createdAt: "2026-04-05T00:00:00.000Z", + updatedAt: "2026-04-05T00:00:00.000Z", + }, + } satisfies Extract); + return yield* Fiber.join(fiber); + }), + ), + ); + + assertTrue(result._tag === "Failure"); + assertTrue(result.failure._tag === "OrchestrationGetSnapshotError"); + assert.include(result.failure.message, "Failed to load orchestration project shell"); + }).pipe(Effect.provide(NodeHttpServer.layerTest)), + ); + it.effect("enriches replayed project events with repository identity metadata", () => Effect.gen(function* () { const repositoryIdentity = { diff --git a/apps/server/src/ws.ts b/apps/server/src/ws.ts index f20f3a68e74c..5353b5e5403b 100644 --- a/apps/server/src/ws.ts +++ b/apps/server/src/ws.ts @@ -118,11 +118,13 @@ import { failEnvironmentAuthInvalid, failEnvironmentInternal } from "./auth/http import * as RelayClient from "@t3tools/shared/relayClient"; import { makeServerConfigHeartbeatStream, shouldSendServerConfigHeartbeat } from "./wsKeepalive.ts"; const isOrchestrationDispatchCommandError = Schema.is(OrchestrationDispatchCommandError); +const isOrchestrationGetSnapshotError = Schema.is(OrchestrationGetSnapshotError); const isOrchestrationScheduledTaskMutationError = Schema.is( OrchestrationScheduledTaskMutationError, ); const nowIso = Effect.map(DateTime.now, DateTime.formatIso); +const SHELL_REPLAY_SNAPSHOT_GAP_THRESHOLD = 1_000; function unexpectedCompatibilityError(error: never): never { throw new Error(`Unhandled compatibility error: ${String(error)}`); @@ -558,7 +560,11 @@ const makeWsRpcLayer = ( const toShellStreamEvent = ( event: OrchestrationEvent, - ): Effect.Effect, never, never> => { + ): Effect.Effect< + Option.Option, + OrchestrationGetSnapshotError, + never + > => { switch (event.type) { case "project.created": case "project.meta-updated": @@ -570,7 +576,13 @@ const makeWsRpcLayer = ( project: nextProject, })), ), - Effect.orElseSucceed(() => Option.none()), + Effect.mapError( + (cause) => + new OrchestrationGetSnapshotError({ + message: "Failed to load orchestration project shell", + cause, + }), + ), ); case "project.deleted": return Effect.succeed( @@ -598,7 +610,13 @@ const makeWsRpcLayer = ( thread: nextThread, })), ), - Effect.orElseSucceed(() => Option.none()), + Effect.mapError( + (cause) => + new OrchestrationGetSnapshotError({ + message: "Failed to load orchestration thread shell", + cause, + }), + ), ); default: if (event.aggregateKind !== "thread") { @@ -614,7 +632,13 @@ const makeWsRpcLayer = ( thread: nextThread, })), ), - Effect.orElseSucceed(() => Option.none()), + Effect.mapError( + (cause) => + new OrchestrationGetSnapshotError({ + message: "Failed to load orchestration thread shell", + cause, + }), + ), ); } }; @@ -831,7 +855,10 @@ const makeWsRpcLayer = ( observeRpcStreamEffect( ORCHESTRATION_WS_METHODS.subscribeShell, Effect.gen(function* () { - const liveStream = orchestrationEngine.streamDomainEvents.pipe( + const liveStream: Stream.Stream< + OrchestrationShellStreamItem, + OrchestrationGetSnapshotError + > = orchestrationEngine.streamDomainEvents.pipe( Stream.mapEffect(toShellStreamEvent), Stream.flatMap((event) => Option.isSome(event) ? Stream.succeed(event.value) : Stream.empty, @@ -851,10 +878,80 @@ const makeWsRpcLayer = ( const afterSequence = input.afterSequence; return Stream.unwrap( Effect.gen(function* () { - const liveBuffer = yield* Queue.unbounded(); + const liveBuffer = yield* Queue.unbounded< + | { + readonly _tag: "item"; + readonly item: OrchestrationShellStreamItem; + } + | { + readonly _tag: "error"; + readonly error: OrchestrationGetSnapshotError; + } + >(); + const liveBufferStream = Stream.fromQueue(liveBuffer).pipe( + Stream.mapEffect((entry) => + entry._tag === "item" + ? Effect.succeed(entry.item) + : Effect.fail(entry.error), + ), + ); yield* Effect.forkScoped( - liveStream.pipe(Stream.runForEach((item) => Queue.offer(liveBuffer, item))), + liveStream.pipe( + Stream.runForEach((item) => + Queue.offer(liveBuffer, { + _tag: "item" as const, + item, + }), + ), + Effect.catch((error) => + Queue.offer(liveBuffer, { + _tag: "error" as const, + error, + }), + ), + ), ); + const currentSequence = yield* projectionSnapshotQuery + .getSnapshotSequence() + .pipe( + Effect.tapError((cause) => + Effect.logError("orchestration shell sequence load failed", { cause }), + ), + Effect.mapError( + (cause) => + new OrchestrationGetSnapshotError({ + message: "Failed to load orchestration shell sequence", + cause, + }), + ), + ); + const isClientCursorAhead = afterSequence > currentSequence.snapshotSequence; + if ( + isClientCursorAhead || + currentSequence.snapshotSequence - afterSequence > + SHELL_REPLAY_SNAPSHOT_GAP_THRESHOLD + ) { + const snapshot = yield* projectionSnapshotQuery.getShellSnapshot().pipe( + Effect.tapError((cause) => + Effect.logError("orchestration shell snapshot load failed", { cause }), + ), + Effect.mapError( + (cause) => + new OrchestrationGetSnapshotError({ + message: "Failed to load orchestration shell snapshot", + cause, + }), + ), + ); + return Stream.concat( + Stream.make({ + kind: "snapshot" as const, + snapshot, + ...(isClientCursorAhead ? { force: true } : {}), + }), + liveBufferStream, + ); + } const catchUpStream = orchestrationEngine .readEvents(afterSequence, Number.MAX_SAFE_INTEGER) .pipe( @@ -862,15 +959,25 @@ const makeWsRpcLayer = ( Stream.flatMap((event) => Option.isSome(event) ? Stream.succeed(event.value) : Stream.empty, ), - Stream.mapError( - (cause) => - new OrchestrationGetSnapshotError({ - message: "Failed to replay orchestration shell events", - cause, - }), + Stream.mapError((cause) => + isOrchestrationGetSnapshotError(cause) + ? cause + : new OrchestrationGetSnapshotError({ + message: "Failed to replay orchestration shell events", + cause, + }), ), ); - return Stream.concat(catchUpStream, Stream.fromQueue(liveBuffer)); + return Stream.concat( + catchUpStream, + Stream.concat( + Stream.make({ + kind: "caught-up" as const, + sequence: currentSequence.snapshotSequence, + }), + liveBufferStream, + ), + ); }), ); } diff --git a/packages/client-runtime/src/state/shell-sync.test.ts b/packages/client-runtime/src/state/shell-sync.test.ts index de240d58bbc6..a8901c6adbd9 100644 --- a/packages/client-runtime/src/state/shell-sync.test.ts +++ b/packages/client-runtime/src/state/shell-sync.test.ts @@ -1,6 +1,8 @@ import { + EnvironmentAuthorizationError, EnvironmentId, ORCHESTRATION_WS_METHODS, + OrchestrationGetSnapshotError, type OrchestrationShellSnapshot, type OrchestrationShellStreamItem, } from "@t3tools/contracts"; @@ -10,6 +12,7 @@ import * as Option from "effect/Option"; import * as Queue from "effect/Queue"; import * as Stream from "effect/Stream"; import * as SubscriptionRef from "effect/SubscriptionRef"; +import * as TestClock from "effect/testing/TestClock"; import { AVAILABLE_CONNECTION_STATE, @@ -193,4 +196,564 @@ describe("environment shell synchronization", () => { expect(yield* SubscriptionRef.get(loaderCalls)).toBe(0); }), ); + + it.effect("moves a stalled warm replay out of live state", () => + Effect.gen(function* () { + const cachedSnapshot: OrchestrationShellSnapshot = { + snapshotSequence: 5, + projects: [], + threads: [], + updatedAt: "2026-06-06T00:00:00.000Z", + }; + const client = { + [ORCHESTRATION_WS_METHODS.subscribeShell]: () => Stream.never, + } as unknown as WsRpcProtocolClient; + const supervisorState = yield* SubscriptionRef.make(AVAILABLE_CONNECTION_STATE); + const activeSession = yield* SubscriptionRef.make>( + Option.some(session(client)), + ); + const supervisor = EnvironmentSupervisor.EnvironmentSupervisor.of({ + target: TARGET, + state: supervisorState, + session: activeSession, + 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(cachedSnapshot)), + saveShell: () => Effect.void, + loadThread: () => Effect.succeed(Option.none()), + saveThread: () => Effect.void, + removeThread: () => Effect.void, + clear: () => Effect.void, + }); + const snapshotLoader = ShellSnapshotLoader.of({ + load: () => Effect.never, + }); + const shellState = yield* makeEnvironmentShellState().pipe( + Effect.provideService(EnvironmentSupervisor.EnvironmentSupervisor, supervisor), + Effect.provideService(Persistence.EnvironmentCacheStore, cache), + Effect.provideService(ShellSnapshotLoader, snapshotLoader), + ); + + yield* SubscriptionRef.changes(shellState).pipe( + Stream.filter((state) => state.status === "live"), + Stream.runHead, + ); + yield* TestClock.adjust("5 seconds"); + yield* Effect.yieldNow; + + const state = yield* SubscriptionRef.get(shellState); + expect(state.status).toBe("synchronizing"); + expect(Option.getOrThrow(state.snapshot)).toEqual(cachedSnapshot); + }), + ); + + it.effect("moves a failed warm replay out of live state without retrying denied access", () => + Effect.gen(function* () { + const cachedSnapshot: OrchestrationShellSnapshot = { + snapshotSequence: 5, + projects: [], + threads: [], + updatedAt: "2026-06-06T00:00:00.000Z", + }; + const subscriptionCalls = yield* SubscriptionRef.make(0); + const client = { + [ORCHESTRATION_WS_METHODS.subscribeShell]: () => + Stream.fromEffect(SubscriptionRef.update(subscriptionCalls, (count) => count + 1)).pipe( + Stream.drain, + Stream.concat( + Stream.fail( + new EnvironmentAuthorizationError({ + message: "Denied", + requiredScope: "orchestration:read", + }), + ), + ), + ), + } as unknown as WsRpcProtocolClient; + const supervisorState = yield* SubscriptionRef.make(AVAILABLE_CONNECTION_STATE); + const activeSession = yield* SubscriptionRef.make>( + Option.some(session(client)), + ); + const supervisor = EnvironmentSupervisor.EnvironmentSupervisor.of({ + target: TARGET, + state: supervisorState, + session: activeSession, + 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(cachedSnapshot)), + saveShell: () => Effect.void, + loadThread: () => Effect.succeed(Option.none()), + saveThread: () => Effect.void, + removeThread: () => Effect.void, + clear: () => Effect.void, + }); + const snapshotLoader = ShellSnapshotLoader.of({ + load: () => Effect.never, + }); + const shellState = yield* makeEnvironmentShellState().pipe( + Effect.provideService(EnvironmentSupervisor.EnvironmentSupervisor, supervisor), + Effect.provideService(Persistence.EnvironmentCacheStore, cache), + Effect.provideService(ShellSnapshotLoader, snapshotLoader), + ); + + yield* SubscriptionRef.changes(shellState).pipe( + Stream.filter((state) => Option.isSome(state.error)), + Stream.runHead, + ); + + const state = yield* SubscriptionRef.get(shellState); + expect(state.status).toBe("synchronizing"); + expect(state.error).toEqual(Option.some("Could not synchronize environment data.")); + expect(Option.getOrThrow(state.snapshot)).toEqual(cachedSnapshot); + yield* TestClock.adjust("1 second"); + yield* Effect.yieldNow; + expect(yield* SubscriptionRef.get(subscriptionCalls)).toBe(1); + }), + ); + + it.effect("retries a recoverable warm replay failure and returns live", () => + Effect.gen(function* () { + const cachedSnapshot: OrchestrationShellSnapshot = { + snapshotSequence: 5, + projects: [], + threads: [], + updatedAt: "2026-06-06T00:00:00.000Z", + }; + const subscriptionCalls = yield* SubscriptionRef.make(0); + const client = { + [ORCHESTRATION_WS_METHODS.subscribeShell]: () => + Stream.unwrap( + SubscriptionRef.updateAndGet(subscriptionCalls, (count) => count + 1).pipe( + Effect.map((call) => + call === 1 + ? Stream.fail( + new OrchestrationGetSnapshotError({ + message: "Replay projection failed", + }), + ) + : Stream.succeed({ + kind: "caught-up", + sequence: 6, + } satisfies OrchestrationShellStreamItem), + ), + ), + ), + } as unknown as WsRpcProtocolClient; + const supervisorState = yield* SubscriptionRef.make(AVAILABLE_CONNECTION_STATE); + const activeSession = yield* SubscriptionRef.make>( + Option.some(session(client)), + ); + const supervisor = EnvironmentSupervisor.EnvironmentSupervisor.of({ + target: TARGET, + state: supervisorState, + session: activeSession, + prepared: yield* SubscriptionRef.make(Option.some(PREPARED)), + connect: Effect.void, + disconnect: Effect.void, + retryNow: Effect.void, + } satisfies EnvironmentSupervisor.EnvironmentSupervisor["Service"]); + const savedSequence = yield* SubscriptionRef.make(cachedSnapshot.snapshotSequence); + const cache = Persistence.EnvironmentCacheStore.of({ + loadShell: () => Effect.succeed(Option.some(cachedSnapshot)), + saveShell: (_environmentId, snapshot) => + SubscriptionRef.set(savedSequence, snapshot.snapshotSequence), + loadThread: () => Effect.succeed(Option.none()), + saveThread: () => Effect.void, + removeThread: () => Effect.void, + clear: () => Effect.void, + }); + const snapshotLoader = ShellSnapshotLoader.of({ + load: () => Effect.never, + }); + const shellState = yield* makeEnvironmentShellState().pipe( + Effect.provideService(EnvironmentSupervisor.EnvironmentSupervisor, supervisor), + Effect.provideService(Persistence.EnvironmentCacheStore, cache), + Effect.provideService(ShellSnapshotLoader, snapshotLoader), + ); + + yield* SubscriptionRef.changes(shellState).pipe( + Stream.filter((state) => Option.isSome(state.error)), + Stream.runHead, + ); + expect(yield* SubscriptionRef.get(subscriptionCalls)).toBe(1); + + yield* TestClock.adjust("250 millis"); + yield* SubscriptionRef.changes(shellState).pipe( + Stream.filter((state) => state.status === "live"), + Stream.runHead, + ); + + const state = yield* SubscriptionRef.get(shellState); + expect(state.status).toBe("live"); + expect(state.error).toEqual(Option.none()); + expect(Option.getOrThrow(state.snapshot).snapshotSequence).toBe(6); + yield* TestClock.adjust("500 millis"); + expect(yield* SubscriptionRef.get(savedSequence)).toBe(6); + expect(yield* SubscriptionRef.get(subscriptionCalls)).toBe(2); + }), + ); + + it.effect("keeps an acknowledged idle warm replay live without snapshot recovery", () => + Effect.gen(function* () { + const cachedSnapshot: OrchestrationShellSnapshot = { + snapshotSequence: 5, + projects: [], + threads: [], + updatedAt: "2026-06-06T00:00:00.000Z", + }; + const acknowledgedSequence = 10; + const events = yield* Queue.unbounded(); + const client = { + [ORCHESTRATION_WS_METHODS.subscribeShell]: () => Stream.fromQueue(events), + } as unknown as WsRpcProtocolClient; + const supervisorState = yield* SubscriptionRef.make(AVAILABLE_CONNECTION_STATE); + const activeSession = yield* SubscriptionRef.make>( + Option.some(session(client)), + ); + const supervisor = EnvironmentSupervisor.EnvironmentSupervisor.of({ + target: TARGET, + state: supervisorState, + session: activeSession, + prepared: yield* SubscriptionRef.make(Option.some(PREPARED)), + connect: Effect.void, + disconnect: Effect.void, + retryNow: Effect.void, + } satisfies EnvironmentSupervisor.EnvironmentSupervisor["Service"]); + const savedSequence = yield* SubscriptionRef.make(cachedSnapshot.snapshotSequence); + const cache = Persistence.EnvironmentCacheStore.of({ + loadShell: () => Effect.succeed(Option.some(cachedSnapshot)), + saveShell: (_environmentId, snapshot) => + SubscriptionRef.update(savedSequence, (current) => + Math.max(current, snapshot.snapshotSequence), + ), + loadThread: () => Effect.succeed(Option.none()), + saveThread: () => Effect.void, + removeThread: () => Effect.void, + clear: () => Effect.void, + }); + const loaderCalls = yield* SubscriptionRef.make(0); + const snapshotLoader = ShellSnapshotLoader.of({ + load: () => + SubscriptionRef.updateAndGet(loaderCalls, (count) => count + 1).pipe( + Effect.as(Option.none()), + ), + }); + const shellState = yield* makeEnvironmentShellState().pipe( + Effect.provideService(EnvironmentSupervisor.EnvironmentSupervisor, supervisor), + Effect.provideService(Persistence.EnvironmentCacheStore, cache), + Effect.provideService(ShellSnapshotLoader, snapshotLoader), + ); + + yield* Queue.offer(events, { kind: "caught-up", sequence: acknowledgedSequence }); + for (let index = 0; index < 10; index += 1) { + yield* Effect.yieldNow; + } + yield* TestClock.adjust("6 seconds"); + yield* Effect.yieldNow; + + const state = yield* SubscriptionRef.get(shellState); + expect(state.status).toBe("live"); + expect(state.error).toEqual(Option.none()); + expect(Option.getOrThrow(state.snapshot)).toEqual({ + ...cachedSnapshot, + snapshotSequence: acknowledgedSequence, + }); + expect(yield* SubscriptionRef.get(savedSequence)).toBe(acknowledgedSequence); + expect(yield* SubscriptionRef.get(loaderCalls)).toBe(0); + }), + ); + + it.effect("accepts a forced snapshot when the server cursor moved backwards", () => + Effect.gen(function* () { + const cachedSnapshot: OrchestrationShellSnapshot = { + snapshotSequence: 10, + projects: [], + threads: [], + updatedAt: "2026-06-06T00:00:00.000Z", + }; + const resetSnapshot: OrchestrationShellSnapshot = { + snapshotSequence: 3, + projects: [], + threads: [], + updatedAt: "2026-06-06T00:00:01.000Z", + }; + const events = yield* Queue.unbounded(); + const capturedAfterSequences: Array = []; + const client = { + [ORCHESTRATION_WS_METHODS.subscribeShell]: (input: { readonly afterSequence?: number }) => { + capturedAfterSequences.push(input.afterSequence); + return Stream.fromQueue(events); + }, + } as unknown as WsRpcProtocolClient; + const supervisorState = yield* SubscriptionRef.make(AVAILABLE_CONNECTION_STATE); + const activeSession = yield* SubscriptionRef.make>( + Option.some(session(client)), + ); + const supervisor = EnvironmentSupervisor.EnvironmentSupervisor.of({ + target: TARGET, + state: supervisorState, + session: activeSession, + 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(cachedSnapshot)), + saveShell: () => Effect.void, + loadThread: () => Effect.succeed(Option.none()), + saveThread: () => Effect.void, + removeThread: () => Effect.void, + clear: () => Effect.void, + }); + const snapshotLoader = ShellSnapshotLoader.of({ + load: () => Effect.never, + }); + const shellState = yield* makeEnvironmentShellState().pipe( + Effect.provideService(EnvironmentSupervisor.EnvironmentSupervisor, supervisor), + Effect.provideService(Persistence.EnvironmentCacheStore, cache), + Effect.provideService(ShellSnapshotLoader, snapshotLoader), + ); + + yield* Queue.offer(events, { kind: "snapshot", snapshot: resetSnapshot, force: true }); + yield* SubscriptionRef.changes(shellState).pipe( + Stream.filter( + (state) => + state.status === "live" && + Option.isSome(state.snapshot) && + state.snapshot.value.snapshotSequence === resetSnapshot.snapshotSequence, + ), + Stream.runHead, + ); + + expect(Option.getOrThrow((yield* SubscriptionRef.get(shellState)).snapshot)).toEqual( + resetSnapshot, + ); + yield* SubscriptionRef.set(activeSession, Option.none()); + yield* Effect.yieldNow; + yield* SubscriptionRef.set(activeSession, Option.some(session(client))); + for (let index = 0; index < 10; index += 1) { + yield* Effect.yieldNow; + } + expect(capturedAfterSequences).toEqual([ + cachedSnapshot.snapshotSequence, + resetSnapshot.snapshotSequence, + ]); + }), + ); + + it.effect( + "keeps HTTP recovery snapshots in synchronizing state until the socket catches up", + () => + Effect.gen(function* () { + const cachedSnapshot: OrchestrationShellSnapshot = { + snapshotSequence: 5, + projects: [], + threads: [], + updatedAt: "2026-06-06T00:00:00.000Z", + }; + const recoverySnapshot: OrchestrationShellSnapshot = { + snapshotSequence: 6, + projects: [], + threads: [], + updatedAt: "2026-06-06T00:00:01.000Z", + }; + const client = { + [ORCHESTRATION_WS_METHODS.subscribeShell]: () => Stream.never, + } as unknown as WsRpcProtocolClient; + const supervisorState = yield* SubscriptionRef.make(AVAILABLE_CONNECTION_STATE); + const activeSession = yield* SubscriptionRef.make>( + Option.some(session(client)), + ); + const supervisor = EnvironmentSupervisor.EnvironmentSupervisor.of({ + target: TARGET, + state: supervisorState, + session: activeSession, + 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(cachedSnapshot)), + saveShell: () => Effect.void, + loadThread: () => Effect.succeed(Option.none()), + saveThread: () => Effect.void, + removeThread: () => Effect.void, + clear: () => Effect.void, + }); + const snapshotLoader = ShellSnapshotLoader.of({ + load: () => Effect.succeed(Option.some(recoverySnapshot)), + }); + const shellState = yield* makeEnvironmentShellState().pipe( + Effect.provideService(EnvironmentSupervisor.EnvironmentSupervisor, supervisor), + Effect.provideService(Persistence.EnvironmentCacheStore, cache), + Effect.provideService(ShellSnapshotLoader, snapshotLoader), + ); + + yield* TestClock.adjust("5 seconds"); + yield* Effect.yieldNow; + + const state = yield* SubscriptionRef.get(shellState); + expect(state.status).toBe("synchronizing"); + expect(Option.getOrThrow(state.snapshot)).toEqual(recoverySnapshot); + }), + ); + + it.effect("rearms stalled replay recovery when the environment reconnects", () => + Effect.gen(function* () { + const cachedSnapshot: OrchestrationShellSnapshot = { + snapshotSequence: 5, + projects: [], + threads: [], + updatedAt: "2026-06-06T00:00:00.000Z", + }; + const recoverySnapshot: OrchestrationShellSnapshot = { + snapshotSequence: 8, + projects: [], + threads: [], + updatedAt: "2026-06-06T00:00:01.000Z", + }; + const events = yield* Queue.unbounded(); + const client = { + [ORCHESTRATION_WS_METHODS.subscribeShell]: () => Stream.fromQueue(events), + } as unknown as WsRpcProtocolClient; + const supervisorState = yield* SubscriptionRef.make(AVAILABLE_CONNECTION_STATE); + const activeSession = yield* SubscriptionRef.make>( + Option.some(session(client)), + ); + const supervisor = EnvironmentSupervisor.EnvironmentSupervisor.of({ + target: TARGET, + state: supervisorState, + session: activeSession, + 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(cachedSnapshot)), + saveShell: () => Effect.void, + loadThread: () => Effect.succeed(Option.none()), + saveThread: () => Effect.void, + removeThread: () => Effect.void, + clear: () => Effect.void, + }); + const snapshotLoader = ShellSnapshotLoader.of({ + load: () => Effect.succeed(Option.some(recoverySnapshot)), + }); + const shellState = yield* makeEnvironmentShellState().pipe( + Effect.provideService(EnvironmentSupervisor.EnvironmentSupervisor, supervisor), + Effect.provideService(Persistence.EnvironmentCacheStore, cache), + Effect.provideService(ShellSnapshotLoader, snapshotLoader), + ); + + yield* Queue.offer(events, { kind: "caught-up", sequence: cachedSnapshot.snapshotSequence }); + for (let index = 0; index < 10; index += 1) { + yield* Effect.yieldNow; + } + yield* SubscriptionRef.set(supervisorState, { + desired: true, + network: "online", + phase: "connecting", + stage: "synchronizing", + attempt: 2, + generation: 1, + lastFailure: null, + retryAt: null, + }); + yield* TestClock.adjust("5 seconds"); + yield* Effect.yieldNow; + + const state = yield* SubscriptionRef.get(shellState); + expect(state.status).toBe("synchronizing"); + expect(Option.getOrThrow(state.snapshot)).toEqual(recoverySnapshot); + }), + ); + + it.effect("does not let stalled replay recovery overwrite a newer socket snapshot", () => + Effect.gen(function* () { + const cachedSnapshot: OrchestrationShellSnapshot = { + snapshotSequence: 5, + projects: [], + threads: [], + updatedAt: "2026-06-06T00:00:00.000Z", + }; + const recoverySnapshot: OrchestrationShellSnapshot = { + snapshotSequence: 6, + projects: [], + threads: [], + updatedAt: "2026-06-06T00:00:01.000Z", + }; + const socketSnapshot: OrchestrationShellSnapshot = { + snapshotSequence: 7, + projects: [], + threads: [], + updatedAt: "2026-06-06T00:00:02.000Z", + }; + const events = yield* Queue.unbounded(); + const client = { + [ORCHESTRATION_WS_METHODS.subscribeShell]: () => Stream.fromQueue(events), + } as unknown as WsRpcProtocolClient; + const supervisorState = yield* SubscriptionRef.make(AVAILABLE_CONNECTION_STATE); + const activeSession = yield* SubscriptionRef.make>( + Option.some(session(client)), + ); + const supervisor = EnvironmentSupervisor.EnvironmentSupervisor.of({ + target: TARGET, + state: supervisorState, + session: activeSession, + 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(cachedSnapshot)), + saveShell: () => Effect.void, + loadThread: () => Effect.succeed(Option.none()), + saveThread: () => Effect.void, + removeThread: () => Effect.void, + clear: () => Effect.void, + }); + const snapshotLoader = ShellSnapshotLoader.of({ + load: () => Effect.sleep("1 second").pipe(Effect.as(Option.some(recoverySnapshot))), + }); + const shellState = yield* makeEnvironmentShellState().pipe( + Effect.provideService(EnvironmentSupervisor.EnvironmentSupervisor, supervisor), + Effect.provideService(Persistence.EnvironmentCacheStore, cache), + Effect.provideService(ShellSnapshotLoader, snapshotLoader), + ); + + yield* SubscriptionRef.changes(shellState).pipe( + Stream.filter((state) => state.status === "live"), + Stream.runHead, + ); + yield* TestClock.adjust("5 seconds"); + yield* Queue.offer(events, { kind: "snapshot", snapshot: socketSnapshot }); + yield* SubscriptionRef.changes(shellState).pipe( + Stream.filter( + (state) => + state.status === "live" && + Option.isSome(state.snapshot) && + state.snapshot.value.snapshotSequence === socketSnapshot.snapshotSequence, + ), + Stream.runHead, + ); + + yield* TestClock.adjust("1 second"); + yield* Effect.yieldNow; + + const state = yield* SubscriptionRef.get(shellState); + expect(Option.getOrThrow(state.snapshot)).toEqual(socketSnapshot); + }), + ); }); diff --git a/packages/client-runtime/src/state/shell.ts b/packages/client-runtime/src/state/shell.ts index faa70bc4f3a7..d1a2523e45bd 100644 --- a/packages/client-runtime/src/state/shell.ts +++ b/packages/client-runtime/src/state/shell.ts @@ -1,4 +1,5 @@ import { + EnvironmentAuthorizationError, ORCHESTRATION_WS_METHODS, type EnvironmentId, type OrchestrationShellSnapshot, @@ -9,6 +10,8 @@ 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 Schema from "effect/Schema"; import * as Stream from "effect/Stream"; import * as SubscriptionRef from "effect/SubscriptionRef"; import { AsyncResult, Atom } from "effect/unstable/reactivity"; @@ -44,7 +47,25 @@ function shellStatusForSnapshot( return Option.isSome(snapshot) ? "cached" : "empty"; } +function synchronizingStatusForSnapshot( + snapshot: Option.Option, +): EnvironmentShellStatus { + return Option.isSome(snapshot) ? "synchronizing" : "empty"; +} + const SHELL_SYNCHRONIZATION_ERROR_MESSAGE = "Could not synchronize environment data."; +const SHELL_REPLAY_STALL_TIMEOUT = "5 seconds"; +const SHELL_EXPECTED_FAILURE_RETRY_DELAY = "250 millis"; +const isEnvironmentAuthorizationError = Schema.is(EnvironmentAuthorizationError); + +function isTerminalShellSubscriptionFailure(cause: Cause.Cause): boolean { + return ( + cause.reasons.length > 0 && + cause.reasons.every( + (reason) => reason._tag === "Fail" && isEnvironmentAuthorizationError(reason.error), + ) + ); +} export const makeEnvironmentShellState = Effect.fn("EnvironmentShellState.make")(function* () { const supervisor = yield* EnvironmentSupervisor; @@ -68,6 +89,9 @@ export const makeEnvironmentShellState = Effect.fn("EnvironmentShellState.make") error: Option.none(), }); const persistence = yield* Queue.sliding(1); + const serverItemSeen = yield* Ref.make(false); + const replayWatchdogEpoch = yield* Ref.make(0); + const subscribeInput: { afterSequence?: number } = {}; const persist = Effect.fn("EnvironmentShellState.persist")(function* ( snapshot: OrchestrationShellSnapshot, @@ -117,19 +141,60 @@ export const makeEnvironmentShellState = Effect.fn("EnvironmentShellState.make") Effect.andThen( SubscriptionRef.update(state, (current) => ({ ...current, - status: shellStatusForSnapshot(current.snapshot), + status: synchronizingStatusForSnapshot(current.snapshot), error: Option.some(SHELL_SYNCHRONIZATION_ERROR_MESSAGE), })), ), ); + const setExpectedStreamError = Effect.fn("EnvironmentShellState.setExpectedStreamError")( + function* (cause: Cause.Cause) { + yield* setStreamError(Cause.squash(cause)); + if (isTerminalShellSubscriptionFailure(cause)) { + return yield* Effect.never; + } + }, + ); + const applyItem = Effect.fn("EnvironmentShellState.applyItem")(function* ( item: OrchestrationShellStreamItem, ) { + if (item.kind === "caught-up") { + const current = yield* SubscriptionRef.get(state); + if (Option.isNone(current.snapshot)) { + yield* SubscriptionRef.set(state, { + ...current, + error: Option.none(), + }); + return; + } + + const nextSnapshot = + item.sequence > current.snapshot.value.snapshotSequence + ? { ...current.snapshot.value, snapshotSequence: item.sequence } + : current.snapshot.value; + yield* SubscriptionRef.set(state, { + snapshot: Option.some(nextSnapshot), + status: "live", + error: Option.none(), + }); + subscribeInput.afterSequence = nextSnapshot.snapshotSequence; + if (nextSnapshot !== current.snapshot.value) { + yield* Queue.offer(persistence, nextSnapshot); + } + return; + } + const current = yield* SubscriptionRef.get(state); const nextSnapshot = item.kind === "snapshot" - ? item.snapshot + ? Option.match(current.snapshot, { + onNone: () => item.snapshot, + onSome: (snapshot) => + item.force === true || item.snapshot.snapshotSequence >= snapshot.snapshotSequence + ? item.snapshot + : snapshot, + }) : Option.match(current.snapshot, { onNone: () => null, onSome: (snapshot) => @@ -146,9 +211,77 @@ export const makeEnvironmentShellState = Effect.fn("EnvironmentShellState.make") status: "live", error: Option.none(), }); + subscribeInput.afterSequence = nextSnapshot.snapshotSequence; yield* Queue.offer(persistence, nextSnapshot); }); + const applyRecoverySnapshot = Effect.fn("EnvironmentShellState.applyRecoverySnapshot")(function* ( + snapshot: OrchestrationShellSnapshot, + ) { + const current = yield* SubscriptionRef.get(state); + const nextSnapshot = Option.match(current.snapshot, { + onNone: () => snapshot, + onSome: (currentSnapshot) => + snapshot.snapshotSequence >= currentSnapshot.snapshotSequence ? snapshot : currentSnapshot, + }); + + yield* SubscriptionRef.set(state, { + snapshot: Option.some(nextSnapshot), + status: "synchronizing", + error: Option.none(), + }); + subscribeInput.afterSequence = nextSnapshot.snapshotSequence; + if (Option.isNone(current.snapshot) || nextSnapshot !== current.snapshot.value) { + yield* Queue.offer(persistence, nextSnapshot); + } + }); + + const recoverFromStalledReplay = Effect.fn("EnvironmentShellState.recoverFromStalledReplay")( + function* (epoch: number) { + if ((yield* Ref.get(replayWatchdogEpoch)) !== epoch || (yield* Ref.get(serverItemSeen))) { + return; + } + + yield* SubscriptionRef.update(state, (current) => ({ + ...current, + status: Option.isSome(current.snapshot) ? "synchronizing" : current.status, + })); + + const prepared = yield* SubscriptionRef.get(supervisor.prepared); + if (Option.isNone(prepared)) { + yield* setStreamError(new Error("Shell replay stalled before the server caught up.")); + return; + } + + const snapshot = yield* snapshotLoader.load(prepared.value); + if ((yield* Ref.get(replayWatchdogEpoch)) !== epoch || (yield* Ref.get(serverItemSeen))) { + return; + } + if (Option.isSome(snapshot)) { + yield* applyRecoverySnapshot(snapshot.value); + return; + } + + yield* setStreamError( + new Error("Shell replay stalled and snapshot refresh was unavailable."), + ); + }, + ); + + const armReplayWatchdog = Effect.fn("EnvironmentShellState.armReplayWatchdog")(function* () { + const current = yield* SubscriptionRef.get(state); + if (Option.isNone(current.snapshot)) { + return; + } + + const epoch = yield* Ref.updateAndGet(replayWatchdogEpoch, (value) => value + 1); + yield* Ref.set(serverItemSeen, false); + yield* Effect.sleep(SHELL_REPLAY_STALL_TIMEOUT).pipe( + Effect.andThen(recoverFromStalledReplay(epoch)), + Effect.forkScoped, + ); + }); + yield* Effect.forkScoped( Effect.gen(function* () { // Establish the base shell snapshot to resume from, minimizing bytes over @@ -176,23 +309,35 @@ export const makeEnvironmentShellState = Effect.fn("EnvironmentShellState.make") if (Option.isSome(base)) { yield* applyItem({ kind: "snapshot", snapshot: base.value }); + yield* armReplayWatchdog(); } - const subscribeInput = Option.match(base, { - onNone: () => ({}), - onSome: (snapshot) => ({ afterSequence: snapshot.snapshotSequence }), + Option.match(base, { + onNone: () => { + delete subscribeInput.afterSequence; + }, + onSome: (snapshot) => { + subscribeInput.afterSequence = snapshot.snapshotSequence; + }, }); yield* subscribe(ORCHESTRATION_WS_METHODS.subscribeShell, subscribeInput, { - onExpectedFailure: (cause) => setStreamError(Cause.squash(cause)), - }).pipe(Stream.runForEach(applyItem)); + onExpectedFailure: setExpectedStreamError, + retryExpectedFailureAfter: SHELL_EXPECTED_FAILURE_RETRY_DELAY, + }).pipe( + Stream.tap(() => Ref.set(serverItemSeen, true)), + Stream.catchCause((cause) => + Stream.fromEffect(setStreamError(Cause.squash(cause))).pipe(Stream.drain), + ), + Stream.runForEach(applyItem), + ); }), ); yield* SubscriptionRef.changes(supervisor.state).pipe( Stream.runForEach((connectionState) => { switch (connectionProjectionPhase(connectionState)) { case "synchronizing": - return setSynchronizing; + return setSynchronizing.pipe(Effect.andThen(armReplayWatchdog())); case "disconnected": return setDisconnected; case "ready": diff --git a/packages/contracts/src/orchestration.ts b/packages/contracts/src/orchestration.ts index a387e11e11c2..37e4accc42af 100644 --- a/packages/contracts/src/orchestration.ts +++ b/packages/contracts/src/orchestration.ts @@ -455,11 +455,19 @@ export const OrchestrationShellStreamEvent = Schema.Union([ ]); export type OrchestrationShellStreamEvent = typeof OrchestrationShellStreamEvent.Type; +export const OrchestrationShellCaughtUp = Schema.Struct({ + kind: Schema.Literal("caught-up"), + sequence: NonNegativeInt, +}); +export type OrchestrationShellCaughtUp = typeof OrchestrationShellCaughtUp.Type; + export const OrchestrationShellStreamItem = Schema.Union([ Schema.Struct({ kind: Schema.Literal("snapshot"), snapshot: OrchestrationShellSnapshot, + force: Schema.optional(Schema.Boolean), }), + OrchestrationShellCaughtUp, OrchestrationShellStreamEvent, ]); export type OrchestrationShellStreamItem = typeof OrchestrationShellStreamItem.Type;