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

fix(sync): preserve pending questions across session switch and directory eviction (closes #918) (#1103)

Browse Source 
* fix(sync): preserve pending questions across session switch and directory eviction

Closes #918, completes the gap left by #909.

The 'agent question disappears after switching session / coming back
later' bug had two root causes that #909 only partially addressed:

1. Directory-eviction TTL (20 min) silently dropped child stores that
   held pending questions/permissions. The discard wasn't gated on
   in-flight blocking-request state, so any 'question.asked' event
   that arrived during the eviction-then-rehydrate window was routed
   to a non-existent store and silently lost.

2. PR #909 re-fetches listPendingQuestions/Permissions only on SSE
   reconnect. Switching sessions within the same socket — including
   navigating back to a directory whose child store was rebuilt after
   eviction — left the UI relying on store state that may have missed
   events that fired while a different session was active.

Three edits, in src/sync:

- eviction.ts / types.ts / child-store.ts: add hasPendingBlockingRequests
  to EvictPlan + DisposeCheck and never evict a directory whose store
  carries a non-empty state.question or state.permission record.
- sync-context.tsx: extract resyncBlockingRequestsForDirectory from the
  reconnect path and call it on currentSessionId changes (debounced
  250ms), reusing PR #909's signature-based merge so concurrent SSE
  updates aren't clobbered.
- __tests__/eviction.test.ts, __tests__/session-switch-resync.test.ts:
  new unit coverage for the eviction guard and resync semantics
  (deduped fetch per switch, in-flight SSE preservation, stale entry
  cleanup, unknown-session filtering).

* fix(sync): refresh store before blocking request resync

---------

Co-authored-by: Alexander Busse <alex@ableph.net>
Co-authored-by: Bohdan Triapitsyn <artmore@protonmail.com>
This commit is contained in:
Alexander Busse
2026-05-05 19:39:35 +03:00
committed by GitHub
co-authored by Alexander Busse Bohdan Triapitsyn
parent a49e2ce1ae
commit 2b0e1ef42e
6 changed files with 533 additions and 157 deletions
Download Patch File Download Diff File Expand all files Collapse all files
packages/ui/src/sync/sync-context.tsx
+204 -155
View File
@@ -690,6 +690,169 @@ const updateRoutingIndexFromEvent = (
}
}
/**
* Re-fetch pending questions and permissions for a directory and merge them
* into the directory's child store, preserving any in-flight SSE updates that
* arrived while the request was pending. Shared between reconnect resync and
* session-switch resync (the latter is a belt-and-suspenders backstop for any
* code path that drops a `question.asked` / `permission.requested` event —
* directory-eviction rehydration, cross-directory session switches, transport
* fallback gaps, etc.). When `candidateSessionIds` is omitted, every session
* known to the directory store is treated as a candidate.
*/
export async function resyncBlockingRequestsForDirectory(
directory: string,
store: StoreApi<DirectoryStore>,
candidateSessionIds?: string[],
) {
const before = store.getState()
const knownSessionIds = new Set<string>([
...before.session.map((session) => session.id),
...Object.keys(before.message ?? {}),
...Object.keys(before.session_status ?? {}),
...Object.keys(before.question ?? {}),
...Object.keys(before.permission ?? {}),
])
const candidates = candidateSessionIds ?? Array.from(knownSessionIds)
if (candidates.length === 0) return
// Re-fetch pending questions — they may have been asked during an SSE gap,
// a directory-eviction window, or a session-switch that bypassed bootstrap.
try {
const beforeSignatures = new Map(
candidates.map((sessionId) => [sessionId, requestSignature(before.question[sessionId])]),
)
const pendingQuestions = await opencodeClient.listPendingQuestions({ directories: [directory] })
const grouped: Record<string, QuestionRequest[]> = {}
for (const q of pendingQuestions) {
if (!q?.id || !q.sessionID) continue
if (!knownSessionIds.has(q.sessionID)) continue
const list = grouped[q.sessionID]
if (list) list.push(q)
else grouped[q.sessionID] = [q]
}
for (const sessionId of Object.keys(grouped)) {
grouped[sessionId].sort((a, b) => (a.id < b.id ? -1 : a.id > b.id ? 1 : 0))
}
for (const [sessionId, questions] of Object.entries(grouped)) {
const knownIds = new Set((before.question[sessionId] ?? []).map((item) => item.id))
const isViewed = isViewedInCurrentSession(directory, sessionId)
if (isViewed) continue
for (const question of questions) {
if (knownIds.has(question.id)) continue
const toastKey = getQuestionToastKey(sessionId, question.id)
if (!toastKey || pendingQuestionToastIds.has(toastKey)) continue
pendingQuestionToastIds.add(toastKey)
const firstQuestion = question.questions?.[0]
const title = firstQuestion?.header?.trim() || "Input needed"
const description = firstQuestion?.question?.trim() || "Agent is waiting for your response"
toast.info(title, {
id: `question-${toastKey}`,
description,
action: {
label: "Open session",
onClick: () => openSessionFromToast(sessionId, directory),
},
})
}
}
store.setState((state: DirectoryStore) => {
const merged = { ...state.question }
for (const [sessionId, questions] of Object.entries(grouped)) {
merged[sessionId] = questions
}
for (const sessionId of candidates) {
if (grouped[sessionId]) continue
const beforeSignature = beforeSignatures.get(sessionId) ?? ""
const currentSignature = requestSignature(state.question[sessionId])
if (currentSignature !== beforeSignature) continue
delete merged[sessionId]
}
return { question: merged }
})
} catch {
// Non-fatal: question resync best-effort
}
// Re-fetch pending permissions — same rationale as questions.
try {
const beforeSignatures = new Map(
candidates.map((sessionId) => [sessionId, requestSignature(before.permission[sessionId])]),
)
const pendingPermissions = await opencodeClient.listPendingPermissions({ directories: [directory] })
const grouped: Record<string, PermissionRequest[]> = {}
for (const permission of pendingPermissions) {
if (!permission?.id || !permission.sessionID) continue
if (!knownSessionIds.has(permission.sessionID)) continue
const list = grouped[permission.sessionID]
if (list) list.push(permission)
else grouped[permission.sessionID] = [permission]
}
for (const sessionId of Object.keys(grouped)) {
grouped[sessionId].sort((a, b) => (a.id < b.id ? -1 : a.id > b.id ? 1 : 0))
}
const permissionStore = usePermissionStore.getState()
const autoAcceptingSessionIds = Object.keys(grouped).filter((sessionId) => permissionStore.isSessionAutoAccepting(sessionId))
if (autoAcceptingSessionIds.length > 0) {
await Promise.all(
autoAcceptingSessionIds.flatMap((sessionId) =>
(grouped[sessionId] ?? []).map((permission) =>
sessionActions.respondToPermission(permission.sessionID, permission.id, "once").catch(() => undefined),
),
),
)
for (const sessionId of autoAcceptingSessionIds) {
delete grouped[sessionId]
}
}
for (const [sessionId, permissions] of Object.entries(grouped)) {
const knownIds = new Set((before.permission[sessionId] ?? []).map((item) => item.id))
const isViewed = isViewedInCurrentSession(directory, sessionId)
if (isViewed) continue
for (const permission of permissions) {
if (knownIds.has(permission.id)) continue
const toastKey = getPermissionToastKey(sessionId, permission.id)
if (!toastKey || pendingPermissionToastIds.has(toastKey)) continue
pendingPermissionToastIds.add(toastKey)
const description = typeof permission.permission === "string" && permission.permission.trim().length > 0
? permission.permission
: "Agent needs your approval"
toast.info("Permission needed", {
id: `permission-${toastKey}`,
description,
action: {
label: "Open session",
onClick: () => openSessionFromToast(sessionId, directory),
},
})
}
}
store.setState((state: DirectoryStore) => {
const merged = { ...state.permission }
for (const [sessionId, permissions] of Object.entries(grouped)) {
merged[sessionId] = permissions
}
for (const sessionId of candidates) {
if (grouped[sessionId]) continue
const beforeSignature = beforeSignatures.get(sessionId) ?? ""
const currentSignature = requestSignature(state.permission[sessionId])
if (currentSignature !== beforeSignature) continue
delete merged[sessionId]
}
return { permission: merged }
})
} catch {
// Non-fatal: permission resync best-effort
}
}
async function resyncDirectoryAfterReconnect(
directory: string,
store: StoreApi<DirectoryStore>,
@@ -805,161 +968,7 @@ async function resyncDirectoryAfterReconnect(
setIndexedSessionMessages(routingIndex, sessionId, directory, nextMessages)
}))
// Re-fetch pending questions on reconnect — they may have been asked
// during the SSE disconnection window and will not arrive via SSE events.
// Overwrite sessions covered by API response, and clear reconnect candidates
// that remain unchanged during the request but are absent from the response.
// If SSE changed a session while the request was in-flight, keep that data.
try {
const before = store.getState()
const knownSessionIds = new Set<string>([
...before.session.map((session) => session.id),
...Object.keys(before.message ?? {}),
...Object.keys(before.session_status ?? {}),
...Object.keys(before.question ?? {}),
...Object.keys(before.permission ?? {}),
])
const beforeSignatures = new Map(
candidateSessionIds.map((sessionId) => [sessionId, requestSignature(before.question[sessionId])]),
)
const pendingQuestions = await opencodeClient.listPendingQuestions({ directories: [directory] })
const grouped: Record<string, QuestionRequest[]> = {}
for (const q of pendingQuestions) {
if (!q?.id || !q.sessionID) continue
if (!knownSessionIds.has(q.sessionID)) continue
const list = grouped[q.sessionID]
if (list) list.push(q)
else grouped[q.sessionID] = [q]
}
// Sort each group by id for binary-search compatibility
for (const sessionId of Object.keys(grouped)) {
grouped[sessionId].sort((a, b) => (a.id < b.id ? -1 : a.id > b.id ? 1 : 0))
}
for (const [sessionId, questions] of Object.entries(grouped)) {
const knownIds = new Set((before.question[sessionId] ?? []).map((item) => item.id))
const isViewed = isViewedInCurrentSession(directory, sessionId)
if (isViewed) continue
for (const question of questions) {
if (knownIds.has(question.id)) continue
const toastKey = getQuestionToastKey(sessionId, question.id)
if (!toastKey || pendingQuestionToastIds.has(toastKey)) continue
pendingQuestionToastIds.add(toastKey)
const firstQuestion = question.questions?.[0]
const title = firstQuestion?.header?.trim() || "Input needed"
const description = firstQuestion?.question?.trim() || "Agent is waiting for your response"
toast.info(title, {
id: `question-${toastKey}`,
description,
action: {
label: "Open session",
onClick: () => openSessionFromToast(sessionId, directory),
},
})
}
}
store.setState((state: DirectoryStore) => {
const merged = { ...state.question }
for (const [sessionId, questions] of Object.entries(grouped)) {
merged[sessionId] = questions
}
for (const sessionId of candidateSessionIds) {
if (grouped[sessionId]) continue
const beforeSignature = beforeSignatures.get(sessionId) ?? ""
const currentSignature = requestSignature(state.question[sessionId])
if (currentSignature !== beforeSignature) continue
delete merged[sessionId]
}
return { question: merged }
})
} catch {
// Non-fatal: question resync best-effort
}
// Re-fetch pending permissions on reconnect — same rationale as questions.
try {
const before = store.getState()
const knownSessionIds = new Set<string>([
...before.session.map((session) => session.id),
...Object.keys(before.message ?? {}),
...Object.keys(before.session_status ?? {}),
...Object.keys(before.question ?? {}),
...Object.keys(before.permission ?? {}),
])
const beforeSignatures = new Map(
candidateSessionIds.map((sessionId) => [sessionId, requestSignature(before.permission[sessionId])]),
)
const pendingPermissions = await opencodeClient.listPendingPermissions({ directories: [directory] })
const grouped: Record<string, PermissionRequest[]> = {}
for (const permission of pendingPermissions) {
if (!permission?.id || !permission.sessionID) continue
if (!knownSessionIds.has(permission.sessionID)) continue
const list = grouped[permission.sessionID]
if (list) list.push(permission)
else grouped[permission.sessionID] = [permission]
}
for (const sessionId of Object.keys(grouped)) {
grouped[sessionId].sort((a, b) => (a.id < b.id ? -1 : a.id > b.id ? 1 : 0))
}
const permissionStore = usePermissionStore.getState()
const autoAcceptingSessionIds = Object.keys(grouped).filter((sessionId) => permissionStore.isSessionAutoAccepting(sessionId))
if (autoAcceptingSessionIds.length > 0) {
await Promise.all(
autoAcceptingSessionIds.flatMap((sessionId) =>
(grouped[sessionId] ?? []).map((permission) =>
sessionActions.respondToPermission(permission.sessionID, permission.id, "once").catch(() => undefined),
),
),
)
for (const sessionId of autoAcceptingSessionIds) {
delete grouped[sessionId]
}
}
for (const [sessionId, permissions] of Object.entries(grouped)) {
const knownIds = new Set((before.permission[sessionId] ?? []).map((item) => item.id))
const isViewed = isViewedInCurrentSession(directory, sessionId)
if (isViewed) continue
for (const permission of permissions) {
if (knownIds.has(permission.id)) continue
const toastKey = getPermissionToastKey(sessionId, permission.id)
if (!toastKey || pendingPermissionToastIds.has(toastKey)) continue
pendingPermissionToastIds.add(toastKey)
const description = typeof permission.permission === "string" && permission.permission.trim().length > 0
? permission.permission
: "Agent needs your approval"
toast.info("Permission needed", {
id: `permission-${toastKey}`,
description,
action: {
label: "Open session",
onClick: () => openSessionFromToast(sessionId, directory),
},
})
}
}
store.setState((state: DirectoryStore) => {
const merged = { ...state.permission }
for (const [sessionId, permissions] of Object.entries(grouped)) {
merged[sessionId] = permissions
}
for (const sessionId of candidateSessionIds) {
if (grouped[sessionId]) continue
const beforeSignature = beforeSignatures.get(sessionId) ?? ""
const currentSignature = requestSignature(state.permission[sessionId])
if (currentSignature !== beforeSignature) continue
delete merged[sessionId]
}
return { permission: merged }
})
} catch {
// Non-fatal: permission resync best-effort
}
await resyncBlockingRequestsForDirectory(directory, store, candidateSessionIds)
ingestDirectoryStateIntoRoutingIndex(routingIndex, directory, store.getState())
}
@@ -1469,6 +1478,46 @@ export function SyncProvider(props: {
return unsubscribe
}, [props.directory, childStores])
// Re-fetch pending questions/permissions on session-switch.
// PR #909 only re-fetches on SSE reconnect, leaving an event-drop gap when
// switching sessions within the same socket — the question.asked event may
// have arrived while a different session was active and the directory store
// was evicted, or the user may navigate back to a directory whose store was
// rebuilt after eviction. A 250ms debounce coalesces rapid switches.
useEffect(() => {
let cancelled = false
let timer: ReturnType<typeof setTimeout> | null = null
let lastSessionId: string | null = null
let unsub: (() => void) | undefined
void import("./session-ui-store")
.then(({ useSessionUIStore }) => {
if (cancelled) return
lastSessionId = useSessionUIStore.getState().currentSessionId
unsub = useSessionUIStore.subscribe((state) => {
const nextSessionId = state.currentSessionId
if (nextSessionId === lastSessionId) return
lastSessionId = nextSessionId
if (!nextSessionId) return
const sessionDirectory = state.getDirectoryForSession(nextSessionId)
?? opencodeClient.getDirectory()
?? props.directory
if (!sessionDirectory) return
if (timer) clearTimeout(timer)
timer = setTimeout(() => {
const currentStore = childStores.getChild(sessionDirectory)
if (!currentStore) return
void resyncBlockingRequestsForDirectory(sessionDirectory, currentStore).catch(() => undefined)
}, 250)
})
})
.catch(() => undefined)
return () => {
cancelled = true
if (timer) clearTimeout(timer)
unsub?.()
}
}, [props.directory, childStores])
return <SyncContext.Provider value={system}>{props.children}</SyncContext.Provider>
}