From f25c1aef7be83be8171fcb682fee1e669f0a8ce2 Mon Sep 17 00:00:00 2001 From: Ross Cawston Date: Mon, 31 Aug 2026 16:14:11 -0700 Subject: [PATCH] fix(server): bound thread activity payload hydration --- .../Layers/ProjectionSnapshotQuery.test.ts | 65 +++++++++ .../Layers/ProjectionSnapshotQuery.ts | 132 +++++++++++++----- 2 files changed, 161 insertions(+), 36 deletions(-) diff --git a/apps/server/src/orchestration/Layers/ProjectionSnapshotQuery.test.ts b/apps/server/src/orchestration/Layers/ProjectionSnapshotQuery.test.ts index 30892c760e77..58cdfa3f599b 100644 --- a/apps/server/src/orchestration/Layers/ProjectionSnapshotQuery.test.ts +++ b/apps/server/src/orchestration/Layers/ProjectionSnapshotQuery.test.ts @@ -2421,6 +2421,71 @@ projectionSnapshotLayer("ProjectionSnapshotQuery windowed thread detail", (it) = }), ); + it.effect("bounds activity hydration by serialized payload bytes", () => + Effect.gen(function* () { + yield* seedFanOutThread(); + const snapshotQuery = yield* ProjectionSnapshotQuery; + const sql = yield* SqlClient.SqlClient; + + yield* sql`DELETE FROM projection_thread_activities`; + yield* sql` + WITH RECURSIVE activity_rows(sequence) AS ( + SELECT 1 + UNION ALL + SELECT sequence + 1 FROM activity_rows WHERE sequence < 31 + ) + INSERT INTO projection_thread_activities ( + activity_id, thread_id, turn_id, tone, kind, summary, payload_json, sequence, created_at + ) + SELECT + printf('activity-%04d', sequence), + 'thread-w', + 'turn-5', + 'tool', + 'tool.completed', + 'ran tool', + json_object( + 'output', + printf('%.*c', CASE WHEN sequence = 31 THEN 700000 ELSE 400000 END, 'x') + ), + sequence, + '2026-03-01T00:04:00.000Z' + FROM activity_rows + `; + + const assertBoundedActivities = ( + activities: ReadonlyArray<{ readonly id: string; readonly payload: unknown }>, + ) => { + assert.equal(activities.length, 20); + assert.equal(activities[0]?.id, asEventId("activity-0012")); + assert.equal(activities.at(-1)?.id, asEventId("activity-0031")); + assert.deepStrictEqual(activities.at(-1)?.payload, { + t3PayloadTruncated: true, + originalBytes: 700013, + }); + const serializedPayloadBytes = activities.reduce( + (total, activity) => total + Buffer.byteLength(JSON.stringify(activity.payload)), + 0, + ); + assert.ok(serializedPayloadBytes <= 8 * 1024 * 1024); + }; + + const fullDetail = yield* snapshotQuery.getThreadDetailById(threadW); + assert.equal(fullDetail._tag, "Some"); + if (fullDetail._tag === "Some") { + assertBoundedActivities(fullDetail.value.activities); + } + + const windowedDetail = yield* snapshotQuery.getThreadDetailSnapshot(threadW, { + turnLimit: 2, + }); + assert.equal(windowedDetail._tag, "Some"); + if (windowedDetail._tag === "Some") { + assertBoundedActivities(windowedDetail.value.thread.activities); + } + }), + ); + it.effect("a thread with no turns returns its content unwindowed on the first page", () => Effect.gen(function* () { const sql = yield* SqlClient.SqlClient; diff --git a/apps/server/src/orchestration/Layers/ProjectionSnapshotQuery.ts b/apps/server/src/orchestration/Layers/ProjectionSnapshotQuery.ts index 0b9698eaf9cf..d50177058953 100644 --- a/apps/server/src/orchestration/Layers/ProjectionSnapshotQuery.ts +++ b/apps/server/src/orchestration/Layers/ProjectionSnapshotQuery.ts @@ -70,10 +70,11 @@ import { const decodeReadModel = Schema.decodeUnknownEffect(OrchestrationReadModel); const decodeShellSnapshot = Schema.decodeUnknownEffect(OrchestrationShellSnapshot); const decodeThread = Schema.decodeUnknownEffect(OrchestrationThread); -// Keep detail reads consistent with the in-memory projector's retained -// activity window. Applying the limit in SQL avoids decoding an unbounded -// payload_json set before the projector can enforce that invariant. +// Apply both row and serialized-payload limits in SQL so oversized tool output +// cannot exhaust the server heap before JSON decoding can enforce a boundary. const THREAD_DETAIL_ACTIVITY_LIMIT = 500; +const THREAD_DETAIL_ACTIVITY_PAYLOAD_LIMIT_BYTES = 512 * 1024; +const THREAD_DETAIL_ACTIVITY_PAYLOAD_BUDGET_BYTES = 8 * 1024 * 1024; const ProjectionProjectDbRowSchema = ProjectionProject.mapFields( Struct.assign({ defaultModelSelection: Schema.NullOr(Schema.fromJsonString(ModelSelection)), @@ -1019,17 +1020,7 @@ const makeProjectionSnapshotQuery = Effect.gen(function* () { Result: ProjectionThreadActivityDbRowSchema, execute: ({ threadId }) => sql` - SELECT - activity_id AS "activityId", - thread_id AS "threadId", - turn_id AS "turnId", - tone, - kind, - summary, - payload_json AS "payload", - sequence, - created_at AS "createdAt" - FROM ( + WITH recent_activities AS ( SELECT activity_id, thread_id, @@ -1042,12 +1033,48 @@ const makeProjectionSnapshotQuery = Effect.gen(function* () { created_at FROM projection_thread_activities WHERE thread_id = ${threadId} - ORDER BY - sequence DESC, - created_at DESC, - activity_id DESC + ORDER BY sequence DESC, created_at DESC, activity_id DESC LIMIT ${THREAD_DETAIL_ACTIVITY_LIMIT} - ) AS recent_activities + ), + ordered_activities AS ( + SELECT + *, + ROW_NUMBER() OVER ( + ORDER BY sequence DESC, created_at DESC, activity_id DESC + ) AS recent_order, + MIN( + length(payload_json), + ${THREAD_DETAIL_ACTIVITY_PAYLOAD_LIMIT_BYTES} + ) AS hydrated_payload_bytes + FROM recent_activities + ), + budgeted_activities AS ( + SELECT + *, + SUM(hydrated_payload_bytes) OVER ( + ORDER BY recent_order ASC + ) AS cumulative_payload_bytes + FROM ordered_activities + ) + SELECT + activity_id AS "activityId", + thread_id AS "threadId", + turn_id AS "turnId", + tone, + kind, + summary, + CASE + WHEN length(payload_json) > ${THREAD_DETAIL_ACTIVITY_PAYLOAD_LIMIT_BYTES} + THEN json_object( + 't3PayloadTruncated', json('true'), + 'originalBytes', length(payload_json) + ) + ELSE payload_json + END AS "payload", + sequence, + created_at AS "createdAt" + FROM budgeted_activities + WHERE cumulative_payload_bytes <= ${THREAD_DETAIL_ACTIVITY_PAYLOAD_BUDGET_BYTES} ORDER BY sequence ASC, created_at ASC, @@ -1342,7 +1369,14 @@ const makeProjectionSnapshotQuery = Effect.gen(function* () { activity.tone, activity.kind, activity.summary, - activity.payload_json AS "payload", + CASE + WHEN length(activity.payload_json) > ${THREAD_DETAIL_ACTIVITY_PAYLOAD_LIMIT_BYTES} + THEN json_object( + 't3PayloadTruncated', json('true'), + 'originalBytes', length(activity.payload_json) + ) + ELSE activity.payload_json + END AS "payload", activity.sequence, activity.created_at AS "createdAt" FROM pinned_activity_ids AS pinned @@ -1357,17 +1391,7 @@ const makeProjectionSnapshotQuery = Effect.gen(function* () { Result: ProjectionThreadActivityDbRowSchema, execute: ({ threadId, minAnchorAt, minTurnKey, beforeAnchorAt, beforeTurnKey }) => sql` - SELECT - activity_id AS "activityId", - thread_id AS "threadId", - turn_id AS "turnId", - tone, - kind, - summary, - payload_json AS "payload", - sequence, - created_at AS "createdAt" - FROM ( + WITH recent_activities AS ( SELECT activity_id, thread_id, @@ -1406,12 +1430,48 @@ const makeProjectionSnapshotQuery = Effect.gen(function* () { AND created_at < ${beforeAnchorAt} ) ) - ORDER BY - sequence DESC, - created_at DESC, - activity_id DESC + ORDER BY sequence DESC, created_at DESC, activity_id DESC LIMIT ${THREAD_DETAIL_ACTIVITY_LIMIT} - ) AS recent_activities + ), + ordered_activities AS ( + SELECT + *, + ROW_NUMBER() OVER ( + ORDER BY sequence DESC, created_at DESC, activity_id DESC + ) AS recent_order, + MIN( + length(payload_json), + ${THREAD_DETAIL_ACTIVITY_PAYLOAD_LIMIT_BYTES} + ) AS hydrated_payload_bytes + FROM recent_activities + ), + budgeted_activities AS ( + SELECT + *, + SUM(hydrated_payload_bytes) OVER ( + ORDER BY recent_order ASC + ) AS cumulative_payload_bytes + FROM ordered_activities + ) + SELECT + activity_id AS "activityId", + thread_id AS "threadId", + turn_id AS "turnId", + tone, + kind, + summary, + CASE + WHEN length(payload_json) > ${THREAD_DETAIL_ACTIVITY_PAYLOAD_LIMIT_BYTES} + THEN json_object( + 't3PayloadTruncated', json('true'), + 'originalBytes', length(payload_json) + ) + ELSE payload_json + END AS "payload", + sequence, + created_at AS "createdAt" + FROM budgeted_activities + WHERE cumulative_payload_bytes <= ${THREAD_DETAIL_ACTIVITY_PAYLOAD_BUDGET_BYTES} ORDER BY sequence ASC, created_at ASC,