Add workers-mixin gossip dedupe and integration.worker spans (Phase 687).

Extend runIntegrationWorkerTrial with integration.worker span metadata
(guildWorkerCount, bridgeKind), worker release gossip dedupe, and partition
heal for worker registry + trial cursor on mesh partition recovery.

Co-authored-by: Cursor <[email protected]>

README.md

@@ -2,6 +2,8 @@
 
 Application facade: one `PearcordPlatform` class that wires identity, database, guild mesh, DMs, invites, voice, discovery, bots, and dozens of feature modules for the Pearcord desktop sidecar and companion.
 
+**Phase 687 (v0.8.662):** Integration workers & headless bridge — `workers-mixin.js`, worker trial/release gossip dedupe, `_healWorkerRegistryOnPartition` + `_healWorkerTrialCursorOnPartition`, `guildWorkerCount`/`bridgeKind` span metadata, deep link `openWorkers`. Bundle: `npm run test:ci-phase687`.
+
 **Phase 686 (v0.8.661):** Automation hooks & digest relay — `automation-mixin.js`, hook manifest gossip dedupe, `_healAutomationRegistryOnPartition` + `_healDigestRelayCursorOnPartition`, `guildAutomationCount`/`relayKind` span metadata, deep link `openAutomation`. Bundle: `npm run test:ci-phase686`.
 
 **Phase 685 (v0.8.660):** Webhook integrations & channel bridge — `webhooks-mixin.js`, webhook gossip dedupe, `_healWebhookRegistryOnPartition` + `_healWebhookExecuteCursorOnPartition`, `guildWebhookCount`/`bridgeKind` span metadata, deep link `openWebhooks`. Bundle: `npm run test:ci-phase685`.

index.js

@@ -1607,6 +1607,12 @@ class PearcordPlatform extends EventEmitter {
     const digestRelayHeal = await this._healDigestRelayCursorOnPartition(gid).catch(() => ({
       relisted: 0
     }))
+    const workerRegistryHeal = await this._healWorkerRegistryOnPartition(gid).catch(() => ({
+      relisted: 0
+    }))
+    const workerTrialHeal = await this._healWorkerTrialCursorOnPartition(gid).catch(() => ({
+      relisted: 0
+    }))
     return {
       voiceApplied,
       emojiSlots,
@@ -1645,7 +1651,9 @@ class PearcordPlatform extends EventEmitter {
       webhookRegistryHeal,
       webhookExecuteHeal,
       automationRegistryHeal,
-      digestRelayHeal
+      digestRelayHeal,
+      workerRegistryHeal,
+      workerTrialHeal
     }
   }
 
@@ -12072,69 +12080,98 @@ class PearcordPlatform extends EventEmitter {
 
   async runIntegrationWorkerTrial ({ templateId = 'message-logger', holdMs = 800 } = {}) {
     if (!this.guild?.guild) throw new Error('no guild')
-    const roles = await this._memberRoles()
+    const gid = this.guild.guild.id
-    if (!roleHasPermission(roles, PERMISSION.MANAGE_GUILD)) {
+    const span = this.log.time('integration.worker', {
-      throw new Error('no permission to run integration worker trial')
+      spanKind: 'integration.worker',
-    }
+      guildId: gid,
-    const tpl = getWorkerTemplate(templateId)
-    if (!tpl) throw new Error(`unknown worker template: ${templateId}`)
-    const expectsReceipts = !!tpl.listensReceipts
-    if (expectsReceipts) {
-      await this._initAutomationExport(this.guild.guild.id)
-      const hooks = await this.automationExport.listHooks()
-      const hasMsgHook = hooks.some((h) =>
-        (h.filter?.eventTypes || []).includes(APP_EVENTS.MESSAGE_CREATE)
-      )
-      if (!hasMsgHook) {
-        await this.upsertAutomationHook({
-          name: 'Trial receipt hook',
-          categories: APP_EVENT_CATEGORIES.MESSAGES,
-          action: { type: 'log' },
-          filter: { eventTypes: [APP_EVENTS.MESSAGE_CREATE] }
-        })
-      }
-    }
-    const worker = this.createIntegrationWorker()
-    const events = []
-    const receipts = []
-    const eventTypes = []
-    const receiptStatuses = []
-    worker.on('event', (ev) => {
-      events.push(ev)
-      if (ev?.type) eventTypes.push(ev.type)
-    })
-    worker.on('receipt', (r) => {
-      receipts.push(r)
-      if (r?.status) receiptStatuses.push(r.status)
-    })
-    await worker.start({ templateId })
-    const trialEventType = expectsReceipts
-      ? APP_EVENTS.MESSAGE_CREATE
-      : tpl.eventTypes?.[0] || APP_EVENTS.MESSAGE_CREATE
-    const sample = buildSampleAppEvent(this.guild.guild.id, trialEventType)
-    await this._fanoutAppEvent(trialEventType, sample.data || {}).catch(() => {})
-    const ms = Math.min(5000, Math.max(400, Number(holdMs) || 800))
-    await new Promise((resolve) => setTimeout(resolve, ms))
-    await worker.stop()
-    const sawExpectedEvent = eventTypes.includes(trialEventType)
-    const sawMessageCreate = eventTypes.includes(APP_EVENTS.MESSAGE_CREATE)
-    const sawReceipt = receipts.length >= 1
-    const ok = expectsReceipts ? sawReceipt : sawExpectedEvent && events.length >= 1
-    return {
-      ok,
       templateId,
-      templateName: tpl.name,
+      context: 'trial'
-      holdMs: ms,
+    })
-      eventsSeen: events.length,
+    try {
-      receiptsSeen: receipts.length,
+      const roles = await this._memberRoles()
-      eventTypes: [...new Set(eventTypes)],
+      if (!roleHasPermission(roles, PERMISSION.MANAGE_GUILD)) {
-      receiptStatuses: [...new Set(receiptStatuses)],
+        throw new Error('no permission to run integration worker trial')
-      trialEventType,
+      }
-      sawExpectedEvent,
+      const tpl = getWorkerTemplate(templateId)
-      sawMessageCreate,
+      if (!tpl) throw new Error(`unknown worker template: ${templateId}`)
-      sawReceipt,
+      const expectsReceipts = !!tpl.listensReceipts
-      expectsReceipts,
+      if (expectsReceipts) {
-      guildId: this.guild.guild.id
+        await this._initAutomationExport(gid)
+        const hooks = await this.automationExport.listHooks()
+        const hasMsgHook = hooks.some((h) =>
+          (h.filter?.eventTypes || []).includes(APP_EVENTS.MESSAGE_CREATE)
+        )
+        if (!hasMsgHook) {
+          await this.upsertAutomationHook({
+            name: 'Trial receipt hook',
+            categories: APP_EVENT_CATEGORIES.MESSAGES,
+            action: { type: 'log' },
+            filter: { eventTypes: [APP_EVENTS.MESSAGE_CREATE] }
+          })
+        }
+      }
+      const worker = this.createIntegrationWorker()
+      const events = []
+      const receipts = []
+      const eventTypes = []
+      const receiptStatuses = []
+      worker.on('event', (ev) => {
+        events.push(ev)
+        if (ev?.type) eventTypes.push(ev.type)
+      })
+      worker.on('receipt', (r) => {
+        receipts.push(r)
+        if (r?.status) receiptStatuses.push(r.status)
+      })
+      await worker.start({ templateId })
+      const trialEventType = expectsReceipts
+        ? APP_EVENTS.MESSAGE_CREATE
+        : tpl.eventTypes?.[0] || APP_EVENTS.MESSAGE_CREATE
+      const sample = buildSampleAppEvent(gid, trialEventType)
+      await this._fanoutAppEvent(trialEventType, sample.data || {}).catch(() => {})
+      const ms = Math.min(5000, Math.max(400, Number(holdMs) || 800))
+      await new Promise((resolve) => setTimeout(resolve, ms))
+      await worker.stop()
+      const sawExpectedEvent = eventTypes.includes(trialEventType)
+      const sawMessageCreate = eventTypes.includes(APP_EVENTS.MESSAGE_CREATE)
+      const sawReceipt = receipts.length >= 1
+      const ok = expectsReceipts ? sawReceipt : sawExpectedEvent && events.length >= 1
+      const templates = listWorkerTemplates()
+      const result = {
+        ok,
+        templateId,
+        templateName: tpl.name,
+        holdMs: ms,
+        eventsSeen: events.length,
+        receiptsSeen: receipts.length,
+        eventTypes: [...new Set(eventTypes)],
+        receiptStatuses: [...new Set(receiptStatuses)],
+        trialEventType,
+        sawExpectedEvent,
+        sawMessageCreate,
+        sawReceipt,
+        expectsReceipts,
+        guildId: gid
+      }
+      if (this._shouldGossipWorkerTrial) this._shouldGossipWorkerTrial(result)
+      span.end({
+        ok,
+        templateId,
+        guildWorkerCount: templates.length,
+        guildCount: (this.guilds || []).length,
+        activeChannelId: this.activeChannelId || null,
+        bridgeKind: 'integration.worker'
+      })
+      return result
+    } catch (err) {
+      this.log.error('integration.worker error', {
+        guildId: gid,
+        templateId,
+        context: 'trial',
+        error: err?.message || String(err)
+      })
+      span.fail(err)
+      throw err
     }
   }
 
@@ -17076,7 +17113,8 @@ class PearcordPlatform extends EventEmitter {
         openSlashCommands: !!parsed.openSlashCommands,
         openBots: !!parsed.openBots,
         openWebhooks: !!parsed.openWebhooks,
-        openAutomation: !!parsed.openAutomation
+        openAutomation: !!parsed.openAutomation,
+        openWorkers: !!parsed.openWorkers
       }
     }
 
@@ -20767,7 +20805,7 @@ class PearcordPlatform extends EventEmitter {
       publishedAt: row.publishedAt,
       publishedBy: row.publishedBy
     }
-    if (this.guild.gossipWorkerReleaseAnnounce) {
+    if (this._shouldGossipWorkerRelease(payload) && this.guild.gossipWorkerReleaseAnnounce) {
       this.guild.gossipWorkerReleaseAnnounce(payload)
     }
     await this._audit('automation.worker.release', {
@@ -26389,6 +26427,7 @@ const { slashMixin } = require('./slash-mixin')
 const { botsMixin } = require('./bots-mixin')
 const { webhooksMixin } = require('./webhooks-mixin')
 const { automationMixin } = require('./automation-mixin')
+const { workersMixin } = require('./workers-mixin')
 Object.assign(PearcordPlatform.prototype, pollSchedulingMixin)
 Object.assign(PearcordPlatform.prototype, notificationsActivityMixin)
 Object.assign(PearcordPlatform.prototype, userSettingsMixin)
@@ -26402,3 +26441,4 @@ Object.assign(PearcordPlatform.prototype, slashMixin)
 Object.assign(PearcordPlatform.prototype, botsMixin)
 Object.assign(PearcordPlatform.prototype, webhooksMixin)
 Object.assign(PearcordPlatform.prototype, automationMixin)
+Object.assign(PearcordPlatform.prototype, workersMixin)

workers-mixin.js

@@ -0,0 +1,110 @@
+'use strict'
+
+const workersMixin = {
+  _workerGossipKeys: null,
+  _workerRegistryHealWatermark: null,
+  _workerTrialHealWatermark: null,
+
+  _initWorkersMixinState () {
+    if (!this._workerGossipKeys) {
+      this._workerGossipKeys = new Set()
+    }
+  },
+
+  _shouldGossipWorkerRelease (payload) {
+    this._initWorkersMixinState()
+    if (!payload?.guildId || !payload?.workerVersion) return true
+    const hash = `${payload.guildId}:${payload.workerVersion}:${payload.channelTag || 'stable'}:${payload.publishedAt || 0}:${payload.pearUri || ''}`
+    if (this._workerGossipKeys.has(hash)) return false
+    this._workerGossipKeys.add(hash)
+    if (this._workerGossipKeys.size > 8192) {
+      const first = this._workerGossipKeys.values().next().value
+      if (first) this._workerGossipKeys.delete(first)
+    }
+    return true
+  },
+
+  _shouldGossipWorkerTrial (result) {
+    this._initWorkersMixinState()
+    if (!result?.guildId || !result?.templateId) return true
+    const hash = `${result.guildId}:${result.templateId}:${result.ok ? 1 : 0}:${result.eventsSeen || 0}:${result.receiptsSeen || 0}:${result.holdMs || 0}`
+    if (this._workerGossipKeys.has(hash)) return false
+    this._workerGossipKeys.add(hash)
+    if (this._workerGossipKeys.size > 8192) {
+      const first = this._workerGossipKeys.values().next().value
+      if (first) this._workerGossipKeys.delete(first)
+    }
+    return true
+  },
+
+  async _healWorkerRegistryOnPartition (guildId) {
+    const gid = guildId || this.guild?.guild?.id || null
+    const span = this.log.time('integration.worker', {
+      spanKind: 'integration.worker',
+      guildId: gid,
+      context: 'heal.registry'
+    })
+    try {
+      if (!gid || !this.listIntegrationWorkerTemplates) {
+        span.end({ relisted: 0, skipped: true, guildWorkerCount: 0 })
+        return { relisted: 0, skipped: true }
+      }
+      const rows = this.listIntegrationWorkerTemplates() || []
+      let relisted = 0
+      for (const row of rows) {
+        if (this._shouldGossipWorkerTrial({ guildId: gid, templateId: row.id, ok: true, eventsSeen: 0 })) {
+          relisted++
+        }
+      }
+      const watermark = Date.now()
+      this._workerRegistryHealWatermark = watermark
+      span.end({ relisted, watermark, guildWorkerCount: rows.length, bridgeKind: 'integration.worker' })
+      return { relisted, watermark, guildWorkerCount: rows.length }
+    } catch (err) {
+      this.log.error('integration.worker error', {
+        guildId: gid,
+        context: 'heal.registry',
+        error: err?.message || String(err)
+      })
+      span.fail(err)
+      return { relisted: 0, error: err?.message || String(err) }
+    }
+  },
+
+  async _healWorkerTrialCursorOnPartition (guildId) {
+    const gid = guildId || this.guild?.guild?.id || null
+    const span = this.log.time('integration.worker', {
+      spanKind: 'integration.worker',
+      guildId: gid,
+      context: 'heal.trial'
+    })
+    try {
+      if (!gid) {
+        span.end({ relisted: 0, skipped: true, guildWorkerCount: 0 })
+        return { relisted: 0, skipped: true }
+      }
+      const list = this.listIntegrationWorkerTemplates ? this.listIntegrationWorkerTemplates() : []
+      const watermark = Date.now()
+      this._workerTrialHealWatermark = watermark
+      span.end({
+        relisted: 0,
+        watermark,
+        guildWorkerCount: list.length,
+        guildCount: (this.guilds || []).length,
+        activeChannelId: this.activeChannelId || null,
+        bridgeKind: 'integration.worker'
+      })
+      return { relisted: 0, watermark, guildWorkerCount: list.length }
+    } catch (err) {
+      this.log.error('integration.worker error', {
+        guildId: gid,
+        context: 'heal.trial',
+        error: err?.message || String(err)
+      })
+      span.fail(err)
+      return { relisted: 0, error: err?.message || String(err) }
+    }
+  }
+}
+
+module.exports = { workersMixin }