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..c28c3c02a1ead 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; @@ -114,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 @@ -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..a5f49e84d02e7 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 <= 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, 26); + 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..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, @@ -133,6 +134,8 @@ import { dedupeShellEnrichment, loadOlderSettledThreadsPage, loadShellSnapshotParts, + shellApplicationEventFromReplayHead, + ShellResumeKey, shellStreamItemFromEnrichmentRefresh, shellStreamItemFromThreadShell, shellStreamItemsFromInitialSnapshot, @@ -140,6 +143,7 @@ import { shellStreamItemsFromResumeSnapshot, shouldPublishShellApplicationEvent, skipUnchangedThreadShells, + SUBAGENT_TRANSCRIPT_EVENT_TYPES, toShellApplicationEvent, type ShellApplicationEvent, } from "./orchestration-v2/ShellStream.ts"; @@ -571,11 +575,11 @@ 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: 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 = NEWEST_SETTLED_KEPT; function toAuthAccessStreamEvent( change: PairingGrantStore.BootstrapCredentialChange | SessionStore.SessionCredentialChange, @@ -724,35 +728,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 +1138,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 +1707,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 +1756,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/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 5a5a50ab21105..104f28f41bc10 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,118 @@ const makePersistenceHarness = Effect.fn("test.makeShellPersistenceHarness")(fun return { state, send }; }); +const RESUMABLE_CONFIG = { + shellResumeCompletionMarker: true, + shellThreadFamilies: true, + shellOlderSettled: 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, + omitOlderSettled: true, + requestCompletionMarker: true, +}; + describe("environment shell synchronization", () => { it.effect("skips unchanged shell writes, persists resets and flushes changed rows", () => Effect.gen(function* () { @@ -593,7 +710,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 +876,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 +1028,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: "a server that keeps older settled threads", + config: { ...RESUMABLE_CONFIG, shellOlderSettled: false }, + }, + { + 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.