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,