From a0b16a1b11b3dff5956c872999e21fadf55da276 Mon Sep 17 00:00:00 2001 From: Stephen Belanger Date: Wed, 28 Oct 2026 18:53:50 +0800 Subject: [PATCH 1/3] fix(server): preserve reconnect sync progress on unreliable networks --- .../OrchestrationEventStore.sequence.test.ts | 43 +++++++++++ .../persistence/OrchestrationEventStore.ts | 22 +++++- apps/server/src/ws.test.ts | 59 +++++++++++++++ apps/server/src/ws.ts | 74 +++++++++++++++---- .../src/state/threads-sync.test.ts | 27 +++++++ 5 files changed, 208 insertions(+), 17 deletions(-) diff --git a/apps/server/src/persistence/OrchestrationEventStore.sequence.test.ts b/apps/server/src/persistence/OrchestrationEventStore.sequence.test.ts index 07f4fe5c9fbfb..c873c58fbf3dd 100644 --- a/apps/server/src/persistence/OrchestrationEventStore.sequence.test.ts +++ b/apps/server/src/persistence/OrchestrationEventStore.sequence.test.ts @@ -68,6 +68,49 @@ const seedEvents = Effect.fn("test.seedSequenceEvents")(function* ( `; }); +it.effect("fetches the next byte-bounded replay page after delivering the previous page", () => + Effect.gen(function* () { + const store = yield* OrchestrationEventStore.OrchestrationEventStore; + const sql = yield* SqlClient.SqlClient; + const threadId = ThreadId.make("byte-paged-replay"); + const rows = [1, 2].map((ordinal) => ({ + ...eventRow(ordinal, "thread", 2, threadId), + payload_json: encodeJson({ + providerSessionId: `session:${ordinal}:${"x".repeat(180_000)}`, + detachedAt: occurredAt, + }), + })); + const inserted = yield* seedEvents(rows); + const updatedPayload = { + providerSessionId: ProviderSessionId.make("session:updated"), + detachedAt: occurredAt, + }; + const items = yield* store + .readAgentEvents({ + threadId, + throughSequence: inserted[1]!.sequence, + maxPageBytes: 256 * 1024, + }) + .pipe( + Stream.tap((item) => + item.sequence === inserted[0]!.sequence + ? sql`UPDATE orchestration_events SET payload_json = ${encodeJson(updatedPayload)} + WHERE sequence = ${inserted[1]!.sequence}` + : Effect.void, + ), + Stream.runCollect, + ); + assert.deepEqual( + items.map((item) => item.sequence), + inserted.map((item) => item.sequence), + ); + assert.deepEqual(items[1]!.event.payload, { + ...updatedPayload, + detachedAt: DateTime.makeUnsafe(occurredAt), + }); + }).pipe(Effect.provide(layerEventStore)), +); + it.effect("keeps application and scoped agent high-water marks separate from legacy history", () => Effect.gen(function* () { const store = yield* OrchestrationEventStore.OrchestrationEventStore; diff --git a/apps/server/src/persistence/OrchestrationEventStore.ts b/apps/server/src/persistence/OrchestrationEventStore.ts index 51c3861a9fca0..9f7eecb6e35c8 100644 --- a/apps/server/src/persistence/OrchestrationEventStore.ts +++ b/apps/server/src/persistence/OrchestrationEventStore.ts @@ -60,6 +60,8 @@ export class OrchestrationEventStore extends Context.Service< readonly eventType?: OrchestrationV2DomainEvent["type"]; readonly limit?: number; readonly skipUnknownEventTypes?: boolean; + /** Bound fetched JSON per page, always allowing one indivisible event. */ + readonly maxPageBytes?: number; }) => Stream.Stream; readonly getAgentReplayStats: (input: { @@ -392,8 +394,9 @@ const makeEventStore = Effect.gen(function* () { readonly eventType?: OrchestrationV2DomainEvent["type"]; readonly onlyAgentEvents?: boolean; readonly limit: number; - }) => - sql` + readonly maxPageBytes?: number; + }) => { + const query = sql` SELECT sequence, event_id, @@ -433,7 +436,19 @@ const makeEventStore = Effect.gen(function* () { ])} ORDER BY sequence ASC LIMIT ${input.limit} + `; + if (input.maxPageBytes === undefined) return query.pipe(NodeSqliteClient.readOnly); + return sql` + SELECT * FROM ( + SELECT *, ROW_NUMBER() OVER (ORDER BY sequence) AS replay_row, + SUM(octet_length(payload_json) + octet_length(metadata_json)) + OVER (ORDER BY sequence ROWS UNBOUNDED PRECEDING) AS replay_bytes + FROM (${query}) + ) + WHERE replay_row = 1 OR replay_bytes <= ${input.maxPageBytes} + ORDER BY sequence `.pipe(NodeSqliteClient.readOnly); + }; const appendAgentEvents: OrchestrationEventStore["Service"]["appendAgentEvents"] = (input) => Effect.forEach( @@ -556,6 +571,7 @@ const makeEventStore = Effect.gen(function* () { ...(input?.eventType === undefined ? {} : { eventType: input.eventType }), onlyAgentEvents: true, limit: pageLimit, + ...(input?.maxPageBytes === undefined ? {} : { maxPageBytes: input.maxPageBytes }), }).pipe( Effect.mapError(toPersistenceSqlError("OrchestrationEventStore.readAgentEvents:query")), Effect.map((rows) => { @@ -564,7 +580,7 @@ const makeEventStore = Effect.gen(function* () { return [ rows, last === undefined || - rows.length < pageLimit || + (input?.maxPageBytes === undefined && rows.length < pageLimit) || nextRemaining <= 0 || (input?.throughSequence !== undefined && last.sequence >= input.throughSequence) ? Option.none() diff --git a/apps/server/src/ws.test.ts b/apps/server/src/ws.test.ts index ac60becc2d78a..6019dfd404bf5 100644 --- a/apps/server/src/ws.test.ts +++ b/apps/server/src/ws.test.ts @@ -303,6 +303,65 @@ it.live.each(["thread", "shell"] as const)( }).pipe(Effect.scoped, Effect.provide(ReplayTestLayer)), ); +it.effect.each([ + { count: 260, padding: 0, unknownPrefix: 0 }, + { count: 3, padding: 600_000, unknownPrefix: 0 }, + { count: 3, padding: 0, unknownPrefix: 140 }, +])( + "resumes catch-up of $count events ($padding bytes, $unknownPrefix unknown rows) after interruption", + ({ count, padding, unknownPrefix }) => + Effect.gen(function* () { + const sql = yield* SqlClient.SqlClient; + const threadId = ThreadId.make("thread:incremental-replay"); + const now = "2026-10-28T00:00:00.000Z"; + const rows = Array.from({ length: count + unknownPrefix }, (_, version) => ({ + event_id: `incremental-replay:${version}`, + aggregate_kind: "thread", + stream_id: threadId, + stream_version: version, + event_type: version < unknownPrefix ? "future.event" : "provider-session.detached", + occurred_at: now, + actor_kind: "server", + payload_json: JSON.stringify({ + providerSessionId: `session:${version}:${"x".repeat(padding)}`, + detachedAt: now, + }), + metadata_json: "{}", + application_event_version: 2, + })); + const inserted = yield* sql<{ sequence: number }>` + INSERT INTO orchestration_events ${sql.insert(rows)} RETURNING sequence + `; + const subscribe = (afterSequence: number) => + subscribeOrchestrationV2Thread({ + threadId, + afterSequence, + requestCompletionMarker: true, + }).pipe( + Effect.provideService(ThreadManagement.ThreadManagementService, { + ensureLegacyTranscript: () => Effect.void, + getThreadSnapshot: () => + Effect.die("Ordinary catch-up must not restart with a snapshot"), + streamStoredEventsFrom: () => Stream.empty, + } as unknown as ThreadManagement.ThreadManagementService["Service"]), + ); + const first = yield* subscribe(0).pipe( + Effect.flatMap((stream) => + stream.pipe(Stream.take(Math.min(10, count - 1)), Stream.runCollect), + ), + ); + const last = first.at(-1); + assert.equal(first.length, Math.min(10, count - 1)); + assert.equal(last?.kind, "event"); + if (last?.kind !== "event") return; + const resumed = yield* subscribe(last.sequence).pipe(Effect.flatMap(Stream.runCollect)); + assert.deepEqual( + [...first, ...resumed].map((item) => (item.kind === "event" ? item.sequence : item.kind)), + [...inserted.slice(unknownPrefix).map((row) => row.sequence), "synchronized"], + ); + }).pipe(Effect.scoped, Effect.provide(ReplayTestLayer)), +); + it.effect("coalesces streaming replay before synchronization and the live tail", () => Effect.gen(function* () { const store = yield* OrchestrationEventStore.OrchestrationEventStore; diff --git a/apps/server/src/ws.ts b/apps/server/src/ws.ts index e1d6303b23506..cda47efa39556 100644 --- a/apps/server/src/ws.ts +++ b/apps/server/src/ws.ts @@ -554,6 +554,7 @@ function projectFileFailureContext( } const PROVIDER_STATUS_DEBOUNCE_MS = 200; +const THREAD_REPLAY_PAGE_BYTES = 256 * 1024; // Middleware added later wraps middleware added earlier, so instrumentation wraps authorization. const ServerWsRpcGroup = WsRpcGroup.middleware(RpcInstrumentation); @@ -823,13 +824,20 @@ export const subscribeOrchestrationV2Thread = Effect.fn("ws.orchestrationV2.subs ), ); - const loadReplayThrough = (afterSequence: number, throughSequence: number) => - applicationEvents + const loadReplayThrough = ( + afterSequence: number, + throughSequence: number, + incremental = false, + ) => { + let encodedBytes = 0; + const replay = applicationEvents .readAgentEvents({ threadId: input.threadId, afterSequence, throughSequence, - limit: THREAD_RESUME_MAX_REPLAY_EVENTS + 1, + ...(incremental + ? { maxPageBytes: THREAD_REPLAY_PAGE_BYTES } + : { limit: THREAD_RESUME_MAX_REPLAY_EVENTS + 1 }), // Downgrades can leave unknown rows; skip them and resume live from highWater so later events survive. skipUnknownEventTypes: true, }) @@ -839,17 +847,32 @@ export const subscribeOrchestrationV2Thread = Effect.fn("ws.orchestrationV2.subs sequence: stored.sequence, event: projectDomainEventForWire(stored.event), })), - Stream.runCollect, - Effect.map((items) => Array.from(items)), - Effect.mapError( - (cause) => - new OrchestrationV2GetThreadProjectionError({ - threadId: input.threadId, - message: `Failed while replaying orchestration V2 thread ${input.threadId}`, - cause, - }), - ), ); + return ( + incremental + ? replay.pipe( + // Count delivered events rather than stored rows: a page of + // unknown future event types must not truncate the catch-up. + Stream.take(THREAD_RESUME_MAX_REPLAY_EVENTS), + Stream.takeUntil((item) => { + encodedBytes += threadReplayEncodedBytes([item]); + return encodedBytes >= THREAD_REPLAY_PAGE_BYTES; + }), + ) + : replay + ).pipe( + Stream.runCollect, + Effect.map((items) => Array.from(items)), + Effect.mapError( + (cause) => + new OrchestrationV2GetThreadProjectionError({ + threadId: input.threadId, + message: `Failed while replaying orchestration V2 thread ${input.threadId}`, + cause, + }), + ), + ); + }; const completionMarker = input.requestCompletionMarker === true @@ -919,7 +942,7 @@ export const subscribeOrchestrationV2Thread = Effect.fn("ws.orchestrationV2.subs threadId: input.threadId, afterSequence: input.afterSequence, throughSequence: highWater, - maxEvents: THREAD_RESUME_MAX_REPLAY_EVENTS, + maxEvents: highWater - input.afterSequence, }) .pipe( Effect.mapError( @@ -931,6 +954,29 @@ export const subscribeOrchestrationV2Thread = Effect.fn("ws.orchestrationV2.subs }), ), ); + if (!stats.hasCreateEvent) { + // Pull each page only as the transport consumes the previous one. The + // client commits each event's cursor before the final completion marker, + // so a dropped connection resumes from its partial catch-up progress. + const replay = Stream.paginate(input.afterSequence, (cursor) => + loadReplayThrough(cursor, highWater, true).pipe( + Effect.map((items) => { + const last = items.at(-1); + return [ + coalesceLiveUpdatedEvents(items), + last === undefined || last.sequence >= highWater + ? Option.none() + : Option.some(last.sequence), + ] as const; + }), + ), + ); + return replay.pipe( + Stream.concat(completionMarker), + Stream.concat(eventStreamFrom(highWater)), + withLiveDedupe, + ); + } // Bound stored and projected payload bytes separately; neither bounds process memory. if ( stats.eventCount > THREAD_RESUME_MAX_REPLAY_EVENTS || diff --git a/packages/client-runtime/src/state/threads-sync.test.ts b/packages/client-runtime/src/state/threads-sync.test.ts index c8c212db41ca1..25bd2524e479f 100644 --- a/packages/client-runtime/src/state/threads-sync.test.ts +++ b/packages/client-runtime/src/state/threads-sync.test.ts @@ -2079,6 +2079,33 @@ describe("EnvironmentThreads", () => { }), ); + it.effect("resumes partial catch-up before its completion marker arrives", () => + Effect.gen(function* () { + const h = yield* makeHarness({ cached: BASE_PROJECTION, completionMarker: true }); + yield* Queue.offer(h.inputs, titleUpdated("Partial progress", CACHED_SNAPSHOT_SEQUENCE + 1)); + yield* awaitThreadState( + h.observed, + (value) => + value.status === "synchronizing" && + Option.isSome(value.data) && + value.data.value.thread.title === "Partial progress", + ); + yield* h.replaceSession; + yield* SubscriptionRef.changes(h.subscriptionCount).pipe( + Stream.filter((count) => count >= 2), + Stream.runHead, + ); + expect(yield* Ref.get(h.lastSubscribeAfterSequence)).toBe(CACHED_SNAPSHOT_SEQUENCE + 1); + expect((yield* Ref.get(h.latest)).status).toBe("synchronizing"); + yield* Queue.offerAll(h.inputs, [ + titleUpdated("Finished", CACHED_SNAPSHOT_SEQUENCE + 2), + synchronized(), + ]); + const live = yield* awaitThreadState(h.observed, (value) => value.status === "live"); + expect(Option.getOrThrow(live.data).thread.title).toBe("Finished"); + }), + ); + it.effect("keeps replayed updates synchronizing until the completion marker arrives", () => Effect.gen(function* () { const harness = yield* makeHarness({ cached: BASE_PROJECTION, completionMarker: true }); From 2edf1ff7e9d1eebc7b3303c995897a5bb812799e Mon Sep 17 00:00:00 2001 From: Stephen Belanger Date: Wed, 28 Oct 2026 19:30:21 +0800 Subject: [PATCH 2/3] perf(server): avoid restoring payloads before reconnect replay --- .../OrchestrationEventStore.sequence.test.ts | 23 ++++++++++++++++ .../persistence/OrchestrationEventStore.ts | 26 ++++++++++++++++++ apps/server/src/ws.test.ts | 6 +++++ apps/server/src/ws.ts | 27 +++++++++++++++---- 4 files changed, 77 insertions(+), 5 deletions(-) diff --git a/apps/server/src/persistence/OrchestrationEventStore.sequence.test.ts b/apps/server/src/persistence/OrchestrationEventStore.sequence.test.ts index c873c58fbf3dd..727e87c9766c9 100644 --- a/apps/server/src/persistence/OrchestrationEventStore.sequence.test.ts +++ b/apps/server/src/persistence/OrchestrationEventStore.sequence.test.ts @@ -68,6 +68,29 @@ const seedEvents = Effect.fn("test.seedSequenceEvents")(function* ( `; }); +it.effect("checks creation ranges without restoring or decoding payloads", () => + Effect.gen(function* () { + const store = yield* OrchestrationEventStore.OrchestrationEventStore; + const threadId = ThreadId.make("creation-existence"); + const inserted = yield* seedEvents([ + { ...eventRow(1, "thread", 2, threadId), event_type: "thread.created", payload_json: "{" }, + eventRow(2, "thread", 2, threadId), + { ...eventRow(3, "thread", 2, "other"), event_type: "thread.created", payload_json: "{" }, + { ...eventRow(4, "thread", 1, threadId), event_type: "thread.created", payload_json: "{" }, + ]); + const range = { + threadId, + afterSequence: 0, + throughSequence: inserted[3]!.sequence, + eventType: "thread.created" as const, + }; + assert.isTrue(yield* store.hasAgentEvent(range)); + assert.isFalse(yield* store.hasAgentEvent({ ...range, afterSequence: inserted[0]!.sequence })); + assert.isFalse(yield* store.hasAgentEvent({ ...range, throughSequence: 0 })); + assert.isTrue(yield* store.hasAgentEvent({ ...range, throughSequence: inserted[0]!.sequence })); + }).pipe(Effect.provide(layerEventStore)), +); + it.effect("fetches the next byte-bounded replay page after delivering the previous page", () => Effect.gen(function* () { const store = yield* OrchestrationEventStore.OrchestrationEventStore; diff --git a/apps/server/src/persistence/OrchestrationEventStore.ts b/apps/server/src/persistence/OrchestrationEventStore.ts index 9f7eecb6e35c8..bf37bd843a81f 100644 --- a/apps/server/src/persistence/OrchestrationEventStore.ts +++ b/apps/server/src/persistence/OrchestrationEventStore.ts @@ -64,6 +64,14 @@ export class OrchestrationEventStore extends Context.Service< readonly maxPageBytes?: number; }) => Stream.Stream; + /** Check a scoped replay range without reading or restoring event payloads. */ + readonly hasAgentEvent: (input: { + readonly threadId: ThreadId; + readonly afterSequence: number; + readonly throughSequence: number; + readonly eventType: OrchestrationV2DomainEvent["type"]; + }) => Effect.Effect; + readonly getAgentReplayStats: (input: { readonly threadId: ThreadId; readonly afterSequence: number; @@ -601,6 +609,23 @@ const makeEventStore = Effect.gen(function* () { ); }; + const hasAgentEvent: OrchestrationEventStore["Service"]["hasAgentEvent"] = (input) => + sql<{ readonly found: number }>` + SELECT EXISTS ( + SELECT 1 FROM orchestration_events + WHERE aggregate_kind = 'thread' + AND stream_id = ${input.threadId} + AND application_event_version = 2 + AND sequence > ${input.afterSequence} + AND sequence <= ${input.throughSequence} + AND event_type = ${input.eventType} + ) AS found + `.pipe( + NodeSqliteClient.readOnly, + Effect.mapError(toPersistenceSqlError("OrchestrationEventStore.hasAgentEvent:query")), + Effect.map((rows) => rows[0]?.found === 1), + ); + const getAgentReplayStats: OrchestrationEventStore["Service"]["getAgentReplayStats"] = (input) => sql<{ readonly eventCount: number; @@ -759,6 +784,7 @@ const makeEventStore = Effect.gen(function* () { appendProjectEvent, appendAgentEvents, readAgentEvents, + hasAgentEvent, getAgentReplayStats, getReplayStats, latestAgentSequence, diff --git a/apps/server/src/ws.test.ts b/apps/server/src/ws.test.ts index 6019dfd404bf5..021b1c11e82cf 100644 --- a/apps/server/src/ws.test.ts +++ b/apps/server/src/ws.test.ts @@ -312,6 +312,7 @@ it.effect.each([ ({ count, padding, unknownPrefix }) => Effect.gen(function* () { const sql = yield* SqlClient.SqlClient; + const eventStore = yield* OrchestrationEventStore.OrchestrationEventStore; const threadId = ThreadId.make("thread:incremental-replay"); const now = "2026-10-28T00:00:00.000Z"; const rows = Array.from({ length: count + unknownPrefix }, (_, version) => ({ @@ -338,6 +339,11 @@ it.effect.each([ afterSequence, requestCompletionMarker: true, }).pipe( + Effect.provideService(OrchestrationEventStore.OrchestrationEventStore, { + ...eventStore, + getAgentReplayStats: () => + Effect.die("Ordinary catch-up must not restore payloads for stats"), + }), Effect.provideService(ThreadManagement.ThreadManagementService, { ensureLegacyTranscript: () => Effect.void, getThreadSnapshot: () => diff --git a/apps/server/src/ws.ts b/apps/server/src/ws.ts index cda47efa39556..5af1d0fd816c3 100644 --- a/apps/server/src/ws.ts +++ b/apps/server/src/ws.ts @@ -937,24 +937,24 @@ export const subscribeOrchestrationV2Thread = Effect.fn("ws.orchestrationV2.subs if (input.afterSequence > highWater) { return yield* snapshotThenLive(); } - const stats = yield* applicationEvents - .getAgentReplayStats({ + const hasCreateEvent = yield* applicationEvents + .hasAgentEvent({ threadId: input.threadId, afterSequence: input.afterSequence, throughSequence: highWater, - maxEvents: highWater - input.afterSequence, + eventType: "thread.created", }) .pipe( Effect.mapError( (cause) => new OrchestrationV2GetThreadProjectionError({ threadId: input.threadId, - message: `Failed to measure orchestration V2 thread ${input.threadId} replay`, + message: `Failed to check orchestration V2 thread ${input.threadId} creation`, cause, }), ), ); - if (!stats.hasCreateEvent) { + if (!hasCreateEvent) { // Pull each page only as the transport consumes the previous one. The // client commits each event's cursor before the final completion marker, // so a dropped connection resumes from its partial catch-up progress. @@ -977,6 +977,23 @@ export const subscribeOrchestrationV2Thread = Effect.fn("ws.orchestrationV2.subs withLiveDedupe, ); } + const stats = yield* applicationEvents + .getAgentReplayStats({ + threadId: input.threadId, + afterSequence: input.afterSequence, + throughSequence: highWater, + maxEvents: THREAD_RESUME_MAX_REPLAY_EVENTS, + }) + .pipe( + Effect.mapError( + (cause) => + new OrchestrationV2GetThreadProjectionError({ + threadId: input.threadId, + message: `Failed to measure orchestration V2 thread ${input.threadId} replay`, + cause, + }), + ), + ); // Bound stored and projected payload bytes separately; neither bounds process memory. if ( stats.eventCount > THREAD_RESUME_MAX_REPLAY_EVENTS || From 03ebabcab3f0b2f2b108469db9ae5b0be44df6a4 Mon Sep 17 00:00:00 2001 From: Stephen Belanger Date: Wed, 28 Oct 2026 19:43:15 +0800 Subject: [PATCH 3/3] perf(server): stop replay sizing when the byte page fills --- .../OrchestrationEventStore.test.ts | 28 ++++++++++-- .../persistence/OrchestrationEventStore.ts | 43 ++++++++++++++----- 2 files changed, 57 insertions(+), 14 deletions(-) diff --git a/apps/server/src/persistence/OrchestrationEventStore.test.ts b/apps/server/src/persistence/OrchestrationEventStore.test.ts index baad3d5c52a05..1843a764581dd 100644 --- a/apps/server/src/persistence/OrchestrationEventStore.test.ts +++ b/apps/server/src/persistence/OrchestrationEventStore.test.ts @@ -113,7 +113,7 @@ layer("OrchestrationEventStore", (it) => { const id = NodeId.make("node:subagent-prompt"); const driver = ProviderDriverKind.make("claudeAgent"); const providerInstanceId = ProviderInstanceId.make("claudeAgent"); - const prompt = "Audit the repository. ".repeat(200); + const prompt = 'Audit "repo"\n漢字 '.repeat(9_000); const subagent = (index: number, progress: string, nextPrompt = prompt) => ({ id: EventId.make(`event:subagent-prompt:subagent:${index}`), @@ -185,14 +185,14 @@ layer("OrchestrationEventStore", (it) => { subagent(3, "Retrying", "Audit again."), item(3, "Retrying", "Audit again."), ]; - yield* store.appendAgentEvents({ events }); + const appended = yield* store.appendAgentEvents({ events }); const stored = yield* sql<{ readonly payload_json: string }>` SELECT payload_json FROM orchestration_events WHERE stream_id = ${threadId} ORDER BY sequence ASC `; assert.deepEqual( - stored.map((row) => row.payload_json.includes(prompt)), + stored.map((row) => row.payload_json.includes(JSON.stringify(prompt).slice(1, -1))), [true, true, false, false, false, false, false, false], ); const replayed = yield* store @@ -214,6 +214,28 @@ layer("OrchestrationEventStore", (it) => { throughSequence: yield* store.latestApplicationSequence, }); assert.isAbove(applicationStats.rawPayloadBytes, prompt.length * 6); + const paged = yield* store + .readAgentEvents({ threadId, afterSequence, maxPageBytes: 256 * 1024 }) + .pipe(Stream.runCollect); + assert.deepEqual( + paged.map((entry) => entry.event), + events, + ); + // The first two updates here reference prompts in earlier anchor rows. + // Sizing must stop before touching a later page's malformed payload. + yield* sql`UPDATE orchestration_events SET payload_json = '{' WHERE sequence = ${appended[4]!.sequence}`; + const firstPage = yield* store + .readAgentEvents({ + threadId, + afterSequence: appended[1]!.sequence, + throughSequence: appended[4]!.sequence, + maxPageBytes: 256 * 1024, + }) + .pipe(Stream.take(1), Stream.runCollect); + assert.deepEqual( + firstPage.map((entry) => entry.event), + [events[2]!], + ); }), ); diff --git a/apps/server/src/persistence/OrchestrationEventStore.ts b/apps/server/src/persistence/OrchestrationEventStore.ts index bf37bd843a81f..c7e07329347e1 100644 --- a/apps/server/src/persistence/OrchestrationEventStore.ts +++ b/apps/server/src/persistence/OrchestrationEventStore.ts @@ -404,8 +404,7 @@ const makeEventStore = Effect.gen(function* () { readonly limit: number; readonly maxPageBytes?: number; }) => { - const query = sql` - SELECT + const columns = sql` sequence, event_id, command_id, @@ -418,6 +417,8 @@ const makeEventStore = Effect.gen(function* () { application_event_version, causation_event_id, correlation_id + `; + const from = sql` FROM orchestration_events ${ // Otherwise the sequence index wins and command lookups scan every V2 event. @@ -442,18 +443,38 @@ const makeEventStore = Effect.gen(function* () { ...(input.commandId === undefined ? [] : [sql`command_id = ${input.commandId}`]), ...(input.eventType === undefined ? [] : [sql`event_type = ${input.eventType}`]), ])} - ORDER BY sequence ASC - LIMIT ${input.limit} `; - if (input.maxPageBytes === undefined) return query.pipe(NodeSqliteClient.readOnly); + if (input.maxPageBytes === undefined) { + return sql` + SELECT ${columns} ${from} ORDER BY sequence ASC LIMIT ${input.limit} + `.pipe(NodeSqliteClient.readOnly); + } + // Materialize only candidate ids. Walk their restored sizes until the page + // fills (at most one overflow row), rather than restoring all 500 bodies. + // Prompt references must be sized after restoration, including JSON escaping. return sql` - SELECT * FROM ( - SELECT *, ROW_NUMBER() OVER (ORDER BY sequence) AS replay_row, - SUM(octet_length(payload_json) + octet_length(metadata_json)) - OVER (ORDER BY sequence ROWS UNBOUNDED PRECEDING) AS replay_bytes - FROM (${query}) + WITH RECURSIVE candidates AS MATERIALIZED ( + SELECT sequence, ROW_NUMBER() OVER (ORDER BY sequence) AS ordinal + ${from} ORDER BY sequence ASC LIMIT ${input.limit} + ), replay_page(cursor, replay_bytes, row_count) AS ( + SELECT orchestration_events.sequence, + octet_length(${restoredPayloadJson}) + octet_length(metadata_json), 1 + FROM orchestration_events + JOIN candidates ON candidates.sequence = orchestration_events.sequence + WHERE candidates.ordinal = 1 + UNION ALL + SELECT orchestration_events.sequence, + page.replay_bytes + octet_length(${restoredPayloadJson}) + octet_length(metadata_json), + page.row_count + 1 + FROM replay_page AS page + JOIN candidates ON candidates.ordinal = page.row_count + 1 + JOIN orchestration_events ON orchestration_events.sequence = candidates.sequence + WHERE page.replay_bytes < ${input.maxPageBytes} ) - WHERE replay_row = 1 OR replay_bytes <= ${input.maxPageBytes} + SELECT ${columns} + FROM orchestration_events + JOIN replay_page ON replay_page.cursor = orchestration_events.sequence + WHERE replay_page.row_count = 1 OR replay_page.replay_bytes <= ${input.maxPageBytes} ORDER BY sequence `.pipe(NodeSqliteClient.readOnly); };