From e475a1ca36d664d3ae7feb75466300e5c2d864ad Mon Sep 17 00:00:00 2001
From: Samuel Newman <mozzius@protonmail.com>
Date: Fri, 12 Jun 2026 10:36:28 +0300
Subject: [PATCH] fix chat event bus error recovery and subscriber error
 handling

- recovering from an error state now re-runs init() when latestRev was
  never seeded (previously polled forever with an undefined cursor,
  silently skipping all events), and resumes from the existing cursor
  when one exists (previously discarded it and skipped the offline gap)
- emit log batches outside poll()'s try/catch so a throwing subscriber
  is logged instead of being misreported as a network/poll failure

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
---
 src/state/messages/events/agent.ts | 64 ++++++++++++++++++++++--------
 1 file changed, 47 insertions(+), 17 deletions(-)

diff --git a/src/state/messages/events/agent.ts b/src/state/messages/events/agent.ts
index 636261bd27..7794204447 100644
--- a/src/state/messages/events/agent.ts
+++ b/src/state/messages/events/agent.ts
@@ -211,17 +211,9 @@ export class MessagesEventBus {
       }
       case MessagesEventBusStatus.Error: {
         switch (action.event) {
-          case MessagesEventBusDispatchEvent.UpdatePoll: {
-            // basically reset
-            this.status = MessagesEventBusStatus.Initializing
-            this.latestRev = undefined
-            this.init()
-            break
-          }
+          case MessagesEventBusDispatchEvent.UpdatePoll:
           case MessagesEventBusDispatchEvent.Resume: {
-            this.status = MessagesEventBusStatus.Ready
-            this.resetPoll()
-            this.emitter.emit('event', {type: 'connect'})
+            this.recoverFromError()
             break
           }
         }
@@ -238,6 +230,31 @@ export class MessagesEventBus {
     })
   }
 
+  private recoverFromError() {
+    logger.debug(`recoverFromError`, {hasRev: !!this.latestRev})
+
+    if (this.latestRev === undefined) {
+      /*
+       * init() never succeeded, so we have no cursor to resume from. Re-run
+       * init() to seed latestRev. Its success path dispatches Ready, which from
+       * Initializing transitions us to Ready + resetPoll + emit connect.
+       */
+      this.status = MessagesEventBusStatus.Initializing
+      this.init()
+    } else {
+      /*
+       * A poll failed mid-session but we still have a valid cursor. Resume
+       * polling from it directly. We must NOT route through init() here: its
+       * seeding logic takes the max of the existing rev and the server's
+       * current cursor, which would skip any events that arrived while we were
+       * offline.
+       */
+      this.status = MessagesEventBusStatus.Ready
+      this.resetPoll()
+      this.emitter.emit('event', {type: 'connect'})
+    }
+  }
+
   private async init() {
     logger.debug(`init`, {})
 
@@ -337,6 +354,9 @@ export class MessagesEventBus {
     //   },
     // )
 
+    let needsEmit = false
+    let batch: ChatBskyConvoGetLog.OutputSchema['logs'] = []
+
     try {
       const response = await networkRetry(2, () => {
         return this.agent.chat.bsky.convo.getLog(
@@ -351,9 +371,6 @@ export class MessagesEventBus {
 
       const {logs: events} = response.data
 
-      let needsEmit = false
-      let batch: ChatBskyConvoGetLog.OutputSchema['logs'] = []
-
       for (const ev of events) {
         /*
          * If there's a rev, we should handle it. If there's not a rev, we don't
@@ -373,10 +390,6 @@ export class MessagesEventBus {
           }
         }
       }
-
-      if (needsEmit) {
-        this.emitter.emit('event', {type: 'logs', logs: batch})
-      }
     } catch (e: any) {
       if (!isNetworkError(e) && !isErrorMaybeAppPasswordPermissions(e)) {
         logger.error(`poll events failed`, {
@@ -397,5 +410,22 @@ export class MessagesEventBus {
     } finally {
       this.isPolling = false
     }
+
+    /*
+     * Emit outside the try/catch above so a throwing subscriber is not
+     * misreported as a poll failure (which would show a network-error banner
+     * and drop the batch, since the revs have already been consumed). poll()
+     * runs from setInterval, so we must not let the exception escape - log it
+     * and move on without dispatching Error.
+     */
+    if (needsEmit) {
+      try {
+        this.emitter.emit('event', {type: 'logs', logs: batch})
+      } catch (e) {
+        logger.error(`subscriber error handling chat events`, {
+          safeMessage: e instanceof Error ? e.message : String(e),
+        })
+      }
+    }
   }
 }