BuzzbeeSCD/openchamber
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

fix(sync): commit first message page before expansion loop (#2084) (#2086)

Browse Source 
* fix(sync): commit first message page before expansion loop (#2084)

loadMessages committed the store only after the full expansion sequence
(50→100→150), so the hydrating skeleton stayed for 3 sequential HTTP
round-trips when a session tail had no user message boundary.

Move the store write (materialize + setState) to happen after the first
fetch. The expansion loop now commits each expanded page incrementally
instead of overwriting a page variable and committing once at the end.

The first commit is gated on hasUserMessage(page.session) || page.complete:
if the tail is assistant-only, deferring to the expansion loop keeps the
skeleton (loading state) instead of rendering an empty chat that looks
like a fresh session. Sessions with a user boundary in the first 50
messages get content after a single round-trip.

* fix(sync): address review nits for #2084

- deferred init uses page.session instead of [] so limit reflects the
  real fetched count if the expansion loop is ever a no-op
- both stale branches in commitMessagesToStore return messages: [] for
  consistency
- add isStale guard between expansion fetch and commit for defense-in-depth

* fix(sync): always commit prepend-mode pages to store (#2084)

The deferred init path (assistant-only tail) skipped commitMessagesToStore
entirely when options.before was set — prepend mode. The fetched older
messages were never written to the store, silently dropping them.

Gate the deferral on !options.before: prepend mode always commits because
messages are already rendered (no skeleton to protect) and skipping the
store write would lose the fetched page.

---------

Co-authored-by: bashrusakh <bashrusakh@users.noreply.github.com>
Co-authored-by: Bohdan Triapitsyn <artmore@protonmail.com>
This commit is contained in:
Leonid
2026-07-11 14:39:49 +03:00
committed by GitHub
co-authored by bashrusakh Bohdan Triapitsyn
parent 37768958f3
commit d8a904954b
2 changed files with 328 additions and 47 deletions
Download Patch File Download Diff File Expand all files Collapse all files
packages/ui/src/sync/use-sync.ts
+112 -46
View File
@@ -127,7 +127,7 @@ function isUserMessage(message: Message): boolean {
return role === "user"
}
function hasUserMessage(messages: Message[] | undefined): boolean {
export function hasUserMessage(messages: Message[] | undefined): boolean {
return Boolean(messages?.some(isUserMessage))
}
@@ -349,75 +349,141 @@ export function useSync() {
setMetaFor(sessionID, { loading: true })
try {
// A resync (no `before`) must fetch at least as many messages as we
// already have on screen. Live events append to the store WITHOUT growing
// m.limit, so reusing the stale m.limit here would under-fetch and make
// the server hand back a spurious "older" cursor — surfacing a phantom
// "load older" button for a session whose full history is already shown
// (e.g. after a reconnect resync following a few new messages).
// Commit a fetched page to the store: merge optimistic items, run
// materialization, and write the result so the UI can render it.
// Returns the committed meta so the caller can update pagination
// state once at the end. The store write happens here (per page) so
// the hydrating skeleton disappears after the first fetch instead of
// waiting for the full expansion sequence.
const commitMessagesToStore = (
page: Awaited<ReturnType<typeof fetchMessages>>,
mode: "replace" | "prepend" | undefined,
isStale?: () => boolean,
) => {
const items = getOptimistic(sessionID)
const merged = mergeOptimisticPage(page, items)
for (const messageID of merged.confirmed) {
clearOptimistic(sessionID, messageID)
}
if (isStale?.()) {
return { messages: [], cursor: merged.cursor, complete: merged.complete }
}
const current = store.getState()
const materialized = materializeSessionSnapshots(
current,
sessionID,
merged.session.map((info) => ({
info,
parts: merged.part.find((item) => item.id === info.id)?.part ?? [],
})),
{ skipPartTypes: SKIP_PARTS, mode: mode === "prepend" ? "prepend" : "merge" },
)
// materializeSessionSnapshots is synchronous today, so this check
// is defense-in-depth: it guards the store write if materialization
// ever becomes async or yields between the check above and setState.
if (isStale?.()) {
return { messages: [], cursor: merged.cursor, complete: merged.complete }
}
if (materialized.messagesChanged || materialized.partsChanged) {
store.setState({
...(materialized.messagesChanged ? { message: materialized.message } : {}),
...(materialized.partsChanged ? { part: materialized.part } : {}),
})
}
return { messages: materialized.messages, cursor: merged.cursor, complete: merged.complete }
}
// Live events can append messages without growing m.limit. A resync
// must cover everything already rendered or it can manufacture an
// "older" cursor for history that is already on screen.
const storeMessageCount = store.getState().message[sessionID]?.length ?? 0
const limit = options?.before ? HISTORY_MESSAGE_PAGE_SIZE : Math.max(m.limit, storeMessageCount)
let page = await fetchMessages(sessionID, limit, options?.before)
const limit = options?.before
? HISTORY_MESSAGE_PAGE_SIZE
: Math.max(m.limit, storeMessageCount)
const page = await fetchMessages(sessionID, limit, options?.before)
// Commit the first page to the store immediately so the hydrating
// skeleton disappears after a single round-trip — but only when the
// page already contains a user message boundary. If the tail is
// assistant/tool-only (a very large final turn), committing now would
// drop the skeleton and render an empty chat (turn projection skips
// assistant messages without a user parent), which looks like a fresh
// session instead of a loading state. In that case defer the first
// commit to the expansion loop below, which fetches older records
// until a user boundary appears and commits that page instead.
//
// The deferral only applies to the initial fetch (no `before`).
// Prepend mode (loading older history) always commits — messages are
// already rendered, so there is no skeleton to protect, and skipping
// the store write would drop the fetched older messages entirely.
//
// The deferred fallback carries page.session (not []) so that if the
// expansion loop is ever a no-op (e.g. all nextLimit <= limit after a
// constant change), the final setMetaFor reflects the real fetched
// count instead of overwriting it with 0.
const deferFirstCommit =
!options?.before && !page.complete && !hasUserMessage(page.session)
let committed = deferFirstCommit
? { messages: page.session, cursor: page.cursor, complete: page.complete }
: commitMessagesToStore(page, options?.mode, options?.isStale)
// If the first commit detected a stale session, bail out immediately
// instead of relying on downstream guards to skip the final setMetaFor.
if (options?.isStale?.()) {
setMetaFor(sessionID, { loading: false })
return
}
// Keep the initial page small for switch performance. Some sessions
// have a very large final turn, so the latest records can
// contain only assistant/tool records and no user boundary. That makes
// turn projection render an empty chat until the user manually loads
// older messages. Expand only this initial tail fetch, with a hard cap.
// Each expanded page is committed to the store incrementally so the
// user sees content as soon as a user boundary appears, instead of
// waiting for the full expansion sequence before the first paint.
if (!options?.before && !page.complete && !hasUserMessage(page.session)) {
for (const nextLimit of getInitialPageExpansionLimits()) {
if (nextLimit <= limit) continue
page = await fetchMessages(sessionID, nextLimit)
if (page.complete || hasUserMessage(page.session)) break
if (options?.isStale?.()) {
setMetaFor(sessionID, { loading: false })
return
}
const expandedPage = await fetchMessages(sessionID, nextLimit)
if (options?.isStale?.()) {
setMetaFor(sessionID, { loading: false })
return
}
committed = commitMessagesToStore(expandedPage, options?.mode, options?.isStale)
if (options?.isStale?.()) {
setMetaFor(sessionID, { loading: false })
return
}
if (expandedPage.complete || hasUserMessage(expandedPage.session)) break
}
}
// Merge optimistic items
const items = getOptimistic(sessionID)
const merged = mergeOptimisticPage(page, items)
for (const messageID of merged.confirmed) {
clearOptimistic(sessionID, messageID)
}
if (options?.isStale?.()) {
setMetaFor(sessionID, { loading: false })
return
}
const current = store.getState()
const materialized = materializeSessionSnapshots(
current,
sessionID,
merged.session.map((info) => ({
info,
parts: merged.part.find((item) => item.id === info.id)?.part ?? [],
})),
{ skipPartTypes: SKIP_PARTS, mode: options?.mode === "prepend" ? "prepend" : "merge" },
)
if (options?.isStale?.()) {
setMetaFor(sessionID, { loading: false })
return
}
setMetaFor(sessionID, {
limit: materialized.messages.length,
cursor: merged.cursor,
complete: merged.complete,
limit: committed.messages.length,
cursor: committed.cursor,
complete: committed.complete,
loading: false,
})
if (materialized.messagesChanged || materialized.partsChanged) {
store.setState({
...(materialized.messagesChanged ? { message: materialized.message } : {}),
...(materialized.partsChanged ? { part: materialized.part } : {}),
})
}
setSessionPrefetch({
directory,
sessionID,
limit: materialized.messages.length,
cursor: merged.cursor,
complete: merged.complete,
limit: committed.messages.length,
cursor: committed.cursor,
complete: committed.complete,
})
} catch {
setMetaFor(sessionID, { loading: false })