fix(sync): prevent race condition when adding new subscriptions

When a pullAllSessions was already running, the reloadTimer triggered
by adding sessions would be skipped by the source-level concurrency
lock, causing newly added sessions to wait until the next scheduled
interval (up to 60 minutes) before syncing.

Fix: trigger individual per-session pulls via triggerPull(sourceId,
sessionId) which bypasses the source-level lock, ensuring immediate
sync for newly added sessions regardless of concurrent pulls.

Also includes: connection pre-check on data source add, removal of
duplicate "browse remote sessions" button, hiding "other" session
types, and full resync fix for deleted local sessions.

Co-authored-by: Cursor <cursoragent@cursor.com>

apps/cli/src/sync/adapters.ts

@@ -77,6 +77,11 @@ export class DirectImporter implements DataImporter {
       return this.incrementalImportFile(targetSessionId, tempFile)
     }
 
+    if (targetSessionId) {
+      this.logger.info(`[DirectImporter] Session ${targetSessionId} not found locally, need full resync`)
+      return { success: false, newMessageCount: 0, needFullResync: true }
+    }
+
     return this.fullImportFile(tempFile, externalId)
   }
 

apps/cli/src/sync/routes.ts

@@ -77,6 +77,9 @@ export function registerAutomationRoutes(server: FastifyInstance, ctx: SyncRoute
       return reply.code(404).send({ error: 'Data source not found' })
     }
     reloadTimer(request.params.id, true)
+    for (const sess of added) {
+      pullEngine.triggerPull(request.params.id, sess.id).catch(() => {})
+    }
     return added
   })
 

apps/desktop/main/ipc/api.ts

@@ -178,6 +178,9 @@ export function registerApiHandlers(_ctx: IpcContext): void {
     (_event, sourceId: string, sessions: Array<{ name: string; remoteSessionId: string }>) => {
       const added = dsManager.addSessions(sourceId, sessions)
       reloadTimer(sourceId, true)
+      for (const sess of added) {
+        pullEngine.triggerPull(sourceId, sess.id).catch(() => {})
+      }
       return added
     }
   )

packages/sync/src/pull-engine.ts

@@ -104,6 +104,7 @@ export class PullEngine {
   private dsManager: DataSourceManager
   private logger: SyncLogger
   private isImporting: () => boolean
+  private pullingSourceIds = new Set<string>()
 
   constructor(options: PullEngineOptions) {
     this.fetcher = options.fetcher
@@ -256,8 +257,17 @@ export class PullEngine {
   }
 
   async pullAllSessions(ds: DataSource): Promise<void> {
-    for (const sess of ds.sessions) {
-      await this.executePullSession(ds.id, ds, sess)
+    if (this.pullingSourceIds.has(ds.id)) {
+      this.logger.info(`[Pull] Skipping pullAllSessions for "${ds.baseUrl}": pull already in progress`)
+      return
+    }
+    this.pullingSourceIds.add(ds.id)
+    try {
+      for (const sess of ds.sessions) {
+        await this.executePullSession(ds.id, ds, sess)
+      }
+    } finally {
+      this.pullingSourceIds.delete(ds.id)
     }
   }
 
@@ -272,15 +282,24 @@ export class PullEngine {
       return { success: result.success, error: result.error }
     }
 
-    const errors: string[] = []
-    for (const sess of ds.sessions) {
-      const result = await this.executePullSession(sourceId, ds, sess)
-      if (!result.success && result.error) errors.push(`${sess.name}: ${result.error}`)
+    if (this.pullingSourceIds.has(sourceId)) {
+      this.logger.info(`[Pull] Skipping triggerPull for source ${sourceId}: pull already in progress`)
+      return { success: true }
     }
-    if (errors.length > 0) {
-      return { success: false, error: errors.join('; ') }
+    this.pullingSourceIds.add(sourceId)
+    try {
+      const errors: string[] = []
+      for (const sess of ds.sessions) {
+        const result = await this.executePullSession(sourceId, ds, sess)
+        if (!result.success && result.error) errors.push(`${sess.name}: ${result.error}`)
+      }
+      if (errors.length > 0) {
+        return { success: false, error: errors.join('; ') }
+      }
+      return { success: true }
+    } finally {
+      this.pullingSourceIds.delete(sourceId)
     }
-    return { success: true }
   }
 
   async triggerPullAll(sourceId: string): Promise<{ success: boolean; error?: string }> {

src/components/common/Settings/API/DataSourceAddModal.vue

@@ -33,6 +33,7 @@ const remoteSessions = ref<RemoteSession[]>([])
 const selectedSessionIds = ref<Set<string>>(new Set())
 const discovering = ref(false)
 const loadingMore = ref(false)
+const submitting = ref(false)
 const discoveryError = ref('')
 const discoveryKeyword = ref('')
 const discoveryNextCursor = ref<string | undefined>()
@@ -75,9 +76,13 @@ watch(
   { immediate: true }
 )
 
+const nonOtherSessions = computed(() =>
+  remoteSessions.value.filter((session) => getSessionTypeSelection(session) !== 'other')
+)
+
 const visibleRemoteSessions = computed(() => {
-  if (activeSessionTypeFilter.value === 'all') return remoteSessions.value
-  return remoteSessions.value.filter((session) => getSessionTypeSelection(session) === activeSessionTypeFilter.value)
+  if (activeSessionTypeFilter.value === 'all') return nonOtherSessions.value
+  return nonOtherSessions.value.filter((session) => getSessionTypeSelection(session) === activeSessionTypeFilter.value)
 })
 
 const visibleAvailableSessions = computed(() =>
@@ -96,14 +101,13 @@ const sessionTypeSelectionOptions = computed(() =>
       { value: 'all', label: t('settings.api.dataSources.discovery.typeAll') },
       { value: 'private', label: t('settings.api.dataSources.discovery.typePrivate') },
       { value: 'group', label: t('settings.api.dataSources.discovery.typeGroup') },
-      { value: 'other', label: t('settings.api.dataSources.discovery.typeOther') },
     ] as Array<{ value: SessionTypeFilter; label: string }>
   ).map((option) => ({
     ...option,
     count:
       option.value === 'all'
-        ? remoteSessions.value.length
-        : remoteSessions.value.filter((session) => getSessionTypeSelection(session) === option.value).length,
+        ? nonOtherSessions.value.length
+        : nonOtherSessions.value.filter((session) => getSessionTypeSelection(session) === option.value).length,
   }))
 )
 
@@ -201,22 +205,33 @@ async function handleSubmit() {
       .map((s) => ({ name: s.name, remoteSessionId: s.id }))
     emit('sessionsAdded', props.manageSource.id, sessions)
     closeModal()
-  } else {
-    const ds = await store.addDataSource({
+    return
+  }
+
+  submitting.value = true
+  discoveryError.value = ''
+  try {
+    await store.fetchRemoteSessions(formData.value.baseUrl, formData.value.token, { limit: 1 })
+  } catch (err: any) {
+    discoveryError.value = err.message || t('settings.api.dataSources.discovery.connectionCheckFailed')
+    return
+  } finally {
+    submitting.value = false
+  }
+
+  submitting.value = true
+  try {
+    await store.addDataSource({
       name: formData.value.name || undefined,
       baseUrl: formData.value.baseUrl,
       token: formData.value.token,
       intervalMinutes: formData.value.intervalMinutes,
       pullLimit: formData.value.pullLimit || undefined,
     })
-    if (ds && selectedSessionIds.value.size > 0) {
-      const sessions = remoteSessions.value
-        .filter((s) => selectedSessionIds.value.has(s.id))
-        .map((s) => ({ name: s.name, remoteSessionId: s.id }))
-      await store.addImportSessions(ds.id, sessions)
-    }
     emit('sourceAdded')
     closeModal()
+  } finally {
+    submitting.value = false
   }
 }
 
@@ -297,19 +312,6 @@ function formatMessageCount(count?: number): string {
                 />
               </div>
             </div>
-
-            <div class="flex items-center gap-2">
-              <UButton
-                color="primary"
-                variant="soft"
-                :loading="discovering"
-                :disabled="!formData.baseUrl"
-                @click="discoverSessions"
-              >
-                <UIcon name="i-heroicons-magnifying-glass" class="mr-1 h-4 w-4" />
-                {{ t('settings.api.dataSources.discovery.browse') }}
-              </UButton>
-            </div>
           </template>
 
           <!-- Error -->
@@ -320,11 +322,11 @@ function formatMessageCount(count?: number): string {
             {{ discoveryError }}
           </div>
 
-          <!-- Session list -->
-          <div v-if="hasDiscoveryRun">
+          <!-- Session list (manage mode only) -->
+          <div v-if="isManageMode && hasDiscoveryRun">
             <div class="mb-2 flex items-center justify-between">
               <span class="text-xs font-medium text-gray-700 dark:text-gray-300">
-                {{ t('settings.api.dataSources.discovery.found', { count: remoteSessions.length }) }}
+                {{ t('settings.api.dataSources.discovery.found', { count: nonOtherSessions.length }) }}
               </span>
               <button
                 class="text-xs text-blue-500 hover:text-blue-700 dark:hover:text-blue-400"
@@ -371,7 +373,7 @@ function formatMessageCount(count?: number): string {
               </div>
             </div>
             <div
-              v-if="remoteSessions.length > 0"
+              v-if="nonOtherSessions.length > 0"
               class="max-h-64 overflow-y-auto rounded-lg border border-gray-200 dark:border-gray-600"
             >
               <div
@@ -435,7 +437,7 @@ function formatMessageCount(count?: number): string {
             >
               {{ t('settings.api.dataSources.discovery.subscribe', { count: selectedSessionIds.size }) }}
             </UButton>
-            <UButton v-else color="primary" :disabled="!formData.baseUrl" @click="handleSubmit">
+            <UButton v-else color="primary" :disabled="!formData.baseUrl" :loading="submitting" @click="handleSubmit">
               {{ t('settings.api.dataSources.addBtn') }}
             </UButton>
           </div>

src/i18n/locales/en-US/settings.json

@@ -528,7 +528,8 @@
         "loadMore": "Load More",
         "subscribe": "Subscribe {count} sessions",
         "messages": "messages",
-        "error": "Failed to connect to remote server"
+        "error": "Failed to connect to remote server",
+        "connectionCheckFailed": "Cannot connect to the remote server. Please check the URL and token."
       },
       "edit": {
         "title": "Edit Data Source",

src/i18n/locales/ja-JP/settings.json

@@ -527,7 +527,8 @@
         "loadMore": "さらに読み込む",
         "subscribe": "{count} 件を購読",
         "messages": "件のメッセージ",
-        "error": "リモートサーバーへの接続に失敗しました"
+        "error": "リモートサーバーへの接続に失敗しました",
+        "connectionCheckFailed": "リモートサーバーに接続できません。URL とトークンを確認してください。"
       },
       "noSessions": "インポートセッションがありません。上の「サブスクリプション管理」から追加してください",
       "edit": {

src/i18n/locales/zh-CN/settings.json

@@ -528,7 +528,8 @@
         "loadMore": "加载更多",
         "subscribe": "订阅 {count} 个会话",
         "messages": "条消息",
-        "error": "连接远程服务器失败"
+        "error": "连接远程服务器失败",
+        "connectionCheckFailed": "无法连接到远程服务器，请检查地址和令牌是否正确"
       },
       "edit": {
         "title": "修改配置",

src/i18n/locales/zh-TW/settings.json

@@ -527,7 +527,8 @@
         "loadMore": "載入更多",
         "subscribe": "訂閱 {count} 個會話",
         "messages": "則訊息",
-        "error": "連線遠端伺服器失敗"
+        "error": "連線遠端伺服器失敗",
+        "connectionCheckFailed": "無法連線到遠端伺服器，請檢查地址和權杖是否正確"
       },
       "noSessions": "暫無匯入會話，點擊上方「管理訂閱」新增",
       "edit": {

src/stores/apiServer.ts

@@ -131,6 +131,10 @@ function createWebTransport(): ApiTransport {
 
   async function fetchJson<T>(url: string, options?: RequestInit): Promise<T> {
     const resp = await fetch(url, options)
+    if (!resp.ok) {
+      const body = await resp.json().catch(() => ({}))
+      throw new Error((body as any)?.error || `HTTP ${resp.status}`)
+    }
     return resp.json()
   }
 
@@ -379,10 +383,9 @@ export const useApiServerStore = defineStore('apiServer', () => {
   async function addImportSessions(sourceId: string, sessions: Array<{ name: string; remoteSessionId: string }>) {
     try {
       const added = await transport.addImportSessions(sourceId, sessions)
+      await fetchDataSources()
       if (added.length > 0) {
-        await triggerPullAll(sourceId)
-      } else {
-        await fetchDataSources()
+        pollDataSourceUpdates(sourceId)
       }
       return added
     } catch (err) {
@@ -391,6 +394,19 @@ export const useApiServerStore = defineStore('apiServer', () => {
     }
   }
 
+  function pollDataSourceUpdates(sourceId: string, maxAttempts = 10, intervalMs = 3000) {
+    let attempt = 0
+    const timer = setInterval(async () => {
+      attempt++
+      await fetchDataSources()
+      const ds = dataSources.value.find((s) => s.id === sourceId)
+      const allDone = ds?.sessions.every((s) => s.lastStatus !== 'idle')
+      if (allDone || attempt >= maxAttempts) {
+        clearInterval(timer)
+      }
+    }, intervalMs)
+  }
+
   async function removeImportSession(sourceId: string, sessionId: string) {
     try {
       const ok = await transport.removeImportSession(sourceId, sessionId)