diff --git a/.changeset/slack-thread-unsubscribe.md b/.changeset/slack-thread-unsubscribe.md new file mode 100644 index 000000000..75e2f98fb --- /dev/null +++ b/.changeset/slack-thread-unsubscribe.md @@ -0,0 +1,5 @@ +--- +"eve": patch +--- + +Added Slack `threadParticipation` configuration for following active threads with arbitrary durable user-defined state. Handlers independently decide whether to respond and which JSON-serializable state is available to the next thread reply. diff --git a/docs/channels/slack.mdx b/docs/channels/slack.mdx index 9d14879a2..807ac786a 100644 --- a/docs/channels/slack.mdx +++ b/docs/channels/slack.mdx @@ -71,24 +71,90 @@ Only the first available handler runs. A message hook returning `null` drops the `onInteraction(action, ctx)` separately handles `block_actions` callbacks not consumed by HITL. eve attaches the triggering Slack user id to the same model message as its text, preserving speaker attribution without profile lookups. -#### Continue conversations without repeated mentions +#### Participate in active threads -Use `onMessage` to handle explicit mentions and continue replies in threads with an active eve session: +The API grows from mention-only behavior to automatic or stateful participation. + +**1. Respond only to explicit mentions.** This is the default: ```ts title="agent/channels/slack.ts" export default slackChannel({ credentials: connectSlackCredentials("slack/my-agent"), - async onMessage(ctx, message) { - if (message.author?.isBot) return null; - const isDirectMessage = message.raw.channel_type === "im"; - return isDirectMessage || ctx.isBotMentioned() || (await ctx.isSubscribed()) - ? { auth: null } - : null; +}); +``` + +**2. Respond to every human reply in active threads.** Enable `threadParticipation` without requiring another mention: + +```ts title="agent/channels/slack.ts" +export default slackChannel({ + credentials: connectSlackCredentials("slack/my-agent"), + threadParticipation: true, +}); +``` + +For Vercel Connect, open **Advanced** when creating the connector and add `message.channels` under **Trigger Event Types** and `channels:history` under **Bot Scopes**. Private channels additionally need `message.groups` and `groups:history`. + +**3. Track your own participation state.** Provide an `initialState` and return the next state from `handle`. This example treats mentioning another person as a handoff: + +```ts title="agent/channels/slack.ts" +type ParticipationState = "followingThread" | "ignoringThread"; + +export default slackChannel({ + credentials: connectSlackCredentials("slack/my-agent"), + threadParticipation: { + initialState: "followingThread" as ParticipationState, + handle(message, ctx) { + if (ctx.isBotMentioned) return { respond: true, state: "followingThread" }; + if (ctx.state === "ignoringThread") return { respond: false, state: ctx.state }; + + const mentionsSomeone = /<@[A-Z0-9]+(?:\|[^>]+)?>/.test(message.text); + return mentionsSomeone + ? { respond: false, state: "ignoringThread" } + : { respond: true, state: ctx.state }; + }, + }, +}); +``` + +The handler runs for replies in a thread, including explicit mentions, but not for the top-level message that starts the thread. It runs inside the durable session after Slack channel state and `defineState` are hydrated. `state` must be JSON-serializable. + +**4. Classify ambiguous replies.** The same state can distinguish a permanent handoff from one ignored message: + +```ts title="agent/channels/slack.ts" +import { generateText, Output } from "ai"; +import { z } from "zod"; + +type ParticipationState = "followingThread" | "ignoringThread"; + +export default slackChannel({ + credentials: connectSlackCredentials("slack/my-agent"), + threadParticipation: { + initialState: "followingThread" as ParticipationState, + async handle(message, ctx) { + if (ctx.isBotMentioned) return { respond: true, state: "followingThread" }; + if (ctx.state === "ignoringThread") return { respond: false, state: ctx.state }; + + await ctx.thread.refresh(); + const conversation = ctx.thread.recentMessages + .map((entry) => `${entry.isMe ? "agent" : (entry.user ?? "unknown")}: ${entry.text}`) + .join("\n"); + const { output } = await generateText({ + model: "google/gemini-3.1-flash-lite", + system: "Decide whether the final Slack message expects the agent to answer.", + prompt: conversation, + output: Output.object({ schema: z.object({ respond: z.boolean() }) }), + timeout: 1_500, + maxRetries: 0, + }); + return { respond: output.respond, state: ctx.state }; + }, }, }); ``` -`isBotMentioned()` identifies an explicit mention. `isSubscribed()` checks whether the message belongs to a thread with an active eve session. For Vercel Connect, open **Advanced** when creating the connector and add `message.channels` under **Trigger Event Types** and `channels:history` under **Bot Scopes**. Private channels additionally need `message.groups` and `groups:history`. +`ctx.thread.refresh()` loads up to 50 Slack messages into `ctx.thread.recentMessages`. AI SDK string model ids use AI Gateway and the same ambient `AI_GATEWAY_API_KEY` or `VERCEL_OIDC_TOKEN` credentials as string models in `defineAgent`; the standalone call does not add classifier input to agent history. + +Use `onMessage` only when you need lower-level webhook-side admission, custom auth, or pre-dispatch behavior. `ctx.isBotMentioned()` identifies explicit mentions, while `ctx.isSubscribed()` is a point-in-time check for an active eve session; it does not inspect `threadParticipation` state. Use `ctx.thread.listParticipants()` when routing depends on who has joined the thread. It fetches the current thread and returns unique human Slack user ids in first-appearance order, so the first id is the starting author for a human-started thread. Bot and system messages are excluded: diff --git a/packages/eve/src/channel/routes.ts b/packages/eve/src/channel/routes.ts index d2f5f2bfc..9c17923e7 100644 --- a/packages/eve/src/channel/routes.ts +++ b/packages/eve/src/channel/routes.ts @@ -44,6 +44,8 @@ export interface RouteHandlerArgs { export interface SendPayload { readonly message?: string | UserContent; readonly inputResponses?: readonly InputResponse[]; + /** @internal Channel-owned serializable data for durable delivery processing. */ + readonly channelData?: unknown; /** * Context strings contributed by the channel. eve appends each entry * as a `role: "user"` message to `session.history` before the delivery diff --git a/packages/eve/src/channel/send.test.ts b/packages/eve/src/channel/send.test.ts index 3ddc8194e..4ccce02c6 100644 --- a/packages/eve/src/channel/send.test.ts +++ b/packages/eve/src/channel/send.test.ts @@ -113,6 +113,26 @@ describe("createSendFn", () => { }); }); + it("forwards channel-owned delivery data through deliver and run inputs", async () => { + const channelData = { kind: "test-message", value: 1 }; + const deliverRuntime = createRuntime(new RuntimeNoActiveSessionError("test:token")); + vi.mocked(deliverRuntime.deliver).mockResolvedValue({ sessionId: "existing-session-id" }); + + const deliverSend = createSendFn(deliverRuntime, ADAPTER, "test"); + await deliverSend( + { channelData, message: "hello" }, + { auth: null, continuationToken: "token" }, + ); + expect(vi.mocked(deliverRuntime.deliver).mock.calls[0]?.[0].payload.channelData).toEqual( + channelData, + ); + + const runRuntime = createRuntime(new RuntimeNoActiveSessionError("test:token")); + const runSend = createSendFn(runRuntime, ADAPTER, "test"); + await runSend({ channelData, message: "hello" }, { auth: null, continuationToken: "token" }); + expect(vi.mocked(runRuntime.run).mock.calls[0]?.[0].input.channelData).toEqual(channelData); + }); + it("adds channel request ids to deliver and run inputs when provided", async () => { const deliverRuntime: Runtime = { cancelTurn: vi.fn(), diff --git a/packages/eve/src/channel/send.ts b/packages/eve/src/channel/send.ts index 2ed42b423..acc252003 100644 --- a/packages/eve/src/channel/send.ts +++ b/packages/eve/src/channel/send.ts @@ -29,6 +29,7 @@ export function createSendFn( const { message: rawMessage, inputResponses, + channelData, context, outputSchema, } = normalizeSendInput(input); @@ -39,7 +40,7 @@ export function createSendFn( auth, continuationToken, requestId: metadata.requestId, - payload: { inputResponses, message, context, outputSchema }, + payload: { inputResponses, channelData, message, context, outputSchema }, }; const { sessionId } = await runtime.deliver(deliverInput); @@ -69,7 +70,7 @@ export function createSendFn( channelName, callback, continuationToken, - input: { message: message ?? "", context, outputSchema }, + input: { channelData, message: message ?? "", context, outputSchema }, mode, requestId: metadata.requestId, title, diff --git a/packages/eve/src/channel/types.ts b/packages/eve/src/channel/types.ts index f8c3e89e6..9f89bce79 100644 --- a/packages/eve/src/channel/types.ts +++ b/packages/eve/src/channel/types.ts @@ -319,6 +319,8 @@ export interface RunInput { readonly initiatorAuth?: SessionAuthContext | null; readonly input: { readonly message: string | UserContent; + /** @internal Channel-owned serializable data for durable delivery processing. */ + readonly channelData?: unknown; readonly context?: readonly string[]; readonly outputSchema?: JsonObject; }; diff --git a/packages/eve/src/execution/workflow-entry.ts b/packages/eve/src/execution/workflow-entry.ts index 3af4d4127..f7270ccee 100644 --- a/packages/eve/src/execution/workflow-entry.ts +++ b/packages/eve/src/execution/workflow-entry.ts @@ -112,6 +112,7 @@ export async function workflowEntry(input: WorkflowEntryInput): Promise { + const results: StepInput[] = []; + for (const payload of delivery.payloads) { + const result = adapter.deliver + ? await adapter.deliver(payload, adapterCtx) + : defaultDeliverResult(payload); + + if (result !== undefined && result !== null) { + results.push(result); + } } - } - resolved = results.length === 0 ? undefined : results.reduce(coalesceTurnInputs); + return { + result: results.length === 0 ? undefined : results.reduce(coalesceTurnInputs), + session: enrichedSession, + }; + }); + resolved = scoped.result; + initialSession = scoped.session; + deliverySessionChanged = initialSession !== beforeDeliverySession; } else if (input.input?.kind === "runtime-action-result") { recordSubagentUsageSpans(input.input.results); resolved = { runtimeActionResults: input.input.results }; @@ -253,7 +266,7 @@ export async function turnStep(rawInput: TurnStepInput): Promise): boolean { * `onInteraction` callback for non-HITL clicks. */ export interface InteractionHandlerDeps { - readonly config: SlackChannelConfig; + readonly config: SlackChannelConfig; } /** diff --git a/packages/eve/src/public/channels/slack/slackChannel.test.ts b/packages/eve/src/public/channels/slack/slackChannel.test.ts index 630e27a9f..049938260 100644 --- a/packages/eve/src/public/channels/slack/slackChannel.test.ts +++ b/packages/eve/src/public/channels/slack/slackChannel.test.ts @@ -1433,7 +1433,7 @@ describe("slackChannel() inbound mention pipeline", () => { }); describe("slackChannel() onMessage", () => { - it("exposes mention and subscription helpers", async () => { + it("exposes mention and active-session helpers", async () => { const observed: Array<{ mentioned: boolean; subscribed: boolean }> = []; const onMessage = vi.fn(async (ctx) => { observed.push({ @@ -1470,7 +1470,80 @@ describe("slackChannel() onMessage", () => { ]); }); - it("allows a simple mention-or-subscription policy", async () => { + it("ignores ordinary thread replies by default", async () => { + const channel = slackChannel({ + credentials: { botToken: "xoxb-test", signingSecret: SIGNING_SECRET }, + }); + const reply = buildEventBody( + { + channel: "C01", + channel_type: "channel", + text: "continue", + thread_ts: "1700000000.000100", + ts: "1700000000.000200", + type: "message", + user: "U01", + }, + { authorizations: [{ is_bot: true, user_id: "U_BOT" }] }, + ); + + const { send } = await firePost(channel, buildSignedRequest({ body: reply })); + + expect(send).not.toHaveBeenCalled(); + }); + + it("does not start a session from an ordinary reply when threadParticipation is enabled", async () => { + const channel = slackChannel({ + credentials: { botToken: "xoxb-test", signingSecret: SIGNING_SECRET }, + threadParticipation: true, + }); + const reply = buildEventBody( + { + channel: "C01", + channel_type: "channel", + text: "continue", + thread_ts: "1700000000.000100", + ts: "1700000000.000200", + type: "message", + user: "U01", + }, + { authorizations: [{ is_bot: true, user_id: "U_BOT" }] }, + ); + + const { send } = await firePost(channel, buildSignedRequest({ body: reply }), { + resolveActiveSession: vi.fn().mockResolvedValue(undefined), + }); + + expect(send).not.toHaveBeenCalled(); + }); + + it("follows active threads when threadParticipation is enabled", async () => { + const channel = slackChannel({ + credentials: { botToken: "xoxb-test", signingSecret: SIGNING_SECRET }, + threadParticipation: true, + }); + const reply = buildEventBody( + { + channel: "C01", + channel_type: "channel", + text: "continue", + thread_ts: "1700000000.000100", + ts: "1700000000.000200", + type: "message", + user: "U01", + }, + { authorizations: [{ is_bot: true, user_id: "U_BOT" }] }, + ); + + const { send } = await firePost(channel, buildSignedRequest({ body: reply })); + + expect(send).toHaveBeenCalledWith( + expect.objectContaining({ message: expect.any(String) }), + expect.objectContaining({ continuationToken: "C01:1700000000.000100" }), + ); + }); + + it("allows an advanced onMessage admission policy", async () => { const channel = slackChannel({ credentials: { botToken: "xoxb-test", signingSecret: SIGNING_SECRET }, async onMessage(ctx) { @@ -1498,6 +1571,211 @@ describe("slackChannel() onMessage", () => { ); }); + it("resolves thread participation inside the hydrated session", async () => { + const participate = vi.fn(() => ({ respond: false, state: "ignoringThread" })); + const channel = slackChannel({ + threadParticipation: { initialState: "followingThread", handle: participate }, + }); + const adapter = withState(getAdapter(channel), { + ...THREAD_STATE, + participationState: "followingThread", + }); + const adapterCtx = buildAdapterContext(adapter, stubAccessor()); + const message = { + attachments: [], + channelId: "C01", + markdown: "handoff to @someone", + raw: {}, + teamId: "T01", + text: "handoff to <@U02>", + threadTs: "1700000000.000001", + ts: "1700000000.000002", + }; + + const result = await contextStorage.run(stubAlsContext, () => + adapter.deliver!( + { + channelData: { + isBotMentioned: false, + kind: "slack-message", + message, + }, + message: "attributed message", + }, + adapterCtx, + ), + ); + + expect(result).toBeUndefined(); + expect(adapterCtx.state.participationState).toBe("ignoringThread"); + expect(participate).toHaveBeenCalledWith( + message, + expect.objectContaining({ + state: "followingThread", + isBotMentioned: false, + session: expect.any(Object), + slack: expect.any(Object), + thread: expect.any(Object), + }), + ); + }); + + it("can ignore one message without changing participation state", async () => { + const channel = slackChannel({ + threadParticipation: { + initialState: "followingThread", + handle: () => ({ respond: false, state: "followingThread" }), + }, + }); + const adapter = withState(getAdapter(channel), { + ...THREAD_STATE, + participationState: "followingThread", + }); + const adapterCtx = buildAdapterContext(adapter, stubAccessor()); + const message = { + attachments: [], + channelId: "C01", + markdown: "thanks", + raw: {}, + teamId: "T01", + text: "thanks", + threadTs: "1700000000.000001", + ts: "1700000000.000002", + }; + + const result = await contextStorage.run(stubAlsContext, () => + adapter.deliver!( + { + channelData: { + isBotMentioned: false, + kind: "slack-message", + message, + }, + message: "attributed message", + }, + adapterCtx, + ), + ); + + expect(result).toBeUndefined(); + expect(adapterCtx.state.participationState).toBe("followingThread"); + }); + + it("does not run thread participation for a top-level session message", async () => { + const participate = vi.fn(() => ({ respond: false, state: "ignoringThread" })); + const channel = slackChannel({ + threadParticipation: { initialState: "followingThread", handle: participate }, + }); + const adapter = withState(getAdapter(channel), { + ...THREAD_STATE, + participationState: "ignoringThread", + }); + const adapterCtx = buildAdapterContext(adapter, stubAccessor()); + const message = { + attachments: [], + channelId: "C01", + markdown: "@eve start", + raw: {}, + teamId: "T01", + text: "<@U_BOT> start", + threadTs: "1700000000.000002", + ts: "1700000000.000002", + }; + + const result = await contextStorage.run(stubAlsContext, () => + adapter.deliver!( + { + channelData: { + isBotMentioned: true, + kind: "slack-message", + message, + }, + message: "attributed message", + }, + adapterCtx, + ), + ); + + expect(participate).not.toHaveBeenCalled(); + expect(result).toEqual({ message: "attributed message" }); + expect(adapterCtx.state.participationState).toBe("followingThread"); + }); + + it("runs thread participation for explicit mentions in a thread", async () => { + const participate = vi.fn(() => ({ respond: true, state: "followingThread" })); + const channel = slackChannel({ + threadParticipation: { initialState: "followingThread", handle: participate }, + }); + const adapter = withState(getAdapter(channel), { + ...THREAD_STATE, + participationState: "ignoringThread", + }); + const adapterCtx = buildAdapterContext(adapter, stubAccessor()); + const message = { + attachments: [], + channelId: "C01", + markdown: "@eve come back", + raw: {}, + teamId: "T01", + text: "<@U_BOT> come back", + threadTs: "1700000000.000001", + ts: "1700000000.000002", + }; + + await contextStorage.run(stubAlsContext, () => + adapter.deliver!( + { + channelData: { + isBotMentioned: true, + kind: "slack-message", + message, + }, + message: "attributed message", + }, + adapterCtx, + ), + ); + + expect(participate).toHaveBeenCalledWith( + message, + expect.objectContaining({ state: "ignoringThread", isBotMentioned: true }), + ); + }); + + it("responds to explicit mentions in a thread by default", async () => { + const adapter = withState(getAdapter(slackChannel()), { + ...THREAD_STATE, + participationState: "ignoringThread", + }); + const adapterCtx = buildAdapterContext(adapter, stubAccessor()); + const message = { + attachments: [], + channelId: "C01", + markdown: "@eve come back", + raw: {}, + teamId: "T01", + text: "<@U_BOT> come back", + threadTs: "1700000000.000001", + ts: "1700000000.000002", + }; + + const result = await contextStorage.run(stubAlsContext, () => + adapter.deliver!( + { + channelData: { + isBotMentioned: true, + kind: "slack-message", + message, + }, + message: "attributed message", + }, + adapterCtx, + ), + ); + + expect(result).toEqual({ message: "attributed message" }); + }); + it("passes messages from other bots to onMessage", async () => { const onMessage = vi.fn((_ctx, _message: { author?: { isBot: boolean } }) => null); const channel = slackChannel({ @@ -2867,6 +3145,7 @@ describe("constrainAuthorizationRequired", () => { const request = vi.fn().mockResolvedValue({ ok: true }); const state: SlackChannelState = { channelId: "C123", + participationState: "followingThread", threadTs: "111.222", teamId: null, triggeringUserId: "U777", diff --git a/packages/eve/src/public/channels/slack/slackChannel.ts b/packages/eve/src/public/channels/slack/slackChannel.ts index 7399bf6b9..3b85189b0 100644 --- a/packages/eve/src/public/channels/slack/slackChannel.ts +++ b/packages/eve/src/public/channels/slack/slackChannel.ts @@ -1,9 +1,12 @@ import { parseSlackWebhookBody } from "#compiled/@chat-adapter/slack/webhook.js"; +import { defaultDeliverResult, type ChannelAdapter } from "#channel/adapter.js"; +import type { CompiledChannel } from "#channel/compiled-channel.js"; import type { CrossChannelReceiveOptions } from "#channel/cross-channel-receive.js"; import type { Session, SessionHandle } from "#channel/session.js"; import type { CancelTurnResult, SessionAuthContext } from "#channel/types.js"; import type { CardElement } from "#compiled/chat/index.js"; +import { buildCallbackContext } from "#context/build-callback-context.js"; import type { SessionContext } from "#public/definitions/callback-context.js"; import type { ChannelSessionOps } from "#public/definitions/channel.js"; @@ -28,6 +31,7 @@ import { defaultInputRequestedHandler, defaultOnAppMention, defaultOnDirectMessage, + defaultSlackAuth, } from "#public/channels/slack/defaults.js"; import { parseMessageEvent, @@ -162,12 +166,77 @@ type SlackSessionFailedHandler = ( channel: SlackEventContext, ) => void | Promise; +function defaultThreadParticipation(): SlackThreadParticipationResult { + return { respond: true, state: true }; +} + +async function defaultSubscribedThreadMessage( + ctx: SlackInboundMessageContext, + message: SlackMessage, +): Promise { + if (message.author?.isBot || !(await ctx.isSubscribed())) return null; + return { auth: defaultSlackAuth(message, ctx) }; +} + +function readSlackMessageDeliveryData(value: unknown): SlackMessageDeliveryData | undefined { + if (typeof value !== "object" || value === null) return undefined; + const data = value as Partial; + if ( + data.kind !== "slack-message" || + typeof data.isBotMentioned !== "boolean" || + typeof data.message !== "object" || + data.message === null + ) { + return undefined; + } + return data as SlackMessageDeliveryData; +} + /** * JSON-serializable per-session state, stored verbatim across workflow * step boundaries. Anything written here must round-trip through * `JSON.stringify` / `JSON.parse`. */ +/** Context for a durable Slack thread participation handler. */ +export interface SlackThreadParticipationContext extends SessionContext { + /** Durable user-defined state returned after the previous thread reply. */ + readonly state: TState; + /** Whether this message explicitly mentions the installed bot. */ + readonly isBotMentioned: boolean; + /** Thread-bound Slack operations, including lazy history refresh. */ + readonly thread: SlackThread; + /** Slack identity and raw Web API access for the bound thread. */ + readonly slack: SlackHandle; +} + +/** Response decision and durable user-defined state after one thread reply. */ +export interface SlackThreadParticipationResult { + /** Whether this message should dispatch an agent turn. */ + readonly respond: boolean; + /** JSON-serializable state made available to the next thread reply. */ + readonly state: TState; +} + +/** Custom durable Slack thread participation behavior. */ +export interface SlackThreadParticipation { + /** JSON-serializable state assigned when a top-level message starts the thread. */ + readonly initialState: TState; + /** Runs for each admitted reply after durable state is hydrated. */ + handle( + message: SlackMessage, + ctx: SlackThreadParticipationContext, + ): SlackThreadParticipationResult | Promise>; +} + +interface SlackMessageDeliveryData { + readonly isBotMentioned: boolean; + readonly kind: "slack-message"; + readonly message: SlackMessage; +} + export interface SlackChannelState { + /** User-defined durable state for thread participation. */ + participationState?: unknown; /** Slack channel id seeded by the inbound mention. */ channelId: string | null; /** Slack thread root ts. */ @@ -353,7 +422,10 @@ export interface SlackInboundMessageContext extends SlackContext { * `"no_active_turn"` are successful outcomes. */ cancel(options?: SlackCancelOptions): Promise; - /** Returns whether this message belongs to a thread with an active eve session. */ + /** + * Returns whether this message belongs to a thread with an active eve session. + * This webhook-side lookup does not read the durable subscription state. + */ isSubscribed(): Promise; /** Returns whether the inbound event explicitly mentions this bot. */ isBotMentioned(): boolean; @@ -487,7 +559,7 @@ export interface SlackChannelInternalEvents extends Omit< readonly "authorization.required"?: SlackEventHandler<"authorization.required">; } -export interface SlackChannelConfig { +export interface SlackChannelConfig { readonly credentials?: SlackChannelCredentials; readonly botName?: string; @@ -511,6 +583,15 @@ export interface SlackChannelConfig { */ readonly threadContext?: LoadThreadContextMessagesOptions; + /** + * Participates in replies to threads that have an eve session. Pass `true` + * to respond to every admitted human reply, or provide user-defined durable + * state and a handler that decides whether to respond. The handler runs for + * thread replies after durable session and Slack channel state are hydrated; + * top-level messages bypass it and initialize the configured state. + */ + readonly threadParticipation?: true | SlackThreadParticipation; + /** * Handles human-authored Slack messages. Specialized `onAppMention` and * `onDirectMessage` handlers take precedence for their event types. Other @@ -634,6 +715,22 @@ export interface SlackChannel extends Channel< SlackInstrumentationMetadata > {} +function defineSlackChannel( + definition: Parameters< + typeof defineChannel< + SlackChannelState, + SlackChannelContext, + SlackReceiveTarget, + SlackInstrumentationMetadata + > + >[0] & { readonly deliver: NonNullable }, +): SlackChannel { + const { deliver, ...baseDefinition } = definition; + const channel = defineChannel(baseDefinition) as SlackChannel & + CompiledChannel; + return { ...channel, adapter: { ...channel.adapter, deliver } } as SlackChannel; +} + /** * Slack channel factory. Wires up the webhook route, mention dispatch, * interaction handling, and a baseline set of typing / error / @@ -644,7 +741,9 @@ export interface SlackChannel extends Channel< * fallback ahead of unsupplied mention and DM defaults; otherwise unsupplied * fields keep their defaults. */ -export function slackChannel(config: SlackChannelConfig = {}): SlackChannel { +export function slackChannel( + config: SlackChannelConfig = {}, +): SlackChannel { const uploadPolicy = mergeUploadPolicy(config.uploadPolicy); const slackFetchFile = createSlackFetchFile({ botToken: config.credentials?.botToken }); const authorizationRequiredOverride = config.events?.["authorization.required"]; @@ -670,12 +769,7 @@ export function slackChannel(config: SlackChannelConfig = {}): SlackChannel { // Light weight dedup mechanism - not reliable across multiple invocations. const handledEvents = new Set(); - return defineChannel< - SlackChannelState, - SlackChannelContext, - SlackReceiveTarget, - SlackInstrumentationMetadata - >({ + return defineSlackChannel({ kindHint: "slack", state: { channelId: null as string | null, @@ -685,8 +779,46 @@ export function slackChannel(config: SlackChannelConfig = {}): SlackChannel { pendingToolCallMessage: null, lastReasoningTypingAtMs: null, lastReasoningTypingStatus: null, + participationState: + typeof config.threadParticipation === "object" + ? structuredClone(config.threadParticipation.initialState) + : config.threadParticipation === true + ? true + : undefined, pendingAuthMessageTs: {}, }, + deliver: async (payload, adapterCtx) => { + const delivery = readSlackMessageDeliveryData(payload.channelData); + if (delivery === undefined) return defaultDeliverResult(payload); + if (delivery.message.ts === delivery.message.threadTs) { + if (typeof config.threadParticipation === "object") { + adapterCtx.state.participationState = structuredClone( + config.threadParticipation.initialState, + ); + } + return defaultDeliverResult(payload); + } + + const sessionCtx = buildCallbackContext(); + const slackCtx = adapterCtx as typeof adapterCtx & SlackChannelContext; + const participation = + typeof config.threadParticipation === "object" + ? await config.threadParticipation.handle(delivery.message, { + ...sessionCtx, + isBotMentioned: delivery.isBotMentioned, + slack: slackCtx.slack, + state: adapterCtx.state.participationState as TParticipationState, + thread: slackCtx.thread, + }) + : defaultThreadParticipation(); + if (typeof participation?.respond !== "boolean" || !("state" in participation)) { + throw new Error( + "slackChannel().threadParticipation.handle must return `respond` and `state` fields.", + ); + } + adapterCtx.state.participationState = participation.state; + return participation.respond ? defaultDeliverResult(payload) : undefined; + }, fetchFile: slackFetchFile, metadata(state): SlackInstrumentationMetadata { return { @@ -844,7 +976,7 @@ async function handleEventPost(input: { readonly continuationToken: string; }) => Promise<{ readonly sessionId: string } | undefined>; readonly waitUntil: (task: Promise) => void; - readonly config: SlackChannelConfig; + readonly config: SlackChannelConfig; readonly uploadPolicy: UploadPolicy; readonly handledEvents: Set; }): Promise { @@ -882,7 +1014,7 @@ async function handleEventPost(input: { const message = slackMessageFromWebhookPayload(payload); if (message !== null && !isSelfAuthoredSlackMessage({ appId, botUserId }, message)) { const dispatchMessageWith = - (handler: NonNullable) => () => + (handler: NonNullable["onAppMention"]>) => () => dispatchSlackMessage({ appId, botUserId, @@ -923,7 +1055,10 @@ async function handleEventPost(input: { } } - if (dispatch === null && config.onMessage !== undefined) { + if ( + dispatch === null && + (config.onMessage !== undefined || config.threadParticipation !== undefined) + ) { const message = parseMessageEvent(envelope); if (message !== null && !isSelfAuthoredSlackMessage({ appId, botUserId }, message)) { // Slack also emits message.channels for an app mention. The app_mention @@ -935,7 +1070,7 @@ async function handleEventPost(input: { botUserId, cancel: input.cancel, credentials: config.credentials, - handler: config.onMessage!, + handler: config.onMessage ?? defaultSubscribedThreadMessage, kind: "channel_message", message, resolveActiveSession: input.resolveActiveSession, @@ -1003,7 +1138,7 @@ async function dispatchSlackMessage(input: { readonly botUserId: string | undefined; readonly cancel: CancelFn; readonly credentials: SlackChannelCredentials | undefined; - readonly handler: NonNullable; + readonly handler: NonNullable["onMessage"]>; readonly kind: "app_mention" | "channel_message" | "direct_message"; readonly message: SlackMessage; readonly resolveActiveSession: (options: { @@ -1023,19 +1158,18 @@ async function dispatchSlackMessage(input: { threadTs: input.message.threadTs, teamId: input.message.teamId, }); + const isBotMentioned = + input.kind === "app_mention" || + (input.botUserId !== undefined && input.message.text.includes(`<@${input.botUserId}>`)); const ctx: SlackInboundMessageContext = { cancel: (options = {}) => input.cancel({ continuationToken, turnId: options.turnId, }), - isBotMentioned: () => - input.kind === "app_mention" || - (input.botUserId !== undefined && input.message.text.includes(`<@${input.botUserId}`)), + isBotMentioned: () => isBotMentioned, isSubscribed: async () => - (await input.resolveActiveSession({ - continuationToken, - })) !== undefined, + (await input.resolveActiveSession({ continuationToken })) !== undefined, reset: (options = {}) => input.reset({ continuationToken, @@ -1058,6 +1192,7 @@ async function dispatchSlackMessage(input: { await deliverSlackMessage({ credentials: input.credentials, + isBotMentioned, kind: input.kind, message: input.message, result, @@ -1073,7 +1208,7 @@ async function dispatchSlackEvent(input: { readonly cancel: CancelFn; readonly credentials: SlackChannelCredentials | undefined; readonly envelope: SlackEventEnvelope; - readonly handler: NonNullable; + readonly handler: NonNullable["onEvent"]>; readonly resolveActiveSession: (options: { readonly continuationToken: string; }) => Promise<{ readonly sessionId: string } | undefined>; @@ -1155,6 +1290,7 @@ async function verifyInbound( async function deliverSlackMessage(input: { readonly credentials: SlackChannelCredentials | undefined; + readonly isBotMentioned: boolean; readonly kind: string; readonly message: SlackMessage; readonly result: Exclude; @@ -1193,10 +1329,15 @@ async function deliverSlackMessage(input: { const channelContext = input.result.context ?? []; + const channelData: SlackMessageDeliveryData = { + isBotMentioned: input.isBotMentioned, + kind: "slack-message", + message: input.message, + }; await input.send( channelContext.length === 0 - ? { message: turnMessage } - : { message: turnMessage, context: channelContext }, + ? { channelData, message: turnMessage } + : { channelData, message: turnMessage, context: channelContext }, { auth: input.result.auth, continuationToken: slackContinuationToken(message.channelId, message.threadTs), diff --git a/pnpm-lock.yaml b/pnpm-lock.yaml index 7e0f56bd4..1bf155eaf 100644 --- a/pnpm-lock.yaml +++ b/pnpm-lock.yaml @@ -363,7 +363,7 @@ importers: version: 0.41.2 '@vercel/connect': specifier: 'catalog:' - version: 0.4.2(@ai-sdk/mcp@2.0.16(zod@4.4.3))(@auth/core@0.41.2)(@chat-adapter/slack@4.34.0(ai@7.0.34(zod@4.4.3))(bufferutil@4.1.0)(debug@4.4.3(supports-color@10.2.2))(supports-color@10.2.2)(zod@4.4.3))(ai@7.0.34(zod@4.4.3))(eve@packages+eve) + version: 0.4.2(@ai-sdk/mcp@2.0.16(zod@4.4.3))(@auth/core@0.41.2)(@chat-adapter/slack@4.34.0(ai@7.0.34(zod@4.4.3))(debug@4.4.3(supports-color@10.2.2))(supports-color@10.2.2)(zod@4.4.3))(ai@7.0.34(zod@4.4.3))(eve@packages+eve) '@vercel/oidc': specifier: 3.8.0 version: 3.8.0 @@ -15643,6 +15643,21 @@ snapshots: - utf-8-validate - zod + '@chat-adapter/slack@4.34.0(ai@7.0.34(zod@4.4.3))(debug@4.4.3(supports-color@10.2.2))(supports-color@10.2.2)(zod@4.4.3)': + dependencies: + '@chat-adapter/shared': 4.34.0(ai@7.0.34(zod@4.4.3))(supports-color@10.2.2)(zod@4.4.3) + '@slack/socket-mode': 2.0.7(bufferutil@4.1.0)(debug@4.4.3(supports-color@10.2.2))(supports-color@10.2.2) + '@slack/web-api': 7.19.0(debug@4.4.3(supports-color@10.2.2))(supports-color@10.2.2) + chat: 4.34.0(ai@7.0.34(zod@4.4.3))(supports-color@10.2.2)(zod@4.4.3) + transitivePeerDependencies: + - ai + - bufferutil + - debug + - supports-color + - utf-8-validate + - zod + optional: true + '@chat-adapter/state-memory@4.34.0(ai@7.0.34(zod@4.4.3))(supports-color@10.2.2)(zod@4.4.3)': dependencies: chat: 4.34.0(ai@7.0.34(zod@4.4.3))(supports-color@10.2.2)(zod@4.4.3) @@ -21170,13 +21185,13 @@ snapshots: ai: 6.0.191(zod@4.4.3) eve: link:packages/eve - '@vercel/connect@0.4.2(@ai-sdk/mcp@2.0.16(zod@4.4.3))(@auth/core@0.41.2)(@chat-adapter/slack@4.34.0(ai@7.0.34(zod@4.4.3))(bufferutil@4.1.0)(debug@4.4.3(supports-color@10.2.2))(supports-color@10.2.2)(zod@4.4.3))(ai@7.0.34(zod@4.4.3))(eve@packages+eve)': + '@vercel/connect@0.4.2(@ai-sdk/mcp@2.0.16(zod@4.4.3))(@auth/core@0.41.2)(@chat-adapter/slack@4.34.0(ai@7.0.34(zod@4.4.3))(debug@4.4.3(supports-color@10.2.2))(supports-color@10.2.2)(zod@4.4.3))(ai@7.0.34(zod@4.4.3))(eve@packages+eve)': dependencies: '@vercel/oidc': 3.8.0 optionalDependencies: '@ai-sdk/mcp': 2.0.16(zod@4.4.3) '@auth/core': 0.41.2 - '@chat-adapter/slack': 4.34.0(ai@7.0.34(zod@4.4.3))(bufferutil@4.1.0)(debug@4.4.3(supports-color@10.2.2))(supports-color@10.2.2)(zod@4.4.3) + '@chat-adapter/slack': 4.34.0(ai@7.0.34(zod@4.4.3))(debug@4.4.3(supports-color@10.2.2))(supports-color@10.2.2)(zod@4.4.3) ai: 7.0.34(zod@4.4.3) eve: link:packages/eve