diff --git a/apps/server/src/cli/project.ts b/apps/server/src/cli/project.ts index 25733a5e35b..39b3b243112 100644 --- a/apps/server/src/cli/project.ts +++ b/apps/server/src/cli/project.ts @@ -337,7 +337,9 @@ const dispatchLiveOrchestrationCommand = ( const getOfflineSnapshot = Effect.fn("getOfflineSnapshot")(function* () { const projectionSnapshotQuery = yield* ProjectionSnapshotQuery.ProjectionSnapshotQuery; - return yield* projectionSnapshotQuery.getSnapshot(); + // Project commands only read the project list, so use the lightweight + // command read model instead of hydrating every thread body in the database. + return yield* projectionSnapshotQuery.getCommandReadModel(); }); const tryResolveLiveProjectExecutionMode = Effect.fn("tryResolveLiveProjectExecutionMode")( diff --git a/apps/server/src/orchestration/http.ts b/apps/server/src/orchestration/http.ts index 659665e47b5..9a5c8c0be39 100644 --- a/apps/server/src/orchestration/http.ts +++ b/apps/server/src/orchestration/http.ts @@ -32,8 +32,13 @@ export const orchestrationHttpApiLayer = HttpApiBuilder.group( Effect.fn("environment.orchestration.snapshot")(function* (args) { yield* annotateEnvironmentRequest(args.endpoint.name); yield* requireEnvironmentScope(AuthOrchestrationReadScope); + // Serve the lightweight command read model (thread bodies empty) + // instead of the fully hydrated snapshot. Hydrating every message + // and activity payload in the database has OOM-killed servers, and + // the route's only consumer (the project CLI) reads projects alone — + // UI clients load the shell and per-thread snapshots instead. return yield* projectionSnapshotQuery - .getSnapshot() + .getCommandReadModel() .pipe( Effect.catch((cause) => failEnvironmentInternal("orchestration_snapshot_failed", cause), diff --git a/apps/server/src/server.test.ts b/apps/server/src/server.test.ts index fff71dbb4e7..569e8a51c37 100644 --- a/apps/server/src/server.test.ts +++ b/apps/server/src/server.test.ts @@ -6005,6 +6005,149 @@ it.layer(NodeServices.layer)("server router seam", (it) => { }).pipe(Effect.provide(NodeHttpServer.layerTest), TestClock.withLive), ); + it.effect("subscribeThread sends a fresh snapshot instead of replaying a large gap", () => + Effect.gen(function* () { + let readEventsCalls = 0; + const thread = makeDefaultOrchestrationReadModel().threads[0]!; + + yield* buildAppUnderTest({ + layers: { + orchestrationEngine: { + // Head is far ahead of the client's afterSequence (gap > 1000). + latestSequence: Effect.succeed(100_000), + readEvents: () => + Stream.sync(() => { + readEventsCalls += 1; + return {} as OrchestrationEvent; + }), + }, + projectionSnapshotQuery: { + getThreadDetailSnapshot: () => + Effect.succeed(Option.some({ snapshotSequence: 100_000, thread })), + }, + }, + }); + + const wsUrl = yield* getWsServerUrl("/ws"); + const items = yield* Effect.scoped( + withWsRpcClient(wsUrl, (client) => + client[ORCHESTRATION_WS_METHODS.subscribeThread]({ + threadId: defaultThreadId, + afterSequence: 5, + requestCompletionMarker: true, + }).pipe(Stream.take(2), Stream.runCollect), + ), + ); + + const [first, second] = Array.from(items); + // Large gap => fresh thread snapshot, and the global replay never starts. + assert.equal(first?.kind, "snapshot"); + if (first?.kind === "snapshot") { + assert.equal(first.snapshot.thread.id, defaultThreadId); + assert.equal(first.snapshot.snapshotSequence, 100_000); + } + assert.equal(second?.kind, "synchronized"); + assert.equal(readEventsCalls, 0); + }).pipe(Effect.provide(NodeHttpServer.layerTest)), + ); + + it.effect("subscribeThread replaces a cursor ahead of the authoritative head", () => + Effect.gen(function* () { + let readEventsCalls = 0; + const thread = makeDefaultOrchestrationReadModel().threads[0]!; + + yield* buildAppUnderTest({ + layers: { + orchestrationEngine: { + latestSequence: Effect.succeed(5), + readEvents: () => + Stream.sync(() => { + readEventsCalls += 1; + return {} as OrchestrationEvent; + }), + }, + projectionSnapshotQuery: { + getThreadDetailSnapshot: () => + Effect.succeed(Option.some({ snapshotSequence: 5, thread })), + }, + }, + }); + + const wsUrl = yield* getWsServerUrl("/ws"); + const first = yield* Effect.scoped( + withWsRpcClient(wsUrl, (client) => + client[ORCHESTRATION_WS_METHODS.subscribeThread]({ + threadId: defaultThreadId, + afterSequence: 10, + }).pipe(Stream.runHead), + ), + ); + + assert.equal(Option.getOrThrow(first).kind, "snapshot"); + assert.equal(readEventsCalls, 0); + }).pipe(Effect.provide(NodeHttpServer.layerTest)), + ); + + it.effect("subscribeThread bounds catch-up replay to the captured head", () => + Effect.gen(function* () { + let replayLimit: number | undefined; + const now = "2026-01-01T00:00:00.000Z"; + const messageEvent = { + sequence: 3, + eventId: EventId.make("event-replay-message"), + aggregateKind: "thread", + aggregateId: defaultThreadId, + occurredAt: now, + commandId: null, + causationEventId: null, + correlationId: null, + metadata: {}, + type: "thread.message-sent", + payload: { + threadId: defaultThreadId, + messageId: MessageId.make("message-replay"), + role: "user", + text: "Replayed message", + turnId: null, + streaming: false, + createdAt: now, + updatedAt: now, + }, + } satisfies Extract; + + yield* buildAppUnderTest({ + layers: { + orchestrationEngine: { + latestSequence: Effect.succeed(50), + readEvents: (_afterSequence, limit) => { + replayLimit = limit; + return Stream.make(messageEvent); + }, + }, + }, + }); + + const wsUrl = yield* getWsServerUrl("/ws"); + const items = yield* Effect.scoped( + withWsRpcClient(wsUrl, (client) => + client[ORCHESTRATION_WS_METHODS.subscribeThread]({ + threadId: defaultThreadId, + afterSequence: 0, + requestCompletionMarker: true, + }).pipe(Stream.take(2), Stream.runCollect), + ), + ); + + const [first, second] = Array.from(items); + assert.equal(first?.kind, "event"); + assert.equal(first?.kind === "event" ? first.event.sequence : null, 3); + assert.equal(second?.kind, "synchronized"); + // The replay is bounded to the head captured before the read, not + // Number.MAX_SAFE_INTEGER. + assert.equal(replayLimit, 50); + }).pipe(Effect.provide(NodeHttpServer.layerTest)), + ); + it.effect("subscribeShell sends a fresh snapshot instead of replaying a large gap", () => Effect.gen(function* () { let readEventsCalls = 0; diff --git a/apps/server/src/ws.ts b/apps/server/src/ws.ts index 06888ef3f70..909a51a4cf5 100644 --- a/apps/server/src/ws.ts +++ b/apps/server/src/ws.ts @@ -298,6 +298,13 @@ const PROVIDER_STATUS_DEBOUNCE_MS = 200; // Matches the event store's default page size (DEFAULT_READ_FROM_SEQUENCE_LIMIT). const SHELL_RESUME_MAX_GAP = 1_000; +// Same bound for thread resume. The replay reads the *global* event range and +// filters per-thread afterwards, so a stale cursor far behind the head would +// otherwise decode every intervening event's payload — reconnects with cursors +// hundreds of thousands of events behind have OOM-killed servers on large +// databases. Past this gap the client is reset with a fresh thread snapshot. +const THREAD_RESUME_MAX_GAP = 1_000; + function toAuthAccessStreamEvent( change: PairingGrantStore.BootstrapCredentialChange | SessionStore.SessionCredentialChange, revision: number, @@ -1281,38 +1288,50 @@ const makeWsRpcLayer = ( // catch-up followed by the buffered/ongoing live events. Overlapping // events are deduped by sequence on the client. // - // Read the full range after the cursor (not the store's default - // page-bounded limit): the range is normally tiny (a fresh HTTP - // snapshot sequence) and the per-thread filter runs after reading, - // so a global cap could otherwise omit this thread's events. + // The replay is bounded to the projection head captured below. The + // catch-up range is normally tiny (a fresh HTTP snapshot sequence), + // but a stale cached cursor can sit hundreds of thousands of global + // events behind — replaying that decodes every intervening event + // (including every other thread's tool payloads) only to discard + // almost all of them, which has OOM-killed servers on large + // databases. A truncated replay would silently drop this thread's + // events, so past the gap cap we reset the client with a fresh + // thread snapshot instead, exactly like subscribeShell above. if (input.afterSequence !== undefined) { const afterSequence = input.afterSequence; - const catchUpStream = orchestrationEngine - .readEvents(afterSequence, Number.MAX_SAFE_INTEGER) - .pipe( - Stream.filter(isThisThreadDetailEvent), - Stream.map((event) => ({ - kind: "event" as const, - event: projectActivityEvent(event), - })), - Stream.mapError( - (cause) => - new OrchestrationGetSnapshotError({ - message: `Failed to replay thread ${input.threadId} events`, - cause, - }), - ), - ); - const afterCatchUp = - input.requestCompletionMarker === true - ? Stream.concat( - Stream.fromEffect( - Queue.offer(liveBuffer, { kind: "synchronized" as const }), - ).pipe(Stream.drain), - bufferedLiveStream, - ) - : bufferedLiveStream; - return Stream.concat(catchUpStream, afterCatchUp); + const headSequence = yield* orchestrationEngine.latestSequence; + const replayGap = headSequence - afterSequence; + if (replayGap >= 0 && replayGap <= THREAD_RESUME_MAX_GAP) { + const catchUpStream = orchestrationEngine + .readEvents(afterSequence, replayGap) + .pipe( + Stream.filter(isThisThreadDetailEvent), + Stream.map((event) => ({ + kind: "event" as const, + event: projectActivityEvent(event), + })), + Stream.mapError( + (cause) => + new OrchestrationGetSnapshotError({ + message: `Failed to replay thread ${input.threadId} events`, + cause, + }), + ), + ); + const afterCatchUp = + input.requestCompletionMarker === true + ? Stream.concat( + Stream.fromEffect( + Queue.offer(liveBuffer, { kind: "synchronized" as const }), + ).pipe(Stream.drain), + bufferedLiveStream, + ) + : bufferedLiveStream; + return Stream.concat(catchUpStream, afterCatchUp); + } + // Gap too large (or cursor ahead of authoritative state): fall + // through to the snapshot path so the client converges from a + // fresh thread detail instead of an unbounded replay. } const snapshot = yield* projectionSnapshotQuery