fix(sync): resync the active session after reconnect transitions (#1011)
* fix(sync): resync the active session after reconnect transitions Include the viewed session in reconnect recovery and trigger a targeted resync on transport switches so the active chat catches up after missed live events. Constraint: Keep the fix in the sync layer instead of adding ChatContainer-only recovery Rejected: Widen reconnect heuristics for every cached session | broader recovery scope than needed Confidence: high Scope-risk: narrow * Avoid no-op reconnect resync store writes --------- Co-authored-by: Bohdan Triapitsyn <artmore@protonmail.com>
This commit is contained in:
jwcrystal
committed by
GitHub
GitHub
co-authored by
Bohdan Triapitsyn
Bohdan Triapitsyn
parent
e9b117c88b
commit
5c7f5aa4f7
@@ -24,6 +24,7 @@ import { setActionRefs } from "./session-actions"
|
||||
import { setSyncRefs } from "./sync-refs"
|
||||
import { stripMessageDiffSnapshots, stripSessionDiffSnapshots } from "./sanitize"
|
||||
import { syncDebug } from "./debug"
|
||||
import { getReconnectCandidateSessionIds } from "./reconnect-recovery"
|
||||
import { opencodeClient } from "@/lib/opencode/client"
|
||||
import { usePermissionStore } from "@/stores/permissionStore"
|
||||
import { useConfigStore } from "@/stores/useConfigStore"
|
||||
@@ -133,6 +134,7 @@ 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) {
|
||||
@@ -160,6 +162,36 @@ function haveEquivalentPartSnapshots(left: Part[] | undefined, right: Part[]): b
|
||||
return true
|
||||
}
|
||||
|
||||
function haveEquivalentMessageSnapshots(left: Message[] | undefined, right: Message[]): boolean {
|
||||
if (!left) {
|
||||
return right.length === 0
|
||||
}
|
||||
|
||||
if (left.length !== right.length) {
|
||||
return false
|
||||
}
|
||||
|
||||
for (let index = 0; index < left.length; index += 1) {
|
||||
const leftMessage = left[index]
|
||||
const rightMessage = right[index]
|
||||
if (!leftMessage || !rightMessage) {
|
||||
return false
|
||||
}
|
||||
if (leftMessage.id !== rightMessage.id) {
|
||||
return false
|
||||
}
|
||||
if (syncSnapshotSignature(leftMessage) !== syncSnapshotSignature(rightMessage)) {
|
||||
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.
|
||||
@@ -273,40 +305,13 @@ function isRecentBoot() {
|
||||
return bootingRoot || Date.now() - bootedAt < BOOT_DEBOUNCE_MS
|
||||
}
|
||||
|
||||
function getReconnectCandidateSessionIds(state: State) {
|
||||
const ids = new Set<string>()
|
||||
|
||||
for (const [sessionId, status] of Object.entries(state.session_status ?? {})) {
|
||||
if (status && status.type !== "idle") ids.add(sessionId)
|
||||
function getViewedSessionRecoveryTarget(directory: string) {
|
||||
if (!_activeDirectory || !_activeSession) return null
|
||||
if (directory !== _activeDirectory) return null
|
||||
return {
|
||||
directory: _activeDirectory,
|
||||
sessionId: _activeSession,
|
||||
}
|
||||
|
||||
for (const [sessionId, messages] of Object.entries(state.message ?? {})) {
|
||||
const lastMessage = messages[messages.length - 1]
|
||||
if (
|
||||
lastMessage
|
||||
&& lastMessage.role === "assistant"
|
||||
&& typeof (lastMessage as { time?: { completed?: number } }).time?.completed !== "number"
|
||||
) {
|
||||
ids.add(sessionId)
|
||||
}
|
||||
}
|
||||
|
||||
// Ensure parent sessions of any child sessions are also resynced.
|
||||
// A child may have completed during the disconnect gap without the
|
||||
// store knowing (SSE events lost). The parent's task tool part needs
|
||||
// to reflect the child's final state.
|
||||
const parentIds = new Set<string>()
|
||||
for (const session of state.session) {
|
||||
const parentId = (session as Session & { parentID?: string | null }).parentID
|
||||
if (parentId) {
|
||||
parentIds.add(parentId)
|
||||
}
|
||||
}
|
||||
for (const pid of parentIds) {
|
||||
ids.add(pid)
|
||||
}
|
||||
|
||||
return Array.from(ids)
|
||||
}
|
||||
|
||||
function toSessionStatus(status: Awaited<ReturnType<typeof opencodeClient.getSessionStatus>>[string]): SessionStatus | undefined {
|
||||
@@ -716,7 +721,10 @@ async function resyncDirectoryAfterReconnect(
|
||||
routingIndex: EventRoutingIndex,
|
||||
) {
|
||||
const current = store.getState()
|
||||
const candidateSessionIds = getReconnectCandidateSessionIds(current)
|
||||
const candidateSessionIds = getReconnectCandidateSessionIds(current, {
|
||||
directory,
|
||||
viewedSession: getViewedSessionRecoveryTarget(directory),
|
||||
})
|
||||
if (candidateSessionIds.length === 0) return
|
||||
|
||||
const nextStatuses = await opencodeClient.getSessionStatusForDirectory(directory)
|
||||
@@ -730,9 +738,23 @@ async function resyncDirectoryAfterReconnect(
|
||||
}
|
||||
|
||||
if (Object.keys(relevantStatuses).length > 0) {
|
||||
store.setState((state: DirectoryStore) => ({
|
||||
session_status: { ...state.session_status, ...relevantStatuses },
|
||||
}))
|
||||
store.setState((state: DirectoryStore) => {
|
||||
let changed = false
|
||||
for (const [sessionId, nextStatus] of Object.entries(relevantStatuses)) {
|
||||
if (!haveEquivalentSyncSnapshots(state.session_status?.[sessionId], nextStatus)) {
|
||||
changed = true
|
||||
break
|
||||
}
|
||||
}
|
||||
|
||||
if (!changed) {
|
||||
return state
|
||||
}
|
||||
|
||||
return {
|
||||
session_status: { ...state.session_status, ...relevantStatuses },
|
||||
}
|
||||
})
|
||||
}
|
||||
|
||||
const scopedClient = opencodeClient.getScopedSdkClient(directory)
|
||||
@@ -752,17 +774,19 @@ async function resyncDirectoryAfterReconnect(
|
||||
.sort((a, b) => cmp(a.id, b.id))
|
||||
|
||||
store.setState((state: DirectoryStore) => {
|
||||
const sessions = [...state.session]
|
||||
const sessionIndex = sessions.findIndex((item) => item.id === nextSession.id)
|
||||
const sessionIndex = state.session.findIndex((item) => item.id === nextSession.id)
|
||||
let sessions = state.session
|
||||
let sessionChanged = false
|
||||
let sessionTotal = state.sessionTotal
|
||||
|
||||
if (sessionIndex >= 0) {
|
||||
if (sessions[sessionIndex] !== nextSession) {
|
||||
if (!haveEquivalentSyncSnapshots(sessions[sessionIndex], nextSession)) {
|
||||
sessions = [...state.session]
|
||||
sessions[sessionIndex] = nextSession
|
||||
sessionChanged = true
|
||||
}
|
||||
} else {
|
||||
sessions = [...state.session]
|
||||
sessions.push(nextSession)
|
||||
sessions.sort((a, b) => cmp(a.id, b.id))
|
||||
if (!nextSession.parentID) sessionTotal += 1
|
||||
@@ -772,19 +796,32 @@ async function resyncDirectoryAfterReconnect(
|
||||
// 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.
|
||||
const nextPartState = { ...state.part }
|
||||
let nextPartState = state.part
|
||||
let partsChanged = false
|
||||
for (const record of records) {
|
||||
const messageId = record?.info?.id
|
||||
if (!messageId) continue
|
||||
nextPartState[messageId] = (record.parts ?? [])
|
||||
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 messagesChanged = !haveEquivalentMessageSnapshots(state.message[sessionId], nextMessages)
|
||||
if (!sessionChanged && !messagesChanged && !partsChanged) {
|
||||
return state
|
||||
}
|
||||
|
||||
return {
|
||||
...(sessionChanged ? { session: sessions, sessionTotal } : {}),
|
||||
message: { ...state.message, [sessionId]: nextMessages },
|
||||
part: nextPartState,
|
||||
...(messagesChanged ? { message: { ...state.message, [sessionId]: nextMessages } } : {}),
|
||||
...(partsChanged ? { part: nextPartState } : {}),
|
||||
}
|
||||
})
|
||||
|
||||
@@ -1365,6 +1402,21 @@ export function SyncProvider(props: {
|
||||
// Abort controller owned by the pipeline closure. Cleanup aborts + flushes.
|
||||
useEffect(() => {
|
||||
const reconnectResyncing = new Set<string>()
|
||||
const triggerRecoveryResync = (directory: string) => {
|
||||
const store = childStores.children.get(directory)
|
||||
if (!store) return
|
||||
if (reconnectResyncing.has(directory)) return
|
||||
|
||||
reconnectResyncing.add(directory)
|
||||
void resyncDirectoryAfterReconnect(directory, store, routingIndex)
|
||||
.catch(() => {
|
||||
// Transient failure during resync — next SSE event, transport switch,
|
||||
// or reconnect will catch up.
|
||||
})
|
||||
.finally(() => {
|
||||
reconnectResyncing.delete(directory)
|
||||
})
|
||||
}
|
||||
|
||||
const { cleanup } = createEventPipeline({
|
||||
sdk: props.sdk,
|
||||
@@ -1381,18 +1433,8 @@ export function SyncProvider(props: {
|
||||
hasEverConnected: true,
|
||||
connectionPhase: "connected",
|
||||
})
|
||||
for (const [dir, store] of childStores.children) {
|
||||
if (reconnectResyncing.has(dir)) continue
|
||||
if (getReconnectCandidateSessionIds(store.getState()).length === 0) continue
|
||||
|
||||
reconnectResyncing.add(dir)
|
||||
void resyncDirectoryAfterReconnect(dir, store, routingIndex)
|
||||
.catch(() => {
|
||||
// Transient failure during resync — next SSE event or reconnect will catch up.
|
||||
})
|
||||
.finally(() => {
|
||||
reconnectResyncing.delete(dir)
|
||||
})
|
||||
for (const dir of childStores.children.keys()) {
|
||||
triggerRecoveryResync(dir)
|
||||
}
|
||||
},
|
||||
onDisconnect: (reason) => {
|
||||
@@ -1404,14 +1446,17 @@ export function SyncProvider(props: {
|
||||
})
|
||||
},
|
||||
onTransportSwitch: () => {
|
||||
// Transport switched (e.g. WS timeout → SSE fallback) without
|
||||
// actual disconnection. No events lost — just update connection
|
||||
// state without triggering a full directory resync.
|
||||
// Transport switched (e.g. WS timeout → SSE fallback) without a full
|
||||
// disconnect. If the active session missed the transition into a busy
|
||||
// turn, force a targeted resync for the viewed directory.
|
||||
useConfigStore.setState({
|
||||
isConnected: true,
|
||||
hasEverConnected: true,
|
||||
connectionPhase: "connected",
|
||||
})
|
||||
if (_activeDirectory) {
|
||||
triggerRecoveryResync(_activeDirectory)
|
||||
}
|
||||
},
|
||||
})
|
||||
return cleanup
|
||||
|
||||
Reference in New Issue
Block a user