From 4b1e05160f53e9a94d4740a10f5c8942c6e1ce84 Mon Sep 17 00:00:00 2001 From: Bohdan Triapitsyn Date: Wed, 1 Jul 2026 18:32:16 +0300 Subject: [PATCH] fix: recover mobile and sync state after resume Reconnect sync stream when native mobile app resumes Materialize incomplete sessions with explicit recovery reasons Add low-noise debug breadcrumb for scoped recovery --- packages/ui/src/apps/MobileApp.tsx | 8 + .../src/sync/__tests__/event-reducer.test.ts | 8 +- packages/ui/src/sync/debug.ts | 8 +- packages/ui/src/sync/event-reducer.ts | 21 ++- packages/ui/src/sync/sync-context.tsx | 142 +++++++++++------- .../server/lib/event-stream/DOCUMENTATION.md | 1 + 6 files changed, 123 insertions(+), 65 deletions(-) diff --git a/packages/ui/src/apps/MobileApp.tsx b/packages/ui/src/apps/MobileApp.tsx index 756e564c..aeeb2a49 100644 --- a/packages/ui/src/apps/MobileApp.tsx +++ b/packages/ui/src/apps/MobileApp.tsx @@ -282,6 +282,8 @@ const mobileInputKeyboardProps = { spellCheck: false, } as const; +const NATIVE_RESUME_SYNC_EVENT_THROTTLE_MS = 1_000; + const getProjectLabel = (path: string): string => { const normalized = normalizePath(path); if (!normalized) return ''; @@ -1946,9 +1948,15 @@ export function MobileApp({ apis }: MobileAppProps) { // exhausted the attempt (then the connect screen shows). const [autoConnectPhase, setAutoConnectPhase] = React.useState<'pending' | 'attempting' | 'done'>('pending'); const isNativeMobileApp = React.useMemo(() => isCapacitorMobileApp(), []); + const lastNativeResumeSyncEventAtRef = React.useRef(0); const handleNativeResume = React.useCallback(() => { if (!getRuntimeApiBaseUrl()) return; + const now = Date.now(); + if (now - lastNativeResumeSyncEventAtRef.current >= NATIVE_RESUME_SYNC_EVENT_THROTTLE_MS) { + lastNativeResumeSyncEventAtRef.current = now; + window.dispatchEvent(new Event('openchamber:system-resume')); + } void initializeApp(); void refreshGitHubAuthStatus(apis.github, { force: true }); if (providersCount === 0) void loadProviders({ source: 'mobileApp:nativeResume' }); diff --git a/packages/ui/src/sync/__tests__/event-reducer.test.ts b/packages/ui/src/sync/__tests__/event-reducer.test.ts index 0c624563..bfeb8438 100644 --- a/packages/ui/src/sync/__tests__/event-reducer.test.ts +++ b/packages/ui/src/sync/__tests__/event-reducer.test.ts @@ -61,7 +61,7 @@ describe("applyDirectoryEvent", () => { expect(result).toEqual({ changed: false, - materialization: { type: "incomplete-session-snapshot", messageID: "msg_1", partID: "prt_1" }, + materialization: { type: "incomplete-session-snapshot", reason: "orphan-delta", messageID: "msg_1", partID: "prt_1" }, }) }) @@ -73,7 +73,7 @@ describe("applyDirectoryEvent", () => { expect(result).toEqual({ changed: false, - materialization: { type: "incomplete-session-snapshot", messageID: "msg_1", partID: "prt_1" }, + materialization: { type: "incomplete-session-snapshot", reason: "missing-delta-part", messageID: "msg_1", partID: "prt_1" }, }) }) @@ -86,6 +86,7 @@ describe("applyDirectoryEvent", () => { changed: true, materialization: { type: "incomplete-session-snapshot", + reason: "missing-owning-message", sessionID: "ses_1", messageID: "msg_1", partID: "prt_1", @@ -102,6 +103,7 @@ describe("applyDirectoryEvent", () => { changed: true, materialization: { type: "incomplete-session-snapshot", + reason: "missing-owning-message", sessionID: "ses_1", messageID: "msg_1", partID: "prt_1", @@ -123,7 +125,7 @@ describe("applyDirectoryEvent", () => { expect(result).toEqual({ changed: false, - materialization: { type: "incomplete-session-snapshot", sessionID: "ses_1", messageID: "msg_1", partID: "prt_1" }, + materialization: { type: "incomplete-session-snapshot", reason: "orphan-delta", sessionID: "ses_1", messageID: "msg_1", partID: "prt_1" }, }) }) diff --git a/packages/ui/src/sync/debug.ts b/packages/ui/src/sync/debug.ts index b792b445..d6b1aad1 100644 --- a/packages/ui/src/sync/debug.ts +++ b/packages/ui/src/sync/debug.ts @@ -23,7 +23,7 @@ function isSyncDebugEnabled(): boolean { } return _enabled } -type SyncDebugCategory = "pipeline" | "reducer" | "dispatch" +type SyncDebugCategory = "pipeline" | "reducer" | "dispatch" | "recovery" function log(cat: SyncDebugCategory, ...args: unknown[]): void { if (!isSyncDebugEnabled()) return @@ -74,4 +74,10 @@ export const syncDebug = { eventApplied: (eventType: string, sessionID?: string, messageID?: string) => log("dispatch", "event → applied", { eventType, sessionID, messageID }), }, + + recovery: { + /** A scoped session snapshot fetch is starting because live state looked incomplete. */ + materializing: (details: { reason: string; directory: string; sessionID: string; messageID?: string; partID?: string }) => + log("recovery", "materializing session", details), + }, } as const diff --git a/packages/ui/src/sync/event-reducer.ts b/packages/ui/src/sync/event-reducer.ts index c925b3d0..b9120b03 100644 --- a/packages/ui/src/sync/event-reducer.ts +++ b/packages/ui/src/sync/event-reducer.ts @@ -151,10 +151,23 @@ export type GlobalEventResult = { project: Project } | null +export type SessionMaterializationReason = + | "missing-owning-message" + | "orphan-delta" + | "missing-delta-part" + | "empty-assistant-message" + | "child-session-idle" + | "child-session-discovered" + | "ensure-session-messages" + | "stream-reconnect" + | "transport-switch" + | "stale-status-resync" + export type DirectoryEventResult = boolean | { changed: boolean materialization: { type: "incomplete-session-snapshot" + reason: SessionMaterializationReason sessionID?: string messageID: string partID?: string @@ -356,7 +369,7 @@ export function applyDirectoryEvent( return missingOwningMessage ? { changed: true, - materialization: { type: "incomplete-session-snapshot", sessionID, messageID, partID: part.id }, + materialization: { type: "incomplete-session-snapshot", reason: "missing-owning-message", sessionID, messageID, partID: part.id }, } : true } @@ -390,7 +403,7 @@ export function applyDirectoryEvent( return missingOwningMessage ? { changed: true, - materialization: { type: "incomplete-session-snapshot", sessionID, messageID, partID: part.id }, + materialization: { type: "incomplete-session-snapshot", reason: "missing-owning-message", sessionID, messageID, partID: part.id }, } : true } @@ -426,7 +439,7 @@ export function applyDirectoryEvent( syncDebug.reducer.partDeltaNoParts(props.messageID, props.partID) return { changed: false, - materialization: { type: "incomplete-session-snapshot", sessionID: props.sessionID, messageID: props.messageID, partID: props.partID }, + materialization: { type: "incomplete-session-snapshot", reason: "orphan-delta", sessionID: props.sessionID, messageID: props.messageID, partID: props.partID }, } } const result = Binary.search(parts, props.partID, (p) => p.id) @@ -434,7 +447,7 @@ export function applyDirectoryEvent( syncDebug.reducer.partDeltaNotFound(props.messageID, props.partID) return { changed: false, - materialization: { type: "incomplete-session-snapshot", sessionID: props.sessionID, messageID: props.messageID, partID: props.partID }, + materialization: { type: "incomplete-session-snapshot", reason: "missing-delta-part", sessionID: props.sessionID, messageID: props.messageID, partID: props.partID }, } } const existing = parts[result.index] as Record diff --git a/packages/ui/src/sync/sync-context.tsx b/packages/ui/src/sync/sync-context.tsx index 423ecf77..4eab7650 100644 --- a/packages/ui/src/sync/sync-context.tsx +++ b/packages/ui/src/sync/sync-context.tsx @@ -9,7 +9,7 @@ import { createEventPipeline } from "./event-pipeline" import { isVSCodeRuntime } from "@/lib/desktop" import { isMobileSurfaceRuntime } from "@/lib/runtimeSurface" import { isCapacitorApp } from "@/lib/platform" -import { reduceGlobalEvent, applyGlobalProject, applyDirectoryEvent } from "./event-reducer" +import { reduceGlobalEvent, applyGlobalProject, applyDirectoryEvent, type SessionMaterializationReason } from "./event-reducer" import { useGlobalSyncStore } from "./global-sync-store" import { ChildStoreManager, type DirectoryStore } from "./child-store" import { @@ -194,24 +194,36 @@ function haveEquivalentSyncSnapshots(left: unknown, right: unknown): boolean { // Tracked per-directory, deduplicated, and auto-expiring. // --------------------------------------------------------------------------- -type PendingSessionMaterialization = { - sessionID: string - directory: string - enqueuedAt: number -} +type PendingSessionMaterialization = { + sessionID: string + directory: string + enqueuedAt: number + request: SessionMaterializationRequest +} + +type SessionMaterializationRequest = { + reason: SessionMaterializationReason + messageID?: string + partID?: string +} const SESSION_MATERIALIZATION_COOLDOWN_MS = 5_000 const pendingSessionMaterializations = new Map() // key: directory:sessionID const materializationKey = (directory: string, sessionID: string) => `${directory}:${sessionID}` -function enqueueSessionMaterialization(directory: string, sessionID: string, childStores: ChildStoreManager) { - if (!directory || directory === "global" || !sessionID) return - const k = materializationKey(directory, sessionID) - const existing = pendingSessionMaterializations.get(k) - if (existing && Date.now() - existing.enqueuedAt < SESSION_MATERIALIZATION_COOLDOWN_MS) return - - pendingSessionMaterializations.set(k, { sessionID, directory, enqueuedAt: Date.now() }) +function enqueueSessionMaterialization( + directory: string, + sessionID: string, + childStores: ChildStoreManager, + request: SessionMaterializationRequest, +) { + if (!directory || directory === "global" || !sessionID) return + const k = materializationKey(directory, sessionID) + const existing = pendingSessionMaterializations.get(k) + if (existing && Date.now() - existing.enqueuedAt < SESSION_MATERIALIZATION_COOLDOWN_MS) return + + pendingSessionMaterializations.set(k, { sessionID, directory, enqueuedAt: Date.now(), request }) // Defer to next microtask so we don't hold up the current event batch void Promise.resolve().then(async () => { @@ -220,8 +232,8 @@ function enqueueSessionMaterialization(directory: string, sessionID: string, chi pendingSessionMaterializations.delete(k) return } - try { - await materializeSessionFromServer(directory, sessionID, store) + try { + await materializeSessionFromServer(directory, sessionID, store, request) } catch { // Transient failure — next SSE event or reconnect will catch up. } finally { @@ -230,13 +242,20 @@ function enqueueSessionMaterialization(directory: string, sessionID: string, chi }) } -async function materializeSessionFromServer( - directory: string, - sessionID: string, - store: StoreApi, - options?: { isStale?: () => boolean }, -) { - const scopedClient = opencodeClient.getScopedSdkClient(directory) +async function materializeSessionFromServer( + directory: string, + sessionID: string, + store: StoreApi, + options?: SessionMaterializationRequest & { isStale?: () => boolean }, +) { + syncDebug.recovery.materializing({ + reason: options?.reason ?? "ensure-session-messages", + directory, + sessionID, + messageID: options?.messageID, + partID: options?.partID, + }) + const scopedClient = opencodeClient.getScopedSdkClient(directory) const result = await retry(async () => { const response = await scopedClient.session.messages({ sessionID, limit: SESSION_MATERIALIZATION_MESSAGE_LIMIT }) assertSdkSuccess(response, "session.messages") @@ -1216,29 +1235,31 @@ export async function resyncBlockingRequestsForDirectory( } } -async function resyncDirectoryAfterReconnect( - directory: string, - store: StoreApi, - routingIndex: EventRoutingIndex, -) { +async function resyncDirectoryAfterReconnect( + directory: string, + store: StoreApi, + routingIndex: EventRoutingIndex, + reason: SessionMaterializationReason, +) { const current = store.getState() const candidateSessionIds = getActiveSessionCandidateIds(directory, current) if (candidateSessionIds.length === 0) return await resyncDirectorySessionStatuses(directory, store, candidateSessionIds, "authoritative") - const scopedClient = opencodeClient.getScopedSdkClient(directory) - await Promise.all(candidateSessionIds.map(async (sessionId) => { - const [sessionResponse, messageResponse] = await Promise.all([ - retry(async () => { - const response = await scopedClient.session.get({ sessionID: sessionId }) + const scopedClient = opencodeClient.getScopedSdkClient(directory) + await Promise.all(candidateSessionIds.map(async (sessionId) => { + syncDebug.recovery.materializing({ reason, directory, sessionID: sessionId }) + const [sessionResponse, messageResponse] = await Promise.all([ + retry(async () => { + const response = await scopedClient.session.get({ sessionID: sessionId }) assertSdkSuccess(response, "session.get") return response - }).catch(() => null), - retry(async () => { - const response = await scopedClient.session.messages({ sessionID: sessionId, limit: RECONNECT_MESSAGE_LIMIT }) - assertSdkSuccess(response, "session.messages") - return response + }).catch(() => null), + retry(async () => { + const response = await scopedClient.session.messages({ sessionID: sessionId, limit: RECONNECT_MESSAGE_LIMIT }) + assertSdkSuccess(response, "session.messages") + return response }).catch(() => null), ]) const session = sessionResponse?.data @@ -1497,9 +1518,9 @@ function handleEvent( const parentID = idleSession ? (idleSession as Session & { parentID?: string | null }).parentID : null - if (parentID) { - enqueueSessionMaterialization(resolvedDirectory, parentID, childStores) - } + if (parentID) { + enqueueSessionMaterialization(resolvedDirectory, parentID, childStores, { reason: "child-session-idle" }) + } } } @@ -1578,10 +1599,13 @@ function handleEvent( // 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)) { - enqueueSessionMaterialization(resolvedDirectory, sessionID, childStores) - } + const info = (payload.properties as { info: Message }).info + if (info.role === "assistant" && (!after.part[messageID] || after.part[messageID].length === 0)) { + enqueueSessionMaterialization(resolvedDirectory, sessionID, childStores, { + reason: "empty-assistant-message", + messageID, + }) + } } } else { const sessionID = getSessionIdFromPayload(payload) ?? undefined @@ -1600,7 +1624,11 @@ function handleEvent( routingIndex, ) if (materializationSessionID) { - enqueueSessionMaterialization(resolvedDirectory, materializationSessionID, childStores) + enqueueSessionMaterialization(resolvedDirectory, materializationSessionID, childStores, { + reason: materializationResult.reason, + messageID: materializationResult.messageID, + partID: materializationResult.partID, + }) } } @@ -1650,7 +1678,7 @@ export function SyncProvider(props: { [childStores, props.sdk, props.directory], ) - const triggerDirectoryResync = useCallback((directory: string) => { + const triggerDirectoryResync = useCallback((directory: string, reason: SessionMaterializationReason) => { const store = childStores.children.get(directory) if (!store) return const resyncing = resyncingDirectoriesRef.current @@ -1658,7 +1686,7 @@ export function SyncProvider(props: { lastFullResyncAtByDirectoryRef.current.set(directory, Date.now()) resyncing.add(directory) - void resyncDirectoryAfterReconnect(directory, store, routingIndex) + void resyncDirectoryAfterReconnect(directory, store, routingIndex, reason) .catch(() => { // Transient failure — the watchdog, next SSE event, or reconnect will catch up. }) @@ -1845,7 +1873,7 @@ export function SyncProvider(props: { return } for (const dir of childStores.children.keys()) { - triggerDirectoryResync(dir) + triggerDirectoryResync(dir, "stream-reconnect") } }, onDisconnect: (reason) => { @@ -1865,7 +1893,7 @@ export function SyncProvider(props: { connectionPhase: "connected", }) for (const dir of childStores.children.keys()) { - triggerDirectoryResync(dir) + triggerDirectoryResync(dir, "transport-switch") } }, }) @@ -1919,11 +1947,11 @@ export function SyncProvider(props: { ) return { session: sessions, limit: Math.max(sessions.length, 50) } }) - // Trigger parent session materialization so the task tool part - // state (metadata, sessionId, output) is refreshed. - for (const pid of parentIdsForMaterialization) { - enqueueSessionMaterialization(directory, pid, childStores) - } + // Trigger parent session materialization so the task tool part + // state (metadata, sessionId, output) is refreshed. + for (const pid of parentIdsForMaterialization) { + enqueueSessionMaterialization(directory, pid, childStores, { reason: "child-session-discovered" }) + } } catch { // Best-effort — next tick will retry. } @@ -1945,7 +1973,7 @@ export function SyncProvider(props: { needsSnapshotAfterStatusPoll(before, sessionId, statuses[sessionId]) )) if (needsSnapshot) { - triggerDirectoryResync(directory) + triggerDirectoryResync(directory, "stale-status-resync") } } finally { polling.delete(directory) @@ -1977,7 +2005,7 @@ export function SyncProvider(props: { const lastFullResyncAt = lastFullResyncAtByDirectoryRef.current.get(directory) ?? 0 if (shouldTriggerStaleResync(lastStreamActivityAtRef.current, lastFullResyncAt, now)) { pipelineReconnectRef.current?.("active_stream_stale") - triggerDirectoryResync(directory) + triggerDirectoryResync(directory, "stale-status-resync") } // Discover child sessions created by other OpenCode instances @@ -2676,7 +2704,7 @@ export function useEnsureSessionMessages(sessionID: string, directory?: string) void (async () => { try { - await materializeSessionFromServer(resolvedDirectory, sessionID, store, { isStale }) + await materializeSessionFromServer(resolvedDirectory, sessionID, store, { reason: "ensure-session-messages", isStale }) } catch { // Transient failure — next navigation or reconnect will retry } finally { diff --git a/packages/web/server/lib/event-stream/DOCUMENTATION.md b/packages/web/server/lib/event-stream/DOCUMENTATION.md index f69938c6..2acc7263 100644 --- a/packages/web/server/lib/event-stream/DOCUMENTATION.md +++ b/packages/web/server/lib/event-stream/DOCUMENTATION.md @@ -42,6 +42,7 @@ This module contains the OpenChamber message-stream WebSocket protocol and runti - The global hub keeps a bounded replay buffer keyed by SSE `eventId` so reconnecting browser clients can receive buffered events after their requested `Last-Event-ID`. - Directory WS clients still attach one upstream `/event?directory=...` SSE reader per connection because directory streams are scoped. - If an upstream SSE stream stalls after the browser WS is already ready, the reader aborts that upstream fetch and reconnects upstream with `Last-Event-ID`, keeping the browser WS alive when recovery is fast. +- When the shared global upstream reconnects after it was previously ready, the global WS bridge sends a fresh `ready` frame to already-ready browser clients. The browser treats this as a reconnect edge and can run scoped state repair without requiring the browser WS to close. - Health checks are reserved for initial upstream connect failures and explicit upstream-unavailable responses, not for ordinary stall recovery on an already-established stream. - Global synthetic events such as `openchamber:session-status`, `openchamber:session-activity`, `openchamber:notification`, and `openchamber:heartbeat` are preserved on the WS path, but heartbeat frames are emitted only while an upstream SSE stream is actively attached. - Global UI broadcasts are fan-out capable across both SSE and WS clients.