From 4d0467362b5423990ffb09019ac45cba9b8f029d Mon Sep 17 00:00:00 2001 From: Theo Browne Date: Sat, 24 Oct 2026 20:20:05 -0700 Subject: [PATCH 1/3] perf(shell): new sessions resume the cached thread list instead of downloading it Every new RPC session loaded the full HTTP shell snapshot (1.97 MB, 157 KB gzip on real data), even when the cache was seconds old. A shell now remembers its resume key (the server process key, the client build and the snapshot flags) and when its latest authoritative snapshot was applied. A new session resumes from the cursor when the key matches and that snapshot is under a day old. The server replays a gap without reading payloads: one grouped query finds each touched project and thread, and their current rows are sent once. Co-Authored-By: Claude Opus 5.5 --- .../environment-cache-store.test.ts | 11 +- .../src/connection/environment-cache-store.ts | 36 ++- apps/mobile/src/connection/platform.ts | 8 + .../src/orchestration-v2/ShellStream.test.ts | 14 + .../src/orchestration-v2/ShellStream.ts | 45 ++- .../OrchestrationEventStore.test.ts | 78 ++++- .../persistence/OrchestrationEventStore.ts | 53 ++-- apps/server/src/ws.test.ts | 102 ++++++- apps/server/src/ws.ts | 97 +++--- apps/web/src/connection/platform.ts | 1 + apps/web/src/connection/storage.ts | 8 +- docs/internals/connection-runtime.md | 11 + .../src/connection/registry.test.ts | 6 +- .../src/platform/capabilities.ts | 5 + .../src/platform/orchestrationCache.ts | 11 + .../src/platform/persistence.ts | 9 +- .../client-runtime/src/rpc/client.test.ts | 71 ++++- packages/client-runtime/src/rpc/client.ts | 47 ++- .../src/state/shell-sync.test.ts | 282 +++++++++++++++++- packages/client-runtime/src/state/shell.ts | 259 ++++++++++------ packages/contracts/src/server.ts | 6 + 21 files changed, 936 insertions(+), 224 deletions(-) diff --git a/apps/mobile/src/connection/environment-cache-store.test.ts b/apps/mobile/src/connection/environment-cache-store.test.ts index 4c88a03ccf2a0..51e8bf57d589d 100644 --- a/apps/mobile/src/connection/environment-cache-store.test.ts +++ b/apps/mobile/src/connection/environment-cache-store.test.ts @@ -204,11 +204,14 @@ describe("mobile SQLite environment cache store", () => { const store = yield* make().pipe( Effect.provideService(MobileDatabase.MobileDatabase, memory.database), ); + const resume = { key: "resume-key", syncedAt: 1_753_790_400_000 }; - yield* store.saveShell(ENVIRONMENT_ID, SHELL_SNAPSHOT); + yield* store.saveShell(ENVIRONMENT_ID, SHELL_SNAPSHOT, resume); yield* store.saveThread(ENVIRONMENT_ID, THREAD_SNAPSHOT); - const shell = Option.getOrThrow(yield* store.loadShell(ENVIRONMENT_ID)); + const cached = Option.getOrThrow(yield* store.loadShell(ENVIRONMENT_ID)); + const shell = cached.snapshot; + expect(cached.resume).toEqual(resume); const thread = Option.getOrThrow(yield* store.loadThread(ENVIRONMENT_ID, THREAD_ID)); expect(DateTime.formatIso(shell.threads[0]!.updatedAt)).toBe("2026-07-29T12:00:00.000Z"); @@ -257,7 +260,7 @@ describe("mobile SQLite environment cache store", () => { threads: [linked, unlinked], }); - const shell = Option.getOrThrow(yield* store.loadShell(ENVIRONMENT_ID)); + const shell = Option.getOrThrow(yield* store.loadShell(ENVIRONMENT_ID)).snapshot; expect(shell.threads.map((thread) => thread.pullRequests)).toEqual([undefined, undefined]); const links = yield* shell.loadPullRequests!; expect([...links]).toEqual([[THREAD_ID, [link]]]); @@ -298,7 +301,7 @@ describe("mobile SQLite environment cache store", () => { payload.snapshot.threads[0].pullRequests = [{ number: -1 }]; memory.values.set(id, JSON.stringify(payload)); - const shell = Option.getOrThrow(yield* store.loadShell(ENVIRONMENT_ID)); + const shell = Option.getOrThrow(yield* store.loadShell(ENVIRONMENT_ID)).snapshot; expect(shell.threads.map((thread) => thread.id)).toEqual([THREAD_ID, readable.id]); expect([...(yield* shell.loadPullRequests!)]).toEqual([[readable.id, [link]]]); expect(messages).toHaveLength(1); diff --git a/apps/mobile/src/connection/environment-cache-store.ts b/apps/mobile/src/connection/environment-cache-store.ts index 8df7fc104df26..8d36485cc5de1 100644 --- a/apps/mobile/src/connection/environment-cache-store.ts +++ b/apps/mobile/src/connection/environment-cache-store.ts @@ -111,7 +111,12 @@ 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({ + snapshot: stored.snapshot, + ...(stored.resume === undefined ? {} : { resume: stored.resume }), + }) + : Option.none(), }).pipe( Effect.zipWith( Effect.promise(() => projectFaviconDatabaseCache.hydrate()), @@ -120,16 +125,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, resume) { + const payload = yield* encodeStoredShellSnapshot({ + schemaVersion: ORCHESTRATION_CACHE_SCHEMA_VERSION, + environmentId, + snapshot, + ...(resume === undefined ? {} : { resume }), + }).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/mobile/src/connection/platform.ts b/apps/mobile/src/connection/platform.ts index 52a6106436389..3385737371b58 100644 --- a/apps/mobile/src/connection/platform.ts +++ b/apps/mobile/src/connection/platform.ts @@ -16,8 +16,10 @@ import * as Layer from "effect/Layer"; import * as Option from "effect/Option"; import * as Queue from "effect/Queue"; import * as Stream from "effect/Stream"; +import * as Application from "expo-application"; import Constants from "expo-constants"; import * as Network from "expo-network"; +import * as Updates from "expo-updates"; import { AppState } from "react-native"; import { authClientMetadata } from "../lib/authClientMetadata"; @@ -213,6 +215,12 @@ const layerCapabilities = Layer.effectContext( ClientCapabilities.ClientPresentation, ClientCapabilities.ClientPresentation.of({ metadata: authClientMetadata(Constants.expoConfig?.version), + // Store builds can share a version, and over-the-air updates keep the native build. + build: [ + Constants.expoConfig?.version, + Application.nativeBuildVersion, + Updates.updateId ?? "embedded", + ].join(" "), }), ), Context.add( diff --git a/apps/server/src/orchestration-v2/ShellStream.test.ts b/apps/server/src/orchestration-v2/ShellStream.test.ts index e308c74a00ff9..6a2dade75ad44 100644 --- a/apps/server/src/orchestration-v2/ShellStream.test.ts +++ b/apps/server/src/orchestration-v2/ShellStream.test.ts @@ -25,6 +25,7 @@ import { omitOlderSettledShells, shellStreamItemFromEnrichmentRefresh, shellStreamItemsFromInitialSnapshot, + ShellResumeKey, shouldPublishShellApplicationEvent, skipUnchangedThreadShells, } from "./ShellStream.ts"; @@ -662,3 +663,16 @@ describe("skipUnchangedThreadShells", () => { }), ); }); + +describe("shell resume key", () => { + it.effect("stays one value for the whole server process", () => + Effect.gen(function* () { + // Each socket reads it on its own fiber, which falls back to the cached default. + const read = Effect.gen(function* () { + return yield* ShellResumeKey; + }); + const [first, second] = yield* Effect.all([read, read], { concurrency: "unbounded" }); + expect(second).toBe(first); + }), + ); +}); diff --git a/apps/server/src/orchestration-v2/ShellStream.ts b/apps/server/src/orchestration-v2/ShellStream.ts index 99ac2b2d89d71..0a23fe3cb047a 100644 --- a/apps/server/src/orchestration-v2/ShellStream.ts +++ b/apps/server/src/orchestration-v2/ShellStream.ts @@ -9,19 +9,20 @@ import type { OrchestrationV2ThreadShellSnapshot, OrchestrationV2ShellStreamItem, OrchestrationV2StoredEvent, - ProjectId, - ThreadId, } from "@t3tools/contracts"; import { isLiveSubagentShell, isOlderSettledThreadShell, OrchestrationProjectShell as ProjectShellSchema, OrchestrationV2ThreadShell as ThreadShellSchema, + ProjectId, + ThreadId, } from "@t3tools/contracts"; import * as NodeSqliteClient from "@t3tools/shared/nodeSqliteClient"; import type * as Arr from "effect/Array"; import type * as Cause from "effect/Cause"; import * as Clock from "effect/Clock"; +import * as Context from "effect/Context"; import * as DateTime from "effect/DateTime"; import * as Deferred from "effect/Deferred"; import * as Effect from "effect/Effect"; @@ -31,6 +32,7 @@ import * as Schema from "effect/Schema"; import type * as SqlClient from "effect/sql/SqlClient"; import * as Stream from "effect/Stream"; +import type { ShellReplayHead } from "../persistence/OrchestrationEventStore.ts"; import { bufferLatestLiveStream } from "./LiveStreamBudget.ts"; type ShellDelta = Extract; @@ -282,7 +284,14 @@ export type ShellApplicationEvent = readonly event: Pick; }; -/** Focused subagents receive these updates through their thread subscription. */ +/** Subagent events the shell skips; focused subagents receive them through their thread subscription. */ +export const SUBAGENT_TRANSCRIPT_EVENT_TYPES: ReadonlySet = new Set([ + "node.updated", + "turn-item.updated", + "provider-turn.updated", + "message.updated", +]); + export function shouldPublishShellApplicationEvent( stored: ApplicationStoredEvent, subagentThreadIds: Set, @@ -300,14 +309,22 @@ export function shouldPublishShellApplicationEvent( return true; } return !( - subagentThreadIds.has(event.threadId) && - (event.type === "node.updated" || - event.type === "turn-item.updated" || - event.type === "provider-turn.updated" || - event.type === "message.updated") + subagentThreadIds.has(event.threadId) && SUBAGENT_TRANSCRIPT_EVENT_TYPES.has(event.type) ); } +/** + * `ServerConfig.shellResumeKey`. The default is computed once and cached on the reference, so it + * is one random value per server process; tests provide their own. + */ +export class ShellResumeKey extends Context.Reference( + "@t3tools/server/orchestration-v2/ShellStream/ShellResumeKey", + { + // @effect-diagnostics-next-line cryptoRandomUUID:off - A reference default runs outside Effect. + defaultValue: () => crypto.randomUUID(), + }, +) {} + /** Shell updates refetch an aggregate; drop transcript bodies before retaining an event. */ export function toShellApplicationEvent(stored: ApplicationStoredEvent): ShellApplicationEvent { return "aggregateKind" in stored @@ -320,6 +337,18 @@ export function toShellApplicationEvent(stored: ApplicationStoredEvent): ShellAp : { sequence: stored.sequence, event: { threadId: stored.event.threadId } }; } +export function shellApplicationEventFromReplayHead(head: ShellReplayHead): ShellApplicationEvent { + return head.aggregateKind === "project" + ? { + aggregateKind: "project", + aggregateId: ProjectId.make(head.aggregateId), + // Only a deletion skips the row read; any other project event, even an unknown one, reads it. + type: head.eventType === "project.deleted" ? "project.deleted" : "project.meta-updated", + sequence: head.sequence, + } + : { sequence: head.sequence, event: { threadId: ThreadId.make(head.aggregateId) } }; +} + export function coalesceShellApplicationEvents( events: ReadonlyArray, ): ReadonlyArray { diff --git a/apps/server/src/persistence/OrchestrationEventStore.test.ts b/apps/server/src/persistence/OrchestrationEventStore.test.ts index baad3d5c52a05..9e0ed8976d282 100644 --- a/apps/server/src/persistence/OrchestrationEventStore.test.ts +++ b/apps/server/src/persistence/OrchestrationEventStore.test.ts @@ -209,11 +209,6 @@ layer("OrchestrationEventStore", (it) => { maxEvents: events.length, }); assert.isAbove(stats.rawPayloadBytes, prompt.length * 6); - const applicationStats = yield* store.getReplayStats({ - afterSequence, - throughSequence: yield* store.latestApplicationSequence, - }); - assert.isAbove(applicationStats.rawPayloadBytes, prompt.length * 6); }), ); @@ -456,14 +451,77 @@ layer("OrchestrationEventStore", (it) => { rawPayloadBytes: 0, hasCreateEvent: false, }); - assert.deepEqual( - yield* store.getReplayStats({ afterSequence: baseline, throughSequence: head }), + }), + ); + + it.effect("finds the newest shell event of each project and thread in a range", () => + Effect.gen(function* () { + const store = yield* OrchestrationEventStore.OrchestrationEventStore; + const sql = yield* SqlClient.SqlClient; + const baseline = yield* store.latestApplicationSequence; + const versions = new Map(); + const insert = Effect.fn(function* ( + aggregateKind: "project" | "thread", + streamId: string, + eventType: string, + applicationEventVersion = 2, + ) { + const streamVersion = versions.get(streamId) ?? 0; + versions.set(streamId, streamVersion + 1); + const rows = yield* sql<{ readonly sequence: number }>` + INSERT INTO orchestration_events ( + event_id, aggregate_kind, stream_id, stream_version, event_type, + occurred_at, actor_kind, payload_json, metadata_json, application_event_version + ) VALUES ( + ${`heads:${streamId}:${streamVersion}`}, ${aggregateKind}, ${streamId}, ${streamVersion}, + ${eventType}, '2026-09-04T00:00:00.000Z', 'server', 'not JSON', '{}', + ${applicationEventVersion} + ) RETURNING sequence + `; + return rows[0]!.sequence; + }); + yield* insert("thread", "heads:parent", "thread.created"); + const projectHead = yield* insert("project", "heads:project", "project.deleted"); + const childHead = yield* insert("thread", "heads:child", "thread.metadata-updated"); + yield* insert("thread", "heads:child", "turn-item.updated"); + // Legacy rows are not application events. + yield* insert("thread", "heads:legacy", "thread.metadata-updated", 1); + const parentHead = yield* insert("thread", "heads:parent", "future.event"); + const head = yield* store.latestApplicationSequence; + const read = (input: { readonly afterSequence?: number; readonly limit?: number } = {}) => + store.readShellReplayHeads({ + afterSequence: input.afterSequence ?? baseline, + throughSequence: head, + transcriptEventTypes: ["turn-item.updated"], + skipTranscriptOf: ["heads:child"], + limit: input.limit ?? 10, + }); + + assert.deepEqual(yield* read(), [ { - eventCount: 3, - rawPayloadBytes: - Buffer.byteLength(unicodePayload) + 2 * Buffer.byteLength(oversizedPayload), + aggregateKind: "project", + aggregateId: "heads:project", + sequence: projectHead, + eventType: "project.deleted", + }, + { + aggregateKind: "thread", + aggregateId: "heads:child", + sequence: childHead, + eventType: "thread.metadata-updated", }, + { + aggregateKind: "thread", + aggregateId: "heads:parent", + sequence: parentHead, + eventType: "future.event", + }, + ]); + assert.deepEqual( + (yield* read({ limit: 2 })).map((row) => row.aggregateId), + ["heads:project", "heads:child"], ); + assert.deepEqual(yield* read({ afterSequence: parentHead }), []); }), ); diff --git a/apps/server/src/persistence/OrchestrationEventStore.ts b/apps/server/src/persistence/OrchestrationEventStore.ts index 51c3861a9fca0..5ce6e089db4a0 100644 --- a/apps/server/src/persistence/OrchestrationEventStore.ts +++ b/apps/server/src/persistence/OrchestrationEventStore.ts @@ -39,6 +39,13 @@ export type UnsequencedProjectEvent = ApplicationProjectEvent extends infer Even : never : never; +export interface ShellReplayHead { + readonly aggregateKind: "project" | "thread"; + readonly aggregateId: string; + readonly sequence: number; + readonly eventType: string; +} + export class OrchestrationEventStore extends Context.Service< OrchestrationEventStore, { @@ -76,16 +83,18 @@ export class OrchestrationEventStore extends Context.Service< OrchestrationEventStoreError >; - readonly getReplayStats: (input: { + /** + * The newest application event of each project and thread in the range, oldest first, without + * reading payloads: a shell resume sends each touched aggregate's current row. Events of + * `transcriptEventTypes` on `skipTranscriptOf` threads do not count. + */ + readonly readShellReplayHeads: (input: { readonly afterSequence: number; readonly throughSequence: number; - }) => Effect.Effect< - { - readonly eventCount: number; - readonly rawPayloadBytes: number; - }, - OrchestrationEventStoreError - >; + readonly transcriptEventTypes: ReadonlyArray; + readonly skipTranscriptOf: ReadonlyArray; + readonly limit: number; + }) => Effect.Effect, OrchestrationEventStoreError>; readonly latestAgentSequence: ( threadId?: ThreadId, @@ -616,11 +625,16 @@ const makeEventStore = Effect.gen(function* () { })), ); - const getReplayStats: OrchestrationEventStore["Service"]["getReplayStats"] = (input) => - sql<{ readonly eventCount: number; readonly rawPayloadBytes: number }>` + // With a single MAX aggregate, SQLite takes the bare `event_type` from the row holding the MAX. + const readShellReplayHeads: OrchestrationEventStore["Service"]["readShellReplayHeads"] = ( + input, + ) => + sql` SELECT - COUNT(*) AS "eventCount", - COALESCE(SUM(octet_length(${restoredPayloadJson})), 0) AS "rawPayloadBytes" + aggregate_kind AS "aggregateKind", + stream_id AS "aggregateId", + MAX(sequence) AS "sequence", + event_type AS "eventType" FROM orchestration_events INDEXED BY idx_orchestration_events_application_high_water WHERE sequence > ${input.afterSequence} AND sequence <= ${input.throughSequence} @@ -628,13 +642,16 @@ const makeEventStore = Effect.gen(function* () { aggregate_kind = 'project' OR (application_event_version = 2 AND aggregate_kind = 'thread') ) + AND NOT ( + event_type IN (SELECT value FROM json_each(${JSON.stringify(input.transcriptEventTypes)})) + AND stream_id IN (SELECT value FROM json_each(${JSON.stringify(input.skipTranscriptOf)})) + ) + GROUP BY aggregate_kind, stream_id + ORDER BY MAX(sequence) + LIMIT ${input.limit} `.pipe( NodeSqliteClient.readOnly, - Effect.mapError(toPersistenceSqlError("OrchestrationEventStore.getReplayStats:query")), - Effect.map((rows) => ({ - eventCount: rows[0]?.eventCount ?? 0, - rawPayloadBytes: rows[0]?.rawPayloadBytes ?? 0, - })), + Effect.mapError(toPersistenceSqlError("OrchestrationEventStore.readShellReplayHeads:query")), ); const latestAgentSequence: OrchestrationEventStore["Service"]["latestAgentSequence"] = ( @@ -744,7 +761,7 @@ const makeEventStore = Effect.gen(function* () { appendAgentEvents, readAgentEvents, getAgentReplayStats, - getReplayStats, + readShellReplayHeads, latestAgentSequence, latestApplicationSequence, readApplicationEvents: catchUpApplicationEvents, diff --git a/apps/server/src/ws.test.ts b/apps/server/src/ws.test.ts index a2c3910d4da5f..8a425f024dcd2 100644 --- a/apps/server/src/ws.test.ts +++ b/apps/server/src/ws.test.ts @@ -180,7 +180,7 @@ const ReplayTestLayer = OrchestrationEventStore.layer.pipe( ); it.live.each(["thread", "shell"] as const)( - "%s resume skips unknown types, synchronizes, and receives the live tail", + "%s resume survives unknown types, synchronizes, and receives the live tail", (scope) => Effect.gen(function* () { const store = yield* OrchestrationEventStore.OrchestrationEventStore; @@ -288,11 +288,109 @@ it.live.each(["thread", "shell"] as const)( ); assert.deepEqual( items.flatMap((item) => (item.sequence === undefined ? [] : [item.sequence])), - [after!.sequence, liveSequence], + // The shell decodes no payloads, so an unknown event still counts as a touch. + [scope === "thread" ? after!.sequence : highWater!.sequence, liveSequence], ); }).pipe(Effect.scoped, Effect.provide(ReplayTestLayer)), ); +it.effect("shell resume sends each touched aggregate once, or a snapshot past its limits", () => + Effect.gen(function* () { + const sql = yield* SqlClient.SqlClient; + const enrichment = yield* PubSub.unbounded(); + const versions = new Map(); + const insert = Effect.fn(function* ( + aggregateKind: "project" | "thread", + streamId: string, + eventType: string, + ) { + const streamVersion = versions.get(streamId) ?? 0; + versions.set(streamId, streamVersion + 1); + const [row] = yield* sql<{ sequence: number }>` + INSERT INTO orchestration_events ( + event_id, aggregate_kind, stream_id, stream_version, event_type, + occurred_at, actor_kind, payload_json, metadata_json, application_event_version + ) VALUES ( + ${`resume:${streamId}:${streamVersion}`}, ${aggregateKind}, ${streamId}, ${streamVersion}, + ${eventType}, '2026-01-05T00:00:00.000Z', 'server', '{}', '{}', 2 + ) RETURNING sequence + `; + return row!.sequence; + }); + const resume = (afterSequence: number) => + subscribeOrchestrationV2Shell({ afterSequence, requestCompletionMarker: true }).pipe( + Effect.flatMap((stream) => + stream.pipe( + Stream.takeUntil((item) => item.kind === "synchronized"), + Stream.runCollect, + ), + ), + Effect.map((items) => + Array.from(items, (item) => + item.kind === "thread.updated" + ? `${item.thread.id}@${item.sequence}` + : item.kind === "thread.removed" + ? `-${item.threadId}@${item.sequence}` + : item.kind === "project.updated" + ? `${item.project.id}@${item.sequence}` + : item.kind, + ), + ), + Effect.provideService(ThreadManagement.ThreadManagementService, { + listActiveSubagentThreadIds: () => Effect.succeed([ThreadId.make("thread:subagent")]), + getThreadShell: (threadId: ThreadId) => + Effect.succeed( + threadId === "thread:gone" + ? null + : ({ id: threadId, archivedAt: null } as unknown as OrchestrationV2ThreadShell), + ), + readShellSnapshot: () => + Effect.succeed( + Effect.succeed({ + schemaVersion: 1, + snapshotSequence: 0, + threads: [], + archivedThreads: [], + }), + ), + } as unknown as ThreadManagement.ThreadManagementService["Service"]), + Effect.provideService(ProjectStore.ProjectStoreV2, { + listShells: () => Effect.succeed([]), + } as unknown as ProjectStore.ProjectStoreV2["Service"]), + Effect.provideService(ProjectService.ProjectService, { + getShell: (projectId: ProjectId) => + Effect.succeedSome({ id: projectId } as unknown as OrchestrationProjectShell), + } as unknown as ProjectService.ProjectService["Service"]), + Effect.provideService(ProjectEnrichment.ProjectEnrichmentService, { + subscribeChanges: PubSub.subscribe(enrichment), + } as ProjectEnrichment.ProjectEnrichmentService["Service"]), + ); + + const baseline = yield* insert("thread", "thread:kept", "thread.created"); + const kept = yield* insert("thread", "thread:kept", "thread.metadata-updated"); + const subagentRun = yield* insert("thread", "thread:subagent", "run.updated"); + yield* insert("thread", "thread:subagent", "turn-item.updated"); + const gone = yield* insert("thread", "thread:gone", "thread.deleted"); + assert.deepEqual(yield* resume(baseline), [ + `thread:kept@${kept}`, + `thread:subagent@${subagentRun}`, + `-thread:gone@${gone}`, + "synchronized", + ]); + // A cursor past the head came from another event log. + assert.deepEqual(yield* resume(gone + 1), ["snapshot", "synchronized"]); + + const first = yield* insert("project", "project:0", "project.meta-updated"); + for (let index = 1; index <= 500; index += 1) { + yield* insert("project", `project:${index}`, "project.meta-updated"); + } + assert.deepEqual(yield* resume(gone), ["snapshot", "synchronized"]); + const capped = yield* resume(first); + assert.equal(capped.length, 501); + assert.equal(capped.at(-1), "synchronized"); + }).pipe(Effect.scoped, Effect.provide(ReplayTestLayer)), +); + it.effect("shell snapshots omit idle children and older settled threads only on request", () => Effect.gen(function* () { const shell = ( diff --git a/apps/server/src/ws.ts b/apps/server/src/ws.ts index 4341cd926d395..412cadfacb55d 100644 --- a/apps/server/src/ws.ts +++ b/apps/server/src/ws.ts @@ -133,6 +133,8 @@ import { dedupeShellEnrichment, loadOlderSettledThreadsPage, loadShellSnapshotParts, + shellApplicationEventFromReplayHead, + ShellResumeKey, shellStreamItemFromEnrichmentRefresh, shellStreamItemFromThreadShell, shellStreamItemsFromInitialSnapshot, @@ -140,6 +142,7 @@ import { shellStreamItemsFromResumeSnapshot, shouldPublishShellApplicationEvent, skipUnchangedThreadShells, + SUBAGENT_TRANSCRIPT_EVENT_TYPES, toShellApplicationEvent, type ShellApplicationEvent, } from "./orchestration-v2/ShellStream.ts"; @@ -571,11 +574,10 @@ const isOtherWsRpc = (rpc: ServerWsRpc): rpc is Exclude const serverWsRpcs = Array.from(ServerWsRpcGroup.requests.values()); const SplitWsRpcGroup = RpcGroup.make(...serverWsRpcs.filter(isSplitWsRpc)); const OtherWsRpcGroup = RpcGroup.make(...serverWsRpcs.filter(isOtherWsRpc)); -// Replaying each intervening event costs a shell refetch; past this gap a -// single O(active-threads) snapshot is cheaper and bounded. -const SHELL_RESUME_MAX_GAP = 1_000; -// Row counts do not bound large tool payloads; sum stored JSON bytes in SQL and snapshot above budget. -const ORCHESTRATION_REPLAY_PAYLOAD_BUDGET_BYTES = 8 * 1024 * 1024; +// A shell resume decodes no payloads: one indexed scan of the gap finds the touched projects and +// threads, then their current rows are read. Past either cap a snapshot is about as cheap. +const SHELL_RESUME_MAX_GAP = 100_000; +const SHELL_RESUME_MAX_TOUCHED = 500; function toAuthAccessStreamEvent( change: PairingGrantStore.BootstrapCredentialChange | SessionStore.SessionCredentialChange, @@ -724,35 +726,6 @@ function readClientOwners( ); } -const canReplayPersistedRange = Effect.fnUntraced(function* ( - afterSequence: number, - headSequence: number, - maxGap: number, -) { - const applicationEvents = yield* OrchestrationEventStore.OrchestrationEventStore; - - const replayGap = headSequence - afterSequence; - if (replayGap < 0 || replayGap > maxGap) { - return false; - } - const stats = yield* applicationEvents.getReplayStats({ - afterSequence, - throughSequence: headSequence, - }); - if (stats.rawPayloadBytes > ORCHESTRATION_REPLAY_PAYLOAD_BUDGET_BYTES) { - yield* Effect.logDebug("orchestration replay replaced by snapshot", { - afterSequence, - headSequence, - replayGap, - eventCount: stats.eventCount, - payloadBytes: stats.rawPayloadBytes, - payloadBudgetBytes: ORCHESTRATION_REPLAY_PAYLOAD_BUDGET_BYTES, - }); - return false; - } - return true; -}); - const enrichProjectShells = Effect.fn("ws.orchestrationV2.enrichProjectShells")( (projects: ReadonlyArray) => Effect.flatMap(ProjectEnrichmentService.ProjectEnrichmentService, (projectEnrichment) => @@ -1163,41 +1136,39 @@ export const subscribeOrchestrationV2Shell = Effect.fn("ws.orchestrationV2.subsc const completionThenLive = (afterSequence: number) => Stream.concat(completionMarker, liveFrom(afterSequence)); + const snapshotThenLive = Effect.gen(function* () { + const loaded = yield* loadSnapshot(); + return composeShellStreamWithEnrichment({ + initial: initialSnapshotItems(loaded), + tail: completionThenLive(loaded.snapshot.snapshotSequence), + enrichment: enrichmentRefreshes, + }); + }); + const stream = yield* Effect.gen(function* () { - if (input.afterSequence === undefined) { - const loaded = yield* loadSnapshot(); - return composeShellStreamWithEnrichment({ - initial: initialSnapshotItems(loaded), - tail: completionThenLive(loaded.snapshot.snapshotSequence), - enrichment: enrichmentRefreshes, - }); - } + if (input.afterSequence === undefined) return yield* snapshotThenLive; const highWater = yield* applicationEvents.latestApplicationSequence; - if (!(yield* canReplayPersistedRange(input.afterSequence, highWater, SHELL_RESUME_MAX_GAP))) { - const loaded = yield* loadSnapshot(); - return composeShellStreamWithEnrichment({ - initial: initialSnapshotItems(loaded), - tail: completionThenLive(loaded.snapshot.snapshotSequence), - enrichment: enrichmentRefreshes, - }); - } + const gap = highWater - input.afterSequence; + // A cursor past the head came from another event log. + if (gap < 0 || gap > SHELL_RESUME_MAX_GAP) return yield* snapshotThenLive; subagentThreadIds = new Set(yield* threadManagement.listActiveSubagentThreadIds()); + // Each touched aggregate is sent once with its current row. Event types this build does not + // know count as touches, since nothing is decoded. + const heads = yield* applicationEvents.readShellReplayHeads({ + afterSequence: input.afterSequence, + throughSequence: highWater, + transcriptEventTypes: [...SUBAGENT_TRANSCRIPT_EVENT_TYPES], + skipTranscriptOf: [...subagentThreadIds], + limit: SHELL_RESUME_MAX_TOUCHED + 1, + }); + if (heads.length > SHELL_RESUME_MAX_TOUCHED) return yield* snapshotThenLive; const loaded = yield* loadProjectMetadataSnapshot(highWater); - const replay = toShellStream( - applicationEvents - .readApplicationEvents({ - afterSequence: input.afterSequence, - throughSequence: highWater, - // A row a newer build wrote (e.g. after a downgrade) must not fail - // every shell resubscribe. The live tail below resumes from - // highWater, so a dropped row never hides later events. - skipUnknownEventTypes: true, - }) - .pipe(Stream.filter(filterSubagentTranscript)), - ); + const replay = Stream.fromEffect( + projectShellItems(heads.map(shellApplicationEventFromReplayHead)), + ).pipe(Stream.flatMap(Stream.fromIterable)); return composeShellStreamWithEnrichment({ initial: initialEnrichmentItems(loaded), tail: Stream.concat(Stream.concat(replay, completionMarker), liveFrom(highWater)), @@ -1734,6 +1705,7 @@ export const layerWsRpc = ( const auth = yield* serverAuth.getDescriptor(); const scratchWorkspaceRoot = yield* managedFolders.scratchRoot; const lastEventLoopStall = yield* eventLoopMonitor.latestNoticeableStall; + const shellResumeKey = yield* ShellResumeKey; const editorConfig = yield* resolveEditorConfig( yield* resolveAvailableEditorsForConfig(externalLauncher.resolveAvailableEditors()), resolveFileManagerRevealKindForConfig(externalLauncher.resolveFileManagerRevealKind()), @@ -1782,6 +1754,7 @@ export const layerWsRpc = ( }, settings, shellResumeCompletionMarker: true, + shellResumeKey, shellThreadFamilies: true, shellOlderSettled: true, threadResumeCompletionMarker: true, diff --git a/apps/web/src/connection/platform.ts b/apps/web/src/connection/platform.ts index 0fe0129f11a1c..0ab1b094e806f 100644 --- a/apps/web/src/connection/platform.ts +++ b/apps/web/src/connection/platform.ts @@ -326,6 +326,7 @@ const layerCapabilities = Layer.effectContext( Effect.sync(() => { const presentation = ClientCapabilities.ClientPresentation.of({ metadata: clientMetadata(), + build: APP_VERSION, }); const cloudSession = ClientCapabilities.CloudSession.of({ identity: Effect.sync(() => diff --git a/apps/web/src/connection/storage.ts b/apps/web/src/connection/storage.ts index 621a8020c5a28..45e9566a350bb 100644 --- a/apps/web/src/connection/storage.ts +++ b/apps/web/src/connection/storage.ts @@ -736,7 +736,10 @@ export const layer = Layer.effectContext( Effect.mapError((cause) => persistenceError("load-shell", cause)), Effect.map((stored) => stored.environmentId === environmentId - ? Option.some(stored.snapshot) + ? Option.some({ + snapshot: stored.snapshot, + ...(stored.resume === undefined ? {} : { resume: stored.resume }), + }) : Option.none(), ), ), @@ -746,11 +749,12 @@ export const layer = Layer.effectContext( Effect.mapError((cause) => persistenceError("load-shell", cause)), ), saveShell: Effect.fn("storage.saveShell")( - function* (environmentId, snapshot) { + function* (environmentId, snapshot, resume) { const encoded = yield* encodeStoredShellSnapshot({ schemaVersion: ORCHESTRATION_CACHE_SCHEMA_VERSION, environmentId, snapshot, + ...(resume === undefined ? {} : { resume }), }).pipe(Effect.mapError((cause) => persistenceError("save-shell", cause))); yield* writeDatabaseValue(database, SHELL_STORE_NAME, environmentId, encoded); }, diff --git a/docs/internals/connection-runtime.md b/docs/internals/connection-runtime.md index 1f3df3ba0aff6..6df06901ec026 100644 --- a/docs/internals/connection-runtime.md +++ b/docs/internals/connection-runtime.md @@ -88,6 +88,17 @@ On-demand reads wait for the session's first authoritative snapshot snapshot must still bump `generation` once verified, and may reuse a cached snapshot only when its omissions match what the session asks for. +A new session resumes the cached [shell](../../packages/client-runtime/src/state/shell.ts) +from its cursor instead of loading a snapshot only when the shell was synchronized +against the same resume key less than a day ago. The key combines the server's +`shellResumeKey`, which is new for each server process, the client build, and +the snapshot flags the session sends. So a change to shell rows that emits no +event, such as a migration, must come with a new server process, and a new +snapshot flag must join the one flag object `makeSubscribeInput` sends to both +HTTP and the socket. A row a flag leaves out must return to a fresh snapshot +only through an event: a resumed shell never receives a row that time alone +brings back. + [Thread detail](../../packages/client-runtime/src/state/threads.ts) separates subscription lifetime from cache lifetime. Mounted consumers share one live stream, which stops when the last consumer unmounts; hidden mounted routes still diff --git a/packages/client-runtime/src/connection/registry.test.ts b/packages/client-runtime/src/connection/registry.test.ts index 0d0327c53af28..b19a662f20b0f 100644 --- a/packages/client-runtime/src/connection/registry.test.ts +++ b/packages/client-runtime/src/connection/registry.test.ts @@ -275,7 +275,11 @@ const makeHarness = Effect.fn("TestEnvironmentRegistry.makeHarness")(function* ( const cacheStore = Persistence.EnvironmentCacheStore.of({ loadShell: (environmentId) => Ref.get(shellCache).pipe( - Effect.map((cache) => Option.fromUndefinedOr(cache.get(environmentId))), + Effect.map((cache) => + Option.map(Option.fromUndefinedOr(cache.get(environmentId)), (snapshot) => ({ + snapshot, + })), + ), ), saveShell: (environmentId, snapshot) => Ref.update(shellCache, (current) => { diff --git a/packages/client-runtime/src/platform/capabilities.ts b/packages/client-runtime/src/platform/capabilities.ts index f05c835dda290..c210de96ecb13 100644 --- a/packages/client-runtime/src/platform/capabilities.ts +++ b/packages/client-runtime/src/platform/capabilities.ts @@ -55,6 +55,11 @@ export class ClientPresentation extends Context.Service< ClientPresentation, { readonly metadata: AuthClientPresentationMetadata; + /** + * Changes whenever the client code changes, including over-the-air updates. A cached shell from + * another build is not resumed: a build drops the row fields it does not know when it decodes. + */ + readonly build?: string; } >()("@t3tools/client-runtime/platform/capabilities/ClientPresentation") {} diff --git a/packages/client-runtime/src/platform/orchestrationCache.ts b/packages/client-runtime/src/platform/orchestrationCache.ts index e7800e20bd33a..7cdbaec7d862d 100644 --- a/packages/client-runtime/src/platform/orchestrationCache.ts +++ b/packages/client-runtime/src/platform/orchestrationCache.ts @@ -13,10 +13,21 @@ 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; +/** + * What a cached shell was synchronized against: the session's resume key and the client time of + * its latest authoritative snapshot. Caches from older clients have none and are not resumed. + */ +export const ShellResumeBasis = Schema.Struct({ + key: Schema.String, + syncedAt: Schema.Number, +}); +export type ShellResumeBasis = typeof ShellResumeBasis.Type; + export const StoredOrchestrationShellSnapshot = Schema.Struct({ schemaVersion: Schema.Literal(ORCHESTRATION_CACHE_SCHEMA_VERSION), environmentId: EnvironmentId, snapshot: OrchestrationV2ShellSnapshotJson, + resume: Schema.optionalKey(ShellResumeBasis), }); 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..c052f15d70e53 100644 --- a/packages/client-runtime/src/platform/persistence.ts +++ b/packages/client-runtime/src/platform/persistence.ts @@ -13,6 +13,7 @@ import * as Schema from "effect/Schema"; import type { ConnectionRegistration } from "../connection/catalog.ts"; import type { DeferredShellSnapshot } from "../state/shellPullRequests.ts"; +import type { ShellResumeBasis } from "./orchestrationCache.ts"; import type { ConnectionTarget, PersistedConnectionTarget } from "../connection/model.ts"; export class ConnectionPersistenceError extends Schema.TaggedError()( @@ -77,15 +78,21 @@ export class ConnectionRegistrationStore extends Context.Service< } >()("@t3tools/client-runtime/platform/persistence/ConnectionRegistrationStore") {} +export interface CachedEnvironmentShell { + readonly snapshot: DeferredShellSnapshot; + readonly resume?: ShellResumeBasis; +} + export class EnvironmentCacheStore extends Context.Service< EnvironmentCacheStore, { readonly loadShell: ( environmentId: EnvironmentId, - ) => Effect.Effect, ConnectionPersistenceError>; + ) => Effect.Effect, ConnectionPersistenceError>; readonly saveShell: ( environmentId: EnvironmentId, snapshot: OrchestrationV2ShellSnapshot, + resume?: ShellResumeBasis, ) => Effect.Effect; readonly loadThread: ( environmentId: EnvironmentId, diff --git a/packages/client-runtime/src/rpc/client.test.ts b/packages/client-runtime/src/rpc/client.test.ts index bbd86f2f79112..88361b855dc0c 100644 --- a/packages/client-runtime/src/rpc/client.test.ts +++ b/packages/client-runtime/src/rpc/client.test.ts @@ -26,7 +26,13 @@ import { import * as EnvironmentSupervisor from "../connection/supervisor.ts"; import * as RpcSession from "../rpc/session.ts"; import type { WsRpcProtocolClient } from "../rpc/protocol.ts"; -import { runStream, subscribe, subscribeDynamicWithSession } from "./client.ts"; +import { + runStream, + subscribe, + subscribeDynamicWithAttempt, + subscribeDynamicWithSession, + type SubscriptionAttempt, +} from "./client.ts"; const TARGET = new PrimaryConnectionTarget({ environmentId: EnvironmentId.make("environment-1"), @@ -219,6 +225,69 @@ describe("environment RPC", () => { }), ); + it.effect("keeps the producer attempt on an old value buffered across a resubscribe", () => + Effect.gen(function* () { + const subscribed = yield* Queue.unbounded(); + const resubscribe = yield* Queue.unbounded(); + const firstValueBlocked = yield* Deferred.make(); + const releaseFirstValue = yield* Deferred.make(); + const firstValue = { attempt: 0, index: 1 } as unknown as ServerLifecycleStreamEvent; + const bufferedFirstValue = { attempt: 0, index: 2 } as unknown as ServerLifecycleStreamEvent; + const secondValue = { attempt: 1, index: 1 } as unknown as ServerLifecycleStreamEvent; + let calls = 0; + const client = { + [WS_METHODS.subscribeServerLifecycle]: () => { + const call = calls++; + return Stream.fromEffect(Queue.offer(subscribed, call)).pipe( + Stream.drain, + Stream.concat( + Stream.fromIterable(call === 0 ? [firstValue, bufferedFirstValue] : [secondValue]), + ), + Stream.concat(Stream.never), + ); + }, + } as unknown as WsRpcProtocolClient; + const { activeSession, supervisor } = yield* makeHarness(); + const attempts: Array = []; + + const resultFiber = yield* subscribeDynamicWithAttempt( + WS_METHODS.subscribeServerLifecycle, + (attempt) => Effect.sync(() => attempts.push(attempt)).pipe(Effect.as({})), + { resubscribe: Stream.fromQueue(resubscribe) }, + ).pipe( + Stream.mapEffect(([attempt, value]) => + value === firstValue + ? Deferred.succeed(firstValueBlocked, undefined).pipe( + Effect.andThen(Deferred.await(releaseFirstValue)), + Effect.as([attempt, value] as const), + ) + : Effect.succeed([attempt, value] as const), + ), + Stream.take(3), + Stream.runCollect, + Effect.provideService(EnvironmentSupervisor.EnvironmentSupervisor, supervisor), + Effect.forkChild, + ); + + yield* SubscriptionRef.set(activeSession, Option.some(session(client))); + expect(yield* Queue.take(subscribed)).toBe(0); + yield* Deferred.await(firstValueBlocked); + yield* Queue.offer(resubscribe, undefined); + expect(yield* Queue.take(subscribed)).toBe(1); + yield* Deferred.succeed(releaseFirstValue, undefined); + + const result = yield* Fiber.join(resultFiber); + const [first, second] = attempts; + // One session, two attempts: only the attempt tells the buffered value apart. + expect(second?.session).toBe(first?.session); + expect(result).toEqual([ + [first, firstValue], + [first, bufferedFirstValue], + [second, secondValue], + ]); + }), + ); + it.effect("keeps durable subscriptions alive across a transport failure and new session", () => Effect.gen(function* () { const subscriptions: string[] = []; diff --git a/packages/client-runtime/src/rpc/client.ts b/packages/client-runtime/src/rpc/client.ts index c364b3448d03d..296a344e0919e 100644 --- a/packages/client-runtime/src/rpc/client.ts +++ b/packages/client-runtime/src/rpc/client.ts @@ -248,11 +248,19 @@ interface SubscriptionOptions { readonly resubscribe?: Stream.Stream; } +/** + * One subscribe call on one session. A session change, a foreground resubscribe and a retry after + * an expected failure each start a new attempt. + */ +export interface SubscriptionAttempt { + readonly session: RpcSession; +} + function subscribeDynamicMapped( tag: TTag, - makeInput: (session: RpcSession) => Effect.Effect>, + makeInput: (attempt: SubscriptionAttempt) => Effect.Effect>, mapStream: ( - session: RpcSession, + attempt: SubscriptionAttempt, stream: Stream.Stream, EnvironmentRpcStreamFailure>, ) => Stream.Stream>, options?: SubscriptionOptions, @@ -295,13 +303,14 @@ function subscribeDynamicMapped( Stream.suspend(() => Stream.unwrap( Effect.gen(function* () { - const input = yield* makeInput(session); + const attempt: SubscriptionAttempt = { session }; + const input = yield* makeInput(attempt); const completeObservation = yield* observer.observe({ environmentId: supervisor.target.environmentId, method: tag, input, }); - const stream = mapStream(session, method(input)).pipe( + const stream = mapStream(attempt, method(input)).pipe( Stream.onFirst(() => Effect.sync(() => { expectedFailureRetries = 0; @@ -399,7 +408,12 @@ export function subscribeDynamic( EnvironmentRpcStreamFailure, EnvironmentSupervisor.EnvironmentSupervisor > { - return subscribeDynamicMapped(tag, makeInput, (_session, stream) => stream, options); + return subscribeDynamicMapped( + tag, + (attempt) => makeInput(attempt.session), + (_attempt, stream) => stream, + options, + ); } /** Tags each value before `switchMap` can buffer it across a session change. */ @@ -411,11 +425,32 @@ export function subscribeDynamicWithSession], EnvironmentRpcStreamFailure, EnvironmentSupervisor.EnvironmentSupervisor +> { + return subscribeDynamicMapped( + tag, + (attempt) => makeInput(attempt.session), + (attempt, stream) => stream.pipe(Stream.map((value) => [attempt.session, value] as const)), + options, + ); +} + +/** + * Tags each value with its attempt before `switchMap` can buffer it into a later attempt, which + * matters even on one session: a resubscribe may load a newer snapshot than buffered values. + */ +export function subscribeDynamicWithAttempt( + tag: TTag, + makeInput: (attempt: SubscriptionAttempt) => Effect.Effect>, + options?: SubscriptionOptions, +): Stream.Stream< + readonly [attempt: SubscriptionAttempt, value: EnvironmentRpcStreamValue], + EnvironmentRpcStreamFailure, + EnvironmentSupervisor.EnvironmentSupervisor > { return subscribeDynamicMapped( tag, makeInput, - (session, stream) => stream.pipe(Stream.map((value) => [session, value] as const)), + (attempt, stream) => stream.pipe(Stream.map((value) => [attempt, value] as const)), options, ); } diff --git a/packages/client-runtime/src/state/shell-sync.test.ts b/packages/client-runtime/src/state/shell-sync.test.ts index 5a5a50ab21105..31356a6053b33 100644 --- a/packages/client-runtime/src/state/shell-sync.test.ts +++ b/packages/client-runtime/src/state/shell-sync.test.ts @@ -1,6 +1,7 @@ import { EnvironmentId, ORCHESTRATION_V2_WS_METHODS, + OrchestrationV2GetShellSnapshotError, type OrchestrationV2ShellSnapshot, type OrchestrationV2ShellStreamItem, ThreadId, @@ -25,6 +26,7 @@ import { type PreparedConnection, } from "../connection/model.ts"; import * as EnvironmentSupervisor from "../connection/supervisor.ts"; +import * as ClientCapabilities from "../platform/capabilities.ts"; import * as Persistence from "../platform/persistence.ts"; import * as RpcSession from "../rpc/session.ts"; import type { WsRpcProtocolClient } from "../rpc/protocol.ts"; @@ -53,10 +55,13 @@ const LIVE_SHELL_SNAPSHOT: OrchestrationV2ShellSnapshot = { snapshotSequence: 1, }; -function session(client: WsRpcProtocolClient): RpcSession.RpcSession { +function session( + client: WsRpcProtocolClient, + config: object = { shellResumeCompletionMarker: true }, +): RpcSession.RpcSession { return { client, - initialConfig: Effect.succeed({ shellResumeCompletionMarker: true } as never), + initialConfig: Effect.succeed(config as never), subscribeServerConfig: (input) => client.subscribeServerConfig(input), ready: Effect.void, probe: Effect.void, @@ -121,6 +126,112 @@ const makePersistenceHarness = Effect.fn("test.makeShellPersistenceHarness")(fun return { state, send }; }); +const RESUMABLE_CONFIG = { + shellResumeCompletionMarker: true, + shellThreadFamilies: true, + shellResumeKey: "server-1", +}; + +/** + * A shell state on one session whose socket replays `events`. The HTTP loader answers at + * `httpSequence`, and saves are recorded. + */ +const makeResumeHarness = Effect.fn(function* (input: { + readonly cache?: Persistence.CachedEnvironmentShell; + readonly config?: object; + readonly build?: string; + readonly httpSequence: number; +}) { + const events = yield* Queue.unbounded< + OrchestrationV2ShellStreamItem | OrchestrationV2GetShellSnapshotError + >(); + const subscriptions = yield* Queue.unbounded(); + const saved = yield* Ref.make>([]); + const loads = yield* Ref.make(0); + const client = { + [ORCHESTRATION_V2_WS_METHODS.subscribeShell]: (subscribeInput: unknown) => + Stream.unwrap( + Queue.offer(subscriptions, subscribeInput).pipe( + Effect.as( + Stream.fromQueue(events).pipe( + Stream.mapEffect((item) => + "_tag" in item ? Effect.fail(item) : Effect.succeed(item), + ), + ), + ), + ), + ), + } as unknown as WsRpcProtocolClient; + const activeSession = yield* SubscriptionRef.make>( + Option.some(session(client, input.config ?? RESUMABLE_CONFIG)), + ); + 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.succeed(Option.fromUndefinedOr(input.cache)), + saveShell: (_environmentId, snapshot, resume) => + Ref.update(saved, (all) => [ + ...all, + { snapshot, ...(resume === undefined ? {} : { resume }) }, + ]), + 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.update(loads, (count) => count + 1).pipe( + Effect.as(Option.some({ ...LIVE_SHELL_SNAPSHOT, snapshotSequence: input.httpSequence })), + ), + }); + const presentation = ClientCapabilities.ClientPresentation.of({ + metadata: { label: "Test" } as never, + ...(input.build === undefined ? {} : { build: input.build }), + }); + const shellState = yield* makeEnvironmentShellState().pipe( + Effect.provideService(EnvironmentSupervisor.EnvironmentSupervisor, supervisor), + Effect.provideService(Persistence.EnvironmentCacheStore, cache), + Effect.provideService(ShellSnapshotLoader.ShellSnapshotLoader, snapshotLoader), + Effect.provideService(ClientCapabilities.ClientPresentation, presentation), + ); + const stateWhere = (predicate: (state: EnvironmentShellState) => boolean) => + SubscriptionRef.changes(shellState).pipe( + Stream.filter(predicate), + Stream.runHead, + Effect.map(Option.getOrThrow), + ); + return { activeSession, client, events, loads, saved, shellState, stateWhere, subscriptions }; +}); + +/** Runs a client once against a resumable server and returns the shell it saved on close. */ +const savedResumableShell = Effect.fn(function* () { + const scope = yield* Scope.make(); + const before = yield* makeResumeHarness({ build: "build-1", httpSequence: 10 }).pipe( + Scope.provide(scope), + ); + yield* Queue.take(before.subscriptions); + yield* Queue.offer(before.events, { kind: "synchronized" }); + yield* before.stateWhere((state) => state.status === "live"); + yield* Scope.close(scope, Exit.void); + return (yield* Ref.get(before.saved)).at(-1)!; +}); + +const RESUMED_INPUT = { afterSequence: 10, omitIdleSubagents: true, requestCompletionMarker: true }; + describe("environment shell synchronization", () => { it.effect("skips unchanged shell writes, persists resets and flushes changed rows", () => Effect.gen(function* () { @@ -593,7 +704,7 @@ describe("environment shell synchronization", () => { retryNow: Effect.void, } satisfies EnvironmentSupervisor.EnvironmentSupervisor["Service"]); const cache = Persistence.EnvironmentCacheStore.of({ - loadShell: () => Effect.succeed(Option.some(cachedSnapshot)), + loadShell: () => Effect.succeedSome({ snapshot: cachedSnapshot }), saveShell: () => Effect.void, loadThread: () => Effect.succeed(Option.none()), saveThread: () => Effect.void, @@ -759,15 +870,17 @@ describe("environment shell synchronization", () => { const cache = Persistence.EnvironmentCacheStore.of({ loadShell: () => Effect.succeedSome({ - ...cachedSnapshot, - loadPullRequests: Deferred.await(linksReady).pipe( - Effect.as( - new Map([ - [v2ThreadShell.id, [link(1)]], - [otherThreadId, [link(2)]], - ]), + snapshot: { + ...cachedSnapshot, + loadPullRequests: Deferred.await(linksReady).pipe( + Effect.as( + new Map([ + [v2ThreadShell.id, [link(1)]], + [otherThreadId, [link(2)]], + ]), + ), ), - ), + }, }), saveShell: () => Effect.void, loadThread: () => Effect.succeedNone, @@ -909,4 +1022,151 @@ describe("environment shell synchronization", () => { expect(filled).toEqual([link(2)]); }), ); + it.effect("a restarted client resumes its saved shell without HTTP and saves nothing new", () => + Effect.gen(function* () { + const cached = yield* savedResumableShell(); + expect(cached.resume?.key).toEqual(expect.any(String)); + + const scope = yield* Scope.make(); + const after = yield* makeResumeHarness({ + cache: cached, + build: "build-1", + httpSequence: 20, + }).pipe(Scope.provide(scope)); + expect(yield* Queue.take(after.subscriptions)).toEqual(RESUMED_INPUT); + expect(yield* Ref.get(after.loads)).toBe(0); + // Thread families wait while the resumed replay runs. + expect((yield* SubscriptionRef.get(after.shellState)).generation).toBe(0); + yield* Queue.offer(after.events, { kind: "synchronized" }); + const live = yield* after.stateWhere((state) => state.status === "live"); + expect(live.generation).toBeGreaterThan(0); + yield* Scope.close(scope, Exit.void); + expect(yield* Ref.get(after.saved)).toEqual([]); + }), + ); + + it.effect.each([ + { name: "another server process", config: { ...RESUMABLE_CONFIG, shellResumeKey: "server-2" } }, + { name: "another client build", build: "build-2" }, + { name: "other snapshot flags", config: { ...RESUMABLE_CONFIG, shellThreadFamilies: false } }, + { + name: "older settled threads left out", + config: { ...RESUMABLE_CONFIG, shellOlderSettled: true }, + }, + { + name: "a server without completion markers", + config: { ...RESUMABLE_CONFIG, shellResumeCompletionMarker: false }, + }, + { name: "an older client's cache", dropResume: true }, + { name: "a basis 24 hours old", elapsed: "24 hours" as const }, + { name: "a basis from the future", future: true }, + ])("loads HTTP instead of resuming $name", (scenario) => + Effect.gen(function* () { + const saved = yield* savedResumableShell(); + const cached: Persistence.CachedEnvironmentShell = + scenario.dropResume === true + ? { snapshot: saved.snapshot } + : scenario.future === true + ? { ...saved, resume: { ...saved.resume!, syncedAt: saved.resume!.syncedAt + 60_000 } } + : saved; + if (scenario.elapsed !== undefined) yield* TestClock.adjust(scenario.elapsed); + const after = yield* makeResumeHarness({ + cache: cached, + build: scenario.build ?? "build-1", + httpSequence: 20, + ...(scenario.config === undefined ? {} : { config: scenario.config }), + }); + expect(yield* Queue.take(after.subscriptions)).toMatchObject({ afterSequence: 20 }); + expect(yield* Ref.get(after.loads)).toBe(1); + }), + ); + + it.effect("a resumed shell counts as authoritative only once its replay completes", () => + Effect.gen(function* () { + const cached = yield* savedResumableShell(); + const after = yield* makeResumeHarness({ cache: cached, build: "build-1", httpSequence: 20 }); + expect(yield* Queue.take(after.subscriptions)).toEqual(RESUMED_INPUT); + yield* Queue.offer(after.events, { + kind: "thread.removed", + sequence: 11, + location: "active", + threadId: v2ThreadShell.id, + }); + const replayed = yield* after.stateWhere((state) => + Option.exists(state.snapshot, (snapshot) => snapshot.threads.length === 0), + ); + expect(replayed.generation).toBe(0); + yield* Queue.offer(after.events, { kind: "synchronized" }); + const live = yield* after.stateWhere((state) => state.status === "live"); + expect(live.generation).toBeGreaterThan(0); + // No family read settles at generation 0, so the new generation starts with no removals. + expect(live.removedAt.size).toBe(0); + }), + ); + + it.effect("errors after a completed resume retry from the cursor without HTTP", () => + Effect.gen(function* () { + const cached = yield* savedResumableShell(); + const after = yield* makeResumeHarness({ cache: cached, build: "build-1", httpSequence: 20 }); + const failure = new OrchestrationV2GetShellSnapshotError({ message: "stream failed" }); + expect(yield* Queue.take(after.subscriptions)).toEqual(RESUMED_INPUT); + yield* Queue.offer(after.events, { kind: "synchronized" }); + yield* after.stateWhere((state) => state.status === "live"); + // After the marker the session holds a complete shell, so neither error drops it. + for (const retryAfter of ["250 millis", "500 millis"] as const) { + yield* Queue.offer(after.events, failure); + yield* after.stateWhere((state) => Option.isSome(state.error)); + yield* TestClock.adjust(retryAfter); + expect(yield* Queue.take(after.subscriptions)).toEqual(RESUMED_INPUT); + } + // The second error came before that attempt's marker; the next session still resumes. + yield* SubscriptionRef.set( + after.activeSession, + Option.some(session(after.client, RESUMABLE_CONFIG)), + ); + expect(yield* Queue.take(after.subscriptions)).toEqual(RESUMED_INPUT); + expect(yield* Ref.get(after.loads)).toBe(0); + }), + ); + + it.effect("a resume that fails before its marker loads HTTP on the retry", () => + Effect.gen(function* () { + const cached = yield* savedResumableShell(); + const after = yield* makeResumeHarness({ cache: cached, build: "build-1", httpSequence: 20 }); + expect(yield* Queue.take(after.subscriptions)).toEqual(RESUMED_INPUT); + // The metadata refresh that opens every resume is not an authoritative snapshot. + yield* Queue.offer(after.events, { + kind: "snapshot", + snapshot: { ...cached.snapshot, threads: [] }, + resolvedRepositoryIdentityRoots: [], + }); + yield* Queue.offer( + after.events, + new OrchestrationV2GetShellSnapshotError({ message: "replay failed" }), + ); + yield* after.stateWhere((state) => Option.isSome(state.error)); + yield* TestClock.adjust("250 millis"); + expect(yield* Queue.take(after.subscriptions)).toMatchObject({ afterSequence: 20 }); + expect(yield* Ref.get(after.loads)).toBe(1); + }), + ); + + it.effect("a new session on the same server resumes the in-memory shell and keeps families", () => + Effect.gen(function* () { + const harness = yield* makeResumeHarness({ build: "build-1", httpSequence: 10 }); + expect(yield* Queue.take(harness.subscriptions)).toEqual(RESUMED_INPUT); + yield* Queue.offer(harness.events, { kind: "synchronized" }); + const { generation } = yield* harness.stateWhere((state) => state.status === "live"); + + yield* SubscriptionRef.set( + harness.activeSession, + Option.some(session(harness.client, RESUMABLE_CONFIG)), + ); + expect(yield* Queue.take(harness.subscriptions)).toEqual(RESUMED_INPUT); + expect(yield* Ref.get(harness.loads)).toBe(1); + yield* Queue.offer(harness.events, { kind: "synchronized" }); + const resumed = yield* harness.stateWhere((state) => state.status === "live"); + expect(resumed.generation).toBe(generation); + }), + ); }); diff --git a/packages/client-runtime/src/state/shell.ts b/packages/client-runtime/src/state/shell.ts index 079590d99bab2..ab1c4bc66d4b7 100644 --- a/packages/client-runtime/src/state/shell.ts +++ b/packages/client-runtime/src/state/shell.ts @@ -9,11 +9,13 @@ 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"; import * as Queue from "effect/Queue"; import * as Ref from "effect/Ref"; +import * as Schema from "effect/Schema"; import * as Semaphore from "effect/Semaphore"; import * as Stream from "effect/Stream"; import * as SubscriptionRef from "effect/SubscriptionRef"; @@ -24,9 +26,11 @@ import { connectionProjectionPhase } from "../connection/model.ts"; import * as EnvironmentSupervisor from "../connection/supervisor.ts"; import * as ConnectionWakeups from "../connection/wakeups.ts"; import { safeErrorLogAttributes } from "../errors/safeLog.ts"; +import * as ClientCapabilities from "../platform/capabilities.ts"; +import type { ShellResumeBasis } from "../platform/orchestrationCache.ts"; import * as Persistence from "../platform/persistence.ts"; import { runCachePersistence } from "./cachePersistence.ts"; -import { subscribeDynamic } from "../rpc/client.ts"; +import { subscribeDynamicWithAttempt, type SubscriptionAttempt } from "../rpc/client.ts"; import type { RpcSession } from "../rpc/session.ts"; import * as ShellSnapshotLoader from "./shellSnapshotHttp.ts"; import type { DeferredShellSnapshot } from "./shellPullRequests.ts"; @@ -46,18 +50,21 @@ export interface EnvironmentShellState { readonly status: EnvironmentShellStatus; readonly error: Option.Option; /** - * Changes with each authoritative snapshot (not enrichment refreshes); 0 until the first one. - * Thread families read for an older generation may miss changes from before the latest snapshot, - * so they are read again. + * Changes with each authoritative snapshot (not enrichment refreshes) and when the resume of a + * cached shell completes; 0 until the first. Thread families read for an older generation may + * miss changes from before it, so they are read again. */ readonly generation: number; - /** Sequences of the thread removals applied since the latest authoritative snapshot. */ + /** Sequences of the thread removals applied since the generation last changed. */ readonly removedAt: ReadonlyMap; } const EMPTY_REMOVALS: ReadonlyMap = new Map(); // Shared by every shell state, so a rebuilt one never reuses a generation another already used. let lastShellGeneration = 0; +// A new session resumes a shell only this long after its latest authoritative snapshot, so rows a +// fresh snapshot would leave out (idle subagent children, settled threads) do not pile up. +const SHELL_RESUME_MAX_AGE_MS = 24 * 60 * 60 * 1000; export const EMPTY_SHELL_STATE: EnvironmentShellState = { snapshot: Option.none(), @@ -79,6 +86,36 @@ function withoutDeferredPullRequests(value: DeferredShellSnapshot): Orchestratio return snapshot; } +/** A snapshot to save, kept with what it was synchronized against. */ +interface PersistedShell { + readonly snapshot: OrchestrationV2ShellSnapshot; + readonly resume: ShellResumeBasis | undefined; +} + +function sameResumeBasis( + left: ShellResumeBasis | undefined, + right: ShellResumeBasis | undefined, +): boolean { + return left?.key === right?.key && left?.syncedAt === right?.syncedAt; +} + +function samePersistedShell(left: PersistedShell, right: PersistedShell | undefined): boolean { + return left.snapshot === right?.snapshot && sameResumeBasis(left.resume, right.resume); +} + +/** The subscription attempt whose items apply. */ +interface ActiveAttempt { + readonly attempt: SubscriptionAttempt; + readonly resumeKey: string | undefined; + readonly synchronized: boolean; +} + +const encodeResumeKey = Schema.encodeSync( + Schema.fromJsonString( + Schema.Struct({ server: Schema.String, client: Schema.String, selection: Schema.Unknown }), + ), +); + const SHELL_SYNCHRONIZATION_ERROR_MESSAGE = "Could not synchronize environment data."; export const makeEnvironmentShellState = Effect.fn("EnvironmentShellState.make")(function* () { @@ -86,6 +123,9 @@ export const makeEnvironmentShellState = Effect.fn("EnvironmentShellState.make") const cache = yield* Persistence.EnvironmentCacheStore; const snapshotLoader = yield* ShellSnapshotLoader.ShellSnapshotLoader; const wakeups = yield* Effect.serviceOption(ConnectionWakeups.ConnectionWakeups); + const build = Option.getOrUndefined( + yield* Effect.serviceOption(ClientCapabilities.ClientPresentation), + )?.build; const environmentId = supervisor.target.environmentId; const cached = yield* cache.loadShell(environmentId).pipe( Effect.catch((error) => @@ -94,26 +134,32 @@ 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 cachedSnapshot = Option.map(cached, (value) => withoutDeferredPullRequests(value.snapshot)); + const cachedResume = Option.getOrUndefined(cached)?.resume; const state = yield* SubscriptionRef.make({ ...EMPTY_SHELL_STATE, snapshot: cachedSnapshot, status: shellStatusForSnapshot(cachedSnapshot), }); const awaitingCompletion = yield* Ref.make(false); + // The session whose own authoritative snapshot or completed resume the shell holds. const lastAuthoritativeSession = yield* Ref.make(null); - const activeSubscriptionSession = yield* Ref.make(null); - const latestLiveSnapshot = yield* Ref.make>( - Option.none(), - ); + const activeAttempt = yield* Ref.make(null); + // What the shell was synchronized against; another session resumes it only with the same key. + const resumeBasis = yield* Ref.make(cachedResume); + const latestLiveShell = yield* Ref.make>(Option.none()); const persistence = yield* Queue.sliding(1); const persistenceLock = yield* Semaphore.make(1); - let lastPersisted: OrchestrationV2ShellSnapshot | undefined; + // The loaded cache is already saved, so a resume that changes nothing writes nothing. + let lastPersisted: PersistedShell | undefined = Option.match(cachedSnapshot, { + onNone: () => undefined, + onSome: (snapshot) => ({ snapshot, resume: cachedResume }), + }); // 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(); @@ -121,24 +167,27 @@ export const makeEnvironmentShellState = Effect.fn("EnvironmentShellState.make") flush: boolean, ) { while (true) { - const latest = yield* Ref.get(latestLiveSnapshot); - if (Option.isNone(latest) || latest.value === lastPersisted) return; - if (latest.value.threads.some((thread) => pendingLinkRows.has(thread))) return; - const snapshot = latest.value; + const latest = yield* Ref.get(latestLiveShell); + if (Option.isNone(latest) || samePersistedShell(latest.value, lastPersisted)) return; + if (latest.value.snapshot.threads.some((thread) => pendingLinkRows.has(thread))) return; + const shell = latest.value; + const { snapshot } = shell; + const previous = lastPersisted?.snapshot; if ( - lastPersisted !== undefined && - snapshot.snapshotSequence >= lastPersisted.snapshotSequence && - snapshot.schemaVersion === lastPersisted.schemaVersion && - snapshot.projects === lastPersisted.projects && - snapshot.threads === lastPersisted.threads && - snapshot.archivedThreads === lastPersisted.archivedThreads + previous !== undefined && + sameResumeBasis(shell.resume, lastPersisted?.resume) && + snapshot.snapshotSequence >= previous.snapshotSequence && + snapshot.schemaVersion === previous.schemaVersion && + snapshot.projects === previous.projects && + snapshot.threads === previous.threads && + snapshot.archivedThreads === previous.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 = shell; return; } - const saved = yield* cache.saveShell(environmentId, snapshot).pipe( + const saved = yield* cache.saveShell(environmentId, snapshot, shell.resume).pipe( Effect.as(true), Effect.catch((error) => Effect.logWarning("Could not persist environment shell cache.").pipe( @@ -148,7 +197,7 @@ export const makeEnvironmentShellState = Effect.fn("EnvironmentShellState.make") ), ); if (!saved) return; - lastPersisted = snapshot; + lastPersisted = shell; if (!flush) return; } }, persistenceLock.withPermit); @@ -156,7 +205,7 @@ export const makeEnvironmentShellState = Effect.fn("EnvironmentShellState.make") const flushLiveShellSnapshot = persistLatest(true); // Register before scoped worker fibers so reverse finalizer order interrupts - // those fibers first and this flush sees a stable latestLiveSnapshot. + // those fibers first and this flush sees a stable latestLiveShell. yield* Effect.addFinalizer(() => flushLiveShellSnapshot); yield* runCachePersistence(persistence, persist).pipe(Effect.forkScoped); @@ -187,6 +236,14 @@ export const makeEnvironmentShellState = Effect.fn("EnvironmentShellState.make") ); const setStreamError = (error: unknown) => Ref.set(awaitingCompletion, false).pipe( + // A resume through the basis that fails before its marker may fail every time, so the next + // attempt loads HTTP. Memory only: the saved pair keeps its key until the next save. + Effect.andThen(Effect.all([Ref.get(activeAttempt), Ref.get(lastAuthoritativeSession)])), + Effect.flatMap(([active, credited]) => + active === null || active.synchronized || active.attempt.session === credited + ? Effect.void + : Ref.set(resumeBasis, undefined), + ), Effect.andThen(Effect.logWarning("Could not synchronize the environment shell.")), Effect.annotateLogs({ environmentId, @@ -206,18 +263,37 @@ export const makeEnvironmentShellState = Effect.fn("EnvironmentShellState.make") const snapshotWriteLock = yield* Semaphore.make(1); const applyItems = Effect.fn("EnvironmentShellState.applyItems")(function* ( - items: ReadonlyArray, + items: ReadonlyArray, pendingLinkIds?: ReadonlySet, ) { + const active = yield* Ref.get(activeAttempt); + if (active === null) return; const initial = yield* SubscriptionRef.get(state); let waiting = yield* Ref.get(awaitingCompletion); + const initialBasis = yield* Ref.get(resumeBasis); + let basis = initialBasis; + let synchronized = active.synchronized; + let credited = false; let next = initial; - let receivedSnapshot = false; - for (const item of items) { + for (const [attempt, item] of items) { + // Buffered from a replaced attempt, which started from an older cursor or snapshot; the + // active attempt subscribed from the cursor before it, so nothing is lost. + if (attempt !== active.attempt) continue; if (item.kind === "synchronized") { waiting = false; + synchronized = true; if (Option.isSome(next.snapshot)) { - next = { ...next, status: "live", error: Option.none() }; + credited = true; + next = { + ...next, + status: "live", + error: Option.none(), + // A resumed cache is authoritative once its replay completes. No family read settles + // at generation 0, so clearing the removals loses nothing. + ...(next.generation === 0 + ? { generation: ++lastShellGeneration, removedAt: EMPTY_REMOVALS } + : {}), + }; } continue; } @@ -242,7 +318,13 @@ export const makeEnvironmentShellState = Effect.fn("EnvironmentShellState.make") : snapshot, }); if (nextSnapshot === null) continue; - receivedSnapshot ||= item.kind === "snapshot"; + if (authoritative) { + credited = true; + basis = + active.resumeKey === undefined + ? undefined + : { key: active.resumeKey, syncedAt: yield* Clock.currentTimeMillis }; + } const removal = item.kind === "thread.removed" && Option.exists(next.snapshot, (snapshot) => item.sequence > snapshot.snapshotSequence) @@ -261,6 +343,12 @@ export const makeEnvironmentShellState = Effect.fn("EnvironmentShellState.make") }; } yield* Ref.set(awaitingCompletion, waiting); + // Written only when a snapshot set it, so a stream error that cleared it meanwhile stands. + if (basis !== initialBasis) yield* Ref.set(resumeBasis, basis); + if (synchronized !== active.synchronized) { + yield* Ref.set(activeAttempt, { ...active, synchronized }); + } + if (credited) yield* Ref.set(lastAuthoritativeSession, active.attempt.session); if (next === initial) return; if (pendingLinkIds !== undefined && Option.isSome(next.snapshot)) { pendingLinkRows = new Set( @@ -268,15 +356,12 @@ export const makeEnvironmentShellState = Effect.fn("EnvironmentShellState.make") ); } if (Option.isSome(next.snapshot)) { - yield* Ref.set(latestLiveSnapshot, next.snapshot); + yield* Ref.set( + latestLiveShell, + Option.some({ snapshot: next.snapshot.value, resume: basis }), + ); } yield* SubscriptionRef.set(state, next); - if (receivedSnapshot) { - const session = yield* Ref.get(activeSubscriptionSession); - if (session !== null) { - yield* Ref.set(lastAuthoritativeSession, session); - } - } if (next.snapshot !== initial.snapshot && Option.isSome(next.snapshot)) { yield* Queue.offer(persistence, next.snapshot.value); } @@ -337,16 +422,16 @@ export const makeEnvironmentShellState = Effect.fn("EnvironmentShellState.make") }, ); pendingLinkRows = new Set(); - const live = yield* Ref.get(latestLiveSnapshot); + const live = yield* Ref.get(latestLiveShell); 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) ? { ...live.value, snapshot: filled.value } : live.value; + yield* Ref.set(latestLiveShell, Option.some(latest)); + yield* Queue.offer(persistence, latest.snapshot); }, (effect) => snapshotWriteLock.withPermit(effect), ); if (Option.isSome(cached) && Option.isSome(cachedSnapshot)) { - yield* fillDeferredPullRequests(cached.value, cachedSnapshot.value); + yield* fillDeferredPullRequests(cached.value.snapshot, cachedSnapshot.value); } const foregroundResubscriptions = Option.match(wakeups, { @@ -357,34 +442,47 @@ export const makeEnvironmentShellState = Effect.fn("EnvironmentShellState.make") yield* setSynchronizing; yield* Effect.forkScoped( - subscribeDynamic( + subscribeDynamicWithAttempt( ORCHESTRATION_V2_WS_METHODS.subscribeShell, - Effect.fn("EnvironmentShellState.makeSubscribeInput")(function* (session) { - yield* Ref.set(activeSubscriptionSession, session); - const { supportsCompletionMarker, omitIdleSubagents, omitOlderSettled } = - yield* session.initialConfig.pipe( - Effect.map((config) => ({ - supportsCompletionMarker: config.shellResumeCompletionMarker === true, - omitIdleSubagents: config.shellThreadFamilies === true, - omitOlderSettled: - config.shellOlderSettled === true && config.shellThreadFamilies === true, - })), - Effect.orElseSucceed(() => ({ - supportsCompletionMarker: false, - omitIdleSubagents: false, - omitOlderSettled: false, - })), - ); - yield* Ref.set(awaitingCompletion, supportsCompletionMarker); + Effect.fn("EnvironmentShellState.makeSubscribeInput")(function* (attempt) { + const { session } = attempt; + const config = yield* session.initialConfig.pipe(Effect.orElseSucceed(() => null)); + const supportsCompletionMarker = config?.shellResumeCompletionMarker === true; + // HTTP and the socket must agree, or omitted rows would return through the socket. A new + // snapshot flag joins this object, which also puts it in the resume key. + const selection = + config?.shellThreadFamilies === true + ? { + omitIdleSubagents: true as const, + ...(config.shellOlderSettled === true ? { omitOlderSettled: true as const } : {}), + } + : {}; + // A resume needs the marker to know when the shell is authoritative again. + const resumeKey = + config?.shellResumeKey === undefined || !supportsCompletionMarker || build === undefined + ? undefined + : encodeResumeKey({ server: config.shellResumeKey, client: build, selection }); + // Under the write lock, so an older attempt's apply finishes before the cursor is read. + yield* snapshotWriteLock.withPermit( + Ref.set(activeAttempt, { attempt, resumeKey, synchronized: false }).pipe( + Effect.andThen(Ref.set(awaitingCompletion, supportsCompletionMarker)), + ), + ); yield* setSynchronizing; - // Foreground resubscriptions on the same live session can resume from - // the in-memory cursor. A new session reloads the authoritative HTTP - // snapshot so a valid cursor cannot preserve incomplete cached data. - const hasAuthoritativeSnapshot = (yield* Ref.get(lastAuthoritativeSession)) === session; - let canResume = hasAuthoritativeSnapshot; + // The session that holds the shell's own snapshot resumes from its cursor. Any other + // session, or a cached shell, resumes only when it was synchronized against the same key + // less than a day ago; otherwise it loads the HTTP snapshot first. + const basis = yield* Ref.get(resumeBasis); + const age = basis === undefined ? -1 : (yield* Clock.currentTimeMillis) - basis.syncedAt; + let canResume = + (yield* Ref.get(lastAuthoritativeSession)) === session || + (resumeKey !== undefined && + basis?.key === resumeKey && + age >= 0 && + age < SHELL_RESUME_MAX_AGE_MS); let current = yield* SubscriptionRef.get(state); - if (!hasAuthoritativeSnapshot || Option.isNone(current.snapshot)) { + if (!canResume || Option.isNone(current.snapshot)) { const prepared = yield* SubscriptionRef.get(supervisor.prepared).pipe( Effect.flatMap( Option.match({ @@ -399,11 +497,7 @@ export const makeEnvironmentShellState = Effect.fn("EnvironmentShellState.make") }), ), ); - // HTTP and the socket must agree, or omitted children would return through the socket. - const httpSnapshot = yield* snapshotLoader.load(prepared, { - omitIdleSubagents, - omitOlderSettled, - }); + const httpSnapshot = yield* snapshotLoader.load(prepared, selection); if (Option.isSome(httpSnapshot)) { yield* cancelActiveFill; const previous = yield* SubscriptionRef.get(state); @@ -416,7 +510,7 @@ export const makeEnvironmentShellState = Effect.fn("EnvironmentShellState.make") ? [] : httpSnapshot.value.threads.map((thread) => thread.id), ); - yield* applyItems([{ kind: "snapshot", snapshot }], pendingLinkIds); + yield* applyItems([[attempt, { kind: "snapshot", snapshot }]], pendingLinkIds); canResume = true; current = yield* SubscriptionRef.get(state); if (Option.isSome(current.snapshot)) { @@ -425,16 +519,11 @@ export const makeEnvironmentShellState = Effect.fn("EnvironmentShellState.make") } } - const omission = { - ...(omitIdleSubagents ? { omitIdleSubagents: true as const } : {}), - ...(omitOlderSettled ? { omitOlderSettled: true as const } : {}), + const input = { + ...selection, + ...(supportsCompletionMarker ? { requestCompletionMarker: true as const } : {}), }; - if (!canResume || Option.isNone(current.snapshot)) { - return { - ...omission, - ...(supportsCompletionMarker ? { requestCompletionMarker: true as const } : {}), - }; - } + if (!canResume || Option.isNone(current.snapshot)) return input; if (!supportsCompletionMarker) { yield* SubscriptionRef.update(state, (value) => ({ ...value, @@ -442,11 +531,7 @@ export const makeEnvironmentShellState = Effect.fn("EnvironmentShellState.make") error: Option.none(), })); } - return { - afterSequence: current.snapshot.value.snapshotSequence, - ...omission, - ...(supportsCompletionMarker ? { requestCompletionMarker: true as const } : {}), - }; + return { afterSequence: current.snapshot.value.snapshotSequence, ...input }; }), { onExpectedFailure: (cause) => setStreamError(Cause.squash(cause)), diff --git a/packages/contracts/src/server.ts b/packages/contracts/src/server.ts index c14b4b4b24eaf..b11e89c862cc2 100644 --- a/packages/contracts/src/server.ts +++ b/packages/contracts/src/server.ts @@ -594,6 +594,12 @@ export const ServerConfig = Schema.Struct({ observability: ServerObservability, settings: ServerSettings, shellResumeCompletionMarker: Schema.optionalKey(Schema.Boolean), + /** + * Opaque, new for each server process. A shell synchronized against one value may resume from its + * cursor in a later session only while the server sends the same value: a new process can carry + * migrations, new code or another database, none of which emit events. + */ + shellResumeKey: Schema.optionalKey(Schema.String), /** * The shell can leave idle subagent children out of snapshots (`omitIdleSubagents`), and * `getThreadFamily` and `getWorktreeThreadIds` serve the rows it leaves out. From 124f10be57cbec60f0f256aae215c9957dbcdd2f Mon Sep 17 00:00:00 2001 From: Theo Browne Date: Sun, 25 Oct 2026 03:40:54 -0700 Subject: [PATCH 2/3] perf(shell): a resume never sends more rows than a fresh snapshot After #24723 a fresh snapshot keeps about 25 settled threads, so the 500-row replay cap let resumes cost more than a snapshot. The cap now matches the settled rows a snapshot keeps. Co-Authored-By: Claude Opus 5.5 --- apps/server/src/orchestration-v2/ShellStream.ts | 2 +- apps/server/src/ws.test.ts | 4 ++-- apps/server/src/ws.ts | 6 ++++-- 3 files changed, 7 insertions(+), 5 deletions(-) diff --git a/apps/server/src/orchestration-v2/ShellStream.ts b/apps/server/src/orchestration-v2/ShellStream.ts index 0a23fe3cb047a..c28c3c02a1ead 100644 --- a/apps/server/src/orchestration-v2/ShellStream.ts +++ b/apps/server/src/orchestration-v2/ShellStream.ts @@ -116,7 +116,7 @@ export function activeShellRowsForSubscriber( }; } -const NEWEST_SETTLED_KEPT = 25; +export const NEWEST_SETTLED_KEPT = 25; /** * Opted-in snapshots keep the 25 newest settled top-level threads (and ties) and count the rest per diff --git a/apps/server/src/ws.test.ts b/apps/server/src/ws.test.ts index 8a425f024dcd2..a5f49e84d02e7 100644 --- a/apps/server/src/ws.test.ts +++ b/apps/server/src/ws.test.ts @@ -381,12 +381,12 @@ it.effect("shell resume sends each touched aggregate once, or a snapshot past it assert.deepEqual(yield* resume(gone + 1), ["snapshot", "synchronized"]); const first = yield* insert("project", "project:0", "project.meta-updated"); - for (let index = 1; index <= 500; index += 1) { + for (let index = 1; index <= 25; index += 1) { yield* insert("project", `project:${index}`, "project.meta-updated"); } assert.deepEqual(yield* resume(gone), ["snapshot", "synchronized"]); const capped = yield* resume(first); - assert.equal(capped.length, 501); + assert.equal(capped.length, 26); assert.equal(capped.at(-1), "synchronized"); }).pipe(Effect.scoped, Effect.provide(ReplayTestLayer)), ); diff --git a/apps/server/src/ws.ts b/apps/server/src/ws.ts index 412cadfacb55d..dd24604b35f16 100644 --- a/apps/server/src/ws.ts +++ b/apps/server/src/ws.ts @@ -124,6 +124,7 @@ import * as ScheduledTasks from "./scheduledTasks/ScheduledTaskService.ts"; import * as SecretRequests from "./secrets/SecretRequests.ts"; import * as GuestShareService from "./guest/GuestShareService.ts"; import { + NEWEST_SETTLED_KEPT, archivedShellStreamItemFromThreadShell, buildActiveShellSnapshot, bufferShellLiveStream, @@ -575,9 +576,10 @@ const serverWsRpcs = Array.from(ServerWsRpcGroup.requests.values()); const SplitWsRpcGroup = RpcGroup.make(...serverWsRpcs.filter(isSplitWsRpc)); const OtherWsRpcGroup = RpcGroup.make(...serverWsRpcs.filter(isOtherWsRpc)); // A shell resume decodes no payloads: one indexed scan of the gap finds the touched projects and -// threads, then their current rows are read. Past either cap a snapshot is about as cheap. +// threads, then their current rows are read. Past either cap a snapshot is about as cheap: a +// snapshot holds at least the newest settled threads, while touched rows are often ones it omits. const SHELL_RESUME_MAX_GAP = 100_000; -const SHELL_RESUME_MAX_TOUCHED = 500; +const SHELL_RESUME_MAX_TOUCHED = NEWEST_SETTLED_KEPT; function toAuthAccessStreamEvent( change: PairingGrantStore.BootstrapCredentialChange | SessionStore.SessionCredentialChange, From 87ddf5b19881bacece5f3638b838627650e9446d Mon Sep 17 00:00:00 2001 From: Theo Browne Date: Sun, 25 Oct 2026 03:49:55 -0700 Subject: [PATCH 3/3] test(shell): resume tests use the flags this server sends Resume tests now include omitOlderSettled, and a completed resume reads the cached older settled counts once more. Co-Authored-By: Claude Opus 5.5 --- .../src/state/olderSettledThreads.test.ts | 26 +++++++++++++++++++ .../src/state/shell-sync.test.ts | 12 ++++++--- 2 files changed, 35 insertions(+), 3 deletions(-) diff --git a/packages/client-runtime/src/state/olderSettledThreads.test.ts b/packages/client-runtime/src/state/olderSettledThreads.test.ts index e8c419f779d26..87d9aa125d9d1 100644 --- a/packages/client-runtime/src/state/olderSettledThreads.test.ts +++ b/packages/client-runtime/src/state/olderSettledThreads.test.ts @@ -428,6 +428,32 @@ describe("createOlderSettledThreadAtoms", () => { }), ); + it.effect("a completed resume reads the cached counts once more", () => + Effect.gen(function* () { + const { replies, requests, atoms, registry, setShell, until } = yield* makeAtoms(); + const olderAtom = atoms.olderSettledAtom(environmentId); + const count = (value: OlderSettled | null) => value?.counts.get(project); + registry.mount(olderAtom); + + // A resume keeps the cached snapshot, whose cursor is past its counts. + const cached = shellSnapshot({ sequence: 5, asOfSequence: 3, count: 5 }); + setShell({ snapshot: cached }); + expect(count(registry.get(olderAtom))).toBe(5); + yield* Queue.offer( + replies, + Effect.succeed({ + snapshotSequence: 6, + threads: [], + counts: [{ projectId: project, count: 4 }], + }), + ); + setShell({ snapshot: cached, generation: 1 }); + yield* TestClock.adjust("500 millis"); + expect(count(yield* until(olderAtom, (value) => count(value) === 4))).toBe(4); + expect(requests).toEqual([{ before: { settledAt: before }, limit: 0 }]); + }), + ); + it.effect("reads pages in order until a short one, and retries a failed page", () => Effect.gen(function* () { const { replies, requests, atoms, registry, setShell, until } = yield* makeAtoms(); diff --git a/packages/client-runtime/src/state/shell-sync.test.ts b/packages/client-runtime/src/state/shell-sync.test.ts index 31356a6053b33..104f28f41bc10 100644 --- a/packages/client-runtime/src/state/shell-sync.test.ts +++ b/packages/client-runtime/src/state/shell-sync.test.ts @@ -129,6 +129,7 @@ const makePersistenceHarness = Effect.fn("test.makeShellPersistenceHarness")(fun const RESUMABLE_CONFIG = { shellResumeCompletionMarker: true, shellThreadFamilies: true, + shellOlderSettled: true, shellResumeKey: "server-1", }; @@ -230,7 +231,12 @@ const savedResumableShell = Effect.fn(function* () { return (yield* Ref.get(before.saved)).at(-1)!; }); -const RESUMED_INPUT = { afterSequence: 10, omitIdleSubagents: true, requestCompletionMarker: true }; +const RESUMED_INPUT = { + afterSequence: 10, + omitIdleSubagents: true, + omitOlderSettled: true, + requestCompletionMarker: true, +}; describe("environment shell synchronization", () => { it.effect("skips unchanged shell writes, persists resets and flushes changed rows", () => @@ -1050,8 +1056,8 @@ describe("environment shell synchronization", () => { { name: "another client build", build: "build-2" }, { name: "other snapshot flags", config: { ...RESUMABLE_CONFIG, shellThreadFamilies: false } }, { - name: "older settled threads left out", - config: { ...RESUMABLE_CONFIG, shellOlderSettled: true }, + name: "a server that keeps older settled threads", + config: { ...RESUMABLE_CONFIG, shellOlderSettled: false }, }, { name: "a server without completion markers",