[SDK] Delete the bridge agent and rework the session bundle onto lex clients (#11385)

Co-authored-by: Claude Fable 5 <noreply@anthropic.com>

src/state/messages/events/agent.ts

@@ -1,9 +1,8 @@
-import {type AtpAgent, type ChatBskyConvoGetLog} from '@atproto/api'
+import {type Client} from '@atproto/lex'
 import {EventEmitter} from 'eventemitter3'
 import {nanoid} from 'nanoid/non-secure'
 
 import {networkRetry} from '#/lib/async/retry'
-import {DM_SERVICE_HEADERS} from '#/lib/constants'
 import {
   isErrorMaybeAppPasswordPermissions,
   isNetworkError,
@@ -21,13 +20,14 @@ import {
   type MessagesEventBusParams,
   MessagesEventBusStatus,
 } from '#/state/messages/events/types'
+import {chat} from '#/lexicons'
 
 const logger = Logger.create(Logger.Context.DMsAgent)
 
 export class MessagesEventBus {
   private id: string
 
-  private agent: AtpAgent
+  private chatClient: Client
   private emitter = new EventEmitter<{event: [MessagesEventBusEvent]}>()
 
   private status: MessagesEventBusStatus = MessagesEventBusStatus.Initializing
@@ -37,11 +37,20 @@ export class MessagesEventBus {
 
   constructor(params: MessagesEventBusParams) {
     this.id = nanoid(3)
-    this.agent = params.agent
+    this.chatClient = params.chatClient
 
     this.init()
   }
 
+  /**
+   * Point the bus at a new chat client, for when the session bundle is
+   * replaced underneath it. The poll cursor and subscribers are unaffected: the
+   * same account is being served over a fresh session.
+   */
+  updateClient(chatClient: Client) {
+    this.chatClient = chatClient
+  }
+
   requestPollInterval(interval: number) {
     const id = nanoid()
     this.requestedPollIntervals.set(id, interval)
@@ -260,14 +269,11 @@ export class MessagesEventBus {
 
     try {
       const response = await networkRetry(2, () => {
-        return this.agent.chat.bsky.convo.getLog(
-          {},
-          {headers: DM_SERVICE_HEADERS},
-        )
+        return this.chatClient.call(chat.bsky.convo.getLog, {})
       })
       // throw new Error('UNCOMMENT TO TEST INIT FAILURE')
 
-      const {cursor} = response.data
+      const {cursor} = response
 
       // should always be defined
       if (cursor) {
@@ -355,21 +361,18 @@ export class MessagesEventBus {
     // )
 
     let needsEmit = false
-    let batch: ChatBskyConvoGetLog.OutputSchema['logs'] = []
+    let batch: chat.bsky.convo.getLog.$OutputBody['logs'] = []
 
     try {
       const response = await networkRetry(2, () => {
-        return this.agent.chat.bsky.convo.getLog(
-          {
-            cursor: this.latestRev,
-          },
-          {headers: DM_SERVICE_HEADERS},
-        )
+        return this.chatClient.call(chat.bsky.convo.getLog, {
+          cursor: this.latestRev,
+        })
       })
 
       // throw new Error('UNCOMMENT TO TEST POLL FAILURE')
 
-      const {logs: events} = response.data
+      const {logs: events} = response
 
       for (const ev of events) {
         /*

src/state/messages/events/index.tsx

@@ -2,7 +2,7 @@ import {createContext, useContext, useEffect, useState} from 'react'
 import {AppState} from 'react-native'
 
 import {MessagesEventBus} from '#/state/messages/events/agent'
-import {useAgent, useSession} from '#/state/session'
+import {useChatClient, useSession} from '#/state/session'
 
 const MessagesEventBusContext = createContext<MessagesEventBus | null>(null)
 MessagesEventBusContext.displayName = 'MessagesEventBusContext'
@@ -42,14 +42,24 @@ export function MessagesEventBusProviderInner({
 }: {
   children: React.ReactNode
 }) {
-  const agent = useAgent()
+  const chatClient = useChatClient()
   const [bus] = useState(
     () =>
       new MessagesEventBus({
-        agent,
+        chatClient,
       }),
   )
 
+  /*
+   * The bus outlives the client it was constructed with: replacing the session
+   * bundle (account switch, cross-tab token sync, expiry rescue) builds fresh
+   * clients over the new session and disposes the old ones, so a bus still
+   * holding the previous client would poll through a dead session.
+   */
+  useEffect(() => {
+    bus.updateClient(chatClient)
+  }, [bus, chatClient])
+
   useEffect(() => {
     bus.resume()
 

src/state/messages/events/types.ts

@@ -1,7 +1,10 @@
-import {type AtpAgent, type ChatBskyConvoGetLog} from '@atproto/api'
+import {type Client} from '@atproto/lex'
+
+import {type chat} from '#/lexicons'
 
 export type MessagesEventBusParams = {
-  agent: AtpAgent
+  /** The chat client, which proxies `chat.bsky.*` to the chat service. */
+  chatClient: Client
 }
 
 export enum MessagesEventBusStatus {
@@ -64,5 +67,5 @@ export type MessagesEventBusEvent =
     }
   | {
       type: 'logs'
-      logs: ChatBskyConvoGetLog.OutputSchema['logs']
+      logs: chat.bsky.convo.getLog.$OutputBody['logs']
     }