From cc798abf9800e68d36533766774d9b9fe88ce64f Mon Sep 17 00:00:00 2001 From: Michael Yong Date: Tue, 25 Aug 2026 16:55:45 -0700 Subject: [PATCH 1/4] Preserve semantic timeline turn groups --- .../thread/timeline/ThreadTimelineRows.tsx | 13 +- .../src/hooks/queries/thread-queries.test.tsx | 6 +- .../thread-detail/thread-detail-queries.ts | 2 +- apps/server/src/services/threads/timeline.ts | 111 ++++++++--- .../threads/timeline-in-turn-window.test.ts | 176 +++++++++++++++++- .../src/timeline/timeline-merge.ts | 99 ++++++---- .../client-core/test/timeline-merge.test.ts | 49 ++++- packages/db/src/data/events.ts | 42 +++++ packages/db/src/data/index.ts | 1 + .../thread-view/src/build-thread-timeline.ts | 61 +++++- .../src/completed-turn-grouping.ts | 72 ++++--- .../test/completed-turn-grouping.test.ts | 24 +-- 12 files changed, 525 insertions(+), 131 deletions(-) diff --git a/apps/app/src/components/thread/timeline/ThreadTimelineRows.tsx b/apps/app/src/components/thread/timeline/ThreadTimelineRows.tsx index 198d678c6f..dc6d88244b 100644 --- a/apps/app/src/components/thread/timeline/ThreadTimelineRows.tsx +++ b/apps/app/src/components/thread/timeline/ThreadTimelineRows.tsx @@ -1493,15 +1493,20 @@ function LazyTurnRowBody({ showAssistantMessageActions, }: LazyTurnRowBodyProps) { const { getViewRows, threadId } = useTimelineRendererStaticContext(); - const { threadId: rowThreadId, turnId: rowTurnId } = row; + const { + sourceSeqEnd, + sourceSeqStart, + threadId: rowThreadId, + turnId: rowTurnId, + } = row; const identity = useMemo( () => ({ - sourceSeqEnd: row.sourceSeqEnd, - sourceSeqStart: row.sourceSeqStart, + sourceSeqEnd, + sourceSeqStart, threadId: threadId ?? rowThreadId, turnId: rowTurnId, }), - [row.sourceSeqEnd, row.sourceSeqStart, rowThreadId, rowTurnId, threadId], + [rowThreadId, rowTurnId, sourceSeqEnd, sourceSeqStart, threadId], ); const { data: detail, diff --git a/apps/app/src/hooks/queries/thread-queries.test.tsx b/apps/app/src/hooks/queries/thread-queries.test.tsx index 7cc8f76093..2f5ace95d9 100644 --- a/apps/app/src/hooks/queries/thread-queries.test.tsx +++ b/apps/app/src/hooks/queries/thread-queries.test.tsx @@ -239,7 +239,11 @@ describe("useThreadTimelineTurnDetails", () => { expect(sdk.threads.timelineTurnDetails).toHaveBeenNthCalledWith( 2, - expect.objectContaining({ cursor: "cursor-2" }), + expect.objectContaining({ + cursor: "cursor-2", + sourceSeqEnd: "2", + sourceSeqStart: "1", + }), ); await waitFor(() => expect( diff --git a/apps/mobile/src/data/thread-detail/thread-detail-queries.ts b/apps/mobile/src/data/thread-detail/thread-detail-queries.ts index a5496b60c4..8e0d4b135d 100644 --- a/apps/mobile/src/data/thread-detail/thread-detail-queries.ts +++ b/apps/mobile/src/data/thread-detail/thread-detail-queries.ts @@ -223,7 +223,7 @@ export function useThreadQueuedMessages( /** * Lazy children of one completed-turn summary row. The server owns page - * boundaries; the client cache identifies only the turn. + * boundaries; the client cache identifies the row's semantic source range. * A history rewrite invalidates every detail page for the thread. */ export function useTimelineTurnDetails( diff --git a/apps/server/src/services/threads/timeline.ts b/apps/server/src/services/threads/timeline.ts index 8367e4d352..0fc1c88ac9 100644 --- a/apps/server/src/services/threads/timeline.ts +++ b/apps/server/src/services/threads/timeline.ts @@ -26,6 +26,7 @@ import type { TimelineTurnSummaryDetailsResponse, } from "@bb/server-contract"; import { + findStoredTurnAssistantMessageContextRows, findStoredTimelineWindowByteBudgetFloor, findTimelineWindowBudgetFloorSequence, getStoredEventRowsByParentToolCallIdsDataBytes, @@ -173,7 +174,8 @@ interface BuildTimelineTurnDetailsPageOptions extends TimelineTurnSummarySelecti interface BuildTimelineTurnSummaryDetailsRangeOptions extends BuildTimelineTurnSummaryDetailsOptions { preloadedEventRows?: readonly StoredEventRow[]; - resourceKind: "exact-range" | "page"; + resourceKind: "legacy-exact-range" | "logical-exact-range" | "page"; + semanticSourceSeqStart?: number; } export const THREAD_TIMELINE_DEFAULT_SEGMENT_LIMIT = 20; @@ -823,6 +825,11 @@ interface SequenceWindowItemRowsArgs extends TimelineWindowRowsArgs { beforeSequence: number | undefined; /** See {@link InlineOutputCharLimit}. */ maxInlineOutputChars: InlineOutputCharLimit; + /** + * Earliest sequence whose items belong to this semantic resource. Items + * starting before it are neighboring-group context, not page-owned items. + */ + itemOwnershipSequenceStart?: number; /** Inclusive lower bound of the window. */ sequenceStart: number; } @@ -890,6 +897,7 @@ function ensureSequenceWindowWholeItemRows( threadId: args.threadId, }); const itemKeysOwnedByNewerWindow = new Set(); + const itemKeysOwnedBeforeSemanticSelection = new Set(); const itemsStartingBeforeWindow = new Map(); for (const span of spans) { const key = scopedItemRefKey(span); @@ -900,6 +908,13 @@ function ensureSequenceWindowWholeItemRows( itemKeysOwnedByNewerWindow.add(key); continue; } + if ( + args.itemOwnershipSequenceStart !== undefined && + span.minSequence < args.itemOwnershipSequenceStart + ) { + itemKeysOwnedBeforeSemanticSelection.add(key); + continue; + } if (span.minSequence < args.sequenceStart) { itemsStartingBeforeWindow.set(key, { itemId: span.itemId, @@ -912,9 +927,12 @@ function ensureSequenceWindowWholeItemRows( const rows = args.rows.filter( (row) => row.itemId === null || - !itemKeysOwnedByNewerWindow.has( + (!itemKeysOwnedByNewerWindow.has( scopedItemRefKey(storedEventRowItemRef(row)), - ), + ) && + !itemKeysOwnedBeforeSemanticSelection.has( + scopedItemRefKey(storedEventRowItemRef(row)), + )), ); if (itemsStartingBeforeWindow.size === 0) { return rows; @@ -1581,14 +1599,10 @@ function buildSequencePageTimelineRows( return [ { ...row, - // A finished turn summary is one logical row even when transport - // budgets split its source events across pages. Other row kinds keep - // page-local identities because equal ids can represent closure - // context rather than disjoint pieces of one row. - id: - row.completedAt === null || row.status === "pending" - ? `${row.id}${suffix}` - : row.id, + // Every transport fragment keeps a page-local identity. The client + // coalesces only adjacent completed-turn fragments at a page seam; + // visible conversation rows therefore remain semantic boundaries. + id: `${row.id}${suffix}`, sourceSeqEnd, sourceSeqStart, }, @@ -1997,6 +2011,8 @@ function buildTimelineTurnSummaryDetailsRange( const includeProviderUnhandledOperations = options.includeProviderUnhandledOperations; + const useSemanticSelectionContext = + options.resourceKind !== "legacy-exact-range"; const detailsWindow = { beforeSequence: options.sourceSeqEnd + 1, excludedTypes: THREAD_TIMELINE_EXCLUDED_EVENT_TYPES, @@ -2110,6 +2126,15 @@ function buildTimelineTurnSummaryDetailsRange( }, useExactEventRowBounds: exactEventRowsForRequestedTurn.removedRows, }); + const assistantContextRows = useSemanticSelectionContext + ? findStoredTurnAssistantMessageContextRows(db, { + afterSequence: sourceRange.sourceSeqEnd, + beforeSequence: sourceRange.sourceSeqStart, + maxInlineOutputChars: DEFAULT_MAX_INLINE_OUTPUT_CHARS, + threadId: thread.id, + turnId: options.turnId, + }) + : { after: null, before: null }; // The same whole-item ownership rule the timeline window applies, for the // same reason. A byte cut can fall between an item's `item/started` and its // `item/completed`, and the timeline gives such an item to the newest slice. @@ -2117,18 +2142,27 @@ function buildTimelineTurnSummaryDetailsRange( // `item/started` row alone and render it "pending" after the turn finished. const wholeItemEventRows = ensureSequenceWindowWholeItemRows(db, { beforeSequence: detailsWindow.beforeSequence, + itemOwnershipSequenceStart: + options.resourceKind === "legacy-exact-range" + ? undefined + : (options.semanticSourceSeqStart ?? sourceRange.sourceSeqStart), maxInlineOutputChars: detailsInlineOutputLimit, rows: mergeStoredEventRowsById([...requestedTurnStartedRows, ...eventRows]), sequenceStart: detailsWindow.sequenceStart, threadId: thread.id, }); + const eventRowsWithSemanticContext = mergeStoredEventRowsById([ + ...(assistantContextRows.before ? [assistantContextRows.before] : []), + ...wholeItemEventRows, + ...(assistantContextRows.after ? [assistantContextRows.after] : []), + ]); // The floor queries measured the slice before closure, and closure backfills // the earlier lifecycle rows of the items this slice owns. Measure what the // 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); + byteLengthOfStoredEventRows(eventRowsWithSemanticContext); const eventRowsWithParentedChildren = ensureTimelineWindowParentedRows(db, { maxInlineOutputChars: detailsInlineOutputLimit, outOfBoundsChildDataByteLimit: @@ -2138,29 +2172,55 @@ function buildTimelineTurnSummaryDetailsRange( sequenceStart: detailsWindow.sequenceStart, }, threadId: thread.id, - rows: wholeItemEventRows, + rows: eventRowsWithSemanticContext, }).rows; const eventRowsWithTurnStarts = ensureTimelineWindowTurnStartedRows(db, { threadId: thread.id, rows: eventRowsWithParentedChildren, }); + const eventRowsWithTurnLifecycle = useSemanticSelectionContext + ? ensureSequenceWindowTurnCompletedRows(db, { + threadId: thread.id, + rows: eventRowsWithTurnStarts, + }) + : eventRowsWithTurnStarts; const eventRowsWithBackgroundTaskState = ensureTimelineWindowBackgroundTaskStateRows(db, { threadId: thread.id, - rows: eventRowsWithTurnStarts, + rows: eventRowsWithTurnLifecycle, }); - const projectionSourceSeqStart = eventRowsWithTurnStarts.reduce( - (sourceSeqStart, row) => - row.type === "turn/started" && row.turnId === options.turnId - ? Math.min(sourceSeqStart, row.sequence) - : sourceSeqStart, - sourceRange.sourceSeqStart, + const contextOnlyMessageSeqs = new Set( + [assistantContextRows.before, assistantContextRows.after].flatMap((row) => + row ? [row.sequence] : [], + ), ); + const projectionSourceSeqStart = useSemanticSelectionContext + ? sourceRange.sourceSeqStart + : eventRowsWithTurnStarts.reduce( + (sourceSeqStart, row) => + row.type === "turn/started" && row.turnId === options.turnId + ? Math.min(sourceSeqStart, row.sequence) + : sourceSeqStart, + sourceRange.sourceSeqStart, + ); const projectionArgs = { events: eventRowsWithBackgroundTaskState.map((row) => toThreadEventWithMeta(row), ), options: { + allowContextExpandedMatch: useSemanticSelectionContext, + contextOnlyCompletedTurnIds: + useSemanticSelectionContext && + eventRowsWithTurnLifecycle.some( + (row) => + row.type === "turn/completed" && + row.turnId === options.turnId && + row.sequence > sourceRange.sourceSeqEnd, + ) + ? new Set([options.turnId]) + : undefined, + contextOnlyMessageSeqs: + contextOnlyMessageSeqs.size === 0 ? undefined : contextOnlyMessageSeqs, includeProviderUnhandledOperations, sourceSeqEnd: sourceRange.sourceSeqEnd, sourceSeqStart: projectionSourceSeqStart, @@ -2197,7 +2257,7 @@ export function buildTimelineTurnSummaryDetails( ): TimelineTurnSummaryDetailsResponse { return buildTimelineTurnSummaryDetailsRange(db, thread, { ...options, - resourceKind: "exact-range", + resourceKind: "legacy-exact-range", }); } @@ -2268,7 +2328,11 @@ function resolveCompletedTurnDetailBounds( `Cannot paginate details for incomplete turn ${selection.turnId}`, ); } - if (selection.sourceSeqStart > selection.sourceSeqEnd) { + if ( + selection.sourceSeqStart > selection.sourceSeqEnd || + selection.sourceSeqStart < started.sequence || + selection.sourceSeqEnd > completed.sequence + ) { throw new ApiError( 400, "invalid_request", @@ -2307,7 +2371,7 @@ export function buildTimelineTurnDetailsPage( includeProviderUnhandledOperations: options.includeProviderUnhandledOperations, providerDisplayName: options.providerDisplayName, - resourceKind: "exact-range", + resourceKind: "logical-exact-range", ...bounds, }); return { rows: details.rows, nextCursor: null }; @@ -2346,6 +2410,7 @@ export function buildTimelineTurnDetailsPage( preloadedEventRows: page.rows, providerDisplayName: options.providerDisplayName, resourceKind: "page", + semanticSourceSeqStart: bounds.sourceSeqStart, sourceSeqEnd, sourceSeqStart, turnId: options.turnId, 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..0fc9c8e6ee 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 @@ -88,6 +88,10 @@ function backgroundTaskData(status: "pending" | "completed"): string { interface SeedOptions { /** Emit an assistant message before this item in the last turn. */ assistantBeforeItem?: number; + /** Emit an assistant response while deferred items are still running. */ + assistantBeforeDeferredCompletion?: boolean; + /** Emit a second, consecutive assistant narration before this item. */ + assistantNarrationBeforeItem?: number; /** * Start a workflow background task at the top of the last turn, and complete * it there too when `"completed"`. Its rows sit far below any in-turn cut. @@ -100,6 +104,8 @@ interface SeedOptions { delegateLastTurn?: boolean; /** Emit `turn/completed` for the last turn. */ completeLastTurn: boolean; + /** Emit the last turn's terminal assistant response. */ + finalAssistant?: boolean; /** Character count for each seeded command, when testing byte limits. */ commandChars?: number; /** @@ -245,6 +251,24 @@ function seedTurns( }), }); } + if (isLastTurn && options.assistantNarrationBeforeItem === item) { + const itemId = `${turnId}-narration`; + push({ + type: "item/completed", + scope: turnScope(turnId), + providerThreadId, + itemId, + itemKind: "agentMessage", + parentToolCallId: null, + data: JSON.stringify({ + item: { + type: "agentMessage", + id: itemId, + text: "Continuing with more work.", + }, + }), + }); + } const itemId = `${turnId}-item-${item}`; const command = options.commandChars === undefined @@ -320,6 +344,24 @@ function seedTurns( }), }); } + if (isLastTurn && options.assistantBeforeDeferredCompletion) { + const itemId = `${turnId}-deferred-assistant`; + push({ + type: "item/completed", + scope: turnScope(turnId), + providerThreadId, + itemId, + itemKind: "agentMessage", + parentToolCallId: null, + data: JSON.stringify({ + item: { + type: "agentMessage", + id: itemId, + text: "Intermediate update while work is running.", + }, + }), + }); + } for (const item of deferred) { const itemId = `${turnId}-item-${item}`; push({ @@ -372,6 +414,25 @@ function seedTurns( }); } + if (isLastTurn && options.finalAssistant) { + const itemId = `${turnId}-final`; + push({ + type: "item/completed", + scope: turnScope(turnId), + providerThreadId, + itemId, + itemKind: "agentMessage", + parentToolCallId: null, + data: JSON.stringify({ + item: { + type: "agentMessage", + id: itemId, + text: "Finished.", + }, + }), + }); + } + if (!isLastTurn || options.completeLastTurn) { push({ type: "turn/completed", @@ -501,6 +562,21 @@ function collectCommandCallIds( return count; } +function collectAssistantTexts(rows: readonly TimelineRow[]): string[] { + return rows.flatMap((row): string[] => { + if (row.kind === "conversation" && row.role === "assistant") { + return [row.text]; + } + if (row.kind === "work" && row.workKind === "delegation") { + return collectAssistantTexts(row.childRows); + } + if (row.kind === "turn" && row.children !== null) { + return collectAssistantTexts(row.children); + } + return []; + }); +} + interface WalkResult { maxEventRowCount: number; pages: number; @@ -943,6 +1019,102 @@ describe("in-turn timeline windows", () => { ]); }); + it("keeps row expansion inside work segments split by visible assistant replies", () => { + const { db, thread } = setup(); + seedTurns(db, thread, { + assistantBeforeItem: 1, + assistantNarrationBeforeItem: 1, + completeLastTurn: true, + finalAssistant: true, + itemsPerTurn: [3], + longRunningItemIndexes: [0], + }); + + const timelineRows = buildPage(db, thread, LARGE_BUDGET, null).response + .rows; + expect( + timelineRows.map((row) => + row.kind === "conversation" ? `${row.kind}:${row.role}` : row.kind, + ), + ).toEqual([ + "conversation:user", + "turn", + "conversation:assistant", + "turn", + "conversation:assistant", + ]); + + const turnRows = timelineRows.filter((row) => row.kind === "turn"); + expect(turnRows).toHaveLength(2); + const detailGroups = turnRows.map((row) => { + const detail = buildTimelineTurnDetailsPage(db, thread, { + includeProviderUnhandledOperations: false, + sourceSeqEnd: row.sourceSeqEnd, + sourceSeqStart: row.sourceSeqStart, + turnId: row.turnId, + }); + expect(detail.nextCursor).toBeNull(); + const ids = new Set(); + collectCommandCallIds(detail.rows, ids); + return { + assistantTexts: detail.rows.flatMap((detailRow) => + detailRow.kind === "conversation" && detailRow.role === "assistant" + ? [detailRow.text] + : [], + ), + commandIds: [...ids], + }; + }); + + expect(detailGroups).toEqual([ + { assistantTexts: [], commandIds: ["turn-1-item-0"] }, + { + assistantTexts: ["Continuing with more work."], + commandIds: ["turn-1-item-1", "turn-1-item-2"], + }, + ]); + }); + + it("uses following assistant context beyond an overlapping work range", () => { + const { db, thread } = setup(); + seedTurns(db, thread, { + assistantBeforeDeferredCompletion: true, + completeLastTurn: true, + finalAssistant: true, + itemsPerTurn: [1], + longRunningItemIndexes: [0], + }); + + const timelineRows = buildPage(db, thread, LARGE_BUDGET, null).response + .rows; + expect( + timelineRows.map((row) => + row.kind === "conversation" ? `${row.kind}:${row.role}` : row.kind, + ), + ).toEqual([ + "conversation:user", + "turn", + "conversation:assistant", + "conversation:assistant", + ]); + + const turnRow = timelineRows.find((row) => row.kind === "turn"); + expect(turnRow).toBeDefined(); + if (!turnRow || turnRow.kind !== "turn") return; + + const detail = buildTimelineTurnDetailsPage(db, thread, { + includeProviderUnhandledOperations: false, + sourceSeqEnd: turnRow.sourceSeqEnd, + sourceSeqStart: turnRow.sourceSeqStart, + turnId: turnRow.turnId, + }); + const assistantTexts = collectAssistantTexts(detail.rows); + expect(assistantTexts).toEqual([]); + const commandIds = new Set(); + collectCommandCallIds(detail.rows, commandIds); + expect([...commandIds]).toEqual(["turn-1-item-0"]); + }); + it("pages through a finished turn that exceeds the event-data byte limit", () => { const { db, thread } = setup(); seedTurns(db, thread, { @@ -963,6 +1135,7 @@ describe("in-turn timeline windows", () => { const commandCallIds = new Set(); const turnRowIds = new Set(); + let turnRowCount = 0; let cursor: TimelinePaginationCursor | null = null; let pages = 0; for (;;) { @@ -978,6 +1151,7 @@ describe("in-turn timeline windows", () => { if (row.kind !== "turn") { continue; } + turnRowCount += 1; expect(row.status).toBe("completed"); turnRowIds.add(row.id); } @@ -995,7 +1169,7 @@ describe("in-turn timeline windows", () => { expect(pages).toBeGreaterThan(2); expect(commandCallIds.size).toBe(BYTE_WINDOW_ITEM_COUNT); - expect(turnRowIds.size).toBe(1); + expect(turnRowIds.size).toBe(turnRowCount); const expandedCommandCallIds = new Set(); let expandedCommandRowCount = 0; diff --git a/packages/client-core/src/timeline/timeline-merge.ts b/packages/client-core/src/timeline/timeline-merge.ts index 3cd0d2292b..3dc35d82a4 100644 --- a/packages/client-core/src/timeline/timeline-merge.ts +++ b/packages/client-core/src/timeline/timeline-merge.ts @@ -98,51 +98,62 @@ function appendTimelineRowsPreservingOrder( target: TimelineRow[], rows: readonly TimelineRow[], ): void { - const indexById = new Map(target.map((row, index) => [row.id, index])); + const seenIds = new Set(target.map((row) => row.id)); for (const row of rows) { - const existingIndex = indexById.get(row.id); - if (existingIndex !== undefined) { - const existing = target[existingIndex]; - if ( - existing?.kind === "turn" && - row.kind === "turn" && - existing.completedAt !== null && - row.completedAt !== null && - (existing.sourceSeqEnd < row.sourceSeqStart || - row.sourceSeqEnd < existing.sourceSeqStart) - ) { - const ordered = - existing.sourceSeqStart <= row.sourceSeqStart - ? [existing, row] - : [row, existing]; - const children = [ - ...new Map( - ordered - .flatMap((part) => part.children ?? []) - .map((child) => [child.id, child]), - ).values(), - ]; - target[existingIndex] = { - ...ordered[1], - children: - existing.children === null && row.children === null - ? null - : children, - completedAt: Math.max(existing.completedAt, row.completedAt), - createdAt: Math.min(existing.createdAt, row.createdAt), - sourceSeqEnd: Math.max(existing.sourceSeqEnd, row.sourceSeqEnd), - sourceSeqStart: Math.min(existing.sourceSeqStart, row.sourceSeqStart), - startedAt: Math.min(existing.startedAt, row.startedAt), - summaryCount: existing.summaryCount + row.summaryCount, - }; - } + if (seenIds.has(row.id)) { continue; } - indexById.set(row.id, target.length); + seenIds.add(row.id); target.push(row); } } +function canCoalesceCompletedTurnPageSeam( + older: TimelineRow | undefined, + newer: TimelineRow | undefined, +): older is Extract { + return ( + older?.kind === "turn" && + newer?.kind === "turn" && + older.threadId === newer.threadId && + older.turnId === newer.turnId && + older.completedAt !== null && + newer.completedAt !== null && + older.sourceSeqEnd < newer.sourceSeqStart + ); +} + +function coalesceCompletedTurnPageSeam( + older: Extract, + newer: Extract, +): Extract { + if (older.completedAt === null || newer.completedAt === null) { + throw new Error("Cannot coalesce unfinished turn rows"); + } + const children = [ + ...new Map( + [older, newer] + .flatMap((part) => part.children ?? []) + .map((child) => [child.id, child]), + ).values(), + ]; + return { + ...newer, + // The older page owns the stable position of the combined transport + // fragment. Visible rows at either side of the seam prevent this helper + // from running, so a semantic conversation boundary is never crossed. + id: older.id, + children: + older.children === null && newer.children === null ? null : children, + completedAt: Math.max(older.completedAt, newer.completedAt), + createdAt: Math.min(older.createdAt, newer.createdAt), + sourceSeqEnd: newer.sourceSeqEnd, + sourceSeqStart: older.sourceSeqStart, + startedAt: Math.min(older.startedAt, newer.startedAt), + summaryCount: older.summaryCount + newer.summaryCount, + }; +} + function timelineRowIdentitySignature(row: TimelineRow): string { const turnRequest = row.kind === "conversation" && row.role === "user" ? row.turnRequest : null; @@ -205,7 +216,17 @@ export function prependOlderTimelineRows({ }: PrependOlderTimelineRowsArgs): TimelineRow[] { const rows: TimelineRow[] = []; appendTimelineRowsPreservingOrder(rows, olderRows); - appendTimelineRowsPreservingOrder(rows, loadedRows); + const older = rows.at(-1); + const newer = loadedRows[0]; + if ( + canCoalesceCompletedTurnPageSeam(older, newer) && + newer?.kind === "turn" + ) { + rows[rows.length - 1] = coalesceCompletedTurnPageSeam(older, newer); + appendTimelineRowsPreservingOrder(rows, loadedRows.slice(1)); + } else { + appendTimelineRowsPreservingOrder(rows, loadedRows); + } return rows; } diff --git a/packages/client-core/test/timeline-merge.test.ts b/packages/client-core/test/timeline-merge.test.ts index 177e9b3774..309fe47831 100644 --- a/packages/client-core/test/timeline-merge.test.ts +++ b/packages/client-core/test/timeline-merge.test.ts @@ -2,6 +2,7 @@ import { describe, expect, it } from "vitest"; import type { ThreadTimelineResponse, TimelineCommandWorkRow, + TimelineConversationRow, TimelineDelegationWorkRow, TimelinePaginationCursor, TimelineRow, @@ -86,6 +87,23 @@ function commandRow(args: TimelineTestRowArgs): TimelineCommandWorkRow { }; } +function assistantRow(args: TimelineTestRowArgs): TimelineConversationRow { + return { + id: args.id, + threadId: "thread-1", + turnId: "turn-1", + sourceSeqStart: args.sequence, + sourceSeqEnd: args.endSequence ?? args.sequence, + startedAt: args.sequence, + createdAt: args.sequence, + kind: "conversation", + role: "assistant", + text: args.id, + attachments: null, + turnRequest: null, + }; +} + function turnSummaryRow(args: TimelineTurnTestRowArgs): TimelineTurnRow { return { id: args.id, @@ -223,13 +241,13 @@ describe("timeline page row merging", () => { commandRow({ id: "command-4", sequence: 21 }), ]; const olderSlice = turnSummaryRow({ - id: "turn-1", + id: "turn-1:0:sequence-page:10", sequence: 10, endSequence: 11, children: olderCommands, }); const latestSlice = turnSummaryRow({ - id: "turn-1", + id: "turn-1:0:sequence-page:20", sequence: 20, endSequence: 21, children: latestCommands, @@ -240,7 +258,7 @@ describe("timeline page row merging", () => { loadedRows: [latestSlice], }); - expect(rows.map((row) => row.id)).toEqual(["turn-1"]); + expect(rows.map((row) => row.id)).toEqual([olderSlice.id]); expect(rows[0]).toEqual( expect.objectContaining({ completedAt: 20, @@ -260,6 +278,31 @@ describe("timeline page row merging", () => { ).toEqual(["command-1", "command-2", "command-3", "command-4"]); }); + it("does not coalesce work across a visible assistant reply", () => { + const firstSlice = turnSummaryRow({ + id: "turn-1:0:sequence-page:10", + sequence: 10, + endSequence: 11, + }); + const visibleReply = assistantRow({ id: "assistant-1", sequence: 20 }); + const secondSlice = turnSummaryRow({ + id: "turn-1:1:sequence-page:20", + sequence: 21, + endSequence: 22, + }); + + const rows = prependOlderTimelineRows({ + olderRows: [firstSlice], + loadedRows: [visibleReply, secondSlice], + }); + + expect(rows.map((row) => row.id)).toEqual([ + firstSlice.id, + visibleReply.id, + secondSlice.id, + ]); + }); + it("merges a delegation shell repeated across forward detail pages", () => { const rows = mergeTimelineTurnDetailPages([ [ diff --git a/packages/db/src/data/events.ts b/packages/db/src/data/events.ts index 5d73c35066..8904a0e266 100644 --- a/packages/db/src/data/events.ts +++ b/packages/db/src/data/events.ts @@ -1,5 +1,6 @@ import { and, + asc, desc, eq, gt, @@ -1226,6 +1227,15 @@ export interface ListStoredTimelineWindowEventRowsArgs { threadId: string; } +export interface FindStoredTurnAssistantMessageContextRowsArgs { + afterSequence: number; + beforeSequence: number; + /** See {@link InlineOutputCharLimit}. */ + maxInlineOutputChars: InlineOutputCharLimit; + threadId: string; + turnId: string; +} + export type GetStoredTimelineWindowEventDataBytesArgs = ListStoredTimelineWindowEventRowsArgs; @@ -1578,6 +1588,38 @@ export function findStoredEventRow( ); } +export function findStoredTurnAssistantMessageContextRows( + db: DbConnection, + args: FindStoredTurnAssistantMessageContextRowsArgs, +): { after: StoredEventRow | null; before: StoredEventRow | null } { + const fields = storedEventRowFieldsWithInlineOutputLimit( + args.maxInlineOutputChars, + ); + const scope = [ + eq(events.threadId, args.threadId), + eq(events.turnId, args.turnId), + eq(events.type, "item/completed"), + eq(events.itemKind, "agentMessage"), + ]; + const before = + db + .select(fields) + .from(events) + .where(and(...scope, lt(events.sequence, args.beforeSequence))) + .orderBy(desc(events.sequence)) + .limit(1) + .get() ?? null; + const after = + db + .select(fields) + .from(events) + .where(and(...scope, gt(events.sequence, args.afterSequence))) + .orderBy(asc(events.sequence)) + .limit(1) + .get() ?? null; + return { after, before }; +} + export function listStoredEventRowsByParentToolCallIds( db: DbConnection, args: ListStoredEventRowsByParentToolCallIdsArgs, diff --git a/packages/db/src/data/index.ts b/packages/db/src/data/index.ts index c902d9033b..7595d8afec 100644 --- a/packages/db/src/data/index.ts +++ b/packages/db/src/data/index.ts @@ -237,6 +237,7 @@ export { deleteThreadEventSuffixInTransaction, getHighWaterMarks, findStoredEventRow, + findStoredTurnAssistantMessageContextRows, getActiveStoredTurnId, hasRootStoredTurnStarted, hasStoredTurnStarted, diff --git a/packages/thread-view/src/build-thread-timeline.ts b/packages/thread-view/src/build-thread-timeline.ts index 4c32d1022b..0b8c1b035b 100644 --- a/packages/thread-view/src/build-thread-timeline.ts +++ b/packages/thread-view/src/build-thread-timeline.ts @@ -143,6 +143,9 @@ interface ThreadTimelineSourceSeqRange { } interface BuildThreadTimelineTurnDetailsFromEventsOptions extends ThreadTimelineSourceSeqRange { + allowContextExpandedMatch?: boolean; + contextOnlyCompletedTurnIds?: ReadonlySet; + contextOnlyMessageSeqs?: ReadonlySet; includeProviderUnhandledOperations: boolean; providerDisplayName?: string; threadStatus: Thread["status"]; @@ -172,6 +175,7 @@ type ThreadTimelineTurnDetailsFromEventsResult = interface BuildTurnRowsArgs { contextOnlyCompletedTurnIds?: ReadonlySet; + contextOnlyMessageSeqs?: ReadonlySet; includeNestedRows: boolean; rowIdPrefix: string; turn: EventProjectionTurn; @@ -208,6 +212,7 @@ interface BuildCompletedTurnSummaryRowsArgs { interface BuildTimelineRowsOptions { contextOnlyCompletedTurnIds?: ReadonlySet; + contextOnlyMessageSeqs?: ReadonlySet; includeNestedRows: boolean; rowIdPrefix: string; workspaceRoot: string | null; @@ -1231,6 +1236,7 @@ function buildCompletedTurnSummaryRows({ function buildTurnRows({ contextOnlyCompletedTurnIds, + contextOnlyMessageSeqs, includeNestedRows, rowIdPrefix, turn, @@ -1254,6 +1260,7 @@ function buildTurnRows({ groupCompletedTurnMessages( turn, contextOnlyCompletedTurnIds?.has(turn.turnId) === true, + contextOnlyMessageSeqs, ); const terminalRows = terminalMessages.flatMap((message) => convertMessage(message, { includeNestedRows, rowIdPrefix, workspaceRoot }), @@ -1275,15 +1282,52 @@ type TimelineTurnSummaryRow = Extract; function findMatchingTurnSummaryRow( rows: TimelineRow[], - range: ThreadTimelineSourceSeqRange, + range: ThreadTimelineSourceSeqRange & { allowContextExpandedMatch?: boolean }, ): TimelineTurnSummaryRow | null { + const turnRows = rows.filter( + (row): row is TimelineTurnSummaryRow => row.kind === "turn", + ); + const exact = turnRows.find( + (row) => + row.sourceSeqStart === range.sourceSeqStart && + row.sourceSeqEnd === range.sourceSeqEnd, + ); + if (exact || !range.allowContextExpandedMatch) { + return exact ?? null; + } + + // Lifecycle closure and context-only rows can shift a semantic group's + // projected bounds in either direction. A parent shell can widen them, + // while removing a boundary message can narrow them. Prefer the group with + // the greatest overlap with the server-validated selection, then the least + // total boundary movement. return ( - rows.find( - (row): row is TimelineTurnSummaryRow => - row.kind === "turn" && - row.sourceSeqStart === range.sourceSeqStart && - row.sourceSeqEnd === range.sourceSeqEnd, - ) ?? null + turnRows + .filter( + (row) => + row.sourceSeqEnd >= range.sourceSeqStart && + row.sourceSeqStart <= range.sourceSeqEnd, + ) + .sort((left, right) => { + const leftOverlap = + Math.min(left.sourceSeqEnd, range.sourceSeqEnd) - + Math.max(left.sourceSeqStart, range.sourceSeqStart) + + 1; + const rightOverlap = + Math.min(right.sourceSeqEnd, range.sourceSeqEnd) - + Math.max(right.sourceSeqStart, range.sourceSeqStart) + + 1; + if (leftOverlap !== rightOverlap) { + return rightOverlap - leftOverlap; + } + const leftMovement = + Math.abs(left.sourceSeqStart - range.sourceSeqStart) + + Math.abs(left.sourceSeqEnd - range.sourceSeqEnd); + const rightMovement = + Math.abs(right.sourceSeqStart - range.sourceSeqStart) + + Math.abs(right.sourceSeqEnd - range.sourceSeqEnd); + return leftMovement - rightMovement; + })[0] ?? null ); } @@ -1376,6 +1420,7 @@ function buildTimelineRows( rows, buildTurnRows({ contextOnlyCompletedTurnIds: options.contextOnlyCompletedTurnIds, + contextOnlyMessageSeqs: options.contextOnlyMessageSeqs, turn: entry.turn, includeNestedRows, rowIdPrefix: options.rowIdPrefix, @@ -1474,6 +1519,8 @@ function buildThreadTimelineTurnDetailRows( turnMessageDetail: "full", }); return buildTimelineRows(projection, { + contextOnlyCompletedTurnIds: args.options.contextOnlyCompletedTurnIds, + contextOnlyMessageSeqs: args.options.contextOnlyMessageSeqs, includeNestedRows: true, rowIdPrefix: ROOT_TIMELINE_ROW_ID_PREFIX, workspaceRoot: args.options.workspaceRoot, diff --git a/packages/thread-view/src/completed-turn-grouping.ts b/packages/thread-view/src/completed-turn-grouping.ts index bf0a80fa03..edfbb5d1ac 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( @@ -206,16 +183,26 @@ function groupCompletedTurnSummaryMessages( turn: EventProjectionTurn, summaryMessages: EventProjectionMessage[], terminalMessage: EventProjectionMessage | undefined, + useTurnBounds: boolean, + contextOnlyMessageSeqs?: ReadonlySet, ): CompletedTurnSummaryItem[] { const externalBoundarySeqs = turn.externalUserBoundarySeqs ?? []; const visibleResponseIds = findVisibleResponseMessageIds( summaryMessages, terminalMessage, ); + const selectedSummaryMessages = contextOnlyMessageSeqs + ? summaryMessages.filter( + (message) => + !contextOnlyMessageSeqs.has(message.sourceSeqStart) && + !contextOnlyMessageSeqs.has(message.sourceSeqEnd), + ) + : summaryMessages; if ( + useTurnBounds && externalBoundarySeqs.length === 0 && visibleResponseIds.size === 0 && - !summaryMessages.some(isTimelineUngroupableMessage) + !selectedSummaryMessages.some(isTimelineUngroupableMessage) ) { return [ { @@ -224,8 +211,10 @@ function groupCompletedTurnSummaryMessages( completedAt: turn.completedAt, rowIdSegmentIndex: null, sourceBounds: "turn", - sourceMessages: summaryMessages, - summaryCount: turn.summaryCount, + sourceMessages: selectedSummaryMessages, + summaryCount: contextOnlyMessageSeqs + ? getProjectionSummaryCount(selectedSummaryMessages, undefined) + : turn.summaryCount, }, ]; } @@ -293,7 +282,7 @@ function groupCompletedTurnSummaryMessages( } } - for (const message of summaryMessages) { + for (const message of selectedSummaryMessages) { flushExternalBoundariesBefore(message); if (visibleResponseIds.has(message.id)) { flushGroupedMessages(); @@ -321,12 +310,13 @@ function groupCompletedTurnSummaryMessages( externalBoundaryIndex += 1; } flushGroupedMessages(); - return combineSummaryGroupsWithoutLaterHumanBoundary(turn, items); + return useTurnBounds ? applySingleSummaryTurnBounds(turn, items) : items; } export function groupCompletedTurnMessages( turn: EventProjectionTurn, completionIsContextOnly = false, + contextOnlyMessageSeqs?: ReadonlySet, ): CompletedTurnMessageGroups { const messages = turn.messages ?? []; const { summaryMessages, terminalMessages, trailingMessages } = @@ -340,6 +330,8 @@ export function groupCompletedTurnMessages( turn, summaryMessages, terminalMessages[0], + !completionIsContextOnly, + contextOnlyMessageSeqs, ), ), terminalMessages, diff --git a/packages/thread-view/test/completed-turn-grouping.test.ts b/packages/thread-view/test/completed-turn-grouping.test.ts index efb777e1fd..0a0768dc37 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" }], + }, ]); }); @@ -307,7 +307,7 @@ describe("groupCompletedTurnMessages", () => { expect(groups.summaryItems).toMatchObject([ { kind: "summary", - rowIdSegmentIndex: null, + rowIdSegmentIndex: 0, sourceMessages: [{ id: "assistant" }, { id: "command" }], }, ]); From 0d1d1c616da99e2f761ab65a0421cecd13f3d6ed Mon Sep 17 00:00:00 2001 From: Michael Yong Date: Tue, 25 Aug 2026 21:11:45 -0700 Subject: [PATCH 2/4] checkpoint timeline window correctness against prod corpus --- .../services/threads/timeline-pagination.ts | 55 +- apps/server/src/services/threads/timeline.ts | 852 +++++++++++++++--- .../threads/timeline-in-turn-window.test.ts | 125 ++- .../threads/timeline-pagination.test.ts | 44 + .../threads/timeline-provider-input.test.ts | 436 +++++++-- .../src/timeline/timeline-merge.ts | 10 +- packages/db/src/data/events.ts | 214 ++++- packages/db/src/data/index.ts | 7 + .../thread-view/src/build-thread-timeline.ts | 72 +- .../src/completed-turn-grouping.ts | 28 +- .../test/build-thread-timeline.test.ts | 42 + .../test/completed-turn-grouping.test.ts | 22 + 12 files changed, 1659 insertions(+), 248 deletions(-) diff --git a/apps/server/src/services/threads/timeline-pagination.ts b/apps/server/src/services/threads/timeline-pagination.ts index 5de76ef269..deb87c81b1 100644 --- a/apps/server/src/services/threads/timeline-pagination.ts +++ b/apps/server/src/services/threads/timeline-pagination.ts @@ -107,17 +107,23 @@ function isTimelineSegmentAnchorRow(row: TimelineRow): boolean { function buildTimelineLogicalSegment( rows: TimelineRow[], + cursorRow: TimelineRow | null = null, ): TimelineLogicalSegment { - const anchorRow = rows[0]; - if (!anchorRow) { + const firstRow = rows[0]; + if (!firstRow) { throw new Error("Cannot build a timeline segment without rows"); } + const segmentCursorRow = cursorRow ?? firstRow; return { cursor: { - anchorSeq: anchorRow.sourceSeqStart, - anchorId: anchorRow.id, + anchorSeq: segmentCursorRow.sourceSeqStart, + anchorId: segmentCursorRow.id, }, + // Projection order is semantic, not always source order: completed-turn + // summaries precede their terminal assistant message even when the + // summary's derived source bounds begin later. Sorting a segment by source + // here moves that assistant to the wrong side of "Worked for...". rows, }; } @@ -127,23 +133,54 @@ function buildTimelineLogicalSegments( ): TimelineLogicalSegment[] { const segments: TimelineLogicalSegment[] = []; let currentRows: TimelineRow[] = []; + let currentCursorRow: TimelineRow | null = null; for (const row of rows) { if ( isTimelineSegmentAnchorRow(row) && currentRows.length > 0 && - currentRows[0]?.sourceSeqStart !== row.sourceSeqStart + (currentCursorRow?.sourceSeqStart ?? currentRows[0]?.sourceSeqStart) !== + row.sourceSeqStart ) { - segments.push(buildTimelineLogicalSegment(currentRows)); - currentRows = [row]; + // Projection groups thread-scoped messages separately from turn rows. + // A thread event can therefore appear immediately before the accepted + // user row that precedes it in source order. Do not leave that trailing + // event in the older segment: a one-segment page would trim it away. + // Move the anchor ahead of every source-newer trailing row so the + // segment also has the user row as its cursor. A row that starts before + // the anchor stays with that earlier segment even if its lifecycle ends + // after the anchor; the server closes that row with targeted context. + const trailingRowIndex = currentRows.findIndex( + (currentRow) => currentRow.sourceSeqStart >= row.sourceSeqStart, + ); + if (trailingRowIndex === -1) { + segments.push( + buildTimelineLogicalSegment(currentRows, currentCursorRow), + ); + currentRows = [row]; + } else { + const olderRows = currentRows.slice(0, trailingRowIndex); + if (olderRows.length > 0) { + segments.push( + buildTimelineLogicalSegment(olderRows, currentCursorRow), + ); + } + currentRows = [...currentRows.slice(trailingRowIndex), row].sort( + (left, right) => left.sourceSeqStart - right.sourceSeqStart, + ); + } + currentCursorRow = row; continue; } + if (isTimelineSegmentAnchorRow(row) && currentCursorRow === null) { + currentCursorRow = row; + } currentRows.push(row); } if (currentRows.length > 0) { - segments.push(buildTimelineLogicalSegment(currentRows)); + segments.push(buildTimelineLogicalSegment(currentRows, currentCursorRow)); } return segments; @@ -188,7 +225,7 @@ export function paginateTimelineRows( // cursor was read and none has to be trimmed off here. const selectedSegments = segments.slice(-page.segmentLimit); const hasOlderRows = - knownHasOlderSegments ?? segments.length > selectedSegments.length; + knownHasOlderSegments === true || segments.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 0fc1c88ac9..df0f5d54d3 100644 --- a/apps/server/src/services/threads/timeline.ts +++ b/apps/server/src/services/threads/timeline.ts @@ -26,8 +26,12 @@ import type { TimelineTurnSummaryDetailsResponse, } from "@bb/server-contract"; import { + findStoredRootTurnAssistantMessageContextRows, findStoredTurnAssistantMessageContextRows, findStoredTimelineWindowByteBudgetFloor, + hasStoredRootNonAssistantItemBetween, + hasStoredRootTimelineRowsBetween, + hasStoredUserClientTurnRequestBetween, findTimelineWindowBudgetFloorSequence, getStoredEventRowsByParentToolCallIdsDataBytes, getEnvironment, @@ -38,8 +42,11 @@ import { listRecentStoredEventRows, readStoredTimelineWindowForwardPage, listStoredConversationOutlineEventRows, + listStoredContextCompactionStartedRowsByTurnIds, listStoredClientTurnRequestIdsInRange, + listStoredClientTurnRequestRowsByKeys, listStoredEventRowsByParentToolCallIds, + listStoredEventRowsAtSequences, isTimelineCursorSequencePresent, listItemEventSpansByItems, listStoredBufferedTextDeltaRowsByItems, @@ -50,6 +57,7 @@ import { listStoredTimelineWindowEventRows, listTodoSnapshotEventRowsForThread, listStoredDelegatingItemRowsByItemIds, + listStoredThreadCompactedRowsByTurnIds, listStoredTurnCompletedRowsByTurnIds, listStoredTurnInputAcceptedRowsByClientRequestIds, listStoredTurnRejectedRowsByClientRequestIds, @@ -249,6 +257,7 @@ interface TimelineEventRowSelection { byteWindowSequenceEnd: number | null; byteWindowSequenceStart: number | null; contextOnlyToolCallIds: Set; + contextOnlyMessageSeqs: Set; /** See {@link paginateTimelineRows}. */ sequenceWindowStart: TimelineSequenceWindowStart | null; /** See {@link paginateTimelineRows}. */ @@ -256,6 +265,11 @@ interface TimelineEventRowSelection { paginationPage: ThreadTimelinePageRequest; responsePageKind: ThreadTimelinePageKind; oversizedEventPlaceholder: TimelineSystemRow | null; + sourceEndExtensions: Array<{ + itemSourceSeqStart: number; + sourceSeqEnd: number; + turnId: string | null; + }>; rows: StoredEventRow[]; strategy: ThreadTimelineEventSelectionStrategy; } @@ -293,6 +307,7 @@ interface SelectClientRequestContextRowsArgs { interface SelectedClientRequestContextRows { acceptedRows: StoredEventRow[]; rejectedRows: StoredEventRow[]; + turnStartedRows: StoredEventRow[]; } export function toThreadEventWithMeta( @@ -340,28 +355,7 @@ function tryReadClientTurnRequestedRequestId( return event.requestId; } -function tryReadSteerClientTurnRequestedRequestId( - row: StoredEventRow, -): ClientTurnRequestId | null { - if (row.type !== "client/turn/requested") { - return null; - } - const event = parseStoredEvent(row); - if (event.type !== "client/turn/requested") { - return null; - } - - switch (event.target.kind) { - case "auto": - case "steer": - return event.target.expectedTurnId === null ? null : event.requestId; - case "new-turn": - case "thread-start": - return null; - } -} - -function collectSteerClientRequestIdsNeedingContext( +function collectClientRequestIdsNeedingContext( rows: readonly StoredEventRow[], ): ClientTurnRequestId[] { const terminalClientRequestIds = new Set(); @@ -379,7 +373,7 @@ function collectSteerClientRequestIdsNeedingContext( clientRequestIds.delete(clientRequestId); continue; } - const clientRequestId = tryReadSteerClientTurnRequestedRequestId(row); + const clientRequestId = tryReadClientTurnRequestedRequestId(row); if ( clientRequestId === null || terminalClientRequestIds.has(clientRequestId) @@ -530,6 +524,85 @@ function ensureTimelineWindowParentedRows( }; } +function findTimelineWindowParentedSourceEndExtensions( + db: DbConnection, + args: TimelineWindowRowsArgs & { + includeProviderUnhandledOperations: boolean; + }, +): TimelineEventRowSelection["sourceEndExtensions"] { + const parentRows = args.rows.filter( + (row) => row.type === "item/started" && isStoredDelegatingItemRow(row), + ); + if (parentRows.length === 0) { + return []; + } + const structuralRows = listStoredEventRowsByParentToolCallIds(db, { + excludedTypes: args.includeProviderUnhandledOperations + ? THREAD_TIMELINE_EXCLUDED_EVENT_TYPES + : [...THREAD_TIMELINE_EXCLUDED_EVENT_TYPES, "provider/unhandled"], + maxInlineOutputChars: 0, + parentToolCallIds: parentRows.flatMap((row) => + row.itemId === null ? [] : [row.itemId], + ), + threadId: args.threadId, + }); + const childRowsByParentId = new Map(); + for (const row of structuralRows) { + const parentId = getStoredEventParentToolCallId(row); + if (!parentId) { + continue; + } + const rows = childRowsByParentId.get(parentId) ?? []; + rows.push(row); + childRowsByParentId.set(parentId, rows); + } + + function findChildSourceEnd(parentId: string): number | null { + const directRows = childRowsByParentId.get(parentId) ?? []; + const directBackgroundCompletion = directRows.reduce( + (minimum, row) => + row.type === "item/backgroundTask/completed" + ? Math.min(minimum ?? row.sequence, row.sequence) + : minimum, + null, + ); + let sourceSeqEnd: number | null = null; + for (const row of directRows) { + // Background tasks are thread state, not rows inside the delegation's + // child projection, even though their lifecycle keeps the parent's id. + // Including their terminal state would widen the visible summary past + // the canonical delegation row. + if (row.itemKind === "backgroundTask") { + continue; + } + if ( + directBackgroundCompletion !== null && + row.sequence >= directBackgroundCompletion + ) { + continue; + } + sourceSeqEnd = Math.max(sourceSeqEnd ?? row.sequence, row.sequence); + } + return sourceSeqEnd; + } + + return parentRows.flatMap((row) => { + if (row.itemId === null) { + return []; + } + const sourceSeqEnd = findChildSourceEnd(row.itemId); + return sourceSeqEnd === null + ? [] + : [ + { + itemSourceSeqStart: row.sequence, + sourceSeqEnd, + turnId: row.turnId, + }, + ]; + }); +} + /** * Lowest sequence any of these requests was made at. * @@ -561,27 +634,35 @@ function selectClientRequestContextRows( db: DbConnection, args: SelectClientRequestContextRowsArgs, ): SelectedClientRequestContextRows { - const clientRequestIds = collectSteerClientRequestIdsNeedingContext( - args.rows, - ); + const clientRequestIds = collectClientRequestIdsNeedingContext(args.rows); if (clientRequestIds.length === 0) { - return { acceptedRows: [], rejectedRows: [] }; + return { acceptedRows: [], rejectedRows: [], turnStartedRows: [] }; } const afterSequence = minSequenceOfClientRequests( args.rows, new Set(clientRequestIds), ); + const acceptedRows = listStoredTurnInputAcceptedRowsByClientRequestIds(db, { + afterSequence, + clientRequestIds, + threadId: args.threadId, + }); return { - acceptedRows: listStoredTurnInputAcceptedRowsByClientRequestIds(db, { - afterSequence, - clientRequestIds, - threadId: args.threadId, - }), + acceptedRows, rejectedRows: listStoredTurnRejectedRowsByClientRequestIds(db, { afterSequence, clientRequestIds, threadId: args.threadId, }), + // Accepted context assigns an in-window request to its real provider turn. + // The acceptance can sit beyond the page's upper bound, so the ordinary + // window lifecycle pass never saw that turn and could not backfill its + // root. Carry the root as projection context or grouping the accepted user + // message has no turn draft to attach to. + turnStartedRows: ensureTimelineWindowTurnStartedRows(db, { + rows: acceptedRows, + threadId: args.threadId, + }).filter((row) => row.type === "turn/started"), }; } @@ -711,11 +792,13 @@ function selectFullTimelineEventRows( byteWindowSequenceEnd: null, byteWindowSequenceStart: null, contextOnlyToolCallIds: new Set(), + contextOnlyMessageSeqs: new Set(), sequenceWindowStart: null, knownHasOlderSegments: null, paginationPage: page, responsePageKind: page.kind, oversizedEventPlaceholder: null, + sourceEndExtensions: [], rows: listRecentStoredEventRows(db, { threadId: thread.id, excludedTypes: THREAD_TIMELINE_EXCLUDED_EVENT_TYPES, @@ -777,7 +860,7 @@ function ensureTimelineWindowTurnStartedRows( return mergeStoredEventRowsById([...turnStartedRows, ...args.rows]); } -function ensureSequenceWindowTurnCompletedRows( +function ensureTimelineWindowTurnCompletedRows( db: DbConnection, args: TimelineWindowRowsArgs, ): StoredEventRow[] { @@ -820,7 +903,7 @@ function storedEventRowItemRef(row: StoredEventRow): ScopedItemRef { }; } -interface SequenceWindowItemRowsArgs extends TimelineWindowRowsArgs { +interface TimelineWindowItemRowsArgs extends TimelineWindowRowsArgs { /** Exclusive upper bound of the window, or undefined for the latest page. */ beforeSequence: number | undefined; /** See {@link InlineOutputCharLimit}. */ @@ -830,10 +913,17 @@ interface SequenceWindowItemRowsArgs extends TimelineWindowRowsArgs { * starting before it are neighboring-group context, not page-owned items. */ itemOwnershipSequenceStart?: number; + /** Which side of a boundary owns an item that straddles it. */ + ownership: "newest-window" | "starting-window"; /** Inclusive lower bound of the window. */ sequenceStart: number; } +interface TimelineWindowWholeItemRows { + rows: StoredEventRow[]; + sourceEndExtensions: TimelineEventRowSelection["sourceEndExtensions"]; +} + function rowIdentifiesBufferedTextItem(row: StoredEventRow): boolean { if (row.type === "item/started") { return ( @@ -851,25 +941,26 @@ function rowIdentifiesBufferedTextItem(row: StoredEventRow): boolean { } /** - * Makes a sequence-cut window own whole items rather than halves of them. + * Makes a bounded timeline window own whole items rather than halves of them. * - * A cut on a user message never lands inside an item. A cut on the event budget - * does: an `npm run dev` that starts at sequence 2,706 and fails at 5,450 - * straddles any cut in between, and each side then projects its own row under - * the same row id — one of them permanently "pending", and whichever the client - * merges last wins. + * Event and byte budgets can cut anywhere. User-message boundaries can also + * bisect an item: a steer may arrive while the provider is finishing an + * assistant message. Projecting either half alone can drop the item or produce + * two rows under the same id, with the client's merge order deciding which + * survives. * - * The rule is that an item belongs to the newest window holding any of its real - * rows. So this window drops the items that a newer window will also show, and - * backfills the earlier lifecycle rows of the ones it keeps. Background-task - * items are left alone: they deliberately outlive their window and - * {@link ensureTimelineWindowBackgroundTaskStateRows} already carries their - * current state forward. + * Sequence cuts give an item to the newest transport window holding any real + * row. Logical user-message cuts give it to the segment where it started: a + * delegation or assistant remains attached to the work/conversation that + * spawned it even if it finishes after a later user message. Only missing + * lifecycle endpoints are fetched. Background-task items are left alone: they + * deliberately outlive their window and + * {@link ensureTimelineWindowBackgroundTaskStateRows} carries their state. */ -function ensureSequenceWindowWholeItemRows( +function ensureTimelineWindowWholeItemRows( db: DbConnection, - args: SequenceWindowItemRowsArgs, -): StoredEventRow[] { + args: TimelineWindowItemRowsArgs, +): TimelineWindowWholeItemRows { // Keyed by scoped identity, not by item id: providers reuse item ids across // turns (a resumed ACP session restarts its synthetic id counter), and a // thread-wide span for such an id makes every window disown the item. @@ -885,7 +976,7 @@ function ensureSequenceWindowWholeItemRows( } } if (windowItems.size === 0) { - return [...args.rows]; + return { rows: [...args.rows], sourceEndExtensions: [] }; } // Spans, not lifecycle rows. An item emits between its start and its end — @@ -899,9 +990,13 @@ function ensureSequenceWindowWholeItemRows( const itemKeysOwnedByNewerWindow = new Set(); const itemKeysOwnedBeforeSemanticSelection = new Set(); const itemsStartingBeforeWindow = new Map(); + const itemsEndingAfterWindow = new Map(); + const sourceEndExtensions: TimelineEventRowSelection["sourceEndExtensions"] = + []; for (const span of spans) { const key = scopedItemRefKey(span); if ( + args.ownership === "newest-window" && args.beforeSequence !== undefined && span.maxSequence >= args.beforeSequence ) { @@ -915,6 +1010,29 @@ function ensureSequenceWindowWholeItemRows( itemKeysOwnedBeforeSemanticSelection.add(key); continue; } + if ( + args.ownership === "starting-window" && + span.minSequence < args.sequenceStart + ) { + itemKeysOwnedBeforeSemanticSelection.add(key); + continue; + } + if ( + args.ownership === "starting-window" && + args.beforeSequence !== undefined && + span.maxSequence >= args.beforeSequence + ) { + itemsEndingAfterWindow.set(key, { + itemId: span.itemId, + scopeKind: span.scopeKind, + turnId: span.turnId, + }); + sourceEndExtensions.push({ + itemSourceSeqStart: span.minSequence, + sourceSeqEnd: span.maxSequence, + turnId: span.turnId, + }); + } if (span.minSequence < args.sequenceStart) { itemsStartingBeforeWindow.set(key, { itemId: span.itemId, @@ -934,29 +1052,44 @@ function ensureSequenceWindowWholeItemRows( scopedItemRefKey(storedEventRowItemRef(row)), )), ); - if (itemsStartingBeforeWindow.size === 0) { - return rows; + if ( + itemsStartingBeforeWindow.size === 0 && + itemsEndingAfterWindow.size === 0 + ) { + return { rows, sourceEndExtensions }; } - // This window owns these items, so it needs the lifecycle rows that fell - // below the cut — without them a finished command renders "pending" and - // carries neither its command line nor its start time. Only the two lifecycle - // types are fetched for ordinary items: the rest of what an item emitted - // below the cut is the older page's content, and pulling all of it back would - // restore exactly the unbounded read this window exists to avoid. + // This window owns these items, so it needs lifecycle rows that fell outside + // the cut — without them a finished command renders "pending" and carries + // neither its command line nor its final state. Only lifecycle endpoints are + // fetched for ordinary items: pulling all emitted rows across the boundary + // would restore exactly the unbounded read this window exists to avoid. // // Unfinished buffered text is the exception. Its deltas are the only current // snapshot of the message, so dropping the prefix would make text disappear // as the event-budget floor advances. Carry that one item's prefix into the // owning page until item/completed supplies the canonical final text. - const backfillRows = listStoredItemLifecycleRowsByItems(db, { - items: [...itemsStartingBeforeWindow.values()], + const lifecycleItems = new Map([ + ...itemsStartingBeforeWindow, + ...itemsEndingAfterWindow, + ]); + const lifecycleRows = listStoredItemLifecycleRowsByItems(db, { + items: [...lifecycleItems.values()], maxInlineOutputChars: args.maxInlineOutputChars, threadId: args.threadId, - }).filter((row) => row.sequence < args.sequenceStart); + }).filter((row) => { + const key = scopedItemRefKey(storedEventRowItemRef(row)); + return ( + (itemsStartingBeforeWindow.has(key) && + row.sequence < args.sequenceStart) || + (itemsEndingAfterWindow.has(key) && + args.beforeSequence !== undefined && + row.sequence >= args.beforeSequence) + ); + }); const completedItemKeys = new Set(); - for (const row of [...rows, ...backfillRows]) { + for (const row of [...rows, ...lifecycleRows]) { if (row.type === "item/completed" && row.itemId !== null) { completedItemKeys.add(scopedItemRefKey(storedEventRowItemRef(row))); } @@ -966,7 +1099,7 @@ function ensureSequenceWindowWholeItemRows( // item/started event. Classify from either the backfilled lifecycle row or // the in-window delta type so those delta-only items keep their prefix too. const bufferedTextItems = new Map(); - for (const row of [...backfillRows, ...rows]) { + for (const row of [...lifecycleRows, ...rows]) { if (row.itemId === null || !rowIdentifiesBufferedTextItem(row)) { continue; } @@ -981,10 +1114,155 @@ function ensureSequenceWindowWholeItemRows( items: [...bufferedTextItems.values()], threadId: args.threadId, }); - const prefixRows = [...backfillRows, ...bufferedTextRows]; - return prefixRows.length === 0 - ? rows - : mergeStoredEventRowsById([...prefixRows, ...rows]); + // A final progress/delta row can extend an item beyond its lifecycle. Carry + // only that row's metadata so grouping sees its real boundary without + // downloading output across windows. + const sourceEndRows = listStoredEventRowsAtSequences(db, { + maxInlineOutputChars: 0, + sequences: sourceEndExtensions.map((extension) => extension.sourceSeqEnd), + threadId: args.threadId, + }); + const contextRows = [...lifecycleRows, ...bufferedTextRows, ...sourceEndRows]; + return { + rows: + contextRows.length === 0 + ? rows + : mergeStoredEventRowsById([...contextRows, ...rows]), + sourceEndExtensions, + }; +} + +function ensureTimelineWindowContextCompactionRows( + db: DbConnection, + args: TimelineWindowRowsArgs, +): StoredEventRow[] { + const turnIds = new Set(); + for (const row of args.rows) { + if ( + row.type === "item/started" && + row.itemKind === "contextCompaction" && + row.turnId !== null + ) { + turnIds.add(row.turnId); + } + } + if (turnIds.size === 0) { + return [...args.rows]; + } + + const completedRows = listStoredThreadCompactedRowsByTurnIds(db, { + threadId: args.threadId, + turnIds: [...turnIds], + }); + return completedRows.length === 0 + ? [...args.rows] + : mergeStoredEventRowsById([...args.rows, ...completedRows]); +} + +function keepTimelineWindowOwnedContextCompactionRows( + db: DbConnection, + args: TimelineWindowRowsArgs & { sequenceStart: number }, +): StoredEventRow[] { + const turnIds = new Set(); + for (const row of args.rows) { + if ( + row.turnId !== null && + ((row.type === "item/started" && row.itemKind === "contextCompaction") || + row.type === "thread/compacted") + ) { + turnIds.add(row.turnId); + } + } + if (turnIds.size === 0) { + return [...args.rows]; + } + + const turnsOwnedByAnOlderWindow = new Set(); + for (const row of listStoredContextCompactionStartedRowsByTurnIds(db, { + threadId: args.threadId, + turnIds: [...turnIds], + })) { + if (row.turnId !== null && row.sequence < args.sequenceStart) { + turnsOwnedByAnOlderWindow.add(row.turnId); + } + } + if (turnsOwnedByAnOlderWindow.size === 0) { + return [...args.rows]; + } + return args.rows.filter( + (row) => + row.turnId === null || + !turnsOwnedByAnOlderWindow.has(row.turnId) || + !( + (row.itemKind === "contextCompaction" && + (row.type === "item/started" || row.type === "item/completed")) || + row.type === "thread/compacted" + ), + ); +} + +function findOwnedItemContextEndSequence( + rows: readonly StoredEventRow[], + beforeSequence: number, +): number | null { + const boundsByItem = new Map< + string, + { maxSequence: number; minSequence: number } + >(); + const compactionBoundsByTurn = new Map< + string, + { maxSequence: number; minSequence: number } + >(); + for (const row of rows) { + if ( + row.turnId !== null && + ((row.type === "item/started" && row.itemKind === "contextCompaction") || + row.type === "thread/compacted") + ) { + const bounds = compactionBoundsByTurn.get(row.turnId); + if (!bounds) { + compactionBoundsByTurn.set(row.turnId, { + maxSequence: row.sequence, + minSequence: row.sequence, + }); + } else { + bounds.maxSequence = Math.max(bounds.maxSequence, row.sequence); + bounds.minSequence = Math.min(bounds.minSequence, row.sequence); + } + } + if (row.itemId === null || row.itemKind === "backgroundTask") { + continue; + } + const key = scopedItemRefKey(storedEventRowItemRef(row)); + const bounds = boundsByItem.get(key); + if (!bounds) { + boundsByItem.set(key, { + maxSequence: row.sequence, + minSequence: row.sequence, + }); + continue; + } + bounds.maxSequence = Math.max(bounds.maxSequence, row.sequence); + bounds.minSequence = Math.min(bounds.minSequence, row.sequence); + } + + let contextEndSequence: number | null = null; + for (const bounds of [ + ...boundsByItem.values(), + ...compactionBoundsByTurn.values(), + ]) { + if ( + bounds.minSequence >= beforeSequence || + bounds.maxSequence < beforeSequence + ) { + continue; + } + contextEndSequence = Math.max( + contextEndSequence ?? beforeSequence, + bounds.maxSequence, + ); + } + return contextEndSequence; } /** @@ -1079,10 +1357,10 @@ interface ResolvedTimelineSegmentWindow { beforeSequence: number | undefined; byteWindowSequenceStart: number | null; /** - * Whether the window boundary needs whole-item lifecycle closure. - * See {@link ensureSequenceWindowWholeItemRows}. + * How the window boundary assigns straddling items, or null when unbounded. + * See {@link ensureTimelineWindowWholeItemRows}. */ - requiresWholeItemClosure: boolean; + wholeItemOwnership: "newest-window" | "starting-window" | null; /** Segments this page will actually return; ≤ `page.segmentLimit`. */ effectiveSegmentLimit: number; hasAnchors: boolean; @@ -1119,6 +1397,7 @@ function applyTimelineWindowByteBudget( return { ...args.window, byteWindowSequenceStart: floor.sequenceStart, + wholeItemOwnership: "newest-window", knownHasOlderSegments: hasOlderRows, oversizedEventPlaceholder: { id: `${args.threadId}:oversized-event:${floor.sequenceStart}`, @@ -1151,7 +1430,7 @@ function applyTimelineWindowByteBudget( return { ...args.window, byteWindowSequenceStart: floor.sequenceStart, - requiresWholeItemClosure: true, + wholeItemOwnership: "newest-window", sequenceWindowStart: { kind: "byte", sequenceStart: floor.sequenceStart, @@ -1292,7 +1571,7 @@ function resolveTimelineSegmentWindow( const noAnchors: ResolvedTimelineSegmentWindow = { beforeSequence: undefined, byteWindowSequenceStart: null, - requiresWholeItemClosure: false, + wholeItemOwnership: null, effectiveSegmentLimit: page.segmentLimit, hasAnchors: false, sequenceWindowStart: null, @@ -1378,8 +1657,10 @@ function resolveTimelineSegmentWindow( beforeSequence: cursor.anchorSeq, byteWindowSequenceStart: sequenceCursor?.kind === "byte" ? bounds.sequenceStart : null, - requiresWholeItemClosure: - sequenceCursor !== null || bounds.sequenceWindowStart !== null, + wholeItemOwnership: + sequenceCursor !== null || bounds.sequenceWindowStart !== null + ? "newest-window" + : "starting-window", effectiveSegmentLimit: bounds.effectiveSegmentLimit, hasAnchors: true, sequenceWindowStart: bounds.sequenceWindowStart, @@ -1410,7 +1691,8 @@ function resolveTimelineSegmentWindow( return { beforeSequence: undefined, byteWindowSequenceStart: null, - requiresWholeItemClosure: bounds.sequenceWindowStart !== null, + wholeItemOwnership: + bounds.sequenceWindowStart === null ? "starting-window" : "newest-window", effectiveSegmentLimit: bounds.effectiveSegmentLimit, hasAnchors: true, sequenceWindowStart: bounds.sequenceWindowStart, @@ -1428,6 +1710,8 @@ function selectStandardTimelineEventRows( page: ThreadTimelinePageRequest, eventBudget: number, maxInlineOutputChars: InlineOutputCharLimit, + includeNestedRows: boolean, + includeProviderUnhandledOperations: boolean, ): TimelineEventRowSelection { const window = applyTimelineWindowByteBudget(db, { maxInlineOutputChars, @@ -1457,26 +1741,37 @@ function selectStandardTimelineEventRows( threadId: thread.id, }; const windowRows = listStoredTimelineWindowEventRows(db, windowArgs); - const wholeItemWindowRows = window.requiresWholeItemClosure - ? ensureSequenceWindowWholeItemRows(db, { + const wholeItemWindow = window.wholeItemOwnership + ? ensureTimelineWindowWholeItemRows(db, { beforeSequence, maxInlineOutputChars, + ownership: window.wholeItemOwnership, rows: windowRows, sequenceStart, threadId: thread.id, }) - : windowRows; + : { rows: windowRows, sourceEndExtensions: [] }; + const wholeItemWindowRows = wholeItemWindow.rows; + const windowRowsWithCompactionLifecycle = + keepTimelineWindowOwnedContextCompactionRows(db, { + rows: ensureTimelineWindowContextCompactionRows(db, { + rows: wholeItemWindowRows, + threadId: thread.id, + }), + sequenceStart, + threadId: thread.id, + }); const selectedRowsWithTurnStarts = ensureTimelineWindowTurnStartedRows(db, { threadId: thread.id, - rows: wholeItemWindowRows, + rows: windowRowsWithCompactionLifecycle, }); - const selectedRowsWithTurnLifecycle = - window.byteWindowSequenceStart === null - ? selectedRowsWithTurnStarts - : ensureSequenceWindowTurnCompletedRows(db, { - threadId: thread.id, - rows: selectedRowsWithTurnStarts, - }); + const selectedRowsWithTurnLifecycle = ensureTimelineWindowTurnCompletedRows( + db, + { + threadId: thread.id, + rows: selectedRowsWithTurnStarts, + }, + ); const selectedRowsWithInWindowTaskState = ensureTimelineWindowBackgroundTaskStateRows(db, { threadId: thread.id, @@ -1507,24 +1802,138 @@ function selectStandardTimelineEventRows( rows: selectedRowsWithParentedContext.rows, }); const selectedRowsWithParentedTurnLifecycle = + ensureTimelineWindowTurnCompletedRows(db, { + threadId: thread.id, + rows: selectedRowsWithParentedTurnStarts, + }); + const semanticContextEndSequenceFromSelectedRows = + beforeSequence !== undefined && + window.wholeItemOwnership === "starting-window" + ? findOwnedItemContextEndSequence( + windowRowsWithCompactionLifecycle, + beforeSequence, + ) + : null; + const semanticContextEndSequence = + wholeItemWindow.sourceEndExtensions.length === 0 + ? semanticContextEndSequenceFromSelectedRows + : wholeItemWindow.sourceEndExtensions.reduce( + (maximum, extension) => Math.max(maximum, extension.sourceSeqEnd), + semanticContextEndSequenceFromSelectedRows ?? 0, + ); + const byteWindowSequenceEnd = window.byteWindowSequenceStart === null - ? selectedRowsWithParentedTurnStarts - : ensureSequenceWindowTurnCompletedRows(db, { - threadId: thread.id, - rows: selectedRowsWithParentedTurnStarts, + ? null + : (windowRows.at(-1)?.sequence ?? window.byteWindowSequenceStart); + const ownedItemBoundaryContextRows = + beforeSequence !== undefined && semanticContextEndSequence !== null + ? listStoredClientTurnRequestRowsByKeys(db, { + keys: listStoredClientTurnRequestIdsInRange(db, { + seqEnd: semanticContextEndSequence, + seqStart: beforeSequence, + threadId: thread.id, + }).map((requestId) => ({ requestId, threadId: thread.id })), + }).filter((row) => { + const event = parseStoredEvent(row); + return ( + event.type === "client/turn/requested" && event.initiator === "user" + ); + }) + : []; + const windowTurnIds = new Set( + windowRows.flatMap((row) => (row.turnId === null ? [] : [row.turnId])), + ); + const byteBoundaryContextEndSequence = + beforeSequence === undefined || window.byteWindowSequenceStart === null + ? null + : selectedRowsWithParentedTurnLifecycle.reduce( + (minimum, row) => + row.type === "turn/completed" && + row.turnId !== null && + windowTurnIds.has(row.turnId) && + row.sequence > beforeSequence + ? Math.min(minimum ?? row.sequence, row.sequence) + : minimum, + null, + ); + const byteBoundaryContextRowCandidate = + beforeSequence !== undefined && byteBoundaryContextEndSequence !== null + ? (listStoredClientTurnRequestRowsByKeys(db, { + keys: listStoredClientTurnRequestIdsInRange(db, { + seqEnd: byteBoundaryContextEndSequence, + seqStart: beforeSequence, + threadId: thread.id, + }).map((requestId) => ({ requestId, threadId: thread.id })), + }) + .filter((row) => { + const event = parseStoredEvent(row); + return ( + event.type === "client/turn/requested" && + event.initiator === "user" + ); + }) + .sort((left, right) => left.sequence - right.sequence)[0] ?? null) + : null; + const byteBoundaryContextRow = + byteBoundaryContextRowCandidate !== null && + byteWindowSequenceEnd !== null && + !hasStoredRootTimelineRowsBetween(db, { + afterSequence: byteWindowSequenceEnd, + beforeSequence: byteBoundaryContextRowCandidate.sequence, + excludedTypes: includeProviderUnhandledOperations + ? THREAD_TIMELINE_EXCLUDED_EVENT_TYPES + : [...THREAD_TIMELINE_EXCLUDED_EVENT_TYPES, "provider/unhandled"], + threadId: thread.id, + }) + ? byteBoundaryContextRowCandidate + : null; + const futureAssistantContextRows = + byteWindowSequenceEnd === null + ? [] + : [...windowTurnIds].flatMap((turnId) => { + const context = findStoredRootTurnAssistantMessageContextRows(db, { + afterSequence: byteWindowSequenceEnd, + beforeSequence: byteWindowSequenceEnd + 1, + maxInlineOutputChars: DEFAULT_MAX_INLINE_OUTPUT_CHARS, + threadId: thread.id, + turnId, + }); + return context.after && + !hasStoredRootNonAssistantItemBetween(db, { + afterSequence: byteWindowSequenceEnd, + beforeSequence: context.after.sequence, + threadId: thread.id, + }) && + !hasStoredUserClientTurnRequestBetween(db, { + afterSequence: byteWindowSequenceEnd, + beforeSequence: context.after.sequence, + threadId: thread.id, + }) + ? [context.after] + : []; }); + const futureBoundaryContextRows = mergeStoredEventRowsById([ + ...ownedItemBoundaryContextRows, + ...(byteBoundaryContextRow ? [byteBoundaryContextRow] : []), + ...futureAssistantContextRows, + ]); + const futureBoundaryAcceptedRows = selectClientRequestContextRows(db, { + rows: futureBoundaryContextRows, + threadId: thread.id, + }).acceptedRows; return { - byteWindowSequenceEnd: - window.byteWindowSequenceStart === null - ? null - : (wholeItemWindowRows.at(-1)?.sequence ?? - window.byteWindowSequenceStart), + byteWindowSequenceEnd, byteWindowSequenceStart: window.byteWindowSequenceStart, contextOnlyToolCallIds: - window.byteWindowSequenceStart === null + window.byteWindowSequenceStart === null || !includeNestedRows ? selectedRowsWithParentedContext.contextOnlyToolCallIds : new Set(), + contextOnlyMessageSeqs: new Set( + [...futureBoundaryContextRows, ...futureBoundaryAcceptedRows].map( + (row) => row.sequence, + ), + ), sequenceWindowStart: window.sequenceWindowStart, knownHasOlderSegments: window.knownHasOlderSegments, paginationPage: @@ -1536,7 +1945,23 @@ function selectStandardTimelineEventRows( }, responsePageKind: page.kind, oversizedEventPlaceholder: window.oversizedEventPlaceholder, - rows: selectedRowsWithParentedTurnLifecycle, + sourceEndExtensions: [ + ...wholeItemWindow.sourceEndExtensions, + ...(window.byteWindowSequenceStart === null + ? [] + : findTimelineWindowParentedSourceEndExtensions(db, { + includeProviderUnhandledOperations, + rows: windowRows, + threadId: thread.id, + })), + ], + rows: + futureBoundaryContextRows.length === 0 + ? selectedRowsWithParentedTurnLifecycle + : mergeStoredEventRowsById([ + ...selectedRowsWithParentedTurnLifecycle, + ...futureBoundaryContextRows, + ]), strategy: sequenceStart === 0 && beforeSequence === undefined ? "full" @@ -1561,8 +1986,38 @@ function buildSequencePageTimelineRows( (left, right) => left.sourceSeqStart - right.sourceSeqStart, ) : [...rows]; + const sourceEndByRowId = new Map(); + for (const extension of selection.sourceEndExtensions) { + const matchingRow = rowsWithPlaceholder + .filter( + (row) => + row.kind === "turn" && + row.turnId === extension.turnId && + extension.itemSourceSeqStart >= row.sourceSeqStart && + extension.itemSourceSeqStart <= row.sourceSeqEnd, + ) + .sort((left, right) => { + if (left.sourceSeqStart !== right.sourceSeqStart) { + return right.sourceSeqStart - left.sourceSeqStart; + } + return left.sourceSeqEnd - right.sourceSeqEnd; + })[0]; + if (!matchingRow) { + continue; + } + sourceEndByRowId.set( + matchingRow.id, + Math.max( + sourceEndByRowId.get(matchingRow.id) ?? matchingRow.sourceSeqEnd, + extension.sourceSeqEnd, + ), + ); + } if (selection.byteWindowSequenceStart === null) { - return rowsWithPlaceholder; + return rowsWithPlaceholder.map((row) => { + const sourceSeqEnd = sourceEndByRowId.get(row.id); + return sourceSeqEnd === undefined ? row : { ...row, sourceSeqEnd }; + }); } const suffix = @@ -1596,6 +2051,10 @@ function buildSequencePageTimelineRows( // page-unique id. return []; } + const extendedSourceSeqEnd = Math.max( + sourceSeqEnd, + sourceEndByRowId.get(row.id) ?? sourceSeqEnd, + ); return [ { ...row, @@ -1603,7 +2062,7 @@ function buildSequencePageTimelineRows( // coalesces only adjacent completed-turn fragments at a page seam; // visible conversation rows therefore remain semantic boundaries. id: `${row.id}${suffix}`, - sourceSeqEnd, + sourceSeqEnd: extendedSourceSeqEnd, sourceSeqStart, }, ]; @@ -1692,6 +2151,8 @@ function buildThreadTimelineInternal( options.page, options.eventBudget, options.maxInlineOutputChars, + includeNestedRows, + includeProviderUnhandledOperations, ), ); const rawEventRows = eventSelection.rows; @@ -1712,7 +2173,13 @@ function buildThreadTimelineInternal( const decodedRawEvents = measureThreadTimelineStage( profile, "event-json-decode", - () => rawEventRows.map((row) => toThreadEventWithMeta(row)), + () => + mergeStoredEventRowsById([ + ...rawEventRows, + ...acceptedClientRequestContextRows.acceptedRows, + ...acceptedClientRequestContextRows.rejectedRows, + ...acceptedClientRequestContextRows.turnStartedRows, + ]).map((row) => toThreadEventWithMeta(row)), ); if (profile) { profile.decodedEventCount = decodedRawEvents.length; @@ -1739,7 +2206,13 @@ function buildThreadTimelineInternal( ); profile.contextWindowEventRowCount = contextWindowUsageRows.length; } + const byteWindowSequenceStart = eventSelection.byteWindowSequenceStart; const byteWindowSequenceEnd = eventSelection.byteWindowSequenceEnd; + const contextOnlyMessageSeqs = new Set([ + ...eventSelection.contextOnlyMessageSeqs, + ...acceptedClientRequestContextRows.acceptedRows.map((row) => row.sequence), + ...acceptedClientRequestContextRows.rejectedRows.map((row) => row.sequence), + ]); const commonProjectionOptions = { contextOnlyCompletedTurnIds: byteWindowSequenceEnd === null @@ -1753,6 +2226,20 @@ function buildThreadTimelineInternal( : [], ), ), + contextOnlyMessageSeqs: + contextOnlyMessageSeqs.size === 0 ? undefined : contextOnlyMessageSeqs, + messageBoundsOnlyTurnIds: + byteWindowSequenceStart === null + ? undefined + : new Set( + rawEventRows.flatMap((row) => + row.type === "turn/started" && + row.turnId !== null && + row.sequence < byteWindowSequenceStart + ? [row.turnId] + : [], + ), + ), includeProviderUnhandledOperations, isLatestPage: options.page.kind === "latest", providerDisplayName: options.providerDisplayName, @@ -2070,6 +2557,19 @@ function buildTimelineTurnSummaryDetailsRange( acceptedInputRows: [...exactAcceptedInputRows, ...futureAcceptedInputRows], turnId: options.turnId, }); + const externalUserBoundaryRows = exactEventRows.filter((row) => { + const requestId = tryReadClientTurnRequestedRequestId(row); + if ( + requestId === null || + !acceptedInputRowsByTurn.acceptedClientRequestIdsForOtherTurns.has( + requestId, + ) + ) { + return false; + } + const event = parseStoredEvent(row); + return event.type === "client/turn/requested" && event.initiator === "user"; + }); const exactEventRowsForRequestedTurn = filterExactEventRowsForRequestedTurn({ acceptedClientRequestIdsForOtherTurns: acceptedInputRowsByTurn.acceptedClientRequestIdsForOtherTurns, @@ -2079,6 +2579,7 @@ function buildTimelineTurnSummaryDetailsRange( const eventRows = mergeStoredEventRowsById([ ...exactEventRowsForRequestedTurn.rows, ...acceptedInputRowsByTurn.requestedTurnRows, + ...externalUserBoundaryRows, ]); const hasTurnScopedRowsForRequestedTurn = eventRows.some( @@ -2126,7 +2627,7 @@ function buildTimelineTurnSummaryDetailsRange( }, useExactEventRowBounds: exactEventRowsForRequestedTurn.removedRows, }); - const assistantContextRows = useSemanticSelectionContext + const unboundedAssistantContextRows = useSemanticSelectionContext ? findStoredTurnAssistantMessageContextRows(db, { afterSequence: sourceRange.sourceSeqEnd, beforeSequence: sourceRange.sourceSeqStart, @@ -2135,25 +2636,42 @@ function buildTimelineTurnSummaryDetailsRange( turnId: options.turnId, }) : { after: null, before: null }; + // A following assistant is useful only when it is the continuation of this + // logical group. If a human request sits between the selected range and that + // assistant, carrying the assistant backward changes which earlier response + // is considered visible and makes expansion repeat that visible response. + const assistantContextRows = + unboundedAssistantContextRows.after !== null && + hasStoredUserClientTurnRequestBetween(db, { + afterSequence: sourceRange.sourceSeqEnd, + beforeSequence: unboundedAssistantContextRows.after.sequence, + threadId: thread.id, + }) + ? { ...unboundedAssistantContextRows, after: null } + : unboundedAssistantContextRows; // The same whole-item ownership rule the timeline window applies, for the // same reason. A byte cut can fall between an item's `item/started` and its // `item/completed`, and the timeline gives such an item to the newest slice. // Without the rule here, the older slice's details project the item from its // `item/started` row alone and render it "pending" after the turn finished. - const wholeItemEventRows = ensureSequenceWindowWholeItemRows(db, { + const wholeItemEventRows = ensureTimelineWindowWholeItemRows(db, { beforeSequence: detailsWindow.beforeSequence, itemOwnershipSequenceStart: options.resourceKind === "legacy-exact-range" ? undefined : (options.semanticSourceSeqStart ?? sourceRange.sourceSeqStart), maxInlineOutputChars: detailsInlineOutputLimit, + ownership: "newest-window", rows: mergeStoredEventRowsById([...requestedTurnStartedRows, ...eventRows]), sequenceStart: detailsWindow.sequenceStart, threadId: thread.id, - }); + }).rows; const eventRowsWithSemanticContext = mergeStoredEventRowsById([ ...(assistantContextRows.before ? [assistantContextRows.before] : []), - ...wholeItemEventRows, + ...ensureTimelineWindowContextCompactionRows(db, { + rows: wholeItemEventRows, + threadId: thread.id, + }), ...(assistantContextRows.after ? [assistantContextRows.after] : []), ]); // The floor queries measured the slice before closure, and closure backfills @@ -2161,9 +2679,10 @@ 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(eventRowsWithSemanticContext); - const eventRowsWithParentedChildren = ensureTimelineWindowParentedRows(db, { + const detailsEventDataBytes = byteLengthOfStoredEventRows( + eventRowsWithSemanticContext, + ); + const parentedContext = ensureTimelineWindowParentedRows(db, { maxInlineOutputChars: detailsInlineOutputLimit, outOfBoundsChildDataByteLimit: THREAD_TIMELINE_EVENT_DATA_BYTE_LIMIT - detailsEventDataBytes, @@ -2173,13 +2692,14 @@ function buildTimelineTurnSummaryDetailsRange( }, threadId: thread.id, rows: eventRowsWithSemanticContext, - }).rows; + }); + const eventRowsWithParentedChildren = parentedContext.rows; const eventRowsWithTurnStarts = ensureTimelineWindowTurnStartedRows(db, { threadId: thread.id, rows: eventRowsWithParentedChildren, }); const eventRowsWithTurnLifecycle = useSemanticSelectionContext - ? ensureSequenceWindowTurnCompletedRows(db, { + ? ensureTimelineWindowTurnCompletedRows(db, { threadId: thread.id, rows: eventRowsWithTurnStarts, }) @@ -2189,11 +2709,12 @@ function buildTimelineTurnSummaryDetailsRange( threadId: thread.id, rows: eventRowsWithTurnLifecycle, }); - const contextOnlyMessageSeqs = new Set( - [assistantContextRows.before, assistantContextRows.after].flatMap((row) => - row ? [row.sequence] : [], + const contextOnlyMessageSeqs = new Set([ + ...[assistantContextRows.before, assistantContextRows.after].flatMap( + (row) => (row ? [row.sequence] : []), ), - ); + ...externalUserBoundaryRows.map((row) => row.sequence), + ]); const projectionSourceSeqStart = useSemanticSelectionContext ? sourceRange.sourceSeqStart : eventRowsWithTurnStarts.reduce( @@ -2210,7 +2731,7 @@ function buildTimelineTurnSummaryDetailsRange( options: { allowContextExpandedMatch: useSemanticSelectionContext, contextOnlyCompletedTurnIds: - useSemanticSelectionContext && + options.resourceKind === "page" && eventRowsWithTurnLifecycle.some( (row) => row.type === "turn/completed" && @@ -2221,7 +2742,22 @@ function buildTimelineTurnSummaryDetailsRange( : undefined, contextOnlyMessageSeqs: contextOnlyMessageSeqs.size === 0 ? undefined : contextOnlyMessageSeqs, + contextOnlyToolCallIds: + options.resourceKind !== "logical-exact-range" || + parentedContext.contextOnlyToolCallIds.size === 0 + ? undefined + : parentedContext.contextOnlyToolCallIds, includeProviderUnhandledOperations, + messageBoundsOnlyTurnIds: + options.resourceKind === "logical-exact-range" && + eventRowsWithTurnLifecycle.some( + (row) => + row.type === "turn/completed" && + row.turnId === options.turnId && + row.sequence > sourceRange.sourceSeqEnd, + ) + ? new Set([options.turnId]) + : undefined, sourceSeqEnd: sourceRange.sourceSeqEnd, sourceSeqStart: projectionSourceSeqStart, providerDisplayName: options.providerDisplayName, @@ -2328,11 +2864,7 @@ function resolveCompletedTurnDetailBounds( `Cannot paginate details for incomplete turn ${selection.turnId}`, ); } - if ( - selection.sourceSeqStart > selection.sourceSeqEnd || - selection.sourceSeqStart < started.sequence || - selection.sourceSeqEnd > completed.sequence - ) { + if (selection.sourceSeqStart > selection.sourceSeqEnd) { throw new ApiError( 400, "invalid_request", @@ -2342,6 +2874,82 @@ function resolveCompletedTurnDetailBounds( return selection; } +function collectTimelineRowIds( + rows: readonly TimelineRow[], + ids = new Set(), +): Set { + for (const row of rows) { + ids.add(row.id); + if (row.kind === "turn" && row.children !== null) { + collectTimelineRowIds(row.children, ids); + } + if (row.kind === "work" && row.workKind === "delegation") { + collectTimelineRowIds(row.childRows, ids); + } + } + return ids; +} + +function filterTimelineRowsByIds( + rows: readonly TimelineRow[], + ids: ReadonlySet, +): TimelineRow[] { + return rows.flatMap((row): TimelineRow[] => { + if (!ids.has(row.id)) { + return []; + } + if (row.kind === "turn" && row.children !== null) { + return [ + { + ...row, + children: filterTimelineRowsByIds(row.children, ids), + }, + ]; + } + if (row.kind === "work" && row.workKind === "delegation") { + return [ + { + ...row, + childRows: filterTimelineRowsByIds(row.childRows, ids), + }, + ]; + } + return [row]; + }); +} + +function buildTimelineTurnDetailOwnershipIds( + db: DbConnection, + thread: Thread, + options: BuildTimelineTurnDetailsPageOptions, +): Set | null { + const structuralRows = listStoredTimelineWindowEventRows(db, { + beforeSequence: options.sourceSeqEnd + 1, + excludedTypes: THREAD_TIMELINE_EXCLUDED_EVENT_TYPES, + maxInlineOutputChars: 0, + sequenceStart: options.sourceSeqStart, + threadId: thread.id, + }); + const ownership = buildTimelineTurnSummaryDetailsRange(db, thread, { + includeProviderUnhandledOperations: + options.includeProviderUnhandledOperations, + preloadedEventRows: structuralRows, + providerDisplayName: options.providerDisplayName, + resourceKind: "logical-exact-range", + sourceSeqEnd: options.sourceSeqEnd, + sourceSeqStart: options.sourceSeqStart, + turnId: options.turnId, + }); + const firstOwnedSequence = ownership.rows.reduce( + (minimum, row) => Math.min(minimum, row.sourceSeqStart), + Number.POSITIVE_INFINITY, + ); + if (firstOwnedSequence !== options.sourceSeqStart) { + return null; + } + return collectTimelineRowIds(ownership.rows); +} + export function buildTimelineTurnDetailsPage( db: DbConnection, thread: Thread, @@ -2415,8 +3023,12 @@ export function buildTimelineTurnDetailsPage( sourceSeqStart, turnId: options.turnId, }); + const ownershipIds = buildTimelineTurnDetailOwnershipIds(db, thread, options); return { - rows: details.rows, + rows: + ownershipIds === null + ? details.rows + : filterTimelineRowsByIds(details.rows, ownershipIds), nextCursor: page.nextSequenceStart === null ? 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 0fc9c8e6ee..881f412ed3 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 @@ -18,6 +18,7 @@ import { import { LOCAL_WORKFLOW_TASK_TYPE } from "@bb/domain"; import type { DbConnection } from "@bb/db"; import type { + ThreadTimelineResponse, TimelinePaginationCursor, TimelineRow, } from "@bb/server-contract"; @@ -115,6 +116,8 @@ interface SeedOptions { longRunningItemIndexes?: readonly number[]; /** Character count for each completed command output. */ outputChars?: number; + /** Emit one root command before this nested item in the last turn. */ + rootCommandBeforeItem?: number; /** * Emit an output delta for each long-running item after every other item, so * the item's presence in a mid-turn window is deltas rather than lifecycle @@ -269,6 +272,47 @@ function seedTurns( }), }); } + if (isLastTurn && options.rootCommandBeforeItem === item) { + const itemId = `${turnId}-root-command`; + push({ + type: "item/started", + scope: turnScope(turnId), + providerThreadId, + itemId, + itemKind: "commandExecution", + parentToolCallId: null, + data: JSON.stringify({ + item: { + type: "commandExecution", + id: itemId, + command: "echo root", + cwd: "/tmp/test", + status: "pending", + approvalStatus: null, + }, + }), + }); + push({ + type: "item/completed", + scope: turnScope(turnId), + providerThreadId, + itemId, + itemKind: "commandExecution", + parentToolCallId: null, + data: JSON.stringify({ + item: { + type: "commandExecution", + id: itemId, + command: "echo root", + cwd: "/tmp/test", + status: "completed", + approvalStatus: null, + exitCode: 0, + aggregatedOutput: "root output", + }, + }), + }); + } const itemId = `${turnId}-item-${item}`; const command = options.commandChars === undefined @@ -1309,7 +1353,8 @@ describe("in-turn timeline windows", () => { }); const commandCallIds = new Set(); - const expandedCommandCallIds = new Set(); + let logicalSourceSeqStart = Number.POSITIVE_INFINITY; + let logicalSourceSeqEnd = 0; let cursor: TimelinePaginationCursor | null = null; let pages = 0; for (;;) { @@ -1326,18 +1371,11 @@ describe("in-turn timeline windows", () => { // range below the byte floor. expect(row.sourceSeqStart).toBeGreaterThan(4); } - const details = buildTimelineTurnSummaryDetails(db, thread, { - includeProviderUnhandledOperations: false, - sourceSeqEnd: row.sourceSeqEnd, - sourceSeqStart: row.sourceSeqStart, - turnId: row.turnId, - }); - const pageDetailCallIds = new Set(); - collectCommandCallIds(details.rows, pageDetailCallIds); - expect(pageDetailCallIds.size).toBeLessThan(BYTE_WINDOW_ITEM_COUNT); - for (const callId of pageDetailCallIds) { - expandedCommandCallIds.add(callId); - } + logicalSourceSeqStart = Math.min( + logicalSourceSeqStart, + row.sourceSeqStart, + ); + logicalSourceSeqEnd = Math.max(logicalSourceSeqEnd, row.sourceSeqEnd); } expect(page.profile.eventDataBytes, `page ${pages}`).toBeLessThanOrEqual( THREAD_TIMELINE_EVENT_DATA_BYTE_LIMIT, @@ -1352,9 +1390,70 @@ describe("in-turn timeline windows", () => { expect(pages).toBeGreaterThan(2); expect(commandCallIds.size).toBe(BYTE_WINDOW_ITEM_COUNT); + + const expandedCommandCallIds = new Set(); + let detailCursor: string | undefined; + let detailPages = 0; + do { + const details = buildTimelineTurnDetailsPage(db, thread, { + ...(detailCursor ? { cursor: detailCursor } : {}), + includeProviderUnhandledOperations: false, + sourceSeqEnd: logicalSourceSeqEnd, + sourceSeqStart: logicalSourceSeqStart, + turnId: "turn-1", + }); + detailPages += 1; + collectCommandCallIds(details.rows, expandedCommandCallIds); + detailCursor = details.nextCursor ?? undefined; + expect(detailPages).toBeLessThan(10); + } while (detailCursor); + + expect(detailPages).toBeGreaterThan(1); expect(expandedCommandCallIds.size).toBe(BYTE_WINDOW_ITEM_COUNT); }, 15_000); + it("does not fold an older delegation into a newer collapsed byte page", () => { + const { db, thread } = setup(); + seedTurns(db, thread, { + assistantBeforeItem: 100, + commandChars: 25_000, + completeLastTurn: true, + delegateLastTurn: true, + finalAssistant: true, + itemsPerTurn: [BYTE_WINDOW_ITEM_COUNT], + rootCommandBeforeItem: 150, + }); + + const turnRows: TimelineRow[] = []; + let cursor: TimelinePaginationCursor | null = null; + let pages = 0; + for (;;) { + const page: ThreadTimelineResponse = buildPage( + db, + thread, + LARGE_BUDGET, + cursor, + ).response; + pages += 1; + turnRows.push(...page.rows.filter((row) => row.kind === "turn")); + if (!page.timelinePage.hasOlderRows) { + break; + } + cursor = page.timelinePage.olderCursor; + expect(cursor).not.toBeNull(); + expect(pages).toBeLessThan(10); + } + + expect(pages).toBeGreaterThan(2); + expect(turnRows).toContainEqual( + expect.objectContaining({ + sourceSeqEnd: 307, + sourceSeqStart: 205, + summaryCount: 2, + }), + ); + }, 15_000); + it("returns a placeholder when one event exceeds the byte limit", () => { const { db, thread } = setup(); insertEvents(db, noopNotifier, [ diff --git a/apps/server/test/services/threads/timeline-pagination.test.ts b/apps/server/test/services/threads/timeline-pagination.test.ts index 98b0d3787b..78861ae49e 100644 --- a/apps/server/test/services/threads/timeline-pagination.test.ts +++ b/apps/server/test/services/threads/timeline-pagination.test.ts @@ -1,6 +1,7 @@ import { describe, expect, it } from "vitest"; import type { TimelineRow, + TimelineSystemRow, TimelineUserConversationRow, } from "@bb/server-contract"; import { paginateTimelineRows } from "../../../src/services/threads/timeline-pagination.js"; @@ -31,7 +32,50 @@ function userRow(args: { }; } +function interruptedRow(args: { id: string; seq: number }): TimelineSystemRow { + return { + id: args.id, + kind: "system", + systemKind: "operation", + operationKind: "thread-interrupted", + threadId: "thread-1", + turnId: null, + sourceSeqStart: args.seq, + sourceSeqEnd: args.seq, + startedAt: args.seq, + createdAt: args.seq, + completedAt: args.seq, + title: "Thread stopped by user request", + detail: null, + status: "interrupted", + }; +} + describe("paginateTimelineRows", () => { + it("keeps an out-of-order thread row with the user segment it follows", () => { + const rows: TimelineRow[] = [ + userRow({ id: "older-user", seq: 1, text: "older" }), + interruptedRow({ id: "interrupted", seq: 3 }), + userRow({ id: "newer-user", seq: 2, text: "newer" }), + ]; + + const page = paginateTimelineRows({ + sequenceWindowStart: null, + knownHasOlderSegments: null, + page: { kind: "latest", segmentLimit: 1 }, + rows, + }); + + expect(page.rows.map((row) => row.id)).toEqual([ + "newer-user", + "interrupted", + ]); + expect(page.olderCursor).toEqual({ + anchorId: "newer-user", + anchorSeq: 2, + }); + }); + it("keeps grouped user rows from one request in the same segment", () => { const rows: TimelineRow[] = [ userRow({ diff --git a/apps/server/test/services/threads/timeline-provider-input.test.ts b/apps/server/test/services/threads/timeline-provider-input.test.ts index 2f0af6d003..49d824d9fd 100644 --- a/apps/server/test/services/threads/timeline-provider-input.test.ts +++ b/apps/server/test/services/threads/timeline-provider-input.test.ts @@ -16,7 +16,10 @@ import { } from "@bb/db"; import type { DbConnection } from "@bb/db"; import type { TimelineRow } from "@bb/server-contract"; -import { buildThreadTimelineWithProfile } from "../../../src/services/threads/timeline.js"; +import { + buildThreadTimelineWithProfile, + buildTimelineTurnDetailsPage, +} from "../../../src/services/threads/timeline.js"; const providerThreadId = "pi-thread-1"; const PROCESS_EVENT = @@ -41,38 +44,28 @@ function setup(): { db: DbConnection; thread: Thread } { } type EventInput = Parameters[2][number]; +type EventWithoutSequence = Omit; +type RequestId = ReturnType; +type RequestTarget = + | { kind: "new-turn" } + | { kind: "thread-start" } + | { expectedTurnId: string; kind: "steer" }; -/** - * A thread the user started once, followed by a turn a Pi extension opened on - * its own: the only `client/turn/requested` is the first message, and the - * second turn's input is the provider-recorded `userMessage` item. - */ -function seedExtensionTriggeredTurn(db: DbConnection, thread: Thread): void { - const events: EventInput[] = []; - let sequence = 0; - const push = (event: Omit): void => { - sequence += 1; - events.push({ ...event, sequence, threadId: thread.id }); - }; - const clientRequestId = encodeClientTurnRequestIdNumber({ value: 1 }); - const providerEvent = ( - type: EventInput["type"], - turnId: string, - data: Record, - item?: { itemId: string; itemKind: EventInput["itemKind"] }, - ): void => { - push({ - type, - scope: turnScope(turnId), - providerThreadId, - itemId: item?.itemId ?? null, - itemKind: item?.itemKind ?? null, - parentToolCallId: null, - data: JSON.stringify(data), - }); - }; +const execution = { + model: "gpt-5", + serviceTier: "default", + reasoningLevel: "medium", + permissionMode: "full", + source: "client/turn/requested", +} as const; - push({ +function clientRequest( + requestId: RequestId, + text: string, + target: RequestTarget, + source: "spawn" | "tell" = "tell", +): EventWithoutSequence { + return { type: "client/turn/requested", scope: threadScope(), itemId: null, @@ -80,66 +73,220 @@ function seedExtensionTriggeredTurn(db: DbConnection, thread: Thread): void { parentToolCallId: null, data: JSON.stringify({ direction: "outbound", - source: "spawn", + source, initiator: "user", - request: { method: "thread/start", params: {} }, - requestId: clientRequestId, - senderThreadId: null, - input: [{ type: "text", text: "Reply only with ok.", mentions: [] }], - target: { kind: "thread-start" }, - execution: { - model: "gpt-5", - serviceTier: "default", - reasoningLevel: "medium", - permissionMode: "full", - source: "client/turn/requested", + request: { + method: target.kind === "thread-start" ? "thread/start" : "turn/start", + params: {}, }, + requestId, + senderThreadId: null, + input: [{ type: "text", text, mentions: [] }], + target, + execution, }), - }); - providerEvent("turn/started", "turn-1", {}); - providerEvent("turn/input/accepted", "turn-1", { clientRequestId }); - providerEvent( - "item/completed", - "turn-1", - { item: { type: "agentMessage", id: "assistant-1", text: "ok" } }, - { itemId: "assistant-1", itemKind: "agentMessage" }, - ); - providerEvent("turn/completed", "turn-1", { - status: "completed", + }; +} + +function turnEvent( + turnId: string, + type: EventInput["type"], + data: Record, + item?: { itemId: string; itemKind: EventInput["itemKind"] }, +): EventWithoutSequence { + return { + type, + scope: turnScope(turnId), providerThreadId, - }); + itemId: item?.itemId ?? null, + itemKind: item?.itemKind ?? null, + parentToolCallId: null, + data: JSON.stringify(data), + }; +} - providerEvent("turn/started", "turn-2", {}); - providerEvent( +function assistant( + turnId: string, + itemId: string, + text: string, +): EventWithoutSequence { + return turnEvent( + turnId, "item/completed", - "turn-2", - { - item: { - type: "userMessage", - id: "provider-input-1", - content: [{ type: "text", text: PROCESS_EVENT }], - }, - }, - { itemId: "provider-input-1", itemKind: "userMessage" }, + { item: { type: "agentMessage", id: itemId, text } }, + { itemId, itemKind: "agentMessage" }, ); - providerEvent( - "item/completed", - "turn-2", - { - item: { - type: "agentMessage", - id: "assistant-2", - text: "The process finished.", - }, - }, - { itemId: "assistant-2", itemKind: "agentMessage" }, +} + +function insertSeedEvents( + db: DbConnection, + thread: Thread, + events: readonly EventWithoutSequence[], +): void { + insertEvents( + db, + noopNotifier, + events.map((event, index) => ({ + ...event, + sequence: index + 1, + threadId: thread.id, + })), ); - providerEvent("turn/completed", "turn-2", { - status: "completed", - providerThreadId, - }); +} - insertEvents(db, noopNotifier, events); +/** + * A thread the user started once, followed by a turn a Pi extension opened on + * its own: the only `client/turn/requested` is the first message, and the + * second turn's input is the provider-recorded `userMessage` item. + */ +function seedExtensionTriggeredTurn(db: DbConnection, thread: Thread): void { + const clientRequestId = encodeClientTurnRequestIdNumber({ value: 1 }); + insertSeedEvents(db, thread, [ + clientRequest( + clientRequestId, + "Reply only with ok.", + { kind: "thread-start" }, + "spawn", + ), + turnEvent("turn-1", "turn/started", {}), + turnEvent("turn-1", "turn/input/accepted", { clientRequestId }), + assistant("turn-1", "assistant-1", "ok"), + turnEvent("turn-1", "turn/completed", { + status: "completed", + providerThreadId, + }), + turnEvent("turn-2", "turn/started", {}), + turnEvent( + "turn-2", + "item/completed", + { + item: { + type: "userMessage", + id: "provider-input-1", + content: [{ type: "text", text: PROCESS_EVENT }], + }, + }, + { itemId: "provider-input-1", itemKind: "userMessage" }, + ), + assistant("turn-2", "assistant-2", "The process finished."), + turnEvent("turn-2", "turn/completed", { + status: "completed", + providerThreadId, + }), + ]); +} + +function seedGroupedClientRequests(db: DbConnection, thread: Thread): void { + const requestId = encodeClientTurnRequestIdNumber({ value: 1 }); + const groupedRequestId = encodeClientTurnRequestIdNumber({ value: 2 }); + insertSeedEvents(db, thread, [ + clientRequest(requestId, "First grouped message", { kind: "new-turn" }), + clientRequest(groupedRequestId, "Second grouped message", { + kind: "new-turn", + }), + turnEvent("turn-1", "turn/started", {}), + turnEvent("turn-1", "turn/input/accepted", { + clientRequestId: requestId, + }), + turnEvent("turn-1", "turn/input/accepted", { + clientRequestId: groupedRequestId, + }), + turnEvent("turn-1", "turn/completed", { status: "completed" }), + ]); +} + +function seedAssistantCrossingSteer(db: DbConnection, thread: Thread): void { + const firstRequestId = encodeClientTurnRequestIdNumber({ value: 1 }); + const steerRequestId = encodeClientTurnRequestIdNumber({ value: 2 }); + const followingRequestId = encodeClientTurnRequestIdNumber({ value: 3 }); + insertSeedEvents(db, thread, [ + clientRequest(firstRequestId, "Initial message", { kind: "new-turn" }), + turnEvent("turn-1", "turn/started", {}), + turnEvent("turn-1", "turn/input/accepted", { + clientRequestId: firstRequestId, + }), + turnEvent( + "turn-1", + "item/started", + { item: { type: "agentMessage", id: "assistant-1", text: "" } }, + { itemId: "assistant-1", itemKind: "agentMessage" }, + ), + turnEvent( + "turn-1", + "item/agentMessage/delta", + { itemId: "assistant-1", delta: "Assistant before steer" }, + { itemId: "assistant-1", itemKind: null }, + ), + clientRequest(steerRequestId, "Steer while assistant is finishing", { + kind: "steer", + expectedTurnId: "turn-1", + }), + assistant("turn-1", "assistant-1", "Assistant before steer"), + turnEvent("turn-1", "turn/completed", { status: "completed" }), + clientRequest(followingRequestId, "Following message", { + kind: "new-turn", + }), + turnEvent("turn-2", "turn/started", {}), + turnEvent("turn-2", "turn/input/accepted", { + clientRequestId: steerRequestId, + }), + turnEvent("turn-2", "turn/completed", { status: "completed" }), + ]); +} + +function seedDelegationCrossingHumanBoundary( + db: DbConnection, + thread: Thread, +): void { + const firstRequestId = encodeClientTurnRequestIdNumber({ value: 1 }); + const followingRequestId = encodeClientTurnRequestIdNumber({ value: 2 }); + insertSeedEvents(db, thread, [ + clientRequest(firstRequestId, "Initial message", { kind: "new-turn" }), + turnEvent("turn-1", "turn/started", {}), + turnEvent("turn-1", "turn/input/accepted", { + clientRequestId: firstRequestId, + }), + assistant("turn-1", "assistant-1", "First response"), + turnEvent( + "turn-1", + "item/started", + { + item: { + type: "toolCall", + id: "delegation-1", + tool: "Agent", + arguments: { prompt: "Do the long task." }, + status: "pending", + }, + }, + { itemId: "delegation-1", itemKind: "toolCall" }, + ), + assistant("turn-1", "assistant-2", "Final response"), + turnEvent("turn-1", "turn/completed", { status: "completed" }), + clientRequest(followingRequestId, "Following human message", { + kind: "new-turn", + }), + turnEvent( + "turn-1", + "item/completed", + { + item: { + type: "toolCall", + id: "delegation-1", + tool: "Agent", + arguments: { prompt: "Do the long task." }, + result: "", + status: "completed", + }, + }, + { itemId: "delegation-1", itemKind: "toolCall" }, + ), + turnEvent("turn-2", "turn/started", {}), + turnEvent("turn-2", "turn/input/accepted", { + clientRequestId: followingRequestId, + }), + turnEvent("turn-2", "turn/completed", { status: "completed" }), + ]); } function conversationTexts(rows: readonly TimelineRow[]): string[] { @@ -155,6 +302,131 @@ function conversationTexts(rows: readonly TimelineRow[]): string[] { } describe("timeline pages with provider-recorded input", () => { + it("uses a human boundary crossed by an owned delegation as context", () => { + const { db, thread } = setup(); + seedDelegationCrossingHumanBoundary(db, thread); + + const { response } = buildThreadTimelineWithProfile(db, thread, { + eventBudget: 1_000_000, + includeProviderUnhandledOperations: false, + includeNestedRows: false, + maxInlineOutputChars: 32_000, + maxSeq: 0, + page: { + kind: "older", + beforeCursor: { + anchorId: `${thread.id}:user-seed:8`, + anchorSeq: 8, + }, + segmentLimit: 1, + }, + }); + + expect(conversationTexts(response.rows)).toEqual([ + "user:Initial message", + "assistant:First response", + "assistant:Final response", + ]); + const details = buildTimelineTurnDetailsPage(db, thread, { + includeProviderUnhandledOperations: false, + sourceSeqEnd: 9, + sourceSeqStart: 5, + turnId: "turn-1", + }); + expect(details.nextCursor).toBeNull(); + expect(conversationTexts(details.rows)).toEqual([]); + expect(details.rows).toEqual([ + expect.objectContaining({ + kind: "work", + workKind: "delegation", + callId: "delegation-1", + }), + ]); + }); + + it("keeps a straddling assistant in the segment where it started", () => { + const { db, thread } = setup(); + seedAssistantCrossingSteer(db, thread); + + const { response } = buildThreadTimelineWithProfile(db, thread, { + eventBudget: 1_000_000, + includeProviderUnhandledOperations: false, + includeNestedRows: false, + maxInlineOutputChars: 32_000, + maxSeq: 0, + page: { + kind: "older", + beforeCursor: { + anchorId: `${thread.id}:user-seed:9`, + anchorSeq: 9, + }, + segmentLimit: 1, + }, + }); + + expect(conversationTexts(response.rows)).toEqual([ + "user:Steer while assistant is finishing", + ]); + const olderCursor = response.timelinePage.olderCursor; + expect(olderCursor).not.toBeNull(); + if (olderCursor === null) { + throw new Error("Expected an older timeline cursor"); + } + + const { response: olderResponse } = buildThreadTimelineWithProfile( + db, + thread, + { + eventBudget: 1_000_000, + includeProviderUnhandledOperations: false, + includeNestedRows: false, + maxInlineOutputChars: 32_000, + maxSeq: 0, + page: { + kind: "older", + beforeCursor: olderCursor, + segmentLimit: 1, + }, + }, + ); + + expect(conversationTexts(olderResponse.rows)).toEqual([ + "user:Initial message", + "assistant:Assistant before steer", + ]); + }); + + it("keeps a grouped message accepted when its acceptance is beyond the page", () => { + const { db, thread } = setup(); + seedGroupedClientRequests(db, thread); + + const { response } = buildThreadTimelineWithProfile(db, thread, { + eventBudget: 1_000_000, + includeProviderUnhandledOperations: false, + includeNestedRows: false, + maxInlineOutputChars: 32_000, + maxSeq: 0, + page: { + kind: "older", + beforeCursor: { + anchorId: `${thread.id}:user-seed:2`, + anchorSeq: 2, + }, + segmentLimit: 1, + }, + }); + + expect(response.rows).toEqual([ + expect.objectContaining({ + kind: "conversation", + role: "user", + text: "First grouped message", + turnId: "turn-1", + turnRequest: expect.objectContaining({ status: "accepted" }), + }), + ]); + }); + it("keeps the user's earlier turn on the latest page and nests the provider input in its turn", () => { const { db, thread } = setup(); seedExtensionTriggeredTurn(db, thread); diff --git a/packages/client-core/src/timeline/timeline-merge.ts b/packages/client-core/src/timeline/timeline-merge.ts index 3dc35d82a4..2d3f4e21e9 100644 --- a/packages/client-core/src/timeline/timeline-merge.ts +++ b/packages/client-core/src/timeline/timeline-merge.ts @@ -118,8 +118,7 @@ function canCoalesceCompletedTurnPageSeam( older.threadId === newer.threadId && older.turnId === newer.turnId && older.completedAt !== null && - newer.completedAt !== null && - older.sourceSeqEnd < newer.sourceSeqStart + newer.completedAt !== null ); } @@ -147,7 +146,7 @@ function coalesceCompletedTurnPageSeam( older.children === null && newer.children === null ? null : children, completedAt: Math.max(older.completedAt, newer.completedAt), createdAt: Math.min(older.createdAt, newer.createdAt), - sourceSeqEnd: newer.sourceSeqEnd, + sourceSeqEnd: Math.max(older.sourceSeqEnd, newer.sourceSeqEnd), sourceSeqStart: older.sourceSeqStart, startedAt: Math.min(older.startedAt, newer.startedAt), summaryCount: older.summaryCount + newer.summaryCount, @@ -269,10 +268,7 @@ export function mergeTimelineTurnDetailPages( : Math.max(existing.completedAt, row.completedAt), createdAt: Math.min(existing.createdAt, row.createdAt), sourceSeqEnd: Math.max(existing.sourceSeqEnd, row.sourceSeqEnd), - sourceSeqStart: Math.min( - existing.sourceSeqStart, - row.sourceSeqStart, - ), + sourceSeqStart: Math.min(existing.sourceSeqStart, row.sourceSeqStart), startedAt: Math.min(existing.startedAt, row.startedAt), }; continue; diff --git a/packages/db/src/data/events.ts b/packages/db/src/data/events.ts index 8904a0e266..46d80b83fa 100644 --- a/packages/db/src/data/events.ts +++ b/packages/db/src/data/events.ts @@ -1148,6 +1148,25 @@ export interface ListStoredClientTurnRequestIdsInRangeArgs { threadId: string; } +export interface HasStoredRootTimelineRowsBetweenArgs { + afterSequence: number; + beforeSequence: number; + excludedTypes: readonly ThreadEventType[]; + threadId: string; +} + +export interface HasStoredRootNonAssistantItemBetweenArgs { + afterSequence: number; + beforeSequence: number; + threadId: string; +} + +export interface HasStoredUserClientTurnRequestBetweenArgs { + afterSequence: number; + beforeSequence: number; + threadId: string; +} + export interface GetStoredTurnRequestEventForTurnArgs { threadId: string; turnId: string; @@ -1190,6 +1209,16 @@ export interface ListStoredTurnCompletedRowsByTurnIdsArgs { turnIds: readonly string[]; } +export interface ListStoredThreadCompactedRowsByTurnIdsArgs { + threadId: string; + turnIds: readonly string[]; +} + +export interface ListStoredContextCompactionStartedRowsByTurnIdsArgs { + threadId: string; + turnIds: readonly string[]; +} + export interface HasStoredTurnStartedArgs { threadId: string; turnId: string; @@ -1588,19 +1617,23 @@ export function findStoredEventRow( ); } -export function findStoredTurnAssistantMessageContextRows( +function findStoredTurnAssistantMessageContextRowsInternal( db: DbConnection, args: FindStoredTurnAssistantMessageContextRowsArgs, + rootOnly: boolean, ): { after: StoredEventRow | null; before: StoredEventRow | null } { const fields = storedEventRowFieldsWithInlineOutputLimit( args.maxInlineOutputChars, ); - const scope = [ + const scope: SQL[] = [ eq(events.threadId, args.threadId), eq(events.turnId, args.turnId), eq(events.type, "item/completed"), eq(events.itemKind, "agentMessage"), ]; + if (rootOnly) { + scope.push(isNull(events.parentToolCallId)); + } const before = db .select(fields) @@ -1620,6 +1653,21 @@ export function findStoredTurnAssistantMessageContextRows( return { after, before }; } +export function findStoredTurnAssistantMessageContextRows( + db: DbConnection, + args: FindStoredTurnAssistantMessageContextRowsArgs, +): { after: StoredEventRow | null; before: StoredEventRow | null } { + return findStoredTurnAssistantMessageContextRowsInternal(db, args, false); +} + +/** Nearest root assistant messages on either side of a sequence range. */ +export function findStoredRootTurnAssistantMessageContextRows( + db: DbConnection, + args: FindStoredTurnAssistantMessageContextRowsArgs, +): { after: StoredEventRow | null; before: StoredEventRow | null } { + return findStoredTurnAssistantMessageContextRowsInternal(db, args, true); +} + export function listStoredEventRowsByParentToolCallIds( db: DbConnection, args: ListStoredEventRowsByParentToolCallIdsArgs, @@ -1869,6 +1917,40 @@ export interface ListStoredItemLifecycleRowsByItemsArgs { threadId: string; } +export interface ListStoredEventRowsAtSequencesArgs { + /** See {@link InlineOutputCharLimit}. */ + maxInlineOutputChars: InlineOutputCharLimit; + sequences: readonly number[]; + threadId: string; +} + +/** Reads exact event rows by their thread-unique sequence numbers. */ +export function listStoredEventRowsAtSequences( + db: DbConnection, + args: ListStoredEventRowsAtSequencesArgs, +): StoredEventRow[] { + return queryInSqliteVariableBatches({ + dedupeKey: String, + fixedVariableCount: 1, + queryBatch: (sequences) => + db + .select( + storedEventRowFieldsWithInlineOutputLimit(args.maxInlineOutputChars), + ) + .from(events) + .where( + and( + eq(events.threadId, args.threadId), + inArray(events.sequence, [...sequences]), + ), + ) + .orderBy(events.sequence) + .all(), + values: args.sequences, + variableCountPerValue: 1, + }); +} + /** * Every `item/started` and `item/completed` row for the given items, wherever * they sit in the thread. @@ -1968,6 +2050,89 @@ export function listStoredClientTurnRequestIdsInRange( return rows.map((row) => clientTurnRequestIdSchema.parse(row.requestId)); } +/** + * Whether a sequence gap contains an event that can contribute to the root + * timeline projection. This deliberately selects only an id and stops at the + * first match: callers use it to validate boundary context without reading the + * potentially large event payloads between two transport windows. + */ +export function hasStoredRootTimelineRowsBetween( + db: DbConnection, + args: HasStoredRootTimelineRowsBetweenArgs, +): boolean { + const conditions: SQL[] = [ + eq(events.threadId, args.threadId), + gt(events.sequence, args.afterSequence), + lt(events.sequence, args.beforeSequence), + isNull(events.parentToolCallId), + isNotNestedTurnUsageEvent, + ]; + if (args.excludedTypes.length > 0) { + conditions.push(notInArray(events.type, [...args.excludedTypes])); + } + + return ( + db + .select({ id: events.id }) + .from(events) + .where(and(...conditions)) + .limit(1) + .get() !== undefined + ); +} + +/** + * Whether a sequence gap contains root work that separates two assistant + * messages in the completed-turn projection. Reasoning is projection state, + * not a timeline message, and agent messages are the boundary being tested. + */ +export function hasStoredRootNonAssistantItemBetween( + db: DbConnection, + args: HasStoredRootNonAssistantItemBetweenArgs, +): boolean { + return ( + db + .select({ id: events.id }) + .from(events) + .where( + and( + eq(events.threadId, args.threadId), + gt(events.sequence, args.afterSequence), + lt(events.sequence, args.beforeSequence), + isNull(events.parentToolCallId), + isNotNestedTurnUsageEvent, + isNotNull(events.itemKind), + notInArray(events.itemKind, ["agentMessage", "reasoning"]), + ), + ) + .limit(1) + .get() !== undefined + ); +} + +/** Whether a user-authored client request occurs strictly inside a range. */ +export function hasStoredUserClientTurnRequestBetween( + db: DbConnection, + args: HasStoredUserClientTurnRequestBetweenArgs, +): boolean { + return ( + db + .select({ id: events.id }) + .from(events) + .where( + and( + eq(events.threadId, args.threadId), + eq(events.type, "client/turn/requested"), + gt(events.sequence, args.afterSequence), + lt(events.sequence, args.beforeSequence), + sql`json_extract(${events.data}, '$.initiator') = 'user'`, + ), + ) + .limit(1) + .get() !== undefined + ); +} + export function getStoredTurnRequestEventForTurn( db: DbQueryConnection, args: GetStoredTurnRequestEventForTurnArgs, @@ -2159,6 +2324,51 @@ export function listStoredTurnCompletedRowsByTurnIds( .all(); } +export function listStoredThreadCompactedRowsByTurnIds( + db: DbConnection, + args: ListStoredThreadCompactedRowsByTurnIdsArgs, +): StoredEventRow[] { + if (args.turnIds.length === 0) { + return []; + } + + return db + .select(storedEventRowFields) + .from(events) + .where( + and( + eq(events.threadId, args.threadId), + eq(events.type, "thread/compacted"), + inArray(events.turnId, [...args.turnIds]), + ), + ) + .orderBy(events.sequence) + .all(); +} + +export function listStoredContextCompactionStartedRowsByTurnIds( + db: DbConnection, + args: ListStoredContextCompactionStartedRowsByTurnIdsArgs, +): StoredEventRow[] { + if (args.turnIds.length === 0) { + return []; + } + + return db + .select(storedEventRowFields) + .from(events) + .where( + and( + eq(events.threadId, args.threadId), + eq(events.type, "item/started"), + eq(events.itemKind, "contextCompaction"), + inArray(events.turnId, [...args.turnIds]), + ), + ) + .orderBy(events.sequence) + .all(); +} + export interface ListLatestBackgroundTaskStateRowsByItemIdsArgs { itemIds: readonly string[]; threadId: string; diff --git a/packages/db/src/data/index.ts b/packages/db/src/data/index.ts index 7595d8afec..3e3e512e0d 100644 --- a/packages/db/src/data/index.ts +++ b/packages/db/src/data/index.ts @@ -255,9 +255,14 @@ export { listEvents, listRecentStoredEventRows, listStoredConversationOutlineEventRows, + listStoredContextCompactionStartedRowsByTurnIds, listTimelineSegmentAnchorsDescending, findTimelineWindowBudgetFloorSequence, findStoredTimelineWindowByteBudgetFloor, + findStoredRootTurnAssistantMessageContextRows, + hasStoredRootNonAssistantItemBetween, + hasStoredRootTimelineRowsBetween, + hasStoredUserClientTurnRequestBetween, readStoredTimelineWindowForwardPage, getStoredEventRowsByParentToolCallIdsDataBytes, findUnfinishedTurnCoveringSequence, @@ -266,6 +271,7 @@ export { listStoredClientTurnRequestIdsInRange, listStoredClientTurnRequestRowsByKeys, listStoredEventRowsByParentToolCallIds, + listStoredEventRowsAtSequences, listStoredEventRows, isTimelineCursorSequencePresent, listItemEventSpansByItems, @@ -275,6 +281,7 @@ export { listStoredThreadProvisioningRowsByProvisioningId, listStoredTimelineWindowEventRows, listStoredDelegatingItemRowsByItemIds, + listStoredThreadCompactedRowsByTurnIds, listStoredTurnInputAcceptedRowsByClientRequestIds, listStoredTurnRejectedRowsByClientRequestIds, listStoredTurnCompletedRowsByTurnIds, diff --git a/packages/thread-view/src/build-thread-timeline.ts b/packages/thread-view/src/build-thread-timeline.ts index 0b8c1b035b..40dd16de21 100644 --- a/packages/thread-view/src/build-thread-timeline.ts +++ b/packages/thread-view/src/build-thread-timeline.ts @@ -74,6 +74,7 @@ type ThreadTimelineTurnMessageDetail = "summary" | "full"; interface ThreadTimelineFromEventsBaseOptions { contextOnlyCompletedTurnIds?: ReadonlySet; + contextOnlyMessageSeqs?: ReadonlySet; contextOnlyToolCallIds?: ReadonlySet; includeProviderUnhandledOperations: boolean; /** @@ -115,6 +116,7 @@ interface ThreadTimelineFromEventsBaseOptions { interface ThreadTimelineFromEventsOptions extends ThreadTimelineFromEventsBaseOptions { includeNestedRows: boolean; + messageBoundsOnlyTurnIds?: ReadonlySet; turnMessageDetail: ThreadTimelineTurnMessageDetail; } @@ -146,7 +148,9 @@ interface BuildThreadTimelineTurnDetailsFromEventsOptions extends ThreadTimeline allowContextExpandedMatch?: boolean; contextOnlyCompletedTurnIds?: ReadonlySet; contextOnlyMessageSeqs?: ReadonlySet; + contextOnlyToolCallIds?: ReadonlySet; includeProviderUnhandledOperations: boolean; + messageBoundsOnlyTurnIds?: ReadonlySet; providerDisplayName?: string; threadStatus: Thread["status"]; /** See {@link ThreadTimelineFromEventsBaseOptions.threadName}. */ @@ -177,6 +181,7 @@ interface BuildTurnRowsArgs { contextOnlyCompletedTurnIds?: ReadonlySet; contextOnlyMessageSeqs?: ReadonlySet; includeNestedRows: boolean; + messageBoundsOnlyTurnIds?: ReadonlySet; rowIdPrefix: string; turn: EventProjectionTurn; workspaceRoot: string | null; @@ -214,6 +219,7 @@ interface BuildTimelineRowsOptions { contextOnlyCompletedTurnIds?: ReadonlySet; contextOnlyMessageSeqs?: ReadonlySet; includeNestedRows: boolean; + messageBoundsOnlyTurnIds?: ReadonlySet; rowIdPrefix: string; workspaceRoot: string | null; } @@ -1238,6 +1244,7 @@ function buildTurnRows({ contextOnlyCompletedTurnIds, contextOnlyMessageSeqs, includeNestedRows, + messageBoundsOnlyTurnIds, rowIdPrefix, turn, workspaceRoot, @@ -1261,6 +1268,7 @@ function buildTurnRows({ turn, contextOnlyCompletedTurnIds?.has(turn.turnId) === true, contextOnlyMessageSeqs, + messageBoundsOnlyTurnIds?.has(turn.turnId) === true, ); const terminalRows = terminalMessages.flatMap((message) => convertMessage(message, { includeNestedRows, rowIdPrefix, workspaceRoot }), @@ -1296,6 +1304,13 @@ function findMatchingTurnSummaryRow( return exact ?? null; } + const matchingStart = turnRows.find( + (row) => row.sourceSeqStart === range.sourceSeqStart, + ); + if (matchingStart) { + return matchingStart; + } + // Lifecycle closure and context-only rows can shift a semantic group's // projected bounds in either direction. A parent shell can widen them, // while removing a boundary message can narrow them. Prefer the group with @@ -1403,6 +1418,33 @@ function orderRowsAfterExternalUserBoundary( return [...rows.slice(0, suffixStartIndex), ...orderedSuffix]; } +/** + * Thread-scoped system events are separate projection entries, so an event + * emitted during a turn is initially appended after all rows from that turn. + * Move only such late system rows back before the first source-newer row. This + * leaves established semantic ordering intact (notably provisioning before an + * initial turn summary) because a system row that is already early never + * moves. + */ +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, @@ -1421,6 +1463,7 @@ function buildTimelineRows( buildTurnRows({ contextOnlyCompletedTurnIds: options.contextOnlyCompletedTurnIds, contextOnlyMessageSeqs: options.contextOnlyMessageSeqs, + messageBoundsOnlyTurnIds: options.messageBoundsOnlyTurnIds, turn: entry.turn, includeNestedRows, rowIdPrefix: options.rowIdPrefix, @@ -1433,9 +1476,11 @@ function buildTimelineRows( } } - return orderRowsAfterExternalUserBoundary( - rows, - collectExternalUserBoundarySeqs(projection), + return restoreLateSystemRowSourceOrder( + orderRowsAfterExternalUserBoundary( + rows, + collectExternalUserBoundarySeqs(projection), + ), ); } @@ -1457,7 +1502,9 @@ export function buildThreadTimelineFromEvents( const rows = [ ...buildTimelineRows(projection, { contextOnlyCompletedTurnIds: args.options.contextOnlyCompletedTurnIds, + contextOnlyMessageSeqs: args.options.contextOnlyMessageSeqs, includeNestedRows: args.options.includeNestedRows, + messageBoundsOnlyTurnIds: args.options.messageBoundsOnlyTurnIds, rowIdPrefix: ROOT_TIMELINE_ROW_ID_PREFIX, workspaceRoot: args.options.workspaceRoot, }), @@ -1466,7 +1513,14 @@ export function buildThreadTimelineFromEvents( args.events, args.options, ), - ]; + ].filter( + (row) => + !( + row.kind === "conversation" && + row.role === "user" && + args.options.contextOnlyMessageSeqs?.has(row.sourceSeqStart) === true + ), + ); return { activePromptMode: !args.options.isLatestPage @@ -1511,6 +1565,7 @@ function buildThreadTimelineTurnDetailRows( args: BuildThreadTimelineTurnDetailsFromEventsArgs, ): TimelineRow[] { const projection = buildEventProjectionEntries(args.events, { + contextOnlyToolCallIds: args.options.contextOnlyToolCallIds, includeProviderUnhandledOperations: args.options.includeProviderUnhandledOperations, providerDisplayName: args.options.providerDisplayName, @@ -1522,9 +1577,16 @@ function buildThreadTimelineTurnDetailRows( contextOnlyCompletedTurnIds: args.options.contextOnlyCompletedTurnIds, contextOnlyMessageSeqs: args.options.contextOnlyMessageSeqs, includeNestedRows: true, + messageBoundsOnlyTurnIds: args.options.messageBoundsOnlyTurnIds, rowIdPrefix: ROOT_TIMELINE_ROW_ID_PREFIX, workspaceRoot: args.options.workspaceRoot, - }); + }).filter( + (row) => + !( + row.kind === "conversation" && + args.options.contextOnlyMessageSeqs?.has(row.sourceSeqStart) === true + ), + ); } export function buildThreadTimelineTurnDetailsFromEvents( diff --git a/packages/thread-view/src/completed-turn-grouping.ts b/packages/thread-view/src/completed-turn-grouping.ts index edfbb5d1ac..05aefae689 100644 --- a/packages/thread-view/src/completed-turn-grouping.ts +++ b/packages/thread-view/src/completed-turn-grouping.ts @@ -191,13 +191,14 @@ function groupCompletedTurnSummaryMessages( summaryMessages, terminalMessage, ); - const selectedSummaryMessages = contextOnlyMessageSeqs - ? summaryMessages.filter( - (message) => - !contextOnlyMessageSeqs.has(message.sourceSeqStart) && - !contextOnlyMessageSeqs.has(message.sourceSeqEnd), - ) - : summaryMessages; + const isContextOnlyMessage = (message: EventProjectionMessage): boolean => + contextOnlyMessageSeqs?.has(message.sourceSeqStart) === true || + contextOnlyMessageSeqs?.has(message.sourceSeqEnd) === true; + const selectedSummaryMessages = summaryMessages.filter( + (message) => !isContextOnlyMessage(message), + ); + const removedContextOnlyMessage = + selectedSummaryMessages.length !== summaryMessages.length; if ( useTurnBounds && externalBoundarySeqs.length === 0 && @@ -212,7 +213,7 @@ function groupCompletedTurnSummaryMessages( rowIdSegmentIndex: null, sourceBounds: "turn", sourceMessages: selectedSummaryMessages, - summaryCount: contextOnlyMessageSeqs + summaryCount: removedContextOnlyMessage ? getProjectionSummaryCount(selectedSummaryMessages, undefined) : turn.summaryCount, }, @@ -282,8 +283,14 @@ function groupCompletedTurnSummaryMessages( } } - for (const message of selectedSummaryMessages) { + for (const message of summaryMessages) { flushExternalBoundariesBefore(message); + if (isContextOnlyMessage(message)) { + if (message.kind === "user" && message.initiator === "user") { + flushGroupedMessages(true); + } + continue; + } if (visibleResponseIds.has(message.id)) { flushGroupedMessages(); items.push({ @@ -317,6 +324,7 @@ export function groupCompletedTurnMessages( turn: EventProjectionTurn, completionIsContextOnly = false, contextOnlyMessageSeqs?: ReadonlySet, + messageBoundsOnly = false, ): CompletedTurnMessageGroups { const messages = turn.messages ?? []; const { summaryMessages, terminalMessages, trailingMessages } = @@ -330,7 +338,7 @@ export function groupCompletedTurnMessages( turn, summaryMessages, terminalMessages[0], - !completionIsContextOnly, + !completionIsContextOnly && !messageBoundsOnly, contextOnlyMessageSeqs, ), ), diff --git a/packages/thread-view/test/build-thread-timeline.test.ts b/packages/thread-view/test/build-thread-timeline.test.ts index 799b987de7..e8eca252df 100644 --- a/packages/thread-view/test/build-thread-timeline.test.ts +++ b/packages/thread-view/test/build-thread-timeline.test.ts @@ -2083,6 +2083,48 @@ describe("buildThreadTimelineFromEvents", () => { ]); }); + it("orders a thread error emitted during a turn before source-newer turn rows", () => { + const rows = buildTimelineRows([ + turnStartedEvent({ seq: 1 }), + toolCallItemEvent({ + seq: 2, + tool: "read_status", + type: "item/completed", + }), + systemErrorEvent({ + code: "provider_turn_start_timeout", + message: "The provider accepted a turn but did not start it", + seq: 3, + }), + { + event: { + type: "item/completed", + threadId: "thread-1", + providerThreadId: "provider-thread-1", + scope: turnScope("turn-1"), + item: { + type: "agentMessage", + id: "assistant-1", + text: "Still working", + }, + }, + meta: { id: "event-4", seq: 4, createdAt: 4 }, + }, + turnCompletedEvent({ seq: 5 }), + ]); + + expect( + rows.map((row) => ({ + kind: row.kind, + sourceSeqStart: row.sourceSeqStart, + })), + ).toEqual([ + { kind: "turn", sourceSeqStart: 1 }, + { kind: "system", sourceSeqStart: 3 }, + { kind: "conversation", sourceSeqStart: 4 }, + ]); + }); + it("uses legacy provider error detail as the title for generic provider errors", () => { const rows = buildTimelineRows([ turnStartedEvent({ seq: 1 }), diff --git a/packages/thread-view/test/completed-turn-grouping.test.ts b/packages/thread-view/test/completed-turn-grouping.test.ts index 0a0768dc37..6cdd179bad 100644 --- a/packages/thread-view/test/completed-turn-grouping.test.ts +++ b/packages/thread-view/test/completed-turn-grouping.test.ts @@ -199,6 +199,28 @@ describe("groupCompletedTurnMessages", () => { ]); }); + it("uses message bounds without hiding the terminal response", () => { + const command = commandMessage({ id: "command", seq: 2 }); + const response = assistantMessage({ id: "response", seq: 3 }); + const groups = groupCompletedTurnMessages( + completedTurn([command, response], response), + false, + undefined, + true, + ); + + expect(groups.summaryItems).toMatchObject([ + { + kind: "summary", + startedAt: 2, + completedAt: null, + sourceBounds: "messages", + sourceMessages: [command], + }, + ]); + expect(groups.terminalMessages).toEqual([response]); + }); + it("keeps an assistant response visible when more assistant text follows it directly", () => { const answer = assistantMessage({ id: "answer", seq: 1 }); const hookReply = assistantMessage({ id: "hook-reply", seq: 2 }); From f29e66b5a4bca0fbfbec79c7253a4514891b3682 Mon Sep 17 00:00:00 2001 From: Michael Yong Date: Tue, 25 Aug 2026 22:12:29 -0700 Subject: [PATCH 3/4] Limit timeline boundary context queries --- apps/server/src/services/threads/timeline.ts | 59 +++++++++++++++++--- 1 file changed, 52 insertions(+), 7 deletions(-) diff --git a/apps/server/src/services/threads/timeline.ts b/apps/server/src/services/threads/timeline.ts index df0f5d54d3..6a1e2fa7f7 100644 --- a/apps/server/src/services/threads/timeline.ts +++ b/apps/server/src/services/threads/timeline.ts @@ -965,6 +965,10 @@ function ensureTimelineWindowWholeItemRows( // turns (a resumed ACP session restarts its synthetic id counter), and a // thread-wide span for such an id makes every window disown the item. const windowItems = new Map(); + const itemLifecycle = new Map< + string, + { hasCompleted: boolean; hasStarted: boolean } + >(); for (const row of args.rows) { if ( row.itemId !== null && @@ -972,19 +976,43 @@ function ensureTimelineWindowWholeItemRows( row.sequence >= args.sequenceStart ) { const ref = storedEventRowItemRef(row); - windowItems.set(scopedItemRefKey(ref), ref); + const key = scopedItemRefKey(ref); + windowItems.set(key, ref); + const lifecycle = itemLifecycle.get(key) ?? { + hasCompleted: false, + hasStarted: false, + }; + if (row.type === "item/started") { + lifecycle.hasStarted = true; + } + if (row.type === "item/completed") { + lifecycle.hasCompleted = true; + } + itemLifecycle.set(key, lifecycle); } } if (windowItems.size === 0) { return { rows: [...args.rows], sourceEndExtensions: [] }; } - // Spans, not lifecycle rows. An item emits between its start and its end — - // output deltas, reasoning text, tool progress — and an unfinished item has - // no end at all, so "does this item reach past the cut" cannot be answered - // from `item/started` and `item/completed`. + // Fully contained lifecycle pairs cannot cross either edge. Ask SQLite for + // global spans only when the selected window is missing an endpoint. This + // keeps ordinary completed commands on the bounded window read instead of + // putting every item id from every older page into another GROUP BY query. + // Delta-only provider items remain ambiguous and therefore stay in this set. + const ambiguousItems = [...windowItems].flatMap(([key, ref]) => { + const lifecycle = itemLifecycle.get(key); + return lifecycle?.hasStarted && lifecycle.hasCompleted ? [] : [ref]; + }); + if (ambiguousItems.length === 0) { + return { rows: [...args.rows], sourceEndExtensions: [] }; + } + + // Spans, not only lifecycle rows. An ambiguous item can emit output deltas, + // reasoning text, or tool progress between its endpoints, and an unfinished + // item has no end at all. const spans = listItemEventSpansByItems(db, { - items: [...windowItems.values()], + items: ambiguousItems, threadId: args.threadId, }); const itemKeysOwnedByNewerWindow = new Set(); @@ -1887,10 +1915,27 @@ function selectStandardTimelineEventRows( }) ? byteBoundaryContextRowCandidate : null; + // Only a turn whose lifecycle crosses the byte window's upper edge can + // contribute a root assistant message beyond that edge. Querying every turn + // merely present in the page repeats two nearest-message lookups per turn, + // even though completed in-window turns cannot possibly supply this + // boundary context. + const byteBoundaryTurnIds = + byteWindowSequenceEnd === null + ? [] + : [...windowTurnIds].filter((turnId) => { + const completedRow = selectedRowsWithParentedTurnLifecycle.find( + (row) => row.type === "turn/completed" && row.turnId === turnId, + ); + return ( + completedRow === undefined || + completedRow.sequence > byteWindowSequenceEnd + ); + }); const futureAssistantContextRows = byteWindowSequenceEnd === null ? [] - : [...windowTurnIds].flatMap((turnId) => { + : byteBoundaryTurnIds.flatMap((turnId) => { const context = findStoredRootTurnAssistantMessageContextRows(db, { afterSequence: byteWindowSequenceEnd, beforeSequence: byteWindowSequenceEnd + 1, From 1ac79cfc26c3c19a071e69f7ffdb20dc845f3a29 Mon Sep 17 00:00:00 2001 From: Michael Yong Date: Tue, 25 Aug 2026 22:28:23 -0700 Subject: [PATCH 4/4] Keep nested delegation byte slices visible --- .../src/completed-turn-grouping.ts | 35 ++++++++++++++----- .../test/completed-turn-grouping.test.ts | 19 ++++++++++ 2 files changed, 45 insertions(+), 9 deletions(-) diff --git a/packages/thread-view/src/completed-turn-grouping.ts b/packages/thread-view/src/completed-turn-grouping.ts index 05aefae689..578838228a 100644 --- a/packages/thread-view/src/completed-turn-grouping.ts +++ b/packages/thread-view/src/completed-turn-grouping.ts @@ -332,16 +332,33 @@ export function groupCompletedTurnMessages( messages, completionIsContextOnly ? undefined : turn.terminalMessage, ); + const messageBoundSummaryItems = groupCompletedTurnSummaryMessages( + turn, + summaryMessages, + terminalMessages[0], + !completionIsContextOnly && !messageBoundsOnly, + contextOnlyMessageSeqs, + ); + // A byte slice can contain only nested delegation children while the root + // parent starts before the slice. The projection still records that work in + // `summaryCount`, but there is no root message from which message bounds can + // be derived. Fall back to the turn bounds for exactly that empty-message + // case; the server clamps those bounds to the byte slice afterward. + const summaryItems = + messageBoundsOnly && + summaryMessages.length === 0 && + messageBoundSummaryItems.length === 0 && + turn.summaryCount > 0 + ? groupCompletedTurnSummaryMessages( + turn, + summaryMessages, + terminalMessages[0], + !completionIsContextOnly, + contextOnlyMessageSeqs, + ) + : messageBoundSummaryItems; return { - summaryItems: unwrapSingletonContextManagementGroups( - groupCompletedTurnSummaryMessages( - turn, - summaryMessages, - terminalMessages[0], - !completionIsContextOnly && !messageBoundsOnly, - contextOnlyMessageSeqs, - ), - ), + summaryItems: unwrapSingletonContextManagementGroups(summaryItems), terminalMessages, trailingMessages, }; diff --git a/packages/thread-view/test/completed-turn-grouping.test.ts b/packages/thread-view/test/completed-turn-grouping.test.ts index 6cdd179bad..305207eed4 100644 --- a/packages/thread-view/test/completed-turn-grouping.test.ts +++ b/packages/thread-view/test/completed-turn-grouping.test.ts @@ -221,6 +221,25 @@ describe("groupCompletedTurnMessages", () => { expect(groups.terminalMessages).toEqual([response]); }); + it("falls back to turn bounds when a byte slice contains only nested work", () => { + const turn = completedTurn([], undefined, 650); + turn.sourceSeqEnd = 1_304; + turn.createdAt = 1_304; + turn.completedAt = 1_304; + + const groups = groupCompletedTurnMessages(turn, false, undefined, true); + + expect(groups.summaryItems).toMatchObject([ + { + kind: "summary", + startedAt: 1, + completedAt: 1_304, + sourceBounds: "turn", + summaryCount: 650, + }, + ]); + }); + it("keeps an assistant response visible when more assistant text follows it directly", () => { const answer = assistantMessage({ id: "answer", seq: 1 }); const hookReply = assistantMessage({ id: "hook-reply", seq: 2 });