diff --git a/cloud/apps/connector/src/connector-events.test.ts b/cloud/apps/connector/src/connector-events.test.ts index 625cd2893f2..6c5bdd6212e 100644 --- a/cloud/apps/connector/src/connector-events.test.ts +++ b/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) diff --git a/cloud/apps/connector/src/connector-inbox.ts b/cloud/apps/connector/src/connector-inbox.ts index 87bce33a675..430b0a1f950 100644 --- a/cloud/apps/connector/src/connector-inbox.ts +++ b/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 }) diff --git a/cloud/apps/connector/src/connector-server-harness.test-fixture.ts b/cloud/apps/connector/src/connector-server-harness.test-fixture.ts index 50663bac011..e1caa95dffd 100644 --- a/cloud/apps/connector/src/connector-server-harness.test-fixture.ts +++ b/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' diff --git a/cloud/apps/connector/src/connector-threads.ts b/cloud/apps/connector/src/connector-threads.ts index b5ec97f82f1..c01f192305e 100644 --- a/cloud/apps/connector/src/connector-threads.ts +++ b/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 diff --git a/cloud/apps/connector/src/machine-routes.ts b/cloud/apps/connector/src/machine-routes.ts index fce702c5221..976b1d176d9 100644 --- a/cloud/apps/connector/src/machine-routes.ts +++ b/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) } }) }) diff --git a/cloud/apps/connector/src/slack-edge.ts b/cloud/apps/connector/src/slack-edge.ts index 0bce63c042e..d243212304f 100644 --- a/cloud/apps/connector/src/slack-edge.ts +++ b/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) }) diff --git a/cloud/apps/connector/src/slack-harness.test-fixture.ts b/cloud/apps/connector/src/slack-harness.test-fixture.ts index c35d40eb183..8b4e17b7ece 100644 --- a/cloud/apps/connector/src/slack-harness.test-fixture.ts +++ b/cloud/apps/connector/src/slack-harness.test-fixture.ts @@ -160,6 +160,7 @@ export async function createSlackHarness(slack: Partial = {}) { agents: ['claude', 'codex'], promptKinds: ['approval'], features: ['answer', 'cancel', 'reply'], + sources: ['slack'], projects }, 'PUT' diff --git a/cloud/apps/connector/src/slack-machine-choice.test.ts b/cloud/apps/connector/src/slack-machine-choice.test.ts index 5785c35bdee..3f58b42293b 100644 --- a/cloud/apps/connector/src/slack-machine-choice.test.ts +++ b/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 on build box', ident)).toMatchObject( diff --git a/cloud/apps/connector/src/slack-machine-choice.ts b/cloud/apps/connector/src/slack-machine-choice.ts index de0e09a784a..392215c6311 100644 --- a/cloud/apps/connector/src/slack-machine-choice.ts +++ b/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' } diff --git a/cloud/apps/connector/src/slack-posting.test.ts b/cloud/apps/connector/src/slack-posting.test.ts index a511df8bf87..c1f97fc295e 100644 --- a/cloud/apps/connector/src/slack-posting.test.ts +++ b/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([ diff --git a/cloud/packages/connector-contract/src/connector-wire-scalars.ts b/cloud/packages/connector-contract/src/connector-wire-scalars.ts index 49f0760d4ce..bb4ef4fb2ed 100644 --- a/cloud/packages/connector-contract/src/connector-wire-scalars.ts +++ b/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 diff --git a/cloud/packages/connector-contract/src/contract.test.ts b/cloud/packages/connector-contract/src/contract.test.ts index e72ef2302b8..47dd6e4da94 100644 --- a/cloud/packages/connector-contract/src/contract.test.ts +++ b/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) + } }) }) diff --git a/cloud/packages/connector-contract/src/event-messages.ts b/cloud/packages/connector-contract/src/event-messages.ts index 2f166e73965..b797863c788 100644 --- a/cloud/packages/connector-contract/src/event-messages.ts +++ b/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') }) ]) diff --git a/cloud/packages/connector-contract/src/hello-messages.ts b/cloud/packages/connector-contract/src/hello-messages.ts index 8be90e27aeb..7f0a58ec434 100644 --- a/cloud/packages/connector-contract/src/hello-messages.ts +++ b/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) })