WIP: 自动恢复群聊初始连接

[gstack-context]
Decisions: 群聊初始快照失败使用独立退避恢复；协议不兼容不重试；已连上事件流重连逻辑保持不变
Remaining: 完整测试、重启 8800、真实页面验收
Tried: 初始 503 恢复后群聊端到端用例先失败后通过
Skill: /investigate
[/gstack-context]

src/client/stores/groups.ts

@@ -47,6 +47,8 @@ export const useGroupsStore = defineStore('groups', () => {
   let rebuildPromise: Promise<void> | undefined
   let reconnectTimer: number | undefined
   let reconnectAttempt = 0
+  let recoveryTimer: number | undefined
+  let recoveryAttempt = 0
   let selectionGeneration = 0
   const loadingRoomEvents = new Map<string, GroupSocketEnvelope[]>()
 
@@ -124,6 +126,18 @@ export const useGroupsStore = defineStore('groups', () => {
     }, delay)
   }
 
+  function scheduleSnapshotRecovery(expectedGeneration: number): void {
+    if (recoveryTimer !== undefined || expectedGeneration !== generation || !auth.isAuthenticated) return
+    connectionState.value = 'reconnecting'
+    const delay = Math.min(15_000, 500 * 2 ** Math.min(recoveryAttempt, 5))
+    recoveryAttempt += 1
+    recoveryTimer = window.setTimeout(() => {
+      recoveryTimer = undefined
+      if (expectedGeneration !== generation || !auth.isAuthenticated) return
+      void rebuildSnapshot().catch(() => undefined)
+    }, delay)
+  }
+
   async function rebuildSnapshot(): Promise<void> {
     if (rebuildPromise) return rebuildPromise
     const expectedGeneration = ++generation
@@ -133,6 +147,8 @@ export const useGroupsStore = defineStore('groups', () => {
       socket.close()
       if (reconnectTimer !== undefined) window.clearTimeout(reconnectTimer)
       reconnectTimer = undefined
+      if (recoveryTimer !== undefined) window.clearTimeout(recoveryTimer)
+      recoveryTimer = undefined
       try {
         const anchor = await getGroupCapabilities()
         if (!isSupportedGroupProtocolVersion(anchor.protocolVersion)) {
@@ -155,12 +171,14 @@ export const useGroupsStore = defineStore('groups', () => {
         if (restoreId) await loadRoomSnapshot(restoreId, expectedGeneration)
         if (expectedGeneration !== generation) return
         availability.value = 'available'
+        recoveryAttempt = 0
         await saveCache()
         await connectEvents(anchor, expectedGeneration)
       } catch (cause) {
         if (expectedGeneration !== generation) return
         if (availability.value !== 'unsupported') availability.value = 'unavailable'
         error.value = errorMessage(cause)
+        if (availability.value !== 'unsupported') scheduleSnapshotRecovery(expectedGeneration)
         throw cause
       } finally {
         if (expectedGeneration === generation) isLoading.value = false
@@ -191,6 +209,8 @@ export const useGroupsStore = defineStore('groups', () => {
     socket.close()
     if (reconnectTimer !== undefined) window.clearTimeout(reconnectTimer)
     reconnectTimer = undefined
+    if (recoveryTimer !== undefined) window.clearTimeout(recoveryTimer)
+    recoveryTimer = undefined
     connectionState.value = 'disconnected'
   }
 

tests/e2e/workspace.spec.ts

@@ -159,6 +159,26 @@ test('pins the named Agent typing status above the group composer', async ({ pag
   await expect(typing).toHaveCount(0, { timeout: 4_000 })
 })
 
+test('reconnects the group event stream after an unexpected close', async ({ page }) => {
+  await page.getByRole('button', { name: '群聊' }).click()
+  await expect(page.getByText('已同步', { exact: true })).toBeVisible()
+  const before = await (await page.request.get('http://127.0.0.1:19119/__test/group-connections')).json() as { count: number }
+  await page.request.post('http://127.0.0.1:19119/__test/groups/disconnect')
+  await expect.poll(async () => {
+    const value = await (await page.request.get('http://127.0.0.1:19119/__test/group-connections')).json() as { count: number }
+    return value.count
+  }, { timeout: 5_000 }).toBeGreaterThan(before.count)
+  await expect(page.getByText('已同步', { exact: true })).toBeVisible()
+})
+
+test('recovers the group page after its initial upstream connection fails', async ({ page }) => {
+  await page.request.post('http://127.0.0.1:19119/__test/groups/availability', { data: { available: false } })
+  await page.getByRole('button', { name: '群聊' }).click()
+  await expect(page.getByRole('heading', { name: '群聊服务暂不可用' })).toBeVisible()
+  await page.request.post('http://127.0.0.1:19119/__test/groups/availability', { data: { available: true } })
+  await expect(page.getByRole('heading', { name: '设计与工程协作' })).toBeVisible({ timeout: 5_000 })
+})
+
 test('previews an octet-stream Markdown file from the file library', async ({ page }) => {
   await page.route('**/api/app/files**', async route => {
     const url = new URL(route.request().url())

tests/fixtures/fake-hermes.mjs

@@ -8,6 +8,8 @@ const agentId = '33333333-3333-4333-8333-333333333333'
 const now = () => Date.now() / 1000
 let groupCursor = 0
 const groupClients = new Set()
+let groupConnectionCount = 0
+let groupAvailable = true
 
 function broadcastGroup(event, payload) {
   groupCursor += 1
@@ -74,6 +76,15 @@ const previewPdf = Buffer.from('JVBERi0xLjQKMSAwIG9iago8PC9UeXBlL0NhdGFsb2cvUGFn
 
 const server = createServer(async (request, response) => {
   const url = new URL(request.url || '/', `http://${request.headers.host}`)
+  if (url.pathname === '/__test/group-connections' && request.method === 'GET') return json(response, 200, { count: groupConnectionCount })
+  if (url.pathname === '/__test/groups/availability' && request.method === 'POST') {
+    groupAvailable = (await body(request)).available !== false
+    return json(response, 200, { available: groupAvailable })
+  }
+  if (url.pathname === '/__test/groups/disconnect' && request.method === 'POST') {
+    for (const client of groupClients) client.terminate()
+    return json(response, 200, { ok: true })
+  }
   if (url.pathname === '/__test/rpc-requests' && request.method === 'GET') return json(response, 200, { requests: rpcRequests })
   if (url.pathname === '/__test/rpc-requests/reset' && request.method === 'POST') {
     rpcRequests.length = 0
@@ -88,6 +99,7 @@ const server = createServer(async (request, response) => {
   }
   if (url.pathname === '/auth/logout' && request.method === 'POST') return json(response, 200, { ok: true }, { 'Set-Cookie': 'fake_session=; Path=/; Max-Age=0; HttpOnly; SameSite=Lax' })
   if (!authenticated(request)) return json(response, 401, { detail: 'Unauthorized' })
+  if (url.pathname.startsWith('/api/plugins/yaoyao/v1/') && !groupAvailable) return json(response, 503, { detail: 'Group service temporarily unavailable' })
   if (url.pathname === '/api/auth/me') return json(response, 200, { user_id: 'demo-user', display_name: '验收用户', email: 'demo@example.invalid', provider: 'basic' })
   if (url.pathname === '/api/auth/ws-ticket' && request.method === 'POST') return json(response, 200, { ticket: 'fake-ticket' })
   if (url.pathname === '/api/profiles') return json(response, 200, { profiles })
@@ -155,6 +167,7 @@ server.on('upgrade', (request, socket, head) => {
   if (url.searchParams.get('ticket') !== 'fake-ticket') return socket.destroy()
   sockets.handleUpgrade(request, socket, head, client => {
     if (url.pathname.endsWith('/events')) {
+      groupConnectionCount += 1
       groupClients.add(client)
       client.once('close', () => groupClients.delete(client))
       client.send(JSON.stringify({ type: 'group.ready', epoch, cursor: groupCursor, heartbeatSeconds: 20 }))