feat: replay session statuses on upstream reconnect and add compact type (#268)

Author: chriswritescode-dev

backend/src/services/sse-aggregator.ts

@@ -37,6 +37,9 @@ interface PendingQuestion {
   [key: string]: unknown
 }
 
+type SessionStatusValue = { type: string } & Record<string, unknown>
+type SessionStatusMap = Record<string, SessionStatusValue>
+
 const OPENCODE_PORT = ENV.OPENCODE.PORT
 const { RECONNECT_DELAY_MS, MAX_RECONNECT_DELAY_MS } = DEFAULTS.SSE
 
@@ -202,6 +205,61 @@ class SSEAggregator {
     await Promise.allSettled(tasks)
   }
 
+  private async replaySessionStatusesForAllClients(): Promise<void> {
+    const fetcher = this.pendingActionsFetcher
+    if (!fetcher) return
+
+    const directories = new Set<string>()
+    this.clients.forEach((client) => {
+      client.directories.forEach(dir => directories.add(dir))
+    })
+
+    if (directories.size === 0) return
+    logger.info(`replay: replaying session statuses for ${directories.size} directory(ies) after upstream reconnect`)
+    await Promise.allSettled(Array.from(directories).map(directory =>
+      this.replaySessionStatusesForDirectory(directory, fetcher)
+    ))
+  }
+
+  private async replaySessionStatusesForDirectory(
+    directory: string,
+    fetcher: PendingActionsFetcher,
+  ): Promise<void> {
+    let statuses: SessionStatusMap
+    try {
+      statuses = await fetcher.getJson<SessionStatusMap>('/session/status', { directory })
+    } catch (error) {
+      logger.warn(`replay: failed to fetch session statuses for ${directory}: ${String(error)}`)
+      return
+    }
+
+    if (!statuses) return
+
+    const previouslyActive = new Set(this.activeSessions.get(directory) ?? [])
+    const nowActive = new Set<string>()
+
+    let replayed = 0
+    for (const [sessionID, status] of Object.entries(statuses)) {
+      if (!sessionID || !status || status.type === 'idle') continue
+      nowActive.add(sessionID)
+      const data = JSON.stringify({ directory, payload: { type: 'session.status', properties: { sessionID, status } } })
+      this.handleUpstreamMessage(data)
+      replayed++
+    }
+
+    let cleared = 0
+    for (const sessionID of previouslyActive) {
+      if (nowActive.has(sessionID)) continue
+      const data = JSON.stringify({ directory, payload: { type: 'session.status', properties: { sessionID, status: { type: 'idle' } } } })
+      this.handleUpstreamMessage(data)
+      cleared++
+    }
+
+    if (replayed > 0 || cleared > 0) {
+      logger.info(`replay: re-emitted ${replayed} active and ${cleared} idle session status(es) for ${directory}`)
+    }
+  }
+
   private async replayPendingActionsForDirectory(
     clientId: string,
     directory: string,
@@ -289,6 +347,7 @@ class SSEAggregator {
       this.everConnected = true
       if (wasConnectedBefore) {
         void this.replayPendingActionsForAllClients()
+        void this.replaySessionStatusesForAllClients()
       }
     }
 

backend/test/services/sse-aggregator.test.ts

@@ -34,13 +34,16 @@ function createCapturingClient() {
   return { callback, writeFrame, events, frames }
 }
 
-function makeFetcher(map: Record<string, { permissions?: unknown[]; questions?: unknown[] }>): PendingActionsFetcher {
+function makeFetcher(
+  map: Record<string, { permissions?: unknown[]; questions?: unknown[]; statuses?: Record<string, { type: string }> }>,
+): PendingActionsFetcher {
   return {
     async getJson<T>(path: string, opts?: { directory?: string }): Promise<T> {
       const directory = opts?.directory ?? ''
       const entry = map[directory] ?? {}
       if (path === '/permission') return (entry.permissions ?? []) as T
       if (path === '/question') return (entry.questions ?? []) as T
+      if (path === '/session/status') return (entry.statuses ?? {}) as T
       throw new Error(`unexpected path: ${path}`)
     },
   }
@@ -197,6 +200,66 @@ describe('SSEAggregator pending replay on connect', () => {
   })
 })
 
+describe('SSEAggregator session status replay on upstream reconnect', () => {
+  beforeEach(() => {
+    sseAggregator.shutdown()
+    sseAggregator.setPendingActionsFetcher(null)
+  })
+
+  it('re-emits active session statuses to subscribed clients', async () => {
+    const fetcher = makeFetcher({
+      '/repo/a': { statuses: { 'sess-1': { type: 'busy' }, 'sess-2': { type: 'idle' } } },
+    })
+    sseAggregator.setPendingActionsFetcher(fetcher)
+
+    const clientA = createCapturingClient()
+    sseAggregator.addClient('status-1', clientA.callback, clientA.writeFrame, ['/repo/a'])
+
+    await (sseAggregator as any).replaySessionStatusesForAllClients()
+    await flushReplay()
+
+    const parsed = clientA.frames.map(f => JSON.parse(f.replace(/^event: message\ndata: /, '').trim()) as {
+      directory: string
+      payload: { type: string; properties: { sessionID: string; status: { type: string } } }
+    })
+
+    const statusEvents = parsed.filter(p => p.payload.type === 'session.status')
+    expect(statusEvents).toHaveLength(1)
+    expect(statusEvents[0]?.payload.properties.sessionID).toBe('sess-1')
+    expect(statusEvents[0]?.payload.properties.status.type).toBe('busy')
+  })
+
+  it('emits idle for sessions that finished during the disconnect', async () => {
+    const clientA = createCapturingClient()
+    sseAggregator.addClient('status-2', clientA.callback, clientA.writeFrame, ['/repo/a'])
+
+    const busy = JSON.stringify({ directory: '/repo/a', payload: { type: 'session.status', properties: { sessionID: 'sess-9', status: { type: 'busy' } } } })
+    ;(sseAggregator as any).handleUpstreamMessage(busy)
+
+    sseAggregator.setPendingActionsFetcher(makeFetcher({ '/repo/a': { statuses: {} } }))
+
+    await (sseAggregator as any).replaySessionStatusesForAllClients()
+    await flushReplay()
+
+    const parsed = clientA.frames.map(f => JSON.parse(f.replace(/^event: message\ndata: /, '').trim()) as {
+      payload: { type: string; properties: { sessionID: string; status: { type: string } } }
+    })
+    const idleEvents = parsed.filter(p => p.payload.type === 'session.status' && p.payload.properties.status.type === 'idle')
+    expect(idleEvents).toHaveLength(1)
+    expect(idleEvents[0]?.payload.properties.sessionID).toBe('sess-9')
+  })
+
+  it('does nothing when no fetcher is configured', async () => {
+    const clientA = createCapturingClient()
+    sseAggregator.addClient('status-3', clientA.callback, clientA.writeFrame, ['/repo/a'])
+
+    await (sseAggregator as any).replaySessionStatusesForAllClients()
+    await flushReplay()
+
+    expect(clientA.frames).toHaveLength(0)
+  })
+})
+
 describe('SSEAggregator directory-indexed broadcast', () => {
   beforeEach(() => {
     sseAggregator.shutdown()

frontend/src/api/types.ts

@@ -182,6 +182,8 @@ export interface SSESessionStatusEvent {
       type: 'idle'
     } | {
       type: 'busy'
+    } | {
+      type: 'compact'
     } | {
       type: 'retry'
       attempt: number

frontend/src/components/message/MessageThread.tsx

@@ -36,8 +36,8 @@ const isMessageStreaming = (msg: Message): boolean => {
   return !('completed' in msg.time && msg.time.completed)
 }
 
-function isSessionInRetry(sessionStatus: { type?: string }): boolean {
-  return sessionStatus?.type === 'retry'
+function isSessionStatusActive(sessionStatus: { type?: string }): boolean {
+  return sessionStatus?.type !== undefined && sessionStatus.type !== 'idle'
 }
 
 const compareMessageIds = (id1: string, id2: string): number => {
@@ -363,7 +363,7 @@ export const MessageThread = memo(function MessageThread({
     return map
   }, [messages])
 
-  const isSessionBusy = !!pendingAssistantId || isSessionInRetry(sessionStatus)
+  const isSessionBusy = !!pendingAssistantId || isSessionStatusActive(sessionStatus)
   const setSessionTodos = useSessionTodos((state) => state.setTodos)
 
   useEffect(() => {