Method runActivitySubscription

src/browser/stores/WorkspaceStore.ts:3207–3309

(signal: AbortSignal)

  }

  private async runActivitySubscription(signal: AbortSignal): Promise<void> {
    let attempt = 0;

    while (!signal.aborted) {
      const client = this.client ?? (await this.waitForClient(signal));
      if (!client || signal.aborted) {
        return;
      }

      const attemptController = new AbortController();
      const releaseAttemptListeners = this.linkAttemptToClientChange(attemptController, signal);
      const watchdog = this.createStallWatchdog(attemptController, "activity subscription");

      try {
        // Open the live delta stream first so no state transition can be lost
        // between the list snapshot fetch and subscribe registration.
        const iterator = await client.workspace.activity.subscribe(undefined, {
          signal: attemptController.signal,
        });

        const snapshots = await client.workspace.activity.list();
        if (signal.aborted) {
          return;
        }
        // Client changed while list() was in flight — retry with the new client
        // instead of exiting permanently. The outer while loop will pick up the
        // replacement client on the next iteration.
        if (attemptController.signal.aborted) {
          continue;
        }

        queueMicrotask(() => {
          if (signal.aborted || attemptController.signal.aborted) {
            return;
          }
          this.applyWorkspaceActivityList(snapshots);
          this.markActivityHydrated();
        });

        // Start watchdog after bootstrap so slow list() doesn't trigger
        // false-positive reconnects.
        watchdog.start();

        for await (const event of iterator) {
          if (signal.aborted) {
            return;
          }

          watchdog.markEvent();

          // Connection is alive again - don't carry old backoff into the next failure.
          attempt = 0;

          if (event.type === "heartbeat") {
            continue;
          }

          queueMicrotask(() => {