feat(connector-contract): conversations from Slack, Discord or Linear; machines list the sources they accept (#33862)

A connector conversation now names its source (slack, discord or linear), and a posted reply reports a neutral messageId. Each machine's hello lists the sources it accepts, and the cloud only routes a Slack conversation to machines that list Slack. Machine events from other sources are filtered out where they arrive, and a reply for a non-Slack conversation is refused, so a Discord or Linear conversation can never post into a Slack thread with the same id. The unused turn-bound cancel fields are removed (they return with Stop). Inert: no shipped desktop uses the contract, and Slack is off unless configured.

cloud/apps/connector/src/connector-events.test.ts

@@ -179,6 +179,34 @@ describe('connector turn events', () => {
     expect(await thread()).toMatchObject({ pending_prompt: null })
   })
 
+  it("leaves a Slack thread alone when another source's conversation has its id", async () => {
+    const { machine, enqueue, events, thread } = await setup()
+    await enqueue(machine.hostId)
+    await events([{ conversation, paneKey: 'slack-pane', kind: 'linked', agent: 'claude' }])
+    const other = { source: 'discord', externalId: conversation.externalId }
+    await events([{ conversation: other, paneKey: 'discord-pane', kind: 'linked', agent: 'codex' }])
+    const prompt = { kind: 'approval', promptId: 'p1', tool: 'Bash', options: [] }
+    await events([
+      {
+        kind: 'snapshot',
+        conversations: [
+          {
+            conversation: { ...other, source: 'linear' },
+            paneKey: 'linear-pane',
+            agent: 'codex',
+            working: true,
+            prompt
+          }
+        ]
+      }
+    ])
+    expect(await thread()).toMatchObject({
+      pane_key: 'slack-pane',
+      agent: 'claude',
+      pending_prompt: null
+    })
+  })
+
   it('reads an open turn missing from the snapshot as unverifiable', async () => {
     const { machine, enqueue, events, thread } = await setup()
     await enqueue(machine.hostId)

cloud/apps/connector/src/connector-inbox.ts

@@ -3,6 +3,7 @@ import {
   CONNECTOR_LIMITS,
   ConnectorDeliverySchema,
   type ConnectorAcksRequest,
+  type ConnectorConversation,
   type ConnectorDelivery
 } from '@orca-cloud/connector-contract'
 
@@ -19,10 +20,7 @@ export type DeliveryBody = ConnectorDelivery extends infer D
 
 export type ExpiredThread = { hostId: string; externalId: string }
 export type EnqueueResult = 'queued' | 'duplicate' | 'dropped'
-export type Unsettled = {
-  deliveryId: string
-  conversation: { source: 'slack'; externalId: string }
-}
+export type Unsettled = { deliveryId: string; conversation: ConnectorConversation }
 
 const conversation = (externalId: string) => ({ source: 'slack' as const, externalId })
 

cloud/apps/connector/src/connector-server-harness.test-fixture.ts

@@ -181,6 +181,7 @@ export async function createConnectorHarness(
             agents: ['claude'],
             promptKinds: ['approval'],
             features: ['answer', 'cancel'],
+            sources: ['slack'],
             projects: []
           },
           'PUT'

cloud/apps/connector/src/connector-threads.ts

@@ -1,5 +1,5 @@
 import type { CloudDatabase } from '@orca-cloud/cloud-database'
-import type { ConnectorEvent } from '@orca-cloud/connector-contract'
+import type { ConnectorConversation, ConnectorEvent } from '@orca-cloud/connector-contract'
 
 // Linked and snapshot (re)bind a conversation's pane. Moving to another pane forgets the old pane's
 // turn and its verdict; one it left open (working or unverifiable) stays unverifiable, never done.
@@ -12,6 +12,20 @@ const rebindParams = (paneKey: string) => [paneKey, paneKey, paneKey, paneKey]
 // Other events only touch the pane the conversation is bound to.
 const ON_BOUND_PANE = 'host_id = ? AND external_id = ? AND (pane_key IS NULL OR pane_key = ?)'
 
+// Only Slack threads are stored so far; another source's conversation never names one of them.
+export const isStored = (conversation: ConnectorConversation) => conversation.source === 'slack'
+
+// A batch narrowed to stored conversations; a snapshot keeps only its stored entries.
+export function storedEvents(events: ConnectorEvent[]): ConnectorEvent[] {
+  return events.flatMap((event): ConnectorEvent[] => {
+    if (event.kind === 'snapshot') {
+      const conversations = event.conversations.filter((entry) => isStored(entry.conversation))
+      return [{ ...event, conversations }]
+    }
+    return isStored(event.conversation) ? [event] : []
+  })
+}
+
 export type NewThread = {
   externalId: string
   teamId: string

cloud/apps/connector/src/machine-routes.ts

@@ -13,7 +13,7 @@ import {
 import type { Context, Hono, MiddlewareHandler } from 'hono'
 import { lockConnectorHost, type ConnectorHosts } from './connector-hosts.js'
 import type { ConnectorInbox, Unsettled } from './connector-inbox.js'
-import type { ConnectorThreads } from './connector-threads.js'
+import { storedEvents, type ConnectorThreads } from './connector-threads.js'
 import type { InboxWaker } from './inbox-waker.js'
 import type { MachineEnv } from './machine-auth.js'
 import type { SlackEdge } from './slack-edge.js'
@@ -162,8 +162,9 @@ export function registerMachineRoutes(
     }
     return await onCurrentHello(context, body.data.helloRev, async (transaction, hostId) => {
       if (await hosts.advanceEventsSeq(transaction, hostId, body.data.seq)) {
-        await threads.applyEvents(transaction, hostId, body.data.events)
-        await slack?.onEvents(transaction, hostId, body.data.events)
+        const events = storedEvents(body.data.events)
+        await threads.applyEvents(transaction, hostId, events)
+        await slack?.onEvents(transaction, hostId, events)
       }
     })
   })

cloud/apps/connector/src/slack-edge.ts

@@ -9,6 +9,7 @@ import type { Context, Hono, MiddlewareHandler } from 'hono'
 import { bodyLimit } from 'hono/body-limit'
 import { every } from './connector-background.js'
 import type { ConnectorInbox, ExpiredThread } from './connector-inbox.js'
+import { isStored } from './connector-threads.js'
 import type { MachineEnv } from './machine-auth.js'
 import type { SlackConfig } from './slack-config.js'
 import { readSlackEnvelope } from './slack-events.js'
@@ -136,6 +137,9 @@ export function createSlackEdge(deps: {
       }
       const { hostId } = context.get('host')
       const { replyId, conversation, turnId, text } = body.data
+      if (!isStored(conversation)) {
+        return context.json({ error: 'not_this_hosts_thread' }, 403)
+      }
       const sent = await database.transaction((transaction) =>
         queueAgentReply(transaction, posts, hostId, {
           replyId,
@@ -149,7 +153,7 @@ export function createSlackEdge(deps: {
       }
       const response: ConnectorReplyResponse =
         sent.state === 'posted' && sent.slackTs !== null
-          ? { status: 'posted', messageTs: sent.slackTs }
+          ? { status: 'posted', messageId: sent.slackTs }
           : { status: 'queued' }
       return context.json(response)
     })

cloud/apps/connector/src/slack-harness.test-fixture.ts

@@ -160,6 +160,7 @@ export async function createSlackHarness(slack: Partial<SlackConfig> = {}) {
         agents: ['claude', 'codex'],
         promptKinds: ['approval'],
         features: ['answer', 'cancel', 'reply'],
+        sources: ['slack'],
         projects
       },
       'PUT'

cloud/apps/connector/src/slack-machine-choice.test.ts

@@ -7,7 +7,8 @@ function host(
   hostId: string,
   hostName: string,
   projects: ConnectorProject[],
-  agents = ['claude']
+  agents = ['claude'],
+  sources = ['slack']
 ): KnownHost {
   return {
     hostId,
@@ -22,6 +23,7 @@ function host(
       agents,
       promptKinds: [],
       features: [],
+      sources,
       projects
     }
   }
@@ -102,6 +104,16 @@ describe('chooseMachine', () => {
     expect(choose('fix orca', identity(), [])).toEqual({ kind: 'no_machine' })
   })
 
+  it('passes over a machine that does not take Slack work', () => {
+    const orca = [project('orca', 'pl_3', 'stablyai/orca')]
+    const discordOnly = host('h2', 'build box', orca, ['claude'], ['discord'])
+    expect(choose('fix orca', identity(), [laptop, discordOnly])).toMatchObject({
+      kind: 'chosen',
+      hostId: 'h1'
+    })
+    expect(choose('fix orca', identity(), [discordOnly])).toEqual({ kind: 'no_machine' })
+  })
+
   it('reads a repository link, and lets no default override the project the message names', () => {
     const ident = identity({ defaultProject: 'web', defaultHostId: 'h1' })
     expect(choose('fix <https://github.com/stablyai/orca|this> on build box', ident)).toMatchObject(

cloud/apps/connector/src/slack-machine-choice.ts

@@ -66,7 +66,7 @@ export function chooseMachine(input: {
 
   const capable = input.hosts.flatMap((host) => {
     const agent = host.hello.agents.find((name) => CONNECTOR_AGENTS.some((known) => known === name))
-    return agent ? [{ ...host, agent }] : []
+    return agent && host.hello.sources.includes('slack') ? [{ ...host, agent }] : []
   })
   if (capable.length === 0) {
     return { kind: 'no_machine' }

cloud/apps/connector/src/slack-posting.test.ts

@@ -69,10 +69,20 @@ describe('Slack thread status and posts', () => {
     await harness.pass()
     expect(await (await harness.reply('r1', 'Here you go')).json()).toEqual({
       status: 'posted',
-      messageTs: expect.any(String)
+      messageId: expect.any(String)
     })
   })
 
+  it("never posts another source's output to a Slack thread with the same id", async () => {
+    const harness = await startedThread()
+    const other = { source: 'discord', externalId: EXTERNAL_ID }
+    await harness.events(turnStarted, { ...turnEnded, conversation: other })
+    const reply = { v: 1, replyId: 'r1', conversation: other, turnId: TURN, text: 'hi' }
+    expect((await harness.machine.call(CONNECTOR_PATHS.replies, reply)).status).toBe(403)
+    expect((await harness.posts()).filter((post) => post.kind === 'message')).toEqual([])
+    expect(await harness.status()).toBe('Working on build-box…')
+  })
+
   it('shows a refused start', async () => {
     const harness = await startedThread()
     await harness.machine.ack([

cloud/packages/connector-contract/src/connector-wire-scalars.ts

@@ -13,8 +13,10 @@ export const ConnectorHostDescription = {
   orcaVersion: z.string().min(1)
 }
 
+export const CONNECTOR_SOURCES = ['slack', 'discord', 'linear'] as const
+
 export const ConnectorConversationSchema = z.object({
-  source: z.literal('slack'),
+  source: z.enum(CONNECTOR_SOURCES),
   externalId: OpaqueIdSchema
 })
 export type ConnectorConversation = z.infer<typeof ConnectorConversationSchema>

cloud/packages/connector-contract/src/contract.test.ts

@@ -48,7 +48,7 @@ describe('poll response', () => {
       { ...base, kind: 'rewind' },
       { ...start, prompt: '' },
       { deliveryId: base.deliveryId },
-      { ...base, conversation: { source: 'linear', externalId: 'L1' }, kind: 'cancel' }
+      { ...base, conversation: { source: 'github', externalId: 'G1' }, kind: 'cancel' }
     ]) {
       expect(ConnectorDeliverySchema.safeParse(delivery).success).toBe(false)
     }
@@ -183,11 +183,13 @@ describe('hello and replies', () => {
       agents: ['claude', 'future-agent'],
       promptKinds: ['approval'],
       features: ['answer', 'future-feature'],
+      sources: ['slack', 'future-source'],
       projects: [
         { placeId: 'pl_1', name: 'orca', kind: 'git', where: { kind: 'ssh', label: 'openclaw' } }
       ]
     })
     expect(hello.features).toEqual(['answer', 'future-feature'])
+    expect(hello.sources).toEqual(['slack', 'future-source'])
   })
 
   it('caps a hello at maxHelloProjects projects with short display strings', () => {
@@ -208,6 +210,7 @@ describe('hello and replies', () => {
         agents: ['claude'],
         promptKinds: [],
         features: [],
+        sources: ['slack'],
         projects
       }).success
     const longest = 'x'.repeat(CONNECTOR_LIMITS.maxDisplayTextLength)
@@ -225,5 +228,23 @@ describe('hello and replies', () => {
 
   it('reads a reply as posted or queued', () => {
     expect(ConnectorReplyResponseSchema.parse({ status: 'queued' })).toEqual({ status: 'queued' })
+    expect(
+      ConnectorReplyResponseSchema.parse({ status: 'posted', messageId: '1793.0001' })
+    ).toEqual({
+      status: 'posted',
+      messageId: '1793.0001'
+    })
+  })
+
+  it('carries a conversation from any connected platform', () => {
+    for (const source of ['slack', 'discord', 'linear']) {
+      expect(
+        ConnectorDeliverySchema.safeParse({
+          ...base,
+          conversation: { source, externalId: 'x1' },
+          kind: 'cancel'
+        }).success
+      ).toBe(true)
+    }
   })
 })

cloud/packages/connector-contract/src/event-messages.ts

@@ -95,7 +95,7 @@ export const ConnectorReplyRequestSchema = z.object({
 })
 
 export const ConnectorReplyResponseSchema = z.discriminatedUnion('status', [
-  z.object({ status: z.literal('posted'), messageTs: z.string().min(1) }),
+  z.object({ status: z.literal('posted'), messageId: z.string().min(1) }),
   z.object({ status: z.literal('queued') })
 ])
 

cloud/packages/connector-contract/src/hello-messages.ts

@@ -32,6 +32,8 @@ export const ConnectorHelloRequestSchema = z.object({
   agents: CapabilityList,
   promptKinds: CapabilityList,
   features: CapabilityList,
+  // The cloud sends a conversation only to a machine that lists its source.
+  sources: CapabilityList,
   projects: z.array(ConnectorProjectSchema).max(CONNECTOR_LIMITS.maxHelloProjects)
 })