+
+ {showCatchUpIndicator ? : null}
+
{leadingContent}
{showLoadOlderRows ? (
) : null}
-
- {showCatchUpIndicator ? (
-
- {CATCH_UP_INDICATOR_LABEL}
-
- }
- className="mt-4 flex min-h-7 items-center"
- />
- ) : null}
-
{
+ const id = window.setTimeout(
+ () => setVisible(true),
+ CATCH_UP_INDICATOR_REVEAL_DELAY_MS,
+ );
+ return () => window.clearTimeout(id);
+ }, []);
+
+ if (!visible) return null;
+
+ return (
+
+ {CATCH_UP_INDICATOR_LABEL}
+
+ );
+}
+
function DelayedThreadLoadingIndicator() {
const [visible, setVisible] = useState(false);
useEffect(() => {
diff --git a/apps/app/src/components/thread/timeline/useThreadTimelineController.test.tsx b/apps/app/src/components/thread/timeline/useThreadTimelineController.test.tsx
index b1906a8eb61..f66e17cd30a 100644
--- a/apps/app/src/components/thread/timeline/useThreadTimelineController.test.tsx
+++ b/apps/app/src/components/thread/timeline/useThreadTimelineController.test.tsx
@@ -376,7 +376,7 @@ function startTimelineRefetch(queryClient: QueryClient): void {
}
function markTimelineHasNewEvents(queryClient: QueryClient): void {
- markThreadTimelineUnseenEvents(queryClient, "thread-1");
+ markThreadTimelineUnseenEvents(queryClient, "thread-1", 2);
void queryClient.invalidateQueries({
queryKey: TIMELINE_QUERY_KEY,
refetchType: "none",
@@ -1480,6 +1480,22 @@ describe("useThreadTimelineController commits", () => {
expect(view.latest().isCatchingUpTimeline).toBe(false);
});
+ it("keeps a timeline stale when a completed fetch does not include the known event", async () => {
+ vi.mocked(sdk.threads.timeline).mockResolvedValueOnce(
+ makeTimelineResponse({ maxSeq: 1, rows: [newestLoadedRow] }),
+ );
+ const { queryClient, wrapper } = createQueryClientTestHarness();
+ queryClient.setQueryData(
+ TIMELINE_QUERY_KEY,
+ makeTimelineResponse({ maxSeq: 1, rows: [newestLoadedRow] }),
+ );
+ markTimelineHasNewEvents(queryClient);
+ const view = renderProfiledController(wrapper);
+ await waitFor(() => expect(queryClient.isFetching()).toBe(0));
+ expect(view.latest().isCatchingUpTimeline).toBe(true);
+ expect(hasThreadTimelineUnseenEvents(queryClient, "thread-1")).toBe(true);
+ });
+
it("clears unseen events once the catch-up fetch succeeds", async () => {
vi.mocked(sdk.threads.timeline).mockResolvedValueOnce(
makeTimelineResponse({
diff --git a/apps/app/src/components/thread/timeline/useThreadTimelineController.ts b/apps/app/src/components/thread/timeline/useThreadTimelineController.ts
index 112de11f450..d3daaa39f84 100644
--- a/apps/app/src/components/thread/timeline/useThreadTimelineController.ts
+++ b/apps/app/src/components/thread/timeline/useThreadTimelineController.ts
@@ -1,5 +1,5 @@
import type { CommitOlderTimelineRows } from "./load-older-timeline-rows.js";
-import { useCallback, useRef, useState } from "react";
+import { useCallback, useState } from "react";
import {
useQueryClient,
type QueryObserverResult,
@@ -59,12 +59,6 @@ export interface UseThreadTimelineControllerResult {
timelineRows: TimelineRow[];
}
-interface MountFetchState {
- hasNewEvents: boolean;
- key: string | null;
- settled: boolean;
-}
-
interface LoadedTimelineTracker {
latestTimeline: ThreadTimelineResponse | undefined;
loaded: LoadedTimelineState;
@@ -119,25 +113,8 @@ export function useThreadTimelineController({
threadId,
}: UseThreadTimelineControllerArgs): UseThreadTimelineControllerResult {
const queryClient = useQueryClient();
- const mountFetchKey = enabled ? threadId : null;
- const [mountFetch, setMountFetch] = useState({
- hasNewEvents: false,
- key: null,
- settled: false,
- });
- const isNewMountFetch = mountFetch.key !== mountFetchKey;
- const mountFetchHasNewEvents = isNewMountFetch
- ? mountFetchKey !== null &&
- hasThreadTimelineUnseenEvents(queryClient, mountFetchKey)
- : mountFetch.hasNewEvents;
- const isTrackingMountFetch =
- mountFetchKey !== null &&
- mountFetchHasNewEvents &&
- (isNewMountFetch || !mountFetch.settled);
- const isTrackingMountFetchRef = useRef(isTrackingMountFetch);
- isTrackingMountFetchRef.current = isTrackingMountFetch;
const notifyOnChangeProps = useCallback((): TimelineQueryResultProp[] => {
- if (isTrackingMountFetchRef.current) {
+ if (hasThreadTimelineUnseenEvents(queryClient, threadId)) {
return TIMELINE_CONTROLLER_PROPS_WITHOUT_ROWS;
}
const cachedTimeline = queryClient.getQueryData(
@@ -294,21 +271,9 @@ export function useThreadTimelineController({
latestTimelineQuery.isLoading ||
(timelineQueryState.status === "loading" && timelineRows.length === 0) ||
(latestTimelineQuery.isFetching && timelineRows.length === 0);
- const isMountFetchIdle =
- mountFetchKey !== null && !latestTimelineQuery.isFetching;
- if (
- isNewMountFetch ||
- (mountFetch.hasNewEvents && isMountFetchIdle && !mountFetch.settled)
- ) {
- setMountFetch({
- hasNewEvents: mountFetchHasNewEvents,
- key: mountFetchKey,
- settled: isMountFetchIdle,
- });
- }
const isCatchingUpTimeline =
- isTrackingMountFetch &&
- latestTimelineQuery.isFetching &&
+ enabled &&
+ hasThreadTimelineUnseenEvents(queryClient, threadId) &&
timelineRows.length > 0;
const timelineError =
timelineLoading || timelineQueryState.status !== "unavailable"
diff --git a/apps/app/src/hooks/cache-owners/realtime-cache-registry.ts b/apps/app/src/hooks/cache-owners/realtime-cache-registry.ts
index 3efb0ecc2d4..5b0e505bd1e 100644
--- a/apps/app/src/hooks/cache-owners/realtime-cache-registry.ts
+++ b/apps/app/src/hooks/cache-owners/realtime-cache-registry.ts
@@ -573,6 +573,7 @@ interface RealtimeDirtyContext {
interface ThreadRealtimeDirtyContext extends RealtimeDirtyContext {
backgroundActivityChanged: boolean | undefined;
eventTypes: readonly ThreadEventType[] | undefined;
+ timelineSequence: number | undefined;
flushOnce: (key: string) => boolean;
hasPendingInteraction: boolean | undefined;
projectId: string | undefined;
@@ -847,6 +848,7 @@ function dirtyThreadSearchQueriesForCompletedTurn({
function dirtyThreadTimelineQueries({
eventTypes,
+ timelineSequence,
queryClient,
threadId,
}: ThreadRealtimeDirtyContext): void {
@@ -858,11 +860,13 @@ function dirtyThreadTimelineQueries({
});
const outlineMayHaveChanged =
eventTypes === undefined || eventTypes.includes("turn/completed");
+ if (threadId !== undefined && timelineSequence !== undefined) {
+ markThreadTimelineUnseenEvents(queryClient, threadId, timelineSequence);
+ }
if (
threadId !== undefined &&
!hasActiveQueries(queryClient, threadTimelineQueryKeyPrefix(threadId))
) {
- markThreadTimelineUnseenEvents(queryClient, threadId);
for (const queryKey of [...timelineQueryKeys, ...outlineQueryKeys]) {
queryClient.invalidateQueries({ queryKey, refetchType: "none" });
}
diff --git a/apps/app/src/hooks/cache-owners/thread-timeline-unseen-events.ts b/apps/app/src/hooks/cache-owners/thread-timeline-unseen-events.ts
index d8eae0ed91a..d0bef1b355d 100644
--- a/apps/app/src/hooks/cache-owners/thread-timeline-unseen-events.ts
+++ b/apps/app/src/hooks/cache-owners/thread-timeline-unseen-events.ts
@@ -1,35 +1,45 @@
import type { QueryClient } from "@tanstack/react-query";
+import type { ThreadTimelineResponse } from "@bb/server-contract";
+import { threadTimelineQueryKey } from "../queries/query-keys";
-const unseenTimelineEventThreadIdsByClient = new WeakMap<
+const unseenTimelineSequencesByClient = new WeakMap<
QueryClient,
- Set
+ Map
>();
export function markThreadTimelineUnseenEvents(
queryClient: QueryClient,
threadId: string,
+ sequence: number,
): void {
- let threadIds = unseenTimelineEventThreadIdsByClient.get(queryClient);
- if (!threadIds) {
- threadIds = new Set();
- unseenTimelineEventThreadIdsByClient.set(queryClient, threadIds);
+ let sequences = unseenTimelineSequencesByClient.get(queryClient);
+ if (!sequences) {
+ sequences = new Map();
+ unseenTimelineSequencesByClient.set(queryClient, sequences);
}
- threadIds.add(threadId);
+ sequences.set(threadId, Math.max(sequences.get(threadId) ?? 0, sequence));
}
export function clearThreadTimelineUnseenEvents(
queryClient: QueryClient,
threadId: string,
+ loadedSequence: number,
): void {
- unseenTimelineEventThreadIdsByClient.get(queryClient)?.delete(threadId);
+ const sequences = unseenTimelineSequencesByClient.get(queryClient);
+ if (loadedSequence >= (sequences?.get(threadId) ?? 0)) {
+ sequences?.delete(threadId);
+ }
}
export function hasThreadTimelineUnseenEvents(
queryClient: QueryClient,
threadId: string,
): boolean {
- return (
- unseenTimelineEventThreadIdsByClient.get(queryClient)?.has(threadId) ===
- true
+ const sequence = unseenTimelineSequencesByClient
+ .get(queryClient)
+ ?.get(threadId);
+ const timeline = queryClient.getQueryData(
+ threadTimelineQueryKey(threadId),
);
+ return sequence !== undefined && sequence > (timeline?.maxSeq ?? 0);
}
diff --git a/apps/app/src/hooks/queries/thread-queries.ts b/apps/app/src/hooks/queries/thread-queries.ts
index 61822aac20a..2d1fa8752eb 100644
--- a/apps/app/src/hooks/queries/thread-queries.ts
+++ b/apps/app/src/hooks/queries/thread-queries.ts
@@ -977,7 +977,7 @@ async function fetchThreadTimeline({
const timeline = await mergeThreadTimelineDelta(previous, response, () =>
sdk.threads.timeline({ threadId, signal, ...pageArgs }),
);
- clearThreadTimelineUnseenEvents(queryClient, threadId);
+ clearThreadTimelineUnseenEvents(queryClient, threadId, timeline.maxSeq);
return timeline;
}
diff --git a/apps/app/src/hooks/realtime-cache-effects.test.ts b/apps/app/src/hooks/realtime-cache-effects.test.ts
index d0a0a419f0b..84eb10e6f53 100644
--- a/apps/app/src/hooks/realtime-cache-effects.test.ts
+++ b/apps/app/src/hooks/realtime-cache-effects.test.ts
@@ -861,63 +861,80 @@ describe("createRealtimeCacheEffects", () => {
effects.dispose();
});
- it("marks the timeline of an unviewed thread stale without scheduling a refetch", async () => {
- vi.useFakeTimers();
- const { effects, queryClient } = createRealtimeEffectsTestContext();
- const viewedTimelineKey = threadTimelineQueryKey("thr_viewed");
- const unviewedTimelineKey = threadTimelineQueryKey("thr_unviewed");
- queryClient.setQueryData(unviewedTimelineKey, { rows: [] });
- const viewedTimelineQueryFn = vi.fn(async () => ({ rows: [] }));
- const viewedObserver = new QueryObserver(queryClient, {
- queryKey: viewedTimelineKey,
- queryFn: viewedTimelineQueryFn,
- staleTime: Infinity,
- });
- const unsubscribeViewed = viewedObserver.subscribe(() => {});
- await vi.advanceTimersByTimeAsync(0);
- viewedTimelineQueryFn.mockClear();
- const invalidateSpy = vi.spyOn(queryClient, "invalidateQueries");
-
- for (const threadId of ["thr_viewed", "thr_unviewed"]) {
- effects.handleChanged({
- type: "changed",
- entity: "thread",
- id: threadId,
- metadata: { eventTypes: ["item/completed"], projectId: "project-1" },
- changes: ["events-appended"],
+ it.each([
+ { cachedSequence: 1, isBehind: true },
+ { cachedSequence: 2, isBehind: false },
+ { cachedSequence: 3, isBehind: false },
+ ])(
+ "compares unviewed timeline sequence $cachedSequence with the notified sequence without refetching",
+ async ({ cachedSequence, isBehind }) => {
+ vi.useFakeTimers();
+ const { effects, queryClient } = createRealtimeEffectsTestContext();
+ const viewedTimelineKey = threadTimelineQueryKey("thr_viewed");
+ const unviewedTimelineKey = threadTimelineQueryKey("thr_unviewed");
+ queryClient.setQueryData(unviewedTimelineKey, {
+ rows: [],
+ maxSeq: cachedSequence,
});
- }
- await vi.advanceTimersByTimeAsync(50);
+ const viewedTimelineQueryFn = vi.fn(async () => ({
+ rows: [],
+ maxSeq: 2,
+ }));
+ const viewedObserver = new QueryObserver(queryClient, {
+ queryKey: viewedTimelineKey,
+ queryFn: viewedTimelineQueryFn,
+ staleTime: Infinity,
+ });
+ const unsubscribeViewed = viewedObserver.subscribe(() => {});
+ await vi.advanceTimersByTimeAsync(0);
+ viewedTimelineQueryFn.mockClear();
+ const invalidateSpy = vi.spyOn(queryClient, "invalidateQueries");
- expect(queryClient.getQueryState(unviewedTimelineKey)?.isInvalidated).toBe(
- true,
- );
- expect(hasThreadTimelineUnseenEvents(queryClient, "thr_unviewed")).toBe(
- true,
- );
- expect(hasThreadTimelineUnseenEvents(queryClient, "thr_viewed")).toBe(
- false,
- );
- const unviewedInvalidations = invalidateSpy.mock.calls.filter(
- ([filters]) =>
- JSON.stringify(filters?.queryKey) ===
- JSON.stringify(threadTimelineQueryKeyPrefix("thr_unviewed")),
- );
- expect(unviewedInvalidations).toHaveLength(1);
- expect(unviewedInvalidations[0]?.[0]?.refetchType).toBe("none");
- const viewedInvalidations = invalidateSpy.mock.calls.filter(
- ([filters]) =>
- JSON.stringify(filters?.queryKey) ===
- JSON.stringify(threadTimelineQueryKeyPrefix("thr_viewed")),
- );
- expect(viewedInvalidations).toHaveLength(1);
- expect(viewedInvalidations[0]?.[0]?.refetchType).toBeUndefined();
- expect(viewedTimelineQueryFn).toHaveBeenCalledTimes(1);
+ for (const threadId of ["thr_viewed", "thr_unviewed"]) {
+ effects.handleChanged({
+ type: "changed",
+ entity: "thread",
+ id: threadId,
+ metadata: {
+ eventTypes: ["item/completed"],
+ projectId: "project-1",
+ timelineSequence: 2,
+ },
+ changes: ["events-appended"],
+ });
+ }
+ await vi.advanceTimersByTimeAsync(50);
- invalidateSpy.mockRestore();
- unsubscribeViewed();
- effects.dispose();
- });
+ expect(
+ queryClient.getQueryState(unviewedTimelineKey)?.isInvalidated,
+ ).toBe(true);
+ expect(hasThreadTimelineUnseenEvents(queryClient, "thr_unviewed")).toBe(
+ isBehind,
+ );
+ expect(hasThreadTimelineUnseenEvents(queryClient, "thr_viewed")).toBe(
+ false,
+ );
+ const unviewedInvalidations = invalidateSpy.mock.calls.filter(
+ ([filters]) =>
+ JSON.stringify(filters?.queryKey) ===
+ JSON.stringify(threadTimelineQueryKeyPrefix("thr_unviewed")),
+ );
+ expect(unviewedInvalidations).toHaveLength(1);
+ expect(unviewedInvalidations[0]?.[0]?.refetchType).toBe("none");
+ const viewedInvalidations = invalidateSpy.mock.calls.filter(
+ ([filters]) =>
+ JSON.stringify(filters?.queryKey) ===
+ JSON.stringify(threadTimelineQueryKeyPrefix("thr_viewed")),
+ );
+ expect(viewedInvalidations).toHaveLength(1);
+ expect(viewedInvalidations[0]?.[0]?.refetchType).toBeUndefined();
+ expect(viewedTimelineQueryFn).toHaveBeenCalledTimes(1);
+
+ invalidateSpy.mockRestore();
+ unsubscribeViewed();
+ effects.dispose();
+ },
+ );
it("refreshes the full conversation outline once at turn completion, not for streaming deltas", async () => {
vi.useFakeTimers();
@@ -1854,7 +1871,11 @@ describe("createRealtimeCacheEffects", () => {
type: "changed",
entity: "thread",
id: "thr_1",
- metadata: { eventTypes: ["item/completed"], projectId: "project-1" },
+ metadata: {
+ eventTypes: ["item/completed"],
+ projectId: "project-1",
+ timelineSequence: 2,
+ },
changes: ["events-appended"],
});
await vi.advanceTimersByTimeAsync(50);
diff --git a/apps/app/src/hooks/realtime-cache-effects.ts b/apps/app/src/hooks/realtime-cache-effects.ts
index e657b730d84..774cc107cfd 100644
--- a/apps/app/src/hooks/realtime-cache-effects.ts
+++ b/apps/app/src/hooks/realtime-cache-effects.ts
@@ -146,6 +146,12 @@ function mergeThreadChangeMetadata({
? next.statusChange
: (next.statusChange ?? current?.statusChange);
const metadata: ThreadChangeMetadata = {};
+ const sequences = [current?.timelineSequence, next.timelineSequence].filter(
+ (sequence): sequence is number => sequence !== undefined,
+ );
+ if (sequences.length > 0) {
+ metadata.timelineSequence = Math.max(...sequences);
+ }
if (eventTypes) {
metadata.eventTypes = eventTypes;
}
@@ -203,6 +209,7 @@ function flushThreadInvalidations(
context: {
backgroundActivityChanged: undefined,
eventTypes: undefined,
+ timelineSequence: undefined,
flushOnce,
hasPendingInteraction: undefined,
projectId: undefined,
@@ -221,6 +228,7 @@ function flushThreadInvalidations(
context: {
backgroundActivityChanged: metadata?.backgroundActivityChanged,
eventTypes: metadata?.eventTypes,
+ timelineSequence: metadata?.timelineSequence,
flushOnce,
hasPendingInteraction: metadata?.hasPendingInteraction,
projectId: metadata?.projectId,
@@ -262,6 +270,7 @@ function applyImmediateThreadChanges({
context: {
backgroundActivityChanged: merged?.backgroundActivityChanged,
eventTypes: merged?.eventTypes,
+ timelineSequence: merged?.timelineSequence,
flushOnce,
hasPendingInteraction: merged?.hasPendingInteraction,
projectId: merged?.projectId,
diff --git a/apps/server/src/internal/events.ts b/apps/server/src/internal/events.ts
index 1371b890d1b..2b73030b471 100644
--- a/apps/server/src/internal/events.ts
+++ b/apps/server/src/internal/events.ts
@@ -4,6 +4,7 @@ import {
appendDaemonEventsInTransaction,
deriveStoredEventItemFields,
getThread,
+ getLatestThreadSequence,
listStoredTurnCompletedKeys,
listThreadEnvironmentAssignmentsOnHost,
MissingStoredTurnStartedError,
@@ -94,6 +95,7 @@ interface ResolveEventsToApplyArgs {
}
interface NotifyInsertedEventThreadsDeps {
+ db: AppDeps["db"];
hub: AppDeps["hub"];
}
@@ -353,6 +355,7 @@ function notifyInsertedEventThreads(
? { backgroundActivityChanged: true }
: {}),
eventTypes: Array.from(eventTypes),
+ timelineSequence: getLatestThreadSequence(deps.db, { threadId }),
});
}
}
diff --git a/apps/server/src/services/environments/environment-engine.ts b/apps/server/src/services/environments/environment-engine.ts
index 260c6c41f7c..f156380fc69 100644
--- a/apps/server/src/services/environments/environment-engine.ts
+++ b/apps/server/src/services/environments/environment-engine.ts
@@ -1,6 +1,6 @@
import { withHostCleanup } from "../hosts/cleanup-context.js";
import { findHostDataDir } from "../lib/entity-lookup.js";
-import { updateThread } from "@bb/db";
+import { getLatestThreadSequence, updateThread } from "@bb/db";
import {
assertEnvironmentPathAvailable,
findBlockingEnvironmentPathClaim,
@@ -253,6 +253,9 @@ function provisioningReporter(
.where(eq(environments.id, row.id))
.run();
deps.hub.notifyThread(owner, ["events-appended"], {
+ timelineSequence: getLatestThreadSequence(deps.db, {
+ threadId: owner,
+ }),
eventTypes: ["system/thread-provisioning"],
});
});
@@ -562,6 +565,9 @@ async function runCreate(
],
});
deps.hub.notifyThread(context.thread.id, ["events-appended"], {
+ timelineSequence: getLatestThreadSequence(deps.db, {
+ threadId: context.thread.id,
+ }),
eventTypes: ["system/thread-provisioning"],
});
}
@@ -1145,6 +1151,9 @@ function appendThreadProvisioningEventToEnvironmentThreadsInTransaction(
threadId: thread.id,
});
deps.hub.notifyThread(thread.id, ["events-appended"], {
+ timelineSequence: getLatestThreadSequence(deps.db, {
+ threadId: thread.id,
+ }),
eventTypes: ["system/thread-provisioning"],
});
}
@@ -1366,6 +1375,9 @@ function settleEnvironmentProvisionOutcome(
entries,
});
args.deps.hub.notifyThread(thread.id, ["events-appended"], {
+ timelineSequence: getLatestThreadSequence(args.deps.db, {
+ threadId: thread.id,
+ }),
eventTypes: ["system/thread-provisioning"],
});
continue;
diff --git a/apps/server/src/services/interactions/pending-interaction-timeline.ts b/apps/server/src/services/interactions/pending-interaction-timeline.ts
index 19f690e155e..f8fae806b15 100644
--- a/apps/server/src/services/interactions/pending-interaction-timeline.ts
+++ b/apps/server/src/services/interactions/pending-interaction-timeline.ts
@@ -14,6 +14,7 @@ import {
threadScope,
} from "@bb/domain";
import {
+ getLatestThreadSequence,
getThread,
type AppendStoredThreadEventArgs,
type DbNotifier,
@@ -186,6 +187,9 @@ export function appendPendingInteractionTimelineEventInTransaction(
appendThreadEventInTransaction(deps.db, write);
}
deps.hub.notifyThread(interaction.threadId, ["events-appended"], {
+ timelineSequence: getLatestThreadSequence(deps.db, {
+ threadId: interaction.threadId,
+ }),
eventTypes: writes.map((write) => write.type),
});
}
diff --git a/apps/server/src/services/threads/background-task-reconciliation.ts b/apps/server/src/services/threads/background-task-reconciliation.ts
index 40b5920d3f8..f26397cc662 100644
--- a/apps/server/src/services/threads/background-task-reconciliation.ts
+++ b/apps/server/src/services/threads/background-task-reconciliation.ts
@@ -1,5 +1,6 @@
import { z } from "zod";
import {
+ getLatestThreadSequence,
listOpenBackgroundTaskItemRowsForHost,
listOpenBackgroundTaskItemRowsForThread,
type DbNotifier,
@@ -70,6 +71,9 @@ export function settleDanglingBackgroundTasks(
for (const threadId of settledThreadIds) {
deps.hub.notifyThread(threadId, ["events-appended"], {
+ timelineSequence: getLatestThreadSequence(deps.db, {
+ threadId: threadId,
+ }),
eventTypes: ["item/backgroundTask/completed"],
});
}
@@ -83,6 +87,9 @@ export function settleDanglingBackgroundTasksForStoppedThreadInTransaction(
const settledThreadIds = appendDanglingBackgroundTaskCompletions(deps, rows);
for (const threadId of settledThreadIds) {
deps.hub.notifyThread(threadId, ["events-appended"], {
+ timelineSequence: getLatestThreadSequence(deps.db, {
+ threadId: threadId,
+ }),
eventTypes: ["item/backgroundTask/completed"],
});
}
diff --git a/apps/server/src/services/threads/parent-system-messages.ts b/apps/server/src/services/threads/parent-system-messages.ts
index 2f362d57256..d44f295eb37 100644
--- a/apps/server/src/services/threads/parent-system-messages.ts
+++ b/apps/server/src/services/threads/parent-system-messages.ts
@@ -1,4 +1,5 @@
import {
+ getLatestThreadSequence,
getEnvironment,
getHost,
getThread,
@@ -331,6 +332,9 @@ async function queueActiveParentSystemMessage(
}
deps.hub.notifyThread(args.thread.id, ["events-appended"], {
+ timelineSequence: getLatestThreadSequence(deps.db, {
+ threadId: args.thread.id,
+ }),
eventTypes: ["client/turn/requested"],
});
startLiveHostCommand(deps, {
@@ -399,6 +403,9 @@ async function queueReadyParentSystemMessage(
{ behavior: "immediate" },
);
deps.hub.notifyThread(args.thread.id, ["events-appended"], {
+ timelineSequence: getLatestThreadSequence(deps.db, {
+ threadId: args.thread.id,
+ }),
eventTypes: ["client/turn/requested"],
});
startLiveHostCommand(deps, {
diff --git a/apps/server/src/services/threads/queued-messages.ts b/apps/server/src/services/threads/queued-messages.ts
index b65b2127d6b..43b2f6cd4c6 100644
--- a/apps/server/src/services/threads/queued-messages.ts
+++ b/apps/server/src/services/threads/queued-messages.ts
@@ -1,4 +1,5 @@
import {
+ getLatestThreadSequence,
claimNextQueuedThreadMessageGroup,
claimQueuedThreadMessageGroup,
createQueuedThreadMessageInTransaction,
@@ -596,6 +597,9 @@ async function sendClaimedQueuedMessageForIdleProviderThread(
thread.id,
["events-appended", "queue-changed", "status-changed"],
{
+ timelineSequence: getLatestThreadSequence(deps.db, {
+ threadId: thread.id,
+ }),
eventTypes: ["client/turn/requested"],
...buildThreadStatusChangeMetadata(deps, activeThread),
},
diff --git a/apps/server/src/services/threads/thread-environment-directory.ts b/apps/server/src/services/threads/thread-environment-directory.ts
index bb547b864da..f33c3b44610 100644
--- a/apps/server/src/services/threads/thread-environment-directory.ts
+++ b/apps/server/src/services/threads/thread-environment-directory.ts
@@ -1,6 +1,7 @@
import { assertEnvironmentPathAvailable } from "../environments/path-admission.js";
import { z } from "zod";
import {
+ getLatestThreadSequence,
createEnvironment,
type EnvironmentRow,
createEventId,
@@ -204,6 +205,9 @@ function attachReadyEnvironment(
if (result.kind === "attached" && result.changed) {
deps.hub.notifyThread(args.thread.id, ["events-appended"], {
+ timelineSequence: getLatestThreadSequence(deps.db, {
+ threadId: args.thread.id,
+ }),
eventTypes: ["system/operation"],
});
}
diff --git a/apps/server/src/services/threads/thread-environment-placement.ts b/apps/server/src/services/threads/thread-environment-placement.ts
index 7169a5e9fd5..02f0c31057c 100644
--- a/apps/server/src/services/threads/thread-environment-placement.ts
+++ b/apps/server/src/services/threads/thread-environment-placement.ts
@@ -1,4 +1,5 @@
import {
+ getLatestThreadSequence,
findProjectEnvironmentByHostPath,
getAppSettings,
getEnvironment,
@@ -833,6 +834,7 @@ function reportMachineProgress(
entries,
});
deps.hub.notifyThread(threadId, ["events-appended"], {
+ timelineSequence: getLatestThreadSequence(deps.db, { threadId: threadId }),
eventTypes: ["system/thread-provisioning"],
});
}
@@ -1229,6 +1231,9 @@ export function prepareProviderEnvironment(
],
});
deps.hub.notifyThread(context.thread.id, ["events-appended"], {
+ timelineSequence: getLatestThreadSequence(deps.db, {
+ threadId: context.thread.id,
+ }),
eventTypes: ["system/thread-provisioning"],
});
}
diff --git a/apps/server/src/services/threads/thread-events.ts b/apps/server/src/services/threads/thread-events.ts
index 87a56760da4..0e3e9efe456 100644
--- a/apps/server/src/services/threads/thread-events.ts
+++ b/apps/server/src/services/threads/thread-events.ts
@@ -241,6 +241,7 @@ interface AppendThreadEventsTransactionResult {
}
interface BuildAppendThreadEventNotificationMetadataArgs {
+ sequence: number;
readStateUpdate: ThreadReadStateUpdate | null;
eventType: ThreadEventType;
}
@@ -368,11 +369,13 @@ function buildAppendThreadEventNotificationChanges(
}
function buildAppendThreadEventNotificationMetadata({
+ sequence,
readStateUpdate,
eventType,
}: BuildAppendThreadEventNotificationMetadataArgs): ThreadChangeMetadata {
const metadata: ThreadChangeMetadata = {
eventTypes: [eventType],
+ timelineSequence: sequence,
};
if (readStateUpdate?.changed === true) {
metadata.projectId = readStateUpdate.projectId;
@@ -473,6 +476,7 @@ export function appendThreadEvent(
args.threadId,
buildAppendThreadEventNotificationChanges(result.readStateUpdate),
buildAppendThreadEventNotificationMetadata({
+ sequence: result.sequence,
readStateUpdate: result.readStateUpdate,
eventType: args.type,
}),
@@ -545,6 +549,7 @@ export function appendPreparedClientTurnRequestedEventWithNotificationInTransact
result.readStateUpdate,
),
notificationMetadata: buildAppendThreadEventNotificationMetadata({
+ sequence: result.sequence,
readStateUpdate: result.readStateUpdate,
eventType: args.type,
}),
@@ -693,6 +698,7 @@ export function appendSystemErrorEventInTransaction(
buildSystemErrorEventArgs(args),
);
deps.hub.notifyThread(args.threadId, ["events-appended"], {
+ timelineSequence: sequence,
eventTypes: ["system/error"],
});
return sequence;
@@ -825,7 +831,9 @@ export function appendThreadOwnershipChangeEventInTransaction(
}
const sequence = appendThreadEventInTransaction(deps.db, eventArgs);
- deps.hub.notifyThread(args.threadId, ["events-appended"]);
+ deps.hub.notifyThread(args.threadId, ["events-appended"], {
+ timelineSequence: sequence,
+ });
return sequence;
}
diff --git a/apps/server/src/services/threads/thread-fork-history.ts b/apps/server/src/services/threads/thread-fork-history.ts
index 8e3742a8996..a359f1ac5a7 100644
--- a/apps/server/src/services/threads/thread-fork-history.ts
+++ b/apps/server/src/services/threads/thread-fork-history.ts
@@ -1,4 +1,5 @@
import {
+ getLatestThreadSequence,
copyStoredThreadEventsInTransaction,
findLastCompletedRootStoredTurn,
findLastRootStoredTurnStarted,
@@ -314,6 +315,9 @@ export function copyForkSourceHistory(
{ behavior: "immediate" },
);
deps.hub.notifyThread(args.fork.id, ["events-appended"], {
+ timelineSequence: getLatestThreadSequence(deps.db, {
+ threadId: args.fork.id,
+ }),
eventTypes: [...new Set(rows.map((row) => row.type))],
});
}
diff --git a/apps/server/src/services/threads/thread-lifecycle.ts b/apps/server/src/services/threads/thread-lifecycle.ts
index d7e6412d810..1049ae152c7 100644
--- a/apps/server/src/services/threads/thread-lifecycle.ts
+++ b/apps/server/src/services/threads/thread-lifecycle.ts
@@ -16,6 +16,7 @@ import {
sql,
} from "drizzle-orm";
import {
+ getLatestThreadSequence,
deleteThread,
environments,
events,
@@ -547,6 +548,7 @@ function appendProvisioningInterruptedEventInTransaction(
entries: [buildProvisioningStoppedEntry()],
});
deps.hub.notifyThread(thread.id, ["events-appended"], {
+ timelineSequence: getLatestThreadSequence(deps.db, { threadId: thread.id }),
eventTypes: ["system/thread-provisioning"],
});
}
@@ -563,6 +565,9 @@ function appendThreadInterruptedEventIfMissingInTransaction(
reason: args.reason,
});
deps.hub.notifyThread(args.threadId, ["events-appended"], {
+ timelineSequence: getLatestThreadSequence(deps.db, {
+ threadId: args.threadId,
+ }),
eventTypes: ["system/thread/interrupted"],
});
return true;
@@ -585,6 +590,9 @@ function markThreadStoppingWithEventInTransaction(
reason: args.reason,
});
deps.hub.notifyThread(args.threadId, ["events-appended"], {
+ timelineSequence: getLatestThreadSequence(deps.db, {
+ threadId: args.threadId,
+ }),
eventTypes: ["system/thread/interrupted"],
});
return true;
@@ -1267,6 +1275,7 @@ function dispatchThreadStartFromRequest(
if (result.completedProvisionSequence !== null) {
deps.hub.notifyThread(args.threadId, ["events-appended"], {
+ timelineSequence: result.completedProvisionSequence,
eventTypes: ["system/thread-provisioning"],
});
}
@@ -1750,6 +1759,9 @@ function interruptActiveTurnForThreadInTransaction(
eventTypes.push("system/thread/interrupted");
}
deps.hub.notifyThread(args.threadId, ["events-appended", "status-changed"], {
+ timelineSequence: getLatestThreadSequence(deps.db, {
+ threadId: args.threadId,
+ }),
eventTypes,
});
@@ -1874,6 +1886,9 @@ function interruptActiveThreads(
result.threadId,
["events-appended", "status-changed"],
{
+ timelineSequence: getLatestThreadSequence(deps.db, {
+ threadId: result.threadId,
+ }),
eventTypes,
...(thread ? buildThreadStatusChangeMetadata(deps, thread) : {}),
},
diff --git a/apps/server/src/services/threads/thread-provisioning-environment.ts b/apps/server/src/services/threads/thread-provisioning-environment.ts
index b38345416b7..e5a205bcf70 100644
--- a/apps/server/src/services/threads/thread-provisioning-environment.ts
+++ b/apps/server/src/services/threads/thread-provisioning-environment.ts
@@ -1,4 +1,5 @@
import {
+ getLatestThreadSequence,
getEnvironment,
getThread,
type EnvironmentRow,
@@ -101,6 +102,9 @@ export function ensureWorkspaceReadyEventInTransaction(
});
if (appendedSequence !== null)
deps.hub.notifyThread(thread.id, ["events-appended"], {
+ timelineSequence: getLatestThreadSequence(deps.db, {
+ threadId: thread.id,
+ }),
eventTypes: ["system/thread-provisioning"],
});
return true;
diff --git a/apps/server/src/services/threads/thread-provisioning.ts b/apps/server/src/services/threads/thread-provisioning.ts
index c2c40113021..9e891e3cbef 100644
--- a/apps/server/src/services/threads/thread-provisioning.ts
+++ b/apps/server/src/services/threads/thread-provisioning.ts
@@ -1,4 +1,7 @@
-import { getNonDestroyedHostByLaunchKey } from "@bb/db";
+import {
+ getLatestThreadSequence,
+ getNonDestroyedHostByLaunchKey,
+} from "@bb/db";
import { sweepProviderMachine } from "../machines/provider-orchestration.js";
import { cancelProviderEnvironmentCreation } from "../environments/environment-engine.js";
import { getPreparingEnvironment } from "@bb/db";
@@ -182,6 +185,9 @@ function settleSeededThreadProvisioning(
);
if (settled.completedProvisioning) {
deps.hub.notifyThread(args.threadId, ["events-appended"], {
+ timelineSequence: getLatestThreadSequence(deps.db, {
+ threadId: args.threadId,
+ }),
eventTypes: ["system/thread-provisioning"],
});
}
diff --git a/apps/server/src/ws/hub.ts b/apps/server/src/ws/hub.ts
index c62844a3725..e7ea6e8cc0a 100644
--- a/apps/server/src/ws/hub.ts
+++ b/apps/server/src/ws/hub.ts
@@ -61,6 +61,7 @@ export type ServerChangedMessage =
type ChangedMessageListener = (message: ServerChangedMessage) => void;
interface PendingThreadListEventsAppended {
+ timelineSequence: number | undefined;
eventTypes: Set;
merged: boolean;
timeout: ReturnType;
@@ -1065,6 +1066,12 @@ export class NotificationHub implements DbNotifier {
for (const eventType of eventTypes) {
pending.eventTypes.add(eventType);
}
+ if (message.metadata?.timelineSequence !== undefined) {
+ pending.timelineSequence = Math.max(
+ pending.timelineSequence ?? 0,
+ message.metadata.timelineSequence,
+ );
+ }
pending.merged = true;
return;
}
@@ -1076,6 +1083,7 @@ export class NotificationHub implements DbNotifier {
timeout.unref?.();
this.pendingThreadListEventsAppendedByThread.set(threadId, {
eventTypes: new Set(eventTypes),
+ timelineSequence: message.metadata?.timelineSequence,
merged: false,
timeout,
});
@@ -1094,8 +1102,17 @@ export class NotificationHub implements DbNotifier {
type: "changed",
entity: "thread",
id: threadId,
- ...(pending.eventTypes.size > 0
- ? { metadata: { eventTypes: [...pending.eventTypes] } }
+ ...(pending.eventTypes.size > 0 || pending.timelineSequence !== undefined
+ ? {
+ metadata: {
+ ...(pending.eventTypes.size > 0
+ ? { eventTypes: [...pending.eventTypes] }
+ : {}),
+ ...(pending.timelineSequence !== undefined
+ ? { timelineSequence: pending.timelineSequence }
+ : {}),
+ },
+ }
: {}),
changes: ["events-appended"],
};
diff --git a/apps/server/test/app/hub.test.ts b/apps/server/test/app/hub.test.ts
index f4c94b1db42..753ceb19d8c 100644
--- a/apps/server/test/app/hub.test.ts
+++ b/apps/server/test/app/hub.test.ts
@@ -495,6 +495,7 @@ describe("NotificationHub events-appended thread-list coalescing", () => {
threadId: "thread-1",
});
+ let timelineSequence = 0;
for (const eventType of [
"item/agentMessage/delta",
"item/agentMessage/delta",
@@ -502,6 +503,7 @@ describe("NotificationHub events-appended thread-list coalescing", () => {
] as const) {
hub.notifyThread("thread-1", ["events-appended"], {
eventTypes: [eventType],
+ timelineSequence: ++timelineSequence,
});
}
@@ -511,7 +513,10 @@ describe("NotificationHub events-appended thread-list coalescing", () => {
expect(messagesOf(listSocket)[0]).toMatchObject({
id: "thread-1",
changes: ["events-appended"],
- metadata: { eventTypes: ["item/agentMessage/delta"] },
+ metadata: {
+ eventTypes: ["item/agentMessage/delta"],
+ timelineSequence: 1,
+ },
});
vi.advanceTimersByTime(999);
@@ -521,7 +526,10 @@ describe("NotificationHub events-appended thread-list coalescing", () => {
expect(messagesOf(listSocket)[1]).toMatchObject({
id: "thread-1",
changes: ["events-appended"],
- metadata: { eventTypes: ["item/agentMessage/delta", "item/started"] },
+ metadata: {
+ eventTypes: ["item/agentMessage/delta", "item/started"],
+ timelineSequence: 3,
+ },
});
expect(detailSocket.messages).toHaveLength(3);
expect(listAndDetailSocket.messages).toHaveLength(3);
diff --git a/apps/server/test/services/threads/thread-events.test.ts b/apps/server/test/services/threads/thread-events.test.ts
index dbeb577fc0c..d16bef61dd2 100644
--- a/apps/server/test/services/threads/thread-events.test.ts
+++ b/apps/server/test/services/threads/thread-events.test.ts
@@ -345,6 +345,7 @@ describe("thread event appends", () => {
["events-appended", "read-state-changed"],
{
eventTypes: ["client/turn/requested"],
+ timelineSequence: 1,
projectId: thread.projectId,
},
);
@@ -392,7 +393,7 @@ describe("thread event appends", () => {
expect(notifyThreadSpy).toHaveBeenCalledWith(
thread.id,
["events-appended"],
- { eventTypes: ["client/turn/requested"] },
+ { eventTypes: ["client/turn/requested"], timelineSequence: 1 },
);
} finally {
await harness.cleanup();
diff --git a/docs/timeline-pagination.md b/docs/timeline-pagination.md
index 1ef0cbe71f1..9f335af5dff 100644
--- a/docs/timeline-pagination.md
+++ b/docs/timeline-pagination.md
@@ -179,3 +179,20 @@ indexed, bounded lookup. Crossing it rebuilds the prefix so empty completed
messages keep the same fallback previews even when child payloads are omitted.
The checkpoint cache retains at most 16 threads and 8 million characters of
serialized previews and identity data; eviction only affects performance.
+
+## Catch-up feedback
+
+Event-append notifications include `metadata.timelineSequence`, the thread's
+stored event sequence after the write. Both server-side notification coalescing
+and client-side debouncing preserve the highest sequence. The client compares
+this with the cached timeline response's `maxSeq`; only a known newer sequence
+shows catch-up feedback. A refresh caused by cache age, or a delayed notification
+already covered by the cached response, does not show it. Notifications without
+a sequence still invalidate the cache but do not claim that messages are missing.
+
+A successful response acknowledges only sequences through its `maxSeq`, so a
+response that predates another known event cannot clear that event. The catch-up
+indicator appears only after the timeline has remained behind for one second,
+disappears immediately when the cache catches up, and floats over the
+timeline without adding or removing scroll height. Initial loads without cached
+rows continue to use the loading skeleton.
diff --git a/packages/db/src/data/events.ts b/packages/db/src/data/events.ts
index 8ab07d82744..5353df0887a 100644
--- a/packages/db/src/data/events.ts
+++ b/packages/db/src/data/events.ts
@@ -491,6 +491,7 @@ export function insertEvents(
for (const [threadId, eventTypes] of eventTypesByThreadId) {
notifier.notifyThread(threadId, ["events-appended"], {
eventTypes: Array.from(eventTypes),
+ timelineSequence: getLatestThreadSequence(db, { threadId }),
});
}
@@ -1024,6 +1025,7 @@ export function appendStoredThreadEvent(
);
notifier.notifyThread(args.threadId, ["events-appended"], {
eventTypes: [args.type],
+ timelineSequence: sequence,
});
return sequence;
}
@@ -3454,7 +3456,7 @@ const isNotDiagnosticEvent = sql`(
)`;
export function getLatestThreadSequence(
- db: DbConnection,
+ db: DbQueryConnection,
args: GetLatestThreadSequenceArgs,
): number {
return getHighWaterMarks(db, [args.threadId])[args.threadId] ?? 0;
diff --git a/packages/db/test/data/events.test.ts b/packages/db/test/data/events.test.ts
index d068eae62f9..c269300be96 100644
--- a/packages/db/test/data/events.test.ts
+++ b/packages/db/test/data/events.test.ts
@@ -4500,6 +4500,7 @@ describe("events", () => {
["events-appended"],
{
eventTypes: ["system/error", "client/turn/requested"],
+ timelineSequence: 2,
},
);
expect(spy.notifyThread).toHaveBeenCalledWith(
@@ -4507,6 +4508,7 @@ describe("events", () => {
["events-appended"],
{
eventTypes: ["system/error"],
+ timelineSequence: 1,
},
);
expect(spy.notifyThread).toHaveBeenCalledTimes(2);
diff --git a/packages/domain/src/change-kinds.ts b/packages/domain/src/change-kinds.ts
index 45cec8b2ceb..77ee83164a0 100644
--- a/packages/domain/src/change-kinds.ts
+++ b/packages/domain/src/change-kinds.ts
@@ -210,6 +210,7 @@ export const threadChangeMetadataSchema = z
.object({
backgroundActivityChanged: z.boolean().optional(),
eventTypes: z.array(threadEventTypeSchema).readonly().optional(),
+ timelineSequence: z.number().int().nonnegative().optional(),
hasPendingInteraction: z.boolean().optional(),
projectId: z.string().optional(),
statusChange: threadStatusChangeMetadataSchema.optional(),
@@ -290,6 +291,7 @@ const knownThreadEventTypes: ReadonlySet = new Set(
);
const threadChangeMetadataLenientSchema = z.object({
+ timelineSequence: z.number().int().nonnegative().optional(),
backgroundActivityChanged: z.boolean().optional(),
eventTypes: z
.array(z.string())
diff --git a/packages/domain/test/change-kinds.test.ts b/packages/domain/test/change-kinds.test.ts
index 3a6955d815b..398a5025ae3 100644
--- a/packages/domain/test/change-kinds.test.ts
+++ b/packages/domain/test/change-kinds.test.ts
@@ -36,6 +36,7 @@ function lenientOptionsByEntity(): Map {
const maximalThreadMetadata: ThreadChangeMetadata = {
backgroundActivityChanged: true,
eventTypes: [...threadEventTypeValues],
+ timelineSequence: 42,
hasPendingInteraction: true,
projectId: "proj_1",
statusChange: {