authorgravatar for me@t3.ggTheo Browne <me@t3.gg> 2026-09-04 10:35:27-07:00
committergravatar for noreply@github.comGitHub <noreply@github.com> 2026-09-04 10:35:27-07:00
log50bfca43d76ced00f5d67cfbab8bc44c50eb0e53
treea1fe8742e34af45f1d19632e17e5aa4a2663a71c
parentc4353bc6b972fa527579b530f01cc8744b2407d2
signature Signed by PGP key B5690EEEBB952194

perf(server): replay only the selected thread (#9726)

Read only the selected thread's events when resuming its detail stream. Measure bounded row counts and serialized payload bytes before replay, using the existing aggregate index and a captured authoritative head. Keep snapshot resets for oversized or invalid cursors. Reset recreated threads when a snapshot exists, and keep bounded replay for deleted threads whose snapshot is absent. Shell replay and the detail-event filter are unchanged. Created with GPT-6 Astra (preview) in Codex.

12 files changed, 781 insertions(+), 94 deletions(-)

apps/server/src/orchestration/Layers/OrchestrationEngine.test.ts+6
...@@ -325,6 +325,8 @@ describe("OrchestrationEngine", () => {...@@ -325,6 +325,8 @@ describe("OrchestrationEngine", () => {
325 }),325 }),
326 ),326 ),
327 hasEventAfter: () => Effect.succeed(false),327 hasEventAfter: () => Effect.succeed(false),
328 readAggregateRange: () => Stream.die("unused aggregate replay"),
329 getAggregateReplayStats: () => Effect.die("unused aggregate replay stats"),
328 };330 };
329331
330 const projectionSnapshot = {332 const projectionSnapshot = {
...@@ -1235,6 +1237,8 @@ describe("OrchestrationEngine", () => {...@@ -1235,6 +1237,8 @@ describe("OrchestrationEngine", () => {
1235 return Stream.fromIterable(events);1237 return Stream.fromIterable(events);
1236 },1238 },
1237 hasEventAfter: () => Effect.succeed(false),1239 hasEventAfter: () => Effect.succeed(false),
1240 readAggregateRange: () => Stream.die("unused aggregate replay"),
1241 getAggregateReplayStats: () => Effect.die("unused aggregate replay stats"),
1238 };1242 };
12391243
1240 const ServerConfigLayer = ServerConfig.layerTest(process.cwd(), {1244 const ServerConfigLayer = ServerConfig.layerTest(process.cwd(), {
...@@ -1473,6 +1477,8 @@ describe("OrchestrationEngine", () => {...@@ -1473,6 +1477,8 @@ describe("OrchestrationEngine", () => {
1473 return Stream.fromIterable(events);1477 return Stream.fromIterable(events);
1474 },1478 },
1475 hasEventAfter: () => Effect.succeed(false),1479 hasEventAfter: () => Effect.succeed(false),
1480 readAggregateRange: () => Stream.die("unused aggregate replay"),
1481 getAggregateReplayStats: () => Effect.die("unused aggregate replay stats"),
1476 };1482 };
14771483
1478 let shouldFailProjection = true;1484 let shouldFailProjection = true;
apps/server/src/orchestration/Layers/OrchestrationEngine.ts+15
...@@ -381,6 +381,19 @@ const makeOrchestrationEngine = Effect.gen(function* () {...@@ -381,6 +381,19 @@ const makeOrchestrationEngine = Effect.gen(function* () {
381 const readEvents: OrchestrationEngineShape["readEvents"] = (fromSequenceExclusive, limit) =>381 const readEvents: OrchestrationEngineShape["readEvents"] = (fromSequenceExclusive, limit) =>
382 eventStore.readFromSequence(fromSequenceExclusive, limit);382 eventStore.readFromSequence(fromSequenceExclusive, limit);
383383
384 const readThreadEvents: OrchestrationEngineShape["readThreadEvents"] = ({ threadId, ...range }) =>
385 eventStore.readAggregateRange({ ...range, aggregateKind: "thread", aggregateId: threadId });
386
387 const getThreadReplayStats: OrchestrationEngineShape["getThreadReplayStats"] = ({
388 threadId,
389 ...range
390 }) =>
391 eventStore.getAggregateReplayStats({
392 ...range,
393 aggregateKind: "thread",
394 aggregateId: threadId,
395 });
396
384 const dispatch: OrchestrationEngineShape["dispatch"] = (command, options) =>397 const dispatch: OrchestrationEngineShape["dispatch"] = (command, options) =>
385 Effect.gen(function* () {398 Effect.gen(function* () {
386 const result = yield* Deferred.make<{ sequence: number }, OrchestrationDispatchError>();399 const result = yield* Deferred.make<{ sequence: number }, OrchestrationDispatchError>();
...@@ -395,6 +408,8 @@ const makeOrchestrationEngine = Effect.gen(function* () {...@@ -395,6 +408,8 @@ const makeOrchestrationEngine = Effect.gen(function* () {
395408
396 return {409 return {
397 readEvents,410 readEvents,
411 readThreadEvents,
412 getThreadReplayStats,
398 dispatch,413 dispatch,
399 subscribeDomainEvents: PubSub.subscribe(eventPubSub).pipe(Effect.map(Stream.fromSubscription)),414 subscribeDomainEvents: PubSub.subscribe(eventPubSub).pipe(Effect.map(Stream.fromSubscription)),
400 // Each access creates a fresh PubSub subscription so that multiple415 // Each access creates a fresh PubSub subscription so that multiple
apps/server/src/orchestration/Layers/ProviderCommandReactor.test.ts+2
...@@ -418,6 +418,8 @@ describe("ProviderCommandReactor", () => {...@@ -418,6 +418,8 @@ describe("ProviderCommandReactor", () => {
418 const engine = yield* OrchestrationEngineService;418 const engine = yield* OrchestrationEngineService;
419 return {419 return {
420 readEvents: engine.readEvents,420 readEvents: engine.readEvents,
421 readThreadEvents: engine.readThreadEvents,
422 getThreadReplayStats: engine.getThreadReplayStats,
421 dispatch: (command) => {423 dispatch: (command) => {
422 if (command.type === "thread.title.regeneration.complete") {424 if (command.type === "thread.title.regeneration.complete") {
423 titleRegenerationCompletionDispatchAttempts += 1;425 titleRegenerationCompletionDispatchAttempts += 1;
apps/server/src/orchestration/Services/OrchestrationEngine.ts+19-2
...@@ -14,6 +14,7 @@ import type {...@@ -14,6 +14,7 @@ import type {
14 OrchestrationClientOrigin,14 OrchestrationClientOrigin,
15 OrchestrationCommand,15 OrchestrationCommand,
16 OrchestrationEvent,16 OrchestrationEvent,
17 ThreadId,
17} from "@t3tools/contracts";18} from "@t3tools/contracts";
18import * as Context from "effect/Context";19import * as Context from "effect/Context";
19import type * as Effect from "effect/Effect";20import type * as Effect from "effect/Effect";
...@@ -22,6 +23,13 @@ import type * as Stream from "effect/Stream";...@@ -22,6 +23,13 @@ import type * as Stream from "effect/Stream";
2223
23import type { OrchestrationDispatchError } from "../Errors.ts";24import type { OrchestrationDispatchError } from "../Errors.ts";
24import type { OrchestrationEventStoreError } from "../../persistence/Errors.ts";25import type { OrchestrationEventStoreError } from "../../persistence/Errors.ts";
26import type { OrchestrationAggregateReplayStats } from "../../persistence/Services/OrchestrationEventStore.ts";
27
28export interface OrchestrationThreadReplayRange {
29 readonly threadId: ThreadId;
30 readonly fromSequenceExclusive: number;
31 readonly toSequenceInclusive: number;
32}
2533
26/**34/**
27 * OrchestrationEngineShape - Service API for orchestration command and event flow.35 * OrchestrationEngineShape - Service API for orchestration command and event flow.
...@@ -33,8 +41,7 @@ export interface OrchestrationEngineShape {...@@ -33,8 +41,7 @@ export interface OrchestrationEngineShape {
33 * @param fromSequenceExclusive - Sequence cursor (exclusive).41 * @param fromSequenceExclusive - Sequence cursor (exclusive).
34 * @param limit - Maximum number of events to read. Defaults to the event42 * @param limit - Maximum number of events to read. Defaults to the event
35 * store's page-bounded default; pass a higher value when the caller must43 * store's page-bounded default; pass a higher value when the caller must
36 * read every event after the cursor (e.g. per-thread catch-up that filters44 * read a wider global range. Thread subscriptions use readThreadEvents.
37 * a small subset out of a potentially larger global range).
38 * @returns Stream containing ordered events.45 * @returns Stream containing ordered events.
39 */46 */
40 readonly readEvents: (47 readonly readEvents: (
...@@ -42,6 +49,16 @@ export interface OrchestrationEngineShape {...@@ -42,6 +49,16 @@ export interface OrchestrationEngineShape {
42 limit?: number,49 limit?: number,
43 ) => Stream.Stream<OrchestrationEvent, OrchestrationEventStoreError, never>;50 ) => Stream.Stream<OrchestrationEvent, OrchestrationEventStoreError, never>;
4451
52 /** Read only this thread's events through a captured authoritative head. */
53 readonly readThreadEvents: (
54 input: OrchestrationThreadReplayRange & { readonly limit?: number },
55 ) => Stream.Stream<OrchestrationEvent, OrchestrationEventStoreError>;
56
57 /** Measure a bounded thread replay without decoding its event bodies. */
58 readonly getThreadReplayStats: (
59 input: OrchestrationThreadReplayRange & { readonly maxEvents: number },
60 ) => Effect.Effect<OrchestrationAggregateReplayStats, OrchestrationEventStoreError>;
61
45 /**62 /**
46 * Dispatch a validated orchestration command.63 * Dispatch a validated orchestration command.
47 *64 *
apps/server/src/persistence/Layers/OrchestrationEventStore.test.ts+183-2
...@@ -1,4 +1,11 @@...@@ -1,4 +1,11 @@
1import { CommandId, EventId, ProjectId } from "@t3tools/contracts";1import {
2 CommandId,
3 EventId,
4 MessageId,
5 ProjectId,
6 ThreadId,
7 type OrchestrationEvent,
8} from "@t3tools/contracts";
2import { assert, it } from "@effect/vitest";9import { assert, it } from "@effect/vitest";
3import * as Effect from "effect/Effect";10import * as Effect from "effect/Effect";
4import * as Layer from "effect/Layer";11import * as Layer from "effect/Layer";
...@@ -12,6 +19,31 @@ import { OrchestrationEventStoreLive } from "./OrchestrationEventStore.ts";...@@ -12,6 +19,31 @@ import { OrchestrationEventStoreLive } from "./OrchestrationEventStore.ts";
12import { SqlitePersistenceMemory } from "./Sqlite.ts";19import { SqlitePersistenceMemory } from "./Sqlite.ts";
13const isPersistenceDecodeError = Schema.is(PersistenceDecodeError);20const isPersistenceDecodeError = Schema.is(PersistenceDecodeError);
1421
22function messageEvent(threadId: ThreadId, id: string): Omit<OrchestrationEvent, "sequence"> {
23 const now = "2026-01-01T00:00:00.000Z";
24 return {
25 type: "thread.message-sent",
26 eventId: EventId.make(id),
27 aggregateKind: "thread",
28 aggregateId: threadId,
29 occurredAt: now,
30 commandId: null,
31 causationEventId: null,
32 correlationId: null,
33 metadata: {},
34 payload: {
35 threadId,
36 messageId: MessageId.make(id),
37 role: "assistant",
38 text: id,
39 turnId: null,
40 streaming: false,
41 createdAt: now,
42 updatedAt: now,
43 },
44 };
45}
46
15const layer = it.layer(47const layer = it.layer(
16 OrchestrationEventStoreLive.pipe(Layer.provideMerge(SqlitePersistenceMemory)),48 OrchestrationEventStoreLive.pipe(Layer.provideMerge(SqlitePersistenceMemory)),
17);49);
...@@ -79,7 +111,7 @@ layer("OrchestrationEventStore", (it) => {...@@ -79,7 +111,7 @@ layer("OrchestrationEventStore", (it) => {
79 const sql = yield* SqlClient.SqlClient;111 const sql = yield* SqlClient.SqlClient;
80 const now = "2026-01-01T00:00:00.000Z";112 const now = "2026-01-01T00:00:00.000Z";
81113
82 yield* sql`114 const invalidRows = yield* sql<{ readonly sequence: number }>`
83 INSERT INTO orchestration_events (115 INSERT INTO orchestration_events (
84 event_id,116 event_id,
85 aggregate_kind,117 aggregate_kind,
...@@ -108,6 +140,7 @@ layer("OrchestrationEventStore", (it) => {...@@ -108,6 +140,7 @@ layer("OrchestrationEventStore", (it) => {
108 ${"{"},140 ${"{"},
109 ${"{}"}141 ${"{}"}
110 )142 )
143 RETURNING sequence
111 `;144 `;
112145
113 const replayResult = yield* Effect.result(146 const replayResult = yield* Effect.result(
...@@ -122,6 +155,154 @@ layer("OrchestrationEventStore", (it) => {...@@ -122,6 +155,154 @@ layer("OrchestrationEventStore", (it) => {
122 ),155 ),
123 );156 );
124 }157 }
158 const scopedResult = yield* eventStore
159 .readAggregateRange({
160 aggregateKind: "project",
161 aggregateId: "project-invalid-json",
162 fromSequenceExclusive: 0,
163 toSequenceInclusive: invalidRows[0]!.sequence,
164 })
165 .pipe(Stream.runCollect, Effect.result);
166 assert.equal(scopedResult._tag, "Failure");
167 if (scopedResult._tag === "Failure") {
168 assert.ok(isPersistenceDecodeError(scopedResult.failure));
169 assert.ok(
170 scopedResult.failure.operation.includes(
171 "OrchestrationEventStore.readAggregateRange:decodeRows",
172 ),
173 );
174 }
175 }),
176 );
177
178 it.effect("reads one aggregate through the captured head across pruned global gaps", () =>
179 Effect.gen(function* () {
180 const store = yield* OrchestrationEventStore;
181 const sql = yield* SqlClient.SqlClient;
182 const threadId = ThreadId.make("shared-stream-id");
183 const first = yield* store.append(messageEvent(threadId, "scoped-first"));
184 const pruned = yield* store.append(
185 messageEvent(ThreadId.make("pruned-thread"), "pruned-event"),
186 );
187 const second = yield* store.append(messageEvent(threadId, "scoped-second"));
188 // The same stream ID in a different aggregate is not part of this thread.
189 // Its invalid JSON must never reach the event decoder.
190 yield* sql`
191 INSERT INTO orchestration_events (
192 event_id, aggregate_kind, stream_id, stream_version, event_type, occurred_at,
193 actor_kind, payload_json, metadata_json
194 ) VALUES (
195 'same-id-project', 'project', ${threadId}, 0, 'project.created',
196 '2026-01-01T00:00:00.000Z', 'server', '{', '{'
197 ), (
198 'unrelated-invalid', 'thread', 'unrelated-invalid-thread', 0, 'thread.activity-appended',
199 '2026-01-01T00:00:00.000Z', 'server', '{', '{'
200 )
201 `;
202 const last = yield* store.append(messageEvent(threadId, "scoped-last"));
203 yield* sql`DELETE FROM orchestration_events WHERE sequence = ${pruned.sequence}`;
204 yield* store.append(messageEvent(threadId, "after-captured-head"));
205
206 const events = yield* store
207 .readAggregateRange({
208 aggregateKind: "thread",
209 aggregateId: threadId,
210 fromSequenceExclusive: first.sequence,
211 toSequenceInclusive: last.sequence,
212 limit: 100,
213 })
214 .pipe(Stream.runCollect);
215 assert.deepEqual(
216 events.map((event) => event.sequence),
217 [second.sequence, last.sequence],
218 );
219 }),
220 );
221
222 it.effect("bounds thread replay metadata and counts UTF-8 bytes without decoding payloads", () =>
223 Effect.gen(function* () {
224 const store = yield* OrchestrationEventStore;
225 const sql = yield* SqlClient.SqlClient;
226 const rows = yield* sql<{ readonly sequence: number }>`
227 INSERT INTO orchestration_events (
228 event_id, aggregate_kind, stream_id, stream_version, event_type, occurred_at,
229 actor_kind, payload_json, metadata_json
230 ) VALUES
231 ('stats-1', 'thread', 'stats-thread', 0, 'thread.message-sent',
232 '2026-01-01T00:00:00.000Z', 'provider', '{"output":"😀"}', '{}'),
233 ('stats-unrelated', 'thread', 'another-thread', 0, 'thread.created',
234 '2026-01-01T00:00:00.000Z', 'provider', printf('%.*c', 10000, 'x'), '{}'),
235 ('stats-2', 'thread', 'stats-thread', 1, 'thread.activity-appended',
236 '2026-01-01T00:00:00.000Z', 'provider', '{', '{}'),
237 ('stats-other-kind', 'project', 'stats-thread', 0, 'project.deleted',
238 '2026-01-01T00:00:00.000Z', 'provider', printf('%.*c', 20000, 'x'), '{}'),
239 ('stats-3', 'thread', 'stats-thread', 2, 'thread.deleted',
240 '2026-01-01T00:00:00.000Z', 'provider', '{"output":"é"}', '{}'),
241 ('stats-4', 'thread', 'stats-thread', 3, 'thread.created',
242 '2026-01-01T00:00:00.000Z', 'provider', printf('%.*c', 2000, 'x'), '{}')
243 RETURNING sequence
244 `;
245 const range = {
246 aggregateKind: "thread" as const,
247 aggregateId: "stats-thread",
248 fromSequenceExclusive: 0,
249 toSequenceInclusive: rows.at(-1)!.sequence,
250 };
251 assert.deepEqual(yield* store.getAggregateReplayStats({ ...range, maxEvents: 2 }), {
252 eventCount: 3,
253 payloadBytes: 33,
254 hasCreateEvent: false,
255 });
256 assert.deepEqual(yield* store.getAggregateReplayStats({ ...range, maxEvents: 10 }), {
257 eventCount: 4,
258 payloadBytes: 2033,
259 hasCreateEvent: true,
260 });
261 assert.deepEqual(
262 yield* store.getAggregateReplayStats({
263 ...range,
264 toSequenceInclusive: rows[2]!.sequence,
265 maxEvents: 10,
266 }),
267 {
268 eventCount: 2,
269 payloadBytes: 18,
270 hasCreateEvent: false,
271 },
272 );
273 }),
274 );
275
276 it.effect("keeps later pages below the captured head when new events are appended", () =>
277 Effect.gen(function* () {
278 const store = yield* OrchestrationEventStore;
279 const threadId = ThreadId.make("paged-thread");
280 const persisted = yield* Effect.forEach(
281 Array.from({ length: 502 }, (_, index) => index),
282 (index) => store.append(messageEvent(threadId, `paged-${index}`)),
283 );
284 const head = persisted.at(-1)!.sequence;
285 let appendedDuringReplay = false;
286 const replayed = yield* store
287 .readAggregateRange({
288 aggregateKind: "thread",
289 aggregateId: threadId,
290 fromSequenceExclusive: 0,
291 toSequenceInclusive: head,
292 limit: 1_000,
293 })
294 .pipe(
295 Stream.tap(() => {
296 if (appendedDuringReplay) return Effect.void;
297 appendedDuringReplay = true;
298 return store.append(messageEvent(threadId, "appended-during-replay"));
299 }),
300 Stream.runCollect,
301 );
302 assert.deepEqual(
303 replayed.map((event) => event.sequence),
304 persisted.map((event) => event.sequence),
305 );
125 }),306 }),
126 );307 );
127});308});
apps/server/src/persistence/Layers/OrchestrationEventStore.ts+132
...@@ -72,6 +72,18 @@ const ReadFromSequenceRequestSchema = Schema.Struct({...@@ -72,6 +72,18 @@ const ReadFromSequenceRequestSchema = Schema.Struct({
72 sequenceExclusive: NonNegativeInt,72 sequenceExclusive: NonNegativeInt,
73 limit: Schema.Number,73 limit: Schema.Number,
74});74});
75const AggregateReplayRequestSchema = Schema.Struct({
76 aggregateKind: OrchestrationAggregateKind,
77 aggregateId: Schema.String,
78 fromSequenceExclusive: NonNegativeInt,
79 toSequenceInclusive: NonNegativeInt,
80 limit: Schema.Number,
81});
82const AggregateReplayStatsRowSchema = Schema.Struct({
83 eventCount: Schema.Number,
84 payloadBytes: Schema.Number,
85 hasCreateEvent: Schema.Number,
86});
75const DEFAULT_READ_FROM_SEQUENCE_LIMIT = 1_000;87const DEFAULT_READ_FROM_SEQUENCE_LIMIT = 1_000;
76const READ_PAGE_SIZE = 500;88const READ_PAGE_SIZE = 500;
7789
...@@ -189,6 +201,57 @@ const makeEventStore = Effect.gen(function* () {...@@ -189,6 +201,57 @@ const makeEventStore = Effect.gen(function* () {
189 `,201 `,
190 });202 });
191203
204 const readAggregateEventRows = SqlSchema.findAll({
205 Request: AggregateReplayRequestSchema,
206 Result: OrchestrationEventPersistedRowSchema,
207 execute: (request) =>
208 sql`
209 SELECT
210 sequence,
211 event_id AS "eventId",
212 event_type AS "type",
213 aggregate_kind AS "aggregateKind",
214 stream_id AS "aggregateId",
215 occurred_at AS "occurredAt",
216 command_id AS "commandId",
217 causation_event_id AS "causationEventId",
218 correlation_id AS "correlationId",
219 payload_json AS "payload",
220 metadata_json AS "metadata"
221 FROM orchestration_events
222 WHERE aggregate_kind = ${request.aggregateKind}
223 AND stream_id = ${request.aggregateId}
224 AND sequence > ${request.fromSequenceExclusive}
225 AND sequence <= ${request.toSequenceInclusive}
226 ORDER BY sequence ASC
227 LIMIT ${request.limit}
228 `,
229 });
230
231 const readAggregateReplayStats = SqlSchema.findOne({
232 Request: AggregateReplayRequestSchema,
233 Result: AggregateReplayStatsRowSchema,
234 execute: (request) =>
235 sql`
236 SELECT
237 COUNT(*) AS "eventCount",
238 COALESCE(SUM(octet_length(payload_json)), 0) AS "payloadBytes",
239 COALESCE(MAX(event_type IN (
240 'thread.created', 'project.created'
241 )), 0) AS "hasCreateEvent"
242 FROM (
243 SELECT payload_json, event_type
244 FROM orchestration_events
245 WHERE aggregate_kind = ${request.aggregateKind}
246 AND stream_id = ${request.aggregateId}
247 AND sequence > ${request.fromSequenceExclusive}
248 AND sequence <= ${request.toSequenceInclusive}
249 ORDER BY sequence ASC
250 LIMIT ${request.limit}
251 )
252 `,
253 });
254
192 const append: OrchestrationEventStoreShape["append"] = (event) =>255 const append: OrchestrationEventStoreShape["append"] = (event) =>
193 appendEventRow({256 appendEventRow({
194 eventId: event.eventId,257 eventId: event.eventId,
...@@ -295,9 +358,78 @@ const makeEventStore = Effect.gen(function* () {...@@ -295,9 +358,78 @@ const makeEventStore = Effect.gen(function* () {
295 ),358 ),
296 );359 );
297360
361 const readAggregateRange: OrchestrationEventStoreShape["readAggregateRange"] = (input) => {
362 const limit = Math.max(0, Math.floor(input.limit ?? DEFAULT_READ_FROM_SEQUENCE_LIMIT));
363 if (limit === 0 || input.fromSequenceExclusive >= input.toSequenceInclusive) {
364 return Stream.empty;
365 }
366 const readPage = (
367 cursor: number,
368 remaining: number,
369 ): Stream.Stream<OrchestrationEvent, OrchestrationEventStoreError> =>
370 Stream.fromEffect(
371 readAggregateEventRows({
372 ...input,
373 fromSequenceExclusive: cursor,
374 limit: Math.min(remaining, READ_PAGE_SIZE),
375 }).pipe(
376 Effect.mapError(
377 toPersistenceSqlOrDecodeError(
378 "OrchestrationEventStore.readAggregateRange:query",
379 "OrchestrationEventStore.readAggregateRange:decodeRows",
380 ),
381 ),
382 Effect.flatMap((rows) =>
383 Effect.forEach(rows, (row) =>
384 decodeEvent(row).pipe(
385 Effect.mapError(
386 toPersistenceDecodeError("OrchestrationEventStore.readAggregateRange:rowToEvent"),
387 ),
388 ),
389 ),
390 ),
391 ),
392 ).pipe(
393 Stream.flatMap((events) => {
394 const last = events.at(-1);
395 if (last === undefined) {
396 return Stream.empty;
397 }
398 const nextRemaining = remaining - events.length;
399 if (
400 events.length < READ_PAGE_SIZE ||
401 nextRemaining === 0 ||
402 last.sequence >= input.toSequenceInclusive
403 ) {
404 return Stream.fromIterable(events);
405 }
406 return Stream.concat(Stream.fromIterable(events), readPage(last.sequence, nextRemaining));
407 }),
408 );
409 return readPage(input.fromSequenceExclusive, limit);
410 };
411
412 const getAggregateReplayStats: OrchestrationEventStoreShape["getAggregateReplayStats"] = (
413 input,
414 ) =>
415 readAggregateReplayStats({
416 ...input,
417 limit: Math.max(0, Math.floor(input.maxEvents)) + 1,
418 }).pipe(
419 Effect.mapError(
420 toPersistenceSqlOrDecodeError(
421 "OrchestrationEventStore.getAggregateReplayStats:query",
422 "OrchestrationEventStore.getAggregateReplayStats:decodeRow",
423 ),
424 ),
425 Effect.map((row) => ({ ...row, hasCreateEvent: row.hasCreateEvent !== 0 })),
426 );
427
298 return {428 return {
299 append,429 append,
300 readFromSequence,430 readFromSequence,
431 readAggregateRange,
432 getAggregateReplayStats,
301 readAll: () => readFromSequence(0, Number.MAX_SAFE_INTEGER),433 readAll: () => readFromSequence(0, Number.MAX_SAFE_INTEGER),
302 hasEventAfter,434 hasEventAfter,
303 } satisfies OrchestrationEventStoreShape;435 } satisfies OrchestrationEventStoreShape;
apps/server/src/persistence/Services/OrchestrationEventStore.ts+27
...@@ -16,6 +16,20 @@ import type * as Stream from "effect/Stream";...@@ -16,6 +16,20 @@ import type * as Stream from "effect/Stream";
1616
17import type { OrchestrationEventStoreError } from "../Errors.ts";17import type { OrchestrationEventStoreError } from "../Errors.ts";
1818
19export interface OrchestrationAggregateReplayRange {
20 readonly aggregateKind: OrchestrationEvent["aggregateKind"];
21 readonly aggregateId: string;
22 readonly fromSequenceExclusive: number;
23 readonly toSequenceInclusive: number;
24}
25
26export interface OrchestrationAggregateReplayStats {
27 readonly eventCount: number;
28 readonly payloadBytes: number;
29 /** A creation in this range does not prove that the aggregate still exists. */
30 readonly hasCreateEvent: boolean;
31}
32
19/**33/**
20 * OrchestrationEventStoreShape - Service API for orchestration event persistence.34 * OrchestrationEventStoreShape - Service API for orchestration event persistence.
21 */35 */
...@@ -46,6 +60,19 @@ export interface OrchestrationEventStoreShape {...@@ -46,6 +60,19 @@ export interface OrchestrationEventStoreShape {
46 limit?: number,60 limit?: number,
47 ) => Stream.Stream<OrchestrationEvent, OrchestrationEventStoreError>;61 ) => Stream.Stream<OrchestrationEvent, OrchestrationEventStoreError>;
4862
63 /** Read one aggregate through a captured global head, without decoding other streams. */
64 readonly readAggregateRange: (
65 input: OrchestrationAggregateReplayRange & { readonly limit?: number },
66 ) => Stream.Stream<OrchestrationEvent, OrchestrationEventStoreError>;
67
68 /**
69 * Measure at most maxEvents + 1 rows without decoding payloads. The extra
70 * row tells the caller to use a snapshot instead of a truncated replay.
71 */
72 readonly getAggregateReplayStats: (
73 input: OrchestrationAggregateReplayRange & { readonly maxEvents: number },
74 ) => Effect.Effect<OrchestrationAggregateReplayStats, OrchestrationEventStoreError>;
75
49 /**76 /**
50 * Read all events from the beginning of the stream.77 * Read all events from the beginning of the stream.
51 *78 *
apps/server/src/relay/AgentAwarenessRelay.test.ts+4
...@@ -459,6 +459,8 @@ describe.sequential("signRelayAgentActivityPublishProof", () => {...@@ -459,6 +459,8 @@ describe.sequential("signRelayAgentActivityPublishProof", () => {
459459
460 const orchestrationEngine = {460 const orchestrationEngine = {
461 readEvents: () => Stream.empty,461 readEvents: () => Stream.empty,
462 readThreadEvents: () => Stream.empty,
463 getThreadReplayStats: () => Effect.die("unused thread replay stats"),
462 dispatch: () => Effect.succeed({ sequence: 1 }),464 dispatch: () => Effect.succeed({ sequence: 1 }),
463 streamDomainEvents: Stream.fromQueue(events),465 streamDomainEvents: Stream.fromQueue(events),
464 subscribeDomainEvents: Effect.succeed(Stream.fromQueue(events)),466 subscribeDomainEvents: Effect.succeed(Stream.fromQueue(events)),
...@@ -653,6 +655,8 @@ describe.sequential("signRelayAgentActivityPublishProof", () => {...@@ -653,6 +655,8 @@ describe.sequential("signRelayAgentActivityPublishProof", () => {
653 }),655 }),
654 Layer.succeed(OrchestrationEngineService, {656 Layer.succeed(OrchestrationEngineService, {
655 readEvents: () => Stream.empty,657 readEvents: () => Stream.empty,
658 readThreadEvents: () => Stream.empty,
659 getThreadReplayStats: () => Effect.die("unused thread replay stats"),
656 dispatch: () => Effect.succeed({ sequence: 1 }),660 dispatch: () => Effect.succeed({ sequence: 1 }),
657 streamDomainEvents: Stream.fromQueue(events),661 streamDomainEvents: Stream.fromQueue(events),
658 subscribeDomainEvents: Effect.succeed(Stream.fromQueue(events)),662 subscribeDomainEvents: Effect.succeed(Stream.fromQueue(events)),
apps/server/src/server.test.ts+337-68
...@@ -115,6 +115,8 @@ import {...@@ -115,6 +115,8 @@ import {
115import * as ProjectionSnapshotQuery from "./orchestration/Services/ProjectionSnapshotQuery.ts";115import * as ProjectionSnapshotQuery from "./orchestration/Services/ProjectionSnapshotQuery.ts";
116import { ThreadDeletionReactor } from "./orchestration/Services/ThreadDeletionReactor.ts";116import { ThreadDeletionReactor } from "./orchestration/Services/ThreadDeletionReactor.ts";
117import { SqlitePersistenceMemory } from "./persistence/Layers/Sqlite.ts";117import { SqlitePersistenceMemory } from "./persistence/Layers/Sqlite.ts";
118import { OrchestrationEventStoreLive } from "./persistence/Layers/OrchestrationEventStore.ts";
119import { OrchestrationEventStore } from "./persistence/Services/OrchestrationEventStore.ts";
118import { PersistenceSqlError } from "./persistence/Errors.ts";120import { PersistenceSqlError } from "./persistence/Errors.ts";
119import * as ProviderRegistry from "./provider/Services/ProviderRegistry.ts";121import * as ProviderRegistry from "./provider/Services/ProviderRegistry.ts";
120import * as ProviderService from "./provider/Services/ProviderService.ts";122import * as ProviderService from "./provider/Services/ProviderService.ts";
...@@ -918,6 +920,13 @@ const buildAppUnderTest = (options?: {...@@ -918,6 +920,13 @@ const buildAppUnderTest = (options?: {
918 Layer.mergeAll(920 Layer.mergeAll(
919 Layer.mock(OrchestrationEngine.OrchestrationEngineService)({921 Layer.mock(OrchestrationEngine.OrchestrationEngineService)({
920 readEvents: () => Stream.empty,922 readEvents: () => Stream.empty,
923 readThreadEvents: () => Stream.empty,
924 getThreadReplayStats: () =>
925 Effect.succeed({
926 eventCount: 0,
927 payloadBytes: 0,
928 hasCreateEvent: false,
929 }),
921 dispatch: () => Effect.succeed({ sequence: 0 }),930 dispatch: () => Effect.succeed({ sequence: 0 }),
922 streamDomainEvents: Stream.empty,931 streamDomainEvents: Stream.empty,
923 latestSequence: Effect.succeed(0),932 latestSequence: Effect.succeed(0),
...@@ -8047,50 +8056,57 @@ it.layer(NodeServices.layer)("server router seam", (it) => {...@@ -8047,50 +8056,57 @@ it.layer(NodeServices.layer)("server router seam", (it) => {
8047 }).pipe(Effect.provide(NodeHttpServer.layerTest), TestClock.withLive),8056 }).pipe(Effect.provide(NodeHttpServer.layerTest), TestClock.withLive),
8048 );8057 );
80498058
8050 it.effect("subscribeThread sends a fresh snapshot instead of replaying a large gap", () =>8059 it.effect(
8051 Effect.gen(function* () {8060 "subscribeThread sends a fresh snapshot when its event count exceeds the replay limit",
8052 let readEventsCalls = 0;8061 () =>
8053 const thread = makeDefaultOrchestrationReadModel().threads[0]!;8062 Effect.gen(function* () {
8063 let readEventsCalls = 0;
8064 const thread = makeDefaultOrchestrationReadModel().threads[0]!;
80548065
8055 yield* buildAppUnderTest({8066 yield* buildAppUnderTest({
8056 layers: {8067 layers: {
8057 orchestrationEngine: {8068 orchestrationEngine: {
8058 // Head is far ahead of the client's afterSequence (gap > 1000).8069 latestSequence: Effect.succeed(100_000),
8059 latestSequence: Effect.succeed(100_000),8070 getThreadReplayStats: () =>
8060 readEvents: () =>8071 Effect.succeed({
8061 Stream.sync(() => {8072 eventCount: 1_001,
8062 readEventsCalls += 1;8073 payloadBytes: 1_000,
8063 return {} as OrchestrationEvent;8074 hasCreateEvent: false,
8064 }),8075 }),
8065 },8076 readThreadEvents: () =>
8066 projectionSnapshotQuery: {8077 Stream.sync(() => {
8067 getThreadDetailSnapshot: () =>8078 readEventsCalls += 1;
8068 Effect.succeed(Option.some({ snapshotSequence: 100_000, thread })),8079 return {} as OrchestrationEvent;
8080 }),
8081 },
8082 projectionSnapshotQuery: {
8083 getThreadDetailSnapshot: () =>
8084 Effect.succeed(Option.some({ snapshotSequence: 100_000, thread })),
8085 },
8069 },8086 },
8070 },8087 });
8071 });
80728088
8073 const wsUrl = yield* getWsServerUrl("/ws");8089 const wsUrl = yield* getWsServerUrl("/ws");
8074 const items = yield* Effect.scoped(8090 const items = yield* Effect.scoped(
8075 withWsRpcClient(wsUrl, (client) =>8091 withWsRpcClient(wsUrl, (client) =>
8076 client[ORCHESTRATION_WS_METHODS.subscribeThread]({8092 client[ORCHESTRATION_WS_METHODS.subscribeThread]({
8077 threadId: defaultThreadId,8093 threadId: defaultThreadId,
8078 afterSequence: 5,8094 afterSequence: 5,
8079 requestCompletionMarker: true,8095 requestCompletionMarker: true,
8080 }).pipe(Stream.take(2), Stream.runCollect),8096 }).pipe(Stream.take(2), Stream.runCollect),
8081 ),8097 ),
8082 );8098 );
80838099
8084 const [first, second] = Array.from(items);8100 const [first, second] = Array.from(items);
8085 // Large gap => fresh thread snapshot, and the global replay never starts.8101 // Never truncate a thread's replay at the event limit.
8086 assert.equal(first?.kind, "snapshot");8102 assert.equal(first?.kind, "snapshot");
8087 if (first?.kind === "snapshot") {8103 if (first?.kind === "snapshot") {
8088 assert.equal(first.snapshot.thread.id, defaultThreadId);8104 assert.equal(first.snapshot.thread.id, defaultThreadId);
8089 assert.equal(first.snapshot.snapshotSequence, 100_000);8105 assert.equal(first.snapshot.snapshotSequence, 100_000);
8090 }8106 }
8091 assert.equal(second?.kind, "synchronized");8107 assert.equal(second?.kind, "synchronized");
8092 assert.equal(readEventsCalls, 0);8108 assert.equal(readEventsCalls, 0);
8093 }).pipe(Effect.provide(NodeHttpServer.layerTest)),8109 }).pipe(Effect.provide(NodeHttpServer.layerTest)),
8094 );8110 );
80958111
8096 it.effect(8112 it.effect(
...@@ -8104,7 +8120,7 @@ it.layer(NodeServices.layer)("server router seam", (it) => {...@@ -8104,7 +8120,7 @@ it.layer(NodeServices.layer)("server router seam", (it) => {
8104 const releaseAck = yield* Deferred.make<void>();8120 const releaseAck = yield* Deferred.make<void>();
8105 const firstApplied = yield* Deferred.make<void>();8121 const firstApplied = yield* Deferred.make<void>();
8106 const replayStarted = yield* Deferred.make<void>();8122 const replayStarted = yield* Deferred.make<void>();
8107 const replayCalls: Array<{ afterSequence: number; limit: number | undefined }> = [];8123 const replayCalls: Array<{ afterSequence: number; headSequence: number }> = [];
8108 let headSequence = 0;8124 let headSequence = 0;
8109 let snapshotCalls = 0;8125 let snapshotCalls = 0;
8110 const message = (sequence: number, text: string) =>8126 const message = (sequence: number, text: string) =>
...@@ -8148,24 +8164,21 @@ it.layer(NodeServices.layer)("server router seam", (it) => {...@@ -8148,24 +8164,21 @@ it.layer(NodeServices.layer)("server router seam", (it) => {
8148 return Stream.fromSubscription(subscription);8164 return Stream.fromSubscription(subscription);
8149 }),8165 }),
8150 ).pipe(Stream.ensuring(Deferred.succeed(detached, undefined))),8166 ).pipe(Stream.ensuring(Deferred.succeed(detached, undefined))),
8151 readEvents: (afterSequence, limit) => {8167 readThreadEvents: ({ fromSequenceExclusive, toSequenceInclusive }) => {
8152 replayCalls.push({ afterSequence, limit });8168 replayCalls.push({
8169 afterSequence: fromSequenceExclusive,
8170 headSequence: toSequenceInclusive,
8171 });
8153 const range = events.filter(8172 const range = events.filter(
8154 (event) => event.sequence > afterSequence && event.sequence <= headSequence,8173 (event) =>
8174 event.sequence > fromSequenceExclusive && event.sequence <= toSequenceInclusive,
8155 );8175 );
8156 return Stream.concat(8176 return Stream.concat(
8157 Stream.fromEffect(Deferred.succeed(replayStarted, undefined)).pipe(Stream.drain),8177 Stream.fromEffect(Deferred.succeed(replayStarted, undefined)).pipe(Stream.drain),
8158 Stream.fromIterable(range),8178 Stream.fromIterable(range),
8159 );8179 );
8160 },8180 },
8161 },8181 getThreadReplayStats: ({ fromSequenceExclusive, toSequenceInclusive }) => {
8162 projectionSnapshotQuery: {
8163 getThreadDetailSnapshot: () =>
8164 Effect.sync(() => {
8165 snapshotCalls += 1;
8166 return Option.none();
8167 }),
8168 getEventReplayStats: ({ fromSequenceExclusive, toSequenceInclusive }) => {
8169 const range = events.filter(8182 const range = events.filter(
8170 (event) =>8183 (event) =>
8171 event.sequence > fromSequenceExclusive && event.sequence <= toSequenceInclusive,8184 event.sequence > fromSequenceExclusive && event.sequence <= toSequenceInclusive,
...@@ -8176,9 +8189,17 @@ it.layer(NodeServices.layer)("server router seam", (it) => {...@@ -8176,9 +8189,17 @@ it.layer(NodeServices.layer)("server router seam", (it) => {
8176 (bytes, event) => bytes + Buffer.byteLength(jsonRequestBody(event.payload)),8189 (bytes, event) => bytes + Buffer.byteLength(jsonRequestBody(event.payload)),
8177 0,8190 0,
8178 ),8191 ),
8192 hasCreateEvent: false,
8179 });8193 });
8180 },8194 },
8181 },8195 },
8196 projectionSnapshotQuery: {
8197 getThreadDetailSnapshot: () =>
8198 Effect.sync(() => {
8199 snapshotCalls += 1;
8200 return Option.none();
8201 }),
8202 },
8182 },8203 },
8183 });8204 });
81848205
...@@ -8242,8 +8263,8 @@ it.layer(NodeServices.layer)("server router seam", (it) => {...@@ -8242,8 +8263,8 @@ it.layer(NodeServices.layer)("server router seam", (it) => {
8242 );8263 );
8243 assert.equal(completed.event.payload.activity.kind, "tool.completed");8264 assert.equal(completed.event.payload.activity.kind, "tool.completed");
8244 assert.deepEqual(replayCalls, [8265 assert.deepEqual(replayCalls, [
8245 { afterSequence: 0, limit: 0 },8266 { afterSequence: 0, headSequence: 0 },
8246 { afterSequence: 1, limit: 3 },8267 { afterSequence: 1, headSequence: 4 },
8247 ]);8268 ]);
8248 assert.equal(snapshotCalls, 0);8269 assert.equal(snapshotCalls, 0);
8249 }),8270 }),
...@@ -8369,7 +8390,9 @@ it.layer(NodeServices.layer)("server router seam", (it) => {...@@ -8369,7 +8390,9 @@ it.layer(NodeServices.layer)("server router seam", (it) => {
8369 layers: {8390 layers: {
8370 orchestrationEngine: {8391 orchestrationEngine: {
8371 latestSequence: Effect.succeed(5),8392 latestSequence: Effect.succeed(5),
8372 readEvents: () =>8393 getThreadReplayStats: () =>
8394 Effect.die("An invalid cursor must not start a replay query"),
8395 readThreadEvents: () =>
8373 Stream.sync(() => {8396 Stream.sync(() => {
8374 readEventsCalls += 1;8397 readEventsCalls += 1;
8375 return {} as OrchestrationEvent;8398 return {} as OrchestrationEvent;
...@@ -8397,9 +8420,225 @@ it.layer(NodeServices.layer)("server router seam", (it) => {...@@ -8397,9 +8420,225 @@ it.layer(NodeServices.layer)("server router seam", (it) => {
8397 }).pipe(Effect.provide(NodeHttpServer.layerTest)),8420 }).pipe(Effect.provide(NodeHttpServer.layerTest)),
8398 );8421 );
83998422
8423 it.effect("subscribeThread replays a small thread range across a large global gap", () =>
8424 Effect.gen(function* () {
8425 const event = makeLiveToolActivityEvent(99_999, "tool.completed");
8426 yield* buildAppUnderTest({
8427 layers: {
8428 orchestrationEngine: {
8429 latestSequence: Effect.succeed(100_000),
8430 getThreadReplayStats: () =>
8431 Effect.succeed({
8432 eventCount: 1,
8433 payloadBytes: Buffer.byteLength(jsonRequestBody(event.payload)),
8434 hasCreateEvent: false,
8435 }),
8436 readThreadEvents: () => Stream.make(event),
8437 readEvents: () => Stream.die("Thread replay must not read the global log"),
8438 },
8439 projectionSnapshotQuery: {
8440 getEventReplayStats: () => Effect.die("Thread replay must not measure the global log"),
8441 getThreadDetailSnapshot: () =>
8442 Effect.die("Unrelated activity must not force a snapshot"),
8443 },
8444 },
8445 });
8446 const wsUrl = yield* getWsServerUrl("/ws");
8447 const items = yield* Effect.scoped(
8448 withWsRpcClient(wsUrl, (client) =>
8449 client[ORCHESTRATION_WS_METHODS.subscribeThread]({
8450 threadId: defaultThreadId,
8451 afterSequence: 5,
8452 requestCompletionMarker: true,
8453 }).pipe(
8454 Stream.takeUntil((item) => item.kind === "synchronized"),
8455 Stream.runCollect,
8456 ),
8457 ),
8458 );
8459 assert.deepEqual(
8460 items.map((item) => (item.kind === "event" ? item.event.sequence : item.kind)),
8461 [99_999, "synchronized"],
8462 );
8463 const first = items[0];
8464 assertTrue(first?.kind === "event" && first.event.type === "thread.activity-appended");
8465 assert.equal(first.event.payload.activity.kind, "tool.completed");
8466 }).pipe(Effect.provide(NodeHttpServer.layerTest)),
8467 );
8468
8469 it.effect("subscribeThread resets cached history when its ID is created again", () =>
8470 Effect.gen(function* () {
8471 const thread = {
8472 ...makeDefaultOrchestrationReadModel().threads[0]!,
8473 title: "Recreated thread",
8474 };
8475 let requestedTurnLimit: number | undefined;
8476 yield* buildAppUnderTest({
8477 layers: {
8478 orchestrationEngine: {
8479 latestSequence: Effect.succeed(5),
8480 getThreadReplayStats: () =>
8481 Effect.succeed({
8482 eventCount: 3,
8483 payloadBytes: 1_000,
8484 hasCreateEvent: true,
8485 }),
8486 readThreadEvents: () => Stream.die("A recreated thread must not reuse cached history"),
8487 },
8488 projectionSnapshotQuery: {
8489 getThreadDetailSnapshot: (_threadId, options) => {
8490 requestedTurnLimit = options?.turnLimit;
8491 return Effect.succeed(Option.some({ snapshotSequence: 5, thread }));
8492 },
8493 },
8494 },
8495 });
8496 const wsUrl = yield* getWsServerUrl("/ws");
8497 const items = yield* Effect.scoped(
8498 withWsRpcClient(wsUrl, (client) =>
8499 client[ORCHESTRATION_WS_METHODS.subscribeThread]({
8500 threadId: defaultThreadId,
8501 afterSequence: 2,
8502 turnLimit: 1,
8503 requestCompletionMarker: true,
8504 }).pipe(
8505 Stream.takeUntil((item) => item.kind === "synchronized"),
8506 Stream.runCollect,
8507 ),
8508 ),
8509 );
8510 const first = items[0];
8511 assertTrue(first?.kind === "snapshot");
8512 assert.equal(first.snapshot.thread.title, "Recreated thread");
8513 assert.equal(first.snapshot.snapshotSequence, 5);
8514 assert.equal(requestedTurnLimit, 1);
8515 assert.deepEqual(items[1], { kind: "synchronized" });
8516 }).pipe(Effect.provide(NodeHttpServer.layerTest)),
8517 );
8518
8519 for (const { createBeforeDelete, oversized } of [
8520 { createBeforeDelete: false, oversized: false },
8521 { createBeforeDelete: true, oversized: false },
8522 { createBeforeDelete: true, oversized: true },
8523 ]) {
8524 it.effect(
8525 oversized
8526 ? "keeps the missing-snapshot error when an absent thread exceeds the replay limit"
8527 : `synchronizes an absent thread and removes its shell after ${createBeforeDelete ? "creation and deletion" : "deletion"}`,
8528 () =>
8529 Effect.gen(function* () {
8530 const store = yield* OrchestrationEventStore;
8531 const base = makeLiveToolActivityEvent(0, "tool.completed");
8532 if (createBeforeDelete) {
8533 const thread = makeDefaultOrchestrationReadModel().threads[0]!;
8534 yield* store.append({
8535 ...base,
8536 eventId: EventId.make(`create-before-final-delete-${oversized}`),
8537 type: "thread.created",
8538 payload: {
8539 threadId: defaultThreadId,
8540 projectId: thread.projectId,
8541 title: thread.title,
8542 modelSelection: thread.modelSelection,
8543 runtimeMode: thread.runtimeMode,
8544 interactionMode: thread.interactionMode,
8545 branch: thread.branch,
8546 worktreePath: thread.worktreePath,
8547 createdAt: thread.createdAt,
8548 updatedAt: thread.updatedAt,
8549 },
8550 });
8551 }
8552 if (oversized) {
8553 yield* Effect.forEach(
8554 Array.from({ length: 1_000 }, (_, index) => index + 1),
8555 (sequence) => store.append(makeLiveToolActivityEvent(sequence, "tool.completed")),
8556 { discard: true },
8557 );
8558 }
8559 const deleted = yield* store.append({
8560 ...base,
8561 eventId: EventId.make(`deleted-replay-${createBeforeDelete}-${oversized}`),
8562 type: "thread.deleted",
8563 payload: { threadId: defaultThreadId, deletedAt: base.occurredAt },
8564 });
8565 yield* buildAppUnderTest({
8566 layers: {
8567 orchestrationEngine: {
8568 latestSequence: Effect.succeed(deleted.sequence),
8569 getThreadReplayStats: ({ threadId, ...range }) =>
8570 store.getAggregateReplayStats({
8571 ...range,
8572 aggregateKind: "thread",
8573 aggregateId: threadId,
8574 }),
8575 readThreadEvents: ({ threadId, ...range }) =>
8576 store.readAggregateRange({
8577 ...range,
8578 aggregateKind: "thread",
8579 aggregateId: threadId,
8580 }),
8581 readEvents: store.readFromSequence,
8582 },
8583 projectionSnapshotQuery: {
8584 getThreadDetailSnapshot: () => Effect.succeed(Option.none()),
8585 },
8586 },
8587 });
8588 const wsUrl = yield* getWsServerUrl("/ws");
8589 yield* Effect.scoped(
8590 withWsRpcClient(wsUrl, (client) =>
8591 Effect.gen(function* () {
8592 const threadResult = yield* client[ORCHESTRATION_WS_METHODS.subscribeThread]({
8593 threadId: defaultThreadId,
8594 afterSequence: 0,
8595 requestCompletionMarker: true,
8596 }).pipe(
8597 Stream.takeUntil((item) => item.kind === "synchronized"),
8598 Stream.runCollect,
8599 Effect.result,
8600 );
8601 if (oversized) {
8602 assertTrue(threadResult._tag === "Failure");
8603 assert.equal(threadResult.failure._tag, "OrchestrationGetSnapshotError");
8604 assert.equal(
8605 threadResult.failure.message,
8606 `Thread ${defaultThreadId} was not found`,
8607 );
8608 return;
8609 }
8610 assertTrue(threadResult._tag === "Success");
8611 assert.deepEqual(threadResult.success, [{ kind: "synchronized" }]);
8612 const shellItems = yield* client[ORCHESTRATION_WS_METHODS.subscribeShell]({
8613 afterSequence: 0,
8614 requestCompletionMarker: true,
8615 }).pipe(
8616 Stream.takeUntil((item) => item.kind === "synchronized"),
8617 Stream.runCollect,
8618 );
8619 assert.deepEqual(shellItems, [
8620 { kind: "thread-removed", sequence: deleted.sequence, threadId: defaultThreadId },
8621 { kind: "synchronized" },
8622 ]);
8623 }),
8624 ),
8625 );
8626 }).pipe(
8627 Effect.provide(
8628 Layer.mergeAll(
8629 OrchestrationEventStoreLive.pipe(Layer.provideMerge(SqlitePersistenceMemory)),
8630 NodeHttpServer.layerTest,
8631 ),
8632 ),
8633 ),
8634 );
8635 }
8636
8400 it.effect("subscribeThread bounds catch-up replay to the captured head", () =>8637 it.effect("subscribeThread bounds catch-up replay to the captured head", () =>
8401 Effect.gen(function* () {8638 Effect.gen(function* () {
8402 let replayLimit: number | undefined;8639 let replayHead: number | undefined;
8640 let headSequence = 50;
8641 const liveEvents = yield* PubSub.unbounded<OrchestrationEvent>();
8403 const now = "2026-01-01T00:00:00.000Z";8642 const now = "2026-01-01T00:00:00.000Z";
8404 const messageEvent = {8643 const messageEvent = {
8405 sequence: 3,8644 sequence: 3,
...@@ -8427,10 +8666,26 @@ it.layer(NodeServices.layer)("server router seam", (it) => {...@@ -8427,10 +8666,26 @@ it.layer(NodeServices.layer)("server router seam", (it) => {
8427 yield* buildAppUnderTest({8666 yield* buildAppUnderTest({
8428 layers: {8667 layers: {
8429 orchestrationEngine: {8668 orchestrationEngine: {
8430 latestSequence: Effect.succeed(50),8669 latestSequence: Effect.sync(() => headSequence),
8431 readEvents: (_afterSequence, limit) => {8670 streamDomainEvents: Stream.fromPubSub(liveEvents),
8432 replayLimit = limit;8671 getThreadReplayStats: () =>
8433 return Stream.make(messageEvent);8672 Effect.sync(() => {
8673 headSequence = 100;
8674 return { eventCount: 1, payloadBytes: 100, hasCreateEvent: false };
8675 }),
8676 readThreadEvents: ({ toSequenceInclusive }) => {
8677 replayHead = toSequenceInclusive;
8678 return Stream.fromEffect(
8679 PubSub.publish(liveEvents, {
8680 ...messageEvent,
8681 sequence: 51,
8682 eventId: EventId.make("event-live-after-head"),
8683 payload: {
8684 ...messageEvent.payload,
8685 messageId: MessageId.make("message-live-after-head"),
8686 },
8687 }),
8688 ).pipe(Stream.flatMap(() => Stream.make(messageEvent)));
8434 },8689 },
8435 },8690 },
8436 },8691 },
...@@ -8443,17 +8698,18 @@ it.layer(NodeServices.layer)("server router seam", (it) => {...@@ -8443,17 +8698,18 @@ it.layer(NodeServices.layer)("server router seam", (it) => {
8443 threadId: defaultThreadId,8698 threadId: defaultThreadId,
8444 afterSequence: 0,8699 afterSequence: 0,
8445 requestCompletionMarker: true,8700 requestCompletionMarker: true,
8446 }).pipe(Stream.take(2), Stream.runCollect),8701 }).pipe(
8702 Stream.takeUntil((item) => item.kind === "synchronized"),
8703 Stream.runCollect,
8704 ),
8447 ),8705 ),
8448 );8706 );
84498707
8450 const [first, second] = Array.from(items);8708 assert.deepEqual(
8451 assert.equal(first?.kind, "event");8709 items.map((item) => (item.kind === "event" ? item.event.sequence : item.kind)),
8452 assert.equal(first?.kind === "event" ? first.event.sequence : null, 3);8710 [3, 51, "synchronized"],
8453 assert.equal(second?.kind, "synchronized");8711 );
8454 // The replay is bounded to the head captured before the read, not8712 assert.equal(replayHead, 50);
8455 // Number.MAX_SAFE_INTEGER.
8456 assert.equal(replayLimit, 50);
8457 }).pipe(Effect.provide(NodeHttpServer.layerTest)),8713 }).pipe(Effect.provide(NodeHttpServer.layerTest)),
8458 );8714 );
84598715
...@@ -8530,6 +8786,19 @@ it.layer(NodeServices.layer)("server router seam", (it) => {...@@ -8530,6 +8786,19 @@ it.layer(NodeServices.layer)("server router seam", (it) => {
8530 layers: {8786 layers: {
8531 orchestrationEngine: {8787 orchestrationEngine: {
8532 latestSequence: Effect.succeed(5),8788 latestSequence: Effect.succeed(5),
8789 getThreadReplayStats: () =>
8790 Effect.sync(() => {
8791 replayStatsCalls += 1;
8792 return {
8793 eventCount: 5,
8794 payloadBytes: 8 * 1024 * 1024 + 1,
8795 hasCreateEvent: false,
8796 };
8797 }),
8798 readThreadEvents: () => {
8799 readEventsCalls += 1;
8800 return Stream.empty;
8801 },
8533 readEvents: () =>8802 readEvents: () =>
8534 Stream.sync(() => {8803 Stream.sync(() => {
8535 readEventsCalls += 1;8804 readEventsCalls += 1;
apps/server/src/serverRuntimeStartup.reconcile.test.ts+4
...@@ -90,6 +90,8 @@ const runReconciliation = (input: {...@@ -90,6 +90,8 @@ const runReconciliation = (input: {
90 Effect.provideService(ProviderSessionDirectory.ProviderSessionDirectory, input.directory),90 Effect.provideService(ProviderSessionDirectory.ProviderSessionDirectory, input.directory),
91 Effect.provideService(OrchestrationEngine.OrchestrationEngineService, {91 Effect.provideService(OrchestrationEngine.OrchestrationEngineService, {
92 readEvents: () => Stream.empty,92 readEvents: () => Stream.empty,
93 readThreadEvents: () => Stream.empty,
94 getThreadReplayStats: () => Effect.die("unused thread replay stats"),
93 dispatch: input.dispatch,95 dispatch: input.dispatch,
94 streamDomainEvents: Stream.empty,96 streamDomainEvents: Stream.empty,
95 subscribeDomainEvents: Effect.succeed(Stream.empty),97 subscribeDomainEvents: Effect.succeed(Stream.empty),
...@@ -653,6 +655,8 @@ it.effect("does not fail startup when the live provider session inventory cannot...@@ -653,6 +655,8 @@ it.effect("does not fail startup when the live provider session inventory cannot
653 }),655 }),
654 Effect.provideService(OrchestrationEngine.OrchestrationEngineService, {656 Effect.provideService(OrchestrationEngine.OrchestrationEngineService, {
655 readEvents: () => Stream.empty,657 readEvents: () => Stream.empty,
658 readThreadEvents: () => Stream.empty,
659 getThreadReplayStats: () => Effect.die("unused thread replay stats"),
656 dispatch: () => Effect.die("unused"),660 dispatch: () => Effect.die("unused"),
657 streamDomainEvents: Stream.empty,661 streamDomainEvents: Stream.empty,
658 subscribeDomainEvents: Effect.succeed(Stream.empty),662 subscribeDomainEvents: Effect.succeed(Stream.empty),
apps/server/src/serverRuntimeStartup.test.ts+6
...@@ -220,6 +220,8 @@ it.effect("resolveAutoBootstrapWelcomeTargets returns existing project and threa...@@ -220,6 +220,8 @@ it.effect("resolveAutoBootstrapWelcomeTargets returns existing project and threa
220 }),220 }),
221 Effect.provideService(OrchestrationEngine.OrchestrationEngineService, {221 Effect.provideService(OrchestrationEngine.OrchestrationEngineService, {
222 readEvents: () => Stream.empty,222 readEvents: () => Stream.empty,
223 readThreadEvents: () => Stream.empty,
224 getThreadReplayStats: () => Effect.die("unused thread replay stats"),
223 dispatch: (command) =>225 dispatch: (command) =>
224 Ref.update(dispatchCalls, (calls) => [...calls, command.type]).pipe(226 Ref.update(dispatchCalls, (calls) => [...calls, command.type]).pipe(
225 Effect.as({ sequence: 1 }),227 Effect.as({ sequence: 1 }),
...@@ -275,6 +277,8 @@ it.effect("resolveAutoBootstrapWelcomeTargets creates a project and thread when...@@ -275,6 +277,8 @@ it.effect("resolveAutoBootstrapWelcomeTargets creates a project and thread when
275 }),277 }),
276 Effect.provideService(OrchestrationEngine.OrchestrationEngineService, {278 Effect.provideService(OrchestrationEngine.OrchestrationEngineService, {
277 readEvents: () => Stream.empty,279 readEvents: () => Stream.empty,
280 readThreadEvents: () => Stream.empty,
281 getThreadReplayStats: () => Effect.die("unused thread replay stats"),
278 dispatch: (command) =>282 dispatch: (command) =>
279 Ref.update(dispatchCalls, (calls) => [...calls, command]).pipe(283 Ref.update(dispatchCalls, (calls) => [...calls, command]).pipe(
280 Effect.as({ sequence: 1 }),284 Effect.as({ sequence: 1 }),
...@@ -339,6 +343,8 @@ it.effect("resolveAutoBootstrapWelcomeTargets preserves typed UUID generation fa...@@ -339,6 +343,8 @@ it.effect("resolveAutoBootstrapWelcomeTargets preserves typed UUID generation fa
339 }),343 }),
340 Effect.provideService(OrchestrationEngine.OrchestrationEngineService, {344 Effect.provideService(OrchestrationEngine.OrchestrationEngineService, {
341 readEvents: () => Stream.empty,345 readEvents: () => Stream.empty,
346 readThreadEvents: () => Stream.empty,
347 getThreadReplayStats: () => Effect.die("unused thread replay stats"),
342 dispatch: (command) =>348 dispatch: (command) =>
343 Ref.update(dispatchCalls, (calls) => [...calls, command.type]).pipe(349 Ref.update(dispatchCalls, (calls) => [...calls, command.type]).pipe(
344 Effect.as({ sequence: 1 }),350 Effect.as({ sequence: 1 }),
apps/server/src/ws.ts+46-22
...@@ -333,12 +333,9 @@ const PROVIDER_STATUS_DEBOUNCE_MS = 200;...@@ -333,12 +333,9 @@ const PROVIDER_STATUS_DEBOUNCE_MS = 200;
333// Matches the event store's default page size (DEFAULT_READ_FROM_SEQUENCE_LIMIT).333// Matches the event store's default page size (DEFAULT_READ_FROM_SEQUENCE_LIMIT).
334const SHELL_RESUME_MAX_GAP = 1_000;334const SHELL_RESUME_MAX_GAP = 1_000;
335335
336// Same bound for thread resume. The replay reads the *global* event range and336// Thread replay counts only this thread's rows. Busy or pruned unrelated
337// filters per-thread afterwards, so a stale cursor far behind the head would337// streams must not force a full thread snapshot.
338// otherwise decode every intervening event's payload — reconnects with cursors338const THREAD_RESUME_MAX_EVENTS = 1_000;
339// hundreds of thousands of events behind have OOM-killed servers on large
340// databases. Past this gap the client is reset with a fresh thread snapshot.
341const THREAD_RESUME_MAX_GAP = 1_000;
342// Row count alone does not bound replay memory: a few events with large tool339// Row count alone does not bound replay memory: a few events with large tool
343// payloads can decode to gigabytes. Before replaying, sum the serialized340// payloads can decode to gigabytes. Before replaying, sum the serialized
344// payload bytes of the range in SQL and reset with a snapshot past this budget.341// payload bytes of the range in SQL and reset with a snapshot past this budget.
...@@ -1594,6 +1591,7 @@ const makeWsRpcLayer = (...@@ -1594,6 +1591,7 @@ const makeWsRpcLayer = (
1594 { startImmediately: true },1591 { startImmediately: true },
1595 );1592 );
1596 const bufferedLiveStream = liveBuffer.stream;1593 const bufferedLiveStream = liveBuffer.stream;
1594 let replayOnMissingSnapshot: typeof bufferedLiveStream | undefined;
15971595
1598 // When the client already loaded the snapshot over HTTP it passes1596 // When the client already loaded the snapshot over HTTP it passes
1599 // that snapshot's sequence, and we resume the live subscription by1597 // that snapshot's sequence, and we resume the live subscription by
...@@ -1608,24 +1606,41 @@ const makeWsRpcLayer = (...@@ -1608,24 +1606,41 @@ const makeWsRpcLayer = (
1608 // catch-up followed by the buffered/ongoing live events. Overlapping1606 // catch-up followed by the buffered/ongoing live events. Overlapping
1609 // events are deduped by sequence on the client.1607 // events are deduped by sequence on the client.
1610 //1608 //
1611 // The replay is bounded to the projection head captured below. The1609 // Measure only this thread's rows. Global sequence gaps can
1612 // catch-up range is normally tiny (a fresh HTTP snapshot sequence),1610 // contain unrelated or pruned streams. Keep an explicit upper
1613 // but a stale cached cursor can sit hundreds of thousands of global1611 // bound so events after the captured head stay in the live tail.
1614 // events behind — replaying that decodes every intervening event
1615 // (including every other thread's tool payloads) only to discard
1616 // almost all of them, which has OOM-killed servers on large
1617 // databases. A truncated replay would silently drop this thread's
1618 // events, so past the gap cap we reset the client with a fresh
1619 // thread snapshot instead, exactly like subscribeShell above.
1620 if (input.afterSequence !== undefined) {1612 if (input.afterSequence !== undefined) {
1621 const afterSequence = input.afterSequence;1613 const afterSequence = input.afterSequence;
1622 const headSequence = yield* orchestrationEngine.latestSequence;1614 const headSequence = yield* orchestrationEngine.latestSequence;
1623 const replayGap = headSequence - afterSequence;1615 const range = {
1616 threadId: input.threadId,
1617 fromSequenceExclusive: afterSequence,
1618 toSequenceInclusive: headSequence,
1619 };
1620 const replayStats =
1621 afterSequence > headSequence
1622 ? null
1623 : yield* orchestrationEngine
1624 .getThreadReplayStats({
1625 ...range,
1626 maxEvents: THREAD_RESUME_MAX_EVENTS,
1627 })
1628 .pipe(
1629 Effect.mapError(
1630 (cause) =>
1631 new OrchestrationGetSnapshotError({
1632 message: `Failed to measure thread ${input.threadId} replay range`,
1633 cause,
1634 }),
1635 ),
1636 );
1624 if (1637 if (
1625 yield* canReplayPersistedRange(afterSequence, headSequence, THREAD_RESUME_MAX_GAP)1638 replayStats !== null &&
1639 replayStats.eventCount <= THREAD_RESUME_MAX_EVENTS &&
1640 replayStats.payloadBytes <= ORCHESTRATION_REPLAY_PAYLOAD_BUDGET_BYTES
1626 ) {1641 ) {
1627 const catchUpStream = orchestrationEngine1642 const catchUpStream = orchestrationEngine
1628 .readEvents(afterSequence, replayGap)1643 .readThreadEvents({ ...range, limit: THREAD_RESUME_MAX_EVENTS })
1629 .pipe(1644 .pipe(
1630 Stream.filter(isThisThreadDetailEvent),1645 Stream.filter(isThisThreadDetailEvent),
1631 Stream.map((event) => ({1646 Stream.map((event) => ({
...@@ -1648,11 +1663,14 @@ const makeWsRpcLayer = (...@@ -1648,11 +1663,14 @@ const makeWsRpcLayer = (
1648 .pipe(Effect.as(bufferedLiveStream)),1663 .pipe(Effect.as(bufferedLiveStream)),
1649 )1664 )
1650 : bufferedLiveStream;1665 : bufferedLiveStream;
1651 return Stream.concat(catchUpStream, afterCatchUp);1666 const replay = Stream.concat(catchUpStream, afterCatchUp);
1667 if (!replayStats.hasCreateEvent) {
1668 return replay;
1669 }
1670 replayOnMissingSnapshot = replay;
1652 }1671 }
1653 // Gap too large (or cursor ahead of authoritative state): fall1672 // A recreated thread needs a fresh snapshot if it still exists.
1654 // through to the snapshot path so the client converges from a1673 // Oversized replays and invalid cursors also use the snapshot path.
1655 // fresh thread detail instead of an unbounded replay.
1656 }1674 }
16571675
1658 const snapshot = yield* projectionSnapshotQuery1676 const snapshot = yield* projectionSnapshotQuery
...@@ -1675,6 +1693,12 @@ const makeWsRpcLayer = (...@@ -1675,6 +1693,12 @@ const makeWsRpcLayer = (
1675 );1693 );
16761694
1677 if (Option.isNone(snapshot)) {1695 if (Option.isNone(snapshot)) {
1696 // The recreated thread can already be deleted. Preserve the
1697 // bounded replay and shell removal instead of retrying a
1698 // snapshot that cannot exist. Oversized ranges still fail.
1699 if (replayOnMissingSnapshot !== undefined) {
1700 return replayOnMissingSnapshot;
1701 }
1678 return yield* new OrchestrationGetSnapshotError({1702 return yield* new OrchestrationGetSnapshotError({
1679 message: `Thread ${input.threadId} was not found`,1703 message: `Thread ${input.threadId} was not found`,
1680 cause: input.threadId,1704 cause: input.threadId,