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

refactor: stabilize live chat sync materialization (#1132)

Browse Source 
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.
This commit is contained in:
Bohdan Triapitsyn
2026-05-07 18:57:44 +03:00
committed by GitHub
parent ff830d3812
commit e892346c6b
17 changed files with 725 additions and 285 deletions
Download Patch File Download Diff File Expand all files Collapse all files
packages/ui/src/components/chat/ChatContainer.tsx
+12 -15
View File
@@ -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 (
<div className="relative flex flex-col h-full bg-background">
{returnToParentButton}
@@ -897,7 +894,7 @@ export const ChatContainer: React.FC = () => {
);
}
if (sessionMessages.length === 0 && !streamingMessageId) {
if (sessionMessages.length === 0 && !sessionIsWorking) {
return (
<div className="relative flex flex-col h-full bg-background transform-gpu">
{returnToParentButton}
packages/ui/src/components/chat/ModelControls.tsx
+8 -7
View File
@@ -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<ModelControlsProps> = ({
const latestLoadedUserChoiceRestoreRef = React.useRef<string | null>(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<ModelControlsProps> = ({
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<ModelControlsProps> = ({
currentAgentName,
contextHydrated,
providers,
hasCurrentSessionMessagesEntry,
hasRenderableCurrentSessionSnapshot,
latestLoadedUserChoice,
setAgent,
tryApplyModelSelection,
@@ -1113,9 +1114,9 @@ export const ModelControls: React.FC<ModelControlsProps> = ({
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<ModelControlsProps> = ({
applyFallbackAgent();
}, [
currentSessionId,
hasCurrentSessionMessagesEntry,
hasRenderableCurrentSessionSnapshot,
latestLoadedUserChoice,
agents,
primaryAgents,
packages/ui/src/components/chat/lib/turns/windowTurns.test.ts
+48
View File
@@ -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');
});
});
packages/ui/src/components/chat/lib/turns/windowTurns.ts
+11 -5
View File
@@ -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;
}
packages/ui/src/components/session/SessionSidebar.tsx
+1 -1
View File
@@ -1087,7 +1087,7 @@ export const SessionSidebar: React.FC<SessionSidebarProps> = ({
currentSessionId,
sortedSessions,
recentSessionIds: recentSessionIdsList,
loadMessages: sync.syncSession,
ensureSessionRenderable: sync.ensureSessionRenderable,
});
const sectionsForSidebarRender = React.useMemo(() => {
packages/ui/src/components/session/sidebar/SessionNodeItem.tsx
+2 -2
View File
@@ -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) {
packages/ui/src/components/session/sidebar/hooks/useSessionPrefetch.ts
+9 -11
View File
@@ -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<unknown>;
ensureSessionRenderable: (sessionId: string) => Promise<unknown>;
};
export const useSessionPrefetch = ({ currentSessionId, sortedSessions, recentSessionIds = [], loadMessages }: Args): void => {
export const useSessionPrefetch = ({ currentSessionId, sortedSessions, recentSessionIds = [], ensureSessionRenderable }: Args): void => {
const sessionPrefetchTimersRef = React.useRef<Map<string, number>>(new Map());
const sessionPrefetchQueueRef = React.useRef<string[]>([]);
const sessionPrefetchInFlightRef = React.useRef<Set<string>>(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;
}
packages/ui/src/sync/__tests__/event-reducer.test.ts
+89
View File
@@ -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> = {}): 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)
})
})
packages/ui/src/sync/__tests__/materialization.test.ts
+137
View File
@@ -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: [],
})
})
})
packages/ui/src/sync/event-pipeline.ts
+9 -1
View File
@@ -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) {
packages/ui/src/sync/event-reducer.ts
+40 -5
View File
@@ -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<string, unknown>
const existingValue = existing[props.field] as string | undefined
packages/ui/src/sync/materialization.ts
+203
View File
@@ -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<string, Message[]>
part: Record<string, Part[]>
}
export type MaterializeSessionSnapshotsOptions = {
skipPartTypes?: ReadonlySet<string>
mode?: "merge" | "prepend"
}
export type MaterializeSessionSnapshotsResult = {
message: Record<string, Message[]>
part: Record<string, Part[]>
messages: Message[]
messagesChanged: boolean
partsChanged: boolean
}
export type SessionMaterializationStatus = {
hasMessages: boolean
renderable: boolean
missingPartMessageIDs: string[]
}
function sortParts(parts: Part[], skipPartTypes: ReadonlySet<string>) {
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<string, unknown>)[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<string, unknown>
mergedRecord[field] = existingValue
}
return merged
}
function mergeMaterializedParts(
existing: Part[] | undefined,
nextParts: Part[],
skipPartTypes: ReadonlySet<string>,
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<string>()
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,
}
}
packages/ui/src/sync/reconnect-recovery.test.ts
+11 -1
View File
@@ -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" },
packages/ui/src/sync/reconnect-recovery.ts
+6 -8
View File
@@ -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<string, SessionStatus>
message?: Record<string, Message[]>
part?: Record<string, Part[]>
}
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<string>()
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)
}
}
packages/ui/src/sync/sync-context.tsx
+115 -201
View File
@@ -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<string, PendingRepair>() // key: directory:sessionID
const SESSION_MATERIALIZATION_COOLDOWN_MS = 5_000
const pendingSessionMaterializations = new Map<string, PendingSessionMaterialization>() // 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<DirectoryStore>,
) {
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<string>()
const triggerRecoveryResync = (directory: string) => {
const reconnectMaterializing = new Set<string>()
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<typeof setTimeout> | 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 <SyncContext.Provider value={system}>{props.children}</SyncContext.Provider>
}
@@ -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<string, Part[]> = {}
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 {
packages/ui/src/sync/sync-refs.ts
+8
View File
@@ -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] ?? []
packages/ui/src/sync/use-sync.ts
+16 -28
View File
@@ -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<string, Part[]> = { ...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<string, unknown> = {
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,