fix: recover chat state after idle reconnects

Resyncs active sessions after hidden upstream stream reconnects
Recovers orphaned streaming parts with active-session snapshots
Adds coverage for event-stream reconnect behavior
This commit is contained in:
Bohdan Triapitsyn
2026-07-01 17:21:19 +03:00
parent 73c9431883
commit b60a794e80
8 changed files with 86 additions and 41 deletions
+1 -1
View File
@@ -46,7 +46,7 @@
"@fontsource/ibm-plex-sans": "^5.1.1",
"@ibm/plex": "^6.4.1",
"@lezer/highlight": "^1.2.3",
"@opencode-ai/sdk": "^1.17.9",
"@opencode-ai/sdk": "^1.17.12",
"@pierre/diffs": "1.3.0-beta.6",
"@replit/codemirror-vim": "^6.3.0",
"@simplewebauthn/browser": "13.3.0",
+59 -23
View File
@@ -874,15 +874,20 @@ const childStoreHasSessionState = (
|| Object.prototype.hasOwnProperty.call(state.session_status ?? {}, sessionID)
}
const childStoreHasMessagePartState = (
childStores: ChildStoreManager,
directory: string,
messageID: string,
): boolean => {
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)
}
return Object.prototype.hasOwnProperty.call(store.getState().part, messageID)
}
const getActiveDirectoryFallback = (childStores: ChildStoreManager): string | null => {
if (!_activeDirectory || !_activeSession) return null
return childStores.getChild(_activeDirectory) ? _activeDirectory : null
}
const resolveDirectoryFromRoutingIndex = (
routingIndex: EventRoutingIndex,
@@ -927,12 +932,21 @@ const resolveDirectoryFromRoutingIndex = (
}
// Scan child stores for a store that has parts for this message
for (const [dir, store] of childStores.children) {
if (Object.prototype.hasOwnProperty.call(store.getState().part, messageID)) {
return dir
}
}
}
for (const [dir, store] of childStores.children) {
if (Object.prototype.hasOwnProperty.call(store.getState().part, messageID)) {
return dir
}
}
// Some reconnect/idle gaps can deliver part events before the matching
// message.updated event and without a sessionID. If the user is actively
// viewing a session, route the orphaned part event there so the reducer can
// trigger HTTP materialization instead of dropping it as a global event.
const activeDirectory = getActiveDirectoryFallback(childStores)
if (activeDirectory) {
return activeDirectory
}
}
// Single-store fallback: if there's only one directory, use it
if (
@@ -946,8 +960,25 @@ const resolveDirectoryFromRoutingIndex = (
}
}
return normalizedDirectory
}
return normalizedDirectory
}
const resolveMaterializationSessionID = (
materializationSessionID: string | undefined,
messageID: string | undefined,
resolvedDirectory: string,
routingIndex: EventRoutingIndex,
): string | undefined => {
if (materializationSessionID) return materializationSessionID
if (messageID) {
const indexedSessionID = routingIndex.messageSessionById.get(messageID)
if (indexedSessionID) return indexedSessionID
}
if (resolvedDirectory && resolvedDirectory === _activeDirectory && _activeSession) {
return _activeSession
}
return undefined
}
const updateRoutingIndexFromEvent = (
routingIndex: EventRoutingIndex,
@@ -1559,14 +1590,19 @@ function handleEvent(
}
// 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)
}
}
// Snapshot materialization is driven by typed reducer outcomes, not by
// inferring meaning from a generic false/no-change result.
if (materializationResult) {
const materializationSessionID = resolveMaterializationSessionID(
materializationResult.sessionID ?? getSessionIdFromPayload(payload) ?? undefined,
materializationResult.messageID ?? getMessageIdFromPayload(payload) ?? undefined,
resolvedDirectory,
routingIndex,
)
if (materializationSessionID) {
enqueueSessionMaterialization(resolvedDirectory, materializationSessionID, childStores)
}
}
updateRoutingIndexFromEvent(routingIndex, resolvedDirectory, payload)
}
+1 -1
View File
@@ -244,7 +244,7 @@
},
"dependencies": {
"@openchamber/ui": "workspace:*",
"@opencode-ai/sdk": "^1.17.9",
"@opencode-ai/sdk": "^1.17.12",
"adm-zip": "^0.5.16",
"jsonc-parser": "^3.3.1",
"react": "^19.1.1",
+1 -1
View File
@@ -25,7 +25,7 @@
"dependencies": {
"@clack/prompts": "^1.1.0",
"@octokit/rest": "^22.0.1",
"@opencode-ai/sdk": "^1.17.9",
"@opencode-ai/sdk": "^1.17.12",
"@simplewebauthn/server": "13.3.1",
"adm-zip": "^0.5.16",
"better-sqlite3": "^12.10.0",
@@ -120,6 +120,17 @@ export function createGlobalMessageStreamWsBridge({
for (const socket of Array.from(clients)) {
if (!readyClients.has(socket)) {
markReady(socket, clientLastEventIds.get(socket) ?? '');
continue;
}
if (status.wasReady) {
const sent = sendMessageStreamWsFrame(socket, {
type: 'ready',
scope: 'global',
});
if (!sent) {
removeClient(socket);
}
}
}
return;
@@ -435,7 +435,7 @@ describe('message stream websocket runtime', () => {
return createSseResponse({
signal: options.signal,
holdOpen: false,
holdOpen: true,
blocks: [
'id: evt-2\ndata: {"type":"server.connected","properties":{}}\n\n',
],
@@ -451,7 +451,7 @@ describe('message stream websocket runtime', () => {
const readyFrames = socket.sent.filter((frame) => frame.type === 'ready');
const eventFrames = socket.sent.filter((frame) => frame.type === 'event' && frame.payload?.type === 'server.connected');
expect(readyFrames).toHaveLength(1);
expect(readyFrames.length).toBeGreaterThanOrEqual(2);
expect(eventFrames.length).toBeGreaterThanOrEqual(2);
expect(fetchCalls.slice(0, 2)).toEqual([null, 'evt-1']);
expect(triggerHealthCheckCalls).toBe(0);