From 75dd923e1e6bbc391568f67e5466bcfa6915e8d1 Mon Sep 17 00:00:00 2001 From: Michael Yong Date: Thu, 17 Sep 2026 15:35:50 -0700 Subject: [PATCH 1/3] Skip hidden command payloads and share timeline expansion context --- apps/server/src/services/threads/timeline.ts | 619 ++++++------------ .../timeline-structural-output.test.ts | 165 +++++ docs/timeline-pagination.md | 27 +- packages/db/src/data/events.ts | 68 +- packages/db/src/data/index.ts | 2 + packages/db/test/query-plans.test.ts | 19 + .../thread-view/src/build-thread-timeline.ts | 108 ++- .../src/event-projection-message.ts | 1 - packages/thread-view/src/exec-lifecycle.ts | 8 +- .../src/tool-activity-projection.ts | 31 - .../test/completed-turn-grouping.test.ts | 1 - .../completed-turn-summary-rendering.test.ts | 3 +- .../test/timeline-row-plan.test.ts | 1 + .../test/tool-activity-projection.test.ts | 6 - 14 files changed, 561 insertions(+), 498 deletions(-) create mode 100644 apps/server/test/services/threads/timeline-structural-output.test.ts diff --git a/apps/server/src/services/threads/timeline.ts b/apps/server/src/services/threads/timeline.ts index 46069a0d2a9..61937f630f5 100644 --- a/apps/server/src/services/threads/timeline.ts +++ b/apps/server/src/services/threads/timeline.ts @@ -25,7 +25,6 @@ import type { ProviderComposerCommand, Thread, ThreadEvent, - ThreadEventItemType, } from "@bb/domain"; import type { ThreadConversationOutlineItem, @@ -49,18 +48,15 @@ import { listContextWindowUsageRows, isTimelineCursorSequencePresent, listStoredConversationOutlineEventRows, - listStoredClientTurnRequestIdsInRange, listStoredEventRows, + listStoredEventRowsByIds, listTimelineInterruptionRows, listStoredClientTurnRequestRowsByKeys, listStoredEventRowsByParentToolCallIds, - listItemEventSpansByItems, - listStoredBufferedTextDeltaRowsByItems, - listStoredItemLifecycleRowsByItems, listLatestBackgroundTaskStateRowsByItemIds, listLatestThreadStateEventRowsByThreadIds, listLatestOpenBackgroundTaskStateRowsForThread, - listStoredTimelineWindowEventRows, + hasTimelineTurnEventsInWindow, listStoredTimelineTurnEventRows, listStoredTimelineThreadWindowEventRows, listTimelineRootWindowTurnIds, @@ -70,13 +66,11 @@ import { listStoredTurnRejectedRowsByClientRequestIds, listStoredTurnStartedRowsByTurnIdsUpToSequence, listTimelineWindowHintsDescending, - scopedItemRefKey, upsertThreadConversationOutlineRecord, } from "@bb/db"; import type { DbConnection, InlineOutputCharLimit, - ScopedItemRef, StoredEventRow, } from "@bb/db"; import { ApiError } from "../../errors.js"; @@ -117,33 +111,6 @@ function resolveThreadWorkspaceRoot( return getEnvironment(db, thread.environmentId)?.path ?? null; } -interface PartitionAcceptedInputRowsByRequestedTurnArgs { - acceptedInputRows: readonly StoredEventRow[]; - turnId: string; -} - -interface PartitionAcceptedInputRowsByRequestedTurnResult { - acceptedClientRequestIdsForOtherTurns: ReadonlySet; - requestedTurnRows: StoredEventRow[]; -} - -interface FilterExactEventRowsForRequestedTurnArgs { - acceptedClientRequestIdsForOtherTurns: ReadonlySet; - exactEventRows: readonly StoredEventRow[]; - turnId: string; -} - -interface FilterExactEventRowsForRequestedTurnResult { - removedRows: boolean; - rows: readonly StoredEventRow[]; -} - -interface ResolveTurnSummaryDetailsSourceRangeArgs { - exactEventRows: readonly StoredEventRow[]; - fallbackRange: TimelineTurnSummarySelection; - useExactEventRowBounds: boolean; -} - interface BuildThreadTimelineOptions { completedTurnDisplay: CompletedTurnDisplay; eventBudget: number; @@ -232,6 +199,7 @@ interface TimelineWindowRowsArgs { } interface TimelineWindowParentedRowsArgs extends TimelineWindowRowsArgs { + includeCommandOutput: boolean; excludeDiagnosticEvents: boolean; maxInlineOutputChars: InlineOutputCharLimit; sequenceBounds: { @@ -527,6 +495,7 @@ function ensureTimelineWindowParentedRows( excludedTypes: THREAD_TIMELINE_EXCLUDED_EVENT_TYPES, excludeDiagnosticEvents: args.excludeDiagnosticEvents, maxInlineOutputChars: args.maxInlineOutputChars, + includeCommandOutput: args.includeCommandOutput, parentToolCallIds: toolCallIdsToFetch, sequenceStart: childSequenceBounds?.sequenceStart, threadId: args.threadId, @@ -607,108 +576,6 @@ function selectClientRequestContextRows( }; } -function partitionAcceptedInputRowsByRequestedTurn( - args: PartitionAcceptedInputRowsByRequestedTurnArgs, -): PartitionAcceptedInputRowsByRequestedTurnResult { - const acceptedClientRequestIdsForOtherTurns = new Set(); - const requestedTurnRows: StoredEventRow[] = []; - for (const row of args.acceptedInputRows) { - if (row.scopeKind !== "turn" || row.turnId === null) { - throw new Error(`Expected turn-scoped turn/input/accepted row ${row.id}`); - } - const clientRequestId = parseAcceptedInputClientRequestId(row); - if (row.turnId === args.turnId) { - requestedTurnRows.push(row); - continue; - } - acceptedClientRequestIdsForOtherTurns.add(clientRequestId); - } - - return { - acceptedClientRequestIdsForOtherTurns, - requestedTurnRows, - }; -} - -const CROSS_TURN_TOOL_ITEM_KINDS: ReadonlySet = new Set([ - "commandExecution", - "toolCall", - "webSearch", - "webFetch", - "imageView", - "fileRead", - "search", - "planSteps", - "delegation", - "extension", -]); - -function filterExactEventRowsForRequestedTurn( - args: FilterExactEventRowsForRequestedTurnArgs, -): FilterExactEventRowsForRequestedTurnResult { - const rows: StoredEventRow[] = []; - let removedRows = false; - const openToolCallIds = new Set(); - for (const row of args.exactEventRows) { - if (row.scopeKind === "turn" && row.turnId !== args.turnId) { - const continuesOpenToolCall = - row.itemId !== null && - row.type.startsWith("item/") && - openToolCallIds.has(row.itemId); - if (!continuesOpenToolCall) { - removedRows = true; - continue; - } - } else if ( - row.type === "item/started" && - row.itemId !== null && - row.itemKind !== null && - CROSS_TURN_TOOL_ITEM_KINDS.has(row.itemKind) - ) { - openToolCallIds.add(row.itemId); - } - if (row.type === "item/completed" && row.itemId !== null) { - openToolCallIds.delete(row.itemId); - } - - const requestId = tryReadClientTurnRequestedRequestId(row); - if ( - requestId !== null && - args.acceptedClientRequestIdsForOtherTurns.has(requestId) - ) { - removedRows = true; - continue; - } - rows.push(row); - } - - return { - removedRows, - rows, - }; -} - -function resolveTurnSummaryDetailsSourceRange( - args: ResolveTurnSummaryDetailsSourceRangeArgs, -): TimelineTurnSummarySelection { - const fallbackRange = args.fallbackRange; - if (!args.useExactEventRowBounds) { - return fallbackRange; - } - - const firstRow = args.exactEventRows[0]; - const lastRow = args.exactEventRows.at(-1); - if (!firstRow || !lastRow) { - return fallbackRange; - } - - return { - sourceSeqEnd: lastRow.sequence, - sourceSeqStart: firstRow.sequence, - turnId: fallbackRange.turnId, - }; -} - function collectTurnIdsMissingStartedRows( rows: readonly StoredEventRow[], ): string[] { @@ -759,124 +626,6 @@ function ensureTimelineWindowTurnStartedRows( return mergeStoredEventRowsById([...turnStartedRows, ...args.rows]); } -function storedEventRowItemRef(row: StoredEventRow): ScopedItemRef { - return { - itemId: row.itemId ?? "", - scopeKind: row.scopeKind, - turnId: row.turnId, - }; -} - -interface SequenceWindowItemRowsArgs extends TimelineWindowRowsArgs { - beforeSequence: number | undefined; - maxInlineOutputChars: InlineOutputCharLimit; - sequenceStart: number; -} - -function rowIdentifiesBufferedTextItem(row: StoredEventRow): boolean { - if (row.type === "item/started") { - return ( - row.itemKind === "agentMessage" || - row.itemKind === "plan" || - row.itemKind === "reasoning" - ); - } - return ( - row.type === "item/agentMessage/delta" || - row.type === "item/plan/delta" || - row.type === "item/reasoning/summaryTextDelta" || - row.type === "item/reasoning/textDelta" - ); -} - -function ensureSequenceWindowWholeItemRows( - db: DbConnection, - args: SequenceWindowItemRowsArgs, -): StoredEventRow[] { - const windowItems = new Map(); - for (const row of args.rows) { - if ( - row.itemId !== null && - row.itemKind !== "backgroundTask" && - row.sequence >= args.sequenceStart - ) { - const ref = storedEventRowItemRef(row); - windowItems.set(scopedItemRefKey(ref), ref); - } - } - if (windowItems.size === 0) { - return [...args.rows]; - } - - const spans = listItemEventSpansByItems(db, { - items: [...windowItems.values()], - threadId: args.threadId, - }); - const itemKeysOwnedByNewerWindow = new Set(); - const itemsStartingBeforeWindow = new Map(); - for (const span of spans) { - const key = scopedItemRefKey(span); - if ( - args.beforeSequence !== undefined && - span.maxSequence >= args.beforeSequence - ) { - itemKeysOwnedByNewerWindow.add(key); - continue; - } - if (span.minSequence < args.sequenceStart) { - itemsStartingBeforeWindow.set(key, { - itemId: span.itemId, - scopeKind: span.scopeKind, - turnId: span.turnId, - }); - } - } - - const rows = args.rows.filter( - (row) => - row.itemId === null || - !itemKeysOwnedByNewerWindow.has( - scopedItemRefKey(storedEventRowItemRef(row)), - ), - ); - if (itemsStartingBeforeWindow.size === 0) { - return rows; - } - - const backfillRows = listStoredItemLifecycleRowsByItems(db, { - items: [...itemsStartingBeforeWindow.values()], - maxInlineOutputChars: args.maxInlineOutputChars, - threadId: args.threadId, - }).filter((row) => row.sequence < args.sequenceStart); - - const completedItemKeys = new Set(); - for (const row of [...rows, ...backfillRows]) { - if (row.type === "item/completed" && row.itemId !== null) { - completedItemKeys.add(scopedItemRefKey(storedEventRowItemRef(row))); - } - } - const bufferedTextItems = new Map(); - for (const row of [...backfillRows, ...rows]) { - if (row.itemId === null || !rowIdentifiesBufferedTextItem(row)) { - continue; - } - const ref = storedEventRowItemRef(row); - const key = scopedItemRefKey(ref); - if (!completedItemKeys.has(key) && itemsStartingBeforeWindow.has(key)) { - bufferedTextItems.set(key, ref); - } - } - const bufferedTextRows = listStoredBufferedTextDeltaRowsByItems(db, { - beforeSequence: args.sequenceStart, - items: [...bufferedTextItems.values()], - threadId: args.threadId, - }); - const prefixRows = [...backfillRows, ...bufferedTextRows]; - return prefixRows.length === 0 - ? rows - : mergeStoredEventRowsById([...prefixRows, ...rows]); -} - function ensureTimelineWindowBackgroundTaskStateRows( db: DbConnection, args: TimelineWindowRowsArgs & { beforeSequence?: number }, @@ -935,6 +684,7 @@ function selectStandardTimelineEventRows( thread: Thread, page: ThreadTimelinePageRequest, eventBudget: number, + includeCommandOutput: boolean, maxInlineOutputChars: InlineOutputCharLimit, epochSequenceStart: number, excludeDiagnosticEvents: boolean, @@ -943,8 +693,6 @@ function selectStandardTimelineEventRows( knownBudgetFloor: TimelineBudgetFloor | null, profile: ThreadTimelineBuildProfileAccumulator, ): StandardTimelineEventRowSelection { - const decode: StoredEventDecoder = (row) => - decodeStoredEventRowCached(db, row); const beforeSequence = contentCursor?.beforeSequence ?? (page.kind === "older" ? page.beforeCursor.anchorSeq : maxSeq + 1); @@ -998,7 +746,72 @@ function selectStandardTimelineEventRows( excludedTypes: THREAD_TIMELINE_EXCLUDED_EVENT_TYPES, excludeDiagnosticEvents, }) !== undefined; + const { fetchedTurnIds, ...context } = loadTimelineContextRows( + db, + thread, + { + sequenceStart, + beforeSequence, + epochSequenceStart, + maxSeq, + maxInlineOutputChars, + excludeDiagnosticEvents, + includeHeadState: page.kind === "latest", + requestedTurnIds: [], + includeCommandOutput, + }, + profile, + ); + return { + hints, + fetchedTurnIds, + selection: { + ...context, + ownedSequenceStart: sequenceStart, + ownedSequenceEnd: beforeSequence, + knownHasOlderSegments: hasOlder ? true : null, + paginationPage: + contentCursor === undefined ? page : { ...page, segmentLimit: 1 }, + responsePageKind: page.kind, + strategy: + !hasOlder && page.kind === "latest" ? "full" : "standard-window", + }, + }; +} + +interface TimelineContextRowsOptions { + includeCommandOutput: boolean; + sequenceStart: number; + beforeSequence: number; + epochSequenceStart: number; + maxSeq: number; + maxInlineOutputChars: InlineOutputCharLimit; + excludeDiagnosticEvents: boolean; + includeHeadState: boolean; + requestedTurnIds: readonly string[]; +} + +function loadTimelineContextRows( + db: DbConnection, + thread: Thread, + options: TimelineContextRowsOptions, + profile: ThreadTimelineBuildProfileAccumulator, +) { + const { + sequenceStart, + beforeSequence, + epochSequenceStart, + maxSeq, + maxInlineOutputChars, + excludeDiagnosticEvents, + includeHeadState, + requestedTurnIds, + includeCommandOutput, + } = options; + const decode: StoredEventDecoder = (row) => + decodeStoredEventRowCached(db, row); const windowArgs = { + includeCommandOutput, threadId: thread.id, sequenceStart, beforeSequence, @@ -1018,6 +831,7 @@ function selectStandardTimelineEventRows( ); let rows = listStoredTimelineThreadWindowEventRows(db, windowArgs); const initialTurnIds = [ + ...requestedTurnIds, ...listTimelineRootWindowTurnIds(db, windowArgs), ...rows.flatMap((row) => { if (row.type !== "client/turn/requested") return []; @@ -1053,6 +867,7 @@ function selectStandardTimelineEventRows( }), ]); selectedRows = ensureTimelineWindowParentedRows(db, { + includeCommandOutput, threadId: thread.id, rows: selectedRows, maxInlineOutputChars, @@ -1068,7 +883,7 @@ function selectStandardTimelineEventRows( rows: selectedRows, beforeSequence: maxSeq + 1, }).filter((row) => row.sequence <= maxSeq); - if (page.kind === "latest") + if (includeHeadState) selectedRows = ensureLatestTimelineOpenBackgroundTaskStateRows(db, { threadId: thread.id, rows: selectedRows, @@ -1160,43 +975,31 @@ function selectStandardTimelineEventRows( [...contextRows, ...rows].map((row) => row.sequence), ); return { - hints, fetchedTurnIds: fetchedTurns, - selection: { - headStateRows: - page.kind === "latest" - ? measureThreadTimelineStage(profile, "group-context-query", () => - listLatestTimelineHeadStateRows(db, thread.id).filter( - (row) => - row.sequence <= maxSeq && !visibleSequences.has(row.sequence), - ), - ) - : [], - contextOnlyInterruptionSequences: new Set( - interruptionRows - .filter((row) => !visibleSequences.has(row.sequence)) - .map((row) => row.sequence), - ), - orderingBoundarySequence: groupingContext.orderingBoundarySequence, - ownedSequenceStart: sequenceStart, - ownedSequenceEnd: beforeSequence, - knownHasOlderSegments: hasOlder ? true : null, - paginationPage: - contentCursor === undefined ? page : { ...page, segmentLimit: 1 }, - responsePageKind: page.kind, - rows: ensureTimelineWindowTurnStartedRows(db, { - threadId: thread.id, - rows: mergeStoredEventRowsById([ - ...interruptionRows, - ...terminalContext, - ...contextRows, - ...requestedRows, - ...rows, - ]), - }), - strategy: - !hasOlder && page.kind === "latest" ? "full" : "standard-window", - }, + headStateRows: includeHeadState + ? measureThreadTimelineStage(profile, "group-context-query", () => + listLatestTimelineHeadStateRows(db, thread.id).filter( + (row) => + row.sequence <= maxSeq && !visibleSequences.has(row.sequence), + ), + ) + : [], + contextOnlyInterruptionSequences: new Set( + interruptionRows + .filter((row) => !visibleSequences.has(row.sequence)) + .map((row) => row.sequence), + ), + orderingBoundarySequence: groupingContext.orderingBoundarySequence, + rows: ensureTimelineWindowTurnStartedRows(db, { + threadId: thread.id, + rows: mergeStoredEventRowsById([ + ...interruptionRows, + ...terminalContext, + ...contextRows, + ...requestedRows, + ...rows, + ]), + }), }; } @@ -1297,12 +1100,18 @@ function buildThreadTimelineInternal( atOrBeforeSequence: snapshot.maxSeq, threadId: thread.id, }); + const includeCommandOutput = + includeNestedRows || + options.completedTurnDisplay === "flat" || + snapshot.status === "active" || + options.maxInlineOutputChars === null; const selectRows = (knownBudgetFloor: TimelineBudgetFloor | null) => selectStandardTimelineEventRows( db, thread, options.page, options.eventBudget, + includeCommandOutput, options.maxInlineOutputChars, contextBoundarySeq ?? 0, !includeDiagnosticOperations, @@ -1357,11 +1166,11 @@ function buildThreadTimelineInternal( rows: hydrateRetainedEventOutputRows(db, storedEventSelection.rows), } : storedEventSelection; - const rawEventRows = eventSelection.rows; + let rawEventRows = eventSelection.rows; profile.eventDataBytes = byteLengthOfStoredEventRows(rawEventRows); profile.eventRowCount = rawEventRows.length; profile.selectionStrategy = eventSelection.strategy; - const decodedRawEvents = measureThreadTimelineStage( + let decodedRawEvents = measureThreadTimelineStage( profile, "event-json-decode", () => @@ -1382,7 +1191,7 @@ function buildThreadTimelineInternal( eventSelection.headStateRows, ); profile.decodedEventCount = decodedRawEvents.length + headStateEvents.length; - const decodedEvents = measureThreadTimelineStage( + let decodedEvents = measureThreadTimelineStage( profile, "summary-compaction", () => compactThreadTimelineSummaryEvents(decodedRawEvents), @@ -1423,22 +1232,74 @@ function buildThreadTimelineInternal( acceptedClientRequestEvents: [], rejectedClientRequestEvents: [], }; - const timeline = measureThreadTimelineStage( + const projectTimeline = () => + buildThreadTimelineFromEvents({ + acceptedClientRequestContext, + contextWindowEvents, + headStateEvents, + events: decodedEvents, + options: { + ...commonProjectionOptions, + includeNestedRows, + providerId: thread.providerId, + }, + }); + let timeline = measureThreadTimelineStage( profile, "thread-view-projection", - () => - buildThreadTimelineFromEvents({ - acceptedClientRequestContext, - contextWindowEvents, - headStateEvents, - events: decodedEvents, - options: { - ...commonProjectionOptions, - includeNestedRows, - providerId: thread.providerId, - }, - }), + projectTimeline, ); + if (!includeCommandOutput) { + const visibleCommandIds = new Set(); + const collectCommands = (rows: readonly TimelineRow[]): void => { + for (const row of rows) { + if (row.kind === "turn") collectCommands(row.children ?? []); + if (row.kind !== "work") continue; + if ("callId" in row) visibleCommandIds.add(row.callId); + if (row.workKind === "delegation") collectCommands(row.childRows); + } + }; + collectCommands(timeline.rows); + const ids = rawEventRows + .filter( + (row) => + row.itemId !== null && + visibleCommandIds.has(row.itemId) && + (row.itemKind === "commandExecution" || + row.type === "item/commandExecution/outputDelta"), + ) + .map((row) => row.id); + if (ids.length > 0) { + const payloadRows = measureThreadTimelineStage( + profile, + "event-query", + () => + listStoredEventRowsByIds(db, { + ids, + maxInlineOutputChars, + }), + ); + const payloads = new Map(payloadRows.map((row) => [row.id, row])); + rawEventRows = rawEventRows.map((row) => payloads.get(row.id) ?? row); + profile.eventDataBytes += byteLengthOfStoredEventRows(payloadRows); + profile.eventRowCount += payloadRows.length; + profile.decodedEventCount += payloadRows.length; + decodedRawEvents = measureThreadTimelineStage( + profile, + "event-json-decode", + () => + rawEventRows.map((row) => + withRowMeta(row, decodeStoredEventRowCached(db, row)), + ), + ); + decodedEvents = compactThreadTimelineSummaryEvents(decodedRawEvents); + timeline = measureThreadTimelineStage( + profile, + "thread-view-projection", + projectTimeline, + ); + } + } const projectedTimelineRows = applyRetainedOutputPreviews( orderTimelineRowsUsingContext( timeline.rows.filter( @@ -1805,107 +1666,48 @@ function buildTimelineTurnSummaryDetailsPage( if (fullDetailsFloor.kind !== "fits") { detailsInlineOutputLimit = DEFAULT_MAX_INLINE_OUTPUT_CHARS; } - const exactEventRows = listStoredTimelineWindowEventRows(db, { - ...detailsWindow, - maxInlineOutputChars: detailsInlineOutputLimit, - }); - const clientRequestIds = listStoredClientTurnRequestIdsInRange(db, { - threadId: thread.id, - seqStart: options.sourceSeqStart, - seqEnd: options.sourceSeqEnd, - }); - const exactAcceptedInputRows = exactEventRows.filter( - (row) => row.type === "turn/input/accepted", - ); - const futureAcceptedInputRows = - listStoredTurnInputAcceptedRowsByClientRequestIds(db, { - threadId: thread.id, - afterSequence: options.sourceSeqEnd, - clientRequestIds, - }).filter((row) => row.sequence <= snapshot.maxSeq); - const acceptedInputRowsByTurn = partitionAcceptedInputRowsByRequestedTurn({ - acceptedInputRows: [...exactAcceptedInputRows, ...futureAcceptedInputRows], - turnId: options.turnId, - }); - const exactEventRowsForRequestedTurn = filterExactEventRowsForRequestedTurn({ - acceptedClientRequestIdsForOtherTurns: - acceptedInputRowsByTurn.acceptedClientRequestIdsForOtherTurns, - exactEventRows, - turnId: options.turnId, - }); - const eventRows = mergeStoredEventRowsById([ - ...exactEventRowsForRequestedTurn.rows, - ...acceptedInputRowsByTurn.requestedTurnRows, - ]); - - const hasTurnScopedRowsForRequestedTurn = eventRows.some( - (row) => row.scopeKind === "turn" && row.turnId === options.turnId, - ); - if (!hasTurnScopedRowsForRequestedTurn) { + if ( + !hasTimelineTurnEventsInWindow(db, { + ...detailsWindow, + turnId: options.turnId, + }) + ) { throw new ApiError( 400, "invalid_request", `Timeline turn summary details range ${options.sourceSeqStart}-${options.sourceSeqEnd} does not include turn ${options.turnId}`, ); } - - const hasCurrentStartedRow = eventRows.some( - (row) => row.type === "turn/started" && row.turnId === options.turnId, - ); - const contextSequenceCutoff = eventRows.reduce( - (maxSequence, row) => Math.max(maxSequence, row.sequence), - options.sourceSeqEnd, + const context = loadTimelineContextRows( + db, + thread, + { + ...detailsWindow, + epochSequenceStart: + getLatestCompletedThreadContextClearSequence(db, { + atOrBeforeSequence: snapshot.maxSeq, + threadId: thread.id, + }) ?? 0, + maxSeq: snapshot.maxSeq, + maxInlineOutputChars: detailsInlineOutputLimit, + includeHeadState: false, + requestedTurnIds: [options.turnId], + includeCommandOutput: true, + }, + createThreadTimelineBuildProfileAccumulator(), ); - const requestedTurnStartedRows = hasCurrentStartedRow - ? [] - : listStoredTurnStartedRowsByTurnIdsUpToSequence(db, { - threadId: thread.id, - sequenceCutoff: contextSequenceCutoff, - turnIds: [options.turnId], - }); - if (!hasCurrentStartedRow && requestedTurnStartedRows.length === 0) { + if ( + !context.rows.some( + (row) => row.type === "turn/started" && row.turnId === options.turnId, + ) + ) { throw new ApiError( 400, "invalid_request", `Timeline turn summary details range ${options.sourceSeqStart}-${options.sourceSeqEnd} cannot resolve turn/started for ${options.turnId}`, ); } - const sourceRange = resolveTurnSummaryDetailsSourceRange({ - exactEventRows: exactEventRowsForRequestedTurn.rows, - fallbackRange: { - sourceSeqEnd: options.sourceSeqEnd, - sourceSeqStart: options.sourceSeqStart, - turnId: options.turnId, - }, - useExactEventRowBounds: exactEventRowsForRequestedTurn.removedRows, - }); - const wholeItemEventRows = ensureSequenceWindowWholeItemRows(db, { - beforeSequence: detailsWindow.beforeSequence, - maxInlineOutputChars: detailsInlineOutputLimit, - rows: mergeStoredEventRowsById([...requestedTurnStartedRows, ...eventRows]), - sequenceStart: detailsWindow.sequenceStart, - threadId: thread.id, - }); - const eventRowsWithParentedChildren = ensureTimelineWindowParentedRows(db, { - excludeDiagnosticEvents: !includeDiagnosticOperations, - maxInlineOutputChars: detailsInlineOutputLimit, - sequenceBounds: { - beforeSequence: snapshot.maxSeq + 1, - sequenceStart: detailsWindow.sequenceStart, - }, - threadId: thread.id, - rows: wholeItemEventRows, - }).rows; - const eventRowsWithTurnStarts = ensureTimelineWindowTurnStartedRows(db, { - threadId: thread.id, - rows: eventRowsWithParentedChildren, - }); - const eventRowsWithBackgroundTaskState = - ensureTimelineWindowBackgroundTaskStateRows(db, { - threadId: thread.id, - rows: eventRowsWithTurnStarts, - beforeSequence: snapshot.maxSeq + 1, - }); + const eventRowsWithBackgroundTaskState = context.rows; const hydratedEventRows = detailsInlineOutputLimit === null ? hydrateRetainedEventOutputRowsWithinDataByteLimit( @@ -1919,23 +1721,26 @@ function buildTimelineTurnSummaryDetailsPage( THREAD_TIMELINE_EVENT_DATA_BYTE_LIMIT ? hydratedEventRows : eventRowsWithBackgroundTaskState; - const projectionSourceSeqStart = eventRowsWithTurnStarts.reduce( - (sourceSeqStart, row) => - row.type === "turn/started" && row.turnId === options.turnId - ? Math.min(sourceSeqStart, row.sequence) - : sourceSeqStart, - sourceRange.sourceSeqStart, + const startedTurnIds = new Set( + projectionEventRows.flatMap((row) => + row.type === "turn/started" ? [row.turnId] : [], + ), ); const projectionEvents = projectionEventRows - .filter((row) => row.sequence <= snapshot.maxSeq) + .filter( + (row) => + row.sequence <= snapshot.maxSeq && + (row.type !== "turn/input/accepted" || startedTurnIds.has(row.turnId)), + ) .map((row) => toThreadEventWithMeta(row)); const children = buildThreadTimelineTurnDetailsFromEvents({ events: projectionEvents, options: { completedTurnDisplay: options.completedTurnDisplay, includeDiagnosticOperations, - sourceSeqEnd: sourceRange.sourceSeqEnd, - sourceSeqStart: projectionSourceSeqStart, + sourceSeqEnd: options.sourceSeqEnd, + sourceSeqStart: options.sourceSeqStart, + turnId: options.turnId, providerDisplayName: options.providerDisplayName, threadStatus: snapshot.status, threadName: thread.title ?? thread.titleFallback ?? "", diff --git a/apps/server/test/services/threads/timeline-structural-output.test.ts b/apps/server/test/services/threads/timeline-structural-output.test.ts new file mode 100644 index 00000000000..4e00f1bbf37 --- /dev/null +++ b/apps/server/test/services/threads/timeline-structural-output.test.ts @@ -0,0 +1,165 @@ +import { describe, expect, it } from "vitest"; +import { turnScope, type Thread } from "@bb/domain"; +import { + createConnection, + createProject, + createThread, + insertEvents, + migrate, + noopNotifier, + upsertHost, + type DbConnection, +} from "@bb/db"; +import { + buildThreadTimelineWithProfile, + buildTimelineTurnSummaryDetails, +} from "../../../src/services/threads/timeline.js"; + +function fixture() { + const db = createConnection(":memory:"); + migrate(db); + const host = upsertHost(db, noopNotifier, { name: "test-host" }); + const { project } = createProject(db, noopNotifier, { + name: "test-project", + source: { type: "local_path", hostId: host.id, path: "/tmp/test" }, + }); + const thread = createThread(db, noopNotifier, { + projectId: project.id, + providerId: "claude-code", + status: "idle", + }); + let sequence = 0; + const add = ( + type: Parameters[2][number]["type"], + data: object, + itemId: string | null = null, + itemKind: "commandExecution" | "agentMessage" | null = null, + ) => { + insertEvents(db, noopNotifier, [ + { + threadId: thread.id, + providerThreadId: "provider", + scope: turnScope("turn"), + sequence: ++sequence, + type, + data: JSON.stringify(data), + itemId, + itemKind, + parentToolCallId: null, + }, + ]); + }; + const command = (id: string, output: string) => ({ + item: { + type: "commandExecution", + id, + command: "cat README.md", + cwd: "/tmp/test", + status: "completed", + approvalStatus: null, + aggregatedOutput: output, + }, + }); + add("turn/started", {}); + for (let index = 0; index < 100; index++) { + const id = `command-${index}`; + add( + "item/completed", + command(id, "hidden output\n".repeat(500)), + id, + "commandExecution", + ); + } + add( + "item/completed", + { item: { type: "agentMessage", id: "answer", text: "Done" } }, + "answer", + "agentMessage", + ); + add( + "item/completed", + command("trailing", "visible output\n".repeat(100)), + "trailing", + "commandExecution", + ); + add("turn/completed", { status: "completed" }); + return { db, thread }; +} + +function build( + db: DbConnection, + thread: Thread, + includeNestedRows: boolean, + responseByteBudget = 20_000_000, +) { + return buildThreadTimelineWithProfile(db, thread, { + completedTurnDisplay: "collapse", + includeDiagnosticOperations: false, + includeNestedRows, + eventBudget: 1500, + maxInlineOutputChars: 32_000, + maxSeq: 0, + page: { kind: "latest", segmentLimit: 8 }, + responseByteBudget, + }); +} + +describe("timeline command output selection", () => { + it("omits hidden payloads while preserving visible output, summary bounds, and expansion", () => { + const { db, thread } = fixture(); + try { + const collapsed = build(db, thread, false); + const expanded = build(db, thread, true); + expect(collapsed.response.rows).toEqual( + expanded.response.rows.map((row) => + row.kind === "turn" ? { ...row, children: null } : row, + ), + ); + expect(collapsed.profile.eventDataBytes).toBeLessThan( + expanded.profile.eventDataBytes / 5, + ); + const summary = expanded.response.rows.find((row) => row.kind === "turn"); + if (summary?.kind !== "turn" || summary.turnId === null) + throw new Error("Missing summary"); + const details = buildTimelineTurnSummaryDetails(db, thread, { + turnId: summary.turnId, + sourceSeqStart: summary.sourceSeqStart, + sourceSeqEnd: summary.sourceSeqEnd, + completedTurnDisplay: "collapse", + includeDiagnosticOperations: false, + }); + expect(details.rows).toEqual(summary.children); + const visible = collapsed.response.rows.find( + (row) => row.kind === "work" && row.workKind === "command", + ); + expect(visible).toMatchObject({ output: "visible output\n".repeat(100) }); + } finally { + db.$client.close(); + } + }); + + it("hydrates visible payloads before applying the response byte budget", () => { + const { db, thread } = fixture(); + try { + const collapsed = build(db, thread, false, 1000).response; + const full = build( + db, + { ...thread, status: "active" }, + false, + 1000, + ).response; + expect(collapsed.rows).toEqual(full.rows); + expect(collapsed.timelinePage.hasOlderRows).toEqual( + full.timelinePage.hasOlderRows, + ); + expect(collapsed.timelinePage.olderCursor?.anchorSeq).toEqual( + full.timelinePage.olderCursor?.anchorSeq, + ); + expect(collapsed.timelinePage.contentPage).toEqual( + full.timelinePage.contentPage, + ); + } finally { + db.$client.close(); + } + }); +}); diff --git a/docs/timeline-pagination.md b/docs/timeline-pagination.md index 354f62f0437..c01edbaf211 100644 --- a/docs/timeline-pagination.md +++ b/docs/timeline-pagination.md @@ -120,15 +120,24 @@ selects that planned summary first and materializes only its children; unrelated summaries are not expanded to find a match. Collapsed rendering no longer needs a separate message-pruning policy that anticipates the grouping rules. -Tool output and nested delegation projections are reconstructed while processing -the loaded events, including for collapsed summaries. The shared row plan avoids -rendering unrelated summaries' child rows when selecting an expansion, but it -does not defer event processing or output reconstruction. - -Selection still reads and decodes event payloads for the required context. Cold -request cost remains dependent on that context; a large collapsed turn is not a -constant-time lookup. Route-cache hits and unchanged deltas are separate cases -and must be benchmarked separately from cold opens and appended updates. +For collapsed inactive timelines, the context query retains command lifecycle +metadata but omits command-output bodies. The same projection and grouping code +selects visible rows. If commands remain visible outside summaries, their payloads +are fetched by event ID and the complete rows are constructed before byte +pagination. Active timelines, flat display, nested-row requests, and unlimited +output requests read complete command payloads directly. Shell-command activity +intents are parsed when a command row is rendered, not for hidden summary children. +Other event payloads and nested delegation structure are still processed upfront. + +Pages and summary expansion use the same conversation-context loader. Expansion +matches the requested turn and exact summary bounds, including nested summaries. +The existing row-output expansion use of the endpoint selects rows owned by its +requested range when the range does not identify a summary; this also handles a +command finishing between preview and expansion. + +Cold request cost still depends on the required context; a large collapsed turn +is not a constant-time lookup. Route-cache hits and unchanged deltas are separate +cases and must be benchmarked separately from cold opens and appended updates. `GET /api/v1/threads/:id/timeline/turn-summary-details` and `sdk.threads.timelineTurnSummaryDetails` retain the existing `turnId`, diff --git a/packages/db/src/data/events.ts b/packages/db/src/data/events.ts index c53cbed1c51..0f78150681f 100644 --- a/packages/db/src/data/events.ts +++ b/packages/db/src/data/events.ts @@ -1160,10 +1160,19 @@ function storedEventRowFieldsWithInlineOutputLimit( }; } -function storedEventRowSqlFields(maxInlineOutputChars: InlineOutputCharLimit) { +function storedEventRowSqlFields( + maxInlineOutputChars: InlineOutputCharLimit, + includeCommandOutput = true, +) { + const data = storedEventRowFieldsWithInlineOutputLimit(maxInlineOutputChars).data; return { createdAt: sql`${events.createdAt}`, - data: sql`${storedEventRowFieldsWithInlineOutputLimit(maxInlineOutputChars).data}`, + data: includeCommandOutput ? sql`${data}` : sql`CASE + WHEN ${events.itemKind} = 'commandExecution' AND ${events.type} IN ('item/started', 'item/completed') + THEN json_remove(${events.data}, '$.item.aggregatedOutput') + WHEN ${events.type} = 'item/commandExecution/outputDelta' + THEN json_set(${events.data}, '$.delta', '') + ELSE ${data} END`, id: sql`${events.id}`, itemId: sql`${events.itemId}`, itemKind: sql`${events.itemKind}`, @@ -1193,6 +1202,7 @@ export interface FindStoredEventRowArgs { } export interface ListStoredEventRowsByParentToolCallIdsArgs { + includeCommandOutput?: boolean; excludeDiagnosticEvents?: boolean; beforeSequence?: number; excludedTypes?: readonly ThreadEventType[]; @@ -1706,7 +1716,7 @@ export function listStoredEventRowsByParentToolCallIds( } return db - .select(storedEventRowSqlFields(args.maxInlineOutputChars)) + .select(storedEventRowSqlFields(args.maxInlineOutputChars, args.includeCommandOutput)) .from( sql`${events} INDEXED BY events_parent_tool_call_thread_parent_sequence_idx`, ) @@ -3166,9 +3176,33 @@ export function findStoredTimelineWindowByteBudgetFloor( return { eventDataBytes: includedDataBytes, kind: "fits" }; } +export function hasTimelineTurnEventsInWindow( + db: DbConnection, + args: Omit & { + turnId: string; + }, +): boolean { + return ( + db + .select({ sequence: events.sequence }) + .from(events) + .where( + and( + ...storedTimelineWindowConditions({ ...args, maxInlineOutputChars: null }), + eq(events.turnId, args.turnId), + ), + ) + .limit(1) + .get() !== undefined + ); +} + export function listStoredTimelineTurnEventRows( db: DbConnection, - args: ListStoredTimelineWindowEventRowsArgs & { turnIds: readonly string[] }, + args: ListStoredTimelineWindowEventRowsArgs & { + turnIds: readonly string[]; + includeCommandOutput?: boolean; + }, ): StoredEventRow[] { if (args.turnIds.length === 0) return []; return queryInSqliteVariableBatches({ @@ -3178,7 +3212,9 @@ export function listStoredTimelineTurnEventRows( fixedVariableCount: 32, queryBatch: (turnIds) => db - .select(storedEventRowSqlFields(args.maxInlineOutputChars)) + .select( + storedEventRowSqlFields(args.maxInlineOutputChars, args.includeCommandOutput), + ) .from( sql`${events} INDEXED BY events_thread_turn_type_item_sequence_idx`, ) @@ -3192,6 +3228,24 @@ export function listStoredTimelineTurnEventRows( }).sort((left, right) => left.sequence - right.sequence); } +export function listStoredEventRowsByIds( + db: DbConnection, + args: { ids: readonly string[]; maxInlineOutputChars: InlineOutputCharLimit }, +): StoredEventRow[] { + return queryInSqliteVariableBatches({ + values: args.ids, + variableCountPerValue: 1, + dedupeKey: (id) => id, + fixedVariableCount: 32, + queryBatch: (ids) => + db + .select(storedEventRowFieldsWithInlineOutputLimit(args.maxInlineOutputChars)) + .from(events) + .where(inArray(events.id, [...ids])) + .all(), + }); +} + export function listTimelineRootWindowTurnIds( db: DbConnection, args: ListStoredTimelineWindowEventRowsArgs, @@ -3217,11 +3271,11 @@ export function listTimelineRootWindowTurnIds( export function listStoredTimelineThreadWindowEventRows( db: DbConnection, - args: ListStoredTimelineWindowEventRowsArgs, + args: ListStoredTimelineWindowEventRowsArgs & { includeCommandOutput?: boolean }, ): StoredEventRow[] { return db .select( - storedEventRowFieldsWithInlineOutputLimit(args.maxInlineOutputChars), + storedEventRowSqlFields(args.maxInlineOutputChars, args.includeCommandOutput), ) .from(events) .where(and(...storedTimelineWindowConditions(args), isNull(events.turnId))) diff --git a/packages/db/src/data/index.ts b/packages/db/src/data/index.ts index 37ec1ba0df9..803c2823def 100644 --- a/packages/db/src/data/index.ts +++ b/packages/db/src/data/index.ts @@ -306,12 +306,14 @@ export { listStoredClientTurnRequestRowsByKeys, listStoredEventRowsByParentToolCallIds, listStoredEventRows, + listStoredEventRowsByIds, listItemEventSpansByItems, listStoredBufferedTextDeltaRowsByItems, listStoredItemLifecycleRowsByItems, scopedItemRefKey, listStoredTimelineWindowEventRows, listStoredTimelineTurnEventRows, + hasTimelineTurnEventsInWindow, listStoredTimelineThreadWindowEventRows, listTimelineRootWindowTurnIds, listStoredDelegatingItemRowsByItemIds, diff --git a/packages/db/test/query-plans.test.ts b/packages/db/test/query-plans.test.ts index 09e6bb523d0..5712843f349 100644 --- a/packages/db/test/query-plans.test.ts +++ b/packages/db/test/query-plans.test.ts @@ -28,6 +28,7 @@ import { listLatestOpenBackgroundTaskStateRowsForThread, listStoredConversationOutlineEventRows, listStoredEventRows, + listStoredEventRowsByIds, listStoredEventRowsByParentToolCallIds, listStoredTurnCompletedKeys, listTodoSnapshotEventRowsForThread, @@ -278,6 +279,24 @@ describe("slow query index plans", () => { } }); + it("resolves selected event payloads through their primary keys", () => { + const { db } = setup(); + try { + const captured = captureStatements(db, () => { + expect( + listStoredEventRowsByIds(db, { + ids: Array.from({ length: 10 }, (_, index) => `event-${index}`), + maxInlineOutputChars: null, + }), + ).toEqual([]); + }); + expect(captured).toHaveLength(1); + expect(queryPlanDetails({ db, ...captured[0]! })).toContain("(id=?)"); + } finally { + db.$client.close(); + } + }); + it("seeks accepted inputs past each thread's latest interruption", () => { const { db, thread } = setup(); try { diff --git a/packages/thread-view/src/build-thread-timeline.ts b/packages/thread-view/src/build-thread-timeline.ts index 4c8db26c65d..fb5ea5720e0 100644 --- a/packages/thread-view/src/build-thread-timeline.ts +++ b/packages/thread-view/src/build-thread-timeline.ts @@ -1,3 +1,4 @@ +import { parseShellCommandIntents } from "./tool-call-parsing.js"; import type { ThreadContextWindowUsage, TimelineActivityIntent, @@ -108,6 +109,7 @@ interface ThreadTimelineSourceSeqRange { } interface BuildThreadTimelineTurnDetailsFromEventsOptions extends ThreadTimelineSourceSeqRange { + turnId: string; completedTurnDisplay: CompletedTurnDisplay; includeDiagnosticOperations: boolean; providerDisplayName?: string; @@ -531,7 +533,9 @@ function convertMessage( exitCode: message.exitCode, completedAt: message.completedAt, approvalStatus: message.approvalStatus, - activityIntents: message.parsedIntents.map(convertActivityIntent), + activityIntents: parseShellCommandIntents(message.command).map( + convertActivityIntent, + ), ...rowPresentation(message), }, ]; @@ -1202,32 +1206,82 @@ export function buildThreadTimelineTurnDetailsFromEvents( rowIdPrefix: ROOT_TIMELINE_ROW_ID_PREFIX, workspaceRoot: args.options.workspaceRoot, }; - const plan = planTimelineRows( - projection, - options.completedTurnDisplay, - options.rowIdPrefix, - ); - const matchingSummary = plan.find( - (item) => - item.kind === "summary" && - item.row.sourceSeqStart === args.options.sourceSeqStart && - item.row.sourceSeqEnd === args.options.sourceSeqEnd, - ); - if (matchingSummary?.kind === "summary") { - return { - kind: "matched", - rows: matchingSummary.messages.flatMap((message) => - convertMessage(message, options), - ), - }; + function findSummary( + current: EventProjection, + rowOptions: BuildTimelineRowsOptions, + ): TimelineRow[] | null { + const plan = planTimelineRows( + current, + rowOptions.completedTurnDisplay, + rowOptions.rowIdPrefix, + ); + for (const item of plan) { + if ( + item.kind === "summary" && + item.row.turnId === args.options.turnId && + item.row.sourceSeqStart === args.options.sourceSeqStart && + item.row.sourceSeqEnd === args.options.sourceSeqEnd + ) { + return item.messages.flatMap((message) => + convertMessage(message, rowOptions), + ); + } + const messages = item.kind === "summary" ? item.messages : [item.message]; + for (const message of messages) { + if (message.kind !== "delegation") continue; + const base = buildTimelineRowBase(message, rowOptions.rowIdPrefix); + const nested = findSummary(message.childProjection, { + ...rowOptions, + rowIdPrefix: `${base.id}:child:`, + }); + if (nested !== null) return nested; + } + } + return null; } - if (plan.some((item) => item.kind === "summary")) { - return { kind: "missing-match" }; + const matchingRows = findSummary(projection, options); + if (matchingRows !== null) return { kind: "matched", rows: matchingRows }; + function selectRange( + current: EventProjection, + rowOptions: BuildTimelineRowsOptions, + ): TimelineRow[] { + const rows: TimelineRow[] = []; + for (const item of planTimelineRows( + current, + rowOptions.completedTurnDisplay, + rowOptions.rowIdPrefix, + )) { + const messages = item.kind === "summary" ? item.messages : [item.message]; + for (const message of messages) { + if ( + getEventProjectionMessageScopeTurnId(message) === + args.options.turnId && + message.sourceSeqStart >= args.options.sourceSeqStart && + message.sourceSeqStart <= args.options.sourceSeqEnd + ) { + rows.push( + ...convertMessage(message, rowOptions).filter( + (row) => !isRootOwnedHumanSteerRow(row), + ), + ); + } else if (message.kind === "delegation") { + const base = buildTimelineRowBase(message, rowOptions.rowIdPrefix); + rows.push( + ...selectRange(message.childProjection, { + ...rowOptions, + rowIdPrefix: `${base.id}:child:`, + }), + ); + } + } + } + return orderRowsAfterExternalUserBoundary( + rows, + collectExternalUserBoundarySeqs(current), + ); } - return { - kind: "ungrouped", - rows: buildTimelineRows(projection, options).filter( - (row) => !isRootOwnedHumanSteerRow(row), - ), - }; + const rows = selectRange(projection, options); + return rows.length > 0 + ? { kind: "ungrouped", rows } + : { kind: "missing-match" }; } diff --git a/packages/thread-view/src/event-projection-message.ts b/packages/thread-view/src/event-projection-message.ts index 1e66ee9c97d..da6a4bff8a6 100644 --- a/packages/thread-view/src/event-projection-message.ts +++ b/packages/thread-view/src/event-projection-message.ts @@ -165,7 +165,6 @@ export interface EventProjectionCommandMessage callId: string; command: string; cwd: string | null; - parsedIntents: EventProjectionToolParsedIntent[]; source: string | null; output: string; exitCode: number | null; diff --git a/packages/thread-view/src/exec-lifecycle.ts b/packages/thread-view/src/exec-lifecycle.ts index 3b76df6d158..3f23c358a2b 100644 --- a/packages/thread-view/src/exec-lifecycle.ts +++ b/packages/thread-view/src/exec-lifecycle.ts @@ -10,12 +10,8 @@ import { getEventParentToolCallId, type EventMeta } from "./event-decode.js"; import type { EventProjectionApprovalLifecycleStatus, EventProjectionToolCallMessage, - EventProjectionToolParsedIntent, } from "./event-projection-types.js"; -import { - extractShellCommandFromString, - parseShellCommandIntents, -} from "./tool-call-parsing.js"; +import { extractShellCommandFromString } from "./tool-call-parsing.js"; interface DelegationMetadata { subagentType?: string; @@ -77,7 +73,6 @@ export interface CommandExecutionUpdate extends ExecutionUpdateBase { kind: "command"; command?: string; cwd?: string | null; - parsedIntents?: EventProjectionToolParsedIntent[]; source?: string | null; exitCode?: number | null; approvalStatus?: EventProjectionApprovalLifecycleStatus | null; @@ -168,7 +163,6 @@ export function parseExecLifecycleEvent( callId, command, cwd: decoded.item.cwd, - parsedIntents: parseShellCommandIntents(command), output: decoded.item.aggregatedOutput, exitCode, completedAt, diff --git a/packages/thread-view/src/tool-activity-projection.ts b/packages/thread-view/src/tool-activity-projection.ts index efe73d2c293..6dd31e1ec86 100644 --- a/packages/thread-view/src/tool-activity-projection.ts +++ b/packages/thread-view/src/tool-activity-projection.ts @@ -8,7 +8,6 @@ import type { EventProjectionMessage, EventProjection, EventProjectionToolCallMessage, - EventProjectionToolParsedIntent, } from "./event-projection-types.js"; import type { EventMeta } from "./event-decode.js"; import type { @@ -92,7 +91,6 @@ interface RunningCommandExecution extends RunningExecutionBase { kind: "command"; command: string; cwd: string | null; - parsedIntents: EventProjectionToolParsedIntent[]; source: string | null; exitCode: number | null; approvalStatus: EventProjectionApprovalLifecycleStatus | null; @@ -214,25 +212,6 @@ export function applyApprovalStatusDelta( } } -function hasSemanticIntent( - intents: EventProjectionToolParsedIntent[], -): boolean { - return intents.some((intent) => intent.type !== "unknown"); -} - -function chooseParsedIntents( - existing: EventProjectionToolParsedIntent[], - incoming: EventProjectionToolParsedIntent[], -): EventProjectionToolParsedIntent[] { - if (incoming.length === 0) return existing; - if (existing.length === 0) return incoming; - if (!hasSemanticIntent(existing) && hasSemanticIntent(incoming)) { - return incoming; - } - if (incoming.length > existing.length) return incoming; - return existing; -} - function isTerminalToolCallStatus( status: EventProjectionToolCallMessage["status"] | undefined, ): boolean { @@ -325,7 +304,6 @@ function createRunningExecCall( kind: "command", command: incoming.command ?? "", cwd: incoming.cwd ?? null, - parsedIntents: incoming.parsedIntents ?? [], source: incoming.source ?? null, exitCode: incoming.exitCode ?? null, approvalStatus: incoming.approvalStatus ?? null, @@ -357,7 +335,6 @@ interface CommandExecutionFieldsTarget { command: string; cwd: string | null; exitCode: number | null; - parsedIntents: EventProjectionToolParsedIntent[]; source: string | null; } @@ -366,7 +343,6 @@ interface CommandExecutionFieldsSource { command?: string; cwd?: string | null; exitCode?: number | null; - parsedIntents?: EventProjectionToolParsedIntent[]; source?: string | null; status?: EventProjectionToolCallMessage["status"]; } @@ -427,12 +403,6 @@ function mergeCommandExecutionFields( if (incoming.source && !target.source) target.source = incoming.source; if (incoming.command && incoming.command !== target.command) { target.command = incoming.command; - target.parsedIntents = incoming.parsedIntents ?? []; - } else { - target.parsedIntents = chooseParsedIntents( - target.parsedIntents, - incoming.parsedIntents ?? [], - ); } if (incoming.exitCode !== undefined) target.exitCode = incoming.exitCode; target.approvalStatus = applyApprovalStatusDelta( @@ -927,7 +897,6 @@ function createExecMessage( kind: "command", command: call.command, cwd: call.cwd, - parsedIntents: call.parsedIntents, source: call.source, exitCode: call.exitCode, approvalStatus: call.approvalStatus, diff --git a/packages/thread-view/test/completed-turn-grouping.test.ts b/packages/thread-view/test/completed-turn-grouping.test.ts index 19d509f4b66..ff52c6e9ae0 100644 --- a/packages/thread-view/test/completed-turn-grouping.test.ts +++ b/packages/thread-view/test/completed-turn-grouping.test.ts @@ -47,7 +47,6 @@ function commandMessage(args: MessageBaseArgs): EventProjectionCommandMessage { callId: args.id, command: "pnpm test", cwd: "/repo", - parsedIntents: [], source: null, output: "", exitCode: 0, diff --git a/packages/thread-view/test/completed-turn-summary-rendering.test.ts b/packages/thread-view/test/completed-turn-summary-rendering.test.ts index 2717e6c41a0..010fdfd0898 100644 --- a/packages/thread-view/test/completed-turn-summary-rendering.test.ts +++ b/packages/thread-view/test/completed-turn-summary-rendering.test.ts @@ -844,6 +844,7 @@ describe("flat completed turn display", () => { const { finishedEvents } = narratedTurn(); const events = fromRows(finishedEvents); const detailOptions = { + turnId: "turn-1", includeDiagnosticOperations: false, sourceSeqEnd: 5, sourceSeqStart: 4, @@ -863,10 +864,8 @@ describe("flat completed turn display", () => { expect(flat.kind).toBe("ungrouped"); expect(flat.kind === "ungrouped" ? rowSignatures(flat.rows) : []).toEqual([ - "conversation:user", "conversation:assistant", "work:command", - "conversation:assistant", ]); expect(collapsed.kind).toBe("matched"); expect( diff --git a/packages/thread-view/test/timeline-row-plan.test.ts b/packages/thread-view/test/timeline-row-plan.test.ts index 8995d615233..f5f11c13ba3 100644 --- a/packages/thread-view/test/timeline-row-plan.test.ts +++ b/packages/thread-view/test/timeline-row-plan.test.ts @@ -150,6 +150,7 @@ describe("timeline row planning", () => { events: fromRows(events), options: { ...options, + turnId: "parent", sourceSeqStart: expected.sourceSeqStart, sourceSeqEnd: expected.sourceSeqEnd, }, diff --git a/packages/thread-view/test/tool-activity-projection.test.ts b/packages/thread-view/test/tool-activity-projection.test.ts index 54f8909df42..d0842ae0552 100644 --- a/packages/thread-view/test/tool-activity-projection.test.ts +++ b/packages/thread-view/test/tool-activity-projection.test.ts @@ -9,7 +9,6 @@ import type { EventProjectionCommandMessage, EventProjectionDelegationMessage, EventProjectionMessage, - EventProjectionToolParsedIntent, } from "../src/event-projection-types.js"; import { createToolActivityState, @@ -26,7 +25,6 @@ interface CommandUpdateArgs { command?: string; completedAt?: number | null; output?: string; - parsedIntents?: EventProjectionToolParsedIntent[]; status: CommandStatus; } @@ -65,7 +63,6 @@ function commandUpdate({ command = "pnpm test", completedAt = null, output, - parsedIntents, status, }: CommandUpdateArgs): CommandExecutionUpdate { return { @@ -77,7 +74,6 @@ function commandUpdate({ exitCode: status === "error" ? 1 : 0, completedAt, ...(output !== undefined ? { output } : {}), - ...(parsedIntents !== undefined ? { parsedIntents } : {}), }; } @@ -247,7 +243,6 @@ describe("tool activity projection", () => { commandUpdate({ command: staleCommand, output: "started\n", - parsedIntents: [{ type: "unknown", cmd: staleCommand }], status: "pending", }), ); @@ -262,7 +257,6 @@ describe("tool activity projection", () => { expect(commandMessages(state)).toMatchObject([ { command: latestCommand, - parsedIntents: [], }, ]); }); From 65889715b3d71f4d783b176cffc4fbfd506d776d Mon Sep 17 00:00:00 2001 From: Michael Yong Date: Thu, 17 Sep 2026 16:24:59 -0700 Subject: [PATCH 2/3] Bound timeline expansion to selected item histories --- apps/server/src/services/threads/timeline.ts | 12 +- .../timeline-structural-output.test.ts | 141 +++++++++++++++++- docs/timeline-pagination.md | 10 +- packages/db/src/data/events.ts | 34 +++++ packages/db/src/data/index.ts | 1 + packages/db/test/query-plans.test.ts | 38 +++++ 6 files changed, 231 insertions(+), 5 deletions(-) diff --git a/apps/server/src/services/threads/timeline.ts b/apps/server/src/services/threads/timeline.ts index 61937f630f5..108494c8dc9 100644 --- a/apps/server/src/services/threads/timeline.ts +++ b/apps/server/src/services/threads/timeline.ts @@ -58,6 +58,7 @@ import { listLatestOpenBackgroundTaskStateRowsForThread, hasTimelineTurnEventsInWindow, listStoredTimelineTurnEventRows, + listTimelineWindowItemIds, listStoredTimelineThreadWindowEventRows, listTimelineRootWindowTurnIds, listTodoSnapshotEventRowsForThread, @@ -830,6 +831,14 @@ function loadTimelineContextRows( }), ); let rows = listStoredTimelineThreadWindowEventRows(db, windowArgs); + const itemContext = + requestedTurnIds.length === 0 + ? undefined + : { + itemIds: listTimelineWindowItemIds(db, windowArgs), + sequenceStart, + beforeSequence, + }; const initialTurnIds = [ ...requestedTurnIds, ...listTimelineRootWindowTurnIds(db, windowArgs), @@ -864,6 +873,7 @@ function loadTimelineContextRows( sequenceStart: epochSequenceStart, beforeSequence: maxSeq + 1, turnIds, + itemContext, }), ]); selectedRows = ensureTimelineWindowParentedRows(db, { @@ -1734,7 +1744,7 @@ function buildTimelineTurnSummaryDetailsPage( ) .map((row) => toThreadEventWithMeta(row)); const children = buildThreadTimelineTurnDetailsFromEvents({ - events: projectionEvents, + events: compactThreadTimelineSummaryEvents(projectionEvents), options: { completedTurnDisplay: options.completedTurnDisplay, includeDiagnosticOperations, diff --git a/apps/server/test/services/threads/timeline-structural-output.test.ts b/apps/server/test/services/threads/timeline-structural-output.test.ts index 4e00f1bbf37..01e8078bb1a 100644 --- a/apps/server/test/services/threads/timeline-structural-output.test.ts +++ b/apps/server/test/services/threads/timeline-structural-output.test.ts @@ -5,6 +5,8 @@ import { createProject, createThread, insertEvents, + listStoredTimelineTurnEventRows, + listTimelineWindowItemIds, migrate, noopNotifier, upsertHost, @@ -15,7 +17,7 @@ import { buildTimelineTurnSummaryDetails, } from "../../../src/services/threads/timeline.js"; -function fixture() { +function fixture(splitTurn = false) { const db = createConnection(":memory:"); migrate(db); const host = upsertHost(db, noopNotifier, { name: "test-host" }); @@ -33,13 +35,18 @@ function fixture() { type: Parameters[2][number]["type"], data: object, itemId: string | null = null, - itemKind: "commandExecution" | "agentMessage" | null = null, + itemKind: + | "commandExecution" + | "agentMessage" + | "contextCompaction" + | null = null, + turnId = "turn", ) => { insertEvents(db, noopNotifier, [ { threadId: thread.id, providerThreadId: "provider", - scope: turnScope("turn"), + scope: turnScope(turnId), sequence: ++sequence, type, data: JSON.stringify(data), @@ -60,6 +67,19 @@ function fixture() { aggregatedOutput: output, }, }); + if (splitTurn) { + add("turn/started", {}, null, null, "overlap"); + for (let index = 0; index < 100; index++) { + const id = `overlap-${index}`; + add( + "item/completed", + command(id, "unrelated output"), + id, + "commandExecution", + "overlap", + ); + } + } add("turn/started", {}); for (let index = 0; index < 100; index++) { const id = `command-${index}`; @@ -76,6 +96,57 @@ function fixture() { "answer", "agentMessage", ); + if (splitTurn) { + add( + "item/completed", + { item: { type: "agentMessage", id: "next", text: "Next task" } }, + "next", + "agentMessage", + ); + add( + "item/commandExecution/outputDelta", + { itemId: "command-0", delta: "late output" }, + "command-0", + ); + add( + "item/completed", + { item: { type: "contextCompaction", id: "compact-1" } }, + "compact-1", + "contextCompaction", + ); + add( + "item/completed", + command("selected", "selected output"), + "selected", + "commandExecution", + ); + add( + "item/completed", + command("overlap-0", "late completion"), + "overlap-0", + "commandExecution", + "overlap", + ); + add( + "item/completed", + command("selected-tail", "tail output"), + "selected-tail", + "commandExecution", + ); + add( + "item/completed", + { item: { type: "agentMessage", id: "final", text: "Finished" } }, + "final", + "agentMessage", + ); + add( + "item/completed", + { item: { type: "contextCompaction", id: "compact-2" } }, + "compact-2", + "contextCompaction", + ); + add("turn/completed", { status: "completed" }, null, null, "overlap"); + } add( "item/completed", command("trailing", "visible output\n".repeat(100)), @@ -105,6 +176,70 @@ function build( } describe("timeline command output selection", () => { + it("expands a small group without selecting unrelated command history", () => { + const { db, thread } = fixture(true); + try { + const expanded = build(db, thread, true).response; + const summary = expanded.rows.find( + (row) => + row.kind === "turn" && + row.children?.some( + (child) => + child.kind === "work" && + child.workKind === "command" && + child.callId === "selected", + ), + ); + if (summary?.kind !== "turn" || summary.turnId === null) + throw new Error("Missing selected summary"); + const details = buildTimelineTurnSummaryDetails(db, thread, { + turnId: summary.turnId, + sourceSeqStart: summary.sourceSeqStart, + sourceSeqEnd: summary.sourceSeqEnd, + completedTurnDisplay: "collapse", + includeDiagnosticOperations: false, + }); + expect(details.rows).toEqual(summary.children); + const context = listStoredTimelineTurnEventRows(db, { + threadId: thread.id, + turnIds: [summary.turnId, "overlap"], + sequenceStart: 0, + beforeSequence: 1000, + maxInlineOutputChars: null, + itemContext: { + itemIds: listTimelineWindowItemIds(db, { + threadId: thread.id, + sequenceStart: summary.sourceSeqStart, + beforeSequence: summary.sourceSeqEnd + 1, + maxInlineOutputChars: null, + }), + sequenceStart: summary.sourceSeqStart, + beforeSequence: summary.sourceSeqEnd + 1, + }, + }); + expect( + context + .filter((row) => row.itemKind === "commandExecution") + .map((row) => row.itemId), + ).toEqual([ + "overlap-0", + "command-0", + "selected", + "overlap-0", + "selected-tail", + ]); + expect( + context.filter((row) => row.itemKind === "contextCompaction"), + ).toHaveLength(2); + expect(context.some((row) => row.type === "turn/completed")).toBe(true); + expect( + context.filter((row) => row.itemKind === "agentMessage"), + ).toHaveLength(3); + } finally { + db.$client.close(); + } + }); + it("omits hidden payloads while preserving visible output, summary bounds, and expansion", () => { const { db, thread } = fixture(); try { diff --git a/docs/timeline-pagination.md b/docs/timeline-pagination.md index c01edbaf211..59ddc1b5b51 100644 --- a/docs/timeline-pagination.md +++ b/docs/timeline-pagination.md @@ -130,7 +130,15 @@ intents are parsed when a command row is rendered, not for hidden summary childr Other event payloads and nested delegation structure are still processed upfront. Pages and summary expansion use the same conversation-context loader. Expansion -matches the requested turn and exact summary bounds, including nested summaries. +first selects item identities from metadata in the requested interval, then loads +their event histories within the snapshot. This preserves updates to items that +started outside the interval without loading unrelated command, reasoning, and +file-change payloads from the same turn. Turn and request state, assistant messages, +and compaction lifecycle events remain context: they affect grouping or represent +state shared across the turn. Parent and child context is still resolved by the +same loader. The requested interval limits detail selection, not those dependencies. +Expansion matches the requested turn and exact summary bounds, including nested +summaries. The existing row-output expansion use of the endpoint selects rows owned by its requested range when the range does not identify a summary; this also handles a command finishing between preview and expansion. diff --git a/packages/db/src/data/events.ts b/packages/db/src/data/events.ts index 0f78150681f..818307d1479 100644 --- a/packages/db/src/data/events.ts +++ b/packages/db/src/data/events.ts @@ -3197,11 +3197,30 @@ export function hasTimelineTurnEventsInWindow( ); } +export function listTimelineWindowItemIds( + db: DbConnection, + args: ListStoredTimelineWindowEventRowsArgs, +): string[] { + return db + .selectDistinct({ itemId: sql`${events.itemId}` }) + .from(events) + .where( + and(...storedTimelineWindowConditions(args), isNotNull(events.itemId)), + ) + .all() + .map((row) => row.itemId); +} + export function listStoredTimelineTurnEventRows( db: DbConnection, args: ListStoredTimelineWindowEventRowsArgs & { turnIds: readonly string[]; includeCommandOutput?: boolean; + itemContext?: { + itemIds: readonly string[]; + sequenceStart: number; + beforeSequence: number; + }; }, ): StoredEventRow[] { if (args.turnIds.length === 0) return []; @@ -3222,6 +3241,21 @@ export function listStoredTimelineTurnEventRows( and( ...storedTimelineWindowConditions(args), inArray(events.turnId, [...turnIds]), + args.itemContext === undefined + ? undefined + : or( + and( + gte(events.sequence, args.itemContext.sequenceStart), + lt(events.sequence, args.itemContext.beforeSequence), + ), + sql`${events.type} NOT LIKE 'item/%'`, + inArray(events.itemKind, [ + "agentMessage", + "contextCompaction", + ]), + eq(events.type, "item/agentMessage/delta"), + sql`${events.itemId} IN (SELECT value FROM json_each(${JSON.stringify(args.itemContext.itemIds)}))`, + ), ), ) .all(), diff --git a/packages/db/src/data/index.ts b/packages/db/src/data/index.ts index 803c2823def..1fafd43b6ea 100644 --- a/packages/db/src/data/index.ts +++ b/packages/db/src/data/index.ts @@ -313,6 +313,7 @@ export { scopedItemRefKey, listStoredTimelineWindowEventRows, listStoredTimelineTurnEventRows, + listTimelineWindowItemIds, hasTimelineTurnEventsInWindow, listStoredTimelineThreadWindowEventRows, listTimelineRootWindowTurnIds, diff --git a/packages/db/test/query-plans.test.ts b/packages/db/test/query-plans.test.ts index 5712843f349..13dc4142c44 100644 --- a/packages/db/test/query-plans.test.ts +++ b/packages/db/test/query-plans.test.ts @@ -29,6 +29,8 @@ import { listStoredConversationOutlineEventRows, listStoredEventRows, listStoredEventRowsByIds, + listTimelineWindowItemIds, + listStoredTimelineTurnEventRows, listStoredEventRowsByParentToolCallIds, listStoredTurnCompletedKeys, listTodoSnapshotEventRowsForThread, @@ -456,6 +458,42 @@ describe("slow query index plans", () => { }, ); + it("bounds expansion item discovery by sequence and context lookup by turn", () => { + const { db, thread } = setup(); + try { + const queries = captureStatements(db, () => { + listTimelineWindowItemIds(db, { + threadId: thread.id, + sequenceStart: 100, + beforeSequence: 200, + maxInlineOutputChars: null, + }); + listStoredTimelineTurnEventRows(db, { + threadId: thread.id, + turnIds: ["selected-turn"], + sequenceStart: 0, + beforeSequence: 1000, + maxInlineOutputChars: null, + itemContext: { + itemIds: ["selected-item"], + sequenceStart: 100, + beforeSequence: 200, + }, + }); + }); + expect(queries).toHaveLength(2); + const plans = queries.map((query) => queryPlanDetails({ db, ...query })); + expect(plans[0]).toMatch( + /events_thread_sequence_idx \(thread_id=\? AND sequence>\? AND sequence<\?\)/u, + ); + expect(plans[1]).toMatch( + /events_thread_turn_type_item_sequence_idx \(thread_id=\? AND turn_id=\?\)/u, + ); + } finally { + db.$client.close(); + } + }); + it("looks up completed turns by thread and turn key", () => { const { db, thread } = setup(); insertEvents( From 63fb6187d42ab4cd447f46f03fd6f2f04a25c88c Mon Sep 17 00:00:00 2001 From: Michael Yong Date: Thu, 17 Sep 2026 18:02:54 -0700 Subject: [PATCH 3/3] Bound expansion dependencies and avoid redundant event processing --- apps/server/src/services/threads/timeline.ts | 64 ++++++++--- .../timeline-parented-pagination.test.ts | 108 +++++++++++++++++- .../timeline-structural-output.test.ts | 31 +++-- docs/timeline-pagination.md | 20 ++-- packages/db/src/data/events.ts | 67 ++++++++--- packages/db/test/query-plans.test.ts | 49 ++++---- 6 files changed, 268 insertions(+), 71 deletions(-) diff --git a/apps/server/src/services/threads/timeline.ts b/apps/server/src/services/threads/timeline.ts index 108494c8dc9..71e0121eabd 100644 --- a/apps/server/src/services/threads/timeline.ts +++ b/apps/server/src/services/threads/timeline.ts @@ -492,6 +492,13 @@ function ensureTimelineWindowParentedRows( const childSequenceBounds = args.sequenceBounds; const childRows = listStoredEventRowsByParentToolCallIds(db, { + excludedEventIds: rows + .filter( + (row) => + row.parentToolCallId !== null && + toolCallIdsToFetch.includes(row.parentToolCallId), + ) + .map((row) => row.id), beforeSequence: childSequenceBounds?.beforeSequence, excludedTypes: THREAD_TIMELINE_EXCLUDED_EVENT_TYPES, excludeDiagnosticEvents: args.excludeDiagnosticEvents, @@ -820,25 +827,51 @@ function loadTimelineContextRows( excludedTypes: THREAD_TIMELINE_EXCLUDED_EVENT_TYPES, maxInlineOutputChars, }; - const groupingContext = measureThreadTimelineStage( - profile, - "ordering-context-query", - () => - getTimelineGroupingContext(db, { - threadId: thread.id, - sequenceStart: epochSequenceStart, - maxSeq, - }), - ); + const groupingContext = + requestedTurnIds.length === 0 + ? measureThreadTimelineStage(profile, "ordering-context-query", () => + getTimelineGroupingContext(db, { + threadId: thread.id, + sequenceStart: epochSequenceStart, + maxSeq, + }), + ) + : null; let rows = listStoredTimelineThreadWindowEventRows(db, windowArgs); const itemContext = requestedTurnIds.length === 0 ? undefined : { - itemIds: listTimelineWindowItemIds(db, windowArgs), + turnIds: requestedTurnIds, + itemIds: listTimelineWindowItemIds(db, { + ...windowArgs, + turnIds: requestedTurnIds, + }), sequenceStart, beforeSequence, }; + const acceptedTurnIds = + groupingContext?.acceptedTurnIds ?? + new Map( + listStoredTurnInputAcceptedRowsByClientRequestIds(db, { + threadId: thread.id, + afterSequence: epochSequenceStart - 1, + clientRequestIds: rows.flatMap((row) => { + if (row.type !== "client/turn/requested") return []; + const requestId = tryReadClientTurnRequestedRequestId(row, decode); + return requestId === null ? [] : [requestId]; + }), + }).flatMap((row) => + row.turnId === null || row.sequence > maxSeq + ? [] + : [ + [ + parseAcceptedInputClientRequestId(row, decode), + row.turnId, + ] as const, + ], + ), + ); const initialTurnIds = [ ...requestedTurnIds, ...listTimelineRootWindowTurnIds(db, windowArgs), @@ -846,9 +879,7 @@ function loadTimelineContextRows( if (row.type !== "client/turn/requested") return []; const requestId = tryReadClientTurnRequestedRequestId(row, decode); const turnId = - requestId === null - ? undefined - : groupingContext.acceptedTurnIds.get(requestId); + requestId === null ? undefined : acceptedTurnIds.get(requestId); return turnId === undefined ? [] : [turnId]; }), ]; @@ -873,6 +904,7 @@ function loadTimelineContextRows( sequenceStart: epochSequenceStart, beforeSequence: maxSeq + 1, turnIds, + excludedEventIds: selectedRows.map((row) => row.id), itemContext, }), ]); @@ -999,7 +1031,7 @@ function loadTimelineContextRows( .filter((row) => !visibleSequences.has(row.sequence)) .map((row) => row.sequence), ), - orderingBoundarySequence: groupingContext.orderingBoundarySequence, + orderingBoundarySequence: groupingContext?.orderingBoundarySequence ?? null, rows: ensureTimelineWindowTurnStartedRows(db, { threadId: thread.id, rows: mergeStoredEventRowsById([ @@ -1742,7 +1774,7 @@ function buildTimelineTurnSummaryDetailsPage( row.sequence <= snapshot.maxSeq && (row.type !== "turn/input/accepted" || startedTurnIds.has(row.turnId)), ) - .map((row) => toThreadEventWithMeta(row)); + .map((row) => withRowMeta(row, decodeStoredEventRowCached(db, row))); const children = buildThreadTimelineTurnDetailsFromEvents({ events: compactThreadTimelineSummaryEvents(projectionEvents), options: { diff --git a/apps/server/test/services/threads/timeline-parented-pagination.test.ts b/apps/server/test/services/threads/timeline-parented-pagination.test.ts index 4d198da85d8..c7185ea70d8 100644 --- a/apps/server/test/services/threads/timeline-parented-pagination.test.ts +++ b/apps/server/test/services/threads/timeline-parented-pagination.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, + buildTimelineTurnSummaryDetails, +} from "../../../src/services/threads/timeline.js"; const providerThreadId = "provider-root"; @@ -57,6 +60,7 @@ function requestId(value: number): ClientTurnRequestId { function insertCrossWindowSubagentEvents( db: DbConnection, thread: Thread, + childStartSequence = 50, ): void { const firstRequestId = requestId(1); const secondRequestId = requestId(2); @@ -269,7 +273,7 @@ function insertCrossWindowSubagentEvents( }, { threadId: thread.id, - sequence: 50, + sequence: childStartSequence, type: "turn/started", scope: turnScope("child-turn"), providerThreadId, @@ -328,6 +332,106 @@ function rowTexts(rows: readonly TimelineRow[]): string[] { } describe("thread timeline parented pagination", () => { + it.each([49, 54])( + "preserves a child command starting at %i whose completion omits its parent call", + (commandSequence) => { + const { db, thread } = setup(); + try { + insertCrossWindowSubagentEvents(db, thread, 48); + insertEvents(db, noopNotifier, [ + { + threadId: thread.id, + sequence: commandSequence, + type: "item/started", + scope: turnScope("child-turn"), + providerThreadId, + itemId: "child-command", + itemKind: "commandExecution", + parentToolCallId: "toolu_agent_1", + data: JSON.stringify({ + item: { + type: "commandExecution", + id: "child-command", + command: "sleep 10", + cwd: "/tmp", + status: "pending", + approvalStatus: null, + }, + }), + }, + { + threadId: thread.id, + sequence: commandSequence + 3, + type: "item/completed", + scope: turnScope("child-turn"), + providerThreadId, + itemId: "child-command", + itemKind: "commandExecution", + parentToolCallId: null, + data: JSON.stringify({ + item: { + type: "commandExecution", + id: "child-command", + command: "sleep 10", + cwd: "/tmp", + status: "failed", + approvalStatus: null, + exitCode: -1, + aggregatedOutput: "stopped", + }, + }), + }, + { + threadId: thread.id, + sequence: 53, + type: "turn/completed", + scope: turnScope("parent-turn"), + providerThreadId, + itemId: null, + itemKind: null, + parentToolCallId: null, + data: JSON.stringify({ status: "completed" }), + }, + ]); + const full = buildThreadTimelineWithProfile(db, thread, { + completedTurnDisplay: "collapse", + eventBudget: 1_000_000, + includeDiagnosticOperations: false, + includeNestedRows: true, + maxInlineOutputChars: null, + maxSeq: Math.max(53, commandSequence + 3), + page: { kind: "latest", segmentLimit: 20 }, + }).response; + const summary = full.rows.find( + (row) => row.kind === "turn" && row.turnId === "parent-turn", + ); + if (summary?.kind !== "turn" || summary.turnId === null) + throw new Error("Missing parent summary"); + const details = buildTimelineTurnSummaryDetails(db, thread, { + turnId: summary.turnId, + sourceSeqStart: summary.sourceSeqStart, + sourceSeqEnd: summary.sourceSeqEnd, + completedTurnDisplay: "collapse", + includeDiagnosticOperations: false, + }); + expect(details.rows).toEqual(summary.children); + const command = flattenRows(details.rows).find( + (row) => + row.kind === "work" && + row.workKind === "command" && + row.callId === "child-command", + ); + expect(command).toMatchObject({ + status: "error", + exitCode: -1, + sourceSeqEnd: commandSequence + 3, + }); + } finally { + db.$client.close(); + } + }, + ); + it("keeps child-only subagent output off the latest page", () => { const { db, thread } = setup(); insertCrossWindowSubagentEvents(db, thread); diff --git a/apps/server/test/services/threads/timeline-structural-output.test.ts b/apps/server/test/services/threads/timeline-structural-output.test.ts index 01e8078bb1a..998aad7ccbb 100644 --- a/apps/server/test/services/threads/timeline-structural-output.test.ts +++ b/apps/server/test/services/threads/timeline-structural-output.test.ts @@ -69,6 +69,19 @@ function fixture(splitTurn = false) { }); if (splitTurn) { add("turn/started", {}, null, null, "overlap"); + add( + "item/completed", + { + item: { + type: "agentMessage", + id: "unrelated-answer", + text: "Other task", + }, + }, + "unrelated-answer", + "agentMessage", + "overlap", + ); for (let index = 0; index < 100; index++) { const id = `overlap-${index}`; add( @@ -207,7 +220,9 @@ describe("timeline command output selection", () => { beforeSequence: 1000, maxInlineOutputChars: null, itemContext: { + turnIds: [summary.turnId], itemIds: listTimelineWindowItemIds(db, { + turnIds: [summary.turnId], threadId: thread.id, sequenceStart: summary.sourceSeqStart, beforeSequence: summary.sourceSeqEnd + 1, @@ -221,17 +236,17 @@ describe("timeline command output selection", () => { context .filter((row) => row.itemKind === "commandExecution") .map((row) => row.itemId), - ).toEqual([ - "overlap-0", - "command-0", - "selected", - "overlap-0", - "selected-tail", - ]); + ).toEqual(["command-0", "selected", "selected-tail"]); expect( context.filter((row) => row.itemKind === "contextCompaction"), ).toHaveLength(2); - expect(context.some((row) => row.type === "turn/completed")).toBe(true); + expect( + context + .filter( + (row) => row.turnId === "overlap" && !row.type.startsWith("item/"), + ) + .map((row) => row.type), + ).toEqual(["turn/started", "turn/completed"]); expect( context.filter((row) => row.itemKind === "agentMessage"), ).toHaveLength(3); diff --git a/docs/timeline-pagination.md b/docs/timeline-pagination.md index 59ddc1b5b51..078c02e0055 100644 --- a/docs/timeline-pagination.md +++ b/docs/timeline-pagination.md @@ -130,15 +130,19 @@ intents are parsed when a command row is rendered, not for hidden summary childr Other event payloads and nested delegation structure are still processed upfront. Pages and summary expansion use the same conversation-context loader. Expansion -first selects item identities from metadata in the requested interval, then loads -their event histories within the snapshot. This preserves updates to items that +first selects item identities from metadata in the requested interval, restricted to +the requested turn and parented child events, then loads their event histories within +the snapshot. This preserves updates to items that started outside the interval without loading unrelated command, reasoning, and -file-change payloads from the same turn. Turn and request state, assistant messages, -and compaction lifecycle events remain context: they affect grouping or represent -state shared across the turn. Parent and child context is still resolved by the -same loader. The requested interval limits detail selection, not those dependencies. -Expansion matches the requested turn and exact summary bounds, including nested -summaries. +file-change payloads from the same turn. Assistant messages and compaction lifecycle +events remain context within the requested turn. Other turns contribute turn and +request state; their item payloads are loaded only for selected item IDs or parented +descendants of the selected work. Selected item histories span turns so a later +completion still updates the command that originally started it. Parent and child context is resolved by the same loader, +with already-loaded event IDs excluded from subsequent turn and child reads. The requested interval limits detail selection, not those dependencies. +Expansion resolves only referenced request IDs, skips page-ordering boundary work, +and reuses the event-decode cache. It matches the requested turn and exact summary +bounds, including nested summaries. The existing row-output expansion use of the endpoint selects rows owned by its requested range when the range does not identify a summary; this also handles a command finishing between preview and expansion. diff --git a/packages/db/src/data/events.ts b/packages/db/src/data/events.ts index 818307d1479..b0b658f8842 100644 --- a/packages/db/src/data/events.ts +++ b/packages/db/src/data/events.ts @@ -1202,6 +1202,7 @@ export interface FindStoredEventRowArgs { } export interface ListStoredEventRowsByParentToolCallIdsArgs { + excludedEventIds?: readonly string[]; includeCommandOutput?: boolean; excludeDiagnosticEvents?: boolean; beforeSequence?: number; @@ -1751,6 +1752,10 @@ function storedEventRowsByParentToolCallIdsConditions( conditions.push(lt(events.sequence, args.beforeSequence)); } + if (args.excludedEventIds?.length) { + conditions.push(sql`${events.id} NOT IN (SELECT value FROM json_each(${JSON.stringify(args.excludedEventIds)}))`); + } + return conditions; } @@ -3199,13 +3204,20 @@ export function hasTimelineTurnEventsInWindow( export function listTimelineWindowItemIds( db: DbConnection, - args: ListStoredTimelineWindowEventRowsArgs, + args: ListStoredTimelineWindowEventRowsArgs & { turnIds: readonly string[] }, ): string[] { return db .selectDistinct({ itemId: sql`${events.itemId}` }) .from(events) .where( - and(...storedTimelineWindowConditions(args), isNotNull(events.itemId)), + and( + ...storedTimelineWindowConditions(args), + or( + inArray(events.turnId, [...args.turnIds]), + isNotNull(events.parentToolCallId), + ), + isNotNull(events.itemId), + ), ) .all() .map((row) => row.itemId); @@ -3215,8 +3227,10 @@ export function listStoredTimelineTurnEventRows( db: DbConnection, args: ListStoredTimelineWindowEventRowsArgs & { turnIds: readonly string[]; + excludedEventIds?: readonly string[]; includeCommandOutput?: boolean; itemContext?: { + turnIds: readonly string[]; itemIds: readonly string[]; sequenceStart: number; beforeSequence: number; @@ -3229,36 +3243,55 @@ export function listStoredTimelineTurnEventRows( variableCountPerValue: 1, dedupeKey: (turnId) => turnId, fixedVariableCount: 32, - queryBatch: (turnIds) => - db + queryBatch: (turnIds) => { + const query = db .select( - storedEventRowSqlFields(args.maxInlineOutputChars, args.includeCommandOutput), + Object.fromEntries( + Object.entries( + storedEventRowSqlFields( + args.maxInlineOutputChars, + args.includeCommandOutput, + ), + ).map(([name, field]) => [name, field.as(name)]), + ), ) .from( sql`${events} INDEXED BY events_thread_turn_type_item_sequence_idx`, ) .where( and( - ...storedTimelineWindowConditions(args), - inArray(events.turnId, [...turnIds]), args.itemContext === undefined ? undefined : or( - and( - gte(events.sequence, args.itemContext.sequenceStart), - lt(events.sequence, args.itemContext.beforeSequence), - ), sql`${events.type} NOT LIKE 'item/%'`, - inArray(events.itemKind, [ - "agentMessage", - "contextCompaction", - ]), - eq(events.type, "item/agentMessage/delta"), sql`${events.itemId} IN (SELECT value FROM json_each(${JSON.stringify(args.itemContext.itemIds)}))`, + and( + inArray(events.turnId, [...args.itemContext.turnIds]), + or( + and( + gte(events.sequence, args.itemContext.sequenceStart), + lt(events.sequence, args.itemContext.beforeSequence), + ), + inArray(events.itemKind, [ + "agentMessage", + "contextCompaction", + ]), + eq(events.type, "item/agentMessage/delta"), + ), + ), ), + ...storedTimelineWindowConditions(args), + inArray(events.turnId, [...turnIds]), + args.excludedEventIds?.length + ? sql`${events.id} NOT IN (SELECT value FROM json_each(${JSON.stringify(args.excludedEventIds)}))` + : undefined, ), ) - .all(), + .toSQL(); + return db.$client + .prepare(query.sql) + .all(...query.params); + }, }).sort((left, right) => left.sequence - right.sequence); } diff --git a/packages/db/test/query-plans.test.ts b/packages/db/test/query-plans.test.ts index 13dc4142c44..77d531f1532 100644 --- a/packages/db/test/query-plans.test.ts +++ b/packages/db/test/query-plans.test.ts @@ -463,6 +463,7 @@ describe("slow query index plans", () => { try { const queries = captureStatements(db, () => { listTimelineWindowItemIds(db, { + turnIds: ["selected-turn"], threadId: thread.id, sequenceStart: 100, beforeSequence: 200, @@ -475,6 +476,7 @@ describe("slow query index plans", () => { beforeSequence: 1000, maxInlineOutputChars: null, itemContext: { + turnIds: ["selected-turn"], itemIds: ["selected-item"], sequenceStart: 100, beforeSequence: 200, @@ -710,29 +712,36 @@ describe("slow query index plans", () => { db.$client.close(); }); - it("loads parented timeline rows through the normalized parent index", () => { - const { db, thread } = setup(); + it.each([ + { excludedEventIds: [] }, + { excludedEventIds: ["already-loaded-event"] }, + ])( + "loads parented timeline rows through the normalized parent index with exclusions %j", + ({ excludedEventIds }) => { + const { db, thread } = setup(); - const [query] = captureStatements(db, () => { + const [query] = captureStatements(db, () => { + expect( + listStoredEventRowsByParentToolCallIds(db, { + maxInlineOutputChars: null, + parentToolCallIds: ["parent-tool-call"], + excludedEventIds, + threadId: thread.id, + }), + ).toEqual([]); + }); + if (!query) { + throw new Error("Expected the parented timeline row lookup SQL"); + } expect( - listStoredEventRowsByParentToolCallIds(db, { - maxInlineOutputChars: null, - parentToolCallIds: ["parent-tool-call"], - threadId: thread.id, - }), - ).toEqual([]); - }); - if (!query) { - throw new Error("Expected the parented timeline row lookup SQL"); - } - expect( - queryPlanDetails({ db, params: query.params, sql: query.sql }), - ).toMatch( - /SEARCH events USING INDEX events_parent_tool_call_thread_parent_sequence_idx/u, - ); + queryPlanDetails({ db, params: query.params, sql: query.sql }), + ).toMatch( + /SEARCH events USING INDEX events_parent_tool_call_thread_parent_sequence_idx/u, + ); - db.$client.close(); - }); + db.$client.close(); + }, + ); it("scans background-task history once without a completed-set join", () => { const { db, thread } = setup();