diff --git a/apps/mobile/src/connection/environment-cache-store.test.ts b/apps/mobile/src/connection/environment-cache-store.test.ts index 2e80259bb8c76..5f48ad04d38de 100644 --- a/apps/mobile/src/connection/environment-cache-store.test.ts +++ b/apps/mobile/src/connection/environment-cache-store.test.ts @@ -498,12 +498,21 @@ describe("mobile SQLite environment cache store", () => { Effect.provideService(MobileDatabase.MobileDatabase, memory.database), ); - yield* store.saveShell(ENVIRONMENT_ID, SHELL_SNAPSHOT); + const origin = { + serverVersion: "1.0.0", + databaseMigration: 70, + omitIdleSubagents: true, + omitOlderSettled: false, + snapshotAt: 1_800_000_000_000, + }; + yield* store.saveShell(ENVIRONMENT_ID, SHELL_SNAPSHOT, origin); yield* store.saveThread(ENVIRONMENT_ID, THREAD_SNAPSHOT); const shell = Option.getOrThrow(yield* store.loadShell(ENVIRONMENT_ID)); const thread = Option.getOrThrow(yield* store.loadThread(ENVIRONMENT_ID, THREAD_ID)); + expect(shell.origin).toEqual(origin); + expect(DateTime.formatIso(shell.threads[0]!.updatedAt)).toBe("2026-07-29T12:00:00.000Z"); expect(DateTime.formatIso(shell.threads[0]!.latestUserMessageAt!)).toBe( "2026-07-29T12:00:00.000Z", diff --git a/apps/mobile/src/connection/environment-cache-store.ts b/apps/mobile/src/connection/environment-cache-store.ts index 6fada3fbeda3d..60ec7c0db2f0d 100644 --- a/apps/mobile/src/connection/environment-cache-store.ts +++ b/apps/mobile/src/connection/environment-cache-store.ts @@ -106,7 +106,13 @@ export const make = Effect.fn("MobileEnvironmentCacheStore.make")(function* () { operation: "load-shell", decode: decodeStoredShellSnapshot, select: (stored) => - stored.environmentId === environmentId ? Option.some(stored.snapshot) : Option.none(), + stored.environmentId === environmentId + ? Option.some( + stored.origin === undefined + ? stored.snapshot + : { ...stored.snapshot, origin: stored.origin }, + ) + : Option.none(), }).pipe( Effect.zipWith( Effect.promise(() => projectFaviconDatabaseCache.hydrate()), @@ -115,16 +121,25 @@ export const make = Effect.fn("MobileEnvironmentCacheStore.make")(function* () { ), ), ), - saveShell: Effect.fn("MobileEnvironmentCache.saveShell")(function* (environmentId, snapshot) { - const payload = yield* encodeStoredShellSnapshot({ - schemaVersion: ORCHESTRATION_CACHE_SCHEMA_VERSION, - environmentId, - snapshot, - }).pipe(Effect.mapError((cause) => persistenceError("save-shell", cause))); - yield* database - .saveCache(environmentId, "shell", "snapshot", ORCHESTRATION_CACHE_SCHEMA_VERSION, payload) - .pipe(Effect.mapError(mapDatabaseError("save-shell"))); - }), + saveShell: Effect.fn("MobileEnvironmentCache.saveShell")( + function* (environmentId, snapshot, origin) { + const payload = yield* encodeStoredShellSnapshot({ + schemaVersion: ORCHESTRATION_CACHE_SCHEMA_VERSION, + environmentId, + snapshot, + ...(origin === undefined ? {} : { origin }), + }).pipe(Effect.mapError((cause) => persistenceError("save-shell", cause))); + yield* database + .saveCache( + environmentId, + "shell", + "snapshot", + ORCHESTRATION_CACHE_SCHEMA_VERSION, + payload, + ) + .pipe(Effect.mapError(mapDatabaseError("save-shell"))); + }, + ), loadThread: Effect.fn("MobileEnvironmentCache.loadThread")((environmentId, threadId) => loadDecodedCache({ database, diff --git a/apps/server/src/ws.test.ts b/apps/server/src/ws.test.ts index 2be9e61c7c916..0ac17248cc3c4 100644 --- a/apps/server/src/ws.test.ts +++ b/apps/server/src/ws.test.ts @@ -1434,10 +1434,15 @@ it.effect("a publish whose push fails still refreshes the new remote", () => ), ); -// Pruned history cannot be replayed: a cursor below the boundary must resync from a snapshot. -it.effect.each(["thread", "shell"] as const)( - "%s resume from a cursor below the pruned boundary loads a snapshot", - (scope) => +// Pruned history cannot be replayed: a cursor below the boundary must resync from a snapshot, +// or, for a shell client that loads it over HTTP, be told to without the server building one. +it.effect.each([ + { scope: "thread", httpSnapshotFallback: false }, + { scope: "shell", httpSnapshotFallback: false }, + { scope: "shell", httpSnapshotFallback: true }, +] as const)( + "$scope resume from a cursor below the pruned boundary (HTTP fallback: $httpSnapshotFallback)", + ({ scope, httpSnapshotFallback }) => Effect.gen(function* () { const sql = yield* SqlClient.SqlClient; const now = "2026-10-13T00:00:00.000Z"; @@ -1461,11 +1466,18 @@ it.effect.each(["thread", "shell"] as const)( ) VALUES (${OrchestrationEventStore.EVENT_LOG_PRUNED_THROUGH}, 2, ${second!.sequence}, ${now}) `; const subscription = Effect.gen(function* () { - const stream = + const items: ReadonlyArray = scope === "thread" - ? yield* subscribeOrchestrationV2Thread({ threadId, afterSequence: first!.sequence }) - : yield* subscribeOrchestrationV2Shell({ afterSequence: first!.sequence }); - return stream; + ? yield* Stream.runCollect( + yield* subscribeOrchestrationV2Thread({ threadId, afterSequence: first!.sequence }), + ) + : yield* Stream.runCollect( + yield* subscribeOrchestrationV2Shell({ + afterSequence: first!.sequence, + ...(httpSnapshotFallback ? { httpSnapshotFallback: true as const } : {}), + }), + ); + return items; }); const exit = yield* subscription.pipe( Effect.provideService(ThreadManagement.ThreadManagementService, { @@ -1487,7 +1499,10 @@ it.effect.each(["thread", "shell"] as const)( } as ProjectEnrichment.ProjectEnrichmentService["Service"]), Effect.exit, ); - assert.deepEqual(Exit.isFailure(exit) ? Cause.squash(exit.cause) : exit, "snapshot"); + assert.deepEqual( + Exit.isFailure(exit) ? Cause.squash(exit.cause) : Array.from(exit.value), + httpSnapshotFallback ? [{ kind: "snapshot-required" }] : "snapshot", + ); }).pipe(Effect.scoped, Effect.provide(ReplayTestLayer)), ); diff --git a/apps/server/src/ws.ts b/apps/server/src/ws.ts index 035f8b1688d44..a9ebc9f35c3b2 100644 --- a/apps/server/src/ws.ts +++ b/apps/server/src/ws.ts @@ -64,6 +64,7 @@ import { OrchestrationV2ThreadLaunchError, type OrchestrationProjectShell, type OrchestrationV2ShellSnapshot, + type OrchestrationV2ShellStreamItem, type OrchestrationV2ThreadStreamItem, type ApplicationStoredEvent, type ProjectEntriesFailure, @@ -1115,6 +1116,7 @@ export const subscribeOrchestrationV2Thread = Effect.fn("ws.orchestrationV2.subs export const subscribeOrchestrationV2Shell = Effect.fn("ws.orchestrationV2.subscribeShell")( function* (input: { readonly afterSequence?: number; + readonly httpSnapshotFallback?: true; readonly requestCompletionMarker?: boolean; readonly omitIdleSubagents?: boolean; readonly omitOlderSettled?: boolean; @@ -1325,14 +1327,20 @@ export const subscribeOrchestrationV2Shell = Effect.fn("ws.orchestrationV2.subsc }); } - const fromSnapshot = Effect.gen(function* () { - const loaded = yield* loadSnapshot(); - return composeShellStreamWithEnrichment({ - initial: initialSnapshotItems(loaded), - tail: completionThenLive(loaded.snapshot.snapshotSequence), - enrichment: enrichmentRefreshes, - }); + const snapshotRequired: Stream.Stream = Stream.make({ + kind: "snapshot-required", }); + const fromSnapshot = + input.httpSnapshotFallback === true + ? Effect.succeed(snapshotRequired) + : Effect.gen(function* () { + const loaded = yield* loadSnapshot(); + return composeShellStreamWithEnrichment({ + initial: initialSnapshotItems(loaded), + tail: completionThenLive(loaded.snapshot.snapshotSequence), + enrichment: enrichmentRefreshes, + }); + }); const highWater = yield* applicationEvents.latestApplicationSequence; if (!(yield* canReplayPersistedRange(input.afterSequence, highWater, SHELL_RESUME_MAX_GAP))) { return yield* fromSnapshot; diff --git a/apps/web/src/connection/storage.ts b/apps/web/src/connection/storage.ts index 78b01fc6d4695..c987488e0cf61 100644 --- a/apps/web/src/connection/storage.ts +++ b/apps/web/src/connection/storage.ts @@ -846,7 +846,11 @@ export const layer = Layer.effectContext( Effect.mapError((cause) => persistenceError("load-shell", cause)), Effect.map((stored) => stored.environmentId === environmentId - ? Option.some(stored.snapshot) + ? Option.some( + stored.origin === undefined + ? stored.snapshot + : { ...stored.snapshot, origin: stored.origin }, + ) : Option.none(), ), ), @@ -857,11 +861,12 @@ export const layer = Layer.effectContext( ); }), saveShell: Effect.fn("storage.saveShell")( - function* (environmentId, snapshot) { + function* (environmentId, snapshot, origin) { const encoded = yield* encodeStoredShellSnapshot({ schemaVersion: ORCHESTRATION_CACHE_SCHEMA_VERSION, environmentId, snapshot, + ...(origin === undefined ? {} : { origin }), }).pipe(Effect.mapError((cause) => persistenceError("save-shell", cause))); yield* writeDatabaseValue(database, SHELL_STORE_NAME, environmentId, encoded); }, diff --git a/packages/client-runtime/src/platform/orchestrationCache.ts b/packages/client-runtime/src/platform/orchestrationCache.ts index e7800e20bd33a..3d24e03f113f9 100644 --- a/packages/client-runtime/src/platform/orchestrationCache.ts +++ b/packages/client-runtime/src/platform/orchestrationCache.ts @@ -13,10 +13,25 @@ export const ORCHESTRATION_CACHE_SCHEMA_VERSION = 3 as const; // Older caches may contain human turns that bypassed server history budgets. export const ORCHESTRATION_THREAD_CACHE_SCHEMA_VERSION = 5 as const; +/** + * The server snapshot the cached rows descend from. Written only for rows that have one, so a + * cache from before it existed decodes and never resumes. + */ +export const ShellCacheOrigin = Schema.Struct({ + serverVersion: Schema.String, + databaseMigration: Schema.NullOr(Schema.Int), + omitIdleSubagents: Schema.Boolean, + omitOlderSettled: Schema.Boolean, + /** Epoch milliseconds of that snapshot; replay does not renew it. */ + snapshotAt: Schema.Number, +}); +export type ShellCacheOrigin = typeof ShellCacheOrigin.Type; + export const StoredOrchestrationShellSnapshot = Schema.Struct({ schemaVersion: Schema.Literal(ORCHESTRATION_CACHE_SCHEMA_VERSION), environmentId: EnvironmentId, snapshot: OrchestrationV2ShellSnapshotJson, + origin: Schema.optionalKey(ShellCacheOrigin), }); export const StoredOrchestrationThreadSnapshot = Schema.Struct({ diff --git a/packages/client-runtime/src/platform/persistence.ts b/packages/client-runtime/src/platform/persistence.ts index e043279d3e576..6476b80221225 100644 --- a/packages/client-runtime/src/platform/persistence.ts +++ b/packages/client-runtime/src/platform/persistence.ts @@ -14,6 +14,13 @@ import * as Schema from "effect/Schema"; import type { ConnectionRegistration } from "../connection/catalog.ts"; import type { DeferredShellSnapshot } from "../state/shellPullRequests.ts"; import type { ConnectionTarget, PersistedConnectionTarget } from "../connection/model.ts"; +import type { ShellCacheOrigin } from "./orchestrationCache.ts"; + +export type { ShellCacheOrigin }; + +export interface CachedShellSnapshot extends DeferredShellSnapshot { + readonly origin?: ShellCacheOrigin; +} export class ConnectionPersistenceError extends Schema.TaggedError()( "ConnectionPersistenceError", @@ -82,10 +89,12 @@ export class EnvironmentCacheStore extends Context.Service< { readonly loadShell: ( environmentId: EnvironmentId, - ) => Effect.Effect, ConnectionPersistenceError>; + ) => Effect.Effect, ConnectionPersistenceError>; + /** `origin` only when the rows descend from a server snapshot. */ readonly saveShell: ( environmentId: EnvironmentId, snapshot: OrchestrationV2ShellSnapshot, + origin?: ShellCacheOrigin, ) => Effect.Effect; readonly loadThread: ( environmentId: EnvironmentId, diff --git a/packages/client-runtime/src/state/shell-sync.test.ts b/packages/client-runtime/src/state/shell-sync.test.ts index 032ede63714d2..062db03809cfc 100644 --- a/packages/client-runtime/src/state/shell-sync.test.ts +++ b/packages/client-runtime/src/state/shell-sync.test.ts @@ -1136,3 +1136,182 @@ describe("environment shell synchronization", () => { }), ); }); + +describe("environment shell cold start", () => { + const NOW = 1_800_000_000_000; + const HOUR = 60 * 60 * 1000; + const CACHED_SEQUENCE = 7; + const HTTP_SEQUENCE = 9; + // What `session()` advertises: server 1.0.0, no omissions, no migration. + const CURRENT_ORIGIN = { + serverVersion: "1.0.0", + databaseMigration: null, + omitIdleSubagents: false, + omitOlderSettled: false, + }; + const RECENT_ORIGIN = { ...CURRENT_ORIGIN, snapshotAt: NOW - HOUR }; + + const makeColdStart = Effect.fn("test.makeColdStartShell")(function* ( + origin: Persistence.ShellCacheOrigin | undefined, + ) { + yield* TestClock.setTime(NOW); + const subscriptions = yield* Queue.unbounded<{ + readonly input: unknown; + readonly events: Queue.Queue; + }>(); + const saved = yield* Queue.unbounded<{ + readonly snapshot: OrchestrationV2ShellSnapshot; + readonly origin: Persistence.ShellCacheOrigin | undefined; + }>(); + const httpLoads = yield* Ref.make(0); + const client = { + [ORCHESTRATION_V2_WS_METHODS.subscribeShell]: (input: unknown) => + Stream.unwrap( + Effect.gen(function* () { + const events = yield* Queue.unbounded(); + yield* Queue.offer(subscriptions, { input, events }); + return Stream.fromQueue(events); + }), + ), + } as unknown as WsRpcProtocolClient; + const activeSession = yield* SubscriptionRef.make(Option.some(session(client))); + const supervisor = EnvironmentSupervisor.EnvironmentSupervisor.of({ + target: TARGET, + state: yield* SubscriptionRef.make(AVAILABLE_CONNECTION_STATE), + 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.succeedSome({ + ...v2ShellSnapshot, + snapshotSequence: CACHED_SEQUENCE, + ...(origin === undefined ? {} : { origin }), + }), + saveShell: (_environmentId, snapshot, origin?: Persistence.ShellCacheOrigin) => + Queue.offer(saved, { snapshot, origin }).pipe(Effect.asVoid), + loadThread: () => Effect.succeedNone, + saveThread: () => Effect.void, + removeThread: () => Effect.void, + loadServerConfig: () => Effect.succeedNone, + saveServerConfig: () => Effect.void, + loadVcsRefs: () => Effect.succeedNone, + saveVcsRefs: () => Effect.void, + removeVcsRefs: () => Effect.void, + clearVcsRefs: () => Effect.void, + clear: () => Effect.void, + }); + const snapshotLoader = ShellSnapshotLoader.ShellSnapshotLoader.of({ + load: () => + Ref.updateAndGet(httpLoads, (count) => count + 1).pipe( + Effect.map((count) => + Option.some({ ...v2ShellSnapshot, snapshotSequence: HTTP_SEQUENCE + count - 1 }), + ), + ), + }); + const shellState = yield* makeEnvironmentShellState().pipe( + Effect.provideService(EnvironmentSupervisor.EnvironmentSupervisor, supervisor), + Effect.provideService(Persistence.EnvironmentCacheStore, cache), + Effect.provideService(ShellSnapshotLoader.ShellSnapshotLoader, snapshotLoader), + ); + return { + subscriptions, + saved, + httpLoads, + reconnect: SubscriptionRef.set(activeSession, Option.some(session(client))), + liveAt: (sequence: number) => + SubscriptionRef.changes(shellState).pipe( + Stream.filter( + (state) => + state.status === "live" && + Option.exists(state.snapshot, (snapshot) => snapshot.snapshotSequence === sequence), + ), + Stream.runHead, + Effect.map(Option.getOrThrow), + ), + }; + }); + + it.effect.each([ + { label: "a recent matching origin", origin: RECENT_ORIGIN, resumes: true }, + { label: "no origin", origin: undefined, resumes: false }, + { + label: "other omissions", + origin: { ...RECENT_ORIGIN, omitIdleSubagents: true }, + resumes: false, + }, + { + label: "another migration", + origin: { ...RECENT_ORIGIN, databaseMigration: 70 }, + resumes: false, + }, + { + label: "another server version", + origin: { ...RECENT_ORIGIN, serverVersion: "0.9.0" }, + resumes: false, + }, + { + label: "a 12 hour old origin", + origin: { ...RECENT_ORIGIN, snapshotAt: NOW - 12 * HOUR }, + resumes: false, + }, + ])("starts from a cache with $label (resumes: $resumes)", ({ origin, resumes }) => + Effect.gen(function* () { + const h = yield* makeColdStart(origin); + const first = yield* Queue.take(h.subscriptions); + const sequence = resumes ? CACHED_SEQUENCE : HTTP_SEQUENCE; + expect(first.input).toMatchObject({ afterSequence: sequence }); + // Only held rows may be refused; a fresh snapshot's cursor never loops back to HTTP. + if (resumes) expect(first.input).toMatchObject({ httpSnapshotFallback: true }); + else expect(first.input).not.toHaveProperty("httpSnapshotFallback"); + expect(yield* Ref.get(h.httpLoads)).toBe(resumes ? 0 : 1); + + // Thread families, older threads, and search wait for a nonzero generation. + yield* Queue.offer(first.events, { kind: "synchronized" }); + expect((yield* h.liveAt(sequence)).generation).toBeGreaterThan(0); + + if (!resumes) { + // The next cold start may resume from this snapshot. + yield* TestClock.adjust("500 millis"); + expect(yield* Queue.take(h.saved)).toMatchObject({ + snapshot: { snapshotSequence: HTTP_SEQUENCE }, + origin: { ...CURRENT_ORIGIN, snapshotAt: NOW }, + }); + } + }), + ); + + it.effect("reloads over HTTP when the server answers a resume with snapshot-required", () => + Effect.gen(function* () { + const h = yield* makeColdStart(RECENT_ORIGIN); + + const coldResume = yield* Queue.take(h.subscriptions); + expect(coldResume.input).toMatchObject({ afterSequence: CACHED_SEQUENCE }); + yield* Queue.offer(coldResume.events, { kind: "snapshot-required" }); + const coldReload = yield* Queue.take(h.subscriptions); + expect(coldReload.input).toMatchObject({ afterSequence: HTTP_SEQUENCE }); + expect(coldReload.input).not.toHaveProperty("httpSnapshotFallback"); + expect(yield* Ref.get(h.httpLoads)).toBe(1); + yield* Queue.offer(coldReload.events, { kind: "synchronized" }); + expect((yield* h.liveAt(HTTP_SEQUENCE)).generation).toBeGreaterThan(0); + + // A warm reconnect resumes the in-memory shell and falls back the same way. + yield* h.reconnect; + const warmResume = yield* Queue.take(h.subscriptions); + expect(warmResume.input).toMatchObject({ + afterSequence: HTTP_SEQUENCE, + httpSnapshotFallback: true, + }); + expect(yield* Ref.get(h.httpLoads)).toBe(1); + yield* Queue.offer(warmResume.events, { kind: "snapshot-required" }); + const warmReload = yield* Queue.take(h.subscriptions); + expect(warmReload.input).toMatchObject({ afterSequence: HTTP_SEQUENCE + 1 }); + expect(yield* Ref.get(h.httpLoads)).toBe(2); + yield* Queue.offer(warmReload.events, { kind: "synchronized" }); + yield* h.liveAt(HTTP_SEQUENCE + 1); + }), + ); +}); diff --git a/packages/client-runtime/src/state/shell.ts b/packages/client-runtime/src/state/shell.ts index d75ea5ff64a05..808fd9fdece56 100644 --- a/packages/client-runtime/src/state/shell.ts +++ b/packages/client-runtime/src/state/shell.ts @@ -9,6 +9,7 @@ import { type ThreadPullRequestLink, } from "@t3tools/contracts"; import * as Cause from "effect/Cause"; +import * as Clock from "effect/Clock"; import * as Effect from "effect/Effect"; import * as Fiber from "effect/Fiber"; import * as Option from "effect/Option"; @@ -82,14 +83,41 @@ function withoutDeferredPullRequests(value: DeferredShellSnapshot): Orchestratio const SHELL_SYNCHRONIZATION_ERROR_MESSAGE = "Could not synchronize environment data."; -function groupBySession(entries: ReadonlyArray) { - const groups: Array<[RpcSession, Array]> = []; - for (const [session, value] of entries) { - const last = groups.at(-1); - if (last?.[0] === session) last[1].push(value); - else groups.push([session, [value]]); - } - return groups; +/** + * Replay does not refresh the older-settled cut, rows that left the newest window, or + * projections the server rebuilt, so drift grows with the age of the last server snapshot. + */ +const CACHE_RESUME_MAX_AGE_MILLIS = 12 * 60 * 60 * 1000; + +type ShellOriginIdentity = Omit; + +interface ShellSubscription { + readonly session: RpcSession; + /** What a snapshot from this subscription descends from; null when the server did not say. */ + readonly identity: ShellOriginIdentity | null; +} + +function canResumeFrom( + origin: Persistence.ShellCacheOrigin | undefined, + identity: ShellOriginIdentity | null, + now: number, +): boolean { + if (origin === undefined || identity === null) return false; + const age = now - origin.snapshotAt; + return ( + origin.serverVersion === identity.serverVersion && + origin.databaseMigration === identity.databaseMigration && + origin.omitIdleSubagents === identity.omitIdleSubagents && + origin.omitOlderSettled === identity.omitOlderSettled && + age >= 0 && + age < CACHE_RESUME_MAX_AGE_MILLIS + ); +} + +interface LiveShell { + readonly snapshot: OrchestrationV2ShellSnapshot; + // Read together with the rows, so a save never pairs them with another snapshot's origin. + readonly origin: Persistence.ShellCacheOrigin | undefined; } export const makeEnvironmentShellState = Effect.fn("EnvironmentShellState.make")(function* () { @@ -105,29 +133,30 @@ export const makeEnvironmentShellState = Effect.fn("EnvironmentShellState.make") environmentId, ...safeErrorLogAttributes(error), }), - Effect.as(Option.none()), + Effect.as(Option.none()), ), ), ); - const cachedSnapshot = Option.map(cached, withoutDeferredPullRequests); + const cachedOrigin = Option.isSome(cached) ? cached.value.origin : undefined; + const cachedSnapshot = Option.map(cached, ({ origin: _origin, ...rows }) => + withoutDeferredPullRequests(rows), + ); const state = yield* SubscriptionRef.make({ ...EMPTY_SHELL_STATE, snapshot: cachedSnapshot, status: shellStatusForSnapshot(cachedSnapshot), }); const awaitingCompletion = yield* Ref.make(false); - // The omissions and server build of the subscription whose snapshot the - // in-memory shell last came from; null while it holds only cached data. - const authoritativeOmission = yield* Ref.make(null); - // Items buffered from a previous session keep that session's key. - const sessionOmissions = new WeakMap(); - const latestLiveSnapshot = yield* Ref.make>( - Option.none(), - ); + const activeSubscription = yield* Ref.make(null); + // The server snapshot the in-memory rows descend from, directly or through the cache. + // Only a server snapshot renews it; `snapshot-required` drops it. + const origin = yield* Ref.make(cachedOrigin); + const reloadRequests = yield* Queue.unbounded(); + const latestLive = yield* Ref.make>(Option.none()); const persistence = yield* Queue.sliding(1); const persistenceLock = yield* Semaphore.make(1); - let lastPersisted: OrchestrationV2ShellSnapshot | undefined; + let lastPersisted: LiveShell | undefined; // Rows still waiting for their deferred pull request links, which must not be saved: // the cache would keep them without links until the next server refresh. let pendingLinkRows: ReadonlySet = new Set(); @@ -135,24 +164,25 @@ export const makeEnvironmentShellState = Effect.fn("EnvironmentShellState.make") flush: boolean, ) { while (true) { - const latest = yield* Ref.get(latestLiveSnapshot); + const latest = yield* Ref.get(latestLive); if (Option.isNone(latest) || latest.value === lastPersisted) return; - if (latest.value.threads.some((thread) => pendingLinkRows.has(thread))) return; - const snapshot = latest.value; + const { snapshot, origin } = latest.value; + if (snapshot.threads.some((thread) => pendingLinkRows.has(thread))) return; if ( lastPersisted !== undefined && - snapshot.snapshotSequence >= lastPersisted.snapshotSequence && - snapshot.schemaVersion === lastPersisted.schemaVersion && - snapshot.projects === lastPersisted.projects && - snapshot.threads === lastPersisted.threads && - snapshot.archivedThreads === lastPersisted.archivedThreads + origin === lastPersisted.origin && + snapshot.snapshotSequence >= lastPersisted.snapshot.snapshotSequence && + snapshot.schemaVersion === lastPersisted.snapshot.schemaVersion && + snapshot.projects === lastPersisted.snapshot.projects && + snapshot.threads === lastPersisted.snapshot.threads && + snapshot.archivedThreads === lastPersisted.snapshot.archivedThreads ) { // Keep the live cursor without rewriting the rows. Remember it here too so a // later authoritative reset to a lower sequence still reaches the cache. - lastPersisted = snapshot; + lastPersisted = latest.value; return; } - const saved = yield* cache.saveShell(environmentId, snapshot).pipe( + const saved = yield* cache.saveShell(environmentId, snapshot, origin).pipe( Effect.as(true), Effect.catch((error) => Effect.logWarning("Could not persist environment shell cache.").pipe( @@ -162,7 +192,7 @@ export const makeEnvironmentShellState = Effect.fn("EnvironmentShellState.make") ), ); if (!saved) return; - lastPersisted = snapshot; + lastPersisted = latest.value; if (!flush) return; } }, persistenceLock.withPermit); @@ -220,14 +250,20 @@ export const makeEnvironmentShellState = Effect.fn("EnvironmentShellState.make") const snapshotWriteLock = yield* Semaphore.make(1); const applyItems = Effect.fn("EnvironmentShellState.applyItems")(function* ( - items: ReadonlyArray, - omission: string | null, + entries: ReadonlyArray, pendingLinkIds?: ReadonlySet, ) { + // Read under the lock: items buffered across a session change must not land on the + // rows the next session loaded. + const subscription = yield* Ref.get(activeSubscription); + const items = entries.flatMap(([session, item]) => + session === subscription?.session ? [item] : [], + ); const initial = yield* SubscriptionRef.get(state); let waiting = yield* Ref.get(awaitingCompletion); let next = initial; let receivedAuthoritativeSnapshot = false; + let snapshotRequired = false; let deltas: Array = []; const flushDeltas = () => { if (deltas.length > 0 && Option.isSome(next.snapshot)) { @@ -260,6 +296,10 @@ export const makeEnvironmentShellState = Effect.fn("EnvironmentShellState.make") deltas = []; }; for (const item of items) { + if (item.kind === "snapshot-required") { + snapshotRequired = true; + continue; + } if (item.kind !== "snapshot" && item.kind !== "synchronized") { deltas.push(item); continue; @@ -269,7 +309,13 @@ export const makeEnvironmentShellState = Effect.fn("EnvironmentShellState.make") if (item.kind === "synchronized") { waiting = false; if (Option.isSome(next.snapshot)) { - next = { ...next, status: "live", error: Option.none() }; + next = { + ...next, + status: "live", + error: Option.none(), + // A cache resume that caught up is as current as a snapshot for generation readers. + generation: next.generation === 0 ? ++lastShellGeneration : next.generation, + }; } continue; } @@ -292,6 +338,20 @@ export const makeEnvironmentShellState = Effect.fn("EnvironmentShellState.make") } flushDeltas(); yield* Ref.set(awaitingCompletion, waiting); + let nextOrigin = yield* Ref.get(origin); + if (snapshotRequired) { + // The rows stay on screen but reload before the next resume; the cache keeps the old + // origin until that reload saves, which costs at most one more refusal. + nextOrigin = undefined; + yield* Ref.set(origin, nextOrigin); + yield* Queue.offer(reloadRequests, undefined); + } else if (receivedAuthoritativeSnapshot) { + nextOrigin = + subscription?.identity == null + ? undefined + : { ...subscription.identity, snapshotAt: yield* Clock.currentTimeMillis }; + yield* Ref.set(origin, nextOrigin); + } if (next === initial) return; if (pendingLinkIds !== undefined && Option.isSome(next.snapshot)) { pendingLinkRows = new Set( @@ -299,12 +359,12 @@ export const makeEnvironmentShellState = Effect.fn("EnvironmentShellState.make") ); } if (Option.isSome(next.snapshot)) { - yield* Ref.set(latestLiveSnapshot, next.snapshot); + yield* Ref.set( + latestLive, + Option.some({ snapshot: next.snapshot.value, origin: nextOrigin }), + ); } yield* SubscriptionRef.set(state, next); - if (receivedAuthoritativeSnapshot && omission !== null) { - yield* Ref.set(authoritativeOmission, omission); - } if (next.snapshot !== initial.snapshot && Option.isSome(next.snapshot)) { yield* Queue.offer(persistence, next.snapshot.value); } @@ -365,11 +425,11 @@ export const makeEnvironmentShellState = Effect.fn("EnvironmentShellState.make") }, ); pendingLinkRows = new Set(); - const live = yield* Ref.get(latestLiveSnapshot); + const live = yield* Ref.get(latestLive); if (Option.isNone(live)) return; - const latest = Option.isSome(filled) ? filled : live; - yield* Ref.set(latestLiveSnapshot, latest); - yield* Queue.offer(persistence, latest.value); + const latest = Option.isSome(filled) ? filled.value : live.value.snapshot; + yield* Ref.set(latestLive, Option.some({ snapshot: latest, origin: live.value.origin })); + yield* Queue.offer(persistence, latest); }, (effect) => snapshotWriteLock.withPermit(effect), ); @@ -377,53 +437,52 @@ export const makeEnvironmentShellState = Effect.fn("EnvironmentShellState.make") yield* fillDeferredPullRequests(cached.value, cachedSnapshot.value); } - const foregroundResubscriptions = Option.match(wakeups, { - onNone: () => Stream.never, - onSome: (service) => - service.changes.pipe(Stream.filter(ConnectionWakeups.shouldResubscribeAfterWakeup)), - }); + const resubscriptions = Stream.merge( + Stream.fromQueue(reloadRequests), + Option.match(wakeups, { + onNone: () => Stream.never, + onSome: (service) => + service.changes.pipe(Stream.filter(ConnectionWakeups.shouldResubscribeAfterWakeup)), + }), + ); yield* setSynchronizing; yield* Effect.forkScoped( subscribeDynamicWithSession( ORCHESTRATION_V2_WS_METHODS.subscribeShell, Effect.fn("EnvironmentShellState.makeSubscribeInput")(function* (session) { - const { omitIdleSubagents, omitOlderSettled, omissionKey } = - yield* session.initialConfig.pipe( - Effect.map((config) => { - const omitIdleSubagents = config.shellThreadFamilies === true; - const omitOlderSettled = config.shellOlderSettled === true && omitIdleSubagents; - // Migrations rewrite projections without events, so an upgraded server reloads. - const { serverVersion, databaseMigration } = config.environment; - return { - omitIdleSubagents, - omitOlderSettled, - omissionKey: JSON.stringify([ - omitIdleSubagents, - omitOlderSettled, - serverVersion, - databaseMigration ?? null, - ]), - }; - }), - Effect.orElseSucceed(() => ({ - omitIdleSubagents: false, - omitOlderSettled: false, - omissionKey: null, - })), - ); - sessionOmissions.set(session, omissionKey); + const identity = yield* session.initialConfig.pipe( + Effect.map((config): ShellOriginIdentity => { + const omitIdleSubagents = config.shellThreadFamilies === true; + return { + serverVersion: config.environment.serverVersion, + // Migrations rewrite projections without events, so a migrated server reloads. + databaseMigration: config.environment.databaseMigration ?? null, + omitIdleSubagents, + omitOlderSettled: config.shellOlderSettled === true && omitIdleSubagents, + }; + }), + Effect.orElseSucceed(() => null), + ); + const { omitIdleSubagents, omitOlderSettled } = identity ?? { + omitIdleSubagents: false, + omitOlderSettled: false, + }; + yield* Ref.set(activeSubscription, { session, identity }); yield* Ref.set(awaitingCompletion, true); yield* setSynchronizing; - // A shell from a server snapshot resumes from its cursor, on this session or a - // new one. Cached data, other omissions, or another server build reload the - // authoritative HTTP snapshot so a valid cursor cannot preserve stale data. - const hasAuthoritativeSnapshot = - omissionKey !== null && (yield* Ref.get(authoritativeOmission)) === omissionKey; - let canResume = hasAuthoritativeSnapshot; + // Rows from a recent server snapshot with the same omissions and server build resume + // from their cursor, in memory or from the cache. Anything else reloads the HTTP + // snapshot so a valid cursor cannot preserve stale or incomplete data. let current = yield* SubscriptionRef.get(state); - if (!hasAuthoritativeSnapshot || Option.isNone(current.snapshot)) { + let canResume = + Option.isSome(current.snapshot) && + canResumeFrom(yield* Ref.get(origin), identity, yield* Clock.currentTimeMillis); + // Only rows the client already held may be refused; a cursor from the snapshot just + // loaded falls back to the socket snapshot, so a refusal cannot loop. + const resumingHeldRows = canResume; + if (!canResume) { const prepared = yield* SubscriptionRef.get(supervisor.prepared).pipe( Effect.flatMap( Option.match({ @@ -455,7 +514,7 @@ export const makeEnvironmentShellState = Effect.fn("EnvironmentShellState.make") ? [] : httpSnapshot.value.threads.map((thread) => thread.id), ); - yield* applyItems([{ kind: "snapshot", snapshot }], omissionKey, pendingLinkIds); + yield* applyItems([[session, { kind: "snapshot", snapshot }]], pendingLinkIds); canResume = true; current = yield* SubscriptionRef.get(state); if (Option.isSome(current.snapshot)) { @@ -473,6 +532,7 @@ export const makeEnvironmentShellState = Effect.fn("EnvironmentShellState.make") } return { afterSequence: current.snapshot.value.snapshotSequence, + ...(resumingHeldRows ? { httpSnapshotFallback: true as const } : {}), ...omission, requestCompletionMarker: true as const, }; @@ -480,17 +540,9 @@ export const makeEnvironmentShellState = Effect.fn("EnvironmentShellState.make") { onExpectedFailure: (cause) => setStreamError(Cause.squash(cause)), retryExpectedFailureAfter: "250 millis", - resubscribe: foregroundResubscriptions, + resubscribe: resubscriptions, }, - ).pipe( - Stream.runForEachArray((entries) => - Effect.forEach( - groupBySession(entries), - ([session, items]) => applyItems(items, sessionOmissions.get(session) ?? null), - { discard: true }, - ), - ), - ), + ).pipe(Stream.runForEachArray((entries) => applyItems(entries))), ); yield* SubscriptionRef.changes(supervisor.state).pipe( Stream.runForEach((connectionState) => { diff --git a/packages/client-runtime/src/state/shellReducer.ts b/packages/client-runtime/src/state/shellReducer.ts index 8491143996908..2ef7dea533f3c 100644 --- a/packages/client-runtime/src/state/shellReducer.ts +++ b/packages/client-runtime/src/state/shellReducer.ts @@ -131,10 +131,7 @@ export function mergeShellSnapshotProjects( export function applyShellStreamEvent( snapshot: OrchestrationV2ShellSnapshot, - event: Exclude< - OrchestrationV2ShellStreamItem, - { readonly kind: "snapshot" } | { readonly kind: "synchronized" } - >, + event: ShellStreamEvent, ): OrchestrationV2ShellSnapshot { if (event.sequence <= snapshot.snapshotSequence) return snapshot; @@ -217,9 +214,9 @@ function batchRows(rows: ReadonlyArray) { }; } -export type ShellStreamEvent = Exclude< +export type ShellStreamEvent = Extract< OrchestrationV2ShellStreamItem, - { readonly kind: "snapshot" } | { readonly kind: "synchronized" } + { readonly sequence: number } >; /** Apply ordered deltas with one index and at most one new array per collection. */ diff --git a/packages/contracts/src/orchestrationV2.ts b/packages/contracts/src/orchestrationV2.ts index 461ec5e4b3f3e..93c45533cbae0 100644 --- a/packages/contracts/src/orchestrationV2.ts +++ b/packages/contracts/src/orchestrationV2.ts @@ -2242,6 +2242,13 @@ export const OrchestrationV2ShellStreamItem = Schema.Union([ Schema.Struct({ kind: Schema.Literal("synchronized"), }), + /** + * The only item of a resume the server cannot replay when the client asked for + * `httpSnapshotFallback`: load the HTTP snapshot and subscribe again from its cursor. + */ + Schema.Struct({ + kind: Schema.Literal("snapshot-required"), + }), Schema.Struct({ kind: Schema.Literal("snapshot"), snapshot: OrchestrationV2ShellSnapshot, @@ -3690,6 +3697,11 @@ export type OrchestrationV2GetThreadProjectionInput = export const OrchestrationV2SubscribeShellInput = Schema.Struct({ afterSequence: Schema.optionalKey(NonNegativeInt), + /** + * With `afterSequence`, a gap the server cannot replay answers `snapshot-required` instead of + * a socket snapshot, which clients without transport compression download over gzip HTTP. + */ + httpSnapshotFallback: Schema.optionalKey(Schema.Literal(true)), requestCompletionMarker: Schema.optionalKey(Schema.Boolean), /** * Snapshots leave out idle subagent children; read them with `getThreadFamily`. Send only to