authorgravatar for me@t3.ggTheo Browne <me@t3.gg> 2026-08-31 20:36:53-07:00
committergravatar for noreply@github.comGitHub <noreply@github.com> 2026-08-31 20:36:53-07:00
logb17cc3d1bf0f4a5deee6dd6a470e36970b864a9b
treee5bcbcae83c0a536d8a0aea5a89efe217196c34b
parent42a8fd5103b4af57b16be721a9864ca3a5c18c5b
signature Signed by PGP key B5690EEEBB952194

perf(server): reduce frequency of full tool call output being loaded into memory from db (#8988)

10 files changed, 363 insertions(+), 98 deletions(-)

apps/server/src/orchestration/Layers/CheckpointReactor.ts+1-1
......@@ -164,7 +164,7 @@ const make = Effect.gen(function* () {
164164
165165 const resolveThreadDetail = Effect.fn("resolveThreadDetail")(function* (threadId: ThreadId) {
166166 return yield* projectionSnapshotQuery
167 .getThreadDetailById(threadId)
167 .getThreadDetailById(threadId, { activityKinds: [] })
168168 .pipe(Effect.map(Option.getOrUndefined));
169169 });
170170
apps/server/src/orchestration/Layers/ProjectionPipeline.test.ts+114-56
......@@ -2235,7 +2235,7 @@ it.layer(BaseTestLayer)("OrchestrationProjectionPipeline", (it) => {
22352235 }),
22362236 );
22372237
2238 it.effect("clears stale pending user input from projected shell summaries", () =>
2238 it.effect("reads only user-input activities when refreshing shell summaries", () =>
22392239 Effect.gen(function* () {
22402240 const projectionPipeline = yield* OrchestrationProjectionPipeline;
22412241 const eventStore = yield* OrchestrationEventStore;
......@@ -2293,70 +2293,128 @@ it.layer(BaseTestLayer)("OrchestrationProjectionPipeline", (it) => {
22932293 },
22942294 });
22952295
2296 // Invalid JSON proves the summary query filters tool rows before decoding payloads.
2297 yield* sql`
2298 INSERT INTO projection_thread_activities (
2299 activity_id,
2300 thread_id,
2301 turn_id,
2302 tone,
2303 kind,
2304 summary,
2305 payload_json,
2306 sequence,
2307 created_at
2308 )
2309 VALUES
2310 (
2311 'activity-malformed-tool-output',
2312 'thread-stale-user-input',
2313 NULL,
2314 'info',
2315 'tool.completed',
2316 'Tool completed',
2317 '{not-json',
2318 NULL,
2319 '2026-02-26T12:35:02.000Z'
2320 ),
2321 (
2322 'activity-user-input-resolved-requested',
2323 'thread-stale-user-input',
2324 NULL,
2325 'info',
2326 'user-input.requested',
2327 'User input requested',
2328 json_object('requestId', 'user-input-resolved'),
2329 NULL,
2330 '2026-02-26T12:35:03.000Z'
2331 ),
2332 (
2333 'activity-user-input-resolved',
2334 'thread-stale-user-input',
2335 NULL,
2336 'info',
2337 'user-input.resolved',
2338 'User input resolved',
2339 json_object('requestId', 'user-input-resolved'),
2340 NULL,
2341 '2026-02-26T12:35:04.000Z'
2342 ),
2343 (
2344 'activity-user-input-stale-requested',
2345 'thread-stale-user-input',
2346 NULL,
2347 'info',
2348 'user-input.requested',
2349 'User input requested',
2350 json_object('requestId', 'user-input-stale'),
2351 NULL,
2352 '2026-02-26T12:35:05.000Z'
2353 ),
2354 (
2355 'activity-user-input-stale-failed',
2356 'thread-stale-user-input',
2357 NULL,
2358 'error',
2359 'provider.user-input.respond.failed',
2360 'Provider user input response failed',
2361 json_object(
2362 'requestId',
2363 'user-input-stale',
2364 'detail',
2365 'Unknown pending Codex user input request: user-input-stale'
2366 ),
2367 NULL,
2368 '2026-02-26T12:35:06.000Z'
2369 ),
2370 (
2371 'activity-user-input-active-requested',
2372 'thread-stale-user-input',
2373 NULL,
2374 'info',
2375 'user-input.requested',
2376 'User input requested',
2377 json_object('requestId', 'user-input-active'),
2378 NULL,
2379 '2026-02-26T12:35:07.000Z'
2380 ),
2381 (
2382 'activity-user-input-active-failed',
2383 'thread-stale-user-input',
2384 NULL,
2385 'error',
2386 'provider.user-input.respond.failed',
2387 'Provider user input response failed',
2388 json_object(
2389 'requestId',
2390 'user-input-active',
2391 'detail',
2392 'Provider is temporarily unavailable'
2393 ),
2394 NULL,
2395 '2026-02-26T12:35:08.000Z'
2396 )
2397 `;
2398
22962399 yield* appendAndProject({
2297 type: "thread.activity-appended",
2400 type: "thread.message-sent",
22982401 eventId: EventId.make("evt-stale-user-input-3"),
22992402 aggregateKind: "thread",
23002403 aggregateId: ThreadId.make("thread-stale-user-input"),
2301 occurredAt: "2026-02-26T12:35:02.000Z",
2404 occurredAt: "2026-02-26T12:35:09.000Z",
23022405 commandId: CommandId.make("cmd-stale-user-input-3"),
23032406 causationEventId: null,
23042407 correlationId: CorrelationId.make("cmd-stale-user-input-3"),
23052408 metadata: {},
23062409 payload: {
23072410 threadId: ThreadId.make("thread-stale-user-input"),
2308 activity: {
2309 id: EventId.make("activity-stale-user-input-requested"),
2310 tone: "info",
2311 kind: "user-input.requested",
2312 summary: "User input requested",
2313 payload: {
2314 requestId: "user-input-request-stale-1",
2315 questions: [
2316 {
2317 id: "sandbox_mode",
2318 header: "Sandbox",
2319 question: "Which mode should be used?",
2320 options: [
2321 {
2322 label: "workspace-write",
2323 description: "Allow workspace writes only",
2324 },
2325 ],
2326 },
2327 ],
2328 },
2329 turnId: null,
2330 createdAt: "2026-02-26T12:35:02.000Z",
2331 },
2332 },
2333 });
2334
2335 yield* appendAndProject({
2336 type: "thread.activity-appended",
2337 eventId: EventId.make("evt-stale-user-input-4"),
2338 aggregateKind: "thread",
2339 aggregateId: ThreadId.make("thread-stale-user-input"),
2340 occurredAt: "2026-02-26T12:35:03.000Z",
2341 commandId: CommandId.make("cmd-stale-user-input-4"),
2342 causationEventId: null,
2343 correlationId: CorrelationId.make("cmd-stale-user-input-4"),
2344 metadata: {},
2345 payload: {
2346 threadId: ThreadId.make("thread-stale-user-input"),
2347 activity: {
2348 id: EventId.make("activity-stale-user-input-failed"),
2349 tone: "error",
2350 kind: "provider.user-input.respond.failed",
2351 summary: "Provider user input response failed",
2352 payload: {
2353 requestId: "user-input-request-stale-1",
2354 detail:
2355 "Provider adapter request failed (codex) for item/tool/requestUserInput: Unknown pending Codex user input request: user-input-request-stale-1",
2356 },
2357 turnId: null,
2358 createdAt: "2026-02-26T12:35:03.000Z",
2359 },
2411 messageId: MessageId.make("message-stale-user-input"),
2412 role: "user",
2413 text: "Continue",
2414 turnId: null,
2415 streaming: false,
2416 createdAt: "2026-02-26T12:35:09.000Z",
2417 updatedAt: "2026-02-26T12:35:09.000Z",
23602418 },
23612419 });
23622420
......@@ -2367,7 +2425,7 @@ it.layer(BaseTestLayer)("OrchestrationProjectionPipeline", (it) => {
23672425 FROM projection_threads
23682426 WHERE thread_id = 'thread-stale-user-input'
23692427 `;
2370 assert.deepEqual(threadRows, [{ pendingUserInputCount: 0 }]);
2428 assert.deepEqual(threadRows, [{ pendingUserInputCount: 1 }]);
23712429 }),
23722430 );
23732431
apps/server/src/orchestration/Layers/ProjectionPipeline.ts+2-2
......@@ -130,7 +130,7 @@ function isStalePendingApprovalFailureDetail(detail: string | null): boolean {
130130 );
131131}
132132
133// A full refresh loads all thread history, so skip events that cannot change the summary.
133// A refresh reads each persisted summary source, so skip events that cannot change the result.
134134function shouldRefreshThreadShellSummary(event: OrchestrationEvent): boolean {
135135 if (event.type === "thread.message-sent") {
136136 return event.payload.role === "user";
......@@ -588,7 +588,7 @@ const makeOrchestrationProjectionPipeline = Effect.fn("makeOrchestrationProjecti
588588 const [messages, proposedPlans, activities, pendingApprovals] = yield* Effect.all([
589589 projectionThreadMessageRepository.listByThreadId({ threadId }),
590590 projectionThreadProposedPlanRepository.listByThreadId({ threadId }),
591 projectionThreadActivityRepository.listByThreadId({ threadId }),
591 projectionThreadActivityRepository.listUserInputLifecycleByThreadId({ threadId }),
592592 projectionPendingApprovalRepository.listByThreadId({ threadId }),
593593 ]);
594594
apps/server/src/orchestration/Layers/ProjectionSnapshotQuery.test.ts+71
......@@ -484,6 +484,77 @@ projectionSnapshotLayer("ProjectionSnapshotQuery", (it) => {
484484 if (threadDetail._tag === "Some") {
485485 assert.deepEqual(threadDetail.value, snapshot.threads[0]);
486486 }
487
488 yield* sql`
489 INSERT INTO projection_thread_activities (
490 activity_id,
491 thread_id,
492 turn_id,
493 tone,
494 kind,
495 summary,
496 payload_json,
497 created_at
498 )
499 VALUES
500 (
501 'activity-task-started',
502 'thread-1',
503 'turn-1',
504 'info',
505 'task.started',
506 'Ship the query filter',
507 '{"taskId":"task-1","detail":"Ship the query filter"}',
508 '2026-02-24T00:00:06.100Z'
509 ),
510 (
511 'activity-malformed-tool',
512 'thread-1',
513 'turn-1',
514 'info',
515 'tool.completed',
516 'Malformed tool output',
517 'not-json',
518 '2026-02-24T00:00:06.200Z'
519 )
520 `;
521
522 const detailWithoutActivities = yield* snapshotQuery.getThreadDetailById(
523 ThreadId.make("thread-1"),
524 { activityKinds: [] },
525 );
526 assert.equal(detailWithoutActivities._tag, "Some");
527 if (detailWithoutActivities._tag === "Some") {
528 assert.deepEqual(detailWithoutActivities.value.activities, []);
529 assert.deepEqual(detailWithoutActivities.value.messages, snapshot.threads[0]?.messages);
530 assert.deepEqual(
531 detailWithoutActivities.value.proposedPlans,
532 snapshot.threads[0]?.proposedPlans,
533 );
534 assert.deepEqual(
535 detailWithoutActivities.value.checkpoints,
536 snapshot.threads[0]?.checkpoints,
537 );
538 }
539
540 const detailWithTaskActivities = yield* snapshotQuery.getThreadDetailById(
541 ThreadId.make("thread-1"),
542 { activityKinds: ["task.started", "task.progress"] },
543 );
544 assert.equal(detailWithTaskActivities._tag, "Some");
545 if (detailWithTaskActivities._tag === "Some") {
546 assert.deepEqual(detailWithTaskActivities.value.activities, [
547 {
548 id: asEventId("activity-task-started"),
549 tone: "info",
550 kind: "task.started",
551 summary: "Ship the query filter",
552 payload: { taskId: "task-1", detail: "Ship the query filter" },
553 turnId: asTurnId("turn-1"),
554 createdAt: "2026-02-24T00:00:06.100Z",
555 },
556 ]);
557 }
487558 }),
488559 );
489560
apps/server/src/orchestration/Layers/ProjectionSnapshotQuery.ts+89-22
......@@ -64,6 +64,7 @@ import {
6464 type ProjectionFullThreadDiffContext,
6565 type ProjectionSnapshotCounts,
6666 type ProjectionThreadCheckpointContext,
67 type ProjectionThreadDetailQuery,
6768 type ProjectionSnapshotQueryShape,
6869} from "../Services/ProjectionSnapshotQuery.ts";
6970
......@@ -141,6 +142,10 @@ const ProjectIdLookupInput = Schema.Struct({
141142const ThreadIdLookupInput = Schema.Struct({
142143 threadId: ThreadId,
143144});
145const ThreadActivityKindsLookupInput = Schema.Struct({
146 threadId: ThreadId,
147 activityKinds: Schema.Array(Schema.String),
148});
144149// Windowed reads order turns by the stable keyset (anchor, turn key), where
145150// anchor is requested_at and turn key is
146151// COALESCE(turn_id, ''). Both are event-derived, so cursors survive the
......@@ -1055,6 +1060,48 @@ const makeProjectionSnapshotQuery = Effect.gen(function* () {
10551060 `,
10561061 });
10571062
1063 const listThreadActivityRowsByThreadAndKinds = SqlSchema.findAll({
1064 Request: ThreadActivityKindsLookupInput,
1065 Result: ProjectionThreadActivityDbRowSchema,
1066 execute: ({ threadId, activityKinds }) =>
1067 sql`
1068 SELECT
1069 activity_id AS "activityId",
1070 thread_id AS "threadId",
1071 turn_id AS "turnId",
1072 tone,
1073 kind,
1074 summary,
1075 payload_json AS "payload",
1076 sequence,
1077 created_at AS "createdAt"
1078 FROM (
1079 SELECT
1080 activity_id,
1081 thread_id,
1082 turn_id,
1083 tone,
1084 kind,
1085 summary,
1086 payload_json,
1087 sequence,
1088 created_at
1089 FROM projection_thread_activities
1090 WHERE thread_id = ${threadId}
1091 AND ${sql.in("kind", activityKinds)}
1092 ORDER BY
1093 sequence DESC,
1094 created_at DESC,
1095 activity_id DESC
1096 LIMIT ${THREAD_DETAIL_ACTIVITY_LIMIT}
1097 ) AS recent_activities
1098 ORDER BY
1099 sequence ASC,
1100 created_at ASC,
1101 activity_id ASC
1102 `,
1103 });
1104
10581105 const getThreadSessionRowByThread = SqlSchema.findOneOption({
10591106 Request: ThreadIdLookupInput,
10601107 Result: ProjectionThreadSessionDbRowSchema,
......@@ -2524,8 +2571,43 @@ const makeProjectionSnapshotQuery = Effect.gen(function* () {
25242571 readonly beforeTurnKey: string;
25252572 }
25262573
2527 const getThreadDetailByIdBounded = (threadId: ThreadId, bounds: ThreadDetailBounds | undefined) =>
2574 const getThreadDetailByIdBounded = (
2575 threadId: ThreadId,
2576 bounds: ThreadDetailBounds | undefined,
2577 query: ProjectionThreadDetailQuery | undefined = undefined,
2578 ) =>
25282579 Effect.gen(function* () {
2580 const activityRowsEffect = (
2581 query?.activityKinds === undefined
2582 ? bounds === undefined
2583 ? listThreadActivityRowsByThread({ threadId })
2584 : listThreadActivityRowsByThreadWindow({ threadId, ...bounds })
2585 : query.activityKinds.length === 0
2586 ? Effect.succeed([])
2587 : listThreadActivityRowsByThreadAndKinds({
2588 threadId,
2589 activityKinds: query.activityKinds,
2590 })
2591 ).pipe(
2592 Effect.mapError(
2593 toPersistenceSqlOrDecodeError(
2594 "ProjectionSnapshotQuery.getThreadDetailById:listActivities:query",
2595 "ProjectionSnapshotQuery.getThreadDetailById:listActivities:decodeRows",
2596 ),
2597 ),
2598 );
2599 const pinnedActivityRowsEffect =
2600 query?.activityKinds === undefined
2601 ? listPinnedThreadActivityRowsByThread({ threadId }).pipe(
2602 Effect.mapError(
2603 toPersistenceSqlOrDecodeError(
2604 "ProjectionSnapshotQuery.getThreadDetailById:listPinnedActivities:query",
2605 "ProjectionSnapshotQuery.getThreadDetailById:listPinnedActivities:decodeRows",
2606 ),
2607 ),
2608 )
2609 : Effect.succeed([]);
2610
25292611 const [
25302612 threadRow,
25312613 messageRows,
......@@ -2563,25 +2645,8 @@ const makeProjectionSnapshotQuery = Effect.gen(function* () {
25632645 ),
25642646 ),
25652647 ),
2566 (bounds === undefined
2567 ? listThreadActivityRowsByThread({ threadId })
2568 : listThreadActivityRowsByThreadWindow({ threadId, ...bounds })
2569 ).pipe(
2570 Effect.mapError(
2571 toPersistenceSqlOrDecodeError(
2572 "ProjectionSnapshotQuery.getThreadDetailById:listActivities:query",
2573 "ProjectionSnapshotQuery.getThreadDetailById:listActivities:decodeRows",
2574 ),
2575 ),
2576 ),
2577 listPinnedThreadActivityRowsByThread({ threadId }).pipe(
2578 Effect.mapError(
2579 toPersistenceSqlOrDecodeError(
2580 "ProjectionSnapshotQuery.getThreadDetailById:listPinnedActivities:query",
2581 "ProjectionSnapshotQuery.getThreadDetailById:listPinnedActivities:decodeRows",
2582 ),
2583 ),
2584 ),
2648 activityRowsEffect,
2649 pinnedActivityRowsEffect,
25852650 listCheckpointRowsByThread({ threadId }).pipe(
25862651 Effect.mapError(
25872652 toPersistenceSqlOrDecodeError(
......@@ -2700,8 +2765,10 @@ const makeProjectionSnapshotQuery = Effect.gen(function* () {
27002765 );
27012766 });
27022767
2703 const getThreadDetailById: ProjectionSnapshotQueryShape["getThreadDetailById"] = (threadId) =>
2704 getThreadDetailByIdBounded(threadId, undefined);
2768 const getThreadDetailById: ProjectionSnapshotQueryShape["getThreadDetailById"] = (
2769 threadId,
2770 query,
2771 ) => getThreadDetailByIdBounded(threadId, undefined, query);
27052772
27062773 // Bounds pathological fan-out: one user turn that spawned hundreds of
27072774 // subagent turns still pages in bounded chunks, at the cost of splitting the
apps/server/src/orchestration/Layers/ProviderCommandReactor.ts+1-1
......@@ -481,7 +481,7 @@ const make = Effect.gen(function* () {
481481
482482 const resolveThread = Effect.fnUntraced(function* (threadId: ThreadId) {
483483 return yield* projectionSnapshotQuery
484 .getThreadDetailById(threadId)
484 .getThreadDetailById(threadId, { activityKinds: [] })
485485 .pipe(Effect.map(Option.getOrUndefined));
486486 });
487487
apps/server/src/orchestration/Layers/ProviderRuntimeIngestion.ts+7-3
......@@ -48,6 +48,7 @@ import { canReplaceThreadTitle } from "../threadTitles.ts";
4848
4949const providerTurnKey = (threadId: ThreadId, turnId: TurnId) => `${threadId}:${turnId}`;
5050const providerTaskKey = (threadId: ThreadId, taskId: string) => `${threadId}:${taskId}`;
51const TASK_TITLE_ACTIVITY_KINDS = ["task.started", "task.progress"] as const;
5152
5253// Fallback when the in-memory description cache no longer has the task name
5354// (server restart, session-exit sweep, TTL/capacity eviction): earlier
......@@ -949,9 +950,12 @@ const make = Effect.gen(function* () {
949950 ),
950951 );
951952
952 const resolveThreadDetail = Effect.fn("resolveThreadDetail")(function* (threadId: ThreadId) {
953 const resolveThreadDetail = Effect.fn("resolveThreadDetail")(function* (
954 threadId: ThreadId,
955 activityKinds: ReadonlyArray<string> = [],
956 ) {
953957 return yield* projectionSnapshotQuery
954 .getThreadDetailById(threadId)
958 .getThreadDetailById(threadId, { activityKinds })
955959 .pipe(Effect.map(Option.getOrUndefined));
956960 });
957961
......@@ -2022,7 +2026,7 @@ const make = Effect.gen(function* () {
20222026 if (event.type === "task.completed") {
20232027 taskTitle = yield* lookupTaskDescription(thread.id, event.payload.taskId);
20242028 if (!taskTitle) {
2025 const threadDetail = yield* getLoadedThreadDetail();
2029 const threadDetail = yield* resolveThreadDetail(thread.id, TASK_TITLE_ACTIVITY_KINDS);
20262030 taskTitle = findTaskTitleInActivities(threadDetail?.activities, event.payload.taskId);
20272031 }
20282032 }
apps/server/src/orchestration/Services/ProjectionSnapshotQuery.ts+10
......@@ -54,6 +54,15 @@ export interface ProjectionFullThreadDiffContext {
5454 readonly toCheckpointRef: CheckpointRef | null;
5555}
5656
57export interface ProjectionThreadDetailQuery {
58 /**
59 * Limit activities before SQLite returns and decodes their payloads.
60 * Any explicit filter omits pinned-request reads. An empty list also skips
61 * the activity query. Omit this option to preserve the full detail response.
62 */
63 readonly activityKinds?: ReadonlyArray<string>;
64}
65
5766/**
5867 * ProjectionSnapshotQueryShape - Service API for read-model snapshots.
5968 */
......@@ -168,6 +177,7 @@ export interface ProjectionSnapshotQueryShape {
168177 */
169178 readonly getThreadDetailById: (
170179 threadId: ThreadId,
180 query?: ProjectionThreadDetailQuery,
171181 ) => Effect.Effect<Option.Option<OrchestrationThread>, ProjectionRepositoryError>;
172182
173183 /**
apps/server/src/persistence/Layers/ProjectionThreadActivities.ts+59-13
......@@ -23,6 +23,21 @@ const ProjectionThreadActivityDbRowSchema = ProjectionThreadActivity.mapFields(
2323 }),
2424);
2525
26const mapActivityRows = (
27 rows: ReadonlyArray<Schema.Schema.Type<typeof ProjectionThreadActivityDbRowSchema>>,
28): ReadonlyArray<ProjectionThreadActivity> =>
29 rows.map((row) => ({
30 activityId: row.activityId,
31 threadId: row.threadId,
32 turnId: row.turnId,
33 tone: row.tone,
34 kind: row.kind,
35 summary: row.summary,
36 payload: row.payload,
37 ...(row.sequence !== null ? { sequence: row.sequence } : {}),
38 createdAt: row.createdAt,
39 }));
40
2641function toPersistenceSqlOrDecodeError(sqlOperation: string, decodeOperation: string) {
2742 return (cause: unknown) =>
2843 Schema.isSchemaError(cause)
......@@ -97,6 +112,36 @@ const makeProjectionThreadActivityRepository = Effect.gen(function* () {
97112 `,
98113 });
99114
115 const listUserInputLifecycleActivityRows = SqlSchema.findAll({
116 Request: ListProjectionThreadActivitiesInput,
117 Result: ProjectionThreadActivityDbRowSchema,
118 execute: ({ threadId }) =>
119 sql`
120 SELECT
121 activity_id AS "activityId",
122 thread_id AS "threadId",
123 turn_id AS "turnId",
124 tone,
125 kind,
126 summary,
127 payload_json AS "payload",
128 sequence,
129 created_at AS "createdAt"
130 FROM projection_thread_activities
131 WHERE thread_id = ${threadId}
132 AND kind IN (
133 'user-input.requested',
134 'user-input.resolved',
135 'provider.user-input.respond.failed'
136 )
137 ORDER BY
138 CASE WHEN sequence IS NULL THEN 0 ELSE 1 END ASC,
139 sequence ASC,
140 created_at ASC,
141 activity_id ASC
142 `,
143 });
144
100145 const deleteProjectionThreadActivityRows = SqlSchema.void({
101146 Request: DeleteProjectionThreadActivitiesInput,
102147 execute: ({ threadId }) =>
......@@ -124,21 +169,21 @@ const makeProjectionThreadActivityRepository = Effect.gen(function* () {
124169 "ProjectionThreadActivityRepository.listByThreadId:decodeRows",
125170 ),
126171 ),
127 Effect.map((rows) =>
128 rows.map((row) => ({
129 activityId: row.activityId,
130 threadId: row.threadId,
131 turnId: row.turnId,
132 tone: row.tone,
133 kind: row.kind,
134 summary: row.summary,
135 payload: row.payload,
136 ...(row.sequence !== null ? { sequence: row.sequence } : {}),
137 createdAt: row.createdAt,
138 })),
139 ),
172 Effect.map(mapActivityRows),
140173 );
141174
175 const listUserInputLifecycleByThreadId: ProjectionThreadActivityRepositoryShape["listUserInputLifecycleByThreadId"] =
176 (input) =>
177 listUserInputLifecycleActivityRows(input).pipe(
178 Effect.mapError(
179 toPersistenceSqlOrDecodeError(
180 "ProjectionThreadActivityRepository.listUserInputLifecycleByThreadId:query",
181 "ProjectionThreadActivityRepository.listUserInputLifecycleByThreadId:decodeRows",
182 ),
183 ),
184 Effect.map(mapActivityRows),
185 );
186
142187 const deleteByThreadId: ProjectionThreadActivityRepositoryShape["deleteByThreadId"] = (input) =>
143188 deleteProjectionThreadActivityRows(input).pipe(
144189 Effect.mapError(
......@@ -149,6 +194,7 @@ const makeProjectionThreadActivityRepository = Effect.gen(function* () {
149194 return {
150195 upsert,
151196 listByThreadId,
197 listUserInputLifecycleByThreadId,
152198 deleteByThreadId,
153199 } satisfies ProjectionThreadActivityRepositoryShape;
154200});
apps/server/src/persistence/Services/ProjectionThreadActivities.ts+9
......@@ -67,6 +67,15 @@ export interface ProjectionThreadActivityRepositoryShape {
6767 input: ListProjectionThreadActivitiesInput,
6868 ) => Effect.Effect<ReadonlyArray<ProjectionThreadActivity>, ProjectionRepositoryError>;
6969
70 /**
71 * List activity rows used to derive pending user-input state.
72 *
73 * Filters in SQLite so unrelated payloads do not enter server memory.
74 */
75 readonly listUserInputLifecycleByThreadId: (
76 input: ListProjectionThreadActivitiesInput,
77 ) => Effect.Effect<ReadonlyArray<ProjectionThreadActivity>, ProjectionRepositoryError>;
78
7079 /**
7180 * Delete projected thread activity rows by thread.
7281 */