From e892346c6b05b7d9636a259e69f52dab82fd611a Mon Sep 17 00:00:00 2001 From: Bohdan Triapitsyn Date: Thu, 7 May 2026 18:57:44 +0300 Subject: [PATCH] refactor: stabilize live chat sync materialization (#1132) Canonicalize session message/part materialization across load, prefetch, reconnect, and recovery paths so OpenChamber restores session snapshots through one consistent merge flow. Preserve live assistant streaming text when stale or delayed snapshots arrive, while still replacing optimistic user parts with confirmed server snapshots to avoid duplicated user messages. Narrow recovery triggers to explicit incomplete snapshot signals instead of broad session-event fallbacks, reducing unnecessary session refetches during active streaming. Keep turn windowing aligned with parented assistant replies and add regression coverage for materialization gaps, stale snapshot protection, optimistic user replacement, reconnect recovery, and turn grouping. --- .../ui/src/components/chat/ChatContainer.tsx | 27 +- .../ui/src/components/chat/ModelControls.tsx | 15 +- .../chat/lib/turns/windowTurns.test.ts | 48 +++ .../components/chat/lib/turns/windowTurns.ts | 16 +- .../src/components/session/SessionSidebar.tsx | 2 +- .../session/sidebar/SessionNodeItem.tsx | 4 +- .../sidebar/hooks/useSessionPrefetch.ts | 20 +- .../src/sync/__tests__/event-reducer.test.ts | 89 +++++ .../sync/__tests__/materialization.test.ts | 137 ++++++++ packages/ui/src/sync/event-pipeline.ts | 10 +- packages/ui/src/sync/event-reducer.ts | 45 ++- packages/ui/src/sync/materialization.ts | 203 +++++++++++ .../ui/src/sync/reconnect-recovery.test.ts | 12 +- packages/ui/src/sync/reconnect-recovery.ts | 14 +- packages/ui/src/sync/sync-context.tsx | 316 +++++++----------- packages/ui/src/sync/sync-refs.ts | 8 + packages/ui/src/sync/use-sync.ts | 44 +-- 17 files changed, 725 insertions(+), 285 deletions(-) create mode 100644 packages/ui/src/components/chat/lib/turns/windowTurns.test.ts create mode 100644 packages/ui/src/sync/__tests__/event-reducer.test.ts create mode 100644 packages/ui/src/sync/__tests__/materialization.test.ts create mode 100644 packages/ui/src/sync/materialization.ts diff --git a/packages/ui/src/components/chat/ChatContainer.tsx b/packages/ui/src/components/chat/ChatContainer.tsx index cd202f93..fdbb7f76 100644 --- a/packages/ui/src/components/chat/ChatContainer.tsx +++ b/packages/ui/src/components/chat/ChatContainer.tsx @@ -39,6 +39,7 @@ import { useSessionStatus, } from '@/sync/sync-context'; import { useSync } from '@/sync/use-sync'; +import { getSessionMaterializationStatus } from '@/sync/materialization'; import { usePlanDetection } from '@/hooks/usePlanDetection'; import { getAllSyncSessions } from '@/sync/sync-refs'; import { useI18n } from '@/lib/i18n'; @@ -329,8 +330,8 @@ export const ChatContainer: React.FC = () => { // Sync actions const sync = useSync(); - const loadMessages = React.useCallback( - (sessionId: string) => sync.syncSession(sessionId), + const ensureSessionRenderable = React.useCallback( + (sessionId: string) => sync.ensureSessionRenderable(sessionId), [sync], ); const loadMoreMessages = React.useCallback( @@ -363,9 +364,9 @@ export const ChatContainer: React.FC = () => { ), ); const sessionMessageCount = useSessionMessageCount(currentSessionId ?? ''); - const hasLoadedSessionMessages = useDirectorySync( + const hasRenderableSessionSnapshot = useDirectorySync( React.useCallback( - (state) => (currentSessionId ? state.message[currentSessionId] !== undefined : false), + (state) => (currentSessionId ? getSessionMaterializationStatus(state, currentSessionId).renderable : false), [currentSessionId], ), ); @@ -425,10 +426,6 @@ export const ChatContainer: React.FC = () => { return false; } - if (streamingMessageId || activeStreamingPhase) { - return true; - } - const statusType = sessionStatusForCurrent.type ?? 'idle'; if (statusType === 'busy' || statusType === 'retry') { return true; @@ -440,7 +437,7 @@ export const ChatContainer: React.FC = () => { && lastMessage.role === 'assistant' && typeof (lastMessage as { time?: { completed?: number } }).time?.completed !== 'number', ); - }, [activeStreamingPhase, currentSessionId, sessionMessages, sessionPermissions.length, sessionQuestions.length, sessionStatusForCurrent.type, streamingMessageId]); + }, [currentSessionId, sessionMessages, sessionPermissions.length, sessionQuestions.length, sessionStatusForCurrent.type]); const activeRetryStatus = React.useMemo(() => { if (!currentSessionId || sessionStatusForCurrent.type !== 'retry') { return null; @@ -751,7 +748,7 @@ export const ChatContainer: React.FC = () => { const isSessionHydrating = Boolean(currentSessionId) - && !hasLoadedSessionMessages; + && !hasRenderableSessionSnapshot; React.useEffect(() => { if (!currentSessionId) { @@ -783,10 +780,10 @@ export const ChatContainer: React.FC = () => { React.useEffect(() => { if (!currentSessionId) return; - if (hasLoadedSessionMessages) return; + if (hasRenderableSessionSnapshot) return; const load = async () => { - await loadMessages(currentSessionId).finally(() => { + await ensureSessionRenderable(currentSessionId).finally(() => { const statusType = sessionStatusForCurrent.type ?? 'idle'; const isActivePhase = statusType === 'busy' || statusType === 'retry'; const hasHashTarget = typeof window !== 'undefined' && window.location.hash.length > 0; @@ -805,7 +802,7 @@ export const ChatContainer: React.FC = () => { }; void load(); - }, [currentSessionId, hasLoadedSessionMessages, isPinned, loadMessages, resumeToLatestInstant, sessionStatusForCurrent.type]); + }, [currentSessionId, ensureSessionRenderable, hasRenderableSessionSnapshot, isPinned, resumeToLatestInstant, sessionStatusForCurrent.type]); if (!currentSessionId && !draftOpen) { return ( @@ -841,7 +838,7 @@ export const ChatContainer: React.FC = () => { return null; } - if (isSessionHydrating && sessionMessages.length === 0 && !streamingMessageId) { + if (isSessionHydrating && sessionMessages.length === 0 && !sessionIsWorking) { return (
{returnToParentButton} @@ -897,7 +894,7 @@ export const ChatContainer: React.FC = () => { ); } - if (sessionMessages.length === 0 && !streamingMessageId) { + if (sessionMessages.length === 0 && !sessionIsWorking) { return (
{returnToParentButton} diff --git a/packages/ui/src/components/chat/ModelControls.tsx b/packages/ui/src/components/chat/ModelControls.tsx index 8073d138..c6357440 100644 --- a/packages/ui/src/components/chat/ModelControls.tsx +++ b/packages/ui/src/components/chat/ModelControls.tsx @@ -63,6 +63,7 @@ import { useSessionUIStore } from '@/sync/session-ui-store'; import { useSelectionStore } from '@/sync/selection-store'; import { useDirectorySync, useSessionMessages } from '@/sync/sync-context'; import { useSync } from '@/sync/use-sync'; +import { getSessionMaterializationStatus } from '@/sync/materialization'; import { useUIStore } from '@/stores/useUIStore'; import { useModelLists } from '@/hooks/useModelLists'; import { useIsTextTruncated } from '@/hooks/useIsTextTruncated'; @@ -768,9 +769,9 @@ export const ModelControls: React.FC = ({ const latestLoadedUserChoiceRestoreRef = React.useRef(null); const currentSessionDirectory = currentSessionId ? getDirectoryForSession(currentSessionId) : undefined; - const hasCurrentSessionMessagesEntry = useDirectorySync( + const hasRenderableCurrentSessionSnapshot = useDirectorySync( React.useCallback( - (state) => (currentSessionId ? state.message[currentSessionId] !== undefined : false), + (state) => (currentSessionId ? getSessionMaterializationStatus(state, currentSessionId).renderable : false), [currentSessionId], ), currentSessionDirectory ?? undefined, @@ -942,7 +943,7 @@ export const ModelControls: React.FC = ({ return; } - if (!contextHydrated || providers.length === 0 || !hasCurrentSessionMessagesEntry || !latestLoadedUserChoice?.providerID || !latestLoadedUserChoice.modelID) { + if (!contextHydrated || providers.length === 0 || !hasRenderableCurrentSessionSnapshot || !latestLoadedUserChoice?.providerID || !latestLoadedUserChoice.modelID) { return; } @@ -990,7 +991,7 @@ export const ModelControls: React.FC = ({ currentAgentName, contextHydrated, providers, - hasCurrentSessionMessagesEntry, + hasRenderableCurrentSessionSnapshot, latestLoadedUserChoice, setAgent, tryApplyModelSelection, @@ -1113,9 +1114,9 @@ export const ModelControls: React.FC = ({ return; } - if (!hasCurrentSessionMessagesEntry) { + if (!hasRenderableCurrentSessionSnapshot) { if (!sync.isLoading(currentSessionId)) { - void sync.syncSession(currentSessionId); + void sync.ensureSessionRenderable(currentSessionId); } return; } @@ -1127,7 +1128,7 @@ export const ModelControls: React.FC = ({ applyFallbackAgent(); }, [ currentSessionId, - hasCurrentSessionMessagesEntry, + hasRenderableCurrentSessionSnapshot, latestLoadedUserChoice, agents, primaryAgents, diff --git a/packages/ui/src/components/chat/lib/turns/windowTurns.test.ts b/packages/ui/src/components/chat/lib/turns/windowTurns.test.ts new file mode 100644 index 00000000..bb819449 --- /dev/null +++ b/packages/ui/src/components/chat/lib/turns/windowTurns.test.ts @@ -0,0 +1,48 @@ +import { describe, expect, test } from 'bun:test'; +import type { Message, Part } from '@opencode-ai/sdk/v2'; +import { buildTurnWindowModel, updateTurnWindowModelIncremental } from './windowTurns'; +import type { ChatMessageEntry } from './types'; + +function message({ id, role, parentID }: { id: string; role: 'user' | 'assistant' | 'system'; parentID?: string }): ChatMessageEntry { + return { + info: { + id, + role, + ...(parentID ? { parentID } : {}), + time: { created: 1 }, + } as Message, + parts: [] as Part[], + }; +} + +describe('windowTurns', () => { + test('does not map assistant messages without a parent to the current turn', () => { + const user = message({ id: 'u1', role: 'user' }); + const assistant = message({ id: 'a1', role: 'assistant' }); + + const model = buildTurnWindowModel([user, assistant]); + + expect(model.messageToTurnId.get('u1')).toBe('u1'); + expect(model.messageToTurnId.has('a1')).toBe(false); + }); + + test('incremental update does not map assistant messages without a parent to the current turn', () => { + const user = message({ id: 'u1', role: 'user' }); + const assistant = message({ id: 'a1', role: 'assistant' }); + const base = buildTurnWindowModel([user]); + + const next = updateTurnWindowModelIncremental(base, [user], [user, assistant]); + + expect(next?.messageToTurnId.get('u1')).toBe('u1'); + expect(next?.messageToTurnId.has('a1')).toBe(false); + }); + + test('maps assistant messages to their parent user turn', () => { + const user = message({ id: 'u1', role: 'user' }); + const assistant = message({ id: 'a1', role: 'assistant', parentID: 'u1' }); + + const model = buildTurnWindowModel([user, assistant]); + + expect(model.messageToTurnId.get('a1')).toBe('u1'); + }); +}); diff --git a/packages/ui/src/components/chat/lib/turns/windowTurns.ts b/packages/ui/src/components/chat/lib/turns/windowTurns.ts index c6c24912..b141b4ec 100644 --- a/packages/ui/src/components/chat/lib/turns/windowTurns.ts +++ b/packages/ui/src/components/chat/lib/turns/windowTurns.ts @@ -119,9 +119,10 @@ export const updateTurnWindowModelIncremental = ( } const parentId = resolveParentMessageId(nextMessage); - const targetTurnIndex = parentId - ? nextModel.turnIndexById.get(parentId) - : nextModel.turnIds.length - 1; + if (!parentId) { + return nextModel; + } + const targetTurnIndex = nextModel.turnIndexById.get(parentId); if (typeof targetTurnIndex !== 'number' || targetTurnIndex < 0) { return null; } @@ -173,8 +174,13 @@ export const buildTurnWindowModel = (messages: ChatMessageEntry[]): TurnWindowMo } const parentId = resolveParentMessageId(message); - const parentTurnIndex = parentId ? userMessageToTurnIndex.get(parentId) : undefined; - const targetTurnIndex = typeof parentTurnIndex === 'number' ? parentTurnIndex : currentTurnIndex; + if (!parentId) { + return; + } + const targetTurnIndex = userMessageToTurnIndex.get(parentId); + if (typeof targetTurnIndex !== 'number') { + return; + } if (targetTurnIndex < 0) { return; } diff --git a/packages/ui/src/components/session/SessionSidebar.tsx b/packages/ui/src/components/session/SessionSidebar.tsx index 20312015..a5c22701 100644 --- a/packages/ui/src/components/session/SessionSidebar.tsx +++ b/packages/ui/src/components/session/SessionSidebar.tsx @@ -1087,7 +1087,7 @@ export const SessionSidebar: React.FC = ({ currentSessionId, sortedSessions, recentSessionIds: recentSessionIdsList, - loadMessages: sync.syncSession, + ensureSessionRenderable: sync.ensureSessionRenderable, }); const sectionsForSidebarRender = React.useMemo(() => { diff --git a/packages/ui/src/components/session/sidebar/SessionNodeItem.tsx b/packages/ui/src/components/session/sidebar/SessionNodeItem.tsx index e9695cd8..bca7c47f 100644 --- a/packages/ui/src/components/session/sidebar/SessionNodeItem.tsx +++ b/packages/ui/src/components/session/sidebar/SessionNodeItem.tsx @@ -347,7 +347,7 @@ function SessionNodeItemComponent(props: Props): React.ReactNode { let skipped = 0; for (const child of children) { try { - await sync.syncSession(child.session.id); + await sync.ensureSessionRenderable(child.session.id); const childRecords = buildSessionMessageRecordsSnapshot(directoryStore.getState(), child.session.id).list; const childTitle = child.session.title || t('sessions.sidebar.session.export.untitledSubagent'); const childAgent = (child.session as Session & { agent?: string }).agent; @@ -379,7 +379,7 @@ function SessionNodeItemComponent(props: Props): React.ReactNode { return; } - await sync.syncSession(session.id); + await sync.ensureSessionRenderable(session.id); const records = buildSessionMessageRecordsSnapshot(directoryStore.getState(), session.id).list; if (records.length === 0) { diff --git a/packages/ui/src/components/session/sidebar/hooks/useSessionPrefetch.ts b/packages/ui/src/components/session/sidebar/hooks/useSessionPrefetch.ts index 211ebd07..26734f36 100644 --- a/packages/ui/src/components/session/sidebar/hooks/useSessionPrefetch.ts +++ b/packages/ui/src/components/session/sidebar/hooks/useSessionPrefetch.ts @@ -1,7 +1,7 @@ import React from 'react'; import type { Session } from '@opencode-ai/sdk/v2'; import { useSessionUIStore } from '@/sync/session-ui-store'; -import { getSyncMessages } from '@/sync/sync-refs'; +import { getSyncSessionMaterializationStatus } from '@/sync/sync-refs'; const SESSION_PREFETCH_HOVER_DELAY_MS = 180; const SESSION_PREFETCH_SETTLE_MS = 600; @@ -12,10 +12,10 @@ type Args = { currentSessionId: string | null; sortedSessions: Session[]; recentSessionIds?: string[]; - loadMessages: (sessionId: string) => Promise; + ensureSessionRenderable: (sessionId: string) => Promise; }; -export const useSessionPrefetch = ({ currentSessionId, sortedSessions, recentSessionIds = [], loadMessages }: Args): void => { +export const useSessionPrefetch = ({ currentSessionId, sortedSessions, recentSessionIds = [], ensureSessionRenderable }: Args): void => { const sessionPrefetchTimersRef = React.useRef>(new Map()); const sessionPrefetchQueueRef = React.useRef([]); const sessionPrefetchInFlightRef = React.useRef>(new Set()); @@ -36,30 +36,28 @@ export const useSessionPrefetch = ({ currentSessionId, sortedSessions, recentSes continue; } - // Check if messages already loaded in sync child store - const hasMessages = getSyncMessages(nextSessionId).length > 0; - if (hasMessages) { + // Check if the session is already renderable in the sync child store. + if (getSyncSessionMaterializationStatus(nextSessionId).renderable) { continue; } sessionPrefetchInFlightRef.current.add(nextSessionId); - void loadMessages(nextSessionId) + void ensureSessionRenderable(nextSessionId) .catch(() => undefined) .finally(() => { sessionPrefetchInFlightRef.current.delete(nextSessionId); pumpSessionPrefetchQueue(); }); } - }, [loadMessages]); + }, [ensureSessionRenderable]); const scheduleSessionPrefetch = React.useCallback((sessionId: string | null | undefined) => { if (!sessionId || sessionId === currentSessionId || typeof window === 'undefined') { return; } - // Already loaded in sync - const hasMessages = getSyncMessages(sessionId).length > 0; - if (hasMessages) { + // Already renderable in sync + if (getSyncSessionMaterializationStatus(sessionId).renderable) { return; } diff --git a/packages/ui/src/sync/__tests__/event-reducer.test.ts b/packages/ui/src/sync/__tests__/event-reducer.test.ts new file mode 100644 index 00000000..bef5e324 --- /dev/null +++ b/packages/ui/src/sync/__tests__/event-reducer.test.ts @@ -0,0 +1,89 @@ +import { describe, expect, test } from "bun:test" +import type { Event, Part } from "@opencode-ai/sdk/v2/client" +import { applyDirectoryEvent } from "../event-reducer" +import { INITIAL_STATE, type State } from "../types" + +function state(overrides: Partial = {}): State { + return { + ...INITIAL_STATE, + message: {}, + part: {}, + ...overrides, + } +} + +function deltaEvent(): Event { + return { + type: "message.part.delta", + properties: { + messageID: "msg_1", + partID: "prt_1", + field: "text", + delta: "hello", + }, + } as Event +} + +function partUpdatedEvent(): Event { + return { + type: "message.part.updated", + properties: { + part: { + id: "prt_1", + messageID: "msg_1", + sessionID: "ses_1", + type: "text", + text: "hello", + }, + }, + } as Event +} + +describe("applyDirectoryEvent", () => { + test("returns typed materialization when delta arrives before parts", () => { + const result = applyDirectoryEvent(state(), deltaEvent()) + + expect(result).toEqual({ + changed: false, + materialization: { type: "incomplete-session-snapshot", messageID: "msg_1", partID: "prt_1" }, + }) + }) + + test("returns typed materialization when delta part is missing", () => { + const result = applyDirectoryEvent( + state({ part: { msg_1: [{ id: "prt_2", messageID: "msg_1", type: "text", text: "" } as Part] } }), + deltaEvent(), + ) + + expect(result).toEqual({ + changed: false, + materialization: { type: "incomplete-session-snapshot", messageID: "msg_1", partID: "prt_1" }, + }) + }) + + test("applies part update and requests materialization when owning message is absent", () => { + const draft = state() + const result = applyDirectoryEvent(draft, partUpdatedEvent()) + + expect(draft.part.msg_1.map((item) => item.id)).toEqual(["prt_1"]) + expect(result).toEqual({ + changed: true, + materialization: { + type: "incomplete-session-snapshot", + sessionID: "ses_1", + messageID: "msg_1", + partID: "prt_1", + }, + }) + }) + + test("applies part update without materialization when owning message exists", () => { + const draft = state({ + message: { ses_1: [{ id: "msg_1", sessionID: "ses_1", role: "assistant", time: { created: 1 } } as never] }, + }) + const result = applyDirectoryEvent(draft, partUpdatedEvent()) + + expect(draft.part.msg_1.map((item) => item.id)).toEqual(["prt_1"]) + expect(result).toBe(true) + }) +}) diff --git a/packages/ui/src/sync/__tests__/materialization.test.ts b/packages/ui/src/sync/__tests__/materialization.test.ts new file mode 100644 index 00000000..83cf709d --- /dev/null +++ b/packages/ui/src/sync/__tests__/materialization.test.ts @@ -0,0 +1,137 @@ +import { describe, expect, test } from "bun:test" +import type { Message, Part } from "@opencode-ai/sdk/v2/client" +import { getSessionMaterializationStatus, materializeSessionSnapshots } from "../materialization" + +function message(id: string, sessionID = "ses_1"): Message { + return { id, sessionID, role: "assistant", time: { created: 1 } } as Message +} + +function userMessage(id: string, sessionID = "ses_1"): Message { + return { id, sessionID, role: "user", time: { created: 1 } } as Message +} + +function part(id: string, messageID: string, type = "text", text = id): Part { + return { id, messageID, sessionID: "ses_1", type, text } as Part +} + +describe("materializeSessionSnapshots", () => { + test("materializes messages and parts together", () => { + const result = materializeSessionSnapshots( + { message: {}, part: {} }, + "ses_1", + [{ info: message("msg_1"), parts: [part("prt_1", "msg_1")] }], + ) + + expect(result.message.ses_1.map((item) => item.id)).toEqual(["msg_1"]) + expect(result.part.msg_1.map((item) => item.id)).toEqual(["prt_1"]) + expect(result.messagesChanged).toBe(true) + expect(result.partsChanged).toBe(true) + }) + + test("preserves unchanged references", () => { + const existingMessage = message("msg_1") + const existingPart = part("prt_1", "msg_1") + const state = { message: { ses_1: [existingMessage] }, part: { msg_1: [existingPart] } } + + const result = materializeSessionSnapshots( + state, + "ses_1", + [{ info: existingMessage, parts: [existingPart] }], + ) + + expect(result.message).toBe(state.message) + expect(result.part).toBe(state.part) + expect(result.messagesChanged).toBe(false) + expect(result.partsChanged).toBe(false) + }) + + test("skips non-rendered part types", () => { + const result = materializeSessionSnapshots( + { message: {}, part: {} }, + "ses_1", + [{ info: message("msg_1"), parts: [part("prt_patch", "msg_1", "patch"), part("prt_text", "msg_1")] }], + { skipPartTypes: new Set(["patch"]) }, + ) + + expect(result.part.msg_1.map((item) => item.id)).toEqual(["prt_text"]) + }) + + test("preserves newer live streaming text when a stale snapshot materializes", () => { + const livePart = part("prt_1", "msg_1", "text", "First chunk ") + const stalePart = part("prt_1", "msg_1", "text", "") + const state = { + message: { ses_1: [message("msg_1")] }, + part: { msg_1: [livePart] }, + } + + const result = materializeSessionSnapshots( + state, + "ses_1", + [{ info: message("msg_1"), parts: [stalePart] }], + ) + + expect(result.part.msg_1[0]).toBe(livePart) + expect((result.part.msg_1[0] as { text?: string })?.text).toBe("First chunk ") + }) + + test("preserves live streaming parts omitted by a stale snapshot", () => { + const livePart = part("prt_1", "msg_1", "text", "First chunk ") + const state = { + message: { ses_1: [message("msg_1")] }, + part: { msg_1: [livePart] }, + } + + const result = materializeSessionSnapshots( + state, + "ses_1", + [{ info: message("msg_1"), parts: [] }], + ) + + expect(result.part.msg_1[0]).toBe(livePart) + }) + + test("does not preserve omitted optimistic user text parts beside server snapshot parts", () => { + const optimisticPart = { id: "prt_optimistic", messageID: "msg_1", type: "text", text: "Hello" } as Part + const serverPart = part("prt_server", "msg_1", "text", "Hello") + const state = { + message: { ses_1: [userMessage("msg_1")] }, + part: { msg_1: [optimisticPart] }, + } + + const result = materializeSessionSnapshots( + state, + "ses_1", + [{ info: userMessage("msg_1"), parts: [serverPart] }], + ) + + expect(result.part.msg_1).toEqual([serverPart]) + }) +}) + +describe("getSessionMaterializationStatus", () => { + test("requires assistant parts for renderable cached state", () => { + const state = { + message: { ses_1: [message("msg_1")] }, + part: {}, + } + + expect(getSessionMaterializationStatus(state, "ses_1")).toEqual({ + hasMessages: true, + renderable: false, + missingPartMessageIDs: ["msg_1"], + }) + }) + + test("treats user-only cached state as renderable", () => { + const state = { + message: { ses_1: [{ ...message("msg_1"), role: "user" } as Message] }, + part: {}, + } + + expect(getSessionMaterializationStatus(state, "ses_1")).toEqual({ + hasMessages: true, + renderable: true, + missingPartMessageIDs: [], + }) + }) +}) diff --git a/packages/ui/src/sync/event-pipeline.ts b/packages/ui/src/sync/event-pipeline.ts index 91346823..e680c536 100644 --- a/packages/ui/src/sync/event-pipeline.ts +++ b/packages/ui/src/sync/event-pipeline.ts @@ -136,7 +136,15 @@ function toWebSocketUrl(candidate: string): string { } function buildGlobalEventWsUrl(lastEventId?: string): string { - const baseUrl = opencodeClient.getBaseUrl() + let baseUrl = "/api" + try { + const client = opencodeClient as { getBaseUrl?: () => string } + if (typeof client.getBaseUrl === "function") { + baseUrl = client.getBaseUrl() + } + } catch { + baseUrl = "/api" + } const normalizedBase = baseUrl.endsWith("/") ? baseUrl : `${baseUrl}/` const httpUrl = new URL("global/event/ws", resolveAbsoluteUrl(normalizedBase)) if (lastEventId && lastEventId.length > 0) { diff --git a/packages/ui/src/sync/event-reducer.ts b/packages/ui/src/sync/event-reducer.ts index a8d71eb8..5ef37a85 100644 --- a/packages/ui/src/sync/event-reducer.ts +++ b/packages/ui/src/sync/event-reducer.ts @@ -101,6 +101,23 @@ export type GlobalEventResult = { project: Project } | null +export type DirectoryEventResult = boolean | { + changed: boolean + materialization: { + type: "incomplete-session-snapshot" + sessionID?: string + messageID: string + partID?: string + } +} + +function hasMessage(draft: State, sessionID: string | undefined, messageID: string): boolean { + if (!sessionID) return false + const messages = draft.message[sessionID] + if (!messages) return false + return Binary.search(messages, messageID, (message) => message.id).found +} + export function reduceGlobalEvent(event: Event): GlobalEventResult { if (event.type === "global.disposed" || event.type === "server.connected") { return { type: "refresh" } @@ -135,7 +152,7 @@ export function applyDirectoryEvent( onLoadLsp?: () => void onSetSessionTodo?: (sessionID: string, todos: Todo[] | undefined) => void }, -): boolean { +): DirectoryEventResult { switch (event.type) { case "server.instance.disposed": { callbacks?.onRefresh?.("") @@ -269,11 +286,18 @@ export function applyDirectoryEvent( return false } const messageID = (part as { messageID: string }).messageID + const sessionID = (part as { sessionID?: string }).sessionID + const missingOwningMessage = !hasMessage(draft, sessionID, messageID) const parts = draft.part[messageID] if (!parts) { syncDebug.reducer.partUpdatedNoExistingParts(messageID, part.id, part.type) draft.part[messageID] = [part] - return true + return missingOwningMessage + ? { + changed: true, + materialization: { type: "incomplete-session-snapshot", sessionID, messageID, partID: part.id }, + } + : true } const next = [...parts] const result = Binary.search(next, part.id, (p) => p.id) @@ -302,7 +326,12 @@ export function applyDirectoryEvent( next.splice(insertResult.index, 0, part) } draft.part[messageID] = next - return true + return missingOwningMessage + ? { + changed: true, + materialization: { type: "incomplete-session-snapshot", sessionID, messageID, partID: part.id }, + } + : true } case "message.part.removed": { @@ -333,12 +362,18 @@ export function applyDirectoryEvent( const parts = draft.part[props.messageID] if (!parts) { syncDebug.reducer.partDeltaNoParts(props.messageID, props.partID) - return false + return { + changed: false, + materialization: { type: "incomplete-session-snapshot", messageID: props.messageID, partID: props.partID }, + } } const result = Binary.search(parts, props.partID, (p) => p.id) if (!result.found) { syncDebug.reducer.partDeltaNotFound(props.messageID, props.partID) - return false + return { + changed: false, + materialization: { type: "incomplete-session-snapshot", messageID: props.messageID, partID: props.partID }, + } } const existing = parts[result.index] as Record const existingValue = existing[props.field] as string | undefined diff --git a/packages/ui/src/sync/materialization.ts b/packages/ui/src/sync/materialization.ts new file mode 100644 index 00000000..4d2c5bc7 --- /dev/null +++ b/packages/ui/src/sync/materialization.ts @@ -0,0 +1,203 @@ +import type { Message, Part } from "@opencode-ai/sdk/v2/client" +import { mergeMessages } from "./optimistic" + +const cmp = (a: string, b: string) => (a < b ? -1 : a > b ? 1 : 0) +const STREAMING_PART_FIELDS = ["text", "output"] as const + +export type MaterializedMessageRecord = { + info: Message + parts: Part[] +} + +export type MaterializedState = { + message: Record + part: Record +} + +export type MaterializeSessionSnapshotsOptions = { + skipPartTypes?: ReadonlySet + mode?: "merge" | "prepend" +} + +export type MaterializeSessionSnapshotsResult = { + message: Record + part: Record + messages: Message[] + messagesChanged: boolean + partsChanged: boolean +} + +export type SessionMaterializationStatus = { + hasMessages: boolean + renderable: boolean + missingPartMessageIDs: string[] +} + +function sortParts(parts: Part[], skipPartTypes: ReadonlySet) { + return parts + .filter((part) => !!part?.id && !skipPartTypes.has(part.type)) + .sort((a, b) => cmp(a.id, b.id)) +} + +function haveEquivalentPartSnapshots(left: Part[] | undefined, right: Part[]): boolean { + if (!left) return right.length === 0 + if (left.length !== right.length) return false + + for (let index = 0; index < left.length; index += 1) { + const leftPart = left[index] + const rightPart = right[index] + if (!leftPart || !rightPart) return false + if (leftPart.id !== rightPart.id) return false + if (JSON.stringify(leftPart) !== JSON.stringify(rightPart)) return false + } + + return true +} + +function getPartEndTime(part: Part): number | undefined { + const stateEnd = (part as { state?: { time?: { end?: unknown } } }).state?.time?.end + if (typeof stateEnd === "number") { + return stateEnd + } + + const timeEnd = (part as { time?: { end?: unknown } }).time?.end + return typeof timeEnd === "number" ? timeEnd : undefined +} + +function getStringField(part: Part, field: "text" | "output"): string | undefined { + const value = (part as Record)[field] + return typeof value === "string" ? value : undefined +} + +function hasLiveStreamingField(part: Part): boolean { + if (getPartEndTime(part) !== undefined) return false + return STREAMING_PART_FIELDS.some((field) => { + const value = getStringField(part, field) + return typeof value === "string" && value.length > 0 + }) +} + +function mergeMaterializedPart(existing: Part | undefined, next: Part): Part { + if (!existing || getPartEndTime(next) !== undefined) return next + + let merged: Part = next + for (const field of STREAMING_PART_FIELDS) { + const existingValue = getStringField(existing, field) + if (!existingValue) continue + + const nextValue = getStringField(next, field) + if (typeof nextValue === "string" && nextValue.length >= existingValue.length) continue + if (typeof nextValue === "string" && nextValue.length > 0 && !existingValue.startsWith(nextValue)) continue + + if (merged === next) merged = { ...next } + const mergedRecord = merged as Record + mergedRecord[field] = existingValue + } + + return merged +} + +function mergeMaterializedParts( + existing: Part[] | undefined, + nextParts: Part[], + skipPartTypes: ReadonlySet, + preserveLiveStreamingParts: boolean, +): Part[] { + if (!existing || existing.length === 0) return nextParts + if (!preserveLiveStreamingParts) return nextParts + + const existingByID = new Map(existing.map((part) => [part.id, part])) + let mergedParts = nextParts + let changed = false + + for (let index = 0; index < nextParts.length; index += 1) { + const nextPart = nextParts[index] + const mergedPart = mergeMaterializedPart(existingByID.get(nextPart.id), nextPart) + if (mergedPart === nextPart) continue + if (!changed) mergedParts = [...nextParts] + mergedParts[index] = mergedPart + changed = true + } + + const snapshotIDs = new Set(nextParts.map((part) => part.id)) + const missingLiveParts = existing.filter( + (part) => !!part?.id && !snapshotIDs.has(part.id) && !skipPartTypes.has(part.type) && hasLiveStreamingField(part), + ) + if (missingLiveParts.length === 0) return mergedParts + + return [...mergedParts, ...missingLiveParts].sort((a, b) => cmp(a.id, b.id)) +} + +export function materializeSessionSnapshots( + state: MaterializedState, + sessionID: string, + records: MaterializedMessageRecord[], + options: MaterializeSessionSnapshotsOptions = {}, +): MaterializeSessionSnapshotsResult { + const skipPartTypes = options.skipPartTypes ?? new Set() + const snapshots = records + .filter((record) => !!record?.info?.id) + .sort((left, right) => cmp(left.info.id, right.info.id)) + const nextMessages = snapshots.map((record) => record.info) + const currentMessages = state.message[sessionID] ?? [] + const messages = mergeMessages(currentMessages, nextMessages) + const messagesChanged = messages !== currentMessages + + let partsChanged = false + const nextPartState = { ...state.part } + const isPrepend = options.mode === "prepend" + + for (const record of snapshots) { + const messageID = record.info.id + if (isPrepend && nextPartState[messageID]) continue + + const existing = nextPartState[messageID] + const nextParts = mergeMaterializedParts( + existing, + sortParts(record.parts ?? [], skipPartTypes), + skipPartTypes, + record.info.role === "assistant", + ) + if (haveEquivalentPartSnapshots(existing, nextParts)) continue + + if (nextParts.length === 0) { + delete nextPartState[messageID] + } else { + nextPartState[messageID] = nextParts + } + partsChanged = true + } + + return { + message: messagesChanged ? { ...state.message, [sessionID]: messages } : state.message, + part: partsChanged ? nextPartState : state.part, + messages, + messagesChanged, + partsChanged, + } +} + +export function getSessionMaterializationStatus( + state: MaterializedState, + sessionID: string, +): SessionMaterializationStatus { + const messages = state.message[sessionID] + if (!messages) { + return { hasMessages: false, renderable: false, missingPartMessageIDs: [] } + } + + const missingPartMessageIDs: string[] = [] + for (const message of messages) { + if (message.role !== "assistant") continue + const parts = state.part[message.id] + if (!parts || parts.length === 0) { + missingPartMessageIDs.push(message.id) + } + } + + return { + hasMessages: true, + renderable: missingPartMessageIDs.length === 0, + missingPartMessageIDs, + } +} diff --git a/packages/ui/src/sync/reconnect-recovery.test.ts b/packages/ui/src/sync/reconnect-recovery.test.ts index fcede14b..1f0c44eb 100644 --- a/packages/ui/src/sync/reconnect-recovery.test.ts +++ b/packages/ui/src/sync/reconnect-recovery.test.ts @@ -1,5 +1,5 @@ import { describe, expect, test } from "bun:test" -import type { Message, SessionStatus } from "@opencode-ai/sdk/v2/client" +import type { Message, Part, SessionStatus } from "@opencode-ai/sdk/v2/client" import type { Session } from "@opencode-ai/sdk/v2" import { getReconnectCandidateSessionIds } from "./reconnect-recovery" @@ -23,6 +23,10 @@ function createAssistantMessage(id: string, sessionID: string, completed?: numbe } as unknown as Message } +function createPart(id: string, messageID: string): Part { + return { id, messageID, sessionID: "active", type: "text", text: "done" } as Part +} + describe("getReconnectCandidateSessionIds", () => { test("includes non-idle, incomplete assistant, and parent sessions", () => { const busyStatus = { type: "busy" } as SessionStatus @@ -48,6 +52,9 @@ describe("getReconnectCandidateSessionIds", () => { message: { active: [createAssistantMessage("m-1", "active", 1)], }, + part: { + "m-1": [createPart("p-1", "m-1")], + }, }, { directory: "/repo", viewedSession: { directory: "/repo", sessionId: "active" }, @@ -72,6 +79,9 @@ describe("getReconnectCandidateSessionIds", () => { message: { active: [createAssistantMessage("m-1", "active", 1)], }, + part: { + "m-1": [createPart("p-1", "m-1")], + }, }, { directory: "/repo-a", viewedSession: { directory: "/repo-b", sessionId: "active" }, diff --git a/packages/ui/src/sync/reconnect-recovery.ts b/packages/ui/src/sync/reconnect-recovery.ts index e7e8c75f..68daaa05 100644 --- a/packages/ui/src/sync/reconnect-recovery.ts +++ b/packages/ui/src/sync/reconnect-recovery.ts @@ -1,24 +1,25 @@ import type { SessionStatus, Message, Part } from "@opencode-ai/sdk/v2/client" import type { Session } from "@opencode-ai/sdk/v2" +import { getSessionMaterializationStatus } from "./materialization" -type ReconnectRecoveryState = { +type ReconnectMaterializationState = { session: Session[] session_status?: Record message?: Record part?: Record } -export type ViewedSessionRecoveryTarget = { +export type ViewedSessionMaterializationTarget = { directory: string sessionId: string } type ReconnectCandidateOptions = { directory?: string - viewedSession?: ViewedSessionRecoveryTarget | null + viewedSession?: ViewedSessionMaterializationTarget | null } -export function getReconnectCandidateSessionIds(state: ReconnectRecoveryState, options?: ReconnectCandidateOptions) { +export function getReconnectCandidateSessionIds(state: ReconnectMaterializationState, options?: ReconnectCandidateOptions) { const ids = new Set() for (const [sessionId, status] of Object.entries(state.session_status ?? {})) { @@ -27,16 +28,13 @@ export function getReconnectCandidateSessionIds(state: ReconnectRecoveryState, o for (const [sessionId, messages] of Object.entries(state.message ?? {})) { const lastMessage = messages[messages.length - 1] - const lastAssistantComplete = lastMessage - && lastMessage.role === "assistant" - && typeof (lastMessage as { time?: { completed?: number } }).time?.completed === "number" if ( lastMessage && lastMessage.role === "assistant" && typeof (lastMessage as { time?: { completed?: number } }).time?.completed !== "number" ) { ids.add(sessionId) - } else if (lastAssistantComplete && state.part && (state.part[lastMessage.id]?.length ?? 0) === 0) { + } else if (!getSessionMaterializationStatus({ message: state.message ?? {}, part: state.part ?? {} }, sessionId).renderable) { ids.add(sessionId) } } diff --git a/packages/ui/src/sync/sync-context.tsx b/packages/ui/src/sync/sync-context.tsx index fc9159cc..428b6e3b 100644 --- a/packages/ui/src/sync/sync-context.tsx +++ b/packages/ui/src/sync/sync-context.tsx @@ -36,7 +36,7 @@ import type { SessionStatus } from "@opencode-ai/sdk/v2/client" import type { PermissionRequest } from "@/types/permission" import type { QuestionRequest } from "@/types/question" import * as sessionActions from "./session-actions" -import { mergeMessages } from "./optimistic" +import { getSessionMaterializationStatus, materializeSessionSnapshots } from "./materialization" // --------------------------------------------------------------------------- // Context @@ -123,7 +123,7 @@ let bootingRoot = false let bootedAt = 0 const BOOT_DEBOUNCE_MS = 1500 const RECONNECT_MESSAGE_LIMIT = 30 -const REPAIR_MESSAGE_LIMIT = 30 +const SESSION_MATERIALIZATION_MESSAGE_LIMIT = 30 const RECONNECT_SKIP_PARTS = new Set(["patch", "step-start", "step-finish"]) const requestSignature = (items: Array<{ id: string }> | undefined): string => { if (!items || items.length === 0) return "" @@ -135,117 +135,77 @@ const requestSignature = (items: Array<{ id: string }> | undefined): string => { const cmp = (a: string, b: string) => (a < b ? -1 : a > b ? 1 : 0) -const partRepairSignature = (part: Part): string => JSON.stringify(part) const syncSnapshotSignature = (value: unknown): string => JSON.stringify(value) -function haveEquivalentPartSnapshots(left: Part[] | undefined, right: Part[]): boolean { - if (!left) { - return right.length === 0 - } - - if (left.length !== right.length) { - return false - } - - for (let index = 0; index < left.length; index += 1) { - const leftPart = left[index] - const rightPart = right[index] - if (!leftPart || !rightPart) { - return false - } - if (leftPart.id !== rightPart.id) { - return false - } - if (partRepairSignature(leftPart) !== partRepairSignature(rightPart)) { - return false - } - } - - return true -} - function haveEquivalentSyncSnapshots(left: unknown, right: unknown): boolean { return syncSnapshotSignature(left) === syncSnapshotSignature(right) } // --------------------------------------------------------------------------- -// Parts-gap recovery — when SSE events arrive but parts are missing, -// trigger a targeted re-fetch for the affected sessions. +// Session materialization scheduler — when local message/part state is incomplete, +// fetch the canonical session snapshot and materialize messages and parts together. // Tracked per-directory, deduplicated, and auto-expiring. // --------------------------------------------------------------------------- -type PendingRepair = { +type PendingSessionMaterialization = { sessionID: string directory: string enqueuedAt: number } -const REPAIR_COOLDOWN_MS = 5_000 -const pendingRepairs = new Map() // key: directory:sessionID +const SESSION_MATERIALIZATION_COOLDOWN_MS = 5_000 +const pendingSessionMaterializations = new Map() // key: directory:sessionID -const repairKey = (directory: string, sessionID: string) => `${directory}:${sessionID}` +const materializationKey = (directory: string, sessionID: string) => `${directory}:${sessionID}` -function enqueuePartsRepair(directory: string, sessionID: string, childStores: ChildStoreManager) { +function enqueueSessionMaterialization(directory: string, sessionID: string, childStores: ChildStoreManager) { if (!directory || directory === "global" || !sessionID) return - const k = repairKey(directory, sessionID) - const existing = pendingRepairs.get(k) - if (existing && Date.now() - existing.enqueuedAt < REPAIR_COOLDOWN_MS) return + const k = materializationKey(directory, sessionID) + const existing = pendingSessionMaterializations.get(k) + if (existing && Date.now() - existing.enqueuedAt < SESSION_MATERIALIZATION_COOLDOWN_MS) return - pendingRepairs.set(k, { sessionID, directory, enqueuedAt: Date.now() }) + pendingSessionMaterializations.set(k, { sessionID, directory, enqueuedAt: Date.now() }) // Defer to next microtask so we don't hold up the current event batch void Promise.resolve().then(async () => { const store = childStores.getChild(directory) if (!store) { - pendingRepairs.delete(k) + pendingSessionMaterializations.delete(k) return } try { - await repairSessionParts(directory, sessionID, store) + await materializeSessionFromServer(directory, sessionID, store) } catch { // Transient failure — next SSE event or reconnect will catch up. } finally { - pendingRepairs.delete(k) + pendingSessionMaterializations.delete(k) } }) } -async function repairSessionParts( +async function materializeSessionFromServer( directory: string, sessionID: string, store: StoreApi, ) { const scopedClient = opencodeClient.getScopedSdkClient(directory) const result = await retry(() => - scopedClient.session.messages({ sessionID, limit: REPAIR_MESSAGE_LIMIT }), + scopedClient.session.messages({ sessionID, limit: SESSION_MATERIALIZATION_MESSAGE_LIMIT }), ) const records = (result.data ?? []).filter((record: { info?: { id?: string } }) => !!record?.info?.id) if (records.length === 0) return store.setState((state: DirectoryStore) => { - const nextMessages = records - .map((record: { info: Message }) => stripMessageDiffSnapshots(record.info)) - .sort((a, b) => cmp(a.id, b.id)) - const currentMessages = state.message[sessionID] ?? [] - const mergedMessages = mergeMessages(currentMessages, nextMessages) - const nextPartState = { ...state.part } - for (const record of records) { - const messageId = record?.info?.id - if (!messageId) continue - const newParts = (record.parts ?? []) - .filter((part: Part) => !!part?.id && !RECONNECT_SKIP_PARTS.has(part.type)) - .sort((a: Part, b: Part) => cmp(a.id, b.id)) - - const existing = nextPartState[messageId] - // Repair when parts are missing, truncated, or stale-but-same-length. - if (!haveEquivalentPartSnapshots(existing, newParts)) { - nextPartState[messageId] = newParts - } - } - return { - message: mergedMessages !== currentMessages ? { ...state.message, [sessionID]: mergedMessages } : state.message, - part: nextPartState, - } + const materialized = materializeSessionSnapshots( + state, + sessionID, + records.map((record: { info: Message; parts?: Part[] }) => ({ + info: stripMessageDiffSnapshots(record.info), + parts: record.parts ?? [], + })), + { skipPartTypes: RECONNECT_SKIP_PARTS }, + ) + return { message: materialized.message, part: materialized.part } }) } @@ -289,7 +249,7 @@ function isRecentBoot() { return bootingRoot || Date.now() - bootedAt < BOOT_DEBOUNCE_MS } -function getViewedSessionRecoveryTarget(directory: string) { +function getViewedSessionMaterializationTarget(directory: string) { if (!_activeDirectory || !_activeSession) return null if (directory !== _activeDirectory) return null return { @@ -582,6 +542,29 @@ const findSessionInChildStores = ( return null } +const childStoreHasSessionState = ( + childStores: ChildStoreManager, + directory: string, + sessionID: string, +): boolean => { + const store = childStores.getChild(directory) + if (!store) return false + const state = store.getState() + return state.session.some((session) => session.id === sessionID) + || Object.prototype.hasOwnProperty.call(state.message, sessionID) + || Object.prototype.hasOwnProperty.call(state.session_status ?? {}, sessionID) +} + +const childStoreHasMessagePartState = ( + childStores: ChildStoreManager, + directory: string, + messageID: string, +): boolean => { + const store = childStores.getChild(directory) + if (!store) return false + return Object.prototype.hasOwnProperty.call(store.getState().part, messageID) +} + const resolveDirectoryFromRoutingIndex = ( routingIndex: EventRoutingIndex, rawDirectory: string, @@ -592,8 +575,13 @@ const resolveDirectoryFromRoutingIndex = ( const sessionID = getSessionIdFromPayload(payload) if (sessionID) { + if (normalizedDirectory && normalizedDirectory !== "global" && childStoreHasSessionState(childStores, normalizedDirectory, sessionID)) { + setIndexedSessionDirectory(routingIndex, sessionID, normalizedDirectory) + return normalizedDirectory + } + const indexedDirectory = routingIndex.sessionDirectoryById.get(sessionID) - if (indexedDirectory) { + if (indexedDirectory && childStores.getChild(indexedDirectory)) { return indexedDirectory } @@ -607,10 +595,14 @@ const resolveDirectoryFromRoutingIndex = ( const messageID = getMessageIdFromPayload(payload) if (messageID) { + if (normalizedDirectory && normalizedDirectory !== "global" && childStoreHasMessagePartState(childStores, normalizedDirectory, messageID)) { + return normalizedDirectory + } + const sessionFromMessage = routingIndex.messageSessionById.get(messageID) if (sessionFromMessage) { const indexedDirectory = routingIndex.sessionDirectoryById.get(sessionFromMessage) - if (indexedDirectory) { + if (indexedDirectory && childStores.getChild(indexedDirectory)) { return indexedDirectory } } @@ -702,12 +694,11 @@ const updateRoutingIndexFromEvent = ( /** * Re-fetch pending questions and permissions for a directory and merge them * into the directory's child store, preserving any in-flight SSE updates that - * arrived while the request was pending. Shared between reconnect resync and - * session-switch resync (the latter is a belt-and-suspenders backstop for any - * code path that drops a `question.asked` / `permission.requested` event — - * directory-eviction rehydration, cross-directory session switches, transport - * fallback gaps, etc.). When `candidateSessionIds` is omitted, every session - * known to the directory store is treated as a candidate. + * arrived while the request was pending. Used by reconnect/materialization + * recovery paths only; normal session switches rely on primary SSE reducer + * state for `question.asked` / `permission.asked` events. When + * `candidateSessionIds` is omitted, every session known to the directory store + * is treated as a candidate. */ export async function resyncBlockingRequestsForDirectory( directory: string, @@ -725,8 +716,8 @@ export async function resyncBlockingRequestsForDirectory( const candidates = candidateSessionIds ?? Array.from(knownSessionIds) if (candidates.length === 0) return - // Re-fetch pending questions — they may have been asked during an SSE gap, - // a directory-eviction window, or a session-switch that bypassed bootstrap. + // Re-fetch pending questions that may have been asked during an SSE gap, + // reconnect window, or directory materialization gap. try { const beforeSignatures = new Map( candidates.map((sessionId) => [sessionId, requestSignature(before.question[sessionId])]), @@ -870,7 +861,7 @@ async function resyncDirectoryAfterReconnect( const current = store.getState() const candidateSessionIds = getReconnectCandidateSessionIds(current, { directory, - viewedSession: getViewedSessionRecoveryTarget(directory), + viewedSession: getViewedSessionMaterializationTarget(directory), }) if (candidateSessionIds.length === 0) return @@ -940,36 +931,25 @@ async function resyncDirectoryAfterReconnect( sessionChanged = true } - // Merge parts: overwrite only messages present in the fetch snapshot. - // Do NOT delete parts for messages that may have been added by SSE - // events arriving between the fetch and the setState — those are more recent. - let nextPartState = state.part - let partsChanged = false - for (const record of records) { - const messageId = record?.info?.id - if (!messageId) continue - const nextParts = (record.parts ?? []) - .filter((part) => !!part?.id && !RECONNECT_SKIP_PARTS.has(part.type)) - .sort((a, b) => cmp(a.id, b.id)) - if (!haveEquivalentPartSnapshots(state.part[messageId], nextParts)) { - if (!partsChanged) { - nextPartState = { ...state.part } - partsChanged = true - } - nextPartState[messageId] = nextParts - } - } - - const mergedMessages = mergeMessages(state.message[sessionId] ?? [], nextMessages) - const messagesChanged = mergedMessages !== (state.message[sessionId] ?? []) + const materialized = materializeSessionSnapshots( + state, + sessionId, + records.map((record) => ({ + info: stripMessageDiffSnapshots(record.info), + parts: record.parts ?? [], + })), + { skipPartTypes: RECONNECT_SKIP_PARTS }, + ) + const messagesChanged = materialized.messagesChanged + const partsChanged = materialized.partsChanged if (!sessionChanged && !messagesChanged && !partsChanged) { return state } return { ...(sessionChanged ? { session: sessions, sessionTotal } : {}), - ...(messagesChanged ? { message: { ...state.message, [sessionId]: mergedMessages } } : {}), - ...(partsChanged ? { part: nextPartState } : {}), + ...(messagesChanged ? { message: materialized.message } : {}), + ...(partsChanged ? { part: materialized.part } : {}), } }) @@ -1146,8 +1126,8 @@ function handleEvent( } } - // Sync-layer parent resync: when a child session goes idle, schedule - // a targeted parts repair for the parent session. This ensures the + // Sync-layer parent resync: when a child session goes idle, recover + // the parent session snapshot. This ensures the // parent's task tool part reflects the child's completion even when // no ToolPart component is mounted. if (payload.type === "session.idle") { @@ -1159,7 +1139,7 @@ function handleEvent( ? (idleSession as Session & { parentID?: string | null }).parentID : null if (parentID) { - enqueuePartsRepair(resolvedDirectory, parentID, childStores) + enqueueSessionMaterialization(resolvedDirectory, parentID, childStores) } } } @@ -1220,24 +1200,28 @@ function handleEvent( break } - if (applyDirectoryEvent(draft, payload, { + const reducerResult = applyDirectoryEvent(draft, payload, { onSetSessionTodo: (sessionID, todos) => { useTodosPersistStore.getState().setSessionTodos(sessionID, todos) }, - })) { + }) + const reducerChanged = typeof reducerResult === "boolean" ? reducerResult : reducerResult.changed + const materializationResult = typeof reducerResult === "boolean" ? undefined : reducerResult.materialization + + if (reducerChanged) { store.setState(draft) const sessionID = getSessionIdFromPayload(payload) ?? undefined const messageID = getMessageIdFromPayload(payload) ?? undefined syncDebug.dispatch.eventApplied(payload.type, sessionID, messageID) - // Parts-gap recovery on message.updated: if the message was inserted or + // Snapshot materialization on message.updated: if the message was inserted or // replaced but draft.part[messageID] is empty, the parts were lost or - // never arrived. Trigger repair so the UI doesn't render a blank bubble. + // never arrived. Recover the session so the UI doesn't render a blank bubble. if (sessionID && messageID && payload.type === "message.updated") { const after = store.getState() const info = (payload.properties as { info: Message }).info if (info.role === "assistant" && (!after.part[messageID] || after.part[messageID].length === 0)) { - enqueuePartsRepair(resolvedDirectory, sessionID, childStores) + enqueueSessionMaterialization(resolvedDirectory, sessionID, childStores) } } } else { @@ -1245,13 +1229,14 @@ function handleEvent( const messageID = getMessageIdFromPayload(payload) ?? undefined syncDebug.dispatch.eventNoChange(payload.type, sessionID, messageID) - // Parts-gap recovery: if a delta event was dropped because the parts array - // was missing or the partID was not found, trigger a repair fetch for the - // session. message.part.updated never needs repair — it only returns false - // for intentionally skipped types (step-start, step-finish, patch) or when - // preserving an existing finished tool part, neither of which indicates missing data. - if (sessionID && messageID && payload.type === "message.part.delta") { - enqueuePartsRepair(resolvedDirectory, sessionID, childStores) + } + + // Snapshot materialization is driven by typed reducer outcomes, not by + // inferring meaning from a generic false/no-change result. + if (materializationResult) { + const materializationSessionID = materializationResult.sessionID ?? getSessionIdFromPayload(payload) ?? undefined + if (materializationSessionID) { + enqueueSessionMaterialization(resolvedDirectory, materializationSessionID, childStores) } } @@ -1395,20 +1380,20 @@ export function SyncProvider(props: { // Event pipeline — created once per mount. No class, no start/stop. // Abort controller owned by the pipeline closure. Cleanup aborts + flushes. useEffect(() => { - const reconnectResyncing = new Set() - const triggerRecoveryResync = (directory: string) => { + const reconnectMaterializing = new Set() + const triggerReconnectMaterialization = (directory: string) => { const store = childStores.children.get(directory) if (!store) return - if (reconnectResyncing.has(directory)) return + if (reconnectMaterializing.has(directory)) return - reconnectResyncing.add(directory) + reconnectMaterializing.add(directory) void resyncDirectoryAfterReconnect(directory, store, routingIndex) .catch(() => { - // Transient failure during resync — next SSE event, transport switch, + // Transient failure during materialization — next SSE event, transport switch, // or reconnect will catch up. }) .finally(() => { - reconnectResyncing.delete(directory) + reconnectMaterializing.delete(directory) }) } @@ -1428,7 +1413,7 @@ export function SyncProvider(props: { connectionPhase: "connected", }) for (const dir of childStores.children.keys()) { - triggerRecoveryResync(dir) + triggerReconnectMaterialization(dir) } }, onDisconnect: (reason) => { @@ -1449,7 +1434,7 @@ export function SyncProvider(props: { connectionPhase: "connected", }) if (_activeDirectory) { - triggerRecoveryResync(_activeDirectory) + triggerReconnectMaterialization(_activeDirectory) } }, }) @@ -1487,46 +1472,6 @@ export function SyncProvider(props: { return unsubscribe }, [props.directory, childStores]) - // Re-fetch pending questions/permissions on session-switch. - // PR #909 only re-fetches on SSE reconnect, leaving an event-drop gap when - // switching sessions within the same socket — the question.asked event may - // have arrived while a different session was active and the directory store - // was evicted, or the user may navigate back to a directory whose store was - // rebuilt after eviction. A 250ms debounce coalesces rapid switches. - useEffect(() => { - let cancelled = false - let timer: ReturnType | null = null - let lastSessionId: string | null = null - let unsub: (() => void) | undefined - void import("./session-ui-store") - .then(({ useSessionUIStore }) => { - if (cancelled) return - lastSessionId = useSessionUIStore.getState().currentSessionId - unsub = useSessionUIStore.subscribe((state) => { - const nextSessionId = state.currentSessionId - if (nextSessionId === lastSessionId) return - lastSessionId = nextSessionId - if (!nextSessionId) return - const sessionDirectory = state.getDirectoryForSession(nextSessionId) - ?? opencodeClient.getDirectory() - ?? props.directory - if (!sessionDirectory) return - if (timer) clearTimeout(timer) - timer = setTimeout(() => { - const currentStore = childStores.getChild(sessionDirectory) - if (!currentStore) return - void resyncBlockingRequestsForDirectory(sessionDirectory, currentStore).catch(() => undefined) - }, 250) - }) - }) - .catch(() => undefined) - return () => { - cancelled = true - if (timer) clearTimeout(timer) - unsub?.() - } - }, [props.directory, childStores]) - return {props.children} } @@ -1997,8 +1942,8 @@ export function useEnsureSessionMessages(sessionID: string, directory?: string) if (!sessionID) return const state = store.getState() - // Already loaded — nothing to do - if (Object.prototype.hasOwnProperty.call(state.message, sessionID)) return + // Already loaded into a renderable message/part snapshot — nothing to do. + if (getSessionMaterializationStatus(state, sessionID).renderable) return // Session doesn't exist — nothing to load if (!state.session.some((s) => s.id === sessionID)) return @@ -2011,38 +1956,7 @@ export function useEnsureSessionMessages(sessionID: string, directory?: string) void (async () => { try { - const scopedClient = opencodeClient.getScopedSdkClient(dir ?? "") - const response = await scopedClient.session.messages({ - sessionID: sessionID, - limit: RECONNECT_MESSAGE_LIMIT, - }) - const records = (response.data ?? []).filter( - (record: { info?: { id?: string } }) => !!record?.info?.id, - ) - if (records.length === 0) return - - const nextMessages = records - .map((record: { info: Message }) => stripMessageDiffSnapshots(record.info)) - .filter((m: Message | null): m is Message => m !== null) - .sort((a: Message, b: Message) => cmp(a.id, b.id)) - - const nextPartState: Record = {} - for (const record of records) { - const messageId = record?.info?.id - if (!messageId) continue - nextPartState[messageId] = (record.parts ?? []) - .filter((part: Part) => !!part?.id && !RECONNECT_SKIP_PARTS.has(part.type)) - .sort((a: Part, b: Part) => cmp(a.id, b.id)) - } - - store.setState((state: DirectoryStore) => { - const currentMessages = state.message[sessionID] ?? [] - const mergedMessages = mergeMessages(currentMessages, nextMessages) - return { - message: mergedMessages !== currentMessages ? { ...state.message, [sessionID]: mergedMessages } : state.message, - part: { ...state.part, ...nextPartState }, - } - }) + await materializeSessionFromServer(dir ?? "", sessionID, store) } catch { // Transient failure — next navigation or reconnect will retry } finally { diff --git a/packages/ui/src/sync/sync-refs.ts b/packages/ui/src/sync/sync-refs.ts index f156abc4..316f2a1a 100644 --- a/packages/ui/src/sync/sync-refs.ts +++ b/packages/ui/src/sync/sync-refs.ts @@ -7,6 +7,7 @@ import type { OpencodeClient } from "@opencode-ai/sdk/v2/client" import type { ChildStoreManager } from "./child-store" +import { getSessionMaterializationStatus } from "./materialization" import type { State } from "./types" let _sdk: OpencodeClient | null = null @@ -83,6 +84,13 @@ export function getSyncMessages(sessionId: string, directory?: string) { return getDirectoryState(directory)?.message[sessionId] ?? [] } +/** Read renderability of a session snapshot from current directory's child store */ +export function getSyncSessionMaterializationStatus(sessionId: string, directory?: string) { + const state = getDirectoryState(directory) + if (!state) return { hasMessages: false, renderable: false, missingPartMessageIDs: [] } + return getSessionMaterializationStatus(state, sessionId) +} + /** Read parts for a message from current directory's child store */ export function getSyncParts(messageId: string, directory?: string) { return getDirectoryState(directory)?.part[messageId] ?? [] diff --git a/packages/ui/src/sync/use-sync.ts b/packages/ui/src/sync/use-sync.ts index f4f04ecb..7ec8ca4e 100644 --- a/packages/ui/src/sync/use-sync.ts +++ b/packages/ui/src/sync/use-sync.ts @@ -6,7 +6,6 @@ import { SESSION_CACHE_LIMIT } from "./types" import { pickSessionCacheEvictions } from "./session-cache" import { mergeOptimisticPage, - mergeMessages, type OptimisticItem, } from "./optimistic" import { useDirectoryStore, useSyncSDK, useSyncDirectory, useChildStoreManager } from "./sync-context" @@ -18,6 +17,7 @@ import { setSessionPrefetch, clearSessionPrefetch, } from "./session-prefetch-cache" +import { getSessionMaterializationStatus, materializeSessionSnapshots } from "./materialization" const SKIP_PARTS = new Set(["patch", "step-start", "step-finish"]) const MESSAGE_PAGE_SIZE = 200 @@ -224,33 +224,19 @@ export function useSync() { } const current = store.getState() - const cached = current.message[sessionID] ?? [] - const messages = options?.mode === "prepend" - ? mergeMessages(cached, merged.session) - : (cached.length > 0 ? mergeMessages(cached, merged.session) : merged.session) + 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" }, + ) - // Build part updates — preserve existing references on prepend to avoid flicker - const isPrepend = options?.mode === "prepend" - let partsChanged = false - const partUpdate: Record = { ...current.part } - for (const p of merged.part) { - if (isPrepend && partUpdate[p.id]) continue // already loaded - const filtered = p.part.filter((x: Part) => !SKIP_PARTS.has(x.type)) - if (filtered.length) { - partUpdate[p.id] = filtered - partsChanged = true - } - } - - const patch: Record = { - message: messages !== cached ? { ...current.message, [sessionID]: messages } : current.message, - } - if (!isPrepend || partsChanged) { - patch.part = partUpdate - } - store.setState(patch) + store.setState({ message: materialized.message, part: materialized.part }) setMetaFor(sessionID, { - limit: messages.length, + limit: materialized.messages.length, cursor: merged.cursor, complete: merged.complete, loading: false, @@ -258,7 +244,7 @@ export function useSync() { setSessionPrefetch({ directory, sessionID, - limit: messages.length, + limit: materialized.messages.length, cursor: merged.cursor, complete: merged.complete, }) @@ -281,7 +267,8 @@ export function useSync() { const current = store.getState() const m = getMetaFor(sessionID) - const cached = current.message[sessionID] !== undefined && m.limit > 0 + const materialization = getSessionMaterializationStatus(current, sessionID) + const cached = materialization.hasMessages && materialization.renderable && m.limit > 0 const hasSession = Binary.search(current.session, sessionID, (s) => s.id).found if (cached && hasSession && !force) return @@ -401,6 +388,7 @@ export function useSync() { return useMemo( () => ({ + ensureSessionRenderable: syncSession, syncSession, loadMore, hasMore,