feat(adapter): enhance error handling in state subscription by logging errors from listeners

Author: gene9831

packages/kit/src/message/adapters/shared.ts

@@ -20,6 +20,14 @@ const normalizeKinds = (kindsOrListener: MessageUpdateKinds | ((currentState: Pu
 export const createStateSubscriptionController = (getState: () => PublicMessageState) => {
   const listeners = new Set<ListenerEntry>()
 
+  const dispatch = (entry: ListenerEntry, snapshot: PublicMessageState) => {
+    try {
+      entry.listener(snapshot)
+    } catch (error) {
+      console.error('Error in message state subscriber:', error)
+    }
+  }
+
   const notify = (kindsOrKind: MessageUpdateKinds) => {
     const kinds = new Set(Array.isArray(kindsOrKind) ? kindsOrKind : [kindsOrKind])
     const snapshot = getState()
@@ -40,7 +48,7 @@ export const createStateSubscriptionController = (getState: () => PublicMessageS
         }
       }
 
-      entry.listener(snapshot)
+      dispatch(entry, snapshot)
     }
   }
 
@@ -60,7 +68,7 @@ export const createStateSubscriptionController = (getState: () => PublicMessageS
     }
 
     listeners.add(entry)
-    listener(getState())
+    dispatch(entry, getState())
 
     return () => {
       listeners.delete(entry)

packages/kit/src/message/test/native.test.ts

@@ -145,6 +145,46 @@ describe('createMessageEngine', () => {
     expect(snapshotsForRequestState).toHaveLength(expectedRequestStateSequence.length)
   })
 
+  it('keeps notifying other subscribers when one subscriber throws', () => {
+    const adapter = createNativeMessageAdapter()
+    adapter.initialize({
+      requestState: 'idle',
+      processingState: undefined,
+      messages: [],
+    })
+
+    const errorSpy = vi.spyOn(console, 'error').mockImplementation(() => {})
+    const healthyListener = vi.fn()
+
+    adapter.subscribe('requestState', () => {
+      throw new Error('subscriber failed')
+    })
+    adapter.subscribe('requestState', healthyListener)
+
+    adapter.mutate('requestState', (draft) => {
+      draft.requestState = 'processing'
+      draft.processingState = 'requesting'
+    })
+
+    expect(healthyListener).toHaveBeenNthCalledWith(
+      1,
+      expect.objectContaining({
+        requestState: 'idle',
+        processingState: undefined,
+      }),
+    )
+    expect(healthyListener).toHaveBeenNthCalledWith(
+      2,
+      expect.objectContaining({
+        requestState: 'processing',
+        processingState: 'requesting',
+      }),
+    )
+    expect(errorSpy).toHaveBeenCalled()
+
+    errorSpy.mockRestore()
+  })
+
   it('runs onBeforeRequest with current request body before assistant is appended', async () => {
     const onBeforeRequest = vi.fn()
     const engine = createTestMessageEngine({