From 70ed15fa99cad8b37d8bbb8f754efbf37ea0dc61 Mon Sep 17 00:00:00 2001 From: Michael Yong Date: Wed, 26 Aug 2026 13:49:06 -0700 Subject: [PATCH 1/2] Page timeline summaries at logical boundaries --- apps/server/src/services/threads/timeline.ts | 55 ++++-- .../test/public/public-thread-data.test.ts | 12 +- ...lic-thread-timeline-output-preview.test.ts | 8 +- .../threads/timeline-in-turn-window.test.ts | 174 +++++++++++++++++- .../timeline-workflow-progress-window.test.ts | 4 +- .../thread-view/src/build-thread-timeline.ts | 28 ++- .../src/completed-turn-grouping.ts | 49 ++--- .../test/completed-turn-grouping.test.ts | 22 +-- 8 files changed, 280 insertions(+), 72 deletions(-) diff --git a/apps/server/src/services/threads/timeline.ts b/apps/server/src/services/threads/timeline.ts index 8367e4d352..9cac006a7e 100644 --- a/apps/server/src/services/threads/timeline.ts +++ b/apps/server/src/services/threads/timeline.ts @@ -1350,6 +1350,12 @@ function resolveTimelineSegmentWindow( segmentLimit: page.segmentLimit, threadId, }); + // Rows before the first user-message anchor are the thread prelude. When + // this page reaches the oldest anchor, include that prelude as one extra + // logical segment instead of leaving it unreachable behind no cursor. + const includesThreadPrelude = + bounds.sequenceWindowStart === null && + precedingAnchors.length <= page.segmentLimit; return { // Every cursor names the first sequence the page that issued it covered, // so this page ends exactly there. Reading up to the *next anchor* past @@ -1362,13 +1368,15 @@ function resolveTimelineSegmentWindow( sequenceCursor?.kind === "byte" ? bounds.sequenceStart : null, requiresWholeItemClosure: sequenceCursor !== null || bounds.sequenceWindowStart !== null, - effectiveSegmentLimit: bounds.effectiveSegmentLimit, + effectiveSegmentLimit: includesThreadPrelude + ? bounds.effectiveSegmentLimit + 1 + : bounds.effectiveSegmentLimit, hasAnchors: true, sequenceWindowStart: bounds.sequenceWindowStart, knownHasOlderSegments: precedingAnchors.length > bounds.affordableAnchorCount, oversizedEventPlaceholder: null, - sequenceStart: bounds.sequenceStart, + sequenceStart: includesThreadPrelude ? 0 : bounds.sequenceStart, }; } @@ -1389,18 +1397,23 @@ function resolveTimelineSegmentWindow( segmentLimit: page.segmentLimit, threadId, }); + const includesThreadPrelude = + bounds.sequenceWindowStart === null && + newestAnchors.length <= page.segmentLimit; return { beforeSequence: undefined, byteWindowSequenceStart: null, requiresWholeItemClosure: bounds.sequenceWindowStart !== null, - effectiveSegmentLimit: bounds.effectiveSegmentLimit, + effectiveSegmentLimit: includesThreadPrelude + ? bounds.effectiveSegmentLimit + 1 + : bounds.effectiveSegmentLimit, hasAnchors: true, sequenceWindowStart: bounds.sequenceWindowStart, // Budgeted windows read exactly the segments they return, so "is there // more" comes from the anchor list rather than an over-read segment. knownHasOlderSegments: newestAnchors.length > bounds.affordableAnchorCount, oversizedEventPlaceholder: null, - sequenceStart: bounds.sequenceStart, + sequenceStart: includesThreadPrelude ? 0 : bounds.sequenceStart, }; } @@ -1410,16 +1423,20 @@ function selectStandardTimelineEventRows( page: ThreadTimelinePageRequest, eventBudget: number, maxInlineOutputChars: InlineOutputCharLimit, + enforceByteBudget = true, ): TimelineEventRowSelection { - const window = applyTimelineWindowByteBudget(db, { - maxInlineOutputChars, + const segmentWindow = resolveTimelineSegmentWindow(db, { + eventBudget, + page, threadId: thread.id, - window: resolveTimelineSegmentWindow(db, { - eventBudget, - page, - threadId: thread.id, - }), }); + const window = enforceByteBudget + ? applyTimelineWindowByteBudget(db, { + maxInlineOutputChars, + threadId: thread.id, + window: segmentWindow, + }) + : segmentWindow; if ( !window.hasAnchors && window.sequenceWindowStart === null && @@ -1666,6 +1683,12 @@ function buildThreadTimelineInternal( ? createThreadTimelineBuildProfileAccumulator() : null; const includeNestedRows = options.includeNestedRows ?? false; + // A default timeline page must never begin at a raw event/byte cut: semantic + // rows can span those cuts, leaving a row as context on one page and outside + // the next. Page the summary resource only at its user-message anchors. + // Nested consumers keep the bounded legacy event windows; completed-turn + // expansion has its own paginated resource. + const useTransportWindows = includeNestedRows; const includeProviderUnhandledOperations = options.includeProviderUnhandledOperations; const eventSelection = measureThreadTimelineStage( @@ -1676,8 +1699,11 @@ function buildThreadTimelineInternal( db, thread, options.page, - options.eventBudget, - options.maxInlineOutputChars, + useTransportWindows ? options.eventBudget : Number.MAX_SAFE_INTEGER, + useTransportWindows || thread.status !== "idle" + ? options.maxInlineOutputChars + : 0, + useTransportWindows, ), ); const rawEventRows = eventSelection.rows; @@ -2127,8 +2153,7 @@ function buildTimelineTurnSummaryDetailsRange( // route actually holds, so the parent expansion spends what is left rather // than a pre-closure estimate of it. The subtraction may go negative, which // is the safe direction: the parent fetch then stays inside its bounds. - const detailsEventDataBytes = - byteLengthOfStoredEventRows(wholeItemEventRows); + const detailsEventDataBytes = byteLengthOfStoredEventRows(wholeItemEventRows); const eventRowsWithParentedChildren = ensureTimelineWindowParentedRows(db, { maxInlineOutputChars: detailsInlineOutputLimit, outOfBoundsChildDataByteLimit: diff --git a/apps/server/test/public/public-thread-data.test.ts b/apps/server/test/public/public-thread-data.test.ts index 83817f94a1..62e45baf2a 100644 --- a/apps/server/test/public/public-thread-data.test.ts +++ b/apps/server/test/public/public-thread-data.test.ts @@ -1668,7 +1668,7 @@ describe("public thread data routes", () => { }); it( - "expands the newest slice when a large delegation parent completes last", + "expands the canonical summary when a large delegation parent completes last", async () => { await withTestHarness(async (harness) => { const { environment, thread } = seedThreadFixture(harness); @@ -1803,12 +1803,18 @@ describe("public thread data routes", () => { if (!turnRow) { throw new Error("Expected a turn row"); } - expect(turnRow.sourceSeqStart).toBeGreaterThan(2); + expect(turnRow.sourceSeqStart).toBe(1); + expect(turnRow.sourceSeqEnd).toBe(sequence); const detailsResponse = await harness.app.request( - `/api/v1/threads/${thread.id}/timeline/turn-summary-details?turnId=${turnRow.turnId}&sourceSeqStart=${turnRow.sourceSeqStart}&sourceSeqEnd=${turnRow.sourceSeqEnd}`, + `/api/v1/threads/${thread.id}/timeline/turn-details?turnId=${turnRow.turnId}&sourceSeqStart=${turnRow.sourceSeqStart}&sourceSeqEnd=${turnRow.sourceSeqEnd}`, ); expect(detailsResponse.status).toBe(200); + const details = timelineTurnDetailsResponseSchema.parse( + await readJson(detailsResponse), + ); + expect(details.rows.length).toBeGreaterThan(0); + expect(details.nextCursor).not.toBeNull(); }); }, 10_000, diff --git a/apps/server/test/public/public-thread-timeline-output-preview.test.ts b/apps/server/test/public/public-thread-timeline-output-preview.test.ts index d1a6184782..f8bc95b6c3 100644 --- a/apps/server/test/public/public-thread-timeline-output-preview.test.ts +++ b/apps/server/test/public/public-thread-timeline-output-preview.test.ts @@ -47,7 +47,9 @@ function findCommandRow(rows: readonly TimelineRow[], command: string) { function seedRunningTurnWithCommands(harness: TestAppHarness): { threadId: string; } { - const { environment, thread } = seedThreadFixture(harness); + const { environment, thread } = seedThreadFixture(harness, { + thread: { status: "active" }, + }); const turn = { threadId: thread.id, environmentId: environment.id, @@ -201,7 +203,9 @@ describe("GET /threads/:id/timeline inline output preview", () => { describe("GET /threads/:id/timeline inline output preview (tool rows)", () => { it("previews a large tool result and row-scoped details return it whole", async () => { await withTestHarness(async (harness) => { - const { environment, thread } = seedThreadFixture(harness); + const { environment, thread } = seedThreadFixture(harness, { + thread: { status: "active" }, + }); const turn = { threadId: thread.id, environmentId: environment.id, diff --git a/apps/server/test/services/threads/timeline-in-turn-window.test.ts b/apps/server/test/services/threads/timeline-in-turn-window.test.ts index 76f7a8f793..775d42a46b 100644 --- a/apps/server/test/services/threads/timeline-in-turn-window.test.ts +++ b/apps/server/test/services/threads/timeline-in-turn-window.test.ts @@ -454,7 +454,7 @@ function buildPage( return buildThreadTimelineWithProfile(db, thread, { eventBudget, includeProviderUnhandledOperations: false, - includeNestedRows: false, + includeNestedRows: true, maxInlineOutputChars: 32_000, maxSeq: 0, page: cursor @@ -463,6 +463,25 @@ function buildPage( }); } +function buildSummaryPage( + db: DbConnection, + thread: Thread, + eventBudget: number, + cursor: TimelinePaginationCursor | null = null, + segmentLimit = 20, +) { + return buildThreadTimelineWithProfile(db, thread, { + eventBudget, + includeProviderUnhandledOperations: false, + includeNestedRows: false, + maxInlineOutputChars: 32_000, + maxSeq: 0, + page: cursor + ? { kind: "older", beforeCursor: cursor, segmentLimit } + : { kind: "latest", segmentLimit }, + }); +} + function buildNestedPage( db: DbConnection, thread: Thread, @@ -543,6 +562,159 @@ function walkAllPages( } describe("in-turn timeline windows", () => { + it("keeps summary rows independent of event and byte transport budgets", () => { + const { db, thread } = setup(); + seedTurns(db, thread, { + commandChars: 25_000, + completeLastTurn: true, + itemsPerTurn: [BYTE_WINDOW_ITEM_COUNT], + }); + + const budgeted = buildSummaryPage(db, thread, 100); + const unbudgeted = buildSummaryPage(db, thread, LARGE_BUDGET); + + expect(budgeted.profile.eventRowCount).toBeGreaterThan(100); + expect(budgeted.profile.eventDataBytes).toBeGreaterThan( + THREAD_TIMELINE_EVENT_DATA_BYTE_LIMIT, + ); + expect(budgeted.response.timelinePage.hasOlderRows).toBe(false); + expect(budgeted.response.rows).toEqual(unbudgeted.response.rows); + }); + + it("recombines logical summary pages into the unpaginated projection", () => { + const { db, thread } = setup(); + seedTurns(db, thread, { + commandChars: 25_000, + completeLastTurn: true, + itemsPerTurn: [1, 1, BYTE_WINDOW_ITEM_COUNT, 1, 1], + }); + + const rowsByPage: string[][] = []; + let cursor: TimelinePaginationCursor | null = null; + let maxEventDataBytes = 0; + let maxEventRowCount = 0; + let pages = 0; + for (;;) { + const { profile, response } = buildSummaryPage( + db, + thread, + 100, + cursor, + 2, + ); + pages += 1; + maxEventDataBytes = Math.max(maxEventDataBytes, profile.eventDataBytes); + maxEventRowCount = Math.max(maxEventRowCount, profile.eventRowCount); + rowsByPage.push(response.rows.map((row) => JSON.stringify(row))); + if (!response.timelinePage.hasOlderRows) break; + cursor = response.timelinePage.olderCursor; + expect(cursor).not.toBeNull(); + expect(pages).toBeLessThan(10); + } + + const unpaginated = buildSummaryPage( + db, + thread, + LARGE_BUDGET, + null, + LARGE_BUDGET, + ); + expect(pages).toBe(3); + expect(maxEventRowCount).toBeGreaterThan(100); + expect(maxEventDataBytes).toBeGreaterThan( + THREAD_TIMELINE_EVENT_DATA_BYTE_LIMIT, + ); + expect(rowsByPage.reverse().flat()).toEqual( + unpaginated.response.rows.map((row) => JSON.stringify(row)), + ); + }); + + it("keeps assistant prelude rows before the first user anchor", () => { + const { db, thread } = setup(); + const firstRequestId = requestId(1); + insertEvents(db, noopNotifier, [ + { + threadId: thread.id, + sequence: 1, + type: "turn/started", + scope: turnScope("prelude-turn"), + providerThreadId, + itemId: null, + itemKind: null, + parentToolCallId: null, + data: JSON.stringify({}), + }, + { + threadId: thread.id, + sequence: 2, + type: "item/completed", + scope: turnScope("prelude-turn"), + providerThreadId, + itemId: "prelude-message", + itemKind: "agentMessage", + parentToolCallId: null, + data: JSON.stringify({ + item: { + type: "agentMessage", + id: "prelude-message", + text: "Assistant greeting before the first prompt.", + }, + }), + }, + { + threadId: thread.id, + sequence: 3, + type: "turn/completed", + scope: turnScope("prelude-turn"), + providerThreadId, + itemId: null, + itemKind: null, + parentToolCallId: null, + data: JSON.stringify({ status: "completed", providerThreadId }), + }, + { + threadId: thread.id, + sequence: 4, + type: "client/turn/requested", + scope: threadScope(), + itemId: null, + itemKind: null, + parentToolCallId: null, + data: JSON.stringify({ + direction: "outbound", + source: "tell", + initiator: "user", + request: { method: "turn/start", params: {} }, + requestId: firstRequestId, + senderThreadId: null, + input: [{ type: "text", text: "First prompt", mentions: [] }], + target: { kind: "thread-start" }, + execution, + }), + }, + ]); + + const { response } = buildSummaryPage(db, thread, 100); + expect(response.timelinePage.hasOlderRows).toBe(false); + expect(response.rows).toEqual( + expect.arrayContaining([ + expect.objectContaining({ + kind: "conversation", + role: "assistant", + text: "Assistant greeting before the first prompt.", + }), + expect.objectContaining({ + kind: "conversation", + role: "user", + text: "First prompt", + }), + ]), + ); + expect(response.rows[0]).toEqual( + expect.objectContaining({ role: "assistant" }), + ); + }); + const expectSteerDetailsOwnership = ( steerStatus: "accepted" | "rejected", ): void => { diff --git a/apps/server/test/services/threads/timeline-workflow-progress-window.test.ts b/apps/server/test/services/threads/timeline-workflow-progress-window.test.ts index e5b703d30f..7f384d9fad 100644 --- a/apps/server/test/services/threads/timeline-workflow-progress-window.test.ts +++ b/apps/server/test/services/threads/timeline-workflow-progress-window.test.ts @@ -305,7 +305,9 @@ function buildPage( return buildThreadTimelineWithProfile(db, thread, { eventBudget, includeProviderUnhandledOperations: false, - includeNestedRows: false, + // This suite exercises the legacy nested event-window reader. The default + // summary timeline projects the full semantic stream before pagination. + includeNestedRows: true, maxInlineOutputChars: 32_000, maxSeq: 0, page: cursor diff --git a/packages/thread-view/src/build-thread-timeline.ts b/packages/thread-view/src/build-thread-timeline.ts index 4c32d1022b..1b8c1c0fa3 100644 --- a/packages/thread-view/src/build-thread-timeline.ts +++ b/packages/thread-view/src/build-thread-timeline.ts @@ -1359,6 +1359,26 @@ function orderRowsAfterExternalUserBoundary( return [...rows.slice(0, suffixStartIndex), ...orderedSuffix]; } +/** Restore source order for thread-scoped system events emitted during a turn. */ +function restoreLateSystemRowSourceOrder(rows: TimelineRow[]): TimelineRow[] { + const orderedRows: TimelineRow[] = []; + for (const row of rows) { + if (row.kind !== "system") { + orderedRows.push(row); + continue; + } + const insertionIndex = orderedRows.findIndex( + (candidate) => candidate.sourceSeqStart > row.sourceSeqStart, + ); + if (insertionIndex === -1) { + orderedRows.push(row); + } else { + orderedRows.splice(insertionIndex, 0, row); + } + } + return orderedRows; +} + function buildTimelineRows( projection: EventProjection, options: BuildTimelineRowsOptions, @@ -1388,9 +1408,11 @@ function buildTimelineRows( } } - return orderRowsAfterExternalUserBoundary( - rows, - collectExternalUserBoundarySeqs(projection), + return restoreLateSystemRowSourceOrder( + orderRowsAfterExternalUserBoundary( + rows, + collectExternalUserBoundarySeqs(projection), + ), ); } diff --git a/packages/thread-view/src/completed-turn-grouping.ts b/packages/thread-view/src/completed-turn-grouping.ts index bf0a80fa03..cbac72e35d 100644 --- a/packages/thread-view/src/completed-turn-grouping.ts +++ b/packages/thread-view/src/completed-turn-grouping.ts @@ -85,48 +85,25 @@ function getSummaryMessageBounds( return { startedAt }; } -function combineSummaryGroupsWithoutLaterHumanBoundary( +function applySingleSummaryTurnBounds( turn: EventProjectionTurn, items: readonly CompletedTurnSummaryItem[], ): CompletedTurnSummaryItem[] { const summaryGroups = items.filter(isCompletedTurnSummaryGroup); - if (summaryGroups.length === 0) { + if (summaryGroups.length !== 1) { return [...items]; } - const canUseCanonicalIdentity = - (turn.externalUserBoundarySeqs?.length ?? 0) === 0 && - !items.some( - (item) => - item.kind === "ungrouped-message" && - isTimelineUngroupableMessage(item.message) && - item.message.sourceSeqStart > turn.sourceSeqStart, - ); - if (!canUseCanonicalIdentity) { - return [...items]; - } - - const firstSummaryGroup = summaryGroups[0]; - if (!firstSummaryGroup) { - return [...items]; - } - const combinedSummaryGroup: CompletedTurnSummaryGroup = { - ...firstSummaryGroup, - startedAt: turn.startedAt, - completedAt: turn.completedAt, - rowIdSegmentIndex: null, - sourceBounds: - summaryGroups.length === 1 ? firstSummaryGroup.sourceBounds : "messages", - sourceMessages: summaryGroups.flatMap((group) => group.sourceMessages), - summaryCount: summaryGroups.reduce( - (count, group) => count + group.summaryCount, - 0, - ), - }; - return items.flatMap((item): CompletedTurnSummaryItem[] => { - if (item === firstSummaryGroup) return [combinedSummaryGroup]; - return isCompletedTurnSummaryGroup(item) ? [] : [item]; - }); + const onlySummaryGroup = summaryGroups[0]; + return items.map((item) => + item === onlySummaryGroup + ? { + ...item, + startedAt: turn.startedAt, + completedAt: turn.completedAt, + } + : item, + ); } function splitCompletedTurnMessages( @@ -321,7 +298,7 @@ function groupCompletedTurnSummaryMessages( externalBoundaryIndex += 1; } flushGroupedMessages(); - return combineSummaryGroupsWithoutLaterHumanBoundary(turn, items); + return applySingleSummaryTurnBounds(turn, items); } export function groupCompletedTurnMessages( diff --git a/packages/thread-view/test/completed-turn-grouping.test.ts b/packages/thread-view/test/completed-turn-grouping.test.ts index efb777e1fd..a880388a58 100644 --- a/packages/thread-view/test/completed-turn-grouping.test.ts +++ b/packages/thread-view/test/completed-turn-grouping.test.ts @@ -226,7 +226,7 @@ describe("groupCompletedTurnMessages", () => { kind: "summary", startedAt: 1, completedAt: 4, - rowIdSegmentIndex: null, + rowIdSegmentIndex: 0, sourceBounds: "messages", sourceMessages: [{ id: "narration" }, { id: "command" }], summaryCount: 2, @@ -236,7 +236,7 @@ describe("groupCompletedTurnMessages", () => { expect(groups.terminalMessages).toEqual([hookReply]); }); - it("uses the canonical row identity when the accepted request starts the turn", () => { + it("keeps segmented row identity when the accepted request starts the turn", () => { const seed = userMessage({ id: "seed", seq: 1 }); const narration = assistantMessage({ id: "narration", seq: 2 }); const command = commandMessage({ id: "command", seq: 3 }); @@ -251,7 +251,7 @@ describe("groupCompletedTurnMessages", () => { { kind: "ungrouped-message", message: { id: "seed" } }, { kind: "summary", - rowIdSegmentIndex: null, + rowIdSegmentIndex: 0, sourceBounds: "messages", sourceMessages: [{ id: "narration" }, { id: "command" }], }, @@ -259,7 +259,7 @@ describe("groupCompletedTurnMessages", () => { ]); }); - it("combines work around visible assistant replies without a later human boundary", () => { + it("keeps work segmented around visible assistant replies", () => { const firstNarration = assistantMessage({ id: "narration-1", seq: 1 }); const firstCommand = commandMessage({ id: "command-1", seq: 2 }); const visibleReply = assistantMessage({ id: "visible-reply", seq: 3 }); @@ -284,15 +284,15 @@ describe("groupCompletedTurnMessages", () => { expect(groups.summaryItems).toMatchObject([ { kind: "summary", - rowIdSegmentIndex: null, - sourceMessages: [ - { id: "narration-1" }, - { id: "command-1" }, - { id: "narration-2" }, - { id: "command-2" }, - ], + rowIdSegmentIndex: 0, + sourceMessages: [{ id: "narration-1" }, { id: "command-1" }], }, { kind: "ungrouped-message", message: { id: "visible-reply" } }, + { + kind: "summary", + rowIdSegmentIndex: 1, + sourceMessages: [{ id: "narration-2" }, { id: "command-2" }], + }, ]); }); From 8e231481a5c96ef41a514fb7a041cf09e83089d7 Mon Sep 17 00:00:00 2001 From: Michael Yong Date: Wed, 26 Aug 2026 14:14:19 -0700 Subject: [PATCH 2/2] Close semantic rows across timeline page seams --- .../services/threads/timeline-pagination.ts | 19 +- apps/server/src/services/threads/timeline.ts | 29 ++- .../threads/timeline-in-turn-window.test.ts | 187 ++++++++++++++++++ packages/db/src/data/events.ts | 20 ++ packages/db/src/data/index.ts | 1 + packages/db/test/data/events.test.ts | 19 ++ 6 files changed, 263 insertions(+), 12 deletions(-) diff --git a/apps/server/src/services/threads/timeline-pagination.ts b/apps/server/src/services/threads/timeline-pagination.ts index 5de76ef269..30302bce8c 100644 --- a/apps/server/src/services/threads/timeline-pagination.ts +++ b/apps/server/src/services/threads/timeline-pagination.ts @@ -184,11 +184,22 @@ export function paginateTimelineRows( rows: [...rows], }; } - // Every window ends strictly before its cursor, so no segment at or past the - // cursor was read and none has to be trimmed off here. - const selectedSegments = segments.slice(-page.segmentLimit); + // An older semantic-summary read can overlap its cursor segment so rows that + // cross the raw event boundary are projected with context from both sides. + // The cursor segment is context only; this page owns everything before it. + const cursorIndex = + page.kind === "older" + ? segments.findIndex( + (segment) => + segment.cursor.anchorSeq === page.beforeCursor.anchorSeq && + segment.cursor.anchorId === page.beforeCursor.anchorId, + ) + : -1; + const eligibleSegments = + cursorIndex === -1 ? segments : segments.slice(0, cursorIndex); + const selectedSegments = eligibleSegments.slice(-page.segmentLimit); const hasOlderRows = - knownHasOlderSegments ?? segments.length > selectedSegments.length; + knownHasOlderSegments ?? eligibleSegments.length > selectedSegments.length; const oldestSelectedSegment = selectedSegments[0]; return { diff --git a/apps/server/src/services/threads/timeline.ts b/apps/server/src/services/threads/timeline.ts index 9cac006a7e..0efb305775 100644 --- a/apps/server/src/services/threads/timeline.ts +++ b/apps/server/src/services/threads/timeline.ts @@ -28,6 +28,7 @@ import type { import { findStoredTimelineWindowByteBudgetFloor, findTimelineWindowBudgetFloorSequence, + findTimelineSegmentAnchorSequenceAfter, getStoredEventRowsByParentToolCallIdsDataBytes, getEnvironment, findUnfinishedTurnCoveringSequence, @@ -1357,12 +1358,9 @@ function resolveTimelineSegmentWindow( bounds.sequenceWindowStart === null && precedingAnchors.length <= page.segmentLimit; return { - // Every cursor names the first sequence the page that issued it covered, - // so this page ends exactly there. Reading up to the *next anchor* past - // the cursor instead — and trimming that segment off after projecting it - // — meant an older page read one whole extra segment beyond its budget: - // on a thread with a 3,900-event turn, 5,513 events against a budget of - // 1,500, all to discard the surplus. + // Transport-window readers end exactly at the cursor. The top-level + // semantic summary can add one segment of projection overlap later, + // after it has deliberately opted out of these event/byte budgets. beforeSequence: cursor.anchorSeq, byteWindowSequenceStart: sequenceCursor?.kind === "byte" ? bounds.sequenceStart : null, @@ -1430,13 +1428,28 @@ function selectStandardTimelineEventRows( page, threadId: thread.id, }); + // A semantic row may start below a user-message cursor and finish above it + // (for example, an assistant message interrupted by a steer). The older + // summary page reads through the following segment anchor, projects that + // overlap, and trims the cursor segment afterward. Nested transport windows + // keep their existing exact upper bound. + const projectionWindow = + !enforceByteBudget && page.kind === "older" + ? { + ...segmentWindow, + beforeSequence: findTimelineSegmentAnchorSequenceAfter(db, { + sequence: page.beforeCursor.anchorSeq, + threadId: thread.id, + }), + } + : segmentWindow; const window = enforceByteBudget ? applyTimelineWindowByteBudget(db, { maxInlineOutputChars, threadId: thread.id, - window: segmentWindow, + window: projectionWindow, }) - : segmentWindow; + : projectionWindow; if ( !window.hasAnchors && window.sequenceWindowStart === null && diff --git a/apps/server/test/services/threads/timeline-in-turn-window.test.ts b/apps/server/test/services/threads/timeline-in-turn-window.test.ts index 775d42a46b..4d14df4992 100644 --- a/apps/server/test/services/threads/timeline-in-turn-window.test.ts +++ b/apps/server/test/services/threads/timeline-in-turn-window.test.ts @@ -629,6 +629,193 @@ describe("in-turn timeline windows", () => { ); }); + it("keeps an assistant message that finishes after the next user anchor", () => { + const { db, thread } = setup(); + const firstRequestId = requestId(1); + const secondRequestId = requestId(2); + insertEvents(db, noopNotifier, [ + { + threadId: thread.id, + sequence: 1, + type: "client/turn/requested", + scope: threadScope(), + itemId: null, + itemKind: null, + parentToolCallId: null, + data: JSON.stringify({ + direction: "outbound", + source: "tell", + initiator: "user", + request: { method: "turn/start", params: {} }, + requestId: firstRequestId, + senderThreadId: null, + input: [{ type: "text", text: "First prompt", mentions: [] }], + target: { kind: "thread-start" }, + execution, + }), + }, + { + threadId: thread.id, + sequence: 2, + type: "turn/started", + scope: turnScope("turn-1"), + providerThreadId, + itemId: null, + itemKind: null, + parentToolCallId: null, + data: JSON.stringify({}), + }, + { + threadId: thread.id, + sequence: 3, + type: "turn/input/accepted", + scope: turnScope("turn-1"), + providerThreadId, + itemId: null, + itemKind: null, + parentToolCallId: null, + data: JSON.stringify({ clientRequestId: firstRequestId }), + }, + { + threadId: thread.id, + sequence: 4, + type: "item/started", + scope: turnScope("turn-1"), + providerThreadId, + itemId: "assistant-1", + itemKind: "agentMessage", + parentToolCallId: null, + data: JSON.stringify({ + item: { type: "agentMessage", id: "assistant-1", text: "" }, + }), + }, + { + threadId: thread.id, + sequence: 5, + type: "item/agentMessage/delta", + scope: turnScope("turn-1"), + providerThreadId, + itemId: "assistant-1", + itemKind: null, + parentToolCallId: null, + data: JSON.stringify({ itemId: "assistant-1", delta: "Answer" }), + }, + { + threadId: thread.id, + sequence: 6, + type: "client/turn/requested", + scope: threadScope(), + itemId: null, + itemKind: null, + parentToolCallId: null, + data: JSON.stringify({ + direction: "outbound", + source: "tell", + initiator: "user", + request: { method: "turn/start", params: {} }, + requestId: secondRequestId, + senderThreadId: null, + input: [{ type: "text", text: "Second prompt", mentions: [] }], + target: { kind: "new-turn" }, + execution, + }), + }, + { + threadId: thread.id, + sequence: 7, + type: "item/completed", + scope: turnScope("turn-1"), + providerThreadId, + itemId: "assistant-1", + itemKind: "agentMessage", + parentToolCallId: null, + data: JSON.stringify({ + item: { + type: "agentMessage", + id: "assistant-1", + text: "Answer completed after the second prompt arrived.", + }, + }), + }, + { + threadId: thread.id, + sequence: 8, + type: "turn/completed", + scope: turnScope("turn-1"), + providerThreadId, + itemId: null, + itemKind: null, + parentToolCallId: null, + data: JSON.stringify({ status: "completed", providerThreadId }), + }, + { + threadId: thread.id, + sequence: 9, + type: "turn/started", + scope: turnScope("turn-2"), + providerThreadId, + itemId: null, + itemKind: null, + parentToolCallId: null, + data: JSON.stringify({}), + }, + { + threadId: thread.id, + sequence: 10, + type: "turn/input/accepted", + scope: turnScope("turn-2"), + providerThreadId, + itemId: null, + itemKind: null, + parentToolCallId: null, + data: JSON.stringify({ clientRequestId: secondRequestId }), + }, + { + threadId: thread.id, + sequence: 11, + type: "turn/completed", + scope: turnScope("turn-2"), + providerThreadId, + itemId: null, + itemKind: null, + parentToolCallId: null, + data: JSON.stringify({ status: "completed", providerThreadId }), + }, + ]); + + const newest = buildSummaryPage(db, thread, 100, null, 1).response; + expect(newest.timelinePage.olderCursor).not.toBeNull(); + const older = buildSummaryPage( + db, + thread, + 100, + newest.timelinePage.olderCursor, + 1, + ).response; + const olderRowIds = new Set(older.rows.map((row) => row.id)); + const paginatedRows = [ + ...older.rows, + ...newest.rows.filter((row) => !olderRowIds.has(row.id)), + ]; + const unpaginatedRows = buildSummaryPage( + db, + thread, + LARGE_BUDGET, + null, + LARGE_BUDGET, + ).response.rows; + + expect(paginatedRows).toEqual(unpaginatedRows); + expect( + paginatedRows.some( + (row) => + row.kind === "conversation" && + row.role === "assistant" && + row.text === "Answer completed after the second prompt arrived.", + ), + ).toBe(true); + }); + it("keeps assistant prelude rows before the first user anchor", () => { const { db, thread } = setup(); const firstRequestId = requestId(1); diff --git a/packages/db/src/data/events.ts b/packages/db/src/data/events.ts index 5d73c35066..695178aba9 100644 --- a/packages/db/src/data/events.ts +++ b/packages/db/src/data/events.ts @@ -2940,6 +2940,26 @@ export interface TimelineSegmentAnchorLookupArgs { sequence: number; } +/** The first segment anchor strictly after `sequence`, if any. */ +export function findTimelineSegmentAnchorSequenceAfter( + db: DbConnection, + args: TimelineSegmentAnchorLookupArgs, +): number | undefined { + const row = db + .select({ sequence: events.sequence }) + .from(events) + .where( + and( + timelineSegmentAnchorConditions(args.threadId), + gt(events.sequence, args.sequence), + ), + ) + .orderBy(events.sequence) + .limit(1) + .get(); + return row?.sequence; +} + /** The segment anchor at exactly `sequence`, if that turn qualifies as one. */ export function getTimelineSegmentAnchorAtSequence( db: DbConnection, diff --git a/packages/db/src/data/index.ts b/packages/db/src/data/index.ts index c902d9033b..48eb00163d 100644 --- a/packages/db/src/data/index.ts +++ b/packages/db/src/data/index.ts @@ -261,6 +261,7 @@ export { getStoredEventRowsByParentToolCallIdsDataBytes, findUnfinishedTurnCoveringSequence, hasParentedEventCrossingSequence, + findTimelineSegmentAnchorSequenceAfter, getTimelineSegmentAnchorAtSequence, listStoredClientTurnRequestIdsInRange, listStoredClientTurnRequestRowsByKeys, diff --git a/packages/db/test/data/events.test.ts b/packages/db/test/data/events.test.ts index ee1e065283..2fe24c33e4 100644 --- a/packages/db/test/data/events.test.ts +++ b/packages/db/test/data/events.test.ts @@ -17,6 +17,7 @@ import { appendStoredThreadEventsInTransaction, findStoredEventRow, findStoredTimelineWindowByteBudgetFloor, + findTimelineSegmentAnchorSequenceAfter, findTimelineWindowBudgetFloorSequence, getActiveStoredTurnId, getHighWaterMarks, @@ -1460,6 +1461,24 @@ describe("events", () => { threadId: thread.id, }), ).toEqual({ rowId: `${thread.id}:user-seed:2`, sequence: 2 }); + expect( + findTimelineSegmentAnchorSequenceAfter(db, { + sequence: 7, + threadId: thread.id, + }), + ).toBe(8); + expect( + findTimelineSegmentAnchorSequenceAfter(db, { + sequence: 10, + threadId: thread.id, + }), + ).toBe(11); + expect( + findTimelineSegmentAnchorSequenceAfter(db, { + sequence: 11, + threadId: thread.id, + }), + ).toBeUndefined(); }); it("loads timeline event windows with sequence bounds and exclusions", () => {