authorgravatar for me@t3.ggTheo Browne <me@t3.gg> 2026-09-04 06:24:41-07:00
committergravatar for noreply@github.comGitHub <noreply@github.com> 2026-09-04 06:24:41-07:00
log8ac5462920c45cdee63af15b2598909736f2ec84
treeb845a2b99f3409aaa33e8b4c06f5df03b34eb8d7
parent560afffdea82000d757c98ea79678aee75f8648c
signature Signed by PGP key B5690EEEBB952194

perf(server): stop loading message bodies for thread summaries (#9662)

7 files changed, 205 insertions(+), 20 deletions(-)

apps/server/src/orchestration/Layers/ProjectionPipeline.test.ts+101-1
...@@ -2534,7 +2534,7 @@ it.layer(BaseTestLayer)("OrchestrationProjectionPipeline", (it) => {...@@ -2534,7 +2534,7 @@ it.layer(BaseTestLayer)("OrchestrationProjectionPipeline", (it) => {
2534 }),2534 }),
2535 );2535 );
25362536
2537 it.effect("maintains shell summary fields across message and activity streams", () =>2537 it.effect("maintains shell summaries without reading message bodies", () =>
2538 Effect.gen(function* () {2538 Effect.gen(function* () {
2539 const projectionPipeline = yield* OrchestrationProjectionPipeline;2539 const projectionPipeline = yield* OrchestrationProjectionPipeline;
2540 const eventStore = yield* OrchestrationEventStore;2540 const eventStore = yield* OrchestrationEventStore;
...@@ -2742,6 +2742,106 @@ it.layer(BaseTestLayer)("OrchestrationProjectionPipeline", (it) => {...@@ -2742,6 +2742,106 @@ it.layer(BaseTestLayer)("OrchestrationProjectionPipeline", (it) => {
2742 updatedAt: "2026-03-01T08:00:05.000Z",2742 updatedAt: "2026-03-01T08:00:05.000Z",
2743 },2743 },
2744 ]);2744 ]);
2745
2746 // Summary refreshes must not decode message bodies or attachment metadata.
2747 yield* sql`
2748 UPDATE projection_thread_messages
2749 SET attachments_json = '{not-json'
2750 WHERE thread_id = 'thread-shell-summary'
2751 `;
2752 yield* sql`
2753 INSERT INTO projection_pending_approvals (
2754 request_id, thread_id, turn_id, status, decision, created_at, resolved_at
2755 ) VALUES
2756 ('summary-pending', 'thread-shell-summary', NULL, 'pending', NULL,
2757 '2026-03-01T08:00:06.000Z', NULL),
2758 ('summary-resolved', 'thread-shell-summary', NULL, 'resolved', 'accept',
2759 '2026-03-01T08:00:06.000Z', '2026-03-01T08:00:06.000Z'),
2760 ('summary-other-thread', 'thread-shell-summary-other', NULL, 'pending', NULL,
2761 '2026-03-01T08:00:06.000Z', NULL)
2762 `;
2763 yield* sql`
2764 INSERT INTO projection_thread_proposed_plans (
2765 plan_id, thread_id, turn_id, plan_markdown, implemented_at,
2766 implementation_thread_id, created_at, updated_at
2767 ) VALUES (
2768 'summary-plan', 'thread-shell-summary', 'turn-shell-summary-1', '# Plan', NULL,
2769 NULL, '2026-03-01T08:00:06.000Z', '2026-03-01T08:00:06.000Z'
2770 )
2771 `;
2772
2773 const refreshEvents = [
2774 {
2775 type: "thread.session-set",
2776 eventId: EventId.make("evt-shell-summary-7"),
2777 aggregateKind: "thread",
2778 aggregateId: ThreadId.make("thread-shell-summary"),
2779 occurredAt: "2026-03-01T08:00:07.000Z",
2780 commandId: CommandId.make("cmd-shell-summary-7"),
2781 causationEventId: null,
2782 correlationId: CorrelationId.make("cmd-shell-summary-7"),
2783 metadata: {},
2784 payload: {
2785 threadId: ThreadId.make("thread-shell-summary"),
2786 session: {
2787 threadId: ThreadId.make("thread-shell-summary"),
2788 status: "ready",
2789 providerName: "codex",
2790 runtimeMode: "approval-required",
2791 activeTurnId: null,
2792 lastError: null,
2793 updatedAt: "2026-03-01T08:00:07.000Z",
2794 },
2795 },
2796 },
2797 {
2798 type: "thread.turn-diff-completed",
2799 eventId: EventId.make("evt-shell-summary-8"),
2800 aggregateKind: "thread",
2801 aggregateId: ThreadId.make("thread-shell-summary"),
2802 occurredAt: "2026-03-01T08:00:08.000Z",
2803 commandId: CommandId.make("cmd-shell-summary-8"),
2804 causationEventId: null,
2805 correlationId: CorrelationId.make("cmd-shell-summary-8"),
2806 metadata: {},
2807 payload: {
2808 threadId: ThreadId.make("thread-shell-summary"),
2809 turnId: TurnId.make("turn-shell-summary-1"),
2810 checkpointTurnCount: 1,
2811 checkpointRef: CheckpointRef.make("refs/t3/checkpoints/thread-shell-summary/1"),
2812 status: "ready",
2813 files: [],
2814 assistantMessageId: MessageId.make("message-shell-summary-assistant"),
2815 completedAt: "2026-03-01T08:00:08.000Z",
2816 },
2817 },
2818 ] satisfies ReadonlyArray<Parameters<typeof eventStore.append>[0]>;
2819
2820 for (const event of refreshEvents) {
2821 yield* appendAndProject(event);
2822 const summary = yield* sql<{
2823 readonly latestUserMessageAt: string | null;
2824 readonly pendingApprovalCount: number;
2825 readonly pendingUserInputCount: number;
2826 readonly hasActionableProposedPlan: number;
2827 }>`
2828 SELECT
2829 latest_user_message_at AS "latestUserMessageAt",
2830 pending_approval_count AS "pendingApprovalCount",
2831 pending_user_input_count AS "pendingUserInputCount",
2832 has_actionable_proposed_plan AS "hasActionableProposedPlan"
2833 FROM projection_threads
2834 WHERE thread_id = 'thread-shell-summary'
2835 `;
2836 assert.deepEqual(summary, [
2837 {
2838 latestUserMessageAt: "2026-03-01T08:00:02.000Z",
2839 pendingApprovalCount: 1,
2840 pendingUserInputCount: 1,
2841 hasActionableProposedPlan: 1,
2842 },
2843 ]);
2844 }
2745 }),2845 }),
2746 );2846 );
27472847
apps/server/src/orchestration/Layers/ProjectionPipeline.ts+7-19
...@@ -587,26 +587,14 @@ const makeOrchestrationProjectionPipeline = Effect.fn("makeOrchestrationProjecti...@@ -587,26 +587,14 @@ const makeOrchestrationProjectionPipeline = Effect.fn("makeOrchestrationProjecti
587 return;587 return;
588 }588 }
589589
590 const [messages, proposedPlans, activities, pendingApprovals] = yield* Effect.all([590 const [latestUserMessageAt, proposedPlans, activities, pendingApprovalCount] =
591 projectionThreadMessageRepository.listByThreadId({ threadId }),591 yield* Effect.all([
592 projectionThreadProposedPlanRepository.listByThreadId({ threadId }),592 projectionThreadMessageRepository.getLatestUserMessageAt({ threadId }),
593 projectionThreadActivityRepository.listUserInputLifecycleByThreadId({ threadId }),593 projectionThreadProposedPlanRepository.listByThreadId({ threadId }),
594 projectionPendingApprovalRepository.listByThreadId({ threadId }),594 projectionThreadActivityRepository.listUserInputLifecycleByThreadId({ threadId }),
595 ]);595 projectionPendingApprovalRepository.countPendingByThreadId({ threadId }),
596596 ]);
597 let latestUserMessageAt: string | null = null;
598 for (const message of messages) {
599 if (
600 message.role === "user" &&
601 (latestUserMessageAt === null || message.createdAt > latestUserMessageAt)
602 ) {
603 latestUserMessageAt = message.createdAt;
604 }
605 }
606597
607 const pendingApprovalCount = pendingApprovals.filter(
608 (approval) => approval.status === "pending",
609 ).length;
610 const pendingUserInputCount = derivePendingUserInputCountFromActivities(activities);598 const pendingUserInputCount = derivePendingUserInputCountFromActivities(activities);
611 const hasActionableProposedPlan = deriveHasActionableProposedPlan({599 const hasActionableProposedPlan = deriveHasActionableProposedPlan({
612 latestTurnId: existingRow.value.latestTurnId,600 latestTurnId: existingRow.value.latestTurnId,
apps/server/src/persistence/Layers/ProjectionPendingApprovals.ts+21
...@@ -2,6 +2,7 @@ import * as SqlClient from "effect/unstable/sql/SqlClient";...@@ -2,6 +2,7 @@ import * as SqlClient from "effect/unstable/sql/SqlClient";
2import * as SqlSchema from "effect/unstable/sql/SqlSchema";2import * as SqlSchema from "effect/unstable/sql/SqlSchema";
3import * as Effect from "effect/Effect";3import * as Effect from "effect/Effect";
4import * as Layer from "effect/Layer";4import * as Layer from "effect/Layer";
5import * as Schema from "effect/Schema";
56
6import { toPersistenceSqlError } from "../Errors.ts";7import { toPersistenceSqlError } from "../Errors.ts";
7import {8import {
...@@ -68,6 +69,16 @@ const makeProjectionPendingApprovalRepository = Effect.gen(function* () {...@@ -68,6 +69,16 @@ const makeProjectionPendingApprovalRepository = Effect.gen(function* () {
68 `,69 `,
69 });70 });
7071
72 const countPendingApprovalRows = SqlSchema.findOne({
73 Request: ListProjectionPendingApprovalsInput,
74 Result: Schema.Struct({ count: Schema.Number }),
75 execute: ({ threadId }) => sql`
76 SELECT COUNT(*) AS count
77 FROM projection_pending_approvals
78 WHERE thread_id = ${threadId} AND status = 'pending'
79 `,
80 });
81
71 const getProjectionPendingApprovalRow = SqlSchema.findOneOption({82 const getProjectionPendingApprovalRow = SqlSchema.findOneOption({
72 Request: GetProjectionPendingApprovalInput,83 Request: GetProjectionPendingApprovalInput,
73 Result: ProjectionPendingApproval,84 Result: ProjectionPendingApproval,
...@@ -116,6 +127,15 @@ const makeProjectionPendingApprovalRepository = Effect.gen(function* () {...@@ -116,6 +127,15 @@ const makeProjectionPendingApprovalRepository = Effect.gen(function* () {
116 ),127 ),
117 );128 );
118129
130 const countPendingByThreadId: ProjectionPendingApprovalRepositoryShape["countPendingByThreadId"] =
131 (input) =>
132 countPendingApprovalRows(input).pipe(
133 Effect.mapError(
134 toPersistenceSqlError("ProjectionPendingApprovalRepository.countPendingByThreadId:query"),
135 ),
136 Effect.map((row) => row.count),
137 );
138
119 const getByRequestId: ProjectionPendingApprovalRepositoryShape["getByRequestId"] = (input) =>139 const getByRequestId: ProjectionPendingApprovalRepositoryShape["getByRequestId"] = (input) =>
120 getProjectionPendingApprovalRow(input).pipe(140 getProjectionPendingApprovalRow(input).pipe(
121 Effect.mapError(141 Effect.mapError(
...@@ -142,6 +162,7 @@ const makeProjectionPendingApprovalRepository = Effect.gen(function* () {...@@ -142,6 +162,7 @@ const makeProjectionPendingApprovalRepository = Effect.gen(function* () {
142 return {162 return {
143 upsert,163 upsert,
144 listByThreadId,164 listByThreadId,
165 countPendingByThreadId,
145 getByRequestId,166 getByRequestId,
146 deleteByRequestId,167 deleteByRequestId,
147 deleteByThreadId,168 deleteByThreadId,
apps/server/src/persistence/Layers/ProjectionThreadMessages.test.ts+43
...@@ -12,6 +12,49 @@ const layer = it.layer(...@@ -12,6 +12,49 @@ const layer = it.layer(
12);12);
1313
14layer("ProjectionThreadMessageRepository", (it) => {14layer("ProjectionThreadMessageRepository", (it) => {
15 it.effect("finds the latest user-message time within one thread", () =>
16 Effect.gen(function* () {
17 const repository = yield* ProjectionThreadMessageRepository;
18 const threadId = ThreadId.make("thread-latest-user-message");
19 assert.isNull(yield* repository.getLatestUserMessageAt({ threadId }));
20
21 const messages = [
22 { role: "user", createdAt: "2026-02-28T19:05:02.000Z" },
23 { role: "user", createdAt: "2026-02-28T19:05:01.000Z" },
24 { role: "assistant", createdAt: "2026-02-28T19:05:03.000Z" },
25 { role: "system", createdAt: "2026-02-28T19:05:04.000Z" },
26 ] as const;
27 for (const [index, message] of messages.entries()) {
28 yield* repository.upsert({
29 messageId: MessageId.make(`latest-user-message-${index}`),
30 threadId,
31 turnId: null,
32 ...message,
33 text: "Message body",
34 isStreaming: false,
35 updatedAt: "2026-02-28T19:06:00.000Z",
36 });
37 }
38 yield* repository.upsert({
39 messageId: MessageId.make("latest-user-message-other-thread"),
40 threadId: ThreadId.make("thread-latest-user-message-other"),
41 turnId: null,
42 role: "user",
43 text: "Other thread",
44 isStreaming: false,
45 createdAt: "2026-02-28T19:05:05.000Z",
46 updatedAt: "2026-02-28T19:05:05.000Z",
47 });
48
49 assert.strictEqual(
50 yield* repository.getLatestUserMessageAt({ threadId }),
51 "2026-02-28T19:05:02.000Z",
52 );
53 yield* repository.deleteByThreadId({ threadId });
54 assert.isNull(yield* repository.getLatestUserMessageAt({ threadId }));
55 }),
56 );
57
15 it.effect("appends streaming text and applies attachment updates", () =>58 it.effect("appends streaming text and applies attachment updates", () =>
16 Effect.gen(function* () {59 Effect.gen(function* () {
17 const repository = yield* ProjectionThreadMessageRepository;60 const repository = yield* ProjectionThreadMessageRepository;
apps/server/src/persistence/Layers/ProjectionThreadMessages.ts+23
...@@ -182,6 +182,18 @@ const makeProjectionThreadMessageRepository = Effect.gen(function* () {...@@ -182,6 +182,18 @@ const makeProjectionThreadMessageRepository = Effect.gen(function* () {
182 `,182 `,
183 });183 });
184184
185 const getLatestUserMessageAtRow = SqlSchema.findOne({
186 Request: ListProjectionThreadMessagesInput,
187 Result: Schema.Struct({
188 latestUserMessageAt: Schema.NullOr(ProjectionThreadMessage.fields.createdAt),
189 }),
190 execute: ({ threadId }) => sql`
191 SELECT MAX(created_at) AS "latestUserMessageAt"
192 FROM projection_thread_messages
193 WHERE thread_id = ${threadId} AND role = 'user'
194 `,
195 });
196
185 const deleteProjectionThreadMessageRows = SqlSchema.void({197 const deleteProjectionThreadMessageRows = SqlSchema.void({
186 Request: DeleteProjectionThreadMessagesInput,198 Request: DeleteProjectionThreadMessagesInput,
187 execute: ({ threadId }) =>199 execute: ({ threadId }) =>
...@@ -219,6 +231,16 @@ const makeProjectionThreadMessageRepository = Effect.gen(function* () {...@@ -219,6 +231,16 @@ const makeProjectionThreadMessageRepository = Effect.gen(function* () {
219 Effect.map((rows) => rows.map(toProjectionThreadMessage)),231 Effect.map((rows) => rows.map(toProjectionThreadMessage)),
220 );232 );
221233
234 const getLatestUserMessageAt: ProjectionThreadMessageRepositoryShape["getLatestUserMessageAt"] = (
235 input,
236 ) =>
237 getLatestUserMessageAtRow(input).pipe(
238 Effect.mapError(
239 toPersistenceSqlError("ProjectionThreadMessageRepository.getLatestUserMessageAt:query"),
240 ),
241 Effect.map((row) => row.latestUserMessageAt),
242 );
243
222 const deleteByThreadId: ProjectionThreadMessageRepositoryShape["deleteByThreadId"] = (input) =>244 const deleteByThreadId: ProjectionThreadMessageRepositoryShape["deleteByThreadId"] = (input) =>
223 deleteProjectionThreadMessageRows(input).pipe(245 deleteProjectionThreadMessageRows(input).pipe(
224 Effect.mapError(246 Effect.mapError(
...@@ -231,6 +253,7 @@ const makeProjectionThreadMessageRepository = Effect.gen(function* () {...@@ -231,6 +253,7 @@ const makeProjectionThreadMessageRepository = Effect.gen(function* () {
231 appendStreaming,253 appendStreaming,
232 getByMessageId,254 getByMessageId,
233 listByThreadId,255 listByThreadId,
256 getLatestUserMessageAt,
234 deleteByThreadId,257 deleteByThreadId,
235 } satisfies ProjectionThreadMessageRepositoryShape;258 } satisfies ProjectionThreadMessageRepositoryShape;
236});259});
apps/server/src/persistence/Services/ProjectionPendingApprovals.ts+5
...@@ -69,6 +69,11 @@ export interface ProjectionPendingApprovalRepositoryShape {...@@ -69,6 +69,11 @@ export interface ProjectionPendingApprovalRepositoryShape {
69 input: ListProjectionPendingApprovalsInput,69 input: ListProjectionPendingApprovalsInput,
70 ) => Effect.Effect<ReadonlyArray<ProjectionPendingApproval>, ProjectionRepositoryError>;70 ) => Effect.Effect<ReadonlyArray<ProjectionPendingApproval>, ProjectionRepositoryError>;
7171
72 /** Count pending approvals without loading resolved request history. */
73 readonly countPendingByThreadId: (
74 input: ListProjectionPendingApprovalsInput,
75 ) => Effect.Effect<number, ProjectionRepositoryError>;
76
72 /**77 /**
73 * Read a pending approval row by request id.78 * Read a pending approval row by request id.
74 */79 */
apps/server/src/persistence/Services/ProjectionThreadMessages.ts+5
...@@ -90,6 +90,11 @@ export interface ProjectionThreadMessageRepositoryShape {...@@ -90,6 +90,11 @@ export interface ProjectionThreadMessageRepositoryShape {
90 input: ListProjectionThreadMessagesInput,90 input: ListProjectionThreadMessagesInput,
91 ) => Effect.Effect<ReadonlyArray<ProjectionThreadMessage>, ProjectionRepositoryError>;91 ) => Effect.Effect<ReadonlyArray<ProjectionThreadMessage>, ProjectionRepositoryError>;
9292
93 /** Read the latest user-message timestamp without loading message bodies. */
94 readonly getLatestUserMessageAt: (
95 input: ListProjectionThreadMessagesInput,
96 ) => Effect.Effect<ProjectionThreadMessage["createdAt"] | null, ProjectionRepositoryError>;
97
93 /**98 /**
94 * Delete projected thread messages by thread.99 * Delete projected thread messages by thread.
95 */100 */