authorgravatar for me@t3.ggTheo Browne <me@t3.gg> 2026-09-06 02:21:03-07:00
committergravatar for noreply@github.comGitHub <noreply@github.com> 2026-09-06 02:21:03-07:00
logeb8ed80300bd719b49fc80ad8708f9839bb48c92
tree8c7f4b67996ad490851b1cb981e0614e2596808c
parent076d753ae6cb97136b6b4199fb8fc0b379282dba
signature Signed by PGP key B5690EEEBB952194

perf(server): skip plan bodies in thread summaries (#10341)

5 files changed, 280 insertions(+), 36 deletions(-)

apps/server/src/orchestration/Layers/ProjectionPipeline.test.ts+3-2
...@@ -2697,7 +2697,7 @@ it.layer(BaseTestLayer)("OrchestrationProjectionPipeline", (it) => {...@@ -2697,7 +2697,7 @@ it.layer(BaseTestLayer)("OrchestrationProjectionPipeline", (it) => {
2697 }),2697 }),
2698 );2698 );
26992699
2700 it.effect("maintains shell summaries without reading message bodies", () =>2700 it.effect("maintains shell summaries without decoding message or plan bodies", () =>
2701 Effect.gen(function* () {2701 Effect.gen(function* () {
2702 const projectionPipeline = yield* OrchestrationProjectionPipeline;2702 const projectionPipeline = yield* OrchestrationProjectionPipeline;
2703 const eventStore = yield* OrchestrationEventStore;2703 const eventStore = yield* OrchestrationEventStore;
...@@ -2923,12 +2923,13 @@ it.layer(BaseTestLayer)("OrchestrationProjectionPipeline", (it) => {...@@ -2923,12 +2923,13 @@ it.layer(BaseTestLayer)("OrchestrationProjectionPipeline", (it) => {
2923 ('summary-other-thread', 'thread-shell-summary-other', NULL, 'pending', NULL,2923 ('summary-other-thread', 'thread-shell-summary-other', NULL, 'pending', NULL,
2924 '2026-03-01T08:00:06.000Z', NULL)2924 '2026-03-01T08:00:06.000Z', NULL)
2925 `;2925 `;
2926 // Empty markdown must not be decoded when the shell only needs plan status.
2926 yield* sql`2927 yield* sql`
2927 INSERT INTO projection_thread_proposed_plans (2928 INSERT INTO projection_thread_proposed_plans (
2928 plan_id, thread_id, turn_id, plan_markdown, implemented_at,2929 plan_id, thread_id, turn_id, plan_markdown, implemented_at,
2929 implementation_thread_id, created_at, updated_at2930 implementation_thread_id, created_at, updated_at
2930 ) VALUES (2931 ) VALUES (
2931 'summary-plan', 'thread-shell-summary', 'turn-shell-summary-1', '# Plan', NULL,2932 'summary-plan', 'thread-shell-summary', 'turn-shell-summary-1', '', NULL,
2932 NULL, '2026-03-01T08:00:06.000Z', '2026-03-01T08:00:06.000Z'2933 NULL, '2026-03-01T08:00:06.000Z', '2026-03-01T08:00:06.000Z'
2933 )2934 )
2934 `;2935 `;
apps/server/src/orchestration/Layers/ProjectionPipeline.ts+5-33
...@@ -197,33 +197,6 @@ function derivePendingUserInputCountFromActivities(...@@ -197,33 +197,6 @@ function derivePendingUserInputCountFromActivities(
197 return openRequestIds.size;197 return openRequestIds.size;
198}198}
199199
200function deriveHasActionableProposedPlan(input: {
201 readonly latestTurnId: string | null;
202 readonly proposedPlans: ReadonlyArray<ProjectionThreadProposedPlan>;
203}): boolean {
204 const sorted = [...input.proposedPlans].toSorted(
205 (left, right) =>
206 left.updatedAt.localeCompare(right.updatedAt) || left.planId.localeCompare(right.planId),
207 );
208
209 let latestForTurn: ProjectionThreadProposedPlan | null = null;
210 if (input.latestTurnId !== null) {
211 for (let index = sorted.length - 1; index >= 0; index -= 1) {
212 const plan = sorted[index];
213 if (plan?.turnId === input.latestTurnId) {
214 latestForTurn = plan;
215 break;
216 }
217 }
218 }
219 if (latestForTurn !== null) {
220 return latestForTurn.implementedAt === null;
221 }
222
223 const latestPlan = sorted.at(-1) ?? null;
224 return latestPlan !== null && latestPlan.implementedAt === null;
225}
226
227function retainProjectionMessagesAfterRevert(200function retainProjectionMessagesAfterRevert(
228 messages: ReadonlyArray<ProjectionThreadMessage>,201 messages: ReadonlyArray<ProjectionThreadMessage>,
229 turns: ReadonlyArray<ProjectionTurn>,202 turns: ReadonlyArray<ProjectionTurn>,
...@@ -595,19 +568,18 @@ const makeOrchestrationProjectionPipeline = Effect.fn("makeOrchestrationProjecti...@@ -595,19 +568,18 @@ const makeOrchestrationProjectionPipeline = Effect.fn("makeOrchestrationProjecti
595 return;568 return;
596 }569 }
597570
598 const [latestUserMessageAt, proposedPlans, activities, pendingApprovalCount] =571 const [latestUserMessageAt, hasActionableProposedPlan, activities, pendingApprovalCount] =
599 yield* Effect.all([572 yield* Effect.all([
600 projectionThreadMessageRepository.getLatestUserMessageAt({ threadId }),573 projectionThreadMessageRepository.getLatestUserMessageAt({ threadId }),
601 projectionThreadProposedPlanRepository.listByThreadId({ threadId }),574 projectionThreadProposedPlanRepository.hasActionableByThreadId({
575 threadId,
576 latestTurnId: existingRow.value.latestTurnId,
577 }),
602 projectionThreadActivityRepository.listUserInputLifecycleByThreadId({ threadId }),578 projectionThreadActivityRepository.listUserInputLifecycleByThreadId({ threadId }),
603 projectionPendingApprovalRepository.countPendingByThreadId({ threadId }),579 projectionPendingApprovalRepository.countPendingByThreadId({ threadId }),
604 ]);580 ]);
605581
606 const pendingUserInputCount = derivePendingUserInputCountFromActivities(activities);582 const pendingUserInputCount = derivePendingUserInputCountFromActivities(activities);
607 const hasActionableProposedPlan = deriveHasActionableProposedPlan({
608 latestTurnId: existingRow.value.latestTurnId,
609 proposedPlans,
610 });
611583
612 yield* projectionThreadRepository.upsert({584 yield* projectionThreadRepository.upsert({
613 ...existingRow.value,585 ...existingRow.value,
apps/server/src/persistence/Layers/ProjectionRepositories.test.ts+210-1
...@@ -1,25 +1,234 @@...@@ -1,25 +1,234 @@
1import { ProjectId, ThreadId, ProviderInstanceId } from "@t3tools/contracts";1import { ProjectId, ThreadId, TurnId, ProviderInstanceId } from "@t3tools/contracts";
2import { assert, it } from "@effect/vitest";2import { assert, it } from "@effect/vitest";
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 Option from "effect/Option";5import * as Option from "effect/Option";
6import * as SqlClient from "effect/unstable/sql/SqlClient";6import * as SqlClient from "effect/unstable/sql/SqlClient";
7import * as Statement from "effect/unstable/sql/Statement";
78
8import { SqlitePersistenceMemory } from "./Sqlite.ts";9import { SqlitePersistenceMemory } from "./Sqlite.ts";
9import { ProjectionProjectRepositoryLive } from "./ProjectionProjects.ts";10import { ProjectionProjectRepositoryLive } from "./ProjectionProjects.ts";
10import { ProjectionThreadRepositoryLive } from "./ProjectionThreads.ts";11import { ProjectionThreadRepositoryLive } from "./ProjectionThreads.ts";
12import { ProjectionThreadProposedPlanRepositoryLive } from "./ProjectionThreadProposedPlans.ts";
11import { ProjectionProjectRepository } from "../Services/ProjectionProjects.ts";13import { ProjectionProjectRepository } from "../Services/ProjectionProjects.ts";
12import { ProjectionThreadRepository } from "../Services/ProjectionThreads.ts";14import { ProjectionThreadRepository } from "../Services/ProjectionThreads.ts";
15import { ProjectionThreadProposedPlanRepository } from "../Services/ProjectionThreadProposedPlans.ts";
1316
14const projectionRepositoriesLayer = it.layer(17const projectionRepositoriesLayer = it.layer(
15 Layer.mergeAll(18 Layer.mergeAll(
16 ProjectionProjectRepositoryLive.pipe(Layer.provideMerge(SqlitePersistenceMemory)),19 ProjectionProjectRepositoryLive.pipe(Layer.provideMerge(SqlitePersistenceMemory)),
17 ProjectionThreadRepositoryLive.pipe(Layer.provideMerge(SqlitePersistenceMemory)),20 ProjectionThreadRepositoryLive.pipe(Layer.provideMerge(SqlitePersistenceMemory)),
21 ProjectionThreadProposedPlanRepositoryLive.pipe(Layer.provideMerge(SqlitePersistenceMemory)),
18 SqlitePersistenceMemory,22 SqlitePersistenceMemory,
19 ),23 ),
20);24);
2125
22projectionRepositoriesLayer("Projection repositories", (it) => {26projectionRepositoriesLayer("Projection repositories", (it) => {
27 it.effect("selects the latest-turn plan before checking implementation status", () =>
28 Effect.gen(function* () {
29 const plans = yield* ProjectionThreadProposedPlanRepository;
30 const threadId = ThreadId.make("thread-plan-status");
31 const latestTurnId = TurnId.make("turn-plan-status-current");
32 const firstPlan = {
33 planId: "plan-status-first",
34 threadId,
35 turnId: latestTurnId,
36 planMarkdown: "# First plan",
37 implementedAt: null,
38 implementationThreadId: null,
39 createdAt: "2026-03-24T00:00:01.000Z",
40 updatedAt: "2026-03-24T00:00:01.000Z",
41 };
42 yield* plans.upsert(firstPlan);
43 yield* plans.upsert({
44 ...firstPlan,
45 planId: "plan-status-implemented",
46 implementedAt: "2026-03-24T00:00:02.000Z",
47 createdAt: "2026-03-24T00:00:02.000Z",
48 updatedAt: "2026-03-24T00:00:02.000Z",
49 });
50 yield* plans.upsert({
51 ...firstPlan,
52 planId: "plan-status-other-turn",
53 turnId: TurnId.make("turn-plan-status-old"),
54 updatedAt: "2026-03-24T00:00:10.000Z",
55 });
56
57 assert.isFalse(yield* plans.hasActionableByThreadId({ threadId, latestTurnId }));
58 assert.isTrue(yield* plans.hasActionableByThreadId({ threadId, latestTurnId: null }));
59
60 yield* plans.upsert({ ...firstPlan, updatedAt: "2026-03-24T00:00:03.000Z" });
61 assert.isTrue(yield* plans.hasActionableByThreadId({ threadId, latestTurnId }));
62 }),
63 );
64
65 it.effect("falls back within the thread when the latest turn has no plan", () =>
66 Effect.gen(function* () {
67 const plans = yield* ProjectionThreadProposedPlanRepository;
68 const threadId = ThreadId.make("thread-plan-fallback");
69 const latestTurnId = TurnId.make("turn-plan-fallback-missing");
70 assert.isFalse(yield* plans.hasActionableByThreadId({ threadId, latestTurnId }));
71 assert.isFalse(yield* plans.hasActionableByThreadId({ threadId, latestTurnId: null }));
72
73 const firstPlan = {
74 planId: "plan-fallback-without-turn",
75 threadId,
76 turnId: null,
77 planMarkdown: "# Old plan",
78 implementedAt: "2026-03-24T00:00:01.000Z",
79 implementationThreadId: null,
80 createdAt: "2026-03-24T00:00:01.000Z",
81 updatedAt: "2026-03-24T00:00:01.000Z",
82 };
83 yield* plans.upsert(firstPlan);
84 yield* plans.upsert({
85 ...firstPlan,
86 planId: "plan-fallback-with-turn",
87 turnId: TurnId.make("turn-plan-fallback-old"),
88 implementedAt: null,
89 updatedAt: "2026-03-24T00:00:02.000Z",
90 });
91 yield* plans.upsert({
92 ...firstPlan,
93 planId: "plan-fallback-other-thread",
94 threadId: ThreadId.make("thread-plan-fallback-other"),
95 turnId: latestTurnId,
96 updatedAt: "2026-03-24T00:00:03.000Z",
97 });
98
99 assert.isTrue(yield* plans.hasActionableByThreadId({ threadId, latestTurnId }));
100 assert.isTrue(yield* plans.hasActionableByThreadId({ threadId, latestTurnId: null }));
101 }),
102 );
103
104 it.effect("preserves locale ordering and stable ties when selecting plan status", () =>
105 Effect.gen(function* () {
106 const plans = yield* ProjectionThreadProposedPlanRepository;
107 const timestamp = "2026-03-24T00:00:00.000Z";
108 const cases = [
109 {
110 name: "mixed-case-ids",
111 expected: "plan-A".localeCompare("plan-a") > 0,
112 rows: [
113 {
114 planId: "plan-a",
115 implementedAt: timestamp,
116 createdAt: timestamp,
117 updatedAt: timestamp,
118 },
119 { planId: "plan-A", implementedAt: null, createdAt: timestamp, updatedAt: timestamp },
120 ],
121 },
122 {
123 name: "equivalent-ids",
124 expected: true,
125 rows: [
126 {
127 planId: "plan-\u00e9",
128 implementedAt: timestamp,
129 createdAt: timestamp,
130 updatedAt: timestamp,
131 },
132 {
133 planId: "plan-e\u0301",
134 implementedAt: null,
135 createdAt: "2026-03-24T00:00:01.000Z",
136 updatedAt: timestamp,
137 },
138 ],
139 },
140 {
141 name: "timestamp-formats",
142 expected: "2026-03-24T00:00:00+00:00".localeCompare("2026-03-24T00:00:00-01:00") > 0,
143 rows: [
144 {
145 planId: "plan-minus",
146 implementedAt: timestamp,
147 createdAt: timestamp,
148 updatedAt: "2026-03-24T00:00:00-01:00",
149 },
150 {
151 planId: "plan-plus",
152 implementedAt: null,
153 createdAt: timestamp,
154 updatedAt: "2026-03-24T00:00:00+00:00",
155 },
156 ],
157 },
158 ];
159 for (const testCase of cases) {
160 const threadId = ThreadId.make(`thread-plan-order-${testCase.name}`);
161 const latestTurnId = TurnId.make(`turn-plan-order-${testCase.name}`);
162 for (const plan of testCase.rows) {
163 yield* plans.upsert({
164 ...plan,
165 planId: `${testCase.name}-${plan.planId}`,
166 threadId,
167 turnId: latestTurnId,
168 planMarkdown: "# Plan",
169 implementationThreadId: null,
170 });
171 }
172 assert.strictEqual(
173 yield* plans.hasActionableByThreadId({ threadId, latestTurnId }),
174 testCase.expected,
175 testCase.name,
176 );
177 assert.strictEqual(
178 yield* plans.hasActionableByThreadId({ threadId, latestTurnId: null }),
179 testCase.expected,
180 testCase.name,
181 );
182 }
183 }),
184 );
185
186 it.effect("returns only current-turn status metadata when old plans have large bodies", () =>
187 Effect.gen(function* () {
188 const plans = yield* ProjectionThreadProposedPlanRepository;
189 const sql = yield* SqlClient.SqlClient;
190 const threadId = ThreadId.make("thread-plan-metadata");
191 const latestTurnId = TurnId.make("turn-plan-metadata-current");
192 const updatedAt = "2026-03-24T00:00:00.000Z";
193 yield* sql`
194 WITH RECURSIVE history(n) AS (
195 SELECT 1 UNION ALL SELECT n + 1 FROM history WHERE n < 256
196 )
197 INSERT INTO projection_thread_proposed_plans (
198 plan_id, thread_id, turn_id, plan_markdown, implemented_at,
199 implementation_thread_id, created_at, updated_at
200 )
201 SELECT 'plan-metadata-old-' || n, ${threadId}, 'turn-plan-metadata-old',
202 ${"# Old plan\n".repeat(1024)}, ${updatedAt}, NULL, ${updatedAt}, ${updatedAt}
203 FROM history
204 `;
205 yield* plans.upsert({
206 planId: "plan-metadata-current",
207 threadId,
208 turnId: latestTurnId,
209 planMarkdown: "# Current plan",
210 implementedAt: null,
211 implementationThreadId: null,
212 createdAt: updatedAt,
213 updatedAt,
214 });
215 const statements: Array<Statement.Statement<unknown>> = [];
216 const actionable = yield* plans.hasActionableByThreadId({ threadId, latestTurnId }).pipe(
217 Effect.provideService(Statement.CurrentTransformer, (statement) => {
218 statements.push(statement);
219 return Effect.succeed(statement);
220 }),
221 );
222 assert.isTrue(actionable);
223 assert.strictEqual(statements.length, 1);
224 const statement = statements[0];
225 if (statement === undefined) return yield* Effect.die("Expected a plan status query.");
226 assert.deepEqual(yield* statement, [
227 { planId: "plan-metadata-current", implementedAt: null, updatedAt },
228 ]);
229 }),
230 );
231
23 it.effect("stores SQL NULL for missing project model options", () =>232 it.effect("stores SQL NULL for missing project model options", () =>
24 Effect.gen(function* () {233 Effect.gen(function* () {
25 const projects = yield* ProjectionProjectRepository;234 const projects = yield* ProjectionProjectRepository;
apps/server/src/persistence/Layers/ProjectionThreadProposedPlans.ts+52
...@@ -1,11 +1,13 @@...@@ -1,11 +1,13 @@
1import * as Effect from "effect/Effect";1import * as Effect from "effect/Effect";
2import * as Layer from "effect/Layer";2import * as Layer from "effect/Layer";
3import * as Schema from "effect/Schema";
3import * as SqlClient from "effect/unstable/sql/SqlClient";4import * as SqlClient from "effect/unstable/sql/SqlClient";
4import * as SqlSchema from "effect/unstable/sql/SqlSchema";5import * as SqlSchema from "effect/unstable/sql/SqlSchema";
56
6import { toPersistenceSqlError } from "../Errors.ts";7import { toPersistenceSqlError } from "../Errors.ts";
7import {8import {
8 DeleteProjectionThreadProposedPlansInput,9 DeleteProjectionThreadProposedPlansInput,
10 HasActionableProjectionThreadProposedPlanInput,
9 ListProjectionThreadProposedPlansInput,11 ListProjectionThreadProposedPlansInput,
10 ProjectionThreadProposedPlan,12 ProjectionThreadProposedPlan,
11 ProjectionThreadProposedPlanRepository,13 ProjectionThreadProposedPlanRepository,
...@@ -77,6 +79,55 @@ const makeProjectionThreadProposedPlanRepository = Effect.gen(function* () {...@@ -77,6 +79,55 @@ const makeProjectionThreadProposedPlanRepository = Effect.gen(function* () {
77 `,79 `,
78 });80 });
7981
82 const listPlanStatusCandidates = SqlSchema.findAll({
83 Request: HasActionableProjectionThreadProposedPlanInput,
84 Result: Schema.Struct({
85 planId: ProjectionThreadProposedPlan.fields.planId,
86 implementedAt: ProjectionThreadProposedPlan.fields.implementedAt,
87 updatedAt: ProjectionThreadProposedPlan.fields.updatedAt,
88 }),
89 execute: ({ threadId, latestTurnId }) => sql`
90 SELECT
91 plan_id AS "planId",
92 implemented_at AS "implementedAt",
93 updated_at AS "updatedAt"
94 FROM projection_thread_proposed_plans
95 WHERE thread_id = ${threadId}
96 AND (
97 turn_id = ${latestTurnId}
98 OR NOT EXISTS (
99 SELECT 1 FROM projection_thread_proposed_plans
100 WHERE thread_id = ${threadId} AND turn_id = ${latestTurnId}
101 )
102 )
103 ORDER BY created_at ASC, plan_id ASC
104 `,
105 });
106
107 const hasActionableByThreadId = Effect.fn(
108 "ProjectionThreadProposedPlanRepository.hasActionableByThreadId",
109 )(
110 function* (input: HasActionableProjectionThreadProposedPlanInput) {
111 const candidates = yield* listPlanStatusCandidates(input);
112 let selected: (typeof candidates)[number] | undefined;
113 // Timestamps and IDs use localeCompare, not SQLite byte order. Replace
114 // equal candidates to preserve the stable order of listByThreadId.
115 for (const candidate of candidates) {
116 if (
117 selected === undefined ||
118 (candidate.updatedAt.localeCompare(selected.updatedAt) ||
119 candidate.planId.localeCompare(selected.planId)) >= 0
120 ) {
121 selected = candidate;
122 }
123 }
124 return selected?.implementedAt === null;
125 },
126 Effect.mapError(
127 toPersistenceSqlError("ProjectionThreadProposedPlanRepository.hasActionableByThreadId:query"),
128 ),
129 );
130
80 const upsert: ProjectionThreadProposedPlanRepositoryShape["upsert"] = (row) =>131 const upsert: ProjectionThreadProposedPlanRepositoryShape["upsert"] = (row) =>
81 upsertProjectionThreadProposedPlanRow(row).pipe(132 upsertProjectionThreadProposedPlanRow(row).pipe(
82 Effect.mapError(toPersistenceSqlError("ProjectionThreadProposedPlanRepository.upsert:query")),133 Effect.mapError(toPersistenceSqlError("ProjectionThreadProposedPlanRepository.upsert:query")),
...@@ -101,6 +152,7 @@ const makeProjectionThreadProposedPlanRepository = Effect.gen(function* () {...@@ -101,6 +152,7 @@ const makeProjectionThreadProposedPlanRepository = Effect.gen(function* () {
101 return {152 return {
102 upsert,153 upsert,
103 listByThreadId,154 listByThreadId,
155 hasActionableByThreadId,
104 deleteByThreadId,156 deleteByThreadId,
105 } satisfies ProjectionThreadProposedPlanRepositoryShape;157 } satisfies ProjectionThreadProposedPlanRepositoryShape;
106});158});
apps/server/src/persistence/Services/ProjectionThreadProposedPlans.ts+10
...@@ -29,6 +29,13 @@ export const ListProjectionThreadProposedPlansInput = Schema.Struct({...@@ -29,6 +29,13 @@ export const ListProjectionThreadProposedPlansInput = Schema.Struct({
29export type ListProjectionThreadProposedPlansInput =29export type ListProjectionThreadProposedPlansInput =
30 typeof ListProjectionThreadProposedPlansInput.Type;30 typeof ListProjectionThreadProposedPlansInput.Type;
3131
32export const HasActionableProjectionThreadProposedPlanInput = Schema.Struct({
33 threadId: ThreadId,
34 latestTurnId: Schema.NullOr(TurnId),
35});
36export type HasActionableProjectionThreadProposedPlanInput =
37 typeof HasActionableProjectionThreadProposedPlanInput.Type;
38
32export const DeleteProjectionThreadProposedPlansInput = Schema.Struct({39export const DeleteProjectionThreadProposedPlansInput = Schema.Struct({
33 threadId: ThreadId,40 threadId: ThreadId,
34});41});
...@@ -42,6 +49,9 @@ export interface ProjectionThreadProposedPlanRepositoryShape {...@@ -42,6 +49,9 @@ export interface ProjectionThreadProposedPlanRepositoryShape {
42 readonly listByThreadId: (49 readonly listByThreadId: (
43 input: ListProjectionThreadProposedPlansInput,50 input: ListProjectionThreadProposedPlansInput,
44 ) => Effect.Effect<ReadonlyArray<ProjectionThreadProposedPlan>, ProjectionRepositoryError>;51 ) => Effect.Effect<ReadonlyArray<ProjectionThreadProposedPlan>, ProjectionRepositoryError>;
52 readonly hasActionableByThreadId: (
53 input: HasActionableProjectionThreadProposedPlanInput,
54 ) => Effect.Effect<boolean, ProjectionRepositoryError>;
45 readonly deleteByThreadId: (55 readonly deleteByThreadId: (
46 input: DeleteProjectionThreadProposedPlansInput,56 input: DeleteProjectionThreadProposedPlansInput,
47 ) => Effect.Effect<void, ProjectionRepositoryError>;57 ) => Effect.Effect<void, ProjectionRepositoryError>;