migrate the stateful chat agents to the chat client

Convo and MessagesEventBus were the last two useAgent consumers holding a
long-lived agent rather than reading one per render. Both now take a chat
client, which carries the atproto-proxy header itself, so the per-call
DM_SERVICE_HEADERS drop out under the double-set rule.

Both providers gained an effect that pushes the current client into the
instance. The instances are created once with useState and outlive the client
they were built with: an account switch, a cross-tab token sync or an expiry
rescue replaces the bundle and disposes the previous clients, so without the
sync a convo would keep sending through a dead session.

`handleSendMessageFailure` narrows on XrpcResponseError rather than XRPCError.
The lex client throws the former, so the status branch (and the recoverable /
unrecoverable split that drives the retry banner) had silently stopped firing.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>

src/screens/Messages/components/MessagesList.tsx

@@ -70,7 +70,7 @@ import {Loader} from '#/components/Loader'
 import {Text} from '#/components/Typography'
 import {useAnalytics} from '#/analytics'
 import {IS_ANDROID, IS_NATIVE, IS_WEB} from '#/env'
-import {app} from '#/lexicons'
+import {app, type chat, type com} from '#/lexicons'
 import * as bsky from '#/types/bsky'
 import {ChatStatusInfo} from './ChatStatusInfo'
 import {groupSystemMessages, type RenderItem} from './groupSystemMessages'
@@ -537,10 +537,7 @@ export function MessagesList({
        */
       rt.detectFacetsWithoutResolution()
 
-      let embed:
-        | $Typed<AppBskyEmbedRecord.Main>
-        | $Typed<ChatBskyEmbedJoinLink.Main>
-        | undefined
+      let embed: chat.bsky.convo.defs.MessageInput['embed']
       let embedView:
         | $Typed<AppBskyEmbedRecord.View>
         | $Typed<ChatBskyEmbedJoinLink.View>
@@ -576,10 +573,15 @@ export function MessagesList({
           if (post) {
             embed = {
               $type: 'app.bsky.embed.record',
+              /*
+               * `getPost` still returns an `@atproto/api` view, whose `uri` and
+               * `cid` are plain strings rather than the branded syntax types
+               * the lexicon input declares.
+               */
               record: {
                 uri: post.uri,
                 cid: post.cid,
-              },
+              } as com.atproto.repo.strongRef.Main,
             }
 
             embedView = {

src/state/messages/convo/agent.ts

@@ -1,20 +1,17 @@
 import {
   type $Typed,
   type AppBskyEmbedRecord,
-  type AtpAgent,
   type ChatBskyActorDefs,
   ChatBskyConvoDefs,
   type ChatBskyConvoGetLog,
-  type ChatBskyConvoSendMessage,
   type ChatBskyEmbedJoinLink,
   type ChatBskyGroupDefs,
 } from '@atproto/api'
-import {XRPCError} from '@atproto/api'
+import {type Client, XrpcResponseError} 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,
@@ -52,6 +49,7 @@ import {
   parseConvoView,
 } from '#/components/dms/util'
 import {IS_NATIVE} from '#/env'
+import {chat} from '#/lexicons'
 
 const logger = Logger.create(Logger.Context.ConversationAgent)
 
@@ -105,7 +103,7 @@ function toDeletedMessageView(
 export class Convo {
   private id: string
 
-  private agent: AtpAgent
+  private chatClient: Client
   private events: MessagesEventBus
   private senderUserDid: string
 
@@ -131,7 +129,7 @@ export class Convo {
     string,
     {
       id: string
-      message: ChatBskyConvoSendMessage.InputSchema['message']
+      message: chat.bsky.convo.defs.MessageInput
       optimisticEmbedView?:
         | $Typed<AppBskyEmbedRecord.View>
         | $Typed<ChatBskyEmbedJoinLink.View>
@@ -166,9 +164,9 @@ export class Convo {
   constructor(params: ConvoParams) {
     this.id = nanoid(3)
     this.convoId = params.convoId
-    this.agent = params.agent
+    this.chatClient = params.chatClient
     this.events = params.events
-    this.senderUserDid = params.agent.assertDid
+    this.senderUserDid = params.chatClient.assertDid
 
     if (params.placeholderData) {
       this.setupPlaceholderData(params.placeholderData)
@@ -197,6 +195,15 @@ export class Convo {
     this.subscribers.forEach(subscriber => subscriber())
   }
 
+  /**
+   * Point the convo at a new chat client, for when the session bundle is
+   * replaced underneath it. Conversation state is unaffected: the same account
+   * is being served over a fresh session.
+   */
+  updateClient(chatClient: Client) {
+    this.chatClient = chatClient
+  }
+
   private subscribers: (() => void)[] = []
 
   subscribe(subscriber: () => void) {
@@ -720,13 +727,12 @@ export class Convo {
     this.pendingFetchConvo = (async () => {
       try {
         const response = await networkRetry(2, () => {
-          return this.agent.chat.bsky.convo.getConvo(
-            {convoId: this.convoId},
-            {headers: DM_SERVICE_HEADERS},
-          )
+          return this.chatClient.call(chat.bsky.convo.getConvo, {
+            convoId: this.convoId,
+          })
         })
 
-        const convo = response.data.convo
+        const convo = response.convo
 
         return {
           convo,
@@ -763,18 +769,15 @@ export class Convo {
     let cursor: string | undefined
     do {
       const result = await networkRetry(2, () => {
-        return this.agent.chat.bsky.convo.getConvoMembers(
-          {
-            convoId: this.convoId,
-            limit: 50,
-            cursor,
-          },
-          {headers: DM_SERVICE_HEADERS},
-        )
+        return this.chatClient.call(chat.bsky.convo.getConvoMembers, {
+          convoId: this.convoId,
+          limit: 50,
+          cursor,
+        })
       })
-      cursor = result.data.cursor
+      cursor = result.cursor
 
-      for (const member of result.data.members) {
+      for (const member of result.members) {
         this.relatedProfiles.set(member.did, member)
       }
     } while (cursor)
@@ -808,16 +811,13 @@ export class Convo {
 
       const nextCursor = this.oldestRev // for TS
       const response = await networkRetry(2, () => {
-        return this.agent.chat.bsky.convo.getMessages(
-          {
-            cursor: nextCursor,
-            convoId: this.convoId,
-            limit: IS_NATIVE ? 30 : 60,
-          },
-          {headers: DM_SERVICE_HEADERS},
-        )
+        return this.chatClient.call(chat.bsky.convo.getMessages, {
+          cursor: nextCursor,
+          convoId: this.convoId,
+          limit: IS_NATIVE ? 30 : 60,
+        })
       })
-      const {cursor, messages, relatedProfiles} = response.data
+      const {cursor, messages, relatedProfiles} = response
 
       // Trust the cursor for pagination. We can't infer "no more pages" from a
       // short page: the server pages by raw rows but strips deleted messages
@@ -1031,7 +1031,7 @@ export class Convo {
   private pendingMessageFailure: 'recoverable' | 'unrecoverable' | null = null
 
   sendMessage(
-    message: ChatBskyConvoSendMessage.InputSchema['message'],
+    message: chat.bsky.convo.defs.MessageInput,
     optimisticEmbedView?:
       | $Typed<AppBskyEmbedRecord.View>
       | $Typed<ChatBskyEmbedJoinLink.View>,
@@ -1165,14 +1165,10 @@ export class Convo {
 
       const {id, message} = pendingMessage
 
-      const response = await this.agent.chat.bsky.convo.sendMessage(
-        {
-          convoId: this.convoId,
-          message,
-        },
-        {encoding: 'application/json', headers: DM_SERVICE_HEADERS},
-      )
-      const res = response.data
+      const res = await this.chatClient.call(chat.bsky.convo.sendMessage, {
+        convoId: this.convoId,
+        message,
+      })
 
       // remove from queue
       this.pendingMessages.delete(id)
@@ -1197,8 +1193,15 @@ export class Convo {
     }
   }
 
-  private handleSendMessageFailure(e: Error | XRPCError) {
-    if (e instanceof XRPCError) {
+  /*
+   * The lex client throws `XrpcResponseError`, not `@atproto/api`'s
+   * `XRPCError`, so the status/message branch narrows on the lex class. Only
+   * a genuine server response carries a status: transport and internal lex
+   * failures are `XrpcInternalError`s and fall through to the generic arm,
+   * where `isNetworkError` keeps them out of the logs.
+   */
+  private handleSendMessageFailure(e: Error | XrpcResponseError) {
+    if (e instanceof XrpcResponseError) {
       if (NETWORK_FAILURE_STATUSES.includes(e.status)) {
         this.pendingMessageFailure = 'recoverable'
       } else {
@@ -1261,16 +1264,15 @@ export class Convo {
     )
 
     try {
-      const {data} = await this.agent.chat.bsky.convo.sendMessageBatch(
+      const {items} = await this.chatClient.call(
+        chat.bsky.convo.sendMessageBatch,
         {
           items: messageArray.map(({message}) => ({
             convoId: this.convoId,
             message,
           })),
         },
-        {encoding: 'application/json', headers: DM_SERVICE_HEADERS},
       )
-      const {items} = data
 
       /*
        * Insert into `newMessages` as soon as we have a real ID. That way, when
@@ -1304,13 +1306,10 @@ export class Convo {
 
     try {
       await networkRetry(2, () => {
-        return this.agent.chat.bsky.convo.deleteMessageForSelf(
-          {
-            convoId: this.convoId,
-            messageId,
-          },
-          {encoding: 'application/json', headers: DM_SERVICE_HEADERS},
-        )
+        return this.chatClient.call(chat.bsky.convo.deleteMessageForSelf, {
+          convoId: this.convoId,
+          messageId,
+        })
       })
     } catch (err) {
       const e = err as Error
@@ -1529,10 +1528,11 @@ export class Convo {
 
     try {
       logger.debug(`Adding reaction ${emoji} to message ${messageId}`)
-      const {data} = await this.agent.chat.bsky.convo.addReaction(
-        {messageId, value: emoji, convoId: this.convoId},
-        {encoding: 'application/json', headers: DM_SERVICE_HEADERS},
-      )
+      const data = await this.chatClient.call(chat.bsky.convo.addReaction, {
+        messageId,
+        value: emoji,
+        convoId: this.convoId,
+      })
       if (ChatBskyConvoDefs.isMessageView(data.message)) {
         if (this.pastMessages.has(messageId)) {
           this.pastMessages.set(messageId, data.message)
@@ -1594,10 +1594,11 @@ export class Convo {
 
     try {
       logger.debug(`Removing reaction ${emoji} from message ${messageId}`)
-      await this.agent.chat.bsky.convo.removeReaction(
-        {messageId, value: emoji, convoId: this.convoId},
-        {encoding: 'application/json', headers: DM_SERVICE_HEADERS},
-      )
+      await this.chatClient.call(chat.bsky.convo.removeReaction, {
+        messageId,
+        value: emoji,
+        convoId: this.convoId,
+      })
     } catch (error) {
       if (restore) restore()
       throw error

src/state/messages/convo/index.tsx

@@ -28,7 +28,7 @@ import {
 } from '#/state/queries/messages/conversation'
 import {RQKEY_ROOT as ListConvosQueryKeyRoot} from '#/state/queries/messages/list-conversations'
 import {RQKEY as createProfileQueryKey} from '#/state/queries/profile'
-import {useAgent} from '#/state/session'
+import {useChatClient} from '#/state/session'
 import {type GroupConvoMember} from '#/components/dms/util'
 
 export * from '#/state/messages/convo/util'
@@ -80,7 +80,7 @@ export function ConvoProvider({
   convoId,
 }: Pick<ConvoParams, 'convoId'> & {children: React.ReactNode}) {
   const queryClient = useQueryClient()
-  const agent = useAgent()
+  const chatClient = useChatClient()
   const events = useMessagesEventBus()
   const [convo] = useState(() => {
     const placeholder = queryClient.getQueryData<ChatBskyConvoDefs.ConvoView>(
@@ -88,7 +88,7 @@ export function ConvoProvider({
     )
     return new Convo({
       convoId,
-      agent,
+      chatClient,
       events,
       placeholderData: placeholder ? {convo: placeholder} : undefined,
     })
@@ -96,6 +96,16 @@ export function ConvoProvider({
   const service = useSyncExternalStore(convo.subscribe, convo.getSnapshot)
   const {mutate: markAsRead} = useMarkAsReadMutation()
 
+  /*
+   * The convo outlives the client it was constructed with: replacing the
+   * session bundle builds fresh clients over the new session and disposes the
+   * old ones, so a convo still holding the previous client would send through a
+   * dead session.
+   */
+  useEffect(() => {
+    convo.updateClient(chatClient)
+  }, [convo, chatClient])
+
   const appState = useAppState()
   const isActive = appState === 'active'
   useFocusEffect(

src/state/messages/convo/types.ts

@@ -1,19 +1,20 @@
 import {
   type $Typed,
   type AppBskyEmbedRecord,
-  type AtpAgent,
   type ChatBskyActorDefs,
   type ChatBskyConvoDefs,
-  type ChatBskyConvoSendMessage,
   type ChatBskyEmbedJoinLink,
 } from '@atproto/api'
+import {type Client} from '@atproto/lex'
 
 import {type MessagesEventBus} from '#/state/messages/events/agent'
 import {type ConvoWithDetails} from '#/components/dms/util'
+import {type chat} from '#/lexicons'
 
 export type ConvoParams = {
   convoId: string
-  agent: AtpAgent
+  /** The chat client, which proxies `chat.bsky.*` to the chat service. */
+  chatClient: Client
   events: MessagesEventBus
   placeholderData?: {
     convo: ChatBskyConvoDefs.ConvoView
@@ -108,7 +109,7 @@ export type ConvoItem =
 
 type DeleteMessage = (messageId: string) => Promise<void>
 type SendMessage = (
-  message: ChatBskyConvoSendMessage.InputSchema['message'],
+  message: chat.bsky.convo.defs.MessageInput,
   optimisticEmbedView:
     | $Typed<AppBskyEmbedRecord.View>
     | $Typed<ChatBskyEmbedJoinLink.View>

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']
     }