From e17e31f177e3e176a223449a8d4556a3c9936bf8 Mon Sep 17 00:00:00 2001 From: Yuruzuu Date: Mon, 2 Nov 2026 14:04:39 +0800 Subject: [PATCH 1/2] perf(server): shell updates read a burst of threads once Each coalesced shell event group read every thread with its own getThreadShell transaction. A group touching N threads repeated about four SELECTs per thread, per subscriber, and fork siblings reread shared ancestors. Groups with more than one distinct thread now make one targeted getShellSnapshot read. Groups with one thread keep the scalar read, which is one SELECT cheaper. Every item still carries its own stored event sequence, never the snapshot's global sequence. Co-Authored-By: Claude Opus 5.5 --- .../orchestration-v2/ProjectionStore.test.ts | 11 ++ apps/server/src/ws.test.ts | 121 ++++++++++++++++++ apps/server/src/ws.ts | 27 +++- 3 files changed, 157 insertions(+), 2 deletions(-) diff --git a/apps/server/src/orchestration-v2/ProjectionStore.test.ts b/apps/server/src/orchestration-v2/ProjectionStore.test.ts index aadd9ecc8be06..063b72a5c3524 100644 --- a/apps/server/src/orchestration-v2/ProjectionStore.test.ts +++ b/apps/server/src/orchestration-v2/ProjectionStore.test.ts @@ -1374,6 +1374,17 @@ const assertProjectScopedShells = Effect.fn("assertProjectScopedShells")(functio all.archivedThreads.map((thread) => thread.id), [archivedThreadId], ); + const missingThreadId = ThreadId.make(`thread:${suffix}:missing`); + const targeted = yield* projectionStore.getShellSnapshot({ + threadIds: [forkThreadId, archivedThreadId, missingThreadId], + }); + for (const threadId of [forkThreadId, archivedThreadId, missingThreadId]) { + assert.deepEqual( + [...targeted.threads, ...targeted.archivedThreads].find((shell) => shell.id === threadId) ?? + null, + yield* projectionStore.getThreadShell(threadId), + ); + } return { projectId, activeThreadId, forkThreadId, otherThreadId }; }); diff --git a/apps/server/src/ws.test.ts b/apps/server/src/ws.test.ts index 97ed92fe21bbc..fe67290661798 100644 --- a/apps/server/src/ws.test.ts +++ b/apps/server/src/ws.test.ts @@ -421,6 +421,127 @@ it.effect("coalesces streaming replay before synchronization and the live tail", }).pipe(Effect.scoped, Effect.provide(ReplayTestLayer)), ); +it.effect("shell groups read distinct threads once and keep each event's own sequence", () => + Effect.gen(function* () { + const sql = yield* SqlClient.SqlClient; + const now = "2026-10-27T00:00:00.000Z"; + const insert = (aggregateKind: "thread" | "project", streamId: string, version: number) => + sql<{ sequence: number }>` + INSERT INTO orchestration_events ( + event_id, aggregate_kind, stream_id, stream_version, event_type, + occurred_at, actor_kind, payload_json, metadata_json, application_event_version + ) VALUES ( + ${`shell-group:${streamId}:${version}`}, ${aggregateKind}, ${streamId}, ${version}, + ${aggregateKind === "thread" ? "provider-session.detached" : "project.meta-updated"}, + ${now}, 'server', ${JSON.stringify( + aggregateKind === "thread" + ? { providerSessionId: "session:shell-group", detachedAt: now } + : { projectId: streamId, updatedAt: now }, + )}, '{}', 2 + ) RETURNING sequence + `.pipe(Effect.map(([row]) => row!.sequence)); + const shell = (id: string, archivedAt: string | null = null) => + ({ id: ThreadId.make(id), archivedAt }) as unknown as OrchestrationV2ThreadShell; + const project = { + id: ProjectId.make("project:group"), + } as unknown as OrchestrationProjectShell; + + const after = yield* insert("project", "project:before", 0); + yield* insert("thread", "thread:active", 0); + const archived = yield* insert("thread", "thread:archived", 0); + const projectUpdated = yield* insert("project", project.id, 0); + const active = yield* insert("thread", "thread:active", 1); + const missing = yield* insert("thread", "thread:missing", 0); + + const shellReads: Array | undefined> = []; + let batchFails = false; + const collect = (afterSequence: number) => + subscribeOrchestrationV2Shell({ afterSequence, requestCompletionMarker: true }).pipe( + Effect.flatMap((stream) => + stream.pipe( + Stream.takeUntil((item) => item.kind === "synchronized"), + Stream.runCollect, + ), + ), + Effect.provideService(ThreadManagement.ThreadManagementService, { + listActiveSubagentThreadIds: () => Effect.succeed([]), + getShellSnapshot: (options?: { readonly threadIds?: ReadonlyArray }) => + Effect.suspend(() => { + shellReads.push(options?.threadIds); + return batchFails + ? Effect.fail("batch failed") + : Effect.succeed({ + schemaVersion: 1, + snapshotSequence: Number.MAX_SAFE_INTEGER, + threads: [shell("thread:active")], + archivedThreads: [shell("thread:archived", now)], + }); + }), + getThreadShell: (threadId: string) => + Effect.sync(() => { + shellReads.push([`scalar:${threadId}`]); + return shell(threadId); + }), + } as unknown as ThreadManagement.ThreadManagementService["Service"]), + Effect.provideService(ProjectStore.ProjectStoreV2, { + listShells: () => Effect.succeed([]), + } as unknown as ProjectStore.ProjectStoreV2["Service"]), + Effect.provideService(ProjectService.ProjectService, { + getShell: () => Effect.succeedSome(project), + } as unknown as ProjectService.ProjectService["Service"]), + Effect.provideServiceEffect( + ProjectEnrichment.ProjectEnrichmentService, + PubSub.unbounded().pipe( + Effect.map( + (changes) => + ({ + subscribeChanges: PubSub.subscribe(changes), + }) as unknown as ProjectEnrichment.ProjectEnrichmentService["Service"], + ), + ), + ), + ); + + const grouped = yield* collect(after); + assert.deepEqual(shellReads, [["thread:archived", "thread:active", "thread:missing"]]); + assert.deepEqual( + grouped.flatMap((item): ReadonlyArray> => + item.kind === "thread.updated" + ? [[item.kind, item.sequence, item.thread.id]] + : item.kind === "thread.removed" + ? [[item.kind, item.sequence, item.threadId]] + : item.kind === "project.updated" + ? [[item.kind, item.sequence, item.project.id]] + : [], + ), + [ + ["thread.removed", archived, "thread:archived"], + ["project.updated", projectUpdated, project.id], + ["thread.updated", active, "thread:active"], + ["thread.removed", missing, "thread:missing"], + ], + ); + + shellReads.length = 0; + const singleton = yield* collect(active); + assert.deepEqual(shellReads, [["scalar:thread:missing"]]); + assert.deepEqual( + singleton.flatMap((item) => (item.kind === "thread.updated" ? [item.sequence] : [])), + [missing], + ); + + shellReads.length = 0; + batchFails = true; + yield* collect(after); + assert.deepEqual(shellReads, [ + ["thread:archived", "thread:active", "thread:missing"], + ["scalar:thread:archived"], + ["scalar:thread:active"], + ["scalar:thread:missing"], + ]); + }).pipe(Effect.scoped, Effect.provide(ReplayTestLayer)), +); + it.effect("shell snapshots omit idle children and older settled threads only on request", () => Effect.gen(function* () { const shell = ( diff --git a/apps/server/src/ws.ts b/apps/server/src/ws.ts index cc06273187a9a..914c4cc23ab22 100644 --- a/apps/server/src/ws.ts +++ b/apps/server/src/ws.ts @@ -1104,14 +1104,37 @@ export const subscribeOrchestrationV2Shell = Effect.fn("ws.orchestrationV2.subsc const projectShellItems = Effect.fn("ws.orchestrationV2.projectShellItems")(function* ( events: ReadonlyArray, ) { + const coalesced = coalesceShellApplicationEvents(events); + const threadIds = coalesced.flatMap((stored) => + "aggregateKind" in stored ? [] : [stored.event.threadId], + ); + // A failed batch can't name its thread, so the scalar reads below report which one is bad. + const batch = + threadIds.length > 1 + ? yield* threadManagement + .getShellSnapshot({ threadIds }) + .pipe(Effect.orElseSucceed(() => null)) + : null; + const batchedShells = + batch === null + ? null + : new Map( + [...batch.threads, ...batch.archivedThreads].map( + (shell) => [shell.id, shell] as const, + ), + ); return yield* Effect.forEach( - coalesceShellApplicationEvents(events), + coalesced, (stored) => Effect.gen(function* () { if ("aggregateKind" in stored) { return yield* projectItem(stored); } - const shell = yield* threadManagement.getThreadShell(stored.event.threadId); + const threadId = stored.event.threadId; + const shell = + batchedShells === null + ? yield* threadManagement.getThreadShell(threadId) + : (batchedShells.get(threadId) ?? null); return shellStreamItemFromThreadShell({ stored, shell }); }), { concurrency: 8 }, From b4bf89d7883755dd0256808ede98d0f8e4a6b304 Mon Sep 17 00:00:00 2001 From: Yuruzuu Date: Mon, 2 Nov 2026 14:59:24 +0800 Subject: [PATCH 2/2] test(server): shell group test models missing threads as removals Co-Authored-By: Claude Opus 5.5 --- apps/server/src/ws.test.ts | 54 ++++++++++++++++++++------------------ 1 file changed, 29 insertions(+), 25 deletions(-) diff --git a/apps/server/src/ws.test.ts b/apps/server/src/ws.test.ts index fe67290661798..32325e4ee4999 100644 --- a/apps/server/src/ws.test.ts +++ b/apps/server/src/ws.test.ts @@ -480,7 +480,9 @@ it.effect("shell groups read distinct threads once and keep each event's own seq getThreadShell: (threadId: string) => Effect.sync(() => { shellReads.push([`scalar:${threadId}`]); - return shell(threadId); + return threadId === "thread:missing" + ? null + : shell(threadId, threadId === "thread:archived" ? now : null); }), } as unknown as ThreadManagement.ThreadManagementService["Service"]), Effect.provideService(ProjectStore.ProjectStoreV2, { @@ -502,37 +504,39 @@ it.effect("shell groups read distinct threads once and keep each event's own seq ), ); - const grouped = yield* collect(after); + const collectSummary = (afterSequence: number) => + collect(afterSequence).pipe( + Effect.map((items) => + items.flatMap((item): ReadonlyArray> => + item.kind === "thread.updated" + ? [[item.kind, item.sequence, item.thread.id]] + : item.kind === "thread.removed" + ? [[item.kind, item.sequence, item.threadId]] + : item.kind === "project.updated" + ? [[item.kind, item.sequence, item.project.id]] + : [], + ), + ), + ); + const expectedGroup = [ + ["thread.removed", archived, "thread:archived"], + ["project.updated", projectUpdated, project.id], + ["thread.updated", active, "thread:active"], + ["thread.removed", missing, "thread:missing"], + ]; + + assert.deepEqual(yield* collectSummary(after), expectedGroup); assert.deepEqual(shellReads, [["thread:archived", "thread:active", "thread:missing"]]); - assert.deepEqual( - grouped.flatMap((item): ReadonlyArray> => - item.kind === "thread.updated" - ? [[item.kind, item.sequence, item.thread.id]] - : item.kind === "thread.removed" - ? [[item.kind, item.sequence, item.threadId]] - : item.kind === "project.updated" - ? [[item.kind, item.sequence, item.project.id]] - : [], - ), - [ - ["thread.removed", archived, "thread:archived"], - ["project.updated", projectUpdated, project.id], - ["thread.updated", active, "thread:active"], - ["thread.removed", missing, "thread:missing"], - ], - ); shellReads.length = 0; - const singleton = yield* collect(active); + assert.deepEqual(yield* collectSummary(active), [ + ["thread.removed", missing, "thread:missing"], + ]); assert.deepEqual(shellReads, [["scalar:thread:missing"]]); - assert.deepEqual( - singleton.flatMap((item) => (item.kind === "thread.updated" ? [item.sequence] : [])), - [missing], - ); shellReads.length = 0; batchFails = true; - yield* collect(after); + assert.deepEqual(yield* collectSummary(after), expectedGroup); assert.deepEqual(shellReads, [ ["thread:archived", "thread:active", "thread:missing"], ["scalar:thread:archived"],