修复离线重连、事件补发和 Hermes 工具授权恢复

src/client/api/chatTranscript.ts

@@ -196,14 +196,13 @@ export class ChatTranscriptClient {
         value.epoch !== current.epoch ||
         typeof cursor !== 'number' ||
         !Number.isSafeInteger(cursor) ||
-        cursor < current.cursor ||
         typeof value.running !== 'boolean' ||
         typeof value.queued !== 'boolean'
       ) throw new TranscriptGap()
-      const nextCursor = cursor
+      if (cursor < current.cursor) return
+      if (cursor > current.cursor) throw new TranscriptGap()
       await this.commit({
         ...current,
-        cursor: nextCursor,
         running: value.running,
         queued: value.queued,
         pendingApproval: value.pendingApproval ?? null,

src/client/api/httpRealtime.ts

@@ -42,13 +42,15 @@ export class HTTPRealtimeChannel {
   private async readLoop(onReady: () => void): Promise<void> {
     let attempt = 0
     while (!this.stopped) {
-      this.abort = new AbortController()
+      const attemptAbort = new AbortController()
+      this.abort = attemptAbort
       let reader: ReadableStreamDefaultReader<Uint8Array> | undefined
       let watchdog: number | undefined
       try {
+        watchdog = window.setTimeout(() => attemptAbort.abort(), 45_000)
         const headers: Record<string, string> = { Accept: 'text/event-stream' }
         if (this.cursor) headers['Last-Event-ID'] = this.cursor
-        const response = await fetch(`/api/realtime/channels/${this.id}/events`, { credentials: 'same-origin', headers, signal: this.abort.signal, cache: 'no-store' })
+        const response = await fetch(`/api/realtime/channels/${this.id}/events`, { credentials: 'same-origin', headers, signal: attemptAbort.signal, cache: 'no-store' })
         if ([401, 403, 404, 409, 428].includes(response.status)) throw new ApiError('实时连接需要重新同步', response.status, 'REALTIME_RESET_REQUIRED')
         if (!response.ok) throw new Error(`SSE HTTP ${response.status}`)
         if (!response.headers.get('content-type')?.startsWith('text/event-stream') || !response.body) throw new ApiError('无效的事件流响应', 502, 'INVALID_EVENT_STREAM')
@@ -56,7 +58,7 @@ export class HTTPRealtimeChannel {
         const parser = new SSEParser(), decoder = new TextDecoder()
         while (!this.stopped) {
           window.clearTimeout(watchdog)
-          watchdog = window.setTimeout(() => this.abort?.abort(), 45_000)
+          watchdog = window.setTimeout(() => attemptAbort.abort(), 45_000)
           const { value, done } = await reader.read()
           if (done) throw new Error('SSE disconnected')
           attempt = 0

src/client/views/ConversationsView.vue

@@ -299,7 +299,7 @@ async function refresh() {
   conversations.value = transcriptStore.conversations
   if (!cursor) { cursor = c.cursor; transcriptStore.cursor = c.cursor }
 }
-async function load(id = selected.value, append = false) {
+async function load(id = selected.value, append = false, signal?: AbortSignal) {
   if (!id) {
     generation++
     active.value = undefined
@@ -311,6 +311,9 @@ async function load(id = selected.value, append = false) {
     return
   }
   const own = ++generation
+  const owner = auth.user?.id, epoch = accountEpoch
+  const current = () => !disposed && own === generation && !signal?.aborted
+    && auth.user?.id === owner && accountEpoch === epoch
   const previousComposer = composerKey.value
   const previousFiles = composer.value?.filesSnapshot() ?? []
   const previousQuote = quoted.value
@@ -324,11 +327,11 @@ async function load(id = selected.value, append = false) {
     activeTask.value = null;tasks.value = [];assignments.value = [];quoted.value = null;older.value = false
   }
   try {
-    const incoming = cached ?? await apiRequest<WorkspaceDetail>(`/api/app/conversations/${id}?limit=50${requestedTask ? `&taskId=${encodeURIComponent(requestedTask)}` : ''}`)
-    if (disposed || own !== generation) return
+    const incoming = cached ?? await apiRequest<WorkspaceDetail>(`/api/app/conversations/${id}?limit=50${requestedTask ? `&taskId=${encodeURIComponent(requestedTask)}` : ''}`, { signal })
+    if (!current()) return
     const r = cached ?? transcriptStore.finishRead(readToken, incoming)
     if (cached && transcriptStore.staleDetails.has(transcriptKey(id, cached.task?.id))) {
-      queueMicrotask(() => { if (!disposed && active.value?.id === id) void load(id, true) })
+      queueMicrotask(() => { if (current() && active.value?.id === id) void load(id, true, signal) })
     }
     const atBottom = timeline.value?.isFollowingBottom() ?? true
     active.value = r.conversation
@@ -350,28 +353,31 @@ async function load(id = selected.value, append = false) {
         } catch { /* Leave the original draft intact when browser storage is unavailable. */ }
       }
       await nextTick()
+      if (!current()) return
       const saved = taskFiles.get(composerKey.value)
       if (saved?.length) await composer.value?.attachFiles(saved)
+      if (!current()) return
       quoted.value = taskQuotes.get(composerKey.value) ?? null
     }
     if (!append) older.value = r.hasOlder ?? r.messages.length === 50
     if (!append || atBottom) {
       await nextTick()
+      if (!current()) return
       if (!append) timeline.value?.scrollToBottom('auto')
-      void markRead().catch(e => { if (own === generation) error.value = e instanceof Error ? e.message : '同步已读失败' })
+      void markRead().catch(e => { if (current()) error.value = e instanceof Error ? e.message : '同步已读失败' })
     }
   } catch (e) {
-    if (own === generation && requestedTask && e instanceof ApiError && e.status === 404) {
+    if (current() && requestedTask && e instanceof ApiError && e.status === 404) {
       try {
-        const current = await apiRequest<{tasks:WorkspaceTask[]}>(`/api/app/conversations/${id}/tasks`)
-        const fallback = current.tasks[0]
-        if (own === generation && fallback && fallback.id !== requestedTask) {
+        const available = await apiRequest<{tasks:WorkspaceTask[]}>(`/api/app/conversations/${id}/tasks`, { signal })
+        const fallback = available.tasks[0]
+        if (current() && fallback && fallback.id !== requestedTask) {
           await router.replace({query:{...route.query,taskId:fallback.id}})
           return
         }
       } catch { /* A deleted conversation still reports its original error. */ }
     }
-    if (own === generation) error.value = e instanceof Error ? e.message : '加载失败'
+    if (current()) error.value = e instanceof Error ? e.message : '加载失败'
   } finally {
     transcriptStore.cancelRead(readToken)
     if (own === generation) loading.value = false
@@ -470,6 +476,12 @@ let patchEvents = false
 let eventSource: EventSource | undefined
 let eventFrame: number | undefined
 let eventQueue: WorkspaceEvent[] = []
+let recoveryAbort: AbortController | undefined
+function cancelRecovery() {
+  recoveryAbort?.abort(); recoveryAbort = undefined
+  if (timer) clearTimeout(timer)
+  timer = undefined
+}
 function presentDetail(detail: WorkspaceDetail) {
   const atBottom = timeline.value?.isFollowingBottom() ?? true
   active.value = detail.conversation
@@ -501,17 +513,17 @@ function flushEvents() {
     }
   }
 }
-async function hydrateWorkspace() {
+async function hydrateWorkspace(signal: AbortSignal) {
   const owner = auth.user?.id, epoch = accountEpoch
-  const capabilities = await apiRequest<{ features: string[]; csrfToken?: string }>('/api/app/capabilities')
-  if (disposed || accountEpoch !== epoch || auth.user?.id !== owner) return
+  const capabilities = await apiRequest<{ features: string[]; csrfToken?: string }>('/api/app/capabilities', { signal })
+  if (signal.aborted || disposed || accountEpoch !== epoch || auth.user?.id !== owner) return
   patchEvents = capabilities.features.includes(WORKSPACE_PATCH_CAPABILITY)
   knowledgeEnabled.value = capabilities.features.includes('bot-file-memory-v1')
   if (capabilities.csrfToken) setApiCsrfToken(capabilities.csrfToken)
   const readToken = transcriptStore.beginRead()
   try {
-  const snapshot = await apiRequest<WorkspaceSnapshot & { serverIdentity?: ServerIdentity; projects?: WorkspaceProject[] }>('/api/app/workspace/snapshot')
-  if (disposed || accountEpoch !== epoch || auth.user?.id !== owner) return
+  const snapshot = await apiRequest<WorkspaceSnapshot & { serverIdentity?: ServerIdentity; projects?: WorkspaceProject[] }>('/api/app/workspace/snapshot', { signal })
+  if (signal.aborted || disposed || accountEpoch !== epoch || auth.user?.id !== owner) return
   transcriptStore.hydrate(snapshot)
   agents.value = snapshot.agents; conversations.value = transcriptStore.conversations; cursor = snapshot.cursor
   projects.value = snapshot.projects ?? []
@@ -536,7 +548,7 @@ function connectEvents() {
     flushEvents(); error.value = ''
     publishServerIdentity(JSON.parse((event as MessageEvent).data).serverIdentity)
   })
-  source.addEventListener('reset', () => { source.close(); void recoverEvents() })
+  source.addEventListener('reset', () => { if (!disposed && eventSource === source) { source.close(); void recoverEvents() } })
   source.onerror = () => {
     if (disposed || eventSource !== source) return
     error.value = '连接中断，正在重连'
@@ -549,34 +561,41 @@ function connectEvents() {
   }
 }
 function suspendEvents() {
+  cancelRecovery()
   eventSource?.close(); eventSource = undefined
-  if (timer) clearTimeout(timer)
   error.value = '网络已断开，恢复连接后将自动同步'
 }
 function resumeEvents() {
   if (!disposed && auth.user?.id) void recoverEvents()
 }
-async function recoverEvents() {
+async function recoverEvents(initial = false) {
+  cancelRecovery()
+  if (disposed || !auth.user?.id) return
   if (!navigator.onLine) { suspendEvents(); return }
+  const controller = new AbortController(), owner = auth.user.id, epoch = accountEpoch
+  recoveryAbort = controller
+  const current = () => !disposed && !controller.signal.aborted && recoveryAbort === controller
+    && accountEpoch === epoch && auth.user?.id === owner
   eventSource?.close(); eventSource = undefined; eventQueue = []
   if (eventFrame !== undefined) cancelAnimationFrame(eventFrame)
   eventFrame = undefined
   try {
-    await hydrateWorkspace()
-    const detail = selected.value ? transcriptStore.get(selected.value, selectedTask.value || activeTask.value?.id) : undefined
+    await hydrateWorkspace(controller.signal)
+    if (!current()) return
+    const detail = !initial && selected.value ? transcriptStore.get(selected.value, selectedTask.value || activeTask.value?.id) : undefined
     if (detail) {
       presentDetail(detail)
-      if (transcriptStore.staleDetails.has(transcriptKey(detail.conversation.id, detail.task?.id))) await load(detail.conversation.id, true)
-    } else await load()
-    connectEvents()
+      if (transcriptStore.staleDetails.has(transcriptKey(detail.conversation.id, detail.task?.id))) await load(detail.conversation.id, true, controller.signal)
+    } else await load(selected.value, false, controller.signal)
+    if (current()) connectEvents()
   }
   catch (cause) {
-    if (disposed) return
+    if (!current()) return
     if (!navigator.onLine) { suspendEvents(); return }
     error.value = cause instanceof Error ? cause.message : '连接中断'
     if (cause instanceof ApiError && [401, 403].includes(cause.status)) return
     timer = setTimeout(() => void recoverEvents(), 1500)
-  }
+  } finally { if (recoveryAbort === controller) recoveryAbort = undefined }
 }
 async function select(id: string) {
   error.value = ''
@@ -929,7 +948,7 @@ watch(() => auth.user?.id, (owner, previous) => {
   knowledgePanel.value?.close(); knowledgeEnabled.value = false; projects.value = []; knowledgeRevision.value++
   accountEpoch++; busy.value = false
   generation++
-  if (timer) clearTimeout(timer)
+  cancelRecovery()
   text.value = ''; files.value = []; mentions.value = []; quoted.value = null
   pendingRequestId = undefined; pendingFingerprint = ''; uploadedSources = []; uploadedReferences = []
   eventSource?.close(); eventSource = undefined; eventQueue = []
@@ -946,13 +965,7 @@ onMounted(async () => {
   window.addEventListener('offline', suspendEvents)
   window.addEventListener('online', resumeEvents)
   void auth.refreshProfileAvatars().catch(() => undefined)
-  try {
-    await hydrateWorkspace()
-    if (!disposed) connectEvents()
-    await load()
-  } catch (e) {
-    error.value = String(e)
-  }
+  await recoverEvents(true)
 })
 onBeforeUnmount(() => {
   window.removeEventListener('offline', suspendEvents)
@@ -963,7 +976,7 @@ onBeforeUnmount(() => {
   eventSource?.close()
   if (eventFrame !== undefined) cancelAnimationFrame(eventFrame)
   generation++
-  if (timer) clearTimeout(timer)
+  cancelRecovery()
 })
 </script>
 <template>

src/server/chatTranscriptApi.ts

@@ -34,7 +34,7 @@ export function chatTranscriptRouter(cache:ChatCacheCoordinator,auth:LocalAuthSt
   router.get('/api/app/chat/sessions/:id/events',ctx=>{
     const {owner,profile,id}=scope(ctx),version=auth.pushAuthorizationVersion(owner)
     const raw=ctx.get('last-event-id')||String(ctx.query.after??''),parts=raw.split(':')
-    let cursor=Number(parts.at(-1)),closed=false,lastFlush=0
+    let cursor=Number(parts.at(-1)),closed=false,lastFlush=0,ready=false
     transcripts.seed(owner,profile,id)
     const epoch=transcripts.epochFor(owner,profile,id)
     if(parts.length!==2||parts[0]!==epoch||!Number.isSafeInteger(cursor)||cursor<0||cursor>transcripts.cursor(owner,profile,id))throw new HttpError(409,'聊天事件需要重新同步','transcript_reset')
@@ -47,14 +47,19 @@ export function chatTranscriptRouter(cache:ChatCacheCoordinator,auth:LocalAuthSt
       if(res.writableLength+Buffer.byteLength(value)>4*1024*1024){res.write('event: reset\ndata: {}\n\n');close();return false}
       res.write(value);return true
     }
+    const control=()=>({epoch,cursor,...transcripts.control(owner,profile,id)})
     const flush=()=>{
       clearTimeout(timer);timer=undefined
+      if(closed)return
       if(!valid()){close();return}
       if(transcripts.epochFor(owner,profile,id)!==epoch){write('event: reset\ndata: {}\n\n');close();return}
       const events=transcripts.events(owner,profile,id,cursor)
       for(const event of events){if(event.epoch!==epoch){write('event: reset\ndata: {}\n\n');close();return}if(!write(`id: ${epoch}:${event.cursor}\nevent: transcript\ndata: ${JSON.stringify(event)}\n\n`))return;cursor=event.cursor}
       if(events.length)lastFlush=Date.now()
-      if(events.length===250)timer=setTimeout(flush,0)
+      if(events.length===250){timer=setTimeout(flush,0);return}
+      // A control frame must never acknowledge events still waiting in another page.
+      if(cursor!==transcripts.cursor(owner,profile,id)){write('event: reset\ndata: {}\n\n');close();return}
+      if(!ready){ready=true;write(`event: ready\ndata: ${JSON.stringify(control())}\n\n`)}
     }
     const changed=()=>{
       if(closed)return
@@ -67,12 +72,13 @@ export function chatTranscriptRouter(cache:ChatCacheCoordinator,auth:LocalAuthSt
     res.writeHead(200,{'Content-Type':'text/event-stream; charset=utf-8','Cache-Control':'no-store','X-Accel-Buffering':'no'})
     res.flushHeaders();res.once('close',close);res.once('error',close)
     off=transcripts.subscribe(changed)
-    const control=()=>({epoch,cursor:transcripts.cursor(owner,profile,id),...transcripts.control(owner,profile,id)})
-    flush();write(`event: ready\ndata: ${JSON.stringify(control())}\n\n`)
+    flush()
+    if(closed)return
     heartbeat=setInterval(()=>{
       if(!valid()){close();return}
       if(transcripts.epochFor(owner,profile,id)!==epoch){write('event: reset\ndata: {}\n\n');close();return}
-      write(`event: state\ndata: ${JSON.stringify(control())}\n\n`)
+      flush()
+      if(!closed&&ready&&timer===undefined)write(`event: state\ndata: ${JSON.stringify(control())}\n\n`)
     },5_000);heartbeat.unref()
   })
   return router

src/server/realtimeBroker.ts

@@ -330,14 +330,43 @@ export class RealtimeBroker {
     }
     return u
   }
+  private authorized(p: RealtimePrincipal): boolean {
+    try { if (!p.valid()) return false; p.authorize?.('chat'); return true } catch { return false }
+  }
+  private connectionPrincipal(u: Upstream): RealtimePrincipal | undefined {
+    if (this.authorized(u.principal)) return u.principal
+    // A service transport may outlive its first browser session and serve other users.
+    const candidates = [
+      ...[...this.channels.values()].filter(c => c.principal.upstreamKey === u.principal.upstreamKey).map(c => c.principal),
+      ...[...u.routes.values()].flatMap(r => [...r.observers.values()]),
+    ]
+    const principal = candidates.find(p => this.authorized(p))
+    if (principal) u.principal = principal
+    return principal
+  }
+  private scheduleReconnect(u: Upstream): void {
+    if (this.closed || this.upstreams.get(u.principal.upstreamKey) !== u || u.retry || !this.connectionPrincipal(u)) return
+    u.retry = setTimeout(() => {
+      u.retry = undefined
+      void this.connect(u).catch(() => {})
+    }, Math.min(15_000, 500 * 2 ** Math.min(u.attempts++, 5)))
+    u.retry.unref()
+  }
   private connect(u: Upstream): Promise<void> {
+    if (this.closed || this.upstreams.get(u.principal.upstreamKey) !== u) return Promise.reject(new Error('Broker transport closed'))
     if (u.connecting) return u.connecting
     if (u.socket?.readyState === WebSocket.OPEN && u.ready) return Promise.resolve()
+    clearTimeout(u.retry); u.retry = undefined
     u.connecting = (async () => {
-      const url = await u.principal.url('chat')
-      if (this.closed) throw new Error('Broker closed')
+      // Let the previous socket's recovery unwind before starting another generation.
+      await u.recovery?.catch(() => {})
+      const principal = this.connectionPrincipal(u)
+      if (!principal) throw new HttpError(401, 'Authentication expired', 'authentication_required')
+      const url = await principal.url('chat')
+      if (this.closed || this.upstreams.get(principal.upstreamKey) !== u) throw new Error('Broker transport closed')
+      if (!this.authorized(principal)) throw new HttpError(401, 'Authentication expired', 'authentication_required')
       await new Promise<void>((resolve, reject) => {
-        const ws = new WebSocket(url, { agent: u.principal.agent, headers: { Origin: `${url.protocol === 'wss:' ? 'https:' : 'http:'}//${url.host}` }, maxPayload: CHAT_MAX_PAYLOAD, handshakeTimeout: 15_000 })
+        const ws = new WebSocket(url, { agent: principal.agent, headers: { Origin: `${url.protocol === 'wss:' ? 'https:' : 'http:'}//${url.host}` }, maxPayload: CHAT_MAX_PAYLOAD, handshakeTimeout: 15_000 })
         u.socket = ws
         const timeout = setTimeout(() => { reject(new Error('Upstream ready timeout')); ws.terminate() }, 20_000)
         let ready = false
@@ -346,12 +375,20 @@ export class RealtimeBroker {
           let f: Frame
           try { f = JSON.parse(data.toString()) } catch { ws.terminate(); return }
           if (f.method === 'event' && f.params?.type === 'gateway.ready') {
-            clearTimeout(timeout); ready = true; u.ready = f; u.attempts = 0
+            if (ready) return
+            clearTimeout(timeout); ready = true; u.ready = f
             const epoch = f.params.payload?.replay_epoch
             const changed = u.epoch !== undefined && epoch !== u.epoch
             u.epoch = epoch
-            if (u.routes.size) u.recovery = this.recover(u, changed).finally(() => { u.recovery = undefined })
-            resolve()
+            const recovery = u.routes.size ? this.recover(u, changed) : Promise.resolve()
+            u.recovery = recovery
+            void recovery.then(() => {
+              if (this.closed || u.socket !== ws || ws.readyState !== WebSocket.OPEN) throw new Error('Upstream disconnected during recovery')
+              u.attempts = 0
+              resolve()
+            }).catch(error => { reject(error); ws.terminate() }).finally(() => {
+              if (u.recovery === recovery) u.recovery = undefined
+            })
           } else if (f.id !== undefined) {
             const pending = u.pending.get(String(f.id))
             if (pending) { clearTimeout(pending.timer); u.pending.delete(String(f.id)); pending.resolve(f) }
@@ -359,23 +396,24 @@ export class RealtimeBroker {
         })
         let alive = true
         ws.on('pong', () => { alive = true })
-        u.ping = setInterval(() => { if (!alive) ws.terminate(); else if (ws.readyState === WebSocket.OPEN) { alive = false; ws.ping() } }, 25_000)
-        u.ping.unref()
+        const ping = setInterval(() => { if (!alive) ws.terminate(); else if (ws.readyState === WebSocket.OPEN) { alive = false; ws.ping() } }, 25_000)
+        u.ping = ping; ping.unref()
         ws.on('error', () => { if (!ready) reject(new Error('Upstream connection failed')) })
         ws.on('close', () => {
-          clearTimeout(timeout); clearInterval(u.ping)
+          clearTimeout(timeout); clearInterval(ping)
           if (u.socket !== ws) return
           u.socket = undefined; u.ready = undefined
-          if (!ready) reject(new Error('Upstream disconnected'))
+          reject(new Error('Upstream disconnected'))
           for (const pending of u.pending.values()) { clearTimeout(pending.timer); pending.reject(new Error('Upstream disconnected; result unknown')) }
           u.pending.clear()
-          if (!this.closed && this.upstreams.get(u.principal.upstreamKey) === u) {
-            u.retry = setTimeout(() => { u.retry = undefined; void this.connect(u).catch(() => {}) }, Math.min(15_000, 500 * 2 ** Math.min(u.attempts++, 5)))
-            u.retry.unref()
-          }
+          this.scheduleReconnect(u)
         })
       })
-    })().finally(() => { u.connecting = undefined })
+    })().catch(error => {
+      // Credentials and readiness can fail before any WebSocket close callback exists.
+      this.scheduleReconnect(u)
+      throw error
+    }).finally(() => { u.connecting = undefined })
     return u.connecting
   }
   private rpc(u: Upstream, method: string, params: Frame, observed?: (frame: Frame) => void, sent?: (frame: Frame) => void): Promise<Frame> {
@@ -496,6 +534,23 @@ export class RealtimeBroker {
       for (const r of u.routes.values()) {
         const lastSeen = r.seq
         const resumed = await this.rpc(u, 'session.resume', { session_id: r.stored, profile: r.profile, close_on_disconnect: false, omit_messages: true })
+        if (resumed.error?.code === 4007 || resumed.error?.code === 'session_not_found') {
+          // Hermes' session.resume uses 4007 for a missing stored session. Retire
+          // only that route; a deleted session must not reconnect the whole service forever.
+          this.reset(u, 'upstream_session_missing')
+          this.event(u, { method: 'event', params: { type: 'error', session_id: r.runtime,
+            payload: { code: 'session_not_found', message: 'Hermes 中的原会话已不存在，无法恢复。' } } }, true)
+          r.active = false
+          const key = this.routeKey(r.profile, r.stored)
+          u.routes.delete(key); u.byRuntime.delete(r.runtime)
+          this.routeOwners.delete(JSON.stringify([u.principal.instanceKey ?? u.principal.upstreamKey, r.profile, r.stored]))
+          for (const c of this.channels.values()) if (c.principal.upstreamKey === u.principal.upstreamKey) c.routes.delete(key)
+          for (const [id, runtime] of this.interactions) if (id.startsWith(`${u.principal.upstreamKey}:`) && runtime === r.runtime) this.interactions.delete(id)
+          continue
+        }
+        if (resumed.error || typeof resumed.result?.session_id !== 'string' || !resumed.result.session_id) {
+          throw new Error('Upstream session recovery failed')
+        }
         r.workingDirectory = undefined
         r.cwd = typeof resumed.result?.info?.cwd === 'string' ? resumed.result.info.cwd : undefined
         if (typeof resumed.result?.running === 'boolean') r.active = resumed.result.running
@@ -506,7 +561,7 @@ export class RealtimeBroker {
               snapshot?resumed.result:{...resumed.result,inflight:undefined})
           }
         }
-        if (!resumed.result || resumed.result.session_id !== r.runtime || changed) {
+        if (resumed.result.session_id !== r.runtime || changed) {
           if (resumed.result?.session_id) { u.byRuntime.delete(r.runtime); r.runtime = resumed.result.session_id; u.byRuntime.set(r.runtime, r) }
           r.seq = 0; this.reset(u, 'upstream_reset'); restore(true); continue
         }
@@ -514,7 +569,7 @@ export class RealtimeBroker {
         if (replay.error || replay.result?.truncated || replay.result?.epoch !== u.epoch) {this.reset(u, 'upstream_replay_unavailable');restore(true)}
         else {for (const p of replay.result?.events ?? []) this.event(u, { jsonrpc: '2.0', method: 'event', params: p }, true);restore(false)}
       }
-    } catch { this.reset(u, 'upstream_recovery_failed') }
+    } catch (error) { this.reset(u, 'upstream_recovery_failed'); throw error }
     finally {
       u.recovering = false
       const deferred = u.deferred; u.deferred = []

src/server/workspaceGateway.ts

@@ -277,6 +277,10 @@ export class WorkspaceNodes {
 
 /** One server-owned transport per execution binding, never tied to a browser. */
 export class WorkspaceGateway {
+  private closed = false
+  private generation = 0
+  private available = false
+  get connected(): boolean { return this.available }
   private serverRequests = new Map<string, {method:'approval'|'clarify';sessionId:string}>()
   private socket?: WebSocket
   private paired?: WorkspacePairedChannel
@@ -289,15 +293,26 @@ export class WorkspaceGateway {
   onDisconnect: () => void = () => {}
   constructor(readonly target: GatewayTarget,readonly scope?:GatewayExecutionScope) {}
   async connect(): Promise<void> {
-    if(this.target.runner){this.runner=await this.target.runner.open(frame=>this.onEvent(frame),()=>this.onDisconnect(),this.scope);return}
+    if (this.closed) throw new Error('Hermes connection closed')
+    const generation = ++this.generation
+    const active = () => !this.closed && generation === this.generation
+    const disconnected = () => { if(active()){this.available=false;this.onDisconnect()} }
+    if(this.target.runner){
+      const runner=await this.target.runner.open(frame=>{if(active())this.onEvent(frame)},disconnected,this.scope)
+      if(!active()){runner.close();throw new Error('Hermes connection closed')}
+      this.runner=runner;this.available=true;return
+    }
     if (this.target.pairedToken) {
       this.paired = new WorkspacePairedChannel(this.target.url, this.target.pairedToken, this.target.client.fetchImpl,
-        frame => this.onEvent(frame), () => this.onDisconnect())
+        frame => {if(active())this.onEvent(frame)}, disconnected)
       await this.paired.connect()
+      if(!active())throw new Error('Hermes connection closed')
+      this.available = true
       return
     }
     const credential = await this.target.session.webSocketCredential(),
       url = new URL(this.target.url)
+    if (!active()) throw new Error('Hermes connection closed')
     url.protocol = url.protocol === 'https:' ? 'wss:' : 'ws:'
     url.pathname = `${url.pathname.replace(/\/$/, '')}/api/ws`
     url.search = ''
@@ -318,7 +333,17 @@ export class WorkspaceGateway {
       socket.terminate()
       failed(new Error('Hermes gateway readiness timed out'))
     }, 20_000)
+    let awaitingPong = false
+    const heartbeat = setInterval(() => {
+      if (!active() || socket.readyState !== WebSocket.OPEN) return
+      if (awaitingPong) { socket.terminate(); return }
+      awaitingPong = true
+      socket.ping()
+    }, 15_000)
+    heartbeat.unref()
+    socket.on('pong', () => { awaitingPong = false })
     socket.on('message', (raw) => {
+      if (!active()) return
       try {
         const frame = JSON.parse(raw.toString()),
           waiter = this.pending.get(String(frame.id))
@@ -360,17 +385,21 @@ export class WorkspaceGateway {
       failed(error)
     })
     socket.on('close', () => {
+      clearInterval(heartbeat)
+      clearTimeout(handshakeTimer)
+      failed(new Error('Hermes connection closed'))
+      if (!active()) return
       for (const waiter of this.pending.values()) {
         clearTimeout(waiter.timer)
         waiter.reject(new Error('Hermes connection closed'))
       }
       this.pending.clear()
       this.serverRequests.clear()
-      this.onDisconnect()
-      clearTimeout(handshakeTimer)
-      failed(new Error('Hermes connection closed'))
+      disconnected()
     })
     await handshake
+    if (!active() || socket.readyState !== WebSocket.OPEN) throw new Error('Hermes connection closed')
+    this.available = true
   }
   private acceptServerRequest(frame:any):boolean {
     if(!['approval','clarify'].includes(frame?.method)||typeof frame.id!=='string'||typeof frame.params?.session_id!=='string')return false
@@ -424,10 +453,18 @@ export class WorkspaceGateway {
     })
   }
   close(): void {
+    this.closed = true
+    this.available = false
+    this.generation++
     this.onDisconnect = () => {}
+    for (const waiter of this.pending.values()) {
+      clearTimeout(waiter.timer)
+      waiter.reject(new Error('Hermes connection closed'))
+    }
+    this.pending.clear()
     this.serverRequests.clear()
     this.paired?.close()
     this.runner?.close()
-    this.socket?.close()
+    this.socket?.terminate()
   }
 }

src/server/workspaceRuntime.ts

@@ -2,6 +2,7 @@ import {type DesktopEnvironments} from './desktopEnvironments.js'
 import {type GrokCloud} from './grokCloud.js'
 import {VmToolSession} from './vmComputer.js'
 import { randomUUID } from 'node:crypto'
+import { setTimeout as delay } from 'node:timers/promises'
 import { readFileSync } from 'node:fs'
 import { z } from 'zod'
 import { WorkspaceStore, parse } from './workspaceStore.js'
@@ -69,6 +70,8 @@ export interface WorkspaceBinding {
   conversationTaskId?: string
   taskId?: string
   contextSeq?: number
+  /** This admitted turn needs its original in-memory tool grant to keep running. */
+  toolsBound?: boolean
 }
 interface LiveTurn {
   gateway: WorkspaceGateway
@@ -385,6 +388,7 @@ export class WorkspaceRuntime extends WorkspaceScheduler {
     let flushTimer: ReturnType<typeof setTimeout> | undefined
     let usageCompletion: Promise<void> | undefined
     let submittedText = ''
+    let reconnecting: Promise<void> | undefined
     const toolController = new AbortController()
     let toolLease: WorkspaceToolLease | undefined
     let pluginLease: OpenBotPlugins | undefined
@@ -460,8 +464,105 @@ export class WorkspaceRuntime extends WorkspaceScheduler {
       } catch (failure) { rejectTurn(failure instanceof Error ? failure : new Error('无法保存执行状态')) }
       finally { vmTools?.close(); if (usageCompletion) void usageCompletion.finally(() => gateway.close()); else gateway.close() }
     }
-    gateway.onDisconnect = () =>
-      finish(completedEvidence ? undefined : new Error('Hermes 连接断开'))
+    const restoreCompleted = async () => {
+      const marker = `[yaoyao-run:${run.runId}:${resultMessage.id}]`
+      let answer: { content?: string; text?: string } | undefined
+      let found = false, seenLastEvent = false
+      for (let page = 0; page < 20 && !found; page++) {
+        const response = await target.session.request(`/api/sessions/${encodeURIComponent(binding!.storedId)}/messages`, {
+          search: new URLSearchParams({ profile: agent.profile, limit: '500', offset: String(page * 500), order: 'latest', include_compacted: 'true' }),
+          maxResponseBytes: 8 * 1024 * 1024,
+        })
+        if (settled) return
+        gateway.scope?.authorize()
+        if (response.status !== 200) throw new Error('无法核对历史')
+        const history = JSON.parse(response.body.toString()).messages
+        if (!Array.isArray(history)) throw new Error('历史响应无效')
+        for (const item of [...history].reverse()) {
+          if (item.role === 'user') {
+            if (String(item.content ?? item.text).includes(marker)) { found = true; break }
+            answer = undefined; seenLastEvent = false
+          } else if (!seenLastEvent && ['assistant', 'tool'].includes(item.role)) {
+            seenLastEvent = true
+            const calls = item.tool_calls
+            const hasCalls = Array.isArray(calls) ? calls.length > 0 : !!calls && calls !== '[]'
+            if (item.role === 'assistant' && !hasCalls && !item.function_call && item.finish_reason !== 'tool_calls' && (item.content || item.text)) answer = item
+          }
+        }
+        if (history.length < 500) break
+      }
+      if (!found || !answer) throw new Error('无法确认本轮的最终回复，正在核对原执行；不会重复提交')
+      resultMessage.content = String(answer.content ?? answer.text)
+      binding!.contextSeq = Math.max(binding!.contextSeq ?? 0, run.contextThroughSeq ?? 0)
+      this.store.put(owner, 'binding', key, binding!)
+      finish()
+    }
+    const lostTools = () => new HttpError(409, '原执行仍在运行，但本轮工具授权已丢失，无法完整恢复；请停止后重新发起。', 'workspace_tool_lease_lost')
+    const recoverTransport = async () => {
+      let attempt = 0
+      while (!settled) {
+        try {
+          gateway.scope?.authorize()
+          await delay(Math.min(30_000, 500 * 2 ** Math.min(attempt++, 6)), undefined, { signal: toolController.signal })
+          gateway.scope?.authorize()
+          if (!gateway.connected) await gateway.connect()
+          if (settled) return
+          gateway.scope?.authorize()
+          const opened = await gateway.rpc('session.resume', { profile: agent.profile, session_id: binding!.storedId, omit_messages: true, close_on_disconnect: false })
+          if (settled) return
+          gateway.scope?.authorize()
+          if (!opened.session_id || (opened.info?.profile_name && opened.info.profile_name !== agent.profile)) throw new Error('Hermes 会话身份不匹配')
+          runtimeId = String(opened.session_id)
+          const previousStoredId = binding!.storedId
+          const storedId = String(opened.stored_session_id ?? opened.session_key ?? opened.resumed ?? previousStoredId)
+          if (!storedId) throw new Error('Hermes 会话身份不匹配')
+          binding!.storedId = storedId
+          binding!.aliases = [...new Set([...binding!.aliases, previousStoredId, storedId])]
+          binding!.runtimeId = runtimeId
+          this.store.put(owner, 'binding', key, binding!)
+          const live = this.live.get(key)
+          if (live?.taskId === run.id) live.runtimeId = runtimeId
+          if (!opened.running) {
+            try { await restoreCompleted() } catch (error) { finish(error instanceof Error ? error : new Error('无法核对原执行')) }
+            return
+          }
+          if (binding!.toolsBound && !toolLease) throw lostTools()
+          if (toolLease) {
+            try { await toolLease.bind() }
+            catch (error) {
+              if (error instanceof HttpError && (error.code === 'team_tools_bind_failed' || error.code === 'hermes_bridge_session_busy')) throw lostTools()
+              throw error
+            }
+          }
+          if (settled) return
+          gateway.scope?.authorize()
+          if (!gateway.connected) continue
+          const current = this.getWork(owner, run.id)
+          current.error = undefined
+          if (current.status === 'uncertain') current.status = 'running'
+          resultMessage.status = 'streaming'; resultMessage.error = undefined
+          this.store.atomic(() => { this.saveWork(owner, current); this.store.saveMessage(owner, resultMessage) })
+          return
+        } catch (error) {
+          if (settled) return
+          try { gateway.scope?.authorize() } catch (revoked) { finish(revoked instanceof Error ? revoked : new Error('本轮授权已结束')); return }
+          if (error instanceof HttpError && error.code === 'workspace_tool_lease_lost') { finish(error); return }
+          // No new prompt is admitted while the original outcome is unknown.
+          const current = this.getWork(owner, run.id)
+          current.error = error instanceof Error ? error.message : '正在恢复 Hermes 连接'
+          this.saveWork(owner, current)
+        }
+      }
+    }
+    gateway.onDisconnect = () => {
+      if (settled || reconnecting) return
+      if (completedEvidence || !submitted || !binding?.storedId || !runtimeId) { finish(completedEvidence ? undefined : new Error('Hermes 连接断开')); return }
+      const current = this.getWork(owner, run.id)
+      current.error = 'Hermes 连接断开，正在恢复原执行'
+      this.saveWork(owner, current)
+      reconnecting = recoverTransport().finally(() => { reconnecting = undefined })
+      void reconnecting.catch(error => finish(error instanceof Error ? error : new Error('无法恢复 Hermes 连接')))
+    }
     gateway.onEvent = (frame: GatewayFrame) => {
       if (settled || !runtimeId || frame.session_id !== runtimeId) return
       const currentTurn = this.getWork(owner, run.id)
@@ -638,6 +739,7 @@ export class WorkspaceRuntime extends WorkspaceScheduler {
       )
         throw new Error('Hermes 会话身份不匹配')
       binding = {
+        toolsBound: recovering ? binding?.toolsBound : undefined,
         memoryVersion: binding?.memoryVersion,
         memoryState: binding?.memoryState,
         memoryPending: binding?.memoryPending,
@@ -671,39 +773,12 @@ export class WorkspaceRuntime extends WorkspaceScheduler {
         throw new Error('运行已停止')
       if (recovering) {
         if (opened.running) {
+          if (binding.toolsBound) throw lostTools()
           const current = this.getWork(owner, run.id)
           current.status = 'running'
           this.saveWork(owner, current)
         } else {
-          const marker = `[yaoyao-run:${run.runId}:${resultMessage.id}]`
-          let answer: { content?: string; text?: string } | undefined
-          let found = false, seenLastEvent = false
-          for (let page = 0; page < 20 && !found; page++) {
-            const response = await target.session.request(`/api/sessions/${encodeURIComponent(storedId)}/messages`, {
-              search: new URLSearchParams({ profile: agent.profile, limit: '500', offset: String(page * 500), order: 'latest', include_compacted: 'true' }),
-              maxResponseBytes: 8 * 1024 * 1024,
-            })
-            if (response.status !== 200) throw new Error('无法核对历史')
-            const history = JSON.parse(response.body.toString()).messages
-            if (!Array.isArray(history)) throw new Error('历史响应无效')
-            for (const item of [...history].reverse()) {
-              if (item.role === 'user') {
-                if (String(item.content ?? item.text).includes(marker)) { found = true; break }
-                answer = undefined; seenLastEvent = false
-              } else if (!seenLastEvent && ['assistant', 'tool'].includes(item.role)) {
-                seenLastEvent = true
-                const calls = item.tool_calls
-                const hasCalls = Array.isArray(calls) ? calls.length > 0 : !!calls && calls !== '[]'
-                if (item.role === 'assistant' && !hasCalls && !item.function_call && item.finish_reason !== 'tool_calls' && (item.content || item.text)) answer = item
-              }
-            }
-            if (history.length < 500) break
-          }
-          if (!found || !answer) throw new Error('无法确认本轮的最终回复，正在核对原执行；不会重复提交')
-          resultMessage.content = String(answer.content ?? answer.text)
-          binding.contextSeq = Math.max(binding.contextSeq ?? 0, run.contextThroughSeq ?? 0)
-          this.store.put(owner, 'binding', key, binding)
-          finish()
+          await restoreCompleted()
         }
       } else {
         if (opened.running) throw new HttpError(409, '上游会话仍在运行', 'session_busy')
@@ -759,6 +834,8 @@ export class WorkspaceRuntime extends WorkspaceScheduler {
             onFailure:error=>{if(!settled)void gateway.rpc('session.interrupt',{session_id:runtimeId}).catch(()=>{}).finally(()=>finish(error))},
           })
           await toolLease.bind();if(settled)return await completion
+          binding.toolsBound = true
+          this.store.put(owner, 'binding', key, binding)
         }
         const workingDirectory = await applyWorkingDirectory((method, params) => gateway.rpc(method, params), runtimeId, cwd, undefined, opened.info?.cwd)
         environment.cwd=workingDirectory?.resolved
@@ -861,6 +938,7 @@ export class WorkspaceRuntime extends WorkspaceScheduler {
       }
       return await completion
     } catch (error) {
+      if (reconnecting && submitted && !settled) return await completion
       finish(error instanceof Error ? error : new Error('执行失败'))
       return await completion
     }
@@ -941,7 +1019,8 @@ export class WorkspaceRuntime extends WorkspaceScheduler {
     const key = this.bindingKey(work.conversationId, work.conversationTaskId, work.agentId)
     const live = this.live.get(key)
     if (live?.taskId === work.id) {
-      await live.gateway.rpc('session.interrupt', { session_id: live.runtimeId })
+      try { await live.gateway.rpc('session.interrupt', { session_id: live.runtimeId }) }
+      catch (error) { if (!live.gateway.connected) live.done(new Error('已停止')); throw error }
       const current = this.getWork(owner, work.id)
       if (['complete', 'failed', 'interrupted'].includes(current.status)) {
         live.done(new Error('已停止'))

src/server/workspaceScheduler.ts

@@ -181,7 +181,7 @@ export abstract class WorkspaceScheduler {
     } else {
       root.status = active.some(t => t.status === 'uncertain') ? 'uncertain' : active.some(t => t.status === 'running') ? 'running' : active.some(t => t.status === 'waiting') ? 'waiting' : tasks.some(t => t.status !== 'queued') ? 'running' : 'queued'
     }
-    root.error = tasks.find(t => t.status === 'uncertain' || t.status === 'failed')?.error
+    root.error = tasks.find(t => t.status === 'uncertain' || t.status === 'failed')?.error ?? active.find(t => t.error)?.error
     this.store.saveRun(owner, root)
     if (done && !wasTerminal) this.onRunSettled(owner, root)
     if (done && !wasTerminal && root.status !== 'interrupted') {

src/server/workspaceToolLease.ts

@@ -30,6 +30,8 @@ async function bridgeRequest(target: GatewayTarget, path: string, body: unknown)
       && /^[a-z0-9][a-z0-9_-]{0,63}$/.test(body.profile) ? ` Profile「${body.profile}」` : ''
     if (path === '/bind' && Object.hasOwn(failures, code))
       throw new HttpError(response.status >= 500 ? 502 : 409, `Hermes${profile}${failures[code]}。`, `hermes_bridge_${code}`)
+    if (response.status >= 500 || response.status === 408 || response.status === 429)
+      throw new HttpError(502, 'Hermes 工具桥暂时不可用，正在重试原授权。', 'team_tools_bridge_unavailable')
     throw new HttpError(409, 'Hermes 工具桥绑定失败，请确认该 Profile 已启用工具桥，并新建会话后重试。', 'team_tools_bind_failed')
   }
   return value
@@ -73,6 +75,7 @@ export async function createWorkspaceToolLease(input: LeaseInput): Promise<Works
   const token = randomBytes(32).toString('base64url'), tokenHash = createHash('sha256').update(token).digest()
   const generation = `${input.workId}:${randomUUID()}`
   let expiresAt = Date.now() + TTL, closed = false, url = ''
+  let renewAfter = Date.now() + 5 * 60_000
   let timer: ReturnType<typeof setInterval> | undefined
   let binding: Promise<void> | undefined, cleanup: Promise<void> | undefined
   let boundRuntimeId = ''
@@ -181,8 +184,9 @@ export async function createWorkspaceToolLease(input: LeaseInput): Promise<Works
     binding = (async () => {
       const bindSession = async () => {
         const current = input.session()
+        const requestedExpiry = Date.now() + TTL
         const result = await bridgeRequest(input.target, '/bind', { native_tools: true, session_id: current.runtimeId,
-          stored_session_id: current.storedId, profile: input.profile, generation, bridge_url: url, token, expires_at: expiresAt,
+          stored_session_id: current.storedId, profile: input.profile, generation, bridge_url: url, token, expires_at: requestedExpiry,
           ...(input.computerPolicy ? {computer_policy: input.computerPolicy} : {}),
           ...(input.workspaceMemory ? {workspace_memory: true} : {}) })
         if(input.workspaceMemory&&result.workspace_memory!==true)
@@ -191,6 +195,10 @@ export async function createWorkspaceToolLease(input: LeaseInput): Promise<Works
           throw new HttpError(409, '请更新 Hermes 的夭夭工具桥，当前版本尚未接管隔离电脑会话。', 'computer_bridge_upgrade_required')
         if (!bound && result.native_tools !== true)
           throw new HttpError(409, 'Hermes 未挂载本轮原生团队工具，请升级或修复工具桥。', 'team_tools_unavailable')
+        // A lost renewal receipt cannot extend the original local grant.
+        assertActive()
+        expiresAt = requestedExpiry
+        renewAfter = Date.now() + 5 * 60_000
         bound = true
       }
       try { await bindSession() }
@@ -214,11 +222,17 @@ export async function createWorkspaceToolLease(input: LeaseInput): Promise<Works
     input.signal.addEventListener('abort', abort, { once: true })
     assertActive()
     timer = setInterval(() => {
-      if (binding) return
       try { assertActive() } catch (error) { void dispose(); input.onFailure(error as Error); return }
-      expiresAt = Date.now() + TTL
-      void bind().catch(error => { void dispose(); input.onFailure(error as Error) })
-    }, 5 * 60_000)
+      if (binding || Date.now() < renewAfter) return
+      void bind().catch(error => {
+        if (closed) return
+        const rejected = error instanceof HttpError && ['team_tools_bind_failed', 'hermes_bridge_session_busy', 'hermes_bridge_agent_initialization_failed'].includes(error.code)
+        const transient = !(error instanceof HttpError) || !rejected && (error.status >= 500 || [408, 429].includes(error.status)
+          || ['hermes_bridge_initializing', 'hermes_bridge_binding_busy', 'hermes_bridge_bridge_catalog_unreachable'].includes(error.code))
+        if (transient && Date.now() < expiresAt) return
+        void dispose(); input.onFailure(error as Error)
+      })
+    }, 30_000)
     timer.unref()
     return { bind, dispose, async profileRequest(path, body) {
       assertActive()

tests/client/conversationRecovery.test.ts

@@ -0,0 +1,99 @@
+import { afterEach, beforeEach, expect, it, vi } from 'vitest'
+import { flushPromises, shallowMount, type VueWrapper } from '@vue/test-utils'
+import { shallowRef } from 'vue'
+import { createMemoryHistory, createRouter, matchedRouteKey } from 'vue-router'
+import ConversationsView from '@/views/ConversationsView.vue'
+import WorkspaceMessageTimeline from '@/components/workspace/WorkspaceMessageTimeline.vue'
+import { apiRequest } from '@/api/client'
+
+vi.mock('@/api/client', async importOriginal => ({ ...await importOriginal<typeof import('@/api/client')>(), apiRequest: vi.fn(), setApiCsrfToken: vi.fn() }))
+vi.mock('@/stores/auth', () => ({ useAuthStore: () => ({ user: { id: 'owner' }, profiles: [], refreshProfileAvatars: async () => {} }) }))
+vi.mock('@/stores/theme', () => ({ useThemeStore: () => ({}) }))
+let wrapper: VueWrapper | undefined
+const sources: Array<{ url: string; close: ReturnType<typeof vi.fn> }> = []
+beforeEach(() => {
+  vi.useFakeTimers()
+  sources.length = 0
+  vi.stubGlobal('EventSource', class extends EventTarget {
+    readonly close = vi.fn()
+    constructor(readonly url: string) { super(); sources.push(this) }
+  })
+})
+afterEach(() => {
+  wrapper?.unmount(); wrapper = undefined
+  vi.mocked(apiRequest).mockReset(); vi.restoreAllMocks(); vi.unstubAllGlobals(); vi.useRealTimers()
+})
+async function mountView(path = '/conversations') {
+  const router = createRouter({ history: createMemoryHistory(), routes: [{ path: '/conversations/:id?', component: { template: '<div />' } }] })
+  await router.push(path); await router.isReady()
+  wrapper = shallowMount(ConversationsView, { global: {
+    provide: { [matchedRouteKey as symbol]: shallowRef(router.currentRoute.value.matched[0]) }, plugins: [router],
+    stubs: { WorkspaceShell: { template: '<div><slot name="sidebar" /><slot /></div>' }, ComposerShell: { template: '<div />', methods: { filesSnapshot: () => [] } }, Teleport: true },
+  } })
+  await flushPromises()
+}
+function successfulResponse(path: string) {
+  if (path === '/api/app/capabilities') return { features: [] }
+  if (path === '/api/app/workspace/snapshot') return { agents: [], conversations: [], details: [], cursor: 17 }
+  throw new Error(`Unexpected request ${path}`)
+}
+it('recovers an initial server outage while the browser stays online', async () => {
+  vi.mocked(apiRequest).mockRejectedValueOnce(new Error('服务暂不可用')).mockImplementation(async path => successfulResponse(path) as never)
+  await mountView()
+  expect(sources).toHaveLength(0)
+  await vi.advanceTimersByTimeAsync(1500); await flushPromises()
+  expect(sources.map(source => source.url)).toEqual(['/api/app/events/stream?after=17'])
+  expect(vi.mocked(apiRequest).mock.calls.filter(([path]) => path === '/api/app/capabilities')).toHaveLength(2)
+})
+it('cancels a scheduled initial recovery when the view is destroyed', async () => {
+  vi.mocked(apiRequest).mockRejectedValue(new Error('服务暂不可用'))
+  await mountView()
+  wrapper!.unmount(); wrapper = undefined
+  await vi.advanceTimersByTimeAsync(10_000)
+  expect(apiRequest).toHaveBeenCalledTimes(1)
+  expect(sources).toHaveLength(0)
+})
+it('aborts an in-flight snapshot and prevents a late response from connecting after destruction', async () => {
+  let signal: AbortSignal | undefined, release!: (value: unknown) => void
+  vi.mocked(apiRequest).mockImplementation(async (path, options) => {
+    if (path === '/api/app/capabilities') return successfulResponse(path) as never
+    signal = options?.signal as AbortSignal
+    return new Promise(resolve => { release = resolve })
+  })
+  await mountView()
+  expect(signal?.aborted).toBe(false)
+  wrapper!.unmount(); wrapper = undefined
+  expect(signal?.aborted).toBe(true)
+  release(successfulResponse('/api/app/workspace/snapshot'))
+  await flushPromises(); await vi.advanceTimersByTimeAsync(10_000)
+  expect(sources).toHaveLength(0)
+  expect(apiRequest).toHaveBeenCalledTimes(2)
+})
+it('cancels recovery while offline and restarts it only when connectivity returns', async () => {
+  let online = true
+  vi.spyOn(navigator, 'onLine', 'get').mockImplementation(() => online)
+  vi.mocked(apiRequest).mockRejectedValueOnce(new Error('服务暂不可用')).mockImplementation(async path => successfulResponse(path) as never)
+  await mountView()
+  online = false; window.dispatchEvent(new Event('offline'))
+  await vi.advanceTimersByTimeAsync(10_000)
+  expect(apiRequest).toHaveBeenCalledTimes(1)
+  online = true; window.dispatchEvent(new Event('online')); await flushPromises()
+  expect(sources.map(source => source.url)).toEqual(['/api/app/events/stream?after=17'])
+})
+it('aborts a pending detail on going offline and ignores its late response', async () => {
+  let online = true, signal: AbortSignal | undefined, release!: (value: unknown) => void
+  vi.spyOn(navigator, 'onLine', 'get').mockImplementation(() => online)
+  vi.mocked(apiRequest).mockImplementation(async (path, options) => {
+    if (!path.startsWith('/api/app/conversations/')) return successfulResponse(path) as never
+    signal = options?.signal as AbortSignal
+    return new Promise(resolve => { release = resolve })
+  })
+  await mountView('/conversations/c')
+  expect(signal?.aborted).toBe(false)
+  online = false; window.dispatchEvent(new Event('offline'))
+  expect(signal?.aborted).toBe(true)
+  release({ conversation: { id: 'c', kind: 'direct', name: '迟到会话', memberIds: [], readSeq: 0, lastSeq: 0 }, messages: [], run: null, interactions: [], context: null })
+  await flushPromises(); await vi.advanceTimersByTimeAsync(5000)
+  expect(wrapper!.findComponent(WorkspaceMessageTimeline).props('title')).not.toBe('迟到会话')
+  expect(sources).toHaveLength(0)
+})

tests/client/httpRealtimeRecovery.test.ts

@@ -0,0 +1,39 @@
+import { afterEach, expect, it, vi } from 'vitest'
+import { HTTPRealtimeChannel } from '@/api/httpRealtime'
+const api = vi.hoisted(() => ({ request: vi.fn() }))
+vi.mock('@/api/client', async importOriginal => ({ ...await importOriginal<typeof import('@/api/client')>(), apiRequest: api.request }))
+afterEach(() => { vi.unstubAllGlobals(); vi.useRealTimers(); api.request.mockReset() })
+
+it('times out response headers on every reconnect and preserves the received cursor', async () => {
+  vi.useFakeTimers()
+  api.request.mockResolvedValue({ id: 'channel' })
+  let firstStream!: ReadableStreamDefaultController<Uint8Array>
+  const attempts: RequestInit[] = []
+  vi.stubGlobal('fetch', vi.fn((_url: unknown, options: RequestInit) => {
+    attempts.push(options)
+    if (attempts.length > 1) return new Promise((_resolve, reject) => {
+      options.signal?.addEventListener('abort', () => reject(new Error('headers timed out')), { once: true })
+    })
+    return Promise.resolve(new Response(new ReadableStream<Uint8Array>({ start(controller) {
+      firstStream = controller
+      controller.enqueue(new TextEncoder().encode('id: epoch:9\nevent: frame\ndata: {"params":{"type":"gateway.ready"}}\n\n'))
+    } }), { headers: { 'content-type': 'text/event-stream' } }))
+  }))
+  const failed = vi.fn(), channel = new HTTPRealtimeChannel(async () => {}, failed)
+  try {
+    await channel.open('chat')
+    firstStream.error(new Error('connection lost'))
+    await vi.advanceTimersByTimeAsync(500)
+    expect(attempts).toHaveLength(2)
+    await vi.advanceTimersByTimeAsync(45_000)
+    expect(attempts[1].signal?.aborted).toBe(true)
+    await vi.advanceTimersByTimeAsync(1000)
+    expect(attempts).toHaveLength(3)
+    expect(attempts[2].headers).toMatchObject({ 'Last-Event-ID': 'epoch:9' })
+    expect(failed).not.toHaveBeenCalled()
+  } finally {
+    channel.close()
+    await vi.advanceTimersByTimeAsync(60_000)
+    expect(attempts).toHaveLength(3)
+  }
+})

tests/client/ordinaryTranscriptV2.test.ts

@@ -62,6 +62,40 @@ describe('ordinary v2 durable Web replica', () => {
     } as unknown as TranscriptSnapshot)
     expect(client.snapshot).toBeUndefined()
   })
+  it('rejects an ahead control cursor and ignores a delayed control without replacing state', async () => {
+    const { client, changed } = setup()
+    const saved = { ...client.snapshot, running: true, queued: true, liveStatus: '进行中' }
+    client.restore(saved)
+    await expect(client.control({ epoch: saved.epoch, cursor: saved.cursor + 1, running: false, queued: false })).rejects.toThrow('版本不连续')
+    await client.control({ epoch: saved.epoch, cursor: saved.cursor - 1, running: false, queued: false })
+    expect(changed).not.toHaveBeenCalled()
+    expect(client.snapshot).toEqual(saved)
+    client.close()
+  })
+  it.each(['ready', 'state'])('repairs a %s cursor gap with a snapshot instead of skipping messages', async event => {
+    vi.useFakeTimers()
+    const { client, changed } = setup()
+    const repaired = { ...fixture.snapshot, cursor: 3, hasOlder: false,
+      messages: [{ ...fixture.snapshot.messages[0], revision: 3, content: '补齐的消息' }] }
+    api.request.mockResolvedValue(repaired)
+    let attempts = 0
+    vi.stubGlobal('fetch', vi.fn((_url: unknown, options: RequestInit) => {
+      attempts++
+      return Promise.resolve(new Response(new ReadableStream<Uint8Array>({ start(controller) {
+        if (attempts === 1) controller.enqueue(new TextEncoder().encode(`event: ${event}\ndata: ${JSON.stringify({ epoch: repaired.epoch, cursor: 3, running: false, queued: false })}\n\n`))
+        options.signal?.addEventListener('abort', () => { try { controller.close() } catch {} }, { once: true })
+      } }), { status: 200 }))
+    }))
+    const running = client.run()
+    try {
+      await vi.advanceTimersByTimeAsync(1000)
+      expect(api.request).toHaveBeenCalledWith(expect.stringContaining('/snapshot?'), expect.anything())
+      expect(changed.mock.lastCall![0]).toMatchObject({ cursor: 3, messages: repaired.messages })
+      expect(attempts).toBe(2)
+    } finally {
+      client.close(); await running; vi.unstubAllGlobals(); vi.useRealTimers()
+    }
+  })
   it('reopens an inactive event connection without dropping the durable cursor', async () => {
     vi.useFakeTimers()
     const { client } = setup()

tests/server/ordinaryTranscriptReplay.test.ts

@@ -0,0 +1,72 @@
+// @vitest-environment node
+import Koa from 'koa'
+import { mkdtempSync, rmSync } from 'node:fs'
+import { tmpdir } from 'node:os'
+import { join } from 'node:path'
+import type { Server } from 'node:http'
+import { afterEach, expect, it, vi } from 'vitest'
+import { ChatCacheStore, type ChatCacheCoordinator } from '../../src/server/chatCache'
+import { chatTranscriptRouter } from '../../src/server/chatTranscriptApi'
+import type { LocalAuthStore } from '../../src/server/localAuth'
+import { ChatTranscriptClient } from '../../src/client/api/chatTranscript'
+import type { TranscriptSnapshot } from '../../src/shared/chatTranscript'
+
+const scope = ['owner', 'p', 's'] as const
+const fixtures: Array<{ home: string; store: ChatCacheStore; server: Server }> = []
+afterEach(async () => {
+  vi.unstubAllGlobals()
+  for (const { home, store, server } of fixtures.splice(0)) {
+    store.close()
+    server.closeAllConnections()
+    await new Promise<void>(resolve => server.close(() => resolve()))
+    rmSync(home, { recursive: true, force: true })
+  }
+})
+
+it.each([300, 1000])('replays %i committed events through the HTTP router before confirming the cursor', async count => {
+  const home = mkdtempSync(join(tmpdir(), 'ordinary-replay-')), store = new ChatCacheStore(home)
+  store.recordCommand(...scope, 'session.create', { source: 'web' })
+  store.transcripts.seed(...scope)
+  const base = store.transcripts.snapshot(...scope, {}, 'current')
+  for (let i = 0; i < count; i++) {
+    store.transcripts.upsert(...scope, { id: `m${i}`, role: 'assistant', content: `消息${i}`, status: 'complete' }, i)
+  }
+  const finalCursor = store.transcripts.cursor(...scope)
+  const auth = {
+    require: () => ({ id: scope[0] }), current: () => ({ id: scope[0] }),
+    canUseSource: () => true, isUserActive: () => true, pushAuthorizationVersion: () => 1,
+  } as unknown as LocalAuthStore
+  const router = chatTranscriptRouter({ store, schedule: () => {} } as unknown as ChatCacheCoordinator, auth)
+  const app = new Koa().use(router.routes()).use(router.allowedMethods())
+  const server = app.listen(0, '127.0.0.1')
+  fixtures.push({ home, store, server })
+  await new Promise<void>(resolve => server.once('listening', resolve))
+  const origin = `http://127.0.0.1:${(server.address() as { port: number }).port}`
+  const nativeFetch = globalThis.fetch
+  const requested: string[] = []
+  vi.stubGlobal('fetch', (input: string | URL | Request, init?: RequestInit) => {
+    requested.push(String(input))
+    return nativeFetch(new URL(String(input), origin), init)
+  })
+  let checkpoint: TranscriptSnapshot = base, commits = 0
+  const failed = vi.fn()
+  const client = new ChatTranscriptClient('p', 's', async value => { checkpoint = value; commits++ }, failed)
+  client.restore(base)
+  const running = client.run()
+  try {
+    // Restored checkpoint, every event, then ready; no snapshot repair is necessary.
+    await vi.waitFor(() => expect(commits).toBe(count + 2), { timeout: 5000 })
+    expect(checkpoint.cursor).toBe(finalCursor)
+    expect(checkpoint.messages).toEqual(store.transcripts.all(...scope))
+    expect(checkpoint.messages).toHaveLength(count)
+    expect(failed).not.toHaveBeenCalled()
+    expect(requested).toEqual(['/api/app/chat/sessions/s/events?profile=p'])
+    store.transcripts.upsert(...scope, { id: 'live', role: 'assistant', content: '重连后新增', status: 'complete' }, count)
+    await vi.waitFor(() => expect(checkpoint.messages).toHaveLength(count + 1))
+    expect(checkpoint.messages.at(-1)?.content).toBe('重连后新增')
+    expect(failed).not.toHaveBeenCalled()
+  } finally {
+    client.close()
+    await running
+  }
+}, 15_000)

tests/server/realtimeBroker.test.ts

@@ -8,7 +8,7 @@ import { RealtimeBroker, type RealtimeActivity, type RealtimePrincipal, type Str
 import { RealtimeReceipts } from '../../src/server/realtimeReceipts.js'
 
 const cleanup: Array<() => void> = []
-afterEach(() => cleanup.splice(0).reverse().forEach(f => f()))
+afterEach(() => { cleanup.splice(0).reverse().forEach(f => f()); vi.restoreAllMocks() })
 async function fixture() {
   const home = mkdtempSync(join(tmpdir(), 'yaoyao-realtime-test-'))
   const ws = new WebSocketServer({ port: 0, host: '127.0.0.1' })
@@ -18,6 +18,8 @@ async function fixture() {
   let count = 0
   let seq = 0, truncate = false, dropPrompt = false
   let cwd = '/old-session', busy = false, rejectCwd = false
+  let loseResumeReply = false
+  let resumeError: number | undefined
   const history: any[] = []
   let peer: WebSocket
   ws.on('connection', socket => {
@@ -26,6 +28,11 @@ async function fixture() {
     socket.on('message', raw => {
       const f = JSON.parse(raw.toString()); commands.push(f)
       if (f.method === 'prompt.submit' && dropPrompt) { socket.terminate(); return }
+      if (f.method === 'session.resume' && loseResumeReply) { loseResumeReply = false; return }
+      if (f.method === 'session.resume' && resumeError !== undefined) {
+        socket.send(JSON.stringify({ id: f.id, error: { code: resumeError, message: resumeError === 4007 ? 'session not found' : 'database temporarily unavailable' } }))
+        resumeError = undefined; return
+      }
       if (f.method === 'profiles.list') {
         socket.send(JSON.stringify({id:f.id,result:{profiles:[{name:'default',ui_meta:{'hermes-bots':{chat:'stored-1'}}}]}}))
       } else if (f.method === 'session.events.since') {
@@ -53,6 +60,8 @@ async function fixture() {
   return { home, broker, principal, commands, activities, count: () => count, advance: (n: number) => { now += n },
     cwd: () => cwd, busy: () => { busy = true }, rejectCwd: () => { rejectCwd = true },
     disconnect: () => peer!.terminate(), truncate: () => { truncate = true }, dropPrompt: () => { dropPrompt = true },
+    loseResumeReply: () => { loseResumeReply = true },
+    failNextResume: (code: number) => { resumeError = code },
     emit: (type: string, payload: unknown) => {
       const params = { type, session_id: 'runtime-1', seq: ++seq, payload }; history.push(params)
       if (peer!.readyState === 1) peer!.send(JSON.stringify({ method: 'event', params }))
@@ -355,6 +364,111 @@ describe('realtime broker', () => {
     expect(f.count()).toBe(2)
     expect(received.filter(e => e.event === 'reset')).toHaveLength(0)
   })
+  it('keeps retrying credential failures and replays missed events without resubmitting prompts', async () => {
+    const f = await fixture(), p = f.principal('alice'), url = vi.fn(p.url)
+    const channel = await f.broker.create({ ...p, url }, 'chat')
+    await f.broker.command(channel, 'open', resume)
+    await f.broker.command(channel, 'send-once', { method: 'prompt.submit', params: { session_id: 'runtime-1', text: 'once' } })
+    const received: StreamEntry[] = []
+    f.broker.subscribe(channel, undefined, entry => received.push(entry))
+    url.mockRejectedValueOnce(new Error('Hermes is restarting')).mockRejectedValueOnce(new Error('Ticket endpoint unavailable'))
+    f.disconnect(); f.emit('message.delta', { text: 'missed-during-restart' })
+    await vi.waitFor(() => expect(received.filter(e => e.data.includes('missed-during-restart'))).toHaveLength(1), { timeout: 5000 })
+    expect(url).toHaveBeenCalledTimes(4)
+    expect(f.count()).toBe(2)
+    expect(f.commands.filter(command => command.method === 'prompt.submit')).toHaveLength(1)
+  }, 7000)
+  it('shares one credential attempt when clients reconnect during the retry delay', async () => {
+    const f = await fixture(), p = f.principal('alice'), url = vi.fn(p.url)
+    await f.broker.create({ ...p, url }, 'chat')
+    let release!: (url: URL) => void
+    url.mockImplementationOnce(() => new Promise(resolve => { release = resolve }))
+    f.disconnect()
+    await vi.waitFor(() => expect(url).toHaveBeenCalledTimes(2))
+    const second = f.broker.create({ ...p, url }, 'chat')
+    const third = f.broker.create({ ...p, url }, 'chat')
+    release(await p.url('chat'))
+    await Promise.all([second, third])
+    expect(url).toHaveBeenCalledTimes(2)
+    expect(f.count()).toBe(2)
+  })
+  it('retries a connected gateway whose session recovery never replies', async () => {
+    const f = await fixture(), channel = await f.broker.create(f.principal('alice'), 'chat')
+    await f.broker.command(channel, 'open', resume)
+    const received: StreamEntry[] = []
+    f.broker.subscribe(channel, undefined, entry => received.push(entry))
+    const setTimeout = globalThis.setTimeout
+    vi.spyOn(globalThis, 'setTimeout').mockImplementation(((callback: (...args: any[]) => void, ms?: number, ...args: any[]) =>
+      setTimeout(callback, ms === 120_000 ? 30 : ms, ...args)) as typeof globalThis.setTimeout)
+    f.loseResumeReply(); f.disconnect(); f.emit('message.delta', { text: 'after-recovery-timeout' })
+    await vi.waitFor(() => expect(received.some(e => e.data.includes('after-recovery-timeout'))).toBe(true), { timeout: 3000 })
+    expect(f.count()).toBe(3)
+    expect(received.some(e => e.data.includes('upstream_recovery_failed'))).toBe(true)
+  })
+  it('retries an explicit temporary session recovery error instead of reporting ready', async () => {
+    const f = await fixture(), channel = await f.broker.create(f.principal('alice'), 'chat')
+    await f.broker.command(channel, 'open', resume)
+    const received: StreamEntry[] = []
+    f.broker.subscribe(channel, undefined, entry => received.push(entry))
+    f.failNextResume(5000); f.disconnect(); f.emit('message.delta', { text: 'after-temporary-resume-error' })
+    await vi.waitFor(() => expect(received.some(e => e.data.includes('after-temporary-resume-error'))).toBe(true), { timeout: 3000 })
+    expect(f.count()).toBe(3)
+    expect(channel.routes.size).toBe(1)
+  })
+  it('retires a missing stored session without endlessly reconnecting the service', async () => {
+    const f = await fixture(), channel = await f.broker.create(f.principal('alice'), 'chat')
+    await f.broker.command(channel, 'open', resume)
+    const received: StreamEntry[] = []
+    f.broker.subscribe(channel, undefined, entry => received.push(entry))
+    f.failNextResume(4007); f.disconnect()
+    await vi.waitFor(() => expect(received.some(e => e.data.includes('session_not_found'))).toBe(true), { timeout: 2500 })
+    expect(channel.routes.size).toBe(0)
+    await new Promise(resolve => setTimeout(resolve, 1200))
+    expect(f.count()).toBe(2)
+    expect(f.commands.filter(command => command.method === 'session.resume')).toHaveLength(2)
+  })
+  it('does not reopen a socket when authorization is revoked during credential lookup', async () => {
+    const f = await fixture(), p = f.principal('alice')
+    let valid = true, release!: (url: URL) => void
+    const url = vi.fn(p.url)
+    await f.broker.create({ ...p, valid: () => valid, url }, 'chat')
+    url.mockImplementationOnce(() => new Promise(resolve => { release = resolve }))
+    f.disconnect()
+    await vi.waitFor(() => expect(url).toHaveBeenCalledTimes(2))
+    valid = false
+    release(await p.url('chat'))
+    await new Promise(resolve => setTimeout(resolve, 650))
+    expect(url).toHaveBeenCalledTimes(2)
+    expect(f.count()).toBe(1)
+  })
+  it('does not reopen a socket after shutdown with credentials still pending', async () => {
+    const f = await fixture(), p = f.principal('alice'), url = vi.fn(p.url)
+    await f.broker.create({ ...p, url }, 'chat')
+    let release!: (url: URL) => void
+    url.mockImplementationOnce(() => new Promise(resolve => { release = resolve }))
+    f.disconnect()
+    await vi.waitFor(() => expect(url).toHaveBeenCalledTimes(2))
+    f.broker.close()
+    release(await p.url('chat'))
+    await new Promise(resolve => setTimeout(resolve, 650))
+    expect(url).toHaveBeenCalledTimes(2)
+    expect(f.count()).toBe(1)
+  })
+  it('recovers a shared service connection for another authorized user after its first user expires', async () => {
+    const f = await fixture()
+    let firstValid = true
+    const firstURL = vi.fn(f.principal('alice').url), secondURL = vi.fn(f.principal('bob').url)
+    await f.broker.create({ ...f.principal('alice'), valid: () => firstValid, url: firstURL }, 'chat')
+    const channel = await f.broker.create({ ...f.principal('bob'), url: secondURL }, 'chat')
+    await f.broker.command(channel, 'open', resume)
+    const received: StreamEntry[] = []
+    f.broker.subscribe(channel, undefined, entry => received.push(entry))
+    firstValid = false
+    f.disconnect(); f.emit('message.delta', { text: 'still-authorized' })
+    await vi.waitFor(() => expect(received.some(e => e.data.includes('still-authorized'))).toBe(true), { timeout: 2500 })
+    expect(firstURL).toHaveBeenCalledTimes(1)
+    expect(secondURL).toHaveBeenCalledTimes(1)
+  })
   it('reports reset when upstream history was truncated', async () => {
     const f = await fixture(), c = await f.broker.create(f.principal('alice'), 'chat')
     await f.broker.command(c, 'open', resume)

tests/server/workspace.test.ts

@@ -570,7 +570,7 @@ describe('Web-owned workspace', () => {
     }
     const run = runtime.send(owner, c.id, { requestId: randomUUID(), content: '修改一次' })
     await vi.waitFor(() =>
-      expect(store.require<WorkspaceRun>(owner, 'run', run.id).status).toBe('uncertain'),
+      expect(store.require<WorkspaceRun>(owner, 'run', run.id).error).toContain('正在恢复原执行'),
     )
     runtime.close()
     const configured = requests.filter(r => r.method === 'config.set').length
@@ -1250,7 +1250,7 @@ it('recovers a partially completed parallel batch after restart without resubmit
   const last=pending.get(c.name)!
   recoveryByStored.set(storedByRuntime.get(last.p.session_id)!,[{role:'user',content:last.p.text},{role:'assistant',content:'丙已完成'}])
   last.socket.terminate()
-  await vi.waitFor(()=>expect(store.require<WorkspaceRun>(owner,'run',root.id).status).toBe('uncertain'))
+  await vi.waitFor(()=>expect(store.require<WorkspaceRun>(owner,'run',root.id).error).toContain('正在恢复原执行'))
   runtime.close();runtime=new WorkspaceRuntime(store,nodes,uploads);runtime.start()
   await finished(root.id)
   expect(admin).toBe(2)
@@ -1530,7 +1530,7 @@ it('upgrades an admitted v0.3.7 execution without submitting its prompt again',
   const a=agent('升级执行'),c=direct(a.id)
   reply=(socket,p)=>{recoveryHistory=[{role:'user',content:p.text},{role:'assistant',content:'升级前已完成'}];socket.terminate()}
   const root=runtime.send(owner,c.id,{requestId:randomUUID(),content:'只执行一次'})
-  await vi.waitFor(()=>expect(store.require<WorkspaceRun>(owner,'run',root.id).status).toBe('uncertain'))
+  await vi.waitFor(()=>expect(store.require<WorkspaceRun>(owner,'run',root.id).error).toContain('正在恢复原执行'))
   runtime.close()
   const old=store.list<any>(owner,'turn')[0]!
   for(const task of store.list<any>(owner,'turn'))store.remove(owner,'turn',task.id)

tests/server/workspaceGateway.test.ts

@@ -1,5 +1,5 @@
 // @vitest-environment node
-import {expect,it} from 'vitest'
+import {expect,it,vi} from 'vitest'
 import {WebSocketServer,type WebSocket} from 'ws'
 import {WorkspaceGateway,type GatewayFrame,type GatewayTarget} from '../../src/server/workspaceGateway'
 
@@ -45,3 +45,64 @@ it('adapts Hermes server requests, scopes answers, restores pending cards and wi
     await new Promise<void>(done=>server.close(()=>done()))
   }
 })
+
+it('ignores frames and closes from the previous socket after a reconnect', async () => {
+  const server = new WebSocketServer({port: 0, host: '127.0.0.1'})
+  await new Promise<void>(resolve => server.once('listening', resolve))
+  let current!: WebSocket, pending: any
+  server.on('connection', socket => {
+    current = socket
+    socket.send(JSON.stringify({method: 'event', params: {type: 'gateway.ready'}}))
+    socket.on('message', raw => { pending = JSON.parse(String(raw)) })
+  })
+  const target = {url: new URL(`http://127.0.0.1:${(server.address() as {port:number}).port}`), client: {},
+    session: {webSocketCredential: async () => ({name: 'ticket', value: 'fixture'})}} as unknown as GatewayTarget
+  const gateway = new WorkspaceGateway(target), events = vi.fn(), disconnect = vi.fn()
+  gateway.onEvent = events; gateway.onDisconnect = disconnect
+  try {
+    await gateway.connect()
+    const previous = (gateway as any).socket as WebSocket
+    current.terminate()
+    await expect.poll(() => disconnect.mock.calls.length).toBe(1)
+    await gateway.connect()
+    const reply = gateway.rpc('session.resume', {session_id: 'stored'})
+    await expect.poll(() => pending?.method).toBe('session.resume')
+    previous.emit('message', Buffer.from(JSON.stringify({method: 'event', params: {type: 'message.complete', session_id: 'stale'}})))
+    previous.emit('close')
+    expect(gateway.connected).toBe(true)
+    expect(disconnect).toHaveBeenCalledOnce()
+    expect(events.mock.calls.some(([frame]) => frame.session_id === 'stale')).toBe(false)
+    current.send(JSON.stringify({id: pending.id, result: {session_id: 'current'}}))
+    await expect(reply).resolves.toEqual({session_id: 'current'})
+  } finally {
+    gateway.close(); for (const client of server.clients) client.terminate()
+    await new Promise<void>(resolve => server.close(() => resolve()))
+  }
+})
+
+it('detects a half-open connection when Hermes stops answering pings', async () => {
+  vi.useFakeTimers({toFake: ['setInterval', 'clearInterval']})
+  const server = new WebSocketServer({port: 0, host: '127.0.0.1', autoPong: false})
+  await new Promise<void>(resolve => server.once('listening', resolve))
+  const ping = vi.fn()
+  server.on('connection', socket => {
+    socket.on('ping', ping)
+    socket.send(JSON.stringify({method: 'event', params: {type: 'gateway.ready'}}))
+  })
+  const target = {url: new URL(`http://127.0.0.1:${(server.address() as {port:number}).port}`), client: {},
+    session: {webSocketCredential: async () => ({name: 'ticket', value: 'fixture'})}} as unknown as GatewayTarget
+  const gateway = new WorkspaceGateway(target), disconnect = vi.fn()
+  gateway.onDisconnect = disconnect
+  try {
+    await gateway.connect()
+    await vi.advanceTimersByTimeAsync(15_000)
+    await expect.poll(() => ping.mock.calls.length).toBe(1)
+    await vi.advanceTimersByTimeAsync(15_000)
+    await expect.poll(() => disconnect.mock.calls.length).toBe(1)
+    expect(gateway.connected).toBe(false)
+  } finally {
+    gateway.close(); for (const client of server.clients) client.terminate()
+    await new Promise<void>(resolve => server.close(() => resolve()))
+    vi.useRealTimers()
+  }
+})

tests/server/workspaceReconnect.test.ts

@@ -0,0 +1,189 @@
+// @vitest-environment node
+import {afterEach, beforeEach, expect, it, vi} from 'vitest'
+import {mkdtempSync, rmSync} from 'node:fs'
+import {tmpdir} from 'node:os'
+import {join} from 'node:path'
+import {randomUUID} from 'node:crypto'
+import {WebSocketServer, type WebSocket} from 'ws'
+import {WorkspaceStore} from '../../src/server/workspaceStore'
+import {WorkspaceRuntime} from '../../src/server/workspaceRuntime'
+import {type GatewayTarget, type WorkspaceNodes} from '../../src/server/workspaceGateway'
+import {type WorkspacePlugins} from '../../src/server/botPlugins/workspacePlugins'
+import {UploadStore} from '../../src/server/uploads'
+
+const owner = 'reconnect-user'
+let home: string, store: WorkspaceStore, uploads: UploadStore, runtime: WorkspaceRuntime
+let server: WebSocketServer, socket: WebSocket, target: GatewayTarget, nodes: WorkspaceNodes
+let running: boolean, authorized: boolean, acknowledgePrompt: boolean, holdCredential: (() => Promise<void>) | undefined
+let holdBind: (() => Promise<void>) | undefined
+let resumedRuntimeId: string, resumedStoredId: string, historyPaths: string[]
+let rpc: Array<{method: string; params: any}>, binds: any[], unbinds: any[], history: any[]
+let pluginDispose: ReturnType<typeof vi.fn>, pluginCall: ReturnType<typeof vi.fn>
+beforeEach(async () => {
+  home = mkdtempSync(join(tmpdir(), 'yaoyao-workspace-reconnect-'))
+  store = new WorkspaceStore(home); uploads = new UploadStore(home)
+  running = true; authorized = true; acknowledgePrompt = true; holdCredential = undefined; holdBind = undefined
+  resumedRuntimeId = 'runtime'; resumedStoredId = 'stored'; historyPaths = []
+  rpc = []; binds = []; unbinds = []; history = []
+  server = new WebSocketServer({port: 0, host: '127.0.0.1'})
+  await new Promise<void>(resolve => server.once('listening', resolve))
+  target = {
+    url: new URL(`http://127.0.0.1:${(server.address() as {port: number}).port}`), client: {directAgent: undefined},
+    session: {
+      webSocketCredential: async () => { await holdCredential?.(); return {name: 'ticket', value: 'fixture'} },
+      request: async (path: string, options: any) => {
+        if (path.endsWith('/bind')) { binds.push(options.body); await holdBind?.() }
+        if (path.endsWith('/unbind')) unbinds.push(options.body)
+        if (path.startsWith('/api/sessions/')) historyPaths.push(path)
+        const value = path.endsWith('/capabilities') ? {ready: true, native_tools: true, in_process: true}
+          : path.startsWith('/api/plugins/yaoyao-bot-bridge/') ? {ok: true, native_tools: true}
+          : path === '/api/config' ? {terminal: {}} : {messages: history}
+        return {status: 200, body: Buffer.from(JSON.stringify(value)), headers: new Headers()}
+      },
+    },
+  } as unknown as GatewayTarget
+  nodes = {requireSource: () => {}, target: () => target} as unknown as WorkspaceNodes
+  runtime = new WorkspaceRuntime(store, nodes, uploads, () => authorized)
+  pluginDispose = vi.fn(async () => {})
+  pluginCall = vi.fn(async () => ({message: '插件仍可使用'}))
+  runtime.plugins = {selected: () => true, open: async () => ({services: () => [],
+    catalog: () => [{id: 'plugin_fixture', name: 'plugin_fixture', description: '回归测试', inputSchema: {type: 'object'}}],
+    call: pluginCall, dispose: pluginDispose,
+  })} as unknown as WorkspacePlugins
+  server.on('connection', connection => {
+    socket = connection
+    connection.send(JSON.stringify({method: 'event', params: {type: 'gateway.ready'}}))
+    connection.on('message', raw => {
+      const frame = JSON.parse(String(raw)); rpc.push(frame)
+      const result = frame.method === 'session.create' ? {session_id: 'runtime', stored_session_id: 'stored', running: false, info: {profile_name: 'default'}}
+        : frame.method === 'session.resume' ? {session_id: resumedRuntimeId, stored_session_id: resumedStoredId, running, info: {profile_name: 'default'}}
+        : {status: 'streaming'}
+      if (frame.method !== 'prompt.submit' || acknowledgePrompt) connection.send(JSON.stringify({id: frame.id, result}))
+    })
+  })
+})
+afterEach(async () => {
+  runtime.close()
+  await new Promise(resolve => setTimeout(resolve, 20))
+  for (const client of server.clients) client.terminate()
+  await new Promise<void>(resolve => server.close(() => resolve()))
+  uploads.close(); store.close(); rmSync(home, {recursive: true, force: true})
+})
+async function start() {
+  const agent = store.createAgent(owner, {name: '恢复测试', profile: 'default'})
+  const conversation = store.list<any>(owner, 'conversation').find(c => c.kind === 'direct' && c.memberIds[0] === agent.id)
+  const run = runtime.send(owner, conversation.id, {requestId: randomUUID(), content: '只执行一次'})
+  await expect.poll(() => rpc.filter(r => r.method === 'prompt.submit')).toHaveLength(1)
+  return {run, conversation, agent}
+}
+async function callback(path: string, body: unknown) {
+  const lease = binds[0]
+  return fetch(lease.bridge_url + path, {method: 'POST', headers: {Authorization: `Bearer ${lease.token}`, 'Content-Type': 'application/json'}, body: JSON.stringify(body)})
+}
+function complete(text = '恢复完成') { socket.send(JSON.stringify({method: 'event', params: {type: 'message.complete', session_id: 'runtime', payload: {text}}})) }
+
+it.each([true, false])('keeps the original tool and plugin grants through running recovery (prompt receipt=%s)', async receipt => {
+  acknowledgePrompt = receipt
+  const {run, conversation, agent} = await start()
+  const first = binds[0]
+  socket.terminate()
+  await expect.poll(() => binds.length).toBe(2)
+  expect(binds[1]).toMatchObject({generation: first.generation, token: first.token, bridge_url: first.bridge_url})
+  expect(unbinds).toHaveLength(0)
+  expect(pluginDispose).not.toHaveBeenCalled()
+  const catalog = await (await callback('/tools/list', {})).json()
+  const tool = catalog.tools.find((t: any) => t.name === 'workspace_list_agents')
+  const result = await (await callback('/tools/call', {toolId: tool.id, arguments: {}, callId: 'after-reconnect'})).json()
+  expect(result.structuredContent.agents).toEqual(expect.arrayContaining([expect.objectContaining({id: agent.id})]))
+  const plugin = catalog.tools.find((t: any) => t.name === 'plugin_fixture')
+  await callback('/tools/call', {toolId: plugin.id, arguments: {}, callId: 'plugin-after-reconnect'})
+  expect(pluginCall).toHaveBeenCalledOnce()
+  complete()
+  await expect.poll(() => store.require<any>(owner, 'run', run.id).status).toBe('complete')
+  expect(store.messages(owner, conversation.id).at(-1)?.content).toBe('恢复完成')
+  expect(rpc.filter(r => r.method === 'prompt.submit')).toHaveLength(1)
+  await expect.poll(() => unbinds.length).toBe(1)
+  expect(pluginDispose).toHaveBeenCalledOnce()
+})
+
+it('reconciles completion during a disconnect without renewing or resubmitting the finished turn', async () => {
+  const {run, conversation} = await start()
+  history = [{role: 'user', content: rpc.find(r => r.method === 'prompt.submit')!.params.text}, {role: 'assistant', content: '断线期间完成'}]
+  running = false; socket.terminate()
+  await expect.poll(() => store.require<any>(owner, 'run', run.id).status).toBe('complete')
+  expect(store.messages(owner, conversation.id).at(-1)?.content).toBe('断线期间完成')
+  expect(binds).toHaveLength(1)
+  expect(rpc.filter(r => r.method === 'prompt.submit')).toHaveLength(1)
+})
+
+it('continues reconnecting if the replacement socket closes while the original grant is renewing', async () => {
+  const {run} = await start()
+  let release!: () => void
+  holdBind = () => new Promise<void>(resolve => { release = resolve })
+  socket.terminate()
+  await expect.poll(() => binds.length).toBe(2)
+  socket.terminate()
+  await expect.poll(() => server.clients.size).toBe(0)
+  holdBind = undefined; release()
+  await expect.poll(() => binds.length, {timeout: 3000}).toBe(3)
+  expect(new Set(binds.map(b => b.generation)).size).toBe(1)
+  complete()
+  await expect.poll(() => store.require<any>(owner, 'run', run.id).status).toBe('complete')
+  expect(rpc.filter(r => r.method === 'prompt.submit')).toHaveLength(1)
+})
+
+it('follows resumed runtime and compacted stored IDs for tool renewal and subsequent history reconciliation', async () => {
+  const {run} = await start()
+  resumedRuntimeId = 'runtime-after-resume'; resumedStoredId = 'compacted-tip'
+  socket.terminate()
+  await expect.poll(() => binds.length).toBe(2)
+  expect(binds[1]).toMatchObject({session_id: resumedRuntimeId, stored_session_id: resumedStoredId, generation: binds[0].generation})
+  expect(store.list<any>(owner, 'binding')[0]).toMatchObject({runtimeId: resumedRuntimeId, storedId: resumedStoredId, aliases: ['stored', 'compacted-tip']})
+  running = false
+  history = [{role: 'user', content: rpc.find(r => r.method === 'prompt.submit')!.params.text}, {role: 'assistant', content: '压缩后完成'}]
+  socket.terminate()
+  await expect.poll(() => store.require<any>(owner, 'run', run.id).status).toBe('complete')
+  expect(historyPaths).toEqual(['/api/sessions/compacted-tip/messages'])
+  expect(rpc.filter(r => r.method === 'prompt.submit')).toHaveLength(1)
+})
+
+it.each(['stop', 'revoke', 'close'] as const)('revokes callbacks and cannot revive a detached turn after %s', async action => {
+  const {run} = await start()
+  socket.terminate()
+  await expect.poll(() => store.require<any>(owner, 'run', run.id).error).toContain('正在恢复原执行')
+  if (action === 'stop') await runtime.stop(owner, run.id)
+  else if (action === 'revoke') authorized = false
+  else runtime.close()
+  await expect.poll(() => unbinds.length).toBe(1)
+  await expect(callback('/tools/list', {})).rejects.toThrow()
+  expect(pluginDispose).toHaveBeenCalledOnce()
+  expect(binds).toHaveLength(1)
+  expect(rpc.filter(r => r.method === 'prompt.submit')).toHaveLength(1)
+  if (action === 'stop') expect(store.require<any>(owner, 'run', run.id).status).toBe('interrupted')
+})
+
+it('does not create a late WebSocket when the service closes during reconnect credentials', async () => {
+  const {run} = await start()
+  let release!: () => void
+  const credential = new Promise<void>(resolve => { release = resolve })
+  let waiting = false
+  holdCredential = () => { waiting = true; return credential }
+  socket.terminate()
+  await expect.poll(() => waiting).toBe(true)
+  runtime.close(); release()
+  await expect.poll(() => unbinds.length).toBe(1)
+  expect(server.clients.size).toBe(0)
+  expect(rpc.filter(r => r.method === 'session.resume')).toHaveLength(0)
+  expect(store.require<any>(owner, 'run', run.id).status).toBe('uncertain')
+})
+
+it('keeps a running turn uncertain after process restart when its original tools cannot be restored', async () => {
+  const {run} = await start()
+  runtime.close()
+  runtime = new WorkspaceRuntime(store, nodes, uploads, () => authorized)
+  runtime.start()
+  await expect.poll(() => store.require<any>(owner, 'run', run.id).error).toContain('本轮工具授权已丢失')
+  expect(store.require<any>(owner, 'run', run.id).status).toBe('uncertain')
+  expect(binds).toHaveLength(1)
+  expect(rpc.filter(r => r.method === 'prompt.submit')).toHaveLength(1)
+})

tests/server/workspaceTeamTools.test.ts

@@ -267,13 +267,13 @@ describe('Bot self rules', () => {
   })
 })
 
-async function mount(signal = new AbortController().signal) {
+async function mount(signal = new AbortController().signal, onFailure: (error: Error) => void = () => {}) {
   const lease = await createWorkspaceToolLease({
     target, profile:manager.profile, workId:work.id,
     session:()=>({runtimeId:'runtime',storedId:'stored'}),signal,
     assertActive:()=>{runtime.teamTools.assertTurn(owner,work.id)},
     catalog:()=>[...runtime.teamTools.catalog(owner,work.id),...runtime.knowledgeTools.catalog(owner,work.id,false)],
-    call:(id,args)=>runtime.knowledgeTools.handles(id)?runtime.knowledgeTools.call(owner,work.id,id,args,false):runtime.teamTools.call(owner,work.id,id,args),onFailure:()=>{},
+    call:(id,args)=>runtime.knowledgeTools.handles(id)?runtime.knowledgeTools.call(owner,work.id,id,args,false):runtime.teamTools.call(owner,work.id,id,args),onFailure,
   })
   leases.push(lease); await lease.bind()
   return lease
@@ -283,6 +283,46 @@ function http(path:string, body:unknown, options:RequestInit={}) {
 }
 
 describe('loopback tool lease', () => {
+  it.each(['503', 'upstream_unavailable'])('retains the original grant through %s renewal failures, then renews after recovery', async failure => {
+    vi.useFakeTimers({toFake: ['Date', 'setInterval', 'clearInterval']})
+    try {
+      const failed = vi.fn()
+      await mount(undefined, failed)
+      const original = {...binding}, request = vi.mocked(target.session.request).getMockImplementation()!
+      let unavailable = true
+      vi.mocked(target.session.request).mockImplementation(async (path, options) => {
+        if (path.endsWith('/bind') && unavailable) {
+          if (failure === 'upstream_unavailable') throw new HttpError(502, 'Unable to reach Hermes', 'upstream_unavailable')
+          return {status: 503, body: Buffer.from('{}'), headers: new Headers()}
+        }
+        return request(path, options)
+      })
+      await vi.advanceTimersByTimeAsync(5 * 60_000)
+      expect(failed).not.toHaveBeenCalled()
+      expect((await http('/tools/list', {})).status).toBe(200)
+      unavailable = false
+      await vi.advanceTimersByTimeAsync(30_000)
+      expect(binding).toMatchObject({generation: original.generation, token: original.token, bridge_url: original.bridge_url})
+      expect(binding!.expires_at).toBeGreaterThan(original.expires_at)
+      expect(failed).not.toHaveBeenCalled()
+    } finally { vi.useRealTimers() }
+  })
+
+  it.each([401, 503])('retires a lease on rejected renewal or its original expiry (HTTP %s)', async status => {
+    vi.useFakeTimers({toFake: ['Date', 'setInterval', 'clearInterval']})
+    try {
+      const failed = vi.fn()
+      await mount(undefined, failed)
+      const request = vi.mocked(target.session.request).getMockImplementation()!
+      vi.mocked(target.session.request).mockImplementation(async (path, options) => path.endsWith('/bind')
+        ? {status, body: Buffer.from('{}'), headers: new Headers()} : request(path, options))
+      await vi.advanceTimersByTimeAsync((status === 401 ? 5 : 30) * 60_000)
+      expect(failed).toHaveBeenCalledOnce()
+      expect(failed.mock.calls[0]![0]).toMatchObject({code: status === 401 ? 'team_tools_bind_failed' : 'team_tools_expired'})
+      await expect(http('/tools/list', {})).rejects.toThrow()
+    } finally { vi.useRealTimers() }
+  })
+
   it('updates self rules through the bridge without ending the current tool lease', async () => {
     await mount()
     const catalog = await (await http('/tools/list',{})).json()