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

fix: recover mobile and sync state after resume

Browse Source 
Reconnect sync stream when native mobile app resumes
Materialize incomplete sessions with explicit recovery reasons
Add low-noise debug breadcrumb for scoped recovery
This commit is contained in:
Bohdan Triapitsyn
2026-07-01 18:32:24 +03:00
parent 0a69c4ebab
commit 4b1e05160f
6 changed files with 123 additions and 65 deletions
Download Patch File Download Diff File Expand all files Collapse all files
packages/ui/src/apps/MobileApp.tsx
+8
View File
@@ -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' });
packages/ui/src/sync/__tests__/event-reducer.test.ts
+5 -3
View File
@@ -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" },
})
})
packages/ui/src/sync/debug.ts
+7 -1
View File
@@ -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
packages/ui/src/sync/event-reducer.ts
+17 -4
View File
@@ -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<string, unknown>
packages/ui/src/sync/sync-context.tsx
+85 -57
View File
@@ -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<string, PendingSessionMaterialization>() // 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<DirectoryStore>,
options?: { isStale?: () => boolean },
) {
const scopedClient = opencodeClient.getScopedSdkClient(directory)
async function materializeSessionFromServer(
directory: string,
sessionID: string,
store: StoreApi<DirectoryStore>,
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<DirectoryStore>,
routingIndex: EventRoutingIndex,
) {
async function resyncDirectoryAfterReconnect(
directory: string,
store: StoreApi<DirectoryStore>,
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 {
packages/web/server/lib/event-stream/DOCUMENTATION.md
+1
View File
@@ -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.