diff --git a/apps/app/src/components/thread/timeline/ThreadTimelineRows.tsx b/apps/app/src/components/thread/timeline/ThreadTimelineRows.tsx index c0d1ae1fc4..c3f083e5d4 100644 --- a/apps/app/src/components/thread/timeline/ThreadTimelineRows.tsx +++ b/apps/app/src/components/thread/timeline/ThreadTimelineRows.tsx @@ -113,8 +113,8 @@ import { } from "./timeline-row-containment.js"; import { NESTED_TIMELINE_GROUP_LINE_CLASS_NAME } from "./timeline-nested-group-line.js"; import { getThreadRoutePath } from "@/lib/route-paths"; -import { useThreadTimelineTurnSummaryDetails } from "@/hooks/queries/thread-queries"; -import { type ThreadTimelineTurnSummaryDetailsQueryIdentity } from "@/hooks/queries/query-keys"; +import { useThreadTimelineTurnDetails } from "@/hooks/queries/thread-queries"; +import { type ThreadTimelineTurnDetailsQueryIdentity } from "@/hooks/queries/query-keys"; import { useSenderThreadMetadataById, type SenderThreadMetadata, @@ -378,14 +378,6 @@ interface TimelineRowTitleRenderStateCache { state: TimelineRowTitleRenderState; } -interface BuildTurnSummaryDetailsIdentityArgs { - rowSourceSeqEnd: TimelineViewTurnRow["sourceSeqEnd"]; - rowSourceSeqStart: TimelineViewTurnRow["sourceSeqStart"]; - rowThreadId: TimelineViewTurnRow["threadId"]; - rowTurnId: TimelineViewTurnRow["turnId"]; - threadId: string | undefined; -} - interface TimelineRowsOwnerKeyArgs { threadId: string | undefined; timelineRows: readonly TimelineRow[]; @@ -646,21 +638,6 @@ function useTimelineSearchExpansionRowIds( }, [inheritedRowIds, location.state, rows, threadId]); } -function buildTurnSummaryDetailsIdentity({ - rowSourceSeqEnd, - rowSourceSeqStart, - rowThreadId, - rowTurnId, - threadId, -}: BuildTurnSummaryDetailsIdentityArgs): ThreadTimelineTurnSummaryDetailsQueryIdentity { - return { - sourceSeqEnd: rowSourceSeqEnd, - sourceSeqStart: rowSourceSeqStart, - threadId: threadId ?? rowThreadId, - turnId: rowTurnId, - }; -} - function timelineRowsOwnerKey({ threadId, timelineRows, @@ -1511,28 +1488,27 @@ function LazyTurnRowBody({ showAssistantMessageActions, }: LazyTurnRowBodyProps) { const { getViewRows, threadId } = useTimelineRendererStaticContext(); - const { - sourceSeqEnd: rowSourceSeqEnd, - sourceSeqStart: rowSourceSeqStart, - threadId: rowThreadId, - turnId: rowTurnId, - } = row; - const identity = useMemo( - () => - buildTurnSummaryDetailsIdentity({ - rowSourceSeqEnd, - rowSourceSeqStart, - rowThreadId, - rowTurnId, - threadId, - }), - [rowSourceSeqEnd, rowSourceSeqStart, rowThreadId, rowTurnId, threadId], + const { threadId: rowThreadId, turnId: rowTurnId } = row; + const identity = useMemo( + () => ({ + threadId: threadId ?? rowThreadId, + turnId: rowTurnId, + }), + [rowThreadId, rowTurnId, threadId], ); const { data: detail, + fetchNextPage, + hasNextPage, isError, + isFetchingNextPage, refetch, - } = useThreadTimelineTurnSummaryDetails(identity); + } = useThreadTimelineTurnDetails(identity); + useEffect(() => { + if (hasNextPage && !isFetchingNextPage && !isError) { + void fetchNextPage(); + } + }, [fetchNextPage, hasNextPage, isError, isFetchingNextPage]); const handleRetry = useCallback((): void => { void refetch(); }, [refetch]); diff --git a/apps/app/src/components/thread/timeline/TimelineRowDetails.output-preview.test.tsx b/apps/app/src/components/thread/timeline/TimelineRowDetails.output-preview.test.tsx index 8cabf3290f..ea7ee458ed 100644 --- a/apps/app/src/components/thread/timeline/TimelineRowDetails.output-preview.test.tsx +++ b/apps/app/src/components/thread/timeline/TimelineRowDetails.output-preview.test.tsx @@ -67,6 +67,7 @@ afterEach(() => { describe("previewed command output", () => { it("loads the full output for an expanded finished row through row-scoped turn details", async () => { timelineTurnSummaryDetails.mockResolvedValue({ + page: null, rows: [ commandRow({ id: "cmd_big", diff --git a/apps/app/src/hooks/queries/query-keys.ts b/apps/app/src/hooks/queries/query-keys.ts index 3d1dd90129..4fdccb947e 100644 --- a/apps/app/src/hooks/queries/query-keys.ts +++ b/apps/app/src/hooks/queries/query-keys.ts @@ -354,6 +354,16 @@ export interface ThreadTimelineTurnSummaryDetailsQueryIdentity { threadId: string; turnId: string; } +export interface ThreadTimelineTurnDetailsQueryIdentity { + threadId: string; + turnId: string; +} +type ThreadTimelineTurnDetailsQueryKey = readonly [ + typeof THREAD_TIMELINE_TURN_SUMMARY_DETAILS_QUERY_KEY, + string, + string, + "pages", +]; type ThreadTimelineTurnSummaryDetailsQueryKey = readonly [ typeof THREAD_TIMELINE_TURN_SUMMARY_DETAILS_QUERY_KEY, string, @@ -932,6 +942,18 @@ export function threadTimelineTurnSummaryDetailsQueryKey({ ]; } +export function threadTimelineTurnDetailsQueryKey({ + threadId, + turnId, +}: ThreadTimelineTurnDetailsQueryIdentity): ThreadTimelineTurnDetailsQueryKey { + return [ + THREAD_TIMELINE_TURN_SUMMARY_DETAILS_QUERY_KEY, + threadId, + turnId, + "pages", + ]; +} + export function threadTimelineQueryKeyPrefix( threadId: string, ): ThreadTimelineQueryKeyPrefix { diff --git a/apps/app/src/hooks/queries/thread-queries.test.tsx b/apps/app/src/hooks/queries/thread-queries.test.tsx index 075fdddf16..5ebf454e15 100644 --- a/apps/app/src/hooks/queries/thread-queries.test.tsx +++ b/apps/app/src/hooks/queries/thread-queries.test.tsx @@ -7,6 +7,7 @@ import type { SidebarBootstrapResponse, ThreadTimelineResponse, ThreadWithIncludesResponse, + TimelineTurnSummaryDetailsResponse, } from "@bb/server-contract"; import { COMPACT_VIEWPORT_QUERY } from "@bb/shared-ui/hooks/use-compact-viewport"; import * as api from "@/lib/api"; @@ -34,6 +35,7 @@ import { useThreadQueuedMessages, useThreadStorageLocation, useThreadTimeline, + useThreadTimelineTurnDetails, } from "./thread-queries"; vi.mock("@/lib/api", async (importOriginal) => { @@ -52,6 +54,7 @@ vi.mock("@/lib/sdk", () => ({ queuedMessages: { list: vi.fn() }, storageLocation: vi.fn(), timeline: vi.fn(), + timelineTurnSummaryDetails: vi.fn(), }, }, })); @@ -179,6 +182,71 @@ beforeEach(() => { }); }); +describe("useThreadTimelineTurnDetails", () => { + it("loads opaque forward pages sequentially and joins their rows", async () => { + vi.mocked(sdk.threads.timelineTurnSummaryDetails).mockImplementation( + async (input) => { + const isFirstPage = input.mode === "page" && input.cursor === undefined; + return { + page: { + nextCursor: isFirstPage ? "cursor-2" : null, + }, + rows: [ + { + id: isFirstPage ? "work-1" : "work-5", + threadId: "thread-1", + turnId: "turn-1", + sourceSeqStart: isFirstPage ? 1 : 5, + sourceSeqEnd: isFirstPage ? 1 : 5, + startedAt: isFirstPage ? 1 : 5, + createdAt: isFirstPage ? 1 : 5, + kind: "system", + systemKind: "debug", + title: "Work", + detail: null, + status: null, + }, + ], + } satisfies TimelineTurnSummaryDetailsResponse; + }, + ); + const { wrapper } = createQueryClientTestHarness(); + + const result = renderHook( + () => + useThreadTimelineTurnDetails({ + threadId: "thread-1", + turnId: "turn-1", + }), + { wrapper }, + ); + + await waitFor(() => { + expect(result.result.current.data?.rows.map((row) => row.id)).toEqual([ + "work-1", + ]); + }); + await act(async () => { + await result.result.current.fetchNextPage(); + }); + await waitFor(() => { + expect(result.result.current.data?.rows.map((row) => row.id)).toEqual([ + "work-1", + "work-5", + ]); + }); + expect(sdk.threads.timelineTurnSummaryDetails).toHaveBeenCalledTimes(2); + const firstInput = vi.mocked(sdk.threads.timelineTurnSummaryDetails).mock + .calls[0]?.[0]; + expect(firstInput).toMatchObject({ mode: "page", turnId: "turn-1" }); + expect(firstInput && "cursor" in firstInput).toBe(false); + expect(sdk.threads.timelineTurnSummaryDetails).toHaveBeenNthCalledWith( + 2, + expect.objectContaining({ mode: "page", cursor: "cursor-2" }), + ); + }); +}); + describe("useThreadDetailBootstrap", () => { it("starts the timeline request before the thread bootstrap settles", async () => { let resolveThread: diff --git a/apps/app/src/hooks/queries/thread-queries.ts b/apps/app/src/hooks/queries/thread-queries.ts index a3c72ceecc..d211dfebbe 100644 --- a/apps/app/src/hooks/queries/thread-queries.ts +++ b/apps/app/src/hooks/queries/thread-queries.ts @@ -25,6 +25,7 @@ import type { TimelineTurnSummaryDetailsResponse, } from "@bb/server-contract"; import { applyTimelineDelta } from "@bb/server-contract"; +import { coalesceTimelineTurnDetailPageRows } from "@bb/thread-view"; import type { ThreadListFilters } from "@bb/client-core"; import type { FilePreview } from "@bb/client-core"; import type { PathListOptions } from "@/lib/path-list-options"; @@ -78,8 +79,10 @@ import { threadHostFilePreviewQueryKey, threadConversationOutlineQueryKey, threadTimelineQueryKey, + threadTimelineTurnDetailsQueryKey, threadTimelineTurnSummaryDetailsQueryKey, threadsQueryKey, + type ThreadTimelineTurnDetailsQueryIdentity, type ThreadTimelineTurnSummaryDetailsQueryIdentity, } from "./query-keys"; import { ARCHIVED_THREADS_PAGE_SIZE } from "./archived-threads-page-size"; @@ -1040,10 +1043,20 @@ export function useThreadTimelineTurnSummaryDetails( identity: ThreadTimelineTurnSummaryDetailsQueryIdentity, options?: ThreadTimelineTurnSummaryDetailsQueryOptions, ) { - return useQuery({ + return useQuery( + threadTimelineTurnSummaryDetailsQueryOptions(identity, options), + ); +} + +function threadTimelineTurnSummaryDetailsQueryOptions( + identity: ThreadTimelineTurnSummaryDetailsQueryIdentity, + options?: ThreadTimelineTurnSummaryDetailsQueryOptions, +) { + return { queryKey: threadTimelineTurnSummaryDetailsQueryKey(identity), - queryFn: ({ signal }) => + queryFn: ({ signal }: { signal: AbortSignal }) => sdk.threads.timelineTurnSummaryDetails({ + mode: "range", threadId: requireThreadId( identity.threadId, "useThreadTimelineTurnSummaryDetails", @@ -1064,7 +1077,47 @@ export function useThreadTimelineTurnSummaryDetails( refetchOnMount: options?.refetchOnMount ?? true, staleTime: options?.staleTime ?? Infinity, ...HEAVY_PAYLOAD_QUERY_POLICY, + }; +} + +/** Load a completed turn's details forward, one bounded server page at a time. */ +export function useThreadTimelineTurnDetails( + identity: ThreadTimelineTurnDetailsQueryIdentity, +) { + const query = useInfiniteQuery({ + queryKey: threadTimelineTurnDetailsQueryKey(identity), + queryFn: ({ pageParam, signal }) => + sdk.threads.timelineTurnSummaryDetails({ + ...(pageParam ? { cursor: pageParam } : {}), + mode: "page", + signal, + threadId: requireThreadId( + identity.threadId, + "useThreadTimelineTurnDetails", + ), + turnId: identity.turnId, + }), + initialPageParam: null as string | null, + getNextPageParam: (lastPage) => lastPage.page?.nextCursor ?? undefined, + enabled: Boolean(identity.threadId) && Boolean(identity.turnId), + meta: { + errorMessage: "Failed to load turn details.", + showErrorToast: false, + }, + refetchOnMount: true, + staleTime: Infinity, + ...HEAVY_PAYLOAD_QUERY_POLICY, }); + return { + ...query, + data: query.data + ? { + rows: coalesceTimelineTurnDetailPageRows( + query.data.pages.map((page) => page.rows), + ), + } + : undefined, + }; } export function getLatestPendingInteraction( diff --git a/apps/cli/src/__tests__/command-output/thread-log.test.ts b/apps/cli/src/__tests__/command-output/thread-log.test.ts index 8e09775726..d0dd16fb0a 100644 --- a/apps/cli/src/__tests__/command-output/thread-log.test.ts +++ b/apps/cli/src/__tests__/command-output/thread-log.test.ts @@ -9,6 +9,7 @@ import { import type { CommandRegistrar } from "../helpers/command-output-harness.js"; import * as fixtures from "../helpers/command-output-fixtures.js"; import { registerThreadCommands } from "../../commands/thread/index.js"; +import type { TimelineTurnRow } from "@bb/server-contract"; describe("bb thread log command output", () => { setupCommandOutputTestEnvironment(); @@ -465,6 +466,62 @@ describe("bb thread log command output", () => { expect(output).not.toContain("older history omitted"); }); + it("bb thread log --all renders byte-window fragments as one worked-for row", async () => { + const turnFragment = (start: number, end: number): TimelineTurnRow => ({ + ...fixtures.makeTimelineBase({ + id: "thread-log:turn-1:turn", + sourceSeqStart: start, + sourceSeqEnd: end, + startedAt: 1_000, + createdAt: 9_000, + }), + turnId: "turn-1", + kind: "turn", + status: "completed", + summaryCount: 1, + completedAt: 9_000, + children: null, + }); + const getTimeline = vi.fn( + async (input: { query: { beforeAnchorSeq?: string } }) => { + if (input.query.beforeAnchorSeq === undefined) { + return { + ...fixtures.makeTimelineResponse([turnFragment(5, 8)]), + timelinePage: { + kind: "latest" as const, + segmentLimit: 100, + returnedSegmentCount: 1, + hasOlderRows: true, + olderCursor: { + anchorSeq: 5, + anchorId: "thread-log:byte-window:5", + }, + }, + }; + } + return { + ...fixtures.makeTimelineResponse([turnFragment(1, 4)]), + timelinePage: { + kind: "older" as const, + segmentLimit: 100, + returnedSegmentCount: 1, + hasOlderRows: false, + olderCursor: null, + }, + }; + }, + ); + stubServerApi({ + "v1.threads.:id.timeline.$get": getTimeline, + }); + + await runCommand(["thread", "log", "thread-log", "--all"], register); + + const output = String(vi.mocked(console.log).mock.calls[0]?.[0]); + expect(output.match(/Worked for/g)).toHaveLength(1); + expect(output).toContain("Worked for (8s)"); + }); + it("bb thread log rejects --all combined with --limit", async () => { stubServerApi({ "v1.threads.:id.timeline.$get": vi.fn(async () => diff --git a/apps/cli/src/commands/thread/show.ts b/apps/cli/src/commands/thread/show.ts index 8f05f7f09b..680c630303 100644 --- a/apps/cli/src/commands/thread/show.ts +++ b/apps/cli/src/commands/thread/show.ts @@ -1,5 +1,6 @@ import { Command } from "commander"; import { + coalesceTimelineTurnPageRows, formatThreadTimelineText, type ThreadTimelineTextFormat, } from "@bb/thread-view"; @@ -528,10 +529,13 @@ export function registerShowCommand( page = older.timelinePage; } const color = process.stdout.isTTY === true && !process.env.NO_COLOR; - const text = formatThreadTimelineText(rows, { - verbose: format === "verbose", - color, - }); + const text = formatThreadTimelineText( + coalesceTimelineTurnPageRows(rows), + { + verbose: format === "verbose", + color, + }, + ); const notice = page.hasOlderRows ? `(Showing the newest ${page.returnedSegmentCount} user-message turns; older history omitted. Use --limit (max ${THREAD_LOG_TIMELINE_SEGMENT_LIMIT_MAX}) or --all to see more.)` : null; diff --git a/apps/mobile/src/data/thread-detail/index.ts b/apps/mobile/src/data/thread-detail/index.ts index c1f414d836..127ef92c30 100644 --- a/apps/mobile/src/data/thread-detail/index.ts +++ b/apps/mobile/src/data/thread-detail/index.ts @@ -5,7 +5,7 @@ export { useThreadDetailBootstrap, useThreadPendingInteractions, useThreadQueuedMessages, - useTimelineTurnSummaryDetails, + useTimelineTurnDetails, } from "./thread-detail-queries"; export { useThreadTimelineController } from "./use-thread-timeline-controller"; export { useChildThreadSummary } from "./use-child-thread-summary"; 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 43370ee394..012bb4af79 100644 --- a/apps/mobile/src/data/thread-detail/thread-detail-queries.ts +++ b/apps/mobile/src/data/thread-detail/thread-detail-queries.ts @@ -5,9 +5,13 @@ import type { ThreadQueuedMessageListResponse, ThreadTimelineResponse, ThreadWithIncludesResponse, - TimelineTurnSummaryDetailsResponse, } from "@bb/server-contract"; -import { useQuery, useQueryClient } from "@tanstack/react-query"; +import { coalesceTimelineTurnDetailPageRows } from "@bb/thread-view"; +import { + useInfiniteQuery, + useQuery, + useQueryClient, +} from "@tanstack/react-query"; import { useProfileClient } from "@/app-shell/ProfilesProvider"; import { shouldRetryTransientReadQuery, @@ -20,8 +24,8 @@ import { threadPendingInteractionsQueryKey, threadQueuedMessagesQueryKey, threadTimelineQueryKey, - threadTimelineTurnSummaryDetailsQueryKey, - type ThreadTimelineTurnSummaryDetailsQueryIdentity, + threadTimelineTurnDetailsQueryKey, + type ThreadTimelineTurnDetailsQueryIdentity, } from "@/lib/query/query-keys"; import { requireEnabledQueryArg } from "../shared/query-helpers"; import { SESSION_STATIC_QUERY_POLICY } from "../shared/query-policies"; @@ -218,44 +222,45 @@ export function useThreadQueuedMessages( }); } -/** - * Lazy children of one completed-turn summary row - * (`GET /threads/:id/timeline/turn-summary-details`). Immutable for the - * identity (turn + source sequence span), so it never goes stale; a history - * rewrite invalidates every window of the thread. - */ -export function useTimelineTurnSummaryDetails( - identity: ThreadTimelineTurnSummaryDetailsQueryIdentity, - options?: QueryOptions, +/** Load a completed turn's details forward, one bounded server page at a time. */ +export function useTimelineTurnDetails( + identity: ThreadTimelineTurnDetailsQueryIdentity, ) { const { sdk } = useProfileClient(); - const enabled = - (options?.enabled ?? true) && - Boolean(identity.threadId) && - Boolean(identity.turnId); - - return useQuery({ - queryKey: threadTimelineTurnSummaryDetailsQueryKey(identity), - queryFn: ({ signal }) => + const query = useInfiniteQuery({ + queryKey: threadTimelineTurnDetailsQueryKey(identity), + queryFn: ({ pageParam, signal }) => sdk.threads.timelineTurnSummaryDetails({ + ...(pageParam ? { cursor: pageParam } : {}), + mode: "page", + signal, threadId: requireEnabledQueryArg({ value: identity.threadId, - hookName: "useTimelineTurnSummaryDetails", - argName: "thread id", + hookName: "useTimelineTurnDetails", + argName: "threadId", + }), + turnId: requireEnabledQueryArg({ + value: identity.turnId, + hookName: "useTimelineTurnDetails", + argName: "turnId", }), - sourceSeqEnd: String(identity.sourceSeqEnd), - sourceSeqStart: String(identity.sourceSeqStart), - turnId: identity.turnId, - signal, }), - enabled, - meta: { - errorMessage: "Failed to load turn summary details.", - showErrorToast: false, - }, + initialPageParam: null as string | null, + getNextPageParam: (lastPage) => lastPage.page?.nextCursor ?? undefined, + enabled: Boolean(identity.threadId) && Boolean(identity.turnId), refetchOnMount: true, staleTime: Infinity, }); + return { + ...query, + data: query.data + ? { + rows: coalesceTimelineTurnDetailPageRows( + query.data.pages.map((page) => page.rows), + ), + } + : undefined, + }; } /** diff --git a/apps/mobile/src/lib/query/query-keys.ts b/apps/mobile/src/lib/query/query-keys.ts index 40c18c81ff..e0203b3dbf 100644 --- a/apps/mobile/src/lib/query/query-keys.ts +++ b/apps/mobile/src/lib/query/query-keys.ts @@ -143,19 +143,15 @@ type ThreadTimelineQueryKey = readonly [ typeof THREAD_TIMELINE_QUERY_KEY, string, ]; -/** Identity of one lazily loaded completed-turn detail window. */ -export interface ThreadTimelineTurnSummaryDetailsQueryIdentity { - sourceSeqEnd: number; - sourceSeqStart: number; +export interface ThreadTimelineTurnDetailsQueryIdentity { threadId: string; turnId: string; } -type ThreadTimelineTurnSummaryDetailsQueryKey = readonly [ +type ThreadTimelineTurnDetailsQueryKey = readonly [ typeof THREAD_TIMELINE_TURN_SUMMARY_DETAILS_QUERY_KEY, string, string, - number, - number, + "pages", ]; type ThreadTimelineTurnSummaryDetailsQueryKeyPrefix = readonly [ typeof THREAD_TIMELINE_TURN_SUMMARY_DETAILS_QUERY_KEY, @@ -386,18 +382,15 @@ export function threadTimelineQueryKey( return [THREAD_TIMELINE_QUERY_KEY, threadId]; } -export function threadTimelineTurnSummaryDetailsQueryKey({ - sourceSeqEnd, - sourceSeqStart, +export function threadTimelineTurnDetailsQueryKey({ threadId, turnId, -}: ThreadTimelineTurnSummaryDetailsQueryIdentity): ThreadTimelineTurnSummaryDetailsQueryKey { +}: ThreadTimelineTurnDetailsQueryIdentity): ThreadTimelineTurnDetailsQueryKey { return [ THREAD_TIMELINE_TURN_SUMMARY_DETAILS_QUERY_KEY, threadId, turnId, - sourceSeqStart, - sourceSeqEnd, + "pages", ]; } diff --git a/apps/mobile/src/screens/thread/timeline/TurnChildrenLoader.tsx b/apps/mobile/src/screens/thread/timeline/TurnChildrenLoader.tsx index f9b384dce0..7b2b162e73 100644 --- a/apps/mobile/src/screens/thread/timeline/TurnChildrenLoader.tsx +++ b/apps/mobile/src/screens/thread/timeline/TurnChildrenLoader.tsx @@ -1,11 +1,11 @@ import { useCallback, useEffect, useState, type ReactElement } from "react"; -import { useTimelineTurnSummaryDetails } from "@/data/thread-detail"; -import type { ThreadTimelineTurnSummaryDetailsQueryIdentity } from "@/lib/query/query-keys"; +import { useTimelineTurnDetails } from "@/data/thread-detail"; +import type { ThreadTimelineTurnDetailsQueryIdentity } from "@/lib/query/query-keys"; import type { TimelineListItem, TimelineTurnChildrenState } from "./rows"; interface TurnChildrenLoaderProps { itemKey: string; - identity: ThreadTimelineTurnSummaryDetailsQueryIdentity; + identity: ThreadTimelineTurnDetailsQueryIdentity; onChange: (itemKey: string, state: TimelineTurnChildrenState | null) => void; } @@ -19,9 +19,18 @@ function TurnChildrenLoader({ identity, onChange, }: TurnChildrenLoaderProps) { - const query = useTimelineTurnSummaryDetails(identity); - const data = query.data; - const isError = query.isError; + const { + data, + fetchNextPage, + hasNextPage, + isError, + isFetchingNextPage, + } = useTimelineTurnDetails(identity); + useEffect(() => { + if (hasNextPage && !isFetchingNextPage && !isError) { + void fetchNextPage(); + } + }, [fetchNextPage, hasNextPage, isError, isFetchingNextPage]); useEffect(() => { if (data) { onChange(itemKey, { status: "loaded", rows: data.rows }); @@ -90,8 +99,6 @@ export function renderTurnChildrenLoaders( key={item.key} itemKey={item.key} identity={{ - sourceSeqEnd: row.sourceSeqEnd, - sourceSeqStart: row.sourceSeqStart, threadId: threadId || row.threadId, turnId: row.turnId, }} diff --git a/apps/mobile/src/screens/thread/timeline/renderers/turn/TurnRow.tsx b/apps/mobile/src/screens/thread/timeline/renderers/turn/TurnRow.tsx index eaa142d8bd..6a1c108a58 100644 --- a/apps/mobile/src/screens/thread/timeline/renderers/turn/TurnRow.tsx +++ b/apps/mobile/src/screens/thread/timeline/renderers/turn/TurnRow.tsx @@ -13,7 +13,7 @@ import { isPastTimelineRow } from "../shared/row-dim"; * `turn` renderer: a completed turn's recap header ("Worked for 8m 14s") or * the live "Working" row. Expanding reveals the turn's rows as flattened * children one level in; turns outside the loaded window fetch them lazily - * (`useTimelineTurnSummaryDetails` through the list's loaders), so the row + * (`useTimelineTurnDetails` through the list's loaders), so the row * shows the load state under its header until they arrive. */ export function TurnRow({ diff --git a/apps/server/src/routes/threads/data.ts b/apps/server/src/routes/threads/data.ts index 6aa5e42cce..904b2900db 100644 --- a/apps/server/src/routes/threads/data.ts +++ b/apps/server/src/routes/threads/data.ts @@ -455,16 +455,36 @@ export function registerThreadDataRoutes(app: Hono, deps: AppDeps): void { deps.config.isDevelopment || getAppSettings(deps.db).showUnhandledProviderEvents; return context.json( - buildTimelineTurnSummaryDetails(deps.db, thread, { - includeProviderUnhandledOperations, - providerDisplayName: resolveThreadProviderDisplayName( - deps, - thread.providerId, - ), - turnId: query.turnId, - sourceSeqStart: parseInteger(query.sourceSeqStart, "sourceSeqStart"), - sourceSeqEnd: parseInteger(query.sourceSeqEnd, "sourceSeqEnd"), - }), + buildTimelineTurnSummaryDetails( + deps.db, + thread, + query.mode === "page" + ? { + ...(query.cursor ? { cursor: query.cursor } : {}), + eventBudget: deps.config.featureFlags.timelineWindowEventBudget, + includeProviderUnhandledOperations, + mode: "page", + providerDisplayName: resolveThreadProviderDisplayName( + deps, + thread.providerId, + ), + turnId: query.turnId, + } + : { + includeProviderUnhandledOperations, + mode: "range", + providerDisplayName: resolveThreadProviderDisplayName( + deps, + thread.providerId, + ), + turnId: query.turnId, + sourceSeqStart: parseInteger( + query.sourceSeqStart, + "sourceSeqStart", + ), + sourceSeqEnd: parseInteger(query.sourceSeqEnd, "sourceSeqEnd"), + }, + ), ); }); diff --git a/apps/server/src/services/threads/timeline.ts b/apps/server/src/services/threads/timeline.ts index af4ed9330a..402adeef03 100644 --- a/apps/server/src/services/threads/timeline.ts +++ b/apps/server/src/services/threads/timeline.ts @@ -4,6 +4,7 @@ import { buildThreadTimelineTurnDetailsFromEvents, compactThreadTimelineSummaryEvents, type AcceptedClientRequestContext, + type EventProjectionTurnWindowCoverage, type ThreadEventWithMeta, } from "@bb/thread-view"; import { LEGACY_CODEX_GOAL_EXTENSION_KIND } from "@bb/domain"; @@ -25,6 +26,7 @@ import type { } from "@bb/server-contract"; import { findStoredTimelineWindowByteBudgetFloor, + findStoredTimelineWindowForwardBudgetCeiling, findTimelineWindowBudgetFloorSequence, getStoredEventRowsByParentToolCallIdsDataBytes, getEnvironment, @@ -60,6 +62,7 @@ import type { StandardTimelineSegmentAnchorRow, StoredEventRow, } from "@bb/db"; +import { z } from "zod"; import { ApiError } from "../../errors.js"; import { roundDurationMs } from "../lib/duration.js"; import { runEventLoopWorkSync } from "../system/event-loop-work.js"; @@ -156,11 +159,29 @@ interface BuildThreadTimelineOptions { planCommand?: ProviderComposerCommand | null; } -interface BuildTimelineTurnSummaryDetailsOptions extends TimelineTurnSummarySelection { +interface BuildTimelineTurnSummaryDetailsBaseOptions { includeProviderUnhandledOperations: boolean; providerDisplayName?: string; } +interface BuildTimelineTurnSummaryDetailsRangeOptions + extends + BuildTimelineTurnSummaryDetailsBaseOptions, + TimelineTurnSummarySelection { + mode: "range"; +} + +interface BuildTimelineTurnSummaryDetailsPageOptions extends BuildTimelineTurnSummaryDetailsBaseOptions { + cursor?: string; + eventBudget: number; + mode: "page"; + turnId: string; +} + +type BuildTimelineTurnSummaryDetailsOptions = + | BuildTimelineTurnSummaryDetailsPageOptions + | BuildTimelineTurnSummaryDetailsRangeOptions; + export const THREAD_TIMELINE_DEFAULT_SEGMENT_LIMIT = 20; export const THREAD_TIMELINE_SEGMENT_LIMIT_MAX = 100; @@ -243,6 +264,37 @@ interface TimelineEventRowSelection { strategy: ThreadTimelineEventSelectionStrategy; } +function resolveTurnWindowCoverage( + selection: TimelineEventRowSelection, +): ReadonlyMap | undefined { + const sequenceStart = selection.byteWindowSequenceStart; + const sequenceEnd = selection.byteWindowSequenceEnd; + if (sequenceStart === null || sequenceEnd === null) { + return undefined; + } + + const completionOwnershipById = new Map(); + for (const row of selection.rows) { + if (row.turnId === null || row.type !== "turn/completed") { + continue; + } + const isOwned = + row.sequence >= sequenceStart && row.sequence <= sequenceEnd; + completionOwnershipById.set( + row.turnId, + (completionOwnershipById.get(row.turnId) ?? false) || isOwned, + ); + } + + const partialCoverage = new Map(); + for (const [turnId, ownsCompletion] of completionOwnershipById) { + if (!ownsCompletion) { + partialCoverage.set(turnId, { ownsCompletion: false }); + } + } + return partialCoverage.size === 0 ? undefined : partialCoverage; +} + interface TimelineWindowRowsArgs { rows: readonly StoredEventRow[]; threadId: string; @@ -1539,6 +1591,7 @@ function buildSequencePageTimelineRows( return rowsWithPlaceholder.flatMap((row): TimelineRow[] => { if ( row.kind !== "turn" || + row.status === "pending" || selection.byteWindowSequenceEnd === null || selection.byteWindowSequenceStart === null ) { @@ -1566,7 +1619,10 @@ function buildSequencePageTimelineRows( return [ { ...row, - id: `${row.id}${suffix}`, + // Byte windows are transport slices of one logical completed turn. + // Keep its canonical identity so clients can coalesce the slices into + // one "Worked for" row. Detail pagination is a separate resource and + // does not depend on these transport boundaries. sourceSeqEnd, sourceSeqStart, }, @@ -1740,6 +1796,7 @@ function buildThreadTimelineInternal( contextOnlyToolCallIds: eventSelection.contextOnlyToolCallIds, includeNestedRows, providerId: thread.providerId, + turnWindowCoverageById: resolveTurnWindowCoverage(eventSelection), turnMessageDetail: includeNestedRows ? "full" : "summary", }, }), @@ -1947,10 +2004,10 @@ export function buildThreadConversationOutline( }); } -export function buildTimelineTurnSummaryDetails( +function buildTimelineTurnSummaryDetailsRange( db: DbConnection, thread: Thread, - options: BuildTimelineTurnSummaryDetailsOptions, + options: BuildTimelineTurnSummaryDetailsRangeOptions, ): TimelineTurnSummaryDetailsResponse { if (options.sourceSeqStart > options.sourceSeqEnd) { throw new ApiError( @@ -2084,8 +2141,7 @@ export function buildTimelineTurnSummaryDetails( // route actually holds, so the parent expansion spends what is left rather // than a pre-closure estimate of it. The subtraction may go negative, which // is the safe direction: the parent fetch then stays inside its bounds. - const detailsEventDataBytes = - byteLengthOfStoredEventRows(wholeItemEventRows); + const detailsEventDataBytes = byteLengthOfStoredEventRows(wholeItemEventRows); const eventRowsWithParentedChildren = ensureTimelineWindowParentedRows(db, { maxInlineOutputChars: detailsInlineOutputLimit, outOfBoundsChildDataByteLimit: @@ -2130,6 +2186,7 @@ export function buildTimelineTurnSummaryDetails( if (children.kind !== "missing-match") { return { + page: null, rows: children.rows, }; } @@ -2138,3 +2195,156 @@ export function buildTimelineTurnSummaryDetails( `Timeline turn summary details could not match range ${options.sourceSeqStart}-${options.sourceSeqEnd}`, ); } + +const TURN_DETAILS_CURSOR_PREFIX = "turn-details-v1:"; +const turnDetailsCursorPayloadSchema = z.object({ + sequenceStart: z.number().int(), + threadId: z.string().min(1), + turnId: z.string().min(1), +}); + +function encodeTurnDetailsCursor(args: { + sequenceStart: number; + threadId: string; + turnId: string; +}): string { + const payload = Buffer.from(JSON.stringify(args), "utf8").toString( + "base64url", + ); + return `${TURN_DETAILS_CURSOR_PREFIX}${payload}`; +} + +function parseTurnDetailsCursor( + cursor: string, + resource: { threadId: string; turnId: string }, + bounds: { sourceSeqEnd: number; sourceSeqStart: number }, +): number { + let decoded: unknown; + try { + const encoded = cursor.slice(TURN_DETAILS_CURSOR_PREFIX.length); + decoded = JSON.parse(Buffer.from(encoded, "base64url").toString("utf8")); + } catch { + decoded = null; + } + const parsed = cursor.startsWith(TURN_DETAILS_CURSOR_PREFIX) + ? turnDetailsCursorPayloadSchema.safeParse(decoded) + : null; + if ( + parsed === null || + !parsed.success || + parsed.data.threadId !== resource.threadId || + parsed.data.turnId !== resource.turnId || + !Number.isSafeInteger(parsed.data.sequenceStart) || + parsed.data.sequenceStart <= bounds.sourceSeqStart || + parsed.data.sequenceStart > bounds.sourceSeqEnd + ) { + throw new ApiError( + 400, + "invalid_request", + "Turn details cursor is no longer available", + ); + } + return parsed.data.sequenceStart; +} + +function resolveCompletedTurnDetailBounds( + db: DbConnection, + threadId: string, + turnId: string, +): TimelineTurnSummarySelection { + const startedRow = listStoredTurnStartedRowsByTurnIdsUpToSequence(db, { + sequenceCutoff: Number.MAX_SAFE_INTEGER, + threadId, + turnIds: [turnId], + })[0]; + const completedRow = listStoredTurnCompletedRowsByTurnIds(db, { + threadId, + turnIds: [turnId], + }).at(-1); + if ( + !startedRow || + !completedRow || + startedRow.sequence > completedRow.sequence + ) { + throw new ApiError( + 400, + "invalid_request", + `Cannot paginate details for incomplete turn ${turnId}`, + ); + } + return { + sourceSeqEnd: completedRow.sequence, + sourceSeqStart: startedRow.sequence, + turnId, + }; +} + +function buildTimelineTurnSummaryDetailsPage( + db: DbConnection, + thread: Thread, + options: BuildTimelineTurnSummaryDetailsPageOptions, +): TimelineTurnSummaryDetailsResponse { + const bounds = resolveCompletedTurnDetailBounds( + db, + thread.id, + options.turnId, + ); + const sourceSeqStart = options.cursor + ? parseTurnDetailsCursor( + options.cursor, + { threadId: thread.id, turnId: options.turnId }, + bounds, + ) + : bounds.sourceSeqStart; + const ceiling = findStoredTimelineWindowForwardBudgetCeiling(db, { + beforeSequence: bounds.sourceSeqEnd + 1, + excludedTypes: THREAD_TIMELINE_EXCLUDED_EVENT_TYPES, + maxDataBytes: THREAD_TIMELINE_EVENT_DATA_BYTE_LIMIT, + maxEventCount: options.eventBudget, + maxInlineOutputChars: DEFAULT_MAX_INLINE_OUTPUT_CHARS, + sequenceStart: sourceSeqStart, + threadId: thread.id, + }); + if (ceiling.kind === "single-event-too-large") { + throw new ApiError( + 413, + "timeline_window_too_large", + `Timeline turn detail event ${ceiling.sequence} exceeds the safe response limit`, + ); + } + const nextCursor = + ceiling.kind === "ceiling" + ? encodeTurnDetailsCursor({ + sequenceStart: ceiling.nextSequenceStart, + threadId: thread.id, + turnId: options.turnId, + }) + : null; + const sourceSeqEnd = + ceiling.kind === "ceiling" + ? ceiling.nextSequenceStart - 1 + : bounds.sourceSeqEnd; + const detail = buildTimelineTurnSummaryDetailsRange(db, thread, { + includeProviderUnhandledOperations: + options.includeProviderUnhandledOperations, + mode: "range", + providerDisplayName: options.providerDisplayName, + sourceSeqEnd, + sourceSeqStart, + turnId: options.turnId, + }); + return { + page: { nextCursor }, + rows: detail.rows, + }; +} + +export function buildTimelineTurnSummaryDetails( + db: DbConnection, + thread: Thread, + options: BuildTimelineTurnSummaryDetailsOptions, +): TimelineTurnSummaryDetailsResponse { + return options.mode === "page" + ? buildTimelineTurnSummaryDetailsPage(db, thread, options) + : buildTimelineTurnSummaryDetailsRange(db, thread, options); +} diff --git a/apps/server/test/public/public-thread-data.test.ts b/apps/server/test/public/public-thread-data.test.ts index 41f204bd43..5ca6d33f20 100644 --- a/apps/server/test/public/public-thread-data.test.ts +++ b/apps/server/test/public/public-thread-data.test.ts @@ -990,7 +990,7 @@ describe("public thread data routes", () => { }); }); - it("hydrates timeline turn-summary details from the summary row identity and range", async () => { + it("hydrates timeline turn-summary details through range and paginated reads", async () => { await withTestHarness(async (harness) => { const { environment, thread } = seedThreadFixture(harness); @@ -1064,7 +1064,7 @@ describe("public thread data routes", () => { expect(turnRow.children).toBeNull(); const toolDetailsResponse = await harness.app.request( - `/api/v1/threads/${thread.id}/timeline/turn-summary-details?turnId=${turnRow.turnId}&sourceSeqStart=${turnRow.sourceSeqStart}&sourceSeqEnd=${turnRow.sourceSeqEnd}`, + `/api/v1/threads/${thread.id}/timeline/turn-summary-details?mode=range&turnId=${turnRow.turnId}&sourceSeqStart=${turnRow.sourceSeqStart}&sourceSeqEnd=${turnRow.sourceSeqEnd}`, ); expect(toolDetailsResponse.status).toBe(200); const toolDetails = timelineTurnSummaryDetailsResponseSchema.parse( @@ -1078,6 +1078,18 @@ describe("public thread data routes", () => { expect(detailRow.workKind).toBe("tool"); expect(detailRow.callId).toBe("tool-1"); } + + const paginatedDetailsResponse = await harness.app.request( + `/api/v1/threads/${thread.id}/timeline/turn-summary-details?mode=page&turnId=${turnRow.turnId}`, + ); + expect(paginatedDetailsResponse.status).toBe(200); + const paginatedDetails = timelineTurnSummaryDetailsResponseSchema.parse( + await readJson(paginatedDetailsResponse), + ); + expect(paginatedDetails).toEqual({ + page: { nextCursor: null }, + rows: toolDetails.rows, + }); }); }); @@ -1171,7 +1183,7 @@ describe("public thread data routes", () => { } const detailsResponse = await harness.app.request( - `/api/v1/threads/${thread.id}/timeline/turn-summary-details?turnId=${turnRow.turnId}&sourceSeqStart=${turnRow.sourceSeqStart}&sourceSeqEnd=${turnRow.sourceSeqEnd}`, + `/api/v1/threads/${thread.id}/timeline/turn-summary-details?mode=range&turnId=${turnRow.turnId}&sourceSeqStart=${turnRow.sourceSeqStart}&sourceSeqEnd=${turnRow.sourceSeqEnd}`, ); expect(detailsResponse.status).toBe(200); const details = timelineTurnSummaryDetailsResponseSchema.parse( @@ -1492,7 +1504,7 @@ describe("public thread data routes", () => { } const detailsResponse = await harness.app.request( - `/api/v1/threads/${thread.id}/timeline/turn-summary-details?turnId=${childTurnRow.turnId}&sourceSeqStart=${childTurnRow.sourceSeqStart}&sourceSeqEnd=${childTurnRow.sourceSeqEnd}`, + `/api/v1/threads/${thread.id}/timeline/turn-summary-details?mode=range&turnId=${childTurnRow.turnId}&sourceSeqStart=${childTurnRow.sourceSeqStart}&sourceSeqEnd=${childTurnRow.sourceSeqEnd}`, ); expect(detailsResponse.status).toBe(200); const details = timelineTurnSummaryDetailsResponseSchema.parse( @@ -1628,7 +1640,7 @@ describe("public thread data routes", () => { } const detailsResponse = await harness.app.request( - `/api/v1/threads/${thread.id}/timeline/turn-summary-details?turnId=${parentTurnRow.turnId}&sourceSeqStart=${parentTurnRow.sourceSeqStart}&sourceSeqEnd=${parentTurnRow.sourceSeqEnd}`, + `/api/v1/threads/${thread.id}/timeline/turn-summary-details?mode=range&turnId=${parentTurnRow.turnId}&sourceSeqStart=${parentTurnRow.sourceSeqStart}&sourceSeqEnd=${parentTurnRow.sourceSeqEnd}`, ); expect(detailsResponse.status).toBe(200); const details = timelineTurnSummaryDetailsResponseSchema.parse( @@ -1793,7 +1805,7 @@ describe("public thread data routes", () => { expect(turnRow.sourceSeqStart).toBeGreaterThan(2); const detailsResponse = await harness.app.request( - `/api/v1/threads/${thread.id}/timeline/turn-summary-details?turnId=${turnRow.turnId}&sourceSeqStart=${turnRow.sourceSeqStart}&sourceSeqEnd=${turnRow.sourceSeqEnd}`, + `/api/v1/threads/${thread.id}/timeline/turn-summary-details?mode=range&turnId=${turnRow.turnId}&sourceSeqStart=${turnRow.sourceSeqStart}&sourceSeqEnd=${turnRow.sourceSeqEnd}`, ); expect(detailsResponse.status).toBe(200); }); @@ -1924,7 +1936,7 @@ describe("public thread data routes", () => { }); const detailsResponse = await harness.app.request( - `/api/v1/threads/${thread.id}/timeline/turn-summary-details?turnId=requested-turn&sourceSeqStart=1&sourceSeqEnd=5`, + `/api/v1/threads/${thread.id}/timeline/turn-summary-details?mode=range&turnId=requested-turn&sourceSeqStart=1&sourceSeqEnd=5`, ); expect(detailsResponse.status).toBe(200); const details = timelineTurnSummaryDetailsResponseSchema.parse( @@ -1979,7 +1991,7 @@ describe("public thread data routes", () => { }); const detailsResponse = await harness.app.request( - `/api/v1/threads/${thread.id}/timeline/turn-summary-details?turnId=turn-1&sourceSeqStart=2&sourceSeqEnd=2`, + `/api/v1/threads/${thread.id}/timeline/turn-summary-details?mode=range&turnId=turn-1&sourceSeqStart=2&sourceSeqEnd=2`, ); expect(detailsResponse.status).toBe(200); const details = timelineTurnSummaryDetailsResponseSchema.parse( @@ -2002,7 +2014,7 @@ describe("public thread data routes", () => { const { thread } = seedThreadFixture(harness); const response = await harness.app.request( - `/api/v1/threads/${thread.id}/timeline/turn-summary-details?turnId=turn-1&sourceSeqStart=oops&sourceSeqEnd=2`, + `/api/v1/threads/${thread.id}/timeline/turn-summary-details?mode=range&turnId=turn-1&sourceSeqStart=oops&sourceSeqEnd=2`, ); expect(response.status).toBe(400); await expect(readJson(response)).resolves.toMatchObject({ diff --git a/apps/server/test/public/public-thread-timeline-output-preview.test.ts b/apps/server/test/public/public-thread-timeline-output-preview.test.ts index d1a6184782..1e5ded9153 100644 --- a/apps/server/test/public/public-thread-timeline-output-preview.test.ts +++ b/apps/server/test/public/public-thread-timeline-output-preview.test.ts @@ -152,7 +152,7 @@ describe("GET /threads/:id/timeline inline output preview", () => { expect(big.turnId).toBe("turn-1"); const response = await harness.app.request( - `/api/v1/threads/${threadId}/timeline/turn-summary-details?turnId=${big.turnId}&sourceSeqStart=${big.sourceSeqStart}&sourceSeqEnd=${big.sourceSeqEnd}`, + `/api/v1/threads/${threadId}/timeline/turn-summary-details?mode=range&turnId=${big.turnId}&sourceSeqStart=${big.sourceSeqStart}&sourceSeqEnd=${big.sourceSeqEnd}`, ); expect(response.status).toBe(200); const details = timelineTurnSummaryDetailsResponseSchema.parse( @@ -183,7 +183,7 @@ describe("GET /threads/:id/timeline inline output preview", () => { }); const response = await harness.app.request( - `/api/v1/threads/${threadId}/timeline/turn-summary-details?turnId=${big.turnId}&sourceSeqStart=${big.sourceSeqStart}&sourceSeqEnd=${big.sourceSeqEnd}`, + `/api/v1/threads/${threadId}/timeline/turn-summary-details?mode=range&turnId=${big.turnId}&sourceSeqStart=${big.sourceSeqStart}&sourceSeqEnd=${big.sourceSeqEnd}`, ); expect(response.status).toBe(200); const details = timelineTurnSummaryDetailsResponseSchema.parse( @@ -244,7 +244,7 @@ describe("GET /threads/:id/timeline inline output preview (tool rows)", () => { ); const response = await harness.app.request( - `/api/v1/threads/${thread.id}/timeline/turn-summary-details?turnId=${row.turnId}&sourceSeqStart=${row.sourceSeqStart}&sourceSeqEnd=${row.sourceSeqEnd}`, + `/api/v1/threads/${thread.id}/timeline/turn-summary-details?mode=range&turnId=${row.turnId}&sourceSeqStart=${row.sourceSeqStart}&sourceSeqEnd=${row.sourceSeqEnd}`, ); expect(response.status).toBe(200); const details = timelineTurnSummaryDetailsResponseSchema.parse( diff --git a/apps/server/test/public/public-thread-timeline-work-rows.test.ts b/apps/server/test/public/public-thread-timeline-work-rows.test.ts index a86611b099..77495608ca 100644 --- a/apps/server/test/public/public-thread-timeline-work-rows.test.ts +++ b/apps/server/test/public/public-thread-timeline-work-rows.test.ts @@ -67,7 +67,7 @@ async function getTurnDetails( turnRow: Extract, ): Promise { const response = await harness.app.request( - `/api/v1/threads/${threadId}/timeline/turn-summary-details?turnId=${turnRow.turnId}&sourceSeqStart=${turnRow.sourceSeqStart}&sourceSeqEnd=${turnRow.sourceSeqEnd}`, + `/api/v1/threads/${threadId}/timeline/turn-summary-details?mode=range&turnId=${turnRow.turnId}&sourceSeqStart=${turnRow.sourceSeqStart}&sourceSeqEnd=${turnRow.sourceSeqEnd}`, ); expect(response.status).toBe(200); return timelineTurnSummaryDetailsResponseSchema.parse( 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 4d0654361e..1d4a085a06 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 @@ -20,7 +20,9 @@ import type { DbConnection } from "@bb/db"; import type { TimelinePaginationCursor, TimelineRow, + ThreadTimelineResponse, } from "@bb/server-contract"; +import { coalesceTimelineTurnDetailPageRows } from "@bb/thread-view"; import { buildThreadTimeline, buildTimelineTurnSummaryDetails, @@ -85,6 +87,8 @@ function backgroundTaskData(status: "pending" | "completed"): string { } interface SeedOptions { + /** Emit this many consecutive progress messages before deferred commands finish. */ + assistantProgressCount?: 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. @@ -106,12 +110,18 @@ interface SeedOptions { longRunningItemIndexes?: readonly number[]; /** Character count for each completed command output. */ outputChars?: number; + /** Give every event a deterministic sequence-derived wall-clock time. */ + eventTimeStepMs?: 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 * rows. */ streamLongRunningOutput?: boolean; + /** Emit one final assistant response after deferred commands finish. */ + terminalAssistant?: boolean; + /** Terminal status stored on each emitted `turn/completed`. */ + turnStatus?: "completed" | "error" | "interrupted"; itemsPerTurn: readonly number[]; } @@ -129,7 +139,14 @@ function seedTurns( let sequence = 0; const push = (event: Omit): void => { sequence += 1; - events.push({ ...event, sequence, threadId: thread.id }); + events.push({ + ...event, + ...(options.eventTimeStepMs === undefined + ? {} + : { createdAt: sequence * options.eventTimeStepMs }), + sequence, + threadId: thread.id, + }); }; options.itemsPerTurn.forEach((items, index) => { @@ -299,6 +316,31 @@ function seedTurns( }), }); } + if (isLastTurn) { + for ( + let progress = 0; + progress < (options.assistantProgressCount ?? 0); + progress += 1 + ) { + const itemId = `${turnId}-progress-${progress}`; + push({ + type: "item/completed", + scope: turnScope(turnId), + providerThreadId, + itemId, + itemKind: "agentMessage", + parentToolCallId: null, + data: JSON.stringify({ + item: { + type: "agentMessage", + id: itemId, + text: `Progress ${progress}`, + }, + }), + }); + } + } + for (const item of deferred) { const itemId = `${turnId}-item-${item}`; push({ @@ -330,6 +372,25 @@ function seedTurns( }); } + if (isLastTurn && options.terminalAssistant === true) { + const itemId = `${turnId}-terminal`; + push({ + type: "item/completed", + scope: turnScope(turnId), + providerThreadId, + itemId, + itemKind: "agentMessage", + parentToolCallId: null, + data: JSON.stringify({ + item: { + type: "agentMessage", + id: itemId, + text: "Terminal response", + }, + }), + }); + } + if (parentToolCallId !== null) { push({ type: "item/completed", @@ -359,7 +420,10 @@ function seedTurns( itemId: null, itemKind: null, parentToolCallId: null, - data: JSON.stringify({ status: "completed", providerThreadId }), + data: JSON.stringify({ + status: options.turnStatus ?? "completed", + providerThreadId, + }), }); } }); @@ -477,6 +541,60 @@ function collectCommandCallIds( } } +function collectAssistantTexts( + rows: readonly TimelineRow[], + target: string[], +): void { + for (const row of rows) { + if (row.kind === "conversation" && row.role === "assistant") { + target.push(row.text); + } + if (row.kind === "work" && row.workKind === "delegation") { + collectAssistantTexts(row.childRows, target); + } + if (row.kind === "turn" && row.children !== null) { + collectAssistantTexts(row.children, target); + } + } +} + +function walkTurnDetails( + db: DbConnection, + thread: Thread, + args: { eventBudget: number; turnId: string }, +): { pageRows: TimelineRow[][]; pages: number; rows: TimelineRow[] } { + const pageRows: TimelineRow[][] = []; + const rows: TimelineRow[] = []; + const seenCursors = new Set(); + let cursor: string | undefined; + let pages = 0; + + for (;;) { + const details = buildTimelineTurnSummaryDetails(db, thread, { + ...(cursor === undefined ? {} : { cursor }), + eventBudget: args.eventBudget, + includeProviderUnhandledOperations: false, + mode: "page", + turnId: args.turnId, + }); + pages += 1; + pageRows.push(details.rows); + rows.push(...details.rows); + const nextCursor = details.page?.nextCursor ?? null; + if (nextCursor === null) { + break; + } + expect(seenCursors.has(nextCursor), `cursor loop at ${nextCursor}`).toBe( + false, + ); + seenCursors.add(nextCursor); + cursor = nextCursor; + expect(pages).toBeLessThan(100); + } + + return { pageRows, pages, rows }; +} + interface WalkResult { maxEventRowCount: number; pages: number; @@ -681,6 +799,7 @@ describe("in-turn timeline windows", () => { const details = buildTimelineTurnSummaryDetails(db, thread, { includeProviderUnhandledOperations: false, + mode: "range", sourceSeqEnd: turnRow.sourceSeqEnd, sourceSeqStart: turnRow.sourceSeqStart, turnId: turnRow.turnId, @@ -791,12 +910,28 @@ describe("in-turn timeline windows", () => { ); }); + it("pages completed-turn details forward under the event-count budget", () => { + const { db, thread } = setup(); + seedTurns(db, thread, { completeLastTurn: true, itemsPerTurn: [20] }); + + const details = walkTurnDetails(db, thread, { + eventBudget: 8, + turnId: "turn-1", + }); + const commandCallIds = new Set(); + collectCommandCallIds(details.rows, commandCallIds); + + expect(details.pages).toBeGreaterThan(1); + expect(commandCallIds.size).toBe(20); + }); + it("pages through a finished turn that exceeds the event-data byte limit", () => { const { db, thread } = setup(); seedTurns(db, thread, { commandChars: 25_000, completeLastTurn: true, itemsPerTurn: [BYTE_WINDOW_ITEM_COUNT], + turnStatus: "interrupted", }); const commandCallIds = new Set(); @@ -812,11 +947,11 @@ describe("in-turn timeline windows", () => { if (row.kind !== "turn") { continue; } - expect(row.status).toBe("completed"); - expect(turnRowIds.has(row.id)).toBe(false); + expect(row.status).toBe("interrupted"); turnRowIds.add(row.id); const details = buildTimelineTurnSummaryDetails(db, thread, { includeProviderUnhandledOperations: false, + mode: "range", sourceSeqEnd: row.sourceSeqEnd, sourceSeqStart: row.sourceSeqStart, turnId: row.turnId, @@ -844,7 +979,134 @@ describe("in-turn timeline windows", () => { expect(pages).toBeGreaterThan(2); expect(commandCallIds.size).toBe(BYTE_WINDOW_ITEM_COUNT); expect(expandedCommandCallIds.size).toBe(BYTE_WINDOW_ITEM_COUNT); - expect(turnRowIds.size).toBe(pages); + expect(turnRowIds.size).toBe(1); + + const paginatedDetails = walkTurnDetails(db, thread, { + eventBudget: LARGE_BUDGET, + turnId: "turn-1", + }); + const paginatedCommandCallIds = new Set(); + collectCommandCallIds(paginatedDetails.rows, paginatedCommandCallIds); + expect(paginatedDetails.pages).toBeGreaterThan(1); + expect(paginatedCommandCallIds.size).toBe(BYTE_WINDOW_ITEM_COUNT); + }, 15_000); + + it("keeps one logical turn identity while only one byte page owns the terminal response", () => { + const { db, thread } = setup(); + seedTurns(db, thread, { + assistantProgressCount: 3, + commandChars: 25_000, + completeLastTurn: true, + eventTimeStepMs: 1_000, + itemsPerTurn: [BYTE_WINDOW_ITEM_COUNT], + longRunningItemIndexes: [0], + terminalAssistant: true, + }); + + const allAssistantTexts: string[] = []; + const topLevelAssistantTexts: string[] = []; + const turnRows: Extract[] = []; + let cursor: TimelinePaginationCursor | null = null; + let pages = 0; + for (;;) { + const page: ThreadTimelineResponse = buildPage( + db, + thread, + LARGE_BUDGET, + cursor, + ).response; + const nestedPage: ThreadTimelineResponse = buildNestedPage( + db, + thread, + LARGE_BUDGET, + cursor, + ).response; + expect(nestedPage.timelinePage).toEqual(page.timelinePage); + pages += 1; + collectAssistantTexts(nestedPage.rows, allAssistantTexts); + for (const row of page.rows) { + if (row.kind === "conversation" && row.role === "assistant") { + topLevelAssistantTexts.push(row.text); + } + if (row.kind === "turn") { + turnRows.push(row); + } + } + if (!page.timelinePage.hasOlderRows) { + break; + } + cursor = page.timelinePage.olderCursor; + expect(cursor).not.toBeNull(); + expect(pages).toBeLessThan(10); + } + + const turnStartedAt = 2_000; + const turnCompletedAt = + getLatestThreadSequence(db, { threadId: thread.id }) * 1_000; + expect(pages).toBeGreaterThan(1); + expect(topLevelAssistantTexts).toEqual(["Terminal response"]); + expect(allAssistantTexts.sort()).toEqual([ + "Progress 0", + "Progress 1", + "Progress 2", + "Terminal response", + ]); + expect(turnRows).not.toHaveLength(0); + expect( + turnRows.every( + (row) => + row.startedAt === turnStartedAt && + row.completedAt === turnCompletedAt, + ), + ).toBe(true); + expect(new Set(turnRows.map((row) => row.id)).size).toBe(1); + }, 15_000); + + it("keeps a sole work summary canonical when completed responses stay visible", () => { + const { db, thread } = setup(); + seedTurns(db, thread, { + assistantProgressCount: 2, + commandChars: 25_000, + completeLastTurn: true, + itemsPerTurn: [BYTE_WINDOW_ITEM_COUNT], + terminalAssistant: true, + }); + + const assistantTexts: string[] = []; + const turnRowIds = new Set(); + let cursor: TimelinePaginationCursor | null = null; + let pages = 0; + for (;;) { + const page: ThreadTimelineResponse = buildPage( + db, + thread, + LARGE_BUDGET, + cursor, + ).response; + pages += 1; + for (const row of page.rows) { + if (row.kind === "conversation" && row.role === "assistant") { + assistantTexts.push(row.text); + } + if (row.kind === "turn") { + turnRowIds.add(row.id); + } + } + if (!page.timelinePage.hasOlderRows) { + break; + } + cursor = page.timelinePage.olderCursor; + expect(cursor).not.toBeNull(); + expect(pages).toBeLessThan(10); + } + + expect(pages).toBeGreaterThan(1); + expect(assistantTexts.sort()).toEqual([ + "Progress 0", + "Progress 1", + "Terminal response", + ]); + expect([...turnRowIds]).toEqual([`${thread.id}:turn-1:turn`]); }, 15_000); it("keeps latest byte-page row identities stable while a turn grows", () => { @@ -918,6 +1180,7 @@ describe("in-turn timeline windows", () => { } const details = buildTimelineTurnSummaryDetails(db, thread, { includeProviderUnhandledOperations: false, + mode: "range", sourceSeqEnd: turnRow.sourceSeqEnd, sourceSeqStart: turnRow.sourceSeqStart, turnId: turnRow.turnId, @@ -965,6 +1228,7 @@ describe("in-turn timeline windows", () => { } const details = buildTimelineTurnSummaryDetails(db, thread, { includeProviderUnhandledOperations: false, + mode: "range", sourceSeqEnd: row.sourceSeqEnd, sourceSeqStart: row.sourceSeqStart, turnId: row.turnId, @@ -990,6 +1254,18 @@ describe("in-turn timeline windows", () => { expect(pages).toBeGreaterThan(2); expect(commandCallIds.size).toBe(BYTE_WINDOW_ITEM_COUNT); expect(expandedCommandCallIds.size).toBe(BYTE_WINDOW_ITEM_COUNT); + + const paginatedDetails = walkTurnDetails(db, thread, { + eventBudget: LARGE_BUDGET, + turnId: "turn-1", + }); + const coalescedCommandCallIds = new Set(); + collectCommandCallIds( + coalesceTimelineTurnDetailPageRows(paginatedDetails.pageRows), + coalescedCommandCallIds, + ); + expect(paginatedDetails.pages).toBeGreaterThan(1); + expect(coalescedCommandCallIds.size).toBe(BYTE_WINDOW_ITEM_COUNT); }, 15_000); it("returns a placeholder when one event exceeds the byte limit", () => { @@ -1090,6 +1366,7 @@ describe("in-turn timeline windows", () => { } const details = buildTimelineTurnSummaryDetails(db, thread, { includeProviderUnhandledOperations: false, + mode: "range", sourceSeqEnd: row.sourceSeqEnd, sourceSeqStart: row.sourceSeqStart, turnId: row.turnId, @@ -1603,6 +1880,7 @@ function collectTurnDetailsAndChildren( children: row.children ?? [], details: buildTimelineTurnSummaryDetails(db, thread, { includeProviderUnhandledOperations: false, + mode: "range", sourceSeqEnd: row.sourceSeqEnd, sourceSeqStart: row.sourceSeqStart, turnId: row.turnId, diff --git a/packages/client-core/src/timeline/timeline-merge.ts b/packages/client-core/src/timeline/timeline-merge.ts index 97d6bfcd95..9b11babb3d 100644 --- a/packages/client-core/src/timeline/timeline-merge.ts +++ b/packages/client-core/src/timeline/timeline-merge.ts @@ -3,6 +3,7 @@ import type { TimelinePaginationCursor, TimelineRow, } from "@bb/server-contract"; +import { mergeTimelineTurnPageRows } from "@bb/thread-view"; import { isOptimisticTimelineRowId } from "./optimistic-timeline-row.js"; type NullableTimelinePaginationCursor = TimelinePaginationCursor | null; @@ -98,12 +99,17 @@ function appendTimelineRowsPreservingOrder( target: TimelineRow[], rows: readonly TimelineRow[], ): void { - const seenIds = new Set(target.map((row) => row.id)); + const rowIndexById = new Map(target.map((row, index) => [row.id, index])); for (const row of rows) { - if (seenIds.has(row.id)) { + const existingIndex = rowIndexById.get(row.id); + if (existingIndex !== undefined) { + const existing = target[existingIndex]; + if (existing?.kind === "turn" && row.kind === "turn") { + target[existingIndex] = mergeTimelineTurnPageRows(existing, row); + } continue; } - seenIds.add(row.id); + rowIndexById.set(row.id, target.length); target.push(row); } } @@ -254,7 +260,14 @@ export function mergeLatestTimelineRows({ if (rowsBefore) { rows.push(...rowsBefore); } - rows.push(latestRowsById.get(row.id) ?? row); + const latestRow = latestRowsById.get(row.id); + rows.push( + row.kind === "turn" && latestRow?.kind === "turn" + ? mergeTimelineTurnPageRows(row, latestRow, { + newerWindowStartSequence: latestWindowStartSequence, + }) + : (latestRow ?? row), + ); } rows.push(...pendingRows); if (areTimelineRowReferencesEqual({ left: loadedRows, right: rows })) { diff --git a/packages/client-core/test/timeline-merge.test.ts b/packages/client-core/test/timeline-merge.test.ts index 353da9243a..c6c219c25c 100644 --- a/packages/client-core/test/timeline-merge.test.ts +++ b/packages/client-core/test/timeline-merge.test.ts @@ -25,6 +25,7 @@ interface TimelineTestRowArgs { interface TimelineTurnTestRowArgs extends TimelineTestRowArgs { children?: TimelineRow[]; endSequence?: number; + summaryCount?: number; } function timelineCursor(args: TimelineTestRowArgs): TimelinePaginationCursor { @@ -95,7 +96,7 @@ function turnSummaryRow(args: TimelineTurnTestRowArgs): TimelineTurnRow { createdAt: args.sequence, kind: "turn", status: "completed", - summaryCount: 1, + summaryCount: args.summaryCount ?? 1, completedAt: args.sequence, children: args.children ?? null, }; @@ -183,7 +184,7 @@ describe("timeline page row merging", () => { ]); }); - it("keeps distinct byte-budget slices of one finished turn", () => { + it("coalesces byte-budget slices into one finished turn", () => { const olderCommands = [ commandRow({ id: "command-1", sequence: 10 }), commandRow({ id: "command-2", sequence: 11 }), @@ -193,13 +194,17 @@ describe("timeline page row merging", () => { commandRow({ id: "command-4", sequence: 21 }), ]; const olderSlice = turnSummaryRow({ - id: "turn-1:sequence-page:10", + id: "turn-1", sequence: 10, + endSequence: 11, + summaryCount: 2, children: olderCommands, }); const latestSlice = turnSummaryRow({ - id: "turn-1:sequence-page:20", + id: "turn-1", sequence: 20, + endSequence: 21, + summaryCount: 2, children: latestCommands, }); @@ -208,10 +213,12 @@ describe("timeline page row merging", () => { loadedRows: [latestSlice], }); - expect(rows.map((row) => row.id)).toEqual([ - "turn-1:sequence-page:10", - "turn-1:sequence-page:20", - ]); + expect(rows.map((row) => row.id)).toEqual(["turn-1"]); + expect(rows[0]).toMatchObject({ + sourceSeqStart: 10, + sourceSeqEnd: 21, + summaryCount: 4, + }); expect( rows.flatMap((row) => row.kind === "turn" && row.children !== null diff --git a/packages/db/src/data/events.ts b/packages/db/src/data/events.ts index b266d57484..81b841129e 100644 --- a/packages/db/src/data/events.ts +++ b/packages/db/src/data/events.ts @@ -1234,6 +1234,28 @@ export interface FindStoredTimelineWindowByteBudgetFloorArgs maxDataBytes: number; } +export interface FindStoredTimelineWindowForwardBudgetCeilingArgs + extends Omit { + /** Exclusive upper bound of the logical resource being paginated. */ + beforeSequence: number; + maxDataBytes: number; + maxEventCount: number; +} + +export type StoredTimelineWindowForwardBudgetCeiling = + | { eventCount: number; eventDataBytes: number; kind: "fits" } + | { + eventCount: number; + eventDataBytes: number; + kind: "ceiling"; + nextSequenceStart: number; + } + | { + eventDataBytes: number; + kind: "single-event-too-large"; + sequence: number; + }; + export type StoredTimelineWindowByteBudgetFloor = | { eventDataBytes: number; kind: "fits" } | { eventDataBytes: number; kind: "floor"; sequenceStart: number } @@ -3104,6 +3126,56 @@ export function findStoredTimelineWindowByteBudgetFloor( return { eventDataBytes: includedDataBytes, kind: "fits" }; } +/** + * Finds the exclusive upper bound of the oldest prefix that fits both the + * event-count and stored-byte budgets. Unlike the main timeline's newest-first + * floor, completed-turn details page forward so clients can render them in + * chronological order as pages arrive. + */ +export function findStoredTimelineWindowForwardBudgetCeiling( + db: DbConnection, + args: FindStoredTimelineWindowForwardBudgetCeilingArgs, +): StoredTimelineWindowForwardBudgetCeiling { + const data = storedTimelineWindowDataColumn(args.maxInlineOutputChars); + const candidates = db + .select({ + dataBytes: sql`length(CAST(${data} AS BLOB))`.as("data_bytes"), + sequence: events.sequence, + }) + .from(events) + .where(and(...storedTimelineWindowConditions(args))) + .orderBy(events.sequence) + .limit(args.maxEventCount + 1) + .all(); + + let eventCount = 0; + let eventDataBytes = 0; + for (const row of candidates) { + if ( + eventCount >= args.maxEventCount || + eventDataBytes + row.dataBytes > args.maxDataBytes + ) { + if (eventCount === 0) { + return { + eventDataBytes: row.dataBytes, + kind: "single-event-too-large", + sequence: row.sequence, + }; + } + return { + eventCount, + eventDataBytes, + kind: "ceiling", + nextSequenceStart: row.sequence, + }; + } + eventCount += 1; + eventDataBytes += row.dataBytes; + } + + return { eventCount, eventDataBytes, kind: "fits" }; +} + export function listStoredTimelineWindowEventRows( db: DbConnection, args: ListStoredTimelineWindowEventRowsArgs, diff --git a/packages/db/src/data/index.ts b/packages/db/src/data/index.ts index 4cadc15b1a..75c9ec8afb 100644 --- a/packages/db/src/data/index.ts +++ b/packages/db/src/data/index.ts @@ -256,6 +256,7 @@ export { listTimelineSegmentAnchorsDescending, findTimelineWindowBudgetFloorSequence, findStoredTimelineWindowByteBudgetFloor, + findStoredTimelineWindowForwardBudgetCeiling, getStoredEventRowsByParentToolCallIdsDataBytes, findUnfinishedTurnCoveringSequence, hasParentedEventCrossingSequence, diff --git a/packages/db/test/data/events.test.ts b/packages/db/test/data/events.test.ts index 1465b3418a..3bcf48227a 100644 --- a/packages/db/test/data/events.test.ts +++ b/packages/db/test/data/events.test.ts @@ -17,6 +17,7 @@ import { appendStoredThreadEventsInTransaction, findStoredEventRow, findStoredTimelineWindowByteBudgetFloor, + findStoredTimelineWindowForwardBudgetCeiling, findTimelineWindowBudgetFloorSequence, getActiveStoredTurnId, getHighWaterMarks, @@ -4743,6 +4744,82 @@ describe("timeline read-boundary output truncation", () => { })); }); + it("finds the oldest forward page that fits both event and byte budgets", () => { + const { db, thread } = setup(); + insertEvents( + db, + noopNotifier, + [100, 200, 300].map((messageChars, index) => ({ + threadId: thread.id, + sequence: index + 1, + type: "system/error" as const, + ...threadEventFields, + itemId: null, + itemKind: null, + parentToolCallId: null, + data: JSON.stringify({ message: "x".repeat(messageChars) }), + })), + ); + const window = { + beforeSequence: 4, + maxInlineOutputChars: null, + sequenceStart: 1, + threadId: thread.id, + } as const; + const rows = listStoredTimelineWindowEventRows(db, window); + const rowBytes = new Map( + rows.map((row) => [row.sequence, Buffer.byteLength(row.data)]), + ); + const firstTwoBytes = (rowBytes.get(1) ?? 0) + (rowBytes.get(2) ?? 0); + + expect( + findStoredTimelineWindowForwardBudgetCeiling(db, { + ...window, + maxDataBytes: Number.MAX_SAFE_INTEGER, + maxEventCount: 2, + }), + ).toEqual({ + eventCount: 2, + eventDataBytes: firstTwoBytes, + kind: "ceiling", + nextSequenceStart: 3, + }); + expect( + findStoredTimelineWindowForwardBudgetCeiling(db, { + ...window, + maxDataBytes: firstTwoBytes, + maxEventCount: 3, + }), + ).toEqual({ + eventCount: 2, + eventDataBytes: firstTwoBytes, + kind: "ceiling", + nextSequenceStart: 3, + }); + expect( + findStoredTimelineWindowForwardBudgetCeiling(db, { + ...window, + maxDataBytes: getStoredTimelineWindowEventDataBytes(db, window), + maxEventCount: 3, + }), + ).toEqual({ + eventCount: 3, + eventDataBytes: getStoredTimelineWindowEventDataBytes(db, window), + kind: "fits", + }); + expect( + findStoredTimelineWindowForwardBudgetCeiling(db, { + ...window, + maxDataBytes: (rowBytes.get(1) ?? 0) - 1, + maxEventCount: 3, + }), + ).toEqual({ + eventDataBytes: rowBytes.get(1), + kind: "single-event-too-large", + sequence: 1, + }); + }); + it("bounds the byte-total preflight before using the early-stopping iterator", () => { const { db, thread } = setup(); const validData = JSON.stringify({ message: "valid" }); diff --git a/packages/sdk/src/areas/threads.ts b/packages/sdk/src/areas/threads.ts index d1a25551d7..245d4b897b 100644 --- a/packages/sdk/src/areas/threads.ts +++ b/packages/sdk/src/areas/threads.ts @@ -248,10 +248,22 @@ export interface ThreadStoragePathsArgs extends ThreadStoragePathsQuery { threadId: string; } -export interface ThreadTimelineTurnSummaryDetailsArgs extends TimelineTurnSummaryDetailsQuery { - signal?: AbortSignal; - threadId: string; -} +type ThreadTimelineTurnSummaryDetailsPageArgs = Extract< + TimelineTurnSummaryDetailsQuery, + { mode: "page" } +>; +type ThreadTimelineTurnSummaryDetailsRangeArgs = Omit< + Extract, + "mode" +> & { mode?: "range" }; + +export type ThreadTimelineTurnSummaryDetailsArgs = ( + | ThreadTimelineTurnSummaryDetailsPageArgs + | ThreadTimelineTurnSummaryDetailsRangeArgs +) & { + signal?: AbortSignal; + threadId: string; + }; export interface ThreadTabsUpdateArgs extends UpdateThreadTabsRequest { threadId: string; @@ -1118,11 +1130,19 @@ export function createThreadsArea(args: CreateSdkAreaArgs): ThreadsArea { transport.api.v1.threads[":id"].timeline["turn-summary-details"].$get( { param: { id: input.threadId }, - query: { - turnId: input.turnId, - sourceSeqStart: input.sourceSeqStart, - sourceSeqEnd: input.sourceSeqEnd, - }, + query: + input.mode === "page" + ? { + ...(input.cursor ? { cursor: input.cursor } : {}), + mode: "page", + turnId: input.turnId, + } + : { + mode: "range", + turnId: input.turnId, + sourceSeqStart: input.sourceSeqStart, + sourceSeqEnd: input.sourceSeqEnd, + }, }, ...signalRequestArgs(input.signal), ), diff --git a/packages/server-contract/src/api/threads.ts b/packages/server-contract/src/api/threads.ts index 280e7c6795..403f9c6516 100644 --- a/packages/server-contract/src/api/threads.ts +++ b/packages/server-contract/src/api/threads.ts @@ -728,11 +728,36 @@ export const threadTimelineQuerySchema = z }); export type ThreadTimelineQuery = z.infer; -export const timelineTurnSummaryDetailsQuerySchema = z.object({ - turnId: z.string().min(1), - sourceSeqStart: z.string().regex(/^\d+$/), - sourceSeqEnd: z.string().regex(/^\d+$/), -}); +const timelineTurnSummaryDetailsTurnIdSchema = z.string().min(1); +const timelineTurnSummaryDetailsSequenceQuerySchema = z.string().regex(/^\d+$/); +const timelineTurnSummaryDetailsRangeQueryFields = { + turnId: timelineTurnSummaryDetailsTurnIdSchema, + sourceSeqStart: timelineTurnSummaryDetailsSequenceQuerySchema, + sourceSeqEnd: timelineTurnSummaryDetailsSequenceQuerySchema, +} as const; + +export const timelineTurnSummaryDetailsQuerySchema = z.union([ + z.object({ + mode: z.literal("page"), + turnId: timelineTurnSummaryDetailsTurnIdSchema, + cursor: z.string().min(1).optional(), + }), + z.object({ + mode: z.literal("range"), + ...timelineTurnSummaryDetailsRangeQueryFields, + }), + z + .object({ + mode: z.never().optional(), + ...timelineTurnSummaryDetailsRangeQueryFields, + }) + .transform((value) => ({ + mode: "range" as const, + turnId: value.turnId, + sourceSeqStart: value.sourceSeqStart, + sourceSeqEnd: value.sourceSeqEnd, + })), +]); export type TimelineTurnSummaryDetailsQuery = z.infer< typeof timelineTurnSummaryDetailsQuerySchema >; @@ -811,14 +836,42 @@ export const threadFilesRawQuerySchema = z.object({ }); export type ThreadFilesRawQuery = z.infer; -export const timelineTurnSummaryDetailsRequestSchema = z.object({ - turnId: z.string().min(1), +const timelineTurnSummaryDetailsRangeRequestFields = { + turnId: timelineTurnSummaryDetailsTurnIdSchema, sourceSeqStart: z.number().int().nonnegative(), sourceSeqEnd: z.number().int().nonnegative(), -}); +} as const; + +export const timelineTurnSummaryDetailsRequestSchema = z.union([ + z.object({ + mode: z.literal("page"), + turnId: timelineTurnSummaryDetailsTurnIdSchema, + cursor: z.string().min(1).optional(), + }), + z.object({ + mode: z.literal("range"), + ...timelineTurnSummaryDetailsRangeRequestFields, + }), + z + .object({ + mode: z.never().optional(), + ...timelineTurnSummaryDetailsRangeRequestFields, + }) + .transform((value) => ({ + mode: "range" as const, + turnId: value.turnId, + sourceSeqStart: value.sourceSeqStart, + sourceSeqEnd: value.sourceSeqEnd, + })), +]); export const timelineTurnSummaryDetailsResponseSchema = z.object({ rows: z.array(timelineRowSchema), + page: z + .object({ + nextCursor: z.string().min(1).nullable(), + }) + .nullable(), }); export type TimelineTurnSummaryDetailsResponse = z.infer< typeof timelineTurnSummaryDetailsResponseSchema diff --git a/packages/server-contract/test/contract.test.ts b/packages/server-contract/test/contract.test.ts index dab2fe5e6f..dcce033b10 100644 --- a/packages/server-contract/test/contract.test.ts +++ b/packages/server-contract/test/contract.test.ts @@ -267,6 +267,16 @@ const OPTIONAL_SERVER_FIELD_GROUPS: readonly OptionalServerFieldGroup[] = [ "threadTimelineResponseSchema.delta.rowOrder", ], }, + { + reason: + "Turn-detail page requests omit the opaque cursor on the first page; legacy range requests omit mode and normalize to explicit range mode at the boundary.", + fields: [ + "timelineTurnSummaryDetailsQuerySchema.cursor", + "timelineTurnSummaryDetailsQuerySchema.mode", + "timelineTurnSummaryDetailsRequestSchema.cursor", + "timelineTurnSummaryDetailsRequestSchema.mode", + ], + }, { reason: "Uploaded attachments may omit mime type when the client could not determine one.", @@ -1067,10 +1077,23 @@ describe("server-contract canonical schemas", () => { ).toThrow("Project path must be an absolute path."); expect( - timelineTurnSummaryDetailsResponseSchema.parse({ rows: [] }), + timelineTurnSummaryDetailsResponseSchema.parse({ page: null, rows: [] }), ).toEqual({ + page: null, rows: [], }); + expect( + contract.timelineTurnSummaryDetailsQuerySchema.parse({ + turnId: "turn_123", + sourceSeqStart: "10", + sourceSeqEnd: "20", + }), + ).toEqual({ + mode: "range", + turnId: "turn_123", + sourceSeqStart: "10", + sourceSeqEnd: "20", + }); }); it("normalizes the deprecated writable alias without widening readonly", () => { @@ -1603,6 +1626,7 @@ describe("server-contract clients", () => { publicClient.threads[":id"].timeline["turn-summary-details"].$url({ param: { id: "thr_123" }, query: { + mode: "range", turnId: "turn_123", sourceSeqStart: "1", sourceSeqEnd: "2", diff --git a/packages/thread-view/src/apply-turn-message-detail.ts b/packages/thread-view/src/apply-turn-message-detail.ts index a18d3fe2a4..c9616045bb 100644 --- a/packages/thread-view/src/apply-turn-message-detail.ts +++ b/packages/thread-view/src/apply-turn-message-detail.ts @@ -98,7 +98,10 @@ function applyTurnMessageDetail( const messages = (turn.messages ?? []).map((message) => withChildProjectionDetail(message), ); - const terminalMessage = findLastTerminalTimelineMessage(messages); + const terminalMessage = + turn.status !== "pending" && turn.windowCoverage?.ownsCompletion === false + ? undefined + : findLastTerminalTimelineMessage(messages); const summaryMessages = terminalMessage ? messages.slice(0, messages.indexOf(terminalMessage)) : messages; @@ -109,7 +112,6 @@ function applyTurnMessageDetail( (turn.externalUserBoundarySeqs?.length ?? 0) > 0 || isSingletonContextManagementOperation(summaryMessages) || shouldIncludeSummaryTurnMessages(messages, terminalMessage); - const detailedTurn: EventProjectionTurn = { turnId: turn.turnId, threadId: turn.threadId, @@ -120,6 +122,7 @@ function applyTurnMessageDetail( completedAt: turn.completedAt, status: turn.status, summaryCount, + ...(turn.windowCoverage ? { windowCoverage: turn.windowCoverage } : {}), ...(turn.externalUserBoundarySeqs ? { externalUserBoundarySeqs: turn.externalUserBoundarySeqs } : {}), diff --git a/packages/thread-view/src/build-event-projection.ts b/packages/thread-view/src/build-event-projection.ts index 086c201140..b934322527 100644 --- a/packages/thread-view/src/build-event-projection.ts +++ b/packages/thread-view/src/build-event-projection.ts @@ -125,6 +125,7 @@ interface BuildDetailedProjectionArgs { contextOnlyToolCallIds?: ReadonlySet; events: ThreadEventWithMeta[]; messages: EventProjectionMessage[]; + turnWindowCoverageById?: BuildEventProjectionOptions["turnWindowCoverageById"]; turnMessageDetail: BuildEventProjectionOptions["turnMessageDetail"]; } @@ -1030,6 +1031,7 @@ function buildDetailedProjection( const projection = groupEventProjectionTurns({ events: args.events, messages: args.messages, + turnWindowCoverageById: args.turnWindowCoverageById, }); const semanticProjection = normalizeEventProjection( { @@ -1069,6 +1071,7 @@ function buildFullEventProjection( contextOnlyToolCallIds: options.contextOnlyToolCallIds, events, messages: flatProjection.messages, + turnWindowCoverageById: options.turnWindowCoverageById, turnMessageDetail: options.turnMessageDetail, }); } @@ -1104,6 +1107,7 @@ export function buildEventProjectionEntries( contextOnlyToolCallIds: options.contextOnlyToolCallIds, events: orderedEvents, messages: flatProjection.messages, + turnWindowCoverageById: options.turnWindowCoverageById, turnMessageDetail: options.turnMessageDetail, }); } diff --git a/packages/thread-view/src/build-thread-timeline.ts b/packages/thread-view/src/build-thread-timeline.ts index 6a350fc33a..4005eb0091 100644 --- a/packages/thread-view/src/build-thread-timeline.ts +++ b/packages/thread-view/src/build-thread-timeline.ts @@ -33,6 +33,7 @@ import type { EventProjectionProvisioningTranscriptEntry, EventProjectionToolParsedIntent, EventProjectionTurn, + EventProjectionTurnWindowCoverage, } from "./event-projection-types.js"; import { assertNever } from "./assert-never.js"; import { @@ -75,6 +76,15 @@ type ThreadTimelineTurnMessageDetail = "summary" | "full"; interface ThreadTimelineFromEventsBaseOptions { contextOnlyToolCallIds?: ReadonlySet; includeProviderUnhandledOperations: boolean; + /** + * Completion edges physically owned by this sequence window. Backfilled + * completions still settle partial turns, but must not select a page-local + * false terminal response. + */ + turnWindowCoverageById?: ReadonlyMap< + string, + EventProjectionTurnWindowCoverage + >; /** * Tail-only state (`pendingTodos`) is only meaningful on the latest page — * this snapshot describes current head state, not historical state. Caller @@ -187,7 +197,8 @@ interface BuildTurnSummaryRowArgs { completedAt: number | null; includeNestedRows: boolean; rowIdPrefix: string; - segmentIndex: number | null; + rowIdSegmentIndex: number | null; + sourceBounds: "messages" | "turn"; sourceMessages: EventProjectionMessage[]; sourceRows: TimelineRow[]; startedAt: number; @@ -1138,7 +1149,8 @@ function buildTurnSummaryRow({ completedAt, includeNestedRows, rowIdPrefix, - segmentIndex, + rowIdSegmentIndex, + sourceBounds, sourceMessages, sourceRows, startedAt, @@ -1150,13 +1162,13 @@ function buildTurnSummaryRow({ } const bounds = - segmentIndex === null || sourceMessages.length === 0 + sourceBounds === "turn" || sourceMessages.length === 0 ? getTurnBounds(turn) : getTimelineMessageBounds(sourceMessages); const rowId = - segmentIndex === null + rowIdSegmentIndex === null ? `${rowIdPrefix}${turn.threadId}:${turn.turnId}:turn` - : `${rowIdPrefix}${turn.threadId}:${turn.turnId}:turn:${segmentIndex}`; + : `${rowIdPrefix}${turn.threadId}:${turn.turnId}:turn:${rowIdSegmentIndex}`; const resolvedCompletedAt = completedAt ?? getTimelineMessageCompletedAt(sourceMessages); @@ -1209,7 +1221,8 @@ function buildCompletedTurnSummaryRows({ completedAt: item.completedAt, includeNestedRows, rowIdPrefix, - segmentIndex: item.segmentIndex, + rowIdSegmentIndex: item.rowIdSegmentIndex, + sourceBounds: item.sourceBounds, sourceMessages: item.sourceMessages, sourceRows, startedAt: item.startedAt, @@ -1395,6 +1408,7 @@ export function buildThreadTimelineFromEvents( providerDisplayName: args.options.providerDisplayName, threadStatus: args.options.threadStatus, threadName: args.options.threadName, + turnWindowCoverageById: args.options.turnWindowCoverageById, turnMessageDetail: args.options.turnMessageDetail, } satisfies Parameters[1]; const projection = buildEventProjection(args.events, projectionOptions); diff --git a/packages/thread-view/src/completed-turn-grouping.ts b/packages/thread-view/src/completed-turn-grouping.ts index 175612c810..8fd5a064f2 100644 --- a/packages/thread-view/src/completed-turn-grouping.ts +++ b/packages/thread-view/src/completed-turn-grouping.ts @@ -14,7 +14,8 @@ interface CompletedTurnSummaryGroup { kind: "summary"; startedAt: number; completedAt: number | null; - segmentIndex: number | null; + rowIdSegmentIndex: number | null; + sourceBounds: "messages" | "turn"; sourceMessages: EventProjectionMessage[]; summaryCount: number; } @@ -84,7 +85,7 @@ function getSummaryMessageBounds( return { startedAt }; } -function applySingleSummaryTurnBounds( +function applySingleSummaryTurnIdentityAndBounds( turn: EventProjectionTurn, items: readonly CompletedTurnSummaryItem[], ): CompletedTurnSummaryItem[] { @@ -92,6 +93,18 @@ function applySingleSummaryTurnBounds( if (summaryGroups.length !== 1) { return [...items]; } + const canUseCanonicalIdentity = + (turn.externalUserBoundarySeqs?.length ?? 0) === 0 && + !items.some( + (item) => + item.kind === "ungrouped-message" && + isTimelineUngroupableMessage(item.message) && + // The accepted request that opened the turn is rendered beside its + // work summary, but it is the start of this exchange rather than a + // boundary inside it. Later human messages still keep their segment + // identity even when only one side contains collapsible work. + item.message.sourceSeqStart > turn.sourceSeqStart, + ); const onlySummaryGroup = summaryGroups[0]; return items.map((item) => @@ -100,6 +113,13 @@ function applySingleSummaryTurnBounds( ...item, startedAt: turn.startedAt, completedAt: turn.completedAt, + // Visible assistant responses may sit beside one collapsed work + // summary, but they do not create another user exchange. Keep that + // sole summary on the canonical turn identity so a byte-window page + // that cannot see the response assigns the same id. + rowIdSegmentIndex: canUseCanonicalIdentity + ? null + : item.rowIdSegmentIndex, } : item, ); @@ -150,15 +170,41 @@ function isAssistantResponseMessage( ); } +function isTimelineWorkActivityMessage( + message: EventProjectionMessage, +): boolean { + return ( + message.kind !== "assistant-text" && + message.kind !== "user" && + message.kind !== "error" + ); +} + +function hasWorkActivityBetweenAssistantMessages( + messages: readonly EventProjectionMessage[], + current: EventProjectionMessage, + next: EventProjectionMessage, +): boolean { + return messages.some( + (message) => + isTimelineWorkActivityMessage(message) && + message.sourceSeqStart < next.sourceSeqStart && + message.sourceSeqEnd > current.sourceSeqEnd, + ); +} + /** * Assistant text that the provider followed directly with more assistant * text, with no work in between, was a complete response, not narration about - * upcoming tool activity. Providers re-query the model after it stops without - * telling bb why (a Claude Code Stop hook injects its reason as a synthetic - * user message that never becomes a thread event), so the turn carries two - * answers and only the last one is the terminal message. The earlier answer - * must stay visible at rest instead of being folded into the collapsed work - * summary. Text followed by work keeps the existing collapse. + * upcoming tool activity. "Between" includes a work item that started before + * both messages and completed after them: source ordering places that item + * before both texts, but it was still active across their interval. Providers + * re-query the model after it stops without telling bb why (a Claude Code Stop + * hook injects its reason as a synthetic user message that never becomes a + * thread event), so the turn carries two answers and only the last one is the + * terminal message. The earlier answer must stay visible at rest instead of + * being folded into the collapsed work summary. Text followed by work keeps + * the existing collapse. */ function findVisibleResponseMessageIds( summaryMessages: readonly EventProjectionMessage[], @@ -170,7 +216,12 @@ function findVisibleResponseMessageIds( const nextMessage = summaryMessages[index + 1] ?? terminalMessage; if ( isAssistantResponseMessage(message) && - isAssistantResponseMessage(nextMessage) + isAssistantResponseMessage(nextMessage) && + !hasWorkActivityBetweenAssistantMessages( + summaryMessages, + message, + nextMessage, + ) ) { visibleIds.add(message.id); } @@ -198,7 +249,8 @@ function groupCompletedTurnSummaryMessages( kind: "summary", startedAt: turn.startedAt, completedAt: turn.completedAt, - segmentIndex: null, + rowIdSegmentIndex: null, + sourceBounds: "turn", sourceMessages: summaryMessages, summaryCount: turn.summaryCount, }, @@ -220,7 +272,8 @@ function groupCompletedTurnSummaryMessages( kind: "summary", startedAt: bounds.startedAt, completedAt: null, - segmentIndex, + rowIdSegmentIndex: segmentIndex, + sourceBounds: "messages", sourceMessages, summaryCount: getProjectionSummaryCount(sourceMessages, undefined), }); @@ -295,7 +348,7 @@ function groupCompletedTurnSummaryMessages( externalBoundaryIndex += 1; } flushGroupedMessages(); - return applySingleSummaryTurnBounds(turn, items); + return applySingleSummaryTurnIdentityAndBounds(turn, items); } export function groupCompletedTurnMessages( diff --git a/packages/thread-view/src/event-projection-types.ts b/packages/thread-view/src/event-projection-types.ts index 7d95b5231f..91f01607e0 100644 --- a/packages/thread-view/src/event-projection-types.ts +++ b/packages/thread-view/src/event-projection-types.ts @@ -3,6 +3,7 @@ export type { EventProjection, EventProjectionEntry, EventProjectionTurn, + EventProjectionTurnWindowCoverage, EventProjectionTurnMessageDetail, EventProjectionTurnStatus, } from "./event-projection.js"; diff --git a/packages/thread-view/src/event-projection.ts b/packages/thread-view/src/event-projection.ts index 98f36903ea..eacfeea321 100644 --- a/packages/thread-view/src/event-projection.ts +++ b/packages/thread-view/src/event-projection.ts @@ -48,9 +48,23 @@ interface EventProjectionState { export interface BuildEventProjectionOptions extends BuildEventProjectionMessagesOptions { acceptedClientRequestContext?: AcceptedClientRequestContext; contextOnlyToolCallIds?: ReadonlySet; + /** + * Whether the current event window owns a completed turn's terminal edge. + * Sequence-window projections backfill turn lifecycle rows so a partial turn + * can settle, but a context-only completion must not select a false terminal + * response from that slice. + */ + turnWindowCoverageById?: ReadonlyMap< + string, + EventProjectionTurnWindowCoverage + >; turnMessageDetail: EventProjectionTurnMessageDetail; } +export interface EventProjectionTurnWindowCoverage { + ownsCompletion: boolean; +} + export type EventProjectionEntry = | EventProjectionMessageEntry | EventProjectionTurnEntry; @@ -75,6 +89,10 @@ export interface EventProjectionTurn { completedAt: number | null; status: EventProjectionTurnStatus; summaryCount: number; + /** + * Present only when a sequence window does not own the completion edge. + */ + windowCoverage?: EventProjectionTurnWindowCoverage; externalUserBoundarySeqs?: number[]; terminalMessage?: EventProjectionMessage; messages?: EventProjectionMessage[]; diff --git a/packages/thread-view/src/group-event-projection-turns.ts b/packages/thread-view/src/group-event-projection-turns.ts index abedb5a52d..eaf36bc478 100644 --- a/packages/thread-view/src/group-event-projection-turns.ts +++ b/packages/thread-view/src/group-event-projection-turns.ts @@ -4,6 +4,7 @@ import type { EventProjection, EventProjectionEntry, EventProjectionTurn, + EventProjectionTurnWindowCoverage, EventProjectionTurnStatus, } from "./event-projection-types.js"; import { requireThreadEventScopeTurnId } from "@bb/domain"; @@ -39,6 +40,10 @@ interface ProjectionTurnBoundsUpdate { interface GroupEventProjectionTurnsArgs { events: ThreadEventWithMeta[]; messages: EventProjectionMessage[]; + turnWindowCoverageById?: ReadonlyMap< + string, + EventProjectionTurnWindowCoverage + >; } interface TurnEntryDraft { @@ -91,6 +96,7 @@ function toEventProjectionTurnStatus( function createProjectionTurn( event: TurnStartedEvent, meta: EventMeta, + windowCoverage: EventProjectionTurnWindowCoverage | undefined, ): ProjectionTurnDraft { const turnId = requireThreadEventScopeTurnId({ type: event.type, @@ -108,6 +114,7 @@ function createProjectionTurn( completedAt: null, status: "pending", summaryCount: 0, + ...(windowCoverage ? { windowCoverage } : {}), }, }; } @@ -210,7 +217,11 @@ function createEventProjectionEntry( ); } - const terminalMessage = findLastTerminalTimelineMessage(turnDraft.messages); + const terminalMessage = + turnDraft.turn.status !== "pending" && + turnDraft.turn.windowCoverage?.ownsCompletion === false + ? undefined + : findLastTerminalTimelineMessage(turnDraft.messages); const turn: EventProjectionTurn = { ...turnDraft.turn, summaryCount: getProjectionSummaryCount( @@ -255,7 +266,14 @@ export function groupEventProjectionTurns( // lifecycle marker must not make the whole timeline unreadable. continue; } - turnsById.set(turnId, createProjectionTurn(event, meta)); + turnsById.set( + turnId, + createProjectionTurn( + event, + meta, + args.turnWindowCoverageById?.get(turnId), + ), + ); entryDrafts.push({ kind: "turn", turnId, diff --git a/packages/thread-view/src/index.ts b/packages/thread-view/src/index.ts index 748a8fb005..4fd0c2810e 100644 --- a/packages/thread-view/src/index.ts +++ b/packages/thread-view/src/index.ts @@ -1,4 +1,9 @@ export { formatThreadTimelineText } from "./format-timeline-text.js"; +export { + coalesceTimelineTurnDetailPageRows, + coalesceTimelineTurnPageRows, + mergeTimelineTurnPageRows, +} from "./merge-timeline-turn-page-rows.js"; export { parseAgentMessageEnvelope } from "./agent-message-envelope.js"; export type { ThreadTimelineTextFormat } from "./format-timeline-text.js"; export { assertNever } from "./assert-never.js"; @@ -56,6 +61,7 @@ export { export { extractThreadTimelineActivePlanTurn } from "./active-prompt-mode-extraction.js"; export { extractThreadTimelineGoal } from "./goal-snapshot-extraction.js"; export type { AcceptedClientRequestContext } from "./accepted-client-request-context.js"; +export type { EventProjectionTurnWindowCoverage } from "./event-projection-types.js"; export { buildTimelineViewRows, createTimelineViewRowsCache, diff --git a/packages/thread-view/src/merge-timeline-turn-page-rows.ts b/packages/thread-view/src/merge-timeline-turn-page-rows.ts new file mode 100644 index 0000000000..d9c8688aec --- /dev/null +++ b/packages/thread-view/src/merge-timeline-turn-page-rows.ts @@ -0,0 +1,168 @@ +import type { TimelineRow, TimelineTurnRow } from "@bb/server-contract"; + +interface MergeTimelineTurnPageRowsOptions { + /** + * Fresh rows are authoritative from this sequence onward. Inline children + * at or beyond the boundary are replaced, not appended. + */ + newerWindowStartSequence?: number; +} + +function mergeInlineChildren( + older: TimelineTurnRow, + newer: TimelineTurnRow, + newerWindowStartSequence: number | undefined, +): TimelineRow[] | null { + if (older.children === null || newer.children === null) { + return null; + } + + const rows = + newerWindowStartSequence === undefined + ? [...older.children] + : older.children.filter( + (row) => row.sourceSeqEnd < newerWindowStartSequence, + ); + return coalesceTimelineDetailRows([rows, newer.children]); +} + +/** + * Merge two byte-window fragments of the same logical completed turn. + * Transport pagination stays bounded while every renderer receives one row. + */ +export function mergeTimelineTurnPageRows( + older: TimelineTurnRow, + newer: TimelineTurnRow, + options: MergeTimelineTurnPageRowsOptions = {}, +): TimelineTurnRow { + if (older.id !== newer.id || older.turnId !== newer.turnId) { + throw new Error("Cannot merge timeline rows from different turns"); + } + const completedAtCandidates = [older.completedAt, newer.completedAt].filter( + (value): value is number => value !== null, + ); + const retainsOlderWindow = + options.newerWindowStartSequence !== undefined && + older.sourceSeqStart < options.newerWindowStartSequence; + + return { + ...newer, + children: mergeInlineChildren( + older, + newer, + options.newerWindowStartSequence, + ), + completedAt: + completedAtCandidates.length === 0 + ? null + : Math.max(...completedAtCandidates), + createdAt: Math.max(older.createdAt, newer.createdAt), + sourceSeqEnd: Math.max(older.sourceSeqEnd, newer.sourceSeqEnd), + sourceSeqStart: Math.min(older.sourceSeqStart, newer.sourceSeqStart), + startedAt: Math.min(older.startedAt, newer.startedAt), + // Completed turns are immutable. A latest-window refresh replaces the + // newest fragment but keeps the count already accumulated from older + // pages; an older-page prepend adds a disjoint fragment. + summaryCount: + options.newerWindowStartSequence === undefined + ? older.summaryCount + newer.summaryCount + : retainsOlderWindow + ? older.summaryCount + : newer.summaryCount, + }; +} + +/** Coalesce byte-window turn fragments after timeline pages are concatenated. */ +export function coalesceTimelineTurnPageRows( + rows: readonly TimelineRow[], +): TimelineRow[] { + const coalescedRows: TimelineRow[] = []; + const turnIndexById = new Map(); + for (const row of rows) { + if (row.kind !== "turn") { + coalescedRows.push(row); + continue; + } + const existingIndex = turnIndexById.get(row.id); + if (existingIndex === undefined) { + turnIndexById.set(row.id, coalescedRows.length); + coalescedRows.push(row); + continue; + } + const existing = coalescedRows[existingIndex]; + if (existing?.kind !== "turn") { + throw new Error(`Timeline row id ${row.id} changed kind across pages`); + } + coalescedRows[existingIndex] = mergeTimelineTurnPageRows(existing, row); + } + return coalescedRows; +} + +function mergeTimelineDetailRows( + older: TimelineRow, + newer: TimelineRow, +): TimelineRow { + if (older.kind === "turn" && newer.kind === "turn") { + return mergeTimelineTurnPageRows(older, newer); + } + if ( + older.kind === "work" && + older.workKind === "delegation" && + newer.kind === "work" && + newer.workKind === "delegation" + ) { + const completedAtCandidates = [older.completedAt, newer.completedAt].filter( + (value): value is number => value !== null, + ); + return { + ...newer, + childRows: coalesceTimelineDetailRows([older.childRows, newer.childRows]), + completedAt: + completedAtCandidates.length === 0 + ? null + : Math.max(...completedAtCandidates), + sourceSeqEnd: Math.max(older.sourceSeqEnd, newer.sourceSeqEnd), + sourceSeqStart: Math.min(older.sourceSeqStart, newer.sourceSeqStart), + startedAt: Math.min(older.startedAt, newer.startedAt), + }; + } + if (older.kind !== newer.kind) { + throw new Error(`Timeline row id ${newer.id} changed kind across pages`); + } + return newer; +} + +function coalesceTimelineDetailRows( + pages: readonly (readonly TimelineRow[])[], +): TimelineRow[] { + const rows: TimelineRow[] = []; + const rowIndexById = new Map(); + for (const page of pages) { + for (const row of page) { + const existingIndex = rowIndexById.get(row.id); + if (existingIndex === undefined) { + rowIndexById.set(row.id, rows.length); + rows.push(row); + continue; + } + const existing = rows[existingIndex]; + if (!existing) { + throw new Error(`Missing timeline row ${row.id} while merging pages`); + } + rows[existingIndex] = mergeTimelineDetailRows(existing, row); + } + } + return rows; +} + +/** Join forward detail pages, with the later page authoritative for closure rows. */ +export function coalesceTimelineTurnDetailPageRows( + pages: readonly (readonly TimelineRow[])[], +): TimelineRow[] { + return coalesceTimelineDetailRows(pages).sort( + (left, right) => + left.sourceSeqStart - right.sourceSeqStart || + left.sourceSeqEnd - right.sourceSeqEnd || + left.id.localeCompare(right.id), + ); +} diff --git a/packages/thread-view/src/normalize-event-projection.ts b/packages/thread-view/src/normalize-event-projection.ts index 200437b371..cd3e10a689 100644 --- a/packages/thread-view/src/normalize-event-projection.ts +++ b/packages/thread-view/src/normalize-event-projection.ts @@ -242,7 +242,11 @@ function buildSourceTurn( sourceTurn: EventProjectionTurn, messages: EventProjectionMessage[], ): EventProjectionTurn { - const terminalMessage = findLastTerminalTimelineMessage(messages); + const terminalMessage = + sourceTurn.status !== "pending" && + sourceTurn.windowCoverage?.ownsCompletion === false + ? undefined + : findLastTerminalTimelineMessage(messages); const turn: EventProjectionTurn = { ...sourceTurn, summaryCount: getProjectionSummaryCount(messages, terminalMessage), diff --git a/packages/thread-view/test/completed-turn-grouping.test.ts b/packages/thread-view/test/completed-turn-grouping.test.ts index 19d509f4b6..4080b30178 100644 --- a/packages/thread-view/test/completed-turn-grouping.test.ts +++ b/packages/thread-view/test/completed-turn-grouping.test.ts @@ -40,9 +40,18 @@ function assistantMessage( }; } -function commandMessage(args: MessageBaseArgs): EventProjectionCommandMessage { +interface CommandMessageArgs extends MessageBaseArgs { + endSeq?: number; +} + +function commandMessage( + args: CommandMessageArgs, +): EventProjectionCommandMessage { + const endSeq = args.endSeq ?? args.seq; return { ...messageBase(args), + sourceSeqEnd: endSeq, + createdAt: endSeq, kind: "command", callId: args.id, command: "pnpm test", @@ -51,7 +60,7 @@ function commandMessage(args: MessageBaseArgs): EventProjectionCommandMessage { source: null, output: "", exitCode: 0, - completedAt: args.seq, + completedAt: endSeq, approvalStatus: null, status: "completed", }; @@ -189,7 +198,8 @@ describe("groupCompletedTurnMessages", () => { kind: "summary", startedAt: 1, completedAt: 2, - segmentIndex: null, + rowIdSegmentIndex: null, + sourceBounds: "turn", summaryCount: 2, }, ]); @@ -225,7 +235,8 @@ describe("groupCompletedTurnMessages", () => { kind: "summary", startedAt: 1, completedAt: 4, - segmentIndex: 0, + rowIdSegmentIndex: null, + sourceBounds: "messages", sourceMessages: [{ id: "narration" }, { id: "command" }], summaryCount: 2, }, @@ -234,6 +245,53 @@ describe("groupCompletedTurnMessages", () => { expect(groups.terminalMessages).toEqual([hookReply]); }); + it("does not treat the accepted request at the turn start as an internal exchange boundary", () => { + const seed = userMessage({ id: "seed", seq: 1 }); + const narration = assistantMessage({ id: "narration", seq: 2 }); + const command = commandMessage({ id: "command", seq: 3 }); + const answer = assistantMessage({ id: "answer", seq: 4 }); + const hookReply = assistantMessage({ id: "hook-reply", seq: 5 }); + const groups = groupCompletedTurnMessages( + completedTurn( + [seed, narration, command, answer, hookReply], + hookReply, + ), + ); + + expect(groups.summaryItems).toMatchObject([ + { kind: "ungrouped-message", message: { id: "seed" } }, + { + kind: "summary", + rowIdSegmentIndex: null, + sourceBounds: "messages", + sourceMessages: [{ id: "narration" }, { id: "command" }], + }, + { kind: "ungrouped-message", message: { id: "answer" } }, + ]); + expect(groups.terminalMessages).toEqual([hookReply]); + }); + + it("keeps a sole work summary segmented across a later human message", () => { + const seed = userMessage({ id: "seed", seq: 1 }); + const command = commandMessage({ id: "command", seq: 2 }); + const followUp = userMessage({ id: "follow-up", seq: 3 }); + const terminal = assistantMessage({ id: "terminal", seq: 4 }); + const groups = groupCompletedTurnMessages( + completedTurn([seed, command, followUp, terminal], terminal), + ); + + expect(groups.summaryItems).toMatchObject([ + { kind: "ungrouped-message", message: { id: "seed" } }, + { + kind: "summary", + rowIdSegmentIndex: 0, + sourceBounds: "messages", + sourceMessages: [{ id: "command" }], + }, + { kind: "ungrouped-message", message: { id: "follow-up" } }, + ]); + }); + it("keeps every response in a run of adjacent assistant texts", () => { const first = assistantMessage({ id: "first", seq: 1 }); const second = assistantMessage({ id: "second", seq: 2 }); @@ -253,6 +311,32 @@ describe("groupCompletedTurnMessages", () => { expect(groups.terminalMessages).toEqual([terminal]); }); + it("folds adjacent progress updates crossed by active work", () => { + const command = commandMessage({ id: "command", seq: 1, endSeq: 5 }); + const first = assistantMessage({ id: "first-progress", seq: 2 }); + const second = assistantMessage({ id: "second-progress", seq: 3 }); + const terminal = assistantMessage({ id: "terminal", seq: 6 }); + const turn = completedTurn([command, first, second, terminal], terminal); + turn.completedAt = 7; + turn.sourceSeqEnd = 7; + + const groups = groupCompletedTurnMessages(turn); + + expect(groups.summaryItems).toMatchObject([ + { + kind: "summary", + startedAt: 1, + completedAt: 7, + rowIdSegmentIndex: null, + sourceBounds: "turn", + }, + ]); + expect(summarySourceMessageIds(groups)).toEqual([ + ["command", "first-progress", "second-progress"], + ]); + expect(groups.terminalMessages).toEqual([terminal]); + }); + it("preserves the last assistant message before an ungroupable user message", () => { const assistantBefore = assistantMessage({ id: "assistant-before", @@ -315,7 +399,8 @@ describe("groupCompletedTurnMessages", () => { kind: "summary", startedAt: 1, completedAt: 4, - segmentIndex: null, + rowIdSegmentIndex: null, + sourceBounds: "turn", summaryCount: 4, }, ]); @@ -340,7 +425,8 @@ describe("groupCompletedTurnMessages", () => { kind: "summary", startedAt: 1, completedAt: null, - segmentIndex: 0, + rowIdSegmentIndex: 0, + sourceBounds: "messages", summaryCount: 1, }, { @@ -353,7 +439,8 @@ describe("groupCompletedTurnMessages", () => { kind: "summary", startedAt: 3, completedAt: null, - segmentIndex: 1, + rowIdSegmentIndex: 1, + sourceBounds: "messages", summaryCount: 1, }, ]); diff --git a/packages/thread-view/test/merge-timeline-turn-page-rows.test.ts b/packages/thread-view/test/merge-timeline-turn-page-rows.test.ts new file mode 100644 index 0000000000..1d09b5708e --- /dev/null +++ b/packages/thread-view/test/merge-timeline-turn-page-rows.test.ts @@ -0,0 +1,169 @@ +import type { + TimelineCommandWorkRow, + TimelineDelegationWorkRow, + TimelineTurnRow, +} from "@bb/server-contract"; +import { describe, expect, it } from "vitest"; +import { + coalesceTimelineTurnDetailPageRows, + coalesceTimelineTurnPageRows, + mergeTimelineTurnPageRows, +} from "../src/merge-timeline-turn-page-rows.js"; + +function command(id: string, sequence: number): TimelineCommandWorkRow { + return { + id, + threadId: "thread-1", + turnId: "turn-1", + sourceSeqStart: sequence, + sourceSeqEnd: sequence, + startedAt: sequence, + createdAt: sequence, + kind: "work", + workKind: "command", + status: "completed", + callId: id, + command: id, + cwd: null, + source: null, + output: "", + exitCode: 0, + completedAt: sequence, + approvalStatus: null, + activityIntents: [], + }; +} + +function fragment(args: { + children?: TimelineCommandWorkRow[]; + end: number; + start: number; + summaryCount: number; +}): TimelineTurnRow { + return { + id: "thread-1:turn-1:turn", + threadId: "thread-1", + turnId: "turn-1", + sourceSeqStart: args.start, + sourceSeqEnd: args.end, + startedAt: 1_000, + createdAt: 9_000, + kind: "turn", + status: "completed", + summaryCount: args.summaryCount, + completedAt: 9_000, + children: args.children ?? null, + }; +} + +function delegation( + childRows: TimelineCommandWorkRow[], +): TimelineDelegationWorkRow { + return { + id: "delegation", + threadId: "thread-1", + turnId: "turn-1", + sourceSeqStart: 1, + sourceSeqEnd: 9, + startedAt: 1, + createdAt: 9, + kind: "work", + workKind: "delegation", + status: "completed", + callId: "delegation", + toolName: "Agent", + childRef: null, + background: false, + subagentType: null, + description: "Delegated work", + output: "Done", + completedAt: 9, + childRows, + }; +} + +describe("timeline turn page row merging", () => { + it("coalesces bounded byte slices into one logical completed turn", () => { + const older = fragment({ + children: [command("older-command", 2)], + start: 1, + end: 4, + summaryCount: 2, + }); + const newer = fragment({ + children: [command("newer-command", 7)], + start: 5, + end: 9, + summaryCount: 3, + }); + + const rows = coalesceTimelineTurnPageRows([older, newer]); + + expect(rows).toHaveLength(1); + expect(rows[0]).toMatchObject({ + id: "thread-1:turn-1:turn", + sourceSeqStart: 1, + sourceSeqEnd: 9, + startedAt: 1_000, + completedAt: 9_000, + summaryCount: 5, + children: [ + expect.objectContaining({ id: "older-command" }), + expect.objectContaining({ id: "newer-command" }), + ], + }); + }); + + it("replaces stale newest slices when a live byte window moves", () => { + const loaded = mergeTimelineTurnPageRows( + fragment({ start: 1, end: 4, summaryCount: 2 }), + fragment({ start: 5, end: 8, summaryCount: 3 }), + ); + const refreshedLatest = fragment({ + start: 5, + end: 10, + summaryCount: 3, + }); + + const merged = mergeTimelineTurnPageRows(loaded, refreshedLatest, { + newerWindowStartSequence: 5, + }); + + expect(merged.sourceSeqStart).toBe(1); + expect(merged.sourceSeqEnd).toBe(10); + expect(merged.summaryCount).toBe(5); + }); + + it("joins forward detail pages in source order with later closure rows authoritative", () => { + const pending = command("command", 2); + pending.status = "pending"; + pending.completedAt = null; + const completed = command("command", 2); + completed.sourceSeqEnd = 7; + + const rows = coalesceTimelineTurnDetailPageRows([ + [command("before", 1), pending], + [completed, command("after", 8)], + ]); + + expect(rows.map((row) => row.id)).toEqual(["before", "command", "after"]); + expect(rows[1]).toMatchObject({ status: "completed", sourceSeqEnd: 7 }); + }); + + it("joins the children of a delegation repeated as page context", () => { + const rows = coalesceTimelineTurnDetailPageRows([ + [delegation([command("older-command", 2)])], + [delegation([command("newer-command", 8)])], + ]); + + expect(rows).toHaveLength(1); + expect(rows[0]).toMatchObject({ + kind: "work", + workKind: "delegation", + childRows: [ + expect.objectContaining({ id: "older-command" }), + expect.objectContaining({ id: "newer-command" }), + ], + }); + }); +});