authorgravatar for 269035359+t3-code[bot]@users.noreply.github.comt3-code[bot] <269035359+t3-code[bot]@users.noreply.github.com> 2026-08-15 12:44:03+02:00
committergravatar for noreply@github.comGitHub <noreply@github.com> 2026-08-15 12:44:03+02:00
log71c6f8248775066ebaf4bfc6680d3e2acb4bb2d1
tree142756263006fd85e7ef1129e349d2d817b2a8ba
parenteaa6c4712fe11f0396e549b1873f163dc202d229
signature Signed by PGP key B5690EEEBB952194

fix(server): bound thread activity hydration (#6153)

Co-authored-by: t3-code[bot] <269035359+t3-code[bot]@users.noreply.github.com>

2 files changed, 299 insertions(+), 28 deletions(-)

apps/server/src/orchestration/Layers/ProjectionSnapshotQuery.test.ts+124
......@@ -2281,6 +2281,130 @@ projectionSnapshotLayer("ProjectionSnapshotQuery windowed thread detail", (it) =
22812281 }),
22822282 );
22832283
2284 it.effect("bounds activity hydration and preserves unresolved requests", () =>
2285 Effect.gen(function* () {
2286 yield* seedFanOutThread();
2287 const snapshotQuery = yield* ProjectionSnapshotQuery;
2288 const sql = yield* SqlClient.SqlClient;
2289
2290 yield* sql`DELETE FROM projection_thread_activities`;
2291 yield* sql`
2292 WITH RECURSIVE activity_rows(sequence) AS (
2293 SELECT 1
2294 UNION ALL
2295 SELECT sequence + 1 FROM activity_rows WHERE sequence < 501
2296 )
2297 INSERT INTO projection_thread_activities (
2298 activity_id, thread_id, turn_id, tone, kind, summary, payload_json, sequence, created_at
2299 )
2300 SELECT
2301 printf('activity-%04d', sequence),
2302 'thread-w',
2303 'turn-5',
2304 'tool',
2305 'tool.completed',
2306 'ran tool',
2307 printf('{"sequence":%d}', sequence),
2308 sequence,
2309 '2026-03-01T00:04:00.000Z'
2310 FROM activity_rows
2311 `;
2312
2313 const fullDetail = yield* snapshotQuery.getThreadDetailById(threadW);
2314 assert.equal(fullDetail._tag, "Some");
2315 if (fullDetail._tag === "Some") {
2316 assert.equal(fullDetail.value.activities.length, 500);
2317 assert.equal(fullDetail.value.activities[0]?.id, asEventId("activity-0002"));
2318 assert.equal(fullDetail.value.activities.at(-1)?.id, asEventId("activity-0501"));
2319 }
2320
2321 const windowedDetail = yield* snapshotQuery.getThreadDetailSnapshot(threadW, {
2322 turnLimit: 2,
2323 });
2324 assert.equal(windowedDetail._tag, "Some");
2325 if (windowedDetail._tag === "Some") {
2326 assert.equal(windowedDetail.value.thread.activities.length, 500);
2327 assert.equal(windowedDetail.value.thread.activities[0]?.id, asEventId("activity-0002"));
2328 assert.equal(windowedDetail.value.thread.activities.at(-1)?.id, asEventId("activity-0501"));
2329 }
2330
2331 yield* sql`
2332 INSERT INTO projection_thread_activities (
2333 activity_id, thread_id, turn_id, tone, kind, summary, payload_json, sequence, created_at
2334 )
2335 VALUES
2336 (
2337 'approval-old', 'thread-w', NULL, 'approval', 'approval.requested',
2338 'Approve old command', '{"requestId":"approval-1"}', NULL,
2339 '2026-03-01T00:00:01.000Z'
2340 ),
2341 (
2342 'user-input-old', 'thread-w', NULL, 'approval', 'user-input.requested',
2343 'Answer old question', '{"requestId":"input-1"}', NULL,
2344 '2026-03-01T00:00:02.000Z'
2345 ),
2346 (
2347 'user-input-closed', 'thread-w', NULL, 'approval', 'user-input.requested',
2348 'Closed question', '{"requestId":"input-closed"}', NULL,
2349 '2026-03-01T00:00:03.000Z'
2350 ),
2351 (
2352 'user-input-closed-resolution', 'thread-w', NULL, 'info', 'user-input.resolved',
2353 'Closed question', '{"requestId":"input-closed"}', NULL,
2354 '2026-03-01T00:00:04.000Z'
2355 ),
2356 (
2357 'user-input-tied-z-request', 'thread-w', NULL, 'approval', 'user-input.requested',
2358 'Tied open question', '{"requestId":"input-tied-open"}', NULL,
2359 '2026-03-01T00:00:05.000Z'
2360 ),
2361 (
2362 'user-input-tied-a-resolution', 'thread-w', NULL, 'info', 'user-input.resolved',
2363 'Tied open question', '{"requestId":"input-tied-open"}', NULL,
2364 '2026-03-01T00:00:05.000Z'
2365 )
2366 `;
2367 yield* sql`
2368 INSERT INTO projection_pending_approvals (
2369 request_id, thread_id, turn_id, status, decision, created_at, resolved_at
2370 )
2371 VALUES (
2372 'approval-1', 'thread-w', NULL, 'pending', NULL,
2373 '2026-03-01T00:00:01.000Z', NULL
2374 )
2375 `;
2376 yield* sql`
2377 UPDATE projection_threads
2378 SET pending_approval_count = 1, pending_user_input_count = 1
2379 WHERE thread_id = 'thread-w'
2380 `;
2381
2382 const detailWithPinnedRequests = yield* snapshotQuery.getThreadDetailById(threadW);
2383 assert.equal(detailWithPinnedRequests._tag, "Some");
2384 if (detailWithPinnedRequests._tag === "Some") {
2385 const ids = detailWithPinnedRequests.value.activities.map((activity) => activity.id);
2386 assert.equal(detailWithPinnedRequests.value.activities.length, 503);
2387 assert.equal(ids.includes(asEventId("approval-old")), true);
2388 assert.equal(ids.includes(asEventId("user-input-old")), true);
2389 assert.equal(ids.includes(asEventId("user-input-closed")), false);
2390 assert.equal(ids.includes(asEventId("user-input-tied-z-request")), true);
2391 }
2392
2393 const windowWithPinnedRequests = yield* snapshotQuery.getThreadDetailSnapshot(threadW, {
2394 turnLimit: 2,
2395 });
2396 assert.equal(windowWithPinnedRequests._tag, "Some");
2397 if (windowWithPinnedRequests._tag === "Some") {
2398 const ids = windowWithPinnedRequests.value.thread.activities.map((activity) => activity.id);
2399 assert.equal(windowWithPinnedRequests.value.thread.activities.length, 503);
2400 assert.equal(ids.includes(asEventId("approval-old")), true);
2401 assert.equal(ids.includes(asEventId("user-input-old")), true);
2402 assert.equal(ids.includes(asEventId("user-input-closed")), false);
2403 assert.equal(ids.includes(asEventId("user-input-tied-z-request")), true);
2404 }
2405 }),
2406 );
2407
22842408 it.effect("a thread with no turns returns its content unwindowed on the first page", () =>
22852409 Effect.gen(function* () {
22862410 const sql = yield* SqlClient.SqlClient;
apps/server/src/orchestration/Layers/ProjectionSnapshotQuery.ts+175-28
......@@ -69,6 +69,10 @@ import {
6969const decodeReadModel = Schema.decodeUnknownEffect(OrchestrationReadModel);
7070const decodeShellSnapshot = Schema.decodeUnknownEffect(OrchestrationShellSnapshot);
7171const decodeThread = Schema.decodeUnknownEffect(OrchestrationThread);
72// Keep detail reads consistent with the in-memory projector's retained
73// activity window. Applying the limit in SQL avoids decoding an unbounded
74// payload_json set before the projector can enforce that invariant.
75const THREAD_DETAIL_ACTIVITY_LIMIT = 500;
7276const ProjectionProjectDbRowSchema = ProjectionProject.mapFields(
7377 Struct.assign({
7478 defaultModelSelection: Schema.NullOr(Schema.fromJsonString(ModelSelection)),
......@@ -1015,8 +1019,25 @@ const makeProjectionSnapshotQuery = Effect.gen(function* () {
10151019 payload_json AS "payload",
10161020 sequence,
10171021 created_at AS "createdAt"
1018 FROM projection_thread_activities
1019 WHERE thread_id = ${threadId}
1022 FROM (
1023 SELECT
1024 activity_id,
1025 thread_id,
1026 turn_id,
1027 tone,
1028 kind,
1029 summary,
1030 payload_json,
1031 sequence,
1032 created_at
1033 FROM projection_thread_activities
1034 WHERE thread_id = ${threadId}
1035 ORDER BY
1036 sequence DESC,
1037 created_at DESC,
1038 activity_id DESC
1039 LIMIT ${THREAD_DETAIL_ACTIVITY_LIMIT}
1040 ) AS recent_activities
10201041 ORDER BY
10211042 sequence ASC,
10221043 created_at ASC,
......@@ -1232,6 +1253,95 @@ const makeProjectionSnapshotQuery = Effect.gen(function* () {
12321253 `,
12331254 });
12341255
1256 // Blocking request payloads must remain available even if they predate the
1257 // recent activity window. Each CTE returns at most one unresolved row per
1258 // request, so the merge below stays bounded by actionable work.
1259 const listPinnedThreadActivityRowsByThread = SqlSchema.findAll({
1260 Request: ThreadIdLookupInput,
1261 Result: ProjectionThreadActivityDbRowSchema,
1262 execute: ({ threadId }) =>
1263 sql`
1264 WITH pending_approval_requests AS (
1265 SELECT request_id, thread_id
1266 FROM projection_pending_approvals
1267 WHERE thread_id = ${threadId}
1268 AND status = 'pending'
1269 ),
1270 pending_approval_activities AS (
1271 SELECT
1272 activity.activity_id,
1273 ROW_NUMBER() OVER (
1274 PARTITION BY pending.request_id
1275 ORDER BY activity.created_at DESC, activity.activity_id DESC
1276 ) AS request_order
1277 FROM pending_approval_requests AS pending
1278 CROSS JOIN projection_thread_activities AS activity
1279 WHERE activity.thread_id = pending.thread_id
1280 AND activity.kind = 'approval.requested'
1281 AND json_extract(activity.payload_json, '$.requestId') = pending.request_id
1282 ),
1283 pending_user_input_thread AS (
1284 SELECT thread_id
1285 FROM projection_threads
1286 WHERE thread_id = ${threadId}
1287 AND pending_user_input_count > 0
1288 ),
1289 user_input_lifecycle AS (
1290 SELECT
1291 activity.activity_id,
1292 activity.kind,
1293 ROW_NUMBER() OVER (
1294 PARTITION BY json_extract(activity.payload_json, '$.requestId')
1295 ORDER BY activity.created_at DESC, activity.activity_id DESC
1296 ) AS request_order
1297 FROM pending_user_input_thread AS pending
1298 CROSS JOIN projection_thread_activities AS activity
1299 WHERE activity.thread_id = pending.thread_id
1300 AND (
1301 activity.kind IN ('user-input.requested', 'user-input.resolved')
1302 OR (
1303 activity.kind = 'provider.user-input.respond.failed'
1304 AND (
1305 lower(COALESCE(json_extract(activity.payload_json, '$.detail'), ''))
1306 LIKE '%stale pending user-input request%'
1307 OR lower(COALESCE(json_extract(activity.payload_json, '$.detail'), ''))
1308 LIKE '%unknown pending user-input request%'
1309 OR lower(COALESCE(json_extract(activity.payload_json, '$.detail'), ''))
1310 LIKE '%unknown pending user input request%'
1311 OR lower(COALESCE(json_extract(activity.payload_json, '$.detail'), ''))
1312 LIKE '%unknown pending codex user input request%'
1313 )
1314 )
1315 )
1316 AND json_extract(activity.payload_json, '$.requestId') IS NOT NULL
1317 ),
1318 pinned_activity_ids AS (
1319 SELECT activity_id
1320 FROM pending_approval_activities
1321 WHERE request_order = 1
1322 UNION ALL
1323 SELECT activity_id
1324 FROM user_input_lifecycle
1325 WHERE request_order = 1
1326 AND kind = 'user-input.requested'
1327 )
1328 SELECT
1329 activity.activity_id AS "activityId",
1330 activity.thread_id AS "threadId",
1331 activity.turn_id AS "turnId",
1332 activity.tone,
1333 activity.kind,
1334 activity.summary,
1335 activity.payload_json AS "payload",
1336 activity.sequence,
1337 activity.created_at AS "createdAt"
1338 FROM pinned_activity_ids AS pinned
1339 INNER JOIN projection_thread_activities AS activity
1340 ON activity.activity_id = pinned.activity_id
1341 ORDER BY activity.created_at ASC, activity.activity_id ASC
1342 `,
1343 });
1344
12351345 const listThreadActivityRowsByThreadWindow = SqlSchema.findAll({
12361346 Request: ThreadTurnRangeLookupInput,
12371347 Result: ProjectionThreadActivityDbRowSchema,
......@@ -1247,34 +1357,51 @@ const makeProjectionSnapshotQuery = Effect.gen(function* () {
12471357 payload_json AS "payload",
12481358 sequence,
12491359 created_at AS "createdAt"
1250 FROM projection_thread_activities
1251 WHERE thread_id = ${threadId}
1252 AND (
1253 turn_id IN (
1254 SELECT turn_id FROM projection_turns
1255 WHERE thread_id = ${threadId}
1256 AND turn_id IS NOT NULL
1257 AND (
1258 requested_at > ${minAnchorAt}
1259 OR (
1260 requested_at = ${minAnchorAt}
1261 AND turn_id >= ${minTurnKey}
1360 FROM (
1361 SELECT
1362 activity_id,
1363 thread_id,
1364 turn_id,
1365 tone,
1366 kind,
1367 summary,
1368 payload_json,
1369 sequence,
1370 created_at
1371 FROM projection_thread_activities
1372 WHERE thread_id = ${threadId}
1373 AND (
1374 turn_id IN (
1375 SELECT turn_id FROM projection_turns
1376 WHERE thread_id = ${threadId}
1377 AND turn_id IS NOT NULL
1378 AND (
1379 requested_at > ${minAnchorAt}
1380 OR (
1381 requested_at = ${minAnchorAt}
1382 AND turn_id >= ${minTurnKey}
1383 )
12621384 )
1263 )
1264 AND (
1265 requested_at < ${beforeAnchorAt}
1266 OR (
1267 requested_at = ${beforeAnchorAt}
1268 AND turn_id < ${beforeTurnKey}
1385 AND (
1386 requested_at < ${beforeAnchorAt}
1387 OR (
1388 requested_at = ${beforeAnchorAt}
1389 AND turn_id < ${beforeTurnKey}
1390 )
12691391 )
1270 )
1271 )
1272 OR (
1273 turn_id IS NULL
1274 AND created_at >= ${minAnchorAt}
1275 AND created_at < ${beforeAnchorAt}
1392 )
1393 OR (
1394 turn_id IS NULL
1395 AND created_at >= ${minAnchorAt}
1396 AND created_at < ${beforeAnchorAt}
1397 )
12761398 )
1277 )
1399 ORDER BY
1400 sequence DESC,
1401 created_at DESC,
1402 activity_id DESC
1403 LIMIT ${THREAD_DETAIL_ACTIVITY_LIMIT}
1404 ) AS recent_activities
12781405 ORDER BY
12791406 sequence ASC,
12801407 created_at ASC,
......@@ -2374,6 +2501,7 @@ const makeProjectionSnapshotQuery = Effect.gen(function* () {
23742501 messageRows,
23752502 proposedPlanRows,
23762503 activityRows,
2504 pinnedActivityRows,
23772505 checkpointRows,
23782506 latestTurnRow,
23792507 sessionRow,
......@@ -2416,6 +2544,14 @@ const makeProjectionSnapshotQuery = Effect.gen(function* () {
24162544 ),
24172545 ),
24182546 ),
2547 listPinnedThreadActivityRowsByThread({ threadId }).pipe(
2548 Effect.mapError(
2549 toPersistenceSqlOrDecodeError(
2550 "ProjectionSnapshotQuery.getThreadDetailById:listPinnedActivities:query",
2551 "ProjectionSnapshotQuery.getThreadDetailById:listPinnedActivities:decodeRows",
2552 ),
2553 ),
2554 ),
24192555 listCheckpointRowsByThread({ threadId }).pipe(
24202556 Effect.mapError(
24212557 toPersistenceSqlOrDecodeError(
......@@ -2446,6 +2582,17 @@ const makeProjectionSnapshotQuery = Effect.gen(function* () {
24462582 return Option.none<OrchestrationThread>();
24472583 }
24482584
2585 const selectedActivityRows = [
2586 ...new Map(
2587 [...activityRows, ...pinnedActivityRows].map((row) => [row.activityId, row] as const),
2588 ).values(),
2589 ].toSorted(
2590 (left, right) =>
2591 (left.sequence ?? -1) - (right.sequence ?? -1) ||
2592 left.createdAt.localeCompare(right.createdAt) ||
2593 left.activityId.localeCompare(right.activityId),
2594 );
2595
24492596 const thread = {
24502597 id: threadRow.value.threadId,
24512598 projectId: threadRow.value.projectId,
......@@ -2483,7 +2630,7 @@ const makeProjectionSnapshotQuery = Effect.gen(function* () {
24832630 return message;
24842631 }),
24852632 proposedPlans: proposedPlanRows.map(mapProposedPlanRow),
2486 activities: activityRows.map((row) => {
2633 activities: selectedActivityRows.map((row) => {
24872634 const activity = {
24882635 id: row.activityId,
24892636 tone: row.tone,