authorgravatar for me@t3.ggTheo Browne <me@t3.gg> 2026-09-01 01:25:24-07:00
committergravatar for noreply@github.comGitHub <noreply@github.com> 2026-09-01 01:25:24-07:00
loga9ffb8279614df6ae2f1f4b7f09a0dc42edef797
tree5239663b66231d7bbfed7a9f26d46576b7823c31
parent73776d4e52087331245e82ebf3c053ae0991935d
signature Signed by PGP key B5690EEEBB952194

perf(server): bound snapshot activity payload memory (#9000)

5 files changed, 449 insertions(+), 94 deletions(-)

apps/server/src/orchestration/ActivityPayloadProjection.test.ts+22
......@@ -64,6 +64,28 @@ describe("projectActivityPayload", () => {
6464 expect(JSON.stringify(projected.payload).length).toBeLessThan(500);
6565 });
6666
67 it("keeps preview normalization and fence-only fallback while scanning lines", () => {
68 const preview = projectActivityPayload(
69 activity({
70 itemType: "command_execution",
71 data: { rawOutput: `\`\`\`\n actual\tresult \n${"x".repeat(5000)}` },
72 }),
73 );
74 const fences = projectActivityPayload(
75 activity({
76 itemType: "command_execution",
77 data: { rawOutput: "```\r\n \t \n```\n" },
78 }),
79 );
80
81 expect((preview.payload as { data: { rawOutput: unknown } }).data.rawOutput).toEqual({
82 content: "actual result",
83 });
84 expect((fences.payload as { data: { rawOutput: unknown } }).data.rawOutput).toEqual({
85 content: "2 lines",
86 });
87 });
88
6789 it("keeps bounded Claude and ACP command output summaries", () => {
6890 const claude = projectActivityPayload(
6991 activity({
apps/server/src/orchestration/ActivityPayloadProjection.ts+19-12
......@@ -144,22 +144,29 @@ function projectCommandValue(data: Record<string, unknown>): unknown {
144144}
145145
146146function summarizeToolTextOutput(value: string): string | null {
147 const lines: string[] = [];
148 for (const rawLine of value.split(/\r?\n/u)) {
149 const line = rawLine.replace(/\s+/g, " ").trim();
147 let meaningfulLineCount = 0;
148 let offset = 0;
149
150 while (offset <= value.length) {
151 const newlineIndex = value.indexOf("\n", offset);
152 const lineEnd = newlineIndex === -1 ? value.length : newlineIndex;
153 const line = value.slice(offset, lineEnd).replace(/\s+/g, " ").trim();
150154 if (line.length > 0) {
151 lines.push(line);
155 meaningfulLineCount += 1;
156 if (line !== "```") {
157 const summary = line.length <= 84 ? line : `${line.slice(0, 83).trimEnd()}…`;
158 // V8 can retain the full tool output behind a short sliced string.
159 // Join a tiny character array so the returned preview owns its bytes.
160 return Array.from(summary).join("");
161 }
162 }
163 if (newlineIndex === -1) {
164 break;
152165 }
166 offset = newlineIndex + 1;
153167 }
154168
155 const firstLine = lines.find((line) => line !== "```");
156 if (firstLine) {
157 return firstLine.length <= 84 ? firstLine : `${firstLine.slice(0, 83).trimEnd()}…`;
158 }
159 if (lines.length > 1) {
160 return `${lines.length.toLocaleString()} lines`;
161 }
162 return null;
169 return meaningfulLineCount > 1 ? `${meaningfulLineCount.toLocaleString()} lines` : null;
163170}
164171
165172/**
apps/server/src/orchestration/Layers/ProjectionSnapshotQuery.test.ts+145-12
......@@ -21,6 +21,7 @@ import * as ThreadBackgroundLiveness from "../ThreadBackgroundLiveness.ts";
2121import * as ThreadPlanProgress from "../ThreadPlanProgress.ts";
2222import { ProjectionSnapshotQuery } from "../Services/ProjectionSnapshotQuery.ts";
2323import { encodeThreadDetailPageCursor } from "../threadDetailCursor.ts";
24import { projectThreadDetailSnapshot } from "../ActivityPayloadProjection.ts";
2425
2526const asProjectId = (value: string): ProjectId => ProjectId.make(value);
2627const asTurnId = (value: string): TurnId => TurnId.make(value);
......@@ -2389,9 +2390,84 @@ projectionSnapshotLayer("ProjectionSnapshotQuery windowed thread detail", (it) =
23892390 'thread-w',
23902391 'turn-5',
23912392 'tool',
2392 'tool.completed',
2393 CASE
2394 WHEN sequence = 2 THEN 'tool.updated'
2395 WHEN sequence IN (3, 70) THEN 'context-window.updated'
2396 ELSE 'tool.completed'
2397 END,
23932398 'ran tool',
2394 printf('{"sequence":%d}', sequence),
2399 CASE
2400 WHEN sequence IN (2, 80) THEN json_object(
2401 'itemType', 'command_execution',
2402 'toolCallId', 'cross-batch-call',
2403 'title', CASE WHEN sequence = 80 THEN 'Build completed' ELSE 'Build' END,
2404 'status', 'completed',
2405 'data', json_object(
2406 'toolCallId', 'cross-batch-call',
2407 'item', json_object(
2408 'command', 'vp test run',
2409 'aggregatedOutput', printf(
2410 'command output%s%s',
2411 char(10),
2412 replace(hex(zeroblob(8192)), '00', 'x')
2413 )
2414 ),
2415 'rawOutput', printf(
2416 'raw output%s%s',
2417 char(10),
2418 replace(hex(zeroblob(8192)), '00', 'y')
2419 ),
2420 'files', json_array(json_object('path', 'apps/server/src/snapshot.ts'))
2421 )
2422 )
2423 WHEN sequence = 10 THEN json_object(
2424 'itemType', 'mcp_tool_call',
2425 'status', 'completed',
2426 'data', json_object(
2427 'item', json_object(
2428 'type', 'mcpToolCall',
2429 'id', 'mcp-item-10',
2430 'tool', 'fetch_pr',
2431 'server', 'github',
2432 'status', 'completed',
2433 'arguments', json_object('pr', 42),
2434 'result', json_object(
2435 'content', json_array(json_object(
2436 'type', 'text',
2437 'text', printf(
2438 'PR body line one%s%s',
2439 char(10),
2440 replace(hex(zeroblob(8192)), '00', 'z')
2441 )
2442 ))
2443 ),
2444 '_meta', json_object('raw', replace(hex(zeroblob(8192)), '00', 'q'))
2445 )
2446 )
2447 )
2448 WHEN sequence = 11 THEN json_object(
2449 'itemType', 'command_execution',
2450 'status', 'completed',
2451 'data', json_object(
2452 'item', json_object(
2453 'status', 'failed',
2454 'command', 'vp test run',
2455 'aggregatedOutput', printf(
2456 'failed command%s%s',
2457 char(10),
2458 replace(hex(zeroblob(8192)), '00', 'w')
2459 )
2460 ),
2461 'rawOutput', json_object('stdout', 'failed output'),
2462 'files', json_array(json_object('path', 'apps/server/src/failed.ts'))
2463 )
2464 )
2465 WHEN sequence IN (3, 70) THEN json_object(
2466 'usedTokens', sequence * 100,
2467 'modelContextWindow', 100000
2468 )
2469 ELSE json_object('sequence', sequence)
2470 END,
23952471 sequence,
23962472 '2026-03-01T00:04:00.000Z'
23972473 FROM activity_rows
......@@ -2469,12 +2545,14 @@ projectionSnapshotLayer("ProjectionSnapshotQuery windowed thread detail", (it) =
24692545 const detailWithPinnedRequests = yield* snapshotQuery.getThreadDetailById(threadW);
24702546 assert.equal(detailWithPinnedRequests._tag, "Some");
24712547 if (detailWithPinnedRequests._tag === "Some") {
2472 const ids = detailWithPinnedRequests.value.activities.map((activity) => activity.id);
2548 const ids = new Set(
2549 detailWithPinnedRequests.value.activities.map((activity) => activity.id),
2550 );
24732551 assert.equal(detailWithPinnedRequests.value.activities.length, 503);
2474 assert.equal(ids.includes(asEventId("approval-old")), true);
2475 assert.equal(ids.includes(asEventId("user-input-old")), true);
2476 assert.equal(ids.includes(asEventId("user-input-closed")), false);
2477 assert.equal(ids.includes(asEventId("user-input-tied-z-request")), true);
2552 assert.equal(ids.has(asEventId("approval-old")), true);
2553 assert.equal(ids.has(asEventId("user-input-old")), true);
2554 assert.equal(ids.has(asEventId("user-input-closed")), false);
2555 assert.equal(ids.has(asEventId("user-input-tied-z-request")), true);
24782556 }
24792557
24802558 const windowWithPinnedRequests = yield* snapshotQuery.getThreadDetailSnapshot(threadW, {
......@@ -2482,12 +2560,67 @@ projectionSnapshotLayer("ProjectionSnapshotQuery windowed thread detail", (it) =
24822560 });
24832561 assert.equal(windowWithPinnedRequests._tag, "Some");
24842562 if (windowWithPinnedRequests._tag === "Some") {
2485 const ids = windowWithPinnedRequests.value.thread.activities.map((activity) => activity.id);
2563 const ids = new Set(
2564 windowWithPinnedRequests.value.thread.activities.map((activity) => activity.id),
2565 );
24862566 assert.equal(windowWithPinnedRequests.value.thread.activities.length, 503);
2487 assert.equal(ids.includes(asEventId("approval-old")), true);
2488 assert.equal(ids.includes(asEventId("user-input-old")), true);
2489 assert.equal(ids.includes(asEventId("user-input-closed")), false);
2490 assert.equal(ids.includes(asEventId("user-input-tied-z-request")), true);
2567 assert.equal(ids.has(asEventId("approval-old")), true);
2568 assert.equal(ids.has(asEventId("user-input-old")), true);
2569 assert.equal(ids.has(asEventId("user-input-closed")), false);
2570 assert.equal(ids.has(asEventId("user-input-tied-z-request")), true);
2571 }
2572
2573 const fullSnapshot = yield* snapshotQuery.getThreadDetailSnapshot(threadW);
2574 assert.equal(fullSnapshot._tag, "Some");
2575 if (
2576 detailWithPinnedRequests._tag === "Some" &&
2577 fullSnapshot._tag === "Some" &&
2578 windowWithPinnedRequests._tag === "Some"
2579 ) {
2580 const projectedFullSnapshot = projectThreadDetailSnapshot(fullSnapshot.value);
2581 const projectedRawBaseline = projectThreadDetailSnapshot({
2582 snapshotSequence: fullSnapshot.value.snapshotSequence,
2583 thread: detailWithPinnedRequests.value,
2584 });
2585 assert.deepStrictEqual(projectedFullSnapshot, projectedRawBaseline);
2586
2587 const rawActivitiesById = new Map(
2588 detailWithPinnedRequests.value.activities.map((activity) => [activity.id, activity]),
2589 );
2590 const projectedWindowSnapshot = projectThreadDetailSnapshot(windowWithPinnedRequests.value);
2591 const projectedWindowBaseline = projectThreadDetailSnapshot({
2592 ...windowWithPinnedRequests.value,
2593 thread: {
2594 ...windowWithPinnedRequests.value.thread,
2595 activities: windowWithPinnedRequests.value.thread.activities.map(
2596 (activity) => rawActivitiesById.get(activity.id) ?? activity,
2597 ),
2598 },
2599 });
2600 assert.deepStrictEqual(projectedWindowSnapshot, projectedWindowBaseline);
2601
2602 const projectedIds = new Set(
2603 projectedFullSnapshot.thread.activities.map((activity) => activity.id),
2604 );
2605 assert.equal(projectedIds.has(asEventId("activity-0002")), false);
2606 assert.equal(projectedIds.has(asEventId("activity-0003")), false);
2607 assert.equal(projectedIds.has(asEventId("activity-0070")), true);
2608
2609 const failedCommand = projectedFullSnapshot.thread.activities.find(
2610 (activity) => activity.id === asEventId("activity-0011"),
2611 );
2612 assert.deepStrictEqual(failedCommand?.payload, {
2613 itemType: "command_execution",
2614 status: "failed",
2615 data: {
2616 item: {
2617 command: "vp test run",
2618 aggregatedOutput: "failed command",
2619 },
2620 files: [{ path: "apps/server/src/failed.ts" }],
2621 rawOutput: { content: "failed output" },
2622 },
2623 });
24912624 }
24922625 }),
24932626 );
apps/server/src/orchestration/Layers/ProjectionSnapshotQuery.ts+259-70
......@@ -57,6 +57,7 @@ import {
5757 decodeThreadDetailPageCursor,
5858 encodeThreadDetailPageCursor,
5959} from "../threadDetailCursor.ts";
60import { projectActivityPayload } from "../ActivityPayloadProjection.ts";
6061import * as RepositoryIdentityResolver from "../../project/RepositoryIdentityResolver.ts";
6162import { ORCHESTRATION_PROJECTOR_NAMES } from "./ProjectionPipeline.ts";
6263import {
......@@ -75,6 +76,9 @@ const decodeThread = Schema.decodeUnknownEffect(OrchestrationThread);
7576// activity window. Applying the limit in SQL avoids decoding an unbounded
7677// payload_json set before the projector can enforce that invariant.
7778const THREAD_DETAIL_ACTIVITY_LIMIT = 500;
79// Snapshot payloads are decoded and projected in small sequential batches so
80// one client read does not retain the raw payloads for the full activity window.
81const THREAD_DETAIL_ACTIVITY_PAYLOAD_BATCH_SIZE = 25;
7882const ProjectionProjectDbRowSchema = ProjectionProject.mapFields(
7983 Struct.assign({
8084 defaultModelSelection: Schema.NullOr(Schema.fromJsonString(ModelSelection)),
......@@ -100,6 +104,9 @@ const ProjectionThreadActivityDbRowSchema = ProjectionThreadActivity.mapFields(
100104 sequence: Schema.NullOr(NonNegativeInt),
101105 }),
102106);
107const ProjectionThreadActivityIdRowSchema = Schema.Struct({
108 activityId: ProjectionThreadActivity.fields.activityId,
109});
103110const ProjectionThreadSessionDbRowSchema = ProjectionThreadSession;
104111const ProjectionCheckpointDbRowSchema = ProjectionCheckpoint.mapFields(
105112 Struct.assign({
......@@ -146,6 +153,9 @@ const ThreadActivityKindsLookupInput = Schema.Struct({
146153 threadId: ThreadId,
147154 activityKinds: Schema.Array(Schema.String),
148155});
156const ThreadActivityIdsLookupInput = Schema.Struct({
157 activityIds: Schema.Array(ProjectionThreadActivity.fields.activityId),
158});
149159// Windowed reads order turns by the stable keyset (anchor, turn key), where
150160// anchor is requested_at and turn key is
151161// COALESCE(turn_id, ''). Both are event-derived, so cursors survive the
......@@ -349,6 +359,21 @@ function mapProposedPlanRow(
349359 };
350360}
351361
362function mapThreadActivityRow(
363 row: Schema.Schema.Type<typeof ProjectionThreadActivityDbRowSchema>,
364): OrchestrationThreadActivity {
365 return {
366 id: row.activityId,
367 tone: row.tone,
368 kind: row.kind,
369 summary: row.summary,
370 payload: row.payload,
371 turnId: row.turnId,
372 createdAt: row.createdAt,
373 ...(row.sequence !== null ? { sequence: row.sequence } : {}),
374 };
375}
376
352377function toPersistenceSqlOrDecodeError(sqlOperation: string, decodeOperation: string) {
353378 return (cause: unknown): ProjectionRepositoryError =>
354379 Schema.isSchemaError(cause)
......@@ -1060,6 +1085,44 @@ const makeProjectionSnapshotQuery = Effect.gen(function* () {
10601085 `,
10611086 });
10621087
1088 const listThreadActivityIdsByThread = SqlSchema.findAll({
1089 Request: ThreadIdLookupInput,
1090 Result: ProjectionThreadActivityIdRowSchema,
1091 execute: ({ threadId }) =>
1092 sql`
1093 SELECT activity_id AS "activityId"
1094 FROM projection_thread_activities
1095 WHERE thread_id = ${threadId}
1096 ORDER BY
1097 sequence DESC,
1098 created_at DESC,
1099 activity_id DESC
1100 LIMIT ${THREAD_DETAIL_ACTIVITY_LIMIT}
1101 `,
1102 });
1103
1104 const listThreadActivityRowsByIds = SqlSchema.findAll({
1105 Request: ThreadActivityIdsLookupInput,
1106 Result: ProjectionThreadActivityDbRowSchema,
1107 execute: ({ activityIds }) =>
1108 sql`
1109 SELECT
1110 activity_id AS "activityId",
1111 thread_id AS "threadId",
1112 turn_id AS "turnId",
1113 tone,
1114 kind,
1115 summary,
1116 payload_json AS "payload",
1117 sequence,
1118 created_at AS "createdAt"
1119 FROM projection_thread_activities
1120 -- The selectors already scoped these globally unique ids to the
1121 -- thread inside this transaction. Keep this as a primary-key lookup.
1122 WHERE ${sql.in("activity_id", activityIds)}
1123 `,
1124 });
1125
10631126 const listThreadActivityRowsByThreadAndKinds = SqlSchema.findAll({
10641127 Request: ThreadActivityKindsLookupInput,
10651128 Result: ProjectionThreadActivityDbRowSchema,
......@@ -1310,15 +1373,8 @@ const makeProjectionSnapshotQuery = Effect.gen(function* () {
13101373 `,
13111374 });
13121375
1313 // Blocking request payloads must remain available even if they predate the
1314 // recent activity window. Each CTE returns at most one unresolved row per
1315 // request, so the merge below stays bounded by actionable work.
1316 const listPinnedThreadActivityRowsByThread = SqlSchema.findAll({
1317 Request: ThreadIdLookupInput,
1318 Result: ProjectionThreadActivityDbRowSchema,
1319 execute: ({ threadId }) =>
1320 sql`
1321 WITH pending_approval_requests AS (
1376 const pinnedThreadActivityIdsCte = (threadId: string) => sql`
1377pending_approval_requests AS (
13221378 SELECT request_id, thread_id
13231379 FROM projection_pending_approvals
13241380 WHERE thread_id = ${threadId}
......@@ -1382,6 +1438,17 @@ const makeProjectionSnapshotQuery = Effect.gen(function* () {
13821438 WHERE request_order = 1
13831439 AND kind = 'user-input.requested'
13841440 )
1441 `;
1442
1443 // Blocking request payloads must remain available even if they predate the
1444 // recent activity window. Each CTE returns at most one unresolved row per
1445 // request, so the merge below stays bounded by actionable work.
1446 const listPinnedThreadActivityRowsByThread = SqlSchema.findAll({
1447 Request: ThreadIdLookupInput,
1448 Result: ProjectionThreadActivityDbRowSchema,
1449 execute: ({ threadId }) =>
1450 sql`
1451 WITH ${pinnedThreadActivityIdsCte(threadId)}
13851452 SELECT
13861453 activity.activity_id AS "activityId",
13871454 activity.thread_id AS "threadId",
......@@ -1399,6 +1466,17 @@ const makeProjectionSnapshotQuery = Effect.gen(function* () {
13991466 `,
14001467 });
14011468
1469 const listPinnedThreadActivityIdsByThread = SqlSchema.findAll({
1470 Request: ThreadIdLookupInput,
1471 Result: ProjectionThreadActivityIdRowSchema,
1472 execute: ({ threadId }) =>
1473 sql`
1474 WITH ${pinnedThreadActivityIdsCte(threadId)}
1475 SELECT activity_id AS "activityId"
1476 FROM pinned_activity_ids
1477 `,
1478 });
1479
14021480 const listThreadActivityRowsByThreadWindow = SqlSchema.findAll({
14031481 Request: ThreadTurnRangeLookupInput,
14041482 Result: ProjectionThreadActivityDbRowSchema,
......@@ -1466,6 +1544,48 @@ const makeProjectionSnapshotQuery = Effect.gen(function* () {
14661544 `,
14671545 });
14681546
1547 const listThreadActivityIdsByThreadWindow = SqlSchema.findAll({
1548 Request: ThreadTurnRangeLookupInput,
1549 Result: ProjectionThreadActivityIdRowSchema,
1550 execute: ({ threadId, minAnchorAt, minTurnKey, beforeAnchorAt, beforeTurnKey }) =>
1551 sql`
1552 SELECT activity_id AS "activityId"
1553 FROM projection_thread_activities
1554 WHERE thread_id = ${threadId}
1555 AND (
1556 turn_id IN (
1557 SELECT turn_id FROM projection_turns
1558 WHERE thread_id = ${threadId}
1559 AND turn_id IS NOT NULL
1560 AND (
1561 requested_at > ${minAnchorAt}
1562 OR (
1563 requested_at = ${minAnchorAt}
1564 AND turn_id >= ${minTurnKey}
1565 )
1566 )
1567 AND (
1568 requested_at < ${beforeAnchorAt}
1569 OR (
1570 requested_at = ${beforeAnchorAt}
1571 AND turn_id < ${beforeTurnKey}
1572 )
1573 )
1574 )
1575 OR (
1576 turn_id IS NULL
1577 AND created_at >= ${minAnchorAt}
1578 AND created_at < ${beforeAnchorAt}
1579 )
1580 )
1581 ORDER BY
1582 sequence DESC,
1583 created_at DESC,
1584 activity_id DESC
1585 LIMIT ${THREAD_DETAIL_ACTIVITY_LIMIT}
1586 `,
1587 });
1588
14691589 const getFullThreadDiffContextRow = SqlSchema.findOneOption({
14701590 Request: FullThreadDiffContextLookupInput,
14711591 Result: ProjectionFullThreadDiffContextRowSchema,
......@@ -2571,49 +2691,136 @@ const makeProjectionSnapshotQuery = Effect.gen(function* () {
25712691 readonly beforeTurnKey: string;
25722692 }
25732693
2574 const getThreadDetailByIdBounded = (
2575 threadId: ThreadId,
2576 bounds: ThreadDetailBounds | undefined,
2577 query: ProjectionThreadDetailQuery | undefined = undefined,
2578 ) =>
2579 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 })
2694 type ThreadDetailActivityRead =
2695 | {
2696 readonly mode: "raw";
2697 readonly query?: ProjectionThreadDetailQuery;
2698 }
2699 | {
2700 readonly mode: "client";
2701 };
2702
2703 const listProjectedThreadActivities = Effect.fn(
2704 "ProjectionSnapshotQuery.listProjectedThreadActivities",
2705 )(function* (threadId: ThreadId, bounds: ThreadDetailBounds | undefined) {
2706 const [activityIdRows, pinnedActivityIdRows] = yield* Effect.all([
2707 (bounds === undefined
2708 ? listThreadActivityIdsByThread({ threadId })
2709 : listThreadActivityIdsByThreadWindow({ threadId, ...bounds })
25912710 ).pipe(
25922711 Effect.mapError(
25932712 toPersistenceSqlOrDecodeError(
2594 "ProjectionSnapshotQuery.getThreadDetailById:listActivities:query",
2595 "ProjectionSnapshotQuery.getThreadDetailById:listActivities:decodeRows",
2713 "ProjectionSnapshotQuery.getThreadDetailById:listActivityIds:query",
2714 "ProjectionSnapshotQuery.getThreadDetailById:listActivityIds:decodeRows",
25962715 ),
25972716 ),
2717 ),
2718 listPinnedThreadActivityIdsByThread({ threadId }).pipe(
2719 Effect.mapError(
2720 toPersistenceSqlOrDecodeError(
2721 "ProjectionSnapshotQuery.getThreadDetailById:listPinnedActivityIds:query",
2722 "ProjectionSnapshotQuery.getThreadDetailById:listPinnedActivityIds:decodeRows",
2723 ),
2724 ),
2725 ),
2726 ]);
2727 const activityIds = [
2728 ...new Set([...activityIdRows, ...pinnedActivityIdRows].map(({ activityId }) => activityId)),
2729 ];
2730 const activities: OrchestrationThreadActivity[] = [];
2731
2732 for (
2733 let offset = 0;
2734 offset < activityIds.length;
2735 offset += THREAD_DETAIL_ACTIVITY_PAYLOAD_BATCH_SIZE
2736 ) {
2737 const batchIds = activityIds.slice(
2738 offset,
2739 offset + THREAD_DETAIL_ACTIVITY_PAYLOAD_BATCH_SIZE,
25982740 );
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",
2741 const batchRows = yield* listThreadActivityRowsByIds({ activityIds: batchIds }).pipe(
2742 Effect.mapError(
2743 toPersistenceSqlOrDecodeError(
2744 "ProjectionSnapshotQuery.getThreadDetailById:listActivityPayloadBatch:query",
2745 "ProjectionSnapshotQuery.getThreadDetailById:listActivityPayloadBatch:decodeRows",
2746 ),
2747 ),
2748 );
2749 for (const row of batchRows) {
2750 activities.push(projectActivityPayload(mapThreadActivityRow(row)));
2751 }
2752 }
2753
2754 return activities.toSorted(
2755 (left, right) =>
2756 (left.sequence ?? -1) - (right.sequence ?? -1) ||
2757 left.createdAt.localeCompare(right.createdAt) ||
2758 left.id.localeCompare(right.id),
2759 );
2760 });
2761
2762 const getThreadDetailByIdBounded = (
2763 threadId: ThreadId,
2764 bounds: ThreadDetailBounds | undefined,
2765 activityRead: ThreadDetailActivityRead = { mode: "raw" },
2766 ) =>
2767 Effect.gen(function* () {
2768 const activitiesEffect =
2769 activityRead.mode === "client"
2770 ? listProjectedThreadActivities(threadId, bounds)
2771 : Effect.all([
2772 (activityRead.query?.activityKinds === undefined
2773 ? bounds === undefined
2774 ? listThreadActivityRowsByThread({ threadId })
2775 : listThreadActivityRowsByThreadWindow({ threadId, ...bounds })
2776 : activityRead.query.activityKinds.length === 0
2777 ? Effect.succeed([])
2778 : listThreadActivityRowsByThreadAndKinds({
2779 threadId,
2780 activityKinds: activityRead.query.activityKinds,
2781 })
2782 ).pipe(
2783 Effect.mapError(
2784 toPersistenceSqlOrDecodeError(
2785 "ProjectionSnapshotQuery.getThreadDetailById:listActivities:query",
2786 "ProjectionSnapshotQuery.getThreadDetailById:listActivities:decodeRows",
2787 ),
26062788 ),
26072789 ),
2608 )
2609 : Effect.succeed([]);
2790 activityRead.query?.activityKinds === undefined
2791 ? listPinnedThreadActivityRowsByThread({ threadId }).pipe(
2792 Effect.mapError(
2793 toPersistenceSqlOrDecodeError(
2794 "ProjectionSnapshotQuery.getThreadDetailById:listPinnedActivities:query",
2795 "ProjectionSnapshotQuery.getThreadDetailById:listPinnedActivities:decodeRows",
2796 ),
2797 ),
2798 )
2799 : Effect.succeed([]),
2800 ]).pipe(
2801 Effect.map(([activityRows, pinnedActivityRows]) =>
2802 [
2803 ...new Map(
2804 [...activityRows, ...pinnedActivityRows].map(
2805 (row) => [row.activityId, row] as const,
2806 ),
2807 ).values(),
2808 ]
2809 .toSorted(
2810 (left, right) =>
2811 (left.sequence ?? -1) - (right.sequence ?? -1) ||
2812 left.createdAt.localeCompare(right.createdAt) ||
2813 left.activityId.localeCompare(right.activityId),
2814 )
2815 .map(mapThreadActivityRow),
2816 ),
2817 );
26102818
26112819 const [
26122820 threadRow,
26132821 messageRows,
26142822 proposedPlanRows,
2615 activityRows,
2616 pinnedActivityRows,
2823 activities,
26172824 checkpointRows,
26182825 latestTurnRow,
26192826 sessionRow,
......@@ -2645,8 +2852,7 @@ const makeProjectionSnapshotQuery = Effect.gen(function* () {
26452852 ),
26462853 ),
26472854 ),
2648 activityRowsEffect,
2649 pinnedActivityRowsEffect,
2855 activitiesEffect,
26502856 listCheckpointRowsByThread({ threadId }).pipe(
26512857 Effect.mapError(
26522858 toPersistenceSqlOrDecodeError(
......@@ -2677,17 +2883,6 @@ const makeProjectionSnapshotQuery = Effect.gen(function* () {
26772883 return Option.none<OrchestrationThread>();
26782884 }
26792885
2680 const selectedActivityRows = [
2681 ...new Map(
2682 [...activityRows, ...pinnedActivityRows].map((row) => [row.activityId, row] as const),
2683 ).values(),
2684 ].toSorted(
2685 (left, right) =>
2686 (left.sequence ?? -1) - (right.sequence ?? -1) ||
2687 left.createdAt.localeCompare(right.createdAt) ||
2688 left.activityId.localeCompare(right.activityId),
2689 );
2690
26912886 const thread = {
26922887 id: threadRow.value.threadId,
26932888 projectId: threadRow.value.projectId,
......@@ -2729,21 +2924,7 @@ const makeProjectionSnapshotQuery = Effect.gen(function* () {
27292924 return message;
27302925 }),
27312926 proposedPlans: proposedPlanRows.map(mapProposedPlanRow),
2732 activities: selectedActivityRows.map((row) => {
2733 const activity = {
2734 id: row.activityId,
2735 tone: row.tone,
2736 kind: row.kind,
2737 summary: row.summary,
2738 payload: row.payload,
2739 turnId: row.turnId,
2740 createdAt: row.createdAt,
2741 };
2742 if (row.sequence !== null) {
2743 return Object.assign(activity, { sequence: row.sequence });
2744 }
2745 return activity;
2746 }),
2927 activities,
27472928 checkpoints: checkpointRows.map((row) => ({
27482929 turnId: row.turnId,
27492930 checkpointTurnCount: row.checkpointTurnCount,
......@@ -2768,7 +2949,11 @@ const makeProjectionSnapshotQuery = Effect.gen(function* () {
27682949 const getThreadDetailById: ProjectionSnapshotQueryShape["getThreadDetailById"] = (
27692950 threadId,
27702951 query,
2771 ) => getThreadDetailByIdBounded(threadId, undefined, query);
2952 ) =>
2953 getThreadDetailByIdBounded(threadId, undefined, {
2954 mode: "raw",
2955 ...(query === undefined ? {} : { query }),
2956 });
27722957
27732958 // Bounds pathological fan-out: one user turn that spawned hundreds of
27742959 // subagent turns still pages in bounded chunks, at the cost of splitting the
......@@ -2792,7 +2977,9 @@ const makeProjectionSnapshotQuery = Effect.gen(function* () {
27922977 .withTransaction(
27932978 Effect.gen(function* () {
27942979 if (window?.turnLimit === undefined) {
2795 const thread = yield* getThreadDetailById(threadId);
2980 const thread = yield* getThreadDetailByIdBounded(threadId, undefined, {
2981 mode: "client",
2982 });
27962983 if (Option.isNone(thread)) {
27972984 return Option.none<OrchestrationThreadDetailSnapshot>();
27982985 }
......@@ -2845,7 +3032,9 @@ const makeProjectionSnapshotQuery = Effect.gen(function* () {
28453032 ? { minAnchorAt: "", minTurnKey: "", beforeAnchorAt: "", beforeTurnKey: "" }
28463033 : undefined;
28473034
2848 const thread = yield* getThreadDetailByIdBounded(threadId, emptyBounds ?? bounds);
3035 const thread = yield* getThreadDetailByIdBounded(threadId, emptyBounds ?? bounds, {
3036 mode: "client",
3037 });
28493038 if (Option.isNone(thread)) {
28503039 return Option.none<OrchestrationThreadDetailSnapshot>();
28513040 }
apps/server/src/orchestration/Services/ProjectionSnapshotQuery.ts+4
......@@ -191,6 +191,10 @@ export interface ProjectionSnapshotQueryShape {
191191 * response carries `page` metadata (see `OrchestrationThreadDetailWindow`).
192192 * Without a window the full thread is returned with no `page` field —
193193 * pagination is strictly opt-in.
194 *
195 * Activity payloads are projected for clients as they are read in small
196 * sequential batches. Callers still apply the full snapshot projector for
197 * collection-level activity pruning.
194198 */
195199 readonly getThreadDetailSnapshot: (
196200 threadId: ThreadId,