diff --git a/packages/ui/src/sync/__tests__/event-pipeline.test.js b/packages/ui/src/sync/__tests__/event-pipeline.test.js index 78fc50a0..99dc7649 100644 --- a/packages/ui/src/sync/__tests__/event-pipeline.test.js +++ b/packages/ui/src/sync/__tests__/event-pipeline.test.js @@ -35,6 +35,22 @@ function createSdkWithSingleEvent(event, hold) { }; } +// Helper to create an SDK that yields multiple events in sequence, then holds. +function createSdkWithEvents(events, hold) { + return { + global: { + event: async () => ({ + stream: (async function* () { + for (const event of events) { + yield event; + } + await hold; + })(), + }), + }, + }; +} + describe('createEventPipeline', () => { it('falls back to payload.properties.directory when the SDK event omits top-level directory', async () => { installDomStubs(); @@ -192,4 +208,136 @@ describe('createEventPipeline', () => { expect(received[0].directory).toBe('global'); expect(received[0].payload.type).toBe('server.connected'); }); + + it('delivers message.part.delta events after a coalesced message.part.updated (no stale-delta skip)', async () => { + installDomStubs(); + + let releaseStream; + const hold = new Promise((resolve) => { + releaseStream = resolve; + }); + + const received = []; + + // Simulate: part.updated arrives first, then delta, then part.updated again (coalesces with first). + // After coalescing, the delta should still be delivered — NOT skipped. + const directory = '/test/dir'; + const sdk = createSdkWithEvents([ + // T0: message.part.updated for part-A + { + payload: { + type: 'message.part.updated', + properties: { + directory, + part: { id: 'part-A', type: 'text', messageID: 'msg-1' }, + }, + }, + }, + // T1: message.part.delta for part-A (should flow through even after coalesce) + { + payload: { + type: 'message.part.delta', + properties: { + directory, + messageID: 'msg-1', + partID: 'part-A', + field: 'text', + delta: ' world', + }, + }, + }, + // T2: message.part.updated for part-A — coalesces with T0 + { + payload: { + type: 'message.part.updated', + properties: { + directory, + part: { id: 'part-A', type: 'text', messageID: 'msg-1' }, + }, + }, + }, + ], hold); + + const delivered = new Promise((resolve) => { + const { cleanup } = createEventPipeline({ + sdk, + onEvent: (dir, payload) => { + received.push({ directory: dir, payload }); + if (received.length === 2) { + cleanup(); + releaseStream(); + resolve(); + } + }, + }); + }); + + await delivered; + + // Coalescing means T0 and T2 merge into one event at T0's queue position. + // The delta is a different event type with no coalesce key, so it gets + // its own queue slot. After coalesce: + // - queue[0] = coalesced part.updated (from T2, replacing T0) + // - queue[1] = part.delta (from T1) + // Total: 2 events delivered + expect(received.length).toBe(2); + + // The first event should be the coalesced message.part.updated + expect(received[0].payload.type).toBe('message.part.updated'); + + // The delta MUST be delivered — it should NOT be skipped + expect(received[1].payload.type).toBe('message.part.delta'); + expect(received[1].payload.properties.delta).toBe(' world'); + }); + + it('coalesces message.part.updated events for the same part', async () => { + installDomStubs(); + + let releaseStream; + const hold = new Promise((resolve) => { + releaseStream = resolve; + }); + + const received = []; + const directory = '/test/dir'; + + const sdk = createSdkWithEvents([ + { + payload: { + type: 'message.part.updated', + properties: { + directory, + part: { id: 'part-A', type: 'text', messageID: 'msg-1' }, + }, + }, + }, + { + payload: { + type: 'message.part.updated', + properties: { + directory, + part: { id: 'part-A', type: 'text', messageID: 'msg-1' }, + }, + }, + }, + ], hold); + + const delivered = new Promise((resolve) => { + const { cleanup } = createEventPipeline({ + sdk, + onEvent: (dir, payload) => { + received.push({ directory: dir, payload }); + cleanup(); + releaseStream(); + resolve(); + }, + }); + }); + + await delivered; + + // Only 1 event should be delivered (coalesced) + expect(received.length).toBe(1); + expect(received[0].payload.type).toBe('message.part.updated'); + }); }); diff --git a/packages/ui/src/sync/debug.ts b/packages/ui/src/sync/debug.ts new file mode 100644 index 00000000..f9dd69ba --- /dev/null +++ b/packages/ui/src/sync/debug.ts @@ -0,0 +1,83 @@ +/** + * Sync debug logging — gated behind localStorage flag. + * + * Enable in browser console: + * localStorage.setItem("openchamber:sync:debug", "1") + * + * Disable: + * localStorage.removeItem("openchamber:sync:debug") + * + * All checks are early-returns on the hot path — zero cost when disabled. + */ + +const FLAG_KEY = "openchamber:sync:debug" + +let _enabled: boolean | undefined + +export function isSyncDebugEnabled(): boolean { + if (_enabled !== undefined) return _enabled + try { + _enabled = typeof localStorage !== "undefined" && localStorage.getItem(FLAG_KEY) === "1" + } catch { + _enabled = false + } + return _enabled +} + +/** Force-refresh the flag (call after user toggles localStorage). */ +export function refreshSyncDebugFlag(): void { + _enabled = undefined +} + +type SyncDebugCategory = "pipeline" | "reducer" | "dispatch" + +function log(cat: SyncDebugCategory, ...args: unknown[]): void { + if (!isSyncDebugEnabled()) return + const tag = `%c[sync:${cat}]` + const style = "color: #888" + console.log(tag, style, ...args) +} + +export const syncDebug = { + pipeline: { + /** Event coalesced (replaced an earlier event in the queue). */ + coalesced: (eventType: string, coalesceKey: string) => + log("pipeline", "coalesced", eventType, coalesceKey), + + /** Flush batch dispatched. */ + flush: (count: number) => + log("pipeline", "flush", `${count} events`), + }, + + reducer: { + /** message.updated skipped because role/finish/completed matched existing. */ + messageUpdatedUnchanged: (sessionID: string, messageID: string, role: string, finish: unknown, completed: unknown) => + log("reducer", "message.updated UNCHANGED (skipped)", { sessionID, messageID, role, finish, completed }), + + /** message.part.updated arrived but no parts array exists for this messageID. */ + partUpdatedNoExistingParts: (messageID: string, partID: string, partType: string) => + log("reducer", "message.part.updated NO EXISTING PARTS", { messageID, partID, partType }), + + /** message.part.delta arrived but parts array missing — silently dropped. */ + partDeltaNoParts: (messageID: string, partID: string) => + log("reducer", "message.part.delta DROPPED (no parts array)", { messageID, partID }), + + /** message.part.delta arrived but partID not found in parts array. */ + partDeltaNotFound: (messageID: string, partID: string) => + log("reducer", "message.part.delta DROPPED (partID not found)", { messageID, partID }), + + /** SKIP_PARTS filtered out a part. */ + partSkipped: (messageID: string, partID: string, partType: string) => + log("reducer", "message.part.updated SKIPPED (type filtered)", { messageID, partID, partType }), + }, + + dispatch: { + /** Event dispatched to store but reducer returned false (no state change). */ + eventNoChange: (eventType: string, sessionID?: string, messageID?: string) => + log("dispatch", "event → no change", { eventType, sessionID, messageID }), + + /** Event applied to store successfully. */ + eventApplied: (eventType: string, sessionID?: string, messageID?: string) => + log("dispatch", "event → applied", { eventType, sessionID, messageID }), + }, +} as const \ No newline at end of file diff --git a/packages/ui/src/sync/event-pipeline.ts b/packages/ui/src/sync/event-pipeline.ts index efc32aca..8cb19a53 100644 --- a/packages/ui/src/sync/event-pipeline.ts +++ b/packages/ui/src/sync/event-pipeline.ts @@ -9,6 +9,7 @@ */ import type { Event, OpencodeClient } from "@opencode-ai/sdk/v2/client" +import { syncDebug } from "./debug" // --------------------------------------------------------------------------- // Types @@ -86,13 +87,9 @@ export function createEventPipeline(input: EventPipelineInput) { let queue: QueuedEvent[] = [] let buffer: QueuedEvent[] = [] const coalesced = new Map() - const staleDeltas = new Set() let timer: ReturnType | undefined let last = 0 - const deltaKey = (directory: string, messageID: string, partID: string) => - `${directory}:${messageID}:${partID}` - // Coalesce key — same-type events for the same entity replace earlier ones const key = (directory: string, payload: Event): string | undefined => { if (payload.type === "session.status") { @@ -109,7 +106,7 @@ export function createEventPipeline(input: EventPipelineInput) { return undefined } - // Flush — swap queue, dispatch events, skip stale deltas + // Flush — swap queue, dispatch events const flush = () => { if (timer) clearTimeout(timer) timer = undefined @@ -117,21 +114,16 @@ export function createEventPipeline(input: EventPipelineInput) { if (queue.length === 0) return const events = queue - const skip = staleDeltas.size > 0 ? new Set(staleDeltas) : undefined queue = buffer buffer = events queue.length = 0 coalesced.clear() - staleDeltas.clear() last = Date.now() + syncDebug.pipeline.flush(events.length) // React 18 batches synchronous setState calls automatically, // equivalent to SolidJS batch() for (const event of events) { - if (skip && event.payload.type === "message.part.delta") { - const props = event.payload.properties as { messageID: string; partID: string } - if (skip.has(deltaKey(event.directory, props.messageID, props.partID))) continue - } onEvent(event.directory, event.payload) } @@ -214,10 +206,7 @@ export function createEventPipeline(input: EventPipelineInput) { const i = coalesced.get(k) if (i !== undefined) { queue[i] = { directory, payload: normalizedPayload } - if (normalizedPayload.type === "message.part.updated") { - const part = (normalizedPayload.properties as { part: { messageID: string; id: string } }).part - staleDeltas.add(deltaKey(directory, part.messageID, part.id)) - } + syncDebug.pipeline.coalesced(normalizedPayload.type, k) continue } coalesced.set(k, queue.length) diff --git a/packages/ui/src/sync/event-reducer.ts b/packages/ui/src/sync/event-reducer.ts index 296a6bff..d0d61830 100644 --- a/packages/ui/src/sync/event-reducer.ts +++ b/packages/ui/src/sync/event-reducer.ts @@ -14,6 +14,7 @@ import { Binary } from "./binary" import type { GlobalState, State } from "./types" import { dropSessionCaches } from "./session-cache" import { stripSessionDiffSnapshots } from "./sanitize" +import { syncDebug } from "./debug" const SKIP_PARTS = new Set(["patch", "step-start", "step-finish"]) @@ -160,6 +161,7 @@ export function applyDirectoryEvent( && (existing as { finish?: unknown }).finish === (info as { finish?: unknown }).finish && (existing.time as { completed?: number })?.completed === (info.time as { completed?: number })?.completed if (unchanged) { + syncDebug.reducer.messageUpdatedUnchanged(info.sessionID, info.id, info.role, (info as { finish?: unknown }).finish, (info.time as { completed?: number })?.completed) return false } const next = [...messages] @@ -190,10 +192,14 @@ export function applyDirectoryEvent( case "message.part.updated": { const part = (event.properties as { part: Part }).part - if (SKIP_PARTS.has(part.type)) return false + if (SKIP_PARTS.has(part.type)) { + syncDebug.reducer.partSkipped((part as { messageID: string }).messageID, part.id, part.type) + return false + } const messageID = (part as { messageID: string }).messageID const parts = draft.part[messageID] if (!parts) { + syncDebug.reducer.partUpdatedNoExistingParts(messageID, part.id, part.type) draft.part[messageID] = [part] return true } @@ -246,9 +252,15 @@ export function applyDirectoryEvent( delta: string } const parts = draft.part[props.messageID] - if (!parts) return false + if (!parts) { + syncDebug.reducer.partDeltaNoParts(props.messageID, props.partID) + return false + } const result = Binary.search(parts, props.partID, (p) => p.id) - if (!result.found) return false + if (!result.found) { + syncDebug.reducer.partDeltaNotFound(props.messageID, props.partID) + return false + } const existing = parts[result.index] as Record const existingValue = existing[props.field] as string | undefined // Create new Part object + new array so React detects the change diff --git a/packages/ui/src/sync/sync-context.tsx b/packages/ui/src/sync/sync-context.tsx index b64f042d..44897deb 100644 --- a/packages/ui/src/sync/sync-context.tsx +++ b/packages/ui/src/sync/sync-context.tsx @@ -15,6 +15,7 @@ import { updateStreamingState } from "./streaming" import { setActionRefs } from "./session-actions" import { setSyncRefs } from "./sync-refs" import { stripMessageDiffSnapshots, stripSessionDiffSnapshots } from "./sanitize" +import { syncDebug } from "./debug" import { opencodeClient } from "@/lib/opencode/client" import { usePermissionStore } from "@/stores/permissionStore" import { autoRespondsPermission, normalizeDirectory } from "@/stores/utils/permissionAutoAccept" @@ -100,6 +101,79 @@ const RECONNECT_SKIP_PARTS = new Set(["patch", "step-start", "step-finish"]) const cmp = (a: string, b: string) => (a < b ? -1 : a > b ? 1 : 0) +// --------------------------------------------------------------------------- +// Parts-gap recovery — when SSE events arrive but parts are missing, +// trigger a targeted re-fetch for the affected sessions. +// Tracked per-directory, deduplicated, and auto-expiring. +// --------------------------------------------------------------------------- + +type PendingRepair = { + sessionID: string + directory: string + enqueuedAt: number +} + +const REPAIR_COOLDOWN_MS = 5_000 +const pendingRepairs = new Map() // key: directory:sessionID + +const repairKey = (directory: string, sessionID: string) => `${directory}:${sessionID}` + +function enqueuePartsRepair(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 + + pendingRepairs.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) + return + } + try { + await repairSessionParts(directory, sessionID, store) + } catch { + // Transient failure — next SSE event or reconnect will catch up. + } finally { + pendingRepairs.delete(k) + } + }) +} + +async function repairSessionParts( + directory: string, + sessionID: string, + store: StoreApi, +) { + const scopedClient = opencodeClient.getScopedSdkClient(directory) + const result = await retry(() => + scopedClient.session.messages({ sessionID, limit: RECONNECT_MESSAGE_LIMIT }), + ) + const records = (result.data ?? []).filter((record: { info?: { id?: string } }) => !!record?.info?.id) + if (records.length === 0) return + + store.setState((state: DirectoryStore) => { + 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] + // Only patch if parts were missing or fewer than server has + if (!existing || existing.length < newParts.length) { + nextPartState[messageId] = newParts + } + } + return { part: nextPartState } + }) +} + // Module-level refs for notification viewed check. // Used to determine if user is currently viewing the session when a notification arrives. let _activeDirectory = "" @@ -804,6 +878,33 @@ function handleEvent( if (applyDirectoryEvent(draft, payload)) { 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 + // 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. + 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) + } + } + } else { + const sessionID = getSessionIdFromPayload(payload) ?? undefined + const messageID = getMessageIdFromPayload(payload) ?? undefined + syncDebug.dispatch.eventNoChange(payload.type, sessionID, messageID) + + // Parts-gap recovery: if a part event was dropped because the parts array + // was missing (message not yet inserted or parts lost), trigger a repair + // fetch for the session. + if (sessionID && messageID && ( + payload.type === "message.part.delta" || payload.type === "message.part.updated" + )) { + enqueuePartsRepair(resolvedDirectory, sessionID, childStores) + } } updateRoutingIndexFromEvent(routingIndex, resolvedDirectory, payload)